use patch generator to build patchersHEADmaster

3 files changed, 526 insertions, 0 deletions

diff --git a/src/steamvr_patches/__init__.py b/src/steamvr_patches/__init__.py
new file mode 100644
index 0000000..be3b696
--- /dev/null
+++ b/src/steamvr_patches/__init__.py
@@ -0,0 +1,125 @@
+# SPDX-FileCopyrightText: 2026 Tomasz Kramkowski <tomasz@kramkow.ski>
+# SPDX-License-Identifier: GPL-3.0-or-later
+
+import argparse
+import hashlib
+import sys
+import tempfile
+import tomllib
+from base64 import b85encode
+from importlib.resources import files
+from io import BytesIO
+from json import dumps
+from pathlib import Path
+from typing import Annotated, BinaryIO, Self
+from zlib import compress
+
+from pydantic import BaseModel, ConfigDict, Field, field_validator
+
+from .elf import Hook, compile_hooks
+
+NonnegativeInt = Annotated[int, Field(strict=True, ge=0)]
+
+
+class Compilation(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    sources: list[Path] = Field(min_length=1)
+    imports: dict[str, NonnegativeInt] = Field(default_factory=dict)
+    hooks: list[Hook] = Field(min_length=1)
+
+
+class FilePatch(BaseModel):
+    model_config = ConfigDict(val_json_bytes="hex", extra="forbid")
+
+    path: Path
+    sha256: bytes = Field(min_length=32, max_length=32)
+    edits: list[tuple[NonnegativeInt, bytes]]
+    compiled: Compilation | None = None
+
+
+class PatchSet(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    target: str
+    targets: list[FilePatch]
+
+    @field_validator("targets")
+    @classmethod
+    def unique_paths(cls, targets: list[FilePatch]) -> list[FilePatch]:
+        paths = [target.path for target in targets]
+        if len(set(paths)) != len(paths):
+            raise ValueError("duplicate patch target")
+        return targets
+
+    @classmethod
+    def parse(cls, config: dict, directory: Path) -> Self:
+        result = cls.model_validate(config)
+        for target in result.targets:
+            if target.compiled is not None:
+                target.compiled.sources = [
+                    directory / source for source in target.compiled.sources
+                ]
+        return result
+
+
+def compile_patch(config: FilePatch, source: BinaryIO, output: Path) -> dict:
+    source.seek(0)
+    data = source.read()
+    if hashlib.sha256(data).digest() != config.sha256:
+        raise ValueError(f"input SHA-256 mismatch: {config.path}")
+    edits = list(config.edits)
+    if config.compiled is not None:
+        compiled = config.compiled
+        edits += compile_hooks(
+            source, compiled.sources, compiled.imports, compiled.hooks, output
+        )
+    end = 0
+    for offset, replacement in sorted(edits):
+        if offset < end:
+            raise ValueError("overlapping edits")
+        end = offset + len(replacement)
+    patched = BytesIO(data)
+    for offset, replacement in edits:
+        patched.seek(offset)
+        patched.write(replacement)
+    return {
+        "original_hash": config.sha256,
+        "patched_hash": hashlib.sha256(patched.getbuffer()).digest(),
+        "edits": edits,
+    }
+
+
+def render(patches: dict[str, dict]) -> str:
+    template = files("steamvr_patches").joinpath("template.py").read_text()
+    placeholder = "PATCH_DATA = {}"
+    if template.count(placeholder) != 1:
+        raise ValueError("missing or duplicate runtime patch-table placeholder")
+    data = dumps(patches, default=bytes.hex, separators=(",", ":")).encode()
+    encoded = b85encode(compress(data, level=9))
+    return template.replace(placeholder, f"PATCH_DATA = decode_patch_data({encoded!r})")
+
+
+def generate(patch_data: Path, input_root: Path) -> str:
+    with (patch_data / "patches.toml").open("rb") as source:
+        definition = PatchSet.parse(tomllib.load(source), patch_data)
+    patches = {}
+    with tempfile.TemporaryDirectory(prefix="generate-patcher-") as temporary:
+        for index, config in enumerate(definition.targets):
+            with (input_root / config.path).open("rb") as source:
+                patches[config.path.as_posix()] = compile_patch(
+                    config, source, Path(temporary) / str(index)
+                )
+    return render(patches)
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser()
+    parser.add_argument(
+        "patch_data", type=Path, help="directory containing patches.toml etc"
+    )
+    parser.add_argument(
+        "input_root", type=Path, help="directory containing the unpatched target files"
+    )
+    args = parser.parse_args()
+    sys.stdout.write(generate(args.patch_data, args.input_root))
diff --git a/src/steamvr_patches/elf.py b/src/steamvr_patches/elf.py
new file mode 100644
index 0000000..e69c567
--- /dev/null
+++ b/src/steamvr_patches/elf.py
@@ -0,0 +1,232 @@
+# SPDX-FileCopyrightText: 2026 Tomasz Kramkowski <tomasz@kramkow.ski>
+# SPDX-License-Identifier: GPL-3.0-or-later
+
+"""Compile C/assembly hooks into one Linux x86-64 ELF patch payload."""
+
+import struct
+import subprocess
+from pathlib import Path
+from typing import BinaryIO
+
+from capstone import CS_ARCH_X86, CS_MODE_64, Cs
+from elftools.construct import Container
+from elftools.elf.constants import P_FLAGS
+from elftools.elf.elffile import ELFFile
+from elftools.elf.relocation import RelocationSection
+from elftools.elf.sections import SymbolTableSection
+from pydantic import BaseModel, ConfigDict, Field
+
+
+class Hook(BaseModel):
+    model_config = ConfigDict(strict=True, frozen=True, extra="forbid")
+
+    address: int = Field(ge=0)
+    length: int = Field(ge=5)
+    target: str = Field(min_length=1)
+
+
+class CompiledPayload:
+    """Position-independent code with resolved entry addresses inside it."""
+
+    code: bytes
+    entries: dict[str, int]
+
+    def __init__(self, source: BinaryIO, targets: list[str]):
+        elf = ELFFile(source)
+        for section in elf.iter_sections():
+            if (
+                section["sh_flags"] & 2
+                and section["sh_size"]
+                and section.name != ".payload"
+            ):
+                raise ValueError(f"unsupported allocated section: {section.name}")
+            # PC32 (2) PLT32 (4) (PC-relative)
+            if isinstance(section, RelocationSection):
+                if any(
+                    r["r_info_type"] not in (2, 4) for r in section.iter_relocations()
+                ):
+                    raise ValueError("helper contains a non-relative relocation")
+        payload = elf.get_section_by_name(".payload")
+        if payload is None:
+            raise ValueError("missing compiled payload")
+        self.code = payload.data()
+        symbols = elf.get_section_by_name(".symtab")
+        if not isinstance(symbols, SymbolTableSection):
+            raise ValueError("missing compiled symbol table")
+        payload_index = elf.get_section_index(".payload")
+        address = payload["sh_addr"]
+        self.entries = {}
+        for target in targets:
+            matches = symbols.get_symbol_by_name(target) or []
+            if len(matches) != 1 or matches[0]["st_shndx"] != payload_index:
+                raise ValueError(
+                    f"hook target is not a unique payload symbol: {target}"
+                )
+            entry = matches[0]["st_value"]
+            if not address <= entry < address + len(self.code):
+                raise ValueError(
+                    f"hook target is outside the compiled payload: {target}"
+                )
+            self.entries[target] = entry
+
+
+def align(value: int, boundary: int) -> int:
+    return (value + boundary - 1) & -boundary
+
+
+def run(*args: str) -> str:
+    return subprocess.check_output(args, text=True)
+
+
+def resolve_hook_sites(elf: ELFFile, hooks: list[Hook]) -> list[tuple[Hook, int]]:
+    """Find file offsets and verify that each hook covers whole instructions."""
+    source = elf.stream
+    sites = []
+    disassembler = Cs(CS_ARCH_X86, CS_MODE_64)
+    for hook in hooks:
+        offset = next(elf.address_offsets(hook.address, hook.length))
+        source.seek(offset)
+        instructions = disassembler.disasm(source.read(hook.length), hook.address)
+        if sum(i.size for i in instructions) != hook.length:
+            raise ValueError("hook length cuts an instruction or contains invalid code")
+        sites.append((hook, offset))
+    return sites
+
+
+def compile_code(
+    sources: list[Path],
+    imports: dict[str, int],
+    targets: list[str],
+    code_va: int,
+    output: Path,
+) -> CompiledPayload:
+    """Compile and link the patch sources at their destination address."""
+    output.mkdir(parents=True, exist_ok=False)
+    objects = []
+    flags = [
+        "-Os",
+        "-ffreestanding",
+        "-fPIE",
+        "-fvisibility=hidden",
+        "-fno-stack-protector",
+        "-fno-strict-aliasing",
+        "-mno-red-zone",
+        "-fno-asynchronous-unwind-tables",
+        "-fno-unwind-tables",
+        "-fcf-protection=none",
+        "-Wall",
+        "-Wextra",
+    ]
+    for number, source_path in enumerate(sources):
+        obj = output / f"{number}.o"
+        source_flags = ["-std=c17"] if source_path.suffix == ".c" else []
+        run("cc", *flags, *source_flags, "-c", str(source_path), "-o", str(obj))
+        objects.append(str(obj))
+
+    linker_script = output / "layout.ld"
+    linker_script.write_text(
+        "PHDRS { code PT_LOAD FLAGS(5); }\n"  # PF_R | PF_X: readable/executable.
+        f"SECTIONS {{ . = {code_va:#x};\n"
+        " .payload : { *(.text.entry) *(.text*) *(.rodata*) } :code\n"
+        " /DISCARD/ : { *(.comment) *(.note.GNU-stack) }\n}\n"
+    )
+    import_args = [f"--defsym={name}={value:#x}" for name, value in imports.items()]
+    linked = output / "helper.elf"
+    run(
+        "ld",
+        "--build-id=none",
+        "--emit-relocs",
+        f"--script={linker_script}",
+        *import_args,
+        f"--output={linked}",
+        *objects,
+    )
+    with linked.open("rb") as compiled:
+        return CompiledPayload(compiled, targets)
+
+
+def make_hook_edits(
+    sites: list[tuple[Hook, int]], entries: dict[str, int]
+) -> list[tuple[int, bytes]]:
+    """Redirect each hook site to its compiled entry with a relative jump."""
+    edits = []
+    for hook, offset in sites:
+        displacement = entries[hook.target] - hook.address - 5
+        if not -(1 << 31) <= displacement < 1 << 31:
+            raise ValueError(
+                f"hook target is outside relative jump range: {hook.target}"
+            )
+        jump = b"\xe9" + struct.pack("<i", displacement) + b"\x90" * (hook.length - 5)
+        edits.append((offset, jump))
+    return edits
+
+
+def compile_segment(
+    elf: ELFFile,
+    sources: list[Path],
+    imports: dict[str, int],
+    targets: list[str],
+    output: Path,
+) -> tuple[CompiledPayload, list[tuple[int, bytes]]]:
+    """Place the compiled code and enlarged program-header table in a new segment."""
+    phdr = elf.structs.Elf_Phdr
+    headers = [segment.header for segment in elf.iter_segments()]
+    # Place the new segment after all existing loadable mappings.
+    load_end = max(
+        header["p_vaddr"] + header["p_memsz"]
+        for header in headers
+        if header["p_type"] == "PT_LOAD"
+    )
+    segment_va = align(load_end, 4096)
+    segment_offset = align(elf.stream_len, 4096)
+    table_size = (len(headers) + 1) * phdr.sizeof()
+    code_delta = align(table_size, 16)
+    code_va = segment_va + code_delta
+    compiled = compile_code(sources, imports, targets, code_va, output)
+
+    # Relocate the enlarged program-header table into the new segment.
+    table_header = next(header for header in headers if header["p_type"] == "PT_PHDR")
+    table_header.update(
+        p_flags=P_FLAGS.PF_R,
+        p_offset=segment_offset,
+        p_vaddr=segment_va,
+        p_paddr=segment_va,
+        p_filesz=table_size,
+        p_memsz=table_size,
+        p_align=8,
+    )
+    segment_size = code_delta + len(compiled.code)
+    headers.append(
+        Container(
+            p_type="PT_LOAD",
+            p_flags=P_FLAGS.PF_R | P_FLAGS.PF_X,
+            p_offset=segment_offset,
+            p_vaddr=segment_va,
+            p_paddr=segment_va,
+            p_filesz=segment_size,
+            p_memsz=segment_size,
+            p_align=4096,
+        )
+    )
+    edits = [
+        (0x20, struct.pack("<Q", segment_offset)),  # Elf64_Ehdr.e_phoff
+        (0x38, struct.pack("<H", len(headers))),  # Elf64_Ehdr.e_phnum
+        (segment_offset, b"".join(phdr.build(header) for header in headers)),
+        (segment_offset + code_delta, compiled.code),
+    ]
+    return compiled, edits
+
+
+def compile_hooks(
+    source: BinaryIO,
+    sources: list[Path],
+    imports: dict[str, int],
+    hooks: list[Hook],
+    output: Path,
+) -> list[tuple[int, bytes]]:
+    elf = ELFFile(source)
+    sites = resolve_hook_sites(elf, hooks)
+    compiled, segment_edits = compile_segment(
+        elf, sources, imports, [hook.target for hook in hooks], output
+    )
+    return make_hook_edits(sites, compiled.entries) + segment_edits
diff --git a/src/steamvr_patches/template.py b/src/steamvr_patches/template.py
new file mode 100644
index 0000000..4f88b10
--- /dev/null
+++ b/src/steamvr_patches/template.py
@@ -0,0 +1,169 @@
+#!/usr/bin/env python3
+# SPDX-FileCopyrightText: 2026 Tomasz Kramkowski <tomasz@kramkow.ski>
+# SPDX-License-Identifier: CC0-1.0
+
+"""Patching wrapper script.
+
+This file is generated.
+
+See https://kramkow.ski/project/steamvr-patches.html and
+https://the-tk.com/cgit/steamvr-patches/about/ for more information.
+
+Use at your own risk!
+
+
+Installation (where <thing> is the thing this patcher targets, e.g. steamvr):
+
+1. Put this `patch-<thing>` script somewhere, e.g. `~/.local/libexec/patch-<thing>`
+2. Mark it executable
+3. Change the launch options of the specified <thing> to:
+   `/absolute/path/to/patch-<thing> %command%`.
+   If you already have something in there, I'm sure you can figure it out.
+4. Launch <thing> normally. You should see a desktop notification with some
+   diagnostics which specify if the script actually did/tried to do something.
+"""
+
+import hashlib
+import os
+import stat
+import subprocess
+import sys
+import tempfile
+from base64 import b85decode
+from dataclasses import dataclass
+from io import BytesIO
+from json import loads as json_loads
+from pathlib import Path
+from zlib import decompress
+
+NAME = Path(__file__).name
+MARKER = b"Valve, please fix."
+
+
+@dataclass(frozen=True)
+class Edit:
+    offset: int
+    replacement: bytes
+
+
+@dataclass(frozen=True)
+class Patch:
+    original_hash: bytes
+    patched_hash: bytes
+    edits: tuple[Edit, ...]
+
+    @classmethod
+    def from_dict(cls, data: dict) -> "Patch":
+        return cls(
+            bytes.fromhex(data["original_hash"]),
+            bytes.fromhex(data["patched_hash"]),
+            tuple(
+                Edit(offset, bytes.fromhex(replacement))
+                for offset, replacement in data["edits"]
+            ),
+        )
+
+    def apply_to(self, data: bytes) -> bytes:
+        payload = data.removesuffix(MARKER)
+        digest = sha256(payload)
+        if digest == self.patched_hash:
+            return payload + MARKER
+        if digest != self.original_hash:
+            if payload != data:
+                raise LookupError(
+                    "already patched with another set of patches: "
+                    "verify file integrity and restart"
+                )
+            raise LookupError("unrecognized file hash: this wrapper may need updating")
+        result = BytesIO(payload)
+        for edit in self.edits:
+            result.seek(edit.offset)
+            result.write(edit.replacement)
+        patched = result.getvalue()
+        if sha256(patched) != self.patched_hash:
+            raise ValueError("patched file checksum mismatch")
+        return patched + MARKER
+
+
+def decode_patch_data(data: bytes) -> dict:
+    return json_loads(decompress(b85decode(data)))
+
+
+# Just to keep the file small, not for obfuscation.
+PATCH_DATA = {}
+PATCHES: dict[str, Patch] = {
+    path: Patch.from_dict(data) for path, data in PATCH_DATA.items()
+}
+
+
+def sha256(data):
+    return hashlib.sha256(data).digest()
+
+
+def notify(message):
+    print(f"{NAME}: {message}", file=sys.stderr)
+    try:
+        subprocess.run(
+            ["notify-send", NAME, message],
+            check=False,
+            timeout=3,
+            stdout=subprocess.DEVNULL,
+            stderr=subprocess.DEVNULL,
+        )
+    except (OSError, subprocess.TimeoutExpired):
+        pass
+
+
+def replace(path, data):
+    mode = stat.S_IMODE(path.stat().st_mode)
+    with tempfile.NamedTemporaryFile(
+        dir=path.parent, prefix=path.name + ".", delete=False
+    ) as f:
+        temporary = Path(f.name)
+        try:
+            f.write(data)
+            f.flush()
+            os.fchmod(f.fileno(), mode)
+            os.fsync(f.fileno())
+            os.replace(temporary, path)
+        finally:
+            temporary.unlink(missing_ok=True)
+
+
+def main(command):
+    diagnostics = []
+    if not command:
+        print(f"Usage: {NAME} command [args...]", file=sys.stderr)
+        return 2
+    try:
+        root = Path.cwd()
+        pending = []
+        for relative, patch in PATCHES.items():
+            path = (root / relative).resolve()
+            try:
+                original = path.read_bytes()
+            except FileNotFoundError as e:
+                diagnostics.append(f"Skipped `{relative}`: {e}")
+                continue
+            try:
+                updated = patch.apply_to(original)
+            except ValueError as e:
+                raise ValueError(f"Patching `{relative}` failed: {e}") from e
+            except LookupError as e:
+                diagnostics.append(f"Skipped `{relative}`: {e}")
+                continue
+            if updated != original:
+                pending.append((path, updated))
+        for path, updated in pending:
+            replace(path, updated)
+            diagnostics.append(f"Patched {path}")
+        if diagnostics:
+            notify("\n".join(diagnostics))
+    except (OSError, ValueError) as e:
+        notify("\n".join([f"Encountered an error: {e}", *diagnostics]))
+        return 1
+    os.execvp(command[0], command)
+
+
+if __name__ == "__main__":
+    sys.exit(main(sys.argv[1:]))