Probe GPS and traffic pathfinding hooks
This commit is contained in:
@@ -62,6 +62,12 @@ struct GameStates
|
||||
bool (*Add)(PluginHandle aHandle, uint32_t aType, GameState* aState);
|
||||
};
|
||||
|
||||
struct Hooking
|
||||
{
|
||||
bool (*Attach)(PluginHandle aHandle, void* aTarget, void* aDetour, void** aOriginal);
|
||||
bool (*Detach)(PluginHandle aHandle, void* aTarget);
|
||||
};
|
||||
|
||||
struct Scripts
|
||||
{
|
||||
bool (*Add)(PluginHandle aHandle, const wchar_t* aPath);
|
||||
@@ -75,7 +81,7 @@ struct Sdk
|
||||
{
|
||||
SemVer* runtime;
|
||||
Logger* logger;
|
||||
void* hooking;
|
||||
Hooking* hooking;
|
||||
GameStates* gameStates;
|
||||
Scripts* scripts;
|
||||
};
|
||||
@@ -84,6 +90,8 @@ struct Sdk
|
||||
namespace
|
||||
{
|
||||
constexpr uint32_t kGameStateRunning = 2;
|
||||
constexpr uintptr_t kGpsSystemTickRva = 0x0E6B62C;
|
||||
constexpr uintptr_t kTrafficSystemPathfindingRva = 0x0AD1234;
|
||||
|
||||
HMODULE gModule = nullptr;
|
||||
std::mutex gLogMutex;
|
||||
@@ -91,6 +99,12 @@ RED4ext::v1::PluginHandle gHandle = nullptr;
|
||||
const RED4ext::v1::Sdk* gSdk = nullptr;
|
||||
uint32_t gUpdateLogCount = 0;
|
||||
bool gScannedExecutable = false;
|
||||
uint32_t gGpsTickLogCount = 0;
|
||||
uint32_t gTrafficPathfindingLogCount = 0;
|
||||
|
||||
using ProbeFunc = void (*)(void* aThis, void* a2, void* a3, void* a4);
|
||||
ProbeFunc gOriginalGpsSystemTick = nullptr;
|
||||
ProbeFunc gOriginalTrafficSystemPathfinding = nullptr;
|
||||
|
||||
std::filesystem::path GetLogPath()
|
||||
{
|
||||
@@ -133,6 +147,14 @@ void LogRed4ext(std::string_view aMessage)
|
||||
}
|
||||
}
|
||||
|
||||
void LogHookCall(const char* aName, uint32_t aCount, void* aThis, void* a2, void* a3, void* a4)
|
||||
{
|
||||
std::ostringstream line;
|
||||
line << "hook " << aName << " call=" << aCount << " this=" << aThis << " a2=" << a2 << " a3=" << a3
|
||||
<< " a4=" << a4;
|
||||
LogRed4ext(line.str());
|
||||
}
|
||||
|
||||
const char* ReasonToString(RED4ext::v1::EMainReason aReason)
|
||||
{
|
||||
switch (aReason)
|
||||
@@ -290,6 +312,66 @@ void ScanExecutableStrings()
|
||||
LogRed4ext("exe-string scan end");
|
||||
}
|
||||
|
||||
void DetourGpsSystemTick(void* aThis, void* a2, void* a3, void* a4)
|
||||
{
|
||||
if (gGpsTickLogCount < 16)
|
||||
{
|
||||
LogHookCall("GPSSystem/Tick", gGpsTickLogCount, aThis, a2, a3, a4);
|
||||
++gGpsTickLogCount;
|
||||
}
|
||||
|
||||
if (gOriginalGpsSystemTick)
|
||||
{
|
||||
gOriginalGpsSystemTick(aThis, a2, a3, a4);
|
||||
}
|
||||
}
|
||||
|
||||
void DetourTrafficSystemPathfinding(void* aThis, void* a2, void* a3, void* a4)
|
||||
{
|
||||
if (gTrafficPathfindingLogCount < 16)
|
||||
{
|
||||
LogHookCall("TrafficSystem_Pathfinding", gTrafficPathfindingLogCount, aThis, a2, a3, a4);
|
||||
++gTrafficPathfindingLogCount;
|
||||
}
|
||||
|
||||
if (gOriginalTrafficSystemPathfinding)
|
||||
{
|
||||
gOriginalTrafficSystemPathfinding(aThis, a2, a3, a4);
|
||||
}
|
||||
}
|
||||
|
||||
void AttachProbeHook(const char* aName, uintptr_t aRva, void* aDetour, void** aOriginal)
|
||||
{
|
||||
if (!gSdk || !gSdk->hooking || !gSdk->hooking->Attach)
|
||||
{
|
||||
LogRed4ext("hook attach skipped: RED4ext hooking API unavailable");
|
||||
return;
|
||||
}
|
||||
|
||||
auto* target = reinterpret_cast<void*>(reinterpret_cast<uintptr_t>(GetModuleHandleW(nullptr)) + aRva);
|
||||
const auto attached = gSdk->hooking->Attach(gHandle, target, aDetour, aOriginal);
|
||||
|
||||
std::ostringstream line;
|
||||
line << "hook attach " << aName << " target=" << target << " rva=0x" << std::hex << aRva << std::dec
|
||||
<< " result=" << (attached ? "ok" : "failed") << " original=" << (aOriginal ? *aOriginal : nullptr);
|
||||
LogRed4ext(line.str());
|
||||
}
|
||||
|
||||
void DetachProbeHook(const char* aName, uintptr_t aRva)
|
||||
{
|
||||
if (!gSdk || !gSdk->hooking || !gSdk->hooking->Detach)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
auto* target = reinterpret_cast<void*>(reinterpret_cast<uintptr_t>(GetModuleHandleW(nullptr)) + aRva);
|
||||
const auto detached = gSdk->hooking->Detach(gHandle, target);
|
||||
|
||||
std::ostringstream line;
|
||||
line << "hook detach " << aName << " target=" << target << " result=" << (detached ? "ok" : "failed");
|
||||
LogRed4ext(line.str());
|
||||
}
|
||||
|
||||
bool OnRunningEnter(void* aApp)
|
||||
{
|
||||
std::ostringstream line;
|
||||
@@ -355,6 +437,18 @@ RED4EXT_C_EXPORT bool RED4EXT_CALL Main(RED4ext::v1::PluginHandle aHandle, RED4e
|
||||
{
|
||||
const auto added = aSdk->gameStates->Add(aHandle, kGameStateRunning, &gRunningState);
|
||||
LogRed4ext(added ? "registered Running game-state callbacks" : "failed to register Running game-state callbacks");
|
||||
|
||||
AttachProbeHook("GPSSystem/Tick", kGpsSystemTickRva, reinterpret_cast<void*>(&DetourGpsSystemTick),
|
||||
reinterpret_cast<void**>(&gOriginalGpsSystemTick));
|
||||
AttachProbeHook("TrafficSystem_Pathfinding", kTrafficSystemPathfindingRva,
|
||||
reinterpret_cast<void*>(&DetourTrafficSystemPathfinding),
|
||||
reinterpret_cast<void**>(&gOriginalTrafficSystemPathfinding));
|
||||
}
|
||||
|
||||
if (aReason == RED4ext::v1::EMainReason::Unload)
|
||||
{
|
||||
DetachProbeHook("GPSSystem/Tick", kGpsSystemTickRva);
|
||||
DetachProbeHook("TrafficSystem_Pathfinding", kTrafficSystemPathfindingRva);
|
||||
}
|
||||
|
||||
return true;
|
||||
|
||||
Executable
+136
@@ -0,0 +1,136 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Find likely x64 RIP-relative references to known string RVAs in a PE file."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import dataclasses
|
||||
import struct
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
@dataclasses.dataclass(frozen=True)
|
||||
class Section:
|
||||
name: str
|
||||
virtual_address: int
|
||||
virtual_size: int
|
||||
raw_offset: int
|
||||
raw_size: int
|
||||
characteristics: int
|
||||
|
||||
@property
|
||||
def is_code(self) -> bool:
|
||||
image_scn_cnt_code = 0x00000020
|
||||
image_scn_mem_execute = 0x20000000
|
||||
return bool(self.characteristics & image_scn_cnt_code) and bool(self.characteristics & image_scn_mem_execute)
|
||||
|
||||
def contains_rva(self, rva: int) -> bool:
|
||||
size = max(self.virtual_size, self.raw_size)
|
||||
return self.virtual_address <= rva < self.virtual_address + size
|
||||
|
||||
def rva_to_offset(self, rva: int) -> int:
|
||||
return self.raw_offset + (rva - self.virtual_address)
|
||||
|
||||
|
||||
def parse_pe(data: bytes) -> tuple[int, list[Section]]:
|
||||
if data[:2] != b"MZ":
|
||||
raise ValueError("not an MZ executable")
|
||||
|
||||
pe_offset = struct.unpack_from("<I", data, 0x3C)[0]
|
||||
if data[pe_offset : pe_offset + 4] != b"PE\0\0":
|
||||
raise ValueError("not a PE executable")
|
||||
|
||||
coff_offset = pe_offset + 4
|
||||
section_count = struct.unpack_from("<H", data, coff_offset + 2)[0]
|
||||
optional_size = struct.unpack_from("<H", data, coff_offset + 16)[0]
|
||||
optional_offset = coff_offset + 20
|
||||
magic = struct.unpack_from("<H", data, optional_offset)[0]
|
||||
if magic != 0x20B:
|
||||
raise ValueError("expected PE32+ executable")
|
||||
|
||||
image_base = struct.unpack_from("<Q", data, optional_offset + 24)[0]
|
||||
section_offset = optional_offset + optional_size
|
||||
|
||||
sections: list[Section] = []
|
||||
for index in range(section_count):
|
||||
offset = section_offset + index * 40
|
||||
name = data[offset : offset + 8].split(b"\0", 1)[0].decode("ascii", "replace")
|
||||
virtual_size, virtual_address, raw_size, raw_offset = struct.unpack_from("<IIII", data, offset + 8)
|
||||
characteristics = struct.unpack_from("<I", data, offset + 36)[0]
|
||||
sections.append(Section(name, virtual_address, virtual_size, raw_offset, raw_size, characteristics))
|
||||
|
||||
return image_base, sections
|
||||
|
||||
|
||||
def section_for_rva(sections: list[Section], rva: int) -> Section | None:
|
||||
for section in sections:
|
||||
if section.contains_rva(rva):
|
||||
return section
|
||||
return None
|
||||
|
||||
|
||||
def iter_rip_xrefs(data: bytes, section: Section, image_base: int, target_vas: set[int]) -> dict[int, list[int]]:
|
||||
start = section.raw_offset
|
||||
end = min(len(data), section.raw_offset + section.raw_size)
|
||||
hits: dict[int, list[int]] = {target_va: [] for target_va in target_vas}
|
||||
|
||||
# On x64, most string references are encoded as a signed 32-bit displacement
|
||||
# relative to the address immediately after the displacement.
|
||||
for offset in range(start, max(start, end - 4)):
|
||||
disp = struct.unpack_from("<i", data, offset)[0]
|
||||
instr_end_va = image_base + section.virtual_address + (offset - section.raw_offset) + 4
|
||||
target_va = instr_end_va + disp
|
||||
if target_va in target_vas:
|
||||
hits[target_va].append(section.virtual_address + (offset - section.raw_offset))
|
||||
|
||||
return hits
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("pe", type=Path)
|
||||
parser.add_argument("targets", nargs="+", help="name=rva_hex pairs")
|
||||
args = parser.parse_args()
|
||||
|
||||
data = args.pe.read_bytes()
|
||||
image_base, sections = parse_pe(data)
|
||||
code_sections = [section for section in sections if section.is_code]
|
||||
|
||||
targets: list[tuple[str, int, int]] = []
|
||||
target_vas: set[int] = set()
|
||||
for item in args.targets:
|
||||
name, _, rva_text = item.partition("=")
|
||||
if not name or not rva_text:
|
||||
raise ValueError(f"target must be name=rva_hex: {item}")
|
||||
target_rva = int(rva_text, 16)
|
||||
target_va = image_base + target_rva
|
||||
targets.append((name, target_rva, target_va))
|
||||
target_vas.add(target_va)
|
||||
|
||||
xrefs: dict[int, list[int]] = {target_va: [] for target_va in target_vas}
|
||||
for section in code_sections:
|
||||
section_hits = iter_rip_xrefs(data, section, image_base, target_vas)
|
||||
for target_va, hits in section_hits.items():
|
||||
xrefs[target_va].extend(hits)
|
||||
|
||||
print(f"image_base=0x{image_base:x}")
|
||||
for name, target_rva, target_va in targets:
|
||||
target_section = section_for_rva(sections, target_rva)
|
||||
section_name = target_section.name if target_section else "<none>"
|
||||
print(f"{name}: target_rva=0x{target_rva:x} target_section={section_name}")
|
||||
|
||||
all_hits = xrefs[target_va]
|
||||
if not all_hits:
|
||||
print(" no code xrefs found")
|
||||
continue
|
||||
|
||||
for hit in all_hits[:32]:
|
||||
print(f" code_xref_disp_rva=0x{hit:x} probable_instr_rva=0x{max(0, hit - 7):x}..0x{hit:x}")
|
||||
if len(all_hits) > 32:
|
||||
print(f" ... {len(all_hits) - 32} more")
|
||||
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user