diff --git a/CHANGELOG.md b/CHANGELOG.md index 7993ec1..ab71095 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,12 @@ All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [2.3.2] - 03/2026 + +### Added + +- **FQDN management endpoints** — `register_fqdn()`, `get_fqdn()`, `delete_fqdn()` for managing the panel's TLS certificate SAN via `/api/v2/dns/fqdn` ([spanio/SPAN-API-Client-Docs#10](https://github.com/spanio/SPAN-API-Client-Docs/issues/10)) + ## [2.3.1] - 03/2026 ### Fixed diff --git a/pyproject.toml b/pyproject.toml index 03285e0..bcf3d7a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "span-panel-api" -version = "2.3.1" +version = "2.3.2" description = "A client library for SPAN Panel API" authors = [ {name = "SpanPanel"} diff --git a/src/span_panel_api/__init__.py b/src/span_panel_api/__init__.py index 8b915ef..fcdcfe5 100644 --- a/src/span_panel_api/__init__.py +++ b/src/span_panel_api/__init__.py @@ -4,7 +4,15 @@ supporting MQTT/Homie (v2) transport. """ -from .auth import download_ca_cert, get_homie_schema, regenerate_passphrase, register_v2 +from .auth import ( + delete_fqdn, + download_ca_cert, + get_fqdn, + get_homie_schema, + regenerate_passphrase, + register_fqdn, + register_v2, +) from .detection import DetectionResult, detect_api_version from .exceptions import ( SpanPanelAPIError, @@ -44,7 +52,7 @@ StreamingCapableProtocol, ) -__version__ = "2.3.0" +__version__ = "2.3.2" # fmt: off __all__ = [ # noqa: RUF022 # Protocols @@ -70,8 +78,11 @@ "V2AuthResponse", "V2HomieSchema", "V2StatusInfo", + "delete_fqdn", "download_ca_cert", + "get_fqdn", "get_homie_schema", + "register_fqdn", "regenerate_passphrase", "register_v2", # Transport diff --git a/src/span_panel_api/auth.py b/src/span_panel_api/auth.py index 94943e1..f4bc3b0 100644 --- a/src/span_panel_api/auth.py +++ b/src/span_panel_api/auth.py @@ -241,6 +241,123 @@ async def regenerate_passphrase(host: str, token: str, timeout: float = 10.0, po return _str(data["ebusBrokerPassword"]) +async def register_fqdn(host: str, token: str, fqdn: str, timeout: float = 10.0, port: int = 80) -> None: + """Register an FQDN with the SPAN Panel for TLS certificate SAN inclusion. + + The panel regenerates its TLS server certificate to include the + provided FQDN in the Subject Alternative Names, allowing MQTTS + clients connecting via the FQDN to pass hostname verification. + + Args: + host: IP address or hostname of the SPAN Panel + token: Valid JWT access token from register_v2 + fqdn: Fully qualified domain name to register + timeout: Request timeout in seconds + port: HTTP port of the panel bootstrap API + + Raises: + SpanPanelAuthError: Token invalid or expired + SpanPanelConnectionError: Cannot reach panel + SpanPanelTimeoutError: Request timed out + SpanPanelAPIError: Unexpected response (including 404 if unsupported) + """ + url = _build_url(host, port, "/api/v2/dns/fqdn") + headers = {"Authorization": f"Bearer {token}"} + payload = {"ebusTlsFqdn": fqdn} + + try: + async with httpx.AsyncClient(timeout=timeout, verify=False) as client: # nosec B501 + response = await client.post(url, json=payload, headers=headers) + 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 in (401, 403): + raise SpanPanelAuthError(f"Authentication failed (HTTP {response.status_code})") + + if response.status_code not in (200, 201, 204): + raise SpanPanelAPIError(f"Failed to register FQDN: HTTP {response.status_code}") + + +async def get_fqdn(host: str, token: str, timeout: float = 10.0, port: int = 80) -> str: + """Retrieve the currently registered FQDN from the SPAN Panel. + + Args: + host: IP address or hostname of the SPAN Panel + token: Valid JWT access token from register_v2 + timeout: Request timeout in seconds + port: HTTP port of the panel bootstrap API + + Returns: + The registered FQDN, or empty string if none is configured + + Raises: + SpanPanelAuthError: Token invalid or expired + SpanPanelConnectionError: Cannot reach panel + SpanPanelTimeoutError: Request timed out + SpanPanelAPIError: Unexpected response + """ + url = _build_url(host, port, "/api/v2/dns/fqdn") + headers = {"Authorization": f"Bearer {token}"} + + try: + async with httpx.AsyncClient(timeout=timeout, verify=False) as client: # nosec B501 + response = await client.get(url, headers=headers) + 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 in (401, 403): + raise SpanPanelAuthError(f"Authentication failed (HTTP {response.status_code})") + + if response.status_code == 404: + return "" + + if response.status_code != 200: + raise SpanPanelAPIError(f"Failed to get FQDN: HTTP {response.status_code}") + + data: dict[str, object] = response.json() + return _str(data.get("ebusTlsFqdn")) + + +async def delete_fqdn(host: str, token: str, timeout: float = 10.0, port: int = 80) -> None: + """Remove the registered FQDN from the SPAN Panel. + + The panel regenerates its TLS certificate without the FQDN in + the SAN list. + + Args: + host: IP address or hostname of the SPAN Panel + token: Valid JWT access token from register_v2 + timeout: Request timeout in seconds + port: HTTP port of the panel bootstrap API + + Raises: + SpanPanelAuthError: Token invalid or expired + SpanPanelConnectionError: Cannot reach panel + SpanPanelTimeoutError: Request timed out + SpanPanelAPIError: Unexpected response + """ + url = _build_url(host, port, "/api/v2/dns/fqdn") + headers = {"Authorization": f"Bearer {token}"} + + try: + async with httpx.AsyncClient(timeout=timeout, verify=False) as client: # nosec B501 + response = await client.delete(url, headers=headers) + 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 in (401, 403): + raise SpanPanelAuthError(f"Authentication failed (HTTP {response.status_code})") + + if response.status_code not in (200, 204): + raise SpanPanelAPIError(f"Failed to delete FQDN: HTTP {response.status_code}") + + async def get_v2_status(host: str, timeout: float = 5.0, port: int = 80) -> V2StatusInfo: """Lightweight v2 status probe (unauthenticated). diff --git a/tests/test_detection_auth.py b/tests/test_detection_auth.py index 647471f..21ecac1 100644 --- a/tests/test_detection_auth.py +++ b/tests/test_detection_auth.py @@ -18,10 +18,13 @@ V2StatusInfo, ) from span_panel_api.auth import ( + delete_fqdn, download_ca_cert, + get_fqdn, get_homie_schema, get_v2_status, regenerate_passphrase, + register_fqdn, register_v2, ) @@ -407,6 +410,282 @@ async def test_regenerate_412_precondition_failed(self): await regenerate_passphrase("192.168.65.70", "") +# =================================================================== +# register_fqdn +# =================================================================== + + +class TestRegisterFqdn: + @pytest.mark.asyncio + async def test_register_fqdn_success(self): + mock_response = _mock_response(200) + with patch("span_panel_api.auth.httpx.AsyncClient") as mock_client_cls: + mock_client = AsyncMock() + mock_client.post.return_value = mock_response + mock_client.__aenter__ = AsyncMock(return_value=mock_client) + mock_client.__aexit__ = AsyncMock(return_value=False) + mock_client_cls.return_value = mock_client + + await register_fqdn("192.168.65.70", "jwt-token", "panel.example.com") + + mock_client.post.assert_called_once() + call_kwargs = mock_client.post.call_args + assert call_kwargs.kwargs["json"] == {"ebusTlsFqdn": "panel.example.com"} + + @pytest.mark.asyncio + async def test_register_fqdn_accepts_201(self): + mock_response = _mock_response(201) + with patch("span_panel_api.auth.httpx.AsyncClient") as mock_client_cls: + mock_client = AsyncMock() + mock_client.post.return_value = mock_response + mock_client.__aenter__ = AsyncMock(return_value=mock_client) + mock_client.__aexit__ = AsyncMock(return_value=False) + mock_client_cls.return_value = mock_client + + await register_fqdn("192.168.65.70", "jwt-token", "panel.example.com") + + @pytest.mark.asyncio + async def test_register_fqdn_accepts_204(self): + mock_response = _mock_response(204) + with patch("span_panel_api.auth.httpx.AsyncClient") as mock_client_cls: + mock_client = AsyncMock() + mock_client.post.return_value = mock_response + mock_client.__aenter__ = AsyncMock(return_value=mock_client) + mock_client.__aexit__ = AsyncMock(return_value=False) + mock_client_cls.return_value = mock_client + + await register_fqdn("192.168.65.70", "jwt-token", "panel.example.com") + + @pytest.mark.asyncio + async def test_register_fqdn_auth_error(self): + mock_response = _mock_response(401) + with patch("span_panel_api.auth.httpx.AsyncClient") as mock_client_cls: + mock_client = AsyncMock() + mock_client.post.return_value = mock_response + 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(SpanPanelAuthError, match="401"): + await register_fqdn("192.168.65.70", "bad-token", "panel.example.com") + + @pytest.mark.asyncio + async def test_register_fqdn_403(self): + mock_response = _mock_response(403) + with patch("span_panel_api.auth.httpx.AsyncClient") as mock_client_cls: + mock_client = AsyncMock() + mock_client.post.return_value = mock_response + 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(SpanPanelAuthError, match="403"): + await register_fqdn("192.168.65.70", "bad-token", "panel.example.com") + + @pytest.mark.asyncio + async def test_register_fqdn_api_error(self): + mock_response = _mock_response(500) + with patch("span_panel_api.auth.httpx.AsyncClient") as mock_client_cls: + mock_client = AsyncMock() + mock_client.post.return_value = mock_response + 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 register_fqdn("192.168.65.70", "jwt-token", "panel.example.com") + + @pytest.mark.asyncio + async def test_register_fqdn_connection_error(self): + with patch("span_panel_api.auth.httpx.AsyncClient") as mock_client_cls: + mock_client = AsyncMock() + mock_client.post.side_effect = httpx.ConnectError("Connection refused") + 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(SpanPanelConnectionError): + await register_fqdn("192.168.65.70", "jwt-token", "panel.example.com") + + @pytest.mark.asyncio + async def test_register_fqdn_timeout(self): + with patch("span_panel_api.auth.httpx.AsyncClient") as mock_client_cls: + mock_client = AsyncMock() + mock_client.post.side_effect = httpx.TimeoutException("Timed out") + 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(SpanPanelTimeoutError): + await register_fqdn("192.168.65.70", "jwt-token", "panel.example.com") + + +# =================================================================== +# get_fqdn +# =================================================================== + + +class TestGetFqdn: + @pytest.mark.asyncio + async def test_get_fqdn_success(self): + mock_response = _mock_response(200, {"ebusTlsFqdn": "panel.example.com"}) + with patch("span_panel_api.auth.httpx.AsyncClient") as mock_client_cls: + mock_client = AsyncMock() + mock_client.get.return_value = mock_response + mock_client.__aenter__ = AsyncMock(return_value=mock_client) + mock_client.__aexit__ = AsyncMock(return_value=False) + mock_client_cls.return_value = mock_client + + result = await get_fqdn("192.168.65.70", "jwt-token") + + assert result == "panel.example.com" + + @pytest.mark.asyncio + async def test_get_fqdn_not_configured_returns_empty(self): + mock_response = _mock_response(404) + with patch("span_panel_api.auth.httpx.AsyncClient") as mock_client_cls: + mock_client = AsyncMock() + mock_client.get.return_value = mock_response + mock_client.__aenter__ = AsyncMock(return_value=mock_client) + mock_client.__aexit__ = AsyncMock(return_value=False) + mock_client_cls.return_value = mock_client + + result = await get_fqdn("192.168.65.70", "jwt-token") + + assert result == "" + + @pytest.mark.asyncio + async def test_get_fqdn_auth_error(self): + mock_response = _mock_response(401) + with patch("span_panel_api.auth.httpx.AsyncClient") as mock_client_cls: + mock_client = AsyncMock() + mock_client.get.return_value = mock_response + 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(SpanPanelAuthError, match="401"): + await get_fqdn("192.168.65.70", "bad-token") + + @pytest.mark.asyncio + async def test_get_fqdn_api_error(self): + mock_response = _mock_response(500) + with patch("span_panel_api.auth.httpx.AsyncClient") as mock_client_cls: + mock_client = AsyncMock() + mock_client.get.return_value = mock_response + 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 get_fqdn("192.168.65.70", "jwt-token") + + @pytest.mark.asyncio + async def test_get_fqdn_connection_error(self): + with patch("span_panel_api.auth.httpx.AsyncClient") as mock_client_cls: + mock_client = AsyncMock() + mock_client.get.side_effect = httpx.ConnectError("Connection refused") + 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(SpanPanelConnectionError): + await get_fqdn("192.168.65.70", "jwt-token") + + @pytest.mark.asyncio + async def test_get_fqdn_timeout(self): + with patch("span_panel_api.auth.httpx.AsyncClient") as mock_client_cls: + mock_client = AsyncMock() + mock_client.get.side_effect = httpx.TimeoutException("Timed out") + 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(SpanPanelTimeoutError): + await get_fqdn("192.168.65.70", "jwt-token") + + +# =================================================================== +# delete_fqdn +# =================================================================== + + +class TestDeleteFqdn: + @pytest.mark.asyncio + async def test_delete_fqdn_success_200(self): + mock_response = _mock_response(200) + with patch("span_panel_api.auth.httpx.AsyncClient") as mock_client_cls: + mock_client = AsyncMock() + mock_client.delete.return_value = mock_response + mock_client.__aenter__ = AsyncMock(return_value=mock_client) + mock_client.__aexit__ = AsyncMock(return_value=False) + mock_client_cls.return_value = mock_client + + await delete_fqdn("192.168.65.70", "jwt-token") + + @pytest.mark.asyncio + async def test_delete_fqdn_success_204(self): + mock_response = _mock_response(204) + with patch("span_panel_api.auth.httpx.AsyncClient") as mock_client_cls: + mock_client = AsyncMock() + mock_client.delete.return_value = mock_response + mock_client.__aenter__ = AsyncMock(return_value=mock_client) + mock_client.__aexit__ = AsyncMock(return_value=False) + mock_client_cls.return_value = mock_client + + await delete_fqdn("192.168.65.70", "jwt-token") + + @pytest.mark.asyncio + async def test_delete_fqdn_auth_error(self): + mock_response = _mock_response(403) + with patch("span_panel_api.auth.httpx.AsyncClient") as mock_client_cls: + mock_client = AsyncMock() + mock_client.delete.return_value = mock_response + 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(SpanPanelAuthError, match="403"): + await delete_fqdn("192.168.65.70", "bad-token") + + @pytest.mark.asyncio + async def test_delete_fqdn_api_error(self): + mock_response = _mock_response(500) + with patch("span_panel_api.auth.httpx.AsyncClient") as mock_client_cls: + mock_client = AsyncMock() + mock_client.delete.return_value = mock_response + 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 delete_fqdn("192.168.65.70", "jwt-token") + + @pytest.mark.asyncio + async def test_delete_fqdn_connection_error(self): + with patch("span_panel_api.auth.httpx.AsyncClient") as mock_client_cls: + mock_client = AsyncMock() + mock_client.delete.side_effect = httpx.ConnectError("Connection refused") + 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(SpanPanelConnectionError): + await delete_fqdn("192.168.65.70", "jwt-token") + + @pytest.mark.asyncio + async def test_delete_fqdn_timeout(self): + with patch("span_panel_api.auth.httpx.AsyncClient") as mock_client_cls: + mock_client = AsyncMock() + mock_client.delete.side_effect = httpx.TimeoutException("Timed out") + 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(SpanPanelTimeoutError): + await delete_fqdn("192.168.65.70", "jwt-token") + + # =================================================================== # get_v2_status # =================================================================== @@ -499,7 +778,14 @@ def test_v2_auth_function_exports(self): assert hasattr(span_panel_api, "get_homie_schema") assert hasattr(span_panel_api, "regenerate_passphrase") + def test_fqdn_function_exports(self): + import span_panel_api + + assert hasattr(span_panel_api, "register_fqdn") + assert hasattr(span_panel_api, "get_fqdn") + assert hasattr(span_panel_api, "delete_fqdn") + def test_version_bumped(self): import span_panel_api - assert span_panel_api.__version__ == "2.3.0" + assert span_panel_api.__version__ == "2.3.2" diff --git a/tests/test_mqtt_homie.py b/tests/test_mqtt_homie.py index 79b8869..64dedd3 100644 --- a/tests/test_mqtt_homie.py +++ b/tests/test_mqtt_homie.py @@ -1161,7 +1161,7 @@ def test_top_level_exports(self): def test_version_beta(self): import span_panel_api - assert span_panel_api.__version__ == "2.3.0" + assert span_panel_api.__version__ == "2.3.2" # ---------------------------------------------------------------------------