X-Git-Url: http://www.git.cypherpunks.ru/?a=blobdiff_plain;f=pyderasn.py;h=6e32db54de851f5089b1ed042f75d57a408498ec;hb=7fccd783e8352d75ffe51bb058f6703abb744057;hp=887f000d20b73afbee0a348af9b2ad3681a79433;hpb=29a10ee9f33812d6ec56e1f59613853c14a63a78;p=pyderasn.git diff --git a/pyderasn.py b/pyderasn.py index 887f000..6e32db5 100755 --- a/pyderasn.py +++ b/pyderasn.py @@ -1,12 +1,11 @@ #!/usr/bin/env python # coding: utf-8 # PyDERASN -- Python ASN.1 DER/BER codec with abstract structures -# Copyright (C) 2017-2019 Sergey Matveev +# Copyright (C) 2017-2020 Sergey Matveev # # This program is free software: you can redistribute it and/or modify # it under the terms of the GNU Lesser General Public License as -# published by the Free Software Foundation, either version 3 of the -# License, or (at your option) any later version. +# published by the Free Software Foundation, version 3 of the License. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of @@ -14,8 +13,7 @@ # GNU Lesser General Public License for more details. # # You should have received a copy of the GNU Lesser General Public -# License along with this program. If not, see -# . +# License along with this program. If not, see . """Python ASN.1 DER/BER codec with abstract structures This library allows you to marshal various structures in ASN.1 DER @@ -348,6 +346,8 @@ DEFINED BY some previously met ObjectIdentifier. This library provides ability to specify mapping between some OID and field that must be decoded with specific specification. +.. _defines: + defines kwarg _____________ @@ -421,15 +421,15 @@ value must be sequence of following tuples:: where ``decode_path`` is a tuple holding so-called decode path to the exact :py:class:`pyderasn.ObjectIdentifier` field you want to apply -``defines``, holding exactly the same value as accepted in its keyword -argument. +``defines``, holding exactly the same value as accepted in its +:ref:`keyword argument `. For example, again for CMS, you want to automatically decode ``SignedData`` and CMC's (:rfc:`5272`) ``PKIData`` and ``PKIResponse`` structures it may hold. Also, automatically decode ``controlSequence`` of ``PKIResponse``:: - content_info, tail = ContentInfo().decode(data, defines_by_path=( + content_info, tail = ContentInfo().decode(data, ctx={"defines_by_path": ( ( ("contentType",), ((("content",), {id_signedData: SignedData()}),), @@ -464,7 +464,7 @@ of ``PKIResponse``:: id_cmc_transactionId: TransactionId(), })), ), - )) + )}) Pay attention for :py:class:`pyderasn.DecodePathDefBy` and ``any``. First function is useful for path construction when some automatic @@ -517,6 +517,11 @@ lengths will be invalid in that case. This option should be used only for skipping some decode errors, just to see the decoded structure somehow. +Base Obj +-------- +.. autoclass:: pyderasn.Obj + :members: + Primitive types --------------- @@ -565,6 +570,7 @@ _____________ PrintableString _______________ .. autoclass:: pyderasn.PrintableString + :members: __init__ UTCTime _______ @@ -626,10 +632,10 @@ Various .. autofunction:: pyderasn.tag_decode .. autofunction:: pyderasn.tag_ctxp .. autofunction:: pyderasn.tag_ctxc -.. autoclass:: pyderasn.Obj .. autoclass:: pyderasn.DecodeError :members: __init__ .. autoclass:: pyderasn.NotEnoughData +.. autoclass:: pyderasn.ExceedingData .. autoclass:: pyderasn.LenIndefForm .. autoclass:: pyderasn.TagMismatch .. autoclass:: pyderasn.InvalidLength @@ -650,6 +656,7 @@ from math import ceil from os import environ from string import ascii_letters from string import digits +from unicodedata import category as unicat from six import add_metaclass from six import binary_type @@ -673,6 +680,7 @@ except ImportError: # pragma: no cover def colored(what, *args, **kwargs): return what +__version__ = "5.6" __all__ = ( "Any", @@ -684,6 +692,7 @@ __all__ = ( "DecodeError", "DecodePathDefBy", "Enumerated", + "ExceedingData", "GeneralizedTime", "GeneralString", "GraphicString", @@ -794,6 +803,18 @@ class NotEnoughData(DecodeError): pass +class ExceedingData(ASN1Error): + def __init__(self, nbytes): + super(ExceedingData, self).__init__() + self.nbytes = nbytes + + def __str__(self): + return "%d trailing bytes" % self.nbytes + + def __repr__(self): + return "%s(%s)" % (self.__class__.__name__, self) + + class LenIndefForm(DecodeError): pass @@ -1019,9 +1040,9 @@ def len_decode(data): ######################################################################## class AutoAddSlots(type): - def __new__(mcs, name, bases, _dict): + def __new__(cls, name, bases, _dict): _dict["__slots__"] = _dict.get("__slots__", ()) - return type.__new__(mcs, name, bases, _dict) + return type.__new__(cls, name, bases, _dict) @add_metaclass(AutoAddSlots) @@ -1095,10 +1116,14 @@ class Obj(object): @property def tlen(self): + """See :ref:`decoding` + """ return len(self.tag) @property def tlvlen(self): + """See :ref:`decoding` + """ return self.tlen + self.llen + self.vlen def __str__(self): # pragma: no cover @@ -1123,6 +1148,10 @@ class Obj(object): raise NotImplementedError() def encode(self): + """Encode the structure + + :returns: DER representation + """ raw = self._encode() if self._expl is None: return raw @@ -1150,6 +1179,8 @@ class Obj(object): determine if tag satisfies the scheme) :param _ctx_immutable: do we need to copy ``ctx`` before using it :returns: (Obj, remaining data) + + .. seealso:: :ref:`decoding` """ if ctx is None: ctx = {} @@ -1165,7 +1196,7 @@ class Obj(object): tag_only=tag_only, ) if tag_only: - return + return None obj, tail = result else: try: @@ -1203,7 +1234,7 @@ class Obj(object): tag_only=tag_only, ) if tag_only: # pragma: no cover - return + return None obj, tail = result eoc_expected, tail = tail[:EOC_LEN], tail[EOC_LEN:] if eoc_expected.tobytes() != EOC: @@ -1238,7 +1269,7 @@ class Obj(object): tag_only=tag_only, ) if tag_only: # pragma: no cover - return + return None obj, tail = result if obj.tlvlen < l and not ctx.get("allow_expl_oob", False): raise DecodeError( @@ -1249,48 +1280,86 @@ class Obj(object): ) return obj, (tail if leavemm else tail.tobytes()) + def decod(self, data, offset=0, decode_path=(), ctx=None): + """Decode the data, check that tail is empty + + :raises ExceedingData: if tail is not empty + + This is just a wrapper over :py:meth:`pyderasn.Obj.decode` + (decode without tail) that also checks that there is no + trailing data left. + """ + obj, tail = self.decode( + data, + offset=offset, + decode_path=decode_path, + ctx=ctx, + leavemm=True, + ) + if len(tail) > 0: + raise ExceedingData(len(tail)) + return obj + @property def expled(self): + """See :ref:`decoding` + """ return self._expl is not None @property def expl_tag(self): + """See :ref:`decoding` + """ return self._expl @property def expl_tlen(self): + """See :ref:`decoding` + """ return len(self._expl) @property def expl_llen(self): + """See :ref:`decoding` + """ if self.expl_lenindef: return 1 return len(len_encode(self.tlvlen)) @property def expl_offset(self): + """See :ref:`decoding` + """ return self.offset - self.expl_tlen - self.expl_llen @property def expl_vlen(self): + """See :ref:`decoding` + """ return self.tlvlen @property def expl_tlvlen(self): + """See :ref:`decoding` + """ return self.expl_tlen + self.expl_llen + self.expl_vlen @property def fulloffset(self): + """See :ref:`decoding` + """ return self.expl_offset if self.expled else self.offset @property def fulllen(self): + """See :ref:`decoding` + """ return self.expl_tlvlen if self.expled else self.tlvlen def pps_lenindef(self, decode_path): if self.lenindef and not ( - getattr(self, "defined", None) is not None and - self.defined[1].lenindef + getattr(self, "defined", None) is not None and + self.defined[1].lenindef ): yield _pp( asn1_type_name="EOC", @@ -1577,10 +1646,10 @@ def pprint( 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 + decode_path_only != () and + tuple( + str(p) for p in pp.decode_path[:len(decode_path_only)] + ) != decode_path_only ): continue if big_blobs: @@ -1594,8 +1663,8 @@ def pprint( decode_path_len_decrease=len(decode_path_only), ) for row in pp_console_blob( - pp, - decode_path_len_decrease=len(decode_path_only), + pp, + decode_path_len_decrease=len(decode_path_only), ): yield row else: @@ -1747,7 +1816,7 @@ class Boolean(Obj): offset=offset, ) if tag_only: - return + return None try: l, _, v = len_decode(lv) except DecodeError as err: @@ -1977,6 +2046,7 @@ class Integer(Obj): for name, value in iteritems(self.specs): if value == self._value: return name + return None def __call__( self, @@ -2056,7 +2126,7 @@ class Integer(Obj): offset=offset, ) if tag_only: - return + return None try: l, llen, v = len_decode(lv) except DecodeError as err: @@ -2276,7 +2346,7 @@ class BitString(Obj): if not frozenset(value) <= SET01: raise ValueError("B's coding contains unacceptable chars") return self._bits2octets(value) - elif value.endswith("'H"): + if value.endswith("'H"): value = value[1:-2] return ( len(value) * 4, @@ -2284,8 +2354,7 @@ class BitString(Obj): ) if isinstance(value, binary_type): return (len(value) * 8, value) - else: - raise InvalidValueType((self.__class__, string_types, binary_type)) + raise InvalidValueType((self.__class__, string_types, binary_type)) if isinstance(value, tuple): if ( len(value) == 2 and @@ -2404,7 +2473,7 @@ class BitString(Obj): octets, )) - def _decode_chunk(self, lv, offset, decode_path, ctx): + def _decode_chunk(self, lv, offset, decode_path): try: l, llen, v = len_decode(lv) except DecodeError as err: @@ -2474,8 +2543,8 @@ class BitString(Obj): ) if t == self.tag: if tag_only: # pragma: no cover - return - return self._decode_chunk(lv, offset, decode_path, ctx) + return None + return self._decode_chunk(lv, offset, decode_path) if t == self.tag_constructed: if not ctx.get("bered", False): raise DecodeError( @@ -2485,7 +2554,7 @@ class BitString(Obj): offset=offset, ) if tag_only: # pragma: no cover - return + return None lenindef = False try: l, llen, v = len_decode(lv) @@ -2683,13 +2752,7 @@ class OctetString(Obj): :param default: set default value. Type same as in ``value`` :param bool optional: is object ``OPTIONAL`` in sequence """ - super(OctetString, self).__init__( - impl, - expl, - default, - optional, - _decoded, - ) + super(OctetString, self).__init__(impl, expl, default, optional, _decoded) self._value = value self._bound_min, self._bound_max = getattr( self, @@ -2794,7 +2857,7 @@ class OctetString(Obj): self._value, )) - def _decode_chunk(self, lv, offset, decode_path, ctx): + def _decode_chunk(self, lv, offset, decode_path): try: l, llen, v = len_decode(lv) except DecodeError as err: @@ -2850,8 +2913,8 @@ class OctetString(Obj): ) if t == self.tag: if tag_only: - return - return self._decode_chunk(lv, offset, decode_path, ctx) + return None + return self._decode_chunk(lv, offset, decode_path) if t == self.tag_constructed: if not ctx.get("bered", False): raise DecodeError( @@ -2861,7 +2924,7 @@ class OctetString(Obj): offset=offset, ) if tag_only: - return + return None lenindef = False try: l, llen, v = len_decode(lv) @@ -3077,7 +3140,7 @@ class Null(Obj): offset=offset, ) if tag_only: # pragma: no cover - return + return None try: l, _, v = len_decode(lv) except DecodeError as err: @@ -3179,13 +3242,7 @@ class ObjectIdentifier(Obj): :param default: set default value. Type same as in ``value`` :param bool optional: is object ``OPTIONAL`` in sequence """ - super(ObjectIdentifier, self).__init__( - impl, - expl, - default, - optional, - _decoded, - ) + super(ObjectIdentifier, self).__init__(impl, expl, default, optional, _decoded) self._value = value if value is not None: self._value = self._value_sanitize(value) @@ -3332,7 +3389,7 @@ class ObjectIdentifier(Obj): offset=offset, ) if tag_only: # pragma: no cover - return + return None try: l, llen, v = len_decode(lv) except DecodeError as err: @@ -3456,13 +3513,7 @@ class Enumerated(Integer): bounds=None, # dummy argument, workability for Integer.decode ): super(Enumerated, self).__init__( - value=value, - impl=impl, - expl=expl, - default=default, - optional=optional, - _specs=_specs, - _decoded=_decoded, + value, bounds, impl, expl, default, optional, _specs, _decoded, ) if len(self.specs) == 0: raise ValueError("schema must be specified") @@ -3523,6 +3574,12 @@ class Enumerated(Integer): ) +def escape_control_unicode(c): + if unicat(c).startswith("C"): + c = repr(c).lstrip("u").strip("'") + return c + + class CommonString(OctetString): """Common class for all strings @@ -3585,7 +3642,7 @@ class CommonString(OctetString): * - :py:class:`pyderasn.BMPString` - utf-16-be """ - __slots__ = ("encoding",) + __slots__ = () def _value_sanitize(self, value): value_raw = None @@ -3641,7 +3698,10 @@ class CommonString(OctetString): def pps(self, decode_path=(), no_unicode=False): value = None if self.ready: - value = hexenc(bytes(self)) if no_unicode else self.__unicode__() + value = ( + hexenc(bytes(self)) if no_unicode else + "".join(escape_control_unicode(c) for c in self.__unicode__()) + ) yield _pp( obj=self, asn1_type_name=self.asn1_type_name, @@ -3720,6 +3780,32 @@ class PrintableString(AllowableCharsMixin, CommonString): _allowable_chars = frozenset( (ascii_letters + digits + " '()+,-./:=?").encode("ascii") ) + _asterisk = frozenset("*".encode("ascii")) + _ampersand = frozenset("&".encode("ascii")) + + def __init__( + self, + value=None, + bounds=None, + impl=None, + expl=None, + default=None, + optional=False, + _decoded=(0, 0, 0), + allow_asterisk=False, + allow_ampersand=False, + ): + """ + :param allow_asterisk: allow asterisk character + :param allow_ampersand: allow ampersand character + """ + if allow_asterisk: + self._allowable_chars |= self._asterisk + if allow_ampersand: + self._allowable_chars |= self._ampersand + super(PrintableString, self).__init__( + value, bounds, impl, expl, default, optional, _decoded, + ) def _value_sanitize(self, value): value = super(PrintableString, self)._value_sanitize(value) @@ -3727,6 +3813,34 @@ class PrintableString(AllowableCharsMixin, CommonString): raise DecodeError("non-printable value") return value + def copy(self): + obj = super(PrintableString, self).copy() + obj._allowable_chars = self._allowable_chars + return obj + + def __call__( + self, + value=None, + bounds=None, + impl=None, + expl=None, + default=None, + optional=None, + ): + return self.__class__( + value=value, + bounds=( + (self._bound_min, self._bound_max) + if bounds is None else bounds + ), + impl=self.tag if impl is None else impl, + expl=self._expl if expl is None else expl, + default=self.default if default is None else default, + optional=self.optional if optional is None else optional, + allow_asterisk=self._asterisk <= self._allowable_chars, + allow_ampersand=self._ampersand <= self._allowable_chars, + ) + class TeletexString(CommonString): __slots__ = () @@ -3801,11 +3915,7 @@ class UTCTime(CommonString): :param bool optional: is object ``OPTIONAL`` in sequence """ super(UTCTime, self).__init__( - impl=impl, - expl=expl, - default=default, - optional=optional, - _decoded=_decoded, + None, None, impl, expl, default, optional, _decoded, ) self._value = value if value is not None: @@ -4250,7 +4360,7 @@ class Choice(Obj): offset=offset, ) if tag_only: # pragma: no cover - return + return None value, tail = spec.decode( tlv, offset=offset, @@ -4476,7 +4586,7 @@ class Any(Obj): _decoded=(offset, 0, tlvlen), ) obj.lenindef = True - obj.tag = t + obj.tag = t.tobytes() return obj, v[EOC_LEN:] except DecodeError as err: raise err.__class__( @@ -4500,7 +4610,7 @@ class Any(Obj): optional=self.optional, _decoded=(offset, 0, tlvlen), ) - obj.tag = t + obj.tag = t.tobytes() return obj, tail def __repr__(self): @@ -4719,9 +4829,8 @@ class Sequence(Obj): if spec.optional: continue return False - else: - if not value.ready: - return False + if not value.ready: + return False return True @property @@ -4833,7 +4942,7 @@ class Sequence(Obj): offset=offset, ) if tag_only: # pragma: no cover - return + return None lenindef = False ctx_bered = ctx.get("bered", False) try: @@ -4885,8 +4994,8 @@ class Sequence(Obj): ctx=ctx, _ctx_immutable=False, ) - except TagMismatch: - if spec.optional: + except TagMismatch as err: + if (len(err.decode_path) == len(decode_path) + 1) and spec.optional: continue raise @@ -5083,7 +5192,7 @@ class Set(Sequence): offset=offset, ) if tag_only: - return + return None lenindef = False ctx_bered = ctx.get("bered", False) try: @@ -5255,13 +5364,7 @@ class SequenceOf(Obj): optional=False, _decoded=(0, 0, 0), ): - super(SequenceOf, self).__init__( - impl, - expl, - default, - optional, - _decoded, - ) + super(SequenceOf, self).__init__(impl, expl, default, optional, _decoded) if schema is None: schema = getattr(self, "schema", None) if schema is None: @@ -5416,7 +5519,7 @@ class SequenceOf(Obj): offset=offset, ) if tag_only: - return + return None lenindef = False ctx_bered = ctx.get("bered", False) try: @@ -5621,8 +5724,8 @@ def generic_decoder(): # pragma: no cover for pp in pps: if hasattr(pp, "_fields"): if ( - decode_path_only != () and - pp.decode_path[:len(decode_path_only)] != decode_path_only + decode_path_only != () and + pp.decode_path[:len(decode_path_only)] != decode_path_only ): continue if pp.asn1_type_name == Choice.asn1_type_name: @@ -5640,8 +5743,8 @@ def generic_decoder(): # pragma: no cover decode_path_len_decrease=len(decode_path_only), ) for row in pp_console_blob( - pp, - decode_path_len_decrease=len(decode_path_only), + pp, + decode_path_len_decrease=len(decode_path_only), ): yield row else: @@ -5720,7 +5823,7 @@ def main(): # pragma: no cover print(pprinter( obj, oid_maps=oid_maps, - with_colours=True if environ.get("NO_COLOR") is None else False, + 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