From 3b9ef5b81b4a7f9b6aaa7d29a5087b8d23aa2acd Mon Sep 17 00:00:00 2001 From: Richard Zowalla Date: Sat, 19 Sep 2026 19:30:43 +0200 Subject: [PATCH 1/2] #2098 URLFrontier: optional TLS for the channels to the frontier ManagedChannelUtil built every gRPC channel with usePlaintext() and the module had no setting for transport security, so the URLs and their metadata, which can include cookies, always crossed the network in the clear, and a frontier which requires TLS could not be reached at all. Spout, StatusUpdaterBolt and QueueRegulatorBolt all create their channel there. The channel is now built with Grpc.newChannelBuilder and credentials read from the topology configuration. urlfrontier.tls.enabled switches to TLS channel credentials. The server certificate is checked against the PEM file in urlfrontier.tls.trust.cert.collection, or against the JVM trust store if that key is not set. urlfrontier.tls.client.cert.chain and urlfrontier.tls.client.private.key (with an optional urlfrontier.tls.client.private.key.password) add a client certificate for mutual TLS. Setting only one of the two, or naming a file which cannot be read, fails the component at startup with a message naming the key or the file. Plaintext stays the default so that existing deployments keep working; the INFO line logged when a plaintext channel is created names the key that enables TLS. The three components pass their configuration to createChannel. No new dependency is needed: the TLS credentials come from grpc-api and the transport from the grpc-netty-shaded jar the module already has. The new test covers the credentials built for each configuration and runs TLS and mutual TLS round trips against a local gRPC server with certificates generated at test time, as well as the failing cases of an untrusted server certificate, a plaintext client and a missing client certificate. --- docs/src/main/asciidoc/configuration.adoc | 5 + external/urlfrontier/README.md | 24 ++ .../stormcrawler/urlfrontier/Constants.java | 33 +++ .../urlfrontier/ManagedChannelUtil.java | 111 +++++++- .../urlfrontier/QueueRegulatorBolt.java | 2 +- .../stormcrawler/urlfrontier/Spout.java | 5 +- .../urlfrontier/StatusUpdaterBolt.java | 2 +- .../urlfrontier/ManagedChannelUtilTest.java | 250 ++++++++++++++++++ .../urlfrontier/QueueRegulatorBoltTest.java | 4 +- 9 files changed, 424 insertions(+), 12 deletions(-) create mode 100644 external/urlfrontier/src/test/java/org/apache/stormcrawler/urlfrontier/ManagedChannelUtilTest.java 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..7c159d1b2 100644 --- a/external/urlfrontier/README.md +++ b/external/urlfrontier/README.md @@ -24,6 +24,30 @@ 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: + +```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. 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); } From 2133f70fa0a21048223f8fdf42f7a19215019a69 Mon Sep 17 00:00:00 2001 From: Richard Zowalla Date: Wed, 23 Sep 2026 10:09:03 +0200 Subject: [PATCH 2/2] Document the TLS proxy needed until the frontier supports TLS and the behaviour on a failed handshake --- external/urlfrontier/README.md | 14 +++++++++++--- 1 file changed, 11 insertions(+), 3 deletions(-) diff --git a/external/urlfrontier/README.md b/external/urlfrontier/README.md index 7c159d1b2..bcce43193 100644 --- a/external/urlfrontier/README.md +++ b/external/urlfrontier/README.md @@ -28,7 +28,13 @@ urlfrontier.max.urls.per.bucket:10 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: +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 @@ -45,8 +51,10 @@ 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. The settings apply to `Spout`, `StatusUpdaterBolt` and -`QueueRegulatorBolt`. +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