diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 9650f77..a8a254f 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -21,7 +21,7 @@ talks to over sockets. 3. The daemon builds a `machine.Spec` (CPUs, memory, the OCI rootfs disk, the network, a vsock device, the serial log), asks the `machine` library for the right backend, and boots the VM. It also brings up gvproxy and, on - macOS, the agent proxy and ssh-agent proxy. + macOS, the agent proxy, the ssh-agent proxy, and the reverse-forward proxy. `clawk` / `clawk run ` then connect to the **in-guest pty-agent over vsock** (AF_VSOCK port 1024). Each attach spawns a fresh child in the guest and @@ -63,6 +63,17 @@ adds the filter hooks). Two attachment modes, abstracted by `machine.UserMode`: between gvproxy's unixgram socket and the guest's TAP across an IP-less L2 bridge (`machine/firecracker/usermode_linux.go`). +Port forwards go both ways, by two different mechanisms. Outbound +(`clawk forward add`) is a gvproxy binding on the host's loopback, fixed when +the VM starts. Inbound (`clawk forward add-reverse`) can't be: a guest process +dialling `127.0.0.1` reaches the guest's own loopback, with no route to the +host's. So the guest agent binds the port itself and tunnels each connection +over vsock to the daemon, which validates the requested port against the +sandbox's configured set before dialling the host service (`internal/revfwd` +for the protocol, `internal/cli/reverse_forward.go` for the host end). The +daemon pushes set changes down the same channel, so those edits apply to a +running guest. vz only — firecracker's vsock is one-way. + Live allow-list edits reach the running daemon over a control socket (`internal/vzdctl`); when the sandbox is down they apply on the next `up`. The same socket carries the VM lifecycle verbs: `clawk pause` / `resume` @@ -93,6 +104,7 @@ because it pins a vendored `gvisor-tap-vsock` fork; everything clawk-specific | `internal/template` | `clawk.mod` lexer + parser (typed `sandbox` / `policy` / `namespace` blocks). | | `internal/agentembed` | The in-guest binaries (clawk-init, pty-agent, time-sync), cross-compiled and injected into the rootfs. | | `internal/vsockproto` / `internal/vsockclient` | The host↔guest vsock framing and the host-side client. | +| `internal/revfwd` | Reverse-forward wire protocol (host loopback services exposed on the guest's loopback), mirrored in the guest agent. | | `internal/netfilter` | Egress allow-list (IPs/CIDRs/domains, DNS-aware) consumed by gvproxy. | | `internal/vzdctl` | Daemon control socket (live policy edits, denial ledger, VM pause/resume/suspend). | | `internal/worktree` / `internal/pr` | Multi-repo branch coordination and PR creation. | diff --git a/CHANGELOG.md b/CHANGELOG.md index 423952b..3493be4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,27 @@ tagged. ### Added +- **Reverse port forwarding: host loopback services, reachable in the guest.** + `clawk forward add-reverse 63342` makes whatever is bound to + `127.0.0.1:63342` on your Mac answer at the same address inside the sandbox + (`5432:15432` maps across ports, host-side first, same as `forward add`). + Allow-listing couldn't do this — `127.0.0.1` in the guest is the guest's own + loopback — so the guest agent binds the port and tunnels each connection to + the daemon over vsock, which dials the host service. Only the ports you list + are reachable; the rest of your loopback isn't. + + Unlike outbound forwards these apply to a running sandbox immediately, which + matters for the case that motivated it: the Claude Code IDE plugins + advertise a per-window websocket port in `~/.claude/ide/.lock`, so + reconnecting after an IDE restart is one `add-reverse`, not a VM cycle. + Share `~/.claude/ide` into the guest and `/ide` works from inside the + sandbox — recipe in [docs/networking.md](docs/networking.md#recipe-the-claude-code-ide-plugin). + + Declarable in `clawk.mod` as a `reverse` entry inside `forwards ( … )`. + `clawk status` and `forward list --json` show both directions. vz only: + firecracker's vsock is one-way, and the CLI says so rather than silently + doing nothing. + - **Linux firecracker sandboxes boot without sudo.** Each sandbox's network now lives in its own unprivileged user + network namespace, where clawk has `CAP_NET_ADMIN` over its own bridge and TAPs without asking the host for diff --git a/README.md b/README.md index f86edec..0795a9a 100644 --- a/README.md +++ b/README.md @@ -235,6 +235,8 @@ Anthropic, …) are pre-allowed, and the filter is DNS-aware, so allowing clawk network allow my-project api.stripe.com '*.internal.mycorp.com' 10.0.0.5 clawk network denials my-project # what the agent tried that got blocked clawk forward add my-project 3000 # localhost:3000 → the guest's dev server +clawk forward add-reverse my-project 63342 # and the other way: a service on YOUR + # localhost, reachable inside the guest ``` Denials are recorded by the *hostname the guest resolved*, so `clawk network diff --git a/docs/commands.md b/docs/commands.md index b3f4e0d..06e2d90 100644 --- a/docs/commands.md +++ b/docs/commands.md @@ -131,7 +131,7 @@ without notice. | Provider | Host | Notes | |--------------------|-------|-----------------------------------------------------------------------| | `vz` (default) | macOS | Apple Virtualization.framework; no sudo. Live-mounts your worktree. | -| `firecracker` (experimental) | Linux | KVM microVM; no sudo on hosts that allow unprivileged user namespaces. Carries the worktree on its own disk (host edits don't propagate live), and skips host-file push, ssh-agent forwarding, and per-phase hooks today. | +| `firecracker` (experimental) | Linux | KVM microVM; no sudo on hosts that allow unprivileged user namespaces. Carries the worktree on its own disk (host edits don't propagate live), and skips host-file push, ssh-agent forwarding, reverse port forwards, and per-phase hooks today. | Pick one with `--provider`; the choice persists with the sandbox. Both run the same OCI rootfs, vsock agent, and egress allow-list — see diff --git a/docs/configuration.md b/docs/configuration.md index 05a1963..524a44e 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -49,6 +49,7 @@ sandbox my-project ( forwards ( 3000 5432:5432 + reverse 63342 # the host's localhost:63342, inside the guest ) files ( @@ -117,7 +118,11 @@ sandbox my-project ( - `network ( … )` — egress policy: `allow` / `deny` a domain or `ip `, plus `use …` chains — see [Networking](networking.md#policies-and-use-chains). -- `forwards ( … )` — port forwards (`PORT` or `HOST:GUEST`). +- `forwards ( … )` — port forwards (`PORT` or `HOST:GUEST`). An entry + prefixed with `reverse` points the other way: a service on the host's + `127.0.0.1` becomes reachable at the same address inside the guest. Same + host-first spelling either way — see + [Networking](networking.md#reverse-forwarding-host-loopback--guest). - `env ( … )` — environment variables to export inside the VM. Secret *values* come from your shell at boot and are never written to disk on the host; only names, defaults, and literals live in the file. Each entry uses diff --git a/docs/linux-quickstart.md b/docs/linux-quickstart.md index 5127ebe..99707c8 100644 --- a/docs/linux-quickstart.md +++ b/docs/linux-quickstart.md @@ -165,6 +165,9 @@ The egress allow-list is enforced in gvproxy on the host, so the guest cannot turn it off from the inside — that holds in rootless mode too (verified: a blocked host is refused and logged, `acl: denied example.com`). +Reverse forwards (`clawk forward add-reverse`) are vz-only: they tunnel over +a host-side vsock listener, and firecracker's vsock only runs the other way. + --- ## 5. Limits diff --git a/docs/networking.md b/docs/networking.md index 58fbb1d..40b9f6e 100644 --- a/docs/networking.md +++ b/docs/networking.md @@ -87,3 +87,62 @@ interface is not visible to the host. Note that an idle-stopped VM's port forwards go away until the next boot — give a sandbox that must keep serving `idle_timeout off` (see [Commands & resource usage](commands.md#resource-usage)). + +### Reverse forwarding (host loopback → guest) + +The other direction: a service bound to `127.0.0.1` on your Mac, reachable +at the *same address* inside the guest. Allow-listing its IP doesn't help — +`127.0.0.1` in the guest is the guest's own loopback, and there is no route +from there to yours. + +```sh +clawk forward add-reverse 63342 # guest 127.0.0.1:63342 → host 127.0.0.1:63342 +clawk forward add-reverse 5432:15432 # guest 127.0.0.1:15432 → host 127.0.0.1:5432 +clawk forward remove-reverse 63342 +``` + +Specs read host-side first in both directions, so `5432:15432` names the +same pair of ports whichever verb you use — only who dials whom changes. + +Two things differ from outbound forwards: + +- **They apply immediately.** Outbound forwards are a gvproxy binding fixed + at VM start; reverse forwards are tunnelled over vsock by the daemon and + pushed to the running guest, so no `down`/`up` cycle is needed. +- **Only the listed ports are reachable.** The guest names a port, never an + address, and the host refuses one that isn't configured — so this opens + exactly the holes you asked for, not your whole loopback. + +vz only. firecracker's vsock is one-way (guest listens, host dials), so +there is nothing for the guest to connect back through; the CLI says so +rather than silently doing nothing. + +Reverse forwards can also be declared in `clawk.mod` — see +[Configuration](configuration.md#reference). + +### Recipe: the Claude Code IDE plugin + +The JetBrains and VS Code plugins run a websocket server on the host's +loopback and advertise it in `~/.claude/ide/.lock`. A `claude` running +inside a sandbox needs two things to find it — the lock file, and a route to +the port: + +```sh +# 1. share the host's lock-file dir into the guest (clawk.mod, or clawk apply) +# shares ( +# ~/.claude/ide /home/agent/.claude/ide ro +# ) + +# 2. reverse-forward the port the lock file names +ls ~/.claude/ide # 63342.lock +clawk forward add-reverse my-project 63342 +``` + +Then `/ide` inside the sandbox connects as it would on the host. The port is +per-IDE-window and changes when the IDE restarts; because reverse forwards +apply live, re-running `add-reverse` with the new port is enough — no +sandbox restart. + +Note that the guest's `~/.claude` is the sandbox's own state directory, not +your host `~/.claude`; the share above is what puts the host's lock files +where the guest's `claude` looks. diff --git a/internal/agentembed/main.go.in b/internal/agentembed/main.go.in index b5e9cec..d04a1fe 100644 --- a/internal/agentembed/main.go.in +++ b/internal/agentembed/main.go.in @@ -18,6 +18,7 @@ package main import ( + "bufio" "context" "encoding/binary" "encoding/json" @@ -623,6 +624,14 @@ func main() { // back to giving the guest its full ceiling. go runMemReporter(logger) + // Reverse port forwarder: holds a control connection to the host, + // binds 127.0.0.1: in here for every reverse forward the host + // publishes, and tunnels each connection back over vsock. Idle unless + // the sandbox has reverse forwards configured; retries forever if the + // host end isn't there (firecracker, or a host that hasn't started its + // proxy yet). + go runReverseForwarder(logger) + var wg sync.WaitGroup for { conn, err := listener.Accept() @@ -716,6 +725,269 @@ func forwardSSHAgent(guest net.Conn, logger *log.Logger) { <-done } +// ──────────────────────────────────────────────────────────────────────── +// Reverse port forwarding +// +// The mirror image of `clawk forward add`: a service on the host's +// 127.0.0.1 made reachable at 127.0.0.1 in here. gvproxy can't do it — +// loopback in the guest is the guest's own — so we bind the port locally +// and tunnel each connection to the host over vsock. +// +// Wire protocol kept in lock-step with internal/revfwd in the parent repo; +// any change there must be mirrored here. Two connection kinds, both +// guest-initiated, both starting with one JSON greeting line: +// +// op=control the host answers with the full forward set now and again +// on every change, so `clawk forward add-reverse` applies to +// a running sandbox. +// op=connect the host validates the requested port, answers ok, then +// pipes bytes to 127.0.0.1: on the Mac. +// ──────────────────────────────────────────────────────────────────────── + +const ( + // reverseForwardVSockPort is the host-side port serving both kinds. + // Mirrors revfwd.VSockPort. Distinct from the pty agent (1024), + // time-sync (1025), ssh-agent (1026), and mem-report (1027). + reverseForwardVSockPort = 1028 + + // reverseForwardProtoVersion mirrors revfwd.ProtoVersion. + reverseForwardProtoVersion = 1 + + // reverseForwardMaxLine mirrors revfwd.MaxLineBytes. + reverseForwardMaxLine = 64 * 1024 + + // reverseForwardMinBackoff / reverseForwardMaxBackoff bound the retry + // delay when the host end isn't answering. It legitimately isn't for + // the first moments of every boot (the daemon starts its proxy after + // the VM is up) and permanently under firecracker, whose vsock is + // one-way — so this loop has to be patient without being chatty. + reverseForwardMinBackoff = 1 * time.Second + reverseForwardMaxBackoff = 30 * time.Second + + // reverseForwardStableSession is how long a control connection must + // last to count as healthy and reset the backoff. Shorter than that + // and we treat it as a failed attempt, so a host that accepts and + // immediately hangs up can't put us in a tight redial loop. + reverseForwardStableSession = 30 * time.Second +) + +type revfwdGreeting struct { + Op string `json:"op"` + V int `json:"v"` + Port int `json:"port,omitempty"` +} + +type revfwdSnapshot struct { + Forwards []revfwdForward `json:"forwards"` +} + +type revfwdForward struct { + GuestPort int `json:"guest"` + HostPort int `json:"host"` +} + +type revfwdReply struct { + OK bool `json:"ok"` + Error string `json:"error,omitempty"` +} + +// reverseForwarder owns the guest-side listeners, keyed by the guest port +// each is bound to. +type reverseForwarder struct { + log *log.Logger + + mu sync.Mutex + listeners map[int]*reverseListener +} + +type reverseListener struct { + hostPort int + ln net.Listener +} + +// runReverseForwarder keeps the control connection to the host alive, +// applying every set the host publishes. Never returns. +func runReverseForwarder(logger *log.Logger) { + f := &reverseForwarder{log: logger, listeners: make(map[int]*reverseListener)} + backoff := reverseForwardMinBackoff + quiet := false + for { + start := time.Now() + err := f.controlSession() + // Every listener exists to serve connections through the host, so + // once the host is gone they must go too: a port left bound would + // accept a client and then fail it, which reads as a broken service + // rather than an absent one. + f.apply(nil) + + if time.Since(start) >= reverseForwardStableSession { + backoff, quiet = reverseForwardMinBackoff, false + } + if err != nil && !quiet { + logger.Printf("reverse-forward: %v (retrying, quietly from here)", err) + quiet = true + } + time.Sleep(backoff) + if backoff *= 2; backoff > reverseForwardMaxBackoff { + backoff = reverseForwardMaxBackoff + } + } +} + +// controlSession dials the host and applies snapshots until the connection +// drops. A clean EOF (the host shutting down) returns nil. +func (f *reverseForwarder) controlSession() error { + conn, err := vsock.Dial(vsockHostCID, reverseForwardVSockPort, nil) + if err != nil { + return fmt.Errorf("vsock dial host:%d: %w", reverseForwardVSockPort, err) + } + defer conn.Close() + + if err := revfwdWriteLine(conn, revfwdGreeting{ + Op: "control", V: reverseForwardProtoVersion, + }); err != nil { + return fmt.Errorf("sending control greeting: %w", err) + } + r := bufio.NewReaderSize(conn, reverseForwardMaxLine) + for { + var snap revfwdSnapshot + if err := revfwdReadLine(r, &snap); err != nil { + if errors.Is(err, io.EOF) || isClosedConn(err) { + return nil + } + return fmt.Errorf("reading forward set: %w", err) + } + f.apply(snap.Forwards) + } +} + +// apply reconciles the bound listeners with want, which is always the +// complete desired set. Listeners whose host port changed are rebound; +// ports that fail to bind are logged and retried on the next snapshot. +func (f *reverseForwarder) apply(want []revfwdForward) { + desired := make(map[int]int, len(want)) // guest port -> host port + for _, fw := range want { + if !validPort(fw.GuestPort) || !validPort(fw.HostPort) { + f.log.Printf("reverse-forward: ignoring out-of-range mapping %d -> %d", + fw.GuestPort, fw.HostPort) + continue + } + desired[fw.GuestPort] = fw.HostPort + } + + f.mu.Lock() + defer f.mu.Unlock() + + for guestPort, l := range f.listeners { + if hostPort, ok := desired[guestPort]; ok && hostPort == l.hostPort { + continue + } + _ = l.ln.Close() + delete(f.listeners, guestPort) + } + for guestPort, hostPort := range desired { + if _, ok := f.listeners[guestPort]; ok { + continue + } + // 127.0.0.1 rather than 0.0.0.0 on purpose: the point is to make + // the host service look local, and binding the guest's routable + // interface would re-export it to whatever else can reach the VM. + ln, err := net.Listen("tcp", fmt.Sprintf("127.0.0.1:%d", guestPort)) + if err != nil { + f.log.Printf("reverse-forward: bind 127.0.0.1:%d: %v", guestPort, err) + continue + } + l := &reverseListener{hostPort: hostPort, ln: ln} + f.listeners[guestPort] = l + f.log.Printf("reverse-forward: 127.0.0.1:%d -> host 127.0.0.1:%d", guestPort, hostPort) + go f.acceptLoop(l) + } +} + +func (f *reverseForwarder) acceptLoop(l *reverseListener) { + for { + conn, err := l.ln.Accept() + if err != nil { + // Closed is the normal path: apply() dropped this forward. + if !errors.Is(err, net.ErrClosed) { + f.log.Printf("reverse-forward: accept: %v", err) + } + return + } + go f.forward(conn, l.hostPort) + } +} + +// forward bridges one guest-local connection to the host service behind +// hostPort. +func (f *reverseForwarder) forward(local net.Conn, hostPort int) { + defer local.Close() + + host, err := vsock.Dial(vsockHostCID, reverseForwardVSockPort, nil) + if err != nil { + f.log.Printf("reverse-forward: vsock dial host:%d: %v", reverseForwardVSockPort, err) + return + } + defer host.Close() + + if err := revfwdWriteLine(host, revfwdGreeting{ + Op: "connect", V: reverseForwardProtoVersion, Port: hostPort, + }); err != nil { + f.log.Printf("reverse-forward: sending connect greeting: %v", err) + return + } + r := bufio.NewReaderSize(host, reverseForwardMaxLine) + var reply revfwdReply + if err := revfwdReadLine(r, &reply); err != nil { + f.log.Printf("reverse-forward: reading connect reply: %v", err) + return + } + if !reply.OK { + f.log.Printf("reverse-forward: host refused port %d: %s", hostPort, reply.Error) + return + } + + // Read the host side through r, not host: the reply reader may already + // hold bytes the service sent immediately after the handshake. + // + // First direction to end tears down both — the host's vsock conn has no + // CloseWrite (vz's *VirtioSocketConnection doesn't implement it), so a + // half-close would never reach this end. Same shape as + // forwardSSHAgent above. + done := make(chan struct{}, 2) + go func() { _, _ = io.Copy(host, local); done <- struct{}{} }() + go func() { _, _ = io.Copy(local, r); done <- struct{}{} }() + <-done +} + +func validPort(p int) bool { return p >= 1 && p <= 65535 } + +// revfwdWriteLine mirrors revfwd.WriteLine. +func revfwdWriteLine(w io.Writer, v any) error { + b, err := json.Marshal(v) + if err != nil { + return err + } + if len(b)+1 > reverseForwardMaxLine { + return errors.New("control line too long") + } + _, err = w.Write(append(b, '\n')) + return err +} + +// revfwdReadLine mirrors revfwd.ReadLine. The reader must be reused for any +// payload that follows, or bytes buffered past the newline are lost. +func revfwdReadLine(r *bufio.Reader, v any) error { + line, err := r.ReadSlice('\n') + if errors.Is(err, bufio.ErrBufferFull) { + return errors.New("control line too long") + } + if err != nil { + return err + } + return json.Unmarshal(line, v) +} + // ──────────────────────────────────────────────────────────────────────── // Memory reporter // diff --git a/internal/agentembed/revfwd_lockstep_test.go b/internal/agentembed/revfwd_lockstep_test.go new file mode 100644 index 0000000..27c95a0 --- /dev/null +++ b/internal/agentembed/revfwd_lockstep_test.go @@ -0,0 +1,70 @@ +package agentembed + +import ( + "fmt" + "reflect" + "testing" + + "github.com/clawkwork/clawk/internal/revfwd" + "github.com/stretchr/testify/require" +) + +// The guest agent builds standalone inside the guest, so it can't import +// internal/revfwd — the reverse-forward protocol is transcribed into +// main.go.in by hand. Nothing else notices when the two drift: a renamed +// JSON field or a moved port compiles fine on both sides and fails only as +// a sandbox where reverse forwards silently never appear. +// +// So: check the parts of the protocol that have to match byte-for-byte +// against the parent-module definition, deriving them from the structs +// rather than restating them, so this test can't drift either. +func TestReverseForwardProtocolMirroredInAgent(t *testing.T) { + src := string(AgentMainGo) + + require.Contains(t, src, fmt.Sprintf("reverseForwardVSockPort = %d", revfwd.VSockPort), + "guest dials a different vsock port than the host listens on") + require.Contains(t, src, fmt.Sprintf("reverseForwardProtoVersion = %d", revfwd.ProtoVersion), + "guest announces a protocol version the host will hang up on") + + // Op values are string literals on the wire, sent by the guest and + // switched on by the host. + for _, op := range []string{revfwd.OpControl, revfwd.OpConnect} { + require.Contains(t, src, fmt.Sprintf("Op: %q", op), + "guest never sends the %q greeting", op) + } + + for _, typ := range []any{ + revfwd.Greeting{}, revfwd.Snapshot{}, revfwd.Forward{}, revfwd.ConnectReply{}, + } { + rt := reflect.TypeOf(typ) + for i := range rt.NumField() { + tag, ok := rt.Field(i).Tag.Lookup("json") + if !ok { + continue + } + want := fmt.Sprintf("`json:%q`", tag) + require.Contains(t, src, want, + "guest is missing the %s.%s field tag %s", + rt.Name(), rt.Field(i).Name, want) + } + } +} + +// The guest's own fixed ports must not collide with each other — a +// duplicate would make one service unreachable in a way that looks like +// the host end being down. +func TestGuestVSockPortsAreDistinct(t *testing.T) { + src := string(AgentMainGo) + seen := map[string]string{} + for _, decl := range []struct{ name, port string }{ + {"sshAgentVSockPort", "1026"}, + {"memReportVSockPort", "1027"}, + {"reverseForwardVSockPort", fmt.Sprint(revfwd.VSockPort)}, + } { + require.Contains(t, src, decl.name+" = "+decl.port, + "%s moved; update this test and every mirror of it", decl.name) + require.NotContains(t, seen, decl.port, + "port %s claimed by both %s and %s", decl.port, seen[decl.port], decl.name) + seen[decl.port] = decl.name + } +} diff --git a/internal/cli/daemon.go b/internal/cli/daemon.go index fcfc45b..2f57103 100644 --- a/internal/cli/daemon.go +++ b/internal/cli/daemon.go @@ -464,12 +464,21 @@ func effectiveUseForLog(sb *config.Sandbox) []string { return effectiveUse(ns, sb) } +// reverseForwardSink publishes a reverse-forward set to the in-guest agent. +// An interface so controlHandlers stays platform-neutral: the only +// implementation is the darwin reverseProxy, and the firecracker daemon +// passes nil (its backend can't accept guest-initiated vsock connections). +type reverseForwardSink interface { + Set([]config.PortForward) +} + // controlHandlers builds the control-socket callbacks shared by both VM // daemons: the denial ledger, a live network-policy reload from the store, -// the VM lifecycle surface (pause/resume/suspend), and (when the allow -// list has one) the interactive gate. -func controlHandlers(sb *config.Sandbox, allow *netfilter.AllowList, lc *vmLifecycle, logger *log.Logger) vzdctl.Handlers { - return vzdctl.Handlers{ +// the VM lifecycle surface (pause/resume/suspend), reverse-forward reloads +// when the daemon has a sink for them, and (when the allow list has one) +// the interactive gate. +func controlHandlers(sb *config.Sandbox, allow *netfilter.AllowList, lc *vmLifecycle, rev reverseForwardSink, logger *log.Logger) vzdctl.Handlers { + h := vzdctl.Handlers{ Denials: allow.Denials, Lifecycle: lc.lifecycleHandlers(), Reload: func() error { @@ -488,6 +497,18 @@ func controlHandlers(sb *config.Sandbox, allow *netfilter.AllowList, lc *vmLifec }, Gate: allow.Gate(), } + if rev != nil { + h.ReloadForwards = func() error { + cur, err := store.Load(sb.Name) + if err != nil { + return fmt.Errorf("reloading sandbox record: %w", err) + } + rev.Set(cur.ReverseForwards) + logger.Printf("reverse forwards reloaded: %d", len(cur.ReverseForwards)) + return nil + } + } + return h } // persistAlwaysAllow records an interactive "allow always" decision in the diff --git a/internal/cli/fcd.go b/internal/cli/fcd.go index 22ae496..a4fcfc1 100644 --- a/internal/cli/fcd.go +++ b/internal/cli/fcd.go @@ -87,7 +87,11 @@ func runFcd(_ *cobra.Command, args []string) (retErr error) { // (`clawk network allow` with no down/up), read the denial ledger, and // drive the VM lifecycle (`clawk pause/resume/snapshot`). // Best-effort — without it, edits apply on the next up, as before. - ctl, err := vzdctl.Start(vzdctl.SocketPath(vmDir), controlHandlers(sb, allow, lc, logger)) + // + // No reverse-forward sink: firecracker's vsock is one-way (guest listens, + // host dials), so there is nothing for the guest agent to dial. The + // endpoint 404s and the CLI says so. + ctl, err := vzdctl.Start(vzdctl.SocketPath(vmDir), controlHandlers(sb, allow, lc, nil, logger)) if err != nil { logger.Printf("control socket: disabled (%v) — network edits apply on next up", err) } else { diff --git a/internal/cli/forward.go b/internal/cli/forward.go index 3ec79e4..2659d8d 100644 --- a/internal/cli/forward.go +++ b/internal/cli/forward.go @@ -1,13 +1,16 @@ package cli import ( + "context" "encoding/json" "errors" "fmt" "strconv" "strings" + "time" "github.com/clawkwork/clawk/internal/config" + "github.com/clawkwork/clawk/internal/vzdctl" "github.com/spf13/cobra" ) @@ -17,6 +20,8 @@ func init() { rootCmd.AddCommand(forwardCmd) forwardCmd.AddCommand(forwardAddCmd) forwardCmd.AddCommand(forwardRemoveCmd) + forwardCmd.AddCommand(forwardAddReverseCmd) + forwardCmd.AddCommand(forwardRemoveReverseCmd) forwardCmd.AddCommand(forwardListCmd) forwardListCmd.Flags().BoolVar(&forwardListJSON, "json", false, "emit JSON (the only supported mode; human path is 'clawk status')") @@ -25,7 +30,11 @@ func init() { var forwardCmd = &cobra.Command{ Use: "forward", Aliases: []string{"fwd"}, - Short: "Manage host-to-guest port forwards (changes apply on next up)", + Short: "Manage port forwards in both directions", + Long: `Two directions, same HOST:GUEST spelling: + + add a guest service, on the host's localhost (applies on next up) + add-reverse a host service, on the guest's localhost (applies immediately)`, } var forwardAddCmd = &cobra.Command{ @@ -97,6 +106,146 @@ var forwardRemoveCmd = &cobra.Command{ }, } +var forwardAddReverseCmd = &cobra.Command{ + ValidArgsFunction: completeSandboxNames, + Use: "add-reverse [port-spec...]", + Aliases: []string{"add-r"}, + Short: "Expose a host loopback service on the guest's loopback", + Long: `The inbound counterpart of 'forward add': a service bound to +127.0.0.1 on the host becomes reachable at the SAME address inside the +guest, which is what tools that assume a shared loopback need (the +JetBrains/VS Code Claude Code plugin's IDE websocket, a local API mock, +a database bound to loopback). + +Port specs read host-side first, exactly like 'forward add': + 12345 — guest 127.0.0.1:12345 reaches host 127.0.0.1:12345 + 5432:15432 — guest 127.0.0.1:15432 reaches host 127.0.0.1:5432 + +Unlike outbound forwards these apply to a running sandbox immediately — +no down/up cycle. Only the ports listed here are reachable; the guest +cannot dial the rest of the host's loopback. + +vz (macOS) only: firecracker's vsock is one-way, so the guest has no +channel to connect back through.`, + Args: cobra.MinimumNArgs(2), + RunE: func(cmd *cobra.Command, args []string) error { + sb, err := store.Load(args[0]) + if err != nil { + return err + } + changed := false + for _, spec := range args[1:] { + fwd, err := parsePortSpec(spec) + if err != nil { + return err + } + if forwardExists(sb.ReverseForwards, fwd) { + fmt.Fprintf(cmd.OutOrStdout(), " (already reverse-forwarded: %s)\n", + describeReverse(fwd)) + continue + } + // A guest port can only be bound once, so a second spec claiming + // it would silently lose to the first inside the guest. Reject it + // here where we can name both. + if prev, dup := reverseByGuestPort(sb.ReverseForwards, fwd.GuestPort); dup { + return fmt.Errorf( + "guest port %d is already reverse-forwarded to host port %d — remove that first", + fwd.GuestPort, prev.HostPort) + } + sb.ReverseForwards = append(sb.ReverseForwards, fwd) + changed = true + fmt.Fprintf(cmd.OutOrStdout(), "Reverse forward added: %s\n", describeReverse(fwd)) + } + if err := store.Save(sb); err != nil { + return err + } + if changed { + applyReverseForwards(cmd, sb.Name) + } + return nil + }, +} + +var forwardRemoveReverseCmd = &cobra.Command{ + ValidArgsFunction: completeSandboxNames, + Use: "remove-reverse [port-spec...]", + Aliases: []string{"rm-reverse", "rm-r"}, + Short: "Remove a host-to-guest-loopback reverse forward", + Args: cobra.MinimumNArgs(2), + RunE: func(cmd *cobra.Command, args []string) error { + sb, err := store.Load(args[0]) + if err != nil { + return err + } + drop := make(map[config.PortForward]bool) + for _, spec := range args[1:] { + fwd, err := parsePortSpec(spec) + if err != nil { + return err + } + drop[fwd] = true + } + var kept []config.PortForward + for _, f := range sb.ReverseForwards { + if drop[f] { + fmt.Fprintf(cmd.OutOrStdout(), "Reverse forward removed: %s\n", describeReverse(f)) + } else { + kept = append(kept, f) + } + } + changed := len(kept) != len(sb.ReverseForwards) + sb.ReverseForwards = kept + if err := store.Save(sb); err != nil { + return err + } + if changed { + applyReverseForwards(cmd, sb.Name) + } + return nil + }, +} + +// applyReverseForwards pushes the just-saved reverse-forward set into the +// running daemon, which relays it to the in-guest agent. Reports what +// happened but never fails the command: the store is already updated, so +// the worst case is that the edit lands on the next boot. Mirrors +// applyNetworkPolicy. +func applyReverseForwards(cmd *cobra.Command, name string) { + ctx, cancel := context.WithTimeout(cmd.Context(), 5*time.Second) + defer cancel() + err := vzdctl.NewClient(vzdctl.SocketPath(store.VMDir(name))).ReloadForwards(ctx) + switch { + case err == nil: + fmt.Fprintln(cmd.OutOrStdout(), "Applied to running sandbox.") + case errors.Is(err, vzdctl.ErrNotRunning): + fmt.Fprintln(cmd.OutOrStdout(), "Sandbox not running — applies on next 'clawk up'.") + case errors.Is(err, vzdctl.ErrReverseForwardsUnsupported): + fmt.Fprintln(cmd.OutOrStdout(), + "Sandbox is running an older daemon — restart it ('clawk down && clawk up') to apply.") + default: + fmt.Fprintf(cmd.ErrOrStderr(), + "clawk: live apply failed (%v) — applies on next 'clawk up'\n", err) + } +} + +// describeReverse spells out a reverse forward in the direction traffic +// actually flows, because the HOST:GUEST spec alone doesn't say which end +// listens. Worth the words: getting the direction backwards is the whole +// confusion this command exists to resolve. +func describeReverse(f config.PortForward) string { + return fmt.Sprintf("guest 127.0.0.1:%d → host 127.0.0.1:%d", f.GuestPort, f.HostPort) +} + +// reverseByGuestPort finds an existing reverse forward bound to guestPort. +func reverseByGuestPort(fs []config.PortForward, guestPort int) (config.PortForward, bool) { + for _, f := range fs { + if f.GuestPort == guestPort { + return f, true + } + } + return config.PortForward{}, false +} + var forwardListCmd = &cobra.Command{ ValidArgsFunction: completeSandboxNames, Use: "list ", @@ -121,12 +270,20 @@ var forwardListCmd = &cobra.Command{ Schema string `json:"schema"` Sandbox string `json:"sandbox"` Forwards []statusJSONForward `json:"forwards"` + // Additive: absent on older clawk, so a script that only + // knows "forwards" keeps parsing unchanged. + ReverseForwards []statusJSONForward `json:"reverse_forwards,omitempty"` }{Schema: "1", Sandbox: sb.Name} for _, f := range sb.Forwards { out.Forwards = append(out.Forwards, statusJSONForward{ HostPort: f.HostPort, GuestPort: f.GuestPort, }) } + for _, f := range sb.ReverseForwards { + out.ReverseForwards = append(out.ReverseForwards, statusJSONForward{ + HostPort: f.HostPort, GuestPort: f.GuestPort, + }) + } enc := json.NewEncoder(cmd.OutOrStdout()) enc.SetIndent("", " ") return enc.Encode(out) diff --git a/internal/cli/forward_test.go b/internal/cli/forward_test.go new file mode 100644 index 0000000..bb9e3a4 --- /dev/null +++ b/internal/cli/forward_test.go @@ -0,0 +1,123 @@ +package cli + +import ( + "encoding/json" + "testing" + + "github.com/clawkwork/clawk/internal/config" + "github.com/stretchr/testify/require" +) + +func TestForwardAddReverse(t *testing.T) { + s, _ := setupTest(t) + require.NoError(t, s.Save(&config.Sandbox{ + Name: "rev", Provider: config.ProviderVZ, VMState: config.VMStateStopped, + })) + + out, err := executeCommand("forward", "add-reverse", "rev", "63342", "5432:15432") + require.NoError(t, err) + // The direction is the whole point of the verb, so it has to be in the + // output — a bare "63342:63342" wouldn't say which end listens. + require.Contains(t, out, "guest 127.0.0.1:63342 → host 127.0.0.1:63342") + require.Contains(t, out, "guest 127.0.0.1:15432 → host 127.0.0.1:5432") + // Nothing is running under the mock provider, so the live push reports + // that instead of claiming an apply that never happened. + require.Contains(t, out, "applies on next 'clawk up'") + + sb, err := s.Load("rev") + require.NoError(t, err) + require.Equal(t, []config.PortForward{ + {HostPort: 63342, GuestPort: 63342}, + {HostPort: 5432, GuestPort: 15432}, + }, sb.ReverseForwards) + require.Empty(t, sb.Forwards, "reverse forwards must not leak into the outbound list") +} + +func TestForwardAddReverseIsIdempotent(t *testing.T) { + s, _ := setupTest(t) + require.NoError(t, s.Save(&config.Sandbox{ + Name: "rev", Provider: config.ProviderVZ, VMState: config.VMStateStopped, + })) + + _, err := executeCommand("forward", "add-reverse", "rev", "63342") + require.NoError(t, err) + out, err := executeCommand("forward", "add-reverse", "rev", "63342") + require.NoError(t, err) + require.Contains(t, out, "already reverse-forwarded") + + sb, err := s.Load("rev") + require.NoError(t, err) + require.Len(t, sb.ReverseForwards, 1) +} + +// Only one thing can bind a given guest port, so a second mapping onto it +// has to be refused rather than silently losing inside the guest. +func TestForwardAddReverseRejectsGuestPortClash(t *testing.T) { + s, _ := setupTest(t) + require.NoError(t, s.Save(&config.Sandbox{ + Name: "rev", Provider: config.ProviderVZ, VMState: config.VMStateStopped, + })) + + _, err := executeCommand("forward", "add-reverse", "rev", "5432:15432") + require.NoError(t, err) + _, err = executeCommand("forward", "add-reverse", "rev", "6432:15432") + require.Error(t, err) + require.Contains(t, err.Error(), "guest port 15432 is already reverse-forwarded") + + sb, err := s.Load("rev") + require.NoError(t, err) + require.Len(t, sb.ReverseForwards, 1) +} + +func TestForwardRemoveReverse(t *testing.T) { + s, _ := setupTest(t) + require.NoError(t, s.Save(&config.Sandbox{ + Name: "rev", Provider: config.ProviderVZ, VMState: config.VMStateStopped, + ReverseForwards: []config.PortForward{ + {HostPort: 63342, GuestPort: 63342}, + {HostPort: 5432, GuestPort: 15432}, + }, + })) + + out, err := executeCommand("forward", "remove-reverse", "rev", "63342") + require.NoError(t, err) + require.Contains(t, out, "Reverse forward removed") + + sb, err := s.Load("rev") + require.NoError(t, err) + require.Equal(t, []config.PortForward{{HostPort: 5432, GuestPort: 15432}}, sb.ReverseForwards) +} + +// Both directions ride the same record, and `forward list --json` is the +// scriptable view of it — they must stay distinguishable there. +func TestForwardListJSONSeparatesDirections(t *testing.T) { + s, _ := setupTest(t) + require.NoError(t, s.Save(&config.Sandbox{ + Name: "rev", Provider: config.ProviderVZ, VMState: config.VMStateStopped, + Forwards: []config.PortForward{{HostPort: 3000, GuestPort: 3000}}, + ReverseForwards: []config.PortForward{{HostPort: 5432, GuestPort: 15432}}, + })) + + out, err := executeCommand("forward", "list", "rev", "--json") + require.NoError(t, err) + var got struct { + Forwards []statusJSONForward `json:"forwards"` + ReverseForwards []statusJSONForward `json:"reverse_forwards"` + } + require.NoError(t, json.Unmarshal([]byte(out), &got), "not JSON\n%s", out) + require.Equal(t, []statusJSONForward{{HostPort: 3000, GuestPort: 3000}}, got.Forwards) + require.Equal(t, []statusJSONForward{{HostPort: 5432, GuestPort: 15432}}, got.ReverseForwards) +} + +func TestStatusShowsReverseForwards(t *testing.T) { + s, _ := setupTest(t) + require.NoError(t, s.Save(&config.Sandbox{ + Name: "rev", Provider: config.ProviderVZ, VMState: config.VMStateStopped, + ReverseForwards: []config.PortForward{{HostPort: 5432, GuestPort: 15432}}, + })) + + out, err := executeCommand("status", "rev") + require.NoError(t, err) + require.Contains(t, out, "Reverse") + require.Contains(t, out, "15432 → 5432") +} diff --git a/internal/cli/here.go b/internal/cli/here.go index b7f9847..1d7eccd 100644 --- a/internal/cli/here.go +++ b/internal/cli/here.go @@ -180,7 +180,7 @@ func loadOrCreateHereSandbox(name, cwd string) (*config.Sandbox, bool, error) { if err != nil { return nil, false, err } - var forwardSpecs []string + var forwardSpecs, reverseForwardSpecs []string var onUp, onCreate []string var requiredEnv []string var instructions []string @@ -189,6 +189,7 @@ func loadOrCreateHereSandbox(name, cwd string) (*config.Sandbox, bool, error) { var shareSources []shareSource if clawkfile != nil { forwardSpecs = append(forwardSpecs, clawkfile.Forwards...) + reverseForwardSpecs = append(reverseForwardSpecs, clawkfile.ReverseForwards...) onUp = append(onUp, clawkfile.OnUp...) onCreate = append(onCreate, clawkfile.OnCreate...) requiredEnv = append(requiredEnv, clawkfile.Env...) @@ -214,6 +215,10 @@ func loadOrCreateHereSandbox(name, cwd string) (*config.Sandbox, bool, error) { if err != nil { return nil, false, err } + reverseForwards, err := parseReverseForwardSpecs(reverseForwardSpecs, cwd) + if err != nil { + return nil, false, err + } files, err := composeFiles(fileSources) if err != nil { return nil, false, err @@ -249,24 +254,25 @@ func loadOrCreateHereSandbox(name, cwd string) (*config.Sandbox, bool, error) { memoryMiB, memoryMaxMiB = normalizeMemory(memoryMiB, memoryMaxMiB) sb := &config.Sandbox{ - Name: name, - Provider: provider, - GuestABI: sandbox.CurrentGuestABI, - Namespace: config.DefaultNamespace, // Phase 2: from clawk.mod / -n - Anchor: cwd, - VMState: config.VMStateStopped, - Network: network, - Forwards: forwards, - Files: files, - Shares: shares, - RequiredEnv: requiredEnv, - Instructions: instructions, - Memory: memory, - NestedVirt: nested, - CPU: cpu, - MemoryMiB: memoryMiB, - MemoryMaxMiB: memoryMaxMiB, - DiskMiB: diskMiB, + Name: name, + Provider: provider, + GuestABI: sandbox.CurrentGuestABI, + Namespace: config.DefaultNamespace, // Phase 2: from clawk.mod / -n + Anchor: cwd, + VMState: config.VMStateStopped, + Network: network, + Forwards: forwards, + ReverseForwards: reverseForwards, + Files: files, + Shares: shares, + RequiredEnv: requiredEnv, + Instructions: instructions, + Memory: memory, + NestedVirt: nested, + CPU: cpu, + MemoryMiB: memoryMiB, + MemoryMaxMiB: memoryMaxMiB, + DiskMiB: diskMiB, // IdleTimeoutSec rides the same snapshot-at-create rule as every // other clawk.mod value; it was the one vm(...) field this path // forgot to copy when idle-stop landed, which silently pinned every @@ -317,6 +323,23 @@ func loadHereClawkfile(cwd string) (*template.Template, []template.PolicyDef) { // config.PortForwards. context is the directory the specs came from — // used only for error messages. func parseForwardSpecs(specs []string, context string) ([]config.PortForward, error) { + return parseDirectedForwardSpecs(specs, context, "forward", + func(f config.PortForward) int { return f.HostPort }) +} + +// parseReverseForwardSpecs is parseForwardSpecs for the inbound direction. +// Duplicates are keyed on the guest port, since that's the end that binds. +func parseReverseForwardSpecs(specs []string, context string) ([]config.PortForward, error) { + return parseDirectedForwardSpecs(specs, context, "reverse forward", + func(f config.PortForward) int { return f.GuestPort }) +} + +// parseDirectedForwardSpecs parses specs, dropping later duplicates of an +// already-bound port. Single-source (one clawk.mod), so a repeat is the +// same file saying it twice — dedup silently rather than erroring the way +// the multi-source workspace path does. +func parseDirectedForwardSpecs(specs []string, context, kind string, + bound func(config.PortForward) int) ([]config.PortForward, error) { if len(specs) == 0 { return nil, nil } @@ -325,12 +348,12 @@ func parseForwardSpecs(specs []string, context string) ([]config.PortForward, er for _, s := range specs { fwd, err := parsePortSpec(s) if err != nil { - return nil, fmt.Errorf("%s forward %q: %w", context, s, err) + return nil, fmt.Errorf("%s %s %q: %w", context, kind, s, err) } - if seen[fwd.HostPort] { + if seen[bound(fwd)] { continue } - seen[fwd.HostPort] = true + seen[bound(fwd)] = true out = append(out, fwd) } return out, nil diff --git a/internal/cli/reverse_forward.go b/internal/cli/reverse_forward.go new file mode 100644 index 0000000..ac770c2 --- /dev/null +++ b/internal/cli/reverse_forward.go @@ -0,0 +1,314 @@ +package cli + +import ( + "bufio" + "context" + "errors" + "fmt" + "io" + "log" + "net" + "slices" + "strconv" + "sync" + "time" + + "github.com/clawkwork/clawk/internal/config" + "github.com/clawkwork/clawk/internal/revfwd" + "github.com/clawkwork/clawk/machine" +) + +// reverseProxy is the host end of reverse port forwarding: it publishes the +// sandbox's reverse-forward set to the in-guest agent and bridges each +// connection the guest opens through to a host loopback service. See +// internal/revfwd for the wire protocol and why gvproxy can't do this. +// +// It is built before the VM exists (the control socket needs something to +// push reloads at) and starts listening once Start is handed a machine. Set +// may be called at any point in that lifecycle; guests connected at the time +// see the new set immediately, and one that connects later reads it from the +// first snapshot. +type reverseProxy struct { + logger *log.Logger + + ctx context.Context + cancel context.CancelFunc + listener net.Listener + wg sync.WaitGroup + + mu sync.Mutex + forwards []config.PortForward + // subs are the live control connections' wakeup channels. Each is + // buffered by one and written non-blockingly, so a burst of edits + // collapses into a single resend of the (complete) current set. + subs map[chan struct{}]struct{} +} + +func newReverseProxy(logger *log.Logger) *reverseProxy { + return &reverseProxy{logger: logger, subs: make(map[chan struct{}]struct{})} +} + +// Set replaces the published forward set and wakes every connected guest. +func (p *reverseProxy) Set(fwds []config.PortForward) { + p.mu.Lock() + p.forwards = slices.Clone(fwds) + for sub := range p.subs { + select { + case sub <- struct{}{}: + default: // already pending — it will read the latest set anyway + } + } + p.mu.Unlock() +} + +// Start begins accepting guest connections on revfwd.VSockPort. It is a +// no-op (with one log line) on a backend that can't accept guest-initiated +// vsock connections, matching startSSHAgentProxy. +func (p *reverseProxy) Start(ctx context.Context, m machine.Machine) error { + listener, ok := m.(machine.VSockListener) + if !ok { + p.logger.Printf("reverse-forward: backend doesn't expose VSockListen; not starting") + return nil + } + pCtx, cancel := context.WithCancel(ctx) + l, err := listener.VSockListen(pCtx, revfwd.VSockPort) + if err != nil { + cancel() + return fmt.Errorf("vsock listen port=%d: %w", revfwd.VSockPort, err) + } + p.serve(pCtx, cancel, l) + + p.mu.Lock() + n := len(p.forwards) + p.mu.Unlock() + p.logger.Printf("reverse-forward: listening on guest vsock port %d (%d forward(s))", + revfwd.VSockPort, n) + return nil +} + +// serve takes ownership of l and starts accepting. Split out from Start so +// the protocol can be exercised over any net.Listener — the vsock listener +// only exists inside a running VM, and everything interesting about this +// type is what it does with the connections, not where they came from. +func (p *reverseProxy) serve(ctx context.Context, cancel context.CancelFunc, l net.Listener) { + p.ctx, p.cancel, p.listener = ctx, cancel, l + p.wg.Add(1) + go p.acceptLoop() +} + +// Stop closes the listener and waits for in-flight forwards. Idempotent, and +// safe on a proxy that was never started. +func (p *reverseProxy) Stop() { + if p == nil || p.listener == nil { + return + } + p.cancel() + _ = p.listener.Close() + p.wg.Wait() +} + +func (p *reverseProxy) acceptLoop() { + defer p.wg.Done() + for { + conn, err := p.listener.Accept() + if err != nil { + if errors.Is(err, net.ErrClosed) || p.ctx.Err() != nil { + return + } + p.logger.Printf("reverse-forward: accept: %v", err) + return + } + p.wg.Add(1) + go p.handle(conn) + } +} + +// handle reads the greeting and dispatches to the control stream or a +// one-shot connection bridge. +func (p *reverseProxy) handle(conn net.Conn) { + defer p.wg.Done() + defer conn.Close() + + // Shutdown has to reach connections that are simply idle — a bridged + // websocket can sit silent for hours, and Stop waits on this goroutine. + // Closing the conn from the context unblocks whatever it is parked in. + defer context.AfterFunc(p.ctx, func() { _ = conn.Close() })() + + r := revfwd.NewReader(conn) + var g revfwd.Greeting + if err := revfwd.ReadLine(r, &g); err != nil { + if !errors.Is(err, io.EOF) { + p.logger.Printf("reverse-forward: reading greeting: %v", err) + } + return + } + if g.V != revfwd.ProtoVersion { + // The guest agent is re-injected from the host's sources every cold + // boot, so this only happens across a resumed suspend state — where + // a down/up is exactly the fix. + p.logger.Printf("reverse-forward: guest speaks protocol v%d, host speaks v%d — "+ + "'clawk down && clawk up' to refresh the guest agent", g.V, revfwd.ProtoVersion) + return + } + switch g.Op { + case revfwd.OpControl: + p.serveControl(conn, r) + case revfwd.OpConnect: + p.serveConnect(conn, r, g.Port) + default: + p.logger.Printf("reverse-forward: unknown op %q", g.Op) + } +} + +// serveControl streams the forward set to one guest until it disconnects or +// the daemon shuts down. The guest treats every snapshot as the complete +// desired state, so no delta bookkeeping is needed on either side. +func (p *reverseProxy) serveControl(conn net.Conn, r *bufio.Reader) { + updates, unsubscribe := p.subscribe() + defer unsubscribe() + + // The guest never writes again on a control connection, so a read that + // returns is the guest going away. Without this the daemon would keep + // a dead subscriber until the next edit tried to write to it. + gone := make(chan struct{}) + go func() { + defer close(gone) + _, _ = io.Copy(io.Discard, r) + }() + + for { + if err := revfwd.WriteLine(conn, p.snapshot()); err != nil { + if p.ctx.Err() == nil && !errors.Is(err, net.ErrClosed) { + p.logger.Printf("reverse-forward: sending snapshot: %v", err) + } + return + } + select { + case <-updates: + case <-gone: + return + case <-p.ctx.Done(): + return + } + } +} + +// serveConnect bridges one guest-initiated connection to a host loopback +// service, after checking the requested port is actually configured. +func (p *reverseProxy) serveConnect(conn net.Conn, r *bufio.Reader, hostPort int) { + if !p.allowed(hostPort) { + // Reachable when an edit races a connection the guest already + // accepted; also the backstop if a guest ever asks for a port + // nobody configured. + p.logger.Printf("reverse-forward: refused connect to host port %d (not configured)", hostPort) + _ = revfwd.WriteLine(conn, revfwd.ConnectReply{ + OK: false, + Error: fmt.Sprintf("host port %d is not reverse-forwarded", hostPort), + }) + return + } + host, err := dialHostLoopback(hostPort) + if err != nil { + p.logger.Printf("reverse-forward: dial host port %d: %v", hostPort, err) + _ = revfwd.WriteLine(conn, revfwd.ConnectReply{ + OK: false, Error: fmt.Sprintf("dial host port %d: %v", hostPort, err), + }) + return + } + defer host.Close() + if err := revfwd.WriteLine(conn, revfwd.ConnectReply{OK: true}); err != nil { + return + } + + // Guest → host reads through r, not conn: the handshake reader may hold + // bytes the guest pipelined behind its greeting, and reading the raw + // conn would drop them. + // + // The first direction to end tears down both, rather than half-closing + // and waiting for the other. Half-close isn't available: vz hands us a + // *VirtioSocketConnection, which implements net.Conn but no CloseWrite, + // so a shutdown on this side would never reach the guest and the other + // copy would block until something timed out. Same shape as the + // ssh-agent forwarder on either end of the same transport. + done := make(chan struct{}, 2) + go func() { _, _ = io.Copy(host, r); done <- struct{}{} }() + go func() { _, _ = io.Copy(conn, host); done <- struct{}{} }() + <-done +} + +// hostLoopbackAddrs are the addresses a reverse forward's host port is +// tried on, in order. +// +// Both families, because "localhost" is not one address. On macOS it +// resolves to ::1 first, and a server told to bind the *name* usually ends +// up on the IPv6 loopback alone — `python3 -m http.server --bind localhost` +// takes the first getaddrinfo result and binds [::1] only; much Node +// tooling lands the same way. Dialling 127.0.0.1 alone makes every one of +// those servers look like nothing is listening, which surfaces in the guest +// as a connection that is accepted and immediately reset. +// +// IPv4 first because it's still where most things bind, and a refused +// connect on loopback costs microseconds. +var hostLoopbackAddrs = []string{"127.0.0.1", "::1"} + +// hostLoopbackDialTimeout bounds one attempt. Loopback either answers or +// refuses at once; the timeout only matters so a filtered address can't +// strand the connection before the other family is tried. +const hostLoopbackDialTimeout = 5 * time.Second + +// dialHostLoopback connects to port on the host's loopback, trying each +// address family. Returns the last failure when none answer — for the +// common "nothing is listening" case every attempt reports the same +// connection-refused, so the last one is as good as a joined list and +// reads better in the guest's log. +func dialHostLoopback(port int) (net.Conn, error) { + d := net.Dialer{Timeout: hostLoopbackDialTimeout} + var lastErr error + for _, host := range hostLoopbackAddrs { + conn, err := d.Dial("tcp", net.JoinHostPort(host, strconv.Itoa(port))) + if err == nil { + return conn, nil + } + lastErr = err + } + return nil, lastErr +} + +// snapshot renders the current set in wire form. +func (p *reverseProxy) snapshot() revfwd.Snapshot { + p.mu.Lock() + defer p.mu.Unlock() + snap := revfwd.Snapshot{Forwards: make([]revfwd.Forward, 0, len(p.forwards))} + for _, f := range p.forwards { + snap.Forwards = append(snap.Forwards, revfwd.Forward{ + GuestPort: f.GuestPort, HostPort: f.HostPort, + }) + } + return snap +} + +// allowed reports whether hostPort is in the current set. The guest names a +// port and never an address, and an unlisted one is refused — otherwise any +// process in the sandbox could reach every service on the Mac's loopback. +func (p *reverseProxy) allowed(hostPort int) bool { + p.mu.Lock() + defer p.mu.Unlock() + for _, f := range p.forwards { + if f.HostPort == hostPort { + return true + } + } + return false +} + +func (p *reverseProxy) subscribe() (<-chan struct{}, func()) { + ch := make(chan struct{}, 1) + p.mu.Lock() + p.subs[ch] = struct{}{} + p.mu.Unlock() + return ch, func() { + p.mu.Lock() + delete(p.subs, ch) + p.mu.Unlock() + } +} diff --git a/internal/cli/reverse_forward_test.go b/internal/cli/reverse_forward_test.go new file mode 100644 index 0000000..4cb4865 --- /dev/null +++ b/internal/cli/reverse_forward_test.go @@ -0,0 +1,215 @@ +package cli + +import ( + "bufio" + "context" + "fmt" + "io" + "log" + "net" + "testing" + "time" + + "github.com/clawkwork/clawk/internal/config" + "github.com/clawkwork/clawk/internal/revfwd" + "github.com/stretchr/testify/require" +) + +// startTestProxy runs a reverseProxy over a loopback TCP listener instead of +// vsock (which only exists inside a running VM) and returns its address. The +// protocol is transport-agnostic, so everything below exercises the real +// accept/handshake/bridge path. +func startTestProxy(t *testing.T, forwards ...config.PortForward) (*reverseProxy, string) { + t.Helper() + ln, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + + p := newReverseProxy(log.New(io.Discard, "", 0)) + p.Set(forwards) + ctx, cancel := context.WithCancel(context.Background()) + p.serve(ctx, cancel, ln) + t.Cleanup(p.Stop) + return p, ln.Addr().String() +} + +// dialProxy opens a connection and sends one greeting, returning the +// connection plus the reader that must be used for everything after it. +func dialProxy(t *testing.T, addr string, g revfwd.Greeting) (net.Conn, *bufio.Reader) { + t.Helper() + conn, err := net.Dial("tcp", addr) + require.NoError(t, err) + t.Cleanup(func() { conn.Close() }) + require.NoError(t, conn.SetDeadline(time.Now().Add(5*time.Second))) + require.NoError(t, revfwd.WriteLine(conn, g)) + return conn, revfwd.NewReader(conn) +} + +// echoServer stands in for a service bound to the host's loopback. +func echoServer(t *testing.T) int { + t.Helper() + return echoServerOn(t, "127.0.0.1:0") +} + +func echoServerOn(t *testing.T, addr string) int { + t.Helper() + ln, err := net.Listen("tcp", addr) + require.NoError(t, err) + t.Cleanup(func() { ln.Close() }) + go func() { + for { + c, err := ln.Accept() + if err != nil { + return + } + go func() { defer c.Close(); _, _ = io.Copy(c, c) }() + } + }() + return ln.Addr().(*net.TCPAddr).Port +} + +// A control connection gets the current set immediately, then a fresh full +// set on every edit — that push is what makes `forward add-reverse` apply +// without a reboot. +func TestReverseProxyControlStreamPushesUpdates(t *testing.T) { + p, addr := startTestProxy(t, config.PortForward{HostPort: 5432, GuestPort: 15432}) + _, r := dialProxy(t, addr, revfwd.Greeting{Op: revfwd.OpControl, V: revfwd.ProtoVersion}) + + var snap revfwd.Snapshot + require.NoError(t, revfwd.ReadLine(r, &snap)) + require.Equal(t, []revfwd.Forward{{GuestPort: 15432, HostPort: 5432}}, snap.Forwards) + + p.Set([]config.PortForward{ + {HostPort: 5432, GuestPort: 15432}, + {HostPort: 63342, GuestPort: 63342}, + }) + require.NoError(t, revfwd.ReadLine(r, &snap)) + require.Equal(t, []revfwd.Forward{ + {GuestPort: 15432, HostPort: 5432}, + {GuestPort: 63342, HostPort: 63342}, + }, snap.Forwards) + + // Removals travel the same way: the set is absolute, never a delta. + p.Set(nil) + require.NoError(t, revfwd.ReadLine(r, &snap)) + require.Empty(t, snap.Forwards) +} + +func TestReverseProxyBridgesConfiguredPort(t *testing.T) { + port := echoServer(t) + _, addr := startTestProxy(t, config.PortForward{HostPort: port, GuestPort: 9999}) + + conn, r := dialProxy(t, addr, + revfwd.Greeting{Op: revfwd.OpConnect, V: revfwd.ProtoVersion, Port: port}) + var reply revfwd.ConnectReply + require.NoError(t, revfwd.ReadLine(r, &reply)) + require.True(t, reply.OK, "reply: %+v", reply) + + _, err := conn.Write([]byte("ping\n")) + require.NoError(t, err) + got, err := r.ReadString('\n') + require.NoError(t, err) + require.Equal(t, "ping\n", got) +} + +// A host service on the IPv6 loopback alone must still be reachable. +// "localhost" resolves to ::1 first on macOS, so binding the name — which +// is what `python3 -m http.server --bind localhost` and much Node tooling +// do — often means [::1] and nothing on 127.0.0.1. Dialling only IPv4 made +// those look dead: the guest accepted the connection, the host dial was +// refused, and curl reported `(56) Recv failure: Connection reset by peer`. +func TestReverseProxyBridgesIPv6OnlyHostService(t *testing.T) { + port := echoServerOn(t, "[::1]:0") + // Precondition for the regression: the service really is invisible over + // IPv4, so this test fails if the fix is reverted. + c, err := net.DialTimeout("tcp", fmt.Sprintf("127.0.0.1:%d", port), 2*time.Second) + if err == nil { + c.Close() + t.Skip("something else holds the IPv4 loopback on this port; can't isolate the case") + } + + _, addr := startTestProxy(t, config.PortForward{HostPort: port, GuestPort: 9999}) + conn, r := dialProxy(t, addr, + revfwd.Greeting{Op: revfwd.OpConnect, V: revfwd.ProtoVersion, Port: port}) + var reply revfwd.ConnectReply + require.NoError(t, revfwd.ReadLine(r, &reply)) + require.True(t, reply.OK, "IPv6-only host service unreachable: %+v", reply) + + _, err = conn.Write([]byte("ping\n")) + require.NoError(t, err) + got, err := r.ReadString('\n') + require.NoError(t, err) + require.Equal(t, "ping\n", got) +} + +// The guest names a port and nothing else, and one that isn't configured is +// refused — otherwise anything in the sandbox could reach every service on +// the host's loopback. +func TestReverseProxyRefusesUnconfiguredPort(t *testing.T) { + port := echoServer(t) + _, addr := startTestProxy(t) // nothing configured + + _, r := dialProxy(t, addr, + revfwd.Greeting{Op: revfwd.OpConnect, V: revfwd.ProtoVersion, Port: port}) + var reply revfwd.ConnectReply + require.NoError(t, revfwd.ReadLine(r, &reply)) + require.False(t, reply.OK) + require.Contains(t, reply.Error, "not reverse-forwarded") +} + +// Only the HOST port authorises a connection; a guest port that happens to +// match some other mapping's host port must not open a hole. +func TestReverseProxyRefusesGuestPortAsHostPort(t *testing.T) { + port := echoServer(t) + _, addr := startTestProxy(t, config.PortForward{HostPort: 1, GuestPort: port}) + + _, r := dialProxy(t, addr, + revfwd.Greeting{Op: revfwd.OpConnect, V: revfwd.ProtoVersion, Port: port}) + var reply revfwd.ConnectReply + require.NoError(t, revfwd.ReadLine(r, &reply)) + require.False(t, reply.OK) +} + +// A configured port whose service isn't up must fail the connection rather +// than hang the caller. +func TestReverseProxyReportsDialFailure(t *testing.T) { + ln, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + dead := ln.Addr().(*net.TCPAddr).Port + require.NoError(t, ln.Close()) + + _, addr := startTestProxy(t, config.PortForward{HostPort: dead, GuestPort: dead}) + _, r := dialProxy(t, addr, + revfwd.Greeting{Op: revfwd.OpConnect, V: revfwd.ProtoVersion, Port: dead}) + var reply revfwd.ConnectReply + require.NoError(t, revfwd.ReadLine(r, &reply)) + require.False(t, reply.OK) + require.Contains(t, reply.Error, fmt.Sprintf("dial host port %d", dead)) + require.Contains(t, reply.Error, "connection refused") +} + +// A guest agent from a different protocol generation is hung up on, not +// guessed at. +func TestReverseProxyRejectsVersionMismatch(t *testing.T) { + _, addr := startTestProxy(t, config.PortForward{HostPort: 5432, GuestPort: 5432}) + _, r := dialProxy(t, addr, + revfwd.Greeting{Op: revfwd.OpControl, V: revfwd.ProtoVersion + 1}) + var snap revfwd.Snapshot + require.ErrorIs(t, revfwd.ReadLine(r, &snap), io.EOF) +} + +// Stop must return even with a control connection parked on an idle stream — +// the daemon's shutdown waits on it. +func TestReverseProxyStopClosesIdleConnections(t *testing.T) { + p, addr := startTestProxy(t, config.PortForward{HostPort: 5432, GuestPort: 5432}) + _, r := dialProxy(t, addr, revfwd.Greeting{Op: revfwd.OpControl, V: revfwd.ProtoVersion}) + var snap revfwd.Snapshot + require.NoError(t, revfwd.ReadLine(r, &snap)) + + done := make(chan struct{}) + go func() { defer close(done); p.Stop() }() + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("Stop blocked on an idle control connection") + } +} diff --git a/internal/cli/run.go b/internal/cli/run.go index 7542792..f7be58d 100644 --- a/internal/cli/run.go +++ b/internal/cli/run.go @@ -341,6 +341,10 @@ func loadOrCreateSandboxFromWorkspace(name string, ws *template.Workspace) (*con if err != nil { return nil, err } + reverseForwards, err := mergeReverseForwards(ws) + if err != nil { + return nil, err + } files, err := mergeFiles(ws) if err != nil { @@ -383,24 +387,25 @@ func loadOrCreateSandboxFromWorkspace(name string, ws *template.Workspace) (*con } sb := &config.Sandbox{ - Name: name, - Provider: provider, - GuestABI: sandbox.CurrentGuestABI, - Profile: runProfile, - Namespace: createNamespace(), - VMState: config.VMStateStopped, - Network: network, - Forwards: forwards, - Files: files, - Shares: shares, - NestedVirt: nested, - CPU: cpu, - MemoryMiB: memoryMiB, - MemoryMaxMiB: memoryMaxMiB, - DiskMiB: disk, - IdleTimeoutSec: resolveIdleTimeout(ws), - Image: image, - Kernel: kernel, + Name: name, + Provider: provider, + GuestABI: sandbox.CurrentGuestABI, + Profile: runProfile, + Namespace: createNamespace(), + VMState: config.VMStateStopped, + Network: network, + Forwards: forwards, + ReverseForwards: reverseForwards, + Files: files, + Shares: shares, + NestedVirt: nested, + CPU: cpu, + MemoryMiB: memoryMiB, + MemoryMaxMiB: memoryMaxMiB, + DiskMiB: disk, + IdleTimeoutSec: resolveIdleTimeout(ws), + Image: image, + Kernel: kernel, // Workspace-level `on up` / `on create` — run at the workspace root, // distinct from the per-repo hooks addPhases folds into each Phase. // Only a true workspace root (a block with includes) carries a @@ -566,45 +571,75 @@ func resolveKernel(ws *template.Workspace) (string, error) { return finalizeKernelRef(picked), nil } -// mergeForwards gathers forward specs from the workspace and each repo's -// Clawkfile, parses them, and errors on duplicate host ports with a message -// naming both sources. +// forwardSource is one port-forward spec plus where it came from, so a +// conflict can name both contributors. +type forwardSource struct { + Origin string // human-readable: "workspace" or a repo name + Spec string +} + +// mergeForwards gathers outbound forward specs from the workspace and each +// repo's Clawkfile, parses them, and errors on duplicate host ports with a +// message naming both sources. func mergeForwards(ws *template.Workspace) ([]config.PortForward, error) { - type src struct { - Origin string // human-readable: "workspace" or a repo name - Spec string - } - var sources []src - for _, s := range ws.File.Forwards { - sources = append(sources, src{Origin: "workspace", Spec: s}) + sources := collectForwardSpecs(ws, func(t *template.Template) []string { return t.Forwards }) + // The host port is the one that gets bound, so it's the one that can + // only be claimed once. + return mergeForwardSources(sources, "forward", "host port", + func(f config.PortForward) int { return f.HostPort }) +} + +// mergeReverseForwards is mergeForwards for the inbound direction. The +// uniqueness check moves to the guest port: that's the end doing the +// binding, so two specs claiming it conflict even when their host ports +// differ. +func mergeReverseForwards(ws *template.Workspace) ([]config.PortForward, error) { + sources := collectForwardSpecs(ws, func(t *template.Template) []string { return t.ReverseForwards }) + return mergeForwardSources(sources, "reverse forward", "guest port", + func(f config.PortForward) int { return f.GuestPort }) +} + +// collectForwardSpecs pulls one flavour of forward spec off the workspace +// file and every repo Clawkfile, tagged with its origin. +func collectForwardSpecs(ws *template.Workspace, pick func(*template.Template) []string) []forwardSource { + var sources []forwardSource + for _, s := range pick(ws.File) { + sources = append(sources, forwardSource{Origin: "workspace", Spec: s}) } for _, r := range ws.Repos { if r.Clawkfile == nil { continue } - for _, s := range r.Clawkfile.Forwards { - sources = append(sources, src{Origin: r.Name, Spec: s}) + for _, s := range pick(r.Clawkfile) { + sources = append(sources, forwardSource{Origin: r.Name, Spec: s}) } } + return sources +} - byHostPort := make(map[int]src) +// mergeForwardSources parses specs and rejects two sources claiming the +// same bound port. kind and portName appear in the error; bound picks the +// port that can only be claimed once for this direction. +func mergeForwardSources(sources []forwardSource, kind, portName string, + bound func(config.PortForward) int) ([]config.PortForward, error) { + byPort := make(map[int]forwardSource) var out []config.PortForward for _, s := range sources { fwd, err := parsePortSpec(s.Spec) if err != nil { - return nil, fmt.Errorf("%s forward %q: %w", s.Origin, s.Spec, err) + return nil, fmt.Errorf("%s %s %q: %w", s.Origin, kind, s.Spec, err) } - if prev, dup := byHostPort[fwd.HostPort]; dup { + if prev, dup := byPort[bound(fwd)]; dup { // Allow identical specs from multiple sources (harmless); reject - // only genuine conflicts where the guest port differs. + // only genuine conflicts where the other port differs. if fwd == mustParseSpec(prev.Spec) { continue } return nil, fmt.Errorf( - "host port %d declared by both %s (%s) and %s (%s) — change one", - fwd.HostPort, prev.Origin, prev.Spec, s.Origin, s.Spec) + "%s %d declared by both %s (%s) and %s (%s) — change one", + portName, bound(fwd), prev.Origin, prev.Spec, s.Origin, s.Spec) } - byHostPort[fwd.HostPort] = s + byPort[bound(fwd)] = s out = append(out, fwd) } return out, nil diff --git a/internal/cli/run_test.go b/internal/cli/run_test.go index 83ad66d..d10e49f 100644 --- a/internal/cli/run_test.go +++ b/internal/cli/run_test.go @@ -351,6 +351,56 @@ func TestRunPortConflict(t *testing.T) { require.True(t, strings.Contains(err.Error(), "host port 3000")) } +// TestRunReverseForwards: `reverse` entries from the workspace and a repo +// union onto the record's ReverseForwards, and never into Forwards. +func TestRunReverseForwards(t *testing.T) { + _, _ = setupTest(t) + + dir := t.TempDir() + repo := filepath.Join(dir, "app") + require.NoError(t, os.MkdirAll(repo, 0o755)) + gitInit(t, repo) + require.NoError(t, os.WriteFile(filepath.Join(repo, "clawk.mod"), + []byte("sandbox (\n forwards (\n 3000\n reverse 5432:15432\n )\n)\n"), 0o644)) + require.NoError(t, os.WriteFile(filepath.Join(dir, "clawk.mod"), + []byte("sandbox (\n includes (\n ./app\n )\n forwards (\n reverse 63342\n )\n)\n"), 0o644)) + + _, err := executeCommand("work", filepath.Join(dir, "clawk.mod"), "REV-1", "--bare") + require.NoError(t, err) + + sb, err := store.Load("REV-1") + require.NoError(t, err) + require.Equal(t, []config.PortForward{{HostPort: 3000, GuestPort: 3000}}, sb.Forwards) + require.ElementsMatch(t, []config.PortForward{ + {HostPort: 63342, GuestPort: 63342}, + {HostPort: 5432, GuestPort: 15432}, + }, sb.ReverseForwards) +} + +// A reverse forward binds inside the guest, so the conflict that matters +// is two sources claiming the same GUEST port — even with different host +// ports, which would pass the outbound direction's check. +func TestRunReverseForwardGuestPortConflict(t *testing.T) { + _, _ = setupTest(t) + + dir := t.TempDir() + for _, sub := range []string{"a", "b"} { + r := filepath.Join(dir, sub) + require.NoError(t, os.MkdirAll(r, 0o755)) + gitInit(t, r) + } + require.NoError(t, os.WriteFile(filepath.Join(dir, "a", "clawk.mod"), + []byte("sandbox (\n forwards (\n reverse 5432:15432\n )\n)\n"), 0o644)) + require.NoError(t, os.WriteFile(filepath.Join(dir, "b", "clawk.mod"), + []byte("sandbox (\n forwards (\n reverse 6432:15432\n )\n)\n"), 0o644)) + require.NoError(t, os.WriteFile(filepath.Join(dir, "clawk.mod"), + []byte("sandbox (\n includes (\n ./a\n ./b\n )\n)\n"), 0o644)) + + _, err := executeCommand("work", filepath.Join(dir, "clawk.mod"), "REV-2", "--bare") + require.Error(t, err) + require.Contains(t, err.Error(), "guest port 15432") +} + // TestRunProviderConflict: two repos disagree on provider and the workspace // doesn't pick one. func TestRunProviderConflict(t *testing.T) { diff --git a/internal/cli/status.go b/internal/cli/status.go index 78d0558..0f2867f 100644 --- a/internal/cli/status.go +++ b/internal/cli/status.go @@ -65,9 +65,10 @@ type statusJSONOutput struct { Branches []statusJSONBranch `json:"branches"` // v2 additive blocks. - Forwards []statusJSONForward `json:"forwards,omitempty"` - Network *statusJSONNetwork `json:"network,omitempty"` - Setup []statusJSONSetup `json:"setup,omitempty"` + Forwards []statusJSONForward `json:"forwards,omitempty"` + ReverseForwards []statusJSONForward `json:"reverse_forwards,omitempty"` + Network *statusJSONNetwork `json:"network,omitempty"` + Setup []statusJSONSetup `json:"setup,omitempty"` } type statusJSONBranch struct { @@ -185,6 +186,11 @@ func renderStatusJSON(w io.Writer, sb *config.Sandbox, liveStatus string) error HostPort: f.HostPort, GuestPort: f.GuestPort, }) } + for _, f := range sb.ReverseForwards { + out.ReverseForwards = append(out.ReverseForwards, statusJSONForward{ + HostPort: f.HostPort, GuestPort: f.GuestPort, + }) + } out.Network = &statusJSONNetwork{ Use: effectiveUseForLog(sb), Blocks: sb.Network.Blocks, @@ -276,6 +282,15 @@ func renderStatusDashboard(w io.Writer, provider sandbox.Provider, sb *config.Sa } fmt.Fprintf(w, " Forwards %s\n", strings.Join(parts, ", ")) } + // Arrow drawn guest→host so the two rows read in the direction traffic + // flows; without that a reader has no way to tell them apart. + if len(sb.ReverseForwards) > 0 { + parts := make([]string, 0, len(sb.ReverseForwards)) + for _, f := range sb.ReverseForwards { + parts = append(parts, fmt.Sprintf("%d → %d", f.GuestPort, f.HostPort)) + } + fmt.Fprintf(w, " Reverse %s (guest → host loopback)\n", strings.Join(parts, ", ")) + } // One line per policy layer, lowest precedence first: the use chain, // then the sandbox's own blocks with entry counts. Full contents live diff --git a/internal/cli/vzd.go b/internal/cli/vzd.go index e789abf..f1ddc53 100644 --- a/internal/cli/vzd.go +++ b/internal/cli/vzd.go @@ -97,12 +97,20 @@ func runVzd(_ *cobra.Command, args []string) (retErr error) { // socket so the endpoints answer "booting" instead of racing. lc := newVMLifecycle(sb.Name, vmDir, logger) + // Reverse-forward proxy: publishes the sandbox's host-loopback exposures + // to the in-guest agent and bridges the connections it opens. Built here, + // before the control socket, so `clawk forward add-reverse` has somewhere + // to push while the VM is still booting; it starts listening once the + // machine exists (below). + rev := newReverseProxy(logger) + rev.Set(sb.ReverseForwards) + // Control socket: lets the CLI push network-policy edits into the live // allow list (`clawk network allow` without a down/up cycle), read - // the denial ledger (`clawk network denials`), and drive the VM - // lifecycle (`clawk pause/resume/snapshot`). Best-effort — without - // it, policy edits apply on the next up, as before. - ctl, err := vzdctl.Start(vzdctl.SocketPath(vmDir), controlHandlers(sb, allow, lc, logger)) + // the denial ledger (`clawk network denials`), push reverse-forward + // edits, and drive the VM lifecycle (`clawk pause/resume/snapshot`). + // Best-effort — without it, policy edits apply on the next up, as before. + ctl, err := vzdctl.Start(vzdctl.SocketPath(vmDir), controlHandlers(sb, allow, lc, rev, logger)) if err != nil { logger.Printf("control socket: disabled (%v) — network edits apply on next up", err) } else { @@ -197,6 +205,16 @@ func runVzd(_ *cobra.Command, args []string) (retErr error) { defer sshAgent.Stop() } + // Reverse forwards: the guest agent dials this listener to learn which + // host loopback ports to bind on its own 127.0.0.1, and again for every + // connection to one. Best-effort — a failure here means those ports + // simply don't appear inside the guest. + if err := rev.Start(ctx, m); err != nil { + logger.Printf("reverse-forward: disabled (%v)", err) + } else { + defer rev.Stop() + } + // 9p cache servers: one per toolchain cache, each serving its host cache // dir over 9p on the guest vsock port from ToolchainCacheShares. A // 9p-capable clawk-init mounts these instead of the caches' virtio-fs diff --git a/internal/config/types.go b/internal/config/types.go index 166d1f9..7c3cfdd 100644 --- a/internal/config/types.go +++ b/internal/config/types.go @@ -361,9 +361,19 @@ var DefaultAllowedDomains = []string{ "*.snapcraftcontent.com", // the CDN snap redirects to for actual .snap downloads } -// PortForward maps a host port to a guest port so services running in the VM -// are reachable from the host (e.g., dev servers). Applied at VM start time; -// changes require an `up` cycle to take effect. +// PortForward maps a host port to a guest port. It expresses both forwarding +// directions; which one applies depends on the field it is stored in. +// +// - Sandbox.Forwards (outbound): host 127.0.0.1:HostPort → guest GuestPort, +// served by gvproxy. Applied at VM start time; changes require an `up` +// cycle to take effect. +// - Sandbox.ReverseForwards (inbound): guest 127.0.0.1:GuestPort → host +// 127.0.0.1:HostPort, tunnelled over vsock (see internal/revfwd). +// Applied live to a running sandbox. +// +// The HostPort:GuestPort spelling is deliberately the same in both — a spec +// always reads "this host port, that guest port", so `3000:80` maps the same +// pair of ports whichever direction is being configured. type PortForward struct { HostPort int `json:"host_port"` GuestPort int `json:"guest_port"` @@ -435,6 +445,12 @@ type Sandbox struct { Phases []Phase `json:"phases"` Network NetworkPolicy `json:"network"` Forwards []PortForward `json:"forwards,omitempty"` + // ReverseForwards are host loopback services exposed on the guest's + // own loopback — the inbound counterpart of Forwards. Unlike Forwards + // (a gvproxy binding fixed at VM start) these are tunnelled over vsock + // by the daemon, so edits apply to a running sandbox. See + // internal/revfwd. + ReverseForwards []PortForward `json:"reverse_forwards,omitempty"` // Files is the list of host->guest file copies refreshed on every // `clawk up`. See HostFile. Empty = no snapshots. Files []HostFile `json:"files,omitempty"` diff --git a/internal/revfwd/revfwd.go b/internal/revfwd/revfwd.go new file mode 100644 index 0000000..26bc1e6 --- /dev/null +++ b/internal/revfwd/revfwd.go @@ -0,0 +1,143 @@ +// Package revfwd is the wire protocol for reverse port forwarding: host +// loopback services made reachable on the guest's own loopback. +// +// `clawk forward add` is the outbound half — a guest port bound on the +// host's 127.0.0.1 by gvproxy. This is the inbound half, and it can't ride +// gvproxy: a guest process that dials 127.0.0.1 reaches the guest's own +// loopback, and no host route exists for that. So the guest binds the port +// itself and tunnels each connection to the host over AF_VSOCK. +// +// Shape, per connection (the guest always dials, the host always listens +// on VSockPort): +// +// guest → host one JSON Greeting, newline-terminated +// op=control host replies with a Snapshot line now and another on +// every change to the forward set, until the connection +// is closed. This is how the guest learns which ports to +// bind, so `clawk forward add-reverse` applies live. +// op=connect host validates Port against the current set, replies +// with one ConnectReply line, and on ok=true pipes raw +// bytes to 127.0.0.1:Port for the rest of the connection. +// +// Validation is host-side on purpose: the guest names a port, never an +// address, and a port the user didn't configure is refused. Otherwise any +// process in the sandbox could reach every service on the Mac's loopback. +// +// JSON lines rather than the binary framing of internal/vsockproto: this +// carries a handful of control messages per connection, not a byte stream +// that needs chunking, and being greppable in a log is worth more here +// than the framing overhead saved. +// +// The guest half is inlined in internal/agentembed/main.go.in (the agent +// builds standalone inside the guest and can't import this package). Any +// change here must be mirrored there — same rule as internal/vsockproto. +package revfwd + +import ( + "bufio" + "encoding/json" + "errors" + "fmt" + "io" +) + +// VSockPort is the host-side AF_VSOCK port the guest dials for both +// connection kinds. Disjoint from the other fixed ports: 1024 pty-agent, +// 1025 time-sync, 1026 ssh-agent, 1027 mem-report, 1100+ 9p caches. +const VSockPort uint32 = 1028 + +// ProtoVersion is the current wire version. Bumped only on a breaking +// change; the guest binary is rebuilt from these sources and re-injected +// on every `clawk up`, so host and guest can't drift within a release. +const ProtoVersion = 1 + +// Greeting is the first line of every connection. +type Greeting struct { + // Op is OpControl or OpConnect. + Op string `json:"op"` + + // V is the sender's ProtoVersion. + V int `json:"v"` + + // Port is the HOST port to dial. Set for OpConnect only. + Port int `json:"port,omitempty"` +} + +// Greeting Op values. +const ( + OpControl = "control" + OpConnect = "connect" +) + +// Snapshot is the host's reply on an OpControl connection: the complete +// set of reverse forwards as of now. Every update resends the full set — +// the guest reconciles against it rather than applying deltas, so a +// dropped-and-redialed control connection converges the same way. +type Snapshot struct { + Forwards []Forward `json:"forwards"` +} + +// Forward is one host-loopback port exposed on the guest's loopback. +type Forward struct { + // GuestPort is bound on 127.0.0.1 inside the guest. + GuestPort int `json:"guest"` + + // HostPort is dialed on 127.0.0.1 on the host. + HostPort int `json:"host"` +} + +// ConnectReply is the host's verdict on an OpConnect greeting. Bytes flow +// only after ok=true; on ok=false the host closes the connection. +type ConnectReply struct { + OK bool `json:"ok"` + // Error is a short human-readable reason when OK is false. The guest + // logs it — it's the only place a misconfigured port surfaces. + Error string `json:"error,omitempty"` +} + +// MaxLineBytes caps one control line. Snapshots are a few dozen bytes per +// forward; the cap exists so a peer can't make the reader allocate without +// bound. +const MaxLineBytes = 64 * 1024 + +// ErrLineTooLong reports a control line past MaxLineBytes. The connection +// must be closed — the reader is out of sync with the framing. +var ErrLineTooLong = errors.New("revfwd: control line exceeds MaxLineBytes") + +// WriteLine JSON-encodes v and writes it as one newline-terminated line. +func WriteLine(w io.Writer, v any) error { + b, err := json.Marshal(v) + if err != nil { + return fmt.Errorf("revfwd: encoding %T: %w", v, err) + } + if len(b)+1 > MaxLineBytes { + return ErrLineTooLong + } + _, err = w.Write(append(b, '\n')) + return err +} + +// ReadLine reads one newline-terminated JSON line into v. +// +// It takes a *bufio.Reader rather than an io.Reader because the same +// connection carries raw proxied bytes right after the handshake: the +// caller must keep reading through this reader, or bytes buffered past the +// newline are lost. +func ReadLine(r *bufio.Reader, v any) error { + line, err := r.ReadSlice('\n') + if errors.Is(err, bufio.ErrBufferFull) { + return ErrLineTooLong + } + if err != nil { + return err + } + if err := json.Unmarshal(line, v); err != nil { + return fmt.Errorf("revfwd: decoding %T: %w", v, err) + } + return nil +} + +// NewReader wraps c with a reader sized for MaxLineBytes, so ReadLine's +// buffer-full case really does mean "line too long" rather than "buffer +// too small". +func NewReader(r io.Reader) *bufio.Reader { return bufio.NewReaderSize(r, MaxLineBytes) } diff --git a/internal/revfwd/revfwd_test.go b/internal/revfwd/revfwd_test.go new file mode 100644 index 0000000..e0d2b4c --- /dev/null +++ b/internal/revfwd/revfwd_test.go @@ -0,0 +1,72 @@ +package revfwd + +import ( + "bytes" + "errors" + "io" + "strings" + "testing" + + "github.com/stretchr/testify/require" +) + +func TestGreetingRoundTrip(t *testing.T) { + var buf bytes.Buffer + require.NoError(t, WriteLine(&buf, Greeting{Op: OpConnect, V: ProtoVersion, Port: 63342})) + + var got Greeting + require.NoError(t, ReadLine(NewReader(&buf), &got)) + require.Equal(t, Greeting{Op: OpConnect, V: ProtoVersion, Port: 63342}, got) +} + +func TestSnapshotRoundTrip(t *testing.T) { + var buf bytes.Buffer + want := Snapshot{Forwards: []Forward{ + {GuestPort: 63342, HostPort: 63342}, + {GuestPort: 15432, HostPort: 5432}, + }} + require.NoError(t, WriteLine(&buf, want)) + + var got Snapshot + require.NoError(t, ReadLine(NewReader(&buf), &got)) + require.Equal(t, want, got) +} + +// The handshake and the proxied payload share one connection, so the +// reader that read the greeting must be the one that keeps reading — a +// peer that pipelines its first bytes behind the newline would otherwise +// lose them to the discarded buffer. +func TestReadLineLeavesTrailingBytesReadable(t *testing.T) { + var buf bytes.Buffer + require.NoError(t, WriteLine(&buf, ConnectReply{OK: true})) + buf.WriteString("GET / HTTP/1.1\r\n") + + r := NewReader(&buf) + var reply ConnectReply + require.NoError(t, ReadLine(r, &reply)) + require.True(t, reply.OK) + + rest, err := io.ReadAll(r) + require.NoError(t, err) + require.Equal(t, "GET / HTTP/1.1\r\n", string(rest)) +} + +// A peer that never sends a newline must not make the reader buffer +// without bound. +func TestReadLineRejectsOverlongLine(t *testing.T) { + r := NewReader(strings.NewReader(strings.Repeat("x", MaxLineBytes+10))) + var g Greeting + require.ErrorIs(t, ReadLine(r, &g), ErrLineTooLong) +} + +func TestReadLineReportsEOF(t *testing.T) { + var g Greeting + require.ErrorIs(t, ReadLine(NewReader(strings.NewReader("")), &g), io.EOF) +} + +func TestReadLineRejectsGarbage(t *testing.T) { + var g Greeting + err := ReadLine(NewReader(strings.NewReader("not json\n")), &g) + require.Error(t, err) + require.False(t, errors.Is(err, io.EOF)) +} diff --git a/internal/sandbox/shares.go b/internal/sandbox/shares.go index 2a02766..3129b37 100644 --- a/internal/sandbox/shares.go +++ b/internal/sandbox/shares.go @@ -241,7 +241,8 @@ func SeedClaudeMemory(stateRoot, seed string) error { // NinepBasePort is the first guest vsock port used by the host 9p cache // servers; each ToolchainCacheShares entry gets NinepBasePort+index. Chosen // clear of the fixed control ports (1024 pty-agent, 1025 time-sync, 1026 -// ssh-agent) with headroom so adding a cache never collides. +// ssh-agent, 1027 mem-report, 1028 reverse-forward) with headroom so adding +// a cache never collides. const NinepBasePort uint32 = 1100 // ToolchainCachesEnabled gates ToolchainCacheShares. It is false: every diff --git a/internal/template/parse.go b/internal/template/parse.go index fdc44b9..4e2b83c 100644 --- a/internal/template/parse.go +++ b/internal/template/parse.go @@ -98,7 +98,11 @@ type Template struct { // block. DenySources []string Forwards []string // port forward specs (PORT or HOST:GUEST) - Env []string // env entries to export in the VM (canonical envspec form; see parseEnvBlock) + // ReverseForwards are the `reverse ` entries of a forwards + // block: host loopback ports exposed on the guest's loopback. Same + // HOST:GUEST spelling as Forwards, opposite direction of travel. + ReverseForwards []string + Env []string // env entries to export in the VM (canonical envspec form; see parseEnvBlock) // Lifecycle hooks. Each is a list of shell commands run inside the // VM at the named moment. @@ -211,6 +215,7 @@ func (t *Template) Merge(over *Template) { t.DenyIPs = append(t.DenyIPs, over.DenyIPs...) t.Use = append(t.Use, over.Use...) t.Forwards = append(t.Forwards, over.Forwards...) + t.ReverseForwards = append(t.ReverseForwards, over.ReverseForwards...) t.Env = append(t.Env, over.Env...) t.OnCreate = append(t.OnCreate, over.OnCreate...) t.OnUp = append(t.OnUp, over.OnUp...) @@ -299,7 +304,7 @@ func (p *parser) parseTemplateDirective(tmpl *Template, t Token) error { case "on": return p.parseOnDirective(tmpl) case "forwards": - return p.parseIdentBlock(&tmpl.Forwards, "forwards") + return p.parseForwardsBlock(tmpl) case "files": return p.parseFilesBlock(tmpl) case "shares": @@ -397,6 +402,78 @@ func (p *parser) parseValidatedIdentBlock(dst *[]string, directive string, valid } } +// parseForwardsBlock parses a `forwards ( … )` block (or the inline +// `forwards ENTRY` form). An entry is a port spec — `PORT` or +// `HOST:GUEST` — optionally prefixed with `reverse`: +// +// forwards ( +// 3000 # guest 3000 reachable at localhost:3000 on the host +// 8080:80 # guest 80 reachable at localhost:8080 on the host +// reverse 63342 # the host's localhost:63342 reachable inside the guest +// ) +// +// `reverse` is a modifier rather than its own top-level directive because +// the two lists are the same thing pointed in opposite directions — +// keeping them in one block is what makes the asymmetry visible. +// +// Specs are validated downstream (internal/cli parses them into +// config.PortForward), same as before: this only sorts entries into the +// two lists. +func (p *parser) parseForwardsBlock(tmpl *Template) error { + p.advance() // consume "forwards" + t := p.peek() + + // Inline form: `forwards ENTRY` / `forwards reverse ENTRY` + if t.Kind == TokIdent { + if err := p.parseForwardEntry(tmpl); err != nil { + return err + } + return p.expectNewlineOrEOF() + } + + if t.Kind != TokLParen { + return p.errorAt(t, "expected '(' or identifier after %q, got %s", "forwards", t) + } + p.advance() + for { + p.skipNewlines() + if t := p.peek(); t.Kind == TokRParen { + p.advance() + return p.expectNewlineOrEOF() + } else if t.Kind != TokIdent { + return p.errorAt(t, "expected entry or ')' in %q, got %s", "forwards", t) + } + if err := p.parseForwardEntry(tmpl); err != nil { + return err + } + } +} + +// parseForwardEntry consumes one forwards entry, with the current token +// already known to be a TokIdent. +func (p *parser) parseForwardEntry(tmpl *Template) error { + t := p.peek() + if t.Val != "reverse" { + if isKeyword(t.Val) { + return p.errorAt(t, "unexpected keyword %q in %q", t.Val, "forwards") + } + tmpl.Forwards = append(tmpl.Forwards, t.Val) + p.advance() + return nil + } + p.advance() // consume "reverse" + // The spec must follow on the same line; a newline token here means + // the user wrote a bare `reverse`, which we reject rather than + // silently swallowing the next entry. + spec := p.peek() + if spec.Kind != TokIdent || isKeyword(spec.Val) { + return p.errorAt(spec, "expected a port spec after 'reverse', got %s", spec) + } + tmpl.ReverseForwards = append(tmpl.ReverseForwards, spec.Val) + p.advance() + return nil +} + // parseEnvBlock parses an `env ( … )` block (or the inline `env ENTRY` // form). Each entry is either a bare name (passthrough of the same host // variable) or `NAME = VALUE`, where VALUE is a ${…} host reference — with diff --git a/internal/template/parse_test.go b/internal/template/parse_test.go index b847ec2..25ff50a 100644 --- a/internal/template/parse_test.go +++ b/internal/template/parse_test.go @@ -117,6 +117,36 @@ func TestParseForwards(t *testing.T) { } } +// TestParseReverseForwards: `reverse` inside a forwards block sorts the +// entry into the inbound list, leaving plain entries where they were. +func TestParseReverseForwards(t *testing.T) { + src := `forwards ( + 3000 + reverse 63342 + reverse 5432:15432 + 8080:80 +) +` + tmpl, err := parseBody(src) + require.NoError(t, err) + require.Equal(t, []string{"3000", "8080:80"}, tmpl.Forwards) + require.Equal(t, []string{"63342", "5432:15432"}, tmpl.ReverseForwards) +} + +func TestParseReverseForwardInline(t *testing.T) { + tmpl, err := parseBody("forwards reverse 63342\n") + require.NoError(t, err) + require.Empty(t, tmpl.Forwards) + require.Equal(t, []string{"63342"}, tmpl.ReverseForwards) +} + +// A bare `reverse` must not silently swallow the next entry. +func TestParseReverseForwardNeedsSpec(t *testing.T) { + _, err := parseBody("forwards (\n reverse\n 3000\n)\n") + require.Error(t, err) + require.Contains(t, err.Error(), "expected a port spec after 'reverse'") +} + // TestParseInlineForms checks the go.mod-style syntax where single-entry // directives can be written without parentheses — `forwards 3000` is // equivalent to `forwards ( 3000 )`. Multi-entry directives still need the diff --git a/internal/vzdctl/vzdctl.go b/internal/vzdctl/vzdctl.go index 01ed155..1df7868 100644 --- a/internal/vzdctl/vzdctl.go +++ b/internal/vzdctl/vzdctl.go @@ -38,6 +38,13 @@ type Handlers struct { // applies it to the live allow list. Reload func() error + // ReloadForwards, if non-nil, re-reads the sandbox's reverse port + // forwards from the store and pushes them to the in-guest agent. Nil + // on backends with no vsock listener (firecracker), where the endpoint + // reports 404 and the client maps it to + // ErrReverseForwardsUnsupported. + ReloadForwards func() error + // Gate, if non-nil, powers the interactive allow/deny endpoints // (/v1/events, /v1/decide, /v1/pending). When nil those endpoints // report 404 and the daemon serves only the denial ledger + reload. @@ -127,6 +134,21 @@ func Start(path string, h Handlers) (*Server, error) { } w.WriteHeader(http.StatusNoContent) }) + mux.HandleFunc("POST /v1/reload-forwards", func(w http.ResponseWriter, _ *http.Request) { + // Its own endpoint rather than a second job for /v1/reload: a + // daemon that predates reverse forwarding answers /v1/reload + // happily, and the CLI would report a live apply that never + // happened. A 404 here is the honest answer. + if h.ReloadForwards == nil { + http.Error(w, "reverse forwarding not supported", http.StatusNotFound) + return + } + if err := h.ReloadForwards(); err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + w.WriteHeader(http.StatusNoContent) + }) mux.HandleFunc("GET /v1/events", func(w http.ResponseWriter, r *http.Request) { serveEvents(w, r, h.Gate) }) @@ -382,6 +404,12 @@ var ErrNotRunning = errors.New("control socket not available (sandbox not runnin // errors.Is and suggest a sandbox restart. var ErrLifecycleUnsupported = errors.New("daemon does not support lifecycle control (restart the sandbox to upgrade its daemon)") +// ErrReverseForwardsUnsupported reports that the daemon answered but has no +// reverse-forward endpoint — either it predates the feature or its backend +// has no host-side vsock listener (firecracker). Callers check with +// errors.Is. +var ErrReverseForwardsUnsupported = errors.New("daemon does not support reverse port forwarding") + // Lifecycle fetches the VM's live lifecycle snapshot. func (c *Client) Lifecycle(ctx context.Context) (LifecycleState, error) { req, err := http.NewRequestWithContext(ctx, http.MethodGet, "http://vzd/v1/lifecycle", nil) @@ -478,6 +506,28 @@ func (c *Client) Reload(ctx context.Context) error { return nil } +// ReloadForwards asks the daemon to re-read the sandbox's reverse port +// forwards from the store and push them to the in-guest agent. +func (c *Client) ReloadForwards(ctx context.Context) error { + req, err := http.NewRequestWithContext(ctx, http.MethodPost, "http://vzd/v1/reload-forwards", nil) + if err != nil { + return fmt.Errorf("building reload-forwards request: %w", err) + } + resp, err := c.do(req) + if err != nil { + return err + } + defer resp.Body.Close() + switch resp.StatusCode { + case http.StatusNoContent: + return nil + case http.StatusNotFound: + return fmt.Errorf("%w: %s", ErrReverseForwardsUnsupported, responseError(resp)) + default: + return fmt.Errorf("reload-forwards: %s", responseError(resp)) + } +} + // Pending fetches the daemon's outstanding interactive holds. func (c *Client) Pending(ctx context.Context) ([]netfilter.Pending, error) { req, err := http.NewRequestWithContext(ctx, http.MethodGet, "http://vzd/v1/pending", nil) diff --git a/internal/vzdctl/vzdctl_test.go b/internal/vzdctl/vzdctl_test.go index 3816542..853e77c 100644 --- a/internal/vzdctl/vzdctl_test.go +++ b/internal/vzdctl/vzdctl_test.go @@ -283,3 +283,49 @@ func TestLifecycleVerbErrorSurfaces(t *testing.T) { // Pause has no handler wired — that specific verb is unsupported. require.ErrorIs(t, c.Pause(context.Background()), ErrLifecycleUnsupported) } + +func TestReloadForwardsRoundTrip(t *testing.T) { + sock := testSocket(t) + var reloaded int + srv, err := Start(sock, Handlers{ + Denials: func() []netfilter.Denial { return nil }, + Reload: func() error { return nil }, + ReloadForwards: func() error { reloaded++; return nil }, + }) + require.NoError(t, err) + t.Cleanup(func() { srv.Close() }) + + require.NoError(t, NewClient(sock).ReloadForwards(context.Background())) + require.Equal(t, 1, reloaded) +} + +// A daemon with no reverse-forward sink (firecracker, or one predating the +// feature) must say so rather than let the CLI report a live apply that +// never happened. +func TestReloadForwardsUnsupported(t *testing.T) { + sock := testSocket(t) + srv, err := Start(sock, Handlers{ + Denials: func() []netfilter.Denial { return nil }, + Reload: func() error { return nil }, + }) + require.NoError(t, err) + t.Cleanup(func() { srv.Close() }) + + err = NewClient(sock).ReloadForwards(context.Background()) + require.ErrorIs(t, err, ErrReverseForwardsUnsupported) +} + +func TestReloadForwardsErrorSurfaces(t *testing.T) { + sock := testSocket(t) + srv, err := Start(sock, Handlers{ + Denials: func() []netfilter.Denial { return nil }, + Reload: func() error { return nil }, + ReloadForwards: func() error { return errors.New("record vanished") }, + }) + require.NoError(t, err) + t.Cleanup(func() { srv.Close() }) + + err = NewClient(sock).ReloadForwards(context.Background()) + require.Error(t, err) + require.Contains(t, err.Error(), "record vanished") +}