diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index da907fe..6be1def 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -55,7 +55,7 @@ jobs: - name: Verify stats and endpoint queries against PostgreSQL 16 env: BEACON_TEST_POSTGRES_DSN: postgres://postgres:backup-ci-only@127.0.0.1:5432/postgres?sslmode=disable - run: go test ./db -run '^Test(Signal|Paths|PacketEndpointsResolveLive|ObserverMetrics|RouteEvidence|RouteEvidenceIndex|AnalyticsRetention|AnalyticsRetentionConcurrent|DeleteOldPacketsBatches)Postgres$' -count=1 -v + run: go test ./db -run '^Test(Signal|Paths|PacketEndpointsResolveLive|ObserverMetrics|RouteEvidence|RouteEvidenceIndex|AnalyticsRetention|AnalyticsRetentionConcurrent|DeleteOldPacketsBatches|MeshMapperCatalogue)Postgres$' -count=1 -v - name: Verify backup command against PostgreSQL 16 run: | diff --git a/cmd/beacon/main.go b/cmd/beacon/main.go index 8e322e6..1fa2ea3 100644 --- a/cmd/beacon/main.go +++ b/cmd/beacon/main.go @@ -29,6 +29,7 @@ import ( "github.com/MeshCore-Beacon/beacon-server/internal/ingest" "github.com/MeshCore-Beacon/beacon-server/internal/keystore" "github.com/MeshCore-Beacon/beacon-server/internal/logging" + "github.com/MeshCore-Beacon/beacon-server/internal/meshmapper" "github.com/MeshCore-Beacon/beacon-server/internal/presence" "github.com/MeshCore-Beacon/beacon-server/internal/scopestore" @@ -189,6 +190,16 @@ func main() { } scopes.Load(scopeEntries) slog.Info(fmt.Sprintf("loaded %d transport scopes", len(scopeEntries)), "component", "startup") + var scopeImporter *meshmapper.Importer + if cfg.MeshMapper.Scopes.Enabled { + restoreCtx, cancelRestore := context.WithTimeout(ctx, 10*time.Second) + scopeImporter, err = meshmapper.New(restoreCtx, cfg.MeshMapper.Scopes, store, scopes, scopeEntries) + cancelRestore() + if err != nil { + slog.Error("failed to restore MeshMapper scope catalogues", "component", "startup", "error", err) + os.Exit(1) + } + } // ── Build channel keystore ────────────────────────────────────────────── entries := make(map[string][]keystore.Entry) @@ -281,6 +292,9 @@ func main() { ) if cr, ok := reader.(*cache.CachedReader); ok { + if scopeImporter != nil { + scopeImporter.SetCacheInvalidator(cr.InvalidateScopeNames) + } broker1.SetCacheInvalidators(cr.InvalidateNode, cr.InvalidateObserver) broker2.SetCacheInvalidators(cr.InvalidateNode, cr.InvalidateObserver) } @@ -301,6 +315,9 @@ func main() { } tasks = append(tasks, background.ObserverCleanupTask(coalescer, resolved.ObserverDeleteAfter, resolved.CleanupInterval, onDelete)) } + if scopeImporter != nil { + tasks = append(tasks, background.Task{Name: "meshmapper.scopes", Interval: meshmapper.PollInterval, Run: scopeImporter.Refresh}) + } scheduler := background.New(tasks) go scheduler.Start(ctx) diff --git a/config.yaml.example b/config.yaml.example index a0a86cc..3e2a66f 100644 --- a/config.yaml.example +++ b/config.yaml.example @@ -214,6 +214,21 @@ cache: # allow_countries: [CA, US] # allow_continents: [NA] +# Optional public MeshMapper scope-name discovery. Manual scopes remain authoritative. +# No API key is needed. Sources must belong to a configured region's IATAs. +# Group endpoints are not supported: their names cannot be attributed to each IATA. +# At most 16 sources, 64 names per source, 64 KiB per response; no page scraping. +# Catalogues and ETags persist in PostgreSQL. Failed refreshes retain known names. +# Refresh success/failure/freshness is logged under component=meshmapper.scopes. +# Removing/turning off a source deactivates its imported matching keys on restart; +# historical scope identities and recorded evidence are retained. +#meshmapper: +# scopes: +# enabled: false +# refresh_interval: 1h # 5m-24h; one source checked per 15s tick +# sources: +# YOW: https://yow.meshmapper.net/get_scopes.php + # Background task intervals. # Shorter intervals are useful during initial deployment to confirm data is # flowing. Back off to 1h or more once stable. diff --git a/db/meshmapper.go b/db/meshmapper.go new file mode 100644 index 0000000..3ddbb28 --- /dev/null +++ b/db/meshmapper.go @@ -0,0 +1,49 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later + +package db + +import ( + "context" + "errors" + + sqlc "github.com/MeshCore-Beacon/beacon-server/db/sqlc" + "github.com/MeshCore-Beacon/beacon-server/internal/meshmapper" + "github.com/MeshCore-Beacon/beacon-server/internal/scopestore" + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgtype" +) + +func (s *Store) GetScopeCatalogue(ctx context.Context, iata, url string) (*meshmapper.Cache, error) { + row, err := s.q.GetScopeCatalogue(ctx, sqlc.GetScopeCatalogueParams{Iata: iata, Url: url}) + if errors.Is(err, pgx.ErrNoRows) { + return nil, nil + } + if err != nil { + return nil, err + } + cache := &meshmapper.Cache{Payload: row.Payload, CheckedAt: row.CheckedAt.Time, AttemptedAt: row.AttemptedAt.Time, NextAttempt: row.NextAttempt.Time, LastError: row.LastError} + if row.Etag != nil { + cache.ETag = *row.Etag + } + return cache, nil +} + +func (s *Store) SaveScopeCatalogue(ctx context.Context, iata, url string, cache meshmapper.Cache, entries []scopestore.Entry) error { + params := sqlc.SaveScopeCatalogueParams{ + Iata: iata, Url: url, Payload: cache.Payload, LastError: cache.LastError, + CheckedAt: pgtype.Timestamptz{Time: cache.CheckedAt, Valid: !cache.CheckedAt.IsZero()}, + AttemptedAt: pgtype.Timestamptz{Time: cache.AttemptedAt, Valid: true}, + NextAttempt: pgtype.Timestamptz{Time: cache.NextAttempt, Valid: true}, + Names: []string{}, Keys: [][]byte{}, Fingerprints: [][]byte{}, + } + if !cache.CheckedAt.IsZero() { + params.Etag = &cache.ETag + } + for _, entry := range entries { + params.Names = append(params.Names, entry.Name) + params.Keys = append(params.Keys, entry.TransportKey) + params.Fingerprints = append(params.Fingerprints, entry.KeyFingerprint) + } + return s.q.SaveScopeCatalogue(ctx, params) +} diff --git a/db/meshmapper_integration_test.go b/db/meshmapper_integration_test.go new file mode 100644 index 0000000..06d7109 --- /dev/null +++ b/db/meshmapper_integration_test.go @@ -0,0 +1,139 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later + +package db + +import ( + "context" + "os" + "testing" + "time" + + sqlc "github.com/MeshCore-Beacon/beacon-server/db/sqlc" + "github.com/MeshCore-Beacon/beacon-server/internal/config" + "github.com/MeshCore-Beacon/beacon-server/internal/meshmapper" + "github.com/MeshCore-Beacon/beacon-server/internal/scopestore" + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" +) + +func TestMeshMapperCataloguePostgres(t *testing.T) { + dsn := os.Getenv("BEACON_TEST_POSTGRES_DSN") + if dsn == "" { + t.Skip("set BEACON_TEST_POSTGRES_DSN") + } + ctx, cancel := context.WithTimeout(context.Background(), time.Minute) + defer cancel() + pool, err := pgxpool.New(ctx, dsn) + if err != nil { + t.Fatal(err) + } + if err := RunMigrations(ctx, pool); err != nil { + pool.Close() + t.Fatal(err) + } + pool.Close() + conn, err := pgx.Connect(ctx, dsn) + if err != nil { + t.Fatal(err) + } + defer conn.Close(context.Background()) + tx, err := conn.Begin(ctx) + if err != nil { + t.Fatal(err) + } + defer tx.Rollback(context.Background()) + for _, table := range []string{"transport_scopes", "meshmapper_scope_catalogues"} { + if _, err := tx.Exec(ctx, "CREATE TEMP TABLE "+table+" (LIKE public."+table+" INCLUDING ALL) ON COMMIT DROP"); err != nil { + t.Fatal(err) + } + } + store := &Store{q: sqlc.New(tx)} + manual := scopestore.FromName("manual") + manual.TransportKey[0] ^= 1 + if err := store.UpsertTransportScope(ctx, manual.Name, "Operator label", manual.TransportKey, manual.KeyFingerprint); err != nil { + t.Fatal(err) + } + url := "https://yow.meshmapper.net/get_scopes.php" + if row, err := store.GetScopeCatalogue(ctx, "YOW", url); err != nil || row != nil { + t.Fatal(row, err) + } + now := time.Now().UTC().Truncate(time.Second) + payload := []byte(`{"generated_at":"2026-09-27T16:58:49Z","region":"YOW","zones":["YOW"],"repeaters":5,"scoped":3,"scopes":[{"name":"yow","repeaters":0,"default":0,"monitored":true,"wardriving":false}]}`) + update := meshmapper.Cache{Payload: payload, ETag: `"v1"`, CheckedAt: now, AttemptedAt: now, NextAttempt: now.Add(time.Hour)} + if err := store.SaveScopeCatalogue(ctx, "YOW", url, update, []scopestore.Entry{scopestore.FromName("manual"), scopestore.FromName("yow")}); err != nil { + t.Fatal(err) + } + manualRows, err := store.GetTransportScopes(ctx) + if err != nil || len(manualRows) != 1 || manualRows[0].TransportKey[0] != manual.TransportKey[0] { + t.Fatal("manual key changed", manualRows, err) + } + var label string + var imported bool + if err := tx.QueryRow(ctx, "SELECT display_name,imported_only FROM transport_scopes WHERE name='#manual'").Scan(&label, &imported); err != nil || label != "Operator label" || imported { + t.Fatal(label, imported, err) + } + row, err := store.GetScopeCatalogue(ctx, "YOW", url) + if err != nil || row.ETag != `"v1"` || !row.CheckedAt.Equal(now) { + t.Fatal(row, err) + } + scopes := scopestore.New() + _, err = meshmapper.New(ctx, config.MeshMapperScopesConfig{Enabled: true, Sources: map[string]string{"YOW": url}}, store, scopes, manualRows) + if err != nil || len(scopes.Entries()) != 2 { + t.Fatal("restart did not restore imported membership", err) + } + // A conditional success preserves the document, advances checked time and clears a failure. + checked := now.Add(time.Hour) + if err := store.SaveScopeCatalogue(ctx, "YOW", url, meshmapper.Cache{ETag: `"v1"`, CheckedAt: checked, AttemptedAt: checked, NextAttempt: checked.Add(time.Hour)}, nil); err != nil { + t.Fatal(err) + } + if err := store.SaveScopeCatalogue(ctx, "YOW", url, meshmapper.Cache{AttemptedAt: checked.Add(time.Minute), NextAttempt: checked.Add(2 * time.Hour), LastError: "HTTP 429"}, nil); err != nil { + t.Fatal(err) + } + row, err = store.GetScopeCatalogue(ctx, "YOW", url) + if err != nil || !row.CheckedAt.Equal(checked) || row.ETag != `"v1"` || row.LastError != "HTTP 429" || len(row.Payload) == 0 { + t.Fatal("failed refresh lost durable snapshot", row, err) + } + // Empty membership deactivates matching but retains the old scope identity for historical rows. + empty := []byte(`{"generated_at":"2026-09-27T17:58:49Z","region":"YOW","zones":["YOW"],"repeaters":5,"scoped":0,"scopes":[]}`) + update.Payload = empty + update.ETag = "" + if err := store.SaveScopeCatalogue(ctx, "YOW", url, update, nil); err != nil { + t.Fatal(err) + } + _, err = meshmapper.New(ctx, config.MeshMapperScopesConfig{Enabled: true, Sources: map[string]string{"YOW": url}}, store, scopes, manualRows) + if err != nil || len(scopes.Entries()) != 1 { + t.Fatal("removed imported membership remained active", err) + } + if _, err := store.GetTransportScopeByName(ctx, "#yow"); err != nil { + t.Fatal("deleted historical identity", err) + } + row, _ = store.GetScopeCatalogue(ctx, "YOW", url) + if row.ETag != "" { + t.Fatal("200 without ETag retained obsolete validator") + } + // Reconfiguring a known imported name as manual promotes it without losing its identity. + yow := scopestore.FromName("yow") + if err := store.UpsertTransportScope(ctx, yow.Name, "Local", yow.TransportKey, yow.KeyFingerprint); err != nil { + t.Fatal(err) + } + manualRows, err = store.GetTransportScopes(ctx) + if err != nil || len(manualRows) != 2 { + t.Fatal("manual promotion failed", err) + } + // A failed cache write also rolls back the identity insertion in the same statement. + if _, err := tx.Exec(ctx, "ALTER TABLE meshmapper_scope_catalogues ADD CHECK(last_error <> 'force-failure'); SAVEPOINT rejected"); err != nil { + t.Fatal(err) + } + update.LastError = "force-failure" + if err := store.SaveScopeCatalogue(ctx, "YOW", url, update, []scopestore.Entry{scopestore.FromName("uncommitted")}); err == nil { + t.Fatal("expected constraint failure") + } + if _, err := tx.Exec(ctx, "ROLLBACK TO SAVEPOINT rejected"); err != nil { + t.Fatal(err) + } + var count int + if err := tx.QueryRow(ctx, "SELECT count(*) FROM transport_scopes WHERE name='#uncommitted'").Scan(&count); err != nil || count != 0 { + t.Fatal("partially committed catalogue", count, err) + } +} diff --git a/db/migrations/042_meshmapper_scopes.sql b/db/migrations/042_meshmapper_scopes.sql new file mode 100644 index 0000000..43b48b6 --- /dev/null +++ b/db/migrations/042_meshmapper_scopes.sql @@ -0,0 +1,17 @@ +-- Copyright 2026 Beacon Contributors +-- SPDX-License-Identifier: AGPL-3.0-or-later + +-- Preserve historical identities while distinguishing import-only matching keys. +ALTER TABLE transport_scopes ADD COLUMN IF NOT EXISTS imported_only BOOLEAN NOT NULL DEFAULT FALSE; + +CREATE TABLE IF NOT EXISTS meshmapper_scope_catalogues ( + iata TEXT NOT NULL, + url TEXT NOT NULL, + payload JSONB, + etag TEXT, + checked_at TIMESTAMPTZ, + attempted_at TIMESTAMPTZ NOT NULL, + next_attempt TIMESTAMPTZ NOT NULL, + last_error TEXT NOT NULL DEFAULT '', + PRIMARY KEY (iata, url) +); diff --git a/db/queries/meshmapper.sql b/db/queries/meshmapper.sql new file mode 100644 index 0000000..16e8243 --- /dev/null +++ b/db/queries/meshmapper.sql @@ -0,0 +1,27 @@ +-- Copyright 2026 Beacon Contributors +-- SPDX-License-Identifier: AGPL-3.0-or-later + +-- name: GetScopeCatalogue :one +SELECT * FROM meshmapper_scope_catalogues WHERE iata = $1 AND url = $2; + +-- name: SaveScopeCatalogue :exec +-- One statement commits the validated snapshot and its lookup identities together. +-- Empty arrays insert nothing. NULL payload/checked_at retain last-known-good data +-- after an error or 304. Imported names never replace existing manual metadata. +WITH inserted AS ( + INSERT INTO transport_scopes (name, transport_key, key_fingerprint, imported_only) + SELECT entry.name, entry.key, entry.fingerprint, TRUE + FROM (SELECT unnest(@names::text[]) AS name, unnest(@keys::bytea[]) AS key, + unnest(@fingerprints::bytea[]) AS fingerprint) AS entry + ON CONFLICT (name) DO NOTHING +) +INSERT INTO meshmapper_scope_catalogues (iata, url, payload, etag, checked_at, attempted_at, next_attempt, last_error) +VALUES (@iata, @url, sqlc.narg(payload)::jsonb, sqlc.narg(etag)::text, + sqlc.narg(checked_at)::timestamptz, @attempted_at, @next_attempt, @last_error) +ON CONFLICT (iata, url) DO UPDATE SET + payload = COALESCE(EXCLUDED.payload, meshmapper_scope_catalogues.payload), + etag = COALESCE(EXCLUDED.etag, meshmapper_scope_catalogues.etag), + checked_at = COALESCE(EXCLUDED.checked_at, meshmapper_scope_catalogues.checked_at), + attempted_at = EXCLUDED.attempted_at, + next_attempt = EXCLUDED.next_attempt, + last_error = EXCLUDED.last_error; diff --git a/db/queries/queries.sql b/db/queries/queries.sql index 05ceacd..37cd8fa 100644 --- a/db/queries/queries.sql +++ b/db/queries/queries.sql @@ -49,10 +49,11 @@ VALUES ($1, $2, $3, $4) ON CONFLICT (name) DO UPDATE SET display_name = EXCLUDED.display_name, transport_key = EXCLUDED.transport_key, - key_fingerprint = EXCLUDED.key_fingerprint; + key_fingerprint = EXCLUDED.key_fingerprint, + imported_only = FALSE; -- name: GetTransportScopes :many -SELECT name, transport_key, key_fingerprint FROM transport_scopes ORDER BY name; +SELECT name, transport_key, key_fingerprint FROM transport_scopes WHERE NOT imported_only ORDER BY name; -- name: GetTransportScopeByName :one SELECT id FROM transport_scopes WHERE name = $1; diff --git a/db/sqlc/meshmapper.sql.go b/db/sqlc/meshmapper.sql.go new file mode 100644 index 0000000..b8ba12d --- /dev/null +++ b/db/sqlc/meshmapper.sql.go @@ -0,0 +1,94 @@ +// Code generated by sqlc. DO NOT EDIT. +// versions: +// sqlc v1.31.1 +// source: meshmapper.sql + +package db + +import ( + "context" + + "github.com/jackc/pgx/v5/pgtype" +) + +const getScopeCatalogue = `-- name: GetScopeCatalogue :one + +SELECT iata, url, payload, etag, checked_at, attempted_at, next_attempt, last_error FROM meshmapper_scope_catalogues WHERE iata = $1 AND url = $2 +` + +type GetScopeCatalogueParams struct { + Iata string `json:"iata"` + Url string `json:"url"` +} + +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later +func (q *Queries) GetScopeCatalogue(ctx context.Context, arg GetScopeCatalogueParams) (MeshmapperScopeCatalogue, error) { + row := q.db.QueryRow(ctx, getScopeCatalogue, arg.Iata, arg.Url) + var i MeshmapperScopeCatalogue + err := row.Scan( + &i.Iata, + &i.Url, + &i.Payload, + &i.Etag, + &i.CheckedAt, + &i.AttemptedAt, + &i.NextAttempt, + &i.LastError, + ) + return i, err +} + +const saveScopeCatalogue = `-- name: SaveScopeCatalogue :exec +WITH inserted AS ( + INSERT INTO transport_scopes (name, transport_key, key_fingerprint, imported_only) + SELECT entry.name, entry.key, entry.fingerprint, TRUE + FROM (SELECT unnest($9::text[]) AS name, unnest($10::bytea[]) AS key, + unnest($11::bytea[]) AS fingerprint) AS entry + ON CONFLICT (name) DO NOTHING +) +INSERT INTO meshmapper_scope_catalogues (iata, url, payload, etag, checked_at, attempted_at, next_attempt, last_error) +VALUES ($1, $2, $3::jsonb, $4::text, + $5::timestamptz, $6, $7, $8) +ON CONFLICT (iata, url) DO UPDATE SET + payload = COALESCE(EXCLUDED.payload, meshmapper_scope_catalogues.payload), + etag = COALESCE(EXCLUDED.etag, meshmapper_scope_catalogues.etag), + checked_at = COALESCE(EXCLUDED.checked_at, meshmapper_scope_catalogues.checked_at), + attempted_at = EXCLUDED.attempted_at, + next_attempt = EXCLUDED.next_attempt, + last_error = EXCLUDED.last_error +` + +type SaveScopeCatalogueParams struct { + Iata string `json:"iata"` + Url string `json:"url"` + Payload []byte `json:"payload"` + Etag *string `json:"etag"` + CheckedAt pgtype.Timestamptz `json:"checked_at"` + AttemptedAt pgtype.Timestamptz `json:"attempted_at"` + NextAttempt pgtype.Timestamptz `json:"next_attempt"` + LastError string `json:"last_error"` + Names []string `json:"names"` + Keys [][]byte `json:"keys"` + Fingerprints [][]byte `json:"fingerprints"` +} + +// One statement commits the validated snapshot and its lookup identities together. +// Empty arrays insert nothing. NULL payload/checked_at retain last-known-good data +// after an error or 304. Imported names never replace existing manual metadata. +func (q *Queries) SaveScopeCatalogue(ctx context.Context, arg SaveScopeCatalogueParams) error { + _, err := q.db.Exec(ctx, saveScopeCatalogue, + arg.Iata, + arg.Url, + arg.Payload, + arg.Etag, + arg.CheckedAt, + arg.AttemptedAt, + arg.NextAttempt, + arg.LastError, + arg.Names, + arg.Keys, + arg.Fingerprints, + ) + return err +} diff --git a/db/sqlc/mock/querier.go b/db/sqlc/mock/querier.go index 9d3fb18..16ea073 100644 --- a/db/sqlc/mock/querier.go +++ b/db/sqlc/mock/querier.go @@ -714,6 +714,21 @@ func (mr *MockQuerierMockRecorder) GetScopeByName(ctx, name any) *gomock.Call { return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetScopeByName", reflect.TypeOf((*MockQuerier)(nil).GetScopeByName), ctx, name) } +// GetScopeCatalogue mocks base method. +func (m *MockQuerier) GetScopeCatalogue(ctx context.Context, arg db.GetScopeCatalogueParams) (db.MeshmapperScopeCatalogue, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "GetScopeCatalogue", ctx, arg) + ret0, _ := ret[0].(db.MeshmapperScopeCatalogue) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// GetScopeCatalogue indicates an expected call of GetScopeCatalogue. +func (mr *MockQuerierMockRecorder) GetScopeCatalogue(ctx, arg any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetScopeCatalogue", reflect.TypeOf((*MockQuerier)(nil).GetScopeCatalogue), ctx, arg) +} + // GetScopeNames mocks base method. func (m *MockQuerier) GetScopeNames(ctx context.Context) ([]string, error) { m.ctrl.T.Helper() @@ -1542,6 +1557,20 @@ func (mr *MockQuerierMockRecorder) ResolvePathHashesP4(ctx, arg any) *gomock.Cal return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ResolvePathHashesP4", reflect.TypeOf((*MockQuerier)(nil).ResolvePathHashesP4), ctx, arg) } +// SaveScopeCatalogue mocks base method. +func (m *MockQuerier) SaveScopeCatalogue(ctx context.Context, arg db.SaveScopeCatalogueParams) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "SaveScopeCatalogue", ctx, arg) + ret0, _ := ret[0].(error) + return ret0 +} + +// SaveScopeCatalogue indicates an expected call of SaveScopeCatalogue. +func (mr *MockQuerierMockRecorder) SaveScopeCatalogue(ctx, arg any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SaveScopeCatalogue", reflect.TypeOf((*MockQuerier)(nil).SaveScopeCatalogue), ctx, arg) +} + // SearchKnownRoutes mocks base method. func (m *MockQuerier) SearchKnownRoutes(ctx context.Context, arg db.SearchKnownRoutesParams) ([]db.SearchKnownRoutesRow, error) { m.ctrl.T.Helper() diff --git a/db/sqlc/models.go b/db/sqlc/models.go index 47cce44..86fe98e 100644 --- a/db/sqlc/models.go +++ b/db/sqlc/models.go @@ -234,6 +234,17 @@ type KnownRoute struct { LastReconfirmedAt pgtype.Timestamptz `json:"last_reconfirmed_at"` } +type MeshmapperScopeCatalogue struct { + Iata string `json:"iata"` + Url string `json:"url"` + Payload []byte `json:"payload"` + Etag *string `json:"etag"` + CheckedAt pgtype.Timestamptz `json:"checked_at"` + AttemptedAt pgtype.Timestamptz `json:"attempted_at"` + NextAttempt pgtype.Timestamptz `json:"next_attempt"` + LastError string `json:"last_error"` +} + type MvHourlyIataStat struct { Iata string `json:"iata"` Hour pgtype.Timestamptz `json:"hour"` @@ -528,4 +539,5 @@ type TransportScope struct { TransportKey []byte `json:"transport_key"` KeyFingerprint []byte `json:"key_fingerprint"` CreatedAt pgtype.Timestamptz `json:"created_at"` + ImportedOnly bool `json:"imported_only"` } diff --git a/db/sqlc/querier.go b/db/sqlc/querier.go index a577064..f96b9a3 100644 --- a/db/sqlc/querier.go +++ b/db/sqlc/querier.go @@ -89,6 +89,9 @@ type Querier interface { GetRegionIATAs(ctx context.Context, regionID int32) ([]string, error) GetRouteEvidenceRoute(ctx context.Context, arg GetRouteEvidenceRouteParams) (KnownRoute, error) GetScopeByName(ctx context.Context, name string) (GetScopeByNameRow, error) + // Copyright 2026 Beacon Contributors + // SPDX-License-Identifier: AGPL-3.0-or-later + GetScopeCatalogue(ctx context.Context, arg GetScopeCatalogueParams) (MeshmapperScopeCatalogue, error) GetScopeNames(ctx context.Context) ([]string, error) // Aggregate matching observations once, separately from node memberships to avoid // a cross-join. Empty IATAs keep the original global counts, including associations @@ -250,6 +253,10 @@ type Querier interface { ResolvePathHashesP2(ctx context.Context, arg ResolvePathHashesP2Params) ([]ResolvePathHashesP2Row, error) ResolvePathHashesP3(ctx context.Context, arg ResolvePathHashesP3Params) ([]ResolvePathHashesP3Row, error) ResolvePathHashesP4(ctx context.Context, arg ResolvePathHashesP4Params) ([]ResolvePathHashesP4Row, error) + // One statement commits the validated snapshot and its lookup identities together. + // Empty arrays insert nothing. NULL payload/checked_at retain last-known-good data + // after an error or 304. Imported names never replace existing manual metadata. + SaveScopeCatalogue(ctx context.Context, arg SaveScopeCatalogueParams) error // Returns known routes containing a subsequence from source to destination hash prefix. // Verifies source appears before destination in the route. SearchKnownRoutes(ctx context.Context, arg SearchKnownRoutesParams) ([]SearchKnownRoutesRow, error) diff --git a/db/sqlc/queries.sql.go b/db/sqlc/queries.sql.go index 054bc00..9195a62 100644 --- a/db/sqlc/queries.sql.go +++ b/db/sqlc/queries.sql.go @@ -2051,7 +2051,7 @@ func (q *Queries) GetTransportScopeByName(ctx context.Context, name string) (int } const getTransportScopes = `-- name: GetTransportScopes :many -SELECT name, transport_key, key_fingerprint FROM transport_scopes ORDER BY name +SELECT name, transport_key, key_fingerprint FROM transport_scopes WHERE NOT imported_only ORDER BY name ` type GetTransportScopesRow struct { @@ -5228,7 +5228,8 @@ VALUES ($1, $2, $3, $4) ON CONFLICT (name) DO UPDATE SET display_name = EXCLUDED.display_name, transport_key = EXCLUDED.transport_key, - key_fingerprint = EXCLUDED.key_fingerprint + key_fingerprint = EXCLUDED.key_fingerprint, + imported_only = FALSE ` type UpsertTransportScopeParams struct { diff --git a/docs/meshmapper-scopes.md b/docs/meshmapper-scopes.md new file mode 100644 index 0000000..7c9a391 --- /dev/null +++ b/docs/meshmapper-scopes.md @@ -0,0 +1,26 @@ +# Optional MeshMapper scope discovery + +Beacon can periodically read the [public Scopes API](https://wiki.meshmapper.net/scopes-api/) to discover additional transport scope names for configured regions. Manual `scopes` remain authoritative. This does not supply channel decryption keys or per-repeater records. + +```yaml +meshmapper: + scopes: + enabled: true + refresh_interval: 1h + sources: + YOW: https://yow.meshmapper.net/get_scopes.php +``` + +The IATA must already occur in `regions[].iatas`. Set the exact published regional HTTPS URL; credentials, query strings, other hosts and redirects are rejected. The response must identify the same region and exactly one zone. Group catalogues are rejected because they cannot assign each scope/count to individual member regions. No API key is required. + +The default is disabled. Refresh defaults to one hour and accepts 5 minutes through 24 hours. One source is considered every 15 seconds, keeping requests below four per minute per process. Multiple deployments share MeshMapper's per-IP allowance. ETags reduce transfer, HTTP 304 advances the last successful check, and 429 pauses all sources until the later of the normal retry time and `Retry-After`. The pause survives restarts. HTTP 503 also honors that source's `Retry-After`. + +Responses are limited to 64 KiB and 64 names per source, with at most 16 sources. Names are case-sensitive and at most 128 bytes before Beacon's usual `#` normalization; existing `#`/`$` prefixes are retained. Zero-count monitored names are kept. Imported matching candidates are limited to their source IATA; manual keys keep their existing global applicability. A short transport code that matches multiple candidate names remains unresolved. + +Migration 042 stores the last validated response, validator and check/retry timestamps in PostgreSQL. Saving a response and its scope identities is atomic. Timeouts, invalid responses and HTTP errors retain the prior catalogue; startup restores it without HTTP. A successful empty response removes that source's active imported membership. Removing or disabling a source takes effect on restart. Scope identities referenced by historical records are retained. Adding an imported name to the manual list promotes it to manual ownership. + +Operator logs under `component=meshmapper.scopes` show the IATA, source, active name count, source generation time, last successful check, next attempt and last error. An empty last error means the latest request succeeded; a zero checked time means no successful response has been stored. A 304 retains the original source generation time because MeshMapper excludes that timestamp from its ETag. Cached names may be old during an outage; use the check/error fields to assess freshness. + +Catalogue counts remain source metadata. They never create Beacon packet counts, observer associations or node defaults. Actual packet-code matching supplies those associations through the existing ingest path. New names affect subsequent packets; no historical scan is started. Channel message tags and import-status UI are separate follow-ups. Source corruption or an unavailable catalogue database at startup is reported as a startup failure; disabling the optional importer restores manual-only behavior without deleting saved data. + +Validate and back up PostgreSQL before applying the migration. Rollback to an older server must restore its matching pre-import database and configuration as well as its binary; an older matcher does not understand imported-only ownership or regional candidate limits. diff --git a/internal/cache/reader.go b/internal/cache/reader.go index a570df9..515adc9 100644 --- a/internal/cache/reader.go +++ b/internal/cache/reader.go @@ -80,6 +80,11 @@ func NewCachedReader(inner api.Reader, c *Client, ttl CacheTTLs) api.Reader { } } +// InvalidateScopeNames makes newly committed catalogue names available to filters. +func (cr *CachedReader) InvalidateScopeNames(ctx context.Context) { + cr.c.del(ctx, keyScopeNames) +} + // InvalidateNode removes the cached entries for a node by UUID. // Should be called from the ingest path after a node upsert. func (cr *CachedReader) InvalidateNode(ctx context.Context, nodeID uuid.UUID) { diff --git a/internal/config/config.go b/internal/config/config.go index b3c6cd3..b8b8fff 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -33,6 +33,7 @@ type Config struct { Routes RoutesConfig `yaml:"routes"` Ingest IngestFilterConfig `yaml:"ingest"` Scopes []ScopeConfig `yaml:"scopes"` + MeshMapper MeshMapperConfig `yaml:"meshmapper"` Cache CacheConfig `yaml:"cache"` CORS CORSConfig `yaml:"cors"` RateLimit RateLimitConfig `yaml:"ratelimit"` @@ -408,6 +409,9 @@ func Load(path string) (*Config, error) { } } configDir := filepath.Dir(path) + if err := cfg.validateMeshMapper(); err != nil { + return nil, err + } for iata, details := range cfg.IATAs { if details.BorderFile != "" && !filepath.IsAbs(details.BorderFile) { details.BorderFile = filepath.Join(configDir, details.BorderFile) diff --git a/internal/config/meshmapper.go b/internal/config/meshmapper.go new file mode 100644 index 0000000..5f36774 --- /dev/null +++ b/internal/config/meshmapper.go @@ -0,0 +1,64 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later + +package config + +import ( + "fmt" + "net/url" + "regexp" + "time" +) + +// MaxScopeSources bounds both network refresh work and imported matching keys. +const MaxScopeSources = 16 + +type MeshMapperConfig struct { + Scopes MeshMapperScopesConfig `yaml:"scopes"` +} + +// MeshMapperScopesConfig augments, but never replaces, the manual scopes list. +type MeshMapperScopesConfig struct { + Enabled bool `yaml:"enabled"` + Sources map[string]string `yaml:"sources"` // configured IATA -> published regional endpoint + RefreshInterval duration `yaml:"refresh_interval"` +} + +func (c MeshMapperScopesConfig) Interval() time.Duration { + if c.RefreshInterval.Duration == 0 { + return time.Hour + } + return c.RefreshInterval.Duration +} + +var meshMapperHost = regexp.MustCompile(`^[a-z0-9][a-z0-9-]*\.meshmapper\.net$`) + +func (c *Config) validateMeshMapper() error { + s := c.MeshMapper.Scopes + if !s.Enabled { + return nil + } + if len(s.Sources) == 0 || len(s.Sources) > MaxScopeSources { + return fmt.Errorf("meshmapper.scopes.sources must contain 1-%d regional endpoints", MaxScopeSources) + } + if s.Interval() < 5*time.Minute || s.Interval() > 24*time.Hour { + return fmt.Errorf("meshmapper.scopes.refresh_interval must be between 5m and 24h") + } + configured := map[string]bool{} + for _, region := range c.Regions { + for _, iata := range region.IATAs { + configured[iata] = true + } + } + for iata, endpoint := range s.Sources { + if !configured[iata] { + return fmt.Errorf("meshmapper.scopes source %q must belong to a configured region", iata) + } + u, err := url.Parse(endpoint) + if err != nil || u.Scheme != "https" || u.User != nil || !meshMapperHost.MatchString(u.Host) || + u.Path != "/get_scopes.php" || u.RawPath != "" || u.RawQuery != "" || u.ForceQuery || u.Fragment != "" { + return fmt.Errorf("meshmapper.scopes source %q must be a published https://.meshmapper.net/get_scopes.php endpoint without credentials or parameters", iata) + } + } + return nil +} diff --git a/internal/config/meshmapper_test.go b/internal/config/meshmapper_test.go new file mode 100644 index 0000000..56a5634 --- /dev/null +++ b/internal/config/meshmapper_test.go @@ -0,0 +1,49 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later + +package config + +import ( + "os" + "path/filepath" + "strings" + "testing" + "time" +) + +func TestMeshMapperConfig(t *testing.T) { + valid := "regions:\n - slug: ottawa\n iatas: [YOW]\nmeshmapper:\n scopes:\n enabled: true\n sources:\n YOW: https://yow.meshmapper.net/get_scopes.php\n" + for _, tc := range []struct { + name, text string + bad bool + }{ + {"valid", valid, false}, + {"unknown region", strings.Replace(valid, "iatas: [YOW]", "iatas: [YVR]", 1), true}, + {"http", strings.Replace(valid, "https:", "http:", 1), true}, + {"private host", strings.Replace(valid, "yow.meshmapper.net", "127.0.0.1", 1), true}, + {"credentials", strings.Replace(valid, "https://", "https://secret@", 1), true}, + {"query", strings.Replace(valid, "get_scopes.php", "get_scopes.php?key=secret", 1), true}, + {"non API", strings.Replace(valid, "get_scopes.php", "index.php", 1), true}, + {"fast", valid + " refresh_interval: 1s\n", true}, + {"negative", valid + " refresh_interval: -1h\n", true}, + {"disabled", strings.Replace(valid, "enabled: true", "enabled: false", 1), false}, + } { + t.Run(tc.name, func(t *testing.T) { + p := filepath.Join(t.TempDir(), "config.yaml") + if err := os.WriteFile(p, []byte(tc.text), 0600); err != nil { + t.Fatal(err) + } + cfg, err := Load(p) + if (err != nil) != tc.bad { + t.Fatalf("error=%v", err) + } + if err == nil && cfg.MeshMapper.Scopes.Interval() != time.Hour { + t.Fatal("wrong default") + } + }) + } + cfg := Config{MeshMapper: MeshMapperConfig{Scopes: MeshMapperScopesConfig{Enabled: true}}} + if err := cfg.validateMeshMapper(); err == nil { + t.Fatal("empty sources accepted") + } +} diff --git a/internal/config/seed.go b/internal/config/seed.go index 53ea371..a419573 100644 --- a/internal/config/seed.go +++ b/internal/config/seed.go @@ -5,12 +5,12 @@ package config import ( "context" - "crypto/sha256" "encoding/json" "fmt" "log/slog" "os" - "strings" + + "github.com/MeshCore-Beacon/beacon-server/internal/scopestore" ) // Seeder is the database interface required to seed config data on startup. @@ -63,29 +63,10 @@ func Seed(ctx context.Context, cfg *Config, db Seeder) error { } // Transport Codes for _, s := range cfg.Scopes { - name := normalizeScopeName(s.Name) - key := deriveScopeKey(name) - h := sha256.Sum256(key) - fingerprint := h[:8] - if err := db.UpsertTransportScope(ctx, name, "", key, fingerprint); err != nil { + entry := scopestore.FromName(s.Name) + if err := db.UpsertTransportScope(ctx, entry.Name, "", entry.TransportKey, entry.KeyFingerprint); err != nil { return err } } return nil } - -// normalizeScopeName ensures the scope name has a # or $ prefix. -// Plain names get # prepended: "bc" → "#bc". -func normalizeScopeName(name string) string { - if strings.HasPrefix(name, "#") || strings.HasPrefix(name, "$") { - return name - } - return "#" + name -} - -// deriveScopeKey derives the 16-byte transport key from a normalized scope name. -// key = SHA256(name)[:16] -func deriveScopeKey(name string) []byte { - h := sha256.Sum256([]byte(name)) - return h[:16] -} diff --git a/internal/config/seed_test.go b/internal/config/seed_test.go index 4f2fa0c..069bbe3 100644 --- a/internal/config/seed_test.go +++ b/internal/config/seed_test.go @@ -8,6 +8,7 @@ import ( "encoding/hex" "encoding/json" "errors" + "github.com/MeshCore-Beacon/beacon-server/internal/scopestore" "os" "testing" ) @@ -198,39 +199,39 @@ func TestSeed_DBError(t *testing.T) { } func TestNormalizeScopeName_WithHash(t *testing.T) { - if normalizeScopeName("#bc") != "#bc" { + if scopestore.FromName("#bc").Name != "#bc" { t.Error("expected #bc unchanged") } } func TestNormalizeScopeName_WithDollar(t *testing.T) { - if normalizeScopeName("$bc") != "$bc" { + if scopestore.FromName("$bc").Name != "$bc" { t.Error("expected $bc unchanged") } } func TestNormalizeScopeName_WithoutPrefix(t *testing.T) { - if normalizeScopeName("bc") != "#bc" { + if scopestore.FromName("bc").Name != "#bc" { t.Error("expected bc to become #bc") } } func TestNormalizeScopeName_Empty(t *testing.T) { - if normalizeScopeName("") != "#" { + if scopestore.FromName("").Name != "#" { t.Error("expected empty string to become #") } } func TestDeriveScopeKey_Length(t *testing.T) { - key := deriveScopeKey("#bc") + key := scopestore.FromName("#bc").TransportKey if len(key) != 16 { t.Errorf("expected 16 bytes, got %d", len(key)) } } func TestDeriveScopeKey_Deterministic(t *testing.T) { - a := deriveScopeKey("#bc") - b := deriveScopeKey("#bc") + a := scopestore.FromName("#bc").TransportKey + b := scopestore.FromName("#bc").TransportKey if hex.EncodeToString(a) != hex.EncodeToString(b) { t.Error("expected same key for same input") } @@ -238,7 +239,7 @@ func TestDeriveScopeKey_Deterministic(t *testing.T) { func TestDeriveScopeKey_KnownValue(t *testing.T) { // SHA256("#bc")[:16] — pin the exact derivation so changes are caught - key := deriveScopeKey("#bc") + key := scopestore.FromName("#bc").TransportKey got := hex.EncodeToString(key) // generate this once: echo -n "#bc" | sha256sum | cut -c1-32 const want = "84509cfe73d94f7f6a8299e6bcdb8a3c" @@ -248,8 +249,8 @@ func TestDeriveScopeKey_KnownValue(t *testing.T) { } func TestDeriveScopeKey_DifferentInputs(t *testing.T) { - a := deriveScopeKey("#bc") - b := deriveScopeKey("#other") + a := scopestore.FromName("#bc").TransportKey + b := scopestore.FromName("#other").TransportKey if hex.EncodeToString(a) == hex.EncodeToString(b) { t.Error("expected different keys for different inputs") } diff --git a/internal/ingest/packet.go b/internal/ingest/packet.go index bcc9ca5..5dfe7d7 100644 --- a/internal/ingest/packet.go +++ b/internal/ingest/packet.go @@ -11,11 +11,13 @@ import ( "encoding/hex" "encoding/json" "fmt" + "slices" "strings" "time" "github.com/MeshCore-Beacon/beacon-server/internal/api" "github.com/MeshCore-Beacon/beacon-server/internal/lora" + "github.com/MeshCore-Beacon/beacon-server/internal/scopestore" "github.com/google/uuid" "github.com/meshcore-go/meshcore-go" ) @@ -694,14 +696,7 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [ var matchedScope *string if packet.RouteType() == meshcore.RouteTypeTransportFlood || packet.RouteType() == meshcore.RouteTypeTransportDirect { - for _, entry := range w.scopes.Entries() { - code := computeTransportCode(entry.TransportKey, packet.PayloadType(), packet.Payload) - if code == packet.TransportCode1 { - s := entry.Name - matchedScope = &s - break - } - } + matchedScope = matchTransportScope(w.scopes.Entries(), iata, packet.PayloadType(), packet.Payload, packet.TransportCode1) } var scopeID *int32 if matchedScope != nil { @@ -930,6 +925,26 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [ } } +// matchTransportScope never treats catalogue membership as forwarding evidence. +// Multiple candidate names matching the short code are ambiguous, not first-wins. +func matchTransportScope(entries []scopestore.Entry, iata string, payloadType uint8, payload []byte, code uint16) *string { + var matched *string + for _, entry := range entries { + if entry.IATAs != nil && !slices.Contains(entry.IATAs, iata) { + continue + } + if computeTransportCode(entry.TransportKey, payloadType, payload) != code { + continue + } + if matched != nil && *matched != entry.Name { + return nil + } + name := entry.Name + matched = &name + } + return matched +} + // computeTransportCode derives transport_code_1 from a transport key and packet payload. // code = HMAC-SHA256(key, payload_type_byte || payload)[0:2] as little-endian uint16. // Coerces reserved values 0x0000 → 0x0001 and 0xFFFF → 0xFFFE per §2.4. diff --git a/internal/ingest/scope_import_test.go b/internal/ingest/scope_import_test.go new file mode 100644 index 0000000..3c085e4 --- /dev/null +++ b/internal/ingest/scope_import_test.go @@ -0,0 +1,95 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later + +package ingest + +import ( + "context" + "fmt" + "testing" + + "github.com/MeshCore-Beacon/beacon-server/internal/scopestore" + "github.com/meshcore-go/meshcore-go" +) + +func TestTransportScopeRegionalAndAmbiguousMatches(t *testing.T) { + entry := scopestore.FromName("yow") + entry.IATAs = []string{"YOW"} + payload := []byte{0xde, 0xad, 0xbe, 0xef} + // Fixed HMAC-SHA256 fixture: SHA256("#yow")[:16], payload type 4. + if got := matchTransportScope([]scopestore.Entry{entry}, "YOW", 4, payload, 21698); got == nil || *got != "#yow" { + t.Fatal(got) + } + if got := matchTransportScope([]scopestore.Entry{entry}, "YVR", 4, payload, 21698); got != nil { + t.Fatal("imported key escaped source region") + } + entry.IATAs = nil + if got := matchTransportScope([]scopestore.Entry{entry}, "YVR", 4, payload, 21698); got == nil { + t.Fatal("manual key was region-restricted") + } + // Distinct keys with the same short code for this payload must remain unknown. + a, b := scopestore.FromName("collision-320"), scopestore.FromName("collision-322") + for _, entries := range [][]scopestore.Entry{{a, b}, {b, a}} { + if computeTransportCode(entries[0].TransportKey, 4, payload) != 37018 { + t.Fatal("collision fixture changed") + } + if got := matchTransportScope(entries, "YOW", 4, payload, 37018); got != nil { + t.Fatal("ambiguous short code resolved", *got) + } + } +} + +type importCaptureDB struct{ *frameCaptureDB } + +func (s *importCaptureDB) GetTransportScopeByName(context.Context, string) (int32, error) { + return 123, nil +} + +func TestImportedScopeFlowsThroughPacketIngest(t *testing.T) { + for _, iata := range []string{"YOW", "YVR"} { + t.Run(iata, func(t *testing.T) { + w, base := newTestWorker() + capture := &importCaptureDB{&frameCaptureDB{stubDB: base}} + w.db = capture + key := scopestore.FromName("yow") + key.IATAs = []string{"YOW"} + store := scopestore.New() + store.Load([]scopestore.Entry{key}) + w.scopes = store + packet := buildGrpTxtPacket(t, 0x1a, make([]byte, 16)) + packet.Header = meshcore.MakeHeader(meshcore.RouteTypeTransportFlood, meshcore.PayloadTypeGrpTxt, 0) + packet.TransportCode1 = computeTransportCode(key.TransportKey, packet.PayloadType(), packet.Payload) + w.handlePacket(context.Background(), iata, "0102", packetEnvelope(t, packet)) + if len(capture.packets) != 1 { + t.Fatal("packet not stored") + } + got := capture.packets[0].ScopeID + if iata == "YOW" && (got == nil || *got != 123) { + t.Fatal("scope not persisted", got) + } + if iata == "YVR" && got != nil { + t.Fatal("scope assigned outside source region", got) + } + }) + } +} + +func BenchmarkTransportScopeCatalogue(b *testing.B) { + for _, count := range []int{0, 64, 1024} { + b.Run(fmt.Sprint(count), func(b *testing.B) { + entries := []scopestore.Entry{scopestore.FromName("manual")} + for n := 0; n < count; n++ { + entry := scopestore.FromName(fmt.Sprintf("scope-%d", n)) + entry.IATAs = []string{fmt.Sprintf("R%02d", n/64)} + entries = append(entries, entry) + } + store := scopestore.New() + store.Load(entries) + payload := make([]byte, 128) + b.ReportAllocs() + for b.Loop() { + matchTransportScope(store.Entries(), "R00", 4, payload, 1234) + } + }) + } +} diff --git a/internal/meshmapper/scopes.go b/internal/meshmapper/scopes.go new file mode 100644 index 0000000..a8b64ce --- /dev/null +++ b/internal/meshmapper/scopes.go @@ -0,0 +1,294 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later + +// Package meshmapper imports the published regional scope catalogue, off the ingest path. +package meshmapper + +import ( + "context" + "encoding/json" + "fmt" + "io" + "log/slog" + "net/http" + "slices" + "strconv" + "strings" + "time" + "unicode" + "unicode/utf8" + + "github.com/MeshCore-Beacon/beacon-server/internal/config" + "github.com/MeshCore-Beacon/beacon-server/internal/scopestore" +) + +const ( + MaxScopes = 64 // per source; at most 16 configured sources (1024 imported identities) + MaxBody = 64 << 10 + PollInterval = 15 * time.Second // one source per tick, at most four requests/minute +) + +// Cache stores source provenance and freshness separately from packet evidence. +type Cache struct { + Payload json.RawMessage + ETag string + CheckedAt, AttemptedAt, NextAttempt time.Time + LastError string +} + +type Store interface { + GetScopeCatalogue(context.Context, string, string) (*Cache, error) + SaveScopeCatalogue(context.Context, string, string, Cache, []scopestore.Entry) error +} + +type source struct { + iata, url string + cache Cache + entries []scopestore.Entry + generated time.Time +} + +// Importer is owned by one background task; ScopeStore synchronizes its consumers. +type Importer struct { + store Store + scopes *scopestore.ScopeStore + manual []scopestore.Entry + sources []source + interval time.Duration + client *http.Client + retryAfter time.Time + onChange func(context.Context) +} + +// SetCacheInvalidator is wired once at startup, before the background task starts. +func (i *Importer) SetCacheInvalidator(fn func(context.Context)) { i.onChange = fn } + +// New restores validated snapshots before ingestion, without making HTTP requests. +func New(ctx context.Context, cfg config.MeshMapperScopesConfig, store Store, scopes *scopestore.ScopeStore, manual []scopestore.Entry) (*Importer, error) { + i := &Importer{store: store, scopes: scopes, manual: manual, interval: cfg.Interval(), client: &http.Client{ + Timeout: 10 * time.Second, + CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }, + }} + if !cfg.Enabled { + return i, nil + } + keys := make([]string, 0, len(cfg.Sources)) + for key := range cfg.Sources { + keys = append(keys, key) + } + slices.Sort(keys) + for _, key := range keys { + s := source{iata: key, url: cfg.Sources[key]} + cached, err := store.GetScopeCatalogue(ctx, s.iata, s.url) + if err != nil { + return nil, fmt.Errorf("restore scope catalogue %s: %w", key, err) + } + if cached != nil { + s.cache = *cached + if len(cached.Payload) > 0 { + s.entries, s.generated, err = decode(cached.Payload, key) + if err != nil { + return nil, fmt.Errorf("invalid saved scope catalogue %s: %w", key, err) + } + } + if cached.LastError == "HTTP 429" && cached.NextAttempt.After(i.retryAfter) { + i.retryAfter = cached.NextAttempt + } + } + i.sources = append(i.sources, s) + i.log(s, "restored") + } + i.publish() + return i, nil +} + +// Refresh checks only one due source. The scheduler serializes calls every 15s. +func (i *Importer) Refresh(ctx context.Context) error { + ctx, cancel := context.WithTimeout(ctx, 15*time.Second) + defer cancel() + now := time.Now().UTC() + if now.Before(i.retryAfter) { + return nil + } + for n := range i.sources { + s := &i.sources[n] + if now.Before(s.cache.NextAttempt) { + continue + } + return i.refresh(ctx, s, now) + } + return nil +} + +func (i *Importer) refresh(ctx context.Context, s *source, now time.Time) error { + update := Cache{AttemptedAt: now, NextAttempt: now.Add(i.interval)} + request, err := http.NewRequestWithContext(ctx, http.MethodGet, s.url, nil) + if err != nil { + return err + } + request.Header.Set("Accept", "application/json") + request.Header.Set("User-Agent", "Beacon-MeshMapper-Scopes/1") + if len(s.cache.Payload) > 0 && s.cache.ETag != "" { + request.Header.Set("If-None-Match", s.cache.ETag) + } + response, err := i.client.Do(request) + var entries []scopestore.Entry + var generated time.Time + if err != nil { + if ctx.Err() != nil { + return ctx.Err() + } + update.LastError = "request failed" // do not persist untrusted response bodies or URLs from errors + } else { + defer response.Body.Close() + switch response.StatusCode { + case http.StatusOK: + body, readErr := io.ReadAll(io.LimitReader(response.Body, MaxBody+1)) + if readErr == nil { + entries, generated, readErr = decode(body, s.iata) + } + if readErr != nil { + update.LastError = "invalid response" + } else { + update.Payload = body + } + case http.StatusNotModified: + if len(s.cache.Payload) == 0 || s.cache.ETag == "" { + update.LastError = "304 without cached catalogue" + } + default: + update.LastError = fmt.Sprintf("HTTP %d", response.StatusCode) + if response.StatusCode == http.StatusTooManyRequests || response.StatusCode == http.StatusServiceUnavailable { + // The published contract uses seconds. Ignore invalid/overflowing headers. + if seconds, parseErr := strconv.ParseInt(response.Header.Get("Retry-After"), 10, 64); parseErr == nil && seconds > 0 && seconds <= int64((1<<63-1)/time.Second) { + until := now.Add(time.Duration(seconds) * time.Second) + if until.After(update.NextAttempt) { + update.NextAttempt = until + } + } + } + } + if update.LastError == "" { + etag := response.Header.Get("ETag") + if len(etag) > 256 || strings.ContainsAny(etag, "\r\n") { + update.LastError = "invalid ETag" + update.Payload = nil + entries = nil + } else { + update.ETag = etag + if response.StatusCode == http.StatusNotModified && etag == "" { + update.ETag = s.cache.ETag + } + update.CheckedAt = now + } + } + } + if update.LastError == "HTTP 429" { + i.retryAfter = update.NextAttempt + } + if err := i.store.SaveScopeCatalogue(ctx, s.iata, s.url, update, entries); err != nil { + // Avoid a retry every scheduler tick during a DB outage; never publish uncommitted keys. + s.cache.NextAttempt = update.NextAttempt + return fmt.Errorf("persist scope catalogue %s: %w", s.iata, err) + } + if len(update.Payload) > 0 { + s.cache.Payload = update.Payload + s.entries = entries + s.generated = generated + } + if !update.CheckedAt.IsZero() { + s.cache.CheckedAt = update.CheckedAt + s.cache.ETag = update.ETag + } + s.cache.AttemptedAt = now + s.cache.NextAttempt = update.NextAttempt + s.cache.LastError = update.LastError + if len(update.Payload) > 0 { + i.publish() + if i.onChange != nil { + i.onChange(ctx) + } + } + i.log(*s, "checked") + return nil +} + +func (i *Importer) publish() { + byName := make(map[string]scopestore.Entry, len(i.manual)) + for _, e := range i.manual { + byName[e.Name] = e + } + for _, s := range i.sources { + for _, candidate := range s.entries { + entry, exists := byName[candidate.Name] + if exists && entry.IATAs == nil { + continue + } // manual metadata/keys win + if !exists { + entry = candidate + } + entry.IATAs = append(entry.IATAs, s.iata) + byName[entry.Name] = entry + } + } + names := make([]string, 0, len(byName)) + for name := range byName { + names = append(names, name) + } + slices.Sort(names) + entries := make([]scopestore.Entry, 0, len(names)) + for _, name := range names { + entries = append(entries, byName[name]) + } + i.scopes.Load(entries) +} + +func (i *Importer) log(s source, action string) { + level := slog.LevelInfo + if s.cache.LastError != "" { + level = slog.LevelWarn + } + slog.Log(context.Background(), level, "MeshMapper scopes "+action, "component", "meshmapper.scopes", "iata", s.iata, "source", s.url, + "names", len(s.entries), "generated_at", s.generated, "checked_at", s.cache.CheckedAt, "next_attempt", s.cache.NextAttempt, "last_error", s.cache.LastError) +} + +func decode(body []byte, iata string) ([]scopestore.Entry, time.Time, error) { + var document struct { + GeneratedAt time.Time `json:"generated_at"` + Region string `json:"region"` + Zones []string `json:"zones"` + Repeaters, Scoped *int + Scopes *[]struct { + Name string + Repeaters, Default *int + Monitored, Wardriving *bool + } + } + if len(body) > MaxBody || !utf8.Valid(body) { + return nil, time.Time{}, fmt.Errorf("invalid catalogue size/encoding") + } + if err := json.Unmarshal(body, &document); err != nil { + return nil, time.Time{}, err + } + if document.GeneratedAt.IsZero() || document.Region != iata || len(document.Zones) != 1 || document.Zones[0] != iata || + document.Repeaters == nil || document.Scoped == nil || *document.Scoped < 0 || *document.Repeaters < *document.Scoped || + document.Scopes == nil || len(*document.Scopes) > MaxScopes { + return nil, time.Time{}, fmt.Errorf("invalid regional catalogue") + } + entries := make([]scopestore.Entry, 0, len(*document.Scopes)) + seen := map[string]bool{} + for _, s := range *document.Scopes { + if s.Name == "" || len(s.Name) > 128 || strings.TrimSpace(s.Name) != s.Name || strings.ContainsFunc(s.Name, unicode.IsControl) || + s.Repeaters == nil || s.Default == nil || s.Monitored == nil || s.Wardriving == nil || + *s.Default < 0 || *s.Repeaters < *s.Default || *s.Repeaters > *document.Scoped { + return nil, time.Time{}, fmt.Errorf("invalid scope entry") + } + entry := scopestore.FromName(s.Name) + if len(entry.Name) < 2 || seen[entry.Name] { + return nil, time.Time{}, fmt.Errorf("empty or duplicate scope name") + } + seen[entry.Name] = true + entries = append(entries, entry) + } + return entries, document.GeneratedAt, nil +} diff --git a/internal/meshmapper/scopes_test.go b/internal/meshmapper/scopes_test.go new file mode 100644 index 0000000..66d854a --- /dev/null +++ b/internal/meshmapper/scopes_test.go @@ -0,0 +1,253 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later + +package meshmapper + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "net/http" + "net/http/httptest" + "reflect" + "strings" + "testing" + "time" + + "github.com/MeshCore-Beacon/beacon-server/internal/config" + "github.com/MeshCore-Beacon/beacon-server/internal/scopestore" +) + +func catalogue(names ...string) []byte { + rows := make([]map[string]any, 0, len(names)) + for _, name := range names { + rows = append(rows, map[string]any{"name": name, "repeaters": 0, "default": 0, "monitored": true, "wardriving": false}) + } + b, _ := json.Marshal(map[string]any{"generated_at": "2026-09-27T16:58:49Z", "region": "YOW", "zones": []string{"YOW"}, "repeaters": 5, "scoped": 3, "scopes": rows}) + return b +} + +type memoryStore struct { + rows map[string]Cache + fail bool + saved []scopestore.Entry +} + +func (s *memoryStore) GetScopeCatalogue(_ context.Context, iata, url string) (*Cache, error) { + row, ok := s.rows[iata+url] + if !ok { + return nil, nil + } + return &row, nil +} +func (s *memoryStore) SaveScopeCatalogue(_ context.Context, iata, url string, next Cache, entries []scopestore.Entry) error { + if s.fail { + return errors.New("offline") + } + old := s.rows[iata+url] + if next.Payload == nil { + next.Payload = old.Payload + } + if next.CheckedAt.IsZero() { + next.CheckedAt = old.CheckedAt + next.ETag = old.ETag + } + s.rows[iata+url] = next + s.saved = append(s.saved, entries...) + return nil +} + +func names(store *scopestore.ScopeStore) []string { + out := []string{} + for _, e := range store.Entries() { + out = append(out, e.Name) + } + return out +} + +func TestCatalogueRefreshFallbackAndRestart(t *testing.T) { + ctx := context.Background() + status, body, etag, after := 200, catalogue("yow", "manual"), `"v1"`, "" + calls := 0 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + calls++ + if calls > 1 && r.Header.Get("If-None-Match") != `"v1"` { + t.Errorf("missing conditional refresh: %v", r.Header) + } + w.Header().Set("ETag", etag) + w.Header().Set("Retry-After", after) + w.WriteHeader(status) + if status != 304 { + _, _ = w.Write(body) + } + })) + defer server.Close() + cfg := config.MeshMapperScopesConfig{Enabled: true, Sources: map[string]string{"YOW": server.URL}} + store := &memoryStore{rows: map[string]Cache{}} + scopes := scopestore.New() + manual := scopestore.FromName("manual") + manual.TransportKey = []byte("manual wins") + imp, err := New(ctx, cfg, store, scopes, []scopestore.Entry{manual}) + if err != nil { + t.Fatal(err) + } + now := time.Now().UTC() + check := func() { + t.Helper() + now = now.Add(time.Minute) + if err := imp.refresh(ctx, &imp.sources[0], now); err != nil { + t.Fatal(err) + } + } + check() + if got := names(scopes); !reflect.DeepEqual(got, []string{"#manual", "#yow"}) { + t.Fatal(got) + } + if string(scopes.Entries()[0].TransportKey) != "manual wins" || scopes.Entries()[0].IATAs != nil { + t.Fatal("manual key replaced/restricted") + } + if !reflect.DeepEqual(scopes.Entries()[1].IATAs, []string{"YOW"}) { + t.Fatal("import not bound to region") + } + generated := imp.sources[0].generated + status = 304 + check() + checked := imp.sources[0].cache.CheckedAt + if checked != now || imp.sources[0].generated != generated { + t.Fatal("304 lost freshness or changed generation time") + } + for _, failure := range []int{200, 404, 503, 302} { + status, body = failure, []byte(`{"scopes":null}`) + check() + if imp.sources[0].cache.CheckedAt != checked || len(scopes.Entries()) != 2 { + t.Fatal("failed refresh replaced known catalogue") + } + } + status, after = 429, "7200" + check() + if !imp.retryAfter.Equal(now.Add(2 * time.Hour)) { + t.Fatal("Retry-After ignored") + } + restored := scopestore.New() + restart, err := New(ctx, cfg, store, restored, []scopestore.Entry{manual}) + if err != nil { + t.Fatal(err) + } + if !reflect.DeepEqual(names(restored), names(scopes)) { + t.Fatal("outage restart lost cached names") + } + before := calls + if err := restart.Refresh(ctx); err != nil { + t.Fatal(err) + } + if calls != before { + t.Fatal("restart ignored shared rate-limit cooldown") + } + status, after, body = 200, "", catalogue() + check() + if got := names(scopes); !reflect.DeepEqual(got, []string{"#manual"}) { + t.Fatalf("empty snapshot: %v", got) + } + store.fail = true + body = catalogue("new") + now = now.Add(time.Hour) + if err := imp.refresh(ctx, &imp.sources[0], now); err == nil || len(scopes.Entries()) != 1 { + t.Fatal("uncommitted names became active") + } +} + +func TestDecodeCatalogueBoundaries(t *testing.T) { + valid := string(catalogue("yow", "YOW", "$private")) + bad := []string{ + `null`, `{}`, `[]`, valid + `{}`, + strings.Replace(valid, `"region":"YOW"`, `"region":"PNW"`, 1), + strings.Replace(valid, `"zones":["YOW"]`, `"zones":["YOW","YVR"]`, 1), + strings.Replace(valid, `"scoped":3`, `"scoped":6`, 1), + strings.Replace(valid, `"repeaters":0`, `"repeaters":-1`, 1), + strings.Replace(valid, `"monitored":true,`, ``, 1), + strings.Replace(valid, `"generated_at":"2026-09-27T16:58:49Z"`, `"generated_at":null`, 1), + string(catalogue("yow", "#yow")), string(catalogue("#")), string(catalogue(" yow")), string(catalogue("a\nb")), + strings.Repeat(" ", MaxBody+1), + } + tooMany := []string{} + for n := 0; n <= MaxScopes; n++ { + tooMany = append(tooMany, fmt.Sprint(n)) + } + bad = append(bad, string(catalogue(tooMany...))) + for n, raw := range bad { + if _, _, err := decode([]byte(raw), "YOW"); err == nil { + t.Errorf("invalid case %d accepted", n) + } + } + entries, _, err := decode([]byte(valid), "YOW") + if err != nil || len(entries) != 3 { + t.Fatal(entries, err) + } + if entries[0].Name != "#yow" || entries[1].Name != "#YOW" || entries[2].Name != "$private" { + t.Fatal("case/prefix changed", entries) + } +} + +func TestNoRedirectAndResponseBounds(t *testing.T) { + targetCalls := 0 + target := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { targetCalls++ })) + defer target.Close() + mode := "redirect" + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch mode { + case "redirect": + http.Redirect(w, r, target.URL, 302) + case "oversize": + _, _ = w.Write([]byte(strings.Repeat("x", MaxBody+1))) + case "timeout": + select { + case <-r.Context().Done(): + case <-time.After(time.Second): + } + case "304": + w.WriteHeader(304) + } + })) + defer server.Close() + store := &memoryStore{rows: map[string]Cache{}} + imp, err := New(context.Background(), config.MeshMapperScopesConfig{Enabled: true, Sources: map[string]string{"YOW": server.URL}}, store, scopestore.New(), nil) + if err != nil { + t.Fatal(err) + } + imp.client.Timeout = 20 * time.Millisecond + for _, value := range []string{"redirect", "oversize", "timeout", "304"} { + mode = value + if err := imp.refresh(context.Background(), &imp.sources[0], time.Now()); err != nil { + t.Fatal(err) + } + if imp.sources[0].cache.LastError == "" || len(imp.sources[0].entries) != 0 { + t.Fatal("invalid response accepted", value) + } + } + if targetCalls != 0 { + t.Fatal("followed redirect") + } +} + +func TestImportedSourcesOverlapWithoutGlobalMembership(t *testing.T) { + ctx := context.Background() + store := &memoryStore{rows: map[string]Cache{}} + for _, iata := range []string{"YOW", "YVR"} { + store.rows[iata+iata] = Cache{Payload: []byte(strings.ReplaceAll(string(catalogue("can")), "YOW", iata))} + } + scopes := scopestore.New() + _, err := New(ctx, config.MeshMapperScopesConfig{Enabled: true, Sources: map[string]string{"YOW": "YOW", "YVR": "YVR"}}, store, scopes, nil) + if err != nil || len(scopes.Entries()) != 1 || !reflect.DeepEqual(scopes.Entries()[0].IATAs, []string{"YOW", "YVR"}) { + t.Fatal(scopes.Entries(), err) + } + _, err = New(ctx, config.MeshMapperScopesConfig{Enabled: true, Sources: map[string]string{"YOW": "changed-source"}}, store, scopes, nil) + if err != nil || len(scopes.Entries()) != 0 { + t.Fatal("removed/changed source remained active", err) + } + // Disabled mode never calls the store, even if URLs remain configured. + _, err = New(ctx, config.MeshMapperScopesConfig{Sources: map[string]string{"YOW": "YOW"}}, nil, scopes, nil) + if err != nil { + t.Fatal(err) + } +} diff --git a/internal/scopestore/scopestore.go b/internal/scopestore/scopestore.go index 8abbac3..b95bad8 100644 --- a/internal/scopestore/scopestore.go +++ b/internal/scopestore/scopestore.go @@ -5,13 +5,29 @@ // loaded from the database at startup. package scopestore -import "sync" +import ( + "crypto/sha256" + "strings" + "sync" +) // Entry holds a single transport scope key and its metadata. type Entry struct { Name string - TransportKey []byte // 16 bytes - KeyFingerprint []byte // 8 bytes + TransportKey []byte // 16 bytes + KeyFingerprint []byte // 8 bytes + IATAs []string // nil for manual keys; imported candidates are regional +} + +// FromName uses the same case-sensitive derivation for manual and imported scopes. +func FromName(name string) Entry { + if !strings.HasPrefix(name, "#") && !strings.HasPrefix(name, "$") { + name = "#" + name + } + h := sha256.Sum256([]byte(name)) + key := h[:16] + fingerprint := sha256.Sum256(key) + return Entry{Name: name, TransportKey: key, KeyFingerprint: fingerprint[:8]} } // ScopeStore holds all known transport scope keys in memory.