feat(cert-checker): add Kubernetes resources for certificate checking functionality

- Implemented a new module for cert-checker with Kubernetes resources including ServiceAccount, Role, RoleBinding, ConfigMap, and CronJob. - Added Python script to check certificate expiry and restart deployments if necessary. - Configured environment variables for deployment name, namespace, and service details. - Included requirements for the Python environment. feat(devolo-exporter): introduce Devolo exporter for Prometheus metrics - Created a new module for Devolo exporter with Kubernetes resources including ServiceAccount, ConfigMap, Deployment, and Service. - Developed Python scripts to scrape metrics from Devolo devices and expose them to Prometheus. - Configured environment variables for Devolo device IP, model, and ports. - Added requirements for the Python environment.
2025-10-13 22:35:35 +02:00
parent 84c8288f8a
commit d069fe3585
25 changed files with 1362 additions and 33 deletions
@@ -0,0 +1,129 @@
+import ssl
+import socket
+from datetime import datetime, timezone
+import os
+import logging
+from kubernetes import client, config
+import tempfile
+
+# Setup logging
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s [%(levelname)s] %(message)s"
+)
+
+DEPLOYMENT = os.getenv("DEPLOYMENT_NAME", "my-deployment")
+NAMESPACE = os.getenv("NAMESPACE", "default")
+SECRET_CERT_PATH = os.getenv("SECRET_CERT_PATH", "/certs/tls.crt")
+SERVICE_HOST = os.getenv("SERVICE_HOST", "my-app.default.svc")
+SERVICE_PORT = int(os.getenv("SERVICE_PORT", 443))
+
+def read_secret_cert():
+    logging.info(f"Reading certificate from secret path: {SECRET_CERT_PATH}")
+    return ssl._ssl._test_decode_cert(SECRET_CERT_PATH)
+
+def get_secret_cert_expiry():
+    cert = read_secret_cert()
+    expiry = datetime.strptime(cert['notAfter'], "%b %d %H:%M:%S %Y %Z")
+    logging.info(f"Secret certificate expires at: {expiry}")
+    return expiry
+
+def get_common_name(cert):
+    """
+    Extract Common Name (CN) from a cert dict as returned by ssl.getpeercert()
+    or ssl._ssl._test_decode_cert(). Returns None if not found.
+    """
+    if not cert:
+        return None
+    subject = cert.get('subject', [])
+    for rdn in subject:
+        # rdn is a tuple of tuples, e.g. (('commonName', 'example.com'),)
+        for attr in rdn:
+            if isinstance(attr, tuple) and len(attr) == 2:
+                key, val = attr
+                if key.lower() in ('commonname', 'cn'):
+                    return val
+    return None
+
+def get_live_cert_expiry():
+    logging.info(f"Connecting to service {SERVICE_HOST}:{SERVICE_PORT} to get live certificate")
+    # Try the secure default context first (will verify hostname)
+    ctx = ssl.create_default_context()
+    try:
+        with ctx.wrap_socket(socket.socket(), server_hostname=SERVICE_HOST) as s:
+            s.connect((SERVICE_HOST, SERVICE_PORT))
+            # request both the dict view and the binary DER form
+            cert = s.getpeercert()  # may be empty
+            der = s.getpeercert(binary_form=True)
+    except ssl.SSLCertVerificationError as e:
+        logging.warning("Certificate verification failed (%s). Falling back to non-verifying connection to read cert (insecure).", e)
+        # Insecure fallback: disable hostname and certificate verification just to read the cert
+        ctx = ssl.create_default_context()
+        ctx.check_hostname = False
+        ctx.verify_mode = ssl.CERT_NONE
+        with ctx.wrap_socket(socket.socket(), server_hostname=SERVICE_HOST) as s:
+            s.connect((SERVICE_HOST, SERVICE_PORT))
+            cert = s.getpeercert()
+            der = s.getpeercert(binary_form=True)
+
+    # If the dict view lacks 'notAfter', try decoding the DER to get a full dict
+    if not cert or 'notAfter' not in cert:
+        if der:
+            pem = ssl.DER_cert_to_PEM_cert(der)
+            # write PEM to a temp file and decode using ssl test decoder to get 'notAfter'
+            with tempfile.NamedTemporaryFile(mode="w", delete=True) as tf:
+                tf.write(pem)
+                tf.flush()
+                cert = ssl._ssl._test_decode_cert(tf.name)
+        else:
+            raise RuntimeError("Failed to obtain certificate from peer")
+
+    expiry = datetime.strptime(cert['notAfter'], "%b %d %H:%M:%S %Y %Z")
+    logging.info(f"Live certificate expires at: {expiry}")
+    return expiry, cert
+
+def restart_deployment():
+    logging.info(f"Restarting deployment {DEPLOYMENT} in namespace {NAMESPACE}")
+    config.load_incluster_config()
+    apps = client.AppsV1Api()
+    body = {
+        "spec": {
+            "template": {
+                "metadata": {
+                    "annotations": {
+                        "cert-checker/restartedAt": datetime.now(timezone.utc).isoformat()
+                    }
+                }
+            }
+        }
+    }
+    apps.patch_namespaced_deployment(
+        name=DEPLOYMENT,
+        namespace=NAMESPACE,
+        body=body
+    )
+    logging.info(f"Deployment {DEPLOYMENT} restarted")
+
+if __name__ == "__main__":
+    logging.info("Starting certificate check")
+    secret_cert = read_secret_cert()
+    secret_expiry = datetime.strptime(secret_cert['notAfter'], "%b %d %H:%M:%S %Y %Z")
+    live_expiry, live_cert = get_live_cert_expiry()
+
+    secret_cn = get_common_name(secret_cert)
+    live_cn = get_common_name(live_cert)
+
+    logging.info(f"Secret expiry: {secret_expiry}")
+    logging.info(f"Live cert expiry: {live_expiry}")
+    logging.info(f"Secret CN: {secret_cn}")
+    logging.info(f"Live CN: {live_cn}")
+
+    if secret_cn != live_cn:
+        logging.warning("Certificate mismatch detected (CN), check your setup.")
+        exit(0)
+
+    if secret_expiry != live_expiry:
+        logging.warning("Certificate mismatch detected (expiry), restarting deployment")
+        restart_deployment()
+    else:
+        logging.info("Certificates match, nothing to do.")