admin_hook_runner.py

r"""admin_hook_runner.py - the SYSTEM elevation primitive (minimal).

A SYSTEM command-drain that gives an AI agent (e.g. Claude Code) an unattended admin
channel WITHOUT a UAC prompt each time - the one capability Claude Code lacks natively
(it otherwise runs as the launching user; see anthropics/claude-code issue #29275).

HOW IT WORKS:
  - Installed to run as NT AUTHORITY\SYSTEM (/RL HIGHEST), either as a scheduled task
    (fires this script every minute, one-shot) or as a Windows service (this script in
    --loop mode, polling every few seconds).
  - Each poll: if the QUEUE file (_sysfix.ps1, next to this script) exists, execute it as
    SYSTEM (powershell -File) and DELETE it. Write logs/heartbeat.txt.
  - The queued .ps1 is generated by the CLIENT (run_elevated.py), which also writes its
    own results log; this runner is agnostic to the payload.

Two runtime shapes, same code:
  (default, one-shot)  drain once + heartbeat + exit     <- scheduled task fires it @60s
  --loop [--interval N] drain forever, every N seconds   <- Windows service (instant-ish)

WHY THIS SHAPE: a periodic privileged executor, kept deliberately minimal because it runs
EVERYTHING as SYSTEM - fewer moving parts = less attack surface. Paths derive from this
file's own folder, so it works wherever it is installed.
"""
import argparse
import os
import subprocess
import sys
import time

BASE = os.path.dirname(os.path.abspath(__file__))
QUEUE = os.path.join(BASE, "_sysfix.ps1")            # one-shot script we execute + delete
HEARTBEAT = os.path.join(BASE, "logs", "heartbeat.txt")
_NOWIN = getattr(subprocess, "CREATE_NO_WINDOW", 0)


def run_queued() -> bool:
    """If a queued command script exists, run it as SYSTEM and delete it. True if it ran."""
    if not os.path.exists(QUEUE):
        return False
    try:
        subprocess.run(
            ["powershell.exe", "-NoProfile", "-ExecutionPolicy", "Bypass", "-File", QUEUE],
            creationflags=_NOWIN, timeout=300,
        )
    finally:
        try:
            os.remove(QUEUE)
        except OSError:
            pass
    return True


def heartbeat(ran: bool) -> None:
    """Prove the runner is alive + confirm the identity it runs as (SYSTEM). Overwritten each tick."""
    try:
        os.makedirs(os.path.dirname(HEARTBEAT), exist_ok=True)
        who = os.environ.get("USERNAME", "?")
        stamp = time.strftime("%Y-%m-%d %H:%M:%S")
        with open(HEARTBEAT, "w", encoding="utf-8") as f:
            f.write(f"{stamp}  ran_as={who}  drained_queue={ran}\n")
    except Exception:
        pass


def main() -> int:
    ap = argparse.ArgumentParser(description="SYSTEM elevation command-drain")
    ap.add_argument("--loop", action="store_true",
                    help="run continuously (Windows-service mode) instead of one-shot")
    ap.add_argument("--interval", type=float, default=3.0,
                    help="seconds between polls when --loop (default 3)")
    args = ap.parse_args()

    if args.loop:
        while True:
            heartbeat(run_queued())
            time.sleep(max(0.5, args.interval))
    else:
        heartbeat(run_queued())
    return 0


if __name__ == "__main__":
    raise SystemExit(main())