diff --git a/python_otbr_api/__init__.py b/python_otbr_api/__init__.py index 9188c89..88bfb83 100644 --- a/python_otbr_api/__init__.py +++ b/python_otbr_api/__init__.py @@ -11,7 +11,14 @@ import aiohttp import voluptuous as vol # type: ignore[import] -from .models import ActiveDataSet, PendingDataSet, Timestamp +from .models import ( + ActiveDataSet, + EphemeralKeyActivationResult, + EphemeralKeyState, + EphemeralKeyStatus, + PendingDataSet, + Timestamp, +) # 5 minutes as recommended by # https://github.com/openthread/openthread/discussions/8567#discussioncomment-4468920 @@ -88,6 +95,15 @@ class ThreadNetworkActiveError(OTBRError): """Raised on attempts to modify the active dataset when thread network is active.""" +class EphemeralKeyNotSupportedError(OTBRError): + """Raised when the router does not support the ephemeral key (ePSKc) feature.""" + + +class EphemeralKeyConflictError(OTBRError): + """Raised when activating an ephemeral key while the feature is disabled or a + key is already active.""" + + def _rewrite_keys(data: Any, mapping: dict[str, str]) -> Any: """Recursively rename dict keys according to mapping; pass through others.""" if not isinstance(data, dict): @@ -388,6 +404,134 @@ async def get_extended_address(self) -> bytes: except ValueError as exc: raise OTBRError("unexpected API response") from exc + async def get_ephemeral_key_enabled(self) -> bool: + """Get whether the ephemeral key (ePSKc) feature is enabled. + + Raises EphemeralKeyNotSupportedError if the router does not support + the ephemeral key feature. + """ + await self._maybe_detect_key_format() + response = await self._session.get( + f"{self._url}/node/ba-epskc/state", + timeout=aiohttp.ClientTimeout(total=self._timeout), + ) + + if response.status == HTTPStatus.NOT_FOUND: + raise EphemeralKeyNotSupportedError + + if response.status != HTTPStatus.OK: + raise OTBRError(f"unexpected http status {response.status}") + + try: + return (await response.json()) == "enabled" + except ValueError as exc: + raise OTBRError("unexpected API response") from exc + + async def set_ephemeral_key_enabled(self, enabled: bool) -> None: + """Enable or disable the ephemeral key (ePSKc) feature. + + The feature must be enabled before an ephemeral key can be activated. + Raises EphemeralKeyNotSupportedError if the router does not support + the ephemeral key feature. + """ + await self._maybe_detect_key_format() + response = await self._session.put( + f"{self._url}/node/ba-epskc/state", + json="enable" if enabled else "disable", + timeout=aiohttp.ClientTimeout(total=self._timeout), + ) + + if response.status == HTTPStatus.NOT_FOUND: + raise EphemeralKeyNotSupportedError + + if response.status != HTTPStatus.OK: + raise OTBRError(f"unexpected http status {response.status}") + + async def get_ephemeral_key_status(self) -> EphemeralKeyStatus: + """Get the status of the current ephemeral key (ePSKc) session. + + Raises EphemeralKeyNotSupportedError if the router does not support + the ephemeral key feature. + """ + await self._maybe_detect_key_format() + response = await self._session.get( + f"{self._url}/node/ba-epskc/key", + timeout=aiohttp.ClientTimeout(total=self._timeout), + ) + + if response.status == HTTPStatus.NOT_FOUND: + raise EphemeralKeyNotSupportedError + + if response.status != HTTPStatus.OK: + raise OTBRError(f"unexpected http status {response.status}") + + try: + data = await response.json() + return EphemeralKeyStatus( + state=EphemeralKeyState(data["state"]), port=data["port"] + ) + except (ValueError, KeyError) as exc: + raise OTBRError("unexpected API response") from exc + + async def activate_ephemeral_key( + self, lifetime: int | None = None, port: int | None = None + ) -> EphemeralKeyActivationResult: + """Generate and activate an ephemeral key (ePSKc). + + Returns the generated 9-digit Thread Administration Passcode (TAP) + and the UDP port the border agent is listening on for the ePSKc + session. `lifetime` is the key lifetime in milliseconds and `port` + the UDP port to use; both default to the router's own defaults when + omitted. + + Raises EphemeralKeyNotSupportedError if the router does not support + the ephemeral key feature, and EphemeralKeyConflictError if the + feature is disabled or a key is already active. + """ + await self._maybe_detect_key_format() + body: dict[str, int] = {} + if lifetime is not None: + body["lifetime"] = lifetime + if port is not None: + body["port"] = port + + response = await self._session.post( + f"{self._url}/node/ba-epskc/key", + json=body, + timeout=aiohttp.ClientTimeout(total=self._timeout), + ) + + if response.status == HTTPStatus.NOT_FOUND: + raise EphemeralKeyNotSupportedError + if response.status == HTTPStatus.CONFLICT: + raise EphemeralKeyConflictError + if response.status != HTTPStatus.OK: + raise OTBRError(f"unexpected http status {response.status}") + + try: + data = await response.json() + return EphemeralKeyActivationResult(tap=data["tap"], port=data["port"]) + except (ValueError, KeyError) as exc: + raise OTBRError("unexpected API response") from exc + + async def deactivate_ephemeral_key(self) -> None: + """Deactivate the currently active ephemeral key (ePSKc), if any. + + Raises EphemeralKeyNotSupportedError if the router does not support + the ephemeral key feature. + """ + await self._maybe_detect_key_format() + response = await self._session.delete( + f"{self._url}/node/ba-epskc/key", + timeout=aiohttp.ClientTimeout(total=self._timeout), + ) + + if response.status == HTTPStatus.NOT_FOUND: + raise EphemeralKeyNotSupportedError + + if response.status != HTTPStatus.OK: + raise OTBRError(f"unexpected http status {response.status}") + async def get_coprocessor_version(self) -> str: """Get the coprocessor firmware version. diff --git a/python_otbr_api/models.py b/python_otbr_api/models.py index e0d9874..9366de5 100644 --- a/python_otbr_api/models.py +++ b/python_otbr_api/models.py @@ -3,11 +3,26 @@ from __future__ import annotations from dataclasses import dataclass +from enum import Enum from typing import Any import voluptuous as vol # type: ignore[import] +class EphemeralKeyState(Enum): + """State of the border agent ephemeral key (ePSKc) session. + + Reported by the `/node/ba-epskc/key` endpoint. See + otBorderAgentEphemeralKeyState in openthread/border_agent.h. + """ + + DISABLED = "disabled" + STOPPED = "stopped" + STARTED = "started" + CONNECTED = "connected" + ACCEPTED = "accepted" + + @dataclass class Timestamp: """Timestamp.""" @@ -199,6 +214,22 @@ def from_json(cls, json_data: Any) -> ActiveDataSet: ) +@dataclass +class EphemeralKeyStatus: + """Status of the border agent ephemeral key (ePSKc) session.""" + + state: EphemeralKeyState + port: int + + +@dataclass +class EphemeralKeyActivationResult: + """Result of activating an ephemeral key (ePSKc).""" + + tap: str + port: int + + @dataclass class PendingDataSet: # pylint: disable=too-many-instance-attributes """Operational dataset.""" diff --git a/tests/test_ephemeral_key.py b/tests/test_ephemeral_key.py new file mode 100644 index 0000000..a82c061 --- /dev/null +++ b/tests/test_ephemeral_key.py @@ -0,0 +1,190 @@ +"""Tests for the ephemeral key (ePSKc) REST API.""" + +from http import HTTPStatus + +import pytest +import python_otbr_api +from python_otbr_api import EphemeralKeyState, KeyFormat + +from tests.test_util.aiohttp import AiohttpClientMocker + +BASE_URL = "http://core-openthread-border-router:8081" + + +def _otbr(aioclient_mock: AiohttpClientMocker) -> python_otbr_api.OTBR: + return python_otbr_api.OTBR( + BASE_URL, aioclient_mock.create_session(), key_format=KeyFormat.CAMEL_CASE + ) + + +async def test_get_ephemeral_key_enabled_true( + aioclient_mock: AiohttpClientMocker, +) -> None: + """A 200 with body "enabled" reports the feature as enabled.""" + otbr = _otbr(aioclient_mock) + aioclient_mock.get(f"{BASE_URL}/node/ba-epskc/state", json="enabled") + + assert await otbr.get_ephemeral_key_enabled() is True + + +async def test_get_ephemeral_key_enabled_false( + aioclient_mock: AiohttpClientMocker, +) -> None: + """A 200 with body "disabled" reports the feature as disabled.""" + otbr = _otbr(aioclient_mock) + aioclient_mock.get(f"{BASE_URL}/node/ba-epskc/state", json="disabled") + + assert await otbr.get_ephemeral_key_enabled() is False + + +async def test_get_ephemeral_key_enabled_not_supported( + aioclient_mock: AiohttpClientMocker, +) -> None: + """A 404 means the router does not support ePSKc.""" + otbr = _otbr(aioclient_mock) + aioclient_mock.get(f"{BASE_URL}/node/ba-epskc/state", status=HTTPStatus.NOT_FOUND) + + with pytest.raises(python_otbr_api.EphemeralKeyNotSupportedError): + await otbr.get_ephemeral_key_enabled() + + +async def test_get_ephemeral_key_enabled_unexpected_status( + aioclient_mock: AiohttpClientMocker, +) -> None: + """Any other non-200 status raises OTBRError.""" + otbr = _otbr(aioclient_mock) + aioclient_mock.get( + f"{BASE_URL}/node/ba-epskc/state", status=HTTPStatus.INTERNAL_SERVER_ERROR + ) + + with pytest.raises(python_otbr_api.OTBRError): + await otbr.get_ephemeral_key_enabled() + + +async def test_set_ephemeral_key_enabled(aioclient_mock: AiohttpClientMocker) -> None: + """Enabling sends body "enable", disabling sends body "disable".""" + otbr = _otbr(aioclient_mock) + aioclient_mock.put(f"{BASE_URL}/node/ba-epskc/state", status=HTTPStatus.OK) + + await otbr.set_ephemeral_key_enabled(True) + assert aioclient_mock.mock_calls[-1][2] == "enable" + + await otbr.set_ephemeral_key_enabled(False) + assert aioclient_mock.mock_calls[-1][2] == "disable" + + +async def test_set_ephemeral_key_enabled_not_supported( + aioclient_mock: AiohttpClientMocker, +) -> None: + """A 404 means the router does not support ePSKc.""" + otbr = _otbr(aioclient_mock) + aioclient_mock.put(f"{BASE_URL}/node/ba-epskc/state", status=HTTPStatus.NOT_FOUND) + + with pytest.raises(python_otbr_api.EphemeralKeyNotSupportedError): + await otbr.set_ephemeral_key_enabled(True) + + +async def test_get_ephemeral_key_status(aioclient_mock: AiohttpClientMocker) -> None: + """A successful status response is parsed into an EphemeralKeyStatus.""" + otbr = _otbr(aioclient_mock) + aioclient_mock.get( + f"{BASE_URL}/node/ba-epskc/key", json={"state": "started", "port": 49152} + ) + + status = await otbr.get_ephemeral_key_status() + assert status.state == EphemeralKeyState.STARTED + assert status.port == 49152 + + +async def test_get_ephemeral_key_status_not_supported( + aioclient_mock: AiohttpClientMocker, +) -> None: + """A 404 means the router does not support ePSKc.""" + otbr = _otbr(aioclient_mock) + aioclient_mock.get(f"{BASE_URL}/node/ba-epskc/key", status=HTTPStatus.NOT_FOUND) + + with pytest.raises(python_otbr_api.EphemeralKeyNotSupportedError): + await otbr.get_ephemeral_key_status() + + +async def test_get_ephemeral_key_status_invalid( + aioclient_mock: AiohttpClientMocker, +) -> None: + """A malformed response body raises OTBRError.""" + otbr = _otbr(aioclient_mock) + aioclient_mock.get(f"{BASE_URL}/node/ba-epskc/key", json={"state": "started"}) + + with pytest.raises(python_otbr_api.OTBRError): + await otbr.get_ephemeral_key_status() + + +async def test_activate_ephemeral_key_defaults( + aioclient_mock: AiohttpClientMocker, +) -> None: + """With no arguments an empty JSON object is posted.""" + otbr = _otbr(aioclient_mock) + aioclient_mock.post( + f"{BASE_URL}/node/ba-epskc/key", + json={"tap": "123456789", "port": 49152}, + ) + + result = await otbr.activate_ephemeral_key() + assert result.tap == "123456789" + assert result.port == 49152 + assert aioclient_mock.mock_calls[-1][2] == {} + + +async def test_activate_ephemeral_key_with_params( + aioclient_mock: AiohttpClientMocker, +) -> None: + """lifetime and port are forwarded in the request body when given.""" + otbr = _otbr(aioclient_mock) + aioclient_mock.post( + f"{BASE_URL}/node/ba-epskc/key", + json={"tap": "123456789", "port": 12345}, + ) + + await otbr.activate_ephemeral_key(lifetime=60000, port=12345) + assert aioclient_mock.mock_calls[-1][2] == {"lifetime": 60000, "port": 12345} + + +async def test_activate_ephemeral_key_conflict( + aioclient_mock: AiohttpClientMocker, +) -> None: + """A 409 means the feature is disabled or a key is already active.""" + otbr = _otbr(aioclient_mock) + aioclient_mock.post(f"{BASE_URL}/node/ba-epskc/key", status=HTTPStatus.CONFLICT) + + with pytest.raises(python_otbr_api.EphemeralKeyConflictError): + await otbr.activate_ephemeral_key() + + +async def test_activate_ephemeral_key_not_supported( + aioclient_mock: AiohttpClientMocker, +) -> None: + """A 404 means the router does not support ePSKc.""" + otbr = _otbr(aioclient_mock) + aioclient_mock.post(f"{BASE_URL}/node/ba-epskc/key", status=HTTPStatus.NOT_FOUND) + + with pytest.raises(python_otbr_api.EphemeralKeyNotSupportedError): + await otbr.activate_ephemeral_key() + + +async def test_deactivate_ephemeral_key(aioclient_mock: AiohttpClientMocker) -> None: + """A successful deactivation returns None.""" + otbr = _otbr(aioclient_mock) + aioclient_mock.delete(f"{BASE_URL}/node/ba-epskc/key", status=HTTPStatus.OK) + + await otbr.deactivate_ephemeral_key() + assert aioclient_mock.call_count == 1 + + +async def test_deactivate_ephemeral_key_not_supported( + aioclient_mock: AiohttpClientMocker, +) -> None: + """A 404 means the router does not support ePSKc.""" + otbr = _otbr(aioclient_mock) + aioclient_mock.delete(f"{BASE_URL}/node/ba-epskc/key", status=HTTPStatus.NOT_FOUND) + + with pytest.raises(python_otbr_api.EphemeralKeyNotSupportedError): + await otbr.deactivate_ephemeral_key()