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 b5ffc4252b7..410829bb16e 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 ] # @@ -294,6 +297,8 @@ def __new__(cls, class BERcodec_Object(Generic[_K], metaclass=BERcodec_metaclass): codec = ASN1_Codecs.BER tag = ASN1_Class_UNIVERSAL.ANY + tagging_enc = staticmethod(BER_tagging_enc) + tagging_dec = staticmethod(BER_tagging_dec) @classmethod def asn1_object(cls, val): @@ -374,6 +379,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,13 +406,16 @@ 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) @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: @@ -427,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: ignore[override] + # type: (int, Optional[int], **Any) -> bytes ls = [] while True: ls.append(i & 0xff) @@ -500,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: ignore[override] + # 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: @@ -519,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: ignore[override] + # 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 @@ -541,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: ignore[override] + # type: (int, Optional[int], **Any) -> bytes if i == 0: return chb(int(cls.tag)) + b"\0" else: @@ -553,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: ignore[override] + # type: (AnyStr, Optional[int], **Any) -> bytes oid = bytes_encode(_oid) if oid: lst = [int(x) for x in oid.strip(b".").split(b".")] @@ -645,12 +654,15 @@ class BERcodec_SEQUENCE(BERcodec_Object[Union[bytes, List[BERcodec_Object[Any]]] tag = ASN1_Class_UNIVERSAL.SEQUENCE @classmethod - def enc(cls, _ll, size_len=0): - # type: (Union[bytes, List[BERcodec_Object[Any]]], Optional[int]) -> 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: ll = b"".join(x.enc(cls.codec) for x in _ll) + # 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 @classmethod @@ -698,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[override] + # 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 7895d7aa1bb..b3a1fd52752 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, @@ -27,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 ( @@ -119,6 +118,79 @@ 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 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): + # type: (ASN1_Packet, bytes, **Any) -> bytes + 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, 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.""" + 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) + ) + if self._use_object_enc(pkt, item): + 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(pkt)) + def i2repr(self, pkt, x): # type: (ASN1_Packet, _I) -> str return repr(x) @@ -141,40 +213,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(pkt)) # 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 @@ -471,16 +524,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 + 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: @@ -572,15 +616,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 = [] @@ -603,8 +639,12 @@ 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: + # 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) def i2repr(self, pkt, x): @@ -663,7 +703,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 +711,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 @@ -748,7 +788,7 @@ 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 self.pktchoices[hash(p.cls)] = (p.implicit_tag, p.explicit_tag) # noqa: E501 else: @@ -762,8 +802,7 @@ 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) + s = self._apply_tagging_dec(s, pkt) tag, _ = BER_id_dec(s) if tag in self.choices: choice = self.choices[tag] @@ -791,13 +830,20 @@ def i2m(self, pkt, x): if x is None: s = b"" else: - s = bytes(x) + # 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) + 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 +895,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) @@ -882,9 +923,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/ber.uts b/test/scapy/layers/ber.uts new file mode 100644 index 00000000000..841c9eb1fbd --- /dev/null +++ b/test/scapy/layers/ber.uts @@ -0,0 +1,506 @@ +% 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") + += 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