diff options
Diffstat (limited to '.venv/lib/python3.12/site-packages/pyasn1/codec/native')
3 files changed, 530 insertions, 0 deletions
diff --git a/.venv/lib/python3.12/site-packages/pyasn1/codec/native/__init__.py b/.venv/lib/python3.12/site-packages/pyasn1/codec/native/__init__.py new file mode 100644 index 00000000..8c3066b2 --- /dev/null +++ b/.venv/lib/python3.12/site-packages/pyasn1/codec/native/__init__.py @@ -0,0 +1 @@ +# This file is necessary to make this directory a package. diff --git a/.venv/lib/python3.12/site-packages/pyasn1/codec/native/decoder.py b/.venv/lib/python3.12/site-packages/pyasn1/codec/native/decoder.py new file mode 100644 index 00000000..9ac01ff6 --- /dev/null +++ b/.venv/lib/python3.12/site-packages/pyasn1/codec/native/decoder.py @@ -0,0 +1,244 @@ +# +# This file is part of pyasn1 software. +# +# Copyright (c) 2005-2020, Ilya Etingof <etingof@gmail.com> +# License: https://pyasn1.readthedocs.io/en/latest/license.html +# +import warnings + +from pyasn1 import debug +from pyasn1 import error +from pyasn1.compat import _MISSING +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'] + +LOG = debug.registerLoggee(__name__, flags=debug.DEBUG_DECODER) + + +class AbstractScalarPayloadDecoder(object): + def __call__(self, pyObject, asn1Spec, decodeFun=None, **options): + return asn1Spec.clone(pyObject) + + +class BitStringPayloadDecoder(AbstractScalarPayloadDecoder): + def __call__(self, pyObject, asn1Spec, decodeFun=None, **options): + return asn1Spec.clone(univ.BitString.fromBinaryString(pyObject)) + + +class SequenceOrSetPayloadDecoder(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 SequenceOfOrSetOfPayloadDecoder(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 ChoicePayloadDecoder(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 + + +TAG_MAP = { + univ.Integer.tagSet: AbstractScalarPayloadDecoder(), + univ.Boolean.tagSet: AbstractScalarPayloadDecoder(), + univ.BitString.tagSet: BitStringPayloadDecoder(), + univ.OctetString.tagSet: AbstractScalarPayloadDecoder(), + univ.Null.tagSet: AbstractScalarPayloadDecoder(), + univ.ObjectIdentifier.tagSet: AbstractScalarPayloadDecoder(), + univ.RelativeOID.tagSet: AbstractScalarPayloadDecoder(), + univ.Enumerated.tagSet: AbstractScalarPayloadDecoder(), + univ.Real.tagSet: AbstractScalarPayloadDecoder(), + univ.Sequence.tagSet: SequenceOrSetPayloadDecoder(), # conflicts with SequenceOf + univ.Set.tagSet: SequenceOrSetPayloadDecoder(), # conflicts with SetOf + univ.Choice.tagSet: ChoicePayloadDecoder(), # conflicts with Any + # character string types + char.UTF8String.tagSet: AbstractScalarPayloadDecoder(), + char.NumericString.tagSet: AbstractScalarPayloadDecoder(), + char.PrintableString.tagSet: AbstractScalarPayloadDecoder(), + char.TeletexString.tagSet: AbstractScalarPayloadDecoder(), + char.VideotexString.tagSet: AbstractScalarPayloadDecoder(), + char.IA5String.tagSet: AbstractScalarPayloadDecoder(), + char.GraphicString.tagSet: AbstractScalarPayloadDecoder(), + char.VisibleString.tagSet: AbstractScalarPayloadDecoder(), + char.GeneralString.tagSet: AbstractScalarPayloadDecoder(), + char.UniversalString.tagSet: AbstractScalarPayloadDecoder(), + char.BMPString.tagSet: AbstractScalarPayloadDecoder(), + # useful types + useful.ObjectDescriptor.tagSet: AbstractScalarPayloadDecoder(), + useful.GeneralizedTime.tagSet: AbstractScalarPayloadDecoder(), + useful.UTCTime.tagSet: AbstractScalarPayloadDecoder() +} + +# Put in ambiguous & non-ambiguous types for faster codec lookup +TYPE_MAP = { + univ.Integer.typeId: AbstractScalarPayloadDecoder(), + univ.Boolean.typeId: AbstractScalarPayloadDecoder(), + univ.BitString.typeId: BitStringPayloadDecoder(), + univ.OctetString.typeId: AbstractScalarPayloadDecoder(), + univ.Null.typeId: AbstractScalarPayloadDecoder(), + univ.ObjectIdentifier.typeId: AbstractScalarPayloadDecoder(), + univ.RelativeOID.typeId: AbstractScalarPayloadDecoder(), + univ.Enumerated.typeId: AbstractScalarPayloadDecoder(), + univ.Real.typeId: AbstractScalarPayloadDecoder(), + # ambiguous base types + univ.Set.typeId: SequenceOrSetPayloadDecoder(), + univ.SetOf.typeId: SequenceOfOrSetOfPayloadDecoder(), + univ.Sequence.typeId: SequenceOrSetPayloadDecoder(), + univ.SequenceOf.typeId: SequenceOfOrSetOfPayloadDecoder(), + univ.Choice.typeId: ChoicePayloadDecoder(), + univ.Any.typeId: AbstractScalarPayloadDecoder(), + # character string types + char.UTF8String.typeId: AbstractScalarPayloadDecoder(), + char.NumericString.typeId: AbstractScalarPayloadDecoder(), + char.PrintableString.typeId: AbstractScalarPayloadDecoder(), + char.TeletexString.typeId: AbstractScalarPayloadDecoder(), + char.VideotexString.typeId: AbstractScalarPayloadDecoder(), + char.IA5String.typeId: AbstractScalarPayloadDecoder(), + char.GraphicString.typeId: AbstractScalarPayloadDecoder(), + char.VisibleString.typeId: AbstractScalarPayloadDecoder(), + char.GeneralString.typeId: AbstractScalarPayloadDecoder(), + char.UniversalString.typeId: AbstractScalarPayloadDecoder(), + char.BMPString.typeId: AbstractScalarPayloadDecoder(), + # useful types + useful.ObjectDescriptor.typeId: AbstractScalarPayloadDecoder(), + useful.GeneralizedTime.typeId: AbstractScalarPayloadDecoder(), + useful.UTCTime.typeId: AbstractScalarPayloadDecoder() +} + + +class SingleItemDecoder(object): + + TAG_MAP = TAG_MAP + TYPE_MAP = TYPE_MAP + + def __init__(self, tagMap=_MISSING, typeMap=_MISSING, **ignored): + self._tagMap = tagMap if tagMap is not _MISSING else self.TAG_MAP + self._typeMap = typeMap if typeMap is not _MISSING else self.TYPE_MAP + + def __call__(self, pyObject, asn1Spec, **options): + + if LOG: + debug.scope.push(type(pyObject).__name__) + LOG('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 LOG: + LOG('calling decoder %s on Python type %s ' + '<%s>' % (type(valueDecoder).__name__, + type(pyObject).__name__, repr(pyObject))) + + value = valueDecoder(pyObject, asn1Spec, self, **options) + + if LOG: + LOG('decoder %s produced ASN.1 type %s ' + '<%s>' % (type(valueDecoder).__name__, + type(value).__name__, repr(value))) + debug.scope.pop() + + return value + + +class Decoder(object): + SINGLE_ITEM_DECODER = SingleItemDecoder + + def __init__(self, **options): + self._singleItemDecoder = self.SINGLE_ITEM_DECODER(**options) + + def __call__(self, pyObject, asn1Spec=None, **kwargs): + return self._singleItemDecoder(pyObject, asn1Spec=asn1Spec, **kwargs) + + +#: 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 +#: ------ +#: ~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() + +def __getattr__(attr: str): + if newAttr := {"tagMap": "TAG_MAP", "typeMap": "TYPE_MAP"}.get(attr): + warnings.warn(f"{attr} is deprecated. Please use {newAttr} instead.", DeprecationWarning) + return globals()[newAttr] + raise AttributeError(attr) diff --git a/.venv/lib/python3.12/site-packages/pyasn1/codec/native/encoder.py b/.venv/lib/python3.12/site-packages/pyasn1/codec/native/encoder.py new file mode 100644 index 00000000..6219ae45 --- /dev/null +++ b/.venv/lib/python3.12/site-packages/pyasn1/codec/native/encoder.py @@ -0,0 +1,285 @@ +# +# This file is part of pyasn1 software. +# +# Copyright (c) 2005-2020, Ilya Etingof <etingof@gmail.com> +# License: https://pyasn1.readthedocs.io/en/latest/license.html +# +from collections import OrderedDict +import warnings + +from pyasn1 import debug +from pyasn1 import error +from pyasn1.compat import _MISSING +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__ = ['encode'] + +LOG = debug.registerLoggee(__name__, flags=debug.DEBUG_ENCODER) + + +class AbstractItemEncoder(object): + def encode(self, value, encodeFun, **options): + raise error.PyAsn1Error('Not implemented') + + +class BooleanEncoder(AbstractItemEncoder): + def encode(self, value, encodeFun, **options): + return bool(value) + + +class IntegerEncoder(AbstractItemEncoder): + def encode(self, value, encodeFun, **options): + return int(value) + + +class BitStringEncoder(AbstractItemEncoder): + def encode(self, value, encodeFun, **options): + return str(value) + + +class OctetStringEncoder(AbstractItemEncoder): + def encode(self, value, encodeFun, **options): + return value.asOctets() + + +class TextStringEncoder(AbstractItemEncoder): + def encode(self, value, encodeFun, **options): + return str(value) + + +class NullEncoder(AbstractItemEncoder): + def encode(self, value, encodeFun, **options): + return None + + +class ObjectIdentifierEncoder(AbstractItemEncoder): + def encode(self, value, encodeFun, **options): + return str(value) + + +class RelativeOIDEncoder(AbstractItemEncoder): + def encode(self, value, encodeFun, **options): + return str(value) + + +class RealEncoder(AbstractItemEncoder): + def encode(self, value, encodeFun, **options): + return float(value) + + +class SetEncoder(AbstractItemEncoder): + protoDict = dict + + def encode(self, value, encodeFun, **options): + inconsistency = value.isInconsistent + if inconsistency: + raise error.PyAsn1Error( + f"ASN.1 object {value.__class__.__name__} is inconsistent") + + namedTypes = value.componentType + substrate = self.protoDict() + + for idx, (key, subValue) in enumerate(value.items()): + if namedTypes and namedTypes[idx].isOptional and not value[idx].isValue: + continue + substrate[key] = encodeFun(subValue, **options) + return substrate + + +class SequenceEncoder(SetEncoder): + protoDict = OrderedDict + + +class SequenceOfEncoder(AbstractItemEncoder): + def encode(self, value, encodeFun, **options): + inconsistency = value.isInconsistent + if inconsistency: + raise error.PyAsn1Error( + f"ASN.1 object {value.__class__.__name__} is inconsistent") + return [encodeFun(x, **options) for x in value] + + +class ChoiceEncoder(SequenceEncoder): + pass + + +class AnyEncoder(AbstractItemEncoder): + def encode(self, value, encodeFun, **options): + return value.asOctets() + + +TAG_MAP = { + univ.Boolean.tagSet: BooleanEncoder(), + univ.Integer.tagSet: IntegerEncoder(), + univ.BitString.tagSet: BitStringEncoder(), + univ.OctetString.tagSet: OctetStringEncoder(), + univ.Null.tagSet: NullEncoder(), + univ.ObjectIdentifier.tagSet: ObjectIdentifierEncoder(), + univ.RelativeOID.tagSet: RelativeOIDEncoder(), + univ.Enumerated.tagSet: IntegerEncoder(), + univ.Real.tagSet: RealEncoder(), + # Sequence & Set have same tags as SequenceOf & SetOf + univ.SequenceOf.tagSet: SequenceOfEncoder(), + univ.SetOf.tagSet: SequenceOfEncoder(), + univ.Choice.tagSet: ChoiceEncoder(), + # character string types + char.UTF8String.tagSet: TextStringEncoder(), + char.NumericString.tagSet: TextStringEncoder(), + char.PrintableString.tagSet: TextStringEncoder(), + char.TeletexString.tagSet: TextStringEncoder(), + char.VideotexString.tagSet: TextStringEncoder(), + char.IA5String.tagSet: TextStringEncoder(), + char.GraphicString.tagSet: TextStringEncoder(), + char.VisibleString.tagSet: TextStringEncoder(), + char.GeneralString.tagSet: TextStringEncoder(), + char.UniversalString.tagSet: TextStringEncoder(), + char.BMPString.tagSet: TextStringEncoder(), + # useful types + useful.ObjectDescriptor.tagSet: OctetStringEncoder(), + useful.GeneralizedTime.tagSet: OctetStringEncoder(), + useful.UTCTime.tagSet: OctetStringEncoder() +} + +# Put in ambiguous & non-ambiguous types for faster codec lookup +TYPE_MAP = { + univ.Boolean.typeId: BooleanEncoder(), + univ.Integer.typeId: IntegerEncoder(), + univ.BitString.typeId: BitStringEncoder(), + univ.OctetString.typeId: OctetStringEncoder(), + univ.Null.typeId: NullEncoder(), + univ.ObjectIdentifier.typeId: ObjectIdentifierEncoder(), + univ.RelativeOID.typeId: RelativeOIDEncoder(), + univ.Enumerated.typeId: IntegerEncoder(), + univ.Real.typeId: RealEncoder(), + # Sequence & Set have same tags as SequenceOf & SetOf + univ.Set.typeId: SetEncoder(), + univ.SetOf.typeId: SequenceOfEncoder(), + univ.Sequence.typeId: SequenceEncoder(), + univ.SequenceOf.typeId: SequenceOfEncoder(), + univ.Choice.typeId: ChoiceEncoder(), + univ.Any.typeId: AnyEncoder(), + # character string types + char.UTF8String.typeId: OctetStringEncoder(), + char.NumericString.typeId: OctetStringEncoder(), + char.PrintableString.typeId: OctetStringEncoder(), + char.TeletexString.typeId: OctetStringEncoder(), + char.VideotexString.typeId: OctetStringEncoder(), + char.IA5String.typeId: OctetStringEncoder(), + char.GraphicString.typeId: OctetStringEncoder(), + char.VisibleString.typeId: OctetStringEncoder(), + char.GeneralString.typeId: OctetStringEncoder(), + char.UniversalString.typeId: OctetStringEncoder(), + char.BMPString.typeId: OctetStringEncoder(), + # useful types + useful.ObjectDescriptor.typeId: OctetStringEncoder(), + useful.GeneralizedTime.typeId: OctetStringEncoder(), + useful.UTCTime.typeId: OctetStringEncoder() +} + + +class SingleItemEncoder(object): + + TAG_MAP = TAG_MAP + TYPE_MAP = TYPE_MAP + + def __init__(self, tagMap=_MISSING, typeMap=_MISSING, **ignored): + self._tagMap = tagMap if tagMap is not _MISSING else self.TAG_MAP + self._typeMap = typeMap if typeMap is not _MISSING else self.TYPE_MAP + + def __call__(self, value, **options): + if not isinstance(value, base.Asn1Item): + raise error.PyAsn1Error( + 'value is not valid (should be an instance of an ASN.1 Item)') + + if LOG: + debug.scope.push(type(value).__name__) + LOG('encoder called for type %s ' + '<%s>' % (type(value).__name__, value.prettyPrint())) + + tagSet = value.tagSet + + try: + concreteEncoder = self._typeMap[value.typeId] + + except KeyError: + # use base type for codec lookup to recover untagged types + baseTagSet = tag.TagSet( + value.tagSet.baseTag, value.tagSet.baseTag) + + try: + concreteEncoder = self._tagMap[baseTagSet] + + except KeyError: + raise error.PyAsn1Error('No encoder for %s' % (value,)) + + if LOG: + LOG('using value codec %s chosen by ' + '%s' % (concreteEncoder.__class__.__name__, tagSet)) + + pyObject = concreteEncoder.encode(value, self, **options) + + if LOG: + LOG('encoder %s produced: ' + '%s' % (type(concreteEncoder).__name__, repr(pyObject))) + debug.scope.pop() + + return pyObject + + +class Encoder(object): + SINGLE_ITEM_ENCODER = SingleItemEncoder + + def __init__(self, **options): + self._singleItemEncoder = self.SINGLE_ITEM_ENCODER(**options) + + def __call__(self, pyObject, asn1Spec=None, **options): + return self._singleItemEncoder( + pyObject, asn1Spec=asn1Spec, **options) + + +#: Turns ASN.1 object into a Python built-in type object(s). +#: +#: Takes any ASN.1 object (e.g. :py:class:`~pyasn1.type.base.PyAsn1Item` derivative) +#: walks all its components recursively and produces a Python built-in type or a tree +#: of those. +#: +#: One exception is that instead of :py:class:`dict`, the :py:class:`OrderedDict` +#: is used to preserve ordering of the components in ASN.1 SEQUENCE. +#: +#: Parameters +#: ---------- +# asn1Value: any pyasn1 object (e.g. :py:class:`~pyasn1.type.base.PyAsn1Item` derivative) +#: pyasn1 object to encode (or a tree of them) +#: +#: Returns +#: ------- +#: : :py:class:`object` +#: Python built-in type instance (or a tree of them) +#: +#: Raises +#: ------ +#: ~pyasn1.error.PyAsn1Error +#: On encoding errors +#: +#: Examples +#: -------- +#: Encode ASN.1 value object into native Python types +#: +#: .. code-block:: pycon +#: +#: >>> seq = SequenceOf(componentType=Integer()) +#: >>> seq.extend([1, 2, 3]) +#: >>> encode(seq) +#: [1, 2, 3] +#: +encode = SingleItemEncoder() + +def __getattr__(attr: str): + if newAttr := {"tagMap": "TAG_MAP", "typeMap": "TYPE_MAP"}.get(attr): + warnings.warn(f"{attr} is deprecated. Please use {newAttr} instead.", DeprecationWarning) + return globals()[newAttr] + raise AttributeError(attr) |