Update for 7.56 (2026.09.01.0000.0000)

Update for 7.55h2 (2026.08.11.0000.0000)

.github/workflows/minor_patch_diff.yml

@@ -0,0 +1,81 @@
+name: Minor Patch Diff Workflow
+
+on:
+  workflow_dispatch:
+
+jobs:
+  check-and-diff:
+    runs-on: ubuntu-latest
+    permissions:
+      contents: write
+      pull-requests: write
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v6
+
+      - name: Set up Python
+        uses: actions/setup-python@v6
+        with:
+          python-version: '3.12'
+          cache: 'pip'
+
+      - name: Fetch Patch Info and Update Versions
+        id: versions
+        run: python automation/ffxiv_info.py >> $GITHUB_OUTPUT
+
+      - name: Download Latest ffxiv_dx11.exe
+        if: steps.versions.outputs.is_new == 'true'
+        run: python automation/download_exe.py --output latest/
+
+      - name: Download Previous ffxiv_dx11.exe
+        if: steps.versions.outputs.is_new == 'true'
+        run: python automation/download_exe.py --version "${{ steps.versions.outputs.thaliak_version_prev }}" --output previous/
+
+      - name: Install Analysis Tools
+        if: steps.versions.outputs.is_new == 'true'
+        run: |
+          curl -L -o radare2.deb https://github.com/radareorg/radare2/releases/download/6.1.0/radare2_6.1.0_amd64.deb
+          sudo dpkg -i radare2.deb || sudo apt-get install -f -y
+          rm radare2.deb
+
+      - name: Install Python Dependencies
+        if: steps.versions.outputs.is_new == 'true'
+        run: |
+          pip install -r requirements-vtable.txt
+
+      - name: Run Diff
+        if: steps.versions.outputs.is_new == 'true'
+        run: |
+          LATEST_EXE=$(find latest -name "ffxiv_dx11.exe" | head -n 1)
+          PREVIOUS_EXE=$(find previous -name "ffxiv_dx11.exe" | head -n 1)
+
+          cp "$LATEST_EXE" "ffxiv_dx11.${{ steps.versions.outputs.retail_new }}.exe"
+          cp "$PREVIOUS_EXE" "ffxiv_dx11.${{ steps.versions.outputs.retail_prev }}.exe"
+
+          # Run the diffing process and capture JSON output
+          python vtable_diff.py "ffxiv_dx11.${{ steps.versions.outputs.retail_prev }}.exe" "ffxiv_dx11.${{ steps.versions.outputs.retail_new }}.exe" > "diffs/${{ steps.versions.outputs.retail_new }}.diff.json"
+
+      - name: Upload Results
+        if: steps.versions.outputs.is_new == 'true'
+        uses: actions/upload-artifact@v6
+        with:
+          name: diff
+          path: |
+            diffs/${{ steps.versions.outputs.retail_new }}.diff.json
+
+      - name: Create Pull Request
+        if: steps.versions.outputs.is_new == 'true'
+        uses: peter-evans/create-pull-request@v8
+        with:
+          token: ${{ secrets.GITHUB_TOKEN }}
+          commit-message: "Update for ${{ steps.versions.outputs.retail_new }} (${{ steps.versions.outputs.thaliak_version_new }})"
+          title: "Update for ${{ steps.versions.outputs.retail_new }} (${{ steps.versions.outputs.thaliak_version_new }})"
+          body: |
+            This automatic update contains the following changes:
+            - Updated `ffxiv_versions_global.json`
+            - Generated opcode diff for retail version ${{ steps.versions.outputs.retail_new }}
+          branch: "auto-update"
+          branch-suffix: "short-commit-hash"
+          base: "main"
+          delete-branch: true
+

.gitignore

@@ -0,0 +1,9 @@
+__pycache__
+/venv
+/ipcs
+/traces
+*.pt
+*.asm
+*.json
+!diffs/*.json
+*.exe

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2023 Flawed
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

README.md

@@ -0,0 +1,94 @@
+# opcodediff
+
+A set of tools for matching opcodes across different versions of the FFXIV
+binary.
+
+## Prerequisites
+
+1. Ensure you have python == 3.12 installed.
+
+1. Set up a python venv:
+   ```sh
+   python -m venv /path/to/venv/dir
+   ```
+
+1. Activate the venv
+
+   Linux:
+   ```sh
+   source venv/bin/activate # csh or fish variants available as well
+   ```
+
+   Windows:
+   ```
+   venv\Scripts\activate
+   ```
+
+1. Install dependencies
+
+   ```sh
+   pip install -r requirements.txt
+   ```
+
+1. Ensure you have [radare](https://github.com/radareorg/radare2) version 6.x somewhere on
+    your PATH
+
+1. For `generate_similarity_matrix.py`, an NVIDIA GPU and CUDA support are highly recommended,
+   but not required. It might just take twice as long to train.
+
+## Usage
+
+Pass `--help` to any of the scripts for usage.
+
+### Workflow for minor patches
+
+Here we simply match the vtable from the previous version to the new minor patch version,
+assuming there is a 1:1 correspondence between the switch cases at the same offset in the function.
+
+Example:
+```sh
+python vtable_diff.py ffxiv_dx11.7.00h.exe ffxiv_dx11.7.01.exe > 7.01.diff.json
+```
+
+Post-diff processing:
+```sh
+python generate_opcodes_file.py 7.00h 7.01 7.01.diff.json Ipcs.7.00h.h
+python generate_act_format.py Ipcs.7.01.h
+```
+
+Sanity checking with the older method as validation:
+```sh
+python minor_patch_diff.py ffxiv_dx11.7.00h.exe ffxiv_dx11.7.01.exe > 7.01.sanitycheck.json
+python sanity_check.py 7.01.diff.json 7.01.sanitycheck.json
+```
+
+### Workflow for major patches
+
+For major patches, we cannot assume a 1:1 correspondence between vtables as before, since there
+are probably insertions in the new version that are scattered throughout the switch case.
+
+To solve this, we can run a sequence alignment algorithm to match the two vtables.
+
+But first, in order to generate a similarity matrix, we must first generate
+"traces" as signatures for every packet handler.  Then we can plug these traces
+into our language model to generate embeddings for each handler. Then, we can
+simply run cross cosine similarity to match these embeddings to generate our
+similarity matrix.
+
+Finally, we run the [Needleman-Wunsch algorithm](https://en.wikipedia.org/wiki/Needleman%E2%80%93Wunsch_algorithm)
+to generate a global sequence alignment of the two vtables.
+
+Example:
+```sh
+python generate_deep_traces.py ffxiv_dx11.6.58h.exe 6.58h-traces
+python generate_deep_traces.py ffxiv_dx11.7.00.exe 7.00-traces
+python generate_similarity_matrix.py 6.58h-traces 7.00-traces 7.00.similarity.json
+
+python vtable_alignment.py ffxiv_dx11.6.58h.exe ffxiv_dx11.7.00.exe 7.00.similarity.json > 7.00.diff.json
+```
+
+### Post-diff processing
+```sh
+python generate_opcodes_file.py 6.58h 7.00 7.00.diff.json Ipcs.6.58h.h
+python generate_act_format.py Ipcs.7.00.h
+```

analysis_utils.py

@@ -0,0 +1,202 @@
+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 ? ? 49 8B ? 41 81"
+    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.5.0") >= 0:
+        offset_reg = "r13"
+        opcode_offset = _get_opcode_offset_post_72(r2, offset_reg)
+    elif semver.compare(sem_ver, "7.4.0") >= 0:
+        offset_reg = "r15"
+        opcode_offset = _get_opcode_offset_post_72(r2, offset_reg)
+    elif semver.compare(sem_ver, "7.3.8") >= 0:
+        offset_reg = "r13"
+        opcode_offset = _get_opcode_offset_post_72(r2, offset_reg)
+    elif 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)
+    elif semver.compare(sem_ver, "7.2.0") >= 0:
+        offset_reg = "r15"
+        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

asm2vec/LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2021 oalieno
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

asm2vec/README.md

@@ -0,0 +1,9 @@
+# Asm2Vec
+
+This package is just a heavily modified version of
+https://github.com/oalieno/asm2vec-pytorch.
+
+Actually that implementation doesn't even work properly, so I had to make
+corrections according to the [original Asm2Vec
+paper](https://www.computer.org/csdl/proceedings-article/sp/2019/666000a038/19skfc3ZfKo)
+to get things to work right.

asm2vec/__init__.py

@@ -0,0 +1,6 @@
+import importlib
+
+__all__ = ["model", "datatype", "utils"]
+
+for module in __all__:
+    importlib.import_module(f".{module}", "asm2vec")

asm2vec/datatype.py

@@ -0,0 +1,205 @@
+import torch
+import random
+import warnings
+import re
+
+
+class Token:
+    def __init__(self, name, index):
+        self.name = name
+        self.index = index
+        self.count = 1
+
+    def __str__(self):
+        return self.name
+
+
+class Tokens:
+    def __init__(self, name_to_index=None, tokens=None):
+        self.name_to_index = name_to_index or {}
+        self.tokens = tokens or []
+
+    def __getitem__(self, key):
+        if type(key) is str:
+            if self.name_to_index.get(key) is None:
+                warnings.warn("Unknown token in training dataset")
+                return self.tokens[self.name_to_index[""]]
+            return self.tokens[self.name_to_index[key]]
+        elif type(key) is int:
+            return self.tokens[key]
+        else:
+            try:
+                return [self[k] for k in key]
+            except:
+                raise ValueError
+
+    def load_state_dict(self, sd):
+        self.name_to_index = sd["name_to_index"]
+        self.tokens = sd["tokens"]
+
+    def state_dict(self):
+        return {"name_to_index": self.name_to_index, "tokens": self.tokens}
+
+    def size(self):
+        return len(self.tokens)
+
+    def add(self, names):
+        if type(names) is not list:
+            names = [names]
+        for name in names:
+            if name not in self.name_to_index:
+                token = Token(name, len(self.tokens))
+                self.name_to_index[name] = token.index
+                self.tokens.append(token)
+            else:
+                self.tokens[self.name_to_index[name]].count += 1
+
+    def update(self, tokens_new):
+        for token in tokens_new:
+            if token.name not in self.name_to_index:
+                token.index = len(self.tokens)
+                self.name_to_index[token.name] = token.index
+                self.tokens.append(token)
+            else:
+                self.tokens[self.name_to_index[token.name]].count += token.count
+
+    def precompute_weights(self, pos):
+        """
+        This process actually takes a long time due to the size of the weights,
+        so precompute them according to the shape of the dataset.
+        """
+        from tqdm import tqdm
+
+        total = sum([token.count for token in self.tokens])
+        token_weights = torch.zeros(len(self.tokens))
+        for token in self.tokens:
+            token_weights[token.index] = (token.count / total) ** 0.75
+        weights = token_weights.repeat(pos.shape[0], 1)
+        for i in tqdm(range(pos.shape[0])):
+            for taken in pos[i]:
+                weights[i][taken] = 0
+        return weights
+
+    def sample(self, all_weights, batch_indices, num=5):
+        weights = all_weights[batch_indices]
+        return torch.multinomial(weights, num, replacement=False)
+
+
+class Function:
+    def __init__(self, insts, blocks, meta):
+        self.insts = insts
+        self.blocks = blocks
+        self.meta = meta
+
+    @classmethod
+    def load(cls, text):
+        """
+        gcc -S format compatiable
+        """
+        label, labels, insts, blocks, meta = None, {}, [], [], {}
+        for line in text.strip("\n").split("\n"):
+            if line[0] in [" ", "\t"]:
+                line = line.strip()
+                # meta data
+                if line[0] == ".":
+                    key, _, value = line[1:].strip().partition(" ")
+                    meta[key] = value
+                # instruction
+                else:
+                    inst = Instruction.load(line)
+                    insts.append(inst)
+                    if len(blocks) == 0 or blocks[-1].end():
+                        blocks.append(BasicBlock())
+                        # link prev and next block
+                        if len(blocks) > 1:
+                            blocks[-2].successors.add(blocks[-1])
+                    if label:
+                        labels[label], label = blocks[-1], None
+                    blocks[-1].add(inst)
+            # label
+            else:
+                label = line.partition(":")[0]
+        # link label
+        for block in blocks:
+            inst = block.insts[-1]
+            if inst.is_jmp() and labels.get(inst.args[0]):
+                block.successors.add(labels[inst.args[0]])
+        # replace label with CONST
+        for inst in insts:
+            for i, arg in enumerate(inst.args):
+                if labels.get(arg):
+                    inst.args[i] = "CONST"
+        return cls(insts, blocks, meta)
+
+    def __hash__(self):
+        return hash("\n".join((str(inst) for inst in self.insts)))
+
+    def __eq__(self, other):
+        if isinstance(other, Function):
+            a = "\n".join((str(inst) for inst in self.insts))
+            b = "\n".join((str(inst) for inst in other.insts))
+            return a == b
+        return False
+
+    def tokens(self):
+        return [token for inst in self.insts for token in inst.tokens()]
+
+    def random_walk(self, num=3):
+        return [self._random_walk() for _ in range(num)]
+
+    def _random_walk(self):
+        current, visited, seq = self.blocks[0], [], []
+        while current not in visited:
+            visited.append(current)
+            seq += current.insts
+            # no following block / hit return
+            if len(current.successors) == 0 or current.insts[-1].op == "ret":
+                break
+            current = random.choice(list(current.successors))
+        return seq
+
+
+class BasicBlock:
+    def __init__(self):
+        self.insts = []
+        self.successors = set()
+
+    def add(self, inst):
+        self.insts.append(inst)
+
+    def end(self):
+        inst = self.insts[-1]
+        return inst.is_jmp() or inst.op == "ret"
+
+
+class Instruction:
+    def __init__(self, op, args):
+        self.op = op
+        self.args = args
+
+    def __str__(self):
+        return f'{self.op} {", ".join([str(arg) for arg in self.args if str(arg)])}'
+
+    @classmethod
+    def load(cls, text):
+        text = text.strip().strip("bnd").strip()  # get rid of BND prefix
+        text = text.replace(" - ", " + ")
+        text = re.sub(r"0x[0-9a-f]+", "CONST", text)
+        text = re.sub(r"\*[0-9]", "*CONST", text)
+        text = re.sub(r" [0-9]", " CONST", text)
+        op, _, args = text.strip().partition(" ")
+        if args:
+            args = [arg.strip() for arg in args.split(",")]
+        else:
+            args = []
+        args = (args + ["", ""])[:2]
+        return cls(op, args)
+
+    def tokens(self):
+        return [self.op] + self.args
+
+    def is_jmp(self):
+        return "jmp" in self.op or self.op[0] == "j"
+
+    def is_call(self):
+        return self.op == "call"

asm2vec/model.py

@@ -0,0 +1,88 @@
+import torch
+import torch.nn as nn
+
+bce, sigmoid, softmax = nn.BCELoss(), nn.Sigmoid(), nn.Softmax(dim=1)
+
+# TODO: Document how this shit works because wtf
+class ASM2VEC(nn.Module):
+    def __init__(self, vocab_size, function_size, embedding_size):
+        super(ASM2VEC, self).__init__()
+        # Dictionary of the token embeddings: v_t
+        self.embeddings = nn.Embedding(
+            vocab_size,
+            embedding_size,
+            _weight=(torch.rand(vocab_size, embedding_size) - 0.5) / embedding_size / 2,
+        )
+        # Dictionary of the function embeddings, \theta_f_s
+        self.embeddings_f = nn.Embedding(
+            function_size,
+            2 * embedding_size,
+            _weight=(torch.rand(function_size, 2 * embedding_size) - 0.5)
+            / embedding_size
+            / 2,
+        )
+        # Dictionary of outputs: Transposed v'_t
+        self.embeddings_r = nn.Embedding(
+            vocab_size,
+            2 * embedding_size,
+            _weight=torch.zeros(vocab_size, 2 * embedding_size),
+        )
+        # Where the old embeddings are stored once the training step is done
+        self.old_embeddings_f = None
+
+    def init_estimation_mode(self, function_size_new):
+        device = self.embeddings.weight.device
+        embedding_size = self.embeddings.embedding_dim
+
+        embedding_weights = self.embeddings.weight
+        self.embeddings = nn.Embedding.from_pretrained(embedding_weights)
+
+        output_weights = self.embeddings_r.weight
+        self.embeddings_r = nn.Embedding.from_pretrained(output_weights)
+
+        self.old_embeddings_f = self.embeddings_f
+        self.embeddings_f = nn.Embedding(
+            function_size_new,
+            2 * embedding_size,
+            _weight=(
+                (torch.rand(function_size_new, 2 * embedding_size) - 0.5)
+                / embedding_size
+                / 2
+            ).to(device),
+        )
+
+    def v(self, inp):
+        # Retrieve the embeddings for all the context tokens
+        e = self.embeddings(inp[:, 1:])
+        # Retrieve the embedding for the function, \theta_f_s
+        v_f = self.embeddings_f(inp[:, 0])
+        # Calculate CT(in_(j-1))
+        v_prev = torch.cat([e[:, 0], (e[:, 1] + e[:, 2]) / 2], dim=1)
+        # Calculate CT(in_(j+1))
+        v_next = torch.cat([e[:, 3], (e[:, 4] + e[:, 5]) / 2], dim=1)
+        # delta(in_j, f_s) = 1/3 * (\theta_f_s + CT(in_(j-1)) + CT(in_(j+1)))
+        v = ((v_f + v_prev + v_next) / 3).unsqueeze(2)
+        return v
+
+    def forward(self, inp, pos, neg):
+        device, batch_size = inp.device, inp.shape[0]
+        v = self.v(inp)
+        # negative sampling loss
+        pred = torch.bmm(self.embeddings_r(torch.cat([pos, neg], dim=1)), v).squeeze()
+        label = torch.cat(
+            [torch.ones(batch_size, 1), torch.zeros(batch_size, neg.shape[1])], dim=1
+        ).to(device)
+        return bce(sigmoid(pred), label)
+
+    def predict(self, inp, pos):
+        device, batch_size = inp.device, inp.shape[0]
+        v = self.v(inp)
+        probs = torch.bmm(
+            self.embeddings_r(
+                torch.arange(self.embeddings_r.num_embeddings)
+                .repeat(batch_size, 1)
+                .to(device)
+            ),
+            v,
+        ).squeeze(dim=2)
+        return softmax(probs)

asm2vec/utils.py

@@ -0,0 +1,348 @@
+import os
+import time
+import torch
+from torch.utils.data import DataLoader, Dataset
+from pathlib import Path
+from .datatype import Tokens, Function, Instruction
+from .model import ASM2VEC
+
+import re
+import json
+
+CONSTANTS_RE = re.compile(r"(-? 0x[0-9a-f]+)|\*([0-9])| ([0-9])")
+PACKET_SIZE_HINT_RE = re.compile(r"mov qword \[rsp \+ 0x20\], (0x[0-9a-f]+)")
+
+
+class TraceData:
+    """
+    A class that stores information about traces read from file.
+
+    Terminology:
+        Trace/Function:
+            The text produced by the generate_deep_traces.py script for each
+            opcode switch case.
+
+        Token:
+            A numerical representation of the each symbol that appears in the
+            trace. The tokens dictionary should be shared across all traces to
+            ensure the same representation matches.
+
+        fn_idx:
+            Index of the Function. This is necessary to index into the
+            training model embeddings. Different switch cases may share
+            the same fn_idx because they may be textually identical but
+            have different constants.
+
+        Opcode Set/opcode_set:
+            A set of opcodes that a single switch case covers.
+
+        Pointer Opcode/ptr_opcode:
+            A single opcode that represents the opcode set.
+
+        Constants vector/constants_vector:
+            Since the training process removes constants from the input text,
+            the constants vector is a "signature" generated by looking at
+            constants used in the trace.
+    """
+
+    def __init__(self, tokens):
+        self.tokens = tokens
+
+        self.__traces = dict()
+        """
+        Maps trace to trace idx
+        """
+
+        self.opcodes = dict()
+        """
+        maps ptr_opcode => { fn_idx, constants_vector, [opcodes...] }.
+        See class docstring for more details.
+        """
+
+        self.__opcode_sets = dict()
+        """
+        maps ptr_opcode => [opcodes...]
+        """
+
+    @property
+    def traces(self):
+        """
+        A list of parsed traces in the TraceData.
+
+        See class docstring for more details.
+        """
+        return list(self.__traces.keys())
+
+    def __process_trace(self, ptr_opcode, text):
+        fn = Function.load(text)
+        if fn in self.__traces:
+            fn_idx = self.__traces[fn]
+        else:
+            fn_idx = len(self.__traces)
+            self.__traces[fn] = fn_idx
+            self.tokens.add(fn.tokens())
+
+        self.opcodes[ptr_opcode] = {
+            "fn_idx": fn_idx,
+            "constants_vector": self.__get_constants_vector(text),
+            "packet_size_hint": self.__get_packet_size_hint(text),
+            "opcodes": self.__opcode_sets[ptr_opcode],
+        }
+
+    def __process_opcode_sets(self, opcode_sets_file):
+        with open(opcode_sets_file) as f:
+            data = json.load(f)
+            self.__opcode_sets = {int(op): ops for op, ops in data.items()}
+
+    @staticmethod
+    def __read_trace_from_file(f):
+        lines = f.readlines()
+        normalized_lines = []
+        for line in lines:
+            normalized_lines.append(" " + line)
+        # cap lines at 200 since Asm2Vec performance goes way down if
+        # the text is too long
+        return "".join(normalized_lines[:200])
+
+    @staticmethod
+    def __get_constants_vector(trace):
+        constants = []
+        for line in trace.strip("\n").split("\n"):
+            match = CONSTANTS_RE.search(line)
+            if not match or not match.lastindex:
+                continue
+
+            const_str = match.group(match.lastindex)
+            const_str = "".join(const_str.split())
+            const = int(const_str, 16)
+
+            # We only really care about constants less than 10000
+            if abs(const) < 10000:
+                constants.append(const)
+        return constants
+
+    @staticmethod
+    def __get_packet_size_hint(trace):
+        """
+        Dumb way of getting the packet size hint from whatever handlers
+        call them
+        """
+        call0_lines = []
+        call0_found = False
+
+        for line in trace.strip("\n").split("\n"):
+            if not call0_found:
+                if "CALL0" in line:
+                    call0_found = True
+                else:
+                    continue
+            if "CALL0_END" in line:
+                break
+            else:
+                call0_lines.append(line)
+
+        for line in call0_lines:
+            match = PACKET_SIZE_HINT_RE.search(line)
+            if match and match.lastindex:
+                const_str = match.group(match.lastindex)
+                const_str = "".join(const_str.split())
+                const = int(const_str, 16)
+                return const
+
+        # Couldn't find anything, so just return 0
+        return 0
+
+    @staticmethod
+    def load_data(paths, tokens):
+        """Reads traces from paths and returns TraceData for that path."""
+        if type(paths) is not list:
+            paths = [paths]
+
+        filenames = []
+        for path in paths:
+            if os.path.isdir(path):
+                filenames += [
+                    Path(path) / filename
+                    for filename in sorted(os.listdir(path))
+                    if os.path.isfile(Path(path) / filename)
+                ]
+            else:
+                filenames += [Path(path)]
+
+        trace_data = TraceData(tokens)
+        # Process opcode_sets.json first, save traces for later
+
+        trace_files = dict()
+        for filepath in filenames:
+            filename = os.path.basename(filepath)
+            file_split = os.path.splitext(filename)
+            file_ext = file_split[-1]
+            if file_ext == ".json" and filename == "opcode_sets.json":
+                trace_data.__process_opcode_sets(filepath)
+            elif file_ext == ".asm":
+                ptr_opcode = int(file_split[0], base=16)
+                trace_files[ptr_opcode] = filepath
+
+        for ptr_opcode, trace_file in trace_files.items():
+            with open(trace_file) as f:
+                text = trace_data.__read_trace_from_file(f)
+                trace_data.__process_trace(ptr_opcode, text)
+
+        return trace_data
+
+
+class AsmDataset(Dataset):
+    def __init__(self, x, y):
+        self.x = x
+        self.y = y
+
+    def __len__(self):
+        return len(self.x)
+
+    def __getitem__(self, index):
+        return index, self.x[index], self.y[index]
+
+
+def preprocess(functions, tokens):
+    x, y = [], []
+    for i, fn in enumerate(functions):
+        for j in range(1, len(fn.insts) - 1):
+            x.append(
+                [i]
+                + [
+                    tokens[token].index
+                    for token in fn.insts[j - 1].tokens() + fn.insts[j + 1].tokens()
+                ]
+            )
+            y.append([tokens[token].index for token in fn.insts[j].tokens()])
+    return torch.tensor(x), torch.tensor(y)
+
+
+def train(
+    trace_data: TraceData,
+    model=None,
+    embedding_size=100,
+    batch_size=1024,
+    epochs=10,
+    neg_sample_num=25,
+    calc_acc=False,
+    device="cpu",
+    mode="train",
+    callback=None,
+    learning_rate=0.02,
+):
+    """Trains the model on the provided trace data."""
+    functions = trace_data.traces
+    tokens = trace_data.tokens
+
+    if mode == "train":
+        if model is None:
+            model = ASM2VEC(
+                tokens.size(),
+                function_size=len(functions),
+                embedding_size=embedding_size,
+            ).to(device)
+        optimizer = torch.optim.Adam(model.parameters(), lr=learning_rate)
+    elif mode == "test":
+        if model is None:
+            raise ValueError("test mode requires a pretrained model")
+        optimizer = torch.optim.Adam(model.embeddings_f.parameters(), lr=learning_rate)
+    else:
+        raise ValueError("Unknown mode")
+
+    # Precompute the token weights so that they are cached for later use
+    inp, pos = preprocess(functions, tokens)
+    token_weights = tokens.precompute_weights(pos)
+    loader = DataLoader(AsmDataset(inp, pos), batch_size=batch_size, shuffle=True)
+    for epoch in range(epochs):
+        start = time.time()
+        loss_sum, loss_count, accs = 0.0, 0, []
+
+        model.train()
+        """
+        Recall that the model consumes context tokens as input and uses
+        embeddings for each output token as an output layer of the network. The
+        model should output a high score for the token that correctly matches
+        with the given context, and a low score for tokens that don't match the
+        given context.
+
+        Our dataloader outputs samples in the form of
+        batch_size x (batch_index, input_context, correct_token).
+        """
+        for i, (batch_indices, inp, pos) in enumerate(loader):
+            for j in range(pos.shape[1]):
+                # Sample tokens that are not the positive token.
+                neg = tokens.sample(token_weights, batch_indices, neg_sample_num)
+                pos_token = torch.unsqueeze(pos[:, j], 1)
+                loss = model(inp.to(device), pos_token.to(device), neg.to(device))
+                loss_sum, loss_count = loss_sum + loss, loss_count + 1
+
+                optimizer.zero_grad()
+                loss.backward()
+                optimizer.step()
+
+                if i == 0 and calc_acc:
+                    probs = model.predict(inp.to(device), pos.to(device))
+                    accs.append(accuracy(pos, probs))
+
+        if callback:
+            callback(
+                {
+                    "model": model,
+                    "tokens": tokens,
+                    "epoch": epoch,
+                    "time": time.time() - start,
+                    "loss": loss_sum / loss_count,
+                    "accuracy": torch.tensor(accs).mean() if calc_acc else None,
+                }
+            )
+
+    return model
+
+
+def save_model(path, model, tokens):
+    torch.save(
+        {
+            "model_params": (
+                model.embeddings.num_embeddings,
+                model.embeddings_f.num_embeddings,
+                model.embeddings.embedding_dim,
+            ),
+            "model": model.state_dict(),
+            "tokens": tokens.state_dict(),
+        },
+        path,
+    )
+
+
+def load_model(path, device="cpu"):
+    checkpoint = torch.load(path, map_location=device)
+    tokens = Tokens()
+    tokens.load_state_dict(checkpoint["tokens"])
+    model = ASM2VEC(*checkpoint["model_params"])
+    model.load_state_dict(checkpoint["model"])
+    model = model.to(device)
+    return model, tokens
+
+
+def accuracy(y, probs):
+    return torch.mean(
+        torch.tensor([torch.sum(probs[i][yi].detach()) for i, yi in enumerate(y)])
+    )
+
+
+def cosine_similarities(model):
+    """
+    Reads the old and new embeddings from the model and returns the pairwise
+    cosine similarities between these embeddings.
+    """
+    old_f = model.to("cpu").old_embeddings_f
+    new_f = model.to("cpu").embeddings_f
+    v_old = old_f(torch.tensor([i for i in range(old_f.num_embeddings)]))
+    v_new = new_f(torch.tensor([i for i in range(new_f.num_embeddings)]))
+
+    cs_matrix = torch.nn.functional.cosine_similarity(
+        v_old[:, :, None], v_new.t()[None, :, :]
+    )
+
+    return cs_matrix.detach().numpy()

automation/download_exe.py

@@ -0,0 +1,391 @@
+#!/usr/bin/env python3
+"""
+download_exe.py - Extract ffxiv_dx11.exe from an FFXIV ZiPatch file by streaming.
+
+Does not require a CLUT file. Queries the Thaliak REST API for the patch URL,
+then streams and parses the ZiPatch format to assemble ffxiv_dx11.exe on the fly.
+
+Usage:
+    python automation/download_exe.py --output latest/
+    python automation/download_exe.py --version D2025.01.14.0000.0000 --output previous/
+"""
+
+import os
+import sys
+import struct
+import zlib
+import argparse
+import urllib.request
+import json
+from typing import Optional
+
+# ---------------------------------------------------------------------------
+# Constants
+# ---------------------------------------------------------------------------
+
+THALIAK_REPO = "4e9a232b"
+THALIAK_API = f"https://thaliak.xiv.dev/api/v2beta/repositories/{THALIAK_REPO}/patches"
+
+# ZiPatch magic: \x91ZIPATCH\r\n\x1A\n
+ZIPATCH_MAGIC = bytes(
+    [0x91, 0x5A, 0x49, 0x50, 0x41, 0x54, 0x43, 0x48, 0x0D, 0x0A, 0x1A, 0x0A]
+)
+
+TARGET_FILE = "ffxiv_dx11.exe"
+
+# SqpkCompressedBlock: compressed_size == 32000 means the block is stored raw
+SQPK_BLOCK_UNCOMPRESSED = 32000
+
+# SqpkFile operation codes
+OP_ADD_FILE = ord("A")
+OP_REMOVE_ALL = ord("R")
+OP_DELETE_FILE = ord("D")
+OP_MAKE_DIR = ord("M")
+
+
+# ---------------------------------------------------------------------------
+# Thaliak helpers
+# ---------------------------------------------------------------------------
+
+
+def _fetch_json(url: str) -> dict:
+    req = urllib.request.Request(
+        url, headers={"User-Agent": "ffxiv-exe-downloader/1.0"}
+    )
+    with urllib.request.urlopen(req) as resp:
+        return json.loads(resp.read().decode("utf-8"))
+
+
+def get_patch_url(version: Optional[str]) -> tuple[str, str]:
+    """Return (version_string, remote_url) for *version* or the latest patch."""
+    data = _fetch_json(THALIAK_API)
+    patches = data.get("patches", [])
+    if not patches:
+        raise RuntimeError("No patches returned from Thaliak API")
+
+    if version is None:
+        return patches[-1]["version_string"], patches[-1]["remote_url"]
+
+    # The version coming from the GHA output may or may not have a D/H prefix.
+    for p in reversed(patches):
+        vs = p["version_string"]
+        bare = vs[1:] if vs[:1] in ("D", "H") else vs
+        if vs == version or bare == version:
+            return vs, p["remote_url"]
+
+    raise RuntimeError(f"Version {version!r} not found in Thaliak patches")
+
+
+# ---------------------------------------------------------------------------
+# Streaming reader
+# ---------------------------------------------------------------------------
+
+
+class StreamReader:
+    """
+    Wraps an HTTP response object and exposes read_exact / skip helpers.
+    Reads are buffered internally so we never buffer more than needed.
+    """
+
+    CHUNK = 65_536  # 64 KiB read chunks for internal buffering
+
+    def __init__(self, response) -> None:
+        self._resp = response
+        self._buf = bytearray()
+        self._eof = False
+
+    def _fill(self, needed: int) -> None:
+        while len(self._buf) < needed and not self._eof:
+            raw = self._resp.read(self.CHUNK)
+            if not raw:
+                self._eof = True
+                break
+            self._buf.extend(raw)
+
+    def read_exact(self, n: int) -> bytes:
+        if n == 0:
+            return b""
+        self._fill(n)
+        if len(self._buf) < n:
+            raise EOFError(
+                f"Stream ended prematurely: needed {n}, got {len(self._buf)}"
+            )
+        data = bytes(self._buf[:n])
+        del self._buf[:n]
+        return data
+
+    def skip(self, n: int) -> None:
+        """Discard n bytes without buffering them all at once."""
+        while n > 0:
+            chunk = min(n, self.CHUNK)
+            self._fill(chunk)
+            take = min(chunk, len(self._buf))
+            if take == 0:
+                raise EOFError("Stream ended prematurely during skip")
+            del self._buf[:take]
+            n -= take
+
+
+# ---------------------------------------------------------------------------
+# ZiPatch parsing helpers
+# ---------------------------------------------------------------------------
+
+
+def _align_up(value: int, alignment: int) -> int:
+    return (value + alignment - 1) & ~(alignment - 1)
+
+
+def _decompress_raw_deflate(data: bytes) -> bytes:
+    """Decompress raw DEFLATE (no zlib/gzip header, wbits=-15)."""
+    return zlib.decompress(data, wbits=-15)
+
+
+def _parse_sqpk_file_body(body: bytes) -> Optional[tuple]:
+    """
+    Parse an SQPK command-'F' body.
+
+    Returns (operation, file_offset, file_size, path, blocks_start_offset) or None on error.
+    `blocks_start_offset` is the index in `body` where compressed blocks begin.
+    """
+    # body layout (after the 4-byte inner_size + 1-byte command already consumed):
+    #   [1: operation] [2: pad] [8: file_offset BE] [8: file_size BE]
+    #   [4: path_len BE] [2: expansion_id BE] [2: pad] [path_len: path]
+    #   ... compressed blocks ...
+    HEADER = 1 + 2 + 8 + 8 + 4 + 2 + 2  # = 27 bytes
+    if len(body) < HEADER:
+        return None
+
+    offset = 0
+    operation = body[offset]
+    offset += 1 + 2  # op + 2 pad bytes
+
+    (file_offset,) = struct.unpack_from(">q", body, offset)
+    offset += 8
+    (file_size,) = struct.unpack_from(">q", body, offset)
+    offset += 8
+    (path_len,) = struct.unpack_from(">i", body, offset)
+    offset += 4
+    offset += 4  # expansion_id (2) + pad (2)
+
+    if path_len < 0 or offset + path_len > len(body):
+        return None
+
+    path = (
+        body[offset : offset + path_len]
+        .decode("utf-8", errors="replace")
+        .rstrip("\x00")
+    )
+    offset += path_len
+
+    return operation, file_offset, file_size, path, offset
+
+
+def _parse_compressed_blocks(
+    body: bytes, blocks_start: int, file_offset: int, out_buf: bytearray
+) -> int:
+    """
+    Parse SqpkCompressedBlock entries from `body[blocks_start:]`, decompressing each
+    into `out_buf` starting at `file_offset`.
+
+    Returns the final file offset after all blocks (i.e. file_offset + total decompressed bytes).
+    """
+    pos = blocks_start
+    current_file_off = file_offset
+
+    while pos < len(body):
+        blk_start = pos
+
+        if pos + 16 > len(body):
+            break  # not enough data for a block header
+
+        (header_size,) = struct.unpack_from("<i", body, pos)
+        pos += 4
+        pos += 4  # pad (uint32)
+        (compressed_size,) = struct.unpack_from("<i", body, pos)
+        pos += 4
+        (data_size,) = struct.unpack_from("<i", body, pos)
+        pos += 4
+
+        # Seek to blk_start + header_size (skip any extra header bytes beyond the 16 we read)
+        pos = blk_start + header_size
+
+        is_compressed = compressed_size != SQPK_BLOCK_UNCOMPRESSED
+
+        if is_compressed:
+            if pos + compressed_size > len(body):
+                break
+            raw = body[pos : pos + compressed_size]
+            decompressed = _decompress_raw_deflate(raw)
+            pos += compressed_size
+        else:
+            if pos + data_size > len(body):
+                break
+            decompressed = body[pos : pos + data_size]
+            pos += data_size
+
+        # Write decompressed bytes into out_buf at current_file_off,
+        # growing the buffer as needed (multiple SqpkFile chunks cover
+        # sequential ranges; each chunk's file_size is only its own share).
+        end = current_file_off + len(decompressed)
+        if end > len(out_buf):
+            out_buf.extend(bytes(end - len(out_buf)))
+        out_buf[current_file_off:end] = decompressed
+
+        current_file_off += data_size  # advance by the *uncompressed* size
+
+        # Align to next 128-byte boundary from block start
+        consumed = pos - blk_start
+        aligned = _align_up(consumed, 128)
+        pos = blk_start + aligned
+
+    return current_file_off
+
+
+# ---------------------------------------------------------------------------
+# Main extraction logic
+# ---------------------------------------------------------------------------
+
+
+def extract_exe(url: str, output_path: str) -> bool:
+    """
+    Stream the ZiPatch at *url*, extract ffxiv_dx11.exe, write to *output_path*.
+    Returns True on success.
+    """
+    print(f"Streaming patch: {url}", file=sys.stderr)
+
+    req = urllib.request.Request(
+        url, headers={"User-Agent": "ffxiv-exe-downloader/1.0"}
+    )
+    with urllib.request.urlopen(req) as resp:
+        reader = StreamReader(resp)
+
+        # --- Verify ZiPatch magic (12 bytes) ---
+        magic = reader.read_exact(12)
+        if magic != ZIPATCH_MAGIC:
+            raise RuntimeError(f"Not a valid ZiPatch file (bad magic: {magic.hex()})")
+
+        out_buf: Optional[bytearray] = None
+
+        while True:
+            # Each chunk: [4: body_size BE] [4: fourcc] [body_size: body] [4: CRC32]
+            header = reader.read_exact(4)
+            (body_size,) = struct.unpack(">I", header)
+            fourcc_bytes = reader.read_exact(4)
+            fourcc = fourcc_bytes.decode("ascii", errors="replace")
+
+            if fourcc == "EOF_":
+                # EndOfFile chunk – we are done
+                reader.skip(body_size + 4)  # body + CRC
+                break
+
+            if fourcc != "SQPK":
+                # Skip non-SQPK chunks (FileHeader, ApplyOption, etc.)
+                reader.skip(body_size + 4)
+                continue
+
+            # --- SQPK chunk ---
+            # First 5 bytes: inner_size (4, BE) + command (1)
+            if body_size < 5:
+                reader.skip(body_size + 4)
+                continue
+
+            preamble = reader.read_exact(5)
+            command = chr(preamble[4])
+
+            if command != "F":
+                # Not a file command – skip the rest
+                reader.skip(body_size - 5 + 4)
+                continue
+
+            # command == 'F' (SqpkFile) – load the full body to parse file path and blocks
+            rest = reader.read_exact(body_size - 5)
+            reader.skip(4)  # CRC
+
+            parsed = _parse_sqpk_file_body(rest)
+            if parsed is None:
+                continue
+
+            operation, file_offset, _chunk_file_size, path, blocks_start = parsed
+
+            # Check whether this chunk targets our exe (case-insensitive suffix match)
+            is_target = path.lower() == TARGET_FILE.lower() or path.lower().endswith(
+                "/" + TARGET_FILE.lower()
+            )
+
+            if not is_target:
+                continue
+
+            if operation == OP_REMOVE_ALL:
+                out_buf = None
+                print(f"  RemoveAll on {path}", file=sys.stderr)
+                continue
+
+            if operation != OP_ADD_FILE:
+                continue
+
+            # AddFile: reset buffer on first write (file_offset == 0 means truncate+rewrite).
+            # Do NOT pre-allocate from chunk_file_size — that's only this chunk's share;
+            # the full exe spans many SqpkFile chunks at increasing file_offset values.
+            if file_offset == 0:
+                out_buf = bytearray()
+                print(f"  Found {path}", file=sys.stderr)
+
+            if out_buf is None:
+                # Shouldn't happen for a well-formed patch
+                continue
+
+            _parse_compressed_blocks(rest, blocks_start, file_offset, out_buf)
+
+        if out_buf is None:
+            print(f"ERROR: {TARGET_FILE} not found in patch!", file=sys.stderr)
+            return False
+
+        os.makedirs(os.path.dirname(os.path.abspath(output_path)), exist_ok=True)
+        with open(output_path, "wb") as f:
+            f.write(out_buf)
+        print(f"  Written {len(out_buf):,} bytes -> {output_path}", file=sys.stderr)
+        return True
+
+
+# ---------------------------------------------------------------------------
+# Entry point
+# ---------------------------------------------------------------------------
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(
+        description=f"Download {TARGET_FILE} from an FFXIV ZiPatch file without a CLUT."
+    )
+    parser.add_argument(
+        "--version",
+        default=None,
+        metavar="VERSION",
+        help="Thaliak version string (default: latest). Accepts D/H-prefixed or bare form.",
+    )
+    parser.add_argument(
+        "--output",
+        required=True,
+        metavar="PATH",
+        help="Output directory or file path.",
+    )
+    args = parser.parse_args()
+
+    # Resolve output path to a file
+    out = args.output
+    if out.endswith(os.sep) or os.path.isdir(out):
+        out = os.path.join(out, TARGET_FILE)
+    elif not out.lower().endswith(".exe"):
+        os.makedirs(out, exist_ok=True)
+        out = os.path.join(out, TARGET_FILE)
+
+    print("Fetching patch metadata from Thaliak...", file=sys.stderr)
+    version_str, patch_url = get_patch_url(args.version)
+    print(f"  Version : {version_str}", file=sys.stderr)
+    print(f"  URL     : {patch_url}", file=sys.stderr)
+
+    success = extract_exe(patch_url, out)
+    sys.exit(0 if success else 1)
+
+
+if __name__ == "__main__":
+    main()

automation/ffxiv_info.py

@@ -0,0 +1,287 @@
+import os
+import sys
+import re
+import json
+import urllib.request
+from typing import Optional, List
+
+# --- Constants ---
+THALIAK_REPO = "4e9a232b"
+THALIAK_API = f"https://thaliak.xiv.dev/api/v2beta/repositories/{THALIAK_REPO}/patches"
+BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+VERSIONS_FILE = os.path.join(BASE_DIR, "automation", "ffxiv_versions_global.json")
+
+# --- Utilities ---
+
+
+def fetch_url(url, is_json=False):
+    print(f"Fetching {url}...", file=sys.stderr)
+    headers = {
+        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36",
+        "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8",
+        "Accept-Language": "en-US,en;q=0.9",
+    }
+    req = urllib.request.Request(url, headers=headers)
+    try:
+        with urllib.request.urlopen(req) as response:
+            content = response.read().decode("utf-8")
+            if is_json:
+                return json.loads(content)
+            return content
+    except urllib.error.HTTPError as e:
+        if e.code == 503:
+            print(
+                f"  Received 503 for {url}. Maintenance mode. Reading body anyway...",
+                file=sys.stderr,
+            )
+            try:
+                content = e.read().decode("utf-8")
+                if is_json:
+                    return json.loads(content)
+                return content
+            except Exception:
+                return None
+        else:
+            print(f"  HTTP Error {e.code}: {e.reason}", file=sys.stderr)
+        return None
+    except Exception as e:
+        print(f"  Error fetching {url}: {e}", file=sys.stderr)
+        return None
+
+
+# --- Version Info Helpers ---
+
+
+def get_version_sort_key(v):
+    """
+    Generates a sortable key for FFXIV versions.
+    Handles both strings (e.g., '7.05h2') and dicts containing 'retail_version'.
+    """
+    if isinstance(v, dict):
+        v = v.get("retail_version", "0.0")
+    # Using a list of (type_priority, value) tuples to avoid comparing list vs int
+    # Priorities: 0 for numeric, 1 for alphabetic
+    parts = re.findall(r"(\d+|[a-zA-Z]+)", v)
+    return [(0, int(p)) if p.isdigit() else (1, [ord(c) for c in p]) for p in parts]
+
+
+def load_versions() -> List[dict]:
+    """Loads all registered versions from the global registry."""
+    if os.path.exists(VERSIONS_FILE):
+        try:
+            with open(VERSIONS_FILE, "r") as f:
+                data = json.load(f)
+                if isinstance(data, list):
+                    data.sort(key=get_version_sort_key)
+                    return data
+        except (json.JSONDecodeError, IOError) as e:
+            print(f"Warning: Failed to load versions file: {e}", file=sys.stderr)
+    return []
+
+
+def fetch_latest_thaliak_patch() -> Optional[str]:
+    """
+    Fetches the latest date-based version string from the Thaliak API.
+    """
+    data = fetch_url(THALIAK_API, is_json=True)
+    if not data or "patches" not in data or not data["patches"]:
+        return None
+    return data["patches"][-1]["version_string"]
+
+
+# --- Lodestone Scraper ---
+
+
+def parse_lodestone_news_list(html):
+    """
+    Parses the Lodestone news list to extract news items (URL, Title, and Timestamp).
+    Handles the structure found on both the news category and the main landing page.
+    """
+    if not html:
+        return []
+
+    # This pattern captures the link and the title following it within the same list item structure.
+    # It accounts for the multi-line whitespace and optional tags like [Maintenance].
+    # Timestamp is optional as fallback pages may not have the JS helper.
+    pattern = re.compile(
+        r'<a [^>]*href="(?P<url>/lodestone/news/detail/[^"]+)"[^>]*>.*?<p [^>]*class="news__list--title"[^>]*>(?P<title_content>.*?)</p>(?:.*?ldst_strftime\((?P<timestamp>\d+))?',
+        re.DOTALL | re.I,
+    )
+
+    items = []
+    for match in pattern.finditer(html):
+        url = "https://na.finalfantasyxiv.com" + match.group("url")
+        # Clean title content of tags and whitespace
+        content = match.group("title_content")
+        title = re.sub(r"<[^>]+>", "", content).strip()
+        timestamp = int(match.group("timestamp")) if match.group("timestamp") else 0
+        items.append({"url": url, "title": title, "timestamp": timestamp})
+
+    return items
+
+
+def extract_patch_version(html):
+    """Extracts the 'X.XX' patch version from maintenance detail HTML."""
+    if not html or not (m := re.search(r"patch\s+(\d+\.\d+)", html, re.I)):
+        return None
+    v = m.group(1)
+    if len(v.split(".")[1]) == 1:
+        v += "0"
+    return v
+
+
+def get_next_hotfix_version(base_v, existing_versions):
+    matching = [
+        v["retail_version"]
+        for v in existing_versions
+        if v["retail_version"].startswith(base_v)
+    ]
+    hotfixes = [v for v in matching if "h" in v]
+    if not hotfixes:
+        return f"{base_v}h"
+
+    # Extract numeric suffix: h -> 1, h2 -> 2, etc.
+    nums = [int(h.split("h")[1]) if h.split("h")[1].isdigit() else 1 for h in hotfixes]
+    return f"{base_v}h{max(nums) + 1}"
+
+
+def determine_retail_version(base_v, existing_versions, thaliak_version_new):
+    matching = [v for v in existing_versions if v["retail_version"].startswith(base_v)]
+    latest = matching[-1] if matching else None
+
+    # If we have a record for this base version but the Thaliak version is new, it's a hotfix.
+    if latest and thaliak_version_new and thaliak_version_new != latest["version_string"]:
+        return get_next_hotfix_version(base_v, existing_versions)
+
+    return latest["retail_version"] if latest else base_v
+
+
+def is_maintenance_post(title):
+    """Returns True if the title represents a primary world maintenance post."""
+    t = title.lower()
+    return "all worlds" in t and "maintenance" in t and "follow-up" not in t
+
+
+def scrape_latest_maintenance(thaliak_version_new=None):
+    """
+    Scrapes the Lodestone to find the most recent 'All Worlds Maintenance' post.
+    """
+    # Step 1: News Category
+    url = "https://na.finalfantasyxiv.com/lodestone/news/category/2"
+    html = fetch_url(url)
+
+    # Step 2: Fallback to Landing Page (common during 503 maintenance)
+    if not html:
+        url = "https://na.finalfantasyxiv.com/lodestone/"
+        html = fetch_url(url)
+
+    if not html:
+        return None
+
+    news_items = parse_lodestone_news_list(html)
+    existing_versions = load_versions()
+
+    for item in news_items:
+        title = item["title"]
+        link = item["url"]
+
+        if not is_maintenance_post(title):
+            continue
+
+        detail_html = fetch_url(link)
+        if not detail_html:
+            continue
+
+        base_version = extract_patch_version(detail_html)
+        if not base_version:
+            continue
+
+        retail_version = determine_retail_version(
+            base_version, existing_versions, thaliak_version_new
+        )
+
+        return {"retail_version": retail_version, "title": title, "url": link}
+
+    return None
+
+
+def get_version_info() -> Optional[dict]:
+    """
+    Returns the latest version information by combining Thaliak and Lodestone data.
+    """
+    thaliak_version_new = fetch_latest_thaliak_patch()
+    maintenance = scrape_latest_maintenance(thaliak_version_new=thaliak_version_new)
+
+    if not maintenance and not thaliak_version_new:
+        return None
+
+    return {
+        "retail_version": maintenance["retail_version"] if maintenance else "Unknown",
+        "version_string": thaliak_version_new,
+        "title": maintenance["title"] if maintenance else "Unknown",
+        "url": maintenance["url"] if maintenance else "Unknown",
+    }
+
+
+def get_patch_context():
+    """
+    Returns a dictionary containing:
+    - retail_prev, thaliak_version_prev: The latest registered version.
+    - retail_new, thaliak_version_new: The latest versions found externally.
+    - is_new: True if thaliak_version_new is not in the registry.
+    """
+    versions = load_versions()
+    prev = versions[-1] if versions else None
+
+    thaliak_version_new = fetch_latest_thaliak_patch()
+    maintenance = scrape_latest_maintenance(thaliak_version_new=thaliak_version_new)
+    retail_new = maintenance["retail_version"] if maintenance else None
+
+    is_new = thaliak_version_new is not None and not any(
+        v["version_string"] == thaliak_version_new for v in versions
+    )
+
+    return {
+        "retail_prev": prev["retail_version"] if prev else None,
+        "thaliak_version_prev": prev["version_string"] if prev else None,
+        "retail_new": retail_new,
+        "thaliak_version_new": thaliak_version_new,
+        "is_new": is_new,
+    }
+
+
+def update_and_get_info():
+    """
+    Discovers current patch context, prints results for GitHub Actions,
+    and updates the local registration if a new version is found.
+    """
+    ctx = get_patch_context()
+
+    if ctx["thaliak_version_prev"]:
+        print(f"thaliak_version_prev={ctx['thaliak_version_prev']}")
+    if ctx["retail_prev"]:
+        print(f"retail_prev={ctx['retail_prev']}")
+    if ctx["thaliak_version_new"]:
+        print(f"thaliak_version_new={ctx['thaliak_version_new']}")
+    if ctx["retail_new"]:
+        print(f"retail_new={ctx['retail_new']}")
+    print(f"is_new={'true' if ctx['is_new'] else 'false'}")
+
+    if ctx["is_new"] and ctx["thaliak_version_new"] and ctx["retail_new"]:
+        data = load_versions()
+        data.append(
+            {
+                "retail_version": ctx["retail_new"],
+                "version_string": ctx["thaliak_version_new"],
+            }
+        )
+        print(
+            f"Added new version mapping: {ctx['retail_new']} -> {ctx['thaliak_version_new']}",
+            file=sys.stderr,
+        )
+        with open(VERSIONS_FILE, "w") as f:
+            json.dump(data, f, indent=2)
+
+
+if __name__ == "__main__":
+    update_and_get_info()

automation/ffxiv_versions_global.json

@@ -0,0 +1,178 @@
+[
+  {
+    "retail_version": "7.00",
+    "version_string": "2024.06.18.0000.0000"
+  },
+  {
+    "retail_version": "7.00h",
+    "version_string": "2024.07.06.0000.0000"
+  },
+  {
+    "retail_version": "7.01",
+    "version_string": "2024.07.10.0001.0000"
+  },
+  {
+    "retail_version": "7.05",
+    "version_string": "2024.07.24.0000.0000"
+  },
+  {
+    "retail_version": "7.05h",
+    "version_string": "2024.08.02.0000.0000"
+  },
+  {
+    "retail_version": "7.05h2",
+    "version_string": "2024.08.21.0000.0000"
+  },
+  {
+    "retail_version": "7.10",
+    "version_string": "2024.11.06.0000.0000"
+  },
+  {
+    "retail_version": "7.11",
+    "version_string": "2024.11.20.0000.0000"
+  },
+  {
+    "retail_version": "7.15",
+    "version_string": "2024.12.07.0000.0000"
+  },
+  {
+    "retail_version": "7.16",
+    "version_string": "2025.01.14.0000.0000"
+  },
+  {
+    "retail_version": "7.16h",
+    "version_string": "2025.01.28.0000.0000"
+  },
+  {
+    "retail_version": "7.18",
+    "version_string": "2025.02.19.0000.0000"
+  },
+  {
+    "retail_version": "7.18h",
+    "version_string": "2025.02.27.0000.0000"
+  },
+  {
+    "retail_version": "7.20",
+    "version_string": "2025.03.18.0000.0000"
+  },
+  {
+    "retail_version": "7.20h",
+    "version_string": "2025.03.27.0000.0000"
+  },
+  {
+    "retail_version": "7.21",
+    "version_string": "2025.04.16.0000.0000"
+  },
+  {
+    "retail_version": "7.25",
+    "version_string": "2025.05.17.0000.0000"
+  },
+  {
+    "retail_version": "7.25h",
+    "version_string": "2025.06.10.0000.0000"
+  },
+  {
+    "retail_version": "7.25h2",
+    "version_string": "2025.06.19.0000.0000"
+  },
+  {
+    "retail_version": "7.25h3",
+    "version_string": "2025.06.28.0000.0000"
+  },
+  {
+    "retail_version": "7.30",
+    "version_string": "2025.07.30.0000.0000"
+  },
+  {
+    "retail_version": "7.30h",
+    "version_string": "2025.08.07.0000.0000"
+  },
+  {
+    "retail_version": "7.31",
+    "version_string": "2025.08.22.0000.0000"
+  },
+  {
+    "retail_version": "7.31h",
+    "version_string": "2025.09.04.0000.0000"
+  },
+  {
+    "retail_version": "7.35",
+    "version_string": "2025.09.30.0000.0000"
+  },
+  {
+    "retail_version": "7.35h",
+    "version_string": "2025.10.13.0000.0000"
+  },
+  {
+    "retail_version": "7.38",
+    "version_string": "2025.10.30.0000.0000"
+  },
+  {
+    "retail_version": "7.40",
+    "version_string": "2025.12.09.0000.0000"
+  },
+  {
+    "retail_version": "7.40h",
+    "version_string": "2025.12.18.0000.0000"
+  },
+  {
+    "retail_version": "7.40h2",
+    "version_string": "2025.12.23.0000.0000"
+  },
+  {
+    "retail_version": "7.41",
+    "version_string": "2026.01.21.0000.0000"
+  },
+  {
+    "retail_version": "7.41h",
+    "version_string": "2026.01.30.0000.0000"
+  },
+  {
+    "retail_version": "7.45",
+    "version_string": "2026.02.20.0000.0000"
+  },
+  {
+    "retail_version": "7.45h",
+    "version_string": "2026.03.07.0000.0000"
+  },
+  {
+    "retail_version": "7.45h2",
+    "version_string": "2026.03.17.0000.0000"
+  },
+  {
+    "retail_version": "7.50",
+    "version_string": "2026.04.21.0000.0000"
+  },
+  {
+    "retail_version": "7.50h",
+    "version_string": "2026.05.01.0000.0000"
+  },
+  {
+    "retail_version": "7.51",
+    "version_string": "2026.05.25.0000.0000"
+  },
+  {
+    "retail_version": "7.51h",
+    "version_string": "2026.06.10.0000.0000"
+  },
+  {
+    "retail_version": "7.51h2",
+    "version_string": "2026.06.18.0000.0000"
+  },
+  {
+    "retail_version": "7.55",
+    "version_string": "2026.07.16.0001.0000"
+  },
+  {
+    "retail_version": "7.55h",
+    "version_string": "2026.08.05.0000.0000"
+  },
+  {
+    "retail_version": "7.55h2",
+    "version_string": "2026.08.11.0000.0000"
+  },
+  {
+    "retail_version": "7.56",
+    "version_string": "2026.09.01.0000.0000"
+  }
+]

debug_similarity_matrix.py

@@ -0,0 +1,67 @@
+import click
+from vtable_alignment import Similarity
+
+from utils import HexIntParamType
+
+
+@click.command()
+@click.argument(
+    "similarity_json_file",
+    type=click.Path(exists=True, dir_okay=False, resolve_path=True),
+)
+@click.argument("opcode", type=HexIntParamType())
+@click.option(
+    "--reverse",
+    is_flag=True,
+    help="Prints a column of the similarity matrix given a new opcode",
+)
+@click.option(
+    "--accept",
+    default=None,
+    help="Modifies the similarity matrix file by accepting the match between the opcode argument and the argument to this option",
+    type=HexIntParamType(),
+)
+def debug_similarity_matrix(similarity_json_file, opcode, reverse, accept):
+    """
+    Given an old opcode, debug prints a row of the similarity matrix generated
+    from generate_similarity_matrix.py.
+
+    Example:
+    python debug_similarity_matrix.py similarity.json 0x200
+    """
+
+    entries = dict()
+    similarity = Similarity(similarity_json_file)
+
+    if accept is not None:
+        similarity.accept(opcode, accept)
+        similarity.write_to_file(similarity_json_file)
+        return
+
+    if reverse:
+        print("Checking column of matrix since --reverse was provided")
+        for old_opcode in similarity.old_opcodes:
+            entries[old_opcode] = similarity.lookup(old_opcode, opcode)
+    else:
+        for new_opcode in similarity.new_opcodes:
+            entries[new_opcode] = similarity.lookup(opcode, new_opcode)
+
+    max_opcode = 0
+    max_score = -9999
+
+    print(f"Similarities for {hex(opcode)}")
+
+    entries = list(entries.items())
+    entries.sort(key=lambda x: x[1], reverse=True)
+    for candidate, similarity in entries:
+        print(f"\t{hex(candidate)} => {similarity}")
+        if similarity > max_score:
+            max_opcode = candidate
+            max_score = similarity
+
+    print("Best match")
+    print(f"{hex(max_opcode)} => {max_score}")
+
+
+if __name__ == "__main__":
+    debug_similarity_matrix()

diffs/README.md

@@ -0,0 +1,5 @@
+# Diffs
+
+A bunch of diffs saved up over the past couple of patches.
+
+**Please note the 6.3, 6.4, 6.5 major version diffs are probably not 100% accurate.**

generate_act_format.py

@@ -0,0 +1,87 @@
+import re
+import click
+
+# Convert opcodes to the ACT expected format
+desired_names = {
+    "StatusEffectList": None,
+    "StatusEffectList2": None,
+    "StatusEffectList3": None,
+    "BossStatusEffectList": None,
+    "StatusEffectList4": "StatusEffectListForay3",
+    "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,
+}
+
+
+def write_act_format(opcodes_lines):
+    output_lines = []
+    opcode_mapping = dict()
+
+    for line in opcodes_lines:
+        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 opcode_val == "UNKNOWN":
+            opcodes = []
+        elif " 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:
+            output_lines.append(f"{desired}|{opcodes[0]:x}")
+        elif len(opcodes) > 1:
+            output_lines.append(f'{desired}|{[f"{opcode:x}" for opcode in opcodes]}')
+        else:
+            output_lines.append(f"{desired}|???")
+
+    return output_lines
+
+
+@click.command()
+@click.argument("opcodes_file", type=click.File("r"))
+def generate_act_format(opcodes_file):
+    """
+    Parses an OPCODES_FILE and outputs opcodes in a format expected by ACT.
+    This is also just a bunch of regexes slapped together.
+
+    Outputs to stdout.
+
+    Example:
+
+    python generate_act_format.py Ipcs.h
+    """
+    lines = write_act_format(opcodes_file.readlines())
+    for line in lines:
+        print(line)
+
+
+if __name__ == "__main__":
+    generate_act_format()

generate_deep_traces.py

@@ -0,0 +1,250 @@
+import click
+import json
+import pathlib
+
+from analysis_utils import (
+    get_correct_switch,
+    get_packet_handler_addr,
+    get_packet_handler_opcode_offset,
+    get_packet_handler_switch_addr,
+)
+
+
+class RefNode:
+    def __init__(self, ea):
+        self.ea = ea
+        self.calls = []
+        self.branch0 = None
+        self.branch1 = None
+
+    def add_call(self, ea):
+        node = RefNode(ea)
+        self.calls.append(node)
+        return node
+
+    def add_b0(self, ea):
+        if self.branch0:
+            raise Exception(f"Node {repr(self)} already has a branch0!")
+        self.branch0 = RefNode(ea)
+        return self.branch0
+
+    def add_b1(self, ea):
+        if self.branch1:
+            raise Exception(f"Node {repr(self)} already has a branch1!")
+        self.branch1 = RefNode(ea)
+        return self.branch1
+
+    def __repr__(self):
+        calls = f", calls: {self.calls}" if self.calls else ""
+        branch0 = f", b0: {self.branch0}" if self.branch0 else ""
+        branch1 = f", b1: {self.branch1}" if self.branch1 else ""
+        return f"{{ ea: {hex(self.ea)}{calls}{branch0}{branch1} }}"
+
+
+class OpcodeCase:
+    def __init__(self, ea, opcodes):
+        self.ref = RefNode(ea)
+        self.opcodes = opcodes
+
+    def __repr__(self):
+        return f"{[hex(opcode) for opcode in self.opcodes]}: { repr(self.ref) }"
+
+
+class BlockDict(dict):
+    def missing_set(self, eas):
+        missing = set()
+        for ea in eas:
+            if ea not in self:
+                missing.add(ea)
+        return missing
+
+    def update_with_missing(self, r2, eas):
+        missing = self.missing_set(eas)
+        if len(missing) == 0:
+            return
+
+        block_json_list = r2.cmd(
+            f"pdbj @@={' '.join((hex(ea) for ea in missing))}"
+        ).splitlines()
+        missing_blocks = [json.loads(block_json) for block_json in block_json_list]
+        additional_blocks = {
+            missing_block[0]["addr"]: missing_block for missing_block in missing_blocks
+        }
+        self.update(additional_blocks)
+
+        if len(additional_blocks) < len(missing):
+            return self.missing_set(eas)
+
+    def mark_missing_as_unknown(self, eas):
+        missing = self.missing_set(eas)
+        self.update({ea: [{"unknown": True}] for ea in missing})
+
+
+def populate_child_refs(blocks, refs):
+    """Runs a single step of a BFS for traversing calls/jumps"""
+    child_refs = []
+    for ref in refs:
+        for insn in blocks[ref.ea]:
+            if "jump" in insn:
+                jump_ea = insn["jump"]
+                if insn["type"] == "call":
+                    child_refs.append(ref.add_call(jump_ea))
+                else:
+                    child_refs.append(ref.add_b1(jump_ea))
+                    if "fail" in insn:
+                        child_refs.append(ref.add_b0(insn["fail"]))
+    return child_refs
+
+
+def generate_opcodes_db(r2, switch, opcode_offset, fn_graph):
+    opcodes_db = dict()
+
+    for case_ea, data in switch.items():
+        opcodes = data["opcodes"]
+        resolved_opcodes = [int(opcode) + opcode_offset for opcode in opcodes]
+        opcodes_db[resolved_opcodes[0]] = OpcodeCase(case_ea, resolved_opcodes)
+
+    blocks = BlockDict()
+    for bb in fn_graph["bbs"]:
+        blocks[bb["addr"]] = bb["ops"]
+
+    # This should be a no-op since it is assumed the fn_graph would have
+    # every block in the function.
+    still_missing = blocks.update_with_missing(r2, switch.keys())
+    if still_missing:
+        raise Exception("There's no way there should be any missing blocks here")
+
+    ref_nodes = []
+    for opcase in opcodes_db.values():
+        ref_nodes.append(opcase.ref)
+
+    for i in range(10):
+        child_refs = populate_child_refs(blocks, ref_nodes)
+        eas = [ref.ea for ref in child_refs]
+        still_missing = blocks.update_with_missing(r2, eas)
+        if still_missing:
+            r2.cmd(f"af @@={' '.join((hex(ea) for ea in still_missing))}")
+            yet_still_missing = blocks.update_with_missing(r2, still_missing)
+            if yet_still_missing:
+                blocks.mark_missing_as_unknown(yet_still_missing)
+        ref_nodes = child_refs
+
+    return opcodes_db, blocks
+
+
+def extract_opcode_data(exe_file):
+    from utils import eprint, sync_r2_output
+
+    import r2pipe
+
+    r2 = r2pipe.open(exe_file, ["-2"])
+    eprint(f"Radare loaded {exe_file}")
+
+    sync_r2_output(r2)
+
+    target = get_packet_handler_addr(r2, exe_file)
+    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")
+
+    eprint(f"  Loaded switch cases")
+
+    ## 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
+
+    ## STEP 3: Grab function graph
+    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
+    )
+    eprint(f"  Found switch at {switch_ea}")
+
+    opcodes_db, blocks = generate_opcodes_db(
+        r2, packet_handler_switch, opcode_offset, fn_graph
+    )
+
+    eprint(f"  Loaded {len(opcodes_db)} cases from packet handler")
+
+    r2.quit()
+
+    return opcodes_db, blocks
+
+
+def bb_lines(blocks, ref):
+    lines = []
+    block = blocks[ref.ea]
+    for insn in block:
+        if "unknown" in insn:
+            lines.append("UNKNOWN_BLOCK")
+        else:
+            lines.append(insn["opcode"])
+    return lines
+
+
+def trace_lines(blocks, ref):
+    # Generate traces in a BFS fashion
+    lines = []
+    refs = [(None, ref)]
+    while refs:
+        (name, ref) = refs.pop(0)
+        if name:
+            lines.append(name)
+        lines.extend(bb_lines(blocks, ref))
+        if name:
+            lines.append(f"{name}_END")
+        for i, call_ref in enumerate(ref.calls):
+            refs.append((f"CALL{i}", call_ref))
+        if ref.branch0:
+            refs.append(("BRANCH0", ref.branch0))
+        if ref.branch1:
+            refs.append(("BRANCH1", ref.branch1))
+    return lines
+
+
+@click.command()
+@click.argument(
+    "exe_file", type=click.Path(exists=True, dir_okay=False, resolve_path=True)
+)
+@click.argument("output_dir", type=click.Path(file_okay=False))
+def generate_deep_traces(exe_file, output_dir):
+    """
+    Generates deep traces for every packet handler in the target EXE_FILE.
+    This outputs an ASM trace as an .asm file for each pointer opcode.
+    It also outputs an `opcode_sets.json` that maps each pointer opcode to the
+    full set of opcodes handled by that trace.
+
+    These aren't normal traces by any measure; they are simply basic blocks
+    printed in BFS order, where children blocks are added to the BFS tree
+    by traversing calls and jumps.
+
+    Example:
+
+    python generate_deep_traces.py ffxiv_dx11.6.28h.exe 6.28h-traces
+    """
+    opcodes_db, blocks = extract_opcode_data(exe_file)
+
+    pathlib.Path(output_dir).mkdir(parents=True, exist_ok=True)
+
+    for opcode, opcase in opcodes_db.items():
+        with open(f"{output_dir}/{hex(opcode)}.asm", "w+") as f:
+            trace = trace_lines(blocks, opcase.ref)
+            f.writelines(s + "\n" for s in trace)
+
+    with open(f"{output_dir}/opcode_sets.json", "w+") as f:
+        opcode_sets = {
+            opcode: opcase.opcodes for (opcode, opcase) in opcodes_db.items()
+        }
+        json.dump(opcode_sets, f)
+
+
+if __name__ == "__main__":
+    generate_deep_traces()

generate_diff_offset.py

@@ -0,0 +1,43 @@
+import click
+import json
+
+
+@click.command()
+@click.argument("diff_file", type=click.File("r"))
+@click.argument("offset", type=int)
+def generate_diff_offset(diff_file, offset):
+    """
+    Given a diff file, produces another diff file where the matches
+    are offset by OFFSET.
+
+    Example:
+    python generate_diff_offset.py -- diff.json -1 > offset.diff.json
+    """
+    diff_json = json.load(diff_file)
+    olds = []
+    news = []
+    for pair in diff_json:
+        if "old" not in pair or "new" not in pair:
+            continue
+        olds.append(pair["old"])
+        news.append(pair["new"])
+
+    if offset < 0:
+        news = [[] for _ in range(-offset)] + news
+    if offset > 0:
+        olds = [[] for _ in range(offset)] + olds
+
+    opcodes_object = []
+    for old, new in zip(olds, news):
+        opcodes_object.append(
+            {
+                "old": old,
+                "new": new,
+            }
+        )
+
+    print(json.dumps(opcodes_object, indent=2))
+
+
+if __name__ == "__main__":
+    generate_diff_offset()

generate_opcodes_file.py

@@ -0,0 +1,101 @@
+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
+
+
+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"
+
+
+def replace_line_with_new_opcode(line, diff, ver, old_ver):
+    match_groups = re.findall(
+        r"^(\s*[^\/]\w+\s*)=\s*(.*),(\s*)\/\/.*" + old_ver + "$", line
+    )
+    if len(match_groups) != 1:
+        return line
+
+    opcode_name = match_groups[0][0]
+    opcode_val = match_groups[0][1]
+    comment_spacing = match_groups[0][2]
+    if " or " in opcode_val:
+        old_opcode = int(opcode_val.split(" or ")[0], 16)
+    else:
+        old_opcode = int(opcode_val, 16)
+    if old_opcode in diff:
+        new_opcodes = diff[old_opcode]
+        return f"{opcode_name}= {opcodes_str(new_opcodes)},{comment_spacing}// updated {ver}\n"
+
+    return f"{line}"
+
+
+@click.command()
+@click.argument("old_version_string")
+@click.argument("new_version_string")
+@click.argument("diff_file", type=click.File("r"))
+@click.argument("opcodes_file", type=click.File("r"))
+@click.option(
+    "--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",
+)
+def generate_opcodes_file(
+    old_version_string, new_version_string, diff_file, opcodes_file, reverse, output
+):
+    """
+    Parses an OPCODES_FILE and applies a JSON DIFF_FILE to generate a new one.
+    The opcodes file is basically anything that has syntax resembling Sapphire's
+    `Ipcs.h`. It doesn't do any C++ header parsing; it's just a bunch of regexes
+    slapped together but it works.
+
+    Example:
+
+    python generate_opcodes_file.py 6.30 6.30h diff.json Ipcs.h
+    """
+    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,
+                old_version_string,
+            )
+        )
+
+    if output:
+        new_filename = output
+    else:
+        new_filename = f"ipcs/Ipcs.{new_version_string}.h"
+    with open(new_filename, "w+") as f:
+        f.writelines(queued_lines)
+
+    print("Wrote to", new_filename)
+
+
+if __name__ == "__main__":
+    generate_opcodes_file()

generate_similarity_matrix.py

@@ -0,0 +1,187 @@
+import click
+import torch
+from asm2vec.utils import (
+    TraceData,
+    train,
+    save_model,
+    cosine_similarities,
+)
+from asm2vec.datatype import Tokens
+import json
+
+
+def length_heuristic(l0, l1, debug=False):
+    """
+    Function length heuristic (since asm2vec is terrible at handling mismatched lengths)
+
+    Returns a similarity metric in the range [0, 1]
+    """
+    length_diff = abs(l0 - l1)
+    # Weight mismatched lengths considerably lower, but clip factor to 0
+    length_factor = max(1 - 4 * (length_diff / (l0 + l1)), 0)
+    if debug:
+        print("Length factor", l0, l1, length_factor)
+    return length_factor
+
+
+def full_similarity_matrix(
+    cosine_similarity_matrix,
+    old_trace_data: TraceData,
+    new_trace_data: TraceData,
+):
+    """
+    Generates a matrix comparing all old opcodes to all new opcodes.
+
+    Returns (old opcodes, new_opcodes, similarity_matrix)
+    """
+    old_fns = old_trace_data.traces
+    new_fns = new_trace_data.traces
+
+    old_opcodes = []
+    new_opcodes = []
+    # Full similarity matrix mapping old_opcodes => new_opcodes
+    similarity_matrix = []
+
+    for old_data in old_trace_data.opcodes.values():
+        similarities = []
+        for new_data in new_trace_data.opcodes.values():
+            old_idx = old_data["fn_idx"]
+            new_idx = new_data["fn_idx"]
+            # Use the length of the instructions for the length heuristic
+            l0 = len(old_fns[old_idx].insts)
+            l1 = len(new_fns[new_idx].insts)
+            length_factor = length_heuristic(l0, l1)
+
+            h0 = old_data["packet_size_hint"]
+            h1 = new_data["packet_size_hint"]
+
+            packet_size_factor = 0
+            if h0 == h1 and h0 != 0:
+                packet_size_factor = 0.5
+            else:
+                packet_size_factor = -0.5
+            # Since cosine similarity is in the range (-1, 1), add 1 to push it
+            # into the range (0, 2).
+            cs = cosine_similarity_matrix[old_idx, new_idx] + 1
+
+            # Multiply the length factor and cosine similarity together to
+            # yield some value in the range (0, 2), then subtract 1 to get a
+            # score from range (-1, 1)
+            score = length_factor * cs - 1
+
+            # Add or subtract score depending on the packet size matching
+            # Also clamp value to between (-1, 1)
+            score = max(min(score + packet_size_factor, 1.0), -1.0)
+
+            # Now we copy this similarity value for all opcodes in the new
+            # switch case
+            for op in new_data["opcodes"]:
+                similarities.append(float(score))
+        # Now we copy this similarity mapping for all opcodes in the old
+        # switch case
+        for op in old_data["opcodes"]:
+            similarity_matrix.append(similarities)
+
+    for old_data in old_trace_data.opcodes.values():
+        for op in old_data["opcodes"]:
+            old_opcodes.append(op)
+
+    for new_data in new_trace_data.opcodes.values():
+        for op in new_data["opcodes"]:
+            new_opcodes.append(op)
+
+    return (old_opcodes, new_opcodes, similarity_matrix)
+
+
+def print_banner(text):
+    print("")
+    print(f"======= {text} =======")
+    print("")
+
+
+def write_matrix_to_file(output_file, old_opcodes, new_opcodes, similarity_matrix):
+    with open(output_file, "w+") as f:
+        json.dump(
+            {
+                "old_opcodes": old_opcodes,
+                "new_opcodes": new_opcodes,
+                "matrix": similarity_matrix,
+            },
+            f,
+            indent=4,
+        )
+        print_banner(f"Output written to {output_file}")
+
+
+@click.command()
+@click.argument(
+    "old_traces", type=click.Path(exists=True, file_okay=False, resolve_path=True)
+)
+@click.argument(
+    "new_traces", type=click.Path(exists=True, file_okay=False, resolve_path=True)
+)
+@click.argument("output_file", type=click.Path(dir_okay=False, resolve_path=True))
+def generate_similarity_matrix(old_traces, new_traces, output_file):
+    """
+    Compares the OLD_TRACES and NEW_TRACES directories generated by the
+    `generate_deep_traces.py` script.
+
+    Creates a JSON OUTPUT_FILE containing a pairwise similarity matrix of all
+    opcodes found.
+
+    \b
+    {
+        "old_opcodes": (list of old opcodes indexing dimension 0),
+        "new_opcodes": (list of new opcodes indexing dimenision 1),
+        "matrix": (m by n array of floats: [[]]),
+    }
+    Example:
+
+    python generate_similarity_matrix.py old-traces/ new-traces/ similarity.json
+    """
+    tokens = Tokens()
+    old_trace_data = TraceData.load_data(old_traces, tokens)
+    new_trace_data = TraceData.load_data(new_traces, tokens)
+
+    opath = "model.pt"
+
+    def training_callback(context):
+        progress = f'{context["epoch"]} | time = {context["time"]:.2f}, loss = {context["loss"]:.4f}'
+        if context["accuracy"]:
+            progress += f', accuracy = {context["accuracy"]:.4f}'
+        print(progress)
+        save_model(opath, context["model"], context["tokens"])
+
+    training_params = {
+        "embedding_size": 100,
+        "batch_size": 1024,
+        "epochs": 20,
+        "neg_sample_num": 25,
+        "calc_acc": True,
+        "device": "cuda" if torch.cuda.is_available() else "cpu",
+        "callback": training_callback,
+        "learning_rate": 0.02,
+    }
+
+    print_banner("Training embeddings from scratch on old trace data")
+    model = train(old_trace_data, **training_params)
+
+    # Prepare the model for new trace data and freeze all training from old trace data
+    model.init_estimation_mode(len(new_trace_data.traces))
+
+    print_banner("Calculating embeddings for new trace data")
+    model = train(new_trace_data, model=model, mode="test", **training_params)
+
+    print_banner("Calculating cosine similarities")
+    csm = cosine_similarities(model)
+
+    print_banner("Computing full similarity matrix")
+    old_opcodes, new_opcodes, similarity_matrix = full_similarity_matrix(
+        csm, old_trace_data, new_trace_data
+    )
+
+    write_matrix_to_file(output_file, old_opcodes, new_opcodes, similarity_matrix)
+
+
+if __name__ == "__main__":
+    generate_similarity_matrix()

minor_patch_diff.py

@@ -0,0 +1,191 @@
+import click
+import json
+
+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_block_sizes(blocks):
+    block_sizes = dict()
+    for block in blocks:
+        block_sizes[block["addr"]] = block["size"]
+    return block_sizes
+
+
+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]] = {
+            "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 extract_opcode_data(exe_file):
+    from utils import eprint, sync_r2_output
+
+    import r2pipe
+
+    r2 = r2pipe.open(exe_file, ["-2"])
+    eprint(f"Radare loaded {exe_file}")
+
+    sync_r2_output(r2)
+
+    target = get_packet_handler_addr(r2, exe_file)
+    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")
+
+    eprint(f"  Loaded switch cases")
+
+    ## 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
+
+    ## STEP 3: Grab blocks from packet handler
+    blocks = r2.cmdj("afbj")
+
+    r2.quit()
+
+    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
+    )
+    eprint(f"  Found switch at {switch_ea}")
+
+    block_sizes = get_block_sizes(blocks)
+    opcodes_db = generate_opcodes_db(
+        packet_handler_ea, packet_handler_switch, opcode_offset, block_sizes
+    )
+
+    eprint(f"  Loaded {len(opcodes_db)} cases from packet handler")
+
+    return opcodes_db
+
+
+def find_closest_rel_ea(opcodes_db, dest):
+    closest = fucked_distance
+    closest_opcode = None
+
+    for opcode, case in opcodes_db.items():
+        rel_ea = case["rel_ea"]
+
+        num = abs(rel_ea - dest)
+
+        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])
+
+
+def add_match_case(cases, case):
+    # check if case already exists
+
+    for c in cases:
+        if c["rel_ea"] == case["rel_ea"]:
+            return
+
+    cases.append(case)
+
+
+def find_opcode_matches(old_opcodes_db, new_opcodes_db):
+    matches = []
+    new_opcodes = list(new_opcodes_db.keys())
+
+    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"])
+
+        if dist == fucked_distance:
+            continue
+
+        order_match_opcode = new_opcodes[k]
+
+        order_match = new_opcodes_db[order_match_opcode]
+        dist_match = new_opcodes_db[dist_match_opcode]
+
+        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"]))
+
+    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)
+)
+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.
+
+    The format of the output is a list (all fields are optional):
+
+    \b
+    [
+        {
+            "old": (list of opcodes in the switch case),
+            "new": (list of opcodes in the switch case),
+        },
+        ...
+    ]
+
+    Example:
+
+    python minor_patch_diff.py ffxiv_dx11.old.exe ffxiv_dx11.new.exe > diff.json
+    """
+    old_opcodes_db = extract_opcode_data(old_exe)
+    new_opcodes_db = extract_opcode_data(new_exe)
+
+    opcodes_found = find_opcode_matches(old_opcodes_db, new_opcodes_db)
+    opcodes_object = []
+
+    for old, new in opcodes_found:
+        opcodes_object.append(
+            {
+                "old": [hex(o) for o in old],
+                "new": [hex(o) for o in new],
+            }
+        )
+
+    print(json.dumps(opcodes_object, indent=2))
+
+
+if __name__ == "__main__":
+    minor_patch_diff()

requirements-vtable.txt

@@ -0,0 +1,3 @@
+click==8.2.1
+r2pipe==1.9.6
+semver==3.0.4

requirements.txt

@@ -0,0 +1,14 @@
+black==25.1.0
+click==8.2.1
+colorama==0.4.6
+mypy-extensions==1.1.0
+numpy==2.3.2
+pathspec==0.12.1
+platformdirs==4.4.0
+r2pipe==1.9.6
+rapidfuzz==3.14.0
+semver==3.0.4
+-f https://download.pytorch.org/whl/cu129
+torch==2.8.0
+tqdm==4.67.1
+typing_extensions==4.15.0

sanity_check.py

@@ -0,0 +1,63 @@
+import json
+import re
+import click
+
+
+def load_diff_file(f):
+    diff = dict()
+    diff_json = json.load(f)
+    for pair in diff_json:
+        if "old" not in pair or "new" not in pair:
+            continue
+        for old_opcode in pair["old"]:
+            diff[int(old_opcode, 16)] = set(
+                (int(new_opcode, 16) for new_opcode in pair["new"])
+            )
+    return diff
+
+
+@click.command()
+@click.argument("vtable_diff", type=click.File("r"))
+@click.argument("minor_patch_diff", type=click.File("r"))
+def sanity_check(vtable_diff, minor_patch_diff):
+    """
+    Compares two JSON diff files to cross-check results from the different
+    scripts in this repo.
+
+    Example:
+
+    python sanity_check.py vtable_diff.json minor_patch_diff.json
+    """
+    diff1 = load_diff_file(vtable_diff)
+    diff2 = load_diff_file(minor_patch_diff)
+
+    all_good = True
+    for old_opcode, new_opcodes in diff2.items():
+        if old_opcode not in diff1:
+            if len(new_opcodes) < 50:
+                print(f"Missing old opcode in vtable diff: {hex(old_opcode)}")
+                all_good = False
+            continue
+
+        other_new_opcodes = diff1[old_opcode]
+        vtable_new_opcode = None
+        if len(other_new_opcodes) > 0:
+            vtable_new_opcode = list(other_new_opcodes)[0]
+        elif len(new_opcodes) == 0:
+            continue
+
+        if vtable_new_opcode not in new_opcodes:
+            new_text = (
+                hex(vtable_new_opcode) if vtable_new_opcode is not None else "None"
+            )
+            print(f"vtable diff mismatch for case {hex(old_opcode)} => {new_text}")
+            all_good = False
+
+    if all_good:
+        print("Sanity check OK")
+    else:
+        print("Errors detected")
+
+
+if __name__ == "__main__":
+    sanity_check()

utils.py

@@ -0,0 +1,56 @@
+import sys
+import time
+import os
+import click
+
+
+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)
+
+
+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.
+
+    Read stuff from the process pipe until it stops being stupid.
+    """
+
+    def set_blocking(fileno, blocking):
+        if hasattr(os, "set_blocking"):
+            os.set_blocking(fileno, blocking)
+
+    time.sleep(1)
+    set_blocking(r2.process.stdout.fileno(), False)
+    p = r2.process.stdout.read(1)
+    set_blocking(r2.process.stdout.fileno(), True)
+    output = r2.cmd(f"?vi 123").strip()
+    if output != "123":
+        raise Exception("R2 state never got synced")
+
+
+class HexIntParamType(click.ParamType):
+    name = "integer"
+
+    def convert(self, value, param, ctx):
+        if isinstance(value, int):
+            return value
+
+        try:
+            if value[:2].lower() == "0x":
+                return int(value[2:], 16)
+            return int(value, 16)
+        except ValueError:
+            self.fail(f"{value!r} is not a valid hex integer", param, ctx)

vtable_alignment.py

@@ -0,0 +1,375 @@
+import click
+import json
+import numpy as np
+
+from vtable_diff import extract_opcode_data
+from utils import eprint
+from generate_similarity_matrix import write_matrix_to_file
+
+
+def needleman_wunsch(old_seq, new_seq, similarity, gap_penalty):
+    """
+    The Needleman-Wunsch algorithm adapted from
+    https://github.com/farhanma/pyseq/blob/master/functions1_3.py
+
+    Returns (alignment, alignment_score)
+    """
+
+    # Stage 1: Create a zero matrix and fills it via algorithm
+    n, m = len(old_seq), len(new_seq)
+    mat = []
+    for i in range(n + 1):
+        mat.append([0] * (m + 1))
+    for j in range(m + 1):
+        mat[0][j] = gap_penalty * j
+    for i in range(n + 1):
+        mat[i][0] = gap_penalty * i
+    for i in range(1, n + 1):
+        for j in range(1, m + 1):
+            # max(Match, Insertion, Deletion)
+            mat[i][j] = max(
+                mat[i - 1][j - 1] + similarity.lookup(old_seq[i - 1], new_seq[j - 1]),
+                mat[i][j - 1] + gap_penalty,
+                mat[i - 1][j] + gap_penalty,
+            )
+
+    # Stage 2: Computes the final alignment, by backtracking through matrix
+    alignment = []
+    i, j = n, m
+    while i and j:
+        score, scoreDiag, scoreUp, scoreLeft = (
+            mat[i][j],
+            mat[i - 1][j - 1],
+            mat[i - 1][j],
+            mat[i][j - 1],
+        )
+        if score == scoreDiag + similarity.lookup(old_seq[i - 1], new_seq[j - 1]):
+            alignment.append((old_seq[i - 1], new_seq[j - 1]))
+            i -= 1
+            j -= 1
+        elif score == scoreUp + gap_penalty:
+            alignment.append((old_seq[i - 1], None))
+            i -= 1
+        elif score == scoreLeft + gap_penalty:
+            alignment.append((None, new_seq[j - 1]))
+            j -= 1
+    while i:
+        alignment.append((old_seq[i - 1], None))
+        i -= 1
+    while j:
+        alignment.append((None, new_seq[j - 1]))
+        j -= 1
+
+    # Since we were backtracking, we reverse the collected alignment
+    alignment.reverse()
+
+    return alignment, mat[n][m]
+
+
+class Placeholder:
+    def __init__(self, old, new):
+        self.old = old
+        self.new = new
+
+
+class Similarity:
+    def __init__(self, similarity_json_file):
+        with open(similarity_json_file) as f:
+            data = json.load(f)
+
+        self.old_opcodes = {
+            opcode: idx for (idx, opcode) in enumerate(data["old_opcodes"])
+        }
+        self.new_opcodes = {
+            opcode: idx for (idx, opcode) in enumerate(data["new_opcodes"])
+        }
+        self.matrix = np.array(data["matrix"])
+        self.warnings = set()
+
+    def lookup(self, old_opcode, new_opcode):
+        if isinstance(old_opcode, Placeholder) or isinstance(new_opcode, Placeholder):
+            return -9999
+
+        if old_opcode not in self.old_opcodes:
+            self.warnings.add(
+                f"WARNING: Could not find old opcode {hex(old_opcode)} in similarity matrix"
+            )
+            return 0
+        if new_opcode not in self.new_opcodes:
+            self.warnings.add(
+                f"WARNING: Could not find new opcode {hex(new_opcode)} in similarity matrix"
+            )
+            return 0
+        i = self.old_opcodes[old_opcode]
+        j = self.new_opcodes[new_opcode]
+        return self.matrix[i][j]
+
+    def accept(self, old_opcode, new_opcode):
+        if old_opcode not in self.old_opcodes:
+            self.warnings.add(
+                f"WARNING: Could not find old opcode {hex(old_opcode)} in similarity matrix"
+            )
+            return
+        if new_opcode not in self.new_opcodes:
+            self.warnings.add(
+                f"WARNING: Could not find new opcode {hex(new_opcode)} in similarity matrix"
+            )
+            return
+        i = self.old_opcodes[old_opcode]
+        j = self.new_opcodes[new_opcode]
+        self.matrix[i][j] = 1
+
+    def get_confident_matches(self, threshold=0.1):
+        """
+        Returns matches in the form of [(old,new), ...] that are confidently
+        above the score threshold and where all pairs in the matching prefer
+        each other over any other opcodes.
+        """
+        scores = {}
+        new_opcodes = np.array([op for op in self.new_opcodes])
+        for old_op, i in self.old_opcodes.items():
+            top_n_idxs = np.argpartition(-self.matrix[i], 2)[:2]
+            scores[old_op] = [(new_opcodes[j], self.matrix[i][j]) for j in top_n_idxs]
+            scores[old_op].sort(key=lambda x: x[1], reverse=True)
+
+        transposed_matrix = self.matrix.transpose()
+        rev_scores = {}
+        old_opcodes = np.array([op for op in self.old_opcodes])
+        for new_op, j in self.new_opcodes.items():
+            top_n_idxs = np.argpartition(-transposed_matrix[j], 2)[:2]
+            rev_scores[new_op] = [
+                (old_opcodes[i], transposed_matrix[j][i]) for i in top_n_idxs
+            ]
+            rev_scores[new_op].sort(key=lambda x: x[1], reverse=True)
+
+        # Accept matches if they pass the threshold in the old => new direction,
+        # and if the new => old direction is a best match
+        matches = []
+        for old_op, top_matches in scores.items():
+            new_op = top_matches[0][0]
+            if top_matches[0][1] - top_matches[1][1] >= threshold:
+                rev_top_matches = rev_scores[new_op]
+                if (
+                    rev_top_matches[0][0] == old_op
+                    and rev_top_matches[0][1] > 0
+                    and rev_top_matches[0][1] - rev_top_matches[1][1] >= threshold
+                ):
+                    matches.append((old_op, new_op))
+
+        return matches
+
+    def clear_warnings(self):
+        self.warnings = set()
+
+    def print_warnings(self):
+        for warning in self.warnings:
+            eprint(warning)
+
+    def write_to_file(self, output_file):
+        write_matrix_to_file(
+            output_file,
+            list(self.old_opcodes.keys()),
+            list(self.new_opcodes.keys()),
+            self.matrix.tolist(),
+        )
+
+
+def find_potential_reorders(similarity: Similarity, alignment):
+    """
+    Given an alignment, determines pairs that are potentially reordered
+    in the alignment.
+    """
+    matches = similarity.get_confident_matches()
+    old_seq = filter(lambda x: x is not None, (old for (old, _) in alignment))
+    new_seq = filter(lambda x: x is not None, (new for (_, new) in alignment))
+    old_seq_set = set(old_seq)
+    new_seq_set = set(new_seq)
+
+    # Ensure matches at least exist somewhere in the seq
+    matches = list(
+        filter(lambda x: x[0] in old_seq_set and x[1] in new_seq_set, matches)
+    )
+
+    eprint(f'Found {len(matches)} "confident" matches')
+
+    reorders = []
+
+    # Check for reorders
+    matched_old = {match[0]: match[1] for match in matches}
+
+    for old, new in alignment:
+        if old in matched_old and matched_old[old] != new:
+            truth = matched_old[old]
+            mismatched = ""
+            if new is not None:
+                mismatched = hex(new)
+
+            eprint(
+                f"Potential reorder detected! {hex(old)} => {hex(truth)}, got {mismatched}"
+            )
+            reorders.append((old, truth))
+
+    return reorders
+
+
+def calculate_score(similarity: Similarity, alignment, gap_penalty=-1):
+    """
+    Given an alignment, determines the score in O(n+m) time.
+    """
+    score = 0
+    for old, new in alignment:
+        if old is None:
+            score += gap_penalty
+        elif new is None:
+            score += gap_penalty
+        else:
+            score += similarity.lookup(old, new)
+    return score
+
+
+def reorder_and_align(similarity: Similarity, old_seq, new_seq, reorders):
+    """
+    Reorders the sequences so that the pairs in the reorders set are forced to
+    match.  Then performs an alignment.
+    """
+    old_matches = set()
+    new_matches = dict()
+    for old, new in reorders:
+        old_matches.add(old)
+        new_matches[new] = Placeholder(old, new)
+
+    old_seq = list(filter(lambda x: x not in old_matches, old_seq))
+    new_seq = list(map(lambda x: new_matches[x] if x in new_matches else x, new_seq))
+
+    similarity.clear_warnings()
+    alignment, _ = needleman_wunsch(old_seq, new_seq, similarity, -1)
+    similarity.print_warnings()
+    fixed_alignment = []
+    for old, target in alignment:
+        if isinstance(target, Placeholder):
+            fixed_alignment.append((target.old, target.new))
+        else:
+            fixed_alignment.append((old, target))
+
+    return fixed_alignment
+
+
+def find_best_alignment(
+    similarity: Similarity, original_alignment, reorders, improvement_threshold=1.0
+):
+    """
+    Iteratively tests each match to see if fixing them would result in a better
+    alignment greater than the improvement_threshold. Returns the best alignment
+    using a subset of the mismatches.
+    """
+
+    original_score = calculate_score(similarity, original_alignment)
+    old_seq = list(
+        filter(lambda x: x is not None, (old for (old, _) in original_alignment))
+    )
+    new_seq = list(
+        filter(lambda x: x is not None, (new for (_, new) in original_alignment))
+    )
+
+    promising_reorders = []
+
+    for old, new in reorders:
+        eprint(f"Testing reorder {hex(old)} => {hex(new)}")
+        candidate_alignment = reorder_and_align(
+            similarity, old_seq, new_seq, [(old, new)]
+        )
+        candidate_score = calculate_score(similarity, candidate_alignment)
+        eprint(f"Alignment score: {candidate_score}")
+        if candidate_score > original_score + improvement_threshold:
+            promising_reorders.append((old, new))
+
+    if len(promising_reorders) == 0:
+        eprint("No promising reorders")
+        return None
+
+    eprint(
+        "Testing promising reorders:",
+        {f"({hex(old)} => {hex(new)}) " for (old, new) in promising_reorders},
+    )
+    promising_alignment = reorder_and_align(
+        similarity, old_seq, new_seq, promising_reorders
+    )
+    candidate_score = calculate_score(similarity, promising_alignment)
+    eprint(f"New alignment score: {candidate_score}")
+    return promising_alignment
+
+
+@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(
+    "similarity_json_file",
+    type=click.Path(exists=True, dir_okay=False, resolve_path=True),
+)
+def vtable_alignment(old_exe, new_exe, similarity_json_file):
+    """
+    A more generalized version of vtable_diff. Generates an opcode
+    diff file by running a sequence alignment algorithm and attempting
+    to find the optimal global alignment of the vtable opcodes
+    from different exe versions.
+
+    Requires a similarity matrix generated from generate_similarity_matrix.py.
+
+    This script outputs to stdout, so pipe it to a json file.
+
+    The format of the output is a list (all fields are optional):
+
+    \b
+    [
+        {
+            "old": [opcode],
+            "new": [opcode],
+        },
+        ...
+    ]
+
+    Example:
+
+    python vtable_alignment.py ffxiv_dx11.old.exe ffxiv_dx11.new.exe similarity.json > diff.json
+    """
+    old_opcodes_db = extract_opcode_data(old_exe)
+    new_opcodes_db = extract_opcode_data(new_exe)
+
+    old_seq = [opcode for opcode in old_opcodes_db.values()]
+    new_seq = [opcode for opcode in new_opcodes_db.values()]
+
+    similarity = Similarity(similarity_json_file)
+
+    eprint("Running initial alignment...")
+    alignment, score = needleman_wunsch(old_seq, new_seq, similarity, -1)
+
+    similarity.print_warnings()
+    eprint(f"Alignment score: {score}")
+
+    eprint("Finding potential reorders")
+    reorders = find_potential_reorders(similarity, alignment)
+
+    new_alignment = find_best_alignment(similarity, alignment, reorders)
+    if new_alignment is not None:
+        alignment = new_alignment
+
+    diff = []
+    for old, new in alignment:
+        if old is None:
+            eprint("New opcode did not find matching old one:", hex(new))
+            diff.append({"old": [], "new": [hex(new)]})
+        elif new is None:
+            eprint("Old opcode did not find matching new one:", hex(old))
+            diff.append({"old": [hex(old)], "new": []})
+        else:
+            diff.append({"old": [hex(old)], "new": [hex(new)]})
+
+    print(json.dumps(diff, indent=2))
+
+
+if __name__ == "__main__":
+    vtable_alignment()

vtable_diff.py

@@ -0,0 +1,159 @@
+import click
+import json
+import sys
+
+from analysis_utils import get_correct_switch
+from utils import eprint, create_r2_byte_pattern, sync_r2_output
+import r2pipe
+
+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):
+    orig_loc = r2.cmd("s")  # Save original spot
+    r2.cmd("aei; aeim; aeip")  # Initialize ESIL VM, stack, instruction pointer
+    r2.cmd('"aesue rax,0x0,>"')  # continue until rax changes
+    r2.cmd('"aesue rax,0x0,>"')  # continue until rax changes again
+    r2.cmd("aeso")  # step
+    r2.cmd("aer rax=0x200")  # set rax to some arbitrary number
+    r2.cmd("aeso")  # step
+    regs = r2.cmdj("arj")
+    new_rax = regs["rax"]
+    opcode_offset = 0x200 - new_rax
+
+    # Clear the ESIL environment
+    r2.cmd("ar0; aeim-; aei-")
+    r2.cmd(f"s {orig_loc}")  # Seek back to original spot
+
+    return opcode_offset
+
+
+def extract_opcode_data(exe_file):
+    r2 = r2pipe.open(exe_file, ["-2"])
+    eprint(f"Radare loaded {exe_file}")
+
+    sync_r2_output(r2)
+
+    p = create_r2_byte_pattern(PACKET_INTERFACE_DISPATCHER_SIG)
+    target = r2.cmd(f"/x {p}").split()[0]  # Find byte pattern
+
+    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")
+
+    eprint(f"  Loaded switch cases")
+
+    ## STEP 2: Grab opcode offset
+
+    opcode_offset = get_opcode_offset(r2)
+    eprint(f"  Found opcode offset: {opcode_offset}")
+
+    r2.quit()
+
+    eprint(f"  Grabbed blocks from packet handler")
+
+    ## STEP 4: Process data
+    opcodes_db = dict()
+    switch_ea, packet_handler_switch = get_correct_switch(target, switch_cases)
+    eprint(f"  Found switch at {switch_ea}")
+
+    vtable_offset = 0x10
+    for data in packet_handler_switch.values():
+        if len(data["opcodes"]) > 10:
+            continue
+        opcodes_db[vtable_offset] = int(data["opcodes"][0]) + opcode_offset
+        vtable_offset += 0x8
+
+    eprint(f"  Loaded {len(opcodes_db)} cases from packet handler")
+    return opcodes_db
+
+
+def find_opcode_matches(old_opcodes_db, new_opcodes_db):
+    matches = []
+
+    for offset, old_opcode in old_opcodes_db.items():
+        if offset in new_opcodes_db:
+            matches.append((old_opcode, new_opcodes_db[offset]))
+
+    return matches
+
+
+def diff_exes(old_exe, new_exe):
+    old_opcodes_db = extract_opcode_data(old_exe)
+    new_opcodes_db = extract_opcode_data(new_exe)
+
+    if len(old_opcodes_db) != len(new_opcodes_db):
+        eprint(
+            f"ERROR: vtables have different sizes: {len(old_opcodes_db)} != {len(new_opcodes_db)}. Refusing to continue."
+        )
+        sys.exit(1)
+
+    opcodes_found = find_opcode_matches(old_opcodes_db, new_opcodes_db)
+    opcodes_object = []
+
+    for old, new in opcodes_found:
+        opcodes_object.append(
+            {
+                "old": [hex(old)],
+                "new": [hex(new)],
+            }
+        )
+
+    return opcodes_object
+
+
+@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)
+)
+def vtable_diff(old_exe, new_exe):
+    """
+    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.
+
+    The format of the output is a list (all fields are optional):
+
+    \b
+    [
+        {
+            "old": [opcode],
+            "new": [opcode],
+        },
+        ...
+    ]
+
+    Example:
+
+    python vtable_diff.py ffxiv_dx11.old.exe ffxiv_dx11.new.exe > diff.json
+    """
+    opcodes_object = diff_exes(old_exe, new_exe)
+    print(json.dumps(opcodes_object, indent=2))
+
+
+if __name__ == "__main__":
+    vtable_diff()