diff --git a/scapy/cbor/__init__.py b/scapy/cbor/__init__.py index dcec5d8ed5d..c969a2a1a05 100644 --- a/scapy/cbor/__init__.py +++ b/scapy/cbor/__init__.py @@ -47,6 +47,7 @@ from scapy.cbor.cborfields import ( CBORF_element, CBORF_field, + CBORF_ANY, CBORF_UNSIGNED_INTEGER, CBORF_NEGATIVE_INTEGER, CBORF_INTEGER, @@ -56,12 +57,19 @@ CBORF_NULL, CBORF_UNDEFINED, CBORF_FLOAT, + CBORF_SEQUENCE, + CBORF_SEQUENCE_OF, CBORF_ARRAY, CBORF_ARRAY_OF, + CBORF_ARRAY_INDEFINITE, CBORF_MAP, CBORF_SEMANTIC_TAG, + CBORF_UNSIGNED_ENUM, + CBORF_UNSIGNED_FLAGS, CBORF_optional, + CBORF_CONDITIONAL, CBORF_PACKET, + CBORF_BYTE_STRING_PACKET, ) __all__ = [ @@ -104,6 +112,7 @@ # Field base classes "CBORF_element", "CBORF_field", + "CBORF_ANY", # Scalar fields "CBORF_UNSIGNED_INTEGER", "CBORF_NEGATIVE_INTEGER", @@ -115,11 +124,18 @@ "CBORF_UNDEFINED", "CBORF_FLOAT", # Structured fields + "CBORF_SEQUENCE", + "CBORF_SEQUENCE_OF", "CBORF_ARRAY", "CBORF_ARRAY_OF", + "CBORF_ARRAY_INDEFINITE", "CBORF_MAP", "CBORF_SEMANTIC_TAG", # Complex fields + "CBORF_UNSIGNED_ENUM", + "CBORF_UNSIGNED_FLAGS", "CBORF_optional", + "CBORF_CONDITIONAL", "CBORF_PACKET", + "CBORF_BYTE_STRING_PACKET", ] diff --git a/scapy/cbor/cbor.py b/scapy/cbor/cbor.py index 1dcff4943f1..1ce68f0ee47 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 @@ -368,6 +369,10 @@ class CBOR_BYTE_STRING(CBOR_Object[bytes]): """CBOR byte string (major type 2)""" tag = CBOR_MajorTypes.BYTE_STRING + def __repr__(self): + # type: () -> str + return "<%s[h'%s']>" % (self.__class__.__name__, self.val.hex() if self.val else '') + class CBOR_TEXT_STRING(CBOR_Object[str]): """CBOR text string (major type 3)""" diff --git a/scapy/cbor/cborcodec.py b/scapy/cbor/cborcodec.py index 5cfa0ccbc92..5436a29cf37 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..0aefdcaf521 100644 --- a/scapy/cbor/cborfields.py +++ b/scapy/cbor/cborfields.py @@ -20,6 +20,7 @@ CBOR_NEGATIVE_INTEGER, CBOR_BYTE_STRING, CBOR_TEXT_STRING, + CBOR_ARRAY, CBOR_SEMANTIC_TAG, CBOR_FALSE, CBOR_TRUE, @@ -36,9 +37,11 @@ CBORcodec_NEGATIVE_INTEGER, CBORcodec_BYTE_STRING, CBORcodec_TEXT_STRING, + CBORcodec_ARRAY, CBORcodec_SIMPLE_AND_FLOAT, ) from scapy.base_classes import BasePacket +from scapy.error import log_runtime from scapy.volatile import ( RandChoice, RandFloat, @@ -47,7 +50,7 @@ RandField, ) -from scapy import packet +from scapy import packet, fields, config from typing import ( Any, @@ -84,6 +87,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] @@ -120,11 +127,13 @@ 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): # 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 +146,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] @@ -153,8 +162,13 @@ def extract_packet(self, c = cls(s, _underlayer=_underlayer) except CBORF_badsequence: c = packet.Raw(s, _underlayer=_underlayer) # type: ignore - cpad = c.getlayer(packet.Raw) + craw = c.getlayer(config.conf.raw_layer) + cpad = c.getlayer(config.conf.padding_layer) s = b"" + if craw is not None: + s = craw.load + if craw.underlayer: + del craw.underlayer.payload if cpad is not None: s = cpad.load if cpad.underlayer: @@ -163,7 +177,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 +199,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]] @@ -208,6 +222,17 @@ def copy(self): return copy.copy(self) +class CBORF_ANY(CBORF_field[CBOR_Object, CBOR_Object]): + """Represent any well-formed CBOR value, including recursion.""" + + def m2i(self, pkt, s): + # type: (CBOR_Packet, bytes) -> Tuple[CBOR_UNSIGNED_INTEGER, bytes] + return CBORcodec_Object.dec(s) # type: ignore + + def _encode(self, x): + # type: (Any) -> bytes + return CBORcodec_Object.enc(x) + ############################# # Simple CBOR Fields # ############################# @@ -218,6 +243,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 +270,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 +298,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 +341,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 +362,58 @@ 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: + log_runtime.exception("Failed to decode byte string content to %s", pkt_cls) + sub = packet.Raw(obj.val, _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 +571,118 @@ 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. + 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). @@ -504,6 +701,9 @@ class MyCBOR(CBOR_Packet): CBOR_tag = CBOR_MajorTypes.ARRAY holds_packets = 1 + encode_indefinite = False + """Set to true to encode using indefinite length.""" + def __init__(self, *seq, **kwargs): # type: (*Any, **Any) -> None # The array itself is a structural field without its own named slot on @@ -529,6 +729,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 +755,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 +775,23 @@ 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) + if self.encode_indefinite: + return ( + CBOR_encode_head(int(CBOR_MajorTypes.ARRAY), None) + items + + CBOR_encode_head(int(CBOR_MajorTypes.SIMPLE_AND_FLOAT), None) + ) + else: + return CBOR_encode_head(int(CBOR_MajorTypes.ARRAY), len(parts)) + items + + +class CBORF_ARRAY_INDEFINITE(CBORF_ARRAY): + """A field to act as an array but to always encode to indefinte-length.""" + + encode_indefinite = True _ARRAY_T = Union[ @@ -574,6 +802,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). @@ -630,9 +941,13 @@ def m2i(self, pkt, s): lst.append(c) return lst, s + def _encode(self, x): + # type: (Any) -> bytes + return self.build(x) + 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) @@ -862,6 +1177,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/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..7d677e1c637 --- /dev/null +++ b/scapy/contrib/bpv7.py @@ -0,0 +1,603 @@ +# 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 + +from dataclasses import dataclass +import datetime +import enum +import struct +from typing import Optional, Union, cast, ClassVar, Callable +from scapy import config, volatile +from scapy.packet import Packet +from scapy.error import log_runtime +from scapy.cbor.cborcodec import CBOR_decode_head, CBOR_MajorTypes +from scapy.cbor import ( + CBORF_field, + CBORF_ANY, + CBORF_UNSIGNED_INTEGER, + CBORF_INTEGER, + CBORF_ARRAY, + CBORF_ARRAY_OF, + CBORF_ARRAY_INDEFINITE, + CBORF_BYTE_STRING, + CBORF_CONDITIONAL, + CBORF_SEQUENCE, + 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, +) +from scapy.cborpacket import ( + CBOR_Packet, +) +from scapy.libs.crc import CRC, CRC_16_X25, CRC_32C + + +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): + """Handled EID scheme names and values.""" + + dtn = 1 + ipn = 2 + + +_DTN_WELL_KNOWN_SSP = { + 0: "none", +} +"""Compressed SSP encoding.""" + + +@dataclass +class EidStruct: + """ + Internal state for the :class:`BundleEidField` class. + """ + + 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) + + 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 + 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("Invalid scheme state") + + 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("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 + 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("Invalid scheme state") + + 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("Invalid scheme state") + + 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 + + +@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.""" + + +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. + + .. py:attribute:: crc_type_name + The name of the CRC-type field. + .. py:attribute:: crc_value_name + The name of the CRC-value field. + """ + + _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 has_crc(self): + """ + Match the signature for CBORF_CONDITIONAL on the CRC Value field. + """ + crc_type = self.getfieldval(self._crc_type_name).val + return crc_type != CrcType.NONE.value + + 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. + """ + crc_type = getattr(self, self._crc_type_name).val + crc_value = getattr(self, self._crc_value_name) + if crc_type == CrcType.NONE: + # there should not be a value + crc_value = None + 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. + """ + + crc_type = getattr(self, self._crc_type_name).val + crc_value = getattr(self, self._crc_value_name) or b"" + if crc_type == CrcType.NONE: + expect = b"" + valid = not crc_value + else: + 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.val == expect + + if not valid: + log_runtime.warning( + "CRC check failed! Expected %s got %s" % (expect.hex(), crc_value.hex()) + ) + + return valid + + +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_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=BundleTimestamp(), 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=AbstractBlock.has_crc + ), + ) + + 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. + """ + + @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. """ + + _reg_types: ClassVar[dict[int, Packet]] = {} + """ Known block types. """ + + @classmethod + def register_type(cls, type_code: int) -> Callable[[type[Packet]], type[Packet]]: + """ + Decorator to register a BTSD decoder for a specific block type + """ + + def reg(pkt_cls: type[Packet]) -> type[Packet]: + cls._reg_types[type_code] = pkt_cls + return pkt_cls + + return reg + + def btsd_class(self, data: bytes): + cls = None + type_code = self.getfieldval("type_code") + if type_code is not None: + try: + cls = self._reg_types[type_code.val] + except KeyError: + pass + return cls + + CBOR_root = CBORF_ARRAY( + CBORF_UNSIGNED_INTEGER("type_code", default=None), + CBORF_UNSIGNED_INTEGER("block_num", default=None), + 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( + CBORF_BYTE_STRING("crc_value", default=None), cond=AbstractBlock.has_crc + ), + ) + + def extract_padding(self, s): + # type: (bytes) -> Tuple[bytes, Optional[bytes]] + return None, s + + def self_build(self): + # type: () -> bytes + + self.update_crc(keep_existing=True) + return super().self_build() + + +@CanonicalBlock.register_type(6) +class PreviousNodeBlock(CBOR_Packet): + """Block data content from Section 4.4.1 of RFC 9171.""" + + CBOR_root = BundleEidField("node", default=None) + + +@CanonicalBlock.register_type(7) +class BundleAgeBlock(CBOR_Packet): + """Block data content from Section 4.4.2 of RFC 9171.""" + + CBOR_root = CBORF_UNSIGNED_INTEGER("age", default=None) + + +@CanonicalBlock.register_type(10) +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), + ) + + +class BpsecKeyValPair(CBOR_Packet): + CBOR_root = CBORF_ARRAY( + CBORF_UNSIGNED_INTEGER("key", default=None), + CBORF_ANY("val", default=None), + ) + + +#@CanonicalBlock.register_type(11) +#@CanonicalBlock.register_type(12) +class AbstractSecurityBock(CBOR_Packet): + """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. """ + + CBOR_root = CBORF_SEQUENCE( + CBORF_ARRAY_OF( + "targets", [], cls=CBORF_UNSIGNED_INTEGER("blk_num", default=None) + ), + CBORF_INTEGER("context_id", default=None), + CBORF_UNSIGNED_FLAGS("flags", default=0, size=64, names=_enum_dict(Flag)), + BundleEidField("source", default=None), + CBORF_CONDITIONAL( + CBORF_ARRAY_OF("parameters", [], cls=CBORF_PACKET('kvp', default=None, cls=BpsecKeyValPair)), + cond=lambda pkt: pkt.flags.val & AbstractSecurityBock.Flag.PARAMETERS, + ), + # one packet in this list per target + CBORF_ARRAY_OF( + "tgt_results", [], cls=CBORF_ARRAY_OF("results", [], cls=CBORF_PACKET('kvp', default=None, cls=BpsecKeyValPair)) + ), + ) + + +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): + """ + 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 + print("new block") + return CanonicalBlock + + CBOR_root = CBORF_ARRAY_INDEFINITE( + CBORF_PACKET("primary", default=PrimaryBlock(), cls=PrimaryBlock), + CBORF_SEQUENCE_OF("blocks", default=[], 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) + + +config.conf.debug_dissector = True diff --git a/scapy/fields.py b/scapy/fields.py index 3e158d8520d..eb671133023 100644 --- a/scapy/fields.py +++ b/scapy/fields.py @@ -440,7 +440,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/scapy/libs/crc.py b/scapy/libs/crc.py index f43e3e7e700..6a0a76cbc9e 100644 --- a/scapy/libs/crc.py +++ b/scapy/libs/crc.py @@ -13,8 +13,10 @@ "CRC", "CRCParam", "CRC_16", - "CRC_32", "CRC_16_CCITT", + "CRC_16_X25", + "CRC_32", + "CRC_32C", "CRC_32_AUTOSAR", "WELL_KNOWN_POLY", ] @@ -22,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 @@ -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 new file mode 100644 index 00000000000..a76ccf4b74a --- /dev/null +++ b/test/contrib/bpv7.uts @@ -0,0 +1,283 @@ +% Bundle Protocol Version 7 test campaign + ++ Syntax check += Import the BPv7 layer +from scapy.contrib.bpv7 import * + ++ EID CODEC + += EID encode + +from scapy.cbor import * +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') + + ++ Block CODEC + += Primary default +from scapy.contrib.bpv7 import * +pkt = PrimaryBlock() +pkt.show() +outdata = bytes(pkt) +print(outdata.hex()) +assert outdata == bytes.fromhex('8a070000820100820100820100820000000000') + += 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()) +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) +pkt.show() +assert pkt.check_crc() +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) +pkt.show() +assert pkt.check_crc() +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) +pkt.show() +assert pkt.check_crc() +assert len(pkt.blocks) == 2 +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) +pkt.show() +assert pkt.check_crc() +assert pkt.primary.source == 'dtn://src/svc' +assert pkt.primary.destination == 'dtn://dst/svc' +assert len(pkt.blocks) == 1 +assert pkt.blocks[0].type_code == 1 + +pkt.clear_cache() +outdata = bytes(pkt) +print(outdata.hex()) +assert outdata == data + + += decoding example from Appendix A.1 of draft-dtn-bpsec-cose + +from scapy.contrib.bpv7 import * +data = bytes.fromhex( + "9f890700028201692f2f6473742f7376638201692f2f7372632f7376638201662f2f" + "7372632f821b000000bd51281400001a000f42404482a081c9850b03000058608101" + "03018201662f2f7372632f818205a2000120018181821158458443a10106a1044a45" + "78616d706c65412e31f65830ec8260a38a1a00fef2cd4aae063f50f01c5645e84c6c" + "4893ca895eed44ef60a5f50f9adf5cc5654499b881e5896378058601010002466568" + "656c6c6f444ec359d2ff" +) +pkt = BundleV7(data) +pkt.show() +assert pkt.check_crc() +assert pkt.primary.source == 'dtn://src/svc' +assert pkt.primary.destination == 'dtn://dst/svc' +assert len(pkt.blocks) == 2 +assert pkt.blocks[0].type_code == 11 +print(pkt.blocks[0].btsd.show()) +assert pkt.blocks[1].type_code == 1 + +#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 +)