From fd76c18eb053c1e62e1bd1baefe4a1b27582249f Mon Sep 17 00:00:00 2001 From: duffyduck Date: Thu, 13 Aug 2026 09:41:26 +0200 Subject: [PATCH] Wildcard-Zertifikate ueber Plesk-DNS und Let's Encrypt Legt den A/AAAA-Record fuer einen Namen in Plesk an bzw. aktualisiert ihn und holt anschliessend per DNS-01-Challenge ein Wildcard-Zertifikat von Let's Encrypt. Alle Bestandteile werden einzeln abgelegt, zusaetzlich als kombinierte bundle.pem und als passwortgeschuetzte cert.pfx. - Plesk-Anbindung ueber die XML-API (Zone finden, Records lesen/anlegen/loeschen) - ACME-Order mit dns-01, Account-Key wird wiederverwendet - Propagations-Check gegen die autoritativen Nameserver der Zone, bricht vor der Validierung ab statt einen Fehlversuch bei Let's Encrypt zu verbrennen - Renewal-Check: laeuft idempotent, taugt so direkt fuer den Cron - Docker-Container, Konfiguration ueber .env, Ausgabe im Projektverzeichnis - Tests ohne echten Plesk-Server bzw. ohne Zertifikatsausstellung Co-Authored-By: Claude Opus 5 (1M context) --- .env.example | 60 +++++++++ .gitignore | 7 + Dockerfile | 23 ++++ README.md | 175 ++++++++++++++++++++++++ app/__init__.py | 3 + app/acme_client.py | 266 +++++++++++++++++++++++++++++++++++++ app/certfiles.py | 293 ++++++++++++++++++++++++++++++++++++++++ app/config.py | 137 +++++++++++++++++++ app/dnsutil.py | 117 ++++++++++++++++ app/main.py | 230 ++++++++++++++++++++++++++++++++ app/plesk.py | 295 +++++++++++++++++++++++++++++++++++++++++ docker-compose.yml | 17 +++ requirements.txt | 5 + run.sh | 32 +++++ tests/test_acme_api.py | 54 ++++++++ tests/test_cli.py | 162 ++++++++++++++++++++++ tests/test_offline.py | 204 ++++++++++++++++++++++++++++ 17 files changed, 2080 insertions(+) create mode 100644 .env.example create mode 100644 .gitignore create mode 100644 Dockerfile create mode 100644 README.md create mode 100644 app/__init__.py create mode 100644 app/acme_client.py create mode 100644 app/certfiles.py create mode 100644 app/config.py create mode 100644 app/dnsutil.py create mode 100644 app/main.py create mode 100644 app/plesk.py create mode 100644 docker-compose.yml create mode 100644 requirements.txt create mode 100755 run.sh create mode 100644 tests/test_acme_api.py create mode 100644 tests/test_cli.py create mode 100644 tests/test_offline.py diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..ed0c47e --- /dev/null +++ b/.env.example @@ -0,0 +1,60 @@ +# --------------------------------------------------------------------------- +# Docker: UID/GID, damit die erzeugten Dateien dir gehoeren (id -u / id -g) +# --------------------------------------------------------------------------- +PUID=1000 +PGID=1000 + +# --------------------------------------------------------------------------- +# Plesk Server +# --------------------------------------------------------------------------- +PLESK_HOST=plesk.example.com +PLESK_PORT=8443 + +# Authentifizierung: entweder API-Key (empfohlen) ODER Login/Passwort. +# API-Key erzeugen (auf dem Plesk-Server): +# plesk bin secret_key --create -ip-address +PLESK_API_KEY= +PLESK_USER=admin +PLESK_PASSWORD= + +# TLS-Zertifikat des Plesk-Panels pruefen (bei self-signed auf false setzen) +PLESK_VERIFY_TLS=false +PLESK_TIMEOUT=60 + +# --------------------------------------------------------------------------- +# ACME / Let's Encrypt +# --------------------------------------------------------------------------- +ACME_EMAIL=admin@example.com +ACME_DIRECTORY_URL=https://acme-v02.api.letsencrypt.org/directory +# true => Staging-Umgebung von Let's Encrypt (zum Testen, keine Rate-Limits) +ACME_STAGING=false +ACME_ACCOUNT_DIR=/app/data + +# --------------------------------------------------------------------------- +# Zertifikat +# --------------------------------------------------------------------------- +# Passwort fuer die .pfx-Datei und den verschluesselten Private Key +CERT_PASSWORD=bitte-aendern +CERT_OUTPUT_DIR=/app/certs + +# rsa oder ec +KEY_TYPE=rsa +RSA_KEY_SIZE=4096 +EC_CURVE=secp256r1 + +# PFX im Legacy-Format (SHA1/3DES) erzeugen - noetig fuer aeltere +# Windows-/Java-Importer. false = modern (AES-256). +PFX_LEGACY_COMPAT=false + +# Erneuerung erst, wenn das vorhandene Zertifikat weniger Tage gueltig ist +RENEW_DAYS_BEFORE_EXPIRY=30 + +# --------------------------------------------------------------------------- +# DNS +# --------------------------------------------------------------------------- +DNS_TTL=300 +# Wartezeit auf DNS-Propagation der _acme-challenge TXT-Records (Sekunden) +DNS_PROPAGATION_TIMEOUT=600 +DNS_PROPAGATION_INTERVAL=15 +# Zusaetzliche Resolver fuer den Propagations-Check (Komma-separiert) +DNS_RESOLVERS=1.1.1.1,8.8.8.8 diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..8e25474 --- /dev/null +++ b/.gitignore @@ -0,0 +1,7 @@ +.env +certs/ +data/ +__pycache__/ +*.pyc +.venv/ +venv/ diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..c7d3808 --- /dev/null +++ b/Dockerfile @@ -0,0 +1,23 @@ +FROM python:3.12-slim + +ENV PYTHONUNBUFFERED=1 \ + PYTHONDONTWRITEBYTECODE=1 \ + CERT_OUTPUT_DIR=/app/certs \ + ACME_ACCOUNT_DIR=/app/data + +WORKDIR /app + +# tzdata, damit die Log-Zeitstempel zur eigenen Zeitzone passen (TZ in .env) +RUN apt-get update \ + && apt-get install -y --no-install-recommends tzdata \ + && rm -rf /var/lib/apt/lists/* + +COPY requirements.txt . +RUN pip install --no-cache-dir -r requirements.txt + +COPY app ./app + +RUN mkdir -p /app/certs /app/data && chmod 777 /app/certs /app/data + +ENTRYPOINT ["python", "-m", "app.main"] +CMD ["--help"] diff --git a/README.md b/README.md new file mode 100644 index 0000000..5bbbde7 --- /dev/null +++ b/README.md @@ -0,0 +1,175 @@ +# Wildcard Let's Encrypt Cert Creator für Plesk + +Legt in Plesk den DNS-Eintrag für einen Namen an (bzw. aktualisiert ihn) und holt +anschließend per **DNS-01-Challenge** ein **Wildcard-Zertifikat** von Let's Encrypt. +Alle Bestandteile des Zertifikats landen als einzelne Dateien im Projektverzeichnis – +zusätzlich als kombinierte `bundle.pem` und als passwortgeschützte `cert.pfx`. + +``` +./run.sh vpn.example.com 192.0.2.10 +``` + +erzeugt ein Zertifikat für `vpn.example.com` **und** `*.vpn.example.com`. + +--- + +## 1. Voraussetzungen + +* Docker (oder lokal Python 3.10+) +* Die Domain muss auf dem Plesk-Server als DNS-Zone liegen und Plesk muss der + **autoritative Nameserver** dafür sein (sonst kann die DNS-Challenge nicht validiert werden). +* Ein Plesk-API-Zugang (API-Key empfohlen). + +### Plesk API-Key erzeugen + +Auf dem Plesk-Server (SSH, als root): + +```bash +plesk bin secret_key --create -ip-address \ + -description "wildcard cert creator" +``` + +Der ausgegebene Key kommt in die `.env` als `PLESK_API_KEY`. +Alternativ geht auch `PLESK_USER=admin` + `PLESK_PASSWORD=...`. + +> Der Port 8443 des Plesk-Panels muss von dem Rechner aus erreichbar sein. + +## 2. Einrichten + +```bash +cp .env.example .env +$EDITOR .env # Plesk-Daten, ACME_EMAIL, CERT_PASSWORD eintragen +chmod +x run.sh +``` + +Wichtig in der `.env`: + +| Variable | Bedeutung | +|---|---| +| `PLESK_HOST` / `PLESK_PORT` | Plesk-Panel (Standard-Port 8443) | +| `PLESK_API_KEY` | API-Key (oder `PLESK_USER`/`PLESK_PASSWORD`) | +| `PLESK_VERIFY_TLS` | bei selbstsigniertem Panel-Zertifikat `false` | +| `ACME_EMAIL` | Kontaktadresse für Let's Encrypt | +| `ACME_STAGING` | `true` zum Testen (kein Rate-Limit-Risiko) | +| `CERT_PASSWORD` | Passwort der `.pfx` und des verschlüsselten Keys | +| `KEY_TYPE` | `rsa` (Standard, 4096 Bit) oder `ec` | +| `PFX_LEGACY_COMPAT` | `true` für alte Windows-/Java-Importer (SHA1/3DES) | + +## 3. Benutzen + +```bash +# DNS-Eintrag anlegen/aktualisieren + Wildcard-Zertifikat holen +./run.sh vpn.example.com 192.0.2.10 + +# erstmal testen (Staging-CA von Let's Encrypt) +./run.sh vpn.example.com 192.0.2.10 --staging + +# nur den DNS-Eintrag setzen, kein Zertifikat +./run.sh vpn.example.com 192.0.2.10 --dns-only + +# Zertifikat erneuern, obwohl das alte noch gültig ist +./run.sh vpn.example.com 192.0.2.10 --force + +# ohne Wildcard, dafür mit weiteren Namen +./run.sh vpn.example.com 192.0.2.10 --no-wildcard --san www.example.com +``` + +Ohne Docker: + +```bash +python3 -m venv .venv && . .venv/bin/activate +pip install -r requirements.txt +python -m app.main vpn.example.com 192.0.2.10 +``` + +### Alle Optionen + +| Option | Wirkung | +|---|---| +| `--ip ` | Alternative zum zweiten Positionsargument (IPv4 → A, IPv6 → AAAA) | +| `--san ` | zusätzlicher Name im Zertifikat (mehrfach möglich) | +| `--no-wildcard` | nur der reine Name, ohne `*.` | +| `--skip-dns` | A/AAAA-Record unangetastet lassen | +| `--dns-only` | nur DNS, kein Zertifikat | +| `--staging` | Let's-Encrypt-Staging-Umgebung | +| `--force` | erneuern, auch wenn noch gültig | +| `--keep-txt` | `_acme-challenge`-Records stehen lassen (Debugging) | +| `--ignore-propagation-timeout` | trotz fehlender DNS-Propagation weitermachen | +| `--env ` | andere `.env` verwenden | +| `-v` | ausführliche Ausgabe (inkl. XML-Requests) | + +## 4. Was wird erzeugt + +Alles unter `certs//`: + +| Datei | Inhalt | +|---|---| +| `privkey.pem` | Private Key, PKCS#8, unverschlüsselt | +| `privkey-traditional.pem` | derselbe Key im klassischen OpenSSL-Format | +| `privkey-encrypted.pem` | Key, verschlüsselt mit `CERT_PASSWORD` | +| `pubkey.pem` | öffentlicher Schlüssel | +| `csr.pem` | der verwendete Certificate Signing Request | +| `cert.pem` / `cert.crt` | das reine Zertifikat (Leaf) | +| `cert.der` | Leaf im DER-Format (für Windows) | +| `chain.pem` | nur die Zwischenzertifikate | +| `chain-01.pem`, … | jedes Zwischenzertifikat einzeln | +| `fullchain.pem` | Leaf + Kette (das, was Webserver meist wollen) | +| `bundle.pem` | Key + Leaf + Kette in einer Datei | +| `cert.pfx` | PKCS#12 mit Key + Kette, Passwort = `CERT_PASSWORD` | +| `cert-info.txt` / `cert-info.json` | Subject, SANs, Gültigkeit, Fingerprints | + +Der ACME-Account-Key liegt in `data/` – **nicht löschen**, sonst wird bei jedem Lauf +ein neuer Let's-Encrypt-Account registriert. + +Schnellprüfung: + +```bash +openssl x509 -in certs/vpn.example.com/cert.pem -noout -text +openssl pkcs12 -info -in certs/vpn.example.com/cert.pfx -nodes -passin pass: +``` + +## 5. Wie es abläuft + +1. Plesk-Zone zum Namen suchen (längste passende Zone, z. B. `example.com` für `vpn.example.com`). +2. A/AAAA-Record anlegen oder auf die neue IP korrigieren. +3. Prüfen, ob überhaupt erneuert werden muss (`RENEW_DAYS_BEFORE_EXPIRY`, Standard 30 Tage). +4. Key + CSR für `` und `*.` erzeugen. +5. ACME-Order bei Let's Encrypt, für jede Autorisierung einen `_acme-challenge`-TXT-Record + in Plesk setzen (alte Reste werden vorher entfernt). +6. Warten, bis **alle autoritativen Nameserver** die TXT-Records ausliefern + (`DNS_PROPAGATION_TIMEOUT`). +7. Challenges beantworten, Zertifikat abholen, TXT-Records wieder löschen. +8. Alle Dateien schreiben und die Zertifikatsdaten ausgeben. + +## 6. Automatische Erneuerung + +Das Skript ist idempotent: Läuft es, obwohl das Zertifikat noch länger als +`RENEW_DAYS_BEFORE_EXPIRY` Tage gültig ist, passiert nichts. Also einfach per Cron: + +```cron +17 3 * * * cd /home/duffy/Dokumente/programmierung/wildcard-lets-encrypt-cert-plesk-creator && ./run.sh vpn.example.com 192.0.2.10 >> renew.log 2>&1 +``` + +## 7. Wenn etwas klemmt + +| Symptom | Ursache / Lösung | +|---|---| +| `Plesk rejected the credentials (HTTP 401)` | API-Key falsch oder nicht für diese Quell-IP erzeugt | +| `No Plesk DNS zone found for …` | Domain liegt nicht auf diesem Plesk oder der API-User sieht sie nicht | +| `DNS propagation timed out` | Plesk ist nicht der autoritative NS, oder die Zone ist deaktiviert. Mit `-v` prüfen, welcher Nameserver antwortet | +| `no dns-01 challenge` | Wildcards gehen ausschließlich über DNS-01 – Domain-Validierung per HTTP ist nicht möglich | +| `urn:ietf:params:acme:error:rateLimited` | Rate-Limit von Let's Encrypt (5 Zertifikate pro Woche pro Domain). Erst mit `--staging` testen | +| PFX lässt sich in Windows nicht importieren | `PFX_LEGACY_COMPAT=true` setzen und neu erzeugen | + +Bei `-v` werden auch die XML-Requests an Plesk geloggt – hilfreich, wenn ein +Record nicht so landet wie erwartet. + +## 8. Tests + +Ohne echten Plesk-Server und ohne Zertifikatsausstellung: + +```bash +.venv/bin/python tests/test_offline.py # XML-Parsing, DNS-Logik, Dateiausgabe, PFX +.venv/bin/python tests/test_cli.py # kompletter Ablauf gegen ein Plesk-Fake +.venv/bin/python tests/test_acme_api.py # prueft die Let's-Encrypt-Anbindung (nur Directory-Abruf) +``` diff --git a/app/__init__.py b/app/__init__.py new file mode 100644 index 0000000..380badf --- /dev/null +++ b/app/__init__.py @@ -0,0 +1,3 @@ +"""Wildcard Let's Encrypt certificate creator for Plesk-managed DNS zones.""" + +__version__ = "1.0.0" diff --git a/app/acme_client.py b/app/acme_client.py new file mode 100644 index 0000000..052b16b --- /dev/null +++ b/app/acme_client.py @@ -0,0 +1,266 @@ +"""ACME (Let's Encrypt) client: account handling, dns-01 orders, certificate issuance.""" + +from __future__ import annotations + +import datetime +import json +import logging +import re +from collections import defaultdict +from pathlib import Path +from typing import Protocol +from urllib.parse import urlparse + +import josepy as jose +from acme import challenges, client, errors, messages +from cryptography import x509 +from cryptography.hazmat.primitives import hashes, serialization +from cryptography.hazmat.primitives.asymmetric import ec, rsa +from cryptography.x509.oid import NameOID + +from .config import Config + +log = logging.getLogger(__name__) + +USER_AGENT = "wildcard-lets-encrypt-cert-plesk-creator/1.0" +ACCOUNT_KEY_BITS = 2048 + +_CURVES = { + "secp256r1": ec.SECP256R1, + "prime256v1": ec.SECP256R1, + "p-256": ec.SECP256R1, + "secp384r1": ec.SECP384R1, + "p-384": ec.SECP384R1, +} + + +class AcmeFailure(Exception): + """Raised when the certificate could not be issued.""" + + +class DnsSolver(Protocol): + """Everything the ACME flow needs from the DNS side.""" + + def add_txt(self, name: str, value: str) -> None: ... + + def wait_for_propagation(self, expected: dict[str, set[str]]) -> None: ... + + def cleanup(self) -> None: ... + + +# -------------------------------------------------------------------------- +# keys & CSR +# -------------------------------------------------------------------------- + + +def generate_private_key(cfg: Config): + if cfg.key_type == "ec": + curve = _CURVES.get(cfg.ec_curve) + if curve is None: + raise AcmeFailure(f"Unsupported EC_CURVE {cfg.ec_curve!r} (use secp256r1 or secp384r1).") + log.info("Generating EC private key (%s)", cfg.ec_curve) + return ec.generate_private_key(curve()) + log.info("Generating RSA private key (%d bit)", cfg.rsa_key_size) + return rsa.generate_private_key(public_exponent=65537, key_size=cfg.rsa_key_size) + + +def build_csr(private_key, domains: list[str]) -> bytes: + """PEM-encoded CSR with *domains* as SANs (first entry also becomes the CN).""" + common_name = next((d for d in domains if not d.startswith("*.")), domains[0]) + builder = ( + x509.CertificateSigningRequestBuilder() + .subject_name(x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, common_name)])) + .add_extension( + x509.SubjectAlternativeName([x509.DNSName(d) for d in domains]), + critical=False, + ) + ) + csr = builder.sign(private_key, hashes.SHA256()) + return csr.public_bytes(serialization.Encoding.PEM) + + +# -------------------------------------------------------------------------- +# ACME +# -------------------------------------------------------------------------- + + +class AcmeManager: + def __init__(self, cfg: Config): + self.cfg = cfg + host = urlparse(cfg.acme_directory_url).hostname or "acme" + self.account_dir = cfg.acme_account_dir / re.sub(r"[^a-z0-9.-]", "_", host.lower()) + self.account_dir.mkdir(parents=True, exist_ok=True) + self.account_key_file = self.account_dir / "account_key.json" + self.account_regr_file = self.account_dir / "account.json" + self.account_key = self._load_or_create_account_key() + self.client = self._connect() + + # -- account ----------------------------------------------------------- + + def _load_or_create_account_key(self) -> jose.JWKRSA: + if self.account_key_file.is_file(): + log.info("Using existing ACME account key %s", self.account_key_file) + return jose.JWKRSA.json_loads(self.account_key_file.read_text()) + + log.info("Creating new ACME account key in %s", self.account_key_file) + key = jose.JWKRSA( + key=rsa.generate_private_key(public_exponent=65537, key_size=ACCOUNT_KEY_BITS) + ) + self.account_key_file.write_text(key.json_dumps_pretty()) + self.account_key_file.chmod(0o600) + return key + + def _connect(self) -> client.ClientV2: + net = client.ClientNetwork(self.account_key, user_agent=USER_AGENT) + url = self.cfg.acme_directory_url + log.info("Connecting to ACME directory %s", url) + try: + if hasattr(client.ClientV2, "get_directory"): + directory = client.ClientV2.get_directory(url, net) + else: # pragma: no cover - older acme releases + directory = messages.Directory.from_json(net.get(url).json()) + except Exception as exc: # noqa: BLE001 - network/protocol errors alike + raise AcmeFailure(f"Cannot read the ACME directory at {url}: {exc}") from exc + + acme_client = client.ClientV2(directory, net=net) + self._register(acme_client) + return acme_client + + def _register(self, acme_client: client.ClientV2) -> None: + if self.account_regr_file.is_file(): + try: + regr = messages.RegistrationResource.from_json( + json.loads(self.account_regr_file.read_text()) + ) + acme_client.net.account = regr + log.info("Reusing ACME account %s", regr.uri) + return + except Exception as exc: # noqa: BLE001 - fall back to registration + log.warning("Stored ACME account is unusable (%s), registering again.", exc) + + log.info("Registering ACME account for %s", self.cfg.acme_email) + try: + regr = acme_client.new_account( + messages.NewRegistration.from_data( + email=self.cfg.acme_email, terms_of_service_agreed=True + ) + ) + except errors.Error as exc: + raise AcmeFailure(f"ACME account registration failed: {exc}") from exc + + self.account_regr_file.write_text(json.dumps(regr.to_json(), indent=2)) + self.account_regr_file.chmod(0o600) + log.info("ACME account registered: %s", regr.uri) + + # -- issuance ---------------------------------------------------------- + + def obtain_certificate(self, domains: list[str], csr_pem: bytes, solver: DnsSolver) -> str: + log.info("Requesting certificate for: %s", ", ".join(domains)) + try: + order = self.client.new_order(csr_pem) + except errors.Error as exc: + raise AcmeFailure(f"ACME order could not be created: {exc}") from exc + + pending: list[tuple[messages.ChallengeBody, str, str]] = [] + expected: dict[str, set[str]] = defaultdict(set) + + for authz in order.authorizations: + identifier = authz.body.identifier.value + if authz.body.status == messages.STATUS_VALID: + log.info("Authorization for %s is still valid - no challenge needed.", identifier) + continue + challb = self._dns_challenge(authz, identifier) + validation = challb.chall.validation(self.account_key) + record_name = challb.chall.validation_domain_name(identifier) + expected[record_name].add(validation) + pending.append((challb, record_name, validation)) + + if not pending: + log.info("All authorizations are already valid.") + try: + for record_name, validation in ((n, v) for n, vs in expected.items() for v in vs): + solver.add_txt(record_name, validation) + + if expected: + solver.wait_for_propagation(dict(expected)) + + for challb, record_name, _validation in pending: + log.info("Answering dns-01 challenge for %s", record_name) + self.client.answer_challenge(challb, challb.chall.response(self.account_key)) + + deadline = datetime.datetime.now() + datetime.timedelta(seconds=300) + try: + order = self.client.poll_and_finalize(order, deadline) + except errors.ValidationError as exc: + raise AcmeFailure(self._validation_error_text(exc)) from exc + except errors.TimeoutError as exc: + raise AcmeFailure( + "Timed out waiting for Let's Encrypt to validate/issue the certificate." + ) from exc + except errors.Error as exc: + raise AcmeFailure(f"ACME finalization failed: {exc}") from exc + finally: + solver.cleanup() + + if not order.fullchain_pem: + raise AcmeFailure("Let's Encrypt returned an empty certificate chain.") + log.info("Certificate issued.") + return order.fullchain_pem + + @staticmethod + def _dns_challenge(authz, identifier: str) -> messages.ChallengeBody: + for challb in authz.body.challenges: + if isinstance(challb.chall, challenges.DNS01): + return challb + raise AcmeFailure( + f"Let's Encrypt offered no dns-01 challenge for {identifier} - " + "a wildcard certificate cannot be issued without it." + ) + + @staticmethod + def _validation_error_text(exc: errors.ValidationError) -> str: + lines = ["Let's Encrypt could not validate the DNS challenge:"] + for authz in exc.failed_authzrs: + domain = authz.body.identifier.value + for challb in authz.body.challenges: + if challb.error is not None: + lines.append(f" - {domain}: {challb.error.detail or challb.error}") + break + else: + lines.append(f" - {domain}: status {authz.body.status}") + lines.append( + "Check that the _acme-challenge TXT records are served by the authoritative " + "nameservers of the zone (is Plesk the DNS master for this domain?)." + ) + return "\n".join(lines) + + +def certificate_domains(fqdn: str, wildcard: bool, extra_sans: list[str]) -> list[str]: + """Build the SAN list: base name, *.base name, plus anything extra.""" + domains = [fqdn] + if wildcard: + domains.append(f"*.{fqdn}") + for san in extra_sans: + san = san.strip().lower() + if san and san not in domains: + domains.append(san) + return domains + + +def csr_for(cfg: Config, domains: list[str]): + key = generate_private_key(cfg) + csr = build_csr(key, domains) + log.debug("CSR built for %s", domains) + return key, csr + + +def parse_chain(fullchain_pem: str) -> list[x509.Certificate]: + """Split a PEM chain into certificate objects (leaf first).""" + pem_blocks = re.findall( + r"-----BEGIN CERTIFICATE-----.*?-----END CERTIFICATE-----", + fullchain_pem, + re.DOTALL, + ) + if not pem_blocks: + raise AcmeFailure("No certificates found in the ACME response.") + return [x509.load_pem_x509_certificate(block.encode()) for block in pem_blocks] diff --git a/app/certfiles.py b/app/certfiles.py new file mode 100644 index 0000000..b1744c8 --- /dev/null +++ b/app/certfiles.py @@ -0,0 +1,293 @@ +"""Write out every part of the issued certificate: key, cert, chain, PEM bundle, PFX, info.""" + +from __future__ import annotations + +import datetime +import json +import logging +from pathlib import Path + +from cryptography import x509 +from cryptography.hazmat.primitives import hashes, serialization +from cryptography.hazmat.primitives.asymmetric import ec, rsa +from cryptography.hazmat.primitives.serialization import pkcs12 +from cryptography.x509.oid import NameOID + +from .config import Config + +log = logging.getLogger(__name__) + + +# -------------------------------------------------------------------------- +# helpers +# -------------------------------------------------------------------------- + + +def not_before(cert: x509.Certificate) -> datetime.datetime: + try: + return cert.not_valid_before_utc + except AttributeError: # pragma: no cover - cryptography < 42 + return cert.not_valid_before.replace(tzinfo=datetime.timezone.utc) + + +def not_after(cert: x509.Certificate) -> datetime.datetime: + try: + return cert.not_valid_after_utc + except AttributeError: # pragma: no cover - cryptography < 42 + return cert.not_valid_after.replace(tzinfo=datetime.timezone.utc) + + +def san_list(cert: x509.Certificate) -> list[str]: + try: + ext = cert.extensions.get_extension_for_class(x509.SubjectAlternativeName) + except x509.ExtensionNotFound: + return [] + return list(ext.value.get_values_for_type(x509.DNSName)) + + +def days_remaining(cert: x509.Certificate) -> int: + delta = not_after(cert) - datetime.datetime.now(datetime.timezone.utc) + return delta.days + + +def _write(path: Path, data: bytes | str, secret: bool = False) -> Path: + mode = "wb" if isinstance(data, bytes) else "w" + with open(path, mode) as fh: + fh.write(data) + path.chmod(0o600 if secret else 0o644) + return path + + +def _key_description(key) -> dict[str, object]: + if isinstance(key, rsa.RSAPrivateKey) or isinstance(key, rsa.RSAPublicKey): + return {"algorithm": "RSA", "size_bits": key.key_size} + if isinstance(key, (ec.EllipticCurvePrivateKey, ec.EllipticCurvePublicKey)): + return {"algorithm": "EC", "curve": key.curve.name, "size_bits": key.key_size} + return {"algorithm": type(key).__name__} + + +# -------------------------------------------------------------------------- +# output +# -------------------------------------------------------------------------- + + +def write_certificate_files( + cfg: Config, + fqdn: str, + private_key, + csr_pem: bytes, + chain: list[x509.Certificate], + domains: list[str], +) -> dict[str, Path]: + """Write all certificate artefacts for *fqdn* and return {label: path}.""" + out_dir = cfg.cert_output_dir / fqdn + out_dir.mkdir(parents=True, exist_ok=True) + + leaf, intermediates = chain[0], chain[1:] + password = cfg.cert_password.encode() + paths: dict[str, Path] = {} + + # -- private key, in the usual flavours -------------------------------- + paths["private key (PKCS#8)"] = _write( + out_dir / "privkey.pem", + private_key.private_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PrivateFormat.PKCS8, + encryption_algorithm=serialization.NoEncryption(), + ), + secret=True, + ) + paths["private key (traditional)"] = _write( + out_dir / "privkey-traditional.pem", + private_key.private_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PrivateFormat.TraditionalOpenSSL, + encryption_algorithm=serialization.NoEncryption(), + ), + secret=True, + ) + paths["private key (encrypted)"] = _write( + out_dir / "privkey-encrypted.pem", + private_key.private_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PrivateFormat.PKCS8, + encryption_algorithm=serialization.BestAvailableEncryption(password), + ), + secret=True, + ) + paths["public key"] = _write( + out_dir / "pubkey.pem", + private_key.public_key().public_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PublicFormat.SubjectPublicKeyInfo, + ), + ) + + # -- CSR --------------------------------------------------------------- + paths["CSR"] = _write(out_dir / "csr.pem", csr_pem) + + # -- certificate parts ------------------------------------------------- + leaf_pem = leaf.public_bytes(serialization.Encoding.PEM) + chain_pem = b"".join(c.public_bytes(serialization.Encoding.PEM) for c in intermediates) + fullchain_pem = leaf_pem + chain_pem + + paths["certificate (leaf)"] = _write(out_dir / "cert.pem", leaf_pem) + paths["certificate (.crt copy)"] = _write(out_dir / "cert.crt", leaf_pem) + paths["certificate (DER/.cer)"] = _write( + out_dir / "cert.der", leaf.public_bytes(serialization.Encoding.DER) + ) + paths["CA chain"] = _write(out_dir / "chain.pem", chain_pem) + paths["fullchain"] = _write(out_dir / "fullchain.pem", fullchain_pem) + + for index, ca in enumerate(intermediates, start=1): + cn = _common_name(ca) or f"ca-{index}" + paths[f"chain part {index} ({cn})"] = _write( + out_dir / f"chain-{index:02d}.pem", ca.public_bytes(serialization.Encoding.PEM) + ) + + # -- combined PEM (key + certificate + chain) -------------------------- + bundle = ( + private_key.private_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PrivateFormat.PKCS8, + encryption_algorithm=serialization.NoEncryption(), + ) + + fullchain_pem + ) + paths["PEM bundle (key + fullchain)"] = _write(out_dir / "bundle.pem", bundle, secret=True) + + # -- PKCS#12 / PFX ----------------------------------------------------- + paths["PKCS#12 (PFX)"] = _write( + out_dir / "cert.pfx", + _build_pfx(cfg, fqdn, private_key, leaf, intermediates, password), + secret=True, + ) + + # -- human/machine readable summary ------------------------------------ + info = certificate_info(chain, private_key, domains) + paths["info (JSON)"] = _write(out_dir / "cert-info.json", json.dumps(info, indent=2) + "\n") + paths["info (text)"] = _write(out_dir / "cert-info.txt", format_info(info) + "\n") + + return paths + + +def _build_pfx(cfg: Config, fqdn: str, private_key, leaf, intermediates, password: bytes) -> bytes: + if cfg.pfx_legacy_compat: + try: + encryption = ( + serialization.PrivateFormat.PKCS12.encryption_builder() + .key_cert_algorithm(pkcs12.PBES.PBESv1SHA1And3KeyTripleDESCBC) + .hmac_hash(hashes.SHA1()) + .build(password) + ) + log.info("Building PFX in legacy format (SHA1 / 3DES)") + except Exception as exc: # noqa: BLE001 - depends on the OpenSSL build + log.warning("Legacy PFX encryption unavailable (%s), falling back to AES-256.", exc) + encryption = serialization.BestAvailableEncryption(password) + else: + encryption = serialization.BestAvailableEncryption(password) + + return pkcs12.serialize_key_and_certificates( + name=fqdn.encode(), + key=private_key, + cert=leaf, + cas=intermediates or None, + encryption_algorithm=encryption, + ) + + +# -------------------------------------------------------------------------- +# certificate information +# -------------------------------------------------------------------------- + + +def _common_name(cert: x509.Certificate) -> str: + try: + return cert.subject.get_attributes_for_oid(NameOID.COMMON_NAME)[0].value + except (IndexError, ValueError): + return "" + + +def certificate_info(chain: list[x509.Certificate], private_key, domains: list[str]) -> dict: + leaf = chain[0] + return { + "requested_domains": domains, + "key": _key_description(private_key), + "leaf": _cert_info(leaf), + "chain": [_cert_info(c) for c in chain[1:]], + "generated_at": datetime.datetime.now(datetime.timezone.utc).isoformat(timespec="seconds"), + } + + +def _cert_info(cert: x509.Certificate) -> dict: + return { + "subject": cert.subject.rfc4514_string(), + "common_name": _common_name(cert), + "issuer": cert.issuer.rfc4514_string(), + "serial_number": format(cert.serial_number, "x"), + "not_before": not_before(cert).isoformat(timespec="seconds"), + "not_after": not_after(cert).isoformat(timespec="seconds"), + "days_remaining": days_remaining(cert), + "subject_alternative_names": san_list(cert), + "signature_algorithm": getattr( + cert.signature_algorithm_oid, "_name", cert.signature_algorithm_oid.dotted_string + ), + "public_key": _key_description(cert.public_key()), + "fingerprint_sha256": cert.fingerprint(hashes.SHA256()).hex(":"), + "fingerprint_sha1": cert.fingerprint(hashes.SHA1()).hex(":"), + } + + +def format_info(info: dict) -> str: + lines = ["Certificate details", "=" * 60] + leaf = info["leaf"] + lines += [ + f"Common name : {leaf['common_name']}", + f"SANs : {', '.join(leaf['subject_alternative_names'])}", + f"Issuer : {leaf['issuer']}", + f"Serial : {leaf['serial_number']}", + f"Valid from : {leaf['not_before']}", + f"Valid until : {leaf['not_after']} ({leaf['days_remaining']} days)", + f"Key : {info['key'].get('algorithm')} " + f"{info['key'].get('size_bits', '')} {info['key'].get('curve', '')}".rstrip(), + f"Signature : {leaf['signature_algorithm']}", + f"SHA-256 : {leaf['fingerprint_sha256']}", + f"SHA-1 : {leaf['fingerprint_sha1']}", + ] + if info["chain"]: + lines += ["", "Chain:"] + for index, ca in enumerate(info["chain"], start=1): + lines.append(f" {index}. {ca['common_name'] or ca['subject']} (until {ca['not_after']})") + return "\n".join(lines) + + +# -------------------------------------------------------------------------- +# renewal check +# -------------------------------------------------------------------------- + + +def load_existing_certificate(cfg: Config, fqdn: str) -> x509.Certificate | None: + path = cfg.cert_output_dir / fqdn / "cert.pem" + if not path.is_file(): + return None + try: + return x509.load_pem_x509_certificate(path.read_bytes()) + except ValueError as exc: + log.warning("Existing certificate %s is unreadable (%s) - issuing a new one.", path, exc) + return None + + +def renewal_needed(cfg: Config, fqdn: str, domains: list[str]) -> tuple[bool, str]: + cert = load_existing_certificate(cfg, fqdn) + if cert is None: + return True, "no certificate found yet" + + have = {d.lower() for d in san_list(cert)} + want = {d.lower() for d in domains} + if not want.issubset(have): + return True, f"missing names in existing certificate: {', '.join(sorted(want - have))}" + + left = days_remaining(cert) + if left <= cfg.renew_days_before_expiry: + return True, f"expires in {left} days" + return False, f"still valid for {left} days" diff --git a/app/config.py b/app/config.py new file mode 100644 index 0000000..f77803a --- /dev/null +++ b/app/config.py @@ -0,0 +1,137 @@ +"""Configuration loading from environment / .env file.""" + +from __future__ import annotations + +import os +from dataclasses import dataclass, field +from pathlib import Path + +from dotenv import load_dotenv + +STAGING_DIRECTORY_URL = "https://acme-staging-v02.api.letsencrypt.org/directory" +PRODUCTION_DIRECTORY_URL = "https://acme-v02.api.letsencrypt.org/directory" + + +def _bool(name: str, default: bool = False) -> bool: + raw = os.getenv(name) + if raw is None or raw.strip() == "": + return default + return raw.strip().lower() in ("1", "true", "yes", "on", "y") + + +def _int(name: str, default: int) -> int: + raw = os.getenv(name) + if raw is None or raw.strip() == "": + return default + try: + return int(raw.strip()) + except ValueError as exc: + raise ConfigError(f"{name} must be an integer, got {raw!r}") from exc + + +def _str(name: str, default: str = "") -> str: + raw = os.getenv(name) + return default if raw is None else raw.strip() + + +class ConfigError(Exception): + """Raised when the configuration is incomplete or contradictory.""" + + +@dataclass +class Config: + # Plesk + plesk_host: str + plesk_port: int + plesk_api_key: str + plesk_user: str + plesk_password: str + plesk_verify_tls: bool + plesk_timeout: int + + # ACME + acme_email: str + acme_directory_url: str + acme_account_dir: Path + + # Certificate + cert_password: str + cert_output_dir: Path + key_type: str + rsa_key_size: int + ec_curve: str + pfx_legacy_compat: bool + renew_days_before_expiry: int + + # DNS + dns_ttl: int + dns_propagation_timeout: int + dns_propagation_interval: int + dns_resolvers: list[str] = field(default_factory=list) + + @property + def plesk_url(self) -> str: + return f"https://{self.plesk_host}:{self.plesk_port}/enterprise/control/agent.php" + + def validate(self) -> None: + if not self.plesk_host: + raise ConfigError("PLESK_HOST is not set (see .env.example).") + if not self.plesk_api_key and not (self.plesk_user and self.plesk_password): + raise ConfigError( + "Plesk credentials missing: set PLESK_API_KEY or PLESK_USER + PLESK_PASSWORD." + ) + if not self.acme_email: + raise ConfigError("ACME_EMAIL is not set - Let's Encrypt requires a contact address.") + if not self.cert_password: + raise ConfigError("CERT_PASSWORD is not set - it is required for the .pfx file.") + if self.key_type not in ("rsa", "ec"): + raise ConfigError(f"KEY_TYPE must be 'rsa' or 'ec', got {self.key_type!r}.") + + +def load_config(env_file: str | os.PathLike[str] | None = None, staging: bool | None = None) -> Config: + """Load the configuration from the environment, optionally seeded by a .env file.""" + if env_file is not None: + load_dotenv(env_file, override=False) + else: + # Search upwards from the project root; harmless if no .env exists. + default_env = Path(__file__).resolve().parent.parent / ".env" + if default_env.is_file(): + load_dotenv(default_env, override=False) + else: + load_dotenv(override=False) + + use_staging = _bool("ACME_STAGING", False) if staging is None else staging + directory_url = _str("ACME_DIRECTORY_URL", PRODUCTION_DIRECTORY_URL) or PRODUCTION_DIRECTORY_URL + if use_staging: + directory_url = STAGING_DIRECTORY_URL + + resolvers = [r.strip() for r in _str("DNS_RESOLVERS", "1.1.1.1,8.8.8.8").split(",") if r.strip()] + + cfg = Config( + plesk_host=_str("PLESK_HOST"), + plesk_port=_int("PLESK_PORT", 8443), + plesk_api_key=_str("PLESK_API_KEY"), + plesk_user=_str("PLESK_USER"), + plesk_password=os.getenv("PLESK_PASSWORD", ""), + plesk_verify_tls=_bool("PLESK_VERIFY_TLS", False), + plesk_timeout=_int("PLESK_TIMEOUT", 60), + acme_email=_str("ACME_EMAIL"), + acme_directory_url=directory_url, + acme_account_dir=Path(_str("ACME_ACCOUNT_DIR", "./data") or "./data"), + cert_password=os.getenv("CERT_PASSWORD", ""), + cert_output_dir=Path(_str("CERT_OUTPUT_DIR", "./certs") or "./certs"), + key_type=_str("KEY_TYPE", "rsa").lower() or "rsa", + rsa_key_size=_int("RSA_KEY_SIZE", 4096), + ec_curve=_str("EC_CURVE", "secp256r1").lower() or "secp256r1", + pfx_legacy_compat=_bool("PFX_LEGACY_COMPAT", False), + renew_days_before_expiry=_int("RENEW_DAYS_BEFORE_EXPIRY", 30), + dns_ttl=_int("DNS_TTL", 300), + dns_propagation_timeout=_int("DNS_PROPAGATION_TIMEOUT", 600), + dns_propagation_interval=_int("DNS_PROPAGATION_INTERVAL", 15), + dns_resolvers=resolvers, + ) + return cfg + + +def is_staging(cfg: Config) -> bool: + return cfg.acme_directory_url == STAGING_DIRECTORY_URL diff --git a/app/dnsutil.py b/app/dnsutil.py new file mode 100644 index 0000000..9bba3cf --- /dev/null +++ b/app/dnsutil.py @@ -0,0 +1,117 @@ +"""DNS propagation checks for the ACME dns-01 challenge.""" + +from __future__ import annotations + +import logging +import time + +import dns.exception +import dns.flags +import dns.message +import dns.query +import dns.rdatatype +import dns.resolver + +log = logging.getLogger(__name__) + +QUERY_TIMEOUT = 5.0 + + +def _resolver(nameservers: list[str] | None = None) -> dns.resolver.Resolver: + res = dns.resolver.Resolver(configure=not nameservers) + if nameservers: + res.nameservers = nameservers + res.lifetime = QUERY_TIMEOUT * 2 + res.timeout = QUERY_TIMEOUT + return res + + +def authoritative_servers(zone: str, fallback_resolvers: list[str]) -> list[str]: + """IP addresses of the authoritative nameservers of *zone*.""" + ips: list[str] = [] + names: list[str] = [] + for res in (_resolver(), _resolver(fallback_resolvers)): + try: + answer = res.resolve(zone, "NS", raise_on_no_answer=False) + names = sorted({str(rdata.target).rstrip(".") for rdata in answer}) + if names: + break + except dns.exception.DNSException as exc: + log.debug("NS lookup for %s failed: %s", zone, exc) + + lookup = _resolver(fallback_resolvers) + for name in names: + for rtype in ("A", "AAAA"): + try: + for rdata in lookup.resolve(name, rtype, raise_on_no_answer=False): + ips.append(str(rdata)) + except dns.exception.DNSException: + continue + + unique = list(dict.fromkeys(ips)) + if unique: + log.info("Authoritative nameservers for %s: %s (%s)", zone, ", ".join(names), ", ".join(unique)) + else: + log.warning("Could not determine authoritative nameservers for %s, using public resolvers.", zone) + return unique + + +def txt_values(name: str, server: str) -> set[str]: + """TXT values for *name* as seen by the DNS server at *server*.""" + query = dns.message.make_query(name, dns.rdatatype.TXT) + values: set[str] = set() + try: + response = dns.query.udp(query, server, timeout=QUERY_TIMEOUT) + if response.flags & dns.flags.TC: + response = dns.query.tcp(query, server, timeout=QUERY_TIMEOUT) + except dns.exception.DNSException as exc: + log.debug("TXT query %s @%s failed: %s", name, server, exc) + return values + + for rrset in response.answer: + if rrset.rdtype != dns.rdatatype.TXT: + continue + for rdata in rrset: + values.add(b"".join(rdata.strings).decode("utf-8", "replace")) + return values + + +def wait_for_txt( + expected: dict[str, set[str]], + servers: list[str], + timeout: int, + interval: int, +) -> bool: + """Block until every server serves every expected TXT value, or *timeout* expires.""" + if not servers: + log.warning("No DNS servers to verify against - waiting %ss blindly.", interval * 2) + time.sleep(interval * 2) + return False + + deadline = time.monotonic() + timeout + attempt = 0 + while True: + attempt += 1 + missing: list[str] = [] + for name, wanted in expected.items(): + for server in servers: + seen = txt_values(name, server) + for value in wanted: + if value not in seen: + missing.append(f"{name} @{server}") + break + if not missing: + log.info("DNS propagation confirmed on all %d nameserver(s).", len(servers)) + return True + + remaining = deadline - time.monotonic() + if remaining <= 0: + log.error("DNS propagation timed out. Still missing: %s", ", ".join(sorted(set(missing)))) + return False + log.info( + "Attempt %d: waiting for DNS propagation (%d pending, %ds left)...", + attempt, + len(set(missing)), + int(remaining), + ) + time.sleep(min(interval, max(1, int(remaining)))) diff --git a/app/main.py b/app/main.py new file mode 100644 index 0000000..83674c3 --- /dev/null +++ b/app/main.py @@ -0,0 +1,230 @@ +"""CLI entry point: update DNS in Plesk, then issue a wildcard Let's Encrypt certificate.""" + +from __future__ import annotations + +import argparse +import ipaddress +import logging +import sys + +from . import __version__ +from .acme_client import AcmeFailure, AcmeManager, certificate_domains, csr_for, parse_chain +from .certfiles import certificate_info, format_info, renewal_needed, write_certificate_files +from .config import Config, ConfigError, is_staging, load_config +from .dnsutil import authoritative_servers, wait_for_txt +from .plesk import PleskClient, PleskError, normalise_name + +log = logging.getLogger("wildcard-cert") + + +class PleskDnsSolver: + """Puts the ACME dns-01 TXT records into Plesk and removes them afterwards.""" + + def __init__(self, plesk: PleskClient, cfg: Config, keep_records: bool = False, + ignore_propagation_timeout: bool = False): + self.plesk = plesk + self.cfg = cfg + self.keep_records = keep_records + self.ignore_propagation_timeout = ignore_propagation_timeout + self.created: list[tuple[str, str, str]] = [] # (record_id, name, value) + self.zones: set[str] = set() + self._purged: set[str] = set() + + def add_txt(self, name: str, value: str) -> None: + name = normalise_name(name) + # Leftovers from an aborted earlier run would only confuse the validation. + # Only once per name: a wildcard order puts two values on the same record + # name, and the second one must not wipe the first. + if name not in self._purged: + self._purged.add(name) + stale = self.plesk.delete_txt_records(name) + if stale: + log.info("Removed %d stale TXT record(s) for %s", stale, name) + record_id, zone = self.plesk.add_txt_record(name, value) + self.created.append((record_id, name, value)) + self.zones.add(zone) + + def wait_for_propagation(self, expected: dict[str, set[str]]) -> None: + for name, values in expected.items(): + for value in values: + if not self.plesk.zone_has_txt(name, value): + raise PleskError( + f"Plesk does not report the TXT record {name} after adding it. " + "Is the DNS zone managed by this Plesk server?" + ) + log.info("Plesk confirms all %d challenge record(s).", sum(len(v) for v in expected.values())) + + servers: list[str] = [] + for zone in self.zones: + servers.extend(authoritative_servers(zone, self.cfg.dns_resolvers)) + servers = list(dict.fromkeys(servers)) or list(self.cfg.dns_resolvers) + + ok = wait_for_txt( + expected, + servers, + self.cfg.dns_propagation_timeout, + self.cfg.dns_propagation_interval, + ) + if not ok and not self.ignore_propagation_timeout: + raise AcmeFailure( + "The _acme-challenge TXT records did not show up on the authoritative " + "nameservers within DNS_PROPAGATION_TIMEOUT. Aborting before Let's Encrypt " + "counts a failed validation. Use --ignore-propagation-timeout to try anyway." + ) + + def cleanup(self) -> None: + if self.keep_records: + log.warning("--keep-txt: leaving %d challenge record(s) in place.", len(self.created)) + return + for record_id, name, _value in self.created: + try: + self.plesk.delete_record(record_id) + except PleskError as exc: + log.warning("Could not remove challenge record %s (%s): %s", record_id, name, exc) + self.created.clear() + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + prog="wildcard-cert", + description="Create/update a DNS record in Plesk and issue a wildcard " + "Let's Encrypt certificate (dns-01) for it.", + ) + parser.add_argument("fqdn", help="DNS name, e.g. vpn.example.com") + parser.add_argument("ip", nargs="?", help="IPv4/IPv6 address for the A/AAAA record") + parser.add_argument("--ip", dest="ip_opt", help="alternative to the positional IP argument") + parser.add_argument("--san", action="append", default=[], + help="additional SAN (repeatable)") + parser.add_argument("--no-wildcard", action="store_true", + help="only the plain name, without *.") + parser.add_argument("--skip-dns", action="store_true", + help="do not touch the A/AAAA record (challenge records are still needed)") + parser.add_argument("--dns-only", action="store_true", + help="only create/update the A/AAAA record, no certificate") + parser.add_argument("--staging", action="store_true", + help="use the Let's Encrypt staging environment") + parser.add_argument("--force", action="store_true", + help="issue even if the existing certificate is still valid") + parser.add_argument("--keep-txt", action="store_true", + help="keep the _acme-challenge records (debugging)") + parser.add_argument("--ignore-propagation-timeout", action="store_true", + help="continue even if the TXT records are not visible in time") + parser.add_argument("--env", help="path to an .env file") + parser.add_argument("-v", "--verbose", action="store_true", help="debug output") + parser.add_argument("--version", action="version", version=f"%(prog)s {__version__}") + return parser + + +def setup_logging(verbose: bool) -> None: + logging.basicConfig( + level=logging.DEBUG if verbose else logging.INFO, + format="%(asctime)s %(levelname)-7s %(message)s", + datefmt="%H:%M:%S", + ) + for noisy in ("urllib3", "requests", "acme.client", "josepy"): + logging.getLogger(noisy).setLevel(logging.DEBUG if verbose else logging.WARNING) + + +def run(args: argparse.Namespace) -> int: + cfg = load_config(args.env, staging=True if args.staging else None) + cfg.validate() + + fqdn = normalise_name(args.fqdn) + if fqdn.startswith("*."): + fqdn = fqdn[2:] + ip = args.ip_opt or args.ip + if ip: + try: + ipaddress.ip_address(ip) + except ValueError: + raise ConfigError(f"{ip!r} is not a valid IP address.") from None + + domains = certificate_domains(fqdn, not args.no_wildcard, args.san) + + log.info("=" * 62) + log.info("Domain : %s", fqdn) + log.info("Certificate : %s", ", ".join(domains)) + log.info("IP address : %s", ip or "(unchanged)") + log.info("ACME : %s%s", cfg.acme_directory_url, " [STAGING]" if is_staging(cfg) else "") + log.info("Output : %s", cfg.cert_output_dir / fqdn) + log.info("=" * 62) + + plesk = PleskClient(cfg) + zone, domain_id = plesk.find_zone(fqdn) + log.info("Plesk DNS zone: %s (domain id %s)", zone, domain_id) + + # -- 1. A/AAAA record -------------------------------------------------- + if ip and not args.skip_dns: + action = plesk.ensure_address_record(fqdn, ip) + log.info("DNS record %s: %s", fqdn, action) + elif not ip: + log.info("No IP given - skipping the address record.") + + if args.dns_only: + log.info("--dns-only: done.") + return 0 + + # -- 2. renewal check -------------------------------------------------- + needed, reason = renewal_needed(cfg, fqdn, domains) + if not needed and not args.force: + log.info("Certificate %s (%s). Nothing to do - use --force to renew anyway.", reason, fqdn) + return 0 + log.info("Issuing certificate: %s", reason if needed else "forced") + + # -- 3. key + CSR ------------------------------------------------------ + private_key, csr_pem = csr_for(cfg, domains) + + # -- 4. ACME order with dns-01 ---------------------------------------- + manager = AcmeManager(cfg) + solver = PleskDnsSolver( + plesk, cfg, + keep_records=args.keep_txt, + ignore_propagation_timeout=args.ignore_propagation_timeout, + ) + try: + fullchain_pem = manager.obtain_certificate(domains, csr_pem, solver) + finally: + # obtain_certificate() cleans up itself; this catches anything that blew up + # before it got that far. cleanup() is idempotent. + solver.cleanup() + chain = parse_chain(fullchain_pem) + + # -- 5. write every artefact ------------------------------------------ + paths = write_certificate_files(cfg, fqdn, private_key, csr_pem, chain, domains) + + width = max(len(label) for label in paths) + print() + print("Files written") + print("=" * 62) + for label, path in paths.items(): + print(f" {label:<{width}} : {path}") + print() + print(format_info(certificate_info(chain, private_key, domains))) + print() + if is_staging(cfg): + print("NOTE: staging certificate - not trusted by browsers.") + print("PFX password: the value of CERT_PASSWORD from your .env") + return 0 + + +def main(argv: list[str] | None = None) -> int: + args = build_parser().parse_args(argv) + setup_logging(args.verbose) + try: + return run(args) + except ConfigError as exc: + log.error("Configuration error: %s", exc) + return 2 + except PleskError as exc: + log.error("Plesk error: %s", exc) + return 1 + except AcmeFailure as exc: + log.error("%s", exc) + return 1 + except KeyboardInterrupt: # pragma: no cover + log.error("Aborted.") + return 130 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/app/plesk.py b/app/plesk.py new file mode 100644 index 0000000..370531d --- /dev/null +++ b/app/plesk.py @@ -0,0 +1,295 @@ +"""Minimal Plesk XML-API client with the DNS operations we need. + +Endpoint: https://:8443/enterprise/control/agent.php +Auth: KEY header (API secret key) or HTTP_AUTH_LOGIN / HTTP_AUTH_PASSWD. +""" + +from __future__ import annotations + +import ipaddress +import logging +import xml.etree.ElementTree as ET +from dataclasses import dataclass + +import requests +import urllib3 + +from .config import Config + +log = logging.getLogger(__name__) + + +class PleskError(Exception): + """Raised when Plesk answers with an error status or unparsable data.""" + + +@dataclass(frozen=True) +class DnsRecord: + id: str + site_id: str + type: str + host: str # normalised: lower case, no trailing dot + value: str + + def __str__(self) -> str: # pragma: no cover - debug helper + return f"[{self.id}] {self.type} {self.host} -> {self.value}" + + +def normalise_name(name: str) -> str: + return name.strip().rstrip(".").lower() + + +def plesk_host(fqdn: str, zone: str) -> str: + """Host value for add_rec: relative label, or the absolute name for the zone apex.""" + relative = relative_host(fqdn, zone) + return relative if relative else normalise_name(fqdn) + "." + + +def relative_host(fqdn: str, zone: str) -> str: + """Return the host part of *fqdn* relative to *zone* ('' for the zone apex).""" + fqdn = normalise_name(fqdn) + zone = normalise_name(zone) + if fqdn == zone: + return "" + if not fqdn.endswith("." + zone): + raise PleskError(f"{fqdn!r} is not inside the DNS zone {zone!r}.") + return fqdn[: -(len(zone) + 1)] + + +class PleskClient: + def __init__(self, cfg: Config): + self.cfg = cfg + self.session = requests.Session() + headers = {"Content-Type": "text/xml", "HTTP_PRETTY_PRINT": "TRUE"} + if cfg.plesk_api_key: + headers["KEY"] = cfg.plesk_api_key + else: + headers["HTTP_AUTH_LOGIN"] = cfg.plesk_user + headers["HTTP_AUTH_PASSWD"] = cfg.plesk_password + self.session.headers.update(headers) + if not cfg.plesk_verify_tls: + urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) + self._zone_cache: dict[str, str] | None = None + + # -- low level --------------------------------------------------------- + + def _request(self, body: ET.Element) -> ET.Element: + packet = ET.Element("packet") + packet.append(body) + payload = ET.tostring(packet, encoding="utf-8", xml_declaration=True) + log.debug("Plesk request: %s", payload.decode("utf-8", "replace")) + try: + resp = self.session.post( + self.cfg.plesk_url, + data=payload, + verify=self.cfg.plesk_verify_tls, + timeout=self.cfg.plesk_timeout, + ) + except requests.RequestException as exc: + raise PleskError(f"Cannot reach Plesk at {self.cfg.plesk_url}: {exc}") from exc + + if resp.status_code == 401: + raise PleskError("Plesk rejected the credentials (HTTP 401). Check PLESK_API_KEY / PLESK_USER.") + if resp.status_code >= 400: + raise PleskError(f"Plesk returned HTTP {resp.status_code}: {resp.text[:500]}") + + log.debug("Plesk response: %s", resp.text) + try: + root = ET.fromstring(resp.content) + except ET.ParseError as exc: + raise PleskError(f"Invalid XML from Plesk: {exc}\n{resp.text[:500]}") from exc + + system_status = root.find("./system/status") + if system_status is not None and system_status.text != "ok": + errtext = root.findtext("./system/errtext", default="unknown error") + errcode = root.findtext("./system/errcode", default="?") + raise PleskError(f"Plesk API error {errcode}: {errtext}") + return root + + @staticmethod + def _check_result(result: ET.Element, context: str) -> None: + status = result.findtext("status", default="") + if status != "ok": + errcode = result.findtext("errcode", default="?") + errtext = result.findtext("errtext", default="unknown error") + raise PleskError(f"{context} failed (code {errcode}): {errtext}") + + # -- zones ------------------------------------------------------------- + + def list_zones(self) -> dict[str, str]: + """Map of DNS zone name -> Plesk domain id, for domains and subscriptions.""" + if self._zone_cache is not None: + return self._zone_cache + + zones: dict[str, str] = {} + for operator in ("site", "webspace"): + op = ET.Element(operator) + get = ET.SubElement(op, "get") + ET.SubElement(get, "filter") + dataset = ET.SubElement(get, "dataset") + ET.SubElement(dataset, "gen_info") + try: + root = self._request(op) + except PleskError as exc: + log.debug("%s.get failed: %s", operator, exc) + continue + + for result in root.findall(f"./{operator}/get/result"): + if result.findtext("status") != "ok": + continue + domain_id = result.findtext("id") + gen_info = result.find("./data/gen_info") + if domain_id is None or gen_info is None: + continue + for tag in ("name", "ascii-name"): + name = gen_info.findtext(tag) + if name: + zones.setdefault(normalise_name(name), domain_id) + + if not zones: + raise PleskError( + "No domains found on the Plesk server. Does the API user have access to any subscription?" + ) + self._zone_cache = zones + log.debug("Known Plesk zones: %s", sorted(zones)) + return zones + + def find_zone(self, fqdn: str) -> tuple[str, str]: + """Find the most specific Plesk zone hosting *fqdn*. Returns (zone_name, domain_id).""" + fqdn = normalise_name(fqdn) + zones = self.list_zones() + candidates = [z for z in zones if fqdn == z or fqdn.endswith("." + z)] + if not candidates: + raise PleskError( + f"No Plesk DNS zone found for {fqdn!r}. " + f"Known zones: {', '.join(sorted(zones)) or ''}" + ) + zone = max(candidates, key=len) + return zone, zones[zone] + + # -- records ----------------------------------------------------------- + + def get_records(self, domain_id: str) -> list[DnsRecord]: + dns = ET.Element("dns") + get_rec = ET.SubElement(dns, "get_rec") + flt = ET.SubElement(get_rec, "filter") + ET.SubElement(flt, "site-id").text = str(domain_id) + root = self._request(dns) + + records: list[DnsRecord] = [] + for result in root.findall("./dns/get_rec/result"): + status = result.findtext("status") + if status != "ok": + errtext = result.findtext("errtext", "") + # An empty zone reports an error instead of an empty list on some versions. + log.debug("get_rec result not ok: %s", errtext) + continue + data = result.find("data") + if data is None: + continue + records.append( + DnsRecord( + id=result.findtext("id", ""), + site_id=data.findtext("site-id", str(domain_id)), + type=(data.findtext("type", "") or "").upper(), + host=normalise_name(data.findtext("host", "") or ""), + value=(data.findtext("value", "") or "").strip().strip('"'), + ) + ) + return records + + def add_record(self, domain_id: str, rtype: str, host: str, value: str) -> str: + dns = ET.Element("dns") + add_rec = ET.SubElement(dns, "add_rec") + ET.SubElement(add_rec, "site-id").text = str(domain_id) + ET.SubElement(add_rec, "type").text = rtype.upper() + ET.SubElement(add_rec, "host").text = host + ET.SubElement(add_rec, "value").text = value + root = self._request(dns) + result = root.find("./dns/add_rec/result") + if result is None: + raise PleskError("Unexpected Plesk answer while adding a DNS record.") + self._check_result(result, f"Adding {rtype} record {host or '@'}") + rec_id = result.findtext("id", "") + log.info("Plesk: added %s record %s -> %s (id %s)", rtype.upper(), host or "@", value, rec_id) + return rec_id + + def delete_record(self, record_id: str) -> None: + dns = ET.Element("dns") + del_rec = ET.SubElement(dns, "del_rec") + flt = ET.SubElement(del_rec, "filter") + ET.SubElement(flt, "id").text = str(record_id) + root = self._request(dns) + result = root.find("./dns/del_rec/result") + if result is None: + raise PleskError("Unexpected Plesk answer while deleting a DNS record.") + self._check_result(result, f"Deleting DNS record {record_id}") + log.info("Plesk: deleted DNS record id %s", record_id) + + # -- high level helpers ------------------------------------------------ + + def ensure_address_record(self, fqdn: str, ip: str) -> str: + """Create or update the A/AAAA record for *fqdn*. + + Returns 'created' | 'updated' | 'unchanged'. + """ + rtype = "AAAA" if ipaddress.ip_address(ip).version == 6 else "A" + zone, domain_id = self.find_zone(fqdn) + host = plesk_host(fqdn, zone) + fqdn_n = normalise_name(fqdn) + + existing = [r for r in self.get_records(domain_id) if r.type == rtype and r.host == fqdn_n] + if any(r.value == ip for r in existing): + for stale in (r for r in existing if r.value != ip): + self.delete_record(stale.id) + log.info("DNS: %s record %s already points to %s", rtype, fqdn_n, ip) + return "unchanged" + + for stale in existing: + log.info("DNS: replacing %s record %s (%s -> %s)", rtype, fqdn_n, stale.value, ip) + self.delete_record(stale.id) + + self.add_record(domain_id, rtype, host, ip) + return "updated" if existing else "created" + + def add_txt_record(self, fqdn: str, value: str) -> tuple[str, str]: + """Add a TXT record. Returns (record_id, zone).""" + zone, domain_id = self.find_zone(fqdn) + host = plesk_host(fqdn, zone) + rec_id = self.add_record(domain_id, "TXT", host, value) + return rec_id, zone + + def delete_txt_records(self, fqdn: str, value: str | None = None) -> int: + """Delete TXT records for *fqdn* (optionally only those carrying *value*).""" + zone, domain_id = self.find_zone(fqdn) + fqdn_n = normalise_name(fqdn) + deleted = 0 + for rec in self.get_records(domain_id): + if rec.type != "TXT" or rec.host != fqdn_n: + continue + if value is not None and rec.value != value: + continue + try: + self.delete_record(rec.id) + deleted += 1 + except PleskError as exc: + log.warning("Could not delete TXT record %s: %s", rec.id, exc) + return deleted + + def zone_has_txt(self, fqdn: str, value: str) -> bool: + _zone, domain_id = self.find_zone(fqdn) + fqdn_n = normalise_name(fqdn) + return any( + r.type == "TXT" and r.host == fqdn_n and r.value == value + for r in self.get_records(domain_id) + ) + + def nameservers(self, fqdn: str) -> list[str]: + """NS records of the zone as configured in Plesk (best effort).""" + zone, domain_id = self.find_zone(fqdn) + zone_n = normalise_name(zone) + return [ + normalise_name(r.value) + for r in self.get_records(domain_id) + if r.type == "NS" and r.host == zone_n + ] diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000..dff6519 --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,17 @@ +version: "3.8" + +services: + wildcard-cert: + build: . + image: wildcard-lets-encrypt-cert-plesk-creator:latest + env_file: + - .env + # Damit die Zertifikate auf dem Host nicht root gehoeren. + user: "${PUID:-1000}:${PGID:-1000}" + volumes: + - ./certs:/app/certs # hier landen alle Zertifikatsdateien + - ./data:/app/data # ACME-Account-Key (wiederverwenden = keine Rate-Limits) + environment: + CERT_OUTPUT_DIR: /app/certs + ACME_ACCOUNT_DIR: /app/data + TZ: "${TZ:-Europe/Berlin}" diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..98c9997 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,5 @@ +acme>=2.9.0 +cryptography>=42.0.0 +requests>=2.31.0 +python-dotenv>=1.0.1 +dnspython>=2.6.1 diff --git a/run.sh b/run.sh new file mode 100755 index 0000000..146bae2 --- /dev/null +++ b/run.sh @@ -0,0 +1,32 @@ +#!/usr/bin/env bash +# Bequemer Wrapper: ./run.sh vpn.example.com 1.2.3.4 [weitere Optionen] +set -euo pipefail + +cd "$(dirname "$0")" + +if [ ! -f .env ]; then + echo "Es gibt noch keine .env - bitte .env.example kopieren und ausfuellen:" >&2 + echo " cp .env.example .env && \$EDITOR .env" >&2 + exit 2 +fi + +mkdir -p certs data + +if docker compose version >/dev/null 2>&1; then + DC="docker compose" +elif command -v docker-compose >/dev/null 2>&1; then + DC="docker-compose" +else + echo "Weder 'docker compose' noch 'docker-compose' gefunden." >&2 + exit 2 +fi + +export PUID="${PUID:-$(id -u)}" +export PGID="${PGID:-$(id -g)}" + +if ! docker image inspect wildcard-lets-encrypt-cert-plesk-creator:latest >/dev/null 2>&1; then + echo ">> Baue Docker-Image ..." + $DC build +fi + +exec $DC run --rm wildcard-cert "$@" diff --git a/tests/test_acme_api.py b/tests/test_acme_api.py new file mode 100644 index 0000000..c9c3f77 --- /dev/null +++ b/tests/test_acme_api.py @@ -0,0 +1,54 @@ +"""Check the acme-library API usage against the real Let's Encrypt staging directory. + +Fetches the directory and builds the client; account registration is stubbed out so +nothing is created on the CA side. +""" +import os +import pathlib +import sys +import tempfile + +sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent.parent)) +os.environ.update( + PLESK_HOST="p", PLESK_API_KEY="k", ACME_EMAIL="a@b.de", CERT_PASSWORD="x", + ACME_ACCOUNT_DIR=tempfile.mkdtemp(), ACME_STAGING="true", +) + +import acme, josepy +from app.config import load_config +from app import acme_client +from acme import challenges, messages + +print("acme:", getattr(acme, "__version__", "?"), "josepy:", getattr(josepy, "__version__", "?")) + +registered = [] +acme_client.AcmeManager._register = lambda self, c: registered.append(c) + +cfg = load_config(env_file="/dev/null") +m = acme_client.AcmeManager(cfg) +print("directory newOrder:", m.client.directory["newOrder"]) +assert registered, "register hook not called" +assert m.account_key_file.is_file() and oct(m.account_key_file.stat().st_mode)[-3:] == "600" + +# reload must reuse the very same account key +m2 = acme_client.AcmeManager(cfg) +assert m2.account_key.thumbprint() == m.account_key.thumbprint() +print("account key reused ok") + +# dns-01 challenge helpers used in obtain_certificate() +chall = challenges.DNS01(token=b"0123456789abcdef0123456789abcdef") +validation = chall.validation(m.account_key) +name = chall.validation_domain_name("example.com") +assert name == "_acme-challenge.example.com", name +assert isinstance(validation, str) and len(validation) == 43, validation +resp = chall.response(m.account_key) +assert isinstance(resp, challenges.DNS01Response) +print("dns-01 helpers ok:", name, validation) + +# the API surface obtain_certificate() relies on +for attr in ("new_order", "answer_challenge", "poll_and_finalize"): + assert hasattr(m.client, attr), attr +assert hasattr(messages, "STATUS_VALID") +assert hasattr(messages.NewRegistration, "from_data") +print("client API surface ok") +print("\nACME API CHECK PASSED") diff --git a/tests/test_cli.py b/tests/test_cli.py new file mode 100644 index 0000000..280070a --- /dev/null +++ b/tests/test_cli.py @@ -0,0 +1,162 @@ +"""End-to-end wiring test of app.main with a stateful fake Plesk and a fake CA.""" +import datetime +import os +import pathlib +import sys +import tempfile +import xml.etree.ElementTree as ET + +sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent.parent)) + +OUT = tempfile.mkdtemp() +os.environ.update( + PLESK_HOST="plesk.example.com", PLESK_API_KEY="dummy", ACME_EMAIL="a@b.de", + CERT_PASSWORD="secret123", CERT_OUTPUT_DIR=OUT, ACME_ACCOUNT_DIR=tempfile.mkdtemp(), + RSA_KEY_SIZE="2048", +) + +from pathlib import Path +from cryptography import x509 +from cryptography.hazmat.primitives import hashes, serialization +from cryptography.hazmat.primitives.asymmetric import rsa +from cryptography.x509.oid import NameOID + +from app import main as appmain +from app import plesk as pleskmod + +# ---------------------------------------------------------------- fake Plesk +class FakeZone: + def __init__(self): + self.next_id = 100 + self.records = [] # dicts: id, type, host, value + + def add(self, rtype, host, value, zone="example.com"): + self.next_id += 1 + fqdn = zone if host in ("", None) else f"{host}.{zone}" + self.records.append({"id": str(self.next_id), "type": rtype, "host": fqdn + ".", "value": value}) + return str(self.next_id) + + +ZONE = FakeZone() +ZONE.add("NS", "", "ns1.example.com.") +ZONE.add("A", "vpn", "203.0.113.1") + + +def fake_request(self, body): + tag = body.tag + if tag == "site": + return ET.fromstring( + 'ok3' + 'example.com' + ) + if tag == "webspace": + return ET.fromstring('') + if body.find("get_rec") is not None: + parts = "".join( + f'ok{r["id"]}3' + f'{r["type"]}{r["host"]}{r["value"]}' + for r in ZONE.records + ) + return ET.fromstring(f"{parts}") + if (add := body.find("add_rec")) is not None: + rid = ZONE.add(add.findtext("type"), add.findtext("host") or "", add.findtext("value")) + return ET.fromstring( + f"ok{rid}" + ) + if (dele := body.find("del_rec")) is not None: + rid = dele.findtext("./filter/id") + before = len(ZONE.records) + ZONE.records[:] = [r for r in ZONE.records if r["id"] != rid] + assert len(ZONE.records) == before - 1, f"record {rid} not found" + return ET.fromstring( + "ok" + ) + raise AssertionError("unexpected " + ET.tostring(body, encoding="unicode")) + + +pleskmod.PleskClient._request = fake_request + +# ------------------------------------------------------------------ fake CA +issued = {} + + +class FakeAcme: + def __init__(self, cfg): + self.cfg = cfg + + def obtain_certificate(self, domains, csr_pem, solver): + csr = x509.load_pem_x509_csr(csr_pem) + issued["domains"] = domains + # One challenge per identifier - wildcard and base name share the record name + # but carry different values, so both must be present at the same time. + expected = {} + for d in domains: + base = d[2:] if d.startswith("*.") else d + name = f"_acme-challenge.{base}" + value = "val-" + d.replace("*.", "wild-").replace(".", "") + expected.setdefault(name, set()).add(value) + solver.add_txt(name, value) + solver.wait_for_propagation(expected) + issued["expected"] = expected + ca_key = rsa.generate_private_key(public_exponent=65537, key_size=2048) + now = datetime.datetime.now(datetime.timezone.utc) + ca = (x509.CertificateBuilder() + .subject_name(x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "Fake CA")])) + .issuer_name(x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "Fake CA")])) + .public_key(ca_key.public_key()).serial_number(x509.random_serial_number()) + .not_valid_before(now).not_valid_after(now + datetime.timedelta(days=3000)) + .add_extension(x509.BasicConstraints(ca=True, path_length=None), critical=True) + .sign(ca_key, hashes.SHA256())) + leaf = (x509.CertificateBuilder().subject_name(csr.subject).issuer_name(ca.subject) + .public_key(csr.public_key()).serial_number(x509.random_serial_number()) + .not_valid_before(now).not_valid_after(now + datetime.timedelta(days=90)) + .add_extension( + csr.extensions.get_extension_for_class(x509.SubjectAlternativeName).value, False) + .sign(ca_key, hashes.SHA256())) + return "".join(c.public_bytes(serialization.Encoding.PEM).decode() for c in (leaf, ca)) + + +appmain.AcmeManager = FakeAcme +appmain.authoritative_servers = lambda zone, resolvers: ["192.0.2.53"] +appmain.wait_for_txt = lambda expected, servers, timeout, interval: True + +# -------------------------------------------------------------------- tests +print("--- run 1: new record + new certificate") +rc = appmain.main(["vpn.example.com", "203.0.113.9", "--env", "/dev/null"]) +assert rc == 0, rc +assert issued["domains"] == ["vpn.example.com", "*.vpn.example.com"] +a_records = [r for r in ZONE.records if r["type"] == "A"] +assert len(a_records) == 1 and a_records[0]["value"] == "203.0.113.9", a_records +assert not [r for r in ZONE.records if r["type"] == "TXT"], "challenge records not cleaned up" +assert (Path(OUT) / "vpn.example.com" / "cert.pfx").is_file() + +print("\n--- run 2: nothing to do (cert still valid)") +issued.clear() +rc = appmain.main(["vpn.example.com", "203.0.113.9", "--env", "/dev/null"]) +assert rc == 0 and not issued, "should not have issued again" + +print("\n--- run 3: --force") +rc = appmain.main(["vpn.example.com", "203.0.113.9", "--force", "--env", "/dev/null"]) +assert rc == 0 and issued + +print("\n--- run 4: --no-wildcard --san + wildcard input normalised") +issued.clear() +rc = appmain.main(["*.vpn.example.com", "--no-wildcard", "--san", "alt.example.com", + "--env", "/dev/null"]) +assert rc == 0 and issued["domains"] == ["vpn.example.com", "alt.example.com"], issued + +print("\n--- run 5: --dns-only") +issued.clear() +rc = appmain.main(["www.example.com", "203.0.113.55", "--dns-only", "--env", "/dev/null"]) +assert rc == 0 and not issued +assert any(r["host"] == "www.example.com." and r["value"] == "203.0.113.55" for r in ZONE.records) + +print("\n--- error paths") +assert appmain.main(["vpn.example.com", "not-an-ip", "--env", "/dev/null"]) == 2 +assert appmain.main(["vpn.otherdomain.tld", "203.0.113.1", "--env", "/dev/null"]) == 1 + +os.environ["PLESK_API_KEY"] = "" +os.environ["PLESK_HOST"] = "" +assert appmain.main(["vpn.example.com", "--env", "/dev/null"]) == 2 + +print("\nALL CLI TESTS PASSED") diff --git a/tests/test_offline.py b/tests/test_offline.py new file mode 100644 index 0000000..fc2a09c --- /dev/null +++ b/tests/test_offline.py @@ -0,0 +1,204 @@ +"""Smoke test: no network, no Plesk. Exercises parsing, DNS logic and file output.""" +import datetime +import os +import pathlib +import sys +import tempfile +import xml.etree.ElementTree as ET +from pathlib import Path + +sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent.parent)) + +os.environ.update( + PLESK_HOST="plesk.example.com", PLESK_API_KEY="dummy", ACME_EMAIL="a@b.de", + CERT_PASSWORD="secret123", KEY_TYPE="ec", CERT_OUTPUT_DIR=tempfile.mkdtemp(), + ACME_ACCOUNT_DIR=tempfile.mkdtemp(), +) + +from app.config import load_config +from app.plesk import PleskClient, relative_host, normalise_name, PleskError +from app.acme_client import certificate_domains, csr_for, parse_chain, build_csr +from app.certfiles import write_certificate_files, certificate_info, format_info, renewal_needed +from cryptography import x509 +from cryptography.hazmat.primitives import hashes, serialization +from cryptography.x509.oid import NameOID + +cfg = load_config(env_file="/dev/null") +cfg.validate() +print("config ok:", cfg.plesk_url) + +# ---- host helpers --------------------------------------------------------- +assert relative_host("vpn.example.com", "example.com") == "vpn" +assert relative_host("example.com", "example.com") == "" +assert relative_host("_acme-challenge.a.b.example.com", "example.com") == "_acme-challenge.a.b" +assert normalise_name("VPN.Example.COM.") == "vpn.example.com" +try: + relative_host("other.de", "example.com") + raise SystemExit("FAIL: expected PleskError") +except PleskError: + pass +print("host helpers ok") + +# ---- Plesk XML parsing with a stubbed transport --------------------------- +SITES = """ +ok3example.com +example.com +ok7sub.example.com +""" +WEBSPACES = """ +ok3example.com +""" +RECORDS = """ +ok1013A +vpn.example.com.10.0.0.1 +ok1023NS +example.com.ns1.example.com. +ok1033TXT +_acme-challenge.vpn.example.com."oldvalue" +""" +ADD_OK = """ok +555""" +DEL_OK = """ok +101""" + +calls = [] + + +class StubPlesk(PleskClient): + def _request(self, body): + payload = ET.tostring(body, encoding="unicode") + calls.append(payload) + if body.tag == "site": + return ET.fromstring(SITES) + if body.tag == "webspace": + return ET.fromstring(WEBSPACES) + if body.find("get_rec") is not None: + return ET.fromstring(RECORDS) + if body.find("add_rec") is not None: + return ET.fromstring(ADD_OK) + if body.find("del_rec") is not None: + return ET.fromstring(DEL_OK) + raise AssertionError("unexpected request " + payload) + + +p = StubPlesk(cfg) +assert p.list_zones() == {"example.com": "3", "sub.example.com": "7"}, p.list_zones() +assert p.find_zone("vpn.example.com") == ("example.com", "3") +assert p.find_zone("x.sub.example.com") == ("sub.example.com", "7") +recs = p.get_records("3") +assert [r.type for r in recs] == ["A", "NS", "TXT"] +assert recs[2].value == "oldvalue", recs[2].value # quotes stripped +assert p.nameservers("example.com") == ["ns1.example.com"] + +calls.clear() +assert p.ensure_address_record("vpn.example.com", "10.0.0.1") == "unchanged" +assert not any("add_rec" in c for c in calls) + +calls.clear() +assert p.ensure_address_record("vpn.example.com", "10.0.0.2") == "updated" +assert any("del_rec" in c for c in calls) and any("add_rec" in c for c in calls) +add = [c for c in calls if "add_rec" in c][0] +assert "vpn" in add and "10.0.0.2" in add and "A" in add, add + +calls.clear() +assert p.ensure_address_record("new.example.com", "2001:db8::1") == "created" +add = [c for c in calls if "add_rec" in c][0] +assert "AAAA" in add, add + +calls.clear() +assert p.ensure_address_record("example.com", "10.0.0.9") == "created" +add = [c for c in calls if "add_rec" in c][0] +assert "example.com." in add, add # apex -> absolute name + +assert p.zone_has_txt("_acme-challenge.vpn.example.com", "oldvalue") is True +assert p.zone_has_txt("_acme-challenge.vpn.example.com", "nope") is False +assert p.delete_txt_records("_acme-challenge.vpn.example.com") == 1 +print("plesk client ok") + +# ---- domains / CSR -------------------------------------------------------- +domains = certificate_domains("vpn.example.com", True, ["extra.example.com", "vpn.example.com"]) +assert domains == ["vpn.example.com", "*.vpn.example.com", "extra.example.com"], domains +key, csr_pem = csr_for(cfg, domains) +csr = x509.load_pem_x509_csr(csr_pem) +sans = csr.extensions.get_extension_for_class(x509.SubjectAlternativeName).value.get_values_for_type(x509.DNSName) +assert sans == domains, sans +assert csr.subject.get_attributes_for_oid(NameOID.COMMON_NAME)[0].value == "vpn.example.com" +print("csr ok:", sans) + +# ---- fake chain -> file output ------------------------------------------- +def selfsigned(cn, subject_key, issuer_name=None, issuer_key=None, sans=None, days=90, ca=False): + now = datetime.datetime.now(datetime.timezone.utc) + subject = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, cn)]) + issuer = issuer_name or subject + b = (x509.CertificateBuilder().subject_name(subject).issuer_name(issuer) + .public_key(subject_key.public_key()).serial_number(x509.random_serial_number()) + .not_valid_before(now - datetime.timedelta(minutes=5)) + .not_valid_after(now + datetime.timedelta(days=days))) + if sans: + b = b.add_extension(x509.SubjectAlternativeName([x509.DNSName(d) for d in sans]), critical=False) + if ca: + b = b.add_extension(x509.BasicConstraints(ca=True, path_length=None), critical=True) + return b.sign(issuer_key or subject_key, hashes.SHA256()) + +from cryptography.hazmat.primitives.asymmetric import rsa +ca_key = rsa.generate_private_key(public_exponent=65537, key_size=2048) +ca_cert = selfsigned("Test CA R1", ca_key, ca=True, days=3000) +leaf = selfsigned("vpn.example.com", key, ca_cert.subject, ca_key, sans=domains) +chain = [leaf, ca_cert] + +# round-trip through the PEM parser used for the ACME answer +pem = b"".join(c.public_bytes(serialization.Encoding.PEM) for c in chain).decode() +assert len(parse_chain(pem)) == 2 + +paths = write_certificate_files(cfg, "vpn.example.com", key, csr_pem, chain, domains) +out = Path(os.environ["CERT_OUTPUT_DIR"]) / "vpn.example.com" +expected_files = { + "privkey.pem", "privkey-traditional.pem", "privkey-encrypted.pem", "pubkey.pem", + "csr.pem", "cert.pem", "cert.crt", "cert.der", "chain.pem", "chain-01.pem", + "fullchain.pem", "bundle.pem", "cert.pfx", "cert-info.json", "cert-info.txt", +} +present = {f.name for f in out.iterdir()} +assert expected_files <= present, expected_files - present +assert (out / "privkey.pem").stat().st_mode & 0o777 == 0o600 +assert (out / "cert.pfx").stat().st_mode & 0o777 == 0o600 +print("files ok:", sorted(present)) + +# PFX must open with the configured password +from cryptography.hazmat.primitives.serialization import pkcs12 +k2, c2, cas = pkcs12.load_key_and_certificates((out / "cert.pfx").read_bytes(), b"secret123") +assert c2.subject == leaf.subject and len(cas) == 1 +try: + pkcs12.load_key_and_certificates((out / "cert.pfx").read_bytes(), b"wrong") + raise SystemExit("FAIL: wrong pfx password accepted") +except ValueError: + pass +# encrypted private key too +serialization.load_pem_private_key((out / "privkey-encrypted.pem").read_bytes(), b"secret123") +# bundle = key + leaf + chain +bundle = (out / "bundle.pem").read_text() +assert bundle.count("BEGIN CERTIFICATE") == 2 and "PRIVATE KEY" in bundle +assert (out / "fullchain.pem").read_text().count("BEGIN CERTIFICATE") == 2 +assert (out / "chain.pem").read_text().count("BEGIN CERTIFICATE") == 1 +print("pfx/bundle ok") + +info = certificate_info(chain, key, domains) +assert info["leaf"]["subject_alternative_names"] == domains +assert info["key"]["algorithm"] == "EC" +print(format_info(info)) + +needed, reason = renewal_needed(cfg, "vpn.example.com", domains) +assert needed is False, reason +needed, reason = renewal_needed(cfg, "vpn.example.com", domains + ["other.example.com"]) +assert needed is True and "missing names" in reason, reason +needed, reason = renewal_needed(cfg, "unknown.example.com", domains) +assert needed is True and "no certificate" in reason +print("renewal check ok") + +# ---- legacy pfx path ------------------------------------------------------ +cfg.pfx_legacy_compat = True +write_certificate_files(cfg, "legacy.example.com", key, csr_pem, chain, domains) +legacy = Path(os.environ["CERT_OUTPUT_DIR"]) / "legacy.example.com" / "cert.pfx" +pkcs12.load_key_and_certificates(legacy.read_bytes(), b"secret123") +print("legacy pfx ok") + +print("\nALL SMOKE TESTS PASSED")