From e8f32df4dcef22e5c09c6d58ad9ecbda2d4c1e0a Mon Sep 17 00:00:00 2001 From: efiten Date: Wed, 2 Sep 2026 11:07:18 +0200 Subject: [PATCH] feat(#1784): gate ingestor neighbor-edge creation on the path-trust threshold (rebase of #1863) (#1930) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Continues #1863. Three of the four commits are @Saarlandpower's and @SaarMesh-Bot's, authorship unchanged. The fourth is mine and is explained below. ## Why a rebase was needed #1863 was stacked on #1824, and #1841 merged instead. Both carried the same pathTrust base from different commits, which is why the two conflicted while each reported MERGEABLE against master. Cherry-picking #1863's own three commits onto master applied cleanly with no conflicts, which confirms its actual work was always independent of that duplicated base. ## The fourth commit, and a correction to something I got wrong The three commits do not build on master: ``` cmd/ingestor/main.go:455:23: cfg.GetPathTrust undefined (type *Config has no field or method GetPathTrust) ``` **#1824 added the pathTrust config and helper to both `cmd/server/config.go` and `cmd/ingestor/config.go`. #1841 carried only the server half** — one of its own commits is titled "remove ingestor side". I then closed #1824 as superseded by #1841, which is true for the server side and wrong for the ingestor side. Master has no pathTrust code in `cmd/ingestor/config.go` at all. The fourth commit restores that half, unchanged from `beae2c1c`: the `packetpath` import, the `PathTrust` field, the `PathTrustConfig` alias, `GetPathTrust`, and `cmd/ingestor/config_test.go` verbatim (28 lines covering the default, an explicit value, and a nil `*Config` receiver). That code is @Bjorkan's and @SaarMesh-Bot's from #1824, not mine; I only put it back. ## Verification - All three original commits cherry-picked onto `b3a306b8` with **no conflicts** - `cmd/ingestor` builds, and its `PathTrust|Neighbor|Config` tests pass - `cmd/server` `Neighbor|PathTrust|AnonReq|Edge` tests pass ## Interaction with #1929 #1929 moves `DefaultMinHashBytesForMapping` from 2 to 1. With that in, this PR's ingestor gate is a no-op by default and only takes effect when an operator sets `minHashBytesForMapping` to 2 or 3, which is the opt-in shape #1784 asks for. The two are complementary; merge order between them does not matter. @Saarlandpower @SaarMesh-Bot — your work, your credit. Say the word and I will close this and hand the rebase back, or push it to the #1863 branch if you would rather that stayed the vehicle. 🤖 Generated with [Claude Code](https://claude.com/claude-code) https://claude.ai/code/session_01Wzwr3eXseyNM7Xj598djjE --------- Co-authored-by: Saarlandpower Co-authored-by: Claude Co-authored-by: SaarMesh-Bot <300107934+SaarMesh-Bot@users.noreply.github.com> --- cmd/ingestor/config.go | 50 ++-- cmd/ingestor/config_test.go | 28 +++ cmd/ingestor/main.go | 6 +- cmd/ingestor/maintenance.go | 19 ++ cmd/ingestor/neighbor_builder.go | 54 ++++- cmd/ingestor/neighbor_builder_delta_test.go | 6 +- cmd/ingestor/neighbor_builder_test.go | 245 +++++++++++++++++++- cmd/server/neighbor_graph.go | 34 ++- cmd/server/neighbor_graph_test.go | 77 ++++++ cmd/server/neighbor_persist.go | 72 ------ 10 files changed, 481 insertions(+), 110 deletions(-) diff --git a/cmd/ingestor/config.go b/cmd/ingestor/config.go index be1d55f4..5074b976 100644 --- a/cmd/ingestor/config.go +++ b/cmd/ingestor/config.go @@ -11,6 +11,7 @@ import ( "github.com/meshcore-analyzer/dbconfig" "github.com/meshcore-analyzer/geofilter" + "github.com/meshcore-analyzer/packetpath" ) // MQTTSource represents a single MQTT broker connection. @@ -43,22 +44,26 @@ type MQTTLegacy struct { // Config holds the ingestor configuration, compatible with the Node.js config.json format. type Config struct { - DBPath string `json:"dbPath"` - MQTT *MQTTLegacy `json:"mqtt,omitempty"` - MQTTSources []MQTTSource `json:"mqttSources,omitempty"` - LogLevel string `json:"logLevel,omitempty"` - ChannelKeysPath string `json:"channelKeysPath,omitempty"` - ChannelKeys map[string]string `json:"channelKeys,omitempty"` - HashChannels []string `json:"hashChannels,omitempty"` - HashRegions []string `json:"hashRegions,omitempty"` - Retention *RetentionConfig `json:"retention,omitempty"` - Metrics *MetricsConfig `json:"metrics,omitempty"` - Runtime *RuntimeConfig `json:"runtime,omitempty"` - ClientRxCoverage *ClientRxCoverageConfig `json:"clientRxCoverage,omitempty"` - GeoFilter *GeoFilterConfig `json:"geo_filter,omitempty"` - ForeignAdverts *ForeignAdvertConfig `json:"foreignAdverts,omitempty"` - ValidateSignatures *bool `json:"validateSignatures,omitempty"` - DB *DBConfig `json:"db,omitempty"` + DBPath string `json:"dbPath"` + MQTT *MQTTLegacy `json:"mqtt,omitempty"` + MQTTSources []MQTTSource `json:"mqttSources,omitempty"` + LogLevel string `json:"logLevel,omitempty"` + ChannelKeysPath string `json:"channelKeysPath,omitempty"` + ChannelKeys map[string]string `json:"channelKeys,omitempty"` + HashChannels []string `json:"hashChannels,omitempty"` + HashRegions []string `json:"hashRegions,omitempty"` + Retention *RetentionConfig `json:"retention,omitempty"` + Metrics *MetricsConfig `json:"metrics,omitempty"` + Runtime *RuntimeConfig `json:"runtime,omitempty"` + ClientRxCoverage *ClientRxCoverageConfig `json:"clientRxCoverage,omitempty"` + GeoFilter *GeoFilterConfig `json:"geo_filter,omitempty"` + // PathTrust configures the minimum path-hash prefix length trusted as + // mapping/topology evidence (issue #1784). Read by the neighbor-edge + // builder; see packetpath.MeetsPathTrust. + PathTrust *PathTrustConfig `json:"pathTrust,omitempty"` + ForeignAdverts *ForeignAdvertConfig `json:"foreignAdverts,omitempty"` + ValidateSignatures *bool `json:"validateSignatures,omitempty"` + DB *DBConfig `json:"db,omitempty"` // ObserverIATAWhitelist restricts which observer IATA regions are processed. // When non-empty, only observers whose IATA code (from the MQTT topic) matches @@ -112,6 +117,10 @@ func (c *Config) IngestBufferSizeOrDefault() int { // GeoFilterConfig is an alias for the shared geofilter.Config type. type GeoFilterConfig = geofilter.Config +// PathTrustConfig is an alias for the shared packetpath.TrustConfig type +// (issue #1784). See packetpath.TrustConfig for the full doc comment. +type PathTrustConfig = packetpath.TrustConfig + // ForeignAdvertConfig controls how the ingestor handles ADVERTs whose GPS lies // outside the configured geofilter polygon (#730). Modes: // - "flag" (default): store the advert/node and tag it foreign for visibility. @@ -238,6 +247,15 @@ func (c *Config) ObserverDaysOrDefault() int { return 14 } +// GetPathTrust returns the effective path-trust config, applying +// DefaultMinHashBytesForMapping when unset (issue #1784). +func (c *Config) GetPathTrust() PathTrustConfig { + if c != nil && c.PathTrust != nil { + return *c.PathTrust + } + return PathTrustConfig{MinHashBytesForMapping: packetpath.DefaultMinHashBytesForMapping} +} + // 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 { diff --git a/cmd/ingestor/config_test.go b/cmd/ingestor/config_test.go index ec55900f..71e9a681 100644 --- a/cmd/ingestor/config_test.go +++ b/cmd/ingestor/config_test.go @@ -4,6 +4,8 @@ import ( "os" "path/filepath" "testing" + + "github.com/meshcore-analyzer/packetpath" ) func TestLoadConfigValidJSON(t *testing.T) { @@ -496,3 +498,29 @@ func TestIngestBufferSizeOrDefault(t *testing.T) { t.Fatalf("invalid negative should fall back to default, got %d", got) } } + +// ─── #1784: GetPathTrust ──────────────────────────────────────────────────── + +func TestGetPathTrustDefaults(t *testing.T) { + cfg := &Config{} + pt := cfg.GetPathTrust() + if pt.MinHashBytesForMapping != packetpath.DefaultMinHashBytesForMapping { + t.Errorf("expected default %d, got %d", packetpath.DefaultMinHashBytesForMapping, pt.MinHashBytesForMapping) + } +} + +func TestGetPathTrustCustom(t *testing.T) { + cfg := &Config{PathTrust: &PathTrustConfig{MinHashBytesForMapping: 3}} + pt := cfg.GetPathTrust() + if pt.MinHashBytesForMapping != 3 { + t.Errorf("expected 3, got %d", pt.MinHashBytesForMapping) + } +} + +func TestGetPathTrustNilConfig(t *testing.T) { + var cfg *Config + pt := cfg.GetPathTrust() + if pt.MinHashBytesForMapping != packetpath.DefaultMinHashBytesForMapping { + t.Errorf("expected default %d for nil *Config, got %d", packetpath.DefaultMinHashBytesForMapping, pt.MinHashBytesForMapping) + } +} diff --git a/cmd/ingestor/main.go b/cmd/ingestor/main.go index f4cc283a..963be045 100644 --- a/cmd/ingestor/main.go +++ b/cmd/ingestor/main.go @@ -445,7 +445,11 @@ func main() { // Neighbor-edges builder (#1287 — Option 4): ingestor owns // neighbor_edges writes. Runs every 60s. Server reads the snapshot // via cmd/server/neighbor_recomputer.go on the same cadence. - stopNeighborBuilder := store.StartNeighborEdgesBuilder(NeighborEdgesBuilderInterval) + // #1784: the neighbor builder is the first real consumer of the + // path-trust threshold. Resolved once here so every tick shares the + // same operator-configured value. + neighborTrust := cfg.GetPathTrust() + stopNeighborBuilder := store.StartNeighborEdgesBuilder(NeighborEdgesBuilderInterval, &neighborTrust) defer stopNeighborBuilder() log.Printf("[neighbor-build] enabled (interval=%s)", NeighborEdgesBuilderInterval) diff --git a/cmd/ingestor/maintenance.go b/cmd/ingestor/maintenance.go index fea189fe..37043056 100644 --- a/cmd/ingestor/maintenance.go +++ b/cmd/ingestor/maintenance.go @@ -252,3 +252,22 @@ func extractPubkeyFromAdvertJSON(s string) string { } return "" } + +// extractPubkeyFromAnonReqJSON parses an ANON_REQ decoded_json blob and +// returns the ephemeralPubKey field, or "" if absent/invalid (#1777). +// ANON_REQ carries the sender's full Ed25519 ephemeral pubkey — the same +// trust level as ADVERT's pubKey — unlike REQ/RESP/PATH/TXT, which only +// carry a 1-byte truncated hash of the originator. +func extractPubkeyFromAnonReqJSON(s string) string { + if s == "" { + return "" + } + var m map[string]interface{} + if err := json.Unmarshal([]byte(s), &m); err != nil { + return "" + } + if v, ok := m["ephemeralPubKey"].(string); ok { + return v + } + return "" +} diff --git a/cmd/ingestor/neighbor_builder.go b/cmd/ingestor/neighbor_builder.go index ff5782fe..7a49ecc9 100644 --- a/cmd/ingestor/neighbor_builder.go +++ b/cmd/ingestor/neighbor_builder.go @@ -8,6 +8,8 @@ import ( "strings" "sync" "time" + + "github.com/meshcore-analyzer/packetpath" ) // NeighborEdgesBuilderInterval is how often the ingestor rescans @@ -35,6 +37,15 @@ const neighborBuilderSlowTickThreshold = 5 * time.Second // independent of the server package. const payloadADVERT = 0x04 +// payloadAnonReq mirrors PayloadANON_REQ in cmd/server/decoder.go (#1777). +// ANON_REQ carries the sender's full Ed25519 ephemeral pubkey +// (decoder.go's EphemeralPubKey field), unlike REQ/RESP/PATH/TXT which +// only carry a 1-byte truncated hash of the originator in src/dst — not +// enough to seed a trustworthy neighbor edge (~1/256 collision odds). +// ANON_REQ is therefore treated like ADVERT for the originator↔path[0] +// edge; other non-ADVERT types are deliberately excluded. +const payloadAnonReq = 0x07 + // edgeRow is one row to upsert into neighbor_edges. (a, b) is already // canonical-ordered (a <= b). type edgeRow struct { @@ -49,7 +60,7 @@ type edgeRow struct { // The function returns a stop closure. Initial build runs synchronously // before the ticker starts so the server's first snapshot load picks // up real data instead of an empty table. -func (s *Store) StartNeighborEdgesBuilder(interval time.Duration) func() { +func (s *Store) StartNeighborEdgesBuilder(interval time.Duration, trust *packetpath.TrustConfig) func() { if interval <= 0 { interval = NeighborEdgesBuilderInterval } @@ -74,7 +85,7 @@ func (s *Store) StartNeighborEdgesBuilder(interval time.Duration) func() { log.Printf("[neighbor-build] initial neighbor-graph refresh error: %v", err) } for { - n, err := s.buildAndPersistNeighborEdges() + n, err := s.buildAndPersistNeighborEdges(trust) if err != nil { log.Printf("[neighbor-build] initial build error: %v", err) break @@ -100,7 +111,7 @@ func (s *Store) StartNeighborEdgesBuilder(interval time.Duration) func() { if err := s.RefreshPrefixIndex(); err != nil { log.Printf("[neighbor-build] prefix-index refresh error: %v", err) } - n, err := s.buildAndPersistNeighborEdges() + n, err := s.buildAndPersistNeighborEdges(trust) // Refresh the neighbor-graph snapshot after the edges // build (#1560) so the context-aware resolver picks up // newly persisted adjacencies on the next ingest. @@ -155,7 +166,7 @@ func (s *Store) StartNeighborEdgesBuilder(interval time.Duration) func() { // SELECT of (lowered) pubkey prefixes from nodes. Prefixes with // multiple candidates are skipped (matches the conservative // resolution rule in cmd/server/extractEdgesFromObs). -func (s *Store) buildAndPersistNeighborEdges() (int, error) { +func (s *Store) buildAndPersistNeighborEdges(trust *packetpath.TrustConfig) (int, error) { prefixIdx, err := buildPrefixIndex(s.db) if err != nil { return 0, fmt.Errorf("build prefix index: %w", err) @@ -202,29 +213,38 @@ func (s *Store) buildAndPersistNeighborEdges() (int, error) { if err := rows.Scan(&payloadType, &decodedJSON, &fromPubkey, &pathJSON, &observerID, &epochTs); err != nil { continue } + isAdvert := payloadType.Valid && payloadType.Int64 == int64(payloadADVERT) + isAnonReq := payloadType.Valid && payloadType.Int64 == int64(payloadAnonReq) + // #1777: ANON_REQ's ephemeralPubKey is as trustworthy an originator + // identity as ADVERT's pubKey — see payloadAnonReq doc comment. + hasFullOriginator := isAdvert || isAnonReq + fromNode := strings.ToLower(fromPubkey) if fromNode == "" { - fromNode = strings.ToLower(extractPubkeyFromAdvertJSON(decodedJSON)) + if isAdvert { + fromNode = strings.ToLower(extractPubkeyFromAdvertJSON(decodedJSON)) + } else if isAnonReq { + fromNode = strings.ToLower(extractPubkeyFromAnonReqJSON(decodedJSON)) + } } - isAdvert := payloadType.Valid && payloadType.Int64 == int64(payloadADVERT) ts := time.Unix(epochTs, 0).UTC().Format(time.RFC3339) observerPK := strings.ToLower(observerID) path := parsePathArray(pathJSON) if len(path) == 0 { - if isAdvert && fromNode != "" && fromNode != observerPK && observerPK != "" { + if hasFullOriginator && fromNode != "" && fromNode != observerPK && observerPK != "" { edges = append(edges, canonEdge(fromNode, observerPK, ts)) } continue } - if isAdvert && fromNode != "" { - if resolved, ok := resolvePrefix(prefixIdx, path[0]); ok && resolved != fromNode { + if hasFullOriginator && fromNode != "" { + if resolved, ok := resolvePrefix(prefixIdx, path[0], trust); ok && resolved != fromNode { edges = append(edges, canonEdge(fromNode, resolved, ts)) } } if observerPK != "" { last := path[len(path)-1] - if resolved, ok := resolvePrefix(prefixIdx, last); ok && resolved != observerPK { + if resolved, ok := resolvePrefix(prefixIdx, last, trust); ok && resolved != observerPK { edges = append(edges, canonEdge(observerPK, resolved, ts)) } } @@ -325,7 +345,19 @@ func buildPrefixIndex(db *sql.DB) (prefixIndex, error) { // candidate matches, otherwise (zero || multiple), it returns ok=false // (matches the conservative server-side resolver in // cmd/server/extractEdgesFromObs). -func resolvePrefix(idx prefixIndex, hop string) (string, bool) { +func resolvePrefix(idx prefixIndex, hop string, trust *packetpath.TrustConfig) (string, bool) { + // #1784: gate on prefix length before consulting the index. A hop + // hash short enough to fall below the operator's trust threshold is + // not mapping evidence even when it happens to resolve to exactly + // one candidate today — uniqueness is relative to the nodes we + // currently know about, so an unknown or newly joined repeater + // sharing that prefix silently turns it into a wrong edge. Gating + // here rather than at each call site means every consumer of the + // resolver (originator edge, observer edge, and any future hop + // pair) inherits the threshold automatically. + if !packetpath.MeetsPathTrust(len(hop)/2, trust) { + return "", false + } h := strings.ToLower(hop) candidates := idx[h] if len(candidates) != 1 { diff --git a/cmd/ingestor/neighbor_builder_delta_test.go b/cmd/ingestor/neighbor_builder_delta_test.go index 307fb978..4e530953 100644 --- a/cmd/ingestor/neighbor_builder_delta_test.go +++ b/cmd/ingestor/neighbor_builder_delta_test.go @@ -92,7 +92,7 @@ func TestNeighborEdgesBuilderDeltaScan(t *testing.T) { // goroutine harness). Full scan allowed because neighbor_edges // starts empty. for { - n, err := store.buildAndPersistNeighborEdges() + n, err := store.buildAndPersistNeighborEdges(trustAllPrefixes()) if err != nil { t.Fatalf("warm-up build: %v", err) } @@ -119,7 +119,7 @@ func TestNeighborEdgesBuilderDeltaScan(t *testing.T) { // Tick #2: NO new observations. Expect no-op + fast. noopStart := time.Now() - n2, err := store.buildAndPersistNeighborEdges() + n2, err := store.buildAndPersistNeighborEdges(trustAllPrefixes()) if err != nil { t.Fatalf("noop build: %v", err) } @@ -166,7 +166,7 @@ func TestNeighborEdgesBuilderDeltaScan(t *testing.T) { } deltaStart := time.Now() - n3, err := store.buildAndPersistNeighborEdges() + n3, err := store.buildAndPersistNeighborEdges(trustAllPrefixes()) if err != nil { t.Fatalf("delta build: %v", err) } diff --git a/cmd/ingestor/neighbor_builder_test.go b/cmd/ingestor/neighbor_builder_test.go index 65b63be7..706e6be0 100644 --- a/cmd/ingestor/neighbor_builder_test.go +++ b/cmd/ingestor/neighbor_builder_test.go @@ -3,6 +3,8 @@ package main import ( "path/filepath" "testing" + + "github.com/meshcore-analyzer/packetpath" ) // TestNeighborEdgesBuilderUpsertsFromObservations enforces issue @@ -66,7 +68,7 @@ func TestNeighborEdgesBuilderUpsertsFromObservations(t *testing.T) { t.Fatal(err) } - n, err := store.buildAndPersistNeighborEdges() + n, err := store.buildAndPersistNeighborEdges(trustAllPrefixes()) if err != nil { t.Fatalf("buildAndPersistNeighborEdges: %v", err) } @@ -85,3 +87,244 @@ func TestNeighborEdgesBuilderUpsertsFromObservations(t *testing.T) { // (test ends here) +// TestNeighborEdgesBuilderUpsertsFromAnonReqEphemeralPubKey verifies #1777: +// ANON_REQ transmissions (payload type 7) carry the sender's full +// ephemeral pubkey in decoded_json ("ephemeralPubKey"), not in the +// from_pubkey column (which is only populated for ADVERT at write time, +// see db.go's #1143 comment). The builder must fall back to parsing +// decoded_json for ANON_REQ, exactly as it already does for ADVERT. +func TestNeighborEdgesBuilderUpsertsFromAnonReqEphemeralPubKey(t *testing.T) { + dir := t.TempDir() + dbPath := filepath.Join(dir, "build.db") + + store, err := OpenStore(dbPath) + if err != nil { + t.Fatalf("OpenStore: %v", err) + } + defer store.Close() + + if _, err := store.db.Exec( + `INSERT INTO nodes (public_key, name) VALUES (?, ?), (?, ?)`, + "aaaaaaaaaa", "sender", + "bbbbbbbbbb", "first-hop", + ); err != nil { + t.Fatal(err) + } + + if _, err := store.db.Exec( + `INSERT INTO observers (id, name) VALUES (?, ?)`, + "obs-1", "observer-1", + ); err != nil { + t.Fatal(err) + } + var obsRowid int64 + if err := store.db.QueryRow(`SELECT rowid FROM observers WHERE id = ?`, "obs-1").Scan(&obsRowid); err != nil { + t.Fatal(err) + } + + // ANON_REQ transmission: from_pubkey left NULL (as real ingest does — + // only ADVERT populates it at write time), sender identity carried in + // decoded_json.ephemeralPubKey instead. + res, err := store.db.Exec( + `INSERT INTO transmissions (raw_hex, hash, first_seen, route_type, payload_type, payload_version, decoded_json) + VALUES (?, ?, ?, ?, ?, ?, ?)`, + "", "h2", "2026-01-01T00:00:00Z", 0, payloadAnonReq, 0, `{"ephemeralPubKey":"aaaaaaaaaa"}`, + ) + if err != nil { + t.Fatal(err) + } + txID, _ := res.LastInsertId() + + if _, err := store.db.Exec( + `INSERT INTO observations (transmission_id, observer_idx, path_json, timestamp) VALUES (?, ?, ?, ?)`, + txID, obsRowid, `["bb"]`, int64(1735689600), + ); err != nil { + t.Fatal(err) + } + + n, err := store.buildAndPersistNeighborEdges(trustAllPrefixes()) + if err != nil { + t.Fatalf("buildAndPersistNeighborEdges: %v", err) + } + if n == 0 { + t.Fatal("expected at least 1 edge upserted, got 0") + } + + var got int + if err := store.db.QueryRow(`SELECT COUNT(*) FROM neighbor_edges WHERE node_a = ? AND node_b = ?`, "aaaaaaaaaa", "bbbbbbbbbb").Scan(&got); err != nil { + t.Fatal(err) + } + if got != 1 { + t.Fatalf("expected the sender\u2194first-hop edge from ANON_REQ to be persisted (#1777); got %d rows", got) + } +} + +// TestNeighborEdgesBuilderExcludesOtherNonAdvertTypes verifies #1777's +// scope boundary: a plain REQ (payload type 2, not ADVERT or ANON_REQ) +// must NOT produce an originator↔path[0] edge, even if from_pubkey happens +// to be set — REQ's src is only a 1-byte truncated hash of the originator, +// not a full pubkey, and was explicitly rejected as an edge source in the +// #1777 discussion (collision odds ~1/256). +func TestNeighborEdgesBuilderExcludesOtherNonAdvertTypes(t *testing.T) { + dir := t.TempDir() + dbPath := filepath.Join(dir, "build.db") + + store, err := OpenStore(dbPath) + if err != nil { + t.Fatalf("OpenStore: %v", err) + } + defer store.Close() + + if _, err := store.db.Exec( + `INSERT INTO nodes (public_key, name) VALUES (?, ?), (?, ?)`, + "aaaaaaaaaa", "sender", + "bbbbbbbbbb", "first-hop", + ); err != nil { + t.Fatal(err) + } + + if _, err := store.db.Exec( + `INSERT INTO observers (id, name) VALUES (?, ?)`, + "obs-1", "observer-1", + ); err != nil { + t.Fatal(err) + } + var obsRowid int64 + if err := store.db.QueryRow(`SELECT rowid FROM observers WHERE id = ?`, "obs-1").Scan(&obsRowid); err != nil { + t.Fatal(err) + } + + const payloadREQ = 2 + res, err := store.db.Exec( + `INSERT INTO transmissions (raw_hex, hash, first_seen, route_type, payload_type, payload_version, decoded_json, from_pubkey) + VALUES (?, ?, ?, ?, ?, ?, ?, ?)`, + "", "h3", "2026-01-01T00:00:00Z", 0, payloadREQ, 0, "{}", "aaaaaaaaaa", + ) + if err != nil { + t.Fatal(err) + } + txID, _ := res.LastInsertId() + + if _, err := store.db.Exec( + `INSERT INTO observations (transmission_id, observer_idx, path_json, timestamp) VALUES (?, ?, ?, ?)`, + txID, obsRowid, `["bb"]`, int64(1735689600), + ); err != nil { + t.Fatal(err) + } + + if _, err := store.buildAndPersistNeighborEdges(trustAllPrefixes()); err != nil { + t.Fatalf("buildAndPersistNeighborEdges: %v", err) + } + + var got int + if err := store.db.QueryRow(`SELECT COUNT(*) FROM neighbor_edges WHERE node_a = ? AND node_b = ?`, "aaaaaaaaaa", "bbbbbbbbbb").Scan(&got); err != nil { + t.Fatal(err) + } + if got != 0 { + t.Fatalf("REQ should not produce an originator\u2194path[0] edge; got %d rows", got) + } +} + +// trustAllPrefixes returns the pre-#1784 threshold, where 1-byte hop +// hashes still count as mapping evidence. The builder tests above +// exercise edge *shape* using 1-byte fixtures; pinning them to the +// legacy threshold keeps their original intent intact, while the gate +// itself is covered by the two tests below. +func trustAllPrefixes() *packetpath.TrustConfig { + return &packetpath.TrustConfig{MinHashBytesForMapping: 1} +} + +// seedTrustFixture builds the minimal DB shape shared by the path-trust +// tests: two nodes, one observer, one ADVERT transmission, and one +// observation carrying hop as its single path element. +func seedTrustFixture(t *testing.T, store *Store, hop string) { + t.Helper() + if _, err := store.db.Exec( + `INSERT INTO nodes (public_key, name) VALUES (?, ?), (?, ?)`, + "aaaaaaaaaa", "from-node", + "bbbbbbbbbb", "first-hop", + ); err != nil { + t.Fatal(err) + } + if _, err := store.db.Exec(`INSERT INTO observers (id, name) VALUES (?, ?)`, "obs-1", "observer-1"); err != nil { + t.Fatal(err) + } + var obsRowid int64 + if err := store.db.QueryRow(`SELECT rowid FROM observers WHERE id = ?`, "obs-1").Scan(&obsRowid); err != nil { + t.Fatal(err) + } + res, err := store.db.Exec( + `INSERT INTO transmissions (raw_hex, hash, first_seen, route_type, payload_type, payload_version, decoded_json, from_pubkey) + VALUES (?, ?, ?, ?, ?, ?, ?, ?)`, + "", "h1", "2026-01-01T00:00:00Z", 0, payloadADVERT, 0, "{}", "aaaaaaaaaa", + ) + if err != nil { + t.Fatal(err) + } + txID, _ := res.LastInsertId() + if _, err := store.db.Exec( + `INSERT INTO observations (transmission_id, observer_idx, path_json, timestamp) VALUES (?, ?, ?, ?)`, + txID, obsRowid, `["`+hop+`"]`, int64(1735689600), + ); err != nil { + t.Fatal(err) + } +} + +// TestNeighborEdgesBuilderPathTrustExcludesOneByte pins #1784 at the +// builder: under the default threshold (2 bytes) a 1-byte hop hash +// produces no edge, even though it resolves to exactly one candidate in +// the nodes table. Uniqueness of a 1-byte prefix is a property of the +// nodes we happen to know about — on a mesh large enough to occupy all +// 256 values (SaarMesh: 1071 nodes, 13 of them uniquely resolvable by +// one byte) a later-joining repeater sharing that byte turns today's +// "unique" resolution into a wrong edge. +func TestNeighborEdgesBuilderPathTrustExcludesOneByte(t *testing.T) { + store, err := OpenStore(filepath.Join(t.TempDir(), "trust1.db")) + if err != nil { + t.Fatalf("OpenStore: %v", err) + } + defer store.Close() + seedTrustFixture(t, store, "bb") + + // nil == package default (MinHashBytesForMapping = 2). + if _, err := store.buildAndPersistNeighborEdges(nil); err != nil { + t.Fatalf("buildAndPersistNeighborEdges: %v", err) + } + + var got int + if err := store.db.QueryRow( + `SELECT COUNT(*) FROM neighbor_edges WHERE node_a = ? AND node_b = ?`, + "aaaaaaaaaa", "bbbbbbbbbb").Scan(&got); err != nil { + t.Fatal(err) + } + if got != 0 { + t.Fatalf("1-byte hop must not produce an edge under the default threshold, got %d", got) + } +} + +// TestNeighborEdgesBuilderPathTrustAllowsTwoByte is the positive half of +// the gate: the same fixture with a 2-byte hop still produces the edge, +// so the threshold narrows the evidence base rather than disabling the +// builder. +func TestNeighborEdgesBuilderPathTrustAllowsTwoByte(t *testing.T) { + store, err := OpenStore(filepath.Join(t.TempDir(), "trust2.db")) + if err != nil { + t.Fatalf("OpenStore: %v", err) + } + defer store.Close() + seedTrustFixture(t, store, "bbbb") + + if _, err := store.buildAndPersistNeighborEdges(nil); err != nil { + t.Fatalf("buildAndPersistNeighborEdges: %v", err) + } + + var got int + if err := store.db.QueryRow( + `SELECT COUNT(*) FROM neighbor_edges WHERE node_a = ? AND node_b = ?`, + "aaaaaaaaaa", "bbbbbbbbbb").Scan(&got); err != nil { + t.Fatal(err) + } + if got != 1 { + t.Fatalf("2-byte hop must still produce the edge, got %d", got) + } +} diff --git a/cmd/server/neighbor_graph.go b/cmd/server/neighbor_graph.go index 47138bf1..209c7e67 100644 --- a/cmd/server/neighbor_graph.go +++ b/cmd/server/neighbor_graph.go @@ -302,6 +302,14 @@ func BuildFromStoreWithOptions(store *PacketStore, opts BuildOptions) *NeighborG // Phase 1: Extract edges from every transmission + observation. for _, tx := range packets { isAdvert := tx.PayloadType != nil && *tx.PayloadType == PayloadADVERT + isAnonReq := tx.PayloadType != nil && *tx.PayloadType == PayloadANON_REQ + // #1777: ANON_REQ's ephemeralPubKey is a full Ed25519 pubkey — the + // same trust level as ADVERT's pubKey — so it can seed an + // originator↔path[0] edge exactly like ADVERT. Other non-ADVERT + // types (REQ/RESP/PATH/TXT) only carry a 1-byte truncated hash of + // the originator in src/dst, which is deliberately excluded here + // (would manufacture false edges at ~1/256 collision odds). + hasFullOriginator := isAdvert || isAnonReq fromNode := extractFromNode(tx) // Pre-compute lowered originator once per tx (not per observation). fromLower := "" @@ -315,7 +323,7 @@ func BuildFromStoreWithOptions(store *PacketStore, opts BuildOptions) *NeighborG if len(path) == 0 { // Zero-hop - if isAdvert && fromLower != "" { + if hasFullOriginator && fromLower != "" { if fromLower != observerPK { // self-edge guard g.upsertEdge(fromLower, observerPK, "", observerPK, obs.SNR, parseTimestamp(obs.Timestamp)) } @@ -323,8 +331,8 @@ func BuildFromStoreWithOptions(store *PacketStore, opts BuildOptions) *NeighborG continue } - // Edge 1: originator ↔ path[0] — ADVERTs only - if isAdvert && fromLower != "" { + // Edge 1: originator ↔ path[0] — ADVERT and ANON_REQ only (#1777) + if hasFullOriginator && fromLower != "" { firstHop := cachedToLower(lowerCache, path[0]) if fromLower != firstHop { // self-edge guard (shouldn't happen but spec says check) if packetpath.MeetsPathTrust(len(path[0])/2, opts.PathTrust) { @@ -359,15 +367,29 @@ func BuildFromStoreWithOptions(store *PacketStore, opts BuildOptions) *NeighborG } // extractFromNode pulls the originator pubkey from a StoreTx's DecodedJSON. -// ADVERTs use "pubKey", other packets may use "from_node" or "from". +// ADVERTs use "pubKey"; other packets may use "from_node" or "from". // Uses the cached ParsedDecoded() accessor to avoid repeated json.Unmarshal. func extractFromNode(tx *StoreTx) string { decoded := tx.ParsedDecoded() if decoded == nil { return "" } - // ADVERTs store the originator pubkey as "pubKey"; other packets may use - // "from_node" or "from". Check all three so we never miss the originator. + // ANON_REQ carries the originator's full Ed25519 pubkey as + // "ephemeralPubKey" (#1777) — the same trust level as ADVERT's "pubKey", + // unlike the 1-byte truncated src/dst hashes on REQ/RESP/PATH/TXT. + // Gated on the actual payload type (rather than just checking whether + // the JSON key happens to be present) so this stays correct even if a + // future decoder change reuses the "ephemeralPubKey" name for a + // different, non-originator field on some other payload type — the + // field name alone would no longer be a safe signal, but the payload + // type check still is. + if tx.PayloadType != nil && *tx.PayloadType == PayloadANON_REQ { + if v, ok := decoded["ephemeralPubKey"]; ok { + if s, ok := v.(string); ok && s != "" { + return s + } + } + } for _, field := range []string{"pubKey", "from_node", "from"} { if v, ok := decoded[field]; ok { if s, ok := v.(string); ok && s != "" { diff --git a/cmd/server/neighbor_graph_test.go b/cmd/server/neighbor_graph_test.go index 661df6a2..e9a02cc7 100644 --- a/cmd/server/neighbor_graph_test.go +++ b/cmd/server/neighbor_graph_test.go @@ -624,6 +624,83 @@ func TestBuildNeighborGraph_ADVERTOnlyConstraint(t *testing.T) { } } +// ngEphemeralPubKeyJSON creates decoded JSON using the real ANON_REQ format +// ("ephemeralPubKey" field) — #1777. +func ngEphemeralPubKeyJSON(pubkey string) string { + b, _ := json.Marshal(map[string]string{"ephemeralPubKey": pubkey}) + return string(b) +} + +// TestBuildNeighborGraph_AnonReqSingleHopPath verifies #1777: ANON_REQ +// (payload type 7) carries the sender's full ephemeral pubkey and should +// produce an originator↔path[0] edge exactly like ADVERT, in addition to +// the always-present observer↔path[last] edge. +func TestBuildNeighborGraph_AnonReqSingleHopPath(t *testing.T) { + nodes := []nodeInfo{ + {Role: "repeater", PublicKey: "aaaa1111", Name: "NodeX"}, + {Role: "repeater", PublicKey: "r1aabbcc", Name: "R1"}, + {Role: "repeater", PublicKey: "obs00001", Name: "Observer"}, + } + tx := ngMakeTx(1, 7, ngEphemeralPubKeyJSON("aaaa1111"), []*StoreObs{ + ngMakeObs("obs00001", `["r1aa"]`, nowStr, ngFloatPtr(-10)), + }) + store := ngTestStore(nodes, []*StoreTx{tx}) + g := BuildFromStore(store) + + edges := g.AllEdges() + if len(edges) != 2 { + t.Fatalf("expected 2 edges (originator↔path[0] + observer↔path[last]), got %d", len(edges)) + } + + found := false + for _, e := range edges { + if (e.NodeA == "aaaa1111" && e.NodeB == "r1aabbcc") || + (e.NodeA == "r1aabbcc" && e.NodeB == "aaaa1111") { + found = true + } + } + if !found { + t.Error("ANON_REQ should produce originator↔path[0] edge (#1777)") + } + + found = false + for _, e := range edges { + if (e.NodeA == "obs00001" && e.NodeB == "r1aabbcc") || + (e.NodeA == "r1aabbcc" && e.NodeB == "obs00001") { + found = true + } + } + if !found { + t.Error("missing observer↔path[last] edge (Observer↔R1)") + } +} + +// TestBuildNeighborGraph_ReqRespStillExcluded verifies #1777's scope +// boundary: only ADVERT and ANON_REQ get an originator↔path[0] edge. +// REQ (payload type 2, reusing the existing ADVERTOnlyConstraint fixture +// shape) must still be excluded — its "from"/"src" is a 1-byte truncated +// hash, not a full pubkey, and manufacturing an edge from it would carry +// ~1/256 collision odds (rejected in the #1777 discussion). +func TestBuildNeighborGraph_ReqRespStillExcluded(t *testing.T) { + nodes := []nodeInfo{ + {Role: "repeater", PublicKey: "aaaa1111", Name: "NodeX"}, + {Role: "repeater", PublicKey: "r1aabbcc", Name: "R1"}, + {Role: "repeater", PublicKey: "obs00001", Name: "Observer"}, + } + tx := ngMakeTx(1, 2, ngFromNodeJSON("aaaa1111"), []*StoreObs{ + ngMakeObs("obs00001", `["r1aa"]`, nowStr, ngFloatPtr(-10)), + }) + store := ngTestStore(nodes, []*StoreTx{tx}) + g := BuildFromStore(store) + + for _, e := range g.AllEdges() { + a, b := e.NodeA, e.NodeB + if (a == "aaaa1111" && b == "r1aabbcc") || (a == "r1aabbcc" && b == "aaaa1111") { + t.Error("REQ (non-ADVERT, non-ANON_REQ) should NOT produce originator↔path[0] edge") + } + } +} + // ngPubKeyJSON creates decoded JSON using the real ADVERT format ("pubKey" field). func ngPubKeyJSON(pubkey string) string { b, _ := json.Marshal(map[string]string{"pubKey": pubkey}) diff --git a/cmd/server/neighbor_persist.go b/cmd/server/neighbor_persist.go index 5204147b..7273ba38 100644 --- a/cmd/server/neighbor_persist.go +++ b/cmd/server/neighbor_persist.go @@ -227,75 +227,3 @@ func unmarshalResolvedPath(s string) []*string { } return result } - -// ─── Shared edge-extraction helper (used by ingestor + tests) ────────────────── - -// edgeCandidate represents an extracted edge. The ingestor uses the -// same logic when computing edges from observations. -type edgeCandidate struct { - A, B, Timestamp string -} - -// extractEdgesFromObs extracts neighbor edge candidates from a single -// observation. For ADVERTs: originator↔path[0] (if unambiguous). For -// ALL types: observer↔path[last] (if unambiguous). Also handles -// zero-hop ADVERTs (originator↔observer direct link). -// -// Kept in cmd/server because the in-memory graph builder -// (neighbor_graph.go) also calls it; it is pure compute and does not -// touch the DB. -func extractEdgesFromObs(obs *StoreObs, tx *StoreTx, pm *prefixMap) []edgeCandidate { - isAdvert := tx.PayloadType != nil && *tx.PayloadType == PayloadADVERT - fromNode := extractFromNode(tx) - path := parsePathJSON(obs.PathJSON) - observerPK := strings.ToLower(obs.ObserverID) - ts := obs.Timestamp - var edges []edgeCandidate - - if len(path) == 0 { - if isAdvert && fromNode != "" { - fromLower := strings.ToLower(fromNode) - if fromLower != observerPK { - a, b := fromLower, observerPK - if a > b { - a, b = b, a - } - edges = append(edges, edgeCandidate{a, b, ts}) - } - } - return edges - } - - if isAdvert && fromNode != "" && pm != nil { - firstHop := strings.ToLower(path[0]) - fromLower := strings.ToLower(fromNode) - candidates := pm.m[firstHop] - if len(candidates) == 1 { - resolved := strings.ToLower(candidates[0].PublicKey) - if resolved != fromLower { - a, b := fromLower, resolved - if a > b { - a, b = b, a - } - edges = append(edges, edgeCandidate{a, b, ts}) - } - } - } - - if pm != nil { - lastHop := strings.ToLower(path[len(path)-1]) - candidates := pm.m[lastHop] - if len(candidates) == 1 { - resolved := strings.ToLower(candidates[0].PublicKey) - if resolved != observerPK { - a, b := observerPK, resolved - if a > b { - a, b = b, a - } - edges = append(edges, edgeCandidate{a, b, ts}) - } - } - } - - return edges -}