wildcard-lets-encrypt-cert-plesk-creator/tests/test_cli.py

"""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 <system>-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"<result><status>ok</status><id>{i}</id><data><gen_info>"
                f"<name>{n}</name></gen_info></data></result>"
                for n, i in selected.items()
            )
        else:
            inner = ("<result><status>error</status><errcode>1013</errcode>"
                     "<errtext>Object not found</errtext></result>")
        return ET.fromstring(f"<packet><{tag}><get>{inner}</get></{tag}></packet>")
    if body.find("get_rec") is not None:
        parts = "".join(
            f'<result><status>ok</status><id>{r["id"]}</id><data><site-id>3</site-id>'
            f'<type>{r["type"]}</type><host>{r["host"]}</host><value>{r["value"]}</value></data></result>'
            for r in ZONE.records
        )
        return ET.fromstring(f"<packet><dns><get_rec>{parts}</get_rec></dns></packet>")
    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"<packet><dns><add_rec><result><status>ok</status><id>{rid}</id></result></add_rec></dns></packet>"
        )
    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(
            "<packet><dns><del_rec><result><status>ok</status></result></del_rec></dns></packet>"
        )
    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")