Add pending-dataset TLV write support

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name         = "python-otbr-api"
-version      = "2.10.0"
+version      = "2.11.0"
 license      = {text = "MIT"}
 description  = "API to interact with an OTBR via its REST API"
 readme       = "README.md"

python_otbr_api/__init__.py

@@ -113,6 +113,15 @@ class EphemeralKeyConflictError(OTBRError):
     key is already active."""
 
 
+class PendingDatasetConflictError(OTBRError):
+    """Raised when a pending dataset write is refused because one is in place.
+
+    Its own type rather than a plain OTBRError: the caller's next step is
+    different from the one a transport or protocol failure calls for, since
+    nothing was written and the state that made the write wrong is readable.
+    """
+
+
 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):
@@ -372,6 +381,43 @@ async def set_active_dataset_tlvs(self, dataset: bytes) -> None:
         if response.status not in (HTTPStatus.CREATED, HTTPStatus.OK):
             raise OTBRError(f"unexpected http status {response.status}")
 
+    async def set_pending_dataset_tlvs(self, dataset: bytes) -> None:
+        """Set the pending operational dataset, when none is in place yet.
+
+        A write while a pending dataset is in flight is refused outright.
+        Superseding one is never safe to do implicitly: the replacement races
+        the delay timer on every device that already holds the old dataset, so
+        a late replacement can split the mesh, and it would also silently undo
+        whatever the in-flight dataset was doing, such as a channel change.
+        A caller that finds the write refused should surface that to whoever
+        asked for it, not retry.
+
+        The check runs locally first, and again on the border router for one
+        that honors If-None-Match on this endpoint
+        (https://github.com/openthread/ot-br-posix/pull/3552), where it is
+        atomic with the write; older border routers ignore the header.
+
+        Raises PendingDatasetConflictError when either check refuses the
+        write, and OTBRError if the http status is 400 or higher for any
+        other reason or the response is invalid.
+        """
+        if await self.get_pending_dataset_tlvs() is not None:
+            raise PendingDatasetConflictError("a pending dataset is already in place")
+        await self._maybe_detect_key_format()
+        response = await self._session.put(
+            f"{self._url}/node/dataset/pending",
+            data=dataset.hex(),
+            headers={"Content-Type": "text/plain", "If-None-Match": "*"},
+            timeout=aiohttp.ClientTimeout(total=10),
+        )
+
+        if response.status == HTTPStatus.CONFLICT:
+            raise ThreadNetworkActiveError
+        if response.status == HTTPStatus.PRECONDITION_FAILED:
+            raise PendingDatasetConflictError("a pending dataset is already in place")
+        if response.status not in (HTTPStatus.CREATED, HTTPStatus.OK):
+            raise OTBRError(f"unexpected http status {response.status}")
+
     async def set_channel(
         self, channel: int, delay: int = PENDING_DATASET_DELAY_TIMER
     ) -> None:

python_otbr_api/tlv_parser.py

@@ -116,11 +116,62 @@ def __post_init__(self) -> None:
         """Decode the timestamp."""
         # The timestamps are packed in 8 bytes:
         # [seconds 48 bits][ticks 15 bits][authoritative flag 1 bit]
-        unpacked: int = struct.unpack("!Q", self.data)[0]
+        try:
+            unpacked: int = struct.unpack("!Q", self.data)[0]
+        except struct.error as err:
+            raise TLVError(f"invalid timestamp '{self.data.hex()}'") from err
         self.authoritative = bool(unpacked & 1)
         self.seconds = unpacked >> 16
         self.ticks = (unpacked >> 1) & 0x7FFF
 
+    @classmethod
+    def from_values(
+        cls,
+        tag: MeshcopTLVType,
+        seconds: int,
+        ticks: int = 0,
+        authoritative: bool = False,
+    ) -> Timestamp:
+        """Construct a timestamp from its field values.
+
+        Raises TLVError if seconds or ticks don't fit the wire format.
+        """
+        if not 0 <= seconds < 2**48:
+            raise TLVError(f"timestamp seconds out of range: {seconds}")
+        if not 0 <= ticks < 2**15:
+            raise TLVError(f"timestamp ticks out of range: {ticks}")
+        packed = (seconds << 16) | (ticks << 1) | int(authoritative)
+        return cls(tag, struct.pack("!Q", packed))
+
+
+@dataclass
+class DelayTimer(MeshcopTLVItem):
+    """Delay timer."""
+
+    delay: int = field(init=False)
+
+    def __post_init__(self) -> None:
+        """Decode the delay in milliseconds.
+
+        Raises TLVError if the data is not the four bytes the wire format
+        defines, so a malformed delay timer is rejected instead of being
+        accepted at whatever width it arrived with.
+        """
+        try:
+            self.delay = struct.unpack("!L", self.data)[0]
+        except struct.error as err:
+            raise TLVError(f"invalid delay timer '{self.data.hex()}'") from err
+
+    @classmethod
+    def from_milliseconds(cls, delay: int) -> DelayTimer:
+        """Construct a delay timer from a delay in milliseconds.
+
+        Raises TLVError if the delay doesn't fit the wire format.
+        """
+        if not 0 <= delay < 2**32:
+            raise TLVError(f"delay timer out of range: {delay}")
+        return cls(MeshcopTLVType.DELAYTIMER, struct.pack("!L", delay))
+
 
 def _encode_item(item: MeshcopTLVItem) -> bytes:
     """Encode a dataset item to TLV format."""
@@ -143,10 +194,12 @@ def encode_tlv(items: dict[MeshcopTLVType | int, MeshcopTLVItem]) -> str:
 
 def _parse_item(tag: MeshcopTLVType | int, data: bytes) -> MeshcopTLVItem:
     """Parse a TLV encoded dataset item."""
-    if tag == MeshcopTLVType.ACTIVETIMESTAMP:
+    if tag in (MeshcopTLVType.ACTIVETIMESTAMP, MeshcopTLVType.PENDINGTIMESTAMP):
         return Timestamp(tag, data)
     if tag == MeshcopTLVType.CHANNEL:
         return Channel(tag, data)
+    if tag == MeshcopTLVType.DELAYTIMER:
+        return DelayTimer(tag, data)
     if tag == MeshcopTLVType.NETWORKNAME:
         return NetworkName(tag, data)
 
@@ -190,9 +243,10 @@ def parse_tlv(data: str) -> dict[MeshcopTLVType | int, MeshcopTLVItem]:
         val = data_bytes[pos : pos + _len]
         pos += _len
 
-        # Once we have the value, we can log a warning about the unknown TLV
+        # Log unknown TLVs by type and length only: the value may hold
+        # network credentials, which don't belong in the log.
         if not isinstance(tag, MeshcopTLVType):
-            _LOGGER.warning("unknown TLV type %d=%r", raw_tag, val)
+            _LOGGER.warning("unknown TLV type %d (%d bytes)", raw_tag, _len)
 
         if tag in result:
             raise TLVError(f"duplicated tag {tag!r}")

tests/test_init_legacy.py

@@ -790,3 +790,90 @@ async def test_get_coprocessor_version_invalid(aioclient_mock: AiohttpClientMock
 
     with pytest.raises(python_otbr_api.OTBRError):
         await otbr.get_coprocessor_version()
+
+
+async def test_set_pending_dataset_tlvs(aioclient_mock: AiohttpClientMocker) -> None:
+    """Test set_pending_dataset_tlvs."""
+    otbr = python_otbr_api.OTBR(
+        BASE_URL, aioclient_mock.create_session(), key_format=KeyFormat.PASCAL_CASE
+    )
+
+    dataset = bytes.fromhex(
+        "0E080000000000010000000300000F35060004001FFFE0020811111111222222220708FDAD"
+        "70BFE5AA15DD051000112233445566778899AABBCCDDEEFF030E4F70656E54687265616444"
+        "656D6F010212340410445F2B5CA6F2A93A55CE570A70EFEECB0C0402A0F7F8"
+    )
+    aioclient_mock.get(f"{BASE_URL}/node/dataset/pending", status=HTTPStatus.NO_CONTENT)
+    aioclient_mock.put(f"{BASE_URL}/node/dataset/pending", status=HTTPStatus.CREATED)
+
+    await otbr.set_pending_dataset_tlvs(dataset)
+    assert aioclient_mock.call_count == 2
+    assert aioclient_mock.mock_calls[-1][0] == "PUT"
+    assert aioclient_mock.mock_calls[-1][1].path == "/node/dataset/pending"
+    assert aioclient_mock.mock_calls[-1][2] == dataset.hex()
+    assert aioclient_mock.mock_calls[-1][3]["If-None-Match"] == "*"
+
+
+async def test_set_pending_dataset_tlvs_thread_active(
+    aioclient_mock: AiohttpClientMocker,
+) -> None:
+    """Test set_pending_dataset_tlvs with error."""
+    otbr = python_otbr_api.OTBR(
+        BASE_URL, aioclient_mock.create_session(), key_format=KeyFormat.PASCAL_CASE
+    )
+
+    aioclient_mock.get(f"{BASE_URL}/node/dataset/pending", status=HTTPStatus.NO_CONTENT)
+    aioclient_mock.put(f"{BASE_URL}/node/dataset/pending", status=HTTPStatus.CONFLICT)
+
+    with pytest.raises(python_otbr_api.ThreadNetworkActiveError):
+        await otbr.set_pending_dataset_tlvs(b"")
+
+
+async def test_set_pending_dataset_tlvs_refused_while_pending(
+    aioclient_mock: AiohttpClientMocker,
+) -> None:
+    """Test a write is refused while a pending dataset is in place."""
+    otbr = python_otbr_api.OTBR(
+        BASE_URL, aioclient_mock.create_session(), key_format=KeyFormat.PASCAL_CASE
+    )
+
+    in_flight = "0E080000000000010000340400006699000300000C"
+    aioclient_mock.get(f"{BASE_URL}/node/dataset/pending", text=in_flight)
+
+    with pytest.raises(python_otbr_api.PendingDatasetConflictError):
+        await otbr.set_pending_dataset_tlvs(bytes.fromhex(in_flight))
+    assert aioclient_mock.call_count == 1
+
+
+async def test_set_pending_dataset_tlvs_refused_by_router(
+    aioclient_mock: AiohttpClientMocker,
+) -> None:
+    """Test the border router refusing the precondition is surfaced."""
+    otbr = python_otbr_api.OTBR(
+        BASE_URL, aioclient_mock.create_session(), key_format=KeyFormat.PASCAL_CASE
+    )
+
+    aioclient_mock.get(f"{BASE_URL}/node/dataset/pending", status=HTTPStatus.NO_CONTENT)
+    aioclient_mock.put(
+        f"{BASE_URL}/node/dataset/pending", status=HTTPStatus.PRECONDITION_FAILED
+    )
+
+    # The dedicated type, so a caller can tell "nothing was written because a
+    # dataset is in place" from a transport or protocol failure.
+    with pytest.raises(python_otbr_api.PendingDatasetConflictError):
+        await otbr.set_pending_dataset_tlvs(b"")
+
+
+async def test_set_pending_dataset_tlvs_202(
+    aioclient_mock: AiohttpClientMocker,
+) -> None:
+    """Test set_pending_dataset_tlvs with error."""
+    otbr = python_otbr_api.OTBR(
+        BASE_URL, aioclient_mock.create_session(), key_format=KeyFormat.PASCAL_CASE
+    )
+
+    aioclient_mock.get(f"{BASE_URL}/node/dataset/pending", status=HTTPStatus.NO_CONTENT)
+    aioclient_mock.put(f"{BASE_URL}/node/dataset/pending", status=HTTPStatus.ACCEPTED)
+
+    with pytest.raises(python_otbr_api.OTBRError):
+        await otbr.set_pending_dataset_tlvs(b"")

tests/test_tlv_parser.py

@@ -5,6 +5,7 @@
 from python_otbr_api.tlv_parser import (
     Timestamp,
     Channel,
+    DelayTimer,
     MeshcopTLVItem,
     MeshcopTLVType,
     NetworkName,
@@ -31,10 +32,10 @@
         MeshcopTLVType.IPV6_ADDRESS_TLV,
         bytes.fromhex("20010db8000000000000000000000001"),
     ),
-    MeshcopTLVType.PENDINGTIMESTAMP: MeshcopTLVItem(
+    MeshcopTLVType.PENDINGTIMESTAMP: Timestamp(
         MeshcopTLVType.PENDINGTIMESTAMP, bytes.fromhex("0000000000010000")
     ),
-    MeshcopTLVType.DELAYTIMER: MeshcopTLVItem(
+    MeshcopTLVType.DELAYTIMER: DelayTimer(
         MeshcopTLVType.DELAYTIMER, bytes.fromhex("00001388")
     ),
     MeshcopTLVType.COUNT: MeshcopTLVItem(MeshcopTLVType.COUNT, bytes.fromhex("03")),
@@ -252,3 +253,96 @@ def test_timestamp_parsing_full_integrity() -> None:
 
     # 3. Check Authoritative: Ensures the lowest bit is read correctly
     assert timestamp.authoritative is True
+
+
+def test_timestamp_from_values() -> None:
+    """Test constructing a timestamp from field values."""
+    timestamp = Timestamp.from_values(
+        MeshcopTLVType.ACTIVETIMESTAMP, seconds=400, ticks=32767, authoritative=True
+    )
+    assert timestamp.data == bytes.fromhex("000000000190FFFF")
+    assert timestamp.seconds == 400
+    assert timestamp.ticks == 32767
+    assert timestamp.authoritative is True
+
+    pending = Timestamp.from_values(MeshcopTLVType.PENDINGTIMESTAMP, seconds=1)
+    assert pending.tag == MeshcopTLVType.PENDINGTIMESTAMP
+    assert pending.data == bytes.fromhex("0000000000010000")
+    assert pending.ticks == 0
+    assert pending.authoritative is False
+
+    # The largest encodable timestamp survives a roundtrip
+    ceiling = Timestamp.from_values(
+        MeshcopTLVType.ACTIVETIMESTAMP, seconds=2**48 - 1, ticks=2**15 - 1
+    )
+    assert ceiling.seconds == 2**48 - 1
+    assert ceiling.ticks == 2**15 - 1
+
+
+@pytest.mark.parametrize(
+    ("seconds", "ticks", "msg"),
+    (
+        (2**48, 0, "timestamp seconds out of range"),
+        (-1, 0, "timestamp seconds out of range"),
+        (0, 2**15, "timestamp ticks out of range"),
+        (0, -1, "timestamp ticks out of range"),
+    ),
+)
+def test_timestamp_from_values_out_of_range(seconds, ticks, msg) -> None:
+    """Test constructing a timestamp from values which don't fit the wire format."""
+    with pytest.raises(TLVError, match=msg):
+        Timestamp.from_values(
+            MeshcopTLVType.ACTIVETIMESTAMP, seconds=seconds, ticks=ticks
+        )
+
+
+def test_timestamp_invalid_data() -> None:
+    """Test a malformed timestamp raises TLVError, not struct.error."""
+    with pytest.raises(TLVError, match="invalid timestamp '00'"):
+        Timestamp(MeshcopTLVType.ACTIVETIMESTAMP, bytes.fromhex("00"))
+    # Also via the parser, as it would arrive from user supplied TLVs
+    with pytest.raises(TLVError, match="invalid timestamp '00000000000100'"):
+        parse_tlv("0E0700000000000100")
+
+
+def test_pending_timestamp_and_delay_timer_parsed() -> None:
+    """Test PENDINGTIMESTAMP and DELAYTIMER decode to typed items."""
+    dataset = parse_tlv("33080000000000010000340400006699")
+    pending = dataset[MeshcopTLVType.PENDINGTIMESTAMP]
+    assert isinstance(pending, Timestamp)
+    assert pending.seconds == 1
+    delay = dataset[MeshcopTLVType.DELAYTIMER]
+    assert isinstance(delay, DelayTimer)
+    assert delay.delay == 0x6699
+
+
+def test_delay_timer_from_milliseconds() -> None:
+    """Test constructing a delay timer from a delay in milliseconds."""
+    delay = DelayTimer.from_milliseconds(5 * 60 * 1000)
+    assert delay.tag == MeshcopTLVType.DELAYTIMER
+    assert delay.data == bytes.fromhex("000493E0")
+    assert delay.delay == 300000
+    assert encode_tlv({MeshcopTLVType.DELAYTIMER: delay}) == "3404000493e0"
+
+
+@pytest.mark.parametrize("delay", (2**32, -1))
+def test_delay_timer_out_of_range(delay) -> None:
+    """Test constructing a delay timer which doesn't fit the wire format."""
+    with pytest.raises(TLVError, match="delay timer out of range"):
+        DelayTimer.from_milliseconds(delay)
+
+
+def test_delay_timer_invalid_data() -> None:
+    """Test a malformed delay timer raises TLVError, not struct.error."""
+    with pytest.raises(TLVError, match="invalid delay timer '00'"):
+        DelayTimer(MeshcopTLVType.DELAYTIMER, bytes.fromhex("00"))
+    # Also via the parser, as it would arrive from user supplied TLVs
+    with pytest.raises(TLVError, match="invalid delay timer '000000'"):
+        parse_tlv("3403000000")
+
+
+def test_unknown_tlv_value_not_logged(caplog: pytest.LogCaptureFixture) -> None:
+    """Test the unknown-TLV warning does not log the value."""
+    parse_tlv("BD03ABCDEF")
+    assert "unknown TLV type 189 (3 bytes)" in caplog.text
+    assert "abcdef" not in caplog.text.lower()