codekingpro/portable-devtools

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.html6#7import warnings8 9from pyasn1 import error10from pyasn1.codec.ber import encoder11from pyasn1.type import univ12from pyasn1.type import useful13 14__all__ = ['Encoder', 'encode']15 16 17class BooleanEncoder(encoder.IntegerEncoder):18    def encodeValue(self, value, asn1Spec, encodeFun, **options):19        if value == 0:20            substrate = (0,)21        else:22            substrate = (255,)23        return substrate, False, False24 25 26class RealEncoder(encoder.RealEncoder):27    def _chooseEncBase(self, value):28        m, b, e = value29        return self._dropFloatingPoint(m, b, e)30 31 32# specialized GeneralStringEncoder here33 34class TimeEncoderMixIn(object):35    Z_CHAR = ord('Z')36    PLUS_CHAR = ord('+')37    MINUS_CHAR = ord('-')38    COMMA_CHAR = ord(',')39    DOT_CHAR = ord('.')40    ZERO_CHAR = ord('0')41 42    MIN_LENGTH = 1243    MAX_LENGTH = 1944 45    def encodeValue(self, value, asn1Spec, encodeFun, **options):46        # CER encoding constraints:47        # - minutes are mandatory, seconds are optional48        # - sub-seconds must NOT be zero / no meaningless zeros49        # - no hanging fraction dot50        # - time in UTC (Z)51        # - only dot is allowed for fractions52 53        if asn1Spec is not None:54            value = asn1Spec.clone(value)55 56        numbers = value.asNumbers()57 58        if self.PLUS_CHAR in numbers or self.MINUS_CHAR in numbers:59            raise error.PyAsn1Error('Must be UTC time: %r' % value)60 61        if numbers[-1] != self.Z_CHAR:62            raise error.PyAsn1Error('Missing "Z" time zone specifier: %r' % value)63 64        if self.COMMA_CHAR in numbers:65            raise error.PyAsn1Error('Comma in fractions disallowed: %r' % value)66 67        if self.DOT_CHAR in numbers:68 69            isModified = False70 71            numbers = list(numbers)72 73            searchIndex = min(numbers.index(self.DOT_CHAR) + 4, len(numbers) - 1)74 75            while numbers[searchIndex] != self.DOT_CHAR:76                if numbers[searchIndex] == self.ZERO_CHAR:77                    del numbers[searchIndex]78                    isModified = True79 80                searchIndex -= 181 82            searchIndex += 183 84            if searchIndex < len(numbers):85                if numbers[searchIndex] == self.Z_CHAR:86                    # drop hanging comma87                    del numbers[searchIndex - 1]88                    isModified = True89 90            if isModified:91                value = value.clone(numbers)92 93        if not self.MIN_LENGTH < len(numbers) < self.MAX_LENGTH:94            raise error.PyAsn1Error('Length constraint violated: %r' % value)95 96        options.update(maxChunkSize=1000)97 98        return encoder.OctetStringEncoder.encodeValue(99            self, value, asn1Spec, encodeFun, **options100        )101 102 103class GeneralizedTimeEncoder(TimeEncoderMixIn, encoder.OctetStringEncoder):104    MIN_LENGTH = 12105    MAX_LENGTH = 20106 107 108class UTCTimeEncoder(TimeEncoderMixIn, encoder.OctetStringEncoder):109    MIN_LENGTH = 10110    MAX_LENGTH = 14111 112 113class SetOfEncoder(encoder.SequenceOfEncoder):114    def encodeValue(self, value, asn1Spec, encodeFun, **options):115        chunks = self._encodeComponents(116            value, asn1Spec, encodeFun, **options)117 118        # sort by serialised and padded components119        if len(chunks) > 1:120            zero = b'\x00'121            maxLen = max(map(len, chunks))122            paddedChunks = [123                (x.ljust(maxLen, zero), x) for x in chunks124            ]125            paddedChunks.sort(key=lambda x: x[0])126 127            chunks = [x[1] for x in paddedChunks]128 129        return b''.join(chunks), True, True130 131 132class SequenceOfEncoder(encoder.SequenceOfEncoder):133    def encodeValue(self, value, asn1Spec, encodeFun, **options):134 135        if options.get('ifNotEmpty', False) and not len(value):136            return b'', True, True137 138        chunks = self._encodeComponents(139            value, asn1Spec, encodeFun, **options)140 141        return b''.join(chunks), True, True142 143 144class SetEncoder(encoder.SequenceEncoder):145    @staticmethod146    def _componentSortKey(componentAndType):147        """Sort SET components by tag148 149        Sort regardless of the Choice value (static sort)150        """151        component, asn1Spec = componentAndType152 153        if asn1Spec is None:154            asn1Spec = component155 156        if asn1Spec.typeId == univ.Choice.typeId and not asn1Spec.tagSet:157            if asn1Spec.tagSet:158                return asn1Spec.tagSet159            else:160                return asn1Spec.componentType.minTagSet161        else:162            return asn1Spec.tagSet163 164    def encodeValue(self, value, asn1Spec, encodeFun, **options):165 166        substrate = b''167 168        comps = []169        compsMap = {}170 171        if asn1Spec is None:172            # instance of ASN.1 schema173            inconsistency = value.isInconsistent174            if inconsistency:175                raise error.PyAsn1Error(176                    f"ASN.1 object {value.__class__.__name__} is inconsistent")177 178            namedTypes = value.componentType179 180            for idx, component in enumerate(value.values()):181                if namedTypes:182                    namedType = namedTypes[idx]183 184                    if namedType.isOptional and not component.isValue:185                            continue186 187                    if namedType.isDefaulted and component == namedType.asn1Object:188                            continue189 190                    compsMap[id(component)] = namedType191 192                else:193                    compsMap[id(component)] = None194 195                comps.append((component, asn1Spec))196 197        else:198            # bare Python value + ASN.1 schema199            for idx, namedType in enumerate(asn1Spec.componentType.namedTypes):200 201                try:202                    component = value[namedType.name]203 204                except KeyError:205                    raise error.PyAsn1Error('Component name "%s" not found in %r' % (namedType.name, value))206 207                if namedType.isOptional and namedType.name not in value:208                    continue209 210                if namedType.isDefaulted and component == namedType.asn1Object:211                    continue212 213                compsMap[id(component)] = namedType214                comps.append((component, asn1Spec[idx]))215 216        for comp, compType in sorted(comps, key=self._componentSortKey):217            namedType = compsMap[id(comp)]218 219            if namedType:220                options.update(ifNotEmpty=namedType.isOptional)221 222            chunk = encodeFun(comp, compType, **options)223 224            # wrap open type blob if needed225            if namedType and namedType.openType:226                wrapType = namedType.asn1Object227                if wrapType.tagSet and not wrapType.isSameTypeWith(comp):228                    chunk = encodeFun(chunk, wrapType, **options)229 230            substrate += chunk231 232        return substrate, True, True233 234 235class SequenceEncoder(encoder.SequenceEncoder):236    omitEmptyOptionals = True237 238 239TAG_MAP = encoder.TAG_MAP.copy()240 241TAG_MAP.update({242    univ.Boolean.tagSet: BooleanEncoder(),243    univ.Real.tagSet: RealEncoder(),244    useful.GeneralizedTime.tagSet: GeneralizedTimeEncoder(),245    useful.UTCTime.tagSet: UTCTimeEncoder(),246    # Sequence & Set have same tags as SequenceOf & SetOf247    univ.SetOf.tagSet: SetOfEncoder(),248    univ.Sequence.typeId: SequenceEncoder()249})250 251TYPE_MAP = encoder.TYPE_MAP.copy()252 253TYPE_MAP.update({254    univ.Boolean.typeId: BooleanEncoder(),255    univ.Real.typeId: RealEncoder(),256    useful.GeneralizedTime.typeId: GeneralizedTimeEncoder(),257    useful.UTCTime.typeId: UTCTimeEncoder(),258    # Sequence & Set have same tags as SequenceOf & SetOf259    univ.Set.typeId: SetEncoder(),260    univ.SetOf.typeId: SetOfEncoder(),261    univ.Sequence.typeId: SequenceEncoder(),262    univ.SequenceOf.typeId: SequenceOfEncoder()263})264 265 266class SingleItemEncoder(encoder.SingleItemEncoder):267    fixedDefLengthMode = False268    fixedChunkSize = 1000269 270    TAG_MAP = TAG_MAP271    TYPE_MAP = TYPE_MAP272 273 274class Encoder(encoder.Encoder):275    SINGLE_ITEM_ENCODER = SingleItemEncoder276 277 278#: Turns ASN.1 object into CER octet stream.279#:280#: Takes any ASN.1 object (e.g. :py:class:`~pyasn1.type.base.PyAsn1Item` derivative)281#: walks all its components recursively and produces a CER octet stream.282#:283#: Parameters284#: ----------285#: value: either a Python or pyasn1 object (e.g. :py:class:`~pyasn1.type.base.PyAsn1Item` derivative)286#:     A Python or pyasn1 object to encode. If Python object is given, `asnSpec`287#:     parameter is required to guide the encoding process.288#:289#: Keyword Args290#: ------------291#: asn1Spec:292#:     Optional ASN.1 schema or value object e.g. :py:class:`~pyasn1.type.base.PyAsn1Item` derivative293#:294#: Returns295#: -------296#: : :py:class:`bytes`297#:     Given ASN.1 object encoded into BER octet-stream298#:299#: Raises300#: ------301#: ~pyasn1.error.PyAsn1Error302#:     On encoding errors303#:304#: Examples305#: --------306#: Encode Python value into CER with ASN.1 schema307#:308#: .. code-block:: pycon309#:310#:    >>> seq = SequenceOf(componentType=Integer())311#:    >>> encode([1, 2, 3], asn1Spec=seq)312#:    b'0\x80\x02\x01\x01\x02\x01\x02\x02\x01\x03\x00\x00'313#:314#: Encode ASN.1 value object into CER315#:316#: .. code-block:: pycon317#:318#:    >>> seq = SequenceOf(componentType=Integer())319#:    >>> seq.extend([1, 2, 3])320#:    >>> encode(seq)321#:    b'0\x80\x02\x01\x01\x02\x01\x02\x02\x01\x03\x00\x00'322#:323encode = Encoder()324 325# EncoderFactory queries class instance and builds a map of tags -> encoders326 327def __getattr__(attr: str):328    if newAttr := {"tagMap": "TAG_MAP", "typeMap": "TYPE_MAP"}.get(attr):329        warnings.warn(f"{attr} is deprecated. Please use {newAttr} instead.", DeprecationWarning, stacklevel=2)330        return globals()[newAttr]331    raise AttributeError(attr)332