diff --git a/docs/src/main/asciidoc/configuration.adoc b/docs/src/main/asciidoc/configuration.adoc index 19df72888..4f595609b 100644 --- a/docs/src/main/asciidoc/configuration.adoc +++ b/docs/src/main/asciidoc/configuration.adoc @@ -527,6 +527,11 @@ probe exercises only representative metadata. | urlfrontier.host | localhost | URLFrontier service hostname. | urlfrontier.port | 7071 | URLFrontier service port. | urlfrontier.address | - | URLFrontier service address (can be used instead of host and port). +| urlfrontier.tls.enabled | false | Use TLS on the channels to URLFrontier. If false, the channels are plaintext. +| urlfrontier.tls.trust.cert.collection | - | PEM file with the certificates trusted to sign the URLFrontier server certificate. The JVM trust store is used if not set. +| urlfrontier.tls.client.cert.chain | - | PEM file with the client certificate chain sent for mutual TLS. Must be set together with `urlfrontier.tls.client.private.key`. +| urlfrontier.tls.client.private.key | - | PKCS#8 PEM file with the private key of the client certificate. Must be set together with `urlfrontier.tls.client.cert.chain`. +| urlfrontier.tls.client.private.key.password | - | Password of the client private key, if it is encrypted. | urlfrontier.max.buckets | 10 | Number of buckets to request from the frontier. | urlfrontier.max.urls.per.bucket | 10 | Maximum URLs per bucket. Must be 1 when robots crawl-delay pacing is enabled so one request cannot hand out several URLs from the same host. | urlfrontier.robots.crawl.delay.enabled | false | Enables forwarding long robots.txt Crawl-delays from the queue stream to URLFrontier. Also requires `fetcher.max.crawl.delay.force: true` and the validated configuration described above. diff --git a/external/urlfrontier/README.md b/external/urlfrontier/README.md index 8c9a78928..bcce43193 100644 --- a/external/urlfrontier/README.md +++ b/external/urlfrontier/README.md @@ -24,6 +24,38 @@ urlfrontier.max.buckets: 10 urlfrontier.max.urls.per.bucket:10 ``` +## Transport security + +The gRPC channels to the frontier are plaintext unless TLS is enabled. Plaintext is kept as the +default so that existing deployments keep working. The channel carries the URLs and their metadata, so enable TLS whenever the +frontier runs on another host. + +URLFrontier 2.6 only listens in plaintext, so until +[crawler-commons/url-frontier#219](https://github.com/crawler-commons/url-frontier/pull/219) is +released, the frontier needs a TLS terminating proxy in front of it, and `urlfrontier.address` or +`urlfrontier.host` and `urlfrontier.port` must point at that proxy. Once the frontier supports TLS +itself, the same settings connect to it directly: + +```yaml +urlfrontier.tls.enabled: true +# PEM file with the certificates trusted to sign the server certificate; +# the JVM trust store is used if not set +urlfrontier.tls.trust.cert.collection: /etc/stormcrawler/frontier-ca.pem +# client certificate and PKCS#8 private key for mutual TLS, both or neither +urlfrontier.tls.client.cert.chain: /etc/stormcrawler/crawler.pem +urlfrontier.tls.client.private.key: /etc/stormcrawler/crawler-key.pem +# only needed if the private key is encrypted +urlfrontier.tls.client.private.key.password: changeit +``` + +The server certificate must be valid for the host name in `urlfrontier.address` or +`urlfrontier.host`. Setting only one of `urlfrontier.tls.client.cert.chain` and +`urlfrontier.tls.client.private.key`, or pointing a key at a file which cannot be read, fails +the component at startup. A failed TLS handshake, on the other hand, does not: the channel keeps +reconnecting, `Spout` and `StatusUpdaterBolt` wait for it without a deadline, and the topology +runs without fetching or updating anything. The settings apply to `Spout`, `StatusUpdaterBolt` +and `QueueRegulatorBolt`. + ## Sending discovered URLs in batches `StatusUpdaterBolt` sends known URLs (fetched, redirections, errors...) to the frontier's diff --git a/external/urlfrontier/src/main/java/org/apache/stormcrawler/urlfrontier/Constants.java b/external/urlfrontier/src/main/java/org/apache/stormcrawler/urlfrontier/Constants.java index 7c3335776..6c61bf1d8 100644 --- a/external/urlfrontier/src/main/java/org/apache/stormcrawler/urlfrontier/Constants.java +++ b/external/urlfrontier/src/main/java/org/apache/stormcrawler/urlfrontier/Constants.java @@ -27,6 +27,39 @@ private Constants() {} public static final String URLFRONTIER_DEFAULT_HOST = "localhost"; public static final int URLFRONTIER_DEFAULT_PORT = 7071; + // Transport security (#2098) + + /** + * Whether the channels to URLFrontier use TLS. Defaults to false, in which case they are + * plaintext. + */ + public static final String URLFRONTIER_TLS_ENABLED_KEY = "urlfrontier.tls.enabled"; + + /** + * Path to a PEM file with the certificates trusted to sign the URLFrontier server certificate. + * If not set, the JVM trust store is used. + */ + public static final String URLFRONTIER_TLS_TRUST_CERT_COLLECTION_KEY = + "urlfrontier.tls.trust.cert.collection"; + + /** + * Path to a PEM file with the client certificate chain sent for mutual TLS. Must be set + * together with {@link #URLFRONTIER_TLS_CLIENT_PRIVATE_KEY_KEY}. + */ + public static final String URLFRONTIER_TLS_CLIENT_CERT_CHAIN_KEY = + "urlfrontier.tls.client.cert.chain"; + + /** + * Path to the PKCS#8 PEM file with the private key of the client certificate. Must be set + * together with {@link #URLFRONTIER_TLS_CLIENT_CERT_CHAIN_KEY}. + */ + public static final String URLFRONTIER_TLS_CLIENT_PRIVATE_KEY_KEY = + "urlfrontier.tls.client.private.key"; + + /** Password of the client private key, if it is encrypted. */ + public static final String URLFRONTIER_TLS_CLIENT_PRIVATE_KEY_PASSWORD_KEY = + "urlfrontier.tls.client.private.key.password"; + // Spout public static final String URLFRONTIER_MAX_URLS_PER_BUCKET_KEY = "urlfrontier.max.urls.per.bucket"; diff --git a/external/urlfrontier/src/main/java/org/apache/stormcrawler/urlfrontier/ManagedChannelUtil.java b/external/urlfrontier/src/main/java/org/apache/stormcrawler/urlfrontier/ManagedChannelUtil.java index 35e127360..02d08766d 100644 --- a/external/urlfrontier/src/main/java/org/apache/stormcrawler/urlfrontier/ManagedChannelUtil.java +++ b/external/urlfrontier/src/main/java/org/apache/stormcrawler/urlfrontier/ManagedChannelUtil.java @@ -18,9 +18,22 @@ package org.apache.stormcrawler.urlfrontier; import static org.apache.stormcrawler.urlfrontier.Constants.URLFRONTIER_DEFAULT_PORT; +import static org.apache.stormcrawler.urlfrontier.Constants.URLFRONTIER_TLS_CLIENT_CERT_CHAIN_KEY; +import static org.apache.stormcrawler.urlfrontier.Constants.URLFRONTIER_TLS_CLIENT_PRIVATE_KEY_KEY; +import static org.apache.stormcrawler.urlfrontier.Constants.URLFRONTIER_TLS_CLIENT_PRIVATE_KEY_PASSWORD_KEY; +import static org.apache.stormcrawler.urlfrontier.Constants.URLFRONTIER_TLS_ENABLED_KEY; +import static org.apache.stormcrawler.urlfrontier.Constants.URLFRONTIER_TLS_TRUST_CERT_COLLECTION_KEY; +import io.grpc.ChannelCredentials; +import io.grpc.Grpc; +import io.grpc.InsecureChannelCredentials; import io.grpc.ManagedChannel; -import io.grpc.ManagedChannelBuilder; +import io.grpc.TlsChannelCredentials; +import java.io.File; +import java.io.IOException; +import java.util.Map; +import org.apache.commons.lang3.StringUtils; +import org.apache.stormcrawler.util.ConfUtils; import org.jetbrains.annotations.NotNull; import org.jetbrains.annotations.Range; import org.slf4j.Logger; @@ -35,20 +48,104 @@ private ManagedChannelUtil() {} private static final Logger LOG = LoggerFactory.getLogger(ManagedChannelUtil.class); - /** Gets a channel for the given host and post. */ + /** Gets a channel for the given host and port. */ @NotNull static ManagedChannel createChannel( - @NotNull String host, @Range(from = 0, to = 65535) int port) { - return createChannel(host + ":" + port); + @NotNull String host, + @Range(from = 0, to = 65535) int port, + @NotNull Map conf) { + return createChannel(host + ":" + port, conf); } + /** + * Gets a channel for the given address. The channel uses TLS when {@code + * urlfrontier.tls.enabled} is true and plaintext otherwise. + */ @NotNull - static ManagedChannel createChannel(@NotNull String address) { + static ManagedChannel createChannel( + @NotNull String address, @NotNull Map conf) { // add the default port if missing if (!address.contains(":")) { address += ":" + URLFRONTIER_DEFAULT_PORT; } - LOG.info("Initialisation of connection to URLFrontier service on {}", address); - return ManagedChannelBuilder.forTarget(address).usePlaintext().build(); + ChannelCredentials credentials = createCredentials(conf); + if (credentials instanceof TlsChannelCredentials) { + LOG.info("Initialisation of TLS connection to URLFrontier service on {}", address); + } else { + LOG.info( + "Initialisation of plaintext connection to URLFrontier service on {}; set {}" + + " to encrypt it", + address, + URLFRONTIER_TLS_ENABLED_KEY); + } + return Grpc.newChannelBuilder(address, credentials).build(); + } + + /** + * Builds the channel credentials from the configuration: plaintext unless {@code + * urlfrontier.tls.enabled} is true. With TLS the server certificate is checked against the + * certificates in {@code urlfrontier.tls.trust.cert.collection}, or against the JVM trust store + * if that key is not set. A client certificate for mutual TLS is sent when both {@code + * urlfrontier.tls.client.cert.chain} and {@code urlfrontier.tls.client.private.key} are set. + * + * @throws IllegalArgumentException if only one of the client certificate chain and private key + * is set, or if one of the configured files cannot be read + */ + @NotNull + static ChannelCredentials createCredentials(@NotNull Map conf) { + if (!ConfUtils.getBoolean(conf, URLFRONTIER_TLS_ENABLED_KEY, false)) { + return InsecureChannelCredentials.create(); + } + + TlsChannelCredentials.Builder builder = TlsChannelCredentials.newBuilder(); + + String trustCerts = ConfUtils.getString(conf, URLFRONTIER_TLS_TRUST_CERT_COLLECTION_KEY); + if (StringUtils.isNotBlank(trustCerts)) { + try { + builder.trustManager(new File(trustCerts)); + } catch (IOException | RuntimeException e) { + throw new IllegalArgumentException( + "Cannot read the certificates in " + + URLFRONTIER_TLS_TRUST_CERT_COLLECTION_KEY + + ": " + + trustCerts, + e); + } + } + + String certChain = ConfUtils.getString(conf, URLFRONTIER_TLS_CLIENT_CERT_CHAIN_KEY); + String privateKey = ConfUtils.getString(conf, URLFRONTIER_TLS_CLIENT_PRIVATE_KEY_KEY); + boolean hasCertChain = StringUtils.isNotBlank(certChain); + boolean hasPrivateKey = StringUtils.isNotBlank(privateKey); + if (hasCertChain != hasPrivateKey) { + throw new IllegalArgumentException( + URLFRONTIER_TLS_CLIENT_CERT_CHAIN_KEY + + " and " + + URLFRONTIER_TLS_CLIENT_PRIVATE_KEY_KEY + + " must be set together for mutual TLS, only " + + (hasCertChain + ? URLFRONTIER_TLS_CLIENT_CERT_CHAIN_KEY + : URLFRONTIER_TLS_CLIENT_PRIVATE_KEY_KEY) + + " is set"); + } + if (hasCertChain) { + String password = + ConfUtils.getString(conf, URLFRONTIER_TLS_CLIENT_PRIVATE_KEY_PASSWORD_KEY); + try { + builder.keyManager( + new File(certChain), + new File(privateKey), + StringUtils.isEmpty(password) ? null : password); + } catch (IOException | RuntimeException e) { + throw new IllegalArgumentException( + "Cannot read the client certificate chain " + + certChain + + " or the private key " + + privateKey, + e); + } + } + + return builder.build(); } } diff --git a/external/urlfrontier/src/main/java/org/apache/stormcrawler/urlfrontier/QueueRegulatorBolt.java b/external/urlfrontier/src/main/java/org/apache/stormcrawler/urlfrontier/QueueRegulatorBolt.java index b5bc5b338..46d9b6eb6 100644 --- a/external/urlfrontier/src/main/java/org/apache/stormcrawler/urlfrontier/QueueRegulatorBolt.java +++ b/external/urlfrontier/src/main/java/org/apache/stormcrawler/urlfrontier/QueueRegulatorBolt.java @@ -223,7 +223,7 @@ public void prepare(Map conf, TopologyContext context, OutputCol Collections.sort(addresses); address = addresses.get(context.getThisTaskIndex() % addresses.size()); } - this.channel = ManagedChannelUtil.createChannel(address); + this.channel = ManagedChannelUtil.createChannel(address, conf); this.frontier = URLFrontierGrpc.newStub(channel).withWaitForReady(); } diff --git a/external/urlfrontier/src/main/java/org/apache/stormcrawler/urlfrontier/Spout.java b/external/urlfrontier/src/main/java/org/apache/stormcrawler/urlfrontier/Spout.java index 9763051ec..f5ef866ca 100644 --- a/external/urlfrontier/src/main/java/org/apache/stormcrawler/urlfrontier/Spout.java +++ b/external/urlfrontier/src/main/java/org/apache/stormcrawler/urlfrontier/Spout.java @@ -124,9 +124,10 @@ public void open( ConfUtils.getString( stormConf, URLFRONTIER_HOST_KEY, URLFRONTIER_DEFAULT_HOST), ConfUtils.getInt( - stormConf, URLFRONTIER_PORT_KEY, URLFRONTIER_DEFAULT_PORT)); + stormConf, URLFRONTIER_PORT_KEY, URLFRONTIER_DEFAULT_PORT), + stormConf); } else { - channel = ManagedChannelUtil.createChannel(address); + channel = ManagedChannelUtil.createChannel(address, stormConf); } frontier = URLFrontierGrpc.newStub(channel).withWaitForReady(); diff --git a/external/urlfrontier/src/main/java/org/apache/stormcrawler/urlfrontier/StatusUpdaterBolt.java b/external/urlfrontier/src/main/java/org/apache/stormcrawler/urlfrontier/StatusUpdaterBolt.java index bd623e0b0..ea9ac92b6 100644 --- a/external/urlfrontier/src/main/java/org/apache/stormcrawler/urlfrontier/StatusUpdaterBolt.java +++ b/external/urlfrontier/src/main/java/org/apache/stormcrawler/urlfrontier/StatusUpdaterBolt.java @@ -288,7 +288,7 @@ public void prepare( address = host + ":" + port; } - channel = ManagedChannelUtil.createChannel(address); + channel = ManagedChannelUtil.createChannel(address, stormConf); channel.notifyWhenStateChanged( ConnectivityState.SHUTDOWN, () -> onChannelStateChange(ConnectivityState.SHUTDOWN)); diff --git a/external/urlfrontier/src/test/java/org/apache/stormcrawler/urlfrontier/ManagedChannelUtilTest.java b/external/urlfrontier/src/test/java/org/apache/stormcrawler/urlfrontier/ManagedChannelUtilTest.java new file mode 100644 index 000000000..1b9e983b2 --- /dev/null +++ b/external/urlfrontier/src/test/java/org/apache/stormcrawler/urlfrontier/ManagedChannelUtilTest.java @@ -0,0 +1,250 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to you under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.stormcrawler.urlfrontier; + +import static org.apache.stormcrawler.urlfrontier.Constants.URLFRONTIER_TLS_CLIENT_CERT_CHAIN_KEY; +import static org.apache.stormcrawler.urlfrontier.Constants.URLFRONTIER_TLS_CLIENT_PRIVATE_KEY_KEY; +import static org.apache.stormcrawler.urlfrontier.Constants.URLFRONTIER_TLS_ENABLED_KEY; +import static org.apache.stormcrawler.urlfrontier.Constants.URLFRONTIER_TLS_TRUST_CERT_COLLECTION_KEY; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import crawlercommons.urlfrontier.URLFrontierGrpc; +import crawlercommons.urlfrontier.Urlfrontier.QueueWithinCrawlParams; +import crawlercommons.urlfrontier.Urlfrontier.Stats; +import io.grpc.Grpc; +import io.grpc.InsecureChannelCredentials; +import io.grpc.ManagedChannel; +import io.grpc.Server; +import io.grpc.ServerCredentials; +import io.grpc.StatusRuntimeException; +import io.grpc.TlsChannelCredentials; +import io.grpc.TlsServerCredentials; +import io.grpc.netty.shaded.io.netty.handler.ssl.util.SelfSignedCertificate; +import io.grpc.stub.StreamObserver; +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.TimeUnit; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; + +class ManagedChannelUtilTest { + + private static SelfSignedCertificate serverCert; + private static SelfSignedCertificate clientCert; + + private Server server; + private ManagedChannel channel; + + @BeforeAll + static void createCertificates() throws Exception { + serverCert = new SelfSignedCertificate("localhost"); + clientCert = new SelfSignedCertificate("client"); + } + + @AfterAll + static void deleteCertificates() { + serverCert.delete(); + clientCert.delete(); + } + + @AfterEach + void shutdown() throws InterruptedException { + if (channel != null) { + channel.shutdownNow().awaitTermination(5, TimeUnit.SECONDS); + } + if (server != null) { + server.shutdownNow().awaitTermination(5, TimeUnit.SECONDS); + } + } + + @Test + void plaintextByDefault() { + assertInstanceOf( + InsecureChannelCredentials.class, + ManagedChannelUtil.createCredentials(new HashMap<>())); + + // the other keys are ignored while TLS is disabled + Map conf = new HashMap<>(); + conf.put(URLFRONTIER_TLS_ENABLED_KEY, false); + conf.put(URLFRONTIER_TLS_CLIENT_CERT_CHAIN_KEY, clientCert.certificate().getPath()); + assertInstanceOf( + InsecureChannelCredentials.class, ManagedChannelUtil.createCredentials(conf)); + } + + @Test + void tlsWithSystemTrustStore() { + Map conf = new HashMap<>(); + conf.put(URLFRONTIER_TLS_ENABLED_KEY, true); + TlsChannelCredentials tls = + assertInstanceOf( + TlsChannelCredentials.class, ManagedChannelUtil.createCredentials(conf)); + assertNull(tls.getRootCertificates()); + assertNull(tls.getCertificateChain()); + assertNull(tls.getPrivateKey()); + } + + @Test + void tlsWithTrustCollectionAndClientCertificate() { + TlsChannelCredentials tls = + assertInstanceOf( + TlsChannelCredentials.class, + ManagedChannelUtil.createCredentials(mutualTlsConf())); + assertNotNull(tls.getRootCertificates()); + assertNotNull(tls.getCertificateChain()); + assertNotNull(tls.getPrivateKey()); + } + + @Test + void clientCertificateWithoutPrivateKeyIsRejected() { + Map conf = mutualTlsConf(); + conf.remove(URLFRONTIER_TLS_CLIENT_PRIVATE_KEY_KEY); + IllegalArgumentException e = + assertThrows( + IllegalArgumentException.class, + () -> ManagedChannelUtil.createCredentials(conf)); + assertTrue(e.getMessage().contains(URLFRONTIER_TLS_CLIENT_PRIVATE_KEY_KEY)); + } + + @Test + void privateKeyWithoutClientCertificateIsRejected() { + Map conf = mutualTlsConf(); + conf.remove(URLFRONTIER_TLS_CLIENT_CERT_CHAIN_KEY); + IllegalArgumentException e = + assertThrows( + IllegalArgumentException.class, + () -> ManagedChannelUtil.createCredentials(conf)); + assertTrue(e.getMessage().contains(URLFRONTIER_TLS_CLIENT_CERT_CHAIN_KEY)); + } + + @Test + void unreadableTrustCollectionIsRejected() { + Map conf = new HashMap<>(); + conf.put(URLFRONTIER_TLS_ENABLED_KEY, true); + conf.put(URLFRONTIER_TLS_TRUST_CERT_COLLECTION_KEY, "/does/not/exist.pem"); + IllegalArgumentException e = + assertThrows( + IllegalArgumentException.class, + () -> ManagedChannelUtil.createCredentials(conf)); + assertTrue(e.getMessage().contains(URLFRONTIER_TLS_TRUST_CERT_COLLECTION_KEY)); + } + + @Test + void tlsRoundTrip() throws Exception { + server = startServer(serverCredentials()); + + Map conf = new HashMap<>(); + conf.put(URLFRONTIER_TLS_ENABLED_KEY, true); + conf.put(URLFRONTIER_TLS_TRUST_CERT_COLLECTION_KEY, serverCert.certificate().getPath()); + channel = ManagedChannelUtil.createChannel("localhost", server.getPort(), conf); + + assertEquals(42, getStats(channel).getSize()); + } + + @Test + void untrustedServerCertificateFails() throws Exception { + server = startServer(serverCredentials()); + + // trusts a certificate which did not sign the one of the server + Map conf = new HashMap<>(); + conf.put(URLFRONTIER_TLS_ENABLED_KEY, true); + conf.put(URLFRONTIER_TLS_TRUST_CERT_COLLECTION_KEY, clientCert.certificate().getPath()); + channel = ManagedChannelUtil.createChannel("localhost", server.getPort(), conf); + + assertThrows(StatusRuntimeException.class, () -> getStats(channel)); + } + + @Test + void plaintextClientCannotTalkToTlsServer() throws Exception { + server = startServer(serverCredentials()); + + channel = ManagedChannelUtil.createChannel("localhost", server.getPort(), new HashMap<>()); + + assertThrows(StatusRuntimeException.class, () -> getStats(channel)); + } + + @Test + void mutualTlsRoundTrip() throws Exception { + server = startServer(mutualTlsServerCredentials()); + + channel = ManagedChannelUtil.createChannel("localhost", server.getPort(), mutualTlsConf()); + + assertEquals(42, getStats(channel).getSize()); + } + + @Test + void mutualTlsWithoutClientCertificateFails() throws Exception { + server = startServer(mutualTlsServerCredentials()); + + Map conf = mutualTlsConf(); + conf.remove(URLFRONTIER_TLS_CLIENT_CERT_CHAIN_KEY); + conf.remove(URLFRONTIER_TLS_CLIENT_PRIVATE_KEY_KEY); + channel = ManagedChannelUtil.createChannel("localhost", server.getPort(), conf); + + assertThrows(StatusRuntimeException.class, () -> getStats(channel)); + } + + private static Map mutualTlsConf() { + Map conf = new HashMap<>(); + conf.put(URLFRONTIER_TLS_ENABLED_KEY, true); + conf.put(URLFRONTIER_TLS_TRUST_CERT_COLLECTION_KEY, serverCert.certificate().getPath()); + conf.put(URLFRONTIER_TLS_CLIENT_CERT_CHAIN_KEY, clientCert.certificate().getPath()); + conf.put(URLFRONTIER_TLS_CLIENT_PRIVATE_KEY_KEY, clientCert.privateKey().getPath()); + return conf; + } + + private static ServerCredentials serverCredentials() throws Exception { + return TlsServerCredentials.create(serverCert.certificate(), serverCert.privateKey()); + } + + private static ServerCredentials mutualTlsServerCredentials() throws Exception { + return TlsServerCredentials.newBuilder() + .keyManager(serverCert.certificate(), serverCert.privateKey()) + .trustManager(clientCert.certificate()) + .clientAuth(TlsServerCredentials.ClientAuth.REQUIRE) + .build(); + } + + private static Server startServer(ServerCredentials credentials) throws Exception { + return Grpc.newServerBuilderForPort(0, credentials) + .addService( + new URLFrontierGrpc.URLFrontierImplBase() { + @Override + public void getStats( + QueueWithinCrawlParams request, + StreamObserver responseObserver) { + responseObserver.onNext(Stats.newBuilder().setSize(42).build()); + responseObserver.onCompleted(); + } + }) + .build() + .start(); + } + + private static Stats getStats(ManagedChannel channel) { + return URLFrontierGrpc.newBlockingStub(channel) + .withDeadlineAfter(10, TimeUnit.SECONDS) + .getStats(QueueWithinCrawlParams.getDefaultInstance()); + } +} diff --git a/external/urlfrontier/src/test/java/org/apache/stormcrawler/urlfrontier/QueueRegulatorBoltTest.java b/external/urlfrontier/src/test/java/org/apache/stormcrawler/urlfrontier/QueueRegulatorBoltTest.java index 0d55c9c14..5bed85c8a 100644 --- a/external/urlfrontier/src/test/java/org/apache/stormcrawler/urlfrontier/QueueRegulatorBoltTest.java +++ b/external/urlfrontier/src/test/java/org/apache/stormcrawler/urlfrontier/QueueRegulatorBoltTest.java @@ -70,7 +70,9 @@ void before() { container = new URLFrontierContainer(image); container.start(); var connection = container.getFrontierConnection(); - channel = ManagedChannelUtil.createChannel(connection.getHost(), connection.getPort()); + channel = + ManagedChannelUtil.createChannel( + connection.getHost(), connection.getPort(), Map.of()); blocking = URLFrontierGrpc.newBlockingStub(channel); }