Add opcode diff applier web page

.github/workflows/minor_patch_diff.yml

@@ -1,81 +0,0 @@
-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

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

LICENSE

@@ -1,21 +0,0 @@
-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

@@ -1,94 +0,0 @@
-# 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

@@ -1,202 +0,0 @@
-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

@@ -1,21 +0,0 @@
-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

@@ -1,9 +0,0 @@
-# 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

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

asm2vec/datatype.py

@@ -1,205 +0,0 @@
-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

@@ -1,88 +0,0 @@
-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

@@ -1,348 +0,0 @@
-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

@@ -1,391 +0,0 @@
-#!/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

@@ -1,287 +0,0 @@
-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

@@ -1,178 +0,0 @@
-[
-  {
-    "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

@@ -1,67 +0,0 @@
-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

@@ -1,5 +0,0 @@
-# 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

@@ -1,87 +0,0 @@
-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

@@ -1,250 +0,0 @@
-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

@@ -1,43 +0,0 @@
-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

@@ -1,101 +0,0 @@
-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

@@ -1,187 +0,0 @@
-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

@@ -1,191 +0,0 @@
-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

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

requirements.txt

@@ -1,14 +0,0 @@
-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

@@ -1,63 +0,0 @@
-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

@@ -1,56 +0,0 @@
-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

@@ -1,375 +0,0 @@
-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

@@ -1,159 +0,0 @@
-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()