From b3a9677c52004bf7ca328345f3f8f6544167cafc Mon Sep 17 00:00:00 2001 From: Kpa-clawbot Date: Fri, 1 May 2026 23:11:27 -0700 Subject: [PATCH] feat(ingestor + server): observerBlacklist config (#962) (#963) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary Implements `observerBlacklist` config — mirrors the existing `nodeBlacklist` pattern for observers. Drop observers by pubkey at ingest, with defense-in-depth filtering on the server side. Closes #962 ## Changes ### Ingestor (`cmd/ingestor/`) - **`config.go`**: Added `ObserverBlacklist []string` field + `IsObserverBlacklisted()` method (case-insensitive, whitespace-trimmed) - **`main.go`**: Early return in `handleMessage` when `parts[2]` (observer ID from MQTT topic) matches blacklist — before status handling, before IATA filter. No UpsertObserver, no observations, no metrics insert. Log line: `observer blacklisted, dropping` ### Server (`cmd/server/`) - **`config.go`**: Same `ObserverBlacklist` field + `IsObserverBlacklisted()` with `sync.Once` cached set (same pattern as `nodeBlacklist`) - **`routes.go`**: Defense-in-depth filtering in `handleObservers` (skip blacklisted in list) and `handleObserverDetail` (404 for blacklisted ID) - **`main.go`**: Startup `softDeleteBlacklistedObservers()` marks matching rows `inactive=1` so historical data is hidden - **`neighbor_persist.go`**: `softDeleteBlacklistedObservers()` implementation ### Tests - `cmd/ingestor/observer_blacklist_test.go`: config method tests (case-insensitive, empty, nil) - `cmd/server/observer_blacklist_test.go`: config tests + HTTP handler tests (list excludes blacklisted, detail returns 404, no-blacklist passes all, concurrent safety) ## Config ```json { "observerBlacklist": [ "EE550DE547D7B94848A952C98F585881FCF946A128E72905E95517475F83CFB1" ] } ``` ## Verification (Rule 18 — actual server output) **Before blacklist** (no config): ``` Total: 31 DUBLIN in list: True ``` **After blacklist** (DUBLIN Observer pubkey in `observerBlacklist`): ``` [observer-blacklist] soft-deleted 1 blacklisted observer(s) Total: 30 DUBLIN in list: False ``` Detail endpoint for blacklisted observer returns **404**. All existing tests pass (`go test ./...` for both server and ingestor). --------- Co-authored-by: you --- cmd/ingestor/config.go | 28 +++++ cmd/ingestor/main.go | 7 ++ cmd/ingestor/observer_blacklist_test.go | 43 +++++++ cmd/server/config.go | 35 ++++++ cmd/server/main.go | 6 + cmd/server/neighbor_persist.go | 35 ++++++ cmd/server/observer_blacklist_test.go | 159 ++++++++++++++++++++++++ cmd/server/routes.go | 11 ++ 8 files changed, 324 insertions(+) create mode 100644 cmd/ingestor/observer_blacklist_test.go create mode 100644 cmd/server/observer_blacklist_test.go diff --git a/cmd/ingestor/config.go b/cmd/ingestor/config.go index 910d3b95..e6ab2d26 100644 --- a/cmd/ingestor/config.go +++ b/cmd/ingestor/config.go @@ -7,6 +7,7 @@ import ( "log" "os" "strings" + "sync" "github.com/meshcore-analyzer/geofilter" ) @@ -42,6 +43,15 @@ type Config struct { GeoFilter *GeoFilterConfig `json:"geo_filter,omitempty"` ValidateSignatures *bool `json:"validateSignatures,omitempty"` DB *DBConfig `json:"db,omitempty"` + + // ObserverBlacklist is a list of observer public keys to drop at ingest. + // Messages from blacklisted observers are silently discarded — no DB writes, + // no UpsertObserver, no observations, no metrics. + ObserverBlacklist []string `json:"observerBlacklist,omitempty"` + + // obsBlacklistSetCached is the lazily-built lowercase set for O(1) lookups. + obsBlacklistSetCached map[string]bool + obsBlacklistOnce sync.Once } // GeoFilterConfig is an alias for the shared geofilter.Config type. @@ -114,6 +124,24 @@ func (c *Config) ObserverDaysOrDefault() int { return 14 } +// IsObserverBlacklisted returns true if the given observer ID is in the observerBlacklist. +func (c *Config) IsObserverBlacklisted(id string) bool { + if c == nil || len(c.ObserverBlacklist) == 0 { + return false + } + c.obsBlacklistOnce.Do(func() { + m := make(map[string]bool, len(c.ObserverBlacklist)) + for _, pk := range c.ObserverBlacklist { + trimmed := strings.ToLower(strings.TrimSpace(pk)) + if trimmed != "" { + m[trimmed] = true + } + } + c.obsBlacklistSetCached = m + }) + return c.obsBlacklistSetCached[strings.ToLower(strings.TrimSpace(id))] +} + // LoadConfig reads configuration from a JSON file, with env var overrides. // If the config file does not exist, sensible defaults are used (zero-config startup). func LoadConfig(path string) (*Config, error) { diff --git a/cmd/ingestor/main.go b/cmd/ingestor/main.go index b9dad90f..9984b13a 100644 --- a/cmd/ingestor/main.go +++ b/cmd/ingestor/main.go @@ -240,6 +240,13 @@ func handleMessage(store *Store, tag string, source MQTTSource, m mqtt.Message, return } + // Observer blacklist: drop ALL messages from blacklisted observers before any + // DB writes (status, metrics, packets). Trumps IATA filter. + if len(parts) > 2 && cfg.IsObserverBlacklisted(parts[2]) { + log.Printf("MQTT [%s] observer %.8s blacklisted, dropping", tag, parts[2]) + return + } + // Status topic: meshcore///status // IATA filter does NOT apply here — observer metadata (noise_floor, battery, etc.) // is region-independent and should be accepted from all observers regardless of diff --git a/cmd/ingestor/observer_blacklist_test.go b/cmd/ingestor/observer_blacklist_test.go new file mode 100644 index 00000000..5f2e7aef --- /dev/null +++ b/cmd/ingestor/observer_blacklist_test.go @@ -0,0 +1,43 @@ +package main + +import ( + "testing" +) + +func TestIngestorIsObserverBlacklisted(t *testing.T) { + cfg := &Config{ + ObserverBlacklist: []string{"OBS1", "obs2"}, + } + + tests := []struct { + id string + want bool + }{ + {"OBS1", true}, + {"obs1", true}, + {"OBS2", true}, + {"obs3", false}, + {"", false}, + } + + for _, tt := range tests { + got := cfg.IsObserverBlacklisted(tt.id) + if got != tt.want { + t.Errorf("IsObserverBlacklisted(%q) = %v, want %v", tt.id, got, tt.want) + } + } +} + +func TestIngestorIsObserverBlacklistedEmpty(t *testing.T) { + cfg := &Config{} + if cfg.IsObserverBlacklisted("anything") { + t.Error("empty blacklist should not match") + } +} + +func TestIngestorIsObserverBlacklistedNil(t *testing.T) { + var cfg *Config + if cfg.IsObserverBlacklisted("anything") { + t.Error("nil config should not match") + } +} diff --git a/cmd/server/config.go b/cmd/server/config.go index f21ef207..d5f5f569 100644 --- a/cmd/server/config.go +++ b/cmd/server/config.go @@ -72,6 +72,15 @@ type Config struct { DebugAffinity bool `json:"debugAffinity,omitempty"` + // ObserverBlacklist is a list of observer public keys to exclude from API + // responses (defense in depth — ingestor drops at ingest, server filters + // any that slipped through from a prior unblocked window). + ObserverBlacklist []string `json:"observerBlacklist,omitempty"` + + // obsBlacklistSetCached is the lazily-built set version of ObserverBlacklist. + obsBlacklistSetCached map[string]bool + obsBlacklistOnce sync.Once + ResolvedPath *ResolvedPathConfig `json:"resolvedPath,omitempty"` NeighborGraph *NeighborGraphConfig `json:"neighborGraph,omitempty"` } @@ -404,3 +413,29 @@ func (c *Config) IsBlacklisted(pubkey string) bool { } return c.blacklistSet()[strings.ToLower(strings.TrimSpace(pubkey))] } + +// obsBlacklistSet lazily builds and caches the observerBlacklist as a set for O(1) lookups. +func (c *Config) obsBlacklistSet() map[string]bool { + c.obsBlacklistOnce.Do(func() { + if len(c.ObserverBlacklist) == 0 { + return + } + m := make(map[string]bool, len(c.ObserverBlacklist)) + for _, pk := range c.ObserverBlacklist { + trimmed := strings.ToLower(strings.TrimSpace(pk)) + if trimmed != "" { + m[trimmed] = true + } + } + c.obsBlacklistSetCached = m + }) + return c.obsBlacklistSetCached +} + +// IsObserverBlacklisted returns true if the given observer ID is in the observerBlacklist. +func (c *Config) IsObserverBlacklisted(id string) bool { + if c == nil || len(c.ObserverBlacklist) == 0 { + return false + } + return c.obsBlacklistSet()[strings.ToLower(strings.TrimSpace(id))] +} diff --git a/cmd/server/main.go b/cmd/server/main.go index aea2c305..17b9c0fe 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -180,6 +180,12 @@ func main() { log.Printf("[store] warning: could not add observers.inactive column: %v", err) } + // Soft-delete observers that are in the blacklist (mark inactive=1) so + // historical data from a prior unblocked window is hidden too. + if len(cfg.ObserverBlacklist) > 0 { + softDeleteBlacklistedObservers(dbPath, cfg.ObserverBlacklist) + } + // WaitGroup for background init steps that gate /api/healthz readiness. var initWg sync.WaitGroup diff --git a/cmd/server/neighbor_persist.go b/cmd/server/neighbor_persist.go index feb77efb..58637e86 100644 --- a/cmd/server/neighbor_persist.go +++ b/cmd/server/neighbor_persist.go @@ -320,6 +320,41 @@ func ensureObserverInactiveColumn(dbPath string) error { return nil } +// softDeleteBlacklistedObservers marks observers matching the blacklist as +// inactive=1 so they are hidden from API responses. Runs once at startup. +func softDeleteBlacklistedObservers(dbPath string, blacklist []string) { + rw, err := openRW(dbPath) + if err != nil { + log.Printf("[observer-blacklist] warning: could not open DB for soft-delete: %v", err) + return + } + defer rw.Close() + + placeholders := make([]string, 0, len(blacklist)) + args := make([]interface{}, 0, len(blacklist)) + for _, pk := range blacklist { + trimmed := strings.TrimSpace(pk) + if trimmed == "" { + continue + } + placeholders = append(placeholders, "LOWER(?)") + args = append(args, trimmed) + } + if len(placeholders) == 0 { + return + } + + query := "UPDATE observers SET inactive = 1 WHERE LOWER(id) IN (" + strings.Join(placeholders, ",") + ") AND (inactive IS NULL OR inactive = 0)" + result, err := rw.Exec(query, args...) + if err != nil { + log.Printf("[observer-blacklist] warning: soft-delete failed: %v", err) + return + } + if n, _ := result.RowsAffected(); n > 0 { + log.Printf("[observer-blacklist] soft-deleted %d blacklisted observer(s)", n) + } +} + // resolvePathForObs resolves hop prefixes to full pubkeys for an observation. // Returns nil if path is empty. func resolvePathForObs(pathJSON, observerID string, tx *StoreTx, pm *prefixMap, graph *NeighborGraph) []*string { diff --git a/cmd/server/observer_blacklist_test.go b/cmd/server/observer_blacklist_test.go new file mode 100644 index 00000000..32ccd71c --- /dev/null +++ b/cmd/server/observer_blacklist_test.go @@ -0,0 +1,159 @@ +package main + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "testing" +) + +func TestConfigIsObserverBlacklisted(t *testing.T) { + cfg := &Config{ + ObserverBlacklist: []string{"OBS1", "obs2", " Obs3 "}, + } + + tests := []struct { + id string + want bool + }{ + {"OBS1", true}, + {"obs1", true}, // case-insensitive + {"OBS2", true}, + {"Obs3", true}, // whitespace trimmed + {"obs4", false}, + {"", false}, + } + + for _, tt := range tests { + got := cfg.IsObserverBlacklisted(tt.id) + if got != tt.want { + t.Errorf("IsObserverBlacklisted(%q) = %v, want %v", tt.id, got, tt.want) + } + } +} + +func TestConfigIsObserverBlacklistedEmpty(t *testing.T) { + cfg := &Config{} + if cfg.IsObserverBlacklisted("anything") { + t.Error("empty blacklist should not match anything") + } +} + +func TestConfigIsObserverBlacklistedNil(t *testing.T) { + var cfg *Config + if cfg.IsObserverBlacklisted("anything") { + t.Error("nil config should not match anything") + } +} + +func TestObserverBlacklistFiltersHandleObservers(t *testing.T) { + db := setupTestDB(t) + db.conn.Exec("INSERT OR IGNORE INTO observers (id, name, iata, last_seen) VALUES ('goodobs', 'GoodObs', 'SFO', datetime('now'))") + db.conn.Exec("INSERT OR IGNORE INTO observers (id, name, iata, last_seen) VALUES ('badobs', 'BadObs', 'LAX', datetime('now'))") + + cfg := &Config{ + ObserverBlacklist: []string{"badobs"}, + } + srv := NewServer(db, cfg, NewHub()) + srv.RegisterRoutes(setupTestRouter(srv)) + + req := httptest.NewRequest("GET", "/api/observers", nil) + w := httptest.NewRecorder() + srv.router.ServeHTTP(w, req) + + if w.Code != http.StatusOK { + t.Fatalf("expected 200, got %d: %s", w.Code, w.Body.String()) + } + + var resp ObserverListResponse + if err := json.Unmarshal(w.Body.Bytes(), &resp); err != nil { + t.Fatalf("failed to parse response: %v", err) + } + + for _, obs := range resp.Observers { + if obs.ID == "badobs" { + t.Error("blacklisted observer should not appear in observers list") + } + } + + foundGood := false + for _, obs := range resp.Observers { + if obs.ID == "goodobs" { + foundGood = true + } + } + if !foundGood { + t.Error("non-blacklisted observer should appear in observers list") + } +} + +func TestObserverBlacklistFiltersObserverDetail(t *testing.T) { + db := setupTestDB(t) + db.conn.Exec("INSERT OR IGNORE INTO observers (id, name, iata, last_seen) VALUES ('badobs', 'BadObs', 'LAX', datetime('now'))") + + cfg := &Config{ + ObserverBlacklist: []string{"badobs"}, + } + srv := NewServer(db, cfg, NewHub()) + srv.RegisterRoutes(setupTestRouter(srv)) + + req := httptest.NewRequest("GET", "/api/observers/badobs", nil) + w := httptest.NewRecorder() + srv.router.ServeHTTP(w, req) + + if w.Code != http.StatusNotFound { + t.Errorf("expected 404 for blacklisted observer detail, got %d", w.Code) + } +} + +func TestNoObserverBlacklistPassesAll(t *testing.T) { + db := setupTestDB(t) + db.conn.Exec("INSERT OR IGNORE INTO observers (id, name, iata, last_seen) VALUES ('someobs', 'SomeObs', 'SFO', datetime('now'))") + + cfg := &Config{} + srv := NewServer(db, cfg, NewHub()) + srv.RegisterRoutes(setupTestRouter(srv)) + + req := httptest.NewRequest("GET", "/api/observers", nil) + w := httptest.NewRecorder() + srv.router.ServeHTTP(w, req) + + if w.Code != http.StatusOK { + t.Fatalf("expected 200, got %d", w.Code) + } + + var resp ObserverListResponse + if err := json.Unmarshal(w.Body.Bytes(), &resp); err != nil { + t.Fatalf("failed to parse response: %v", err) + } + + foundSome := false + for _, obs := range resp.Observers { + if obs.ID == "someobs" { + foundSome = true + } + } + if !foundSome { + t.Error("without blacklist, observer should appear") + } +} + +func TestObserverBlacklistConcurrent(t *testing.T) { + cfg := &Config{ + ObserverBlacklist: []string{"AA", "BB", "CC"}, + } + + done := make(chan struct{}) + for i := 0; i < 50; i++ { + go func() { + defer func() { done <- struct{}{} }() + for j := 0; j < 100; j++ { + cfg.IsObserverBlacklisted("AA") + cfg.IsObserverBlacklisted("DD") + } + }() + } + for i := 0; i < 50; i++ { + <-done + } +} diff --git a/cmd/server/routes.go b/cmd/server/routes.go index 03ac7542..5bd0562f 100644 --- a/cmd/server/routes.go +++ b/cmd/server/routes.go @@ -1932,6 +1932,10 @@ func (s *Server) handleObservers(w http.ResponseWriter, r *http.Request) { result := make([]ObserverResp, 0, len(observers)) for _, o := range observers { + // Defense in depth: skip observers that are in the blacklist + if s.cfg != nil && s.cfg.IsObserverBlacklisted(o.ID) { + continue + } plh := 0 if c, ok := pktCounts[o.ID]; ok { plh = c @@ -1963,6 +1967,13 @@ func (s *Server) handleObservers(w http.ResponseWriter, r *http.Request) { func (s *Server) handleObserverDetail(w http.ResponseWriter, r *http.Request) { id := mux.Vars(r)["id"] + + // Defense in depth: reject blacklisted observer + if s.cfg != nil && s.cfg.IsObserverBlacklisted(id) { + writeError(w, 404, "Observer not found") + return + } + obs, err := s.db.GetObserverByID(id) if err != nil || obs == nil { writeError(w, 404, "Observer not found")