diff --git a/src/ndn/encoding/__init__.py b/src/ndn/encoding/__init__.py index a3646e9..d5b788d 100644 --- a/src/ndn/encoding/__init__.py +++ b/src/ndn/encoding/__init__.py @@ -3,6 +3,7 @@ from .name import * from .signer import * from .tlv_model import * +from .tlv_model_v2 import tlv_encode, tlv_parse, NDNName, tlv_get_arg, tlv_set_arg from .ndn_format_0_3 import * from .ndnlp_v2 import * @@ -13,6 +14,7 @@ __all__.extend(name.__all__) __all__.extend(signer.__all__) __all__.extend(tlv_model.__all__) +__all__ += ['tlv_encode', 'tlv_parse', 'NDNName', 'tlv_get_arg', 'tlv_set_arg'] __all__.extend(ndn_format_0_3.__all__) __all__.extend(ndnlp_v2.__all__) diff --git a/src/ndn/encoding/tlv_model_v2.py b/src/ndn/encoding/tlv_model_v2.py index e69de29..a3a3ace 100644 --- a/src/ndn/encoding/tlv_model_v2.py +++ b/src/ndn/encoding/tlv_model_v2.py @@ -0,0 +1,977 @@ +# ----------------------------------------------------------------------------- +# Copyright (C) 2019-2020 The python-ndn authors +# +# This file is part of python-ndn. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# ----------------------------------------------------------------------------- +""" +Dataclass-based TLV encoding/decoding (v2 API). + +Usage:: + + from dataclasses import dataclass, field + from typing import List, Optional + from ndn.encoding import tlv_encode, tlv_parse, NDNName + + @dataclass + class Inner: + value: int = field(default=None, metadata={'tlv_type': 0x01}) + + @dataclass + class Outer: + name: NDNName = field(default=None, metadata={'tlv_type': 0x07}) + count: int = field(default=None, metadata={'tlv_type': 0x0a}) + payload: bytes = field(default=None, metadata={'tlv_type': 0x15}) + sub: Inner = field(default=None, metadata={'tlv_type': 0x16}) + tags: List[bytes] = field(default_factory=list, + metadata={'tlv_type': 0x17}) + + wire = tlv_encode(obj) + obj = tlv_parse(Outer, wire) + +Field-kind inference from Python annotation +------------------------------------------- ++--------------------------------------------+----------+------------------+ +| Annotation | Kind | Old equivalent | ++============================================+==========+==================+ +| int / Enum / Flag subclass | uint | UintField | ++--------------------------------------------+----------+------------------+ +| bool | bool | BoolField | ++--------------------------------------------+----------+------------------+ +| bytes / bytearray / memoryview | bytes | BytesField | ++--------------------------------------------+----------+------------------+ +| str | str | BytesField | +| | | (is_string=True) | ++--------------------------------------------+----------+------------------+ +| NDNName (sentinel) | name | NameField | ++--------------------------------------------+----------+------------------+ +| Any @dataclass type | model | ModelField | ++--------------------------------------------+----------+------------------+ +| List[T] | repeated | RepeatedField | ++--------------------------------------------+----------+------------------+ +| Dict[K, V] | map | MapField | ++--------------------------------------------+----------+------------------+ +| None + field_type='offset_marker' | (zero) | OffsetMarker | ++--------------------------------------------+----------+------------------+ +| bytes + field_type='sig_value' | (special)| SignatureValue | ++--------------------------------------------+----------+------------------+ +| NDNName + field_type='interest_name' | (special)| InterestNameField| ++--------------------------------------------+----------+------------------+ + +Supported metadata keys +----------------------- +``'tlv_type'`` int TLV type number (required except for offset_marker) +``'fixed_len'`` int Force uint value width: 1, 2, 4, or 8 bytes +``'ignore_critical' bool Suppress DecodeError for nested model parsing +``'field_type'`` str Explicit kind override when inference is insufficient + +For **map** fields (``Dict[K, V]``): +``'val_tlv_type'`` int TLV type for map values (required) + +For **sig_value** fields: +``'cover_start'`` str Name of the offset_marker field where sig coverage begins +``'digest_cover_start' str Same or different offset_marker; where digest coverage begins +``'digest_cover_end'`` str Offset_marker after sig_value; where digest coverage ends + +Signature machinery markers (set by caller before tlv_encode / tlv_parse): +``markers['##signer']`` Signer instance; absent means unsigned +``markers['##need_digest']`` True ⟹ insert/compute ParametersSha256DigestComponent + +Signature machinery markers (set by tlv_encode / tlv_parse internally): +``markers['##sig_covered_part']`` list[memoryview | bytes]: regions covered by sig +``markers['##sig_value_buf']`` writable memoryview into the placeholder bytes +``markers['##shrink_len']`` int: bytes trimmed from end after sig finalization +``markers['##digest_buf']`` writable memoryview into the digest component value +``markers[fname]`` int: recorded byte offset for each offset_marker field +""" +import dataclasses +import struct +import typing +from enum import Enum, Flag +from hashlib import sha256 + +from .tlv_type import BinaryStr, VarBinaryStr, is_binary_str +from .tlv_var import write_tl_num, parse_tl_num, get_tl_num_size +from .name import Name, Component +from .tlv_model import DecodeError + + +__all__ = [ + 'tlv_encode', 'tlv_parse', 'NDNName', 'DecodeError', + 'tlv_get_arg', 'tlv_set_arg', +] + +# Kinds that occupy zero wire bytes and may not have a 'tlv_type' metadata key. +_ZERO_WIRE_KINDS = frozenset({'offset_marker'}) + + +# --------------------------------------------------------------------------- +# NDNName sentinel — used as a type annotation for NDN Name fields +# --------------------------------------------------------------------------- + +class NDNName: + """ + Sentinel annotation type that marks a field as an NDN Name. + + Use it wherever you would have used :class:`~ndn.encoding.NameField` in + the old metaclass API:: + + name: NDNName = field(default=None, metadata={'tlv_type': 0x07}) + # repeated Names: + names: List[NDNName] = field(default_factory=list, + metadata={'tlv_type': 0x07}) + + The actual runtime value is :any:`FormalName` (a list of encoded + component bytes), exactly as returned by the old NameField. + """ + + +# --------------------------------------------------------------------------- +# Annotation helpers +# --------------------------------------------------------------------------- + +def _unwrap_optional(annotation): + """Return T for Optional[T] = Union[T, None]; otherwise return unchanged.""" + if typing.get_origin(annotation) is typing.Union: + args = [a for a in typing.get_args(annotation) if a is not type(None)] + if len(args) == 1: + return args[0] + return annotation + + +def _infer_kind(annotation, metadata: dict) -> str: + """ + Determine the TLV field kind from a Python type annotation plus metadata. + + Returns one of: ``'uint'``, ``'bool'``, ``'bytes'``, ``'str'``, + ``'name'``, ``'model'``, ``'repeated'``. + + The ``'field_type'`` metadata key overrides automatic inference. + """ + if 'field_type' in metadata: + return metadata['field_type'] + + annotation = _unwrap_optional(annotation) + origin = typing.get_origin(annotation) + + if origin is list: + return 'repeated' + if origin is dict: + return 'map' + if annotation is NDNName: + return 'name' + # bool must be checked before int since bool is a subclass of int + if annotation is bool: + return 'bool' + if annotation is int or ( + isinstance(annotation, type) + and issubclass(annotation, (int, Enum, Flag)) + and annotation is not bool): + return 'uint' + if annotation in (bytes, bytearray, memoryview): + return 'bytes' + if annotation is str: + return 'str' + if dataclasses.is_dataclass(annotation): + return 'model' + + raise TypeError( + f'Cannot infer TLV field kind from annotation {annotation!r}. ' + f"Use metadata key 'field_type' to override." + ) + + +def _element_annotation(annotation): + """Extract T from List[T]; falls back to bytes.""" + annotation = _unwrap_optional(annotation) + args = typing.get_args(annotation) + return args[0] if args else bytes + + +def _map_annotations(annotation): + """Extract (K, V) from Dict[K, V]; falls back to (str, bytes).""" + annotation = _unwrap_optional(annotation) + args = typing.get_args(annotation) + if len(args) == 2: + return args[0], args[1] + return str, bytes + + +def _map_key_meta(metadata: dict) -> dict: + """Build a synthetic metadata dict for a map key sub-field.""" + return {'tlv_type': metadata['tlv_type']} + + +def _map_val_meta(metadata: dict) -> dict: + """Build a synthetic metadata dict for a map value sub-field.""" + m = {'tlv_type': metadata['val_tlv_type']} + if 'ignore_critical' in metadata: + m['ignore_critical'] = metadata['ignore_critical'] + return m + + +# --------------------------------------------------------------------------- +# Interest-name helpers (used by both pass-1 and pass-2) +# --------------------------------------------------------------------------- + +def _encoded_length_interest_name(fname: str, val, metadata: dict, + markers: dict) -> int: + """ + Size pass for an Interest Name field. + + Mirrors ``InterestNameField.encoded_length``. If ``markers['##need_digest']`` + is truthy and the name does not already contain a + ``ParametersSha256DigestComponent``, 34 extra bytes are reserved for one. + """ + if val is None: + return 0 + type_num = metadata['tlv_type'] + need_digest = markers.get('##need_digest', False) + + # Normalize to a list of component bytes. + if isinstance(val, str): + name = Name.from_str(val) + elif is_binary_str(val): + name = Name.decode(val)[0] + else: + name = list(val) + for i, comp in enumerate(name): + if isinstance(comp, str): + name[i] = Component.from_str(Component.escape_str(comp)) + elif not is_binary_str(comp): + raise TypeError(f'{fname}: invalid name component {comp!r}') + + # Locate an existing ParametersSha256DigestComponent (at most one allowed). + digest_pos = None + if need_digest: + for i, comp in enumerate(name): + if Component.get_type(comp) == Component.TYPE_PARAMETERS_SHA256: + if digest_pos is None: + digest_pos = i + else: + raise ValueError( + f'{fname}: multiple ParametersSha256DigestComponent in name') + + markers[f'{fname}##digest_pos'] = digest_pos + markers[f'{fname}##preprocessed_name'] = name + + comp_total = sum(len(c) for c in name) + if need_digest and digest_pos is None: + # Reserve space for a new digest component: T(1B) + L(1B) + V(32B). + comp_total += (get_tl_num_size(Component.TYPE_PARAMETERS_SHA256) + + get_tl_num_size(32) + 32) + + markers[f'{fname}##name_value_len'] = comp_total + return get_tl_num_size(type_num) + get_tl_num_size(comp_total) + comp_total + + +def _encode_into_interest_name(fname: str, val, metadata: dict, markers: dict, + wire: VarBinaryStr, offset: int) -> int: + """ + Write pass for an Interest Name field. + + Mirrors ``InterestNameField.encode_into``. Appends non-digest name + components to ``markers['##sig_covered_part']`` (wire slices) and stores + the writable digest-value buffer in ``markers['##digest_buf']``. + """ + if val is None: + return 0 + type_num = metadata['tlv_type'] + name = markers[f'{fname}##preprocessed_name'] + comp_total = markers[f'{fname}##name_value_len'] + digest_pos = markers[f'{fname}##digest_pos'] + need_digest = markers.get('##need_digest', False) + sig_covered_part = markers.setdefault('##sig_covered_part', []) + + origin = offset + t_sz = write_tl_num(type_num, wire, offset); offset += t_sz + l_sz = write_tl_num(comp_total, wire, offset); offset += l_sz + cover_start = offset + + for i, comp in enumerate(name): + comp_len = len(comp) + wire[offset:offset + comp_len] = comp + if i == digest_pos: + if offset > cover_start: + sig_covered_part.append(wire[cover_start:offset]) + # Value of the digest component sits after T + L (each 1 byte for + # TYPE_PARAMETERS_SHA256=2 < 253 and length=32 < 253). + c_t_sz = get_tl_num_size(Component.TYPE_PARAMETERS_SHA256) + c_l_sz = get_tl_num_size(32) + markers['##digest_buf'] = wire[offset + c_t_sz + c_l_sz:offset + comp_len] + cover_start = offset + comp_len + offset += comp_len + + if offset > cover_start: + sig_covered_part.append(wire[cover_start:offset]) + + if need_digest and digest_pos is None: + # Append a new ParametersSha256DigestComponent at the end of the name. + c_t_sz = write_tl_num(Component.TYPE_PARAMETERS_SHA256, wire, offset) + offset += c_t_sz + c_l_sz = write_tl_num(32, wire, offset) + offset += c_l_sz + markers['##digest_buf'] = wire[offset:offset + 32] + # Keep the preprocessed name up-to-date for get_final_name use. + name.append(bytes(wire[offset - c_t_sz - c_l_sz:offset + 32])) + offset += 32 + + return offset - origin + + +# --------------------------------------------------------------------------- +# Post-encoding finalization (signature + SHA-256 digest) +# --------------------------------------------------------------------------- + +def _finalize_encode(markers: dict, mv, model_end: int) -> int: + """ + Called by :func:`tlv_encode` after all bytes have been written. + + 1. Asks the signer to fill in the signature-value placeholder, updates the + inline length byte if the actual signature is shorter (ECDSA), and + records ``markers['##shrink_len']``. + 2. If ``markers['##need_digest']`` is set, computes ``SHA-256`` over the + digest-covered range and writes it into the name's digest-component + placeholder (``markers['##digest_buf']``). + + Returns *shrink_size* (0 for fixed-length signature schemes like HMAC/EdDSA). + All offsets in *markers* are absolute positions within *mv*. + """ + signer = markers.get('##signer') + shrink_size = 0 + + if signer is not None and '##sig_value_buf' in markers: + sig_value_buf = markers['##sig_value_buf'] + alloc_size = len(sig_value_buf) + real_size = signer.write_signature_value( + sig_value_buf, markers.get('##sig_covered_part', [])) + shrink_size = alloc_size - real_size + markers['##shrink_len'] = shrink_size + if shrink_size > 0: + if alloc_size >= 253: + raise ValueError( + f'Signature with variable length ≥ 253 bytes is not supported ' + f'(allocated {alloc_size})') + markers['##sig_wire_l_field'][0] = real_size + + if markers.get('##need_digest') and '##digest_buf' in markers: + d_start_field = markers.get('##_digest_cover_start_field') + d_end_field = markers.get('##_digest_cover_end_field') + d_start = markers[d_start_field] if (d_start_field and d_start_field in markers) else 0 + d_end = markers[d_end_field] if (d_end_field and d_end_field in markers) else model_end + d_end -= shrink_size + markers['##digest_buf'][:] = sha256(bytes(mv[d_start:d_end])).digest() + + return shrink_size + + +# --------------------------------------------------------------------------- +# Encoding — pass 1: size computation +# --------------------------------------------------------------------------- + +def _uint_value_len(val: int, fname: str, fixed_len) -> int: + if fixed_len is not None: + n = fixed_len + elif val <= 0xFF: + n = 1 + elif val <= 0xFFFF: + n = 2 + elif val <= 0xFFFFFFFF: + n = 4 + else: + n = 8 + if val >= 0x100 ** n: + raise ValueError(f'{fname}={val!r} cannot be encoded into {n} bytes') + return n + + +def _encoded_length_field(fname: str, val, kind: str, annotation, metadata: dict, + markers: dict) -> int: + """ + Compute the encoded byte count of one TLV field (T + L + V). + + Intermediate values are cached in *markers* under ``fname##...`` keys, + exactly mirroring the convention used by the v1 :class:`~ndn.encoding.Field` + subclasses. Returns 0 when the field is absent (*val* is ``None``/falsy + for bool). + """ + # Zero-wire kinds: handled before looking up tlv_type. + if kind == 'offset_marker': + return 0 + + if kind == 'sig_value': + signer = markers.get('##signer') + if signer is None: + return 0 + type_num = metadata['tlv_type'] + sig_size = signer.get_signature_value_size() + markers[f'{fname}##sig_size'] = sig_size + markers.setdefault('##sig_covered_part', []) + return get_tl_num_size(type_num) + get_tl_num_size(sig_size) + sig_size + + if kind == 'interest_name': + return _encoded_length_interest_name(fname, val, metadata, markers) + + type_num = metadata['tlv_type'] + + # BoolField: present if truthy, absent otherwise + if kind == 'bool': + return (get_tl_num_size(type_num) + 1) if val else 0 + + if val is None: + return 0 + + if kind == 'uint': + if isinstance(val, (Enum, Flag)): + val = val.value + if not isinstance(val, int) or val < 0: + raise TypeError(f'{fname}={val!r} is not a non-negative integer') + fixed_len = metadata.get('fixed_len') + vlen = _uint_value_len(val, fname, fixed_len) + markers[f'{fname}##encoded_length'] = vlen + # L for uint is always 1 byte because vlen ∈ {1,2,4,8} < 253 + return get_tl_num_size(type_num) + 1 + vlen + + if kind in ('bytes', 'str'): + if isinstance(val, str): + raw = val.encode('utf-8') + markers[f'{fname}##encoded_str'] = raw + else: + raw = val + n = len(raw) + return get_tl_num_size(type_num) + get_tl_num_size(n) + n + + if kind == 'name': + # Normalise to list-of-components or a pre-encoded binary blob + name_val = val + if isinstance(name_val, str): + name_val = Name.from_str(name_val) + elif not is_binary_str(name_val): + if hasattr(name_val, '__iter__'): + name_val = list(name_val) + for i, comp in enumerate(name_val): + if isinstance(comp, str): + name_val[i] = Component.from_str(Component.escape_str(comp)) + elif not is_binary_str(comp): + raise TypeError(f'{fname}: invalid name component type') + else: + raise TypeError(f'{fname}: invalid name type') + if isinstance(name_val, list): + total_with_tl = Name.encoded_length(name_val) + else: + total_with_tl = len(name_val) + markers[f'{fname}##preprocessed_name'] = name_val + markers[f'{fname}##encoded_length_with_tl'] = total_with_tl + return total_with_tl + + if kind == 'model': + inner_markers: dict = {} + length = _encoded_length_model(val, inner_markers) + markers[f'{fname}##inner_markers'] = inner_markers + markers[f'{fname}##encoded_length'] = length + return get_tl_num_size(type_num) + get_tl_num_size(length) + length + + if kind == 'repeated': + if not val: + return 0 + elem_ann = _element_annotation(annotation) + elem_kind = _infer_kind(elem_ann, metadata) + total = 0 + for i, ele in enumerate(val): + total += _encoded_length_field( + f'{fname}[{i}]', ele, elem_kind, elem_ann, metadata, markers) + return total + + if kind == 'map': + if not val: + return 0 + key_ann, val_ann = _map_annotations(annotation) + key_meta = _map_key_meta(metadata) + vl_meta = _map_val_meta(metadata) + key_kind = _infer_kind(key_ann, key_meta) + vl_kind = _infer_kind(val_ann, vl_meta) + total = 0 + for i, (k, v) in enumerate(val.items()): + total += _encoded_length_field( + f'{fname}[{i}#k]', k, key_kind, key_ann, key_meta, markers) + total += _encoded_length_field( + f'{fname}[{i}#v]', v, vl_kind, val_ann, vl_meta, markers) + return total + + raise TypeError(f'Unknown field kind {kind!r} for {fname!r}') + + +def _encoded_length_model(obj, markers: dict) -> int: + """Compute the total encoded length for all TLV fields of a dataclass object.""" + cls = type(obj) + hints = typing.get_type_hints(cls) + total = 0 + for f in dataclasses.fields(cls): + ann = hints[f.name] + kind = _infer_kind(ann, f.metadata) + if kind not in _ZERO_WIRE_KINDS and 'tlv_type' not in f.metadata: + continue + total += _encoded_length_field( + f.name, getattr(obj, f.name), kind, ann, f.metadata, markers) + markers['##encoded_length'] = total + return total + + +# --------------------------------------------------------------------------- +# Encoding — pass 2: write bytes +# --------------------------------------------------------------------------- + +def _encode_into_field(fname: str, val, kind: str, annotation, metadata: dict, + markers: dict, wire: VarBinaryStr, offset: int) -> int: + """ + Write one TLV field into *wire* at *offset*. + + *wire* must be a writable :class:`memoryview` (or :class:`bytearray`). + Returns the number of bytes written. Must be called after the matching + :func:`_encoded_length_field` call so that ``markers`` is populated. + """ + # Zero-wire kinds: handled before looking up tlv_type. + if kind == 'offset_marker': + markers[fname] = offset + return 0 + + if kind == 'sig_value': + signer = markers.get('##signer') + if signer is None: + return 0 + type_num = metadata['tlv_type'] + sig_size = markers[f'{fname}##sig_size'] + # Collect the covered region: from cover_start up to current offset. + cover_start_field = metadata.get('cover_start') + cover_start = markers.get(cover_start_field, 0) if cover_start_field else 0 + markers.setdefault('##sig_covered_part', []).append(wire[cover_start:offset]) + # Store digest-coverage field names for _finalize_encode. + for mkey in ('digest_cover_start', 'digest_cover_end'): + if mkey in metadata: + markers[f'##_{mkey}_field'] = metadata[mkey] + # Write T + L (stored for in-place shrink) + placeholder V. + t_sz = write_tl_num(type_num, wire, offset) + l_off = offset + t_sz + l_sz = write_tl_num(sig_size, wire, l_off) + markers['##sig_wire_l_field'] = wire[l_off:l_off + l_sz] + v_start = l_off + l_sz + markers['##sig_value_buf'] = wire[v_start:v_start + sig_size] + return t_sz + l_sz + sig_size + + if kind == 'interest_name': + return _encode_into_interest_name(fname, val, metadata, markers, wire, offset) + + type_num = metadata['tlv_type'] + + if kind == 'bool': + if val: + t_size = write_tl_num(type_num, wire, offset) + wire[offset + t_size] = 0 # L = 0 + return t_size + 1 + return 0 + + if val is None: + return 0 + + if kind == 'uint': + if isinstance(val, (Enum, Flag)): + val = val.value + vlen = markers[f'{fname}##encoded_length'] + t_size = write_tl_num(type_num, wire, offset) + if vlen == 1: + struct.pack_into('!BB', wire, offset + t_size, 1, val) + elif vlen == 2: + struct.pack_into('!BH', wire, offset + t_size, 2, val) + elif vlen == 4: + struct.pack_into('!BI', wire, offset + t_size, 4, val) + else: + struct.pack_into('!BQ', wire, offset + t_size, 8, val) + return t_size + 1 + vlen # T + L(1 byte) + V + + if kind in ('bytes', 'str'): + raw = markers.get(f'{fname}##encoded_str') + if raw is None: + raw = val.encode('utf-8') if isinstance(val, str) else val + n = len(raw) + t_size = write_tl_num(type_num, wire, offset) + l_size = write_tl_num(n, wire, offset + t_size) + v_start = offset + t_size + l_size + wire[v_start:v_start + n] = raw # zero-copy slice assignment + return t_size + l_size + n + + if kind == 'name': + name_val = markers[f'{fname}##preprocessed_name'] + name_len = markers[f'{fname}##encoded_length_with_tl'] + if isinstance(name_val, list): + Name.encode(name_val, wire, offset) + else: + wire[offset:offset + name_len] = name_val + return name_len + + if kind == 'model': + inner_markers = markers[f'{fname}##inner_markers'] + length = markers[f'{fname}##encoded_length'] + t_size = write_tl_num(type_num, wire, offset) + l_size = write_tl_num(length, wire, offset + t_size) + _encode_into_model(val, inner_markers, wire, offset + t_size + l_size) + return t_size + l_size + length + + if kind == 'repeated': + if not val: + return 0 + elem_ann = _element_annotation(annotation) + elem_kind = _infer_kind(elem_ann, metadata) + total = 0 + for i, ele in enumerate(val): + total += _encode_into_field( + f'{fname}[{i}]', ele, elem_kind, elem_ann, metadata, markers, + wire, offset + total) + return total + + if kind == 'map': + if not val: + return 0 + key_ann, val_ann = _map_annotations(annotation) + key_meta = _map_key_meta(metadata) + vl_meta = _map_val_meta(metadata) + key_kind = _infer_kind(key_ann, key_meta) + vl_kind = _infer_kind(val_ann, vl_meta) + total = 0 + for i, (k, v) in enumerate(val.items()): + total += _encode_into_field( + f'{fname}[{i}#k]', k, key_kind, key_ann, key_meta, markers, + wire, offset + total) + total += _encode_into_field( + f'{fname}[{i}#v]', v, vl_kind, val_ann, vl_meta, markers, + wire, offset + total) + return total + + raise TypeError(f'Unknown field kind {kind!r} for {fname!r}') + + +def _encode_into_model(obj, markers: dict, wire: VarBinaryStr, offset: int) -> None: + """Write all TLV fields of a dataclass object into *wire* starting at *offset*.""" + cls = type(obj) + hints = typing.get_type_hints(cls) + for f in dataclasses.fields(cls): + ann = hints[f.name] + kind = _infer_kind(ann, f.metadata) + if kind not in _ZERO_WIRE_KINDS and 'tlv_type' not in f.metadata: + continue + offset += _encode_into_field( + f.name, getattr(obj, f.name), kind, ann, f.metadata, markers, wire, offset) + + +# --------------------------------------------------------------------------- +# Public encode entry point +# --------------------------------------------------------------------------- + +def tlv_encode(obj, wire=None, offset: int = 0, markers: dict = None): + """ + Encode a dataclass TLV object. + + **Allocating form** — ``tlv_encode(obj)`` + Allocates a new :class:`bytearray`, fills it, and returns it. + + **In-place form** — ``tlv_encode(obj, wire, offset=0)`` + Encodes into an existing *wire* (:class:`bytearray` or writable + :class:`memoryview`) starting at *offset*. Returns a zero-copy + :class:`memoryview` slice of the written region. + + :param obj: dataclass instance to encode. + :param wire: optional writable buffer. + :param offset: starting byte offset within *wire*. + :param markers: optional shared markers dict (for multi-model coordination). + :return: :class:`bytearray` (allocating) or :class:`memoryview` (in-place). + """ + if markers is None: + markers = {} + total = _encoded_length_model(obj, markers) + if wire is None: + buf = bytearray(total) + mv = memoryview(buf) + _encode_into_model(obj, markers, mv, 0) + shrink = _finalize_encode(markers, mv, total) + if shrink: + # Can't resize bytearray while memoryview exports are live (the sig/digest + # slices in markers still reference mv). Return a trimmed copy instead. + return bytearray(mv[:total - shrink]) + return buf + mv = memoryview(wire) + _encode_into_model(obj, markers, mv, offset) + shrink = _finalize_encode(markers, mv, offset + total) + return mv[offset:offset + total - shrink] + + +# --------------------------------------------------------------------------- +# Parsing +# --------------------------------------------------------------------------- + +def _make_default_instance(cls): + """ + Create a dataclass instance with all fields set to their defaults. + + Uses ``object.__new__`` to bypass ``__init__``, then sets each field: + - ``field(default=X)`` → X + - ``field(default_factory=F)`` → F() + - no default → None (same behaviour as old TlvModel.parse) + """ + obj = object.__new__(cls) + for f in dataclasses.fields(cls): + if f.default is not dataclasses.MISSING: + object.__setattr__(obj, f.name, f.default) + elif f.default_factory is not dataclasses.MISSING: + object.__setattr__(obj, f.name, f.default_factory()) + else: + object.__setattr__(obj, f.name, None) + return obj + + +def _parse_value(fname: str, kind: str, annotation, metadata: dict, + wire, offset: int, length: int, offset_btl: int, + ignore_critical: bool): + """ + Parse a single TLV *value* (V only, not T or L) from *wire*. + + :param fname: field name (for error messages). + :param kind: field kind string. + :param annotation: resolved Python type annotation. + :param metadata: dataclass field metadata dict. + :param wire: memoryview of the full wire buffer. + :param offset: byte offset of V within *wire*. + :param length: byte length of V. + :param offset_btl: byte offset of the TLV's T field within *wire* + (used by NameField to pass to ``Name.decode``). + :param ignore_critical: forwarded to nested ``tlv_parse`` calls. + :return: the parsed Python value. + """ + if kind == 'bool': + return True + + if kind == 'uint': + if length == 1: + raw = struct.unpack_from('!B', wire, offset)[0] + elif length == 2: + raw = struct.unpack_from('!H', wire, offset)[0] + elif length == 4: + raw = struct.unpack_from('!I', wire, offset)[0] + elif length == 8: + raw = struct.unpack_from('!Q', wire, offset)[0] + else: + raise ValueError( + f'{fname}: uint value length must be 1, 2, 4, or 8; got {length}') + # Auto-convert to the annotated Enum/Flag type if applicable + inner = _unwrap_optional(annotation) + if (isinstance(inner, type) + and issubclass(inner, (Enum, Flag)) + and inner is not int): + try: + return inner(raw) + except ValueError: + pass + return raw + + if kind == 'bytes': + return wire[offset:offset + length] # zero-copy memoryview slice + + if kind == 'str': + return bytes(wire[offset:offset + length]).decode('utf-8') + + if kind == 'name': + return Name.decode(wire, offset_btl)[0] + + if kind == 'model': + inner_cls = _unwrap_optional(annotation) + ignore = metadata.get('ignore_critical', ignore_critical) + return tlv_parse(inner_cls, wire[offset:offset + length], ignore) + + raise TypeError(f'Unknown kind {kind!r} for {fname!r}') + + +def tlv_parse(cls, wire, ignore_critical: bool = False, markers: dict = None): + """ + Parse a TLV-encoded buffer into a fresh dataclass instance. + + Matching follows NDN ordering rules — fields are matched in their + declaration order within *cls* (parent class fields come first, as per + standard Python dataclass inheritance). + + Unknown critical TLV types (odd type numbers) raise + :exc:`~ndn.encoding.DecodeError` unless *ignore_critical* is ``True``. + + Bytes-typed fields (``bytes``, ``bytearray``, ``memoryview`` annotations) + are returned as zero-copy :class:`memoryview` slices into *wire*. + + :param cls: dataclass class to parse into. + :param wire: TLV-encoded buffer + (:class:`bytes`, :class:`bytearray`, or :class:`memoryview`). + :param ignore_critical: suppress :exc:`DecodeError` for unknown critical + TLV types. + :param markers: optional dict for out-of-band state (offset_marker positions, + sig/digest buffers). A fresh ``{}`` is used when ``None``. + :return: populated dataclass instance. + :raises DecodeError: unknown critical TLV type encountered. + """ + if markers is None: + markers = {} + + # Wrap in memoryview for zero-copy slicing throughout the parse + if isinstance(wire, memoryview): + mv = wire + else: + mv = memoryview(wire if isinstance(wire, (bytes, bytearray)) else bytes(wire)) + + hints = typing.get_type_hints(cls) + ordered = [] + for f in dataclasses.fields(cls): + ann = hints[f.name] + kind = _infer_kind(ann, f.metadata) + if kind not in _ZERO_WIRE_KINDS and 'tlv_type' not in f.metadata: + continue + ordered.append((f.name, f.metadata, kind, ann)) + + obj = _make_default_instance(cls) + offset = 0 + field_pos = 0 # lowest index still eligible for matching + + while offset < len(mv): + offset_btl = offset + typ, sz_t = parse_tl_num(mv, offset) + offset += sz_t + length, sz_l = parse_tl_num(mv, offset) + offset += sz_l + + found = False + for i in range(field_pos, len(ordered)): + fname, meta, kind, ann = ordered[i] + if kind == 'offset_marker': + continue # never matches a wire TLV type + + if meta['tlv_type'] != typ: + continue + + # Advance any offset_markers between field_pos and i. + for j in range(field_pos, i): + jname, _, jkind, _ = ordered[j] + if jkind == 'offset_marker': + markers[jname] = offset_btl + + if kind == 'repeated': + elem_ann = _element_annotation(ann) + elem_kind = _infer_kind(elem_ann, meta) + val = _parse_value(fname, elem_kind, elem_ann, meta, + mv, offset, length, offset_btl, ignore_critical) + lst = getattr(obj, fname) + if lst is None: + lst = [] + object.__setattr__(obj, fname, lst) + lst.append(val) + field_pos = i # stay at i to accept more elements + + elif kind == 'map': + # Two-phase parse: consume key, then immediately read value TLV. + key_ann, val_ann = _map_annotations(ann) + key_meta = _map_key_meta(meta) + vl_meta = _map_val_meta(meta) + key_kind = _infer_kind(key_ann, key_meta) + vl_kind = _infer_kind(val_ann, vl_meta) + + dct = getattr(obj, fname) + if dct is None: + dct = {} + object.__setattr__(obj, fname, dct) + idx = len(dct) + + key = _parse_value(f'{fname}[{idx}#k]', key_kind, key_ann, key_meta, + mv, offset, length, offset_btl, ignore_critical) + + # advance past key value → now at the value TLV + offset += length + offset_btl = offset + _val_typ, _sz_t2 = parse_tl_num(mv, offset) + offset += _sz_t2 + length, _sz_l2 = parse_tl_num(mv, offset) + offset += _sz_l2 + + val = _parse_value(f'{fname}[{idx}#v]', vl_kind, val_ann, vl_meta, + mv, offset, length, offset_btl, ignore_critical) + dct[key] = val + field_pos = i # stay at i to accept more pairs + + elif kind == 'sig_value': + # Extract sig buffer; append covered region to ##sig_covered_part. + sig_buf = mv[offset:offset + length] + markers['##sig_value_buf'] = sig_buf + cover_start_field = meta.get('cover_start') + if cover_start_field is not None: + cover_start = markers.get(cover_start_field) + if cover_start is not None: + markers.setdefault('##sig_covered_part', []).append( + mv[cover_start:offset_btl]) + object.__setattr__(obj, fname, sig_buf) + field_pos = i + 1 + + elif kind == 'interest_name': + # Decode name; split into sig-covered components and digest buf. + name = Name.decode(mv, offset_btl)[0] + sig_cp = markers.setdefault('##sig_covered_part', []) + for comp in name: + if Component.get_type(comp) == Component.TYPE_PARAMETERS_SHA256: + markers['##digest_buf'] = Component.get_value(comp) + else: + sig_cp.append(comp) + object.__setattr__(obj, fname, name) + field_pos = i + 1 + + else: + val = _parse_value(fname, kind, ann, meta, + mv, offset, length, offset_btl, ignore_critical) + object.__setattr__(obj, fname, val) + field_pos = i + 1 + + found = True + break + + if not found and (typ & 1) and not ignore_critical: + raise DecodeError( + f'unknown critical TLV type {typ:#x} is unrecognized, ' + f'redundant, or out-of-order') + + offset += length + + return obj + + +# --------------------------------------------------------------------------- +# Marker helpers (convenience wrappers for the markers dict) +# --------------------------------------------------------------------------- + +def tlv_get_arg(markers: dict, key: str, default=None): + """ + Read a value from the *markers* dict used by :func:`tlv_encode` / + :func:`tlv_parse`. + + Equivalent to ``markers.get(key, default)``. + """ + return markers.get(key, default) + + +def tlv_set_arg(markers: dict, key: str, val) -> None: + """ + Write a value into the *markers* dict used by :func:`tlv_encode` / + :func:`tlv_parse`. + + Equivalent to ``markers[key] = val``. + """ + markers[key] = val diff --git a/tests/encoding/tlv_model_v2_test.py b/tests/encoding/tlv_model_v2_test.py new file mode 100644 index 0000000..06f55ed --- /dev/null +++ b/tests/encoding/tlv_model_v2_test.py @@ -0,0 +1,1238 @@ +# ----------------------------------------------------------------------------- +# Copyright (C) 2019-2020 The python-ndn authors +# +# This file is part of python-ndn. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# ----------------------------------------------------------------------------- +"""Tests for the dataclass-based TLV v2 API (tlv_encode / tlv_parse).""" +from dataclasses import dataclass, field +from enum import IntEnum, IntFlag +from hashlib import sha256 +from typing import Dict, List, Optional + +import pytest + +from ndn.encoding import ( + tlv_encode, tlv_parse, NDNName, DecodeError, tlv_get_arg, tlv_set_arg, + # v1 equivalents used for binary-compatibility checks + TlvModel, UintField, BoolField, BytesField, NameField, ModelField, + RepeatedField, Name, Component, + # Signer interface + Signer, +) + + +# --------------------------------------------------------------------------- +# Shared dataclass fixtures (defined at module scope for get_type_hints) +# --------------------------------------------------------------------------- + +@dataclass +class _Inner: + val: int = field(default=None, metadata={'tlv_type': 0x01}) + + +@dataclass +class _Outer: + inner: _Inner = field(default=None, metadata={'tlv_type': 0x02}) + + +@dataclass +class _RepeatedUint: + words: List[int] = field(default_factory=list, + metadata={'tlv_type': 0x01, 'fixed_len': 2}) + + +@dataclass +class _RepeatedModel: + items: List[_Inner] = field(default_factory=list, metadata={'tlv_type': 0x10}) + + +# --------------------------------------------------------------------------- +# TestUintField +# --------------------------------------------------------------------------- + +class TestUintField: + """UintField: variable-width and fixed-width non-negative integers.""" + + def test_min_width_1_byte(self): + @dataclass + class M: + x: int = field(default=None, metadata={'tlv_type': 0x03}) + + obj = M(x=0) + wire = tlv_encode(obj) + assert wire == b'\x03\x01\x00' + assert tlv_parse(M, wire).x == 0 + + def test_min_width_2_bytes(self): + @dataclass + class M: + x: int = field(default=None, metadata={'tlv_type': 0x03}) + + obj = M(x=0x0100) + wire = tlv_encode(obj) + assert wire == b'\x03\x02\x01\x00' + assert tlv_parse(M, wire).x == 0x0100 + + def test_min_width_4_bytes(self): + @dataclass + class M: + x: int = field(default=None, metadata={'tlv_type': 0x03}) + + obj = M(x=0x00010000) + wire = tlv_encode(obj) + assert wire == b'\x03\x04\x00\x01\x00\x00' + assert tlv_parse(M, wire).x == 0x00010000 + + def test_min_width_8_bytes(self): + @dataclass + class M: + x: int = field(default=None, metadata={'tlv_type': 0x03}) + + obj = M(x=0x0000000100000000) + wire = tlv_encode(obj) + assert wire == b'\x03\x08\x00\x00\x00\x01\x00\x00\x00\x00' + assert tlv_parse(M, wire).x == 0x0000000100000000 + + def test_fixed_len_1(self): + @dataclass + class M: + x: int = field(default=None, metadata={'tlv_type': 0x1b, 'fixed_len': 1}) + + obj = M(x=3) + wire = tlv_encode(obj) + # T=0x1b L=0x01 V=0x03 + assert wire == b'\x1b\x01\x03' + assert tlv_parse(M, wire).x == 3 + + def test_fixed_len_2(self): + @dataclass + class M: + x: int = field(default=None, metadata={'tlv_type': 0x03, 'fixed_len': 2}) + + obj = M(x=5) + wire = tlv_encode(obj) + assert wire == b'\x03\x02\x00\x05' + assert tlv_parse(M, wire).x == 5 + + def test_none_omitted(self): + @dataclass + class M: + x: int = field(default=None, metadata={'tlv_type': 0x03}) + y: int = field(default=None, metadata={'tlv_type': 0x05}) + + wire = tlv_encode(M(x=None, y=7)) + assert wire == b'\x05\x01\x07' + p = tlv_parse(M, wire) + assert p.x is None + assert p.y == 7 + + def test_optional_annotation(self): + @dataclass + class M: + x: Optional[int] = field(default=None, metadata={'tlv_type': 0x03}) + + obj = M(x=42) + wire = tlv_encode(obj) + assert tlv_parse(M, wire).x == 42 + + +class TestUintFieldEnum: + """IntEnum and IntFlag auto-conversion on parse.""" + + def test_intenum_roundtrip(self): + class FaceType(IntEnum): + PERSISTENT = 1 + ON_DEMAND = 2 + + @dataclass + class M: + face_type: FaceType = field(default=None, metadata={'tlv_type': 0x84}) + + obj = M(face_type=FaceType.PERSISTENT) + wire = tlv_encode(obj) + p = tlv_parse(M, wire) + assert p.face_type == FaceType.PERSISTENT + assert isinstance(p.face_type, FaceType) + + def test_intflag_roundtrip(self): + class Flags(IntFlag): + A = 1 + B = 2 + + @dataclass + class M: + f: Flags = field(default=None, metadata={'tlv_type': 0x06}) + + obj = M(f=Flags.A | Flags.B) + wire = tlv_encode(obj) + p = tlv_parse(M, wire) + assert p.f == Flags.A | Flags.B + assert isinstance(p.f, Flags) + + def test_unknown_enum_value_returns_int(self): + class Color(IntEnum): + RED = 1 + + @dataclass + class M: + c: Color = field(default=None, metadata={'tlv_type': 0x01}) + + # Wire with value 99, which is not a valid Color + wire = b'\x01\x01\x63' + p = tlv_parse(M, wire) + assert p.c == 99 + assert type(p.c) is int + + +# --------------------------------------------------------------------------- +# TestBoolField +# --------------------------------------------------------------------------- + +class TestBoolField: + """BoolField: 0-length TLV present when truthy, absent otherwise.""" + + def test_present(self): + @dataclass + class M: + flag: bool = field(default=None, metadata={'tlv_type': 0x12}) + + wire = tlv_encode(M(flag=True)) + # T=0x12 L=0x00 + assert wire == b'\x12\x00' + assert tlv_parse(M, wire).flag is True + + def test_absent_when_false(self): + @dataclass + class M: + flag: bool = field(default=None, metadata={'tlv_type': 0x12}) + + assert tlv_encode(M(flag=False)) == b'' + assert tlv_encode(M(flag=None)) == b'' + + def test_absent_field_returns_none(self): + @dataclass + class M: + flag: bool = field(default=None, metadata={'tlv_type': 0x12}) + x: int = field(default=None, metadata={'tlv_type': 0x14}) + + wire = tlv_encode(M(flag=None, x=1)) + p = tlv_parse(M, wire) + assert p.flag is None + assert p.x == 1 + + def test_optional_annotation(self): + @dataclass + class M: + flag: Optional[bool] = field(default=None, metadata={'tlv_type': 0x12}) + + wire = tlv_encode(M(flag=True)) + assert tlv_parse(M, wire).flag is True + + +# --------------------------------------------------------------------------- +# TestBytesField +# --------------------------------------------------------------------------- + +class TestBytesField: + """BytesField: raw bytes and UTF-8 strings.""" + + def test_bytes_roundtrip(self): + @dataclass + class M: + data: bytes = field(default=None, metadata={'tlv_type': 0x15}) + + obj = M(data=b'\x01\x02\x03') + wire = tlv_encode(obj) + assert wire == b'\x15\x03\x01\x02\x03' + p = tlv_parse(M, wire) + assert bytes(p.data) == b'\x01\x02\x03' + + def test_bytes_parse_is_memoryview(self): + @dataclass + class M: + data: bytes = field(default=None, metadata={'tlv_type': 0x15}) + + wire = b'\x15\x02\xde\xad' + p = tlv_parse(M, wire) + assert isinstance(p.data, memoryview) + + def test_str_field_roundtrip(self): + @dataclass + class M: + label: str = field(default=None, metadata={'tlv_type': 0x16}) + + obj = M(label='hello') + wire = tlv_encode(obj) + assert wire == b'\x16\x05hello' + assert tlv_parse(M, wire).label == 'hello' + + def test_str_field_unicode(self): + @dataclass + class M: + s: str = field(default=None, metadata={'tlv_type': 0x16}) + + obj = M(s='日本語') + wire = tlv_encode(obj) + assert tlv_parse(M, wire).s == '日本語' + + def test_bytearray_annotation(self): + @dataclass + class M: + data: bytearray = field(default=None, metadata={'tlv_type': 0x15}) + + wire = tlv_encode(M(data=bytearray(b'abc'))) + p = tlv_parse(M, wire) + assert bytes(p.data) == b'abc' + + def test_memoryview_annotation(self): + @dataclass + class M: + data: memoryview = field(default=None, metadata={'tlv_type': 0x15}) + + src = bytearray(b'\xca\xfe') + wire = tlv_encode(M(data=memoryview(src))) + p = tlv_parse(M, wire) + assert isinstance(p.data, memoryview) + assert bytes(p.data) == b'\xca\xfe' + + def test_large_value_multibyte_length(self): + """Length field uses multi-byte varint when value ≥ 253 bytes.""" + @dataclass + class M: + data: bytes = field(default=None, metadata={'tlv_type': 0x15}) + + payload = bytes(range(256)) + wire = tlv_encode(M(data=payload)) + # Length 256 → encoded as 0xFD 0x01 0x00 (3 bytes) + assert wire[1:4] == b'\xfd\x01\x00' + p = tlv_parse(M, wire) + assert bytes(p.data) == payload + + +# --------------------------------------------------------------------------- +# TestNameField +# --------------------------------------------------------------------------- + +class TestNameField: + """NDNName: NDN Name TLV via string, FormalName list, or binary.""" + + def test_from_string(self): + @dataclass + class M: + name: NDNName = field(default=None, metadata={'tlv_type': 0x07}) + + obj = M(name='/foo/bar') + wire = tlv_encode(obj) + # 0x07 0x0a [0x08 0x03 foo] [0x08 0x03 bar] + assert wire == b'\x07\x0a\x08\x03foo\x08\x03bar' + p = tlv_parse(M, wire) + assert Name.to_str(p.name) == '/foo/bar' + + def test_from_formal_name(self): + @dataclass + class M: + name: NDNName = field(default=None, metadata={'tlv_type': 0x07}) + + formal = Name.from_str('/a/b') + wire = tlv_encode(M(name=formal)) + p = tlv_parse(M, wire) + assert Name.to_str(p.name) == '/a/b' + + def test_empty_name(self): + @dataclass + class M: + name: NDNName = field(default=None, metadata={'tlv_type': 0x07}) + + wire = tlv_encode(M(name='/')) + p = tlv_parse(M, wire) + assert p.name == [] + + def test_none_omitted(self): + @dataclass + class M: + name: NDNName = field(default=None, metadata={'tlv_type': 0x07}) + + assert tlv_encode(M(name=None)) == b'' + + def test_repeated_names(self): + """List[NDNName] — multiple Name TLVs with the same type number.""" + @dataclass + class M: + names: List[NDNName] = field(default_factory=list, + metadata={'tlv_type': 0x07}) + + obj = M(names=['/foo', '/bar']) + wire = tlv_encode(obj) + p = tlv_parse(M, wire) + assert len(p.names) == 2 + assert Name.to_str(p.names[0]) == '/foo' + assert Name.to_str(p.names[1]) == '/bar' + + +# --------------------------------------------------------------------------- +# TestModelField +# --------------------------------------------------------------------------- + +class TestModelField: + """ModelField: nested dataclass, recursively encoded.""" + + def test_basic_nested(self): + wire = tlv_encode(_Outer(inner=_Inner(val=255))) + # 0x02 (outer T) 0x03 (outer L) 0x01 0x01 0xFF + assert wire == b'\x02\x03\x01\x01\xff' + p = tlv_parse(_Outer, wire) + assert p.inner.val == 255 + + def test_absent_nested(self): + wire = tlv_encode(_Outer(inner=None)) + assert wire == b'' + p = tlv_parse(_Outer, wire) + assert p.inner is None + + def test_deeply_nested(self): + @dataclass + class Level3: + x: int = field(default=None, metadata={'tlv_type': 0x01}) + + @dataclass + class Level2: + sub: Level3 = field(default=None, metadata={'tlv_type': 0x10}) + + @dataclass + class Level1: + sub: Level2 = field(default=None, metadata={'tlv_type': 0x20}) + + obj = Level1(sub=Level2(sub=Level3(x=7))) + wire = tlv_encode(obj) + p = tlv_parse(Level1, wire) + assert p.sub.sub.x == 7 + + def test_ignore_critical_propagated(self): + """ignore_critical in metadata is forwarded to nested parse.""" + @dataclass + class Inner: + x: int = field(default=None, metadata={'tlv_type': 0x01}) + + @dataclass + class Outer: + sub: Inner = field(default=None, + metadata={'tlv_type': 0x10, + 'ignore_critical': True}) + + # Inject an unknown critical TLV (type 0x03, odd) inside sub + inner_wire = b'\x03\x01\x00' + sub_wire = bytes([0x10, len(inner_wire)]) + inner_wire + # With ignore_critical via metadata, this must not raise + p = tlv_parse(Outer, sub_wire) + assert p.sub.x is None + + +# --------------------------------------------------------------------------- +# TestRepeatedField +# --------------------------------------------------------------------------- + +class TestRepeatedField: + """RepeatedField: multiple TLVs of the same type, no outer wrapper.""" + + def test_uint_elements(self): + wire = tlv_encode(_RepeatedUint(words=[0, 1, 2])) + # Each word: T=0x01 L=0x02 V=2-byte big-endian + assert wire == b'\x01\x02\x00\x00\x01\x02\x00\x01\x01\x02\x00\x02' + p = tlv_parse(_RepeatedUint, wire) + assert p.words == [0, 1, 2] + + def test_bytes_elements(self): + @dataclass + class M: + tags: List[bytes] = field(default_factory=list, + metadata={'tlv_type': 0x17}) + + obj = M(tags=[b'a', b'bb', b'ccc']) + wire = tlv_encode(obj) + assert wire == b'\x17\x01a\x17\x02bb\x17\x03ccc' + p = tlv_parse(M, wire) + assert [bytes(t) for t in p.tags] == [b'a', b'bb', b'ccc'] + + def test_str_elements(self): + @dataclass + class M: + labels: List[str] = field(default_factory=list, + metadata={'tlv_type': 0x16}) + + obj = M(labels=['hello', 'world']) + wire = tlv_encode(obj) + p = tlv_parse(M, wire) + assert p.labels == ['hello', 'world'] + + def test_model_elements(self): + wire = tlv_encode(_RepeatedModel(items=[_Inner(val=10), _Inner(val=20)])) + p = tlv_parse(_RepeatedModel, wire) + assert [i.val for i in p.items] == [10, 20] + + def test_empty_list_produces_no_bytes(self): + wire = tlv_encode(_RepeatedUint(words=[])) + assert wire == b'' + + def test_default_factory_list_initialised_on_parse(self): + """A repeated field with no default_factory should still get a list on parse.""" + @dataclass + class M: + items: List[int] = field(metadata={'tlv_type': 0x05}) + + wire = b'\x05\x01\x01\x05\x01\x02' + p = tlv_parse(M, wire) + assert p.items == [1, 2] + + +# --------------------------------------------------------------------------- +# TestOrdering +# --------------------------------------------------------------------------- + +class TestOrdering: + """NDN TLV ordering rules: forward-only matching, critical type handling.""" + + def test_unknown_even_type_skipped(self): + """Unknown even TLV types are silently ignored (non-critical).""" + @dataclass + class M: + x: int = field(default=None, metadata={'tlv_type': 0x04}) + + # Wire: unknown even type 0x02, then known type 0x04 + wire = b'\x02\x01\xff\x04\x01\x07' + p = tlv_parse(M, wire) + assert p.x == 7 + + def test_unknown_odd_type_raises(self): + """Unknown odd TLV types are critical — must raise DecodeError.""" + @dataclass + class M: + x: int = field(default=None, metadata={'tlv_type': 0x04}) + + wire = b'\x03\x01\x00\x04\x01\x07' + with pytest.raises(DecodeError): + tlv_parse(M, wire) + + def test_ignore_critical_suppresses_error(self): + @dataclass + class M: + x: int = field(default=None, metadata={'tlv_type': 0x04}) + + wire = b'\x03\x01\x00\x04\x01\x07' + p = tlv_parse(M, wire, ignore_critical=True) + assert p.x == 7 + + def test_out_of_order_critical_field_raises(self): + """ + A critical (odd type) field appearing out-of-order is unrecognised and + must raise DecodeError. Even types are non-critical and silently dropped. + """ + @dataclass + class M: + a: int = field(default=None, metadata={'tlv_type': 0x03}) # odd = critical + b: int = field(default=None, metadata={'tlv_type': 0x05}) # odd = critical + + # b (0x05) comes first; after matching it, field_pos advances past a (0x03). + # The parser then sees 0x03 as unknown critical → DecodeError. + wire = b'\x05\x01\x02\x03\x01\x01' + with pytest.raises(DecodeError): + tlv_parse(M, wire) + + def test_out_of_order_even_field_silently_dropped(self): + """Even (non-critical) fields seen out-of-order are silently skipped.""" + @dataclass + class M: + a: int = field(default=None, metadata={'tlv_type': 0x02}) # even + b: int = field(default=None, metadata={'tlv_type': 0x04}) # even + + # b before a — a is dropped silently (non-critical) + wire = b'\x04\x01\x02\x02\x01\x01' + p = tlv_parse(M, wire) + assert p.b == 2 + assert p.a is None + + def test_out_of_order_non_critical_silently_skipped(self): + @dataclass + class M: + a: int = field(default=None, metadata={'tlv_type': 0x04}) # even, non-critical when out of order + b: int = field(default=None, metadata={'tlv_type': 0x06}) + + # b before a — a is even so silently skipped + wire = b'\x06\x01\x02\x04\x01\x01' + p = tlv_parse(M, wire) + assert p.b == 2 + assert p.a is None + + +# --------------------------------------------------------------------------- +# TestInheritance +# --------------------------------------------------------------------------- + +class TestInheritance: + """Dataclass inheritance: parent fields come first (no IncludeBase needed).""" + + def test_parent_fields_encoded_first(self): + @dataclass + class Base: + x: int = field(default=None, metadata={'tlv_type': 0x01}) + + @dataclass + class Child(Base): + y: int = field(default=None, metadata={'tlv_type': 0x03}) + + wire = tlv_encode(Child(x=1, y=2)) + # x (0x01) must appear before y (0x03) + assert wire == b'\x01\x01\x01\x03\x01\x02' + p = tlv_parse(Child, wire) + assert p.x == 1 and p.y == 2 + + def test_child_only_fields(self): + @dataclass + class Base: + x: int = field(default=None, metadata={'tlv_type': 0x01}) + + @dataclass + class Child(Base): + y: int = field(default=None, metadata={'tlv_type': 0x03}) + + wire = tlv_encode(Child(x=None, y=5)) + assert wire == b'\x03\x01\x05' + p = tlv_parse(Child, wire) + assert p.x is None + assert p.y == 5 + + def test_non_tlv_fields_ignored(self): + """Fields without 'tlv_type' in metadata are silently skipped.""" + @dataclass + class M: + internal: str = field(default='ignored') # no metadata + x: int = field(default=None, metadata={'tlv_type': 0x01}) + + wire = tlv_encode(M(internal='should_not_appear', x=42)) + assert wire == b'\x01\x01\x2a' + p = tlv_parse(M, wire) + assert p.x == 42 + + +# --------------------------------------------------------------------------- +# TestZeroCopyAndInPlace +# --------------------------------------------------------------------------- + +class TestZeroCopyAndInPlace: + """Zero-copy memoryview slices and in-place buffer encoding.""" + + def test_bytes_parse_shares_buffer(self): + """Parsed bytes field is a memoryview slice — no copy.""" + @dataclass + class M: + data: bytes = field(default=None, metadata={'tlv_type': 0x15}) + + wire = bytearray(b'\x15\x04\xde\xad\xbe\xef') + p = tlv_parse(M, wire) + assert isinstance(p.data, memoryview) + assert bytes(p.data) == b'\xde\xad\xbe\xef' + + def test_inplace_encode_returns_memoryview(self): + @dataclass + class M: + x: int = field(default=None, metadata={'tlv_type': 0x03}) + + buf = bytearray(20) + mv = tlv_encode(M(x=7), buf, offset=5) + assert isinstance(mv, memoryview) + assert bytes(mv) == b'\x03\x01\x07' + # Bytes written at correct position + assert buf[5:8] == b'\x03\x01\x07' + # Surrounding bytes untouched + assert buf[:5] == b'\x00' * 5 + assert buf[8:] == b'\x00' * 12 + + def test_inplace_matches_standalone(self): + @dataclass + class M: + x: int = field(default=None, metadata={'tlv_type': 0x03}) + y: bytes = field(default=None, metadata={'tlv_type': 0x15}) + + obj = M(x=300, y=b'hello') + standalone = tlv_encode(obj) + buf = bytearray(len(standalone) + 10) + mv = tlv_encode(obj, buf, offset=3) + assert bytes(mv) == bytes(standalone) + + def test_inplace_memoryview_buffer(self): + """In-place encoding also works with a memoryview target.""" + @dataclass + class M: + x: int = field(default=None, metadata={'tlv_type': 0x05}) + + buf = bytearray(10) + mv_buf = memoryview(buf) + result = tlv_encode(M(x=1), mv_buf, offset=2) + assert bytes(result) == b'\x05\x01\x01' + + +# --------------------------------------------------------------------------- +# TestDefaultHandling +# --------------------------------------------------------------------------- + +class TestDefaultHandling: + """Default values and field initialisation during parse.""" + + def test_field_with_explicit_default_preserved_if_absent(self): + @dataclass + class M: + x: int = field(default=42, metadata={'tlv_type': 0x01}) + + # Wire that does not contain field x + wire = b'' + p = tlv_parse(M, wire) + assert p.x == 42 + + def test_field_without_default_is_none_if_absent(self): + @dataclass + class M: + x: int = field(metadata={'tlv_type': 0x01}) + + wire = b'' + p = tlv_parse(M, wire) + assert p.x is None + + def test_default_factory_list_preserved_if_absent(self): + wire = tlv_encode(_RepeatedUint(words=[])) + p = tlv_parse(_RepeatedUint, wire) + assert p.words == [] + + +# --------------------------------------------------------------------------- +# TestBinaryCompatibility +# --------------------------------------------------------------------------- + +class TestBinaryCompatibility: + """Byte-for-byte compatibility with the v1 TlvModel metaclass API.""" + + def test_uint_compat(self): + class V1(TlvModel): + sig_type = UintField(0x1b, fixed_len=1) + nonce = UintField(0x26) + + @dataclass + class V2: + sig_type: int = field(default=None, + metadata={'tlv_type': 0x1b, 'fixed_len': 1}) + nonce: int = field(default=None, metadata={'tlv_type': 0x26}) + + v1 = V1(); v1.sig_type = 3; v1.nonce = 42 + assert bytes(v1.encode()) == bytes(tlv_encode(V2(sig_type=3, nonce=42))) + + def test_bool_compat(self): + class V1(TlvModel): + flag = BoolField(0x12) + count = UintField(0x0a) + + @dataclass + class V2: + flag: bool = field(default=None, metadata={'tlv_type': 0x12}) + count: int = field(default=None, metadata={'tlv_type': 0x0a}) + + for flag_val in (True, False, None): + v1 = V1(); v1.flag = flag_val; v1.count = 5 + v2 = V2(flag=flag_val, count=5) + assert bytes(v1.encode()) == bytes(tlv_encode(v2)) + + def test_bytes_compat(self): + class V1(TlvModel): + raw = BytesField(0x15) + label = BytesField(0x16, is_string=True) + + @dataclass + class V2: + raw: bytes = field(default=None, metadata={'tlv_type': 0x15}) + label: str = field(default=None, metadata={'tlv_type': 0x16}) + + v1 = V1(); v1.raw = b'\x01\x02\x03'; v1.label = 'hi' + v2 = V2(raw=b'\x01\x02\x03', label='hi') + assert bytes(v1.encode()) == bytes(tlv_encode(v2)) + + def test_name_compat(self): + class V1(TlvModel): + name = NameField() + + @dataclass + class V2: + name: NDNName = field(default=None, metadata={'tlv_type': 0x07}) + + v1 = V1(); v1.name = '/foo/bar' + v2 = V2(name='/foo/bar') + assert bytes(v1.encode()) == bytes(tlv_encode(v2)) + + def test_model_compat(self): + class V1Inner(TlvModel): + val = UintField(0x01) + + class V1Outer(TlvModel): + inner = ModelField(0x10, V1Inner) + + @dataclass + class V2Inner: + val: int = field(default=None, metadata={'tlv_type': 0x01}) + + @dataclass + class V2Outer: + inner: V2Inner = field(default=None, metadata={'tlv_type': 0x10}) + + v1 = V1Outer(); v1.inner = V1Inner(); v1.inner.val = 99 + v2 = V2Outer(inner=V2Inner(val=99)) + assert bytes(v1.encode()) == bytes(tlv_encode(v2)) + + def test_repeated_uint_compat(self): + class V1(TlvModel): + words = RepeatedField(UintField(0x01, fixed_len=2)) + + v1 = V1(); v1.words = [0, 1, 2] + v2 = _RepeatedUint(words=[0, 1, 2]) + assert bytes(v1.encode()) == bytes(tlv_encode(v2)) + + def test_repeated_model_compat(self): + class V1Inner(TlvModel): + val = UintField(0x01) + + class V1Rep(TlvModel): + items = RepeatedField(ModelField(0x10, V1Inner)) + + v1 = V1Rep() + r1 = V1Inner(); r1.val = 10 + r2 = V1Inner(); r2.val = 20 + v1.items = [r1, r2] + + v2 = _RepeatedModel(items=[_Inner(val=10), _Inner(val=20)]) + assert bytes(v1.encode()) == bytes(tlv_encode(v2)) + + def test_parse_interop(self): + """Wire produced by v1 can be parsed by v2 and vice-versa.""" + class V1(TlvModel): + name = NameField() + count = UintField(0x0a) + + @dataclass + class V2: + name: NDNName = field(default=None, metadata={'tlv_type': 0x07}) + count: int = field(default=None, metadata={'tlv_type': 0x0a}) + + v1 = V1(); v1.name = '/test'; v1.count = 7 + wire_from_v1 = bytes(v1.encode()) + + p = tlv_parse(V2, wire_from_v1) + assert Name.to_str(p.name) == '/test' + assert p.count == 7 + + v2 = V2(name='/test', count=7) + wire_from_v2 = bytes(tlv_encode(v2)) + + p2 = V1.parse(wire_from_v2) + assert Name.to_str(p2.name) == '/test' + assert p2.count == 7 + + +# --------------------------------------------------------------------------- +# MapField tests +# --------------------------------------------------------------------------- + +@dataclass +class _StrBytesMap: + entries: Dict[str, bytes] = field(default_factory=dict, metadata={ + 'tlv_type': 0x21, + 'val_tlv_type': 0x23, + }) + + +@dataclass +class _Inner2: + value: int = field(default=None, metadata={'tlv_type': 0x01}) + + +@dataclass +class _StrModelMap: + entries: Dict[str, _Inner2] = field(default_factory=dict, metadata={ + 'tlv_type': 0x21, + 'val_tlv_type': 0x22, + }) + + +class TestMapField: + def test_str_bytes_roundtrip(self): + obj = _StrBytesMap(entries={'alpha': b'\x01\x02', 'beta': b'\x03'}) + wire = tlv_encode(obj) + p = tlv_parse(_StrBytesMap, wire) + assert list(p.entries.keys()) == ['alpha', 'beta'] + assert bytes(p.entries['alpha']) == b'\x01\x02' + assert bytes(p.entries['beta']) == b'\x03' + + def test_insertion_order_preserved(self): + """Dict round-trip must preserve the original key insertion order.""" + obj = _StrBytesMap(entries={'z': b'\x00', 'a': b'\x01', 'm': b'\x02'}) + p = tlv_parse(_StrBytesMap, tlv_encode(obj)) + assert list(p.entries.keys()) == ['z', 'a', 'm'] + + def test_empty_map_produces_no_bytes(self): + obj = _StrBytesMap(entries={}) + assert tlv_encode(obj) == b'' + + def test_none_map_produces_no_bytes(self): + obj = _StrBytesMap(entries=None) + assert tlv_encode(obj) == b'' + + def test_none_map_defaults_to_empty_on_parse(self): + """Parsing wire with no map TLVs leaves entries as the default_factory value.""" + p = tlv_parse(_StrBytesMap, b'') + assert p.entries == {} + + def test_str_model_map_roundtrip(self): + obj = _StrModelMap(entries={'x': _Inner2(value=7), 'y': _Inner2(value=99)}) + wire = tlv_encode(obj) + p = tlv_parse(_StrModelMap, wire) + assert list(p.entries.keys()) == ['x', 'y'] + assert p.entries['x'].value == 7 + assert p.entries['y'].value == 99 + + def test_v1_compat_wire(self): + """v2 map encoding must be byte-for-byte identical to v1 MapField.""" + from ndn.encoding import MapField, BytesField + + class V1Map(TlvModel): + entries = MapField(BytesField(0x21, is_string=True), BytesField(0x23)) + + v1 = V1Map() + v1.entries['alpha'] = b'\x01\x02' + v1.entries['beta'] = b'\x03' + v1_wire = bytes(v1.encode()) + + v2 = _StrBytesMap(entries={'alpha': b'\x01\x02', 'beta': b'\x03'}) + v2_wire = bytes(tlv_encode(v2)) + + assert v1_wire == v2_wire + + def test_v1_produced_wire_parsed_by_v2(self): + from ndn.encoding import MapField, BytesField + + class V1Map(TlvModel): + entries = MapField(BytesField(0x21, is_string=True), BytesField(0x23)) + + v1 = V1Map() + v1.entries['hello'] = b'\xde\xad' + wire = bytes(v1.encode()) + + p = tlv_parse(_StrBytesMap, wire) + assert bytes(p.entries['hello']) == b'\xde\xad' + + def test_bytes_values_are_memoryview_zero_copy(self): + obj = _StrBytesMap(entries={'k': b'\xca\xfe'}) + wire = tlv_encode(obj) + p = tlv_parse(_StrBytesMap, wire) + assert isinstance(p.entries['k'], memoryview) + + +# --------------------------------------------------------------------------- +# Signature machinery tests +# --------------------------------------------------------------------------- + +# ── Simple fixed-length mock signer (HMAC-like) ────────────────────────────── + +class _HmacSigner(Signer): + """Deterministic 32-byte 'HMAC' signer using SHA-256(key || content).""" + SIG_SIZE = 32 + + def __init__(self, key: bytes = b'secret'): + self._key = key + + def write_signature_info(self, sig_info): + sig_info.signature_type = 4 # HMAC_WITH_SHA256 + + def get_signature_value_size(self) -> int: + return self.SIG_SIZE + + def write_signature_value(self, wire, contents) -> int: + h = sha256(self._key) + for blk in contents: + h.update(bytes(blk)) + sig = h.digest() + wire[:] = sig + return len(sig) + + def verify(self, sig: bytes, contents) -> bool: + buf = bytearray(self.SIG_SIZE) + mv = memoryview(buf) + self.write_signature_value(mv, contents) + return bytes(buf) == bytes(sig) + + +# ── Variable-length mock signer (ECDSA-like, sometimes shorter) ─────────────── + +class _EcdsaSigner(Signer): + """Always signs with 71 bytes, but reports max 72 (tests shrink path).""" + MAX_SIZE = 72 + REAL_SIZE = 71 + + def write_signature_info(self, sig_info): + sig_info.signature_type = 3 # SHA256_WITH_ECDSA + + def get_signature_value_size(self): + return self.MAX_SIZE + + def write_signature_value(self, wire, contents): + for i in range(self.REAL_SIZE): + wire[i] = i & 0xFF + return self.REAL_SIZE + + +# ── Data-like model (no digest, no interest-name) ──────────────────────────── + +@dataclass +class _SigInfo: + signature_type: int = field(default=None, metadata={'tlv_type': 0x1b, 'fixed_len': 1}) + + +@dataclass +class _DataValue: + _sig_cover_start: None = field(default=None, metadata={'field_type': 'offset_marker'}) + name: NDNName = field(default=None, metadata={'tlv_type': 0x07}) + content: Optional[bytes] = field(default=None, metadata={'tlv_type': 0x15}) + signature_info: Optional[_SigInfo] = field(default=None, metadata={'tlv_type': 0x16}) + signature_value: Optional[bytes] = field(default=None, metadata={ + 'tlv_type': 0x17, + 'field_type': 'sig_value', + 'cover_start': '_sig_cover_start', + }) + + +# ── Interest-like model (interest_name + digest) ────────────────────────────── + +@dataclass +class _InterestValue: + name: NDNName = field(default=None, metadata={ + 'tlv_type': 0x07, 'field_type': 'interest_name'}) + nonce: Optional[int] = field(default=None, metadata={ + 'tlv_type': 0x0a, 'fixed_len': 4}) + _sig_cover_start: None = field(default=None, metadata={'field_type': 'offset_marker'}) + application_parameters: Optional[bytes] = field(default=None, metadata={'tlv_type': 0x24}) + signature_info: Optional[_SigInfo] = field(default=None, metadata={'tlv_type': 0x2c}) + signature_value: Optional[bytes] = field(default=None, metadata={ + 'tlv_type': 0x2e, + 'field_type': 'sig_value', + 'cover_start': '_sig_cover_start', + 'digest_cover_start': '_sig_cover_start', + 'digest_cover_end': '_digest_cover_end', + }) + _digest_cover_end: None = field(default=None, metadata={'field_type': 'offset_marker'}) + + +class TestSignatureMachinery: + # ── offset_marker ────────────────────────────────────────────────────────── + + def test_offset_marker_produces_no_bytes(self): + @dataclass + class M: + _mark: None = field(default=None, metadata={'field_type': 'offset_marker'}) + v: int = field(default=None, metadata={'tlv_type': 0x01}) + + wire = tlv_encode(M(v=7)) + assert wire == bytes(tlv_encode(_Inner(val=7))) # only the uint TLV, no extra bytes + + def test_offset_marker_records_position_during_encode(self): + @dataclass + class M: + a: int = field(default=None, metadata={'tlv_type': 0x01}) + _mark: None = field(default=None, metadata={'field_type': 'offset_marker'}) + b: int = field(default=None, metadata={'tlv_type': 0x03}) + + markers = {} + tlv_encode(M(a=1, b=2), markers=markers) + # 'a' occupies 3 bytes (T=1, L=1, V=1), so _mark records offset 3. + assert markers['_mark'] == 3 + + def test_offset_marker_records_position_during_parse(self): + @dataclass + class M: + a: int = field(default=None, metadata={'tlv_type': 0x01}) + _mark: None = field(default=None, metadata={'field_type': 'offset_marker'}) + b: int = field(default=None, metadata={'tlv_type': 0x03}) + + wire = tlv_encode(M(a=1, b=2)) + markers = {} + tlv_parse(M, wire, markers=markers) + # offset_btl of 'b' = 3 (after 'a'), so _mark records 3. + assert markers.get('_mark') == 3 + + # ── sig_value / Data-like encoding ──────────────────────────────────────── + + def test_data_encode_produces_signature(self): + signer = _HmacSigner() + obj = _DataValue(name='/test', content=b'hello') + obj.signature_info = _SigInfo() + signer.write_signature_info(obj.signature_info) + + markers = {'##signer': signer} + wire = tlv_encode(obj, markers=markers) + + # Signature value TLV must be present. + assert 0x17 in bytes(wire) + # Wire must end with 32 sig bytes (preceded by TL 17 20). + assert wire[-34:-32] == b'\x17\x20' + + def test_data_encode_decode_roundtrip(self): + signer = _HmacSigner() + obj = _DataValue(name='/test/data', content=b'payload') + obj.signature_info = _SigInfo() + signer.write_signature_info(obj.signature_info) + + markers = {'##signer': signer} + wire = tlv_encode(obj, markers=markers) + + parse_markers = {} + p = tlv_parse(_DataValue, wire, markers=parse_markers) + assert Name.to_str(p.name) == '/test/data' + assert bytes(p.content) == b'payload' + assert p.signature_info.signature_type == 4 # HMAC_WITH_SHA256 + + def test_data_signature_verifies(self): + signer = _HmacSigner() + obj = _DataValue(name='/verify/me', content=b'data') + obj.signature_info = _SigInfo() + signer.write_signature_info(obj.signature_info) + + enc_markers = {'##signer': signer} + wire = tlv_encode(obj, markers=enc_markers) + enc_covered = enc_markers['##sig_covered_part'] + + parse_markers = {} + p = tlv_parse(_DataValue, wire, markers=parse_markers) + parse_covered = parse_markers.get('##sig_covered_part', []) + sig_buf = parse_markers['##sig_value_buf'] + + assert signer.verify(bytes(sig_buf), parse_covered) + + def test_data_signature_is_deterministic(self): + """Same object encoded twice with the same signer → identical wires.""" + signer = _HmacSigner() + obj1 = _DataValue(name='/det/test', content=b'hello') + obj1.signature_info = _SigInfo() + signer.write_signature_info(obj1.signature_info) + + obj2 = _DataValue(name='/det/test', content=b'hello') + obj2.signature_info = _SigInfo() + signer.write_signature_info(obj2.signature_info) + + w1 = tlv_encode(obj1, markers={'##signer': signer}) + w2 = tlv_encode(obj2, markers={'##signer': signer}) + assert bytes(w1) == bytes(w2) + + def test_ecdsa_signer_shrinks_wire(self): + signer = _EcdsaSigner() + obj = _DataValue(name='/shrink', content=b'x') + obj.signature_info = _SigInfo() + signer.write_signature_info(obj.signature_info) + + markers = {'##signer': signer} + wire = tlv_encode(obj, markers=markers) + + # Allocated 72 bytes, actual 71 → last byte trimmed. + assert markers['##shrink_len'] == 1 + # The sig_value TLV's L byte should now read 71 (0x47). + sig_tlv_idx = bytes(wire).index(0x17) # find sig_value type byte + assert wire[sig_tlv_idx + 1] == 71 + + def test_unsigned_data_produces_no_sig_tlv(self): + obj = _DataValue(name='/unsigned', content=b'ok') + wire = tlv_encode(obj) + assert b'\x17' not in bytes(wire) + + # ── interest_name + digest ──────────────────────────────────────────────── + + def test_interest_name_without_digest(self): + obj = _InterestValue(name='/plain/interest', nonce=42) + wire = tlv_encode(obj) + p = tlv_parse(_InterestValue, wire) + assert Name.to_str(p.name) == '/plain/interest' + assert p.nonce == 42 + + def test_interest_with_digest_appended(self): + """When ##need_digest is True and no digest component exists, one is appended.""" + app_param = b'\x01\x02\x03' + obj = _InterestValue(name='/digest/test', application_parameters=app_param) + obj.signature_info = _SigInfo() + signer = _HmacSigner() + signer.write_signature_info(obj.signature_info) + + markers = {'##signer': signer, '##need_digest': True} + wire = tlv_encode(obj, markers=markers) + + # Parse back and check digest component exists in name. + p = tlv_parse(_InterestValue, wire) + name_str = Name.to_str(p.name) + assert 'params-sha256=' in name_str + + def test_interest_digest_value_is_sha256(self): + """The ParametersSha256DigestComponent must equal SHA-256 of the digest-covered part.""" + from ndn.encoding.name import Component as C + app_param = b'\xde\xad\xbe\xef' + obj = _InterestValue(name='/verify/digest', application_parameters=app_param) + obj.signature_info = _SigInfo() + signer = _HmacSigner() + signer.write_signature_info(obj.signature_info) + + markers = {'##signer': signer, '##need_digest': True} + wire = tlv_encode(obj, markers=markers) + + # Locate the ParametersSha256DigestComponent in the encoded name. + p = tlv_parse(_InterestValue, wire) + digest_comp = None + for comp in p.name: + if C.get_type(comp) == C.TYPE_PARAMETERS_SHA256: + digest_comp = comp + break + assert digest_comp is not None + + digest_val = bytes(C.get_value(digest_comp)) + # Determine what the digest should cover: find where _sig_cover_start landed. + raw = bytes(wire) + sig_cover_start = markers.get('_sig_cover_start', 0) + d_end_field = '_digest_cover_end' + sig_cover_end = markers.get(d_end_field, len(raw)) + expected = sha256(raw[sig_cover_start:sig_cover_end]).digest() + assert digest_val == expected + + def test_interest_sig_covered_part_set_on_parse(self): + """After parsing an Interest, ##sig_covered_part is populated.""" + obj = _InterestValue(name='/parse/sig', application_parameters=b'\x00') + obj.signature_info = _SigInfo() + signer = _HmacSigner() + signer.write_signature_info(obj.signature_info) + + markers = {'##signer': signer, '##need_digest': True} + wire = tlv_encode(obj, markers=markers) + + parse_markers = {} + tlv_parse(_InterestValue, wire, markers=parse_markers) + assert '##sig_covered_part' in parse_markers + assert len(parse_markers['##sig_covered_part']) > 0 + + # ── tlv_get_arg / tlv_set_arg ───────────────────────────────────────────── + + def test_tlv_get_arg_missing_returns_default(self): + m = {} + assert tlv_get_arg(m, 'x', 42) == 42 + + def test_tlv_set_arg_stores_value(self): + m = {} + tlv_set_arg(m, 'key', 'value') + assert tlv_get_arg(m, 'key') == 'value'