Phase 0: scaffold microsandbox-based rewrite

Start a from-scratch rewrite of agent-vm on top of microsandbox. The original
Bash/Python tree is removed on this branch and continues to live on main until
the rewrite reaches v1.

This phase only sets up the build:

- PLAN.md and ARCHITECTURE.md document the phased roadmap and the
  design-decision log; ARCHITECTURE grows after each phase.
- microsandbox added as a git submodule at vendor/microsandbox so the Phase 3
  SecretValue::File extension can branch the fork in place.
- Cargo workspace at the root with crates/agent-vm/ as the binary crate.
- Hello-world main.rs that boots an alpine sandbox via the microsandbox SDK
  to prove the wiring is correct end-to-end (cargo check passes).
This commit is contained in:
Evgeny Boger 2026-05-17 17:05:07 +03:00
parent ae6d450f17
commit cb4be40448
21 changed files with 6461 additions and 7337 deletions

7
.gitignore vendored
View file

@ -1,3 +1,8 @@
/target/
**/*.rs.bk
Cargo.lock.bak
.DS_Store
# Project-local agent hook (carries over from the original agent-vm layout).
.claude-vm.runtime.sh
github-git-proxy.log
.agent-vm.runtime.sh

3
.gitmodules vendored Normal file
View file

@ -0,0 +1,3 @@
[submodule "vendor/microsandbox"]
path = vendor/microsandbox
url = https://github.com/wirenboard/microsandbox.git

54
ARCHITECTURE.md Normal file
View file

@ -0,0 +1,54 @@
# agent-vm — ARCHITECTURE
How the rewrite is put together and *why*. Reading this top-to-bottom should
tell you what every nontrivial design choice in the codebase exists for.
Updated after each phase lands. Section per phase; subsection per major
decision.
## Phase 0 — Scaffolding
### Repository layout
```
microsandbox-rewrite/
├── PLAN.md # phased roadmap
├── ARCHITECTURE.md # this file
├── Cargo.toml # workspace
├── crates/
│ └── agent-vm/
│ ├── Cargo.toml
│ └── src/main.rs # hello-world sandbox boot
├── vendor/
│ └── microsandbox/ # git submodule, wirenboard/microsandbox
└── .gitmodules
```
### Why a Cargo workspace from day one
The binary is small today but we already know we'll need at least one
internal crate per concern (creds, image, session). A workspace lets us add
those without restructuring later, and keeps `vendor/microsandbox` out of
our crate's manifest noise.
### Why a git submodule for microsandbox (vs. crates.io, vs. path dep)
- **Phase 3 requires extending microsandbox.** The new `SecretValue::File`
variant lives in `microsandbox-network`. A path dep against a sibling
checkout works for one developer but not for CI or contributors. A submodule
pinned to a branch on our fork (`wirenboard/microsandbox`) makes the
checkout self-contained and the upstream diff reviewable.
- **`[patch]` against crates.io** also works, but it duplicates the source-of-
truth pointer (Cargo.lock + patch table) and hides the fact that we are
shipping a fork. Submodule is more explicit.
### Why depend on the path under `vendor/microsandbox` even before we fork
Phase 0 doesn't change microsandbox, but we point at the submodule path so
the build wiring we set up here is the same wiring Phase 3 uses. Avoids a
mid-rewrite refactor of `Cargo.toml`.
### Why `Sandbox::builder("hello").image("alpine")` for the smoke test
Smallest possible exercise of the SDK that proves we can talk to the runtime.
Alpine is in the microsandbox examples, downloads quickly, and exits cleanly.
No need to involve our own image (that's Phase 1).

6179
Cargo.lock generated Normal file

File diff suppressed because it is too large Load diff

17
Cargo.toml Normal file
View file

@ -0,0 +1,17 @@
[workspace]
resolver = "3"
members = ["crates/agent-vm"]
# microsandbox declares its own workspace under vendor/; keep it separate
# so Cargo doesn't try to absorb it into ours.
exclude = ["vendor/microsandbox"]
[workspace.package]
version = "0.1.0"
edition = "2024"
license = "MIT"
repository = "https://github.com/wirenboard/agent-vm"
[profile.release]
codegen-units = 1
lto = true
strip = true

141
PLAN.md Normal file
View file

@ -0,0 +1,141 @@
# agent-vm rewrite — PLAN
Living roadmap for rewriting `agent-vm` on top of
[microsandbox](https://github.com/wirenboard/microsandbox). The plan is locked
*now* but is updated as phases land. Each phase ends at a stop point so we can
inspect, adjust, then proceed.
The architecture details and design rationale live in `ARCHITECTURE.md` and are
written after each phase, not up front.
## Why a rewrite
The existing `agent-vm` (Bash, 2.4kloc + Python helpers, Lima full VMs) is
mature but heavy: 30-second cold start, 16 GB disk template, host-side `mitm`
chain, balloon daemon, custom GitHub App. microsandbox boots microVMs in
~100 ms from OCI images, has a first-class Rust SDK, and ships TLS interception
+ placeholder-substituted secrets at the network layer. Most of `agent-vm`'s
infrastructure becomes either unnecessary or moves into a small Rust binary.
## v1 scope (in)
- Subcommands: `setup`, `claude`, `codex`, `opencode`, `shell`.
- Project working directory mounted into the sandbox as `/workspace`.
- Per-project session persistence for `~/.claude/`, `~/.codex/`,
`~/.local/share/opencode/` under `${XDG_STATE_HOME}/agent-vm/<project-hash>/`.
- Host-rooted credentials with refresh: real tokens never enter the VM; host
`claude -p` / `codex exec` are used to rotate; the VM picks up the new token
on the next request without restarting the sandbox.
- Pre-baked Debian-based OCI image with the three agent CLIs and dev tools.
- Interactive attach for the agent TUIs.
## v1 scope (out, may revisit)
- GitHub App device flow + per-repo scoped tokens.
- USB passthrough.
- Dynamic memory / balloon daemon.
- Clipboard bridge.
- `ccusage` wrapper.
- Chrome DevTools MCP wiring.
- GitHub Copilot API token acquisition.
- `--mount` for extra host directories.
- `AI_HTTPS_PROXY` upstream proxy chaining.
- Apple Silicon / macOS-VZ specifics.
- WSL2-on-Windows specifics.
## Phased roadmap
Each row is one PR; we stop after each phase, fill in `ARCHITECTURE.md`, then
the user signs off on the next.
### Phase 0 — Scaffolding
- Worktree on `rewrite-microsandbox`.
- microsandbox added as a git submodule at `vendor/microsandbox` (tracking
`wirenboard/microsandbox @ main`; we'll branch off here in Phase 3).
- Cargo workspace at the worktree root; `crates/agent-vm/` binary crate.
- Hello-world `main.rs`: `Sandbox::builder("hello").image("alpine").create()`,
run `echo`, stop.
- `cargo check -p agent-vm` succeeds (runtime exec needs KVM and is out of
scope for the lint).
**Done when:** scaffold compiles, PLAN and ARCHITECTURE files exist, submodule
is registered.
### Phase 1 — OCI image
- `images/Dockerfile`: Debian 13 base + `git curl wget jq build-essential
python3 python3-pip nodejs(22) ripgrep fd-find docker-cli` + the three agent
CLIs (`@anthropic-ai/claude-code`, `opencode-ai`, `@openai/codex`).
- `images/build.sh` builds locally and tags `agent-vm:latest`. No registry
push in v1.
- `agent-vm setup` subcommand wraps `images/build.sh`.
**Done when:** `agent-vm setup` builds the image and `msb run agent-vm:latest
-- claude --version` succeeds.
### Phase 2 — Launcher MVP
- clap-based subcommand parser: `setup | claude | codex | opencode | shell`.
- Project hash + state dir helper (`${XDG_STATE_HOME:-~/.local/state}/agent-vm
/<hash>/`).
- Mount `cwd` at `/workspace` inside the sandbox.
- Symlink session dirs from the persisted state dir into the guest home.
- `attach_shell` for interactive agents.
- Credentials: env-var only (`ANTHROPIC_API_KEY`, `OPENAI_API_KEY`) so we can
test the launcher end-to-end without the refresh machinery.
**Done when:** `cd repo && agent-vm claude -p "say hi"` returns a real Claude
response from inside the sandbox.
### Phase 3 — Static host-rooted secrets
- Branch `vendor/microsandbox` to add a `SecretValue::File(PathBuf)` variant
alongside the existing `String` value. The TLS-intercept proxy re-reads the
file on every substitution.
- `agent-vm` on startup snapshots host `~/.claude/.credentials.json` and
`~/.codex/auth.json` into `<state>/tokens/{anthropic,openai}.token` files.
- Register file-backed secret entries for `api.anthropic.com`,
`api.openai.com`, `chatgpt.com`, `platform.claude.com`, etc.
- Cross-instance lock so two `agent-vm` processes don't fight over the token
files.
**Done when:** inside the guest, `cat /proc/$$/environ | tr '\0' '\n' | grep
ANTHROPIC` shows only placeholders, while a real Claude request succeeds.
### Phase 4 — Refresh semantics
- inotify (Linux) / kqueue (macOS) watcher on host creds files: when host
Claude/Codex rotates tokens, we re-snapshot to the file microsandbox watches.
- Proactive expiry watch: when the access token is < 5 minutes from expiry
and no host activity has refreshed it, spawn `claude -p "ping" --model
sonnet` / `codex exec --skip-git-repo-check "Reply OK"` on the host.
- Single-flight per credential; cross-instance lockfile so concurrent
`agent-vm` instances don't all kick off a refresh at once.
**Done when:** a multi-hour session crosses a token rotation without the agent
seeing an auth error.
### Phase 5 — Polish & docs
- `.agent-vm.runtime.sh` project hook.
- `--memory N` flag (passed through to microsandbox builder).
- Full arg passthrough to the agent command.
- README rewrite: install, setup, usage, troubleshooting.
- Smoke tests: at least one end-to-end test that boots the image and runs a
trivial agent command.
**Done when:** README is publishable; one CI smoke test green.
## Working agreements
1. **One phase = one PR.** Stop after each.
2. **ARCHITECTURE.md is the source of truth for the *why*.** Every major
design choice in a phase gets a short subsection: what was chosen, what was
rejected, why.
3. **Don't touch old `agent-vm`** (Bash, Python helpers) on the rewrite
branch. The old tree stays on `main` until v1 is shipped from the new
branch.
4. **microsandbox changes go into the submodule, not vendored copies.** If we
need to fork, we do it on a branch of `wirenboard/microsandbox` so the
diff stays reviewable upstream.

377
README.md
View file

@ -1,368 +1,31 @@
# agent-vm
# agent-vm (microsandbox rewrite)
Run AI coding agents inside sandboxed Linux VMs. The agent gets full autonomy while your host system stays safe.
Work-in-progress rewrite of [`agent-vm`](https://github.com/wirenboard/agent-vm)
on top of [microsandbox](https://github.com/wirenboard/microsandbox). The goal
is sub-second cold-start microVMs for AI coding agents (Claude Code, Codex CLI,
OpenCode) with host-rooted credentials and no in-VM proxy.
Uses [Lima](https://lima-vm.io/) to create lightweight Debian VMs on macOS and Linux. Ships with dev tools, Docker, and a headless Chrome browser with [Chrome DevTools MCP](https://github.com/ChromeDevTools/chrome-devtools-mcp) pre-configured.
This is a long-running rewrite. The current state lives on
`rewrite-microsandbox`. The original `agent-vm` continues to live on `main`
until v1 of the rewrite ships.
Supports [Claude Code](https://claude.ai/code), [OpenCode](https://opencode.ai/), and [Codex CLI](https://developers.openai.com/codex/cli/).
## Where to look
Feedbacks welcome!
- [`PLAN.md`](PLAN.md) — phased roadmap, what's in/out of v1.
- [`ARCHITECTURE.md`](ARCHITECTURE.md) — running record of design decisions,
filled in as each phase lands.
## Prerequisites
## Status
- macOS or Linux
- [Lima](https://lima-vm.io/docs/installation/) (installed automatically via Homebrew if available)
- A [Claude subscription](https://claude.ai/) (Pro, Max, or Team), an [OpenCode](https://opencode.ai/) compatible provider, and/or an OpenAI Codex-capable account or API key
Phase 0 (scaffolding): in progress.
## Install
## Building
```bash
git clone https://github.com/sylvinus/agent-vm.git
cd agent-vm
# Add to your shell config
echo "source $(pwd)/claude-vm.sh" >> ~/.zshrc # zsh
echo "source $(pwd)/claude-vm.sh" >> ~/.bashrc # or bash
git submodule update --init vendor/microsandbox
cargo check -p agent-vm
```
## Usage
### One-time setup
```bash
agent-vm setup
```
Creates a VM template with dev tools, Docker, Chromium, Claude Code, OpenCode, and Codex pre-installed. During setup, Claude will launch once for authentication. After it responds, type `/exit` to continue with the rest of the setup. (We haven't found a way to automate this step yet.)
Options:
| Flag | Description | Default |
|------|-------------|---------|
| `--minimal` | Only install git, curl, jq, Claude Code, OpenCode, and Codex. Skips Docker, Node.js, Python, Chromium, and the Chrome MCP server. | off |
| `--disk GB` | VM disk size in GB | 30 |
| `--memory GB` | VM memory ceiling in GB | 16 (Linux), 4 (macOS) |
```bash
agent-vm setup --minimal # Lightweight VM with only core CLIs
agent-vm setup --disk 50 --memory 16 # Larger VM for heavy workloads
```
### Run Claude in a VM
```bash
cd your-project
agent-vm claude
```
Clones the template into a fresh VM, mounts your current directory, and runs `claude --dangerously-skip-permissions` with `IS_SANDBOX=1` to suppress the dangerous mode confirmation prompt (the VM itself is the sandbox). The VM is deleted when Claude exits.
The default model is `opus[1m]` (Opus with 1M context window). You can override it by passing `--model`:
```bash
agent-vm claude -p "fix all lint errors" # Run with a prompt
agent-vm claude --resume # Resume previous session
agent-vm claude -c "explain this codebase" # Continue conversation
agent-vm claude --memory 8 # Start with 8G instead of default 2G
agent-vm claude --model sonnet # Use Sonnet instead of Opus
```
### Run OpenCode in a VM
```bash
cd your-project
agent-vm opencode
```
Runs [OpenCode](https://opencode.ai/) inside a sandboxed VM instead of Claude Code. Uses the same VM template, proxies, and security model. OpenCode is configured to use the Anthropic provider through the host proxy with all permissions auto-approved (the VM is the sandbox).
When launching OpenCode, `agent-vm` prefers native OpenCode OAuth for OpenAI when host-side Codex ChatGPT auth is available and `OPENAI_API_KEY` is not set:
1. It checks the host's Codex auth state and runs a minimal host `codex exec` to verify the login still works and to refresh tokens if needed.
2. If that succeeds, it writes a placeholder OpenCode `openai` OAuth credential into `~/.local/share/opencode/auth.json` inside the VM and routes the real bearer token through the host credential proxy.
3. This keeps the real OpenAI tokens out of the VM, but the OpenCode ChatGPT session only works until that host bearer token expires.
4. If host-side Codex auth is unavailable or invalid, OpenCode falls back to its other configured providers unless you provide `OPENAI_API_KEY`.
Any arguments are forwarded to the `opencode` command:
```bash
agent-vm opencode run "fix all lint errors" # Non-interactive mode
agent-vm opencode --continue # Continue last session
agent-vm opencode -m anthropic/claude-sonnet-4-5 # Specify a model
```
### Run Codex in a VM
```bash
cd your-project
agent-vm codex
```
Runs [Codex CLI](https://developers.openai.com/codex/cli/) inside the same sandboxed VM model. `agent-vm` persists `~/.codex/` per project, configures Codex for `danger-full-access` + `approval_policy = "never"` on first use inside the VM, and updates Codex before launch.
When launching Codex, `agent-vm` now prefers host-side Codex ChatGPT auth when available:
1. It checks the host's Codex auth state and runs a minimal host `codex exec` to verify the login still works and to refresh tokens if needed.
2. If that succeeds, it injects the refreshed bearer token through the host credential proxy and writes a placeholder `~/.codex/auth.json` inside the VM so the VM never sees the real tokens.
3. If host-side Codex auth is unavailable or invalid, it falls back to the old behavior: the VM keeps its project-local `~/.codex/` state and users can run `codex login` inside the VM.
If you set `OPENAI_API_KEY` on the host, `agent-vm` still uses the API-key-based proxy flow for Codex instead of host ChatGPT auth.
Any arguments are forwarded to the `codex` command:
```bash
agent-vm codex "fix all lint errors" # Interactive TUI with an initial prompt
agent-vm codex exec "explain this codebase" # Non-interactive mode
agent-vm codex resume --last # Resume the last Codex session
agent-vm codex --model gpt-5.1-codex-mini # Specify a model
```
### Debug shell
```bash
agent-vm shell # Plain shell with API proxy configured
agent-vm shell claude # Shell pre-configured for Claude
agent-vm shell opencode # Shell pre-configured for OpenCode
agent-vm shell codex # Shell pre-configured for Codex
```
Drops you into a bash shell inside a fresh VM instead of launching an agent. Useful for debugging or manual testing. When an agent is specified, the shell has that agent's configuration (env vars, MCP servers, etc.) pre-applied.
### Dynamic memory (Linux)
On Linux, VMs use a [virtio-balloon](https://www.linux-kvm.org/page/Projects/auto-ballooning) device to dynamically adjust memory. The VM is created with a 16G ceiling but starts with only 2G of usable RAM. As the guest needs more memory, the balloon daemon automatically deflates to give it more, up to the full 16G. When memory pressure drops, unused memory is reclaimed back to the host.
This means you can run multiple VMs without each one reserving its full allocation upfront.
```bash
agent-vm claude --memory 5 --max-memory 10 # Start with 5G, grow up to 10G
agent-vm claude --memory 8 # Start with 8G, ceiling from template (16G)
agent-vm memory # Show current memory of all running VMs
agent-vm memory 12G # Manually set memory to 12G (auto-detect VM)
agent-vm memory my-vm 8G # Set specific VM's memory
```
On macOS (Apple Silicon with VZ backend), QEMU is not used and balloon is not available. VMs use a fixed 4G memory allocation instead. You can still use `--memory` to set a different fixed size. The `agent-vm memory` subcommand (live adjustment) is Linux-only.
### Common options
| Flag | Applies to | Description |
|------|-----------|-------------|
| `--memory GB` | claude, opencode, codex, shell | Initial memory (default: 2G with balloon, 4G without) |
| `--max-memory GB` | claude, opencode, codex, shell | Memory ceiling for balloon (default: from template) |
| `--no-git` | claude, opencode, codex, shell | Skip GitHub integration |
| `--mount DIR` | claude, opencode, codex, shell | Mount extra host directory into VM (repeatable) |
| `--usb DEVICE` | claude, opencode, codex, shell | Pass USB device to VM (repeatable) |
## Customization
### Per-user: `~/.claude-vm.setup.sh`
Create this file in your home directory to install extra tools into the VM template. It runs once during `agent-vm setup`, as the default VM user (with sudo available):
```bash
# ~/.claude-vm.setup.sh
sudo apt-get install -y postgresql-client
pip install pandas numpy
```
### Per-project: `.claude-vm.runtime.sh`
Create this file at the root of any project. It runs inside the cloned VM each time you run an agent, just before the agent starts. Use it for project-specific setup like installing dependencies or starting services:
```bash
# your-project/.claude-vm.runtime.sh
npm install
docker compose up -d
```
## Session persistence
VMs are ephemeral — each invocation creates a fresh clone and deletes it on exit. To make session resume work across VM launches, agent state is persisted outside your project directory.
**How it works:**
1. A per-project state directory is created under `${XDG_STATE_HOME:-~/.local/state}/agent-vm/`
2. Claude's session-related directories (`projects`, `file-history`, `todos`, `plans`) and `history.jsonl` are symlinked from `~/.claude/` to `<state-dir>/claude-sessions/`
3. Ephemeral config (`CLAUDE.md`) stays in-VM and is not persisted
4. `~/.claude.json` is persisted as `<state-dir>/claude-sessions/claude.json` to preserve onboarding/project state
5. `hasCompletedOnboarding=true` is enforced before launch to prevent first-run greeting loops
Set `AGENT_VM_STATE_DIR` to override the state root location.
```bash
agent-vm claude -p "remember ZEBRA" # First session
agent-vm claude --continue # Picks up where the last session left off
```
`agent-vm claude` now launches Claude directly without a hidden priming request.
### OpenCode sessions and configuration
OpenCode session data is persisted in `<state-dir>/opencode-sessions/` by symlinking `~/.local/share/opencode/` inside the VM.
OpenCode configuration is stored in `<state-dir>/opencode-config/opencode.json` and referenced via the `OPENCODE_CONFIG` env var. It configures:
- The Anthropic provider pointing to the host proxy
- All permissions set to `"allow"` (the VM is the sandbox)
- GitHub MCP server (when available)
- Autoupdates disabled
OpenCode authentication state is stored in `<state-dir>/opencode-sessions/auth.json`. When host-side Codex ChatGPT auth is available, `agent-vm opencode` writes a placeholder OAuth credential for the `openai` provider there; the real bearer token stays on the host and is injected by the credential proxy.
### Codex sessions and configuration
Codex state is persisted in `<state-dir>/codex-home/` by symlinking `~/.codex/` inside the VM.
On first use, `agent-vm` writes a minimal `config.toml` there with:
- `sandbox_mode = "danger-full-access"`
- `approval_policy = "never"`
That matches the existing model for Claude/OpenCode: the VM is the sandbox, so Codex itself should not stop to ask for extra approval inside it. Existing Codex config is preserved on subsequent runs.
## Credential Proxy
A two-layer proxy chain keeps all API credentials out of the VM:
1. **mitmproxy** (inside VM, port 8080) — transparently intercepts HTTPS traffic to configured domains, redirects it to the host-side credential proxy. Requests to non-configured domains pass through unchanged. Requests to blocked domains (e.g. `datadoghq.com`) are rejected with 403.
2. **credential-proxy.py** (on host) — receives redirected requests, injects real auth headers based on domain/path matching rules, and forwards to the real upstream. Supports per-repo GitHub tokens via path-prefix matching.
The VM only ever sees placeholder tokens. Real credentials live in the host process's memory. A per-instance shared secret (via standard `Proxy-Authorization`) prevents cross-VM credential theft.
### Anthropic auth via host Claude credentials
When your host has `~/.claude/.credentials.json` (from a normal Claude Code login), agent-vm uses it as the single source of truth for Anthropic auth:
- A placeholder `~/.claude/.credentials.json` is written inside the VM with the host's real `expiresAt` but fake access/refresh tokens.
- The credential proxy reads the host file on every `api.anthropic.com` request and swaps in the real Bearer token — but only when the VM presented that placeholder bearer (or no `Authorization` at all). If Claude Code set the header itself with a different value — e.g. the remote-control bridge polling `/v1/environments/.../work/*` with its `environment_secret` (which carries `org:external_poll_sessions`, a scope the user OAuth token lacks) — that value is forwarded untouched.
- When the VM's Claude Code decides to refresh (POST to `platform.claude.com/v1/oauth/token`), the proxy intercepts it:
- If the host token is still valid, it short-circuits and returns placeholders with the real remaining lifetime.
- Otherwise, it spawns `claude -p "say hi and exit" --model sonnet` on the host so host Claude performs the rotation and writes the new tokens to its own file. The proxy never writes the credentials file itself — host Claude remains the sole writer.
This means token rotation "just works" across multiple concurrent VMs, and real tokens never enter any VM.
### OpenAI auth via host Codex credentials
Same pattern for Codex (and OpenCode when using host ChatGPT auth). When `OPENAI_API_KEY` is unset and host `~/.codex/auth.json` contains a valid ChatGPT login, the proxy:
- Reads `tokens.access_token` from the host file on every `api.openai.com` / `chatgpt.com` request and injects it as the Bearer.
- MITMs `auth.openai.com/oauth/token`: decodes the host JWT's `exp` claim; if still valid, synthesizes a response with forged placeholder JWTs whose claims (`chatgpt_account_id`, `chatgpt_plan_type`, `email`, `exp`) mirror the host's. Otherwise spawns `codex exec --skip-git-repo-check --dangerously-bypass-approvals-and-sandbox "Reply with OK"` on the host — host Codex does the real rotation, the proxy re-reads and forges fresh placeholders.
VM Codex updates its own `auth.json` from the forged response but never sees a real `access_token` / `refresh_token`.
### Configuration
| Env var | Description | Default |
|---------|-------------|---------|
| `OPENAI_API_KEY` | OpenAI API token to inject for Codex / `api.openai.com` | — |
| `AI_HTTPS_PROXY` | Upstream proxy for AI API traffic only (e.g. `http://user:pass@host:8082`) | — |
| `AI_SSL_CERT_FILE` | Extra CA cert PEM for `AI_HTTPS_PROXY` | — |
| `CREDENTIAL_PROXY_DEBUG` | Set to `1` for verbose logging | `0` |
| `CREDENTIAL_PROXY_LOG_DIR` | Directory for log file | `.` |
When `AI_HTTPS_PROXY` is set, only AI API requests (`api.anthropic.com`, `api.openai.com`) are routed through it. GitHub and other traffic goes direct.
## GitHub Integration
When you run `agent-vm claude`, `agent-vm opencode`, or `agent-vm codex` inside a git repo with a GitHub remote, it automatically:
1. Detects the repository (and submodules) from `git remote`
2. Scans any `--mount` directories for additional GitHub repos and their submodules
3. Checks push access via `git push --dry-run`
4. Obtains repo-scoped GitHub tokens via the device flow (browser-based OAuth)
5. Configures the credential proxy with per-repo path-prefix rules
6. Rewrites SSH URLs to HTTPS so all git traffic goes through mitmproxy
7. Writes instructions to `~/.claude/CLAUDE.md` in the VM so the agent knows git is available
This means mounted git repos get full GitHub integration (push, pull, `gh` CLI) just like the main project directory. For example:
```bash
agent-vm claude --mount ../other-repo --mount ../shared-lib
```
All three repos (current directory, `other-repo`, and `shared-lib`) will get their own scoped tokens if they have GitHub remotes with push access.
No credentials are ever exposed to the VM. The credential proxy injects tokens on the host side based on the request path (e.g. `/owner/repo` for git, `/repos/owner/repo` for the GitHub API).
### Token generation and scoping
Tokens are generated via a [GitHub App](https://docs.github.com/en/apps) using the [device flow](https://docs.github.com/en/apps/oauth-apps/building-oauth-apps/authorizing-oauth-apps#device-flow):
1. **One-time setup**: Create a GitHub App with `contents: write` permission and install it on your org/account. The App's Client ID is configured in `claude-vm.sh`.
2. **Per-session**: The device flow runs for each repo that has push access — you approve in a browser.
3. **Multi-repo**: Each repo gets its own scoped token. The credential proxy uses path-prefix matching to inject the right token per request.
4. **Caching**: Tokens are cached in `~/.cache/claude-vm/` and automatically refreshed when expired.
### GitHub Copilot API
Agents get access to the [GitHub Copilot API](https://docs.github.com/en/copilot/using-github-copilot/using-github-copilot-in-the-command-line) (`api.githubcopilot.com`) via a `gho_*` OAuth token injected by the credential proxy.
Token acquisition strategy (in order):
1. **Dedicated Copilot cache** — a previously cached token is loaded from `~/.cache/claude-vm/copilot-token.json`.
2. **Device flow** — if no cached token is found, an OAuth device flow runs using the OpenCode OAuth App (`Ov23li8tweQw6odWQebz`) with `read:user` scope. This app grants access to the full model list (Claude, Gemini, GPT-5, etc.). The token is cached for future sessions.
Copilot access requires your GitHub account to have an active Copilot subscription.
## How it works
1. **`agent-vm setup`** creates a Debian 13 VM with Lima, installs dev tools + Chrome + Claude Code + OpenCode + Codex, and stops it as a reusable template
2. **`agent-vm claude [args]`** clones the template, mounts your working directory read-write, starts the balloon daemon (Linux), runs optional `.claude-vm.runtime.sh`, then launches Claude with full permissions (forwarding any arguments to the `claude` command)
3. **`agent-vm opencode [args]`** same as above but launches OpenCode instead, with its own config and session persistence
4. **`agent-vm codex [args]`** same as above but launches Codex instead, with its own `~/.codex/` persistence and OpenAI proxy wiring
5. **`agent-vm shell [agent]`** same VM setup but drops into a bash shell for debugging
6. On exit, the cloned VM is stopped and deleted. The template persists for reuse
Ports opened inside the VM (e.g. by Docker containers) are automatically forwarded to your host by Lima.
## What's in the VM
| Category | Packages |
|----------|----------|
| Core | git, curl, wget, build-essential, jq |
| Python | python3, pip, venv |
| Node.js | Node.js 22 (via NodeSource) |
| Search | ripgrep, fd-find |
| Browser | Chromium (headless), xvfb |
| Containers | Docker Engine, Docker Compose |
| AI | Claude Code, OpenCode, Codex, Chrome DevTools MCP server |
| Memory | virtio-balloon auto-scaling (Linux only) |
## Why a VM?
Running an AI agent with full permissions is powerful but risky. Here's how the options compare:
| | No sandbox | Docker | VM (agent-vm) |
|---|---|---|---|
| Agent can run any command | Yes | Yes | Yes |
| File system isolation | None | Partial (shared kernel) | Full |
| Network isolation | None | Partial | Full |
| Can run Docker inside | Yes | Requires DinD or socket mount | Yes (native) |
| Kernel-level isolation | None | None (shares host kernel) | Full (separate kernel) |
| Protection from container escapes | None | None | Yes |
| Browser / GUI tools | Host only | Complex setup | Built-in (headless Chromium) |
Docker containers share the host kernel, so a motivated agent could exploit kernel vulnerabilities or misconfigurations to escape. A VM runs its own kernel — even if the agent gains root inside the VM, it can't reach the host.
A VM also avoids the practical headaches of Docker sandboxing. Docker runs natively inside the VM without Docker-in-Docker hacks or socket mounts. Headless Chromium works out of the box without fiddling with `--no-sandbox` flags or shared memory settings. Lima automatically forwards ports from the VM to your host, so if the agent starts a server on port 3000, it's immediately accessible at `localhost:3000`. The agent gets a normal Linux environment where everything just works.
Finally, using a VM means you don't need Node.js, npm, Docker, or any other dev tooling installed on your host machine. The only host dependency is Lima. All the tools (and their vulnerabilities) live inside the VM.
For AI agents running with `--dangerously-skip-permissions`, a VM is the only sandbox that provides meaningful security.
## Usage tracking
Since each VM instance stores Claude session data in its own state directory rather than the standard `~/.claude/`, tools like [ccusage](https://github.com/ryoppippi/ccusage) won't find them by default. A wrapper script is included that discovers all agent-vm session directories automatically:
```bash
bin/ccusage # Daily report (default)
bin/ccusage monthly # Monthly report
bin/ccusage --since 20260401 # Filter by date
bin/ccusage -i # Break down by project
```
The wrapper sets `CLAUDE_CONFIG_DIR` to include `~/.claude`, `~/.config/claude`, and all `claude-sessions/` directories under `~/.local/state/agent-vm/`.
## License
MIT
Actually running a sandbox requires the microsandbox runtime prerequisites
(Linux with KVM, or macOS Apple Silicon). See microsandbox's
[`README`](vendor/microsandbox/README.md).

View file

@ -1,354 +0,0 @@
#!/usr/bin/env python3
"""
balloon-daemon.py QEMU virtio-balloon controller.
Connects to a QEMU QMP socket and manages the virtio-balloon device.
Usage:
balloon-daemon.py <socket> get
balloon-daemon.py <socket> set <size>
balloon-daemon.py <socket> daemon [options]
Arguments:
socket Path to the QMP Unix socket (e.g. ~/.lima/vm/qmp.sock)
Commands:
get Query and print the current balloon size.
set <size> Set the balloon to a fixed target size (e.g. 8G, 4096M).
daemon [options] Continuously monitor guest memory pressure and auto-adjust
the balloon to reclaim unused memory from the host while
keeping enough headroom for the guest.
Daemon options:
--max-memory SIZE Upper bound / ceiling for the balloon. The guest will never
be given more than this. Default: auto-detected from the
current balloon size at startup (i.e. QEMU -m value).
--min-memory SIZE Lower bound / floor. The guest will always have at least
this much memory. Default: 1G.
--initial-target SIZE
Set balloon to this size before entering the monitor loop.
Useful for pre-inflating the balloon at boot. Default: none.
--headroom PCT Keep at least this percentage of *used* memory as free
headroom. E.g. 40 means desired = used * 1.4.
Default: 40.
--min-free SIZE Minimum free memory to maintain. The balloon target will
always be at least used + min-free, regardless of headroom
percentage. Also serves as the emergency deflate threshold:
if available memory drops below this, the balloon grows
aggressively (2x step). Default: 1G.
--step SIZE Base step size for emergency deflation. Normal adjustments
jump directly to the desired target. Default: 256M.
--interval SECS How often to poll guest memory stats and re-evaluate the
balloon target, in seconds. Default: 5.
-v, --verbose Log balloon decisions to stderr.
Algorithm:
On each cycle the daemon computes:
used = cur - avail (actual app memory usage, excludes cache/balloon)
desired = max(used * (1 + headroom%), used + min_free)
desired = clamp(desired, min_memory, max_memory)
Then:
- If avail < min_free: emergency GROW by 2x step (guest critically low)
- If cur < desired - step: GROW jump balloon to desired immediately
- If cur > desired + step: SHRINK jump balloon to desired immediately
Both grow and shrink are fast (single-step jump to target).
Sizes can be specified as: 8G, 8GiB, 8GB, 4096M, 4096MiB, 4096MB, or raw bytes.
"""
import argparse
import json
import signal
import socket
import sys
import time
GiB = 1 << 30
MiB = 1 << 20
def parse_size(s):
"""Parse human-readable size (e.g. '8G', '512M') to bytes."""
s = s.strip().upper()
for suffix, mult in [("GIB", GiB), ("GIG", GiB), ("GB", GiB), ("G", GiB),
("MIB", MiB), ("MB", MiB), ("M", MiB)]:
if s.endswith(suffix):
return int(float(s[:-len(suffix)]) * mult)
return int(s)
def fmt(b):
"""Format bytes as human-readable size."""
if b >= GiB:
return f"{b / GiB:.1f}G"
return f"{b // MiB}M"
class QMP:
"""Minimal QMP (QEMU Machine Protocol) client."""
def __init__(self, path):
self.path = path
self.sock = None
self.buf = b""
def connect(self, retries=1, delay=1):
for i in range(retries):
try:
self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
self.sock.settimeout(10)
self.sock.connect(self.path)
self._recv() # greeting
self._send({"execute": "qmp_capabilities"})
r = self._recv()
if "error" in r:
raise RuntimeError(r["error"])
return
except (ConnectionRefusedError, FileNotFoundError, OSError):
if self.sock:
self.sock.close()
self.sock = None
if i < retries - 1:
time.sleep(delay)
else:
raise
def _send(self, obj):
self.sock.sendall(json.dumps(obj).encode() + b"\n")
def _recv(self):
while True:
nl = self.buf.find(b"\n")
if nl >= 0:
line = self.buf[:nl].strip()
self.buf = self.buf[nl + 1:]
if line:
return json.loads(line)
continue
data = self.sock.recv(4096)
if not data:
raise ConnectionError("QMP connection closed")
self.buf += data
def cmd(self, execute, **kw):
msg = {"execute": execute}
if kw:
msg["arguments"] = kw
self._send(msg)
# Skip async events, wait for command response
while True:
r = self._recv()
if "event" not in r:
return r
def close(self):
if self.sock:
self.sock.close()
self.sock = None
# ── Balloon device discovery ─────────────────────────────────────────────────
# QOM paths where the balloon device may appear
BALLOON_PATHS = [
"/machine/peripheral/balloon0", # explicit id=balloon0
"/machine/peripheral-anon/balloon0", # anonymous device
]
def find_balloon(qmp):
"""Find the balloon device QOM path."""
for path in BALLOON_PATHS:
r = qmp.cmd("qom-get", path=path, property="guest-stats-polling-interval")
if "return" in r:
return path
return None
# ── Subcommands ──────────────────────────────────────────────────────────────
def cmd_get(args):
q = QMP(args.socket)
q.connect(retries=5)
r = q.cmd("query-balloon").get("return", {})
print(fmt(r.get("actual", 0)))
q.close()
def cmd_set(args):
target = parse_size(args.size)
q = QMP(args.socket)
q.connect(retries=5)
q.cmd("balloon", value=target)
print(f"Balloon set to {fmt(target)}")
q.close()
def _qmp_poll(sock_path, retries=5):
"""Connect, return QMP client. Caller must close after use."""
q = QMP(sock_path)
q.connect(retries=retries)
return q
def cmd_daemon(args):
min_mem = parse_size(args.min_memory)
max_mem = parse_size(args.max_memory) if args.max_memory else None
initial = parse_size(args.initial_target) if args.initial_target else None
step = parse_size(args.step)
headroom = args.headroom / 100.0
min_free = parse_size(args.min_free)
iv = args.interval
def log(msg):
if args.verbose:
print(msg, file=sys.stderr, flush=True)
running = True
def stop(*_):
nonlocal running
running = False
signal.signal(signal.SIGTERM, stop)
signal.signal(signal.SIGINT, stop)
# Initial setup: determine max memory, find balloon device, enable stats
q = _qmp_poll(args.socket, retries=30)
log(f"Connected to {args.socket}")
# Determine max memory from QEMU's configured RAM (not balloon, which may
# have been shrunk by a previous daemon run)
if max_mem is None:
max_mem = (q.cmd("query-memory-size-summary")
.get("return", {}).get("base-memory", 0))
if not max_mem:
# Fallback: use current balloon size
max_mem = q.cmd("query-balloon").get("return", {}).get("actual", 0)
if not max_mem:
print("Error: cannot determine VM memory size", file=sys.stderr)
sys.exit(1)
log(f"Max memory: {fmt(max_mem)}")
# Find balloon device and enable stats polling
bp = find_balloon(q)
if not bp:
print("Error: balloon device not found (is -device virtio-balloon-pci enabled?)",
file=sys.stderr)
sys.exit(1)
q.cmd("qom-set", path=bp, property="guest-stats-polling-interval", value=iv)
log(f"Stats polling on {bp} every {iv}s")
# Set initial balloon target if requested
if initial is not None:
target = max(min_mem, min(max_mem, initial))
q.cmd("balloon", value=target)
log(f"Initial target: {fmt(target)}")
q.close() # Release QMP for other tools
# Wait for guest to boot and first stats to be collected
time.sleep(iv + 1)
# Main loop: reconnect on each cycle to avoid holding the QMP socket
while running:
try:
q = _qmp_poll(args.socket)
except (ConnectionRefusedError, FileNotFoundError, OSError):
log("QMP connection failed, VM may have stopped")
break
try:
# Re-enable stats polling (gets reset on reconnect)
q.cmd("qom-set", path=bp, property="guest-stats-polling-interval", value=iv)
guest_stats = q.cmd("qom-get", path=bp, property="guest-stats").get("return", {})
stats = guest_stats.get("stats", {})
avail = stats.get("stat-available-memory", -1)
total = stats.get("stat-total-memory", -1)
if total <= 0 or avail < 0:
log("Waiting for guest stats...")
q.close()
time.sleep(iv)
continue
# Base decisions on actual usage, not ceiling.
# avail = MemAvailable (excludes reclaimable cache).
# Use balloon target (cur), not MemTotal, since MemTotal is fixed
# at boot and includes balloon-claimed pages.
cur = q.cmd("query-balloon").get("return", {}).get("actual", max_mem)
used = max(0, cur - avail)
desired = max(int(used * (1 + headroom)), used + min_free)
desired = max(min_mem, min(max_mem, desired))
log(f"used={fmt(used)} avail={fmt(avail)} "
f"cur={fmt(cur)} desired={fmt(desired)}")
new = cur
if avail < min_free and cur < max_mem:
# Critically low — jump to desired (at minimum +2*step)
new = max(desired, cur + step * 2)
new = min(new, max_mem)
log(f" CRITICAL: {fmt(cur)} -> {fmt(new)}")
elif cur < desired - step:
# Guest needs more memory — jump to desired immediately
new = min(desired, max_mem)
log(f" GROW: {fmt(cur)} -> {fmt(new)}")
elif cur > desired + step:
# Excess — reclaim immediately
new = max(desired, min_mem)
log(f" SHRINK: {fmt(cur)} -> {fmt(new)}")
if new != cur:
q.cmd("balloon", value=new)
except ConnectionError:
log("QMP connection lost")
q.close()
break
except Exception as e:
log(f"Error: {e}")
q.close()
time.sleep(iv)
log("Stopped")
def main():
ap = argparse.ArgumentParser(description="QEMU virtio-balloon controller")
ap.add_argument("socket", help="QMP Unix socket path")
sp = ap.add_subparsers(dest="command", required=True)
sp.add_parser("get", help="Query current balloon size")
ps = sp.add_parser("set", help="Set balloon target")
ps.add_argument("size", help="Target size (e.g. 8G, 4096M)")
pd = sp.add_parser("daemon", help="Auto-adjust balloon based on memory pressure")
pd.add_argument("--max-memory", default=None,
help="Maximum memory / ceiling (default: auto-detect from QEMU -m)")
pd.add_argument("--initial-target", default=None,
help="Set balloon to this size before entering monitor loop (e.g. 8G)")
pd.add_argument("--min-memory", default="1G",
help="Minimum balloon size / floor (default: 1G)")
pd.add_argument("--step", default="256M",
help="Adjustment step size (default: 256M)")
pd.add_argument("--headroom", type=float, default=40,
help="Keep this %% of used memory as free headroom (default: 40)")
pd.add_argument("--min-free", default="1G",
help="Emergency deflate threshold (default: 1G)")
pd.add_argument("--interval", type=int, default=5,
help="Stats polling interval in seconds (default: 5)")
pd.add_argument("-v", "--verbose", action="store_true",
help="Log decisions to stderr")
a = ap.parse_args()
{"get": cmd_get, "set": cmd_set, "daemon": cmd_daemon}[a.command](a)
if __name__ == "__main__":
main()

View file

@ -1,5 +0,0 @@
#!/bin/bash
# Wrapper for ccusage that includes all agent-vm session directories
agent_vm_dirs=$(find "$HOME/.local/state/agent-vm" -maxdepth 2 -name "claude-sessions" -type d 2>/dev/null | paste -sd,)
export CLAUDE_CONFIG_DIR="$HOME/.claude,$HOME/.config/claude${agent_vm_dirs:+,$agent_vm_dirs}"
exec npx ccusage@latest "$@"

File diff suppressed because it is too large Load diff

View file

@ -1,185 +0,0 @@
#!/usr/bin/env python3
"""
PTY wrapper that intercepts Ctrl+V to save host clipboard images to a shared mount.
Sits between the host terminal and a child process (typically limactl shell).
When Ctrl+V (0x16) is detected in stdin, reads the clipboard image on the host
and writes it to $CLIPBOARD_DIR/clipboard.png. All bytes (including 0x16) are
always forwarded to the child.
Usage:
CLIPBOARD_DIR=/path/to/shared python3 clipboard-pty.py limactl shell ...
Supports macOS (osascript/pbpaste) and Linux (wl-paste / xclip).
"""
import os
import pty
import select
import shutil
import signal
import struct
import subprocess
import sys
import fcntl
import termios
CTRL_V = 0x16
def _read_clipboard_image():
"""Try to read PNG image data from the host clipboard. Returns bytes or None."""
if sys.platform == "darwin":
# macOS: use osascript to write clipboard PNGf data to a temp file
try:
result = subprocess.run(
[
"osascript", "-e",
'set png to the clipboard as «class PNGf»',
"-e",
'set f to open for access POSIX file "/tmp/.clipboard-pty.png" with write permission',
"-e",
'set eof f to 0',
"-e",
'write png to f',
"-e",
'close access f',
],
capture_output=True, timeout=3,
)
if result.returncode == 0 and os.path.exists("/tmp/.clipboard-pty.png"):
with open("/tmp/.clipboard-pty.png", "rb") as fh:
data = fh.read()
os.unlink("/tmp/.clipboard-pty.png")
if data[:4] == b"\x89PNG":
return data
except (subprocess.TimeoutExpired, OSError):
pass
return None
# Linux: try wl-paste (Wayland), then xclip (X11)
for cmd in (
["wl-paste", "-t", "image/png"],
["xclip", "-selection", "clipboard", "-t", "image/png", "-o"],
):
if not shutil.which(cmd[0]):
continue
try:
result = subprocess.run(cmd, capture_output=True, timeout=3)
if result.returncode == 0 and result.stdout[:4] == b"\x89PNG":
return result.stdout
except (subprocess.TimeoutExpired, OSError):
pass
return None
def _save_clipboard(clipboard_dir):
"""Read clipboard image and save to clipboard_dir/clipboard.png."""
data = _read_clipboard_image()
if data:
path = os.path.join(clipboard_dir, "clipboard.png")
with open(path, "wb") as fh:
fh.write(data)
def _resize_pty(child_fd):
"""Forward the current terminal size to the child pty."""
try:
sz = fcntl.ioctl(sys.stdin.fileno(), termios.TIOCGWINSZ, b"\x00" * 8)
fcntl.ioctl(child_fd, termios.TIOCSWINSZ, sz)
except OSError:
pass
def main():
if len(sys.argv) < 2:
print("Usage: CLIPBOARD_DIR=/path python3 clipboard-pty.py <command> [args...]",
file=sys.stderr)
sys.exit(1)
clipboard_dir = os.environ.get("CLIPBOARD_DIR", "")
if not clipboard_dir:
print("Error: CLIPBOARD_DIR not set", file=sys.stderr)
sys.exit(1)
os.makedirs(clipboard_dir, exist_ok=True)
# Fork a child with a pty
child_pid, child_fd = pty.fork()
if child_pid == 0:
# Child: exec the command
os.execvp(sys.argv[1], sys.argv[1:])
# Parent: relay I/O between terminal and child pty
is_tty = os.isatty(sys.stdin.fileno())
if is_tty:
# Forward SIGWINCH to child
def on_winch(signum, frame):
_resize_pty(child_fd)
os.kill(child_pid, signal.SIGWINCH)
signal.signal(signal.SIGWINCH, on_winch)
# Set initial size
_resize_pty(child_fd)
# Save and set raw mode on stdin (only when stdin is a real terminal)
old_attrs = termios.tcgetattr(sys.stdin) if is_tty else None
try:
if is_tty:
new_attrs = termios.tcgetattr(sys.stdin)
# Enter raw mode: disable echo, canonical mode, signals, etc.
new_attrs[0] = 0 # iflag
new_attrs[1] = 0 # oflag
new_attrs[2] &= ~(termios.CSIZE | termios.PARENB) # cflag
new_attrs[2] |= termios.CS8
new_attrs[3] = 0 # lflag
new_attrs[6][termios.VMIN] = 1
new_attrs[6][termios.VTIME] = 0
termios.tcsetattr(sys.stdin, termios.TCSAFLUSH, new_attrs)
stdin_fd = sys.stdin.fileno()
stdout_fd = sys.stdout.fileno()
watch_fds = [stdin_fd, child_fd]
while child_fd in watch_fds:
try:
rlist, _, _ = select.select(watch_fds, [], [])
except (OSError, ValueError):
break
if stdin_fd in rlist:
try:
data = os.read(stdin_fd, 1024)
except OSError:
data = b""
if not data:
watch_fds.remove(stdin_fd)
else:
if CTRL_V in data:
_save_clipboard(clipboard_dir)
os.write(child_fd, data)
if child_fd in rlist:
try:
data = os.read(child_fd, 1024)
except OSError:
break
if not data:
break
os.write(stdout_fd, data)
finally:
if old_attrs is not None:
termios.tcsetattr(sys.stdin, termios.TCSAFLUSH, old_attrs)
# Wait for child and exit with its status
_, status = os.waitpid(child_pid, 0)
sys.exit(os.waitstatus_to_exitcode(status))
if __name__ == "__main__":
main()

View file

@ -1,151 +0,0 @@
#!/usr/bin/env python3
"""
Obtain and cache a GitHub OAuth token for the Copilot API.
Uses the OpenCode OAuth App (Ov23li8tweQw6odWQebz) with read:user scope.
This app is the only one that grants access to the full Copilot model list
(Claude, Gemini, GPT-5, etc.).
Usage:
python3 copilot_token.py <cache_file>
Exits 0 and prints the token to stdout on success.
Exits 1 on failure (device code expired, auth error, etc.).
The cache file is read first; if it contains a valid token it is used
without running the device flow. On success the token is written back to
the cache file for future invocations.
"""
import json
import os
import sys
import time
import urllib.error
import urllib.request
CLIENT_ID = "Ov23li8tweQw6odWQebz"
SCOPE = "read:user"
# Bypass any HTTP proxy for GitHub auth endpoints
_opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))
def _post(url, params):
data = json.dumps(params).encode()
req = urllib.request.Request(
url, data=data,
headers={"Accept": "application/json", "Content-Type": "application/json"},
)
with _opener.open(req, timeout=10) as r:
return json.load(r)
def _load_cached(cache_file):
"""Return cached token string if present and valid, else None."""
try:
with open(cache_file) as f:
token = json.load(f).get("access_token", "")
except (FileNotFoundError, json.JSONDecodeError):
return None
if not token:
return None
# Validate against GitHub
req = urllib.request.Request("https://api.github.com/user")
req.add_header("Authorization", f"token {token}")
req.add_header("Accept", "application/vnd.github+json")
try:
with _opener.open(req, timeout=10) as r:
r.read()
except urllib.error.HTTPError as e:
if e.code == 401:
print(" Cached Copilot token rejected (401), discarding", file=sys.stderr)
try:
os.unlink(cache_file)
except OSError:
pass
return None
except Exception:
pass # Network error — keep token, let it fail later
return token
def _save(cache_file, token):
os.makedirs(os.path.dirname(cache_file), exist_ok=True)
with open(cache_file, "w") as f:
json.dump({"access_token": token}, f)
os.chmod(cache_file, 0o600)
def _device_flow(cache_file):
"""Run OAuth device flow. Prints token to stdout and exits 0 on success."""
resp = _post(
"https://github.com/login/device/code",
{"client_id": CLIENT_ID, "scope": SCOPE},
)
interval = resp.get("interval", 5)
print(file=sys.stderr)
print(" " + "=" * 50, file=sys.stderr)
print(f" Open: https://github.com/login/device", file=sys.stderr)
print(f" Enter: {resp['user_code']}", file=sys.stderr)
print(" " + "=" * 50, file=sys.stderr)
print(file=sys.stderr)
print(f" Waiting for authorization (expires in {resp['expires_in']}s)...", file=sys.stderr)
deadline = time.time() + resp["expires_in"]
while time.time() < deadline:
time.sleep(interval)
try:
token_resp = _post(
"https://github.com/login/oauth/access_token",
{
"client_id": CLIENT_ID,
"device_code": resp["device_code"],
"grant_type": "urn:ietf:params:oauth:grant-type:device_code",
},
)
except Exception:
continue
if "access_token" in token_resp:
token = token_resp["access_token"]
_save(cache_file, token)
print(" Authorization successful!", file=sys.stderr)
print(token)
sys.exit(0)
error = token_resp.get("error", "")
if error == "slow_down":
interval = token_resp.get("interval", interval + 5)
elif error == "authorization_pending":
continue
else:
print(f" Error: {error}", file=sys.stderr)
sys.exit(1)
print(" Error: device code expired", file=sys.stderr)
sys.exit(1)
def main():
if len(sys.argv) != 2:
print(f"Usage: {sys.argv[0]} <cache_file>", file=sys.stderr)
sys.exit(1)
cache_file = sys.argv[1]
token = _load_cached(cache_file)
if token:
print("Using cached Copilot token", file=sys.stderr)
print(token)
sys.exit(0)
print("Requesting GitHub token for Copilot API access...", file=sys.stderr)
_device_flow(cache_file)
if __name__ == "__main__":
main()

View file

@ -0,0 +1,15 @@
[package]
name = "agent-vm"
description = "Sandboxed VMs for AI coding agents, built on microsandbox."
version.workspace = true
edition.workspace = true
license.workspace = true
repository.workspace = true
[[bin]]
name = "agent-vm"
path = "src/main.rs"
[dependencies]
microsandbox = { path = "../../vendor/microsandbox/crates/microsandbox" }
tokio = { version = "1.42", features = ["macros", "rt-multi-thread"] }

View file

@ -0,0 +1,25 @@
//! agent-vm — sandboxed VMs for AI coding agents.
//!
//! Phase 0: hello-world. Boots a throwaway alpine microsandbox, runs `echo`,
//! and exits. This exists only to prove the SDK wiring is correct end-to-end.
//! Real subcommands land in Phase 2.
use microsandbox::Sandbox;
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
let sandbox = Sandbox::builder("agent-vm-hello")
.image("alpine")
.cpus(1)
.memory(256)
.replace()
.create()
.await?;
let output = sandbox.shell("echo hello from alpine").await?;
println!("{}", output.stdout()?.trim_end());
sandbox.stop_and_wait().await?;
Sandbox::remove("agent-vm-hello").await?;
Ok(())
}

View file

@ -1,860 +0,0 @@
#!/usr/bin/env python3
"""
Host-side unified credential proxy for agent VMs.
Receives HTTP requests from the VM's mitmproxy with X-Original-Host header,
matches against credential rules, injects auth headers, and forwards to
the real upstream over HTTPS.
The VM never sees real credentials.
Usage:
CREDENTIAL_PROXY_RULES='[{"domain":"api.anthropic.com","headers":{"Authorization":"Bearer sk-..."}}]' \
python3 credential-proxy.py
# Prints the listening port to stdout, then serves until SIGTERM.
Optional env vars:
CREDENTIAL_PROXY_SECRET Shared secret; if set, requests must include
Proxy-Authorization header (407 otherwise)
CREDENTIAL_PROXY_DEBUG Set to "1" for verbose logging
CREDENTIAL_PROXY_LOG_DIR Directory for log file (default: current dir)
AI_HTTPS_PROXY Upstream proxy for AI API connections only
(e.g. http://user:pass@localhost:8082)
Only used for rules with "use_proxy": true
AI_SSL_CERT_FILE Path to additional CA certificate PEM file
(for AI_HTTPS_PROXY's TLS)
"""
import base64
import fcntl
import http.client
import http.server
import json
import os
import select
import signal
import socket
import socketserver
import ssl
import subprocess
import sys
import threading
import time
from urllib.parse import parse_qsl, urlparse, unquote
MAX_REQUEST_BODY = 32 * 1024 * 1024 # 32 MB
UPSTREAM_TIMEOUT = 300 # seconds
INBOUND_TIMEOUT = 60 # seconds
WEBSOCKET_IDLE_TIMEOUT = 300 # seconds
# Server-Sent Events long-polls (e.g. the Claude Code worker event channel)
# are idle by design — applying UPSTREAM_TIMEOUT as a socket read timeout would
# tear the channel down every few minutes. Use a much larger bound so a truly
# dead connection still gets reaped, but normal idle periods don't trip it.
EVENT_STREAM_IDLE_TIMEOUT = 3600 # seconds
PROXY_SECRET = os.environ.get("CREDENTIAL_PROXY_SECRET", "")
DEBUG = os.environ.get("CREDENTIAL_PROXY_DEBUG", "0") == "1"
LOG_DIR = os.environ.get("CREDENTIAL_PROXY_LOG_DIR", ".")
AI_SSL_CERT_FILE = os.environ.get("AI_SSL_CERT_FILE", "")
_log_file = None
# Upstream proxy for AI API connections (rules with "use_proxy": true)
_ai_proxy = None
_AI_HTTPS_PROXY = os.environ.get("AI_HTTPS_PROXY", "")
if _AI_HTTPS_PROXY:
_p = urlparse(_AI_HTTPS_PROXY)
_ai_proxy = {"host": _p.hostname, "port": _p.port or 8080}
if _p.username:
_creds = base64.b64encode(
f"{unquote(_p.username)}:{unquote(_p.password or '')}".encode()
).decode()
_ai_proxy["auth"] = f"Basic {_creds}"
# SSL contexts: one for AI proxy connections, one for direct connections
_ssl_ctx = ssl.create_default_context()
_ai_ssl_ctx = ssl.create_default_context()
if AI_SSL_CERT_FILE:
_ai_ssl_ctx.load_verify_locations(AI_SSL_CERT_FILE)
def _get_log_file():
global _log_file
if _log_file is None:
log_path = os.path.join(LOG_DIR, "credential-proxy.log")
fd = os.open(log_path, os.O_WRONLY | os.O_APPEND | os.O_CREAT, 0o600)
_log_file = os.fdopen(fd, "a")
return _log_file
def debug(msg):
if DEBUG:
line = f"[credential-proxy {time.strftime('%H:%M:%S')}] {msg}\n"
f = _get_log_file()
f.write(line)
f.flush()
def redact(value, show=8):
if not value:
return "<empty>"
if len(value) <= show:
return value
return value[:show] + "..."
def _read_credentials_file(path):
"""Read and parse a credentials JSON file under an advisory shared lock.
flock(LOCK_SH) already permits concurrent in-process readers; no extra
threading.Lock needed. The shared lock blocks only while host Claude
holds LOCK_EX to rewrite the file i.e. exactly when we want readers
to wait for the new tokens.
"""
try:
with open(path, "r") as f:
fcntl.flock(f.fileno(), fcntl.LOCK_SH)
try:
return json.load(f)
finally:
fcntl.flock(f.fileno(), fcntl.LOCK_UN)
except (OSError, json.JSONDecodeError) as e:
debug(f" credentials read failed: {path}: {e}")
return None
def _json_path_get(data, dotted):
cur = data
for part in dotted.split("."):
if not isinstance(cur, dict) or part not in cur:
return None
cur = cur[part]
return cur
def _jwt_payload(token):
"""Decode a JWT payload (no signature verification) or return {}."""
try:
parts = (token or "").split(".")
if len(parts) < 2:
return {}
pad = "=" * (-len(parts[1]) % 4)
return json.loads(base64.urlsafe_b64decode(parts[1] + pad))
except (ValueError, json.JSONDecodeError, UnicodeDecodeError):
return {}
def _jwt_exp_seconds(token):
exp = _jwt_payload(token).get("exp")
try:
return int(exp) if exp is not None else 0
except (TypeError, ValueError):
return 0
def _jwt_forge(payload):
"""Emit an unsigned JWT (header.payload.placeholder) — host Codex's own
placeholder writer uses the same shape and Codex reads it back fine."""
def b64(obj):
return base64.urlsafe_b64encode(
json.dumps(obj, separators=(",", ":")).encode()
).decode().rstrip("=")
header = {"alg": "RS256", "typ": "JWT"}
return f"{b64(header)}.{b64(payload)}.placeholder"
# Serializes refresh invocations per credentials file: we never want two
# refresh probes racing for the same file.
_refresh_locks = {}
_refresh_locks_guard = threading.Lock()
def _refresh_lock(path):
with _refresh_locks_guard:
lock = _refresh_locks.get(path)
if lock is None:
lock = threading.Lock()
_refresh_locks[path] = lock
return lock
_HOST_REFRESH_ENV_DROP = {
"HTTPS_PROXY", "HTTP_PROXY", "NO_PROXY",
"https_proxy", "http_proxy", "no_proxy",
"CLAUDE_CODE_OAUTH_TOKEN",
}
def _trigger_host_refresh(command, timeout):
"""Spawn a host agent (claude / codex) to force its credential refresh.
The spawned process inherits the user env but with proxy vars stripped so
it talks to the real upstream, not back to us.
"""
env = {k: v for k, v in os.environ.items()
if k not in _HOST_REFRESH_ENV_DROP}
debug(f" oauth_refresh: spawning {command}")
try:
result = subprocess.run(
command, env=env,
stdin=subprocess.DEVNULL,
stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
timeout=timeout, check=False,
)
if result.returncode != 0:
debug(f" oauth_refresh: refresh command rc={result.returncode}: "
f"{result.stdout[:500]!r}")
return False
debug(f" oauth_refresh: refresh command succeeded")
return True
except subprocess.TimeoutExpired:
debug(f" oauth_refresh: refresh command timed out after {timeout}s")
return False
except (FileNotFoundError, OSError) as e:
debug(f" oauth_refresh: refresh command failed to launch: {e}")
return False
# Parse credential rules from env. See _handle_oauth_refresh() and
# _build_upstream_headers() for the supported rule fields (headers, path_prefix,
# use_proxy, auth_from_file, oauth_refresh).
_RULES = {}
_rules_json = os.environ.get("CREDENTIAL_PROXY_RULES", "[]")
try:
for rule in json.loads(_rules_json):
domain = rule["domain"]
entry = {
"headers": rule.get("headers", {}),
"path_prefix": rule.get("path_prefix"),
"use_proxy": rule.get("use_proxy", False),
"auth_from_file": rule.get("auth_from_file"),
"oauth_refresh": rule.get("oauth_refresh"),
"oauth_refresh_codex": rule.get("oauth_refresh_codex"),
}
_RULES.setdefault(domain, []).append(entry)
# Sort each domain's rules: longest path_prefix first, None last
for domain in _RULES:
_RULES[domain].sort(
key=lambda r: (r["path_prefix"] is None, -(len(r["path_prefix"] or ""))))
except (json.JSONDecodeError, KeyError, TypeError) as e:
print(f"Error: invalid CREDENTIAL_PROXY_RULES: {e}", file=sys.stderr)
sys.exit(1)
def _match_rule(domain, path):
"""Find the best matching rule for domain + path."""
candidates = _RULES.get(domain)
if not candidates:
return None
for rule in candidates:
prefix = rule["path_prefix"]
if prefix is None or path.startswith(prefix):
return rule
return None
def _build_upstream_headers(request_headers, original_host, rule):
"""Copy inbound headers, remove proxy metadata, and apply injected headers."""
headers = {}
header_keys_lower = {}
for key, value in request_headers.items():
lower = key.lower()
if lower in (
"host", "accept-encoding",
"x-original-host", "x-original-port", "x-original-scheme",
"proxy-authorization",
):
continue
if lower in header_keys_lower:
actual_key = header_keys_lower[lower]
headers[actual_key] = f"{headers[actual_key]},{value}"
else:
headers[key] = value
header_keys_lower[lower] = key
headers["Host"] = original_host
def _set_header(name, value):
to_remove = []
for existing_key in headers:
if existing_key.lower() == name.lower():
to_remove.append(existing_key)
for key in to_remove:
del headers[key]
header_keys_lower.pop(key.lower(), None)
headers[name] = value
header_keys_lower[name.lower()] = name
if rule:
for header_name, header_value in rule["headers"].items():
_set_header(header_name, header_value)
debug(f" injected {header_name}: {redact(header_value)}")
auth_src = rule.get("auth_from_file")
if auth_src:
header_name = auth_src["header"]
# The VM ships a fixed placeholder credential and we rewrite it on
# the wire. But some clients set this header themselves with a value
# we must not clobber — notably Claude Code's remote-control bridge,
# which authenticates /v1/environments/.../work/* with the
# environment_secret it got at registration (that secret carries
# org:external_poll_sessions; the user OAuth token does not). So if
# "overwrite_only" is configured, only replace the header when its
# inbound value is absent/empty or one of those known placeholders;
# otherwise pass the client's value through untouched. Without
# "overwrite_only" the header is always replaced (legacy behaviour).
overwrite_only = auth_src.get("overwrite_only")
current = next(
(headers[k] for k in headers if k.lower() == header_name.lower()),
None,
)
if overwrite_only is not None and current and current not in overwrite_only:
debug(f" auth_from_file: kept client-set {header_name} "
f"({redact(current)}) — not a proxy-managed placeholder")
else:
creds = _read_credentials_file(auth_src["path"])
token = _json_path_get(creds or {}, auth_src["json_path"])
if token:
header_value = auth_src["format"].replace("{token}", token)
_set_header(header_name, header_value)
debug(f" injected {header_name} from {auth_src['path']}: {redact(header_value)}")
else:
debug(f" auth_from_file missing token at {auth_src['json_path']} in {auth_src['path']}")
if DEBUG:
for key, value in headers.items():
lower = key.lower()
if lower in ("authorization", "x-api-key"):
debug(f" > {key}: {redact(value)}")
else:
debug(f" > {key}: {value}")
return headers
def _open_upstream_connection(original_host, original_port, original_scheme, use_proxy):
"""Open an upstream HTTP(S) connection for non-upgrade requests."""
if original_scheme == "https" and use_proxy:
conn = http.client.HTTPSConnection(
_ai_proxy["host"], _ai_proxy["port"], context=_ai_ssl_ctx, timeout=UPSTREAM_TIMEOUT
)
tunnel_headers = {}
if "auth" in _ai_proxy:
tunnel_headers["Proxy-Authorization"] = _ai_proxy["auth"]
conn.set_tunnel(original_host, original_port, headers=tunnel_headers)
return conn
if original_scheme == "https":
return http.client.HTTPSConnection(
original_host, original_port, context=_ssl_ctx, timeout=UPSTREAM_TIMEOUT
)
return http.client.HTTPConnection(original_host, original_port, timeout=UPSTREAM_TIMEOUT)
def _open_upstream_socket(original_host, original_port, original_scheme, use_proxy):
"""Open a raw socket suitable for WebSocket upgrade forwarding."""
if use_proxy:
sock = socket.create_connection((_ai_proxy["host"], _ai_proxy["port"]), timeout=UPSTREAM_TIMEOUT)
sock.settimeout(UPSTREAM_TIMEOUT)
connect_lines = [f"CONNECT {original_host}:{original_port} HTTP/1.1"]
connect_lines.append(f"Host: {original_host}:{original_port}")
if "auth" in _ai_proxy:
connect_lines.append(f"Proxy-Authorization: {_ai_proxy['auth']}")
connect_lines.append("")
connect_lines.append("")
sock.sendall("\r\n".join(connect_lines).encode())
response = b""
while b"\r\n\r\n" not in response:
chunk = sock.recv(4096)
if not chunk:
raise OSError("proxy closed CONNECT response")
response += chunk
if len(response) > 65536:
raise OSError("CONNECT response too large")
header_block, remainder = response.split(b"\r\n\r\n", 1)
status_line = header_block.split(b"\r\n", 1)[0].decode("iso-8859-1", errors="replace")
try:
status_code = int(status_line.split(" ", 2)[1])
except (IndexError, ValueError) as exc:
raise OSError(f"invalid CONNECT response: {status_line}") from exc
if status_code != 200:
raise OSError(f"CONNECT failed: {status_line}")
if remainder:
raise OSError("unexpected buffered data after CONNECT")
if original_scheme == "https":
server_hostname = original_host
ctx = _ai_ssl_ctx
sock = ctx.wrap_socket(sock, server_hostname=server_hostname)
return sock
sock = socket.create_connection((original_host, original_port), timeout=UPSTREAM_TIMEOUT)
sock.settimeout(UPSTREAM_TIMEOUT)
if original_scheme == "https":
sock = _ssl_ctx.wrap_socket(sock, server_hostname=original_host)
return sock
def _read_http_response(sock):
"""Read an HTTP response head and return status/header info plus buffered body bytes."""
response = b""
while b"\r\n\r\n" not in response:
chunk = sock.recv(4096)
if not chunk:
raise OSError("upstream closed before sending response headers")
response += chunk
if len(response) > 65536:
raise OSError("response headers too large")
header_block, remainder = response.split(b"\r\n\r\n", 1)
lines = header_block.decode("iso-8859-1").split("\r\n")
status_line = lines[0]
parts = status_line.split(" ", 2)
if len(parts) < 2:
raise OSError(f"invalid upstream status line: {status_line}")
status_code = int(parts[1])
reason = parts[2] if len(parts) > 2 else ""
headers = []
for line in lines[1:]:
if not line or ":" not in line:
continue
key, value = line.split(":", 1)
headers.append((key.strip(), value.lstrip()))
return status_code, reason, headers, remainder
def _tunnel_bidirectional(client_sock, upstream_sock, initial_upstream_data=b""):
sockets = [client_sock, upstream_sock]
if initial_upstream_data:
client_sock.sendall(initial_upstream_data)
while sockets:
readable, _, exceptional = select.select(sockets, [], sockets, WEBSOCKET_IDLE_TIMEOUT)
if exceptional:
break
if not readable:
debug(" websocket tunnel idle timeout")
break
for current in readable:
peer = upstream_sock if current is client_sock else client_sock
try:
data = current.recv(65536)
except (socket.timeout, ssl.SSLWantReadError):
continue
if not data:
return
peer.sendall(data)
class CredentialProxyHandler(http.server.BaseHTTPRequestHandler):
protocol_version = "HTTP/1.1"
def log_message(self, format, *args):
pass
def do_request(self):
# Verify shared secret via standard proxy auth (prevents cross-VM credential theft)
if PROXY_SECRET:
proxy_auth = self.headers.get("Proxy-Authorization", "")
expected = "Basic " + base64.b64encode(f"_:{PROXY_SECRET}".encode()).decode()
if proxy_auth != expected:
debug(f"REJECTED: invalid Proxy-Authorization from {self.client_address[0]}")
self.send_error_response(407, "Proxy authentication required")
return
# Extract original host from mitmproxy header
original_host = self.headers.get("X-Original-Host", "")
original_port = int(self.headers.get("X-Original-Port", "443"))
original_scheme = self.headers.get("X-Original-Scheme", "https")
if not original_host:
self.send_error_response(400, "Missing X-Original-Host header")
return
# Read request body with size limit
try:
content_length = int(self.headers.get("Content-Length", 0))
except (ValueError, TypeError):
self.send_error_response(400, "Invalid Content-Length")
return
if content_length < 0 or content_length > MAX_REQUEST_BODY:
self.send_error_response(413, "Request body too large")
return
body = self.rfile.read(content_length) if content_length else None
debug(f">>> {self.command} {self.path} (host={original_host}, {content_length} bytes)")
# Apply credential rules for this domain (+ optional path prefix)
rule = _match_rule(original_host, self.path)
if not rule:
debug(f" no credential rule for {original_host}, forwarding as-is")
if rule and rule.get("oauth_refresh"):
self._handle_oauth_refresh(rule["oauth_refresh"], body)
return
if rule and rule.get("oauth_refresh_codex"):
self._handle_codex_oauth_refresh(rule["oauth_refresh_codex"], body)
return
headers = _build_upstream_headers(self.headers, original_host, rule)
# Connect to upstream (route through AI proxy if rule says so)
upstream_port = original_port
use_proxy = rule and rule.get("use_proxy") and _ai_proxy
is_websocket = (
self.headers.get("Upgrade", "").lower() == "websocket"
and "upgrade" in self.headers.get("Connection", "").lower()
)
if is_websocket:
try:
t0 = time.monotonic()
upstream_sock = _open_upstream_socket(
original_host, upstream_port, original_scheme, use_proxy
)
request_lines = [f"{self.command} {self.path} HTTP/1.1"]
for key, value in headers.items():
request_lines.append(f"{key}: {value}")
request_lines.append("")
request_lines.append("")
upstream_sock.sendall("\r\n".join(request_lines).encode())
status_code, reason, upstream_headers, buffered = _read_http_response(upstream_sock)
self.send_response(status_code, reason)
for key, value in upstream_headers:
lower = key.lower()
if lower in ("transfer-encoding", "content-length", "keep-alive"):
continue
self.send_header(key, value)
if status_code != 101:
self.send_header("Connection", "close")
self.end_headers()
self.wfile.flush()
latency_ms = (time.monotonic() - t0) * 1000
debug(f"<<< {status_code} websocket-upgrade ({latency_ms:.0f}ms)")
if status_code == 101:
self.connection.settimeout(WEBSOCKET_IDLE_TIMEOUT)
upstream_sock.settimeout(WEBSOCKET_IDLE_TIMEOUT)
try:
_tunnel_bidirectional(self.connection, upstream_sock, buffered)
finally:
upstream_sock.close()
else:
if buffered:
self.wfile.write(buffered)
self.wfile.flush()
upstream_sock.close()
self.close_connection = True
return
except Exception as e:
debug(f" WEBSOCKET UPSTREAM ERROR: {e}")
self.send_error_response(502, f"Failed to connect to {original_host}")
return
try:
t0 = time.monotonic()
conn = _open_upstream_connection(original_host, upstream_port, original_scheme, use_proxy)
conn.request(self.command, self.path, body=body, headers=headers)
upstream = conn.getresponse()
latency_ms = (time.monotonic() - t0) * 1000
except Exception as e:
debug(f" UPSTREAM ERROR: {e}")
self.send_error_response(502, f"Failed to connect to {original_host}")
return
# Detect streaming: chunked transfer-encoding, or a Server-Sent Events
# response (text/event-stream is a deliberately-idle long-poll, e.g. the
# Claude Code worker event channel that delivers web-typed prompts).
is_streaming = False
is_event_stream = False
upstream_headers = upstream.getheaders()
for key, value in upstream_headers:
lower_key = key.lower()
if lower_key == "transfer-encoding":
is_streaming = True
elif lower_key == "content-type" and "text/event-stream" in value.lower():
is_streaming = True
is_event_stream = True
debug(f"<<< {upstream.status} {'event-stream' if is_event_stream else 'streaming' if is_streaming else 'complete'} ({latency_ms:.0f}ms)")
if is_streaming:
# A streaming response (SSE long-poll, slow token stream, …) can sit
# idle far longer than UPSTREAM_TIMEOUT between bytes; that timeout
# is meant for connection setup, not for an open stream. Relax the
# upstream read timeout so we don't tear the stream — and the client,
# which then reconnects — down every few minutes.
if getattr(conn, "sock", None) is not None:
try:
conn.sock.settimeout(EVENT_STREAM_IDLE_TIMEOUT)
except OSError:
pass
# Streaming: re-chunk the decoded body
self.send_response(upstream.status)
for key, value in upstream_headers:
lower = key.lower()
if lower in ("transfer-encoding", "content-length",
"connection", "keep-alive"):
continue
self.send_header(key, value)
self.send_header("Transfer-Encoding", "chunked")
self.send_header("Connection", "close")
self.close_connection = True
self.end_headers()
total_bytes = 0
try:
while True:
# read1(): forward each batch of bytes as it arrives instead
# of waiting for an 8 KiB buffer to fill — essential for
# Server-Sent Events, which dribble small events with long
# gaps in between.
data = upstream.read1(8192)
if not data:
break
total_bytes += len(data)
self.wfile.write(f"{len(data):x}\r\n".encode())
self.wfile.write(data)
self.wfile.write(b"\r\n")
self.wfile.flush()
self.wfile.write(b"0\r\n\r\n")
self.wfile.flush()
elapsed_ms = (time.monotonic() - t0) * 1000
debug(f" streamed {total_bytes} bytes ({elapsed_ms:.0f}ms total)")
except (BrokenPipeError, ConnectionResetError) as e:
debug(f" stream broken: {e} after {total_bytes} bytes")
except OSError as e:
# socket.timeout (TimeoutError) and other socket errors land
# here; headers are already sent, so just log and close.
debug(f" stream interrupted: {type(e).__name__}: {e} after {total_bytes} bytes")
finally:
conn.close()
else:
# Non-streaming: read full body, send with Content-Length
body_data = upstream.read()
conn.close()
debug(f" body: {len(body_data)} bytes")
self.send_response(upstream.status)
for key, value in upstream_headers:
lower = key.lower()
if lower in ("transfer-encoding", "content-length",
"connection", "keep-alive"):
continue
self.send_header(key, value)
self.send_header("Content-Length", str(len(body_data)))
self.end_headers()
self.wfile.write(body_data)
self.wfile.flush()
def send_error_response(self, code, message):
body = json.dumps({"error": {"type": "proxy_error", "message": message}}).encode()
self.send_response(code)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
def _handle_oauth_refresh(self, cfg, body):
"""Synthesize an OAuth refresh response by consulting the host file.
If the host's access token is still valid, return it (as placeholders
plus the real remaining lifetime). Otherwise, spawn host Claude to
perform the actual OAuth refresh (host Claude writes the file), then
re-read and return the fresh window again as placeholders.
"""
creds_path = cfg["credentials_file"]
json_path = cfg.get("json_path", "claudeAiOauth")
expiry_grace_ms = int(cfg.get("expiry_grace_ms", 60_000))
placeholder_access = cfg.get(
"placeholder_access_token", "sk-ant-oat01-placeholder-proxy-managed")
placeholder_refresh = cfg.get(
"placeholder_refresh_token", "sk-ant-ort01-placeholder-proxy-managed")
refresh_command = cfg.get(
"refresh_command", ["claude", "-p", "say hi and exit", "--model", "sonnet"])
refresh_timeout = int(cfg.get("refresh_timeout", 120))
try:
body_json = json.loads(body) if body else {}
except (json.JSONDecodeError, UnicodeDecodeError):
body_json = {}
if body_json.get("grant_type") not in (None, "refresh_token"):
debug(f" oauth_refresh: unsupported grant_type {body_json.get('grant_type')!r}")
self.send_error_response(400, "unsupported grant_type")
return
def _current_oauth():
creds = _read_credentials_file(creds_path) or {}
return _json_path_get(creds, json_path) or {}
def _expiry_ms(oauth):
try:
return int(oauth.get("expiresAt") or 0)
except (TypeError, ValueError):
return 0
oauth = _current_oauth()
if _expiry_ms(oauth) <= int(time.time() * 1000) + expiry_grace_ms:
with _refresh_lock(creds_path):
# Re-check: another thread may have refreshed while we waited.
oauth = _current_oauth()
if _expiry_ms(oauth) <= int(time.time() * 1000) + expiry_grace_ms:
ok = _trigger_host_refresh(refresh_command, refresh_timeout)
oauth = _current_oauth()
if not ok or _expiry_ms(oauth) <= int(time.time() * 1000):
debug(" oauth_refresh: host refresh did not update credentials")
self.send_error_response(502, "host refresh failed")
return
expires_in = max(1, (_expiry_ms(oauth) - int(time.time() * 1000)) // 1000)
scopes = oauth.get("scopes") or []
response_body = {
"token_type": "Bearer",
"access_token": placeholder_access,
"refresh_token": placeholder_refresh,
"expires_in": int(expires_in),
}
if scopes:
response_body["scope"] = " ".join(scopes)
encoded = json.dumps(response_body).encode()
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(encoded)))
self.send_header("Cache-Control", "no-store")
self.end_headers()
self.wfile.write(encoded)
debug(f" oauth_refresh: returned placeholder tokens, expires_in={expires_in}s")
def _handle_codex_oauth_refresh(self, cfg, body):
"""MITM for auth.openai.com/oauth/token.
The request body is application/x-www-form-urlencoded (Codex convention,
not JSON). We read JWT `exp` from host's tokens.access_token; if stale,
spawn host `codex exec` which does the real rotation and rewrites the
file. Response is OAuth-shaped JSON with forged placeholder JWTs whose
claims mirror the host's (account_id, plan_type, email, exp) so the VM
Codex writes a valid-looking auth.json real tokens stay on the host.
"""
creds_path = cfg["credentials_file"]
expiry_grace_ms = int(cfg.get("expiry_grace_ms", 60_000))
refresh_command = cfg.get(
"refresh_command", ["codex", "exec",
"--skip-git-repo-check",
"--dangerously-bypass-approvals-and-sandbox",
"--color", "never",
"Reply with exactly OK and nothing else."])
refresh_timeout = int(cfg.get("refresh_timeout", 120))
client_id = cfg.get("audience", "app_EMoamEEZ73f0CkXaXp7hrann")
placeholder_refresh = cfg.get(
"placeholder_refresh_token",
"placeholder-refresh-token-injected-by-proxy")
try:
params = dict(parse_qsl((body or b"").decode("utf-8"), keep_blank_values=True))
except UnicodeDecodeError:
params = {}
if params.get("grant_type") not in (None, "refresh_token"):
debug(f" oauth_refresh_codex: unsupported grant_type {params.get('grant_type')!r}")
self.send_error_response(400, "unsupported grant_type")
return
def _host_access_jwt():
creds = _read_credentials_file(creds_path) or {}
return _json_path_get(creds, "tokens.access_token") or ""
now_ms = lambda: int(time.time() * 1000)
if _jwt_exp_seconds(_host_access_jwt()) * 1000 <= now_ms() + expiry_grace_ms:
with _refresh_lock(creds_path):
if _jwt_exp_seconds(_host_access_jwt()) * 1000 <= now_ms() + expiry_grace_ms:
ok = _trigger_host_refresh(refresh_command, refresh_timeout)
if not ok or _jwt_exp_seconds(_host_access_jwt()) * 1000 <= now_ms():
debug(" oauth_refresh_codex: host refresh did not update credentials")
self.send_error_response(502, "host refresh failed")
return
creds = _read_credentials_file(creds_path) or {}
tokens = creds.get("tokens") or {}
access_payload = _jwt_payload(tokens.get("access_token"))
id_payload = _jwt_payload(tokens.get("id_token"))
auth_claims = (access_payload.get("https://api.openai.com/auth")
or id_payload.get("https://api.openai.com/auth") or {})
account_id = (tokens.get("account_id")
or auth_claims.get("chatgpt_account_id"))
plan_type = auth_claims.get("chatgpt_plan_type")
email = (id_payload.get("email")
or (id_payload.get("https://api.openai.com/profile") or {}).get("email"))
exp = int(access_payload.get("exp") or 0)
iat = max(0, exp - 3600)
shared_auth = {"chatgpt_account_id": account_id, "chatgpt_plan_type": plan_type}
vm_id_payload = {
"iss": "https://auth.openai.com",
"aud": [client_id],
"iat": iat, "exp": exp,
"https://api.openai.com/auth": shared_auth,
}
if email:
vm_id_payload["email"] = email
vm_access_payload = {
"iss": "https://auth.openai.com",
"aud": ["https://api.openai.com/v1"],
"iat": iat, "nbf": iat, "exp": exp,
"scp": access_payload.get("scp") or [
"openid", "profile", "email", "offline_access"],
"https://api.openai.com/auth": shared_auth,
}
expires_in = max(1, exp - int(time.time()))
response_body = {
"token_type": "Bearer",
"access_token": _jwt_forge(vm_access_payload),
"id_token": _jwt_forge(vm_id_payload),
"refresh_token": placeholder_refresh,
"expires_in": expires_in,
}
encoded = json.dumps(response_body).encode()
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(encoded)))
self.send_header("Cache-Control", "no-store")
self.end_headers()
self.wfile.write(encoded)
debug(f" oauth_refresh_codex: returned forged JWTs, expires_in={expires_in}s")
do_GET = do_request
do_POST = do_request
do_PUT = do_request
do_PATCH = do_request
do_DELETE = do_request
do_HEAD = do_request
do_OPTIONS = do_request
class QuietServer(socketserver.ThreadingMixIn, http.server.HTTPServer):
allow_reuse_address = False
daemon_threads = True
timeout = INBOUND_TIMEOUT
def main():
if not _RULES:
print("Error: CREDENTIAL_PROXY_RULES is empty or not set", file=sys.stderr)
sys.exit(1)
server = QuietServer(("127.0.0.1", 0), CredentialProxyHandler)
port = server.server_address[1]
print(port, flush=True)
def handle_signal(signum, frame):
threading.Thread(target=server.shutdown, daemon=True).start()
signal.signal(signal.SIGTERM, handle_signal)
signal.signal(signal.SIGINT, handle_signal)
rule_summary = []
for domain, rules in _RULES.items():
for r in rules:
prefix = r["path_prefix"]
rule_summary.append(f"{domain}{prefix or ''}")
debug(f"Listening on port {port}")
debug(f"Credential rules for: {', '.join(rule_summary)}")
if _ai_proxy:
debug(f"AI upstream proxy: {_ai_proxy['host']}:{_ai_proxy['port']}")
if AI_SSL_CERT_FILE:
debug(f"AI SSL cert: {AI_SSL_CERT_FILE}")
server.serve_forever()
if __name__ == "__main__":
main()

View file

@ -1,313 +0,0 @@
# Архитектурное решение: Unified MITM Proxy
## 1. Введение и цели
### Задача
Безопасная изоляция учётных данных (API-ключей Anthropic, токенов GitHub) от агентных виртуальных машин, при этом обеспечивая прозрачный доступ к API.
### Ключевые требования качества
| Приоритет | Атрибут | Описание |
|-----------|---------|----------|
| 1 | Безопасность | ВМ никогда не видит реальные токены |
| 2 | Прозрачность | Приложения внутри ВМ работают со стандартными URL |
| 3 | Поддержка стриминга | SSE/chunked-ответы от Anthropic API |
| 4 | Простота | Минимум компонентов, нет внешних зависимостей на хосте |
### Заинтересованные стороны
| Роль | Ожидание |
|------|----------|
| Оператор | Простой запуск и настройка |
| Агент (Claude/OpenCode) | Прозрачный доступ к API без модификаций |
| Безопасник | Учётные данные не утекают в ВМ |
## 2. Ограничения
- ВМ управляются через Lima/QEMU
- Хост ↔ ВМ связь через `host.lima.internal`
- На хосте только стандартная библиотека Python
- В ВМ допускается установка пакетов через pip
## 3. Контекст системы
```
┌──────────────────────────────────────────────────────────────────────┐
│ Хост │
│ │
│ ~/.claude/.credentials.json ◄─── claude -p (refresh) │
│ │ ▲ │
│ │ чтение per request │ spawn при истечении │
│ ▼ │ │
│ credential-proxy.py ──────────► api.anthropic.com │
│ (127.0.0.1:PORT) platform.claude.com (oauth) │
│ ▲ github.com, api.github.com │
│ │ HTTP │
│ ─ ─ ─ ─ ─ │ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ │
│ Lima VM │ │
│ │ │
│ Claude ──► mitmdump ────────────► pypi.org, npm и др. │
│ git (HTTPS_PROXY) (passthrough, без MITM) │
│ gh │
└──────────────────────────────────────────────────────────────────────┘
```
## 4. Стратегия решения
**Было (3 прокси + URL-хаки):**
- `claude-vm-proxy.py` — прокси для Anthropic API
- `github-git-proxy.py` — прокси для git через `url.insteadOf`
- `github-mcp-proxy.py` — MCP-прокси для GitHub
- Подмена `ANTHROPIC_BASE_URL`, `git config url.*.insteadOf`
**Стало (2-уровневый MITM):**
- `mitmproxy-addon.py` — перехват HTTPS внутри ВМ
- `credential-proxy.py` — инъекция токенов на хосте
**Причина замены:** 3 прокси + URL-хаки — хрупкая архитектура. Приложения ожидают реальные URL, а подмена ломает валидацию, редиректы и конфигурацию. MITM-подход прозрачен для приложений.
## 5. Строительные блоки
### 5.1. credential-proxy.py (хост)
Единый HTTP-прокси, заменяющий три предыдущих.
**Конфигурация** (`CREDENTIAL_PROXY_RULES` env):
```json
[
{"domain": "api.anthropic.com",
"auth_from_file": {"path": "~/.claude/.credentials.json",
"json_path": "claudeAiOauth.accessToken",
"header": "Authorization", "format": "Bearer {token}"},
"use_proxy": true},
{"domain": "platform.claude.com", "path_prefix": "/v1/oauth/token",
"oauth_refresh": {"credentials_file": "~/.claude/.credentials.json",
"placeholder_access_token": "sk-ant-oat01-placeholder-proxy-managed",
"placeholder_refresh_token": "sk-ant-ort01-placeholder-proxy-managed",
"refresh_command": ["claude", "-p", "say hi and exit", "--model", "sonnet"]}},
{"domain": "github.com", "path_prefix": "/org1/repo1", "headers": {"Authorization": "Basic <base64-token1>"}},
{"domain": "github.com", "path_prefix": "/org2/repo2", "headers": {"Authorization": "Basic <base64-token2>"}},
{"domain": "github.com", "headers": {"Authorization": "Basic <base64-fallback>"}},
{"domain": "api.github.com", "path_prefix": "/repos/org1/repo1", "headers": {"Authorization": "token TOKEN1"}},
{"domain": "api.github.com", "path_prefix": "/repos/org2/repo2", "headers": {"Authorization": "token TOKEN2"}},
{"domain": "api.github.com", "headers": {"Authorization": "token FALLBACK"}}
]
```
Опциональное поле `path_prefix` позволяет разные токены для разных репозиториев.
Правила сортируются по длине префикса (longest match first), правила без префикса — fallback.
Поле `use_proxy` маршрутизирует запрос через `AI_HTTPS_PROXY` (upstream proxy для AI API).
**Источники заголовков авторизации:**
- `headers` — статические значения из конфигурации (например, Basic для GitHub);
- `auth_from_file` — значение считывается из JSON-файла на хосте при каждом запросе
(путь к полю указан через `json_path`); используется для Anthropic — токен берётся
из `~/.claude/.credentials.json` хоста, поэтому VM видит актуальный Bearer и после
ротации без перезапуска.
**OAuth refresh MITM (`oauth_refresh`):**
Перехватывает POST к `platform.claude.com/v1/oauth/token`. Вместо того чтобы ходить на
upstream, прокси читает `credentials_file` хоста:
1. Если `claudeAiOauth.expiresAt > now + 60s` — возвращает VM фиктивные токены
(`placeholder_access_token` / `placeholder_refresh_token`) и реальный остаток времени
(`expires_in`).
2. Иначе запускает `refresh_command` (по умолчанию `claude -p "say hi and exit" --model sonnet`),
дожидается окончания (timeout 120s), перечитывает файл, возвращает placeholders.
3. Если команда не обновила файл — отвечает 502.
Прокси **никогда не пишет** файл `credentials.json` — хост-Claude остаётся единственным
писателем, поэтому логика ротации полностью совпадает с логикой хоста. На уровне процесса
используется per-file `threading.Lock`, чтобы одновременно не запускать несколько
`claude -p` для одного файла.
**Поток обработки запроса:**
1. Получить HTTP-запрос от mitmproxy
2. Проверить `Proxy-Authorization: Basic` (если `CREDENTIAL_PROXY_SECRET` задан) → 403 при несовпадении
3. Извлечь `X-Original-Host`, `X-Original-Port`, `X-Original-Scheme`
4. Сопоставить домен + путь с правилами (longest path_prefix first)
5. Перезаписать заголовки авторизации (удаляя существующие)
6. Убрать `Accept-Encoding`, `Proxy-Authorization` из upstream-запроса
7. Переслать на upstream по HTTPS
8. Вернуть ответ (с re-chunking для стриминга)
**Upstream proxy:** Правила с `"use_proxy": true` маршрутизируются через `AI_HTTPS_PROXY` (HTTP CONNECT tunnel). `AI_SSL_CERT_FILE` добавляет CA-сертификат для TLS-соединения с upstream proxy. Остальные правила (GitHub) идут напрямую.
**Лимиты:** 32 МБ тело запроса, 300с таймаут upstream, 60с таймаут входящих.
### 5.2. mitmproxy-addon.py (ВМ)
Аддон для mitmdump (~30 строк). Перехватывает HTTPS-трафик для настроенных доменов.
**Принцип работы:**
- Фильтрация по `CREDENTIAL_PROXY_DOMAINS` (comma-separated) — только перечисленные домены перенаправляются на credential-proxy
- Остальной трафик (pypi, npm, Docker Hub) проходит через mitmproxy без изменений к реальным upstream
- Для перехваченных запросов: добавить `X-Original-*` и `Proxy-Authorization: Basic` заголовки, перенаправить на хост-прокси по HTTP
- Запросы к заблокированным доменам (`BLOCKED_DOMAINS`) отклоняются с 403
**Запуск:** через `systemd-run --user` для выживания после завершения SSH-сессии `limactl shell`.
### 5.3. Интеграция в claude-vm.sh
**Последовательность запуска:**
1. Получить токены:
- Anthropic: проверить наличие валидного `~/.claude/.credentials.json` хоста;
- GitHub App через device auth.
2. Собрать `CREDENTIAL_PROXY_RULES` JSON (пропускается, если нет токенов)
3. Запустить `credential-proxy.py` на хосте → получить порт (пропускается, если правила пусты)
4. Клонировать и загрузить ВМ
5. Настроить учётные данные Claude Code внутри ВМ:
- если используется хостовый файл — записать placeholder `~/.claude/.credentials.json`
(реальный `expiresAt` из хостового файла, подменные токены);
- иначе — `CLAUDE_CODE_OAUTH_TOKEN=placeholder` при `AI_HTTPS_PROXY`.
6. Сгенерировать CA-сертификат mitmproxy, установить в доверенные
7. Установить прокси env vars через `/etc/profile.d/credential-proxy.sh`
8. Запустить mitmdump с аддоном через `systemd-run --user`
9. Настроить git: credential helper (placeholder-токен) + `url.insteadOf` (SSH → HTTPS)
10. Настроить gh CLI: `hosts.yml` + `config.yml` с `version: "1"` (предотвращает миграцию)
11. Запустить агента (без `ANTHROPIC_BASE_URL`)
**Проверка push-доступа:**
Перед запуском device auth flow для каждого репозитория (основного и подмодулей) проверяется наличие write-доступа через `git push --dry-run --no-verify origin HEAD`. Работает с любым типом хост-аутентификации (SSH-ключи, HTTPS-токены). Успешный exit code или отказ "non-fast-forward" / "up to date" означают наличие push-доступа. Репозитории без write-доступа пропускаются — device auth flow не запускается.
**Условный запуск:**
- Хостовый `~/.claude/.credentials.json` валиден → правила `auth_from_file`
(api.anthropic.com) + `oauth_refresh` (platform.claude.com); в ВМ пишется placeholder
`~/.claude/.credentials.json`; `CLAUDE_CODE_OAUTH_TOKEN` не создаётся, чтобы Claude
использовал файл и проходил стандартный OAuth refresh через прокси.
- Хостового файла нет, но `AI_HTTPS_PROXY` задан → `CLAUDE_CODE_OAUTH_TOKEN=placeholder`
(трафик маршрутизируется через upstream proxy).
- Ничего из перечисленного → `CLAUDE_CODE_OAUTH_TOKEN` не создаётся.
- Нет ни одного токена (ни Anthropic, ни GitHub) и нет `AI_HTTPS_PROXY` → credential-proxy
и mitmproxy не запускаются, ВМ работает без прокси.
## 6. Решения об архитектуре
### ADR-1: mitmproxy вместо URL-подмены
| | URL-подмена (было) | MITM-прокси (стало) |
|---|---|---|
| Прозрачность | Приложения видят подменённые URL | Приложения видят реальные URL |
| Компонентов | 3 прокси + конфиг-хаки | 2 компонента |
| Новые домены | Добавить прокси + insteadOf | Добавить правило в JSON |
| Зависимости | Нет | mitmproxy в ВМ (~15 МБ) |
**Решение:** MITM-подход. Дополнительная зависимость оправдана радикальным упрощением.
### ADR-2: Per-repo авторизация через path_prefix
- `api.anthropic.com``Authorization: Bearer`
- `github.com/owner/repo``Authorization: Basic <base64(x-access-token:TOKEN_FOR_REPO)>` (git HTTP)
- `api.github.com/repos/owner/repo``Authorization: token <TOKEN_FOR_REPO>` (gh CLI, API)
- `github.com` (fallback) → первый доступный токен
- `api.github.com` (fallback) → первый доступный токен
Правила с `path_prefix` сортируются по длине (longest match first). Запросы к неизвестным репозиториям или к `/user`, `/search` и т.д. используют fallback.
**Причина:** GitHub App installation tokens привязаны к конкретным репозиториям. Проект с подмодулями из разных организаций требует разные токены.
### ADR-3: Фильтрация доменов в аддоне
Аддон проверяет `CREDENTIAL_PROXY_DOMAINS` (comma-separated) и перенаправляет только запросы к перечисленным доменам. Остальные проходят через mitmproxy без изменений к реальным upstream.
**Было:** `--ignore-hosts` с негативным lookahead regex. Не работало — mitmproxy туннелировал CONNECT-запросы без вызова `request()` хука.
**Стало:** Аддон сам фильтрует домены. mitmproxy перехватывает весь HTTPS (MITMs всё), но аддон перенаправляет на credential-proxy только настроенные домены.
**Причина:** Простота и надёжность. Regex + `--ignore-hosts` ломался из-за shell escaping и несовместимости между версиями mitmproxy.
### ADR-4: SSH → HTTPS через git insteadOf
ВМ настроена с `git config --global url."https://github.com/".insteadOf "git@github.com:"`.
**Причина:** Протокол SSH (порт 22) нельзя перехватить через HTTP-прокси. mitmproxy работает только с HTTP/HTTPS. Rewrite на HTTPS обеспечивает прохождение всего git-трафика через MITM-цепочку для инъекции токенов.
### ADR-5: Отказ от retry-логики
Прокси не повторяет запросы при ошибках upstream.
**Причина:** Клиенты (Claude SDK, git) имеют свою логику повторов. Двойные retry создают каскадные проблемы.
### ADR-7: Хост-агент — единственный писатель файла учётных данных
**Контекст:** VM-агент (Claude или Codex) должен рефрешить OAuth-токен, но не должен
видеть ни реальный access, ни реальный refresh. На хосте параллельно может работать
пользовательский агент, который тоже владеет тем же файлом.
**Решение:** Прокси обрабатывает refresh-запросы VM **без обращения к upstream** — вместо
этого он запускает локального агента (`claude -p ...` для Anthropic, `codex exec ...`
для OpenAI). Хост-агент выполняет свою обычную логику ротации и сам пишет обновлённый
файл. Прокси затем перечитывает файл и возвращает VM placeholder-токены с реальным
`expires_in`.
**Почему так:**
- **Один писатель** — исключает конфликт между прокси и хост-агентом при одновременной
ротации.
- **Логика ротации не дублируется** — если поставщик изменит формат refresh-ответа,
обновление хост-агента автоматически покроет и VM.
- **Никаких утечек** — прокси читает реальные токены только в памяти, файлом владеет хост.
**Для Codex дополнительно:** `auth.json` содержит JWT-токены. Прокси подделывает свежие
placeholder-JWT с корректными claim'ами (`chatgpt_account_id`, `plan_type`, `email`, `exp`),
подписанные литералом `.placeholder`. Codex внутри VM принимает их (подпись проверяется
только сервером), но реальные JWT остаются только в файле хоста.
**Инъекция только placeholder'а (`auth_from_file.overwrite_only`):** для `api.anthropic.com`
прокси переписывает `Authorization` **только** если ВМ прислала тот placeholder-bearer,
которым она была провижинена (или заголовок отсутствует). Если Claude Code сам выставил
заголовок другим значением — например, remote-control bridge ходит в
`/v1/environments/.../work/poll` с `environment_secret`, у которого есть scope
`org:external_poll_sessions` (его нет у пользовательского OAuth-токена), — это значение
проходит без изменений, иначе bridge получает 403 и завершается внутри ВМ.
**Альтернатива:** Прокси сам делает HTTP-запрос к `platform.claude.com` /
`auth.openai.com` и переписывает файл. Отвергнуто — усложняет инвариант "у файла один
писатель" и дублирует код OAuth.
### ADR-6: Per-instance секрет для изоляции ВМ
**Угроза:** На одном хосте несколько ВМ с разными наборами репозиториев. ВМ-1 может обратиться к credential-proxy ВМ-2 через `host.lima.internal:PORT` и получить токены чужих репозиториев.
**Решение:** При каждом запуске генерируется 256-бит случайный секрет (`secrets.token_hex(32)`). Секрет передаётся:
- credential-proxy через `CREDENTIAL_PROXY_SECRET` env var
- mitmproxy через `CREDENTIAL_PROXY_SECRET` env var
Аддон добавляет `Proxy-Authorization: Basic <secret>` к каждому запросу. Прокси проверяет его перед инъекцией — несовпадение → 403. Заголовок удаляется перед пересылкой upstream.
**Альтернатива:** Привязка к IP/порту — ненадёжна (ВМ делят сетевой стек хоста через Lima). Секрет проще и надёжнее.
## 7. Риски и технический долг
| Риск | Вероятность | Влияние | Митигация |
|------|-------------|---------|-----------|
| mitmproxy обновление ломает аддон | Низкая | Средняя | Аддон использует стабильный API (request hook) |
| Новый домен требует MITM | Средняя | Низкая | Добавить правило в JSON + домен в `CREDENTIAL_PROXY_DOMAINS` |
| CA-сертификат просрочится | Низкая | Высокая | Генерируется при каждом запуске ВМ |
| Upstream меняет формат авторизации | Низкая | Высокая | Правила конфигурируемы через JSON |
| mitmproxy не стартует в ВМ | Средняя | Высокая | `systemd-run --user` + проверка готовности + journalctl диагностика |
## 8. Тестирование
60 автоматических тестов (`test_credential_proxy.py`):
- **TestCredentialProxy** (11) — инъекция заголовков, HTTP-методы, пути, дубликаты
- **TestCredentialProxyErrors** (4) — 400/413/502 ошибки
- **TestCredentialProxySecret** (4) — Proxy-Authorization проверка, 403 при несовпадении, strip перед upstream
- **TestCredentialProxyUnmatched** (1) — домены без правил
- **TestCredentialProxyStreaming** (1) — chunked-ответы
- **TestCredentialProxyWebSocket** (1) — websocket upgrade tunnel
- **TestCredentialProxyPathMatching** (4) — per-repo маршрутизация по path_prefix
- **TestCredentialProxyStartup** (3) — запуск, некорректный JSON, SIGTERM
- **TestCredentialProxyUpstreamProxy** (1) — per-rule маршрутизация через AI_HTTPS_PROXY
- **TestCredentialProxyConnectTunnel** (1) — CONNECT-туннель с авторизацией upstream proxy
- **TestCredentialProxyExtraCACerts** (2) — загрузка дополнительных CA-сертификатов
- **TestCredentialProxyHostClaudeCredentials** (7) — `auth_from_file` читает per-request;
`oauth_refresh` (Claude) при валидном токене не запускает `refresh_command`; при
просроченном — запускает и перечитывает файл; отказ → 502; `oauth_refresh_codex`
аналогично (JWT-expiry check, форма form-encoded, подделка JWT-ответа)
- **TestBuildCredentialRules** (11) — генерация per-repo правил из токенов (включая
always-emit api.anthropic.com, `auth_from_file` + `oauth_refresh` при хостовом
Claude-файле, и Codex `auth_from_file` + OPENAI_API_KEY precedence)
- **TestCodexHostAuthHelpers** (3) — чтение/валидация хостовых Codex auth
- **TestMitmproxyAddon** (6) — перезапись flow, секрет, блокировка доменов
Запуск: `python3 -m pytest test_credential_proxy.py -v`

View file

@ -1,542 +0,0 @@
#!/usr/bin/env python3
"""
GitHub App User Token Generator
Generates repo-scoped GitHub user access tokens via the device flow.
Only requires the GitHub App's Client ID (no secrets needed).
Usage:
# Print instructions to create the GitHub App (one-time setup)
python3 github_app_token_demo.py create-app
# Generate a user access token via device flow
python3 github_app_token_demo.py user-token --client-id Iv23liXXXXXX
# Scope the token to a single repo (writes only)
python3 github_app_token_demo.py user-token --client-id Iv23liXXXXXX --repo wirenboard/agent-vm
Add -v/--verbose to any command for detailed HTTP logging.
"""
import argparse
import json
import os
import sys
import re
import time
import urllib.parse
import urllib.request
import urllib.error
TARGET_OWNER = os.environ.get("TARGET_OWNER", "wirenboard")
TARGET_REPO = os.environ.get("TARGET_REPO", "agent-vm")
VERBOSE = False
# Retry settings for transient GitHub errors (5xx)
MAX_RETRIES = 3
RETRY_DELAYS = [1, 2, 4]
ERROR_LOG_DIR = os.environ.get("GITHUB_TOKEN_LOG_DIR", ".")
def verbose(msg):
if VERBOSE:
print(f" [verbose] {msg}", file=sys.stderr)
def _looks_like_html(body_str):
"""Check if a response body is HTML."""
prefix = body_str[:256].lstrip()
return prefix.startswith(("<", "<!DOCTYPE", "<!doctype"))
def _save_error_body(status, body_str, context=""):
"""Save an error response body to a file for debugging. Returns the path."""
ts = time.strftime("%Y%m%d-%H%M%S")
filename = f"github-token-error-{ts}-{status}.html"
path = os.path.join(ERROR_LOG_DIR, filename)
try:
with open(path, "w") as f:
if context:
f.write(f"<!-- {context} -->\n")
f.write(body_str)
verbose(f"saved error response to {path}")
except OSError as e:
verbose(f"failed to save error response: {e}")
path = None
return path
def _copy_to_clipboard(text):
"""Try to copy text to the system clipboard. Fails silently."""
import subprocess
import shutil
for cmd in (["pbcopy"], ["xclip", "-selection", "clipboard"], ["xsel", "--clipboard", "--input"], ["wl-copy"]):
if shutil.which(cmd[0]):
try:
subprocess.run(cmd, input=text.encode(), check=True, timeout=5)
verbose(f"copied to clipboard via {cmd[0]}")
return True
except (subprocess.SubprocessError, OSError):
pass
verbose("no clipboard tool found")
return False
def parse_repo(repo_str):
"""Parse a repo URL/slug into (owner, name). Accepts:
- owner/name
- https://github.com/owner/name
- git@github.com:owner/name.git
"""
# git@github.com:owner/name.git
m = re.match(r'git@github\.com:([^/]+)/([^/]+?)(?:\.git)?$', repo_str)
if m:
return m.group(1), m.group(2)
# https://github.com/owner/name[.git]
m = re.match(r'https?://github\.com/([^/]+)/([^/]+?)(?:\.git)?(?:/.*)?$', repo_str)
if m:
return m.group(1), m.group(2)
# owner/name
m = re.match(r'^([^/]+)/([^/]+)$', repo_str)
if m:
return m.group(1), m.group(2)
print(f"Error: cannot parse repo: {repo_str}", file=sys.stderr)
sys.exit(1)
def resolve_repo_id(owner, name):
"""Resolve a repo's numeric ID. Tries public API first, falls back to gh CLI."""
import subprocess
url = f"https://api.github.com/repos/{owner}/{name}"
req = urllib.request.Request(url)
req.add_header("Accept", "application/vnd.github+json")
verbose(f"Resolving repo ID: {owner}/{name}")
try:
with urllib.request.urlopen(req) as resp:
data = json.loads(resp.read().decode())
repo_id = data["id"]
verbose(f"Resolved {owner}/{name} -> {repo_id}")
return str(repo_id)
except urllib.error.HTTPError:
verbose(f"Public API failed, trying gh CLI...")
try:
result = subprocess.run(
["gh", "api", f"repos/{owner}/{name}", "--jq", ".id"],
capture_output=True, text=True, check=True,
)
repo_id = result.stdout.strip()
verbose(f"Resolved {owner}/{name} -> {repo_id} (via gh)")
return repo_id
except (subprocess.CalledProcessError, FileNotFoundError) as e:
print(f"Error: cannot resolve repo {owner}/{name}", file=sys.stderr)
print(f" Public API returned 404 and gh CLI failed: {e}", file=sys.stderr)
print(f" Use --repository-id to pass the numeric ID directly", file=sys.stderr)
sys.exit(1)
# --- App creation helper ---
APP_MANIFEST = {
"name": "WB Agent VM Token Generator",
"url": "https://github.com/wirenboard/agent-vm",
"hook_attributes": {"active": False},
"redirect_url": "",
"public": False,
"default_permissions": {"contents": "write"},
"default_events": [],
}
def cmd_create_app():
"""Print instructions for creating the GitHub App via the UI."""
print("=" * 60)
print("GitHub App Creation Instructions")
print("=" * 60)
print()
print("1. Go to: https://github.com/organizations/wirenboard/settings/apps/new")
print(" (Or for personal account: https://github.com/settings/apps/new)")
print()
print("2. Fill in the following settings:")
print(f" - GitHub App name: {APP_MANIFEST['name']}")
print(f" - Homepage URL: {APP_MANIFEST['url']}")
print(" - Uncheck 'Active' under Webhook")
print(" - Under 'Repository permissions':")
print(" - Contents: Read & write")
print(" - Under 'Where can this GitHub App be installed?':")
print(" - Select 'Only on this account'")
print()
print(" After creating the app, go to Optional features:")
print(" - User-to-server token expiration: Opt-out")
print(" (Expiring tokens break when multiple repos are used,")
print(" because each device flow revokes previous refresh tokens)")
print()
print("3. Click 'Create GitHub App'")
print()
print("4. Note the Client ID (starts with Iv...)")
print()
print("5. Install the app:")
print(" - Click 'Install App' in the left sidebar")
print(" - Choose the target organization/account")
print(" - Select repositories to grant access to")
print()
print("6. Generate a token:")
print(f" python3 {sys.argv[0]} user-token --client-id <CLIENT_ID> --repo <owner/repo>")
print()
# --- GitHub API helper ---
def github_api(method, path, token, token_type="Bearer", body=None):
"""Make a GitHub API request with retry on 5xx errors."""
url = f"https://api.github.com{path}"
data = json.dumps(body).encode() if body else None
verbose(f">>> {method} {url}")
if body:
verbose(f">>> Body: {json.dumps(body)}")
last_error = None
for attempt in range(MAX_RETRIES + 1):
if attempt > 0:
delay = RETRY_DELAYS[min(attempt - 1, len(RETRY_DELAYS) - 1)]
verbose(f"retry {attempt}/{MAX_RETRIES} after {delay}s")
time.sleep(delay)
req = urllib.request.Request(url, data=data, method=method)
req.add_header("Accept", "application/vnd.github+json")
req.add_header("Authorization", f"{token_type} {token}")
req.add_header("X-GitHub-Api-Version", "2022-11-28")
if data:
req.add_header("Content-Type", "application/json")
try:
with urllib.request.urlopen(req) as resp:
resp_body = resp.read().decode()
verbose(f"<<< {resp.status} {resp.reason}")
verbose(f"<<< Body: {resp_body[:500]}{'...' if len(resp_body) > 500 else ''}")
return json.loads(resp_body)
except urllib.error.HTTPError as e:
error_body = e.read().decode()
verbose(f"<<< {e.code} {e.reason}")
verbose(f"<<< Body: {error_body}")
if e.code >= 500:
if _looks_like_html(error_body):
_save_error_body(e.code, error_body, f"{method} {url} attempt {attempt + 1}")
last_error = f"{e.code} {e.reason} (HTML error page saved to disk)"
else:
last_error = f"{e.code} {e.reason}"
print(f"GitHub API error: {e.code} {e.reason} (attempt {attempt + 1}/{MAX_RETRIES + 1}, retrying...)", file=sys.stderr)
continue
# Non-5xx error — don't retry
print(f"GitHub API error: {e.code} {e.reason}", file=sys.stderr)
print(f"URL: {url}", file=sys.stderr)
if _looks_like_html(error_body):
saved = _save_error_body(e.code, error_body, f"{method} {url}")
print(f"Response body saved to: {saved}", file=sys.stderr)
else:
print(f"Response: {error_body}", file=sys.stderr)
sys.exit(1)
except urllib.error.URLError as e:
last_error = str(e.reason)
print(f"GitHub API error: {last_error} (attempt {attempt + 1}/{MAX_RETRIES + 1}, retrying...)", file=sys.stderr)
continue
print(f"GitHub API error: {last_error} (all {MAX_RETRIES + 1} attempts failed)", file=sys.stderr)
print(f"URL: {url}", file=sys.stderr)
sys.exit(1)
def verify_token(token, token_type="token"):
"""Verify the token works by reading the target repo."""
repo = github_api(
"GET",
f"/repos/{TARGET_OWNER}/{TARGET_REPO}",
token,
token_type=token_type,
)
return repo
# --- Token cache ---
DEFAULT_CACHE_DIR = os.path.expanduser("~/.cache/claude-vm")
def _cache_path(cache_dir, client_id, owner, repo):
"""Return the cache file path for a given client/repo combo."""
safe = f"{owner}_{repo}".replace("/", "_")
return os.path.join(cache_dir, f"github-token-{safe}.json")
def load_cached_token(cache_dir, client_id, owner, repo):
"""Try to load a valid token from cache. Returns token string or None."""
path = _cache_path(cache_dir, client_id, owner, repo)
try:
with open(path, "r") as f:
data = json.load(f)
except (FileNotFoundError, json.JSONDecodeError):
return None
token = data.get("access_token")
if not token:
return None
# Discard old-format cache files that had expiring tokens
if data.get("refresh_token") or data.get("expires_at"):
verbose("Discarding old cache with expiring token format")
try:
os.unlink(path)
except OSError:
pass
return None
# Validate the token is still accepted by GitHub
req = urllib.request.Request("https://api.github.com/user")
req.add_header("Authorization", f"token {token}")
req.add_header("Accept", "application/vnd.github+json")
try:
with urllib.request.urlopen(req) as resp:
resp.read()
except urllib.error.HTTPError as e:
if e.code == 401:
verbose("Cached token rejected by GitHub (401), discarding")
try:
os.unlink(path)
except OSError:
pass
return None
except Exception:
pass # Network error — keep token, let it fail later
verbose(f"Using cached token")
return token
def save_cached_token(cache_dir, client_id, owner, repo, access_token):
"""Save token to cache file."""
path = _cache_path(cache_dir, client_id, owner, repo)
os.makedirs(os.path.dirname(path), exist_ok=True)
data = {"access_token": access_token}
with open(path, "w") as f:
json.dump(data, f)
# Restrict permissions — token file
os.chmod(path, 0o600)
verbose(f"Token cached to {path}")
# --- Device flow (user access token) ---
def device_flow_request(url, params):
"""POST to GitHub's OAuth endpoints (form-encoded, JSON response).
Retries on 5xx errors; saves HTML error pages to disk."""
encoded_data = urllib.parse.urlencode(params).encode()
verbose(f">>> POST {url}")
verbose(f">>> Body: {urllib.parse.urlencode(params)}")
last_error = None
for attempt in range(MAX_RETRIES + 1):
if attempt > 0:
delay = RETRY_DELAYS[min(attempt - 1, len(RETRY_DELAYS) - 1)]
verbose(f"retry {attempt}/{MAX_RETRIES} after {delay}s")
time.sleep(delay)
req = urllib.request.Request(url, data=encoded_data, method="POST")
req.add_header("Accept", "application/json")
try:
with urllib.request.urlopen(req) as resp:
resp_body = resp.read().decode()
verbose(f"<<< {resp.status} {resp.reason}")
verbose(f"<<< Body: {resp_body[:500]}{'...' if len(resp_body) > 500 else ''}")
return json.loads(resp_body)
except urllib.error.HTTPError as e:
error_body = e.read().decode()
verbose(f"<<< {e.code} {e.reason}")
verbose(f"<<< Body: {error_body}")
if e.code >= 500:
if _looks_like_html(error_body):
_save_error_body(e.code, error_body, f"POST {url} attempt {attempt + 1}")
last_error = f"{e.code} {e.reason} (HTML error page saved to disk)"
else:
last_error = f"{e.code} {e.reason}"
print(f" GitHub returned {e.code} (attempt {attempt + 1}/{MAX_RETRIES + 1}, retrying...)", file=sys.stderr)
continue
# Non-5xx error — don't retry
print(f"GitHub OAuth error: {e.code} {e.reason}", file=sys.stderr)
if _looks_like_html(error_body):
saved = _save_error_body(e.code, error_body, f"POST {url}")
print(f"Response body saved to: {saved}", file=sys.stderr)
else:
print(f"Response: {error_body}", file=sys.stderr)
sys.exit(1)
except urllib.error.URLError as e:
last_error = str(e.reason)
print(f" Connection error: {last_error} (attempt {attempt + 1}/{MAX_RETRIES + 1}, retrying...)", file=sys.stderr)
continue
print(f"GitHub OAuth failed: {last_error} (all {MAX_RETRIES + 1} attempts failed)", file=sys.stderr)
sys.exit(1)
def cmd_user_token(client_id, repository_id=None, token_only=False,
cache_dir=None):
"""Generate a user access token via the device flow."""
out = sys.stderr if token_only else sys.stdout
# Try cached token first
if cache_dir:
cached = load_cached_token(cache_dir, client_id, TARGET_OWNER, TARGET_REPO)
if cached:
print(f"Using cached token for {TARGET_OWNER}/{TARGET_REPO}", file=out)
if token_only:
print(cached)
else:
print("=" * 60)
print(f"GITHUB_TOKEN={cached}")
print("=" * 60)
return
print("Requesting device code...", file=out)
resp = device_flow_request(
"https://github.com/login/device/code",
{"client_id": client_id},
)
device_code = resp["device_code"]
user_code = resp["user_code"]
verification_uri = resp["verification_uri"]
expires_in = resp["expires_in"]
interval = resp.get("interval", 5)
verbose(f"Device code: {device_code}")
verbose(f"Polling interval: {interval}s, expires in: {expires_in}s")
# Copy the user code to clipboard if possible
copied = _copy_to_clipboard(user_code)
clip_hint = " (copied to clipboard)" if copied else ""
print(file=out)
print("=" * 60, file=out)
print(f" Open: {verification_uri}", file=out)
print(f" Enter: {user_code}{clip_hint}", file=out)
print("=" * 60, file=out)
print(file=out)
print(f"Waiting for authorization (expires in {expires_in}s)...", file=out)
# Poll for the token
poll_count = 0
while True:
time.sleep(interval)
poll_count += 1
verbose(f"Polling attempt #{poll_count}...")
poll_params = {
"client_id": client_id,
"device_code": device_code,
"grant_type": "urn:ietf:params:oauth:grant-type:device_code",
}
if repository_id:
poll_params["repository_id"] = repository_id
token_resp = device_flow_request(
"https://github.com/login/oauth/access_token",
poll_params,
)
error = token_resp.get("error")
if error == "authorization_pending":
verbose("Status: authorization_pending")
continue
elif error == "slow_down":
interval = token_resp.get("interval", interval + 5)
verbose(f"Status: slow_down, new interval: {interval}s")
continue
elif error == "expired_token":
print("Error: Device code expired. Please try again.", file=sys.stderr)
sys.exit(1)
elif error == "access_denied":
print("Error: Authorization was denied by the user.", file=sys.stderr)
sys.exit(1)
elif error:
print(f"Error: {error} - {token_resp.get('error_description', '')}", file=sys.stderr)
sys.exit(1)
# Success
token = token_resp["access_token"]
token_type = token_resp.get("token_type", "bearer")
verbose(f"Token type: {token_type}")
verbose(f"Full token response keys: {list(token_resp.keys())}")
break
print("Authorization successful!", file=out)
print(file=out)
print("Verifying token...", file=out)
repo = verify_token(token, token_type="Bearer")
print(f"Verified: can access {repo['full_name']}", file=out)
print(file=out)
# Cache the token
if cache_dir:
save_cached_token(cache_dir, client_id, TARGET_OWNER, TARGET_REPO,
token)
if token_only:
print(token)
else:
print("=" * 60)
print(f"GITHUB_TOKEN={token}")
print("=" * 60)
# --- CLI ---
def main():
global VERBOSE
parser = argparse.ArgumentParser(
description="Generate repo-scoped GitHub user tokens via a GitHub App"
)
parser.add_argument("-v", "--verbose", action="store_true",
help="Enable verbose HTTP logging to stderr")
sub = parser.add_subparsers(dest="command")
sub.add_parser("create-app", help="Print instructions to create the GitHub App")
user_parser = sub.add_parser("user-token", help="Generate a user access token via device flow")
user_parser.add_argument("--client-id", required=True, help="GitHub App Client ID (Iv23li...)")
user_parser.add_argument("--repo", type=str, default=None,
help="Repository to scope the token to (owner/name, URL, or git@github.com:owner/name.git)")
user_parser.add_argument("--repository-id", type=str, default=None,
help="Numeric repo ID (use for private repos where --repo can't resolve)")
user_parser.add_argument("--token-only", action="store_true",
help="Print only the token to stdout (status messages go to stderr)")
user_parser.add_argument("--cache-dir", type=str, default=None,
help="Directory to cache tokens (default: no caching)")
args = parser.parse_args()
VERBOSE = args.verbose
if args.command == "create-app":
cmd_create_app()
elif args.command == "user-token":
global TARGET_OWNER, TARGET_REPO
repository_id = args.repository_id
if args.repo:
TARGET_OWNER, TARGET_REPO = parse_repo(args.repo)
if not repository_id:
repository_id = resolve_repo_id(TARGET_OWNER, TARGET_REPO)
cmd_user_token(args.client_id, repository_id=repository_id,
token_only=args.token_only,
cache_dir=args.cache_dir)
else:
parser.print_help()
sys.exit(1)
if __name__ == "__main__":
main()

View file

@ -1,69 +0,0 @@
"""
mitmproxy addon for agent VMs.
Intercepts HTTPS requests to configured domains and forwards them as plain
HTTP to the host-side credential proxy, which injects real auth tokens.
Requests to non-configured domains pass through to the real upstream unchanged.
Usage:
CREDENTIAL_PROXY_HOST=host.lima.internal \
CREDENTIAL_PROXY_PORT=12345 \
CREDENTIAL_PROXY_DOMAINS=api.anthropic.com,github.com,api.github.com \
mitmdump -s mitmproxy-addon.py
"""
import base64
import os
from mitmproxy import http
PROXY_HOST = os.environ.get("CREDENTIAL_PROXY_HOST", "host.lima.internal")
PROXY_PORT = int(os.environ.get("CREDENTIAL_PROXY_PORT", "0"))
PROXY_SECRET = os.environ.get("CREDENTIAL_PROXY_SECRET", "")
PROXY_DOMAINS = set(
d.strip()
for d in os.environ.get("CREDENTIAL_PROXY_DOMAINS", "").split(",")
if d.strip()
)
BLOCKED_DOMAINS = set(
d.strip()
for d in os.environ.get("BLOCKED_DOMAINS", "").split(",")
if d.strip()
)
def request(flow: http.HTTPFlow) -> None:
# Block requests to blacklisted domains
if flow.request.host in BLOCKED_DOMAINS:
flow.response = http.Response.make(403, b"Blocked by proxy policy")
return
# Only redirect configured domains to the credential proxy
if flow.request.host not in PROXY_DOMAINS:
return
# Add original destination info as headers
flow.request.headers["X-Original-Host"] = flow.request.host
flow.request.headers["X-Original-Port"] = str(flow.request.port)
flow.request.headers["X-Original-Scheme"] = flow.request.scheme
# Add shared secret for cross-VM isolation (standard proxy auth)
if PROXY_SECRET:
creds = base64.b64encode(f"_:{PROXY_SECRET}".encode()).decode()
flow.request.headers["Proxy-Authorization"] = f"Basic {creds}"
# Redirect to host-side credential proxy over plain HTTP
flow.request.scheme = "http"
flow.request.host = PROXY_HOST
flow.request.port = PROXY_PORT
def responseheaders(flow: http.HTTPFlow) -> None:
# Stream Server-Sent Events straight through to the client. mitmproxy
# buffers response bodies under `stream_large_bodies` by default; doing
# that to an SSE long-poll (e.g. the Claude Code worker event channel that
# carries web-typed prompts into the VM) holds the connection open forever
# with nothing reaching the client, so the client times out and retries.
ctype = flow.response.headers.get("content-type", "")
if "text/event-stream" in ctype.lower():
flow.response.stream = True

File diff suppressed because it is too large Load diff

View file

@ -1,140 +0,0 @@
#!/usr/bin/env python3
"""Full E2E test: host proxy + VM mitmproxy + gh CLI."""
import http.client, json, os, signal, subprocess, sys, time
SECRET = "e2e-secret-" + os.urandom(8).hex()
RULES = [
{"domain": "api.github.com", "path_prefix": "/repos/wirenboard/agent-vm",
"headers": {"Authorization": "token TOK_AGENT_VM"}},
{"domain": "api.github.com", "path_prefix": "/repos/wirenboard/wb-cloud-ui",
"headers": {"Authorization": "token TOK_CLOUD_UI"}},
{"domain": "api.github.com",
"headers": {"Authorization": "token TOK_FALLBACK"}},
]
# Start credential proxy
env = os.environ.copy()
env["CREDENTIAL_PROXY_RULES"] = json.dumps(RULES)
env["CREDENTIAL_PROXY_SECRET"] = SECRET
proc = subprocess.Popen([sys.executable, "credential-proxy.py"],
stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=env)
port_line = proc.stdout.readline().decode().strip()
PROXY_PORT = int(port_line)
print(f"Credential proxy on :{PROXY_PORT}, secret={SECRET[:20]}...")
# Copy addon to VM
subprocess.run(["bash", "-c",
"cat mitmproxy-addon.py | limactl shell test-e2e bash -c 'cat > ~/.mitmproxy/addon.py'"],
check=True)
# Kill any old mitmproxy
subprocess.run(["limactl", "shell", "test-e2e", "bash", "-c",
"pkill -f mitmdump; sleep 1"], capture_output=True)
# Start mitmproxy in VM
start_cmd = (
f"CREDENTIAL_PROXY_HOST=host.lima.internal "
f"CREDENTIAL_PROXY_PORT={PROXY_PORT} "
f"CREDENTIAL_PROXY_SECRET={SECRET} "
f"CREDENTIAL_PROXY_DOMAINS=api.github.com,github.com "
f"nohup mitmdump "
f" --listen-port 8080 "
f" --set connection_strategy=lazy "
f" -s ~/.mitmproxy/addon.py "
f" > /tmp/mitmproxy.log 2>&1 & "
f"sleep 3; echo READY"
)
r = subprocess.run(["limactl", "shell", "test-e2e", "bash", "-c", start_cmd],
capture_output=True, text=True, timeout=15)
print(f"mitmproxy: {r.stdout.strip()}")
results = []
# Test from VM using curl through mitmproxy
for name, path in [
("agent-vm repo", "/repos/wirenboard/agent-vm"),
("wb-cloud-ui repo", "/repos/wirenboard/wb-cloud-ui"),
("user (fallback)", "/user"),
]:
cmd = (
f"curl -s -w '\\nHTTP_CODE:%{{http_code}}' "
f"--proxy http://127.0.0.1:8080 "
f"https://api.github.com{path} 2>/dev/null"
)
r = subprocess.run(["limactl", "shell", "test-e2e", "bash", "-c", cmd],
capture_output=True, text=True, timeout=15)
http_code = ""
for line in r.stdout.strip().split('\n'):
if line.startswith("HTTP_CODE:"):
http_code = line.split(":")[1]
# 401 = fake token sent and rejected; 200 = public repo (no auth needed)
ok = http_code in ("401", "200")
print(f" {name}: HTTP {http_code} -> {'PASS' if ok else 'FAIL'}")
results.append(ok)
# Test gh CLI from VM
cmd = (
"export HTTPS_PROXY=http://127.0.0.1:8080; "
"export GH_TOKEN=placeholder; "
"gh api /repos/wirenboard/agent-vm --jq .full_name 2>&1; echo EXIT:$?"
)
r = subprocess.run(["limactl", "shell", "test-e2e", "bash", "-c", cmd],
capture_output=True, text=True, timeout=15)
gh_output = r.stdout.strip()
# gh gets 401 (Bad credentials) or succeeds (200 for public repo)
ok = "agent-vm" in gh_output or "Bad credentials" in gh_output
print(f" gh CLI: {gh_output[:60]} -> {'PASS' if ok else 'FAIL'}")
results.append(ok)
# Test wrong secret from VM (direct curl to proxy, bypassing mitmproxy)
cmd = (
f"curl -s -w '\\nHTTP_CODE:%{{http_code}}' "
f"-H 'X-Original-Host: api.github.com' "
f"-H 'X-Original-Port: 443' "
f"-H 'X-Original-Scheme: https' "
f"-H 'X-Proxy-Token: WRONG' "
f"http://host.lima.internal:{PROXY_PORT}/repos/wirenboard/agent-vm 2>/dev/null"
)
r = subprocess.run(["limactl", "shell", "test-e2e", "bash", "-c", cmd],
capture_output=True, text=True, timeout=10)
code = ""
for line in r.stdout.strip().split('\n'):
if line.startswith("HTTP_CODE:"):
code = line.split(":")[1]
ok = code == "403"
print(f" wrong secret: HTTP {code} -> {'PASS' if ok else 'FAIL'}")
results.append(ok)
# Test no secret from VM
cmd = (
f"curl -s -w '\\nHTTP_CODE:%{{http_code}}' "
f"-H 'X-Original-Host: api.github.com' "
f"-H 'X-Original-Port: 443' "
f"-H 'X-Original-Scheme: https' "
f"http://host.lima.internal:{PROXY_PORT}/repos/wirenboard/agent-vm 2>/dev/null"
)
r = subprocess.run(["limactl", "shell", "test-e2e", "bash", "-c", cmd],
capture_output=True, text=True, timeout=10)
code = ""
for line in r.stdout.strip().split('\n'):
if line.startswith("HTTP_CODE:"):
code = line.split(":")[1]
ok = code == "403"
print(f" no secret: HTTP {code} -> {'PASS' if ok else 'FAIL'}")
results.append(ok)
# Cleanup
proc.send_signal(signal.SIGTERM)
try:
proc.wait(timeout=3)
except subprocess.TimeoutExpired:
proc.kill()
proc.wait()
passed = sum(results)
total = len(results)
print(f"\nResults: {passed}/{total}")
if passed == total:
print("All E2E VM tests passed!")
sys.exit(0 if passed == total else 1)

1
vendor/microsandbox vendored Submodule

@ -0,0 +1 @@
Subproject commit 229db2ab4fc774758f4bd1898a75da19585430d2