| 1 | // Package agentsock is the agent's own listening socket — the ONE channel |
| 2 | // between a sidecar and the platform, and it points inward: children dial |
| 3 | // the agent, the agent never dials a child. That is what lets a sidecar |
| 4 | // bind nothing, and what lets the identical artifact run off-platform |
| 5 | // behind NAT with no protocol difference. |
| 6 | // |
| 7 | // Every exec child runs as the agent's user, so the socket's file mode is |
| 8 | // not an identity boundary. Identity is the per-wake token the activator |
| 9 | // mints and passes in the child's environment (CARLOS_SIDECAR_TOKEN), |
| 10 | // scoped to one host for one wake: a token for A cannot speak about B, and |
| 11 | // a token from a previous wake matches nothing. |
| 12 | // |
| 13 | // # What the socket trusts |
| 14 | // |
| 15 | // Nothing the caller sends, except as material for the one check. The |
| 16 | // Carlos-Sidecar-Host header names a host, but naming it proves nothing — |
| 17 | // only holding that host's live token does, and every effect is then applied |
| 18 | // to the host the TOKEN authorized. No body carries a host field, and none |
| 19 | // ever should: that is the confused-deputy shape this design exists to |
| 20 | // exclude. Bodies are read under a 4KiB cap, parsed strictly, and refused on |
| 21 | // anything unexpected; a refused call — for any reason, at any stage — has no |
| 22 | // effect at all. |
| 23 | // |
| 24 | // Refusals are deliberately uniform. Wrong token, absent token, unknown host, |
| 25 | // and a host whose worker is asleep all produce one identical 401, so the |
| 26 | // socket cannot be used to enumerate hosts or probe which ones are awake. |
| 27 | package agentsock |
| 28 | |
| 29 | import ( |
| 30 | "context" |
| 31 | "crypto/rand" |
| 32 | "crypto/subtle" |
| 33 | "encoding/json" |
| 34 | "errors" |
| 35 | "fmt" |
| 36 | "io" |
| 37 | "log" |
| 38 | "net" |
| 39 | "net/http" |
| 40 | "os" |
| 41 | "strings" |
| 42 | "time" |
| 43 | ) |
| 44 | |
| 45 | // Workers is the activator's half of this conversation. It is declared here, |
| 46 | // not imported, so the socket stays testable and the direction stays plain: |
| 47 | // this package is told about workers, and never reaches for one. |
| 48 | // |
| 49 | // WorkerToken returns "" for a host that is unknown, asleep, or stopped. That |
| 50 | // is a REFUSAL, never "no token required" — see authorize. |
| 51 | type Workers interface { |
| 52 | WorkerToken(host string) string |
| 53 | NoteWorkerContact(host string) |
| 54 | SetWorkerNextDue(host string, at int64) error |
| 55 | SleepWorker(host string) |
| 56 | NoteWorkerReady(host, version string) |
| 57 | } |
| 58 | |
| 59 | const ( |
| 60 | // hostHeader names the host a caller claims to be. Untrusted on its own. |
| 61 | hostHeader = "Carlos-Sidecar-Host" |
| 62 | |
| 63 | // maxBody bounds every body we read, following the next-due decode in |
| 64 | // activator/backend_exec.go. These messages are two small fields; a |
| 65 | // caller sending more is malformed, not generous. |
| 66 | maxBody = 1 << 12 |
| 67 | |
| 68 | // maxVersion bounds the build string a child reports. It is displayed to |
| 69 | // members, so it is length- and character-checked before it is recorded. |
| 70 | maxVersion = 200 |
| 71 | |
| 72 | // socketMode keeps the socket off the world. It is defence in depth, not |
| 73 | // the boundary: exec children share the agent's uid, so the mode cannot |
| 74 | // tell one child from another. The token does that. |
| 75 | socketMode = 0o660 |
| 76 | |
| 77 | // Local calls, all of them. Nothing here should ever be slow, and a |
| 78 | // child that stalls mid-request must not pin a connection. |
| 79 | readHeaderTimeout = 2 * time.Second |
| 80 | readTimeout = 5 * time.Second |
| 81 | writeTimeout = 5 * time.Second |
| 82 | idleTimeout = 30 * time.Second |
| 83 | shutdownGrace = 3 * time.Second |
| 84 | |
| 85 | // dialProbe is how long we wait to learn whether an existing socket file |
| 86 | // still has somebody listening on it. |
| 87 | dialProbe = 250 * time.Millisecond |
| 88 | ) |
| 89 | |
| 90 | // Server serves the agent socket. Zero per-request state: one instance |
| 91 | // answers every child on the box concurrently. |
| 92 | type Server struct { |
| 93 | w Workers |
| 94 | } |
| 95 | |
| 96 | // New builds the socket server over w. |
| 97 | func New(w Workers) *Server { return &Server{w: w} } |
| 98 | |
| 99 | // Handler is the routing table, exported so tests can exercise it without |
| 100 | // binding anything. Every verb is POST-only: the method pattern means a GET |
| 101 | // is turned away by the router, before authorize and before any effect. |
| 102 | func (s *Server) Handler() http.Handler { |
| 103 | mux := http.NewServeMux() |
| 104 | mux.Handle("POST /v1/ready", s.guard(s.ready)) |
| 105 | mux.Handle("POST /v1/next-due", s.guard(s.nextDue)) |
| 106 | mux.Handle("POST /v1/idle", s.guard(s.idle)) |
| 107 | mux.Handle("POST /v1/heartbeat", s.guard(s.heartbeat)) |
| 108 | return mux |
| 109 | } |
| 110 | |
| 111 | // guard is the single gate. Nothing reaches a verb without passing it, and a |
| 112 | // verb never sees the request's claimed host — only the authorized one, so |
| 113 | // there is no way for a handler to act on a host it was not entitled to. |
| 114 | func (s *Server) guard(fn func(http.ResponseWriter, *http.Request, string)) http.Handler { |
| 115 | return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { |
| 116 | host, ok := s.authorize(r) |
| 117 | if !ok { |
| 118 | refuse(w) |
| 119 | return |
| 120 | } |
| 121 | fn(w, r, host) |
| 122 | }) |
| 123 | } |
| 124 | |
| 125 | // authorize resolves the caller to the one host it may speak about. |
| 126 | // |
| 127 | // The dangerous shape here is WorkerToken returning "" — unknown host, asleep, |
| 128 | // stopped — being read as "this host needs no token", which would be a total |
| 129 | // bypass for every host that is not currently awake. So the empty case is not |
| 130 | // handled by a guard that a later edit could delete: an absent token is |
| 131 | // replaced with fresh random bytes, and the comparison then simply cannot |
| 132 | // succeed. Refusal is a property of the compare, not of a branch above it. |
| 133 | // |
| 134 | // A caller that names no host at all is refused here, not left to WorkerToken. |
| 135 | // Asking a collaborator what "" means is the same mistake in a different |
| 136 | // place: this package would be safe only for as long as the activator's map |
| 137 | // happens to hold no "" key, which is not a property this package can see or |
| 138 | // keep. An unnamed caller has nothing to authorize, so it never reaches the |
| 139 | // lookup — and gets the same 401 as every other refusal. |
| 140 | // |
| 141 | // The compare itself is constant-time, and both failure paths — no live token, |
| 142 | // wrong live token — run the same code, so the structure carries no timing |
| 143 | // signal beyond the length leak inherent to any comparison. |
| 144 | func (s *Server) authorize(r *http.Request) (string, bool) { |
| 145 | host := r.Header.Get(hostHeader) |
| 146 | presented := bearer(r.Header.Get("Authorization")) |
| 147 | |
| 148 | want := "" |
| 149 | if host != "" { |
| 150 | want = s.w.WorkerToken(host) |
| 151 | } |
| 152 | if want == "" { |
| 153 | want = unmatchable() |
| 154 | } |
| 155 | return host, subtle.ConstantTimeCompare([]byte(presented), []byte(want)) == 1 |
| 156 | } |
| 157 | |
| 158 | // unmatchable returns a value no caller can present. crypto/rand.Read never |
| 159 | // fails (it panics on a broken system rather than returning short), so this |
| 160 | // has no error path that could degrade into a guessable constant. |
| 161 | func unmatchable() string { |
| 162 | var b [32]byte |
| 163 | rand.Read(b[:]) |
| 164 | return string(b[:]) |
| 165 | } |
| 166 | |
| 167 | // bearer extracts the credential from an Authorization header, returning "" |
| 168 | // for anything that is not exactly a non-empty Bearer credential. "Bearer " |
| 169 | // with nothing after it yields "", which authorize then cannot match. |
| 170 | func bearer(h string) string { |
| 171 | const scheme = "Bearer " |
| 172 | if len(h) > len(scheme) && strings.EqualFold(h[:len(scheme)], scheme) { |
| 173 | return h[len(scheme):] |
| 174 | } |
| 175 | return "" |
| 176 | } |
| 177 | |
| 178 | // refuse is the only 401 in the package. One status, one body, for every |
| 179 | // reason — a caller learns that it was refused and nothing else. |
| 180 | func refuse(w http.ResponseWriter) { |
| 181 | http.Error(w, "unauthorized", http.StatusUnauthorized) |
| 182 | } |
| 183 | |
| 184 | // bad is the only 400 in the package, for the same reason. |
| 185 | func bad(w http.ResponseWriter) { |
| 186 | http.Error(w, "bad request", http.StatusBadRequest) |
| 187 | } |
| 188 | |
| 189 | // ready ends the settle window and records the build the child says it is |
| 190 | // running. The version is the child's own claim — §9.2's honesty rule — so it |
| 191 | // is bounded and character-checked, never trusted. The field must be PRESENT |
| 192 | // (a typo'd body is a loud 400, not a silent "unknown"), but may be empty: a |
| 193 | // child that genuinely does not know its build says so, and is still ready. |
| 194 | func (s *Server) ready(w http.ResponseWriter, r *http.Request, host string) { |
| 195 | var body struct { |
| 196 | Version *string `json:"version"` |
| 197 | } |
| 198 | if !decode(r, &body) { |
| 199 | bad(w) |
| 200 | return |
| 201 | } |
| 202 | if body.Version == nil || len(*body.Version) > maxVersion || !displayable(*body.Version) { |
| 203 | bad(w) |
| 204 | return |
| 205 | } |
| 206 | s.w.NoteWorkerReady(host, *body.Version) |
| 207 | w.WriteHeader(http.StatusNoContent) |
| 208 | } |
| 209 | |
| 210 | // nextDue records when the child expects to work again, while it is alive to |
| 211 | // say so. "at" must be present and non-negative: an absent field decoding to |
| 212 | // zero would silently reschedule the worker to the epoch. |
| 213 | func (s *Server) nextDue(w http.ResponseWriter, r *http.Request, host string) { |
| 214 | var body struct { |
| 215 | At *int64 `json:"at"` |
| 216 | } |
| 217 | if !decode(r, &body) { |
| 218 | bad(w) |
| 219 | return |
| 220 | } |
| 221 | if body.At == nil || *body.At < 0 { |
| 222 | bad(w) |
| 223 | return |
| 224 | } |
| 225 | if err := s.w.SetWorkerNextDue(host, *body.At); err != nil { |
| 226 | // The child must not read a 204 for a schedule that did not land. |
| 227 | log.Printf("agentsock: set next-due for %s: %v", host, err) |
| 228 | http.Error(w, "internal error", http.StatusInternalServerError) |
| 229 | return |
| 230 | } |
| 231 | w.WriteHeader(http.StatusNoContent) |
| 232 | } |
| 233 | |
| 234 | // idle is "I have nothing to do". It carries no arguments, so nothing in the |
| 235 | // request can steer it: the authorized host is the whole message. Whether the |
| 236 | // request is honoured at all is the activator's call — an always-on worker's |
| 237 | // idle is refused there. |
| 238 | func (s *Server) idle(w http.ResponseWriter, _ *http.Request, host string) { |
| 239 | s.w.SleepWorker(host) |
| 240 | w.WriteHeader(http.StatusNoContent) |
| 241 | } |
| 242 | |
| 243 | // heartbeat is a worker's only sign of life: nothing routes to a sidecar, so |
| 244 | // no request will ever stamp one. Argument-free for the same reason as idle. |
| 245 | func (s *Server) heartbeat(w http.ResponseWriter, _ *http.Request, host string) { |
| 246 | s.w.NoteWorkerContact(host) |
| 247 | w.WriteHeader(http.StatusNoContent) |
| 248 | } |
| 249 | |
| 250 | // decode parses a bounded body into v. The cap is applied to the READER, so an |
| 251 | // oversized body is truncated into a parse error rather than accepted: there is |
| 252 | // no size at which a caller gets a partial message honoured. |
| 253 | func decode(r *http.Request, v any) bool { |
| 254 | return json.NewDecoder(io.LimitReader(r.Body, maxBody)).Decode(v) == nil |
| 255 | } |
| 256 | |
| 257 | // displayable rejects control characters in a string bound for member-facing |
| 258 | // output — a version line should not be able to carry a NUL, a newline, or a |
| 259 | // terminal escape into a log or a console page. |
| 260 | func displayable(s string) bool { |
| 261 | for _, r := range s { |
| 262 | if r < 0x20 || r == 0x7f { |
| 263 | return false |
| 264 | } |
| 265 | } |
| 266 | return true |
| 267 | } |
| 268 | |
| 269 | // Serve binds path and answers on it until ctx is cancelled, then shuts down |
| 270 | // and leaves no socket file behind. |
| 271 | // |
| 272 | // A socket file that outlives its agent is the ordinary result of a hard kill, |
| 273 | // and the next start must take it over rather than die on "address already in |
| 274 | // use" — so a stale file is removed before binding. A file somebody is still |
| 275 | // LISTENING on is a different thing entirely: unlinking that would orphan a |
| 276 | // live agent, whose children would go on talking to a process nothing else can |
| 277 | // reach. So we probe first, and refuse rather than steal — and, for the same |
| 278 | // reason, this function unlinks nothing it did not create (see below). |
| 279 | func (s *Server) Serve(ctx context.Context, path string) error { |
| 280 | if listening(path) { |
| 281 | return fmt.Errorf("agentsock: %s is already served by a live listener", path) |
| 282 | } |
| 283 | if err := os.Remove(path); err != nil && !errors.Is(err, os.ErrNotExist) { |
| 284 | return fmt.Errorf("agentsock: remove stale socket %s: %w", path, err) |
| 285 | } |
| 286 | ln, err := net.Listen("unix", path) |
| 287 | if err != nil { |
| 288 | return fmt.Errorf("agentsock: listen on %s: %w", path, err) |
| 289 | } |
| 290 | // Cleanup is the LISTENER's job and nothing else's. Go unlinks a unix |
| 291 | // socket it created when the listener closes, and that unlink happens |
| 292 | // inside Close — indivisibly, from the point of view of a successor |
| 293 | // trying to bind the same path. |
| 294 | // |
| 295 | // A deferred os.Remove(path) here would not be. Between the listener's |
| 296 | // own unlink and the deferred call, a successor agent can have bound the |
| 297 | // path, and we would then unlink ITS socket: the successor serves on an |
| 298 | // inode nothing can reach, its children never arrive, and the box has the |
| 299 | // same split-brain the probe above exists to refuse to cause. Guarding |
| 300 | // one door and leaving the other open is worse than guarding neither, |
| 301 | // because it reads as guarded. So there is no unconditional remove, and |
| 302 | // every return below closes the listener instead. |
| 303 | if ul, ok := ln.(*net.UnixListener); ok { |
| 304 | ul.SetUnlinkOnClose(true) // the default; pinned because it is load-bearing |
| 305 | } |
| 306 | |
| 307 | if err := os.Chmod(path, socketMode); err != nil { |
| 308 | ln.Close() // unlinks the socket we just created, and only that one |
| 309 | return fmt.Errorf("agentsock: chmod %s: %w", path, err) |
| 310 | } |
| 311 | |
| 312 | srv := &http.Server{ |
| 313 | Handler: s.Handler(), |
| 314 | ReadHeaderTimeout: readHeaderTimeout, |
| 315 | ReadTimeout: readTimeout, |
| 316 | WriteTimeout: writeTimeout, |
| 317 | IdleTimeout: idleTimeout, |
| 318 | MaxHeaderBytes: 1 << 14, |
| 319 | } |
| 320 | |
| 321 | errc := make(chan error, 1) |
| 322 | go func() { errc <- srv.Serve(ln) }() |
| 323 | |
| 324 | select { |
| 325 | case err := <-errc: |
| 326 | if errors.Is(err, http.ErrServerClosed) { |
| 327 | return nil |
| 328 | } |
| 329 | return err |
| 330 | case <-ctx.Done(): |
| 331 | // ctx is already cancelled, so Shutdown gets its own deadline. |
| 332 | sctx, cancel := context.WithTimeout(context.Background(), shutdownGrace) |
| 333 | defer cancel() |
| 334 | err := srv.Shutdown(sctx) |
| 335 | <-errc |
| 336 | return err |
| 337 | } |
| 338 | } |
| 339 | |
| 340 | // listening reports whether something is currently accepting on path. A stale |
| 341 | // socket file, a regular file, and a missing path all answer false. |
| 342 | func listening(path string) bool { |
| 343 | c, err := net.DialTimeout("unix", path, dialProbe) |
| 344 | if err != nil { |
| 345 | return false |
| 346 | } |
| 347 | c.Close() |
| 348 | return true |
| 349 | } |
| 350 | |
amadan