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: http://snmplabs.com/pyasn1/license.html
6#
7import os
9from pyasn1 import debug
10from pyasn1 import error
11from pyasn1.codec.ber import eoo
12from pyasn1.codec.streaming import asSeekableStream
13from pyasn1.codec.streaming import isEndOfStream
14from pyasn1.codec.streaming import peekIntoStream
15from pyasn1.codec.streaming import readFromStream
16from pyasn1.compat.integer import from_bytes
17from pyasn1.compat.octets import oct2int, octs2ints, ints2octs, null
18from pyasn1.error import PyAsn1Error
19from pyasn1.type import base
20from pyasn1.type import char
21from pyasn1.type import tag
22from pyasn1.type import tagmap
23from pyasn1.type import univ
24from pyasn1.type import useful
26__all__ = ['StreamingDecoder', 'Decoder', 'decode']
28LOG = debug.registerLoggee(__name__, flags=debug.DEBUG_DECODER)
30noValue = base.noValue
32SubstrateUnderrunError = error.SubstrateUnderrunError
35class AbstractPayloadDecoder(object):
36 protoComponent = None
38 def valueDecoder(self, substrate, asn1Spec,
39 tagSet=None, length=None, state=None,
40 decodeFun=None, substrateFun=None,
41 **options):
42 """Decode value with fixed byte length.
44 The decoder is allowed to consume as many bytes as necessary.
45 """
46 raise error.PyAsn1Error('SingleItemDecoder not implemented for %s' % (tagSet,)) # TODO: Seems more like an NotImplementedError?
48 def indefLenValueDecoder(self, substrate, asn1Spec,
49 tagSet=None, length=None, state=None,
50 decodeFun=None, substrateFun=None,
51 **options):
52 """Decode value with undefined length.
54 The decoder is allowed to consume as many bytes as necessary.
55 """
56 raise error.PyAsn1Error('Indefinite length mode decoder not implemented for %s' % (tagSet,)) # TODO: Seems more like an NotImplementedError?
58 @staticmethod
59 def _passAsn1Object(asn1Object, options):
60 if 'asn1Object' not in options:
61 options['asn1Object'] = asn1Object
63 return options
66class AbstractSimplePayloadDecoder(AbstractPayloadDecoder):
67 @staticmethod
68 def substrateCollector(asn1Object, substrate, length, options):
69 for chunk in readFromStream(substrate, length, options):
70 yield chunk
72 def _createComponent(self, asn1Spec, tagSet, value, **options):
73 if options.get('native'):
74 return value
75 elif asn1Spec is None:
76 return self.protoComponent.clone(value, tagSet=tagSet)
77 elif value is noValue:
78 return asn1Spec
79 else:
80 return asn1Spec.clone(value)
83class RawPayloadDecoder(AbstractSimplePayloadDecoder):
84 protoComponent = univ.Any('')
86 def valueDecoder(self, substrate, asn1Spec,
87 tagSet=None, length=None, state=None,
88 decodeFun=None, substrateFun=None,
89 **options):
90 if substrateFun:
91 asn1Object = self._createComponent(asn1Spec, tagSet, '', **options)
93 for chunk in substrateFun(asn1Object, substrate, length, options):
94 yield chunk
96 return
98 for value in decodeFun(substrate, asn1Spec, tagSet, length, **options):
99 yield value
101 def indefLenValueDecoder(self, substrate, asn1Spec,
102 tagSet=None, length=None, state=None,
103 decodeFun=None, substrateFun=None,
104 **options):
105 if substrateFun:
106 asn1Object = self._createComponent(asn1Spec, tagSet, '', **options)
108 for chunk in substrateFun(asn1Object, substrate, length, options):
109 yield chunk
111 return
113 while True:
114 for value in decodeFun(
115 substrate, asn1Spec, tagSet, length,
116 allowEoo=True, **options):
118 if value is eoo.endOfOctets:
119 return
121 yield value
124rawPayloadDecoder = RawPayloadDecoder()
127class IntegerPayloadDecoder(AbstractSimplePayloadDecoder):
128 protoComponent = univ.Integer(0)
130 def valueDecoder(self, substrate, asn1Spec,
131 tagSet=None, length=None, state=None,
132 decodeFun=None, substrateFun=None,
133 **options):
135 if tagSet[0].tagFormat != tag.tagFormatSimple:
136 raise error.PyAsn1Error('Simple tag format expected')
138 for chunk in readFromStream(substrate, length, options):
139 if isinstance(chunk, SubstrateUnderrunError):
140 yield chunk
142 if chunk:
143 value = from_bytes(chunk, signed=True)
145 else:
146 value = 0
148 yield self._createComponent(asn1Spec, tagSet, value, **options)
151class BooleanPayloadDecoder(IntegerPayloadDecoder):
152 protoComponent = univ.Boolean(0)
154 def _createComponent(self, asn1Spec, tagSet, value, **options):
155 return IntegerPayloadDecoder._createComponent(
156 self, asn1Spec, tagSet, value and 1 or 0, **options)
159class BitStringPayloadDecoder(AbstractSimplePayloadDecoder):
160 protoComponent = univ.BitString(())
161 supportConstructedForm = True
163 def valueDecoder(self, substrate, asn1Spec,
164 tagSet=None, length=None, state=None,
165 decodeFun=None, substrateFun=None,
166 **options):
168 if substrateFun:
169 asn1Object = self._createComponent(asn1Spec, tagSet, noValue, **options)
171 for chunk in substrateFun(asn1Object, substrate, length, options):
172 yield chunk
174 return
176 if not length:
177 raise error.PyAsn1Error('Empty BIT STRING substrate')
179 for chunk in isEndOfStream(substrate):
180 if isinstance(chunk, SubstrateUnderrunError):
181 yield chunk
183 if chunk:
184 raise error.PyAsn1Error('Empty BIT STRING substrate')
186 if tagSet[0].tagFormat == tag.tagFormatSimple: # XXX what tag to check?
188 for trailingBits in readFromStream(substrate, 1, options):
189 if isinstance(trailingBits, SubstrateUnderrunError):
190 yield trailingBits
192 trailingBits = ord(trailingBits)
193 if trailingBits > 7:
194 raise error.PyAsn1Error(
195 'Trailing bits overflow %s' % trailingBits
196 )
198 for chunk in readFromStream(substrate, length - 1, options):
199 if isinstance(chunk, SubstrateUnderrunError):
200 yield chunk
202 value = self.protoComponent.fromOctetString(
203 chunk, internalFormat=True, padding=trailingBits)
205 yield self._createComponent(asn1Spec, tagSet, value, **options)
207 return
209 if not self.supportConstructedForm:
210 raise error.PyAsn1Error('Constructed encoding form prohibited '
211 'at %s' % self.__class__.__name__)
213 if LOG:
214 LOG('assembling constructed serialization')
216 # All inner fragments are of the same type, treat them as octet string
217 substrateFun = self.substrateCollector
219 bitString = self.protoComponent.fromOctetString(null, internalFormat=True)
221 current_position = substrate.tell()
223 while substrate.tell() - current_position < length:
224 for component in decodeFun(
225 substrate, self.protoComponent, substrateFun=substrateFun,
226 **options):
227 if isinstance(component, SubstrateUnderrunError):
228 yield component
230 trailingBits = oct2int(component[0])
231 if trailingBits > 7:
232 raise error.PyAsn1Error(
233 'Trailing bits overflow %s' % trailingBits
234 )
236 bitString = self.protoComponent.fromOctetString(
237 component[1:], internalFormat=True,
238 prepend=bitString, padding=trailingBits
239 )
241 yield self._createComponent(asn1Spec, tagSet, bitString, **options)
243 def indefLenValueDecoder(self, substrate, asn1Spec,
244 tagSet=None, length=None, state=None,
245 decodeFun=None, substrateFun=None,
246 **options):
248 if substrateFun:
249 asn1Object = self._createComponent(asn1Spec, tagSet, noValue, **options)
251 for chunk in substrateFun(asn1Object, substrate, length, options):
252 yield chunk
254 return
256 # All inner fragments are of the same type, treat them as octet string
257 substrateFun = self.substrateCollector
259 bitString = self.protoComponent.fromOctetString(null, internalFormat=True)
261 while True: # loop over fragments
263 for component in decodeFun(
264 substrate, self.protoComponent, substrateFun=substrateFun,
265 allowEoo=True, **options):
267 if component is eoo.endOfOctets:
268 break
270 if isinstance(component, SubstrateUnderrunError):
271 yield component
273 if component is eoo.endOfOctets:
274 break
276 trailingBits = oct2int(component[0])
277 if trailingBits > 7:
278 raise error.PyAsn1Error(
279 'Trailing bits overflow %s' % trailingBits
280 )
282 bitString = self.protoComponent.fromOctetString(
283 component[1:], internalFormat=True,
284 prepend=bitString, padding=trailingBits
285 )
287 yield self._createComponent(asn1Spec, tagSet, bitString, **options)
290class OctetStringPayloadDecoder(AbstractSimplePayloadDecoder):
291 protoComponent = univ.OctetString('')
292 supportConstructedForm = True
294 def valueDecoder(self, substrate, asn1Spec,
295 tagSet=None, length=None, state=None,
296 decodeFun=None, substrateFun=None,
297 **options):
298 if substrateFun:
299 asn1Object = self._createComponent(asn1Spec, tagSet, noValue, **options)
301 for chunk in substrateFun(asn1Object, substrate, length, options):
302 yield chunk
304 return
306 if tagSet[0].tagFormat == tag.tagFormatSimple: # XXX what tag to check?
307 for chunk in readFromStream(substrate, length, options):
308 if isinstance(chunk, SubstrateUnderrunError):
309 yield chunk
311 yield self._createComponent(asn1Spec, tagSet, chunk, **options)
313 return
315 if not self.supportConstructedForm:
316 raise error.PyAsn1Error('Constructed encoding form prohibited at %s' % self.__class__.__name__)
318 if LOG:
319 LOG('assembling constructed serialization')
321 # All inner fragments are of the same type, treat them as octet string
322 substrateFun = self.substrateCollector
324 header = null
326 original_position = substrate.tell()
327 # head = popSubstream(substrate, length)
328 while substrate.tell() - original_position < length:
329 for component in decodeFun(
330 substrate, self.protoComponent, substrateFun=substrateFun,
331 **options):
332 if isinstance(component, SubstrateUnderrunError):
333 yield component
335 header += component
337 yield self._createComponent(asn1Spec, tagSet, header, **options)
339 def indefLenValueDecoder(self, substrate, asn1Spec,
340 tagSet=None, length=None, state=None,
341 decodeFun=None, substrateFun=None,
342 **options):
343 if substrateFun and substrateFun is not self.substrateCollector:
344 asn1Object = self._createComponent(asn1Spec, tagSet, noValue, **options)
346 for chunk in substrateFun(asn1Object, substrate, length, options):
347 yield chunk
349 return
351 # All inner fragments are of the same type, treat them as octet string
352 substrateFun = self.substrateCollector
354 header = null
356 while True: # loop over fragments
358 for component in decodeFun(
359 substrate, self.protoComponent, substrateFun=substrateFun,
360 allowEoo=True, **options):
362 if isinstance(component, SubstrateUnderrunError):
363 yield component
365 if component is eoo.endOfOctets:
366 break
368 if component is eoo.endOfOctets:
369 break
371 header += component
373 yield self._createComponent(asn1Spec, tagSet, header, **options)
376class NullPayloadDecoder(AbstractSimplePayloadDecoder):
377 protoComponent = univ.Null('')
379 def valueDecoder(self, substrate, asn1Spec,
380 tagSet=None, length=None, state=None,
381 decodeFun=None, substrateFun=None,
382 **options):
384 if tagSet[0].tagFormat != tag.tagFormatSimple:
385 raise error.PyAsn1Error('Simple tag format expected')
387 for chunk in readFromStream(substrate, length, options):
388 if isinstance(chunk, SubstrateUnderrunError):
389 yield chunk
391 component = self._createComponent(asn1Spec, tagSet, '', **options)
393 if chunk:
394 raise error.PyAsn1Error('Unexpected %d-octet substrate for Null' % length)
396 yield component
399class ObjectIdentifierPayloadDecoder(AbstractSimplePayloadDecoder):
400 protoComponent = univ.ObjectIdentifier(())
402 def valueDecoder(self, substrate, asn1Spec,
403 tagSet=None, length=None, state=None,
404 decodeFun=None, substrateFun=None,
405 **options):
406 if tagSet[0].tagFormat != tag.tagFormatSimple:
407 raise error.PyAsn1Error('Simple tag format expected')
409 for chunk in readFromStream(substrate, length, options):
410 if isinstance(chunk, SubstrateUnderrunError):
411 yield chunk
413 if not chunk:
414 raise error.PyAsn1Error('Empty substrate')
416 chunk = octs2ints(chunk)
418 oid = ()
419 index = 0
420 substrateLen = len(chunk)
421 while index < substrateLen:
422 subId = chunk[index]
423 index += 1
424 if subId < 128:
425 oid += (subId,)
426 elif subId > 128:
427 # Construct subid from a number of octets
428 nextSubId = subId
429 subId = 0
430 while nextSubId >= 128:
431 subId = (subId << 7) + (nextSubId & 0x7F)
432 if index >= substrateLen:
433 raise error.SubstrateUnderrunError(
434 'Short substrate for sub-OID past %s' % (oid,)
435 )
436 nextSubId = chunk[index]
437 index += 1
438 oid += ((subId << 7) + nextSubId,)
439 elif subId == 128:
440 # ASN.1 spec forbids leading zeros (0x80) in OID
441 # encoding, tolerating it opens a vulnerability. See
442 # https://www.esat.kuleuven.be/cosic/publications/article-1432.pdf
443 # page 7
444 raise error.PyAsn1Error('Invalid octet 0x80 in OID encoding')
446 # Decode two leading arcs
447 if 0 <= oid[0] <= 39:
448 oid = (0,) + oid
449 elif 40 <= oid[0] <= 79:
450 oid = (1, oid[0] - 40) + oid[1:]
451 elif oid[0] >= 80:
452 oid = (2, oid[0] - 80) + oid[1:]
453 else:
454 raise error.PyAsn1Error('Malformed first OID octet: %s' % chunk[0])
456 yield self._createComponent(asn1Spec, tagSet, oid, **options)
459class RealPayloadDecoder(AbstractSimplePayloadDecoder):
460 protoComponent = univ.Real()
462 def valueDecoder(self, substrate, asn1Spec,
463 tagSet=None, length=None, state=None,
464 decodeFun=None, substrateFun=None,
465 **options):
466 if tagSet[0].tagFormat != tag.tagFormatSimple:
467 raise error.PyAsn1Error('Simple tag format expected')
469 for chunk in readFromStream(substrate, length, options):
470 if isinstance(chunk, SubstrateUnderrunError):
471 yield chunk
473 if not chunk:
474 yield self._createComponent(asn1Spec, tagSet, 0.0, **options)
475 return
477 fo = oct2int(chunk[0])
478 chunk = chunk[1:]
479 if fo & 0x80: # binary encoding
480 if not chunk:
481 raise error.PyAsn1Error("Incomplete floating-point value")
483 if LOG:
484 LOG('decoding binary encoded REAL')
486 n = (fo & 0x03) + 1
488 if n == 4:
489 n = oct2int(chunk[0])
490 chunk = chunk[1:]
492 eo, chunk = chunk[:n], chunk[n:]
494 if not eo or not chunk:
495 raise error.PyAsn1Error('Real exponent screwed')
497 e = oct2int(eo[0]) & 0x80 and -1 or 0
499 while eo: # exponent
500 e <<= 8
501 e |= oct2int(eo[0])
502 eo = eo[1:]
504 b = fo >> 4 & 0x03 # base bits
506 if b > 2:
507 raise error.PyAsn1Error('Illegal Real base')
509 if b == 1: # encbase = 8
510 e *= 3
512 elif b == 2: # encbase = 16
513 e *= 4
514 p = 0
516 while chunk: # value
517 p <<= 8
518 p |= oct2int(chunk[0])
519 chunk = chunk[1:]
521 if fo & 0x40: # sign bit
522 p = -p
524 sf = fo >> 2 & 0x03 # scale bits
525 p *= 2 ** sf
526 value = (p, 2, e)
528 elif fo & 0x40: # infinite value
529 if LOG:
530 LOG('decoding infinite REAL')
532 value = fo & 0x01 and '-inf' or 'inf'
534 elif fo & 0xc0 == 0: # character encoding
535 if not chunk:
536 raise error.PyAsn1Error("Incomplete floating-point value")
538 if LOG:
539 LOG('decoding character encoded REAL')
541 try:
542 if fo & 0x3 == 0x1: # NR1
543 value = (int(chunk), 10, 0)
545 elif fo & 0x3 == 0x2: # NR2
546 value = float(chunk)
548 elif fo & 0x3 == 0x3: # NR3
549 value = float(chunk)
551 else:
552 raise error.SubstrateUnderrunError(
553 'Unknown NR (tag %s)' % fo
554 )
556 except ValueError:
557 raise error.SubstrateUnderrunError(
558 'Bad character Real syntax'
559 )
561 else:
562 raise error.SubstrateUnderrunError(
563 'Unknown encoding (tag %s)' % fo
564 )
566 yield self._createComponent(asn1Spec, tagSet, value, **options)
569class AbstractConstructedPayloadDecoder(AbstractPayloadDecoder):
570 protoComponent = None
573class ConstructedPayloadDecoderBase(AbstractConstructedPayloadDecoder):
574 protoRecordComponent = None
575 protoSequenceComponent = None
577 def _getComponentTagMap(self, asn1Object, idx):
578 raise NotImplementedError()
580 def _getComponentPositionByType(self, asn1Object, tagSet, idx):
581 raise NotImplementedError()
583 def _decodeComponentsSchemaless(
584 self, substrate, tagSet=None, decodeFun=None,
585 length=None, **options):
587 asn1Object = None
589 components = []
590 componentTypes = set()
592 original_position = substrate.tell()
594 while length == -1 or substrate.tell() < original_position + length:
595 for component in decodeFun(substrate, **options):
596 if isinstance(component, SubstrateUnderrunError):
597 yield component
599 if length == -1 and component is eoo.endOfOctets:
600 break
602 components.append(component)
603 componentTypes.add(component.tagSet)
605 # Now we have to guess is it SEQUENCE/SET or SEQUENCE OF/SET OF
606 # The heuristics is:
607 # * 1+ components of different types -> likely SEQUENCE/SET
608 # * otherwise -> likely SEQUENCE OF/SET OF
609 if len(componentTypes) > 1:
610 protoComponent = self.protoRecordComponent
612 else:
613 protoComponent = self.protoSequenceComponent
615 asn1Object = protoComponent.clone(
616 # construct tagSet from base tag from prototype ASN.1 object
617 # and additional tags recovered from the substrate
618 tagSet=tag.TagSet(protoComponent.tagSet.baseTag, *tagSet.superTags)
619 )
621 if LOG:
622 LOG('guessed %r container type (pass `asn1Spec` to guide the '
623 'decoder)' % asn1Object)
625 for idx, component in enumerate(components):
626 asn1Object.setComponentByPosition(
627 idx, component,
628 verifyConstraints=False,
629 matchTags=False, matchConstraints=False
630 )
632 yield asn1Object
634 def valueDecoder(self, substrate, asn1Spec,
635 tagSet=None, length=None, state=None,
636 decodeFun=None, substrateFun=None,
637 **options):
638 if tagSet[0].tagFormat != tag.tagFormatConstructed:
639 raise error.PyAsn1Error('Constructed tag format expected')
641 original_position = substrate.tell()
643 if substrateFun:
644 if asn1Spec is not None:
645 asn1Object = asn1Spec.clone()
647 elif self.protoComponent is not None:
648 asn1Object = self.protoComponent.clone(tagSet=tagSet)
650 else:
651 asn1Object = self.protoRecordComponent, self.protoSequenceComponent
653 for chunk in substrateFun(asn1Object, substrate, length, options):
654 yield chunk
656 return
658 if asn1Spec is None:
659 for asn1Object in self._decodeComponentsSchemaless(
660 substrate, tagSet=tagSet, decodeFun=decodeFun,
661 length=length, **options):
662 if isinstance(asn1Object, SubstrateUnderrunError):
663 yield asn1Object
665 if substrate.tell() < original_position + length:
666 if LOG:
667 for trailing in readFromStream(substrate, context=options):
668 if isinstance(trailing, SubstrateUnderrunError):
669 yield trailing
671 LOG('Unused trailing %d octets encountered: %s' % (
672 len(trailing), debug.hexdump(trailing)))
674 yield asn1Object
676 return
678 asn1Object = asn1Spec.clone()
679 asn1Object.clear()
681 options = self._passAsn1Object(asn1Object, options)
683 if asn1Spec.typeId in (univ.Sequence.typeId, univ.Set.typeId):
685 namedTypes = asn1Spec.componentType
687 isSetType = asn1Spec.typeId == univ.Set.typeId
688 isDeterministic = not isSetType and not namedTypes.hasOptionalOrDefault
690 if LOG:
691 LOG('decoding %sdeterministic %s type %r chosen by type ID' % (
692 not isDeterministic and 'non-' or '', isSetType and 'SET' or '',
693 asn1Spec))
695 seenIndices = set()
696 idx = 0
697 while substrate.tell() - original_position < length:
698 if not namedTypes:
699 componentType = None
701 elif isSetType:
702 componentType = namedTypes.tagMapUnique
704 else:
705 try:
706 if isDeterministic:
707 componentType = namedTypes[idx].asn1Object
709 elif namedTypes[idx].isOptional or namedTypes[idx].isDefaulted:
710 componentType = namedTypes.getTagMapNearPosition(idx)
712 else:
713 componentType = namedTypes[idx].asn1Object
715 except IndexError:
716 raise error.PyAsn1Error(
717 'Excessive components decoded at %r' % (asn1Spec,)
718 )
720 for component in decodeFun(substrate, componentType, **options):
721 if isinstance(component, SubstrateUnderrunError):
722 yield component
724 if not isDeterministic and namedTypes:
725 if isSetType:
726 idx = namedTypes.getPositionByType(component.effectiveTagSet)
728 elif namedTypes[idx].isOptional or namedTypes[idx].isDefaulted:
729 idx = namedTypes.getPositionNearType(component.effectiveTagSet, idx)
731 asn1Object.setComponentByPosition(
732 idx, component,
733 verifyConstraints=False,
734 matchTags=False, matchConstraints=False
735 )
737 seenIndices.add(idx)
738 idx += 1
740 if LOG:
741 LOG('seen component indices %s' % seenIndices)
743 if namedTypes:
744 if not namedTypes.requiredComponents.issubset(seenIndices):
745 raise error.PyAsn1Error(
746 'ASN.1 object %s has uninitialized '
747 'components' % asn1Object.__class__.__name__)
749 if namedTypes.hasOpenTypes:
751 openTypes = options.get('openTypes', {})
753 if LOG:
754 LOG('user-specified open types map:')
756 for k, v in openTypes.items():
757 LOG('%s -> %r' % (k, v))
759 if openTypes or options.get('decodeOpenTypes', False):
761 for idx, namedType in enumerate(namedTypes.namedTypes):
762 if not namedType.openType:
763 continue
765 if namedType.isOptional and not asn1Object.getComponentByPosition(idx).isValue:
766 continue
768 governingValue = asn1Object.getComponentByName(
769 namedType.openType.name
770 )
772 try:
773 openType = openTypes[governingValue]
775 except KeyError:
777 if LOG:
778 LOG('default open types map of component '
779 '"%s.%s" governed by component "%s.%s"'
780 ':' % (asn1Object.__class__.__name__,
781 namedType.name,
782 asn1Object.__class__.__name__,
783 namedType.openType.name))
785 for k, v in namedType.openType.items():
786 LOG('%s -> %r' % (k, v))
788 try:
789 openType = namedType.openType[governingValue]
791 except KeyError:
792 if LOG:
793 LOG('failed to resolve open type by governing '
794 'value %r' % (governingValue,))
795 continue
797 if LOG:
798 LOG('resolved open type %r by governing '
799 'value %r' % (openType, governingValue))
801 containerValue = asn1Object.getComponentByPosition(idx)
803 if containerValue.typeId in (
804 univ.SetOf.typeId, univ.SequenceOf.typeId):
806 for pos, containerElement in enumerate(
807 containerValue):
809 stream = asSeekableStream(containerValue[pos].asOctets())
811 for component in decodeFun(stream, asn1Spec=openType, **options):
812 if isinstance(component, SubstrateUnderrunError):
813 yield component
815 containerValue[pos] = component
817 else:
818 stream = asSeekableStream(asn1Object.getComponentByPosition(idx).asOctets())
820 for component in decodeFun(stream, asn1Spec=openType, **options):
821 if isinstance(component, SubstrateUnderrunError):
822 yield component
824 asn1Object.setComponentByPosition(idx, component)
826 else:
827 inconsistency = asn1Object.isInconsistent
828 if inconsistency:
829 raise inconsistency
831 else:
832 componentType = asn1Spec.componentType
834 if LOG:
835 LOG('decoding type %r chosen by given `asn1Spec`' % componentType)
837 idx = 0
839 while substrate.tell() - original_position < length:
840 for component in decodeFun(substrate, componentType, **options):
841 if isinstance(component, SubstrateUnderrunError):
842 yield component
844 asn1Object.setComponentByPosition(
845 idx, component,
846 verifyConstraints=False,
847 matchTags=False, matchConstraints=False
848 )
850 idx += 1
852 yield asn1Object
854 def indefLenValueDecoder(self, substrate, asn1Spec,
855 tagSet=None, length=None, state=None,
856 decodeFun=None, substrateFun=None,
857 **options):
858 if tagSet[0].tagFormat != tag.tagFormatConstructed:
859 raise error.PyAsn1Error('Constructed tag format expected')
861 if substrateFun is not None:
862 if asn1Spec is not None:
863 asn1Object = asn1Spec.clone()
865 elif self.protoComponent is not None:
866 asn1Object = self.protoComponent.clone(tagSet=tagSet)
868 else:
869 asn1Object = self.protoRecordComponent, self.protoSequenceComponent
871 for chunk in substrateFun(asn1Object, substrate, length, options):
872 yield chunk
874 return
876 if asn1Spec is None:
877 for asn1Object in self._decodeComponentsSchemaless(
878 substrate, tagSet=tagSet, decodeFun=decodeFun,
879 length=length, **dict(options, allowEoo=True)):
880 if isinstance(asn1Object, SubstrateUnderrunError):
881 yield asn1Object
883 yield asn1Object
885 return
887 asn1Object = asn1Spec.clone()
888 asn1Object.clear()
890 options = self._passAsn1Object(asn1Object, options)
892 if asn1Spec.typeId in (univ.Sequence.typeId, univ.Set.typeId):
894 namedTypes = asn1Object.componentType
896 isSetType = asn1Object.typeId == univ.Set.typeId
897 isDeterministic = not isSetType and not namedTypes.hasOptionalOrDefault
899 if LOG:
900 LOG('decoding %sdeterministic %s type %r chosen by type ID' % (
901 not isDeterministic and 'non-' or '', isSetType and 'SET' or '',
902 asn1Spec))
904 seenIndices = set()
906 idx = 0
908 while True: # loop over components
909 if len(namedTypes) <= idx:
910 asn1Spec = None
912 elif isSetType:
913 asn1Spec = namedTypes.tagMapUnique
915 else:
916 try:
917 if isDeterministic:
918 asn1Spec = namedTypes[idx].asn1Object
920 elif namedTypes[idx].isOptional or namedTypes[idx].isDefaulted:
921 asn1Spec = namedTypes.getTagMapNearPosition(idx)
923 else:
924 asn1Spec = namedTypes[idx].asn1Object
926 except IndexError:
927 raise error.PyAsn1Error(
928 'Excessive components decoded at %r' % (asn1Object,)
929 )
931 for component in decodeFun(substrate, asn1Spec, allowEoo=True, **options):
933 if isinstance(component, SubstrateUnderrunError):
934 yield component
936 if component is eoo.endOfOctets:
937 break
939 if component is eoo.endOfOctets:
940 break
942 if not isDeterministic and namedTypes:
943 if isSetType:
944 idx = namedTypes.getPositionByType(component.effectiveTagSet)
946 elif namedTypes[idx].isOptional or namedTypes[idx].isDefaulted:
947 idx = namedTypes.getPositionNearType(component.effectiveTagSet, idx)
949 asn1Object.setComponentByPosition(
950 idx, component,
951 verifyConstraints=False,
952 matchTags=False, matchConstraints=False
953 )
955 seenIndices.add(idx)
956 idx += 1
958 if LOG:
959 LOG('seen component indices %s' % seenIndices)
961 if namedTypes:
962 if not namedTypes.requiredComponents.issubset(seenIndices):
963 raise error.PyAsn1Error(
964 'ASN.1 object %s has uninitialized '
965 'components' % asn1Object.__class__.__name__)
967 if namedTypes.hasOpenTypes:
969 openTypes = options.get('openTypes', {})
971 if LOG:
972 LOG('user-specified open types map:')
974 for k, v in openTypes.items():
975 LOG('%s -> %r' % (k, v))
977 if openTypes or options.get('decodeOpenTypes', False):
979 for idx, namedType in enumerate(namedTypes.namedTypes):
980 if not namedType.openType:
981 continue
983 if namedType.isOptional and not asn1Object.getComponentByPosition(idx).isValue:
984 continue
986 governingValue = asn1Object.getComponentByName(
987 namedType.openType.name
988 )
990 try:
991 openType = openTypes[governingValue]
993 except KeyError:
995 if LOG:
996 LOG('default open types map of component '
997 '"%s.%s" governed by component "%s.%s"'
998 ':' % (asn1Object.__class__.__name__,
999 namedType.name,
1000 asn1Object.__class__.__name__,
1001 namedType.openType.name))
1003 for k, v in namedType.openType.items():
1004 LOG('%s -> %r' % (k, v))
1006 try:
1007 openType = namedType.openType[governingValue]
1009 except KeyError:
1010 if LOG:
1011 LOG('failed to resolve open type by governing '
1012 'value %r' % (governingValue,))
1013 continue
1015 if LOG:
1016 LOG('resolved open type %r by governing '
1017 'value %r' % (openType, governingValue))
1019 containerValue = asn1Object.getComponentByPosition(idx)
1021 if containerValue.typeId in (
1022 univ.SetOf.typeId, univ.SequenceOf.typeId):
1024 for pos, containerElement in enumerate(
1025 containerValue):
1027 stream = asSeekableStream(containerValue[pos].asOctets())
1029 for component in decodeFun(stream, asn1Spec=openType,
1030 **dict(options, allowEoo=True)):
1031 if isinstance(component, SubstrateUnderrunError):
1032 yield component
1034 if component is eoo.endOfOctets:
1035 break
1037 containerValue[pos] = component
1039 else:
1040 stream = asSeekableStream(asn1Object.getComponentByPosition(idx).asOctets())
1041 for component in decodeFun(stream, asn1Spec=openType,
1042 **dict(options, allowEoo=True)):
1043 if isinstance(component, SubstrateUnderrunError):
1044 yield component
1046 if component is eoo.endOfOctets:
1047 break
1049 asn1Object.setComponentByPosition(idx, component)
1051 else:
1052 inconsistency = asn1Object.isInconsistent
1053 if inconsistency:
1054 raise inconsistency
1056 else:
1057 componentType = asn1Spec.componentType
1059 if LOG:
1060 LOG('decoding type %r chosen by given `asn1Spec`' % componentType)
1062 idx = 0
1064 while True:
1066 for component in decodeFun(
1067 substrate, componentType, allowEoo=True, **options):
1069 if isinstance(component, SubstrateUnderrunError):
1070 yield component
1072 if component is eoo.endOfOctets:
1073 break
1075 if component is eoo.endOfOctets:
1076 break
1078 asn1Object.setComponentByPosition(
1079 idx, component,
1080 verifyConstraints=False,
1081 matchTags=False, matchConstraints=False
1082 )
1084 idx += 1
1086 yield asn1Object
1089class SequenceOrSequenceOfPayloadDecoder(ConstructedPayloadDecoderBase):
1090 protoRecordComponent = univ.Sequence()
1091 protoSequenceComponent = univ.SequenceOf()
1094class SequencePayloadDecoder(SequenceOrSequenceOfPayloadDecoder):
1095 protoComponent = univ.Sequence()
1098class SequenceOfPayloadDecoder(SequenceOrSequenceOfPayloadDecoder):
1099 protoComponent = univ.SequenceOf()
1102class SetOrSetOfPayloadDecoder(ConstructedPayloadDecoderBase):
1103 protoRecordComponent = univ.Set()
1104 protoSequenceComponent = univ.SetOf()
1107class SetPayloadDecoder(SetOrSetOfPayloadDecoder):
1108 protoComponent = univ.Set()
1111class SetOfPayloadDecoder(SetOrSetOfPayloadDecoder):
1112 protoComponent = univ.SetOf()
1115class ChoicePayloadDecoder(ConstructedPayloadDecoderBase):
1116 protoComponent = univ.Choice()
1118 def valueDecoder(self, substrate, asn1Spec,
1119 tagSet=None, length=None, state=None,
1120 decodeFun=None, substrateFun=None,
1121 **options):
1122 if asn1Spec is None:
1123 asn1Object = self.protoComponent.clone(tagSet=tagSet)
1125 else:
1126 asn1Object = asn1Spec.clone()
1128 if substrateFun:
1129 for chunk in substrateFun(asn1Object, substrate, length, options):
1130 yield chunk
1132 return
1134 options = self._passAsn1Object(asn1Object, options)
1136 if asn1Object.tagSet == tagSet:
1137 if LOG:
1138 LOG('decoding %s as explicitly tagged CHOICE' % (tagSet,))
1140 for component in decodeFun(
1141 substrate, asn1Object.componentTagMap, **options):
1142 if isinstance(component, SubstrateUnderrunError):
1143 yield component
1145 else:
1146 if LOG:
1147 LOG('decoding %s as untagged CHOICE' % (tagSet,))
1149 for component in decodeFun(
1150 substrate, asn1Object.componentTagMap, tagSet, length,
1151 state, **options):
1152 if isinstance(component, SubstrateUnderrunError):
1153 yield component
1155 effectiveTagSet = component.effectiveTagSet
1157 if LOG:
1158 LOG('decoded component %s, effective tag set %s' % (component, effectiveTagSet))
1160 asn1Object.setComponentByType(
1161 effectiveTagSet, component,
1162 verifyConstraints=False,
1163 matchTags=False, matchConstraints=False,
1164 innerFlag=False
1165 )
1167 yield asn1Object
1169 def indefLenValueDecoder(self, substrate, asn1Spec,
1170 tagSet=None, length=None, state=None,
1171 decodeFun=None, substrateFun=None,
1172 **options):
1173 if asn1Spec is None:
1174 asn1Object = self.protoComponent.clone(tagSet=tagSet)
1176 else:
1177 asn1Object = asn1Spec.clone()
1179 if substrateFun:
1180 for chunk in substrateFun(asn1Object, substrate, length, options):
1181 yield chunk
1183 return
1185 options = self._passAsn1Object(asn1Object, options)
1187 isTagged = asn1Object.tagSet == tagSet
1189 if LOG:
1190 LOG('decoding %s as %stagged CHOICE' % (
1191 tagSet, isTagged and 'explicitly ' or 'un'))
1193 while True:
1195 if isTagged:
1196 iterator = decodeFun(
1197 substrate, asn1Object.componentType.tagMapUnique,
1198 **dict(options, allowEoo=True))
1200 else:
1201 iterator = decodeFun(
1202 substrate, asn1Object.componentType.tagMapUnique,
1203 tagSet, length, state, **dict(options, allowEoo=True))
1205 for component in iterator:
1207 if isinstance(component, SubstrateUnderrunError):
1208 yield component
1210 if component is eoo.endOfOctets:
1211 break
1213 effectiveTagSet = component.effectiveTagSet
1215 if LOG:
1216 LOG('decoded component %s, effective tag set '
1217 '%s' % (component, effectiveTagSet))
1219 asn1Object.setComponentByType(
1220 effectiveTagSet, component,
1221 verifyConstraints=False,
1222 matchTags=False, matchConstraints=False,
1223 innerFlag=False
1224 )
1226 if not isTagged:
1227 break
1229 if not isTagged or component is eoo.endOfOctets:
1230 break
1232 yield asn1Object
1235class AnyPayloadDecoder(AbstractSimplePayloadDecoder):
1236 protoComponent = univ.Any()
1238 def valueDecoder(self, substrate, asn1Spec,
1239 tagSet=None, length=None, state=None,
1240 decodeFun=None, substrateFun=None,
1241 **options):
1242 if asn1Spec is None:
1243 isUntagged = True
1245 elif asn1Spec.__class__ is tagmap.TagMap:
1246 isUntagged = tagSet not in asn1Spec.tagMap
1248 else:
1249 isUntagged = tagSet != asn1Spec.tagSet
1251 if isUntagged:
1252 fullPosition = substrate.markedPosition
1253 currentPosition = substrate.tell()
1255 substrate.seek(fullPosition, os.SEEK_SET)
1256 length += currentPosition - fullPosition
1258 if LOG:
1259 for chunk in peekIntoStream(substrate, length):
1260 if isinstance(chunk, SubstrateUnderrunError):
1261 yield chunk
1262 LOG('decoding as untagged ANY, substrate '
1263 '%s' % debug.hexdump(chunk))
1265 if substrateFun:
1266 for chunk in substrateFun(
1267 self._createComponent(asn1Spec, tagSet, noValue, **options),
1268 substrate, length, options):
1269 yield chunk
1271 return
1273 for chunk in readFromStream(substrate, length, options):
1274 if isinstance(chunk, SubstrateUnderrunError):
1275 yield chunk
1277 yield self._createComponent(asn1Spec, tagSet, chunk, **options)
1279 def indefLenValueDecoder(self, substrate, asn1Spec,
1280 tagSet=None, length=None, state=None,
1281 decodeFun=None, substrateFun=None,
1282 **options):
1283 if asn1Spec is None:
1284 isTagged = False
1286 elif asn1Spec.__class__ is tagmap.TagMap:
1287 isTagged = tagSet in asn1Spec.tagMap
1289 else:
1290 isTagged = tagSet == asn1Spec.tagSet
1292 if isTagged:
1293 # tagged Any type -- consume header substrate
1294 chunk = null
1296 if LOG:
1297 LOG('decoding as tagged ANY')
1299 else:
1300 # TODO: Seems not to be tested
1301 fullPosition = substrate.markedPosition
1302 currentPosition = substrate.tell()
1304 substrate.seek(fullPosition, os.SEEK_SET)
1305 for chunk in readFromStream(substrate, currentPosition - fullPosition, options):
1306 if isinstance(chunk, SubstrateUnderrunError):
1307 yield chunk
1309 if LOG:
1310 LOG('decoding as untagged ANY, header substrate %s' % debug.hexdump(chunk))
1312 # Any components do not inherit initial tag
1313 asn1Spec = self.protoComponent
1315 if substrateFun and substrateFun is not self.substrateCollector:
1316 asn1Object = self._createComponent(
1317 asn1Spec, tagSet, noValue, **options)
1319 for chunk in substrateFun(
1320 asn1Object, chunk + substrate, length + len(chunk), options):
1321 yield chunk
1323 return
1325 if LOG:
1326 LOG('assembling constructed serialization')
1328 # All inner fragments are of the same type, treat them as octet string
1329 substrateFun = self.substrateCollector
1331 while True: # loop over fragments
1333 for component in decodeFun(
1334 substrate, asn1Spec, substrateFun=substrateFun,
1335 allowEoo=True, **options):
1337 if isinstance(component, SubstrateUnderrunError):
1338 yield component
1340 if component is eoo.endOfOctets:
1341 break
1343 if component is eoo.endOfOctets:
1344 break
1346 chunk += component
1348 if substrateFun:
1349 yield chunk # TODO: Weird
1351 else:
1352 yield self._createComponent(asn1Spec, tagSet, chunk, **options)
1355# character string types
1356class UTF8StringPayloadDecoder(OctetStringPayloadDecoder):
1357 protoComponent = char.UTF8String()
1360class NumericStringPayloadDecoder(OctetStringPayloadDecoder):
1361 protoComponent = char.NumericString()
1364class PrintableStringPayloadDecoder(OctetStringPayloadDecoder):
1365 protoComponent = char.PrintableString()
1368class TeletexStringPayloadDecoder(OctetStringPayloadDecoder):
1369 protoComponent = char.TeletexString()
1372class VideotexStringPayloadDecoder(OctetStringPayloadDecoder):
1373 protoComponent = char.VideotexString()
1376class IA5StringPayloadDecoder(OctetStringPayloadDecoder):
1377 protoComponent = char.IA5String()
1380class GraphicStringPayloadDecoder(OctetStringPayloadDecoder):
1381 protoComponent = char.GraphicString()
1384class VisibleStringPayloadDecoder(OctetStringPayloadDecoder):
1385 protoComponent = char.VisibleString()
1388class GeneralStringPayloadDecoder(OctetStringPayloadDecoder):
1389 protoComponent = char.GeneralString()
1392class UniversalStringPayloadDecoder(OctetStringPayloadDecoder):
1393 protoComponent = char.UniversalString()
1396class BMPStringPayloadDecoder(OctetStringPayloadDecoder):
1397 protoComponent = char.BMPString()
1400# "useful" types
1401class ObjectDescriptorPayloadDecoder(OctetStringPayloadDecoder):
1402 protoComponent = useful.ObjectDescriptor()
1405class GeneralizedTimePayloadDecoder(OctetStringPayloadDecoder):
1406 protoComponent = useful.GeneralizedTime()
1409class UTCTimePayloadDecoder(OctetStringPayloadDecoder):
1410 protoComponent = useful.UTCTime()
1413TAG_MAP = {
1414 univ.Integer.tagSet: IntegerPayloadDecoder(),
1415 univ.Boolean.tagSet: BooleanPayloadDecoder(),
1416 univ.BitString.tagSet: BitStringPayloadDecoder(),
1417 univ.OctetString.tagSet: OctetStringPayloadDecoder(),
1418 univ.Null.tagSet: NullPayloadDecoder(),
1419 univ.ObjectIdentifier.tagSet: ObjectIdentifierPayloadDecoder(),
1420 univ.Enumerated.tagSet: IntegerPayloadDecoder(),
1421 univ.Real.tagSet: RealPayloadDecoder(),
1422 univ.Sequence.tagSet: SequenceOrSequenceOfPayloadDecoder(), # conflicts with SequenceOf
1423 univ.Set.tagSet: SetOrSetOfPayloadDecoder(), # conflicts with SetOf
1424 univ.Choice.tagSet: ChoicePayloadDecoder(), # conflicts with Any
1425 # character string types
1426 char.UTF8String.tagSet: UTF8StringPayloadDecoder(),
1427 char.NumericString.tagSet: NumericStringPayloadDecoder(),
1428 char.PrintableString.tagSet: PrintableStringPayloadDecoder(),
1429 char.TeletexString.tagSet: TeletexStringPayloadDecoder(),
1430 char.VideotexString.tagSet: VideotexStringPayloadDecoder(),
1431 char.IA5String.tagSet: IA5StringPayloadDecoder(),
1432 char.GraphicString.tagSet: GraphicStringPayloadDecoder(),
1433 char.VisibleString.tagSet: VisibleStringPayloadDecoder(),
1434 char.GeneralString.tagSet: GeneralStringPayloadDecoder(),
1435 char.UniversalString.tagSet: UniversalStringPayloadDecoder(),
1436 char.BMPString.tagSet: BMPStringPayloadDecoder(),
1437 # useful types
1438 useful.ObjectDescriptor.tagSet: ObjectDescriptorPayloadDecoder(),
1439 useful.GeneralizedTime.tagSet: GeneralizedTimePayloadDecoder(),
1440 useful.UTCTime.tagSet: UTCTimePayloadDecoder()
1441}
1443# Type-to-codec map for ambiguous ASN.1 types
1444TYPE_MAP = {
1445 univ.Set.typeId: SetPayloadDecoder(),
1446 univ.SetOf.typeId: SetOfPayloadDecoder(),
1447 univ.Sequence.typeId: SequencePayloadDecoder(),
1448 univ.SequenceOf.typeId: SequenceOfPayloadDecoder(),
1449 univ.Choice.typeId: ChoicePayloadDecoder(),
1450 univ.Any.typeId: AnyPayloadDecoder()
1451}
1453# Put in non-ambiguous types for faster codec lookup
1454for typeDecoder in TAG_MAP.values():
1455 if typeDecoder.protoComponent is not None:
1456 typeId = typeDecoder.protoComponent.__class__.typeId
1457 if typeId is not None and typeId not in TYPE_MAP:
1458 TYPE_MAP[typeId] = typeDecoder
1461(stDecodeTag,
1462 stDecodeLength,
1463 stGetValueDecoder,
1464 stGetValueDecoderByAsn1Spec,
1465 stGetValueDecoderByTag,
1466 stTryAsExplicitTag,
1467 stDecodeValue,
1468 stDumpRawValue,
1469 stErrorCondition,
1470 stStop) = [x for x in range(10)]
1473EOO_SENTINEL = ints2octs((0, 0))
1476class SingleItemDecoder(object):
1477 defaultErrorState = stErrorCondition
1478 #defaultErrorState = stDumpRawValue
1479 defaultRawDecoder = AnyPayloadDecoder()
1481 supportIndefLength = True
1483 TAG_MAP = TAG_MAP
1484 TYPE_MAP = TYPE_MAP
1486 def __init__(self, **options):
1487 self._tagMap = options.get('tagMap', self.TAG_MAP)
1488 self._typeMap = options.get('typeMap', self.TYPE_MAP)
1490 # Tag & TagSet objects caches
1491 self._tagCache = {}
1492 self._tagSetCache = {}
1494 def __call__(self, substrate, asn1Spec=None,
1495 tagSet=None, length=None, state=stDecodeTag,
1496 decodeFun=None, substrateFun=None,
1497 **options):
1499 allowEoo = options.pop('allowEoo', False)
1501 if LOG:
1502 LOG('decoder called at scope %s with state %d, working with up '
1503 'to %s octets of substrate: '
1504 '%s' % (debug.scope, state, length, substrate))
1506 # Look for end-of-octets sentinel
1507 if allowEoo and self.supportIndefLength:
1509 for eoo_candidate in readFromStream(substrate, 2, options):
1510 if isinstance(eoo_candidate, SubstrateUnderrunError):
1511 yield eoo_candidate
1513 if eoo_candidate == EOO_SENTINEL:
1514 if LOG:
1515 LOG('end-of-octets sentinel found')
1516 yield eoo.endOfOctets
1517 return
1519 else:
1520 substrate.seek(-2, os.SEEK_CUR)
1522 tagMap = self._tagMap
1523 typeMap = self._typeMap
1524 tagCache = self._tagCache
1525 tagSetCache = self._tagSetCache
1527 value = noValue
1529 substrate.markedPosition = substrate.tell()
1531 while state is not stStop:
1533 if state is stDecodeTag:
1534 # Decode tag
1535 isShortTag = True
1537 for firstByte in readFromStream(substrate, 1, options):
1538 if isinstance(firstByte, SubstrateUnderrunError):
1539 yield firstByte
1541 firstOctet = ord(firstByte)
1543 try:
1544 lastTag = tagCache[firstOctet]
1546 except KeyError:
1547 integerTag = firstOctet
1548 tagClass = integerTag & 0xC0
1549 tagFormat = integerTag & 0x20
1550 tagId = integerTag & 0x1F
1552 if tagId == 0x1F:
1553 isShortTag = False
1554 lengthOctetIdx = 0
1555 tagId = 0
1557 while True:
1558 for integerByte in readFromStream(substrate, 1, options):
1559 if isinstance(integerByte, SubstrateUnderrunError):
1560 yield integerByte
1562 if not integerByte:
1563 raise error.SubstrateUnderrunError(
1564 'Short octet stream on long tag decoding'
1565 )
1567 integerTag = ord(integerByte)
1568 lengthOctetIdx += 1
1569 tagId <<= 7
1570 tagId |= (integerTag & 0x7F)
1572 if not integerTag & 0x80:
1573 break
1575 lastTag = tag.Tag(
1576 tagClass=tagClass, tagFormat=tagFormat, tagId=tagId
1577 )
1579 if isShortTag:
1580 # cache short tags
1581 tagCache[firstOctet] = lastTag
1583 if tagSet is None:
1584 if isShortTag:
1585 try:
1586 tagSet = tagSetCache[firstOctet]
1588 except KeyError:
1589 # base tag not recovered
1590 tagSet = tag.TagSet((), lastTag)
1591 tagSetCache[firstOctet] = tagSet
1592 else:
1593 tagSet = tag.TagSet((), lastTag)
1595 else:
1596 tagSet = lastTag + tagSet
1598 state = stDecodeLength
1600 if LOG:
1601 LOG('tag decoded into %s, decoding length' % tagSet)
1603 if state is stDecodeLength:
1604 # Decode length
1605 for firstOctet in readFromStream(substrate, 1, options):
1606 if isinstance(firstOctet, SubstrateUnderrunError):
1607 yield firstOctet
1609 firstOctet = ord(firstOctet)
1611 if firstOctet < 128:
1612 length = firstOctet
1614 elif firstOctet > 128:
1615 size = firstOctet & 0x7F
1616 # encoded in size bytes
1617 for encodedLength in readFromStream(substrate, size, options):
1618 if isinstance(encodedLength, SubstrateUnderrunError):
1619 yield encodedLength
1620 encodedLength = list(encodedLength)
1621 # missing check on maximum size, which shouldn't be a
1622 # problem, we can handle more than is possible
1623 if len(encodedLength) != size:
1624 raise error.SubstrateUnderrunError(
1625 '%s<%s at %s' % (size, len(encodedLength), tagSet)
1626 )
1628 length = 0
1629 for lengthOctet in encodedLength:
1630 length <<= 8
1631 length |= oct2int(lengthOctet)
1632 size += 1
1634 else: # 128 means indefinite
1635 length = -1
1637 if length == -1 and not self.supportIndefLength:
1638 raise error.PyAsn1Error('Indefinite length encoding not supported by this codec')
1640 state = stGetValueDecoder
1642 if LOG:
1643 LOG('value length decoded into %d' % length)
1645 if state is stGetValueDecoder:
1646 if asn1Spec is None:
1647 state = stGetValueDecoderByTag
1649 else:
1650 state = stGetValueDecoderByAsn1Spec
1651 #
1652 # There're two ways of creating subtypes in ASN.1 what influences
1653 # decoder operation. These methods are:
1654 # 1) Either base types used in or no IMPLICIT tagging has been
1655 # applied on subtyping.
1656 # 2) Subtype syntax drops base type information (by means of
1657 # IMPLICIT tagging.
1658 # The first case allows for complete tag recovery from substrate
1659 # while the second one requires original ASN.1 type spec for
1660 # decoding.
1661 #
1662 # In either case a set of tags (tagSet) is coming from substrate
1663 # in an incremental, tag-by-tag fashion (this is the case of
1664 # EXPLICIT tag which is most basic). Outermost tag comes first
1665 # from the wire.
1666 #
1667 if state is stGetValueDecoderByTag:
1668 try:
1669 concreteDecoder = tagMap[tagSet]
1671 except KeyError:
1672 concreteDecoder = None
1674 if concreteDecoder:
1675 state = stDecodeValue
1677 else:
1678 try:
1679 concreteDecoder = tagMap[tagSet[:1]]
1681 except KeyError:
1682 concreteDecoder = None
1684 if concreteDecoder:
1685 state = stDecodeValue
1686 else:
1687 state = stTryAsExplicitTag
1689 if LOG:
1690 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'))
1691 debug.scope.push(concreteDecoder is None and '?' or concreteDecoder.protoComponent.__class__.__name__)
1693 if state is stGetValueDecoderByAsn1Spec:
1695 if asn1Spec.__class__ is tagmap.TagMap:
1696 try:
1697 chosenSpec = asn1Spec[tagSet]
1699 except KeyError:
1700 chosenSpec = None
1702 if LOG:
1703 LOG('candidate ASN.1 spec is a map of:')
1705 for firstOctet, v in asn1Spec.presentTypes.items():
1706 LOG(' %s -> %s' % (firstOctet, v.__class__.__name__))
1708 if asn1Spec.skipTypes:
1709 LOG('but neither of: ')
1710 for firstOctet, v in asn1Spec.skipTypes.items():
1711 LOG(' %s -> %s' % (firstOctet, v.__class__.__name__))
1712 LOG('new candidate ASN.1 spec is %s, chosen by %s' % (chosenSpec is None and '<none>' or chosenSpec.prettyPrintType(), tagSet))
1714 elif tagSet == asn1Spec.tagSet or tagSet in asn1Spec.tagMap:
1715 chosenSpec = asn1Spec
1716 if LOG:
1717 LOG('candidate ASN.1 spec is %s' % asn1Spec.__class__.__name__)
1719 else:
1720 chosenSpec = None
1722 if chosenSpec is not None:
1723 try:
1724 # ambiguous type or just faster codec lookup
1725 concreteDecoder = typeMap[chosenSpec.typeId]
1727 if LOG:
1728 LOG('value decoder chosen for an ambiguous type by type ID %s' % (chosenSpec.typeId,))
1730 except KeyError:
1731 # use base type for codec lookup to recover untagged types
1732 baseTagSet = tag.TagSet(chosenSpec.tagSet.baseTag, chosenSpec.tagSet.baseTag)
1733 try:
1734 # base type or tagged subtype
1735 concreteDecoder = tagMap[baseTagSet]
1737 if LOG:
1738 LOG('value decoder chosen by base %s' % (baseTagSet,))
1740 except KeyError:
1741 concreteDecoder = None
1743 if concreteDecoder:
1744 asn1Spec = chosenSpec
1745 state = stDecodeValue
1747 else:
1748 state = stTryAsExplicitTag
1750 else:
1751 concreteDecoder = None
1752 state = stTryAsExplicitTag
1754 if LOG:
1755 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'))
1756 debug.scope.push(chosenSpec is None and '?' or chosenSpec.__class__.__name__)
1758 if state is stDecodeValue:
1759 if not options.get('recursiveFlag', True) and not substrateFun: # deprecate this
1760 substrateFun = lambda a, b, c: (a, b[:c])
1762 original_position = substrate.tell()
1764 if length == -1: # indef length
1765 for value in concreteDecoder.indefLenValueDecoder(
1766 substrate, asn1Spec,
1767 tagSet, length, stGetValueDecoder,
1768 self, substrateFun, **options):
1769 if isinstance(value, SubstrateUnderrunError):
1770 yield value
1772 else:
1773 for value in concreteDecoder.valueDecoder(
1774 substrate, asn1Spec,
1775 tagSet, length, stGetValueDecoder,
1776 self, substrateFun, **options):
1777 if isinstance(value, SubstrateUnderrunError):
1778 yield value
1780 bytesRead = substrate.tell() - original_position
1781 if bytesRead != length:
1782 raise PyAsn1Error(
1783 "Read %s bytes instead of expected %s." % (bytesRead, length))
1785 if LOG:
1786 LOG('codec %s yields type %s, value:\n%s\n...' % (
1787 concreteDecoder.__class__.__name__, value.__class__.__name__,
1788 isinstance(value, base.Asn1Item) and value.prettyPrint() or value))
1790 state = stStop
1791 break
1793 if state is stTryAsExplicitTag:
1794 if (tagSet and
1795 tagSet[0].tagFormat == tag.tagFormatConstructed and
1796 tagSet[0].tagClass != tag.tagClassUniversal):
1797 # Assume explicit tagging
1798 concreteDecoder = rawPayloadDecoder
1799 state = stDecodeValue
1801 else:
1802 concreteDecoder = None
1803 state = self.defaultErrorState
1805 if LOG:
1806 LOG('codec %s chosen, decoding %s' % (concreteDecoder and concreteDecoder.__class__.__name__ or "<none>", state is stDecodeValue and 'value' or 'as failure'))
1808 if state is stDumpRawValue:
1809 concreteDecoder = self.defaultRawDecoder
1811 if LOG:
1812 LOG('codec %s chosen, decoding value' % concreteDecoder.__class__.__name__)
1814 state = stDecodeValue
1816 if state is stErrorCondition:
1817 raise error.PyAsn1Error(
1818 '%s not in asn1Spec: %r' % (tagSet, asn1Spec)
1819 )
1821 if LOG:
1822 debug.scope.pop()
1823 LOG('decoder left scope %s, call completed' % debug.scope)
1825 yield value
1828class StreamingDecoder(object):
1829 """Create an iterator that turns BER/CER/DER byte stream into ASN.1 objects.
1831 On each iteration, consume whatever BER/CER/DER serialization is
1832 available in the `substrate` stream-like object and turns it into
1833 one or more, possibly nested, ASN.1 objects.
1835 Parameters
1836 ----------
1837 substrate: :py:class:`file`, :py:class:`io.BytesIO`
1838 BER/CER/DER serialization in form of a byte stream
1840 Keyword Args
1841 ------------
1842 asn1Spec: :py:class:`~pyasn1.type.base.PyAsn1Item`
1843 A pyasn1 type object to act as a template guiding the decoder.
1844 Depending on the ASN.1 structure being decoded, `asn1Spec` may
1845 or may not be required. One of the reasons why `asn1Spec` may
1846 me required is that ASN.1 structure is encoded in the *IMPLICIT*
1847 tagging mode.
1849 Yields
1850 ------
1851 : :py:class:`~pyasn1.type.base.PyAsn1Item`, :py:class:`~pyasn1.error.SubstrateUnderrunError`
1852 Decoded ASN.1 object (possibly, nested) or
1853 :py:class:`~pyasn1.error.SubstrateUnderrunError` object indicating
1854 insufficient BER/CER/DER serialization on input to fully recover ASN.1
1855 objects from it.
1857 In the latter case the caller is advised to ensure some more data in
1858 the input stream, then call the iterator again. The decoder will resume
1859 the decoding process using the newly arrived data.
1861 The `context` property of :py:class:`~pyasn1.error.SubstrateUnderrunError`
1862 object might hold a reference to the partially populated ASN.1 object
1863 being reconstructed.
1865 Raises
1866 ------
1867 ~pyasn1.error.PyAsn1Error, ~pyasn1.error.EndOfStreamError
1868 `PyAsn1Error` on deserialization error, `EndOfStreamError` on
1869 premature stream closure.
1871 Examples
1872 --------
1873 Decode BER serialisation without ASN.1 schema
1875 .. code-block:: pycon
1877 >>> stream = io.BytesIO(
1878 ... b'0\t\x02\x01\x01\x02\x01\x02\x02\x01\x03')
1879 >>>
1880 >>> for asn1Object in StreamingDecoder(stream):
1881 ... print(asn1Object)
1882 >>>
1883 SequenceOf:
1884 1 2 3
1886 Decode BER serialisation with ASN.1 schema
1888 .. code-block:: pycon
1890 >>> stream = io.BytesIO(
1891 ... b'0\t\x02\x01\x01\x02\x01\x02\x02\x01\x03')
1892 >>>
1893 >>> schema = SequenceOf(componentType=Integer())
1894 >>>
1895 >>> decoder = StreamingDecoder(stream, asn1Spec=schema)
1896 >>> for asn1Object in decoder:
1897 ... print(asn1Object)
1898 >>>
1899 SequenceOf:
1900 1 2 3
1901 """
1903 SINGLE_ITEM_DECODER = SingleItemDecoder
1905 def __init__(self, substrate, asn1Spec=None, **options):
1906 self._singleItemDecoder = self.SINGLE_ITEM_DECODER(**options)
1907 self._substrate = asSeekableStream(substrate)
1908 self._asn1Spec = asn1Spec
1909 self._options = options
1911 def __iter__(self):
1912 while True:
1913 for asn1Object in self._singleItemDecoder(
1914 self._substrate, self._asn1Spec, **self._options):
1915 yield asn1Object
1917 for chunk in isEndOfStream(self._substrate):
1918 if isinstance(chunk, SubstrateUnderrunError):
1919 yield
1921 break
1923 if chunk:
1924 break
1927class Decoder(object):
1928 """Create a BER decoder object.
1930 Parse BER/CER/DER octet-stream into one, possibly nested, ASN.1 object.
1931 """
1932 STREAMING_DECODER = StreamingDecoder
1934 @classmethod
1935 def __call__(cls, substrate, asn1Spec=None, **options):
1936 """Turns BER/CER/DER octet stream into an ASN.1 object.
1938 Takes BER/CER/DER octet-stream in form of :py:class:`bytes` (Python 3)
1939 or :py:class:`str` (Python 2) and decode it into an ASN.1 object
1940 (e.g. :py:class:`~pyasn1.type.base.PyAsn1Item` derivative) which
1941 may be a scalar or an arbitrary nested structure.
1943 Parameters
1944 ----------
1945 substrate: :py:class:`bytes` (Python 3) or :py:class:`str` (Python 2)
1946 BER/CER/DER octet-stream to parse
1948 Keyword Args
1949 ------------
1950 asn1Spec: :py:class:`~pyasn1.type.base.PyAsn1Item`
1951 A pyasn1 type object (:py:class:`~pyasn1.type.base.PyAsn1Item`
1952 derivative) to act as a template guiding the decoder.
1953 Depending on the ASN.1 structure being decoded, `asn1Spec` may or
1954 may not be required. Most common reason for it to require is that
1955 ASN.1 structure is encoded in *IMPLICIT* tagging mode.
1957 Returns
1958 -------
1959 : :py:class:`tuple`
1960 A tuple of :py:class:`~pyasn1.type.base.PyAsn1Item` object
1961 recovered from BER/CER/DER substrate and the unprocessed trailing
1962 portion of the `substrate` (may be empty)
1964 Raises
1965 ------
1966 : :py:class:`~pyasn1.error.PyAsn1Error`
1967 :py:class:`~pyasn1.error.SubstrateUnderrunError` on insufficient
1968 input or :py:class:`~pyasn1.error.PyAsn1Error` on decoding error.
1970 Examples
1971 --------
1972 Decode BER/CER/DER serialisation without ASN.1 schema
1974 .. code-block:: pycon
1976 >>> s, unprocessed = decode(b'0\t\x02\x01\x01\x02\x01\x02\x02\x01\x03')
1977 >>> str(s)
1978 SequenceOf:
1979 1 2 3
1981 Decode BER/CER/DER serialisation with ASN.1 schema
1983 .. code-block:: pycon
1985 >>> seq = SequenceOf(componentType=Integer())
1986 >>> s, unprocessed = decode(
1987 b'0\t\x02\x01\x01\x02\x01\x02\x02\x01\x03', asn1Spec=seq)
1988 >>> str(s)
1989 SequenceOf:
1990 1 2 3
1992 """
1993 substrate = asSeekableStream(substrate)
1995 streamingDecoder = cls.STREAMING_DECODER(
1996 substrate, asn1Spec, **options)
1998 for asn1Object in streamingDecoder:
1999 if isinstance(asn1Object, SubstrateUnderrunError):
2000 raise error.SubstrateUnderrunError('Short substrate on input')
2002 try:
2003 tail = next(readFromStream(substrate))
2005 except error.EndOfStreamError:
2006 tail = null
2008 return asn1Object, tail
2011#: Turns BER octet stream into an ASN.1 object.
2012#:
2013#: Takes BER octet-stream and decode it into an ASN.1 object
2014#: (e.g. :py:class:`~pyasn1.type.base.PyAsn1Item` derivative) which
2015#: may be a scalar or an arbitrary nested structure.
2016#:
2017#: Parameters
2018#: ----------
2019#: substrate: :py:class:`bytes` (Python 3) or :py:class:`str` (Python 2)
2020#: BER octet-stream
2021#:
2022#: Keyword Args
2023#: ------------
2024#: asn1Spec: any pyasn1 type object e.g. :py:class:`~pyasn1.type.base.PyAsn1Item` derivative
2025#: A pyasn1 type object to act as a template guiding the decoder. Depending on the ASN.1 structure
2026#: being decoded, *asn1Spec* may or may not be required. Most common reason for
2027#: it to require is that ASN.1 structure is encoded in *IMPLICIT* tagging mode.
2028#:
2029#: Returns
2030#: -------
2031#: : :py:class:`tuple`
2032#: A tuple of pyasn1 object recovered from BER substrate (:py:class:`~pyasn1.type.base.PyAsn1Item` derivative)
2033#: and the unprocessed trailing portion of the *substrate* (may be empty)
2034#:
2035#: Raises
2036#: ------
2037#: ~pyasn1.error.PyAsn1Error, ~pyasn1.error.SubstrateUnderrunError
2038#: On decoding errors
2039#:
2040#: Notes
2041#: -----
2042#: This function is deprecated. Please use :py:class:`Decoder` or
2043#: :py:class:`StreamingDecoder` class instance.
2044#:
2045#: Examples
2046#: --------
2047#: Decode BER serialisation without ASN.1 schema
2048#:
2049#: .. code-block:: pycon
2050#:
2051#: >>> s, _ = decode(b'0\t\x02\x01\x01\x02\x01\x02\x02\x01\x03')
2052#: >>> str(s)
2053#: SequenceOf:
2054#: 1 2 3
2055#:
2056#: Decode BER serialisation with ASN.1 schema
2057#:
2058#: .. code-block:: pycon
2059#:
2060#: >>> seq = SequenceOf(componentType=Integer())
2061#: >>> s, _ = decode(b'0\t\x02\x01\x01\x02\x01\x02\x02\x01\x03', asn1Spec=seq)
2062#: >>> str(s)
2063#: SequenceOf:
2064#: 1 2 3
2065#:
2066decode = Decoder()