diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/steamvr_patches/__init__.py | 125 | ||||
| -rw-r--r-- | src/steamvr_patches/elf.py | 232 | ||||
| -rw-r--r-- | src/steamvr_patches/template.py | 169 |
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:])) |