diff --git a/src/span_panel_api/auth.py b/src/span_panel_api/auth.py index 4df1a89..26bc336 100644 --- a/src/span_panel_api/auth.py +++ b/src/span_panel_api/auth.py @@ -8,6 +8,7 @@ from __future__ import annotations +import asyncio import hashlib import json import uuid @@ -33,6 +34,30 @@ def _int(val: object) -> int: return int(str(val)) +HTTP_TOO_MANY_REQUESTS = 429 + +#: Default attempts and base backoff used when the panel rate-limits a request. +CA_CERT_MAX_ATTEMPTS = 5 +CA_CERT_BACKOFF_S = 1.5 + + +def _retry_delay(retry_after: str | None, attempt: int, backoff_s: float) -> float: + """Return how long to wait before retrying a rate-limited request. + + Prefers the panel's ``Retry-After`` header (delta-seconds form) and falls + back to exponential backoff. Malformed or negative values fall back too, + so a bad header can never stall or skip the retry. + """ + fallback = backoff_s * (2.0 ** (attempt - 1)) + if retry_after is None: + return fallback + try: + parsed = float(retry_after) + except (TypeError, ValueError): + return fallback + return parsed if parsed >= 0 else fallback + + async def register_v2( host: str, name: str, @@ -116,14 +141,24 @@ async def download_ca_cert( timeout: float = 10.0, port: int = 80, httpx_client: httpx.AsyncClient | None = None, + max_attempts: int = CA_CERT_MAX_ATTEMPTS, + backoff_s: float = CA_CERT_BACKOFF_S, ) -> str: """Download the PEM CA certificate from the SPAN Panel. + The panel rate-limits this endpoint and replies with HTTP 429 once the + limit is hit. A single reconnect storm — or another client polling the + same panel — is enough to trigger it, and a one-shot request would turn + that transient condition into a hard setup failure. Retry with + exponential backoff, honouring ``Retry-After`` when the panel sends it. + Args: host: IP address or hostname of the SPAN Panel timeout: Request timeout in seconds when ``httpx_client`` is None; ignored when injected. port: HTTP port of the panel bootstrap API httpx_client: Optional shared ``httpx.AsyncClient``; not closed by this function. + max_attempts: Total attempts made when the panel replies HTTP 429. + backoff_s: Base delay for exponential backoff between 429 retries. Returns: PEM-encoded CA certificate as a string @@ -134,23 +169,32 @@ async def download_ca_cert( SpanPanelAPIError: Unexpected response or invalid PEM """ url = _build_url(host, port, "/api/v2/certificate/ca") + last_status: int | None = None - try: - async with _get_client(httpx_client, timeout) as client: - response = await client.get(url) - except httpx.ConnectError as exc: - raise SpanPanelConnectionError(f"Cannot reach panel at {host}") from exc - except httpx.TimeoutException as exc: - raise SpanPanelTimeoutError(f"Timed out connecting to {host}") from exc + for attempt in range(1, max_attempts + 1): + try: + async with _get_client(httpx_client, timeout) as client: + response = await client.get(url) + except httpx.ConnectError as exc: + raise SpanPanelConnectionError(f"Cannot reach panel at {host}") from exc + except httpx.TimeoutException as exc: + raise SpanPanelTimeoutError(f"Timed out connecting to {host}") from exc - if response.status_code != 200: - raise SpanPanelAPIError(f"Failed to download CA cert: HTTP {response.status_code}") + if response.status_code == 200: + pem = response.text + if not pem.startswith("-----BEGIN"): + raise SpanPanelAPIError("Response is not a valid PEM certificate") + return pem + + last_status = response.status_code + + if response.status_code == HTTP_TOO_MANY_REQUESTS and attempt < max_attempts: + await asyncio.sleep(_retry_delay(response.headers.get("retry-after"), attempt, backoff_s)) + continue - pem = response.text - if not pem.startswith("-----BEGIN"): - raise SpanPanelAPIError("Response is not a valid PEM certificate") + break - return pem + raise SpanPanelAPIError(f"Failed to download CA cert: HTTP {last_status}") async def get_homie_schema( diff --git a/src/span_panel_api/mqtt/connection.py b/src/span_panel_api/mqtt/connection.py index 212c81e..b989c9c 100644 --- a/src/span_panel_api/mqtt/connection.py +++ b/src/span_panel_api/mqtt/connection.py @@ -47,10 +47,23 @@ def _build_ssl_context(ca_pem: str) -> ssl.SSLContext: The panel issues a private CA and a server cert signed by it. We do not want to trust system CAs for this connection, so the context is built fresh rather than via ``ssl.create_default_context()``. + + The panel's CA is a minimal self-signed certificate that omits the + Authority Key Identifier (AKI) X.509v3 extension. Python 3.13 enabled + ``VERIFY_X509_STRICT`` by default, and that flag rejects such a + certificate with "Missing Authority Key Identifier", which makes the + MQTTS handshake fail on otherwise healthy panels. The flag is cleared + here so the library keeps working across Python versions. + + This does not weaken the parts of verification that matter for this + connection: the trust anchor is still only the panel's own CA (fetched + over the local network immediately before use), hostname checking stays + enabled, and signature/expiry validation is unchanged. """ ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) ctx.verify_mode = ssl.CERT_REQUIRED ctx.check_hostname = True + ctx.verify_flags &= ~ssl.VERIFY_X509_STRICT ctx.load_verify_locations(cadata=ca_pem) return ctx diff --git a/tests/test_detection_auth.py b/tests/test_detection_auth.py index 8131bb5..a9adfd1 100644 --- a/tests/test_detection_auth.py +++ b/tests/test_detection_auth.py @@ -20,6 +20,7 @@ ) from span_panel_api.auth import ( delete_fqdn, + _retry_delay, download_ca_cert, get_fqdn, get_homie_schema, @@ -392,6 +393,110 @@ async def test_download_http_error(self): with pytest.raises(SpanPanelAPIError, match="500"): await download_ca_cert("192.168.65.70") + @pytest.mark.asyncio + async def test_download_retries_after_429_then_succeeds(self): + """A transient rate-limit must not fail setup — it should be retried.""" + responses = [_mock_response(429), _mock_response(429), _mock_response(200, text=PEM_CERT)] + sleeps: list[float] = [] + + async def _fake_sleep(delay: float) -> None: + sleeps.append(delay) + + with ( + patch("span_panel_api._http.httpx.AsyncClient") as mock_client_cls, + patch("span_panel_api.auth.asyncio.sleep", _fake_sleep), + ): + mock_client = AsyncMock() + mock_client.get = AsyncMock(side_effect=responses) + mock_client.__aenter__ = AsyncMock(return_value=mock_client) + mock_client.__aexit__ = AsyncMock(return_value=False) + mock_client_cls.return_value = mock_client + + result = await download_ca_cert("192.168.65.70", backoff_s=1.5) + + assert result.startswith("-----BEGIN") + assert mock_client.get.await_count == 3 + # exponential backoff between the two retries + assert sleeps == [1.5, 3.0] + + @pytest.mark.asyncio + async def test_download_gives_up_after_max_attempts(self): + """Persistent 429s surface as an API error rather than retrying forever.""" + sleeps: list[float] = [] + + async def _fake_sleep(delay: float) -> None: + sleeps.append(delay) + + with ( + patch("span_panel_api._http.httpx.AsyncClient") as mock_client_cls, + patch("span_panel_api.auth.asyncio.sleep", _fake_sleep), + ): + mock_client = AsyncMock() + mock_client.get = AsyncMock(return_value=_mock_response(429)) + mock_client.__aenter__ = AsyncMock(return_value=mock_client) + mock_client.__aexit__ = AsyncMock(return_value=False) + mock_client_cls.return_value = mock_client + + with pytest.raises(SpanPanelAPIError, match="429"): + await download_ca_cert("192.168.65.70", max_attempts=3) + + assert mock_client.get.await_count == 3 + assert len(sleeps) == 2 + + @pytest.mark.asyncio + async def test_download_honours_retry_after_header(self): + """When the panel sends Retry-After, obey it instead of the backoff curve.""" + rate_limited = _mock_response(429) + rate_limited.headers["retry-after"] = "7" + sleeps: list[float] = [] + + async def _fake_sleep(delay: float) -> None: + sleeps.append(delay) + + with ( + patch("span_panel_api._http.httpx.AsyncClient") as mock_client_cls, + patch("span_panel_api.auth.asyncio.sleep", _fake_sleep), + ): + mock_client = AsyncMock() + mock_client.get = AsyncMock(side_effect=[rate_limited, _mock_response(200, text=PEM_CERT)]) + mock_client.__aenter__ = AsyncMock(return_value=mock_client) + mock_client.__aexit__ = AsyncMock(return_value=False) + mock_client_cls.return_value = mock_client + + result = await download_ca_cert("192.168.65.70") + + assert result.startswith("-----BEGIN") + assert sleeps == [7.0] + + @pytest.mark.asyncio + async def test_download_does_not_retry_non_429(self): + """Only rate limiting is retried; other failures fail fast.""" + with patch("span_panel_api._http.httpx.AsyncClient") as mock_client_cls: + mock_client = AsyncMock() + mock_client.get = AsyncMock(return_value=_mock_response(500)) + mock_client.__aenter__ = AsyncMock(return_value=mock_client) + mock_client.__aexit__ = AsyncMock(return_value=False) + mock_client_cls.return_value = mock_client + + with pytest.raises(SpanPanelAPIError, match="500"): + await download_ca_cert("192.168.65.70") + + assert mock_client.get.await_count == 1 + + +class TestRetryDelay: + def test_malformed_retry_after_falls_back_to_backoff(self): + assert _retry_delay("soon", attempt=1, backoff_s=1.5) == 1.5 + + def test_negative_retry_after_falls_back_to_backoff(self): + assert _retry_delay("-5", attempt=2, backoff_s=1.5) == 3.0 + + def test_missing_retry_after_uses_exponential_backoff(self): + assert _retry_delay(None, attempt=3, backoff_s=1.5) == 6.0 + + def test_zero_retry_after_is_respected(self): + assert _retry_delay("0", attempt=4, backoff_s=1.5) == 0.0 + # =================================================================== # get_homie_schema diff --git a/tests/test_ssl_context.py b/tests/test_ssl_context.py new file mode 100644 index 0000000..7e6df0e --- /dev/null +++ b/tests/test_ssl_context.py @@ -0,0 +1,234 @@ +"""SSL context construction for the panel's private CA. + +SPAN panels serve a minimal self-signed CA that omits the Authority Key +Identifier (AKI) X.509v3 extension. Python 3.13 turned on +``VERIFY_X509_STRICT`` by default, and that flag rejects such certificates +with "Missing Authority Key Identifier" — which broke MQTTS against +perfectly healthy panels. These tests pin the behaviour so the regression +cannot come back silently. +""" + +from __future__ import annotations + +import contextlib +import datetime +import socket +import ssl +import tempfile +import threading +from pathlib import Path + +import pytest + +from span_panel_api.mqtt.connection import _build_ssl_context + +cryptography = pytest.importorskip("cryptography", reason="cryptography needed to mint a test CA") + +from cryptography import x509 # noqa: E402 +from cryptography.hazmat.primitives import hashes, serialization # noqa: E402 +from cryptography.hazmat.primitives.asymmetric import ec # noqa: E402 +from cryptography.x509.oid import NameOID # noqa: E402 + + +def _self_signed_ca(*, with_aki: bool) -> str: + """Mint a self-signed CA, mirroring what the panel serves. + + With ``with_aki=False`` the certificate omits the Authority Key + Identifier extension, exactly like the SPAN panel's CA. + """ + key = ec.generate_private_key(ec.SECP256R1()) + subject = x509.Name( + [ + x509.NameAttribute(NameOID.COUNTRY_NAME, "US"), + x509.NameAttribute(NameOID.ORGANIZATION_NAME, "SPAN.io"), + x509.NameAttribute(NameOID.COMMON_NAME, "test-panel CA"), + ] + ) + now = datetime.datetime.now(datetime.timezone.utc) + builder = ( + x509.CertificateBuilder() + .subject_name(subject) + .issuer_name(subject) + .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=365)) + .add_extension(x509.BasicConstraints(ca=True, path_length=None), critical=True) + .add_extension( + x509.KeyUsage( + digital_signature=True, + content_commitment=False, + key_encipherment=False, + data_encipherment=False, + key_agreement=False, + key_cert_sign=True, + crl_sign=True, + encipher_only=False, + decipher_only=False, + ), + critical=True, + ) + ) + if with_aki: + builder = builder.add_extension( + x509.AuthorityKeyIdentifier.from_issuer_public_key(key.public_key()), + critical=False, + ) + + cert = builder.sign(key, hashes.SHA256()) + return cert.public_bytes(serialization.Encoding.PEM).decode() + + +def _ca_and_leaf(*, with_aki: bool) -> tuple[str, str, str]: + """Mint a CA plus a localhost leaf signed by it. + + Returns ``(ca_pem, leaf_pem, leaf_key_pem)``. When ``with_aki`` is False + neither certificate carries an Authority Key Identifier, reproducing the + chain a SPAN panel actually presents. + """ + ca_key = ec.generate_private_key(ec.SECP256R1()) + ca_name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "test-panel CA")]) + now = datetime.datetime.now(datetime.timezone.utc) + + ca_builder = ( + 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 - datetime.timedelta(days=1)) + .not_valid_after(now + datetime.timedelta(days=365)) + .add_extension(x509.BasicConstraints(ca=True, path_length=None), critical=True) + ) + if with_aki: + ca_builder = ca_builder.add_extension(x509.SubjectKeyIdentifier.from_public_key(ca_key.public_key()), critical=False) + ca_cert = ca_builder.sign(ca_key, hashes.SHA256()) + + leaf_key = ec.generate_private_key(ec.SECP256R1()) + leaf_builder = ( + x509.CertificateBuilder() + .subject_name(x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "localhost")])) + .issuer_name(ca_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) + .add_extension(x509.BasicConstraints(ca=False, path_length=None), critical=True) + ) + if with_aki: + leaf_builder = leaf_builder.add_extension( + x509.AuthorityKeyIdentifier.from_issuer_public_key(ca_key.public_key()), critical=False + ) + leaf_cert = leaf_builder.sign(ca_key, hashes.SHA256()) + + return ( + ca_cert.public_bytes(serialization.Encoding.PEM).decode(), + leaf_cert.public_bytes(serialization.Encoding.PEM).decode(), + leaf_key.private_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PrivateFormat.PKCS8, + encryption_algorithm=serialization.NoEncryption(), + ).decode(), + ) + + +@contextlib.contextmanager +def _tls_server(leaf_pem: str, leaf_key_pem: str): + """Run a throwaway TLS server on localhost, yielding ``(host, port)``.""" + with tempfile.TemporaryDirectory() as tmp: + cert_path = Path(tmp) / "leaf.pem" + key_path = Path(tmp) / "leaf.key" + cert_path.write_text(leaf_pem) + key_path.write_text(leaf_key_pem) + + server_ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) + server_ctx.load_cert_chain(certfile=str(cert_path), keyfile=str(key_path)) + + listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + listener.bind(("127.0.0.1", 0)) + listener.listen(1) + listener.settimeout(5) + + def _serve() -> None: + try: + raw, _ = listener.accept() + except OSError: + return + try: + with server_ctx.wrap_socket(raw, server_side=True): + pass + except OSError: + pass + finally: + raw.close() + + thread = threading.Thread(target=_serve, daemon=True) + thread.start() + try: + yield listener.getsockname() + finally: + listener.close() + thread.join(timeout=5) + + +class TestBuildSslContext: + def test_loads_ca_without_authority_key_identifier(self) -> None: + """The panel's AKI-less CA must load — this is the actual regression.""" + ctx = _build_ssl_context(_self_signed_ca(with_aki=False)) + + assert ctx.verify_mode is ssl.CERT_REQUIRED + assert ctx.check_hostname is True + assert ctx.get_ca_certs(), "panel CA should be installed as a trust anchor" + + def test_handshake_succeeds_against_panel_style_cert(self) -> None: + """End-to-end proof: a TLS handshake completes against an AKI-less chain.""" + ca_pem, leaf_pem, leaf_key_pem = _ca_and_leaf(with_aki=False) + ctx = _build_ssl_context(ca_pem) + + with _tls_server(leaf_pem, leaf_key_pem) as (host, port): + with socket.create_connection((host, port), timeout=5) as raw: + with ctx.wrap_socket(raw, server_hostname="localhost") as tls: + assert tls.getpeercert() is not None + + def test_strict_x509_would_reject_the_panel_chain(self) -> None: + """Guard the premise: with VERIFY_X509_STRICT the same handshake fails. + + The AKI requirement is enforced during chain building at handshake + time, not when the CA is loaded. If a future Python stops rejecting + AKI-less chains this test fails loudly, telling us the workaround is + no longer needed. + """ + ca_pem, leaf_pem, leaf_key_pem = _ca_and_leaf(with_aki=False) + strict = _build_ssl_context(ca_pem) + strict.verify_flags |= ssl.VERIFY_X509_STRICT + + with _tls_server(leaf_pem, leaf_key_pem) as (host, port): + with socket.create_connection((host, port), timeout=5) as raw: + with pytest.raises(ssl.SSLCertVerificationError, match="Authority Key Identifier"): + strict.wrap_socket(raw, server_hostname="localhost") + + def test_strict_flag_is_cleared(self) -> None: + ctx = _build_ssl_context(_self_signed_ca(with_aki=False)) + assert not (ctx.verify_flags & ssl.VERIFY_X509_STRICT) + + def test_hostname_and_peer_verification_stay_enabled(self) -> None: + """Clearing the strict flag must not weaken the checks that matter.""" + ctx = _build_ssl_context(_self_signed_ca(with_aki=True)) + + assert ctx.check_hostname is True + assert ctx.verify_mode is ssl.CERT_REQUIRED + + def test_system_ca_bundle_is_not_trusted(self) -> None: + """Only the panel CA is a trust anchor — no system roots.""" + ctx = _build_ssl_context(_self_signed_ca(with_aki=False)) + assert len(ctx.get_ca_certs()) == 1 + + def test_conventional_ca_still_loads(self) -> None: + ctx = _build_ssl_context(_self_signed_ca(with_aki=True)) + assert ctx.get_ca_certs() + + def test_malformed_pem_raises(self) -> None: + with pytest.raises((ssl.SSLError, ValueError)): + _build_ssl_context("-----BEGIN CERTIFICATE-----\nnot base64\n-----END CERTIFICATE-----\n")