test_store.py

#!/usr/bin/env python3
"""
test_store.py - Certificate store for testing mod_dynssl.

Serves certificates from a local ./certs/ directory.

Directory structure:
    certs/
        example.com.crt   <- PEM certificate
        example.com.key   <- PEM private key

Usage:
    # Without auth (open, for local testing):
    python3 test_store.py

    # With Bearer token auth:
    DYNSSL_TOKEN=mysecrettoken python3 test_store.py

    # Run HTTPS store (self-signed cert example):
    openssl req -x509 -newkey rsa:2048 \
        -keyout certs/store.key \
        -out certs/store.crt \
        -days 365 -nodes -subj "/CN=localhost"
    DYNSSL_USE_HTTPS=1 DYNSSL_TLS_CERT=certs/store.crt DYNSSL_TLS_KEY=certs/store.key \
        python3 test_store.py

    # Test with token over HTTPS:
    curl -H "Authorization: Bearer mysecrettoken" \
         --cacert certs/store.crt \
         https://localhost:8888/certs/example.com

    # Test HTTPS:
    curl --cacert certs/store.crt https://localhost:8888/certs/example.com

Generate a test certificate:
    mkdir -p certs
    openssl req -x509 -newkey rsa:2048 \
        -keyout certs/example.com.key \
        -out certs/example.com.crt \
        -days 365 -nodes -subj "/CN=example.com"
"""

import http.server
import json
import os
import ssl
import sys

CERT_DIR   = "./certs"
PORT       = 8888
AUTH_TOKEN = os.environ.get("DYNSSL_TOKEN", "")
USE_HTTPS  = os.environ.get("DYNSSL_USE_HTTPS", "").lower() in ("1", "true", "yes", "on")
TLS_CERT   = os.environ.get("DYNSSL_TLS_CERT", os.path.join(CERT_DIR, "store.crt"))
TLS_KEY    = os.environ.get("DYNSSL_TLS_KEY", os.path.join(CERT_DIR, "store.key"))


class CertStoreHandler(http.server.BaseHTTPRequestHandler):

    def do_GET(self):

        # Bearer token auth -- only enforced if DYNSSL_TOKEN is set
        if AUTH_TOKEN:
            auth_header = self.headers.get("Authorization", "")
            expected    = f"Bearer {AUTH_TOKEN}"
            if auth_header != expected:
                self.send_error(401, "Unauthorized")
                print(f"  [auth] rejected request -- missing or invalid token")
                return

        # Expected path: /certs/{domain}
        parts = self.path.strip("/").split("/")
        if len(parts) < 2 or parts[0] != "certs":
            self.send_error(404, "Not found")
            return

        domain = parts[1]

        # Sanitize domain
        safe_chars = set("abcdefghijklmnopqrstuvwxyz"
                         "ABCDEFGHIJKLMNOPQRSTUVWXYZ"
                         "0123456789.-_")
        if not all(c in safe_chars for c in domain):
            self.send_error(400, "Invalid domain name")
            return

        cert_path = os.path.join(CERT_DIR, f"{domain}.crt")
        key_path  = os.path.join(CERT_DIR, f"{domain}.key")

        if not os.path.exists(cert_path) or not os.path.exists(key_path):
            print(f"  [miss] {domain}")
            self.send_error(404, f"No certificate for: {domain}")
            return

        with open(cert_path, "r") as f:
            cert_pem = f.read()

        with open(key_path, "r") as f:
            key_pem = f.read()

        response = json.dumps({
            "cert": cert_pem,
            "key":  key_pem
        })

        print(f"  [hit]  {domain}")

        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(response)))
        self.end_headers()
        self.wfile.write(response.encode("utf-8"))

    def log_message(self, format, *args):
        pass  # Suppress default access log, we print our own above


if __name__ == "__main__":
    os.makedirs(CERT_DIR, exist_ok=True)
    server = http.server.HTTPServer(("localhost", PORT), CertStoreHandler)
    scheme = "http"
    if USE_HTTPS:
        if not os.path.exists(TLS_CERT):
            print(f"ERROR: TLS cert file not found: {TLS_CERT}")
            sys.exit(1)
        if not os.path.exists(TLS_KEY):
            print(f"ERROR: TLS key file not found: {TLS_KEY}")
            sys.exit(1)

        context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
        context.load_cert_chain(certfile=TLS_CERT, keyfile=TLS_KEY)
        server.socket = context.wrap_socket(server.socket, server_side=True)
        scheme = "https"

    print(f"Certificate store running on {scheme}://localhost:{PORT}")
    print(f"Serving certs from: {os.path.abspath(CERT_DIR)}/")
    if AUTH_TOKEN:
        print("Auth: Bearer token enabled")
    else:
        print("Auth: none (set DYNSSL_TOKEN env var to enable)")
    if USE_HTTPS:
        print(f"TLS cert: {TLS_CERT}")
        print(f"TLS key : {TLS_KEY}")
    else:
        print("TLS: disabled (set DYNSSL_USE_HTTPS=1 to enable)")
    print()

    try:
        server.serve_forever()
    except KeyboardInterrupt:
        print("\nStore stopped.")