| 1 | package console |
| 2 | |
| 3 | // The operator doors for platform GeoIP (geoip spec §4.2, §5.5): |
| 4 | // |
| 5 | // PUT /api/cli/system/geoip stage an upload, answer 202 + a job id |
| 6 | // GET /api/cli/system/geoip the pointer, every box, the rollout verdict |
| 7 | // DELETE /api/cli/system/geoip off: delete the pointer, sweep the blobs |
| 8 | // |
| 9 | // All three open with apiSystemRoller (an ops owner or CARLOS_SYSTEM_OPERATORS, |
| 10 | // a person, never a token) and are registered with the zero svcrole.Set. |
| 11 | |
| 12 | import ( |
| 13 | "context" |
| 14 | "errors" |
| 15 | "fmt" |
| 16 | "io" |
| 17 | "log" |
| 18 | "net/http" |
| 19 | "os" |
| 20 | "sort" |
| 21 | "strings" |
| 22 | "time" |
| 23 | |
| 24 | "github.com/carlosframework/platform/internal/geoip" |
| 25 | "github.com/carlosframework/platform/internal/release" |
| 26 | "github.com/carlosframework/platform/internal/status" |
| 27 | ) |
| 28 | |
| 29 | // geoipCommitDelay is a test seam: a job sleeps this long before Commit, so |
| 30 | // a test can prove the 202 arrives before verification finishes. |
| 31 | var geoipCommitDelay time.Duration |
| 32 | |
| 33 | // geoipRetiredAfter is how old a heartbeat may be, for a box with no edges |
| 34 | // record and no fleet registration, before it is left out as retired. |
| 35 | const geoipRetiredAfter = 7 * 24 * time.Hour |
| 36 | |
| 37 | func (s *Server) handleAPISystemGeoipPut(w http.ResponseWriter, r *http.Request) { |
| 38 | addr, _, ok := s.apiSystemRoller(w, r) |
| 39 | if !ok { |
| 40 | return |
| 41 | } |
| 42 | if !s.systemBucket(w) { |
| 43 | return |
| 44 | } |
| 45 | // The server-wide ReadTimeout is 5 minutes (cmd/console/main.go); a |
| 46 | // 1 GiB upload gets its own deadline, the way the ship door does it. |
| 47 | rc := http.NewResponseController(w) |
| 48 | _ = rc.SetReadDeadline(time.Now().Add(15 * time.Minute)) |
| 49 | defer rc.SetReadDeadline(time.Time{}) |
| 50 | body := http.MaxBytesReader(w, r.Body, geoip.MaxBlobBytes) |
| 51 | // Stage takes the one publish slot BEFORE reading any of the body |
| 52 | // (codex R5-1): a second upload is refused here, unread. |
| 53 | st, err := s.geoip.Stage(r.Context(), body) |
| 54 | var mbe *http.MaxBytesError |
| 55 | switch { |
| 56 | case errors.Is(err, errGeoipBusy): |
| 57 | jsonError(w, http.StatusConflict, "another geoip publish is in progress. try again when it finishes") |
| 58 | return |
| 59 | case errors.As(err, &mbe): |
| 60 | jsonError(w, http.StatusRequestEntityTooLarge, "the database is larger than 1 GiB") |
| 61 | return |
| 62 | case errors.Is(err, errGeoipRefused): |
| 63 | jsonError(w, http.StatusUnprocessableEntity, err.Error()) |
| 64 | return |
| 65 | case err != nil: |
| 66 | log.Printf("console: geoip upload: %v", err) |
| 67 | jsonError(w, http.StatusBadGateway, "could not receive the database") |
| 68 | return |
| 69 | } |
| 70 | opts := geoipPublishOpts{Origin: "manual", Force: r.URL.Query().Get("force") == "1"} |
| 71 | id, job := s.newSystemJob() |
| 72 | job.step("received %d bytes (sha256 %s)", st.size, st.sha[:8]) |
| 73 | // The job owns the slot and the staged directory from here; Commit |
| 74 | // releases both on every path. A fresh context: the request's dies with |
| 75 | // the 202, and a backhauled request has only 30s after its body to see |
| 76 | // headers (codex R4-2). |
| 77 | go func() { |
| 78 | ctx, cancel := context.WithTimeout(context.Background(), 15*time.Minute) |
| 79 | defer cancel() |
| 80 | if geoipCommitDelay > 0 { |
| 81 | time.Sleep(geoipCommitDelay) |
| 82 | } |
| 83 | job.step("verifying the database") |
| 84 | ptr, err := s.geoip.Commit(ctx, st, opts) |
| 85 | if err == nil { |
| 86 | job.setResult(ptr) |
| 87 | job.step("published %s: %s, built %s", ptr.SHA256[:8], ptr.DatabaseType, ptr.Build().Format("2006-01-02")) |
| 88 | s.auditSystem(addr, "geoip publish", ptr.SHA256[:8]+" "+ptr.DatabaseType) |
| 89 | } else if errors.Is(err, errGeoipRefused) { |
| 90 | log.Printf("console: geoip publish refused: %v", err) |
| 91 | } else { |
| 92 | log.Printf("console: geoip publish: %v", err) |
| 93 | } |
| 94 | job.finish(err) |
| 95 | }() |
| 96 | writeJSON(w, http.StatusAccepted, map[string]string{"job": id}) |
| 97 | } |
| 98 | |
| 99 | func (s *Server) handleAPISystemGeoipDelete(w http.ResponseWriter, r *http.Request) { |
| 100 | addr, _, ok := s.apiSystemRoller(w, r) |
| 101 | if !ok { |
| 102 | return |
| 103 | } |
| 104 | if !s.systemBucket(w) { |
| 105 | return |
| 106 | } |
| 107 | if err := s.geoip.Off(r.Context()); errors.Is(err, errGeoipBusy) { |
| 108 | jsonError(w, http.StatusConflict, "a geoip publish is in progress. run off again when it finishes") |
| 109 | return |
| 110 | } else if err != nil { |
| 111 | log.Printf("console: geoip off: %v", err) |
| 112 | jsonError(w, http.StatusBadGateway, "could not turn geoip off") |
| 113 | return |
| 114 | } |
| 115 | s.auditSystem(addr, "geoip off", "") |
| 116 | writeJSON(w, http.StatusOK, map[string]any{"off": true, "feed_set": os.Getenv("CONSOLE_GEOIP_FEED") != ""}) |
| 117 | } |
| 118 | |
| 119 | // geoipBoxRow is one box in the status answer. |
| 120 | type geoipBoxRow struct { |
| 121 | Label string `json:"label"` |
| 122 | Fleet string `json:"fleet,omitempty"` |
| 123 | State string `json:"state"` // ok | stale | old binary | no heartbeat | other sha | tunnel | none |
| 124 | LastSeen *time.Time `json:"last_seen,omitempty"` |
| 125 | GeoIP *status.GeoIPState `json:"geoip,omitempty"` |
| 126 | Expected bool `json:"expected"` |
| 127 | Blocking string `json:"blocking,omitempty"` |
| 128 | } |
| 129 | |
| 130 | func (s *Server) handleAPISystemGeoipGet(w http.ResponseWriter, r *http.Request) { |
| 131 | if _, _, ok := s.apiSystemRoller(w, r); !ok { |
| 132 | return |
| 133 | } |
| 134 | if !s.systemBucket(w) { |
| 135 | return |
| 136 | } |
| 137 | ctx := r.Context() |
| 138 | cur, _, present, perr := s.geoip.Current(ctx) |
| 139 | out := map[string]any{"pointer": nil} |
| 140 | want := geoipWant{state: geoipOff} |
| 141 | switch { |
| 142 | case perr != nil: |
| 143 | out["pointer_error"] = "the published pointer is unreadable" |
| 144 | want.state = geoipUnknown |
| 145 | case present && (cur.Expired || geoip.Expired(cur.DatabaseType, cur.Build(), time.Now())): |
| 146 | // Expired, marked or not yet marked by the daily job: every edge |
| 147 | // drops it by the clock, so every mirror must be empty (codex R10 |
| 148 | // on R9-1). |
| 149 | want.state = geoipOff |
| 150 | case present: |
| 151 | want = geoipWant{state: geoipLive, sha: cur.SHA256} |
| 152 | } |
| 153 | if perr == nil && present { |
| 154 | p := map[string]any{"sha256": cur.SHA256, "database_type": cur.DatabaseType, "build": cur.Build(), |
| 155 | "published": cur.Published, "origin": cur.Origin, "edition": cur.Edition, "size": cur.Size, "expired": cur.Expired} |
| 156 | if at, ok := geoip.ExpiresAt(cur.DatabaseType, cur.Build()); ok { |
| 157 | p["expires_at"] = at |
| 158 | } |
| 159 | if a, ok := geoip.AttributionFor(cur.DatabaseType); ok { |
| 160 | p["attribution"] = a.Text |
| 161 | } |
| 162 | out["pointer"] = p |
| 163 | } |
| 164 | inv, err := s.geoipInventory(ctx) |
| 165 | if err != nil { |
| 166 | log.Printf("console: geoip status inventory: %v", err) |
| 167 | jsonError(w, http.StatusBadGateway, "box heartbeats unreadable") |
| 168 | return |
| 169 | } |
| 170 | ready, rows, blocking := geoipVerdict(want, inv, time.Now()) |
| 171 | out["boxes"], out["ready"], out["blocking"] = rows, ready, blocking |
| 172 | out["feed_set"] = os.Getenv("CONSOLE_GEOIP_FEED") != "" |
| 173 | if raw, _, err := release.OpenObject(ctx, s.cfg.Releases, geoip.FeedKey); err == nil { |
| 174 | b, rerr := io.ReadAll(io.LimitReader(raw, geoip.MaxPointerBytes+1)) |
| 175 | raw.Close() |
| 176 | if rerr == nil && len(b) <= geoip.MaxPointerBytes { |
| 177 | if fs, err := geoip.ParseFeedStatus(b); err == nil { |
| 178 | out["feed"] = map[string]any{"name": fs.Feed, "last_check": fs.LastCheck, "last_success": fs.LastSuccess, |
| 179 | "edition": fs.LastEdition, "failures": fs.ConsecutiveFailures, "not_found": fs.ConsecutiveNotFound, |
| 180 | "last_failure": fs.LastFailure, "failing": fs.Failing()} |
| 181 | } |
| 182 | } |
| 183 | } |
| 184 | writeJSON(w, http.StatusOK, out) |
| 185 | } |
| 186 | |
| 187 | // geoipBoxKey is a box's identity: a label is unique only within its |
| 188 | // namespace, the platform ("") or one fleet (codex R8-1). Two fleets may |
| 189 | // both have node-1, and a fleet label may equal a platform label. |
| 190 | type geoipBoxKey struct{ Fleet, Label string } |
| 191 | |
| 192 | func (k geoipBoxKey) String() string { |
| 193 | if k.Fleet == "" { |
| 194 | return k.Label |
| 195 | } |
| 196 | return k.Fleet + "/" + k.Label |
| 197 | } |
| 198 | |
| 199 | // geoipInventoryRead is everything the verdict reconciles (spec §5.5). |
| 200 | // expectedComplete is false when the list of expected boxes itself could |
| 201 | // not be read (a fleet registry or fleet listing failed): that always |
| 202 | // blocks. heartbeatsComplete is false when some heartbeat record could not |
| 203 | // be read: informational only, because an expected box whose record was |
| 204 | // unreadable is missing from boxes and blocks as "no heartbeat" on its own |
| 205 | // (codex R9 on R1-16). |
| 206 | // |
| 207 | // fleetsListed is true only once the fleet registry was opened and its |
| 208 | // fleets listed; fleetsUnread says why not, for the operator log. The GeoIP |
| 209 | // verdict never reads it: it treats an unset or absent registry as a |
| 210 | // platform-only deployment, as it always has. The update settings gate |
| 211 | // (updatesEnforced) does read it, because a fleet list it never saw is not |
| 212 | // evidence that there are no fleet boxes to wait for (spec §4.4). |
| 213 | type geoipInventoryRead struct { |
| 214 | expected map[geoipBoxKey]bool |
| 215 | boxes map[geoipBoxKey]status.Box |
| 216 | expectedComplete bool |
| 217 | heartbeatsComplete bool |
| 218 | fleetsListed bool |
| 219 | fleetsUnread string |
| 220 | } |
| 221 | |
| 222 | // geoipWantState is what the published pointer asks of every mirror. |
| 223 | type geoipWantState int |
| 224 | |
| 225 | const ( |
| 226 | geoipUnknown geoipWantState = iota // the pointer could not be read: never ready |
| 227 | geoipOff // no pointer, or an expired one: mirrors must hold nothing |
| 228 | geoipLive // mirrors must hold sha |
| 229 | ) |
| 230 | |
| 231 | type geoipWant struct { |
| 232 | state geoipWantState |
| 233 | sha string |
| 234 | } |
| 235 | |
| 236 | // geoipInventory reads the expected boxes (platform edges records, active |
| 237 | // fleet registrations) and every heartbeat prefix with LoadBoxesComplete, |
| 238 | // never LoadBoxes, which skips unreadable records silently. |
| 239 | func (s *Server) geoipInventory(ctx context.Context) (geoipInventoryRead, error) { |
| 240 | inv := geoipInventoryRead{expected: map[geoipBoxKey]bool{}, boxes: map[geoipBoxKey]status.Box{}, expectedComplete: true, heartbeatsComplete: true} |
| 241 | keys, err := s.cfg.Releases.List(ctx, "control/edges/") |
| 242 | if err != nil { |
| 243 | return inv, fmt.Errorf("edges records: %w", err) |
| 244 | } |
| 245 | for _, k := range keys { |
| 246 | label := strings.TrimSuffix(strings.TrimPrefix(k, "control/edges/"), ".json") |
| 247 | if label != "" && !strings.Contains(label, "/") && strings.HasSuffix(k, ".json") { |
| 248 | inv.expected[geoipBoxKey{Label: label}] = true |
| 249 | } |
| 250 | } |
| 251 | add := func(store release.Store, fleet string) error { |
| 252 | boxes, complete, err := status.LoadBoxesComplete(ctx, store) |
| 253 | if err != nil { |
| 254 | return err |
| 255 | } |
| 256 | if !complete { |
| 257 | inv.heartbeatsComplete = false |
| 258 | } |
| 259 | for _, b := range boxes { |
| 260 | inv.boxes[geoipBoxKey{fleet, b.Label}] = b |
| 261 | } |
| 262 | return nil |
| 263 | } |
| 264 | if err := add(s.cfg.Releases, ""); err != nil { |
| 265 | return inv, err |
| 266 | } |
| 267 | if s.cfg.Registry == "" { |
| 268 | inv.fleetsUnread = "no registry path is configured" |
| 269 | return inv, nil |
| 270 | } |
| 271 | reg, err := s.openPassRegistry() |
| 272 | switch { |
| 273 | case errors.Is(err, os.ErrNotExist): |
| 274 | inv.fleetsUnread = err.Error() |
| 275 | return inv, nil |
| 276 | case err != nil: |
| 277 | inv.expectedComplete = false |
| 278 | inv.fleetsUnread = err.Error() |
| 279 | return inv, nil |
| 280 | } |
| 281 | fleets, err := reg.ListFleets() |
| 282 | if err != nil { |
| 283 | inv.expectedComplete = false |
| 284 | inv.fleetsUnread = "listing fleets: " + err.Error() |
| 285 | return inv, nil |
| 286 | } |
| 287 | inv.fleetsListed = true |
| 288 | for _, f := range fleets { |
| 289 | fbs, err := reg.ListFleetBoxes(f.Name) |
| 290 | if err != nil { |
| 291 | inv.expectedComplete = false |
| 292 | continue |
| 293 | } |
| 294 | for _, fb := range fbs { |
| 295 | if fb.Status == "active" { |
| 296 | inv.expected[geoipBoxKey{f.Name, fb.Label}] = true |
| 297 | } |
| 298 | } |
| 299 | if err := add(release.NewPrefixStore(s.cfg.Releases, "fleets/"+f.Name+"/"), f.Name); err != nil { |
| 300 | // That fleet's heartbeats are unknown: its expected boxes are |
| 301 | // missing from boxes and block as "no heartbeat". |
| 302 | inv.heartbeatsComplete = false |
| 303 | } |
| 304 | } |
| 305 | return inv, nil |
| 306 | } |
| 307 | |
| 308 | // geoipVerdict is spec §5.5's reconciliation. ready only when every listed |
| 309 | // box is fresh, readable and GeoIP-aware, every mirroring box holds the |
| 310 | // live pointer's sha, and every read was complete. |
| 311 | func geoipVerdict(want geoipWant, inv geoipInventoryRead, now time.Time) (bool, []geoipBoxRow, []string) { |
| 312 | var rows []geoipBoxRow |
| 313 | var blocking []string |
| 314 | keys := map[geoipBoxKey]bool{} |
| 315 | for k := range inv.expected { |
| 316 | keys[k] = true |
| 317 | } |
| 318 | for k, b := range inv.boxes { |
| 319 | if inv.expected[k] || now.Sub(b.GeneratedAt) <= geoipRetiredAfter { |
| 320 | keys[k] = true |
| 321 | } |
| 322 | } |
| 323 | sorted := make([]geoipBoxKey, 0, len(keys)) |
| 324 | for k := range keys { |
| 325 | sorted = append(sorted, k) |
| 326 | } |
| 327 | sort.Slice(sorted, func(i, j int) bool { return sorted[i].String() < sorted[j].String() }) |
| 328 | for _, k := range sorted { |
| 329 | name, expected := k.String(), inv.expected[k] |
| 330 | row := geoipBoxRow{Label: k.Label, Fleet: k.Fleet, Expected: expected} |
| 331 | b, seen := inv.boxes[k] |
| 332 | retire := ". if this box is retired, delete its edges record (control/edges/" + k.Label + ".json)" |
| 333 | if k.Fleet != "" { |
| 334 | retire = ". if this box is retired, detach it from fleet " + k.Fleet |
| 335 | } |
| 336 | switch { |
| 337 | case !seen: |
| 338 | row.State = "no heartbeat" |
| 339 | row.Blocking = name + ": no heartbeat" + retire |
| 340 | case status.BoxStale(b, now, status.SnapshotPeriod): |
| 341 | t := b.GeneratedAt |
| 342 | row.LastSeen = &t |
| 343 | row.State = "stale" |
| 344 | row.Blocking = name + ": no heartbeat since " + t.UTC().Format(time.RFC3339) + retire |
| 345 | case b.Metrics == nil || b.Metrics.GeoIP == nil: |
| 346 | t := b.GeneratedAt |
| 347 | row.LastSeen = &t |
| 348 | row.State = "old binary" |
| 349 | row.Blocking = name + ": runs a binary that does not strip forged geo headers" |
| 350 | default: |
| 351 | t := b.GeneratedAt |
| 352 | row.LastSeen, row.GeoIP = &t, b.Metrics.GeoIP |
| 353 | switch g := b.Metrics.GeoIP; { |
| 354 | case g.Mode == "tunnel", g.Mode == "none": |
| 355 | row.State = g.Mode |
| 356 | case want.state == geoipLive && g.SHA256 != want.sha: |
| 357 | row.State = "other sha" |
| 358 | row.Blocking = name + ": has " + short8(g.SHA256) + ", not the published " + short8(want.sha) |
| 359 | case want.state == geoipOff && g.SHA256 != "": |
| 360 | // Off or expired: a mirror still serving a generation has |
| 361 | // not caught up (codex R9-1). |
| 362 | row.State = "still loaded" |
| 363 | row.Blocking = name + ": still has " + short8(g.SHA256) + " loaded" |
| 364 | default: |
| 365 | row.State = "ok" |
| 366 | } |
| 367 | } |
| 368 | // Only an EXPECTED box (an edges record, an active fleet |
| 369 | // registration) can hold the verdict up: deleting a retired box's |
| 370 | // edges record unblocks at once, though its last heartbeat stays |
| 371 | // listed for information (codex R8 on R1-16). |
| 372 | if !expected { |
| 373 | row.Blocking = "" |
| 374 | } |
| 375 | if row.Blocking != "" { |
| 376 | blocking = append(blocking, row.Blocking) |
| 377 | } |
| 378 | rows = append(rows, row) |
| 379 | } |
| 380 | if want.state == geoipUnknown { |
| 381 | blocking = append(blocking, "database: the published pointer could not be read") |
| 382 | } |
| 383 | if !inv.expectedComplete { |
| 384 | blocking = append(blocking, "boxes: the list of fleet boxes could not be read") |
| 385 | } |
| 386 | if rows == nil { |
| 387 | rows = []geoipBoxRow{} |
| 388 | } |
| 389 | if blocking == nil { |
| 390 | blocking = []string{} |
| 391 | } |
| 392 | return len(blocking) == 0, rows, blocking |
| 393 | } |
| 394 | |
| 395 | func short8(sha string) string { |
| 396 | if sha == "" { |
| 397 | return "nothing" |
| 398 | } |
| 399 | if len(sha) > 8 { |
| 400 | return sha[:8] |
| 401 | } |
| 402 | return sha |
| 403 | } |
| 404 | |
amadan