From 669646406a09579b5ed4532f172f6b27e7e6e5e2 Mon Sep 17 00:00:00 2001 From: Nils Weiss Date: Sun, 19 Jul 2026 20:54:10 +0200 Subject: [PATCH 01/11] asn1: prepare fields for pluggable codecs Dispatch tagging via codec stems and pass codec kwargs through ASN1F encode/decode so additional codecs can register without hardcoding BER in every field path. AI-Assisted: yes (Cursor) --- scapy/asn1/ber.py | 13 +- scapy/asn1fields.py | 336 +++++++++++++++++++++---------- test/scapy/layers/asn1.uts | 40 ++++ test/scapy/layers/ber_codec.py | 275 +++++++++++++++++++++++++ test/scapy/layers/ber_packets.py | 150 ++++++++++++++ 5 files changed, 709 insertions(+), 105 deletions(-) create mode 100644 test/scapy/layers/ber_codec.py create mode 100644 test/scapy/layers/ber_packets.py diff --git a/scapy/asn1/ber.py b/scapy/asn1/ber.py index b5ffc4252b7..75887d2b938 100644 --- a/scapy/asn1/ber.py +++ b/scapy/asn1/ber.py @@ -11,6 +11,7 @@ # Good read: https://luca.ntop.org/Teaching/Appunti/asn1.html +from scapy.config import conf from scapy.error import warning from scapy.compat import chb, orb, bytes_encode from scapy.utils import binrepr, inet_aton, inet_ntoa @@ -266,7 +267,9 @@ def BER_tagging_enc(s, implicit_tag=None, explicit_tag=None): if implicit_tag is not None: s = BER_id_enc(implicit_tag) + s[1:] elif explicit_tag is not None: - s = BER_id_enc(explicit_tag) + BER_len_enc(len(s)) + s + s = BER_id_enc(explicit_tag) + BER_len_enc( + len(s), size=conf.ASN1_default_long_size, + ) + s return s # [ BER classes ] # @@ -374,6 +377,7 @@ def dec(cls, context=None, # type: Optional[Type[ASN1_Class]] safe=False, # type: bool _depth=0, # type: int + **_kwargs # type: Any ): # type: (...) -> Tuple[Union[_ASN1_ERROR, ASN1_Object[_K]], bytes] if _depth > MAX_BER_DEPTH: @@ -400,6 +404,7 @@ def safedec(cls, s, # type: bytes context=None, # type: Optional[Type[ASN1_Class]] _depth=0, # type: int + **_kwargs # type: Any ): # type: (...) -> Tuple[Union[_ASN1_ERROR, ASN1_Object[_K]], bytes] return cls.dec(s, context, safe=True, _depth=_depth) @@ -418,6 +423,10 @@ def enc(cls, s, size_len=0): ASN1_Codecs.BER.register_stem(BERcodec_Object) +BERcodec_Object.tagging_enc = staticmethod(BER_tagging_enc) +BERcodec_Object.tagging_dec = staticmethod(BER_tagging_dec) +BERcodec_Object.skip_tagging = False + ########################## # BERcodec objects # @@ -651,6 +660,8 @@ def enc(cls, _ll, size_len=0): ll = _ll else: ll = b"".join(x.enc(cls.codec) for x in _ll) + if not size_len: + size_len = conf.ASN1_default_long_size return chb(int(cls.tag)) + BER_len_enc(len(ll), size=size_len) + ll @classmethod diff --git a/scapy/asn1fields.py b/scapy/asn1fields.py index 7895d7aa1bb..19aa85907bc 100644 --- a/scapy/asn1fields.py +++ b/scapy/asn1fields.py @@ -17,6 +17,7 @@ ASN1_BOOLEAN, ASN1_Class, ASN1_Class_UNIVERSAL, + ASN1_Decoding_Error, ASN1_Error, ASN1_INTEGER, ASN1_NULL, @@ -119,6 +120,50 @@ def register_owner(self, cls): # type: (Type[ASN1_Packet]) -> None self.owners.append(cls) + def _apply_diff_tag(self, diff_tag): + # type: (Optional[int]) -> None + # this implies that flexible_tag was True + if diff_tag is not None: + if self.implicit_tag is not None: + self.implicit_tag = diff_tag + elif self.explicit_tag is not None: + self.explicit_tag = diff_tag + + def _tagging_dec(self, pkt, s, **kwargs): + # type: (ASN1_Packet, bytes, **Any) -> Tuple[Optional[int], bytes] + stem = pkt.ASN1_codec.get_stem() + if getattr(stem, "skip_tagging", False): + return None, s + fn = getattr(stem, "tagging_dec", None) + if fn is None: + return BER_tagging_dec(s, **kwargs) + return fn(s, **kwargs) + + def _tagging_enc(self, pkt, s, **kwargs): + # type: (ASN1_Packet, bytes, **Any) -> bytes + stem = pkt.ASN1_codec.get_stem() + if getattr(stem, "skip_tagging", False): + return s + fn = getattr(stem, "tagging_enc", None) + if fn is None: + return BER_tagging_enc(s, **kwargs) + return fn(s, **kwargs) + + def _codec_kwargs(self, size_len=None): + # type: (Optional[int]) -> Dict[str, Any] + return { + "size_len": self.size_len if size_len is None else size_len, + } + + def _encode_item(self, pkt, item): + # type: (ASN1_Packet, Any) -> bytes + if isinstance(item, ASN1_Object): + return item.enc(pkt.ASN1_codec) + if hasattr(item, "self_build"): + return cast("ASN1_Packet", item).self_build() + codec = self.ASN1_tag.get_codec(pkt.ASN1_codec) + return codec.enc(item, **self._codec_kwargs()) + def i2repr(self, pkt, x): # type: (ASN1_Packet, _I) -> str return repr(x) @@ -141,22 +186,24 @@ def m2i(self, pkt, s): as expected or not. Noticeably, input methods from cert.py expect certain exceptions to be raised. Hence default flexible_tag is False. """ - diff_tag, s = BER_tagging_dec(s, hidden_tag=self.ASN1_tag, - implicit_tag=self.implicit_tag, - explicit_tag=self.explicit_tag, - safe=self.flexible_tag, - _fname=self.name) - if diff_tag is not None: - # this implies that flexible_tag was True - if self.implicit_tag is not None: - self.implicit_tag = diff_tag - elif self.explicit_tag is not None: - self.explicit_tag = diff_tag + diff_tag, s = self._tagging_dec( + pkt, s, + hidden_tag=self.ASN1_tag, + implicit_tag=self.implicit_tag, + explicit_tag=self.explicit_tag, + safe=self.flexible_tag, + _fname=self.name, + ) + self._apply_diff_tag(diff_tag) codec = self.ASN1_tag.get_codec(pkt.ASN1_codec) if self.flexible_tag: - return codec.safedec(s, context=self.context) # type: ignore + return codec.safedec( + s, context=self.context, **self._codec_kwargs() + ) # type: ignore else: - return codec.dec(s, context=self.context) # type: ignore + return codec.dec( + s, context=self.context, **self._codec_kwargs() + ) # type: ignore def i2m(self, pkt, x): # type: (ASN1_Packet, Union[bytes, _I, _A]) -> bytes @@ -171,10 +218,14 @@ def i2m(self, pkt, x): else: raise ASN1_Error("Encoding Error: got %r instead of an %r for field [%s]" % (x, self.ASN1_tag, self.name)) # noqa: E501 else: - s = self.ASN1_tag.get_codec(pkt.ASN1_codec).enc(x, size_len=self.size_len) - return BER_tagging_enc(s, - implicit_tag=self.implicit_tag, - explicit_tag=self.explicit_tag) + s = self.ASN1_tag.get_codec(pkt.ASN1_codec).enc( + x, **self._codec_kwargs() + ) + return self._tagging_enc( + pkt, s, + implicit_tag=self.implicit_tag, + explicit_tag=self.explicit_tag, + ) def any2i(self, pkt, x): # type: (ASN1_Packet, Any) -> _I @@ -461,6 +512,45 @@ def get_fields_list(self): return reduce(lambda x, y: x + y.get_fields_list(), self.seq, []) + def _apply_tagging_dec(self, s, pkt): + # type: (bytes, Any) -> bytes + diff_tag, s = self._tagging_dec( + pkt, s, + hidden_tag=self.ASN1_tag, + implicit_tag=self.implicit_tag, + explicit_tag=self.explicit_tag, + safe=self.flexible_tag, + _fname=pkt.name, + ) + self._apply_diff_tag(diff_tag) + return s + + def _dissect_sequence_children(self, pkt, s): + # type: (Any, bytes) -> bytes + if len(s) == 0: + for obj in self.seq: + obj.set_val(pkt, None) + return s + for obj in self.seq: + try: + s = obj.dissect(pkt, s) + except ASN1F_badsequence: + break + return s + + def _m2i_ber(self, pkt, s): + # type: (Any, bytes) -> Tuple[Any, bytes] + s = self._apply_tagging_dec(s, pkt) + codec = self.ASN1_tag.get_codec(pkt.ASN1_codec) + i, s, remain = codec.check_type_check_len(s) + s = self._dissect_sequence_children(pkt, s) + if len(s) > 0: + raise BER_Decoding_Error( + "unexpected remainder in %s" % pkt.name, + remaining=s, + ) + return [], remain + def m2i(self, pkt, s): # type: (Any, bytes) -> Tuple[Any, bytes] """ @@ -471,33 +561,7 @@ def m2i(self, pkt, s): Thus m2i returns an empty list (along with the proper remainder). It is discarded by dissect() and should not be missed elsewhere. """ - diff_tag, s = BER_tagging_dec(s, hidden_tag=self.ASN1_tag, - implicit_tag=self.implicit_tag, - explicit_tag=self.explicit_tag, - safe=self.flexible_tag, - _fname=pkt.name) - if diff_tag is not None: - if self.implicit_tag is not None: - self.implicit_tag = diff_tag - elif self.explicit_tag is not None: - self.explicit_tag = diff_tag - codec = self.ASN1_tag.get_codec(pkt.ASN1_codec) - i, s, remain = codec.check_type_check_len(s) - if len(s) == 0: - for obj in self.seq: - obj.set_val(pkt, None) - else: - for obj in self.seq: - try: - s = obj.dissect(pkt, s) - except ASN1F_badsequence: - break - if len(s) > 0: - raise BER_Decoding_Error( - "unexpected remainder in %s" % pkt.name, - remaining=s, - ) - return [], remain + return self._m2i_ber(pkt, s) def dissect(self, pkt, s): # type: (Any, bytes) -> bytes @@ -572,15 +636,14 @@ def m2i(self, s, # type: bytes ): # type: (...) -> Tuple[List[Any], bytes] - diff_tag, s = BER_tagging_dec(s, hidden_tag=self.ASN1_tag, - implicit_tag=self.implicit_tag, - explicit_tag=self.explicit_tag, - safe=self.flexible_tag) - if diff_tag is not None: - if self.implicit_tag is not None: - self.implicit_tag = diff_tag - elif self.explicit_tag is not None: - self.explicit_tag = diff_tag + diff_tag, s = self._tagging_dec( + pkt, s, + hidden_tag=self.ASN1_tag, + implicit_tag=self.implicit_tag, + explicit_tag=self.explicit_tag, + safe=self.flexible_tag, + ) + self._apply_diff_tag(diff_tag) codec = self.ASN1_tag.get_codec(pkt.ASN1_codec) i, s, remain = codec.check_type_check_len(s) lst = [] @@ -603,8 +666,10 @@ def build(self, pkt): s = cast(Union[List[_SEQ_T], bytes], val) elif val is None: s = b"" - else: + elif self.holds_packets: s = b"".join(bytes(i) for i in val) + else: + s = b"".join(self.fld._encode_item(pkt, i) for i in val) return self.i2m(pkt, s) def i2repr(self, pkt, x): @@ -663,7 +728,7 @@ def m2i(self, pkt, s): # type: (ASN1_Packet, bytes) -> Tuple[Any, bytes] try: return self._field.m2i(pkt, s) - except (ASN1_Error, ASN1F_badsequence, BER_Decoding_Error): + except (ASN1_Error, ASN1F_badsequence, ASN1_Decoding_Error): # ASN1_Error may be raised by ASN1F_CHOICE return None, s @@ -671,7 +736,7 @@ def dissect(self, pkt, s): # type: (ASN1_Packet, bytes) -> bytes try: return self._field.dissect(pkt, s) - except (ASN1_Error, ASN1F_badsequence, BER_Decoding_Error): + except (ASN1_Error, ASN1F_badsequence, ASN1_Decoding_Error): self._field.set_val(pkt, None) return s @@ -689,6 +754,23 @@ def i2repr(self, pkt, x): # type: (ASN1_Packet, Any) -> str return self._field.i2repr(pkt, x) + def set_val(self, pkt, val): + # type: (ASN1_Packet, Any) -> None + self._field.set_val(pkt, val) + + def set_absent(self, pkt): + # type: (ASN1_Packet) -> None + self.set_val(pkt, None) + + def is_empty(self, pkt): + # type: (ASN1_Packet) -> bool + val = getattr(pkt, self._field.name, None) + if val is None: + return True + if getattr(self._field, "islist", 0) and val == []: + return True + return False + class ASN1F_omit(ASN1F_field[None, None]): """ @@ -731,6 +813,8 @@ def __init__(self, name, default, *args, **kwargs): self.default = default self.current_choice = None self.choices = {} # type: Dict[int, _CHOICE_T] + self.choice_order = [] # type: List[int] + self.choice_list = [] # type: List[_CHOICE_T] self.pktchoices = {} for p in args: if hasattr(p, "ASN1_root"): @@ -738,21 +822,63 @@ def __init__(self, name, default, *args, **kwargs): # should be ASN1_Packet if hasattr(p.ASN1_root, "choices"): root = cast(ASN1F_CHOICE, p.ASN1_root) - for k, v in root.choices.items(): - # ASN1F_CHOICE recursion - self.choices[k] = v + for k in root.choice_order: + self._register_choice(k, root.choices[k]) else: - self.choices[p.ASN1_root.network_tag] = p + self._register_choice(p.ASN1_root.network_tag, p) elif hasattr(p, "ASN1_tag"): if isinstance(p, type): # should be ASN1F_field class - self.choices[int(p.ASN1_tag)] = p + self._register_choice(int(p.ASN1_tag), p) else: # should be ASN1F_field instance - self.choices[p.network_tag] = p - self.pktchoices[hash(p.cls)] = (p.implicit_tag, p.explicit_tag) # noqa: E501 + self._register_choice(p.network_tag, p) + if hasattr(p, "cls"): + self.pktchoices[hash(p.cls)] = (p.implicit_tag, p.explicit_tag) # noqa: E501 else: raise ASN1_Error("ASN1F_CHOICE: no tag found for one field") + self._tag_to_index = { + tag: idx for idx, tag in enumerate(self.choice_order) + } + + def _register_choice(self, tag, choice): + # type: (int, _CHOICE_T) -> None + self.choices[tag] = choice + self.choice_order.append(tag) + self.choice_list.append(choice) + + def _dissect_choice_payload(self, pkt, choice, payload): + # type: (ASN1_Packet, _CHOICE_T, bytes) -> Tuple[ASN1_Object[Any], bytes] + if hasattr(choice, "ASN1_root"): + return self.extract_packet(choice, payload, _underlayer=pkt) # type: ignore + if isinstance(choice, type): + return choice(self.name, b"").m2i(pkt, payload) + return choice.m2i(pkt, payload) + + def _m2i_ber(self, pkt, s): + # type: (ASN1_Packet, bytes) -> Tuple[ASN1_Object[Any], bytes] + _, s = self._tagging_dec( + pkt, s, + hidden_tag=self.ASN1_tag, + explicit_tag=self.explicit_tag, + ) + tag, _ = BER_id_dec(s) + return self._m2i_tagged(pkt, tag, s) + + def _m2i_tagged(self, pkt, tag, payload): + # type: (ASN1_Packet, int, bytes) -> Tuple[ASN1_Object[Any], bytes] + if tag in self.choices: + choice = self.choices[tag] + elif self.flexible_tag: + choice = ASN1F_field + else: + raise ASN1_Error( + "ASN1F_CHOICE: unexpected field in '%s' " + "(tag %s not in possible tags %s)" % ( + self.name, tag, list(self.choices.keys()) + ) + ) + return self._dissect_choice_payload(pkt, choice, payload) def m2i(self, pkt, s): # type: (ASN1_Packet, bytes) -> Tuple[ASN1_Object[Any], bytes] @@ -762,42 +888,43 @@ def m2i(self, pkt, s): """ if len(s) == 0: raise ASN1_Error("ASN1F_CHOICE: got empty string") - _, s = BER_tagging_dec(s, hidden_tag=self.ASN1_tag, - explicit_tag=self.explicit_tag) - tag, _ = BER_id_dec(s) - if tag in self.choices: - choice = self.choices[tag] - else: - if self.flexible_tag: - choice = ASN1F_field - else: - raise ASN1_Error( - "ASN1F_CHOICE: unexpected field in '%s' " - "(tag %s not in possible tags %s)" % ( - self.name, tag, list(self.choices.keys()) - ) - ) - if hasattr(choice, "ASN1_root"): - # we don't want to import ASN1_Packet in this module... - return self.extract_packet(choice, s, _underlayer=pkt) # type: ignore - elif isinstance(choice, type): - return choice(self.name, b"").m2i(pkt, s) - else: - # XXX check properly if this is an ASN1F_PACKET - return choice.m2i(pkt, s) + return self._m2i_ber(pkt, s) + + def _choice_tag_for(self, x): + # type: (Any) -> Optional[int] + index = self._choice_index_for(x) + return None if index is None else self.choice_order[index] + + def _choice_index_for(self, x): + # type: (Any) -> Optional[int] + for index, choice in enumerate(self.choice_list): + if isinstance(choice, type) and hasattr(choice, "ASN1_root"): + if isinstance(x, choice): + return index + elif hasattr(choice, "ASN1_tag"): + if isinstance(x, ASN1_Object) and x.tag == choice.ASN1_tag: + return index + return None def i2m(self, pkt, x): # type: (ASN1_Packet, Any) -> bytes if x is None: s = b"" else: - s = bytes(x) + if isinstance(x, ASN1_Object): + s = x.enc(pkt.ASN1_codec) + elif hasattr(x, "self_build"): + s = cast("ASN1_Packet", x).self_build() + else: + s = bytes(x) if hash(type(x)) in self.pktchoices: imp, exp = self.pktchoices[hash(type(x))] - s = BER_tagging_enc(s, - implicit_tag=imp, - explicit_tag=exp) - return BER_tagging_enc(s, explicit_tag=self.explicit_tag) + s = self._tagging_enc( + pkt, s, + implicit_tag=imp, + explicit_tag=exp, + ) + return self._tagging_enc(pkt, s, explicit_tag=self.explicit_tag) def randval(self): # type: () -> RandChoice @@ -849,16 +976,15 @@ def m2i(self, pkt, s): if not hasattr(cls, "ASN1_root"): # A normal Packet (!= ASN1) return self.extract_packet(cls, s, _underlayer=pkt) - diff_tag, s = BER_tagging_dec(s, hidden_tag=cls.ASN1_root.ASN1_tag, # noqa: E501 - implicit_tag=self.implicit_tag, - explicit_tag=self.explicit_tag, - safe=self.flexible_tag, - _fname=self.name) - if diff_tag is not None: - if self.implicit_tag is not None: - self.implicit_tag = diff_tag - elif self.explicit_tag is not None: - self.explicit_tag = diff_tag + diff_tag, s = self._tagging_dec( + pkt, s, + hidden_tag=cls.ASN1_root.ASN1_tag, # noqa: E501 + implicit_tag=self.implicit_tag, + explicit_tag=self.explicit_tag, + safe=self.flexible_tag, + _fname=self.name, + ) + self._apply_diff_tag(diff_tag) if not s: return None, s return self.extract_packet(cls, s, _underlayer=pkt) @@ -882,9 +1008,11 @@ def i2m(self, if not hasattr(x, "ASN1_root"): # A normal Packet (!= ASN1) return s - return BER_tagging_enc(s, - implicit_tag=self.implicit_tag, - explicit_tag=self.explicit_tag) + return self._tagging_enc( + pkt, s, + implicit_tag=self.implicit_tag, + explicit_tag=self.explicit_tag, + ) def any2i(self, pkt, # type: ASN1_Packet diff --git a/test/scapy/layers/asn1.uts b/test/scapy/layers/asn1.uts index 9fa0bad0f44..e07e18e8165 100644 --- a/test/scapy/layers/asn1.uts +++ b/test/scapy/layers/asn1.uts @@ -101,3 +101,43 @@ ASN1_UTC_TIME(datetime(2020, 12, 31)).val == "201231000000" ASN1_UTC_TIME(datetime(2020, 12, 31, tzinfo=timezone.utc)).val == "201231000000Z" = UTC datetime construction (offset) ASN1_UTC_TIME(datetime(2020, 12, 31, tzinfo=timezone(timedelta(hours=-23, minutes=-59)))).val == "201231000000-2359" + ++ ASN.1 BER packets and fields += BER field explicit tag +__import__('test.scapy.layers.ber_packets', fromlist=['check_ber_field_explicit_tag']).check_ber_field_explicit_tag() += BER field fixed size +__import__('test.scapy.layers.ber_packets', fromlist=['check_ber_field_fixed_size']).check_ber_field_fixed_size() += BER field optional +__import__('test.scapy.layers.ber_packets', fromlist=['check_ber_field_optional']).check_ber_field_optional() += BER field sequence of +__import__('test.scapy.layers.ber_packets', fromlist=['check_ber_field_sequence_of']).check_ber_field_sequence_of() += BER field choice +__import__('test.scapy.layers.ber_packets', fromlist=['check_ber_field_choice']).check_ber_field_choice() += BER packet record +__import__('test.scapy.layers.ber_packets', fromlist=['check_ber_packet_record']).check_ber_packet_record() + ++ ASN.1 BER codec += BER error formatting +__import__('test.scapy.layers.ber_codec', fromlist=['check_ber_error_str']).check_ber_error_str() += BER length encoding +__import__('test.scapy.layers.ber_codec', fromlist=['check_ber_len_enc_dec']).check_ber_len_enc_dec() += BER number encoding +__import__('test.scapy.layers.ber_codec', fromlist=['check_ber_num_enc_dec']).check_ber_num_enc_dec() += BER identifier encoding +__import__('test.scapy.layers.ber_codec', fromlist=['check_ber_id_enc_dec']).check_ber_id_enc_dec() += BER tagging +__import__('test.scapy.layers.ber_codec', fromlist=['check_ber_tagging']).check_ber_tagging() += BER integer codec +__import__('test.scapy.layers.ber_codec', fromlist=['check_ber_integer']).check_ber_integer() += BER bit string codec +__import__('test.scapy.layers.ber_codec', fromlist=['check_ber_bit_string']).check_ber_bit_string() += BER string and null codec +__import__('test.scapy.layers.ber_codec', fromlist=['check_ber_string_and_null']).check_ber_string_and_null() += BER OID codec +__import__('test.scapy.layers.ber_codec', fromlist=['check_ber_oid']).check_ber_oid() += BER sequence and set codec +__import__('test.scapy.layers.ber_codec', fromlist=['check_ber_sequence_and_set']).check_ber_sequence_and_set() += BER IP address codec +__import__('test.scapy.layers.ber_codec', fromlist=['check_ber_ipaddress']).check_ber_ipaddress() += BER object dispatch +__import__('test.scapy.layers.ber_codec', fromlist=['check_ber_object_dispatch']).check_ber_object_dispatch() diff --git a/test/scapy/layers/ber_codec.py b/test/scapy/layers/ber_codec.py new file mode 100644 index 00000000000..e6939f7a27e --- /dev/null +++ b/test/scapy/layers/ber_codec.py @@ -0,0 +1,275 @@ +# SPDX-License-Identifier: GPL-2.0-only +# This file is part of Scapy +# See https://scapy.net/ for more information + +""" +BER codec and helper coverage tests. +""" + +from typing import Any + + +def _raises(exc, func): + # type: (type, Any) -> None + try: + func() + except exc: + return + raise AssertionError("Expected %s" % exc.__name__) + + +from scapy.asn1.asn1 import ( + ASN1_Class_UNIVERSAL, + ASN1_DECODING_ERROR, + ASN1_INTEGER, + ASN1_Object, +) +from scapy.asn1.ber import ( + BER_BadTag_Decoding_Error, + BER_Decoding_Error, + BER_Encoding_Error, + BER_Exception, + BER_id_dec, + BER_id_enc, + BER_len_dec, + BER_len_enc, + BER_num_dec, + BER_num_enc, + BER_tagging_dec, + BER_tagging_enc, + BERcodec_BIT_STRING, + BERcodec_INTEGER, + BERcodec_IPADDRESS, + BERcodec_NULL, + BERcodec_Object, + BERcodec_OID, + BERcodec_SEQUENCE, + BERcodec_SET, + BERcodec_STRING, +) +from scapy.config import conf + + +def check_ber_error_str(): + # type: () -> None + obj = ASN1_INTEGER(1) + enc_err = BER_Encoding_Error("enc", encoded=obj, remaining=b"rest") + assert "Already encoded" in str(enc_err) + enc_err2 = BER_Encoding_Error("enc", encoded="raw", remaining=b"") + assert "raw" in str(enc_err2) + + dec_err = BER_Decoding_Error("dec", decoded=obj, remaining=b"tail") + assert "Already decoded" in str(dec_err) + dec_err2 = BER_Decoding_Error("dec", decoded=[1], remaining=b"") + assert "[1]" in str(dec_err2) + + +def check_ber_len_enc_dec(): + # type: () -> None + for value in [0, 1, 127, 128, 999]: + encoded = BER_len_enc(value) + length, remain = BER_len_dec(encoded) + assert length == value + assert remain == b"" + + assert BER_len_enc(45, size=None) == BER_len_enc(45, size=0) + assert BER_len_enc(45, size=4) == b"\x84\x00\x00\x00-" + + _raises(BER_Exception, lambda: BER_len_enc(0, size=128)) + + _raises(BER_Decoding_Error, lambda: BER_len_dec(b"\x82")) + + +def check_ber_num_enc_dec(): + # type: () -> None + for value in [0, 1, 127, 256, 16384]: + encoded = BER_num_enc(value) + decoded, remain = BER_num_dec(encoded) + assert decoded == value + assert remain == b"" + + _raises(BER_Decoding_Error, lambda: BER_num_dec(b"")) + + _raises(BER_Decoding_Error, lambda: BER_num_dec(b"\x80\x80")) + + +def check_ber_id_enc_dec(): + # type: () -> None + for tag in [0x02, 0x30, 0x81, 0xA0]: + encoded = BER_id_enc(tag) + decoded, remain = BER_id_dec(encoded) + assert decoded == tag + assert remain == b"" + + high_tag = (0x03 << 5) + 0x22 + encoded = BER_id_enc(high_tag) + decoded, remain = BER_id_dec(encoded) + assert decoded == high_tag + assert remain == b"" + + +def check_ber_tagging(): + # type: () -> None + inner = BERcodec_INTEGER.enc(7) + implicit = BER_tagging_enc(inner, implicit_tag=0xA0) + assert implicit.startswith(b"\xa0") + real_tag, payload = BER_tagging_dec( + implicit, + hidden_tag=ASN1_Class_UNIVERSAL.INTEGER, + implicit_tag=0xA0, + ) + assert real_tag is None + assert payload[0] == int(ASN1_Class_UNIVERSAL.INTEGER) + + conf.ASN1_default_long_size = 4 + try: + explicit = BER_tagging_enc(inner, explicit_tag=0xA1) + assert explicit.startswith(b"\xa1\x84") + real_tag, payload = BER_tagging_dec( + explicit, + explicit_tag=0xA1, + ) + assert real_tag is None + assert payload == inner + finally: + conf.ASN1_default_long_size = 0 + + _raises(BER_Decoding_Error, lambda: BER_tagging_dec( + implicit, + hidden_tag=ASN1_Class_UNIVERSAL.INTEGER, + implicit_tag=0xA1, + )) + + safe_tag, _ = BER_tagging_dec( + implicit, + hidden_tag=ASN1_Class_UNIVERSAL.INTEGER, + implicit_tag=0xA1, + safe=True, + ) + assert safe_tag == 0xA0 + + +def check_ber_integer(): + # type: () -> None + for value in [0, 1, 127, 128, 255, -1, -128, -129]: + encoded = BERcodec_INTEGER.enc(value) + obj, remain = BERcodec_INTEGER.do_dec(encoded) + assert obj.val == value + assert remain == b"" + + _raises(BER_BadTag_Decoding_Error, lambda: BERcodec_INTEGER.do_dec(BERcodec_STRING.enc(b"x"))) + + _raises(BER_Decoding_Error, lambda: BERcodec_INTEGER.check_type_get_len(b"\x02")) + + +def check_ber_bit_string(): + # type: () -> None + encoded = BERcodec_BIT_STRING.enc("1011") + obj, remain = BERcodec_BIT_STRING.do_dec(encoded) + assert obj.val == "1011" + assert remain == b"" + + padded = BERcodec_BIT_STRING.enc("10110000") + obj2, _ = BERcodec_BIT_STRING.do_dec(padded) + assert obj2.val == "10110000" + + _raises(BER_Decoding_Error, lambda: BERcodec_BIT_STRING.do_dec(b"\x03\x01\x08", safe=True)) + + _raises(BER_Decoding_Error, lambda: BERcodec_BIT_STRING.do_dec(b"\x03\x00")) + + +def check_ber_string_and_null(): + # type: () -> None + encoded = BERcodec_STRING.enc(b"hello") + obj, remain = BERcodec_STRING.do_dec(encoded) + assert obj.val == b"hello" + assert remain == b"" + + null = BERcodec_NULL.enc(0) + assert null == b"\x05\x00" + obj, remain = BERcodec_NULL.do_dec(null) + assert obj.val == 0 + + non_null = BERcodec_NULL.enc(42) + obj, remain = BERcodec_NULL.do_dec(non_null) + assert obj.val == 42 + + +def check_ber_oid(): + # type: () -> None + encoded = BERcodec_OID.enc("1.2.840.113556.1.4.529") + obj, remain = BERcodec_OID.do_dec(encoded) + assert obj.val == "1.2.840.113556.1.4.529" + assert remain == b"" + + empty, remain = BERcodec_OID.do_dec(BERcodec_OID.enc("")) + assert empty.val == "" + assert remain == b"" + + +def check_ber_sequence_and_set(): + # type: () -> None + payload = BERcodec_INTEGER.enc(1) + BERcodec_INTEGER.enc(2) + seq = BERcodec_SEQUENCE.enc(payload) + obj, remain = BERcodec_SEQUENCE.do_dec(seq) + assert len(obj.val) == 2 + assert obj.val[0].val == 1 + assert obj.val[1].val == 2 + assert remain == b"" + + as_list = BERcodec_SEQUENCE.enc([ASN1_INTEGER(3), ASN1_INTEGER(4)]) + obj2, remain2 = BERcodec_SEQUENCE.do_dec(as_list) + assert [x.val for x in obj2.val] == [3, 4] + assert remain2 == b"" + + st = BERcodec_SET.enc(payload) + obj3, remain3 = BERcodec_SET.do_dec(st) + assert len(obj3.val) == 2 + assert remain3 == b"" + + conf.ASN1_default_long_size = 4 + try: + long_seq = BERcodec_SEQUENCE.enc(payload) + assert long_seq.startswith(b"0\x84") + finally: + conf.ASN1_default_long_size = 0 + + _raises(BER_Decoding_Error, lambda: BERcodec_SEQUENCE.do_dec(b"\x30\x05" + BERcodec_INTEGER.enc(1))) + + +def check_ber_ipaddress(): + # type: () -> None + encoded = BERcodec_IPADDRESS.enc("192.168.0.1") + obj, remain = BERcodec_IPADDRESS.do_dec(encoded) + assert obj.val == "192.168.0.1" + assert remain == b"" + + _raises(BER_Encoding_Error, lambda: BERcodec_IPADDRESS.enc("not-an-ip")) + + _raises(BER_Decoding_Error, lambda: BERcodec_IPADDRESS.do_dec(BERcodec_STRING.enc(b"bad"))) + + +def check_ber_object_dispatch(): + # type: () -> None + encoded = BERcodec_INTEGER.enc(99) + obj, remain = BERcodec_Object.do_dec(encoded) + assert obj.val == 99 + assert remain == b"" + + _raises(BER_Decoding_Error, lambda: BERcodec_Object.check_string(b"")) + + _raises(BER_Decoding_Error, lambda: BERcodec_Object.do_dec(b"\xff\x00")) + + bad, remain = BERcodec_Object.safedec(b"\x02\x01\x01") + assert isinstance(bad, ASN1_INTEGER) + assert bad.val == 1 + + unknown, remain = BERcodec_Object.safedec(b"\xff\x00") + assert isinstance(unknown, ASN1_DECODING_ERROR) + + truncated, remain = BERcodec_Object.dec(b"\x02\x05\x01", safe=True) + assert isinstance(truncated, ASN1_DECODING_ERROR) + assert remain == b"" + + _raises(TypeError, lambda: BERcodec_Object.enc(object())) + assert BERcodec_Object.enc("42") == BERcodec_STRING.enc("42") diff --git a/test/scapy/layers/ber_packets.py b/test/scapy/layers/ber_packets.py new file mode 100644 index 00000000000..ccb503e2a60 --- /dev/null +++ b/test/scapy/layers/ber_packets.py @@ -0,0 +1,150 @@ +# SPDX-License-Identifier: GPL-2.0-only +# This file is part of Scapy +# See https://scapy.net/ for more information + +""" +BER ASN1_Packet and ASN1F_field build tests. +""" + +from scapy.asn1.asn1 import ASN1_Codecs, ASN1_INTEGER, ASN1_STRING +from scapy.asn1fields import ( + ASN1F_BOOLEAN, + ASN1F_CHOICE, + ASN1F_INTEGER, + ASN1F_SEQUENCE, + ASN1F_SEQUENCE_OF, + ASN1F_STRING, + ASN1F_optional, +) +from scapy.asn1packet import ASN1_Packet +from scapy.packet import raw + + +class BERTaggedInteger(ASN1_Packet): + ASN1_codec = ASN1_Codecs.BER + ASN1_root = ASN1F_INTEGER("n", 0, explicit_tag=0xA1) + + +class BERFixedFields(ASN1_Packet): + ASN1_codec = ASN1_Codecs.BER + ASN1_root = ASN1F_SEQUENCE( + ASN1F_INTEGER("n", 0, size_len=1), + ASN1F_STRING("s", "", size_len=3), + ) + + +class BEROptionalField(ASN1_Packet): + ASN1_codec = ASN1_Codecs.BER + ASN1_root = ASN1F_SEQUENCE( + ASN1F_INTEGER("id", 0), + ASN1F_optional(ASN1F_INTEGER("extra", 0, explicit_tag=0xA0)), + ) + + +class BERSequenceOfIntegers(ASN1_Packet): + ASN1_codec = ASN1_Codecs.BER + ASN1_root = ASN1F_SEQUENCE_OF("values", [], ASN1F_INTEGER) + + +class BERChoiceField(ASN1_Packet): + ASN1_codec = ASN1_Codecs.BER + ASN1_root = ASN1F_CHOICE( + "c", ASN1_INTEGER(0), ASN1F_INTEGER, ASN1F_STRING, + ) + + +class BERRecord(ASN1_Packet): + ASN1_codec = ASN1_Codecs.BER + ASN1_root = ASN1F_SEQUENCE( + ASN1F_INTEGER("id", 0), + ASN1F_BOOLEAN("flag", False), + ASN1F_STRING("label", ""), + ASN1F_optional(ASN1F_INTEGER("extra", 0, explicit_tag=0xA0)), + ASN1F_SEQUENCE_OF("values", [], ASN1F_INTEGER), + ) + + +def _roundtrip(cls, pkt): + # type: (type, ASN1_Packet) -> ASN1_Packet + return cls(raw(pkt)) + + +def check_ber_field_explicit_tag(): + # type: () -> None + pkt = BERTaggedInteger(n=5) + assert raw(pkt) == b"\xa1\x03\x02\x01\x05" + decoded = _roundtrip(BERTaggedInteger, pkt) + assert decoded.n.val == 5 + + +def check_ber_field_fixed_size(): + # type: () -> None + pkt = BERFixedFields(n=200, s=b"ABC") + assert raw(pkt) == bytes.fromhex("300d02810200c80483000003414243") + decoded = _roundtrip(BERFixedFields, pkt) + assert decoded.n.val == 200 + assert decoded.s.val == b"ABC" + + +def check_ber_field_optional(): + # type: () -> None + present = BEROptionalField(id=1, extra=7) + assert raw(present) == bytes.fromhex("3008020101a003020107") + decoded = _roundtrip(BEROptionalField, present) + assert decoded.id.val == 1 + assert decoded.extra.val == 7 + + absent = BEROptionalField(id=1, extra=None) + assert raw(absent) == bytes.fromhex("3003020101") + decoded = _roundtrip(BEROptionalField, absent) + assert decoded.id.val == 1 + assert decoded.extra is None + + +def check_ber_field_sequence_of(): + # type: () -> None + pkt = BERSequenceOfIntegers(values=[1, 2, 3]) + assert raw(pkt) == b"\x30\x09\x02\x01\x01\x02\x01\x02\x02\x01\x03" + decoded = _roundtrip(BERSequenceOfIntegers, pkt) + assert [x.val for x in decoded.values] == [1, 2, 3] + + +def check_ber_field_choice(): + # type: () -> None + as_int = BERChoiceField(c=ASN1_INTEGER(99)) + assert raw(as_int) == b"\x02\x01c" + decoded = _roundtrip(BERChoiceField, as_int) + assert decoded.c.val == 99 + + as_str = BERChoiceField(c=ASN1_STRING("x")) + assert raw(as_str) == b"\x04\x01x" + decoded = _roundtrip(BERChoiceField, as_str) + assert decoded.c.val == b"x" + + +def check_ber_packet_record(): + # type: () -> None + pkt = BERRecord( + id=42, flag=True, label="hi", extra=7, values=[1, 2, 3], + ) + expected = bytes.fromhex( + "301a02012a01010104026869" + "a003020107" + "3009020101020102020103" + ) + assert raw(pkt) == expected + decoded = _roundtrip(BERRecord, pkt) + assert decoded.id.val == 42 + assert decoded.flag.val == 1 + assert decoded.label.val == b"hi" + assert decoded.extra.val == 7 + assert [x.val for x in decoded.values] == [1, 2, 3] + + empty = BERRecord(id=1, flag=False, label="", extra=None, values=[]) + assert raw(empty) == bytes.fromhex("300a02010101010004003000") + decoded = _roundtrip(BERRecord, empty) + assert decoded.id.val == 1 + assert decoded.flag.val == 0 + assert decoded.label.val == b"" + assert decoded.extra is None + assert [x.val for x in decoded.values] == [] From 7b57e8fd1349af5cb127b8d9e0464f1aedaef2e0 Mon Sep 17 00:00:00 2001 From: Nils Weiss Date: Sun, 19 Jul 2026 21:23:00 +0200 Subject: [PATCH 02/11] Fix mypy AI-Assisted: yes (Cursor) --- scapy/asn1/ber.py | 7 +++---- scapy/asn1fields.py | 9 +++++---- 2 files changed, 8 insertions(+), 8 deletions(-) diff --git a/scapy/asn1/ber.py b/scapy/asn1/ber.py index 75887d2b938..799aec18a2c 100644 --- a/scapy/asn1/ber.py +++ b/scapy/asn1/ber.py @@ -297,6 +297,9 @@ def __new__(cls, class BERcodec_Object(Generic[_K], metaclass=BERcodec_metaclass): codec = ASN1_Codecs.BER tag = ASN1_Class_UNIVERSAL.ANY + skip_tagging = False + tagging_enc = staticmethod(BER_tagging_enc) + tagging_dec = staticmethod(BER_tagging_dec) @classmethod def asn1_object(cls, val): @@ -423,10 +426,6 @@ def enc(cls, s, size_len=0): ASN1_Codecs.BER.register_stem(BERcodec_Object) -BERcodec_Object.tagging_enc = staticmethod(BER_tagging_enc) -BERcodec_Object.tagging_dec = staticmethod(BER_tagging_dec) -BERcodec_Object.skip_tagging = False - ########################## # BERcodec objects # diff --git a/scapy/asn1fields.py b/scapy/asn1fields.py index 19aa85907bc..fa8059f737d 100644 --- a/scapy/asn1fields.py +++ b/scapy/asn1fields.py @@ -13,6 +13,7 @@ from functools import reduce from scapy.asn1.asn1 import ( + ASN1Codec, ASN1_BIT_STRING, ASN1_BOOLEAN, ASN1_Class, @@ -131,23 +132,23 @@ def _apply_diff_tag(self, diff_tag): def _tagging_dec(self, pkt, s, **kwargs): # type: (ASN1_Packet, bytes, **Any) -> Tuple[Optional[int], bytes] - stem = pkt.ASN1_codec.get_stem() + stem = cast(ASN1Codec, pkt.ASN1_codec).get_stem() if getattr(stem, "skip_tagging", False): return None, s fn = getattr(stem, "tagging_dec", None) if fn is None: return BER_tagging_dec(s, **kwargs) - return fn(s, **kwargs) + return cast(Tuple[Optional[int], bytes], fn(s, **kwargs)) def _tagging_enc(self, pkt, s, **kwargs): # type: (ASN1_Packet, bytes, **Any) -> bytes - stem = pkt.ASN1_codec.get_stem() + stem = cast(ASN1Codec, pkt.ASN1_codec).get_stem() if getattr(stem, "skip_tagging", False): return s fn = getattr(stem, "tagging_enc", None) if fn is None: return BER_tagging_enc(s, **kwargs) - return fn(s, **kwargs) + return cast(bytes, fn(s, **kwargs)) def _codec_kwargs(self, size_len=None): # type: (Optional[int]) -> Dict[str, Any] From ec765a0b7a74c2ec4219272e300edcd3480d7277 Mon Sep 17 00:00:00 2001 From: Nils Weiss Date: Mon, 20 Jul 2026 09:08:58 +0200 Subject: [PATCH 03/11] Fix review AI-Assisted: yes (Cursor) --- scapy/asn1fields.py | 7 +------ test/scapy/layers/asn1.uts | 2 ++ test/scapy/layers/ber_packets.py | 34 ++++++++++++++++++++++++++++++++ 3 files changed, 37 insertions(+), 6 deletions(-) diff --git a/scapy/asn1fields.py b/scapy/asn1fields.py index fa8059f737d..eb000e35af1 100644 --- a/scapy/asn1fields.py +++ b/scapy/asn1fields.py @@ -765,12 +765,7 @@ def set_absent(self, pkt): def is_empty(self, pkt): # type: (ASN1_Packet) -> bool - val = getattr(pkt, self._field.name, None) - if val is None: - return True - if getattr(self._field, "islist", 0) and val == []: - return True - return False + return self._field.is_empty(pkt) class ASN1F_omit(ASN1F_field[None, None]): diff --git a/test/scapy/layers/asn1.uts b/test/scapy/layers/asn1.uts index e07e18e8165..d6c9725d729 100644 --- a/test/scapy/layers/asn1.uts +++ b/test/scapy/layers/asn1.uts @@ -109,6 +109,8 @@ __import__('test.scapy.layers.ber_packets', fromlist=['check_ber_field_explicit_ __import__('test.scapy.layers.ber_packets', fromlist=['check_ber_field_fixed_size']).check_ber_field_fixed_size() = BER field optional __import__('test.scapy.layers.ber_packets', fromlist=['check_ber_field_optional']).check_ber_field_optional() += BER optional SEQUENCE is_empty +__import__('test.scapy.layers.ber_packets', fromlist=['check_ber_optional_sequence_is_empty']).check_ber_optional_sequence_is_empty() = BER field sequence of __import__('test.scapy.layers.ber_packets', fromlist=['check_ber_field_sequence_of']).check_ber_field_sequence_of() = BER field choice diff --git a/test/scapy/layers/ber_packets.py b/test/scapy/layers/ber_packets.py index ccb503e2a60..09cb02fe6f1 100644 --- a/test/scapy/layers/ber_packets.py +++ b/test/scapy/layers/ber_packets.py @@ -64,6 +64,18 @@ class BERRecord(ASN1_Packet): ) +class BEROptionalSequence(ASN1_Packet): + ASN1_codec = ASN1_Codecs.BER + ASN1_root = ASN1F_SEQUENCE( + ASN1F_INTEGER("hdr", 0), + ASN1F_optional(ASN1F_SEQUENCE( + ASN1F_INTEGER("id", None), + ASN1F_STRING("label", None), + explicit_tag=0xA0, + )), + ) + + def _roundtrip(cls, pkt): # type: (type, ASN1_Packet) -> ASN1_Packet return cls(raw(pkt)) @@ -101,6 +113,28 @@ def check_ber_field_optional(): assert decoded.extra is None +def check_ber_optional_sequence_is_empty(): + # type: () -> None + """Optional ASN1F_SEQUENCE must use the wrapped field's is_empty(). + + SEQUENCE stores children under their own names (not dummy_seq_name), so + inspecting pkt.dummy_seq_name incorrectly reports present children as empty + and makes the parent SEQUENCE look empty. + """ + opt = BEROptionalSequence.ASN1_root.seq[1] + + present = BEROptionalSequence(hdr=1, id=42, label=b"abc") + assert opt._field.is_empty(present) is False + assert opt.is_empty(present) is False + assert BEROptionalSequence.ASN1_root.is_empty(present) is False + assert raw(present) == bytes.fromhex("300f020101a00a300802012a0403616263") + + absent = BEROptionalSequence(hdr=1, id=None, label=None) + assert opt._field.is_empty(absent) is True + assert opt.is_empty(absent) is True + assert raw(absent) == bytes.fromhex("3003020101") + + def check_ber_field_sequence_of(): # type: () -> None pkt = BERSequenceOfIntegers(values=[1, 2, 3]) From 190f03031d49b39186fd6ecf657afe87eb8a8d8a Mon Sep 17 00:00:00 2001 From: Nils Weiss Date: Tue, 21 Jul 2026 15:10:21 +0200 Subject: [PATCH 04/11] simplify code AI-Assisted: yes (Cursor) --- scapy/asn1fields.py | 112 ++++++++++++++++---------------------------- 1 file changed, 41 insertions(+), 71 deletions(-) diff --git a/scapy/asn1fields.py b/scapy/asn1fields.py index eb000e35af1..4ddbb55687d 100644 --- a/scapy/asn1fields.py +++ b/scapy/asn1fields.py @@ -130,26 +130,39 @@ def _apply_diff_tag(self, diff_tag): elif self.explicit_tag is not None: self.explicit_tag = diff_tag + def _codec_stem(self, pkt): + # type: (ASN1_Packet) -> type + return cast(ASN1Codec, pkt.ASN1_codec).get_stem() + def _tagging_dec(self, pkt, s, **kwargs): # type: (ASN1_Packet, bytes, **Any) -> Tuple[Optional[int], bytes] - stem = cast(ASN1Codec, pkt.ASN1_codec).get_stem() + stem = self._codec_stem(pkt) if getattr(stem, "skip_tagging", False): return None, s - fn = getattr(stem, "tagging_dec", None) - if fn is None: - return BER_tagging_dec(s, **kwargs) + fn = getattr(stem, "tagging_dec", BER_tagging_dec) return cast(Tuple[Optional[int], bytes], fn(s, **kwargs)) def _tagging_enc(self, pkt, s, **kwargs): # type: (ASN1_Packet, bytes, **Any) -> bytes - stem = cast(ASN1Codec, pkt.ASN1_codec).get_stem() + stem = self._codec_stem(pkt) if getattr(stem, "skip_tagging", False): return s - fn = getattr(stem, "tagging_enc", None) - if fn is None: - return BER_tagging_enc(s, **kwargs) + fn = getattr(stem, "tagging_enc", BER_tagging_enc) return cast(bytes, fn(s, **kwargs)) + def _apply_tagging_dec(self, s, pkt, **kwargs): + # type: (bytes, ASN1_Packet, **Any) -> bytes + tag_kwargs = { + "hidden_tag": self.ASN1_tag, + "implicit_tag": self.implicit_tag, + "explicit_tag": self.explicit_tag, + "safe": self.flexible_tag, + } # type: Dict[str, Any] + tag_kwargs.update(kwargs) + diff_tag, s = self._tagging_dec(pkt, s, **tag_kwargs) + self._apply_diff_tag(diff_tag) + return s + def _codec_kwargs(self, size_len=None): # type: (Optional[int]) -> Dict[str, Any] return { @@ -187,24 +200,10 @@ def m2i(self, pkt, s): as expected or not. Noticeably, input methods from cert.py expect certain exceptions to be raised. Hence default flexible_tag is False. """ - diff_tag, s = self._tagging_dec( - pkt, s, - hidden_tag=self.ASN1_tag, - implicit_tag=self.implicit_tag, - explicit_tag=self.explicit_tag, - safe=self.flexible_tag, - _fname=self.name, - ) - self._apply_diff_tag(diff_tag) + s = self._apply_tagging_dec(s, pkt, _fname=self.name) codec = self.ASN1_tag.get_codec(pkt.ASN1_codec) - if self.flexible_tag: - return codec.safedec( - s, context=self.context, **self._codec_kwargs() - ) # type: ignore - else: - return codec.dec( - s, context=self.context, **self._codec_kwargs() - ) # type: ignore + dec = codec.safedec if self.flexible_tag else codec.dec + return dec(s, context=self.context, **self._codec_kwargs()) # type: ignore def i2m(self, pkt, x): # type: (ASN1_Packet, Union[bytes, _I, _A]) -> bytes @@ -513,19 +512,6 @@ def get_fields_list(self): return reduce(lambda x, y: x + y.get_fields_list(), self.seq, []) - def _apply_tagging_dec(self, s, pkt): - # type: (bytes, Any) -> bytes - diff_tag, s = self._tagging_dec( - pkt, s, - hidden_tag=self.ASN1_tag, - implicit_tag=self.implicit_tag, - explicit_tag=self.explicit_tag, - safe=self.flexible_tag, - _fname=pkt.name, - ) - self._apply_diff_tag(diff_tag) - return s - def _dissect_sequence_children(self, pkt, s): # type: (Any, bytes) -> bytes if len(s) == 0: @@ -541,7 +527,7 @@ def _dissect_sequence_children(self, pkt, s): def _m2i_ber(self, pkt, s): # type: (Any, bytes) -> Tuple[Any, bytes] - s = self._apply_tagging_dec(s, pkt) + s = self._apply_tagging_dec(s, pkt, _fname=pkt.name) codec = self.ASN1_tag.get_codec(pkt.ASN1_codec) i, s, remain = codec.check_type_check_len(s) s = self._dissect_sequence_children(pkt, s) @@ -556,11 +542,9 @@ def m2i(self, pkt, s): # type: (Any, bytes) -> Tuple[Any, bytes] """ ASN1F_SEQUENCE behaves transparently, with nested ASN1_objects being - dissected one by one. Because we use obj.dissect (see loop below) - instead of obj.m2i (as we trust dissect to do the appropriate set_vals) - we do not directly retrieve the list of nested objects. - Thus m2i returns an empty list (along with the proper remainder). - It is discarded by dissect() and should not be missed elsewhere. + dissected one by one. m2i returns an empty list (along with the proper + remainder). It is discarded by dissect() and should not be missed + elsewhere. """ return self._m2i_ber(pkt, s) @@ -637,14 +621,7 @@ def m2i(self, s, # type: bytes ): # type: (...) -> Tuple[List[Any], bytes] - diff_tag, s = self._tagging_dec( - pkt, s, - hidden_tag=self.ASN1_tag, - implicit_tag=self.implicit_tag, - explicit_tag=self.explicit_tag, - safe=self.flexible_tag, - ) - self._apply_diff_tag(diff_tag) + s = self._apply_tagging_dec(s, pkt) codec = self.ASN1_tag.get_codec(pkt.ASN1_codec) i, s, remain = codec.check_type_check_len(s) lst = [] @@ -833,9 +810,6 @@ def __init__(self, name, default, *args, **kwargs): self.pktchoices[hash(p.cls)] = (p.implicit_tag, p.explicit_tag) # noqa: E501 else: raise ASN1_Error("ASN1F_CHOICE: no tag found for one field") - self._tag_to_index = { - tag: idx for idx, tag in enumerate(self.choice_order) - } def _register_choice(self, tag, choice): # type: (int, _CHOICE_T) -> None @@ -853,11 +827,7 @@ def _dissect_choice_payload(self, pkt, choice, payload): def _m2i_ber(self, pkt, s): # type: (ASN1_Packet, bytes) -> Tuple[ASN1_Object[Any], bytes] - _, s = self._tagging_dec( - pkt, s, - hidden_tag=self.ASN1_tag, - explicit_tag=self.explicit_tag, - ) + s = self._apply_tagging_dec(s, pkt) tag, _ = BER_id_dec(s) return self._m2i_tagged(pkt, tag, s) @@ -886,11 +856,6 @@ def m2i(self, pkt, s): raise ASN1_Error("ASN1F_CHOICE: got empty string") return self._m2i_ber(pkt, s) - def _choice_tag_for(self, x): - # type: (Any) -> Optional[int] - index = self._choice_index_for(x) - return None if index is None else self.choice_order[index] - def _choice_index_for(self, x): # type: (Any) -> Optional[int] for index, choice in enumerate(self.choice_list): @@ -902,6 +867,15 @@ def _choice_index_for(self, x): return index return None + def _choice_for_index(self, index): + # type: (int) -> _CHOICE_T + return self.choice_list[index] + + def _choice_tag_for(self, x): + # type: (Any) -> Optional[int] + index = self._choice_index_for(x) + return None if index is None else self.choice_order[index] + def i2m(self, pkt, x): # type: (ASN1_Packet, Any) -> bytes if x is None: @@ -972,15 +946,11 @@ def m2i(self, pkt, s): if not hasattr(cls, "ASN1_root"): # A normal Packet (!= ASN1) return self.extract_packet(cls, s, _underlayer=pkt) - diff_tag, s = self._tagging_dec( - pkt, s, + s = self._apply_tagging_dec( + s, pkt, hidden_tag=cls.ASN1_root.ASN1_tag, # noqa: E501 - implicit_tag=self.implicit_tag, - explicit_tag=self.explicit_tag, - safe=self.flexible_tag, _fname=self.name, ) - self._apply_diff_tag(diff_tag) if not s: return None, s return self.extract_packet(cls, s, _underlayer=pkt) From 121435b4d7fdc9fdc0ef10574eb0e8581740c3bc Mon Sep 17 00:00:00 2001 From: Nils Weiss Date: Tue, 21 Jul 2026 16:00:05 +0200 Subject: [PATCH 05/11] asn1: address PR review on tagging and codec kwargs Apply field tagging in SEQUENCE OF builds, honor size_len for ASN1_Object values, and trim redundant optional/tagging helpers. AI-Assisted: yes (Cursor) --- scapy/asn1fields.py | 62 ++++++++++++++++---------------- test/scapy/layers/asn1.uts | 4 +++ test/scapy/layers/ber_packets.py | 30 ++++++++++++++++ 3 files changed, 64 insertions(+), 32 deletions(-) diff --git a/scapy/asn1fields.py b/scapy/asn1fields.py index 4ddbb55687d..864aa611bae 100644 --- a/scapy/asn1fields.py +++ b/scapy/asn1fields.py @@ -134,21 +134,24 @@ def _codec_stem(self, pkt): # type: (ASN1_Packet) -> type return cast(ASN1Codec, pkt.ASN1_codec).get_stem() - def _tagging_dec(self, pkt, s, **kwargs): - # type: (ASN1_Packet, bytes, **Any) -> Tuple[Optional[int], bytes] + def _tagging(self, pkt, s, encode, **kwargs): + # type: (ASN1_Packet, bytes, bool, **Any) -> Any stem = self._codec_stem(pkt) if getattr(stem, "skip_tagging", False): - return None, s + return s if encode else (None, s) + if encode: + fn = getattr(stem, "tagging_enc", BER_tagging_enc) + return cast(bytes, fn(s, **kwargs)) fn = getattr(stem, "tagging_dec", BER_tagging_dec) return cast(Tuple[Optional[int], bytes], fn(s, **kwargs)) + def _tagging_dec(self, pkt, s, **kwargs): + # type: (ASN1_Packet, bytes, **Any) -> Tuple[Optional[int], bytes] + return self._tagging(pkt, s, False, **kwargs) + def _tagging_enc(self, pkt, s, **kwargs): # type: (ASN1_Packet, bytes, **Any) -> bytes - stem = self._codec_stem(pkt) - if getattr(stem, "skip_tagging", False): - return s - fn = getattr(stem, "tagging_enc", BER_tagging_enc) - return cast(bytes, fn(s, **kwargs)) + return self._tagging(pkt, s, True, **kwargs) def _apply_tagging_dec(self, s, pkt, **kwargs): # type: (bytes, ASN1_Packet, **Any) -> bytes @@ -171,10 +174,22 @@ def _codec_kwargs(self, size_len=None): def _encode_item(self, pkt, item): # type: (ASN1_Packet, Any) -> bytes - if isinstance(item, ASN1_Object): - return item.enc(pkt.ASN1_codec) + """Encode a field value with codec kwargs, without field tagging.""" + if item is None: + return b"" if hasattr(item, "self_build"): return cast("ASN1_Packet", item).self_build() + if isinstance(item, ASN1_Object): + if (self.ASN1_tag == ASN1_Class_UNIVERSAL.ANY or + item.tag == ASN1_Class_UNIVERSAL.RAW or + item.tag == ASN1_Class_UNIVERSAL.ERROR): + return item.enc(pkt.ASN1_codec) + if self.ASN1_tag != item.tag: + raise ASN1_Error( + "Encoding Error: got %r instead of an %r for field [%s]" % + (item, self.ASN1_tag, self.name) + ) + item = item.val codec = self.ASN1_tag.get_codec(pkt.ASN1_codec) return codec.enc(item, **self._codec_kwargs()) @@ -209,18 +224,7 @@ def i2m(self, pkt, x): # type: (ASN1_Packet, Union[bytes, _I, _A]) -> bytes if x is None: return b"" - if isinstance(x, ASN1_Object): - if (self.ASN1_tag == ASN1_Class_UNIVERSAL.ANY or - x.tag == ASN1_Class_UNIVERSAL.RAW or - x.tag == ASN1_Class_UNIVERSAL.ERROR or - self.ASN1_tag == x.tag): - s = x.enc(pkt.ASN1_codec) - else: - raise ASN1_Error("Encoding Error: got %r instead of an %r for field [%s]" % (x, self.ASN1_tag, self.name)) # noqa: E501 - else: - s = self.ASN1_tag.get_codec(pkt.ASN1_codec).enc( - x, **self._codec_kwargs() - ) + s = self._encode_item(pkt, x) return self._tagging_enc( pkt, s, implicit_tag=self.implicit_tag, @@ -647,7 +651,8 @@ def build(self, pkt): elif self.holds_packets: s = b"".join(bytes(i) for i in val) else: - s = b"".join(self.fld._encode_item(pkt, i) for i in val) + # Use i2m so element implicit/explicit tags match m2i()/fld.m2i() + s = b"".join(self.fld.i2m(pkt, i) for i in val) return self.i2m(pkt, s) def i2repr(self, pkt, x): @@ -732,17 +737,10 @@ def i2repr(self, pkt, x): # type: (ASN1_Packet, Any) -> str return self._field.i2repr(pkt, x) - def set_val(self, pkt, val): - # type: (ASN1_Packet, Any) -> None - self._field.set_val(pkt, val) - def set_absent(self, pkt): # type: (ASN1_Packet) -> None - self.set_val(pkt, None) - - def is_empty(self, pkt): - # type: (ASN1_Packet) -> bool - return self._field.is_empty(pkt) + # Used by codecs that track optionality explicitly (e.g. PER). + self._field.set_val(pkt, None) class ASN1F_omit(ASN1F_field[None, None]): diff --git a/test/scapy/layers/asn1.uts b/test/scapy/layers/asn1.uts index d6c9725d729..5ae5b61e308 100644 --- a/test/scapy/layers/asn1.uts +++ b/test/scapy/layers/asn1.uts @@ -113,6 +113,10 @@ __import__('test.scapy.layers.ber_packets', fromlist=['check_ber_field_optional' __import__('test.scapy.layers.ber_packets', fromlist=['check_ber_optional_sequence_is_empty']).check_ber_optional_sequence_is_empty() = BER field sequence of __import__('test.scapy.layers.ber_packets', fromlist=['check_ber_field_sequence_of']).check_ber_field_sequence_of() += BER SEQUENCE OF tagged elements +__import__('test.scapy.layers.ber_packets', fromlist=['check_ber_sequence_of_tagged_elements']).check_ber_sequence_of_tagged_elements() += BER ASN1_Object codec kwargs +__import__('test.scapy.layers.ber_packets', fromlist=['check_ber_asn1_object_codec_kwargs']).check_ber_asn1_object_codec_kwargs() = BER field choice __import__('test.scapy.layers.ber_packets', fromlist=['check_ber_field_choice']).check_ber_field_choice() = BER packet record diff --git a/test/scapy/layers/ber_packets.py b/test/scapy/layers/ber_packets.py index 09cb02fe6f1..68e8c0a227b 100644 --- a/test/scapy/layers/ber_packets.py +++ b/test/scapy/layers/ber_packets.py @@ -76,6 +76,18 @@ class BEROptionalSequence(ASN1_Packet): ) +class BERSequenceOfTaggedIntegers(ASN1_Packet): + ASN1_codec = ASN1_Codecs.BER + ASN1_root = ASN1F_SEQUENCE_OF( + "values", [], ASN1F_INTEGER("v", 0, explicit_tag=0xA0), + ) + + +class BERSizedInteger(ASN1_Packet): + ASN1_codec = ASN1_Codecs.BER + ASN1_root = ASN1F_INTEGER("n", 0, size_len=1) + + def _roundtrip(cls, pkt): # type: (type, ASN1_Packet) -> ASN1_Packet return cls(raw(pkt)) @@ -143,6 +155,24 @@ def check_ber_field_sequence_of(): assert [x.val for x in decoded.values] == [1, 2, 3] +def check_ber_sequence_of_tagged_elements(): + # type: () -> None + """SEQUENCE OF must apply the element field's tagging on build.""" + pkt = BERSequenceOfTaggedIntegers(values=[1, 2]) + assert raw(pkt) == bytes.fromhex("300aa003020101a003020102") + decoded = _roundtrip(BERSequenceOfTaggedIntegers, pkt) + assert [x.val for x in decoded.values] == [1, 2] + + +def check_ber_asn1_object_codec_kwargs(): + # type: () -> None + """ASN1_Object values must honor field codec kwargs such as size_len.""" + as_int = BERSizedInteger(n=5) + as_obj = BERSizedInteger(n=ASN1_INTEGER(5)) + assert raw(as_int) == raw(as_obj) == b"\x02\x81\x01\x05" + assert _roundtrip(BERSizedInteger, as_obj).n.val == 5 + + def check_ber_field_choice(): # type: () -> None as_int = BERChoiceField(c=ASN1_INTEGER(99)) From d0b76c5dc1666e22a83ab228316a860dfcb7148d Mon Sep 17 00:00:00 2001 From: Nils Weiss Date: Tue, 21 Jul 2026 16:10:37 +0200 Subject: [PATCH 06/11] =?UTF-8?q?Fixed.=20The=20Kerberos=20DCE=5FSTYLE=20f?= =?UTF-8?q?ailures=20came=20from=20=5Fencode=5Fitem:=20Packet=20values=20(?= =?UTF-8?q?e.g.=20KRB=5FAuthenticatorChecksum=20in=20an=20ASN1F=5FSTRING)?= =?UTF-8?q?=20used=20=20=20self=5Fbuild()=20alone=20and=20skipped=20the=20?= =?UTF-8?q?BER=20OCTET=20STRING=20wrap=20(04=20=E2=80=A6).=20Optional=20ck?= =?UTF-8?q?sum=20then=20failed=20to=20decode,=20so=20the=20stream=20still?= =?UTF-8?q?=20had=200xa3=20=20=20when=20cusec=20(0xa4)=20was=20expected.?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit AI-Assisted: yes (Cursor) --- scapy/asn1fields.py | 13 +++++++++---- test/scapy/layers/asn1.uts | 2 ++ test/scapy/layers/ber_packets.py | 17 +++++++++++++++++ 3 files changed, 28 insertions(+), 4 deletions(-) diff --git a/scapy/asn1fields.py b/scapy/asn1fields.py index 864aa611bae..2f99ec4474a 100644 --- a/scapy/asn1fields.py +++ b/scapy/asn1fields.py @@ -147,11 +147,14 @@ def _tagging(self, pkt, s, encode, **kwargs): def _tagging_dec(self, pkt, s, **kwargs): # type: (ASN1_Packet, bytes, **Any) -> Tuple[Optional[int], bytes] - return self._tagging(pkt, s, False, **kwargs) + return cast( + Tuple[Optional[int], bytes], + self._tagging(pkt, s, False, **kwargs), + ) def _tagging_enc(self, pkt, s, **kwargs): # type: (ASN1_Packet, bytes, **Any) -> bytes - return self._tagging(pkt, s, True, **kwargs) + return cast(bytes, self._tagging(pkt, s, True, **kwargs)) def _apply_tagging_dec(self, s, pkt, **kwargs): # type: (bytes, ASN1_Packet, **Any) -> bytes @@ -177,8 +180,6 @@ def _encode_item(self, pkt, item): """Encode a field value with codec kwargs, without field tagging.""" if item is None: return b"" - if hasattr(item, "self_build"): - return cast("ASN1_Packet", item).self_build() if isinstance(item, ASN1_Object): if (self.ASN1_tag == ASN1_Class_UNIVERSAL.ANY or item.tag == ASN1_Class_UNIVERSAL.RAW or @@ -190,6 +191,10 @@ def _encode_item(self, pkt, item): (item, self.ASN1_tag, self.name) ) item = item.val + elif hasattr(item, "self_build"): + # Packet values (e.g. ASN1F_STRING_PacketField) must still go through + # the BER type codec so the universal tag/length are applied. + item = item.self_build() codec = self.ASN1_tag.get_codec(pkt.ASN1_codec) return codec.enc(item, **self._codec_kwargs()) diff --git a/test/scapy/layers/asn1.uts b/test/scapy/layers/asn1.uts index 5ae5b61e308..b00caa345db 100644 --- a/test/scapy/layers/asn1.uts +++ b/test/scapy/layers/asn1.uts @@ -117,6 +117,8 @@ __import__('test.scapy.layers.ber_packets', fromlist=['check_ber_field_sequence_ __import__('test.scapy.layers.ber_packets', fromlist=['check_ber_sequence_of_tagged_elements']).check_ber_sequence_of_tagged_elements() = BER ASN1_Object codec kwargs __import__('test.scapy.layers.ber_packets', fromlist=['check_ber_asn1_object_codec_kwargs']).check_ber_asn1_object_codec_kwargs() += BER STRING field packet value +__import__('test.scapy.layers.ber_packets', fromlist=['check_ber_string_field_packet_value']).check_ber_string_field_packet_value() = BER field choice __import__('test.scapy.layers.ber_packets', fromlist=['check_ber_field_choice']).check_ber_field_choice() = BER packet record diff --git a/test/scapy/layers/ber_packets.py b/test/scapy/layers/ber_packets.py index 68e8c0a227b..ddf68d3dd0a 100644 --- a/test/scapy/layers/ber_packets.py +++ b/test/scapy/layers/ber_packets.py @@ -173,6 +173,23 @@ def check_ber_asn1_object_codec_kwargs(): assert _roundtrip(BERSizedInteger, as_obj).n.val == 5 +def check_ber_string_field_packet_value(): + # type: () -> None + """Packet values in ASN1F_STRING must still get a BER universal STRING tag.""" + from scapy.fields import StrFixedLenField + from scapy.packet import Packet + + class Blob(Packet): + fields_desc = [StrFixedLenField("data", b"ABCD", 4)] + + class P(ASN1_Packet): + ASN1_codec = ASN1_Codecs.BER + ASN1_root = ASN1F_STRING("s", "") + + enc = P.ASN1_root.i2m(P(), Blob()) + assert enc == b"\x04\x04ABCD" + + def check_ber_field_choice(): # type: () -> None as_int = BERChoiceField(c=ASN1_INTEGER(99)) From 2a2cfb0b409bc0434aa1f22150e0a4276f750997 Mon Sep 17 00:00:00 2001 From: Nils Weiss Date: Tue, 21 Jul 2026 20:42:23 +0200 Subject: [PATCH 07/11] =?UTF-8?q?Fixed.=20=5Fencode=5Fitem=20was=20always?= =?UTF-8?q?=20calling=20codec.enc(...,=20size=5Flen=3DNone),=20so=20conf.A?= =?UTF-8?q?SN1=5Fdefault=5Flong=5Fsize=20padded=20every=20BER=20length=20?= =?UTF-8?q?=E2=80=94=20=20=20including=20INTEGER/STRING=20=E2=80=94=20inst?= =?UTF-8?q?ead=20of=20only=20SEQUENCE/SET.?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit AI-Assisted: yes (Cursor) --- scapy/asn1fields.py | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/scapy/asn1fields.py b/scapy/asn1fields.py index 2f99ec4474a..f8d09c909df 100644 --- a/scapy/asn1fields.py +++ b/scapy/asn1fields.py @@ -190,6 +190,11 @@ def _encode_item(self, pkt, item): "Encoding Error: got %r instead of an %r for field [%s]" % (item, self.ASN1_tag, self.name) ) + # Without an explicit field size_len, keep ASN1_Object.enc() so + # conf.ASN1_default_long_size only affects SEQUENCE/SET (via their + # bytes payload path) and explicit tagging — Microsoft LDAP style. + if self.size_len is None: + return item.enc(pkt.ASN1_codec) item = item.val elif hasattr(item, "self_build"): # Packet values (e.g. ASN1F_STRING_PacketField) must still go through From ef80e3d6f63b227d7c104fc972fcfa001ef919cf Mon Sep 17 00:00:00 2001 From: Nils Weiss Date: Thu, 23 Jul 2026 12:51:07 +0200 Subject: [PATCH 08/11] Implemented now (feasible) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 1. Copilot: size_len=0 vs unspecified — Real bug. BERcodec_SEQUENCE.enc now defaults to size_len=None and only applies conf.ASN1_default_long_size when size_len is None, so size_len=0 can force short-form lengths. Regression test added. 2. guedou: drop unused helpers — Removed set_absent, SEQUENCE/CHOICE _m2i_ber, _register_choice, choice_order/choice_list, and unused _choice_* APIs. Logic inlined back into m2i / __init__ so this PR stays focused on codec-stem hooks. Kept on purpose • SEQUENCE OF holds_packets split — Needed for correct BER with tagged element fields (not for other encodings). Clarified the comment. • Prior encoding bugfixes (ASN1_Object/Packet in _encode_item) — Required so LDAP/Kerberos still work with the new encode path. asn1.uts, kerberos.uts, and ldap.uts all pass. Changes are uncommitted if you want a commit next. AI-Assisted: yes (Cursor) --- scapy/asn1/ber.py | 5 +- scapy/asn1fields.py | 152 ++++++++++++--------------------- test/scapy/layers/ber_codec.py | 4 + 3 files changed, 60 insertions(+), 101 deletions(-) diff --git a/scapy/asn1/ber.py b/scapy/asn1/ber.py index 799aec18a2c..95954c701a7 100644 --- a/scapy/asn1/ber.py +++ b/scapy/asn1/ber.py @@ -653,13 +653,14 @@ class BERcodec_SEQUENCE(BERcodec_Object[Union[bytes, List[BERcodec_Object[Any]]] tag = ASN1_Class_UNIVERSAL.SEQUENCE @classmethod - def enc(cls, _ll, size_len=0): + def enc(cls, _ll, size_len=None): # type: (Union[bytes, List[BERcodec_Object[Any]]], Optional[int]) -> bytes if isinstance(_ll, bytes): ll = _ll else: ll = b"".join(x.enc(cls.codec) for x in _ll) - if not size_len: + # None = apply conf; explicit 0 keeps short-form lengths. + if size_len is None: size_len = conf.ASN1_default_long_size return chb(int(cls.tag)) + BER_len_enc(len(ll), size=size_len) + ll diff --git a/scapy/asn1fields.py b/scapy/asn1fields.py index f8d09c909df..9233494b023 100644 --- a/scapy/asn1fields.py +++ b/scapy/asn1fields.py @@ -526,41 +526,34 @@ def get_fields_list(self): return reduce(lambda x, y: x + y.get_fields_list(), self.seq, []) - def _dissect_sequence_children(self, pkt, s): - # type: (Any, bytes) -> bytes - if len(s) == 0: - for obj in self.seq: - obj.set_val(pkt, None) - return s - for obj in self.seq: - try: - s = obj.dissect(pkt, s) - except ASN1F_badsequence: - break - return s - - def _m2i_ber(self, pkt, s): - # type: (Any, bytes) -> Tuple[Any, bytes] - s = self._apply_tagging_dec(s, pkt, _fname=pkt.name) - codec = self.ASN1_tag.get_codec(pkt.ASN1_codec) - i, s, remain = codec.check_type_check_len(s) - s = self._dissect_sequence_children(pkt, s) - if len(s) > 0: - raise BER_Decoding_Error( - "unexpected remainder in %s" % pkt.name, - remaining=s, - ) - return [], remain - def m2i(self, pkt, s): # type: (Any, bytes) -> Tuple[Any, bytes] """ ASN1F_SEQUENCE behaves transparently, with nested ASN1_objects being - dissected one by one. m2i returns an empty list (along with the proper - remainder). It is discarded by dissect() and should not be missed - elsewhere. + dissected one by one. Because we use obj.dissect (see loop below) + instead of obj.m2i (as we trust dissect to do the appropriate set_vals) + we do not directly retrieve the list of nested objects. + Thus m2i returns an empty list (along with the proper remainder). + It is discarded by dissect() and should not be missed elsewhere. """ - return self._m2i_ber(pkt, s) + s = self._apply_tagging_dec(s, pkt, _fname=pkt.name) + codec = self.ASN1_tag.get_codec(pkt.ASN1_codec) + i, s, remain = codec.check_type_check_len(s) + if len(s) == 0: + for obj in self.seq: + obj.set_val(pkt, None) + else: + for obj in self.seq: + try: + s = obj.dissect(pkt, s) + except ASN1F_badsequence: + break + if len(s) > 0: + raise BER_Decoding_Error( + "unexpected remainder in %s" % pkt.name, + remaining=s, + ) + return [], remain def dissect(self, pkt, s): # type: (Any, bytes) -> bytes @@ -661,7 +654,8 @@ def build(self, pkt): elif self.holds_packets: s = b"".join(bytes(i) for i in val) else: - # Use i2m so element implicit/explicit tags match m2i()/fld.m2i() + # BER: element fields may carry implicit/explicit tags; i2m + # matches m2i()/fld.m2i(). (Packet elements use bytes() above.) s = b"".join(self.fld.i2m(pkt, i) for i in val) return self.i2m(pkt, s) @@ -747,11 +741,6 @@ def i2repr(self, pkt, x): # type: (ASN1_Packet, Any) -> str return self._field.i2repr(pkt, x) - def set_absent(self, pkt): - # type: (ASN1_Packet) -> None - # Used by codecs that track optionality explicitly (e.g. PER). - self._field.set_val(pkt, None) - class ASN1F_omit(ASN1F_field[None, None]): """ @@ -794,8 +783,6 @@ def __init__(self, name, default, *args, **kwargs): self.default = default self.current_choice = None self.choices = {} # type: Dict[int, _CHOICE_T] - self.choice_order = [] # type: List[int] - self.choice_list = [] # type: List[_CHOICE_T] self.pktchoices = {} for p in args: if hasattr(p, "ASN1_root"): @@ -803,57 +790,23 @@ def __init__(self, name, default, *args, **kwargs): # should be ASN1_Packet if hasattr(p.ASN1_root, "choices"): root = cast(ASN1F_CHOICE, p.ASN1_root) - for k in root.choice_order: - self._register_choice(k, root.choices[k]) + for k, v in root.choices.items(): + # ASN1F_CHOICE recursion + self.choices[k] = v else: - self._register_choice(p.ASN1_root.network_tag, p) + self.choices[p.ASN1_root.network_tag] = p elif hasattr(p, "ASN1_tag"): if isinstance(p, type): # should be ASN1F_field class - self._register_choice(int(p.ASN1_tag), p) + self.choices[int(p.ASN1_tag)] = p else: # should be ASN1F_field instance - self._register_choice(p.network_tag, p) + self.choices[p.network_tag] = p if hasattr(p, "cls"): self.pktchoices[hash(p.cls)] = (p.implicit_tag, p.explicit_tag) # noqa: E501 else: raise ASN1_Error("ASN1F_CHOICE: no tag found for one field") - def _register_choice(self, tag, choice): - # type: (int, _CHOICE_T) -> None - self.choices[tag] = choice - self.choice_order.append(tag) - self.choice_list.append(choice) - - def _dissect_choice_payload(self, pkt, choice, payload): - # type: (ASN1_Packet, _CHOICE_T, bytes) -> Tuple[ASN1_Object[Any], bytes] - if hasattr(choice, "ASN1_root"): - return self.extract_packet(choice, payload, _underlayer=pkt) # type: ignore - if isinstance(choice, type): - return choice(self.name, b"").m2i(pkt, payload) - return choice.m2i(pkt, payload) - - def _m2i_ber(self, pkt, s): - # type: (ASN1_Packet, bytes) -> Tuple[ASN1_Object[Any], bytes] - s = self._apply_tagging_dec(s, pkt) - tag, _ = BER_id_dec(s) - return self._m2i_tagged(pkt, tag, s) - - def _m2i_tagged(self, pkt, tag, payload): - # type: (ASN1_Packet, int, bytes) -> Tuple[ASN1_Object[Any], bytes] - if tag in self.choices: - choice = self.choices[tag] - elif self.flexible_tag: - choice = ASN1F_field - else: - raise ASN1_Error( - "ASN1F_CHOICE: unexpected field in '%s' " - "(tag %s not in possible tags %s)" % ( - self.name, tag, list(self.choices.keys()) - ) - ) - return self._dissect_choice_payload(pkt, choice, payload) - def m2i(self, pkt, s): # type: (ASN1_Packet, bytes) -> Tuple[ASN1_Object[Any], bytes] """ @@ -862,27 +815,28 @@ def m2i(self, pkt, s): """ if len(s) == 0: raise ASN1_Error("ASN1F_CHOICE: got empty string") - return self._m2i_ber(pkt, s) - - def _choice_index_for(self, x): - # type: (Any) -> Optional[int] - for index, choice in enumerate(self.choice_list): - if isinstance(choice, type) and hasattr(choice, "ASN1_root"): - if isinstance(x, choice): - return index - elif hasattr(choice, "ASN1_tag"): - if isinstance(x, ASN1_Object) and x.tag == choice.ASN1_tag: - return index - return None - - def _choice_for_index(self, index): - # type: (int) -> _CHOICE_T - return self.choice_list[index] - - def _choice_tag_for(self, x): - # type: (Any) -> Optional[int] - index = self._choice_index_for(x) - return None if index is None else self.choice_order[index] + s = self._apply_tagging_dec(s, pkt) + tag, _ = BER_id_dec(s) + if tag in self.choices: + choice = self.choices[tag] + else: + if self.flexible_tag: + choice = ASN1F_field + else: + raise ASN1_Error( + "ASN1F_CHOICE: unexpected field in '%s' " + "(tag %s not in possible tags %s)" % ( + self.name, tag, list(self.choices.keys()) + ) + ) + if hasattr(choice, "ASN1_root"): + # we don't want to import ASN1_Packet in this module... + return self.extract_packet(choice, s, _underlayer=pkt) # type: ignore + elif isinstance(choice, type): + return choice(self.name, b"").m2i(pkt, s) + else: + # XXX check properly if this is an ASN1F_PACKET + return choice.m2i(pkt, s) def i2m(self, pkt, x): # type: (ASN1_Packet, Any) -> bytes diff --git a/test/scapy/layers/ber_codec.py b/test/scapy/layers/ber_codec.py index e6939f7a27e..06c5ee53abe 100644 --- a/test/scapy/layers/ber_codec.py +++ b/test/scapy/layers/ber_codec.py @@ -231,6 +231,10 @@ def check_ber_sequence_and_set(): try: long_seq = BERcodec_SEQUENCE.enc(payload) assert long_seq.startswith(b"0\x84") + # Explicit size_len=0 must keep short-form lengths even when the + # LDAP-style default long size is set. + short_seq = BERcodec_SEQUENCE.enc(payload, size_len=0) + assert short_seq[0] == 0x30 and short_seq[1] == len(payload) finally: conf.ASN1_default_long_size = 0 From 6fdafdfd1cb4277cb2f4c441a11863bac1815703 Mon Sep 17 00:00:00 2001 From: Nils Weiss Date: Sat, 1 Aug 2026 12:44:20 +0200 Subject: [PATCH 09/11] apply feedback AI-Assisted: yes (Cursor) --- scapy/asn1/ber.py | 1 - scapy/asn1fields.py | 70 ++--- test/scapy/layers/asn1.uts | 48 ---- test/scapy/layers/ber.uts | 452 +++++++++++++++++++++++++++++++ test/scapy/layers/ber_codec.py | 279 ------------------- test/scapy/layers/ber_packets.py | 231 ---------------- 6 files changed, 475 insertions(+), 606 deletions(-) create mode 100644 test/scapy/layers/ber.uts delete mode 100644 test/scapy/layers/ber_codec.py delete mode 100644 test/scapy/layers/ber_packets.py diff --git a/scapy/asn1/ber.py b/scapy/asn1/ber.py index 95954c701a7..9ad739295ec 100644 --- a/scapy/asn1/ber.py +++ b/scapy/asn1/ber.py @@ -297,7 +297,6 @@ def __new__(cls, class BERcodec_Object(Generic[_K], metaclass=BERcodec_metaclass): codec = ASN1_Codecs.BER tag = ASN1_Class_UNIVERSAL.ANY - skip_tagging = False tagging_enc = staticmethod(BER_tagging_enc) tagging_dec = staticmethod(BER_tagging_dec) diff --git a/scapy/asn1fields.py b/scapy/asn1fields.py index 9233494b023..c883b4f4fbe 100644 --- a/scapy/asn1fields.py +++ b/scapy/asn1fields.py @@ -13,7 +13,6 @@ from functools import reduce from scapy.asn1.asn1 import ( - ASN1Codec, ASN1_BIT_STRING, ASN1_BOOLEAN, ASN1_Class, @@ -29,8 +28,6 @@ from scapy.asn1.ber import ( BER_Decoding_Error, BER_id_dec, - BER_tagging_dec, - BER_tagging_enc, ) from scapy.base_classes import BasePacket from scapy.volatile import ( @@ -130,51 +127,31 @@ def _apply_diff_tag(self, diff_tag): elif self.explicit_tag is not None: self.explicit_tag = diff_tag - def _codec_stem(self, pkt): - # type: (ASN1_Packet) -> type - return cast(ASN1Codec, pkt.ASN1_codec).get_stem() - - def _tagging(self, pkt, s, encode, **kwargs): - # type: (ASN1_Packet, bytes, bool, **Any) -> Any - stem = self._codec_stem(pkt) - if getattr(stem, "skip_tagging", False): - return s if encode else (None, s) - if encode: - fn = getattr(stem, "tagging_enc", BER_tagging_enc) - return cast(bytes, fn(s, **kwargs)) - fn = getattr(stem, "tagging_dec", BER_tagging_dec) - return cast(Tuple[Optional[int], bytes], fn(s, **kwargs)) - def _tagging_dec(self, pkt, s, **kwargs): # type: (ASN1_Packet, bytes, **Any) -> Tuple[Optional[int], bytes] - return cast( - Tuple[Optional[int], bytes], - self._tagging(pkt, s, False, **kwargs), - ) + return pkt.ASN1_codec.get_stem().tagging_dec(s, **kwargs) # type: ignore def _tagging_enc(self, pkt, s, **kwargs): # type: (ASN1_Packet, bytes, **Any) -> bytes - return cast(bytes, self._tagging(pkt, s, True, **kwargs)) - - def _apply_tagging_dec(self, s, pkt, **kwargs): - # type: (bytes, ASN1_Packet, **Any) -> bytes - tag_kwargs = { - "hidden_tag": self.ASN1_tag, - "implicit_tag": self.implicit_tag, - "explicit_tag": self.explicit_tag, - "safe": self.flexible_tag, - } # type: Dict[str, Any] - tag_kwargs.update(kwargs) - diff_tag, s = self._tagging_dec(pkt, s, **tag_kwargs) + return pkt.ASN1_codec.get_stem().tagging_enc(s, **kwargs) # type: ignore + + def _apply_tagging_dec(self, s, pkt, hidden_tag=None, **kwargs): + # type: (bytes, ASN1_Packet, Optional[Any], **Any) -> bytes + # Always pass the field tags; callers may override hidden_tag (PACKET) + # or add decode metadata such as _fname. + if hidden_tag is None: + hidden_tag = self.ASN1_tag + diff_tag, s = self._tagging_dec( + pkt, s, + hidden_tag=hidden_tag, + implicit_tag=self.implicit_tag, + explicit_tag=self.explicit_tag, + safe=self.flexible_tag, + **kwargs, + ) self._apply_diff_tag(diff_tag) return s - def _codec_kwargs(self, size_len=None): - # type: (Optional[int]) -> Dict[str, Any] - return { - "size_len": self.size_len if size_len is None else size_len, - } - def _encode_item(self, pkt, item): # type: (ASN1_Packet, Any) -> bytes """Encode a field value with codec kwargs, without field tagging.""" @@ -201,7 +178,7 @@ def _encode_item(self, pkt, item): # the BER type codec so the universal tag/length are applied. item = item.self_build() codec = self.ASN1_tag.get_codec(pkt.ASN1_codec) - return codec.enc(item, **self._codec_kwargs()) + return codec.enc(item, size_len=self.size_len) def i2repr(self, pkt, x): # type: (ASN1_Packet, _I) -> str @@ -228,7 +205,7 @@ def m2i(self, pkt, s): s = self._apply_tagging_dec(s, pkt, _fname=self.name) codec = self.ASN1_tag.get_codec(pkt.ASN1_codec) dec = codec.safedec if self.flexible_tag else codec.dec - return dec(s, context=self.context, **self._codec_kwargs()) # type: ignore + return dec(s, context=self.context, size_len=self.size_len) # type: ignore def i2m(self, pkt, x): # type: (ASN1_Packet, Union[bytes, _I, _A]) -> bytes @@ -800,10 +777,9 @@ def __init__(self, name, default, *args, **kwargs): # should be ASN1F_field class self.choices[int(p.ASN1_tag)] = p else: - # should be ASN1F_field instance + # should be ASN1F_PACKET instance self.choices[p.network_tag] = p - if hasattr(p, "cls"): - self.pktchoices[hash(p.cls)] = (p.implicit_tag, p.explicit_tag) # noqa: E501 + self.pktchoices[hash(p.cls)] = (p.implicit_tag, p.explicit_tag) # noqa: E501 else: raise ASN1_Error("ASN1F_CHOICE: no tag found for one field") @@ -843,10 +819,10 @@ def i2m(self, pkt, x): if x is None: s = b"" else: + # Use the packet codec for ASN1_Object values; bytes(x) would + # follow conf.ASN1_default_codec instead. if isinstance(x, ASN1_Object): s = x.enc(pkt.ASN1_codec) - elif hasattr(x, "self_build"): - s = cast("ASN1_Packet", x).self_build() else: s = bytes(x) if hash(type(x)) in self.pktchoices: diff --git a/test/scapy/layers/asn1.uts b/test/scapy/layers/asn1.uts index b00caa345db..9fa0bad0f44 100644 --- a/test/scapy/layers/asn1.uts +++ b/test/scapy/layers/asn1.uts @@ -101,51 +101,3 @@ ASN1_UTC_TIME(datetime(2020, 12, 31)).val == "201231000000" ASN1_UTC_TIME(datetime(2020, 12, 31, tzinfo=timezone.utc)).val == "201231000000Z" = UTC datetime construction (offset) ASN1_UTC_TIME(datetime(2020, 12, 31, tzinfo=timezone(timedelta(hours=-23, minutes=-59)))).val == "201231000000-2359" - -+ ASN.1 BER packets and fields -= BER field explicit tag -__import__('test.scapy.layers.ber_packets', fromlist=['check_ber_field_explicit_tag']).check_ber_field_explicit_tag() -= BER field fixed size -__import__('test.scapy.layers.ber_packets', fromlist=['check_ber_field_fixed_size']).check_ber_field_fixed_size() -= BER field optional -__import__('test.scapy.layers.ber_packets', fromlist=['check_ber_field_optional']).check_ber_field_optional() -= BER optional SEQUENCE is_empty -__import__('test.scapy.layers.ber_packets', fromlist=['check_ber_optional_sequence_is_empty']).check_ber_optional_sequence_is_empty() -= BER field sequence of -__import__('test.scapy.layers.ber_packets', fromlist=['check_ber_field_sequence_of']).check_ber_field_sequence_of() -= BER SEQUENCE OF tagged elements -__import__('test.scapy.layers.ber_packets', fromlist=['check_ber_sequence_of_tagged_elements']).check_ber_sequence_of_tagged_elements() -= BER ASN1_Object codec kwargs -__import__('test.scapy.layers.ber_packets', fromlist=['check_ber_asn1_object_codec_kwargs']).check_ber_asn1_object_codec_kwargs() -= BER STRING field packet value -__import__('test.scapy.layers.ber_packets', fromlist=['check_ber_string_field_packet_value']).check_ber_string_field_packet_value() -= BER field choice -__import__('test.scapy.layers.ber_packets', fromlist=['check_ber_field_choice']).check_ber_field_choice() -= BER packet record -__import__('test.scapy.layers.ber_packets', fromlist=['check_ber_packet_record']).check_ber_packet_record() - -+ ASN.1 BER codec -= BER error formatting -__import__('test.scapy.layers.ber_codec', fromlist=['check_ber_error_str']).check_ber_error_str() -= BER length encoding -__import__('test.scapy.layers.ber_codec', fromlist=['check_ber_len_enc_dec']).check_ber_len_enc_dec() -= BER number encoding -__import__('test.scapy.layers.ber_codec', fromlist=['check_ber_num_enc_dec']).check_ber_num_enc_dec() -= BER identifier encoding -__import__('test.scapy.layers.ber_codec', fromlist=['check_ber_id_enc_dec']).check_ber_id_enc_dec() -= BER tagging -__import__('test.scapy.layers.ber_codec', fromlist=['check_ber_tagging']).check_ber_tagging() -= BER integer codec -__import__('test.scapy.layers.ber_codec', fromlist=['check_ber_integer']).check_ber_integer() -= BER bit string codec -__import__('test.scapy.layers.ber_codec', fromlist=['check_ber_bit_string']).check_ber_bit_string() -= BER string and null codec -__import__('test.scapy.layers.ber_codec', fromlist=['check_ber_string_and_null']).check_ber_string_and_null() -= BER OID codec -__import__('test.scapy.layers.ber_codec', fromlist=['check_ber_oid']).check_ber_oid() -= BER sequence and set codec -__import__('test.scapy.layers.ber_codec', fromlist=['check_ber_sequence_and_set']).check_ber_sequence_and_set() -= BER IP address codec -__import__('test.scapy.layers.ber_codec', fromlist=['check_ber_ipaddress']).check_ber_ipaddress() -= BER object dispatch -__import__('test.scapy.layers.ber_codec', fromlist=['check_ber_object_dispatch']).check_ber_object_dispatch() diff --git a/test/scapy/layers/ber.uts b/test/scapy/layers/ber.uts new file mode 100644 index 00000000000..2bd00deda53 --- /dev/null +++ b/test/scapy/layers/ber.uts @@ -0,0 +1,452 @@ +% Tests for ASN.1 BER encoding + +# +# Try me with: +# bash test/run_tests -t test/scapy/layers/ber.uts -F + +########### ASN.1 BER packets and fields ####################################### + ++ ASN.1 BER packets and fields += prepare BER packet classes +class BERTaggedInteger(ASN1_Packet): + ASN1_codec = ASN1_Codecs.BER + ASN1_root = ASN1F_INTEGER("n", 0, explicit_tag=0xA1) + +class BERFixedFields(ASN1_Packet): + ASN1_codec = ASN1_Codecs.BER + ASN1_root = ASN1F_SEQUENCE( + ASN1F_INTEGER("n", 0, size_len=1), + ASN1F_STRING("s", "", size_len=3), + ) + +class BEROptionalField(ASN1_Packet): + ASN1_codec = ASN1_Codecs.BER + ASN1_root = ASN1F_SEQUENCE( + ASN1F_INTEGER("id", 0), + ASN1F_optional(ASN1F_INTEGER("extra", 0, explicit_tag=0xA0)), + ) + +class BERSequenceOfIntegers(ASN1_Packet): + ASN1_codec = ASN1_Codecs.BER + ASN1_root = ASN1F_SEQUENCE_OF("values", [], ASN1F_INTEGER) + +class BERChoiceField(ASN1_Packet): + ASN1_codec = ASN1_Codecs.BER + ASN1_root = ASN1F_CHOICE( + "c", ASN1_INTEGER(0), ASN1F_INTEGER, ASN1F_STRING, + ) + +class BERRecord(ASN1_Packet): + ASN1_codec = ASN1_Codecs.BER + ASN1_root = ASN1F_SEQUENCE( + ASN1F_INTEGER("id", 0), + ASN1F_BOOLEAN("flag", False), + ASN1F_STRING("label", ""), + ASN1F_optional(ASN1F_INTEGER("extra", 0, explicit_tag=0xA0)), + ASN1F_SEQUENCE_OF("values", [], ASN1F_INTEGER), + ) + +class BEROptionalSequence(ASN1_Packet): + ASN1_codec = ASN1_Codecs.BER + ASN1_root = ASN1F_SEQUENCE( + ASN1F_INTEGER("hdr", 0), + ASN1F_optional(ASN1F_SEQUENCE( + ASN1F_INTEGER("id", None), + ASN1F_STRING("label", None), + explicit_tag=0xA0, + )), + ) + +class BERSequenceOfTaggedIntegers(ASN1_Packet): + ASN1_codec = ASN1_Codecs.BER + ASN1_root = ASN1F_SEQUENCE_OF( + "values", [], ASN1F_INTEGER("v", 0, explicit_tag=0xA0), + ) + +class BERSizedInteger(ASN1_Packet): + ASN1_codec = ASN1_Codecs.BER + ASN1_root = ASN1F_INTEGER("n", 0, size_len=1) + += BER field explicit tag +pkt = BERTaggedInteger(n=5) +assert raw(pkt) == b"\xa1\x03\x02\x01\x05" +decoded = BERTaggedInteger(raw(pkt)) +decoded.n.val == 5 + += BER field fixed size +pkt = BERFixedFields(n=200, s=b"ABC") +assert raw(pkt) == bytes.fromhex("300d02810200c80483000003414243") +decoded = BERFixedFields(raw(pkt)) +assert decoded.n.val == 200 +decoded.s.val == b"ABC" + += BER field optional +present = BEROptionalField(id=1, extra=7) +assert raw(present) == bytes.fromhex("3008020101a003020107") +decoded = BEROptionalField(raw(present)) +assert decoded.id.val == 1 +assert decoded.extra.val == 7 +absent = BEROptionalField(id=1, extra=None) +assert raw(absent) == bytes.fromhex("3003020101") +decoded = BEROptionalField(raw(absent)) +assert decoded.id.val == 1 +decoded.extra is None + += BER optional SEQUENCE is_empty +opt = BEROptionalSequence.ASN1_root.seq[1] +present = BEROptionalSequence(hdr=1, id=42, label=b"abc") +assert opt._field.is_empty(present) is False +assert opt.is_empty(present) is False +assert BEROptionalSequence.ASN1_root.is_empty(present) is False +assert raw(present) == bytes.fromhex("300f020101a00a300802012a0403616263") +absent = BEROptionalSequence(hdr=1, id=None, label=None) +assert opt._field.is_empty(absent) is True +assert opt.is_empty(absent) is True +raw(absent) == bytes.fromhex("3003020101") + += BER field sequence of +pkt = BERSequenceOfIntegers(values=[1, 2, 3]) +assert raw(pkt) == b"\x30\x09\x02\x01\x01\x02\x01\x02\x02\x01\x03" +decoded = BERSequenceOfIntegers(raw(pkt)) +[x.val for x in decoded.values] == [1, 2, 3] + += BER SEQUENCE OF tagged elements +pkt = BERSequenceOfTaggedIntegers(values=[1, 2]) +assert raw(pkt) == bytes.fromhex("300aa003020101a003020102") +decoded = BERSequenceOfTaggedIntegers(raw(pkt)) +[x.val for x in decoded.values] == [1, 2] + += BER ASN1_Object codec kwargs +as_int = BERSizedInteger(n=5) +as_obj = BERSizedInteger(n=ASN1_INTEGER(5)) +assert raw(as_int) == raw(as_obj) == b"\x02\x81\x01\x05" +BERSizedInteger(raw(as_obj)).n.val == 5 + += BER STRING field packet value +class Blob(Packet): + fields_desc = [StrFixedLenField("data", b"ABCD", 4)] + +class P(ASN1_Packet): + ASN1_codec = ASN1_Codecs.BER + ASN1_root = ASN1F_STRING("s", "") + +P.ASN1_root.i2m(P(), Blob()) == b"\x04\x04ABCD" + += BER field choice +as_int = BERChoiceField(c=ASN1_INTEGER(99)) +assert raw(as_int) == b"\x02\x01c" +decoded = BERChoiceField(raw(as_int)) +assert decoded.c.val == 99 +as_str = BERChoiceField(c=ASN1_STRING("x")) +assert raw(as_str) == b"\x04\x01x" +decoded = BERChoiceField(raw(as_str)) +decoded.c.val == b"x" + += BER packet record +pkt = BERRecord(id=42, flag=True, label="hi", extra=7, values=[1, 2, 3]) +expected = bytes.fromhex( + "301a02012a01010104026869" + "a003020107" + "3009020101020102020103" +) +assert raw(pkt) == expected +decoded = BERRecord(raw(pkt)) +assert decoded.id.val == 42 +assert decoded.flag.val == 1 +assert decoded.label.val == b"hi" +assert decoded.extra.val == 7 +assert [x.val for x in decoded.values] == [1, 2, 3] +empty = BERRecord(id=1, flag=False, label="", extra=None, values=[]) +assert raw(empty) == bytes.fromhex("300a02010101010004003000") +decoded = BERRecord(raw(empty)) +assert decoded.id.val == 1 +assert decoded.flag.val == 0 +assert decoded.label.val == b"" +assert decoded.extra is None +[x.val for x in decoded.values] == [] + +########### ASN.1 BER codec ####################################### + ++ ASN.1 BER codec += BER error formatting +obj = ASN1_INTEGER(1) +enc_err = BER_Encoding_Error("enc", encoded=obj, remaining=b"rest") +assert "Already encoded" in str(enc_err) +enc_err2 = BER_Encoding_Error("enc", encoded="raw", remaining=b"") +assert "raw" in str(enc_err2) +dec_err = BER_Decoding_Error("dec", decoded=obj, remaining=b"tail") +assert "Already decoded" in str(dec_err) +dec_err2 = BER_Decoding_Error("dec", decoded=[1], remaining=b"") +"[1]" in str(dec_err2) + += BER length encoding +results = [] +for value in [0, 1, 127, 128, 999]: + encoded = BER_len_enc(value) + length, remain = BER_len_dec(encoded) + results.append(length == value and remain == b"") + +assert all(results) +assert BER_len_enc(45, size=None) == BER_len_enc(45, size=0) +BER_len_enc(45, size=4) == b"\x84\x00\x00\x00-" + += BER length encoding errors +try: + BER_len_enc(0, size=128) + False +except BER_Exception: + True + += BER length decoding truncated +try: + BER_len_dec(b"\x82") + False +except BER_Decoding_Error: + True + += BER number encoding +results = [] +for value in [0, 1, 127, 256, 16384]: + encoded = BER_num_enc(value) + decoded, remain = BER_num_dec(encoded) + results.append(decoded == value and remain == b"") + +all(results) + += BER number decoding errors +try: + BER_num_dec(b"") + False +except BER_Decoding_Error: + True + += BER number decoding unfinished +try: + BER_num_dec(b"\x80\x80") + False +except BER_Decoding_Error: + True + += BER identifier encoding +results = [] +for tag in [0x02, 0x30, 0x81, 0xA0]: + encoded = BER_id_enc(tag) + decoded, remain = BER_id_dec(encoded) + results.append(decoded == tag and remain == b"") + +assert all(results) +high_tag = (0x03 << 5) + 0x22 +encoded = BER_id_enc(high_tag) +decoded, remain = BER_id_dec(encoded) +decoded == high_tag and remain == b"" + += BER tagging +inner = BERcodec_INTEGER.enc(7) +implicit = BER_tagging_enc(inner, implicit_tag=0xA0) +assert implicit.startswith(b"\xa0") +real_tag, payload = BER_tagging_dec( + implicit, + hidden_tag=ASN1_Class_UNIVERSAL.INTEGER, + implicit_tag=0xA0, +) +assert real_tag is None +payload[0] == int(ASN1_Class_UNIVERSAL.INTEGER) + += BER tagging with long size +inner = BERcodec_INTEGER.enc(7) +conf.ASN1_default_long_size = 4 +try: + explicit = BER_tagging_enc(inner, explicit_tag=0xA1) + assert explicit.startswith(b"\xa1\x84") + real_tag, payload = BER_tagging_dec(explicit, explicit_tag=0xA1) + assert real_tag is None + assert payload == inner +finally: + conf.ASN1_default_long_size = 0 + += BER tagging mismatch +inner = BERcodec_INTEGER.enc(7) +implicit = BER_tagging_enc(inner, implicit_tag=0xA0) +try: + BER_tagging_dec( + implicit, + hidden_tag=ASN1_Class_UNIVERSAL.INTEGER, + implicit_tag=0xA1, + ) + False +except BER_Decoding_Error: + True + += BER tagging safe mismatch +inner = BERcodec_INTEGER.enc(7) +implicit = BER_tagging_enc(inner, implicit_tag=0xA0) +safe_tag, _ = BER_tagging_dec( + implicit, + hidden_tag=ASN1_Class_UNIVERSAL.INTEGER, + implicit_tag=0xA1, + safe=True, +) +safe_tag == 0xA0 + += BER integer codec +results = [] +for value in [0, 1, 127, 128, 255, -1, -128, -129]: + encoded = BERcodec_INTEGER.enc(value) + obj, remain = BERcodec_INTEGER.do_dec(encoded) + results.append(obj.val == value and remain == b"") + +all(results) + += BER integer bad tag +try: + BERcodec_INTEGER.do_dec(BERcodec_STRING.enc(b"x")) + False +except BER_BadTag_Decoding_Error: + True + += BER integer truncated +try: + BERcodec_INTEGER.check_type_get_len(b"\x02") + False +except BER_Decoding_Error: + True + += BER bit string codec +encoded = BERcodec_BIT_STRING.enc("1011") +obj, remain = BERcodec_BIT_STRING.do_dec(encoded) +assert obj.val == "1011" +assert remain == b"" +padded = BERcodec_BIT_STRING.enc("10110000") +obj2, _ = BERcodec_BIT_STRING.do_dec(padded) +obj2.val == "10110000" + += BER bit string errors +try: + BERcodec_BIT_STRING.do_dec(b"\x03\x01\x08", safe=True) + False +except BER_Decoding_Error: + True + += BER bit string empty +try: + BERcodec_BIT_STRING.do_dec(b"\x03\x00") + False +except BER_Decoding_Error: + True + += BER string and null codec +encoded = BERcodec_STRING.enc(b"hello") +obj, remain = BERcodec_STRING.do_dec(encoded) +assert obj.val == b"hello" +assert remain == b"" +null = BERcodec_NULL.enc(0) +assert null == b"\x05\x00" +obj, remain = BERcodec_NULL.do_dec(null) +assert obj.val == 0 +non_null = BERcodec_NULL.enc(42) +obj, remain = BERcodec_NULL.do_dec(non_null) +obj.val == 42 + += BER OID codec +encoded = BERcodec_OID.enc("1.2.840.113556.1.4.529") +obj, remain = BERcodec_OID.do_dec(encoded) +assert obj.val == "1.2.840.113556.1.4.529" +assert remain == b"" +empty, remain = BERcodec_OID.do_dec(BERcodec_OID.enc("")) +assert empty.val == "" +remain == b"" + += BER sequence and set codec +payload = BERcodec_INTEGER.enc(1) + BERcodec_INTEGER.enc(2) +seq = BERcodec_SEQUENCE.enc(payload) +obj, remain = BERcodec_SEQUENCE.do_dec(seq) +assert len(obj.val) == 2 +assert obj.val[0].val == 1 +assert obj.val[1].val == 2 +assert remain == b"" +as_list = BERcodec_SEQUENCE.enc([ASN1_INTEGER(3), ASN1_INTEGER(4)]) +obj2, remain2 = BERcodec_SEQUENCE.do_dec(as_list) +assert [x.val for x in obj2.val] == [3, 4] +assert remain2 == b"" +st = BERcodec_SET.enc(payload) +obj3, remain3 = BERcodec_SET.do_dec(st) +assert len(obj3.val) == 2 +remain3 == b"" + += BER sequence long size default +payload = BERcodec_INTEGER.enc(1) + BERcodec_INTEGER.enc(2) +conf.ASN1_default_long_size = 4 +try: + long_seq = BERcodec_SEQUENCE.enc(payload) + assert long_seq.startswith(b"0\x84") + short_seq = BERcodec_SEQUENCE.enc(payload, size_len=0) + assert short_seq[0] == 0x30 and short_seq[1] == len(payload) +finally: + conf.ASN1_default_long_size = 0 + += BER sequence truncated +try: + BERcodec_SEQUENCE.do_dec(b"\x30\x05" + BERcodec_INTEGER.enc(1)) + False +except BER_Decoding_Error: + True + += BER IP address codec +encoded = BERcodec_IPADDRESS.enc("192.168.0.1") +obj, remain = BERcodec_IPADDRESS.do_dec(encoded) +assert obj.val == "192.168.0.1" +remain == b"" + += BER IP address encoding error +try: + BERcodec_IPADDRESS.enc("not-an-ip") + False +except BER_Encoding_Error: + True + += BER IP address decoding error +try: + BERcodec_IPADDRESS.do_dec(BERcodec_STRING.enc(b"bad")) + False +except BER_Decoding_Error: + True + += BER object dispatch +encoded = BERcodec_INTEGER.enc(99) +obj, remain = BERcodec_Object.do_dec(encoded) +assert obj.val == 99 +remain == b"" + += BER object empty string +try: + BERcodec_Object.check_string(b"") + False +except BER_Decoding_Error: + True + += BER object unknown tag +try: + BERcodec_Object.do_dec(b"\xff\x00") + False +except BER_Decoding_Error: + True + += BER object safedec +bad, remain = BERcodec_Object.safedec(b"\x02\x01\x01") +assert isinstance(bad, ASN1_INTEGER) +assert bad.val == 1 +unknown, remain = BERcodec_Object.safedec(b"\xff\x00") +assert isinstance(unknown, ASN1_DECODING_ERROR) +truncated, remain = BERcodec_Object.dec(b"\x02\x05\x01", safe=True) +assert isinstance(truncated, ASN1_DECODING_ERROR) +remain == b"" + += BER object enc +try: + BERcodec_Object.enc(object()) + False +except TypeError: + True + += BER object enc string +BERcodec_Object.enc("42") == BERcodec_STRING.enc("42") diff --git a/test/scapy/layers/ber_codec.py b/test/scapy/layers/ber_codec.py deleted file mode 100644 index 06c5ee53abe..00000000000 --- a/test/scapy/layers/ber_codec.py +++ /dev/null @@ -1,279 +0,0 @@ -# SPDX-License-Identifier: GPL-2.0-only -# This file is part of Scapy -# See https://scapy.net/ for more information - -""" -BER codec and helper coverage tests. -""" - -from typing import Any - - -def _raises(exc, func): - # type: (type, Any) -> None - try: - func() - except exc: - return - raise AssertionError("Expected %s" % exc.__name__) - - -from scapy.asn1.asn1 import ( - ASN1_Class_UNIVERSAL, - ASN1_DECODING_ERROR, - ASN1_INTEGER, - ASN1_Object, -) -from scapy.asn1.ber import ( - BER_BadTag_Decoding_Error, - BER_Decoding_Error, - BER_Encoding_Error, - BER_Exception, - BER_id_dec, - BER_id_enc, - BER_len_dec, - BER_len_enc, - BER_num_dec, - BER_num_enc, - BER_tagging_dec, - BER_tagging_enc, - BERcodec_BIT_STRING, - BERcodec_INTEGER, - BERcodec_IPADDRESS, - BERcodec_NULL, - BERcodec_Object, - BERcodec_OID, - BERcodec_SEQUENCE, - BERcodec_SET, - BERcodec_STRING, -) -from scapy.config import conf - - -def check_ber_error_str(): - # type: () -> None - obj = ASN1_INTEGER(1) - enc_err = BER_Encoding_Error("enc", encoded=obj, remaining=b"rest") - assert "Already encoded" in str(enc_err) - enc_err2 = BER_Encoding_Error("enc", encoded="raw", remaining=b"") - assert "raw" in str(enc_err2) - - dec_err = BER_Decoding_Error("dec", decoded=obj, remaining=b"tail") - assert "Already decoded" in str(dec_err) - dec_err2 = BER_Decoding_Error("dec", decoded=[1], remaining=b"") - assert "[1]" in str(dec_err2) - - -def check_ber_len_enc_dec(): - # type: () -> None - for value in [0, 1, 127, 128, 999]: - encoded = BER_len_enc(value) - length, remain = BER_len_dec(encoded) - assert length == value - assert remain == b"" - - assert BER_len_enc(45, size=None) == BER_len_enc(45, size=0) - assert BER_len_enc(45, size=4) == b"\x84\x00\x00\x00-" - - _raises(BER_Exception, lambda: BER_len_enc(0, size=128)) - - _raises(BER_Decoding_Error, lambda: BER_len_dec(b"\x82")) - - -def check_ber_num_enc_dec(): - # type: () -> None - for value in [0, 1, 127, 256, 16384]: - encoded = BER_num_enc(value) - decoded, remain = BER_num_dec(encoded) - assert decoded == value - assert remain == b"" - - _raises(BER_Decoding_Error, lambda: BER_num_dec(b"")) - - _raises(BER_Decoding_Error, lambda: BER_num_dec(b"\x80\x80")) - - -def check_ber_id_enc_dec(): - # type: () -> None - for tag in [0x02, 0x30, 0x81, 0xA0]: - encoded = BER_id_enc(tag) - decoded, remain = BER_id_dec(encoded) - assert decoded == tag - assert remain == b"" - - high_tag = (0x03 << 5) + 0x22 - encoded = BER_id_enc(high_tag) - decoded, remain = BER_id_dec(encoded) - assert decoded == high_tag - assert remain == b"" - - -def check_ber_tagging(): - # type: () -> None - inner = BERcodec_INTEGER.enc(7) - implicit = BER_tagging_enc(inner, implicit_tag=0xA0) - assert implicit.startswith(b"\xa0") - real_tag, payload = BER_tagging_dec( - implicit, - hidden_tag=ASN1_Class_UNIVERSAL.INTEGER, - implicit_tag=0xA0, - ) - assert real_tag is None - assert payload[0] == int(ASN1_Class_UNIVERSAL.INTEGER) - - conf.ASN1_default_long_size = 4 - try: - explicit = BER_tagging_enc(inner, explicit_tag=0xA1) - assert explicit.startswith(b"\xa1\x84") - real_tag, payload = BER_tagging_dec( - explicit, - explicit_tag=0xA1, - ) - assert real_tag is None - assert payload == inner - finally: - conf.ASN1_default_long_size = 0 - - _raises(BER_Decoding_Error, lambda: BER_tagging_dec( - implicit, - hidden_tag=ASN1_Class_UNIVERSAL.INTEGER, - implicit_tag=0xA1, - )) - - safe_tag, _ = BER_tagging_dec( - implicit, - hidden_tag=ASN1_Class_UNIVERSAL.INTEGER, - implicit_tag=0xA1, - safe=True, - ) - assert safe_tag == 0xA0 - - -def check_ber_integer(): - # type: () -> None - for value in [0, 1, 127, 128, 255, -1, -128, -129]: - encoded = BERcodec_INTEGER.enc(value) - obj, remain = BERcodec_INTEGER.do_dec(encoded) - assert obj.val == value - assert remain == b"" - - _raises(BER_BadTag_Decoding_Error, lambda: BERcodec_INTEGER.do_dec(BERcodec_STRING.enc(b"x"))) - - _raises(BER_Decoding_Error, lambda: BERcodec_INTEGER.check_type_get_len(b"\x02")) - - -def check_ber_bit_string(): - # type: () -> None - encoded = BERcodec_BIT_STRING.enc("1011") - obj, remain = BERcodec_BIT_STRING.do_dec(encoded) - assert obj.val == "1011" - assert remain == b"" - - padded = BERcodec_BIT_STRING.enc("10110000") - obj2, _ = BERcodec_BIT_STRING.do_dec(padded) - assert obj2.val == "10110000" - - _raises(BER_Decoding_Error, lambda: BERcodec_BIT_STRING.do_dec(b"\x03\x01\x08", safe=True)) - - _raises(BER_Decoding_Error, lambda: BERcodec_BIT_STRING.do_dec(b"\x03\x00")) - - -def check_ber_string_and_null(): - # type: () -> None - encoded = BERcodec_STRING.enc(b"hello") - obj, remain = BERcodec_STRING.do_dec(encoded) - assert obj.val == b"hello" - assert remain == b"" - - null = BERcodec_NULL.enc(0) - assert null == b"\x05\x00" - obj, remain = BERcodec_NULL.do_dec(null) - assert obj.val == 0 - - non_null = BERcodec_NULL.enc(42) - obj, remain = BERcodec_NULL.do_dec(non_null) - assert obj.val == 42 - - -def check_ber_oid(): - # type: () -> None - encoded = BERcodec_OID.enc("1.2.840.113556.1.4.529") - obj, remain = BERcodec_OID.do_dec(encoded) - assert obj.val == "1.2.840.113556.1.4.529" - assert remain == b"" - - empty, remain = BERcodec_OID.do_dec(BERcodec_OID.enc("")) - assert empty.val == "" - assert remain == b"" - - -def check_ber_sequence_and_set(): - # type: () -> None - payload = BERcodec_INTEGER.enc(1) + BERcodec_INTEGER.enc(2) - seq = BERcodec_SEQUENCE.enc(payload) - obj, remain = BERcodec_SEQUENCE.do_dec(seq) - assert len(obj.val) == 2 - assert obj.val[0].val == 1 - assert obj.val[1].val == 2 - assert remain == b"" - - as_list = BERcodec_SEQUENCE.enc([ASN1_INTEGER(3), ASN1_INTEGER(4)]) - obj2, remain2 = BERcodec_SEQUENCE.do_dec(as_list) - assert [x.val for x in obj2.val] == [3, 4] - assert remain2 == b"" - - st = BERcodec_SET.enc(payload) - obj3, remain3 = BERcodec_SET.do_dec(st) - assert len(obj3.val) == 2 - assert remain3 == b"" - - conf.ASN1_default_long_size = 4 - try: - long_seq = BERcodec_SEQUENCE.enc(payload) - assert long_seq.startswith(b"0\x84") - # Explicit size_len=0 must keep short-form lengths even when the - # LDAP-style default long size is set. - short_seq = BERcodec_SEQUENCE.enc(payload, size_len=0) - assert short_seq[0] == 0x30 and short_seq[1] == len(payload) - finally: - conf.ASN1_default_long_size = 0 - - _raises(BER_Decoding_Error, lambda: BERcodec_SEQUENCE.do_dec(b"\x30\x05" + BERcodec_INTEGER.enc(1))) - - -def check_ber_ipaddress(): - # type: () -> None - encoded = BERcodec_IPADDRESS.enc("192.168.0.1") - obj, remain = BERcodec_IPADDRESS.do_dec(encoded) - assert obj.val == "192.168.0.1" - assert remain == b"" - - _raises(BER_Encoding_Error, lambda: BERcodec_IPADDRESS.enc("not-an-ip")) - - _raises(BER_Decoding_Error, lambda: BERcodec_IPADDRESS.do_dec(BERcodec_STRING.enc(b"bad"))) - - -def check_ber_object_dispatch(): - # type: () -> None - encoded = BERcodec_INTEGER.enc(99) - obj, remain = BERcodec_Object.do_dec(encoded) - assert obj.val == 99 - assert remain == b"" - - _raises(BER_Decoding_Error, lambda: BERcodec_Object.check_string(b"")) - - _raises(BER_Decoding_Error, lambda: BERcodec_Object.do_dec(b"\xff\x00")) - - bad, remain = BERcodec_Object.safedec(b"\x02\x01\x01") - assert isinstance(bad, ASN1_INTEGER) - assert bad.val == 1 - - unknown, remain = BERcodec_Object.safedec(b"\xff\x00") - assert isinstance(unknown, ASN1_DECODING_ERROR) - - truncated, remain = BERcodec_Object.dec(b"\x02\x05\x01", safe=True) - assert isinstance(truncated, ASN1_DECODING_ERROR) - assert remain == b"" - - _raises(TypeError, lambda: BERcodec_Object.enc(object())) - assert BERcodec_Object.enc("42") == BERcodec_STRING.enc("42") diff --git a/test/scapy/layers/ber_packets.py b/test/scapy/layers/ber_packets.py deleted file mode 100644 index ddf68d3dd0a..00000000000 --- a/test/scapy/layers/ber_packets.py +++ /dev/null @@ -1,231 +0,0 @@ -# SPDX-License-Identifier: GPL-2.0-only -# This file is part of Scapy -# See https://scapy.net/ for more information - -""" -BER ASN1_Packet and ASN1F_field build tests. -""" - -from scapy.asn1.asn1 import ASN1_Codecs, ASN1_INTEGER, ASN1_STRING -from scapy.asn1fields import ( - ASN1F_BOOLEAN, - ASN1F_CHOICE, - ASN1F_INTEGER, - ASN1F_SEQUENCE, - ASN1F_SEQUENCE_OF, - ASN1F_STRING, - ASN1F_optional, -) -from scapy.asn1packet import ASN1_Packet -from scapy.packet import raw - - -class BERTaggedInteger(ASN1_Packet): - ASN1_codec = ASN1_Codecs.BER - ASN1_root = ASN1F_INTEGER("n", 0, explicit_tag=0xA1) - - -class BERFixedFields(ASN1_Packet): - ASN1_codec = ASN1_Codecs.BER - ASN1_root = ASN1F_SEQUENCE( - ASN1F_INTEGER("n", 0, size_len=1), - ASN1F_STRING("s", "", size_len=3), - ) - - -class BEROptionalField(ASN1_Packet): - ASN1_codec = ASN1_Codecs.BER - ASN1_root = ASN1F_SEQUENCE( - ASN1F_INTEGER("id", 0), - ASN1F_optional(ASN1F_INTEGER("extra", 0, explicit_tag=0xA0)), - ) - - -class BERSequenceOfIntegers(ASN1_Packet): - ASN1_codec = ASN1_Codecs.BER - ASN1_root = ASN1F_SEQUENCE_OF("values", [], ASN1F_INTEGER) - - -class BERChoiceField(ASN1_Packet): - ASN1_codec = ASN1_Codecs.BER - ASN1_root = ASN1F_CHOICE( - "c", ASN1_INTEGER(0), ASN1F_INTEGER, ASN1F_STRING, - ) - - -class BERRecord(ASN1_Packet): - ASN1_codec = ASN1_Codecs.BER - ASN1_root = ASN1F_SEQUENCE( - ASN1F_INTEGER("id", 0), - ASN1F_BOOLEAN("flag", False), - ASN1F_STRING("label", ""), - ASN1F_optional(ASN1F_INTEGER("extra", 0, explicit_tag=0xA0)), - ASN1F_SEQUENCE_OF("values", [], ASN1F_INTEGER), - ) - - -class BEROptionalSequence(ASN1_Packet): - ASN1_codec = ASN1_Codecs.BER - ASN1_root = ASN1F_SEQUENCE( - ASN1F_INTEGER("hdr", 0), - ASN1F_optional(ASN1F_SEQUENCE( - ASN1F_INTEGER("id", None), - ASN1F_STRING("label", None), - explicit_tag=0xA0, - )), - ) - - -class BERSequenceOfTaggedIntegers(ASN1_Packet): - ASN1_codec = ASN1_Codecs.BER - ASN1_root = ASN1F_SEQUENCE_OF( - "values", [], ASN1F_INTEGER("v", 0, explicit_tag=0xA0), - ) - - -class BERSizedInteger(ASN1_Packet): - ASN1_codec = ASN1_Codecs.BER - ASN1_root = ASN1F_INTEGER("n", 0, size_len=1) - - -def _roundtrip(cls, pkt): - # type: (type, ASN1_Packet) -> ASN1_Packet - return cls(raw(pkt)) - - -def check_ber_field_explicit_tag(): - # type: () -> None - pkt = BERTaggedInteger(n=5) - assert raw(pkt) == b"\xa1\x03\x02\x01\x05" - decoded = _roundtrip(BERTaggedInteger, pkt) - assert decoded.n.val == 5 - - -def check_ber_field_fixed_size(): - # type: () -> None - pkt = BERFixedFields(n=200, s=b"ABC") - assert raw(pkt) == bytes.fromhex("300d02810200c80483000003414243") - decoded = _roundtrip(BERFixedFields, pkt) - assert decoded.n.val == 200 - assert decoded.s.val == b"ABC" - - -def check_ber_field_optional(): - # type: () -> None - present = BEROptionalField(id=1, extra=7) - assert raw(present) == bytes.fromhex("3008020101a003020107") - decoded = _roundtrip(BEROptionalField, present) - assert decoded.id.val == 1 - assert decoded.extra.val == 7 - - absent = BEROptionalField(id=1, extra=None) - assert raw(absent) == bytes.fromhex("3003020101") - decoded = _roundtrip(BEROptionalField, absent) - assert decoded.id.val == 1 - assert decoded.extra is None - - -def check_ber_optional_sequence_is_empty(): - # type: () -> None - """Optional ASN1F_SEQUENCE must use the wrapped field's is_empty(). - - SEQUENCE stores children under their own names (not dummy_seq_name), so - inspecting pkt.dummy_seq_name incorrectly reports present children as empty - and makes the parent SEQUENCE look empty. - """ - opt = BEROptionalSequence.ASN1_root.seq[1] - - present = BEROptionalSequence(hdr=1, id=42, label=b"abc") - assert opt._field.is_empty(present) is False - assert opt.is_empty(present) is False - assert BEROptionalSequence.ASN1_root.is_empty(present) is False - assert raw(present) == bytes.fromhex("300f020101a00a300802012a0403616263") - - absent = BEROptionalSequence(hdr=1, id=None, label=None) - assert opt._field.is_empty(absent) is True - assert opt.is_empty(absent) is True - assert raw(absent) == bytes.fromhex("3003020101") - - -def check_ber_field_sequence_of(): - # type: () -> None - pkt = BERSequenceOfIntegers(values=[1, 2, 3]) - assert raw(pkt) == b"\x30\x09\x02\x01\x01\x02\x01\x02\x02\x01\x03" - decoded = _roundtrip(BERSequenceOfIntegers, pkt) - assert [x.val for x in decoded.values] == [1, 2, 3] - - -def check_ber_sequence_of_tagged_elements(): - # type: () -> None - """SEQUENCE OF must apply the element field's tagging on build.""" - pkt = BERSequenceOfTaggedIntegers(values=[1, 2]) - assert raw(pkt) == bytes.fromhex("300aa003020101a003020102") - decoded = _roundtrip(BERSequenceOfTaggedIntegers, pkt) - assert [x.val for x in decoded.values] == [1, 2] - - -def check_ber_asn1_object_codec_kwargs(): - # type: () -> None - """ASN1_Object values must honor field codec kwargs such as size_len.""" - as_int = BERSizedInteger(n=5) - as_obj = BERSizedInteger(n=ASN1_INTEGER(5)) - assert raw(as_int) == raw(as_obj) == b"\x02\x81\x01\x05" - assert _roundtrip(BERSizedInteger, as_obj).n.val == 5 - - -def check_ber_string_field_packet_value(): - # type: () -> None - """Packet values in ASN1F_STRING must still get a BER universal STRING tag.""" - from scapy.fields import StrFixedLenField - from scapy.packet import Packet - - class Blob(Packet): - fields_desc = [StrFixedLenField("data", b"ABCD", 4)] - - class P(ASN1_Packet): - ASN1_codec = ASN1_Codecs.BER - ASN1_root = ASN1F_STRING("s", "") - - enc = P.ASN1_root.i2m(P(), Blob()) - assert enc == b"\x04\x04ABCD" - - -def check_ber_field_choice(): - # type: () -> None - as_int = BERChoiceField(c=ASN1_INTEGER(99)) - assert raw(as_int) == b"\x02\x01c" - decoded = _roundtrip(BERChoiceField, as_int) - assert decoded.c.val == 99 - - as_str = BERChoiceField(c=ASN1_STRING("x")) - assert raw(as_str) == b"\x04\x01x" - decoded = _roundtrip(BERChoiceField, as_str) - assert decoded.c.val == b"x" - - -def check_ber_packet_record(): - # type: () -> None - pkt = BERRecord( - id=42, flag=True, label="hi", extra=7, values=[1, 2, 3], - ) - expected = bytes.fromhex( - "301a02012a01010104026869" - "a003020107" - "3009020101020102020103" - ) - assert raw(pkt) == expected - decoded = _roundtrip(BERRecord, pkt) - assert decoded.id.val == 42 - assert decoded.flag.val == 1 - assert decoded.label.val == b"hi" - assert decoded.extra.val == 7 - assert [x.val for x in decoded.values] == [1, 2, 3] - - empty = BERRecord(id=1, flag=False, label="", extra=None, values=[]) - assert raw(empty) == bytes.fromhex("300a02010101010004003000") - decoded = _roundtrip(BERRecord, empty) - assert decoded.id.val == 1 - assert decoded.flag.val == 0 - assert decoded.label.val == b"" - assert decoded.extra is None - assert [x.val for x in decoded.values] == [] From 915521b8b3dcaf963d22eb2fa2222fcd5519633a Mon Sep 17 00:00:00 2001 From: Nils Weiss Date: Sat, 1 Aug 2026 12:52:05 +0200 Subject: [PATCH 10/11] Make codecs UPER and OER ready AI-Assisted: yes (Cursor) --- scapy/asn1/ber.py | 34 ++++++++++++------------ scapy/asn1fields.py | 23 ++++++++++++----- test/scapy/layers/ber.uts | 54 +++++++++++++++++++++++++++++++++++++++ 3 files changed, 89 insertions(+), 22 deletions(-) diff --git a/scapy/asn1/ber.py b/scapy/asn1/ber.py index 9ad739295ec..1c4362414b2 100644 --- a/scapy/asn1/ber.py +++ b/scapy/asn1/ber.py @@ -412,8 +412,10 @@ def safedec(cls, return cls.dec(s, context, safe=True, _depth=_depth) @classmethod - def enc(cls, s, size_len=0): - # type: (_K, Optional[int]) -> bytes + def enc(cls, s, size_len=0, **_kwargs): + # type: (_K, Optional[int], **Any) -> bytes + # Ignore unknown kwargs so shared field._codec_kwargs() dicts (OER/UPER + # keys) do not TypeError on BER packets. if isinstance(s, (str, bytes)): return BERcodec_STRING.enc(s, size_len=size_len) else: @@ -434,8 +436,8 @@ class BERcodec_INTEGER(BERcodec_Object[int]): tag = ASN1_Class_UNIVERSAL.INTEGER @classmethod - def enc(cls, i, size_len=0): - # type: (int, Optional[int]) -> bytes + def enc(cls, i, size_len=0, **_kwargs): + # type: (int, Optional[int], **Any) -> bytes ls = [] while True: ls.append(i & 0xff) @@ -507,8 +509,8 @@ def do_dec(cls, ) @classmethod - def enc(cls, _s, size_len=0): - # type: (AnyStr, Optional[int]) -> bytes + def enc(cls, _s, size_len=0, **_kwargs): + # type: (AnyStr, Optional[int], **Any) -> bytes # /!\ this is DER encoding (bit strings are only zero-bit padded) s = bytes_encode(_s) if len(s) % 8 == 0: @@ -526,8 +528,8 @@ class BERcodec_STRING(BERcodec_Object[str]): tag = ASN1_Class_UNIVERSAL.STRING @classmethod - def enc(cls, _s, size_len=0): - # type: (Union[str, bytes], Optional[int]) -> bytes + def enc(cls, _s, size_len=0, **_kwargs): + # type: (Union[str, bytes], Optional[int], **Any) -> bytes s = bytes_encode(_s) # Be sure we are encoding bytes return chb(int(cls.tag)) + BER_len_enc(len(s), size=size_len) + s @@ -548,8 +550,8 @@ class BERcodec_NULL(BERcodec_INTEGER): tag = ASN1_Class_UNIVERSAL.NULL @classmethod - def enc(cls, i, size_len=0): - # type: (int, Optional[int]) -> bytes + def enc(cls, i, size_len=0, **_kwargs): + # type: (int, Optional[int], **Any) -> bytes if i == 0: return chb(int(cls.tag)) + b"\0" else: @@ -560,8 +562,8 @@ class BERcodec_OID(BERcodec_Object[bytes]): tag = ASN1_Class_UNIVERSAL.OID @classmethod - def enc(cls, _oid, size_len=0): - # type: (AnyStr, Optional[int]) -> bytes + def enc(cls, _oid, size_len=0, **_kwargs): + # type: (AnyStr, Optional[int], **Any) -> bytes oid = bytes_encode(_oid) if oid: lst = [int(x) for x in oid.strip(b".").split(b".")] @@ -652,8 +654,8 @@ class BERcodec_SEQUENCE(BERcodec_Object[Union[bytes, List[BERcodec_Object[Any]]] tag = ASN1_Class_UNIVERSAL.SEQUENCE @classmethod - def enc(cls, _ll, size_len=None): - # type: (Union[bytes, List[BERcodec_Object[Any]]], Optional[int]) -> bytes + def enc(cls, _ll, size_len=None, **_kwargs): + # type: (Union[bytes, List[BERcodec_Object[Any]]], Optional[int], **Any) -> bytes if isinstance(_ll, bytes): ll = _ll else: @@ -708,8 +710,8 @@ class BERcodec_IPADDRESS(BERcodec_STRING): tag = ASN1_Class_UNIVERSAL.IPADDRESS @classmethod - def enc(cls, ipaddr_ascii, size_len=0): # type: ignore - # type: (str, Optional[int]) -> bytes + def enc(cls, ipaddr_ascii, size_len=0, **_kwargs): # type: ignore + # type: (str, Optional[int], **Any) -> bytes try: s = inet_aton(ipaddr_ascii) except Exception: diff --git a/scapy/asn1fields.py b/scapy/asn1fields.py index c883b4f4fbe..b3a1fd52752 100644 --- a/scapy/asn1fields.py +++ b/scapy/asn1fields.py @@ -129,6 +129,8 @@ def _apply_diff_tag(self, diff_tag): def _tagging_dec(self, pkt, s, **kwargs): # type: (ASN1_Packet, bytes, **Any) -> Tuple[Optional[int], bytes] + # Stem must provide tagging_*; OER implements real tags, UPER/PER use + # identity helpers (no BER-style tagging). return pkt.ASN1_codec.get_stem().tagging_dec(s, **kwargs) # type: ignore def _tagging_enc(self, pkt, s, **kwargs): @@ -152,6 +154,18 @@ def _apply_tagging_dec(self, s, pkt, hidden_tag=None, **kwargs): self._apply_diff_tag(diff_tag) return s + def _codec_kwargs(self, pkt): + # type: (ASN1_Packet) -> Dict[str, Any] + # OER/UPER need extra constraints (oer_unsigned, uper_min/max, …) on + # every enc/dec call; override this instead of hardcoding BER size_len. + return {"size_len": self.size_len} + + def _use_object_enc(self, pkt, item): + # type: (ASN1_Packet, ASN1_Object[Any]) -> bool + # BER/LDAP: item.enc() when size_len is unset. UPER must override to + # False so constrained integers go through codec.enc(**kwargs). + return self.size_len is None + def _encode_item(self, pkt, item): # type: (ASN1_Packet, Any) -> bytes """Encode a field value with codec kwargs, without field tagging.""" @@ -167,10 +181,7 @@ def _encode_item(self, pkt, item): "Encoding Error: got %r instead of an %r for field [%s]" % (item, self.ASN1_tag, self.name) ) - # Without an explicit field size_len, keep ASN1_Object.enc() so - # conf.ASN1_default_long_size only affects SEQUENCE/SET (via their - # bytes payload path) and explicit tagging — Microsoft LDAP style. - if self.size_len is None: + if self._use_object_enc(pkt, item): return item.enc(pkt.ASN1_codec) item = item.val elif hasattr(item, "self_build"): @@ -178,7 +189,7 @@ def _encode_item(self, pkt, item): # the BER type codec so the universal tag/length are applied. item = item.self_build() codec = self.ASN1_tag.get_codec(pkt.ASN1_codec) - return codec.enc(item, size_len=self.size_len) + return codec.enc(item, **self._codec_kwargs(pkt)) def i2repr(self, pkt, x): # type: (ASN1_Packet, _I) -> str @@ -205,7 +216,7 @@ def m2i(self, pkt, s): s = self._apply_tagging_dec(s, pkt, _fname=self.name) codec = self.ASN1_tag.get_codec(pkt.ASN1_codec) dec = codec.safedec if self.flexible_tag else codec.dec - return dec(s, context=self.context, size_len=self.size_len) # type: ignore + return dec(s, context=self.context, **self._codec_kwargs(pkt)) # type: ignore def i2m(self, pkt, x): # type: (ASN1_Packet, Union[bytes, _I, _A]) -> bytes diff --git a/test/scapy/layers/ber.uts b/test/scapy/layers/ber.uts index 2bd00deda53..841c9eb1fbd 100644 --- a/test/scapy/layers/ber.uts +++ b/test/scapy/layers/ber.uts @@ -450,3 +450,57 @@ except TypeError: = BER object enc string BERcodec_Object.enc("42") == BERcodec_STRING.enc("42") + += BER enc ignores extra codec kwargs +assert BERcodec_INTEGER.enc(5, oer_unsigned=True, uper_min=0) == BERcodec_INTEGER.enc(5) +assert BERcodec_STRING.enc(b"x", uper_max=10) == BERcodec_STRING.enc(b"x") +BERcodec_SEQUENCE.enc(BERcodec_INTEGER.enc(1), uper_min=0) == BERcodec_SEQUENCE.enc(BERcodec_INTEGER.enc(1)) + ++ ASN.1 codec stem contract += identity tagging stem for PER-style codecs +def _id_tagging_enc(s, **kwargs): + return s + +def _id_tagging_dec(s, **kwargs): + return None, s + +class FakeStem(object): + tagging_enc = staticmethod(_id_tagging_enc) + tagging_dec = staticmethod(_id_tagging_dec) + +stem = FakeStem() +assert stem.tagging_enc(b"\x02\x01\x05", implicit_tag=0xA0) == b"\x02\x01\x05" +diff, payload = stem.tagging_dec(b"\x02\x01\x05", hidden_tag=2, explicit_tag=0xA1) +diff is None and payload == b"\x02\x01\x05" + += field _codec_kwargs and object-enc hooks +class P(ASN1_Packet): + ASN1_codec = ASN1_Codecs.BER + ASN1_root = ASN1F_INTEGER("n", 0) + +fld = P.ASN1_root +assert fld._codec_kwargs(P()) == {"size_len": None} +assert fld._use_object_enc(P(), ASN1_INTEGER(5)) is True +assert raw(P(n=ASN1_INTEGER(5))) == b"\x02\x01\x05" +assert raw(P(n=5)) == b"\x02\x01\x05" + +class Sized(ASN1_Packet): + ASN1_codec = ASN1_Codecs.BER + ASN1_root = ASN1F_INTEGER("n", 0, size_len=1) + +sfld = Sized.ASN1_root +assert sfld._use_object_enc(Sized(), ASN1_INTEGER(5)) is False +assert raw(Sized(n=ASN1_INTEGER(5))) == raw(Sized(n=5)) == b"\x02\x81\x01\x05" + += field encode with extra kwargs via _codec_kwargs override +class ExtraKwField(ASN1F_INTEGER): + def _codec_kwargs(self, pkt): + return {"size_len": self.size_len, "oer_unsigned": True, "uper_min": 0} + +class ExtraPkt(ASN1_Packet): + ASN1_codec = ASN1_Codecs.BER + ASN1_root = ExtraKwField("n", 0) + +# BER enc swallows unknown kwargs; round-trip still works. +assert raw(ExtraPkt(n=7)) == b"\x02\x01\x07" +ExtraPkt(raw(ExtraPkt(n=7))).n.val == 7 From 208f0298d181a618185f085dca020c87b89bdda3 Mon Sep 17 00:00:00 2001 From: Nils Weiss Date: Sat, 1 Aug 2026 13:02:14 +0200 Subject: [PATCH 11/11] fix mypy and flake8 AI-Assisted: no --- .config/codespell_ignore.txt | 1 + scapy/asn1/ber.py | 16 ++++++++-------- 2 files changed, 9 insertions(+), 8 deletions(-) diff --git a/.config/codespell_ignore.txt b/.config/codespell_ignore.txt index 510f3e0c090..e5d65af4708 100644 --- a/.config/codespell_ignore.txt +++ b/.config/codespell_ignore.txt @@ -48,6 +48,7 @@ temporaere tim ue uint +uper vas wan wanna diff --git a/scapy/asn1/ber.py b/scapy/asn1/ber.py index 1c4362414b2..410829bb16e 100644 --- a/scapy/asn1/ber.py +++ b/scapy/asn1/ber.py @@ -436,7 +436,7 @@ class BERcodec_INTEGER(BERcodec_Object[int]): tag = ASN1_Class_UNIVERSAL.INTEGER @classmethod - def enc(cls, i, size_len=0, **_kwargs): + def enc(cls, i, size_len=0, **_kwargs): # type: ignore[override] # type: (int, Optional[int], **Any) -> bytes ls = [] while True: @@ -509,7 +509,7 @@ def do_dec(cls, ) @classmethod - def enc(cls, _s, size_len=0, **_kwargs): + def enc(cls, _s, size_len=0, **_kwargs): # type: ignore[override] # type: (AnyStr, Optional[int], **Any) -> bytes # /!\ this is DER encoding (bit strings are only zero-bit padded) s = bytes_encode(_s) @@ -528,7 +528,7 @@ class BERcodec_STRING(BERcodec_Object[str]): tag = ASN1_Class_UNIVERSAL.STRING @classmethod - def enc(cls, _s, size_len=0, **_kwargs): + def enc(cls, _s, size_len=0, **_kwargs): # type: ignore[override] # type: (Union[str, bytes], Optional[int], **Any) -> bytes s = bytes_encode(_s) # Be sure we are encoding bytes @@ -550,7 +550,7 @@ class BERcodec_NULL(BERcodec_INTEGER): tag = ASN1_Class_UNIVERSAL.NULL @classmethod - def enc(cls, i, size_len=0, **_kwargs): + def enc(cls, i, size_len=0, **_kwargs): # type: ignore[override] # type: (int, Optional[int], **Any) -> bytes if i == 0: return chb(int(cls.tag)) + b"\0" @@ -562,7 +562,7 @@ class BERcodec_OID(BERcodec_Object[bytes]): tag = ASN1_Class_UNIVERSAL.OID @classmethod - def enc(cls, _oid, size_len=0, **_kwargs): + def enc(cls, _oid, size_len=0, **_kwargs): # type: ignore[override] # type: (AnyStr, Optional[int], **Any) -> bytes oid = bytes_encode(_oid) if oid: @@ -654,8 +654,8 @@ class BERcodec_SEQUENCE(BERcodec_Object[Union[bytes, List[BERcodec_Object[Any]]] tag = ASN1_Class_UNIVERSAL.SEQUENCE @classmethod - def enc(cls, _ll, size_len=None, **_kwargs): - # type: (Union[bytes, List[BERcodec_Object[Any]]], Optional[int], **Any) -> bytes + def enc(cls, _ll, size_len=None, **_kwargs): # type: ignore[override] + # type: (Union[bytes, List[BERcodec_Object[Any]]], Optional[int], **Any) -> bytes # noqa: E501 if isinstance(_ll, bytes): ll = _ll else: @@ -710,7 +710,7 @@ class BERcodec_IPADDRESS(BERcodec_STRING): tag = ASN1_Class_UNIVERSAL.IPADDRESS @classmethod - def enc(cls, ipaddr_ascii, size_len=0, **_kwargs): # type: ignore + def enc(cls, ipaddr_ascii, size_len=0, **_kwargs): # type: ignore[override] # type: (str, Optional[int], **Any) -> bytes try: s = inet_aton(ipaddr_ascii)