Operation-Lovecraft-Fallen-doll/frida_hooks/capture_handshake.py

#!/usr/bin/env python3
"""
capture_handshake.py - Capture the full KEY_A/KEY_B handshake
Run this, then start/restart the game. It will attach once the process appears.
"""

import frida
import sys
import os
import time
import struct
import json

SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
JS_PATH = os.path.join(SCRIPT_DIR, 'capture_handshake.js')
PROCESS_NAME = 'Paralogue-Win64-Shipping.exe'
OUTPUT_DIR = os.path.join(SCRIPT_DIR, 'captures')
os.makedirs(OUTPUT_DIR, exist_ok=True)

packets = []
session = None


def on_message(message, data):
    if message['type'] == 'error':
        print(f'[!] {message["description"]}')
        return

    p = message.get('payload', {})
    if not isinstance(p, dict):
        return

    t = p.get('t', '')

    if t == 'ready':
        print(f'[*] {p["msg"]}')
        return

    if t in ('connect', 'discovered'):
        fd = p.get('fd', '?')
        ip = p.get('ip', '?')
        port = p.get('port', '?')
        print(f'[+] Game socket fd={fd} -> {ip}:{port}')
        return

    if t in ('recv', 'send') and data:
        raw = bytes(data)
        n = p.get('n', 0)
        direction = 'S->C' if t == 'recv' else 'C->S'
        ts = time.time()
        packets.append({'ts': ts, 'dir': direction, 'data': raw, 'n': n, 'req': p.get('req', 0)})

        # Decode if it looks like a handshake packet (starts with 00 08)
        label = ''
        if len(raw) >= 10 and raw[0] == 0x00 and raw[1] == 0x08:
            # Protobuf envelope
            hdr = raw[2:10]
            fields = {}
            pos = 0
            while pos < len(hdr):
                byte = hdr[pos]
                fn = byte >> 3
                wt = byte & 7
                pos += 1
                if wt == 0:
                    val = 0; shift = 0
                    while pos < len(hdr):
                        b = hdr[pos]; val |= (b & 0x7f) << shift; shift += 7; pos += 1
                        if (b & 0x80) == 0: break
                    fields[fn] = val
            msg_type = fields.get(2, 0)
            names = {1001: 'KEY_A', 1002: 'KEY_B'}
            label = f' [{names.get(msg_type, f"type_{msg_type}")} route={fields.get(1,0)} f4={fields.get(4,0)}]'

        # Check for protobuf-like content (might be decrypted)
        if not label and len(raw) > 4:
            # Check if first byte is a valid protobuf field tag
            tag = raw[0]
            fn = tag >> 3
            wt = tag & 7
            if fn > 0 and fn < 100 and wt in (0, 2):
                label = f' [maybe protobuf: field{fn} wire{wt}]'

        sz = len(raw)
        hex_preview = raw[:32].hex()
        if sz > 32:
            hex_preview += '...'
        print(f'[#{n:04d}] {direction} {sz:5d}B{label}')
        print(f'        {hex_preview}')

        # Save incrementally
        capfile = os.path.join(OUTPUT_DIR, f'handshake_{int(ts)}.jsonl')
        with open(capfile, 'a') as f:
            entry = {'n': n, 'dir': direction, 'ts': ts, 'sz': sz, 'hex': raw.hex()}
            if label:
                entry['label'] = label.strip(' []')
            f.write(json.dumps(entry) + '\n')


def wait_for_process():
    print(f'[*] Waiting for {PROCESS_NAME}...')
    while True:
        try:
            device = frida.get_local_device()
            for proc in device.enumerate_processes():
                if proc.name == PROCESS_NAME:
                    return proc.pid
        except:
            pass
        time.sleep(1)


def main():
    global session

    print('=' * 60)
    print('  Handshake Capture Tool')
    print('  Close the game, then start it. This tool will auto-attach.')
    print('=' * 60)

    with open(JS_PATH, 'r') as f:
        js_code = f.read()

    # Check if already running
    try:
        device = frida.get_local_device()
        for proc in device.enumerate_processes():
            if proc.name == PROCESS_NAME:
                print(f'[!] Game already running (PID {proc.pid}). Attaching...')
                print(f'[!] Note: handshake may already be done. Restart game for full capture.')
                session = frida.attach(proc.pid)
                break
    except:
        pass

    if not session:
        pid = wait_for_process()
        print(f'[*] Found {PROCESS_NAME} (PID {pid}), attaching...')
        # Small delay to let the process initialize
        time.sleep(2)
        session = frida.attach(pid)

    print(f'[*] Attached. Installing hooks...')
    script = session.create_script(js_code)
    script.on('message', on_message)
    script.load()

    print('[*] Press Ctrl+C to stop.\n')

    try:
        while True:
            time.sleep(0.5)
    except KeyboardInterrupt:
        pass

    print(f'\n[*] Captured {len(packets)} packets.')

    # Save summary
    if packets:
        summary_path = os.path.join(OUTPUT_DIR, 'handshake_summary.txt')
        with open(summary_path, 'w') as f:
            for pkt in packets:
                f.write(f'{pkt["dir"]} {pkt["n"]:4d} {len(pkt["data"]):5d}B {pkt["data"][:32].hex()}\n')
        print(f'[*] Summary saved to {summary_path}')

        # Save raw binary capture
        bin_path = os.path.join(OUTPUT_DIR, 'handshake.bin')
        with open(bin_path, 'wb') as f:
            for pkt in packets:
                d = 0 if pkt['dir'] == 'C->S' else 1
                f.write(struct.pack('<dBI', pkt['ts'], d, len(pkt['data'])))
                f.write(pkt['data'])
        print(f'[*] Binary capture saved to {bin_path}')

    session.detach()
    print('[*] Done.')


if __name__ == '__main__':
    main()