credential-server

#!/usr/bin/env python3
"""Credential server for CLI tools inside safehouse sandbox.

Listens on a Unix socket and serves credentials on-demand with approval.
Supports GitHub CLI (gh), AWS CLI, Azure CLI (az), SSH agent gating, Docker, and netrc.
Run this OUTSIDE the sandbox.

Protocol (JSON over Unix socket):
  Request:  {"type": "gh"}
  Request:  {"type": "aws", "profile": "my-profile"}
  Request:  {"type": "az"}
  Request:  {"type": "ssh"}
  Request:  {"type": "docker"}
  Request:  {"type": "netrc", "machine": "pypi.fury.io"}
  Request:  {"type": "open", "url": "https://..."}
  Response: {"ok": true, ...credentials...}
  Response: {"ok": false}

Remote mode (aws and open): a request with "origin": "remote" arrives over an
ssh -R socket forward. It carries self-reported "host"/"cmd"/"cwd" and is
scoped to a synthetic "remote:<host>" identity. See AGENTS.md "Remote mode".

Approval form (single screen, hotkeys toggle each row, Enter confirms):
  scope:    [o] once  [r] reads  [p] pattern  [a] all
  duration: [1] 1min  [5] 5min  [s] session   (shown for r/p/a; default 5min)
  pattern:  [g] cycles granularity            (shown for p with >1 option:
                                               e.g. "aws s3 cp" vs "aws s3" vs exact)
  [t] include sensitive                       (shown for r/p/a; default on when the
                                               displayed command is itself sensitive)
  grants:   live summary of the pending approval
  [Enter] confirm  [d/Esc] deny  [?] info
Defaults to once, so a bare Enter allows the single request.

Sensitive escalation (secretsmanager/kms/keyvault except list-*/describe-* metadata
ops; ssm only with --with-decryption):
  A persistent approval without "include sensitive" still gates sensitive commands
  later, once each, even when otherwise auto-approved:
    [Enter]  once     - approve this one request
    [d/Esc]  deny     - reject
    [r]      remember - auto-approve sensitive for remaining approval duration
    [s]      session  - auto-approve sensitive for this credential until the
                        scope ends (sandbox exit / server restart), surviving
                        later re-approvals
"""
import sys

if "--help" in sys.argv or "-h" in sys.argv:
    print("Usage: credential-server [options]")
    print()
    print("Options:")
    print("  --no-auto-git-reads  Prompt for git/gh read commands")
    print("                       By default git/gh reads are auto-approved")
    print("  --no-auto-git-push   Prompt for git push/gh pr write commands")
    print("                       By default non-protected-branch pushes are auto-approved")
    print("  --no-auto-docker-reads")
    print("                       Prompt for docker read commands")
    print("                       By default docker reads are auto-approved")
    print("  --no-auto-netrc      Prompt for netrc credential requests")
    print("                       By default netrc requests are auto-approved")
    sys.exit(0)
auto_approve_git_reads = "--no-auto-git-reads" not in sys.argv
auto_approve_git_push = "--no-auto-git-push" not in sys.argv
auto_approve_docker_reads = "--no-auto-docker-reads" not in sys.argv
auto_approve_netrc = "--no-auto-netrc" not in sys.argv
import ctypes
import ctypes.util
import fnmatch
import json
import netrc as netrc_mod
import os
import re
import select
import shlex
import shutil
import signal
import socket
import struct
import subprocess
import sys
import termios
import threading
import time
import tty
import urllib.parse
from datetime import datetime, timedelta
try:
    import tomllib
except ModuleNotFoundError:
    tomllib = None

SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
SOCKET_PATH = os.path.join(SCRIPT_DIR, ".credential-server.sock")
AUDIT_LOG = os.path.join(SCRIPT_DIR, "credential-audit.log")
CONFIG_PATH = os.path.join(SCRIPT_DIR, "config.toml")
APPROVALS_PATH = os.path.join(SCRIPT_DIR, ".approvals.toml")
DOCKER_PROXY_PATH = os.path.join(SCRIPT_DIR, ".docker-proxy.sock")

# How long an OAuth loopback tunnel for a remote `open` stays up.
LOOPBACK_TUNNEL_TTL = 300

# Click-to-focus notifications via alerter (brew install vjeantet/tap/alerter).
# --sender activates terminal app on click; wezterm cli or AXRaise focuses the pane/window.
WINDOW_TITLE = "credential-server"
TERMINAL_BUNDLE_IDS = {
    "Apple_Terminal": "com.apple.Terminal",
    "iTerm.app": "com.googlecode.iterm2",
    "ghostty": "com.mitchellh.ghostty",
    "WezTerm": "com.github.wez.wezterm",
    "Alacritty": "org.alacritty",
    "kitty": "net.kovidgoyal.kitty",
}
TERMINAL_BUNDLE_ID = TERMINAL_BUNDLE_IDS.get(os.environ.get("TERM_PROGRAM", ""))
HAS_ALERTER = shutil.which("alerter") is not None
WEZTERM_PANE = os.environ.get("WEZTERM_PANE")

# Approval memory: {(safehouse_pid, cred_key): [grant, ...]}. A credential holds
# several grants so approving one pattern doesn't evict another.
# grant: {"mode": ..., "expires": ..., "pattern": ..., "sensitive": ...}
# cred_key examples: "gh", "aws:dev", "ssh:root@host"
# mode: "read" = reads only, "similar" = matching pattern only, "all" = everything
# pattern: pattern string from pattern_options (only for "similar" mode), e.g. "aws s3 cp"
# expires: datetime | None (None = until safehouse process exits)
approvals = {}

# Sensitive escalations remembered for the whole scope, outliving the approval
# entry they were granted under: {(safehouse_pid, cred_key)}
sensitive_scopes = set()

# PIDs of approved commands — children inherit approval
approved_pids = set()

def _pid_alive(pid):
    try:
        os.kill(pid, 0)
        return True
    except OSError:
        return False

def audit_log(cmd, cred_key, approval, cwd=None, origin=None):
    """Append approved credential request to audit log.

    origin is set for remote requests ("remote:<host>") so self-reported
    remote commands are distinguishable from verified local ones."""
    ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    cwd_tag = f" cwd={cwd}" if cwd else ""
    origin_tag = f" origin={origin}" if origin else ""
    with open(AUDIT_LOG, "a") as f:
        f.write(f"{ts} {approval} {cred_key} {cmd}{cwd_tag}{origin_tag}\n")


SOL_LOCAL = 0
LOCAL_PEERPID = 2

DIM = "\033[2m"
BOLD = "\033[1m"
GREEN = "\033[32m"
RED = "\033[31m"
YELLOW = "\033[33m"
CYAN = "\033[36m"
RESET = "\033[0m"


SENSITIVE_SERVICES = {"secretsmanager", "kms", "keyvault", "secret"}

PROMPT_TIMEOUT = 60
REMINDER_INTERVAL = 15


def load_config():
    """Load config from config.toml. Returns dict with defaults for missing keys."""
    defaults = {"auto-approve": {"aws-profiles": []}}
    if tomllib is None:
        if os.path.exists(CONFIG_PATH):
            print(f"  {RED}warning: config.toml exists but tomllib unavailable (needs Python 3.11+){RESET}")
        return defaults
    if not os.path.exists(CONFIG_PATH):
        return defaults
    try:
        with open(CONFIG_PATH, "rb") as f:
            return tomllib.load(f)
    except (tomllib.TOMLDecodeError, OSError) as e:
        print(f"  {RED}warning: failed to load config.toml: {e}{RESET}")
        return defaults


_config = load_config()
_ap = _config.get("auto-approve", {}).get("aws-profiles", [])
auto_approve_aws_profiles = {p for p in _ap if isinstance(p, str)} if isinstance(_ap, list) else set()
_mp = _config.get("auto-approve", {}).get("main-push-repos", [])
auto_approve_main_push_repos = {r.lower() for r in _mp if isinstance(r, str)} if isinstance(_mp, list) else set()


def is_auto_approved_profile(profile):
    return any(fnmatch.fnmatchcase(profile, pat) for pat in auto_approve_aws_profiles)


def is_sensitive(parts):
    """Check if command accesses sensitive services (secrets, keys, etc.).

    Metadata-only operations (list-*, describe-*) are exempt; get-* is not,
    since get-secret-value reveals secret material. SSM only gates
    --with-decryption reads (the only path exposing SecureString values)."""
    parts = parts or ()
    if "ssm" in parts and "--with-decryption" in parts:
        return True
    for i, part in enumerate(parts):
        if part in SENSITIVE_SERVICES:
            nxt = parts[i + 1] if i + 1 < len(parts) else ""
            if not nxt.startswith(("list-", "describe-")):
                return True
    return False


def _is_s3_download(parts):
    """True if an `aws s3 cp/sync` only reads from S3: its first non-flag operand
    is an s3:// URI and no later token is one, so an s3:// destination (or a flag
    value hiding one) always stays a write. `-` counts as an operand, keeping
    `cp - s3://...` an upload. `mv` is excluded: it deletes the source object.
    Local filesystem writes (including `sync --delete`) are out of scope."""
    if "s3" not in parts:
        return False
    i = parts.index("s3") + 1
    if i >= len(parts) or parts[i] not in ("cp", "sync"):
        return False
    operands = [p for p in parts[i + 1:] if p == "-" or not p.startswith("-")]
    if not operands or not operands[0].startswith("s3://"):
        return False
    return not any(p.startswith("s3://") for p in operands[1:])


def is_read_only(parts):
    """Classify if command is read-only (safe to auto-approve in read mode).

    `parts` is a tokenized argv list (use get_argv() output, not cmd.split())."""
    if not parts:
        return False

    aws_read_prefixes = ("describe-", "list-", "get-", "head-", "batch-get-", "filter-", "lookup-", "search-")
    aws_read_exact = {"ls", "tail", "scan", "query", "search", "start-query", "start-live-tail",
                      "validate-template"}
    if parts[0] == "aws":
        if is_sensitive(parts):
            return False
        if _is_s3_download(parts):
            return True
        for prefix in aws_read_prefixes:
            if any(part.startswith(prefix) for part in parts[1:]):
                return True
        if any(part in aws_read_exact for part in parts[1:]):
            return True
        return False

    gh_read_verbs = {"list", "view", "status", "diff", "checks", "search", "clone", "watch"}
    gh_help_flags = {"--version", "--help", "-v", "-h"}
    if parts[0] == "gh":
        if len(parts) > 1 and parts[1] in gh_help_flags:
            return True
        if len(parts) > 1 and parts[1] in gh_read_verbs:
            return True
        if len(parts) > 2 and parts[2] in gh_read_verbs:
            return True
        if len(parts) > 1 and parts[1] == "api":
            mutating = {"-X", "--method"}
            for i, p in enumerate(parts[2:], 2):
                if p in mutating and i + 1 < len(parts) and parts[i + 1].upper() != "GET":
                    return False
            return True
        return False

    az_read_verbs = {"list", "show", "get"}
    if parts[0] == "az":
        if is_sensitive(parts):
            return False
        return any(part in az_read_verbs for part in parts[1:])

    docker_read_cmds = {"ps", "images", "logs", "inspect", "stats", "top", "port",
                        "version", "info", "diff", "history", "pull"}
    docker_read_mgmt = {"ls", "list", "inspect", "logs"}
    if parts[0] == "docker":
        # docker compose reads
        if len(parts) >= 3 and parts[1] == "compose":
            return parts[2] in ("ps", "logs", "config", "ls", "images", "version", "pull")
        # docker <management> <read-verb> (e.g. docker container ls)
        if len(parts) >= 3 and parts[1] in ("container", "image", "network", "volume", "buildx"):
            return parts[2] in docker_read_mgmt
        # docker <subcommand>
        if len(parts) >= 2:
            return parts[1] in docker_read_cmds
        return False

    git_read_subcommands = {"fetch", "pull", "clone", "ls-remote", "remote"}
    if parts[0] == "git" or parts[0].endswith("/git"):
        i = _git_subcommand_index(parts)
        return i is not None and parts[i] in git_read_subcommands

    return False


def rw_tag(parts):
    """'read'/'write' for tools is_read_only classifies, None for the rest."""
    if not parts or os.path.basename(parts[0]) not in ("aws", "gh", "az", "docker", "git"):
        return None
    return "read" if is_read_only(parts) else "write"


def approved_line(tags, cmd_parts, cmd, actual_cmd=None):
    rw = rw_tag(cmd_parts)
    if rw:
        tags = tags + [rw]
    head = actual_cmd or cmd
    if rw == "write":
        head = f"{RED}{head}{RESET}"
    body = f"{head}\n    {DIM}via: {cmd}{RESET}" if actual_cmd else head
    return f"\n  {GREEN}✓{RESET} {DIM}({', '.join(tags)}){RESET}\n    {body}"


def _git_subcommand_index(parts):
    """Index of the git subcommand in parts, skipping global `-c key=val` /
    `-C path` flags. Returns None if no subcommand found."""
    i = 1
    while i < len(parts) - 1 and parts[i] in ("-c", "-C"):
        i += 2
    return i if i < len(parts) else None


def is_safe_docker_write(parts):
    """Check if docker write command is safe to auto-approve."""
    if not parts or len(parts) < 2 or parts[0] != "docker":
        return False

    # docker exec is always safe (existing container's security context)
    if parts[1] == "exec":
        return True

    if parts[1] != "run":
        return False

    unsafe_flags = {"--privileged", "--cap-add", "--device", "--security-opt"}
    volume_flags = {"-v", "--volume", "--mount"}
    namespace_flags = {"--net", "--network", "--pid", "--ipc", "--uts", "--userns"}

    for i, part in enumerate(parts[2:], 2):
        flag = part.split("=")[0]
        if flag in unsafe_flags or flag in volume_flags:
            return False
        if flag in namespace_flags:
            val = part.split("=", 1)[1] if "=" in part else (parts[i + 1] if i + 1 < len(parts) else "")
            if val == "host":
                return False
    return True


def _ssh_targets_git_host(parts):
    """Check if SSH command targets a known git host."""
    git_hosts = {"github.com", "gitlab.com", "bitbucket.org"}
    return any(a in git_hosts or a.split("@")[-1] in git_hosts for a in parts[1:])


def is_git_gh_read(parts):
    """Check if command is a git, gh, or git-related SSH read operation."""
    if not parts:
        return False
    if parts[0] == "gh" or parts[0] == "git" or parts[0].endswith("/git"):
        return is_read_only(parts)
    # ssh -G only prints config (no connection) - safe regardless of caller
    if parts[0] == "ssh" and "-G" in parts and _ssh_targets_git_host(parts):
        return True
    return False

def is_claude_code(chain):
    """Check if process chain originates from Claude Code."""
    return any(name == "claude" for _, name, _, _ in chain)

def get_git_branch(cwd):
    """Get current git branch name from working directory. Returns branch name or None."""
    if not cwd:
        return None
    try:
        return subprocess.check_output(
            ["git", "-C", cwd, "branch", "--show-current"],
            stderr=subprocess.DEVNULL,
        ).decode().strip() or None
    except (subprocess.CalledProcessError, OSError):
        return None


def git_target_cwd(parts, safehouse_cwd):
    """Dir where a git command actually operates: `-C <path>` if present (resolved
    against safehouse_cwd when relative), else safehouse_cwd. Returns safehouse_cwd
    for non-git commands."""
    if not parts or not (parts[0] == "git" or parts[0].endswith("/git")):
        return safehouse_cwd
    target = None
    i = 1
    while i < len(parts) - 1 and parts[i] in ("-c", "-C"):
        if parts[i] == "-C":
            target = parts[i + 1]
        i += 2
    if target is None:
        return safehouse_cwd
    if not os.path.isabs(target):
        if not safehouse_cwd:
            return safehouse_cwd
        target = os.path.normpath(os.path.join(safehouse_cwd, target))
    return target


_OWNER_REPO_ISSUE_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9-]*/[A-Za-z0-9._-]+#\d+$")


def get_gh_target_repo(parts):
    """Extract target repo from gh command. Checks -R/--repo flag, URLs, and owner/repo#N args."""
    for i, part in enumerate(parts):
        if part in ("-R", "--repo") and i + 1 < len(parts):
            return parts[i + 1]
        if part.startswith("--repo="):
            return part.split("=", 1)[1]
    # Check positional args for URL or owner/repo#N patterns
    for part in parts:
        if part.startswith("-"):
            continue
        if part.startswith("https://github.com/"):
            segments = part[len("https://github.com/"):].split("/")
            if len(segments) >= 2:
                return "/".join(segments[:2])
        if _OWNER_REPO_ISSUE_RE.match(part):
            return part.split("#")[0]
    return None


def get_current_repo(cwd):
    """Get owner/repo from git remote origin. Returns None on failure."""
    if not cwd:
        return None
    try:
        url = subprocess.check_output(
            ["git", "-C", cwd, "remote", "get-url", "origin"],
            stderr=subprocess.DEVNULL,
        ).decode().strip()
        for prefix in ("https://github.com/", "git@github.com:"):
            if url.startswith(prefix):
                return url[len(prefix):].removesuffix(".git")
        return None
    except (subprocess.CalledProcessError, OSError):
        return None


def is_same_repo(parts, safehouse_cwd):
    """Check if gh command targets the checked-out repo. False if targeting a different repo."""
    target = get_gh_target_repo(parts)
    if target is None:
        return True
    current = get_current_repo(safehouse_cwd)
    if current is None:
        return False
    return target.lower().rstrip("/") == current.lower().rstrip("/")

def is_safe_gh_api_write(parts, safehouse_cwd):
    """Check if gh api write targets a safe endpoint (comments/reviews) on the same repo."""
    # Find API path (first non-flag arg after 'api')
    api_path = None
    for part in parts[2:]:
        if not part.startswith("-"):
            api_path = part
            break
    if not api_path:
        return False
    segments = api_path.split("/")
    # repos/owner/repo/.../comments or .../reviews
    if len(segments) < 4 or segments[0] != "repos":
        return False
    if segments[-1] not in ("comments", "reviews"):
        return False
    api_repo = "/".join(segments[1:3])
    current = get_current_repo(safehouse_cwd)
    if current is None:
        return False
    return api_repo.lower() == current.lower()

def is_safe_git_push(parts, safehouse_cwd):
    """Check if command is a safe push/PR op (not targeting main/master,
    unless the repo is listed in main-push-repos). Best-effort."""
    if not parts:
        return False

    protected = {"main", "master"}

    # gh pr/issue writes and run rerun (not merge)
    if parts[0] == "gh":
        if not is_same_repo(parts, safehouse_cwd):
            return False
        if len(parts) >= 3 and parts[1] == "pr" and parts[2] in ("create", "edit", "close", "comment"):
            return True
        if len(parts) >= 3 and parts[1] == "issue" and parts[2] == "comment":
            return True
        if len(parts) >= 3 and parts[1] == "run" and parts[2] == "rerun":
            return True
        if len(parts) >= 3 and parts[1] == "api":
            return is_safe_gh_api_write(parts, safehouse_cwd)
        return False

    # git push only
    if not (parts[0] == "git" or parts[0].endswith("/git")):
        return False
    ci = _git_subcommand_index(parts)
    if ci is None or parts[ci] != "push":
        return False

    # Don't auto-approve --all or --mirror (pushes all branches including protected)
    rest = parts[ci + 1:]
    if any(p in ("--all", "--mirror") for p in rest):
        return False

    # Parse: git push [flags...] [remote] [refspec...]
    args = [p for p in rest if not p.startswith("-")]

    # Exemption is keyed to the origin URL, so only lift protection for pushes to origin
    if auto_approve_main_push_repos and (not args or args[0] == "origin"):
        repo = get_current_repo(git_target_cwd(parts, safehouse_cwd))
        if repo and repo.lower() in auto_approve_main_push_repos:
            protected = set()

    # Check explicit refspecs for protected branch targets
    if len(args) >= 2:
        for refspec in args[1:]:
            if ":" in refspec:
                target = refspec.split(":")[-1].removeprefix("refs/heads/")
            else:
                target = refspec.removeprefix("refs/heads/")
            if target in protected:
                return False
            if target.upper() == "HEAD":
                break  # HEAD resolves to current branch — fall through
        else:
            return True  # All explicit refspecs are non-protected

    # No explicit branch or HEAD used - determine current branch from target cwd
    branch = get_git_branch(git_target_cwd(parts, safehouse_cwd))
    if branch:
        return branch not in protected
    return False

def notify(message):
    """Send macOS notification with sound. Click focuses terminal if alerter available."""
    def _alerter():
        """Run alerter, focus terminal on click."""
        try:
            result = subprocess.run(
                ["alerter", "--title", "Credential Request",
                 "--message", message, "--group", "credential-server",
                 "--sender", TERMINAL_BUNDLE_ID, "--timeout", "30"],
                capture_output=True, text=True,
            )
            if result.stdout.strip() in ("@CONTENTCLICKED", "@ACTIONCLICKED"):
                if WEZTERM_PANE:
                    # Switch tab and tag window title for AXRaise
                    subprocess.run(
                        ["wezterm", "cli", "activate-pane", "--pane-id", WEZTERM_PANE],
                        stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
                    )
                    subprocess.run(
                        ["wezterm", "cli", "set-window-title", "--pane-id", WEZTERM_PANE, WINDOW_TITLE],
                        stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
                    )
                # Raise specific window by title
                if TERMINAL_BUNDLE_ID:
                    subprocess.Popen(
                        ["osascript", "-e",
                         'tell application "System Events"'
                         f' to tell (first process whose bundle identifier is "{TERMINAL_BUNDLE_ID}")'
                         f' to perform action "AXRaise" of (first window whose name contains "{WINDOW_TITLE}")'],
                        stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
                    )
        except OSError:
            pass

    try:
        subprocess.Popen(
            ["afplay", "/System/Library/Sounds/Funk.aiff"],
            stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
        )
        if HAS_ALERTER and TERMINAL_BUNDLE_ID:
            threading.Thread(target=_alerter, daemon=True).start()
        else:
            subprocess.Popen(
                ["osascript",
                 "-e", f'display notification "{message}" with title "Credential Request"'],
                stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
            )
    except OSError:
        pass


def cleanup(*_):
    try:
        os.unlink(SOCKET_PATH)
    except OSError:
        pass
    try:
        os.unlink(DOCKER_PROXY_PATH)
    except OSError:
        pass
    sys.exit(0)


signal.signal(signal.SIGINT, cleanup)
signal.signal(signal.SIGTERM, cleanup)


_libc = ctypes.CDLL(ctypes.util.find_library("c"), use_errno=True)
# macOS sysctl mib: CTL_KERN=1, KERN_PROCARGS2=49. Returns int32 argc, exec_path,
# optional alignment padding, then argv[0..argc-1], then envp, then applep.
_KERN_PROCARGS2 = 49
_CTL_KERN = 1


def get_argv(pid):
    """Get real argv via KERN_PROCARGS2. Returns list[str] or None if unavailable.

    Unlike `ps -o args=`, this preserves the original token boundaries: a
    `-c key="val with spaces"` flag stays a single argv element instead of being
    re-split on whitespace.
    """
    try:
        mib = (ctypes.c_int * 3)(_CTL_KERN, _KERN_PROCARGS2, pid)
        size = ctypes.c_size_t(0)
        if _libc.sysctl(mib, 3, None, ctypes.byref(size), None, 0) != 0:
            return None
        buf = (ctypes.c_char * size.value)()
        if _libc.sysctl(mib, 3, buf, ctypes.byref(size), None, 0) != 0:
            return None
        data = buf.raw[:size.value]
        if len(data) < 4:
            return None
        argc = int.from_bytes(data[:4], "little")
        # Skip exec_path (null-terminated) and any zero-padding before argv[0].
        nul = data.find(b"\x00", 4)
        if nul == -1:
            return None
        offset = nul + 1
        while offset < len(data) and data[offset] == 0:
            offset += 1
        argv = []
        for _ in range(argc):
            nul = data.find(b"\x00", offset)
            if nul == -1:
                return None
            argv.append(data[offset:nul].decode("utf-8", "replace"))
            offset = nul + 1
        return argv
    except OSError:
        return None


def get_process_chain(pid, max_depth=16):
    """Walk the process tree up from pid.

    Each entry: (pid, name, args, argv). `args` is the ps display string (lossy
    for quoted values with spaces). `argv` is the real argv list; use it for
    parsing decisions, and `args` for display only. Falls back to args.split()
    when argv is unavailable (process exited, etc.)."""
    chain = []
    current = pid
    for _ in range(max_depth):
        if current <= 1:
            break
        try:
            out = subprocess.check_output(
                ["ps", "-p", str(current), "-o", "pid=,ppid=,args="],
                stderr=subprocess.DEVNULL,
            ).decode().strip()
            parts = out.split(None, 2)
            if len(parts) < 3:
                break
            args = parts[2]
            name = os.path.basename(args.split()[0]) if args else "?"
            argv = get_argv(int(parts[0]))
            if argv is None:
                argv = args.split()
            chain.append((int(parts[0]), name, args, argv))
            current = int(parts[1])
        except (subprocess.CalledProcessError, ValueError):
            break
    return chain


def get_cwd(pid):
    """Get working directory of a process."""
    try:
        out = subprocess.check_output(
            ["lsof", "-a", "-d", "cwd", "-p", str(pid), "-Fn"],
            stderr=subprocess.DEVNULL,
        ).decode()
        return next((l[1:] for l in out.splitlines() if l.startswith("n/")), None)
    except subprocess.CalledProcessError:
        return None


def _find_wrapper(argv, wrapper):
    """Return index of an argv token whose path ends with `wrapper` (e.g. 'bin/gh')."""
    for i, tok in enumerate(argv or ()):
        if tok.endswith("/" + wrapper) or tok == wrapper:
            return i
    return None


def parse_chain(chain):
    """Extract command and safehouse info from process chain.

    Returns (cmd, cmd_argv, cmd_pid, safehouse_pid, safehouse_cwd, actual_cmd).
    `cmd`/`actual_cmd` are display strings; `cmd_argv` is the canonical
    tokenized argv. Pass it to classifiers, not cmd.split().
    """
    cmd = None
    cmd_argv = None
    cmd_pid = None
    ssh_fallback = None
    ssh_fallback_argv = None
    safehouse_pid = None
    safehouse_cwd = None
    # Outermost ansible ancestor (workers may match too via fork-inherited argv;
    # we want the long-lived top-level so approval inheritance survives worker churn).
    ansible_match = None
    for pid, name, args, argv in chain:
        if ".credential-server.sock" in args:
            continue
        if "git-credential-helper" in args:
            continue
        if "libexec/git-core/" in args:
            continue
        if "aws-credential-process" in args:
            continue
        if "safehouse" in args:
            safehouse_pid = pid
            safehouse_cwd = get_cwd(pid)
            break
        # ansible/ansible-playbook is a python script, so argv[0] may be the
        # interpreter and argv[1] the script path. Track outermost match.
        if ssh_fallback is not None and argv:
            for i, token in enumerate(argv[:2]):
                base = os.path.basename(token)
                if base in ("ansible-playbook", "ansible"):
                    new_argv = [base] + argv[i + 1:]
                    ansible_match = (shlex.join(new_argv), new_argv, pid)
                    break
        # First non-skipped entry is the actual command
        if cmd is None:
            for wrapper in ("bin/gh", "bin/aws", "bin/ssh", "bin/docker"):
                idx = _find_wrapper(argv, wrapper)
                if idx is None:
                    continue
                tool = wrapper.split("/")[1]
                new_argv = [tool] + argv[idx + 1:]
                extracted = shlex.join(new_argv)
                if wrapper == "bin/ssh":
                    # Don't set cmd yet, look for parent (e.g. git)
                    ssh_fallback = extracted
                    ssh_fallback_argv = new_argv
                else:
                    cmd = extracted
                    cmd_argv = new_argv
                    cmd_pid = pid
                break
            else:
                if ssh_fallback is not None:
                    # ssh under git: attribute to the parent git command.
                    # Otherwise keep walking past intermediates (git-lfs, hooks, sh -c, etc.)
                    is_git = argv and (argv[0] == "git" or argv[0].endswith("/git"))
                    if is_git:
                        gi = _git_subcommand_index(argv)
                        if gi is not None and argv[gi] in (
                            "push", "fetch", "pull", "clone", "ls-remote",
                        ):
                            cmd = shlex.join(argv)
                            cmd_argv = list(argv)
                            cmd_pid = pid
                else:
                    cmd_argv = list(argv) if argv else args.split()
                    cmd = shlex.join(cmd_argv)
                    cmd_pid = pid
    if cmd is None and ansible_match is not None:
        cmd, cmd_argv, cmd_pid = ansible_match
    actual_cmd = ssh_fallback if ssh_fallback and cmd else None
    if cmd is None and ssh_fallback is not None:
        cmd = ssh_fallback
        cmd_argv = ssh_fallback_argv
    return cmd, cmd_argv, cmd_pid, safehouse_pid, safehouse_cwd, actual_cmd


def _sanitize_host(host):
    """Clamp a self-reported remote host label to a safe, short token.

    Remote requests supply their own host name; restrict it so it can't
    inject ANSI codes or newlines into the scope key, prompt, or audit log."""
    cleaned = re.sub(r"[^A-Za-z0-9._-]", "", host)[:64]
    return cleaned or "unknown"


def loopback_callback_port(url):
    """Port of a loopback redirect_uri in an OAuth authorize URL, else None."""
    try:
        params = urllib.parse.parse_qs(urllib.parse.urlparse(url).query)
        for value in params.get("redirect_uri", []):
            parsed = urllib.parse.urlparse(value)
            if parsed.hostname in ("127.0.0.1", "localhost", "::1") and parsed.port:
                return parsed.port
    except ValueError:
        pass
    return None


def _cancel_forward(host, spec):
    """Tear down a forward set up by forward_loopback."""
    try:
        subprocess.run(["ssh", "-O", "cancel", "-L", spec, host],
                       stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL,
                       stderr=subprocess.DEVNULL, timeout=10)
    except (subprocess.TimeoutExpired, OSError):
        pass


def forward_loopback(host, port):
    """Tunnel localhost:<port> to the same port on host, cancelled after a TTL.

    An OAuth authorize URL from a remote host redirects to a loopback listener
    that lives on that host, but we open the URL in the Mac's browser — without
    this forward the callback lands on the Mac's own (empty) port. Needs the
    shared ssh ControlMaster to be up; if it isn't, print the manual command."""
    spec = f"{port}:127.0.0.1:{port}"
    try:
        result = subprocess.run(["ssh", "-O", "forward", "-L", spec, host],
                                stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL,
                                stderr=subprocess.PIPE, timeout=10)
    except (subprocess.TimeoutExpired, OSError) as e:
        result = None
        error = str(e)
    else:
        error = result.stderr.decode(errors="replace").strip()
    if result is None or result.returncode != 0:
        print(f"  {RED}callback tunnel failed:{RESET} {error}")
        print(f"  {DIM}run manually: ssh -L {spec} {host}{RESET}")
        return
    mins = LOOPBACK_TUNNEL_TTL // 60
    print(f"  {GREEN}callback tunnel{RESET} {DIM}localhost:{port} → {host} "
          f"(cancels in {mins}min){RESET}")
    timer = threading.Timer(LOOPBACK_TUNNEL_TTL, _cancel_forward, args=(host, spec))
    timer.daemon = True
    timer.start()


# ssh flags that take a separate value argument (vs boolean or attached -pVAL form)
SSH_ARG_FLAGS = {"-B", "-b", "-c", "-D", "-E", "-e", "-F", "-I", "-i", "-J", "-L",
                 "-l", "-m", "-O", "-o", "-p", "-Q", "-R", "-S", "-W", "-w"}


def _ssh_host(parts):
    """Find destination host token in an ssh argv (skipping flag/value pairs)."""
    i = 1  # skip "ssh"
    while i < len(parts):
        tok = parts[i]
        if not tok.startswith("-"):
            return tok
        if tok in SSH_ARG_FLAGS:
            i += 2  # bare arg-flag consumes next token as its value
        else:
            i += 1  # boolean flag or attached form (e.g. -p22, -oKey=val)
    return None


def pattern_options(parts):
    """Pattern granularities for parts, recommended-first. Empty if too short."""
    if not parts:
        return []
    program = parts[0]
    options = []
    if program == "ssh":
        host = _ssh_host(parts)
        if host:
            options.append(("host", f"ssh {host}"))
    elif program in ("aws", "gh") and len(parts) >= 3:
        options.append(("service+verb", " ".join(parts[:3])))
        options.append(("service", " ".join(parts[:2])))
    elif len(parts) >= 2:
        options.append(("subcommand", " ".join(parts[:2])))
    exact = " ".join(parts)
    if not any(p == exact for _, p in options):
        options.append(("exact", exact))
    return options


def pattern_matches(parts, stored):
    """True if any of parts' available patterns equals the stored pattern."""
    return any(p == stored for _, p in pattern_options(parts))


def grant_time_left(safehouse_pid, grant):
    """Remaining-time label for a grant, or None once it is dead."""
    if grant["expires"] is not None:
        remaining = (grant["expires"] - datetime.now()).total_seconds()
        if remaining <= 0:
            return None
        mins, secs = divmod(int(remaining), 60)
        return f"{mins}m{secs:02d}s"
    # Remote scope (string identity): no live PID to probe. A session approval
    # lasts until this server process restarts.
    if isinstance(safehouse_pid, int) and not _pid_alive(safehouse_pid):
        return None
    return "sess"


def check_approval(safehouse_pid, cred_key, parts):
    """Find a live grant covering this command. Returns (grant, reason).

    `parts` is the tokenized argv list (used for read/pattern classification)."""
    key = (safehouse_pid, cred_key)
    if safehouse_pid is None or key not in approvals:
        return None, None

    live = [g for g in approvals[key] if grant_time_left(safehouse_pid, g) is not None]
    if len(live) != len(approvals[key]):
        if live:
            approvals[key] = live
        else:
            del approvals[key]
        save_approvals()

    for grant in live:
        time_left = grant_time_left(safehouse_pid, grant)
        mode = grant.get("mode", "all")
        if mode == "all":
            return grant, time_left
        if mode == "read" and is_read_only(parts):
            return grant, f"read {time_left}"
        if mode == "similar":
            stored = grant.get("pattern")
            if stored and pattern_matches(parts, stored):
                return grant, f"{stored} {time_left}"
    return None, None


def store_approval(safehouse_pid, cred_key, mode, duration, pattern=None, sensitive=False):
    """Add a grant to a credential. mode: read/similar/all, duration: 1/5/session.

    Grants accumulate: approving a second pattern keeps the first, so alternating
    commands under one credential don't evict each other. Re-approving the same
    (mode, pattern) replaces it.

    safehouse_pid is an int (local sandbox) or a "remote:<host>" string
    (remote mode) — both are valid scope identities. duration "session"
    leaves expires=None; for an int that means until the process exits, for
    a remote string until the server restarts (see check_approval)."""
    if safehouse_pid is None:
        return
    key = (safehouse_pid, cred_key)
    durations = {"1": timedelta(minutes=1), "5": timedelta(minutes=5)}
    expires = datetime.now() + durations[duration] if duration in durations else None
    grant = {"mode": mode, "expires": expires}
    if mode == "similar" and pattern:
        grant["pattern"] = pattern
    if sensitive:
        grant["sensitive"] = True
    kept = [g for g in approvals.get(key, [])
            if (g["mode"], g.get("pattern")) != (mode, grant.get("pattern"))
            and grant_time_left(safehouse_pid, g) is not None]
    approvals[key] = kept + [grant]
    save_approvals()


def save_approvals():
    """Persist current approvals to .approvals.toml."""
    entries = []
    now = datetime.now()
    for (pid, cred_key), grants in list(approvals.items()):
        # Remote scopes (string identity) are session-only — never persisted.
        if not isinstance(pid, int):
            continue
        for grant in grants:
            # Skip already-expired grants
            if grant["expires"] is not None and grant["expires"] <= now:
                continue
            rec = {"safehouse_pid": pid, "cred_key": cred_key, "mode": grant["mode"]}
            if grant["expires"] is not None:
                rec["expires"] = grant["expires"].strftime("%Y-%m-%dT%H:%M:%S")
            if "pattern" in grant:
                rec["pattern"] = grant["pattern"]
            if grant.get("sensitive"):
                rec["sensitive"] = True
            entries.append(rec)
    lines = ["# Auto-managed by credential-server. Edit at your own risk.", ""]
    for rec in entries:
        lines.append("[[approvals]]")
        for k, v in rec.items():
            if isinstance(v, bool):
                lines.append(f"{k} = {str(v).lower()}")
            elif isinstance(v, int):
                lines.append(f"{k} = {v}")
            else:
                lines.append(f'{k} = "{v}"')
        lines.append("")
    for pid, cred_key in sorted(k for k in sensitive_scopes if isinstance(k[0], int)):
        lines.append("[[sensitive]]")
        lines.append(f"safehouse_pid = {pid}")
        lines.append(f'cred_key = "{cred_key}"')
        lines.append("")
    try:
        with open(APPROVALS_PATH, "w") as f:
            f.write("\n".join(lines))
    except OSError as e:
        print(f"  {RED}warning: failed to save approvals: {e}{RESET}")


def load_approvals():
    """Load .approvals.toml. Returns (approvals, sensitive_scopes), pruning
    expired and dead-PID entries."""
    if tomllib is None or not os.path.exists(APPROVALS_PATH):
        return {}, set()
    try:
        with open(APPROVALS_PATH, "rb") as f:
            data = tomllib.load(f)
    except (tomllib.TOMLDecodeError, OSError) as e:
        print(f"  {RED}warning: failed to load approvals: {e}{RESET}")
        return {}, set()
    now = datetime.now()
    result = {}
    for rec in data.get("approvals", []):
        pid = rec.get("safehouse_pid")
        cred_key = rec.get("cred_key")
        if not pid or not cred_key:
            continue
        # Only int safehouse PIDs persist; reject any hand-edited string scope.
        if not isinstance(pid, int):
            continue
        # Check PID still alive
        if not _pid_alive(pid):
            continue
        # Check expiry
        expires = None
        if "expires" in rec:
            try:
                expires = datetime.strptime(rec["expires"], "%Y-%m-%dT%H:%M:%S")
            except ValueError:
                continue
            if expires <= now:
                continue
        grant = {"mode": rec.get("mode", "all"), "expires": expires}
        if "pattern" in rec:
            grant["pattern"] = rec["pattern"]
        if rec.get("sensitive"):
            grant["sensitive"] = True
        result.setdefault((pid, cred_key), []).append(grant)
    scopes = set()
    for rec in data.get("sensitive", []):
        pid = rec.get("safehouse_pid")
        cred_key = rec.get("cred_key")
        if isinstance(pid, int) and cred_key and _pid_alive(pid):
            scopes.add((pid, cred_key))
    return result, scopes

def read_key(timeout=None):
    """Read a single keypress without requiring Enter. Returns None on timeout."""
    fd = sys.stdin.fileno()
    if os.isatty(fd):
        old = termios.tcgetattr(fd)
        try:
            tty.setraw(fd)
            if timeout is not None:
                ready, _, _ = select.select([fd], [], [], timeout)
                if not ready:
                    return None
            ch = sys.stdin.read(1)
        finally:
            termios.tcsetattr(fd, termios.TCSADRAIN, old)
        return ch
    else:
        # Fallback for non-TTY (testing)
        line = sys.stdin.readline()
        return line[0] if line else "d"


def _print_chain_debug(chain):
    """Print process chain for debugging approval decisions."""
    print(f"  {DIM}--- process chain (child -> parent) ---{RESET}")
    for i, (pid, name, args, _argv) in enumerate(chain):
        # Tag known process types
        tags = []
        if name == "claude":
            tags.append(f"{CYAN}claude-code{RESET}")
        elif name == "opencode":
            tags.append(f"{CYAN}opencode{RESET}")
        elif "safehouse" in args:
            tags.append(f"{GREEN}sandbox{RESET}")
        elif name == "git" or args.split()[0].endswith("/git"):
            tags.append("git")
        elif name in ("node", "python3", "bash", "zsh", "sh"):
            tags.append(name)
        for wrapper in ("bin/gh", "bin/aws", "bin/ssh", "bin/docker"):
            if wrapper in args:
                tags.append(f"{CYAN}wrapper{RESET}")
                break
        tag_str = f" {DIM}({', '.join(tags)}){RESET}" if tags else ""
        indent = "    " + "  " * i
        # Truncate long args
        display_args = args if len(args) <= 120 else args[:117] + "..."
        print(f"{indent}{DIM}pid={pid}{RESET} {BOLD}{name}{RESET}{tag_str}")
        print(f"{indent}  {DIM}{display_args}{RESET}")
    print(f"  {DIM}---{RESET}")


def _radio(segments):
    """Join [(key, label, active)] into a row, active option green, rest dim."""
    out = []
    for key, label, active in segments:
        if active:
            out.append(f"{DIM}[{key}]{RESET} {GREEN}{BOLD}{label}{RESET}")
        else:
            out.append(f"{DIM}[{key}] {label}{RESET}")
    return "  ".join(out)


def _visible_len(s):
    """Length of s ignoring ANSI escape sequences."""
    return len(re.sub(r"\x1b\[[0-9;]*m", "", s))


def _truncate_visible(s, width):
    """Truncate s to width visible columns, preserving ANSI codes and appending RESET."""
    if _visible_len(s) <= width:
        return s
    out, seen = [], 0
    for tok in re.split(r"(\x1b\[[0-9;]*m)", s):
        if tok.startswith("\x1b"):
            out.append(tok)
            continue
        for ch in tok:
            if seen >= width:
                return "".join(out) + RESET
            out.append(ch)
            seen += 1
    return "".join(out) + RESET


def prompt_approval(cmd, parts, chain=(), cred_label="", origin="this sandbox"):
    """Single-screen approval form. Returns (action, mode, duration, pattern, sensitive).
    action: 'once'/'deny'/'approve'
    mode: 'read'/'similar'/'all' (only for 'approve')
    duration: '1'/'5'/'session' (only for 'approve')
    pattern: chosen pattern string (only for mode='similar')
    sensitive: include sensitive commands in a persistent approval (only for 'approve')

    `cmd` is the display string (shown in notifications); `parts` is the argv
    list used for classification. `cred_label` names the credential the
    approval is keyed on (e.g. "ssh coding26", "aws my-profile"); `origin`
    names the approval's session scope.
    """
    pattern_opts = pattern_options(parts)

    scope = "once"
    duration = "5"
    pat_idx = 0
    sensitive = is_sensitive(parts)
    cred = cred_label or "these"

    def grant_summary():
        if scope == "once":
            return f"{DIM}this command only{RESET}"
        if scope == "reads":
            what = f"read-only {CYAN}{cred}{RESET}{DIM} commands"
        elif scope == "pattern":
            what = f"{CYAN}{cred}{RESET}{DIM} commands matching {RESET}{CYAN}{pattern_opts[pat_idx][1]}{RESET}{DIM}"
        else:
            what = f"all {CYAN}{cred}{RESET}{DIM} commands"
        dur = {"1": "for 1min", "5": "for 5min", "session": "until session ends"}[duration]
        bits = [what, origin, dur]
        if sensitive:
            bits.append("incl. sensitive")
        return f"{DIM}" + " · ".join(bits) + RESET

    def render_lines():
        persistent = scope in ("reads", "pattern", "all")
        scopes = [("o", "once"), ("r", "reads")]
        if pattern_opts:
            scopes.append(("p", "pattern"))
        scopes.append(("a", "all"))
        lines = ["  scope:    " + _radio([(k, l, scope == l) for k, l in scopes])]
        if persistent:
            lines.append("  duration: " + _radio([
                ("1", "1min", duration == "1"),
                ("5", "5min", duration == "5"),
                ("s", "session", duration == "session"),
            ]))
        if scope == "pattern" and len(pattern_opts) > 1:
            label, pat = pattern_opts[pat_idx]
            lines.append(f"  pattern:  {DIM}[g]{RESET} {label}: {CYAN}{pat}{RESET} "
                         f"{DIM}({pat_idx + 1}/{len(pattern_opts)}){RESET}")
        if persistent:
            box = f"{GREEN}[x]{RESET}" if sensitive else f"{DIM}[ ]{RESET}"
            lines.append(f"  {DIM}[t]{RESET} include sensitive {box}")
        lines.append(f"  grants:   {grant_summary()}")
        hint = f"  {DIM}[Enter] confirm  [d] deny" + ("  [?] info" if chain else "") + RESET
        lines.append(hint)
        width = shutil.get_terminal_size().columns - 1
        return [_truncate_visible(line, width) for line in lines]

    def draw(prev_n):
        # Cursor sits at the start of the last drawn line; step back to line 0.
        if prev_n > 1:
            sys.stdout.write(f"\033[{prev_n - 1}A")
        sys.stdout.write("\r\033[J")
        lines = render_lines()
        sys.stdout.write("\r\n".join(lines) + "\r")
        sys.stdout.flush()
        return len(lines)

    def confirm():
        dur_label = {"1": "1min", "5": "5min", "session": "session"}[duration]
        if scope == "reads":
            bits = ["reads", cred, dur_label]
        elif scope == "pattern":
            bits = [pattern_opts[pat_idx][1], cred, dur_label]
        elif scope == "all":
            bits = ["all", cred, dur_label]
        else:
            bits = ["once"]
        if scope != "once" and sensitive:
            bits.append("+sensitive")
        sys.stdout.write("\r\n" + f"  {GREEN}\u2713 " + " \u00b7 ".join(bits) + RESET + "\r\n")
        sys.stdout.flush()

    def deny():
        sys.stdout.write("\r\n" + f"  {RED}denied{RESET}\r\n")
        sys.stdout.flush()
        return "deny", None, None, None, False

    deadline = time.monotonic() + PROMPT_TIMEOUT
    next_reminder = time.monotonic() + REMINDER_INTERVAL
    prev_n = draw(0)
    while True:
        now = time.monotonic()
        remaining = deadline - now
        if remaining <= 0:
            sys.stdout.write("\r\n")
            print(f"  {RED}timed out{RESET}")
            return "deny", None, None, None, False
        if now >= next_reminder:
            notify(cmd)
            next_reminder = now + REMINDER_INTERVAL
        wait = min(remaining, next_reminder - now)
        ch = read_key(timeout=max(0.1, wait))
        if ch is None:
            continue
        if ch == "\x03":  # Ctrl+C
            sys.stdout.write("\r\n")
            cleanup()
        elif ch == "\x1b" or ch in ("d", "D"):
            return deny()
        elif ch == "?" and chain:
            sys.stdout.write("\r\n")
            _print_chain_debug(chain)
            prev_n = draw(0)
            continue
        elif ch in ("\r", "\n"):
            confirm()
            if scope == "once":
                return "once", None, None, None, False
            if scope == "reads":
                return "approve", "read", duration, None, sensitive
            if scope == "pattern":
                return "approve", "similar", duration, pattern_opts[pat_idx][1], sensitive
            return "approve", "all", duration, None, sensitive
        elif ch in ("o", "O"):
            scope = "once"
        elif ch in ("r", "R"):
            scope = "reads"
        elif ch in ("p", "P") and pattern_opts:
            scope = "pattern"
        elif ch in ("a", "A"):
            scope = "all"
        elif ch == "1":
            duration = "1"
        elif ch == "5":
            duration = "5"
        elif ch in ("s", "S"):
            duration = "session"
        elif ch in ("g", "G") and scope == "pattern" and len(pattern_opts) > 1:
            pat_idx = (pat_idx + 1) % len(pattern_opts)
        elif ch in ("t", "T") and scope in ("reads", "pattern", "all"):
            sensitive = not sensitive
        else:
            continue
        prev_n = draw(prev_n)


def prompt_open(cmd):
    """Prompt for a remote URL open. Returns True to open."""
    sys.stdout.write(f"  {DIM}[Enter]{RESET}open  {DIM}[d]{RESET}eny ▸ ")
    sys.stdout.flush()
    deadline = time.monotonic() + PROMPT_TIMEOUT
    next_reminder = time.monotonic() + REMINDER_INTERVAL
    while True:
        now = time.monotonic()
        remaining = deadline - now
        if remaining <= 0:
            print(f"{RED}timed out{RESET}")
            return False
        if now >= next_reminder:
            notify(cmd)
            next_reminder = now + REMINDER_INTERVAL
        wait = min(remaining, next_reminder - time.monotonic())
        ch = read_key(timeout=max(0.1, wait))
        if ch is None:
            continue
        if ch == "\x03":
            print()
            cleanup()
        elif ch == "\x1b" or ch in ("d", "D"):
            print(f"{RED}denied{RESET}")
            return False
        elif ch in ("\r", "\n"):
            print(f"{GREEN}open{RESET}")
            return True


def prompt_sensitive(cmd=None):
    """Prompt for sensitive command escalation.

    Returns 'once'/'deny'/'remember' (rest of the current approval) or
    'session' (rest of the scope, surviving re-approvals)."""
    sys.stdout.write(f"  {DIM}[Enter]{RESET}once  {DIM}[d]{RESET}eny  "
                     f"{DIM}[r]{RESET}emember  {DIM}[s]{RESET}ession ▸ ")
    sys.stdout.flush()
    deadline = time.monotonic() + PROMPT_TIMEOUT
    next_reminder = time.monotonic() + REMINDER_INTERVAL
    while True:
        now = time.monotonic()
        remaining = deadline - now
        if remaining <= 0:
            print(f"{RED}timed out{RESET}")
            return "deny"
        if now >= next_reminder:
            if cmd:
                notify(cmd)
            next_reminder = now + REMINDER_INTERVAL
        wait = min(remaining, next_reminder - time.monotonic())
        ch = read_key(timeout=max(0.1, wait))
        if ch is None:
            continue
        if ch == "\x03":
            print()
            cleanup()
        elif ch == "\x1b" or ch in ("d", "D"):
            print(f"{RED}denied{RESET}")
            return "deny"
        elif ch in ("\r", "\n"):
            print(f"{GREEN}once{RESET}")
            return "once"
        elif ch in ("r", "R"):
            print(f"{GREEN}remember{RESET}")
            return "remember"
        elif ch in ("s", "S"):
            print(f"{GREEN}session{RESET}")
            return "session"


def fetch_gh_token():
    """Fetch gh auth token. Returns token string or None."""
    try:
        return subprocess.check_output(
            ["gh", "auth", "token"], stderr=subprocess.DEVNULL
        ).decode().strip()
    except subprocess.CalledProcessError:
        return None


# AWS credential cache: {profile: {"creds": {...}, "expires": datetime}}
aws_cache = {}


EXPIRED_MARKERS = ("sso session", "token has expired", "sso login")


def fetch_aws_credentials(profile, quiet=False):
    """Fetch temporary AWS credentials for profile, with caching.
    Returns (status, creds) with status in ("ok", "expired", "error")."""
    # Check cache
    if profile in aws_cache:
        entry = aws_cache[profile]
        remaining = (entry["expires"] - datetime.now()).total_seconds()
        # Use cache if >60s remaining
        if remaining > 60:
            return "ok", entry["creds"]
        del aws_cache[profile]

    try:
        result = subprocess.run(
            ["aws", "configure", "export-credentials", "--profile", profile],
            capture_output=True,
            text=True,
            check=True,
        )
        data = json.loads(result.stdout)
        creds = {
            "access_key_id": data["AccessKeyId"],
            "secret_access_key": data["SecretAccessKey"],
            "session_token": data.get("SessionToken", ""),
            "expiration": data.get("Expiration", ""),
        }
        # Cache until expiration
        if "Expiration" in data:
            expires_utc = datetime.fromisoformat(data["Expiration"].replace("Z", "+00:00"))
            expires_local = expires_utc.astimezone().replace(tzinfo=None)
            aws_cache[profile] = {"creds": creds, "expires": expires_local}
        return "ok", creds
    except subprocess.CalledProcessError as e:
        stderr = (e.stderr or "").lower()
        if any(m in stderr for m in EXPIRED_MARKERS):
            return "expired", None
        if not quiet:
            detail = (e.stderr or "").strip() or str(e)
            print(f"  {RED}error: failed to get AWS credentials for {profile}: {detail}{RESET}")
        return "error", None
    except (json.JSONDecodeError, KeyError) as e:
        if not quiet:
            print(f"  {RED}error: failed to parse AWS credentials for {profile}: {e}{RESET}")
        return "error", None


def run_aws_sso_login(profile, timeout=180):
    """Run `aws sso login --profile X` in-terminal. Returns creds dict or None."""
    cred_key = f"aws:{profile}"
    cmd_repr = f"aws sso login --profile {profile}"
    print(f"\n  {YELLOW}\u26a0 AWS session expired: {profile}{RESET}")
    print(f"  {DIM}Running: {cmd_repr}{RESET}")
    notify(f"AWS login needed: {profile}")
    audit_log(cmd_repr, cred_key, "aws-sso-login:start")

    outcome, reason, creds = "fail", None, None
    try:
        result = subprocess.run(
            ["aws", "sso", "login", "--profile", profile],
            timeout=timeout,
        )
        if result.returncode != 0:
            reason = f"SSO login exited {result.returncode}"
        else:
            status, creds = fetch_aws_credentials(profile, quiet=True)
            if creds is not None:
                outcome = "success"
            else:
                reason = f"login succeeded but credentials still unavailable ({status})"
    except subprocess.TimeoutExpired:
        outcome, reason = "timeout", "SSO login timed out"
    except OSError as e:
        reason = f"SSO login failed: {e}"

    audit_log(cmd_repr, cred_key, f"aws-sso-login:{outcome}")
    if outcome == "success":
        print(f"  {GREEN}\u2713 signed in{RESET}")
        return creds
    print(f"  {RED}\u2717 {reason}{RESET}")
    return None


def handle_request(conn, request, safehouse_pid, safehouse_cwd, cmd, cmd_argv=None, cmd_pid=None, chain=(), actual_cmd=None):
    """Handle a credential request. Returns response dict.

    `cmd` is the display string; `cmd_argv` is the canonical tokenized argv
    (passed to classifiers). When cmd_argv is None, falls back to cmd.split()."""
    global approved_pids
    req_type = request.get("type")
    is_remote = request.get("origin") == "remote"

    # Orphaned request: peer's safehouse ancestor is gone (sandbox exited while
    # request sat in the accept queue). Drop silently rather than prompting the
    # user about a dead session. Remote requests legitimately have no safehouse
    # ancestor — the main loop gives them a synthetic "remote:<host>" identity.
    if safehouse_pid is None and not is_remote:
        print(f"\n  {DIM}skipped (orphaned, no safehouse ancestor): {cmd}{RESET}")
        return {"ok": False}

    # Remote mode serves AWS and URL opens (gh/git/docker remote is out of scope).
    if is_remote and req_type not in ("aws", "open"):
        print(f"\n  {RED}remote: only aws and open supported, got {req_type}{RESET}")
        return {"ok": False}

    # URL open: http(s):// only, runs /usr/bin/open outside sandbox. Sandbox
    # opens are auto-approved; remote ones prompt, since any process on the
    # remote host can reach the forwarded socket.
    if req_type == "open":
        url = request.get("url", "")
        if not url.startswith(("http://", "https://")):
            print(f"\n  {RED}open: rejected non-http(s) URL{RESET}")
            return {"ok": False}
        if is_remote:
            host_label = safehouse_pid.split(":", 1)[1]
            port = loopback_callback_port(url)
            try:
                conn.getpeername()
            except OSError:
                print(f"  {DIM}skipped (disconnected): open{RESET}")
                return {"ok": False}
            notify(cmd)
            loc = f"{CYAN}{safehouse_cwd}{RESET}  " if safehouse_cwd else ""
            print(f"\n  {loc}{YELLOW}remote/{host_label}{RESET}")
            print(f"  {YELLOW}[remote]{RESET} {BOLD}open{RESET} {url}")
            if port:
                print(f"  {DIM}callback: tunnels localhost:{port} → {host_label}{RESET}")
            if not prompt_open(cmd):
                audit_log(f"open {url}", "open", "denied", safehouse_cwd, origin=safehouse_pid)
                return {"ok": False}
            if port:
                forward_loopback(host_label, port)
        try:
            result = subprocess.run(
                ["/usr/bin/open", url],
                stdin=subprocess.DEVNULL,
                stdout=subprocess.DEVNULL,
                stderr=subprocess.PIPE,
                timeout=5,
            )
            if result.returncode != 0:
                print(f"\n  {RED}open failed: {result.stderr.decode(errors='replace').strip()}{RESET}")
                return {"ok": False}
        except (subprocess.TimeoutExpired, OSError) as e:
            print(f"\n  {RED}open failed: {e}{RESET}")
            return {"ok": False}
        cwd_short = os.path.basename(safehouse_cwd) if safehouse_cwd else ""
        tags = ["open"] + (["remote"] if is_remote else []) + ([cwd_short] if cwd_short else [])
        display_url = url if len(url) <= 100 else url[:97] + "..."
        print(f"\n  {GREEN}✓{RESET} {DIM}({', '.join(tags)}){RESET}\n    open {display_url}")
        audit_log(f"open {url}", "open", "once" if is_remote else "auto-open",
                  safehouse_cwd, origin=safehouse_pid if is_remote else None)
        return {"ok": True}

    # For remote requests safehouse_pid is the "remote:<host>" scope string.
    audit_origin = safehouse_pid if is_remote else None

    cmd_parts = list(cmd_argv) if cmd_argv is not None else (cmd.split() if cmd else [])
    cmd_cwd = get_cwd(cmd_pid) if cmd_pid else None
    effective_cwd = cmd_cwd or safehouse_cwd
    target_cwd = git_target_cwd(cmd_parts, effective_cwd)

    display_cwd = target_cwd or safehouse_cwd
    if is_remote:
        host_label = safehouse_pid.split(":", 1)[1]
        loc = f"{CYAN}{display_cwd}{RESET}  " if display_cwd else ""
        session_label = f"{loc}{YELLOW}remote/{host_label}{RESET}"
    elif display_cwd:
        session_label = f"{CYAN}{display_cwd}{RESET}  {DIM}safehouse/{safehouse_pid}{RESET}"
    else:
        session_label = f"safehouse/{safehouse_pid}"

    git_branch = None
    if req_type == "gh":
        cred_key = "gh"
        detail = ""
    elif req_type == "aws":
        profile = request.get("profile", "default")
        cred_key = f"aws:{profile}"
        detail = f"\n  profile: {CYAN}{profile}{RESET}"
    elif req_type == "az":
        cred_key = "az"
        detail = ""
    elif req_type == "ssh":
        ssh_host = _ssh_host(cmd_parts) if cmd_parts[:1] == ["ssh"] else None
        cred_key = f"ssh:{ssh_host}" if ssh_host else "ssh"
        detail = ""
    elif req_type == "docker":
        cred_key = "docker"
        detail = ""
    elif req_type == "netrc":
        machine = request.get("machine", "")
        cred_key = f"netrc:{machine}"
        detail = f"\n  machine: {CYAN}{machine}{RESET}"
    else:
        return {"ok": False}

    # Resolve git branch for git/gh commands (any credential type)
    if cmd_parts and (cmd_parts[0] in ("git", "gh") or cmd_parts[0].endswith("/git")):
        git_branch = get_git_branch(target_cwd)
        if git_branch:
            detail = f"\n  branch: {CYAN}{git_branch}{RESET}"

    # Check pre-approval
    key = (safehouse_pid, cred_key)
    grant, reason = check_approval(safehouse_pid, cred_key, cmd_parts)
    if grant:
        # Sensitive commands need additional escalation even when auto-approved
        if (is_sensitive(cmd_parts) and key not in sensitive_scopes
                and not grant.get("sensitive")):
            try:
                conn.getpeername()
            except OSError:
                print(f"  {DIM}skipped (disconnected): {cmd}{RESET}")
                return {"ok": False}
            notify(cmd)
            print(f"\n  {session_label}{detail}")
            print(f"  {BOLD}⚠ sensitive:{RESET} {RED}{cmd}{RESET}")
            print(f"  {DIM}approved: {reason}{RESET}")
            sensitive_choice = prompt_sensitive(cmd)
            if sensitive_choice == "deny":
                audit_log(cmd, cred_key, "sensitive:denied", safehouse_cwd, origin=audit_origin)
                return {"ok": False}
            if sensitive_choice == "remember":
                grant["sensitive"] = True
                save_approvals()
            elif sensitive_choice == "session":
                sensitive_scopes.add(key)
                save_approvals()
            audit_log(cmd, cred_key, f"sensitive:{sensitive_choice}", safehouse_cwd, origin=audit_origin)
        else:
            tags = [reason]
            if is_remote:
                tags.append("remote")
            if is_sensitive(cmd_parts):
                tags.append("sensitive")
            if req_type == "aws":
                tags.insert(0, request.get("profile", "default"))
            if git_branch:
                tags.append(f"branch:{git_branch}")
            cwd_short = os.path.basename(safehouse_cwd) if safehouse_cwd else ""
            if cwd_short:
                tags.append(cwd_short)
            print(approved_line(tags, cmd_parts, cmd, actual_cmd))
            audit_log(cmd, cred_key, reason, safehouse_cwd, origin=audit_origin)
        creds = _fetch_creds(req_type, request)
        if creds is not None:
            return {"ok": True, **creds}
        return {"ok": False}


    # Auto-approve git/gh reads and safe pushes
    auto_reason = None
    if auto_approve_git_reads and is_git_gh_read(cmd_parts):
        auto_reason = "read"
    elif cmd_parts[:3] == ["gh", "auth", "token"] and is_claude_code(chain):
        auto_reason = "claude-auth"
    # ssh -T to git hosts from Claude Code (connectivity probe)
    elif (auto_approve_git_reads and cmd_parts and cmd_parts[0] == "ssh"
          and "-T" in cmd_parts and _ssh_targets_git_host(cmd_parts)
          and is_claude_code(chain)):
        auto_reason = "claude-ssh"
    elif auto_approve_git_push and is_safe_git_push(cmd_parts, effective_cwd):
        auto_reason = "push"
    elif auto_approve_docker_reads and cmd_parts and cmd_parts[0] == "docker" and is_read_only(cmd_parts):
        auto_reason = "read"
    elif auto_approve_docker_reads and is_safe_docker_write(cmd_parts):
        auto_reason = "safe-run"
    elif auto_approve_netrc and req_type == "netrc":
        auto_reason = "netrc"

    # Check if any ancestor process was already approved
    if auto_reason is None and approved_pids:
        chain_pids = {pid for pid, _, _, _ in chain}
        matched = chain_pids & approved_pids
        if matched:
            # Clean up dead PIDs while we're here
            approved_pids -= {p for p in matched if not _pid_alive(p)}
            if chain_pids & approved_pids:
                auto_reason = "child"

    if auto_reason:
        creds = _fetch_creds(req_type, request)
        if creds is not None:
            if cmd_pid:
                approved_pids.add(cmd_pid)
            tags = ["auto", auto_reason] + ([f"branch:{git_branch}"] if git_branch else [])
            print(approved_line(tags, cmd_parts, cmd, actual_cmd))
            audit_log(cmd, cred_key, f"auto-{auto_reason}", safehouse_cwd, origin=audit_origin)
            return {"ok": True, **creds}

    # Auto-approve configured AWS profiles
    if req_type == "aws":
        profile = request.get("profile", "default")
        if is_auto_approved_profile(profile):
            creds = _fetch_creds(req_type, request)
            # Prompting can't help \u2014 approval would just re-run the same failing fetch
            if creds is None:
                audit_log(cmd, cred_key, "auto-config-failed", safehouse_cwd, origin=audit_origin)
                return {"ok": False}
            tags = ["auto", "config", profile] + (["remote"] if is_remote else [])
            print(approved_line(tags, cmd_parts, cmd))
            audit_log(cmd, cred_key, "auto-config", safehouse_cwd, origin=audit_origin)
            return {"ok": True, **creds}

    # Skip if caller already disconnected (queued requests)
    try:
        conn.getpeername()
    except OSError:
        print(f"  {DIM}skipped (disconnected): {cmd}{RESET}")
        return {"ok": False}

    notify(cmd)
    is_read = is_read_only(cmd_parts)
    if is_read:
        cmd_display = f"{BOLD}{cmd}{RESET} {DIM}(read){RESET}"
    else:
        cmd_display = f"{RED}{BOLD}{cmd}{RESET} {DIM}(write){RESET}"
    print(f"\n  {session_label}")
    remote_tag = f"{YELLOW}[remote]{RESET} " if is_remote else ""
    print(f"  {remote_tag}{cmd_display}{detail}")
    cred_label = cred_key.replace(":", " ", 1)
    origin = f"any session on {host_label}" if is_remote else "this sandbox"
    action, mode, duration, chosen_pattern, sensitive = prompt_approval(
        cmd, cmd_parts, chain, cred_label=cred_label, origin=origin)

    if action == "deny":
        return {"ok": False}

    creds = _fetch_creds(req_type, request)
    if creds is None:
        return {"ok": False}

    if action == "approve":
        store_approval(safehouse_pid, cred_key, mode, duration, chosen_pattern, sensitive)

    if cmd_pid:
        approved_pids.add(cmd_pid)

    audit_log(cmd, cred_key, f"{action}:{mode}:{duration}" if mode else action, safehouse_cwd, origin=audit_origin)
    return {"ok": True, **creds}


def _fetch_creds(req_type, request):
    """Fetch credentials based on request type."""
    if req_type == "gh":
        token = fetch_gh_token()
        return {"token": token} if token else None
    elif req_type == "aws":
        profile = request.get("profile", "default")
        status, creds = fetch_aws_credentials(profile)
        if creds is None and status == "expired":
            creds = run_aws_sso_login(profile)
        return creds
    elif req_type == "az":
        return {}
    elif req_type == "ssh":
        return {}
    elif req_type == "docker":
        return {}  # approval-only, access gated by proxy
    elif req_type == "netrc":
        return _fetch_netrc(request.get("machine", ""))
    return None


def _fetch_netrc(machine):
    """Read credentials for machine from ~/.netrc."""
    try:
        nrc = netrc_mod.netrc()
        auth = nrc.authenticators(machine)
        if auth:
            login, _, password = auth
            return {"login": login or "", "password": password or ""}
    except (FileNotFoundError, netrc_mod.NetrcParseError) as e:
        print(f"  {RED}netrc error: {e}{RESET}")
    return None


def _detect_docker_socket():
    """Find the real Docker socket. Returns path or None."""
    docker_host = os.environ.get("DOCKER_HOST", "")
    if docker_host.startswith("unix://"):
        path = docker_host[len("unix://"):]
        if os.path.exists(path):
            return path
    for path in [
        "/var/run/docker.sock",
        os.path.expanduser("~/.docker/run/docker.sock"),
        os.path.expanduser("~/.colima/default/docker.sock"),
        os.path.expanduser("~/.orbstack/run/docker.sock"),
    ]:
        if os.path.exists(path):
            return path
    return None


def _docker_proxy_conn(client, docker_sock_path):
    """Proxy a single connection to the real Docker socket."""
    docker = None
    try:
        peer_pid = struct.unpack("i", client.getsockopt(SOL_LOCAL, LOCAL_PEERPID, 4))[0]
        chain = get_process_chain(peer_pid)
        chain_pids = {pid for pid, _, _, _ in chain}

        if not (chain_pids & approved_pids):
            return

        docker = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        docker.connect(docker_sock_path)

        def forward(src, dst):
            try:
                while True:
                    data = src.recv(65536)
                    if not data:
                        break
                    dst.sendall(data)
            except OSError:
                pass
            try:
                dst.shutdown(socket.SHUT_WR)
            except OSError:
                pass

        t1 = threading.Thread(target=forward, args=(client, docker), daemon=True)
        t2 = threading.Thread(target=forward, args=(docker, client), daemon=True)
        t1.start()
        t2.start()
        t1.join()
        t2.join()
    except OSError:
        pass
    finally:
        try:
            client.close()
        except OSError:
            pass
        if docker:
            try:
                docker.close()
            except OSError:
                pass


def _docker_proxy_listener(proxy_sock, docker_sock_path):
    """Accept and proxy Docker connections in a loop."""
    while True:
        try:
            client, _ = proxy_sock.accept()
            threading.Thread(
                target=_docker_proxy_conn,
                args=(client, docker_sock_path),
                daemon=True,
            ).start()
        except OSError:
            break


# --- Main ---

# Restore persisted approvals
restored, restored_sensitive = load_approvals()
if restored:
    flat = [(pid, cred_key, grant)
            for (pid, cred_key), grants in restored.items() for grant in grants]
    n_timed = sum(1 for _, _, g in flat if g["expires"] is not None)
    n_session = len(flat) - n_timed
    parts = []
    if n_timed:
        parts.append(f"{n_timed} timed")
    if n_session:
        parts.append(f"{n_session} session")
    print(f"found {len(flat)} persisted approval(s) ({', '.join(parts)})")
    for pid, cred_key, grant in flat:
        if grant["expires"]:
            remaining = (grant["expires"] - datetime.now()).total_seconds()
            mins, secs = divmod(int(remaining), 60)
            exp = f"{mins}m{secs:02d}s"
        else:
            exp = "session"
        label = grant.get("pattern", grant["mode"])
        sens = grant.get("sensitive") or (pid, cred_key) in restored_sensitive
        print(f"  pid={pid} {cred_key} {label} {exp}{' +sensitive' if sens else ''}")
    sys.stdout.write(f"restore these? {DIM}[Enter]{RESET}yes  {DIM}[r]{RESET}eset \u25b8 ")
    sys.stdout.flush()
    ch = read_key()
    if ch in ("r", "R"):
        print(f"{RED}reset{RESET}")
        try:
            os.unlink(APPROVALS_PATH)
        except OSError:
            pass
    else:
        approvals.update(restored)
        sensitive_scopes.update(restored_sensitive)
        print(f"{GREEN}restored{RESET}")

if os.path.exists(SOCKET_PATH):
    os.unlink(SOCKET_PATH)

sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
sock.bind(SOCKET_PATH)
os.chmod(SOCKET_PATH, 0o600)
sock.listen(5)

# Docker socket proxy
_docker_sock_path = _detect_docker_socket()
if _docker_sock_path:
    if os.path.exists(DOCKER_PROXY_PATH):
        os.unlink(DOCKER_PROXY_PATH)
    _docker_proxy_sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
    _docker_proxy_sock.bind(DOCKER_PROXY_PATH)
    os.chmod(DOCKER_PROXY_PATH, 0o600)
    _docker_proxy_sock.listen(5)
    threading.Thread(
        target=_docker_proxy_listener,
        args=(_docker_proxy_sock, _docker_sock_path),
        daemon=True,
    ).start()

print("credential server listening")
if auto_approve_git_reads:
    print("git/gh reads auto-approved (disable: --no-auto-git-reads)")
if auto_approve_git_push:
    print("git push/gh pr writes auto-approved, except main/master (disable: --no-auto-git-push)")
if auto_approve_aws_profiles:
    print(f"auto-approved AWS profiles: {', '.join(sorted(auto_approve_aws_profiles))}")
if auto_approve_git_push and auto_approve_main_push_repos:
    print(f"main pushes auto-approved for: {', '.join(sorted(auto_approve_main_push_repos))}")
if _docker_sock_path:
    print(f"docker proxy active ({_docker_sock_path})")
    if auto_approve_docker_reads:
        print("docker reads auto-approved (disable: --no-auto-docker-reads)")
print("Press Ctrl+C to stop\n")

# Set terminal window title for AXRaise targeting
sys.stdout.write(f"\033]0;{WINDOW_TITLE}\007")
sys.stdout.flush()

try:
    while True:
        conn, _ = sock.accept()
        try:
            # Read request
            data = conn.recv(4096).decode().strip()
            if not data:
                conn.sendall(json.dumps({"ok": False}).encode() + b"\n")
                continue
            request = json.loads(data)

            # Get process info
            peer_pid = struct.unpack("i", conn.getsockopt(SOL_LOCAL, LOCAL_PEERPID, 4))[0]
            chain = get_process_chain(peer_pid)
            cmd, cmd_argv, cmd_pid, safehouse_pid, safehouse_cwd, actual_cmd = parse_chain(chain)
            cmd = cmd or request.get("type", "unknown")

            # Remote requests arrive over an `ssh -R` socket forward: the peer
            # is the local ssh client, so process introspection finds no
            # safehouse ancestor and the wrong command. Trust the wrapper's
            # self-reported context instead, scoped to a synthetic
            # "remote:<host>" identity. See AGENTS.md "remote mode".
            if request.get("origin") == "remote":
                host = _sanitize_host(str(request.get("host", "unknown")))
                safehouse_pid = f"remote:{host}"
                remote_cmd = str(request.get("cmd", "")).strip()
                cmd = remote_cmd or request.get("type", "unknown")
                try:
                    cmd_argv = shlex.split(remote_cmd) if remote_cmd else []
                except ValueError:
                    cmd_argv = remote_cmd.split()
                safehouse_cwd = request.get("cwd") or None
                cmd_pid = None
                actual_cmd = None
                chain = ()

            # Handle
            response = handle_request(conn, request, safehouse_pid, safehouse_cwd, cmd, cmd_argv, cmd_pid, chain, actual_cmd)
            conn.sendall(json.dumps(response).encode() + b"\n")

        except (json.JSONDecodeError, EOFError, OSError):
            try:
                conn.sendall(json.dumps({"ok": False}).encode() + b"\n")
            except OSError:
                pass
        except Exception as e:
            import traceback
            print(f"\n  {RED}error: {e}{RESET}")
            traceback.print_exc()
            try:
                conn.sendall(json.dumps({"ok": False}).encode() + b"\n")
            except OSError:
                pass
        finally:
            conn.close()
finally:
    cleanup()