forked from s434650/CatOrNot
215 lines
7.6 KiB
Python
215 lines
7.6 KiB
Python
#
|
|
# This file is part of pyasn1 software.
|
|
#
|
|
# Copyright (c) 2005-2018, Ilya Etingof <etingof@gmail.com>
|
|
# License: http://snmplabs.com/pyasn1/license.html
|
|
#
|
|
from pyasn1 import debug
|
|
from pyasn1 import error
|
|
from pyasn1.type import base
|
|
from pyasn1.type import char
|
|
from pyasn1.type import tag
|
|
from pyasn1.type import univ
|
|
from pyasn1.type import useful
|
|
|
|
__all__ = ['decode']
|
|
|
|
|
|
class AbstractScalarDecoder(object):
|
|
def __call__(self, pyObject, asn1Spec, decodeFun=None, **options):
|
|
return asn1Spec.clone(pyObject)
|
|
|
|
|
|
class BitStringDecoder(AbstractScalarDecoder):
|
|
def __call__(self, pyObject, asn1Spec, decodeFun=None, **options):
|
|
return asn1Spec.clone(univ.BitString.fromBinaryString(pyObject))
|
|
|
|
|
|
class SequenceOrSetDecoder(object):
|
|
def __call__(self, pyObject, asn1Spec, decodeFun=None, **options):
|
|
asn1Value = asn1Spec.clone()
|
|
|
|
componentsTypes = asn1Spec.componentType
|
|
|
|
for field in asn1Value:
|
|
if field in pyObject:
|
|
asn1Value[field] = decodeFun(pyObject[field], componentsTypes[field].asn1Object, **options)
|
|
|
|
return asn1Value
|
|
|
|
|
|
class SequenceOfOrSetOfDecoder(object):
|
|
def __call__(self, pyObject, asn1Spec, decodeFun=None, **options):
|
|
asn1Value = asn1Spec.clone()
|
|
|
|
for pyValue in pyObject:
|
|
asn1Value.append(decodeFun(pyValue, asn1Spec.componentType), **options)
|
|
|
|
return asn1Value
|
|
|
|
|
|
class ChoiceDecoder(object):
|
|
def __call__(self, pyObject, asn1Spec, decodeFun=None, **options):
|
|
asn1Value = asn1Spec.clone()
|
|
|
|
componentsTypes = asn1Spec.componentType
|
|
|
|
for field in pyObject:
|
|
if field in componentsTypes:
|
|
asn1Value[field] = decodeFun(pyObject[field], componentsTypes[field].asn1Object, **options)
|
|
break
|
|
|
|
return asn1Value
|
|
|
|
|
|
tagMap = {
|
|
univ.Integer.tagSet: AbstractScalarDecoder(),
|
|
univ.Boolean.tagSet: AbstractScalarDecoder(),
|
|
univ.BitString.tagSet: BitStringDecoder(),
|
|
univ.OctetString.tagSet: AbstractScalarDecoder(),
|
|
univ.Null.tagSet: AbstractScalarDecoder(),
|
|
univ.ObjectIdentifier.tagSet: AbstractScalarDecoder(),
|
|
univ.Enumerated.tagSet: AbstractScalarDecoder(),
|
|
univ.Real.tagSet: AbstractScalarDecoder(),
|
|
univ.Sequence.tagSet: SequenceOrSetDecoder(), # conflicts with SequenceOf
|
|
univ.Set.tagSet: SequenceOrSetDecoder(), # conflicts with SetOf
|
|
univ.Choice.tagSet: ChoiceDecoder(), # conflicts with Any
|
|
# character string types
|
|
char.UTF8String.tagSet: AbstractScalarDecoder(),
|
|
char.NumericString.tagSet: AbstractScalarDecoder(),
|
|
char.PrintableString.tagSet: AbstractScalarDecoder(),
|
|
char.TeletexString.tagSet: AbstractScalarDecoder(),
|
|
char.VideotexString.tagSet: AbstractScalarDecoder(),
|
|
char.IA5String.tagSet: AbstractScalarDecoder(),
|
|
char.GraphicString.tagSet: AbstractScalarDecoder(),
|
|
char.VisibleString.tagSet: AbstractScalarDecoder(),
|
|
char.GeneralString.tagSet: AbstractScalarDecoder(),
|
|
char.UniversalString.tagSet: AbstractScalarDecoder(),
|
|
char.BMPString.tagSet: AbstractScalarDecoder(),
|
|
# useful types
|
|
useful.ObjectDescriptor.tagSet: AbstractScalarDecoder(),
|
|
useful.GeneralizedTime.tagSet: AbstractScalarDecoder(),
|
|
useful.UTCTime.tagSet: AbstractScalarDecoder()
|
|
}
|
|
|
|
# Put in ambiguous & non-ambiguous types for faster codec lookup
|
|
typeMap = {
|
|
univ.Integer.typeId: AbstractScalarDecoder(),
|
|
univ.Boolean.typeId: AbstractScalarDecoder(),
|
|
univ.BitString.typeId: BitStringDecoder(),
|
|
univ.OctetString.typeId: AbstractScalarDecoder(),
|
|
univ.Null.typeId: AbstractScalarDecoder(),
|
|
univ.ObjectIdentifier.typeId: AbstractScalarDecoder(),
|
|
univ.Enumerated.typeId: AbstractScalarDecoder(),
|
|
univ.Real.typeId: AbstractScalarDecoder(),
|
|
# ambiguous base types
|
|
univ.Set.typeId: SequenceOrSetDecoder(),
|
|
univ.SetOf.typeId: SequenceOfOrSetOfDecoder(),
|
|
univ.Sequence.typeId: SequenceOrSetDecoder(),
|
|
univ.SequenceOf.typeId: SequenceOfOrSetOfDecoder(),
|
|
univ.Choice.typeId: ChoiceDecoder(),
|
|
univ.Any.typeId: AbstractScalarDecoder(),
|
|
# character string types
|
|
char.UTF8String.typeId: AbstractScalarDecoder(),
|
|
char.NumericString.typeId: AbstractScalarDecoder(),
|
|
char.PrintableString.typeId: AbstractScalarDecoder(),
|
|
char.TeletexString.typeId: AbstractScalarDecoder(),
|
|
char.VideotexString.typeId: AbstractScalarDecoder(),
|
|
char.IA5String.typeId: AbstractScalarDecoder(),
|
|
char.GraphicString.typeId: AbstractScalarDecoder(),
|
|
char.VisibleString.typeId: AbstractScalarDecoder(),
|
|
char.GeneralString.typeId: AbstractScalarDecoder(),
|
|
char.UniversalString.typeId: AbstractScalarDecoder(),
|
|
char.BMPString.typeId: AbstractScalarDecoder(),
|
|
# useful types
|
|
useful.ObjectDescriptor.typeId: AbstractScalarDecoder(),
|
|
useful.GeneralizedTime.typeId: AbstractScalarDecoder(),
|
|
useful.UTCTime.typeId: AbstractScalarDecoder()
|
|
}
|
|
|
|
|
|
class Decoder(object):
|
|
|
|
# noinspection PyDefaultArgument
|
|
def __init__(self, tagMap, typeMap):
|
|
self.__tagMap = tagMap
|
|
self.__typeMap = typeMap
|
|
|
|
def __call__(self, pyObject, asn1Spec, **options):
|
|
if debug.logger & debug.flagDecoder:
|
|
logger = debug.logger
|
|
else:
|
|
logger = None
|
|
if logger:
|
|
debug.scope.push(type(pyObject).__name__)
|
|
logger('decoder called at scope %s, working with type %s' % (debug.scope, type(pyObject).__name__))
|
|
|
|
if asn1Spec is None or not isinstance(asn1Spec, base.Asn1Item):
|
|
raise error.PyAsn1Error('asn1Spec is not valid (should be an instance of an ASN.1 Item, not %s)' % asn1Spec.__class__.__name__)
|
|
|
|
try:
|
|
valueDecoder = self.__typeMap[asn1Spec.typeId]
|
|
|
|
except KeyError:
|
|
# use base type for codec lookup to recover untagged types
|
|
baseTagSet = tag.TagSet(asn1Spec.tagSet.baseTag, asn1Spec.tagSet.baseTag)
|
|
|
|
try:
|
|
valueDecoder = self.__tagMap[baseTagSet]
|
|
except KeyError:
|
|
raise error.PyAsn1Error('Unknown ASN.1 tag %s' % asn1Spec.tagSet)
|
|
|
|
if logger:
|
|
logger('calling decoder %s on Python type %s <%s>' % (type(valueDecoder).__name__, type(pyObject).__name__, repr(pyObject)))
|
|
|
|
value = valueDecoder(pyObject, asn1Spec, self, **options)
|
|
|
|
if logger:
|
|
logger('decoder %s produced ASN.1 type %s <%s>' % (type(valueDecoder).__name__, type(value).__name__, repr(value)))
|
|
debug.scope.pop()
|
|
|
|
return value
|
|
|
|
|
|
#: Turns Python objects of built-in types into ASN.1 objects.
|
|
#:
|
|
#: Takes Python objects of built-in types and turns them into a tree of
|
|
#: ASN.1 objects (e.g. :py:class:`~pyasn1.type.base.PyAsn1Item` derivative) which
|
|
#: may be a scalar or an arbitrary nested structure.
|
|
#:
|
|
#: Parameters
|
|
#: ----------
|
|
#: pyObject: :py:class:`object`
|
|
#: A scalar or nested Python objects
|
|
#:
|
|
#: Keyword Args
|
|
#: ------------
|
|
#: asn1Spec: any pyasn1 type object e.g. :py:class:`~pyasn1.type.base.PyAsn1Item` derivative
|
|
#: A pyasn1 type object to act as a template guiding the decoder. It is required
|
|
#: for successful interpretation of Python objects mapping into their ASN.1
|
|
#: representations.
|
|
#:
|
|
#: Returns
|
|
#: -------
|
|
#: : :py:class:`~pyasn1.type.base.PyAsn1Item` derivative
|
|
#: A scalar or constructed pyasn1 object
|
|
#:
|
|
#: Raises
|
|
#: ------
|
|
#: :py:class:`~pyasn1.error.PyAsn1Error`
|
|
#: On decoding errors
|
|
#:
|
|
#: Examples
|
|
#: --------
|
|
#: Decode native Python object into ASN.1 objects with ASN.1 schema
|
|
#:
|
|
#: .. code-block:: pycon
|
|
#:
|
|
#: >>> seq = SequenceOf(componentType=Integer())
|
|
#: >>> s, _ = decode([1, 2, 3], asn1Spec=seq)
|
|
#: >>> str(s)
|
|
#: SequenceOf:
|
|
#: 1 2 3
|
|
#:
|
|
decode = Decoder(tagMap, typeMap)
|