X-Git-Url: http://www.git.cypherpunks.ru/?a=blobdiff_plain;f=pyderasn.py;h=581af5f87ee2923ac183dadb5817ee2cd22ee808;hb=d3141c4def089fc213dac0fe788ce03c67c191ea;hp=88579558b731b42143a1363c07226e54f274a87e;hpb=5bbbeb6269f0402dbeb51200120f1057e2f6a9b5;p=pyderasn.git diff --git a/pyderasn.py b/pyderasn.py index 8857955..581af5f 100755 --- a/pyderasn.py +++ b/pyderasn.py @@ -515,6 +515,7 @@ Various ------- .. autofunction:: pyderasn.abs_decode_path +.. autofunction:: pyderasn.colonize_hex .. autofunction:: pyderasn.hexenc .. autofunction:: pyderasn.hexdec .. autofunction:: pyderasn.tag_encode @@ -539,9 +540,11 @@ 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 six import add_metaclass @@ -554,12 +557,13 @@ from six import iterbytes 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: +except ImportError: # pragma: no cover def colored(what, *args): return what @@ -641,7 +645,11 @@ 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 @@ -696,7 +704,7 @@ class InvalidOID(DecodeError): pass -class ObjUnknown(ValueError): +class ObjUnknown(ASN1Error): def __init__(self, name): super(ObjUnknown, self).__init__() self.name = name @@ -708,7 +716,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 @@ -720,7 +728,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 @@ -734,7 +742,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 @@ -958,16 +966,16 @@ class Obj(object): """ raise NotImplementedError() + def _assert_ready(self): + 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 - def _assert_ready(self): - if not self.ready: - raise ObjNotReady(self.__class__.__name__) - @property def decoded(self): """Is object decoded? @@ -1022,6 +1030,7 @@ class Obj(object): decode_path=(), ctx=None, tag_only=False, + _ctx_immutable=True, ): """Decode the data @@ -1029,14 +1038,17 @@ class Obj(object): :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 ``ctx`` before using it :returns: (Obj, remaining data) """ if ctx is None: ctx = {} + elif _ctx_immutable: + ctx = copy(ctx) tlv = memoryview(data) if self._expl is None: result = self._decode( @@ -1084,7 +1096,7 @@ class Obj(object): ctx=ctx, tag_only=tag_only, ) - if tag_only: + if tag_only: # pragma: no cover return obj, tail = result eoc_expected, tail = tail[:EOC_LEN], tail[EOC_LEN:] @@ -1119,7 +1131,7 @@ class Obj(object): ctx=ctx, tag_only=tag_only, ) - if tag_only: + if tag_only: # pragma: no cover return obj, tail = result if obj.tlvlen < l and not ctx.get("allow_expl_oob", False): @@ -1170,7 +1182,10 @@ class Obj(object): return self.expl_tlvlen if self.expled else self.tlvlen def pps_lenindef(self, decode_path): - if self.lenindef: + 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="", @@ -1227,6 +1242,7 @@ class DecodePathDefBy(object): ######################################################################## PP = namedtuple("PP", ( + "obj", "asn1_type_name", "obj_name", "decode_path", @@ -1252,6 +1268,7 @@ PP = namedtuple("PP", ( def _pp( + obj=None, asn1_type_name="unknown", obj_name="unknown", decode_path=(), @@ -1275,6 +1292,7 @@ def _pp( bered=False, ): return PP( + obj, asn1_type_name, obj_name, decode_path, @@ -1303,6 +1321,12 @@ 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 range(0, len(hexed), 2)) + + def pp_console_row( pp, oids=None, @@ -1373,6 +1397,15 @@ def pp_console_row( value in oids ): cols.append(_colourize("(%s)" % oids[value], "green", with_colours)) + 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)) @@ -1400,9 +1433,7 @@ def pp_console_blob(pp, decode_path_len_decrease=0): blob = hexenc(pp.blob).upper() for i in range(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)]) @@ -1655,6 +1686,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, @@ -1982,6 +2014,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, @@ -2314,7 +2347,7 @@ class BitString(Obj): offset=offset, ) if t == self.tag: - if tag_only: + if tag_only: # pragma: no cover return return self._decode_chunk(lv, offset, decode_path, ctx) if t == self.tag_constructed: @@ -2325,7 +2358,7 @@ class BitString(Obj): decode_path=decode_path, offset=offset, ) - if tag_only: + if tag_only: # pragma: no cover return lenindef = False try: @@ -2379,6 +2412,7 @@ class BitString(Obj): decode_path=sub_decode_path, leavemm=True, ctx=ctx, + _ctx_immutable=False, ) except TagMismatch: raise DecodeError( @@ -2443,6 +2477,7 @@ class BitString(Obj): 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, @@ -2743,6 +2778,7 @@ class OctetString(Obj): decode_path=sub_decode_path, leavemm=True, ctx=ctx, + _ctx_immutable=False, ) except TagMismatch: raise DecodeError( @@ -2793,6 +2829,7 @@ class OctetString(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, @@ -2907,7 +2944,7 @@ class Null(Obj): decode_path=decode_path, offset=offset, ) - if tag_only: + if tag_only: # pragma: no cover return try: l, _, v = len_decode(lv) @@ -2938,6 +2975,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, @@ -3158,7 +3196,7 @@ class ObjectIdentifier(Obj): decode_path=decode_path, offset=offset, ) - if tag_only: + if tag_only: # pragma: no cover return try: l, llen, v = len_decode(lv) @@ -3228,6 +3266,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, @@ -3454,6 +3493,7 @@ class CommonString(OctetString): if self.ready: value = hexenc(bytes(self)) if no_unicode else self.__unicode__() yield _pp( + obj=self, asn1_type_name=self.asn1_type_name, obj_name=self.__class__.__name__, decode_path=decode_path, @@ -3485,29 +3525,57 @@ 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 set(six_unichr(c) for c in self._allowable_chars) + + +class NumericString(AllowableCharsMixin, CommonString): """Numeric string - Its value is properly sanitized: only ASCII digits can be stored. + Its value is properly sanitized: only ASCII digits with spaces can + be stored. + + >>> NumericString().allowable_chars + set(['3', '4', '7', '5', '1', '0', '8', '9', ' ', '6', '2']) """ __slots__ = () tag_default = tag_encode(18) encoding = "ascii" asn1_type_name = "NumericString" - allowable_chars = set(digits.encode("ascii")) + _allowable_chars = set(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 set(value) <= self._allowable_chars: raise DecodeError("non-numeric value") return value -class PrintableString(CommonString): +class PrintableString(AllowableCharsMixin, CommonString): + """Printable string + + Its value is properly sanitized: see X.680 41.4 table 10. + + >>> PrintableString().allowable_chars + >>> set([' ', "'", ..., 'z']) + """ __slots__ = () tag_default = tag_encode(19) encoding = "ascii" asn1_type_name = "PrintableString" + _allowable_chars = set( + (ascii_letters + digits + " '()+,-./:=?").encode("ascii") + ) + + def _value_sanitize(self, value): + value = super(PrintableString, self)._value_sanitize(value) + if not set(value) <= self._allowable_chars: + raise DecodeError("non-printable value") + return value class TeletexString(CommonString): @@ -3658,6 +3726,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, @@ -3971,6 +4040,7 @@ class Choice(Obj): decode_path=sub_decode_path, ctx=ctx, tag_only=True, + _ctx_immutable=False, ) except TagMismatch: continue @@ -3981,7 +4051,7 @@ class Choice(Obj): decode_path=decode_path, offset=offset, ) - if tag_only: + if tag_only: # pragma: no cover return value, tail = spec.decode( tlv, @@ -3989,6 +4059,7 @@ class Choice(Obj): leavemm=True, decode_path=sub_decode_path, ctx=ctx, + _ctx_immutable=False, ) obj = self.__class__( schema=self.specs, @@ -4008,6 +4079,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, @@ -4190,6 +4262,7 @@ class Any(Obj): decode_path=decode_path + (str(chunk_i),), leavemm=True, ctx=ctx, + _ctx_immutable=False, ) vlen += chunk.tlvlen sub_offset += chunk.tlvlen @@ -4234,6 +4307,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, @@ -4283,8 +4357,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 @@ -4555,7 +4628,7 @@ class Sequence(Obj): decode_path=decode_path, offset=offset, ) - if tag_only: + if tag_only: # pragma: no cover return lenindef = False ctx_bered = ctx.get("bered", False) @@ -4606,6 +4679,7 @@ class Sequence(Obj): leavemm=True, decode_path=sub_decode_path, ctx=ctx, + _ctx_immutable=False, ) except TagMismatch: if spec.optional: @@ -4630,6 +4704,7 @@ class Sequence(Obj): leavemm=True, decode_path=sub_sub_decode_path, ctx=ctx, + _ctx_immutable=False, ) if len(defined_tail) > 0: raise DecodeError( @@ -4649,6 +4724,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( @@ -4730,6 +4806,7 @@ class Sequence(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, @@ -4850,6 +4927,7 @@ class Set(Sequence): decode_path=sub_decode_path, ctx=ctx, tag_only=True, + _ctx_immutable=False, ) except TagMismatch: continue @@ -4866,6 +4944,7 @@ class Set(Sequence): leavemm=True, decode_path=sub_decode_path, ctx=ctx, + _ctx_immutable=False, ) value_len = value.fulllen if value_prev.tobytes() > v[:value_len].tobytes(): @@ -5175,6 +5254,7 @@ class SequenceOf(Obj): leavemm=True, decode_path=sub_decode_path, ctx=ctx, + _ctx_immutable=False, ) value_len = value.fulllen if ordering_check: @@ -5232,6 +5312,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,