#!/usr/bin/env python3 """ sg_bytecode_ir.py — parses a decrypted SourceGuardian body.bin (the output of sg_container_decode.py) into a normalized JSON structure and a readable opcode listing. Implements section 08 of "SourceGuardian PHP 8.1 — Static Container & Bytecode Recovery Without Execution" (https://www.amariei.org/sourceguardian-php81-static-decode.html): SOURCEGUARDIAN magic -> function/class/main op_array records -> per-opcode operand decoding, resolved against literals and compiled-variable names. Opcode identity uses PHP 8.1's own public Zend/zend_vm_opcodes.h numbering (SourceGuardian does not renumber opcodes — see the article, section 08). Usage: python sg_bytecode_ir.py dump/article_fixture.body.bin -> writes dump/article_fixture.body.icdump.json and .icdump.txt """ import json import os import struct import sys # ── PHP 8.1 opcode names, from Zend/zend_vm_opcodes.h ────────────────────── OPCODE_NAMES = { 0: 'ZEND_NOP', 1: 'ZEND_ADD', 2: 'ZEND_SUB', 3: 'ZEND_MUL', 4: 'ZEND_DIV', 5: 'ZEND_MOD', 6: 'ZEND_SL', 7: 'ZEND_SR', 8: 'ZEND_CONCAT', 9: 'ZEND_BW_OR', 10: 'ZEND_BW_AND', 11: 'ZEND_BW_XOR', 12: 'ZEND_POW', 13: 'ZEND_BW_NOT', 14: 'ZEND_BOOL_NOT', 15: 'ZEND_BOOL_XOR', 16: 'ZEND_IS_IDENTICAL', 17: 'ZEND_IS_NOT_IDENTICAL', 18: 'ZEND_IS_EQUAL', 19: 'ZEND_IS_NOT_EQUAL', 20: 'ZEND_IS_SMALLER', 21: 'ZEND_IS_SMALLER_OR_EQUAL', 22: 'ZEND_ASSIGN', 23: 'ZEND_ASSIGN_DIM', 24: 'ZEND_ASSIGN_OBJ', 25: 'ZEND_ASSIGN_STATIC_PROP', 26: 'ZEND_ASSIGN_OP', 27: 'ZEND_ASSIGN_DIM_OP', 28: 'ZEND_ASSIGN_OBJ_OP', 29: 'ZEND_ASSIGN_STATIC_PROP_OP', 30: 'ZEND_ASSIGN_REF', 31: 'ZEND_QM_ASSIGN', 32: 'ZEND_ASSIGN_OBJ_REF', 33: 'ZEND_ASSIGN_STATIC_PROP_REF', 34: 'ZEND_PRE_INC', 35: 'ZEND_PRE_DEC', 36: 'ZEND_POST_INC', 37: 'ZEND_POST_DEC', 38: 'ZEND_PRE_INC_STATIC_PROP', 39: 'ZEND_PRE_DEC_STATIC_PROP', 40: 'ZEND_POST_INC_STATIC_PROP', 41: 'ZEND_POST_DEC_STATIC_PROP', 42: 'ZEND_JMP', 43: 'ZEND_JMPZ', 44: 'ZEND_JMPNZ', 45: 'ZEND_JMPZNZ', 46: 'ZEND_JMPZ_EX', 47: 'ZEND_JMPNZ_EX', 48: 'ZEND_CASE', 49: 'ZEND_CHECK_VAR', 50: 'ZEND_SEND_VAR_NO_REF_EX', 51: 'ZEND_CAST', 52: 'ZEND_BOOL', 53: 'ZEND_FAST_CONCAT', 54: 'ZEND_ROPE_INIT', 55: 'ZEND_ROPE_ADD', 56: 'ZEND_ROPE_END', 57: 'ZEND_BEGIN_SILENCE', 58: 'ZEND_END_SILENCE', 59: 'ZEND_INIT_FCALL_BY_NAME', 60: 'ZEND_DO_FCALL', 61: 'ZEND_INIT_FCALL', 62: 'ZEND_RETURN', 63: 'ZEND_RECV', 64: 'ZEND_RECV_INIT', 65: 'ZEND_SEND_VAL', 66: 'ZEND_SEND_VAR_EX', 67: 'ZEND_SEND_REF', 68: 'ZEND_NEW', 69: 'ZEND_INIT_NS_FCALL_BY_NAME', 70: 'ZEND_FREE', 71: 'ZEND_INIT_ARRAY', 72: 'ZEND_ADD_ARRAY_ELEMENT', 73: 'ZEND_INCLUDE_OR_EVAL', 74: 'ZEND_UNSET_VAR', 75: 'ZEND_UNSET_DIM', 76: 'ZEND_UNSET_OBJ', 77: 'ZEND_FE_RESET_R', 78: 'ZEND_FE_FETCH_R', 79: 'ZEND_EXIT', 80: 'ZEND_FETCH_R', 81: 'ZEND_FETCH_DIM_R', 82: 'ZEND_FETCH_OBJ_R', 83: 'ZEND_FETCH_W', 84: 'ZEND_FETCH_DIM_W', 85: 'ZEND_FETCH_OBJ_W', 86: 'ZEND_FETCH_RW', 87: 'ZEND_FETCH_DIM_RW', 88: 'ZEND_FETCH_OBJ_RW', 89: 'ZEND_FETCH_IS', 90: 'ZEND_FETCH_DIM_IS', 91: 'ZEND_FETCH_OBJ_IS', 92: 'ZEND_FETCH_FUNC_ARG', 93: 'ZEND_FETCH_DIM_FUNC_ARG', 94: 'ZEND_FETCH_OBJ_FUNC_ARG', 95: 'ZEND_FETCH_UNSET', 96: 'ZEND_FETCH_DIM_UNSET', 97: 'ZEND_FETCH_OBJ_UNSET', 98: 'ZEND_FETCH_LIST_R', 99: 'ZEND_FETCH_CONSTANT', 100: 'ZEND_CHECK_FUNC_ARG', 101: 'ZEND_EXT_STMT', 102: 'ZEND_EXT_FCALL_BEGIN', 103: 'ZEND_EXT_FCALL_END', 104: 'ZEND_EXT_NOP', 105: 'ZEND_TICKS', 106: 'ZEND_SEND_VAR_NO_REF', 107: 'ZEND_CATCH', 108: 'ZEND_THROW', 109: 'ZEND_FETCH_CLASS', 110: 'ZEND_CLONE', 111: 'ZEND_RETURN_BY_REF', 112: 'ZEND_INIT_METHOD_CALL', 113: 'ZEND_INIT_STATIC_METHOD_CALL', 114: 'ZEND_ISSET_ISEMPTY_VAR', 115: 'ZEND_ISSET_ISEMPTY_DIM_OBJ', 116: 'ZEND_SEND_VAL_EX', 117: 'ZEND_SEND_VAR', 118: 'ZEND_INIT_USER_CALL', 119: 'ZEND_SEND_ARRAY', 120: 'ZEND_SEND_USER', 121: 'ZEND_STRLEN', 122: 'ZEND_DEFINED', 123: 'ZEND_TYPE_CHECK', 124: 'ZEND_VERIFY_RETURN_TYPE', 125: 'ZEND_FE_RESET_RW', 126: 'ZEND_FE_FETCH_RW', 127: 'ZEND_FE_FREE', 128: 'ZEND_INIT_DYNAMIC_CALL', 129: 'ZEND_DO_ICALL', 130: 'ZEND_DO_UCALL', 131: 'ZEND_DO_FCALL_BY_NAME', 132: 'ZEND_PRE_INC_OBJ', 133: 'ZEND_PRE_DEC_OBJ', 134: 'ZEND_POST_INC_OBJ', 135: 'ZEND_POST_DEC_OBJ', 136: 'ZEND_ECHO', 137: 'ZEND_OP_DATA', 138: 'ZEND_INSTANCEOF', 139: 'ZEND_GENERATOR_CREATE', 140: 'ZEND_MAKE_REF', 141: 'ZEND_DECLARE_FUNCTION', 142: 'ZEND_DECLARE_LAMBDA_FUNCTION', 143: 'ZEND_DECLARE_CONST', 144: 'ZEND_DECLARE_CLASS', 145: 'ZEND_DECLARE_CLASS_DELAYED', 146: 'ZEND_DECLARE_ANON_CLASS', 147: 'ZEND_ADD_ARRAY_UNPACK', 148: 'ZEND_ISSET_ISEMPTY_PROP_OBJ', 149: 'ZEND_HANDLE_EXCEPTION', 150: 'ZEND_USER_OPCODE', 151: 'ZEND_ASSERT_CHECK', 152: 'ZEND_JMP_SET', 153: 'ZEND_UNSET_CV', 154: 'ZEND_ISSET_ISEMPTY_CV', 155: 'ZEND_FETCH_LIST_W', 156: 'ZEND_SEPARATE', 157: 'ZEND_FETCH_CLASS_NAME', 158: 'ZEND_CALL_TRAMPOLINE', 159: 'ZEND_DISCARD_EXCEPTION', 160: 'ZEND_YIELD', 161: 'ZEND_GENERATOR_RETURN', 162: 'ZEND_FAST_CALL', 163: 'ZEND_FAST_RET', 164: 'ZEND_RECV_VARIADIC', 165: 'ZEND_SEND_UNPACK', 166: 'ZEND_YIELD_FROM', 167: 'ZEND_COPY_TMP', 168: 'ZEND_BIND_GLOBAL', 169: 'ZEND_COALESCE', 170: 'ZEND_SPACESHIP', 171: 'ZEND_FUNC_NUM_ARGS', 172: 'ZEND_FUNC_GET_ARGS', 173: 'ZEND_FETCH_STATIC_PROP_R', 174: 'ZEND_FETCH_STATIC_PROP_W', 175: 'ZEND_FETCH_STATIC_PROP_RW', 176: 'ZEND_FETCH_STATIC_PROP_IS', 177: 'ZEND_FETCH_STATIC_PROP_FUNC_ARG', 178: 'ZEND_FETCH_STATIC_PROP_UNSET', 179: 'ZEND_UNSET_STATIC_PROP', 180: 'ZEND_ISSET_ISEMPTY_STATIC_PROP', 181: 'ZEND_FETCH_CLASS_CONSTANT', 182: 'ZEND_BIND_LEXICAL', 183: 'ZEND_BIND_STATIC', 184: 'ZEND_FETCH_THIS', 185: 'ZEND_SEND_FUNC_ARG', 186: 'ZEND_ISSET_ISEMPTY_THIS', 187: 'ZEND_SWITCH_LONG', 188: 'ZEND_SWITCH_STRING', 189: 'ZEND_IN_ARRAY', 190: 'ZEND_COUNT', 191: 'ZEND_GET_CLASS', 192: 'ZEND_GET_CALLED_CLASS', 193: 'ZEND_GET_TYPE', 194: 'ZEND_ARRAY_KEY_EXISTS', 195: 'ZEND_MATCH', 196: 'ZEND_CASE_STRICT', 197: 'ZEND_MATCH_ERROR', 198: 'ZEND_JMP_NULL', 199: 'ZEND_CHECK_UNDEF_ARGS', 200: 'ZEND_FETCH_GLOBALS', 201: 'ZEND_VERIFY_NEVER_TYPE', 202: 'ZEND_CALLABLE_CONVERT', } IS_UNUSED, IS_CONST, IS_TMP_VAR, IS_VAR, IS_CV = 0, 1, 2, 4, 8 IS_NAMES = {0: 'IS_UNUSED', 1: 'IS_CONST', 2: 'IS_TMP_VAR', 4: 'IS_VAR', 8: 'IS_CV'} class ParseError(Exception): pass class StreamReader: def __init__(self, data: bytes): self.data, self.pos = data, 0 def read_u8(self): if self.pos >= len(self.data): raise ParseError(f'EOF read_u8 @{self.pos}') v = self.data[self.pos] self.pos += 1 return v def read_u32(self): if self.pos + 4 > len(self.data): raise ParseError(f'EOF read_u32 @{self.pos}') v = struct.unpack_from(' len(self.data): raise ParseError(f'EOF read_double @{self.pos}') v = struct.unpack_from(' len(self.data): raise ParseError(f'EOF read_bytes({n}) @{self.pos}') v = self.data[self.pos:self.pos + n] self.pos += n return v def read_zstr(self): n = self.read_u32() if n == 0xFFFFFFFF: return None data = self.read_bytes(n) try: return data.decode('utf-8') except UnicodeDecodeError: return data.decode('latin-1') def skip(self, n): if self.pos + n > len(self.data): raise ParseError(f'skip({n}) past EOF @{self.pos}') self.pos += n def version_gate(r: StreamReader, ver: int): if ver < 0x17: return while True: b = r.read_u8() if b == 0xFF: break if b == 0xF0: r.skip(r.read_u32()) break # ── operand slot primitives ───────────────────────────────────────────────── def _slot(r: StreamReader): """IS_TMP/VAR/CV slot: u32 index -> (raw_idx, byte_offset).""" idx = r.read_u32() return idx, 16 * (idx + 3) def _literal_idx(r: StreamReader) -> int: return r.read_u32() def _cache_slot(r: StreamReader, ext_handles: int = 0) -> int: return 4 * (ext_handles + r.read_u32()) def _generic_operand(r: StreamReader): """type_byte + dispatch -> (type_byte, raw_idx, zend_value).""" tb = r.read_u8() t = tb & 0xF if t == IS_CONST: idx = _literal_idx(r) return tb, idx, idx if t in (IS_TMP_VAR, IS_VAR, IS_CV): raw, slot = _slot(r) return tb, raw, slot val = r.read_u32() return tb, 0, val def _op(tp, **kw): base = {'type': tp, 'type_name': IS_NAMES.get(tp, f'IS_{tp}'), 'zend_value': 0} base.update(kw) return base def _make_operand(tp_byte, raw_idx, zend_val, cv_names, literals): t = tp_byte & 0xF o = _op(t, zend_value=zend_val) if t == IS_CONST: o['literal_index'] = raw_idx if 0 <= raw_idx < len(literals): o['literal_value'] = literals[raw_idx].get('value') elif t == IS_CV: o['variable_index'] = raw_idx name = cv_names[raw_idx] if 0 <= raw_idx < len(cv_names) else f'cv{raw_idx}' o['variable_name'] = '$' + name elif t in (IS_TMP_VAR, IS_VAR): o['variable_index'] = raw_idx return o def _unused(val=0): return _op(IS_UNUSED, zend_value=val) def _jump_operand(target): o = _op(IS_UNUSED, zend_value=target) o['jump_target'] = target return o def _dec_result(r, nibble, ext_handles, cv_names, literals): if nibble == 0xF: return _unused() if nibble == 0: tb, ri, zv = _generic_operand(r) return _make_operand(tb, ri, zv, cv_names, literals) if nibble == 1: tb = r.read_u8() ri, zv = _slot(r) return _make_operand(tb, ri, zv, cv_names, literals) if nibble == 2: tb = r.read_u8() v = r.read_u32() return _make_operand(tb, 0, v, cv_names, literals) if nibble == 3: slot = _cache_slot(r, ext_handles) return _op(IS_VAR, zend_value=slot) if nibble == 4: tb = r.read_u8() slot = _cache_slot(r, ext_handles) if tb == 1 else _slot(r)[1] return _make_operand(tb, 0, slot, cv_names, literals) raise ParseError(f'unknown result nibble {nibble:#x}') def _dec_op1(r, nibble, ext_handles, cv_names, literals): if nibble == 0xF: return _unused() if nibble == 0: tb, ri, zv = _generic_operand(r) return _make_operand(tb, ri, zv, cv_names, literals) if nibble == 1: tb = r.read_u8() ri, zv = _slot(r) return _make_operand(tb, ri, zv, cv_names, literals) if nibble == 2: tb = r.read_u8() v = r.read_u32() return _make_operand(tb, 0, v, cv_names, literals) if nibble == 3: r.read_u8() li = _literal_idx(r) return _make_operand(IS_CONST, li, li, cv_names, literals) if nibble == 4: return _jump_operand(r.read_u32()) if nibble == 5: tb = r.read_u8() if tb == 0: return _unused(r.read_u32()) if tb == 1: li = _literal_idx(r) return _make_operand(IS_CONST, li, li, cv_names, literals) ri, zv = _slot(r) return _make_operand(tb, ri, zv, cv_names, literals) raise ParseError(f'unknown op1 nibble {nibble:#x}') def _dec_op2(r, nibble, ext_handles, cv_names, literals): if nibble == 0xF: return _unused() if nibble == 0: tb, ri, zv = _generic_operand(r) return _make_operand(tb, ri, zv, cv_names, literals) if nibble == 1: tb = r.read_u8() ri, zv = _slot(r) return _make_operand(tb, ri, zv, cv_names, literals) if nibble == 2: tb = r.read_u8() v = r.read_u32() return _make_operand(tb, 0, v, cv_names, literals) if nibble == 3: r.read_u8() li = _literal_idx(r) return _make_operand(IS_CONST, li, li, cv_names, literals) if nibble == 4: return _jump_operand(r.read_u32()) if nibble == 5: tb = r.read_u8() if tb == 0: return _unused(r.read_u32()) if tb == 1: li = _literal_idx(r) return _make_operand(IS_CONST, li, li, cv_names, literals) ri, zv = _slot(r) return _make_operand(tb, ri, zv, cv_names, literals) if nibble == 6: slot = _cache_slot(r, ext_handles) return _op(IS_VAR, zend_value=slot) raise ParseError(f'unknown op2 nibble {nibble:#x}') def _parse_opcode(r: StreamReader, ver: int, index: int, literals: list, cv_names: list, ext_handles: int = 0) -> dict: opcode = r.read_u8() type_word = r.read_u32() if ver >= 0x1B else 0xFFFFFF ext_n = (type_word >> 12) & 0xF res_n = (type_word >> 8) & 0xF op1_n = type_word & 0xF op2_n = (type_word >> 4) & 0xF ext_val = 0 if ext_n == 0xF: pass elif ext_n in (2, 3): ext_val = r.read_u32() elif ext_n == 1: _, ext_val = _slot(r) elif ext_n == 4: ext_val = _cache_slot(r, ext_handles) elif ext_n == 5: ext_val = _cache_slot(r, ext_handles) flags = r.read_u8() ext_val = (ext_val & ~3) | (flags & 3) elif ext_n == 6: ext_val = 24 * r.read_u32() flags = r.read_u8() ext_val = (ext_val & ~3) | (flags & 3) else: raise ParseError(f'opcode {opcode}@{index}: bad ext nibble {ext_n:#x}') # Dynamic-call opcodes (205/206/207, all serialized as ZEND_SG_FCALL) # carry the resolved callee name as an extra length-prefixed string # right here, before result/op1/op2 — easy to miss since it's the one # opcode shape that isn't a fixed nibble-driven operand triplet. call_name = None if opcode in (205, 206, 207) and _next_zstr_plausible(r): call_name = r.read_zstr() result = _dec_result(r, res_n, ext_handles, cv_names, literals) op1 = _dec_op1(r, op1_n, ext_handles, cv_names, literals) op2 = _dec_op2(r, op2_n, ext_handles, cv_names, literals) lineno = r.read_u32() opcode_name = OPCODE_NAMES.get(opcode, f'ZEND_UNKNOWN_{opcode}') if opcode in (205, 206, 207) and call_name: opcode_name = 'ZEND_SG_FCALL' return { 'index': index, 'opcode': opcode, 'opcode_name': opcode_name, 'lineno': lineno, 'extended_value': ext_val, 'call_name': call_name, 'op1': op1, 'op2': op2, 'result': result, } # ── zval / hashtable ───────────────────────────────────────────────────────── def _parse_zval(r: StreamReader, ver: int) -> dict: type_info = r.read_u32() tail = r.read_u32() if tail == 0xFFFFFFFF: r.read_u32() base = type_info & 0xFF if base in (127, 126): # SG_FILE / SG_DIR marker (no extra bytes) return {'type': 'sg_special', 'tag': base, 'value': None} if base in (125, 124): # SG path constant: a string + one extra u32 s = r.read_zstr() r.read_u32() return {'type': 'string', 'value': s} if base == 0: return {'type': 'undef', 'value': None} if base == 1: return {'type': 'null', 'value': None} if base == 2: return {'type': 'bool', 'value': False} if base == 3: return {'type': 'bool', 'value': True} if base == 4: v = r.read_u32() return {'type': 'int', 'value': v - 0x100000000 if v >= 0x80000000 else v} if base == 5: return {'type': 'float', 'value': r.read_double()} if base == 6: return {'type': 'string', 'value': r.read_zstr()} if base == 7: ht = _parse_hashtable(r, ver) r.read_u32() r.read_u32() return {'type': 'array', 'value': ht} return {'type': 'unknown', 'value': base} def _parse_hashtable(r: StreamReader, ver: int) -> dict: count = r.read_u32() if count == 0: return {} if ver >= 0x20: r.read_u8() # packed flag result = {} for _ in range(count): klen = r.read_u32() if klen == 0xFFFFFFFF: key = r.read_u32() else: raw = r.read_bytes(klen) try: key = raw.decode('utf-8') except UnicodeDecodeError: key = raw.decode('latin-1') result[key] = _parse_zval(r, ver) return result def _skip_type_info(r: StreamReader, ver: int): hi = r.read_u32() if hi & 0x40000000: count = r.read_u32() for _ in range(count): entry_hi = r.read_u32() if ver >= 0x21 else hi if (entry_hi & 0x10000000) or ver < 0x21: r.read_zstr() elif entry_hi & 0x40000000: _skip_type_info(r, ver) elif hi & 0x11000000: r.read_zstr() def _skip_attributes(r: StreamReader, ver: int): count = r.read_u32() for _ in range(count): r.read_zstr() r.read_u32() r.read_u32() r.read_u32() arg_count = r.read_u32() for _ in range(arg_count): r.read_zstr() _parse_zval(r, ver) # ── op_array / class / container ──────────────────────────────────────────── def _read_raw_oparray_counts(r: StreamReader): return (r.read_u32(), r.read_u32(), r.read_u32(), r.read_u32(), r.read_u32()) def _oparray_counts_plausible(r: StreamReader, counts) -> bool: _ext_handles_count, literal_slot_size, opcode_count, T, num_cvs = counts remaining = len(r.data) - r.pos if literal_slot_size not in (16, 24, 28, 32): return False if opcode_count > 200000 or T > 200000 or num_cvs > 10000: return False if opcode_count == 0 and (T != 0 or num_cvs != 0): return False return num_cvs * 4 + opcode_count * 9 <= remaining def _read_oparray_counts(r: StreamReader, ver: int, literals: list): """A small number of SG PHP 8.1 bodies serialize 1-16 extra zvals immediately before this fixed-shape counts block (an off-by-N in the encoder's literal-table bookkeeping for certain literal shapes). Probe for the count that makes the following block plausible, the same way the reference implementation does, rather than assuming zero.""" counts_pos = r.pos for extra_count in range(0, 17): r.pos = counts_pos extras = [] try: for _ in range(extra_count): extras.append(_parse_zval(r, ver)) counts = _read_raw_oparray_counts(r) except ParseError: continue if _oparray_counts_plausible(r, counts): literals.extend(extras) return counts r.pos = counts_pos return _read_raw_oparray_counts(r) def parse_op_array(r: StreamReader, ver: int, has_hybrid_vm: bool = False) -> dict: version_gate(r, ver) type_byte = r.read_u8() # 1 = function stub, 2 = full op_array if type_byte not in (1, 2): raise ParseError(f'bad op_array type {type_byte} @{r.pos}') fn_flags = r.read_u32() r.read_u8(); r.read_u8(); r.read_u8() num_args = r.read_u32() required_num_args = r.read_u32() arg_names = [] if fn_flags & 0x6000: # return-type info precedes params r.read_zstr() _skip_type_info(r, ver) for _ in range(num_args): arg_names.append(r.read_zstr() or '') _skip_type_info(r, ver) function_name = r.read_zstr() r.read_zstr() # scope name (consumed, not needed here) if type_byte == 1: return {'kind': 'function_stub', 'function_name': function_name or '', 'arg_names': arg_names, 'opcodes': [], 'literals': [], 'cv_names': []} literal_count = r.read_u32() literals = [_parse_zval(r, ver) for _ in range(literal_count)] _, _, opcode_count, T, num_cvs = _read_oparray_counts(r, ver, literals) cv_names = [r.read_zstr() or '' for _ in range(num_cvs)] pre_data = [r.read_u32() for _ in range(opcode_count)] if has_hybrid_vm else [] opcodes = [_parse_opcode(r, ver, i, literals, cv_names) for i in range(opcode_count)] lr_count = r.read_u32() for _ in range(lr_count): if ver >= 0x1B: r.read_u8(); _slot(r) elif ver >= 0x1A: _slot(r) else: r.read_u32() r.read_u32(); r.read_u32() tc_count = r.read_u32() try_catches = [] for _ in range(tc_count): try_catches.append({ 'try_op': r.read_u32(), 'catch_op': r.read_u32(), 'finally_op': r.read_u32(), 'finally_end': r.read_u32(), }) _parse_hashtable(r, ver) # static variables doc_comment = r.read_zstr() line_start = r.read_u32() line_end = r.read_u32() r.read_u32() # prototype depth _skip_attributes(r, ver) nested_count = r.read_u32() nested = [parse_op_array(r, ver, has_hybrid_vm) for _ in range(nested_count)] return { 'kind': 'op_array', 'fn_flags': fn_flags, 'function_name': function_name or '', 'arg_names': arg_names, 'literals': literals, 'opcode_count': opcode_count, 'T': T, 'num_cvs': num_cvs, 'cv_names': cv_names, 'pre_data': pre_data, 'opcodes': opcodes, 'try_catches': try_catches, 'doc_comment': doc_comment, 'line_start': line_start, 'line_end': line_end, 'nested': nested, } def parse_container(data: bytes) -> dict: r = StreamReader(data) magic = r.read_bytes(15) if magic != b'SOURCEGUARDIAN\x00': raise ParseError(f'bad magic {magic!r}') ver, flags = r.read_u32(), r.read_u32() r.read_u32(); r.read_u32() # reserved words if ver > 0x21: raise ParseError(f'unsupported container version {ver}') has_hybrid_vm = bool(flags & 2) functions, classes = [], [] for phase in range(3): if phase == 1: while (tag := r.read_u8()) != 0: if tag != 1: raise ParseError(f'class tag={tag:#x} @{r.pos}') r.read_bytes(r.read_u32()) # SG composite key, opaque classes.append(parse_class(r, ver, has_hybrid_vm)) if ver >= 0x1D: break else: while (tag := r.read_u8()) != 0: if tag != 2: raise ParseError(f'func tag={tag:#x} @{r.pos}') r.read_bytes(r.read_u32()) functions.append(parse_op_array(r, ver, has_hybrid_vm)) if ver < 0x1D: break main = parse_op_array(r, ver, has_hybrid_vm) return {'ver': ver, 'flags': flags, 'functions': functions, 'classes': classes, 'main': main} # ── class property table (two PHP 8.1 build variants) ────────────────────── def _next_u32_is_probable_prop_slot(r: StreamReader) -> bool: if r.pos + 8 > len(r.data): return False first = struct.unpack_from('= 0x10000: return False if 0 < second <= 512 and r.pos + 8 + second <= len(r.data): return True if r.pos + 9 <= len(r.data): shifted = struct.unpack_from(' 512 or length > len(r.data) - r.pos) and shifted <= 512: r.read_u8() return r.read_zstr() def _next_zstr_plausible(r: StreamReader, limit: int = 512) -> bool: if r.pos + 4 > len(r.data): return False length = struct.unpack_from(' limit or r.pos + 4 + length > len(r.data): return False data = r.data[r.pos + 4:r.pos + 4 + length] return all(b in (9, 10, 13) or 32 <= b < 127 for b in data) def _parse_property_zval(r: StreamReader, ver: int) -> dict: type_info = r.read_u32() tail = r.read_u32() if tail == 0xFFFFFFFF: r.read_u32() base = type_info & 0xFF if base == 7: ht = _parse_hashtable(r, ver) r.read_u32() return {'type': 'array', 'value': ht} rr = StreamReader(r.data) rr.pos = r.pos - (12 if tail == 0xFFFFFFFF else 8) value = _parse_zval(rr, ver) r.pos = rr.pos return value def _skip_class_props(r: StreamReader, ver: int): total = r.read_u32() r.read_u32() # static count (loop still runs `total` times) for _ in range(total): if _next_u32_is_probable_prop_slot(r): r.read_u32() # property slot/offset _read_prop_zstr(r) # property name r.read_u32() # flags _read_prop_zstr(r) # mangled/typed-property name _read_prop_zstr(r) # doc comment _parse_property_zval(r, ver) _skip_type_info(r, ver) _skip_attributes(r, ver) def _attributes_plausible_at(r: StreamReader, pos: int, ver: int, require_nonempty: bool = False) -> bool: if pos + 4 > len(r.data): return False rr = StreamReader(r.data) rr.pos = pos try: count = rr.read_u32() if require_nonempty and count == 0: return False if count > 64: return False for _ in range(count): if not _next_zstr_plausible(rr, 512): return False name = rr.read_zstr() if not isinstance(name, str) or not name: return False if rr.pos + 16 > len(rr.data): return False rr.read_u32(); rr.read_u32(); rr.read_u32() arg_count = rr.read_u32() if arg_count > 64: return False for _ in range(arg_count): if not _next_zstr_plausible(rr, 512): return False rr.read_zstr() _parse_zval(rr, ver) except ParseError: return False return True def _skip_class_attributes(r: StreamReader, ver: int): """One SG PHP 8.1 build inserts an empty 5-byte aux table before a class's own attribute list (observed with #[AllowDynamicProperties]); probe for a real, non-empty attribute list after it before consuming those 5 bytes.""" if (r.pos + 9 <= len(r.data) and struct.unpack_from(' bool: if pos + 4 > len(r.data): return False count = struct.unpack_from(' 4096: return False if count == 0: return True name_pos = pos + 4 if name_pos + 4 > len(r.data): return False length = struct.unpack_from(' 256 or name_pos + 4 + length > len(r.data): return False name = r.data[name_pos + 4:name_pos + 4 + length] if not all(b in (92, 95, 36) or 48 <= b <= 57 or 65 <= b <= 90 or 97 <= b <= 122 for b in name): return False gate_pos = name_pos + 4 + length return r.data[gate_pos] in (0xF0, 0xFF) if gate_pos < len(r.data) else False def _skip_optional_class_aux_table(r: StreamReader, ver: int) -> None: """One SG PHP 8.1 build inserts an extra empty HashTable + packed-flag byte between the static-properties table and the method table when the class has constants; older/simpler classes place method_count directly here. Probe both layouts rather than assuming one.""" pos = r.pos if _method_count_plausible(r, pos) and struct.unpack_from('= 0x20 else 4 if pos + aux_skip > len(r.data): return if struct.unpack_from('= 0x20 and r.data[pos + 4] != 0: return if _method_count_plausible(r, pos + aux_skip): r.skip(aux_skip) def parse_class(r: StreamReader, ver: int, has_hybrid_vm: bool = False) -> dict: version_gate(r, ver) type_byte = r.read_u8() if type_byte != 2: raise ParseError(f'class type_byte={type_byte} (expected 2) @{r.pos}') class_name = r.read_zstr() or '' r.read_u32() # class flags parent_name = r.read_zstr() interfaces = [r.read_zstr() for _ in range(r.read_u32())] traits = [r.read_zstr() for _ in range(r.read_u32())] for _ in range(r.read_u32()): # trait aliases mcount = r.read_u32() r.read_zstr(); r.read_zstr() for _ in range(mcount): r.read_zstr() for _ in range(r.read_u32()): # trait precedences r.read_zstr(); r.read_zstr(); r.read_zstr(); r.read_u32() doc_comment = r.read_zstr() constants = {} for _ in range(r.read_u32()): cname = r.read_zstr() or '' cval = _parse_zval(r, ver) r.read_u32(); r.read_zstr() _skip_attributes(r, ver) r.read_u32() constants[cname] = cval _skip_class_props(r, ver) _skip_class_attributes(r, ver) r.read_u32() # trailing flags/line word _parse_hashtable(r, ver) # static properties table _skip_optional_class_aux_table(r, ver) methods = [] for _ in range(r.read_u32()): method_name = r.read_zstr() or '' oa = parse_op_array(r, ver, has_hybrid_vm) oa['method_name'] = method_name methods.append(oa) return { 'class_name': class_name, 'parent_name': parent_name, 'interfaces': interfaces, 'traits': traits, 'doc_comment': doc_comment, 'constants': constants, 'methods': methods, } # ── JSON / text output ─────────────────────────────────────────────────────── def _fmt_operand(op: dict) -> str: t = op['type_name'] if t == 'IS_UNUSED': return f'UNUSED({op.get("jump_target", op["zend_value"])})' if 'jump_target' in op else 'UNUSED' if t == 'IS_CONST': return f'CONST#{op.get("literal_index")}={op.get("literal_value")!r}' if t == 'IS_CV': return op.get('variable_name', '?') return f'{t}#{op.get("variable_index", op["zend_value"])}' def render_text(oa: dict, header: str = '') -> str: lines = [header] if header else [] for op in oa.get('opcodes', []): lines.append( f" {op['index']:>4} {op['opcode_name']:<28} " f"result={_fmt_operand(op['result'])} op1={_fmt_operand(op['op1'])} " f"op2={_fmt_operand(op['op2'])} line={op['lineno']}" ) return '\n'.join(lines) def process_file(path: str): data = open(path, 'rb').read() container = parse_container(data) out_base = os.path.splitext(path)[0] with open(out_base + '.icdump.json', 'w', encoding='utf-8') as fh: json.dump(container, fh, indent=2, default=str) with open(out_base + '.icdump.txt', 'w', encoding='utf-8') as fh: for fn in container['functions']: fh.write(render_text(fn, f"function {fn['function_name']}()") + '\n\n') fh.write(render_text(container['main'], 'MAIN') + '\n') print(f'{path}: ver={container["ver"]:#x} functions={len(container["functions"])} ' f'classes={len(container["classes"])} main_ops={container["main"]["opcode_count"]}') def main(): if len(sys.argv) < 2: print('usage: python sg_bytecode_ir.py [...]', file=sys.stderr) sys.exit(1) for path in sys.argv[1:]: try: process_file(path) except ParseError as e: print(f'{path}: PARSE ERROR: {e}', file=sys.stderr) if __name__ == '__main__': main()