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
43 changes: 32 additions & 11 deletions python_otbr_api/tlv_parser.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,9 @@
from dataclasses import dataclass, field
from enum import IntEnum
import struct
import logging

_LOGGER = logging.getLogger(__name__)


class TLVError(Exception):
Expand Down Expand Up @@ -124,7 +127,7 @@ def _encode_item(item: MeshcopTLVItem) -> bytes:
return struct.pack(f"!BB{data_len}s", item.tag, data_len, item.data)


def encode_tlv(items: dict[MeshcopTLVType, MeshcopTLVItem]) -> str:
def encode_tlv(items: dict[MeshcopTLVType | int, MeshcopTLVItem]) -> str:
"""Encode a TLV encoded dataset to a hex string.

Raises if the TLV is invalid.
Expand All @@ -137,7 +140,7 @@ def encode_tlv(items: dict[MeshcopTLVType, MeshcopTLVItem]) -> str:
return result.hex()


def _parse_item(tag: MeshcopTLVType, data: bytes) -> MeshcopTLVItem:
def _parse_item(tag: MeshcopTLVType | int, data: bytes) -> MeshcopTLVItem:
"""Parse a TLV encoded dataset item."""
if tag == MeshcopTLVType.ACTIVETIMESTAMP:
return Timestamp(tag, data)
Expand All @@ -149,7 +152,7 @@ def _parse_item(tag: MeshcopTLVType, data: bytes) -> MeshcopTLVItem:
return MeshcopTLVItem(tag, data)


def parse_tlv(data: str) -> dict[MeshcopTLVType, MeshcopTLVItem]:
def parse_tlv(data: str) -> dict[MeshcopTLVType | int, MeshcopTLVItem]:
"""Parse a TLV encoded dataset.

Raises if the TLV is invalid.
Expand All @@ -160,19 +163,37 @@ def parse_tlv(data: str) -> dict[MeshcopTLVType, MeshcopTLVItem]:
raise TLVError("invalid tlvs") from err
result = {}
pos = 0
while pos < len(data_bytes):
try:
tag = MeshcopTLVType(data_bytes[pos])
except ValueError as err:
raise TLVError(f"unknown type {data_bytes[pos]}") from err
length = len(data_bytes)

while pos < length:
if pos + 2 > length:
raise TLVError("truncated tlv header")

raw_tag = data_bytes[pos]
pos += 1

# Unknown tags should still be passed through
tag: MeshcopTLVType | int

try:
tag = MeshcopTLVType(raw_tag)
except ValueError:
tag = raw_tag

_len = data_bytes[pos]
pos += 1

if pos + _len > length:
raise TLVError(f"expected {_len} bytes for tag {tag!r}, got {length - pos}")

val = data_bytes[pos : pos + _len]
if len(val) < _len:
raise TLVError(f"expected {_len} bytes for {tag.name}, got {len(val)}")
pos += _len

# Once we have the value, we can log a warning about the unknown TLV
if not isinstance(tag, MeshcopTLVType):
_LOGGER.warning("unknown TLV type %d=%r", raw_tag, val)

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Considering the point of the PR it to let unknown tags pass through, I'm not sure we should log a warning? I'd suggest to lower severity to debug since this shouldn't matter to the user.


if tag in result:
raise TLVError(f"duplicated tag {tag.name}")
raise TLVError(f"duplicated tag {tag!r}")
result[tag] = _parse_item(tag, val)
return result
15 changes: 11 additions & 4 deletions tests/test_tlv_parser.py

@emontnemery emontnemery Feb 3, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why do we modify the existing tests instead of adding new tests?
Also, is all the the new code covered by tests?

Edit: The only added code not covered is the handling of duplicated tags

Original file line number Diff line number Diff line change
Expand Up @@ -47,14 +47,15 @@ def test_encode_tlv() -> None:
MeshcopTLVType.SECURITYPOLICY: MeshcopTLVItem(
MeshcopTLVType.SECURITYPOLICY, bytes.fromhex("02a0f7f8")
),
189: MeshcopTLVItem(189, bytes.fromhex("abcdef")),
}
dataset_tlv = encode_tlv(dataset)
assert (
dataset_tlv
== (
"0E080000000000010000000300000F35060004001FFFE0020811111111222222220708FDAD"
"70BFE5AA15DD051000112233445566778899AABBCCDDEEFF030E4F70656E54687265616444"
"656D6F010212340410445F2B5CA6F2A93A55CE570A70EFEECB0C0402A0F7F8"
"656D6F010212340410445F2B5CA6F2A93A55CE570A70EFEECB0C0402A0F7F8BD03ABCDEF"
).lower()
)

Expand All @@ -64,7 +65,7 @@ def test_parse_tlv() -> None:
dataset_tlv = (
"0E080000000000010000000300000F35060004001FFFE0020811111111222222220708FDAD70BF"
"E5AA15DD051000112233445566778899AABBCCDDEEFF030E4F70656E54687265616444656D6F01"
"0212340410445F2B5CA6F2A93A55CE570A70EFEECB0C0402A0F7F8"
"0212340410445F2B5CA6F2A93A55CE570A70EFEECB0C0402A0F7F8BD03ABCDEF"
)
dataset = parse_tlv(dataset_tlv)
assert dataset == {
Expand Down Expand Up @@ -98,6 +99,7 @@ def test_parse_tlv() -> None:
MeshcopTLVType.CHANNELMASK: MeshcopTLVItem(
MeshcopTLVType.CHANNELMASK, bytes.fromhex("0004001fffe0")
),
189: MeshcopTLVItem(189, bytes.fromhex("abcdef")),
}


Expand Down Expand Up @@ -138,12 +140,17 @@ def test_parse_tlv_apple() -> None:
(
"FF",
TLVError,
"unknown type 255",
"truncated tlv header",
),
(
"FF01",
TLVError,
"expected 1 bytes for tag 255, got 0",
),
(
"030E4F70656E54687265616444656D",
TLVError,
"expected 14 bytes for NETWORKNAME, got 13",
"expected 14 bytes for tag <MeshcopTLVType.NETWORKNAME: 3>, got 13",
),
(
"030E4F70656E54687265616444656DFF",
Expand Down