imp-proxy

#!/usr/bin/env python3
"""imp — lend a remote machine your Claude Max subscription without ever
giving it the credential.

The token stays on this machine. A local proxy holds it and injects it into
every request; the far side gets only a per-session capability string and a
URL that resolves to this machine's loopback. Stop the proxy and the far
side's access dies instantly — no watchdog, no teardown, nothing to scrub,
and no need to kill running `claude` processes to revoke.

Any host you can ssh to is the default, and ssh's own -R carries the
traffic:

    this machine                         far side (ssh HOST)
    ------------------------------------------------------------
    injecting proxy  <------ ssh -R ------> claude
      (real token)                          ANTHROPIC_BASE_URL=
                                            http://127.0.0.1:22376

A Fly Sprite (-s) is supported too. `sprite proxy --ssh` supports neither -R
nor -A, so for a sprite we reimplement remote forwarding in userspace over
the one ssh stdio channel, which is 8-bit clean in both directions
(verified): a relay on the sprite, framed and multiplexed back to us.

Runs on macOS and Linux. Requires only python3 and ssh here. The far side
needs nothing for --no-settings, python3 to have its settings.json written,
and for a sprite, python3 there and the `sprite` CLI here.
"""

import argparse
import base64
import http.client
import json
import os
import platform
import posixpath
import queue
import re
import shlex
import signal
import socket
import socketserver
import struct
import subprocess
import sys
import tempfile
import threading
import time
import urllib.parse

UPSTREAM = "https://api.anthropic.com"

# Claude Code drops oauth-2025-04-20 from its beta list when it has no local
# OAuth credentials -- which is exactly the state the far side is in. Without it
# the Max entitlement is not honoured, so we put it back on the way out.
REQUIRED_BETAS = ("claude-code-20250219", "oauth-2025-04-20")

# A hostile far side gets to choose the request path, so the proxy is otherwise an
# open relay to your account for the length of the session. Only inference
# endpoints are worth reaching through here; everything else is refused before
# the credential is ever attached. Extend with --allow, escape with
# --allow-any-path.
DEFAULT_ALLOWED = (
    "/v1/messages",
    "/v1/messages/count_tokens",
    "/v1/models",
    "/v1/models/*",
)

# The port the far side listens on. Nothing but `claude` and imp ever dials
# it -- it is on that machine's loopback, reached only through our own ssh --
# so the only property that matters is that nothing else there wants it.
# 8080, which this was, is the first port a dev server takes, and a taken
# port is a relay that cannot bind or a -R that silently has no proxy behind
# it. This one is a random pick from the range above the dev-server crowd and
# below Linux's ephemeral ports, so neither a service nor an outbound socket
# is likely to have it first. Nothing depends on the number; -p changes it.
DEFAULT_REMOTE_PORT = 22376

HDR = struct.Struct("!IBI")
T_OPEN, T_DATA, T_CLOSE, T_PING = 1, 2, 3, 4

PING_EVERY = 10.0
LINK_TIMEOUT = 60.0
RELAY_IDLE_TIMEOUT = 30.0

# A hostile far side gets to declare how much we should allocate for it. None of
# these are tuning knobs -- they exist so that a number chosen on the far end
# cannot exhaust memory, threads or descriptors on this machine.
MAX_BODY = 64 * 1024 * 1024     # generous next to a large image request
MAX_FRAME = 1024 * 1024         # the relay never sends more than 64KB
MAX_STREAMS = 256               # concurrent proxied connections
# Per-stream write buffer before a consumer is written off as stuck. A claude
# that pauses to render a long response is slow, not stuck -- be generous.
MAX_STREAM_BUFFER = 8 * 1024 * 1024
SETTINGS_TIMEOUT = 30.0         # ssh round trip for settings.json
STATE_TTL = 24 * 3600.0         # how long a session is worth reattaching to

# A long inference call streams for minutes, so the socket timeout has to be
# generous; it exists to stop a wedged connection pinning a thread forever.
UPSTREAM_TIMEOUT = 600.0
# How long a kept connection may sit idle before we open a new one instead.
# A connection the far end closed while we were not looking costs a retry to
# discover; below the usual edge idle timeout, we mostly do not find out.
UPSTREAM_MAX_IDLE = 55.0


def log(*a):
    sys.stderr.write("[imp] " + " ".join(str(x) for x in a) + "\n")
    sys.stderr.flush()


def die(*lines):
    for line in lines:
        log(line)
    raise SystemExit(2)


# --------------------------------------------------------------------------
# credential lookup (macOS + Linux)
# --------------------------------------------------------------------------

def _run(cmd):
    try:
        return subprocess.run(cmd, capture_output=True, text=True)
    except FileNotFoundError:
        return None


def _extract_token(blob):
    """A credential blob is either a bare token or JSON holding accessToken."""
    blob = blob.strip()
    if not blob:
        return None
    try:
        data = json.loads(blob)
    except ValueError:
        return blob

    found = []

    def walk(o):
        if isinstance(o, dict):
            for k, v in o.items():
                if k == "accessToken" and isinstance(v, str):
                    found.append(v)
                else:
                    walk(v)
        elif isinstance(o, list):
            for v in o:
                walk(v)

    walk(data)
    return found[0] if found else None


def load_credential():
    """Return (token, human-readable source). Never logged, never written out.

    Preferred: the dedicated setup-token minted by imp-auth, which is
    long-lived and inference-only. Falls back to this machine's live Claude
    login, which works but carries a short-lived access token -- see --help.
    """
    darwin = platform.system() == "Darwin"
    user = os.environ.get("USER") or os.environ.get("LOGNAME") or ""

    if darwin:
        candidates = [
            (["security", "find-generic-password", "-s", "imp-oauth",
              "-a", user, "-w"], "macOS keychain (imp-oauth, setup-token)"),
            (["security", "find-generic-password", "-s", "Claude Code-credentials",
              "-a", user, "-w"], "macOS keychain (live Claude login)"),
        ]
        for cmd, source in candidates:
            r = _run(cmd)
            if r is not None and r.returncode == 0:
                tok = _extract_token(r.stdout)
                if tok:
                    return tok, source
    else:
        path = os.path.expanduser("~/.claude/.credentials.json")
        if os.path.exists(path):
            with open(path) as f:
                tok = _extract_token(f.read())
            if tok:
                return tok, path
        for service in ("imp-oauth", "Claude Code-credentials"):
            r = _run(["secret-tool", "lookup", "service", service])
            if r is not None and r.returncode == 0:
                tok = _extract_token(r.stdout)
                if tok:
                    return tok, "secret-tool (%s)" % service

    env = os.environ.get("CLAUDE_CODE_OAUTH_TOKEN")
    if env:
        return env.strip(), "CLAUDE_CODE_OAUTH_TOKEN in this environment"

    raise SystemExit(
        "no Claude credential found.\n"
        "  Mint a dedicated one:  claude setup-token   (then store it as\n"
        "  'imp-oauth' in your keychain), or just log in with `claude`."
    )


# --------------------------------------------------------------------------
# the injecting proxy — the only thing that ever sees the real token
# --------------------------------------------------------------------------

# Decoding is repeated until it stops changing, so %252e%252e cannot hide a
# traversal behind one round of unescaping. A handful of rounds is far more
# than any real target needs; the cap only stops a pathological input.
MAX_DECODE_ROUNDS = 5


def _fully_decoded(path):
    """Percent-decode until stable, or None if it will not settle."""
    for _ in range(MAX_DECODE_ROUNDS):
        nxt = urllib.parse.unquote(path)
        if nxt == path:
            return path
        path = nxt
    return None


def normalize_path(raw):
    """Canonical path to authorize *and* to forward, or None if unsafe.

    The value returned here is what actually goes upstream. Authorizing one
    spelling and forwarding another is how an allowlist gets walked past:
    %2e%2e survives posixpath.normpath untouched, so a check on the raw target
    sees nothing wrong while whatever decodes it downstream sees `..`.

    So: decode first, refuse anything whose decoded form is not already
    normal, and re-encode that decoded form for the wire. What is checked and
    what is sent are then the same string by construction.
    """
    if not raw.startswith("/"):
        return None
    path = raw.split("#", 1)[0].split("?", 1)[0]

    decoded = _fully_decoded(path)
    if decoded is None or not decoded.startswith("/"):
        return None
    # Backslashes and control bytes have no place in a target and are a
    # standard way to confuse whatever parses it next.
    if "\\" in decoded or any(c < " " or c == "\x7f" for c in decoded):
        return None

    norm = posixpath.normpath(decoded)
    if norm != decoded and norm != decoded.rstrip("/"):
        return None
    return urllib.parse.quote(norm, safe="/")


def _glob_to_re(pat):
    """`*` matches within one path segment, `**` spans segments.

    fnmatch's `*` crosses `/`, which makes /v1/models/* mean "anything under
    /v1/models" rather than "one model id" -- a much wider grant than it
    looks. Path globs should behave the way shell globs do.
    """
    out, i = [], 0
    while i < len(pat):
        c = pat[i]
        if c == "*":
            if pat.startswith("**", i):
                out.append(".*")
                i += 2
            else:
                out.append("[^/]*")
                i += 1
        elif c == "?":
            out.append("[^/]")
            i += 1
        else:
            out.append(re.escape(c))
            i += 1
    return re.compile("".join(out) + r"\Z")


_GLOB_CACHE = {}


def path_allowed(path, patterns):
    if not patterns:            # --allow-any-path
        return True
    for pat in patterns:
        rx = _GLOB_CACHE.get(pat)
        if rx is None:
            rx = _GLOB_CACHE[pat] = _glob_to_re(pat)
        if rx.match(path):
            return True
    return False


HOP = {"host", "accept-encoding", "content-length", "connection", "keep-alive",
       "proxy-authenticate", "proxy-authorization", "te", "trailers",
       "transfer-encoding", "upgrade"}


class Upstream(object):
    """One kept-alive connection to the API per handler thread.

    http.client rather than urllib.request, for two reasons.

    Redirects: urllib's default handler copies Authorization into the
    redirected request without checking the origin, so a 3xx would hand the
    real token to whatever host the Location names. It took a handler
    subclass to say no. http.client does not follow redirects at all -- the
    3xx comes back like any other response and goes to the far side verbatim,
    for it to decide about without our credential.

    Reuse: urllib opens a fresh connection per request, so every message paid
    a full TLS handshake to api.anthropic.com. That handshake was most of
    what imp added on top of talking to the API directly. Threads here are
    one per client connection, so a thread-local connection lines the two
    lifetimes up: one `claude` conversation, one connection at each end, and
    nothing shared between threads to lock.
    """

    def __init__(self):
        self._local = threading.local()

    def _target(self):
        u = urllib.parse.urlsplit(UPSTREAM)
        return u.scheme, u.hostname, u.port

    def _open(self):
        scheme, host, port = self._target()
        cls = (http.client.HTTPSConnection if scheme == "https"
               else http.client.HTTPConnection)
        conn = cls(host, port, timeout=UPSTREAM_TIMEOUT)
        self._local.conn = conn
        self._local.key = (scheme, host, port)
        self._local.used = time.time()
        return conn

    def discard(self):
        """Throw the kept connection away. Safe to call at any point."""
        conn = getattr(self._local, "conn", None)
        self._local.conn = None
        if conn is not None:
            try:
                conn.close()
            except Exception:
                pass

    def send(self, method, target, body, headers):
        """Return (response, error); exactly one of the two is None.

        A kept connection the far end has already closed is indistinguishable
        from a live one until we write to it. That failure costs one retry on
        a fresh connection -- safe to repeat regardless of method, because a
        request that died on the way out was never seen upstream.
        """
        conn = getattr(self._local, "conn", None)
        if conn is not None and (
                getattr(self._local, "key", None) != self._target()
                or time.time() - getattr(self._local, "used", 0.0) > UPSTREAM_MAX_IDLE):
            self.discard()
            conn = None
        reused = conn is not None
        if conn is None:
            conn = self._open()
        while True:
            try:
                conn.request(method, target, body=body, headers=headers)
                resp = conn.getresponse()
                self._local.used = time.time()
                return resp, None
            except Exception as e:
                self.discard()
                if not reused:
                    return None, e
                reused = False
                conn = self._open()

    def release(self, resp, complete):
        """Hand the connection back, or throw it away.

        Only a response read to the end leaves the connection at a message
        boundary. A client that hung up mid-stream, or an upstream that said
        `Connection: close`, leaves bytes we would otherwise read as the head
        of the next response.
        """
        if resp is None:
            return
        will_close = getattr(resp, "will_close", True)
        try:
            resp.close()
        except Exception:
            pass
        if will_close or not complete:
            self.discard()


UPSTREAM_CONN = Upstream()


def _drop(conn):
    """Close a proxied socket that another thread may be blocked reading.

    Same reason as the relay's drop(): close() alone does not reliably wake a
    thread parked in recv(), nor guarantee the peer sees EOF.
    """
    try:
        conn.shutdown(socket.SHUT_RDWR)
    except OSError:
        pass
    try:
        conn.close()
    except OSError:
        pass


class Credential(object):
    """Holds the live token. Re-read on a 401 so a long session survives the
    access token rolling underneath it -- Claude Code on this machine refreshes
    the keychain entry, we just notice."""

    def __init__(self):
        self.token, self.source = load_credential()
        self.lock = threading.Lock()

    def refresh(self):
        with self.lock:
            old = self.token
            try:
                self.token, self.source = load_credential()
            except (Exception, SystemExit) as e:
                # SystemExit is "no credential found", but the keychain and
                # ~/.claude/.credentials.json can also be unreadable or
                # half-written under us. Losing a refresh is survivable;
                # letting it out of this method is not -- it would surface in
                # the handler thread as a dropped connection, where passing
                # the upstream 401 back was the answer.
                log("credential refresh failed: %s" % e)
                return False
            return self.token != old


def make_proxy_handler(cred, capability, verbose, allowed):
    import http.server

    class Handler(http.server.BaseHTTPRequestHandler):
        protocol_version = "HTTP/1.1"

        def log_message(self, fmt, *args):
            pass

        def finish(self):
            # A thread here is one client connection, so when it ends the
            # kept upstream connection has no next request to serve. Closed
            # on the way out rather than left to the garbage collector --
            # which is where it otherwise surfaced, as a ResourceWarning and
            # a socket held open for as long as the collector felt like it.
            try:
                http.server.BaseHTTPRequestHandler.finish(self)
            finally:
                UPSTREAM_CONN.discard()

        # The target is the far side's to choose, so it reaches our terminal
        # only as printable ASCII, clipped: a request line can carry 64 KB of
        # escape sequences, and a rejection is exactly when it is unvetted.
        def _safe_target(self):
            raw = getattr(self, "path", None) or "-"
            clean = "".join(c if " " <= c <= "~" else "?" for c in raw)
            return clean if len(clean) <= 120 else clean[:120] + "..."

        def _reject(self, code, why):
            # Logged, not returned: the body says what was wrong, and the
            # far side already knows what it asked for.
            log("rejected: %s (%s %s)" % (why, self.command, self._safe_target()))
            body = json.dumps({"type": "error", "error": {
                "type": "authentication_error", "message": why}}).encode()
            self.send_response(code)
            self.send_header("Content-Type", "application/json")
            self.send_header("Content-Length", str(len(body)))
            # We may be refusing before reading the request body, so this
            # connection cannot be reused -- whatever is still in flight would
            # be parsed as the next request line.
            self.send_header("Connection", "close")
            self.close_connection = True
            self.end_headers()
            self.wfile.write(body)

        def _proxy(self):
            path = normalize_path(self.path)
            if path is None:
                self._reject(400, "malformed request path")
                return
            if not path_allowed(path, allowed):
                self._reject(403, "path is not on the allowlist")
                return
            # Forward the target we authorized, not the one we were handed.
            target = path
            q = self.path.split("#", 1)[0].split("?", 1)
            if len(q) == 2:
                target += "?" + q[1]

            # Folded to lower case on the way in. Header names are
            # case-insensitive but a dict is not, and every lookup below is a
            # dict lookup: with the far side's spelling kept, an
            # `Anthropic-Beta` slid past the `anthropic-beta` merge and its
            # betas were dropped on the floor. One spelling from here on, so
            # a repeated header collapses here rather than somewhere further
            # down -- and so popping two keys strips every credential header
            # the far side sent, whatever case it chose for them.
            fwd = {}
            for k, v in self.headers.items():
                k = k.lower()
                if k not in HOP:
                    fwd[k] = v
            fwd["accept-encoding"] = "identity"

            presented = ""
            for k in ("authorization", "x-api-key"):
                v = fwd.pop(k, None)
                if v is not None:
                    presented = presented or v
            presented = presented[7:] if presented.lower().startswith("bearer ") else presented

            # Before the body is read, not after: Content-Length is chosen by
            # the far side, and reading it first hands an unauthenticated caller
            # a blocking allocation of whatever size it likes.
            if not presented or presented.strip() != capability:
                self._reject(403, "bad or missing session capability")
                return

            n = self.headers.get("Content-Length")
            try:
                length = int(n) if n else 0
            except ValueError:
                self._reject(400, "malformed Content-Length")
                return
            if length < 0:
                # rfile.read(-1) would read to EOF -- unbounded, by way of a
                # value that merely looks like a number.
                self._reject(400, "negative Content-Length")
                return
            if length > MAX_BODY:
                self._reject(413, "request body of %d bytes exceeds the %d byte limit"
                             % (length, MAX_BODY))
                return
            body = self.rfile.read(length) if length else b""

            fwd["authorization"] = "Bearer " + cred.token
            betas = [b.strip() for b in (fwd.get("anthropic-beta") or "").split(",")]
            betas = [b for b in betas if b]
            for b in REQUIRED_BETAS:
                if b not in betas:
                    betas.insert(0, b)
            if betas:
                fwd["anthropic-beta"] = ",".join(betas)

            if verbose:
                log("%s %s (%d bytes up)" % (self.command, target, len(body)))

            resp, err = UPSTREAM_CONN.send(self.command, target, body, fwd)
            if err is None and resp.status == 401 and cred.refresh():
                log("upstream 401 -- reloaded credential, retrying")
                # Read the 401 body out before reusing the connection; an
                # unread body is the next response as far as the socket knows.
                try:
                    resp.read()
                    drained = True
                except Exception:
                    drained = False
                UPSTREAM_CONN.release(resp, drained)
                fwd["authorization"] = "Bearer " + cred.token
                resp, err = UPSTREAM_CONN.send(self.command, target, body, fwd)
            if err is not None:
                log("upstream error: %r" % (err,))
                self._reject(502, "upstream error")
                return

            self.send_response(resp.status)
            for k, v in resp.getheaders():
                if k.lower() not in HOP:
                    self.send_header(k, v)
            self.send_header("Transfer-Encoding", "chunked")
            self.end_headers()

            # Streamed verbatim so SSE arrives token-by-token, not buffered.
            # `complete` is set the moment upstream is drained, before the
            # terminating chunk goes out: a client that hangs up on that last
            # write has cost us nothing, and the connection is still good.
            complete = False
            try:
                while True:
                    chunk = resp.read(4096)
                    if not chunk:
                        break
                    self.wfile.write(b"%x\r\n" % len(chunk) + chunk + b"\r\n")
                    self.wfile.flush()
                complete = True
                self.wfile.write(b"0\r\n\r\n")
                self.wfile.flush()
            finally:
                UPSTREAM_CONN.release(resp, complete)

        do_GET = do_POST = do_PUT = do_DELETE = _proxy

    return Handler


class ProxyServer(socketserver.ThreadingTCPServer):
    daemon_threads = True
    allow_reuse_address = True

    def handle_error(self, request, client_address):
        """A client hanging up mid-request is routine -- `claude` does it on
        every Ctrl-C. socketserver's default would dump a traceback for it.
        Anything genuinely unexpected gets one line, never headers or bodies."""
        exc = sys.exc_info()[1]
        if isinstance(exc, (ConnectionResetError, BrokenPipeError,
                            ConnectionAbortedError, socket.timeout)):
            return
        log("connection error: %r" % (exc,))


# --------------------------------------------------------------------------
# the relay that runs on the sprite (shipped as base64 in argv, never a file)
# --------------------------------------------------------------------------

RELAY_SRC = r'''
import os, queue, socket, struct, sys, threading, time

PORT = int(sys.argv[1])
IDLE = float(sys.argv[2])
# Per-stream buffer before a consumer is written off as stuck. Generous: a
# claude that pauses to render a long response is slow, not stuck.
MAX_BUF = 8 * 1024 * 1024
HDR = struct.Struct("!IBI")
T_OPEN, T_DATA, T_CLOSE, T_PING = 1, 2, 3, 4

out = sys.stdout.buffer
inp = sys.stdin.buffer
streams = {}
slock = threading.Lock()
last = [time.time()]

# A bounded queue drained by one writer thread. Producers are per-stream pump
# threads, so blocking one when the channel backs up is correct backpressure
# for that stream alone -- unlike holding a lock across the write, which
# stalled every stream at once.
outq = queue.Queue(maxsize=256)


def drop(conn):
    # shutdown() before close(): another thread is blocked in recv() on this
    # socket, and close() alone neither reliably wakes it nor guarantees the
    # peer sees EOF. shutdown() does both.
    try:
        conn.shutdown(socket.SHUT_RDWR)
    except OSError:
        pass
    try:
        conn.close()
    except OSError:
        pass


def send(sid, t, payload=b""):
    outq.put(HDR.pack(sid, t, len(payload)) + payload)


def chan_writer():
    while True:
        buf = outq.get()
        try:
            out.write(buf)
            out.flush()
        except (OSError, ValueError):
            os._exit(0)


class Stream(object):
    """A socket plus its own writer thread.

    The frame reader must never block: one consumer that stops draining would
    otherwise stall every other stream, and -- because the reader is also what
    refreshes the idle timestamp -- trip the watchdog into killing the whole
    session 30s later.
    """

    def __init__(self, sid, conn):
        self.sid = sid
        self.conn = conn
        self.q = queue.Queue()
        self.buffered = 0
        self.lock = threading.Lock()
        self.closed = False
        threading.Thread(target=self.writer, daemon=True).start()

    def feed(self, data):
        """Hand off without blocking. True if accepted."""
        with self.lock:
            if self.closed:
                return False
            if self.buffered + len(data) > MAX_BUF:
                return False        # genuinely stuck; caller drops the stream
            self.buffered += len(data)
        self.q.put(data)
        return True

    def writer(self):
        while True:
            data = self.q.get()
            if data is None:
                break
            try:
                self.conn.sendall(data)
            except OSError:
                break
            with self.lock:
                self.buffered -= len(data)
        self.close()

    def close(self):
        with self.lock:
            if self.closed:
                return
            self.closed = True
        self.q.put(None)
        drop(self.conn)


def pump(sid, st):
    try:
        while True:
            b = st.conn.recv(65536)
            if not b:
                break
            send(sid, T_DATA, b)     # may block on outq; that is this stream's
    except OSError:                  # backpressure, not everyone else's
        pass
    finally:
        with slock:
            streams.pop(sid, None)
        send(sid, T_CLOSE)
        st.close()


def acceptor():
    srv = socket.socket()
    srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    try:
        srv.bind(("127.0.0.1", PORT))
    except OSError as e:
        sys.stderr.write("relay: cannot bind 127.0.0.1:%d: %s\n" % (PORT, e))
        sys.stderr.flush()
        os._exit(1)
    srv.listen(64)
    sys.stderr.write("relay listening on 127.0.0.1:%d\n" % PORT)
    sys.stderr.flush()
    # Tell the other end the link is genuinely up. Nothing is "ready" before
    # this frame arrives.
    send(0, T_PING)
    n = 0
    while True:
        conn, _ = srv.accept()
        n += 1
        st = Stream(n, conn)
        with slock:
            streams[n] = st
        send(n, T_OPEN)
        threading.Thread(target=pump, args=(n, st), daemon=True).start()


def watchdog():
    # The sprite keeps remote commands alive after the link drops, so nothing
    # tears us down for free. Silence means the operator is gone: exit, which
    # closes every proxied connection and frees the port.
    while True:
        time.sleep(2)
        if time.time() - last[0] > IDLE:
            sys.stderr.write("relay: %.0fs silent, exiting\n" % IDLE)
            os._exit(0)


def readexact(n):
    buf = b""
    while len(buf) < n:
        c = inp.read(n - len(buf))
        if not c:
            return None
        buf += c
    return buf


threading.Thread(target=chan_writer, daemon=True).start()
threading.Thread(target=acceptor, daemon=True).start()
threading.Thread(target=watchdog, daemon=True).start()

while True:
    h = readexact(HDR.size)
    if h is None:
        break
    sid, t, ln = HDR.unpack(h)
    payload = readexact(ln) if ln else b""
    if payload is None:
        break
    last[0] = time.time()
    if t == T_PING:
        continue
    with slock:
        st = streams.get(sid)
    if st is None:
        continue
    if t == T_DATA:
        if not st.feed(payload):
            # Buffered past MAX_BUF: this consumer is not coming back. Drop
            # it alone rather than letting it hold up the demultiplexer.
            with slock:
                streams.pop(sid, None)
            send(sid, T_CLOSE)
            st.close()
    elif t == T_CLOSE:
        with slock:
            streams.pop(sid, None)
        st.close()
os._exit(0)
'''


# --------------------------------------------------------------------------
# this side of the mux
# --------------------------------------------------------------------------

class Stream(object):
    """A proxied socket plus its own writer thread.

    The frame reader must never block on one of these. With a single `claude`
    on the sprite a blocking write is just backpressure, but with several,
    one that stops draining would stall every other stream -- and since the
    reader is also what refreshes the relay's idle timestamp, the watchdog
    would then tear the whole session down 30s later.
    """

    def __init__(self, conn):
        self.conn = conn
        self.q = queue.Queue()
        self.buffered = 0
        self.lock = threading.Lock()
        self.closed = False
        threading.Thread(target=self._writer, daemon=True).start()

    def feed(self, data):
        """Hand off without blocking. False means the peer is not draining."""
        with self.lock:
            if self.closed:
                return False
            if self.buffered + len(data) > MAX_STREAM_BUFFER:
                return False
            self.buffered += len(data)
        self.q.put(data)
        return True

    def _writer(self):
        while True:
            data = self.q.get()
            if data is None:
                break
            try:
                self.conn.sendall(data)
            except OSError:
                break
            with self.lock:
                self.buffered -= len(data)
        self.close()

    def close(self):
        with self.lock:
            if self.closed:
                return
            self.closed = True
        self.q.put(None)
        _drop(self.conn)


class Mux(object):
    """Demultiplexes the sprite's framed stdout into local proxy connections."""

    def __init__(self, proc, proxy_port, verbose):
        self.proc = proc
        self.proxy_port = proxy_port
        self.verbose = verbose
        self.socks = {}
        self.lock = threading.Lock()
        self.alive = True
        self.ready = threading.Event()
        # Bounded, drained by one writer thread. Producers are per-stream
        # pump threads, so blocking one when the pipe backs up is that
        # stream's backpressure -- not a lock every other stream waits on.
        self.outq = queue.Queue(maxsize=256)
        threading.Thread(target=self._chan_writer, daemon=True).start()

    def send(self, sid, t, payload=b""):
        try:
            self.outq.put(HDR.pack(sid, t, len(payload)) + payload)
        except Exception:
            self.alive = False

    def _chan_writer(self):
        while True:
            buf = self.outq.get()
            if buf is None:
                break
            try:
                self.proc.stdin.write(buf)
                self.proc.stdin.flush()
            except (OSError, ValueError):
                self.alive = False
                break

    def _pump(self, sid, st):
        try:
            while True:
                b = st.conn.recv(65536)
                if not b:
                    break
                self.send(sid, T_DATA, b)
        except OSError:
            pass
        finally:
            with self.lock:
                self.socks.pop(sid, None)
            self.send(sid, T_CLOSE)
            st.close()

    def _open(self, sid):
        with self.lock:
            if len(self.socks) >= MAX_STREAMS:
                log("refusing stream %d: %d already open" % (sid, MAX_STREAMS))
                self.send(sid, T_CLOSE)
                return
            stale = self.socks.pop(sid, None)
        if stale is not None:
            # A repeated SID used to overwrite the entry and leak the socket.
            stale.close()
        try:
            conn = socket.create_connection(("127.0.0.1", self.proxy_port), timeout=10)
            conn.settimeout(None)
        except OSError as e:
            log("could not reach local proxy: %r" % (e,))
            self.send(sid, T_CLOSE)
            return
        st = Stream(conn)
        with self.lock:
            self.socks[sid] = st
        threading.Thread(target=self._pump, args=(sid, st), daemon=True).start()

    def pinger(self):
        while self.alive:
            time.sleep(PING_EVERY)
            self.send(0, T_PING)

    def _readexact(self, n):
        # Chunks into a list rather than repeatedly concatenating: `buf += c`
        # over a large frame is quadratic.
        parts, got = [], 0
        while got < n:
            c = self.proc.stdout.read(n - got)
            if not c:
                return None
            parts.append(c)
            got += len(c)
        return b"".join(parts)

    def run(self):
        threading.Thread(target=self.pinger, daemon=True).start()
        while True:
            h = self._readexact(HDR.size)
            if h is None:
                break
            sid, t, ln = HDR.unpack(h)
            if ln > MAX_FRAME:
                # The length is a uint32 chosen by the far end. Allocating on
                # its say-so is how the sprite OOMs this machine; the relay
                # never legitimately sends more than 64KB.
                log("frame of %d bytes exceeds the %d byte limit -- "
                    "closing the link" % (ln, MAX_FRAME))
                break
            payload = self._readexact(ln) if ln else b""
            if payload is None:
                break
            self.ready.set()
            if t == T_OPEN:
                self._open(sid)
            elif t == T_DATA:
                with self.lock:
                    st = self.socks.get(sid)
                if st is not None and not st.feed(payload):
                    log("stream %d buffered past %d bytes without draining -- "
                        "dropping it" % (sid, MAX_STREAM_BUFFER))
                    with self.lock:
                        self.socks.pop(sid, None)
                    self.send(sid, T_CLOSE)
                    st.close()
            elif t == T_CLOSE:
                with self.lock:
                    st = self.socks.pop(sid, None)
                if st is not None:
                    st.close()
        self.alive = False
        self.ready.set()


# --------------------------------------------------------------------------
# far-side plumbing
# --------------------------------------------------------------------------

def resolve_target(host, sprite):
    """Which machine to fund. A host is the default and needs no flag; a
    sprite is asked for with -s, or selected for this directory by `sprite
    use`, which is the sprite CLI's own default and so is honoured here.

    Settled before anything else happens rather than discovered as an ssh
    failure after the credential is already loaded."""
    if host and sprite:
        die("-s and a host name two different machines; pick one.")
    if host:
        return SSHTarget(host)
    if sprite:
        return SpriteTarget(sprite)
    if os.path.exists(".sprite"):
        return SpriteTarget("")  # the sprite CLI reads it, as it would anyway
    die("no machine named.",
        "  imp HOST         any ssh destination, ssh_config aliases included",
        "  imp -s SPRITE    a Fly Sprite")


def wait_for_link(mux, proc, timeout):
    """The relay pings as soon as it is listening; that frame is the only
    honest signal that the sprite side exists."""
    deadline = time.time() + timeout
    while time.time() < deadline:
        if mux.ready.wait(0.25):
            return mux.alive
        if proc.poll() is not None:
            return False
    return False


def wait_for_marker(proc, timeout):
    """Same question for a native -R forward, same answer: wait to be told.

    ssh returning success only means it connected. The marker is printed by
    the remote shell, which does not run until the forward has been set up --
    and with ExitOnForwardFailure, not at all if it could not be.
    """
    ready = threading.Event()

    def reader():
        for line in iter(proc.stdout.readline, b""):
            if line.strip() == LINK_MARKER.encode():
                ready.set()
                return

    threading.Thread(target=reader, daemon=True).start()
    deadline = time.time() + timeout
    while time.time() < deadline:
        if ready.wait(0.25):
            return True
        if proc.poll() is not None:
            return False
    return False


# A shared ControlMaster would outlive revocation -- the whole point is that
# stopping this process ends the far side's access, and a multiplexed socket
# someone else is holding open does not end.
SSH_COMMON = [
    "-o", "ControlMaster=no",
    "-o", "ControlPath=none",
    "-o", "ServerAliveInterval=10",
    "-o", "ServerAliveCountMax=3",
]

# Printed by the far side once its end of the forward exists, which is the
# only honest signal that it does. POSIX sh, no python: with a real -R there
# is nothing else the remote has to be able to run.
LINK_MARKER = "IMP_LINK_UP"
LINK_CMD = "echo %s; exec cat >/dev/null" % LINK_MARKER


class SpriteTarget(object):
    """A Fly Sprite, reached through the sprite CLI.

    `sprite proxy --ssh` supports neither -R nor -A, so the reverse forward
    has to be the userspace relay in RELAY_SRC. Host key checking is off
    because the ProxyCommand builds a fresh transport per connection -- there
    is no stable key to pin, and the CLI has already authenticated it.
    """

    needs_relay = True
    kind = "sprite"

    def __init__(self, name):
        self.name = name

    def label(self):
        return self.name or "(from .sprite)"

    def short_label(self):
        """A name for a tmux window. `.sprite` is read here and only here --
        for a label, where being wrong is cosmetic."""
        if self.name:
            return self.name
        try:
            with open(".sprite") as f:
                return f.readline().strip() or "sprite"
        except OSError:
            return "sprite"

    def console_argv(self):
        argv = ["sprite", "console"]
        if self.name:
            argv += ["-s", self.name]
        return argv

    def exec_argv(self, remote_cmd):
        proxy_cmd = "sprite proxy --ssh"
        if self.name:
            proxy_cmd += " -s " + self.name
        return [
            "ssh",
            "-o", "ProxyCommand=" + proxy_cmd,
            "-o", "StrictHostKeyChecking=no",
            "-o", "UserKnownHostsFile=/dev/null",
            "-o", "LogLevel=ERROR",
        ] + SSH_COMMON + ["sprite", remote_cmd]

    def link_argv(self, remote_port, local_port):
        return self.exec_argv(remote_python(
            RELAY_SRC, (str(remote_port), str(RELAY_IDLE_TIMEOUT))))


class SSHTarget(object):
    """Any host that speaks ordinary ssh, where -R does natively what the
    relay was written to work around.

    Host key checking is left alone here. The sprite's is a throwaway
    transport; a machine you own is not, and turning off the check for it
    would be trading a real protection for nothing.
    """

    needs_relay = False
    kind = "ssh"

    def __init__(self, dest):
        self.dest = dest

    def label(self):
        return self.dest

    def short_label(self):
        return self.dest

    def console_argv(self):
        # -t because this is a person's shell: without a tty you get no
        # prompt, no job control and no `claude`.
        return ["ssh", "-t", self.dest]

    def exec_argv(self, remote_cmd):
        return ["ssh"] + SSH_COMMON + [self.dest, remote_cmd]

    def link_argv(self, remote_port, local_port):
        # 127.0.0.1 spelled out on both ends. `-R 22376:...` binds 0.0.0.0 on
        # a host configured with GatewayPorts yes, which would put the
        # capability endpoint on the network -- the one way this transport
        # can be less careful than the relay, which only ever binds loopback.
        fwd = "127.0.0.1:%d:127.0.0.1:%d" % (remote_port, local_port)
        # Without ExitOnForwardFailure a taken remote port is a warning and a
        # session that silently has no proxy behind it. The relay reports the
        # same condition by failing to bind and never pinging.
        return ["ssh", "-R", fwd,
                "-o", "ExitOnForwardFailure=yes"] + SSH_COMMON + [
                self.dest, LINK_CMD]


# Compare-and-swap on the capability, which is already unique per session.
# Without it, a second imp against the same host (only possible with -p,
# since the relay's bind on the default port already refuses a collision)
# silently repoints settings.json at itself, and whichever session exits
# first strips the env block out from under the other.
#
# A capability that is not ours is only worth respecting if the session
# behind it still exists, and this script runs on the far side -- where that
# question is answerable by connecting to the port the pointer names. A
# crashed session leaves an inert pointer and nothing listening, so it is
# taken over silently rather than blocking every future run.
SETTINGS_SCRIPT = r'''
import json, os, socket, sys, tempfile

base_url = sys.stdin.readline().strip()
capability = sys.stdin.readline().strip()
# Clearing normally removes only our own pointer. --clear passes force, which
# is the escape hatch for one nothing is behind.
force = sys.stdin.readline().strip() == "force"


def holder_is_live(url):
    """Is anything actually listening on the port this pointer names?"""
    try:
        port = int(url.rsplit(":", 1)[-1])
    except (ValueError, AttributeError):
        return False          # not a pointer imp wrote; nothing to protect
    s = socket.socket()
    s.settimeout(0.5)
    try:
        s.connect(("127.0.0.1", port))
        return True
    except OSError:
        return False
    finally:
        s.close()

path = os.path.expanduser("~/.claude/settings.json")
try:
    with open(path) as f:
        data = json.load(f)
except (IOError, OSError):
    data = {}
except ValueError:
    sys.exit("existing settings.json is not valid JSON; refusing to touch it")

env = data.setdefault("env", {})
held = env.get("ANTHROPIC_AUTH_TOKEN")
warn = ""
if "CLAUDE_CODE_OAUTH_TOKEN" in env:
    warn = "WARNING: a stored CLAUDE_CODE_OAUTH_TOKEN is present in settings.json"

if base_url:
    if held and held != capability:
        held_url = env.get("ANTHROPIC_BASE_URL")
        if holder_is_live(held_url):
            sys.exit("BUSY: another imp session is live on %s. Stop it first."
                     % held_url)
        warn = (warn + " | " if warn else "") + "took over a dead session's pointer"
    env["ANTHROPIC_BASE_URL"] = base_url
    env["ANTHROPIC_AUTH_TOKEN"] = capability
else:
    if held is None:
        warn = (warn + " | " if warn else "") + "no pointer was set"
    elif held != capability and not force:
        # Another session took over while we were running. Its pointer is
        # live and ours is already gone; removing it would break them.
        print("left another session's capability in place")
        sys.exit(0)
    elif held != capability:
        # Say what was taken and whether anything is behind it, because the
        # answer decides whether there is a session somewhere to go and stop.
        note = ("something is STILL listening on it"
                if holder_is_live(env.get("ANTHROPIC_BASE_URL"))
                else "nothing was listening on it")
        warn = (warn + " | " if warn else "") + (
            "cleared a pointer we did not own (%s...); %s" % (held[:8], note))
    env.pop("ANTHROPIC_BASE_URL", None)
    env.pop("ANTHROPIC_AUTH_TOKEN", None)
    if not env:
        data.pop("env", None)

d = os.path.dirname(path)
os.makedirs(d, exist_ok=True)
# A unique temp name: two writers sharing a fixed one interleave, and each
# os.replace is atomic only with respect to its own content.
fd, tmp = tempfile.mkstemp(prefix=".imp.", suffix=".tmp", dir=d)
try:
    with os.fdopen(fd, "w") as f:
        json.dump(data, f, indent=2)
        f.write("\n")
    os.chmod(tmp, 0o600)
    os.replace(tmp, path)
except BaseException:
    try:
        os.unlink(tmp)
    except OSError:
        pass
    raise
print("ok" + ((" | " + warn) if warn else ""))
'''


def remote_python(src, args=()):
    b64 = base64.b64encode(src.encode()).decode()
    argv = " ".join("'%s'" % a for a in args)
    return ("python3 -u -c \"exec(__import__('base64').b64decode('%s'))\" %s"
            % (b64, argv))


def write_remote_settings(target, base_url, capability, force=False):
    """Install (or clear) the far side's pointer at us. Neither value is a
    secret.

    Installing refuses to displace another session's capability while that
    session is still listening; clearing only ever removes our own.
    """
    argv = target.exec_argv(remote_python(SETTINGS_SCRIPT))
    try:
        r = subprocess.run(argv,
                           input="%s\n%s\n%s\n" % (base_url, capability,
                                                    "force" if force else ""),
                           capture_output=True, text=True,
                           timeout=SETTINGS_TIMEOUT)
    except subprocess.TimeoutExpired:
        # The far side controls when this command finishes and how much
        # it prints. Teardown must not be something it can hold open.
        return False, "timed out after %.0fs" % SETTINGS_TIMEOUT
    if r.returncode != 0:
        return False, (r.stderr.strip() or "ssh failed")
    return True, r.stdout.strip()


def clear_target(target):
    """Force-remove the far side's pointer, whoever wrote it.

    There is nothing running on the far side to kill -- an imp session is a
    process on somebody's machine plus a line in a settings.json -- so this
    removes the line and reports whether anything is still answering behind
    it. Something listening means a session really is live, on a machine that
    is not necessarily this one; nothing listening means the pointer was the
    whole of it.
    """
    ok, msg = write_remote_settings(target, "", "", force=True)
    if not ok:
        log("could not clear %s: %s" % (target.label(), msg))
        return 1
    # The script reports through the same "ok | note" line every other call
    # uses; for a clear, the note is the whole of the news.
    log("%s: %s" % (target.label(),
                    msg[5:] if msg.startswith("ok | ") else "cleared"))
    return 0


# --------------------------------------------------------------------------

def port_arg(s):
    """A port, checked here rather than discovered as a bind failure -- or,
    for 0, as a settings.json pointing at http://127.0.0.1:0."""
    try:
        n = int(s)
    except ValueError:
        raise argparse.ArgumentTypeError("%r is not a number" % s)
    if not 1 <= n <= 65535:
        raise argparse.ArgumentTypeError("port must be 1-65535, not %d" % n)
    return n


def consoles_arg(s):
    try:
        n = int(s)
    except ValueError:
        raise argparse.ArgumentTypeError("%r is not a number" % s)
    if not 1 <= n <= 32:
        raise argparse.ArgumentTypeError("consoles must be 1-32, not %d" % n)
    return n


def set_ready(path, ready):
    """Raise or lower the readiness flag `imp` waits on.

    A file, not a line on stderr. The thing waiting is a shell script, and
    `test -e` is the whole of what it has to be able to do -- scraping the
    log for the word `ready` would make that log a format nobody meant to
    promise. Best-effort in both directions: nothing here is worth failing a
    session over, and the file lives in a directory `imp` cleans up anyway.
    """
    if not path:
        return
    try:
        if ready:
            with open(path, "w") as f:
                f.write("ready\n")
        else:
            os.remove(path)
    except OSError:
        pass


def tmux_label(name):
    """A window name `imp` can pass to tmux without quoting worries."""
    safe = "".join(c if (c.isalnum() or c in "_-") else "-" for c in name)
    return safe or "sprite"


# --------------------------------------------------------------------------
# Remembering a session, so that Ctrl-C can be an accident
#
# The capability is minted here and handed over once, through the far side's
# settings.json. Stopping the proxy takes our half of that away and leaves
# theirs -- the processes keep running, pointed at a port with nothing behind
# it. Which is exactly what revoking should look like, and exactly what a
# mistyped Ctrl-C looks like too.
#
# A running `claude` does re-read settings.json, so a fresh proxy with a
# fresh capability is picked up by one that was already going. That is Claude
# Code's behaviour and not ours to promise, and it is worth not depending on:
# the capability is written down here, and --reattach hands the same one
# back, so a proxy that comes back is a proxy the far side never noticed had
# gone.
# It is not a credential. It names one session, buys nothing without a proxy
# of ours listening on the far side's port, and the far side has held it
# since the moment it was minted. 0600 all the same, and only ever reused
# because someone asked.

def new_capability():
    """The far side's whole share of a session: a name for it, and nothing
    that is worth anything without a proxy of ours behind the port."""
    return base64.urlsafe_b64encode(os.urandom(24)).decode().rstrip("=")


# What one looks like, taken from one rather than written down twice.
CAPABILITY_LEN = len(new_capability())


def state_dir():
    base = os.environ.get("XDG_STATE_HOME") or os.path.expanduser("~/.local/state")
    return os.path.join(base, "imp")


def state_path(target, remote_port):
    """One file per label per port -- keyed the way the tmux session is.

    The same host on two ports is two sessions funding two sets of consoles,
    and handing one's capability back to the other would fund neither.
    """
    return os.path.join(state_dir(), "%s-%d.json"
                        % (tmux_label(target.short_label()), remote_port))


def save_capability(target, remote_port, capability):
    """Best effort, deliberately: a session that cannot write this is a
    session that works, minus the ability to be reattached to."""
    d = state_dir()
    try:
        os.makedirs(d, mode=0o700, exist_ok=True)
        fd, tmp = tempfile.mkstemp(prefix=".imp.", suffix=".tmp", dir=d)
        try:
            with os.fdopen(fd, "w") as f:
                json.dump({"capability": capability,
                           "port": remote_port,
                           "target": target.label(),
                           # Which command reaches it, for anything that has
                           # only the label -- `imp --meter` samples a sprite
                           # and an ssh host differently.
                           "kind": target.kind,
                           "written": int(time.time())}, f)
                f.write("\n")
            os.chmod(tmp, 0o600)
            os.replace(tmp, state_path(target, remote_port))
        except BaseException:
            try:
                os.unlink(tmp)
            except OSError:
                pass
            raise
    except OSError:
        pass


def load_capability(target, remote_port):
    """The capability this target's last session used, or None and why not.

    The reason is printed rather than swallowed: the difference between
    consoles that resume and consoles that have to be restarted is the whole
    of what --reattach promises, and it should not be something you discover
    by watching one fail.
    """
    try:
        with open(state_path(target, remote_port)) as f:
            rec = json.load(f)
    except (IOError, OSError):
        return None, ("no earlier session on %s port %d was written down"
                      % (target.label(), remote_port))
    except ValueError:
        return None, "what was written down is not readable"
    cap = rec.get("capability") or ""
    # Shaped like something we minted. A file someone has edited to hold a
    # real ANTHROPIC_API_KEY is the case worth refusing outright: it would
    # otherwise be installed on the far side as a capability, which is the
    # one thing this program exists not to do.
    if not re.match(r"\A[A-Za-z0-9_-]{%d}\Z" % CAPABILITY_LEN, cap):
        return None, "what was written down is not a capability of ours"
    age = time.time() - rec.get("written", 0)
    if age > STATE_TTL:
        return None, ("the last session there ended %.0f hours ago"
                      % (age / 3600.0))
    return cap, ""


def forget_capability(target, remote_port):
    """--clear ends a session, and this is the part of one that lives here.

    A pointer that has been taken out is not something to reattach to; the
    consoles it funded are gone or unfunded either way.
    """
    try:
        os.remove(state_path(target, remote_port))
    except OSError:
        pass


def main():
    ap = argparse.ArgumentParser(
        prog="imp-proxy",
        description="Lend a remote machine -- any host you can ssh to, or a "
                    "Fly Sprite -- your Claude Max subscription without "
                    "giving it the credential.",
        epilog="The credential never leaves this machine. Stop the proxy "
               "(Ctrl-C) and the far side's access dies immediately -- running "
               "`claude` processes there simply start failing API calls, so "
               "revoking costs you nothing but a reconnect.")
    ap.add_argument("dest", nargs="?", default="", metavar="HOST",
                    help="an ssh destination -- anything `ssh` accepts, "
                         "including a Host from your ssh config")
    ap.add_argument("-H", "--host", default="",
                    help="the same as HOST, spelled as an option")
    ap.add_argument("-s", "--sprite", default="",
                    help="a Fly Sprite instead of an ssh host (default when "
                         "neither is given: the one `sprite use` selected in "
                         "this directory)")
    ap.add_argument("-p", "--remote-port", type=port_arg,
                    default=DEFAULT_REMOTE_PORT,
                    help="port to listen on at the far end (default %d)"
                         % DEFAULT_REMOTE_PORT)
    ap.add_argument("--no-settings", action="store_true",
                    help="do not touch the far side's ~/.claude/settings.json; "
                         "print the env vars for you to set yourself")
    ap.add_argument("--allow", action="append", default=[], metavar="GLOB",
                    help="allow an extra request path through; repeatable, "
                         "globs accepted")
    ap.add_argument("--allow-any-path", action="store_true",
                    help="disable the path allowlist entirely -- the proxy "
                         "becomes an open relay to your account for the "
                         "length of the session")
    ap.add_argument("--clear", action="store_true",
                    help="remove the far side's ANTHROPIC_* pointer, whoever "
                         "left it, and say whether anything is still behind "
                         "it. The way out of a pointer with nothing running "
                         "under it; starts no proxy and reads no credential")
    ap.add_argument("-n", "--consoles", type=consoles_arg, default=3,
                    help="how many console windows `imp` should open "
                         "(default 3). Ignored when imp-proxy is run "
                         "directly -- it opens no windows")
    ap.add_argument("--reattach", action="store_true",
                    help="reuse the capability the last session on this "
                         "target held, so `claude` processes still running "
                         "there are funded again without being restarted. "
                         "`imp --reattach` also puts the proxy back in the "
                         "tmux session it was killed out of")
    ap.add_argument("-v", "--verbose", action="store_true",
                    help="log each proxied request")
    # Written when the far side is actually funded, removed on the way out.
    # `imp` waits on it before starting `claude` in its console windows.
    ap.add_argument("--ready-file", default="", help=argparse.SUPPRESS)
    # For `imp`, which needs to know which console to open beside us. Not in
    # --help: it is one program talking to another, not a thing to run.
    ap.add_argument("--print-console", action="store_true",
                    help=argparse.SUPPRESS)
    args = ap.parse_args()
    if args.dest and args.host and args.dest != args.host:
        die("two hosts named (%s and %s); pick one." % (args.dest, args.host))
    target = resolve_target(args.dest or args.host, args.sprite)

    if args.print_console:
        # `imp` asks us instead of parsing argv itself. argparse accepts
        # -sNAME, --sprite=NAME and any unambiguous abbreviation, and a shell
        # loop that only knew "-s NAME" answered "no sprite named" to all
        # three -- opening a console on a different machine than the one
        # being funded. One parser, one answer, and the drift cannot recur.
        #
        # Three lines: whether this invocation is a session at all, a window
        # label, and the console command ready for a shell to re-parse.
        print("mode=%s" % ("oneshot" if args.clear else "session"))
        print("reattach=%d" % (1 if args.reattach else 0))
        print("label=%s" % tmux_label(target.short_label()))
        print("consoles=%d" % args.consoles)
        print("console=%s" % " ".join(shlex.quote(a)
                                      for a in target.console_argv()))
        return 0

    if args.clear:
        forget_capability(target, args.remote_port)
        return clear_target(target)

    cred = Credential()
    capability = new_capability()
    log("credential loaded (%d chars) from %s" % (len(cred.token), cred.source))
    if args.reattach:
        prior, why = load_capability(target, args.remote_port)
        if prior:
            capability = prior
            log("reattached -- reusing the capability %s already holds"
                % target.label())
        else:
            log("nothing to reattach to: %s" % why)
            log("a new capability instead -- `claude` already running there "
                "has to be restarted to pick it up")
    save_capability(target, args.remote_port, capability)

    allowed = () if args.allow_any_path else DEFAULT_ALLOWED + tuple(args.allow)
    if allowed:
        log("path allowlist: %s" % " ".join(allowed))
    else:
        log("path allowlist DISABLED -- any path reaches your account")

    handler = make_proxy_handler(cred, capability, args.verbose, allowed)
    proxy = ProxyServer(("127.0.0.1", 0), handler)
    proxy_port = proxy.server_address[1]
    threading.Thread(target=proxy.serve_forever, daemon=True).start()
    log("injecting proxy on 127.0.0.1:%d (loopback only)" % proxy_port)

    base_url = "http://127.0.0.1:%d" % args.remote_port

    # Ownership is settled before the link exists, and the ordering is the
    # whole of the check. Once our own forward is listening on the far side's
    # port, holder_is_live() connects to it and reports *us* as the incumbent
    # -- so a pointer left behind by a killed session would refuse every
    # future run, permanently, with nothing there to go and stop.
    settings_written = False
    if not args.no_settings:
        ok, msg = write_remote_settings(target, base_url, capability)
        if ok:
            settings_written = True
            log("remote settings.json points at us (%s)" % msg)
        elif "BUSY:" in msg:
            # Another live session owns the far side. Breaking it is never the
            # right default -- stop here rather than half-taking-over.
            log(msg.replace("BUSY: ", ""))
            log("if nothing is really running there: imp-proxy --clear")
            proxy.shutdown()
            return 1
        else:
            log("could not write remote settings.json: %s" % msg)
            log("set these on the far side yourself:")
            log("  ANTHROPIC_BASE_URL=%s" % base_url)
            log("  ANTHROPIC_AUTH_TOKEN=%s" % capability)
    else:
        log("set these on the far side yourself:")
        log("  ANTHROPIC_BASE_URL=%s" % base_url)
        log("  ANTHROPIC_AUTH_TOKEN=%s" % capability)

    proc = subprocess.Popen(target.link_argv(args.remote_port, proxy_port),
                            stdin=subprocess.PIPE, stdout=subprocess.PIPE,
                            stderr=None, bufsize=0)

    def on_signal(signum, frame):
        try:
            proc.terminate()
        except OSError:
            pass

    # SIGHUP too, and not for tidiness: Python's default for it kills the
    # process outright, so the teardown below -- the one that clears the
    # far side's settings.json -- never runs. A terminal closing or a tmux pane
    # being destroyed both arrive that way, which makes the ordinary way to
    # end a session the one path that skips cleanup.
    for sig in (signal.SIGINT, signal.SIGTERM, signal.SIGHUP):
        signal.signal(sig, on_signal)

    # The sprite gets the userspace relay and its demultiplexer; a plain ssh
    # host gets -R, where the kernel on both ends is the demultiplexer and
    # there is nothing here to run.
    mux = None
    if target.needs_relay:
        mux = Mux(proc, proxy_port, args.verbose)
        threading.Thread(target=mux.run, daemon=True).start()
        linked = wait_for_link(mux, proc, LINK_TIMEOUT)
    else:
        linked = wait_for_marker(proc, LINK_TIMEOUT)
    if not linked:
        log("no link to '%s' -- it failed or timed out." % target.label())
        if target.needs_relay:
            log("check `sprite ls` and that python3 exists on the sprite.")
        else:
            log("check `ssh %s true`, and that port %d is free there."
                % (target.label(), args.remote_port))
        try:
            proc.terminate()
        except OSError:
            pass
        # The pointer went in before the link was tried, so a link that never
        # came up leaves one to take back out. Said out loud when it fails,
        # the way the teardown below does -- a pointer left behind is the
        # thing the next run will refuse to start over.
        if settings_written:
            ok, msg = write_remote_settings(target, "", capability)
            if not ok:
                log("left an inert capability in settings.json (%s)" % msg)
        proxy.shutdown()
        return 1

    log("ready -- `claude` on %s now runs on your Max subscription."
        % target.label())
    log("Ctrl-C revokes instantly. Nothing to clean up there.")
    set_ready(args.ready_file, True)

    try:
        while proc.poll() is None and (mux is None or mux.alive):
            time.sleep(0.25)
    finally:
        set_ready(args.ready_file, False)
        log("link closed -- access revoked")
        try:
            proc.terminate()
        except OSError:
            pass
        if settings_written:
            ok, msg = write_remote_settings(target, "", capability)
            log("cleared remote settings.json" if ok
                else "left an inert capability in settings.json (%s)" % msg)
        proxy.shutdown()
    return 0


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