asn1: prepare fields for pluggable codecs

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)
@@ -627,12 +635,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):
+    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)
+        # 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

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
@@ -471,16 +536,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:
@@ -569,15 +625,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 +645,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):
@@ -657,15 +709,15 @@ 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
 
     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
 
@@ -744,7 +796,8 @@ def __init__(self, name, default, *args, **kwargs):
                 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
+                    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")
 
@@ -756,8 +809,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]
@@ -785,13 +837,20 @@ def i2m(self, pkt, x):
         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 +902,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 +930,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