amadan

carlos / platform Public

Let sidecars dial the agent, and let nothing dial a sidecar

Paul Campbell pushed by paul@keymail.dev e049d99439479a72e92dc6e1e6f29317fe570148 parent b2db5e7
2 files changed, +883 −0
  • internal/agentsock/agentsock.go +325 −0
  • internal/agentsock/agentsock_test.go +558 −0
diff --git a/internal/agentsock/agentsock.go b/internal/agentsock/agentsock.go
new file mode 100644
index 000000000..3c091e561
--- /dev/null
+++ b/internal/agentsock/agentsock.go
@@ -0,0 +1,325 @@
+// Package agentsock is the agent's own listening socket — the ONE channel
+// between a sidecar and the platform, and it points inward: children dial
+// the agent, the agent never dials a child. That is what lets a sidecar
+// bind nothing, and what lets the identical artifact run off-platform
+// behind NAT with no protocol difference.
+//
+// Every exec child runs as the agent's user, so the socket's file mode is
+// not an identity boundary. Identity is the per-wake token the activator
+// mints and passes in the child's environment (CARLOS_SIDECAR_TOKEN),
+// scoped to one host for one wake: a token for A cannot speak about B, and
+// a token from a previous wake matches nothing.
+//
+// # What the socket trusts
+//
+// Nothing the caller sends, except as material for the one check. The
+// Carlos-Sidecar-Host header names a host, but naming it proves nothing —
+// only holding that host's live token does, and every effect is then applied
+// to the host the TOKEN authorized. No body carries a host field, and none
+// ever should: that is the confused-deputy shape this design exists to
+// exclude. Bodies are read under a 4KiB cap, parsed strictly, and refused on
+// anything unexpected; a refused call — for any reason, at any stage — has no
+// effect at all.
+//
+// Refusals are deliberately uniform. Wrong token, absent token, unknown host,
+// and a host whose worker is asleep all produce one identical 401, so the
+// socket cannot be used to enumerate hosts or probe which ones are awake.
+package agentsock
+
+import (
+ "context"
+ "crypto/rand"
+ "crypto/subtle"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "io"
+ "log"
+ "net"
+ "net/http"
+ "os"
+ "strings"
+ "time"
+)
+
+// Workers is the activator's half of this conversation. It is declared here,
+// not imported, so the socket stays testable and the direction stays plain:
+// this package is told about workers, and never reaches for one.
+//
+// WorkerToken returns "" for a host that is unknown, asleep, or stopped. That
+// is a REFUSAL, never "no token required" — see authorize.
+type Workers interface {
+ WorkerToken(host string) string
+ NoteWorkerContact(host string)
+ SetWorkerNextDue(host string, at int64) error
+ SleepWorker(host string)
+ NoteWorkerReady(host, version string)
+}
+
+const (
+ // hostHeader names the host a caller claims to be. Untrusted on its own.
+ hostHeader = "Carlos-Sidecar-Host"
+
+ // maxBody bounds every body we read, following the next-due decode in
+ // activator/backend_exec.go. These messages are two small fields; a
+ // caller sending more is malformed, not generous.
+ maxBody = 1 << 12
+
+ // maxVersion bounds the build string a child reports. It is displayed to
+ // members, so it is length- and character-checked before it is recorded.
+ maxVersion = 200
+
+ // socketMode keeps the socket off the world. It is defence in depth, not
+ // the boundary: exec children share the agent's uid, so the mode cannot
+ // tell one child from another. The token does that.
+ socketMode = 0o660
+
+ // Local calls, all of them. Nothing here should ever be slow, and a
+ // child that stalls mid-request must not pin a connection.
+ readHeaderTimeout = 2 * time.Second
+ readTimeout = 5 * time.Second
+ writeTimeout = 5 * time.Second
+ idleTimeout = 30 * time.Second
+ shutdownGrace = 3 * time.Second
+
+ // dialProbe is how long we wait to learn whether an existing socket file
+ // still has somebody listening on it.
+ dialProbe = 250 * time.Millisecond
+)
+
+// Server serves the agent socket. Zero per-request state: one instance
+// answers every child on the box concurrently.
+type Server struct {
+ w Workers
+}
+
+// New builds the socket server over w.
+func New(w Workers) *Server { return &Server{w: w} }
+
+// Handler is the routing table, exported so tests can exercise it without
+// binding anything. Every verb is POST-only: the method pattern means a GET
+// is turned away by the router, before authorize and before any effect.
+func (s *Server) Handler() http.Handler {
+ mux := http.NewServeMux()
+ mux.Handle("POST /v1/ready", s.guard(s.ready))
+ mux.Handle("POST /v1/next-due", s.guard(s.nextDue))
+ mux.Handle("POST /v1/idle", s.guard(s.idle))
+ mux.Handle("POST /v1/heartbeat", s.guard(s.heartbeat))
+ return mux
+}
+
+// guard is the single gate. Nothing reaches a verb without passing it, and a
+// verb never sees the request's claimed host — only the authorized one, so
+// there is no way for a handler to act on a host it was not entitled to.
+func (s *Server) guard(fn func(http.ResponseWriter, *http.Request, string)) http.Handler {
+ return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ host, ok := s.authorize(r)
+ if !ok {
+ refuse(w)
+ return
+ }
+ fn(w, r, host)
+ })
+}
+
+// authorize resolves the caller to the one host it may speak about.
+//
+// The dangerous shape here is WorkerToken returning "" — unknown host, asleep,
+// stopped — being read as "this host needs no token", which would be a total
+// bypass for every host that is not currently awake. So the empty case is not
+// handled by a guard that a later edit could delete: an absent token is
+// replaced with fresh random bytes, and the comparison then simply cannot
+// succeed. Refusal is a property of the compare, not of a branch above it.
+//
+// The compare itself is constant-time, and both failure paths — no live token,
+// wrong live token — run the same code, so the structure carries no timing
+// signal beyond the length leak inherent to any comparison.
+func (s *Server) authorize(r *http.Request) (string, bool) {
+ host := r.Header.Get(hostHeader)
+ presented := bearer(r.Header.Get("Authorization"))
+
+ want := s.w.WorkerToken(host)
+ if want == "" {
+ want = unmatchable()
+ }
+ return host, subtle.ConstantTimeCompare([]byte(presented), []byte(want)) == 1
+}
+
+// unmatchable returns a value no caller can present. crypto/rand.Read never
+// fails (it panics on a broken system rather than returning short), so this
+// has no error path that could degrade into a guessable constant.
+func unmatchable() string {
+ var b [32]byte
+ rand.Read(b[:])
+ return string(b[:])
+}
+
+// bearer extracts the credential from an Authorization header, returning ""
+// for anything that is not exactly a non-empty Bearer credential. "Bearer "
+// with nothing after it yields "", which authorize then cannot match.
+func bearer(h string) string {
+ const scheme = "Bearer "
+ if len(h) > len(scheme) && strings.EqualFold(h[:len(scheme)], scheme) {
+ return h[len(scheme):]
+ }
+ return ""
+}
+
+// refuse is the only 401 in the package. One status, one body, for every
+// reason — a caller learns that it was refused and nothing else.
+func refuse(w http.ResponseWriter) {
+ http.Error(w, "unauthorized", http.StatusUnauthorized)
+}
+
+// bad is the only 400 in the package, for the same reason.
+func bad(w http.ResponseWriter) {
+ http.Error(w, "bad request", http.StatusBadRequest)
+}
+
+// ready ends the settle window and records the build the child says it is
+// running. The version is the child's own claim — §9.2's honesty rule — so it
+// is bounded and character-checked, never trusted. The field must be PRESENT
+// (a typo'd body is a loud 400, not a silent "unknown"), but may be empty: a
+// child that genuinely does not know its build says so, and is still ready.
+func (s *Server) ready(w http.ResponseWriter, r *http.Request, host string) {
+ var body struct {
+ Version *string `json:"version"`
+ }
+ if !decode(r, &body) {
+ bad(w)
+ return
+ }
+ if body.Version == nil || len(*body.Version) > maxVersion || !displayable(*body.Version) {
+ bad(w)
+ return
+ }
+ s.w.NoteWorkerReady(host, *body.Version)
+ w.WriteHeader(http.StatusNoContent)
+}
+
+// nextDue records when the child expects to work again, while it is alive to
+// say so. "at" must be present and non-negative: an absent field decoding to
+// zero would silently reschedule the worker to the epoch.
+func (s *Server) nextDue(w http.ResponseWriter, r *http.Request, host string) {
+ var body struct {
+ At *int64 `json:"at"`
+ }
+ if !decode(r, &body) {
+ bad(w)
+ return
+ }
+ if body.At == nil || *body.At < 0 {
+ bad(w)
+ return
+ }
+ if err := s.w.SetWorkerNextDue(host, *body.At); err != nil {
+ // The child must not read a 204 for a schedule that did not land.
+ log.Printf("agentsock: set next-due for %s: %v", host, err)
+ http.Error(w, "internal error", http.StatusInternalServerError)
+ return
+ }
+ w.WriteHeader(http.StatusNoContent)
+}
+
+// idle is "I have nothing to do". It carries no arguments, so nothing in the
+// request can steer it: the authorized host is the whole message. Whether the
+// request is honoured at all is the activator's call — an always-on worker's
+// idle is refused there.
+func (s *Server) idle(w http.ResponseWriter, _ *http.Request, host string) {
+ s.w.SleepWorker(host)
+ w.WriteHeader(http.StatusNoContent)
+}
+
+// heartbeat is a worker's only sign of life: nothing routes to a sidecar, so
+// no request will ever stamp one. Argument-free for the same reason as idle.
+func (s *Server) heartbeat(w http.ResponseWriter, _ *http.Request, host string) {
+ s.w.NoteWorkerContact(host)
+ w.WriteHeader(http.StatusNoContent)
+}
+
+// decode parses a bounded body into v. The cap is applied to the READER, so an
+// oversized body is truncated into a parse error rather than accepted: there is
+// no size at which a caller gets a partial message honoured.
+func decode(r *http.Request, v any) bool {
+ return json.NewDecoder(io.LimitReader(r.Body, maxBody)).Decode(v) == nil
+}
+
+// displayable rejects control characters in a string bound for member-facing
+// output — a version line should not be able to carry a NUL, a newline, or a
+// terminal escape into a log or a console page.
+func displayable(s string) bool {
+ for _, r := range s {
+ if r < 0x20 || r == 0x7f {
+ return false
+ }
+ }
+ return true
+}
+
+// Serve binds path and answers on it until ctx is cancelled, then shuts down
+// and leaves no socket file behind.
+//
+// A socket file that outlives its agent is the ordinary result of a hard kill,
+// and the next start must take it over rather than die on "address already in
+// use" — so a stale file is removed before binding. A file somebody is still
+// LISTENING on is a different thing entirely: unlinking that would orphan a
+// live agent, whose children would go on talking to a process nothing else can
+// reach. So we probe first, and refuse rather than steal.
+func (s *Server) Serve(ctx context.Context, path string) error {
+ if listening(path) {
+ return fmt.Errorf("agentsock: %s is already served by a live listener", path)
+ }
+ if err := os.Remove(path); err != nil && !errors.Is(err, os.ErrNotExist) {
+ return fmt.Errorf("agentsock: remove stale socket %s: %w", path, err)
+ }
+ ln, err := net.Listen("unix", path)
+ if err != nil {
+ return fmt.Errorf("agentsock: listen on %s: %w", path, err)
+ }
+ // Go unlinks a unix socket it created when the listener closes; this is
+ // the belt to that braces, so no exit path can leave the file behind.
+ defer func() { _ = os.Remove(path) }()
+
+ if err := os.Chmod(path, socketMode); err != nil {
+ ln.Close()
+ return fmt.Errorf("agentsock: chmod %s: %w", path, err)
+ }
+
+ srv := &http.Server{
+ Handler: s.Handler(),
+ ReadHeaderTimeout: readHeaderTimeout,
+ ReadTimeout: readTimeout,
+ WriteTimeout: writeTimeout,
+ IdleTimeout: idleTimeout,
+ MaxHeaderBytes: 1 << 14,
+ }
+
+ errc := make(chan error, 1)
+ go func() { errc <- srv.Serve(ln) }()
+
+ select {
+ case err := <-errc:
+ if errors.Is(err, http.ErrServerClosed) {
+ return nil
+ }
+ return err
+ case <-ctx.Done():
+ // ctx is already cancelled, so Shutdown gets its own deadline.
+ sctx, cancel := context.WithTimeout(context.Background(), shutdownGrace)
+ defer cancel()
+ err := srv.Shutdown(sctx)
+ <-errc
+ return err
+ }
+}
+
+// listening reports whether something is currently accepting on path. A stale
+// socket file, a regular file, and a missing path all answer false.
+func listening(path string) bool {
+ c, err := net.DialTimeout("unix", path, dialProbe)
+ if err != nil {
+ return false
+ }
+ c.Close()
+ return true
+}
diff --git a/internal/agentsock/agentsock_test.go b/internal/agentsock/agentsock_test.go
new file mode 100644
index 000000000..24e7633c8
--- /dev/null
+++ b/internal/agentsock/agentsock_test.go
@@ -0,0 +1,558 @@
+package agentsock
+
+import (
+ "context"
+ "errors"
+ "net"
+ "net/http"
+ "net/http/httptest"
+ "os"
+ "path/filepath"
+ "strings"
+ "sync"
+ "testing"
+ "time"
+)
+
+// fakeWorkers records every effect so a test can assert the ABSENCE of one.
+// "the call was refused" is only half the property; "and nothing happened" is
+// the half that catches an auth check placed after the action.
+type fakeWorkers struct {
+ mu sync.Mutex
+ tokens map[string]string
+ contacts []string
+ due map[string]int64
+ dueErr error
+ slept []string
+ ready map[string]string
+}
+
+func (f *fakeWorkers) WorkerToken(host string) string {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ return f.tokens[host]
+}
+
+func (f *fakeWorkers) NoteWorkerContact(host string) {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ f.contacts = append(f.contacts, host)
+}
+
+func (f *fakeWorkers) SetWorkerNextDue(host string, at int64) error {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ if f.dueErr != nil {
+ return f.dueErr
+ }
+ if f.due == nil {
+ f.due = map[string]int64{}
+ }
+ f.due[host] = at
+ return nil
+}
+
+func (f *fakeWorkers) SleepWorker(host string) {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ f.slept = append(f.slept, host)
+}
+
+func (f *fakeWorkers) NoteWorkerReady(host, version string) {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ if f.ready == nil {
+ f.ready = map[string]string{}
+ }
+ f.ready[host] = version
+}
+
+// effects renders every recorded side effect, so a test can assert a refused
+// call changed NOTHING rather than checking one slice and missing the rest.
+func (f *fakeWorkers) effects() (contacts, slept int, due map[string]int64, ready map[string]string) {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ due = map[string]int64{}
+ for k, v := range f.due {
+ due[k] = v
+ }
+ ready = map[string]string{}
+ for k, v := range f.ready {
+ ready[k] = v
+ }
+ return len(f.contacts), len(f.slept), due, ready
+}
+
+func (f *fakeWorkers) assertNoEffects(t *testing.T, why string) {
+ t.Helper()
+ contacts, slept, due, ready := f.effects()
+ if contacts != 0 || slept != 0 || len(due) != 0 || len(ready) != 0 {
+ t.Fatalf("%s: a refused call must have no effect at all; contacts=%d slept=%d due=%v ready=%v",
+ why, contacts, slept, due, ready)
+ }
+}
+
+func post(h http.Handler, path, host, tok, body string) *httptest.ResponseRecorder {
+ req := httptest.NewRequest("POST", path, strings.NewReader(body))
+ if host != "" {
+ req.Header.Set("Carlos-Sidecar-Host", host)
+ }
+ if tok != "" {
+ req.Header.Set("Authorization", "Bearer "+tok)
+ }
+ rec := httptest.NewRecorder()
+ h.ServeHTTP(rec, req)
+ return rec
+}
+
+func waitFor(t *testing.T, what string, cond func() bool) {
+ t.Helper()
+ deadline := time.Now().Add(5 * time.Second)
+ for time.Now().Before(deadline) {
+ if cond() {
+ return
+ }
+ time.Sleep(2 * time.Millisecond)
+ }
+ t.Fatalf("timed out waiting for %s", what)
+}
+
+var allVerbs = []string{"/v1/ready", "/v1/next-due", "/v1/idle", "/v1/heartbeat"}
+
+func TestAgentSockRejectsWrongToken(t *testing.T) {
+ f := &fakeWorkers{tokens: map[string]string{"bot.x.internal": "good"}}
+ h := New(f).Handler()
+
+ for _, tc := range []struct{ name, host, tok string }{
+ {"wrong token", "bot.x.internal", "bad"},
+ {"no token", "bot.x.internal", ""},
+ {"unknown host", "nope.x.internal", "good"},
+ {"token for another host", "other.x.internal", "good"},
+ {"no host header", "", "good"},
+ {"empty-string token against an asleep host", "asleep.x.internal", ""},
+ } {
+ rec := post(h, "/v1/heartbeat", tc.host, tc.tok, "")
+ if rec.Code != http.StatusUnauthorized {
+ t.Fatalf("%s: code = %d, want 401", tc.name, rec.Code)
+ }
+ }
+ f.assertNoEffects(t, "wrong-token sweep")
+}
+
+// The inversion this guards is total: WorkerToken("") means "unknown, asleep,
+// or stopped", never "no token required". A caller who presents nothing must
+// not match a host that holds nothing.
+func TestAgentSockEmptyStoredTokenIsNotABypass(t *testing.T) {
+ // "asleep" is present in the map with the empty token the real activator
+ // returns for a host that is not currently awake or waking.
+ f := &fakeWorkers{tokens: map[string]string{"asleep.x.internal": ""}}
+ h := New(f).Handler()
+
+ for _, tok := range []string{"", " ", "Bearer", "anything"} {
+ for _, path := range allVerbs {
+ body := ""
+ switch path {
+ case "/v1/ready":
+ body = `{"version":"v1"}`
+ case "/v1/next-due":
+ body = `{"at":1756}`
+ }
+ rec := post(h, path, "asleep.x.internal", tok, body)
+ if rec.Code != http.StatusUnauthorized {
+ t.Fatalf("%s with token %q: code = %d, want 401", path, tok, rec.Code)
+ }
+ }
+ }
+ // The bare header form, which no post() helper shape produces: an
+ // Authorization header whose credential is literally empty.
+ req := httptest.NewRequest("POST", "/v1/heartbeat", nil)
+ req.Header.Set("Carlos-Sidecar-Host", "asleep.x.internal")
+ req.Header.Set("Authorization", "Bearer ")
+ rec := httptest.NewRecorder()
+ h.ServeHTTP(rec, req)
+ if rec.Code != http.StatusUnauthorized {
+ t.Fatalf("empty bearer credential: code = %d, want 401", rec.Code)
+ }
+ f.assertNoEffects(t, "empty stored token")
+}
+
+func TestAgentSockTokenIsScopedToOneHost(t *testing.T) {
+ // The security property: possession of A's token says nothing about B.
+ f := &fakeWorkers{tokens: map[string]string{"a.x.internal": "tok-a", "b.x.internal": "tok-b"}}
+ h := New(f).Handler()
+ if rec := post(h, "/v1/idle", "b.x.internal", "tok-a", ""); rec.Code != http.StatusUnauthorized {
+ t.Fatalf("A's token spoke about B: code = %d", rec.Code)
+ }
+ if rec := post(h, "/v1/ready", "b.x.internal", "tok-a", `{"version":"pwned"}`); rec.Code != http.StatusUnauthorized {
+ t.Fatalf("A's token reported B's version: code = %d", rec.Code)
+ }
+ if rec := post(h, "/v1/next-due", "b.x.internal", "tok-a", `{"at":1}`); rec.Code != http.StatusUnauthorized {
+ t.Fatalf("A's token rescheduled B: code = %d", rec.Code)
+ }
+ if rec := post(h, "/v1/heartbeat", "b.x.internal", "tok-a", ""); rec.Code != http.StatusUnauthorized {
+ t.Fatalf("A's token kept B alive: code = %d", rec.Code)
+ }
+ f.assertNoEffects(t, "cross-host token")
+}
+
+// Every refusal reason must be indistinguishable on the wire: status, headers
+// worth reading, and body. A caller must not be able to probe "does host X
+// exist" or "is host X awake" by reading the difference.
+func TestAgentSockRefusalsAreIndistinguishable(t *testing.T) {
+ f := &fakeWorkers{tokens: map[string]string{
+ "awake.x.internal": "good",
+ "asleep.x.internal": "",
+ }}
+ h := New(f).Handler()
+
+ type shape struct {
+ code int
+ ctype, body string
+ }
+ var seen []shape
+ for _, tc := range []struct{ name, host, tok string }{
+ {"wrong token", "awake.x.internal", "bad"},
+ {"wrong token, same length", "awake.x.internal", "goo1"},
+ {"absent token", "awake.x.internal", ""},
+ {"unknown host", "ghost.x.internal", "good"},
+ {"asleep host", "asleep.x.internal", "good"},
+ {"absent host header", "", "good"},
+ } {
+ rec := post(h, "/v1/heartbeat", tc.host, tc.tok, "")
+ seen = append(seen, shape{rec.Code, rec.Header().Get("Content-Type"), rec.Body.String()})
+ if rec.Code != http.StatusUnauthorized {
+ t.Fatalf("%s: code = %d, want 401", tc.name, rec.Code)
+ }
+ }
+ for i := 1; i < len(seen); i++ {
+ if seen[i] != seen[0] {
+ t.Fatalf("refusals are distinguishable: %+v vs %+v", seen[0], seen[i])
+ }
+ }
+ f.assertNoEffects(t, "indistinguishable refusals")
+}
+
+func TestAgentSockVerbs(t *testing.T) {
+ f := &fakeWorkers{tokens: map[string]string{"bot.x.internal": "good"}}
+ h := New(f).Handler()
+
+ if rec := post(h, "/v1/ready", "bot.x.internal", "good", `{"version":"abc123"}`); rec.Code != http.StatusNoContent {
+ t.Fatalf("ready: %d", rec.Code)
+ }
+ if f.ready["bot.x.internal"] != "abc123" {
+ t.Fatal("ready must record the running version — it is a sidecar's only version signal")
+ }
+ if rec := post(h, "/v1/next-due", "bot.x.internal", "good", `{"at":1756}`); rec.Code != http.StatusNoContent {
+ t.Fatalf("next-due: %d", rec.Code)
+ }
+ if f.due["bot.x.internal"] != 1756 {
+ t.Fatalf("due = %d", f.due["bot.x.internal"])
+ }
+ if rec := post(h, "/v1/idle", "bot.x.internal", "good", ""); rec.Code != http.StatusNoContent {
+ t.Fatalf("idle: %d", rec.Code)
+ }
+ if len(f.slept) != 1 {
+ t.Fatal("idle must ask for a stop")
+ }
+ if rec := post(h, "/v1/heartbeat", "bot.x.internal", "good", ""); rec.Code != http.StatusNoContent {
+ t.Fatalf("heartbeat: %d", rec.Code)
+ }
+ if len(f.contacts) != 1 || f.contacts[0] != "bot.x.internal" {
+ t.Fatalf("heartbeat must stamp contact for the CALLER's host; contacts = %v", f.contacts)
+ }
+}
+
+// The verb acts on the host the TOKEN authorized, never on a host named
+// anywhere in the body — there is no host field in any body, and adding one
+// would be the confused-deputy bug this pins.
+func TestAgentSockActsOnlyOnTheAuthorizedHost(t *testing.T) {
+ f := &fakeWorkers{tokens: map[string]string{"a.x.internal": "tok-a", "b.x.internal": "tok-b"}}
+ h := New(f).Handler()
+
+ if rec := post(h, "/v1/next-due", "a.x.internal", "tok-a", `{"at":99,"host":"b.x.internal"}`); rec.Code != http.StatusNoContent {
+ t.Fatalf("next-due: %d", rec.Code)
+ }
+ if _, ok := f.due["b.x.internal"]; ok {
+ t.Fatal("a body field must never redirect an effect onto another host")
+ }
+ if f.due["a.x.internal"] != 99 {
+ t.Fatalf("due for the authorized host = %d, want 99", f.due["a.x.internal"])
+ }
+}
+
+func TestAgentSockRefusesGarbage(t *testing.T) {
+ f := &fakeWorkers{tokens: map[string]string{"bot.x.internal": "good"}}
+ h := New(f).Handler()
+ for _, body := range []string{"not json", `{"at":"soon"}`, strings.Repeat("x", 1<<20), `{"at":-1}`} {
+ rec := post(h, "/v1/next-due", "bot.x.internal", "good", body)
+ if rec.Code != http.StatusBadRequest {
+ t.Fatalf("body %.20q: code = %d, want 400", body, rec.Code)
+ }
+ }
+ // An oversized body on a JSON-bearing verb must be refused, not truncated
+ // into something that happens to parse.
+ big := `{"version":"` + strings.Repeat("v", 1<<20) + `"}`
+ if rec := post(h, "/v1/ready", "bot.x.internal", "good", big); rec.Code != http.StatusBadRequest {
+ t.Fatalf("oversized ready: code = %d, want 400", rec.Code)
+ }
+ // The SIZE CAP itself, not the field bounds behind it: a body that is
+ // perfectly well-formed and whose fields are all in range, but megabytes
+ // long. Without a bounded reader this parses and acts.
+ pad := strings.Repeat("x", 1<<20)
+ if rec := post(h, "/v1/next-due", "bot.x.internal", "good", `{"at":1756,"pad":"`+pad+`"}`); rec.Code != http.StatusBadRequest {
+ t.Fatalf("padded next-due: code = %d, want 400 — the body must be read under a cap", rec.Code)
+ }
+ if rec := post(h, "/v1/ready", "bot.x.internal", "good", `{"version":"abc","pad":"`+pad+`"}`); rec.Code != http.StatusBadRequest {
+ t.Fatalf("padded ready: code = %d, want 400 — the body must be read under a cap", rec.Code)
+ }
+ for _, body := range []string{"not json", `{"version":42}`, "{\"version\":\"a\\u0000b\"}"} {
+ if rec := post(h, "/v1/ready", "bot.x.internal", "good", body); rec.Code != http.StatusBadRequest {
+ t.Fatalf("ready body %.20q: code = %d, want 400", body, rec.Code)
+ }
+ }
+ if _, ok := f.due["bot.x.internal"]; ok {
+ t.Fatal("a malformed body must leave no effect")
+ }
+ if len(f.ready) != 0 {
+ t.Fatalf("a malformed ready must leave no effect: %v", f.ready)
+ }
+
+ if rec := post(h, "/v1/nope", "bot.x.internal", "good", ""); rec.Code != http.StatusNotFound {
+ t.Fatalf("unknown verb: %d", rec.Code)
+ }
+ req := httptest.NewRequest("GET", "/v1/idle", nil)
+ // GET on a POST-only verb
+ rec := httptest.NewRecorder()
+ h.ServeHTTP(rec, req)
+ if rec.Code != http.StatusMethodNotAllowed && rec.Code != http.StatusUnauthorized {
+ t.Fatalf("GET: %d", rec.Code)
+ }
+}
+
+// A GET must not act even when it carries a perfectly good credential — the
+// method gate has to come before the effect, not merely before the 200.
+func TestAgentSockGetNeverActs(t *testing.T) {
+ f := &fakeWorkers{tokens: map[string]string{"bot.x.internal": "good"}}
+ h := New(f).Handler()
+ for _, path := range allVerbs {
+ for _, method := range []string{"GET", "HEAD", "PUT", "DELETE"} {
+ req := httptest.NewRequest(method, path, nil)
+ req.Header.Set("Carlos-Sidecar-Host", "bot.x.internal")
+ req.Header.Set("Authorization", "Bearer good")
+ rec := httptest.NewRecorder()
+ h.ServeHTTP(rec, req)
+ if rec.Code == http.StatusNoContent {
+ t.Fatalf("%s %s succeeded; only POST may act", method, path)
+ }
+ }
+ }
+ f.assertNoEffects(t, "non-POST methods")
+}
+
+// An empty next-due body is malformed: "at" is the whole message, and a
+// zero-valued default would silently reschedule the worker to the epoch.
+func TestAgentSockNextDueRequiresAt(t *testing.T) {
+ f := &fakeWorkers{tokens: map[string]string{"bot.x.internal": "good"}}
+ h := New(f).Handler()
+ for _, body := range []string{"", "{}", "null"} {
+ if rec := post(h, "/v1/next-due", "bot.x.internal", "good", body); rec.Code != http.StatusBadRequest {
+ t.Fatalf("next-due body %q: code = %d, want 400", body, rec.Code)
+ }
+ }
+ if len(f.due) != 0 {
+ t.Fatalf("due must be untouched: %v", f.due)
+ }
+}
+
+// The version field must be PRESENT — a typo'd body is a loud 400, never a
+// silent "unknown" that also ends the settle window. It may be empty, which is
+// a child honestly saying it does not know its own build.
+func TestAgentSockReadyRequiresTheVersionField(t *testing.T) {
+ f := &fakeWorkers{tokens: map[string]string{"bot.x.internal": "good"}}
+ h := New(f).Handler()
+ for _, body := range []string{"", "{}", "null", `{"verison":"abc"}`} {
+ if rec := post(h, "/v1/ready", "bot.x.internal", "good", body); rec.Code != http.StatusBadRequest {
+ t.Fatalf("ready body %q: code = %d, want 400", body, rec.Code)
+ }
+ }
+ if len(f.ready) != 0 {
+ t.Fatalf("ready must be untouched: %v", f.ready)
+ }
+ if rec := post(h, "/v1/ready", "bot.x.internal", "good", `{"version":""}`); rec.Code != http.StatusNoContent {
+ t.Fatalf("explicitly-unknown version: code = %d, want 204", rec.Code)
+ }
+ if v, ok := f.ready["bot.x.internal"]; !ok || v != "" {
+ t.Fatalf("ready = %q, %v; an explicit empty version is honest, and still ready", v, ok)
+ }
+}
+
+// Bodyless verbs stay bodyless: idle and heartbeat carry no arguments, so
+// nothing a caller sends can steer them.
+func TestAgentSockBodylessVerbsIgnoreTheirBody(t *testing.T) {
+ f := &fakeWorkers{tokens: map[string]string{"bot.x.internal": "good"}}
+ h := New(f).Handler()
+ if rec := post(h, "/v1/heartbeat", "bot.x.internal", "good", `{"host":"other.x.internal"}`); rec.Code != http.StatusNoContent {
+ t.Fatalf("heartbeat with a body: %d", rec.Code)
+ }
+ if len(f.contacts) != 1 || f.contacts[0] != "bot.x.internal" {
+ t.Fatalf("contacts = %v", f.contacts)
+ }
+}
+
+func TestAgentSockNextDueStoreFailureIsNot204(t *testing.T) {
+ f := &fakeWorkers{
+ tokens: map[string]string{"bot.x.internal": "good"},
+ dueErr: errors.New("index write failed"),
+ }
+ h := New(f).Handler()
+ rec := post(h, "/v1/next-due", "bot.x.internal", "good", `{"at":1756}`)
+ if rec.Code != http.StatusInternalServerError {
+ t.Fatalf("code = %d, want 500 — a child must learn its schedule did not land", rec.Code)
+ }
+}
+
+func TestAgentSockServeCleansUpItsSocket(t *testing.T) {
+ dir := t.TempDir()
+ path := filepath.Join(dir, "agent.sock")
+ ctx, cancel := context.WithCancel(context.Background())
+ done := make(chan error, 1)
+ go func() { done <- New(&fakeWorkers{}).Serve(ctx, path) }()
+ waitFor(t, "socket to appear", func() bool { _, err := os.Stat(path); return err == nil })
+ cancel()
+ if err := <-done; err != nil {
+ t.Fatalf("Serve returned %v on a clean shutdown", err)
+ }
+ if _, err := os.Stat(path); !os.IsNotExist(err) {
+ t.Fatal("a stale socket file must not survive shutdown — the next start would fail to bind")
+ }
+}
+
+// A hard kill leaves the socket file behind. The next start must take it over,
+// not die on "address already in use".
+func TestAgentSockServeReplacesAStaleSocketFile(t *testing.T) {
+ dir := t.TempDir()
+ path := filepath.Join(dir, "agent.sock")
+ if err := os.WriteFile(path, []byte("stale"), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+ done := make(chan error, 1)
+ go func() { done <- New(&fakeWorkers{}).Serve(ctx, path) }()
+ waitFor(t, "socket to appear", func() bool {
+ c, err := net.Dial("unix", path)
+ if err != nil {
+ return false
+ }
+ c.Close()
+ return true
+ })
+ cancel()
+ <-done
+}
+
+// The other half of stale-socket removal: a socket somebody is LISTENING on is
+// not stale. Unlinking it would orphan the live agent — its children would go
+// on talking to a process nothing else can reach — so refuse instead.
+func TestAgentSockServeRefusesALiveSocket(t *testing.T) {
+ dir := t.TempDir()
+ path := filepath.Join(dir, "agent.sock")
+ ln, err := net.Listen("unix", path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer ln.Close()
+ go func() {
+ for {
+ c, err := ln.Accept()
+ if err != nil {
+ return
+ }
+ c.Close()
+ }
+ }()
+
+ ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+ defer cancel()
+ if err := New(&fakeWorkers{}).Serve(ctx, path); err == nil {
+ t.Fatal("Serve must refuse a socket another listener already owns")
+ }
+ if _, err := os.Stat(path); err != nil {
+ t.Fatalf("the live socket was unlinked: %v", err)
+ }
+}
+
+// End to end over a real unix socket: the shape a sidecar actually dials.
+func TestAgentSockOverTheRealSocket(t *testing.T) {
+ dir := t.TempDir()
+ path := filepath.Join(dir, "agent.sock")
+ f := &fakeWorkers{tokens: map[string]string{"bot.x.internal": "good"}}
+ ctx, cancel := context.WithCancel(context.Background())
+ done := make(chan error, 1)
+ go func() { done <- New(f).Serve(ctx, path) }()
+ defer func() { cancel(); <-done }()
+ waitFor(t, "socket to appear", func() bool { _, err := os.Stat(path); return err == nil })
+
+ if fi, err := os.Stat(path); err != nil {
+ t.Fatal(err)
+ } else if perm := fi.Mode().Perm(); perm&0o007 != 0 {
+ t.Fatalf("socket mode %o is reachable by other — the token is the boundary, but the mode should not widen the audience", perm)
+ }
+
+ cl := &http.Client{
+ Transport: &http.Transport{
+ DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) {
+ return (&net.Dialer{}).DialContext(ctx, "unix", path)
+ },
+ },
+ Timeout: 5 * time.Second,
+ }
+ call := func(tok string) int {
+ req, err := http.NewRequest("POST", "http://agent/v1/heartbeat", nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ req.Header.Set("Carlos-Sidecar-Host", "bot.x.internal")
+ req.Header.Set("Authorization", "Bearer "+tok)
+ resp, err := cl.Do(req)
+ if err != nil {
+ t.Fatal(err)
+ }
+ resp.Body.Close()
+ return resp.StatusCode
+ }
+ if code := call("good"); code != http.StatusNoContent {
+ t.Fatalf("live heartbeat: %d", code)
+ }
+ if code := call("bad"); code != http.StatusUnauthorized {
+ t.Fatalf("live bad token: %d", code)
+ }
+ contacts, _, _, _ := f.effects()
+ if contacts != 1 {
+ t.Fatalf("contacts = %d, want exactly the authorized one", contacts)
+ }
+}
+
+// Many children hammering one agent at once — the handler holds no per-request
+// state, and -race proves it.
+func TestAgentSockConcurrentCallers(t *testing.T) {
+ f := &fakeWorkers{tokens: map[string]string{"a.x.internal": "tok-a", "b.x.internal": "tok-b"}}
+ h := New(f).Handler()
+ var wg sync.WaitGroup
+ for i := range 50 {
+ wg.Add(1)
+ go func(i int) {
+ defer wg.Done()
+ if i%2 == 0 {
+ post(h, "/v1/heartbeat", "a.x.internal", "tok-a", "")
+ } else {
+ post(h, "/v1/heartbeat", "b.x.internal", "tok-a", "")
+ }
+ }(i)
+ }
+ wg.Wait()
+ contacts, _, _, _ := f.effects()
+ if contacts != 25 {
+ t.Fatalf("contacts = %d, want 25 — only a.x.internal's own token may stamp it", contacts)
+ }
+}