diff --git a/src/ndn/app_support/security_v2_2.py b/src/ndn/app_support/security_v2_2.py new file mode 100644 index 0000000..4b585b5 --- /dev/null +++ b/src/ndn/app_support/security_v2_2.py @@ -0,0 +1,187 @@ +# ----------------------------------------------------------------------------- +# 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. +# ----------------------------------------------------------------------------- +import dataclasses as dc +from datetime import datetime, timedelta, UTC +from typing import Optional + +from ..utils import timestamp +from ..encoding import ( + Component, + FormalName, + Name, + VarBinaryStr, + parse_and_check_tl, +) +from ..encoding.tlv_model_v2 import tlv_encode, tlv_parse +from ..encoding.ndn_format_0_3_2 import ( + ContentType, + DataPacketValue, + KeyLocator, + MetaInfo, + SignatureInfo, + TypeNumber, +) + + +KEY_COMPONENT = Component.from_str('KEY') +SELF_COMPONENT = Component.from_str('self') +SIGN_REQ_COMPONENT = Component.from_str('cert-request') + + +class SecurityV2TypeNumber: + VALIDITY_PERIOD = 0xFD + NOT_BEFORE = 0xFE + NOT_AFTER = 0xFF + ADDITIONAL_DESCRIPTION = 0x0102 + DESCRIPTION_ENTRY = 0x0200 + DESCRIPTION_KEY = 0x0201 + DESCRIPTION_VALUE = 0x0202 + + SAFE_BAG = 0x80 + ENCRYPTED_KEY_BAG = 0x81 + + +@dc.dataclass +class DescriptionEntry: + description_key: Optional[bytes] = dc.field( + default=None, metadata={ + 'tlv_type': SecurityV2TypeNumber.DESCRIPTION_KEY}) + description_value: Optional[bytes] = dc.field( + default=None, metadata={ + 'tlv_type': SecurityV2TypeNumber.DESCRIPTION_VALUE}) + + +@dc.dataclass +class AdditionalDescription: + description_entry: list[DescriptionEntry] = dc.field( + default_factory=list, metadata={ + 'tlv_type': SecurityV2TypeNumber.DESCRIPTION_ENTRY}) + + +@dc.dataclass +class CertificateV2Extension: + additional_description: Optional[AdditionalDescription] = dc.field( + default=None, metadata={ + 'tlv_type': SecurityV2TypeNumber.ADDITIONAL_DESCRIPTION}) + + +@dc.dataclass +class ValidityPeriod: + not_before: Optional[bytes] = dc.field( + default=None, metadata={'tlv_type': SecurityV2TypeNumber.NOT_BEFORE}) + not_after: Optional[bytes] = dc.field( + default=None, metadata={'tlv_type': SecurityV2TypeNumber.NOT_AFTER}) + + +@dc.dataclass +class CertificateV2SignatureInfo(SignatureInfo): + validity_period: Optional[ValidityPeriod] = dc.field( + default=None, metadata={ + 'tlv_type': SecurityV2TypeNumber.VALIDITY_PERIOD}) + additional_description: Optional[AdditionalDescription] = dc.field( + default=None, metadata={ + 'tlv_type': SecurityV2TypeNumber.ADDITIONAL_DESCRIPTION}) + + +@dc.dataclass +class CertificateV2Value(DataPacketValue): + signature_info: Optional[CertificateV2SignatureInfo] = dc.field( + default=None, metadata={ + 'tlv_type': TypeNumber.SIGNATURE_INFO, + 'ignore_critical': True, + }) + + +@dc.dataclass +class SafeBag: + certificate_v2: Optional[bytes] = dc.field( + default=None, metadata={'tlv_type': TypeNumber.DATA}) + # We do not use ModelField due to 2 reasons: + # 1. The encoded length of CertificateV2 is unknown. + # 2. Generally we already have an encoded certificate when exporting a + # SafeBag. + encrypted_key_bag: Optional[bytes] = dc.field( + default=None, metadata={ + 'tlv_type': SecurityV2TypeNumber.ENCRYPTED_KEY_BAG}) + + +@dc.dataclass +class _CertificateEnvelope: + value: bytes = dc.field(metadata={'tlv_type': TypeNumber.DATA}) + + +def parse_certificate(wire) -> CertificateV2Value: + wire = parse_and_check_tl(wire, TypeNumber.DATA) + return tlv_parse(CertificateV2Value, wire) + + +def new_cert(key_name, issuer_id_component, pub_key, signer, + start_time, end_time) -> tuple[FormalName, VarBinaryStr]: + cert_name = Name.normalize(key_name) + [ + issuer_id_component, + Component.from_version(timestamp()), + ] + not_before = start_time.strftime('%Y%m%dT%H%M%S').encode() + not_after = end_time.strftime('%Y%m%dT%H%M%S').encode() + signature_info = CertificateV2SignatureInfo( + validity_period=ValidityPeriod( + not_before=not_before, + not_after=not_after, + ), + ) + signer.write_signature_info(signature_info) + if (signature_info.key_locator is not None + and not isinstance(signature_info.key_locator, KeyLocator)): + old_key_locator = signature_info.key_locator + signature_info.key_locator = KeyLocator( + name=old_key_locator.name, + key_digest=old_key_locator.key_digest, + ) + cert_val = CertificateV2Value( + name=cert_name, + content=pub_key, + meta_info=MetaInfo( + content_type=ContentType.KEY, + freshness_period=3600000, + ), + signature_info=signature_info, + ) + value = tlv_encode(cert_val, markers={'##signer': signer}) + return cert_name, tlv_encode(_CertificateEnvelope(value=value)) + + +def self_sign(key_name, pub_key, signer) -> tuple[FormalName, VarBinaryStr]: + end_time = datetime.now(UTC) + end_time = end_time.replace(year=end_time.year + 20) + return new_cert(key_name, SELF_COMPONENT, pub_key, signer, + datetime.fromisoformat('1970-01-01T00:00:00'), end_time) + + +def sign_req(key_name, pub_key, signer) -> tuple[FormalName, VarBinaryStr]: + start_time = datetime.now(UTC) + end_time = start_time + timedelta(days=10) + return new_cert(key_name, SIGN_REQ_COMPONENT, pub_key, signer, + datetime.now(UTC), end_time) + + +def derive_cert(key_name, issuer_id, pub_key, signer, + start_time, expire_sec) -> tuple[FormalName, VarBinaryStr]: + end_time = start_time + timedelta(seconds=expire_sec) + if isinstance(issuer_id, str): + issuer_id = Component.from_str(issuer_id) + return new_cert(key_name, issuer_id, pub_key, signer, start_time, end_time) 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/ndn_format_0_3_2.py b/src/ndn/encoding/ndn_format_0_3_2.py new file mode 100644 index 0000000..181fc4c --- /dev/null +++ b/src/ndn/encoding/ndn_format_0_3_2.py @@ -0,0 +1,329 @@ +# ----------------------------------------------------------------------------- +# Copyright (C) 2019-2020 The python-ndn authors +# Licensed under the Apache License, Version 2.0 (the "License"); +# ----------------------------------------------------------------------------- +"""NDN Packet Format v0.3 models using the dataclass TLV API.""" +import dataclasses as dc +from typing import Optional + +from .name import Name, Component +from .signer import Signer +from .tlv_model_v2 import NDNName, tlv_encode, tlv_parse +from .tlv_type import BinaryStr, VarBinaryStr, NonStrictName, FormalName +from .tlv_var import get_tl_num_size, parse_and_check_tl, write_tl_num + +__all__ = [ + 'TypeNumber', 'ContentType', 'SignatureType', 'KeyLocator', + 'SignatureInfo', + 'Links', 'MetaInfo', 'InterestParam', 'SignaturePtrs', 'make_interest', + 'make_data', 'parse_interest', 'parse_data', 'Interest', 'Data', +] + + +class TypeNumber: + INTEREST = 0x05 + DATA = 0x06 + NAME = Name.TYPE_NAME + GENERIC_NAME_COMPONENT = Component.TYPE_GENERIC + IMPLICIT_SHA256_DIGEST_COMPONENT = Component.TYPE_IMPLICIT_SHA256 + PARAMETERS_SHA256_DIGEST_COMPONENT = Component.TYPE_PARAMETERS_SHA256 + CAN_BE_PREFIX = 0x21 + MUST_BE_FRESH = 0x12 + FORWARDING_HINT = 0x1e + NONCE = 0x0a + INTEREST_LIFETIME = 0x0c + HOP_LIMIT = 0x22 + APPLICATION_PARAMETERS = 0x24 + INTEREST_SIGNATURE_INFO = 0x2c + INTEREST_SIGNATURE_VALUE = 0x2e + META_INFO = 0x14 + CONTENT = 0x15 + SIGNATURE_INFO = 0x16 + SIGNATURE_VALUE = 0x17 + CONTENT_TYPE = 0x18 + FRESHNESS_PERIOD = 0x19 + FINAL_BLOCK_ID = 0x1a + SIGNATURE_TYPE = 0x1b + KEY_LOCATOR = 0x1c + KEY_DIGEST = 0x1d + SIGNATURE_NONCE = 0x26 + SIGNATURE_TIME = 0x28 + SIGNATURE_SEQ_NUM = 0x2a + DELEGATION = 0x1f + PREFERENCE = 0x1e + + +class ContentType: + BLOB = 0 + LINK = 1 + KEY = 2 + NACK = 3 + + +class SignatureType: + NOT_SIGNED = None + DIGEST_SHA256 = 0 + SHA256_WITH_RSA = 1 + SHA256_WITH_ECDSA = 3 + HMAC_WITH_SHA256 = 4 + ED25519 = 5 + NULL = 200 + + +@dc.dataclass +class KeyLocator: + name: NDNName = dc.field( + default=None, metadata={'tlv_type': TypeNumber.NAME}) + key_digest: Optional[bytes] = dc.field( + default=None, metadata={'tlv_type': TypeNumber.KEY_DIGEST}) + + +@dc.dataclass +class SignatureInfo: + signature_type: Optional[int] = dc.field( + default=None, metadata={ + 'tlv_type': TypeNumber.SIGNATURE_TYPE, 'fixed_len': 1}) + key_locator: Optional[KeyLocator] = dc.field( + default=None, metadata={'tlv_type': TypeNumber.KEY_LOCATOR}) + signature_nonce: Optional[int] = dc.field( + default=None, metadata={'tlv_type': TypeNumber.SIGNATURE_NONCE}) + signature_time: Optional[int] = dc.field( + default=None, metadata={'tlv_type': TypeNumber.SIGNATURE_TIME}) + signature_seq_num: Optional[int] = dc.field( + default=None, metadata={'tlv_type': TypeNumber.SIGNATURE_SEQ_NUM}) + + +@dc.dataclass +class Links: + names: list[NDNName] = dc.field( + default_factory=list, metadata={'tlv_type': TypeNumber.NAME}) + + +@dc.dataclass +class InterestPacketValue: + name: NDNName = dc.field(default='/', metadata={ + 'tlv_type': TypeNumber.NAME, 'field_type': 'interest_name'}) + can_be_prefix: bool = dc.field( + default=False, metadata={'tlv_type': TypeNumber.CAN_BE_PREFIX}) + must_be_fresh: bool = dc.field( + default=False, metadata={'tlv_type': TypeNumber.MUST_BE_FRESH}) + forwarding_hint: Optional[Links] = dc.field( + default=None, metadata={'tlv_type': TypeNumber.FORWARDING_HINT}) + nonce: Optional[int] = dc.field(default=None, metadata={ + 'tlv_type': TypeNumber.NONCE, 'fixed_len': 4}) + lifetime: Optional[int] = dc.field( + default=None, metadata={'tlv_type': TypeNumber.INTEREST_LIFETIME}) + hop_limit: Optional[int] = dc.field(default=None, metadata={ + 'tlv_type': TypeNumber.HOP_LIMIT, 'fixed_len': 1}) + _sig_cover_start: None = dc.field( + default=None, metadata={'field_type': 'offset_marker'}) + _digest_cover_start: None = dc.field( + default=None, metadata={'field_type': 'offset_marker'}) + application_parameters: Optional[bytes] = dc.field( + default=None, metadata={'tlv_type': TypeNumber.APPLICATION_PARAMETERS}) + signature_info: Optional[SignatureInfo] = dc.field( + default=None, metadata={ + 'tlv_type': TypeNumber.INTEREST_SIGNATURE_INFO}) + signature_value: Optional[bytes] = dc.field(default=None, metadata={ + 'tlv_type': TypeNumber.INTEREST_SIGNATURE_VALUE, + 'field_type': 'sig_value', + 'cover_start': '_sig_cover_start', + 'digest_cover_start': '_digest_cover_start', + 'digest_cover_end': '_digest_cover_end', + }) + _digest_cover_end: None = dc.field( + default=None, metadata={'field_type': 'offset_marker'}) + + +@dc.dataclass +class InterestPacket: + interest: Optional[InterestPacketValue] = dc.field( + default=None, metadata={'tlv_type': TypeNumber.INTEREST}) + + +@dc.dataclass(init=False) +class MetaInfo: + content_type: Optional[int] = dc.field( + default=None, metadata={'tlv_type': TypeNumber.CONTENT_TYPE}) + freshness_period: Optional[int] = dc.field( + default=None, metadata={'tlv_type': TypeNumber.FRESHNESS_PERIOD}) + final_block_id: Optional[bytes] = dc.field( + default=None, metadata={'tlv_type': TypeNumber.FINAL_BLOCK_ID}) + + def __init__(self, + content_type: Optional[int] = ContentType.BLOB, + freshness_period: Optional[int] = None, + final_block_id: Optional[BinaryStr] = None): + self.content_type = content_type + self.freshness_period = freshness_period + self.final_block_id = final_block_id + + @staticmethod + def from_dict(kwargs): + return MetaInfo(**{ + f.name: kwargs[f.name] + for f in dc.fields(MetaInfo) + if f.name in kwargs + }) + + +@dc.dataclass +class DataPacketValue: + _sig_cover_start: None = dc.field( + default=None, metadata={'field_type': 'offset_marker'}) + name: NDNName = dc.field( + default='/', metadata={'tlv_type': TypeNumber.NAME}) + meta_info: Optional[MetaInfo] = dc.field( + default=None, metadata={'tlv_type': TypeNumber.META_INFO}) + content: Optional[bytes] = dc.field( + default=None, metadata={'tlv_type': TypeNumber.CONTENT}) + signature_info: Optional[SignatureInfo] = dc.field(default=None, metadata={ + 'tlv_type': TypeNumber.SIGNATURE_INFO, 'ignore_critical': True}) + signature_value: Optional[bytes] = dc.field(default=None, metadata={ + 'tlv_type': TypeNumber.SIGNATURE_VALUE, + 'field_type': 'sig_value', + 'cover_start': '_sig_cover_start', + }) + + +@dc.dataclass +class DataPacket: + data: Optional[DataPacketValue] = dc.field( + default=None, metadata={'tlv_type': TypeNumber.DATA}) + + +@dc.dataclass +class InterestParam: + can_be_prefix: bool = False + must_be_fresh: bool = False + nonce: Optional[int] = None + lifetime: Optional[int] = 4000 + hop_limit: Optional[int] = None + forwarding_hint: list[NonStrictName] = dc.field(default_factory=list) + + @staticmethod + def from_dict(kwargs): + return InterestParam(**{ + f.name: kwargs[f.name] + for f in dc.fields(InterestParam) + if f.name in kwargs + }) + + +@dc.dataclass +class SignaturePtrs: + signature_info: Optional[SignatureInfo] = None + signature_covered_part: list[BinaryStr] = dc.field(default_factory=list) + signature_value_buf: Optional[BinaryStr] = None + digest_covered_part: list[BinaryStr] = dc.field(default_factory=list) + digest_value_buf: Optional[BinaryStr] = None + + +Interest = tuple[FormalName, InterestParam, Optional[BinaryStr], SignaturePtrs] +Data = tuple[FormalName, MetaInfo, Optional[BinaryStr], SignaturePtrs] + + +def _wrap_tlv(type_num: int, value: BinaryStr) -> VarBinaryStr: + total = ( + get_tl_num_size(type_num) + + get_tl_num_size(len(value)) + + len(value) + ) + wire = bytearray(total) + offset = write_tl_num(type_num, wire, 0) + offset += write_tl_num(len(value), wire, offset) + wire[offset:] = value + return wire + + +def make_interest(name: NonStrictName, + interest_param: InterestParam, + app_param: Optional[BinaryStr] = None, + signer: Optional[Signer] = None, + need_final_name: bool = False): + value = InterestPacketValue( + name=name, + can_be_prefix=interest_param.can_be_prefix, + must_be_fresh=interest_param.must_be_fresh, + nonce=interest_param.nonce, + lifetime=interest_param.lifetime, + hop_limit=interest_param.hop_limit, + application_parameters=app_param, + ) + if interest_param.forwarding_hint: + value.forwarding_hint = Links( + names=list(interest_param.forwarding_hint)) + if signer is not None: + value.signature_info = SignatureInfo() + signer.write_signature_info(value.signature_info) + if value.application_parameters is None: + value.application_parameters = b'' + + markers = { + '##signer': signer, + '##need_digest': value.application_parameters is not None, + '##_digest_cover_start_field': '_digest_cover_start', + '##_digest_cover_end_field': '_digest_cover_end', + } + encoded_value = tlv_encode(value, markers=markers) + wire = _wrap_tlv(TypeNumber.INTEREST, encoded_value) + if need_final_name: + final_value = tlv_parse(InterestPacketValue, encoded_value) + return wire, final_value.name + return wire + + +def make_data(name: NonStrictName, + meta_info: MetaInfo, + content: Optional[BinaryStr] = None, + signer: Optional[Signer] = None) -> VarBinaryStr: + value = DataPacketValue(name=name, meta_info=meta_info, content=content) + if signer is not None: + value.signature_info = SignatureInfo() + signer.write_signature_info(value.signature_info) + encoded_value = tlv_encode(value, markers={'##signer': signer}) + return _wrap_tlv(TypeNumber.DATA, encoded_value) + + +def parse_interest(wire: BinaryStr, with_tl: bool = True) -> Interest: + value_wire = ( + parse_and_check_tl(wire, TypeNumber.INTEREST) + if with_tl else wire + ) + markers = {} + ret = tlv_parse(InterestPacketValue, value_wire, markers=markers) + params = InterestParam( + can_be_prefix=ret.can_be_prefix, + must_be_fresh=ret.must_be_fresh, + nonce=ret.nonce, + lifetime=ret.lifetime, + hop_limit=ret.hop_limit, + ) + if ret.forwarding_hint: + params.forwarding_hint.extend(ret.forwarding_hint.names) + + digest_parts = [] + digest_start = markers.get('_digest_cover_start') + if digest_start is not None: + digest_parts.append(memoryview(value_wire)[digest_start:]) + sig_ptrs = SignaturePtrs( + signature_info=ret.signature_info, + signature_covered_part=markers.get('##sig_covered_part', []), + signature_value_buf=ret.signature_value, + digest_covered_part=digest_parts, + digest_value_buf=markers.get('##digest_buf'), + ) + return ret.name, params, ret.application_parameters, sig_ptrs + + +def parse_data(wire: BinaryStr, with_tl: bool = True) -> Data: + value_wire = parse_and_check_tl(wire, TypeNumber.DATA) if with_tl else wire + markers = {} + ret = tlv_parse(DataPacketValue, value_wire, markers=markers) + meta_info = ret.meta_info if ret.meta_info is not None else MetaInfo() + sig_ptrs = SignaturePtrs( + signature_info=ret.signature_info, + signature_covered_part=markers.get('##sig_covered_part', []), + signature_value_buf=ret.signature_value, + ) + return ret.name, meta_info, ret.content, sig_ptrs diff --git a/src/ndn/encoding/ndnlp_v2_2.py b/src/ndn/encoding/ndnlp_v2_2.py new file mode 100644 index 0000000..9939c00 --- /dev/null +++ b/src/ndn/encoding/ndnlp_v2_2.py @@ -0,0 +1,133 @@ +# ----------------------------------------------------------------------------- +# Copyright (C) 2019-2020 The python-ndn authors +# Licensed under the Apache License, Version 2.0 (the "License"); +# ----------------------------------------------------------------------------- +"""NDNLPv2 models using the dataclass TLV API.""" +import dataclasses as dc +from typing import Optional + +from .tlv_model import DecodeError +from .tlv_model_v2 import tlv_encode, tlv_parse +from .tlv_type import BinaryStr, VarBinaryStr +from .tlv_var import parse_and_check_tl + +__all__ = [ + 'LpTypeNumber', 'NackReason', 'NetworkNack', 'CachePolicy', + 'LpPacketValue', 'LpPacket', 'parse_network_nack', 'make_network_nack', + 'parse_lp_packet', 'parse_lp_packet_v2', +] + + +class LpTypeNumber: + FRAGMENT = 0x50 + SEQUENCE = 0x51 + FRAG_INDEX = 0x52 + FRAG_COUNT = 0x53 + HOP_COUNT = 0x54 + PIT_TOKEN = 0x62 + LP_PACKET = 0x64 + NACK = 0x0320 + NACK_REASON = 0x0321 + INCOMING_FACE_ID = 0x032C + NEXT_HOP_FACE_ID = 0x0330 + CACHE_POLICY = 0x0334 + CACHE_POLICY_TYPE = 0x0335 + CONGESTION_MARK = 0x0340 + ACK = 0x0344 + TX_SEQUENCE = 0x0348 + NON_DISCOVERY = 0x034C + PREFIX_ANNOUNCEMENT = 0x0350 + + +class NackReason: + NONE = 0 + CONGESTION = 50 + DUPLICATE = 100 + NO_ROUTE = 150 + + +@dc.dataclass +class NetworkNack: + nack_reason: Optional[int] = dc.field( + default=None, metadata={'tlv_type': LpTypeNumber.NACK_REASON}) + + +@dc.dataclass +class CachePolicy: + cache_policy_type: Optional[int] = dc.field( + default=None, metadata={'tlv_type': LpTypeNumber.CACHE_POLICY_TYPE}) + + +@dc.dataclass +class LpPacketValue: + frag_index: Optional[int] = dc.field( + default=None, metadata={'tlv_type': LpTypeNumber.FRAG_INDEX}) + frag_count: Optional[int] = dc.field( + default=None, metadata={'tlv_type': LpTypeNumber.FRAG_COUNT}) + pit_token: Optional[bytes] = dc.field( + default=None, metadata={'tlv_type': LpTypeNumber.PIT_TOKEN}) + nack: Optional[NetworkNack] = dc.field( + default=None, metadata={ + 'tlv_type': LpTypeNumber.NACK, 'ignore_critical': False}) + incoming_face_id: Optional[int] = dc.field( + default=None, metadata={'tlv_type': LpTypeNumber.INCOMING_FACE_ID}) + next_hop_face_id: Optional[int] = dc.field( + default=None, metadata={'tlv_type': LpTypeNumber.NEXT_HOP_FACE_ID}) + cache_policy: Optional[CachePolicy] = dc.field( + default=None, metadata={ + 'tlv_type': LpTypeNumber.CACHE_POLICY, 'ignore_critical': False}) + congestion_mark: Optional[int] = dc.field( + default=None, metadata={'tlv_type': LpTypeNumber.CONGESTION_MARK}) + tx_sequence: Optional[bytes] = dc.field( + default=None, metadata={'tlv_type': LpTypeNumber.TX_SEQUENCE}) + ack: Optional[bytes] = dc.field( + default=None, metadata={'tlv_type': LpTypeNumber.ACK}) + non_discovery: bool = dc.field( + default=False, metadata={'tlv_type': LpTypeNumber.NON_DISCOVERY}) + prefix_announcement: Optional[bytes] = dc.field( + default=None, metadata={'tlv_type': LpTypeNumber.PREFIX_ANNOUNCEMENT}) + fragment: Optional[bytes] = dc.field( + default=None, metadata={'tlv_type': LpTypeNumber.FRAGMENT}) + + +@dc.dataclass +class LpPacket: + lp_packet: Optional[LpPacketValue] = dc.field( + default=None, metadata={'tlv_type': LpTypeNumber.LP_PACKET}) + + +def parse_lp_packet(wire: BinaryStr, + with_tl: bool = True + ) -> tuple[Optional[int], Optional[BinaryStr]]: + ret = parse_lp_packet_v2(wire, with_tl) + reason = ret.nack.nack_reason if ret.nack is not None else None + return reason, ret.fragment + + +def parse_lp_packet_v2(wire: BinaryStr, with_tl: bool = True) -> LpPacketValue: + if with_tl: + wire = parse_and_check_tl(wire, LpTypeNumber.LP_PACKET) + ret = tlv_parse(LpPacketValue, wire, ignore_critical=True) + if ret.frag_index is not None or ret.frag_count is not None: + raise DecodeError('NDNLP fragmentation is not implemented yet.') + return ret + + +def parse_network_nack( + wire: BinaryStr, + with_tl: bool = True) -> tuple[Optional[int], Optional[BinaryStr]]: + if with_tl: + wire = parse_and_check_tl(wire, LpTypeNumber.LP_PACKET) + ret = tlv_parse(LpPacketValue, wire, ignore_critical=True) + if ret.nack is not None: + return ret.nack.nack_reason, ret.fragment + return None, None + + +def make_network_nack(encoded_interest: BinaryStr, + nack_reason: int) -> VarBinaryStr: + value = LpPacketValue( + nack=NetworkNack(nack_reason=nack_reason), + fragment=encoded_interest, + ) + return tlv_encode(LpPacket(lp_packet=value)) diff --git a/src/ndn/encoding/tlv_model_v2.py b/src/ndn/encoding/tlv_model_v2.py index e69de29..155fd20 100644 --- a/src/ndn/encoding/tlv_model_v2.py +++ b/src/ndn/encoding/tlv_model_v2.py @@ -0,0 +1,981 @@ +# ----------------------------------------------------------------------------- +# 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 types import UnionType + +from .tlv_type import 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) in (typing.Union, UnionType): + 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 + for i, comp in enumerate(name): + if Component.get_type(comp) == Component.TYPE_PARAMETERS_SHA256: + if len(Component.get_value(comp)) != 32: + raise ValueError( + f'{fname}: ParametersSha256DigestComponent must be 32 bytes') + if need_digest: + 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: memoryview, 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/ndn_format_0_3_2_test.py b/tests/encoding/ndn_format_0_3_2_test.py new file mode 100644 index 0000000..87a186b --- /dev/null +++ b/tests/encoding/ndn_format_0_3_2_test.py @@ -0,0 +1,73 @@ +import hashlib + +from ndn.encoding import Name +from ndn.encoding.ndn_format_0_3_2 import ( + ContentType, + InterestParam, + MetaInfo, + SignatureType, + make_data, + make_interest, + parse_data, + parse_interest, +) +from ndn.security import DigestSha256Signer + + +def test_default_interest_wire_format(): + wire = make_interest('/local/ndn/prefix', InterestParam()) + assert wire == ( + b'\x05\x1a\x07\x14\x08\x05local\x08\x03ndn\x08\x06prefix' + b'\x0c\x02\x0f\xa0' + ) + + name, params, app_params, sig = parse_interest(wire) + assert name == Name.from_str('/local/ndn/prefix') + assert params.lifetime == 4000 + assert app_params is None + assert sig.signature_info is None + + +def test_signed_interest_wire_format_and_coverage(): + wire = make_interest( + '/local/ndn/prefix', + InterestParam(nonce=0x6c211166), + b'\x01\x02\x03\x04', + DigestSha256Signer(), + ) + assert wire == ( + b'\x05\x6f\x07\x36\x08\x05local\x08\x03ndn\x08\x06prefix' + b'\x02 \x8e\x6e\x36\xd7\xea\xbc\xde\x43\x75\x61\x40\xc9' + b'\x0b\xda\x09\xd5' + b'\x00\xd2\xa5\x77\xf2\xf5\x33\xb5\x69\xf0\x44\x1d\xf0\xa7\xf9\xe2' + b'\x0a\x04\x6c\x21\x11\x66\x0c\x02\x0f\xa0' + b'\x24\x04\x01\x02\x03\x04\x2c\x03\x1b\x01\x00' + b'\x2e \xea\xa8\xf0\x99\x08\x63\x78\x95\x1d\xe0\x5f\xf1' + b'\xde\xbb\xc1\x18' + b'\xb5\x21\x8b\x2f\xca\xa0\xb5\x1d\x18\xfa\xbc\x29\xf5\x4d\x58\xff' + ) + + _, _, _, sig = parse_interest(wire) + assert sig.signature_info.signature_type == SignatureType.DIGEST_SHA256 + signature = hashlib.sha256(b''.join(sig.signature_covered_part)).digest() + digest = hashlib.sha256(b''.join(sig.digest_covered_part)).digest() + assert signature == sig.signature_value_buf + assert digest == sig.digest_value_buf + + +def test_data_wire_format_and_coverage(): + wire = make_data( + '/local/ndn/prefix', MetaInfo(), signer=DigestSha256Signer()) + assert wire == ( + b"\x06\x42\x07\x14\x08\x05local\x08\x03ndn\x08\x06prefix" + b"\x14\x03\x18\x01\x00\x16\x03\x1b\x01\x00" + b"\x17 \x7f1\xe4\t\xc5z/\x1d\r\xdaVh8\xfd\xd9\x94" + b"\xd8\'S\x13[\xd7\x15\xa5\x9d%^\x80\xf2\xab\xf0\xb5" + ) + + name, meta_info, content, sig = parse_data(wire) + assert name == Name.from_str('/local/ndn/prefix') + assert meta_info.content_type == ContentType.BLOB + assert content is None + signature = hashlib.sha256(b''.join(sig.signature_covered_part)).digest() + assert signature == sig.signature_value_buf diff --git a/tests/encoding/ndnlp_v2_2_test.py b/tests/encoding/ndnlp_v2_2_test.py new file mode 100644 index 0000000..8c6c188 --- /dev/null +++ b/tests/encoding/ndnlp_v2_2_test.py @@ -0,0 +1,58 @@ +from ndn.encoding.ndnlp_v2_2 import ( + LpPacketValue, + LpTypeNumber, + NackReason, + NetworkNack, + make_network_nack, + parse_network_nack, + parse_lp_packet_v2, +) +from ndn.encoding.ndn_format_0_3_2 import ( + InterestParam, + make_interest, + parse_interest, +) +from ndn.encoding import DecodeError, Name, tlv_encode, write_tl_num +import pytest + + +def test_network_nack_wire_format(): + interest = make_interest( + '/localhost/nfd/faces/events', + InterestParam(must_be_fresh=True, can_be_prefix=True), + ) + lp_packet = make_network_nack(interest, NackReason.NO_ROUTE) + + assert lp_packet == ( + b"\x64\x36\xfd\x03\x20\x05\xfd\x03\x21\x01\x96" + b"\x50\x2b\x05\x29\x07\x1f\x08\tlocalhost\x08\x03nfd" + b"\x08\x05faces\x08\x06events\x21\x00\x12\x00\x0c\x02\x0f\xa0" + ) + + reason, encoded_interest = parse_network_nack(lp_packet) + name, params, _, _ = parse_interest(encoded_interest) + assert reason == NackReason.NO_ROUTE + assert name == Name.from_str('/localhost/nfd/faces/events') + assert params.can_be_prefix + assert params.must_be_fresh + + +def test_network_nack_parser_accepts_fragment_metadata(): + value = tlv_encode(LpPacketValue( + frag_index=0, + frag_count=1, + nack=NetworkNack(nack_reason=NackReason.NO_ROUTE), + fragment=b'\x05\x00', + )) + wire = bytearray(2 + len(value)) + offset = write_tl_num(LpTypeNumber.LP_PACKET, wire, 0) + offset += write_tl_num(len(value), wire, offset) + wire[offset:] = value + + assert parse_network_nack(wire) == (NackReason.NO_ROUTE, b'\x05\x00') + + +def test_nested_unknown_critical_field_is_rejected(): + wire = b'\x64\x06\xfd\x03\x20\x02\x01\x00' + with pytest.raises(DecodeError): + parse_lp_packet_v2(wire) diff --git a/tests/encoding/tlv_model_v2_test.py b/tests/encoding/tlv_model_v2_test.py new file mode 100644 index 0000000..d58ac88 --- /dev/null +++ b/tests/encoding/tlv_model_v2_test.py @@ -0,0 +1,1235 @@ +# ----------------------------------------------------------------------------- +# 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 + +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, 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: int | None = 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: bool | None = 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: bytes | None = field(default=None, metadata={'tlv_type': 0x15}) + signature_info: _SigInfo | None = field(default=None, metadata={'tlv_type': 0x16}) + signature_value: bytes | None = 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: int | None = field(default=None, metadata={ + 'tlv_type': 0x0a, 'fixed_len': 4}) + _sig_cover_start: None = field(default=None, metadata={'field_type': 'offset_marker'}) + application_parameters: bytes | None = field(default=None, metadata={'tlv_type': 0x24}) + signature_info: _SigInfo | None = field(default=None, metadata={'tlv_type': 0x2c}) + signature_value: bytes | None = 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' diff --git a/tests/misc/security_v2_2_test.py b/tests/misc/security_v2_2_test.py new file mode 100644 index 0000000..bc22220 --- /dev/null +++ b/tests/misc/security_v2_2_test.py @@ -0,0 +1,83 @@ +import dataclasses as dc +import hashlib +from datetime import UTC, datetime + +from ndn.app_support.security_v2_2 import ( + CertificateV2SignatureInfo, + CertificateV2Value, + ContentType, + SafeBag, + SecurityV2TypeNumber, + new_cert, + parse_certificate, +) +from ndn.encoding import ( + Component, + Name, + SignatureType, + tlv_encode, + tlv_parse, +) +from ndn.encoding.ndn_format_0_3_2 import parse_data +from ndn.security import DigestSha256Signer, HmacSha256Signer + + +def test_certificate_models_use_dataclass_tlv_format(): + assert dc.is_dataclass(CertificateV2SignatureInfo) + assert dc.is_dataclass(CertificateV2Value) + assert dc.is_dataclass(SafeBag) + + +def test_new_cert_round_trip(): + start = datetime(2025, 1, 2, 3, 4, 5, tzinfo=UTC) + end = datetime(2026, 2, 3, 4, 5, 6, tzinfo=UTC) + cert_name, wire = new_cert( + '/test/KEY/key-id', + Component.from_str('issuer'), + b'public-key', + DigestSha256Signer(), + start, + end, + ) + + cert = parse_certificate(wire) + assert cert.name == cert_name + assert cert.content == b'public-key' + assert cert.meta_info.content_type == ContentType.KEY + assert cert.meta_info.freshness_period == 3600000 + assert cert.signature_info.signature_type == SignatureType.DIGEST_SHA256 + assert cert.signature_info.validity_period.not_before == b'20250102T030405' + assert cert.signature_info.validity_period.not_after == b'20260203T040506' + assert Name.is_prefix(Name.from_str('/test/KEY/key-id'), cert_name) + + _, _, _, sig = parse_data(wire) + covered = b''.join(sig.signature_covered_part) + assert hashlib.sha256(covered).digest() == sig.signature_value_buf + + +def test_new_cert_converts_legacy_signer_key_locator(): + _, wire = new_cert( + '/test/KEY/key-id', + Component.from_str('issuer'), + b'public-key', + HmacSha256Signer('/signer/key', b'secret'), + datetime(2025, 1, 1, tzinfo=UTC), + datetime(2026, 1, 1, tzinfo=UTC), + ) + + cert = parse_certificate(wire) + assert cert.signature_info.key_locator.name == Name.from_str('/signer/key') + + +def test_safe_bag_round_trip(): + safe_bag = SafeBag(certificate_v2=b'\x06\x00', encrypted_key_bag=b'key') + wire = tlv_encode(safe_bag) + + assert wire == ( + bytes([0x06, 0x02, 0x06, 0x00]) + + bytes([SecurityV2TypeNumber.ENCRYPTED_KEY_BAG, 0x03]) + + b'key' + ) + parsed = tlv_parse(SafeBag, wire) + assert parsed.certificate_v2 == b'\x06\x00' + assert parsed.encrypted_key_bag == b'key'