X-Git-Url: http://www.git.cypherpunks.ru/?a=blobdiff_plain;f=pyderasn.py;h=15ca386b89894c7c4278fc2914f5ca9c3b44e199;hb=858cfc57f01b2b31b707fb4f808319ab877003dc;hp=960256f1e38a1e191a384a9fc8eaa40f0cd6ecb7;hpb=fd4251b19318fa6821e3e5c68eaed7e8f875535e;p=pyderasn.git diff --git a/pyderasn.py b/pyderasn.py index 960256f..15ca386 100755 --- a/pyderasn.py +++ b/pyderasn.py @@ -1,6 +1,8 @@ #!/usr/bin/env python # coding: utf-8 # cython: language_level=3 +# pylint: disable=line-too-long,superfluous-parens,protected-access,too-many-lines +# pylint: disable=too-many-return-statements,too-many-branches,too-many-statements # PyDERASN -- Python ASN.1 DER/CER/BER codec with abstract structures # Copyright (C) 2017-2020 Sergey Matveev # @@ -18,7 +20,7 @@ """Python ASN.1 DER/BER codec with abstract structures This library allows you to marshal various structures in ASN.1 DER -format, unmarshal them in BER/CER/DER ones. +format, unmarshal BER/CER/DER ones. >>> i = Integer(123) >>> raw = i.encode() @@ -663,6 +665,11 @@ parse the CRL above with fully assembled ``RevokedCertificate``:: ): print("serial number:", int(obj["userCertificate"])) +.. note:: + + SEQUENCE/SET values with DEFAULT specified are automatically decoded + without evgen mode. + .. _mmap: mmap-ed file @@ -761,6 +768,8 @@ forcefully encoded in DER during CER encoding, by specifying bounds = (1, 32) der_forced = True +.. _agg_octet_string: + agg_octet_string ________________ @@ -827,6 +836,8 @@ copy the payload (without BER/CER encoding interleaved overhead) in it. Virtually it won't take memory more than for keeping small structures and 1 KB binary chunks. +.. _seqof-iterators: + SEQUENCE OF iterators _____________________ @@ -842,6 +853,60 @@ generator taking necessary data from the database and giving the ``RevokedCertificate`` objects. Only binary representation of that objects will take memory during DER encoding. +2-pass DER encoding +------------------- + +There is ability to do 2-pass encoding to DER, writing results directly +to specified writer (buffer, file, whatever). It could be 1.5+ times +slower than ordinary encoding, but it takes little memory for 1st pass +state storing. For example, 1st pass state for CACert.org's CRL with +~416K of certificate entries takes nearly 3.5 MB of memory. +``SignedData`` with several gigabyte ``EncapsulatedContentInfo`` takes +nearly 0.5 KB of memory. + +If you use :ref:`mmap-ed ` memoryviews, :ref:`SEQUENCE OF +iterators ` and write directly to opened file, then +there is very small memory footprint. + +1st pass traverses through all the objects of the structure and returns +the size of DER encoded structure, together with 1st pass state object. +That state contains precalculated lengths for various objects inside the +structure. + +:: + + fulllen, state = obj.encode1st() + +2nd pass takes the writer and 1st pass state. It traverses through all +the objects again, but writes their encoded representation to the writer. + +:: + + opener = io.open if PY2 else open + with opener("result", "wb") as fd: + obj.encode2nd(fd.write, iter(state)) + +.. warning:: + + You **MUST NOT** use 1st pass state if anything is changed in the + objects. It is intended to be used immediately after 1st pass is + done! + +If you use :ref:`SEQUENCE OF iterators `, then you +have to reinitialize the values after the 1st pass. And you **have to** +be sure that the iterator gives exactly the same values as previously. +Yes, you have to run your iterator twice -- because this is two pass +encoding mode. + +If you want to encode to the memory, then you can use convenient +:py:func:`pyderasn.encode2pass` helper. + +.. _browser: + +ASN.1 browser +------------- +.. autofunction:: pyderasn.browse + Base Obj -------- .. autoclass:: pyderasn.Obj @@ -858,7 +923,7 @@ _______ Integer _______ .. autoclass:: pyderasn.Integer - :members: __init__, named + :members: __init__, named, tohex BitString _________ @@ -897,6 +962,14 @@ _______________ .. autoclass:: pyderasn.PrintableString :members: __init__, allow_asterisk, allow_ampersand +IA5String +_________ +.. autoclass:: pyderasn.IA5String + +VisibleString +_____________ +.. autoclass:: pyderasn.VisibleString + UTCTime _______ .. autoclass:: pyderasn.UTCTime @@ -952,11 +1025,14 @@ Various .. autofunction:: pyderasn.abs_decode_path .. autofunction:: pyderasn.agg_octet_string +.. autofunction:: pyderasn.ascii_visualize .. autofunction:: pyderasn.colonize_hex +.. autofunction:: pyderasn.encode2pass .. autofunction:: pyderasn.encode_cer .. autofunction:: pyderasn.file_mmaped .. autofunction:: pyderasn.hexenc .. autofunction:: pyderasn.hexdec +.. autofunction:: pyderasn.hexdump .. autofunction:: pyderasn.tag_encode .. autofunction:: pyderasn.tag_decode .. autofunction:: pyderasn.tag_ctxp @@ -1099,6 +1175,7 @@ from mmap import PROT_READ from operator import attrgetter from string import ascii_letters from string import digits +from sys import maxsize as sys_maxsize from sys import version_info from unicodedata import category as unicat @@ -1124,7 +1201,7 @@ except ImportError: # pragma: no cover def colored(what, *args, **kwargs): return what -__version__ = "7.1" +__version__ = "7.6" __all__ = ( "agg_octet_string", @@ -1134,8 +1211,10 @@ __all__ = ( "Boolean", "BoundsError", "Choice", + "colonize_hex", "DecodeError", "DecodePathDefBy", + "encode2pass", "encode_cer", "Enumerated", "ExceedingData", @@ -1216,11 +1295,13 @@ def file_mmaped(fd): """ return memoryview(mmap(fd.fileno(), 0, prot=PROT_READ)) + 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) @@ -1233,7 +1314,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 (not p1 is any) and (p1 != p2): + if (p1 is not any) and (p1 != p2): break else: return define @@ -1518,9 +1599,33 @@ def len_decode(data): return l, 1 + octets_num, data[1 + octets_num:] +LEN0 = len_encode(0) +LEN1 = len_encode(1) LEN1K = len_encode(1000) +def len_size(l): + """How many bytes length field will take + """ + if l < 128: + return 1 + if l < 256: # 1 << 8 + return 2 + if l < 65536: # 1 << 16 + return 3 + if l < 16777216: # 1 << 24 + return 4 + if l < 4294967296: # 1 << 32 + return 5 + if l < 1099511627776: # 1 << 40 + return 6 + if l < 281474976710656: # 1 << 48 + return 7 + if l < 72057594037927936: # 1 << 56 + return 8 + raise OverflowError("too big length") + + def write_full(writer, data): """Fully write provided data @@ -1539,6 +1644,17 @@ def write_full(writer, data): written += n +# If it is 64-bit system, then use compact 64-bit array of unsigned +# longs. Use an ordinary list with universal integers otherwise, that +# is slower. +if sys_maxsize > 2 ** 32: + def state_2pass_new(): + return array("L") +else: + def state_2pass_new(): + return [] + + ######################################################################## # Base class ######################################################################## @@ -1669,13 +1785,13 @@ class Obj(object): @property def tlen(self): - """See :ref:`decoding` + """.. seealso:: :ref:`decoding` """ return len(self.tag) @property def tlvlen(self): - """See :ref:`decoding` + """.. seealso:: :ref:`decoding` """ return self.tlen + self.llen + self.vlen @@ -1697,9 +1813,18 @@ class Obj(object): def _encode(self): # pragma: no cover raise NotImplementedError() + def _encode_cer(self, writer): + write_full(writer, self._encode()) + def _decode(self, tlv, offset, decode_path, ctx, tag_only, evgen_mode): # pragma: no cover yield NotImplemented + def _encode1st(self, state): + raise NotImplementedError() + + def _encode2nd(self, writer, state_iter): + raise NotImplementedError() + def encode(self): """DER encode the structure @@ -1710,6 +1835,36 @@ class Obj(object): return raw return b"".join((self._expl, len_encode(len(raw)), raw)) + def encode1st(self, state=None): + """Do the 1st pass of 2-pass encoding + + :rtype: (int, array("L")) + :returns: full length of encoded data and precalculated various + objects lengths + """ + if state is None: + state = state_2pass_new() + if self._expl is None: + return self._encode1st(state) + state.append(0) + idx = len(state) - 1 + vlen, _ = self._encode1st(state) + state[idx] = vlen + fulllen = len(self._expl) + len_size(vlen) + vlen + return fulllen, state + + def encode2nd(self, writer, state_iter): + """Do the 2nd pass of 2-pass encoding + + :param writer: must comply with ``io.RawIOBase.write`` behaviour + :param state_iter: iterator over the 1st pass state (``iter(state)``) + """ + if self._expl is None: + self._encode2nd(writer, state_iter) + else: + write_full(writer, self._expl + len_encode(next(state_iter))) + self._encode2nd(writer, state_iter) + def encode_cer(self, writer): """CER encode the structure to specified writer @@ -1727,9 +1882,6 @@ class Obj(object): if self._expl is not None: write_full(writer, EOC) - def _encode_cer(self, writer): - write_full(writer, self._encode()) - def hexencode(self): """Do hexadecimal encoded :py:meth:`pyderasn.Obj.encode` """ @@ -1795,7 +1947,8 @@ class Obj(object): That method is identical to :py:meth:`pyderasn.Obj.decode`, but it returns the generator producing ``(decode_path, obj, tail)`` - values. See :ref:`evgen mode `. + values. + .. seealso:: :ref:`evgen mode `. """ if ctx is None: ctx = {} @@ -1820,7 +1973,7 @@ class Obj(object): yield None return _decode_path, obj, tail = result - if not _decode_path is decode_path: + if _decode_path is not decode_path: yield result else: try: @@ -1862,7 +2015,7 @@ class Obj(object): yield None return _decode_path, obj, tail = result - if not _decode_path is decode_path: + if _decode_path is not decode_path: yield result eoc_expected, tail = tail[:EOC_LEN], tail[EOC_LEN:] if eoc_expected.tobytes() != EOC: @@ -1901,7 +2054,7 @@ class Obj(object): yield None return _decode_path, obj, tail = result - if not _decode_path is decode_path: + if _decode_path is not decode_path: yield result if obj.tlvlen < l and not ctx.get("allow_expl_oob", False): raise DecodeError( @@ -1944,25 +2097,25 @@ class Obj(object): @property def expled(self): - """See :ref:`decoding` + """.. seealso:: :ref:`decoding` """ return self._expl is not None @property def expl_tag(self): - """See :ref:`decoding` + """.. seealso:: :ref:`decoding` """ return self._expl @property def expl_tlen(self): - """See :ref:`decoding` + """.. seealso:: :ref:`decoding` """ return len(self._expl) @property def expl_llen(self): - """See :ref:`decoding` + """.. seealso:: :ref:`decoding` """ if self.expl_lenindef: return 1 @@ -1970,31 +2123,31 @@ class Obj(object): @property def expl_offset(self): - """See :ref:`decoding` + """.. seealso:: :ref:`decoding` """ return self.offset - self.expl_tlen - self.expl_llen @property def expl_vlen(self): - """See :ref:`decoding` + """.. seealso:: :ref:`decoding` """ return self.tlvlen @property def expl_tlvlen(self): - """See :ref:`decoding` + """.. seealso:: :ref:`decoding` """ return self.expl_tlen + self.expl_llen + self.expl_vlen @property def fulloffset(self): - """See :ref:`decoding` + """.. seealso:: :ref:`decoding` """ return self.expl_offset if self.expled else self.offset @property def fulllen(self): - """See :ref:`decoding` + """.. seealso:: :ref:`decoding` """ return self.expl_tlvlen if self.expled else self.tlvlen @@ -2041,6 +2194,17 @@ def encode_cer(obj): return buf.getvalue() +def encode2pass(obj): + """Encode (2-pass mode) to DER in memory buffer + + :returns bytes: memory buffer contents + """ + buf = BytesIO() + _, state = obj.encode1st() + obj.encode2nd(buf.write, iter(state)) + return buf.getvalue() + + class DecodePathDefBy(object): """DEFINED BY representation inside decode path """ @@ -2154,6 +2318,15 @@ def colonize_hex(hexed): return ":".join(hexed[i:i + 2] for i in six_xrange(0, len(hexed), 2)) +def find_oid_name(asn1_type_name, oid_maps, value): + if len(oid_maps) > 0 and asn1_type_name == ObjectIdentifier.asn1_type_name: + for oid_map in oid_maps: + oid_name = oid_map.get(value) + if oid_name is not None: + return oid_name + return None + + def pp_console_row( pp, oid_maps=(), @@ -2177,9 +2350,7 @@ def pp_console_row( col += _colourize("B", "red", with_colours) if pp.bered else " " cols.append(col) col = "[%d,%d,%4d]%s" % ( - pp.tlen, - pp.llen, - pp.vlen, + pp.tlen, pp.llen, pp.vlen, LENINDEF_PP_CHAR if pp.lenindef else " " ) col = _colourize(col, "green", with_colours, ()) @@ -2191,19 +2362,11 @@ def pp_console_row( if isinstance(ent, DecodePathDefBy): cols.append(_colourize("DEFINED BY", "red", with_colours, ("reverse",))) value = str(ent.defined_by) - oid_name = None - if ( - len(oid_maps) > 0 and - ent.defined_by.asn1_type_name == - ObjectIdentifier.asn1_type_name - ): - for oid_map in oid_maps: - oid_name = oid_map.get(value) - if oid_name is not None: - cols.append(_colourize("%s:" % oid_name, "green", with_colours)) - break + oid_name = find_oid_name(ent.defined_by.asn1_type_name, oid_maps, value) if oid_name is None: cols.append(_colourize("%s:" % value, "white", with_colours, ("reverse",))) + else: + cols.append(_colourize("%s:" % oid_name, "green", with_colours)) else: cols.append(_colourize("%s:" % ent, "yellow", with_colours, ("reverse",))) if pp.expl is not None: @@ -2222,23 +2385,12 @@ def pp_console_row( if pp.value is not None: value = pp.value cols.append(_colourize(value, "white", with_colours, ("reverse",))) - if ( - len(oid_maps) > 0 and - pp.asn1_type_name == ObjectIdentifier.asn1_type_name - ): - for oid_map in oid_maps: - oid_name = oid_map.get(value) - if oid_name is not None: - cols.append(_colourize("(%s)" % oid_name, "green", with_colours)) - break + oid_name = find_oid_name(pp.asn1_type_name, oid_maps, pp.value) + if oid_name is not None: + cols.append(_colourize("(%s)" % oid_name, "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, + "(%s)" % colonize_hex(pp.obj.tohex()), "green", with_colours, )) if with_blob: if pp.blob.__class__ == binary_type: @@ -2284,7 +2436,7 @@ def pprint( """Pretty print object :param Obj obj: object you want to pretty print - :param oid_maps: list of ``str(OID) <-> human readable string`` dictionary. + :param oid_maps: list of ``str(OID) <-> human readable string`` dictionaries. Its human readable form is printed when OID is met :param big_blobs: if large binary objects are met (like OctetString values), do we need to print them too, on separate @@ -2458,11 +2610,14 @@ class Boolean(Obj): def _encode(self): self._assert_ready() - return b"".join(( - self.tag, - len_encode(1), - (b"\xFF" if self._value else b"\x00"), - )) + return b"".join((self.tag, LEN1, (b"\xFF" if self._value else b"\x00"))) + + def _encode1st(self, state): + return len(self.tag) + 2, state + + def _encode2nd(self, writer, state_iter): + self._assert_ready() + write_full(writer, self._encode()) def _decode(self, tlv, offset, decode_path, ctx, tag_only, evgen_mode): try: @@ -2702,13 +2857,23 @@ class Integer(Obj): self._assert_ready() return int(self._value) + def tohex(self): + """Hexadecimal representation + + Use :py:func:`pyderasn.colonize_hex` for colonizing it. + """ + hex_repr = hex(int(self))[2:].upper() + if len(hex_repr) % 2 != 0: + hex_repr = "0" + hex_repr + return hex_repr + def __hash__(self): self._assert_ready() - return hash( - self.tag + - bytes(self._expl or b"") + + return hash(b"".join(( + self.tag, + bytes(self._expl or b""), str(self._value).encode("ascii"), - ) + ))) def __eq__(self, their): if isinstance(their, integer_types): @@ -2755,7 +2920,7 @@ class Integer(Obj): _specs=self.specs, ) - def _encode(self): + def _encode_payload(self): self._assert_ready() value = self._value if PY2: @@ -2792,8 +2957,19 @@ class Integer(Obj): bytes_len += 1 else: break + return octets + + def _encode(self): + octets = self._encode_payload() return b"".join((self.tag, len_encode(len(octets)), octets)) + def _encode1st(self, state): + l = len(self._encode_payload()) + return len(self.tag) + len_size(l) + l, state + + def _encode2nd(self, writer, state_iter): + write_full(writer, self._encode()) + def _decode(self, tlv, offset, decode_path, ctx, tag_only, evgen_mode): try: t, _, lv = tag_strip(tlv) @@ -3178,6 +3354,21 @@ class BitString(Obj): octets, )) + def _encode1st(self, state): + self._assert_ready() + _, octets = self._value + l = len(octets) + 1 + return len(self.tag) + len_size(l) + l, state + + def _encode2nd(self, writer, state_iter): + bit_len, octets = self._value + write_full(writer, b"".join(( + self.tag, + len_encode(len(octets) + 1), + int2byte((8 - bit_len % 8) % 8), + ))) + write_full(writer, octets) + def _encode_cer(self, writer): bit_len, octets = self._value if len(octets) + 1 <= 1000: @@ -3192,7 +3383,7 @@ class BitString(Obj): int2byte(0), octets[offset:offset + 999], ))) - tail = octets[offset+999:] + tail = octets[offset + 999:] if len(tail) > 0: tail = int2byte((8 - bit_len % 8) % 8) + tail write_full(writer, b"".join(( @@ -3629,6 +3820,16 @@ class OctetString(Obj): self._value, )) + def _encode1st(self, state): + self._assert_ready() + l = len(self._value) + return len(self.tag) + len_size(l) + l, state + + def _encode2nd(self, writer, state_iter): + value = self._value + write_full(writer, self.tag + len_encode(len(value))) + write_full(writer, value) + def _encode_cer(self, writer): octets = self._value if len(octets) <= 1000: @@ -3642,7 +3843,7 @@ class OctetString(Obj): LEN1K, octets[offset:offset + 1000], ))) - tail = octets[offset+1000:] + tail = octets[offset + 1000:] if len(tail) > 0: write_full(writer, b"".join(( OctetString.tag_default, @@ -3897,6 +4098,8 @@ def agg_octet_string(evgens, decode_path, raw, writer): :param writer: buffer.write where string is going to be saved :param writer: where string is going to be saved. Must comply with ``io.RawIOBase.write`` behaviour + + .. seealso:: :ref:`agg_octet_string` """ decode_path_len = len(decode_path) for dp, obj, _ in evgens: @@ -3985,7 +4188,13 @@ class Null(Obj): ) def _encode(self): - return self.tag + len_encode(0) + return self.tag + LEN0 + + def _encode1st(self, state): + return len(self.tag) + 1, state + + def _encode2nd(self, writer, state_iter): + write_full(writer, self.tag + LEN0) def _decode(self, tlv, offset, decode_path, ctx, tag_only, evgen_mode): try: @@ -4108,7 +4317,9 @@ class ObjectIdentifier(Obj): tuple element is ``{OID: pyderasn.Obj()}`` dictionary, mapping between current OID value and structure applied to defined field. - :ref:`Read about DEFINED BY ` + + .. seealso:: :ref:`definedby` + :param bytes impl: override default tag with ``IMPLICIT`` one :param bytes expl: override default tag with ``EXPLICIT`` one :param default: set default value. Type same as in ``value`` @@ -4201,11 +4412,11 @@ class ObjectIdentifier(Obj): def __hash__(self): self._assert_ready() - return hash( - self.tag + - bytes(self._expl or b"") + + return hash(b"".join(( + self.tag, + bytes(self._expl or b""), str(self._value).encode("ascii"), - ) + ))) def __eq__(self, their): if their.__class__ == tuple: @@ -4239,7 +4450,7 @@ class ObjectIdentifier(Obj): optional=self.optional if optional is None else optional, ) - def _encode(self): + def _encode_octets(self): self._assert_ready() value = self._value first_value = value[1] @@ -4255,9 +4466,19 @@ class ObjectIdentifier(Obj): octets = [zero_ended_encode(first_value)] for arc in value[2:]: octets.append(zero_ended_encode(arc)) - v = b"".join(octets) + return b"".join(octets) + + def _encode(self): + v = self._encode_octets() return b"".join((self.tag, len_encode(len(v)), v)) + def _encode1st(self, state): + l = len(self._encode_octets()) + return len(self.tag) + len_size(l) + l, state + + def _encode2nd(self, writer, state_iter): + write_full(writer, self._encode()) + def _decode(self, tlv, offset, decode_path, ctx, tag_only, evgen_mode): try: t, _, lv = tag_strip(tlv) @@ -4497,27 +4718,25 @@ class CommonString(OctetString): :header-rows: 1 * - Class - - Text Encoding + - Text Encoding, validation * - :py:class:`pyderasn.UTF8String` - utf-8 * - :py:class:`pyderasn.NumericString` - - ascii + - proper alphabet validation * - :py:class:`pyderasn.PrintableString` - - ascii + - proper alphabet validation * - :py:class:`pyderasn.TeletexString` - - ascii + - iso-8859-1 * - :py:class:`pyderasn.T61String` - - ascii + - iso-8859-1 * - :py:class:`pyderasn.VideotexString` - iso-8859-1 * - :py:class:`pyderasn.IA5String` - - ascii + - proper alphabet validation * - :py:class:`pyderasn.GraphicString` - iso-8859-1 - * - :py:class:`pyderasn.VisibleString` - - ascii - * - :py:class:`pyderasn.ISO646String` - - ascii + * - :py:class:`pyderasn.VisibleString`, :py:class:`pyderasn.ISO646String` + - proper alphabet validation * - :py:class:`pyderasn.GeneralString` - iso-8859-1 * - :py:class:`pyderasn.UniversalString` @@ -4625,6 +4844,12 @@ class AllowableCharsMixin(object): return self._allowable_chars return frozenset(six_unichr(c) for c in self._allowable_chars) + def _value_sanitize(self, value): + value = super(AllowableCharsMixin, self)._value_sanitize(value) + if not frozenset(value) <= self._allowable_chars: + raise DecodeError("non satisfying alphabet value") + return value + class NumericString(AllowableCharsMixin, CommonString): """Numeric string @@ -4641,12 +4866,6 @@ class NumericString(AllowableCharsMixin, CommonString): asn1_type_name = "NumericString" _allowable_chars = frozenset(digits.encode("ascii") + b" ") - def _value_sanitize(self, value): - value = super(NumericString, self)._value_sanitize(value) - if not frozenset(value) <= self._allowable_chars: - raise DecodeError("non-numeric value") - return value - PrintableStringState = namedtuple( "PrintableStringState", @@ -4714,12 +4933,6 @@ class PrintableString(AllowableCharsMixin, CommonString): """ return self._ampersand <= self._allowable_chars - def _value_sanitize(self, value): - value = super(PrintableString, self)._value_sanitize(value) - if not frozenset(value) <= self._allowable_chars: - raise DecodeError("non-printable value") - return value - def __getstate__(self): return PrintableStringState( *super(PrintableString, self).__getstate__(), @@ -4757,7 +4970,7 @@ class PrintableString(AllowableCharsMixin, CommonString): class TeletexString(CommonString): __slots__ = () tag_default = tag_encode(20) - encoding = "ascii" + encoding = "iso-8859-1" asn1_type_name = "TeletexString" @@ -4773,23 +4986,58 @@ class VideotexString(CommonString): asn1_type_name = "VideotexString" -class IA5String(CommonString): +class IA5String(AllowableCharsMixin, CommonString): + """IA5 string + + Its value is properly sanitized: it is a mix of + + * http://www.itscj.ipsj.or.jp/iso-ir/006.pdf (G) + * http://www.itscj.ipsj.or.jp/iso-ir/001.pdf (C0) + * DEL character (0x7F) + + It is just 7-bit ASCII. + + >>> IA5String().allowable_chars + frozenset(["NUL", ... "DEL"]) + """ __slots__ = () tag_default = tag_encode(22) encoding = "ascii" asn1_type_name = "IA5" + _allowable_chars = frozenset(b"".join( + six_unichr(c).encode("ascii") for c in six_xrange(128) + )) LEN_YYMMDDHHMMSSZ = len("YYMMDDHHMMSSZ") +LEN_LEN_YYMMDDHHMMSSZ = len_encode(LEN_YYMMDDHHMMSSZ) +LEN_YYMMDDHHMMSSZ_WITH_LEN = len(LEN_LEN_YYMMDDHHMMSSZ) + LEN_YYMMDDHHMMSSZ LEN_YYYYMMDDHHMMSSDMZ = len("YYYYMMDDHHMMSSDMZ") LEN_YYYYMMDDHHMMSSZ = len("YYYYMMDDHHMMSSZ") +LEN_LEN_YYYYMMDDHHMMSSZ = len_encode(LEN_YYYYMMDDHHMMSSZ) -class VisibleString(CommonString): +class VisibleString(AllowableCharsMixin, CommonString): + """Visible string + + Its value is properly sanitized. ASCII subset from space to tilde is + allowed: http://www.itscj.ipsj.or.jp/iso-ir/006.pdf + + >>> VisibleString().allowable_chars + frozenset([" ", ... "~"]) + """ __slots__ = () tag_default = tag_encode(26) encoding = "ascii" asn1_type_name = "VisibleString" + _allowable_chars = frozenset(b"".join( + six_unichr(c).encode("ascii") for c in six_xrange(ord(" "), ord("~") + 1) + )) + + +class ISO646String(VisibleString): + __slots__ = () + asn1_type_name = "ISO646String" UTCTimeState = namedtuple( @@ -4970,6 +5218,7 @@ class UTCTime(VisibleString): if self.ber_encoded: value += " (%s)" % self.ber_raw return value + return None def __unicode__(self): if self.ready: @@ -5011,8 +5260,14 @@ class UTCTime(VisibleString): def _encode(self): self._assert_ready() - value = self._encode_time() - return b"".join((self.tag, len_encode(len(value)), value)) + return b"".join((self.tag, LEN_LEN_YYMMDDHHMMSSZ, self._encode_time())) + + def _encode1st(self, state): + return len(self.tag) + LEN_YYMMDDHHMMSSZ_WITH_LEN, state + + def _encode2nd(self, writer, state_iter): + self._assert_ready() + write_full(writer, self._encode()) def _encode_cer(self, writer): write_full(writer, self._encode()) @@ -5178,6 +5433,22 @@ class GeneralizedTime(UTCTime): encoded += (".%06d" % value.microsecond).rstrip("0") return (encoded + "Z").encode("ascii") + def _encode(self): + self._assert_ready() + value = self._value + if value.microsecond > 0: + encoded = self._encode_time() + return b"".join((self.tag, len_encode(len(encoded)), encoded)) + return b"".join((self.tag, LEN_LEN_YYYYMMDDHHMMSSZ, self._encode_time())) + + def _encode1st(self, state): + self._assert_ready() + vlen = len(self._encode_time()) + return len(self.tag) + len_size(vlen) + vlen, state + + def _encode2nd(self, writer, state_iter): + write_full(writer, self._encode()) + class GraphicString(CommonString): __slots__ = () @@ -5186,11 +5457,6 @@ class GraphicString(CommonString): asn1_type_name = "GraphicString" -class ISO646String(VisibleString): - __slots__ = () - asn1_type_name = "ISO646String" - - class GeneralString(CommonString): __slots__ = () tag_default = tag_encode(27) @@ -5422,6 +5688,13 @@ class Choice(Obj): self._assert_ready() return self._value[1].encode() + def _encode1st(self, state): + self._assert_ready() + return self._value[1].encode1st(state) + + def _encode2nd(self, writer, state_iter): + self._value[1].encode2nd(writer, state_iter) + def _encode_cer(self, writer): self._assert_ready() self._value[1].encode_cer(writer) @@ -5601,7 +5874,7 @@ class Any(Obj): def _value_sanitize(self, value): if value.__class__ == binary_type: if len(value) == 0: - raise ValueError("Any value can not be empty") + raise ValueError("%s value can not be empty" % self.__class__.__name__) return value if isinstance(value, self.__class__): return value._value @@ -5690,6 +5963,20 @@ class Any(Obj): return value return value.encode() + def _encode1st(self, state): + self._assert_ready() + value = self._value + if value.__class__ == binary_type: + return len(value), state + return value.encode1st(state) + + def _encode2nd(self, writer, state_iter): + value = self._value + if value.__class__ == binary_type: + write_full(writer, value) + else: + value.encode2nd(writer, state_iter) + def _encode_cer(self, writer): self._assert_ready() value = self._value @@ -5848,7 +6135,19 @@ SequenceState = namedtuple( ) -class Sequence(Obj): +class SequenceEncode1stMixing(object): + def _encode1st(self, state): + state.append(0) + idx = len(state) - 1 + vlen = 0 + for v in self._values_for_encoding(): + l, _ = v.encode1st(state) + vlen += l + state[idx] = vlen + return len(self.tag) + len_size(vlen) + vlen, state + + +class Sequence(SequenceEncode1stMixing, Obj): """``SEQUENCE`` structure type You have to make specification of sequence:: @@ -5936,11 +6235,10 @@ class Sequence(Obj): defaulted values existence validation by setting ``"allow_default_values": True`` :ref:`context ` option. - .. warning:: - - Check for default value existence is not performed in - ``evgen_mode``, because previously decoded values are not stored - in memory, to be able to compare them. + All values with DEFAULT specified are decoded atomically in + :ref:`evgen mode `. If DEFAULT value is some kind of + SEQUENCE, then it will be yielded as a single element, not + disassembled. That is required for DEFAULT existence check. Two sequences are equal if they have equal specification (schema), implicit/explicit tagging and the same values. @@ -6095,6 +6393,11 @@ class Sequence(Obj): v = b"".join(v.encode() for v in self._values_for_encoding()) return b"".join((self.tag, len_encode(len(v)), v)) + def _encode2nd(self, writer, state_iter): + write_full(writer, self.tag + len_encode(next(state_iter))) + for v in self._values_for_encoding(): + v.encode2nd(writer, state_iter) + def _encode_cer(self, writer): write_full(writer, self.tag + LENINDEF) for v in self._values_for_encoding(): @@ -6161,9 +6464,10 @@ class Sequence(Obj): len(v) == 0 ): continue + spec_defaulted = spec.default is not None sub_decode_path = decode_path + (name,) try: - if evgen_mode: + if evgen_mode and not spec_defaulted: for _decode_path, value, v_tail in spec.decode_evgen( v, sub_offset, @@ -6241,9 +6545,10 @@ class Sequence(Obj): vlen += value_len sub_offset += value_len v = v_tail - if not evgen_mode: - if spec.default is not None and value == spec.default: - # This will not work in evgen_mode + if spec_defaulted: + if evgen_mode: + yield sub_decode_path, value, v_tail + if value == spec.default: if ctx_bered or ctx_allow_default_values: ber_encoded = True else: @@ -6253,6 +6558,7 @@ class Sequence(Obj): decode_path=sub_decode_path, offset=sub_offset, ) + if not evgen_mode: values[name] = value spec_defines = getattr(spec, "defines", ()) if len(spec_defines) == 0: @@ -6339,7 +6645,7 @@ class Sequence(Obj): yield pp -class Set(Sequence): +class Set(Sequence, SequenceEncode1stMixing): """``SET`` structure type Its usage is identical to :py:class:`pyderasn.Sequence`. @@ -6356,17 +6662,16 @@ class Set(Sequence): tag_default = tag_encode(form=TagFormConstructed, num=17) asn1_type_name = "SET" - def _encode(self): - v = b"".join(value.encode() for value in sorted( - self._values_for_encoding(), + def _values_for_encoding(self): + return sorted( + super(Set, self)._values_for_encoding(), key=attrgetter("tag_order"), - )) - return b"".join((self.tag, len_encode(len(v)), v)) + ) def _encode_cer(self, writer): write_full(writer, self.tag + LENINDEF) for v in sorted( - self._values_for_encoding(), + super(Set, self)._values_for_encoding(), key=attrgetter("tag_order_cer"), ): v.encode_cer(writer) @@ -6453,7 +6758,8 @@ class Set(Sequence): decode_path=decode_path, offset=offset, ) - if evgen_mode: + spec_defaulted = spec.default is not None + if evgen_mode and not spec_defaulted: for _decode_path, value, v_tail in spec.decode_evgen( v, sub_offset, @@ -6485,17 +6791,20 @@ class Set(Sequence): decode_path=sub_decode_path, offset=sub_offset, ) - if spec.default is None or value != spec.default: - pass - elif ctx_bered or ctx_allow_default_values: - ber_encoded = True - else: - raise DecodeError( - "DEFAULT value met", - klass=self.__class__, - decode_path=sub_decode_path, - offset=sub_offset, - ) + if spec_defaulted: + if evgen_mode: + yield sub_decode_path, value, v_tail + if value != spec.default: + pass + elif ctx_bered or ctx_allow_default_values: + ber_encoded = True + else: + raise DecodeError( + "DEFAULT value met", + klass=self.__class__, + decode_path=sub_decode_path, + offset=sub_offset, + ) values[name] = value del _specs_items[name] tag_order_prev = value_tag_order @@ -6542,7 +6851,7 @@ SequenceOfState = namedtuple( ) -class SequenceOf(Obj): +class SequenceOf(SequenceEncode1stMixing, Obj): """``SEQUENCE OF`` sequence type For that kind of type you must specify the object it will carry on @@ -6631,7 +6940,6 @@ class SequenceOf(Obj): value = value._value elif hasattr(value, NEXT_ATTR_NAME): iterator = True - value = value elif hasattr(value, "__iter__"): value = list(value) else: @@ -6772,6 +7080,31 @@ class SequenceOf(Obj): value = b"".join(v.encode() for v in self._values_for_encoding()) return b"".join((self.tag, len_encode(len(value)), value)) + def _encode1st(self, state): + state = super(SequenceOf, self)._encode1st(state) + if hasattr(self._value, NEXT_ATTR_NAME): + self._value = [] + return state + + def _encode2nd(self, writer, state_iter): + write_full(writer, self.tag + len_encode(next(state_iter))) + iterator = hasattr(self._value, NEXT_ATTR_NAME) + if iterator: + values_count = 0 + class_expected = self.spec.__class__ + values_for_encoding = self._values_for_encoding() + self._value = [] + for v in values_for_encoding: + if not isinstance(v, class_expected): + raise InvalidValueType((class_expected,)) + v.encode2nd(writer, state_iter) + values_count += 1 + if not self._bound_min <= values_count <= self._bound_max: + raise BoundsError(self._bound_min, values_count, self._bound_max) + else: + for v in self._values_for_encoding(): + v.encode2nd(writer, state_iter) + def _encode_cer(self, writer): write_full(writer, self.tag + LENINDEF) iterator = hasattr(self._value, NEXT_ATTR_NAME) @@ -6995,6 +7328,17 @@ class SetOf(SequenceOf): v = b"".join(sorted(v.encode() for v in self._values_for_encoding())) return b"".join((self.tag, len_encode(len(v)), v)) + def _encode2nd(self, writer, state_iter): + write_full(writer, self.tag + len_encode(next(state_iter))) + values = [] + for v in self._values_for_encoding(): + buf = BytesIO() + v.encode2nd(buf.write, state_iter) + values.append(buf.getvalue()) + values.sort() + for v in values: + write_full(writer, v) + def _encode_cer(self, writer): write_full(writer, self.tag + LENINDEF) for v in sorted(encode_cer(v) for v in self._values_for_encoding()): @@ -7054,6 +7398,7 @@ def generic_decoder(): # pragma: no cover with_colours=False, with_decode_path=False, decode_path_only=(), + decode_path=(), ): def _pprint_pps(pps): for pp in pps: @@ -7085,10 +7430,404 @@ def generic_decoder(): # pragma: no cover else: for row in _pprint_pps(pp): yield row - return "\n".join(_pprint_pps(obj.pps())) + return "\n".join(_pprint_pps(obj.pps(decode_path))) return SEQUENCEOF(), pprint_any +def ascii_visualize(ba): + """Output only ASCII printable characters, like in hexdump -C + + Example output for given binary string (right part):: + + 92 2b 39 20 65 91 e6 8e 95 93 1a 58 df 02 78 ea |.+9 e......X..x.| + ^^^^^^^^^^^^^^^^ + """ + return "".join((six_unichr(b) if 0x20 <= b <= 0x7E else ".") for b in ba) + + +def hexdump(raw): + """Generate ``hexdump -C`` like output + + Rendered example:: + + 00000000 30 80 30 80 a0 80 02 01 02 00 00 02 14 54 a5 18 |0.0..........T..| + 00000010 69 ef 8b 3f 15 fd ea ad bd 47 e0 94 81 6b 06 6a |i..?.....G...k.j| + + Result of that function is a generator of lines, where each line is + a list of columns:: + + [ + [...], + ["00000010 ", " 69", " ef", " 8b", " 3f", " 15", " fd", " ea", " ad ", + " bd", " 47", " e0", " 94", " 81", " 6b", " 06", " 6a ", + " |i..?.....G...k.j|"] + [...], + ] + """ + hexed = hexenc(raw).upper() + addr, cols = 0, ["%08x " % 0] + for i in six_xrange(0, len(hexed), 2): + if i != 0 and i // 2 % 8 == 0: + cols[-1] += " " + if i != 0 and i // 2 % 16 == 0: + cols.append(" |%s|" % ascii_visualize(bytearray(raw[addr:addr + 16]))) + yield cols + addr += 16 + cols = ["%08x " % addr] + cols.append(" " + hexed[i:i + 2]) + if len(cols) > 0: + cols.append(" |%s|" % ascii_visualize(bytearray(raw[addr:]))) + yield cols + + +def browse(raw, obj, oid_maps=()): + """Interactive browser + + :param bytes raw: binary data you decoded + :param obj: decoded :py:class:`pyderasn.Obj` + :param oid_maps: list of ``str(OID) <-> human readable string`` dictionaries. + Its human readable form is printed when OID is met + + .. note:: `urwid `__ dependency required + + This browser is an interactive terminal application for browsing + structures of your decoded ASN.1 objects. You can quit it with **q** + key. It consists of three windows: + + :tree: + View of ASN.1 elements hierarchy. You can navigate it using **Up**, + **Down**, **PageUp**, **PageDown**, **Home**, **End** keys. + **Left** key goes to constructed element above. **Plus**/**Minus** + keys collapse/uncollapse constructed elements. **Space** toggles it + :info: + window with various information about element. You can scroll it + with **h**/**l** (down, up) (**H**/**L** for triple speed) keys + :hexdump: + window with raw data hexdump and highlighted current element's + contents. It automatically focuses on element's data. You can + scroll it with **j**/**k** (down, up) (**J**/**K** for triple + speed) keys. If element has explicit tag, then it also will be + highlighted with different colour + + Window's header contains current decode path and progress bars with + position in *info* and *hexdump* windows. + + If you press **d**, then current element will be saved in the + current directory under its decode path name (adding ".0", ".1", etc + suffix if such file already exists). **D** will save it with explicit tag. + + You can also invoke it with ``--browse`` command line argument. + """ + from copy import deepcopy + from os.path import exists as path_exists + import urwid + + class TW(urwid.TreeWidget): + def __init__(self, state, *args, **kwargs): + self.state = state + self.scrolled = {"info": False, "hexdump": False} + super(TW, self).__init__(*args, **kwargs) + + def _get_pp(self): + pp = self.get_node().get_value() + constructed = len(pp) > 1 + return (pp if hasattr(pp, "_fields") else pp[0]), constructed + + def _state_update(self): + pp, _ = self._get_pp() + self.state["decode_path"].set_text( + ":".join(str(p) for p in pp.decode_path) + ) + lines = deepcopy(self.state["hexed"]) + + def attr_set(i, attr): + line = lines[i // 16] + idx = 1 + (i - 16 * (i // 16)) + line[idx] = (attr, line[idx]) + + if pp.expl_offset is not None: + for i in six_xrange( + pp.expl_offset, + pp.expl_offset + pp.expl_tlen + pp.expl_llen, + ): + attr_set(i, "select-expl") + for i in six_xrange(pp.offset, pp.offset + pp.tlen + pp.llen + pp.vlen): + attr_set(i, "select-value") + self.state["hexdump"]._set_body([urwid.Text(line) for line in lines]) + self.state["hexdump"].set_focus(pp.offset // 16) + self.state["hexdump"].set_focus_valign("middle") + self.state["hexdump_bar"].set_completion( + (100 * pp.offset // 16) // + len(self.state["hexdump"]._body.positions()) + ) + + lines = [ + [("header", "Name: "), pp.obj_name], + [("header", "Type: "), pp.asn1_type_name], + [("header", "Offset: "), "%d (0x%x)" % (pp.offset, pp.offset)], + [("header", "[TLV]len: "), "%d/%d/%d" % ( + pp.tlen, pp.llen, pp.vlen, + )], + [("header", "TLVlen: "), "%d" % sum(( + pp.tlen, pp.llen, pp.vlen, + ))], + [("header", "Slice: "), "[%d:%d]" % ( + pp.offset, pp.offset + pp.tlen + pp.llen + pp.vlen, + )], + ] + if pp.lenindef: + lines.append([("warning", "LENINDEF")]) + if pp.ber_encoded: + lines.append([("warning", "BER encoded")]) + if pp.bered: + lines.append([("warning", "BERed")]) + if pp.expl is not None: + lines.append([("header", "EXPLICIT")]) + klass, _, num = pp.expl + lines.append([" Tag: %s%d" % (TagClassReprs[klass], num)]) + if pp.expl_offset is not None: + lines.append([" Offset: %d" % pp.expl_offset]) + lines.append([" [TLV]len: %d/%d/%d" % ( + pp.expl_tlen, pp.expl_llen, pp.expl_vlen, + )]) + lines.append([" TLVlen: %d" % sum(( + pp.expl_tlen, pp.expl_llen, pp.expl_vlen, + ))]) + lines.append([" Slice: [%d:%d]" % ( + pp.expl_offset, + pp.expl_offset + pp.expl_tlen + pp.expl_llen + pp.expl_vlen, + )]) + if pp.impl is not None: + klass, _, num = pp.impl + lines.append([ + ("header", "IMPLICIT: "), "%s%d" % (TagClassReprs[klass], num), + ]) + if pp.optional: + lines.append(["OPTIONAL"]) + if pp.default: + lines.append(["DEFAULT"]) + if len(pp.decode_path) > 0: + ent = pp.decode_path[-1] + if isinstance(ent, DecodePathDefBy): + lines.append([""]) + value = str(ent.defined_by) + oid_name = find_oid_name( + ent.defined_by.asn1_type_name, oid_maps, value, + ) + lines.append([("header", "DEFINED BY: "), "%s" % ( + value if oid_name is None + else "%s (%s)" % (oid_name, value) + )]) + lines.append([""]) + if pp.value is not None: + lines.append([("header", "Value: "), pp.value]) + if ( + len(oid_maps) > 0 and + pp.asn1_type_name == ObjectIdentifier.asn1_type_name + ): + for oid_map in oid_maps: + oid_name = oid_map.get(pp.value) + if oid_name is not None: + lines.append([("header", "Human: "), oid_name]) + break + if pp.asn1_type_name == Integer.asn1_type_name: + lines.append([ + ("header", "Decimal: "), "%d" % int(pp.obj), + ]) + lines.append([ + ("header", "Hexadecimal: "), colonize_hex(pp.obj.tohex()), + ]) + if pp.blob.__class__ == binary_type: + blob = hexenc(pp.blob).upper() + for i in six_xrange(0, len(blob), 32): + lines.append([colonize_hex(blob[i:i + 32])]) + elif pp.blob.__class__ == tuple: + lines.append([", ".join(pp.blob)]) + self.state["info"]._set_body([urwid.Text(line) for line in lines]) + self.state["info_bar"].set_completion(0) + + def selectable(self): + if self.state["widget_current"] != self: + self.state["widget_current"] = self + self.scrolled["info"] = False + self.scrolled["hexdump"] = False + self._state_update() + return super(TW, self).selectable() + + def get_display_text(self): + pp, constructed = self._get_pp() + style = "constructed" if constructed else "" + if len(pp.decode_path) == 0: + return (style, pp.obj_name) + if pp.asn1_type_name == "EOC": + return ("eoc", "EOC") + ent = pp.decode_path[-1] + if isinstance(ent, DecodePathDefBy): + value = str(ent.defined_by) + oid_name = find_oid_name( + ent.defined_by.asn1_type_name, oid_maps, value, + ) + return ("defby", "DEFBY:" + ( + value if oid_name is None else oid_name + )) + return (style, ent) + + def _scroll(self, what, step): + self.state[what]._invalidate() + pos = self.state[what].focus_position + if not self.scrolled[what]: + self.scrolled[what] = True + pos -= 2 + pos = max(0, pos + step) + pos = min(pos, len(self.state[what]._body.positions()) - 1) + self.state[what].set_focus(pos) + self.state[what].set_focus_valign("top") + self.state[what + "_bar"].set_completion( + (100 * pos) // len(self.state[what]._body.positions()) + ) + + def keypress(self, size, key): + if key == "q": + raise urwid.ExitMainLoop() + + if key == " ": + self.expanded = not self.expanded + self.update_expanded_icon() + return None + + hexdump_steps = {"j": 1, "k": -1, "J": 5, "K": -5} + if key in hexdump_steps: + self._scroll("hexdump", hexdump_steps[key]) + return None + + info_steps = {"h": 1, "l": -1, "H": 5, "L": -5} + if key in info_steps: + self._scroll("info", info_steps[key]) + return None + + if key in ("d", "D"): + pp, _ = self._get_pp() + dp = ":".join(str(p) for p in pp.decode_path) + dp = dp.replace(" ", "_") + if dp == "": + dp = "root" + if key == "d" or pp.expl_offset is None: + data = self.state["raw"][pp.offset:( + pp.offset + pp.tlen + pp.llen + pp.vlen + )] + else: + data = self.state["raw"][pp.expl_offset:( + pp.expl_offset + pp.expl_tlen + pp.expl_llen + pp.expl_vlen + )] + ctr = 0 + + def duplicate_path(dp, ctr): + if ctr == 0: + return dp + return "%s.%d" % (dp, ctr) + + while True: + if not path_exists(duplicate_path(dp, ctr)): + break + ctr += 1 + dp = duplicate_path(dp, ctr) + with open(dp, "wb") as fd: + fd.write(data) + self.state["decode_path"].set_text( + ("warning", "Saved to: " + dp) + ) + return None + return super(TW, self).keypress(size, key) + + class PN(urwid.ParentNode): + def __init__(self, state, value, *args, **kwargs): + self.state = state + if not hasattr(value, "_fields"): + value = list(value) + super(PN, self).__init__(value, *args, **kwargs) + + def load_widget(self): + return TW(self.state, self) + + def load_child_keys(self): + value = self.get_value() + if hasattr(value, "_fields"): + return [] + return range(len(value[1:])) + + def load_child_node(self, key): + return PN( + self.state, + self.get_value()[key + 1], + parent=self, + key=key, + depth=self.get_depth() + 1, + ) + + class LabeledPG(urwid.ProgressBar): + def __init__(self, label, *args, **kwargs): + self.label = label + super(LabeledPG, self).__init__(*args, **kwargs) + + def get_text(self): + return "%s: %s" % (self.label, super(LabeledPG, self).get_text()) + + WinHexdump = urwid.ListBox([urwid.Text("")]) + WinInfo = urwid.ListBox([urwid.Text("")]) + WinDecodePath = urwid.Text("", "center") + WinInfoBar = LabeledPG("info", "pg-normal", "pg-complete") + WinHexdumpBar = LabeledPG("hexdump", "pg-normal", "pg-complete") + WinTree = urwid.TreeListBox(urwid.TreeWalker(PN( + { + "raw": raw, + "hexed": list(hexdump(raw)), + "widget_current": None, + "info": WinInfo, + "info_bar": WinInfoBar, + "hexdump": WinHexdump, + "hexdump_bar": WinHexdumpBar, + "decode_path": WinDecodePath, + }, + list(obj.pps()), + ))) + help_text = " ".join(( + "q:quit", + "space:(un)collapse", + "(pg)up/down/home/end:nav", + "jkJK:hexdump hlHL:info", + "dD:dump", + )) + urwid.MainLoop( + urwid.Frame( + urwid.Columns([ + ("weight", 1, WinTree), + ("weight", 2, urwid.Pile([ + urwid.LineBox(WinInfo), + urwid.LineBox(WinHexdump), + ])), + ]), + header=urwid.Columns([ + ("weight", 2, urwid.AttrWrap(WinDecodePath, "header")), + ("weight", 1, WinInfoBar), + ("weight", 1, WinHexdumpBar), + ]), + footer=urwid.AttrWrap(urwid.Text(help_text), "help") + ), + [ + ("header", "bold", ""), + ("constructed", "bold", ""), + ("help", "light magenta", ""), + ("warning", "light red", ""), + ("defby", "light red", ""), + ("eoc", "dark red", ""), + ("select-value", "light green", ""), + ("select-expl", "light red", ""), + ("pg-normal", "", "light blue"), + ("pg-complete", "black", "yellow"), + ], + ).run() + + def main(): # pragma: no cover import argparse parser = argparse.ArgumentParser(description="PyDERASN ASN.1 BER/CER/DER decoder") @@ -7134,6 +7873,11 @@ def main(): # pragma: no cover action="store_true", help="Turn on event generation mode", ) + parser.add_argument( + "--browse", + action="store_true", + help="Start ASN.1 browser", + ) parser.add_argument( "RAWFile", type=argparse.FileType("rb"), @@ -7150,9 +7894,9 @@ def main(): # pragma: no cover [obj_by_path(_path) for _path in (args.oids or "").split(",")] if args.oids else () ) + from functools import partial if args.schema: schema = obj_by_path(args.schema) - from functools import partial pprinter = partial(pprint, big_blobs=True) else: schema, pprinter = generic_decoder() @@ -7162,6 +7906,11 @@ def main(): # pragma: no cover } if args.defines_by_path is not None: ctx["defines_by_path"] = obj_by_path(args.defines_by_path) + if args.browse: + obj, _ = schema().decode(raw, ctx=ctx) + browse(raw, obj, oid_maps) + from sys import exit as sys_exit + sys_exit(0) from os import environ pprinter = partial( pprinter, @@ -7184,4 +7933,5 @@ def main(): # pragma: no cover if __name__ == "__main__": + from pyderasn import * main()