feat(ingestor + server): observerBlacklist config (#962) (#963)

## 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 <pubkey-short> 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 <you@example.com>

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) {

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/<region>/<observer_id>/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

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")
+	}
+}

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))]
+}

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
 

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 {

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
+	}
+}

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")