X-Git-Url: http://www.git.cypherpunks.ru/?a=blobdiff_plain;f=pyderasn.py;h=21b4306319db9630c936577b5a7d8595afa717d3;hb=cc4cd876b3603073deba6bec63bb3b8b02fecd89;hp=0d3a6637840115c15fdcf90bc898f095f0c640f7;hpb=517f9ab0bb42c9d89fade2aa69ec5da9501efc0a;p=pyderasn.git diff --git a/pyderasn.py b/pyderasn.py index 0d3a663..21b4306 100755 --- a/pyderasn.py +++ b/pyderasn.py @@ -1,5 +1,6 @@ #!/usr/bin/env python # coding: utf-8 +# cython: language_level=3 # PyDERASN -- Python ASN.1 DER/BER codec with abstract structures # Copyright (C) 2017-2020 Sergey Matveev # @@ -665,6 +666,7 @@ from math import ceil from os import environ from string import ascii_letters from string import digits +from sys import version_info from unicodedata import category as unicat from six import add_metaclass @@ -689,7 +691,7 @@ except ImportError: # pragma: no cover def colored(what, *args, **kwargs): return what -__version__ = "6.1" +__version__ = "6.3" __all__ = ( "Any", @@ -763,6 +765,20 @@ EOC = b"\x00\x00" EOC_LEN = len(EOC) LENINDEF = b"\x80" # length indefinite mark LENINDEF_PP_CHAR = "I" if PY2 else "∞" +NAMEDTUPLE_KWARGS = {} if version_info < (3, 6) else {"module": __name__} +SET01 = frozenset("01") +DECIMALS = frozenset(digits) +DECIMAL_SIGNS = ".," + + +def pureint(value): + if not set(value) <= DECIMALS: + raise ValueError("non-pure integer") + return int(value) + +def fractions2float(fractions_raw): + pureint(fractions_raw) + return float("0." + fractions_raw) ######################################################################## @@ -1479,7 +1495,7 @@ PP = namedtuple("PP", ( "lenindef", "ber_encoded", "bered", -)) +), **NAMEDTUPLE_KWARGS) def _pp( @@ -1629,9 +1645,9 @@ def pp_console_row( with_colours, )) if with_blob: - if isinstance(pp.blob, binary_type): + if pp.blob.__class__ == binary_type: cols.append(hexenc(pp.blob)) - elif isinstance(pp.blob, tuple): + elif pp.blob.__class__ == tuple: cols.append(", ".join(pp.blob)) if pp.optional: cols.append(_colourize("OPTIONAL", "red", with_colours)) @@ -1651,12 +1667,12 @@ def pp_console_blob(pp, decode_path_len_decrease=0): decode_path_len = len(pp.decode_path) - decode_path_len_decrease if decode_path_len > 0: cols.append(" ." * (decode_path_len + 1)) - if isinstance(pp.blob, binary_type): + if pp.blob.__class__ == binary_type: blob = hexenc(pp.blob).upper() for i in six_xrange(0, len(blob), 32): chunk = blob[i:i + 32] yield " ".join(cols + [colonize_hex(chunk)]) - elif isinstance(pp.blob, tuple): + elif pp.blob.__class__ == tuple: yield " ".join(cols + [", ".join(pp.blob)]) @@ -1739,7 +1755,7 @@ BooleanState = namedtuple("BooleanState", ( "expl_lenindef", "lenindef", "ber_encoded", -)) +), **NAMEDTUPLE_KWARGS) class Boolean(Obj): @@ -1786,7 +1802,7 @@ class Boolean(Obj): self._value = default def _value_sanitize(self, value): - if isinstance(value, bool): + if value.__class__ == bool: return value if issubclass(value.__class__, Boolean): return value._value @@ -1835,7 +1851,7 @@ class Boolean(Obj): return self._value def __eq__(self, their): - if isinstance(their, bool): + if their.__class__ == bool: return self._value == their if not issubclass(their.__class__, Boolean): return False @@ -1983,7 +1999,7 @@ IntegerState = namedtuple("IntegerState", ( "expl_lenindef", "lenindef", "ber_encoded", -)) +), **NAMEDTUPLE_KWARGS) class Integer(Obj): @@ -2049,7 +2065,7 @@ class Integer(Obj): super(Integer, self).__init__(impl, expl, default, optional, _decoded) self._value = value specs = getattr(self, "schema", {}) if _specs is None else _specs - self.specs = specs if isinstance(specs, dict) else dict(specs) + self.specs = specs if specs.__class__ == dict else dict(specs) self._bound_min, self._bound_max = getattr( self, "bounds", @@ -2073,7 +2089,7 @@ class Integer(Obj): pass elif issubclass(value.__class__, Integer): value = value._value - elif isinstance(value, str): + elif value.__class__ == str: value = self.specs.get(value) if value is None: raise ObjUnknown("integer value: %s" % value) @@ -2336,7 +2352,6 @@ class Integer(Obj): yield pp -SET01 = frozenset(("0", "1")) BitStringState = namedtuple("BitStringState", ( "version", "specs", @@ -2353,7 +2368,7 @@ BitStringState = namedtuple("BitStringState", ( "ber_encoded", "tag_constructed", "defined", -)) +), **NAMEDTUPLE_KWARGS) class BitString(Obj): @@ -2431,7 +2446,7 @@ class BitString(Obj): """ super(BitString, self).__init__(impl, expl, default, optional, _decoded) specs = getattr(self, "schema", {}) if _specs is None else _specs - self.specs = specs if isinstance(specs, dict) else dict(specs) + self.specs = specs if specs.__class__ == dict else dict(specs) self._value = None if value is None else self._value_sanitize(value) if default is not None: default = self._value_sanitize(default) @@ -2477,14 +2492,14 @@ class BitString(Obj): len(value) * 4, hexdec(value + ("" if len(value) % 2 == 0 else "0")), ) - if isinstance(value, binary_type): + if value.__class__ == binary_type: return (len(value) * 8, value) raise InvalidValueType((self.__class__, string_types, binary_type)) - if isinstance(value, tuple): + if value.__class__ == tuple: if ( len(value) == 2 and isinstance(value[0], integer_types) and - isinstance(value[1], binary_type) + value[1].__class__ == binary_type ): return value bits = [] @@ -2559,7 +2574,7 @@ class BitString(Obj): return self._value[1] def __eq__(self, their): - if isinstance(their, bytes): + if their.__class__ == bytes: return self._value[1] == their if not issubclass(their.__class__, BitString): return False @@ -2591,7 +2606,7 @@ class BitString(Obj): ) def __getitem__(self, key): - if isinstance(key, int): + if key.__class__ == int: bit_len, octets = self._value if key >= bit_len: return False @@ -2616,64 +2631,6 @@ class BitString(Obj): octets, )) - def _decode_chunk(self, lv, offset, decode_path): - try: - l, llen, v = len_decode(lv) - except DecodeError as err: - raise err.__class__( - msg=err.msg, - klass=self.__class__, - decode_path=decode_path, - offset=offset, - ) - if l > len(v): - raise NotEnoughData( - "encoded length is longer than data", - klass=self.__class__, - decode_path=decode_path, - offset=offset, - ) - if l == 0: - raise NotEnoughData( - "zero length", - klass=self.__class__, - decode_path=decode_path, - offset=offset, - ) - pad_size = byte2int(v) - if l == 1 and pad_size != 0: - raise DecodeError( - "invalid empty value", - klass=self.__class__, - decode_path=decode_path, - offset=offset, - ) - if pad_size > 7: - raise DecodeError( - "too big pad", - klass=self.__class__, - decode_path=decode_path, - offset=offset, - ) - if byte2int(v[l - 1:l]) & ((1 << pad_size) - 1) != 0: - raise DecodeError( - "invalid pad", - klass=self.__class__, - decode_path=decode_path, - offset=offset, - ) - v, tail = v[:l], v[l:] - obj = self.__class__( - value=((len(v) - 1) * 8 - pad_size, v[1:].tobytes()), - impl=self.tag, - expl=self._expl, - default=self.default, - optional=self.optional, - _specs=self.specs, - _decoded=(offset, llen, l), - ) - return obj, tail - def _decode(self, tlv, offset, decode_path, ctx, tag_only): try: t, tlen, lv = tag_strip(tlv) @@ -2687,23 +2644,8 @@ class BitString(Obj): if t == self.tag: if tag_only: # pragma: no cover return None - return self._decode_chunk(lv, offset, decode_path) - if t == self.tag_constructed: - if not ctx.get("bered", False): - raise DecodeError( - "unallowed BER constructed encoding", - klass=self.__class__, - decode_path=decode_path, - offset=offset, - ) - if tag_only: # pragma: no cover - return None - lenindef = False try: l, llen, v = len_decode(lv) - except LenIndefForm: - llen, l, v = 1, 0, lv[1:] - lenindef = True except DecodeError as err: raise err.__class__( msg=err.msg, @@ -2718,90 +2660,160 @@ class BitString(Obj): decode_path=decode_path, offset=offset, ) - if not lenindef and l == 0: + if 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 lenindef: - if v[:EOC_LEN].tobytes() == EOC: - break - else: - if vlen == l: - break - if vlen > l: - raise DecodeError( - "chunk out of bounds", - klass=self.__class__, - decode_path=decode_path + (str(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, - _ctx_immutable=False, - ) - except TagMismatch: - raise DecodeError( - "expected BitString encoded chunk", - klass=self.__class__, - 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: + pad_size = byte2int(v) + if l == 1 and pad_size != 0: raise DecodeError( - "no chunks", + "invalid empty value", klass=self.__class__, 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( - "BitString chunk is not multiple of 8 bits", - klass=self.__class__, - 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 + if pad_size > 7: + raise DecodeError( + "too big pad", + klass=self.__class__, + decode_path=decode_path, + offset=offset, + ) + if byte2int(v[l - 1:l]) & ((1 << pad_size) - 1) != 0: + raise DecodeError( + "invalid pad", + klass=self.__class__, + decode_path=decode_path, + offset=offset, + ) + v, tail = v[:l], v[l:] obj = self.__class__( - value=(bit_len, b"".join(values)), + value=((len(v) - 1) * 8 - pad_size, v[1:].tobytes()), impl=self.tag, expl=self._expl, default=self.default, optional=self.optional, _specs=self.specs, - _decoded=(offset, llen, vlen + (EOC_LEN if lenindef else 0)), + _decoded=(offset, llen, l), ) - obj.lenindef = lenindef - obj.ber_encoded = True - return obj, (v[EOC_LEN:] if lenindef else v) - raise TagMismatch( - klass=self.__class__, - decode_path=decode_path, - offset=offset, + return obj, tail + if t != self.tag_constructed: + raise TagMismatch( + klass=self.__class__, + decode_path=decode_path, + offset=offset, + ) + if not ctx.get("bered", False): + raise DecodeError( + "unallowed BER constructed encoding", + klass=self.__class__, + decode_path=decode_path, + offset=offset, + ) + if tag_only: # pragma: no cover + return None + lenindef = False + try: + l, llen, v = len_decode(lv) + except LenIndefForm: + llen, l, v = 1, 0, lv[1:] + lenindef = True + except DecodeError as err: + raise err.__class__( + msg=err.msg, + klass=self.__class__, + decode_path=decode_path, + offset=offset, + ) + if l > len(v): + raise NotEnoughData( + "encoded length is longer than data", + klass=self.__class__, + decode_path=decode_path, + offset=offset, + ) + if not lenindef 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 lenindef: + if v[:EOC_LEN].tobytes() == EOC: + break + else: + if vlen == l: + break + if vlen > l: + raise DecodeError( + "chunk out of bounds", + klass=self.__class__, + decode_path=decode_path + (str(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, + _ctx_immutable=False, + ) + except TagMismatch: + raise DecodeError( + "expected BitString encoded chunk", + klass=self.__class__, + 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( + "no chunks", + klass=self.__class__, + 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( + "BitString chunk is not multiple of 8 bits", + klass=self.__class__, + 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 lenindef else 0)), ) + obj.lenindef = lenindef + obj.ber_encoded = True + return obj, (v[EOC_LEN:] if lenindef else v) def __repr__(self): return pp_console_row(next(self.pps())) @@ -2864,7 +2876,7 @@ OctetStringState = namedtuple("OctetStringState", ( "ber_encoded", "tag_constructed", "defined", -)) +), **NAMEDTUPLE_KWARGS) class OctetString(Obj): @@ -2943,7 +2955,7 @@ class OctetString(Obj): ) def _value_sanitize(self, value): - if isinstance(value, binary_type): + if value.__class__ == binary_type: pass elif issubclass(value.__class__, OctetString): value = value._value @@ -3000,7 +3012,7 @@ class OctetString(Obj): return self._value def __eq__(self, their): - if isinstance(their, binary_type): + if their.__class__ == binary_type: return self._value == their if not issubclass(their.__class__, OctetString): return False @@ -3042,51 +3054,6 @@ class OctetString(Obj): self._value, )) - def _decode_chunk(self, lv, offset, decode_path, ctx): - try: - l, llen, v = len_decode(lv) - except DecodeError as err: - raise err.__class__( - msg=err.msg, - klass=self.__class__, - decode_path=decode_path, - offset=offset, - ) - if l > len(v): - raise NotEnoughData( - "encoded length is longer than data", - klass=self.__class__, - decode_path=decode_path, - offset=offset, - ) - v, tail = v[:l], v[l:] - try: - obj = self.__class__( - value=v.tobytes(), - bounds=(self._bound_min, self._bound_max), - impl=self.tag, - expl=self._expl, - default=self.default, - optional=self.optional, - _decoded=(offset, llen, l), - ctx=ctx, - ) - 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, - ) - return obj, tail - def _decode(self, tlv, offset, decode_path, ctx, tag_only): try: t, tlen, lv = tag_strip(tlv) @@ -3100,23 +3067,8 @@ class OctetString(Obj): if t == self.tag: if tag_only: return None - return self._decode_chunk(lv, offset, decode_path, ctx) - if t == self.tag_constructed: - if not ctx.get("bered", False): - raise DecodeError( - "unallowed BER constructed encoding", - klass=self.__class__, - decode_path=decode_path, - offset=offset, - ) - if tag_only: - return None - lenindef = False try: l, llen, v = len_decode(lv) - except LenIndefForm: - llen, l, v = 1, 0, lv[1:] - lenindef = True except DecodeError as err: raise err.__class__( msg=err.msg, @@ -3131,53 +3083,16 @@ class OctetString(Obj): decode_path=decode_path, offset=offset, ) - chunks = [] - sub_offset = offset + tlen + llen - vlen = 0 - while True: - if lenindef: - if v[:EOC_LEN].tobytes() == EOC: - break - else: - if vlen == l: - break - if vlen > l: - raise DecodeError( - "chunk out of bounds", - klass=self.__class__, - decode_path=decode_path + (str(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, - _ctx_immutable=False, - ) - except TagMismatch: - raise DecodeError( - "expected OctetString encoded chunk", - klass=self.__class__, - decode_path=sub_decode_path, - offset=sub_offset, - ) - chunks.append(chunk) - sub_offset += chunk.tlvlen - vlen += chunk.tlvlen - v = v_tail + v, tail = v[:l], v[l:] try: obj = self.__class__( - value=b"".join(bytes(chunk) for chunk in chunks), + value=v.tobytes(), 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 lenindef else 0)), + _decoded=(offset, llen, l), ctx=ctx, ) except DecodeError as err: @@ -3194,14 +3109,108 @@ class OctetString(Obj): decode_path=decode_path, offset=offset, ) - obj.lenindef = lenindef - obj.ber_encoded = True - return obj, (v[EOC_LEN:] if lenindef else v) - raise TagMismatch( - klass=self.__class__, - decode_path=decode_path, - offset=offset, - ) + return obj, tail + if t != self.tag_constructed: + raise TagMismatch( + klass=self.__class__, + decode_path=decode_path, + offset=offset, + ) + if not ctx.get("bered", False): + raise DecodeError( + "unallowed BER constructed encoding", + klass=self.__class__, + decode_path=decode_path, + offset=offset, + ) + if tag_only: + return None + lenindef = False + try: + l, llen, v = len_decode(lv) + except LenIndefForm: + llen, l, v = 1, 0, lv[1:] + lenindef = True + except DecodeError as err: + raise err.__class__( + msg=err.msg, + klass=self.__class__, + decode_path=decode_path, + offset=offset, + ) + if l > len(v): + raise NotEnoughData( + "encoded length is longer than data", + klass=self.__class__, + decode_path=decode_path, + offset=offset, + ) + chunks = [] + sub_offset = offset + tlen + llen + vlen = 0 + while True: + if lenindef: + if v[:EOC_LEN].tobytes() == EOC: + break + else: + if vlen == l: + break + if vlen > l: + raise DecodeError( + "chunk out of bounds", + klass=self.__class__, + decode_path=decode_path + (str(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, + _ctx_immutable=False, + ) + except TagMismatch: + raise DecodeError( + "expected OctetString encoded chunk", + klass=self.__class__, + decode_path=sub_decode_path, + offset=sub_offset, + ) + chunks.append(chunk) + sub_offset += chunk.tlvlen + vlen += chunk.tlvlen + v = v_tail + 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 lenindef else 0)), + ctx=ctx, + ) + 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.lenindef = lenindef + obj.ber_encoded = True + return obj, (v[EOC_LEN:] if lenindef else v) def __repr__(self): return pp_console_row(next(self.pps())) @@ -3252,7 +3261,7 @@ NullState = namedtuple("NullState", ( "expl_lenindef", "lenindef", "ber_encoded", -)) +), **NAMEDTUPLE_KWARGS) class Null(Obj): @@ -3422,14 +3431,7 @@ ObjectIdentifierState = namedtuple("ObjectIdentifierState", ( "lenindef", "ber_encoded", "defines", -)) - - -def pureint(value): - i = int(value) - if (value[0] in "+- ") or (value[-1] == " "): - raise ValueError("non-pure integer") - return i +), **NAMEDTUPLE_KWARGS) class ObjectIdentifier(Obj): @@ -3498,10 +3500,10 @@ class ObjectIdentifier(Obj): self.defines = defines def __add__(self, their): + if their.__class__ == tuple: + return self.__class__(self._value + their) if isinstance(their, self.__class__): return self.__class__(self._value + their._value) - if isinstance(their, tuple): - return self.__class__(self._value + their) raise InvalidValueType((self.__class__, tuple)) def _value_sanitize(self, value): @@ -3512,7 +3514,7 @@ class ObjectIdentifier(Obj): value = tuple(pureint(arc) for arc in value.split(".")) except ValueError: raise InvalidOID("unacceptable arcs values") - if isinstance(value, tuple): + if value.__class__ == tuple: if len(value) < 2: raise InvalidOID("less than 2 arcs") first_arc = value[0] @@ -3580,7 +3582,7 @@ class ObjectIdentifier(Obj): ) def __eq__(self, their): - if isinstance(their, tuple): + if their.__class__ == tuple: return self._value == their if not issubclass(their.__class__, ObjectIdentifier): return False @@ -3816,7 +3818,7 @@ class Enumerated(Integer): def escape_control_unicode(c): - if unicat(c).startswith("C"): + if unicat(c)[0] == "C": c = repr(c).lstrip("u").strip("'") return c @@ -3890,9 +3892,9 @@ class CommonString(OctetString): value_decoded = None if isinstance(value, self.__class__): value_raw = value._value - elif isinstance(value, text_type): + elif value.__class__ == text_type: value_decoded = value - elif isinstance(value, binary_type): + elif value.__class__ == binary_type: value_raw = value else: raise InvalidValueType((self.__class__, text_type, binary_type)) @@ -3916,9 +3918,9 @@ class CommonString(OctetString): return value_raw def __eq__(self, their): - if isinstance(their, binary_type): + if their.__class__ == binary_type: return self._value == their - if isinstance(their, text_type): + if their.__class__ == text_type: return self._value == their.encode(self.encoding) if not isinstance(their, self.__class__): return False @@ -4009,6 +4011,7 @@ class NumericString(AllowableCharsMixin, CommonString): PrintableStringState = namedtuple( "PrintableStringState", OctetStringState._fields + ("allowable_chars",), + **NAMEDTUPLE_KWARGS ) @@ -4142,11 +4145,6 @@ LEN_YYYYMMDDHHMMSSDMZ = len("YYYYMMDDHHMMSSDMZ") LEN_YYYYMMDDHHMMSSZ = len("YYYYMMDDHHMMSSZ") -def fractions2float(fractions_raw): - pureint(fractions_raw) - return float("0." + fractions_raw) - - class VisibleString(CommonString): __slots__ = () tag_default = tag_encode(26) @@ -4154,7 +4152,21 @@ class VisibleString(CommonString): asn1_type_name = "VisibleString" -UTCTimeState = namedtuple("UTCTimeState", OctetStringState._fields + ("ber_raw",)) +UTCTimeState = namedtuple( + "UTCTimeState", + OctetStringState._fields + ("ber_raw",), + **NAMEDTUPLE_KWARGS +) + + +def str_to_time_fractions(value): + v = pureint(value) + year, v = (v // 10**10), (v % 10**10) + month, v = (v // 10**8), (v % 10**8) + day, v = (v // 10**6), (v % 10**6) + hour, v = (v // 10**4), (v % 10**4) + minute, second = (v // 100), (v % 100) + return year, month, day, hour, minute, second class UTCTime(VisibleString): @@ -4171,6 +4183,9 @@ class UTCTime(VisibleString): >>> UTCTime(datetime(2057, 9, 30, 22, 7, 50)).todatetime() datetime.datetime(1957, 9, 30, 22, 7, 50) + If BER encoded value was met, then ``ber_raw`` attribute will hold + its raw representation. + .. warning:: Pay attention that UTCTime can not hold full year, so all years @@ -4184,7 +4199,7 @@ class UTCTime(VisibleString): * minutes are not exceeding 60 * offset value is not exceeding 14 hours """ - __slots__ = ("_ber_raw",) + __slots__ = ("ber_raw",) tag_default = tag_encode(23) encoding = "ascii" asn1_type_name = "UTCTime" @@ -4212,10 +4227,10 @@ class UTCTime(VisibleString): None, None, impl, expl, None, optional, _decoded, ctx, ) self._value = value - self._ber_raw = None + self.ber_raw = None if value is not None: - self._value, self._ber_raw = self._value_sanitize(value, ctx) - self.ber_encoded = self._ber_raw is not None + self._value, self.ber_raw = self._value_sanitize(value, ctx) + self.ber_encoded = self.ber_raw is not None if default is not None: default, _ = self._value_sanitize(default) self.default = self.__class__( @@ -4229,18 +4244,12 @@ class UTCTime(VisibleString): self.optional = optional def _strptime_bered(self, value): - year = pureint(value[:2]) - year += 2000 if year < 50 else 1900 - decoded = datetime( - year, # %Y - pureint(value[2:4]), # %m - pureint(value[4:6]), # %d - pureint(value[6:8]), # %H - pureint(value[8:10]), # %M - ) + year, month, day, hour, minute, _ = str_to_time_fractions(value[:10] + "00") value = value[10:] if len(value) == 0: raise ValueError("no timezone") + year += 2000 if year < 50 else 1900 + decoded = datetime(year, month, day, hour, minute) offset = 0 if value[-1] == "Z": value = value[:-1] @@ -4253,10 +4262,11 @@ class UTCTime(VisibleString): sign = 1 else: raise ValueError("invalid UTC offset") - offset = 60 * pureint(value[-2:]) + v = pureint(value[-4:]) + offset, v = (60 * (v % 100)), v // 100 if offset >= 3600: raise ValueError("invalid UTC offset minutes") - offset += 3600 * pureint(value[-4:-2]) + offset += 3600 * v if offset > 14 * 3600: raise ValueError("too big UTC offset") offset *= sign @@ -4268,8 +4278,7 @@ class UTCTime(VisibleString): seconds = pureint(value) if seconds >= 60: raise ValueError("invalid seconds value") - decoded += timedelta(seconds=seconds) - return offset, decoded + return offset, decoded + timedelta(seconds=seconds) def _strptime(self, value): # datetime.strptime's format: %y%m%d%H%M%SZ @@ -4277,16 +4286,9 @@ class UTCTime(VisibleString): raise ValueError("invalid UTCTime length") if value[-1] != "Z": raise ValueError("non UTC timezone") - year = pureint(value[:2]) + year, month, day, hour, minute, second = str_to_time_fractions(value[:-1]) year += 2000 if year < 50 else 1900 - return datetime( - year, # %y - pureint(value[2:4]), # %m - pureint(value[4:6]), # %d - pureint(value[6:8]), # %H - pureint(value[8:10]), # %M - pureint(value[10:12]), # %S - ) + return datetime(year, month, day, hour, minute, second) def _dt_sanitize(self, value): if value.year < 1950 or value.year > 2049: @@ -4294,7 +4296,7 @@ class UTCTime(VisibleString): return value.replace(microsecond=0) def _value_sanitize(self, value, ctx=None): - if isinstance(value, binary_type): + if value.__class__ == binary_type: try: value_decoded = value.decode("ascii") except (UnicodeEncodeError, UnicodeDecodeError) as err: @@ -4308,7 +4310,7 @@ class UTCTime(VisibleString): try: offset, _value = self._strptime_bered(value_decoded) _value = _value - timedelta(seconds=offset) - return self._dt_sanitize(_value), value_decoded + return self._dt_sanitize(_value), value except (TypeError, ValueError, OverflowError) as _err: err = _err raise DecodeError( @@ -4317,7 +4319,7 @@ class UTCTime(VisibleString): ) if isinstance(value, self.__class__): return value._value, None - if isinstance(value, datetime): + if value.__class__ == datetime: return self._dt_sanitize(value), None raise InvalidValueType((self.__class__, datetime)) @@ -4325,35 +4327,35 @@ class UTCTime(VisibleString): if self.ready: value = self._value.isoformat() if self.ber_encoded: - value += " (%s)" % self._ber_raw + value += " (%s)" % self.ber_raw return value def __unicode__(self): if self.ready: value = self._value.isoformat() if self.ber_encoded: - value += " (%s)" % self._ber_raw + value += " (%s)" % self.ber_raw return value return text_type(self._pp_value()) def __getstate__(self): return UTCTimeState( *super(UTCTime, self).__getstate__(), - **{"ber_raw": self._ber_raw} + **{"ber_raw": self.ber_raw} ) def __setstate__(self, state): super(UTCTime, self).__setstate__(state) - self._ber_raw = state.ber_raw + self.ber_raw = state.ber_raw def __bytes__(self): self._assert_ready() return self._encode_time() def __eq__(self, their): - if isinstance(their, binary_type): + if their.__class__ == binary_type: return self._encode_time() == their - if isinstance(their, datetime): + if their.__class__ == datetime: return self.todatetime() == their if not isinstance(their, self.__class__): return False @@ -4446,14 +4448,9 @@ class GeneralizedTime(UTCTime): def _strptime_bered(self, value): if len(value) < 4 + 3 * 2: raise ValueError("invalid GeneralizedTime") - decoded = datetime( - pureint(value[:4]), # %Y - pureint(value[4:6]), # %m - pureint(value[6:8]), # %d - pureint(value[8:10]), # %H - ) - value = value[10:] - offset = 0 + year, month, day, hour, _, _ = str_to_time_fractions(value[:10] + "0000") + decoded = datetime(year, month, day, hour) + offset, value = 0, value[10:] if len(value) == 0: return offset, decoded if value[-1] == "Z": @@ -4463,22 +4460,24 @@ class GeneralizedTime(UTCTime): idx = value.rfind(char) if idx == -1: continue - offset_raw = value[idx + 1:].replace(":", "") - if len(offset_raw) not in (2, 4): + offset_raw, value = value[idx + 1:].replace(":", ""), value[:idx] + v = pureint(offset_raw) + if len(offset_raw) == 4: + offset, v = (60 * (v % 100)), v // 100 + if offset >= 3600: + raise ValueError("invalid UTC offset minutes") + elif len(offset_raw) == 2: + pass + else: raise ValueError("invalid UTC offset") - value = value[:idx] - offset = 60 * pureint(offset_raw[2:] or "0") - if offset >= 3600: - raise ValueError("invalid UTC offset minutes") - offset += 3600 * pureint(offset_raw[:2]) + offset += 3600 * v if offset > 14 * 3600: raise ValueError("too big UTC offset") offset *= sign break if len(value) == 0: return offset, decoded - decimal_signs = ".," - if value[0] in decimal_signs: + if value[0] in DECIMAL_SIGNS: return offset, ( decoded + timedelta(seconds=3600 * fractions2float(value[1:])) ) @@ -4488,7 +4487,7 @@ class GeneralizedTime(UTCTime): value = value[2:] if len(value) == 0: return offset, decoded - if value[0] in decimal_signs: + if value[0] in DECIMAL_SIGNS: return offset, ( decoded + timedelta(seconds=60 * fractions2float(value[1:])) ) @@ -4498,7 +4497,7 @@ class GeneralizedTime(UTCTime): value = value[2:] if len(value) == 0: return offset, decoded - if value[0] not in decimal_signs: + if value[0] not in DECIMAL_SIGNS: raise ValueError("invalid format after seconds") return offset, ( decoded + timedelta(microseconds=10**6 * fractions2float(value[1:])) @@ -4510,14 +4509,7 @@ class GeneralizedTime(UTCTime): # datetime.strptime's format: %Y%m%d%H%M%SZ if value[-1] != "Z": raise ValueError("non UTC timezone") - return datetime( - pureint(value[:4]), # %Y - pureint(value[4:6]), # %m - pureint(value[6:8]), # %d - pureint(value[8:10]), # %H - pureint(value[10:12]), # %M - pureint(value[12:14]), # %S - ) + return datetime(*str_to_time_fractions(value[:-1])) if l >= LEN_YYYYMMDDHHMMSSDMZ: # datetime.strptime's format: %Y%m%d%H%M%S.%fZ if value[-1] != "Z": @@ -4531,16 +4523,8 @@ class GeneralizedTime(UTCTime): if us_len > 6: raise ValueError("only microsecond fractions are supported") us = pureint(us + ("0" * (6 - us_len))) - decoded = datetime( - pureint(value[:4]), # %Y - pureint(value[4:6]), # %m - pureint(value[6:8]), # %d - pureint(value[8:10]), # %H - pureint(value[10:12]), # %M - pureint(value[12:14]), # %S - us, # %f - ) - return decoded + year, month, day, hour, minute, second = str_to_time_fractions(value[:14]) + return datetime(year, month, day, hour, minute, second, us) raise ValueError("invalid GeneralizedTime length") def _encode_time(self): @@ -4598,7 +4582,7 @@ ChoiceState = namedtuple("ChoiceState", ( "expl_lenindef", "lenindef", "ber_encoded", -)) +), **NAMEDTUPLE_KWARGS) class Choice(Obj): @@ -4662,7 +4646,7 @@ class Choice(Obj): if len(schema) == 0: raise ValueError("schema must be specified") self.specs = ( - schema if isinstance(schema, OrderedDict) else OrderedDict(schema) + schema if schema.__class__ == OrderedDict else OrderedDict(schema) ) self._value = None if value is not None: @@ -4677,7 +4661,7 @@ class Choice(Obj): self._value = copy(default_obj._value) def _value_sanitize(self, value): - if isinstance(value, tuple) and len(value) == 2: + if (value.__class__ == tuple) and len(value) == 2: choice, obj = value spec = self.specs.get(choice) if spec is None: @@ -4732,7 +4716,7 @@ class Choice(Obj): self.ber_encoded = state.ber_encoded def __eq__(self, their): - if isinstance(their, tuple) and len(their) == 2: + if (their.__class__ == tuple) and len(their) == 2: return self._value == their if not isinstance(their, self.__class__): return False @@ -4916,7 +4900,7 @@ AnyState = namedtuple("AnyState", ( "lenindef", "ber_encoded", "defined", -)) +), **NAMEDTUPLE_KWARGS) class Any(Obj): @@ -4953,7 +4937,7 @@ class Any(Obj): self.defined = None def _value_sanitize(self, value): - if isinstance(value, binary_type): + if value.__class__ == binary_type: return value if isinstance(value, self.__class__): return value._value @@ -5004,7 +4988,7 @@ class Any(Obj): self.defined = state.defined def __eq__(self, their): - if isinstance(their, binary_type): + if their.__class__ == binary_type: return self._value == their if issubclass(their.__class__, Any): return self._value == their._value @@ -5150,7 +5134,7 @@ def get_def_by_path(defines_by_path, sub_decode_path): if len(path) != len(sub_decode_path): continue for p1, p2 in zip(path, sub_decode_path): - if (p1 != any) and (p1 != p2): + if (not p1 is any) and (p1 != p2): break else: return define @@ -5195,7 +5179,7 @@ SequenceState = namedtuple("SequenceState", ( "expl_lenindef", "lenindef", "ber_encoded", -)) +), **NAMEDTUPLE_KWARGS) class Sequence(Obj): @@ -5307,7 +5291,7 @@ class Sequence(Obj): if schema is None: schema = getattr(self, "schema", ()) self.specs = ( - schema if isinstance(schema, OrderedDict) else OrderedDict(schema) + schema if schema.__class__ == OrderedDict else OrderedDict(schema) ) self._value = {} if value is not None: @@ -5435,19 +5419,17 @@ class Sequence(Obj): return spec.default return None - def _encoded_values(self): - raws = [] + def _values_for_encoding(self): for name, spec in iteritems(self.specs): value = self._value.get(name) if value is None: if spec.optional: continue raise ObjNotReady(name) - raws.append(value.encode()) - return raws + yield value def _encode(self): - v = b"".join(self._encoded_values()) + v = b"".join(v.encode() for v in self._values_for_encoding()) return b"".join((self.tag, len_encode(len(v)), v)) def _decode(self, tlv, offset, decode_path, ctx, tag_only): @@ -5682,8 +5664,8 @@ class Set(Sequence): .. _allow_unordered_set_ctx: DER prohibits unordered values encoding and will raise an error - during decode. If If :ref:`bered ` context option is set, - then no error will occure. Also you can disable strict values + during decode. If :ref:`bered ` context option is set, + then no error will occur. Also you can disable strict values ordering check by setting ``"allow_unordered_set": True`` :ref:`context ` option. """ @@ -5692,7 +5674,7 @@ class Set(Sequence): asn1_type_name = "SET" def _encode(self): - raws = self._encoded_values() + raws = [v.encode() for v in self._values_for_encoding()] raws.sort() v = b"".join(raws) return b"".join((self.tag, len_encode(len(v)), v)) @@ -5833,13 +5815,14 @@ class Set(Sequence): tail = v[EOC_LEN:] obj.lenindef = True obj._value = values - if not obj.ready: - raise DecodeError( - "not all values are ready", - klass=self.__class__, - decode_path=decode_path, - offset=offset, - ) + for name, spec in iteritems(self.specs): + if name not in values and not spec.optional: + raise DecodeError( + "%s value is not ready" % name, + klass=self.__class__, + decode_path=decode_path, + offset=offset, + ) obj.ber_encoded = ber_encoded return obj, tail @@ -5860,7 +5843,7 @@ SequenceOfState = namedtuple("SequenceOfState", ( "expl_lenindef", "lenindef", "ber_encoded", -)) +), **NAMEDTUPLE_KWARGS) class SequenceOf(Obj): @@ -6058,11 +6041,11 @@ class SequenceOf(Obj): def __getitem__(self, key): return self._value[key] - def _encoded_values(self): - return [v.encode() for v in self._value] + def _values_for_encoding(self): + return iter(self._value) def _encode(self): - v = b"".join(self._encoded_values()) + v = b"".join(v.encode() for v in self._values_for_encoding()) return b"".join((self.tag, len_encode(len(v)), v)) def _decode(self, tlv, offset, decode_path, ctx, tag_only, ordering_check=False): @@ -6225,7 +6208,7 @@ class SetOf(SequenceOf): asn1_type_name = "SET OF" def _encode(self): - raws = self._encoded_values() + raws = [v.encode() for v in self._values_for_encoding()] raws.sort() v = b"".join(raws) return b"".join((self.tag, len_encode(len(v)), v))