#

# This file is part of pyasn1 software.

#

# Copyright (c) 2005-2020, Ilya Etingof <[email protected]>

# License: https://pyasn1.readthedocs.io/en/latest/license.html

#

import io

import os

import sys

import warnings

from pyasn1 import debug

from pyasn1 import error

from pyasn1.codec.ber import eoo

from pyasn1.codec.streaming import asSeekableStream

from pyasn1.codec.streaming import isEndOfStream

from pyasn1.codec.streaming import peekIntoStream

from pyasn1.codec.streaming import readFromStream

from pyasn1.compat import _MISSING

from pyasn1.error import PyAsn1Error

from pyasn1.type import base

from pyasn1.type import char

from pyasn1.type import tag

from pyasn1.type import tagmap

from pyasn1.type import univ

from pyasn1.type import useful

__all__ = ['StreamingDecoder', 'Decoder', 'decode']

LOG = debug.registerLoggee(__name__, flags=debug.DEBUG_DECODER)

noValue = base.noValue

SubstrateUnderrunError = error.SubstrateUnderrunError

# 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

# Maximum number of bytes in a BER length field (8 bytes = up to 2^64-1)

MAX_LENGTH_OCTETS = 8

class AbstractPayloadDecoder(object):

protoComponent = None

def valueDecoder(self, substrate, asn1Spec,

tagSet=None, length=None, state=None,

decodeFun=None, substrateFun=None,

**options):

"""Decode value with fixed byte length.

The decoder is allowed to consume as many bytes as necessary.

"""

raise error.PyAsn1Error('SingleItemDecoder not implemented for %s' % (tagSet,)) # TODO: Seems more like an NotImplementedError?

def indefLenValueDecoder(self, substrate, asn1Spec,

tagSet=None, length=None, state=None,

decodeFun=None, substrateFun=None,

**options):

"""Decode value with undefined length.

The decoder is allowed to consume as many bytes as necessary.

"""

raise error.PyAsn1Error('Indefinite length mode decoder not implemented for %s' % (tagSet,)) # TODO: Seems more like an NotImplementedError?

@staticmethod

def _passAsn1Object(asn1Object, options):

if 'asn1Object' not in options:

options['asn1Object'] = asn1Object

return options

class AbstractSimplePayloadDecoder(AbstractPayloadDecoder):

@staticmethod

def substrateCollector(asn1Object, substrate, length, options):

for chunk in readFromStream(substrate, length, options):

yield chunk

def _createComponent(self, asn1Spec, tagSet, value, **options):

if options.get('native'):

return value

elif asn1Spec is None:

return self.protoComponent.clone(value, tagSet=tagSet)

elif value is noValue:

return asn1Spec

else:

return asn1Spec.clone(value)

class RawPayloadDecoder(AbstractSimplePayloadDecoder):

protoComponent = univ.Any('')

def valueDecoder(self, substrate, asn1Spec,

tagSet=None, length=None, state=None,

decodeFun=None, substrateFun=None,

**options):

if substrateFun:

asn1Object = self._createComponent(asn1Spec, tagSet, '', **options)

for chunk in substrateFun(asn1Object, substrate, length, options):

yield chunk

return

for value in decodeFun(substrate, asn1Spec, tagSet, length, **options):

yield value

def indefLenValueDecoder(self, substrate, asn1Spec,

tagSet=None, length=None, state=None,

decodeFun=None, substrateFun=None,

**options):

if substrateFun:

asn1Object = self._createComponent(asn1Spec, tagSet, '', **options)

for chunk in substrateFun(asn1Object, substrate, length, options):

yield chunk

return

while True:

for value in decodeFun(

substrate, asn1Spec, tagSet, length,

allowEoo=True, **options):

if value is eoo.endOfOctets:

return

yield value

rawPayloadDecoder = RawPayloadDecoder()

class IntegerPayloadDecoder(AbstractSimplePayloadDecoder):

protoComponent = univ.Integer(0)

def valueDecoder(self, substrate, asn1Spec,

tagSet=None, length=None, state=None,

decodeFun=None, substrateFun=None,

**options):

if tagSet[0].tagFormat != tag.tagFormatSimple:

raise error.PyAsn1Error('Simple tag format expected')

for chunk in readFromStream(substrate, length, options):

if isinstance(chunk, SubstrateUnderrunError):

yield chunk

if chunk:

value = int.from_bytes(bytes(chunk), 'big', signed=True)

else:

value = 0

yield self._createComponent(asn1Spec, tagSet, value, **options)

class BooleanPayloadDecoder(IntegerPayloadDecoder):

protoComponent = univ.Boolean(0)

def _createComponent(self, asn1Spec, tagSet, value, **options):

return IntegerPayloadDecoder._createComponent(

self, asn1Spec, tagSet, value and 1 or 0, **options)

class BitStringPayloadDecoder(AbstractSimplePayloadDecoder):

protoComponent = univ.BitString(())

supportConstructedForm = True

def valueDecoder(self, substrate, asn1Spec,

tagSet=None, length=None, state=None,

decodeFun=None, substrateFun=None,

**options):

if substrateFun:

asn1Object = self._createComponent(asn1Spec, tagSet, noValue, **options)

for chunk in substrateFun(asn1Object, substrate, length, options):

yield chunk

return

if not length:

raise error.PyAsn1Error('Empty BIT STRING substrate')

for chunk in isEndOfStream(substrate):

if isinstance(chunk, SubstrateUnderrunError):

yield chunk

if chunk:

raise error.PyAsn1Error('Empty BIT STRING substrate')

if tagSet[0].tagFormat == tag.tagFormatSimple: # XXX what tag to check?

for trailingBits in readFromStream(substrate, 1, options):

if isinstance(trailingBits, SubstrateUnderrunError):

yield trailingBits

trailingBits = ord(trailingBits)

if trailingBits > 7:

raise error.PyAsn1Error(

'Trailing bits overflow %s' % trailingBits

)

for chunk in readFromStream(substrate, length - 1, options):

if isinstance(chunk, SubstrateUnderrunError):

yield chunk

value = self.protoComponent.fromOctetString(

chunk, internalFormat=True, padding=trailingBits)

yield self._createComponent(asn1Spec, tagSet, value, **options)

return

if not self.supportConstructedForm:

raise error.PyAsn1Error('Constructed encoding form prohibited '

'at %s' % self.__class__.__name__)

if LOG:

LOG('assembling constructed serialization')

# All inner fragments are of the same type, treat them as octet string

substrateFun = self.substrateCollector

bitString = self.protoComponent.fromOctetString(b'', internalFormat=True)

current_position = substrate.tell()

while substrate.tell() - current_position < length:

for component in decodeFun(

substrate, self.protoComponent, substrateFun=substrateFun,

**options):

if isinstance(component, SubstrateUnderrunError):

yield component

trailingBits = component[0]

if trailingBits > 7:

raise error.PyAsn1Error(

'Trailing bits overflow %s' % trailingBits

)

bitString = self.protoComponent.fromOctetString(

component[1:], internalFormat=True,

prepend=bitString, padding=trailingBits

)

yield self._createComponent(asn1Spec, tagSet, bitString, **options)

def indefLenValueDecoder(self, substrate, asn1Spec,

tagSet=None, length=None, state=None,

decodeFun=None, substrateFun=None,

**options):

if substrateFun:

asn1Object = self._createComponent(asn1Spec, tagSet, noValue, **options)

for chunk in substrateFun(asn1Object, substrate, length, options):

yield chunk

return

# All inner fragments are of the same type, treat them as octet string

substrateFun = self.substrateCollector

bitString = self.protoComponent.fromOctetString(b'', internalFormat=True)

while True: # loop over fragments

for component in decodeFun(

substrate, self.protoComponent, substrateFun=substrateFun,

allowEoo=True, **options):

if component is eoo.endOfOctets:

break

if isinstance(component, SubstrateUnderrunError):

yield component

if component is eoo.endOfOctets:

break

trailingBits = component[0]

if trailingBits > 7:

raise error.PyAsn1Error(

'Trailing bits overflow %s' % trailingBits

)

bitString = self.protoComponent.fromOctetString(

component[1:], internalFormat=True,

prepend=bitString, padding=trailingBits

)

yield self._createComponent(asn1Spec, tagSet, bitString, **options)

class OctetStringPayloadDecoder(AbstractSimplePayloadDecoder):

protoComponent = univ.OctetString('')

supportConstructedForm = True

def valueDecoder(self, substrate, asn1Spec,

tagSet=None, length=None, state=None,

decodeFun=None, substrateFun=None,

**options):

if substrateFun:

asn1Object = self._createComponent(asn1Spec, tagSet, noValue, **options)

for chunk in substrateFun(asn1Object, substrate, length, options):

yield chunk

return

if tagSet[0].tagFormat == tag.tagFormatSimple: # XXX what tag to check?

for chunk in readFromStream(substrate, length, options):

if isinstance(chunk, SubstrateUnderrunError):

yield chunk

yield self._createComponent(asn1Spec, tagSet, chunk, **options)

return

if not self.supportConstructedForm:

raise error.PyAsn1Error('Constructed encoding form prohibited at %s' % self.__class__.__name__)

if LOG:

LOG('assembling constructed serialization')

# All inner fragments are of the same type, treat them as octet string

substrateFun = self.substrateCollector

header = b''

original_position = substrate.tell()

# head = popSubstream(substrate, length)

while substrate.tell() - original_position < length:

for component in decodeFun(

substrate, self.protoComponent, substrateFun=substrateFun,

**options):

if isinstance(component, SubstrateUnderrunError):

yield component

header += component

yield self._createComponent(asn1Spec, tagSet, header, **options)

def indefLenValueDecoder(self, substrate, asn1Spec,

tagSet=None, length=None, state=None,

decodeFun=None, substrateFun=None,

**options):

if substrateFun and substrateFun is not self.substrateCollector:

asn1Object = self._createComponent(asn1Spec, tagSet, noValue, **options)

for chunk in substrateFun(asn1Object, substrate, length, options):

yield chunk

return

# All inner fragments are of the same type, treat them as octet string

substrateFun = self.substrateCollector

header = b''

while True: # loop over fragments

for component in decodeFun(

substrate, self.protoComponent, substrateFun=substrateFun,

allowEoo=True, **options):

if isinstance(component, SubstrateUnderrunError):

yield component

if component is eoo.endOfOctets:

break

if component is eoo.endOfOctets:

break

header += component

yield self._createComponent(asn1Spec, tagSet, header, **options)

class NullPayloadDecoder(AbstractSimplePayloadDecoder):

protoComponent = univ.Null('')

def valueDecoder(self, substrate, asn1Spec,

tagSet=None, length=None, state=None,

decodeFun=None, substrateFun=None,

**options):

if tagSet[0].tagFormat != tag.tagFormatSimple:

raise error.PyAsn1Error('Simple tag format expected')

for chunk in readFromStream(substrate, length, options):

if isinstance(chunk, SubstrateUnderrunError):

yield chunk

component = self._createComponent(asn1Spec, tagSet, '', **options)

if chunk:

raise error.PyAsn1Error('Unexpected %d-octet substrate for Null' % length)

yield component

class ObjectIdentifierPayloadDecoder(AbstractSimplePayloadDecoder):

protoComponent = univ.ObjectIdentifier(())

def valueDecoder(self, substrate, asn1Spec,

tagSet=None, length=None, state=None,

decodeFun=None, substrateFun=None,

**options):

if tagSet[0].tagFormat != tag.tagFormatSimple:

raise error.PyAsn1Error('Simple tag format expected')

for chunk in readFromStream(substrate, length, options):

if isinstance(chunk, SubstrateUnderrunError):

yield chunk

if not chunk:

raise error.PyAsn1Error('Empty substrate')

oid = ()

index = 0

substrateLen = len(chunk)

while index < substrateLen:

subId = chunk[index]

index += 1

if subId < 128:

oid += (subId,)

elif subId > 128:

# Construct subid from a number of octets

nextSubId = subId

subId = 0

continuationOctetCount = 0

while nextSubId >= 128:

continuationOctetCount += 1

if continuationOctetCount > MAX_OID_ARC_CONTINUATION_OCTETS:

raise error.PyAsn1Error(

'OID arc exceeds maximum continuation octets limit (%d) '

'at position %d' % (MAX_OID_ARC_CONTINUATION_OCTETS, index)

)

subId = (subId << 7) + (nextSubId & 0x7F)

if index >= substrateLen:

raise error.SubstrateUnderrunError(

'Short substrate for sub-OID past %s' % (oid,)

)

nextSubId = chunk[index]

index += 1

oid += ((subId << 7) + nextSubId,)

elif subId == 128:

# ASN.1 spec forbids leading zeros (0x80) in OID

# encoding, tolerating it opens a vulnerability. See

# https://www.esat.kuleuven.be/cosic/publications/article-1432.pdf

# page 7

raise error.PyAsn1Error('Invalid octet 0x80 in OID encoding')

# Decode two leading arcs

if 0 <= oid[0] <= 39:

oid = (0,) + oid

elif 40 <= oid[0] <= 79:

oid = (1, oid[0] - 40) + oid[1:]

elif oid[0] >= 80:

oid = (2, oid[0] - 80) + oid[1:]

else:

raise error.PyAsn1Error('Malformed first OID octet: %s' % chunk[0])

yield self._createComponent(asn1Spec, tagSet, oid, **options)

class RelativeOIDPayloadDecoder(AbstractSimplePayloadDecoder):

protoComponent = univ.RelativeOID(())

def valueDecoder(self, substrate, asn1Spec,

tagSet=None, length=None, state=None,

decodeFun=None, substrateFun=None,

**options):

if tagSet[0].tagFormat != tag.tagFormatSimple:

raise error.PyAsn1Error('Simple tag format expected')

for chunk in readFromStream(substrate, length, options):

if isinstance(chunk, SubstrateUnderrunError):

yield chunk

if not chunk:

raise error.PyAsn1Error('Empty substrate')

reloid = ()

index = 0

substrateLen = len(chunk)

while index < substrateLen:

subId = chunk[index]

index += 1

if subId < 128:

reloid += (subId,)

elif subId > 128:

# Construct subid from a number of octets

nextSubId = subId

subId = 0

continuationOctetCount = 0

while nextSubId >= 128:

continuationOctetCount += 1

if continuationOctetCount > MAX_OID_ARC_CONTINUATION_OCTETS:

raise error.PyAsn1Error(

'RELATIVE-OID arc exceeds maximum continuation octets limit (%d) '

'at position %d' % (MAX_OID_ARC_CONTINUATION_OCTETS, index)

)

subId = (subId << 7) + (nextSubId & 0x7F)

if index >= substrateLen:

raise error.SubstrateUnderrunError(

'Short substrate for sub-OID past %s' % (reloid,)

)

nextSubId = chunk[index]

index += 1

reloid += ((subId << 7) + nextSubId,)

elif subId == 128:

# ASN.1 spec forbids leading zeros (0x80) in OID

# encoding, tolerating it opens a vulnerability. See

# https://www.esat.kuleuven.be/cosic/publications/article-1432.pdf

# page 7

raise error.PyAsn1Error('Invalid octet 0x80 in RELATIVE-OID encoding')

yield self._createComponent(asn1Spec, tagSet, reloid, **options)

class RealPayloadDecoder(AbstractSimplePayloadDecoder):

protoComponent = univ.Real()

def valueDecoder(self, substrate, asn1Spec,

tagSet=None, length=None, state=None,

decodeFun=None, substrateFun=None,

**options):

if tagSet[0].tagFormat != tag.tagFormatSimple:

raise error.PyAsn1Error('Simple tag format expected')

for chunk in readFromStream(substrate, length, options):

if isinstance(chunk, SubstrateUnderrunError):

yield chunk

if not chunk:

yield self._createComponent(asn1Spec, tagSet, 0.0, **options)

return

fo = chunk[0]

chunk = chunk[1:]

if fo & 0x80: # binary encoding

if not chunk:

raise error.PyAsn1Error("Incomplete floating-point value")

if LOG:

LOG('decoding binary encoded REAL')

n = (fo & 0x03) + 1

if n == 4:

n = chunk[0]

chunk = chunk[1:]

eo, chunk = chunk[:n], chunk[n:]

if not eo or not chunk:

raise error.PyAsn1Error('Real exponent screwed')

e = eo[0] & 0x80 and -1 or 0

while eo: # exponent

e <<= 8

e |= eo[0]

eo = eo[1:]

b = fo >> 4 & 0x03 # base bits

if b > 2:

raise error.PyAsn1Error('Illegal Real base')

if b == 1: # encbase = 8

e *= 3

elif b == 2: # encbase = 16

e *= 4

p = 0

while chunk: # value

p <<= 8

p |= chunk[0]

chunk = chunk[1:]

if fo & 0x40: # sign bit

p = -p

sf = fo >> 2 & 0x03 # scale bits

p *= 2 ** sf

value = (p, 2, e)

elif fo & 0x40: # infinite value

if LOG:

LOG('decoding infinite REAL')

value = fo & 0x01 and '-inf' or 'inf'

elif fo & 0xc0 == 0: # character encoding

if not chunk:

raise error.PyAsn1Error("Incomplete floating-point value")

if LOG:

LOG('decoding character encoded REAL')

try:

if fo & 0x3 == 0x1: # NR1

value = (int(chunk), 10, 0)

elif fo & 0x3 == 0x2: # NR2

value = float(chunk)

elif fo & 0x3 == 0x3: # NR3

value = float(chunk)

else:

raise error.SubstrateUnderrunError(

'Unknown NR (tag %s)' % fo

)

except ValueError:

raise error.SubstrateUnderrunError(

'Bad character Real syntax'

)

else:

raise error.SubstrateUnderrunError(

'Unknown encoding (tag %s)' % fo

)

yield self._createComponent(asn1Spec, tagSet, value, **options)

class AbstractConstructedPayloadDecoder(AbstractPayloadDecoder):

protoComponent = None

class ConstructedPayloadDecoderBase(AbstractConstructedPayloadDecoder):

protoRecordComponent = None

protoSequenceComponent = None

def _getComponentTagMap(self, asn1Object, idx):

raise NotImplementedError

def _getComponentPositionByType(self, asn1Object, tagSet, idx):

raise NotImplementedError

def _decodeComponentsSchemaless(

self, substrate, tagSet=None, decodeFun=None,

length=None, **options):

asn1Object = None

components = []

componentTypes = set()

original_position = substrate.tell()

while length == -1 or substrate.tell() < original_position + length:

for component in decodeFun(substrate, **options):

if isinstance(component, SubstrateUnderrunError):

yield component

if length == -1 and component is eoo.endOfOctets:

break

components.append(component)

componentTypes.add(component.tagSet)

# Now we have to guess is it SEQUENCE/SET or SEQUENCE OF/SET OF

# The heuristics is:

# * 1+ components of different types -> likely SEQUENCE/SET

# * otherwise -> likely SEQUENCE OF/SET OF

if len(componentTypes) > 1:

protoComponent = self.protoRecordComponent

else:

protoComponent = self.protoSequenceComponent

asn1Object = protoComponent.clone(

# construct tagSet from base tag from prototype ASN.1 object

# and additional tags recovered from the substrate

tagSet=tag.TagSet(protoComponent.tagSet.baseTag, *tagSet.superTags)

)

if LOG:

LOG('guessed %r container type (pass `asn1Spec` to guide the '

'decoder)' % asn1Object)

for idx, component in enumerate(components):

asn1Object.setComponentByPosition(

idx, component,

verifyConstraints=False,

matchTags=False, matchConstraints=False

)

yield asn1Object

def valueDecoder(self, substrate, asn1Spec,

tagSet=None, length=None, state=None,

decodeFun=None, substrateFun=None,

**options):

if tagSet[0].tagFormat != tag.tagFormatConstructed:

raise error.PyAsn1Error('Constructed tag format expected')

original_position = substrate.tell()

if substrateFun:

if asn1Spec is not None:

asn1Object = asn1Spec.clone()

elif self.protoComponent is not None:

asn1Object = self.protoComponent.clone(tagSet=tagSet)

else:

asn1Object = self.protoRecordComponent, self.protoSequenceComponent

for chunk in substrateFun(asn1Object, substrate, length, options):

yield chunk

return

if asn1Spec is None:

for asn1Object in self._decodeComponentsSchemaless(

substrate, tagSet=tagSet, decodeFun=decodeFun,

length=length, **options):

if isinstance(asn1Object, SubstrateUnderrunError):

yield asn1Object

if substrate.tell() < original_position + length:

if LOG:

for trailing in readFromStream(substrate, context=options):

if isinstance(trailing, SubstrateUnderrunError):

yield trailing

LOG('Unused trailing %d octets encountered: %s' % (

len(trailing), debug.hexdump(trailing)))

yield asn1Object

return

asn1Object = asn1Spec.clone()

asn1Object.clear()

options = self._passAsn1Object(asn1Object, options)

if asn1Spec.typeId in (univ.Sequence.typeId, univ.Set.typeId):

namedTypes = asn1Spec.componentType

isSetType = asn1Spec.typeId == univ.Set.typeId

isDeterministic = not isSetType and not namedTypes.hasOptionalOrDefault

if LOG:

LOG('decoding %sdeterministic %s type %r chosen by type ID' % (

not isDeterministic and 'non-' or '', isSetType and 'SET' or '',

asn1Spec))

seenIndices = set()

idx = 0

while substrate.tell() - original_position < length:

if not namedTypes:

componentType = None

elif isSetType:

componentType = namedTypes.tagMapUnique

else:

try:

if isDeterministic:

componentType = namedTypes[idx].asn1Object

elif namedTypes[idx].isOptional or namedTypes[idx].isDefaulted:

componentType = namedTypes.getTagMapNearPosition(idx)

else:

componentType = namedTypes[idx].asn1Object

except IndexError:

raise error.PyAsn1Error(

'Excessive components decoded at %r' % (asn1Spec,)

)

for component in decodeFun(substrate, componentType, **options):

if isinstance(component, SubstrateUnderrunError):

yield component

if not isDeterministic and namedTypes:

if isSetType:

idx = namedTypes.getPositionByType(component.effectiveTagSet)

elif namedTypes[idx].isOptional or namedTypes[idx].isDefaulted:

idx = namedTypes.getPositionNearType(component.effectiveTagSet, idx)

asn1Object.setComponentByPosition(

idx, component,

verifyConstraints=False,

matchTags=False, matchConstraints=False

)

seenIndices.add(idx)

idx += 1

if LOG:

LOG('seen component indices %s' % seenIndices)

if namedTypes:

if not namedTypes.requiredComponents.issubset(seenIndices):

raise error.PyAsn1Error(

'ASN.1 object %s has uninitialized '

'components' % asn1Object.__class__.__name__)

if namedTypes.hasOpenTypes:

openTypes = options.get('openTypes', {})

if LOG:

LOG('user-specified open types map:')

for k, v in openTypes.items():

LOG('%s -> %r' % (k, v))

if openTypes or options.get('decodeOpenTypes', False):

for idx, namedType in enumerate(namedTypes.namedTypes):

if not namedType.openType:

continue

if namedType.isOptional and not asn1Object.getComponentByPosition(idx).isValue:

continue

governingValue = asn1Object.getComponentByName(

namedType.openType.name

)

try:

openType = openTypes[governingValue]

except KeyError:

if LOG:

LOG('default open types map of component '

'"%s.%s" governed by component "%s.%s"'

':' % (asn1Object.__class__.__name__,

namedType.name,

asn1Object.__class__.__name__,

namedType.openType.name))

for k, v in namedType.openType.items():

LOG('%s -> %r' % (k, v))

try:

openType = namedType.openType[governingValue]

except KeyError:

if LOG:

LOG('failed to resolve open type by governing '

'value %r' % (governingValue,))

continue

if LOG:

LOG('resolved open type %r by governing '

'value %r' % (openType, governingValue))

containerValue = asn1Object.getComponentByPosition(idx)

if containerValue.typeId in (

univ.SetOf.typeId, univ.SequenceOf.typeId):

for pos, containerElement in enumerate(

containerValue):

stream = asSeekableStream(containerValue[pos].asOctets())

for component in decodeFun(stream, asn1Spec=openType, **options):

if isinstance(component, SubstrateUnderrunError):

yield component

containerValue[pos] = component

else:

stream = asSeekableStream(asn1Object.getComponentByPosition(idx).asOctets())

for component in decodeFun(stream, asn1Spec=openType, **options):

if isinstance(component, SubstrateUnderrunError):

yield component

asn1Object.setComponentByPosition(idx, component)

else:

inconsistency = asn1Object.isInconsistent

if inconsistency:

raise error.PyAsn1Error(

f"ASN.1 object {asn1Object.__class__.__name__} is inconsistent")

else:

componentType = asn1Spec.componentType

if LOG:

LOG('decoding type %r chosen by given `asn1Spec`' % componentType)

idx = 0

while substrate.tell() - original_position < length:

for component in decodeFun(substrate, componentType, **options):

if isinstance(component, SubstrateUnderrunError):

yield component

asn1Object.setComponentByPosition(

idx, component,

verifyConstraints=False,

matchTags=False, matchConstraints=False

)

idx += 1

yield asn1Object

def indefLenValueDecoder(self, substrate, asn1Spec,

tagSet=None, length=None, state=None,

decodeFun=None, substrateFun=None,

**options):

if tagSet[0].tagFormat != tag.tagFormatConstructed:

raise error.PyAsn1Error('Constructed tag format expected')

if substrateFun is not None:

if asn1Spec is not None:

asn1Object = asn1Spec.clone()

elif self.protoComponent is not None:

asn1Object = self.protoComponent.clone(tagSet=tagSet)

else:

asn1Object = self.protoRecordComponent, self.protoSequenceComponent

for chunk in substrateFun(asn1Object, substrate, length, options):

yield chunk

return

if asn1Spec is None:

for asn1Object in self._decodeComponentsSchemaless(

substrate, tagSet=tagSet, decodeFun=decodeFun,

length=length, **dict(options, allowEoo=True)):

if isinstance(asn1Object, SubstrateUnderrunError):

yield asn1Object

yield asn1Object

return

asn1Object = asn1Spec.clone()

asn1Object.clear()

options = self._passAsn1Object(asn1Object, options)

if asn1Spec.typeId in (univ.Sequence.typeId, univ.Set.typeId):

namedTypes = asn1Object.componentType

isSetType = asn1Object.typeId == univ.Set.typeId

isDeterministic = not isSetType and not namedTypes.hasOptionalOrDefault

if LOG:

LOG('decoding %sdeterministic %s type %r chosen by type ID' % (

not isDeterministic and 'non-' or '', isSetType and 'SET' or '',

asn1Spec))

seenIndices = set()

idx = 0

while True: # loop over components

if len(namedTypes) <= idx:

asn1Spec = None

elif isSetType:

asn1Spec = namedTypes.tagMapUnique

else:

try:

if isDeterministic:

asn1Spec = namedTypes[idx].asn1Object

elif namedTypes[idx].isOptional or namedTypes[idx].isDefaulted:

asn1Spec = namedTypes.getTagMapNearPosition(idx)

else:

asn1Spec = namedTypes[idx].asn1Object

except IndexError:

raise error.PyAsn1Error(

'Excessive components decoded at %r' % (asn1Object,)

)