diff --git a/README.md b/README.md index 47eae62..1e2db3a 100644 --- a/README.md +++ b/README.md @@ -63,8 +63,8 @@ docker run --memory=5g --memory-swap=5g -p 8080:8080 \ -v routewatch:/var/lib/berlin.sneak.app.routewatch routewatch ``` -The memory limit is explained in Memory. The volume holds the database and the -copies of the IP lists. +The memory limit is explained in Memory. The volume holds the database, the +copies of the IP lists and the geolocation files. To run the binary on the host instead, build it with `make build`. The build needs Go and a C compiler, because the SQLite library is written in C: @@ -293,6 +293,7 @@ address. Every entry has the same fields: an empty list when it is on none - `ip_lists`: the public IP lists the address is on (see IP lists), or an empty list when it is on none +- `geo`: where the address is (see Geolocation), or null when no source answers Every other field is about the prefix in the routing table that holds the address. Each is null when no prefix in the routing table holds the address: @@ -509,6 +510,72 @@ How RouteWatch keeps the answers: - It keeps answers for at most 250,000 addresses, and keeps no answer for a new address while it holds that many. +## Geolocation + +An IP lookup finds where each address is, on a best-effort basis. It uses two +sources: + +- [geojs](https://www.geojs.io/) is asked first. One request carries every + address of the lookup, and it has 1 second to answer. +- [DB-IP City Lite](https://db-ip.com/db/download/ip-to-city-lite) answers when + geojs has no answer for the address. It is a database file that RouteWatch + reads itself. + +geojs has no answer when its request fails or runs out of time. An answer with +no country, city or coordinates also counts as no answer, and RouteWatch does +not cache it. + +The `geo` object of a result has these fields: + +- `country_code`: the two-letter country code +- `country`: the country's name in English +- `continent_code`: the two-letter continent code +- `region`: the region or state +- `city`: the city +- `latitude` and `longitude`: the location in degrees +- `accuracy_radius_km`: how far from that location the address may be, in km +- `time_zone`: the time zone's name, such as `Europe/Berlin` +- `source`: the source of the answer, `geojs` or `dbip` +- `updated`: when the answer was fetched from geojs, or when the database copy + was downloaded + +A field the source does not know is null. DB-IP has no `accuracy_radius_km` and +no `time_zone`, so both are null in its answers. `geo` itself is null when +neither source answers. A reserved address, one whose `is_bogon` flag is true, +goes to neither source, so its `geo` is null. + +What goes outside: geojs receives each address that the cache does not hold. For +`/ip/` and the home page, that address is the visitor's own. A lookup answered +from the cache sends nothing. DB-IP receives only the monthly request for its +file. + +How RouteWatch keeps the answers of geojs: + +- It holds at most 250,000 addresses in memory, and drops the least recently + used first. That takes about 100 MB of the Go heap. +- It writes new answers every 5 seconds to `cache.db` in the state directory. +- It loads the newest 250,000 answers from `cache.db` at startup. +- An address not in memory is read from `cache.db` before geojs is asked. +- An answer expires 7 days after its fetch. Every 10 minutes RouteWatch deletes + expired answers from memory and from `cache.db`. + +Answers from the database are not cached, because reading the file is local. + +How RouteWatch keeps the database: + +- It keeps one copy in the state directory, named `dbip-city-lite.mmdb`. +- It downloads the current month's file at the first start, and then once a + month. Until the first download finishes, only geojs answers. +- A download fails when it takes over 10 minutes or unpacks to over 200 MiB. +- A new file replaces the old copy only once it opens. After a failed download + RouteWatch keeps the old copy and tries again an hour later. +- The file is about 130 MB. RouteWatch maps it into memory, outside the Go heap. + The kernel loads the parts that are read, and can drop them again when memory + is short. + +The home page's footer says "IP Geolocation by DB-IP" with a link to +[db-ip.com](https://db-ip.com). + ## Configuration RouteWatch reads its settings from environment variables. It takes no @@ -560,7 +627,9 @@ are not read. The state directory holds: - `db.sqlite`: the database -- `cache.db`: the answers of PTR lookups +- `cache.db`: the answers of PTR lookups and of geojs +- `dbip-city-lite.mmdb`: the copy of DB-IP's City Lite database (see + Geolocation) - `iplists`: the IP list copies It is at: @@ -585,8 +654,9 @@ a memory limit of 5 GiB and swap disabled: docker run --memory=5g --memory-swap=5g ... ``` -The 5 GiB leaves room above the ceilings for spikes and for the kernel's cache -of the database file. +The 5 GiB leaves room above the ceilings for spikes, for the kernel's cache of +the database file, and for the mapped copy of DB-IP's database (see +Geolocation). The ceilings add up to 3.7 GiB: @@ -722,6 +792,8 @@ The parts: address. - The PTR cache looks up the PTR names of addresses and keeps the answers (see PTR lookups). +- The geolocation part finds where addresses are, keeps the answers of geojs and + downloads DB-IP's database (see Geolocation). - The database maintainer checkpoints the database every 5 seconds: it folds the write-ahead log, where SQLite first writes changes, into the database file. It reclaims free space every 10 minutes. Every hour it refreshes the statistics @@ -831,7 +903,8 @@ On `SIGTERM` or `SIGINT` the daemon stops its parts in this order: 4. the database maintainer 5. the IP list downloads 6. the HTTP server -7. the PTR cache, which writes what it holds +7. the geolocation part, which writes the geojs answers it holds +8. the PTR cache, which writes what it holds Last, the daemon checkpoints and closes the database. It gives all this 8 seconds and then exits with status 0. A stop that is still running after 9 @@ -854,6 +927,8 @@ What a stop can lose: SQLite rolls back any write the exit cut off. - When the 8 seconds run out, the PTR answers not yet in `cache.db` are lost. RouteWatch asks the resolvers for them again the next time they are needed. +- When the 8 seconds run out, the geojs answers not yet in `cache.db` are lost. + RouteWatch asks geojs for them again the next time they are needed. ### Code layout @@ -870,6 +945,7 @@ routewatch/ │ ├── cachedb/ # cache.db, the caches of outside answers │ ├── config/ # Settings from the environment │ ├── database/ # SQLite storage +│ ├── geo/ # Geolocation: geojs, DB-IP City Lite, the cache │ ├── iplist/ # Public IP lists: downloads and lookups │ ├── logger/ # Structured logging │ ├── metrics/ # Feed and route update figures @@ -922,11 +998,16 @@ announcements(id PRIMARY KEY, prefix_id, peer_asn, origin_asn, path, The indexes are in `internal/database/schema.sql`. -`cache.db` is a separate file, with one table: +`cache.db` is a separate file, with two tables: ```sql -- The answers of PTR lookups. names is empty for an address with no PTR names. hostnames(address PRIMARY KEY, names, expires_at) + +-- The answers of geojs. A part geojs does not know is NULL. +geolocations(address PRIMARY KEY, country_code, country, continent_code, + region, city, latitude, longitude, accuracy_radius_km, time_zone, + fetched_at) ``` ## Development diff --git a/TODO.md b/TODO.md index 7cf65ff..ea02720 100644 --- a/TODO.md +++ b/TODO.md @@ -53,6 +53,10 @@ problem the run finds. # Completed Steps +- 2026-11-10: each result of the IP lookup has `geo`, where the address is. It + comes from geojs, with DB-IP's City Lite database as the fallback. The geojs + answers stay in memory and in `cache.db` for 7 days (closes + https://git.eeqj.de/sneak/routewatch/issues/239). - 2026-11-10: the IP lookup checks the `protonvpn` list of ProtonVPN's /24 networks and the `protonvpn_entry` list of its servers' IPv4 and IPv6 addresses, from a public mirror of Proton's server list (closes diff --git a/go.mod b/go.mod index fe4c33b..1a8279c 100644 --- a/go.mod +++ b/go.mod @@ -8,8 +8,11 @@ require ( github.com/go-chi/chi/v5 v5.2.2 github.com/google/btree v1.1.3 github.com/google/uuid v1.6.0 + github.com/hashicorp/golang-lru/v2 v2.0.7 github.com/mattn/go-sqlite3 v1.14.29 + github.com/maxmind/mmdbwriter v1.2.0 github.com/miekg/dns v1.1.73 + github.com/oschwald/maxminddb-golang/v2 v2.9.0 github.com/rcrowley/go-metrics v0.0.0-20250401214520-65e299d6c5c9 go.uber.org/fx v1.24.0 go4.org/netipx v0.0.0-20260823151212-3075585bcbeb @@ -22,5 +25,5 @@ require ( go.uber.org/multierr v1.10.0 // indirect go.uber.org/zap v1.26.0 // indirect golang.org/x/net v0.57.0 // indirect - golang.org/x/sys v0.47.0 // indirect + golang.org/x/sys v0.49.0 // indirect ) diff --git a/go.sum b/go.sum index 3eb4d33..8d5968a 100644 --- a/go.sum +++ b/go.sum @@ -1,7 +1,5 @@ github.com/bits-and-blooms/bitset v1.26.0 h1:WPvcPZbP1yuKofmjs2SYIFO1EjzIGhCP9ducElLsh5Q= github.com/bits-and-blooms/bitset v1.26.0/go.mod h1:7hO7Gc7Pp1vODcmWvKMRA9BNmbv6a/7QIWpPxHddWR8= -github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= -github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY= github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto= github.com/go-chi/chi/v5 v5.2.2 h1:CMwsvRVTbXVytCk1Wd72Zy1LAsAh9GxMmSNWLHCG618= @@ -10,16 +8,20 @@ github.com/google/btree v1.1.3 h1:CVpQJjYgC4VbzxeGVHfvZrv1ctoYCAI8vbl07Fcxlyg= github.com/google/btree v1.1.3/go.mod h1:qOPhT0dTNdNzV6Z/lhRX0YXUafgPLFUh+gZMl761Gm4= github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +github.com/hashicorp/golang-lru/v2 v2.0.7 h1:a+bsQ5rvGLjzHuww6tVxozPZFVghXaHOwFs4luLUK2k= +github.com/hashicorp/golang-lru/v2 v2.0.7/go.mod h1:QeFd9opnmA6QUJc5vARoKUSoFhyfM2/ZepoAG6RGpeM= github.com/mattn/go-sqlite3 v1.14.29 h1:1O6nRLJKvsi1H2Sj0Hzdfojwt8GiGKm+LOfLaBFaouQ= github.com/mattn/go-sqlite3 v1.14.29/go.mod h1:Uh1q+B4BYcTPb+yiD3kU8Ct7aC0hY9fxUwlHK0RXw+Y= +github.com/maxmind/mmdbwriter v1.2.0 h1:hyvDopImmgvle3aR8AaddxXnT0iQH2KWJX3vNfkwzYM= +github.com/maxmind/mmdbwriter v1.2.0/go.mod h1:EQmKHhk2y9DRVvyNxwCLKC5FrkXZLx4snc5OlLY5XLE= github.com/miekg/dns v1.1.73 h1:uhT8nJxmTrPJYClxVxTCX+CVn6qnzSiybRk72Z6DgrE= github.com/miekg/dns v1.1.73/go.mod h1:RW2Obtfd5NZHvOFe3zYG0W8koWOQtAzyHaLo8vASBuQ= -github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= -github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/oschwald/maxminddb-golang/v2 v2.9.0 h1:YpIl3WoA9oIuNvqNaO/aD1N/7Ci2wOqoKH5GWo53H5Y= +github.com/oschwald/maxminddb-golang/v2 v2.9.0/go.mod h1:3TxexMBcR0ss3SbyYvnqfsDzDjN4o1KCg/nTsaMAivU= github.com/rcrowley/go-metrics v0.0.0-20250401214520-65e299d6c5c9 h1:bsUq1dX0N8AOIL7EB/X911+m4EHsnWEHeJ0c+3TTBrg= github.com/rcrowley/go-metrics v0.0.0-20250401214520-65e299d6c5c9/go.mod h1:bCqnVzQkZxMG4s8nGwiZ5l3QUCyqpo9Y+/ZMZ9VjZe4= -github.com/stretchr/testify v1.8.1 h1:w7B6lhMri9wdJUVmEZPGGhZzrYTPvgJArz7wNPgYKsk= -github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4= +github.com/stretchr/testify v1.12.1 h1:EuwCh5fleGS7H32xRwO3wRGT7DxrDhLAT6FF8MpWDWE= +github.com/stretchr/testify v1.12.1/go.mod h1:MDEgiDPPsNp5cuIrHPPCyornHKgEVbtFUmoNlxoYthg= go.uber.org/dig v1.19.0 h1:BACLhebsYdpQ7IROQ1AGPjrXcP5dF80U3gKoFzbaq/4= go.uber.org/dig v1.19.0/go.mod h1:Us0rSJiThwCv2GteUN0Q7OKvU7n5J4dxZ9JKUXozFdE= go.uber.org/fx v1.24.0 h1:wE8mruvpg2kiiL1Vqd0CC+tr0/24XIB10Iwp2lLWzkg= @@ -30,14 +32,16 @@ go.uber.org/multierr v1.10.0 h1:S0h4aNzvfcFsC3dRF1jLoaov7oRaKqRGC/pUEJ2yvPQ= go.uber.org/multierr v1.10.0/go.mod h1:20+QtiLqy0Nd6FdQB9TLXag12DsQkrbs3htMFfDN80Y= go.uber.org/zap v1.26.0 h1:sI7k6L95XOKS281NhVKOFCUNIvv9e0w4BF8N3u+tCRo= go.uber.org/zap v1.26.0/go.mod h1:dtElttAiwGvoJ/vj4IwHBS/gXsEu/pZ50mUIRWuG0so= +go.yaml.in/yaml/v3 v3.0.5 h1:N6y/pJk8buWs9NY5ERU2HSMfm+IuD/OtfdAnq6kESPw= +go.yaml.in/yaml/v3 v3.0.5/go.mod h1:HVTZu1O7/Vkt2N+BFy8Zza+lnLsABggaTM2ZpNIGuKg= go4.org/netipx v0.0.0-20260823151212-3075585bcbeb h1:XBM4hvfwGAttkkiTIFfeigdfcL1xIfdKXqFdgiHGtDs= go4.org/netipx v0.0.0-20260823151212-3075585bcbeb/go.mod h1:PLyyIXexvUFg3Owu6p/WfdlivPbZJsZdgWZlrGope/Y= golang.org/x/net v0.57.0 h1:K5+3DljvIuDG9/Jv9rvyMywYNFCQ9RSUY6OOTTkT+tE= golang.org/x/net v0.57.0/go.mod h1:KpXc8iv+r3XplLAG/f7Jsf9RPszJzdR0f58q9vGOuEU= golang.org/x/sync v0.24.0 h1:iqiDrGoDgyuvAKrhLQthUEzU/R378N78gXbBjxed7oA= golang.org/x/sync v0.24.0/go.mod h1:sUUOizhqBxiL6pEWpqNLUiaJn1ShEbZ6BBqskPbjZm0= -golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs= -golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= +golang.org/x/sys v0.49.0 h1:XbzkgYJHdqh/8m2Uu0W/dQv8nktxx4BFHp1M0gROTXA= +golang.org/x/sys v0.49.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og= golang.org/x/term v0.45.0 h1:NwWyBmoJCbfTHpxrWoZ9C6/VxOf7ic219I8xZZFdrf0= golang.org/x/term v0.45.0/go.mod h1:9aqxs0blBcrm/n0L9QW0aRVD+ktan8ssZromtqJC43w= golang.org/x/text v0.43.0 h1:1QivrlhwAsnMOcqgOdl9my7s9OyZl8O5/UGUTJVXoWc= diff --git a/internal/cachedb/schema.sql b/internal/cachedb/schema.sql index 4d5d8bc..5125a6d 100644 --- a/internal/cachedb/schema.sql +++ b/internal/cachedb/schema.sql @@ -6,3 +6,19 @@ CREATE TABLE IF NOT EXISTS hostnames ( names TEXT NOT NULL, expires_at INTEGER NOT NULL ); + +-- geolocations holds the answers of geojs. A part geojs does not know is NULL. +-- fetched_at is in seconds since 1970. +CREATE TABLE IF NOT EXISTS geolocations ( + address TEXT PRIMARY KEY, + country_code TEXT, + country TEXT, + continent_code TEXT, + region TEXT, + city TEXT, + latitude REAL, + longitude REAL, + accuracy_radius_km INTEGER, + time_zone TEXT, + fetched_at INTEGER NOT NULL +); diff --git a/internal/geo/cache.go b/internal/geo/cache.go new file mode 100644 index 0000000..ff78311 --- /dev/null +++ b/internal/geo/cache.go @@ -0,0 +1,253 @@ +package geo + +import ( + "context" + "database/sql" + "errors" + "fmt" + "maps" + "time" +) + +const ( + // entryLifetime is how long a geojs answer is kept after its fetch. + entryLifetime = 7 * 24 * time.Hour + + // flushInterval is how often new answers are written to cache.db. + flushInterval = 5 * time.Second + + // sweepInterval is how often expired answers are deleted. + sweepInterval = 10 * time.Minute + + // rowColumns are the columns of the geolocations table in the order + // scanRow reads them. + rowColumns = "address, country_code, country, continent_code, region, " + + "city, latitude, longitude, accuracy_radius_km, time_zone, fetched_at" +) + +// rowScanner is the Scan method of *sql.Row and *sql.Rows. +type rowScanner interface { + Scan(dest ...any) error +} + +// cached returns the current answer held for address, from memory, then from +// the answers waiting to be written, then from cache.db. It returns nil when +// there is none. +func (l *Locator) cached(ctx context.Context, address string) *Location { + found, ok := l.entries.Get(address) + if ok { + if !l.expired(found) { + return found + } + + l.entries.Remove(address) + } + + l.pendingMu.Lock() + waiting := l.pending[address] + l.pendingMu.Unlock() + + if waiting != nil && !l.expired(waiting) { + l.entries.Add(address, waiting) + + return waiting + } + + stored, err := l.readRow(ctx, address) + if err != nil { + l.logger.Warn("Failed to read a geolocation cache row", + "address", address, "error", err) + + return nil + } + + if stored != nil { + l.entries.Add(address, stored) + } + + return stored +} + +// expired says whether the answer is at the end of its lifetime. +func (l *Locator) expired(answer *Location) bool { + return !l.now().Before(answer.Updated.Add(entryLifetime)) +} + +// cutoff returns the fetch time at or before which an answer has expired. +func (l *Locator) cutoff() int64 { + return l.now().Add(-entryLifetime).Unix() +} + +// store keeps an answer in memory and queues it for cache.db. While cache.db +// cannot be written, the queue stops growing at maxEntries answers. +func (l *Locator) store(address string, answer *Location) { + l.entries.Add(address, answer) + + l.pendingMu.Lock() + defer l.pendingMu.Unlock() + + if len(l.pending) < l.maxEntries { + l.pending[address] = answer + } +} + +// keepCache writes new answers and deletes expired ones until ctx ends. +func (l *Locator) keepCache(ctx context.Context) { + flushes := time.NewTicker(flushInterval) + defer flushes.Stop() + + sweeps := time.NewTicker(sweepInterval) + defer sweeps.Stop() + + for { + select { + case <-ctx.Done(): + l.flush(context.WithoutCancel(ctx)) + + return + case <-flushes.C: + l.flush(ctx) + case <-sweeps.C: + l.sweep(ctx) + } + } +} + +// load reads the answers that have not expired into memory, at most maxEntries +// of them, the newest last so that they are the last to be dropped. +func (l *Locator) load(ctx context.Context) error { + rows, err := l.db.QueryContext(ctx, + "SELECT "+rowColumns+" FROM (SELECT "+rowColumns+" FROM geolocations "+ + "WHERE fetched_at > ? ORDER BY fetched_at DESC LIMIT ?) "+ + "ORDER BY fetched_at", + l.cutoff(), l.maxEntries) + if err != nil { + return fmt.Errorf("reading the geolocation cache: %w", err) + } + + defer func() { _ = rows.Close() }() + + for rows.Next() { + address, answer, err := scanRow(rows) + if err != nil { + return fmt.Errorf("reading a geolocation cache row: %w", err) + } + + l.entries.Add(address, answer) + } + + err = rows.Err() + if err != nil { + return fmt.Errorf("reading the geolocation cache: %w", err) + } + + return nil +} + +// readRow returns the current answer for address in cache.db, or nil when it +// has none. +func (l *Locator) readRow( + ctx context.Context, address string, +) (*Location, error) { + row := l.db.QueryRowContext(ctx, + "SELECT "+rowColumns+" FROM geolocations "+ + "WHERE address = ? AND fetched_at > ?", + address, l.cutoff()) + + _, answer, err := scanRow(row) + if errors.Is(err, sql.ErrNoRows) { + return nil, nil //nolint:nilnil // no row is not a failure + } + + return answer, err +} + +// scanRow reads one row of the geolocations table. +func scanRow(row rowScanner) (string, *Location, error) { + var ( + address string + fetched int64 + answer Location + ) + + err := row.Scan(&address, &answer.CountryCode, &answer.Country, + &answer.ContinentCode, &answer.Region, &answer.City, &answer.Latitude, + &answer.Longitude, &answer.AccuracyRadiusKm, &answer.TimeZone, &fetched) + if err != nil { + return "", nil, err + } + + answer.Source = sourceGeoJS + answer.Updated = time.Unix(fetched, 0).UTC() + + return address, &answer, nil +} + +// flush writes the pending answers to cache.db. Those it cannot write stay +// pending, unless a newer answer for the address has arrived meanwhile. +func (l *Locator) flush(ctx context.Context) { + l.pendingMu.Lock() + batch := l.pending + l.pending = make(map[string]*Location) + l.pendingMu.Unlock() + + if len(batch) == 0 { + return + } + + err := l.writeRows(ctx, batch) + if err == nil { + return + } + + l.logger.Error("Failed to write the geolocation cache", "error", err) + + l.pendingMu.Lock() + defer l.pendingMu.Unlock() + + maps.Insert(batch, maps.All(l.pending)) + l.pending = batch +} + +// writeRows writes batch to cache.db in one transaction. +func (l *Locator) writeRows( + ctx context.Context, batch map[string]*Location, +) error { + tx, err := l.db.BeginTx(ctx, nil) + if err != nil { + return err + } + + defer func() { _ = tx.Rollback() }() + + for address, answer := range batch { + _, err = tx.ExecContext(ctx, + "INSERT OR REPLACE INTO geolocations ("+rowColumns+") "+ + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", + address, answer.CountryCode, answer.Country, answer.ContinentCode, + answer.Region, answer.City, answer.Latitude, answer.Longitude, + answer.AccuracyRadiusKm, answer.TimeZone, answer.Updated.Unix()) + if err != nil { + return err + } + } + + return tx.Commit() +} + +// sweep deletes the expired answers from memory and cache.db. +func (l *Locator) sweep(ctx context.Context) { + for _, address := range l.entries.Keys() { + answer, ok := l.entries.Peek(address) + if ok && l.expired(answer) { + l.entries.Remove(address) + } + } + + _, err := l.db.ExecContext(ctx, + "DELETE FROM geolocations WHERE fetched_at <= ?", l.cutoff()) + if err != nil { + l.logger.Error("Failed to delete expired geolocation cache rows", + "error", err) + } +} diff --git a/internal/geo/database.go b/internal/geo/database.go new file mode 100644 index 0000000..25528ad --- /dev/null +++ b/internal/geo/database.go @@ -0,0 +1,349 @@ +package geo + +import ( + "compress/gzip" + "context" + "errors" + "fmt" + "io" + "io/fs" + "net/http" + "net/netip" + "os" + "path/filepath" + "time" + + "github.com/oschwald/maxminddb-golang/v2" +) + +const ( + // DatabaseDownloadFormat is the URL of DB-IP's City Lite database, with %s + // for the year and month. + DatabaseDownloadFormat = "https://download.db-ip.com/free/" + + "dbip-city-lite-%s.mmdb.gz" + + // databaseMonthFormat is the year and month in the URL. + databaseMonthFormat = "2006-01" + + // databaseFileName is the copy of the database in the state directory. + databaseFileName = "dbip-city-lite.mmdb" + + // databaseDownloadSuffix ends the name of the file a download is written + // to, until it has opened. + databaseDownloadSuffix = ".download" + + // maxDatabaseSize is the largest database accepted once unpacked, in + // bytes. + maxDatabaseSize = 200 << 20 + + // databaseDownloadTimeout limits one download. + databaseDownloadTimeout = 10 * time.Minute + + // databaseCheckInterval is how often RouteWatch checks whether a new + // month's database is due. A new month's file appears some time on the + // first of the month, so a failed download is tried again this much later. + databaseCheckInterval = time.Hour +) + +var ( + errDatabaseStatus = errors.New("unexpected HTTP status from DB-IP") + errDatabaseTooLong = errors.New("database is over the size limit") +) + +// dbipName is the part of a database record that holds an English name. +type dbipName struct { + Names struct { + English string `maxminddb:"en"` + } `maxminddb:"names"` +} + +// dbipRecord is the part of a database record RouteWatch reads. +type dbipRecord struct { + City dbipName `maxminddb:"city"` + Continent struct { + Code string `maxminddb:"code"` + } `maxminddb:"continent"` + Country struct { + dbipName + + ISOCode string `maxminddb:"iso_code"` + } `maxminddb:"country"` + Location struct { + Latitude *float64 `maxminddb:"latitude"` + Longitude *float64 `maxminddb:"longitude"` + } `maxminddb:"location"` + Subdivisions []dbipName `maxminddb:"subdivisions"` +} + +// location returns the record as a Location fetched at the time given. The +// first subdivision is the region. The database has no accuracy radius and no +// time zone. +func (r *dbipRecord) location(fetched time.Time) *Location { + found := &Location{ + CountryCode: nilIfEmpty(r.Country.ISOCode), + Country: nilIfEmpty(r.Country.Names.English), + ContinentCode: nilIfEmpty(r.Continent.Code), + City: nilIfEmpty(r.City.Names.English), + Latitude: r.Location.Latitude, + Longitude: r.Location.Longitude, + Source: sourceDBIP, + Updated: fetched, + } + + if len(r.Subdivisions) > 0 { + found.Region = nilIfEmpty(r.Subdivisions[0].Names.English) + } + + return found +} + +// nilIfEmpty returns a pointer to text, or nil when text is empty. +func nilIfEmpty(text string) *string { + if text == "" { + return nil + } + + return &text +} + +// fromDatabase returns where the copy of the database places addr. It returns +// nil when there is no copy yet or the copy has no record for addr. It also +// returns nil, and logs the error, when reading the copy fails. +func (l *Locator) fromDatabase(addr netip.Addr) *Location { + l.databaseMu.RLock() + defer l.databaseMu.RUnlock() + + if l.database == nil { + return nil + } + + result := l.database.Lookup(addr) + + err := result.Err() + if err != nil { + l.logger.Error("Failed to look up an address in the database", + "address", addr, "error", err) + + return nil + } + + if !result.Found() { + return nil + } + + var record dbipRecord + + err = result.Decode(&record) + if err != nil { + l.logger.Error("Failed to read a database record", + "address", addr, "error", err) + + return nil + } + + return record.location(l.databaseFetched) +} + +// keepDatabaseCurrent downloads the database when it is due, at once and then +// every databaseCheckInterval, until ctx ends. +func (l *Locator) keepDatabaseCurrent(ctx context.Context) { + l.downloadDatabaseIfDue(ctx) + + checks := time.NewTicker(databaseCheckInterval) + defer checks.Stop() + + for { + select { + case <-ctx.Done(): + return + case <-checks.C: + l.downloadDatabaseIfDue(ctx) + } + } +} + +// downloadDatabaseIfDue downloads this month's database, unless the copy in +// use was downloaded this month. A failure leaves the copy as it is. +func (l *Locator) downloadDatabaseIfDue(ctx context.Context) { + now := l.now().UTC() + + if !l.databaseDue(now) { + return + } + + err := l.downloadDatabase(ctx, now) + if err != nil && ctx.Err() == nil { + l.logger.Error("Database download failed; keeping the previous copy", + "error", err) + } +} + +// databaseDue says whether there is no open copy or it is from an earlier +// month than now. +func (l *Locator) databaseDue(now time.Time) bool { + l.databaseMu.RLock() + defer l.databaseMu.RUnlock() + + return l.database == nil || + l.databaseFetched.Format(databaseMonthFormat) != + now.Format(databaseMonthFormat) +} + +// downloadDatabase fetches the database of now's month. The file replaces the +// copy, on disk and in use, only once it has opened. +func (l *Locator) downloadDatabase(ctx context.Context, now time.Time) error { + ctx, cancel := context.WithTimeout(ctx, databaseDownloadTimeout) + defer cancel() + + url := fmt.Sprintf(l.databaseURLFormat, now.Format(databaseMonthFormat)) + + request, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil) + if err != nil { + return err + } + + request.Header.Set(userAgentHeader, userAgent) + + response, err := l.client.Do(request) + if err != nil { + return err + } + + defer func() { _ = response.Body.Close() }() + + if response.StatusCode != http.StatusOK { + return fmt.Errorf("%w: %s", errDatabaseStatus, response.Status) + } + + return l.replaceDatabase(response.Body, now) +} + +// replaceDatabase unpacks the gzip stream body beside the copy, then makes the +// result the copy once it has opened. Whatever fails, the file it wrote is +// removed and the copy stays. +func (l *Locator) replaceDatabase(body io.Reader, now time.Time) error { + root, err := os.OpenRoot(l.dir) + if err != nil { + return err + } + + defer func() { _ = root.Close() }() + + downloadName := databaseFileName + databaseDownloadSuffix + + err = unpackDatabase(root, downloadName, body, l.maxDatabaseBytes) + if err != nil { + _ = root.Remove(downloadName) + + return err + } + + opened, err := maxminddb.Open(filepath.Join(l.dir, downloadName)) + if err != nil { + _ = root.Remove(downloadName) + + return fmt.Errorf("opening the downloaded database: %w", err) + } + + err = root.Rename(downloadName, databaseFileName) + if err != nil { + _ = opened.Close() + _ = root.Remove(downloadName) + + return err + } + + l.useDatabase(opened, now.Truncate(time.Second)) + l.logger.Info("Downloaded a new copy of the database") + + return nil +} + +// unpackDatabase writes the gzip stream body to the file name in root, up to +// limit bytes once unpacked. +func unpackDatabase( + root *os.Root, name string, body io.Reader, limit int64, +) error { + unpacked, err := gzip.NewReader(body) + if err != nil { + return fmt.Errorf("unpacking the database: %w", err) + } + + f, err := root.Create(name) + if err != nil { + return err + } + + defer func() { _ = f.Close() }() + + written, err := io.Copy(f, io.LimitReader(unpacked, limit+1)) + if err != nil { + return fmt.Errorf("unpacking the database: %w", err) + } + + if written > limit { + return errDatabaseTooLong + } + + return nil +} + +// openDatabaseCopy opens the copy left by an earlier run, if there is one. A +// copy that does not open is treated as no copy, so it is downloaded again. +func (l *Locator) openDatabaseCopy() { + path := filepath.Join(l.dir, databaseFileName) + + info, err := os.Stat(path) + if errors.Is(err, fs.ErrNotExist) { + return + } + + if err != nil { + l.logger.Error("Failed to read the database copy", "error", err) + + return + } + + opened, err := maxminddb.Open(path) + if err != nil { + l.logger.Error("Failed to open the database copy", "error", err) + + return + } + + l.useDatabase(opened, info.ModTime().UTC().Truncate(time.Second)) +} + +// useDatabase makes opened the copy in use and closes the one it replaces. +func (l *Locator) useDatabase(opened *maxminddb.Reader, fetched time.Time) { + l.databaseMu.Lock() + defer l.databaseMu.Unlock() + + if l.database != nil { + err := l.database.Close() + if err != nil { + l.logger.Error("Failed to close the old database copy", "error", err) + } + } + + l.database = opened + l.databaseFetched = fetched +} + +// closeDatabase closes the copy in use, after which no source can answer from +// it. +func (l *Locator) closeDatabase() { + l.databaseMu.Lock() + defer l.databaseMu.Unlock() + + if l.database == nil { + return + } + + err := l.database.Close() + if err != nil { + l.logger.Error("Failed to close the database copy", "error", err) + } + + l.database = nil +} diff --git a/internal/geo/geojs.go b/internal/geo/geojs.go new file mode 100644 index 0000000..f9c6520 --- /dev/null +++ b/internal/geo/geojs.go @@ -0,0 +1,151 @@ +package geo + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "io" + "math" + "net/http" + "net/netip" + "strconv" + "strings" + "time" +) + +const ( + // GeoJSEndpoint is the URL of the geojs geo endpoint. + GeoJSEndpoint = "https://get.geojs.io/v1/ip/geo.json" + + // geojsTimeLimit limits one request to geojs. + geojsTimeLimit = time.Second + + // geojsAddressSeparator separates the addresses of one request. + geojsAddressSeparator = "," + + // geojsUnknown is what geojs sends for a value it does not know. + geojsUnknown = "nil" + + // maxGeoJSAnswerBytes is far more than an answer for the most addresses + // an IP lookup can ask about. + maxGeoJSAnswerBytes = 1 << 20 +) + +var errGeoJSStatus = errors.New("unexpected HTTP status from geojs") + +// geojsAnswer is one address's entry in an answer from geojs. +// +//nolint:tagliatelle // the keys geojs sends +type geojsAnswer struct { + IP string `json:"ip"` + CountryCode string `json:"country_code"` + Country string `json:"country"` + ContinentCode string `json:"continent_code"` + Region string `json:"region"` + City string `json:"city"` + Latitude string `json:"latitude"` + Longitude string `json:"longitude"` + Accuracy *int `json:"accuracy"` + TimeZone string `json:"timezone"` +} + +// askGeoJS asks geojs about addresses in one request, and returns its answers +// by address. An address geojs left out, or gave no country, city or +// coordinates for, has no entry. +func (l *Locator) askGeoJS( + ctx context.Context, addresses []string, +) (map[string]*Location, error) { + ctx, cancel := context.WithTimeout(ctx, l.timeLimit) + defer cancel() + + // A comma and a colon are legal in a query, and the escaped forms are + // not what geojs documents. + url := l.geojsURL + "?ip=" + strings.Join(addresses, geojsAddressSeparator) + + request, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil) + if err != nil { + return nil, err + } + + request.Header.Set(userAgentHeader, userAgent) + + response, err := l.client.Do(request) + if err != nil { + return nil, err + } + + defer func() { _ = response.Body.Close() }() + + if response.StatusCode != http.StatusOK { + return nil, fmt.Errorf("%w: %d", errGeoJSStatus, response.StatusCode) + } + + var parsed []geojsAnswer + + err = json.NewDecoder(io.LimitReader(response.Body, maxGeoJSAnswerBytes)). + Decode(&parsed) + if err != nil { + return nil, fmt.Errorf("reading the geojs answer: %w", err) + } + + fetched := l.now().UTC().Truncate(time.Second) + answers := make(map[string]*Location, len(parsed)) + + for _, answer := range parsed { + addr, err := netip.ParseAddr(answer.IP) + if err != nil { + continue + } + + location := answer.location(fetched) + if hasLocation(location) { + answers[plainAddress(addr)] = location + } + } + + return answers, nil +} + +// hasLocation says whether found has a country, a city or a coordinate. +func hasLocation(found *Location) bool { + return found.CountryCode != nil || found.Country != nil || + found.City != nil || found.Latitude != nil || found.Longitude != nil +} + +// location returns the answer as a Location fetched at the time given. +func (a geojsAnswer) location(fetched time.Time) *Location { + return &Location{ + CountryCode: knownText(a.CountryCode), + Country: knownText(a.Country), + ContinentCode: knownText(a.ContinentCode), + Region: knownText(a.Region), + City: knownText(a.City), + Latitude: knownNumber(a.Latitude), + Longitude: knownNumber(a.Longitude), + AccuracyRadiusKm: a.Accuracy, + TimeZone: knownText(a.TimeZone), + Source: sourceGeoJS, + Updated: fetched, + } +} + +// knownText returns a pointer to text, or nil when geojs does not know it. +func knownText(text string) *string { + if text == "" || text == geojsUnknown { + return nil + } + + return &text +} + +// knownNumber returns text as a number, or nil when geojs does not know it. NaN +// and infinity parse, but JSON cannot carry them. +func knownNumber(text string) *float64 { + number, err := strconv.ParseFloat(text, 64) + if err != nil || math.IsNaN(number) || math.IsInf(number, 0) { + return nil + } + + return &number +} diff --git a/internal/geo/locator.go b/internal/geo/locator.go new file mode 100644 index 0000000..f167d69 --- /dev/null +++ b/internal/geo/locator.go @@ -0,0 +1,232 @@ +// Package geo finds where IP addresses are. It asks geojs.io, and keeps those +// answers in memory and in cache.db. When geojs.io gives no location for an +// address, it reads a copy of DB-IP's City Lite database, which it downloads +// once a month. +package geo + +import ( + "context" + "database/sql" + "fmt" + "log/slog" + "net/http" + "net/netip" + "sync" + "time" + + lru "github.com/hashicorp/golang-lru/v2" + "github.com/oschwald/maxminddb-golang/v2" +) + +// The values of Location.Source. +const ( + sourceGeoJS = "geojs" + sourceDBIP = "dbip" +) + +const ( + // maxEntries limits how many geojs answers are held in memory. + maxEntries = 250_000 + + // userAgentHeader and userAgent name routewatch to geojs and DB-IP. + userAgentHeader = "User-Agent" + userAgent = "routewatch (+https://git.eeqj.de/sneak/routewatch)" +) + +// Params is what New needs. +type Params struct { + // DB is the open cache.db. + DB *sql.DB + + // Dir is the directory that holds the copy of the database. + Dir string + + // GeoJSURL is the URL of the geojs geo endpoint, GeoJSEndpoint in use. + GeoJSURL string + + // DatabaseURLFormat is the URL of the database with %s for the year and + // month, DatabaseDownloadFormat in use. + DatabaseURLFormat string + + Logger *slog.Logger +} + +// Location is where an address is. A part the source does not know is nil. +// +//nolint:tagliatelle // snake_case keys, as the Go style guide asks +type Location struct { + CountryCode *string `json:"country_code"` + Country *string `json:"country"` + ContinentCode *string `json:"continent_code"` + Region *string `json:"region"` + City *string `json:"city"` + Latitude *float64 `json:"latitude"` + Longitude *float64 `json:"longitude"` + AccuracyRadiusKm *int `json:"accuracy_radius_km"` + TimeZone *string `json:"time_zone"` + + // Source is geojs or dbip. + Source string `json:"source"` + + // Updated is when the value was fetched: the geojs answer, or the copy of + // the database. + Updated time.Time `json:"updated"` +} + +// Locator answers where addresses are. Start loads its cache and starts its +// goroutines. +type Locator struct { + db *sql.DB + dir string + geojsURL string + databaseURLFormat string + timeLimit time.Duration + maxEntries int + maxDatabaseBytes int64 + logger *slog.Logger + client *http.Client + now func() time.Time + + // entries holds geojs answers by address, and drops the least recently + // used first. + entries *lru.Cache[string, *Location] + + // pending holds the answers that cache.db does not have yet. + pendingMu sync.Mutex + pending map[string]*Location + + // database is the open copy, nil until one has opened, and databaseFetched + // is when that copy was downloaded. A lookup holds the read lock, so that + // a new copy never closes the old one under it. + databaseMu sync.RWMutex + database *maxminddb.Reader + databaseFetched time.Time + + // stop ends the goroutines Start runs, and done is closed when they have + // returned. + stop context.CancelFunc + done chan struct{} +} + +// New returns a Locator with an empty cache. Start loads it. +func New(params Params) (*Locator, error) { + entries, err := lru.New[string, *Location](maxEntries) + if err != nil { + return nil, fmt.Errorf("creating the geolocation cache: %w", err) + } + + return &Locator{ + db: params.DB, + dir: params.Dir, + geojsURL: params.GeoJSURL, + databaseURLFormat: params.DatabaseURLFormat, + timeLimit: geojsTimeLimit, + maxEntries: maxEntries, + maxDatabaseBytes: maxDatabaseSize, + logger: params.Logger.With("component", "geo"), + client: &http.Client{}, + now: time.Now, + entries: entries, + pending: make(map[string]*Location), + done: make(chan struct{}), + }, nil +} + +// Lookup returns where each of addrs is, in the same order. The entry for an +// address neither source answers for is nil. Addresses with no current cached +// answer go to geojs in one request. The returned Locations are shared with +// the cache and must not be changed. +func (l *Locator) Lookup(ctx context.Context, addrs []netip.Addr) []*Location { + locations := make([]*Location, len(addrs)) + + var unanswered []string + + for i, addr := range addrs { + address := plainAddress(addr) + + locations[i] = l.cached(ctx, address) + if locations[i] == nil { + unanswered = append(unanswered, address) + } + } + + if len(unanswered) == 0 { + return locations + } + + answers, err := l.askGeoJS(ctx, unanswered) + if err != nil { + l.logger.Warn("geojs lookup failed; using the database", "error", err) + } + + for i, addr := range addrs { + if locations[i] != nil { + continue + } + + address := plainAddress(addr) + + answer, ok := answers[address] + if ok { + l.store(address, answer) + locations[i] = answer + + continue + } + + locations[i] = l.fromDatabase(addr) + } + + return locations +} + +// plainAddress returns addr as the requests to geojs and the cache name it: an +// IPv4 address mapped into IPv6 as plain IPv4, and with no zone. A client +// chooses the zone, so it must not reach the request. +func plainAddress(addr netip.Addr) string { + return addr.Unmap().WithZone("").String() +} + +// Start loads the cached answers that have not expired and opens the copy of +// the database, if there is one. It then starts the goroutines that write new +// answers, delete expired ones and download the database. They run until Stop: +// the end of ctx does not stop them. +func (l *Locator) Start(ctx context.Context) error { + err := l.load(ctx) + if err != nil { + return err + } + + l.logger.Info("Geolocation cache loaded", "entries", l.entries.Len()) + + l.openDatabaseCopy() + + ctx, l.stop = context.WithCancel(context.WithoutCancel(ctx)) + + go func() { + defer close(l.done) + defer l.closeDatabase() + + var workers sync.WaitGroup + + workers.Go(func() { l.keepCache(ctx) }) + workers.Go(func() { l.keepDatabaseCurrent(ctx) }) + workers.Wait() + }() + + return nil +} + +// Stop cancels a download in progress, writes the answers cache.db does not +// have yet, and waits until the goroutines have returned or ctx ends. +func (l *Locator) Stop(ctx context.Context) { + l.stop() + + select { + case <-l.done: + l.logger.Info("Geolocation stopped") + case <-ctx.Done(): + l.logger.Warn("Stop deadline passed before the geolocation cache " + + "was written") + } +} diff --git a/internal/geo/locator_internal_test.go b/internal/geo/locator_internal_test.go new file mode 100644 index 0000000..17cc853 --- /dev/null +++ b/internal/geo/locator_internal_test.go @@ -0,0 +1,1009 @@ +package geo + +import ( + "bytes" + "compress/gzip" + "context" + "database/sql" + "errors" + "io/fs" + "log/slog" + "net" + "net/http" + "net/http/httptest" + "net/netip" + "os" + "path/filepath" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "git.eeqj.de/sneak/routewatch/internal/cachedb" + lru "github.com/hashicorp/golang-lru/v2" + "github.com/maxmind/mmdbwriter" + "github.com/maxmind/mmdbwriter/mmdbtype" +) + +// The addresses and networks of the tests, from the documentation ranges. +const ( + testAddress = "192.0.2.10" + otherAddress = "2001:db8::10" + thirdAddress = "203.0.113.10" + coveredNetwork = "192.0.2.0/24" + testCity = "Testville" + databaseCity = "Databaseville" + databaseMonth = "2026-01" + databasePath = "/dbip-city-lite-" + databaseMonth + ".mmdb.gz" + geojsPath = "/geo.json" + + // testTimeLimit replaces geojsTimeLimit in the tests that use a real + // server, so that a stall on a busy host cannot use it up. + testTimeLimit = 2 * time.Minute + + // shortTimeLimit is the limit of the test in which geojs never answers. + shortTimeLimit = 50 * time.Millisecond +) + +// The bodies the fake geojs sends. +const ( + geojsTwoAddresses = `[ + {"accuracy": 20, "city": "` + testCity + `", "continent_code": "EU", + "country": "Germany", "country_code": "DE", "ip": "` + testAddress + `", + "latitude": "52.6171", "longitude": "13.1207", + "region": "Brandenburg", "timezone": "Europe/Berlin"}, + {"country": "Germany", "ip": "` + otherAddress + `", + "latitude": "nil", "longitude": "nil"} + ]` + geojsOneAddress = `[{"ip": "` + testAddress + `", "city": "` + testCity + `"}]` + + // geojsNoLocation has no country, city or coordinates, as geojs sends for + // some addresses. Its other parts do not count as a location. + geojsNoLocation = `[{"accuracy": 1000, "area_code": "0", "asn": 64496, + "continent_code": "EU", "ip": "` + testAddress + `", + "latitude": "nil", "longitude": "nil", + "organization": "AS64496 Example", "organization_name": "Example", + "region": "Brandenburg", "timezone": "Europe/Berlin"}]` +) + +// fixedTime returns the time a test's clock starts at. +func fixedTime() time.Time { + return time.Date(2026, time.January, 2, 3, 4, 5, 0, time.UTC) +} + +// parseAddresses returns the given addresses parsed. +func parseAddresses(t *testing.T, addresses ...string) []netip.Addr { + t.Helper() + + addrs := make([]netip.Addr, len(addresses)) + + for i, address := range addresses { + addr, err := netip.ParseAddr(address) + if err != nil { + t.Fatalf("parsing %s: %v", address, err) + } + + addrs[i] = addr + } + + return addrs +} + +// fakeGeoJS is a geojs server. It sends body, or status when that is set, or +// waits for the request to end when hang is true. It keeps each request's +// address list. +type fakeGeoJS struct { + server *httptest.Server + body string + status int + hang bool + + mu sync.Mutex + requests []string +} + +// newFakeGeoJS starts a fake geojs that sends body. +func newFakeGeoJS(t *testing.T, body string) *fakeGeoJS { + t.Helper() + + fake := &fakeGeoJS{body: body} + fake.server = httptest.NewServer(http.HandlerFunc(fake.serve)) + t.Cleanup(fake.server.Close) + + return fake +} + +func (f *fakeGeoJS) serve(w http.ResponseWriter, r *http.Request) { + f.mu.Lock() + f.requests = append(f.requests, r.URL.Query().Get("ip")) + f.mu.Unlock() + + if f.hang { + <-r.Context().Done() + + return + } + + if f.status != 0 { + w.WriteHeader(f.status) + + return + } + + _, _ = w.Write([]byte(f.body)) +} + +// requestsSeen returns the address list of each request so far. +func (f *fakeGeoJS) requestsSeen() []string { + f.mu.Lock() + defer f.mu.Unlock() + + return append([]string{}, f.requests...) +} + +// openCacheDB opens a new cache.db in a temporary directory. +func openCacheDB(t *testing.T, dir string) *sql.DB { + t.Helper() + + db, err := cachedb.Open(t.Context(), dir) + if err != nil { + t.Fatalf("opening cache.db: %v", err) + } + + t.Cleanup(func() { _ = db.Close() }) + + return db +} + +// newTestLocator returns a Locator that asks fake and uses db and dir, with +// its clock stopped at fixedTime. +func newTestLocator( + t *testing.T, fake *fakeGeoJS, db *sql.DB, dir string, +) *Locator { + t.Helper() + + locator, err := New(Params{ + DB: db, + Dir: dir, + GeoJSURL: fake.server.URL + geojsPath, + DatabaseURLFormat: fake.server.URL + "/dbip-city-lite-%s.mmdb.gz", + Logger: slog.New(slog.DiscardHandler), + }) + if err != nil { + t.Fatalf("New: %v", err) + } + + locator.timeLimit = testTimeLimit + locator.now = fixedTime + + t.Cleanup(locator.closeDatabase) + + return locator +} + +// lookupOne returns the location of one address. +func lookupOne(t *testing.T, locator *Locator, address string) *Location { + t.Helper() + + return locator.Lookup(t.Context(), parseAddresses(t, address))[0] +} + +// databaseFile returns an MMDB file with the shape of City Lite. It covers +// coveredNetwork, with no time zone or accuracy radius, as the real file has +// none. +func databaseFile(t *testing.T) []byte { + t.Helper() + + tree, err := mmdbwriter.New(mmdbwriter.Options{ + DatabaseType: "DBIP-City-Lite", + IncludeReservedNetworks: true, + }) + if err != nil { + t.Fatalf("creating the test database: %v", err) + } + + _, network, err := net.ParseCIDR(coveredNetwork) + if err != nil { + t.Fatalf("parsing the network: %v", err) + } + + names := func(name string) mmdbtype.Map { + return mmdbtype.Map{"names": mmdbtype.Map{"en": mmdbtype.String(name)}} + } + + country := names("Germany") + country["iso_code"] = mmdbtype.String("DE") + + err = tree.Insert(network, mmdbtype.Map{ + "city": names(databaseCity), + "continent": mmdbtype.Map{"code": mmdbtype.String("EU")}, + "country": country, + "subdivisions": mmdbtype.Slice{names("Brandenburg")}, + "location": mmdbtype.Map{ + "latitude": mmdbtype.Float64(52.5), + "longitude": mmdbtype.Float64(13.5), + }, + }) + if err != nil { + t.Fatalf("inserting the test record: %v", err) + } + + var file bytes.Buffer + + _, err = tree.WriteTo(&file) + if err != nil { + t.Fatalf("writing the test database: %v", err) + } + + return file.Bytes() +} + +// gzipped returns data compressed with gzip. +func gzipped(t *testing.T, data []byte) []byte { + t.Helper() + + var packed bytes.Buffer + + writer := gzip.NewWriter(&packed) + + _, err := writer.Write(data) + if err != nil { + t.Fatalf("compressing: %v", err) + } + + err = writer.Close() + if err != nil { + t.Fatalf("compressing: %v", err) + } + + return packed.Bytes() +} + +// giveDatabase puts a copy of the test database in the locator's directory, and +// opens it. +func giveDatabase(t *testing.T, locator *Locator) { + t.Helper() + + path := filepath.Join(locator.dir, databaseFileName) + + err := os.WriteFile(path, databaseFile(t), 0o600) + if err != nil { + t.Fatalf("writing the database copy: %v", err) + } + + locator.openDatabaseCopy() +} + +// requireCity checks where a location says it is, and its source. +func requireCity(t *testing.T, got *Location, city, source string) { + t.Helper() + + if got == nil { + t.Fatal("no location") + } + + if got.City == nil || *got.City != city { + t.Errorf("city %v, want %s", got.City, city) + } + + if got.Source != source { + t.Errorf("source %s, want %s", got.Source, source) + } +} + +// TestLookupAsksGeoJSOnceForEveryAddress looks up two addresses, and expects a +// single request that names both, and each answer in its own place. +func TestLookupAsksGeoJSOnceForEveryAddress(t *testing.T) { + t.Parallel() + + fake := newFakeGeoJS(t, geojsTwoAddresses) + dir := t.TempDir() + locator := newTestLocator(t, fake, openCacheDB(t, dir), dir) + + got := locator.Lookup(t.Context(), + parseAddresses(t, otherAddress, testAddress)) + + requests := fake.requestsSeen() + if len(requests) != 1 || requests[0] != otherAddress+","+testAddress { + t.Fatalf("requests %v, want one naming both addresses", requests) + } + + first := got[1] + requireCity(t, first, testCity, sourceGeoJS) + + if *first.CountryCode != "DE" || *first.Country != "Germany" || + *first.ContinentCode != "EU" || *first.Region != "Brandenburg" || + *first.Latitude != 52.6171 || *first.Longitude != 13.1207 || + *first.AccuracyRadiusKm != 20 || *first.TimeZone != "Europe/Berlin" { + t.Errorf("wrong location %+v", first) + } + + if !first.Updated.Equal(fixedTime()) { + t.Errorf("updated %v, want %v", first.Updated, fixedTime()) + } +} + +// TestLookupGivesUnknownPartsAsNil checks an answer with only a country name, +// whose coordinates are the string "nil": the other parts are nil. +func TestLookupGivesUnknownPartsAsNil(t *testing.T) { + t.Parallel() + + fake := newFakeGeoJS(t, geojsTwoAddresses) + dir := t.TempDir() + locator := newTestLocator(t, fake, openCacheDB(t, dir), dir) + + got := lookupOne(t, locator, otherAddress) + if got == nil { + t.Fatal("no location for an address geojs answered") + } + + if got.Latitude != nil || got.Longitude != nil || got.City != nil || + got.CountryCode != nil || got.AccuracyRadiusKm != nil || + got.TimeZone != nil { + t.Errorf("unknown parts are not nil: %+v", got) + } + + if got.Source != sourceGeoJS { + t.Errorf("source %s, want %s", got.Source, sourceGeoJS) + } +} + +// TestLookupRejectsNumbersJSONCannotCarry checks that a coordinate that is not +// a finite number is nil. +func TestLookupRejectsNumbersJSONCannotCarry(t *testing.T) { + t.Parallel() + + for _, text := range []string{"NaN", "Inf", "-Inf", "not a number", ""} { + if got := knownNumber(text); got != nil { + t.Errorf("knownNumber(%q) = %v, want nil", text, *got) + } + } +} + +// TestLookupFallsBackToTheDatabase makes geojs fail in each way the fallback +// covers, and expects the database's answer. +func TestLookupFallsBackToTheDatabase(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + body string + status int + hang bool + }{ + {name: "timeout", hang: true}, + {name: "error status", status: http.StatusBadGateway}, + {name: "unreadable answer", body: "not JSON"}, + {name: "address left out", body: "[]"}, + {name: "answer with no location", body: geojsNoLocation}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + fake := newFakeGeoJS(t, tt.body) + fake.status = tt.status + fake.hang = tt.hang + dir := t.TempDir() + locator := newTestLocator(t, fake, openCacheDB(t, dir), dir) + locator.timeLimit = shortTimeLimit + + if !tt.hang { + locator.timeLimit = testTimeLimit + } + + giveDatabase(t, locator) + + got := lookupOne(t, locator, testAddress) + requireCity(t, got, databaseCity, sourceDBIP) + + if *got.CountryCode != "DE" || *got.Country != "Germany" || + *got.ContinentCode != "EU" || *got.Region != "Brandenburg" || + *got.Latitude != 52.5 || *got.Longitude != 13.5 { + t.Errorf("wrong location %+v", got) + } + + if got.TimeZone != nil || got.AccuracyRadiusKm != nil { + t.Errorf("parts the database lacks are not nil: %+v", got) + } + + // The fallback is local, so it is not cached: the next lookup + // asks geojs again. + lookupOne(t, locator, testAddress) + + if requests := fake.requestsSeen(); len(requests) != 2 { + t.Errorf("%d requests, want 2", len(requests)) + } + }) + } +} + +// TestAnyCountryCityOrCoordinateIsAnAnswer checks that each of those parts, +// alone, makes an answer from geojs that is returned and kept. +func TestAnyCountryCityOrCoordinateIsAnAnswer(t *testing.T) { + t.Parallel() + + parts := []string{ + `"country": "Germany"`, + `"country_code": "DE"`, + `"city": "` + testCity + `"`, + `"latitude": "52.5"`, + `"longitude": "13.5"`, + } + + for _, part := range parts { + t.Run(part, func(t *testing.T) { + t.Parallel() + + fake := newFakeGeoJS(t, `[{"ip": "`+testAddress+`", `+part+`}]`) + dir := t.TempDir() + locator := newTestLocator(t, fake, openCacheDB(t, dir), dir) + + got := lookupOne(t, locator, testAddress) + if got == nil || got.Source != sourceGeoJS { + t.Fatalf("location %+v, want the answer from geojs", got) + } + + if !locator.entries.Contains(testAddress) { + t.Error("the answer is not in memory") + } + }) + } +} + +// TestLookupSendsNoZoneToGeoJS looks up an address with a zone, which is text +// a client chooses. geojs receives only the plain address, and the answer is +// kept under it. +func TestLookupSendsNoZoneToGeoJS(t *testing.T) { + t.Parallel() + + fake := newFakeGeoJS(t, geojsTwoAddresses) + dir := t.TempDir() + locator := newTestLocator(t, fake, openCacheDB(t, dir), dir) + + got := lookupOne(t, locator, otherAddress+"%x&callback=evil") + + requests := fake.requestsSeen() + if len(requests) != 1 || requests[0] != otherAddress { + t.Errorf("requests %v, want one naming only %s", requests, otherAddress) + } + + if got == nil || got.Source != sourceGeoJS { + t.Errorf("location %+v, want the answer from geojs", got) + } + + if !locator.entries.Contains(otherAddress) { + t.Error("the answer is not kept under the plain address") + } +} + +// TestFailedDatabaseLookupIsLogged closes the copy under the locator, so that +// the lookup in it fails. The location is nil and the failure is logged. +func TestFailedDatabaseLookupIsLogged(t *testing.T) { + t.Parallel() + + fake := newFakeGeoJS(t, "[]") + dir := t.TempDir() + locator := newTestLocator(t, fake, openCacheDB(t, dir), dir) + giveDatabase(t, locator) + + err := locator.database.Close() + if err != nil { + t.Fatalf("closing the copy: %v", err) + } + + var logged bytes.Buffer + + locator.logger = slog.New(slog.NewTextHandler(&logged, nil)) + + if got := locator.fromDatabase(parseAddresses(t, testAddress)[0]); got != nil { + t.Errorf("location %+v from a failed lookup, want nil", got) + } + + if !strings.Contains(logged.String(), "address="+testAddress) { + t.Errorf("the failure was not logged: %q", logged.String()) + } +} + +// TestLookupIsNilWhenNeitherSourceAnswers makes geojs fail with no database +// copy, and then with a copy that does not cover the address. +func TestLookupIsNilWhenNeitherSourceAnswers(t *testing.T) { + t.Parallel() + + fake := newFakeGeoJS(t, "") + fake.status = http.StatusBadGateway + dir := t.TempDir() + locator := newTestLocator(t, fake, openCacheDB(t, dir), dir) + + if got := lookupOne(t, locator, testAddress); got != nil { + t.Errorf("location %+v before the first download, want nil", got) + } + + giveDatabase(t, locator) + + if got := lookupOne(t, locator, thirdAddress); got != nil { + t.Errorf("location %+v for an address outside the database, want nil", got) + } +} + +// TestAnsweredAddressesAreNotAskedAgain checks that an address held in memory +// makes no request, and that only the others go to geojs. +func TestAnsweredAddressesAreNotAskedAgain(t *testing.T) { + t.Parallel() + + fake := newFakeGeoJS(t, geojsTwoAddresses) + dir := t.TempDir() + locator := newTestLocator(t, fake, openCacheDB(t, dir), dir) + + lookupOne(t, locator, testAddress) + + second := locator.Lookup(t.Context(), + parseAddresses(t, testAddress, otherAddress)) + + requireCity(t, second[0], testCity, sourceGeoJS) + + requests := fake.requestsSeen() + if len(requests) != 2 || requests[1] != otherAddress { + t.Errorf("requests %v, want the second to name only %s", + requests, otherAddress) + } +} + +// TestCacheSurvivesARestart writes an answer, then starts a new Locator on the +// same cache.db. It answers with no request, each part as it was. +func TestCacheSurvivesARestart(t *testing.T) { + t.Parallel() + + fake := newFakeGeoJS(t, geojsTwoAddresses) + dir := t.TempDir() + db := openCacheDB(t, dir) + locator := newTestLocator(t, fake, db, dir) + + want := locator.Lookup(t.Context(), + parseAddresses(t, testAddress, otherAddress)) + locator.flush(t.Context()) + + restarted := newTestLocator(t, fake, db, dir) + + err := restarted.load(t.Context()) + if err != nil { + t.Fatalf("load: %v", err) + } + + got := restarted.Lookup(t.Context(), + parseAddresses(t, testAddress, otherAddress)) + + if requests := fake.requestsSeen(); len(requests) != 1 { + t.Errorf("%d requests, want only the first", len(requests)) + } + + requireCity(t, got[0], testCity, sourceGeoJS) + + if *got[0].Latitude != *want[0].Latitude || + *got[0].AccuracyRadiusKm != *want[0].AccuracyRadiusKm || + !got[0].Updated.Equal(want[0].Updated) { + t.Errorf("reloaded %+v, want %+v", got[0], want[0]) + } + + if got[1].Latitude != nil || got[1].City != nil { + t.Errorf("unknown parts came back set: %+v", got[1]) + } +} + +// TestStopWritesWhatIsPending starts a Locator, takes an answer and stops it. +// The answer is in cache.db afterwards. +func TestStopWritesWhatIsPending(t *testing.T) { + t.Parallel() + + fake := newFakeGeoJS(t, geojsTwoAddresses) + dir := t.TempDir() + db := openCacheDB(t, dir) + locator := newTestLocator(t, fake, db, dir) + + err := locator.Start(t.Context()) + if err != nil { + t.Fatalf("Start: %v", err) + } + + lookupOne(t, locator, testAddress) + locator.Stop(t.Context()) + + stored, err := locator.readRow(t.Context(), testAddress) + if err != nil { + t.Fatalf("readRow: %v", err) + } + + requireCity(t, stored, testCity, sourceGeoJS) +} + +// TestAnswersExpireAfterSevenDays checks an answer just before and at the end +// of its lifetime, and that the sweep deletes its row and memory entry. +func TestAnswersExpireAfterSevenDays(t *testing.T) { + t.Parallel() + + fake := newFakeGeoJS(t, geojsOneAddress) + dir := t.TempDir() + db := openCacheDB(t, dir) + locator := newTestLocator(t, fake, db, dir) + + lookupOne(t, locator, testAddress) + locator.flush(t.Context()) + + locator.now = func() time.Time { return fixedTime().Add(entryLifetime - time.Second) } + lookupOne(t, locator, testAddress) + + if requests := fake.requestsSeen(); len(requests) != 1 { + t.Fatalf("%d requests just before expiry, want 1", len(requests)) + } + + expiry := fixedTime().Add(entryLifetime) + locator.now = func() time.Time { return expiry } + + locator.sweep(t.Context()) + + if locator.entries.Len() != 0 { + t.Errorf("%d entries in memory after the sweep, want 0", locator.entries.Len()) + } + + var rows int + + err := db.QueryRowContext(t.Context(), + "SELECT COUNT(*) FROM geolocations").Scan(&rows) + if err != nil { + t.Fatalf("counting rows: %v", err) + } + + if rows != 0 { + t.Errorf("%d rows after the sweep, want 0", rows) + } + + lookupOne(t, locator, testAddress) + + if requests := fake.requestsSeen(); len(requests) != 2 { + t.Errorf("%d requests after expiry, want 2", len(requests)) + } +} + +// limitMemory makes the locator's memory hold at most size answers. +func limitMemory(t *testing.T, locator *Locator, size int) { + t.Helper() + + entries, err := lru.New[string, *Location](size) + if err != nil { + t.Fatalf("lru.New: %v", err) + } + + locator.entries = entries + locator.maxEntries = size +} + +// answerFor returns a location for store to keep, fetched at the time given. +func answerFor(fetched time.Time) *Location { + return &Location{City: new(testCity), Source: sourceGeoJS, Updated: fetched} +} + +// TestMemoryDropsTheLeastRecentlyUsedAnswer fills memory with two answers, +// uses the older one and stores a third. The unused one leaves memory, and its +// row still answers without a request. +func TestMemoryDropsTheLeastRecentlyUsedAnswer(t *testing.T) { + t.Parallel() + + fake := newFakeGeoJS(t, "[]") + dir := t.TempDir() + locator := newTestLocator(t, fake, openCacheDB(t, dir), dir) + limitMemory(t, locator, 2) + + locator.store(testAddress, answerFor(fixedTime())) + locator.store(otherAddress, answerFor(fixedTime())) + locator.flush(t.Context()) + + lookupOne(t, locator, testAddress) + locator.store(thirdAddress, answerFor(fixedTime())) + + if locator.entries.Contains(otherAddress) { + t.Fatal("the least recently used answer stayed in memory") + } + + if !locator.entries.Contains(testAddress) { + t.Fatal("the recently used answer left memory") + } + + requireCity(t, lookupOne(t, locator, otherAddress), testCity, sourceGeoJS) + + if requests := fake.requestsSeen(); len(requests) != 0 { + t.Errorf("requests %v, want the row to answer", requests) + } +} + +// TestLoadKeepsTheNewestAnswers loads more rows than memory holds. The newest +// are in memory, and the newest of all is the last to be dropped. +func TestLoadKeepsTheNewestAnswers(t *testing.T) { + t.Parallel() + + fake := newFakeGeoJS(t, "[]") + dir := t.TempDir() + locator := newTestLocator(t, fake, openCacheDB(t, dir), dir) + + locator.store(testAddress, answerFor(fixedTime().Add(-3*time.Hour))) + locator.store(otherAddress, answerFor(fixedTime().Add(-2*time.Hour))) + locator.store(thirdAddress, answerFor(fixedTime().Add(-time.Hour))) + locator.flush(t.Context()) + + limitMemory(t, locator, 2) + + err := locator.load(t.Context()) + if err != nil { + t.Fatalf("load: %v", err) + } + + if locator.entries.Contains(testAddress) { + t.Error("the oldest answer was loaded") + } + + locator.entries.Add("192.0.2.99", answerFor(fixedTime())) + + if !locator.entries.Contains(thirdAddress) { + t.Error("the newest answer was the first to be dropped") + } +} + +// fakeDownload serves file for the January 2026 database and counts requests. +func fakeDownload(t *testing.T, file []byte) *httptest.Server { + t.Helper() + + server := httptest.NewServer(http.HandlerFunc( + func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != databasePath { + http.NotFound(w, r) + + return + } + + _, _ = w.Write(file) + })) + t.Cleanup(server.Close) + + return server +} + +// newDownloadLocator returns a locator that downloads file. +func newDownloadLocator(t *testing.T, file []byte) *Locator { + t.Helper() + + fake := newFakeGeoJS(t, "[]") + fake.status = http.StatusBadGateway + dir := t.TempDir() + locator := newTestLocator(t, fake, openCacheDB(t, dir), dir) + locator.databaseURLFormat = fakeDownload(t, file).URL + + "/dbip-city-lite-%s.mmdb.gz" + + return locator +} + +// requireNoDownloadFile checks that a download left no temporary file. +func requireNoDownloadFile(t *testing.T, locator *Locator) { + t.Helper() + + _, err := os.Stat(filepath.Join(locator.dir, + databaseFileName+databaseDownloadSuffix)) + if !errors.Is(err, fs.ErrNotExist) { + t.Errorf("the download's file is still there: %v", err) + } +} + +// TestDownloadPutsTheDatabaseInUse downloads this month's file, unpacks it, and +// answers from it. A restart opens the copy. +func TestDownloadPutsTheDatabaseInUse(t *testing.T) { + t.Parallel() + + file := gzipped(t, databaseFile(t)) + locator := newDownloadLocator(t, file) + + err := locator.downloadDatabase(t.Context(), fixedTime()) + if err != nil { + t.Fatalf("downloadDatabase: %v", err) + } + + requireNoDownloadFile(t, locator) + + got := lookupOne(t, locator, testAddress) + requireCity(t, got, databaseCity, sourceDBIP) + + if !got.Updated.Equal(fixedTime().Truncate(time.Second)) { + t.Errorf("updated %v, want the download's time", got.Updated) + } + + restarted := newDownloadLocator(t, file) + restarted.dir = locator.dir + restarted.openDatabaseCopy() + requireCity(t, lookupOne(t, restarted, testAddress), databaseCity, sourceDBIP) +} + +// TestDownloadIsDueOnceAMonth checks when a copy needs replacing: with no copy, +// not again in the month it was fetched, and in the next month. +func TestDownloadIsDueOnceAMonth(t *testing.T) { + t.Parallel() + + locator := newDownloadLocator(t, gzipped(t, databaseFile(t))) + + if !locator.databaseDue(fixedTime()) { + t.Error("no download due with no copy") + } + + err := locator.downloadDatabase(t.Context(), fixedTime()) + if err != nil { + t.Fatalf("downloadDatabase: %v", err) + } + + if locator.databaseDue(fixedTime().AddDate(0, 0, 20)) { + t.Error("a download is due in the same month") + } + + if !locator.databaseDue(fixedTime().AddDate(0, 1, 0)) { + t.Error("no download due in the next month") + } +} + +// TestFailedDownloadKeepsTheCopy serves a copy that cannot replace the one in +// use, in each way, and expects the old copy to go on answering. +func TestFailedDownloadKeepsTheCopy(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + file func(t *testing.T) []byte + max int64 + }{ + { + name: "not gzip", + file: func(*testing.T) []byte { return []byte("not gzip") }, + }, + { + name: "not a database", + file: func(t *testing.T) []byte { + t.Helper() + + return gzipped(t, []byte("not a database")) + }, + }, + { + name: "cut off", + file: func(t *testing.T) []byte { + t.Helper() + + packed := gzipped(t, databaseFile(t)) + + return packed[:len(packed)/2] + }, + }, + { + name: "over the size limit", + file: func(t *testing.T) []byte { + t.Helper() + + return gzipped(t, databaseFile(t)) + }, + max: 10, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + locator := newDownloadLocator(t, tt.file(t)) + if tt.max != 0 { + locator.maxDatabaseBytes = tt.max + } + + giveDatabase(t, locator) + + err := locator.downloadDatabase(t.Context(), fixedTime()) + if err == nil { + t.Fatal("the download succeeded") + } + + requireNoDownloadFile(t, locator) + requireCity(t, lookupOne(t, locator, testAddress), + databaseCity, sourceDBIP) + }) + } +} + +// TestMissingFileIsAFailedDownload checks that a month whose file is not +// published yet is an error and leaves no copy. +func TestMissingFileIsAFailedDownload(t *testing.T) { + t.Parallel() + + locator := newDownloadLocator(t, nil) + + err := locator.downloadDatabase(t.Context(), fixedTime().AddDate(0, 1, 0)) + if !errors.Is(err, errDatabaseStatus) { + t.Fatalf("error %v, want %v", err, errDatabaseStatus) + } + + if !locator.databaseDue(fixedTime()) { + t.Error("a copy is in use after a failed download") + } +} + +// TestUnopenableCopyIsDownloadedAgain leaves a damaged copy from this month. It +// counts as no copy, so the download is due. +func TestUnopenableCopyIsDownloadedAgain(t *testing.T) { + t.Parallel() + + locator := newDownloadLocator(t, nil) + + err := os.WriteFile(filepath.Join(locator.dir, databaseFileName), + []byte("damaged"), 0o600) + if err != nil { + t.Fatalf("writing the copy: %v", err) + } + + locator.openDatabaseCopy() + + if !locator.databaseDue(fixedTime()) { + t.Error("no download due with a copy that does not open") + } +} + +// TestLookupDuringReplaceIsSafe looks up addresses while the copy is replaced. +// The race detector and the mapped file both catch a lookup of a closed copy. +func TestLookupDuringReplaceIsSafe(t *testing.T) { + t.Parallel() + + file := gzipped(t, databaseFile(t)) + locator := newDownloadLocator(t, file) + giveDatabase(t, locator) + + addrs := parseAddresses(t, testAddress) + + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + + var readers sync.WaitGroup + + readers.Go(func() { + for ctx.Err() == nil { + locator.fromDatabase(addrs[0]) + } + }) + + for range 5 { + err := locator.downloadDatabase(t.Context(), fixedTime()) + if err != nil { + t.Fatalf("downloadDatabase: %v", err) + } + } + + cancel() + readers.Wait() +} + +// TestRequestNamesRouteWatch checks the header each outside request carries, +// which tells geojs and DB-IP who is asking. +func TestRequestNamesRouteWatch(t *testing.T) { + t.Parallel() + + var seen atomic.Pointer[string] + + server := httptest.NewServer(http.HandlerFunc( + func(_ http.ResponseWriter, r *http.Request) { + seen.Store(new(r.Header.Get(userAgentHeader))) + })) + t.Cleanup(server.Close) + + fake := newFakeGeoJS(t, "[]") + dir := t.TempDir() + locator := newTestLocator(t, fake, openCacheDB(t, dir), dir) + locator.geojsURL = server.URL + geojsPath + + lookupOne(t, locator, testAddress) + + got := seen.Load() + if got == nil || !strings.HasPrefix(*got, "routewatch") { + t.Errorf("User-Agent %v", got) + } +} diff --git a/internal/routewatch/app.go b/internal/routewatch/app.go index 35ef1b4..4279136 100644 --- a/internal/routewatch/app.go +++ b/internal/routewatch/app.go @@ -19,6 +19,7 @@ import ( "git.eeqj.de/sneak/routewatch/internal/cachedb" "git.eeqj.de/sneak/routewatch/internal/config" "git.eeqj.de/sneak/routewatch/internal/database" + "git.eeqj.de/sneak/routewatch/internal/geo" "git.eeqj.de/sneak/routewatch/internal/iplist" "git.eeqj.de/sneak/routewatch/internal/logger" "git.eeqj.de/sneak/routewatch/internal/metrics" @@ -67,6 +68,7 @@ type RouteWatch struct { ipLists *iplist.IPLists cacheDB *sql.DB ptrCache *ptr.Cache + geoLocator *geo.Locator } // ipListsDirName is the directory in the state directory that holds the IP @@ -160,6 +162,11 @@ func (rw *RouteWatch) Run(ctx context.Context) error { return err } + err = rw.startGeoLocator(ctx) + if err != nil { + return err + } + // Start HTTP server err = rw.server.Start(ctx) if err != nil { @@ -178,8 +185,9 @@ func (rw *RouteWatch) Run(ctx context.Context) error { // Shutdown stops all services in order: the streamer, the handlers, which // write what they hold, the WHOIS fetcher, the database maintainer, the -// IP list downloads, the HTTP server and the PTR cache, and last the database, -// which it checkpoints and closes. Each stage waits at most until ctx ends, so +// IP list downloads, the HTTP server, the geolocation cache and the PTR cache, +// and last the database, which it checkpoints and closes. Each stage waits at +// most until ctx ends, so // Shutdown returns soon after: a handler logs what it could not write by // then, and the HTTP server closes the connections of requests still // running. @@ -239,6 +247,10 @@ func (rw *RouteWatch) Shutdown(ctx context.Context) { "error", err) } + if rw.geoLocator != nil { + rw.geoLocator.Stop(ctx) + } + rw.stopPTRCache(ctx) rw.closeDatabase(ctx) @@ -283,6 +295,32 @@ func (rw *RouteWatch) startPTRCache(ctx context.Context) error { return nil } +// startGeoLocator loads the geolocation cache from cache.db, which +// startPTRCache has opened, and gives the server the locator. The HTTP handlers +// read it without a lock, so it is set before they run. +func (rw *RouteWatch) startGeoLocator(ctx context.Context) error { + locator, err := geo.New(geo.Params{ + DB: rw.cacheDB, + Dir: rw.config.StateDir, + GeoJSURL: geo.GeoJSEndpoint, + DatabaseURLFormat: geo.DatabaseDownloadFormat, + Logger: rw.logger.Logger, + }) + if err != nil { + return err + } + + err = locator.Start(ctx) + if err != nil { + return err + } + + rw.geoLocator = locator + rw.server.SetGeoLocator(locator) + + return nil +} + // stopPTRCache stops the PTR cache, which writes the answers cache.db does not // have yet, and closes cache.db, unless ctx has ended and a write may still be // running. diff --git a/internal/server/handlers.go b/internal/server/handlers.go index 01d4509..7c4632a 100644 --- a/internal/server/handlers.go +++ b/internal/server/handlers.go @@ -17,6 +17,7 @@ import ( "git.eeqj.de/sneak/routewatch/internal/bogon" "git.eeqj.de/sneak/routewatch/internal/database" + "git.eeqj.de/sneak/routewatch/internal/geo" "git.eeqj.de/sneak/routewatch/internal/iplist" "git.eeqj.de/sneak/routewatch/internal/metrics" "git.eeqj.de/sneak/routewatch/internal/ristypes" @@ -764,9 +765,9 @@ type IPLookupResponse struct { } // IPLookupResult is the answer for one address: its PTR names, its flags, its -// live route and that route's AS, and the IP lists that contain it. Every -// result has every field. The fields of IPInfo are null when the address has -// no live route. +// live route and that route's AS, the IP lists that contain it, and where it +// is. Every result has every field. The fields of IPInfo are null when the +// address has no live route. // //nolint:tagliatelle // snake_case keys, as the Go style guide asks type IPLookupResult struct { @@ -784,6 +785,9 @@ type IPLookupResult struct { // IPLists has one entry for each IP list the address is on. IPLists []iplist.Listing `json:"ip_lists"` + + // Geo is where the address is. It is nil when no source answers. + Geo *geo.Location `json:"geo"` } // IPLookupFlags are the yes-or-no facts about an address. Every result has @@ -839,6 +843,8 @@ func (s *Server) handleIPInfo() http.HandlerFunc { return } + locations := s.geoLocations(ctx, ipsToLookup) + // Track ASNs that need WHOIS refresh refreshASNs := make(map[int]bool) @@ -874,7 +880,9 @@ func (s *Server) handleIPInfo() http.HandlerFunc { return } - response.Results = append(response.Results, s.ipLookupResult(ip, ptrs, ipInfo)) + result := s.ipLookupResult(ip, ptrs, ipInfo) + result.Geo = locations[ip] + response.Results = append(response.Results, result) if ipInfo.NeedsWHOISRefresh { refreshASNs[*ipInfo.ASN] = true @@ -1029,6 +1037,45 @@ func ipLookupFlags(ip string, listings []iplist.Listing) IPLookupFlags { } } +// geoLocations returns where ips are, by address. An address that does not +// parse has no entry. Nor has a reserved one, which is sent to no outside +// source and has no location. +func (s *Server) geoLocations( + ctx context.Context, ips []string, +) map[string]*geo.Location { + if s.geoLocator == nil { + return nil + } + + var ( + kept []string + addrs []netip.Addr + ) + + for _, ip := range ips { + addr, err := netip.ParseAddr(ip) + if err != nil || isBogon(ip) { + continue + } + + kept = append(kept, ip) + addrs = append(addrs, addr) + } + + if len(addrs) == 0 { + return nil + } + + answers := s.geoLocator.Lookup(ctx, addrs) + locations := make(map[string]*geo.Location, len(kept)) + + for i, ip := range kept { + locations[ip] = answers[i] + } + + return locations +} + // isBogon says whether the address ip is reserved, as package bogon tells. func isBogon(ip string) bool { addr, err := netip.ParseAddr(ip) diff --git a/internal/server/handlers_internal_test.go b/internal/server/handlers_internal_test.go index a1898b8..2cfc3c6 100644 --- a/internal/server/handlers_internal_test.go +++ b/internal/server/handlers_internal_test.go @@ -15,6 +15,7 @@ import ( "slices" "strconv" "strings" + "sync" "sync/atomic" "testing" "testing/synctest" @@ -22,6 +23,7 @@ import ( "git.eeqj.de/sneak/routewatch/internal/config" "git.eeqj.de/sneak/routewatch/internal/database" + "git.eeqj.de/sneak/routewatch/internal/geo" "git.eeqj.de/sneak/routewatch/internal/iplist" "git.eeqj.de/sneak/routewatch/internal/logger" "git.eeqj.de/sneak/routewatch/internal/metrics" @@ -1613,6 +1615,131 @@ func TestIPLookupFlags(t *testing.T) { } } +// fakeGeoLocator answers every address with location, and keeps the addresses +// of each call. It is safe for the concurrent subtests that share it. +type fakeGeoLocator struct { + location *geo.Location + + mu sync.Mutex + calls [][]netip.Addr +} + +func (f *fakeGeoLocator) Lookup( + _ context.Context, addrs []netip.Addr, +) []*geo.Location { + f.mu.Lock() + defer f.mu.Unlock() + + f.calls = append(f.calls, addrs) + + answers := make([]*geo.Location, len(addrs)) + for i := range addrs { + answers[i] = f.location + } + + return answers +} + +// TestIPLookupGivesTheGeoObject looks up two public addresses. They go to the +// locator in one call, and each result has the geo object under the keys the +// API documents. These addresses are public, because a reserved one is not +// looked up. +func TestIPLookupGivesTheGeoObject(t *testing.T) { + t.Parallel() + + s := newServerWithTestRoutes(t) + useFakeResolver(s, &fakeResolver{}) + + locator := &fakeGeoLocator{location: &geo.Location{ + CountryCode: new("DE"), + City: new("Berlin"), + Latitude: new(52.5), + Source: "geojs", + Updated: testRouteTime(0), + }} + s.SetGeoLocator(locator) + + path := "/api/v1/ip/1.1.1.1,8.8.8.8" + + results := resultFields(t, get(t, s, path).Body.Bytes()) + if len(results) != 2 { + t.Fatalf("%s: %d results, want 2", path, len(results)) + } + + if len(locator.calls) != 1 || len(locator.calls[0]) != 2 { + t.Errorf("locator calls %v, want one with both addresses", locator.calls) + } + + var got map[string]json.RawMessage + + err := json.Unmarshal(results[0]["geo"], &got) + if err != nil { + t.Fatalf("decoding geo %s: %v", results[0]["geo"], err) + } + + wantKeys := []string{ + "accuracy_radius_km", "city", "continent_code", "country", + "country_code", "latitude", "longitude", "region", "source", + "time_zone", "updated", + } + if keys := slices.Sorted(maps.Keys(got)); !slices.Equal(keys, wantKeys) { + t.Errorf("geo keys %v, want %v", keys, wantKeys) + } + + if string(got["city"]) != `"Berlin"` || string(got["latitude"]) != "52.5" || + string(got["region"]) != jsonNull { + t.Errorf("wrong geo %s", results[0]["geo"]) + } +} + +// TestIPLookupOfAReservedAddressHasNoGeo looks up a reserved address alone and +// beside a public one. The reserved one never reaches the locator, and its geo +// is null. +func TestIPLookupOfAReservedAddressHasNoGeo(t *testing.T) { + t.Parallel() + + s := newServerWithTestRoutes(t) + useFakeResolver(s, &fakeResolver{}) + + locator := &fakeGeoLocator{location: &geo.Location{Source: "geojs"}} + s.SetGeoLocator(locator) + + results := resultFields(t, get(t, s, "/api/v1/ip/"+addrInTestPrefixV4).Body.Bytes()) + if len(results) != 1 || string(results[0]["geo"]) != jsonNull { + t.Fatalf("results %v, want one with geo null", results) + } + + if len(locator.calls) != 0 { + t.Errorf("locator calls %v, want none", locator.calls) + } + + results = resultFields(t, + get(t, s, "/api/v1/ip/"+addrInTestPrefixV4+",1.1.1.1").Body.Bytes()) + if len(results) != 2 || string(results[0]["geo"]) != jsonNull || + string(results[1]["geo"]) == jsonNull { + t.Errorf("results %v, want geo only for the public address", results) + } + + if len(locator.calls) != 1 || len(locator.calls[0]) != 1 { + t.Errorf("locator calls %v, want one with one address", locator.calls) + } +} + +// TestIPLookupGeoIsNullWhenNoSourceAnswers checks an address the locator has +// no answer for. +func TestIPLookupGeoIsNullWhenNoSourceAnswers(t *testing.T) { + t.Parallel() + + s := newServerWithTestRoutes(t) + useFakeResolver(s, &fakeResolver{}) + s.SetGeoLocator(&fakeGeoLocator{}) + + results := resultFields(t, get(t, s, "/api/v1/ip/1.1.1.1").Body.Bytes()) + if len(results) != 1 || string(results[0]["geo"]) != jsonNull { + t.Errorf("results %v, want one with geo null", results) + } +} + // TestIPLookupReportsVPNProviders looks up addresses on no list, on a list // that is not a VPN list, on one VPN list and on two. is_vpn says whether any // VPN list holds the address, and vpn_providers names only those lists. An diff --git a/internal/server/server.go b/internal/server/server.go index 9e7936b..e291076 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -24,6 +24,7 @@ import ( "git.eeqj.de/sneak/routewatch/internal/config" "git.eeqj.de/sneak/routewatch/internal/database" + "git.eeqj.de/sneak/routewatch/internal/geo" "git.eeqj.de/sneak/routewatch/internal/iplist" "git.eeqj.de/sneak/routewatch/internal/logger" "git.eeqj.de/sneak/routewatch/internal/routetable" @@ -63,6 +64,12 @@ type PTRResolver interface { LookupAddr(ctx context.Context, addr string) ([]string, error) } +// GeoLocator finds where addresses are, as *geo.Locator does. It returns one +// entry for each address, nil when no source answers. +type GeoLocator interface { + Lookup(ctx context.Context, addrs []netip.Addr) []*geo.Location +} + // Server provides HTTP endpoints for status monitoring type Server struct { router *chi.Mux @@ -86,6 +93,10 @@ type Server struct { // ptrResolver looks up the IP lookup's PTR names. SetPTRResolver sets it. ptrResolver PTRResolver + // geoLocator finds where the IP lookup's addresses are. SetGeoLocator + // sets it. + geoLocator GeoLocator + // cards is the copy of the home page's cards that the pages serve, nil // until the server first works them out cards atomic.Pointer[homeCards] @@ -209,6 +220,12 @@ func (s *Server) SetPTRResolver(resolver PTRResolver) { s.ptrResolver = resolver } +// SetGeoLocator sets what finds where the IP lookup's addresses are. Call it +// before Start: the handlers read it without a lock. +func (s *Server) SetGeoLocator(locator GeoLocator) { + s.geoLocator = locator +} + // SetASNFetcher sets the ASN WHOIS fetcher for on-demand lookups. Call it // before Start: the handlers read the fetcher without a lock. func (s *Server) SetASNFetcher(fetcher ASNFetcher) { diff --git a/internal/templates/index.html b/internal/templates/index.html index c940576..f0856b8 100644 --- a/internal/templates/index.html +++ b/internal/templates/index.html @@ -544,6 +544,8 @@ {{appLicense}} | {{appGitRevision}} + | + IP Geolocation by DB-IP