| 1 | package console |
| 2 | |
| 3 | // The GeoIP feed (geoip spec §10): the console fetches new editions of |
| 4 | // DB-IP Lite or GeoLite2 and publishes them through the same slot as |
| 5 | // `carlos system geoip publish`, with two extra rules the feed may never |
| 6 | // skip: the database's type must be the configured feed's, and an edition |
| 7 | // must be newer than the last one this feed published. |
| 8 | |
| 9 | import ( |
| 10 | "context" |
| 11 | "crypto/sha256" |
| 12 | "encoding/hex" |
| 13 | "encoding/json" |
| 14 | "errors" |
| 15 | "io" |
| 16 | "log" |
| 17 | mrand "math/rand/v2" |
| 18 | "net/http" |
| 19 | "os" |
| 20 | "strings" |
| 21 | "time" |
| 22 | |
| 23 | "github.com/carlosframework/platform/internal/geoip" |
| 24 | "github.com/carlosframework/platform/internal/release" |
| 25 | ) |
| 26 | |
| 27 | const ( |
| 28 | geoipFeedMaxCompressed = int64(512 << 20) |
| 29 | geoipFeedMaxExpanded = int64(3 << 29) // 1.5 GiB |
| 30 | geoipFeedMaxChecksum = 4 << 10 |
| 31 | ) |
| 32 | |
| 33 | type geoipFeeder struct { |
| 34 | pub *geoipPublisher |
| 35 | store release.Store |
| 36 | spec geoip.FeedSpec |
| 37 | account string |
| 38 | key string |
| 39 | box string |
| 40 | client *http.Client |
| 41 | now func() time.Time |
| 42 | maxExpanded int64 |
| 43 | // Test seams; "" means the real hosts. |
| 44 | dbipBase, maxmindBase string |
| 45 | } |
| 46 | |
| 47 | // newGeoipFeeder reads the feed configuration. false means no feed: unset, |
| 48 | // or misconfigured (logged once; the console starts either way). |
| 49 | func newGeoipFeeder(cfg Config, pub *geoipPublisher) (*geoipFeeder, bool) { |
| 50 | name := strings.TrimSpace(cfg.GeoipFeed) |
| 51 | if name == "" || pub == nil { |
| 52 | return nil, false |
| 53 | } |
| 54 | spec, ok := geoip.ParseFeed(name) |
| 55 | if !ok { |
| 56 | log.Printf("console: geoip feed misconfigured: CONSOLE_GEOIP_FEED %q is not one of dbip-city-lite, dbip-country-lite, geolite2-city, geolite2-country, geoip2-city, geoip2-country; the feed is off", name) |
| 57 | return nil, false |
| 58 | } |
| 59 | if spec.Provider == "maxmind" && (cfg.GeoipMaxMindAccount == "" || cfg.GeoipMaxMindKey == "") { |
| 60 | log.Printf("console: geoip feed misconfigured: %s needs CONSOLE_GEOIP_MAXMIND_ACCOUNT and CONSOLE_GEOIP_MAXMIND_KEY; the feed is off", name) |
| 61 | return nil, false |
| 62 | } |
| 63 | return &geoipFeeder{ |
| 64 | pub: pub, store: pub.store, spec: spec, |
| 65 | account: cfg.GeoipMaxMindAccount, key: cfg.GeoipMaxMindKey, box: cfg.BoxLabel, |
| 66 | client: &http.Client{Timeout: 15 * time.Minute, CheckRedirect: geoipFeedRedirect}, |
| 67 | now: time.Now, |
| 68 | maxExpanded: geoipFeedMaxExpanded, |
| 69 | }, true |
| 70 | } |
| 71 | |
| 72 | // geoipFeedRedirect follows at most 3 redirects, https only, and drops the |
| 73 | // Basic credentials on any cross-host hop (MaxMind redirects downloads to |
| 74 | // presigned R2 URLs; spec §10.2). |
| 75 | func geoipFeedRedirect(req *http.Request, via []*http.Request) error { |
| 76 | if len(via) > 3 { |
| 77 | return errors.New("geoip feed: too many redirects") |
| 78 | } |
| 79 | if req.URL.Scheme != "https" { |
| 80 | return errors.New("geoip feed: refusing a redirect away from https") |
| 81 | } |
| 82 | if len(via) > 0 && req.URL.Host != via[0].URL.Host { |
| 83 | req.Header.Del("Authorization") |
| 84 | } |
| 85 | return nil |
| 86 | } |
| 87 | |
| 88 | // geoipFeedFailure is a check that failed, with a fixed category for |
| 89 | // feed.json and the log (never an error text: it could carry a URL). |
| 90 | type geoipFeedFailure struct{ category string } |
| 91 | |
| 92 | func (f geoipFeedFailure) Error() string { return "geoip feed: " + f.category } |
| 93 | |
| 94 | func feedFail(category string) error { return geoipFeedFailure{category} } |
| 95 | |
| 96 | var errFeedNotFound = errors.New("geoip feed: no edition available yet") |
| 97 | |
| 98 | // Check is one feed check. It always writes feed.json. |
| 99 | func (f *geoipFeeder) Check(ctx context.Context) error { |
| 100 | // Status I/O gets its own short deadline, detached from the check's: a |
| 101 | // check that ran out its 15 minutes on a stalled provider must still |
| 102 | // record the failure, or seven timeouts never add up to an alert (code |
| 103 | // review R2-2). |
| 104 | sctx, scancel := context.WithTimeout(context.WithoutCancel(ctx), 30*time.Second) |
| 105 | defer scancel() |
| 106 | st, etag, known := f.readStatus(sctx) |
| 107 | st.V, st.Feed, st.Box, st.LastCheck = 1, f.spec.Name, f.box, f.now().UTC().Truncate(time.Second) |
| 108 | edition, err := f.check(ctx) |
| 109 | var ff geoipFeedFailure |
| 110 | switch { |
| 111 | case err == nil && edition != "": |
| 112 | st.LastSuccess, st.LastEdition = st.LastCheck, edition |
| 113 | st.ConsecutiveFailures, st.ConsecutiveNotFound, st.LastFailure = 0, 0, "" |
| 114 | case err == nil: |
| 115 | // Nothing new: a success with nothing to do. |
| 116 | st.ConsecutiveFailures, st.ConsecutiveNotFound, st.LastFailure = 0, 0, "" |
| 117 | case errors.Is(err, errGeoipBusy): |
| 118 | // Someone is publishing by hand: not a failure, try tomorrow. |
| 119 | case errors.Is(err, errFeedNotFound): |
| 120 | st.ConsecutiveNotFound++ |
| 121 | st.LastFailure = "not found" |
| 122 | case errors.As(err, &ff): |
| 123 | st.ConsecutiveFailures++ |
| 124 | st.LastFailure = ff.category |
| 125 | log.Printf("console: geoip feed %s: edition %s rejected: %s", f.spec.Name, orNone(edition), ff.category) |
| 126 | default: |
| 127 | st.ConsecutiveFailures++ |
| 128 | st.LastFailure = "store" |
| 129 | log.Printf("console: geoip feed %s: check failed: store", f.spec.Name) |
| 130 | } |
| 131 | if !known { |
| 132 | // The existing history could not be read: overwriting it would turn |
| 133 | // seven recorded failures into one and clear a firing alert with no |
| 134 | // success behind it (code review R3-2). Next check, then. |
| 135 | log.Printf("console: geoip feed status unreadable; this check's outcome was not recorded") |
| 136 | return err |
| 137 | } |
| 138 | wctx, wcancel := context.WithTimeout(context.WithoutCancel(ctx), 30*time.Second) |
| 139 | defer wcancel() |
| 140 | f.writeStatus(wctx, st, etag) |
| 141 | return err |
| 142 | } |
| 143 | |
| 144 | func orNone(s string) string { |
| 145 | if s == "" { |
| 146 | return "(none)" |
| 147 | } |
| 148 | return s |
| 149 | } |
| 150 | |
| 151 | // readStatus reads feed.json capped (never whole: code review R2-1) and |
| 152 | // returns its etag for the conditional write. "" etag = absent. |
| 153 | // known=false means the existing object could not be read at all (a |
| 154 | // transient store error): the caller must not overwrite it. |
| 155 | func (f *geoipFeeder) readStatus(ctx context.Context) (st geoip.FeedStatus, etag string, known bool) { |
| 156 | b, etag, err := readCappedKey(ctx, f.store, geoip.FeedKey, geoip.MaxPointerBytes) |
| 157 | switch { |
| 158 | case errors.Is(err, os.ErrNotExist): |
| 159 | return geoip.FeedStatus{}, "", true // confirmed absent |
| 160 | case errors.Is(err, errPointerTooLarge): |
| 161 | return geoip.FeedStatus{}, etag, true // replace an oversized object, never decode it |
| 162 | case err != nil: |
| 163 | return geoip.FeedStatus{}, "", false |
| 164 | } |
| 165 | s, err := geoip.ParseFeedStatus(b) |
| 166 | if err != nil || s.Feed != f.spec.Name { |
| 167 | return geoip.FeedStatus{}, etag, true // garbage, or another feed's history: replace it |
| 168 | } |
| 169 | return s, etag, true |
| 170 | } |
| 171 | |
| 172 | func (f *geoipFeeder) writeStatus(ctx context.Context, st geoip.FeedStatus, etag string) { |
| 173 | b, _ := json.MarshalIndent(st, "", " ") |
| 174 | if _, err := f.store.Put(ctx, geoip.FeedKey, b, etag); err != nil { |
| 175 | if errors.Is(err, release.ErrCASConflict) && etag == "" { |
| 176 | // Present but unreadable before (oversized under DirStore, or a |
| 177 | // race): a delete then create-only put, best effort. |
| 178 | f.store.Delete(ctx, geoip.FeedKey) |
| 179 | if _, err := f.store.Put(ctx, geoip.FeedKey, b, ""); err == nil { |
| 180 | return |
| 181 | } |
| 182 | } |
| 183 | log.Printf("console: geoip feed status not written") |
| 184 | } |
| 185 | } |
| 186 | |
| 187 | // check returns the edition published ("" when there was nothing new). |
| 188 | func (f *geoipFeeder) check(ctx context.Context) (string, error) { |
| 189 | cur, _, present, curErr := f.pub.Current(ctx) |
| 190 | ours := present && curErr == nil && cur.Origin == "feed:"+f.spec.Name |
| 191 | if f.spec.Provider == "dbip" { |
| 192 | return f.checkDBIP(ctx, cur, ours) |
| 193 | } |
| 194 | return f.checkMaxMind(ctx, cur, ours) |
| 195 | } |
| 196 | |
| 197 | func (f *geoipFeeder) checkDBIP(ctx context.Context, cur geoip.Pointer, ours bool) (string, error) { |
| 198 | now := f.now().UTC() |
| 199 | this := time.Date(now.Year(), now.Month(), 1, 0, 0, 0, 0, time.UTC) |
| 200 | if ours && cur.Edition == this.Format("2006-01") { |
| 201 | return cur.Edition, nil // confirmed current: a success (code review R7-2) |
| 202 | } |
| 203 | months := []time.Time{this} |
| 204 | if !ours { |
| 205 | // No pointer of ours yet: last month is the best there is until |
| 206 | // this month's appears. A pointer of ours already holds it. |
| 207 | months = append(months, this.AddDate(0, -1, 0)) |
| 208 | } |
| 209 | for _, m := range months { |
| 210 | res, err := f.get(ctx, geoip.DBIPURL(f.dbipBase, f.spec, m), false) |
| 211 | if err != nil { |
| 212 | return "", feedFail("download") |
| 213 | } |
| 214 | if res.StatusCode == http.StatusNotFound || res.StatusCode == http.StatusGone { |
| 215 | res.Body.Close() |
| 216 | continue |
| 217 | } |
| 218 | if res.StatusCode != http.StatusOK { |
| 219 | res.Body.Close() |
| 220 | return "", feedFail("download") |
| 221 | } |
| 222 | edition := m.Format("2006-01") |
| 223 | err = f.publish(ctx, res.Body, edition, "", func(r io.Reader) (io.Reader, error) { |
| 224 | return geoip.GunzipReader(r, f.maxExpanded) |
| 225 | }) |
| 226 | res.Body.Close() |
| 227 | return edition, err |
| 228 | } |
| 229 | return "", errFeedNotFound |
| 230 | } |
| 231 | |
| 232 | func (f *geoipFeeder) checkMaxMind(ctx context.Context, cur geoip.Pointer, ours bool) (string, error) { |
| 233 | head, err := f.request(ctx, http.MethodHead, geoip.MaxMindURL(f.maxmindBase, f.spec, "tar.gz"), true) |
| 234 | if err != nil { |
| 235 | return "", feedFail("download") |
| 236 | } |
| 237 | head.Body.Close() |
| 238 | if head.StatusCode == http.StatusNotFound || head.StatusCode == http.StatusGone { |
| 239 | return "", errFeedNotFound |
| 240 | } |
| 241 | if head.StatusCode != http.StatusOK { |
| 242 | return "", feedFail("download") |
| 243 | } |
| 244 | // The edition is the provider's Last-Modified, but only as a real HTTP |
| 245 | // date, re-rendered by us: an arbitrary header value must never reach |
| 246 | // the pointer or feed.json (code review R7-1). |
| 247 | lm, err := http.ParseTime(head.Header.Get("Last-Modified")) |
| 248 | if err != nil { |
| 249 | return "", feedFail("download") |
| 250 | } |
| 251 | edition := lm.UTC().Format(http.TimeFormat) |
| 252 | if ours && cur.Edition == edition { |
| 253 | return edition, nil // confirmed current: a success (code review R7-2) |
| 254 | } |
| 255 | sumRes, err := f.get(ctx, geoip.MaxMindURL(f.maxmindBase, f.spec, "tar.gz.sha256"), true) |
| 256 | if err != nil { |
| 257 | return edition, feedFail("download") |
| 258 | } |
| 259 | sumBody, err := io.ReadAll(io.LimitReader(sumRes.Body, geoipFeedMaxChecksum+1)) |
| 260 | sumRes.Body.Close() |
| 261 | if err != nil || sumRes.StatusCode != http.StatusOK || len(sumBody) > geoipFeedMaxChecksum { |
| 262 | return edition, feedFail("checksum") |
| 263 | } |
| 264 | want := strings.ToLower(strings.TrimSpace(strings.SplitN(string(sumBody), " ", 2)[0])) |
| 265 | if !geoip.IsSHA256Hex(want) { |
| 266 | return edition, feedFail("checksum") |
| 267 | } |
| 268 | res, err := f.get(ctx, geoip.MaxMindURL(f.maxmindBase, f.spec, "tar.gz"), true) |
| 269 | if err != nil || res.StatusCode != http.StatusOK { |
| 270 | if res != nil { |
| 271 | res.Body.Close() |
| 272 | } |
| 273 | return edition, feedFail("download") |
| 274 | } |
| 275 | defer res.Body.Close() |
| 276 | return edition, f.publish(ctx, res.Body, edition, want, func(r io.Reader) (io.Reader, error) { |
| 277 | return geoip.TarGzMMDB(r, f.maxExpanded) |
| 278 | }) |
| 279 | } |
| 280 | |
| 281 | // publish streams body (capped, hashed) through open into the publish |
| 282 | // slot. wantSHA, when set, is the archive's published SHA-256, checked |
| 283 | // after the whole archive has been read and before anything is committed. |
| 284 | func (f *geoipFeeder) publish(ctx context.Context, body io.Reader, edition, wantSHA string, open func(io.Reader) (io.Reader, error)) error { |
| 285 | h := sha256.New() |
| 286 | archive := io.TeeReader(geoip.CappedReader(body, geoipFeedMaxCompressed), h) |
| 287 | db, err := open(archive) |
| 288 | if err != nil { |
| 289 | return classifyFeedErr(err, "invalid") |
| 290 | } |
| 291 | st, err := f.pub.Stage(ctx, db) |
| 292 | if err != nil { |
| 293 | if errors.Is(err, errGeoipBusy) { |
| 294 | return err |
| 295 | } |
| 296 | return classifyFeedErr(err, "download") |
| 297 | } |
| 298 | if wantSHA != "" { |
| 299 | // The tar walk stopped at the database; the checksum covers the |
| 300 | // whole archive, so read the rest through the same capped tee. |
| 301 | if _, err := io.Copy(io.Discard, archive); err != nil { |
| 302 | f.pub.Abandon(st) |
| 303 | return classifyFeedErr(err, "download") |
| 304 | } |
| 305 | if hex.EncodeToString(h.Sum(nil)) != wantSHA { |
| 306 | f.pub.Abandon(st) |
| 307 | return feedFail("checksum") |
| 308 | } |
| 309 | } |
| 310 | _, err = f.pub.Commit(ctx, st, geoipPublishOpts{Origin: "feed:" + f.spec.Name, Edition: edition, RequireType: f.spec.Type, Monotonic: true}) |
| 311 | if err == nil { |
| 312 | return nil |
| 313 | } |
| 314 | msg := err.Error() |
| 315 | switch { |
| 316 | case strings.Contains(msg, "not newer"): |
| 317 | return feedFail("not newer") |
| 318 | case strings.Contains(msg, "the feed returned"): |
| 319 | return feedFail("wrong type") |
| 320 | case strings.Contains(msg, "expired"): |
| 321 | return feedFail("expired") |
| 322 | case errors.Is(err, errGeoipRefused): |
| 323 | return feedFail("invalid") |
| 324 | } |
| 325 | return feedFail("store") |
| 326 | } |
| 327 | |
| 328 | func classifyFeedErr(err error, fallback string) error { |
| 329 | if errors.Is(err, geoip.ErrTooLarge) || errors.Is(err, errGeoipRefused) && strings.Contains(err.Error(), "larger than") { |
| 330 | return feedFail("too large") |
| 331 | } |
| 332 | return feedFail(fallback) |
| 333 | } |
| 334 | |
| 335 | func (f *geoipFeeder) get(ctx context.Context, url string, auth bool) (*http.Response, error) { |
| 336 | return f.request(ctx, http.MethodGet, url, auth) |
| 337 | } |
| 338 | |
| 339 | func (f *geoipFeeder) request(ctx context.Context, method, url string, auth bool) (*http.Response, error) { |
| 340 | req, err := http.NewRequestWithContext(ctx, method, url, nil) |
| 341 | if err != nil { |
| 342 | return nil, err |
| 343 | } |
| 344 | if auth { |
| 345 | // Basic auth in a header, never in the URL (spec §10.1). |
| 346 | req.SetBasicAuth(f.account, f.key) |
| 347 | } |
| 348 | return f.client.Do(req) |
| 349 | } |
| 350 | |
| 351 | // run is the feed's schedule: 5 minutes after start, then daily. |
| 352 | func (f *geoipFeeder) run(ctx context.Context) { |
| 353 | wait := 5 * time.Minute |
| 354 | for { |
| 355 | select { |
| 356 | case <-ctx.Done(): |
| 357 | return |
| 358 | case <-time.After(wait): |
| 359 | } |
| 360 | cctx, cancel := context.WithTimeout(ctx, f.pub.deadline) |
| 361 | f.Check(cctx) |
| 362 | cancel() |
| 363 | wait = 24*time.Hour + mrand.N(time.Hour) |
| 364 | } |
| 365 | } |
| 366 | |
| 367 | // clearFeedStatus removes feed.json when no feed is configured, so a feed |
| 368 | // switched off never alerts anyone. |
| 369 | func clearFeedStatus(ctx context.Context, s release.Store) error { |
| 370 | if err := s.Delete(ctx, geoip.FeedKey); err != nil && !errors.Is(err, os.ErrNotExist) { |
| 371 | return err |
| 372 | } |
| 373 | return nil |
| 374 | } |
| 375 | |
| 376 | // retryClearFeedStatus keeps trying until feed.json is gone or ctx ends: |
| 377 | // a failed delete must not leave a disabled feed's alert firing for good |
| 378 | // (code review R6-1). Each attempt is bounded; the wait doubles from |
| 379 | // first to max. |
| 380 | func retryClearFeedStatus(ctx context.Context, s release.Store, first, max time.Duration) { |
| 381 | wait := first |
| 382 | for { |
| 383 | actx, cancel := context.WithTimeout(ctx, 30*time.Second) |
| 384 | err := clearFeedStatus(actx, s) |
| 385 | cancel() |
| 386 | if err == nil { |
| 387 | return |
| 388 | } |
| 389 | log.Printf("console: geoip feed status could not be cleared; retrying in %s", wait) |
| 390 | select { |
| 391 | case <-ctx.Done(): |
| 392 | return |
| 393 | case <-time.After(wait): |
| 394 | } |
| 395 | if wait *= 2; wait > max { |
| 396 | wait = max |
| 397 | } |
| 398 | } |
| 399 | } |
| 400 | |
amadan