From 4cca6d1c822c831ee310fcc18c7f5e36ac069fba Mon Sep 17 00:00:00 2001 From: Brian Krabach Date: Fri, 3 Apr 2026 21:36:38 -0700 Subject: [PATCH] feat: add muxplex.tls module with self-signed cert generation and inspection - Add muxplex/tls.py with generate_self_signed() and get_cert_info() - generate_self_signed() creates RSA 2048-bit key + X.509 cert with SAN entries (DNS names + 127.0.0.1 + ::1), sets key perms to 0o600, creates parent dirs, returns metadata dict - get_cert_info() inspects PEM certs, returns expires/not_before/ hostnames/serial, returns None for missing/unreadable files - Add muxplex/tests/test_tls.py with 9 tests covering all acceptance criteria - Install cryptography library dependency Co-authored-by: Amplifier --- muxplex/tests/test_tls.py | 158 +++++++++++++++++++++++++++++++++++++ muxplex/tls.py | 161 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 319 insertions(+) create mode 100644 muxplex/tests/test_tls.py create mode 100644 muxplex/tls.py diff --git a/muxplex/tests/test_tls.py b/muxplex/tests/test_tls.py new file mode 100644 index 0000000..6fc0632 --- /dev/null +++ b/muxplex/tests/test_tls.py @@ -0,0 +1,158 @@ +""" +Tests for muxplex/tls.py — TLS certificate generation and inspection. +9 tests covering generate_self_signed() and get_cert_info(). +""" + +import stat + + +# --------------------------------------------------------------------------- +# 1. Importability +# --------------------------------------------------------------------------- + + +def test_tls_module_importable(): + """muxplex.tls must be importable.""" + import muxplex.tls # noqa: F401 + + assert hasattr(muxplex.tls, "generate_self_signed") + assert hasattr(muxplex.tls, "get_cert_info") + + +# --------------------------------------------------------------------------- +# 2–7. generate_self_signed() tests +# --------------------------------------------------------------------------- + + +def test_generate_self_signed_creates_cert_and_key(tmp_path): + """generate_self_signed() must create cert.pem and key.pem at the given paths.""" + from muxplex.tls import generate_self_signed + + cert_path = tmp_path / "cert.pem" + key_path = tmp_path / "key.pem" + + generate_self_signed(cert_path, key_path, hostnames=["localhost"]) + + assert cert_path.exists(), "cert.pem was not created" + assert key_path.exists(), "key.pem was not created" + + +def test_generate_self_signed_cert_is_valid_pem(tmp_path): + """Generated cert must start with '-----BEGIN CERTIFICATE-----'.""" + from muxplex.tls import generate_self_signed + + cert_path = tmp_path / "cert.pem" + key_path = tmp_path / "key.pem" + + generate_self_signed(cert_path, key_path, hostnames=["localhost"]) + + content = cert_path.read_text() + assert content.startswith("-----BEGIN CERTIFICATE-----"), ( + f"cert.pem must start with '-----BEGIN CERTIFICATE-----', got: {content[:50]!r}" + ) + + +def test_generate_self_signed_key_is_valid_pem(tmp_path): + """Generated key must start with '-----BEGIN'.""" + from muxplex.tls import generate_self_signed + + cert_path = tmp_path / "cert.pem" + key_path = tmp_path / "key.pem" + + generate_self_signed(cert_path, key_path, hostnames=["localhost"]) + + content = key_path.read_text() + assert content.startswith("-----BEGIN"), ( + f"key.pem must start with '-----BEGIN', got: {content[:50]!r}" + ) + + +def test_generate_self_signed_key_permissions(tmp_path): + """Key file must have permissions 0o600 (owner read/write only).""" + from muxplex.tls import generate_self_signed + + cert_path = tmp_path / "cert.pem" + key_path = tmp_path / "key.pem" + + generate_self_signed(cert_path, key_path, hostnames=["localhost"]) + + mode = stat.S_IMODE(key_path.stat().st_mode) + assert mode == 0o600, f"key.pem permissions must be 0o600, got: {oct(mode)}" + + +def test_generate_self_signed_returns_metadata(tmp_path): + """generate_self_signed() must return a dict with method, cert_path, key_path, hostnames, expires.""" + from muxplex.tls import generate_self_signed + + cert_path = tmp_path / "cert.pem" + key_path = tmp_path / "key.pem" + hostnames = ["myhost", "myhost.local", "localhost"] + + result = generate_self_signed(cert_path, key_path, hostnames=hostnames) + + assert isinstance(result, dict), "generate_self_signed() must return a dict" + assert result.get("method") == "selfsigned", ( + f"method must be 'selfsigned', got: {result.get('method')!r}" + ) + assert ( + result.get("cert_path") == str(cert_path) + or result.get("cert_path") == cert_path + ), f"cert_path missing or wrong in result: {result.get('cert_path')!r}" + assert ( + result.get("key_path") == str(key_path) or result.get("key_path") == key_path + ), f"key_path missing or wrong in result: {result.get('key_path')!r}" + assert isinstance(result.get("hostnames"), list) and len(result["hostnames"]) > 0, ( + f"hostnames must be a non-empty list, got: {result.get('hostnames')!r}" + ) + assert "expires" in result, "expires key must be in result" + + +def test_generate_self_signed_creates_parent_dirs(tmp_path): + """generate_self_signed() must create parent directories if they don't exist.""" + from muxplex.tls import generate_self_signed + + cert_path = tmp_path / "a" / "b" / "cert.pem" + key_path = tmp_path / "a" / "b" / "key.pem" + + # Parent dirs do not exist yet + assert not cert_path.parent.exists() + + generate_self_signed(cert_path, key_path, hostnames=["localhost"]) + + assert cert_path.exists(), "cert.pem was not created (parent dirs not created)" + assert key_path.exists(), "key.pem was not created (parent dirs not created)" + + +# --------------------------------------------------------------------------- +# 8–9. get_cert_info() tests +# --------------------------------------------------------------------------- + + +def test_get_cert_info_returns_expiry(tmp_path): + """get_cert_info() must return a dict with expires and hostnames for a valid cert.""" + from muxplex.tls import generate_self_signed, get_cert_info + + cert_path = tmp_path / "cert.pem" + key_path = tmp_path / "key.pem" + generate_self_signed(cert_path, key_path, hostnames=["testhost", "localhost"]) + + info = get_cert_info(cert_path) + + assert info is not None, "get_cert_info() must not return None for a valid cert" + assert isinstance(info, dict), "get_cert_info() must return a dict" + assert "expires" in info, "get_cert_info() result must have 'expires' key" + assert "hostnames" in info, "get_cert_info() result must have 'hostnames' key" + assert isinstance(info["hostnames"], list), "hostnames must be a list" + + +def test_get_cert_info_returns_none_for_missing_file(tmp_path): + """get_cert_info() must return None for a missing or unreadable file.""" + from muxplex.tls import get_cert_info + + missing_path = tmp_path / "nonexistent.pem" + + result = get_cert_info(missing_path) + + assert result is None, ( + f"get_cert_info() must return None for missing file, got: {result!r}" + ) diff --git a/muxplex/tls.py b/muxplex/tls.py new file mode 100644 index 0000000..f79cfe6 --- /dev/null +++ b/muxplex/tls.py @@ -0,0 +1,161 @@ +""" +muxplex/tls.py — TLS certificate generation and inspection. + +Provides: + generate_self_signed(cert_path, key_path, hostnames=None, days_valid=3650) + get_cert_info(cert_path) +""" + +import ipaddress +import socket +from datetime import datetime, timezone +from pathlib import Path + + +def _default_hostnames() -> list[str]: + hostname = socket.gethostname() + return [hostname, f"{hostname}.local", "localhost"] + + +def generate_self_signed( + cert_path, + key_path, + hostnames: list[str] | None = None, + days_valid: int = 3650, +) -> dict: + """Generate a self-signed TLS certificate and private key. + + Args: + cert_path: Destination path for the certificate PEM file. + key_path: Destination path for the private key PEM file. + hostnames: DNS names to include. Defaults to [hostname, hostname.local, localhost]. + days_valid: Certificate validity period in days. Default 3650 (≈10 years). + + Returns: + dict with keys: method, cert_path, key_path, hostnames, expires. + """ + from cryptography import x509 + from cryptography.hazmat.primitives import hashes, serialization + from cryptography.hazmat.primitives.asymmetric import rsa + from cryptography.x509.oid import NameOID + + if hostnames is None: + hostnames = _default_hostnames() + + cert_path = Path(cert_path) + key_path = Path(key_path) + + # Create parent directories if needed + cert_path.parent.mkdir(parents=True, exist_ok=True) + key_path.parent.mkdir(parents=True, exist_ok=True) + + # Generate RSA 2048-bit private key + private_key = rsa.generate_private_key( + public_exponent=65537, + key_size=2048, + ) + + # Build subject / issuer with CN = first hostname, O = muxplex + subject = issuer = x509.Name( + [ + x509.NameAttribute(NameOID.COMMON_NAME, hostnames[0]), + x509.NameAttribute(NameOID.ORGANIZATION_NAME, "muxplex"), + ] + ) + + # Build SAN extension: DNS names for all hostnames + loopback IPs + san_entries: list[x509.GeneralName] = [x509.DNSName(h) for h in hostnames] + san_entries.append(x509.IPAddress(ipaddress.IPv4Address("127.0.0.1"))) + san_entries.append(x509.IPAddress(ipaddress.IPv6Address("::1"))) + + now = datetime.now(timezone.utc) + + # Build the certificate + cert = ( + x509.CertificateBuilder() + .subject_name(subject) + .issuer_name(issuer) + .public_key(private_key.public_key()) + .serial_number(x509.random_serial_number()) + .not_valid_before(now) + .not_valid_after( + datetime.fromtimestamp( + now.timestamp() + days_valid * 86400, + tz=timezone.utc, + ) + ) + .add_extension( + x509.SubjectAlternativeName(san_entries), + critical=False, + ) + .sign(private_key, hashes.SHA256()) + ) + + expires = cert.not_valid_after_utc # type: ignore[attr-defined] + + # Write key PEM — create file with restricted permissions before writing + key_pem = private_key.private_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PrivateFormat.TraditionalOpenSSL, + encryption_algorithm=serialization.NoEncryption(), + ) + key_path.touch(mode=0o600, exist_ok=True) + key_path.write_bytes(key_pem) + key_path.chmod(0o600) + + # Write cert PEM + cert_pem = cert.public_bytes(serialization.Encoding.PEM) + cert_path.write_bytes(cert_pem) + + return { + "method": "selfsigned", + "cert_path": str(cert_path), + "key_path": str(key_path), + "hostnames": hostnames, + "expires": expires, + } + + +def get_cert_info(cert_path) -> dict | None: + """Inspect a PEM certificate and return metadata. + + Args: + cert_path: Path to the PEM certificate file. + + Returns: + dict with expires, not_before, hostnames (DNS names + IPs from SANs), serial. + Returns None if the file is missing or cannot be parsed. + """ + from cryptography import x509 + from cryptography.x509.extensions import ExtensionNotFound + + cert_path = Path(cert_path) + + try: + pem_data = cert_path.read_bytes() + except (FileNotFoundError, PermissionError, OSError): + return None + + try: + cert = x509.load_pem_x509_certificate(pem_data) + except Exception: + return None + + # Extract hostnames from SAN extension + hostnames: list[str] = [] + try: + san = cert.extensions.get_extension_for_class(x509.SubjectAlternativeName) + for entry in san.value: + if isinstance(entry, x509.DNSName): + hostnames.append(entry.value) + elif isinstance(entry, x509.IPAddress): + hostnames.append(str(entry.value)) + except ExtensionNotFound: + pass + + return { + "expires": cert.not_valid_after_utc, # type: ignore[attr-defined] + "not_before": cert.not_valid_before_utc, # type: ignore[attr-defined] + "hostnames": hostnames, + "serial": cert.serial_number, + }