diff --git a/.gitignore b/.gitignore index a92a048..c8863dd 100644 --- a/.gitignore +++ b/.gitignore @@ -1,5 +1,7 @@ __pycache__ /venv +/ipcs +/traces *.pt *.asm *.json diff --git a/analysis_utils.py b/analysis_utils.py new file mode 100644 index 0000000..e5c5f9e --- /dev/null +++ b/analysis_utils.py @@ -0,0 +1,199 @@ +import re +import semver + +from utils import create_r2_byte_pattern + + +def get_sem_ver(exe_file: str) -> str: + """Parses the semver from the exe_file name and returns it. + + Args: + exe_file (str): The path to the executable file. + + Returns: + str: The semantic version string. + """ + res = re.match(r".*ffxiv_dx11\.(\d).(\d)(\d)(\w?)(\d?)\.exe", exe_file) + sem_ver = f"{res.group(1)}.{res.group(2)}.{res.group(3)}" + if res.group(4) != "": + sem_ver = f"{sem_ver}+{res.group(4)}" + return sem_ver + + +def packet_handler_sig(sem_ver: str) -> str: + """Returns the signature for Client::Network::PacketDispatcher_OnReceivePacket + for Zone packets. + + Args: + sem_ver (str): The semantic version string. + """ + if semver.compare(sem_ver, "7.3.0") >= 0: + return "48 89 5C 24 ? 55 56 57 41 54 41 55 41 56 41 57 48 8D AC 24 ? ? ? ? B8 ? ? ? ? E8 ? ? ? ? 48 2B E0 48 8B 05 ? ? ? ? 48 33 C4 48 89 85 ? ? ? ? 45 0F B7" + elif semver.compare(sem_ver, "7.2.0") >= 0: + return "40 55 53 56 57 41 55 41 56 41 57 48 8D AC 24 ? ? ? ? B8 ? ? ? ? E8 ? ? ? ? 48 2B E0 48 8B 05 ? ? ? ? 48 33 C4 48 89 85 ? ? ? ? 45 0F B7 78 ?" + elif semver.compare(sem_ver, "6.4.0") >= 0: + return "40 53 56 48 81 EC ? ? ? ? 48 8B 05 ? ? ? ? 48 33 C4 48 89 44 24 ? 8B F2" + else: + return "48 89 ? 24 ? ? 48 83 EC 50 8B F2 49 8B" + + +def get_packet_handler_addr(r2, exe_file: str) -> str: + """Get the address for the Client::Network::PacketDispatcher_OnReceivePacket + for Zone packets. + + Args: + r2: An instance of r2pipe. + exe_file (str): The path to the executable file. + + Returns: + str: The hex address of the packet dispatch handler. + """ + sem_ver = get_sem_ver(exe_file) + p = create_r2_byte_pattern(packet_handler_sig(sem_ver)) + result = r2.cmd(f"/x {p}").split() # Find byte pattern + if len(result) == 0: + raise ValueError( + "Could not find packet handler address. This could be " + "because of an incorrect signature or a transient Radare failure." + ) + return result[0] + + +def packet_handler_switch_sig(sem_ver: str) -> str: + """Returns the signature for approximately a couple instructions before the + actual zone packet handler switch. + + Args: + sem_ver (str): The semantic version string. + """ + if semver.compare(sem_ver, "7.3.0") > 0: + return "E8 ? ? ? ? 41 83 C7 ? 49 8B FD" + elif semver.compare(sem_ver, "7.3.0") == 0: + if semver.parse(sem_ver)["build"]: + return "E8 ? ? ? ? 41 83 C7 ? 49 8B FD" + return "E8 ? ? ? ? 41 83 C5 ? 49 8B FC" + elif semver.compare(sem_ver, "7.2.0") >= 0: + return "E8 ? ? ? ? 41 83 C7 ? EB 1B" + elif semver.compare(sem_ver, "6.4.0") >= 0: + return "40 53 56 48 81 EC ? ? ? ? 48 8B 05 ? ? ? ? 48 33 C4 48 89 44 24 ? 8B F2" + else: + return "48 89 ? 24 ? ? 48 83 EC 50 8B F2 49 8B" + + +def get_packet_handler_switch_addr(r2, exe_file: str) -> str: + """Gets the address of the approximate position where the zone packet + handler switch is located. + + Args: + r2 - An instance of r2pipe. + exe_file (str): The path to the executable file. + + Returns: + str: The hex address that can be used to find the opcode offset. + """ + sem_ver = get_sem_ver(exe_file) + p = create_r2_byte_pattern(packet_handler_switch_sig(sem_ver)) + result = r2.cmd(f"/x {p}").split() # Find byte pattern + if len(result) == 0: + raise ValueError( + "Could not find packet handler switch address. This could be " + "because of an incorrect signature or a transient Radare failure." + ) + return result[0] + + +def _get_opcode_offset_post_72(r2, reg): + r2.cmd("aeso") # step over call + r2.cmd(f"aer {reg}=0x500") # set reg to some arbitrary number + r2.cmd("aeso") # step + + regs = r2.cmdj("arj") + return 0x500 - regs[reg] + + +def _get_opcode_offset_pre_72(r2): + r2.cmd("aecc") # continue until call + r2.cmd("aer rax=0x0") # set rax to 0 + 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") + return regs["rdx"] - regs["rax"] + + +def get_packet_handler_opcode_offset(r2, exe_file: str): + """Gets the offset that maps the actual opcode to a switch case in the + zone packet handler. + + Args: + r2 - An instance of r2pipe. + exe_file (str): The path to the executable file. + """ + sem_ver = get_sem_ver(exe_file) + # The opcode offset can be found somewhere right before the packet handler + # switch + opcode_offset_target = get_packet_handler_switch_addr(r2, exe_file) + + orig_loc = r2.cmd("s") # Save original spot + r2.cmd(f"s {opcode_offset_target}") + r2.cmd("aei; aeim; aeip") # Initialize ESIL VM, stack, and instruction pointer + + if semver.compare(sem_ver, "7.2.0") >= 0: + offset_reg = "r15" + if ( + semver.compare(sem_ver, "7.3.0") == 0 + and semver.parse(sem_ver)["build"] == "" + ): + offset_reg = "r13" + opcode_offset = _get_opcode_offset_post_72(r2, offset_reg) + else: + opcode_offset = _get_opcode_offset_pre_72(r2) + + # Clear the ESIL environment + r2.cmd("ar0; aeim-; aei-") + r2.cmd(f"s {orig_loc}") # Seek back to original spot + + return opcode_offset + + +def get_correct_switch(approx_ea: str, switch_cases: list): + """Despite seeking directly to the packet switch, we still get multiple + switches from analysis, so we need to find the right one. + + Args: + approx_ea (str): The approximate address of the packet switch. + switch_cases (list): The list of switch cases to search through. + + Returns: + (str, dict): The found switch address and the corresponding switch + """ + switches = dict() + approx_ea = int(approx_ea, 16) + pattern = re.compile(r"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["addr"] + + 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]) + + found_switch = None + longest_switch = dict() + for switch_ea in switches: + int_switch_ea = int(switch_ea, 16) + if int_switch_ea < approx_ea or int_switch_ea > approx_ea + 0x100: + continue + if len(switches[switch_ea].keys()) > len(longest_switch): + found_switch = switch_ea + longest_switch = switches[switch_ea] + + return found_switch, longest_switch diff --git a/generate_deep_traces.py b/generate_deep_traces.py index 8e19b3c..b03701c 100644 --- a/generate_deep_traces.py +++ b/generate_deep_traces.py @@ -1,16 +1,12 @@ import click import json import pathlib -import semver -from minor_patch_diff import ( - get_opcode_offset_7_30, - get_sem_ver, - get_opcode_offset, - get_opcode_offset_7_20, +from analysis_utils import ( get_correct_switch, - get_zone_proto_down_sig, - get_opcode_offset_sig, + get_packet_handler_addr, + get_packet_handler_opcode_offset, + get_packet_handler_switch_addr, ) @@ -137,7 +133,7 @@ def generate_opcodes_db(r2, switch, opcode_offset, fn_graph): def extract_opcode_data(exe_file): - from utils import eprint, create_r2_byte_pattern, sync_r2_output + from utils import eprint, sync_r2_output import r2pipe @@ -146,11 +142,7 @@ def extract_opcode_data(exe_file): sync_r2_output(r2) - sem_ver = get_sem_ver(exe_file) - p = create_r2_byte_pattern(get_zone_proto_down_sig(sem_ver)) - target = r2.cmd(f"/x {p}").split()[0] # Find byte pattern - packet_handler_ea = int(target, 16) - + target = get_packet_handler_addr(r2, exe_file) r2.cmd(f"s {target}") # Seek to target ## STEP 1: Grab switch cases @@ -160,18 +152,8 @@ def extract_opcode_data(exe_file): eprint(f" Loaded switch cases") - ## STEP 2: Grab opcode offset - p = create_r2_byte_pattern(get_opcode_offset_sig(sem_ver)) - opcode_offset_target = r2.cmd(f"/x {p}").split()[0] # Find byte pattern - packet_handler_ea = int(opcode_offset_target, 16) - r2.cmd(f"s {opcode_offset_target}") # Seek to target - - if semver.compare(sem_ver, "7.3.0") >= 0: - opcode_offset = get_opcode_offset_7_30(r2) - elif semver.compare(sem_ver, "7.2.0") >= 0: - opcode_offset = get_opcode_offset_7_20(r2) - else: - opcode_offset = get_opcode_offset(r2) + ## STEP 2: Get opcode offset + opcode_offset = get_packet_handler_opcode_offset(r2, exe_file) eprint(f" Found opcode offset: {opcode_offset}") r2.cmd(f"s {target}") # Seek to original target @@ -180,6 +162,7 @@ def extract_opcode_data(exe_file): fn_graph = r2.cmdj(f"pdrj") ## STEP 4: Process data + packet_handler_ea = get_packet_handler_switch_addr(r2, exe_file) switch_ea, packet_handler_switch = get_correct_switch( packet_handler_ea, switch_cases ) diff --git a/generate_opcodes_file.py b/generate_opcodes_file.py index 339a36d..d0000ee 100644 --- a/generate_opcodes_file.py +++ b/generate_opcodes_file.py @@ -57,7 +57,10 @@ def replace_line_with_new_opcode(line, diff, ver, old_ver): "--reverse", is_flag=True, help="Applies the diff file in the opposite direction" ) @click.option( - "-o", "--output", type=click.Path(exists = False, resolve_path=True), help="Output filename" + "-o", + "--output", + type=click.Path(exists=False, resolve_path=True), + help="Output filename", ) def generate_opcodes_file( old_version_string, new_version_string, diff_file, opcodes_file, reverse, output @@ -87,7 +90,7 @@ def generate_opcodes_file( if output: new_filename = output else: - new_filename = f"Ipcs.{new_version_string}.h" + new_filename = f"ipcs/Ipcs.{new_version_string}.h" with open(new_filename, "w+") as f: f.writelines(queued_lines) diff --git a/minor_patch_diff.py b/minor_patch_diff.py index c6830c4..c98b796 100644 --- a/minor_patch_diff.py +++ b/minor_patch_diff.py @@ -1,140 +1,17 @@ import click import json -import re -import semver + +from analysis_utils import ( + get_correct_switch, + get_packet_handler_addr, + get_packet_handler_opcode_offset, + get_packet_handler_switch_addr, +) fucked_distance = 0xFFFFFFFF max_size_diff = 10 -def get_sem_ver(exe_file: str): - res = re.match(r".*ffxiv_dx11\.(\d).(\d)(\d)(\w?)(\d?)\.exe", exe_file) - sem_ver = f"{res.group(1)}.{res.group(2)}.{res.group(3)}" - if res.group(4) != "": - sem_ver = f"{sem_ver}+{res.group(4)}" - return sem_ver - - -def get_zone_proto_down_sig(sem_ver: str): - if semver.compare(sem_ver, "7.3.0") >= 0: - return "48 89 5C 24 ? 55 56 57 41 54 41 55 41 56 41 57 48 8D AC 24 ? ? ? ? B8 ? ? ? ? E8 ? ? ? ? 48 2B E0 48 8B 05 ? ? ? ? 48 33 C4 48 89 85 ? ? ? ? 45 0F B7 68 ?" - elif semver.compare(sem_ver, "7.2.0") >= 0: - return "40 55 53 56 57 41 55 41 56 41 57 48 8D AC 24 ? ? ? ? B8 ? ? ? ? E8 ? ? ? ? 48 2B E0 48 8B 05 ? ? ? ? 48 33 C4 48 89 85 ? ? ? ? 45 0F B7 78 ?" - elif semver.compare(sem_ver, "6.4.0") >= 0: - return "40 53 56 48 81 EC ? ? ? ? 48 8B 05 ? ? ? ? 48 33 C4 48 89 44 24 ? 8B F2" - else: - return "48 89 ? 24 ? ? 48 83 EC 50 8B F2 49 8B" - - -def get_opcode_offset_sig(sem_ver: str): - if semver.compare(sem_ver, "7.3.0") >= 0: - return "E8 ? ? ? ? 41 83 C5 ? 49 8B FC" - elif semver.compare(sem_ver, "7.2.0") >= 0: - return "E8 ? ? ? ? 41 83 C7 ? EB 1B" - elif semver.compare(sem_ver, "6.4.0") >= 0: - return "40 53 56 48 81 EC ? ? ? ? 48 8B 05 ? ? ? ? 48 33 C4 48 89 44 24 ? 8B F2" - else: - return "48 89 ? 24 ? ? 48 83 EC 50 8B F2 49 8B" - - -def get_opcode_offset_7_30(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 - r2.cmd("aeso") # step over call - r2.cmd("aer r13=0x500") # set r15 to some arbitrary number - r2.cmd("aeso") # step - - regs = r2.cmdj("arj") - opcode_offset = 0x500 - regs["r13"] - - # 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 - - -def get_opcode_offset_7_20(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 - r2.cmd("aeso") # step over call - r2.cmd("aer r15=0x500") # set r15 to some arbitrary number - r2.cmd("aeso") # step - - regs = r2.cmdj("arj") - opcode_offset = 0x500 - regs["r15"] - - # 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 - - -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 - - r2.cmd("aecc") # continue until call - r2.cmd("aer rax=0x0") # set rax to 0 - 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"] - - # 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 - - -def get_correct_switch(approx_ea, switch_cases): - switches = dict() - - pattern = re.compile(r"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["addr"] - - 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]) - - found_switch = None - longest_switch = dict() - for switch_ea in switches: - int_switch_ea = int(switch_ea, 16) - if int_switch_ea < approx_ea or int_switch_ea > approx_ea + 0x100: - continue - if len(switches[switch_ea].keys()) > len(longest_switch): - found_switch = switch_ea - longest_switch = switches[switch_ea] - - return found_switch, longest_switch - - def get_block_sizes(blocks): block_sizes = dict() for block in blocks: @@ -144,7 +21,7 @@ def get_block_sizes(blocks): def generate_opcodes_db(packet_handler_ea, switch, opcode_offset, block_sizes): opcodes_db = dict() - + packet_handler_ea = int(packet_handler_ea, 16) for case_ea, data in switch.items(): resolved_opcodes = [int(opcode) + opcode_offset for opcode in data["opcodes"]] opcodes_db[resolved_opcodes[0]] = { @@ -157,7 +34,7 @@ def generate_opcodes_db(packet_handler_ea, switch, opcode_offset, block_sizes): def extract_opcode_data(exe_file): - from utils import eprint, create_r2_byte_pattern, sync_r2_output + from utils import eprint, sync_r2_output import r2pipe @@ -166,10 +43,7 @@ def extract_opcode_data(exe_file): sync_r2_output(r2) - sem_ver = get_sem_ver(exe_file) - p = create_r2_byte_pattern(get_zone_proto_down_sig(sem_ver)) - target = r2.cmd(f"/x {p}").split()[0] # Find byte pattern - packet_handler_ea = int(target, 16) + target = get_packet_handler_addr(r2, exe_file) r2.cmd(f"s {target}") # Seek to target ## STEP 1: Grab switch cases @@ -179,16 +53,8 @@ def extract_opcode_data(exe_file): eprint(f" Loaded switch cases") - ## STEP 2: Grab opcode offset - p = create_r2_byte_pattern(get_opcode_offset_sig(sem_ver)) - opcode_offset_target = r2.cmd(f"/x {p}").split()[0] # Find byte pattern - packet_handler_ea = int(opcode_offset_target, 16) - r2.cmd(f"s {opcode_offset_target}") # Seek to target - - if semver.compare(sem_ver, "7.2.0") >= 0: - opcode_offset = get_opcode_offset_7_20(r2) - else: - opcode_offset = get_opcode_offset(r2) + ## STEP 2: Get opcode offset + opcode_offset = get_packet_handler_opcode_offset(r2, exe_file) eprint(f" Found opcode offset: {opcode_offset}") r2.cmd(f"s {target}") # Seek to original target @@ -201,6 +67,7 @@ def extract_opcode_data(exe_file): eprint(f" Grabbed blocks from packet handler") ## STEP 4: Process data + packet_handler_ea = get_packet_handler_switch_addr(r2, exe_file) switch_ea, packet_handler_switch = get_correct_switch( packet_handler_ea, switch_cases ) @@ -281,6 +148,9 @@ def find_opcode_matches(old_opcodes_db, new_opcodes_db): ) def minor_patch_diff(old_exe, new_exe): """ + DEPRECATED. Use vtable_diff.py instead. It's a more reliable method of + getting the diff for a minor patch. + Generates an opcode diff file for minor patches (e.g 6.30 => 6.30h). This script outputs to stdout, so pipe it to a json file. diff --git a/vtable_diff.py b/vtable_diff.py index 91c9572..4514a25 100644 --- a/vtable_diff.py +++ b/vtable_diff.py @@ -1,11 +1,28 @@ import click import json -from minor_patch_diff import get_correct_switch +from analysis_utils import get_correct_switch from utils import eprint, create_r2_byte_pattern, sync_r2_output import r2pipe -ON_RECEIVE_PACKET_SIG = "49 8B 40 10 4C 8B 50 38" +PACKET_INTERFACE_DISPATCHER_SIG = "49 8B 40 10 4C 8B 50 38" +""" +Signature for some dispatcher function that delegates packet opcodes to +a ZoneDown packet interface. In IDA it has a length of approximately 0x28D5. +It should just contain a large switch statement with handlers that call a +function pointer and look like this: + +``` +mov rax, [rcx] +lea r9, [r10+10h] +jmp qword ptr [rax+10h] +``` + +In reality this dispatcher actually does nothing, as it delegates to vtable +containing only nullsubs. However, the order of these nullsubs in the vtable +is fairly stable across versions for the ZoneDown opcodes. For minor patches, +they might also work for ZoneUp opcodes. +""" def get_opcode_offset(r2): @@ -33,9 +50,8 @@ def extract_opcode_data(exe_file): sync_r2_output(r2) - p = create_r2_byte_pattern(ON_RECEIVE_PACKET_SIG) + p = create_r2_byte_pattern(PACKET_INTERFACE_DISPATCHER_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 @@ -57,7 +73,7 @@ def extract_opcode_data(exe_file): ## STEP 4: Process data opcodes_db = dict() - switch_ea, packet_handler_switch = get_correct_switch(packet_handler_ea, switch_cases) + switch_ea, packet_handler_switch = get_correct_switch(target, switch_cases) eprint(f" Found switch at {switch_ea}") vtable_offset = 0x10 @@ -113,7 +129,8 @@ def diff_exes(old_exe, new_exe): ) def vtable_diff(old_exe, new_exe): """ - Generates an opcode diff file by comparing vtables. + Generates an opcode diff file by comparing vtables. See comments in file + for a detailed explanation. This script outputs to stdout, so pipe it to a json file.