| new file mode 100644 |
| index 000000000..015859ca6 |
| --- /dev/null |
| +++ b/internal/edge/geoipmirror.go |
| @@ -0,0 +1,437 @@ |
| +package edge |
| + |
| +import ( |
| + "context" |
| + "crypto/sha256" |
| + "encoding/hex" |
| + "encoding/json" |
| + "errors" |
| + "fmt" |
| + "io" |
| + "log" |
| + "math/rand/v2" |
| + "os" |
| + "path/filepath" |
| + "strings" |
| + "sync" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/geo" |
| + "github.com/carlosframework/platform/internal/geoip" |
| + "github.com/carlosframework/platform/internal/release" |
| +) |
| + |
| +// The GeoIP mirror (geoip spec §5): poll control/geoip/current.json, fetch |
| +// the content-addressed blob it names into an immutable local generation, |
| +// and hot-swap the edge's reader. genMu serialises everything that chooses, |
| +// expires or deletes a generation (codex R3-1); network I/O stays outside it. |
| + |
| +var ( |
| + geoMirrorInterval = 10 * time.Minute |
| + geoMirrorJitter = time.Minute |
| + geoMirrorDeadline = 10 * time.Minute |
| + geoExpiryInterval = time.Minute |
| +) |
| + |
| +type geoCurrent struct { |
| + SHA256 string `json:"sha256,omitempty"` |
| + DatabaseType string `json:"database_type,omitempty"` |
| + BuildEpoch int64 `json:"build_epoch,omitempty"` |
| +} |
| + |
| +type geoMirror struct { |
| + store release.Store |
| + dir string |
| + sw *geo.Swappable |
| + now func() time.Time |
| + maxBytes int64 |
| + |
| + genMu sync.Mutex |
| + loadedSHA string |
| + loadedType string |
| + loadedBuild time.Time |
| + skipSHA map[string]string // sha -> why (rejected | expired); cleared when the pointer moves on |
| + lastErr string // last poll's error category, "" when healthy |
| + failingAt time.Time |
| + lastFailLog time.Time |
| +} |
| + |
| +func newGeoMirror(store release.Store, dir string, sw *geo.Swappable) *geoMirror { |
| + return &geoMirror{store: store, dir: dir, sw: sw, now: time.Now, maxBytes: geoip.MaxBlobBytes, skipSHA: map[string]string{}} |
| +} |
| + |
| +// GeoIPState is the heartbeat's view of this edge's GeoIP (spec §5.5). |
| +type GeoIPState struct { |
| + Mode string `json:"mode"` // mirror | tunnel | none |
| + SHA256 string `json:"sha256,omitempty"` |
| + Type string `json:"type,omitempty"` |
| + Error string `json:"error,omitempty"` |
| +} |
| + |
| +func (m *geoMirror) State() GeoIPState { |
| + m.genMu.Lock() |
| + defer m.genMu.Unlock() |
| + return GeoIPState{Mode: "mirror", SHA256: m.loadedSHA, Type: m.loadedType, Error: m.lastErr} |
| +} |
| + |
| +func (m *geoMirror) path(sha string) string { return filepath.Join(m.dir, sha+".mmdb") } |
| + |
| +// Boot installs the local generation `current` names, if any, without |
| +// touching the bucket, and clears everything else (spec §5.3 "At boot"). |
| +func (m *geoMirror) Boot() { |
| + if err := os.MkdirAll(m.dir, 0o755); err != nil { |
| + log.Printf("edge: geoip directory unusable; GeoIP stays off until it is: %v", err) |
| + return |
| + } |
| + os.Chmod(m.dir, 0o755) |
| + m.genMu.Lock() |
| + defer m.genMu.Unlock() |
| + cur, err := m.readCurrent() |
| + keep := "" |
| + switch { |
| + case err != nil && !errors.Is(err, os.ErrNotExist): |
| + log.Printf("edge: geoip local state unreadable; starting with no database") |
| + os.Remove(filepath.Join(m.dir, "current")) |
| + case cur.SHA256 == "": |
| + case geoip.Expired(cur.DatabaseType, time.Unix(cur.BuildEpoch, 0), m.now()): |
| + log.Printf("edge: geoip database %s expired (GeoLite2, 30 days after build) and was removed", short(cur.SHA256)) |
| + m.writeCurrentLocked(geoCurrent{}) |
| + default: |
| + db, err := geo.OpenMMDBTrusted(m.path(cur.SHA256)) |
| + if err != nil { |
| + log.Printf("edge: geoip local generation %s unopenable; starting with no database", short(cur.SHA256)) |
| + os.Remove(filepath.Join(m.dir, "current")) |
| + break |
| + } |
| + m.sw.Swap(db) |
| + m.loadedSHA, m.loadedType, m.loadedBuild = cur.SHA256, db.Source(), db.BuildEpoch() |
| + keep = cur.SHA256 |
| + log.Printf("edge: geoip database %s (%s, built %s) loaded", short(cur.SHA256), db.Source(), db.BuildEpoch().Format("2006-01-02")) |
| + } |
| + m.cleanLocked(keep) |
| +} |
| + |
| +func short(sha string) string { |
| + if len(sha) > 8 { |
| + return sha[:8] |
| + } |
| + return sha |
| +} |
| + |
| +func (m *geoMirror) readCurrent() (geoCurrent, error) { |
| + var c geoCurrent |
| + b, err := os.ReadFile(filepath.Join(m.dir, "current")) |
| + if err != nil { |
| + return c, err |
| + } |
| + if err := json.Unmarshal(b, &c); err != nil { |
| + return c, err |
| + } |
| + if c.SHA256 != "" && !geoip.IsSHA256Hex(c.SHA256) { |
| + return geoCurrent{}, errors.New("bad sha") |
| + } |
| + return c, nil |
| +} |
| + |
| +// writeCurrentLocked replaces `current` atomically: temp, fsync, rename, |
| +// fsync the directory (spec §5.2). |
| +func (m *geoMirror) writeCurrentLocked(c geoCurrent) error { |
| + b, _ := json.Marshal(c) |
| + tmp, err := os.CreateTemp(m.dir, ".current-*") |
| + if err != nil { |
| + return err |
| + } |
| + defer os.Remove(tmp.Name()) |
| + if _, err := tmp.Write(b); err != nil { |
| + tmp.Close() |
| + return err |
| + } |
| + if err := tmp.Sync(); err != nil { |
| + tmp.Close() |
| + return err |
| + } |
| + if err := tmp.Close(); err != nil { |
| + return err |
| + } |
| + if err := os.Chmod(tmp.Name(), 0o644); err != nil { |
| + return err |
| + } |
| + if err := os.Rename(tmp.Name(), filepath.Join(m.dir, "current")); err != nil { |
| + return err |
| + } |
| + return syncDir(m.dir) |
| +} |
| + |
| +func syncDir(dir string) error { |
| + d, err := os.Open(dir) |
| + if err != nil { |
| + return err |
| + } |
| + defer d.Close() |
| + return d.Sync() |
| +} |
| + |
| +// cleanLocked deletes every generation but keep, every .incoming-* and |
| +// every stray .current-*. |
| +func (m *geoMirror) cleanLocked(keep string) { |
| + entries, _ := os.ReadDir(m.dir) |
| + for _, e := range entries { |
| + n := e.Name() |
| + if n == "current" || (keep != "" && n == keep+".mmdb") { |
| + continue |
| + } |
| + if strings.HasSuffix(n, ".mmdb") || strings.HasPrefix(n, ".incoming-") || strings.HasPrefix(n, ".current-") { |
| + os.Remove(filepath.Join(m.dir, n)) |
| + } |
| + } |
| +} |
| + |
| +// clearLocked turns GeoIP off on this box: empty current, nil reader, no |
| +// generations. |
| +func (m *geoMirror) clearLocked() { |
| + if err := m.writeCurrentLocked(geoCurrent{}); err != nil { |
| + os.Remove(filepath.Join(m.dir, "current")) |
| + } |
| + m.sw.Swap(nil) |
| + m.loadedSHA, m.loadedType, m.loadedBuild = "", "", time.Time{} |
| + m.cleanLocked("") |
| +} |
| + |
| +// Poll is one pass (spec §5.3). The caller bounds ctx by geoMirrorDeadline. |
| +func (m *geoMirror) Poll(ctx context.Context) error { |
| + if err := os.MkdirAll(m.dir, 0o755); err != nil { |
| + return m.fail("local directory unusable", err) |
| + } |
| + raw, err := readCapped(ctx, m.store, geoip.PointerKey, geoip.MaxPointerBytes) |
| + if errors.Is(err, os.ErrNotExist) { |
| + m.genMu.Lock() |
| + if m.loadedSHA != "" { |
| + log.Printf("edge: geoip turned off; no database published") |
| + } |
| + m.clearLocked() |
| + m.lastErr = "" |
| + m.genMu.Unlock() |
| + return nil |
| + } |
| + if err != nil { |
| + return m.fail("pointer unreadable", err) |
| + } |
| + p, err := geoip.ParsePointer(raw) |
| + if err != nil { |
| + return m.fail("pointer invalid", err) |
| + } |
| + m.genMu.Lock() |
| + if p.Expired || geoip.Expired(p.DatabaseType, p.Build(), m.now()) { |
| + if m.skipSHA[p.SHA256] != "expired" { |
| + log.Printf("edge: geoip database %s expired (GeoLite2, 30 days after build) and was removed", short(p.SHA256)) |
| + } |
| + m.skipSHA = map[string]string{p.SHA256: "expired"} |
| + m.clearLocked() |
| + m.lastErr = "" |
| + m.genMu.Unlock() |
| + return nil |
| + } |
| + if p.SHA256 == m.loadedSHA { |
| + m.lastErr = "" |
| + m.genMu.Unlock() |
| + return nil |
| + } |
| + if why, ok := m.skipSHA[p.SHA256]; ok { |
| + m.genMu.Unlock() |
| + _ = why |
| + return nil |
| + } |
| + m.skipSHA = map[string]string{} // the pointer moved on |
| + m.genMu.Unlock() |
| + |
| + tmp, err := m.download(ctx, p) |
| + if err != nil { |
| + if errors.Is(err, errGeoDigest) || errors.Is(err, errGeoSize) { |
| + m.genMu.Lock() |
| + m.skipSHA[p.SHA256] = "rejected" |
| + m.genMu.Unlock() |
| + log.Printf("edge: geoip blob %s rejected: %v", short(p.SHA256), err) |
| + } |
| + return m.fail("blob unavailable", err) |
| + } |
| + return m.install(tmp, p) |
| +} |
| + |
| +var ( |
| + errGeoDigest = errors.New("digest mismatch") |
| + errGeoSize = errors.New("size mismatch") |
| +) |
| + |
| +// download streams the blob to .incoming-<sha>, checks size and digest, |
| +// fsyncs, and returns its path. Outside genMu. |
| +func (m *geoMirror) download(ctx context.Context, p geoip.Pointer) (string, error) { |
| + rc, _, err := release.OpenObject(ctx, m.store, p.Key) |
| + if err != nil { |
| + return "", err |
| + } |
| + defer rc.Close() |
| + tmp := filepath.Join(m.dir, ".incoming-"+p.SHA256) |
| + f, err := os.OpenFile(tmp, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0o644) |
| + if err != nil { |
| + return "", err |
| + } |
| + limit := p.Size |
| + if m.maxBytes < limit { |
| + limit = m.maxBytes |
| + } |
| + h := sha256.New() |
| + n, err := io.Copy(io.MultiWriter(f, h), io.LimitReader(rc, limit+1)) |
| + if err == nil { |
| + err = f.Sync() |
| + } |
| + if cerr := f.Close(); err == nil { |
| + err = cerr |
| + } |
| + switch { |
| + case err != nil: |
| + os.Remove(tmp) |
| + return "", err |
| + case n > limit || n != p.Size: |
| + // n > limit: the blob ran past the cap (the box's maxBytes may be |
| + // below what the pointer claims), however the counts compare. |
| + os.Remove(tmp) |
| + return "", errGeoSize |
| + case hex.EncodeToString(h.Sum(nil)) != p.SHA256: |
| + os.Remove(tmp) |
| + return "", errGeoDigest |
| + } |
| + return tmp, nil |
| +} |
| + |
| +// install commits a verified download under genMu (spec §5.3 steps 3-5), |
| +// re-checking expiry against the file's own metadata first. |
| +func (m *geoMirror) install(tmp string, p geoip.Pointer) error { |
| + m.genMu.Lock() |
| + defer m.genMu.Unlock() |
| + final := m.path(p.SHA256) |
| + if err := os.Rename(tmp, final); err != nil { |
| + os.Remove(tmp) |
| + return m.failLocked("install failed", err) |
| + } |
| + syncDir(m.dir) |
| + db, err := geo.OpenMMDBTrusted(final) |
| + if err != nil { |
| + os.Remove(final) |
| + m.skipSHA[p.SHA256] = "rejected" |
| + log.Printf("edge: geoip blob %s rejected: unopenable", short(p.SHA256)) |
| + return m.failLocked("blob unopenable", err) |
| + } |
| + if geoip.Expired(db.Source(), db.BuildEpoch(), m.now()) { |
| + db.Close() |
| + os.Remove(final) |
| + m.skipSHA[p.SHA256] = "expired" |
| + log.Printf("edge: geoip database %s expired (GeoLite2, 30 days after build) and was removed", short(p.SHA256)) |
| + return nil |
| + } |
| + if err := m.writeCurrentLocked(geoCurrent{SHA256: p.SHA256, DatabaseType: db.Source(), BuildEpoch: db.BuildEpoch().Unix()}); err != nil { |
| + db.Close() |
| + os.Remove(final) |
| + return m.failLocked("install failed", err) |
| + } |
| + m.sw.Swap(db) |
| + m.loadedSHA, m.loadedType, m.loadedBuild = p.SHA256, db.Source(), db.BuildEpoch() |
| + m.cleanLocked(p.SHA256) |
| + m.lastErr, m.failingAt = "", time.Time{} |
| + log.Printf("edge: geoip database %s (%s, built %s) loaded", short(p.SHA256), db.Source(), db.BuildEpoch().Format("2006-01-02")) |
| + return nil |
| +} |
| + |
| +// ExpireTick is D1 on the loaded generation, under genMu, so it can only |
| +// ever remove the generation that is current at that moment (codex R3-1). |
| +func (m *geoMirror) ExpireTick() { |
| + m.genMu.Lock() |
| + defer m.genMu.Unlock() |
| + if m.loadedSHA == "" || !geoip.Expired(m.loadedType, m.loadedBuild, m.now()) { |
| + return |
| + } |
| + log.Printf("edge: geoip database %s expired (GeoLite2, 30 days after build) and was removed", short(m.loadedSHA)) |
| + m.skipSHA[m.loadedSHA] = "expired" |
| + m.clearLocked() |
| +} |
| + |
| +func (m *geoMirror) fail(category string, err error) error { |
| + m.genMu.Lock() |
| + defer m.genMu.Unlock() |
| + return m.failLocked(category, err) |
| +} |
| + |
| +// failLocked records the category for the heartbeat and logs at most hourly |
| +// while failing. The error text is never logged: it may carry a key or |
| +// bytes; the category is the message (spec §12). |
| +func (m *geoMirror) failLocked(category string, err error) error { |
| + now := m.now() |
| + if m.lastErr == "" { |
| + m.failingAt = now |
| + } |
| + m.lastErr = category |
| + if now.Sub(m.lastFailLog) >= time.Hour { |
| + m.lastFailLog = now |
| + log.Printf("edge: geoip %s; keeping the loaded database", category) |
| + } |
| + return fmt.Errorf("%s: %w", category, err) |
| +} |
| + |
| +// readCapped reads key through OpenObject, refusing more than max bytes. |
| +func readCapped(ctx context.Context, s release.Store, key string, max int) ([]byte, error) { |
| + rc, _, err := release.OpenObject(ctx, s, key) |
| + if err != nil { |
| + return nil, err |
| + } |
| + defer rc.Close() |
| + b, err := io.ReadAll(io.LimitReader(rc, int64(max)+1)) |
| + if err != nil { |
| + return nil, err |
| + } |
| + if len(b) > max { |
| + return nil, fmt.Errorf("%s larger than %d bytes", key, max) |
| + } |
| + return b, nil |
| +} |
| + |
| +// run is the process-lifetime loop: a poll every interval+jitter, each |
| +// bounded by geoMirrorDeadline, and an independent expiry ticker that never |
| +// touches the network. |
| +func (m *geoMirror) run(stop <-chan struct{}) { |
| + go func() { |
| + t := time.NewTicker(geoExpiryInterval) |
| + defer t.Stop() |
| + for { |
| + select { |
| + case <-stop: |
| + return |
| + case <-t.C: |
| + m.ExpireTick() |
| + } |
| + } |
| + }() |
| + for { |
| + ctx, cancel := context.WithTimeout(context.Background(), geoMirrorDeadline) |
| + m.Poll(ctx) |
| + cancel() |
| + select { |
| + case <-stop: |
| + return |
| + case <-time.After(geoMirrorInterval + rand.N(geoMirrorJitter)): |
| + } |
| + } |
| +} |
| + |
| +// WithGeoTunnelClient marks this edge as a tunnel client's: it never mirrors |
| +// the GeoIP database and takes the hub's answer instead (spec §5.1). |
| +func WithGeoTunnelClient() RunOption { return func(e *Edge) { e.geoTunnelClient = true } } |
| + |
| +// GeoIPState is the heartbeat block (spec §5.5). |
| +func (e *Edge) GeoIPState() GeoIPState { |
| + if e.geoMirror != nil { |
| + return e.geoMirror.State() |
| + } |
| + if e.geoMode == "" { |
| + return GeoIPState{Mode: "none"} |
| + } |
| + return GeoIPState{Mode: e.geoMode} |
| +} |
amadan