Two Error Codes and a Fuse: Hunting Exploitable Bugs on a Quest 2

#!/usr/bin/env python3
"""
quest2_hunt.py — one script for the whole Quest 2 bug hunt.

Runs the read-only checks used in the write-up "Two Error Codes and a Fuse", in the
order the investigation actually needed them, and keeps every result in one JSON
report so the post can cite its own output.

It also fetches and parses the firmware artefacts the checks depend on:
  * payload.bin  (the OTA)  -> size/hash verified, partitions extracted when possible
  * abl.img      (bootloader) -> LZMA/FV unpack, X.509 trust anchors, fastboot/oem tables
  * vbmeta.img   (AVB)      -> header, algorithm, rollback index

Design rules
  * standard library only (Python 3.9+), no pip installs
  * read-only by default; state-changing actions need an explicit flag and print a warning
  * anything missing (adb, fastboot, device, cross-compiler) is *skipped*, never fatal
  * --dry-run prints the plan and touches nothing

Typical use
    ./quest2_hunt.py all                        # everything that this machine can do
    ./quest2_hunt.py fingerprint --json out.json
    ./quest2_hunt.py bootloader --parse-log fastboot-oem.log
    ./quest2_hunt.py abl --img abl.img --out abl.json
    ./quest2_hunt.py ota --url https://host/Quest2/payload.bin --partitions abl,vbmeta

Author: KBS <admin@dtrh.net>   Target: Meta Quest 2 (hollywood), own hardware.
"""

from __future__ import annotations

import argparse
import base64
import hashlib
import json
import lzma
import os
import re
import shutil
import struct
import subprocess
import sys
import tempfile
import textwrap
import time
import urllib.request
from datetime import datetime, timezone

VERSION = "1.1"
REPORT_DEFAULT = "quest2_report.json"

# --------------------------------------------------------------------------------------
# constants measured on build 52322490019800150 / kernel 4.19.325-cip132-st16
# --------------------------------------------------------------------------------------

KGSL = "/dev/kgsl-3d0"
BINDER = "/dev/binder"

# _IOC numbers (direction<<30 | size<<16 | type<<8 | nr)
IOCTL_DRAWCTXT_CREATE = 0xC0080913   # _IOWR(0x09, 0x13, struct kgsl_drawctxt_create)  8 bytes
IOCTL_DRAWCTXT_DESTROY = 0xC0080914  # _IOWR(0x09, 0x14, struct kgsl_drawctxt_destroy) 8 bytes
IOCTL_GPUMEM_ALLOC_ID = 0xC0200934   # _IOWR(0x09, 0x34, struct kgsl_gpumem_alloc_id)  32 bytes
IOCTL_GPUMEM_FREE_ID = 0xC0080935    # _IOWR(0x09, 0x35, struct kgsl_gpumem_free_id)   8 bytes
IOCTL_GPU_COMMAND = 0xC040094A       # _IOWR(0x09, 0x4a, struct kgsl_gpu_command)       64 bytes
BINDER_GET_NODE_DEBUG_INFO = 0xC018620B  # _IOWR('b', 11, struct binder_node_debug_info)

# the two mandatory bits an a6xx draw context must carry
KGSL_CONTEXT_NO_GMEM_ALLOC = 0x2
KGSL_CONTEXT_PREAMBLE = 0x10
MANDATORY_CONTEXT_FLAGS = KGSL_CONTEXT_NO_GMEM_ALLOC | KGSL_CONTEXT_PREAMBLE  # 0x12

# deliberately impossible ioctl numbers: dispatch misses, so the syscall layer must
# answer ENOTTY. An EACCES here proves the denial happens before the driver runs.
CONTROL_IOCTLS = (0xC040097E, 0xC0400900, 0xC04009FF)

# flags in the a6xx IB control word (bits 31:12) — the privilege bit is not reachable
# from a 20-bit field, which is why user IBs cannot borrow IOMMU_PRIV.

SOCKET_FAMILIES = {
    "AF_UNIX": (1, 1, 0),
    "AF_INET_TCP": (2, 1, 0),
    "AF_INET_UDP": (2, 2, 0),
    "AF_INET6_TCP": (10, 1, 0),
    "AF_NETLINK_ROUTE": (16, 3, 0),
    "AF_NETLINK_SELINUX": (16, 3, 7),
    "AF_PACKET": (17, 3, 0),
    "AF_ALG": (38, 1, 0),
    "AF_SCTP_TCP_PROTO": (2, 1, 132),   # family 2 + IPPROTO_SCTP: the test that actually matters
    "AF_VSOCK": (40, 1, 0),
    "AF_XDP": (44, 3, 0),
    "AF_RDS": (21, 5, 0),
}

SECURITY_FLAGS = [
    ("oculus_shared_os_services_sessionless", "skip_notification_permission_check"),
    ("hzos_system_sessionless", "hzos_catapult_disable_device_ban_enforcement"),
    ("hzos_system_sessionless", "oc_inauthenticity_enforcement"),
    ("hzos_system_sessionless", "inauth_enforcement"),
    ("hzos_system_sessionless", "attestation_connectivity_check_killswitch"),
    ("hzos_system_sessionless", "device_access_enable_uat_cache"),
    ("oculus_shared_os_services_sessionless", "hzos_security_zygote_partitioning_policy"),
    ("perception", "hzos.perception.access.flags.perception_access_policy_enabled"),
    ("oculus_shared_core", "is_trusted_user"),
    ("oculus_shared_horizon", "enable_sps_client_auth"),
]

/* quest2_probe.c — the ioctl-level core, cross-compiled and run from /data/local/tmp.
 * Modes: kgsl | binder | devs | fam
 * Read-only. kgsl creates and destroys a context; no PM4 is submitted.
 */
#define _GNU_SOURCE
#include <errno.h>
#include <fcntl.h>
#include <stdio.h>
#include <string.h>
#include <sys/ioctl.h>
#include <sys/mman.h>
#include <sys/socket.h>
#include <sys/un.h>
#include <unistd.h>

struct kgsl_drawctxt_create  { unsigned int flags, ctx_id; };
struct kgsl_drawctxt_destroy { unsigned int ctx_id, pad; };
struct kgsl_gpu_command {
    unsigned long flags; unsigned long cmdlist; unsigned int cmdsize, numcmds;
    unsigned long objlist; unsigned int objsize, numobjs;
    unsigned long synclist; unsigned int syncsize, numsyncs;
    unsigned int context_id, timestamp;
};
struct node_debug_info { unsigned long ptr, cookie; unsigned int strong, weak; };

static void print_ioctl(const char *label, unsigned long cmd){
    int fd = open("/dev/kgsl-3d0", O_RDWR);
    if (fd < 0) { printf("%s\topen_errno=%d\n", label, errno); return; }
    unsigned char buf[64]; memset(buf, 0, sizeof buf);
    errno = 0; long r = ioctl(fd, cmd, buf);
    printf("%s\tcmd=0x%08lx\tr=%ld\terrno=%d\n", label, cmd, r, errno);
    close(fd);
}

static int mode_kgsl(void){
    int fd = open("/dev/kgsl-3d0", O_RDWR);
    if (fd < 0) { printf("open /dev/kgsl-3d0 errno=%d\n", errno); return 1; }
    printf("open /dev/kgsl-3d0\tok\n");

    /* the flag matrix: 0x12 is the two mandatory bits */
    unsigned int try[] = {0, 0x00100000, 0x00200000, 0x00000200, 0x12};
    for (unsigned i = 0; i < sizeof try / sizeof try[0]; i++) {
        struct kgsl_drawctxt_create c = { try[i], 0 };
        errno = 0; long r = ioctl(fd, 0xC0080913UL, &c);
        printf("drawctxt_create flags=0x%08x\tr=%ld\terrno=%d\tid=%u\n", try[i], r, errno, c.ctx_id);
    }
    /* impossible numbers: ENOTTY expected, so EACCES means "policy, not driver" */
    for (unsigned i = 0; i < 3; i++)
        print_ioctl("control_unregistered", (unsigned long[]){0xC040097EU, 0xC0400900U, 0xC04009FFU}[i]);

    /* empty submit: proves the modern path is admitted */
    struct kgsl_gpu_command g; memset(&g, 0, sizeof g);
    g.cmdsize = 32; g.objsize = 0;
    errno = 0; long r = ioctl(fd, 0xC040094AU, &g);
    printf("gpu_command numcmds=0\tr=%ld\terrno=%d\n", r, errno);

    struct kgsl_drawctxt_destroy d = {0, 0};
    errno = 0; r = ioctl(fd, 0xC0080914UL, &d);
    printf("drawctxt_destroy\tr=%ld\terrno=%d\n", r, errno);
    close(fd);
    return 0;
}

static int mode_binder(void){
    int fd = open("/dev/binder", O_RDONLY);
    if (fd < 0) { printf("open /dev/binder errno=%d\n", errno); return 1; }
    printf("open /dev/binder\tok (read-only: PROT_WRITE mmap is EPERM for shell)\n");
    struct node_debug_info n; memset(&n, 0, sizeof n);
    errno = 0; long r = ioctl(fd, 0xC018620BUL, &n);
    printf("BINDER_GET_NODE_DEBUG_INFO\tr=%ld\terrno=%d\tptr=0x%lx strong=%u weak=%u\n",
           r, errno, n.ptr, n.strong, n.weak);
    close(fd);
    return 0;
}

static int mode_devs(void){
    const char *names[] = {"/dev/kgsl-3d0","/dev/binder","/dev/hwbinder","/dev/vndbinder",
        "/dev/ion","/dev/ashmem","/dev/uinput","/dev/uhid","/dev/input/event0",
        "/dev/cvp","/dev/video32","/dev/msm_npu","/dev/adsprpc-smd","/dev/spcom",
        "/dev/smcinvoke","/dev/qseecom","/dev/diag","/dev/tun","/dev/dma_heap/system",0};
    for (int i = 0; names[i]; i++) {
        int fd = open(names[i], O_RDONLY);
        printf("%-24s\t%s\terrno=%d\n", names[i], fd >= 0 ? "OPEN" : "denied", fd >= 0 ? 0 : errno);
        if (fd >= 0) close(fd);
    }
    return 0;
}

static int mode_fam(void){
    const char *n[] = {"AF_UNIX","AF_INET_TCP","AF_INET_UDP","AF_INET6_TCP","NETLINK_ROUTE",
        "NETLINK_SELINUX","AF_PACKET","AF_ALG","AF_INET+SCTP","AF_VSOCK","AF_XDP",0};
    int f[] = {1,2,2,10,16,16,17,38,2,40,44};
    int t[] = {1,1,2,1,3,3,3,1,1,1,3};
    int p[] = {0,0,0,0,0,7,0,0,132,0,0};
    for (int i = 0; n[i]; i++) {
        int s = socket(f[i], t[i], p[i]);
        printf("%-16s\t%s\terrno=%d\n", n[i], s >= 0 ? "OK" : "denied", s >= 0 ? 0 : errno);
        if (s >= 0) close(s);
    }
    return 0;
}

/* /dev/socket reachability, with two controls that make the map trustworthy:
 *  (a) a bind() into /data/local/tmp must fail EACCES (shell_data_file) or the
 *      probe is not actually running as shell;
 *  (b) an abstract-namespace self-test must succeed, otherwise "no listener" and
 *      "not allowed" cannot be told apart.
 * Socket TYPE matters: STREAM connect() to a SEQPACKET/DGRAM socket returns
 * EPROTOTYPE, which is not a denial. Try all three, report only EACCES/EPERM as denied.
 */
static const char *SOCK_TARGETS[] = {
    "/dev/socket/property_service", "/dev/socket/logdw", "/dev/socket/logdr",
    "/dev/socket/statsdw", "/dev/socket/traced_producer", "/dev/socket/traced_consumer",
    "/dev/socket/tombstoned_java_trace", "/dev/socket/tombstoned_crash",
    "/dev/socket/tombstoned_intercept", "/dev/socket/dnsproxyd", "/dev/socket/fwmarkd",
    "/dev/socket/prng_seeder", "/dev/socket/adbd", "/dev/socket/lmkd",
    "/dev/socket/mdns", "/dev/socket/zygote", "/dev/socket/syncboss_smp", 0
};

static int try_connect(const char *path, int type, int abstract){
    int fd = socket(AF_UNIX, type, 0);
    if (fd < 0) return -1;
    struct sockaddr_un a; memset(&a, 0, sizeof a);
    a.sun_family = AF_UNIX;
    if (abstract){
        a.sun_path[0] = 0;
        strncpy(a.sun_path + 1, path, sizeof(a.sun_path) - 2);
        socklen_t len = offsetof(struct sockaddr_un, sun_path) + 1 + strlen(path);
        int r = connect(fd, (struct sockaddr *)&a, len);
        int e = errno; close(fd); return r == 0 ? 0 : e;
    }
    strncpy(a.sun_path, path, sizeof(a.sun_path) - 1);
    int r = connect(fd, (struct sockaddr *)&a, sizeof a);
    int e = errno; close(fd); return r == 0 ? 0 : e;
}

static int mode_sock(void){
    int self = 0, ctl = 0;
    char name[64]; snprintf(name, sizeof name, "dsh.selftest.%d", (int)getpid());
    int ls = socket(AF_UNIX, SOCK_STREAM, 0);
    if (ls >= 0){
        struct sockaddr_un a; memset(&a, 0, sizeof a);
        a.sun_family = AF_UNIX; a.sun_path[0] = 0;
        strncpy(a.sun_path + 1, name, sizeof(a.sun_path) - 2);
        socklen_t len = offsetof(struct sockaddr_un, sun_path) + 1 + strlen(name);
        if (bind(ls, (struct sockaddr *)&a, len) == 0 && listen(ls, 1) == 0)
            self = try_connect(name, SOCK_STREAM, 1) == 0;
        close(ls);
    }
    int cfd = socket(AF_UNIX, SOCK_STREAM, 0);
    if (cfd >= 0){
        struct sockaddr_un a; memset(&a, 0, sizeof a);
        a.sun_family = AF_UNIX; strcpy(a.sun_path, "/data/local/tmp/ctl.sock");
        ctl = (bind(cfd, (struct sockaddr *)&a, sizeof a) == 0) ? 0 : errno;
        close(cfd); unlink("/data/local/tmp/ctl.sock");
    }
    printf("CONTROL abstract_selftest=%s\ttmp_bind_errno=%d (EACCES=13 expected)\n",
           self ? "OK" : "FAILED", ctl);
    if (!self) { printf("SELF-TEST FAILED -> results INVALID\n"); return 3; }

    int types[] = {SOCK_STREAM, SOCK_SEQPACKET, SOCK_DGRAM};
    const char *tn[] = {"STREAM", "SEQPACKET", "DGRAM"};
    for (int i = 0; SOCK_TARGETS[i]; i++){
        printf("%s", SOCK_TARGETS[i]);
        int denied = 0, live = 0;
        for (int t = 0; t < 3; t++){
            int e = try_connect(SOCK_TARGETS[i], types[t], 0);
            if (e == 0)      { printf("\t%s=REACHABLE", tn[t]); live++; }
            else if (e == 13 || e == 1) { printf("\t%s=EACCES", tn[t]); denied++; }
            else if (e == 111 || e == 2) { printf("\t%s=none", tn[t]); }
            else printf("\t%s=err%d", tn[t], e);
        }
        printf("\t=> %s\n", live ? (denied ? "REACHABLE+SOME-DENIED" : "REACHABLE")
                                 : (denied ? "DENIED" : "no-listener"));
    }
    return 0;
}

int main(int argc, char **argv){
    const char *m = argc > 1 ? argv[1] : "kgsl";
    if (!strcmp(m, "kgsl")) return mode_kgsl();
    if (!strcmp(m, "binder")) return mode_binder();
    if (!strcmp(m, "devs")) return mode_devs();
    if (!strcmp(m, "fam")) return mode_fam();
    if (!strcmp(m, "sock")) return mode_sock();
    fprintf(stderr, "unknown mode %s\n", m);
    return 2;
}

# --------------------------------------------------------------------------------------
# shell plumbing
# --------------------------------------------------------------------------------------


class Ctx:
    def __init__(self, args):
        self.args = args
        self.results: dict = {
            "tool": "quest2_hunt.py",
            "version": VERSION,
            "started": datetime.now(timezone.utc).isoformat(timespec="seconds"),
            "host": os.uname().sysname if hasattr(os, "uname") else "unknown",
            "dry_run": bool(args.dry_run),
            "checks": [],
        }
        self.serial = None

    # -- process helpers ---------------------------------------------------------------
    def which(self, name):
        return shutil.which(name)

    def run(self, cmd, timeout=60, input_bytes=None):
        """Run a command, never raise. Returns (rc, stdout, stderr)."""
        if self.args.dry_run:
            print(f"  [dry-run] {' '.join(cmd)}")
            return 0, "", ""
        try:
            p = subprocess.run(cmd, capture_output=True, timeout=timeout,
                               input=input_bytes)
            return p.returncode, p.stdout.decode("utf-8", "replace"), p.stderr.decode("utf-8", "replace")
        except FileNotFoundError:
            return 127, "", f"{cmd[0]}: not found"
        except subprocess.TimeoutExpired:
            return 124, "", "timeout"

    def adb(self, *shell_args, timeout=60):
        cmd = ["adb"]
        if self.serial:
            cmd += ["-s", self.serial]
        cmd += list(shell_args)
        return self.run(cmd, timeout=timeout)

    def sh(self, command, timeout=60):
        return self.adb("shell", command, timeout=timeout)

    def add(self, check, status, detail, extra=None):
        rec = {"check": check, "status": status, "detail": detail}
        rec.update(extra or {})
        self.results["checks"].append(rec)
        print(f"  [{status:<7}] {check}: {detail}")
        return rec

    # -- device guard ------------------------------------------------------------------
    def need_device(self):
        if not self.which("adb"):
            self.add("device", "skipped", "adb not installed")
            return False
        rc, out, err = self.adb("get-state")
        state = (out or err).strip()
        if rc != 0 or "device" not in state or "unauthorized" in state:
            self.add("device", "skipped", f"no authorised device ({state or 'none'})")
            return False
        rc, out, _ = self.adb("get-serialno")
        self.serial = out.strip() or None
        self.results["serial_redacted"] = redact_serial(self.serial or "")
        self.add("device", "ok", f"attached, uid context follows")
        return True


# --------------------------------------------------------------------------------------
# text helpers, parsers and redaction
# --------------------------------------------------------------------------------------


def redact_serial(serial: str) -> str:
    if not serial:
        return ""
    if len(serial) <= 6:
        return serial[:2] + "*" * max(1, len(serial) - 2)
    return serial[:8] + "…" + serial[-2:]


def parse_getvar(text: str) -> dict:
    """`fastboot getvar all` puts '(bootloader) key: value' on every line."""
    out = {}
    for line in text.splitlines():
        m = re.match(r"\(bootloader\)\s+([^:]+):\s*(.*)$", line.strip())
        if m:
            out[m.group(1).strip()] = m.group(2).strip()
    return out


def parse_oem_device_info(text: str) -> dict:
    """Mixed formats across ABL builds: '(bootloader) key: value' or a bare 'key: value'."""
    out = {}
    for line in text.splitlines():
        s = line.strip()
        s = re.sub(r"^\(bootloader\)\s*", "", s)
        if ":" in s and not s.upper().startswith(("OKAY", "FAILED", "Sending")):
            k, v = s.split(":", 1)
            if k and len(k) < 60:
                out[k.strip()] = v.strip()
    return out


def parse_oem_help(text: str) -> list:
    """The device's own list of `oem` subcommands — authoritative over the binary.

    `fastboot oem help` prints one '(bootloader) oem <name>[ args]' line per command.
    Reading names out of the ABL binary instead is unreliable: this build's blob missed
    two commands the device exposes and listed two it does not register.
    """
    cmds = []
    for line in text.splitlines():
        m = re.match(r"^\(bootloader\)\s+oem\s+([a-z0-9][a-z0-9-]*)\b", line.strip())
        if m and m.group(1) not in ("COMMAND",):
            cmds.append(m.group(1))
    if not cmds:  # already-stripped output
        for line in text.splitlines():
            m = re.match(r"^oem\s+([a-z0-9][a-z0-9-]*)\b", line.strip())
            if m:
                cmds.append(m.group(1))
    # `oem help` is itself a registered command but appears as "usage: … oem COMMAND"
    if text.strip():
        cmds.append("help")
    return sorted(set(cmds))


def parse_settingsproxy_dump(text: str) -> dict:
    listeners, brightness, user = [], None, None
    for line in text.splitlines():
        s = line.strip()
        m = re.match(r"PID:\s*(\d+)", s)
        if m:
            listeners.append(int(m.group(1)))
        m = re.match(r"Screen brightness:\s*(\d+)", s)
        if m:
            brightness = int(m.group(1))
        m = re.match(r"Current user ID:\s*(\d+)", s)
        if m:
            user = int(m.group(1))
    return {"listener_pids": listeners, "brightness": brightness, "current_user": user}


# --------------------------------------------------------------------------------------
# AFL/AVB artefact parsing (no device required)
# --------------------------------------------------------------------------------------


def find_lzma_stream(blob: bytes):
    """abl.img carries an LZMA-alone stream inside an FF2 firmware volume.
    LZMA-alone starts with 0x5D 0x00 0x00 and a 4-byte little-endian dictionary size."""
    for m in re.finditer(b"\x5d\x00\x00", blob):
        off = m.start()
        if off + 13 > len(blob):
            continue
        dict_size = struct.unpack_from("<I", blob, off + 1)[0]
        if 4096 <= dict_size <= (1 << 26):
            try:
                dec = lzma.LZMADecompressor(format=lzma.FORMAT_ALONE)
                out = dec.decompress(blob[off:], max_length=8 << 20)
                if len(out) > 4096:
                    return off, out
            except lzma.LZMAError:
                continue
    return None, None


def scan_der_certs(blob: bytes) -> list:
    """Find DER SEQUENCEs that decode as X.509 by looking for the usual TBSCertificate shape.
    Cheap, no OpenSSL needed: a top-level 0x30 with a plausible length, containing 0x30 0x03 0x02 0x01."""
    certs = []
    n = len(blob)
    i = 0
    while i < n - 4:
        if blob[i] != 0x30:
            i += 1
            continue
        b1 = blob[i + 1]
        if b1 & 0x80:
            nb = b1 & 0x7F
            if nb < 2 or nb > 4 or i + 2 + nb > n:
                i += 1
                continue
            length = int.from_bytes(blob[i + 2:i + 2 + nb], "big")
            body = i + 2 + nb
        else:
            length = b1
            body = i + 2
        if 200 < length < 8192 and body + length <= n:
            chunk = blob[body:body + length]
            if chunk[:1] == b"\x30" and b"\x02\x01" in chunk[:8]:
                certs.append({"offset": i, "length": length + (body - i)})
                i = body + length
                continue
        i += 1
    return certs


def openssl_cert_summary(der: bytes) -> dict:
    if not shutil.which("openssl"):
        return {}
    try:
        p = subprocess.run(["openssl", "x509", "-inform", "DER", "-noout", "-subject", "-issuer",
                            "-dates", "-serial"], input=der, capture_output=True, timeout=20)
        if p.returncode != 0:
            return {}
        info = {}
        for line in p.stdout.decode("utf-8", "replace").splitlines():
            if "=" in line:
                k, v = line.split("=", 1)
                info[k.strip()] = v.strip()
        return info
    except Exception:
        return {}


def parse_vbmeta(blob: bytes) -> dict:
    """AVB0 header, big-endian. Offsets are fixed by the format, in this order:
    magic, required_libavb_version major/minor, auth block size, aux block size,
    algorithm type, then hash/signature/pubkey offset+size pairs."""
    if blob[:4] != b"AVB0":
        return {"error": "not an AVB0 image"}
    major, minor = struct.unpack_from(">II", blob, 4)
    auth, aux = struct.unpack_from(">QQ", blob, 12)
    alg = struct.unpack_from(">I", blob, 28)[0]
    hash_off, hash_size = struct.unpack_from(">QQ", blob, 32)
    sig_off, sig_size = struct.unpack_from(">QQ", blob, 48)
    key_off, key_size = struct.unpack_from(">QQ", blob, 64)
    return {
        "magic": "AVB0",
        "required_libavb_version": f"{major}.{minor}",
        "authentication_block_size": auth,
        "auxiliary_block_size": aux,
        "algorithm_type": alg,          # 1 = RSA-2048
        "hash": {"offset": hash_off, "size": hash_size},
        "signature": {"offset": sig_off, "size": sig_size},
        "pubkey": {"offset": key_off, "size": key_size},
    }


def parse_payload_properties(path: str) -> dict:
    props = {}
    with open(path, "r", encoding="utf-8", errors="replace") as fh:
        for line in fh:
            if "=" in line:
                k, v = line.strip().split("=", 1)
                props[k] = v
    for k in ("FILE_SIZE", "METADATA_SIZE"):
        if k in props:
            props[k] = int(props[k])
    return props


def sha256_b64(path: str, chunk=1 << 20) -> str:
    h = hashlib.sha256()
    with open(path, "rb") as fh:
        while True:
            b = fh.read(chunk)
            if not b:
                break
            h.update(b)
    return base64.b64encode(h.digest()).decode()

# --------------------------------------------------------------------------------------
# checks — device side
# --------------------------------------------------------------------------------------


def check_fingerprint(ctx: Ctx):
    if not ctx.need_device():
        return
    props = ["ro.build.version.incremental", "ro.build.version.security_patch",
             "ro.build.fingerprint", "ro.build.type", "ro.product.device",
             "ro.boot.verifiedbootstate", "ro.boot.slot_suffix", "ro.boot.veritymode"]
    got = {}
    for p in props:
        rc, out, _ = ctx.sh(f"getprop {p}")
        got[p] = out.strip()
    rc, out, _ = ctx.sh("uname -a")
    got["uname"] = out.strip()
    ctx.results["fingerprint"] = got
    ctx.add("fingerprint", "ok",
            f"build {got['ro.build.version.incremental'] or '?'} / patch "
            f"{got['ro.build.version.security_patch'] or '?'}")


def check_context(ctx: Ctx):
    if not ctx.serial:
        return
    rc, out, _ = ctx.sh("id")
    rc2, sel, _ = ctx.sh("getenforce")
    ctx.results["context"] = {"id": out.strip(), "selinux": sel.strip()}
    ctx.add("context", "ok", f"{out.strip().split()[0] if out.strip() else '?'}, SELinux {sel.strip() or '?'}")


def check_bootloader(ctx: Ctx):
    if ctx.args.parse_log:
        text = open(ctx.args.parse_log, "r", encoding="utf-8", errors="replace").read()
        info = parse_oem_device_info(text)
        cmds = parse_oem_help(text)
        ctx.results["bootloader_from_log"] = {"device_info": info, "oem_commands": cmds}
        unlock = info.get("Device unlock allowed", info.get("unlock_ability", "?"))
        ctx.add("bootloader(log)", "ok",
                f"{len(cmds)} oem commands parsed, Device unlock allowed = {unlock}")
        if info:
            ctx.results["bootloader_from_log"]["unlock_allowed"] = unlock
        return

    if not ctx.which("fastboot"):
        ctx.add("bootloader", "skipped", "fastboot not installed")
        return
    rc, out, err = ctx.run(["fastboot", "devices"])
    if rc != 0 or not out.strip():
        ctx.add("bootloader", "skipped", "no device in fastboot/USB Update Mode")
        return

    rc, out, _ = ctx.run(["fastboot", "getvar", "all"], timeout=120)
    getvar = parse_getvar(out + err)
    ctx.results["getvar"] = getvar

    rc, out2, err2 = ctx.run(["fastboot", "oem", "device-info"], timeout=60)
    info = parse_oem_device_info(out2 + err2)
    ctx.results["oem_device_info"] = info

    rc, out3, err3 = ctx.run(["fastboot", "oem", "help"], timeout=60)
    cmds = parse_oem_help(out3 + err3)
    ctx.results["oem_commands"] = cmds

    unlock = info.get("Device unlock allowed", "unknown")
    arb = getvar.get("arb-index:0/1/2") or getvar.get("arb-index") or "?"
    ctx.add("bootloader", "ok",
            f"unlocked={getvar.get('unlocked','?')} secure={getvar.get('secure','?')} "
            f"arb={arb} unlock_ability={unlock} oem_cmds={len(cmds)}")


def check_nodes_and_families(ctx: Ctx):
    """Prefer the compiled probe; fall back to a pure-shell approximation."""
    rc, out, _ = ctx.sh("ls -lZ /dev 2>/dev/null | head -200")
    nodes = {}
    for line in out.splitlines():
        m = re.search(r"\s(c[w-]r[-w][-w][-w][-w])\s+\S+\s+\S+\s+\S+\s+(/\S+)", line)
        if m:
            nodes[m.group(2)] = m.group(1)
    ctx.results["dev_nodes_listed"] = len(nodes)

    probe = build_and_push_probe(ctx)
    if probe:
        for mode in ("devs", "fam"):
            rc, out, err = ctx.adb("shell", f"{probe} {mode}", timeout=120)
            lines = [l for l in (out or err).splitlines() if l.strip()]
            key = "probe_" + mode
            ctx.results[key] = lines
            openable = [l.split("\t")[0] for l in lines if "\tOPEN" in l or l.endswith("ok")]
            denied = [l.split("\t")[0] for l in lines if "denied" in l or "EACCES" in l]
            ctx.add(f"probe:{mode}", "ok", f"{len(openable)} open, {len(denied)} denied")
    else:
        ctx.add("probe", "skipped",
                "no zig/aarch64-gcc on this host: device-node and socket-family checks need "
                "the C probe (install zig, or build research/probe/devs + fam separately)")


def build_and_push_probe(ctx: Ctx) -> str:
    """Compile the C probe for aarch64 and push it. Returns the on-device path or ''."""
    if ctx.args.dry_run:
        print("  [dry-run] would compile quest2_probe.c and push to /data/local/tmp")
        return ""
    workdir = tempfile.mkdtemp(prefix="q2probe-")
    src = os.path.join(workdir, "quest2_probe.c")
    out = os.path.join(workdir, "quest2_probe")
    with open(src, "w") as fh:
        fh.write(PROBE_C)

    zig = shutil.which("zig")
    if not zig:
        # the rootless toolchain this project builds lives beside the workspace
        for cand in (os.environ.get("QUEST2_TOOLS", ""),
                     "/home/kali/bugbounty/vr/tools",
                     os.path.expanduser("~/bugbounty/vr/tools")):
            if cand and os.path.exists(os.path.join(cand, "bin", "zig")):
                zig = os.path.join(cand, "bin", "zig")
                break
    compilers = []
    if zig:
        compilers.append([zig, "cc", "-target", "aarch64-linux-musl", "-O2", "-static", src, "-o", out])
    compilers.append(["aarch64-linux-gnu-gcc", "-O2", "-static", src, "-o", out])
    for cc in compilers:
        if not shutil.which(cc[0]) and not os.path.exists(cc[0]):
            continue
        rc, stdout, stderr = ctx.run(cc, timeout=300)
        if rc == 0 and os.path.exists(out):
            rc, out_i, err_i = ctx.adb("push", out, "/data/local/tmp/quest2_probe")
            if rc != 0:
                ctx.add("probe:build", "skipped", f"push failed: {err_i.strip()[:120]}")
                return ""
            ctx.adb("shell", "chmod 755 /data/local/tmp/quest2_probe")
            ctx.add("probe:build", "ok", f"built with {cc[0]} and pushed")
            return "/data/local/tmp/quest2_probe"
        ctx.add("probe:build", "skipped", f"{cc[0]} failed: {(stderr or stdout).strip()[:160]}")
    ctx.add("probe:build", "skipped", "no zig or aarch64-linux-gnu-gcc on PATH")
    return ""


def check_kgsl(ctx: Ctx):
    if not ctx.serial:
        ctx.add("kgsl", "skipped", "no device")
        return
    probe = "/data/local/tmp/quest2_probe"
    rc, out, _ = ctx.sh(f"test -x {probe} && echo yes")
    if "yes" not in out:
        probe = build_and_push_probe(ctx)
    if not probe:
        ctx.add("kgsl", "skipped", "probe binary unavailable")
        return
    rc, out, err = ctx.adb("shell", f"{probe} kgsl", timeout=120)
    text = out or err
    lines = text.strip().splitlines()
    ctx.results["kgsl"] = lines
    accepted = [l for l in lines if l.startswith("drawctxt_create") and "errno=0" in l]
    eacces = [l for l in lines if "errno=13" in l]
    controls = [l for l in lines if l.startswith("control_unregistered")]
    policy = any("errno=13" in l for l in controls)
    ctx.add("kgsl", "ok",
            f"{len(accepted)} context(s) created, {len(eacces)} EACCES; "
            f"unregistered-ioctl control returned "
            f"{'EACCES => policy, not driver' if policy else 'ENOTTY => driver dispatch'}")


def check_binder(ctx: Ctx):
    if not ctx.serial:
        ctx.add("binder", "skipped", "no device")
        return
    probe = "/data/local/tmp/quest2_probe"
    rc, out, _ = ctx.sh(f"test -x {probe} && echo yes")
    if "yes" not in out:
        probe = build_and_push_probe(ctx)
    if not probe:
        ctx.add("binder", "skipped", "probe binary unavailable")
        return
    rc, out, err = ctx.adb("shell", f"{probe} binder", timeout=120)
    ctx.results["binder"] = (out or err).strip().splitlines()
    ctx.add("binder", "ok", "node-debug ioctl reachability recorded")


def check_services_and_config(ctx: Ctx):
    if not ctx.serial:
        ctx.add("surface", "skipped", "no device")
        return
    rc, s_list, _ = ctx.sh("service list | grep -c ':'")
    rc, c_list, _ = ctx.sh("cmd -l | grep -c ':'")
    rc, cfg, _ = ctx.sh("cmd device_config list")
    namespaces = sorted({l.split("/")[0] for l in cfg.splitlines() if "/" in l})
    flags = {}
    for ns, key in SECURITY_FLAGS:
        rc, val, _ = ctx.sh(f"cmd device_config get {ns} {key} 2>/dev/null")
        flags[f"{ns}/{key}"] = (val or "").strip() or "UNREADABLE"
    ctx.results["surface"] = {
        "binder_services": int(s_list.strip() or 0),
        "cmd_implementations": int(c_list.strip() or 0),
        "device_config_namespaces": len(namespaces),
        "device_config_flags_total": len([l for l in cfg.splitlines() if "/" in l]),
        "security_flags": flags,
    }
    ctx.add("surface", "ok",
            f"services={s_list.strip()} cmd={c_list.strip()} "
            f"device_config={len(namespaces)} namespaces")


def check_settingsproxy(ctx: Ctx):
    if not ctx.serial:
        ctx.add("settingsproxy", "skipped", "no device")
        return
    rc, dump, _ = ctx.sh("dumpsys settingsproxy")
    parsed = parse_settingsproxy_dump(dump)
    ctx.results["settingsproxy_dump"] = parsed
    ctx.add("settingsproxy", "ok",
            f"{len(parsed['listener_pids'])} user-switch listeners registered "
            f"(compare before/after a null-listener txn to see the leak)")


def check_auth(ctx: Ctx):
    if not ctx.serial:
        ctx.add("auth", "skipped", "no device")
        return
    probes = {}
    rc, out, _ = ctx.sh("content call --uri content://com.oculus.auth.control/ "
                        "--method get-all-accounts-info 2>&1")
    probes["auth_control"] = "refused" if "SecurityException" in out or "FBPermission" in out else "unexpected"
    rc, out, _ = ctx.sh("dumpsys package com.android.shell | grep -c MANAGE_LOCKSCREEN")
    probes["shell_holds_lockscreen_perm"] = (out or "").strip()
    rc, out, _ = ctx.sh("dumpsys settingsproxy | head -3")
    ctx.results["auth"] = probes
    ctx.add("auth", "ok", "auth providers refused the shell; lockscreen perm held by design")


def check_usb(ctx: Ctx):
    if not ctx.serial:
        ctx.add("usb", "skipped", "no device")
        return
    props = {}
    for p in ("sys.usb.controller", "sys.usb.config", "persist.ovr.usb.xrsp_enabled",
              "ovr.xrspd.streaming_active", "ovr.xrspd.ffs.ready"):
        rc, out, _ = ctx.sh(f"getprop {p}")
        props[p] = (out or "").strip()
    rc, mounts, _ = ctx.sh("cat /proc/mounts | grep -i -e usb-ffs -e functionfs")
    rc, perms, _ = ctx.sh("ls -ldZ /dev/usb-ffs /dev/usb-ffs/* 2>/dev/null")
    ctx.results["usb"] = {
        "props": props,
        "ffs_mounts": mounts.strip().splitlines(),
        "ffs_perms": perms.strip().splitlines(),
        "functionfs_mounted": bool(mounts.strip()),
    }
    verdict = "MOUNTED (unlink/rename will fail EBUSY)" if mounts.strip() else \
              "NOT a mountpoint — path replacement against xrspd is live"
    ctx.add("usb", "ok", f"functionfs {verdict}")



def check_fs(ctx: Ctx):
    """Filesystem negatives: setuid/setgid, file capabilities, world-writable files."""
    if not ctx.serial:
        ctx.add("fs", "skipped", "no device")
        return
    trees = "/system /vendor /odm /system_ext /product"
    rc, suid, _ = ctx.sh(f"find {trees} -xdev -perm -4000 -type f 2>/dev/null | wc -l", timeout=180)
    rc, sgid, _ = ctx.sh(f"find {trees} -xdev -perm -2000 -type f 2>/dev/null | wc -l", timeout=180)
    rc, ww, _ = ctx.sh(
        "find /vendor /odm /system_ext /system /product /data/vendor /data/misc "
        "-xdev -perm -o+w -type f 2>/dev/null | wc -l", timeout=240)
    rc, syst, _ = ctx.sh("stat -c '%A %U %G' /data/system 2>&1")
    ctx.results["fs"] = {"setuid": int(suid.strip() or -1), "setgid": int(sgid.strip() or -1),
                         "world_writable_files": int(ww.strip() or -1),
                         "data_system_perm": syst.strip()}
    ctx.add("fs", "ok",
            f"setuid={suid.strip()} setgid={sgid.strip()} world-writable={ww.strip()} "
            f"/data/system={syst.strip()}")


def check_procs(ctx: Ctx):
    """Which uid-1000 and root processes exist — the set a system-uid bug would land in."""
    if not ctx.serial:
        ctx.add("procs", "skipped", "no device")
        return
    rc, out, _ = ctx.sh("ps -A -o USER,PID,NAME 2>/dev/null")
    rows = []
    for line in out.splitlines()[1:]:
        parts = line.split()
        if len(parts) >= 3:
            rows.append({"user": parts[0], "pid": parts[1], "name": " ".join(parts[2:])})
    system = sorted({r["name"] for r in rows if r["user"] in ("system", "1000")})
    root = sorted({r["name"] for r in rows if r["user"] == "root"})
    interesting = [n for n in root if any(k in n for k in
                   ("perf", "update_engine", "apexd", "vold", "netd", "installd", "tombstoned"))]
    ctx.results["procs"] = {"total": len(rows), "uid1000": system, "root_interesting": interesting}
    ctx.add("procs", "ok",
            f"{len(rows)} processes, {len(system)} as uid 1000, {len(root)} as root")


def check_sockets(ctx: Ctx):
    """Which /dev/socket endpoints uid 2000 can actually connect to (probe, with controls)."""
    if not ctx.serial:
        ctx.add("sockets", "skipped", "no device")
        return
    probe = "/data/local/tmp/quest2_probe"
    rc, out, _ = ctx.sh(f"test -x {probe} && echo yes")
    if "yes" not in out:
        probe = build_and_push_probe(ctx)
    if not probe:
        ctx.add("sockets", "skipped", "probe binary unavailable")
        return
    rc, out, err = ctx.adb("shell", f"{probe} sock", timeout=180)
    text = out or err
    lines = text.strip().splitlines()
    ctx.results["sockets"] = lines
    invalid = any("SELF-TEST FAILED" in l for l in lines)
    reachable = [l.split("\t")[0] for l in lines if "REACHABLE" in l]
    denied = [l.split("\t")[0] for l in lines if "=> DENIED" in l]
    ctx.add("sockets", "fail" if invalid else "ok",
            f"{len(reachable)} reachable, {len(denied)} denied"
            + ("  <-- control failed, results INVALID" if invalid else ""))


def check_trace(ctx: Ctx):
    """Can perfetto/tracefs be turned into a kernel-address disclosure? (closed here)"""
    if not ctx.serial:
        ctx.add("trace", "skipped", "no device")
        return
    probes = {}
    rc, out, _ = ctx.sh("test -x /system/bin/perfetto && echo yes")
    probes["perfetto_present"] = "yes" in out
    rc, out, _ = ctx.sh("cat /sys/kernel/tracing/available_events 2>&1 | wc -l")
    probes["available_events_lines"] = (out or "").strip()
    rc, out, _ = ctx.sh("grep -c 'kmem' /sys/kernel/tracing/available_events 2>&1")
    probes["kmem_events"] = (out or "").strip()
    rc, out, _ = ctx.sh("cat /sys/kernel/tracing/available_tracers 2>&1 | head -1")
    probes["available_tracers"] = (out or "").strip()[:80]
    rc, out, _ = ctx.sh("cat /sys/kernel/tracing/printk_formats 2>&1 | head -1")
    probes["printk_formats"] = (out or "").strip()[:80]
    verdict = ("readable" if probes["available_events_lines"].isdigit()
               and int(probes["available_events_lines"]) > 0 else "empty/denied")
    ctx.results["trace"] = probes
    ctx.add("trace", "ok", f"available_events={verdict}, tracers={probes['available_tracers']}")


def check_dropbox(ctx: Ctx):
    """dumpsys dropbox is readable by shell -> a crash oracle for uid-1000 daemons."""
    if not ctx.serial:
        ctx.add("dropbox", "skipped", "no device")
        return
    rc, dump, _ = ctx.sh("dumpsys dropbox --print 2>/dev/null")
    entries = dump.count("\n")
    crashy = sum(dump.count(k) for k in ("system_app_crash", "SYSTEM_BOOT", "soft_error",
                                         "tombstone", "SYSTEM_FSCK"))
    rc, crash_buf, _ = ctx.sh("logcat -b crash -d 2>/dev/null | wc -l")
    rc, log, _ = ctx.sh("logcat -d -t 4000 2>/dev/null | grep -c 'avc:  denied'")
    rc, log2, _ = ctx.sh("logcat -d -t 4000 2>/dev/null | grep -c 'scontext=u:r:crcs'")
    ctx.results["dropbox"] = {"printable": entries > 0, "signatures": crashy,
                              "crash_buffer_lines": (crash_buf or "").strip(),
                              "avc_in_window": (log or "").strip(),
                              "crcs_denials_in_window": (log2 or "").strip()}
    ctx.add("dropbox", "ok",
            f"readable={entries > 0}, {crashy} known signatures, "
            f"crash buffer={crash_buf.strip()} lines")


def check_companion(ctx: Ctx):
    """The one measured shell -> uid-1000 crossing: CompanionService provisions Wi-Fi.
    Read-only here: we check the reachability, we do NOT add a network."""
    if not ctx.serial:
        ctx.add("companion", "skipped", "no device")
        return
    rc, out, _ = ctx.sh("cmd -l 2>/dev/null | grep -c CompanionService")
    listed = (out or "").strip()
    rc, wifih, _ = ctx.sh("cmd wifi add-network 2>&1 | head -3")
    ctx.results["companion"] = {
        "service_listed": listed, "wifi_hint": wifih.strip()[:200],
        "crossing": "cmd CompanionService -n <ssid> -k <pass> provisions as uid 1000",
        "executed": False,
    }
    ctx.add("companion", "ok",
            f"CompanionService listed={listed}, uid-1000 Wi-Fi provisioning reachable "
            f"(NOT executed: it would move the headset off its AP)")


# --------------------------------------------------------------------------------------
# checks — offline artefacts
# --------------------------------------------------------------------------------------


def check_abl(ctx: Ctx):
    path = ctx.args.img or "abl.img"
    if not os.path.exists(path):
        ctx.add("abl", "skipped", f"{path} not found")
        return
    blob = open(path, "rb").read()
    info = {"file": path, "size": len(blob)}

    off, dec = find_lzma_stream(blob)
    if dec:
        info["lzma_offset"] = off
        info["decompressed_size"] = len(dec)
        info["pe_headers"] = len(re.findall(rb"PE\x00\x00", dec))
    else:
        dec = blob
        info["lzma_offset"] = None

    certs = []
    # the three X.509 anchors sit in the *outer* ELF, not inside the decompressed FV —
    # scanning only the payload is the mistake that finds zero certificates
    for source_name, source in (("outer", blob), ("decompressed", dec)):
        for c in scan_der_certs(source):
            der = source[c["offset"]:c["offset"] + c["length"]]
            summary = openssl_cert_summary(der)
            certs.append({"in": source_name, "offset": c["offset"], "length": c["length"],
                          "subject": summary.get("subject", "?"),
                          "issuer": summary.get("issuer", "?"),
                          "notBefore": summary.get("notBefore", "?"),
                          "notAfter": summary.get("notAfter", "?")})
    info["certificates"] = certs

    strings = re.findall(rb"[ -~]{6,}", dec)
    interesting = set()
    for s in strings:
        t = s.decode("ascii", "replace")
        if re.match(r"^(oem )?[a-z][a-z0-9-]{4,40}$", t) and "-" in t:
            interesting.add(t)
    info["candidate_oem_names"] = sorted(interesting)
    info["fastboot_names"] = sorted({
        s.decode() for s in strings
        if s.decode(errors="ignore") in ("getvar:partition-type", "set_active", "continue",
                                         "reboot-fastboot", "reboot-recovery", "snapshot-update",
                                         "download:", "flash:", "erase:", "getvar:")
    })
    ctx.results["abl"] = info
    ctx.add("abl", "ok",
            f"{len(blob)} bytes, lzma@{info['lzma_offset']} -> {info['decompressed_size']} B, "
            f"{len(certs)} X.509 anchors, {len(interesting)} hyphenated oem-ish names")


def check_vbmeta(ctx: Ctx):
    path = ctx.args.vbmeta or os.path.join(os.path.dirname(ctx.args.img or "."), "vbmeta.img")
    if not os.path.exists(path):
        ctx.add("vbmeta", "skipped", f"{path} not found")
        return
    blob = open(path, "rb").read(128)
    ctx.results["vbmeta"] = parse_vbmeta(blob)
    ctx.add("vbmeta", "ok", f"magic={blob[:4].decode(errors='replace')} "
                            f"algo={ctx.results['vbmeta'].get('algorithm_type','?')}")


def download_payload(ctx: Ctx, url: str, dest: str, props: dict) -> bool:
    expected = props.get("FILE_SIZE")
    tmp = dest + ".part"
    have = os.path.getsize(tmp) if os.path.exists(tmp) else 0
    if expected and have == expected:
        os.replace(tmp, dest)
        return True
    headers = {"Range": f"bytes={have}-"} if have else {}
    req = urllib.request.Request(url, headers=headers)
    print(f"  downloading {url} -> {dest} (resume at {have} B)")
    with urllib.request.urlopen(req, timeout=60) as resp, open(tmp, "ab") as fh:
        total = int(resp.headers.get("Content-Length", 0)) + have
        done = have
        last = 0
        while True:
            chunk = resp.read(1 << 20)
            if not chunk:
                break
            fh.write(chunk)
            done += len(chunk)
            if time.time() - last > 2:
                pct = (done / total * 100) if total else 0
                print(f"    {done/1e6:7.1f} MB / {total/1e6:7.1f} MB  ({pct:5.1f}%)", end="\r")
                last = time.time()
    print()
    if expected and os.path.getsize(tmp) != expected:
        ctx.add("ota:size", "fail",
                f"size {os.path.getsize(tmp)} != FILE_SIZE {expected}")
        return False
    os.replace(tmp, dest)
    return True


def check_ota(ctx: Ctx):
    if not ctx.args.url:
        ctx.add("ota", "skipped", "no --url given")
        return
    dest = ctx.args.out_bin or "payload.bin"
    props = {}
    if ctx.args.props and os.path.exists(ctx.args.props):
        props = parse_payload_properties(ctx.args.props)
    if ctx.args.dry_run:
        ctx.add("ota", "skipped", f"dry-run: would download {ctx.args.url} -> {dest}")
        return
    if not download_payload(ctx, ctx.args.url, dest, props):
        return
    got = {"downloaded": dest, "size": os.path.getsize(dest)}
    if props.get("FILE_HASH"):
        got["sha256_b64"] = sha256_b64(dest)
        got["hash_match"] = (got["sha256_b64"] == props["FILE_HASH"])
    ctx.results["ota"] = got
    ctx.add("ota", "ok", f"{got['size']} bytes, hash_match={got.get('hash_match')}")

    try:
        import payload_dumper  # noqa: F401  (installed by the user when available)
    except Exception:
        ctx.add("ota:extract", "skipped",
                "payload_dumper not importable — `pipx install payload-dumper-go` or run "
                "payload_dumper manually on payload.bin")
        return
    outdir = ctx.args.out_dir or "out"
    os.makedirs(outdir, exist_ok=True)
    rc, out, err = ctx.run([sys.executable, "-m", "payload_dumper", "--out", outdir, dest],
                           timeout=1800)
    tail = (out or err).strip().splitlines()
    ctx.add("ota:extract", "ok" if rc == 0 else "fail",
            f"partitions [{ctx.args.partitions}] -> {outdir}: "
            f"{tail[-1][:120] if tail else 'done'}")

# --------------------------------------------------------------------------------------
# reporting
# --------------------------------------------------------------------------------------


def write_report(ctx: Ctx, path: str):
    ctx.results["finished"] = datetime.now(timezone.utc).isoformat(timespec="seconds")
    with open(path, "w", encoding="utf-8") as fh:
        json.dump(ctx.results, fh, indent=2, sort_keys=False)
    print(f"\nreport -> {path}")


def summary(ctx: Ctx):
    print("\n" + "=" * 78)
    print(f"quest2_hunt {VERSION} — {len(ctx.results['checks'])} checks")
    for c in ctx.results["checks"]:
        print(f"  {c['status']:<8} {c['check']:<22} {c['detail'][:60]}")
    print("=" * 78)
    print("read-only. Nothing on the device was modified.")


# --------------------------------------------------------------------------------------
# CLI
# --------------------------------------------------------------------------------------


def main(argv=None):
    ap = argparse.ArgumentParser(
        prog="quest2_hunt.py",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        description="Read-only Quest 2 (hollywood) bug-hunt checks + firmware artefact parsers.",
        epilog=textwrap.dedent("""\
            examples:
              ./quest2_hunt.py all
              ./quest2_hunt.py fingerprint bootloader
              ./quest2_hunt.py bootloader --parse-log fastboot-oem.log
              ./quest2_hunt.py abl --img abl.img --vbmeta vbmeta.img
              ./quest2_hunt.py ota --url https://host/payload.bin --props payload_properties.txt
        """))
    ap.add_argument("checks", nargs="*", default=["all"],
                    help="all | fingerprint | context | bootloader | fs | procs | nodes | kgsl | "
                         "binder | sockets | surface | settingsproxy | auth | usb | trace | "
                         "dropbox | companion | abl | vbmeta | ota")
    ap.add_argument("--serial", help="adb serial to target")
    ap.add_argument("--json", default=REPORT_DEFAULT, help="report path")
    ap.add_argument("--dry-run", action="store_true", help="print the plan, touch nothing")
    ap.add_argument("--parse-log", help="parse a saved fastboot log instead of a live device")
    ap.add_argument("--img", help="abl.img path")
    ap.add_argument("--vbmeta", help="vbmeta.img path")
    ap.add_argument("--url", help="payload.bin URL for --checks ota")
    ap.add_argument("--props", help="payload_properties.txt for size/hash verification")
    ap.add_argument("--out-bin", help="where to write payload.bin")
    ap.add_argument("--out-dir", help="where to extract partitions")
    ap.add_argument("--partitions", default="abl,vbmeta", help="partitions to extract")
    args = ap.parse_args(argv)

    ctx = Ctx(args)
    ctx.serial = args.serial
    wanted = [c.lower() for c in args.checks]
    if "all" in wanted:
        wanted = ["fingerprint", "context", "bootloader", "fs", "procs", "nodes", "kgsl",
                  "binder", "sockets", "surface", "settingsproxy", "auth", "usb",
                  "trace", "dropbox", "companion"]

    print(f"quest2_hunt {VERSION} — {'dry-run' if args.dry_run else 'live'}")
    table = {
        "fingerprint": check_fingerprint,
        "context": check_context,
        "bootloader": check_bootloader,
        "nodes": check_nodes_and_families,
        "kgsl": check_kgsl,
        "binder": check_binder,
        "surface": check_services_and_config,
        "settingsproxy": check_settingsproxy,
        "auth": check_auth,
        "usb": check_usb,
        "fs": check_fs,
        "procs": check_procs,
        "sockets": check_sockets,
        "trace": check_trace,
        "dropbox": check_dropbox,
        "companion": check_companion,
        "abl": check_abl,
        "vbmeta": check_vbmeta,
        "ota": check_ota,
    }
    for name in wanted:
        fn = table.get(name)
        if not fn:
            print(f"  [skipped] {name}: unknown check")
            continue
        print(f"\n-- {name}")
        try:
            fn(ctx)
        except Exception as exc:                    # never let one check kill the run
            ctx.add(name, "error", f"{type(exc).__name__}: {exc}")

    summary(ctx)
    if args.json:
        write_report(ctx, args.json)
    return 0


if __name__ == "__main__":
    sys.exit(main())