lozy360/opcodediff
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Format with black

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