Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
2 changes: 1 addition & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
@@ -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"}
Expand Down
15 changes: 13 additions & 2 deletions src/span_panel_api/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -44,7 +52,7 @@
StreamingCapableProtocol,
)

__version__ = "2.3.0"
__version__ = "2.3.2"
# fmt: off
__all__ = [ # noqa: RUF022
# Protocols
Expand All @@ -70,8 +78,11 @@
"V2AuthResponse",
"V2HomieSchema",
"V2StatusInfo",
"delete_fqdn",
"download_ca_cert",
"get_fqdn",
"get_homie_schema",
"register_fqdn",
"regenerate_passphrase",
"register_v2",
# Transport
Expand Down
117 changes: 117 additions & 0 deletions src/span_panel_api/auth.py
Original file line number Diff line number Diff line change
Expand Up @@ -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).

Expand Down
Loading