"""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") # Wie beim echten Server von Stefan: die Zone haengt an einem Abo (webspace), # 'site' kennt sie nicht, und dieser API-User darf ueberhaupt nichts auflisten. SITE_ZONES = {} WEBSPACE_ZONES = {"example.com": "3"} ENUMERATION_ALLOWED = False def fake_request(self, body): tag = body.tag if tag in ("site", "webspace"): wanted = body.findtext("./get/filter/name") if wanted is None and not ENUMERATION_ALLOWED: # _request() wandelt einen -Fehler in eine PleskError um raise pleskmod.PleskError(f"Plesk API error 1005: Permission denied ({tag}.get)") zones = SITE_ZONES if tag == "site" else WEBSPACE_ZONES selected = {n: i for n, i in zones.items() if wanted is None or n == wanted} if selected: inner = "".join( f"ok{i}" f"{n}" for n, i in selected.items() ) else: inner = ("error1013" "Object not found") return ET.fromstring(f"<{tag}>{inner}") 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--- run 6: --zone erzwingt die Zone") issued.clear() rc = appmain.main(["endian.example.com", "203.0.113.77", "--zone", "example.com", "--env", "/dev/null"]) assert rc == 0 and issued["domains"] == ["endian.example.com", "*.endian.example.com"], issued assert any(r["host"] == "endian.example.com." and r["value"] == "203.0.113.77" for r in ZONE.records), 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")