Refactor commonly used utils out of minor_patch_diff

Plus some documentation

.gitignore

@@ -1,5 +1,7 @@
 __pycache__
 /venv
+/ipcs
+/traces
 *.pt
 *.asm
 *.json

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

generate_deep_traces.py

@@ -1,16 +1,12 @@
 import click
 import json
 import pathlib
-import semver
 
-from minor_patch_diff import (
+from analysis_utils import (
-    get_opcode_offset_7_30,
-    get_sem_ver,
-    get_opcode_offset,
-    get_opcode_offset_7_20,
     get_correct_switch,
-    get_zone_proto_down_sig,
+    get_packet_handler_addr,
-    get_opcode_offset_sig,
+    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)
+    target = get_packet_handler_addr(r2, 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)
-
     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
+    ## STEP 2: Get opcode offset
-    p = create_r2_byte_pattern(get_opcode_offset_sig(sem_ver))
+    opcode_offset = get_packet_handler_opcode_offset(r2, exe_file)
-    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)
     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
     )

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)
 

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)
+    target = get_packet_handler_addr(r2, 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)
     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
+    ## STEP 2: Get opcode offset
-    p = create_r2_byte_pattern(get_opcode_offset_sig(sem_ver))
+    opcode_offset = get_packet_handler_opcode_offset(r2, exe_file)
-    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)
     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.

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.