X-Git-Url: http://www.git.cypherpunks.ru/?a=blobdiff_plain;f=pyderasn.py;h=6e772e09b5870c743b5c9228d3027a8d1cef35bf;hb=038b5d9eab3e5dc2a203f064f22caa854f5b68ae;hp=b4a10d9fc04ba14fcf91a86ce21f2490cf5c41fb;hpb=78833daa9b9637827bfb570135e699b5871aefd3;p=pyderasn.git diff --git a/pyderasn.py b/pyderasn.py index b4a10d9..6e772e0 100755 --- a/pyderasn.py +++ b/pyderasn.py @@ -207,6 +207,7 @@ decoding process. Currently available context options: * :ref:`defines_by_path ` +* :ref:`strict_default_existence ` .. _pprinting: @@ -268,11 +269,11 @@ for AlgorithmIdentifier of X.509's ``tbsCertificate.subjectPublicKeyInfo.algorithm.algorithm``:: ( - (('parameters',), { + (("parameters",), { id_ecPublicKey: ECParameters(), id_GostR3410_2001: GostR34102001PublicKeyParameters(), }), - (('..', 'subjectPublicKey'), { + (("..", "subjectPublicKey"), { id_rsaEncryption: RSAPublicKey(), id_GostR3410_2001: OctetString(), }), @@ -288,7 +289,7 @@ Following types can be automatically decoded (DEFINED BY): * :py:class:`pyderasn.BitString` (that is multiple of 8 bits) * :py:class:`pyderasn.OctetString` * :py:class:`pyderasn.SequenceOf`/:py:class:`pyderasn.SetOf` - ``Any``/``OctetString``-s + ``Any``/``BitString``/``OctetString``-s When any of those fields is automatically decoded, then ``.defined`` attribute contains ``(OID, value)`` tuple. ``OID`` tells by which OID it @@ -328,7 +329,7 @@ of ``PKIResponse``:: ( ( "content", - decode_path_defby(id_signedData), + DecodePathDefBy(id_signedData), "encapContentInfo", "eContentType", ), @@ -340,10 +341,10 @@ of ``PKIResponse``:: ( ( "content", - decode_path_defby(id_signedData), + DecodePathDefBy(id_signedData), "encapContentInfo", "eContent", - decode_path_defby(id_cct_PKIResponse), + DecodePathDefBy(id_cct_PKIResponse), "controlSequence", any, "attrType", @@ -357,7 +358,7 @@ of ``PKIResponse``:: ), )) -Pay attention for :py:func:`pyderasn.decode_path_defby` and ``any``. +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. @@ -471,6 +472,8 @@ from collections import namedtuple from collections import OrderedDict from datetime import datetime from math import ceil +from os import environ +from string import digits from six import add_metaclass from six import binary_type @@ -485,6 +488,13 @@ from six import text_type from six.moves import xrange as six_xrange +try: + from termcolor import colored +except ImportError: + def colored(what, *args): + return what + + __all__ = ( "Any", "BitString", @@ -492,8 +502,8 @@ __all__ = ( "Boolean", "BoundsError", "Choice", - "decode_path_defby", "DecodeError", + "DecodePathDefBy", "Enumerated", "GeneralizedTime", "GeneralString", @@ -551,6 +561,8 @@ TagClassReprs = { TagClassPrivate: "PRIVATE ", TagClassUniversal: "UNIV ", } +EOC = b"\x00\x00" +EOC_LEN = len(EOC) ######################################################################## @@ -580,7 +592,7 @@ class DecodeError(Exception): c for c in ( "" if self.klass is None else self.klass.__name__, ( - ("(%s)" % ".".join(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 "", @@ -596,6 +608,10 @@ class NotEnoughData(DecodeError): pass +class LenIndefiniteForm(DecodeError): + pass + + class TagMismatch(DecodeError): pass @@ -796,7 +812,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 LenIndefiniteForm() if byte2int(data[1:]) == 0: raise DecodeError("leading zeros") l = 0 @@ -833,6 +849,7 @@ class Obj(object): "offset", "llen", "vlen", + "bered", ) def __init__( @@ -854,6 +871,7 @@ class Obj(object): self.optional = optional self.offset, self.llen, self.vlen = _decoded self.default = None + self.bered = False @property def ready(self): # pragma: no cover @@ -902,7 +920,7 @@ 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): @@ -911,7 +929,15 @@ class Obj(object): 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 decode( + self, + data, + offset=0, + leavemm=False, + decode_path=(), + ctx=None, + tag_only=False, + ): """Decode the data :param data: either binary or memoryview @@ -919,18 +945,25 @@ class Obj(object): :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 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) :returns: (Obj, remaining data) """ if ctx is None: 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 + obj, tail = result else: try: t, tlen, lv = tag_strip(tlv) @@ -963,12 +996,16 @@ class Obj(object): decode_path=decode_path, offset=offset, ) - obj, tail = self._decode( + result = self._decode( v, offset=offset + tlen + llen, decode_path=decode_path, ctx=ctx, + tag_only=tag_only, ) + if tag_only: + return + obj, tail = result return obj, (tail if leavemm else tail.tobytes()) @property @@ -1000,10 +1037,27 @@ class Obj(object): return self.expl_tlen + self.expl_llen + self.expl_vlen -def decode_path_defby(defined_by): +class DecodePathDefBy(object): """DEFINED BY representation inside decode path """ - return "DEFINED BY (%s)" % defined_by + __slots__ = ("defined_by",) + + 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 + return self.defined_by == their.defined_by + + def __str__(self): + return "DEFINED BY " + str(self.defined_by) + + def __repr__(self): + return "<%s: %s>" % (self.__class__.__name__, self.defined_by) ######################################################################## @@ -1071,49 +1125,75 @@ def _pp( ) -def pp_console_row(pp, oids=None, with_offsets=False, with_blob=True): +def _colorize(what, colour, with_colours, attrs=("bold",)): + return colored(what, colour, attrs=attrs) if with_colours else what + + +def pp_console_row( + pp, + oids=None, + with_offsets=False, + with_blob=True, + with_colours=False, +): cols = [] if with_offsets: - cols.append("%5d%s [%d,%d,%4d]" % ( + col = "%5d%s" % ( pp.offset, ( " " if pp.expl_offset is None else ("-%d" % (pp.offset - pp.expl_offset)) ), - pp.tlen, - pp.llen, - pp.vlen, - )) + ) + 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))) - cols.append("%s:" % pp.decode_path[-1]) + ent = pp.decode_path[-1] + if isinstance(ent, DecodePathDefBy): + cols.append(_colorize("DEFINED BY", "red", with_colours, ("reverse",))) + value = str(ent.defined_by) + if ( + oids is not None and + ent.defined_by.asn1_type_name == + ObjectIdentifier.asn1_type_name and + value in oids + ): + cols.append(_colorize("%s:" % oids[value], "green", with_colours)) + else: + cols.append(_colorize("%s:" % value, "white", with_colours, ("reverse",))) + else: + cols.append(_colorize("%s:" % ent, "yellow", with_colours, ("reverse",))) if pp.expl is not None: klass, _, num = pp.expl - cols.append("[%s%d] EXPLICIT" % (TagClassReprs[klass], num)) + col = "[%s%d] EXPLICIT" % (TagClassReprs[klass], num) + cols.append(_colorize(col, "blue", with_colours)) if pp.impl is not None: klass, _, num = pp.impl - cols.append("[%s%d]" % (TagClassReprs[klass], num)) + col = "[%s%d]" % (TagClassReprs[klass], num) + cols.append(_colorize(col, "blue", with_colours)) if pp.asn1_type_name.replace(" ", "") != pp.obj_name.upper(): - cols.append(pp.obj_name) - cols.append(pp.asn1_type_name) + cols.append(_colorize(pp.obj_name, "magenta", with_colours)) + cols.append(_colorize(pp.asn1_type_name, "cyan", with_colours)) if pp.value is not None: value = pp.value + cols.append(_colorize(value, "white", with_colours, ("reverse",))) if ( oids is not None and pp.asn1_type_name == ObjectIdentifier.asn1_type_name and value in oids ): - value = "%s (%s)" % (oids[value], pp.value) - cols.append(value) + cols.append(_colorize("(%s)" % oids[value], "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("OPTIONAL") + cols.append(_colorize("OPTIONAL", "red", with_colours)) if pp.default: - cols.append("DEFAULT") + cols.append(_colorize("DEFAULT", "red", with_colours)) return " ".join(cols) @@ -1132,7 +1212,7 @@ def pp_console_blob(pp): yield " ".join(cols + [", ".join(pp.blob)]) -def pprint(obj, oids=None, big_blobs=False): +def pprint(obj, oids=None, big_blobs=False, with_colours=False): """Pretty print object :param Obj obj: object you want to pretty print @@ -1141,6 +1221,8 @@ def pprint(obj, oids=None, big_blobs=False): :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 """ def _pprint_pps(pps): for pp in pps: @@ -1151,11 +1233,18 @@ def pprint(obj, oids=None, big_blobs=False): oids=oids, with_offsets=True, with_blob=False, + with_colours=with_colours, ) for row in pp_console_blob(pp): yield row else: - yield pp_console_row(pp, oids=oids, with_offsets=True) + yield pp_console_row( + pp, + oids=oids, + with_offsets=True, + with_blob=True, + with_colours=with_colours, + ) else: for row in _pprint_pps(pp): yield row @@ -1275,7 +1364,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: @@ -1291,6 +1380,8 @@ class Boolean(Obj): decode_path=decode_path, offset=offset, ) + if tag_only: + return try: l, _, v = len_decode(lv) except DecodeError as err: @@ -1315,10 +1406,14 @@ class Boolean(Obj): offset=offset, ) first_octet = byte2int(v) + bered = False if first_octet == 0: value = False elif first_octet == 0xFF: value = True + elif ctx.get("bered", False): + value = True + bered = True else: raise DecodeError( "unacceptable Boolean value", @@ -1334,6 +1429,7 @@ class Boolean(Obj): optional=self.optional, _decoded=(offset, 1, 1), ) + obj.bered = bered return obj, v[1:] def __repr__(self): @@ -1568,7 +1664,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: @@ -1584,6 +1680,8 @@ class Integer(Obj): decode_path=decode_path, offset=offset, ) + if tag_only: + return try: l, llen, v = len_decode(lv) except DecodeError as err: @@ -1707,19 +1805,19 @@ class BitString(Obj): class KeyUsage(BitString): schema = ( - ('digitalSignature', 0), - ('nonRepudiation', 1), - ('keyEncipherment', 2), + ("digitalSignature", 0), + ("nonRepudiation", 1), + ("keyEncipherment", 2), ) - >>> b = KeyUsage(('keyEncipherment', 'nonRepudiation')) + >>> b = KeyUsage(("keyEncipherment", "nonRepudiation")) KeyUsage BIT STRING 3 bits nonRepudiation, keyEncipherment >>> b.named ['nonRepudiation', 'keyEncipherment'] >>> b.specs {'nonRepudiation': 1, 'digitalSignature': 0, 'keyEncipherment': 2} """ - __slots__ = ("specs", "defined") + __slots__ = ("tag_constructed", "specs", "defined") tag_default = tag_encode(3) asn1_type_name = "BIT STRING" @@ -1757,6 +1855,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: @@ -1902,22 +2006,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, ctx): try: l, llen, v = len_decode(lv) except DecodeError as err: @@ -1956,7 +2045,7 @@ class BitString(Obj): decode_path=decode_path, offset=offset, ) - if byte2int(v[-1:]) & ((1 << pad_size) - 1) != 0: + if byte2int(v[l - 1:l]) & ((1 << pad_size) - 1) != 0: raise DecodeError( "invalid pad", klass=self.__class__, @@ -1975,6 +2064,128 @@ class BitString(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 + return self._decode_chunk(lv, offset, decode_path, ctx) + if t == self.tag_constructed: + if not ctx.get("bered", False): + raise DecodeError( + msg="unallowed BER constructed encoding", + decode_path=decode_path, + offset=offset, + ) + if tag_only: + return + eoc_expected = False + try: + l, llen, v = len_decode(lv) + except LenIndefiniteForm: + llen, l, v = 1, 0, lv[1:] + eoc_expected = True + except DecodeError as err: + raise err.__class__( + msg=err.msg, + klass=self.__class__, + decode_path=decode_path, + offset=offset, + ) + if l > 0 and l > len(v): + raise NotEnoughData( + "encoded length is longer than data", + klass=self.__class__, + decode_path=decode_path, + offset=offset, + ) + if not eoc_expected 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 eoc_expected: + if v[:EOC_LEN].tobytes() == EOC: + break + else: + if vlen == l: + break + if vlen > l: + raise DecodeError( + msg="chunk out of bounds", + decode_path=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, + ) + except TagMismatch: + raise DecodeError( + msg="expected BitString encoded chunk", + 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( + msg="no chunks", + 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( + msg="BitString chunk is not multiple of 8 bit", + 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 eoc_expected else 0)), + ) + obj.bered = True + return obj, v[EOC_LEN if eoc_expected else 0:] + raise TagMismatch( + klass=self.__class__, + decode_path=decode_path, + offset=offset, + ) + def __repr__(self): return pp_console_row(next(self.pps())) @@ -2008,7 +2219,7 @@ class BitString(Obj): defined_by, defined = self.defined or (None, None) if defined_by is not None: yield defined.pps( - decode_path=decode_path + (decode_path_defby(defined_by),) + decode_path=decode_path + (DecodePathDefBy(defined_by),) ) @@ -2028,7 +2239,7 @@ class OctetString(Obj): >>> OctetString(b"hell", bounds=(4, 4)) OCTET STRING 4 bytes 68656c6c """ - __slots__ = ("_bound_min", "_bound_max", "defined") + __slots__ = ("tag_constructed", "_bound_min", "_bound_max", "defined") tag_default = tag_encode(4) asn1_type_name = "OCTET STRING" @@ -2077,6 +2288,12 @@ 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): @@ -2154,22 +2371,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, ctx): try: l, llen, v = len_decode(lv) except DecodeError as err: @@ -2197,6 +2399,13 @@ class OctetString(Obj): optional=self.optional, _decoded=(offset, llen, l), ) + 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), @@ -2206,6 +2415,129 @@ 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 + return self._decode_chunk(lv, offset, decode_path, ctx) + if t == self.tag_constructed: + if not ctx.get("bered", False): + raise DecodeError( + msg="unallowed BER constructed encoding", + decode_path=decode_path, + offset=offset, + ) + if tag_only: + return + eoc_expected = False + try: + l, llen, v = len_decode(lv) + except LenIndefiniteForm: + llen, l, v = 1, 0, lv[1:] + eoc_expected = True + except DecodeError as err: + raise err.__class__( + msg=err.msg, + klass=self.__class__, + decode_path=decode_path, + offset=offset, + ) + if l > 0 and l > len(v): + raise NotEnoughData( + "encoded length is longer than data", + klass=self.__class__, + decode_path=decode_path, + offset=offset, + ) + if not eoc_expected 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 eoc_expected: + if v[:EOC_LEN].tobytes() == EOC: + break + else: + if vlen == l: + break + if vlen > l: + raise DecodeError( + msg="chunk out of bounds", + decode_path=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, + ) + except TagMismatch: + raise DecodeError( + msg="expected OctetString encoded chunk", + 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( + msg="no chunks", + decode_path=decode_path, + offset=offset, + ) + 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 eoc_expected 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.bered = True + return obj, v[EOC_LEN if eoc_expected else 0:] + raise TagMismatch( + klass=self.__class__, + decode_path=decode_path, + offset=offset, + ) + def __repr__(self): return pp_console_row(next(self.pps())) @@ -2232,7 +2564,7 @@ class OctetString(Obj): defined_by, defined = self.defined or (None, None) if defined_by is not None: yield defined.pps( - decode_path=decode_path + (decode_path_defby(defined_by),) + decode_path=decode_path + (DecodePathDefBy(defined_by),) ) @@ -2303,7 +2635,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: @@ -2319,6 +2651,8 @@ class Null(Obj): decode_path=decode_path, offset=offset, ) + if tag_only: + return try: l, _, v = len_decode(lv) except DecodeError as err: @@ -2548,7 +2882,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: @@ -2564,6 +2898,8 @@ class ObjectIdentifier(Obj): decode_path=decode_path, offset=offset, ) + if tag_only: + return try: l, llen, v = len_decode(lv) except DecodeError as err: @@ -2753,7 +3089,7 @@ class CommonString(OctetString): >>> PrintableString("привет мир") Traceback (most recent call last): - UnicodeEncodeError: 'ascii' codec can't encode characters in position 0-5: ordinal not in range(128) + pyderasn.DecodeError: 'ascii' codec can't encode characters in position 0-5: ordinal not in range(128) >>> BMPString("ада", bounds=(2, 2)) Traceback (most recent call last): @@ -2809,14 +3145,17 @@ class CommonString(OctetString): value_raw = value else: raise InvalidValueType((self.__class__, text_type, binary_type)) - value_raw = ( - value_decoded.encode(self.encoding) - if value_raw is None else value_raw - ) - value_decoded = ( - value_raw.decode(self.encoding) - if value_decoded is None else value_decoded - ) + try: + value_raw = ( + value_decoded.encode(self.encoding) + if value_raw is None else value_raw + ) + value_decoded = ( + value_raw.decode(self.encoding) + if value_decoded is None else value_decoded + ) + except (UnicodeEncodeError, UnicodeDecodeError) as err: + raise DecodeError(str(err)) if not self._bound_min <= len(value_decoded) <= self._bound_max: raise BoundsError( self._bound_min, @@ -2878,6 +3217,13 @@ class NumericString(CommonString): tag_default = tag_encode(18) encoding = "ascii" asn1_type_name = "NumericString" + allowable_chars = set(digits.encode("ascii")) + + def _value_sanitize(self, value): + value = super(NumericString, self)._value_sanitize(value) + if not set(value) <= self.allowable_chars: + raise DecodeError("non-numeric value") + return value class PrintableString(CommonString): @@ -3152,8 +3498,8 @@ class Choice(Obj): class GeneralName(Choice): schema = ( - ('rfc822Name', IA5String(impl=tag_ctxp(1))), - ('dNSName', IA5String(impl=tag_ctxp(2))), + ("rfc822Name", IA5String(impl=tag_ctxp(1))), + ("dNSName", IA5String(impl=tag_ctxp(2))), ) >>> gn = GeneralName() @@ -3315,32 +3661,45 @@ class Choice(Obj): self._assert_ready() return self._value[1].encode() - def _decode(self, tlv, offset, decode_path, ctx): + def _decode(self, tlv, offset, decode_path, ctx, tag_only): for choice, spec in self.specs.items(): + 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, ) 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: + return + value, tail = spec.decode( + tlv, offset=offset, + leavemm=True, + decode_path=sub_decode_path, + ctx=ctx, + ) + obj = self.__class__( + schema=self.specs, + expl=self._expl, + default=self.default, + optional=self.optional, + _decoded=(offset, 0, value.tlvlen), ) + obj._value = (choice, value) + return obj, tail def __repr__(self): value = pp_console_row(next(self.pps())) @@ -3490,7 +3849,7 @@ 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) l, llen, v = len_decode(lv) @@ -3544,7 +3903,7 @@ class Any(Obj): defined_by, defined = self.defined or (None, None) if defined_by is not None: yield defined.pps( - decode_path=decode_path + (decode_path_defby(defined_by),) + decode_path=decode_path + (DecodePathDefBy(defined_by),) ) @@ -3617,7 +3976,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 @@ -3641,7 +4000,17 @@ class Sequence(Obj): >>> tbs = TBSCertificate() >>> tbs["version"] = Version("v2") # no need to explicitly add ``expl`` - You can know if value exists/set in the sequence and take its value: + Assign ``None`` to remove value from sequence. + + You can set values in Sequence during its initialization: + + >>> AlgorithmIdentifier(( + ("algorithm", ObjectIdentifier("1.2.3")), + ("parameters", Any(Null())) + )) + AlgorithmIdentifier SEQUENCE[OBJECT IDENTIFIER 1.2.3, 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) @@ -3660,13 +4029,18 @@ class Sequence(Obj): All defaulted values are always optional. + .. _strict_default_existence_ctx: + .. warning:: When decoded DER contains defaulted value inside, then - technically this is not valid DER encoding. But we allow - and pass it. Of course reencoding of that kind of DER will + 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). + 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. Two sequences are equal if they have equal specification (schema), implicit/explicit tagging and the same values. @@ -3693,9 +4067,17 @@ 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 @@ -3703,11 +4085,6 @@ class Sequence(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 - @property def ready(self): for name, spec in self.specs.items(): @@ -3804,7 +4181,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: @@ -3820,6 +4197,8 @@ class Sequence(Obj): decode_path=decode_path, offset=offset, ) + if tag_only: + return try: l, llen, v = len_decode(lv) except DecodeError as err: @@ -3863,11 +4242,14 @@ class Sequence(Obj): for i, _value in enumerate(value): sub_sub_decode_path = sub_decode_path + ( str(i), - decode_path_defby(defined_by), + DecodePathDefBy(defined_by), ) defined_value, defined_tail = defined_spec.decode( memoryview(bytes(_value)), - sub_offset + value.tlen + value.llen, + sub_offset + ( + (value.tlen + value.llen + value.expl_tlen + value.expl_llen) + if value.expled else (value.tlen + value.llen) + ), leavemm=True, decode_path=sub_sub_decode_path, ctx=ctx, @@ -3883,16 +4265,19 @@ class Sequence(Obj): else: defined_value, defined_tail = defined_spec.decode( memoryview(bytes(value)), - sub_offset + value.tlen + value.llen, + sub_offset + ( + (value.tlen + value.llen + value.expl_tlen + value.expl_llen) + if value.expled else (value.tlen + value.llen) + ), leavemm=True, - decode_path=sub_decode_path + (decode_path_defby(defined_by),), + decode_path=sub_decode_path + (DecodePathDefBy(defined_by),), ctx=ctx, ) if len(defined_tail) > 0: raise DecodeError( "remaining data", klass=self.__class__, - decode_path=sub_decode_path + (decode_path_defby(defined_by),), + decode_path=sub_decode_path + (DecodePathDefBy(defined_by),), offset=offset, ) value.defined = (defined_by, defined_value) @@ -3900,9 +4285,15 @@ class Sequence(Obj): sub_offset += (value.expl_tlvlen if value.expled else value.tlvlen) v = v_tail if spec.default is not None and value == spec.default: - # Encoded default values are not valid in DER, - # but we allow that anyway - continue + if ctx.get("strict_default_existence", False): + 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", ()) @@ -3986,7 +4377,7 @@ 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 _decode(self, tlv, offset, decode_path, ctx, tag_only): try: t, tlen, lv = tag_strip(tlv) except DecodeError as err: @@ -4002,6 +4393,8 @@ class Set(Sequence): decode_path=decode_path, offset=offset, ) + if tag_only: + return try: l, llen, v = len_decode(lv) except DecodeError as err: @@ -4023,23 +4416,18 @@ class Set(Sequence): specs_items = self.specs.items while len(v) > 0: for name, spec in 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, ) 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( @@ -4047,6 +4435,20 @@ 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, + ) + 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 obj = self.__class__( schema=self.specs, impl=self.tag, @@ -4239,7 +4641,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): try: t, tlen, lv = tag_strip(tlv) except DecodeError as err: @@ -4255,6 +4657,8 @@ class SequenceOf(Obj): decode_path=decode_path, offset=offset, ) + if tag_only: + return try: l, llen, v = len_decode(lv) except DecodeError as err: @@ -4377,7 +4781,7 @@ def generic_decoder(): # pragma: no cover __slots__ = () schema = choice - def pprint_any(obj, oids=None): + def pprint_any(obj, oids=None, with_colours=False): def _pprint_pps(pps): for pp in pps: if hasattr(pp, "_fields"): @@ -4391,6 +4795,7 @@ def generic_decoder(): # pragma: no cover oids=oids, with_offsets=True, with_blob=False, + with_colours=with_colours, ) for row in pp_console_blob(pp): yield row @@ -4445,7 +4850,11 @@ def main(): # pragma: no cover {"defines_by_path": obj_by_path(args.defines_by_path)} ), ) - print(pprinter(obj, oids=oids)) + print(pprinter( + obj, + oids=oids, + with_colours=True if environ.get("NO_COLOR") is None else False, + )) if tail != b"": print("\nTrailing data: %s" % hexenc(tail))