#!/usr/bin/env python3
# Host-side credential-injecting proxy for sandboxed sessions.
#
# The container only ever holds a placeholder token; this proxy (launched on
# the host by bin/dejima, reached via a bind-mounted unix socket and an
# in-container loopback forwarder) strips the placeholder and injects the real
# credential, so the credential itself never enters the container. See
# docs/security.md.
#
# NOTHING HERE NAMES A BACKEND. Which upstream, which credential, how it is
# read, which header carries it and what shape that header takes are all
# arguments (#171). The Anthropic arrangement this started as is one set of
# those arguments and no longer a special case in the code, because the moment
# a second backend exists, a proxy that knows one backend's name has to grow
# a branch for every other.
#
# Two credential kinds, and the difference is refresh. A static key is read and
# sent. An OAuth credential expires, so it is refreshed against an endpoint and
# the rotated token written back to the host file atomically - which is what
# keeps a host-side session logged in rather than being signed out by a sandbox
# run. A backend says which kind it has; this decides nothing on its own.
import argparse
import http.client
import json
import os
import socketserver
import sys
import tempfile
import threading
import time
import zlib
from http.server import BaseHTTPRequestHandler
from urllib.parse import urlsplit

# Hop-by-hop headers describe the connection to this proxy rather than the
# request, so they stop here (RFC 9110 7.6.1). Host is deliberately not in
# this set: it has to be rewritten for the upstream, and is replaced in place
# rather than dropped and re-added, which keeps the header block in the order
# the client wrote it - see _upstream.
HOP_BY_HOP = {
    "connection", "keep-alive", "proxy-authenticate", "proxy-authorization",
    "te", "trailer", "trailers", "transfer-encoding", "upgrade",
}
# Casing for the headers this proxy has to supply when the client sent none of
# its own. Everything the client did send keeps the client's casing; these are
# only consulted for a header being added rather than replaced. The auth and
# ensure-header names are added to this at startup, in the spelling the
# backend declared, for the same reason.
SUPPLIED_NAMES = {
    "host": "Host",
    "content-length": "Content-Length",
}


# What an operator has to do when the stored credential can no longer be
# refreshed, in one place. Both the startup check and a refresh that the
# provider rejects reach this, and an operator who reads one and not the other
# would otherwise get the remedy from whichever happened to fire.
# Written to be read in a crowded terminal: the action first, in one short
# sentence, because this is competing with a wall of harness output and an
# operator scanning it should not have to reach the end of a clause to find the
# verb. What went wrong is the subordinate half - by the time this is printed,
# what to do about it is the only part that changes anything.
REAUTH_REMEDY = ("run `claude` on the host to log in again - the stored "
                 "credential expired and cannot be refreshed")


class CredentialNeedsOperator(RuntimeError):
    """A credential failure whose remedy is a command, not a diagnosis.

    Carries two texts. The message is written here rather than derived from a
    provider's response, which is what makes it safe to send to the container
    as well as the host; `detail` is whatever the provider actually said and
    stays host-side. The rule being respected is that upstream bytes and
    exception detail do not cross into the session - not that the session must
    be told nothing.
    """

    def __init__(self, message, detail=None):
        super().__init__(message)
        self.detail = detail or message


class StaticCredential:
    """A key in a file, sent as-is. No expiry, so nothing to refresh."""

    # Trailing newline stripped, because a key file is written by an editor or
    # a `>` redirection as often as by a tool, and a newline inside a header
    # value is not a cosmetic difference: it would either be rejected or split
    # the header block. Refused rather than trimmed if anything is left that
    # cannot appear in a header, since a key that needs sanitising is a key
    # that was pasted wrong, and guessing at it would send a mangled
    # credential upstream and report the backend's rejection as the fault.
    # A field, when the key is one value inside a document rather than the
    # whole file. Claude Code signed in with a console API token keeps its key
    # as primaryApiKey in ~/.claude.json, beside unrelated app state, so a
    # backend has to be able to say which value is the credential.
    def __init__(self, path, field=None):
        self.path = path
        self.field = field

    def token(self):
        with open(self.path) as f:
            raw = f.read()
        if self.field is None:
            value = raw.strip()
        else:
            try:
                doc = json.loads(raw)
            except ValueError as e:
                raise RuntimeError("%s is not JSON: %s" % (self.path, e))
            cur = doc
            for seg in self.field.split("."):
                if not isinstance(cur, dict) or seg not in cur:
                    raise RuntimeError(
                        "%s has no %s" % (self.path, self.field))
                cur = cur[seg]
            if not isinstance(cur, str):
                raise RuntimeError(
                    "%s: %s is not a string" % (self.path, self.field))
            value = cur.strip()
        if not value:
            raise RuntimeError("%s is empty" % self.path)
        bad = [c for c in value if ord(c) < 0x20 or ord(c) == 0x7F]
        if bad:
            raise RuntimeError("%s contains a control character" % self.path)
        return value

    def refresh(self, stale_token):
        # Nothing to refresh. A 401 from a static key means the key is wrong,
        # and retrying with the same bytes would turn one rejection into two.
        raise RuntimeError("this credential cannot be refreshed")

    @property
    def refreshable(self):
        return False

    def startup_check(self):
        self.token()


class OAuthCredential:
    def __init__(self, path, oauth_endpoint, client_id, margin):
        self.path = path
        self.oauth_endpoint = oauth_endpoint
        self.client_id = client_id
        self.margin = margin
        self.lock = threading.Lock()

    @property
    def refreshable(self):
        return True

    def startup_check(self):
        # A plain read, not token(): this is a startup sanity check and must
        # not trigger a refresh (that belongs to the first real request, so an
        # external refresher gets a chance to win first).
        self._read()["claudeAiOauth"]["accessToken"]
        expires = self._read()["claudeAiOauth"].get("refreshTokenExpiresAt")
        if expires is not None and expires / 1000 < time.time():
            # Concrete, not generic. This class reads Claude Code's own
            # credential shape, so the remedy is a command the operator can run
            # rather than an instruction to work out.
            raise CredentialNeedsOperator(REAUTH_REMEDY)

    def _read(self):
        with open(self.path) as f:
            return json.load(f)

    def token(self):
        oauth = self._read()["claudeAiOauth"]
        if oauth["expiresAt"] / 1000 - time.time() < self.margin:
            self.refresh(stale_token=oauth["accessToken"])
            oauth = self._read()["claudeAiOauth"]
        return oauth["accessToken"]

    def refresh(self, stale_token):
        with self.lock:
            data = self._read()
            oauth = data["claudeAiOauth"]
            # A concurrent refresher (host-side Claude Code, another sandbox
            # session, another thread here) may have replaced the token
            # already - if the file no longer holds the stale one and isn't
            # near expiry, use what's there instead of refreshing again.
            if (oauth["accessToken"] != stale_token
                    and oauth["expiresAt"] / 1000 - time.time() >= self.margin):
                return
            u = urlsplit(self.oauth_endpoint)
            cls = http.client.HTTPSConnection if u.scheme == "https" else http.client.HTTPConnection
            conn = cls(u.netloc, timeout=30)
            req = json.dumps({
                "grant_type": "refresh_token",
                "refresh_token": oauth["refreshToken"],
                "client_id": self.client_id,
            })
            conn.request("POST", u.path, body=req,
                         headers={"Content-Type": "application/json"})
            resp = conn.getresponse()
            payload = resp.read()
            conn.close()
            if resp.status != 200:
                # invalid_grant is the provider saying this refresh token is
                # finished - not a transient failure, and not something waiting
                # or retrying improves. It is what an overnight session meets in
                # the morning, and the raw body says `invalid_grant` without
                # saying that logging in on the host is the way out.
                #
                # The body is still logged in full: this only decides which
                # remedy to name.
                if b"invalid_grant" in payload:
                    raise CredentialNeedsOperator(
                        REAUTH_REMEDY,
                        detail="token refresh failed: HTTP %d %.200s"
                               % (resp.status, payload))
                raise RuntimeError("token refresh failed: HTTP %d %.200s"
                                   % (resp.status, payload))
            fresh = json.loads(payload)
            oauth["accessToken"] = fresh["access_token"]
            if fresh.get("refresh_token"):
                oauth["refreshToken"] = fresh["refresh_token"]
            oauth["expiresAt"] = int((time.time() + fresh["expires_in"]) * 1000)
            self._write(data)

    def _write(self, data):
        # Atomic and permission-preserving: host-side Claude Code reads this
        # same file and must never observe a partial write or a looser mode.
        # Staged in a dedicated subdirectory rather than ~/.claude directly,
        # which keeps the plaintext window out of any directory the sandbox
        # can see, and keeps the rename below same-filesystem (a subdirectory
        # of the same parent) so atomicity is unaffected.
        #
        # A bind mount over the credential would not make this safe: the
        # rename below is unaffected by the container's mount, but the
        # container's mount is destroyed by the rename, which unlinks the
        # dentry it is pinned to. ~/.claude is not mounted whole, so neither
        # this directory nor the credential has a path into the container
        # (#50, docs/security.md).
        directory = os.path.join(os.path.dirname(self.path) or ".", ".credentials-proxy-tmp")
        os.makedirs(directory, mode=0o700, exist_ok=True)
        fd, tmp = tempfile.mkstemp(dir=directory)
        try:
            with os.fdopen(fd, "w") as f:
                json.dump(data, f)
                f.flush()
                os.fsync(f.fileno())
            os.chmod(tmp, 0o600)
            os.replace(tmp, self.path)
        except BaseException:
            os.unlink(tmp)
            raise


class AuthShape:
    """How a backend wants its credential presented.

    header/format are the injection; strip is what must not travel beside it;
    ensure is a comma-list header that has to contain a value, used for the
    Anthropic OAuth beta and general enough to need no name of its own here.
    """

    def __init__(self, header, fmt, strip, ensure):
        self.header = header
        self.format = fmt
        # The auth header is never also stripped: it is replaced, and listing
        # it in both places would drop the injected value instead of sending
        # it.
        self.strip = frozenset(n.lower() for n in strip) - {header.lower()}
        self.ensure = ensure


class BodyTooLarge(Exception):
    """A request body over the cap. Its own type, not ValueError: _read_body
    raises ValueError for a genuinely malformed request (a chunk length that
    is not hex), and that must still surface as a fault rather than be
    answered as if the session had merely sent too much."""


# What an upstream error body says, as text fit to put on a terminal.
#
# Two things stand between the bytes and the log. The client's own
# Accept-Encoding is forwarded verbatim - a Node harness asks for gzip - so
# what comes back is compressed, and the first version wrote those bytes
# straight out: mojibake on the operator's terminal and a bash warning about
# a null byte in a command substitution. And whatever the encoding, this is
# upstream-controlled text landing where dejima-audit already refuses to put
# session-controlled text unescaped.
#
# So: decompress what the stdlib can, then keep only printable characters.
# The sanitising is the half that has to hold - it is not conditional on
# recognising the encoding, and an unknown one degrades to a note rather than
# to bytes.
def _brotli_decompress(raw):
    """A brotli decoder if the host has one, else None.

    Optional on purpose. This runs on the host as the operator's user, and a
    diagnostic is not worth a hard dependency on a module Debian ships
    separately - so a host without it gets a note naming what to install
    rather than a launch that refuses.

    Not solved by asking the API for an encoding this can read: the request
    the API sees is deliberately the harness's own header block, byte for
    byte, and Accept-Encoding is part of what makes a session look like a
    stock install. Rewriting it to make a log line readable trades a property
    the sandbox is for against a convenience.
    """
    for name in ("brotli", "brotlicffi"):
        try:
            mod = __import__(name)
        except ImportError:
            continue
        try:
            # A bounded prefix is a partial stream. The one-shot decompress
            # wants a whole one, so this is the incremental decoder in both
            # modules - and a prefix that ends mid-symbol yields what it can.
            return mod.Decompressor().process(raw)
        except Exception:
            return None
    return None


def _readable(raw, encoding):
    enc = (encoding or "").strip().lower()
    if enc in ("gzip", "x-gzip", "deflate"):
        try:
            # A bounded prefix is a partial stream, so decompressobj rather
            # than gzip.decompress, which wants a whole member.
            wbits = 16 + zlib.MAX_WBITS if enc != "deflate" else zlib.MAX_WBITS
            raw = zlib.decompressobj(wbits).decompress(raw)
        except zlib.error:
            return "(%d bytes, %s, undecodable)" % (len(raw), enc)
    elif enc == "br":
        decoded = _brotli_decompress(raw)
        if decoded is None:
            return ("(%d bytes, br-encoded - no brotli module on this host, so "
                    "the message cannot be read here; pip install brotli, or "
                    "apt install python3-brotli)" % len(raw))
        raw = decoded
    elif enc:
        # zstd and anything else. Saying which is more use than printing the
        # bytes, and printing the bytes is what this exists to stop.
        return "(%d bytes, %s-encoded)" % (len(raw), enc)
    # Decodes only. Making the text fit for a terminal is _diag's, so that
    # every producer is covered rather than this one path.
    return raw.decode("utf-8", "replace").strip()


# One line, timestamped, appended. Falls back to stderr when the file cannot
# be written rather than losing the message: a diagnostic that disappears
# because the thing that reports problems had a problem is the worst kind.
def _diag(path, line):
    # Sanitised at the sink, not at each producer. bin/dejima prints this file
    # to the operator's terminal after a session ends, so it is a terminal-
    # bound channel carrying text chosen by an upstream API and, through the
    # request path and exception messages, by the session itself. That is the
    # hazard docs/security.md describes for audit findings, which are
    # hex-encoded for it.
    #
    # Nothing gets through today - CPython's request-line parser and
    # http.client both refuse a control character in a URL, and both repr what
    # they refuse - but that is two coincidences rather than a property, and
    # neither of them covers a producer added later.
    line = "".join(c if c.isprintable() or c == " " else " " for c in str(line))
    stamped = "%s %s\n" % (time.strftime("%H:%M:%S"), line)
    if path:
        try:
            with open(path, "a") as fh:
                fh.write(stamped)
            return
        except OSError:
            pass
    sys.stderr.write(stamped)


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

    def log_request(self, *args):
        pass  # per-request lines would interleave with the session's own UI

    def log_error(self, fmt, *args):
        _diag(getattr(self.server, "diag_path", None),
              "dejima-credential-proxy: " + fmt % args)

    # How much of an upstream error body is worth keeping. An API's rejection
    # is a short JSON object; anything longer is not one, and this runs on the
    # host with the session on the other end of it.
    MAX_DIAG_BODY = 2048

    # Capped, because the client is the sandboxed session and this runs on the
    # host as the operator's user. Every other session-controlled size in this
    # repo is bounded (dejima-audit's STAGE_FILE_CAP and DIFF_BYTE_CAP,
    # dejima-project's declare_entry, the briefing fragment limit), several
    # with explicit reasoning about $TMPDIR being tmpfs; this was the one
    # place that discipline was not applied, and a request body goes straight
    # into host RAM.
    #
    # 32 MiB is far above any real Messages API request - the largest thing
    # one carries is a conversation plus attachments - and far below anything
    # that troubles a host. Refused rather than truncated: half a request
    # forwarded upstream is a worse outcome than a rejected one.
    MAX_BODY_BYTES = 32 * 1024 * 1024

    def _length(self, raw, what):
        """A byte count from the wire, or BodyTooLarge.

        `int()` alone accepts a negative, and a negative passes a `> MAX` test
        and then means "read to EOF" to a buffered reader - so the cap above,
        whose whole reason is that this runs on the host as the operator, was
        bypassed by sending `Content-Length: -1`.

        A value that is not a number still raises out of here, deliberately: the
        suppression around handle_error is meant to be specific to the client
        going away, and a malformed chunk length is the fault the test for that
        uses to prove it. Refusing it cleanly here would remove the canary and
        trade noise for blindness.
        """
        n = int(raw, 16) if what == "chunk size" else int(raw)
        if n < 0 or n > self.MAX_BODY_BYTES:
            # The offending value, not the cap. The host log said "refused a
            # request body: 33554432 bytes" whatever arrived, so a hostile -1
            # and a genuinely oversized upload read identically - and the
            # comment on the 413 path promises the log says what came in.
            raise BodyTooLarge("%s of %d, cap %d"
                               % (what, n, self.MAX_BODY_BYTES))
        return n

    def _read_body(self):
        if self.headers.get("Transfer-Encoding", "").lower() == "chunked":
            chunks = []
            total = 0
            while True:
                size = self._length(self.rfile.readline().split(b";")[0], "chunk size")
                if size == 0:
                    self.rfile.readline()
                    return b"".join(chunks)
                total += size
                if total > self.MAX_BODY_BYTES:
                    raise BodyTooLarge("chunked body of %d or more, cap %d"
                                       % (total, self.MAX_BODY_BYTES))
                chunks.append(self.rfile.read(size))
                self.rfile.read(2)
        length = self._length(self.headers.get("Content-Length", 0), "Content-Length")
        return self.rfile.read(length)

    def _fail(self, status, message, log_message=None):
        # log_message (host stderr) may contain upstream response bytes or
        # exception detail; message (sent to the container in the response body)
        # never should. The one thing that does cross is text written in this
        # file naming something the operator must do - see CredentialNeedsOperator,
        # and docs/security.md for why that is not an exception to the rule.
        self.log_error("%s", log_message if log_message is not None else message)
        body = json.dumps(
            {"type": "error", "error": {"type": "proxy_error", "message": message}}
        ).encode()
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def _upstream(self, token, body):
        u = self.server.upstream
        cls = http.client.HTTPSConnection if u.scheme == "https" else http.client.HTTPConnection
        conn = cls(u.netloc, timeout=600)
        # Emitted header by header rather than handed to conn.request() as a
        # dict, so what the API receives is the client's own header block -
        # same names, same casing, same order, duplicates intact - with only
        # the values that must change changed. The container half of this
        # sandbox goes to some length to make the session indistinguishable
        # from a stock install (no ANTHROPIC_* variables, no explicit base
        # URL - see docs/security.md); a request rewritten into CPython's
        # house style would put the fingerprint straight back on the wire,
        # where it is visible to the API rather than only to the client.
        #
        # conn.request() does exactly that in three ways: a dict collapses
        # repeated headers to the last one, dict order puts every header this
        # proxy touches at the end of the block, and putrequest() supplies
        # `Accept-Encoding: identity` - a CPython tell no browser or Node
        # client sends - whenever the caller passed no accept-encoding. The
        # skip_* flags below suppress that and the generated Host header,
        # which is re-emitted in the client's own position instead.
        conn.putrequest(self.command, self.path,
                        skip_host=True, skip_accept_encoding=True)
        auth = self.server.auth
        supplied = {
            "host": u.netloc,
            auth.header.lower(): auth.format.replace("{token}", token),
            "content-length": str(len(body)),
        }
        # An ensure-header is left completely alone when its value is already
        # among the client's own - as anthropic-beta's is in the tls transport,
        # where the harness believes it holds an OAuth credential. Re-joining a
        # list that needs no change would normalize the client's spacing, which
        # is the same class of tell as the casing.
        for name, wanted in auth.ensure:
            low = name.lower()
            present = [v.strip() for header in self.headers.get_all(low, [])
                       for v in header.split(",")]
            if wanted in present:
                continue
            first = self.headers.get(low)
            supplied[low] = first + "," + wanted if first else wanted
        seen = set()
        for name, value in self.headers.items():
            low = name.lower()
            # The strip set is the backend's, not a fixed list. It exists so a
            # header the client used for a credential cannot travel alongside
            # the one this proxy injects: x-api-key beside an Authorization is
            # two answers to the same question, and which one an API prefers is
            # not this program's to guess.
            if low in HOP_BY_HOP or low in auth.strip:
                continue
            if low in supplied:
                # One replacement even for a repeated header, or the injected
                # token would be sent as many times as the client sent its
                # placeholder.
                if low in seen:
                    continue
                value = supplied[low]
            seen.add(low)
            conn.putheader(name, value)
        # Only names registered above are emitted, so a value put in `supplied`
        # whose name is not in SUPPLIED_NAMES is computed and then silently
        # dropped. That is deliberate - it is what keeps this pass from
        # inventing a header the backend never declared - but it means the two
        # have to be populated from the same place, which is why main() derives
        # SUPPLIED_NAMES from the auth header and the ensure list rather than
        # listing them.
        for low, name in SUPPLIED_NAMES.items():
            if low in supplied and low not in seen:
                # Content-Length only when there is a body: a bodyless GET
                # that carried none must not gain one. A chunked request has
                # been de-framed by now, so it lands here and does gain one.
                if low == "content-length" and not body:
                    continue
                conn.putheader(name, supplied[low])
        conn.endheaders(body)
        return conn, conn.getresponse()

    def _proxy(self):
        try:
            body = self._read_body()
        except BodyTooLarge as e:
            # 413 rather than a traceback and a 500: the session gets a status
            # that says what to do about it, and the host log says what came
            # in. Nothing is forwarded upstream.
            #
            # And the connection ends here. The body was refused before it was
            # read, so its bytes are still queued; left open, the next parse
            # takes them as a request line and answers a keep-alive client with
            # garbage 400s instead of one clean refusal.
            self.close_connection = True
            return self._fail(413, "credential proxy: request body too large",
                              log_message="refused a request body: %s" % e)
        try:
            token = self.server.creds.token()
        except CredentialNeedsOperator as e:
            # This one names the remedy in the response as well. The session is
            # where the failure is met, the host terminal is behind a window the
            # agent cannot read, and "see host terminal" costs an operator the
            # trip to find one sentence. Safe to send because the text is this
            # file's own - see the class.
            # Two lines, detail first. bin/dejima's exit audit prints the LAST
            # line of this log when a session ends, so the one an operator reads
            # without going looking is the one that says what to do. The
            # provider's own words stay directly above it for anyone who does go
            # looking.
            self.log_error("%s", "cannot load credentials: %s" % e.detail)
            return self._fail(502, "credential proxy: %s" % e, log_message=str(e))
        except Exception as e:
            return self._fail(502, "credential proxy: could not load or refresh the "
                                    "credential - see host terminal",
                               log_message="cannot load credentials: %s" % e)
        try:
            conn, resp = self._upstream(token, body)
            if resp.status == 401 and self.server.creds.refreshable:
                # Upstream disagrees with expiresAt - refresh once and retry.
                # Only for a credential that can be refreshed: retrying a
                # static key sends the same bytes and turns one rejection into
                # two, with the second one looking like a flapping upstream.
                resp.read()
                conn.close()
                self.server.creds.refresh(stale_token=token)
                token = self.server.creds.token()
                conn, resp = self._upstream(token, body)
        except CredentialNeedsOperator as e:
            # The refresh on the 401 retry above can reach this, and it is not
            # an upstream failure - reporting it as one sends whoever reads it
            # looking at the provider's availability instead of at their own
            # login. Same remedy as the load path, for the same reason.
            self.log_error("%s", "cannot refresh credentials: %s" % e.detail)
            return self._fail(502, "credential proxy: %s" % e, log_message=str(e))
        except Exception as e:
            return self._fail(502, "credential proxy: upstream request failed - "
                                    "see host terminal",
                               log_message="upstream request failed: %s" % e)
        try:
            length = resp.getheader("Content-Length")
            # An upstream rejection, said out loud. The proxy is the only
            # thing that sees both the request the session made and the answer
            # the API gave, and it used to forward a 4xx verbatim and record
            # nothing - so a session that "spun for a bit and gave an error"
            # left the operator with no way to find out why. It is not the
            # container's job to explain: the harness shows whatever it shows.
            #
            # Read whole only when upstream declared a small length. A
            # streaming or oversized body is logged by status alone rather
            # than buffered here, which keeps this off the success path
            # entirely - a 200 never gets read into memory.
            early = None
            if resp.status >= 400:
                # Read a bounded prefix whatever the framing. The first
                # version did this only where upstream declared a small
                # Content-Length, and the first real rejection it caught -
                # three 429s from a rate-limited account - recorded the status
                # and nothing else, which is the half an operator cannot act
                # on. read1 returns what has arrived rather than waiting for a
                # full buffer, so this costs one chunk and is forwarded below
                # before the stream continues normally.
                early = resp.read1(self.MAX_DIAG_BODY)
                detail = _readable(early, resp.getheader("content-encoding"))
                # The headers carry more than the body does, which is not a
                # guess: a real rate-limited call came back
                # {"type":"rate_limit_error","message":"Error"} - the word
                # Error and nothing else. request-id is what Anthropic support
                # asks for, and retry-after is the only thing that says how
                # long; neither is in the body, and a reader should not have to
                # know to go looking for them.
                #
                # Deliberately not the account and workspace ids the same
                # response carries. They identify the operator's org, this
                # line gets pasted into bug reports, and request-id is enough
                # to find the request from the other side.
                extra = ""
                for name in ("request-id", "retry-after"):
                    value = resp.getheader(name)
                    if value:
                        extra += " %s=%s" % (name, value)
                self.log_error("upstream answered %s for %s%s%s",
                               resp.status, self.path, extra,
                               " " + detail if detail else "")
            self.send_response(resp.status)
            for k, v in resp.getheaders():
                if k.lower() in HOP_BY_HOP or k.lower() in ("content-length", "date", "server"):
                    continue
                self.send_header(k, v)
            if length is not None:
                self.send_header("Content-Length", length)
            else:
                # Unknown length (chunked SSE): stream until upstream EOF,
                # delimited by closing our side too.
                self.close_connection = True
                self.send_header("Connection", "close")
            self.end_headers()
            self.wfile.flush()
            if early:
                # Already read for the log above; written out before the loop
                # picks up where it left off, so the session receives the same
                # bytes in the same order whether or not anything was logged.
                self.wfile.write(early)
                self.wfile.flush()
            # Stream response body, flushing after each chunk for unbuffered delivery.
            # Use read1() to get partial reads immediately rather than waiting for
            # a full buffer or EOF, ensuring SSE events stream unbuffered. HTTPResponse.read1()
            # transparently handles chunked transfer-encoding and content-length termination.
            while True:
                chunk = resp.read1(4096)
                if not chunk:
                    break
                self.wfile.write(chunk)
                self.wfile.flush()
        finally:
            conn.close()

    # HEAD and OPTIONS included: the method set this accepts should be the
    # upstream API's, not whichever handlers happen to be defined here. A
    # method with no do_* lands in BaseHTTPRequestHandler's 501, which the
    # container sees as the API rejecting it and the host sees as a logged
    # error - both misattributed to a request the proxy simply didn't route.
    do_GET = do_POST = do_PUT = do_DELETE = do_PATCH = do_HEAD = do_OPTIONS = _proxy


class ProxyServer(socketserver.ThreadingMixIn, socketserver.UnixStreamServer):
    daemon_threads = True

    def __init__(self, socket_path, upstream, creds, auth, diag_path=None):
        self.upstream = upstream
        self.creds = creds
        self.auth = auth
        self.diag_path = diag_path
        super().__init__(socket_path, Handler)

    def get_request(self):
        # BaseHTTPRequestHandler expects a (host, port) client_address;
        # a unix-socket accept() returns '' - substitute a dummy pair.
        request, _ = self.socket.accept()
        return request, ("sandbox", 0)

    def handle_error(self, request, client_address):
        # The container side going away mid-response - the session
        # interrupting a streaming reply, exiting with requests still in
        # flight, or abandoning a request it no longer wants - reaches us as a
        # broken pipe or a reset while writing the rest of the body. Nothing
        # is left to deliver and nothing is wrong, but socketserver's default
        # dumps a full traceback for it, onto the same terminal the session is
        # drawing its UI on: this proxy runs on the host, sharing that
        # terminal. Stay silent for exactly that case. Every other exception
        # is a real fault and still gets its traceback.
        if isinstance(sys.exc_info()[1], (BrokenPipeError, ConnectionResetError)):
            return
        super().handle_error(request, client_address)


def main():
    p = argparse.ArgumentParser()
    p.add_argument("--socket", required=True)
    p.add_argument("--credentials", required=True)
    # No default for any of these, deliberately. A default upstream or a
    # default credential kind would be one backend's arrangement wearing the
    # name of every backend's, which is the shape #171 was filed about: the
    # caller says which backend this is, and this program has no opinion.
    p.add_argument("--upstream", required=True)
    p.add_argument("--credential-kind", required=True,
                   choices=("static", "anthropic-oauth"))
    # Which value in the credentials file is the key, for the static kind.
    # Absent means the whole file is the key.
    p.add_argument("--credential-field", default=None)
    p.add_argument("--auth-header", default="Authorization")
    # "{token}" rather than a scheme name, so a backend that wants something
    # other than "Bearer x" - a bare key, a prefix of its own - needs no code
    # here.
    p.add_argument("--auth-format", default="Bearer {token}")
    # Repeatable, and replacing rather than adding to a default: a backend
    # states what must not travel beside its credential, and inheriting
    # another backend's list would be a guess.
    p.add_argument("--strip-header", action="append", default=[],
                   metavar="NAME")
    p.add_argument("--ensure-header", action="append", default=[],
                   metavar="NAME=VALUE")
    p.add_argument("--oauth-endpoint")
    p.add_argument("--oauth-client-id")
    p.add_argument("--refresh-margin", type=int, default=300)
    # Where diagnostics go. A file rather than stderr, because stderr here is
    # the operator's terminal and the session owns it: a line written mid-run
    # lands in the middle of whatever the harness is drawing. bin/dejima reads
    # this back after the container exits, alongside the exit audit.
    #
    # Absent, they go to stderr, which is what a proxy run by hand wants.
    p.add_argument("--log", default=None,
                   help="append diagnostics here instead of stderr")
    args = p.parse_args()

    if "{token}" not in args.auth_format:
        sys.stderr.write("dejima-credential-proxy: --auth-format must contain "
                         "{token}, or the credential is never sent\n")
        return 2
    ensure = []
    for item in args.ensure_header:
        if "=" not in item:
            sys.stderr.write("dejima-credential-proxy: --ensure-header wants "
                             "NAME=VALUE, got %r\n" % item)
            return 2
        name, _, value = item.partition("=")
        if not name or not value:
            sys.stderr.write("dejima-credential-proxy: --ensure-header wants "
                             "NAME=VALUE, got %r\n" % item)
            return 2
        ensure.append((name, value))

    if args.credential_kind == "anthropic-oauth":
        if not args.oauth_endpoint or not args.oauth_client_id:
            sys.stderr.write("dejima-credential-proxy: anthropic-oauth needs "
                             "--oauth-endpoint and --oauth-client-id\n")
            return 2
        creds = OAuthCredential(args.credentials, args.oauth_endpoint,
                                args.oauth_client_id, args.refresh_margin)
    else:
        creds = StaticCredential(args.credentials, args.credential_field)

    try:
        creds.startup_check()
    except Exception as e:
        sys.stderr.write(
            "dejima-credential-proxy: unusable credential %s: %s\n"
            % (args.credentials, e))
        return 1

    auth = AuthShape(args.auth_header, args.auth_format,
                     args.strip_header, ensure)
    # In the spelling the backend declared, for headers this has to add
    # rather than replace. A client that sent its own keeps its own casing.
    SUPPLIED_NAMES[auth.header.lower()] = auth.header
    for name, _ in ensure:
        SUPPLIED_NAMES.setdefault(name.lower(), name)

    if os.path.exists(args.socket):
        os.unlink(args.socket)
    server = ProxyServer(args.socket, urlsplit(args.upstream), creds, auth,
                         diag_path=args.log)
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        pass
    return 0


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