X-Git-Url: http://www.git.cypherpunks.ru/?p=pyderasn.git;a=blobdiff_plain;f=pyderasn.py;h=5adc5b068d17915f0c543144b66557b9fc355f30;hp=3b5c4d824a47eda126121f843281c0050b2eed6a;hb=3b5f3e7f4219b79410bd7b3e1a3d41be51f78f0a;hpb=88a72bcbbb16c2e0d58b6c6fdc6c51b8e517ee7f diff --git a/pyderasn.py b/pyderasn.py index 3b5c4d8..5adc5b0 100755 --- a/pyderasn.py +++ b/pyderasn.py @@ -1,12 +1,11 @@ #!/usr/bin/env python # coding: utf-8 -# PyDERASN -- Python ASN.1 DER codec with abstract structures -# Copyright (C) 2017-2018 Sergey Matveev +# PyDERASN -- Python ASN.1 DER/BER codec with abstract structures +# 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,16 +13,15 @@ # 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 -# . -"""Python ASN.1 DER codec with abstract structures +# License along with this program. If not, see . +"""Python ASN.1 DER/BER codec with abstract structures -This library allows you to marshal and unmarshal various structures in -ASN.1 DER format, like this: +This library allows you to marshal various structures in ASN.1 DER +format, unmarshal them in BER/CER/DER ones. >>> i = Integer(123) >>> raw = i.encode() - >>> Integer().decode(raw) == i + >>> Integer().decod(raw) == i True There are primitive types, holding single values @@ -67,10 +65,11 @@ ____ Most types in ASN.1 has specific tag for them. ``Obj.tag_default`` is the default tag used during coding process. You can override it with -either ``IMPLICIT`` (using ``impl`` keyword argument), or -``EXPLICIT`` one (using ``expl`` keyword argument). Both arguments take -raw binary string, containing that tag. You can **not** set implicit and -explicit tags simultaneously. +either ``IMPLICIT`` (using either ``impl`` keyword argument or ``impl`` +class attribute), or ``EXPLICIT`` one (using either ``expl`` keyword +argument or ``expl`` class attribute). Both arguments take raw binary +string, containing that tag. You can **not** set implicit and explicit +tags simultaneously. There are :py:func:`pyderasn.tag_ctxp` and :py:func:`pyderasn.tag_ctxc` functions, allowing you to easily create ``CONTEXT`` @@ -163,21 +162,27 @@ All objects have ``ready`` boolean property, that tells if object is ready to be encoded. If that kind of action is performed on unready object, then :py:exc:`pyderasn.ObjNotReady` exception will be raised. -All objects have ``copy()`` method, that returns their copy, that can be +All objects are friendly to ``copy.copy()`` and copied objects can be safely mutated. +Also all objects can be safely ``pickle``-d, but pay attention that +pickling among different PyDERASN versions is prohibited. + .. _decoding: Decoding -------- -Decoding is performed using ``decode()`` method. ``offset`` optional -argument could be used to set initial object's offset in the binary -data, for convenience. It returns decoded object and remaining -unmarshalled data (tail). Internally all work is done on +Decoding is performed using :py:meth:`pyderasn.Obj.decode` method. +``offset`` optional argument could be used to set initial object's +offset in the binary data, for convenience. It returns decoded object +and remaining unmarshalled data (tail). Internally all work is done on ``memoryview(data)``, and you can leave returning tail as a memoryview, by specifying ``leavemm=True`` argument. +Also note convenient :py:meth:`pyderasn.Obj.decod` method, that +immediately checks and raises if there is non-empty tail. + When object is decoded, ``decoded`` property is true and you can safely use following properties: @@ -189,25 +194,35 @@ use following properties: Pay attention that those values do **not** include anything related to explicit tag. If you want to know information about it, then use: -``expled`` (to know if explicit tag is set), ``expl_offset`` (it is -lesser than ``offset``), ``expl_tlen``, ``expl_llen``, ``expl_vlen`` -(that actually equals to ordinary ``tlvlen``). -When error occurs, then :py:exc:`pyderasn.DecodeError` is raised. +* ``expled`` -- to know if explicit tag is set +* ``expl_offset`` (it is lesser than ``offset``) +* ``expl_tlen``, +* ``expl_llen`` +* ``expl_vlen`` (that actually equals to ordinary ``tlvlen``) +* ``fulloffset`` -- it equals to ``expl_offset`` if explicit tag is set, + ``offset`` otherwise +* ``fulllen`` -- it equals to ``expl_len`` if explicit tag is set, + ``tlvlen`` otherwise + +When error occurs, :py:exc:`pyderasn.DecodeError` is raised. .. _ctx: Context _______ -You can specify so called context keyword argument during ``decode()`` -invocation. It is dictionary containing various options governing -decoding process. +You can specify so called context keyword argument during +:py:meth:`pyderasn.Obj.decode` invocation. It is dictionary containing +various options governing decoding process. Currently available context options: +* :ref:`allow_default_values ` +* :ref:`allow_expl_oob ` +* :ref:`allow_unordered_set ` +* :ref:`bered ` * :ref:`defines_by_path ` -* :ref:`strict_default_existence ` .. _pprinting: @@ -229,6 +244,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 @@ -239,6 +353,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 _____________ @@ -266,7 +382,7 @@ You can specify multiple fields, that will be autodecoded -- that is why ``defines`` kwarg is a sequence. You can specify defined field relatively or absolutely to current decode path. For example ``defines`` for AlgorithmIdentifier of X.509's -``tbsCertificate.subjectPublicKeyInfo.algorithm.algorithm``:: +``tbsCertificate:subjectPublicKeyInfo:algorithm:algorithm``:: ( (("parameters",), { @@ -294,8 +410,7 @@ Following types can be automatically decoded (DEFINED BY): When any of those fields is automatically decoded, then ``.defined`` attribute contains ``(OID, value)`` tuple. ``OID`` tells by which OID it was defined, ``value`` contains corresponding decoded value. For example -above, ``content_info["content"].defined == (id_signedData, -signed_data)``. +above, ``content_info["content"].defined == (id_signedData, signed_data)``. .. _defines_by_path_ctx: @@ -313,15 +428,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 = ContentInfo().decod(data, ctx={"defines_by_path": ( ( ("contentType",), ((("content",), {id_signedData: SignedData()}),), @@ -356,13 +471,64 @@ 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 decoding is already done. ``any`` means literally any value it meet -- useful for SEQUENCE/SET OF-s. +.. _bered_ctx: + +BER encoding +------------ + +By default PyDERASN accepts only DER encoded data. It always encodes to +DER. But you can optionally enable BER decoding with setting ``bered`` +:ref:`context ` argument to True. Indefinite lengths and +constructed primitive types should be parsed successfully. + +* If object is encoded in BER form (not the DER one), then ``ber_encoded`` + attribute is set to True. Only ``BOOLEAN``, ``BIT STRING``, ``OCTET + STRING``, ``OBJECT IDENTIFIER``, ``SEQUENCE``, ``SET``, ``SET OF`` + can contain it. +* If object has an indefinite length encoding, then its ``lenindef`` + attribute is set to True. Only ``BIT STRING``, ``OCTET STRING``, + ``SEQUENCE``, ``SET``, ``SEQUENCE OF``, ``SET OF``, ``ANY`` can + contain it. +* If object has an indefinite length encoded explicit tag, then + ``expl_lenindef`` is set to True. +* If object has either any of BER-related encoding (explicit tag + indefinite length, object's indefinite length, BER-encoding) or any + underlying component has that kind of encoding, then ``bered`` + attribute is set to True. For example SignedData CMS can have + ``ContentInfo:content:signerInfos:*`` ``bered`` value set to True, but + ``ContentInfo:content:signerInfos:*:signedAttrs`` won't. + +EOC (end-of-contents) token's length is taken in advance in object's +value length. + +.. _allow_expl_oob_ctx: + +Allow explicit tag out-of-bound +------------------------------- + +Invalid BER encoding could contain ``EXPLICIT`` tag containing more than +one value, more than one object. If you set ``allow_expl_oob`` context +option to True, then no error will be raised and that invalid encoding +will be silently further processed. But pay attention that offsets and +lengths will be invalid in that case. + +.. warning:: + + 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 --------------- @@ -404,6 +570,15 @@ CommonString ____________ .. autoclass:: pyderasn.CommonString +NumericString +_____________ +.. autoclass:: pyderasn.NumericString + +PrintableString +_______________ +.. autoclass:: pyderasn.PrintableString + :members: __init__ + UTCTime _______ .. autoclass:: pyderasn.UTCTime @@ -457,23 +632,38 @@ Various ------- .. autofunction:: pyderasn.abs_decode_path +.. autofunction:: pyderasn.colonize_hex .. autofunction:: pyderasn.hexenc .. autofunction:: pyderasn.hexdec .. autofunction:: pyderasn.tag_encode .. 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 +.. autoclass:: pyderasn.InvalidOID +.. autoclass:: pyderasn.ObjUnknown +.. autoclass:: pyderasn.ObjNotReady +.. autoclass:: pyderasn.InvalidValueType +.. autoclass:: pyderasn.BoundsError """ from codecs import getdecoder from codecs import getencoder from collections import namedtuple from collections import OrderedDict +from copy import copy from datetime import datetime 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 @@ -482,18 +672,22 @@ 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 +from six import unichr as six_unichr from six.moves import xrange as six_xrange try: from termcolor import colored -except ImportError: - def colored(what, *args): +except ImportError: # pragma: no cover + def colored(what, *args, **kwargs): return what +__version__ = "6.0" __all__ = ( "Any", @@ -505,6 +699,7 @@ __all__ = ( "DecodeError", "DecodePathDefBy", "Enumerated", + "ExceedingData", "GeneralizedTime", "GeneralString", "GraphicString", @@ -516,6 +711,7 @@ __all__ = ( "InvalidOID", "InvalidValueType", "ISO646String", + "LenIndefForm", "NotEnoughData", "Null", "NumericString", @@ -561,13 +757,21 @@ TagClassReprs = { TagClassPrivate: "PRIVATE ", TagClassUniversal: "UNIV ", } +EOC = b"\x00\x00" +EOC_LEN = len(EOC) +LENINDEF = b"\x80" # length indefinite mark +LENINDEF_PP_CHAR = "I" if PY2 else "∞" ######################################################################## # Errors ######################################################################## -class DecodeError(Exception): +class ASN1Error(ValueError): + pass + + +class DecodeError(ASN1Error): def __init__(self, msg="", klass=None, decode_path=(), offset=0): """ :param str msg: reason of decode failing @@ -590,7 +794,7 @@ class DecodeError(Exception): c for c in ( "" if self.klass is None else self.klass.__name__, ( - ("(%s)" % ".".join(str(dp) for dp in self.decode_path)) + ("(%s)" % ":".join(str(dp) for dp in self.decode_path)) if len(self.decode_path) > 0 else "" ), ("(at %d)" % self.offset) if self.offset > 0 else "", @@ -606,6 +810,22 @@ 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 + + class TagMismatch(DecodeError): pass @@ -618,7 +838,7 @@ class InvalidOID(DecodeError): pass -class ObjUnknown(ValueError): +class ObjUnknown(ASN1Error): def __init__(self, name): super(ObjUnknown, self).__init__() self.name = name @@ -630,7 +850,7 @@ class ObjUnknown(ValueError): return "%s(%s)" % (self.__class__.__name__, self) -class ObjNotReady(ValueError): +class ObjNotReady(ASN1Error): def __init__(self, name): super(ObjNotReady, self).__init__() self.name = name @@ -642,7 +862,7 @@ class ObjNotReady(ValueError): return "%s(%s)" % (self.__class__.__name__, self) -class InvalidValueType(ValueError): +class InvalidValueType(ASN1Error): def __init__(self, expected_types): super(InvalidValueType, self).__init__() self.expected_types = expected_types @@ -656,7 +876,7 @@ class InvalidValueType(ValueError): return "%s(%s)" % (self.__class__.__name__, self) -class BoundsError(ValueError): +class BoundsError(ASN1Error): def __init__(self, bound_min, value, bound_max): super(BoundsError, self).__init__() self.bound_min = bound_min @@ -797,6 +1017,11 @@ def len_encode(l): def len_decode(data): + """Decode length + + :returns: (decoded length, length's length, remaining data) + :raises LenIndefForm: if indefinite form encoding is met + """ if len(data) == 0: raise NotEnoughData("no data at all") first_octet = byte2int(data) @@ -806,7 +1031,7 @@ def len_decode(data): if octets_num + 1 > len(data): raise NotEnoughData("encoded length is longer than data") if octets_num == 0: - raise DecodeError("long form instead of short one") + raise LenIndefForm() if byte2int(data[1:]) == 0: raise DecodeError("leading zeros") l = 0 @@ -822,9 +1047,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) @@ -843,6 +1068,9 @@ class Obj(object): "offset", "llen", "vlen", + "expl_lenindef", + "lenindef", + "ber_encoded", ) def __init__( @@ -856,14 +1084,15 @@ class Obj(object): self.tag = getattr(self, "impl", self.tag_default) if impl is None else impl self._expl = getattr(self, "expl", None) if expl is None else expl if self.tag != self.tag_default and self._expl is not None: - raise ValueError( - "implicit and explicit tags can not be set simultaneously" - ) + raise ValueError("implicit and explicit tags can not be set simultaneously") if default is not None: optional = True self.optional = optional self.offset, self.llen, self.vlen = _decoded self.default = None + self.expl_lenindef = False + self.lenindef = False + self.ber_encoded = False @property def ready(self): # pragma: no cover @@ -875,23 +1104,48 @@ class Obj(object): if not self.ready: raise ObjNotReady(self.__class__.__name__) + @property + def bered(self): + """Is either object or any elements inside is BER encoded? + """ + return self.expl_lenindef or self.lenindef or self.ber_encoded + @property def decoded(self): """Is object decoded? """ return (self.llen + self.vlen) > 0 - def copy(self): # pragma: no cover - """Make a copy of object, safe to be mutated + def __getstate__(self): # pragma: no cover + """Used for making safe to be mutable pickleable copies """ raise NotImplementedError() + def __setstate__(self, state): + if state.version != __version__: + raise ValueError("data is pickled by different PyDERASN version") + self.tag = self.tag_default + self._value = None + self._expl = None + self.default = None + self.optional = False + self.offset = 0 + self.llen = 0 + self.vlen = 0 + self.expl_lenindef = False + self.lenindef = False + self.ber_encoded = False + @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 @@ -912,35 +1166,66 @@ class Obj(object): def _encode(self): # pragma: no cover raise NotImplementedError() - def _decode(self, tlv, offset, decode_path, ctx): # pragma: no cover + def _decode(self, tlv, offset, decode_path, ctx, tag_only): # pragma: no cover raise NotImplementedError() def encode(self): + """Encode the structure + + :returns: DER representation + """ raw = self._encode() if self._expl is None: return raw return b"".join((self._expl, len_encode(len(raw)), raw)) - def decode(self, data, offset=0, leavemm=False, decode_path=(), ctx=None): + def hexencode(self): + """Do hexadecimal encoded :py:meth:`pyderasn.Obj.encode` + """ + return hexenc(self.encode()) + + def decode( + self, + data, + offset=0, + leavemm=False, + decode_path=(), + ctx=None, + tag_only=False, + _ctx_immutable=True, + ): """Decode the data :param data: either binary or memoryview :param int offset: initial data's offset :param bool leavemm: do we need to leave memoryview of remaining data as is, or convert it to bytes otherwise - :param ctx: optional :ref:`context ` governing decoding process. + :param ctx: optional :ref:`context ` governing decoding process + :param tag_only: decode only the tag, without length and contents + (used only in Choice and Set structures, trying to + determine if tag satisfies the scheme) + :param _ctx_immutable: do we need to ``copy.copy()`` ``ctx`` + before using it? :returns: (Obj, remaining data) + + .. seealso:: :ref:`decoding` """ if ctx is None: ctx = {} + elif _ctx_immutable: + ctx = copy(ctx) tlv = memoryview(data) if self._expl is None: - obj, tail = self._decode( + result = self._decode( tlv, offset, decode_path=decode_path, ctx=ctx, + tag_only=tag_only, ) + if tag_only: + return None + obj, tail = result else: try: t, tlen, lv = tag_strip(tlv) @@ -959,6 +1244,36 @@ class Obj(object): ) try: l, llen, v = len_decode(lv) + except LenIndefForm as err: + if not ctx.get("bered", False): + raise err.__class__( + msg=err.msg, + klass=self.__class__, + decode_path=decode_path, + offset=offset, + ) + llen, v = 1, lv[1:] + offset += tlen + llen + result = self._decode( + v, + offset=offset, + decode_path=decode_path, + ctx=ctx, + tag_only=tag_only, + ) + if tag_only: # pragma: no cover + return None + obj, tail = result + eoc_expected, tail = tail[:EOC_LEN], tail[EOC_LEN:] + if eoc_expected.tobytes() != EOC: + raise DecodeError( + "no EOC", + klass=self.__class__, + decode_path=decode_path, + offset=offset, + ) + obj.vlen += EOC_LEN + obj.expl_lenindef = True except DecodeError as err: raise err.__class__( msg=err.msg, @@ -966,49 +1281,151 @@ class Obj(object): decode_path=decode_path, offset=offset, ) - if l > len(v): - raise NotEnoughData( - "encoded length is longer than data", - klass=self.__class__, + else: + if l > len(v): + raise NotEnoughData( + "encoded length is longer than data", + klass=self.__class__, + decode_path=decode_path, + offset=offset, + ) + result = self._decode( + v, + offset=offset + tlen + llen, decode_path=decode_path, - offset=offset, + ctx=ctx, + tag_only=tag_only, ) - obj, tail = self._decode( - v, - offset=offset + tlen + llen, - decode_path=decode_path, - ctx=ctx, - ) + if tag_only: # pragma: no cover + return None + obj, tail = result + if obj.tlvlen < l and not ctx.get("allow_expl_oob", False): + raise DecodeError( + "explicit tag out-of-bound, longer than data", + klass=self.__class__, + decode_path=decode_path, + offset=offset, + ) 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 + + def hexdecode(self, data, *args, **kwargs): + """Do :py:meth:`pyderasn.Obj.decode` with hexadecimal decoded data + """ + return self.decode(hexdec(data), *args, **kwargs) + + def hexdecod(self, data, *args, **kwargs): + """Do :py:meth:`pyderasn.Obj.decod` with hexadecimal decoded data + """ + return self.decod(hexdec(data), *args, **kwargs) + @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): + if self.lenindef and not ( + getattr(self, "defined", None) is not None and + self.defined[1].lenindef + ): + yield _pp( + asn1_type_name="EOC", + obj_name="", + decode_path=decode_path, + offset=( + self.offset + self.tlvlen - + (EOC_LEN * 2 if self.expl_lenindef else EOC_LEN) + ), + tlen=1, + llen=1, + vlen=0, + ber_encoded=True, + bered=True, + ) + if self.expl_lenindef: + yield _pp( + asn1_type_name="EOC", + obj_name="EXPLICIT", + decode_path=decode_path, + offset=self.expl_offset + self.expl_tlvlen - EOC_LEN, + tlen=1, + llen=1, + vlen=0, + ber_encoded=True, + bered=True, + ) + class DecodePathDefBy(object): """DEFINED BY representation inside decode path @@ -1018,6 +1435,9 @@ class DecodePathDefBy(object): def __init__(self, defined_by): self.defined_by = defined_by + def __ne__(self, their): + return not(self == their) + def __eq__(self, their): if not isinstance(their, self.__class__): return False @@ -1035,6 +1455,7 @@ class DecodePathDefBy(object): ######################################################################## PP = namedtuple("PP", ( + "obj", "asn1_type_name", "obj_name", "decode_path", @@ -1052,10 +1473,15 @@ PP = namedtuple("PP", ( "expl_tlen", "expl_llen", "expl_vlen", + "expl_lenindef", + "lenindef", + "ber_encoded", + "bered", )) def _pp( + obj=None, asn1_type_name="unknown", obj_name="unknown", decode_path=(), @@ -1073,8 +1499,13 @@ def _pp( expl_tlen=None, expl_llen=None, expl_vlen=None, + expl_lenindef=False, + lenindef=False, + ber_encoded=False, + bered=False, ): return PP( + obj, asn1_type_name, obj_name, decode_path, @@ -1092,128 +1523,196 @@ def _pp( expl_tlen, expl_llen, expl_vlen, + expl_lenindef, + lenindef, + ber_encoded, + bered, ) -def _colorize(what, colour, with_colours, attrs=("bold",)): +def _colourize(what, colour, with_colours, attrs=("bold",)): return colored(what, colour, attrs=attrs) if with_colours else what +def colonize_hex(hexed): + """Separate hexadecimal string with colons + """ + 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, + with_decode_path=False, + decode_path_len_decrease=0, ): cols = [] if with_offsets: - col = "%5d%s" % ( + col = "%5d%s%s" % ( pp.offset, ( " " if pp.expl_offset is None else ("-%d" % (pp.offset - pp.expl_offset)) ), + LENINDEF_PP_CHAR if pp.expl_lenindef else " ", ) - cols.append(_colorize(col, "red", with_colours, ())) - col = "[%d,%d,%4d]" % (pp.tlen, pp.llen, pp.vlen) - cols.append(_colorize(col, "green", with_colours, ())) - if len(pp.decode_path) > 0: - cols.append(" ." * (len(pp.decode_path))) + col = _colourize(col, "red", with_colours, ()) + col += _colourize("B", "red", with_colours) if pp.bered else " " + cols.append(col) + col = "[%d,%d,%4d]%s" % ( + pp.tlen, + pp.llen, + pp.vlen, + LENINDEF_PP_CHAR if pp.lenindef else " " + ) + col = _colourize(col, "green", with_colours, ()) + cols.append(col) + decode_path_len = len(pp.decode_path) - decode_path_len_decrease + if decode_path_len > 0: + cols.append(" ." * decode_path_len) ent = pp.decode_path[-1] if isinstance(ent, DecodePathDefBy): - cols.append(_colorize("DEFINED BY", "red", with_colours, ("reverse",))) + 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(_colorize("%s:" % oids[value], "green", with_colours)) - else: - cols.append(_colorize("%s:" % value, "white", with_colours, ("reverse",))) + 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(_colorize("%s:" % ent, "yellow", with_colours, ("reverse",))) + cols.append(_colourize("%s:" % ent, "yellow", with_colours, ("reverse",))) if pp.expl is not None: klass, _, num = pp.expl col = "[%s%d] EXPLICIT" % (TagClassReprs[klass], num) - cols.append(_colorize(col, "blue", with_colours)) + cols.append(_colourize(col, "blue", with_colours)) if pp.impl is not None: klass, _, num = pp.impl col = "[%s%d]" % (TagClassReprs[klass], num) - cols.append(_colorize(col, "blue", with_colours)) + cols.append(_colourize(col, "blue", with_colours)) if pp.asn1_type_name.replace(" ", "") != pp.obj_name.upper(): - cols.append(_colorize(pp.obj_name, "magenta", with_colours)) - cols.append(_colorize(pp.asn1_type_name, "cyan", with_colours)) + cols.append(_colourize(pp.obj_name, "magenta", with_colours)) + if pp.ber_encoded: + cols.append(_colourize("BER", "red", with_colours)) + cols.append(_colourize(pp.asn1_type_name, "cyan", with_colours)) if pp.value is not None: value = pp.value - cols.append(_colorize(value, "white", with_colours, ("reverse",))) + 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(_colorize("(%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: + hex_repr = "0" + hex_repr + cols.append(_colourize( + "(%s)" % colonize_hex(hex_repr), + "green", + with_colours, + )) if with_blob: if isinstance(pp.blob, binary_type): cols.append(hexenc(pp.blob)) elif isinstance(pp.blob, tuple): cols.append(", ".join(pp.blob)) if pp.optional: - cols.append(_colorize("OPTIONAL", "red", with_colours)) + cols.append(_colourize("OPTIONAL", "red", with_colours)) if pp.default: - cols.append(_colorize("DEFAULT", "red", with_colours)) + cols.append(_colourize("DEFAULT", "red", with_colours)) + if with_decode_path: + cols.append(_colourize( + "[%s]" % ":".join(str(p) for p in pp.decode_path), + "grey", + with_colours, + )) return " ".join(cols) -def pp_console_blob(pp): - cols = [" " * len("XXXXXYY [X,X,XXXX]")] - if len(pp.decode_path) > 0: - cols.append(" ." * (len(pp.decode_path) + 1)) +def pp_console_blob(pp, decode_path_len_decrease=0): + cols = [" " * len("XXXXXYYZZ [X,X,XXXX]Z")] + decode_path_len = len(pp.decode_path) - decode_path_len_decrease + if decode_path_len > 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 + [":".join( - chunk[j:j + 2] for j in range(0, len(chunk), 2) - )]) + yield " ".join(cols + [colonize_hex(chunk)]) elif isinstance(pp.blob, tuple): yield " ".join(cols + [", ".join(pp.blob)]) -def pprint(obj, oids=None, big_blobs=False, with_colours=False): +def pprint( + obj, + oid_maps=(), + big_blobs=False, + with_colours=False, + with_decode_path=False, + decode_path_only=(), +): """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 ``str(OID) <-> human readable string`` dictionary. + Its human readable form is printed when OID is met :param big_blobs: if large binary objects are met (like OctetString values), do we need to print them too, on separate lines :param with_colours: colourize output, if ``termcolor`` library is available + :param with_decode_path: print decode path + :param decode_path_only: print only that specified decode path """ def _pprint_pps(pps): for pp in pps: if hasattr(pp, "_fields"): + if ( + decode_path_only != () and + tuple( + str(p) for p in pp.decode_path[:len(decode_path_only)] + ) != decode_path_only + ): + continue if big_blobs: yield pp_console_row( pp, - oids=oids, + oid_maps=oid_maps, with_offsets=True, with_blob=False, with_colours=with_colours, + with_decode_path=with_decode_path, + decode_path_len_decrease=len(decode_path_only), ) - for row in pp_console_blob(pp): + for row in pp_console_blob( + pp, + decode_path_len_decrease=len(decode_path_only), + ): yield row else: yield pp_console_row( pp, - oids=oids, + oid_maps=oid_maps, with_offsets=True, with_blob=True, with_colours=with_colours, + with_decode_path=with_decode_path, + decode_path_len_decrease=len(decode_path_only), ) else: for row in _pprint_pps(pp): @@ -1225,6 +1724,22 @@ def pprint(obj, oids=None, big_blobs=False, with_colours=False): # ASN.1 primitive types ######################################################################## +BooleanState = namedtuple("BooleanState", ( + "version", + "value", + "tag", + "expl", + "default", + "optional", + "offset", + "llen", + "vlen", + "expl_lenindef", + "lenindef", + "ber_encoded", +)) + + class Boolean(Obj): """``BOOLEAN`` boolean type @@ -1269,27 +1784,45 @@ 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 def ready(self): return self._value is not None - def copy(self): - obj = self.__class__() - obj._value = self._value - obj.tag = self.tag - obj._expl = self._expl - obj.default = self.default - obj.optional = self.optional - obj.offset = self.offset - obj.llen = self.llen - obj.vlen = self.vlen - return obj + def __getstate__(self): + return BooleanState( + __version__, + self._value, + self.tag, + self._expl, + self.default, + self.optional, + self.offset, + self.llen, + self.vlen, + self.expl_lenindef, + self.lenindef, + self.ber_encoded, + ) + + def __setstate__(self, state): + super(Boolean, self).__setstate__(state) + self._value = state.value + self.tag = state.tag + self._expl = state.expl + self.default = state.default + self.optional = state.optional + self.offset = state.offset + self.llen = state.llen + self.vlen = state.vlen + self.expl_lenindef = state.expl_lenindef + self.lenindef = state.lenindef + self.ber_encoded = state.ber_encoded def __nonzero__(self): self._assert_ready() @@ -1334,7 +1867,7 @@ class Boolean(Obj): (b"\xFF" if self._value else b"\x00"), )) - def _decode(self, tlv, offset, decode_path, ctx): + def _decode(self, tlv, offset, decode_path, ctx, tag_only): try: t, _, lv = tag_strip(tlv) except DecodeError as err: @@ -1350,6 +1883,8 @@ class Boolean(Obj): decode_path=decode_path, offset=offset, ) + if tag_only: + return None try: l, _, v = len_decode(lv) except DecodeError as err: @@ -1374,10 +1909,14 @@ class Boolean(Obj): offset=offset, ) first_octet = byte2int(v) + ber_encoded = False if first_octet == 0: value = False elif first_octet == 0xFF: value = True + elif ctx.get("bered", False): + value = True + ber_encoded = True else: raise DecodeError( "unacceptable Boolean value", @@ -1393,6 +1932,7 @@ class Boolean(Obj): optional=self.optional, _decoded=(offset, 1, 1), ) + obj.ber_encoded = ber_encoded return obj, v[1:] def __repr__(self): @@ -1400,6 +1940,7 @@ class Boolean(Obj): def pps(self, decode_path=()): yield _pp( + obj=self, asn1_type_name=self.asn1_type_name, obj_name=self.__class__.__name__, decode_path=decode_path, @@ -1416,7 +1957,31 @@ class Boolean(Obj): expl_tlen=self.expl_tlen if self.expled else None, expl_llen=self.expl_llen if self.expled else None, expl_vlen=self.expl_vlen if self.expled else None, + expl_lenindef=self.expl_lenindef, + ber_encoded=self.ber_encoded, + bered=self.bered, ) + for pp in self.pps_lenindef(decode_path): + yield pp + + +IntegerState = namedtuple("IntegerState", ( + "version", + "specs", + "value", + "bound_min", + "bound_max", + "tag", + "expl", + "default", + "optional", + "offset", + "llen", + "vlen", + "expl_lenindef", + "lenindef", + "ber_encoded", +)) class Integer(Obj): @@ -1502,10 +2067,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: @@ -1520,19 +2085,41 @@ class Integer(Obj): def ready(self): return self._value is not None - def copy(self): - obj = self.__class__(_specs=self.specs) - obj._value = self._value - obj._bound_min = self._bound_min - obj._bound_max = self._bound_max - obj.tag = self.tag - obj._expl = self._expl - obj.default = self.default - obj.optional = self.optional - obj.offset = self.offset - obj.llen = self.llen - obj.vlen = self.vlen - return obj + def __getstate__(self): + return IntegerState( + __version__, + self.specs, + self._value, + self._bound_min, + self._bound_max, + self.tag, + self._expl, + self.default, + self.optional, + self.offset, + self.llen, + self.vlen, + self.expl_lenindef, + self.lenindef, + self.ber_encoded, + ) + + def __setstate__(self, state): + super(Integer, self).__setstate__(state) + self.specs = state.specs + self._value = state.value + self._bound_min = state.bound_min + self._bound_max = state.bound_max + self.tag = state.tag + self._expl = state.expl + self.default = state.default + self.optional = state.optional + self.offset = state.offset + self.llen = state.llen + self.vlen = state.vlen + self.expl_lenindef = state.expl_lenindef + self.lenindef = state.lenindef + self.ber_encoded = state.ber_encoded def __int__(self): self._assert_ready() @@ -1562,9 +2149,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, @@ -1627,7 +2215,7 @@ class Integer(Obj): break return b"".join((self.tag, len_encode(len(octets)), octets)) - def _decode(self, tlv, offset, decode_path, ctx): + def _decode(self, tlv, offset, decode_path, ctx, tag_only): try: t, _, lv = tag_strip(tlv) except DecodeError as err: @@ -1643,6 +2231,8 @@ class Integer(Obj): decode_path=decode_path, offset=offset, ) + if tag_only: + return None try: l, llen, v = len_decode(lv) except DecodeError as err: @@ -1720,6 +2310,7 @@ class Integer(Obj): def pps(self, decode_path=()): yield _pp( + obj=self, asn1_type_name=self.asn1_type_name, obj_name=self.__class__.__name__, decode_path=decode_path, @@ -1736,7 +2327,31 @@ class Integer(Obj): expl_tlen=self.expl_tlen if self.expled else None, expl_llen=self.expl_llen if self.expled else None, expl_vlen=self.expl_vlen if self.expled else None, + expl_lenindef=self.expl_lenindef, + bered=self.bered, ) + for pp in self.pps_lenindef(decode_path): + yield pp + + +SET01 = frozenset(("0", "1")) +BitStringState = namedtuple("BitStringState", ( + "version", + "specs", + "value", + "tag", + "expl", + "default", + "optional", + "offset", + "llen", + "vlen", + "expl_lenindef", + "lenindef", + "ber_encoded", + "tag_constructed", + "defined", +)) class BitString(Obj): @@ -1751,6 +2366,8 @@ class BitString(Obj): >>> b.bit_len 88 + >>> BitString("'0A3B5F291CD'H") + BIT STRING 44 bits 0a3b5f291cd0 >>> b = BitString("'010110000000'B") BIT STRING 12 bits 5800 >>> b.bit_len @@ -1777,8 +2394,16 @@ class BitString(Obj): ['nonRepudiation', 'keyEncipherment'] >>> b.specs {'nonRepudiation': 1, 'digitalSignature': 0, 'keyEncipherment': 2} + + .. note:: + + Pay attention that BIT STRING can be encoded both in primitive + and constructed forms. Decoder always checks constructed form tag + additionally to specified primitive one. If BER decoding is + :ref:`not enabled `, then decoder will fail, because + of DER restrictions. """ - __slots__ = ("specs", "defined") + __slots__ = ("tag_constructed", "specs", "defined") tag_default = tag_encode(3) asn1_type_name = "BIT STRING" @@ -1816,6 +2441,12 @@ class BitString(Obj): if value is None: self._value = default self.defined = None + tag_klass, _, tag_num = tag_decode(self.tag) + self.tag_constructed = tag_encode( + klass=tag_klass, + form=TagFormConstructed, + num=tag_num, + ) def _bits2octets(self, bits): if len(self.specs) > 0: @@ -1828,26 +2459,25 @@ 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 - value.startswith("'") and - value.endswith("'B") + value.startswith("'") ): - value = value[1:-2] - if not set(value) <= set(("0", "1")): - raise ValueError("B's coding contains unacceptable chars") - return self._bits2octets(value) - elif isinstance(value, binary_type): + if value.endswith("'B"): + value = value[1:-2] + if not frozenset(value) <= SET01: + raise ValueError("B's coding contains unacceptable chars") + return self._bits2octets(value) + if value.endswith("'H"): + value = value[1:-2] + return ( + len(value) * 4, + hexdec(value + ("" if len(value) % 2 == 0 else "0")), + ) + 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 @@ -1863,31 +2493,54 @@ 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 def ready(self): return self._value is not None - def copy(self): - obj = self.__class__(_specs=self.specs) - value = self._value - if value is not None: - value = (value[0], value[1]) - obj._value = value - obj.tag = self.tag - obj._expl = self._expl - obj.default = self.default - obj.optional = self.optional - obj.offset = self.offset - obj.llen = self.llen - obj.vlen = self.vlen - return obj + def __getstate__(self): + return BitStringState( + __version__, + self.specs, + self._value, + self.tag, + self._expl, + self.default, + self.optional, + self.offset, + self.llen, + self.vlen, + self.expl_lenindef, + self.lenindef, + self.ber_encoded, + self.tag_constructed, + self.defined, + ) + + def __setstate__(self, state): + super(BitString, self).__setstate__(state) + self.specs = state.specs + self._value = state.value + self.tag = state.tag + self._expl = state.expl + self.default = state.default + self.optional = state.optional + self.offset = state.offset + self.llen = state.llen + self.vlen = state.vlen + self.expl_lenindef = state.expl_lenindef + self.lenindef = state.lenindef + self.ber_encoded = state.ber_encoded + self.tag_constructed = state.tag_constructed + self.defined = state.defined def __iter__(self): self._assert_ready() @@ -1916,7 +2569,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, @@ -1961,22 +2614,7 @@ class BitString(Obj): octets, )) - def _decode(self, tlv, offset, decode_path, ctx): - try: - t, _, lv = tag_strip(tlv) - except DecodeError as err: - raise err.__class__( - msg=err.msg, - klass=self.__class__, - decode_path=decode_path, - offset=offset, - ) - if t != self.tag: - raise TagMismatch( - klass=self.__class__, - decode_path=decode_path, - offset=offset, - ) + def _decode_chunk(self, lv, offset, decode_path): try: l, llen, v = len_decode(lv) except DecodeError as err: @@ -2034,41 +2672,197 @@ class BitString(Obj): ) return obj, tail - def __repr__(self): - return pp_console_row(next(self.pps())) - - def pps(self, decode_path=()): - value = None - blob = None - if self.ready: - bit_len, blob = self._value - value = "%d bits" % bit_len - if len(self.specs) > 0: - blob = tuple(self.named) - yield _pp( - asn1_type_name=self.asn1_type_name, - obj_name=self.__class__.__name__, - decode_path=decode_path, - value=value, - blob=blob, - optional=self.optional, - default=self == self.default, - impl=None if self.tag == self.tag_default else tag_decode(self.tag), - expl=None if self._expl is None else tag_decode(self._expl), - offset=self.offset, - tlen=self.tlen, - llen=self.llen, - vlen=self.vlen, - expl_offset=self.expl_offset if self.expled else None, - expl_tlen=self.expl_tlen if self.expled else None, - expl_llen=self.expl_llen if self.expled else None, - expl_vlen=self.expl_vlen if self.expled else None, - ) - defined_by, defined = self.defined or (None, None) - if defined_by is not None: - yield defined.pps( + def _decode(self, tlv, offset, decode_path, ctx, tag_only): + try: + t, tlen, lv = tag_strip(tlv) + except DecodeError as err: + raise err.__class__( + msg=err.msg, + klass=self.__class__, + decode_path=decode_path, + offset=offset, + ) + if t == self.tag: + if tag_only: # pragma: no cover + return None + return self._decode_chunk(lv, offset, decode_path) + if t == self.tag_constructed: + if not ctx.get("bered", False): + raise DecodeError( + "unallowed BER constructed encoding", + klass=self.__class__, + decode_path=decode_path, + offset=offset, + ) + if tag_only: # pragma: no cover + return None + lenindef = False + try: + l, llen, v = len_decode(lv) + except LenIndefForm: + llen, l, v = 1, 0, lv[1:] + lenindef = True + except DecodeError as err: + raise err.__class__( + msg=err.msg, + klass=self.__class__, + decode_path=decode_path, + offset=offset, + ) + if l > len(v): + raise NotEnoughData( + "encoded length is longer than data", + klass=self.__class__, + decode_path=decode_path, + offset=offset, + ) + if not lenindef and l == 0: + raise NotEnoughData( + "zero length", + klass=self.__class__, + decode_path=decode_path, + offset=offset, + ) + chunks = [] + sub_offset = offset + tlen + llen + vlen = 0 + while True: + if lenindef: + if v[:EOC_LEN].tobytes() == EOC: + break + else: + if vlen == l: + break + if vlen > l: + raise DecodeError( + "chunk out of bounds", + klass=self.__class__, + decode_path=decode_path + (str(len(chunks) - 1),), + offset=chunks[-1].offset, + ) + sub_decode_path = decode_path + (str(len(chunks)),) + try: + chunk, v_tail = BitString().decode( + v, + offset=sub_offset, + decode_path=sub_decode_path, + leavemm=True, + ctx=ctx, + _ctx_immutable=False, + ) + except TagMismatch: + raise DecodeError( + "expected BitString encoded chunk", + klass=self.__class__, + decode_path=sub_decode_path, + offset=sub_offset, + ) + chunks.append(chunk) + sub_offset += chunk.tlvlen + vlen += chunk.tlvlen + v = v_tail + if len(chunks) == 0: + raise DecodeError( + "no chunks", + klass=self.__class__, + decode_path=decode_path, + offset=offset, + ) + values = [] + bit_len = 0 + for chunk_i, chunk in enumerate(chunks[:-1]): + if chunk.bit_len % 8 != 0: + raise DecodeError( + "BitString chunk is not multiple of 8 bits", + klass=self.__class__, + decode_path=decode_path + (str(chunk_i),), + offset=chunk.offset, + ) + values.append(bytes(chunk)) + bit_len += chunk.bit_len + chunk_last = chunks[-1] + values.append(bytes(chunk_last)) + bit_len += chunk_last.bit_len + obj = self.__class__( + value=(bit_len, b"".join(values)), + impl=self.tag, + expl=self._expl, + default=self.default, + optional=self.optional, + _specs=self.specs, + _decoded=(offset, llen, vlen + (EOC_LEN if lenindef else 0)), + ) + obj.lenindef = lenindef + obj.ber_encoded = True + return obj, (v[EOC_LEN:] if lenindef else v) + raise TagMismatch( + klass=self.__class__, + decode_path=decode_path, + offset=offset, + ) + + def __repr__(self): + return pp_console_row(next(self.pps())) + + def pps(self, decode_path=()): + value = None + blob = None + if self.ready: + bit_len, blob = self._value + value = "%d bits" % bit_len + if len(self.specs) > 0: + blob = tuple(self.named) + yield _pp( + obj=self, + asn1_type_name=self.asn1_type_name, + obj_name=self.__class__.__name__, + decode_path=decode_path, + value=value, + blob=blob, + optional=self.optional, + default=self == self.default, + impl=None if self.tag == self.tag_default else tag_decode(self.tag), + expl=None if self._expl is None else tag_decode(self._expl), + offset=self.offset, + tlen=self.tlen, + llen=self.llen, + vlen=self.vlen, + expl_offset=self.expl_offset if self.expled else None, + expl_tlen=self.expl_tlen if self.expled else None, + expl_llen=self.expl_llen if self.expled else None, + expl_vlen=self.expl_vlen if self.expled else None, + expl_lenindef=self.expl_lenindef, + lenindef=self.lenindef, + ber_encoded=self.ber_encoded, + bered=self.bered, + ) + defined_by, defined = self.defined or (None, None) + if defined_by is not None: + yield defined.pps( decode_path=decode_path + (DecodePathDefBy(defined_by),) ) + for pp in self.pps_lenindef(decode_path): + yield pp + + +OctetStringState = namedtuple("OctetStringState", ( + "version", + "value", + "bound_min", + "bound_max", + "tag", + "expl", + "default", + "optional", + "offset", + "llen", + "vlen", + "expl_lenindef", + "lenindef", + "ber_encoded", + "tag_constructed", + "defined", +)) class OctetString(Obj): @@ -2086,8 +2880,16 @@ class OctetString(Obj): pyderasn.BoundsError: unsatisfied bounds: 4 <= 5 <= 4 >>> OctetString(b"hell", bounds=(4, 4)) OCTET STRING 4 bytes 68656c6c + + .. note:: + + Pay attention that OCTET STRING can be encoded both in primitive + and constructed forms. Decoder always checks constructed form tag + additionally to specified primitive one. If BER decoding is + :ref:`not enabled `, then decoder will fail, because + of DER restrictions. """ - __slots__ = ("_bound_min", "_bound_max", "defined") + __slots__ = ("tag_constructed", "_bound_min", "_bound_max", "defined") tag_default = tag_encode(4) asn1_type_name = "OCTET STRING" @@ -2111,13 +2913,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, @@ -2136,12 +2932,18 @@ class OctetString(Obj): if self._value is None: self._value = default self.defined = None + tag_klass, _, tag_num = tag_decode(self.tag) + self.tag_constructed = tag_encode( + klass=tag_klass, + form=TagFormConstructed, + num=tag_num, + ) 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: @@ -2152,19 +2954,43 @@ class OctetString(Obj): def ready(self): return self._value is not None - def copy(self): - obj = self.__class__() - obj._value = self._value - obj._bound_min = self._bound_min - obj._bound_max = self._bound_max - obj.tag = self.tag - obj._expl = self._expl - obj.default = self.default - obj.optional = self.optional - obj.offset = self.offset - obj.llen = self.llen - obj.vlen = self.vlen - return obj + def __getstate__(self): + return OctetStringState( + __version__, + self._value, + self._bound_min, + self._bound_max, + self.tag, + self._expl, + self.default, + self.optional, + self.offset, + self.llen, + self.vlen, + self.expl_lenindef, + self.lenindef, + self.ber_encoded, + self.tag_constructed, + self.defined, + ) + + def __setstate__(self, state): + super(OctetString, self).__setstate__(state) + self._value = state.value + self._bound_min = state.bound_min + self._bound_max = state.bound_max + self.tag = state.tag + self._expl = state.expl + self.default = state.default + self.optional = state.optional + self.offset = state.offset + self.llen = state.llen + self.vlen = state.vlen + self.expl_lenindef = state.expl_lenindef + self.lenindef = state.lenindef + self.ber_encoded = state.ber_encoded + self.tag_constructed = state.tag_constructed + self.defined = state.defined def __bytes__(self): self._assert_ready() @@ -2213,22 +3039,7 @@ class OctetString(Obj): self._value, )) - def _decode(self, tlv, offset, decode_path, ctx): - try: - t, _, lv = tag_strip(tlv) - except DecodeError as err: - raise err.__class__( - msg=err.msg, - klass=self.__class__, - decode_path=decode_path, - offset=offset, - ) - if t != self.tag: - raise TagMismatch( - klass=self.__class__, - decode_path=decode_path, - offset=offset, - ) + def _decode_chunk(self, lv, offset, decode_path): try: l, llen, v = len_decode(lv) except DecodeError as err: @@ -2272,11 +3083,127 @@ class OctetString(Obj): ) return obj, tail + def _decode(self, tlv, offset, decode_path, ctx, tag_only): + try: + t, tlen, lv = tag_strip(tlv) + except DecodeError as err: + raise err.__class__( + msg=err.msg, + klass=self.__class__, + decode_path=decode_path, + offset=offset, + ) + if t == self.tag: + if tag_only: + return None + return self._decode_chunk(lv, offset, decode_path) + if t == self.tag_constructed: + if not ctx.get("bered", False): + raise DecodeError( + "unallowed BER constructed encoding", + klass=self.__class__, + decode_path=decode_path, + offset=offset, + ) + if tag_only: + return None + lenindef = False + try: + l, llen, v = len_decode(lv) + except LenIndefForm: + llen, l, v = 1, 0, lv[1:] + lenindef = True + except DecodeError as err: + raise err.__class__( + msg=err.msg, + klass=self.__class__, + decode_path=decode_path, + offset=offset, + ) + if l > len(v): + raise NotEnoughData( + "encoded length is longer than data", + klass=self.__class__, + decode_path=decode_path, + offset=offset, + ) + chunks = [] + sub_offset = offset + tlen + llen + vlen = 0 + while True: + if lenindef: + if v[:EOC_LEN].tobytes() == EOC: + break + else: + if vlen == l: + break + if vlen > l: + raise DecodeError( + "chunk out of bounds", + klass=self.__class__, + decode_path=decode_path + (str(len(chunks) - 1),), + offset=chunks[-1].offset, + ) + sub_decode_path = decode_path + (str(len(chunks)),) + try: + chunk, v_tail = OctetString().decode( + v, + offset=sub_offset, + decode_path=sub_decode_path, + leavemm=True, + ctx=ctx, + _ctx_immutable=False, + ) + except TagMismatch: + raise DecodeError( + "expected OctetString encoded chunk", + klass=self.__class__, + decode_path=sub_decode_path, + offset=sub_offset, + ) + chunks.append(chunk) + sub_offset += chunk.tlvlen + vlen += chunk.tlvlen + v = v_tail + try: + obj = self.__class__( + value=b"".join(bytes(chunk) for chunk in chunks), + bounds=(self._bound_min, self._bound_max), + impl=self.tag, + expl=self._expl, + default=self.default, + optional=self.optional, + _decoded=(offset, llen, vlen + (EOC_LEN if lenindef else 0)), + ) + except DecodeError as err: + raise DecodeError( + msg=err.msg, + klass=self.__class__, + decode_path=decode_path, + offset=offset, + ) + except BoundsError as err: + raise DecodeError( + msg=str(err), + klass=self.__class__, + decode_path=decode_path, + offset=offset, + ) + obj.lenindef = lenindef + obj.ber_encoded = True + return obj, (v[EOC_LEN:] if lenindef else v) + raise TagMismatch( + klass=self.__class__, + decode_path=decode_path, + offset=offset, + ) + def __repr__(self): return pp_console_row(next(self.pps())) def pps(self, decode_path=()): yield _pp( + obj=self, asn1_type_name=self.asn1_type_name, obj_name=self.__class__.__name__, decode_path=decode_path, @@ -2294,12 +3221,33 @@ class OctetString(Obj): expl_tlen=self.expl_tlen if self.expled else None, expl_llen=self.expl_llen if self.expled else None, expl_vlen=self.expl_vlen if self.expled else None, + expl_lenindef=self.expl_lenindef, + lenindef=self.lenindef, + ber_encoded=self.ber_encoded, + bered=self.bered, ) defined_by, defined = self.defined or (None, None) if defined_by is not None: yield defined.pps( decode_path=decode_path + (DecodePathDefBy(defined_by),) ) + for pp in self.pps_lenindef(decode_path): + yield pp + + +NullState = namedtuple("NullState", ( + "version", + "tag", + "expl", + "default", + "optional", + "offset", + "llen", + "vlen", + "expl_lenindef", + "lenindef", + "ber_encoded", +)) class Null(Obj): @@ -2334,16 +3282,33 @@ class Null(Obj): def ready(self): return True - def copy(self): - obj = self.__class__() - obj.tag = self.tag - obj._expl = self._expl - obj.default = self.default - obj.optional = self.optional - obj.offset = self.offset - obj.llen = self.llen - obj.vlen = self.vlen - return obj + def __getstate__(self): + return NullState( + __version__, + self.tag, + self._expl, + self.default, + self.optional, + self.offset, + self.llen, + self.vlen, + self.expl_lenindef, + self.lenindef, + self.ber_encoded, + ) + + def __setstate__(self, state): + super(Null, self).__setstate__(state) + self.tag = state.tag + self._expl = state.expl + self.default = state.default + self.optional = state.optional + self.offset = state.offset + self.llen = state.llen + self.vlen = state.vlen + self.expl_lenindef = state.expl_lenindef + self.lenindef = state.lenindef + self.ber_encoded = state.ber_encoded def __eq__(self, their): if not issubclass(their.__class__, Null): @@ -2369,7 +3334,7 @@ class Null(Obj): def _encode(self): return self.tag + len_encode(0) - def _decode(self, tlv, offset, decode_path, ctx): + def _decode(self, tlv, offset, decode_path, ctx, tag_only): try: t, _, lv = tag_strip(tlv) except DecodeError as err: @@ -2385,6 +3350,8 @@ class Null(Obj): decode_path=decode_path, offset=offset, ) + if tag_only: # pragma: no cover + return None try: l, _, v = len_decode(lv) except DecodeError as err: @@ -2414,6 +3381,7 @@ class Null(Obj): def pps(self, decode_path=()): yield _pp( + obj=self, asn1_type_name=self.asn1_type_name, obj_name=self.__class__.__name__, decode_path=decode_path, @@ -2428,7 +3396,28 @@ class Null(Obj): expl_tlen=self.expl_tlen if self.expled else None, expl_llen=self.expl_llen if self.expled else None, expl_vlen=self.expl_vlen if self.expled else None, + expl_lenindef=self.expl_lenindef, + bered=self.bered, ) + for pp in self.pps_lenindef(decode_path): + yield pp + + +ObjectIdentifierState = namedtuple("ObjectIdentifierState", ( + "version", + "value", + "tag", + "expl", + "default", + "optional", + "offset", + "llen", + "vlen", + "expl_lenindef", + "lenindef", + "ber_encoded", + "defines", +)) class ObjectIdentifier(Obj): @@ -2481,13 +3470,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) @@ -2535,18 +3518,37 @@ class ObjectIdentifier(Obj): def ready(self): return self._value is not None - def copy(self): - obj = self.__class__() - obj._value = self._value - obj.defines = self.defines - obj.tag = self.tag - obj._expl = self._expl - obj.default = self.default - obj.optional = self.optional - obj.offset = self.offset - obj.llen = self.llen - obj.vlen = self.vlen - return obj + def __getstate__(self): + return ObjectIdentifierState( + __version__, + self._value, + self.tag, + self._expl, + self.default, + self.optional, + self.offset, + self.llen, + self.vlen, + self.expl_lenindef, + self.lenindef, + self.ber_encoded, + self.defines, + ) + + def __setstate__(self, state): + super(ObjectIdentifier, self).__setstate__(state) + self._value = state.value + self.tag = state.tag + self._expl = state.expl + self.default = state.default + self.optional = state.optional + self.offset = state.offset + self.llen = state.llen + self.vlen = state.vlen + self.expl_lenindef = state.expl_lenindef + self.lenindef = state.lenindef + self.ber_encoded = state.ber_encoded + self.defines = state.defines def __iter__(self): self._assert_ready() @@ -2614,7 +3616,7 @@ class ObjectIdentifier(Obj): v = b"".join(octets) return b"".join((self.tag, len_encode(len(v)), v)) - def _decode(self, tlv, offset, decode_path, ctx): + def _decode(self, tlv, offset, decode_path, ctx, tag_only): try: t, _, lv = tag_strip(tlv) except DecodeError as err: @@ -2630,6 +3632,8 @@ class ObjectIdentifier(Obj): decode_path=decode_path, offset=offset, ) + if tag_only: # pragma: no cover + return None try: l, llen, v = len_decode(lv) except DecodeError as err: @@ -2655,11 +3659,17 @@ class ObjectIdentifier(Obj): ) v, tail = v[:l], v[l:] arcs = [] + ber_encoded = False while len(v) > 0: i = 0 arc = 0 while True: octet = indexbytes(v, i) + if i == 0 and octet == 0x80: + if ctx.get("bered", False): + ber_encoded = True + else: + raise DecodeError("non normalized arc encoding") arc = (arc << 7) | (octet & 0x7F) if octet & 0x80 == 0: arcs.append(arc) @@ -2691,6 +3701,8 @@ class ObjectIdentifier(Obj): optional=self.optional, _decoded=(offset, llen, l), ) + if ber_encoded: + obj.ber_encoded = True return obj, tail def __repr__(self): @@ -2698,6 +3710,7 @@ class ObjectIdentifier(Obj): def pps(self, decode_path=()): yield _pp( + obj=self, asn1_type_name=self.asn1_type_name, obj_name=self.__class__.__name__, decode_path=decode_path, @@ -2714,7 +3727,12 @@ class ObjectIdentifier(Obj): expl_tlen=self.expl_tlen if self.expled else None, expl_llen=self.expl_llen if self.expled else None, expl_vlen=self.expl_vlen if self.expled else None, + expl_lenindef=self.expl_lenindef, + ber_encoded=self.ber_encoded, + bered=self.bered, ) + for pp in self.pps_lenindef(decode_path): + yield pp class Enumerated(Integer): @@ -2739,13 +3757,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") @@ -2754,7 +3766,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__, @@ -2767,20 +3782,6 @@ class Enumerated(Integer): raise InvalidValueType((self.__class__, int, str)) return value - def copy(self): - obj = self.__class__(_specs=self.specs) - obj._value = self._value - obj._bound_min = self._bound_min - obj._bound_max = self._bound_max - obj.tag = self.tag - obj._expl = self._expl - obj.default = self.default - obj.optional = self.optional - obj.offset = self.offset - obj.llen = self.llen - obj.vlen = self.vlen - return obj - def __call__( self, value=None, @@ -2800,6 +3801,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 @@ -2862,7 +3869,7 @@ class CommonString(OctetString): * - :py:class:`pyderasn.BMPString` - utf-16-be """ - __slots__ = ("encoding",) + __slots__ = () def _value_sanitize(self, value): value_raw = None @@ -2918,8 +3925,12 @@ 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, obj_name=self.__class__.__name__, decode_path=decode_path, @@ -2932,7 +3943,16 @@ class CommonString(OctetString): tlen=self.tlen, llen=self.llen, vlen=self.vlen, + expl_offset=self.expl_offset if self.expled else None, + expl_tlen=self.expl_tlen if self.expled else None, + expl_llen=self.expl_llen if self.expled else None, + expl_vlen=self.expl_vlen if self.expled else None, + expl_lenindef=self.expl_lenindef, + ber_encoded=self.ber_encoded, + bered=self.bered, ) + for pp in self.pps_lenindef(decode_path): + yield pp class UTF8String(CommonString): @@ -2942,25 +3962,138 @@ class UTF8String(CommonString): asn1_type_name = "UTF8String" -class NumericString(CommonString): +class AllowableCharsMixin(object): + @property + def allowable_chars(self): + if PY2: + return self._allowable_chars + return frozenset(six_unichr(c) for c in self._allowable_chars) + + +class NumericString(AllowableCharsMixin, CommonString): + """Numeric string + + Its value is properly sanitized: only ASCII digits with spaces can + be stored. + + >>> NumericString().allowable_chars + 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")) + _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 -class PrintableString(CommonString): +PrintableStringState = namedtuple( + "PrintableStringState", + OctetStringState._fields + ("allowable_chars",), +) + + +class PrintableString(AllowableCharsMixin, CommonString): + """Printable string + + Its value is properly sanitized: see X.680 41.4 table 10. + + >>> PrintableString().allowable_chars + frozenset([' ', "'", ..., 'z']) + >>> obj = PrintableString("foo*bar", allow_asterisk=True) + PrintableString PrintableString foo*bar + >>> obj.allow_asterisk, obj.allow_ampersand + (True, False) + """ __slots__ = () tag_default = tag_encode(19) encoding = "ascii" asn1_type_name = "PrintableString" + _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, + ) + + @property + def allow_asterisk(self): + """Is asterisk character allowed? + """ + return self._asterisk <= self._allowable_chars + + @property + def allow_ampersand(self): + """Is ampersand character allowed? + """ + return self._ampersand <= self._allowable_chars + + def _value_sanitize(self, value): + value = super(PrintableString, self)._value_sanitize(value) + if not frozenset(value) <= self._allowable_chars: + raise DecodeError("non-printable value") + return value + + def __getstate__(self): + return PrintableStringState( + *super(PrintableString, self).__getstate__(), + **{"allowable_chars": self._allowable_chars} + ) + + def __setstate__(self, state): + super(PrintableString, self).__setstate__(state) + self._allowable_chars = state.allowable_chars + + 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.allow_asterisk, + allow_ampersand=self.allow_ampersand, + ) class TeletexString(CommonString): @@ -3007,14 +4140,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, @@ -3034,11 +4169,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: @@ -3053,21 +4184,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, binary_type): + try: + value_decoded = value.decode("ascii") + except (UnicodeEncodeError, UnicodeDecodeError) as err: + 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(self.fmt).encode("ascii") - if isinstance(value, binary_type): - value_decoded = value.decode("ascii") - if len(value_decoded) == LEN_YYMMDDHHMMSSZ: - try: - datetime.strptime(value_decoded, self.fmt) - except ValueError: - raise DecodeError("invalid UTCTime format") - return value - else: - raise DecodeError("invalid UTCTime length") + return value.strftime("%y%m%d%H%M%SZ").encode("ascii") raise InvalidValueType((self.__class__, datetime)) def __eq__(self, their): @@ -3092,7 +4238,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), @@ -3108,6 +4254,7 @@ class UTCTime(CommonString): def pps(self, decode_path=()): yield _pp( + obj=self, asn1_type_name=self.asn1_type_name, obj_name=self.__class__.__name__, decode_path=decode_path, @@ -3120,7 +4267,16 @@ class UTCTime(CommonString): tlen=self.tlen, llen=self.llen, vlen=self.vlen, + expl_offset=self.expl_offset if self.expled else None, + expl_tlen=self.expl_tlen if self.expled else None, + expl_llen=self.expl_llen if self.expled else None, + expl_vlen=self.expl_vlen if self.expled else None, + expl_lenindef=self.expl_lenindef, + ber_encoded=self.ber_encoded, + bered=self.bered, ) + for pp in self.pps_lenindef(decode_path): + yield pp class GeneralizedTime(UTCTime): @@ -3134,51 +4290,89 @@ 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. + + .. warning:: + + Zero year is unsupported. """ __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): - value_decoded = value.decode("ascii") - if len(value_decoded) == LEN_YYYYMMDDHHMMSSZ: - try: - datetime.strptime(value_decoded, self.fmt) - except 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 ValueError: - raise DecodeError( - "invalid GeneralizedTime (with ms) format", - ) - return value - else: + try: + value_decoded = value.decode("ascii") + except (UnicodeEncodeError, UnicodeDecodeError) as err: + 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): @@ -3221,6 +4415,23 @@ class BMPString(CommonString): asn1_type_name = "BMPString" +ChoiceState = namedtuple("ChoiceState", ( + "version", + "specs", + "value", + "tag", + "expl", + "default", + "optional", + "offset", + "llen", + "vlen", + "expl_lenindef", + "lenindef", + "ber_encoded", +)) + + class Choice(Obj): """``CHOICE`` special type @@ -3294,11 +4505,9 @@ class Choice(Obj): default_obj._value = default_value self.default = default_obj if value is None: - self._value = default_obj.copy()._value + self._value = copy(default_obj._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) @@ -3307,24 +4516,51 @@ 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 def ready(self): return self._value is not None and self._value[1].ready - def copy(self): - obj = self.__class__(schema=self.specs) - obj._expl = self._expl - obj.default = self.default - obj.optional = self.optional - obj.offset = self.offset - obj.llen = self.llen - obj.vlen = self.vlen - value = self._value - if value is not None: - obj._value = (value[0], value[1].copy()) - return obj + @property + def bered(self): + return self.expl_lenindef or ( + (self._value is not None) and + self._value[1].bered + ) + + def __getstate__(self): + return ChoiceState( + __version__, + self.specs, + copy(self._value), + self.tag, + self._expl, + self.default, + self.optional, + self.offset, + self.llen, + self.vlen, + self.expl_lenindef, + self.lenindef, + self.ber_encoded, + ) + + def __setstate__(self, state): + super(Choice, self).__setstate__(state) + self.specs = state.specs + self._value = state.value + self._expl = state.expl + self.default = state.default + self.optional = state.optional + self.offset = state.offset + self.llen = state.llen + self.vlen = state.vlen + self.expl_lenindef = state.expl_lenindef + self.lenindef = state.lenindef + self.ber_encoded = state.ber_encoded def __eq__(self, their): if isinstance(their, tuple) and len(their) == 2: @@ -3391,32 +4627,47 @@ class Choice(Obj): self._assert_ready() return self._value[1].encode() - def _decode(self, tlv, offset, decode_path, ctx): - for choice, spec in self.specs.items(): + def _decode(self, tlv, offset, decode_path, ctx, tag_only): + for choice, spec in iteritems(self.specs): + sub_decode_path = decode_path + (choice,) try: - value, tail = spec.decode( + spec.decode( tlv, offset=offset, leavemm=True, - decode_path=decode_path + (choice,), + decode_path=sub_decode_path, ctx=ctx, + tag_only=True, + _ctx_immutable=False, ) except TagMismatch: continue - obj = self.__class__( - schema=self.specs, - expl=self._expl, - default=self.default, - optional=self.optional, - _decoded=(offset, 0, value.tlvlen), + break + else: + raise TagMismatch( + klass=self.__class__, + decode_path=decode_path, + offset=offset, ) - obj._value = (choice, value) - return obj, tail - raise TagMismatch( - klass=self.__class__, - decode_path=decode_path, + if tag_only: # pragma: no cover + return None + value, tail = spec.decode( + tlv, offset=offset, + leavemm=True, + decode_path=sub_decode_path, + ctx=ctx, + _ctx_immutable=False, + ) + obj = self.__class__( + schema=self.specs, + expl=self._expl, + default=self.default, + optional=self.optional, + _decoded=(offset, 0, value.fulllen), ) + obj._value = (choice, value) + return obj, tail def __repr__(self): value = pp_console_row(next(self.pps())) @@ -3426,6 +4677,7 @@ class Choice(Obj): def pps(self, decode_path=()): yield _pp( + obj=self, asn1_type_name=self.asn1_type_name, obj_name=self.__class__.__name__, decode_path=decode_path, @@ -3438,9 +4690,13 @@ class Choice(Obj): tlen=self.tlen, llen=self.llen, vlen=self.vlen, + expl_lenindef=self.expl_lenindef, + bered=self.bered, ) if self.ready: yield self.value.pps(decode_path=decode_path + (self.choice,)) + for pp in self.pps_lenindef(decode_path): + yield pp class PrimitiveTypes(Choice): @@ -3448,9 +4704,9 @@ class PrimitiveTypes(Choice): It could be useful for general decoding of some unspecified values: - >>> PrimitiveTypes().decode(hexdec("0403666f6f"))[0].value + >>> PrimitiveTypes().decod(hexdec("0403666f6f")).value OCTET STRING 3 bytes 666f6f - >>> PrimitiveTypes().decode(hexdec("0203123456"))[0].value + >>> PrimitiveTypes().decod(hexdec("0203123456")).value INTEGER 1193046 """ __slots__ = () @@ -3478,6 +4734,22 @@ class PrimitiveTypes(Choice): )) +AnyState = namedtuple("AnyState", ( + "version", + "value", + "tag", + "expl", + "optional", + "offset", + "llen", + "vlen", + "expl_lenindef", + "lenindef", + "ber_encoded", + "defined", +)) + + class Any(Obj): """``ANY`` special type @@ -3512,28 +4784,55 @@ 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 def ready(self): return self._value is not None - def copy(self): - obj = self.__class__() - obj._value = self._value - obj.tag = self.tag - obj._expl = self._expl - obj.optional = self.optional - obj.offset = self.offset - obj.llen = self.llen - obj.vlen = self.vlen - return obj + @property + def bered(self): + if self.expl_lenindef or self.lenindef: + return True + if self.defined is None: + return False + return self.defined[1].bered + + def __getstate__(self): + return AnyState( + __version__, + self._value, + self.tag, + self._expl, + self.optional, + self.offset, + self.llen, + self.vlen, + self.expl_lenindef, + self.lenindef, + self.ber_encoded, + self.defined, + ) + + def __setstate__(self, state): + super(Any, self).__setstate__(state) + self._value = state.value + self.tag = state.tag + self._expl = state.expl + self.optional = state.optional + self.offset = state.offset + self.llen = state.llen + self.vlen = state.vlen + self.expl_lenindef = state.expl_lenindef + self.lenindef = state.lenindef + self.ber_encoded = state.ber_encoded + self.defined = state.defined def __eq__(self, their): if isinstance(their, binary_type): @@ -3566,10 +4865,51 @@ class Any(Obj): self._assert_ready() return self._value - def _decode(self, tlv, offset, decode_path, ctx): + def _decode(self, tlv, offset, decode_path, ctx, tag_only): try: t, tlen, lv = tag_strip(tlv) + except DecodeError as err: + raise err.__class__( + msg=err.msg, + klass=self.__class__, + decode_path=decode_path, + offset=offset, + ) + try: l, llen, v = len_decode(lv) + except LenIndefForm as err: + if not ctx.get("bered", False): + raise err.__class__( + msg=err.msg, + klass=self.__class__, + decode_path=decode_path, + offset=offset, + ) + llen, vlen, v = 1, 0, lv[1:] + sub_offset = offset + tlen + llen + chunk_i = 0 + while v[:EOC_LEN].tobytes() != EOC: + chunk, v = Any().decode( + v, + offset=sub_offset, + decode_path=decode_path + (str(chunk_i),), + leavemm=True, + ctx=ctx, + _ctx_immutable=False, + ) + vlen += chunk.tlvlen + sub_offset += chunk.tlvlen + chunk_i += 1 + tlvlen = tlen + llen + vlen + EOC_LEN + obj = self.__class__( + value=tlv[:tlvlen].tobytes(), + expl=self._expl, + optional=self.optional, + _decoded=(offset, 0, tlvlen), + ) + obj.lenindef = True + obj.tag = t.tobytes() + return obj, v[EOC_LEN:] except DecodeError as err: raise err.__class__( msg=err.msg, @@ -3592,7 +4932,7 @@ class Any(Obj): optional=self.optional, _decoded=(offset, 0, tlvlen), ) - obj.tag = t + obj.tag = t.tobytes() return obj, tail def __repr__(self): @@ -3600,6 +4940,7 @@ class Any(Obj): def pps(self, decode_path=()): yield _pp( + obj=self, asn1_type_name=self.asn1_type_name, obj_name=self.__class__.__name__, decode_path=decode_path, @@ -3616,12 +4957,17 @@ class Any(Obj): expl_tlen=self.expl_tlen if self.expled else None, expl_llen=self.expl_llen if self.expled else None, expl_vlen=self.expl_vlen if self.expled else None, + expl_lenindef=self.expl_lenindef, + lenindef=self.lenindef, + bered=self.bered, ) defined_by, defined = self.defined or (None, None) if defined_by is not None: yield defined.pps( decode_path=decode_path + (DecodePathDefBy(defined_by),) ) + for pp in self.pps_lenindef(decode_path): + yield pp ######################################################################## @@ -3644,8 +4990,7 @@ def get_def_by_path(defines_by_path, sub_decode_path): def abs_decode_path(decode_path, rel_path): """Create an absolute decode path from current and relative ones - :param decode_path: current decode path, starting point. - Tuple of strings + :param decode_path: current decode path, starting point. Tuple of strings :param rel_path: relative path to ``decode_path``. Tuple of strings. If first tuple's element is "/", then treat it as an absolute path, ignoring ``decode_path`` as @@ -3667,6 +5012,23 @@ def abs_decode_path(decode_path, rel_path): return decode_path + rel_path +SequenceState = namedtuple("SequenceState", ( + "version", + "specs", + "value", + "tag", + "expl", + "default", + "optional", + "offset", + "llen", + "vlen", + "expl_lenindef", + "lenindef", + "ber_encoded", +)) + + class Sequence(Obj): """``SEQUENCE`` structure type @@ -3693,7 +5055,7 @@ class Sequence(Obj): pyderasn.InvalidValueType: invalid value type, expected: >>> ext["extnID"] = ObjectIdentifier("1.2.3") - You can know if sequence is ready to be encoded: + You can determine if sequence is ready to be encoded: >>> ext.ready False @@ -3719,7 +5081,15 @@ class Sequence(Obj): Assign ``None`` to remove value from sequence. - You can know if value exists/set in the sequence and take its value: + You can set values in Sequence during its initialization: + + >>> AlgorithmIdentifier(( + ("algorithm", ObjectIdentifier("1.2.3")), + ("parameters", Any(Null())) + )) + AlgorithmIdentifier SEQUENCE[algorithm: OBJECT IDENTIFIER 1.2.3; parameters: ANY 0500 OPTIONAL] + + You can determine if value exists/set in the sequence and take its value: >>> "extnID" in ext, "extnValue" in ext, "critical" in ext (True, True, False) @@ -3738,18 +5108,14 @@ class Sequence(Obj): All defaulted values are always optional. - .. _strict_default_existence_ctx: - - .. warning:: + .. _allow_default_values_ctx: - When decoded DER contains defaulted value inside, then - technically this is not valid DER encoding. But we allow and pass - it **by default**. Of course reencoding of that kind of DER will - result in different binary representation (validly without - defaulted value inside). You can enable strict defaulted values - existence validation by setting ``"strict_default_existence": - True`` :ref:`context ` option -- decoding process will raise - an exception if defaulted value is met. + DER prohibits default value encoding and will raise an error if + default value is unexpectedly met during decode. + If :ref:`bered ` context option is set, then no error + will be raised, but ``bered`` attribute set. You can disable strict + defaulted values existence validation by setting + ``"allow_default_values": True`` :ref:`context ` option. Two sequences are equal if they have equal specification (schema), implicit/explicit tagging and the same values. @@ -3776,45 +5142,73 @@ class Sequence(Obj): ) self._value = {} if value is not None: - self._value = self._value_sanitize(value) + if issubclass(value.__class__, Sequence): + self._value = value._value + elif hasattr(value, "__iter__"): + for seq_key, seq_value in value: + self[seq_key] = seq_value + else: + raise InvalidValueType((Sequence,)) if default is not None: - default_value = self._value_sanitize(default) + if not issubclass(default.__class__, Sequence): + raise InvalidValueType((Sequence,)) + default_value = default._value default_obj = self.__class__(impl=self.tag, expl=self._expl) default_obj.specs = self.specs default_obj._value = default_value self.default = default_obj if value is None: - self._value = default_obj.copy()._value - - def _value_sanitize(self, value): - if not issubclass(value.__class__, Sequence): - raise InvalidValueType((Sequence,)) - return value._value + self._value = copy(default_obj._value) @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 - def copy(self): - obj = self.__class__(schema=self.specs) - obj.tag = self.tag - obj._expl = self._expl - obj.default = self.default - obj.optional = self.optional - obj.offset = self.offset - obj.llen = self.llen - obj.vlen = self.vlen - obj._value = {k: v.copy() for k, v in self._value.items()} - return obj + @property + def bered(self): + if self.expl_lenindef or self.lenindef or self.ber_encoded: + return True + return any(value.bered for value in itervalues(self._value)) + + def __getstate__(self): + return SequenceState( + __version__, + self.specs, + {k: copy(v) for k, v in iteritems(self._value)}, + self.tag, + self._expl, + self.default, + self.optional, + self.offset, + self.llen, + self.vlen, + self.expl_lenindef, + self.lenindef, + self.ber_encoded, + ) + + def __setstate__(self, state): + super(Sequence, self).__setstate__(state) + self.specs = state.specs + self._value = state.value + self.tag = state.tag + self._expl = state.expl + self.default = state.default + self.optional = state.optional + self.offset = state.offset + self.llen = state.llen + self.vlen = state.vlen + self.expl_lenindef = state.expl_lenindef + self.lenindef = state.lenindef + self.ber_encoded = state.ber_encoded def __eq__(self, their): if not isinstance(their, self.__class__): @@ -3874,7 +5268,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: @@ -3887,7 +5281,7 @@ class Sequence(Obj): v = b"".join(self._encoded_values()) return b"".join((self.tag, len_encode(len(v)), v)) - def _decode(self, tlv, offset, decode_path, ctx): + def _decode(self, tlv, offset, decode_path, ctx, tag_only): try: t, tlen, lv = tag_strip(tlv) except DecodeError as err: @@ -3903,8 +5297,22 @@ class Sequence(Obj): decode_path=decode_path, offset=offset, ) + if tag_only: # pragma: no cover + return None + lenindef = False + ctx_bered = ctx.get("bered", False) try: l, llen, v = len_decode(lv) + except LenIndefForm as err: + if not ctx_bered: + raise err.__class__( + msg=err.msg, + klass=self.__class__, + decode_path=decode_path, + offset=offset, + ) + l, llen, v = 0, 1, lv[1:] + lenindef = True except DecodeError as err: raise err.__class__( msg=err.msg, @@ -3919,11 +5327,18 @@ class Sequence(Obj): decode_path=decode_path, offset=offset, ) - v, tail = v[:l], v[l:] + if not lenindef: + v, tail = v[:l], v[l:] + vlen = 0 sub_offset = offset + tlen + llen values = {} - for name, spec in self.specs.items(): - if len(v) == 0 and spec.optional: + ber_encoded = False + ctx_allow_default_values = ctx.get("allow_default_values", False) + for name, spec in iteritems(self.specs): + if spec.optional and ( + (lenindef and v[:EOC_LEN].tobytes() == EOC) or + len(v) == 0 + ): continue sub_decode_path = decode_path + (name,) try: @@ -3933,13 +5348,14 @@ class Sequence(Obj): leavemm=True, decode_path=sub_decode_path, 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 - defined = get_def_by_path(ctx.get("defines", ()), sub_decode_path) + defined = get_def_by_path(ctx.get("_defines", ()), sub_decode_path) if defined is not None: defined_by, defined_spec = defined if issubclass(value.__class__, SequenceOf): @@ -3957,6 +5373,7 @@ class Sequence(Obj): leavemm=True, decode_path=sub_sub_decode_path, ctx=ctx, + _ctx_immutable=False, ) if len(defined_tail) > 0: raise DecodeError( @@ -3976,6 +5393,7 @@ class Sequence(Obj): leavemm=True, decode_path=sub_decode_path + (DecodePathDefBy(defined_by),), ctx=ctx, + _ctx_immutable=False, ) if len(defined_tail) > 0: raise DecodeError( @@ -3986,18 +5404,20 @@ class Sequence(Obj): ) value.defined = (defined_by, defined_value) - sub_offset += (value.expl_tlvlen if value.expled else value.tlvlen) + value_len = value.fulllen + vlen += value_len + sub_offset += value_len v = v_tail if spec.default is not None and value == spec.default: - if ctx.get("strict_default_existence", False): + if ctx_bered or ctx_allow_default_values: + ber_encoded = True + else: raise DecodeError( "DEFAULT value met", klass=self.__class__, decode_path=sub_decode_path, offset=sub_offset, ) - else: - continue values[name] = value spec_defines = getattr(spec, "defines", ()) @@ -4009,11 +5429,21 @@ class Sequence(Obj): for rel_path, schema in spec_defines: defined = schema.get(value, None) if defined is not None: - ctx.setdefault("defines", []).append(( + ctx.setdefault("_defines", []).append(( abs_decode_path(sub_decode_path[:-1], rel_path), (value, defined), )) - if len(v) > 0: + if lenindef: + if v[:EOC_LEN].tobytes() != EOC: + raise DecodeError( + "no EOC", + klass=self.__class__, + decode_path=decode_path, + offset=offset, + ) + tail = v[EOC_LEN:] + vlen += EOC_LEN + elif len(v) > 0: raise DecodeError( "remaining data", klass=self.__class__, @@ -4026,9 +5456,11 @@ class Sequence(Obj): expl=self._expl, default=self.default, optional=self.optional, - _decoded=(offset, llen, l), + _decoded=(offset, llen, vlen), ) obj._value = values + obj.lenindef = lenindef + obj.ber_encoded = ber_encoded return obj, tail def __repr__(self): @@ -4038,11 +5470,12 @@ class Sequence(Obj): _value = self._value.get(name) if _value is None: continue - cols.append(repr(_value)) - return "%s[%s]" % (value, ", ".join(cols)) + cols.append("%s: %s" % (name, repr(_value))) + return "%s[%s]" % (value, "; ".join(cols)) def pps(self, decode_path=()): yield _pp( + obj=self, asn1_type_name=self.asn1_type_name, obj_name=self.__class__.__name__, decode_path=decode_path, @@ -4058,18 +5491,32 @@ class Sequence(Obj): expl_tlen=self.expl_tlen if self.expled else None, expl_llen=self.expl_llen if self.expled else None, expl_vlen=self.expl_vlen if self.expled else None, + expl_lenindef=self.expl_lenindef, + lenindef=self.lenindef, + ber_encoded=self.ber_encoded, + bered=self.bered, ) for name in self.specs: value = self._value.get(name) if value is None: continue yield value.pps(decode_path=decode_path + (name,)) + for pp in self.pps_lenindef(decode_path): + yield pp class Set(Sequence): """``SET`` structure type Its usage is identical to :py:class:`pyderasn.Sequence`. + + .. _allow_unordered_set_ctx: + + DER prohibits unordered values encoding and will raise an error + during decode. If If :ref:`bered ` context option is set, + then no error will occure. Also you can disable strict values + ordering check by setting ``"allow_unordered_set": True`` + :ref:`context ` option. """ __slots__ = () tag_default = tag_encode(form=TagFormConstructed, num=17) @@ -4081,7 +5528,10 @@ class Set(Sequence): v = b"".join(raws) return b"".join((self.tag, len_encode(len(v)), v)) - def _decode(self, tlv, offset, decode_path, ctx): + 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) except DecodeError as err: @@ -4097,8 +5547,22 @@ class Set(Sequence): decode_path=decode_path, offset=offset, ) + if tag_only: + return None + lenindef = False + ctx_bered = ctx.get("bered", False) try: l, llen, v = len_decode(lv) + except LenIndefForm as err: + if not ctx_bered: + raise err.__class__( + msg=err.msg, + klass=self.__class__, + decode_path=decode_path, + offset=offset, + ) + l, llen, v = 0, 1, lv[1:] + lenindef = True except DecodeError as err: raise err.__class__( msg=err.msg, @@ -4112,29 +5576,33 @@ class Set(Sequence): klass=self.__class__, offset=offset, ) - v, tail = v[:l], v[l:] + if not lenindef: + v, tail = v[:l], v[l:] + vlen = 0 sub_offset = offset + tlen + llen values = {} - specs_items = self.specs.items + ber_encoded = False + 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]) + while len(v) > 0: - for name, spec in specs_items(): + if lenindef and v[:EOC_LEN].tobytes() == EOC: + break + for name, spec in self._specs_items(): + sub_decode_path = decode_path + (name,) try: - value, v_tail = spec.decode( + spec.decode( v, sub_offset, leavemm=True, - decode_path=decode_path + (name,), + decode_path=sub_decode_path, ctx=ctx, + tag_only=True, + _ctx_immutable=False, ) except TagMismatch: continue - sub_offset += ( - value.expl_tlvlen if value.expled else value.tlvlen - ) - v = v_tail - if spec.default is None or value != spec.default: # pragma: no cover - # SeqMixing.test_encoded_default_accepted covers that place - values[name] = value break else: raise TagMismatch( @@ -4142,18 +5610,90 @@ class Set(Sequence): decode_path=decode_path, offset=offset, ) + value, v_tail = spec.decode( + v, + sub_offset, + leavemm=True, + decode_path=sub_decode_path, + ctx=ctx, + _ctx_immutable=False, + ) + value_len = value.fulllen + if value_prev.tobytes() > v[:value_len].tobytes(): + if ctx_bered or ctx_allow_unordered_set: + ber_encoded = True + else: + raise DecodeError( + "unordered " + self.asn1_type_name, + klass=self.__class__, + decode_path=sub_decode_path, + offset=sub_offset, + ) + if spec.default is None or value != spec.default: + pass + elif ctx_bered or ctx_allow_default_values: + ber_encoded = True + else: + raise DecodeError( + "DEFAULT value met", + klass=self.__class__, + decode_path=sub_decode_path, + offset=sub_offset, + ) + values[name] = value + value_prev = v[:value_len] + sub_offset += value_len + vlen += value_len + v = v_tail obj = self.__class__( schema=self.specs, impl=self.tag, expl=self._expl, default=self.default, optional=self.optional, - _decoded=(offset, llen, l), + _decoded=(offset, llen, vlen + (EOC_LEN if lenindef else 0)), ) + if lenindef: + if v[:EOC_LEN].tobytes() != EOC: + raise DecodeError( + "no EOC", + klass=self.__class__, + decode_path=decode_path, + offset=offset, + ) + tail = v[EOC_LEN:] + obj.lenindef = True obj._value = values + if not obj.ready: + raise DecodeError( + "not all values are ready", + klass=self.__class__, + decode_path=decode_path, + offset=offset, + ) + obj.ber_encoded = ber_encoded return obj, tail +SequenceOfState = namedtuple("SequenceOfState", ( + "version", + "spec", + "value", + "bound_min", + "bound_max", + "tag", + "expl", + "default", + "optional", + "offset", + "llen", + "vlen", + "expl_lenindef", + "lenindef", + "ber_encoded", +)) + + class SequenceOf(Obj): """``SEQUENCE OF`` sequence type @@ -4199,13 +5739,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: @@ -4229,7 +5763,7 @@ class SequenceOf(Obj): default_obj._value = default_value self.default = default_obj if value is None: - self._value = default_obj.copy()._value + self._value = copy(default_obj._value) def _value_sanitize(self, value): if issubclass(value.__class__, SequenceOf): @@ -4249,19 +5783,47 @@ class SequenceOf(Obj): def ready(self): return all(v.ready for v in self._value) - def copy(self): - obj = self.__class__(schema=self.spec) - obj._bound_min = self._bound_min - obj._bound_max = self._bound_max - obj.tag = self.tag - obj._expl = self._expl - obj.default = self.default - obj.optional = self.optional - obj.offset = self.offset - obj.llen = self.llen - obj.vlen = self.vlen - obj._value = [v.copy() for v in self._value] - return obj + @property + def bered(self): + if self.expl_lenindef or self.lenindef or self.ber_encoded: + return True + return any(v.bered for v in self._value) + + def __getstate__(self): + return SequenceOfState( + __version__, + self.spec, + [copy(v) for v in self._value], + self._bound_min, + self._bound_max, + self.tag, + self._expl, + self.default, + self.optional, + self.offset, + self.llen, + self.vlen, + self.expl_lenindef, + self.lenindef, + self.ber_encoded, + ) + + def __setstate__(self, state): + super(SequenceOf, self).__setstate__(state) + self.spec = state.spec + self._value = state.value + self._bound_min = state.bound_min + self._bound_max = state.bound_max + self.tag = state.tag + self._expl = state.expl + self.default = state.default + self.optional = state.optional + self.offset = state.offset + self.llen = state.llen + self.vlen = state.vlen + self.expl_lenindef = state.expl_lenindef + self.lenindef = state.lenindef + self.ber_encoded = state.ber_encoded def __eq__(self, their): if isinstance(their, self.__class__): @@ -4334,7 +5896,7 @@ class SequenceOf(Obj): v = b"".join(self._encoded_values()) return b"".join((self.tag, len_encode(len(v)), v)) - def _decode(self, tlv, offset, decode_path, ctx): + def _decode(self, tlv, offset, decode_path, ctx, tag_only, ordering_check=False): try: t, tlen, lv = tag_strip(tlv) except DecodeError as err: @@ -4350,8 +5912,22 @@ class SequenceOf(Obj): decode_path=decode_path, offset=offset, ) + if tag_only: + return None + lenindef = False + ctx_bered = ctx.get("bered", False) try: l, llen, v = len_decode(lv) + except LenIndefForm as err: + if not ctx_bered: + raise err.__class__( + msg=err.msg, + klass=self.__class__, + decode_path=decode_path, + offset=offset, + ) + l, llen, v = 0, 1, lv[1:] + lenindef = True except DecodeError as err: raise err.__class__( msg=err.msg, @@ -4366,31 +5942,73 @@ class SequenceOf(Obj): decode_path=decode_path, offset=offset, ) - v, tail = v[:l], v[l:] + if not lenindef: + v, tail = v[:l], v[l:] + vlen = 0 sub_offset = offset + tlen + llen _value = [] + ctx_allow_unordered_set = ctx.get("allow_unordered_set", False) + value_prev = memoryview(v[:0]) + ber_encoded = False spec = self.spec while len(v) > 0: + if lenindef and v[:EOC_LEN].tobytes() == EOC: + break + sub_decode_path = decode_path + (str(len(_value)),) value, v_tail = spec.decode( v, sub_offset, leavemm=True, - decode_path=decode_path + (str(len(_value)),), + decode_path=sub_decode_path, ctx=ctx, + _ctx_immutable=False, ) - sub_offset += (value.expl_tlvlen if value.expled else value.tlvlen) - v = v_tail + value_len = value.fulllen + if ordering_check: + if value_prev.tobytes() > v[:value_len].tobytes(): + if ctx_bered or ctx_allow_unordered_set: + ber_encoded = True + else: + raise DecodeError( + "unordered " + self.asn1_type_name, + klass=self.__class__, + decode_path=sub_decode_path, + offset=sub_offset, + ) + value_prev = v[:value_len] _value.append(value) - obj = self.__class__( - value=_value, - schema=spec, - bounds=(self._bound_min, self._bound_max), - impl=self.tag, - expl=self._expl, - default=self.default, - optional=self.optional, - _decoded=(offset, llen, l), - ) + sub_offset += value_len + vlen += value_len + v = v_tail + try: + obj = self.__class__( + value=_value, + schema=spec, + bounds=(self._bound_min, self._bound_max), + impl=self.tag, + expl=self._expl, + default=self.default, + optional=self.optional, + _decoded=(offset, llen, vlen + (EOC_LEN if lenindef else 0)), + ) + except BoundsError as err: + raise DecodeError( + msg=str(err), + klass=self.__class__, + decode_path=decode_path, + offset=offset, + ) + if lenindef: + if v[:EOC_LEN].tobytes() != EOC: + raise DecodeError( + "no EOC", + klass=self.__class__, + decode_path=decode_path, + offset=offset, + ) + obj.lenindef = True + tail = v[EOC_LEN:] + obj.ber_encoded = ber_encoded return obj, tail def __repr__(self): @@ -4401,6 +6019,7 @@ class SequenceOf(Obj): def pps(self, decode_path=()): yield _pp( + obj=self, asn1_type_name=self.asn1_type_name, obj_name=self.__class__.__name__, decode_path=decode_path, @@ -4416,9 +6035,15 @@ class SequenceOf(Obj): expl_tlen=self.expl_tlen if self.expled else None, expl_llen=self.expl_llen if self.expled else None, expl_vlen=self.expl_vlen if self.expled else None, + expl_lenindef=self.expl_lenindef, + lenindef=self.lenindef, + ber_encoded=self.ber_encoded, + bered=self.bered, ) for i, value in enumerate(self._value): yield value.pps(decode_path=decode_path + (str(i),)) + for pp in self.pps_lenindef(decode_path): + yield pp class SetOf(SequenceOf): @@ -4436,6 +6061,16 @@ class SetOf(SequenceOf): v = b"".join(raws) return b"".join((self.tag, len_encode(len(v)), v)) + def _decode(self, tlv, offset, decode_path, ctx, tag_only): + return super(SetOf, self)._decode( + tlv, + offset, + decode_path, + ctx, + tag_only, + ordering_check=True, + ) + def obj_by_path(pypath): # pragma: no cover """Import object specified as string Python path @@ -4460,7 +6095,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), @@ -4472,10 +6107,21 @@ def generic_decoder(): # pragma: no cover __slots__ = () schema = choice - def pprint_any(obj, oids=None, with_colours=False): + def pprint_any( + obj, + oid_maps=(), + with_colours=False, + with_decode_path=False, + decode_path_only=(), + ): def _pprint_pps(pps): for pp in pps: if hasattr(pp, "_fields"): + if ( + decode_path_only != () and + pp.decode_path[:len(decode_path_only)] != decode_path_only + ): + continue if pp.asn1_type_name == Choice.asn1_type_name: continue pp_kwargs = pp._asdict() @@ -4483,12 +6129,17 @@ 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, + with_decode_path=with_decode_path, + decode_path_len_decrease=len(decode_path_only), ) - for row in pp_console_blob(pp): + for row in pp_console_blob( + pp, + decode_path_len_decrease=len(decode_path_only), + ): yield row else: for row in _pprint_pps(pp): @@ -4499,7 +6150,7 @@ def generic_decoder(): # pragma: no cover def main(): # pragma: no cover import argparse - parser = argparse.ArgumentParser(description="PyDERASN ASN.1 DER decoder") + parser = argparse.ArgumentParser(description="PyDERASN ASN.1 BER/DER decoder") parser.add_argument( "--skip", type=int, @@ -4508,7 +6159,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", @@ -4518,6 +6169,25 @@ def main(): # pragma: no cover "--defines-by-path", help="Python path to decoder's defines_by_path", ) + parser.add_argument( + "--nobered", + action="store_true", + help="Disallow BER encoding", + ) + parser.add_argument( + "--print-decode-path", + action="store_true", + help="Print decode paths", + ) + parser.add_argument( + "--decode-path-only", + help="Print only specified decode path", + ) + parser.add_argument( + "--allow-expl-oob", + action="store_true", + help="Allow explicit tag out-of-bound", + ) parser.add_argument( "DERFile", type=argparse.FileType("rb"), @@ -4527,24 +6197,32 @@ 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 pprinter = partial(pprint, big_blobs=True) else: schema, pprinter = generic_decoder() - obj, tail = schema().decode( - der, - ctx=( - None if args.defines_by_path is None else - {"defines_by_path": obj_by_path(args.defines_by_path)} - ), - ) + ctx = { + "bered": not args.nobered, + "allow_expl_oob": args.allow_expl_oob, + } + if args.defines_by_path is not None: + ctx["defines_by_path"] = obj_by_path(args.defines_by_path) + 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 + tuple(args.decode_path_only.split(":")) + ), )) if tail != b"": print("\nTrailing data: %s" % hexenc(tail))