X-Git-Url: http://www.git.cypherpunks.ru/?a=blobdiff_plain;f=pyderasn.py;h=f4f38a2c7121329bbd4ba797a55b87a2cf09e56a;hb=de2299f02a411f3b805058afa84118cf361c99c8;hp=1c7f3ed94603bda04df4aa52c38e8f2d293c0055;hpb=afc0f9f65430bed928619c783373ae3c6a82be1b;p=pyderasn.git diff --git a/pyderasn.py b/pyderasn.py index 1c7f3ed..f4f38a2 100755 --- a/pyderasn.py +++ b/pyderasn.py @@ -1,12 +1,11 @@ #!/usr/bin/env python # coding: utf-8 # PyDERASN -- Python ASN.1 DER/BER codec with abstract structures -# Copyright (C) 2017-2019 Sergey Matveev +# Copyright (C) 2017-2020 Sergey Matveev # # This program is free software: you can redistribute it and/or modify # it under the terms of the GNU Lesser General Public License as -# published by the Free Software Foundation, either version 3 of the -# License, or (at your option) any later version. +# published by the Free Software Foundation, version 3 of the License. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of @@ -14,8 +13,7 @@ # GNU Lesser General Public License for more details. # # You should have received a copy of the GNU Lesser General Public -# License along with this program. If not, see -# . +# License along with this program. If not, see . """Python ASN.1 DER/BER codec with abstract structures This library allows you to marshal various structures in ASN.1 DER @@ -239,6 +237,105 @@ all object ``repr``. But it is easy to write custom formatters. >>> print(pprint(obj)) 0 [1,1, 2] INTEGER -12345 +.. _pprint_example: + +Example certificate:: + + >>> print(pprint(crt)) + 0 [1,3,1604] Certificate SEQUENCE + 4 [1,3,1453] . tbsCertificate: TBSCertificate SEQUENCE + 10-2 [1,1, 1] . . version: [0] EXPLICIT Version INTEGER v3 OPTIONAL + 13 [1,1, 3] . . serialNumber: CertificateSerialNumber INTEGER 61595 + 18 [1,1, 13] . . signature: AlgorithmIdentifier SEQUENCE + 20 [1,1, 9] . . . algorithm: OBJECT IDENTIFIER 1.2.840.113549.1.1.5 + 31 [0,0, 2] . . . parameters: [UNIV 5] ANY OPTIONAL + . . . . 05:00 + 33 [0,0, 278] . . issuer: Name CHOICE rdnSequence + 33 [1,3, 274] . . . rdnSequence: RDNSequence SEQUENCE OF + 37 [1,1, 11] . . . . 0: RelativeDistinguishedName SET OF + 39 [1,1, 9] . . . . . 0: AttributeTypeAndValue SEQUENCE + 41 [1,1, 3] . . . . . . type: AttributeType OBJECT IDENTIFIER 2.5.4.6 + 46 [0,0, 4] . . . . . . value: [UNIV 19] AttributeValue ANY + . . . . . . . 13:02:45:53 + [...] + 1461 [1,1, 13] . signatureAlgorithm: AlgorithmIdentifier SEQUENCE + 1463 [1,1, 9] . . algorithm: OBJECT IDENTIFIER 1.2.840.113549.1.1.5 + 1474 [0,0, 2] . . parameters: [UNIV 5] ANY OPTIONAL + . . . 05:00 + 1476 [1,2, 129] . signatureValue: BIT STRING 1024 bits + . . 68:EE:79:97:97:DD:3B:EF:16:6A:06:F2:14:9A:6E:CD + . . 9E:12:F7:AA:83:10:BD:D1:7C:98:FA:C7:AE:D4:0E:2C + [...] + + Trailing data: 0a + +Let's parse that output, human:: + + 10-2 [1,1, 1] . . version: [0] EXPLICIT Version INTEGER v3 OPTIONAL + ^ ^ ^ ^ ^ ^ ^ ^ ^ ^ ^ ^ + 0 1 2 3 4 5 6 7 8 9 10 11 + +:: + + 20 [1,1, 9] . . . algorithm: OBJECT IDENTIFIER 1.2.840.113549.1.1.5 + ^ ^ ^ ^ ^ ^ ^ ^ + 0 2 3 4 5 6 9 10 + +:: + + 33 [0,0, 278] . . issuer: Name CHOICE rdnSequence + ^ ^ ^ ^ ^ ^ ^ ^ ^ + 0 2 3 4 5 6 8 9 10 + +:: + + 52-2∞ B [1,1,1054]∞ . . . . eContent: [0] EXPLICIT BER OCTET STRING 1046 bytes + ^ ^ ^ ^ ^ + 12 13 14 9 10 + +:0: + Offset of the object, where its DER/BER encoding begins. + Pay attention that it does **not** include explicit tag. +:1: + If explicit tag exists, then this is its length (tag + encoded length). +:2: + Length of object's tag. For example CHOICE does not have its own tag, + so it is zero. +:3: + Length of encoded length. +:4: + Length of encoded value. +:5: + Visual indentation to show the depth of object in the hierarchy. +:6: + Object's name inside SEQUENCE/CHOICE. +:7: + If either IMPLICIT or EXPLICIT tag is set, then it will be shown + here. "IMPLICIT" is omitted. +:8: + Object's class name, if set. Omitted if it is just an ordinary simple + value (like with ``algorithm`` in example above). +:9: + Object's ASN.1 type. +:10: + Object's value, if set. Can consist of multiple words (like OCTET/BIT + STRINGs above). We see ``v3`` value in Version, because it is named. + ``rdnSequence`` is the choice of CHOICE type. +:11: + Possible other flags like OPTIONAL and DEFAULT, if value equals to the + default one, specified in the schema. +:12: + Shows does object contains any kind of BER encoded data (possibly + Sequence holding BER-encoded underlying value). +:13: + Only applicable to BER encoded data. Indefinite length encoding mark. +:14: + Only applicable to BER encoded data. If object has BER-specific + encoding, then ``BER`` will be shown. It does not depend on indefinite + length encoding. ``EOC``, ``BOOLEAN``, ``BIT STRING``, ``OCTET STRING`` + (and its derivatives), ``SET``, ``SET OF`` could be BERed. + + .. _definedby: DEFINED BY @@ -249,6 +346,8 @@ DEFINED BY some previously met ObjectIdentifier. This library provides ability to specify mapping between some OID and field that must be decoded with specific specification. +.. _defines: + defines kwarg _____________ @@ -322,15 +421,15 @@ value must be sequence of following tuples:: where ``decode_path`` is a tuple holding so-called decode path to the exact :py:class:`pyderasn.ObjectIdentifier` field you want to apply -``defines``, holding exactly the same value as accepted in its keyword -argument. +``defines``, holding exactly the same value as accepted in its +:ref:`keyword argument `. For example, again for CMS, you want to automatically decode ``SignedData`` and CMC's (:rfc:`5272`) ``PKIData`` and ``PKIResponse`` structures it may hold. Also, automatically decode ``controlSequence`` of ``PKIResponse``:: - content_info, tail = ContentInfo().decode(data, defines_by_path=( + content_info, tail = ContentInfo().decode(data, ctx={"defines_by_path": ( ( ("contentType",), ((("content",), {id_signedData: SignedData()}),), @@ -365,7 +464,7 @@ of ``PKIResponse``:: id_cmc_transactionId: TransactionId(), })), ), - )) + )}) Pay attention for :py:class:`pyderasn.DecodePathDefBy` and ``any``. First function is useful for path construction when some automatic @@ -418,6 +517,11 @@ lengths will be invalid in that case. This option should be used only for skipping some decode errors, just to see the decoded structure somehow. +Base Obj +-------- +.. autoclass:: pyderasn.Obj + :members: + Primitive types --------------- @@ -463,6 +567,11 @@ NumericString _____________ .. autoclass:: pyderasn.NumericString +PrintableString +_______________ +.. autoclass:: pyderasn.PrintableString + :members: __init__ + UTCTime _______ .. autoclass:: pyderasn.UTCTime @@ -523,10 +632,10 @@ Various .. autofunction:: pyderasn.tag_decode .. autofunction:: pyderasn.tag_ctxp .. autofunction:: pyderasn.tag_ctxc -.. autoclass:: pyderasn.Obj .. autoclass:: pyderasn.DecodeError :members: __init__ .. autoclass:: pyderasn.NotEnoughData +.. autoclass:: pyderasn.ExceedingData .. autoclass:: pyderasn.LenIndefForm .. autoclass:: pyderasn.TagMismatch .. autoclass:: pyderasn.InvalidLength @@ -547,6 +656,7 @@ from math import ceil from os import environ from string import ascii_letters from string import digits +from unicodedata import category as unicat from six import add_metaclass from six import binary_type @@ -555,6 +665,8 @@ from six import indexbytes from six import int2byte from six import integer_types from six import iterbytes +from six import iteritems +from six import itervalues from six import PY2 from six import string_types from six import text_type @@ -565,9 +677,10 @@ from six.moves import xrange as six_xrange try: from termcolor import colored except ImportError: # pragma: no cover - def colored(what, *args): + def colored(what, *args, **kwargs): return what +__version__ = "5.6" __all__ = ( "Any", @@ -579,6 +692,7 @@ __all__ = ( "DecodeError", "DecodePathDefBy", "Enumerated", + "ExceedingData", "GeneralizedTime", "GeneralString", "GraphicString", @@ -689,6 +803,18 @@ class NotEnoughData(DecodeError): pass +class ExceedingData(ASN1Error): + def __init__(self, nbytes): + super(ExceedingData, self).__init__() + self.nbytes = nbytes + + def __str__(self): + return "%d trailing bytes" % self.nbytes + + def __repr__(self): + return "%s(%s)" % (self.__class__.__name__, self) + + class LenIndefForm(DecodeError): pass @@ -914,9 +1040,9 @@ def len_decode(data): ######################################################################## class AutoAddSlots(type): - def __new__(mcs, name, bases, _dict): + def __new__(cls, name, bases, _dict): _dict["__slots__"] = _dict.get("__slots__", ()) - return type.__new__(mcs, name, bases, _dict) + return type.__new__(cls, name, bases, _dict) @add_metaclass(AutoAddSlots) @@ -990,10 +1116,14 @@ class Obj(object): @property def tlen(self): + """See :ref:`decoding` + """ return len(self.tag) @property def tlvlen(self): + """See :ref:`decoding` + """ return self.tlen + self.llen + self.vlen def __str__(self): # pragma: no cover @@ -1018,6 +1148,10 @@ class Obj(object): raise NotImplementedError() def encode(self): + """Encode the structure + + :returns: DER representation + """ raw = self._encode() if self._expl is None: return raw @@ -1045,6 +1179,8 @@ class Obj(object): determine if tag satisfies the scheme) :param _ctx_immutable: do we need to copy ``ctx`` before using it :returns: (Obj, remaining data) + + .. seealso:: :ref:`decoding` """ if ctx is None: ctx = {} @@ -1060,7 +1196,7 @@ class Obj(object): tag_only=tag_only, ) if tag_only: - return + return None obj, tail = result else: try: @@ -1098,7 +1234,7 @@ class Obj(object): tag_only=tag_only, ) if tag_only: # pragma: no cover - return + return None obj, tail = result eoc_expected, tail = tail[:EOC_LEN], tail[EOC_LEN:] if eoc_expected.tobytes() != EOC: @@ -1133,7 +1269,7 @@ class Obj(object): tag_only=tag_only, ) if tag_only: # pragma: no cover - return + return None obj, tail = result if obj.tlvlen < l and not ctx.get("allow_expl_oob", False): raise DecodeError( @@ -1144,42 +1280,80 @@ class Obj(object): ) return obj, (tail if leavemm else tail.tobytes()) + def decod(self, data, offset=0, decode_path=(), ctx=None): + """Decode the data, check that tail is empty + + :raises ExceedingData: if tail is not empty + + This is just a wrapper over :py:meth:`pyderasn.Obj.decode` + (decode without tail) that also checks that there is no + trailing data left. + """ + obj, tail = self.decode( + data, + offset=offset, + decode_path=decode_path, + ctx=ctx, + leavemm=True, + ) + if len(tail) > 0: + raise ExceedingData(len(tail)) + return obj + @property def expled(self): + """See :ref:`decoding` + """ return self._expl is not None @property def expl_tag(self): + """See :ref:`decoding` + """ return self._expl @property def expl_tlen(self): + """See :ref:`decoding` + """ return len(self._expl) @property def expl_llen(self): + """See :ref:`decoding` + """ if self.expl_lenindef: return 1 return len(len_encode(self.tlvlen)) @property def expl_offset(self): + """See :ref:`decoding` + """ return self.offset - self.expl_tlen - self.expl_llen @property def expl_vlen(self): + """See :ref:`decoding` + """ return self.tlvlen @property def expl_tlvlen(self): + """See :ref:`decoding` + """ return self.expl_tlen + self.expl_llen + self.expl_vlen @property def fulloffset(self): + """See :ref:`decoding` + """ return self.expl_offset if self.expled else self.offset @property def fulllen(self): + """See :ref:`decoding` + """ return self.expl_tlvlen if self.expled else self.tlvlen def pps_lenindef(self, decode_path): @@ -1325,12 +1499,12 @@ def _colourize(what, colour, with_colours, attrs=("bold",)): def colonize_hex(hexed): """Separate hexadecimal string with colons """ - return ":".join(hexed[i:i + 2] for i in range(0, len(hexed), 2)) + return ":".join(hexed[i:i + 2] for i in six_xrange(0, len(hexed), 2)) def pp_console_row( pp, - oids=None, + oid_maps=(), with_offsets=False, with_blob=True, with_colours=False, @@ -1365,14 +1539,18 @@ def pp_console_row( if isinstance(ent, DecodePathDefBy): cols.append(_colourize("DEFINED BY", "red", with_colours, ("reverse",))) value = str(ent.defined_by) + oid_name = None if ( - oids is not None and + len(oid_maps) > 0 and ent.defined_by.asn1_type_name == - ObjectIdentifier.asn1_type_name and - value in oids + ObjectIdentifier.asn1_type_name ): - cols.append(_colourize("%s:" % oids[value], "green", with_colours)) - else: + for oid_map in oid_maps: + oid_name = oid_map.get(value) + if oid_name is not None: + cols.append(_colourize("%s:" % oid_name, "green", with_colours)) + break + if oid_name is None: cols.append(_colourize("%s:" % value, "white", with_colours, ("reverse",))) else: cols.append(_colourize("%s:" % ent, "yellow", with_colours, ("reverse",))) @@ -1393,11 +1571,14 @@ def pp_console_row( value = pp.value cols.append(_colourize(value, "white", with_colours, ("reverse",))) if ( - oids is not None and - pp.asn1_type_name == ObjectIdentifier.asn1_type_name and - value in oids + len(oid_maps) > 0 and + pp.asn1_type_name == ObjectIdentifier.asn1_type_name ): - cols.append(_colourize("(%s)" % oids[value], "green", with_colours)) + for oid_map in oid_maps: + oid_name = oid_map.get(value) + if oid_name is not None: + cols.append(_colourize("(%s)" % oid_name, "green", with_colours)) + break if pp.asn1_type_name == Integer.asn1_type_name: hex_repr = hex(int(pp.obj._value))[2:].upper() if len(hex_repr) % 2 != 0: @@ -1432,7 +1613,7 @@ def pp_console_blob(pp, decode_path_len_decrease=0): cols.append(" ." * (decode_path_len + 1)) if isinstance(pp.blob, binary_type): blob = hexenc(pp.blob).upper() - for i in range(0, len(blob), 32): + for i in six_xrange(0, len(blob), 32): chunk = blob[i:i + 32] yield " ".join(cols + [colonize_hex(chunk)]) elif isinstance(pp.blob, tuple): @@ -1441,7 +1622,7 @@ def pp_console_blob(pp, decode_path_len_decrease=0): def pprint( obj, - oids=None, + oid_maps=(), big_blobs=False, with_colours=False, with_decode_path=False, @@ -1450,8 +1631,9 @@ def pprint( """Pretty print object :param Obj obj: object you want to pretty print - :param oids: ``OID <-> humand readable string`` dictionary. When OID - from it is met, then its humand readable form is printed + :param oid_maps: list of ``OID <-> humand readable string`` dictionary. + When OID from it is met, then its humand readable form + is printed :param big_blobs: if large binary objects are met (like OctetString values), do we need to print them too, on separate lines @@ -1473,7 +1655,7 @@ def pprint( if big_blobs: yield pp_console_row( pp, - oids=oids, + oid_maps=oid_maps, with_offsets=True, with_blob=False, with_colours=with_colours, @@ -1488,7 +1670,7 @@ def pprint( else: yield pp_console_row( pp, - oids=oids, + oid_maps=oid_maps, with_offsets=True, with_blob=True, with_colours=with_colours, @@ -1549,10 +1731,10 @@ class Boolean(Obj): self._value = default def _value_sanitize(self, value): - if issubclass(value.__class__, Boolean): - return value._value if isinstance(value, bool): return value + if issubclass(value.__class__, Boolean): + return value._value raise InvalidValueType((self.__class__, bool)) @property @@ -1634,7 +1816,7 @@ class Boolean(Obj): offset=offset, ) if tag_only: - return + return None try: l, _, v = len_decode(lv) except DecodeError as err: @@ -1798,10 +1980,10 @@ class Integer(Obj): self._value = default def _value_sanitize(self, value): - if issubclass(value.__class__, Integer): - value = value._value - elif isinstance(value, integer_types): + if isinstance(value, integer_types): pass + elif issubclass(value.__class__, Integer): + value = value._value elif isinstance(value, str): value = self.specs.get(value) if value is None: @@ -1861,9 +2043,10 @@ class Integer(Obj): @property def named(self): - for name, value in self.specs.items(): + for name, value in iteritems(self.specs): if value == self._value: return name + return None def __call__( self, @@ -1943,7 +2126,7 @@ class Integer(Obj): offset=offset, ) if tag_only: - return + return None try: l, llen, v = len_decode(lv) except DecodeError as err: @@ -2045,6 +2228,9 @@ class Integer(Obj): yield pp +SET01 = frozenset(("0", "1")) + + class BitString(Obj): """``BIT STRING`` bit string type @@ -2150,8 +2336,6 @@ class BitString(Obj): return bit_len, bytes(octets) def _value_sanitize(self, value): - if issubclass(value.__class__, BitString): - return value._value if isinstance(value, (string_types, binary_type)): if ( isinstance(value, string_types) and @@ -2159,10 +2343,10 @@ class BitString(Obj): ): if value.endswith("'B"): value = value[1:-2] - if not set(value) <= set(("0", "1")): + if not frozenset(value) <= SET01: raise ValueError("B's coding contains unacceptable chars") return self._bits2octets(value) - elif value.endswith("'H"): + if value.endswith("'H"): value = value[1:-2] return ( len(value) * 4, @@ -2170,8 +2354,7 @@ class BitString(Obj): ) if isinstance(value, binary_type): return (len(value) * 8, value) - else: - raise InvalidValueType((self.__class__, string_types, binary_type)) + raise InvalidValueType((self.__class__, string_types, binary_type)) if isinstance(value, tuple): if ( len(value) == 2 and @@ -2187,11 +2370,13 @@ class BitString(Obj): bits.append(bit) if len(bits) == 0: return self._bits2octets("") - bits = set(bits) + bits = frozenset(bits) return self._bits2octets("".join( ("1" if bit in bits else "0") for bit in six_xrange(max(bits) + 1) )) + if issubclass(value.__class__, BitString): + return value._value raise InvalidValueType((self.__class__, binary_type, string_types)) @property @@ -2243,7 +2428,7 @@ class BitString(Obj): @property def named(self): - return [name for name, bit in self.specs.items() if self[bit]] + return [name for name, bit in iteritems(self.specs) if self[bit]] def __call__( self, @@ -2288,7 +2473,7 @@ class BitString(Obj): octets, )) - def _decode_chunk(self, lv, offset, decode_path, ctx): + def _decode_chunk(self, lv, offset, decode_path): try: l, llen, v = len_decode(lv) except DecodeError as err: @@ -2358,8 +2543,8 @@ class BitString(Obj): ) if t == self.tag: if tag_only: # pragma: no cover - return - return self._decode_chunk(lv, offset, decode_path, ctx) + return None + return self._decode_chunk(lv, offset, decode_path) if t == self.tag_constructed: if not ctx.get("bered", False): raise DecodeError( @@ -2369,7 +2554,7 @@ class BitString(Obj): offset=offset, ) if tag_only: # pragma: no cover - return + return None lenindef = False try: l, llen, v = len_decode(lv) @@ -2567,13 +2752,7 @@ class OctetString(Obj): :param default: set default value. Type same as in ``value`` :param bool optional: is object ``OPTIONAL`` in sequence """ - super(OctetString, self).__init__( - impl, - expl, - default, - optional, - _decoded, - ) + super(OctetString, self).__init__(impl, expl, default, optional, _decoded) self._value = value self._bound_min, self._bound_max = getattr( self, @@ -2600,10 +2779,10 @@ class OctetString(Obj): ) def _value_sanitize(self, value): - if issubclass(value.__class__, OctetString): - value = value._value - elif isinstance(value, binary_type): + if isinstance(value, binary_type): pass + elif issubclass(value.__class__, OctetString): + value = value._value else: raise InvalidValueType((self.__class__, bytes)) if not self._bound_min <= len(value) <= self._bound_max: @@ -2678,7 +2857,7 @@ class OctetString(Obj): self._value, )) - def _decode_chunk(self, lv, offset, decode_path, ctx): + def _decode_chunk(self, lv, offset, decode_path): try: l, llen, v = len_decode(lv) except DecodeError as err: @@ -2734,8 +2913,8 @@ class OctetString(Obj): ) if t == self.tag: if tag_only: - return - return self._decode_chunk(lv, offset, decode_path, ctx) + return None + return self._decode_chunk(lv, offset, decode_path) if t == self.tag_constructed: if not ctx.get("bered", False): raise DecodeError( @@ -2745,7 +2924,7 @@ class OctetString(Obj): offset=offset, ) if tag_only: - return + return None lenindef = False try: l, llen, v = len_decode(lv) @@ -2961,7 +3140,7 @@ class Null(Obj): offset=offset, ) if tag_only: # pragma: no cover - return + return None try: l, _, v = len_decode(lv) except DecodeError as err: @@ -3063,13 +3242,7 @@ class ObjectIdentifier(Obj): :param default: set default value. Type same as in ``value`` :param bool optional: is object ``OPTIONAL`` in sequence """ - super(ObjectIdentifier, self).__init__( - impl, - expl, - default, - optional, - _decoded, - ) + super(ObjectIdentifier, self).__init__(impl, expl, default, optional, _decoded) self._value = value if value is not None: self._value = self._value_sanitize(value) @@ -3216,7 +3389,7 @@ class ObjectIdentifier(Obj): offset=offset, ) if tag_only: # pragma: no cover - return + return None try: l, llen, v = len_decode(lv) except DecodeError as err: @@ -3340,13 +3513,7 @@ class Enumerated(Integer): bounds=None, # dummy argument, workability for Integer.decode ): super(Enumerated, self).__init__( - value=value, - impl=impl, - expl=expl, - default=default, - optional=optional, - _specs=_specs, - _decoded=_decoded, + value, bounds, impl, expl,default, optional, _specs, _decoded, ) if len(self.specs) == 0: raise ValueError("schema must be specified") @@ -3355,7 +3522,10 @@ class Enumerated(Integer): if isinstance(value, self.__class__): value = value._value elif isinstance(value, integer_types): - if value not in list(self.specs.values()): + for _value in itervalues(self.specs): + if _value == value: + break + else: raise DecodeError( "unknown integer value: %s" % value, klass=self.__class__, @@ -3404,6 +3574,12 @@ class Enumerated(Integer): ) +def escape_control_unicode(c): + if unicat(c).startswith("C"): + c = repr(c).lstrip("u").strip("'") + return c + + class CommonString(OctetString): """Common class for all strings @@ -3522,7 +3698,10 @@ class CommonString(OctetString): def pps(self, decode_path=(), no_unicode=False): value = None if self.ready: - value = hexenc(bytes(self)) if no_unicode else self.__unicode__() + value = ( + hexenc(bytes(self)) if no_unicode else + "".join(escape_control_unicode(c) for c in self.__unicode__()) + ) yield _pp( obj=self, asn1_type_name=self.asn1_type_name, @@ -3561,7 +3740,7 @@ class AllowableCharsMixin(object): def allowable_chars(self): if PY2: return self._allowable_chars - return set(six_unichr(c) for c in self._allowable_chars) + return frozenset(six_unichr(c) for c in self._allowable_chars) class NumericString(AllowableCharsMixin, CommonString): @@ -3571,17 +3750,17 @@ class NumericString(AllowableCharsMixin, CommonString): be stored. >>> NumericString().allowable_chars - set(['3', '4', '7', '5', '1', '0', '8', '9', ' ', '6', '2']) + frozenset(['0', '1', '2', '3', '4', '5', '6', '7', '8', '9', ' ']) """ __slots__ = () tag_default = tag_encode(18) encoding = "ascii" asn1_type_name = "NumericString" - _allowable_chars = set(digits.encode("ascii") + b" ") + _allowable_chars = frozenset(digits.encode("ascii") + b" ") def _value_sanitize(self, value): value = super(NumericString, self)._value_sanitize(value) - if not set(value) <= self._allowable_chars: + if not frozenset(value) <= self._allowable_chars: raise DecodeError("non-numeric value") return value @@ -3592,22 +3771,76 @@ class PrintableString(AllowableCharsMixin, CommonString): Its value is properly sanitized: see X.680 41.4 table 10. >>> PrintableString().allowable_chars - >>> set([' ', "'", ..., 'z']) + frozenset([' ', "'", ..., 'z']) """ __slots__ = () tag_default = tag_encode(19) encoding = "ascii" asn1_type_name = "PrintableString" - _allowable_chars = set( + _allowable_chars = frozenset( (ascii_letters + digits + " '()+,-./:=?").encode("ascii") ) + _asterisk = frozenset("*".encode("ascii")) + _ampersand = frozenset("&".encode("ascii")) + + def __init__( + self, + value=None, + bounds=None, + impl=None, + expl=None, + default=None, + optional=False, + _decoded=(0, 0, 0), + allow_asterisk=False, + allow_ampersand=False, + ): + """ + :param allow_asterisk: allow asterisk character + :param allow_ampersand: allow ampersand character + """ + if allow_asterisk: + self._allowable_chars |= self._asterisk + if allow_ampersand: + self._allowable_chars |= self._ampersand + super(PrintableString, self).__init__( + value, bounds, impl, expl, default, optional, _decoded, + ) def _value_sanitize(self, value): value = super(PrintableString, self)._value_sanitize(value) - if not set(value) <= self._allowable_chars: + if not frozenset(value) <= self._allowable_chars: raise DecodeError("non-printable value") return value + def copy(self): + obj = super(PrintableString, self).copy() + obj._allowable_chars = self._allowable_chars + return obj + + def __call__( + self, + value=None, + bounds=None, + impl=None, + expl=None, + default=None, + optional=None, + ): + return self.__class__( + value=value, + bounds=( + (self._bound_min, self._bound_max) + if bounds is None else bounds + ), + impl=self.tag if impl is None else impl, + expl=self._expl if expl is None else expl, + default=self.default if default is None else default, + optional=self.optional if optional is None else optional, + allow_asterisk=self._asterisk <= self._allowable_chars, + allow_ampersand=self._ampersand <= self._allowable_chars, + ) + class TeletexString(CommonString): __slots__ = () @@ -3653,14 +3886,16 @@ class UTCTime(CommonString): datetime.datetime(2017, 9, 30, 22, 7, 50) >>> UTCTime(datetime(2057, 9, 30, 22, 7, 50)).todatetime() datetime.datetime(1957, 9, 30, 22, 7, 50) + + .. warning:: + + BER encoding is unsupported. """ __slots__ = () tag_default = tag_encode(23) encoding = "ascii" asn1_type_name = "UTCTime" - fmt = "%y%m%d%H%M%SZ" - def __init__( self, value=None, @@ -3680,11 +3915,7 @@ class UTCTime(CommonString): :param bool optional: is object ``OPTIONAL`` in sequence """ super(UTCTime, self).__init__( - impl=impl, - expl=expl, - default=default, - optional=optional, - _decoded=_decoded, + None, None, impl, expl, default, optional, _decoded, ) self._value = value if value is not None: @@ -3699,24 +3930,36 @@ class UTCTime(CommonString): if self._value is None: self._value = default + def _strptime(self, value): + # datetime.strptime's format: %y%m%d%H%M%SZ + if len(value) != LEN_YYMMDDHHMMSSZ: + raise ValueError("invalid UTCTime length") + if value[-1] != "Z": + raise ValueError("non UTC timezone") + return datetime( + 2000 + int(value[:2]), # %y + int(value[2:4]), # %m + int(value[4:6]), # %d + int(value[6:8]), # %H + int(value[8:10]), # %M + int(value[10:12]), # %S + ) + def _value_sanitize(self, value): - if isinstance(value, self.__class__): - return value._value - if isinstance(value, datetime): - return value.strftime(self.fmt).encode("ascii") if isinstance(value, binary_type): try: value_decoded = value.decode("ascii") except (UnicodeEncodeError, UnicodeDecodeError) as err: - raise DecodeError("invalid UTCTime encoding") - if len(value_decoded) == LEN_YYMMDDHHMMSSZ: - try: - datetime.strptime(value_decoded, self.fmt) - except (TypeError, ValueError): - raise DecodeError("invalid UTCTime format") - return value - else: - raise DecodeError("invalid UTCTime length") + raise DecodeError("invalid UTCTime encoding: %r" % err) + try: + self._strptime(value_decoded) + except (TypeError, ValueError) as err: + raise DecodeError("invalid UTCTime format: %r" % err) + return value + if isinstance(value, self.__class__): + return value._value + if isinstance(value, datetime): + return value.strftime("%y%m%d%H%M%SZ").encode("ascii") raise InvalidValueType((self.__class__, datetime)) def __eq__(self, their): @@ -3741,7 +3984,7 @@ class UTCTime(CommonString): having < 50 years are treated as 20xx, 19xx otherwise, according to X.509 recomendation. """ - value = datetime.strptime(self._value.decode("ascii"), self.fmt) + value = self._strptime(self._value.decode("ascii")) year = value.year % 100 return datetime( year=(2000 + year) if year < 50 else (1900 + year), @@ -3793,54 +4036,85 @@ class GeneralizedTime(UTCTime): '20170930220750.000123Z' >>> t = GeneralizedTime(datetime(2057, 9, 30, 22, 7, 50)) GeneralizedTime GeneralizedTime 2057-09-30T22:07:50 + + .. warning:: + + BER encoding is unsupported. + + .. warning:: + + Only microsecond fractions are supported. + :py:exc:`pyderasn.DecodeError` will be raised during decoding of + higher precision values. """ __slots__ = () tag_default = tag_encode(24) asn1_type_name = "GeneralizedTime" - fmt = "%Y%m%d%H%M%SZ" - fmt_ms = "%Y%m%d%H%M%S.%fZ" + def _strptime(self, value): + l = len(value) + if l == LEN_YYYYMMDDHHMMSSZ: + # datetime.strptime's format: %y%m%d%H%M%SZ + if value[-1] != "Z": + raise ValueError("non UTC timezone") + return datetime( + int(value[:4]), # %Y + int(value[4:6]), # %m + int(value[6:8]), # %d + int(value[8:10]), # %H + int(value[10:12]), # %M + int(value[12:14]), # %S + ) + if l >= LEN_YYYYMMDDHHMMSSDMZ: + # datetime.strptime's format: %Y%m%d%H%M%S.%fZ + if value[-1] != "Z": + raise ValueError("non UTC timezone") + if value[14] != ".": + raise ValueError("no fractions separator") + us = value[15:-1] + if us[-1] == "0": + raise ValueError("trailing zero") + us_len = len(us) + if us_len > 6: + raise ValueError("only microsecond fractions are supported") + us = int(us + ("0" * (6 - us_len))) + decoded = datetime( + int(value[:4]), # %Y + int(value[4:6]), # %m + int(value[6:8]), # %d + int(value[8:10]), # %H + int(value[10:12]), # %M + int(value[12:14]), # %S + us, # %f + ) + return decoded + raise ValueError("invalid GeneralizedTime length") def _value_sanitize(self, value): - if isinstance(value, self.__class__): - return value._value - if isinstance(value, datetime): - return value.strftime( - self.fmt_ms if value.microsecond > 0 else self.fmt - ).encode("ascii") if isinstance(value, binary_type): try: value_decoded = value.decode("ascii") except (UnicodeEncodeError, UnicodeDecodeError) as err: - raise DecodeError("invalid GeneralizedTime encoding") - if len(value_decoded) == LEN_YYYYMMDDHHMMSSZ: - try: - datetime.strptime(value_decoded, self.fmt) - except (TypeError, ValueError): - raise DecodeError( - "invalid GeneralizedTime (without ms) format", - ) - return value - elif len(value_decoded) >= LEN_YYYYMMDDHHMMSSDMZ: - try: - datetime.strptime(value_decoded, self.fmt_ms) - except (TypeError, ValueError): - raise DecodeError( - "invalid GeneralizedTime (with ms) format", - ) - return value - else: + raise DecodeError("invalid GeneralizedTime encoding: %r" % err) + try: + self._strptime(value_decoded) + except (TypeError, ValueError) as err: raise DecodeError( - "invalid GeneralizedTime length", + "invalid GeneralizedTime format: %r" % err, klass=self.__class__, ) + return value + if isinstance(value, self.__class__): + return value._value + if isinstance(value, datetime): + encoded = value.strftime("%Y%m%d%H%M%S") + if value.microsecond > 0: + encoded = encoded + (".%06d" % value.microsecond).rstrip("0") + return (encoded + "Z").encode("ascii") raise InvalidValueType((self.__class__, datetime)) def todatetime(self): - value = self._value.decode("ascii") - if len(value) == LEN_YYYYMMDDHHMMSSZ: - return datetime.strptime(value, self.fmt) - return datetime.strptime(value, self.fmt_ms) + return self._strptime(self._value.decode("ascii")) class GraphicString(CommonString): @@ -3959,8 +4233,6 @@ class Choice(Obj): self._value = default_obj.copy()._value def _value_sanitize(self, value): - if isinstance(value, self.__class__): - return value._value if isinstance(value, tuple) and len(value) == 2: choice, obj = value spec = self.specs.get(choice) @@ -3969,6 +4241,8 @@ class Choice(Obj): if not isinstance(obj, spec.__class__): raise InvalidValueType((spec,)) return (choice, spec(obj)) + if isinstance(value, self.__class__): + return value._value raise InvalidValueType((self.__class__, tuple)) @property @@ -4064,7 +4338,7 @@ class Choice(Obj): return self._value[1].encode() def _decode(self, tlv, offset, decode_path, ctx, tag_only): - for choice, spec in self.specs.items(): + for choice, spec in iteritems(self.specs): sub_decode_path = decode_path + (choice,) try: spec.decode( @@ -4086,7 +4360,7 @@ class Choice(Obj): offset=offset, ) if tag_only: # pragma: no cover - return + return None value, tail = spec.decode( tlv, offset=offset, @@ -4204,12 +4478,12 @@ class Any(Obj): self.defined = None def _value_sanitize(self, value): + if isinstance(value, binary_type): + return value if isinstance(value, self.__class__): return value._value if isinstance(value, Obj): return value.encode() - if isinstance(value, binary_type): - return value raise InvalidValueType((self.__class__, Obj, binary_type)) @property @@ -4549,22 +4823,21 @@ class Sequence(Obj): @property def ready(self): - for name, spec in self.specs.items(): + for name, spec in iteritems(self.specs): value = self._value.get(name) if value is None: if spec.optional: continue return False - else: - if not value.ready: - return False + if not value.ready: + return False return True @property def bered(self): if self.expl_lenindef or self.lenindef or self.ber_encoded: return True - return any(value.bered for value in self._value.values()) + return any(value.bered for value in itervalues(self._value)) def copy(self): obj = self.__class__(schema=self.specs) @@ -4578,7 +4851,7 @@ class Sequence(Obj): obj.expl_lenindef = self.expl_lenindef obj.lenindef = self.lenindef obj.ber_encoded = self.ber_encoded - obj._value = {k: v.copy() for k, v in self._value.items()} + obj._value = {k: v.copy() for k, v in iteritems(self._value)} return obj def __eq__(self, their): @@ -4639,7 +4912,7 @@ class Sequence(Obj): def _encoded_values(self): raws = [] - for name, spec in self.specs.items(): + for name, spec in iteritems(self.specs): value = self._value.get(name) if value is None: if spec.optional: @@ -4669,7 +4942,7 @@ class Sequence(Obj): offset=offset, ) if tag_only: # pragma: no cover - return + return None lenindef = False ctx_bered = ctx.get("bered", False) try: @@ -4705,7 +4978,7 @@ class Sequence(Obj): values = {} ber_encoded = False ctx_allow_default_values = ctx.get("allow_default_values", False) - for name, spec in self.specs.items(): + for name, spec in iteritems(self.specs): if spec.optional and ( (lenindef and v[:EOC_LEN].tobytes() == EOC) or len(v) == 0 @@ -4721,8 +4994,8 @@ class Sequence(Obj): ctx=ctx, _ctx_immutable=False, ) - except TagMismatch: - if spec.optional: + except TagMismatch as err: + if (len(err.decode_path) == len(decode_path) + 1) and spec.optional: continue raise @@ -4899,6 +5172,9 @@ class Set(Sequence): v = b"".join(raws) return b"".join((self.tag, len_encode(len(v)), v)) + def _specs_items(self): + return iteritems(self.specs) + def _decode(self, tlv, offset, decode_path, ctx, tag_only): try: t, tlen, lv = tag_strip(tlv) @@ -4916,7 +5192,7 @@ class Set(Sequence): offset=offset, ) if tag_only: - return + return None lenindef = False ctx_bered = ctx.get("bered", False) try: @@ -4953,11 +5229,11 @@ class Set(Sequence): ctx_allow_default_values = ctx.get("allow_default_values", False) ctx_allow_unordered_set = ctx.get("allow_unordered_set", False) value_prev = memoryview(v[:0]) - specs_items = self.specs.items + while len(v) > 0: if lenindef and v[:EOC_LEN].tobytes() == EOC: break - for name, spec in specs_items(): + for name, spec in self._specs_items(): sub_decode_path = decode_path + (name,) try: spec.decode( @@ -5088,13 +5364,7 @@ class SequenceOf(Obj): optional=False, _decoded=(0, 0, 0), ): - super(SequenceOf, self).__init__( - impl, - expl, - default, - optional, - _decoded, - ) + super(SequenceOf, self).__init__(impl, expl, default, optional, _decoded) if schema is None: schema = getattr(self, "schema", None) if schema is None: @@ -5249,7 +5519,7 @@ class SequenceOf(Obj): offset=offset, ) if tag_only: - return + return None lenindef = False ctx_bered = ctx.get("bered", False) try: @@ -5431,7 +5701,7 @@ def generic_decoder(): # pragma: no cover choice = PrimitiveTypes() choice.specs["SequenceOf"] = SequenceOf(schema=choice) choice.specs["SetOf"] = SetOf(schema=choice) - for i in range(31): + for i in six_xrange(31): choice.specs["SequenceOf%d" % i] = SequenceOf( schema=choice, expl=tag_ctxc(i), @@ -5445,7 +5715,7 @@ def generic_decoder(): # pragma: no cover def pprint_any( obj, - oids=None, + oid_maps=(), with_colours=False, with_decode_path=False, decode_path_only=(), @@ -5465,7 +5735,7 @@ def generic_decoder(): # pragma: no cover pp = _pp(**pp_kwargs) yield pp_console_row( pp, - oids=oids, + oid_maps=oid_maps, with_offsets=True, with_blob=False, with_colours=with_colours, @@ -5495,7 +5765,7 @@ def main(): # pragma: no cover ) parser.add_argument( "--oids", - help="Python path to dictionary with OIDs", + help="Python paths to dictionary with OIDs, comma separated", ) parser.add_argument( "--schema", @@ -5533,7 +5803,10 @@ def main(): # pragma: no cover args.DERFile.seek(args.skip) der = memoryview(args.DERFile.read()) args.DERFile.close() - oids = obj_by_path(args.oids) if args.oids else {} + oid_maps = ( + [obj_by_path(_path) for _path in (args.oids or "").split(",")] + if args.oids else () + ) if args.schema: schema = obj_by_path(args.schema) from functools import partial @@ -5549,8 +5822,8 @@ def main(): # pragma: no cover obj, tail = schema().decode(der, ctx=ctx) print(pprinter( obj, - oids=oids, - with_colours=True if environ.get("NO_COLOR") is None else False, + oid_maps=oid_maps, + with_colours=environ.get("NO_COLOR") is None, with_decode_path=args.print_decode_path, decode_path_only=( () if args.decode_path_only is None else