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 def valueDecoder(self, substrate, asn1Spec,
178 tagSet=None, length=None, state=None,
179 decodeFun=None, substrateFun=None,
180 **options):
182 if substrateFun:
183 asn1Object = self._createComponent(asn1Spec, tagSet, noValue, **options)
185 for chunk in substrateFun(asn1Object, substrate, length, options):
186 yield chunk
188 return
190 if not length:
191 raise error.PyAsn1Error('Empty BIT STRING substrate')
193 for chunk in isEndOfStream(substrate):
194 if isinstance(chunk, SubstrateUnderrunError):
195 yield chunk
197 if chunk:
198 raise error.PyAsn1Error('Empty BIT STRING substrate')
200 if tagSet[0].tagFormat == tag.tagFormatSimple: # XXX what tag to check?
202 for trailingBits in readFromStream(substrate, 1, options):
203 if isinstance(trailingBits, SubstrateUnderrunError):
204 yield trailingBits
206 trailingBits = ord(trailingBits)
207 if trailingBits > 7:
208 raise error.PyAsn1Error(
209 'Trailing bits overflow %s' % trailingBits
210 )
212 for chunk in readFromStream(substrate, length - 1, options):
213 if isinstance(chunk, SubstrateUnderrunError):
214 yield chunk
216 value = self.protoComponent.fromOctetString(
217 chunk, internalFormat=True, padding=trailingBits)
219 yield self._createComponent(asn1Spec, tagSet, value, **options)
221 return
223 if not self.supportConstructedForm:
224 raise error.PyAsn1Error('Constructed encoding form prohibited '
225 'at %s' % self.__class__.__name__)
227 if LOG:
228 LOG('assembling constructed serialization')
230 # All inner fragments are of the same type, treat them as octet string
231 substrateFun = self.substrateCollector
233 bitString = self.protoComponent.fromOctetString(b'', internalFormat=True)
235 current_position = substrate.tell()
237 while substrate.tell() - current_position < length:
238 for component in decodeFun(
239 substrate, self.protoComponent, substrateFun=substrateFun,
240 **options):
241 if isinstance(component, SubstrateUnderrunError):
242 yield component
244 trailingBits = component[0]
245 if trailingBits > 7:
246 raise error.PyAsn1Error(
247 'Trailing bits overflow %s' % trailingBits
248 )
250 bitString = self.protoComponent.fromOctetString(
251 component[1:], internalFormat=True,
252 prepend=bitString, padding=trailingBits
253 )
255 yield self._createComponent(asn1Spec, tagSet, bitString, **options)
257 def indefLenValueDecoder(self, substrate, asn1Spec,
258 tagSet=None, length=None, state=None,
259 decodeFun=None, substrateFun=None,
260 **options):
262 if substrateFun:
263 asn1Object = self._createComponent(asn1Spec, tagSet, noValue, **options)
265 for chunk in substrateFun(asn1Object, substrate, length, options):
266 yield chunk
268 return
270 # All inner fragments are of the same type, treat them as octet string
271 substrateFun = self.substrateCollector
273 bitString = self.protoComponent.fromOctetString(b'', internalFormat=True)
275 while True: # loop over fragments
277 for component in decodeFun(
278 substrate, self.protoComponent, substrateFun=substrateFun,
279 allowEoo=True, **options):
281 if component is eoo.endOfOctets:
282 break
284 if isinstance(component, SubstrateUnderrunError):
285 yield component
287 if component is eoo.endOfOctets:
288 break
290 trailingBits = component[0]
291 if trailingBits > 7:
292 raise error.PyAsn1Error(
293 'Trailing bits overflow %s' % trailingBits
294 )
296 bitString = self.protoComponent.fromOctetString(
297 component[1:], internalFormat=True,
298 prepend=bitString, padding=trailingBits
299 )
301 yield self._createComponent(asn1Spec, tagSet, bitString, **options)
304class OctetStringPayloadDecoder(AbstractSimplePayloadDecoder):
305 protoComponent = univ.OctetString('')
306 supportConstructedForm = True
308 def valueDecoder(self, substrate, asn1Spec,
309 tagSet=None, length=None, state=None,
310 decodeFun=None, substrateFun=None,
311 **options):
312 if substrateFun:
313 asn1Object = self._createComponent(asn1Spec, tagSet, noValue, **options)
315 for chunk in substrateFun(asn1Object, substrate, length, options):
316 yield chunk
318 return
320 if tagSet[0].tagFormat == tag.tagFormatSimple: # XXX what tag to check?
321 for chunk in readFromStream(substrate, length, options):
322 if isinstance(chunk, SubstrateUnderrunError):
323 yield chunk
325 yield self._createComponent(asn1Spec, tagSet, chunk, **options)
327 return
329 if not self.supportConstructedForm:
330 raise error.PyAsn1Error('Constructed encoding form prohibited at %s' % self.__class__.__name__)
332 if LOG:
333 LOG('assembling constructed serialization')
335 # All inner fragments are of the same type, treat them as octet string
336 substrateFun = self.substrateCollector
338 header = b''
340 original_position = substrate.tell()
341 # head = popSubstream(substrate, length)
342 while substrate.tell() - original_position < length:
343 for component in decodeFun(
344 substrate, self.protoComponent, substrateFun=substrateFun,
345 **options):
346 if isinstance(component, SubstrateUnderrunError):
347 yield component
349 header += component
351 yield self._createComponent(asn1Spec, tagSet, header, **options)
353 def indefLenValueDecoder(self, substrate, asn1Spec,
354 tagSet=None, length=None, state=None,
355 decodeFun=None, substrateFun=None,
356 **options):
357 if substrateFun and substrateFun is not self.substrateCollector:
358 asn1Object = self._createComponent(asn1Spec, tagSet, noValue, **options)
360 for chunk in substrateFun(asn1Object, substrate, length, options):
361 yield chunk
363 return
365 # All inner fragments are of the same type, treat them as octet string
366 substrateFun = self.substrateCollector
368 header = b''
370 while True: # loop over fragments
372 for component in decodeFun(
373 substrate, self.protoComponent, substrateFun=substrateFun,
374 allowEoo=True, **options):
376 if isinstance(component, SubstrateUnderrunError):
377 yield component
379 if component is eoo.endOfOctets:
380 break
382 if component is eoo.endOfOctets:
383 break
385 header += component
387 yield self._createComponent(asn1Spec, tagSet, header, **options)
390class NullPayloadDecoder(AbstractSimplePayloadDecoder):
391 protoComponent = univ.Null('')
393 def valueDecoder(self, substrate, asn1Spec,
394 tagSet=None, length=None, state=None,
395 decodeFun=None, substrateFun=None,
396 **options):
398 if tagSet[0].tagFormat != tag.tagFormatSimple:
399 raise error.PyAsn1Error('Simple tag format expected')
401 for chunk in readFromStream(substrate, length, options):
402 if isinstance(chunk, SubstrateUnderrunError):
403 yield chunk
405 component = self._createComponent(asn1Spec, tagSet, '', **options)
407 if chunk:
408 raise error.PyAsn1Error('Unexpected %d-octet substrate for Null' % length)
410 yield component
413class ObjectIdentifierPayloadDecoder(AbstractSimplePayloadDecoder):
414 protoComponent = univ.ObjectIdentifier(())
416 def valueDecoder(self, substrate, asn1Spec,
417 tagSet=None, length=None, state=None,
418 decodeFun=None, substrateFun=None,
419 **options):
420 if tagSet[0].tagFormat != tag.tagFormatSimple:
421 raise error.PyAsn1Error('Simple tag format expected')
423 for chunk in readFromStream(substrate, length, options):
424 if isinstance(chunk, SubstrateUnderrunError):
425 yield chunk
427 if not chunk:
428 raise error.PyAsn1Error('Empty substrate')
430 oid = []
431 index = 0
432 substrateLen = len(chunk)
433 while index < substrateLen:
434 subId = chunk[index]
435 index += 1
436 if subId < 128:
437 oid.append(subId)
438 elif subId > 128:
439 # Construct subid from a number of octets
440 nextSubId = subId
441 subId = 0
442 continuationOctetCount = 0
443 while nextSubId >= 128:
444 continuationOctetCount += 1
445 if continuationOctetCount > MAX_OID_ARC_CONTINUATION_OCTETS:
446 raise error.PyAsn1Error(
447 'OID arc exceeds maximum continuation octets limit (%d) '
448 'at position %d' % (MAX_OID_ARC_CONTINUATION_OCTETS, index)
449 )
450 subId = (subId << 7) + (nextSubId & 0x7F)
451 if index >= substrateLen:
452 raise error.SubstrateUnderrunError(
453 'Short substrate for sub-OID past %s' % (tuple(oid),)
454 )
455 nextSubId = chunk[index]
456 index += 1
457 oid.append((subId << 7) + nextSubId)
458 elif subId == 128:
459 # ASN.1 spec forbids leading zeros (0x80) in OID
460 # encoding, tolerating it opens a vulnerability. See
461 # https://www.esat.kuleuven.be/cosic/publications/article-1432.pdf
462 # page 7
463 raise error.PyAsn1Error('Invalid octet 0x80 in OID encoding')
465 # Decode two leading arcs
466 if 0 <= oid[0] <= 39:
467 oid.insert(0, 0)
468 elif 40 <= oid[0] <= 79:
469 oid[0] -= 40
470 oid.insert(0, 1)
471 elif oid[0] >= 80:
472 oid[0] -= 80
473 oid.insert(0, 2)
474 else:
475 raise error.PyAsn1Error('Malformed first OID octet: %s' % chunk[0])
477 yield self._createComponent(asn1Spec, tagSet, tuple(oid), **options)
480class RelativeOIDPayloadDecoder(AbstractSimplePayloadDecoder):
481 protoComponent = univ.RelativeOID(())
483 def valueDecoder(self, substrate, asn1Spec,
484 tagSet=None, length=None, state=None,
485 decodeFun=None, substrateFun=None,
486 **options):
487 if tagSet[0].tagFormat != tag.tagFormatSimple:
488 raise error.PyAsn1Error('Simple tag format expected')
490 for chunk in readFromStream(substrate, length, options):
491 if isinstance(chunk, SubstrateUnderrunError):
492 yield chunk
494 if not chunk:
495 raise error.PyAsn1Error('Empty substrate')
497 reloid = []
498 index = 0
499 substrateLen = len(chunk)
500 while index < substrateLen:
501 subId = chunk[index]
502 index += 1
503 if subId < 128:
504 reloid.append(subId)
505 elif subId > 128:
506 # Construct subid from a number of octets
507 nextSubId = subId
508 subId = 0
509 continuationOctetCount = 0
510 while nextSubId >= 128:
511 continuationOctetCount += 1
512 if continuationOctetCount > MAX_OID_ARC_CONTINUATION_OCTETS:
513 raise error.PyAsn1Error(
514 'RELATIVE-OID arc exceeds maximum continuation octets limit (%d) '
515 'at position %d' % (MAX_OID_ARC_CONTINUATION_OCTETS, index)
516 )
517 subId = (subId << 7) + (nextSubId & 0x7F)
518 if index >= substrateLen:
519 raise error.SubstrateUnderrunError(
520 'Short substrate for sub-OID past %s' % (tuple(reloid),)
521 )
522 nextSubId = chunk[index]
523 index += 1
524 reloid.append((subId << 7) + nextSubId)
525 elif subId == 128:
526 # ASN.1 spec forbids leading zeros (0x80) in OID
527 # encoding, tolerating it opens a vulnerability. See
528 # https://www.esat.kuleuven.be/cosic/publications/article-1432.pdf
529 # page 7
530 raise error.PyAsn1Error('Invalid octet 0x80 in RELATIVE-OID encoding')
532 yield self._createComponent(asn1Spec, tagSet, tuple(reloid), **options)
535class RealPayloadDecoder(AbstractSimplePayloadDecoder):
536 protoComponent = univ.Real()
538 def valueDecoder(self, substrate, asn1Spec,
539 tagSet=None, length=None, state=None,
540 decodeFun=None, substrateFun=None,
541 **options):
542 if tagSet[0].tagFormat != tag.tagFormatSimple:
543 raise error.PyAsn1Error('Simple tag format expected')
545 for chunk in readFromStream(substrate, length, options):
546 if isinstance(chunk, SubstrateUnderrunError):
547 yield chunk
549 if not chunk:
550 yield self._createComponent(asn1Spec, tagSet, 0.0, **options)
551 return
553 fo = chunk[0]
554 chunk = chunk[1:]
555 if fo & 0x80: # binary encoding
556 if not chunk:
557 raise error.PyAsn1Error("Incomplete floating-point value")
559 if LOG:
560 LOG('decoding binary encoded REAL')
562 n = (fo & 0x03) + 1
564 if n == 4:
565 n = chunk[0]
566 chunk = chunk[1:]
568 eo, chunk = chunk[:n], chunk[n:]
570 if not eo or not chunk:
571 raise error.PyAsn1Error('Real exponent screwed')
573 e = eo[0] & 0x80 and -1 or 0
575 while eo: # exponent
576 e <<= 8
577 e |= eo[0]
578 eo = eo[1:]
580 b = fo >> 4 & 0x03 # base bits
582 if b > 2:
583 raise error.PyAsn1Error('Illegal Real base')
585 if b == 1: # encbase = 8
586 e *= 3
588 elif b == 2: # encbase = 16
589 e *= 4
590 p = 0
592 while chunk: # value
593 p <<= 8
594 p |= chunk[0]
595 chunk = chunk[1:]
597 if fo & 0x40: # sign bit
598 p = -p
600 sf = fo >> 2 & 0x03 # scale bits
601 p *= 2 ** sf
602 value = (p, 2, e)
604 elif fo & 0x40: # infinite value
605 if LOG:
606 LOG('decoding infinite REAL')
608 value = fo & 0x01 and '-inf' or 'inf'
610 elif fo & 0xc0 == 0: # character encoding
611 if not chunk:
612 raise error.PyAsn1Error("Incomplete floating-point value")
614 if LOG:
615 LOG('decoding character encoded REAL')
617 try:
618 if fo & 0x3 == 0x1: # NR1
619 value = (int(chunk), 10, 0)
621 elif fo & 0x3 == 0x2: # NR2
622 value = float(chunk)
624 elif fo & 0x3 == 0x3: # NR3
625 value = float(chunk)
627 else:
628 raise error.SubstrateUnderrunError(
629 'Unknown NR (tag %s)' % fo
630 )
632 except ValueError:
633 raise error.SubstrateUnderrunError(
634 'Bad character Real syntax'
635 )
637 else:
638 raise error.SubstrateUnderrunError(
639 'Unknown encoding (tag %s)' % fo
640 )
642 yield self._createComponent(asn1Spec, tagSet, value, **options)
645class AbstractConstructedPayloadDecoder(AbstractPayloadDecoder):
646 protoComponent = None
649class ConstructedPayloadDecoderBase(AbstractConstructedPayloadDecoder):
650 protoRecordComponent = None
651 protoSequenceComponent = None
653 def _getComponentTagMap(self, asn1Object, idx):
654 raise NotImplementedError
656 def _getComponentPositionByType(self, asn1Object, tagSet, idx):
657 raise NotImplementedError
659 def _decodeComponentsSchemaless(
660 self, substrate, tagSet=None, decodeFun=None,
661 length=None, **options):
663 asn1Object = None
665 components = []
666 componentTypes = set()
668 original_position = substrate.tell()
670 while length == -1 or substrate.tell() < original_position + length:
671 for component in decodeFun(substrate, **options):
672 if isinstance(component, SubstrateUnderrunError):
673 yield component
675 if length == -1 and component is eoo.endOfOctets:
676 break
678 components.append(component)
679 componentTypes.add(component.tagSet)
681 # Now we have to guess is it SEQUENCE/SET or SEQUENCE OF/SET OF
682 # The heuristics is:
683 # * 1+ components of different types -> likely SEQUENCE/SET
684 # * otherwise -> likely SEQUENCE OF/SET OF
685 if len(componentTypes) > 1:
686 protoComponent = self.protoRecordComponent
688 else:
689 protoComponent = self.protoSequenceComponent
691 asn1Object = protoComponent.clone(
692 # construct tagSet from base tag from prototype ASN.1 object
693 # and additional tags recovered from the substrate
694 tagSet=tag.TagSet(protoComponent.tagSet.baseTag, *tagSet.superTags)
695 )
697 if LOG:
698 LOG('guessed %r container type (pass `asn1Spec` to guide the '
699 'decoder)' % asn1Object)
701 for idx, component in enumerate(components):
702 asn1Object.setComponentByPosition(
703 idx, component,
704 verifyConstraints=False,
705 matchTags=False, matchConstraints=False
706 )
708 yield asn1Object
710 def valueDecoder(self, substrate, asn1Spec,
711 tagSet=None, length=None, state=None,
712 decodeFun=None, substrateFun=None,
713 **options):
714 if tagSet[0].tagFormat != tag.tagFormatConstructed:
715 raise error.PyAsn1Error('Constructed tag format expected')
717 original_position = substrate.tell()
719 if substrateFun:
720 if asn1Spec is not None:
721 asn1Object = asn1Spec.clone()
723 elif self.protoComponent is not None:
724 asn1Object = self.protoComponent.clone(tagSet=tagSet)
726 else:
727 asn1Object = self.protoRecordComponent, self.protoSequenceComponent
729 for chunk in substrateFun(asn1Object, substrate, length, options):
730 yield chunk
732 return
734 if asn1Spec is None:
735 for asn1Object in self._decodeComponentsSchemaless(
736 substrate, tagSet=tagSet, decodeFun=decodeFun,
737 length=length, **options):
738 if isinstance(asn1Object, SubstrateUnderrunError):
739 yield asn1Object
741 if substrate.tell() < original_position + length:
742 if LOG:
743 for trailing in readFromStream(substrate, context=options):
744 if isinstance(trailing, SubstrateUnderrunError):
745 yield trailing
747 LOG('Unused trailing %d octets encountered: %s' % (
748 len(trailing), debug.hexdump(trailing)))
750 yield asn1Object
752 return
754 asn1Object = asn1Spec.clone()
755 asn1Object.clear()
757 options = self._passAsn1Object(asn1Object, options)
759 if asn1Spec.typeId in (univ.Sequence.typeId, univ.Set.typeId):
761 namedTypes = asn1Spec.componentType
763 isSetType = asn1Spec.typeId == univ.Set.typeId
764 isDeterministic = not isSetType and not namedTypes.hasOptionalOrDefault
766 if LOG:
767 LOG('decoding %sdeterministic %s type %r chosen by type ID' % (
768 not isDeterministic and 'non-' or '', isSetType and 'SET' or '',
769 asn1Spec))
771 seenIndices = set()
772 idx = 0
773 while substrate.tell() - original_position < length:
774 if not namedTypes:
775 componentType = None
777 elif isSetType:
778 componentType = namedTypes.tagMapUnique
780 else:
781 try:
782 if isDeterministic:
783 componentType = namedTypes[idx].asn1Object
785 elif namedTypes[idx].isOptional or namedTypes[idx].isDefaulted:
786 componentType = namedTypes.getTagMapNearPosition(idx)
788 else:
789 componentType = namedTypes[idx].asn1Object
791 except IndexError:
792 raise error.PyAsn1Error(
793 'Excessive components decoded at %r' % (asn1Spec,)
794 )
796 for component in decodeFun(substrate, componentType, **options):
797 if isinstance(component, SubstrateUnderrunError):
798 yield component
800 if not isDeterministic and namedTypes:
801 if isSetType:
802 idx = namedTypes.getPositionByType(component.effectiveTagSet)
804 elif namedTypes[idx].isOptional or namedTypes[idx].isDefaulted:
805 idx = namedTypes.getPositionNearType(component.effectiveTagSet, idx)
807 asn1Object.setComponentByPosition(
808 idx, component,
809 verifyConstraints=False,
810 matchTags=False, matchConstraints=False
811 )
813 seenIndices.add(idx)
814 idx += 1
816 if LOG:
817 LOG('seen component indices %s' % seenIndices)
819 if namedTypes:
820 if not namedTypes.requiredComponents.issubset(seenIndices):
821 raise error.PyAsn1Error(
822 'ASN.1 object %s has uninitialized '
823 'components' % asn1Object.__class__.__name__)
825 if namedTypes.hasOpenTypes:
827 openTypes = options.get('openTypes', {})
829 if LOG:
830 LOG('user-specified open types map:')
832 for k, v in openTypes.items():
833 LOG('%s -> %r' % (k, v))
835 if openTypes or options.get('decodeOpenTypes', False):
837 for idx, namedType in enumerate(namedTypes.namedTypes):
838 if not namedType.openType:
839 continue
841 if namedType.isOptional and not asn1Object.getComponentByPosition(idx).isValue:
842 continue
844 governingValue = asn1Object.getComponentByName(
845 namedType.openType.name
846 )
848 try:
849 openType = openTypes[governingValue]
851 except KeyError:
853 if LOG:
854 LOG('default open types map of component '
855 '"%s.%s" governed by component "%s.%s"'
856 ':' % (asn1Object.__class__.__name__,
857 namedType.name,
858 asn1Object.__class__.__name__,
859 namedType.openType.name))
861 for k, v in namedType.openType.items():
862 LOG('%s -> %r' % (k, v))
864 try:
865 openType = namedType.openType[governingValue]
867 except KeyError:
868 if LOG:
869 LOG('failed to resolve open type by governing '
870 'value %r' % (governingValue,))
871 continue
873 if LOG:
874 LOG('resolved open type %r by governing '
875 'value %r' % (openType, governingValue))
877 containerValue = asn1Object.getComponentByPosition(idx)
879 if containerValue.typeId in (
880 univ.SetOf.typeId, univ.SequenceOf.typeId):
882 for pos, containerElement in enumerate(
883 containerValue):
885 stream = asSeekableStream(containerValue[pos].asOctets())
887 for component in decodeFun(stream, asn1Spec=openType, **options):
888 if isinstance(component, SubstrateUnderrunError):
889 yield component
891 containerValue[pos] = component
893 else:
894 stream = asSeekableStream(asn1Object.getComponentByPosition(idx).asOctets())
896 for component in decodeFun(stream, asn1Spec=openType, **options):
897 if isinstance(component, SubstrateUnderrunError):
898 yield component
900 asn1Object.setComponentByPosition(idx, component)
902 else:
903 inconsistency = asn1Object.isInconsistent
904 if inconsistency:
905 raise error.PyAsn1Error(
906 f"ASN.1 object {asn1Object.__class__.__name__} is inconsistent")
908 else:
909 componentType = asn1Spec.componentType
911 if LOG:
912 LOG('decoding type %r chosen by given `asn1Spec`' % componentType)
914 idx = 0
916 while substrate.tell() - original_position < length:
917 for component in decodeFun(substrate, componentType, **options):
918 if isinstance(component, SubstrateUnderrunError):
919 yield component
921 asn1Object.setComponentByPosition(
922 idx, component,
923 verifyConstraints=False,
924 matchTags=False, matchConstraints=False
925 )
927 idx += 1
929 yield asn1Object
931 def indefLenValueDecoder(self, substrate, asn1Spec,
932 tagSet=None, length=None, state=None,
933 decodeFun=None, substrateFun=None,
934 **options):
935 if tagSet[0].tagFormat != tag.tagFormatConstructed:
936 raise error.PyAsn1Error('Constructed tag format expected')
938 if substrateFun is not None:
939 if asn1Spec is not None:
940 asn1Object = asn1Spec.clone()
942 elif self.protoComponent is not None:
943 asn1Object = self.protoComponent.clone(tagSet=tagSet)
945 else:
946 asn1Object = self.protoRecordComponent, self.protoSequenceComponent
948 for chunk in substrateFun(asn1Object, substrate, length, options):
949 yield chunk
951 return
953 if asn1Spec is None:
954 for asn1Object in self._decodeComponentsSchemaless(
955 substrate, tagSet=tagSet, decodeFun=decodeFun,
956 length=length, **dict(options, allowEoo=True)):
957 if isinstance(asn1Object, SubstrateUnderrunError):
958 yield asn1Object
960 yield asn1Object
962 return
964 asn1Object = asn1Spec.clone()
965 asn1Object.clear()
967 options = self._passAsn1Object(asn1Object, options)
969 if asn1Spec.typeId in (univ.Sequence.typeId, univ.Set.typeId):
971 namedTypes = asn1Object.componentType
973 isSetType = asn1Object.typeId == univ.Set.typeId
974 isDeterministic = not isSetType and not namedTypes.hasOptionalOrDefault
976 if LOG:
977 LOG('decoding %sdeterministic %s type %r chosen by type ID' % (
978 not isDeterministic and 'non-' or '', isSetType and 'SET' or '',
979 asn1Spec))
981 seenIndices = set()
983 idx = 0
985 while True: # loop over components
986 if len(namedTypes) <= idx:
987 asn1Spec = None
989 elif isSetType:
990 asn1Spec = namedTypes.tagMapUnique
992 else:
993 try:
994 if isDeterministic:
995 asn1Spec = namedTypes[idx].asn1Object
997 elif namedTypes[idx].isOptional or namedTypes[idx].isDefaulted:
998 asn1Spec = namedTypes.getTagMapNearPosition(idx)
1000 else:
1001 asn1Spec = namedTypes[idx].asn1Object
1003 except IndexError:
1004 raise error.PyAsn1Error(
1005 'Excessive components decoded at %r' % (asn1Object,)
1006 )
1008 for component in decodeFun(substrate, asn1Spec, allowEoo=True, **options):
1010 if isinstance(component, SubstrateUnderrunError):
1011 yield component
1013 if component is eoo.endOfOctets:
1014 break
1016 if component is eoo.endOfOctets:
1017 break
1019 if not isDeterministic and namedTypes:
1020 if isSetType:
1021 idx = namedTypes.getPositionByType(component.effectiveTagSet)
1023 elif namedTypes[idx].isOptional or namedTypes[idx].isDefaulted:
1024 idx = namedTypes.getPositionNearType(component.effectiveTagSet, idx)
1026 asn1Object.setComponentByPosition(
1027 idx, component,
1028 verifyConstraints=False,
1029 matchTags=False, matchConstraints=False
1030 )
1032 seenIndices.add(idx)
1033 idx += 1
1035 if LOG:
1036 LOG('seen component indices %s' % seenIndices)
1038 if namedTypes:
1039 if not namedTypes.requiredComponents.issubset(seenIndices):
1040 raise error.PyAsn1Error(
1041 'ASN.1 object %s has uninitialized '
1042 'components' % asn1Object.__class__.__name__)
1044 if namedTypes.hasOpenTypes:
1046 openTypes = options.get('openTypes', {})
1048 if LOG:
1049 LOG('user-specified open types map:')
1051 for k, v in openTypes.items():
1052 LOG('%s -> %r' % (k, v))
1054 if openTypes or options.get('decodeOpenTypes', False):
1056 for idx, namedType in enumerate(namedTypes.namedTypes):
1057 if not namedType.openType:
1058 continue
1060 if namedType.isOptional and not asn1Object.getComponentByPosition(idx).isValue:
1061 continue
1063 governingValue = asn1Object.getComponentByName(
1064 namedType.openType.name
1065 )
1067 try:
1068 openType = openTypes[governingValue]
1070 except KeyError:
1072 if LOG:
1073 LOG('default open types map of component '
1074 '"%s.%s" governed by component "%s.%s"'
1075 ':' % (asn1Object.__class__.__name__,
1076 namedType.name,
1077 asn1Object.__class__.__name__,
1078 namedType.openType.name))
1080 for k, v in namedType.openType.items():
1081 LOG('%s -> %r' % (k, v))
1083 try:
1084 openType = namedType.openType[governingValue]
1086 except KeyError:
1087 if LOG:
1088 LOG('failed to resolve open type by governing '
1089 'value %r' % (governingValue,))
1090 continue
1092 if LOG:
1093 LOG('resolved open type %r by governing '
1094 'value %r' % (openType, governingValue))
1096 containerValue = asn1Object.getComponentByPosition(idx)
1098 if containerValue.typeId in (
1099 univ.SetOf.typeId, univ.SequenceOf.typeId):
1101 for pos, containerElement in enumerate(
1102 containerValue):
1104 stream = asSeekableStream(containerValue[pos].asOctets())
1106 for component in decodeFun(stream, asn1Spec=openType,
1107 **dict(options, allowEoo=True)):
1108 if isinstance(component, SubstrateUnderrunError):
1109 yield component
1111 if component is eoo.endOfOctets:
1112 break
1114 containerValue[pos] = component
1116 else:
1117 stream = asSeekableStream(asn1Object.getComponentByPosition(idx).asOctets())
1118 for component in decodeFun(stream, asn1Spec=openType,
1119 **dict(options, allowEoo=True)):
1120 if isinstance(component, SubstrateUnderrunError):
1121 yield component
1123 if component is eoo.endOfOctets:
1124 break
1126 asn1Object.setComponentByPosition(idx, component)
1128 else:
1129 inconsistency = asn1Object.isInconsistent
1130 if inconsistency:
1131 raise error.PyAsn1Error(
1132 f"ASN.1 object {asn1Object.__class__.__name__} is inconsistent")
1134 else:
1135 componentType = asn1Spec.componentType
1137 if LOG:
1138 LOG('decoding type %r chosen by given `asn1Spec`' % componentType)
1140 idx = 0
1142 while True:
1144 for component in decodeFun(
1145 substrate, componentType, allowEoo=True, **options):
1147 if isinstance(component, SubstrateUnderrunError):
1148 yield component
1150 if component is eoo.endOfOctets:
1151 break
1153 if component is eoo.endOfOctets:
1154 break
1156 asn1Object.setComponentByPosition(
1157 idx, component,
1158 verifyConstraints=False,
1159 matchTags=False, matchConstraints=False
1160 )
1162 idx += 1
1164 yield asn1Object
1167class SequenceOrSequenceOfPayloadDecoder(ConstructedPayloadDecoderBase):
1168 protoRecordComponent = univ.Sequence()
1169 protoSequenceComponent = univ.SequenceOf()
1172class SequencePayloadDecoder(SequenceOrSequenceOfPayloadDecoder):
1173 protoComponent = univ.Sequence()
1176class SequenceOfPayloadDecoder(SequenceOrSequenceOfPayloadDecoder):
1177 protoComponent = univ.SequenceOf()
1180class SetOrSetOfPayloadDecoder(ConstructedPayloadDecoderBase):
1181 protoRecordComponent = univ.Set()
1182 protoSequenceComponent = univ.SetOf()
1185class SetPayloadDecoder(SetOrSetOfPayloadDecoder):
1186 protoComponent = univ.Set()
1189class SetOfPayloadDecoder(SetOrSetOfPayloadDecoder):
1190 protoComponent = univ.SetOf()
1193class ChoicePayloadDecoder(ConstructedPayloadDecoderBase):
1194 protoComponent = univ.Choice()
1196 def valueDecoder(self, substrate, asn1Spec,
1197 tagSet=None, length=None, state=None,
1198 decodeFun=None, substrateFun=None,
1199 **options):
1200 if asn1Spec is None:
1201 asn1Object = self.protoComponent.clone(tagSet=tagSet)
1203 else:
1204 asn1Object = asn1Spec.clone()
1206 if substrateFun:
1207 for chunk in substrateFun(asn1Object, substrate, length, options):
1208 yield chunk
1210 return
1212 options = self._passAsn1Object(asn1Object, options)
1214 if asn1Object.tagSet == tagSet:
1215 if LOG:
1216 LOG('decoding %s as explicitly tagged CHOICE' % (tagSet,))
1218 for component in decodeFun(
1219 substrate, asn1Object.componentTagMap, **options):
1220 if isinstance(component, SubstrateUnderrunError):
1221 yield component
1223 else:
1224 if LOG:
1225 LOG('decoding %s as untagged CHOICE' % (tagSet,))
1227 for component in decodeFun(
1228 substrate, asn1Object.componentTagMap, tagSet, length,
1229 state, **options):
1230 if isinstance(component, SubstrateUnderrunError):
1231 yield component
1233 effectiveTagSet = component.effectiveTagSet
1235 if LOG:
1236 LOG('decoded component %s, effective tag set %s' % (component, effectiveTagSet))
1238 asn1Object.setComponentByType(
1239 effectiveTagSet, component,
1240 verifyConstraints=False,
1241 matchTags=False, matchConstraints=False,
1242 innerFlag=False
1243 )
1245 yield asn1Object
1247 def indefLenValueDecoder(self, substrate, asn1Spec,
1248 tagSet=None, length=None, state=None,
1249 decodeFun=None, substrateFun=None,
1250 **options):
1251 if asn1Spec is None:
1252 asn1Object = self.protoComponent.clone(tagSet=tagSet)
1254 else:
1255 asn1Object = asn1Spec.clone()
1257 if substrateFun:
1258 for chunk in substrateFun(asn1Object, substrate, length, options):
1259 yield chunk
1261 return
1263 options = self._passAsn1Object(asn1Object, options)
1265 isTagged = asn1Object.tagSet == tagSet
1267 if LOG:
1268 LOG('decoding %s as %stagged CHOICE' % (
1269 tagSet, isTagged and 'explicitly ' or 'un'))
1271 while True:
1273 if isTagged:
1274 iterator = decodeFun(
1275 substrate, asn1Object.componentType.tagMapUnique,
1276 **dict(options, allowEoo=True))
1278 else:
1279 iterator = decodeFun(
1280 substrate, asn1Object.componentType.tagMapUnique,
1281 tagSet, length, state, **dict(options, allowEoo=True))
1283 for component in iterator:
1285 if isinstance(component, SubstrateUnderrunError):
1286 yield component
1288 if component is eoo.endOfOctets:
1289 break
1291 effectiveTagSet = component.effectiveTagSet
1293 if LOG:
1294 LOG('decoded component %s, effective tag set '
1295 '%s' % (component, effectiveTagSet))
1297 asn1Object.setComponentByType(
1298 effectiveTagSet, component,
1299 verifyConstraints=False,
1300 matchTags=False, matchConstraints=False,
1301 innerFlag=False
1302 )
1304 if not isTagged:
1305 break
1307 if not isTagged or component is eoo.endOfOctets:
1308 break
1310 yield asn1Object
1313class AnyPayloadDecoder(AbstractSimplePayloadDecoder):
1314 protoComponent = univ.Any()
1316 def valueDecoder(self, substrate, asn1Spec,
1317 tagSet=None, length=None, state=None,
1318 decodeFun=None, substrateFun=None,
1319 **options):
1320 if asn1Spec is None:
1321 isUntagged = True
1323 elif asn1Spec.__class__ is tagmap.TagMap:
1324 isUntagged = tagSet not in asn1Spec.tagMap
1326 else:
1327 isUntagged = tagSet != asn1Spec.tagSet
1329 if isUntagged:
1330 fullPosition = substrate.markedPosition
1331 currentPosition = substrate.tell()
1333 substrate.seek(fullPosition, os.SEEK_SET)
1334 length += currentPosition - fullPosition
1336 if LOG:
1337 for chunk in peekIntoStream(substrate, length):
1338 if isinstance(chunk, SubstrateUnderrunError):
1339 yield chunk
1340 LOG('decoding as untagged ANY, substrate '
1341 '%s' % debug.hexdump(chunk))
1343 if substrateFun:
1344 for chunk in substrateFun(
1345 self._createComponent(asn1Spec, tagSet, noValue, **options),
1346 substrate, length, options):
1347 yield chunk
1349 return
1351 for chunk in readFromStream(substrate, length, options):
1352 if isinstance(chunk, SubstrateUnderrunError):
1353 yield chunk
1355 yield self._createComponent(asn1Spec, tagSet, chunk, **options)
1357 def indefLenValueDecoder(self, substrate, asn1Spec,
1358 tagSet=None, length=None, state=None,
1359 decodeFun=None, substrateFun=None,
1360 **options):
1361 if asn1Spec is None:
1362 isTagged = False
1364 elif asn1Spec.__class__ is tagmap.TagMap:
1365 isTagged = tagSet in asn1Spec.tagMap
1367 else:
1368 isTagged = tagSet == asn1Spec.tagSet
1370 if isTagged:
1371 # tagged Any type -- consume header substrate
1372 chunk = b''
1374 if LOG:
1375 LOG('decoding as tagged ANY')
1377 else:
1378 # TODO: Seems not to be tested
1379 fullPosition = substrate.markedPosition
1380 currentPosition = substrate.tell()
1382 substrate.seek(fullPosition, os.SEEK_SET)
1383 for chunk in readFromStream(substrate, currentPosition - fullPosition, options):
1384 if isinstance(chunk, SubstrateUnderrunError):
1385 yield chunk
1387 if LOG:
1388 LOG('decoding as untagged ANY, header substrate %s' % debug.hexdump(chunk))
1390 # Any components do not inherit initial tag
1391 asn1Spec = self.protoComponent
1393 if substrateFun and substrateFun is not self.substrateCollector:
1394 asn1Object = self._createComponent(
1395 asn1Spec, tagSet, noValue, **options)
1397 for chunk in substrateFun(
1398 asn1Object, chunk + substrate, length + len(chunk), options):
1399 yield chunk
1401 return
1403 if LOG:
1404 LOG('assembling constructed serialization')
1406 # All inner fragments are of the same type, treat them as octet string
1407 substrateFun = self.substrateCollector
1409 while True: # loop over fragments
1411 for component in decodeFun(
1412 substrate, asn1Spec, substrateFun=substrateFun,
1413 allowEoo=True, **options):
1415 if isinstance(component, SubstrateUnderrunError):
1416 yield component
1418 if component is eoo.endOfOctets:
1419 break
1421 if component is eoo.endOfOctets:
1422 break
1424 chunk += component
1426 if substrateFun:
1427 yield chunk # TODO: Weird
1429 else:
1430 yield self._createComponent(asn1Spec, tagSet, chunk, **options)
1433# character string types
1434class UTF8StringPayloadDecoder(OctetStringPayloadDecoder):
1435 protoComponent = char.UTF8String()
1438class NumericStringPayloadDecoder(OctetStringPayloadDecoder):
1439 protoComponent = char.NumericString()
1442class PrintableStringPayloadDecoder(OctetStringPayloadDecoder):
1443 protoComponent = char.PrintableString()
1446class TeletexStringPayloadDecoder(OctetStringPayloadDecoder):
1447 protoComponent = char.TeletexString()
1450class VideotexStringPayloadDecoder(OctetStringPayloadDecoder):
1451 protoComponent = char.VideotexString()
1454class IA5StringPayloadDecoder(OctetStringPayloadDecoder):
1455 protoComponent = char.IA5String()
1458class GraphicStringPayloadDecoder(OctetStringPayloadDecoder):
1459 protoComponent = char.GraphicString()
1462class VisibleStringPayloadDecoder(OctetStringPayloadDecoder):
1463 protoComponent = char.VisibleString()
1466class GeneralStringPayloadDecoder(OctetStringPayloadDecoder):
1467 protoComponent = char.GeneralString()
1470class UniversalStringPayloadDecoder(OctetStringPayloadDecoder):
1471 protoComponent = char.UniversalString()
1474class BMPStringPayloadDecoder(OctetStringPayloadDecoder):
1475 protoComponent = char.BMPString()
1478# "useful" types
1479class ObjectDescriptorPayloadDecoder(OctetStringPayloadDecoder):
1480 protoComponent = useful.ObjectDescriptor()
1483class GeneralizedTimePayloadDecoder(OctetStringPayloadDecoder):
1484 protoComponent = useful.GeneralizedTime()
1487class UTCTimePayloadDecoder(OctetStringPayloadDecoder):
1488 protoComponent = useful.UTCTime()
1491TAG_MAP = {
1492 univ.Integer.tagSet: IntegerPayloadDecoder(),
1493 univ.Boolean.tagSet: BooleanPayloadDecoder(),
1494 univ.BitString.tagSet: BitStringPayloadDecoder(),
1495 univ.OctetString.tagSet: OctetStringPayloadDecoder(),
1496 univ.Null.tagSet: NullPayloadDecoder(),
1497 univ.ObjectIdentifier.tagSet: ObjectIdentifierPayloadDecoder(),
1498 univ.RelativeOID.tagSet: RelativeOIDPayloadDecoder(),
1499 univ.Enumerated.tagSet: IntegerPayloadDecoder(),
1500 univ.Real.tagSet: RealPayloadDecoder(),
1501 univ.Sequence.tagSet: SequenceOrSequenceOfPayloadDecoder(), # conflicts with SequenceOf
1502 univ.Set.tagSet: SetOrSetOfPayloadDecoder(), # conflicts with SetOf
1503 univ.Choice.tagSet: ChoicePayloadDecoder(), # conflicts with Any
1504 # character string types
1505 char.UTF8String.tagSet: UTF8StringPayloadDecoder(),
1506 char.NumericString.tagSet: NumericStringPayloadDecoder(),
1507 char.PrintableString.tagSet: PrintableStringPayloadDecoder(),
1508 char.TeletexString.tagSet: TeletexStringPayloadDecoder(),
1509 char.VideotexString.tagSet: VideotexStringPayloadDecoder(),
1510 char.IA5String.tagSet: IA5StringPayloadDecoder(),
1511 char.GraphicString.tagSet: GraphicStringPayloadDecoder(),
1512 char.VisibleString.tagSet: VisibleStringPayloadDecoder(),
1513 char.GeneralString.tagSet: GeneralStringPayloadDecoder(),
1514 char.UniversalString.tagSet: UniversalStringPayloadDecoder(),
1515 char.BMPString.tagSet: BMPStringPayloadDecoder(),
1516 # useful types
1517 useful.ObjectDescriptor.tagSet: ObjectDescriptorPayloadDecoder(),
1518 useful.GeneralizedTime.tagSet: GeneralizedTimePayloadDecoder(),
1519 useful.UTCTime.tagSet: UTCTimePayloadDecoder()
1520}
1522# Type-to-codec map for ambiguous ASN.1 types
1523TYPE_MAP = {
1524 univ.Set.typeId: SetPayloadDecoder(),
1525 univ.SetOf.typeId: SetOfPayloadDecoder(),
1526 univ.Sequence.typeId: SequencePayloadDecoder(),
1527 univ.SequenceOf.typeId: SequenceOfPayloadDecoder(),
1528 univ.Choice.typeId: ChoicePayloadDecoder(),
1529 univ.Any.typeId: AnyPayloadDecoder()
1530}
1532# Put in non-ambiguous types for faster codec lookup
1533for typeDecoder in TAG_MAP.values():
1534 if typeDecoder.protoComponent is not None:
1535 typeId = typeDecoder.protoComponent.__class__.typeId
1536 if typeId is not None and typeId not in TYPE_MAP:
1537 TYPE_MAP[typeId] = typeDecoder
1540(stDecodeTag,
1541 stDecodeLength,
1542 stGetValueDecoder,
1543 stGetValueDecoderByAsn1Spec,
1544 stGetValueDecoderByTag,
1545 stTryAsExplicitTag,
1546 stDecodeValue,
1547 stDumpRawValue,
1548 stErrorCondition,
1549 stStop) = [x for x in range(10)]
1552EOO_SENTINEL = bytes((0, 0))
1555class SingleItemDecoder(object):
1556 defaultErrorState = stErrorCondition
1557 #defaultErrorState = stDumpRawValue
1558 defaultRawDecoder = AnyPayloadDecoder()
1560 supportIndefLength = True
1562 TAG_MAP = TAG_MAP
1563 TYPE_MAP = TYPE_MAP
1565 def __init__(self, tagMap=_MISSING, typeMap=_MISSING, **ignored):
1566 self._tagMap = tagMap if tagMap is not _MISSING else self.TAG_MAP
1567 self._typeMap = typeMap if typeMap is not _MISSING else self.TYPE_MAP
1569 # Tag & TagSet objects caches
1570 self._tagCache = {}
1571 self._tagSetCache = {}
1573 def __call__(self, substrate, asn1Spec=None,
1574 tagSet=None, length=None, state=stDecodeTag,
1575 decodeFun=None, substrateFun=None,
1576 **options):
1578 _nestingLevel = options.get('_nestingLevel', 0)
1580 if _nestingLevel > MAX_NESTING_DEPTH:
1581 raise error.PyAsn1Error(
1582 'ASN.1 structure nesting depth exceeds limit (%d)' % MAX_NESTING_DEPTH
1583 )
1585 options['_nestingLevel'] = _nestingLevel + 1
1587 allowEoo = options.pop('allowEoo', False)
1589 if LOG:
1590 LOG('decoder called at scope %s with state %d, working with up '
1591 'to %s octets of substrate: '
1592 '%s' % (debug.scope, state, length, substrate))
1594 # Look for end-of-octets sentinel
1595 if allowEoo and self.supportIndefLength:
1597 for eoo_candidate in readFromStream(substrate, 2, options):
1598 if isinstance(eoo_candidate, SubstrateUnderrunError):
1599 yield eoo_candidate
1601 if eoo_candidate == EOO_SENTINEL:
1602 if LOG:
1603 LOG('end-of-octets sentinel found')
1604 yield eoo.endOfOctets
1605 return
1607 else:
1608 substrate.seek(-2, os.SEEK_CUR)
1610 tagMap = self._tagMap
1611 typeMap = self._typeMap
1612 tagCache = self._tagCache
1613 tagSetCache = self._tagSetCache
1615 value = noValue
1617 substrate.markedPosition = substrate.tell()
1619 while state is not stStop:
1621 if state is stDecodeTag:
1622 # Decode tag
1623 isShortTag = True
1625 for firstByte in readFromStream(substrate, 1, options):
1626 if isinstance(firstByte, SubstrateUnderrunError):
1627 yield firstByte
1629 firstOctet = ord(firstByte)
1631 try:
1632 lastTag = tagCache[firstOctet]
1634 except KeyError:
1635 integerTag = firstOctet
1636 tagClass = integerTag & 0xC0
1637 tagFormat = integerTag & 0x20
1638 tagId = integerTag & 0x1F
1640 if tagId == 0x1F:
1641 isShortTag = False
1642 tagOctetCount = 0
1643 tagId = 0
1645 while True:
1646 for integerByte in readFromStream(substrate, 1, options):
1647 if isinstance(integerByte, SubstrateUnderrunError):
1648 yield integerByte
1650 if not integerByte:
1651 raise error.SubstrateUnderrunError(
1652 'Short octet stream on long tag decoding'
1653 )
1655 integerTag = ord(integerByte)
1656 tagOctetCount += 1
1657 if tagOctetCount > MAX_TAG_OCTETS:
1658 raise error.PyAsn1Error(
1659 'Tag ID octet count exceeds limit (%d)' % (
1660 MAX_TAG_OCTETS,)
1661 )
1662 tagId <<= 7
1663 tagId |= (integerTag & 0x7F)
1665 if not integerTag & 0x80:
1666 break
1668 lastTag = tag.Tag(
1669 tagClass=tagClass, tagFormat=tagFormat, tagId=tagId
1670 )
1672 if isShortTag:
1673 # cache short tags
1674 tagCache[firstOctet] = lastTag
1676 if tagSet is None:
1677 if isShortTag:
1678 try:
1679 tagSet = tagSetCache[firstOctet]
1681 except KeyError:
1682 # base tag not recovered
1683 tagSet = tag.TagSet((), lastTag)
1684 tagSetCache[firstOctet] = tagSet
1685 else:
1686 tagSet = tag.TagSet((), lastTag)
1688 else:
1689 tagSet = lastTag + tagSet
1691 state = stDecodeLength
1693 if LOG:
1694 LOG('tag decoded into %s, decoding length' % tagSet)
1696 if state is stDecodeLength:
1697 # Decode length
1698 for firstOctet in readFromStream(substrate, 1, options):
1699 if isinstance(firstOctet, SubstrateUnderrunError):
1700 yield firstOctet
1702 firstOctet = ord(firstOctet)
1704 if firstOctet < 128:
1705 length = firstOctet
1707 elif firstOctet > 128:
1708 size = firstOctet & 0x7F
1710 if size > MAX_LENGTH_OCTETS:
1711 raise error.PyAsn1Error(
1712 'BER length field size %d exceeds limit (%d)' % (
1713 size, MAX_LENGTH_OCTETS)
1714 )
1716 # encoded in size bytes
1717 for encodedLength in readFromStream(substrate, size, options):
1718 if isinstance(encodedLength, SubstrateUnderrunError):
1719 yield encodedLength
1720 encodedLength = list(encodedLength)
1721 if len(encodedLength) != size:
1722 raise error.SubstrateUnderrunError(
1723 '%s<%s at %s' % (size, len(encodedLength), tagSet)
1724 )
1726 length = 0
1727 for lengthOctet in encodedLength:
1728 length <<= 8
1729 length |= lengthOctet
1730 size += 1
1732 else: # 128 means indefinite
1733 length = -1
1735 if length == -1 and not self.supportIndefLength:
1736 raise error.PyAsn1Error('Indefinite length encoding not supported by this codec')
1738 state = stGetValueDecoder
1740 if LOG:
1741 LOG('value length decoded into %d' % length)
1743 if state is stGetValueDecoder:
1744 if asn1Spec is None:
1745 state = stGetValueDecoderByTag
1747 else:
1748 state = stGetValueDecoderByAsn1Spec
1749 #
1750 # There're two ways of creating subtypes in ASN.1 what influences
1751 # decoder operation. These methods are:
1752 # 1) Either base types used in or no IMPLICIT tagging has been
1753 # applied on subtyping.
1754 # 2) Subtype syntax drops base type information (by means of
1755 # IMPLICIT tagging.
1756 # The first case allows for complete tag recovery from substrate
1757 # while the second one requires original ASN.1 type spec for
1758 # decoding.
1759 #
1760 # In either case a set of tags (tagSet) is coming from substrate
1761 # in an incremental, tag-by-tag fashion (this is the case of
1762 # EXPLICIT tag which is most basic). Outermost tag comes first
1763 # from the wire.
1764 #
1765 if state is stGetValueDecoderByTag:
1766 try:
1767 concreteDecoder = tagMap[tagSet]
1769 except KeyError:
1770 concreteDecoder = None
1772 if concreteDecoder:
1773 state = stDecodeValue
1775 else:
1776 try:
1777 concreteDecoder = tagMap[tagSet[:1]]
1779 except KeyError:
1780 concreteDecoder = None
1782 if concreteDecoder:
1783 state = stDecodeValue
1784 else:
1785 state = stTryAsExplicitTag
1787 if LOG:
1788 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'))
1789 debug.scope.push(concreteDecoder is None and '?' or concreteDecoder.protoComponent.__class__.__name__)
1791 if state is stGetValueDecoderByAsn1Spec:
1793 if asn1Spec.__class__ is tagmap.TagMap:
1794 try:
1795 chosenSpec = asn1Spec[tagSet]
1797 except KeyError:
1798 chosenSpec = None
1800 if LOG:
1801 LOG('candidate ASN.1 spec is a map of:')
1803 for firstOctet, v in asn1Spec.presentTypes.items():
1804 LOG(' %s -> %s' % (firstOctet, v.__class__.__name__))
1806 if asn1Spec.skipTypes:
1807 LOG('but neither of: ')
1808 for firstOctet, v in asn1Spec.skipTypes.items():
1809 LOG(' %s -> %s' % (firstOctet, v.__class__.__name__))
1810 LOG('new candidate ASN.1 spec is %s, chosen by %s' % (chosenSpec is None and '<none>' or chosenSpec.prettyPrintType(), tagSet))
1812 elif tagSet == asn1Spec.tagSet or tagSet in asn1Spec.tagMap:
1813 chosenSpec = asn1Spec
1814 if LOG:
1815 LOG('candidate ASN.1 spec is %s' % asn1Spec.__class__.__name__)
1817 else:
1818 chosenSpec = None
1820 if chosenSpec is not None:
1821 try:
1822 # ambiguous type or just faster codec lookup
1823 concreteDecoder = typeMap[chosenSpec.typeId]
1825 if LOG:
1826 LOG('value decoder chosen for an ambiguous type by type ID %s' % (chosenSpec.typeId,))
1828 except KeyError:
1829 # use base type for codec lookup to recover untagged types
1830 baseTagSet = tag.TagSet(chosenSpec.tagSet.baseTag, chosenSpec.tagSet.baseTag)
1831 try:
1832 # base type or tagged subtype
1833 concreteDecoder = tagMap[baseTagSet]
1835 if LOG:
1836 LOG('value decoder chosen by base %s' % (baseTagSet,))
1838 except KeyError:
1839 concreteDecoder = None
1841 if concreteDecoder:
1842 asn1Spec = chosenSpec
1843 state = stDecodeValue
1845 else:
1846 state = stTryAsExplicitTag
1848 else:
1849 concreteDecoder = None
1850 state = stTryAsExplicitTag
1852 if LOG:
1853 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'))
1854 debug.scope.push(chosenSpec is None and '?' or chosenSpec.__class__.__name__)
1856 if state is stDecodeValue:
1857 if not options.get('recursiveFlag', True) and not substrateFun: # deprecate this
1858 def substrateFun(asn1Object, _substrate, _length, _options):
1859 """Legacy hack to keep the recursiveFlag=False option supported.
1861 The decode(..., substrateFun=userCallback) option was introduced in 0.1.4 as a generalization
1862 of the old recursiveFlag=False option. Users should pass their callback instead of using
1863 recursiveFlag.
1864 """
1865 yield asn1Object
1867 original_position = substrate.tell()
1869 if length == -1: # indef length
1870 for value in concreteDecoder.indefLenValueDecoder(
1871 substrate, asn1Spec,
1872 tagSet, length, stGetValueDecoder,
1873 self, substrateFun, **options):
1874 if isinstance(value, SubstrateUnderrunError):
1875 yield value
1877 else:
1878 for value in concreteDecoder.valueDecoder(
1879 substrate, asn1Spec,
1880 tagSet, length, stGetValueDecoder,
1881 self, substrateFun, **options):
1882 if isinstance(value, SubstrateUnderrunError):
1883 yield value
1885 bytesRead = substrate.tell() - original_position
1886 if not substrateFun and bytesRead != length:
1887 raise PyAsn1Error(
1888 "Read %s bytes instead of expected %s." % (bytesRead, length))
1889 elif substrateFun and bytesRead > length:
1890 # custom substrateFun may be used for partial decoding, reading less is expected there
1891 raise PyAsn1Error(
1892 "Read %s bytes are more than expected %s." % (bytesRead, length))
1894 if LOG:
1895 LOG('codec %s yields type %s, value:\n%s\n...' % (
1896 concreteDecoder.__class__.__name__, value.__class__.__name__,
1897 isinstance(value, base.Asn1Item) and value.prettyPrint() or value))
1899 state = stStop
1900 break
1902 if state is stTryAsExplicitTag:
1903 if (tagSet and
1904 tagSet[0].tagFormat == tag.tagFormatConstructed and
1905 tagSet[0].tagClass != tag.tagClassUniversal):
1906 # Assume explicit tagging
1907 concreteDecoder = rawPayloadDecoder
1908 state = stDecodeValue
1910 else:
1911 concreteDecoder = None
1912 state = self.defaultErrorState
1914 if LOG:
1915 LOG('codec %s chosen, decoding %s' % (concreteDecoder and concreteDecoder.__class__.__name__ or "<none>", state is stDecodeValue and 'value' or 'as failure'))
1917 if state is stDumpRawValue:
1918 concreteDecoder = self.defaultRawDecoder
1920 if LOG:
1921 LOG('codec %s chosen, decoding value' % concreteDecoder.__class__.__name__)
1923 state = stDecodeValue
1925 if state is stErrorCondition:
1926 raise error.PyAsn1Error(
1927 '%s not in asn1Spec: %r' % (tagSet, asn1Spec)
1928 )
1930 if LOG:
1931 debug.scope.pop()
1932 LOG('decoder left scope %s, call completed' % debug.scope)
1934 yield value
1937class StreamingDecoder(object):
1938 """Create an iterator that turns BER/CER/DER byte stream into ASN.1 objects.
1940 On each iteration, consume whatever BER/CER/DER serialization is
1941 available in the `substrate` stream-like object and turns it into
1942 one or more, possibly nested, ASN.1 objects.
1944 Parameters
1945 ----------
1946 substrate: :py:class:`file`, :py:class:`io.BytesIO`
1947 BER/CER/DER serialization in form of a byte stream
1949 Keyword Args
1950 ------------
1951 asn1Spec: :py:class:`~pyasn1.type.base.PyAsn1Item`
1952 A pyasn1 type object to act as a template guiding the decoder.
1953 Depending on the ASN.1 structure being decoded, `asn1Spec` may
1954 or may not be required. One of the reasons why `asn1Spec` may
1955 me required is that ASN.1 structure is encoded in the *IMPLICIT*
1956 tagging mode.
1958 Yields
1959 ------
1960 : :py:class:`~pyasn1.type.base.PyAsn1Item`, :py:class:`~pyasn1.error.SubstrateUnderrunError`
1961 Decoded ASN.1 object (possibly, nested) or
1962 :py:class:`~pyasn1.error.SubstrateUnderrunError` object indicating
1963 insufficient BER/CER/DER serialization on input to fully recover ASN.1
1964 objects from it.
1966 In the latter case the caller is advised to ensure some more data in
1967 the input stream, then call the iterator again. The decoder will resume
1968 the decoding process using the newly arrived data.
1970 The `context` property of :py:class:`~pyasn1.error.SubstrateUnderrunError`
1971 object might hold a reference to the partially populated ASN.1 object
1972 being reconstructed.
1974 Raises
1975 ------
1976 ~pyasn1.error.PyAsn1Error, ~pyasn1.error.EndOfStreamError
1977 `PyAsn1Error` on deserialization error, `EndOfStreamError` on
1978 premature stream closure.
1980 Examples
1981 --------
1982 Decode BER serialisation without ASN.1 schema
1984 .. code-block:: pycon
1986 >>> stream = io.BytesIO(
1987 ... b'0\t\x02\x01\x01\x02\x01\x02\x02\x01\x03')
1988 >>>
1989 >>> for asn1Object in StreamingDecoder(stream):
1990 ... print(asn1Object)
1991 >>>
1992 SequenceOf:
1993 1 2 3
1995 Decode BER serialisation with ASN.1 schema
1997 .. code-block:: pycon
1999 >>> stream = io.BytesIO(
2000 ... b'0\t\x02\x01\x01\x02\x01\x02\x02\x01\x03')
2001 >>>
2002 >>> schema = SequenceOf(componentType=Integer())
2003 >>>
2004 >>> decoder = StreamingDecoder(stream, asn1Spec=schema)
2005 >>> for asn1Object in decoder:
2006 ... print(asn1Object)
2007 >>>
2008 SequenceOf:
2009 1 2 3
2010 """
2012 SINGLE_ITEM_DECODER = SingleItemDecoder
2014 def __init__(self, substrate, asn1Spec=None, **options):
2015 self._singleItemDecoder = self.SINGLE_ITEM_DECODER(**options)
2016 self._substrate = asSeekableStream(substrate)
2017 self._asn1Spec = asn1Spec
2018 self._options = options
2020 def __iter__(self):
2021 while True:
2022 for asn1Object in self._singleItemDecoder(
2023 self._substrate, self._asn1Spec, **self._options):
2024 yield asn1Object
2026 for chunk in isEndOfStream(self._substrate):
2027 if isinstance(chunk, SubstrateUnderrunError):
2028 yield
2030 break
2032 if chunk:
2033 break
2036class Decoder(object):
2037 """Create a BER decoder object.
2039 Parse BER/CER/DER octet-stream into one, possibly nested, ASN.1 object.
2040 """
2041 STREAMING_DECODER = StreamingDecoder
2043 @classmethod
2044 def __call__(cls, substrate, asn1Spec=None, **options):
2045 """Turns BER/CER/DER octet stream into an ASN.1 object.
2047 Takes BER/CER/DER octet-stream in form of :py:class:`bytes`
2048 and decode it into an ASN.1 object
2049 (e.g. :py:class:`~pyasn1.type.base.PyAsn1Item` derivative) which
2050 may be a scalar or an arbitrary nested structure.
2052 Parameters
2053 ----------
2054 substrate: :py:class:`bytes`
2055 BER/CER/DER octet-stream to parse
2057 Keyword Args
2058 ------------
2059 asn1Spec: :py:class:`~pyasn1.type.base.PyAsn1Item`
2060 A pyasn1 type object (:py:class:`~pyasn1.type.base.PyAsn1Item`
2061 derivative) to act as a template guiding the decoder.
2062 Depending on the ASN.1 structure being decoded, `asn1Spec` may or
2063 may not be required. Most common reason for it to require is that
2064 ASN.1 structure is encoded in *IMPLICIT* tagging mode.
2066 substrateFun: :py:class:`Union[
2067 Callable[[pyasn1.type.base.PyAsn1Item, bytes, int],
2068 Tuple[pyasn1.type.base.PyAsn1Item, bytes]],
2069 Callable[[pyasn1.type.base.PyAsn1Item, io.BytesIO, int, dict],
2070 Generator[Union[pyasn1.type.base.PyAsn1Item,
2071 pyasn1.error.SubstrateUnderrunError],
2072 None, None]]
2073 ]`
2074 User callback meant to generalize special use cases like non-recursive or
2075 partial decoding. A 3-arg non-streaming variant is supported for backwards
2076 compatiblilty in addition to the newer 4-arg streaming variant.
2077 The callback will receive the uninitialized object recovered from substrate
2078 as 1st argument, the uninterpreted payload as 2nd argument, and the length
2079 of the uninterpreted payload as 3rd argument. The streaming variant will
2080 additionally receive the decode(..., **options) kwargs as 4th argument.
2081 The non-streaming variant shall return an object that will be propagated
2082 as decode() return value as 1st item, and the remainig payload for further
2083 decode passes as 2nd item.
2084 The streaming variant shall yield an object that will be propagated as
2085 decode() return value, and leave the remaining payload in the stream.
2087 Returns
2088 -------
2089 : :py:class:`tuple`
2090 A tuple of :py:class:`~pyasn1.type.base.PyAsn1Item` object
2091 recovered from BER/CER/DER substrate and the unprocessed trailing
2092 portion of the `substrate` (may be empty)
2094 Raises
2095 ------
2096 : :py:class:`~pyasn1.error.PyAsn1Error`
2097 :py:class:`~pyasn1.error.SubstrateUnderrunError` on insufficient
2098 input or :py:class:`~pyasn1.error.PyAsn1Error` on decoding error.
2100 Examples
2101 --------
2102 Decode BER/CER/DER serialisation without ASN.1 schema
2104 .. code-block:: pycon
2106 >>> s, unprocessed = decode(b'0\t\x02\x01\x01\x02\x01\x02\x02\x01\x03')
2107 >>> str(s)
2108 SequenceOf:
2109 1 2 3
2111 Decode BER/CER/DER serialisation with ASN.1 schema
2113 .. code-block:: pycon
2115 >>> seq = SequenceOf(componentType=Integer())
2116 >>> s, unprocessed = decode(
2117 b'0\t\x02\x01\x01\x02\x01\x02\x02\x01\x03', asn1Spec=seq)
2118 >>> str(s)
2119 SequenceOf:
2120 1 2 3
2122 """
2123 substrate = asSeekableStream(substrate)
2125 if "substrateFun" in options:
2126 origSubstrateFun = options["substrateFun"]
2128 def substrateFunWrapper(asn1Object, substrate, length, options=None):
2129 """Support both 0.4 and 0.5 style APIs.
2131 substrateFun API has changed in 0.5 for use with streaming decoders. To stay backwards compatible,
2132 we first try if we received a streaming user callback. If that fails,we assume we've received a
2133 non-streaming v0.4 user callback and convert it for streaming on the fly
2134 """
2135 try:
2136 substrate_gen = origSubstrateFun(asn1Object, substrate, length, options)
2137 except TypeError as _value:
2138 if _value.__traceback__.tb_next:
2139 # Traceback depth > 1 means TypeError from inside user provided function
2140 raise
2141 # invariant maintained at Decoder.__call__ entry
2142 assert isinstance(substrate, io.BytesIO) # nosec assert_used
2143 substrate_gen = Decoder._callSubstrateFunV4asV5(origSubstrateFun, asn1Object, substrate, length)
2144 for value in substrate_gen:
2145 yield value
2147 options["substrateFun"] = substrateFunWrapper
2149 streamingDecoder = cls.STREAMING_DECODER(
2150 substrate, asn1Spec, **options)
2152 for asn1Object in streamingDecoder:
2153 if isinstance(asn1Object, SubstrateUnderrunError):
2154 raise error.SubstrateUnderrunError('Short substrate on input')
2156 try:
2157 tail = next(readFromStream(substrate))
2159 except error.EndOfStreamError:
2160 tail = b''
2162 return asn1Object, tail
2164 @staticmethod
2165 def _callSubstrateFunV4asV5(substrateFunV4, asn1Object, substrate, length):
2166 substrate_bytes = substrate.read()
2167 if length == -1:
2168 length = len(substrate_bytes)
2169 value, nextSubstrate = substrateFunV4(asn1Object, substrate_bytes, length)
2170 nbytes = substrate.write(nextSubstrate)
2171 substrate.truncate()
2172 substrate.seek(-nbytes, os.SEEK_CUR)
2173 yield value
2175#: Turns BER octet stream into an ASN.1 object.
2176#:
2177#: Takes BER octet-stream and decode it into an ASN.1 object
2178#: (e.g. :py:class:`~pyasn1.type.base.PyAsn1Item` derivative) which
2179#: may be a scalar or an arbitrary nested structure.
2180#:
2181#: Parameters
2182#: ----------
2183#: substrate: :py:class:`bytes`
2184#: BER octet-stream
2185#:
2186#: Keyword Args
2187#: ------------
2188#: asn1Spec: any pyasn1 type object e.g. :py:class:`~pyasn1.type.base.PyAsn1Item` derivative
2189#: A pyasn1 type object to act as a template guiding the decoder. Depending on the ASN.1 structure
2190#: being decoded, *asn1Spec* may or may not be required. Most common reason for
2191#: it to require is that ASN.1 structure is encoded in *IMPLICIT* tagging mode.
2192#:
2193#: Returns
2194#: -------
2195#: : :py:class:`tuple`
2196#: A tuple of pyasn1 object recovered from BER substrate (:py:class:`~pyasn1.type.base.PyAsn1Item` derivative)
2197#: and the unprocessed trailing portion of the *substrate* (may be empty)
2198#:
2199#: Raises
2200#: ------
2201#: ~pyasn1.error.PyAsn1Error, ~pyasn1.error.SubstrateUnderrunError
2202#: On decoding errors
2203#:
2204#: Notes
2205#: -----
2206#: This function is deprecated. Please use :py:class:`Decoder` or
2207#: :py:class:`StreamingDecoder` class instance.
2208#:
2209#: Examples
2210#: --------
2211#: Decode BER serialisation without ASN.1 schema
2212#:
2213#: .. code-block:: pycon
2214#:
2215#: >>> s, _ = decode(b'0\t\x02\x01\x01\x02\x01\x02\x02\x01\x03')
2216#: >>> str(s)
2217#: SequenceOf:
2218#: 1 2 3
2219#:
2220#: Decode BER serialisation with ASN.1 schema
2221#:
2222#: .. code-block:: pycon
2223#:
2224#: >>> seq = SequenceOf(componentType=Integer())
2225#: >>> s, _ = decode(b'0\t\x02\x01\x01\x02\x01\x02\x02\x01\x03', asn1Spec=seq)
2226#: >>> str(s)
2227#: SequenceOf:
2228#: 1 2 3
2229#:
2230decode = Decoder()
2232def __getattr__(attr: str):
2233 if newAttr := {"tagMap": "TAG_MAP", "typeMap": "TYPE_MAP"}.get(attr):
2234 warnings.warn(f"{attr} is deprecated. Please use {newAttr} instead.", DeprecationWarning, stacklevel=2)
2235 return globals()[newAttr]
2236 raise AttributeError(attr)