Asn1 oer uper contrib

.config/codespell_ignore.txt

@@ -54,3 +54,5 @@ wan
 wanna
 webp
 widgits
+UPER
+uPER

scapy/asn1/asn1.py

@@ -28,15 +28,11 @@
     Type,
     Union,
     cast,
-    TYPE_CHECKING,
 )
 from typing import (
     TypeVar,
 )
 
-if TYPE_CHECKING:
-    from scapy.asn1.ber import BERcodec_Object
-
 try:
     from datetime import timezone
 except ImportError:
@@ -110,35 +106,96 @@ class ASN1_Error(Scapy_Exception):
 
 
 class ASN1_Encoding_Error(ASN1_Error):
-    pass
+    def __init__(self,
+                 msg,  # type: str
+                 encoded=None,  # type: Any
+                 remaining=b""  # type: bytes
+                 ):
+        # type: (...) -> None
+        Scapy_Exception.__init__(self, msg)
+        self.remaining = remaining
+        self.encoded = encoded
+
+    def __str__(self):
+        # type: () -> str
+        s = Scapy_Exception.__str__(self)
+        if self.encoded is not None:
+            if isinstance(self.encoded, ASN1_Object):
+                s += "\n### Already encoded ###\n%s" % self.encoded.strshow()
+            else:
+                s += "\n### Already encoded ###\n%r" % self.encoded
+        if self.remaining is not None:
+            s += "\n### Remaining ###\n%r" % self.remaining
+        return s
 
 
 class ASN1_Decoding_Error(ASN1_Error):
-    pass
+    def __init__(self,
+                 msg,  # type: str
+                 decoded=None,  # type: Any
+                 remaining=b""  # type: bytes
+                 ):
+        # type: (...) -> None
+        Scapy_Exception.__init__(self, msg)
+        self.remaining = remaining
+        self.decoded = decoded
+
+    def __str__(self):
+        # type: () -> str
+        s = Scapy_Exception.__str__(self)
+        if self.decoded is not None:
+            if isinstance(self.decoded, ASN1_Object):
+                s += "\n### Already decoded ###\n%s" % self.decoded.strshow()
+            else:
+                s += "\n### Already decoded ###\n%r" % self.decoded
+        if self.remaining is not None:
+            s += "\n### Remaining ###\n%r" % self.remaining
+        return s
 
 
 class ASN1_BadTag_Decoding_Error(ASN1_Decoding_Error):
     pass
 
 
+class ASN1Codec_metaclass(type):
+    def __new__(cls,
+                name,  # type: str
+                bases,  # type: Tuple[type, ...]
+                dct  # type: Dict[str, Any]
+                ):
+        # type: (...) -> type
+        c = super(ASN1Codec_metaclass, cls).__new__(cls, name, bases, dct)
+        try:
+            c.tag.register(c.codec, c)  # type: ignore
+        except Exception:
+            warning("Error registering %r for %r" % (c.tag, c.codec))  # type: ignore
+        return c
+
+
 class ASN1Codec(EnumElement):
     def register_stem(cls, stem):
-        # type: (Type[BERcodec_Object[Any]]) -> None
+        # type: (Type[Any]) -> None
         cls._stem = stem
 
     def register_tagging(cls, enc, dec):
         # type: (Any, Any) -> None
-        # Codec-level implicit/explicit tagging (BER/OER) or identity (UPER/PER).
+        # Codec-level implicit/explicit tagging (BER) or identity (OER/PER).
         cls._tagging_enc = enc
         cls._tagging_dec = dec
 
     def tagging_enc(cls, s, **kwargs):
         # type: (bytes, **Any) -> bytes
-        return cls._tagging_enc(s, **kwargs)  # type: ignore
+        enc = getattr(cls, "_tagging_enc", None)
+        if enc is None:
+            return s
+        return cast(bytes, enc(s, **kwargs))
 
     def tagging_dec(cls, s, **kwargs):
         # type: (bytes, **Any) -> Tuple[Optional[int], bytes]
-        return cls._tagging_dec(s, **kwargs)  # type: ignore
+        dec = getattr(cls, "_tagging_dec", None)
+        if dec is None:
+            return None, s
+        return cast(Tuple[Optional[int], bytes], dec(s, **kwargs))
 
     def dec(cls, s, context=None, _depth=0):
         # type: (bytes, Optional[Type[ASN1_Class]], int) -> ASN1_Object[Any]
@@ -174,7 +231,7 @@ def __init__(self,
                  key,  # type: str
                  value,  # type: int
                  context=None,  # type: Optional[Type[ASN1_Class]]
-                 codec=None  # type: Optional[Dict[ASN1Codec, Type[BERcodec_Object[Any]]]]  # noqa: E501
+                 codec=None  # type: Optional[Dict[ASN1Codec, Type[Any]]]
                  ):
         # type: (...) -> None
         EnumElement.__init__(self, key, value)
@@ -199,11 +256,11 @@ def asn1_object(self, val):
         raise ASN1_Error("%r does not have any assigned ASN1 object" % self)
 
     def register(self, codecnum, codec):
-        # type: (ASN1Codec, Type[BERcodec_Object[Any]]) -> None
+        # type: (ASN1Codec, Type[Any]) -> None
         self._codec[codecnum] = codec
 
     def get_codec(self, codec):
-        # type: (Any) -> Type[BERcodec_Object[Any]]
+        # type: (Any) -> Type[Any]
         try:
             c = self._codec[codec]
         except KeyError:
@@ -322,7 +379,7 @@ def __init__(self, val):
 
     def enc(self, codec):
         # type: (Any) -> bytes
-        return self.tag.get_codec(codec).enc(self.val)
+        return cast(bytes, self.tag.get_codec(codec).enc(self.val))
 
     def __repr__(self):
         # type: () -> str

scapy/asn1/ber.py

@@ -12,10 +12,10 @@
 # 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, bytes_encode
 from scapy.utils import binrepr, inet_aton, inet_ntoa
 from scapy.asn1.asn1 import (
+    ASN1Codec_metaclass,
     ASN1Tag,
     ASN1_BADTAG,
     ASN1_BadTag_Decoding_Error,
@@ -33,7 +33,6 @@
 from typing import (
     Any,
     AnyStr,
-    Dict,
     Generic,
     List,
     Optional,
@@ -59,47 +58,11 @@ class BER_Exception(Exception):
 
 
 class BER_Encoding_Error(ASN1_Encoding_Error):
-    def __init__(self,
-                 msg,  # type: str
-                 encoded=None,  # type: Optional[Union[BERcodec_Object[Any], str]]  # noqa: E501
-                 remaining=b""  # type: bytes
-                 ):
-        # type: (...) -> None
-        Exception.__init__(self, msg)
-        self.remaining = remaining
-        self.encoded = encoded
-
-    def __str__(self):
-        # type: () -> str
-        s = Exception.__str__(self)
-        if isinstance(self.encoded, ASN1_Object):
-            s += "\n### Already encoded ###\n%s" % self.encoded.strshow()
-        else:
-            s += "\n### Already encoded ###\n%r" % self.encoded
-        s += "\n### Remaining ###\n%r" % self.remaining
-        return s
+    pass
 
 
 class BER_Decoding_Error(ASN1_Decoding_Error):
-    def __init__(self,
-                 msg,  # type: str
-                 decoded=None,  # type: Optional[Any]
-                 remaining=b""  # type: bytes
-                 ):
-        # type: (...) -> None
-        Exception.__init__(self, msg)
-        self.remaining = remaining
-        self.decoded = decoded
-
-    def __str__(self):
-        # type: () -> str
-        s = Exception.__str__(self)
-        if isinstance(self.decoded, ASN1_Object):
-            s += "\n### Already decoded ###\n%s" % self.decoded.strshow()
-        else:
-            s += "\n### Already decoded ###\n%r" % self.decoded
-        s += "\n### Remaining ###\n%r" % self.remaining
-        return s
+    pass
 
 
 class BER_BadTag_Decoding_Error(BER_Decoding_Error,
@@ -164,7 +127,6 @@ def BER_num_dec(s, cls_id=0, max_pow=32):
         raise BER_Decoding_Error("BER_num_dec: got empty string", remaining=s)
     x = cls_id
     for i, c in enumerate(s):
-        c = c
         x <<= 7
         x |= c & 0x7f
         if not c & 0x80:
@@ -275,20 +237,8 @@ def BER_tagging_enc(s, implicit_tag=None, explicit_tag=None):
 #    [ BER classes ]    #
 
 
-class BERcodec_metaclass(type):
-    def __new__(cls,
-                name,  # type: str
-                bases,  # type: Tuple[type, ...]
-                dct  # type: Dict[str, Any]
-                ):
-        # type: (...) -> Type[BERcodec_Object[Any]]
-        c = cast('Type[BERcodec_Object[Any]]',
-                 super(BERcodec_metaclass, cls).__new__(cls, name, bases, dct))
-        try:
-            c.tag.register(c.codec, c)
-        except Exception:
-            warning("Error registering %r for %r" % (c.tag, c.codec))
-        return c
+class BERcodec_metaclass(ASN1Codec_metaclass):
+    pass
 
 
 _K = TypeVar('_K')
@@ -412,13 +362,14 @@ def safedec(cls,
     @classmethod
     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.
+        # Ignore unknown kwargs (field=/pkt=/constraint keys) so BER packets
+        # do not TypeError when shared field call sites pass them through.
+        size_len = 0 if size_len is None else int(size_len)
         if isinstance(s, (str, bytes)):
             return BERcodec_STRING.enc(s, size_len=size_len)
         else:
             try:
-                return BERcodec_INTEGER.enc(int(s), size_len=size_len)  # type: ignore
+                return BERcodec_INTEGER.enc(int(s), size_len=size_len)  # type: ignore  # noqa: E501
             except TypeError:
                 raise TypeError("Trying to encode an invalid value !")
 
@@ -437,6 +388,7 @@ class BERcodec_INTEGER(BERcodec_Object[int]):
     @classmethod
     def enc(cls, i, size_len=0, **_kwargs):  # type: ignore[override]
         # type: (int, Optional[int], **Any) -> bytes
+        size_len = 0 if size_len is None else int(size_len)
         ls = []
         while True:
             ls.append(i & 0xff)
@@ -510,6 +462,7 @@ def do_dec(cls,
     @classmethod
     def enc(cls, _s, size_len=0, **_kwargs):  # type: ignore[override]
         # type: (AnyStr, Optional[int], **Any) -> bytes
+        size_len = 0 if size_len is None else int(size_len)
         # /!\ this is DER encoding (bit strings are only zero-bit padded)
         s = bytes_encode(_s)
         if len(s) % 8 == 0:
@@ -529,6 +482,7 @@ class BERcodec_STRING(BERcodec_Object[str]):
     @classmethod
     def enc(cls, _s, size_len=0, **_kwargs):  # type: ignore[override]
         # type: (Union[str, bytes], Optional[int], **Any) -> bytes
+        size_len = 0 if size_len is None else int(size_len)
         s = bytes_encode(_s)
         # Be sure we are encoding bytes
         return chb(int(cls.tag)) + BER_len_enc(len(s), size=size_len) + s
@@ -563,6 +517,7 @@ class BERcodec_OID(BERcodec_Object[bytes]):
     @classmethod
     def enc(cls, _oid, size_len=0, **_kwargs):  # type: ignore[override]
         # type: (AnyStr, Optional[int], **Any) -> bytes
+        size_len = 0 if size_len is None else int(size_len)
         oid = bytes_encode(_oid)
         if oid:
             lst = [int(x) for x in oid.strip(b".").split(b".")]
@@ -588,11 +543,12 @@ def do_dec(cls,
             l, s = BER_num_dec(s)
             lst.append(l)
         if lst:
-            # X.690 sect 8.19.4
-            lst.insert(0, lst[0] // 40)
-            lst[1] %= 40
+            from scapy.asn1.oid import oid_subidentifiers_to_dotted
+            oid = oid_subidentifiers_to_dotted(lst)
+        else:
+            oid = b""
         return (
-            cls.asn1_object(b".".join(str(k).encode('ascii') for k in lst)),
+            cls.asn1_object(oid),
             t,
         )
 
@@ -711,6 +667,7 @@ class BERcodec_IPADDRESS(BERcodec_STRING):
     @classmethod
     def enc(cls, ipaddr_ascii, size_len=0, **_kwargs):  # type: ignore[override]
         # type: (str, Optional[int], **Any) -> bytes
+        size_len = 0 if size_len is None else int(size_len)
         try:
             s = inet_aton(ipaddr_ascii)
         except Exception:

scapy/asn1/compound.py

@@ -0,0 +1,31 @@
+# SPDX-License-Identifier: GPL-2.0-only
+# This file is part of Scapy
+# See https://scapy.net/ for more information
+
+"""Shared helpers for ASN.1 compound-type encode/decode.
+
+Codec-specific SEQUENCE / CHOICE / SEQUENCE OF / PACKET hooks live in
+``compound_ber``, ``compound_oer``, and ``compound_uper``. Those modules
+are bound as methods on the encoder/decoder contexts in
+``scapy.asn1.context``.
+"""
+
+from typing import Any, Callable, List
+
+
+def sequence_decode_children(field, pkt, presence, dissect):
+    # type: (Any, Any, List[bool], Callable[[Any], None]) -> None
+    from scapy.asn1fields import ASN1F_badsequence, ASN1F_optional
+
+    opt_index = 0
+    for obj in field.seq:
+        if isinstance(obj, ASN1F_optional):
+            if not presence[opt_index]:
+                obj.set_missing(pkt)
+                opt_index += 1
+                continue
+            opt_index += 1
+        try:
+            dissect(obj)
+        except ASN1F_badsequence:
+            break