Merge pull request #17587 from vbotbuildovich/backport-pr-17539-v23.3…

….x-159

[v23.3.x] net/tls_probe: Introduce "trust_file_crc32c" metric

src/v/net/probes.cc

@@ -8,6 +8,7 @@
 // by the Apache License, Version 2.0
 
 #include "config/configuration.h"
+#include "hashing/crc32.h"
 #include "metrics/metrics.h"
 #include "net/client_probe.h"
 #include "net/server_probe.h"
@@ -340,15 +341,17 @@ ss::future<ss::shared_ptr<T>> build_reloadable_credentials_with_probe(
   ss::sstring detail,
   ss::tls::reload_callback cb) {
     auto probe = ss::make_lw_shared<net::tls_certificate_probe>();
-    auto wrap_cb = [probe, cb = std::move(cb)](
-                     const std::unordered_set<ss::sstring>& updated,
-                     const ss::tls::certificate_credentials& creds,
-                     const std::exception_ptr& eptr) {
-        if (cb) {
-            cb(updated, eptr);
-        }
-        probe->loaded(creds, eptr);
-    };
+    auto wrap_cb =
+      [probe, cb = std::move(cb)](
+        const std::unordered_set<ss::sstring>& updated,
+        const ss::tls::certificate_credentials& creds,
+        const std::exception_ptr& eptr,
+        std::optional<ss::tls::blob> trust_file_contents = std::nullopt) {
+          if (cb) {
+              cb(updated, eptr);
+          }
+          probe->loaded(creds, eptr, trust_file_contents);
+      };
 
     ss::shared_ptr<T> cred;
     if constexpr (std::is_same<T, ss::tls::server_credentials>::value) {
@@ -359,7 +362,7 @@ ss::future<ss::shared_ptr<T>> build_reloadable_credentials_with_probe(
     }
 
     probe->setup_metrics(std::move(area), std::move(detail));
-    probe->loaded(*cred, nullptr);
+    probe->loaded(*cred, nullptr, builder.get_trust_file_blob());
     co_return cred;
 }
 
@@ -378,7 +381,9 @@ build_reloadable_credentials_with_probe(
   ss::tls::reload_callback cb);
 
 void tls_certificate_probe::loaded(
-  const ss::tls::certificate_credentials& creds, std::exception_ptr ex) {
+  const ss::tls::certificate_credentials& creds,
+  std::exception_ptr ex,
+  std::optional<ss::tls::blob> trust_file_contents) {
     _load_time = clock_type::now();
     reset();
 
@@ -418,6 +423,17 @@ void tls_certificate_probe::loaded(
             _ca.emplace(cert{.expiry = exp, .serial = srl});
         }
     }
+
+    _trust_file_crc32c = [&trust_file_contents]() -> uint32_t {
+        if (!trust_file_contents.has_value()) {
+            return 0u;
+        }
+        crc::crc32c hash;
+        hash.extend(
+          trust_file_contents.value().data(),
+          trust_file_contents.value().size());
+        return hash.value();
+    }();
 }
 
 void tls_certificate_probe::setup_metrics(
@@ -482,6 +498,16 @@ void tls_certificate_probe::setup_metrics(
                             "the given truststore, otherwise zero."),
             labels)
             .aggregate(aggregate_labels));
+        defs.emplace_back(
+          sm::make_gauge(
+            "trust_file_crc32c",
+            [this] { return cert_valid() ? _trust_file_crc32c : 0; },
+            sm::description(
+              "crc32c calculated from the contents of the trust "
+              "file if loaded certificate is valid and trust store "
+              "is present, otherwise zero."),
+            labels)
+            .aggregate(aggregate_labels));
         return defs;
     };
 

src/v/net/tls_certificate_probe.h

@@ -39,7 +39,9 @@ class tls_certificate_probe {
     ~tls_certificate_probe() = default;
 
     void loaded(
-      const ss::tls::certificate_credentials& creds, std::exception_ptr ex);
+      const ss::tls::certificate_credentials& creds,
+      std::exception_ptr ex,
+      std::optional<ss::tls::blob> trust_file_contents);
 
     void setup_metrics(std::string_view area, std::string_view detail);
 
@@ -55,6 +57,7 @@ class tls_certificate_probe {
     std::optional<cert> _cert;
     std::optional<cert> _ca;
     bool _cert_loaded{false};
+    uint32_t _trust_file_crc32c;
 
     bool cert_valid() const {
         auto now = clock_type::now();
@@ -66,6 +69,7 @@ class tls_certificate_probe {
 
     void reset() {
         _cert_loaded = false;
+        _trust_file_crc32c = 0;
         _cert.reset();
         _ca.reset();
     }

tests/rptest/tests/tls_metrics_test.py

@@ -12,6 +12,7 @@
 import time
 from datetime import datetime, timedelta
 from typing import Optional, Callable
+import crc32c
 
 from ducktape.cluster.cluster import ClusterNode
 
@@ -70,6 +71,7 @@ class TLSMetricsTestBase(RedpandaTest):
         'loaded_at_timestamp_seconds',
         'certificate_valid',
         'certificate_serial',
+        'trust_file_crc32c',
     ]
 
     EXPECTED_LABELS: list[str] = [
@@ -300,6 +302,39 @@ def test_expiry_reload(self):
         assert status_before['expiry'] + five_days < status_after[
             'expiry'], f"Unexpected status after reload: {json.dumps(status_after)}"
 
+    @cluster(num_nodes=3)
+    def test_crc32c(self):
+        node = self.redpanda.nodes[0]
+
+        def check_crc():
+            metrics_samples = self._get_metrics_from_node(
+                node, ['trust_file_crc32c'])
+            assert metrics_samples is not None, "Failed to get metrics"
+            vals = self._unpack_samples(metrics_samples)['trust_file_crc32c']
+
+            assert len(vals) > 0, "Missing crc metrics for some reason"
+
+            expected = crc32c.crc32c(
+                open(self.security.tls_provider.ca.crt, 'rb').read())
+
+            for v in vals:
+                got = int(v['value'])
+                assert got == expected, f"Expected {expected}; Got {got}"
+
+            return expected
+
+        original = check_crc()
+
+        self.security.tls_provider = FaketimeTLSProvider(
+            tls=tls.TLSCertManager(self.logger, cert_expiry_days=10))
+
+        self.redpanda.set_security_settings(self.security)
+        self.redpanda.write_tls_certs()
+
+        reloaded = check_crc()
+
+        assert original != reloaded, f"Checksums unexpectedly equal"
+
 
 class TLSMetricsTestChain(TLSMetricsTestBase):
     def __init__(self, *args, **kwargs):