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 <amplifier@anthropic.com>
2026-04-03 21:36:38 -07:00
parent 3a8673690c
commit 4cca6d1c82
2 changed files with 319 additions and 0 deletions
@@ -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}"
+    )
@@ -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,
+    }