Feat: Per-VM socket handover and control channel for MOCA SBX

CLAUDE.md

@@ -463,6 +463,94 @@ every plugin that has config. Counters are per-process and reset on restart
 
 **Plugin classification.** Protocol parsers (`mcp-parser`, `a2a-parser`, `inference-parser`) populate an `IsAction bool` field on their respective extensions to classify each request as either a user-meaningful action or protocol mechanics. Default-false means "not classified as action" — guardrails treat it as bypass. Parsers explicitly set `IsAction = true` for the small set of action methods (`tools/call` / `prompts/get` / `resources/read` for MCP; `message/send` / `message/stream` for A2A; every populated case for inference). Guardrails (`ibac` today; future rate limiters, audit loggers, etc.) read the aggregated verdict via `pctx.Classification()` which returns `(anyAction, anyBypass)`. A defense-in-depth guardrail skips on `anyBypass`, passes through on `!anyAction` (no parser claimed this traffic), and judges only when `anyAction && !anyBypass`. This puts the protocol-specific bypass-vs-action vocabulary in each parser — adding a new guardrail or new protocol does not multiply work at the guardrail layer. See [`docs/plugin-reference.md` "Classifying requests"](docs/plugin-reference.md#classifying-requests-as-actions-vs-protocol-mechanics) for the contract.
 
+### MOCA SBX VM control channel (`core/listener/vmcontrol`)
+
+For MOCA SBX deployments (#1477): SBX microVMs have **no network**. Each VM's
+egress lands on its **own host Unix socket**, bound by MOCA's `vmpool` (the only
+process that knows the jail layout) and handed to the Cortex forward proxy at
+run time over a control socket as SCM_RIGHTS. **The descriptor is the
+identity**: which listener accepted a connection says which VM is calling, and
+VMs come and go every few seconds, so listeners appear and disappear without a
+restart.
+
+**Environment (env-only for now — the conformance harness's channel; operator
+YAML follows when there is an operator to configure it):**
+
+| Variable | Meaning |
+|---|---|
+| `MOCA_EGRESS_CONTROL_SOCK` | The control socket path. Unset/empty = the channel is **off**, and today's behavior is unchanged byte for byte. |
+| `MOCA_EGRESS_CONTROL_UIDS` | Comma-separated uids allowed to connect to the control socket (peer-credential gate). Required when the socket is set — a control socket that admits nobody is a misconfiguration, and one that admits everybody would be worse. |
+| `MOCA_TENANCY` | `unset`/`single` = this slice; `multi` = **boot failure** naming the variable (single-tenant only, MI1 S2). Checked only when the socket is set. |
+
+**The control protocol** (`vmcontrol`): 4-byte big-endian length + protobuf,
+≤64 KiB frames; `ControlRequest{Register{vm_id, sid, grant} | Unregister{vm_id}}`
+(oneof; field 3 reserved for S2's Rebind — the decoder skips unknown fields),
+`ControlReply{ok, code, message}`; refusal codes `bad_request`,
+`duplicate_vm`, `handler_error`. A Register's fd rides as SCM_RIGHTS on the
+sendmsg carrying the frame's first byte. The protobuf codec is hand-rolled
+(moca's `gen/go` module is unpublished, so it cannot be imported) and pinned by
+golden tests against bytes generated by moca's own `proto.Marshal`. The whole
+package is ported from moca's `remote-worker/internal/egress/control` (also
+internal, also unimportable) — Unix-only (`SCM_RIGHTS`, `SO_PEERCRED` on
+Linux / `LOCAL_PEERCRED` on darwin, fail-closed on other unixes).
+
+**Semantics worth knowing:**
+
+- One `forwardproxy.Server` — one pipeline, one session store — serves every
+  VM listener. Identity flows through `Server.TrustedSessionID`: a hook the
+  vmcontrol handler installs that answers "which session is this request on"
+  from the request context (which socket accepted the connection). A non-empty
+  answer **wins over every header- and process-derived id** and is recorded as
+  `SessionSourceConnection` — a VM's traffic cannot be filed under a session
+  its guest merely names. The wrapper also strips `X-Moca-*`,
+  `Proxy-Authorization` and every session-id header from guest requests, so
+  nothing client-asserted reaches the pipeline or the upstream.
+  **The rule covers refusals, not only requests that reach the pipeline**, and
+  that took a second pass to get right: a request turned away *before*
+  hydration — an off-list egress destination on either the plain or CONNECT
+  path, a body too large to buffer, a bridged request whose `Host` is not its
+  tunnel's — records a row too, and those paths originally resolved by
+  header/process/`ActiveSession()`, filing a VM's denied-egress row in another
+  VM's session or in `default`. The connection's answer is now resolved once
+  per request (`trustedSessionOf`) and threaded into every pre-pipeline
+  recording site, with attribution pinned empty there as hydration does it, so
+  no VM's row is stamped with the host process that owns the control socket.
+  **It also survives `session.enabled: false`**: the trusted id needs no store
+  (`sessionViewFor` falls back to an ID-only view), so turning the store off
+  stops recording without un-identifying the VMs.
+- A Register carrying a non-empty **grant** is refused with `bad_request`
+  (single-tenant slice); tenancy is decided once at boot by `MOCA_TENANCY`.
+- **Unregister closes the listener and every live connection — hijacked
+  CONNECT tunnels included**, which `http.Server.Close()` alone would leave
+  open (hence `trackedListener`/`trackedConn`), and returns only after each
+  in-flight request's wrapper has emitted its terminal audit line. Bindings are
+  owned by the control connection that registered them: a vmpool crash or
+  restart unbinds its VMs automatically (`dropConn`), and a redialled
+  vmpool's Register supersedes the stale binding instead of failing as a
+  duplicate.
+- **Audit seed** (#1477 scope; #1482 completes it): one terminal JSON line per
+  request on stdout, in moca's `Event` shape (`vm_id`, `sid`, `host`,
+  `method`, `path`, `status`, `decision` of `tunnel`/`forward`/`aborted`,
+  `via: "moca-egress"`). `sub`, `dial_addr`, `credential`, byte counts and the
+  journald/async seam are #1482's. A request that arrives inside the
+  **aborted window** — after Unregister set the flag, before it closed the
+  conns — is answered `503` with `Connection: close` and emits its own
+  `aborted` line, rather than returning silently and letting `net/http` send
+  an empty `200` the guest would read as success.
+- **A temporary accept error on the control socket is retried**, 5ms doubling
+  to 1s, as `net/http`'s own accept loop does. Only a permanent error comes
+  back from `Serve` — and it must, because `startVMControl` hands that to
+  `fatalf`: an `EMFILE` on a proxy holding a listener per VM plus their
+  tunnels would otherwise exit the process and take every VM's egress with it.
+  Each per-VM `http.Server` also carries bootstrap's `ReadHeaderTimeout`
+  (10s), so a guest cannot hold a goroutine and an fd open by dripping
+  headers; per-VM caps are #1485's.
+- Local test runs: the macOS seatbelt sandbox refuses `AF_UNIX` binds, so
+  `server_test.go` skips there (CI's Linux runners run it); the registry and
+  wire tests run everywhere, with the registry tests driving the real
+  forwardproxy handler over TCP listeners standing in for the handed-over
+  Unix ones.
+
 ### init-iptables.sh
 
 Extensively documented shell script that sets up iptables for transparent traffic interception. Key features:

cmd/cortex/main.go

@@ -1042,6 +1042,13 @@ func main() {
 	// The transparent (enforce-redirect) listener rides with the forward proxy;
 	// declared here so shutdown can close it whether or not the forward role ran.
 	var transparentLn *net.TCPListener
+	// The MOCA SBX VM control channel (env-only, #1477): nil unless
+	// MOCA_EGRESS_CONTROL_SOCK is set. Like transparentLn, closed on shutdown.
+	var vmCtrl vmControlServer
+	ctrlCfg, ctrlErr := resolveControlConfig(os.Getenv)
+	if ctrlErr != nil {
+		fatalf("vm control: %v", ctrlErr)
+	}
 	if roles[config.RoleForward] {
 		fpSrv, ferr := forwardproxy.NewServer(outboundH, sessions, fpMTLS, forwardproxy.WithEgressGuard(egress))
 		if ferr != nil {
@@ -1118,6 +1125,21 @@ func main() {
 				slog.Info("process attribution on: header-less requests are filed by the process that sent them")
 			}
 		}
+		// The VM control channel serves each handed-over listener through this
+		// same forward proxy — one pipeline, one session store, identity from
+		// the connection — so it starts only where the forward role does.
+		//
+		// BEFORE the listeners below start serving. startVMControl assigns
+		// fpSrv.TrustedSessionID and fpSrv.AdoptBridgedRequest, which nothing
+		// locks; ServeHTTPServer hands its already-bound listener to a serving
+		// goroutine at once, so registering first would leave a window where a
+		// request reads those fields while this writes them — a data race, and
+		// one whose losing side resolves a VM's session from the guest's own
+		// headers. Both hooks are set before any request can reach them.
+		if ctrlCfg.Socket != "" {
+			vmCtrl = startVMControl(ctrlCfg, fpSrv)
+		}
+
 		httpServers = append(httpServers, bootstrap.ServeHTTPServer("forward-proxy", fpSrv.Handler(),
 			listeners.take("forward-proxy"), bootstrap.WithConnContext(fpSrv.ConnContext)))
 
@@ -1131,6 +1153,9 @@ func main() {
 			transparentLn = startTransparentProxy(fpSrv, listeners.takeTCP("transparent-proxy"))
 		}
 	}
+	if ctrlCfg.Socket != "" && vmCtrl == nil {
+		fatalf("vm control: %s is set but the forward proxy role is not enabled", envControlSock)
+	}
 
 	_ = mtlsMetrics // TODO Phase 2: surface metrics through /stats
 
@@ -1216,6 +1241,12 @@ func main() {
 		if transparentLn != nil {
 			_ = transparentLn.Close()
 		}
+		// Before the pipelines stop: closing unregisters every VM, closing its
+		// connections and emitting their aborted audit lines, and the requests
+		// it wakes still need a running pipeline to finish against.
+		if vmCtrl != nil {
+			_ = vmCtrl.Close()
+		}
 		statSrv.Shutdown(shutdownCtx)
 		healthSrv.Shutdown(shutdownCtx)
 		if sessionAPISrv != nil {
@@ -1230,6 +1261,9 @@ func main() {
 		if transparentLn != nil {
 			_ = transparentLn.Close()
 		}
+		if vmCtrl != nil {
+			_ = vmCtrl.Close()
+		}
 		_ = statSrv.Close()
 		_ = healthSrv.Close()
 		if sessionAPISrv != nil {

cmd/cortex/vmcontrol.go

@@ -0,0 +1,66 @@
+package main
+
+import (
+	"fmt"
+	"strconv"
+	"strings"
+)
+
+// The MOCA SBX VM control channel (#1477), configured env-only: the conformance
+// harness drives it through MOCA_EGRESS_CONTROL_SOCK / _UIDS, and operator-facing
+// YAML fields follow once there is an operator to configure them. Off unless
+// MOCA_EGRESS_CONTROL_SOCK names a path, so a proxy with no MOCA around it keeps
+// today's behavior byte for byte.
+
+const (
+	envTenancy     = "MOCA_TENANCY"
+	envControlSock = "MOCA_EGRESS_CONTROL_SOCK"
+	envControlUIDs = "MOCA_EGRESS_CONTROL_UIDS"
+)
+
+// controlConfig is the resolved VM control channel; a zero Socket means off.
+type controlConfig struct {
+	Socket      string
+	AllowedUIDs []int
+}
+
+// vmControlServer is the subset of the control server main needs — closing on
+// shutdown, which unregisters every VM and waits for their aborted audit lines.
+// An interface so the non-unix stub can exist without the package.
+type vmControlServer interface{ Close() error }
+
+// resolveControlConfig reads the VM control channel's environment through
+// getenv. Every error names its variable; an unset MOCA_EGRESS_CONTROL_SOCK
+// turns the whole channel off without looking at anything else, so the
+// tenancy check only fires where MOCA actually deployed this proxy.
+func resolveControlConfig(getenv func(string) string) (controlConfig, error) {
+	var cc controlConfig
+	cc.Socket = strings.TrimSpace(getenv(envControlSock))
+	if cc.Socket == "" {
+		return cc, nil // feature off
+	}
+	// Tenancy is decided once, here, not per request: a Register carrying a
+	// grant is refused by the registry with the same answer (MI1 S2).
+	switch t := getenv(envTenancy); t {
+	case "", "single":
+	case "multi":
+		return cc, fmt.Errorf("%s=multi: single-tenant only (MI1 S2)", envTenancy)
+	default:
+		return cc, fmt.Errorf("%s=%q: want unset or \"single\" (MI1 S2)", envTenancy, t)
+	}
+	// No default uid list: a control socket that admits nobody is a
+	// misconfiguration, and one that admits everybody would be worse.
+	raw := strings.TrimSpace(getenv(envControlUIDs))
+	if raw == "" {
+		return cc, fmt.Errorf("%s is required when %s is set", envControlUIDs, envControlSock)
+	}
+	for _, item := range strings.Split(raw, ",") {
+		item = strings.TrimSpace(item)
+		uid, err := strconv.Atoi(item)
+		if err != nil || uid < 0 {
+			return cc, fmt.Errorf("%s: %q is not a non-negative integer", envControlUIDs, item)
+		}
+		cc.AllowedUIDs = append(cc.AllowedUIDs, uid)
+	}
+	return cc, nil
+}

cmd/cortex/vmcontrol_other.go

@@ -0,0 +1,15 @@
+//go:build !unix
+
+package main
+
+import (
+	"github.com/rossoctl/cortex/core/listener/forwardproxy"
+)
+
+// startVMControl refuses: the per-VM socket handover needs SCM_RIGHTS and peer
+// credentials, so a configured channel on a non-Unix platform fails closed
+// rather than running silently without it.
+func startVMControl(cfg controlConfig, fp *forwardproxy.Server) vmControlServer {
+	fatalf("vm control: %s: per-VM socket handover requires a Unix platform (SCM_RIGHTS)", envControlSock)
+	return nil
+}

cmd/cortex/vmcontrol_test.go

@@ -0,0 +1,92 @@
+package main
+
+import (
+	"strings"
+	"testing"
+)
+
+func TestResolveControlConfig(t *testing.T) {
+	env := func(kv ...string) func(string) string {
+		m := map[string]string{}
+		for i := 0; i+1 < len(kv); i += 2 {
+			m[kv[i]] = kv[i+1]
+		}
+		return func(k string) string { return m[k] }
+	}
+
+	cases := []struct {
+		name    string
+		getenv  func(string) string
+		wantErr string // "" = no error; matched as a substring
+		wantOn  bool
+	}{
+		{
+			name:   "unset socket: off, tenancy not even read",
+			getenv: env(envTenancy, "multi"),
+		},
+		{
+			name:   "single tenancy",
+			getenv: env(envControlSock, "/run/c.sock", envControlUIDs, "501", envTenancy, "single"),
+			wantOn: true,
+		},
+		{
+			name:   "tenancy unset",
+			getenv: env(envControlSock, "/run/c.sock", envControlUIDs, "0"),
+			wantOn: true,
+		},
+		{
+			name:    "multi tenancy is a boot failure naming the variable",
+			getenv:  env(envControlSock, "/run/c.sock", envControlUIDs, "0", envTenancy, "multi"),
+			wantErr: envTenancy,
+		},
+		{
+			name:    "unknown tenancy value names the variable",
+			getenv:  env(envControlSock, "/run/c.sock", envControlUIDs, "0", envTenancy, "dual"),
+			wantErr: envTenancy,
+		},
+		{
+			name:    "socket without uids is a misconfiguration",
+			getenv:  env(envControlSock, "/run/c.sock"),
+			wantErr: envControlUIDs,
+		},
+		{
+			name:    "a non-integer uid names the variable",
+			getenv:  env(envControlSock, "/run/c.sock", envControlUIDs, "0,root"),
+			wantErr: envControlUIDs,
+		},
+		{
+			name:    "a negative uid is refused",
+			getenv:  env(envControlSock, "/run/c.sock", envControlUIDs, "-1"),
+			wantErr: envControlUIDs,
+		},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			cc, err := resolveControlConfig(tc.getenv)
+			if tc.wantErr == "" {
+				if err != nil {
+					t.Fatalf("unexpected error: %v", err)
+				}
+				if (cc.Socket != "") != tc.wantOn {
+					t.Errorf("on = %v (socket %q), want %v", cc.Socket != "", cc.Socket, tc.wantOn)
+				}
+				return
+			}
+			if err == nil {
+				t.Fatalf("want an error naming %s, got config %+v", tc.wantErr, cc)
+			}
+			if !strings.Contains(err.Error(), tc.wantErr) {
+				t.Errorf("error %q does not name %s", err, tc.wantErr)
+			}
+		})
+	}
+
+	// The uid list parses comma-separated with whitespace.
+	cc, err := resolveControlConfig(env(envControlSock, "/run/c.sock", envControlUIDs, "0, 501 ,2"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(cc.AllowedUIDs) != 3 || cc.AllowedUIDs[0] != 0 || cc.AllowedUIDs[1] != 501 || cc.AllowedUIDs[2] != 2 {
+		t.Errorf("uids = %v, want [0 501 2]", cc.AllowedUIDs)
+	}
+}

cmd/cortex/vmcontrol_unix.go

@@ -0,0 +1,37 @@
+//go:build unix
+
+package main
+
+import (
+	"log/slog"
+	"os"
+
+	"github.com/rossoctl/cortex/core/listener/forwardproxy"
+	"github.com/rossoctl/cortex/core/listener/vmcontrol"
+)
+
+// startVMControl brings up the VM control channel: the control socket vmpool
+// hands per-VM egress listeners over, a Registry binding each to fp, and the
+// trusted-session hook that makes which-socket-accepted-the-connection the
+// session identity for everything a VM sends.
+func startVMControl(cfg controlConfig, fp *forwardproxy.Server) vmControlServer {
+	// The listener's own answer, never the request's: see TrustedSessionID.
+	fp.TrustedSessionID = vmcontrol.SessionIDFromRequest
+	// A TLS-bridged tunnel's decrypted requests are parsed fresh and bypass the
+	// per-VM wrapper, so the binding has to be re-stamped onto each of them or
+	// the guest's own session headers decide. See AdoptBridgedRequest.
+	fp.AdoptBridgedRequest = vmcontrol.AdoptBridgedRequest
+	reg := vmcontrol.NewRegistry(fp, vmcontrol.NewJSONSink(os.Stdout))
+	srv, err := vmcontrol.NewServer(cfg.Socket, cfg.AllowedUIDs, reg)
+	if err != nil {
+		fatalf("vm control: %v", err)
+	}
+	go func() {
+		if err := srv.Serve(); err != nil {
+			fatalf("vm control: %v", err)
+		}
+	}()
+	slog.Info("vm control listening — MOCA SBX per-VM egress sockets",
+		"socket", cfg.Socket, "allowed_uids", cfg.AllowedUIDs)
+	return srv
+}