Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1#
2# This file is part of pyasn1 software.
3#
4# Copyright (c) 2005-2020, Ilya Etingof <etingof@gmail.com>
5# License: https://pyasn1.readthedocs.io/en/latest/license.html
6#
7import io
8import os
9import sys
10import warnings
12from pyasn1 import debug
13from pyasn1 import error
14from pyasn1.codec.ber import eoo
15from pyasn1.codec.streaming import asSeekableStream
16from pyasn1.codec.streaming import isEndOfStream
17from pyasn1.codec.streaming import peekIntoStream
18from pyasn1.codec.streaming import readFromStream
19from pyasn1.compat import _MISSING
20from pyasn1.error import PyAsn1Error
21from pyasn1.type import base
22from pyasn1.type import char
23from pyasn1.type import tag
24from pyasn1.type import tagmap
25from pyasn1.type import univ
26from pyasn1.type import useful
28__all__ = ['StreamingDecoder', 'Decoder', 'decode']
30LOG = debug.registerLoggee(__name__, flags=debug.DEBUG_DECODER)
32noValue = base.noValue
34SubstrateUnderrunError = error.SubstrateUnderrunError
36# Maximum number of continuation octets (high-bit set) allowed per OID arc.
37# 20 octets allows up to 140-bit integers, supporting UUID-based OIDs
38MAX_OID_ARC_CONTINUATION_OCTETS = 20
40# Maximum number of octets in a long-form tag ID (20 octets = up to
41# 140-bit tag IDs, matching the OID arc limit)
42MAX_TAG_OCTETS = 20
43MAX_NESTING_DEPTH = 100
45# Maximum number of bytes in a BER length field (8 bytes = up to 2^64-1)
46MAX_LENGTH_OCTETS = 8
49class AbstractPayloadDecoder(object):
50 protoComponent = None
52 def valueDecoder(self, substrate, asn1Spec,
53 tagSet=None, length=None, state=None,
54 decodeFun=None, substrateFun=None,
55 **options):
56 """Decode value with fixed byte length.
58 The decoder is allowed to consume as many bytes as necessary.
59 """
60 raise error.PyAsn1Error('SingleItemDecoder not implemented for %s' % (tagSet,)) # TODO: Seems more like an NotImplementedError?
62 def indefLenValueDecoder(self, substrate, asn1Spec,
63 tagSet=None, length=None, state=None,
64 decodeFun=None, substrateFun=None,
65 **options):
66 """Decode value with undefined length.
68 The decoder is allowed to consume as many bytes as necessary.
69 """
70 raise error.PyAsn1Error('Indefinite length mode decoder not implemented for %s' % (tagSet,)) # TODO: Seems more like an NotImplementedError?
72 @staticmethod
73 def _passAsn1Object(asn1Object, options):
74 if 'asn1Object' not in options:
75 options['asn1Object'] = asn1Object
77 return options
80class AbstractSimplePayloadDecoder(AbstractPayloadDecoder):
81 @staticmethod
82 def substrateCollector(asn1Object, substrate, length, options):
83 for chunk in readFromStream(substrate, length, options):
84 yield chunk
86 def _createComponent(self, asn1Spec, tagSet, value, **options):
87 if options.get('native'):
88 return value
89 elif asn1Spec is None:
90 return self.protoComponent.clone(value, tagSet=tagSet)
91 elif value is noValue:
92 return asn1Spec
93 else:
94 return asn1Spec.clone(value)
97class RawPayloadDecoder(AbstractSimplePayloadDecoder):
98 protoComponent = univ.Any('')
100 def valueDecoder(self, substrate, asn1Spec,
101 tagSet=None, length=None, state=None,
102 decodeFun=None, substrateFun=None,
103 **options):
104 if substrateFun:
105 asn1Object = self._createComponent(asn1Spec, tagSet, '', **options)
107 for chunk in substrateFun(asn1Object, substrate, length, options):
108 yield chunk
110 return
112 for value in decodeFun(substrate, asn1Spec, tagSet, length, **options):
113 yield value
115 def indefLenValueDecoder(self, substrate, asn1Spec,
116 tagSet=None, length=None, state=None,
117 decodeFun=None, substrateFun=None,
118 **options):
119 if substrateFun:
120 asn1Object = self._createComponent(asn1Spec, tagSet, '', **options)
122 for chunk in substrateFun(asn1Object, substrate, length, options):
123 yield chunk
125 return
127 while True:
128 for value in decodeFun(
129 substrate, asn1Spec, tagSet, length,
130 allowEoo=True, **options):
132 if value is eoo.endOfOctets:
133 return
135 yield value
138rawPayloadDecoder = RawPayloadDecoder()
141class IntegerPayloadDecoder(AbstractSimplePayloadDecoder):
142 protoComponent = univ.Integer(0)
144 def valueDecoder(self, substrate, asn1Spec,
145 tagSet=None, length=None, state=None,
146 decodeFun=None, substrateFun=None,
147 **options):
149 if tagSet[0].tagFormat != tag.tagFormatSimple:
150 raise error.PyAsn1Error('Simple tag format expected')
152 for chunk in readFromStream(substrate, length, options):
153 if isinstance(chunk, SubstrateUnderrunError):
154 yield chunk
156 if chunk:
157 value = int.from_bytes(bytes(chunk), 'big', signed=True)
159 else:
160 value = 0
162 yield self._createComponent(asn1Spec, tagSet, value, **options)
165class BooleanPayloadDecoder(IntegerPayloadDecoder):
166 protoComponent = univ.Boolean(0)
168 def _createComponent(self, asn1Spec, tagSet, value, **options):
169 return IntegerPayloadDecoder._createComponent(
170 self, asn1Spec, tagSet, value and 1 or 0, **options)
173class BitStringPayloadDecoder(AbstractSimplePayloadDecoder):
174 protoComponent = univ.BitString(())
175 supportConstructedForm = True
177 @staticmethod
178 def _appendFragment(assembled, component, bitLength):
179 trailingBits = component[0]
180 if trailingBits > 7:
181 raise error.PyAsn1Error(
182 'Trailing bits overflow %s' % trailingBits
183 )
185 payload = component[1:]
186 if not payload:
187 if trailingBits:
188 raise error.PyAsn1Error('Trailing bits overflow %s' % trailingBits)
189 return bitLength
191 if not trailingBits and not bitLength % 8:
192 assembled.extend(payload)
193 return bitLength + len(payload) * 8
195 for idx, byte in enumerate(payload):
196 significantBits = 8
197 if idx == len(payload) - 1:
198 significantBits -= trailingBits
199 byte >>= trailingBits
201 usedBits = bitLength % 8
202 if not usedBits:
203 assembled.append(byte << (8 - significantBits))
204 elif usedBits + significantBits <= 8:
205 assembled[-1] |= byte << (8 - usedBits - significantBits)
206 else:
207 remainingBits = usedBits + significantBits - 8
208 assembled[-1] |= byte >> remainingBits
209 assembled.append((byte << (8 - remainingBits)) & 0xff)
211 bitLength += significantBits
213 return bitLength
215 def valueDecoder(self, substrate, asn1Spec,
216 tagSet=None, length=None, state=None,
217 decodeFun=None, substrateFun=None,
218 **options):
220 if substrateFun:
221 asn1Object = self._createComponent(asn1Spec, tagSet, noValue, **options)
223 for chunk in substrateFun(asn1Object, substrate, length, options):
224 yield chunk
226 return
228 if not length:
229 raise error.PyAsn1Error('Empty BIT STRING substrate')
231 for chunk in isEndOfStream(substrate):
232 if isinstance(chunk, SubstrateUnderrunError):
233 yield chunk
235 if chunk:
236 raise error.PyAsn1Error('Empty BIT STRING substrate')
238 if tagSet[0].tagFormat == tag.tagFormatSimple: # XXX what tag to check?
240 for trailingBits in readFromStream(substrate, 1, options):
241 if isinstance(trailingBits, SubstrateUnderrunError):
242 yield trailingBits
244 trailingBits = ord(trailingBits)
245 if trailingBits > 7:
246 raise error.PyAsn1Error(
247 'Trailing bits overflow %s' % trailingBits
248 )
250 for chunk in readFromStream(substrate, length - 1, options):
251 if isinstance(chunk, SubstrateUnderrunError):
252 yield chunk
254 value = self.protoComponent.fromOctetString(
255 chunk, internalFormat=True, padding=trailingBits)
257 yield self._createComponent(asn1Spec, tagSet, value, **options)
259 return
261 if not self.supportConstructedForm:
262 raise error.PyAsn1Error('Constructed encoding form prohibited '
263 'at %s' % self.__class__.__name__)
265 if LOG:
266 LOG('assembling constructed serialization')
268 # All inner fragments are of the same type, treat them as octet string
269 substrateFun = self.substrateCollector
271 assembled = bytearray()
272 bitLength = 0
274 current_position = substrate.tell()
276 while substrate.tell() - current_position < length:
277 for component in decodeFun(
278 substrate, self.protoComponent, substrateFun=substrateFun,
279 **options):
280 if isinstance(component, SubstrateUnderrunError):
281 yield component
283 bitLength = self._appendFragment(assembled, component, bitLength)
285 bitString = self.protoComponent.fromOctetString(
286 assembled, internalFormat=True, padding=(-bitLength) % 8
287 )
289 yield self._createComponent(asn1Spec, tagSet, bitString, **options)
291 def indefLenValueDecoder(self, substrate, asn1Spec,
292 tagSet=None, length=None, state=None,
293 decodeFun=None, substrateFun=None,
294 **options):
296 if substrateFun:
297 asn1Object = self._createComponent(asn1Spec, tagSet, noValue, **options)
299 for chunk in substrateFun(asn1Object, substrate, length, options):
300 yield chunk
302 return
304 # All inner fragments are of the same type, treat them as octet string
305 substrateFun = self.substrateCollector
307 assembled = bytearray()
308 bitLength = 0
310 while True: # loop over fragments
312 for component in decodeFun(
313 substrate, self.protoComponent, substrateFun=substrateFun,
314 allowEoo=True, **options):
316 if component is eoo.endOfOctets:
317 break
319 if isinstance(component, SubstrateUnderrunError):
320 yield component
322 if component is eoo.endOfOctets:
323 break
325 bitLength = self._appendFragment(assembled, component, bitLength)
327 bitString = self.protoComponent.fromOctetString(
328 assembled, internalFormat=True, padding=(-bitLength) % 8
329 )
331 yield self._createComponent(asn1Spec, tagSet, bitString, **options)
334class OctetStringPayloadDecoder(AbstractSimplePayloadDecoder):
335 protoComponent = univ.OctetString('')
336 supportConstructedForm = True
338 def valueDecoder(self, substrate, asn1Spec,
339 tagSet=None, length=None, state=None,
340 decodeFun=None, substrateFun=None,
341 **options):
342 if substrateFun:
343 asn1Object = self._createComponent(asn1Spec, tagSet, noValue, **options)
345 for chunk in substrateFun(asn1Object, substrate, length, options):
346 yield chunk
348 return
350 if tagSet[0].tagFormat == tag.tagFormatSimple: # XXX what tag to check?
351 for chunk in readFromStream(substrate, length, options):
352 if isinstance(chunk, SubstrateUnderrunError):
353 yield chunk
355 yield self._createComponent(asn1Spec, tagSet, chunk, **options)
357 return
359 if not self.supportConstructedForm:
360 raise error.PyAsn1Error('Constructed encoding form prohibited at %s' % self.__class__.__name__)
362 if LOG:
363 LOG('assembling constructed serialization')
365 # Assemble the logical value of each fragment, including nested ones.
366 fragments = []
368 original_position = substrate.tell()
369 # head = popSubstream(substrate, length)
370 while substrate.tell() - original_position < length:
371 for component in decodeFun(
372 substrate, self.protoComponent, **options):
373 if isinstance(component, SubstrateUnderrunError):
374 yield component
376 fragments.append(bytes(component))
378 yield self._createComponent(asn1Spec, tagSet, b''.join(fragments), **options)
380 def indefLenValueDecoder(self, substrate, asn1Spec,
381 tagSet=None, length=None, state=None,
382 decodeFun=None, substrateFun=None,
383 **options):
384 if substrateFun and substrateFun is not self.substrateCollector:
385 asn1Object = self._createComponent(asn1Spec, tagSet, noValue, **options)
387 for chunk in substrateFun(asn1Object, substrate, length, options):
388 yield chunk
390 return
392 # Assemble the logical value of each fragment, including nested ones.
393 fragments = []
395 while True: # loop over fragments
397 for component in decodeFun(
398 substrate, self.protoComponent, allowEoo=True, **options):
400 if isinstance(component, SubstrateUnderrunError):
401 yield component
403 if component is eoo.endOfOctets:
404 break
406 if component is eoo.endOfOctets:
407 break
409 fragments.append(bytes(component))
411 yield self._createComponent(asn1Spec, tagSet, b''.join(fragments), **options)
414class NullPayloadDecoder(AbstractSimplePayloadDecoder):
415 protoComponent = univ.Null('')
417 def valueDecoder(self, substrate, asn1Spec,
418 tagSet=None, length=None, state=None,
419 decodeFun=None, substrateFun=None,
420 **options):
422 if tagSet[0].tagFormat != tag.tagFormatSimple:
423 raise error.PyAsn1Error('Simple tag format expected')
425 for chunk in readFromStream(substrate, length, options):
426 if isinstance(chunk, SubstrateUnderrunError):
427 yield chunk
429 component = self._createComponent(asn1Spec, tagSet, '', **options)
431 if chunk:
432 raise error.PyAsn1Error('Unexpected %d-octet substrate for Null' % length)
434 yield component
437class ObjectIdentifierPayloadDecoder(AbstractSimplePayloadDecoder):
438 protoComponent = univ.ObjectIdentifier(())
440 def valueDecoder(self, substrate, asn1Spec,
441 tagSet=None, length=None, state=None,
442 decodeFun=None, substrateFun=None,
443 **options):
444 if tagSet[0].tagFormat != tag.tagFormatSimple:
445 raise error.PyAsn1Error('Simple tag format expected')
447 for chunk in readFromStream(substrate, length, options):
448 if isinstance(chunk, SubstrateUnderrunError):
449 yield chunk
451 if not chunk:
452 raise error.PyAsn1Error('Empty substrate')
454 oid = []
455 index = 0
456 substrateLen = len(chunk)
457 while index < substrateLen:
458 subId = chunk[index]
459 index += 1
460 if subId < 128:
461 oid.append(subId)
462 elif subId > 128:
463 # Construct subid from a number of octets
464 nextSubId = subId
465 subId = 0
466 continuationOctetCount = 0
467 while nextSubId >= 128:
468 continuationOctetCount += 1
469 if continuationOctetCount > MAX_OID_ARC_CONTINUATION_OCTETS:
470 raise error.PyAsn1Error(
471 'OID arc exceeds maximum continuation octets limit (%d) '
472 'at position %d' % (MAX_OID_ARC_CONTINUATION_OCTETS, index)
473 )
474 subId = (subId << 7) + (nextSubId & 0x7F)
475 if index >= substrateLen:
476 raise error.SubstrateUnderrunError(
477 'Short substrate for sub-OID past %s' % (tuple(oid),)
478 )
479 nextSubId = chunk[index]
480 index += 1
481 oid.append((subId << 7) + nextSubId)
482 elif subId == 128:
483 # ASN.1 spec forbids leading zeros (0x80) in OID
484 # encoding, tolerating it opens a vulnerability. See
485 # https://www.esat.kuleuven.be/cosic/publications/article-1432.pdf
486 # page 7
487 raise error.PyAsn1Error('Invalid octet 0x80 in OID encoding')
489 # Decode two leading arcs
490 if 0 <= oid[0] <= 39:
491 oid.insert(0, 0)
492 elif 40 <= oid[0] <= 79:
493 oid[0] -= 40
494 oid.insert(0, 1)
495 elif oid[0] >= 80:
496 oid[0] -= 80
497 oid.insert(0, 2)
498 else:
499 raise error.PyAsn1Error('Malformed first OID octet: %s' % chunk[0])
501 yield self._createComponent(asn1Spec, tagSet, tuple(oid), **options)
504class RelativeOIDPayloadDecoder(AbstractSimplePayloadDecoder):
505 protoComponent = univ.RelativeOID(())
507 def valueDecoder(self, substrate, asn1Spec,
508 tagSet=None, length=None, state=None,
509 decodeFun=None, substrateFun=None,
510 **options):
511 if tagSet[0].tagFormat != tag.tagFormatSimple:
512 raise error.PyAsn1Error('Simple tag format expected')
514 for chunk in readFromStream(substrate, length, options):
515 if isinstance(chunk, SubstrateUnderrunError):
516 yield chunk
518 if not chunk:
519 raise error.PyAsn1Error('Empty substrate')
521 reloid = []
522 index = 0
523 substrateLen = len(chunk)
524 while index < substrateLen:
525 subId = chunk[index]
526 index += 1
527 if subId < 128:
528 reloid.append(subId)
529 elif subId > 128:
530 # Construct subid from a number of octets
531 nextSubId = subId
532 subId = 0
533 continuationOctetCount = 0
534 while nextSubId >= 128:
535 continuationOctetCount += 1
536 if continuationOctetCount > MAX_OID_ARC_CONTINUATION_OCTETS:
537 raise error.PyAsn1Error(
538 'RELATIVE-OID arc exceeds maximum continuation octets limit (%d) '
539 'at position %d' % (MAX_OID_ARC_CONTINUATION_OCTETS, index)
540 )
541 subId = (subId << 7) + (nextSubId & 0x7F)
542 if index >= substrateLen:
543 raise error.SubstrateUnderrunError(
544 'Short substrate for sub-OID past %s' % (tuple(reloid),)
545 )
546 nextSubId = chunk[index]
547 index += 1
548 reloid.append((subId << 7) + nextSubId)
549 elif subId == 128:
550 # ASN.1 spec forbids leading zeros (0x80) in OID
551 # encoding, tolerating it opens a vulnerability. See
552 # https://www.esat.kuleuven.be/cosic/publications/article-1432.pdf
553 # page 7
554 raise error.PyAsn1Error('Invalid octet 0x80 in RELATIVE-OID encoding')
556 yield self._createComponent(asn1Spec, tagSet, tuple(reloid), **options)
559class RealPayloadDecoder(AbstractSimplePayloadDecoder):
560 protoComponent = univ.Real()
562 def valueDecoder(self, substrate, asn1Spec,
563 tagSet=None, length=None, state=None,
564 decodeFun=None, substrateFun=None,
565 **options):
566 if tagSet[0].tagFormat != tag.tagFormatSimple:
567 raise error.PyAsn1Error('Simple tag format expected')
569 for chunk in readFromStream(substrate, length, options):
570 if isinstance(chunk, SubstrateUnderrunError):
571 yield chunk
573 if not chunk:
574 yield self._createComponent(asn1Spec, tagSet, 0.0, **options)
575 return
577 fo = chunk[0]
578 chunk = chunk[1:]
579 if fo & 0x80: # binary encoding
580 if not chunk:
581 raise error.PyAsn1Error("Incomplete floating-point value")
583 if LOG:
584 LOG('decoding binary encoded REAL')
586 n = (fo & 0x03) + 1
588 if n == 4:
589 n = chunk[0]
590 chunk = chunk[1:]
592 eo, chunk = chunk[:n], chunk[n:]
594 if not eo or not chunk:
595 raise error.PyAsn1Error('Real exponent screwed')
597 e = eo[0] & 0x80 and -1 or 0
599 while eo: # exponent
600 e <<= 8
601 e |= eo[0]
602 eo = eo[1:]
604 b = fo >> 4 & 0x03 # base bits
606 if b > 2:
607 raise error.PyAsn1Error('Illegal Real base')
609 if b == 1: # encbase = 8
610 e *= 3
612 elif b == 2: # encbase = 16
613 e *= 4
614 p = int.from_bytes(bytes(chunk), 'big')
616 if fo & 0x40: # sign bit
617 p = -p
619 sf = fo >> 2 & 0x03 # scale bits
620 p *= 2 ** sf
621 value = (p, 2, e)
623 elif fo & 0x40: # infinite value
624 if LOG:
625 LOG('decoding infinite REAL')
627 value = fo & 0x01 and '-inf' or 'inf'
629 elif fo & 0xc0 == 0: # character encoding
630 if not chunk:
631 raise error.PyAsn1Error("Incomplete floating-point value")
633 if LOG:
634 LOG('decoding character encoded REAL')
636 try:
637 if fo & 0x3 == 0x1: # NR1
638 value = (int(chunk), 10, 0)
640 elif fo & 0x3 == 0x2: # NR2
641 value = float(chunk)
643 elif fo & 0x3 == 0x3: # NR3
644 value = float(chunk)
646 else:
647 raise error.SubstrateUnderrunError(
648 'Unknown NR (tag %s)' % fo
649 )
651 except ValueError:
652 raise error.SubstrateUnderrunError(
653 'Bad character Real syntax'
654 )
656 else:
657 raise error.SubstrateUnderrunError(
658 'Unknown encoding (tag %s)' % fo
659 )
661 yield self._createComponent(asn1Spec, tagSet, value, **options)
664class AbstractConstructedPayloadDecoder(AbstractPayloadDecoder):
665 protoComponent = None
668class ConstructedPayloadDecoderBase(AbstractConstructedPayloadDecoder):
669 protoRecordComponent = None
670 protoSequenceComponent = None
672 def _getComponentTagMap(self, asn1Object, idx):
673 raise NotImplementedError
675 def _getComponentPositionByType(self, asn1Object, tagSet, idx):
676 raise NotImplementedError
678 def _decodeComponentsSchemaless(
679 self, substrate, tagSet=None, decodeFun=None,
680 length=None, **options):
682 asn1Object = None
684 components = []
685 componentTypes = set()
687 original_position = substrate.tell()
689 while length == -1 or substrate.tell() < original_position + length:
690 for component in decodeFun(substrate, **options):
691 if isinstance(component, SubstrateUnderrunError):
692 yield component
694 if length == -1 and component is eoo.endOfOctets:
695 break
697 components.append(component)
698 # Two distinct types settle the choice; keep the set bounded
699 # to avoid quadratic comparisons for colliding tag hashes.
700 if len(componentTypes) < 2:
701 componentTypes.add(component.tagSet)
703 # Now we have to guess is it SEQUENCE/SET or SEQUENCE OF/SET OF
704 # The heuristics is:
705 # * 1+ components of different types -> likely SEQUENCE/SET
706 # * otherwise -> likely SEQUENCE OF/SET OF
707 if len(componentTypes) > 1:
708 protoComponent = self.protoRecordComponent
710 else:
711 protoComponent = self.protoSequenceComponent
713 asn1Object = protoComponent.clone(
714 # construct tagSet from base tag from prototype ASN.1 object
715 # and additional tags recovered from the substrate
716 tagSet=tag.TagSet(protoComponent.tagSet.baseTag, *tagSet.superTags)
717 )
719 if LOG:
720 LOG('guessed %r container type (pass `asn1Spec` to guide the '
721 'decoder)' % asn1Object)
723 for idx, component in enumerate(components):
724 asn1Object.setComponentByPosition(
725 idx, component,
726 verifyConstraints=False,
727 matchTags=False, matchConstraints=False
728 )
730 yield asn1Object
732 def valueDecoder(self, substrate, asn1Spec,
733 tagSet=None, length=None, state=None,
734 decodeFun=None, substrateFun=None,
735 **options):
736 if tagSet[0].tagFormat != tag.tagFormatConstructed:
737 raise error.PyAsn1Error('Constructed tag format expected')
739 original_position = substrate.tell()
741 if substrateFun:
742 if asn1Spec is not None:
743 asn1Object = asn1Spec.clone()
745 elif self.protoComponent is not None:
746 asn1Object = self.protoComponent.clone(tagSet=tagSet)
748 else:
749 asn1Object = self.protoRecordComponent, self.protoSequenceComponent
751 for chunk in substrateFun(asn1Object, substrate, length, options):
752 yield chunk
754 return
756 if asn1Spec is None:
757 for asn1Object in self._decodeComponentsSchemaless(
758 substrate, tagSet=tagSet, decodeFun=decodeFun,
759 length=length, **options):
760 if isinstance(asn1Object, SubstrateUnderrunError):
761 yield asn1Object
763 if substrate.tell() < original_position + length:
764 if LOG:
765 for trailing in readFromStream(substrate, context=options):
766 if isinstance(trailing, SubstrateUnderrunError):
767 yield trailing
769 LOG('Unused trailing %d octets encountered: %s' % (
770 len(trailing), debug.hexdump(trailing)))
772 yield asn1Object
774 return
776 asn1Object = asn1Spec.clone()
777 asn1Object.clear()
779 options = self._passAsn1Object(asn1Object, options)
781 if asn1Spec.typeId in (univ.Sequence.typeId, univ.Set.typeId):
783 namedTypes = asn1Spec.componentType
785 isSetType = asn1Spec.typeId == univ.Set.typeId
786 isDeterministic = not isSetType and not namedTypes.hasOptionalOrDefault
788 if LOG:
789 LOG('decoding %sdeterministic %s type %r chosen by type ID' % (
790 not isDeterministic and 'non-' or '', isSetType and 'SET' or '',
791 asn1Spec))
793 seenIndices = set()
794 idx = 0
795 while substrate.tell() - original_position < length:
796 if not namedTypes:
797 componentType = None
799 elif isSetType:
800 componentType = namedTypes.tagMapUnique
802 else:
803 try:
804 if isDeterministic:
805 componentType = namedTypes[idx].asn1Object
807 elif namedTypes[idx].isOptional or namedTypes[idx].isDefaulted:
808 componentType = namedTypes.getTagMapNearPosition(idx)
810 else:
811 componentType = namedTypes[idx].asn1Object
813 except IndexError:
814 raise error.PyAsn1Error(
815 'Excessive components decoded at %r' % (asn1Spec,)
816 )
818 for component in decodeFun(substrate, componentType, **options):
819 if isinstance(component, SubstrateUnderrunError):
820 yield component
822 if not isDeterministic and namedTypes:
823 if isSetType:
824 idx = namedTypes.getPositionByType(component.effectiveTagSet)
826 elif namedTypes[idx].isOptional or namedTypes[idx].isDefaulted:
827 idx = namedTypes.getPositionNearType(component.effectiveTagSet, idx)
829 asn1Object.setComponentByPosition(
830 idx, component,
831 verifyConstraints=False,
832 matchTags=False, matchConstraints=False
833 )
835 seenIndices.add(idx)
836 idx += 1
838 if LOG:
839 LOG('seen component indices %s' % seenIndices)
841 if namedTypes:
842 if not namedTypes.requiredComponents.issubset(seenIndices):
843 raise error.PyAsn1Error(
844 'ASN.1 object %s has uninitialized '
845 'components' % asn1Object.__class__.__name__)
847 if namedTypes.hasOpenTypes:
849 openTypes = options.get('openTypes', {})
851 if LOG:
852 LOG('user-specified open types map:')
854 for k, v in openTypes.items():
855 LOG('%s -> %r' % (k, v))
857 if openTypes or options.get('decodeOpenTypes', False):
859 for idx, namedType in enumerate(namedTypes.namedTypes):
860 if not namedType.openType:
861 continue
863 if namedType.isOptional and not asn1Object.getComponentByPosition(idx).isValue:
864 continue
866 governingValue = asn1Object.getComponentByName(
867 namedType.openType.name
868 )
870 try:
871 openType = openTypes[governingValue]
873 except KeyError:
875 if LOG:
876 LOG('default open types map of component '
877 '"%s.%s" governed by component "%s.%s"'
878 ':' % (asn1Object.__class__.__name__,
879 namedType.name,
880 asn1Object.__class__.__name__,
881 namedType.openType.name))
883 for k, v in namedType.openType.items():
884 LOG('%s -> %r' % (k, v))
886 try:
887 openType = namedType.openType[governingValue]
889 except KeyError:
890 if LOG:
891 LOG('failed to resolve open type by governing '
892 'value %r' % (governingValue,))
893 continue
895 if LOG:
896 LOG('resolved open type %r by governing '
897 'value %r' % (openType, governingValue))
899 containerValue = asn1Object.getComponentByPosition(idx)
901 if containerValue.typeId in (
902 univ.SetOf.typeId, univ.SequenceOf.typeId):
904 for pos, containerElement in enumerate(
905 containerValue):
907 stream = asSeekableStream(containerValue[pos].asOctets())
909 for component in decodeFun(stream, asn1Spec=openType, **options):
910 if isinstance(component, SubstrateUnderrunError):
911 yield component
913 containerValue[pos] = component
915 else:
916 stream = asSeekableStream(asn1Object.getComponentByPosition(idx).asOctets())
918 for component in decodeFun(stream, asn1Spec=openType, **options):
919 if isinstance(component, SubstrateUnderrunError):
920 yield component
922 asn1Object.setComponentByPosition(idx, component)
924 else:
925 inconsistency = asn1Object.isInconsistent
926 if inconsistency:
927 raise error.PyAsn1Error(
928 f"ASN.1 object {asn1Object.__class__.__name__} is inconsistent")
930 else:
931 componentType = asn1Spec.componentType
933 if LOG:
934 LOG('decoding type %r chosen by given `asn1Spec`' % componentType)
936 idx = 0
938 while substrate.tell() - original_position < length:
939 for component in decodeFun(substrate, componentType, **options):
940 if isinstance(component, SubstrateUnderrunError):
941 yield component
943 asn1Object.setComponentByPosition(
944 idx, component,
945 verifyConstraints=False,
946 matchTags=False, matchConstraints=False
947 )
949 idx += 1
951 yield asn1Object
953 def indefLenValueDecoder(self, substrate, asn1Spec,
954 tagSet=None, length=None, state=None,
955 decodeFun=None, substrateFun=None,
956 **options):
957 if tagSet[0].tagFormat != tag.tagFormatConstructed:
958 raise error.PyAsn1Error('Constructed tag format expected')
960 if substrateFun is not None:
961 if asn1Spec is not None:
962 asn1Object = asn1Spec.clone()
964 elif self.protoComponent is not None:
965 asn1Object = self.protoComponent.clone(tagSet=tagSet)
967 else:
968 asn1Object = self.protoRecordComponent, self.protoSequenceComponent
970 for chunk in substrateFun(asn1Object, substrate, length, options):
971 yield chunk
973 return
975 if asn1Spec is None:
976 for asn1Object in self._decodeComponentsSchemaless(
977 substrate, tagSet=tagSet, decodeFun=decodeFun,
978 length=length, **dict(options, allowEoo=True)):
979 if isinstance(asn1Object, SubstrateUnderrunError):
980 yield asn1Object
982 yield asn1Object
984 return
986 asn1Object = asn1Spec.clone()
987 asn1Object.clear()
989 options = self._passAsn1Object(asn1Object, options)
991 if asn1Spec.typeId in (univ.Sequence.typeId, univ.Set.typeId):
993 namedTypes = asn1Object.componentType
995 isSetType = asn1Object.typeId == univ.Set.typeId
996 isDeterministic = not isSetType and not namedTypes.hasOptionalOrDefault
998 if LOG:
999 LOG('decoding %sdeterministic %s type %r chosen by type ID' % (
1000 not isDeterministic and 'non-' or '', isSetType and 'SET' or '',
1001 asn1Spec))
1003 seenIndices = set()
1005 idx = 0
1007 while True: # loop over components
1008 if len(namedTypes) <= idx:
1009 asn1Spec = None
1011 elif isSetType:
1012 asn1Spec = namedTypes.tagMapUnique
1014 else:
1015 try:
1016 if isDeterministic:
1017 asn1Spec = namedTypes[idx].asn1Object
1019 elif namedTypes[idx].isOptional or namedTypes[idx].isDefaulted:
1020 asn1Spec = namedTypes.getTagMapNearPosition(idx)
1022 else:
1023 asn1Spec = namedTypes[idx].asn1Object
1025 except IndexError:
1026 raise error.PyAsn1Error(
1027 'Excessive components decoded at %r' % (asn1Object,)
1028 )
1030 for component in decodeFun(substrate, asn1Spec, allowEoo=True, **options):
1032 if isinstance(component, SubstrateUnderrunError):
1033 yield component
1035 if component is eoo.endOfOctets:
1036 break
1038 if component is eoo.endOfOctets:
1039 break
1041 if not isDeterministic and namedTypes:
1042 if isSetType:
1043 idx = namedTypes.getPositionByType(component.effectiveTagSet)
1045 elif namedTypes[idx].isOptional or namedTypes[idx].isDefaulted:
1046 idx = namedTypes.getPositionNearType(component.effectiveTagSet, idx)
1048 asn1Object.setComponentByPosition(
1049 idx, component,
1050 verifyConstraints=False,
1051 matchTags=False, matchConstraints=False
1052 )
1054 seenIndices.add(idx)
1055 idx += 1
1057 if LOG:
1058 LOG('seen component indices %s' % seenIndices)
1060 if namedTypes:
1061 if not namedTypes.requiredComponents.issubset(seenIndices):
1062 raise error.PyAsn1Error(
1063 'ASN.1 object %s has uninitialized '
1064 'components' % asn1Object.__class__.__name__)
1066 if namedTypes.hasOpenTypes:
1068 openTypes = options.get('openTypes', {})
1070 if LOG:
1071 LOG('user-specified open types map:')
1073 for k, v in openTypes.items():
1074 LOG('%s -> %r' % (k, v))
1076 if openTypes or options.get('decodeOpenTypes', False):
1078 for idx, namedType in enumerate(namedTypes.namedTypes):
1079 if not namedType.openType:
1080 continue
1082 if namedType.isOptional and not asn1Object.getComponentByPosition(idx).isValue:
1083 continue
1085 governingValue = asn1Object.getComponentByName(
1086 namedType.openType.name
1087 )
1089 try:
1090 openType = openTypes[governingValue]
1092 except KeyError:
1094 if LOG:
1095 LOG('default open types map of component '
1096 '"%s.%s" governed by component "%s.%s"'
1097 ':' % (asn1Object.__class__.__name__,
1098 namedType.name,
1099 asn1Object.__class__.__name__,
1100 namedType.openType.name))
1102 for k, v in namedType.openType.items():
1103 LOG('%s -> %r' % (k, v))
1105 try:
1106 openType = namedType.openType[governingValue]
1108 except KeyError:
1109 if LOG:
1110 LOG('failed to resolve open type by governing '
1111 'value %r' % (governingValue,))
1112 continue
1114 if LOG:
1115 LOG('resolved open type %r by governing '
1116 'value %r' % (openType, governingValue))
1118 containerValue = asn1Object.getComponentByPosition(idx)
1120 if containerValue.typeId in (
1121 univ.SetOf.typeId, univ.SequenceOf.typeId):
1123 for pos, containerElement in enumerate(
1124 containerValue):
1126 stream = asSeekableStream(containerValue[pos].asOctets())
1128 for component in decodeFun(stream, asn1Spec=openType,
1129 **dict(options, allowEoo=True)):
1130 if isinstance(component, SubstrateUnderrunError):
1131 yield component
1133 if component is eoo.endOfOctets:
1134 break
1136 containerValue[pos] = component
1138 else:
1139 stream = asSeekableStream(asn1Object.getComponentByPosition(idx).asOctets())
1140 for component in decodeFun(stream, asn1Spec=openType,
1141 **dict(options, allowEoo=True)):
1142 if isinstance(component, SubstrateUnderrunError):
1143 yield component
1145 if component is eoo.endOfOctets:
1146 break
1148 asn1Object.setComponentByPosition(idx, component)
1150 else:
1151 inconsistency = asn1Object.isInconsistent
1152 if inconsistency:
1153 raise error.PyAsn1Error(
1154 f"ASN.1 object {asn1Object.__class__.__name__} is inconsistent")
1156 else:
1157 componentType = asn1Spec.componentType
1159 if LOG:
1160 LOG('decoding type %r chosen by given `asn1Spec`' % componentType)
1162 idx = 0
1164 while True:
1166 for component in decodeFun(
1167 substrate, componentType, allowEoo=True, **options):
1169 if isinstance(component, SubstrateUnderrunError):
1170 yield component
1172 if component is eoo.endOfOctets:
1173 break
1175 if component is eoo.endOfOctets:
1176 break
1178 asn1Object.setComponentByPosition(
1179 idx, component,
1180 verifyConstraints=False,
1181 matchTags=False, matchConstraints=False
1182 )
1184 idx += 1
1186 yield asn1Object
1189class SequenceOrSequenceOfPayloadDecoder(ConstructedPayloadDecoderBase):
1190 protoRecordComponent = univ.Sequence()
1191 protoSequenceComponent = univ.SequenceOf()
1194class SequencePayloadDecoder(SequenceOrSequenceOfPayloadDecoder):
1195 protoComponent = univ.Sequence()
1198class SequenceOfPayloadDecoder(SequenceOrSequenceOfPayloadDecoder):
1199 protoComponent = univ.SequenceOf()
1202class SetOrSetOfPayloadDecoder(ConstructedPayloadDecoderBase):
1203 protoRecordComponent = univ.Set()
1204 protoSequenceComponent = univ.SetOf()
1207class SetPayloadDecoder(SetOrSetOfPayloadDecoder):
1208 protoComponent = univ.Set()
1211class SetOfPayloadDecoder(SetOrSetOfPayloadDecoder):
1212 protoComponent = univ.SetOf()
1215class ChoicePayloadDecoder(ConstructedPayloadDecoderBase):
1216 protoComponent = univ.Choice()
1218 def valueDecoder(self, substrate, asn1Spec,
1219 tagSet=None, length=None, state=None,
1220 decodeFun=None, substrateFun=None,
1221 **options):
1222 if asn1Spec is None:
1223 asn1Object = self.protoComponent.clone(tagSet=tagSet)
1225 else:
1226 asn1Object = asn1Spec.clone()
1228 if substrateFun:
1229 for chunk in substrateFun(asn1Object, substrate, length, options):
1230 yield chunk
1232 return
1234 options = self._passAsn1Object(asn1Object, options)
1236 if asn1Object.tagSet == tagSet:
1237 if LOG:
1238 LOG('decoding %s as explicitly tagged CHOICE' % (tagSet,))
1240 for component in decodeFun(
1241 substrate, asn1Object.componentTagMap, **options):
1242 if isinstance(component, SubstrateUnderrunError):
1243 yield component
1245 else:
1246 if LOG:
1247 LOG('decoding %s as untagged CHOICE' % (tagSet,))
1249 for component in decodeFun(
1250 substrate, asn1Object.componentTagMap, tagSet, length,
1251 state, **options):
1252 if isinstance(component, SubstrateUnderrunError):
1253 yield component
1255 effectiveTagSet = component.effectiveTagSet
1257 if LOG:
1258 LOG('decoded component %s, effective tag set %s' % (component, effectiveTagSet))
1260 asn1Object.setComponentByType(
1261 effectiveTagSet, component,
1262 verifyConstraints=False,
1263 matchTags=False, matchConstraints=False,
1264 innerFlag=False
1265 )
1267 yield asn1Object
1269 def indefLenValueDecoder(self, substrate, asn1Spec,
1270 tagSet=None, length=None, state=None,
1271 decodeFun=None, substrateFun=None,
1272 **options):
1273 if asn1Spec is None:
1274 asn1Object = self.protoComponent.clone(tagSet=tagSet)
1276 else:
1277 asn1Object = asn1Spec.clone()
1279 if substrateFun:
1280 for chunk in substrateFun(asn1Object, substrate, length, options):
1281 yield chunk
1283 return
1285 options = self._passAsn1Object(asn1Object, options)
1287 isTagged = asn1Object.tagSet == tagSet
1289 if LOG:
1290 LOG('decoding %s as %stagged CHOICE' % (
1291 tagSet, isTagged and 'explicitly ' or 'un'))
1293 while True:
1295 if isTagged:
1296 iterator = decodeFun(
1297 substrate, asn1Object.componentType.tagMapUnique,
1298 **dict(options, allowEoo=True))
1300 else:
1301 iterator = decodeFun(
1302 substrate, asn1Object.componentType.tagMapUnique,
1303 tagSet, length, state, **dict(options, allowEoo=True))
1305 for component in iterator:
1307 if isinstance(component, SubstrateUnderrunError):
1308 yield component
1310 if component is eoo.endOfOctets:
1311 break
1313 effectiveTagSet = component.effectiveTagSet
1315 if LOG:
1316 LOG('decoded component %s, effective tag set '
1317 '%s' % (component, effectiveTagSet))
1319 asn1Object.setComponentByType(
1320 effectiveTagSet, component,
1321 verifyConstraints=False,
1322 matchTags=False, matchConstraints=False,
1323 innerFlag=False
1324 )
1326 if not isTagged:
1327 break
1329 if not isTagged or component is eoo.endOfOctets:
1330 break
1332 yield asn1Object
1335class AnyPayloadDecoder(AbstractSimplePayloadDecoder):
1336 protoComponent = univ.Any()
1338 def valueDecoder(self, substrate, asn1Spec,
1339 tagSet=None, length=None, state=None,
1340 decodeFun=None, substrateFun=None,
1341 **options):
1342 if asn1Spec is None:
1343 isUntagged = True
1345 elif asn1Spec.__class__ is tagmap.TagMap:
1346 isUntagged = tagSet not in asn1Spec.tagMap
1348 else:
1349 isUntagged = tagSet != asn1Spec.tagSet
1351 if isUntagged:
1352 fullPosition = substrate.markedPosition
1353 currentPosition = substrate.tell()
1355 substrate.seek(fullPosition, os.SEEK_SET)
1356 length += currentPosition - fullPosition
1358 if LOG:
1359 for chunk in peekIntoStream(substrate, length):
1360 if isinstance(chunk, SubstrateUnderrunError):
1361 yield chunk
1362 LOG('decoding as untagged ANY, substrate '
1363 '%s' % debug.hexdump(chunk))
1365 if substrateFun:
1366 for chunk in substrateFun(
1367 self._createComponent(asn1Spec, tagSet, noValue, **options),
1368 substrate, length, options):
1369 yield chunk
1371 return
1373 for chunk in readFromStream(substrate, length, options):
1374 if isinstance(chunk, SubstrateUnderrunError):
1375 yield chunk
1377 yield self._createComponent(asn1Spec, tagSet, chunk, **options)
1379 def indefLenValueDecoder(self, substrate, asn1Spec,
1380 tagSet=None, length=None, state=None,
1381 decodeFun=None, substrateFun=None,
1382 **options):
1383 if asn1Spec is None:
1384 isTagged = False
1386 elif asn1Spec.__class__ is tagmap.TagMap:
1387 isTagged = tagSet in asn1Spec.tagMap
1389 else:
1390 isTagged = tagSet == asn1Spec.tagSet
1392 if isTagged:
1393 # tagged Any type -- consume header substrate
1394 chunk = b''
1396 if LOG:
1397 LOG('decoding as tagged ANY')
1399 else:
1400 # TODO: Seems not to be tested
1401 fullPosition = substrate.markedPosition
1402 currentPosition = substrate.tell()
1404 substrate.seek(fullPosition, os.SEEK_SET)
1405 for chunk in readFromStream(substrate, currentPosition - fullPosition, options):
1406 if isinstance(chunk, SubstrateUnderrunError):
1407 yield chunk
1409 if LOG:
1410 LOG('decoding as untagged ANY, header substrate %s' % debug.hexdump(chunk))
1412 # Any components do not inherit initial tag
1413 asn1Spec = self.protoComponent
1415 if substrateFun and substrateFun is not self.substrateCollector:
1416 asn1Object = self._createComponent(
1417 asn1Spec, tagSet, noValue, **options)
1419 for chunk in substrateFun(
1420 asn1Object, chunk + substrate, length + len(chunk), options):
1421 yield chunk
1423 return
1425 if LOG:
1426 LOG('assembling constructed serialization')
1428 # All inner fragments are of the same type, treat them as octet string
1429 substrateFun = self.substrateCollector
1431 fragments = [chunk]
1433 while True: # loop over fragments
1435 for component in decodeFun(
1436 substrate, asn1Spec, substrateFun=substrateFun,
1437 allowEoo=True, **options):
1439 if isinstance(component, SubstrateUnderrunError):
1440 yield component
1442 if component is eoo.endOfOctets:
1443 break
1445 if component is eoo.endOfOctets:
1446 break
1448 fragments.append(component)
1450 chunk = b''.join(fragments)
1452 if substrateFun:
1453 yield chunk # TODO: Weird
1455 else:
1456 yield self._createComponent(asn1Spec, tagSet, chunk, **options)
1459# character string types
1460class UTF8StringPayloadDecoder(OctetStringPayloadDecoder):
1461 protoComponent = char.UTF8String()
1464class NumericStringPayloadDecoder(OctetStringPayloadDecoder):
1465 protoComponent = char.NumericString()
1468class PrintableStringPayloadDecoder(OctetStringPayloadDecoder):
1469 protoComponent = char.PrintableString()
1472class TeletexStringPayloadDecoder(OctetStringPayloadDecoder):
1473 protoComponent = char.TeletexString()
1476class VideotexStringPayloadDecoder(OctetStringPayloadDecoder):
1477 protoComponent = char.VideotexString()
1480class IA5StringPayloadDecoder(OctetStringPayloadDecoder):
1481 protoComponent = char.IA5String()
1484class GraphicStringPayloadDecoder(OctetStringPayloadDecoder):
1485 protoComponent = char.GraphicString()
1488class VisibleStringPayloadDecoder(OctetStringPayloadDecoder):
1489 protoComponent = char.VisibleString()
1492class GeneralStringPayloadDecoder(OctetStringPayloadDecoder):
1493 protoComponent = char.GeneralString()
1496class UniversalStringPayloadDecoder(OctetStringPayloadDecoder):
1497 protoComponent = char.UniversalString()
1500class BMPStringPayloadDecoder(OctetStringPayloadDecoder):
1501 protoComponent = char.BMPString()
1504# "useful" types
1505class ObjectDescriptorPayloadDecoder(OctetStringPayloadDecoder):
1506 protoComponent = useful.ObjectDescriptor()
1509class GeneralizedTimePayloadDecoder(OctetStringPayloadDecoder):
1510 protoComponent = useful.GeneralizedTime()
1513class UTCTimePayloadDecoder(OctetStringPayloadDecoder):
1514 protoComponent = useful.UTCTime()
1517TAG_MAP = {
1518 univ.Integer.tagSet: IntegerPayloadDecoder(),
1519 univ.Boolean.tagSet: BooleanPayloadDecoder(),
1520 univ.BitString.tagSet: BitStringPayloadDecoder(),
1521 univ.OctetString.tagSet: OctetStringPayloadDecoder(),
1522 univ.Null.tagSet: NullPayloadDecoder(),
1523 univ.ObjectIdentifier.tagSet: ObjectIdentifierPayloadDecoder(),
1524 univ.RelativeOID.tagSet: RelativeOIDPayloadDecoder(),
1525 univ.Enumerated.tagSet: IntegerPayloadDecoder(),
1526 univ.Real.tagSet: RealPayloadDecoder(),
1527 univ.Sequence.tagSet: SequenceOrSequenceOfPayloadDecoder(), # conflicts with SequenceOf
1528 univ.Set.tagSet: SetOrSetOfPayloadDecoder(), # conflicts with SetOf
1529 univ.Choice.tagSet: ChoicePayloadDecoder(), # conflicts with Any
1530 # character string types
1531 char.UTF8String.tagSet: UTF8StringPayloadDecoder(),
1532 char.NumericString.tagSet: NumericStringPayloadDecoder(),
1533 char.PrintableString.tagSet: PrintableStringPayloadDecoder(),
1534 char.TeletexString.tagSet: TeletexStringPayloadDecoder(),
1535 char.VideotexString.tagSet: VideotexStringPayloadDecoder(),
1536 char.IA5String.tagSet: IA5StringPayloadDecoder(),
1537 char.GraphicString.tagSet: GraphicStringPayloadDecoder(),
1538 char.VisibleString.tagSet: VisibleStringPayloadDecoder(),
1539 char.GeneralString.tagSet: GeneralStringPayloadDecoder(),
1540 char.UniversalString.tagSet: UniversalStringPayloadDecoder(),
1541 char.BMPString.tagSet: BMPStringPayloadDecoder(),
1542 # useful types
1543 useful.ObjectDescriptor.tagSet: ObjectDescriptorPayloadDecoder(),
1544 useful.GeneralizedTime.tagSet: GeneralizedTimePayloadDecoder(),
1545 useful.UTCTime.tagSet: UTCTimePayloadDecoder()
1546}
1548# Type-to-codec map for ambiguous ASN.1 types
1549TYPE_MAP = {
1550 univ.Set.typeId: SetPayloadDecoder(),
1551 univ.SetOf.typeId: SetOfPayloadDecoder(),
1552 univ.Sequence.typeId: SequencePayloadDecoder(),
1553 univ.SequenceOf.typeId: SequenceOfPayloadDecoder(),
1554 univ.Choice.typeId: ChoicePayloadDecoder(),
1555 univ.Any.typeId: AnyPayloadDecoder()
1556}
1558# Put in non-ambiguous types for faster codec lookup
1559for typeDecoder in TAG_MAP.values():
1560 if typeDecoder.protoComponent is not None:
1561 typeId = typeDecoder.protoComponent.__class__.typeId
1562 if typeId is not None and typeId not in TYPE_MAP:
1563 TYPE_MAP[typeId] = typeDecoder
1566(stDecodeTag,
1567 stDecodeLength,
1568 stGetValueDecoder,
1569 stGetValueDecoderByAsn1Spec,
1570 stGetValueDecoderByTag,
1571 stTryAsExplicitTag,
1572 stDecodeValue,
1573 stDumpRawValue,
1574 stErrorCondition,
1575 stStop) = [x for x in range(10)]
1578EOO_SENTINEL = bytes((0, 0))
1581class SingleItemDecoder(object):
1582 defaultErrorState = stErrorCondition
1583 #defaultErrorState = stDumpRawValue
1584 defaultRawDecoder = AnyPayloadDecoder()
1586 supportIndefLength = True
1588 TAG_MAP = TAG_MAP
1589 TYPE_MAP = TYPE_MAP
1591 def __init__(self, tagMap=_MISSING, typeMap=_MISSING, **ignored):
1592 self._tagMap = tagMap if tagMap is not _MISSING else self.TAG_MAP
1593 self._typeMap = typeMap if typeMap is not _MISSING else self.TYPE_MAP
1595 # Tag & TagSet objects caches
1596 self._tagCache = {}
1597 self._tagSetCache = {}
1599 def __call__(self, substrate, asn1Spec=None,
1600 tagSet=None, length=None, state=stDecodeTag,
1601 decodeFun=None, substrateFun=None,
1602 **options):
1604 _nestingLevel = options.get('_nestingLevel', 0)
1606 if _nestingLevel > MAX_NESTING_DEPTH:
1607 raise error.PyAsn1Error(
1608 'ASN.1 structure nesting depth exceeds limit (%d)' % MAX_NESTING_DEPTH
1609 )
1611 options['_nestingLevel'] = _nestingLevel + 1
1613 allowEoo = options.pop('allowEoo', False)
1615 if LOG:
1616 LOG('decoder called at scope %s with state %d, working with up '
1617 'to %s octets of substrate: '
1618 '%s' % (debug.scope, state, length, substrate))
1620 # Look for end-of-octets sentinel
1621 if allowEoo and self.supportIndefLength:
1623 for eoo_candidate in readFromStream(substrate, 2, options):
1624 if isinstance(eoo_candidate, SubstrateUnderrunError):
1625 yield eoo_candidate
1627 if eoo_candidate == EOO_SENTINEL:
1628 if LOG:
1629 LOG('end-of-octets sentinel found')
1630 yield eoo.endOfOctets
1631 return
1633 else:
1634 substrate.seek(-2, os.SEEK_CUR)
1636 tagMap = self._tagMap
1637 typeMap = self._typeMap
1638 tagCache = self._tagCache
1639 tagSetCache = self._tagSetCache
1641 value = noValue
1643 substrate.markedPosition = substrate.tell()
1645 while state is not stStop:
1647 if state is stDecodeTag:
1648 # Decode tag
1649 isShortTag = True
1651 for firstByte in readFromStream(substrate, 1, options):
1652 if isinstance(firstByte, SubstrateUnderrunError):
1653 yield firstByte
1655 firstOctet = ord(firstByte)
1657 try:
1658 lastTag = tagCache[firstOctet]
1660 except KeyError:
1661 integerTag = firstOctet
1662 tagClass = integerTag & 0xC0
1663 tagFormat = integerTag & 0x20
1664 tagId = integerTag & 0x1F
1666 if tagId == 0x1F:
1667 isShortTag = False
1668 tagOctetCount = 0
1669 tagId = 0
1671 while True:
1672 for integerByte in readFromStream(substrate, 1, options):
1673 if isinstance(integerByte, SubstrateUnderrunError):
1674 yield integerByte
1676 if not integerByte:
1677 raise error.SubstrateUnderrunError(
1678 'Short octet stream on long tag decoding'
1679 )
1681 integerTag = ord(integerByte)
1682 tagOctetCount += 1
1683 if tagOctetCount > MAX_TAG_OCTETS:
1684 raise error.PyAsn1Error(
1685 'Tag ID octet count exceeds limit (%d)' % (
1686 MAX_TAG_OCTETS,)
1687 )
1688 tagId <<= 7
1689 tagId |= (integerTag & 0x7F)
1691 if not integerTag & 0x80:
1692 break
1694 lastTag = tag.Tag(
1695 tagClass=tagClass, tagFormat=tagFormat, tagId=tagId
1696 )
1698 if isShortTag:
1699 # cache short tags
1700 tagCache[firstOctet] = lastTag
1702 if tagSet is None:
1703 if isShortTag:
1704 try:
1705 tagSet = tagSetCache[firstOctet]
1707 except KeyError:
1708 # base tag not recovered
1709 tagSet = tag.TagSet((), lastTag)
1710 tagSetCache[firstOctet] = tagSet
1711 else:
1712 tagSet = tag.TagSet((), lastTag)
1714 else:
1715 tagSet = lastTag + tagSet
1717 state = stDecodeLength
1719 if LOG:
1720 LOG('tag decoded into %s, decoding length' % tagSet)
1722 if state is stDecodeLength:
1723 # Decode length
1724 for firstOctet in readFromStream(substrate, 1, options):
1725 if isinstance(firstOctet, SubstrateUnderrunError):
1726 yield firstOctet
1728 firstOctet = ord(firstOctet)
1730 if firstOctet < 128:
1731 length = firstOctet
1733 elif firstOctet > 128:
1734 size = firstOctet & 0x7F
1736 if size > MAX_LENGTH_OCTETS:
1737 raise error.PyAsn1Error(
1738 'BER length field size %d exceeds limit (%d)' % (
1739 size, MAX_LENGTH_OCTETS)
1740 )
1742 # encoded in size bytes
1743 for encodedLength in readFromStream(substrate, size, options):
1744 if isinstance(encodedLength, SubstrateUnderrunError):
1745 yield encodedLength
1746 encodedLength = list(encodedLength)
1747 if len(encodedLength) != size:
1748 raise error.SubstrateUnderrunError(
1749 '%s<%s at %s' % (size, len(encodedLength), tagSet)
1750 )
1752 length = 0
1753 for lengthOctet in encodedLength:
1754 length <<= 8
1755 length |= lengthOctet
1756 size += 1
1758 else: # 128 means indefinite
1759 length = -1
1761 if length == -1 and not self.supportIndefLength:
1762 raise error.PyAsn1Error('Indefinite length encoding not supported by this codec')
1764 state = stGetValueDecoder
1766 if LOG:
1767 LOG('value length decoded into %d' % length)
1769 if state is stGetValueDecoder:
1770 if asn1Spec is None:
1771 state = stGetValueDecoderByTag
1773 else:
1774 state = stGetValueDecoderByAsn1Spec
1775 #
1776 # There're two ways of creating subtypes in ASN.1 what influences
1777 # decoder operation. These methods are:
1778 # 1) Either base types used in or no IMPLICIT tagging has been
1779 # applied on subtyping.
1780 # 2) Subtype syntax drops base type information (by means of
1781 # IMPLICIT tagging.
1782 # The first case allows for complete tag recovery from substrate
1783 # while the second one requires original ASN.1 type spec for
1784 # decoding.
1785 #
1786 # In either case a set of tags (tagSet) is coming from substrate
1787 # in an incremental, tag-by-tag fashion (this is the case of
1788 # EXPLICIT tag which is most basic). Outermost tag comes first
1789 # from the wire.
1790 #
1791 if state is stGetValueDecoderByTag:
1792 try:
1793 concreteDecoder = tagMap[tagSet]
1795 except KeyError:
1796 concreteDecoder = None
1798 if concreteDecoder:
1799 state = stDecodeValue
1801 else:
1802 try:
1803 concreteDecoder = tagMap[tagSet[:1]]
1805 except KeyError:
1806 concreteDecoder = None
1808 if concreteDecoder:
1809 state = stDecodeValue
1810 else:
1811 state = stTryAsExplicitTag
1813 if LOG:
1814 LOG('codec %s chosen by a built-in type, decoding %s' % (concreteDecoder and concreteDecoder.__class__.__name__ or "<none>", state is stDecodeValue and 'value' or 'as explicit tag'))
1815 debug.scope.push(concreteDecoder is None and '?' or concreteDecoder.protoComponent.__class__.__name__)
1817 if state is stGetValueDecoderByAsn1Spec:
1819 if asn1Spec.__class__ is tagmap.TagMap:
1820 try:
1821 chosenSpec = asn1Spec[tagSet]
1823 except KeyError:
1824 chosenSpec = None
1826 if LOG:
1827 LOG('candidate ASN.1 spec is a map of:')
1829 for firstOctet, v in asn1Spec.presentTypes.items():
1830 LOG(' %s -> %s' % (firstOctet, v.__class__.__name__))
1832 if asn1Spec.skipTypes:
1833 LOG('but neither of: ')
1834 for firstOctet, v in asn1Spec.skipTypes.items():
1835 LOG(' %s -> %s' % (firstOctet, v.__class__.__name__))
1836 LOG('new candidate ASN.1 spec is %s, chosen by %s' % (chosenSpec is None and '<none>' or chosenSpec.prettyPrintType(), tagSet))
1838 elif tagSet == asn1Spec.tagSet or tagSet in asn1Spec.tagMap:
1839 chosenSpec = asn1Spec
1840 if LOG:
1841 LOG('candidate ASN.1 spec is %s' % asn1Spec.__class__.__name__)
1843 else:
1844 chosenSpec = None
1846 if chosenSpec is not None:
1847 try:
1848 # ambiguous type or just faster codec lookup
1849 concreteDecoder = typeMap[chosenSpec.typeId]
1851 if LOG:
1852 LOG('value decoder chosen for an ambiguous type by type ID %s' % (chosenSpec.typeId,))
1854 except KeyError:
1855 # use base type for codec lookup to recover untagged types
1856 baseTagSet = tag.TagSet(chosenSpec.tagSet.baseTag, chosenSpec.tagSet.baseTag)
1857 try:
1858 # base type or tagged subtype
1859 concreteDecoder = tagMap[baseTagSet]
1861 if LOG:
1862 LOG('value decoder chosen by base %s' % (baseTagSet,))
1864 except KeyError:
1865 concreteDecoder = None
1867 if concreteDecoder:
1868 asn1Spec = chosenSpec
1869 state = stDecodeValue
1871 else:
1872 state = stTryAsExplicitTag
1874 else:
1875 concreteDecoder = None
1876 state = stTryAsExplicitTag
1878 if LOG:
1879 LOG('codec %s chosen by ASN.1 spec, decoding %s' % (state is stDecodeValue and concreteDecoder.__class__.__name__ or "<none>", state is stDecodeValue and 'value' or 'as explicit tag'))
1880 debug.scope.push(chosenSpec is None and '?' or chosenSpec.__class__.__name__)
1882 if state is stDecodeValue:
1883 if not options.get('recursiveFlag', True) and not substrateFun: # deprecate this
1884 def substrateFun(asn1Object, _substrate, _length, _options):
1885 """Legacy hack to keep the recursiveFlag=False option supported.
1887 The decode(..., substrateFun=userCallback) option was introduced in 0.1.4 as a generalization
1888 of the old recursiveFlag=False option. Users should pass their callback instead of using
1889 recursiveFlag.
1890 """
1891 yield asn1Object
1893 original_position = substrate.tell()
1895 if length == -1: # indef length
1896 for value in concreteDecoder.indefLenValueDecoder(
1897 substrate, asn1Spec,
1898 tagSet, length, stGetValueDecoder,
1899 self, substrateFun, **options):
1900 if isinstance(value, SubstrateUnderrunError):
1901 yield value
1903 else:
1904 for value in concreteDecoder.valueDecoder(
1905 substrate, asn1Spec,
1906 tagSet, length, stGetValueDecoder,
1907 self, substrateFun, **options):
1908 if isinstance(value, SubstrateUnderrunError):
1909 yield value
1911 bytesRead = substrate.tell() - original_position
1912 if not substrateFun and bytesRead != length:
1913 raise PyAsn1Error(
1914 "Read %s bytes instead of expected %s." % (bytesRead, length))
1915 elif substrateFun and bytesRead > length:
1916 # custom substrateFun may be used for partial decoding, reading less is expected there
1917 raise PyAsn1Error(
1918 "Read %s bytes are more than expected %s." % (bytesRead, length))
1920 if LOG:
1921 LOG('codec %s yields type %s, value:\n%s\n...' % (
1922 concreteDecoder.__class__.__name__, value.__class__.__name__,
1923 isinstance(value, base.Asn1Item) and value.prettyPrint() or value))
1925 state = stStop
1926 break
1928 if state is stTryAsExplicitTag:
1929 if (tagSet and
1930 tagSet[0].tagFormat == tag.tagFormatConstructed and
1931 tagSet[0].tagClass != tag.tagClassUniversal):
1932 # Assume explicit tagging
1933 concreteDecoder = rawPayloadDecoder
1934 state = stDecodeValue
1936 else:
1937 concreteDecoder = None
1938 state = self.defaultErrorState
1940 if LOG:
1941 LOG('codec %s chosen, decoding %s' % (concreteDecoder and concreteDecoder.__class__.__name__ or "<none>", state is stDecodeValue and 'value' or 'as failure'))
1943 if state is stDumpRawValue:
1944 concreteDecoder = self.defaultRawDecoder
1946 if LOG:
1947 LOG('codec %s chosen, decoding value' % concreteDecoder.__class__.__name__)
1949 state = stDecodeValue
1951 if state is stErrorCondition:
1952 raise error.PyAsn1Error(
1953 '%s not in asn1Spec: %r' % (tagSet, asn1Spec)
1954 )
1956 if LOG:
1957 debug.scope.pop()
1958 LOG('decoder left scope %s, call completed' % debug.scope)
1960 yield value
1963class StreamingDecoder(object):
1964 """Create an iterator that turns BER/CER/DER byte stream into ASN.1 objects.
1966 On each iteration, consume whatever BER/CER/DER serialization is
1967 available in the `substrate` stream-like object and turns it into
1968 one or more, possibly nested, ASN.1 objects.
1970 Parameters
1971 ----------
1972 substrate: :py:class:`file`, :py:class:`io.BytesIO`
1973 BER/CER/DER serialization in form of a byte stream
1975 Keyword Args
1976 ------------
1977 asn1Spec: :py:class:`~pyasn1.type.base.PyAsn1Item`
1978 A pyasn1 type object to act as a template guiding the decoder.
1979 Depending on the ASN.1 structure being decoded, `asn1Spec` may
1980 or may not be required. One of the reasons why `asn1Spec` may
1981 me required is that ASN.1 structure is encoded in the *IMPLICIT*
1982 tagging mode.
1984 Yields
1985 ------
1986 : :py:class:`~pyasn1.type.base.PyAsn1Item`, :py:class:`~pyasn1.error.SubstrateUnderrunError`
1987 Decoded ASN.1 object (possibly, nested) or
1988 :py:class:`~pyasn1.error.SubstrateUnderrunError` object indicating
1989 insufficient BER/CER/DER serialization on input to fully recover ASN.1
1990 objects from it.
1992 In the latter case the caller is advised to ensure some more data in
1993 the input stream, then call the iterator again. The decoder will resume
1994 the decoding process using the newly arrived data.
1996 The `context` property of :py:class:`~pyasn1.error.SubstrateUnderrunError`
1997 object might hold a reference to the partially populated ASN.1 object
1998 being reconstructed.
2000 Raises
2001 ------
2002 ~pyasn1.error.PyAsn1Error, ~pyasn1.error.EndOfStreamError
2003 `PyAsn1Error` on deserialization error, `EndOfStreamError` on
2004 premature stream closure.
2006 Examples
2007 --------
2008 Decode BER serialisation without ASN.1 schema
2010 .. code-block:: pycon
2012 >>> stream = io.BytesIO(
2013 ... b'0\t\x02\x01\x01\x02\x01\x02\x02\x01\x03')
2014 >>>
2015 >>> for asn1Object in StreamingDecoder(stream):
2016 ... print(asn1Object)
2017 >>>
2018 SequenceOf:
2019 1 2 3
2021 Decode BER serialisation with ASN.1 schema
2023 .. code-block:: pycon
2025 >>> stream = io.BytesIO(
2026 ... b'0\t\x02\x01\x01\x02\x01\x02\x02\x01\x03')
2027 >>>
2028 >>> schema = SequenceOf(componentType=Integer())
2029 >>>
2030 >>> decoder = StreamingDecoder(stream, asn1Spec=schema)
2031 >>> for asn1Object in decoder:
2032 ... print(asn1Object)
2033 >>>
2034 SequenceOf:
2035 1 2 3
2036 """
2038 SINGLE_ITEM_DECODER = SingleItemDecoder
2040 def __init__(self, substrate, asn1Spec=None, **options):
2041 self._singleItemDecoder = self.SINGLE_ITEM_DECODER(**options)
2042 self._substrate = asSeekableStream(substrate)
2043 self._asn1Spec = asn1Spec
2044 self._options = options
2046 def __iter__(self):
2047 while True:
2048 for asn1Object in self._singleItemDecoder(
2049 self._substrate, self._asn1Spec, **self._options):
2050 yield asn1Object
2052 for chunk in isEndOfStream(self._substrate):
2053 if isinstance(chunk, SubstrateUnderrunError):
2054 yield
2056 break
2058 if chunk:
2059 break
2062class Decoder(object):
2063 """Create a BER decoder object.
2065 Parse BER/CER/DER octet-stream into one, possibly nested, ASN.1 object.
2066 """
2067 STREAMING_DECODER = StreamingDecoder
2069 @classmethod
2070 def __call__(cls, substrate, asn1Spec=None, **options):
2071 """Turns BER/CER/DER octet stream into an ASN.1 object.
2073 Takes BER/CER/DER octet-stream in form of :py:class:`bytes`
2074 and decode it into an ASN.1 object
2075 (e.g. :py:class:`~pyasn1.type.base.PyAsn1Item` derivative) which
2076 may be a scalar or an arbitrary nested structure.
2078 Parameters
2079 ----------
2080 substrate: :py:class:`bytes`
2081 BER/CER/DER octet-stream to parse
2083 Keyword Args
2084 ------------
2085 asn1Spec: :py:class:`~pyasn1.type.base.PyAsn1Item`
2086 A pyasn1 type object (:py:class:`~pyasn1.type.base.PyAsn1Item`
2087 derivative) to act as a template guiding the decoder.
2088 Depending on the ASN.1 structure being decoded, `asn1Spec` may or
2089 may not be required. Most common reason for it to require is that
2090 ASN.1 structure is encoded in *IMPLICIT* tagging mode.
2092 substrateFun: :py:class:`Union[
2093 Callable[[pyasn1.type.base.PyAsn1Item, bytes, int],
2094 Tuple[pyasn1.type.base.PyAsn1Item, bytes]],
2095 Callable[[pyasn1.type.base.PyAsn1Item, io.BytesIO, int, dict],
2096 Generator[Union[pyasn1.type.base.PyAsn1Item,
2097 pyasn1.error.SubstrateUnderrunError],
2098 None, None]]
2099 ]`
2100 User callback meant to generalize special use cases like non-recursive or
2101 partial decoding. A 3-arg non-streaming variant is supported for backwards
2102 compatiblilty in addition to the newer 4-arg streaming variant.
2103 The callback will receive the uninitialized object recovered from substrate
2104 as 1st argument, the uninterpreted payload as 2nd argument, and the length
2105 of the uninterpreted payload as 3rd argument. The streaming variant will
2106 additionally receive the decode(..., **options) kwargs as 4th argument.
2107 The non-streaming variant shall return an object that will be propagated
2108 as decode() return value as 1st item, and the remainig payload for further
2109 decode passes as 2nd item.
2110 The streaming variant shall yield an object that will be propagated as
2111 decode() return value, and leave the remaining payload in the stream.
2113 Returns
2114 -------
2115 : :py:class:`tuple`
2116 A tuple of :py:class:`~pyasn1.type.base.PyAsn1Item` object
2117 recovered from BER/CER/DER substrate and the unprocessed trailing
2118 portion of the `substrate` (may be empty)
2120 Raises
2121 ------
2122 : :py:class:`~pyasn1.error.PyAsn1Error`
2123 :py:class:`~pyasn1.error.SubstrateUnderrunError` on insufficient
2124 input or :py:class:`~pyasn1.error.PyAsn1Error` on decoding error.
2126 Examples
2127 --------
2128 Decode BER/CER/DER serialisation without ASN.1 schema
2130 .. code-block:: pycon
2132 >>> s, unprocessed = decode(b'0\t\x02\x01\x01\x02\x01\x02\x02\x01\x03')
2133 >>> str(s)
2134 SequenceOf:
2135 1 2 3
2137 Decode BER/CER/DER serialisation with ASN.1 schema
2139 .. code-block:: pycon
2141 >>> seq = SequenceOf(componentType=Integer())
2142 >>> s, unprocessed = decode(
2143 b'0\t\x02\x01\x01\x02\x01\x02\x02\x01\x03', asn1Spec=seq)
2144 >>> str(s)
2145 SequenceOf:
2146 1 2 3
2148 """
2149 substrate = asSeekableStream(substrate)
2151 if "substrateFun" in options:
2152 origSubstrateFun = options["substrateFun"]
2154 def substrateFunWrapper(asn1Object, substrate, length, options=None):
2155 """Support both 0.4 and 0.5 style APIs.
2157 substrateFun API has changed in 0.5 for use with streaming decoders. To stay backwards compatible,
2158 we first try if we received a streaming user callback. If that fails,we assume we've received a
2159 non-streaming v0.4 user callback and convert it for streaming on the fly
2160 """
2161 try:
2162 substrate_gen = origSubstrateFun(asn1Object, substrate, length, options)
2163 except TypeError as _value:
2164 if _value.__traceback__.tb_next:
2165 # Traceback depth > 1 means TypeError from inside user provided function
2166 raise
2167 # invariant maintained at Decoder.__call__ entry
2168 assert isinstance(substrate, io.BytesIO) # nosec assert_used
2169 substrate_gen = Decoder._callSubstrateFunV4asV5(origSubstrateFun, asn1Object, substrate, length)
2170 for value in substrate_gen:
2171 yield value
2173 options["substrateFun"] = substrateFunWrapper
2175 streamingDecoder = cls.STREAMING_DECODER(
2176 substrate, asn1Spec, **options)
2178 for asn1Object in streamingDecoder:
2179 if isinstance(asn1Object, SubstrateUnderrunError):
2180 raise error.SubstrateUnderrunError('Short substrate on input')
2182 try:
2183 tail = next(readFromStream(substrate))
2185 except error.EndOfStreamError:
2186 tail = b''
2188 return asn1Object, tail
2190 @staticmethod
2191 def _callSubstrateFunV4asV5(substrateFunV4, asn1Object, substrate, length):
2192 substrate_bytes = substrate.read()
2193 if length == -1:
2194 length = len(substrate_bytes)
2195 value, nextSubstrate = substrateFunV4(asn1Object, substrate_bytes, length)
2196 nbytes = substrate.write(nextSubstrate)
2197 substrate.truncate()
2198 substrate.seek(-nbytes, os.SEEK_CUR)
2199 yield value
2201#: Turns BER octet stream into an ASN.1 object.
2202#:
2203#: Takes BER octet-stream and decode it into an ASN.1 object
2204#: (e.g. :py:class:`~pyasn1.type.base.PyAsn1Item` derivative) which
2205#: may be a scalar or an arbitrary nested structure.
2206#:
2207#: Parameters
2208#: ----------
2209#: substrate: :py:class:`bytes`
2210#: BER octet-stream
2211#:
2212#: Keyword Args
2213#: ------------
2214#: asn1Spec: any pyasn1 type object e.g. :py:class:`~pyasn1.type.base.PyAsn1Item` derivative
2215#: A pyasn1 type object to act as a template guiding the decoder. Depending on the ASN.1 structure
2216#: being decoded, *asn1Spec* may or may not be required. Most common reason for
2217#: it to require is that ASN.1 structure is encoded in *IMPLICIT* tagging mode.
2218#:
2219#: Returns
2220#: -------
2221#: : :py:class:`tuple`
2222#: A tuple of pyasn1 object recovered from BER substrate (:py:class:`~pyasn1.type.base.PyAsn1Item` derivative)
2223#: and the unprocessed trailing portion of the *substrate* (may be empty)
2224#:
2225#: Raises
2226#: ------
2227#: ~pyasn1.error.PyAsn1Error, ~pyasn1.error.SubstrateUnderrunError
2228#: On decoding errors
2229#:
2230#: Notes
2231#: -----
2232#: This function is deprecated. Please use :py:class:`Decoder` or
2233#: :py:class:`StreamingDecoder` class instance.
2234#:
2235#: Examples
2236#: --------
2237#: Decode BER serialisation without ASN.1 schema
2238#:
2239#: .. code-block:: pycon
2240#:
2241#: >>> s, _ = decode(b'0\t\x02\x01\x01\x02\x01\x02\x02\x01\x03')
2242#: >>> str(s)
2243#: SequenceOf:
2244#: 1 2 3
2245#:
2246#: Decode BER serialisation with ASN.1 schema
2247#:
2248#: .. code-block:: pycon
2249#:
2250#: >>> seq = SequenceOf(componentType=Integer())
2251#: >>> s, _ = decode(b'0\t\x02\x01\x01\x02\x01\x02\x02\x01\x03', asn1Spec=seq)
2252#: >>> str(s)
2253#: SequenceOf:
2254#: 1 2 3
2255#:
2256decode = Decoder()
2258def __getattr__(attr: str):
2259 if newAttr := {"tagMap": "TAG_MAP", "typeMap": "TYPE_MAP"}.get(attr):
2260 warnings.warn(f"{attr} is deprecated. Please use {newAttr} instead.", DeprecationWarning, stacklevel=2)
2261 return globals()[newAttr]
2262 raise AttributeError(attr)