X-Git-Url: http://www.git.cypherpunks.ru/?a=blobdiff_plain;f=pyderasn.py;h=00defa5ab0e6926f7221e97de99e8b3a8043f60c;hb=ac1628691fa68bcc61a0374b219ead802b94e17b;hp=f563afef86964240e57971d85c91fa35d40de0c8;hpb=5699682993d2c2b789640c20cd68e7bd1eb0cf7c;p=pyderasn.git diff --git a/pyderasn.py b/pyderasn.py index f563afe..00defa5 100755 --- a/pyderasn.py +++ b/pyderasn.py @@ -21,7 +21,7 @@ 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 @@ -65,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`` @@ -169,13 +170,16 @@ safely mutated. 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: @@ -205,9 +209,9 @@ When error occurs, :py:exc:`pyderasn.DecodeError` is raised. 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: @@ -429,7 +433,7 @@ For example, again for CMS, you want to automatically decode structures it may hold. Also, automatically decode ``controlSequence`` of ``PKIResponse``:: - content_info, tail = ContentInfo().decode(data, ctx={"defines_by_path": ( + content_info = ContentInfo().decod(data, ctx={"defines_by_path": ( ( ("contentType",), ((("content",), {id_signedData: SignedData()}),), @@ -570,6 +574,7 @@ _____________ PrintableString _______________ .. autoclass:: pyderasn.PrintableString + :members: __init__ UTCTime _______ @@ -634,6 +639,7 @@ Various .. autoclass:: pyderasn.DecodeError :members: __init__ .. autoclass:: pyderasn.NotEnoughData +.. autoclass:: pyderasn.ExceedingData .. autoclass:: pyderasn.LenIndefForm .. autoclass:: pyderasn.TagMismatch .. autoclass:: pyderasn.InvalidLength @@ -654,6 +660,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 @@ -677,7 +684,7 @@ except ImportError: # pragma: no cover def colored(what, *args, **kwargs): return what -__version__ = "5.5" +__version__ = "5.6" __all__ = ( "Any", @@ -689,6 +696,7 @@ __all__ = ( "DecodeError", "DecodePathDefBy", "Enumerated", + "ExceedingData", "GeneralizedTime", "GeneralString", "GraphicString", @@ -799,6 +807,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 @@ -1264,6 +1284,26 @@ 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` @@ -1322,8 +1362,8 @@ class Obj(object): 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", @@ -1610,10 +1650,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: @@ -1627,8 +1667,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: @@ -2437,7 +2477,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: @@ -2508,7 +2548,7 @@ class BitString(Obj): if t == self.tag: if tag_only: # pragma: no cover return None - return self._decode_chunk(lv, offset, decode_path, ctx) + return self._decode_chunk(lv, offset, decode_path) if t == self.tag_constructed: if not ctx.get("bered", False): raise DecodeError( @@ -2716,13 +2756,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, @@ -2827,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: @@ -2884,7 +2918,7 @@ class OctetString(Obj): if t == self.tag: if tag_only: return None - return self._decode_chunk(lv, offset, decode_path, ctx) + return self._decode_chunk(lv, offset, decode_path) if t == self.tag_constructed: if not ctx.get("bered", False): raise DecodeError( @@ -3212,13 +3246,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) @@ -3489,13 +3517,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") @@ -3556,6 +3578,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 @@ -3618,7 +3646,7 @@ class CommonString(OctetString): * - :py:class:`pyderasn.BMPString` - utf-16-be """ - __slots__ = ("encoding",) + __slots__ = () def _value_sanitize(self, value): value_raw = None @@ -3674,7 +3702,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, @@ -3753,6 +3784,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) @@ -3760,6 +3817,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__ = () @@ -3834,11 +3919,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: @@ -4337,9 +4418,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__ = () @@ -4509,7 +4590,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__( @@ -4533,7 +4614,7 @@ class Any(Obj): optional=self.optional, _decoded=(offset, 0, tlvlen), ) - obj.tag = t + obj.tag = t.tobytes() return obj, tail def __repr__(self): @@ -5287,13 +5368,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: @@ -5653,8 +5728,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: @@ -5672,8 +5747,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: @@ -5752,7 +5827,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