Switching to config file monitoring

.github/workflows/docker-publish.yml

@@ -1,37 +1,49 @@
-name: Docker Build and Push to GHCR
+name: Publish Docker image
 
 on:
-  push:
-    branches:
-      - master  # Trigger the workflow on push to the main branch
-  pull_request:
-    branches:
-      - master  # Trigger the workflow on pull requests to the main branch
-  workflow_dispatch:
+  release:
+    types: [published]
 
 jobs:
-  build:
+  push_to_registry:
+    name: Push Docker image to Docker Hub
     runs-on: ubuntu-latest
-
+    permissions:
+      packages: write
+      contents: read
+      attestations: write
+      id-token: write
     steps:
-      - name: Checkout repository
-        uses: actions/checkout@v2
-
-      - name: Set up Docker Buildx
-        uses: docker/setup-buildx-action@v2
+      - name: Check out the repo
+        uses: actions/checkout@v4
 
-      - name: Log in to GitHub Container Registry
-        run: echo "${{ secrets.GHCR_PAT }}" | docker login ghcr.io -u ${{ github.actor }} --password-stdin
+      - name: Log in to Docker Hub
+        uses: docker/login-action@f4ef78c080cd8ba55a85445d5b36e214a81df20a
+        with:
+          username: ${{ secrets.DOCKER_USERNAME }}
+          password: ${{ secrets.DOCKER_PASSWORD }}
 
-      - name: Set up QEMU
-        uses: docker/setup-qemu-action@v2
+      - name: Extract metadata (tags, labels) for Docker
+        id: meta
+        uses: docker/metadata-action@9ec57ed1fcdbf14dcef7dfbe97b2010124a938b7
+        with:
+          images: waschinski/cosmos-cert-extractor
 
       - name: Build and push Docker image
-        uses: docker/build-push-action@v5
+        id: push
+        uses: docker/build-push-action@3b5e8027fcad23fda98b2e3ac259d8d67585f671
         with:
           context: .
+          file: ./Dockerfile
           push: true
-          tags: ghcr.io/lilkidsuave/cosmos-cert-extractor:latest
-
-      - name: Log out from GitHub Container Registry
-        run: docker logout ghcr.io
+          tags: ${{ steps.meta.outputs.tags }}
+          labels: ${{ steps.meta.outputs.labels }}
+
+
+      * name: Generate artifact attestation
+        uses: actions/attest-build-provenance@v1
+        with:
+          subject-name: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME}}
+          subject-digest: ${{ steps.push.outputs.digest }}
+          push-to-registry: true
+

Dockerfile

@@ -1,11 +1,11 @@
 # Use an appropriate base image
 FROM python:3.12-slim
 # Copy the script into the container
-COPY extract.py /extract.py 
+COPY extract.py ./extract.py 
 # Install any necessary dependencies
-RUN pip install pyOpenSSL
-# Set default environment variable
-ENV CHECK_INTERVAL=3600
+RUN pip install watchdog
+# Send stdout and stderr straight to terminal
+ENV PYTHONUNBUFFERED 1
 # Make sure the script is executable (if necessary)
 RUN chmod +x /extract.py
 # Command to run the script

README.md

@@ -1,5 +1,8 @@
 # cosmos-cert-extractor
-This is a python script periodically running (every 30 minutes[configurable]) to extract the TLS certificate from the Cosmos config file. The use case I set this up for is in order to use the certificate in my Adguard Home instance.
+This is a python script monitoring your Cosmos config file for changes in order to extract the TLS certificate from it. The use case I set this up for is in order to use the certificate in my Adguard Home instance.
+> [!NOTE]
+> The script is being triggered on *any* configuration change being done in Cosmos and currently does not verify whether the certificate has actually changed.
+
 ## How to use
 Make sure your volume mounts are set up correctly:
 * The `cosmos` volume or path must be mapped to `/input`.

docker-compose.yaml

@@ -1,9 +1,14 @@
 version: '3.3'
+volumes:
+  cosmos:
+    external: true
+  adguard-config:
+    external: true
 services:
-  cert_checker:
-    image: ghcr.io/lilkidsuave/cosmos-cert-extractor:latest
-    environment:
-      - CHECK_INTERVAL=1800  # Override the check interval to 1800 seconds (30 minutes) Default is 1 hour
+  cosmos-cert-extractor:
+    container_name: cosmos-cert-extractor
+    restart: always
     volumes:
-      - #CosmosDirectory:/input
-      - #AdguardDirectory:/output/certs
+      - cosmos:/input:ro
+      - adguard-config:/output
+    image: 'waschinski/cosmos-cert-extractor:latest'

extract.py

@@ -1,81 +1,64 @@
 #!/usr/bin/env python3
 
-import json
 import os
+import sys
+import json
 import time
-from datetime import datetime, timezone
-from OpenSSL import crypto
+from watchdog.observers import Observer
+from watchdog.events import FileSystemEventHandler
+
+INPUT_PATH = "/input"
+CERTS_PATH = "/output/certs"
+
+class ConfigFileHandler(FileSystemEventHandler):
+    def on_modified(self, event):
+        if event.src_path == INPUT_PATH + "/cosmos.config.json":
+            extract_cert()
 
-CONFIG_PATH = "/input/cosmos.config.json"
-CERT_PATH = "/output/certs/cert.pem"
-KEY_PATH = "/output/certs/key.pem"
-DEFAULT_CHECK_INTERVAL = 3600  # Default check interval (1 hour)
+def extract_cert():
+    config_object = load_config()
+    if config_object:
+        cert = config_object["HTTPConfig"]["TLSCert"]
+        key = config_object["HTTPConfig"]["TLSKey"]
+        write_certificates(cert, key)
+    else:
+        print("Cosmos config file not found.")
+        sys.exit()
 
 def load_config():
     try:
-        with open(CONFIG_PATH, "r") as conf_file:
+        with open(INPUT_PATH + "/cosmos.config.json", "r") as conf_file:
             return json.load(conf_file)
     except OSError:
         return None
 
-def load_certificates():
-    try:
-        with open(CERT_PATH, "r") as cert_file:
-            cert_data = cert_file.read()
-        with open(KEY_PATH, "r") as key_file:
-            key_data = key_file.read()
-        return cert_data, key_data
-    except OSError:
-        return None, None
-
 def write_certificates(cert, key):
-    with open(CERT_PATH, "w") as cert_file:
+    with open(CERTS_PATH + "/cert.pem", "w") as cert_file:
         cert_file.write(cert)
 
-    with open(KEY_PATH, "w") as key_file:
+    with open(CERTS_PATH + "/key.pem", "w") as key_file:
         key_file.write(key)
 
-    print("Cert extracted successfully. Checking again in {check_interval} seconds")
-
-def is_cert_expired(cert_data):
-    cert = crypto.load_certificate(crypto.FILETYPE_PEM, cert_data)
-    expiry_date_str = cert.get_notAfter().decode('ascii')
-    expiry_date = datetime.strptime(expiry_date_str, '%Y%m%d%H%M%SZ').replace(tzinfo=timezone.utc)
-    return expiry_date < datetime.now(timezone.utc)
-
-
-def get_check_interval():
-    try:
-        return int(os.getenv('CHECK_INTERVAL', DEFAULT_CHECK_INTERVAL))
-    except ValueError:
-        print(f"Invalid CHECK_INTERVAL value. Using default: {DEFAULT_CHECK_INTERVAL} seconds.")
-        return DEFAULT_CHECK_INTERVAL
+    print("Cert extracted successfully.")
 
 def main():
-    # Ensure it runs at least once
-    run_once = False
-    check_interval = get_check_interval()
-
-    while True:
-        cert_data, key_data = load_certificates()
-        if not cert_data or not key_data:
-            print(f"Couldn't read the certificate or key file. Checking again in {check_interval} seconds")
-            time.sleep(check_interval)
-            continue
+    if not os.path.isdir(INPUT_PATH):
+        print("Config folder not found.")
+        sys.exit()
+    if not os.path.isdir(CERTS_PATH):
+        print("Certs output folder not found.")
+        sys.exit()
 
-        if not run_once or is_cert_expired(cert_data):
-            config_object = load_config()
-            if config_object:
-                cert = config_object["HTTPConfig"]["TLSCert"]
-                key = config_object["HTTPConfig"]["TLSKey"]
-                write_certificates(cert, key)
-                run_once = True
-            else:
-                print(f"Couldn't read the config file. Checking again in {check_interval} seconds")
-        else:
-            print(f"Certificate is still valid. Checking again in {check_interval} seconds")
-
-        time.sleep(check_interval)
+    observer = Observer()
+    event_handler = ConfigFileHandler()
+    observer.schedule(event_handler, INPUT_PATH, recursive=False)
+    observer.start()
+    try:
+        while True:
+            time.sleep(1)
+    except KeyboardInterrupt:
+        observer.stop()
+    observer.join()
 
 if __name__ == "__main__":
     main()