#!/usr/bin/env python3 # -*- coding: utf-8 -*- """Decode a protobuf payload that has NO schema into a structured JSON tree. WHY THIS EXISTS --------------- `references/protocol-reverse.md` states the wire-format rules and says a payload can be walked field by field with no `.proto` in hand, but the skill shipped no tool that does it -- the only evidence behind that section was a workbench script. The honest part of "walk it without a schema" is what a naive decoder hides: * four different things share wire type 2 (nested message, UTF-8 string, packed repeated array, opaque bytes), and nothing on the wire separates them; * a proto3 varint 0 is byte-identical to an absent field, so a decoded `0` is not evidence that the sender set the field. A decoder that prints one interpretation per field manufactures conclusions. This one prints every interpretation consistent with the bytes, labels each with a heuristic confidence, and marks the ones the format cannot decide. WHAT IT DOES ------------ * varint, with the 10-byte ceiling and 64-bit overflow made explicit, plus the non-canonical (redundant) encoding case; * wire types 0/1/2/5 and the deprecated 3/4 group pair; * length-delimited values as candidate sets, each with a confidence and, when two candidates are equally consistent, an explicit note that no schema separates them; * proto3 explicit zero flagged at every occurrence; * several frames in one input: a `|` separator inside --hex, or a varint length prefix with --split varint-length (gRPC/WebSocket-style framing); * --reencode: write a decoded JSON tree -- possibly hand-edited -- back to bytes and compare with the original. That comparison is the round-trip proof that the walk lost nothing. No third-party dependency: `google.protobuf` parses a *stream against a schema* and cannot walk a bare one, so the parsing here is written out. (It is still useful to *generate* reference bytes from a real schema -- see the verification record -- but the tool never imports it.) USAGE ----- python protobuf_decode_raw.py --hex "08 96 01 12 07 74 65 73 74 69 6e 67" python protobuf_decode_raw.py body.bin python protobuf_decode_raw.py body.bin --json > tree.json python protobuf_decode_raw.py - < body.bin python protobuf_decode_raw.py --hex "089601|120774657374696e67" # two frames python protobuf_decode_raw.py framed.bin --split varint-length python protobuf_decode_raw.py --reencode tree.json --out patched.bin --check body.bin python protobuf_decode_raw.py --selftest Stdlib only, Python 3.9+. """ import argparse import json import re import struct import sys MAX_VARINT_BYTES = 10 MASK64 = (1 << 64) - 1 MASK32 = (1 << 32) - 1 WIRE_NAMES = { 0: 'varint', 1: 'fixed64', 2: 'length_delimited', 3: 'group_start', 4: 'group_end', 5: 'fixed32', } WIRE_SEMANTICS = { 0: 'int32/int64/uint32/uint64/sint32/sint64/bool/enum', 1: 'fixed64/sfixed64/double', 2: 'message/string/bytes/packed repeated', 3: 'deprecated group start', 4: 'deprecated group end', 5: 'fixed32/sfixed32/float', } # Tie-break order for "which candidate to expand by default". Structural # candidates come first because they carry more information; `bytes` is last # because it always fits. This is a display default, never a conclusion. VIEW_PRIORITY = ['nested_message', 'utf8_string', 'packed_varint', 'packed_fixed32', 'packed_fixed64', 'bytes', 'empty'] CAVEATS = [ 'wire type 2 carries four different things; without a schema the bytes cannot ' 'separate them, so every length-delimited field lists all candidates that fit.', 'a packed repeated field is indistinguishable from an opaque bytes value: the ' 'total length is known, the element boundaries are not.', 'a varint 0 on the wire was emitted on purpose by some writer; for a field ' 'WITHOUT presence, value 0 and "never set" are the same bytes (nothing at ' 'all), so a field missing from a decode is not evidence that 0 was the value.', 'field numbers are local to their message; nothing on the wire says which ' 'message a field number belongs to.', 'confidence values are a documented heuristic ordering, not probabilities.', ] class UsageError(Exception): """Bad input or bad flags: printed as one line, exit code 2.""" # --------------------------------------------------------------- primitives def varint_len(value): """Number of bytes a canonical (minimal) varint needs for an unsigned value.""" if value < 0: value &= MASK64 n = 1 while value >= 0x80: value >>= 7 n += 1 return n def encode_varint(value): """Canonical base-128 varint. Negative values are written as 64-bit two's complement.""" if value < 0: value &= MASK64 out = bytearray() while True: b = value & 0x7F value >>= 7 if value: out.append(b | 0x80) else: out.append(b) return bytes(out) def read_varint(buf, i, end): """Read one varint from buf[i:end]. Returns (info, next_index, None) on success and (None, i, reason) when no complete varint could be read. `info` always carries: value the decoded integer (may exceed 64 bits when overflow) hex / bytes the exact bytes consumed canonical_bytes what the minimal encoding of `value` would cost non_canonical True when the wire used more bytes than necessary overflow True when byte 10 carried bits above 2**64 too_long True when the varint ran past 10 bytes """ start = i value = 0 count = 0 overflow = False too_long = False while True: if i >= end: return None, start, 'truncated varint at offset %d' % start b = buf[i] i += 1 count += 1 if count <= 9: value |= (b & 0x7F) << (7 * (count - 1)) elif count == 10: if (b & 0x7F) > 1: overflow = True value |= (b & 0x7F) << 63 else: too_long = True if not (b & 0x80): break raw = buf[start:i] canonical = varint_len(value & MASK64) if not overflow else varint_len(value) info = { 'value': value, 'hex': raw.hex(), 'bytes': count, 'canonical_bytes': canonical, 'non_canonical': count > canonical, 'overflow': overflow, 'too_long': too_long, } return info, i, None def zigzag_decode(value): return (value >> 1) ^ -(value & 1) def int32_view(value): v = value & MASK32 return v - (1 << 32) if v >= (1 << 31) else v def int64_view(value): v = value & MASK64 return v - (1 << 64) if v >= (1 << 63) else v def _float_repr(x): """repr() of a float, never NaN/Infinity literals (illegal in strict JSON).""" return repr(x) def ascii_escape(text): """Make a decoded string safe to print on any console: pure ASCII output.""" return text.encode('unicode_escape').decode('ascii') # ----------------------------------------------------------------- decode class Ctx(object): """Collects notes (ambiguities) and errors while walking.""" def __init__(self, max_depth=5, expand=True, silent=False): self.max_depth = max_depth self.expand = expand self.silent = silent self.notes = [] self.errors = [] def note(self, kind, path, offset, detail, **extra): if self.silent: return item = {'kind': kind, 'path': path, 'offset': offset, 'detail': detail} item.update(extra) self.notes.append(item) def error(self, kind, path, offset, detail): self.errors.append({'kind': kind, 'path': path, 'offset': offset, 'detail': detail}) def walk(buf, start, end, depth, ctx, path, until_group=None, analyse=True): """Walk buf[start:end] as one message body. Returns (fields, next_index, status) where status is 'end_of_buffer' when the region was consumed exactly, 'end_group' when the matching deprecated end-group tag closed it, or 'stopped:' when it could not continue. """ fields = [] i = start while i < end: tag_off = i info, j, reason = read_varint(buf, i, end) if info is None: ctx.error('truncated_tag', path, tag_off, reason) return fields, i, 'stopped:truncated_tag' if info['too_long'] or info['overflow']: ctx.error('varint_overflow', path, tag_off, 'tag varint at offset %d is %s (%d bytes, value needs %d); ' 'the walk cannot resynchronise here' % (tag_off, 'longer than 10 bytes' if info['too_long'] else 'wider than 64 bits', info['bytes'], info['canonical_bytes'])) return fields, j, 'stopped:varint_overflow' key = info['value'] i = j field_no = key >> 3 wire = key & 7 if field_no == 0: ctx.note('field_number_zero', path, tag_off, 'tag 0x%x decodes to field number 0, which is illegal in ' 'protobuf; this is a decode error, not a field' % key) fpath = '%s.%d' % (path, field_no) if wire == 4: if until_group is not None and field_no == until_group: return fields, i, 'end_group' ctx.error('stray_end_group', path, tag_off, 'end-group tag for field %d at offset %d has no matching ' 'group start' % (field_no, tag_off)) return fields, i, 'stopped:stray_end_group' if wire not in (0, 1, 2, 3, 5): ctx.error('invalid_wire_type', path, tag_off, 'wire type %d at offset %d does not exist in protobuf' % (wire, tag_off)) return fields, i, 'stopped:invalid_wire_type' field = { 'field': field_no, 'wire': wire, 'wire_name': WIRE_NAMES[wire], 'wire_semantics': WIRE_SEMANTICS[wire], 'offset': tag_off, 'value_offset': i, 'raw_hex': '', } if wire == 0: vinfo, j2, vreason = read_varint(buf, i, end) if vinfo is None: ctx.error('truncated_value', path, i, vreason) return fields, i, 'stopped:truncated_value' value = vinfo['value'] field['view'] = 'varint' field['value'] = value field['varint'] = { 'hex': vinfo['hex'], 'bytes': vinfo['bytes'], 'canonical_bytes': vinfo['canonical_bytes'], 'non_canonical': vinfo['non_canonical'], 'overflow': vinfo['overflow'], 'too_long': vinfo['too_long'], 'as_uint64': value & MASK64, 'as_int64': int64_view(value), 'as_int32': int32_view(value), 'as_sint64_zigzag': zigzag_decode(value), 'as_bool_if_0_or_1': None if value > 1 else bool(value), } i = j2 if vinfo['non_canonical']: ctx.note('non_canonical_varint', fpath, field['value_offset'], 'field %d uses %d bytes where the minimal varint encoding ' 'needs %d; a round-trip re-encode emits the minimal form' % (field_no, vinfo['bytes'], vinfo['canonical_bytes'])) if vinfo['overflow'] or vinfo['too_long']: ctx.note('varint_over_64_bits', fpath, field['value_offset'], 'field %d consumed %d bytes carrying more than 64 bits; ' 'protobuf int64/uint64 cannot hold this value, so the ' 'field is either not a varint or the stream is corrupt' % (field_no, vinfo['bytes'])) if value == 0: ctx.note('proto3_explicit_zero', fpath, field['value_offset'], 'field %d carries an explicit varint 0, so some writer ' 'chose to emit it (a field WITH presence -- proto2 or ' 'proto3 optional -- or a hand-rolled writer). The ' 'converse matters more: for a field WITHOUT presence, ' 'value 0 and "never set" are the same bytes (nothing at ' 'all), so a field missing from this decode is not ' 'evidence that the value in use was 0' % field_no) elif wire == 1: if i + 8 > end: ctx.error('truncated_fixed64', path, i, 'fixed64 for field %d needs 8 bytes, only %d remain' % (field_no, end - i)) return fields, i, 'stopped:truncated_fixed64' raw = buf[i:i + 8] i += 8 field['view'] = 'fixed64' field['fixed64_hex'] = raw.hex() field['fixed64'] = { 'as_uint64': struct.unpack(' end: ctx.error('truncated_fixed32', path, i, 'fixed32 for field %d needs 4 bytes, only %d remain' % (field_no, end - i)) return fields, i, 'stopped:truncated_fixed32' raw = buf[i:i + 4] i += 4 field['view'] = 'fixed32' field['fixed32_hex'] = raw.hex() field['fixed32'] = { 'as_uint32': struct.unpack(' end: ctx.error('length_overruns_buffer', path, value_off, 'field %d declares %d payload bytes at offset %d but ' 'only %d remain' % (field_no, ln, value_off, end - value_off)) return fields, i, 'stopped:length_overruns_buffer' chunk = buf[j2:payload_end] field['view'] = 'bytes' field['length'] = ln field['payload_hex'] = chunk.hex() field['length_prefix'] = { 'hex': linfo['hex'], 'bytes': linfo['bytes'], 'non_canonical': linfo['non_canonical'], } if analyse: cands, view = analyse_payload(chunk, depth, ctx, fpath, value_off) field['candidates'] = cands field['view'] = view field['view_is_heuristic'] = True if view == 'nested_message': if not ctx.expand: field['expansion'] = 'suppressed by --no-expand' else: children, _nxt, _status = walk(chunk, 0, len(chunk), depth + 1, ctx, fpath, analyse=True) field['children'] = children elif view == 'utf8_string': field['string'] = chunk.decode('utf-8', 'replace') elif view == 'packed_varint': field['packed_values'] = _packed_varint_values(chunk) elif view in ('packed_fixed32', 'packed_fixed64'): step = 4 if view == 'packed_fixed32' else 8 field['packed_values_hex'] = [ chunk[k:k + step].hex() for k in range(0, len(chunk), step)] i = payload_end else: # wire == 3, deprecated group if depth >= ctx.max_depth: ctx.note('depth_limit', fpath, tag_off, 'group at depth %d was not walked: --max-depth is %d' % (depth, ctx.max_depth)) field['view'] = 'group' field['children'] = [] field['truncated_by_depth'] = True i = end else: children, j2, status = walk(buf, i, end, depth + 1, ctx, fpath, until_group=field_no, analyse=analyse) field['view'] = 'group' field['children'] = children field['group_deprecated'] = True ctx.note('deprecated_group', fpath, tag_off, 'field %d uses the deprecated group wire type (3/4); ' 'groups were removed from the language, modern writers ' 'do not emit them' % field_no) if status != 'end_group': ctx.error('unterminated_group', fpath, tag_off, 'group for field %d was never closed by a matching ' 'end-group tag' % field_no) i = j2 else: i = j2 field['end'] = i field['raw_hex'] = buf[tag_off:i].hex() fields.append(field) return fields, i, ('end_of_buffer' if i == end else 'stopped:partial') def _packed_varint_values(chunk): values = [] k = 0 while k < len(chunk): info, k2, _reason = read_varint(chunk, k, len(chunk)) if info is None: break values.append(info['value']) k = k2 return values def analyse_payload(chunk, depth, ctx, path, offset): """Every interpretation of a length-delimited payload that fits the bytes. Returns (candidates, view). `view` is the highest-confidence candidate -- a default for expansion and re-encoding, explicitly not a conclusion. """ n = len(chunk) cands = [] if n == 0: cands.append({'kind': 'empty', 'confidence': 1.0, 'note': 'zero-length payload: every interpretation is ' 'vacuous, nothing can be ranked'}) return cands, 'empty' # --- nested message ------------------------------------------------- nested_ok = False if depth < ctx.max_depth: sub = Ctx(ctx.max_depth, ctx.expand, silent=True) sub_fields, nxt, status = walk(chunk, 0, n, depth + 1, sub, path, analyse=False) if status == 'end_of_buffer' and nxt == n and not sub.errors and sub_fields: nested_ok = True cands.append({ 'kind': 'nested_message', 'confidence': 0.85, 'fields_count': len(sub_fields), 'note': 'the payload consumes exactly as a message body (%d field(s))' % len(sub_fields), }) else: reason = ('walk stopped: %s' % status if status != 'end_of_buffer' else 'walk left %d byte(s) unread' % (n - nxt)) cands.append({'kind': 'nested_message', 'confidence': 0.0, 'rejected': reason, 'note': 'rejected: %s' % reason}) else: cands.append({'kind': 'nested_message', 'confidence': 0.0, 'rejected': 'depth limit', 'note': 'not attempted: --max-depth %d reached' % ctx.max_depth}) ctx.note('depth_limit', path, offset, 'a nested reading of this field was not attempted: --max-depth %d ' 'reached. The bytes here may well be a message; re-run with a ' 'larger --max-depth before concluding that they are not' % ctx.max_depth) # --- UTF-8 string --------------------------------------------------- text = None try: text = chunk.decode('utf-8') except UnicodeDecodeError as exc: cands.append({'kind': 'utf8_string', 'confidence': 0.0, 'rejected': 'not valid UTF-8 (%s)' % exc.reason, 'note': 'rejected: %s at byte %d' % (exc.reason, exc.start)}) if text is not None: controls = [ch for ch in text if ord(ch) < 0x20 and ch not in '\t\n\r'] if not controls: conf = 0.9 note = 'valid UTF-8 with no control characters' else: conf = 0.35 note = ('valid UTF-8 but carries %d control character(s) (0x00-0x1f); ' 'binary data commonly decodes as UTF-8 by accident' % len(controls)) cands.append({'kind': 'utf8_string', 'confidence': conf, 'length_chars': len(text), 'note': note}) # --- packed arrays -------------------------------------------------- if n % 4 == 0 and n >= 8: cands.append({'kind': 'packed_fixed32', 'confidence': 0.3, 'elements': n // 4, 'element_hex': _element_preview(chunk, 4), 'note': 'length divisible by 4: could be %d packed fixed32 ' 'or float values -- element boundaries are only a ' 'guess without a schema' % (n // 4)}) if n % 8 == 0 and n >= 16: cands.append({'kind': 'packed_fixed64', 'confidence': 0.3, 'elements': n // 8, 'element_hex': _element_preview(chunk, 8), 'note': 'length divisible by 8: could be %d packed fixed64 ' 'or double values -- element boundaries are only a ' 'guess without a schema' % (n // 8)}) packed_values = _packed_varint_values(chunk) if packed_values and len(b''.join(encode_varint(v) for v in packed_values)) == n \ and len(chunk) == sum(varint_len(v) for v in packed_values): conf = 0.55 if len(packed_values) >= 2 else 0.25 cands.append({'kind': 'packed_varint', 'confidence': conf, 'elements': len(packed_values), 'note': 'the payload is exactly a sequence of %d varint(s). ' 'Without a schema this is a candidate, not a ' 'reading: a packed repeated field is ' 'indistinguishable from an opaque bytes value, so ' 'the element boundaries are not knowable from the ' 'bytes alone' % len(packed_values)}) # --- opaque bytes --------------------------------------------------- # A fallback reading, not an equal-ranking candidate: it is consistent with # every payload, so scoring it like the others would let it win ties on # alphabetical order and silently suppress the useful readings. cands.append({'kind': 'bytes', 'confidence': 0.2, 'note': 'opaque bytes: always consistent with the payload, and ' 'the only honest reading when no other candidate can be ' 'confirmed against a schema'}) # --- resolve ties --------------------------------------------------- live = [c for c in cands if c['confidence'] > 0.0] kinds = set(c['kind'] for c in live) competing = [k for k in ('nested_message', 'utf8_string', 'packed_varint') if k in kinds] if len(competing) > 1: msg = ('%s are equally consistent with these bytes; without a schema ' 'nothing on the wire separates them, so the view below is a ' 'display default and NOT a reading' % ' and '.join(sorted(competing))) for c in live: if c['kind'] != 'bytes': c['confidence'] = min(c['confidence'], 0.5) c['ambiguous_with'] = sorted(k for k in competing if k != c['kind']) c['tie_note'] = msg cands.sort(key=lambda c: (-c['confidence'], VIEW_PRIORITY.index(c['kind']) if c['kind'] in VIEW_PRIORITY else 99)) view = cands[0]['kind'] if cands else 'bytes' if view == 'empty': view = 'bytes' return cands, view def _element_preview(chunk, step, limit=4): out = [chunk[k:k + step].hex() for k in range(0, min(len(chunk), step * limit), step)] if len(chunk) > step * limit: out.append('...') return out # ---------------------------------------------------------------- re-encode def payload_from_view(field): view = field.get('view') or 'bytes' if view in ('nested_message', 'group'): return b''.join(reencode_field(c, None, '') for c in field.get('children') or []) if view == 'utf8_string': return (field.get('string') or '').encode('utf-8') if view == 'packed_varint': return b''.join(encode_varint(int(v)) for v in (field.get('packed_values') or [])) if view in ('packed_fixed32', 'packed_fixed64'): out = b'' for h in field.get('packed_values_hex') or []: out += bytes.fromhex(h) return out return bytes.fromhex(field.get('payload_hex') or '') def reencode_field(field, report, path): """Encode one field node. When `report` is given it is filled with the diff.""" wire = field['wire'] num = int(field['field']) key = encode_varint((num << 3) | wire) here = path or str(num) if wire == 0: body = encode_varint(int(field['value'])) elif wire == 1: body = bytes.fromhex(field['fixed64_hex']) elif wire == 5: body = bytes.fromhex(field['fixed32_hex']) elif wire == 2: payload = payload_from_view(field) body = encode_varint(len(payload)) + payload elif wire == 3: inner = b'' for child in field.get('children') or []: inner += reencode_field(child, report, '%s.%s' % (here, child['field'])) body = inner + encode_varint((num << 3) | 4) else: raise UsageError('wire type %d cannot be re-encoded' % wire) raw = key + body if report is not None: report['fields'] += 1 expected = field.get('raw_hex') if expected is not None and raw.hex() != expected: reason = 'the tree was edited' vinfo = field.get('varint') or {} if vinfo.get('non_canonical'): reason = ('the original varint was non-canonical (%d bytes for a ' '%d-byte value); the re-encode emits the minimal form' % (vinfo.get('bytes'), vinfo.get('canonical_bytes'))) report['mismatched'].append({ 'path': here, 'expected_hex': expected, 'got_hex': raw.hex(), 'reason': reason, }) else: report['identical'] += 1 return raw def reencode_tree(tree, report): frames = tree.get('frames') or [] out = b'' for frame in frames: for field in frame.get('fields') or []: out += reencode_field(field, report, str(field['field'])) return out # ------------------------------------------------------------------- input HEX_ESCAPE_RE = re.compile(r'\\(x[0-9A-Fa-f]{2}|[nrt0\\])') HEX_ONLY_RE = re.compile(r'[0-9A-Fa-f\s,;|]') ESCAPE_MAP = {'n': b'\n', 'r': b'\r', 't': b'\t', '0': b'\x00', '\\': b'\\'} def _unescape(text): """Turn \\xNN / \\n / \\t / \\\\ escapes into bytes.""" out = bytearray() i = 0 while i < len(text): ch = text[i] if ch == '\\' and i + 1 < len(text): nxt = text[i + 1] if nxt == 'x' and i + 3 < len(text) + 1: pair = text[i + 2:i + 4] if len(pair) == 2 and re.fullmatch(r'[0-9A-Fa-f]{2}', pair): out.append(int(pair, 16)) i += 4 continue if nxt in ESCAPE_MAP: out += ESCAPE_MAP[nxt] i += 2 continue out += ch.encode('utf-8') i += 1 return bytes(out) def parse_hex_frames(text): """Split on '|' / ';' and decode each part as hex bytes. Accepts `08 96 01`, `0x08 0x96 0x01`, newlines, commas and `\\x08\\x96\\x01`. A part that is not hex at all is taken as literal text bytes, so a pasted ASCII body still decodes. """ frames = [] for part in re.split(r'[|;]', text): if not part.strip(): continue if '\\x' in part or '\\n' in part or '\\t' in part: # Whitespace and commas are separators here too: in `08 96 | \x12\x07`, # the space after the bar is layout, not a 0x20 byte. frames.append(_unescape(re.sub(r'[\s,]', '', part))) continue cleaned = re.sub(r'0[xX]', '', part) cleaned = re.sub(r'[\s,]', '', cleaned) if not cleaned: continue if not re.fullmatch(r'[0-9A-Fa-f]+', cleaned): frames.append(part.encode('utf-8')) continue if len(cleaned) % 2: raise UsageError('odd number of hex digits in a frame (%d): a hex ' 'string must describe whole bytes' % len(cleaned)) frames.append(bytes.fromhex(cleaned)) if not frames: raise UsageError('no bytes in the hex input after stripping separators') return frames def looks_like_hex_text(text): stripped = text.strip() if not stripped: return False try: stripped.encode('ascii') except UnicodeEncodeError: return False body = HEX_ESCAPE_RE.sub('aa', stripped) body = re.sub(r'0[xX]', '', body) body = re.sub(r'[\s,;|]', '', body) if not body or not re.fullmatch(r'[0-9A-Fa-f]+', body): return False return len(body) % 2 == 0 def split_varint_length(buf): """Frames of the form , as gRPC/WebSocket use.""" frames = [] i = 0 while i < len(buf): info, j, reason = read_varint(buf, i, len(buf)) if info is None: raise UsageError('frame length prefix at offset %d: %s' % (i, reason)) if info['too_long'] or info['overflow']: raise UsageError('frame length prefix at offset %d overflows 64 bits' % i) ln = info['value'] if j + ln > len(buf): raise UsageError('frame at offset %d declares %d payload bytes but only ' '%d remain' % (i, ln, len(buf) - j)) frames.append((j, j + ln, i)) i = j + ln return frames def load_input(args): """Return (buffer, frames) where frames is a list of (start, end, prefix_offset).""" if args.hex is not None: parts = parse_hex_frames(args.hex) buf = b''.join(parts) frames = [] pos = 0 for part in parts: frames.append((pos, pos + len(part), None)) pos += len(part) return buf, frames, '--hex' if args.input is None or args.input == '-': data = sys.stdin.buffer.read() source = 'stdin' try: text = data.decode('utf-8') except UnicodeDecodeError: text = None if text is not None and looks_like_hex_text(text): parts = parse_hex_frames(text) buf = b''.join(parts) frames = [] pos = 0 for part in parts: frames.append((pos, pos + len(part), None)) pos += len(part) return buf, frames, 'stdin (hex text)' return data, [(0, len(data), None)], 'stdin (raw bytes)' try: with open(args.input, 'rb') as fh: data = fh.read() except OSError as exc: raise UsageError('cannot read %s: %s' % (args.input, exc)) return data, [(0, len(data), None)], args.input def apply_split(buf, frames, mode): if mode == 'none': return frames split = split_varint_length(buf) if not split: raise UsageError('--split varint-length found no frames in %d bytes' % len(buf)) return split # ------------------------------------------------------------------ output def field_line(field, indent): pad = ' ' * indent if field['wire'] == 0: v = field['varint'] extra = '' if v['non_canonical']: extra = ' [non-canonical: %d bytes for a %d-byte value]' % ( v['bytes'], v['canonical_bytes']) if v['overflow'] or v['too_long']: extra += ' [OVERFLOW: wider than 64 bits]' if field['value'] == 0: extra += ' [proto3: 0 is byte-identical to an absent field]' return ['%sf%d varint %d%s' % (pad, field['field'], field['value'], extra)] if field['wire'] == 1: d = field['fixed64'] return ['%sf%d fixed64 %s (u64=%d, double=%s)' % (pad, field['field'], field['fixed64_hex'], d['as_uint64'], d['as_double_repr'])] if field['wire'] == 5: d = field['fixed32'] return ['%sf%d fixed32 %s (u32=%d, float=%s)' % (pad, field['field'], field['fixed32_hex'], d['as_uint32'], d['as_float_repr'])] if field['wire'] == 2: lines = ['%sf%d len-delimited(%d) view=%s' % ( pad, field['field'], field['length'], field.get('view'))] ties = [] for cand in field.get('candidates') or []: if cand['confidence'] > 0.0: lines.append('%s candidate %-16s %.2f %s' % (pad, cand['kind'], cand['confidence'], ascii_escape(cand['note']))) elif cand.get('rejected'): lines.append('%s candidate %-16s rejected: %s' % (pad, cand['kind'], ascii_escape(str(cand['rejected'])))) if cand.get('tie_note') and cand['tie_note'] not in ties: ties.append(cand['tie_note']) for t in ties: lines.append('%s tie: %s' % (pad, ascii_escape(t))) if field.get('view') == 'utf8_string': lines.append('%s string: %r' % (pad, ascii_escape(field['string']))) elif field.get('view') == 'nested_message': for child in field.get('children') or []: lines.extend(field_line(child, indent + 2)) if field.get('expansion'): lines.append('%s expansion: %s' % (pad, field['expansion'])) elif field.get('view') == 'packed_varint': lines.append('%s packed varints: %s' % (pad, field['packed_values'])) elif field.get('view') in ('packed_fixed32', 'packed_fixed64'): lines.append('%s elements: %s' % (pad, field['packed_values_hex'])) elif field.get('view') == 'bytes': lines.append('%s bytes: %s' % (pad, field['payload_hex'])) return lines if field['wire'] == 3: lines = ['%sf%d group (deprecated wire type)' % (pad, field['field'])] for child in field.get('children') or []: lines.extend(field_line(child, indent + 1)) return lines return ['%sf%d wire%s' % (pad, field['field'], field['wire'])] def render_human(tree): out = [] inp = tree['input'] out.append('input: %s, %d byte(s)' % (inp['source'], inp['byte_length'])) out.append('hex: %s' % inp['hex']) for frame in tree['frames']: note = '' if frame.get('prefix_offset') is not None: note = ' (length-prefixed at offset %d)' % frame['prefix_offset'] out.append('') out.append('frame %d bytes %d..%d (%d B)%s status=%s' % (frame['index'], frame['start'], frame['end'], frame['length'], note, frame['walk_status'])) if not frame['fields']: out.append(' (no fields walked)') for field in frame['fields']: out.extend(field_line(field, 1)) if tree['ambiguities']: out.append('') out.append('ambiguities the wire format cannot resolve here:') for n in tree['ambiguities']: out.append(' [%s] %s @%d: %s' % (n['kind'], n['path'], n['offset'], ascii_escape(n['detail']))) if tree['errors']: out.append('') out.append('errors:') for e in tree['errors']: out.append(' [%s] %s @%d: %s' % (e['kind'], e['path'], e['offset'], ascii_escape(e['detail']))) out.append('') out.append('fixed caveats (true of every decode in this output):') for c in CAVEATS: out.append(' - %s' % c) return '\n'.join(out) def decode(args): buf, frames, source = load_input(args) frames = apply_split(buf, frames, args.split) ctx = Ctx(args.max_depth, not args.no_expand) frame_nodes = [] head = buf[:4096] input_hex = head.hex() truncated = len(buf) > len(head) for idx, frame in enumerate(frames): start, end = frame[0], frame[1] prefix = frame[2] if len(frame) > 2 else None fields, consumed, status = walk(buf, start, end, 0, ctx, 'frame%d' % idx, analyse=True) frame_nodes.append({ 'index': idx, 'start': start, 'end': end, 'length': end - start, 'prefix_offset': prefix, 'fields': fields, 'walk_status': status, 'bytes_consumed': consumed - start, }) tree = { 'tool': 'protobuf_decode_raw.py', 'schema': None, 'input': {'source': source, 'byte_length': len(buf), 'hex': input_hex, 'hex_truncated': truncated, 'frame_count': len(frame_nodes), 'split': args.split}, 'frames': frame_nodes, 'ambiguities': ctx.notes, 'errors': ctx.errors, 'caveats': CAVEATS, } if args.json: print(json.dumps(tree, indent=2, sort_keys=False, ensure_ascii=True)) else: print(render_human(tree)) return 0 if not ctx.errors else 1 def do_reencode(args): try: with open(args.reencode, 'r', encoding='utf-8') as fh: tree = json.load(fh) except OSError as exc: raise UsageError('cannot read %s: %s' % (args.reencode, exc)) except ValueError as exc: raise UsageError('%s is not valid JSON: %s' % (args.reencode, exc)) if 'frames' not in tree: raise UsageError('%s does not look like protobuf_decode_raw.py output ' '(no "frames" key)' % args.reencode) report = {'fields': 0, 'identical': 0, 'mismatched': []} out = reencode_tree(tree, report) if args.out: with open(args.out, 'wb') as fh: fh.write(out) result = { 'bytes': len(out), 'hex': out.hex(), 'fields': report['fields'], 'fields_byte_identical': report['identical'], 'mismatched': report['mismatched'], 'out_path': args.out, } if args.check: try: with open(args.check, 'rb') as fh: original = fh.read() except OSError as exc: raise UsageError('cannot read %s: %s' % (args.check, exc)) result['check_path'] = args.check result['check_bytes'] = len(original) split = (tree.get('input') or {}).get('split') original_cmp = original scope = 'the whole file' if split == 'varint-length': try: orig_frames = split_varint_length(original) except UsageError as exc: raise UsageError('%s does not carry varint-length framing: %s' % (args.check, exc)) original_cmp = b''.join(original[s:e] for s, e, _p in orig_frames) scope = ('frame bodies only: %d of %d byte(s) are length prefixes ' '(framing, not message data)' % (len(original) - len(original_cmp), len(original))) result['check_scope'] = scope if original_cmp == out: result['check'] = 'MATCH' else: result['check'] = 'MISMATCH' diff = -1 for k in range(min(len(original_cmp), len(out))): if original_cmp[k] != out[k]: diff = k break if diff < 0: diff = min(len(original_cmp), len(out)) result['first_difference_offset'] = diff result['original_at_diff'] = original_cmp[diff:diff + 8].hex() result['reencoded_at_diff'] = out[diff:diff + 8].hex() if args.json: print(json.dumps(result, indent=2, ensure_ascii=True)) else: print('re-encoded %d byte(s) from %d field(s)' % (result['bytes'], result['fields'])) print(' fields re-encoded byte-identically: %d/%d' % (result['fields_byte_identical'], result['fields'])) for m in result['mismatched']: print(' CHANGED field %s: expected %s, got %s (%s)' % (m['path'], m['expected_hex'], m['got_hex'], m['reason'])) if args.out: print(' written to %s' % args.out) else: print(' hex: %s' % result['hex']) if args.check: print(' check against %s (%d B, %s): %s' % (result['check_path'], result['check_bytes'], result.get('check_scope', 'the whole file'), result['check'])) if result['check'] == 'MISMATCH': off = result['first_difference_offset'] print(' first difference at offset %d: original %s vs re-encoded %s' % (off, result['original_at_diff'], result['reencoded_at_diff'])) return 0 if result.get('check', 'MATCH') == 'MATCH' else 1 # ----------------------------------------------------------------- selftest def _fixtures(): """Known-answer fixtures. Every byte here is produced by hand in this file.""" def tag(field, wire): return encode_varint((field << 3) | wire) def ld(field, payload): return tag(field, 2) + encode_varint(len(payload)) + payload nested_inner = tag(2, 0) + encode_varint(7) nested_outer = tag(1, 0) + encode_varint(1) + ld(2, nested_inner) packed = b''.join(encode_varint(v) for v in (3, 270, 86942)) main = b''.join([ tag(1, 0) + encode_varint(150), ld(2, b'testing'), ld(3, nested_outer), ld(4, packed), tag(5, 0) + encode_varint(0), tag(6, 1) + struct.pack(' re-encode -> byte comparison ==') for name, payload, _e, rt in fixtures: checks += 1 frames = _fixture_frames(name, payload) tree = {'frames': [{'fields': walk(payload, s, e, 0, Ctx(6), 'f', analyse=True)[0]} for s, e, _p in frames]} report = {'fields': 0, 'identical': 0, 'mismatched': []} out = reencode_tree(tree, report) expect_bytes = b''.join(payload[s:e] for s, e, _p in frames) if rt == 'non-canonical': ok = (out != expect_bytes and out.hex() == '089601' and report['mismatched'] and 'non-canonical' in report['mismatched'][0]['reason']) verdict = 'MATCH' if ok else 'FAIL' detail = ('re-encode emits the minimal form %s, original %s, and the ' 'report states why' % (out.hex(), expect_bytes.hex())) else: ok = out == expect_bytes verdict = 'MATCH' if ok else 'FAIL' detail = '' if not ok: failures.append('%s: round trip differs (%s vs %s)' % (name, expect_bytes.hex(), out.hex())) print(' %-45s %-6s (%d fields, %d identical) %s' % (name, verdict, report['fields'], report['identical'], detail)) # the two documented traps, asserted rather than narrated print('') print('== the two wire-format traps ==') checks += 2 zero = tag_0(5, 0) + encode_varint(0) absent = b'' same = (zero != absent) and walk(zero, 0, len(zero), 0, Ctx(6), 'f')[0][0]['value'] == 0 ok = walk(absent, 0, 0, 0, Ctx(6), 'f')[0] == [] if not (same and ok): failures.append('proto3 explicit zero vs absent field not distinguished correctly') print(' explicit zero bytes %s decodes to value 0; absent field decodes to no ' 'fields at all: %s' % (zero.hex(), 'PASS' if same and ok else 'FAIL')) packed = b''.join(encode_varint(v) for v in (3, 270, 86942)) cands, view = analyse_payload(packed, 0, Ctx(6), 'f', 0) live_kinds = sorted(c['kind'] for c in cands if c['confidence'] > 0.0) all_kinds = sorted(c['kind'] for c in cands) ok = ('packed_varint' in live_kinds and 'nested_message' in all_kinds and view in live_kinds and view == 'packed_varint') if not ok: failures.append('packed payload did not offer both candidates: %r' % all_kinds) print(' packed payload %s offers live=%s, rejected=%s -> view=%s : %s' % (packed.hex(), live_kinds, sorted(set(all_kinds) - set(live_kinds)), view, 'PASS' if ok else 'FAIL')) print(' -> both readings fit the same bytes; a schema is what decides.') print('') total = len(fixtures) * 2 + 2 print('selftest: %d/%d PASS' % (len(fixtures) * 2 + 2 - len(failures), total)) if failures: print('failures:') for f in failures: print(' - %s' % f) return 1 return 0 def tag_0(field, wire): return encode_varint((field << 3) | wire) def _fixture_frames(name, payload): """Framing used by both the decode and the round-trip pass of the selftest.""" if name == 'varint length framing': return split_varint_length(payload) return [(0, len(payload), None)] def _selftest_lookup(fields, frames, payload, label, num): """Pull the expected value out of a walked tree for one fixture assertion.""" m = re.match(r'^field (\d+) varint value$', label) if m: return _find_field(fields, int(m.group(1)))['value'] m = re.match(r'^field (\d+) varint overflow$', label) if m: return _find_field(fields, int(m.group(1)))['varint']['overflow'] m = re.match(r'^field (\d+) nested child count$', label) if m: return len(_find_field(fields, int(m.group(1))).get('children') or []) m = re.match(r'^field (\d+) has children$', label) if m: return len(_find_field(fields, int(m.group(1))).get('children') or []) m = re.match(r'^field (\d+) fixed64 hex$', label) if m: return _find_field(fields, int(m.group(1)))['fixed64_hex'] m = re.match(r'^field (\d+) packed element count$', label) if m: return len(_find_field(fields, int(m.group(1))).get('packed_values') or []) if label == 'packed element count': return len(_packed_varint_values(bytes.fromhex('038e029ea705'))) if label == 'top-level field count': return len(fields) if label == 'frame body hex': return payload[frames[0][0]:frames[0][1]].hex() raise UsageError('unknown selftest label %s' % label) def _find_field(fields, num): for f in fields: if f['field'] == num: return f raise UsageError('fixture field %d not found' % num) # --------------------------------------------------------------------- main def build_parser(): ap = argparse.ArgumentParser( prog='protobuf_decode_raw.py', description='Decode a schema-less protobuf payload into a JSON tree, with ' 'every length-delimited field reported as a candidate set ' 'instead of one guess. Supports hex, raw files, stdin, several ' 'frames and a round-trip re-encode.', epilog='Examples:\n' ' protobuf_decode_raw.py --hex "08 96 01 12 07 74657374696e67"\n' ' protobuf_decode_raw.py body.bin --json > tree.json\n' ' protobuf_decode_raw.py framed.bin --split varint-length\n' ' protobuf_decode_raw.py --reencode tree.json --out patched.bin ' '--check body.bin\n' ' protobuf_decode_raw.py --selftest\n', formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument('input', nargs='?', default=None, help="binary file to decode, or '-' for stdin") ap.add_argument('--hex', metavar='STR', default=None, help='decode a hex string instead of a file: accepts "0x" ' 'prefixes, spaces, newlines, commas and \\xNN escapes; ' "'|' separates independent frames") ap.add_argument('--json', action='store_true', help='print the JSON tree instead of the human tree view') ap.add_argument('--max-depth', type=int, default=5, metavar='N', help='maximum nesting depth to expand (default 5)') ap.add_argument('--no-expand', action='store_true', help='list candidates but do not recurse into nested messages') ap.add_argument('--split', choices=('none', 'varint-length'), default='none', help='frame the input: "none" (default, one message) or ' '"varint-length" for framing') ap.add_argument('--reencode', metavar='JSON', default=None, help='re-encode a decoded JSON tree back to bytes ' '(round-trip and edit-and-rebuild)') ap.add_argument('--out', metavar='PATH', default=None, help='with --reencode: where to write the bytes') ap.add_argument('--check', metavar='PATH', default=None, help='with --reencode: compare the result against this file') ap.add_argument('--selftest', action='store_true', help='run the built-in known-answer fixtures and their ' 'round-trip checks') return ap def main(argv=None): args = build_parser().parse_args(argv) if args.selftest: return selftest(args) if args.reencode: return do_reencode(args) return decode(args) if __name__ == '__main__': try: sys.exit(main()) except UsageError as exc: sys.stderr.write('error: %s\n' % exc) sys.exit(2) except BrokenPipeError: sys.exit(0)