amadan

carlos / platform Public

cli: carlos system geoip publish/status/off (raw streamed upload, job follow, verdict) (geoip spec §4.2, §5.5)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 1dbb4c9de6545168ab9eb0327ef4657b75548cce parent ca3f353
3 files changed, +412 −1
  • system.go +12 −1
  • system_geoip.go +272 −0
  • system_geoip_test.go +128 −0
diff --git a/system.go b/system.go
index 8253c34ab..f1d2443c3 100644
--- a/system.go
+++ b/system.go
@@ -47,7 +47,7 @@ func systemVerb(args []string) (verb string, rest []string, component string) {
switch args[0] {
// antivirus is a FAMILY, not a leaf: its own sub-verbs come back
// untouched in rest for runSystemAntivirusVerb to read.
- case "update", "status", "rollback", "apply", "antivirus", "help", "-h", "--help":
+ case "update", "status", "rollback", "apply", "antivirus", "geoip", "help", "-h", "--help":
return args[0], args[1:], component
}
return "", args, component
@@ -66,6 +66,8 @@ func runSystem(args []string) {
runSystemRollback(rest)
case "antivirus":
runSystemAntivirusVerb(rest)
+ case "geoip":
+ runSystemGeoipVerb(rest)
case "help", "-h", "--help":
systemUsage()
default:
@@ -97,6 +99,7 @@ func systemUsage() {
fmt.Fprint(os.Stderr, `usage: carlos system <verb> [flags]
carlos system router <verb> [flags]
carlos system antivirus <status|enable|disable> [flags]
+ carlos system geoip <publish|status|off> [flags]
The platform's own binary (ops account owners and CARLOS_SYSTEM_OPERATORS):
@@ -118,6 +121,14 @@ Deployment-wide settings (CARLOS_STORE_OPERATORS):
which is the point of the switch living in a record
rather than in the unit file.
+Platform GeoIP (ops account owners and CARLOS_SYSTEM_OPERATORS):
+
+ geoip publish <file.mmdb> publish a MaxMind-format database (DB-IP Lite or
+ GeoLite2) to every edge. -force skips the shrink check
+ geoip status the published database, and whether every box
+ strips forged headers and has it
+ geoip off stop: edges drop the database and send no geo headers
+
Enabling antivirus scans nothing by itself: a member still requests scanning
for their app, and an operator still grants it per store.
diff --git a/system_geoip.go b/system_geoip.go
new file mode 100644
index 000000000..d175c7aef
--- /dev/null
+++ b/system_geoip.go
@@ -0,0 +1,272 @@
+package main
+
+// carlos system geoip publish|status|off (geoip spec §4.2, §5.5). The
+// strings here await Paul's copy review (plan 1, Task 13) before merge.
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "flag"
+ "fmt"
+ "io"
+ "log"
+ "net/http"
+ "os"
+ "strings"
+ "time"
+)
+
+func runSystemGeoipVerb(rest []string) {
+ if len(rest) == 0 {
+ fmt.Fprint(os.Stderr, "system geoip: no verb given\n\n")
+ systemUsage()
+ os.Exit(2)
+ }
+ switch rest[0] {
+ case "publish":
+ runSystemGeoipPublish(rest[1:])
+ case "status":
+ runSystemGeoipStatus(rest[1:])
+ case "off":
+ runSystemGeoipOff(rest[1:])
+ default:
+ fmt.Fprintf(os.Stderr, "system geoip: unknown verb %q\n\n", rest[0])
+ systemUsage()
+ os.Exit(2)
+ }
+}
+
+func runSystemGeoipPublish(args []string) {
+ fs := flag.NewFlagSet("system geoip publish", flag.ExitOnError)
+ consoleFlag := consoleOption(fs, "console origin (default: the only logged-in one)")
+ force := fs.Bool("force", false, "publish even if the database has under half the current one's nodes")
+ fs.Usage = flagUsage(fs, "<file.mmdb>")
+ fs.Parse(args)
+ if fs.NArg() != 1 {
+ fs.Usage()
+ os.Exit(2)
+ }
+ sess := requireDeployAPISession(*consoleFlag)
+ printUsing(stdout, sess)
+ ptr, err := sess.publishGeoip(context.Background(), fs.Arg(0), *force, stdout)
+ if err != nil {
+ log.Fatalf("carlos system geoip publish: %v", err)
+ }
+ fmt.Fprintf(stdout, "%s %s: %s, built %s. every edge picks it up within 11 minutes\n",
+ colorFor(stdout).ok("published"), short8cli(ptr.SHA256), ptr.DatabaseType, time.Unix(ptr.BuildEpoch, 0).UTC().Format("2006-01-02"))
+}
+
+// geoipPointerView is the pointer as the CLI reads it back from a job.
+type geoipPointerView struct {
+ SHA256 string `json:"sha256"`
+ DatabaseType string `json:"database_type"`
+ BuildEpoch int64 `json:"build_epoch"`
+}
+
+// publishGeoip streams the file in one raw PUT (the console stages it and
+// answers 202 with a job), then follows the job to its result.
+func (s *apiSession) publishGeoip(ctx context.Context, path string, force bool, w io.Writer) (geoipPointerView, error) {
+ f, err := os.Open(path)
+ if err != nil {
+ return geoipPointerView{}, err
+ }
+ defer f.Close()
+ fi, err := f.Stat()
+ if err != nil {
+ return geoipPointerView{}, err
+ }
+ q := ""
+ if force {
+ q = "?force=1"
+ }
+ req, err := http.NewRequestWithContext(ctx, http.MethodPut, s.c.base+"/api/cli/system/geoip"+q, f)
+ if err != nil {
+ return geoipPointerView{}, err
+ }
+ req.ContentLength = fi.Size()
+ req.Header.Set("Content-Type", "application/octet-stream")
+ if s.c.token != "" {
+ req.Header.Set("Authorization", "Bearer "+s.c.token)
+ }
+ // The upload's own client, like doMultipartMethod's: c.httpc's 30s
+ // absolute timeout is sized for JSON, not a 130 MB database.
+ client := &http.Client{Timeout: shipHTTPClientTimeout, CheckRedirect: neverFollow}
+ res, err := client.Do(req)
+ if err != nil {
+ return geoipPointerView{}, err
+ }
+ defer res.Body.Close()
+ if merr := consoleMoved(s.c.base, res.StatusCode, res.Header.Get("Location")); merr != nil {
+ return geoipPointerView{}, offerConsoleMoved(merr)
+ }
+ body, _ := io.ReadAll(io.LimitReader(res.Body, 1<<20))
+ if res.StatusCode != http.StatusAccepted {
+ return geoipPointerView{}, apiErrorFromBody(res.StatusCode, body)
+ }
+ var started struct {
+ Job string `json:"job"`
+ }
+ if err := json.Unmarshal(body, &started); err != nil || started.Job == "" {
+ return geoipPointerView{}, errors.New("the console accepted the upload but named no job")
+ }
+ return s.watchGeoipJob(ctx, started.Job, w)
+}
+
+func (s *apiSession) watchGeoipJob(ctx context.Context, id string, w io.Writer) (geoipPointerView, error) {
+ printed := 0
+ deadline := time.Now().Add(16 * time.Minute)
+ for {
+ st, body, err := s.getJSON(ctx, "/api/cli/system/jobs/"+id)
+ if err != nil {
+ return geoipPointerView{}, fmt.Errorf("job %s: %v", id, err)
+ }
+ if st != http.StatusOK {
+ return geoipPointerView{}, fmt.Errorf("job %s: %s", id, apiErrorFromBody(st, body))
+ }
+ var j struct {
+ Steps []string `json:"steps"`
+ Done bool `json:"done"`
+ Error string `json:"error"`
+ Result *geoipPointerView `json:"result"`
+ }
+ _ = json.Unmarshal(body, &j)
+ for ; printed < len(j.Steps); printed++ {
+ fmt.Fprintf(w, " %s\n", j.Steps[printed])
+ }
+ if j.Done {
+ if j.Error != "" {
+ return geoipPointerView{}, errors.New(j.Error)
+ }
+ if j.Result == nil {
+ return geoipPointerView{}, errors.New("the job finished without naming what it published")
+ }
+ return *j.Result, nil
+ }
+ if time.Now().After(deadline) {
+ return geoipPointerView{}, errors.New("the console's job did not finish in 16 minutes")
+ }
+ time.Sleep(geoipPollInterval)
+ }
+}
+
+// geoipPollInterval is a var so a test need not wait seconds per poll.
+var geoipPollInterval = 2 * time.Second
+
+type geoipStatusResponse struct {
+ Pointer *struct {
+ SHA256 string `json:"sha256"`
+ DatabaseType string `json:"database_type"`
+ Build time.Time `json:"build"`
+ Published time.Time `json:"published"`
+ Origin string `json:"origin"`
+ Expired bool `json:"expired"`
+ ExpiresAt time.Time `json:"expires_at"`
+ Attribution string `json:"attribution"`
+ } `json:"pointer"`
+ PointerError string `json:"pointer_error"`
+ Boxes []struct {
+ Label string `json:"label"`
+ Fleet string `json:"fleet"`
+ State string `json:"state"`
+ } `json:"boxes"`
+ Ready bool `json:"ready"`
+ Blocking []string `json:"blocking"`
+ FeedSet bool `json:"feed_set"`
+}
+
+func runSystemGeoipStatus(args []string) {
+ fs := flag.NewFlagSet("system geoip status", flag.ExitOnError)
+ consoleFlag := consoleOption(fs, "console origin (default: the only logged-in one)")
+ fs.Usage = flagUsage(fs, "")
+ fs.Parse(args)
+ sess := requireDeployAPISession(*consoleFlag)
+ st, body, err := sess.getJSON(context.Background(), "/api/cli/system/geoip")
+ if err != nil {
+ log.Fatalf("carlos system geoip status: %v", err)
+ }
+ if st != http.StatusOK {
+ log.Fatalf("carlos system geoip status: %s", apiErrorFromBody(st, body))
+ }
+ var resp geoipStatusResponse
+ if err := json.Unmarshal(body, &resp); err != nil {
+ log.Fatalf("carlos system geoip status: unreadable response: %v", err)
+ }
+ printGeoipStatus(stdout, resp, time.Now())
+}
+
+// printGeoipStatus is the verb minus the fetch, so it is testable.
+func printGeoipStatus(w io.Writer, resp geoipStatusResponse, now time.Time) {
+ c := colorFor(w)
+ switch p := resp.Pointer; {
+ case resp.PointerError != "":
+ fmt.Fprintf(w, "database: %s\n", c.bad(resp.PointerError))
+ case p == nil:
+ fmt.Fprintln(w, "database: none published. geoip is off")
+ case p.Expired:
+ fmt.Fprintf(w, "database: %s %s expired. publish a newer one or turn geoip off\n", short8cli(p.SHA256), p.DatabaseType)
+ default:
+ age := int(now.Sub(p.Build).Hours() / 24)
+ fmt.Fprintf(w, "database: %s %s, built %s (%d days ago), %s\n", short8cli(p.SHA256), p.DatabaseType, p.Build.Format("2006-01-02"), age, p.Origin)
+ if !p.ExpiresAt.IsZero() {
+ fmt.Fprintf(w, " expires %s (GeoLite2 must be replaced within 30 days of its build)\n", p.ExpiresAt.Format("2006-01-02"))
+ }
+ if p.Attribution != "" {
+ fmt.Fprintf(w, " pages that show this data must credit: %s\n", p.Attribution)
+ }
+ }
+ if resp.FeedSet {
+ fmt.Fprintln(w, "feed: set (CONSOLE_GEOIP_FEED)")
+ }
+ fmt.Fprintln(w)
+ for _, b := range resp.Boxes {
+ name := b.Label
+ if b.Fleet != "" {
+ name = b.Fleet + "/" + b.Label
+ }
+ fmt.Fprintf(w, " %-28s %s\n", name, b.State)
+ }
+ fmt.Fprintln(w)
+ if resp.Ready {
+ fmt.Fprintf(w, "%s: every box strips forged headers and has the current database\n", c.ok("ready"))
+ return
+ }
+ fmt.Fprintf(w, "%s: %d in the way\n", c.bad("not ready"), len(resp.Blocking))
+ for _, b := range resp.Blocking {
+ fmt.Fprintf(w, " %s\n", b)
+ }
+}
+
+func runSystemGeoipOff(args []string) {
+ fs := flag.NewFlagSet("system geoip off", flag.ExitOnError)
+ consoleFlag := consoleOption(fs, "console origin (default: the only logged-in one)")
+ fs.Usage = flagUsage(fs, "")
+ fs.Parse(args)
+ sess := requireDeployAPISession(*consoleFlag)
+ st, body, err := sess.deleteJSON(context.Background(), "/api/cli/system/geoip")
+ if err != nil {
+ log.Fatalf("carlos system geoip off: %v", err)
+ }
+ if st != http.StatusOK {
+ log.Fatalf("carlos system geoip off: %s", apiErrorFromBody(st, body))
+ }
+ var out struct {
+ FeedSet bool `json:"feed_set"`
+ }
+ _ = json.Unmarshal(body, &out)
+ printGeoipOff(stdout, out.FeedSet)
+}
+
+func printGeoipOff(w io.Writer, feedSet bool) {
+ fmt.Fprintln(w, "geoip is off. every edge drops the database within 11 minutes")
+ if feedSet {
+ fmt.Fprintln(w, "the feed is still set and will publish again within a day. unset CONSOLE_GEOIP_FEED to keep it off")
+ }
+}
+
+func short8cli(sha string) string {
+ if len(sha) > 8 {
+ return sha[:8]
+ }
+ return strings.TrimSpace(sha)
+}
diff --git a/system_geoip_test.go b/system_geoip_test.go
new file mode 100644
index 000000000..057f6884e
--- /dev/null
+++ b/system_geoip_test.go
@@ -0,0 +1,128 @@
+package main
+
+import (
+ "bytes"
+ "context"
+ "encoding/json"
+ "io"
+ "net/http"
+ "net/http/httptest"
+ "os"
+ "path/filepath"
+ "strings"
+ "sync/atomic"
+ "testing"
+ "time"
+)
+
+func TestSystemVerbKnowsGeoip(t *testing.T) {
+ if verb, rest, _ := systemVerb([]string{"geoip", "publish", "x.mmdb"}); verb != "geoip" || len(rest) != 2 {
+ t.Fatalf("systemVerb = %q %v", verb, rest)
+ }
+}
+
+func TestPublishGeoipStreamsARawBodyAndFollowsTheJob(t *testing.T) {
+ geoipPollInterval = time.Millisecond
+ t.Cleanup(func() { geoipPollInterval = 2 * time.Second })
+ db := bytes.Repeat([]byte("m"), 4096)
+ path := filepath.Join(t.TempDir(), "db.mmdb")
+ if err := os.WriteFile(path, db, 0o600); err != nil {
+ t.Fatal(err)
+ }
+ var polls atomic.Int32
+ ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ switch {
+ case r.Method == http.MethodPut && r.URL.Path == "/api/cli/system/geoip":
+ if r.Header.Get("Content-Type") != "application/octet-stream" || r.ContentLength != int64(len(db)) {
+ http.Error(w, "bad upload headers", http.StatusBadRequest)
+ return
+ }
+ if r.URL.Query().Get("force") != "1" {
+ http.Error(w, "force not passed", http.StatusBadRequest)
+ return
+ }
+ b, _ := io.ReadAll(r.Body)
+ if !bytes.Equal(b, db) {
+ http.Error(w, "body mangled", http.StatusBadRequest)
+ return
+ }
+ w.WriteHeader(http.StatusAccepted)
+ w.Write([]byte(`{"job":"j1"}`))
+ case r.URL.Path == "/api/cli/system/jobs/j1":
+ if polls.Add(1) < 2 {
+ w.Write([]byte(`{"steps":["received"],"done":false,"error":""}`))
+ return
+ }
+ w.Write([]byte(`{"steps":["received","verifying"],"done":true,"error":"","result":{"sha256":"9f2c0000aaaa","database_type":"DBIP-City-Lite","build_epoch":1790812800}}`))
+ default:
+ http.NotFound(w, r)
+ }
+ }))
+ t.Cleanup(ts.Close)
+ sess := &apiSession{c: newConsoleClient(ts.URL, "tok"), console: ts.URL}
+ var out bytes.Buffer
+ ptr, err := sess.publishGeoip(context.Background(), path, true, &out)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if ptr.DatabaseType != "DBIP-City-Lite" || ptr.SHA256 != "9f2c0000aaaa" {
+ t.Fatalf("pointer = %+v", ptr)
+ }
+ if !strings.Contains(out.String(), "verifying") {
+ t.Fatalf("steps not printed: %q", out.String())
+ }
+}
+
+func TestPublishGeoipReportsTheJobsRefusal(t *testing.T) {
+ geoipPollInterval = time.Millisecond
+ t.Cleanup(func() { geoipPollInterval = 2 * time.Second })
+ path := filepath.Join(t.TempDir(), "db.mmdb")
+ os.WriteFile(path, []byte("junk"), 0o600)
+ ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ if r.Method == http.MethodPut {
+ io.Copy(io.Discard, r.Body)
+ w.WriteHeader(http.StatusAccepted)
+ w.Write([]byte(`{"job":"j2"}`))
+ return
+ }
+ json.NewEncoder(w).Encode(map[string]any{"steps": []string{}, "done": true, "error": "geoip database refused: not a valid MaxMind-format database"})
+ }))
+ t.Cleanup(ts.Close)
+ sess := &apiSession{c: newConsoleClient(ts.URL, "tok"), console: ts.URL}
+ if _, err := sess.publishGeoip(context.Background(), path, false, io.Discard); err == nil || !strings.Contains(err.Error(), "not a valid MaxMind-format database") {
+ t.Fatalf("err = %v", err)
+ }
+}
+
+func TestPrintGeoipStatusVerdicts(t *testing.T) {
+ var resp geoipStatusResponse
+ json.Unmarshal([]byte(`{"pointer":{"sha256":"9f2c0000aaaa","database_type":"GeoLite2-City","build":"2026-10-01T00:00:00Z","origin":"manual","expires_at":"2026-10-31T00:00:00Z","attribution":"This product includes GeoLite2 data"},
+ "boxes":[{"label":"ie-1","state":"ok"},{"label":"node-1","fleet":"tito","state":"old binary"}],
+ "ready":false,"blocking":["tito/node-1: runs a binary that does not strip forged geo headers"]}`), &resp)
+ var out bytes.Buffer
+ printGeoipStatus(&out, resp, time.Date(2026, 10, 6, 0, 0, 0, 0, time.UTC))
+ s := out.String()
+ for _, want := range []string{"GeoLite2-City", "5 days ago", "expires 2026-10-31", "tito/node-1", "not ready", "1 in the way"} {
+ if !strings.Contains(s, want) {
+ t.Errorf("status lacks %q:\n%s", want, s)
+ }
+ }
+ out.Reset()
+ printGeoipStatus(&out, geoipStatusResponse{Ready: true}, time.Now())
+ if !strings.Contains(out.String(), "none published") || !strings.Contains(out.String(), "ready") {
+ t.Errorf("empty status = %q", out.String())
+ }
+}
+
+func TestPrintGeoipOffWarnsAboutTheFeed(t *testing.T) {
+ var out bytes.Buffer
+ printGeoipOff(&out, true)
+ if !strings.Contains(out.String(), "unset CONSOLE_GEOIP_FEED") {
+ t.Fatalf("off with the feed set = %q", out.String())
+ }
+ out.Reset()
+ printGeoipOff(&out, false)
+ if strings.Contains(out.String(), "CONSOLE_GEOIP_FEED") {
+ t.Fatalf("off without a feed mentioned it: %q", out.String())
+ }
+}