akadmin/headroom
1
0
Fork 0
mirror of https://github.com/headroomlabs-ai/headroom.git synced 2026-08-27 14:17:10 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge 73f77da262 into 8884d87378

Browse source
This commit is contained in:
Ayush Kumar Jha 2026-08-27 05:08:25 -04:00 • committed by GitHub
commit b13ea8df2f
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
8 changed files with 859 additions and 11 deletions
Download patch file Download diff file Expand all files Collapse all files

60
docs/content/docs/proxy.mdx
View file

@ -32,6 +32,7 @@ Telemetry is **local-only and off by default**. `HEADROOM_TELEMETRY=on` (or `--t
|--------|---------|-------------|
| `--host` | `127.0.0.1` | Host to bind to |
| `--port` | `8787` | Port to bind to |
| `--uds` | None | Serve on a Unix domain socket at this path instead of `--host`/`--port`. POSIX only — Windows has no `AF_UNIX` support in Python or asyncio. See [Serving on a Unix socket](#serving-on-a-unix-socket) |
| `--workers` | `1` | Number of Uvicorn worker processes |
| `--limit-concurrency` | `1000` | Maximum concurrent connections before Uvicorn returns 503 |
| `--max-connections` | `500` | Maximum upstream HTTP connections |
@ -193,6 +194,65 @@ HEADROOM_SAVINGS_PROFILE=balanced HEADROOM_TARGET_RATIO=0.15 headroom proxy
For permanent custom profiles, see the profile definitions in `headroom/agent_savings.py`. Each profile is an `AgentSavingsProfile` dataclass with fields for compression mode, target ratio, turn protection, and pipeline toggles.
## Serving on a Unix socket
`--uds` binds an `AF_UNIX` socket instead of a TCP port. Request handling is
identical — only the transport changes.
```bash
headroom proxy --uds ~/.headroom/run/proxy.sock
```
Two reasons to prefer it over a loopback port:
- **No port to collide with, and no port exposed.** Access is governed by
filesystem permissions rather than by anything listening on the network.
- **Deployments where a port is awkward.** Containers and systemd units can share a
socket through a bind mount or a runtime directory without publishing a port.
<Callout type="warning">
A socket does **not** re-enable Claude Code's Remote Control. Setting
`ANTHROPIC_UNIX_SOCKET` passes that feature's host check but makes Claude Code
treat the session as API-key auth, which fails a separate subscription-auth
requirement — see [Why the Unix socket route does not
help](/docs/troubleshooting#why-the-unix-socket-route-does-not-help).
</Callout>
Point any Unix-socket-capable client at the path. With curl:
```bash
curl --unix-socket ~/.headroom/run/proxy.sock http://localhost/health
```
### Permissions and lifecycle
A Unix socket carries no credentials of its own, so **the mode of the directory
holding it is the access-control boundary**.
If the parent directory does not exist, Headroom creates it `0700`. If it already
exists, Headroom does not touch its permissions — something else owns that
directory's policy, and silently tightening it could lock out whatever put it
there. Instead it refuses to start when an existing parent is writable by other
users and not sticky, since anyone on the host could then replace the socket.
(A sticky directory such as `/tmp` is accepted: others may create their own
entries there but cannot unlink or rename yours.)
On startup Headroom removes a stale socket left behind by a crashed process, but
it refuses to start when:
- something is still listening on the path (two proxies would silently split traffic),
- the path exists and is not a socket (far more likely a typo'd argument pointing
at real data than a leftover), or
- the path exceeds the platform's `sun_path` limit — 108 bytes on Linux, 104 on
macOS and the BSDs. Keep it short; a long path fails inside `bind()` with an
error that names nothing.
<Callout type="warn">
`--uds` is POSIX-only. On Windows the command exits immediately with an error
telling you to use `--port`; Python exposes no `socket.AF_UNIX` there and
asyncio has no Windows UDS transport.
</Callout>
## Configuration in depth
Proxy behavior is set by three layers, **each overriding the one before**:

43
docs/content/docs/troubleshooting.mdx
View file

@ -215,11 +215,48 @@ See [issue #746](https://github.com/headroomlabs-ai/headroom/issues/746) for the
## Remote Control unavailable through custom ANTHROPIC_BASE_URL
**Symptom**: When Claude Code runs with `ANTHROPIC_BASE_URL` set to a custom host (for example, Headroom), the Remote Control menu is absent.
**Symptom**: When Claude Code runs with `ANTHROPIC_BASE_URL` set to a custom host (for example, Headroom), the Remote Control menu is absent. Invoking `/remote-control` reports `Remote Control is only available when using Claude via api.anthropic.com`.
**Cause**: This is a Claude-side gate. Headroom only receives normal API traffic and can still compress it, but Claude evaluates Remote Control availability before proxy traffic reaches the server.
**Cause**: A client-side gate, evaluated before any traffic reaches Headroom. Claude Code v2.1.196+ parses `ANTHROPIC_BASE_URL` and compares the host against `api.anthropic.com` exactly, so a loopback address fails it. The check is satisfied by any one of:
**Fix**: Use Headroom for normal proxied API sessions, and launch Claude directly (without `ANTHROPIC_BASE_URL`) when you need Claude Remote Control.
- `ANTHROPIC_BASE_URL` unset, or naming `api.anthropic.com` as its host, or
- `ANTHROPIC_UNIX_SOCKET` pointing at a Unix domain socket — but see [below](#why-the-unix-socket-route-does-not-help): it passes this check and fails a different one.
Note that Remote Control does **not** honor `_CLAUDE_CODE_ASSUME_FIRST_PARTY_BASE_URL`, the flag that re-enables the 1M window and on-demand tool loading — it reads the base URL directly. Headroom cannot re-enable Remote Control from the server side the way `--1m` restores the `context-1m` beta header, because the eligibility decision and the bridge connection both happen in the client.
**Fix**: Launch Claude directly, without `ANTHROPIC_BASE_URL`, for sessions that need Remote Control, and use Headroom for the rest.
### Why the Unix socket route does not help
`ANTHROPIC_UNIX_SOCKET` satisfies the host check above, so it looks like a way to keep a first-party base URL while routing through Headroom. It does not work, and the reason is worth recording so nobody re-derives it.
That variable is reserved for `claude ssh`, where the process on the other end of the socket supplies credentials. Claude Code 2.1.198 carries the diagnostic verbatim:
```
ANTHROPIC_UNIX_SOCKET is set (claude ssh remote), and the local proxy is API-key-authed.
```
Setting it therefore makes Claude Code classify the session as API-key auth. Remote Control separately requires *claude.ai subscription* auth — the binary's eligibility messages include `Remote Control requires claude.ai subscription auth.` — so the one variable opens the host gate and closes the subscription gate at the same time.
Confirmed on Linux against Claude Code 2.1.198 with `headroom proxy --uds` and no API key set. The session starts with:
```
⚠ claude.ai connectors are disabled because ANTHROPIC_API_KEY or another auth source is set
and takes precedence over your claude.ai login
```
sends `Not logged in · Please run /login`, and after a successful subscription login still fails inside the SDK, because nothing ever attached a credential:
```
API invalid_api_key: Could not resolve authentication method. Expected one of apiKey,
authToken, credentials, config, or profile to be set.
```
No proxy-side change can repair this: the eligibility decision happens in the client before any request is made.
**What remains**: a transport that leaves `ANTHROPIC_BASE_URL` unset and does not set `ANTHROPIC_UNIX_SOCKET` — an `HTTPS_PROXY` CONNECT/MITM front end with a locally-generated CA trusted through `NODE_EXTRA_CA_CERTS`. That keeps the host genuinely `api.anthropic.com` and leaves subscription auth untouched. It is tracked on [issue #1779](https://github.com/headroomlabs-ai/headroom/issues/1779); `headroom proxy --uds` is still useful as a transport, just not for this.
The upstream request to allow loopback proxies is tracked at [anthropics/claude-code#76653](https://github.com/anthropics/claude-code/issues/76653).
`ENABLE_TOOL_SEARCH` is unaffected and can stay enabled for context-window savings while routing through Headroom.

45
headroom/cli/proxy.py
View file

@ -228,6 +228,17 @@ def dashboard(port: int, no_open: bool) -> None:
envvar="HEADROOM_HOST",
help="Host to bind to (default: 127.0.0.1, env: HEADROOM_HOST)",
)
@click.option(
"--uds",
default=None,
envvar="HEADROOM_UDS",
metavar="PATH",
help=(
"Serve on a Unix domain socket instead of --host/--port. POSIX only. "
"Lets a client keep a first-party base URL while its traffic still "
"reaches Headroom (env: HEADROOM_UDS)."
),
)
@click.option(
"--port",
"-p",
@ -1014,6 +1025,7 @@ def proxy(
mode: str | None,
target_ratio: float | None,
host: str,
uds: str | None,
port: int,
workers: int,
limit_concurrency: int,
@ -1116,6 +1128,17 @@ def proxy(
OPENAI_BASE_URL=http://localhost:8787/v1 your-app
"""
_reexec_with_malloc_tuning()
# Fail before any dependency loading or config work: an unusable --uds is a
# typo or an unsupported platform, and both are cheaper to report up front.
if uds:
from headroom.proxy.uds import UdsError, require_uds_support
try:
require_uds_support()
except UdsError as exc:
raise click.ClickException(str(exc)) from exc
ensure_proxy_dependencies()
# Import here to avoid slow startup
@ -1296,6 +1319,7 @@ def proxy(
config = ProxyConfig(
host=host,
port=port,
uds=uds,
rollout=rollout_snapshot,
anthropic_api_url=provider_api_overrides.anthropic,
anthropic_extra_headers=resolved_anthropic_extra_headers,
@ -1612,6 +1636,22 @@ Memory (Multi-Provider):
else:
tuning_section = ""
# A socket has no URL, and no per-agent recipe belongs here — see
# uds.socket_usage_lines() for why the banner stays transport-neutral.
if config.uds:
from headroom.proxy.uds import socket_usage_lines
listen_display = f"unix:{config.uds}"
usage_section = "\n".join(socket_usage_lines(config.uds))
else:
listen_display = f"http://{config.host}:{config.port}"
usage_section = "\n".join(
(
f" Claude Code: ANTHROPIC_BASE_URL=http://{config.host}:{config.port} claude",
f" Codex / OpenAI: OPENAI_BASE_URL=http://{config.host}:{config.port}/v1 your-app",
)
)
click.echo(f"""
╔═══════════════════════════════════════════════════════════════════════╗
║ HEADROOM PROXY ║
@ -1620,7 +1660,7 @@ Memory (Multi-Provider):
Starting proxy server...
URL: http://{config.host}:{config.port}
URL: {listen_display}
Mode: {config.mode}
Optimization: {"ENABLED" if config.optimize else "DISABLED"}
Caching: {"ENABLED" if config.cache_enabled else "DISABLED"}
@ -1641,8 +1681,7 @@ Routing:
/v1/projects/.../publishers/... → {vertex_url}
Usage:
Claude Code: ANTHROPIC_BASE_URL=http://{config.host}:{config.port} claude
Codex / OpenAI: OPENAI_BASE_URL=http://{config.host}:{config.port}/v1 your-app
{usage_section}
{memory_section}
Endpoints:
GET /livez Process liveness

3
headroom/proxy/models.py
View file

@ -140,6 +140,9 @@ class ProxyConfig:
# Server
host: str = "127.0.0.1"
port: int = 8787
# Serve on this Unix domain socket instead of host:port. POSIX only; see
# headroom/proxy/uds.py for why a socket transport exists at all (GH #1779).
uds: str | None = None
# Resolved at this configuration boundary and then injected unchanged.
rollout: RolloutSnapshot | None = None
anthropic_api_url: str | None = None # Custom Anthropic API URL override

57
headroom/proxy/server.py
View file

@ -5496,13 +5496,23 @@ def run_server(
# Resolve upstream API targets for display in the banner (#583).
api_targets = resolve_api_targets(config.provider_api_overrides)
if config.uds:
# No per-agent recipe on a socket bind; see uds.socket_usage_lines().
listen_display = f"unix:{config.uds}"
usage_label = "Client: "
usage_display = "must support HTTP over a Unix socket natively"
else:
listen_display = f"http://{config.host}:{config.port}"
usage_label = "Claude Code:"
usage_display = f"ANTHROPIC_BASE_URL=http://{config.host}:{config.port} claude"
if print_banner:
print(f"""
╔══════════════════════════════════════════════════════════════════════╗
║ HEADROOM PROXY SERVER ║
╠══════════════════════════════════════════════════════════════════════╣
║ Version: 1.0.0 ║
║ Listening: http://{config.host}:{config.port:<5} ║
║ Listening: {listen_display:<57}║
║ Workers: {workers:<3} Concurrency Limit: {limit_concurrency:<5} ║
║ Backend: {backend_status:<59}║
╠══════════════════════════════════════════════════════════════════════╣
@ -5524,7 +5534,7 @@ def run_server(
║ Conn Pool: {pool_info:<52}║
╠══════════════════════════════════════════════════════════════════════╣
║ USAGE: ║
║ Claude Code: ANTHROPIC_BASE_URL=http://{config.host}:{config.port} claude ║
║ {usage_label} {usage_display:<51}║
║ Cursor: Set base URL in settings ║
╠══════════════════════════════════════════════════════════════════════╣
║ ENDPOINTS: ║
@ -5603,11 +5613,48 @@ def run_server(
# and no CLI flag to change it. Overridable now; the default is unchanged.
uvicorn_log_level = _resolve_uvicorn_log_level()
# Bind target: a Unix socket when one is configured, otherwise host:port.
# uvicorn treats `uds` and `host`/`port` as alternatives, so they are built
# here rather than passed together.
bind_kwargs: dict[str, Any]
uds_path: Path | None = None
if config.uds:
from headroom.proxy.uds import prepare_uds_path
uds_path = prepare_uds_path(config.uds)
bind_kwargs = {"uds": str(uds_path)}
else:
bind_kwargs = {"host": config.host, "port": config.port}
try:
_run_uvicorn(
app_target,
bind_kwargs,
workers,
limit_concurrency,
uvicorn_log_level,
uvicorn_kwargs,
)
finally:
if uds_path is not None:
from headroom.proxy.uds import remove_uds_path
remove_uds_path(uds_path)
def _run_uvicorn(
app_target: Any,
bind_kwargs: dict[str, Any],
workers: int,
limit_concurrency: int,
log_level: str,
uvicorn_kwargs: dict[str, Any],
) -> None:
"""Hand off to uvicorn. Split out so the bind target stays testable."""
uvicorn.run(
app_target,
host=config.host,
port=config.port,
log_level=uvicorn_log_level,
**bind_kwargs,
log_level=log_level,
workers=workers if workers > 1 else None, # None = single process (default)
limit_concurrency=limit_concurrency,
# Defense-in-depth: the loopback guard for /debug/* endpoints trusts

250
headroom/proxy/uds.py Normal file
View file

@ -0,0 +1,250 @@
"""Unix-domain-socket transport for the Headroom proxy.
``headroom proxy --uds PATH`` serves the same ASGI app over an ``AF_UNIX``
socket instead of a TCP port. Nothing about request handling changes — this is
purely the transport shell.
Why a socket at all, when a loopback port already works: no port to collide
with, nothing listening on the network, and access governed by filesystem
permissions rather than by anything reachable over TCP. That suits container
and systemd deployments, and any client that can dial an ``AF_UNIX`` path.
Note for anyone arriving from GH #1779: this does **not** restore Claude Code's
Remote Control. ``ANTHROPIC_UNIX_SOCKET`` does satisfy that feature's
``api.anthropic.com`` host check, but it is reserved for ``claude ssh``, where
the process on the other end of the socket supplies credentials. Setting it
makes Claude Code classify the session as API-key auth, and Remote Control
separately requires claude.ai subscription auth — so the same variable opens one
gate and closes the other. Verified on Claude Code 2.1.198; see
``docs/content/docs/troubleshooting.mdx``.
Access control is filesystem permissions. A socket inherits no credentials of
its own, so the mode of the directory holding it *is* the security boundary.
A parent this module creates is made ``0700``; a parent that already exists is
never modified, only checked, since something else owns its policy.
"""
from __future__ import annotations
import errno
import os
import socket
import stat
import sys
from pathlib import Path
__all__ = [
"UDS_SUPPORTED",
"UdsError",
"socket_usage_lines",
"max_uds_path_length",
"prepare_uds_path",
"remove_uds_path",
"require_uds_support",
]
# ``AF_UNIX`` is absent from CPython on Windows even where the OS supports the
# address family, and asyncio has no Windows UDS transport either. Reading the
# constant through getattr keeps this module importable — and type-checkable —
# on Windows, where every call site is already behind UDS_SUPPORTED.
_AF_UNIX: int = getattr(socket, "AF_UNIX", -1)
#: The single capability check the rest of the module keys off.
UDS_SUPPORTED = _AF_UNIX != -1
# ``sockaddr_un.sun_path`` is a fixed-size buffer: 108 bytes on Linux, 104 on
# the BSDs and macOS. Overrunning it fails inside bind() with a bare ENAMETOOLONG
# that says nothing about which path was too long, so check it up front.
_SUN_PATH_MAX_LINUX = 108
_SUN_PATH_MAX_BSD = 104
class UdsError(RuntimeError):
"""A Unix socket path cannot be used for the reason described."""
def max_uds_path_length(platform: str | None = None) -> int:
"""Longest usable socket path, including the trailing NUL, for *platform*."""
plat = sys.platform if platform is None else platform
if plat.startswith("linux"):
return _SUN_PATH_MAX_LINUX
return _SUN_PATH_MAX_BSD
def require_uds_support(platform: str | None = None) -> None:
"""Raise :class:`UdsError` when this interpreter cannot serve on a socket."""
plat = sys.platform if platform is None else platform
if plat == "win32" or not UDS_SUPPORTED:
raise UdsError(
"--uds needs Unix domain sockets, which are unavailable on this "
"platform (Python has no socket.AF_UNIX and asyncio has no Windows "
"UDS transport). Use --port instead."
)
def _is_live_socket(path: Path) -> bool:
"""True when something is already accepting connections on *path*.
A leftover socket file from a crashed proxy looks identical to a live one
on disk; the only way to tell them apart is to try connecting.
"""
sock = socket.socket(_AF_UNIX, socket.SOCK_STREAM)
try:
sock.settimeout(0.5)
sock.connect(str(path))
except OSError as exc:
# ECONNREFUSED: nothing is listening, the inode is stale.
# ENOENT: it vanished between the stat and the connect.
if exc.errno in (errno.ECONNREFUSED, errno.ENOENT):
return False
# EACCES, ETIMEDOUT, anything else: something is there, or we cannot
# tell. Either way, refuse to unlink it.
return True
else:
return True
finally:
sock.close()
# Docs page carrying the client-compatibility detail the banner has no room for.
UDS_DOCS_URL = "https://headroom-docs.vercel.app/docs/proxy#serving-on-a-unix-socket"
def socket_usage_lines(path: str | os.PathLike[str]) -> tuple[str, ...]:
"""How the startup banner describes a socket bind, for every banner.
Deliberately names no agent and hands out no environment variables. An
earlier revision printed an ``ANTHROPIC_UNIX_SOCKET=... claude`` recipe here,
which is a configuration that does not work: it satisfies Claude Code's
``api.anthropic.com`` host check but reclassifies the session as API-key
auth, and the session then fails to authenticate. Printing it at startup
turned a known-negative result into first-party guidance. The rule this
encodes is that the banner states a transport requirement and points at the
docs; per-client wiring belongs in the docs, where it can be qualified.
"""
return (
f" Socket: {path}",
" Client: must support HTTP over a Unix socket natively",
f" Example: curl --unix-socket {path} http://localhost/health",
f" Details: {UDS_DOCS_URL}",
)
def _missing_ancestors(target: Path) -> list[Path]:
"""Directories along *target* that do not exist yet, shallowest first."""
missing: list[Path] = []
node = target
while not node.exists():
missing.append(node)
if node.parent == node: # reached the filesystem root
break
node = node.parent
return list(reversed(missing))
def _require_safe_existing_parent(parent: Path) -> None:
"""Reject a pre-existing parent that lets other users swap the socket.
Deliberately does not repair the mode. The directory predates this call, so
something else owns its policy — silently tightening a shared directory
would lock out whatever put it there.
Group/world-writable is tolerated when the sticky bit is set, which is the
``/tmp`` case: others may create their own entries but cannot unlink or
rename ours, so the socket cannot be swapped out from under us.
"""
try:
mode = parent.stat().st_mode
except OSError:
return # unreadable; bind() will produce the authoritative error
if not mode & (stat.S_IWGRP | stat.S_IWOTH):
return
if mode & stat.S_ISVTX:
return
raise UdsError(
f"{parent} is writable by other users and not sticky, so anyone on this "
f"host could replace the socket inside it (mode {stat.S_IMODE(mode):04o}). "
"Point --uds at a directory only you can write, or chmod this one to 0700. "
"Headroom will not change the permissions of a directory it did not create."
)
def _prepare_parent_dir(parent: Path) -> None:
"""Create *parent* ``0700`` if absent; otherwise validate without mutating."""
to_create = _missing_ancestors(parent)
if not to_create:
_require_safe_existing_parent(parent)
return
parent.mkdir(parents=True, exist_ok=True)
for created in to_create:
# mkdir's mode is masked by the process umask, so set it explicitly --
# but only on the directories this call brought into existence.
try:
created.chmod(0o700)
except OSError:
pass
def prepare_uds_path(path: str | os.PathLike[str], *, platform: str | None = None) -> Path:
"""Validate *path*, create its parent ``0700``, and clear a stale socket.
Returns the resolved path, ready to hand to uvicorn's ``uds=``.
The parent is created ``0700`` when it does not exist. An existing parent is
left exactly as it is -- see :func:`_require_safe_existing_parent`.
Raises :class:`UdsError` when the platform has no Unix sockets, the path is
too long for ``sun_path``, an existing parent directory is writable by other
users, something is already listening there, or the path exists as anything
other than a socket. That last case matters: a regular file at the target is
far more likely to be a typo'd argument pointing at real data than a
leftover, so it is never removed.
"""
require_uds_support(platform)
resolved = Path(path).expanduser()
if not resolved.is_absolute():
resolved = (Path.cwd() / resolved).resolve()
limit = max_uds_path_length(platform)
encoded = len(str(resolved).encode("utf-8")) + 1 # + trailing NUL
if encoded > limit:
raise UdsError(
f"Socket path is {encoded} bytes, over this platform's {limit}-byte "
f"sun_path limit: {resolved}. Use a shorter path, e.g. under $TMPDIR."
)
_prepare_parent_dir(resolved.parent)
if resolved.exists() or resolved.is_symlink():
mode = resolved.lstat().st_mode
if not stat.S_ISSOCK(mode):
raise UdsError(
f"Refusing to replace {resolved}: it exists and is not a socket. "
"Point --uds at a path Headroom owns."
)
if _is_live_socket(resolved):
raise UdsError(
f"Another process is already listening on {resolved}. Stop it, or "
"choose a different --uds path."
)
resolved.unlink()
return resolved
def remove_uds_path(path: str | os.PathLike[str]) -> None:
"""Unlink *path* if it is still a socket. Never raises.
uvicorn removes its own socket on a clean shutdown; this covers the paths
where it does not get the chance.
"""
try:
target = Path(path)
if target.is_socket():
target.unlink()
except OSError:
pass

15
headroom/settings_store.py
View file

@ -260,6 +260,21 @@ SETTINGS: tuple[SettingField, ...] = (
help="Bind port. Managed by the install manifest on docker/service installs.",
tier="advanced",
),
SettingField(
"HEADROOM_UDS",
"uds",
"Unix socket path",
"Networking",
"str",
default=None,
manifest_managed=True,
help=(
"Serve on a Unix domain socket at this path instead of host/port. "
"POSIX only; leave empty to bind host/port. Managed by the install "
"manifest on docker/service installs."
),
tier="advanced",
),
# --- Logging ---
SettingField(
"HEADROOM_LOG_MESSAGES",

397
tests/test_proxy_uds.py Normal file
View file

@ -0,0 +1,397 @@
"""Tests for serving the proxy on a Unix domain socket (`headroom proxy --uds`).
The socket transport is a plain alternative to a TCP port — see
`headroom/proxy/uds.py` for the rationale and for why it does not restore
Claude Code's Remote Control (GH #1779).
"""
from __future__ import annotations
import socket
import stat
from pathlib import Path
from unittest.mock import patch
import pytest
from click.testing import CliRunner
from headroom.cli.proxy import proxy as proxy_cmd
from headroom.proxy.uds import (
UDS_SUPPORTED,
UdsError,
_missing_ancestors,
_require_safe_existing_parent,
max_uds_path_length,
prepare_uds_path,
remove_uds_path,
require_uds_support,
socket_usage_lines,
)
requires_uds = pytest.mark.skipif(
not UDS_SUPPORTED, reason="platform has no socket.AF_UNIX (Windows)"
)
try: # `headroom.proxy.server` pulls in the compiled Rust core.
import headroom._core # noqa: F401
_CORE_BUILT = True
except ImportError: # pragma: no cover - depends on the local build
_CORE_BUILT = False
requires_core = pytest.mark.skipif(
not _CORE_BUILT, reason="headroom._core is not built in this environment"
)
# --------------------------------------------------------------------------
# Platform capability — runs everywhere, since the platform is a parameter.
# --------------------------------------------------------------------------
def test_require_uds_support_rejects_windows() -> None:
"""Windows has neither socket.AF_UNIX nor an asyncio UDS transport."""
with pytest.raises(UdsError, match="unavailable on this platform"):
require_uds_support(platform="win32")
def test_require_uds_support_accepts_posix() -> None:
if not UDS_SUPPORTED:
pytest.skip("AF_UNIX missing; the platform argument cannot override that")
require_uds_support(platform="linux")
def test_sun_path_limit_is_platform_specific() -> None:
"""Linux allows 108 bytes, the BSDs and macOS 104. Guessing high truncates."""
assert max_uds_path_length("linux") == 108
assert max_uds_path_length("darwin") == 104
def test_cli_rejects_uds_on_windows() -> None:
"""The CLI fails fast with a readable error, not a bind-time OSError."""
with patch("headroom.proxy.uds.UDS_SUPPORTED", False):
result = CliRunner().invoke(proxy_cmd, ["--uds", "/tmp/headroom-test.sock"])
assert result.exit_code != 0
assert "Unix domain sockets" in result.output
assert "--port instead" in result.output
# --------------------------------------------------------------------------
# Parent-directory policy — pure logic, so it runs on every platform.
# --------------------------------------------------------------------------
def test_missing_ancestors_lists_only_absent_levels(tmp_path: Path) -> None:
"""Only these get chmod 0700; anything already on disk is left alone."""
existing = tmp_path / "existing"
existing.mkdir()
missing = _missing_ancestors(existing / "a" / "b")
assert missing == [existing / "a", existing / "a" / "b"]
def test_missing_ancestors_is_empty_for_an_existing_dir(tmp_path: Path) -> None:
assert _missing_ancestors(tmp_path) == []
class _FakeStat:
def __init__(self, mode: int) -> None:
self.st_mode = mode
@pytest.mark.parametrize(
("mode", "accepted"),
[
(0o700, True), # owner only
(0o750, True), # group may read/traverse, not write
(0o755, True), # the common shared-parent case
(0o770, False), # any group member could swap the socket
(0o777, False), # any local user could
(0o1777, True), # /tmp: sticky, so others cannot unlink ours
(0o1770, True), # sticky group-writable
],
)
def test_existing_parent_accepted_only_when_others_cannot_swap_the_socket(
tmp_path: Path, mode: int, accepted: bool
) -> None:
"""Windows chmod is a no-op, so the mode is injected rather than applied."""
with patch.object(Path, "stat", return_value=_FakeStat(stat.S_IFDIR | mode)):
if accepted:
_require_safe_existing_parent(tmp_path)
else:
with pytest.raises(UdsError, match="writable by other users"):
_require_safe_existing_parent(tmp_path)
def test_unreadable_existing_parent_defers_to_bind(tmp_path: Path) -> None:
"""A stat we cannot perform is not evidence of a problem; let bind() rule."""
with patch.object(Path, "stat", side_effect=PermissionError):
_require_safe_existing_parent(tmp_path)
# --------------------------------------------------------------------------
# Startup banner — a socket bind must not advertise a broken recipe.
# --------------------------------------------------------------------------
def test_socket_usage_lines_omit_the_unsupported_claude_code_recipe() -> None:
"""Regression: the banner once printed a configuration that cannot work.
`ANTHROPIC_UNIX_SOCKET=... claude` passes Claude Code's api.anthropic.com
host check but reclassifies the session as API-key auth, and the session
then fails to authenticate. Printing it at startup turned a known-negative
field result into first-party runtime guidance.
"""
rendered = "\n".join(socket_usage_lines("/run/headroom/proxy.sock"))
assert "ANTHROPIC_UNIX_SOCKET" not in rendered
assert "ANTHROPIC_BASE_URL" not in rendered
assert "claude" not in rendered.lower()
def test_socket_usage_lines_state_the_transport_requirement() -> None:
"""What replaces the recipe has to be useful, not merely absent."""
path = "/run/headroom/proxy.sock"
rendered = "\n".join(socket_usage_lines(path))
assert path in rendered
assert "HTTP over a Unix socket" in rendered
assert "curl --unix-socket" in rendered
assert "serving-on-a-unix-socket" in rendered
def test_socket_usage_lines_name_no_agent() -> None:
"""Transport-neutral: the banner singles out no client."""
rendered = "\n".join(socket_usage_lines("/run/headroom/proxy.sock")).lower()
for agent in ("claude", "codex", "opencode", "cursor", "aider", "copilot"):
assert agent not in rendered, f"banner should not name {agent}"
# --------------------------------------------------------------------------
# Path preparation — needs a real AF_UNIX platform.
# --------------------------------------------------------------------------
@requires_uds
def test_prepare_creates_parent_owner_only(tmp_path: Path) -> None:
"""The directory mode is the access-control boundary for the socket."""
target = tmp_path / "run" / "headroom.sock"
resolved = prepare_uds_path(target)
assert resolved == target
assert target.parent.is_dir()
assert stat.S_IMODE(target.parent.stat().st_mode) == 0o700
@requires_uds
def test_prepare_preserves_an_existing_parents_mode(tmp_path: Path) -> None:
"""A caller-owned directory must not be silently tightened to 0700.
Regression test: `--uds /run/shared/hr.sock` where `/run/shared` is a
directory someone else set up at 0755 would have locked out every other
user of that directory.
"""
parent = tmp_path / "shared"
parent.mkdir()
parent.chmod(0o755)
bystander = parent / "someone-elses.txt"
bystander.write_text("theirs", encoding="utf-8")
prepare_uds_path(parent / "headroom.sock")
assert stat.S_IMODE(parent.stat().st_mode) == 0o755
assert bystander.read_text(encoding="utf-8") == "theirs"
@requires_uds
def test_prepare_only_chmods_directories_it_creates(tmp_path: Path) -> None:
"""The 0700 applies to the new levels, not to the existing root above them."""
root = tmp_path / "existing"
root.mkdir()
root.chmod(0o755)
prepare_uds_path(root / "a" / "b" / "headroom.sock")
assert stat.S_IMODE(root.stat().st_mode) == 0o755
assert stat.S_IMODE((root / "a").stat().st_mode) == 0o700
assert stat.S_IMODE((root / "a" / "b").stat().st_mode) == 0o700
@requires_uds
def test_prepare_refuses_a_world_writable_existing_parent(tmp_path: Path) -> None:
"""Without the sticky bit, any local user could swap the socket out."""
parent = tmp_path / "open"
parent.mkdir()
parent.chmod(0o777)
with pytest.raises(UdsError, match="writable by other users"):
prepare_uds_path(parent / "headroom.sock")
assert stat.S_IMODE(parent.stat().st_mode) == 0o777, "the refusal must not mutate"
@requires_uds
def test_prepare_accepts_a_sticky_world_writable_parent(tmp_path: Path) -> None:
"""`/tmp` is 1777: others can add entries but cannot unlink ours."""
parent = tmp_path / "sticky"
parent.mkdir()
parent.chmod(0o1777)
resolved = prepare_uds_path(parent / "headroom.sock")
assert resolved.parent == parent
assert stat.S_IMODE(parent.stat().st_mode) == 0o1777
@requires_uds
def test_prepare_clears_a_stale_socket(tmp_path: Path) -> None:
"""A crashed proxy leaves an inode behind; a restart must not trip on it."""
target = tmp_path / "stale.sock"
dead = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
dead.bind(str(target))
dead.close() # closing without unlinking is exactly the crash case
assert target.exists()
prepare_uds_path(target)
assert not target.exists()
@requires_uds
def test_prepare_refuses_a_live_socket(tmp_path: Path) -> None:
"""Two proxies on one socket would silently steal each other's traffic."""
target = tmp_path / "live.sock"
live = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
live.bind(str(target))
live.listen(1)
try:
with pytest.raises(UdsError, match="already listening"):
prepare_uds_path(target)
assert target.exists(), "the live socket must survive the refusal"
finally:
live.close()
target.unlink(missing_ok=True)
@requires_uds
def test_prepare_never_deletes_a_regular_file(tmp_path: Path) -> None:
"""A typo'd --uds pointing at real data must not destroy it."""
target = tmp_path / "notes.txt"
target.write_text("important", encoding="utf-8")
with pytest.raises(UdsError, match="is not a socket"):
prepare_uds_path(target)
assert target.read_text(encoding="utf-8") == "important"
@requires_uds
def test_prepare_rejects_an_oversized_path(tmp_path: Path) -> None:
"""Past sun_path, bind() fails with an ENAMETOOLONG that names nothing."""
target = tmp_path / ("d" * 120) / "headroom.sock"
with pytest.raises(UdsError, match="sun_path limit"):
prepare_uds_path(target)
@requires_uds
def test_remove_uds_path_is_socket_only(tmp_path: Path) -> None:
"""Cleanup runs in a finally block, so it must be narrow and never raise."""
sock_path = tmp_path / "gone.sock"
sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
sock.bind(str(sock_path))
sock.close()
regular = tmp_path / "keep.txt"
regular.write_text("keep", encoding="utf-8")
remove_uds_path(sock_path)
remove_uds_path(regular)
remove_uds_path(tmp_path / "does-not-exist.sock")
assert not sock_path.exists()
assert regular.exists()
# --------------------------------------------------------------------------
# Server wiring — uvicorn is mocked, so this runs on every platform.
# --------------------------------------------------------------------------
def _bind_kwargs_for(**config_kwargs: object) -> dict[str, object]:
"""Run run_server far enough to capture what it would bind to."""
from headroom.proxy.server import ProxyConfig, run_server
captured: dict[str, object] = {}
def fake_run_uvicorn( # noqa: ANN202
app_target, # noqa: ANN001
bind_kwargs, # noqa: ANN001
workers, # noqa: ANN001
limit_concurrency, # noqa: ANN001
log_level, # noqa: ANN001
uvicorn_kwargs, # noqa: ANN001
):
captured.update(bind_kwargs)
with (
patch("headroom.proxy.server._run_uvicorn", side_effect=fake_run_uvicorn),
patch("headroom.proxy.server.create_app"),
):
run_server(ProxyConfig(**config_kwargs), print_banner=False) # type: ignore[arg-type]
return captured
@requires_core
def test_run_server_binds_host_and_port_by_default() -> None:
bind = _bind_kwargs_for(host="127.0.0.1", port=9123)
assert bind == {"host": "127.0.0.1", "port": 9123}
@requires_uds
@requires_core
def test_run_server_binds_the_socket_instead_of_a_port(tmp_path: Path) -> None:
"""uvicorn treats uds and host/port as alternatives; passing both is an error."""
target = tmp_path / "headroom.sock"
bind = _bind_kwargs_for(host="127.0.0.1", port=9123, uds=str(target))
assert bind == {"uds": str(target)}
assert "host" not in bind and "port" not in bind
@requires_uds
@requires_core
def test_run_server_removes_the_socket_on_exit(tmp_path: Path) -> None:
"""A crash inside uvicorn must not leave an inode that blocks the restart."""
from headroom.proxy.server import ProxyConfig, run_server
target = tmp_path / "headroom.sock"
def bind_then_fail( # noqa: ANN202
app_target, # noqa: ANN001
bind_kwargs, # noqa: ANN001
workers, # noqa: ANN001
limit_concurrency, # noqa: ANN001
log_level, # noqa: ANN001
uvicorn_kwargs, # noqa: ANN001
):
sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
sock.bind(bind_kwargs["uds"])
sock.close()
raise KeyboardInterrupt
with (
patch("headroom.proxy.server._run_uvicorn", side_effect=bind_then_fail),
patch("headroom.proxy.server.create_app"),
pytest.raises(KeyboardInterrupt),
):
run_server(ProxyConfig(uds=str(target)), print_banner=False)
assert not target.exists()