From e783c72d113a45cf8fe961170d8f41f466455640 Mon Sep 17 00:00:00 2001 From: Flawed <33593723+ff14wed@users.noreply.github.com> Date: Tue, 17 Jan 2023 11:48:11 -0800 Subject: [PATCH] Format with black --- generate_act_format.py | 96 +++++++------- generate_opcodes_file.py | 74 ++++++----- minor_patch_diff.py | 262 +++++++++++++++++++++------------------ utils.py | 47 +++---- 4 files changed, 255 insertions(+), 224 deletions(-) diff --git a/generate_act_format.py b/generate_act_format.py index 0f50f75..8ca4b80 100644 --- a/generate_act_format.py +++ b/generate_act_format.py @@ -3,59 +3,61 @@ import click # Convert opcodes to the ACT expected format desired_names = { - "StatusEffectList": None, - "StatusEffectList2": None, - "StatusEffectList3": None, - "BossStatusEffectList": None, - "Effect": "Ability1", - "AoeEffect8": "Ability8", - "AoeEffect16": "Ability16", - "AoeEffect24": "Ability24", - "AoeEffect32": "Ability32", - "ActorCast": None, - "EffectResult": None, - "EffectResultBasic": None, - "ActorControl": None, - "ActorControlSelf": None, - "ActorControlTarget": None, - "UpdateHpMpTp": None, - "PlayerSpawn": None, - "NpcSpawn": None, - "NpcSpawn2": None, - "ActorMove": None, - "ActorSetPos": None, - "ActorGauge": None, - "PlaceFieldMarkerPreset": "PresetWaymark", - "PlaceFieldMarker": "Waymark", - "SystemLogMessage": None + "StatusEffectList": None, + "StatusEffectList2": None, + "StatusEffectList3": None, + "BossStatusEffectList": None, + "Effect": "Ability1", + "AoeEffect8": "Ability8", + "AoeEffect16": "Ability16", + "AoeEffect24": "Ability24", + "AoeEffect32": "Ability32", + "ActorCast": None, + "EffectResult": None, + "EffectResultBasic": None, + "ActorControl": None, + "ActorControlSelf": None, + "ActorControlTarget": None, + "UpdateHpMpTp": None, + "PlayerSpawn": None, + "NpcSpawn": None, + "NpcSpawn2": None, + "ActorMove": None, + "ActorSetPos": None, + "ActorGauge": None, + "PlaceFieldMarkerPreset": "PresetWaymark", + "PlaceFieldMarker": "Waymark", + "SystemLogMessage": None, } + @click.command() -@click.argument("opcodes_file", type=click.File('r')) +@click.argument("opcodes_file", type=click.File("r")) def generate_act_format(opcodes_file): - opcode_mapping = dict() + opcode_mapping = dict() - for line in opcodes_file.readlines(): - match_groups = re.findall(r'^\s*([^\/].*)=\s*(.*),\s*\/\/.*$', line) - if len(match_groups) != 1: - continue + for line in opcodes_file.readlines(): + match_groups = re.findall(r"^\s*([^\/].*)=\s*(.*),\s*\/\/.*$", line) + if len(match_groups) != 1: + continue - opcode_name = match_groups[0][0].strip() - opcode_val = match_groups[0][1] - if " or " in opcode_val: - opcodes = [int(v, 16) for v in opcode_val.split(" or ")] - else: - opcodes = [int(opcode_val, 16)] - opcode_mapping[opcode_name] = opcodes + opcode_name = match_groups[0][0].strip() + opcode_val = match_groups[0][1] + if " or " in opcode_val: + opcodes = [int(v, 16) for v in opcode_val.split(" or ")] + else: + opcodes = [int(opcode_val, 16)] + opcode_mapping[opcode_name] = opcodes + + for name, desired in desired_names.items(): + if desired == None: + desired = name + opcodes = opcode_mapping[name] + if len(opcodes) == 1: + print(f"{desired}|{opcodes[0]:x}") + elif len(opcodes) > 1: + print(f'{desired}|{[f"{opcode:x}" for opcode in opcodes]}') - for name, desired in desired_names.items(): - if desired == None: - desired = name - opcodes = opcode_mapping[name] - if len(opcodes) == 1: - print(f'{desired}|{opcodes[0]:x}') - elif len(opcodes) > 1: - print(f'{desired}|{[f"{opcode:x}" for opcode in opcodes]}') if __name__ == "__main__": - generate_act_format() \ No newline at end of file + generate_act_format() diff --git a/generate_opcodes_file.py b/generate_opcodes_file.py index 86bbc6c..9a36cc9 100644 --- a/generate_opcodes_file.py +++ b/generate_opcodes_file.py @@ -2,60 +2,68 @@ import json import re import click + def load_diff_file(f, reverse=False): - diff = dict() - diff_json = json.load(f) - for pair in diff_json: - if "old" not in pair or "new" not in pair: - continue - old_key = "new" if reverse else "old" - new_key = "old" if reverse else "new" - for old_opcode in pair[old_key]: - diff[int(old_opcode, 16)] = set((int(new_opcode, 16) for new_opcode in pair[new_key])) - return diff + diff = dict() + diff_json = json.load(f) + for pair in diff_json: + if "old" not in pair or "new" not in pair: + continue + old_key = "new" if reverse else "old" + new_key = "old" if reverse else "new" + for old_opcode in pair[old_key]: + diff[int(old_opcode, 16)] = set( + (int(new_opcode, 16) for new_opcode in pair[new_key]) + ) + return diff + def opcodes_str(opcodes): - if len(opcodes) == 1: - return hex(list(opcodes)[0]) - elif len(opcodes) > 1: - return " or ".join((hex(opcode) for opcode in opcodes)) - else: - return "UNKNOWN" + if len(opcodes) == 1: + return hex(list(opcodes)[0]) + elif len(opcodes) > 1: + return " or ".join((hex(opcode) for opcode in opcodes)) + else: + return "UNKNOWN" + def replace_line_with_new_opcode(line, diff, ver): - match_groups = re.findall(r'^\s*([^\/].*)=\s*(.*),\s*\/\/.*$', line) + match_groups = re.findall(r"^\s*([^\/].*)=\s*(.*),\s*\/\/.*$", line) if len(match_groups) != 1: - return line + return line opcode_name = match_groups[0][0] opcode_val = match_groups[0][1] if " or " in opcode_val: - old_opcode = int(opcode_val.split(" or ")[0], 16) + old_opcode = int(opcode_val.split(" or ")[0], 16) else: - old_opcode = int(opcode_val, 16) + old_opcode = int(opcode_val, 16) if old_opcode in diff: - new_opcodes = diff[old_opcode] - return f"{opcode_name}= {opcodes_str(new_opcodes)}, // updated {ver}\n" + new_opcodes = diff[old_opcode] + return f"{opcode_name}= {opcodes_str(new_opcodes)}, // updated {ver}\n" return f"// {line}" @click.command() @click.argument("new_version_string") -@click.argument("diff_file", type=click.File('r')) -@click.argument("opcodes_file", type=click.File('r')) +@click.argument("diff_file", type=click.File("r")) +@click.argument("opcodes_file", type=click.File("r")) @click.option("--reverse", is_flag=True) def generate_opcodes_file(new_version_string, diff_file, opcodes_file, reverse): - diff = load_diff_file(diff_file, reverse) - queued_lines = [] - for line in opcodes_file.readlines(): - queued_lines.append(replace_line_with_new_opcode(line, diff, new_version_string)) + diff = load_diff_file(diff_file, reverse) + queued_lines = [] + for line in opcodes_file.readlines(): + queued_lines.append( + replace_line_with_new_opcode(line, diff, new_version_string) + ) - new_filename = f"{new_version_string}_opcodes.txt" - with open(new_filename, "w+") as f: - f.writelines(queued_lines) + new_filename = f"{new_version_string}_opcodes.txt" + with open(new_filename, "w+") as f: + f.writelines(queued_lines) + + print("Wrote to", new_filename) - print("Wrote to", new_filename) if __name__ == "__main__": - generate_opcodes_file() \ No newline at end of file + generate_opcodes_file() diff --git a/minor_patch_diff.py b/minor_patch_diff.py index 4361c96..b998337 100644 --- a/minor_patch_diff.py +++ b/minor_patch_diff.py @@ -4,192 +4,210 @@ import re ZONE_PROTO_DOWN_SIG = "48 89 ? 24 ? ? 48 83 EC 50 8B F2 49 8B" -fucked_distance = 0xffffffff +fucked_distance = 0xFFFFFFFF max_size_diff = 10 + def get_opcode_offset(r2): - orig_loc = r2.cmd("s") # Save original spot - r2.cmd("aei") # Initialize ESIL VM - r2.cmd("aeim") # Initialize ESIL VM stack - r2.cmd("aeip") # Initialize ESIL VM IP to curseek + orig_loc = r2.cmd("s") # Save original spot + r2.cmd("aei") # Initialize ESIL VM + r2.cmd("aeim") # Initialize ESIL VM stack + r2.cmd("aeip") # Initialize ESIL VM IP to curseek - r2.cmd("aecc") # continue until call - r2.cmd('"aesue rax,0x0,>"') # continue until rax changes? - r2.cmd("aer rdx=0x200") # set rdx to some arbitrary number - r2.cmd("aeso") # step + r2.cmd("aecc") # continue until call + r2.cmd('"aesue rax,0x0,>"') # continue until rax changes? + r2.cmd("aer rdx=0x200") # set rdx to some arbitrary number + r2.cmd("aeso") # step - regs = r2.cmdj("arj") - opcode_offset = regs["rdx"] - regs["rax"] + regs = r2.cmdj("arj") + opcode_offset = regs["rdx"] - regs["rax"] - # Clear the ESIL environment - r2.cmd('ar0') - r2.cmd('aeim-') - r2.cmd('aei-') - r2.cmd(f"s {orig_loc}") # Seek back to original spot + # Clear the ESIL environment + r2.cmd("ar0") + r2.cmd("aeim-") + r2.cmd("aei-") + r2.cmd(f"s {orig_loc}") # Seek back to original spot + + return opcode_offset - return opcode_offset def get_longest_switch(switch_cases): - switches = dict() + switches = dict() - pattern = re.compile("case\.(0x[0-9a-fA-F]+)\.(\d+)") + pattern = re.compile("case\.(0x[0-9a-fA-F]+)\.(\d+)") - for l in switch_cases: - match = pattern.match(l["name"]) - if match is not None: - switch_ea = match[1] - case_ea = l["offset"] + for l in switch_cases: + match = pattern.match(l["name"]) + if match is not None: + switch_ea = match[1] + case_ea = l["offset"] - if switch_ea not in switches: - switches[switch_ea] = dict() - if case_ea not in switches[switch_ea]: - switches[switch_ea][case_ea] = { - "opcodes": [], - } - switches[switch_ea][case_ea]["opcodes"].append(match[2]) + if switch_ea not in switches: + switches[switch_ea] = dict() + if case_ea not in switches[switch_ea]: + switches[switch_ea][case_ea] = { + "opcodes": [], + } + switches[switch_ea][case_ea]["opcodes"].append(match[2]) - longest_switch = dict() - for switch_ea in switches: - if len(switches[switch_ea].keys()) > len(longest_switch): - longest_switch = switches[switch_ea] + longest_switch = dict() + for switch_ea in switches: + if len(switches[switch_ea].keys()) > len(longest_switch): + longest_switch = switches[switch_ea] + + return longest_switch - return longest_switch def get_block_sizes(blocks): - block_sizes = dict() - for block in blocks: - block_sizes[block["addr"]] = block["size"] - return block_sizes + block_sizes = dict() + for block in blocks: + block_sizes[block["addr"]] = block["size"] + return block_sizes + def generate_opcode_db(packet_handler_ea, switch, opcode_offset, block_sizes): - opcodes_db = dict() + opcodes_db = dict() + + for case_ea, data in switch.items(): + resolved_opcodes = [int(opcode) + opcode_offset for opcode in data["opcodes"]] + opcodes_db[resolved_opcodes[0]] = { + "case_ea": case_ea, + "rel_ea": case_ea - packet_handler_ea, + "opcodes": resolved_opcodes, + "size": block_sizes[case_ea] if case_ea in block_sizes else 0, + } + return opcodes_db - for case_ea, data in switch.items(): - resolved_opcodes = [int(opcode) + opcode_offset for opcode in data["opcodes"]] - opcodes_db[resolved_opcodes[0]] = { - "case_ea": case_ea, - "rel_ea": case_ea - packet_handler_ea, - "opcodes": resolved_opcodes, - "size": block_sizes[case_ea] if case_ea in block_sizes else 0, - } - return opcodes_db def get_opcodes_db(exe_file): - from utils import eprint, create_r2_byte_pattern, sync_r2_output + from utils import eprint, create_r2_byte_pattern, sync_r2_output - import r2pipe + import r2pipe - r2 = r2pipe.open(exe_file, ["-2"]) - eprint(f"Radare loaded {exe_file}") + r2 = r2pipe.open(exe_file, ["-2"]) + eprint(f"Radare loaded {exe_file}") - sync_r2_output(r2) + sync_r2_output(r2) - p = create_r2_byte_pattern(ZONE_PROTO_DOWN_SIG) - target = r2.cmd(f"/x {p}").split()[0] # Find byte pattern - packet_handler_ea = int(target, 16) + p = create_r2_byte_pattern(ZONE_PROTO_DOWN_SIG) + target = r2.cmd(f"/x {p}").split()[0] # Find byte pattern + packet_handler_ea = int(target, 16) - r2.cmd(f"s {target}") # Seek to target + r2.cmd(f"s {target}") # Seek to target - ## STEP 1: Grab switch cases - r2.cmd("f--") # Delete existing flags - r2.cmd("afr") # Analyze function recursively - switch_cases = r2.cmdj(f"fj") + ## STEP 1: Grab switch cases + r2.cmd("f--") # Delete existing flags + r2.cmd("afr") # Analyze function recursively + switch_cases = r2.cmdj(f"fj") - eprint(f" Loaded switch cases") + eprint(f" Loaded switch cases") - ## STEP 2: Grab opcode offset + ## STEP 2: Grab opcode offset - opcode_offset = get_opcode_offset(r2) - eprint(f" Found opcode offset: {opcode_offset}") + opcode_offset = get_opcode_offset(r2) + eprint(f" Found opcode offset: {opcode_offset}") - ## STEP 3: Grab blocks from packet handler - blocks = r2.cmdj("afbj") + ## STEP 3: Grab blocks from packet handler + blocks = r2.cmdj("afbj") - r2.quit() + r2.quit() - eprint(f" Grabbed blocks from packet handler") + eprint(f" Grabbed blocks from packet handler") - ## STEP 4: Process data - packet_handler_switch = get_longest_switch(switch_cases) - block_sizes = get_block_sizes(blocks) - opcode_db = generate_opcode_db(packet_handler_ea, packet_handler_switch, opcode_offset, block_sizes) + ## STEP 4: Process data + packet_handler_switch = get_longest_switch(switch_cases) + block_sizes = get_block_sizes(blocks) + opcode_db = generate_opcode_db( + packet_handler_ea, packet_handler_switch, opcode_offset, block_sizes + ) - eprint(f" Loaded {len(opcode_db)} cases from packet handler") + eprint(f" Loaded {len(opcode_db)} cases from packet handler") + + return opcode_db - return opcode_db def find_closest_rel_ea(opcodes_db, dest): - closest = fucked_distance - closest_opcode = None + closest = fucked_distance + closest_opcode = None - for opcode, case in opcodes_db.items(): - rel_ea = case['rel_ea'] + for opcode, case in opcodes_db.items(): + rel_ea = case["rel_ea"] - num = abs(rel_ea - dest) + num = abs(rel_ea - dest) + + if num < closest: + closest = num + closest_opcode = opcode + return (closest, closest_opcode) - if num < closest: - closest = num - closest_opcode = opcode - return (closest, closest_opcode) def get_opcodes_str(opcodes): - return ', '.join([hex(o) for o in opcodes]) + return ", ".join([hex(o) for o in opcodes]) + def add_match_case(cases, case): - # check if case already exists + # check if case already exists - for c in cases: - if c['rel_ea'] == case['rel_ea']: - return + for c in cases: + if c["rel_ea"] == case["rel_ea"]: + return + + cases.append(case) - cases.append(case) def find_opcode_matches(old_opcodes_db, new_opcodes_db): - matches = [] - new_opcodes = list(new_opcodes_db.keys()) + matches = [] + new_opcodes = list(new_opcodes_db.keys()) - for k, case in enumerate(old_opcodes_db.values()): - old_opcodes = case['opcodes'] + for k, case in enumerate(old_opcodes_db.values()): + old_opcodes = case["opcodes"] - # see if we can get a match for the relative ea first - dist, dist_match_opcode = find_closest_rel_ea(new_opcodes_db, case['rel_ea']) + # see if we can get a match for the relative ea first + dist, dist_match_opcode = find_closest_rel_ea(new_opcodes_db, case["rel_ea"]) - if dist == fucked_distance: - continue + if dist == fucked_distance: + continue - order_match_opcode = new_opcodes[k] + order_match_opcode = new_opcodes[k] - order_match = new_opcodes_db[order_match_opcode] - dist_match = new_opcodes_db[dist_match_opcode] + order_match = new_opcodes_db[order_match_opcode] + dist_match = new_opcodes_db[dist_match_opcode] - size_diff = abs(dist_match['size'] - case['size']) + size_diff = abs(dist_match["size"] - case["size"]) - # see if the rva matches for the cases found by the distance and order - if dist_match['rel_ea'] == order_match['rel_ea'] and size_diff < max_size_diff: - matches.append((old_opcodes, order_match["opcodes"])) + # see if the rva matches for the cases found by the distance and order + if dist_match["rel_ea"] == order_match["rel_ea"] and size_diff < max_size_diff: + matches.append((old_opcodes, order_match["opcodes"])) - return matches + return matches @click.command() -@click.argument("old_exe", type=click.Path(exists=True, dir_okay=False, resolve_path=True)) -@click.argument("new_exe", type=click.Path(exists=True, dir_okay=False, resolve_path=True)) +@click.argument( + "old_exe", type=click.Path(exists=True, dir_okay=False, resolve_path=True) +) +@click.argument( + "new_exe", type=click.Path(exists=True, dir_okay=False, resolve_path=True) +) def minor_patch_diff(old_exe, new_exe): - old_opcodes_db = get_opcodes_db(old_exe) - new_opcodes_db = get_opcodes_db(new_exe) + old_opcodes_db = get_opcodes_db(old_exe) + new_opcodes_db = get_opcodes_db(new_exe) - opcodes_found = find_opcode_matches(old_opcodes_db, new_opcodes_db) - opcodes_object = [] + opcodes_found = find_opcode_matches(old_opcodes_db, new_opcodes_db) + opcodes_object = [] - for k, v in enumerate(opcodes_found): - old, new = v + for k, v in enumerate(opcodes_found): + old, new = v - opcodes_object.append({ - "old": [hex(o) for o in old], - "new": [hex(o) for o in new], - }) + opcodes_object.append( + { + "old": [hex(o) for o in old], + "new": [hex(o) for o in new], + } + ) + + print(json.dumps(opcodes_object, indent=2)) - print(json.dumps(opcodes_object, indent=2)) if __name__ == "__main__": - minor_patch_diff() \ No newline at end of file + minor_patch_diff() diff --git a/utils.py b/utils.py index defc1a8..c035384 100644 --- a/utils.py +++ b/utils.py @@ -1,32 +1,35 @@ import sys import time + def eprint(*args, **kwargs): print(*args, file=sys.stderr, **kwargs) + def create_r2_byte_pattern(sig): - tokens = sig.split() - r2_tokens = [] - for token in tokens: - if token == "?": - r2_tokens.append("..") - else: - r2_tokens.append(token) - return "".join(r2_tokens) + tokens = sig.split() + r2_tokens = [] + for token in tokens: + if token == "?": + r2_tokens.append("..") + else: + r2_tokens.append(token) + return "".join(r2_tokens) + def sync_r2_output(r2): - """ - For some fucking reason r2pipe output gets desynced from the start, - making the result of every command what the previous command should - have returned. + """ + For some fucking reason r2pipe output gets desynced from the start, + making the result of every command what the previous command should + have returned. - Read stuff from the process pipe until it stops being stupid. - """ - for i in range(10): - p = r2.process.stdout.read(1) - if len(p) > 0: - break - time.sleep(1) - output = r2.cmd(f"?vi 123").strip() - if output != "123": - raise Exception("R2 state never got synced") \ No newline at end of file + Read stuff from the process pipe until it stops being stupid. + """ + for i in range(10): + p = r2.process.stdout.read(1) + if len(p) > 0: + break + time.sleep(1) + output = r2.cmd(f"?vi 123").strip() + if output != "123": + raise Exception("R2 state never got synced")