]> Cypherpunks.ru repositories - pyderasn.git/blobdiff - pyderasn.py
Convenient decod() helper method
[pyderasn.git] / pyderasn.py
index 1c5b53f1f617810c1e8c3cb906710a282cf23126..5f0c0343ccd0d4153d29e6328fde4a73882229ec 100755 (executable)
@@ -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 <stargrave@stargrave.org>
+# Copyright (C) 2017-2020 Sergey Matveev <stargrave@stargrave.org>
 #
 # 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
-# <http://www.gnu.org/licenses/>.
+# License along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """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 <defines>`.
 
 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
 ---------------
 
@@ -626,10 +631,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
@@ -670,9 +675,10 @@ from six.moves import xrange as six_xrange
 try:
     from termcolor import colored
 except ImportError:  # pragma: no cover
-    def colored(what, *args):
+    def colored(what, *args, **kwargs):
         return what
 
+__version__ = "5.6"
 
 __all__ = (
     "Any",
@@ -684,6 +690,7 @@ __all__ = (
     "DecodeError",
     "DecodePathDefBy",
     "Enumerated",
+    "ExceedingData",
     "GeneralizedTime",
     "GeneralString",
     "GraphicString",
@@ -794,6 +801,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 +1038,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 +1114,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 +1146,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 +1177,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 +1194,7 @@ class Obj(object):
                 tag_only=tag_only,
             )
             if tag_only:
-                return
+                return None
             obj, tail = result
         else:
             try:
@@ -1203,7 +1232,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 +1267,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,42 +1278,80 @@ class Obj(object):
                     )
         return obj, (tail if leavemm else tail.tobytes())
 
+    def decod(self, data, offset=0, decode_path=(), ctx=None):
+        """Decode the data, check that tail is empty
+
+        :raises ExceedingData: if tail is not empty
+
+        This is just a wrapper over :py:meth:`pyderasn.Obj.decode`
+        (decode without tail) that also checks that there is no
+        trailing data left.
+        """
+        obj, tail = self.decode(
+            data,
+            offset=offset,
+            decode_path=decode_path,
+            ctx=ctx,
+            leavemm=True,
+        )
+        if len(tail) > 0:
+            raise ExceedingData(len(tail))
+        return obj
+
     @property
     def expled(self):
+        """See :ref:`decoding`
+        """
         return self._expl is not None
 
     @property
     def expl_tag(self):
+        """See :ref:`decoding`
+        """
         return self._expl
 
     @property
     def expl_tlen(self):
+        """See :ref:`decoding`
+        """
         return len(self._expl)
 
     @property
     def expl_llen(self):
+        """See :ref:`decoding`
+        """
         if self.expl_lenindef:
             return 1
         return len(len_encode(self.tlvlen))
 
     @property
     def expl_offset(self):
+        """See :ref:`decoding`
+        """
         return self.offset - self.expl_tlen - self.expl_llen
 
     @property
     def expl_vlen(self):
+        """See :ref:`decoding`
+        """
         return self.tlvlen
 
     @property
     def expl_tlvlen(self):
+        """See :ref:`decoding`
+        """
         return self.expl_tlen + self.expl_llen + self.expl_vlen
 
     @property
     def fulloffset(self):
+        """See :ref:`decoding`
+        """
         return self.expl_offset if self.expled else self.offset
 
     @property
     def fulllen(self):
+        """See :ref:`decoding`
+        """
         return self.expl_tlvlen if self.expled else self.tlvlen
 
     def pps_lenindef(self, decode_path):
@@ -1435,7 +1502,7 @@ def colonize_hex(hexed):
 
 def pp_console_row(
         pp,
-        oids=None,
+        oid_maps=(),
         with_offsets=False,
         with_blob=True,
         with_colours=False,
@@ -1470,14 +1537,18 @@ def pp_console_row(
         if isinstance(ent, DecodePathDefBy):
             cols.append(_colourize("DEFINED BY", "red", with_colours, ("reverse",)))
             value = str(ent.defined_by)
+            oid_name = None
             if (
-                    oids is not None and
+                    len(oid_maps) > 0 and
                     ent.defined_by.asn1_type_name ==
-                    ObjectIdentifier.asn1_type_name and
-                    value in oids
+                    ObjectIdentifier.asn1_type_name
             ):
-                cols.append(_colourize("%s:" % oids[value], "green", with_colours))
-            else:
+                for oid_map in oid_maps:
+                    oid_name = oid_map.get(value)
+                    if oid_name is not None:
+                        cols.append(_colourize("%s:" % oid_name, "green", with_colours))
+                        break
+            if oid_name is None:
                 cols.append(_colourize("%s:" % value, "white", with_colours, ("reverse",)))
         else:
             cols.append(_colourize("%s:" % ent, "yellow", with_colours, ("reverse",)))
@@ -1498,11 +1569,14 @@ def pp_console_row(
         value = pp.value
         cols.append(_colourize(value, "white", with_colours, ("reverse",)))
         if (
-                oids is not None and
-                pp.asn1_type_name == ObjectIdentifier.asn1_type_name and
-                value in oids
+                len(oid_maps) > 0 and
+                pp.asn1_type_name == ObjectIdentifier.asn1_type_name
         ):
-            cols.append(_colourize("(%s)" % oids[value], "green", with_colours))
+            for oid_map in oid_maps:
+                oid_name = oid_map.get(value)
+                if oid_name is not None:
+                    cols.append(_colourize("(%s)" % oid_name, "green", with_colours))
+                    break
         if pp.asn1_type_name == Integer.asn1_type_name:
             hex_repr = hex(int(pp.obj._value))[2:].upper()
             if len(hex_repr) % 2 != 0:
@@ -1546,7 +1620,7 @@ def pp_console_blob(pp, decode_path_len_decrease=0):
 
 def pprint(
         obj,
-        oids=None,
+        oid_maps=(),
         big_blobs=False,
         with_colours=False,
         with_decode_path=False,
@@ -1555,8 +1629,9 @@ def pprint(
     """Pretty print object
 
     :param Obj obj: object you want to pretty print
-    :param oids: ``OID <-> humand readable string`` dictionary. When OID
-                 from it is met, then its humand readable form is printed
+    :param oid_maps: list of ``OID <-> humand readable string`` dictionary.
+                     When OID from it is met, then its humand readable form
+                     is printed
     :param big_blobs: if large binary objects are met (like OctetString
                       values), do we need to print them too, on separate
                       lines
@@ -1578,7 +1653,7 @@ def pprint(
                 if big_blobs:
                     yield pp_console_row(
                         pp,
-                        oids=oids,
+                        oid_maps=oid_maps,
                         with_offsets=True,
                         with_blob=False,
                         with_colours=with_colours,
@@ -1593,7 +1668,7 @@ def pprint(
                 else:
                     yield pp_console_row(
                         pp,
-                        oids=oids,
+                        oid_maps=oid_maps,
                         with_offsets=True,
                         with_blob=True,
                         with_colours=with_colours,
@@ -1739,7 +1814,7 @@ class Boolean(Obj):
                 offset=offset,
             )
         if tag_only:
-            return
+            return None
         try:
             l, _, v = len_decode(lv)
         except DecodeError as err:
@@ -1969,6 +2044,7 @@ class Integer(Obj):
         for name, value in iteritems(self.specs):
             if value == self._value:
                 return name
+        return None
 
     def __call__(
             self,
@@ -2048,7 +2124,7 @@ class Integer(Obj):
                 offset=offset,
             )
         if tag_only:
-            return
+            return None
         try:
             l, llen, v = len_decode(lv)
         except DecodeError as err:
@@ -2268,7 +2344,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,
@@ -2276,8 +2352,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
@@ -2396,7 +2471,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:
@@ -2466,8 +2541,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(
@@ -2477,7 +2552,7 @@ class BitString(Obj):
                     offset=offset,
                 )
             if tag_only:  # pragma: no cover
-                return
+                return None
             lenindef = False
             try:
                 l, llen, v = len_decode(lv)
@@ -2786,7 +2861,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:
@@ -2842,8 +2917,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(
@@ -2853,7 +2928,7 @@ class OctetString(Obj):
                     offset=offset,
                 )
             if tag_only:
-                return
+                return None
             lenindef = False
             try:
                 l, llen, v = len_decode(lv)
@@ -3069,7 +3144,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:
@@ -3324,7 +3399,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:
@@ -3832,7 +3907,7 @@ class UTCTime(CommonString):
             try:
                 value_decoded = value.decode("ascii")
             except (UnicodeEncodeError, UnicodeDecodeError) as err:
-                raise DecodeError("invalid UTCTime encoding")
+                raise DecodeError("invalid UTCTime encoding: %r" % err)
             try:
                 self._strptime(value_decoded)
             except (TypeError, ValueError) as err:
@@ -3977,7 +4052,7 @@ class GeneralizedTime(UTCTime):
             try:
                 value_decoded = value.decode("ascii")
             except (UnicodeEncodeError, UnicodeDecodeError) as err:
-                raise DecodeError("invalid GeneralizedTime encoding")
+                raise DecodeError("invalid GeneralizedTime encoding: %r" % err)
             try:
                 self._strptime(value_decoded)
             except (TypeError, ValueError) as err:
@@ -4242,7 +4317,7 @@ class Choice(Obj):
                 offset=offset,
             )
         if tag_only:  # pragma: no cover
-            return
+            return None
         value, tail = spec.decode(
             tlv,
             offset=offset,
@@ -4711,9 +4786,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
@@ -4825,7 +4899,7 @@ class Sequence(Obj):
                 offset=offset,
             )
         if tag_only:  # pragma: no cover
-            return
+            return None
         lenindef = False
         ctx_bered = ctx.get("bered", False)
         try:
@@ -4877,8 +4951,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
 
@@ -5075,7 +5149,7 @@ class Set(Sequence):
                 offset=offset,
             )
         if tag_only:
-            return
+            return None
         lenindef = False
         ctx_bered = ctx.get("bered", False)
         try:
@@ -5408,7 +5482,7 @@ class SequenceOf(Obj):
                 offset=offset,
             )
         if tag_only:
-            return
+            return None
         lenindef = False
         ctx_bered = ctx.get("bered", False)
         try:
@@ -5604,7 +5678,7 @@ def generic_decoder():  # pragma: no cover
 
     def pprint_any(
             obj,
-            oids=None,
+            oid_maps=(),
             with_colours=False,
             with_decode_path=False,
             decode_path_only=(),
@@ -5624,7 +5698,7 @@ def generic_decoder():  # pragma: no cover
                     pp = _pp(**pp_kwargs)
                     yield pp_console_row(
                         pp,
-                        oids=oids,
+                        oid_maps=oid_maps,
                         with_offsets=True,
                         with_blob=False,
                         with_colours=with_colours,
@@ -5654,7 +5728,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",
@@ -5692,7 +5766,10 @@ def main():  # pragma: no cover
     args.DERFile.seek(args.skip)
     der = memoryview(args.DERFile.read())
     args.DERFile.close()
-    oids = obj_by_path(args.oids) if args.oids else {}
+    oid_maps = (
+        [obj_by_path(_path) for _path in (args.oids or "").split(",")]
+        if args.oids else ()
+    )
     if args.schema:
         schema = obj_by_path(args.schema)
         from functools import partial
@@ -5708,8 +5785,8 @@ def main():  # pragma: no cover
     obj, tail = schema().decode(der, ctx=ctx)
     print(pprinter(
         obj,
-        oids=oids,
-        with_colours=True if environ.get("NO_COLOR") is None else False,
+        oid_maps=oid_maps,
+        with_colours=environ.get("NO_COLOR") is None,
         with_decode_path=args.print_decode_path,
         decode_path_only=(
             () if args.decode_path_only is None else