#2098 URLFrontier: optional TLS for the channels to the frontier

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.

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

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";

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<String, Object> 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<String, Object> 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<String, Object> 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();
     }
 }

external/urlfrontier/src/main/java/org/apache/stormcrawler/urlfrontier/QueueRegulatorBolt.java

@@ -223,7 +223,7 @@ public void prepare(Map<String, Object> 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();
     }
 

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();

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));