Add certificate renewal infrastructure for Collectors

graylog2-server/src/main/java/org/graylog/collectors/CollectorCaCache.java

@@ -0,0 +1,155 @@
+/*
+ * Copyright (C) 2020 Graylog, Inc.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the Server Side Public License, version 1,
+ * as published by MongoDB, Inc.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * Server Side Public License for more details.
+ *
+ * You should have received a copy of the Server Side Public License
+ * along with this program. If not, see
+ * <http://www.mongodb.com/licensing/server-side-public-license>.
+ */
+package org.graylog.collectors;
+
+import com.github.benmanes.caffeine.cache.Cache;
+import com.github.benmanes.caffeine.cache.Caffeine;
+import com.github.benmanes.caffeine.cache.Expiry;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.eventbus.EventBus;
+import com.google.common.eventbus.Subscribe;
+import com.google.common.util.concurrent.AbstractIdleService;
+import jakarta.inject.Inject;
+import jakarta.inject.Singleton;
+import org.graylog.collectors.events.CollectorCaConfigUpdated;
+import org.graylog.security.pki.CertificateEntry;
+import org.graylog.security.pki.CertificateService;
+import org.graylog.security.pki.PemUtils;
+import org.graylog2.security.encryption.EncryptedValueService;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.security.PrivateKey;
+import java.security.cert.X509Certificate;
+import java.time.Clock;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.Optional;
+import java.util.function.Supplier;
+
+import static org.graylog2.shared.utilities.StringUtils.requireNonBlank;
+
+/**
+ * Provides a CA cache that caches {@link CertificateEntry} instances based on their expiration date.
+ */
+@Singleton
+public class CollectorCaCache extends AbstractIdleService {
+    private static final Logger LOG = LoggerFactory.getLogger(CollectorCaCache.class);
+
+    private static final String SERVER_KEY = "_static_:server";
+    private static final String SIGNING_KEY = "_static_:signing";
+    private static final String CA_KEY = "_static_:ca";
+
+    private final CollectorCaService caService;
+    private final CertificateService certificateService;
+    private final EncryptedValueService encryptedValueService;
+    private final EventBus eventBus;
+    private final Cache<String, CacheEntry> cache;
+
+    public record CacheEntry(PrivateKey privateKey, X509Certificate cert, String fingerprint) {
+    }
+
+    @Inject
+    public CollectorCaCache(CollectorCaService caService,
+                            CertificateService certificateService,
+                            EncryptedValueService encryptedValueService,
+                            EventBus eventBus,
+                            Clock clock) {
+        this.caService = caService;
+        this.certificateService = certificateService;
+        this.encryptedValueService = encryptedValueService;
+        this.eventBus = eventBus;
+        this.cache = Caffeine.newBuilder()
+                .expireAfter(Expiry.<String, CacheEntry>creating((key, value) ->
+                        Duration.between(Instant.now(clock), value.cert().getNotAfter().toInstant())))
+                .initialCapacity(3)
+                .build();
+    }
+
+    /**
+     * Get entry by certificate Subject Key Identifier.
+     *
+     * @param ski the cert Subject Key Identifier value
+     * @return the cache entry or an empty optional
+     */
+    public Optional<CacheEntry> getBySubjectKeyIdentifier(String ski) {
+        requireNonBlank(ski, "Subject Key Identifier can't be blank");
+
+        return Optional.ofNullable(cache.get(ski, key -> getCacheEntry(
+                () -> certificateService.findBySubjectKeyIdentifier(ski).orElse(null)
+        ).orElse(null)));
+    }
+
+    /**
+     * Get the server entry.
+     *
+     * @return the server entry
+     */
+    public CacheEntry getServer() {
+        return cache.get(SERVER_KEY, key -> getCacheEntry(caService::getOtlpServerCert).orElseThrow(() -> new IllegalStateException("Server certificate not found")));
+    }
+
+    /**
+     * Get the signing entry.
+     *
+     * @return the signing entry
+     */
+    public CacheEntry getSigning() {
+        return cache.get(SIGNING_KEY, key -> getCacheEntry(caService::getSigningCert).orElseThrow(() -> new IllegalStateException("Signing certificate not found")));
+    }
+
+    /**
+     * Get the CA entry.
+     *
+     * @return the CA entry
+     */
+    public CacheEntry getCa() {
+        return cache.get(CA_KEY, key -> getCacheEntry(caService::getCaCert).orElseThrow(() -> new IllegalStateException("CA certificate not found")));
+    }
+
+    private Optional<CacheEntry> getCacheEntry(Supplier<CertificateEntry> certSupplier) {
+        try {
+            final var certEntry = certSupplier.get();
+            if (certEntry == null) {
+                return Optional.empty();
+            }
+            final var cert = PemUtils.parseCertificate(certEntry.certificate());
+            final var privateKey = PemUtils.parsePrivateKey(encryptedValueService.decrypt(certEntry.privateKey()));
+            LOG.debug("Loaded cert <{}>", certEntry.fingerprint());
+            return Optional.of(new CacheEntry(privateKey, cert, certEntry.fingerprint()));
+        } catch (Exception e) {
+            LOG.error("Couldn't load certificate", e);
+            throw new RuntimeException(e);
+        }
+    }
+
+    @Override
+    protected void startUp() throws Exception {
+        eventBus.register(this);
+    }
+
+    @Override
+    protected void shutDown() throws Exception {
+        eventBus.unregister(this);
+    }
+
+    @Subscribe
+    @VisibleForTesting
+    void handleCollectorsConfigEvent(CollectorCaConfigUpdated ignored) {
+        cache.invalidateAll();
+    }
+}

graylog2-server/src/main/java/org/graylog/collectors/CollectorCaKeyManager.java

@@ -0,0 +1,107 @@
+/*
+ * Copyright (C) 2020 Graylog, Inc.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the Server Side Public License, version 1,
+ * as published by MongoDB, Inc.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * Server Side Public License for more details.
+ *
+ * You should have received a copy of the Server Side Public License
+ * along with this program. If not, see
+ * <http://www.mongodb.com/licensing/server-side-public-license>.
+ */
+package org.graylog.collectors;
+
+import jakarta.inject.Inject;
+import jakarta.inject.Singleton;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Set;
+
+import javax.net.ssl.SSLEngine;
+import javax.net.ssl.X509ExtendedKeyManager;
+import java.net.Socket;
+import java.security.Principal;
+import java.security.PrivateKey;
+import java.security.cert.X509Certificate;
+
+/**
+ * Custom key manager that dynamically retrieves the server and signing certificates. This behavior is required
+ * for certificate renewal.
+ * <p>
+ * Extends {@link X509ExtendedKeyManager} rather than implementing {@link javax.net.ssl.X509KeyManager} because
+ * Netty uses {@link javax.net.ssl.SSLEngine}-based handshakes. The JDK wraps a plain {@code X509KeyManager} in
+ * an adapter that adds endpoint identification checks; extending the "Extended" variant avoids that wrapper.
+ */
+@Singleton
+public class CollectorCaKeyManager extends X509ExtendedKeyManager {
+    private static final Logger LOG = LoggerFactory.getLogger(CollectorCaKeyManager.class);
+    private static final String ALIAS = "server";
+    private static final Set<String> ED25519_KEY_TYPES = Set.of("EdDSA", "Ed25519");
+
+    private final CollectorCaCache caCache;
+
+    @Inject
+    public CollectorCaKeyManager(CollectorCaCache caCache) {
+        this.caCache = caCache;
+    }
+
+    @Override
+    public String chooseServerAlias(String keyType, Principal[] issuers, Socket socket) {
+        if (ED25519_KEY_TYPES.contains(keyType)) {
+            LOG.debug("Returning <{}> as the server alias for key type <{}>", ALIAS, keyType);
+            return ALIAS;
+        }
+        LOG.debug("Returning null for key type <{}>", keyType);
+        return null;
+    }
+
+    @Override
+    public X509Certificate[] getCertificateChain(String alias) {
+        if (ALIAS.equals(alias)) {
+            final var serverEntry = caCache.getServer();
+            final var signingEntry = caCache.getSigning();
+            LOG.debug("Returning certificate chain for alias <{}>: server-cert={} signing-cert={}",
+                    alias, serverEntry.fingerprint(), signingEntry.fingerprint());
+            return new X509Certificate[]{serverEntry.cert(), signingEntry.cert()};
+        }
+        LOG.debug("Returning null certificate chain for alias <{}>", alias);
+        return null;
+    }
+
+    @Override
+    public PrivateKey getPrivateKey(String alias) {
+        if (ALIAS.equals(alias)) {
+            final var serverEntry = caCache.getServer();
+            LOG.debug("Returning private key for server certificate <{}>", serverEntry.fingerprint());
+            return serverEntry.privateKey();
+        }
+        LOG.debug("Returning null private key for alias <{}>", alias);
+        return null;
+    }
+
+    @Override
+    public String chooseEngineServerAlias(String keyType, Principal[] issuers, SSLEngine engine) {
+        return chooseServerAlias(keyType, issuers, null);
+    }
+
+    @Override
+    public String[] getClientAliases(String keyType, Principal[] issuers) {
+        return null;
+    }
+
+    @Override
+    public String chooseClientAlias(String[] keyType, Principal[] issuers, Socket socket) {
+        return null;
+    }
+
+    @Override
+    public String[] getServerAliases(String keyType, Principal[] issuers) {
+        return null;
+    }
+}