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

"""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 ---------------------------
# Only these two exist on the fake Plesk; 'sub.example.com' is an addon domain
# with its own zone, everything else is just a record inside a zone.
SITE_ZONES = {"example.com": "3", "sub.example.com": "7"}
WEBSPACE_ZONES = {"example.com": "3"}


def domains_response(operator, zones, wanted):
    """Mimic <site>/<webspace> get, with and without a name filter."""
    selected = {n: i for n, i in zones.items() if wanted is None or n == wanted}
    if not selected:
        body = ('<result><status>error</status><errcode>1013</errcode>'
                '<errtext>Object not found</errtext></result>')
    else:
        body = "".join(
            f'<result><status>ok</status><id>{i}</id><data><gen_info>'
            f'<name>{n}</name><ascii-name>{n}</ascii-name></gen_info></data></result>'
            for n, i in selected.items()
        )
    return f"<packet><{operator}><get>{body}</get></{operator}></packet>"
RECORDS = """<?xml version="1.0"?><packet><dns><get_rec>
<result><status>ok</status><id>101</id><data><site-id>3</site-id><type>A</type>
<host>vpn.example.com.</host><value>10.0.0.1</value></data></result>
<result><status>ok</status><id>102</id><data><site-id>3</site-id><type>NS</type>
<host>example.com.</host><value>ns1.example.com.</value></data></result>
<result><status>ok</status><id>103</id><data><site-id>3</site-id><type>TXT</type>
<host>_acme-challenge.vpn.example.com.</host><value>"oldvalue"</value></data></result>
</get_rec></dns></packet>"""
ADD_OK = """<?xml version="1.0"?><packet><dns><add_rec><result><status>ok</status>
<id>555</id></result></add_rec></dns></packet>"""
DEL_OK = """<?xml version="1.0"?><packet><dns><del_rec><result><status>ok</status>
<id>101</id></result></del_rec></dns></packet>"""

calls = []


class StubPlesk(PleskClient):
    def _request(self, body):
        payload = ET.tostring(body, encoding="unicode")
        calls.append(payload)
        if body.tag in ("site", "webspace"):
            wanted = body.findtext("./get/filter/name")
            zones = SITE_ZONES if body.tag == "site" else WEBSPACE_ZONES
            return ET.fromstring(domains_response(body.tag, zones, wanted))
        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()

# A name that is only a record inside a zone must resolve to that zone ...
assert p.find_zone("vpn.example.com") == ("example.com", "3")
assert p.find_zone("deep.down.example.com") == ("example.com", "3")
# ... while a real addon domain wins over its parent (most specific first).
assert p.find_zone("sub.example.com") == ("sub.example.com", "7")
assert p.find_zone("x.sub.example.com") == ("sub.example.com", "7")
assert p.find_zone("example.com") == ("example.com", "3")

# The zone lookup must not depend on being allowed to enumerate all domains.
enumerating = StubPlesk(cfg)
enumerating.list_zones = lambda: (_ for _ in ()).throw(AssertionError("enumerated all domains"))
assert enumerating.find_zone("vpn.example.com") == ("example.com", "3")

# --zone overrides the detection
hinted = StubPlesk(cfg)
hinted.zone_hint = "example.com"
assert hinted.find_zone("x.sub.example.com") == ("example.com", "3")
try:
    hinted.find_zone("vpn.other.tld")
    raise SystemExit("FAIL: zone hint mismatch not detected")
except PleskError:
    pass
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 "<host>vpn</host>" in add and "<value>10.0.0.2</value>" in add and "<type>A</type>" 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 "<type>AAAA</type>" 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 "<host>example.com.</host>" 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")