From be17572a8e3aa47c2a9529403a4a2de249cd270e Mon Sep 17 00:00:00 2001 From: Brian Sipos Date: Tue, 4 Aug 2026 22:46:30 -0400 Subject: [PATCH 1/6] Checkpoint of new BPv7 with working UTS and lots of patched cbor lib --- scapy/cbor/__init__.py | 6 + scapy/cbor/cborcodec.py | 10 +- scapy/cbor/cborfields.py | 253 ++++++++++++++++- scapy/cborpacket.py | 5 + scapy/contrib/bpv7.py | 590 +++++++++++++++++++++++++++++++++++++++ test/contrib/bpv7.uts | 176 ++++++++++++ 6 files changed, 1023 insertions(+), 17 deletions(-) create mode 100644 scapy/contrib/bpv7.py create mode 100644 test/contrib/bpv7.uts diff --git a/scapy/cbor/__init__.py b/scapy/cbor/__init__.py index dcec5d8ed5d..05005fc256a 100644 --- a/scapy/cbor/__init__.py +++ b/scapy/cbor/__init__.py @@ -56,12 +56,15 @@ CBORF_NULL, CBORF_UNDEFINED, CBORF_FLOAT, + CBORF_SEQUENCE, + CBORF_SEQUENCE_OF, CBORF_ARRAY, CBORF_ARRAY_OF, CBORF_MAP, CBORF_SEMANTIC_TAG, CBORF_optional, CBORF_PACKET, + CBORF_BYTE_STRING_PACKET, ) __all__ = [ @@ -115,6 +118,8 @@ "CBORF_UNDEFINED", "CBORF_FLOAT", # Structured fields + "CBORF_SEQUENCE", + "CBORF_SEQUENCE_OF", "CBORF_ARRAY", "CBORF_ARRAY_OF", "CBORF_MAP", @@ -122,4 +127,5 @@ # Complex fields "CBORF_optional", "CBORF_PACKET", + "CBORF_BYTE_STRING_PACKET", ] diff --git a/scapy/cbor/cborcodec.py b/scapy/cbor/cborcodec.py index b49b9d38b30..c02eb764fd0 100644 --- a/scapy/cbor/cborcodec.py +++ b/scapy/cbor/cborcodec.py @@ -69,13 +69,15 @@ def __init__(self, def CBOR_encode_head(major_type, value): - # type: (int, int) -> bytes + # type: (int, Optional[int]) -> bytes """ Encode CBOR initial byte and additional info. Format: 3 bits major type + 5 bits additional info """ - if value < 24: + if value is None or value < 24: # Value fits in 5 bits + if value is None: + value = 0x1f return chb((major_type << 5) | value) elif value < 256: # 1-byte value follows @@ -92,7 +94,7 @@ def CBOR_encode_head(major_type, value): def CBOR_decode_head(s): - # type: (bytes) -> Tuple[int, int, bytes] + # type: (bytes) -> Tuple[int, Optional[int], bytes] """ Decode CBOR initial byte and additional info. Returns: (major_type, value, remaining_bytes) @@ -134,6 +136,8 @@ def CBOR_decode_head(s): "Not enough bytes for 8-byte value", remaining=s) value = struct.unpack(">Q", s[1:9])[0] return major_type, value, s[9:] + elif additional_info == 31: + return major_type, None, s[1:] else: raise CBOR_Codec_Decoding_Error( "Invalid additional info: %d" % additional_info, remaining=s) diff --git a/scapy/cbor/cborfields.py b/scapy/cbor/cborfields.py index 536424728ec..d01956a1c3b 100644 --- a/scapy/cbor/cborfields.py +++ b/scapy/cbor/cborfields.py @@ -124,7 +124,7 @@ def i2h(self, pkt, x): def m2i(self, pkt, s): # type: (CBOR_Packet, bytes) -> Tuple[_A, bytes] - raise NotImplementedError("Subclasses must implement m2i") + raise NotImplementedError(f"Subclasses must implement m2i for {type(self)}") def i2m(self, pkt, x): # type: (CBOR_Packet, Union[bytes, _I, _A]) -> bytes @@ -137,11 +137,11 @@ def i2m(self, pkt, x): def _encode(self, x): # type: (Any) -> bytes """Encode a raw Python value to CBOR bytes.""" - raise NotImplementedError("Subclasses must implement _encode") + raise NotImplementedError(f"Subclasses must implement _encode for {type(self)}") def any2i(self, pkt, x): # type: (CBOR_Packet, Any) -> _I - return cast(_I, x) + return self._wrap(x) def extract_packet(self, cls, # type: Type[CBOR_Packet] @@ -150,7 +150,7 @@ def extract_packet(self, ): # type: (...) -> Tuple[CBOR_Packet, bytes] try: - c = cls(s, _underlayer=_underlayer) + c = cls(s, stop_dissection_after=cls, _underlayer=_underlayer) except CBORF_badsequence: c = packet.Raw(s, _underlayer=_underlayer) # type: ignore cpad = c.getlayer(packet.Raw) @@ -163,7 +163,7 @@ def extract_packet(self, def build(self, pkt): # type: (CBOR_Packet) -> bytes - return self.i2m(pkt, getattr(pkt, self.name)) + return self.i2m(pkt, pkt.getfieldval(self.name)) def dissect(self, pkt, s): # type: (CBOR_Packet, bytes) -> bytes @@ -185,11 +185,11 @@ def do_copy(self, x): def set_val(self, pkt, val): # type: (CBOR_Packet, Any) -> None - setattr(pkt, self.name, val) + pkt.setfieldval(self.name, val) def is_empty(self, pkt): # type: (CBOR_Packet) -> bool - return getattr(pkt, self.name) is None + return pkt.getfieldval(self.name) is None def get_fields_list(self): # type: () -> List[CBORF_field[Any, Any]] @@ -218,6 +218,8 @@ class CBORF_UNSIGNED_INTEGER(CBORF_field[int, CBOR_UNSIGNED_INTEGER]): def _wrap(self, val): # type: (Any) -> CBOR_UNSIGNED_INTEGER + if val is None: + return None if isinstance(val, CBOR_UNSIGNED_INTEGER): return val return CBOR_UNSIGNED_INTEGER(int(val)) @@ -243,6 +245,8 @@ class CBORF_NEGATIVE_INTEGER(CBORF_field[int, CBOR_NEGATIVE_INTEGER]): def _wrap(self, val): # type: (Any) -> CBOR_NEGATIVE_INTEGER + if val is None: + return None if isinstance(val, CBOR_NEGATIVE_INTEGER): return val return CBOR_NEGATIVE_INTEGER(int(val)) @@ -269,6 +273,8 @@ class CBORF_INTEGER(CBORF_field[int, def _wrap(self, val): # type: (Any) -> Union[CBOR_UNSIGNED_INTEGER, CBOR_NEGATIVE_INTEGER] + if val is None: + return None if isinstance(val, (CBOR_UNSIGNED_INTEGER, CBOR_NEGATIVE_INTEGER)): return val i = int(val) @@ -310,6 +316,8 @@ class CBORF_BYTE_STRING(CBORF_field[bytes, CBOR_BYTE_STRING]): def _wrap(self, val): # type: (Any) -> CBOR_BYTE_STRING + if val is None: + return None if isinstance(val, CBOR_BYTE_STRING): return val return CBOR_BYTE_STRING(bytes(val)) @@ -329,6 +337,57 @@ def randval(self): return RandString(RandNum(0, 1000)) +class CBORF_BYTE_STRING_PACKET(CBORF_field['Packet', CBOR_BYTE_STRING]): + """CBOR byte string which wraps another packet field. + The inner packet may or may not itself be CBOR or CBOR sequence data. + """ + CBOR_tag = CBOR_MajorTypes.BYTE_STRING + + def __init__(self, + name, # type: str + default, # type: Optional[BasePacket] + pkt_cls=None, # type: Optional[Type[Packet]] + cls_cb=None, # type: Optional[Callable[[Packet, bytes], Optional[Type[Packet]]]] + ): + # type: (...) -> None + if pkt_cls is None and cls_cb is None: + raise ValueError('Must give one of pkt_cls or cls_cb') + super(CBORF_BYTE_STRING_PACKET, self).__init__(name, default) + self.pkt_cls = pkt_cls + self.cls_cb = cls_cb + + def _wrap(self, val): + # type: (Any) -> Any + return val + + def m2i(self, pkt, s): + # type: (CBOR_Packet, bytes) -> Tuple[Packet, bytes] + obj, remain = CBORcodec_BYTE_STRING.dec(s) # type: ignore + if not isinstance(obj, CBOR_BYTE_STRING): + raise CBOR_Decoding_Error( + "Expected bstr, got %r" % obj) + + if self.pkt_cls is not None: + pkt_cls = self.pkt_cls + elif self.cls_cb is not None: + pkt_cls = self.cls_cb(pkt, obj.val) + if pkt_cls is None: + pkt_cls = packet.Raw + + try: + sub = pkt_cls(obj.val, _underlayer=pkt) + except Exception: + sub = packet.Raw(s, _underlayer=pkt) # type: ignore + + return sub, remain + + def _encode(self, x): + # type: (Any) -> bytes + return CBORcodec_BYTE_STRING.enc( + x if isinstance(x, CBOR_Object) else CBOR_BYTE_STRING(bytes(x)) + ) + + class CBORF_TEXT_STRING(CBORF_field[str, CBOR_TEXT_STRING]): """CBOR text string field (major type 3).""" CBOR_tag = CBOR_MajorTypes.TEXT_STRING @@ -486,6 +545,73 @@ def randval(self): # Structured CBOR Fields # ############################## +class CBORF_SEQUENCE(CBORF_field[List[Any], List[Any]]): + """ + Unframed fixed sequence of named, typed fields. + Analogous to ASN1F_SEQUENCE: each positional element corresponds to a + specific CBORF_field. + + Example:: + + class MyCBOR(CBOR_Packet): + CBOR_root = CBORF_SEQUENCE( + CBORF_INTEGER("version", 1), + CBORF_TEXT_STRING("name", ""), + ) + """ + CBOR_tag = None + holds_packets = 1 + + def __init__(self, *seq, **kwargs): + # type: (*Any, **Any) -> None + # The array itself is a structural field without its own named slot on + # the packet; a placeholder name is used so the base class __init__ + # stays happy. Individual element fields are the ones that carry names. + name = "_cbor_sequence" + default = [field.default for field in seq] + super(CBORF_SEQUENCE, self).__init__(name, None) + self.default = default + self.seq = seq + self.islist = len(seq) > 1 + + def __repr__(self): + # type: () -> str + return "<%s%r>" % (self.__class__.__name__, self.seq) + + def is_empty(self, pkt): + # type: (CBOR_Packet) -> bool + return all(f.is_empty(pkt) for f in self.seq) + + def get_fields_list(self): + # type: () -> List[CBORF_field[Any, Any]] + return reduce(lambda x, y: x + y.get_fields_list(), + self.seq, []) + + def m2i(self, pkt, s): + # type: (Any, bytes) -> Tuple[Any, bytes] + """ + Decode a fixed-count CBOR sequence. + Each element is decoded by its corresponding + field in ``self.seq``. The decoded values are set directly on the + packet by each field's ``dissect`` call, so this method returns an + empty list (which is discarded by ``dissect``). + """ + for obj in self.seq: + try: + s = obj.dissect(pkt, s) + except CBORF_badsequence: + break + return [], s + + def dissect(self, pkt, s): + # type: (Any, bytes) -> bytes + _, x = self.m2i(pkt, s) + return x + + def build(self, pkt): + # type: (CBOR_Packet) -> bytes + return b"".join(obj.build(pkt) for obj in self.seq) + class CBORF_ARRAY(CBORF_field[List[Any], List[Any]]): """ CBOR array with a fixed sequence of named, typed fields (major type 4). @@ -529,6 +655,17 @@ def get_fields_list(self): return reduce(lambda x, y: x + y.get_fields_list(), self.seq, []) + def _consume_break(self, s, need): + # need an indefinite break, peek + major_type, arg, remain = CBOR_decode_head(s) + if major_type == CBOR_MajorTypes.SIMPLE_AND_FLOAT and arg is None: + return True, remain + + if need: + raise CBOR_Decoding_Error("Needed indefinite break and did not see one") + else: + return False, s + def m2i(self, pkt, s): # type: (Any, bytes) -> Tuple[Any, bytes] """ @@ -544,15 +681,17 @@ def m2i(self, pkt, s): if major_type != 4: raise CBOR_Decoding_Error( "Expected major type 4 (array), got %d" % major_type) - if count != len(self.seq): - raise CBOR_Decoding_Error( - "Array length mismatch: expected %d, got %d" % - (len(self.seq), count)) for obj in self.seq: + if count is None: + got, s = self._consume_break(s, False) + if got: + break try: s = obj.dissect(pkt, s) except CBORF_badsequence: break + if count is None: + _, s = self._consume_break(s, True) return [], s def dissect(self, pkt, s): @@ -562,8 +701,11 @@ def dissect(self, pkt, s): def build(self, pkt): # type: (CBOR_Packet) -> bytes - items = b"".join(obj.build(pkt) for obj in self.seq) - return CBOR_encode_head(4, len(self.seq)) + items + parts = (obj.build(pkt) for obj in self.seq) + parts = tuple(filter(lambda s: bool(s), parts)) + # ignore conditional fields which produce no data + items = b"".join(parts) + return CBOR_encode_head(4, len(parts)) + items _ARRAY_T = Union[ @@ -574,6 +716,89 @@ def build(self, pkt): ] +class CBORF_SEQUENCE_OF(CBORF_field[List[_ARRAY_T], List[CBOR_Object[Any]]]): + """ + CBOR sequence of homogeneous elements (no enveloping head). + Analogous to ASN1F_SEQUENCE_OF: variable-length array where every + element shares the same type, specified by ``cls``. + + ``cls`` may be a :class:`CBORF_field` class/instance (leaf type) or a + :class:`CBOR_Packet` subclass (structured type). + """ + CBOR_tag = None + islist = 1 + + def __init__(self, + name, # type: str + default, # type: Any + cls=None, # type: _ARRAY_T + cls_cb=None, # type: Optional[Callable[[Packet, bytes], Optional[Type[Packet]]]] + ): + # type: (...) -> None + if isinstance(cls, type) and issubclass(cls, CBORF_field) or \ + isinstance(cls, CBORF_field): + if isinstance(cls, type): + self.fld = cls("_item", None) # type: ignore + else: + self.fld = cls + self._extract_item = lambda s, pkt: self.fld.m2i(pkt, s) + self.holds_packets = 0 + elif hasattr(cls, "CBOR_root") or callable(cls): + self.cls = cast("Type[CBOR_Packet]", cls) + self._extract_item = lambda s, pkt: self.extract_packet( + self.cls, s, _underlayer=pkt) + self.holds_packets = 1 + elif cls_cb is not None: + def extract(s, pkt): + pkt_cls = cls_cb(pkt, s) + if pkt_cls is not None: + return self.extract_packet(pkt_cls, s, _underlayer=pkt) + else: + return None, s + self._extract_item = extract + self.holds_packets = 1 + else: + raise ValueError("cls must be a CBORF_field or CBOR_Packet") + super(CBORF_SEQUENCE_OF, self).__init__(name, None) + self.default = default + + def is_empty(self, pkt): + # type: (CBOR_Packet) -> bool + return CBORF_field.is_empty(self, pkt) + + def m2i(self, pkt, s): + # type: (CBOR_Packet, bytes) -> Tuple[List[Any], bytes] + lst = [] + while s: + c, s = self._extract_item(s, pkt) # type: ignore + if c is not None: + lst.append(c) + else: + break + return lst, s + + def build(self, pkt): + # type: (CBOR_Packet) -> bytes + val = pkt.getfieldval(self.name) + if val is None: + val = [] + return b"".join(bytes(item) for item in val) + + def i2repr(self, pkt, x): + # type: (CBOR_Packet, Any) -> str + if self.holds_packets: + return repr(x) + elif x is None: + return "()" + else: + return "(%s)" % ", ".join( + self.fld.i2repr(pkt, item) for item in x # type: ignore + ) + + def __repr__(self): + # type: () -> str + return "<%s %s>" % (self.__class__.__name__, self.name) + class CBORF_ARRAY_OF(CBORF_field[List[_ARRAY_T], List[CBOR_Object[Any]]]): """ CBOR array of homogeneous elements (major type 4). @@ -632,7 +857,7 @@ def m2i(self, pkt, s): def build(self, pkt): # type: (CBOR_Packet) -> bytes - val = getattr(pkt, self.name) + val = pkt.getfieldval(self.name) if val is None: val = [] items = b"".join(bytes(item) for item in val) diff --git a/scapy/cborpacket.py b/scapy/cborpacket.py index eb12bedaea9..a05eef6fc9d 100644 --- a/scapy/cborpacket.py +++ b/scapy/cborpacket.py @@ -43,6 +43,11 @@ def __new__(cls, class CBOR_Packet(Packet, metaclass=CBORPacket_metaclass): CBOR_root = cast('CBORF_field[Any, Any]', None) + def setfieldval(self, attr, val): + fld = cast('CBORF_field', self.get_field(attr)) + val = fld._wrap(val) + super().setfieldval(attr, val) + def self_build(self): # type: () -> bytes """Build this CBOR packet to wire bytes using CBOR_root. diff --git a/scapy/contrib/bpv7.py b/scapy/contrib/bpv7.py new file mode 100644 index 00000000000..00be89f1e73 --- /dev/null +++ b/scapy/contrib/bpv7.py @@ -0,0 +1,590 @@ +# SPDX-License-Identifier: GPL-2.0-only +# This file is part of Scapy +# See https://scapy.net/ for more information +# Copyright (C) Brian Sipos + +# scapy.contrib.description = Bundle Protocol Version 7 (BPv7) +# scapy.contrib.status = loads + +import crcmod +from dataclasses import dataclass +import datetime +import enum +import logging +import struct +from typing import Any, List, Optional, Tuple, Union, cast +from scapy import volatile +from scapy.config import conf +from scapy.packet import Packet, bind_layers +from scapy.fields import ConditionalField +from scapy.cbor.cborcodec import CBOR_encode_head, CBOR_decode_head, CBOR_MajorTypes +from scapy.cbor import ( + CBORF_field, + CBORF_UNSIGNED_INTEGER, CBORF_INTEGER, CBORF_ARRAY, CBORF_BYTE_STRING, + CBORF_SEQUENCE, CBORF_SEQUENCE_OF, CBORF_PACKET, CBORF_BYTE_STRING_PACKET, + CBORcodec_ARRAY, CBOR_UNSIGNED_INTEGER, CBOR_TEXT_STRING, CBOR_ARRAY, + CBOR_Object +) +from scapy.cborpacket import ( + CBOR_Packet, +) + +LOG_RUNTIME = logging.getLogger("scapy.runtime") + + +class CBORF_INDEFINITE_ARRAY(CBORF_ARRAY): + """A field to act as an array but to always encode to indefinte-length.""" + + def build(self, pkt): + # type: (CBOR_Packet) -> bytes + items = b"".join(obj.build(pkt) for obj in self.seq) + return ( + CBOR_encode_head(int(CBOR_MajorTypes.ARRAY), None) + items + + CBOR_encode_head(int(CBOR_MajorTypes.SIMPLE_AND_FLOAT), None) + ) + + +class DtnTimeField(CBORF_INTEGER): + ''' A DTN time value representing number of milliseconds from the + DTN epoch 2000-01-01T00:00:00Z. + + This value is automatically converted from a + :py:cls:`datetime.datetime` object and human friendly text in ISO8601 + format. + The special human value "zero" represents the zero value time. + ''' + + # Epoch reference for DTN Time + DTN_EPOCH = datetime.datetime(2000, 1, 1, 0, 0, 0, 0, datetime.timezone.utc) + + @staticmethod + def datetime_to_dtntime(val: 'Optional[datetime.datetime]') -> int: + if val is None: + return 0 + delta = val - DtnTimeField.DTN_EPOCH + return int(delta / datetime.timedelta(milliseconds=1)) + + @staticmethod + def dtntime_to_datetime(val): + if val == 0 or val is None: + return None + delta = datetime.timedelta(milliseconds=val) + return delta + DtnTimeField.DTN_EPOCH + + def i2h(self, pkt, x): + dtval = DtnTimeField.dtntime_to_datetime(x) + if dtval is None: + return 'zero' + return dtval.isoformat(timespec='milliseconds') + + def i2repr(self, pkt, x): + return self.i2h(pkt, x) + + def h2i(self, pkt, x): + return self.any2i(pkt, x) + + def any2i(self, pkt, x): + if x is None: + return None + + elif isinstance(x, datetime.datetime): + return DtnTimeField.datetime_to_dtntime(x) + + elif isinstance(x, (str, bytes)): + return DtnTimeField.datetime_to_dtntime( + datetime.datetime.fromisoformat(x) + ) + + elif isinstance(x, CBOR_UNSIGNED_INTEGER): + return x.val + + return int(x) + + def randval(self): + return volatile.RandNum(0, int(2 ** 16)) + + +class BundleTimestamp(CBOR_Packet): + ''' A structured representation of an DTN Timestamp. + The timestamp is a two-tuple of (time, sequence number) + The creation time portion is automatically converted from a + :py:cls:`datetime.datetime` object and text. + ''' + CBOR_root = CBORF_ARRAY( + DtnTimeField('dtntime', default=0), + CBORF_UNSIGNED_INTEGER('seqno', default=0), + ) + +@enum.unique +class EidScheme(enum.IntEnum): + dtn = 1 + ipn = 2 + +_DTN_WELL_KNOWN_SSP = { + 0: 'none', +} +"""Compressed SSP encoding.""" + +@dataclass +class EidStruct: + scheme: EidScheme + ''' Scheme code point ''' + ssp: Union[int, str, list[int]] + ''' Scheme-specific part ''' + + @staticmethod + def from_text(text: str) -> 'EidStruct': + scheme_name, ssp_text = text.split(':', 1) + + scheme = EidScheme[scheme_name.lower()] + ssp = None + if scheme == EidScheme.dtn: + for key, val in _DTN_WELL_KNOWN_SSP.items(): + if ssp_text == val: + ssp = key + break + if ssp is None: + ssp = ssp_text + + elif scheme == EidScheme.ipn: + # force handling as decimal + parts = [int(part, 10) for part in ssp_text.split('.')] + if not 2 <= len(parts) <= 3: + raise ValueError('IPN SSP must be 2 or 3 elements') + + ssp = parts + else: + raise ValueError(f'BP EID scheme {scheme} not understood') + + return EidStruct( + scheme=scheme, + ssp=ssp + ) + + def to_text(self) -> str: + if self.scheme == EidScheme.dtn: + # DTN scheme + if isinstance(self.ssp, int): + ssp = _DTN_WELL_KNOWN_SSP[self.ssp] + else: + ssp = str(self.ssp) + return 'dtn:' + ssp + elif self.scheme == EidScheme.ipn: + # IPN scheme, 2 or 3 element forms + return 'ipn:' + '.'.join(['{:d}'.format(part) for part in self.ssp]) + else: + raise ValueError(f'BP EID scheme {self.scheme_id} not understood') + + @staticmethod + def from_cbor(item: CBOR_Object) -> 'EidStruct': + if not isinstance(item, CBOR_ARRAY): + raise TypeError(f"Need an array, have {item}") + scheme_id, ssp_item = item.val + scheme = EidScheme(scheme_id) + if scheme == EidScheme.dtn: + ssp = ssp_item.val + elif scheme == EidScheme.ipn: + ssp = [int(item.val) for item in ssp_item.val] + else: + raise ValueError + + return EidStruct(scheme=scheme, ssp=ssp) + + def to_cbor(self) -> CBOR_Object: + if self.scheme == EidScheme.dtn: + if isinstance(self.ssp, int): + ssp_item = CBOR_UNSIGNED_INTEGER(self.ssp) + else: + ssp_item = CBOR_TEXT_STRING(self.ssp) + elif self.scheme == EidScheme.ipn: + ssp_item = [CBOR_UNSIGNED_INTEGER(part) for part in self.ssp] + else: + raise ValueError + + return CBOR_ARRAY([ + CBOR_UNSIGNED_INTEGER(int(self.scheme)), + ssp_item + ]) + +class BundleEidField(CBORF_field[EidStruct, CBOR_ARRAY]): + ''' Provide a human-friendly representation of a BP Endpoint ID (EID) as + a single field. + The EID is a two-item array of (scheme ID, scheme-specific part). + ''' + + def _wrap(self, val): + # type: (Any) -> _A + return self.any2i(None, val) + + def i2h(self, _pkt, x): + # type: (CBOR_Packet, _I) -> Any + # Translate to text form for known schemes + if x is None: + return None + + if not isinstance(x, EidStruct): + raise ValueError(f'EID must be decoded into an EidStruct') + x = cast(EidStruct, x) + + return x.to_text() + + def h2i(self, _pkt, x): + # type: (Optional[Packet], Any) -> I + if x is None: + return None + + return EidStruct.from_text(x) + + def any2i(self, pkt, x): + if x is None: + return None + + if isinstance(x, str): + return self.h2i(pkt, x) + return x + + def i2repr(self, pkt, x): + return self.i2h(pkt, x) + + def _encode(self, x): + # type: (Any) -> bytes + if isinstance(x, str): + x = EidStruct.from_text(x) + return CBORcodec_ARRAY.enc(x) if isinstance(x, CBOR_Object) else x.to_cbor().enc() + + def m2i(self, pkt, s): + # type: (CBOR_Packet, bytes) -> Tuple[_A, bytes] + item, remain = CBORcodec_ARRAY.dec(s) + return EidStruct.from_cbor(item), remain + +class AbstractBlock: + ''' Represent an abstract block internal interface mixin. + + .. py:attribute:: crc_type_name + The name of the CRC-type field. + .. py:attribute:: crc_value_name + The name of the CRC-value field. + ''' + + @enum.unique + class CrcType(enum.IntEnum): + ''' CRC type values. + ''' + NONE = 0 + CRC16 = 1 + CRC32 = 2 + + # Map from CRC type to algorithm + CRC_DEFN = { + CrcType.CRC16: { # BPv7 CRC-16 X.25 + 'func': crcmod.predefined.mkPredefinedCrcFun('x-25'), + 'encode': lambda val: struct.pack('>H', val) + }, + CrcType.CRC32: { # BPv7 CRC-32 Castagnoli + 'func': crcmod.predefined.mkPredefinedCrcFun('crc-32c'), + 'encode': lambda val: struct.pack('>L', val) + }, + } + + _crc_type_name = 'crc_type' + ''' Field name of the CRC Type in the leaf packet class. ''' + _crc_value_name = 'crc_value' + ''' Field name of the CRC Value in the leaf packet class. ''' + + def fill_fields(self): + ''' Fill all fields so that the block is the full size it needs + to be for encoding encoding with build(). + Derived classes should populate their block-type-specific-data also. + ''' + crc_type = self.getfieldval(self._crc_type_name).val + crc_value = self.getfieldval(self._crc_value_name) + if crc_type and not crc_value: + defn = AbstractBlock.CRC_DEFN[crc_type] + # Encode with a zero-valued CRC field + self.setfieldval(self._crc_value_name, defn['encode'](0)) + + def update_crc(self, keep_existing=True): + ''' Update this block's CRC field from the current field data + only if the current CRC (field not default) value is None. + ''' + # class-level configuration + if self._crc_type_name is None or self._crc_value_name is None: + return + + crc_type = self.getfieldval(self._crc_type_name).val + if crc_type == 0: + crc_value = None + else: + crc_value = self.fields.get(self._crc_value_name) + if not keep_existing or crc_value is None: + defn = AbstractBlock.CRC_DEFN[crc_type] + # Encode with a zero-valued CRC field + self.fields[self._crc_value_name] = defn['encode'](0) + pre_crc = self.build() + crc_int = defn['func'](pre_crc) + crc_value = defn['encode'](crc_int) + + self.fields[self._crc_value_name] = crc_value + + def check_crc(self): + ''' Check the current CRC value, if enabled. + :return: True if the CRC is disabled or it is valid. + ''' + if self._crc_type_name is None or self._crc_value_name is None: + return True + + crc_type = self.getfieldval(self._crc_type_name).val + crc_value = self.fields.get(self._crc_value_name) + if crc_type == 0: + valid = crc_value is None + else: + defn = AbstractBlock.CRC_DEFN[crc_type] + # Encode with a zero-valued CRC field + self.fields[self._crc_value_name] = defn['encode'](0) + pre_crc = self.build() + crc_int = defn['func'](pre_crc) + valid = crc_value == defn['encode'](crc_int) + # Restore old value + self.fields[self._crc_value_name] = crc_value + + return valid + +class CBORF_CONDITIONAL(CBORF_field[Any, Any], ConditionalField): + """Derive from ConditionalField to trigger builtin logic.""" + + def __init__(self, + fld, # type: CBORF_field + cond, # type: Callable[[Packet], bool] + ): + ConditionalField.__init__(self, fld, cond) + + def __repr__(self): + # type: () -> str + return "<%s%r>" % (self.__class__.__name__, self.fld) + + @property + def owners(self): + return self.fld.owners + + def _wrap(self, x): + return self.fld._wrap(x) + + def _encode(self, x): + return self.fld._encode(x) + + def m2i(self, pkt, s): + # type: (CBOR_Packet, bytes) -> Tuple[_A, bytes] + if self._evalcond(pkt): + return self.fld.m2i(pkt, s) + else: + return None, s + + +class PrimaryBlock(CBOR_Packet, AbstractBlock): + ''' The primary block definition ''' + + @enum.unique + class Flag(enum.IntFlag): + ''' Bundle processing control flags. + ''' + REQ_DELETION_REPORT = 0x040000 + ''' bundle deletion status reports are requested. ''' + REQ_DELIVERY_REPORT = 0x020000 + ''' bundle delivery status reports are requested. ''' + REQ_FORWARDING_REPORT = 0x010000 + ''' bundle forwarding status reports are requested. ''' + REQ_RECEPTION_REPORT = 0x004000 + ''' bundle reception status reports are requested. ''' + REQ_STATUS_TIME = 0x000040 + ''' status time is requested in all status reports. ''' + USER_APP_ACK = 0x000020 + ''' user application acknowledgement is requested. ''' + NO_FRAGMENT = 0x000004 + ''' bundle must not be fragmented. ''' + PAYLOAD_ADMIN = 0x000002 + ''' payload is an administrative record. ''' + IS_FRAGMENT = 0x000001 + ''' bundle is a fragment. ''' + + def is_fragment(self) -> bool: + """Determine if this bundle is an ADU fragment.""" + flags = self.getfieldval('bundle_flags').val + return bool(flags & PrimaryBlock.Flag.IS_FRAGMENT) + + CBOR_root = CBORF_ARRAY( + CBORF_UNSIGNED_INTEGER('version', default=7), + CBORF_UNSIGNED_INTEGER('bundle_flags', default=0), #FIXME FLAGS + CBORF_UNSIGNED_INTEGER('crc_type', default=AbstractBlock.CrcType.NONE), #FIXME ENUM + BundleEidField('destination', default='dtn:none'), + BundleEidField('source', default='dtn:none'), + BundleEidField('report_to', default='dtn:none'), + CBORF_PACKET('create_ts', default=None, cls=BundleTimestamp), + CBORF_UNSIGNED_INTEGER('lifetime', default=0), + CBORF_CONDITIONAL( + CBORF_UNSIGNED_INTEGER('fragment_offset', default=0), + cond=is_fragment + ), + CBORF_CONDITIONAL( + CBORF_UNSIGNED_INTEGER('total_app_data_len', default=0), + cond=is_fragment + ), + CBORF_CONDITIONAL( + CBORF_BYTE_STRING('crc_value', default=None), + cond=lambda block: block.getfieldval('crc_type').val != 0 + ), + ) + +class CanonicalBlock(CBOR_Packet, AbstractBlock): + ''' The canonical block definition with a block-type-specific data (BTSD) + field containing a dissected Packet. + ''' + + @enum.unique + class Flag(enum.IntFlag): + ''' Block processing control flags ''' + REMOVE_IF_NO_PROCESS = 0x10 + ''' block must be removed from bundle if it can't be processed. ''' + DELETE_IF_NO_PROCESS = 0x04 + ''' bundle must be deleted if block can't be processed. ''' + STATUS_IF_NO_PROCESS = 0x02 + ''' transmission of a status report is requested if block can't be + processed. ''' + REPLICATE_IN_FRAGMENT = 0x01 + ''' block must be replicated in every fragment. ''' + + def btsd_class(self, data: bytes): + cls = super().guess_payload_class(data) + return cls + + CBOR_root = CBORF_ARRAY( + CBORF_UNSIGNED_INTEGER('type_code', default=None), + CBORF_UNSIGNED_INTEGER('block_num', default=None), + CBORF_UNSIGNED_INTEGER('block_flags', default=0), #FIXME FLAGS + CBORF_UNSIGNED_INTEGER('crc_type', default=AbstractBlock.CrcType.NONE), #FIXME ENUM + CBORF_BYTE_STRING_PACKET('btsd', default=None, + cls_cb=btsd_class), + CBORF_CONDITIONAL( + CBORF_BYTE_STRING('crc_value', default=None), + cond=lambda block: block.getfieldval('crc_type') != 0 + ), + ) + + def self_build(self, *args, **kwargs): + # derive the block type from BTSD packet class + if 'block_type' not in self.fields and 'btsd' in self.fields: + fld, fval = self.getfield_and_val('btsd') + fval = fval._overload_fields.get(CanonicalBlock) + print('overload', fval) + if fval and 'block_type' in fval: + self.fields['block_type'] = fval['block_type'] + + self.fill_fields() + self.update_crc(keep_existing=True) + + return super().self_build(*args, **kwargs) + + +class PreviousNodeBlock(CBOR_Packet): + ''' Block data content from Section 4.4.1 of RFC 9171. + ''' + CBOR_root = BundleEidField('node', default=None) + + +class BundleAgeBlock(CBOR_Packet): + ''' Block data content from Section 4.4.2 of RFC 9171. + ''' + CBOR_root = CBORF_UNSIGNED_INTEGER('age', default=None) + + +class HopCountBlock(CBOR_Packet): + ''' Block data content from Section 4.4.3 of RFC 9171. + ''' + CBOR_root = CBORF_ARRAY( + CBORF_UNSIGNED_INTEGER('limit', default=None), + CBORF_UNSIGNED_INTEGER('count', default=0), + ) + + +bind_layers(CanonicalBlock, PreviousNodeBlock, type_code=6) +bind_layers(CanonicalBlock, BundleAgeBlock, type_code=7) +bind_layers(CanonicalBlock, HopCountBlock, type_code=10) + +""" +class BpsecKeyValPair(CborArrayPacket): + fields_desc = ( + CborUintField('key'), + CborAnyField('val'), + ) + + +class BpsecKeyValList(CborArrayPacket): + fields_desc = ( + PacketListField('pairs', [], pkt_cls=BpsecKeyValPair, + count_from=lambda pkt: pkt.array_head_arg), + ) + + +class BpsecKeyValListList(CborArrayPacket): + fields_desc = ( + PacketListField('items', [], pkt_cls=BpsecKeyValList, + count_from=lambda pkt: pkt.array_head_arg), + ) + + +class AbstractSecurityBock(CborSequencePacket): + ''' Block data content from Section 3.6 of RFC 9172. + ''' + + @enum.unique + class Flag(enum.IntFlag): + ''' ASB flags. + Defined in Section 3.6 of RFC 9172. + ''' + PARAMETERS = 0x01 + ''' Security context parameters present. ''' + + fields_desc = ( + CborFieldArrayField('targets', [], field=CborIntField('blk_num')), + CborIntField('context_id'), + CborFlagsField('flags', 0, flags=Flag), + BundleEidField('source', default=None), + ConditionalField( + PacketField('parameters', [], pkt_cls=BpsecKeyValList), + cond=lambda pkt: pkt.flags & AbstractSecurityBock.Flag.PARAMETERS + ), + # one packet in this list per target + PacketField('tgt_results', [], pkt_cls=BpsecKeyValListList), + ) + + +bind_layers(CanonicalBlock, AbstractSecurityBock, type_code=11) +bind_layers(CanonicalBlock, AbstractSecurityBock, type_code=12) +""" + + +class BundleV7(CBOR_Packet): + ''' An entire decoded bundle contents. + + Bundles with administrative records are handled specially in that the + AdminRecord object will be made a (scapy) payload of the "payload block" + which is block type code 1. + ''' + + BLOCK_TYPE_PAYLOAD = 1 + BLOCK_NUM_PAYLOAD = 1 + + def block_until_break(self, data: bytes): + major_type, arg, _ = CBOR_decode_head(data) + if major_type == CBOR_MajorTypes.SIMPLE_AND_FLOAT and arg is None: + return None + print('new block') + return CanonicalBlock + + CBOR_root = CBORF_INDEFINITE_ARRAY( + CBORF_PACKET('primary', default=PrimaryBlock(), cls=PrimaryBlock), + CBORF_SEQUENCE_OF('blocks', default=[], + cls_cb=block_until_break), + ) + + +conf.debug_dissector = True diff --git a/test/contrib/bpv7.uts b/test/contrib/bpv7.uts new file mode 100644 index 00000000000..c81f0ff0aa4 --- /dev/null +++ b/test/contrib/bpv7.uts @@ -0,0 +1,176 @@ +% Bundle Protocol Version 7 tests for Scapy + ++ EID CODEC + += EID construct default + +from scapy.cbor import CBORF_SEQUENCE +from scapy.cborpacket import CBOR_Packet +from scapy.contrib.bpv7 import BundleEidField + +class TestPkt(CBOR_Packet): + CBOR_root = CBORF_SEQUENCE( + BundleEidField('eid', default='dtn:none'), + ) + +pkt = TestPkt() +pkt.show() +outdata = bytes(pkt) +print(outdata.hex()) +assert outdata == bytes.fromhex('820100') + +pkt = TestPkt( + eid="dtn://n/s" +) +pkt.show() +outdata = bytes(pkt) +print(outdata.hex()) +assert outdata == bytes.fromhex('8201652F2F6E2F73') + +pkt = TestPkt( + eid="ipn:1.2.3" +) +pkt.show() +outdata = bytes(pkt) +print(outdata.hex()) +assert outdata == bytes.fromhex('820283010203') + + ++ BPv7 CODEC + += construct default + +from scapy.contrib.bpv7 import * +pkt = BundleV7() +pkt.show() +outdata = bytes(pkt) +print(outdata.hex()) +assert outdata[:1] == b'\x9f' +assert outdata[-1:] == b'\xff' + += construct only payload + +from scapy.contrib.bpv7 import * + +pkt = BundleV7( + primary=PrimaryBlock( + crc_type=2, + destination='ipn:1.2.3', + create_ts=BundleTimestamp( + dtntime='2025-11-26T15:00:00Z', + seqno=1, + ), + lifetime=3600000, + ), + blocks=[ + CanonicalBlock( + type_code=1, + block_num=1, + crc_type=2, + btsd=Raw(b'hi'), + ), + ] +) +pkt.show() +outdata = bytes(pkt) +print(outdata.hex()) +assert outdata[:1] == b'\x9f' +assert outdata[-1:] == b'\xff' + += construct with extensions + +pkt = BundleV7( + primary=PrimaryBlock( + crc_type=2, + destination='ipn:1.2.3', + create_ts=BundleTimestamp( + dtntime='2025-11-26T15:00:00Z', + seqno=1, + ), + lifetime=3600000, + ), + blocks=[ + CanonicalBlock( + block_num=2, + crc_type=2, + btsd=PreviousNodeBlock(node='ipn:3.2.0'), + ), + CanonicalBlock( + type_code=1, + block_num=1, + crc_type=2, + btsd=Raw(b'hi'), + ), + ] +) +pkt.show() +outdata = bytes(pkt) +print(outdata.hex()) +assert len(outdata) > 5 +assert outdata[:1] == b'\x9f' +assert outdata[-1:] == b'\xff' +wrpcap('/tmp/foo.pcap', Ether()/IP()/UDP(sport=4556,dport=4556)/pkt) + + += decoding example from Appendix A.1.1.3 of RFC 9173 + +from scapy.contrib.bpv7 import * +data = bytes.fromhex('9f88070000820282010282028202018202820201820018281a000f424085010100005823526561647920746f2067656e657261746520612033322d62797465207061796c6f6164ff') +pkt = BundleV7(data) +print(repr(pkt)) +pkt.show() +assert pkt.primary.source == 'ipn:2.1' +assert pkt.primary.destination == 'ipn:1.2' + +pkt.clear_cache() +outdata = bytes(pkt) +print(outdata.hex()) +assert outdata == data + += decoding example from Appendix A.1.1.4 of RFC 9173 + +from scapy.contrib.bpv7 import * +data = bytes.fromhex('9f88070000820282010282028202018202820201820018281a000f424085070200004319012c85010100005823526561647920746f2067656e657261746520612033322d62797465207061796c6f6164ff') +pkt = BundleV7(data) +print(repr(pkt)) +pkt.show() +assert pkt.primary.source == 'ipn:2.1' +assert pkt.primary.destination == 'ipn:1.2' + +pkt.clear_cache() +outdata = bytes(pkt) +print(outdata.hex()) +assert outdata == data + += decoding example from Appendix A.1.4 of RFC 9173 + +from scapy.contrib.bpv7 import * +data = bytes.fromhex('9f88070000820282010282028202018202820201820018281a000f4240850b0200005856810101018202820201828201078203008181820158403bdc69b3a34a2b5d3a8554368bd1e808f606219d2a10a846eae3886ae4ecc83c4ee550fdfb1cc636b904e2f1a73e303dcd4b6ccece003e95e8164dcc89a156e185010100005823526561647920746f2067656e657261746520612033322d62797465207061796c6f6164ff') +pkt = BundleV7(data) +print(repr(pkt)) +pkt.show() +assert pkt.blocks[0].type_code == 11 +#assert pkt.blocks[0].btsd.targets == [1] +#assert pkt.blocks[0].btsd.context_id == 1 +#assert pkt.blocks[0].btsd.source == 'ipn:2.1' + +pkt.clear_cache() +outdata = bytes(pkt) +print(outdata.hex()) +assert outdata == data + + += decoding example from Appendix A of draft-dtn-bpsec-cose + +from scapy.contrib.bpv7 import * +data = bytes.fromhex('9f880700008201692f2f6473742f7376638201692f2f7372632f7376638201662f2f7372632f821b000000bd51281400001a000f42408501010000466568656c6c6fff') +pkt = BundleV7(data) +print(repr(pkt)) +pkt.show() +assert pkt.primary.source == 'dtn://src/svc' +assert pkt.primary.destination == 'dtn://dst/svc' + +pkt.clear_cache() +outdata = bytes(pkt) +print(outdata.hex()) +assert outdata == data From d5417af18fafc1f24e8e0d96d5092b0c794a9df5 Mon Sep 17 00:00:00 2001 From: Brian Sipos Date: Tue, 4 Aug 2026 23:12:05 -0400 Subject: [PATCH 2/6] Move common conditional field into cbor library --- scapy/cbor/__init__.py | 2 ++ scapy/cbor/cborfields.py | 38 ++++++++++++++++++++- scapy/contrib/bpv7.py | 71 ++++++++++++++-------------------------- 3 files changed, 63 insertions(+), 48 deletions(-) diff --git a/scapy/cbor/__init__.py b/scapy/cbor/__init__.py index 05005fc256a..7a9935c3e04 100644 --- a/scapy/cbor/__init__.py +++ b/scapy/cbor/__init__.py @@ -63,6 +63,7 @@ CBORF_MAP, CBORF_SEMANTIC_TAG, CBORF_optional, + CBORF_CONDITIONAL, CBORF_PACKET, CBORF_BYTE_STRING_PACKET, ) @@ -126,6 +127,7 @@ "CBORF_SEMANTIC_TAG", # Complex fields "CBORF_optional", + "CBORF_CONDITIONAL", "CBORF_PACKET", "CBORF_BYTE_STRING_PACKET", ] diff --git a/scapy/cbor/cborfields.py b/scapy/cbor/cborfields.py index d01956a1c3b..d44a59dcad9 100644 --- a/scapy/cbor/cborfields.py +++ b/scapy/cbor/cborfields.py @@ -47,7 +47,7 @@ RandField, ) -from scapy import packet +from scapy import packet, fields from typing import ( Any, @@ -1087,6 +1087,42 @@ def i2repr(self, pkt, x): return self._field.i2repr(pkt, x) +class CBORF_CONDITIONAL(CBORF_field[Any, Any], fields.ConditionalField): + """ + Wrapper making a :class:`CBORF_field` conditional on some other packet state. + + Derive from ConditionalField to trigger builtin logic. + """ + + def __init__(self, + fld, # type: CBORF_field + cond, # type: Callable[[Packet], bool] + ): + fields.ConditionalField.__init__(self, fld, cond) + # Leave CBORF_field uninitialized + + def __repr__(self): + # type: () -> str + return "<%s%r>" % (self.__class__.__name__, self.fld) + + @property + def owners(self): + return self.fld.owners + + def _wrap(self, x): + return self.fld._wrap(x) + + def _encode(self, x): + return self.fld._encode(x) + + def m2i(self, pkt, s): + # type: (CBOR_Packet, bytes) -> Tuple[_A, bytes] + if self._evalcond(pkt): + return self.fld.m2i(pkt, s) + else: + return None, s + + class CBORF_PACKET(CBORF_field['CBOR_Packet', Optional['CBOR_Packet']]): """ CBOR field that encapsulates a nested :class:`CBOR_Packet`. diff --git a/scapy/contrib/bpv7.py b/scapy/contrib/bpv7.py index 00be89f1e73..1a0734f9699 100644 --- a/scapy/contrib/bpv7.py +++ b/scapy/contrib/bpv7.py @@ -6,22 +6,20 @@ # scapy.contrib.description = Bundle Protocol Version 7 (BPv7) # scapy.contrib.status = loads -import crcmod from dataclasses import dataclass import datetime import enum import logging import struct -from typing import Any, List, Optional, Tuple, Union, cast +from typing import Optional, Union, cast from scapy import volatile from scapy.config import conf -from scapy.packet import Packet, bind_layers -from scapy.fields import ConditionalField +from scapy.packet import bind_layers from scapy.cbor.cborcodec import CBOR_encode_head, CBOR_decode_head, CBOR_MajorTypes from scapy.cbor import ( CBORF_field, CBORF_UNSIGNED_INTEGER, CBORF_INTEGER, CBORF_ARRAY, CBORF_BYTE_STRING, - CBORF_SEQUENCE, CBORF_SEQUENCE_OF, CBORF_PACKET, CBORF_BYTE_STRING_PACKET, + CBORF_CONDITIONAL, CBORF_SEQUENCE_OF, CBORF_PACKET, CBORF_BYTE_STRING_PACKET, CBORcodec_ARRAY, CBOR_UNSIGNED_INTEGER, CBOR_TEXT_STRING, CBOR_ARRAY, CBOR_Object ) @@ -117,6 +115,7 @@ class BundleTimestamp(CBOR_Packet): @enum.unique class EidScheme(enum.IntEnum): + """Handled EID scheme names and values.""" dtn = 1 ipn = 2 @@ -127,6 +126,9 @@ class EidScheme(enum.IntEnum): @dataclass class EidStruct: + """ + Internal state for the :class:`BundleEidField` class. + """ scheme: EidScheme ''' Scheme code point ''' ssp: Union[int, str, list[int]] @@ -136,9 +138,13 @@ class EidStruct: def from_text(text: str) -> 'EidStruct': scheme_name, ssp_text = text.split(':', 1) - scheme = EidScheme[scheme_name.lower()] + try: + scheme = EidScheme[scheme_name.lower()] + except KeyError: + raise ValueError(f'BP EID scheme {scheme_name} not understood') ssp = None if scheme == EidScheme.dtn: + # some SSP values are well-known and compressed for key, val in _DTN_WELL_KNOWN_SSP.items(): if ssp_text == val: ssp = key @@ -151,15 +157,12 @@ def from_text(text: str) -> 'EidStruct': parts = [int(part, 10) for part in ssp_text.split('.')] if not 2 <= len(parts) <= 3: raise ValueError('IPN SSP must be 2 or 3 elements') - ssp = parts + else: - raise ValueError(f'BP EID scheme {scheme} not understood') + raise ValueError("Invalid scheme state") - return EidStruct( - scheme=scheme, - ssp=ssp - ) + return EidStruct(scheme=scheme, ssp=ssp) def to_text(self) -> str: if self.scheme == EidScheme.dtn: @@ -173,20 +176,24 @@ def to_text(self) -> str: # IPN scheme, 2 or 3 element forms return 'ipn:' + '.'.join(['{:d}'.format(part) for part in self.ssp]) else: - raise ValueError(f'BP EID scheme {self.scheme_id} not understood') + raise ValueError("Invalid scheme state") @staticmethod def from_cbor(item: CBOR_Object) -> 'EidStruct': if not isinstance(item, CBOR_ARRAY): raise TypeError(f"Need an array, have {item}") scheme_id, ssp_item = item.val - scheme = EidScheme(scheme_id) + try: + scheme = EidScheme(scheme_id) + except ValueError: + raise ValueError(f'BP EID scheme {scheme_id} not understood') + if scheme == EidScheme.dtn: ssp = ssp_item.val elif scheme == EidScheme.ipn: ssp = [int(item.val) for item in ssp_item.val] else: - raise ValueError + raise ValueError("Invalid scheme state") return EidStruct(scheme=scheme, ssp=ssp) @@ -199,8 +206,8 @@ def to_cbor(self) -> CBOR_Object: elif self.scheme == EidScheme.ipn: ssp_item = [CBOR_UNSIGNED_INTEGER(part) for part in self.ssp] else: - raise ValueError - + raise ValueError("Invalid scheme state") + return CBOR_ARRAY([ CBOR_UNSIGNED_INTEGER(int(self.scheme)), ssp_item @@ -349,36 +356,6 @@ def check_crc(self): return valid -class CBORF_CONDITIONAL(CBORF_field[Any, Any], ConditionalField): - """Derive from ConditionalField to trigger builtin logic.""" - - def __init__(self, - fld, # type: CBORF_field - cond, # type: Callable[[Packet], bool] - ): - ConditionalField.__init__(self, fld, cond) - - def __repr__(self): - # type: () -> str - return "<%s%r>" % (self.__class__.__name__, self.fld) - - @property - def owners(self): - return self.fld.owners - - def _wrap(self, x): - return self.fld._wrap(x) - - def _encode(self, x): - return self.fld._encode(x) - - def m2i(self, pkt, s): - # type: (CBOR_Packet, bytes) -> Tuple[_A, bytes] - if self._evalcond(pkt): - return self.fld.m2i(pkt, s) - else: - return None, s - class PrimaryBlock(CBOR_Packet, AbstractBlock): ''' The primary block definition ''' From c526c850d0ea233c960747701a53e26353298fed Mon Sep 17 00:00:00 2001 From: Brian Sipos Date: Tue, 4 Aug 2026 23:34:40 -0400 Subject: [PATCH 3/6] Workaround builtin error --- scapy/cbor/cbor.py | 11 ++++++----- scapy/contrib/bpv7.py | 2 +- test/contrib/bpv7.uts | 9 ++++++--- 3 files changed, 13 insertions(+), 9 deletions(-) diff --git a/scapy/cbor/cbor.py b/scapy/cbor/cbor.py index 5588dba664d..2e7707fc975 100644 --- a/scapy/cbor/cbor.py +++ b/scapy/cbor/cbor.py @@ -296,11 +296,12 @@ def __new__(cls, 'Type[CBOR_Object[Any]]', super(CBOR_Object_metaclass, cls).__new__(cls, name, bases, dct) ) - try: - c.tag.register_cbor_object(c) - except Exception: - # Some objects may not have tags yet - log_runtime.warning("Failed to register CBOR object %r" % c) + if c.tag is not None: + try: + c.tag.register_cbor_object(c) + except Exception: + # Some objects may not have tags yet + log_runtime.exception("Failed to register CBOR object %r" % c) return c diff --git a/scapy/contrib/bpv7.py b/scapy/contrib/bpv7.py index 1a0734f9699..0bdb5bbc3e5 100644 --- a/scapy/contrib/bpv7.py +++ b/scapy/contrib/bpv7.py @@ -5,7 +5,7 @@ # scapy.contrib.description = Bundle Protocol Version 7 (BPv7) # scapy.contrib.status = loads - +import crcmod from dataclasses import dataclass import datetime import enum diff --git a/test/contrib/bpv7.uts b/test/contrib/bpv7.uts index c81f0ff0aa4..06aee1d03c2 100644 --- a/test/contrib/bpv7.uts +++ b/test/contrib/bpv7.uts @@ -1,11 +1,14 @@ -% Bundle Protocol Version 7 tests for Scapy +% Bundle Protocol Version 7 test campaign + ++ Syntax check += Import the BPv7 layer +from scapy.contrib.bpv7 import * + EID CODEC = EID construct default -from scapy.cbor import CBORF_SEQUENCE -from scapy.cborpacket import CBOR_Packet +from scapy.cbor import * from scapy.contrib.bpv7 import BundleEidField class TestPkt(CBOR_Packet): From 9ffc3489a3f5fa844f4769d4906f1e7c234785f5 Mon Sep 17 00:00:00 2001 From: Brian Sipos Date: Wed, 5 Aug 2026 10:37:58 -0400 Subject: [PATCH 4/6] Use scapy native CRC processing --- scapy/cbor/cborfields.py | 2 + scapy/contrib/bpv7.py | 163 ++++++++++++++++++++------------------- scapy/libs/crc.py | 22 ++++++ test/contrib/bpv7.uts | 52 ++++++++++++- 4 files changed, 160 insertions(+), 79 deletions(-) diff --git a/scapy/cbor/cborfields.py b/scapy/cbor/cborfields.py index d44a59dcad9..0a598b3e111 100644 --- a/scapy/cbor/cborfields.py +++ b/scapy/cbor/cborfields.py @@ -120,6 +120,8 @@ def i2repr(self, pkt, x): def i2h(self, pkt, x): # type: (CBOR_Packet, _I) -> Any + if isinstance(x, CBOR_Object): + return x.val return x def m2i(self, pkt, s): diff --git a/scapy/contrib/bpv7.py b/scapy/contrib/bpv7.py index 0bdb5bbc3e5..2102adbc0f6 100644 --- a/scapy/contrib/bpv7.py +++ b/scapy/contrib/bpv7.py @@ -5,15 +5,15 @@ # scapy.contrib.description = Bundle Protocol Version 7 (BPv7) # scapy.contrib.status = loads -import crcmod + from dataclasses import dataclass import datetime import enum -import logging import struct from typing import Optional, Union, cast from scapy import volatile from scapy.config import conf +from scapy.error import log_runtime from scapy.packet import bind_layers from scapy.cbor.cborcodec import CBOR_encode_head, CBOR_decode_head, CBOR_MajorTypes from scapy.cbor import ( @@ -26,8 +26,7 @@ from scapy.cborpacket import ( CBOR_Packet, ) - -LOG_RUNTIME = logging.getLogger("scapy.runtime") +from scapy.libs.crc import CRC, CRC_16_X25, CRC_32C class CBORF_INDEFINITE_ARRAY(CBORF_ARRAY): @@ -264,6 +263,37 @@ def m2i(self, pkt, s): item, remain = CBORcodec_ARRAY.dec(s) return EidStruct.from_cbor(item), remain +@enum.unique +class CrcType(enum.IntEnum): + """ + CRC type values defined in RFC 9171. + """ + NONE = 0 + CRC16 = 1 + CRC32 = 2 + +@dataclass +class CrcInfo: + """ + Processing for a specific :class:`CrcType` + """ + cls: CRC + encode: 'Callable[[int], bytes]' + +_CRC_DEFN: dict[CrcType, CrcInfo] = { + CrcType.CRC16: CrcInfo( + # BPv7 CRC-16 X.25 + cls=CRC_16_X25, + encode=lambda val: struct.pack('>H', val) + ), + CrcType.CRC32: CrcInfo( + # BPv7 CRC-32 Castagnoli + cls=CRC_32C, + encode=lambda val: struct.pack('>L', val) + ), +} +"""Map from available CRC type to info.""" + class AbstractBlock: ''' Represent an abstract block internal interface mixin. @@ -273,86 +303,63 @@ class AbstractBlock: The name of the CRC-value field. ''' - @enum.unique - class CrcType(enum.IntEnum): - ''' CRC type values. - ''' - NONE = 0 - CRC16 = 1 - CRC32 = 2 - - # Map from CRC type to algorithm - CRC_DEFN = { - CrcType.CRC16: { # BPv7 CRC-16 X.25 - 'func': crcmod.predefined.mkPredefinedCrcFun('x-25'), - 'encode': lambda val: struct.pack('>H', val) - }, - CrcType.CRC32: { # BPv7 CRC-32 Castagnoli - 'func': crcmod.predefined.mkPredefinedCrcFun('crc-32c'), - 'encode': lambda val: struct.pack('>L', val) - }, - } - _crc_type_name = 'crc_type' ''' Field name of the CRC Type in the leaf packet class. ''' _crc_value_name = 'crc_value' ''' Field name of the CRC Value in the leaf packet class. ''' - def fill_fields(self): - ''' Fill all fields so that the block is the full size it needs - to be for encoding encoding with build(). - Derived classes should populate their block-type-specific-data also. - ''' + def has_crc(self): + """ + Match the signature for CBORF_CONDITIONAL on the CRC Value field. + """ crc_type = self.getfieldval(self._crc_type_name).val - crc_value = self.getfieldval(self._crc_value_name) - if crc_type and not crc_value: - defn = AbstractBlock.CRC_DEFN[crc_type] - # Encode with a zero-valued CRC field - self.setfieldval(self._crc_value_name, defn['encode'](0)) + return crc_type != CrcType.NONE.value def update_crc(self, keep_existing=True): - ''' Update this block's CRC field from the current field data + """ + Update this block's CRC field from the current field data only if the current CRC (field not default) value is None. - ''' - # class-level configuration - if self._crc_type_name is None or self._crc_value_name is None: - return - - crc_type = self.getfieldval(self._crc_type_name).val - if crc_type == 0: + """ + crc_type = getattr(self, self._crc_type_name) + crc_value = getattr(self, self._crc_value_name) + if crc_type == CrcType.NONE: + # there should not be a value crc_value = None - else: - crc_value = self.fields.get(self._crc_value_name) - if not keep_existing or crc_value is None: - defn = AbstractBlock.CRC_DEFN[crc_type] - # Encode with a zero-valued CRC field - self.fields[self._crc_value_name] = defn['encode'](0) - pre_crc = self.build() - crc_int = defn['func'](pre_crc) - crc_value = defn['encode'](crc_int) - - self.fields[self._crc_value_name] = crc_value - - def check_crc(self): + elif crc_value is None or not keep_existing: + # there should be a value + defn = _CRC_DEFN[crc_type] + # Encode with a zero-valued CRC field + self.setfieldval(self._crc_value_name, defn.encode(0)) + pre_crc = self.do_build() + crc_int = defn.cls(pre_crc) + crc_value = defn.encode(crc_int) + + self.setfieldval(self._crc_value_name, crc_value) + + def check_crc(self) -> bool: ''' Check the current CRC value, if enabled. :return: True if the CRC is disabled or it is valid. ''' - if self._crc_type_name is None or self._crc_value_name is None: - return True - crc_type = self.getfieldval(self._crc_type_name).val - crc_value = self.fields.get(self._crc_value_name) - if crc_type == 0: - valid = crc_value is None + crc_type = getattr(self, self._crc_type_name) + crc_value = getattr(self, self._crc_value_name) or b"" + if crc_type == CrcType.NONE: + expect = b"" + valid = not crc_value else: - defn = AbstractBlock.CRC_DEFN[crc_type] - # Encode with a zero-valued CRC field - self.fields[self._crc_value_name] = defn['encode'](0) - pre_crc = self.build() - crc_int = defn['func'](pre_crc) - valid = crc_value == defn['encode'](crc_int) - # Restore old value - self.fields[self._crc_value_name] = crc_value + defn = _CRC_DEFN[crc_type] + # Encode and substitute with a zero-valued CRC field + pre_crc = self.do_build() + + crc_obj: CRC = defn.cls.create_context() + crc_obj.update(pre_crc[:-(crc_obj.size // 8)]) + crc_obj.update(defn.encode(0)) + crc_int = crc_obj.finish() + expect = defn.encode(crc_int) + valid = crc_value == expect + + if not valid: + log_runtime.warning('CRC check failed! Expected %s got %s' % (expect.hex(), crc_value.hex())) return valid @@ -391,7 +398,7 @@ def is_fragment(self) -> bool: CBOR_root = CBORF_ARRAY( CBORF_UNSIGNED_INTEGER('version', default=7), CBORF_UNSIGNED_INTEGER('bundle_flags', default=0), #FIXME FLAGS - CBORF_UNSIGNED_INTEGER('crc_type', default=AbstractBlock.CrcType.NONE), #FIXME ENUM + CBORF_UNSIGNED_INTEGER('crc_type', default=CrcType.NONE), #FIXME ENUM BundleEidField('destination', default='dtn:none'), BundleEidField('source', default='dtn:none'), BundleEidField('report_to', default='dtn:none'), @@ -407,7 +414,7 @@ def is_fragment(self) -> bool: ), CBORF_CONDITIONAL( CBORF_BYTE_STRING('crc_value', default=None), - cond=lambda block: block.getfieldval('crc_type').val != 0 + cond=AbstractBlock.has_crc ), ) @@ -437,28 +444,28 @@ def btsd_class(self, data: bytes): CBORF_UNSIGNED_INTEGER('type_code', default=None), CBORF_UNSIGNED_INTEGER('block_num', default=None), CBORF_UNSIGNED_INTEGER('block_flags', default=0), #FIXME FLAGS - CBORF_UNSIGNED_INTEGER('crc_type', default=AbstractBlock.CrcType.NONE), #FIXME ENUM + CBORF_UNSIGNED_INTEGER('crc_type', default=CrcType.NONE), #FIXME ENUM CBORF_BYTE_STRING_PACKET('btsd', default=None, cls_cb=btsd_class), CBORF_CONDITIONAL( CBORF_BYTE_STRING('crc_value', default=None), - cond=lambda block: block.getfieldval('crc_type') != 0 + cond=AbstractBlock.has_crc ), ) - def self_build(self, *args, **kwargs): + def self_build(self): + # type: () -> bytes + # derive the block type from BTSD packet class if 'block_type' not in self.fields and 'btsd' in self.fields: - fld, fval = self.getfield_and_val('btsd') + _fld, fval = self.getfield_and_val('btsd') fval = fval._overload_fields.get(CanonicalBlock) - print('overload', fval) if fval and 'block_type' in fval: self.fields['block_type'] = fval['block_type'] - self.fill_fields() self.update_crc(keep_existing=True) - return super().self_build(*args, **kwargs) + return super().self_build() class PreviousNodeBlock(CBOR_Packet): diff --git a/scapy/libs/crc.py b/scapy/libs/crc.py index f43e3e7e700..3838b73ce25 100644 --- a/scapy/libs/crc.py +++ b/scapy/libs/crc.py @@ -14,6 +14,7 @@ "CRCParam", "CRC_16", "CRC_32", + "CRC_32C", "CRC_16_CCITT", "CRC_32_AUTOSAR", "WELL_KNOWN_POLY", @@ -383,6 +384,17 @@ class CRC_32(CRC): reflect_output = True test_vectors = [(b"123456789", 0xcbf43926)] +class CRC_32C(CRC): + "aka Castagnoli" + name = "CRC-32C" + size = 32 + poly = 0x1edc6f41 + init_crc = 0xffffffff + xor = 0xffffffff + reflect_input = True + reflect_output = True + test_vectors = [(b"123456789", 0xe3069283)] + class CRC_16_CCITT(CRC): "aka KERMIT CRC" @@ -395,6 +407,16 @@ class CRC_16_CCITT(CRC): reflect_output = True test_vectors = [(b"\xcb\x37", 0x6b3e)] +class CRC_16_X25(CRC): + name = "CRC-16 X-25" + size = 16 + poly = 0x1021 + init_crc = 0xffff + xor = 0xffff + reflect_input = True + reflect_output = True + test_vectors = [(b"123456789", 0x906e)] + class CRC_32_AUTOSAR(CRC): name = "CRC32 AUTOSAR" diff --git a/test/contrib/bpv7.uts b/test/contrib/bpv7.uts index 06aee1d03c2..e055b22ba5f 100644 --- a/test/contrib/bpv7.uts +++ b/test/contrib/bpv7.uts @@ -6,7 +6,7 @@ from scapy.contrib.bpv7 import * + EID CODEC -= EID construct default += EID encode from scapy.cbor import * from scapy.contrib.bpv7 import BundleEidField @@ -39,12 +39,62 @@ print(outdata.hex()) assert outdata == bytes.fromhex('820283010203') ++ Block CODEC + += Primary default +from scapy.contrib.bpv7 import * +pkt = PrimaryBlock() +pkt.show() +outdata = bytes(pkt) +print(outdata.hex()) +assert outdata == bytes.fromhex('89070000820100820100820100000000') + += Canonical payload encode +from scapy.contrib.bpv7 import * +pkt = CanonicalBlock( + type_code=1, + block_num=1, + btsd=Raw(load=b""), +) +pkt.show() +outdata = bytes(pkt) +print(outdata.hex()) +assert outdata == bytes.fromhex('850101000040') + += Canonical with CRC encode +from scapy.contrib.bpv7 import * +pkt = CanonicalBlock( + type_code=1, + block_num=1, + crc_type=2, + btsd=Raw(load=b"hi"), +) +pkt.show() +pkt.show2() +outdata = bytes(pkt) +print(outdata.hex()) +assert outdata == bytes.fromhex('8601010002426869441ff585a4') + += Canonical empty decode +from scapy.contrib.bpv7 import * +pkt = CanonicalBlock(bytes.fromhex('850101000040')) +pkt.show() + += Canonical with CRC decode +from scapy.contrib.bpv7 import * +pkt = CanonicalBlock(bytes.fromhex('8601010002426869441ff585a4')) +pkt.show() +assert pkt.check_crc() + + + BPv7 CODEC = construct default from scapy.contrib.bpv7 import * pkt = BundleV7() +assert pkt.primary.crc_type == 0 + pkt.show() outdata = bytes(pkt) print(outdata.hex()) From 4ae09730e9a03a12c1d76f2b3624cc431eedc8bb Mon Sep 17 00:00:00 2001 From: Brian Sipos Date: Wed, 5 Aug 2026 21:41:00 -0400 Subject: [PATCH 5/6] Test and validate CRC from real bundles --- scapy/cbor/cborfields.py | 4 ++++ scapy/contrib/bpv7.py | 15 ++++++++++++--- scapy/fields.py | 5 ++++- test/contrib/bpv7.uts | 12 ++++-------- 4 files changed, 24 insertions(+), 12 deletions(-) diff --git a/scapy/cbor/cborfields.py b/scapy/cbor/cborfields.py index 0a598b3e111..d4d6128e570 100644 --- a/scapy/cbor/cborfields.py +++ b/scapy/cbor/cborfields.py @@ -84,6 +84,10 @@ class CBORF_element(object): class CBORF_field(CBORF_element, Generic[_I, _A]): + """Base class for CBOR items in packet fields. + The human form of values prefers the unwrapped, non :class:`CBOR_Object` value. + The internal form prefers the :class:`CBOR_Object` instance. + """ holds_packets = 0 islist = 0 CBOR_tag = None # type: Optional[Any] diff --git a/scapy/contrib/bpv7.py b/scapy/contrib/bpv7.py index 2102adbc0f6..708ae779193 100644 --- a/scapy/contrib/bpv7.py +++ b/scapy/contrib/bpv7.py @@ -418,6 +418,11 @@ def is_fragment(self) -> bool: ), ) + def self_build(self): + # type: () -> bytes + self.update_crc(keep_existing=True) + return super().self_build() + class CanonicalBlock(CBOR_Packet, AbstractBlock): ''' The canonical block definition with a block-type-specific data (BTSD) field containing a dissected Packet. @@ -464,7 +469,6 @@ def self_build(self): self.fields['block_type'] = fval['block_type'] self.update_crc(keep_existing=True) - return super().self_build() @@ -557,7 +561,10 @@ class BundleV7(CBOR_Packet): BLOCK_TYPE_PAYLOAD = 1 BLOCK_NUM_PAYLOAD = 1 - def block_until_break(self, data: bytes): + def _block_until_break(self, data: bytes): + """ + Callback to read canonical blocks until the outer indefinite break + """ major_type, arg, _ = CBOR_decode_head(data) if major_type == CBOR_MajorTypes.SIMPLE_AND_FLOAT and arg is None: return None @@ -567,8 +574,10 @@ def block_until_break(self, data: bytes): CBOR_root = CBORF_INDEFINITE_ARRAY( CBORF_PACKET('primary', default=PrimaryBlock(), cls=PrimaryBlock), CBORF_SEQUENCE_OF('blocks', default=[], - cls_cb=block_until_break), + cls_cb=_block_until_break), ) + def check_crc(self) -> bool: + return self.primary.check_crc() and all(blk.check_crc() for blk in self.blocks) conf.debug_dissector = True diff --git a/scapy/fields.py b/scapy/fields.py index 23c8fa774c3..48c9395eb5b 100644 --- a/scapy/fields.py +++ b/scapy/fields.py @@ -430,7 +430,10 @@ def addfield(self, pkt, s, val): def __getattr__(self, attr): # type: (str) -> Any - return getattr(self.fld, attr) + try: + return getattr(self.fld, attr) + except AttributeError: + return super().__getattr__(attr) class MultipleTypeField(_FieldContainer): diff --git a/test/contrib/bpv7.uts b/test/contrib/bpv7.uts index e055b22ba5f..130ab874fbd 100644 --- a/test/contrib/bpv7.uts +++ b/test/contrib/bpv7.uts @@ -90,11 +90,9 @@ assert pkt.check_crc() + BPv7 CODEC = construct default - from scapy.contrib.bpv7 import * pkt = BundleV7() assert pkt.primary.crc_type == 0 - pkt.show() outdata = bytes(pkt) print(outdata.hex()) @@ -102,9 +100,7 @@ assert outdata[:1] == b'\x9f' assert outdata[-1:] == b'\xff' = construct only payload - from scapy.contrib.bpv7 import * - pkt = BundleV7( primary=PrimaryBlock( crc_type=2, @@ -170,8 +166,8 @@ wrpcap('/tmp/foo.pcap', Ether()/IP()/UDP(sport=4556,dport=4556)/pkt) from scapy.contrib.bpv7 import * data = bytes.fromhex('9f88070000820282010282028202018202820201820018281a000f424085010100005823526561647920746f2067656e657261746520612033322d62797465207061796c6f6164ff') pkt = BundleV7(data) -print(repr(pkt)) pkt.show() +pkt.check_crc() assert pkt.primary.source == 'ipn:2.1' assert pkt.primary.destination == 'ipn:1.2' @@ -185,8 +181,8 @@ assert outdata == data from scapy.contrib.bpv7 import * data = bytes.fromhex('9f88070000820282010282028202018202820201820018281a000f424085070200004319012c85010100005823526561647920746f2067656e657261746520612033322d62797465207061796c6f6164ff') pkt = BundleV7(data) -print(repr(pkt)) pkt.show() +pkt.check_crc() assert pkt.primary.source == 'ipn:2.1' assert pkt.primary.destination == 'ipn:1.2' @@ -200,8 +196,8 @@ assert outdata == data from scapy.contrib.bpv7 import * data = bytes.fromhex('9f88070000820282010282028202018202820201820018281a000f4240850b0200005856810101018202820201828201078203008181820158403bdc69b3a34a2b5d3a8554368bd1e808f606219d2a10a846eae3886ae4ecc83c4ee550fdfb1cc636b904e2f1a73e303dcd4b6ccece003e95e8164dcc89a156e185010100005823526561647920746f2067656e657261746520612033322d62797465207061796c6f6164ff') pkt = BundleV7(data) -print(repr(pkt)) pkt.show() +pkt.check_crc() assert pkt.blocks[0].type_code == 11 #assert pkt.blocks[0].btsd.targets == [1] #assert pkt.blocks[0].btsd.context_id == 1 @@ -218,8 +214,8 @@ assert outdata == data from scapy.contrib.bpv7 import * data = bytes.fromhex('9f880700008201692f2f6473742f7376638201692f2f7372632f7376638201662f2f7372632f821b000000bd51281400001a000f42408501010000466568656c6c6fff') pkt = BundleV7(data) -print(repr(pkt)) pkt.show() +pkt.check_crc() assert pkt.primary.source == 'dtn://src/svc' assert pkt.primary.destination == 'dtn://dst/svc' From 7b8c96eddfee6af5f13dea5a7682291365a6eb79 Mon Sep 17 00:00:00 2001 From: Brian Sipos Date: Wed, 5 Aug 2026 22:49:16 -0400 Subject: [PATCH 6/6] Implement enum and flags field types over uint --- scapy/cbor/__init__.py | 4 ++++ scapy/cbor/cborfields.py | 45 ++++++++++++++++++++++++++++++++++++++++ scapy/contrib/bpv7.py | 17 ++++++++++----- scapy/libs/crc.py | 4 ++-- test/contrib/bpv7.uts | 30 ++++++++++++++++++++++++++- 5 files changed, 92 insertions(+), 8 deletions(-) diff --git a/scapy/cbor/__init__.py b/scapy/cbor/__init__.py index 7a9935c3e04..113b2a0d29c 100644 --- a/scapy/cbor/__init__.py +++ b/scapy/cbor/__init__.py @@ -62,6 +62,8 @@ CBORF_ARRAY_OF, CBORF_MAP, CBORF_SEMANTIC_TAG, + CBORF_UNSIGNED_ENUM, + CBORF_UNSIGNED_FLAGS, CBORF_optional, CBORF_CONDITIONAL, CBORF_PACKET, @@ -126,6 +128,8 @@ "CBORF_MAP", "CBORF_SEMANTIC_TAG", # Complex fields + "CBORF_UNSIGNED_ENUM", + "CBORF_UNSIGNED_FLAGS", "CBORF_optional", "CBORF_CONDITIONAL", "CBORF_PACKET", diff --git a/scapy/cbor/cborfields.py b/scapy/cbor/cborfields.py index d4d6128e570..fb7092c802b 100644 --- a/scapy/cbor/cborfields.py +++ b/scapy/cbor/cborfields.py @@ -551,6 +551,51 @@ def randval(self): # Structured CBOR Fields # ############################## +class CBORF_UNSIGNED_ENUM(CBORF_UNSIGNED_INTEGER): + """ + Display like EnumField, codec like CBORF + """ + def __init__(self, + name, # type: str + default, # type: Optional[int] + enum, # type: _EnumType[int] + ): + # type: (...) -> None + CBORF_UNSIGNED_INTEGER.__init__(self, name, default) + + self._enum = fields.EnumField(name, default, enum, "Q") + + def i2repr(self, pkt, x): + return self._enum.i2repr(pkt, x.val) + + def any2i(self, pkt, x): + x = x if isinstance(x, CBOR_Object) else self._enum.any2i(pkt, x) + return super().any2i(pkt, x) + + +class CBORF_UNSIGNED_FLAGS(CBORF_UNSIGNED_INTEGER): + """ + Display like FlagsField, codec like CBORF + """ + def __init__(self, + name, # type: str + default, # type: Optional[Union[int, FlagValue]] + size, # type: int + names, # type: Union[List[str], str, Dict[int, str]] + ): + # type: (...) -> None + CBORF_UNSIGNED_INTEGER.__init__(self, name, default) + + self._flags = fields.FlagsField(name, default, size, names) + + def i2repr(self, pkt, x): + return self._flags.i2repr(pkt, x.val) + + def any2i(self, pkt, x): + x = x if isinstance(x, CBOR_Object) else self._flags.any2i(pkt, x) + return super().any2i(pkt, x) + + class CBORF_SEQUENCE(CBORF_field[List[Any], List[Any]]): """ Unframed fixed sequence of named, typed fields. diff --git a/scapy/contrib/bpv7.py b/scapy/contrib/bpv7.py index 708ae779193..66ef3cf9d68 100644 --- a/scapy/contrib/bpv7.py +++ b/scapy/contrib/bpv7.py @@ -20,6 +20,7 @@ CBORF_field, CBORF_UNSIGNED_INTEGER, CBORF_INTEGER, CBORF_ARRAY, CBORF_BYTE_STRING, CBORF_CONDITIONAL, CBORF_SEQUENCE_OF, CBORF_PACKET, CBORF_BYTE_STRING_PACKET, + CBORF_UNSIGNED_ENUM, CBORF_UNSIGNED_FLAGS, CBORcodec_ARRAY, CBOR_UNSIGNED_INTEGER, CBOR_TEXT_STRING, CBOR_ARRAY, CBOR_Object ) @@ -294,6 +295,12 @@ class CrcInfo: } """Map from available CRC type to info.""" +def _enum_dict(cls: type[enum.IntEnum]) -> dict[int, str]: + return { + item.value: item.name for item in cls + } + + class AbstractBlock: ''' Represent an abstract block internal interface mixin. @@ -397,12 +404,12 @@ def is_fragment(self) -> bool: CBOR_root = CBORF_ARRAY( CBORF_UNSIGNED_INTEGER('version', default=7), - CBORF_UNSIGNED_INTEGER('bundle_flags', default=0), #FIXME FLAGS - CBORF_UNSIGNED_INTEGER('crc_type', default=CrcType.NONE), #FIXME ENUM + CBORF_UNSIGNED_FLAGS('bundle_flags', default=0, size=64, names=_enum_dict(Flag)), + CBORF_UNSIGNED_ENUM('crc_type', default=CrcType.NONE, enum=CrcType), BundleEidField('destination', default='dtn:none'), BundleEidField('source', default='dtn:none'), BundleEidField('report_to', default='dtn:none'), - CBORF_PACKET('create_ts', default=None, cls=BundleTimestamp), + CBORF_PACKET('create_ts', default=BundleTimestamp(), cls=BundleTimestamp), CBORF_UNSIGNED_INTEGER('lifetime', default=0), CBORF_CONDITIONAL( CBORF_UNSIGNED_INTEGER('fragment_offset', default=0), @@ -448,8 +455,8 @@ def btsd_class(self, data: bytes): CBOR_root = CBORF_ARRAY( CBORF_UNSIGNED_INTEGER('type_code', default=None), CBORF_UNSIGNED_INTEGER('block_num', default=None), - CBORF_UNSIGNED_INTEGER('block_flags', default=0), #FIXME FLAGS - CBORF_UNSIGNED_INTEGER('crc_type', default=CrcType.NONE), #FIXME ENUM + CBORF_UNSIGNED_FLAGS('block_flags', default=0, size=64, names=_enum_dict(Flag)), + CBORF_UNSIGNED_ENUM('crc_type', default=CrcType.NONE, enum=CrcType), CBORF_BYTE_STRING_PACKET('btsd', default=None, cls_cb=btsd_class), CBORF_CONDITIONAL( diff --git a/scapy/libs/crc.py b/scapy/libs/crc.py index 3838b73ce25..6a0a76cbc9e 100644 --- a/scapy/libs/crc.py +++ b/scapy/libs/crc.py @@ -13,9 +13,10 @@ "CRC", "CRCParam", "CRC_16", + "CRC_16_CCITT", + "CRC_16_X25", "CRC_32", "CRC_32C", - "CRC_16_CCITT", "CRC_32_AUTOSAR", "WELL_KNOWN_POLY", ] @@ -23,7 +24,6 @@ from functools import lru_cache from collections import defaultdict import itertools -from typing import Set, List, Tuple, Any # Taken from https://en.wikipedia.org/wiki/Cyclic_redundancy_check diff --git a/test/contrib/bpv7.uts b/test/contrib/bpv7.uts index 130ab874fbd..c99fc559d53 100644 --- a/test/contrib/bpv7.uts +++ b/test/contrib/bpv7.uts @@ -47,7 +47,7 @@ pkt = PrimaryBlock() pkt.show() outdata = bytes(pkt) print(outdata.hex()) -assert outdata == bytes.fromhex('89070000820100820100820100000000') +assert outdata == bytes.fromhex('8a070000820100820100820100820000000000') = Canonical payload encode from scapy.contrib.bpv7 import * @@ -223,3 +223,31 @@ pkt.clear_cache() outdata = bytes(pkt) print(outdata.hex()) assert outdata == data + + ++ BPv7 User API + += bundle flags +from scapy.contrib.bpv7 import * + +pkt = PrimaryBlock( + bundle_flags="PAYLOAD_ADMIN", +) +pkt.show() +assert pkt.bundle_flags == PrimaryBlock.Flag.PAYLOAD_ADMIN + +pkt = PrimaryBlock( + bundle_flags=0x004002, +) +pkt.show() +assert pkt.bundle_flags == ( + PrimaryBlock.Flag.PAYLOAD_ADMIN | PrimaryBlock.Flag.REQ_RECEPTION_REPORT +) + +pkt = PrimaryBlock( + bundle_flags="PAYLOAD_ADMIN+REQ_RECEPTION_REPORT", +) +pkt.show() +assert pkt.bundle_flags == ( + PrimaryBlock.Flag.PAYLOAD_ADMIN | PrimaryBlock.Flag.REQ_RECEPTION_REPORT +)