diff --git a/CHANGELOG.md b/CHANGELOG.md index 1a33f78..cc2384a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,35 @@ # Changelog +## 1.1.0 — 2026-08-28 — a fixed certificate authority, so a firmware upgrade looks like one + +**The certificate authority is now shipped with the package rather than generated at startup**, and is identical to the one panelbench ships. Stopping this +simulator and starting panelbench rehearses a firmware upgrade on one panel — the config carries the serial across, and the panel keeps its address and ports — +but every install minted its own authority, so the swap presented a new trust anchor and read as a panel substitution to anything pinned to the old one. A +firmware upgrade does not rotate a panel's certificate authority, and now neither does the swap. + +Its SHA-256, the value Home Assistant pins and displays, is `3cf8c14a78900b8736870c95adcc931cdcb3a51bc3029c96efafd0a4cb790d97`. + +**Upgrading raises one "SPAN Panel certificate authority changed" repair, once, and it is expected.** The generated authority this install pinned is being +replaced by the shipped one; open the repair, check the fingerprint against the value above, and accept it. Doing this now, at an ordinary add-on upgrade, is +why the later swap to panelbench needs no repair at all — so upgrade this simulator before moving across, not after. + +The authority's private key is committed deliberately and is not a leaked secret. It signs nothing that chains to a real panel, which mints its own authority in +firmware and is pinned per config entry, and a real panel ever reporting the fingerprint above would be conclusive evidence of tampering. + +### Fixed + +- **A changed advertised address or container hostname now re-signs only the server certificate**, where it used to regenerate the certificate authority along + with it and present a pinned consumer with a trust anchor that had rotated for no reason. +- **A server certificate signed by a superseded authority is detected by signature rather than by issuer name**, which is the only way to tell two of these + authorities apart: every authority the simulator ever generated carries the same subject and none carries a key identifier. +- **An expired or nearly expired server certificate is re-signed at startup** instead of being served until a handshake fails against an anchor that never + changed — a failure a pinned consumer correctly reports as retryable and then retries forever. +- **A corrupt or unreadable server certificate is replaced rather than raised out of startup**, where it previously put the add-on into a restart loop. +- **An advertised address that is not an IP address is ignored with a warning** instead of aborting certificate generation. +- **Certificate files are written atomically**, so a container killed mid-write cannot leave a truncated certificate behind. +- **The authority's private key is no longer written into the certificate directory**, and one left there by an earlier build is removed; nothing reads it, and + it was world-readable in a directory that survives upgrades. + ## 1.0.18 — 2026-08-28 — an unreachable Home Assistant no longer aborts startup **`HAClient.async_validate` now degrades to `False` on a connection timeout** instead of letting the exception escape and take startup with it. The client sets diff --git a/pyproject.toml b/pyproject.toml index b885951..36a4725 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "span-panel-simulator" -version = "1.0.18" +version = "1.1.0" description = "Standalone eBus simulator for SPAN panels" requires-python = ">=3.14" dependencies = [ diff --git a/scripts/sync-version.sh b/scripts/sync-version.sh index a4c8eca..7d93c08 100755 --- a/scripts/sync-version.sh +++ b/scripts/sync-version.sh @@ -23,7 +23,10 @@ DIRTY=0 sync_file() { local file="$1" pattern="$2" replacement="$3" if ! grep -qF "$replacement" "$file"; then - sed -i '' "s|${pattern}|${replacement}|" "$file" + # No `-i`: BSD wants `-i ''` and GNU wants `-i` alone, and getting it + # wrong fails only on the platform you are not developing on. A temp + # file and a move behave the same everywhere. + sed "s|${pattern}|${replacement}|" "$file" > "$file.tmp" && mv "$file.tmp" "$file" git add "$file" echo "sync-version: updated $file -> $VERSION" DIRTY=1 diff --git a/span_panel_simulator/Dockerfile b/span_panel_simulator/Dockerfile index 482a25a..4dc3553 100644 --- a/span_panel_simulator/Dockerfile +++ b/span_panel_simulator/Dockerfile @@ -32,7 +32,7 @@ EXPOSE 18883 8081 18080 LABEL io.hass.name="SPAN Panel Simulator" \ io.hass.description="Simulates a SPAN electrical panel for testing and upgrade modeling" \ io.hass.type="addon" \ - io.hass.version="1.0.18" \ + io.hass.version="1.1.0" \ io.hass.arch="aarch64|amd64" CMD ["/run.sh"] diff --git a/span_panel_simulator/config.yaml b/span_panel_simulator/config.yaml index 114bc56..546c0d1 100644 --- a/span_panel_simulator/config.yaml +++ b/span_panel_simulator/config.yaml @@ -1,6 +1,6 @@ name: "SPAN Panel Simulator" description: "Simulates a SPAN electrical panel for testing and upgrade modeling" -version: "1.0.18" +version: "1.1.0" slug: "span_panel_simulator" url: "https://github.com/SpanPanel/simulator" image: "ghcr.io/spanpanel/simulator/{arch}" diff --git a/src/span_panel_simulator/__init__.py b/src/span_panel_simulator/__init__.py index 2368c01..f4bcd18 100644 --- a/src/span_panel_simulator/__init__.py +++ b/src/span_panel_simulator/__init__.py @@ -1,3 +1,3 @@ """Standalone eBus simulator for SPAN panels.""" -__version__ = "1.0.18" +__version__ = "1.1.0" diff --git a/src/span_panel_simulator/_ca/README.md b/src/span_panel_simulator/_ca/README.md new file mode 100644 index 0000000..8f16c40 --- /dev/null +++ b/src/span_panel_simulator/_ca/README.md @@ -0,0 +1,49 @@ +# The emulator's certificate authority + +`ca.crt` and `ca.key` are a fixed, deliberately public certificate authority, committed on +purpose. **This is not a leaked secret.** It is test material, in the tradition of Debian's +`ssl-cert-snakeoil` key, and secret scanners flagging it should be allowlisted rather than +obeyed. + +SHA-256 of the certificate's DER bytes — the value the Home Assistant integration pins, +reports under `panel_ca` in diagnostics, and displays in a certificate-authority-changed +repair: + +``` +3cf8c14a78900b8736870c95adcc931cdcb3a51bc3029c96efafd0a4cb790d97 +``` + +## Why it is fixed rather than generated + +The simulator emulates SPAN firmware before r202633 and panelbench emulates r202633 and +later, so stopping one and starting the other rehearses a firmware upgrade on a single +panel. A firmware upgrade does not rotate a panel's certificate authority, and a consumer +that pins the authority is right to treat a change as worth stopping for. When each +emulator minted its own CA the swap looked like a panel substitution, which is the one +thing the rehearsal must not simulate. + +Both repositories ship these bytes identically. They are the only shared state between two +otherwise decoupled projects, and nothing but this directory couples them: neither imports +the other, and the simulator's eventual archival leaves panelbench unaffected. + +## What the public key does and does not cost + +Anyone holding `ca.key` can mint a certificate that an integration entry pinned to *this* +authority will trust, and so can impersonate an emulated panel or read its traffic. What +that buys is control of synthetic circuits on an entry someone created for testing, and +the broker credentials it would expose are already public constants in `run.sh`. + +It buys nothing at all against a real panel. A real panel mints its own authority in +firmware, the integration pins per config entry with no shared trust store, and this key +signs nothing that chains to it. A real panel's entry reporting the fingerprint above +would be conclusive evidence of tampering — a check the generated-CA arrangement could +never offer. + +## Validity + +Valid from 2026-01-01 to 2126-01-01. A static CA's expiry is a same-day, fleet-wide event +for every install at once, and no replacement can ship once the simulator is archived, so +the window is set far past the point where either emulator could still be in use. + +The server certificate signed by it is *not* static: each install mints its own leaf and +key, because a leaf has to name that install's own address and hostname in its SAN. diff --git a/src/span_panel_simulator/_ca/ca.crt b/src/span_panel_simulator/_ca/ca.crt new file mode 100644 index 0000000..8f479d6 --- /dev/null +++ b/src/span_panel_simulator/_ca/ca.crt @@ -0,0 +1,20 @@ +-----BEGIN CERTIFICATE----- +MIIDMDCCAhigAwIBAgIUad8iDgDJTr6QoYDp9lM+1tuwzCswDQYJKoZIhvcNAQEL +BQAwRTEdMBsGA1UECgwUU1BBTiBQYW5lbCBFbXVsYXRvcnMxJDAiBgNVBAMMG1NQ +QU4gUGFuZWwgRW11bGF0b3IgVGVzdCBDQTAgFw0yNjAxMDEwMDAwMDBaGA8yMTI2 +MDEwMTAwMDAwMFowRTEdMBsGA1UECgwUU1BBTiBQYW5lbCBFbXVsYXRvcnMxJDAi +BgNVBAMMG1NQQU4gUGFuZWwgRW11bGF0b3IgVGVzdCBDQTCCASIwDQYJKoZIhvcN +AQEBBQADggEPADCCAQoCggEBALGBGL8YmpvmTdeChNiIJCSkxMa6AijXdouLXanJ +QyfhKDKsmmSTCgtVAdlFfqFoMz/E9QxPDg6Ej6rXerKuaOdcZax8EU1TvhHXuuYA +TTjlGuLWcEOSh7pbReFLxU7Wr38M2yiRcW1NJznv4IGaSIboqOYtD+F23Y4RtuT3 +j4xi4b9UAYY+psCHNF4jiw4MR/XAi+5wDA7RVujyrzRnfYd0XVcs69g7zGQrDiKQ +hGcu8daSixXKbLGDcq8p/rSB3liDxK7Ji38vDB5oNyzvEE0IIXWi9dqqzPBRp25g +IsQjq8ufoZ6953BmM5prnBdyFB6i5XJ+tbPEvXOKFjm9W9UCAwEAAaMWMBQwEgYD +VR0TAQH/BAgwBgEB/wIBADANBgkqhkiG9w0BAQsFAAOCAQEAqgENM+PcNvMcXqBf +vsWlysA5Fm9gGZrTgpV0iZI6ei+YoL9hQE0mLMO3/YOtoJjQbGFeW3w5vrBpQRCS +kPZ8Id1f6Ern5zyaNLF5Kz9ZGp9/VAz5NMsB0KOmRhYBLzJejLpnnazOAFiMbCxt +CetLHKMbyPSla5OoEkEhK2oPdHYcC4qvZrcNo+WMB2UHZYQejixQdQagexZgGXyE +7liPP+KngYcUV0g3XXD45rjQ0czCkxzuJ8kLsqFRMMIW/zF/R9zXG0+8WcF/qhLG +7mj4o5WXYu67qnHVJ0MQ7QZbbvNeqY/KkqgtGvLZ72YgkWy0EWU1atj1KDLSAS1e +5Kr66g== +-----END CERTIFICATE----- diff --git a/src/span_panel_simulator/_ca/ca.key b/src/span_panel_simulator/_ca/ca.key new file mode 100644 index 0000000..a4bd73f --- /dev/null +++ b/src/span_panel_simulator/_ca/ca.key @@ -0,0 +1,27 @@ +-----BEGIN RSA PRIVATE KEY----- +MIIEowIBAAKCAQEAsYEYvxiam+ZN14KE2IgkJKTExroCKNd2i4tdqclDJ+EoMqya +ZJMKC1UB2UV+oWgzP8T1DE8ODoSPqtd6sq5o51xlrHwRTVO+Ede65gBNOOUa4tZw +Q5KHultF4UvFTtavfwzbKJFxbU0nOe/ggZpIhuio5i0P4XbdjhG25PePjGLhv1QB +hj6mwIc0XiOLDgxH9cCL7nAMDtFW6PKvNGd9h3RdVyzr2DvMZCsOIpCEZy7x1pKL +FcpssYNyryn+tIHeWIPErsmLfy8MHmg3LO8QTQghdaL12qrM8FGnbmAixCOry5+h +nr3ncGYzmmucF3IUHqLlcn61s8S9c4oWOb1b1QIDAQABAoIBAASNbpCMICmqb578 +3SJej4AUbRWPCg95GIm+And1zs/0MYnnLQiHpY9319XLjKgKxPG4P/QCXQvDRkSt +RRuAZ0wJGHR4NCC7FUSzjJAUkZEe0yCJO7+H6mI9sAj2CV1FxUzUDBTkx2IfGJ1J +noTvFjgbjyVkIiRNXk5OpwBarGb/IUSAmz2+pGidzReiNQNRaxoXNXrAwKMBwBde +JMXcg5jsC5x/5E8J1sR982tsW4SEpu9C+JsPE+QRecl//gcmRKbEcU/UDnAzt8nE +74l8VYvFIGPN8ensiVePF6AOeD+sNurJr7+4bB6dp+oho+mwlx+lnaQFnNE3fyNP +BeZIH6sCgYEA5vckI0gxdHqbdIjKwDSa3UrvLYn6uNMERYf9xqOPqRamVpwvI/8b +2u7gsU2WVctzLOGki62BtLJ/LxWbMjHfC0P462W0rGLxM5RWW/I8Eg8YTbZ6TnI7 +eKJ6Kwp6XOLctrMkdvG8Uhb1xmqJhhcXuUm/5eAFGQRj0wscSE4N6fcCgYEAxL6C +Z+x8ACaQlpGM/ETHLULscWk4eHVJxAPWP46KSYw/sRpCBGzwqBKdNHB9WfVJ4KOC +6/vdjdyKMBvDAo6fgQ7xipQWLSCieG/DUlgWnxYBrFxjdhs5Pup/c2Gstqc0Ik35 +rRa7FyIVv9hZlpKxTIEOkIvCqr+goHfJiWM3VZMCgYEA3UU0B+EoEBFOUUykAZPk +MdGSZqxj2SHAW+S3CdC01ydt4FftCa1jwFFOTk6okjVBeGQMQgzN0JGZK0Cl2GbS +6nYKSWq8ttJrhiC8w4XEawCvJfSGJBVoSGBQbl+xbr+g+B+T9agN5Q8+EwMKEb6Z +6ddv0wjVmWwNPqjMRbERv20CgYAGtmlEmnDwoGA2gujIgqA5Dxt6Exj2IypNIcRh +jr9xGvPhWv/9nqnnUhtC+sOMGi5XF70TyBgbCvc8BPZ2U99lS9k53SuUhpSEf2uu +l4rOCrYr50sNcIhXeTC4AxEdej7xHK/0E952P5d2YC9OT9r0vUJ99AbERJ16XBvo +MBZnOQKBgAaV1JHmRveHj2WSKv4QAj5j7GhmZ5x7kbCBsupvOgEWl/WWIbZ0FQc4 +cASSqXrSOagLa1DeuvuPNgcrgTtECDZL5tufT3WODLqOt0KnsT4v/705S2C+qbbe +SUi0YAzIQB9rDewY9dY6UP6lJfIbX+GmaUmgkKF++tYyNwBSWS2C +-----END RSA PRIVATE KEY----- diff --git a/src/span_panel_simulator/certs.py b/src/span_panel_simulator/certs.py index 7587a54..986994d 100644 --- a/src/span_panel_simulator/certs.py +++ b/src/span_panel_simulator/certs.py @@ -1,7 +1,28 @@ -"""TLS certificate generation for the simulator. - -Generates a self-signed CA and a server certificate at startup, matching -the real SPAN panel's TLS provisioning flow. +"""TLS certificate provisioning for the emulator. + +The certificate authority is fixed and ships with the package; the server +certificate is minted per install. That split is the whole design, and it +follows from what the emulators are for. + +The simulator emulates SPAN firmware before r202633 and panelbench emulates +r202633 and later, so stopping one and starting the other rehearses a firmware +upgrade on a single panel. A firmware upgrade does not rotate a panel's +certificate authority, and a consumer that pins that authority -- as the Home +Assistant integration does -- is right to stop and ask when one changes. While +each emulator minted its own CA the swap presented a new anchor, so the +rehearsal simulated a panel substitution: the one event it must not. + +The leaf cannot be shared the same way, because it has to name the address and +hostname *this* install answers on. So each install mints its own leaf and its +own key against the packaged authority, and the packaged private key is the only +material the two repositories hold in common. See ``_ca/README.md`` for why that +key is public on purpose and what it does and does not cost. + +Nothing on disk is ever read to decide trust. ``ca.crt`` is written from the +packaged bytes on every start, so a generated anchor left in a persistent +``/data/certs`` by an older build cannot outlive an upgrade, and the certificate +this module serves, the one mosquitto trusts and the one the leaf chains to are +the same certificate by construction rather than by hope. """ from __future__ import annotations @@ -9,204 +30,370 @@ import datetime import ipaddress import logging +import os import socket from dataclasses import dataclass +from importlib import resources from typing import TYPE_CHECKING from cryptography import x509 +from cryptography.exceptions import InvalidSignature, UnsupportedAlgorithm +from cryptography.hazmat.primitives import hashes, serialization +from cryptography.hazmat.primitives.asymmetric import padding, rsa if TYPE_CHECKING: from pathlib import Path -from cryptography.hazmat.primitives import hashes, serialization -from cryptography.hazmat.primitives.asymmetric import rsa + from cryptography.x509.oid import NameOID _LOGGER = logging.getLogger(__name__) -_CA_VALIDITY_DAYS = 3650 _SERVER_VALIDITY_DAYS = 365 _KEY_SIZE = 2048 +# A leaf is re-signed once it is inside this margin of expiring, rather than +# after it already has. Re-signing is free -- the authority is constant and the +# emulator holds its key -- while an expired leaf is a silent, permanent +# outage: a pinned consumer sees a handshake failure against an unchanged +# anchor, reports it as retryable, and retries for as long as the process runs. +_LEAF_RENEWAL_MARGIN = datetime.timedelta(days=30) + +_CA_PACKAGE_DIR = "_ca" + + +class StaticCAUnavailableError(RuntimeError): + """The packaged certificate authority is missing or unreadable. + + Fatal on purpose, and the one place this module refuses to self-heal. + Absent or corrupt package data means the build or the image is broken, and + the self-healing response -- mint a fresh CA -- would hand this install a + private anchor that no other install shares, which is precisely the + divergence the packaged CA exists to prevent. Failing loudly loses a test + panel until the image is fixed; recovering quietly loses the property the + design is for, and does it invisibly. + """ + @dataclass(frozen=True, slots=True) class CertificateBundle: - """Paths to generated certificate files.""" + """Paths to the certificate files this emulator serves. + + There is no ``ca_key_path``. The authority's key is package data and is + never written into the certificate directory: nothing at runtime reads it + from there -- mosquitto is given ``cafile`` and the server pair, and the + bootstrap endpoint serves the certificate -- so writing it would publish a + world-readable private key into a persistent volume for no purpose, and add + a fourth file that could drift from the other three. + """ ca_cert_path: Path - ca_key_path: Path server_cert_path: Path server_key_path: Path ca_cert_pem: bytes -def _cert_has_san(cert_path: Path, address: str | None, dns_name: str | None) -> bool: - """Check whether an existing server certificate contains required SAN entries.""" - cert = x509.load_pem_x509_certificate(cert_path.read_bytes()) +def _packaged_ca() -> tuple[bytes, x509.Certificate, rsa.RSAPrivateKey]: + """Load the shipped authority, or refuse to start. + + Returns the certificate's PEM bytes verbatim alongside the parsed pair. The + raw bytes are what gets written to disk and served, so that what a consumer + fingerprints is the shipped file rather than a re-serialisation of it. + """ + anchor = resources.files(__package__).joinpath(_CA_PACKAGE_DIR) try: - san = cert.extensions.get_extension_for_class(x509.SubjectAlternativeName) - except x509.ExtensionNotFound: - return False - if address: - target = ipaddress.ip_address(address) - if target not in san.value.get_values_for_type(x509.IPAddress): - return False - return not (dns_name and dns_name not in san.value.get_values_for_type(x509.DNSName)) + ca_pem = anchor.joinpath("ca.crt").read_bytes() + ca_key_pem = anchor.joinpath("ca.key").read_bytes() + except (FileNotFoundError, OSError) as err: + raise StaticCAUnavailableError( + f"The packaged certificate authority is missing from {_CA_PACKAGE_DIR}/. " + "This build is incomplete; it cannot be repaired by generating one, " + "because a generated authority would not match any other install." + ) from err + + try: + ca_cert = x509.load_pem_x509_certificate(ca_pem) + ca_key = serialization.load_pem_private_key(ca_key_pem, password=None) + except (ValueError, TypeError) as err: + raise StaticCAUnavailableError( + f"The packaged certificate authority in {_CA_PACKAGE_DIR}/ cannot be read: {err}" + ) from err + + if not isinstance(ca_key, rsa.RSAPrivateKey): + raise StaticCAUnavailableError( + f"The packaged authority key in {_CA_PACKAGE_DIR}/ is {type(ca_key).__name__}, " + "not the RSA key this module signs with." + ) + return ca_pem, ca_cert, ca_key -def _load_existing( - output_dir: Path, advertise_address: str | None = None -) -> CertificateBundle | None: - """Return a CertificateBundle if all expected files already exist. +def _write_atomic(path: Path, data: bytes, *, mode: int = 0o644) -> None: + """Write a file so that a reader never sees it half-written. - When *advertise_address* is provided, the existing server certificate - is checked for a matching SAN entry. If the IP is missing the cached - certs are considered stale and ``None`` is returned so that fresh - certificates are generated. + Both mosquitto and the emulator read this directory, and an add-on killed + mid-write would otherwise leave a truncated certificate that reads back as + corrupt on the next start. Written to a temporary name in the same + directory and moved into place, which is atomic on the same filesystem. """ - ca_cert_path = output_dir / "ca.crt" - ca_key_path = output_dir / "ca.key" - server_cert_path = output_dir / "server.crt" - server_key_path = output_dir / "server.key" + tmp = path.with_name(f".{path.name}.tmp") + tmp.write_bytes(data) + os.chmod(tmp, mode) + os.replace(tmp, path) - if not all(p.exists() for p in (ca_cert_path, ca_key_path, server_cert_path, server_key_path)): - return None - container_host = socket.gethostname() - if not _cert_has_san(server_cert_path, advertise_address, container_host): - _LOGGER.info( - "Existing certificate missing SAN for %s / %s — regenerating", - advertise_address or "(no IP)", - container_host, - ) - return None +def _install_ca(output_dir: Path, ca_pem: bytes) -> Path: + """Put the packaged authority on disk, unconditionally. - _LOGGER.info("Reusing existing TLS certificates in %s", output_dir) - return CertificateBundle( - ca_cert_path=ca_cert_path, - ca_key_path=ca_key_path, - server_cert_path=server_cert_path, - server_key_path=server_key_path, - ca_cert_pem=ca_cert_path.read_bytes(), - ) + Written on every start rather than when absent, because an add-on's data + directory survives upgrades: an install that ran a build which generated + its own CA still has that file, and any rule of the form "keep what is + already there" would serve it forever. Comparing first keeps the common + case a read. + """ + ca_cert_path = output_dir / "ca.crt" + try: + existing: bytes | None = ca_cert_path.read_bytes() + except (FileNotFoundError, OSError): + existing = None + + if existing != ca_pem: + if existing: + _LOGGER.warning( + "Replacing the certificate authority in %s with the packaged one. A consumer " + "pinned to the previous authority will report that it changed, once.", + ca_cert_path, + ) + _write_atomic(ca_cert_path, ca_pem) + + # An older build wrote the authority's private key here. It is + # world-readable, nothing reads it any more, and leaving it behind invites + # somebody to sign with it. + stale_key = output_dir / "ca.key" + if stale_key.exists(): + stale_key.unlink(missing_ok=True) + _LOGGER.info("Removed the authority private key left in %s by an earlier build", stale_key) + + return ca_cert_path + + +def _issued_by(leaf: x509.Certificate, ca_cert: x509.Certificate) -> bool: + """Whether ``ca_cert`` actually signed ``leaf``, by signature. + + Cryptographic rather than a comparison of issuer names, and that is + load-bearing. Every authority these emulators have ever generated carries + the same subject, and neither certificate carries a Subject or Authority + Key Identifier, so names cannot tell two authorities apart. A name-based + check would pass for a leaf signed by a superseded CA -- exactly what sits + in the certificate directory of every install upgrading to the packaged + authority -- and the emulator would then serve a chain its own published + authority rejects, which a pinned consumer cannot recover from. + """ + public_key = ca_cert.public_key() + if not isinstance(public_key, rsa.RSAPublicKey): + return False + try: + # Read inside the guard: this raises `UnsupportedAlgorithm` for a + # signature OID `cryptography` does not know, and returns None for the + # Ed25519/Ed448 family. Neither is a leaf this authority signed, and + # neither may escape into a startup that cannot survive an exception. + algorithm = leaf.signature_hash_algorithm + if algorithm is None: + return False + public_key.verify( + leaf.signature, + leaf.tbs_certificate_bytes, + padding.PKCS1v15(), + algorithm, + ) + except (InvalidSignature, ValueError, TypeError, UnsupportedAlgorithm): + return False + return True -def generate_certificates( - output_dir: Path, - hostname: str = "span-simulator", - advertise_address: str | None = None, -) -> CertificateBundle: - """Generate a self-signed CA and server certificate. +def _key_matches(cert: x509.Certificate, key_path: Path) -> bool: + """Whether `key_path` holds the private key for `cert`. - If all certificate files already exist in *output_dir*, they are loaded - and returned without regeneration. This avoids overwriting certs that - Mosquitto is already using. + A certificate and a key from two different generations both parse, and the + mismatch surfaces only when mosquitto refuses to start with a message about + its keyfile rather than anything naming the real problem. Compared by public + key, which is the only thing that actually has to agree. + """ + try: + key = serialization.load_pem_private_key(key_path.read_bytes(), password=None) + except (FileNotFoundError, OSError, ValueError, TypeError, UnsupportedAlgorithm): + return False + # Compared as DER public bytes rather than by key-type-specific numbers, so + # that a key of some other algorithm answers False like any other mismatch + # instead of raising out of a predicate that promises never to. + encoding = serialization.Encoding.DER + fmt = serialization.PublicFormat.SubjectPublicKeyInfo + try: + return key.public_key().public_bytes(encoding, fmt) == cert.public_key().public_bytes( + encoding, fmt + ) + except (ValueError, UnsupportedAlgorithm): + return False - Args: - output_dir: Directory to write PEM files into. - hostname: Server hostname for the certificate SAN. - advertise_address: Optional IP address to include in the SAN so - that TLS clients connecting by IP pass verification. - Returns: - CertificateBundle with file paths and raw CA PEM bytes. +def _leaf_is_fit( + cert_path: Path, + key_path: Path, + ca_cert: x509.Certificate, + address: str | None, + dns_name: str | None, +) -> bool: + """Whether the leaf on disk can still be served as it stands. + + One predicate rather than a chain of special cases, because every way a + leaf can be unfit has the same remedy: sign a new one. It is unfit if it + cannot be parsed, if this authority did not sign it, if its private key is + missing or is not the one it was issued for, if it has expired or is about + to, or if it does not name the address and hostname this install answers + on. + + Every failure is answered False rather than raised. This runs at startup + under ``set -euo pipefail``; an exception here is not a diagnostic, it is a + container that will not boot. """ - existing = _load_existing(output_dir, advertise_address=advertise_address) - if existing is not None: - return existing + try: + cert = x509.load_pem_x509_certificate(cert_path.read_bytes()) + except (FileNotFoundError, OSError, ValueError): + return False - output_dir.mkdir(parents=True, exist_ok=True) + if not _issued_by(cert, ca_cert): + _LOGGER.info("The server certificate in %s was signed by another authority", cert_path) + return False - # --- CA key + certificate --- - ca_key = rsa.generate_private_key(public_exponent=65537, key_size=_KEY_SIZE) + if not _key_matches(cert, key_path): + _LOGGER.info("The server certificate in %s has no matching private key", cert_path) + return False - ca_name = x509.Name( - [ - x509.NameAttribute(NameOID.ORGANIZATION_NAME, "SPAN Simulator"), - x509.NameAttribute(NameOID.COMMON_NAME, "SPAN Simulator CA"), - ] - ) + if cert.not_valid_after_utc - _LEAF_RENEWAL_MARGIN <= datetime.datetime.now(datetime.UTC): + _LOGGER.info("The server certificate in %s has expired or is about to", cert_path) + return False - now = datetime.datetime.now(datetime.UTC) - ca_cert = ( - x509.CertificateBuilder() - .subject_name(ca_name) - .issuer_name(ca_name) - .public_key(ca_key.public_key()) - .serial_number(x509.random_serial_number()) - .not_valid_before(now) - .not_valid_after(now + datetime.timedelta(days=_CA_VALIDITY_DAYS)) - .add_extension(x509.BasicConstraints(ca=True, path_length=0), critical=True) - .sign(ca_key, hashes.SHA256()) - ) + try: + san = cert.extensions.get_extension_for_class(x509.SubjectAlternativeName) + except x509.ExtensionNotFound: + return False - ca_cert_path = output_dir / "ca.crt" - ca_key_path = output_dir / "ca.key" + if address: + try: + target = ipaddress.ip_address(address) + except ValueError: + # Not an address at all. Nothing can be asserted about a SAN entry + # for it, and inventing one would put a name in the certificate + # that the emulator was never reached by. + _LOGGER.warning("Ignoring advertise address %r: not an IP address", address) + else: + if target not in san.value.get_values_for_type(x509.IPAddress): + return False - ca_cert_pem = ca_cert.public_bytes(serialization.Encoding.PEM) - ca_cert_path.write_bytes(ca_cert_pem) - ca_key_path.write_bytes( - ca_key.private_bytes( - serialization.Encoding.PEM, - serialization.PrivateFormat.TraditionalOpenSSL, - serialization.NoEncryption(), - ) - ) + return not (dns_name and dns_name not in san.value.get_values_for_type(x509.DNSName)) - # --- Server key + certificate --- + +def _sign_server_cert( + output_dir: Path, + ca_cert: x509.Certificate, + ca_key: rsa.RSAPrivateKey, + hostname: str, + advertise_address: str | None, +) -> tuple[Path, Path]: + """Mint a server certificate and key against the packaged authority.""" server_key = rsa.generate_private_key(public_exponent=65537, key_size=_KEY_SIZE) - server_name = x509.Name( - [ - x509.NameAttribute(NameOID.ORGANIZATION_NAME, "SPAN Simulator"), - x509.NameAttribute(NameOID.COMMON_NAME, hostname), - ] - ) + names: list[x509.GeneralName] = [ + x509.DNSName(hostname), + x509.DNSName("localhost"), + x509.DNSName(socket.gethostname()), + x509.IPAddress(ipaddress.ip_address("127.0.0.1")), + ] + if advertise_address: + try: + names.append(x509.IPAddress(ipaddress.ip_address(advertise_address))) + except ValueError: + _LOGGER.warning("Ignoring advertise address %r: not an IP address", advertise_address) + now = datetime.datetime.now(datetime.UTC) server_cert = ( x509.CertificateBuilder() - .subject_name(server_name) - .issuer_name(ca_name) + .subject_name( + x509.Name( + [ + x509.NameAttribute(NameOID.ORGANIZATION_NAME, "SPAN Simulator"), + x509.NameAttribute(NameOID.COMMON_NAME, hostname), + ] + ) + ) + .issuer_name(ca_cert.subject) .public_key(server_key.public_key()) .serial_number(x509.random_serial_number()) .not_valid_before(now) .not_valid_after(now + datetime.timedelta(days=_SERVER_VALIDITY_DAYS)) - .add_extension( - x509.SubjectAlternativeName( - [ - x509.DNSName(hostname), - x509.DNSName("localhost"), - x509.DNSName(socket.gethostname()), - x509.IPAddress(ipaddress.ip_address("127.0.0.1")), - ] - + ( - [x509.IPAddress(ipaddress.ip_address(advertise_address))] - if advertise_address - else [] - ) - ), - critical=False, - ) + .add_extension(x509.SubjectAlternativeName(names), critical=False) .sign(ca_key, hashes.SHA256()) ) server_cert_path = output_dir / "server.crt" server_key_path = output_dir / "server.key" - - server_cert_path.write_bytes(server_cert.public_bytes(serialization.Encoding.PEM)) - server_key_path.write_bytes( + _write_atomic(server_cert_path, server_cert.public_bytes(serialization.Encoding.PEM)) + _write_atomic( + server_key_path, server_key.private_bytes( serialization.Encoding.PEM, serialization.PrivateFormat.TraditionalOpenSSL, serialization.NoEncryption(), - ) + ), ) + return server_cert_path, server_key_path + + +def generate_certificates( + output_dir: Path, + hostname: str = "span-simulator", + advertise_address: str | None = None, +) -> CertificateBundle: + """Install the packaged authority and ensure a usable server certificate. - _LOGGER.info("Generated TLS certificates in %s", output_dir) + The authority is written every time; the server certificate is reused when + it is still fit to serve and re-signed otherwise. A leaf that has gone + stale -- because the advertised address changed, because it expired, or + because it was signed by an authority an older build generated -- costs a + new leaf and nothing else. It never costs a new authority, which is what + a consumer pinned to this panel would have to be asked about. + + Args: + output_dir: Directory to write the certificate files into. + hostname: Server hostname for the certificate's SAN. + advertise_address: Optional IP address to include in the SAN so that + TLS clients connecting by IP pass verification. + + Returns: + CertificateBundle with the file paths and the authority's PEM bytes. + + Raises: + StaticCAUnavailableError: the packaged authority is missing or unreadable. + """ + output_dir.mkdir(parents=True, exist_ok=True) + + ca_pem, ca_cert, ca_key = _packaged_ca() + ca_cert_path = _install_ca(output_dir, ca_pem) + + server_cert_path = output_dir / "server.crt" + server_key_path = output_dir / "server.key" + + if _leaf_is_fit( + server_cert_path, server_key_path, ca_cert, advertise_address, socket.gethostname() + ): + _LOGGER.info("Reusing the server certificate in %s", output_dir) + else: + server_cert_path, server_key_path = _sign_server_cert( + output_dir, ca_cert, ca_key, hostname, advertise_address + ) + _LOGGER.info("Signed a server certificate in %s", output_dir) return CertificateBundle( ca_cert_path=ca_cert_path, - ca_key_path=ca_key_path, server_cert_path=server_cert_path, server_key_path=server_key_path, - ca_cert_pem=ca_cert_pem, + ca_cert_pem=ca_pem, ) diff --git a/tests/test_certs.py b/tests/test_certs.py index 622c498..43c56e5 100644 --- a/tests/test_certs.py +++ b/tests/test_certs.py @@ -1,23 +1,463 @@ -"""Tests for TLS certificate generation.""" +"""Tests for TLS certificate provisioning. + +The property under test throughout is that the certificate authority is fixed +and the server certificate is not. A consumer that pins the authority -- the +Home Assistant integration does -- treats a change as worth stopping for, so +every path that re-signs a leaf must leave the anchor alone, and the anchor +served must always be the packaged one. +""" from __future__ import annotations -from typing import TYPE_CHECKING +import base64 +import datetime +import hashlib +import ipaddress +import socket +from pathlib import Path + +import pytest +from cryptography import x509 +from cryptography.hazmat.primitives import hashes, serialization +from cryptography.hazmat.primitives.asymmetric import ed25519, padding, rsa +from cryptography.x509.oid import NameOID + +from span_panel_simulator.certs import ( + CertificateBundle, + StaticCAUnavailableError, + generate_certificates, +) + +# The published identity of the packaged authority, over the certificate's DER +# bytes -- the same value the integration pins, reports under `panel_ca` in +# diagnostics and shows in a certificate-authority-changed repair. +# +# Hardcoded here on purpose. This is the one form of drift no runtime check can +# catch: the simulator and panelbench each carry a copy of these bytes, and a +# well-meaning "refresh the certs" in either repository would fork the two +# fleets silently. Both repositories assert the same constant, so such a change +# fails a test instead of shipping. +STATIC_CA_FINGERPRINT = "3cf8c14a78900b8736870c95adcc931cdcb3a51bc3029c96efafd0a4cb790d97" + + +def _fingerprint(pem: bytes) -> str: + cert = x509.load_pem_x509_certificate(pem) + return hashlib.sha256(cert.public_bytes(serialization.Encoding.DER)).hexdigest() + + +def _foreign_ca() -> tuple[x509.Certificate, rsa.RSAPrivateKey]: + """Mint an authority carrying the subject every generated CA used to have. + + Deliberately named identically to the authority older builds generated, so + that a check comparing issuer names cannot tell it from the packaged one. + """ + key = rsa.generate_private_key(public_exponent=65537, key_size=2048) + name = x509.Name( + [ + x509.NameAttribute(NameOID.ORGANIZATION_NAME, "SPAN Simulator"), + x509.NameAttribute(NameOID.COMMON_NAME, "SPAN Simulator CA"), + ] + ) + now = datetime.datetime.now(datetime.UTC) + cert = ( + x509.CertificateBuilder() + .subject_name(name) + .issuer_name(name) + .public_key(key.public_key()) + .serial_number(x509.random_serial_number()) + .not_valid_before(now - datetime.timedelta(days=1)) + .not_valid_after(now + datetime.timedelta(days=3650)) + .add_extension(x509.BasicConstraints(ca=True, path_length=0), critical=True) + .sign(key, hashes.SHA256()) + ) + return cert, key + + +def _leaf_signed_by( + ca_cert: x509.Certificate, + ca_key: rsa.RSAPrivateKey, + *, + names: list[x509.GeneralName] | None = None, + not_after: datetime.datetime | None = None, +) -> tuple[bytes, bytes]: + key = rsa.generate_private_key(public_exponent=65537, key_size=2048) + now = datetime.datetime.now(datetime.UTC) + # An already-expired leaf still needs a validity window that makes sense, + # so back-date the start rather than letting it collide with the end. + not_before = min( + now - datetime.timedelta(days=1), (not_after or now) - datetime.timedelta(days=1) + ) + cert = ( + x509.CertificateBuilder() + .subject_name(x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "span-simulator")])) + .issuer_name(ca_cert.subject) + .public_key(key.public_key()) + .serial_number(x509.random_serial_number()) + .not_valid_before(not_before) + .not_valid_after(not_after or (now + datetime.timedelta(days=365))) + .add_extension( + x509.SubjectAlternativeName( + names + if names is not None + else [ + x509.DNSName("span-simulator"), + x509.DNSName("localhost"), + x509.DNSName(socket.gethostname()), + x509.IPAddress(ipaddress.ip_address("127.0.0.1")), + ] + ), + critical=False, + ) + .sign(ca_key, hashes.SHA256()) + ) + return ( + cert.public_bytes(serialization.Encoding.PEM), + key.private_bytes( + serialization.Encoding.PEM, + serialization.PrivateFormat.TraditionalOpenSSL, + serialization.NoEncryption(), + ), + ) + + +def _leaf_of(bundle: CertificateBundle) -> x509.Certificate: + return x509.load_pem_x509_certificate(bundle.server_cert_path.read_bytes()) + + +class TestPackagedAuthority: + def test_fingerprint_is_the_published_constant(self, tmp_path: Path) -> None: + """Guards the one drift no runtime check can catch: the two repos forking.""" + bundle = generate_certificates(tmp_path / "certs") + + assert _fingerprint(bundle.ca_cert_pem) == STATIC_CA_FINGERPRINT + + def test_packaged_key_pairs_with_packaged_certificate(self, tmp_path: Path) -> None: + """A mismatched pair would sign leaves the served authority rejects.""" + bundle = generate_certificates(tmp_path / "certs") + leaf = _leaf_of(bundle) + ca = x509.load_pem_x509_certificate(bundle.ca_cert_pem) + + # The leaf was signed with the packaged key; verifying it against the + # packaged certificate's public key proves the two are a pair. + ca.public_key().verify( # type: ignore[union-attr] + leaf.signature, + leaf.tbs_certificate_bytes, + padding.PKCS1v15(), + leaf.signature_hash_algorithm, + ) + + def test_authority_does_not_expire_within_the_century(self, tmp_path: Path) -> None: + bundle = generate_certificates(tmp_path / "certs") + ca = x509.load_pem_x509_certificate(bundle.ca_cert_pem) + + assert ca.not_valid_after_utc.year == 2126 + + def test_authority_private_key_is_never_written_to_disk(self, tmp_path: Path) -> None: + """It is package data; writing it would publish it into a data volume.""" + certs = tmp_path / "certs" + generate_certificates(certs) + + assert not (certs / "ca.key").exists() + + def test_missing_package_data_is_fatal_not_self_healing( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + """Generating a replacement would mint an anchor no other install shares.""" + import span_panel_simulator.certs as certs_mod + + def _absent(_package: str) -> Path: + return tmp_path / "does-not-exist" + + monkeypatch.setattr(certs_mod.resources, "files", _absent) + + with pytest.raises(StaticCAUnavailableError): + generate_certificates(tmp_path / "certs") + + +class TestAuthorityIsInstalledUnconditionally: + def test_a_generated_authority_left_behind_is_replaced(self, tmp_path: Path) -> None: + """An add-on's data directory survives upgrades; a stale anchor must not.""" + certs = tmp_path / "certs" + certs.mkdir() + foreign_cert, _ = _foreign_ca() + (certs / "ca.crt").write_bytes(foreign_cert.public_bytes(serialization.Encoding.PEM)) + + bundle = generate_certificates(certs) + + assert _fingerprint((certs / "ca.crt").read_bytes()) == STATIC_CA_FINGERPRINT + assert _fingerprint(bundle.ca_cert_pem) == STATIC_CA_FINGERPRINT + + def test_a_stale_authority_key_is_removed(self, tmp_path: Path) -> None: + """Left by an older build, world-readable, and now signs nothing.""" + certs = tmp_path / "certs" + certs.mkdir() + (certs / "ca.key").write_bytes(b"-----BEGIN RSA PRIVATE KEY-----\nstale\n") + + generate_certificates(certs) + + assert not (certs / "ca.key").exists() + + def test_a_truncated_authority_file_is_overwritten(self, tmp_path: Path) -> None: + certs = tmp_path / "certs" + certs.mkdir() + (certs / "ca.crt").write_bytes(b"-----BEGIN CERTIFICATE-----\ntrunc") + + generate_certificates(certs) + + assert _fingerprint((certs / "ca.crt").read_bytes()) == STATIC_CA_FINGERPRINT + + +class TestLeafIsResignedWithoutTouchingTheAuthority: + """The original defect: a stale leaf rotated the anchor along with it.""" + + def test_a_new_advertise_address_resigns_only_the_leaf(self, tmp_path: Path) -> None: + certs = tmp_path / "certs" + first = generate_certificates(certs, advertise_address="10.0.0.5") + first_leaf = _leaf_of(first).serial_number + + second = generate_certificates(certs, advertise_address="10.0.0.9") + + assert _fingerprint(second.ca_cert_pem) == STATIC_CA_FINGERPRINT + assert _fingerprint(first.ca_cert_pem) == _fingerprint(second.ca_cert_pem) + assert _leaf_of(second).serial_number != first_leaf + san = _leaf_of(second).extensions.get_extension_for_class(x509.SubjectAlternativeName) + assert ipaddress.ip_address("10.0.0.9") in san.value.get_values_for_type(x509.IPAddress) -from span_panel_simulator.certs import generate_certificates + def test_an_unchanged_address_reuses_the_leaf(self, tmp_path: Path) -> None: + certs = tmp_path / "certs" + first = generate_certificates(certs, advertise_address="10.0.0.5") + first_leaf = _leaf_of(first).serial_number -if TYPE_CHECKING: - from pathlib import Path + second = generate_certificates(certs, advertise_address="10.0.0.5") + assert _leaf_of(second).serial_number == first_leaf -class TestCertificateGeneration: - """Verify certificate bundle creation.""" + def test_a_leaf_signed_by_a_superseded_authority_is_resigned(self, tmp_path: Path) -> None: + """The upgrade path every existing install takes. - def test_generates_all_files(self, tmp_path: Path) -> None: + The superseded authority carries the same subject as the packaged one, + so a check comparing issuer names would keep this leaf and leave the + emulator serving a chain its own published authority rejects -- which a + pinned consumer cannot recover from. + """ + certs = tmp_path / "certs" + certs.mkdir() + foreign_cert, foreign_key = _foreign_ca() + leaf_pem, leaf_key_pem = _leaf_signed_by(foreign_cert, foreign_key) + (certs / "server.crt").write_bytes(leaf_pem) + (certs / "server.key").write_bytes(leaf_key_pem) + + bundle = generate_certificates(certs) + + ca = x509.load_pem_x509_certificate(bundle.ca_cert_pem) + leaf = _leaf_of(bundle) + assert leaf.public_bytes(serialization.Encoding.PEM) != leaf_pem + # The served chain validates against the served authority. + ca.public_key().verify( # type: ignore[union-attr] + leaf.signature, + leaf.tbs_certificate_bytes, + padding.PKCS1v15(), + leaf.signature_hash_algorithm, + ) + + def test_an_expired_leaf_is_resigned(self, tmp_path: Path) -> None: + certs = tmp_path / "certs" + first = generate_certificates(certs) + expired = datetime.datetime.now(datetime.UTC) - datetime.timedelta(days=1) + ca = x509.load_pem_x509_certificate(first.ca_cert_pem) + # Re-sign with the packaged key is not possible from here, so stand in + # a foreign expired leaf: unfit for two reasons, re-signed for either. + foreign_cert, foreign_key = _foreign_ca() + leaf_pem, _ = _leaf_signed_by(foreign_cert, foreign_key, not_after=expired) + (certs / "server.crt").write_bytes(leaf_pem) + + second = generate_certificates(certs) + + assert _leaf_of(second).not_valid_after_utc > datetime.datetime.now(datetime.UTC) + assert _fingerprint(second.ca_cert_pem) == _fingerprint(first.ca_cert_pem) + assert ca.subject == _leaf_of(second).issuer + + def test_an_unparseable_leaf_is_resigned_rather_than_raising(self, tmp_path: Path) -> None: + """This runs under `set -euo pipefail`; a raise is a boot loop.""" + certs = tmp_path / "certs" + certs.mkdir() + (certs / "server.crt").write_bytes(b"-----BEGIN CERTIFICATE-----\nnot a cert\n") + + bundle = generate_certificates(certs) + + assert _leaf_of(bundle).subject is not None + + def test_a_key_that_does_not_match_the_leaf_forces_a_resign(self, tmp_path: Path) -> None: + """A torn write can leave a certificate and a key from two generations. + + Both parse, so nothing here objects; the mismatch surfaces only when + mosquitto refuses to start with a message about its keyfile, naming + nothing that would lead anyone to the real problem. + """ + certs = tmp_path / "certs" + first = generate_certificates(certs) + first_leaf = _leaf_of(first).serial_number + stray = rsa.generate_private_key(public_exponent=65537, key_size=2048) + (certs / "server.key").write_bytes( + stray.private_bytes( + serialization.Encoding.PEM, + serialization.PrivateFormat.TraditionalOpenSSL, + serialization.NoEncryption(), + ) + ) + + second = generate_certificates(certs) + + assert _leaf_of(second).serial_number != first_leaf + # The pair that ends up on disk actually is a pair. + key = serialization.load_pem_private_key( + (certs / "server.key").read_bytes(), password=None + ) + assert key.public_key().public_numbers() == _leaf_of(second).public_key().public_numbers() + + def test_a_missing_leaf_key_forces_a_resign(self, tmp_path: Path) -> None: + """A certificate without its key cannot be served.""" + certs = tmp_path / "certs" + first = generate_certificates(certs) + first_leaf = _leaf_of(first).serial_number + (certs / "server.key").unlink() + + second = generate_certificates(certs) + + assert (certs / "server.key").exists() + assert _leaf_of(second).serial_number != first_leaf + + +class TestForeignKeyMaterialIsAnsweredNotRaised: + """`_leaf_is_fit` promises never to raise; startup cannot survive one. + + Both paths here run under `set -euo pipefail`, where an exception is not a + diagnostic but an add-on that will not boot. + """ + + def test_a_server_key_of_another_algorithm_forces_a_resign(self, tmp_path: Path) -> None: + certs = tmp_path / "certs" + first = generate_certificates(certs) + first_leaf = _leaf_of(first).serial_number + (certs / "server.key").write_bytes( + ed25519.Ed25519PrivateKey.generate().private_bytes( + serialization.Encoding.PEM, + serialization.PrivateFormat.PKCS8, + serialization.NoEncryption(), + ) + ) + + second = generate_certificates(certs) + + assert _leaf_of(second).serial_number != first_leaf + + def test_a_leaf_with_an_unknown_signature_oid_forces_a_resign(self, tmp_path: Path) -> None: + """`signature_hash_algorithm` raises for an OID cryptography cannot name. + + Distinct from the Ed25519 case below, which returns None rather than + raising. This is the arm that would escape the predicate and take + startup with it, so it is driven directly rather than assumed. + + Built by replacing the sha256WithRSAEncryption OID with an unassigned + arc of the same length, which keeps every DER length prefix valid; the + certificate still parses, and the algorithm is only rejected on access. + """ + certs = tmp_path / "certs" + first = generate_certificates(certs) + first_leaf = _leaf_of(first).serial_number + + der = _leaf_of(first).public_bytes(serialization.Encoding.DER) + mutated = der.replace( + bytes.fromhex("06092a864886f70d01010b"), # sha256WithRSAEncryption + bytes.fromhex("06092a864886f70d01017f"), # unassigned, same length + ) + assert mutated != der, "the signature OID should have been present to patch" + (certs / "server.crt").write_bytes( + b"-----BEGIN CERTIFICATE-----\n" + + base64.encodebytes(mutated) + + b"-----END CERTIFICATE-----\n" + ) + + # The contract is "unfit, re-sign" -- never an exception out of startup. + second = generate_certificates(certs) + + assert _leaf_of(second).serial_number != first_leaf + assert _fingerprint(second.ca_cert_pem) == STATIC_CA_FINGERPRINT + + def test_a_leaf_signed_with_ed25519_forces_a_resign(self, tmp_path: Path) -> None: + """`signature_hash_algorithm` is None for Ed25519, not an error to raise.""" + certs = tmp_path / "certs" + certs.mkdir() + ca_key = ed25519.Ed25519PrivateKey.generate() + name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "SPAN Simulator CA")]) + now = datetime.datetime.now(datetime.UTC) + ca = ( + x509.CertificateBuilder() + .subject_name(name) + .issuer_name(name) + .public_key(ca_key.public_key()) + .serial_number(x509.random_serial_number()) + .not_valid_before(now - datetime.timedelta(days=1)) + .not_valid_after(now + datetime.timedelta(days=365)) + .add_extension(x509.BasicConstraints(ca=True, path_length=0), critical=True) + .sign(ca_key, None) + ) + leaf_key = ed25519.Ed25519PrivateKey.generate() + leaf = ( + x509.CertificateBuilder() + .subject_name(x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "panelbench")])) + .issuer_name(name) + .public_key(leaf_key.public_key()) + .serial_number(x509.random_serial_number()) + .not_valid_before(now - datetime.timedelta(days=1)) + .not_valid_after(now + datetime.timedelta(days=365)) + .add_extension( + x509.SubjectAlternativeName([x509.DNSName("localhost")]), critical=False + ) + .sign(ca_key, None) + ) + (certs / "server.crt").write_bytes(leaf.public_bytes(serialization.Encoding.PEM)) + (certs / "server.key").write_bytes( + leaf_key.private_bytes( + serialization.Encoding.PEM, + serialization.PrivateFormat.PKCS8, + serialization.NoEncryption(), + ) + ) + + bundle = generate_certificates(certs) + + assert _fingerprint(bundle.ca_cert_pem) == STATIC_CA_FINGERPRINT + assert _leaf_of(bundle).signature_hash_algorithm is not None + assert ca.subject != _leaf_of(bundle).issuer + + +class TestSanEdgeCases: + def test_a_non_ip_advertise_address_does_not_crash(self, tmp_path: Path) -> None: + """`ipaddress.ip_address` used to raise straight out of startup.""" + bundle = generate_certificates(tmp_path / "certs", advertise_address="not-an-ip") + + assert bundle.server_cert_path.exists() + + def test_a_non_ip_advertise_address_is_not_named(self, tmp_path: Path) -> None: + bundle = generate_certificates(tmp_path / "certs", advertise_address="not-an-ip") + san = _leaf_of(bundle).extensions.get_extension_for_class(x509.SubjectAlternativeName) + + assert "not-an-ip" not in san.value.get_values_for_type(x509.DNSName) + + def test_the_hostname_and_loopback_are_always_named(self, tmp_path: Path) -> None: + bundle = generate_certificates(tmp_path / "certs") + san = _leaf_of(bundle).extensions.get_extension_for_class(x509.SubjectAlternativeName) + + assert "localhost" in san.value.get_values_for_type(x509.DNSName) + assert ipaddress.ip_address("127.0.0.1") in san.value.get_values_for_type(x509.IPAddress) + + +class TestBundleShape: + def test_generates_the_files_it_reports(self, tmp_path: Path) -> None: bundle = generate_certificates(tmp_path / "certs") assert bundle.ca_cert_path.exists() - assert bundle.ca_key_path.exists() assert bundle.server_cert_path.exists() assert bundle.server_key_path.exists() @@ -30,5 +470,12 @@ def test_ca_pem_is_valid(self, tmp_path: Path) -> None: def test_creates_output_directory(self, tmp_path: Path) -> None: target = tmp_path / "nested" / "deep" / "certs" bundle = generate_certificates(target) + assert target.exists() assert bundle.ca_cert_path.parent == target + + def test_no_temporary_files_are_left_behind(self, tmp_path: Path) -> None: + certs = tmp_path / "certs" + generate_certificates(certs) + + assert not [p for p in certs.iterdir() if p.name.startswith(".")] diff --git a/uv.lock b/uv.lock index d1606a2..bd7a8b3 100644 --- a/uv.lock +++ b/uv.lock @@ -912,7 +912,7 @@ wheels = [ [[package]] name = "span-panel-simulator" -version = "1.0.17" +version = "1.1.0" source = { editable = "." } dependencies = [ { name = "aiohttp" },