Index: pyasn1-0.1.9/pyasn1/codec/ber/decoder.py =================================================================== --- pyasn1-0.1.9.orig/pyasn1/codec/ber/decoder.py +++ pyasn1-0.1.9/pyasn1/codec/ber/decoder.py @@ -7,15 +7,17 @@ from pyasn1 import debug, error # Maximum number of continuation octets (high-bit set) allowed per OID arc. # 20 octets allows up to 140-bit integers, supporting UUID-based OIDs MAX_OID_ARC_CONTINUATION_OCTETS = 20 +MAX_NESTING_DEPTH = 100 class AbstractDecoder: protoComponent = None def valueDecoder(self, fullSubstrate, substrate, asn1Spec, tagSet, - length, state, decodeFun, substrateFun): + length, state, decodeFun, substrateFun, **options): raise error.PyAsn1Error('Decoder not implemented for %s' % (tagSet,)) def indefLenValueDecoder(self, fullSubstrate, substrate, asn1Spec, tagSet, - length, state, decodeFun, substrateFun): + length, state, decodeFun, substrateFun, + **options): raise error.PyAsn1Error('Indefinite length mode decoder not implemented for %s' % (tagSet,)) class AbstractSimpleDecoder(AbstractDecoder): @@ -44,25 +46,28 @@ class ExplicitTagDecoder(AbstractSimpleD protoComponent = univ.Any('') tagFormats = (tag.tagFormatConstructed,) def valueDecoder(self, fullSubstrate, substrate, asn1Spec, tagSet, - length, state, decodeFun, substrateFun): + length, state, decodeFun, substrateFun, **options): if substrateFun: return substrateFun( self._createComponent(asn1Spec, tagSet, ''), substrate, length ) head, tail = substrate[:length], substrate[length:] - value, _ = decodeFun(head, asn1Spec, tagSet, length) + value, _ = decodeFun(head, asn1Spec, tagSet, length, **options) return value, tail def indefLenValueDecoder(self, fullSubstrate, substrate, asn1Spec, tagSet, - length, state, decodeFun, substrateFun): + length, state, decodeFun, substrateFun, + **options): if substrateFun: return substrateFun( self._createComponent(asn1Spec, tagSet, ''), substrate, length ) - value, substrate = decodeFun(substrate, asn1Spec, tagSet, length) - terminator, substrate = decodeFun(substrate, allowEoo=True) + value, substrate = decodeFun(substrate, asn1Spec, tagSet, + length, **options) + terminator, substrate = decodeFun(substrate, allowEoo=True, + **options) if eoo.endOfOctets.isSameTypeWith(terminator) and \ terminator == eoo.endOfOctets: return value, substrate @@ -92,7 +97,7 @@ class IntegerDecoder(AbstractSimpleDecod } def valueDecoder(self, fullSubstrate, substrate, asn1Spec, tagSet, length, - state, decodeFun, substrateFun): + state, decodeFun, substrateFun, **options): head, tail = substrate[:length], substrate[length:] if not head: return self._createComponent(asn1Spec, tagSet, 0), tail @@ -117,7 +122,7 @@ class BitStringDecoder(AbstractSimpleDec protoComponent = univ.BitString(()) tagFormats = (tag.tagFormatSimple, tag.tagFormatConstructed) def valueDecoder(self, fullSubstrate, substrate, asn1Spec, tagSet, length, - state, decodeFun, substrateFun): + state, decodeFun, substrateFun, **options): head, tail = substrate[:length], substrate[length:] if tagSet[0][1] == tag.tagFormatSimple: # XXX what tag to check? if not head: @@ -143,18 +148,20 @@ class BitStringDecoder(AbstractSimpleDec if substrateFun: return substrateFun(r, substrate, length) while head: - component, head = decodeFun(head, self.protoComponent) + component, head = decodeFun(head, self.protoComponent, + **options) r = r + component return r, tail def indefLenValueDecoder(self, fullSubstrate, substrate, asn1Spec, tagSet, - length, state, decodeFun, substrateFun): + length, state, decodeFun, substrateFun, + **options): r = self._createComponent(asn1Spec, tagSet, '') if substrateFun: return substrateFun(r, substrate, length) while substrate: component, substrate = decodeFun(substrate, self.protoComponent, - allowEoo=True) + allowEoo=True, **options) if eoo.endOfOctets.isSameTypeWith(component) and \ component == eoo.endOfOctets: break @@ -169,7 +176,7 @@ class OctetStringDecoder(AbstractSimpleD protoComponent = univ.OctetString('') tagFormats = (tag.tagFormatSimple, tag.tagFormatConstructed) def valueDecoder(self, fullSubstrate, substrate, asn1Spec, tagSet, length, - state, decodeFun, substrateFun): + state, decodeFun, substrateFun, **options): head, tail = substrate[:length], substrate[length:] if tagSet[0][1] == tag.tagFormatSimple: # XXX what tag to check? return self._createComponent(asn1Spec, tagSet, head), tail @@ -177,18 +184,20 @@ class OctetStringDecoder(AbstractSimpleD if substrateFun: return substrateFun(r, substrate, length) while head: - component, head = decodeFun(head, self.protoComponent) + component, head = decodeFun(head, self.protoComponent, + **options) r = r + component return r, tail def indefLenValueDecoder(self, fullSubstrate, substrate, asn1Spec, tagSet, - length, state, decodeFun, substrateFun): + length, state, decodeFun, substrateFun, + **options): r = self._createComponent(asn1Spec, tagSet, '') if substrateFun: return substrateFun(r, substrate, length) while substrate: component, substrate = decodeFun(substrate, self.protoComponent, - allowEoo=True) + allowEoo=True, **options) if eoo.endOfOctets.isSameTypeWith(component) and \ component == eoo.endOfOctets: break @@ -202,7 +211,7 @@ class OctetStringDecoder(AbstractSimpleD class NullDecoder(AbstractSimpleDecoder): protoComponent = univ.Null('') def valueDecoder(self, fullSubstrate, substrate, asn1Spec, tagSet, - length, state, decodeFun, substrateFun): + length, state, decodeFun, substrateFun, **options): head, tail = substrate[:length], substrate[length:] r = self._createComponent(asn1Spec, tagSet) if head: @@ -212,7 +221,7 @@ class NullDecoder(AbstractSimpleDecoder) class ObjectIdentifierDecoder(AbstractSimpleDecoder): protoComponent = univ.ObjectIdentifier(()) def valueDecoder(self, fullSubstrate, substrate, asn1Spec, tagSet, length, - state, decodeFun, substrateFun): + state, decodeFun, substrateFun, **options): head, tail = substrate[:length], substrate[length:] if not head: raise error.PyAsn1Error('Empty substrate') @@ -267,7 +276,7 @@ class ObjectIdentifierDecoder(AbstractSi class RealDecoder(AbstractSimpleDecoder): protoComponent = univ.Real() def valueDecoder(self, fullSubstrate, substrate, asn1Spec, tagSet, - length, state, decodeFun, substrateFun): + length, state, decodeFun, substrateFun, **options): head, tail = substrate[:length], substrate[length:] if not head: return self._createComponent(asn1Spec, tagSet, 0.0), tail @@ -342,7 +351,7 @@ class SequenceDecoder(AbstractConstructe return r.getComponentPositionNearType(t, idx) def valueDecoder(self, fullSubstrate, substrate, asn1Spec, tagSet, - length, state, decodeFun, substrateFun): + length, state, decodeFun, substrateFun, **options): head, tail = substrate[:length], substrate[length:] r = self._createComponent(asn1Spec, tagSet) idx = 0 @@ -350,7 +359,7 @@ class SequenceDecoder(AbstractConstructe return substrateFun(r, substrate, length) while head: asn1Spec = self._getComponentTagMap(r, idx) - component, head = decodeFun(head, asn1Spec) + component, head = decodeFun(head, asn1Spec, **options) idx = self._getComponentPositionByType( r, component.getEffectiveTagSet(), idx ) @@ -361,14 +370,16 @@ class SequenceDecoder(AbstractConstructe return r, tail def indefLenValueDecoder(self, fullSubstrate, substrate, asn1Spec, tagSet, - length, state, decodeFun, substrateFun): + length, state, decodeFun, substrateFun, + **options): r = self._createComponent(asn1Spec, tagSet) if substrateFun: return substrateFun(r, substrate, length) idx = 0 while substrate: asn1Spec = self._getComponentTagMap(r, idx) - component, substrate = decodeFun(substrate, asn1Spec, allowEoo=True) + component, substrate = decodeFun(substrate, asn1Spec, + allowEoo=True, **options) if eoo.endOfOctets.isSameTypeWith(component) and \ component == eoo.endOfOctets: break @@ -388,7 +399,7 @@ class SequenceDecoder(AbstractConstructe class SequenceOfDecoder(AbstractConstructedDecoder): protoComponent = univ.SequenceOf() def valueDecoder(self, fullSubstrate, substrate, asn1Spec, tagSet, - length, state, decodeFun, substrateFun): + length, state, decodeFun, substrateFun, **options): head, tail = substrate[:length], substrate[length:] r = self._createComponent(asn1Spec, tagSet) if substrateFun: @@ -396,21 +407,23 @@ class SequenceOfDecoder(AbstractConstruc asn1Spec = r.getComponentType() idx = 0 while head: - component, head = decodeFun(head, asn1Spec) + component, head = decodeFun(head, asn1Spec, **options) r.setComponentByPosition(idx, component, asn1Spec is None) idx = idx + 1 r.verifySizeSpec() return r, tail def indefLenValueDecoder(self, fullSubstrate, substrate, asn1Spec, tagSet, - length, state, decodeFun, substrateFun): + length, state, decodeFun, substrateFun, + **options): r = self._createComponent(asn1Spec, tagSet) if substrateFun: return substrateFun(r, substrate, length) asn1Spec = r.getComponentType() idx = 0 while substrate: - component, substrate = decodeFun(substrate, asn1Spec, allowEoo=True) + component, substrate = decodeFun(substrate, asn1Spec, + allowEoo=True, **options) if eoo.endOfOctets.isSameTypeWith(component) and \ component == eoo.endOfOctets: break @@ -442,18 +455,19 @@ class ChoiceDecoder(AbstractConstructedD protoComponent = univ.Choice() tagFormats = (tag.tagFormatSimple, tag.tagFormatConstructed) def valueDecoder(self, fullSubstrate, substrate, asn1Spec, tagSet, - length, state, decodeFun, substrateFun): + length, state, decodeFun, substrateFun, **options): head, tail = substrate[:length], substrate[length:] r = self._createComponent(asn1Spec, tagSet) if substrateFun: return substrateFun(r, substrate, length) if r.getTagSet() == tagSet: # explicitly tagged Choice component, head = decodeFun( - head, r.getComponentTagMap() + head, r.getComponentTagMap(), **options ) else: component, head = decodeFun( - head, r.getComponentTagMap(), tagSet, length, state + head, r.getComponentTagMap(), tagSet, length, state, + **options ) if isinstance(component, univ.Choice): effectiveTagSet = component.getEffectiveTagSet() @@ -463,20 +477,24 @@ class ChoiceDecoder(AbstractConstructedD return r, tail def indefLenValueDecoder(self, fullSubstrate, substrate, asn1Spec, tagSet, - length, state, decodeFun, substrateFun): + length, state, decodeFun, substrateFun, **options): r = self._createComponent(asn1Spec, tagSet) if substrateFun: return substrateFun(r, substrate, length) if r.getTagSet() == tagSet: # explicitly tagged Choice - component, substrate = decodeFun(substrate, r.getComponentTagMap()) + component, substrate = decodeFun(substrate, + r.getComponentTagMap(), + **options) # eat up EOO marker - eooMarker, substrate = decodeFun(substrate, allowEoo=True) + eooMarker, substrate = decodeFun(substrate, allowEoo=True, + **options) if not eoo.endOfOctets.isSameTypeWith(eooMarker) or \ eooMarker != eoo.endOfOctets: raise error.PyAsn1Error('No EOO seen before substrate ends') else: component, substrate= decodeFun( - substrate, r.getComponentTagMap(), tagSet, length, state + substrate, r.getComponentTagMap(), tagSet, length, + state, **options ) if isinstance(component, univ.Choice): effectiveTagSet = component.getEffectiveTagSet() @@ -489,7 +507,7 @@ class AnyDecoder(AbstractSimpleDecoder): protoComponent = univ.Any() tagFormats = (tag.tagFormatSimple, tag.tagFormatConstructed) def valueDecoder(self, fullSubstrate, substrate, asn1Spec, tagSet, - length, state, decodeFun, substrateFun): + length, state, decodeFun, substrateFun, **options): if asn1Spec is None or \ asn1Spec is not None and tagSet != asn1Spec.getTagSet(): # untagged Any container, recover inner header substrate @@ -502,7 +520,8 @@ class AnyDecoder(AbstractSimpleDecoder): return self._createComponent(asn1Spec, tagSet, value=head), tail def indefLenValueDecoder(self, fullSubstrate, substrate, asn1Spec, tagSet, - length, state, decodeFun, substrateFun): + length, state, decodeFun, substrateFun, + **options): if asn1Spec is not None and tagSet == asn1Spec.getTagSet(): # tagged Any type -- consume header substrate header = '' @@ -518,7 +537,8 @@ class AnyDecoder(AbstractSimpleDecoder): if substrateFun: return substrateFun(r, substrate, length) while substrate: - component, substrate = decodeFun(substrate, asn1Spec, allowEoo=True) + component, substrate = decodeFun(substrate, asn1Spec, + allowEoo=True, **options) if eoo.endOfOctets.isSameTypeWith(component) and \ component == eoo.endOfOctets: break @@ -619,7 +639,16 @@ class Decoder: def __call__(self, substrate, asn1Spec=None, tagSet=None, length=None, state=stDecodeTag, recursiveFlag=1, - substrateFun=None, allowEoo=False): + substrateFun=None, allowEoo=False, **options): + _nestingLevel = options.get('_nestingLevel', 0) + + if _nestingLevel > MAX_NESTING_DEPTH: + raise error.PyAsn1Error( + 'ASN.1 structure nesting depth exceeds limit (%d)' % MAX_NESTING_DEPTH + ) + + options['_nestingLevel'] = _nestingLevel + 1 + if debug.logger & debug.flagDecoder: debug.logger('decoder called at scope %s with state %d, working with up to %d octets of substrate: %s' % (debug.scope, state, len(substrate), debug.hexdump(substrate))) fullSubstrate = substrate @@ -828,12 +857,12 @@ class Decoder: if length == -1: # indef length value, substrate = concreteDecoder.indefLenValueDecoder( fullSubstrate, substrate, asn1Spec, tagSet, length, - stGetValueDecoder, self, substrateFun + stGetValueDecoder, self, substrateFun, **options ) else: value, substrate = concreteDecoder.valueDecoder( fullSubstrate, substrate, asn1Spec, tagSet, length, - stGetValueDecoder, self, substrateFun + stGetValueDecoder, self, substrateFun, **options ) state = stStop debug.logger and debug.logger & debug.flagDecoder and debug.logger('codec %s yields type %s, value:\n%s\n...remaining substrate is: %s' % (concreteDecoder.__class__.__name__, value.__class__.__name__, value.prettyPrint(), substrate and debug.hexdump(substrate) or '')) Index: pyasn1-0.1.9/test/codec/ber/test_decoder.py =================================================================== --- pyasn1-0.1.9.orig/test/codec/ber/test_decoder.py +++ pyasn1-0.1.9/test/codec/ber/test_decoder.py @@ -13,6 +13,23 @@ if version_info[0:2] < (2, 7) or \ else: import unittest + +def bit_length(n): + s = bin(n) + s = s.lstrip('-0b') + return len(s) + +def to_bytes(n, length=1, byteorder='big', signed=False): + if byteorder == 'little': + order = range(length) + elif byteorder == 'big': + order = reversed(range(length)) + else: + raise ValueError("byteorder must be either 'little' or 'big'") + + return bytes((n >> i*8) & 0xff for i in order) + + class LargeTagDecoderTestCase(unittest.TestCase): def testLargeTag(self): assert decoder.decode(ints2octs((127, 141, 245, 182, 253, 47, 3, 2, 1, 1))) == (1, null) @@ -791,4 +808,119 @@ class EndOfOctetsTestCase(unittest.TestC else: assert 0, 'end-of-contents octets accepted with unexpected data' +class NestingDepthLimitTestCase(unittest.TestCase): + """Test protection against deeply nested ASN.1 structures (CVE prevention).""" + + def testIndefLenSequenceNesting(self): + """Deeply nested indefinite-length SEQUENCEs must raise PyAsn1Error.""" + # Each \x30\x80 opens a new indefinite-length SEQUENCE + payload = b'\x30\x80' * 200 + try: + decoder.decode(payload) + except error.PyAsn1Error: + pass + else: + assert False, 'Deeply nested indef-length SEQUENCEs not rejected' + + def testIndefLenSetNesting(self): + """Deeply nested indefinite-length SETs must raise PyAsn1Error.""" + # Each \x31\x80 opens a new indefinite-length SET + payload = b'\x31\x80' * 200 + try: + decoder.decode(payload) + except error.PyAsn1Error: + pass + else: + assert False, 'Deeply nested indef-length SETs not rejected' + + def testDefiniteLenNesting(self): + """Deeply nested definite-length SEQUENCEs must raise PyAsn1Error.""" + inner = b'\x05\x00' # NULL + for _ in range(200): + length = len(inner) + if length < 128: + inner = b'\x30' + bytes([length]) + inner + else: + length_bytes = to_bytes(length, + (bit_length(length) + 7) // 8, 'big') + inner = b'\x30' + bytes([0x80 | len(length_bytes)]) + \ + length_bytes + inner + try: + decoder.decode(inner) + except error.PyAsn1Error: + pass + else: + assert False, 'Deeply nested definite-length SEQUENCEs not rejected' + + def testNestingUnderLimitWorks(self): + """Nesting within the limit must decode successfully.""" + inner = b'\x05\x00' # NULL + for _ in range(50): + length = len(inner) + if length < 128: + inner = b'\x30' + bytes([length]) + inner + else: + length_bytes = to_bytes(length, + (bit_length(length) + 7) // 8, 'big') + inner = b'\x30' + bytes([0x80 | len(length_bytes)]) + \ + length_bytes + inner + asn1Object, _ = decoder.decode(inner) + assert asn1Object is not None, 'Valid nested structure rejected' + + def testSiblingsDontIncreaseDepth(self): + """Sibling elements at the same level must not inflate depth count.""" + # SEQUENCE containing 200 INTEGER siblings - should decode fine + components = b'\x02\x01\x01' * 200 # 200 x INTEGER(1) + length = len(components) + length_bytes = to_bytes(length, + (bit_length(length) + 7) // 8, 'big') + payload = b'\x30' + bytes([0x80 | len(length_bytes)]) + \ + length_bytes + components + asn1Object, _ = decoder.decode(payload) + assert asn1Object is not None, 'Siblings incorrectly rejected' + + def testErrorMessageContainsLimit(self): + """Error message must indicate the nesting depth limit.""" + payload = b'\x30\x80' * 200 + try: + decoder.decode(payload) + except error.PyAsn1Error as exc: + assert 'nesting depth' in str(exc).lower(), \ + 'Error message missing depth info: %s' % exc + else: + assert False, 'Expected PyAsn1Error' + + def testNoRecursionError(self): + """Must raise PyAsn1Error, not RecursionError.""" + payload = b'\x30\x80' * 50000 + try: + decoder.decode(payload) + except error.PyAsn1Error: + pass + except RecursionError: + assert False, 'Got RecursionError instead of PyAsn1Error' + + def testMixedNesting(self): + """Mixed SEQUENCE and SET nesting must be caught.""" + # Alternate SEQUENCE (0x30) and SET (0x31) with indef length + payload = b'' + for i in range(200): + payload += b'\x30\x80' if i % 2 == 0 else b'\x31\x80' + try: + decoder.decode(payload) + except error.PyAsn1Error: + pass + else: + assert False, 'Mixed nesting not rejected' + + def testWithSchema(self): + """Deeply nested structures must be caught even with schema.""" + payload = b'\x30\x80' * 200 + try: + decoder.decode(payload, asn1Spec=univ.Sequence()) + except error.PyAsn1Error: + pass + else: + assert False, 'Deeply nested with schema not rejected' + if __name__ == '__main__': unittest.main() Index: pyasn1-0.1.9/test/codec/cer/test_decoder.py =================================================================== --- pyasn1-0.1.9.orig/test/codec/cer/test_decoder.py +++ pyasn1-0.1.9/test/codec/cer/test_decoder.py @@ -12,6 +12,7 @@ if version_info[0:2] < (2, 7) or \ else: import unittest + class BooleanDecoderTestCase(unittest.TestCase): def testTrue(self): assert decoder.decode(ints2octs((1, 1, 255))) == (1, null) @@ -38,4 +39,28 @@ class OctetStringDecoderTestCase(unittes ints2octs((36, 128, 4, 130, 3, 232) + (81,)*1000 + (4, 1, 81, 0, 0)) ) == (str2octs('Q'*1001), null) +class NestingDepthLimitTestCase(unittest.TestCase): + """Test CER decoder protection against deeply nested structures.""" + + def testIndefLenNesting(self): + """Deeply nested indefinite-length SEQUENCEs must raise PyAsn1Error.""" + payload = b'\x30\x80' * 200 + try: + decoder.decode(payload) + except PyAsn1Error: + pass + else: + assert False, 'Deeply nested indef-length SEQUENCEs not rejected' + + def testNoRecursionError(self): + """Must raise PyAsn1Error, not RecursionError.""" + payload = b'\x30\x80' * 50000 + try: + decoder.decode(payload) + except PyAsn1Error: + pass + except RecursionError: + assert False, 'Got RecursionError instead of PyAsn1Error' + + if __name__ == '__main__': unittest.main() Index: pyasn1-0.1.9/test/codec/der/test_decoder.py =================================================================== --- pyasn1-0.1.9.orig/test/codec/der/test_decoder.py +++ pyasn1-0.1.9/test/codec/der/test_decoder.py @@ -12,6 +12,23 @@ if version_info[0:2] < (2, 7) or \ else: import unittest + +def bit_length(n): + s = bin(n) + s = s.lstrip('-0b') + return len(s) + +def to_bytes(n, length=1, byteorder='big', signed=False): + if byteorder == 'little': + order = range(length) + elif byteorder == 'big': + order = reversed(range(length)) + else: + raise ValueError("byteorder must be either 'little' or 'big'") + + return bytes((n >> i*8) & 0xff for i in order) + + class OctetStringDecoderTestCase(unittest.TestCase): def testShortMode(self): assert decoder.decode( @@ -28,4 +45,45 @@ class OctetStringDecoderTestCase(unittes else: assert 0, 'indefinite length encoding tolerated' +class NestingDepthLimitTestCase(unittest.TestCase): + """Test DER decoder protection against deeply nested structures.""" + + def testDefiniteLenNesting(self): + """Deeply nested definite-length SEQUENCEs must raise PyAsn1Error.""" + inner = b'\x05\x00' # NULL + for _ in range(200): + length = len(inner) + if length < 128: + inner = b'\x30' + bytes([length]) + inner + else: + length_bytes = to_bytes(length, + (bit_length(length) + 7) // 8, 'big') + inner = b'\x30' + bytes([0x80 | len(length_bytes)]) + \ + length_bytes + inner + try: + decoder.decode(inner) + except PyAsn1Error: + pass + else: + assert False, 'Deeply nested definite-length SEQUENCEs not rejected' + + def testNoRecursionError(self): + """Must raise PyAsn1Error, not RecursionError.""" + inner = b'\x05\x00' + for _ in range(200): + length = len(inner) + if length < 128: + inner = b'\x30' + bytes([length]) + inner + else: + length_bytes = to_bytes(length, + (bit_length(length) + 7) // 8, 'big') + inner = b'\x30' + bytes([0x80 | len(length_bytes)]) + \ + length_bytes + inner + try: + decoder.decode(inner) + except PyAsn1Error: + pass + except RecursionError: + assert False, 'Got RecursionError instead of PyAsn1Error' + if __name__ == '__main__': unittest.main() Index: pyasn1-0.1.9/pyasn1/codec/cer/decoder.py =================================================================== --- pyasn1-0.1.9.orig/pyasn1/codec/cer/decoder.py +++ pyasn1-0.1.9/pyasn1/codec/cer/decoder.py @@ -7,7 +7,7 @@ from pyasn1 import error class BooleanDecoder(decoder.AbstractSimpleDecoder): protoComponent = univ.Boolean(0) def valueDecoder(self, fullSubstrate, substrate, asn1Spec, tagSet, length, - state, decodeFun, substrateFun): + state, decodeFun, substrateFun, **options): head, tail = substrate[:length], substrate[length:] if not head or length != 1: raise error.PyAsn1Error('Not single-octet Boolean payload')