diff --git a/scapy/asn1/asn1.py b/scapy/asn1/asn1.py index bd58715e26d..053baf9b1fd 100644 --- a/scapy/asn1/asn1.py +++ b/scapy/asn1/asn1.py @@ -126,13 +126,13 @@ def register_stem(cls, stem): # type: (Type[BERcodec_Object[Any]]) -> None cls._stem = stem - def dec(cls, s, context=None): - # type: (bytes, Optional[Type[ASN1_Class]]) -> ASN1_Object[Any] - return cls._stem.dec(s, context=context) # type: ignore + def dec(cls, s, context=None, _depth=0): + # type: (bytes, Optional[Type[ASN1_Class]], int) -> ASN1_Object[Any] + return cls._stem.dec(s, context=context, _depth=_depth) # type: ignore - def safedec(cls, s, context=None): - # type: (bytes, Optional[Type[ASN1_Class]]) -> ASN1_Object[Any] - return cls._stem.safedec(s, context=context) # type: ignore + def safedec(cls, s, context=None, _depth=0): + # type: (bytes, Optional[Type[ASN1_Class]], int) -> ASN1_Object[Any] + return cls._stem.safedec(s, context=context, _depth=_depth) # type: ignore def get_stem(cls): # type: () -> type diff --git a/scapy/asn1/ber.py b/scapy/asn1/ber.py index 23899274e4a..b5ffc4252b7 100644 --- a/scapy/asn1/ber.py +++ b/scapy/asn1/ber.py @@ -50,6 +50,8 @@ # [ BER tools ] # +MAX_BER_DEPTH = 32 + class BER_Exception(Exception): pass @@ -155,8 +157,8 @@ def BER_num_enc(ll, size=1): return b"".join(chb(k) for k in x) -def BER_num_dec(s, cls_id=0): - # type: (bytes, int) -> Tuple[int, bytes] +def BER_num_dec(s, cls_id=0, max_pow=32): + # type: (bytes, int, int) -> Tuple[int, bytes] if len(s) == 0: raise BER_Decoding_Error("BER_num_dec: got empty string", remaining=s) x = cls_id @@ -166,6 +168,9 @@ def BER_num_dec(s, cls_id=0): x |= c & 0x7f if not c & 0x80: break + if i > max_pow: + raise BER_Decoding_Error("BER_num_dec: maximum value reached (2^32-1)", + remaining=s) if c & 0x80: raise BER_Decoding_Error("BER_num_dec: unfinished number description", remaining=s) @@ -338,7 +343,8 @@ def check_type_check_len(cls, s): def do_dec(cls, s, # type: bytes context=None, # type: Optional[Type[ASN1_Class]] - safe=False # type: bool + safe=False, # type: bool + _depth=0, # type: int ): # type: (...) -> Tuple[ASN1_Object[Any], bytes] if context is not None: @@ -360,22 +366,28 @@ def do_dec(cls, # Value type defined as Unknown l, s = BER_num_dec(remainder) return ASN1_BADTAG(s[:l]), s[l:] - return codec.dec(s, _context, safe) + return codec.dec(s, _context, safe, _depth=_depth) @classmethod def dec(cls, s, # type: bytes context=None, # type: Optional[Type[ASN1_Class]] safe=False, # type: bool + _depth=0, # type: int ): # type: (...) -> Tuple[Union[_ASN1_ERROR, ASN1_Object[_K]], bytes] + if _depth > MAX_BER_DEPTH: + raise BER_Exception("Reached maximum BER recursion limit") if not safe: - return cls.do_dec(s, context, safe) + return cls.do_dec(s, context, safe, _depth=_depth) try: - return cls.do_dec(s, context, safe) + return cls.do_dec(s, context, safe, _depth=_depth) except BER_BadTag_Decoding_Error as e: o, remain = BERcodec_Object.dec( - e.remaining, context, safe + e.remaining, + context=context, + safe=safe, + _depth=_depth + 1, ) # type: Tuple[ASN1_Object[Any], bytes] return ASN1_BADTAG(o), remain except BER_Decoding_Error as e: @@ -387,9 +399,10 @@ def dec(cls, def safedec(cls, s, # type: bytes context=None, # type: Optional[Type[ASN1_Class]] + _depth=0, # type: int ): # type: (...) -> Tuple[Union[_ASN1_ERROR, ASN1_Object[_K]], bytes] - return cls.dec(s, context, safe=True) + return cls.dec(s, context, safe=True, _depth=_depth) @classmethod def enc(cls, s, size_len=0): @@ -437,6 +450,7 @@ def do_dec(cls, s, # type: bytes context=None, # type: Optional[Type[ASN1_Class]] safe=False, # type: bool + _depth=0, # type: int ): # type: (...) -> Tuple[ASN1_Object[int], bytes] l, s, t = cls.check_type_check_len(s) @@ -461,7 +475,8 @@ class BERcodec_BIT_STRING(BERcodec_Object[str]): def do_dec(cls, s, # type: bytes context=None, # type: Optional[Type[ASN1_Class]] - safe=False # type: bool + safe=False, # type: bool + _depth=0, # type: int ): # type: (...) -> Tuple[ASN1_Object[str], bytes] # /!\ the unused_bits information is lost after this decoding @@ -515,6 +530,7 @@ def do_dec(cls, s, # type: bytes context=None, # type: Optional[Type[ASN1_Class]] safe=False, # type: bool + _depth=0, # type: int ): # type: (...) -> Tuple[ASN1_Object[Any], bytes] l, s, t = cls.check_type_check_len(s) @@ -555,6 +571,7 @@ def do_dec(cls, s, # type: bytes context=None, # type: Optional[Type[ASN1_Class]] safe=False, # type: bool + _depth=0, # type: int ): # type: (...) -> Tuple[ASN1_Object[bytes], bytes] l, s, t = cls.check_type_check_len(s) @@ -562,7 +579,8 @@ def do_dec(cls, while s: l, s = BER_num_dec(s) lst.append(l) - if (len(lst) > 0): + if lst: + # X.690 sect 8.19.4 lst.insert(0, lst[0] // 40) lst[1] %= 40 return ( @@ -639,18 +657,24 @@ def enc(cls, _ll, size_len=0): def do_dec(cls, s, # type: bytes context=None, # type: Optional[Type[ASN1_Class]] - safe=False # type: bool + safe=False, # type: bool + _depth=0, # type: int ): # type: (...) -> Tuple[ASN1_Object[Union[bytes, List[Any]]], bytes] if context is None: context = cls.tag.context + if _depth > MAX_BER_DEPTH: + raise BER_Exception("Reached maximum BER recursion limit") ll, st = cls.check_type_get_len(s) # we may have len(s) < ll s, t = st[:ll], st[ll:] obj = [] while s: try: o, remain = BERcodec_Object.dec( - s, context, safe + s, + context=context, + safe=safe, + _depth=_depth + 1, ) # type: Tuple[ASN1_Object[Any], bytes] s = remain except BER_Decoding_Error as err: @@ -683,8 +707,13 @@ def enc(cls, ipaddr_ascii, size_len=0): # type: ignore return chb(int(cls.tag)) + BER_len_enc(len(s), size=size_len) + s @classmethod - def do_dec(cls, s, context=None, safe=False): - # type: (bytes, Optional[Any], bool) -> Tuple[ASN1_Object[str], bytes] + def do_dec(cls, + s, # type: bytes + context=None, # type: Optional[Any] + safe=False, # type: bool + _depth=0, # type: int + ): + # type: (...) -> Tuple[ASN1_Object[str], bytes] l, s, t = cls.check_type_check_len(s) try: ipaddr_ascii = inet_ntoa(s) diff --git a/scapy/cbor/cbor.py b/scapy/cbor/cbor.py index 5588dba664d..1dcff4943f1 100644 --- a/scapy/cbor/cbor.py +++ b/scapy/cbor/cbor.py @@ -185,13 +185,13 @@ def register_stem(cls, stem): # type: (Type[CBORcodec_Object[Any]]) -> None cls._stem = stem - def dec(cls, s, context=None): - # type: (bytes, Optional[Any]) -> CBOR_Object[Any] - return cls._stem.dec(s, context=context) # type: ignore + def dec(cls, s, context=None, _depth=0): + # type: (bytes, Optional[Any], int) -> CBOR_Object[Any] + return cls._stem.dec(s, context=context, _depth=_depth) # type: ignore - def safedec(cls, s, context=None): - # type: (bytes, Optional[Any]) -> CBOR_Object[Any] - return cls._stem.safedec(s, context=context) # type: ignore + def safedec(cls, s, context=None, _depth=0): + # type: (bytes, Optional[Any], int) -> CBOR_Object[Any] + return cls._stem.safedec(s, context=context, _depth=_depth) # type: ignore def get_stem(cls): # type: () -> type diff --git a/scapy/cbor/cborcodec.py b/scapy/cbor/cborcodec.py index b49b9d38b30..5cfa0ccbc92 100644 --- a/scapy/cbor/cborcodec.py +++ b/scapy/cbor/cborcodec.py @@ -184,7 +184,8 @@ def check_string(cls, s): def do_dec(cls, s, # type: bytes context=None, # type: Optional[Any] - safe=False # type: bool + safe=False, # type: bool + _depth=0, # type: int ): # type: (...) -> Tuple[CBOR_Object[Any], bytes] """Decode CBOR data using automatic dispatch based on major type.""" @@ -195,12 +196,13 @@ def dec(cls, s, # type: bytes context=None, # type: Optional[Any] safe=False, # type: bool + _depth=0, # type: int ): # type: (...) -> Tuple[Union[_CBOR_ERROR, CBOR_Object[_K]], bytes] if not safe: - return cls.do_dec(s, context, safe) + return cls.do_dec(s, context, safe, _depth=_depth) try: - return cls.do_dec(s, context, safe) + return cls.do_dec(s, context, safe, _depth=_depth) except CBOR_Codec_Decoding_Error as e: return CBOR_DECODING_ERROR(s, exc=e), b"" except CBOR_Error as e: @@ -210,9 +212,10 @@ def dec(cls, def safedec(cls, s, # type: bytes context=None, # type: Optional[Any] + _depth=0, # type: int ): # type: (...) -> Tuple[Union[_CBOR_ERROR, CBOR_Object[_K]], bytes] - return cls.dec(s, context, safe=True) + return cls.dec(s, context, safe=True, _depth=_depth) @classmethod def enc(cls, s): @@ -248,6 +251,7 @@ def do_dec(cls, s, # type: bytes context=None, # type: Optional[Any] safe=False, # type: bool + _depth=0, # type: int ): # type: (...) -> Tuple[CBOR_Object[int], bytes] cls.check_string(s) @@ -280,6 +284,7 @@ def do_dec(cls, s, # type: bytes context=None, # type: Optional[Any] safe=False, # type: bool + _depth=0, # type: int ): # type: (...) -> Tuple[CBOR_Object[int], bytes] cls.check_string(s) @@ -310,6 +315,7 @@ def do_dec(cls, s, # type: bytes context=None, # type: Optional[Any] safe=False, # type: bool + _depth=0, # type: int ): # type: (...) -> Tuple[CBOR_Object[bytes], bytes] cls.check_string(s) @@ -345,6 +351,7 @@ def do_dec(cls, s, # type: bytes context=None, # type: Optional[Any] safe=False, # type: bool + _depth=0, # type: int ): # type: (...) -> Tuple[CBOR_Object[str], bytes] cls.check_string(s) @@ -384,6 +391,7 @@ def do_dec(cls, s, # type: bytes context=None, # type: Optional[Any] safe=False, # type: bool + _depth=0, # type: int ): # type: (...) -> Tuple[CBOR_Object[List[Any]], bytes] cls.check_string(s) @@ -425,6 +433,7 @@ def do_dec(cls, s, # type: bytes context=None, # type: Optional[Any] safe=False, # type: bool + _depth=0, # type: int ): # type: (...) -> Tuple[CBOR_Object[Dict[Any, Any]], bytes] cls.check_string(s) @@ -475,6 +484,7 @@ def do_dec(cls, s, # type: bytes context=None, # type: Optional[Any] safe=False, # type: bool + _depth=0, # type: int ): # type: (...) -> Tuple[CBOR_Object[Tuple[int, Any]], bytes] cls.check_string(s) @@ -540,6 +550,7 @@ def do_dec(cls, s, # type: bytes context=None, # type: Optional[Any] safe=False, # type: bool + _depth=0, # type: int ): # type: (...) -> Tuple[CBOR_Object[Any], bytes] from scapy.cbor.cbor import ( diff --git a/test/regression.uts b/test/regression.uts index 18525b0e075..32a75bb4af6 100644 --- a/test/regression.uts +++ b/test/regression.uts @@ -4252,16 +4252,21 @@ r1 == b'@\x04\x08\x08\x08\x08' r2 = b.dec(r1)[0] r2.val == '8.8.8.8' -a = b'\x1f\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\x01\x01\x00C\x02\x01U0\x0f0\r\x06\x08+\x06\x01\x02\x01\x02\x01\x00\x02\x01!' -ret = False +a = b'\x1f\xfe\xfe\xfe\xfe\xfe\x01\x01\x00C\x02\x01U0\x0f0\r\x06\x08+\x06\x01\x02\x01\x02\x01\x00\x02\x01!' try: BERcodec_Object.check_type(a) -except BER_BadTag_Decoding_Error: - ret = True +except BER_BadTag_Decoding_Error as ex: + assert "Got tag" in str(ex) else: - ret = False + assert False -assert ret +a = b'\x1f\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\x01\x01\x00C\x02\x01U0\x0f0\r\x06\x08+\x06\x01\x02\x01\x02\x01\x00\x02\x01!' +try: + BERcodec_Object.check_type(a) +except BER_Decoding_Error as ex: + assert "maximum value" in str(ex) +else: + assert False = BER trigger failures @@ -4272,6 +4277,17 @@ except BER_Decoding_Error: pass += BER tests - 2 + +a = b'0c\x02\x01\x01\x04\x06public\xa2V\x02\x01\x01\x02\x01\x00\x02\x01\x000K0I\x06\x03+\x06\x010B0@0>0<0:08060402000.0,0*0(0&0$0"0 0\x1e0\x1c0\x1a0\x180\x160\x140\x120\x100\x0e0\x0c0\n0\x080\x060\x040\x020\x00' +try: + SNMP(a) +except BER_Exception as ex: + assert "recursion limit" in str(ex) +else: + assert False + + ############ ############ + Fields