diff --git a/CLAUDE.md b/CLAUDE.md index 226beb1f3..230fc405b 100644 --- a/CLAUDE.md +++ b/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: diff --git a/cmd/cortex/main.go b/cmd/cortex/main.go index 2e613ada4..17c13fe27 100644 --- a/cmd/cortex/main.go +++ b/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 { diff --git a/cmd/cortex/vmcontrol.go b/cmd/cortex/vmcontrol.go new file mode 100644 index 000000000..65360bb66 --- /dev/null +++ b/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 +} diff --git a/cmd/cortex/vmcontrol_other.go b/cmd/cortex/vmcontrol_other.go new file mode 100644 index 000000000..5f5069ce8 --- /dev/null +++ b/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 +} diff --git a/cmd/cortex/vmcontrol_test.go b/cmd/cortex/vmcontrol_test.go new file mode 100644 index 000000000..b3c1abdc2 --- /dev/null +++ b/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) + } +} diff --git a/cmd/cortex/vmcontrol_unix.go b/cmd/cortex/vmcontrol_unix.go new file mode 100644 index 000000000..c257a34d1 --- /dev/null +++ b/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 +} diff --git a/core/listener/forwardproxy/server.go b/core/listener/forwardproxy/server.go index 59bce0e2e..a7ab56696 100644 --- a/core/listener/forwardproxy/server.go +++ b/core/listener/forwardproxy/server.go @@ -175,6 +175,31 @@ type Server struct { // resolution byte for byte. Processes peerproc.Resolver + // TrustedSessionID, when set, is the listener's own answer to "which + // session is this request on" — resolved from how the connection arrived + // (which per-VM socket accepted it, vmcontrol), not from anything the + // client sent. A non-empty answer wins over every header- and + // process-derived id, and the client's own session headers never override + // it. #1476 will generalise this into full connection-derived identity; + // this is deliberately the narrow seam it needs. + TrustedSessionID func(r *http.Request) string + + // AdoptBridgedRequest, when set, re-applies a connection-derived identity + // to each request a TLS-bridged tunnel decrypts, given the CONNECT that + // opened it. It exists because a bridged request is NOT the CONNECT: the + // bytes are re-parsed into fresh *http.Requests inside tlsbridge.ServeConn's + // own http.Server, so neither the CONNECT's context nor any header the + // accepting listener stripped survives into them. A listener that derives + // identity from the connection (vmcontrol: which per-VM socket accepted it) + // must therefore be given a second chance to stamp it, or TrustedSessionID + // answers "" on the decrypted request and resolution falls through to the + // guest's own session headers — which is the whole seam inverted. + // + // connect is the CONNECT request as the listener's wrapper saw it, so a + // hook can read the identity it stamped there; r is the decrypted request + // to decorate. The returned request replaces r. + AdoptBridgedRequest func(connect, r *http.Request) *http.Request + // bufferedFallbackOnce keeps the SSE-buffered-path notice to one line per // process; the condition is a supported chain shape, not an error. bufferedFallbackOnce sync.Once @@ -442,6 +467,19 @@ func (s *Server) serveOutbound(w http.ResponseWriter, r *http.Request, tl *tunne // — and silent on the day one does, which is why it is pinned rather than watched. pctx.ResolveClient() + // The connection's own answer to "which session", resolved HERE rather than at + // hydration, because a request can be refused before it ever reaches hydration + // — an off-list egress destination, a body too large to buffer, a bridged + // request whose Host is not its tunnel's. Each of those records a row, and a + // row filed by header/process/ActiveSession() for a VM whose identity the + // connection already settled lands in another VM's session or in `default`: + // the denied-egress event you most want attributed correctly, attributed to + // someone else. Resolving once up here and threading it into every recording + // site is what makes "identity from the connection wins over every header- and + // process-derived id" true for refusals too, not just for requests that make + // it to the pipeline (#1569 review). + trusted := s.trustedSessionOf(r) + // A bridged request is dialed to the tunnel's authority, so policy must judge that // host. The client chose both the CONNECT target and the Host header, and a Host // naming anywhere else would have every host-keyed plugin, skip_hosts included, @@ -470,7 +508,7 @@ func (s *Server) serveOutbound(w http.ResponseWriter, r *http.Request, tl *tunne host := hostOnly(r.URL.Host) action := egressAllowRefusal(pctx, host) chain := s.clientChain(r) - s.recordOutboundRejectIn(tl, pctx, action, s.recordingSessionID("", r.Header, chain)) + s.recordOutboundRejectIn(tl, pctx, action, s.recordingSessionID(trusted, r.Header, chain)) // Sanitized: host (the URL authority), Method and Path are all client-controlled, // and a control rune forged into any of them would forge a log entry — the same // CWE-150 shape #1539's egress guard was flagged for. @@ -559,9 +597,15 @@ func (s *Server) serveOutbound(w http.ResponseWriter, r *http.Request, tl *tunne // its paired response — which resolveOutboundSessionID's own contract forbids, // and which would fail invisibly. The listener's answer wins; the pin is // assigned from this local after the pipeline has run. + // The store's absence gates the header/process resolution, not the trusted + // answer: that one costs no store (sessionViewFor falls back to an ID-only + // view) and is the VM's identity, which plugins must see whether or not + // anyone is recording events. Throwing it away under `session.enabled: + // false` would quietly un-identify every VM in a MOCA deployment that turned + // the store off (#1569 review). var sessionID string - if !skipped && s.Sessions != nil { - sessionID = s.hydrateSession(pctx, r.Header, chain) + if !skipped && (s.Sessions != nil || trusted != "") { + sessionID = s.hydrateSession(pctx, trusted, r.Header, chain) } if !skipped { @@ -1329,7 +1373,7 @@ func (s *Server) refuseMisdirected(w http.ResponseWriter, r *http.Request, pctx } pctx.Host = r.URL.Host chain := s.clientChain(r) - sid := s.attributeForRecording(pctx, r.Header, chain) + sid := s.attributeForRecording(pctx, s.trustedSessionOf(r), r.Header, chain) pctx.OutboundSessionID = sid if chain != nil { defer s.touch(sid, chain) @@ -1490,6 +1534,15 @@ func (s *Server) bridgedHandler(authority string, tl *tunnelLog) http.Handler { if tl.conn != nil { r = r.WithContext(context.WithValue(r.Context(), connProcKey{}, tl.conn)) } + // Before anything reads identity off r. A decrypted request is a NEW + // request — the accepting listener's wrapper never saw it, so whatever + // it stamped on or stripped from the CONNECT has to be redone here, or + // the guest's own headers decide the session (#1569 review). + if s.AdoptBridgedRequest != nil && tl.connect != nil { + if adopted := s.AdoptBridgedRequest(tl.connect, r); adopted != nil { + r = adopted + } + } r.URL.Scheme = "https" r.URL.Host = authority // host:port — preserves non-443 origins s.serveOutbound(w, r, tl) @@ -1542,10 +1595,38 @@ func (s *Server) recordingSessionID(resolved string, clientHeaders http.Header, return s.resolveOutboundSessionID(clientHeaders, chain) } +// trustedSessionOf is the listener's own answer to "which session is this request +// on", from how the connection arrived (vmcontrol installs the hook; the socket +// that accepted the connection names the VM). Empty when no hook is installed, or +// when it does not recognise this request — a transparently redirected connection +// has no request at all. +// +// Separate from hydrateSession so the pre-pipeline refusal paths can ask the same +// question: those never reach hydration, and a refusal filed under a header- or +// process-derived id defeats the point of deriving identity from the connection. +func (s *Server) trustedSessionOf(r *http.Request) string { + if s.TrustedSessionID == nil || r == nil { + return "" + } + return s.TrustedSessionID(r) +} + // attributeForRecording resolves a request that never reached hydration — refused before the // pipeline, or unreadable — pins its attribution on pctx so its rows carry the process, and -// returns the bucket they go in: the attribution's, else ActiveSession(), else default. -func (s *Server) attributeForRecording(pctx *pipeline.Context, clientHeaders http.Header, chain []attribution.Proc) string { +// returns the bucket they go in: the trusted sid, else the attribution's, else +// ActiveSession(), else default. +// +// A trusted sid SHORT-CIRCUITS, and pins attribution EMPTY rather than skipping the +// pin: SetAttribution is write-once and AttributedProc() is recorded straight into +// SessionEvent.Proc, so claiming the slot with nothing is what stops a VM's refusal +// row from being stamped with the host process that merely happens to own the +// control socket — the same reasoning hydrateSession's trusted branch gives, applied +// to the paths that never get there (#1569 review). +func (s *Server) attributeForRecording(pctx *pipeline.Context, trusted string, clientHeaders http.Header, chain []attribution.Proc) string { + if trusted != "" { + pctx.SetAttribution("", "", "") + return trusted + } at := s.attribute(clientHeaders, chain) pctx.SetAttribution(at.Proc, at.ClientProc, at.Header) if sid := at.RecordID(); sid != "" { @@ -1742,7 +1823,7 @@ func (s *Server) recordUnbufferedRequest(tl *tunnelLog, pctx *pipeline.Context, if s.Sessions == nil { return } - sid := s.attributeForRecording(pctx, r.Header, chain) + sid := s.attributeForRecording(pctx, s.trustedSessionOf(r), r.Header, chain) pctx.OutboundSessionID = sid if chain != nil { defer s.touch(sid, chain) @@ -1834,7 +1915,42 @@ type pluginSession struct { // hydrateSession resolves the plugin session for a request and, when there is one, // sets pctx.Session and how it was resolved. It returns the id for the caller's // local, which recording reads instead of pctx. -func (s *Server) hydrateSession(pctx *pipeline.Context, clientHeaders http.Header, chain []attribution.Proc) string { +// trusted is the connection's own answer (trustedSessionOf), resolved by the +// caller — which also keeps the r.Header read at the CALL SITE rather than in +// here. The headers arrive as a parameter for the same reason trusted does: +// the caller resolves both once per request, so the pre-hydration refusal +// paths (attributeForRecording and its call sites) share the caller's values +// instead of re-deriving them. CodeQL note, so nobody re-litigates the +// placement: moving the r.Header read inside this function made CodeQL's +// diff-informed analysis report 28 pre-existing go/clear-text-logging and +// go/log-injection alerts in untouched plugins (sessionbudget, contextguru, +// inferencerouter) as new, every one sourced to the moved line. They are +// false positives — a header id passes validHeaderSessionID before anything +// logs it, and a session id is not a secret — and taking the read back out +// did NOT clear them; they need dismissing on the PR (#1569 review). +func (s *Server) hydrateSession(pctx *pipeline.Context, trusted string, clientHeaders http.Header, chain []attribution.Proc) string { + // The listener's own answer first, and it ends the search: a session + // resolved from how the connection arrived is not something the client has + // any say in, and letting a header or a process chain override it would + // defeat the point of deriving identity from the connection (vmcontrol). + // Hydration still happens — plugins get their pctx.Session view — but the + // claims and affinity work below is skipped: a VM's traffic must not be + // filed by a process it never ran as. A transparently redirected connection + // has no request at all (r == nil), so there is nothing to claim either. + // + // Attribution is pinned EMPTY rather than left unset. SetAttribution is + // write-once and AttributedProc() is recorded straight into + // SessionEvent.Proc, so claiming the slot here is what stops a later + // attributeForRecording from filling a VM's row in with a host process + // that merely happens to own the control socket. The VM's identity is the + // session id itself, which the connection already settled. + if trusted != "" { + pctx.Session = s.sessionViewFor(trusted) + pctx.SessionSource = pipeline.SessionSourceConnection + pctx.SessionHeader = "" + pctx.SetAttribution("", "", "") + return trusted + } at := s.attribute(clientHeaders, chain) pctx.SetAttribution(at.Proc, at.ClientProc, at.Header) sid, how := s.pluginSessionOf(at) @@ -2539,6 +2655,11 @@ func (s *Server) handleConnect(w http.ResponseWriter, r *http.Request) { // attributed to the agent that opened it. pctx.ResolveClient() + // The connection's own answer, resolved once for this CONNECT — the egress + // refusal below is recorded before hydration, so it needs the same trusted + // sid the hydrated path gets (see serveOutbound's, same reasoning). + trusted := s.trustedSessionOf(r) + // Egress allow list (#1483): when on, an off-list CONNECT target is refused // here — before the SkipHosts match, so a skip entry cannot exempt an // off-list host (the allowlist is the outer boundary), and before the @@ -2553,10 +2674,14 @@ func (s *Server) handleConnect(w http.ResponseWriter, r *http.Request) { host := hostOnly(r.Host) action := egressAllowRefusal(pctx, host) // Session resolution the recordUnbufferedRequest way — pre-pipeline, - // from the client's headers and process only — so the denied row lands - // in the bucket the request would have recorded under. + // from the connection's own answer first, then the client's headers and + // process — so the denied row lands in the bucket the request would have + // recorded under. The trusted sid has to come first here for the same + // reason it does on the plain path: this CONNECT is refused before + // hydration, and a VM's denied row filed by header or process lands in + // another VM's session (#1569 review). chain := s.clientChain(r) - s.recordOutboundReject(pctx, action, s.recordingSessionID("", r.Header, chain)) + s.recordOutboundReject(pctx, action, s.recordingSessionID(trusted, r.Header, chain)) // Sanitized like the serveOutbound refusal: host is client-controlled and // lands in a log file. slog.Info("forward-proxy: CONNECT destination refused by egress allow list", @@ -2601,8 +2726,10 @@ func (s *Server) handleConnect(w http.ResponseWriter, r *http.Request) { // announces its session on CONNECT is honored; most do not, and those // fall through to ActiveSession() exactly as before. chain = s.clientChain(r) - if s.Sessions != nil { - sessionID = s.hydrateSession(pctx, r.Header, chain) + // Same widening as the request path: the trusted answer needs no store, + // and a VM's tunnel must carry its identity even with the store off. + if s.Sessions != nil || trusted != "" { + sessionID = s.hydrateSession(pctx, trusted, r.Header, chain) } // Run the outbound pipeline. Plugins that policy on host/identity @@ -2627,9 +2754,13 @@ func (s *Server) handleConnect(w http.ResponseWriter, r *http.Request) { // request files under this pin, when it settles. Pinned only now, after the pipeline, // for the reason sessionID is a local; left unpinned where affinity had no answer, for // the reason in tunnelSessionID. See recordTunnelOpened. With process attribution the - // pin is the process's answer, which is never empty. + // pin is the process's answer, which is never empty. A trusted connection answer is + // pinned ahead of both, whatever those knobs say: identity from the connection must + // not need session.client_affinity to reach a tunnel's own rows (#1569 review). if !skipped && s.Sessions != nil { switch { + case trusted != "": + pctx.OutboundSessionID = trusted case chain != nil && s.processesOn(): // The client's process is known, so the tunnel's own row goes where that // process's requests go — default included, for a process of no agent. @@ -2659,6 +2790,10 @@ func (s *Server) handleConnect(w http.ResponseWriter, r *http.Request) { // plugin ran, so there is nothing to attribute them to. tl := s.newTunnelLog(pctx, skipped) tl.conn = connProcOf(r.Context()) + // Kept for AdoptBridgedRequest: the requests this tunnel decrypts are parsed + // fresh and inherit nothing from here, so a connection-derived identity has + // to be re-stamped from the CONNECT onto each of them. + tl.connect = r // Plain TCP dial. See package-level comment on why mTLS doesn't // apply here. r.Host on a CONNECT carries "host:port" already. @@ -3177,6 +3312,10 @@ type tunnelLog struct { // conn is the CONNECT's client process slot, handed to the requests a bridged tunnel // decrypts; nil off the CONNECT path. conn *connProc + // connect is the CONNECT request itself, kept for AdoptBridgedRequest so a + // listener that derived identity from the accepting connection can stamp it + // onto each decrypted request too. nil off the CONNECT path. + connect *http.Request // program is the client's program when the bridge rules named it, nil when they did // not. bridgeServe records a handshake's failure against it when set and against the // host when not, which is what the transparent listener and a nameless client get. diff --git a/core/listener/forwardproxy/session_source_test.go b/core/listener/forwardproxy/session_source_test.go index 27fcf31a9..9e24c23ba 100644 --- a/core/listener/forwardproxy/session_source_test.go +++ b/core/listener/forwardproxy/session_source_test.go @@ -58,6 +58,52 @@ func TestSessionSourceThroughTheHandler(t *testing.T) { } } +// TrustedSessionID, when it answers, ends the search: the session is resolved +// from how the connection arrived, and a header naming another session — or +// the process chain, or affinity — must not override it. vmcontrol's registry +// tests drive the same seam end to end; this pins the resolution itself. +func TestTrustedSessionIDOverridesEveryOtherSource(t *testing.T) { + store := session.New(5*time.Minute, 100, 0) + defer store.Close() + s := &Server{ + Sessions: store, + SessionIDHeaders: []string{session.ClaudeCodeSessionHeader}, + TrustedSessionID: func(r *http.Request) string { + if r.Header.Get("X-Which-VM") == "vm-a" { + return "trusted-sid" + } + return "" + }, + } + agent := attribution.Proc{PID: 100, Start: 1} + store.Append("trusted-sid", pipeline.SessionEvent{At: time.Now(), Phase: pipeline.SessionRequest}) + + h := http.Header{} + h.Set(session.ClaudeCodeSessionHeader, "header-sid") + h.Set("X-Which-VM", "vm-a") + pctx := &pipeline.Context{} + if sid := s.hydrateSession(pctx, s.trustedSessionOf(&http.Request{Header: h}), h, []attribution.Proc{agent}); sid != "trusted-sid" { + t.Fatalf("session %q, want the trusted trusted-sid", sid) + } + if pctx.Session == nil || pctx.Session.ID != "trusted-sid" { + t.Fatalf("pctx.Session = %+v, want trusted-sid", pctx.Session) + } + if pctx.SessionSource != pipeline.SessionSourceConnection || pctx.SessionHeader != "" { + t.Errorf("source %q/%q, want connection with no header named", pctx.SessionSource, pctx.SessionHeader) + } + + // A request the hook does not claim falls through to the old resolution + // unchanged. + h.Del("X-Which-VM") + pctx = &pipeline.Context{} + if sid := s.hydrateSession(pctx, s.trustedSessionOf(&http.Request{Header: h}), h, nil); sid != "header-sid" { + t.Fatalf("unclaimed request: session %q, want header-sid", sid) + } + if pctx.SessionSource != pipeline.SessionSourceHeader { + t.Errorf("unclaimed request: source %q, want header", pctx.SessionSource) + } +} + // A process that never sent a session header takes the session its ancestor // named, and that is recorded as a process resolution. func TestSessionSourceProcessChain(t *testing.T) { @@ -78,7 +124,7 @@ func TestSessionSourceProcessChain(t *testing.T) { t.Fatalf("agent: %q via %q, want claude-1 via header", sid, how.source) } pctx := &pipeline.Context{} - if sid := s.hydrateSession(pctx, http.Header{}, []attribution.Proc{tool, agent}); sid != "claude-1" { + if sid := s.hydrateSession(pctx, "", http.Header{}, []attribution.Proc{tool, agent}); sid != "claude-1" { t.Fatalf("tool: session %q, want claude-1", sid) } if pctx.SessionSource != pipeline.SessionSourceProcess || pctx.SessionHeader != "" { @@ -107,14 +153,16 @@ func TestChainSessionOverridesAChildsOwnSession(t *testing.T) { return h } pctx := &pipeline.Context{} - s.hydrateSession(pctx, named("claude-1"), []attribution.Proc{agent, shell}) + hAgent := named("claude-1") + s.hydrateSession(pctx, s.trustedSessionOf(&http.Request{Header: hAgent}), hAgent, []attribution.Proc{agent, shell}) store.Append("claude-1", pipeline.SessionEvent{At: time.Now(), Phase: pipeline.SessionRequest}) if pctx.ChainSession != "" { t.Errorf("agent: ChainSession %q, want none", pctx.ChainSession) } pctx = &pipeline.Context{} - sid := s.hydrateSession(pctx, named("fresh"), []attribution.Proc{child, shell, agent}) + hChild := named("fresh") + sid := s.hydrateSession(pctx, s.trustedSessionOf(&http.Request{Header: hChild}), hChild, []attribution.Proc{child, shell, agent}) store.Append("fresh", pipeline.SessionEvent{At: time.Now(), Phase: pipeline.SessionRequest}) if sid != "fresh" || pctx.SessionSource != pipeline.SessionSourceHeader { t.Fatalf("child: recorded under %q via %q, want fresh via header", sid, pctx.SessionSource) @@ -124,7 +172,7 @@ func TestChainSessionOverridesAChildsOwnSession(t *testing.T) { } pctx = &pipeline.Context{} - if sid := s.hydrateSession(pctx, http.Header{}, []attribution.Proc{child, shell, agent}); sid != "fresh" { + if sid := s.hydrateSession(pctx, "", http.Header{}, []attribution.Proc{child, shell, agent}); sid != "fresh" { t.Fatalf("child without header: %q, want its own claim fresh", sid) } if pctx.ChainSession != "claude-1" { diff --git a/core/listener/forwardproxy/transparent.go b/core/listener/forwardproxy/transparent.go index 151a242a3..caa29e799 100644 --- a/core/listener/forwardproxy/transparent.go +++ b/core/listener/forwardproxy/transparent.go @@ -118,7 +118,7 @@ func (s *Server) HandleTransparentConn(clientConn net.Conn, dst string) { // would reopen the same flip window the other paths were fixed for, reduced to // ActiveSession@T0 versus ActiveSession@T1. resolvePluginSessionID with nil // headers is that same resolver with nothing to read a header from. - sessionID := s.hydrateSession(pctx, nil, nil) + sessionID := s.hydrateSession(pctx, "", nil, nil) // Gate on host/identity before opening the tunnel — identical to the // CONNECT path. Parsers see no body and degrade gracefully. @@ -290,7 +290,8 @@ func (s *Server) recordTunnelOpened(pctx *pipeline.Context, reason pipeline.Tunn // row goes to, and returns its bucket; nil when session tracking is off. It is // recordTunnelOpened's filing rule on its own, for a bridged tunnel that settles with no // request having taken its deferred open (tunnelLog.settleLocked). Of pctx it reads only -// OutboundSessionID, which the listener pins before bridging and nothing writes after. +// OutboundSessionID — which the listener pins before bridging and nothing writes after — +// and SessionSource, to recognize the connection's own trusted answer. func (s *Server) appendTunnelOpen(pctx *pipeline.Context, ev pipeline.SessionEvent) *session.Bucket { if s.Sessions == nil { return nil @@ -298,9 +299,11 @@ func (s *Server) appendTunnelOpen(pctx *pipeline.Context, ev pipeline.SessionEve // Without client affinity this reads ActiveSession() at recording time, as it always // has, ignoring the identity the tunnel was gated under (#1187). With it, the pin set // after gating wins; see handleConnect. Process attribution pins it too, with the - // client process's answer. + // client process's answer, and so does the connection itself: a session resolved from + // how the connection arrived is not a knob like affinity, so its pin is honored even + // with affinity and process attribution both off (#1569 review). var sid string - if s.ClientAffinity || s.processesOn() { + if s.ClientAffinity || s.processesOn() || pctx.SessionSource == pipeline.SessionSourceConnection { sid = pctx.OutboundSessionID } if sid == "" { diff --git a/core/listener/vmcontrol/audit.go b/core/listener/vmcontrol/audit.go new file mode 100644 index 000000000..b3638736d --- /dev/null +++ b/core/listener/vmcontrol/audit.go @@ -0,0 +1,86 @@ +package vmcontrol + +import ( + "encoding/json" + "io" + "net/url" + "sync" + "time" +) + +// Decision values recorded in Event.Decision. This slice uses the three +// terminal decisions a VM listener's own wrapper emits; the full moca set +// (destination_denied, conn_cap, credential_*, …) arrives with #1482, which +// also fills the fields left empty here. +const ( + DecisionTunnel = "tunnel" + DecisionForward = "forward" + DecisionAborted = "aborted" +) + +// viaName is the fixed value of Event.Via. +const viaName = "moca-egress" + +// Event is one audit record, in moca's shape (remote-worker/internal/egress +// audit.go): same fields, same JSON tags, so a reader built for moca-egress +// lines parses ours. Path carries no query string, fragment or userinfo (see +// RedactPath); Credential is a name, never a value. +type Event struct { + Time time.Time `json:"time"` + Via string `json:"via"` + VMID string `json:"vm_id"` + SID string `json:"sid"` + Sub string `json:"sub,omitempty"` + Host string `json:"host"` + DialAddr string `json:"dial_addr,omitempty"` + Method string `json:"method"` + Path string `json:"path,omitempty"` + Status int `json:"status"` + BytesUp int64 `json:"bytes_up"` + BytesDown int64 `json:"bytes_down"` + Credential string `json:"credential,omitempty"` + Decision string `json:"decision"` + Dropped uint64 `json:"dropped,omitempty"` + Coalesced uint64 `json:"coalesced,omitempty"` +} + +// AuditSink receives audit events. +type AuditSink interface{ Emit(Event) } + +// jsonSink writes one JSON line per event, synchronously. moca wraps this in +// an async sink because its stdout can stall journald; the seam that matters +// here is the Event shape, and the async layer is #1482's to add where the +// sink actually lands. +type jsonSink struct { + mu sync.Mutex + enc *json.Encoder +} + +// NewJSONSink returns a sink writing one JSON line per event to w. Via is +// forced to "moca-egress" and a zero Time is filled with the current UTC time. +// Write errors are ignored: auditing must not stall the proxy. +func NewJSONSink(w io.Writer) AuditSink { + return &jsonSink{enc: json.NewEncoder(w)} +} + +func (s *jsonSink) Emit(e Event) { + e.Via = viaName + if e.Time.IsZero() { + e.Time = time.Now().UTC() + } + s.mu.Lock() + defer s.mu.Unlock() + _ = s.enc.Encode(e) +} + +// RedactPath returns the escaped path of u and nothing else: no query, +// fragment or userinfo. An empty path or nil u yields "/". +func RedactPath(u *url.URL) string { + if u == nil { + return "/" + } + if p := u.EscapedPath(); p != "" { + return p + } + return "/" +} diff --git a/core/listener/vmcontrol/bridge_binding_test.go b/core/listener/vmcontrol/bridge_binding_test.go new file mode 100644 index 000000000..bea8d807f --- /dev/null +++ b/core/listener/vmcontrol/bridge_binding_test.go @@ -0,0 +1,229 @@ +package vmcontrol + +import ( + "bufio" + "crypto/tls" + "crypto/x509" + "encoding/pem" + "fmt" + "net" + "net/http" + "net/http/httptest" + "strconv" + "strings" + "sync" + "testing" + "time" + + "github.com/rossoctl/cortex/core/listener/forwardproxy" + "github.com/rossoctl/cortex/core/pipeline" + "github.com/rossoctl/cortex/core/session" + "github.com/rossoctl/cortex/core/tlsbridge" +) + +// newBridgedVMEgressStack is newVMEgressStack with a TLS bridge configured, so +// a guest's CONNECT to the origin is decrypted and its inner requests run +// through the forward proxy's own handler rather than being relayed opaquely. +// Returns the stack plus the origin's authority and the root pool a guest needs +// to trust the forged leaf. +func newBridgedVMEgressStack(t *testing.T) (*Registry, *captureSink, *session.Store, *sessionProbe, string, *x509.CertPool, *http.Header) { + t.Helper() + store := session.New(5*time.Minute, 100, 0) + t.Cleanup(store.Close) + probe := &sessionProbe{} + p, err := pipeline.New([]pipeline.Plugin{probe}) + if err != nil { + t.Fatal(err) + } + + var originMu sync.Mutex + originSaw := http.Header{} + origin := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + originMu.Lock() + for k, v := range r.Header { + originSaw[k] = v + } + originMu.Unlock() + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte("ok")) + })) + t.Cleanup(origin.Close) + authority := strings.TrimPrefix(origin.URL, "https://") + + originCAPEM := pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: origin.Certificate().Raw}) + src, err := tlsbridge.NewEphemeralSource() + if err != nil { + t.Fatalf("NewEphemeralSource: %v", err) + } + up, err := tlsbridge.NewUpstreamClient(originCAPEM, false) + if err != nil { + t.Fatalf("NewUpstreamClient: %v", err) + } + port, err := strconv.Atoi(portOfAuthority(authority)) + if err != nil { + t.Fatalf("port of %q: %v", authority, err) + } + decision, err := tlsbridge.NewDecision(tlsbridge.DecisionOpts{Ports: map[int]bool{port: true}}) + if err != nil { + t.Fatalf("NewDecision: %v", err) + } + engine := &tlsbridge.Engine{ + Decision: decision, + Term: tlsbridge.NewTerminator(tlsbridge.NewMinter(src, tlsbridge.MinterOpts{})), + Skip: tlsbridge.NewSkipSet(), + Upstream: up, + CAPEM: src.CACertPEM(), + } + issuer, _ := src.Issuer() + roots := x509.NewCertPool() + roots.AddCert(issuer) + roots.AddCert(origin.Certificate()) + + fp := &forwardproxy.Server{ + OutboundPipeline: pipeline.NewHolder(p), + Sessions: store, + Client: http.DefaultClient, + SessionIDHeaders: []string{session.ClaudeCodeSessionHeader, session.SessionIDHeader}, + TrustedSessionID: SessionIDFromRequest, + // The wiring under test, matching cmd/cortex/vmcontrol_unix.go. + AdoptBridgedRequest: AdoptBridgedRequest, + TLSBridge: engine, + } + sink := &captureSink{} + return NewRegistry(fp, sink), sink, store, probe, authority, roots, &originSaw +} + +// bridgedGuestGet opens a CONNECT through a VM socket, completes TLS against +// the bridge's forged leaf, and sends one request inside the decrypted stream +// carrying the given headers. It returns the inner response. +func bridgedGuestGet(t *testing.T, vmAddr, authority string, roots *x509.CertPool, path string, headers map[string]string) *http.Response { + t.Helper() + raw := guest(t, vmAddr) + resp := guestConnect(t, raw, authority) + if resp.StatusCode != http.StatusOK { + t.Fatalf("CONNECT answered %d, want 200", resp.StatusCode) + } + + tc := tls.Client(raw, &tls.Config{ServerName: hostOf(authority), RootCAs: roots}) + if err := tc.Handshake(); err != nil { + t.Fatalf("TLS handshake over the tunnel: %v", err) + } + t.Cleanup(func() { _ = tc.Close() }) + + var b strings.Builder + fmt.Fprintf(&b, "GET %s HTTP/1.1\r\nHost: %s\r\n", path, authority) + for k, v := range headers { + fmt.Fprintf(&b, "%s: %s\r\n", k, v) + } + b.WriteString("Connection: close\r\n\r\n") + if _, err := tc.Write([]byte(b.String())); err != nil { + t.Fatalf("write inner request: %v", err) + } + inner, err := http.ReadResponse(bufio.NewReader(tc), nil) + if err != nil { + t.Fatalf("read inner response: %v", err) + } + return inner +} + +func hostOf(authority string) string { + if h, _, err := net.SplitHostPort(authority); err == nil { + return h + } + return authority +} + +func portOfAuthority(authority string) string { + if _, p, err := net.SplitHostPort(authority); err == nil { + return p + } + return "443" +} + +// A DECRYPTED request from a VM guest must be bound to its VM exactly as a +// plaintext one is. The bridged path parses its requests off the forged TLS +// connection inside a fresh http.Server (tlsbridge.ServeConn), so the request +// context the per-VM wrapper stamped on the CONNECT does not reach them on its +// own — and without the binding, hydrateSession falls through to header +// resolution and a guest's own X-Claude-Code-Session-Id picks the session. +// +// That is the whole point of the feature inverted: the descriptor is supposed +// to BE the identity. The plaintext case is covered by +// TestRegistryIdentityFromConnection; this is the same contract one layer +// down, where TestRegistryIdentityFromConnection's stack has no bridge and so +// could never have caught it (#1569 review). +func TestBridgedRequestKeepsTheVMBinding(t *testing.T) { + reg, sink, store, probe, authority, roots, originSaw := newBridgedVMEgressStack(t) + vmA := registerTCPVM(t, reg, "vm-a", "sid-a") + + resp := bridgedGuestGet(t, vmA, authority, roots, "/bridged", map[string]string{ + // Everything a guest might assert to escape its own session. + session.ClaudeCodeSessionHeader: "forged-cc", + session.SessionIDHeader: "forged-x", + "X-Moca-Session": "sid-b", + "X-Moca-Vm-Id": "vm-b", + "Proxy-Authorization": "Basic Zm9yZ2Vk", + }) + if resp.StatusCode != http.StatusOK { + t.Fatalf("inner response %d, want 200", resp.StatusCode) + } + _ = resp.Body.Close() + + // The pipeline must have heard the REGISTERED sid, from the connection. + probe.mu.Lock() + ids, sources, headers := probe.ids, probe.sources, probe.headers + probe.mu.Unlock() + if len(ids) == 0 { + t.Fatal("no pipeline pass recorded for the decrypted request") + } + for i, id := range ids { + if id != "sid-a" || sources[i] != pipeline.SessionSourceConnection || headers[i] != "" { + t.Errorf("decrypted request %d: plugin saw session %q via %q/%q, want sid-a via connection", + i, id, sources[i], headers[i]) + } + } + + // Nothing the guest named may hold rows. + for _, forged := range []string{"forged-cc", "forged-x", "sid-b", "default"} { + if v := store.View(forged); v != nil && len(v.Events) > 0 { + t.Errorf("forged session %q has %d rows — the guest moved its traffic", forged, len(v.Events)) + } + } + if v := store.View("sid-a"); v == nil || len(v.Events) == 0 { + t.Error("sid-a has no rows; the decrypted request was filed elsewhere") + } + + // The origin must not see the guest's identity assertions: the wrapper + // strips them on the decrypted request too, not only on the CONNECT. + for _, name := range []string{ + session.ClaudeCodeSessionHeader, session.SessionIDHeader, + "X-Moca-Session", "X-Moca-Vm-Id", "Proxy-Authorization", + } { + if got := originSaw.Get(name); got != "" { + t.Errorf("origin saw guest header %s: %q", name, got) + } + } + + // And the audit line names the VM that owns the socket. + // + // POLLED, not read once: emit runs in the server's handler goroutine after + // inner.ServeHTTP returns, which is not ordered against the client having + // its response — so reading the sink straight after resp.Body.Close() is a + // race the client usually wins on a fast machine and loses on a loaded CI + // runner (observed as "no audit event for vm-a" in CI while passing 20/20 + // locally). Same deadline loop as the tunnel-audit test above, for the same + // reason (#1569 review). + deadline := time.Now().Add(5 * time.Second) + for time.Now().Before(deadline) { + if len(sink.forVM("vm-a")) > 0 { + break + } + time.Sleep(10 * time.Millisecond) + } + if events := sink.forVM("vm-a"); len(events) == 0 { + t.Error("no audit event for vm-a") + } + for _, e := range sink.forVM("vm-b") { + t.Errorf("audit event attributed to vm-b: %+v", e) + } +} diff --git a/core/listener/vmcontrol/control.go b/core/listener/vmcontrol/control.go new file mode 100644 index 000000000..89a19d215 --- /dev/null +++ b/core/listener/vmcontrol/control.go @@ -0,0 +1,57 @@ +// Package vmcontrol carries the per-VM listening socket from MOCA's vmpool to +// the Cortex forward proxy (#1477). +// +// SBX microVMs have no network: each VM's egress lands on its own host Unix +// socket, bound by vmpool (the only process that knows the jail layout) and +// handed to the proxy at run time over a Unix control socket as SCM_RIGHTS. +// The proxy accepts guest connections on that fd without ever touching the +// jail path, and vmpool never sees a credential. The descriptor is the +// identity: which listener accepted a connection says which VM is calling, +// and nothing inside the request can move it. +// +// Wire format: a 4-byte big-endian length followed by a marshalled +// ControlRequest; the server answers each request with exactly one framed +// ControlReply, in order. A Register's fd rides on the sendmsg that carries +// the frame's first byte. +// +// The transport is Unix-only (SCM_RIGHTS, peer credentials); on other +// platforms the package holds only the shared types so that importers still +// compile, and the control server itself does not build. +// +// Ported from rossoctl/moca remote-worker/internal/egress/control @ c13d556 +// (v0.9.0), which is internal to that repo and cannot be imported here. +package vmcontrol + +import ( + "fmt" + "net" +) + +// maxFrame bounds a frame body. Control messages are tiny; a larger length +// means a confused or hostile peer, and the stream cannot be resynchronised +// after it. +const maxFrame = 64 << 10 + +// Refusal codes carried in ControlReply.Code. +const ( + CodeBadRequest = "bad_request" + CodeDuplicateVM = "duplicate_vm" + CodeHandlerError = "handler_error" +) + +// Handler is the proxy's side of the channel. +type Handler interface { + // Register binds vmID's listener. The handler owns ln only when it returns + // nil; on error the server closes it. + Register(vmID, sessionID string, grant []byte, ln net.Listener) error + // Unregister drops vmID's binding. An unknown vmID is not an error. + Unregister(vmID string) error +} + +// RefusalError is a ControlReply{ok:false}. A Handler may return one to choose +// the code; any other handler error is reported as CodeHandlerError. +type RefusalError struct{ Code, Message string } + +func (e *RefusalError) Error() string { + return fmt.Sprintf("vm control refused: %s: %s", e.Code, e.Message) +} diff --git a/core/listener/vmcontrol/handler.go b/core/listener/vmcontrol/handler.go new file mode 100644 index 000000000..fb7676aeb --- /dev/null +++ b/core/listener/vmcontrol/handler.go @@ -0,0 +1,222 @@ +package vmcontrol + +import ( + "bufio" + "context" + "errors" + "net" + "net/http" + "strings" + "sync" + "time" + + "github.com/rossoctl/cortex/core/listener/forwardproxy" + "github.com/rossoctl/cortex/core/session" +) + +// vmBinding is one registered VM: its listener, the http.Server accepting on +// it, and the state Unregister needs to abort it. +type vmBinding struct { + vmID string + sid string + inner http.Handler // the forward proxy's own handler + sink AuditSink + ln *trackedListener + srv *http.Server + + // mu guards aborted and the live-set join below it. + mu sync.Mutex + aborted bool + live sync.WaitGroup // in-flight request wrappers +} + +func newVMBinding(fp *forwardproxy.Server, sink AuditSink, vmID, sid string, ln net.Listener) *vmBinding { + b := &vmBinding{ + vmID: vmID, sid: sid, + inner: fp.Handler(), sink: sink, + ln: newTrackedListener(ln), + } + // ReadHeaderTimeout matches bootstrap's, which every other http.Server in + // the binary gets. Without it a guest can open connections and drip headers + // forever, each one holding a goroutine and an fd until Unregister — and the + // fds are the scarcer half, since one proxy holds a listener per VM plus + // their tunnels. Per-VM caps are #1485's; this is the one-line floor (also + // gosec G112) and it costs the conformance cases nothing (#1569 review). + b.srv = &http.Server{Handler: b, ReadHeaderTimeout: 10 * time.Second} + return b +} + +// requestBinding is what a VM listener stamps into the request context. +type requestBinding struct{ vmID, sid string } + +// bindingCtxKey is a NAMED type, like the forward proxy's connProcKey and the +// transparent listener's origDstKey. An untyped `var bindingCtxKey struct{}` +// would be a zero-size value whose type is plain struct{}, so every package +// doing the same thing shares one context key: a foreign reader would see this +// VM's binding, and a foreign writer would overwrite it with a value this +// package would then type-assert. Naming the type makes the key unique to +// vmcontrol (and satisfies SA1029) (#1569 review). +type bindingCtxKey struct{} + +// SessionIDFromRequest returns the session id the VM listener resolved for r +// — from which handed-over socket its connection arrived — or "" when r did +// not come through one. It is the forward proxy's TrustedSessionID hook: the +// answer is the listener's own, so nothing the client sent participates. +func SessionIDFromRequest(r *http.Request) string { + if b, ok := r.Context().Value(bindingCtxKey{}).(requestBinding); ok { + return b.sid + } + return "" +} + +// AdoptBridgedRequest is the forward proxy's AdoptBridgedRequest hook: it +// carries a VM's binding from the CONNECT onto each request that tunnel +// decrypts, and strips the guest's identity headers from them. +// +// It is needed because a bridged request never passes through vmBinding. +// ServeHTTP: tlsbridge.ServeConn parses the decrypted bytes into fresh +// requests inside its own http.Server, so the context the wrapper stamped on +// the CONNECT is not theirs and the headers it stripped were a different +// header set. Without this, SessionIDFromRequest answers "" on a decrypted +// request, hydrateSession falls through to header resolution, and a guest +// sending X-Claude-Code-Session-Id files its traffic in any session it likes +// — the trusted-session seam inverted, and the guest headers reach the origin +// too. Pinned by TestBridgedRequestKeepsTheVMBinding (#1569 review). +// +// Returns r unchanged when the CONNECT carried no binding, so a bridged +// tunnel on a non-VM listener is untouched. +func AdoptBridgedRequest(connect, r *http.Request) *http.Request { + b, ok := connect.Context().Value(bindingCtxKey{}).(requestBinding) + if !ok { + return r + } + stripGuestHeaders(r) + return r.WithContext(context.WithValue(r.Context(), bindingCtxKey{}, b)) +} + +// guestStrippedHeaders are the client-asserted identity headers a VM guest +// cannot be allowed to speak. Session headers would otherwise be honored by +// the proxy's own header bucketing; Proxy-Authorization reaches nothing that +// authenticates a VM (the forward proxy deletes it too, but the guest's copy +// must not survive even to the pipeline). X-Moca-* is stripped by prefix below. +var guestStrippedHeaders = []string{ + "Proxy-Authorization", + session.ClaudeCodeSessionHeader, + session.BobSessionHeader, + session.SessionIDHeader, + session.CodexSessionHeader, +} + +// stripGuestHeaders removes everything from r that could move the request to +// another session or assert an identity the connection did not establish. +func stripGuestHeaders(r *http.Request) { + for _, name := range guestStrippedHeaders { + r.Header.Del(name) + } + for name := range r.Header { + if strings.HasPrefix(strings.ToLower(name), "x-moca-") { + r.Header.Del(name) + } + } +} + +// ServeHTTP is the per-VM wrapper around the forward proxy's handler: it +// strips guest-asserted identity headers, carries the binding in the request +// context (the trusted-session seam), and emits exactly one terminal audit +// line per request when the inner handler returns. +func (b *vmBinding) ServeHTTP(w http.ResponseWriter, r *http.Request) { + stripGuestHeaders(r) + + // Join the live set under mu, or refuse: an Add racing Unregister's Wait + // could be missed and hang nothing but its own accounting. Past this gate, + // Unregister's closeAll is what wakes the inner handler (its connections + // are closed under it), and live.Wait does not return until the aborted + // line below has been emitted. + b.mu.Lock() + if b.aborted { + b.mu.Unlock() + // The binding is gone and no request ran — but a bare return is not how + // to say so. net/http would send an empty 200 OK on a connection that is + // still open in the window between close() setting aborted and closeAll() + // closing the conns, so the guest reads "success" for a request that + // never left, and the one-terminal-line-per-request promise has a hole + // exactly where the audit matters. 503 with Connection: close says it to + // the guest, and the line says it to the operator (#1569 review). + w.Header().Set("Connection", "close") + http.Error(w, "egress binding closed", http.StatusServiceUnavailable) + b.emit(r, http.StatusServiceUnavailable) + return + } + b.live.Add(1) + b.mu.Unlock() + defer b.live.Done() + + r = r.WithContext(context.WithValue(r.Context(), bindingCtxKey{}, requestBinding{b.vmID, b.sid})) + sw := newStatusWriter(w) + b.inner.ServeHTTP(sw, r) + b.emit(r, sw.status) +} + +// emit writes the request's one terminal audit line. A CONNECT that was +// established reports 200: the proxy writes that status itself on the raw +// connection after hijacking, where statusWriter cannot see it. +func (b *vmBinding) emit(r *http.Request, status int) { + e := Event{ + VMID: b.vmID, SID: b.sid, + Host: r.Host, Method: r.Method, + Status: status, + } + b.mu.Lock() + aborted := b.aborted + b.mu.Unlock() + switch { + case aborted: + e.Decision = DecisionAborted + case r.Method == http.MethodConnect: + e.Decision = DecisionTunnel + default: + e.Decision = DecisionForward + e.Path = RedactPath(r.URL) + } + b.sink.Emit(e) +} + +func (b *vmBinding) close() { + b.mu.Lock() + b.aborted = true + b.mu.Unlock() + b.ln.closeAll() // wakes every in-flight handler, hijacked tunnels included + b.live.Wait() // the aborted lines are emitted before Unregister returns + _ = b.srv.Close() +} + +// statusWriter captures the status the client saw. It forwards Hijack — an +// opaque CONNECT tunnel is the one request shape whose bytes never pass +// through it — and Flush, so streaming responses keep streaming. +type statusWriter struct { + http.ResponseWriter + status int +} + +func newStatusWriter(w http.ResponseWriter) *statusWriter { + return &statusWriter{ResponseWriter: w, status: http.StatusOK} +} + +func (w *statusWriter) WriteHeader(code int) { + w.status = code + w.ResponseWriter.WriteHeader(code) +} + +func (w *statusWriter) Hijack() (net.Conn, *bufio.ReadWriter, error) { + h, ok := w.ResponseWriter.(http.Hijacker) + if !ok { + return nil, nil, errors.New("vmcontrol: inner ResponseWriter does not support Hijack") + } + return h.Hijack() +} + +func (w *statusWriter) Flush() { + if f, ok := w.ResponseWriter.(http.Flusher); ok { + f.Flush() + } +} diff --git a/core/listener/vmcontrol/peercred_darwin.go b/core/listener/vmcontrol/peercred_darwin.go new file mode 100644 index 000000000..dd35a3268 --- /dev/null +++ b/core/listener/vmcontrol/peercred_darwin.go @@ -0,0 +1,34 @@ +//go:build darwin + +package vmcontrol + +import ( + "fmt" + "net" + + "golang.org/x/sys/unix" +) + +// peerUID returns the effective UID of the process that connected c +// (LOCAL_PEERCRED). +func peerUID(c *net.UnixConn) (int, error) { + raw, err := c.SyscallConn() + if err != nil { + return 0, err + } + var uid int + var cerr error + if err := raw.Control(func(fd uintptr) { + var x *unix.Xucred + x, cerr = unix.GetsockoptXucred(int(fd), unix.SOL_LOCAL, unix.LOCAL_PEERCRED) + if cerr == nil { + uid = int(x.Uid) + } + }); err != nil { + return 0, err + } + if cerr != nil { + return 0, fmt.Errorf("LOCAL_PEERCRED: %w", cerr) + } + return uid, nil +} diff --git a/core/listener/vmcontrol/peercred_linux.go b/core/listener/vmcontrol/peercred_linux.go new file mode 100644 index 000000000..d5dbaae45 --- /dev/null +++ b/core/listener/vmcontrol/peercred_linux.go @@ -0,0 +1,34 @@ +//go:build linux + +package vmcontrol + +import ( + "fmt" + "net" + + "golang.org/x/sys/unix" +) + +// peerUID returns the effective UID of the process that connected c +// (SO_PEERCRED). +func peerUID(c *net.UnixConn) (int, error) { + raw, err := c.SyscallConn() + if err != nil { + return 0, err + } + var uid int + var cerr error + if err := raw.Control(func(fd uintptr) { + var cred *unix.Ucred + cred, cerr = unix.GetsockoptUcred(int(fd), unix.SOL_SOCKET, unix.SO_PEERCRED) + if cerr == nil { + uid = int(cred.Uid) + } + }); err != nil { + return 0, err + } + if cerr != nil { + return 0, fmt.Errorf("SO_PEERCRED: %w", cerr) + } + return uid, nil +} diff --git a/core/listener/vmcontrol/peercred_other.go b/core/listener/vmcontrol/peercred_other.go new file mode 100644 index 000000000..2b462e9e7 --- /dev/null +++ b/core/listener/vmcontrol/peercred_other.go @@ -0,0 +1,14 @@ +//go:build unix && !linux && !darwin + +package vmcontrol + +import ( + "errors" + "net" +) + +// peerUID refuses: without a verified peer credential the control socket would +// admit anyone who can reach it, so unsupported unixes fail closed. +func peerUID(*net.UnixConn) (int, error) { + return 0, errors.New("peer credentials unsupported on this platform") +} diff --git a/core/listener/vmcontrol/registry.go b/core/listener/vmcontrol/registry.go new file mode 100644 index 000000000..a37c1cca7 --- /dev/null +++ b/core/listener/vmcontrol/registry.go @@ -0,0 +1,154 @@ +package vmcontrol + +import ( + "net" + "sync" + + "github.com/rossoctl/cortex/core/listener/forwardproxy" +) + +// Registry is the control channel's Handler: it binds each handed-over +// listener to the forward proxy, one binding per VM, and tears bindings down +// on Unregister. One forwardproxy.Server — one pipeline, one session store — +// serves every VM; which listener accepted the connection is the identity +// (#1477), carried into the pipeline through the request context. +type Registry struct { + fp *forwardproxy.Server + sink AuditSink + + mu sync.Mutex + vms map[string]*vmBinding +} + +// NewRegistry returns a Registry serving every registered VM's traffic +// through fp, emitting audit events to sink. +func NewRegistry(fp *forwardproxy.Server, sink AuditSink) *Registry { + return &Registry{fp: fp, sink: sink, vms: map[string]*vmBinding{}} +} + +// Register binds vmID's listener. The Registry owns ln from here on. +// +// A non-empty grant is refused: this slice has no grants (tenancy is decided +// once at boot, by MOCA_TENANCY in cmd/cortex), and refusing here rather than +// ignoring the grant keeps a vmpool that thinks it sent one from running a VM +// with weaker isolation than it configured. +func (r *Registry) Register(vmID, sessionID string, grant []byte, ln net.Listener) error { + if len(grant) != 0 { + return &RefusalError{Code: CodeBadRequest, Message: "grants are not supported in this slice (MI1 S2)"} + } + b := newVMBinding(r.fp, r.sink, vmID, sessionID, ln) + r.mu.Lock() + old := r.vms[vmID] + r.vms[vmID] = b + r.mu.Unlock() + // The control server already refuses a duplicate on the same connection + // and supersedes across connections; this covers a Register that reaches + // the Registry by any other path. + if old != nil { + old.close() + } + go func() { _ = b.srv.Serve(b.ln) }() // returns ErrServerClosed in close + return nil +} + +// Unregister drops vmID's binding. An unknown vmID is not an error. +// +// By the time it returns, the listener refuses new dials, every live +// connection — hijacked tunnels included, which http.Server.Close would leave +// open — has been closed, and each in-flight request's wrapper has emitted its +// aborted audit line. +func (r *Registry) Unregister(vmID string) error { + r.mu.Lock() + b := r.vms[vmID] + delete(r.vms, vmID) + r.mu.Unlock() + if b != nil { + b.close() + } + return nil +} + +// Close unregisters everything (process shutdown). +func (r *Registry) Close() { + r.mu.Lock() + vms := r.vms + r.vms = map[string]*vmBinding{} + r.mu.Unlock() + for _, b := range vms { + b.close() + } +} + +// trackedListener records every accepted connection so Unregister can close +// them all. http.Server.Close closes only the conns it still owns; a hijacked +// CONNECT tunnel is the handler's, and an aborted VM's tunnel must not outlive +// its Unregister. +type trackedListener struct { + inner net.Listener + + mu sync.Mutex + conns map[*trackedConn]struct{} // nil once closed +} + +func newTrackedListener(inner net.Listener) *trackedListener { + return &trackedListener{inner: inner, conns: map[*trackedConn]struct{}{}} +} + +func (l *trackedListener) Accept() (net.Conn, error) { + c, err := l.inner.Accept() + if err != nil { + return nil, err + } + l.mu.Lock() + if l.conns == nil { + l.mu.Unlock() + _ = c.Close() + return nil, net.ErrClosed + } + tc := &trackedConn{Conn: c, ln: l} + l.conns[tc] = struct{}{} + l.mu.Unlock() + return tc, nil +} + +// Close is the net.Listener method: it stops accepting but leaves live +// connections alone, matching http.Server's use of it. Unregister goes through +// closeAll instead. +func (l *trackedListener) Close() error { + return l.inner.Close() +} + +func (l *trackedListener) Addr() net.Addr { return l.inner.Addr() } + +// closeAll closes the listener and every tracked connection. +func (l *trackedListener) closeAll() { + _ = l.inner.Close() + l.mu.Lock() + conns := l.conns + l.conns = nil + l.mu.Unlock() + for c := range conns { + _ = c.Close() + } +} + +func (l *trackedListener) remove(c *trackedConn) { + l.mu.Lock() + if l.conns != nil { + delete(l.conns, c) + } + l.mu.Unlock() +} + +// trackedConn is an accepted connection that removes itself from its listener +// on Close. +type trackedConn struct { + net.Conn + ln *trackedListener + once sync.Once +} + +func (c *trackedConn) Close() error { + c.once.Do(func() { c.ln.remove(c) }) + return c.Conn.Close() +} diff --git a/core/listener/vmcontrol/registry_test.go b/core/listener/vmcontrol/registry_test.go new file mode 100644 index 000000000..bc672c849 --- /dev/null +++ b/core/listener/vmcontrol/registry_test.go @@ -0,0 +1,790 @@ +package vmcontrol + +import ( + "bufio" + "context" + "encoding/json" + "fmt" + "io" + "net" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + "time" + + "github.com/rossoctl/cortex/core/listener/egressallow" + "github.com/rossoctl/cortex/core/listener/forwardproxy" + "github.com/rossoctl/cortex/core/pipeline" + "github.com/rossoctl/cortex/core/session" +) + +// These are the in-tree stand-ins for moca's egress conformance suite +// (remote-worker/egress/conformance), which cannot be imported while moca's +// modules are unpublished. The three done-when cases live here: +// identity_from_connection, unregister_aborts_inflight, grant_refused_under_single. +// +// They drive the Registry directly, with TCP listeners standing in for the +// handed-over Unix ones: everything under test — identity from the +// connection, abort semantics, audit shape — is listener-type-agnostic, and +// TCP keeps the tests runnable where Unix binds are unavailable. The +// SCM_RIGHTS handover itself is covered by server_test.go. + +// sessionProbe records the session identity the pipeline handed a plugin — +// the side every session-aware plugin keys on, as distinct from the store. +type sessionProbe struct { + mu sync.Mutex + ids []string + sources []pipeline.SessionSource + headers []string +} + +func (p *sessionProbe) Name() string { return "session-probe" } +func (p *sessionProbe) Capabilities() pipeline.PluginCapabilities { + return pipeline.PluginCapabilities{} +} +func (p *sessionProbe) OnRequest(_ context.Context, pctx *pipeline.Context) pipeline.Action { + p.mu.Lock() + defer p.mu.Unlock() + if pctx.Session == nil { + p.ids = append(p.ids, "") + p.sources = append(p.sources, "") + p.headers = append(p.headers, "") + } else { + p.ids = append(p.ids, pctx.Session.ID) + p.sources = append(p.sources, pctx.SessionSource) + p.headers = append(p.headers, pctx.SessionHeader) + } + return pipeline.Action{Type: pipeline.Continue} +} +func (p *sessionProbe) OnResponse(_ context.Context, _ *pipeline.Context) pipeline.Action { + return pipeline.Action{Type: pipeline.Continue} +} + +// captureSink collects audit events for assertions. +type captureSink struct { + mu sync.Mutex + events []Event +} + +func (s *captureSink) Emit(e Event) { + s.mu.Lock() + defer s.mu.Unlock() + s.events = append(s.events, e) +} + +func (s *captureSink) forVM(vm string) []Event { + s.mu.Lock() + defer s.mu.Unlock() + var out []Event + for _, e := range s.events { + if e.VMID == vm { + out = append(out, e) + } + } + return out +} + +// newVMEgressStack wires the real handler stack a VM's traffic crosses: a +// forward proxy with header bucketing on (so a forged header would otherwise +// win) and the trusted-session hook set the way cmd/cortex sets it. +func newVMEgressStack(t *testing.T) (*Registry, *captureSink, *session.Store, *sessionProbe) { + t.Helper() + store := session.New(5*time.Minute, 100, 0) + t.Cleanup(store.Close) + probe := &sessionProbe{} + p, err := pipeline.New([]pipeline.Plugin{probe}) + if err != nil { + t.Fatal(err) + } + fp := &forwardproxy.Server{ + OutboundPipeline: pipeline.NewHolder(p), + Sessions: store, + Client: http.DefaultClient, + // Forged-header coverage needs the headers to matter when not trusted. + SessionIDHeaders: []string{session.ClaudeCodeSessionHeader, session.SessionIDHeader}, + TrustedSessionID: SessionIDFromRequest, + } + sink := &captureSink{} + return NewRegistry(fp, sink), sink, store, probe +} + +// registerTCPVM binds a listener and hands it to the registry as a VM's +// egress socket, returning the address its guests dial. +func registerTCPVM(t *testing.T, reg *Registry, vm, sid string) string { + t.Helper() + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = ln.Close() }) + if err := reg.Register(vm, sid, nil, ln); err != nil { + t.Fatal(err) + } + return ln.Addr().String() +} + +// guest dials a VM's socket the way a microVM's egress does. +func guest(t *testing.T, addr string) net.Conn { + t.Helper() + c, err := net.Dial("tcp", addr) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = c.Close() }) + return c +} + +// guestRequest writes one absolute-form GET through a VM socket without +// reading the response. +func guestRequest(conn net.Conn, target string) { + host := target + if i := strings.Index(host, "://"); i >= 0 { + host = host[i+3:] + } + if i := strings.Index(host, "/"); i >= 0 { + host = host[:i] + } + _, _ = fmt.Fprintf(conn, "GET %s HTTP/1.1\r\nHost: %s\r\n\r\n", target, host) +} + +// guestGet sends one absolute-form GET through a VM socket and returns the +// response, leaving the connection open for the caller to close. +func guestGet(t *testing.T, conn net.Conn, target string, headers map[string]string) *http.Response { + t.Helper() + host := target + if i := strings.Index(host, "://"); i >= 0 { + host = host[i+3:] + } + if i := strings.Index(host, "/"); i >= 0 { + host = host[:i] + } + var b strings.Builder + fmt.Fprintf(&b, "GET %s HTTP/1.1\r\nHost: %s\r\n", target, host) + for k, v := range headers { + fmt.Fprintf(&b, "%s: %s\r\n", k, v) + } + b.WriteString("\r\n") + if _, err := conn.Write([]byte(b.String())); err != nil { + t.Fatal(err) + } + resp, err := http.ReadResponse(bufio.NewReader(conn), nil) + if err != nil { + t.Fatal(err) + } + return resp +} + +// guestConnect opens an opaque CONNECT tunnel through a VM socket and returns +// the established connection plus its response. +func guestConnect(t *testing.T, conn net.Conn, target string) *http.Response { + t.Helper() + req, _ := http.NewRequest(http.MethodConnect, "http://"+target, nil) + if err := req.Write(conn); err != nil { + t.Fatal(err) + } + resp, err := http.ReadResponse(bufio.NewReader(conn), req) + if err != nil { + t.Fatal(err) + } + return resp +} + +// echoTCPOrigin is a TCP server echoing every byte, the far end of an opaque +// tunnel. +func echoTCPOrigin(t *testing.T) string { + t.Helper() + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(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().String() +} + +// identity_from_connection: two VMs, two sids; every audit event and every +// recorded row names the registered vm_id/sid and no other, and nothing the +// guest sent — session headers, X-Moca-*, Proxy-Authorization — moves a +// request to another session or reaches the upstream. +func TestRegistryIdentityFromConnection(t *testing.T) { + reg, sink, store, probe := newVMEgressStack(t) + vmA := registerTCPVM(t, reg, "vm-a", "sid-a") + vmB := registerTCPVM(t, reg, "vm-b", "sid-b") + + var backendMu sync.Mutex + var backendSaw http.Header + backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + backendMu.Lock() + backendSaw = r.Header.Clone() + backendMu.Unlock() + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte("ok")) + })) + t.Cleanup(backend.Close) + target := backend.URL + "/from-vm" + + for _, vm := range []struct{ addr, sid, wantVM string }{ + {vmA, "sid-a", "vm-a"}, + {vmB, "sid-b", "vm-b"}, + } { + conn := guest(t, vm.addr) + resp := guestGet(t, conn, target, map[string]string{ + "X-Moca-Session": "sid-b", + "X-Moca-Vm-Id": "vm-b", + "X-Moca-Anything": "x", + "Proxy-Authorization": "Basic Zm9yZ2Vk", + session.ClaudeCodeSessionHeader: "forged-cc", + session.SessionIDHeader: "forged-x", + }) + if resp.StatusCode != http.StatusOK { + t.Fatalf("%s: status %d", vm.wantVM, resp.StatusCode) + } + _ = resp.Body.Close() + _ = conn.Close() + + events := sink.forVM(vm.wantVM) + if len(events) == 0 { + t.Fatalf("%s: no audit event", vm.wantVM) + } + e := events[len(events)-1] + if e.SID != vm.sid || e.Decision != DecisionForward || e.Status != http.StatusOK { + t.Errorf("%s: audit = %+v, want sid %s forward 200", vm.wantVM, e, vm.sid) + } + if e.Method != http.MethodGet || e.Path != "/from-vm" { + t.Errorf("%s: audit = %+v, want GET /from-vm", vm.wantVM, e) + } + } + + // The pipeline heard the registered sid, resolved from the connection. + probe.mu.Lock() + for i, id := range probe.ids { + want := "sid-a" + if i > 0 { + want = "sid-b" + } + if id != want || probe.sources[i] != pipeline.SessionSourceConnection || probe.headers[i] != "" { + t.Errorf("plugin saw session %q via %q/%q, want %q via connection", + id, probe.sources[i], probe.headers[i], want) + } + } + probe.mu.Unlock() + + // Rows are filed under the registered sids, and under nothing the guest named. + for _, sid := range []string{"sid-a", "sid-b"} { + if v := store.View(sid); v == nil || len(v.Events) == 0 { + t.Errorf("session %s has no recorded rows", sid) + } + } + for _, forged := range []string{"forged-cc", "forged-x", "sid-b", "default"} { + if sid := forged; sid != "sid-b" && store.View(sid) != nil { + t.Errorf("forged session %q has rows", forged) + } + } + + // The upstream saw none of the guest's identity assertions. + backendMu.Lock() + for _, name := range []string{ + "X-Moca-Session", "X-Moca-Vm-Id", "X-Moca-Anything", + "Proxy-Authorization", session.ClaudeCodeSessionHeader, session.SessionIDHeader, + } { + if _, ok := backendSaw[http.CanonicalHeaderKey(name)]; ok { + t.Errorf("upstream saw %s", name) + } + } + backendMu.Unlock() +} + +// A CONNECT tunnel through a VM records one terminal tunnel line with the +// CONNECT's own status, and the bytes cross. +func TestRegistryTunnelAudit(t *testing.T) { + reg, sink, _, _ := newVMEgressStack(t) + vmB := registerTCPVM(t, reg, "vm-b", "sid-b") + origin := echoTCPOrigin(t) + + conn := guest(t, vmB) + resp := guestConnect(t, conn, origin) + if resp.StatusCode != http.StatusOK { + t.Fatalf("CONNECT status %d", resp.StatusCode) + } + if _, err := conn.Write([]byte("ping")); err != nil { + t.Fatal(err) + } + buf := make([]byte, 4) + if _, err := io.ReadFull(conn, buf); err != nil { + t.Fatalf("tunnel echo: %v", err) + } + if string(buf) != "ping" { + t.Fatalf("tunnel echoed %q", buf) + } + _ = conn.Close() + + deadline := time.Now().Add(5 * time.Second) + for time.Now().Before(deadline) { + if evs := sink.forVM("vm-b"); len(evs) > 0 { + break + } + time.Sleep(10 * time.Millisecond) + } + events := sink.forVM("vm-b") + if len(events) != 1 { + t.Fatalf("audit events = %+v, want exactly one tunnel line", events) + } + e := events[0] + if e.Decision != DecisionTunnel || e.Method != http.MethodConnect || e.Status != http.StatusOK { + t.Errorf("tunnel audit = %+v, want CONNECT tunnel 200", e) + } + if e.Path != "" { + t.Errorf("tunnel audit path = %q, want none (opaque bytes carry no request line)", e.Path) + } +} + +// unregister_aborts_inflight: Unregister closes the VM's listener and every +// live connection — an opaque tunnel and a slow in-flight GET, hijacked +// tunnels included — before returning; each live request emits exactly one +// aborted line, no tunnel/forward line is emitted for the VM, and the other +// VM's tunnel is untouched. +func TestRegistryUnregisterAbortsInflight(t *testing.T) { + reg, sink, _, _ := newVMEgressStack(t) + vmA := registerTCPVM(t, reg, "vm-a", "sid-a") + vmB := registerTCPVM(t, reg, "vm-b", "sid-b") + origin := echoTCPOrigin(t) + + // A live tunnel on vm-a, established and proven. + tunnelA := guest(t, vmA) + if resp := guestConnect(t, tunnelA, origin); resp.StatusCode != http.StatusOK { + t.Fatalf("vm-a CONNECT status %d", resp.StatusCode) + } + if _, err := tunnelA.Write([]byte("ping")); err != nil { + t.Fatal(err) + } + ping := make([]byte, 4) + if _, err := io.ReadFull(tunnelA, ping); err != nil { + t.Fatalf("vm-a tunnel echo: %v", err) + } + + // A slow in-flight GET on vm-a: the backend holds the response until the + // test releases it, so the request is provably inside the proxy when + // Unregister runs. + arrived := make(chan struct{}, 1) + release := make(chan struct{}) + backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + arrived <- struct{}{} + <-release + w.WriteHeader(http.StatusOK) + })) + t.Cleanup(backend.Close) + slowA := guest(t, vmA) + go func() { + // The response is expected to die with the connection when vm-a is + // unregistered; only the request needs to get in, so no failure is + // reported from in here. + guestRequest(slowA, backend.URL+"/slow") + resp, err := http.ReadResponse(bufio.NewReader(slowA), nil) + if err == nil { + _ = resp.Body.Close() + } + }() + select { + case <-arrived: + case <-time.After(5 * time.Second): + t.Fatal("slow request never reached the backend") + } + + // A live tunnel on vm-b, which must survive vm-a's Unregister. + tunnelB := guest(t, vmB) + if resp := guestConnect(t, tunnelB, origin); resp.StatusCode != http.StatusOK { + t.Fatalf("vm-b CONNECT status %d", resp.StatusCode) + } + + if err := reg.Unregister("vm-a"); err != nil { + t.Fatal(err) + } + close(release) + + // vm-a's connections are closed. + if _, err := tunnelA.Read(make([]byte, 1)); err == nil { + t.Error("vm-a tunnel still readable after Unregister") + } + if _, err := slowA.Read(make([]byte, 1)); err == nil { + t.Error("vm-a slow request still readable after Unregister") + } + // The VM's listener refuses new guests. + if c, err := net.Dial("tcp", vmA); err == nil { + _ = c.Close() + t.Error("vm-a socket still accepts after Unregister") + } + + // Exactly one aborted line per live request, and nothing else, for vm-a. + events := sink.forVM("vm-a") + methods := map[string]bool{} + for _, e := range events { + if e.Decision != DecisionAborted { + t.Errorf("vm-a audit = %+v, want only aborted lines", e) + } + methods[e.Method] = true + } + if len(events) != 2 || !methods[http.MethodConnect] || !methods[http.MethodGet] { + t.Errorf("vm-a audit = %+v, want one aborted CONNECT and one aborted GET", events) + } + + // vm-b's tunnel still carries bytes. + if _, err := tunnelB.Write([]byte("pong")); err != nil { + t.Fatalf("vm-b tunnel write after vm-a Unregister: %v", err) + } + pong := make([]byte, 4) + if _, err := io.ReadFull(tunnelB, pong); err != nil || string(pong) != "pong" { + t.Fatalf("vm-b tunnel echo after vm-a Unregister: %q %v", pong, err) + } +} + +// grant_refused_under_single: a Register carrying a grant is refused with +// bad_request and binds nothing; the same vm_id registers cleanly after. +func TestRegistryGrantRefusedUnderSingle(t *testing.T) { + reg, _, _, _ := newVMEgressStack(t) + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = ln.Close() }) + + err = reg.Register("vm-g", "sid-g", []byte("a-grant"), ln) + if err == nil { + t.Fatal("grant accepted under single tenancy") + } + re, ok := err.(*RefusalError) + if !ok || re.Code != CodeBadRequest { + t.Fatalf("grant refusal = %v, want bad_request", err) + } + if want := "grants are not supported in this slice (MI1 S2)"; re.Message != want { + t.Errorf("grant refusal message = %q, want %q", re.Message, want) + } + // Nothing was bound: Unregister of the refused VM is a no-op, and the + // socket the refused fd served is closed (the server's contract for a + // refused Register; the Registry never adopted it). + if err := reg.Unregister("vm-g"); err != nil { + t.Errorf("unregister of refused vm: %v", err) + } + + // The same vm_id registers cleanly once the grant is gone. + if err := reg.Register("vm-g", "sid-g", nil, ln); err != nil { + t.Fatalf("clean register after refusal: %v", err) + } +} + +// The audit line is moca's JSON shape: same keys, via forced, path without a +// query string. +func TestAuditEventJSONShape(t *testing.T) { + var buf syncBuffer + sink := NewJSONSink(&buf) + sink.Emit(Event{ + VMID: "vm-a", SID: "sid-a", + Host: "example.com", Method: http.MethodGet, + Path: "/x", Status: 200, Decision: DecisionForward, + }) + + var got map[string]any + if err := json.Unmarshal(buf.bytes(), &got); err != nil { + t.Fatalf("not one JSON line: %v (%q)", err, buf.bytes()) + } + for _, key := range []string{"time", "via", "vm_id", "sid", "host", "method", "path", "status", "bytes_up", "bytes_down", "decision"} { + if _, ok := got[key]; !ok { + t.Errorf("audit line missing %q: %v", key, got) + } + } + if got["via"] != "moca-egress" { + t.Errorf("via = %v, want moca-egress", got["via"]) + } + // Fields this slice leaves empty are absent, not zero: a reader must not + // mistake an unset value for a real one. + for _, key := range []string{"sub", "dial_addr", "credential", "dropped", "coalesced"} { + if _, ok := got[key]; ok { + t.Errorf("audit line carries %q, which this slice never sets", key) + } + } +} + +// syncBuffer is a locked byte buffer for the JSON sink test. +type syncBuffer struct { + mu sync.Mutex + b []byte +} + +func (s *syncBuffer) Write(p []byte) (int, error) { + s.mu.Lock() + defer s.mu.Unlock() + s.b = append(s.b, p...) + return len(p), nil +} + +func (s *syncBuffer) bytes() []byte { + s.mu.Lock() + defer s.mu.Unlock() + return append([]byte(nil), s.b...) +} + +// newVMEgressStackWithAllow is newVMEgressStack with a deny-by-default egress +// allow list armed, for the refusal paths that record BEFORE hydration. +func newVMEgressStackWithAllow(t *testing.T, entries ...string) (*Registry, *captureSink, *session.Store) { + t.Helper() + store := session.New(5*time.Minute, 100, 0) + t.Cleanup(store.Close) + p, err := pipeline.New([]pipeline.Plugin{&sessionProbe{}}) + if err != nil { + t.Fatal(err) + } + allow, err := egressallow.New(entries) + if err != nil { + t.Fatal(err) + } + fp := &forwardproxy.Server{ + OutboundPipeline: pipeline.NewHolder(p), + Sessions: store, + Client: http.DefaultClient, + SessionIDHeaders: []string{session.ClaudeCodeSessionHeader, session.SessionIDHeader}, + TrustedSessionID: SessionIDFromRequest, + EgressAllow: allow, + } + sink := &captureSink{} + return NewRegistry(fp, sink), sink, store +} + +// rowsIn is how many session rows a bucket holds. +func rowsIn(store *session.Store, sid string) int { + if v := store.View(sid); v != nil { + return len(v.Events) + } + return 0 +} + +// TestRegistry_EgressDeniedRowLandsInTheVMSession: a VM request refused by the +// egress allow list is refused BEFORE hydration, so nothing on that path used to +// consult TrustedSessionID — the denied row went to ActiveSession() (possibly +// another VM's) or to `default`. +// +// That is the headline guarantee inverted on exactly the event you most want +// attributed correctly: "identity from the connection wins over every header- +// and process-derived id" has to hold for refusals too, not only for requests +// that reach the pipeline. Measured before the fix: sid-a rows=0, the unrelated +// active session rows=2 (#1569 review). +func TestRegistry_EgressDeniedRowLandsInTheVMSession(t *testing.T) { + reg, _, store := newVMEgressStackWithAllow(t, "allowed.example:80") + + // Another session is active, as one is on any busy proxy. + store.Append("someone-else", pipeline.SessionEvent{ + At: time.Now(), Direction: pipeline.Outbound, Phase: pipeline.SessionRequest, + Host: "other.example", StatusCode: 200, + }) + before := rowsIn(store, "someone-else") + + addr := registerTCPVM(t, reg, "vm-a", "sid-a") + c := guest(t, addr) + defer func() { _ = c.Close() }() + fmt.Fprint(c, "GET http://denied.example:80/x HTTP/1.1\r\nHost: denied.example\r\n\r\n") + resp, err := http.ReadResponse(bufio.NewReader(c), nil) + if err != nil { + t.Fatalf("read response: %v", err) + } + defer func() { _ = resp.Body.Close() }() + if resp.StatusCode != http.StatusForbidden { + t.Fatalf("status = %d, want 403", resp.StatusCode) + } + + if got := rowsIn(store, "sid-a"); got == 0 { + t.Error("the VM's session holds no row for its own denied egress; the refusal was filed elsewhere") + } + if got := rowsIn(store, "someone-else"); got != before { + t.Errorf("the unrelated active session gained %d row(s): a VM's denial landed in another session", + got-before) + } + if got := rowsIn(store, session.DefaultSessionID); got != 0 { + t.Errorf("default bucket holds %d row(s); the VM's identity was known", got) + } +} + +// TestRegistry_EgressDeniedCONNECTRowLandsInTheVMSession is the same hole on the +// CONNECT path, which refuses before hydration at its own call site. +func TestRegistry_EgressDeniedCONNECTRowLandsInTheVMSession(t *testing.T) { + reg, _, store := newVMEgressStackWithAllow(t, "allowed.example:443") + store.Append("someone-else", pipeline.SessionEvent{ + At: time.Now(), Direction: pipeline.Outbound, Phase: pipeline.SessionRequest, + Host: "other.example", StatusCode: 200, + }) + before := rowsIn(store, "someone-else") + + addr := registerTCPVM(t, reg, "vm-b", "sid-b") + c := guest(t, addr) + defer func() { _ = c.Close() }() + fmt.Fprint(c, "CONNECT denied.example:443 HTTP/1.1\r\nHost: denied.example:443\r\n\r\n") + line, err := bufio.NewReader(c).ReadString('\n') + if err != nil { + t.Fatalf("read CONNECT response: %v", err) + } + if !strings.Contains(line, "403") { + t.Fatalf("CONNECT = %q, want 403", line) + } + + if got := rowsIn(store, "sid-b"); got == 0 { + t.Error("the VM's session holds no row for its own denied CONNECT") + } + if got := rowsIn(store, "someone-else"); got != before { + t.Errorf("the unrelated active session gained %d row(s)", got-before) + } +} + +// TestRegistry_OpaqueTunnelRowsLandInTheVMSession: an opaque CONNECT's own rows +// (open and close) are filed through appendTunnelOpen, which read +// pctx.OutboundSessionID only under client affinity or process attribution. +// With both off — newVMEgressStack sets neither — a VM's allowed tunnel fell +// back to ActiveSession() at recording time, so its rows landed in whatever +// session was last active, possibly another VM's. Identity from the connection +// must not depend on session.client_affinity to reach a tunnel's own rows +// (#1569 review). +func TestRegistry_OpaqueTunnelRowsLandInTheVMSession(t *testing.T) { + reg, _, store, _ := newVMEgressStack(t) + + // Another session is active, as one is on any busy proxy. + store.Append("someone-else", pipeline.SessionEvent{ + At: time.Now(), Direction: pipeline.Outbound, Phase: pipeline.SessionRequest, + Host: "other.example", StatusCode: 200, + }) + before := rowsIn(store, "someone-else") + + addr := registerTCPVM(t, reg, "vm-b", "sid-b") + origin := echoTCPOrigin(t) + c := guest(t, addr) + resp := guestConnect(t, c, origin) + if resp.StatusCode != http.StatusOK { + t.Fatalf("CONNECT status %d", resp.StatusCode) + } + if _, err := c.Write([]byte("ping")); err != nil { + t.Fatal(err) + } + buf := make([]byte, 4) + if _, err := io.ReadFull(c, buf); err != nil { + t.Fatalf("tunnel echo: %v", err) + } + _ = c.Close() + + // The close row is recorded after the tunnel ends; poll for both rows. + deadline := time.Now().Add(5 * time.Second) + for time.Now().Before(deadline) { + if rowsIn(store, "sid-b") >= 2 { + break + } + time.Sleep(10 * time.Millisecond) + } + if got := rowsIn(store, "sid-b"); got != 2 { + t.Errorf("the VM's session holds %d tunnel row(s), want the open and the close", got) + } + if got := rowsIn(store, "someone-else"); got != before { + t.Errorf("the unrelated active session gained %d row(s): a VM's tunnel rows landed in another session", + got-before) + } +} + +// TestRegistry_TrustedSessionSurvivesSessionsDisabled: with session.enabled +// false the proxy has no store, and the guard that skipped hydration took the +// VM's identity with it — a registered VM's request reached the pipeline with +// pctx.Session nil even though sessionViewFor builds an ID-only view without a +// store. A MOCA deployment that turned the store off would quietly lose every +// VM's identity, which is the one thing the control channel exists to carry +// (#1569 review). +func TestRegistry_TrustedSessionSurvivesSessionsDisabled(t *testing.T) { + probe := &sessionProbe{} + p, err := pipeline.New([]pipeline.Plugin{probe}) + if err != nil { + t.Fatal(err) + } + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusOK) + })) + defer upstream.Close() + + fp := &forwardproxy.Server{ + OutboundPipeline: pipeline.NewHolder(p), + Sessions: nil, // session.enabled: false + Client: http.DefaultClient, + TrustedSessionID: SessionIDFromRequest, + } + reg := NewRegistry(fp, &captureSink{}) + addr := registerTCPVM(t, reg, "vm-c", "sid-c") + + c := guest(t, addr) + defer func() { _ = c.Close() }() + fmt.Fprintf(c, "GET %s/x HTTP/1.1\r\nHost: %s\r\n\r\n", + upstream.URL, strings.TrimPrefix(upstream.URL, "http://")) + resp, err := http.ReadResponse(bufio.NewReader(c), nil) + if err != nil { + t.Fatalf("read response: %v", err) + } + defer func() { _ = resp.Body.Close() }() + + probe.mu.Lock() + ids := append([]string(nil), probe.ids...) + probe.mu.Unlock() + if len(ids) == 0 { + t.Fatal("the pipeline never ran for the VM's request") + } + if ids[0] != "sid-c" { + t.Errorf("plugin saw session %q, want sid-c — the trusted id costs no store and must survive session.enabled: false", ids[0]) + } +} + +// TestVMBinding_AbortedWindowAnswers503: between close() setting aborted and +// closeAll() closing the conns, a request that reaches ServeHTTP used to return +// without writing anything — so net/http sent an empty 200 OK on a still-open +// connection and no audit line was emitted. The guest read "success" for a +// request that never left, and the one-terminal-line-per-request promise had a +// hole exactly where the audit matters (#1569 review). +func TestVMBinding_AbortedWindowAnswers503(t *testing.T) { + p, err := pipeline.New([]pipeline.Plugin{&sessionProbe{}}) + if err != nil { + t.Fatal(err) + } + fp := &forwardproxy.Server{ + OutboundPipeline: pipeline.NewHolder(p), + Client: http.DefaultClient, + TrustedSessionID: SessionIDFromRequest, + } + sink := &captureSink{} + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = ln.Close() }) + b := newVMBinding(fp, sink, "vm-d", "sid-d", ln) + + // Enter the window: aborted set, connections not yet closed. + b.close() + + rec := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodGet, "http://denied.example/x", nil) + b.ServeHTTP(rec, req) + + if rec.Code != http.StatusServiceUnavailable { + t.Errorf("status = %d, want 503 — an empty 200 tells the guest its request succeeded", rec.Code) + } + if got := rec.Header().Get("Connection"); got != "close" { + t.Errorf("Connection = %q, want close", got) + } + evs := sink.forVM("vm-d") + if len(evs) != 1 { + t.Fatalf("audit lines = %d, want exactly 1 — one terminal line per request", len(evs)) + } + if evs[0].Decision != DecisionAborted { + t.Errorf("decision = %q, want %q", evs[0].Decision, DecisionAborted) + } +} diff --git a/core/listener/vmcontrol/server.go b/core/listener/vmcontrol/server.go new file mode 100644 index 000000000..5fc7703e2 --- /dev/null +++ b/core/listener/vmcontrol/server.go @@ -0,0 +1,337 @@ +//go:build unix + +package vmcontrol + +import ( + "errors" + "fmt" + "log/slog" + "net" + "os" + "slices" + "sync" + "syscall" + "time" + + "golang.org/x/sys/unix" +) + +const replyTimeout = 10 * time.Second + +// Server accepts control connections on a Unix socket and drives a Handler. +type Server struct { + ln *net.UnixListener + allowed []int + h Handler + + // hmu serialises every bindings change together with the Handler call that + // matches it. Without it, dropConn could clear a VM's binding, release mu, + // and call Unregister after a redialled client had already re-registered + // that VM: the handler would then refuse the replay as a duplicate, or tear + // down the binding the new connection had just made. Lock order is hmu then + // mu; Handler calls happen under hmu, never under mu. + hmu sync.Mutex + + mu sync.Mutex + closed bool + conns map[*serverConn]struct{} + bindings map[string]*serverConn // vmID -> owning connection + wg sync.WaitGroup +} + +type serverConn struct { + c *net.UnixConn + vms map[string]struct{} +} + +// NewServer listens on path (replacing a stale socket left by a crashed +// predecessor) and restricts the socket to its owner. Peers must additionally +// have a UID in allowedUIDs. +func NewServer(path string, allowedUIDs []int, h Handler) (*Server, error) { + if fi, err := os.Lstat(path); err == nil { + if fi.Mode()&os.ModeSocket == 0 { + return nil, fmt.Errorf("vm control: %s exists and is not a socket", path) + } + if err := os.Remove(path); err != nil { + return nil, err + } + } + ln, err := net.ListenUnix("unix", &net.UnixAddr{Name: path, Net: "unix"}) + if err != nil { + return nil, err + } + // Peer credentials are the real gate; the mode keeps unrelated users from + // even connecting. The window before chmod is covered by the parent + // directory's mode. + if err := os.Chmod(path, 0o600); err != nil { + _ = ln.Close() + return nil, err + } + return &Server{ + ln: ln, allowed: slices.Clone(allowedUIDs), h: h, + conns: map[*serverConn]struct{}{}, + bindings: map[string]*serverConn{}, + }, nil +} + +// acceptRetryMin and acceptRetryMax bound the backoff for a TEMPORARY accept +// error, the same 5ms-doubling-to-1s shape net/http.Server.Serve uses. +const ( + acceptRetryMin = 5 * time.Millisecond + acceptRetryMax = 1 * time.Second +) + +// Serve accepts connections until Close, then returns nil. +// +// A TEMPORARY accept error is retried with bounded backoff rather than returned. +// What makes this load-bearing is what the caller does with a returned error: +// startVMControl hands it to fatalf, which exits the process — so an EMFILE or +// ENFILE (plausible on a proxy holding tens of VM listeners plus their tunnels) +// or an ECONNABORTED would kill Cortex and every VM's egress with it, for a +// condition that clears on its own. net/http's own accept loop has retried +// these since forever, for the same reason. Only a permanent error is worth the +// whole flotilla (#1569 review). +func (s *Server) Serve() error { + var delay time.Duration + for { + c, err := s.ln.AcceptUnix() + if err != nil { + s.mu.Lock() + closed := s.closed + s.mu.Unlock() + if closed || errors.Is(err, net.ErrClosed) { + return nil + } + // net.Error.Temporary() is deprecated and lies about some errors, so + // the fd-exhaustion syscall errnos are named explicitly; Temporary() + // is still consulted for anything else that self-describes as + // transient (ECONNABORTED arrives that way). + if isTemporaryAccept(err) { + if delay == 0 { + delay = acceptRetryMin + } else if delay *= 2; delay > acceptRetryMax { + delay = acceptRetryMax + } + slog.Warn("vmcontrol: temporary accept error on the control socket; retrying", + "error", err, "retry_in", delay) + time.Sleep(delay) + continue + } + return err + } + delay = 0 + sc := &serverConn{c: c, vms: map[string]struct{}{}} + s.mu.Lock() + if s.closed { + s.mu.Unlock() + _ = c.Close() + continue + } + s.conns[sc] = struct{}{} + s.wg.Add(1) + s.mu.Unlock() + go s.serveConn(sc) + } +} + +// Close stops accepting, drops every connection and waits until each +// connection's bindings have been handed to Handler.Unregister. +func (s *Server) Close() error { + s.mu.Lock() + s.closed = true + conns := make([]*serverConn, 0, len(s.conns)) + for sc := range s.conns { + conns = append(conns, sc) + } + s.mu.Unlock() + err := s.ln.Close() + for _, sc := range conns { + _ = sc.c.Close() + } + s.wg.Wait() + return err +} + +func (s *Server) serveConn(sc *serverConn) { + defer s.wg.Done() + defer s.dropConn(sc) + + uid, err := peerUID(sc.c) + if err != nil || !slices.Contains(s.allowed, uid) { + return // closed without a reply: do not tell an unauthorised peer anything + } + for { + body, fds, err := readFrame(sc.c) + if err != nil { + return + } + reply := s.handle(sc, body, fds) + _ = sc.c.SetWriteDeadline(time.Now().Add(replyTimeout)) + if err := writeFrame(sc.c, marshalControlReply(reply), -1); err != nil { + return + } + } +} + +func (s *Server) handle(sc *serverConn, body []byte, fds []int) *ControlReply { + req, err := decodeControlRequest(body) + if err != nil { + closeFDs(fds) + return refusal(CodeBadRequest, "malformed request") + } + switch { + case req.Register != nil: + return s.register(sc, req.Register, fds) + case req.Unregister != nil: + closeFDs(fds) + vm := req.Unregister.VmID + s.hmu.Lock() + defer s.hmu.Unlock() + s.release(vm) + if err := s.h.Unregister(vm); err != nil { + return handlerRefusal(err) + } + return &ControlReply{Ok: true} + default: + closeFDs(fds) + return refusal(CodeBadRequest, "empty request") + } +} + +func (s *Server) register(sc *serverConn, r *Register, fds []int) *ControlReply { + if len(fds) == 0 { + return refusal(CodeBadRequest, "register carries no fd") + } + closeFDs(fds[1:]) // only the first fd is meaningful + ln, err := listenerFromFD(fds[0]) + if err != nil { + return refusal(CodeBadRequest, err.Error()) + } + vm := r.VmID + if vm == "" { + _ = ln.Close() + return refusal(CodeBadRequest, "empty vm_id") + } + + s.hmu.Lock() + defer s.hmu.Unlock() + s.mu.Lock() + owner := s.bindings[vm] + if owner == sc { + s.mu.Unlock() + _ = ln.Close() + return refusal(CodeDuplicateVM, "vm already registered on this connection") + } + // A different connection owning vm is a client that redialled before we + // noticed its old connection die; the new registration supersedes the stale + // one. + s.bindings[vm] = sc + sc.vms[vm] = struct{}{} + s.mu.Unlock() + if owner != nil { + s.mu.Lock() + delete(owner.vms, vm) + s.mu.Unlock() + _ = s.h.Unregister(vm) + } + + if err := s.h.Register(vm, r.Sid, r.Grant, ln); err != nil { + _ = ln.Close() + s.release(vm) + return handlerRefusal(err) + } + return &ControlReply{Ok: true} +} + +// release forgets vm's ownership record. +func (s *Server) release(vm string) { + s.mu.Lock() + if o := s.bindings[vm]; o != nil { + delete(o.vms, vm) + delete(s.bindings, vm) + } + s.mu.Unlock() +} + +// dropConn runs when a control connection ends: bindings are owned by the +// connection that registered them, so a vmpool crash or restart cannot leave +// the proxy serving VMs nobody manages. +func (s *Server) dropConn(sc *serverConn) { + _ = sc.c.Close() + s.hmu.Lock() + defer s.hmu.Unlock() + s.mu.Lock() + delete(s.conns, sc) + owned := make([]string, 0, len(sc.vms)) + for vm := range sc.vms { + if s.bindings[vm] == sc { + delete(s.bindings, vm) + owned = append(owned, vm) + } + } + sc.vms = nil + s.mu.Unlock() + for _, vm := range owned { + _ = s.h.Unregister(vm) + } +} + +// listenerFromFD takes ownership of fd, which must be a listening socket. +func listenerFromFD(fd int) (net.Listener, error) { + acc, err := unix.GetsockoptInt(fd, unix.SOL_SOCKET, unix.SO_ACCEPTCONN) + switch { + case errors.Is(err, unix.ENOPROTOOPT): + // darwin does not report SO_ACCEPTCONN for AF_UNIX sockets (Linux, the + // deployment target, does). Probing with accept(2) would steal a queued + // connection, so development builds accept any stream socket here and a + // non-listener fails at the handler's first Accept instead. + if typ, terr := unix.GetsockoptInt(fd, unix.SOL_SOCKET, unix.SO_TYPE); terr != nil || typ != unix.SOCK_STREAM { + _ = unix.Close(fd) + return nil, errors.New("fd is not a stream socket") + } + case err != nil: + _ = unix.Close(fd) + return nil, fmt.Errorf("fd is not a socket: %w", err) + case acc == 0: + _ = unix.Close(fd) + return nil, errors.New("fd is not a listening socket") + } + f := os.NewFile(uintptr(fd), "vm-egress-listener") + defer f.Close() // FileListener dups + ln, err := net.FileListener(f) + if err != nil { + return nil, fmt.Errorf("fd is not a usable listener: %w", err) + } + return ln, nil +} + +func refusal(code, msg string) *ControlReply { + return &ControlReply{Ok: false, Code: code, Message: msg} +} + +func handlerRefusal(err error) *ControlReply { + var re *RefusalError + if errors.As(err, &re) { + return refusal(re.Code, re.Message) + } + return refusal(CodeHandlerError, err.Error()) +} + +// isTemporaryAccept reports whether an accept error is worth retrying. +// +// EMFILE/ENFILE are the ones that matter here — the process or the system is out +// of descriptors, which a proxy holding a listener per VM plus their tunnels can +// reach, and which clears as soon as something closes. ECONNABORTED is the peer +// giving up mid-handshake. Anything else that still implements the deprecated +// Temporary() and says yes is taken at its word; everything else is permanent. +func isTemporaryAccept(err error) bool { + if errors.Is(err, syscall.EMFILE) || errors.Is(err, syscall.ENFILE) || + errors.Is(err, syscall.ECONNABORTED) || errors.Is(err, syscall.EINTR) { + return true + } + var ne net.Error + if errors.As(err, &ne) { + return ne.Temporary() //nolint:staticcheck // SA1019: no replacement for the accept-retry decision; the errnos above carry the cases that matter. + } + return false +} diff --git a/core/listener/vmcontrol/server_test.go b/core/listener/vmcontrol/server_test.go new file mode 100644 index 000000000..8088481ef --- /dev/null +++ b/core/listener/vmcontrol/server_test.go @@ -0,0 +1,364 @@ +//go:build unix + +package vmcontrol + +import ( + "errors" + "fmt" + "net" + "os" + "path/filepath" + "slices" + "sync" + "syscall" + "testing" + "time" +) + +// These tests drive the control server the way vmpool's client does: one +// framed request, one framed reply, the Register fd riding the sendmsg. They +// are the in-tree stand-ins for moca's own control tests (same semantics, +// different repo — the module boundary is why they live here at all). + +// testClient is a minimal control client: synchronous round trips, no +// reconnect logic. Calls are sequential; each sends one frame and reads one. +type testClient struct { + t *testing.T + conn *net.UnixConn +} + +func dialControl(t *testing.T, path string) *testClient { + t.Helper() + c, err := net.Dial("unix", path) + if err != nil { + t.Fatal(err) + } + tc := &testClient{t: t, conn: c.(*net.UnixConn)} + t.Cleanup(func() { _ = tc.conn.Close() }) + return tc +} + +func (tc *testClient) call(req *ControlRequest, fd int) *ControlReply { + tc.t.Helper() + if err := writeFrame(tc.conn, marshalControlRequest(req), fd); err != nil { + tc.t.Fatal(err) + } + body, fds, err := readFrame(tc.conn) + if err != nil { + tc.t.Fatal(err) + } + closeFDs(fds) // replies never carry fds + rep, err := decodeControlReply(body) + if err != nil { + tc.t.Fatal(err) + } + return rep +} + +func (tc *testClient) register(vm, sid string, grant []byte, f *os.File) *ControlReply { + tc.t.Helper() + fd := -1 + if f != nil { + fd = int(f.Fd()) + } + return tc.call(&ControlRequest{Register: &Register{VmID: vm, Sid: sid, Grant: grant}}, fd) +} + +// handover registers vm the way vmpool does: ln.File() dups the fd, the +// SCM_RIGHTS send duplicates it again into the server, and once the reply is +// in the test closes both of its own copies. From then on the handler's +// listener is the only thing keeping the socket alive — which is what the +// refuses-new-dials assertions below depend on, and what moca's harness does. +func (tc *testClient) handover(vm, sid string, ln *net.UnixListener) *ControlReply { + tc.t.Helper() + f, err := ln.File() + if err != nil { + tc.t.Fatal(err) + } + rep := tc.register(vm, sid, nil, f) + _ = f.Close() + _ = ln.Close() + return rep +} + +func (tc *testClient) unregister(vm string) *ControlReply { + tc.t.Helper() + return tc.call(&ControlRequest{Unregister: &Unregister{VmID: vm}}, -1) +} + +// recordingHandler is a Handler that records calls, for the semantics that are +// the server's to enforce. It is called from the server's goroutines, so +// everything it touches is locked, and it owns its listeners the way the real +// handler does — closing them on Unregister — so a socket really does stop +// accepting once its binding is gone. +type recordingHandler struct { + mu sync.Mutex + registered []string + unregistered []string + lns map[string]net.Listener +} + +func newRecordingHandler() *recordingHandler { + return &recordingHandler{lns: map[string]net.Listener{}} +} + +func (h *recordingHandler) Register(vmID, _ string, _ []byte, ln net.Listener) error { + h.mu.Lock() + defer h.mu.Unlock() + h.registered = append(h.registered, vmID) + h.lns[vmID] = ln + return nil +} + +func (h *recordingHandler) Unregister(vmID string) error { + h.mu.Lock() + defer h.mu.Unlock() + h.unregistered = append(h.unregistered, vmID) + if ln := h.lns[vmID]; ln != nil { + _ = ln.Close() + delete(h.lns, vmID) + } + return nil +} + +func (h *recordingHandler) calls() (registered, unregistered []string) { + h.mu.Lock() + defer h.mu.Unlock() + return slices.Clone(h.registered), slices.Clone(h.unregistered) +} + +// requireUnixBind skips the test when this environment refuses to bind Unix +// sockets — the macOS seatbelt sandbox does, and no flag inside the session +// lifts it. CI's Linux runners bind fine. The control channel is Unix-only by +// construction (SCM_RIGHTS, peer credentials), so there is no fallback path to +// test through; the registry's behavior is covered separately over TCP. +func requireUnixBind(t *testing.T) { + t.Helper() + path := filepath.Join(shortDir(t, "bindprobe", len("/probe.sock")), "probe.sock") + ln, err := net.ListenUnix("unix", &net.UnixAddr{Name: path, Net: "unix"}) + if err != nil { + t.Skipf("unix socket bind unavailable in this environment: %v", err) + } + _ = ln.Close() +} + +// newTestServer starts a control server on a fresh socket, admitting this +// process's uid (overridable), and returns it with the socket path. +func newTestServer(t *testing.T, h Handler, allowedUIDs ...int) (*Server, string) { + t.Helper() + requireUnixBind(t) + path := filepath.Join(shortDir(t, "ectl", len("/control.sock")), "control.sock") + if len(allowedUIDs) == 0 { + allowedUIDs = []int{os.Getuid()} + } + srv, err := NewServer(path, allowedUIDs, h) + if err != nil { + t.Fatal(err) + } + go func() { _ = srv.Serve() }() + t.Cleanup(func() { _ = srv.Close() }) + return srv, path +} + +// vmSocket binds a listener the way vmpool does, on a path the test's guests +// can dial. Unlink-on-close is off so the path survives the test closing its +// own copy after a successful handover. +func vmSocket(t *testing.T, name string) (*net.UnixListener, string) { + t.Helper() + requireUnixBind(t) + path := filepath.Join(shortDir(t, "vmsock", 1+len(name)), name) + ln, err := net.ListenUnix("unix", &net.UnixAddr{Name: path, Net: "unix"}) + if err != nil { + t.Fatal(err) + } + ln.SetUnlinkOnClose(false) + return ln, path +} + +func TestServerRegisterUnregister(t *testing.T) { + h := newRecordingHandler() + _, path := newTestServer(t, h) + tc := dialControl(t, path) + + ln, _ := vmSocket(t, "vm-a.sock") + if rep := tc.handover("vm-a", "sid-a", ln); !rep.Ok { + t.Fatalf("register: %s: %s", rep.Code, rep.Message) + } + if rep := tc.unregister("vm-a"); !rep.Ok { + t.Fatalf("unregister: %s: %s", rep.Code, rep.Message) + } + registered, unregistered := h.calls() + if len(registered) != 1 || registered[0] != "vm-a" { + t.Errorf("registered %v, want [vm-a]", registered) + } + if len(unregistered) != 1 || unregistered[0] != "vm-a" { + t.Errorf("unregistered %v, want [vm-a]", unregistered) + } +} + +// An unknown uid is closed without a reply: the peer learns nothing about the +// gate, not even that there is one to fail. +func TestServerUIDGate(t *testing.T) { + h := newRecordingHandler() + _, path := newTestServer(t, h, os.Getuid()+1000) // nobody + tc := dialControl(t, path) + + _ = writeFrame(tc.conn, marshalControlRequest(&ControlRequest{Register: &Register{VmID: "vm-a"}}), -1) + if _, _, err := readFrame(tc.conn); err == nil { + t.Fatal("unauthorized peer got a reply") + } + if registered, _ := h.calls(); len(registered) != 0 { + t.Errorf("handler ran for an unauthorized peer: %v", registered) + } +} + +func TestServerRegisterWithoutFD(t *testing.T) { + h := newRecordingHandler() + _, path := newTestServer(t, h) + tc := dialControl(t, path) + + rep := tc.register("vm-a", "sid-a", nil, nil) + if rep.Ok || rep.Code != CodeBadRequest || rep.Message != "register carries no fd" { + t.Fatalf("got %+v, want bad_request \"register carries no fd\"", rep) + } +} + +func TestServerDuplicateVMOnSameConnection(t *testing.T) { + h := newRecordingHandler() + _, path := newTestServer(t, h) + tc := dialControl(t, path) + + ln, _ := vmSocket(t, "vm-a.sock") + if rep := tc.handover("vm-a", "sid-a", ln); !rep.Ok { + t.Fatalf("first register: %s", rep.Message) + } + ln2, _ := vmSocket(t, "vm-a-2.sock") + rep := tc.handover("vm-a", "sid-a", ln2) + if rep.Ok || rep.Code != CodeDuplicateVM { + t.Fatalf("got %+v, want duplicate_vm", rep) + } +} + +// Bindings are owned by the connection that registered them: when vmpool's +// connection dies, its VMs are unregistered even though nobody sent an +// Unregister — a crashed vmpool cannot leave the proxy serving VMs nobody +// manages. +func TestServerDropConnUnbinds(t *testing.T) { + h := newRecordingHandler() + _, path := newTestServer(t, h) + tc := dialControl(t, path) + + ln, sockPath := vmSocket(t, "vm-a.sock") + if rep := tc.handover("vm-a", "sid-a", ln); !rep.Ok { + t.Fatalf("register: %s", rep.Message) + } + + _ = tc.conn.Close() // vmpool goes away + deadline := time.Now().Add(5 * time.Second) + for { + _, unregistered := h.calls() + if len(unregistered) > 0 || !time.Now().Before(deadline) { + break + } + time.Sleep(10 * time.Millisecond) + } + _, unregistered := h.calls() + if len(unregistered) != 1 || unregistered[0] != "vm-a" { + t.Fatalf("unregistered %v, want [vm-a]", unregistered) + } + // The socket the VM dialed refuses new connections once the binding is gone. + if c, err := net.Dial("unix", sockPath); err == nil { + _ = c.Close() + t.Error("vm socket still accepts after dropConn") + } +} + +// A redialled vmpool re-registering a VM it still owns supersedes the stale +// binding from the old connection instead of being refused as a duplicate. +func TestServerRedialSupersedes(t *testing.T) { + h := newRecordingHandler() + _, path := newTestServer(t, h) + tc1 := dialControl(t, path) + + ln1, sock1 := vmSocket(t, "vm-a.sock") + if rep := tc1.handover("vm-a", "sid-a", ln1); !rep.Ok { + t.Fatalf("first register: %s", rep.Message) + } + + tc2 := dialControl(t, path) + ln2, sock2 := vmSocket(t, "vm-a-new.sock") + if rep := tc2.handover("vm-a", "sid-a", ln2); !rep.Ok { + t.Fatalf("redial register: %s: %s", rep.Code, rep.Message) + } + + // The old listener is unbound (its socket refuses); the new one serves. + if c, err := net.Dial("unix", sock1); err == nil { + _ = c.Close() + t.Error("stale vm socket still accepts after supersede") + } + if c, err := net.Dial("unix", sock2); err != nil { + t.Errorf("new vm socket refuses: %v", err) + } else { + _ = c.Close() + } + if _, unregistered := h.calls(); len(unregistered) != 1 || unregistered[0] != "vm-a" { + t.Errorf("unregistered %v, want exactly the superseded [vm-a]", unregistered) + } +} + +// The malformed-frame test above needs the reply read inline; this keeps its +// assertion honest by checking the reply really was a refusal. +func TestServerMalformedFrameReply(t *testing.T) { + h := newRecordingHandler() + _, path := newTestServer(t, h) + tc := dialControl(t, path) + + if err := writeFrame(tc.conn, []byte{0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff}, -1); err != nil { + t.Fatal(err) + } + body, fds, err := readFrame(tc.conn) + if err != nil { + t.Fatal(err) + } + closeFDs(fds) + rep, err := decodeControlReply(body) + if err != nil { + t.Fatal(err) + } + if rep.Ok || rep.Code != CodeBadRequest || rep.Message != "malformed request" { + t.Fatalf("got %+v, want bad_request \"malformed request\"", rep) + } + // The connection survives a bad frame: the next request still answers. + ln, _ := vmSocket(t, "vm-ok.sock") + if rep := tc.handover("vm-ok", "sid", ln); !rep.Ok { + t.Fatalf("register after malformed frame: %s", rep.Message) + } +} + +// TestIsTemporaryAccept: the accept loop must retry what clears on its own and +// return only what does not. A returned error reaches fatalf, which exits the +// process and takes every VM's egress with it, so an EMFILE on a proxy holding +// a listener per VM must not be fatal (#1569 review). It lives in this +// unix-tagged file because isTemporaryAccept does; an untagged home broke the +// non-unix test build with `undefined: isTemporaryAccept` (#1569 review). +func TestIsTemporaryAccept(t *testing.T) { + for _, tc := range []struct { + name string + err error + want bool + }{ + {"EMFILE: this process is out of fds", syscall.EMFILE, true}, + {"ENFILE: the system is", syscall.ENFILE, true}, + {"ECONNABORTED: peer gave up mid-handshake", syscall.ECONNABORTED, true}, + {"EINTR", syscall.EINTR, true}, + {"wrapped EMFILE", fmt.Errorf("accept: %w", syscall.EMFILE), true}, + {"EINVAL is permanent", syscall.EINVAL, false}, + {"a plain error is permanent", errors.New("boom"), false}, + } { + t.Run(tc.name, func(t *testing.T) { + if got := isTemporaryAccept(tc.err); got != tc.want { + t.Errorf("isTemporaryAccept(%v) = %v, want %v", tc.err, got, tc.want) + } + }) + } +} diff --git a/core/listener/vmcontrol/shortdir_test.go b/core/listener/vmcontrol/shortdir_test.go new file mode 100644 index 000000000..8f242ef88 --- /dev/null +++ b/core/listener/vmcontrol/shortdir_test.go @@ -0,0 +1,87 @@ +//go:build unix + +package vmcontrol + +import ( + "fmt" + "os" + "path/filepath" + "strings" + "testing" +) + +// The temporary directories these tests bind Unix sockets in. Ported from +// moca's remote-worker/internal/shortdir @ c13d556: a socket path is bounded +// by sockaddr_un.sun_path (104 bytes on darwin, 108 on Linux) and bind fails +// with EINVAL past it, so t.TempDir() — whose name embeds the test's — is +// regularly too long on a Mac (/var/folders/... is 49 bytes before the test +// name even starts). But hardcoding /tmp fails wherever /tmp is not +// writable, and go test honours TMPDIR, which a sandbox sets. So: prefer +// TMPDIR when a directory under it leaves room for the caller's socket name, +// and fall back to /tmp and the OS default otherwise. + +// maxSocketPath is the smaller of the two platform limits, so a path that +// fits here fits everywhere this code runs. +const maxSocketPath = 104 + +// shortDir returns a directory for a test's Unix sockets, removed when the +// test ends. need is how many bytes the caller will append for its longest +// socket path, separator included. +func shortDir(t *testing.T, prefix string, need int) string { + t.Helper() + dir, err := makeShortDir(prefix, need) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = os.RemoveAll(dir) }) + return dir +} + +func makeShortDir(prefix string, need int) (string, error) { + var tried []string + for _, base := range shortDirCandidates() { + // The length that matters is the directory that actually exists: + // MkdirTemp's random suffix is up to 10 digits, so neither the base + // alone nor a fixed allowance predicts it. Create first, measure, and + // hand the directory back if it does not fit. + dir, err := os.MkdirTemp(base, prefix) + if err != nil { + tried = append(tried, fmt.Sprintf("%s (%v)", base, err)) + continue + } + if len(dir)+need > maxSocketPath { + _ = os.RemoveAll(dir) + tried = append(tried, fmt.Sprintf( + "%s (%s is %d bytes, %d with the caller's %d-byte socket name, over the %d limit)", + base, dir, len(dir), len(dir)+need, need, maxSocketPath, + )) + continue + } + return dir, nil + } + return "", fmt.Errorf( + "shortdir: no directory both writable and short enough to hold a %d-byte socket name under "+ + "it (sun_path limit %d); tried: %s", + need, maxSocketPath, strings.Join(tried, "; "), + ) +} + +// shortDirCandidates lists the parent directories to try, in order of +// preference: TMPDIR first (the knob everything else honours), then /tmp, +// then the Go default. Duplicates and empties are dropped. +func shortDirCandidates() []string { + var out []string + seen := map[string]bool{} + for _, d := range []string{os.Getenv("TMPDIR"), "/tmp", os.TempDir()} { + if d == "" { + continue + } + d = filepath.Clean(d) + if seen[d] { + continue + } + seen[d] = true + out = append(out, d) + } + return out +} diff --git a/core/listener/vmcontrol/wire.go b/core/listener/vmcontrol/wire.go new file mode 100644 index 000000000..87db8fd41 --- /dev/null +++ b/core/listener/vmcontrol/wire.go @@ -0,0 +1,467 @@ +//go:build unix + +package vmcontrol + +import ( + "encoding/binary" + "errors" + "fmt" + "io" + "net" + + "golang.org/x/sys/unix" +) + +// The control protocol's three protobuf messages, hand-rolled. Cortex cannot +// import moca's gen module (its modules use a local replace and no submodule +// tags are published), so these are encoded and decoded here, wire-compatible +// byte for byte with egressv1: +// +// message ControlRequest { +// oneof msg { +// Register register = 1; // { string vm_id = 1; string sid = 2; bytes grant = 3; } +// Unregister unregister = 2; // { string vm_id = 1; } +// reserved 3; // Rebind, MI1 S2 +// } +// } +// message ControlReply { +// bool ok = 1; string code = 2; string message = 3; +// } +// +// The golden-byte tests in wire_test.go pin the encoding against bytes +// produced by moca's own proto.Marshal. + +// ControlRequest is one framed request. The oneof is modelled as two pointer +// fields of which at most one is set, matching how the decoder keeps the last +// case it saw. +type ControlRequest struct { + Register *Register + Unregister *Unregister +} + +// Register asks the proxy to bind vmID's listener; the fd rides beside the +// frame as SCM_RIGHTS (see writeFrame). +type Register struct { + VmID string + Sid string + Grant []byte +} + +// Unregister asks the proxy to drop vmID's binding. +type Unregister struct { + VmID string +} + +// ControlReply is the server's answer to one ControlRequest. +type ControlReply struct { + Ok bool + Code string + Message string +} + +// Wire types, for skipUnknown. +const ( + wireVarint = 0 + wire64Bit = 1 + wireBytes = 2 + wireStart = 3 // groups — removed in proto3, refused below + wireEnd = 4 + wire32Bit = 5 +) + +func appendTag(b []byte, field, wire int) []byte { + return appendVarint(b, uint64(field)<<3|uint64(wire)) +} + +func appendVarint(b []byte, v uint64) []byte { + for v >= 0x80 { + b = append(b, byte(v)|0x80) + v >>= 7 + } + return append(b, byte(v)) +} + +// appendString appends a proto3 string field, omitting the empty value as +// proto3 marshal does. +func appendString(b []byte, field int, s string) []byte { + if s == "" { + return b + } + b = appendTag(b, field, wireBytes) + b = appendVarint(b, uint64(len(s))) + return append(b, s...) +} + +func appendBytesField(b []byte, field int, v []byte) []byte { + if len(v) == 0 { + return b + } + b = appendTag(b, field, wireBytes) + b = appendVarint(b, uint64(len(v))) + return append(b, v...) +} + +func appendSubmessage(b []byte, field int, sub []byte) []byte { + b = appendTag(b, field, wireBytes) + b = appendVarint(b, uint64(len(sub))) + return append(b, sub...) +} + +// Marshal encodes req. A set oneof case is emitted even when its submessage is +// empty, as proto3 does for oneof fields. +func marshalControlRequest(req *ControlRequest) []byte { + var b []byte + switch { + case req.Register != nil: + var sub []byte + sub = appendString(sub, 1, req.Register.VmID) + sub = appendString(sub, 2, req.Register.Sid) + sub = appendBytesField(sub, 3, req.Register.Grant) + b = appendSubmessage(b, 1, sub) + case req.Unregister != nil: + var sub []byte + sub = appendString(sub, 1, req.Unregister.VmID) + b = appendSubmessage(b, 2, sub) + } + return b +} + +func marshalControlReply(rep *ControlReply) []byte { + var b []byte + if rep.Ok { + b = appendTag(b, 1, wireVarint) + b = appendVarint(b, 1) + } + b = appendString(b, 2, rep.Code) + b = appendString(b, 3, rep.Message) + return b +} + +// decoder walks a protobuf message body. +type decoder struct { + b []byte +} + +func (d *decoder) more() bool { return len(d.b) > 0 } + +// varint reads one base-128 value. A run longer than 10 bytes never +// terminates and is refused rather than spun on. +func (d *decoder) varint() (uint64, error) { + var v uint64 + for i := 0; i < 10; i++ { + if len(d.b) == 0 { + return 0, io.ErrUnexpectedEOF + } + c := d.b[0] + d.b = d.b[1:] + if c < 0x80 { + return v | uint64(c), nil + } + v |= uint64(c&0x7f) << (7 * uint(i)) + } + return 0, errors.New("vm control: unterminated varint") +} + +// field reads a field key and returns the field number and wire type. +func (d *decoder) field() (num, wire int, err error) { + key, err := d.varint() + if err != nil { + return 0, 0, err + } + num, wire = int(key>>3), int(key&7) + if num == 0 { + return 0, 0, errors.New("vm control: field number 0") + } + return num, wire, nil +} + +// length reads a LEN-delimited value and returns it without copying. +func (d *decoder) length() ([]byte, error) { + n, err := d.varint() + if err != nil { + return nil, err + } + if n > uint64(len(d.b)) { + return nil, io.ErrUnexpectedEOF + } + v := d.b[:n] + d.b = d.b[n:] + return v, nil +} + +// skipUnknown advances past one field the decoder does not know, so future +// revisions (S2's Rebind among them) do not break us. Group wire types are +// refused: proto3 removed them, and matching their end markers faithfully is +// more work than refusing a shape no encoder we interoperate with emits. +func (d *decoder) skipUnknown(wire int) error { + switch wire { + case wireVarint: + _, err := d.varint() + return err + case wire64Bit: + if len(d.b) < 8 { + return io.ErrUnexpectedEOF + } + d.b = d.b[8:] + return nil + case wireBytes: + _, err := d.length() + return err + case wire32Bit: + if len(d.b) < 4 { + return io.ErrUnexpectedEOF + } + d.b = d.b[4:] + return nil + default: + return fmt.Errorf("vm control: unsupported wire type %d", wire) + } +} + +// decodeControlRequest parses a request body. Unknown fields are skipped; +// within the oneof, the last case seen wins, as proto3 unmarshal does. +func decodeControlRequest(body []byte) (*ControlRequest, error) { + req := &ControlRequest{} + d := &decoder{b: body} + for d.more() { + num, wire, err := d.field() + if err != nil { + return nil, err + } + if wire != wireBytes { + if err := d.skipUnknown(wire); err != nil { + return nil, err + } + continue + } + switch num { + case 1: + sub, err := d.length() + if err != nil { + return nil, err + } + reg, err := decodeRegister(sub) + if err != nil { + return nil, err + } + req.Register, req.Unregister = reg, nil + case 2: + sub, err := d.length() + if err != nil { + return nil, err + } + un, err := decodeUnregister(sub) + if err != nil { + return nil, err + } + req.Register, req.Unregister = nil, un + default: + if err := d.skipUnknown(wire); err != nil { + return nil, err + } + } + } + return req, nil +} + +func decodeRegister(body []byte) (*Register, error) { + reg := &Register{} + d := &decoder{b: body} + for d.more() { + num, wire, err := d.field() + if err != nil { + return nil, err + } + if wire != wireBytes { + if err := d.skipUnknown(wire); err != nil { + return nil, err + } + continue + } + v, err := d.length() + if err != nil { + return nil, err + } + switch num { + case 1: + reg.VmID = string(v) + case 2: + reg.Sid = string(v) + case 3: + reg.Grant = append([]byte(nil), v...) + } + } + return reg, nil +} + +func decodeUnregister(body []byte) (*Unregister, error) { + un := &Unregister{} + d := &decoder{b: body} + for d.more() { + num, wire, err := d.field() + if err != nil { + return nil, err + } + if wire != wireBytes { + if err := d.skipUnknown(wire); err != nil { + return nil, err + } + continue + } + v, err := d.length() + if err != nil { + return nil, err + } + if num == 1 { + un.VmID = string(v) + } + } + return un, nil +} + +func decodeControlReply(body []byte) (*ControlReply, error) { + rep := &ControlReply{} + d := &decoder{b: body} + for d.more() { + num, wire, err := d.field() + if err != nil { + return nil, err + } + switch wire { + case wireVarint: + v, err := d.varint() + if err != nil { + return nil, err + } + if num == 1 { + rep.Ok = v != 0 + } + case wireBytes: + v, err := d.length() + if err != nil { + return nil, err + } + switch num { + case 2: + rep.Code = string(v) + case 3: + rep.Message = string(v) + } + default: + if err := d.skipUnknown(wire); err != nil { + return nil, err + } + } + } + return rep, nil +} + +var errFrameTooLarge = errors.New("vm control: frame exceeds 64 KiB") + +// maxFDsPerRead bounds the fds accepted on one receive. A well-formed +// Register carries one; the headroom lets us receive and close surplus fds +// instead of the kernel dropping them silently. +const maxFDsPerRead = 8 + +// writeFrame sends body as one frame. If fd >= 0 it rides as SCM_RIGHTS on the +// sendmsg that carries the frame's first byte, which is the pairing readFrame +// relies on. +func writeFrame(c *net.UnixConn, body []byte, fd int) error { + if len(body) > maxFrame { + return errFrameTooLarge + } + buf := make([]byte, 4+len(body)) + binary.BigEndian.PutUint32(buf, uint32(len(body))) + copy(buf[4:], body) + + var oob []byte + if fd >= 0 { + oob = unix.UnixRights(fd) + } + n, _, err := c.WriteMsgUnix(buf, oob, nil) + if err != nil { + return err + } + // A stream sendmsg may be partial; the fd was attached to byte 0 and has + // been sent. + for n < len(buf) { + m, err := c.Write(buf[n:]) + if err != nil { + return err + } + n += m + } + return nil +} + +// readFrame returns one frame body and every fd received while reading it. +// The caller owns the fds (they are raw ints, close-on-exec) and must close +// those it does not use. +func readFrame(c *net.UnixConn) (body []byte, fds []int, err error) { + defer func() { + if err != nil { + closeFDs(fds) + fds = nil + } + }() + var hdr [4]byte + if err = readFull(c, hdr[:], &fds); err != nil { + return nil, fds, err + } + n := binary.BigEndian.Uint32(hdr[:]) + if n > maxFrame { + return nil, fds, errFrameTooLarge + } + body = make([]byte, n) + if err = readFull(c, body, &fds); err != nil { + if errors.Is(err, io.EOF) { + err = io.ErrUnexpectedEOF + } + return nil, fds, err + } + return body, fds, nil +} + +// readFull fills p, collecting fds from the ancillary data of every receive, +// since the kernel may split a frame across reads and the fd arrives with +// whichever read covers its byte. +func readFull(c *net.UnixConn, p []byte, fds *[]int) error { + oob := make([]byte, unix.CmsgSpace(4*maxFDsPerRead)) + for off := 0; off < len(p); { + n, oobn, flags, _, err := c.ReadMsgUnix(p[off:], oob) + if oobn > 0 { + msgs, perr := unix.ParseSocketControlMessage(oob[:oobn]) + if perr != nil { + return fmt.Errorf("parse control message: %w", perr) + } + for i := range msgs { + rights, perr := unix.ParseUnixRights(&msgs[i]) + if perr != nil { + continue // not SCM_RIGHTS + } + for _, fd := range rights { + // net's ReadMsgUnix cannot pass MSG_CMSG_CLOEXEC, so there is a short + // window between the receive and this call in which a concurrent + // fork+exec could inherit the fd. Accepted: the proxy execs nothing on + // this path. + unix.CloseOnExec(fd) + } + *fds = append(*fds, rights...) + } + } + if flags&unix.MSG_CTRUNC != 0 { + return errors.New("vm control: ancillary data truncated") + } + off += n + if err != nil { + return err + } + } + return nil +} + +func closeFDs(fds []int) { + for _, fd := range fds { + _ = unix.Close(fd) + } +} diff --git a/core/listener/vmcontrol/wire_test.go b/core/listener/vmcontrol/wire_test.go new file mode 100644 index 000000000..8075d0f41 --- /dev/null +++ b/core/listener/vmcontrol/wire_test.go @@ -0,0 +1,270 @@ +//go:build unix + +package vmcontrol + +import ( + "bytes" + "encoding/hex" + "io" + "net" + "os" + "testing" + + "golang.org/x/sys/unix" +) + +// The golden bytes below were produced by moca's own proto.Marshal +// (egressv1.ControlRequest / ControlReply, remote-worker @ c13d556), so what +// this codec emits is pinned byte for byte against the far end of the wire — +// which Cortex cannot generate at test time, since moca's gen module is +// unpublished and the sandbox has no protoc. + +func hexBytes(t *testing.T, s string) []byte { + t.Helper() + b, err := hex.DecodeString(s) + if err != nil { + t.Fatalf("bad hex fixture: %v", err) + } + return b +} + +func TestGoldenMarshal(t *testing.T) { + cases := []struct { + name string + got []byte + want string + }{ + { + "register(vm-a, sid-a)", + marshalControlRequest(&ControlRequest{Register: &Register{VmID: "vm-a", Sid: "sid-a"}}), + "0a0d0a04766d2d6112057369642d61", + }, + { + "register(vm-g, sid-a, grant)", + marshalControlRequest(&ControlRequest{Register: &Register{VmID: "vm-g", Sid: "sid-a", Grant: []byte("a-grant")}}), + "0a160a04766d2d6712057369642d611a07612d6772616e74", + }, + { + // a set oneof case is emitted even when every field is empty + "register(empty)", + marshalControlRequest(&ControlRequest{Register: &Register{}}), + "0a00", + }, + { + "unregister(vm-a)", + marshalControlRequest(&ControlRequest{Unregister: &Unregister{VmID: "vm-a"}}), + "12060a04766d2d61", + }, + { + "empty request", + marshalControlRequest(&ControlRequest{}), + "", + }, + { + "reply ok", + marshalControlReply(&ControlReply{Ok: true}), + "0801", + }, + { + "reply refused(bad_request, register carries no fd)", + marshalControlReply(&ControlReply{Code: "bad_request", Message: "register carries no fd"}), + "120b6261645f726571756573741a1672656769737465722063617272696573206e6f206664", + }, + } + for _, tc := range cases { + if !bytes.Equal(tc.got, hexBytes(t, tc.want)) { + t.Errorf("%s: got %x, want %s", tc.name, tc.got, tc.want) + } + } +} + +func TestGoldenUnmarshal(t *testing.T) { + req, err := decodeControlRequest(hexBytes(t, "0a0d0a04766d2d6112057369642d61")) + if err != nil { + t.Fatal(err) + } + if req.Register == nil || req.Register.VmID != "vm-a" || req.Register.Sid != "sid-a" || req.Unregister != nil { + t.Errorf("register decode: %+v", req) + } + req, err = decodeControlRequest(hexBytes(t, "0a160a04766d2d6712057369642d611a07612d6772616e74")) + if err != nil { + t.Fatal(err) + } + if req.Register == nil || string(req.Register.Grant) != "a-grant" { + t.Errorf("grant decode: %+v", req) + } + req, err = decodeControlRequest(hexBytes(t, "12060a04766d2d61")) + if err != nil { + t.Fatal(err) + } + if req.Unregister == nil || req.Unregister.VmID != "vm-a" || req.Register != nil { + t.Errorf("unregister decode: %+v", req) + } + rep, err := decodeControlReply(hexBytes(t, "120b6261645f726571756573741a1672656769737465722063617272696573206e6f206664")) + if err != nil { + t.Fatal(err) + } + if rep.Ok || rep.Code != "bad_request" || rep.Message != "register carries no fd" { + t.Errorf("refusal decode: %+v", rep) + } +} + +func TestRequestRoundTrip(t *testing.T) { + cases := []struct { + name string + req ControlRequest + }{ + {"register", ControlRequest{Register: &Register{VmID: "vm-7", Sid: "run-9", Grant: []byte{0, 1, 2}}}}, + {"register empty", ControlRequest{Register: &Register{}}}, + {"unregister", ControlRequest{Unregister: &Unregister{VmID: "vm-7"}}}, + {"empty", ControlRequest{}}, + } + for _, tc := range cases { + got, err := decodeControlRequest(marshalControlRequest(&tc.req)) + if err != nil { + t.Fatalf("%s: %v", tc.name, err) + } + want := tc.req + switch { + case want.Register == nil && got.Register != nil, want.Register != nil && got.Register == nil, + want.Unregister == nil && got.Unregister != nil, want.Unregister != nil && got.Unregister == nil: + t.Errorf("%s: oneof shape changed: %+v", tc.name, got) + case want.Register != nil: + if got.Register.VmID != want.Register.VmID || got.Register.Sid != want.Register.Sid || + !bytes.Equal(got.Register.Grant, want.Register.Grant) { + t.Errorf("%s: got %+v want %+v", tc.name, *got.Register, *want.Register) + } + case want.Unregister != nil: + if got.Unregister.VmID != want.Unregister.VmID { + t.Errorf("%s: got %+v want %+v", tc.name, *got.Unregister, *want.Unregister) + } + } + } +} + +// Unknown fields must be skipped, not fatal: S2's Rebind rides field 3 of +// ControlRequest, and future revisions may add more. Within the oneof, the +// last case seen wins, as proto3 unmarshal does. +func TestDecodeSkipsUnknownFields(t *testing.T) { + base := hexBytes(t, "0a0d0a04766d2d6112057369642d61") // register(vm-a, sid-a) + suffixes := map[string][]byte{ + "rebind-shaped (field 3, empty)": {0x1a, 0x00}, + "unknown varint (field 4)": {0x20, 0x2a}, + "unknown 64-bit (field 5)": {0x29, 1, 2, 3, 4, 5, 6, 7, 8}, + "unknown LEN (field 6)": {0x32, 0x03, 'a', 'b', 'c'}, + "unknown 32-bit (field 7)": {0x35, 1, 2, 3, 4}, + } + for name, suffix := range suffixes { + req, err := decodeControlRequest(append(append([]byte{}, base...), suffix...)) + if err != nil { + t.Errorf("%s: %v", name, err) + continue + } + if req.Register == nil || req.Register.VmID != "vm-a" || req.Register.Sid != "sid-a" { + t.Errorf("%s: payload lost: %+v", name, req) + } + } + // The same skipping inside a submessage: Register with an unknown field 4. + req, err := decodeControlRequest([]byte{0x0a, 0x07, 0x0a, 0x03, 'v', 'm', 'x', 0x20, 0x01}) + if err != nil { + t.Fatal(err) + } + if req.Register == nil || req.Register.VmID != "vmx" { + t.Errorf("submessage unknown field: %+v", req) + } +} + +func TestDecodeRejectsGroupsAndTruncation(t *testing.T) { + if _, err := decodeControlRequest([]byte{0x0b}); err == nil { // field 1, group start + t.Error("group wire type accepted") + } + // A submessage whose length runs past the body: length 5, one byte present. + if _, err := decodeControlRequest([]byte{0x0a, 0x05, 0x0a}); err == nil { + t.Error("truncated submessage accepted") + } + // An unterminated varint (a run of continuation bytes) is refused, not spun on. + if _, err := decodeControlRequest(bytes.Repeat([]byte{0xff}, 15)); err == nil { + t.Error("unterminated varint accepted") + } +} + +func TestFrameRoundTrip(t *testing.T) { + a, b := socketPair(t) + defer a.Close() + defer b.Close() + + body := []byte("a control frame body") + if err := writeFrame(a, body, -1); err != nil { + t.Fatal(err) + } + got, fds, err := readFrame(b) + if err != nil { + t.Fatal(err) + } + if !bytes.Equal(got, body) || len(fds) != 0 { + t.Errorf("got %q fds %v", got, fds) + } +} + +func TestFrameOversize(t *testing.T) { + a, b := socketPair(t) + defer a.Close() + defer b.Close() + + if err := writeFrame(a, make([]byte, maxFrame+1), -1); err != errFrameTooLarge { + t.Fatalf("write: got %v, want errFrameTooLarge", err) + } + // A length header lying about it is caught on read: the connection dies + // rather than the reader allocating to a hostile size. + hdr := make([]byte, 4) + hdr[0] = 0xff // > maxFrame in the first byte alone + if _, err := a.Write(hdr); err != nil { + t.Fatal(err) + } + if _, _, err := readFrame(b); err != errFrameTooLarge { + t.Fatalf("read: got %v, want errFrameTooLarge", err) + } +} + +func TestFrameTruncatedEOF(t *testing.T) { + a, b := socketPair(t) + defer a.Close() + + hdr := []byte{0, 0, 0, 8} + if _, err := a.Write(hdr); err != nil { + t.Fatal(err) + } + if _, err := a.Write([]byte("four")); err != nil { + t.Fatal(err) + } + a.Close() // half a frame, then the stream ends + if _, _, err := readFrame(b); err != io.ErrUnexpectedEOF { + t.Fatalf("got %v, want ErrUnexpectedEOF", err) + } +} + +// socketPair is a connected Unix socketpair, the transport readFrame and +// writeFrame speak over. +func socketPair(t *testing.T) (*net.UnixConn, *net.UnixConn) { + t.Helper() + fds, err := unix.Socketpair(unix.AF_UNIX, unix.SOCK_STREAM, 0) + if err != nil { + t.Fatal(err) + } + return fileConn(t, fds[0]), fileConn(t, fds[1]) +} + +func fileConn(t *testing.T, fd int) *net.UnixConn { + t.Helper() + f := os.NewFile(uintptr(fd), "test-sock") + c, err := net.FileConn(f) + if err != nil { + t.Fatal(err) + } + _ = f.Close() // FileConn dups; the original must not hold the socket open + uc, ok := c.(*net.UnixConn) + if !ok { + t.Fatalf("file conn is %T", c) + } + return uc +} diff --git a/core/pipeline/session.go b/core/pipeline/session.go index 051eb3f45..dae9862cd 100644 --- a/core/pipeline/session.go +++ b/core/pipeline/session.go @@ -756,6 +756,11 @@ const ( SessionSourceAffinity SessionSource = "affinity" // SessionSourceActive: the most recently active session. SessionSourceActive SessionSource = "active" + // SessionSourceConnection: the listener's own trusted answer, resolved + // from how the connection arrived rather than anything the client sent + // (which per-VM socket accepted it, vmcontrol). Client-asserted headers + // cannot override it. + SessionSourceConnection SessionSource = "connection" ) // SessionView is a read-only snapshot of a session, safe to pass to plugins.