diff --git a/scapy/asn1/ber.py b/scapy/asn1/ber.py index 23899274e4a..294c4ec0ee9 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 @@ -261,7 +262,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 ] # @@ -289,6 +292,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): @@ -367,6 +373,7 @@ def dec(cls, s, # type: bytes context=None, # type: Optional[Type[ASN1_Class]] safe=False, # type: bool + **_kwargs # type: Any ): # type: (...) -> Tuple[Union[_ASN1_ERROR, ASN1_Object[_K]], bytes] if not safe: @@ -387,6 +394,7 @@ def dec(cls, def safedec(cls, s, # type: bytes context=None, # type: Optional[Type[ASN1_Class]] + **_kwargs # type: Any ): # type: (...) -> Tuple[Union[_ASN1_ERROR, ASN1_Object[_K]], bytes] return cls.dec(s, context, safe=True) @@ -633,6 +641,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 9a5284bddfc..cf4d49c4766 100644 --- a/scapy/asn1fields.py +++ b/scapy/asn1fields.py @@ -13,10 +13,12 @@ from functools import reduce from scapy.asn1.asn1 import ( + ASN1Codec, ASN1_BIT_STRING, ASN1_BOOLEAN, ASN1_Class, ASN1_Class_UNIVERSAL, + ASN1_Decoding_Error, ASN1_Error, ASN1_INTEGER, ASN1_NULL, @@ -119,6 +121,88 @@ 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 _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), + ) + + 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) + 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.""" + if item is None: + return b"" + 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) + ) + # 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 + # 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()) + def i2repr(self, pkt, x): # type: (ASN1_Packet, _I) -> str return repr(x) @@ -141,40 +225,21 @@ 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 + 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) # type: ignore - else: - return codec.dec(s, context=self.context) # 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 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, size_len=self.size_len) - return BER_tagging_enc(s, - implicit_tag=self.implicit_tag, - explicit_tag=self.explicit_tag) + s = self._encode_item(pkt, x) + 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,40 +526,38 @@ 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", 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. 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. """ - 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", remaining=s) - return [], remain + return self._m2i_ber(pkt, s) def dissect(self, pkt, s): # type: (Any, bytes) -> bytes @@ -569,15 +632,7 @@ 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 + 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 = [] @@ -597,8 +652,11 @@ 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: + # 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): @@ -657,7 +715,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 @@ -665,7 +723,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 @@ -683,6 +741,11 @@ 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]): """ @@ -725,6 +788,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"): @@ -732,22 +797,57 @@ 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") + 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] """ @@ -756,42 +856,47 @@ 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_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] 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 @@ -843,16 +948,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 = 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 + s = self._apply_tagging_dec( + s, pkt, + hidden_tag=cls.ASN1_root.ASN1_tag, # noqa: E501 + _fname=self.name, + ) if not s: return None, s return self.extract_packet(cls, s, _underlayer=pkt) @@ -876,9 +976,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..b00caa345db 100644 --- a/test/scapy/layers/asn1.uts +++ b/test/scapy/layers/asn1.uts @@ -101,3 +101,51 @@ 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_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..ddf68d3dd0a --- /dev/null +++ b/test/scapy/layers/ber_packets.py @@ -0,0 +1,231 @@ +# 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] == []