From c554238485e24ce6b2cd3db200dd86c967414088 Mon Sep 17 00:00:00 2001 From: n30nex Date: Sat, 3 Oct 2026 23:58:00 -0400 Subject: [PATCH] fix(traces): validate trace frames and preserve ambiguous identities --- db/packets.go | 23 +++- db/queries/queries.sql | 21 ++- db/sqlc/mock/querier.go | 17 ++- db/sqlc/querier.go | 3 + db/sqlc/queries.sql.go | 109 +++++++++++++--- db/trace_quality_integration_test.go | 77 +++++++++++ db/traces.go | 179 ++++++++++++++++++-------- db/traces_integration_test.go | 13 +- db/traces_test.go | 13 +- docs/docs.go | 31 +++++ docs/swagger.json | 31 +++++ docs/swagger.yaml | 21 +++ internal/api/packets.go | 31 +++-- internal/api/traces.go | 36 +++--- internal/ingest/ingest.go | 1 + internal/ingest/ingest_test.go | 3 + internal/ingest/packet.go | 58 +++++++-- internal/ingest/trace_quality_test.go | 75 +++++++++++ internal/tracequality/quality.go | 96 ++++++++++++++ internal/tracequality/quality_test.go | 51 ++++++++ 20 files changed, 763 insertions(+), 126 deletions(-) create mode 100644 db/trace_quality_integration_test.go create mode 100644 internal/ingest/trace_quality_test.go create mode 100644 internal/tracequality/quality.go create mode 100644 internal/tracequality/quality_test.go diff --git a/db/packets.go b/db/packets.go index 620acce..a142ab6 100644 --- a/db/packets.go +++ b/db/packets.go @@ -15,6 +15,7 @@ import ( sqlc "github.com/MeshCore-Beacon/beacon-server/db/sqlc" "github.com/MeshCore-Beacon/beacon-server/internal/api" "github.com/MeshCore-Beacon/beacon-server/internal/ingest" + "github.com/MeshCore-Beacon/beacon-server/internal/tracequality" "github.com/google/uuid" "github.com/jackc/pgx/v5/pgtype" "github.com/meshcore-go/meshcore-go" @@ -406,7 +407,15 @@ func (s *Store) GetPacket(ctx context.Context, packetHash []byte) (*api.Packet, // compute it once rather than per observation. var traceRawHashes [][]byte if row.PayloadType == int16(meshcore.PayloadTypeTrace) { - if trace, err := meshcore.TraceFromBytes(row.RawPayload); err == nil { + quality := tracequality.Assess(row.RawPayload, int(row.PayloadVersion), int(row.RouteType), -1) + for _, obs := range obsRows { + q := tracequality.Assess(row.RawPayload, int(row.PayloadVersion), int(row.RouteType), int(obs.PathLengthByte)) + for _, reason := range q.Reasons { + quality.Add(reason) + } + } + p.TraceQuality = &quality + if trace, err := meshcore.TraceFromBytes(row.RawPayload); err == nil && quality.Status == "supported" { hashSize := int(trace.PathHashSize()) for i := 0; i+hashSize <= len(trace.PathHashes); i += hashSize { traceRawHashes = append(traceRawHashes, trace.PathHashes[i:i+hashSize]) @@ -442,8 +451,11 @@ func (s *Store) GetPacket(ctx context.Context, packetHash []byte) (*api.Packet, obs.PropagationTimeMs = &prop resolvedPath := []api.ResolvedHop{} if row.PayloadType == int16(meshcore.PayloadTypeTrace) { + rawPath := hex.EncodeToString(v.PathBytes) + obs.WirePathBytes = &rawPath + obs.TraceQuality = p.TraceQuality if len(traceRawHashes) > 0 { - resolved, err := s.ResolvePathHashes(ctx, v.Iata, traceRawHashes) + resolved, err := s.ResolveTracePathHashes(ctx, v.Iata, traceRawHashes) if err != nil { slog.Error(fmt.Sprintf("store: path resolution failed for observation %d", v.ID), "component", "db", "error", err) } else { @@ -491,7 +503,7 @@ func (s *Store) GetPacket(ctx context.Context, packetHash []byte) (*api.Packet, } p.Observations = append(p.Observations, obs) } - if row.PayloadType == 9 && len(obsRows) > 0 { + if row.PayloadType == 9 && len(obsRows) > 0 && p.TraceQuality.Status == "supported" { iatas := make([]string, 0, len(obsRows)) seen := make(map[string]struct{}) for _, v := range obsRows { @@ -503,6 +515,11 @@ func (s *Store) GetPacket(ctx context.Context, packetHash []byte) (*api.Packet, var parsed tracePayload if err := json.Unmarshal(row.ParsedPayload, &parsed); err == nil { p.ResolvedRoute = s.resolveTraceRoute(ctx, &parsed, iatas) + for _, hop := range p.ResolvedRoute { + if hop.Confidence == "ambiguous" { + p.TraceQuality.Add("ambiguous_prefix") + } + } } } return p, nil diff --git a/db/queries/queries.sql b/db/queries/queries.sql index 2a2c470..9f513b1 100644 --- a/db/queries/queries.sql +++ b/db/queries/queries.sql @@ -455,6 +455,8 @@ WHERE p.packet_hash = $1; -- Return distinct observation IATAs in first-heard order for path resolution, -- without fetching full observations separately for every trace packet. SELECT encode(p.packet_hash, 'hex') AS packet_hash_hex, + p.raw_payload, p.payload_version, + EXISTS(SELECT 1 FROM packet_observations po WHERE po.packet_hash=p.packet_hash AND po.path_length_byte>=64) AS invalid_snr_path, p.route_type, p.first_heard_at, p.last_heard_at, @@ -1422,7 +1424,11 @@ SELECT WHERE ti.trace_tag = t.trace_tag AND (COALESCE(cardinality($1::bpchar[]), 0) = 0 OR ti.iata = ANY($1::bpchar[]))) AS iata_count, COALESCE(t.trace_type, '')::text AS trace_type, - t.best_payload + t.best_payload, + ARRAY(SELECT ti.iata FROM trace_iatas ti WHERE ti.trace_tag=t.trace_tag)::bpchar[] AS iatas, + EXISTS(SELECT 1 FROM packets p WHERE p.trace_tag=t.trace_tag AND p.payload_version<>0) AS unsupported_version, + EXISTS(SELECT 1 FROM packets p WHERE p.trace_tag=t.trace_tag AND p.route_type NOT IN (2,3)) AS non_direct, + EXISTS(SELECT 1 FROM packets p JOIN packet_observations po ON po.packet_hash=p.packet_hash WHERE p.trace_tag=t.trace_tag AND po.path_length_byte>=64) AS invalid_snr_path FROM trace_tags t WHERE (COALESCE(cardinality($1::bpchar[]), 0) = 0 OR EXISTS ( SELECT 1 FROM trace_iatas ti WHERE ti.trace_tag = t.trace_tag AND ti.iata = ANY($1::bpchar[]))) @@ -1834,3 +1840,16 @@ ON CONFLICT (country) DO UPDATE SET -- name: PruneZoneBoundaries :many -- Drops imports for IATAs no longer configured, so their manual border returns. DELETE FROM meshmapper_zone_boundaries WHERE NOT (iata = ANY(@keep::text[])) RETURNING iata; + +-- name: ResolveTraceHashes :many +-- TRACE can terminate at companions. Preserve the full prefix (including eight +-- bytes), and return every candidate across heard regions instead of picking one. +SELECT h.hash::bytea AS hash, si.iata, n.id AS node_id, n.name, n.latitude, n.longitude, n.public_key +FROM unnest(@hashes::bytea[]) AS h(hash) +JOIN node_short_ids si ON si.iata = ANY(@iatas::bpchar[]) AND + ((octet_length(h.hash)=1 AND si.prefix_1=h.hash) OR + (octet_length(h.hash)=2 AND si.prefix_2=h.hash) OR + (octet_length(h.hash) IN (4,8) AND si.prefix_4=substring(h.hash FROM 1 FOR 4))) +JOIN nodes n ON n.id=si.node_id +WHERE substring(n.public_key FROM 1 FOR octet_length(h.hash))=h.hash +ORDER BY si.iata, h.hash, n.id; diff --git a/db/sqlc/mock/querier.go b/db/sqlc/mock/querier.go index b34664f..6813cc5 100644 --- a/db/sqlc/mock/querier.go +++ b/db/sqlc/mock/querier.go @@ -3,7 +3,7 @@ // // Generated by this command: // -// mockgen -source db/sqlc/querier.go -destination db/sqlc/mock/querier.go -package mockdb +// mockgen -source=db/sqlc/querier.go -destination=db/sqlc/mock/querier.go -package=mockdb // // Package mockdb is a generated GoMock package. @@ -1896,6 +1896,21 @@ func (mr *MockQuerierMockRecorder) ResolvePathHashesP4(ctx, arg any) *gomock.Cal return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ResolvePathHashesP4", reflect.TypeOf((*MockQuerier)(nil).ResolvePathHashesP4), ctx, arg) } +// ResolveTraceHashes mocks base method. +func (m *MockQuerier) ResolveTraceHashes(ctx context.Context, arg db.ResolveTraceHashesParams) ([]db.ResolveTraceHashesRow, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "ResolveTraceHashes", ctx, arg) + ret0, _ := ret[0].([]db.ResolveTraceHashesRow) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// ResolveTraceHashes indicates an expected call of ResolveTraceHashes. +func (mr *MockQuerierMockRecorder) ResolveTraceHashes(ctx, arg any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ResolveTraceHashes", reflect.TypeOf((*MockQuerier)(nil).ResolveTraceHashes), ctx, arg) +} + // RollAdvertHearings mocks base method. func (m *MockQuerier) RollAdvertHearings(ctx context.Context, hour pgtype.Timestamptz) error { m.ctrl.T.Helper() diff --git a/db/sqlc/querier.go b/db/sqlc/querier.go index f3c54d3..5912fa4 100644 --- a/db/sqlc/querier.go +++ b/db/sqlc/querier.go @@ -314,6 +314,9 @@ 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) + // TRACE can terminate at companions. Preserve the full prefix (including eight + // bytes), and return every candidate across heard regions instead of picking one. + ResolveTraceHashes(ctx context.Context, arg ResolveTraceHashesParams) ([]ResolveTraceHashesRow, error) RollAdvertHearings(ctx context.Context, hour pgtype.Timestamptz) error // Each distinct ADVERT packet counts once, under the exact set of IATAs that heard it this hour. RollAdvertSets(ctx context.Context, hour pgtype.Timestamptz) error diff --git a/db/sqlc/queries.sql.go b/db/sqlc/queries.sql.go index 46b13b5..92a6a17 100644 --- a/db/sqlc/queries.sql.go +++ b/db/sqlc/queries.sql.go @@ -1405,6 +1405,8 @@ func (q *Queries) GetPacketByHash(ctx context.Context, packetHash []byte) (GetPa const getPacketsByTraceTag = `-- name: GetPacketsByTraceTag :many SELECT encode(p.packet_hash, 'hex') AS packet_hash_hex, + p.raw_payload, p.payload_version, + EXISTS(SELECT 1 FROM packet_observations po WHERE po.packet_hash=p.packet_hash AND po.path_length_byte>=64) AS invalid_snr_path, p.route_type, p.first_heard_at, p.last_heard_at, @@ -1425,14 +1427,17 @@ ORDER BY p.first_heard_at ASC ` type GetPacketsByTraceTagRow struct { - PacketHashHex string `json:"packet_hash_hex"` - RouteType int16 `json:"route_type"` - FirstHeardAt pgtype.Timestamptz `json:"first_heard_at"` - LastHeardAt pgtype.Timestamptz `json:"last_heard_at"` - ParsedPayload []byte `json:"parsed_payload"` - ScopeID *int32 `json:"scope_id"` - ScopeName *string `json:"scope_name"` - Iatas []string `json:"iatas"` + PacketHashHex string `json:"packet_hash_hex"` + RawPayload []byte `json:"raw_payload"` + PayloadVersion int16 `json:"payload_version"` + InvalidSnrPath bool `json:"invalid_snr_path"` + RouteType int16 `json:"route_type"` + FirstHeardAt pgtype.Timestamptz `json:"first_heard_at"` + LastHeardAt pgtype.Timestamptz `json:"last_heard_at"` + ParsedPayload []byte `json:"parsed_payload"` + ScopeID *int32 `json:"scope_id"` + ScopeName *string `json:"scope_name"` + Iatas []string `json:"iatas"` } // Return distinct observation IATAs in first-heard order for path resolution, @@ -1448,6 +1453,9 @@ func (q *Queries) GetPacketsByTraceTag(ctx context.Context, decode string) ([]Ge var i GetPacketsByTraceTagRow if err := rows.Scan( &i.PacketHashHex, + &i.RawPayload, + &i.PayloadVersion, + &i.InvalidSnrPath, &i.RouteType, &i.FirstHeardAt, &i.LastHeardAt, @@ -4137,7 +4145,11 @@ SELECT WHERE ti.trace_tag = t.trace_tag AND (COALESCE(cardinality($1::bpchar[]), 0) = 0 OR ti.iata = ANY($1::bpchar[]))) AS iata_count, COALESCE(t.trace_type, '')::text AS trace_type, - t.best_payload + t.best_payload, + ARRAY(SELECT ti.iata FROM trace_iatas ti WHERE ti.trace_tag=t.trace_tag)::bpchar[] AS iatas, + EXISTS(SELECT 1 FROM packets p WHERE p.trace_tag=t.trace_tag AND p.payload_version<>0) AS unsupported_version, + EXISTS(SELECT 1 FROM packets p WHERE p.trace_tag=t.trace_tag AND p.route_type NOT IN (2,3)) AS non_direct, + EXISTS(SELECT 1 FROM packets p JOIN packet_observations po ON po.packet_hash=p.packet_hash WHERE p.trace_tag=t.trace_tag AND po.path_length_byte>=64) AS invalid_snr_path FROM trace_tags t WHERE (COALESCE(cardinality($1::bpchar[]), 0) = 0 OR EXISTS ( SELECT 1 FROM trace_iatas ti WHERE ti.trace_tag = t.trace_tag AND ti.iata = ANY($1::bpchar[]))) @@ -4165,13 +4177,17 @@ type ListTraceTagsParams struct { } type ListTraceTagsRow struct { - TraceTag string `json:"trace_tag"` - FirstHeardAt pgtype.Timestamptz `json:"first_heard_at"` - LastHeardAt pgtype.Timestamptz `json:"last_heard_at"` - PacketCount int64 `json:"packet_count"` - IataCount int64 `json:"iata_count"` - TraceType string `json:"trace_type"` - BestPayload []byte `json:"best_payload"` + TraceTag string `json:"trace_tag"` + FirstHeardAt pgtype.Timestamptz `json:"first_heard_at"` + LastHeardAt pgtype.Timestamptz `json:"last_heard_at"` + PacketCount int64 `json:"packet_count"` + IataCount int64 `json:"iata_count"` + TraceType string `json:"trace_type"` + BestPayload []byte `json:"best_payload"` + Iatas []string `json:"iatas"` + UnsupportedVersion bool `json:"unsupported_version"` + NonDirect bool `json:"non_direct"` + InvalidSnrPath bool `json:"invalid_snr_path"` } // ============================================================ @@ -4204,6 +4220,10 @@ func (q *Queries) ListTraceTags(ctx context.Context, arg ListTraceTagsParams) ([ &i.IataCount, &i.TraceType, &i.BestPayload, + &i.Iatas, + &i.UnsupportedVersion, + &i.NonDirect, + &i.InvalidSnrPath, ); err != nil { return nil, err } @@ -4856,6 +4876,63 @@ func (q *Queries) ResolvePathHashesP4(ctx context.Context, arg ResolvePathHashes return items, nil } +const resolveTraceHashes = `-- name: ResolveTraceHashes :many +SELECT h.hash::bytea AS hash, si.iata, n.id AS node_id, n.name, n.latitude, n.longitude, n.public_key +FROM unnest($1::bytea[]) AS h(hash) +JOIN node_short_ids si ON si.iata = ANY($2::bpchar[]) AND + ((octet_length(h.hash)=1 AND si.prefix_1=h.hash) OR + (octet_length(h.hash)=2 AND si.prefix_2=h.hash) OR + (octet_length(h.hash) IN (4,8) AND si.prefix_4=substring(h.hash FROM 1 FOR 4))) +JOIN nodes n ON n.id=si.node_id +WHERE substring(n.public_key FROM 1 FOR octet_length(h.hash))=h.hash +ORDER BY si.iata, h.hash, n.id +` + +type ResolveTraceHashesParams struct { + Hashes [][]byte `json:"hashes"` + Iatas []string `json:"iatas"` +} + +type ResolveTraceHashesRow struct { + Hash []byte `json:"hash"` + Iata string `json:"iata"` + NodeID uuid.UUID `json:"node_id"` + Name *string `json:"name"` + Latitude *float64 `json:"latitude"` + Longitude *float64 `json:"longitude"` + PublicKey []byte `json:"public_key"` +} + +// TRACE can terminate at companions. Preserve the full prefix (including eight +// bytes), and return every candidate across heard regions instead of picking one. +func (q *Queries) ResolveTraceHashes(ctx context.Context, arg ResolveTraceHashesParams) ([]ResolveTraceHashesRow, error) { + rows, err := q.db.Query(ctx, resolveTraceHashes, arg.Hashes, arg.Iatas) + if err != nil { + return nil, err + } + defer rows.Close() + items := []ResolveTraceHashesRow{} + for rows.Next() { + var i ResolveTraceHashesRow + if err := rows.Scan( + &i.Hash, + &i.Iata, + &i.NodeID, + &i.Name, + &i.Latitude, + &i.Longitude, + &i.PublicKey, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + const saveChannelCatalogue = `-- name: SaveChannelCatalogue :exec INSERT INTO meshmapper_channel_catalogues (iata, url, payload, etag, checked_at, attempted_at, next_attempt, last_error) VALUES ($1, $2, $3::jsonb, $4::text, diff --git a/db/trace_quality_integration_test.go b/db/trace_quality_integration_test.go new file mode 100644 index 0000000..d623ab6 --- /dev/null +++ b/db/trace_quality_integration_test.go @@ -0,0 +1,77 @@ +package db + +import ( + "context" + "encoding/hex" + "os" + "testing" + + sqlc "github.com/MeshCore-Beacon/beacon-server/db/sqlc" + "github.com/jackc/pgx/v5" +) + +func TestTraceQualityPostgres(t *testing.T) { + dsn := os.Getenv("BEACON_TEST_POSTGRES_DSN") + if dsn == "" { + t.Skip("set BEACON_TEST_POSTGRES_DSN") + } + ctx := context.Background() + conn, err := pgx.Connect(ctx, dsn) + if err != nil { + t.Fatal(err) + } + defer conn.Close(ctx) + tx, err := conn.Begin(ctx) + if err != nil { + t.Fatal(err) + } + defer tx.Rollback(ctx) + for _, name := range []string{"nodes", "node_short_ids"} { + if _, err := tx.Exec(ctx, "CREATE TEMP TABLE "+name+" (LIKE public."+name+" INCLUDING ALL) ON COMMIT DROP"); err != nil { + t.Fatal(err) + } + } + _, err = tx.Exec(ctx, ` + INSERT INTO nodes (id,public_key,node_type,name) VALUES + ('00000000-0000-0000-0000-000000000001',decode('aabbccdd11223344'||repeat('00',24),'hex'),1,'Companion'), + ('00000000-0000-0000-0000-000000000002',decode('aabbccdd55667788'||repeat('00',24),'hex'),2,'Repeater'); + INSERT INTO node_short_ids (node_id,iata,prefix_4) VALUES + ('00000000-0000-0000-0000-000000000001','YKF','\xaabbccdd'), + ('00000000-0000-0000-0000-000000000002','YKF','\xaabbccdd'), + ('00000000-0000-0000-0000-000000000001','YHM','\xaabbccdd');`) + if err != nil { + t.Fatal(err) + } + store := &Store{q: sqlc.New(tx)} + for _, width := range []int{1, 2, 4, 8} { + full := "aabbccdd11223344" + h := full[:width*2] + b, _ := hex.DecodeString(h) + candidates, err := store.traceCandidates(ctx, []string{"YKF", "YHM"}, [][]byte{b}) + if err != nil { + t.Fatal(err) + } + flags := byte(0) + switch width { + case 2: + flags = 1 + case 4: + flags = 2 + case 8: + flags = 3 + } + route := traceRoute(&tracePayload{Flags: flags, PathHashes: []string{h}}, []string{"YKF", "YHM"}, candidates) + count := 2 + confidence := "ambiguous" + if width == 8 { + count = 1 + confidence = "high" + } + if len(route) != 1 || len(route[0].Nodes) != count || route[0].Confidence != confidence { + t.Fatalf("width %d: %+v", width, route) + } + if route[0].Nodes[0].PublicKey[:16] != "aabbccdd11223344" { + t.Fatal("companion excluded or eight-byte hash truncated") + } + } +} diff --git a/db/traces.go b/db/traces.go index 93ed18c..cc933b1 100644 --- a/db/traces.go +++ b/db/traces.go @@ -8,6 +8,8 @@ import ( "encoding/hex" "encoding/json" "fmt" + "github.com/MeshCore-Beacon/beacon-server/internal/tracequality" + "github.com/google/uuid" "time" sqlc "github.com/MeshCore-Beacon/beacon-server/db/sqlc" @@ -17,9 +19,11 @@ import ( ) type tracePayload struct { - PathHashes []string `json:"pathHashes"` - Flags byte `json:"flags"` - SNRValues []float32 `json:"snrValues"` + Raw string `json:"raw"` + Quality tracequality.Quality `json:"quality"` + PathHashes []string `json:"pathHashes"` + Flags byte `json:"flags"` + SNRValues []float32 `json:"snrValues"` } func (s *Store) RecordTrace(ctx context.Context, h ingest.TraceHearing) error { @@ -71,12 +75,41 @@ func (s *Store) ListTraceTags(ctx context.Context, iatas []string, scope, traceT return nil, err } items := make([]api.TraceTagSummary, 0, len(rows)) + payloads := make([]tracePayload, 0, len(rows)) + allIATAs := map[string]bool{} + allHashes := map[string][]byte{} for _, r := range rows { var best tracePayload if len(r.BestPayload) > 0 { _ = json.Unmarshal(r.BestPayload, &best) } + quality := tracequality.Stored(best.Raw, best.Flags, best.PathHashes, best.SNRValues) + for _, reason := range best.Quality.Reasons { + if reason != "ambiguous_prefix" { + quality.Add(reason) + } + } + if r.UnsupportedVersion { + quality.Add("unsupported_version") + } + if r.InvalidSnrPath { + quality.Add("invalid_snr_path") + } + if r.NonDirect { + quality.Add("non_direct_trace") + } + payloads = append(payloads, best) + if quality.Status == "supported" { + for _, iata := range r.Iatas { + allIATAs[iata] = true + } + for _, h := range best.PathHashes { + b, _ := hex.DecodeString(h) + allHashes[h] = b + } + } items = append(items, api.TraceTagSummary{ + Quality: quality, TraceTag: r.TraceTag, FirstHeardAt: r.FirstHeardAt.Time.UnixMilli(), LastHeardAt: r.LastHeardAt.Time.UnixMilli(), @@ -87,6 +120,30 @@ func (s *Store) ListTraceTags(ctx context.Context, iatas []string, scope, traceT SNRValues: best.SNRValues, }) } + // One bounded candidate query for the entire page, including all hash widths. + regions := make([]string, 0, len(allIATAs)) + for iata := range allIATAs { + regions = append(regions, iata) + } + hashes := make([][]byte, 0, len(allHashes)) + for _, h := range allHashes { + hashes = append(hashes, h) + } + candidates, err := s.traceCandidates(ctx, regions, hashes) + if err != nil { + return nil, err + } + for i := range items { + if items[i].Quality.Status != "supported" { + continue + } + for _, hop := range traceRoute(&payloads[i], rows[i].Iatas, candidates) { + if hop.Confidence == "ambiguous" { + items[i].Quality.Add("ambiguous_prefix") + break + } + } + } return items, nil } @@ -125,76 +182,94 @@ func (s *Store) GetTraceByTag(ctx context.Context, tag string) (*api.TraceDetail } packet.RawPath = rawPath } - if len(r.Iatas) > 0 { + packet.Quality = tracequality.Assess(r.RawPayload, int(r.PayloadVersion), int(r.RouteType), len(parsed.SNRValues)) + if r.InvalidSnrPath { + packet.Quality.Add("invalid_snr_path") + } + stored := tracequality.Stored(parsed.Raw, parsed.Flags, parsed.PathHashes, parsed.SNRValues) + for _, reason := range stored.Reasons { + packet.Quality.Add(reason) + } + if len(r.Iatas) > 0 && packet.Quality.Status == "supported" { packet.ResolvedRoute = s.resolveTraceRoute(ctx, &parsed, r.Iatas) } + for _, hop := range packet.ResolvedRoute { + if hop.Confidence == "ambiguous" { + packet.Quality.Add("ambiguous_prefix") + } + } detail.Packets = append(detail.Packets, packet) } return detail, nil } +// Candidates remain separated by hearing region until a trace is assembled. +// Never replace an ambiguous region with a convenient singleton from another. +type traceCandidateMap map[string]map[string][]api.ResolvedPathEntry + +func (s *Store) traceCandidates(ctx context.Context, iatas []string, hashes [][]byte) (traceCandidateMap, error) { + out := traceCandidateMap{} + if len(iatas) == 0 || len(hashes) == 0 { + return out, nil + } + rows, err := s.q.ResolveTraceHashes(ctx, sqlc.ResolveTraceHashesParams{Iatas: iatas, Hashes: hashes}) + if err != nil { + return nil, err + } + for _, r := range rows { + if out[r.Iata] == nil { + out[r.Iata] = map[string][]api.ResolvedPathEntry{} + } + h := hex.EncodeToString(r.Hash) + out[r.Iata][h] = append(out[r.Iata][h], api.ResolvedPathEntry{NodeID: r.NodeID, Name: r.Name, Latitude: r.Latitude, Longitude: r.Longitude, PublicKey: r.PublicKey}) + } + return out, nil +} + +func (s *Store) ResolveTracePathHashes(ctx context.Context, iata string, hashes [][]byte) (map[string][]api.ResolvedPathEntry, error) { + candidates, err := s.traceCandidates(ctx, []string{iata}, hashes) + return candidates[iata], err +} + func (s *Store) resolveTraceRoute(ctx context.Context, payload *tracePayload, iatas []string) []api.ResolvedHop { - if payload == nil || len(payload.PathHashes) == 0 { + if payload == nil || tracequality.Stored(payload.Raw, payload.Flags, payload.PathHashes, payload.SNRValues).Status != "supported" { return nil } - hashSize := int(1 << (payload.Flags & 0x03)) hashes := make([][]byte, 0, len(payload.PathHashes)) for _, h := range payload.PathHashes { - b, err := hex.DecodeString(h) - if err == nil { - hashes = append(hashes, b) - } + b, _ := hex.DecodeString(h) + hashes = append(hashes, b) } - confidenceRank := map[string]int{"none": 0, "ambiguous": 1, "high": 2} - type hopResult struct { - confidence string - entries []api.ResolvedPathEntry + candidates, err := s.traceCandidates(ctx, iatas, hashes) + if err != nil { + return nil } - merged := make([]hopResult, len(hashes)) - for i := range merged { - merged[i] = hopResult{confidence: "none"} - } - for _, iata := range iatas { - resolved, err := s.ResolvePathHashes(ctx, iata, hashes) - if err != nil { - continue - } - for i, hash := range hashes { - key := hex.EncodeToString(hash[:hashSize]) - entries := resolved[key] - var confidence string - switch len(entries) { - case 0: - confidence = "none" - case 1: - confidence = "high" - default: - confidence = "ambiguous" - } - if confidenceRank[confidence] > confidenceRank[merged[i].confidence] { - merged[i] = hopResult{confidence: confidence, entries: entries} + return traceRoute(payload, iatas, candidates) +} + +func traceRoute(payload *tracePayload, iatas []string, candidates traceCandidateMap) []api.ResolvedHop { + route := make([]api.ResolvedHop, 0, len(payload.PathHashes)) + for i, h := range payload.PathHashes { + hop := api.ResolvedHop{Confidence: "none", Nodes: []api.ResolvedNode{}} + seen := map[uuid.UUID]bool{} + for _, iata := range iatas { + for _, e := range candidates[iata][h] { + if seen[e.NodeID] { + continue + } + seen[e.NodeID] = true + hop.Nodes = append(hop.Nodes, api.ResolvedNode{ID: e.NodeID, Name: e.Name, Latitude: e.Latitude, Longitude: e.Longitude, PublicKey: hex.EncodeToString(e.PublicKey)}) } } - } - route := make([]api.ResolvedHop, 0, len(hashes)) - for i, hr := range merged { - hop := api.ResolvedHop{ - Confidence: hr.confidence, - Nodes: make([]api.ResolvedNode, 0, len(hr.entries)), + if len(hop.Nodes) == 1 { + hop.Confidence = "high" + } else if len(hop.Nodes) > 1 { + hop.Confidence = "ambiguous" } if i < len(payload.SNRValues) { snr := payload.SNRValues[i] hop.SNR = &snr } - for _, e := range hr.entries { - hop.Nodes = append(hop.Nodes, api.ResolvedNode{ - ID: e.NodeID, - Name: e.Name, - Latitude: e.Latitude, - Longitude: e.Longitude, - PublicKey: hex.EncodeToString(e.PublicKey), - }) - } route = append(route, hop) } return route diff --git a/db/traces_integration_test.go b/db/traces_integration_test.go index 5130544..1d671af 100644 --- a/db/traces_integration_test.go +++ b/db/traces_integration_test.go @@ -59,7 +59,7 @@ INSERT INTO node_short_ids (node_id,iata,prefix_4) VALUES ('00000000-0000-0000-0000-000000000001','YVR','\xaa000000'), ('00000000-0000-0000-0000-000000000002','YYJ','\xaa010000'); INSERT INTO packets (packet_hash,payload_type,payload_version,route_type,raw_payload,raw_header,trace_tag,parsed_payload,first_heard_at,last_heard_at) -SELECT decode(lpad(to_hex(i),64,'0'),'hex'),9,0,1,'\x00','\x00','\x01020304', +SELECT decode(lpad(to_hex(i),64,'0'),'hex'),9,0,2,'\x010203040000000000aa','\x26','\x01020304', CASE WHEN i=100 THEN '{}'::jsonb ELSE '{"flags":0,"pathHashes":["aa"],"snrValues":[12.5]}'::jsonb END, '2026-01-01 00:00:00+00'::timestamptz+i*interval '1 second', '2026-01-01 00:00:00+00'::timestamptz+i*interval '1 second'+interval '1 second' FROM generate_series(1,100) i; @@ -89,7 +89,7 @@ ANALYZE packets; ANALYZE packet_observations; ANALYZE nodes; ANALYZE node_short_ anchor := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC) for i, p := range detail.Packets { number := i + 1 - if p.PacketHash != fmt.Sprintf("%064x", number) || p.RouteType != 1 || p.FirstHeardAt != anchor.Add(time.Duration(number)*time.Second).UnixMilli() || p.LastHeardAt != p.FirstHeardAt+1000 { + if p.PacketHash != fmt.Sprintf("%064x", number) || p.RouteType != 2 || p.FirstHeardAt != anchor.Add(time.Duration(number)*time.Second).UnixMilli() || p.LastHeardAt != p.FirstHeardAt+1000 { t.Fatalf("wrong packet fields or order at %d", number) } if number == 100 { @@ -109,7 +109,10 @@ ANALYZE packets; ANALYZE packet_observations; ANALYZE nodes; ANALYZE node_short_ } want := "YVR relay" if number == 1 { - want = "YYJ relay" + if len(p.ResolvedRoute) != 1 || p.ResolvedRoute[0].Confidence != "ambiguous" || len(p.ResolvedRoute[0].Nodes) != 2 || p.Quality.Status != "ambiguous" { + t.Fatal("conflicting regions must retain both candidates") + } + continue } if len(p.ResolvedRoute) != 1 || p.ResolvedRoute[0].Confidence != "high" || len(p.ResolvedRoute[0].Nodes) != 1 || p.ResolvedRoute[0].Nodes[0].Name == nil || *p.ResolvedRoute[0].Nodes[0].Name != want { t.Fatalf("wrong region precedence/resolution at %d", number) @@ -122,8 +125,8 @@ ANALYZE packets; ANALYZE packet_observations; ANALYZE nodes; ANALYZE node_short_ if counter.observations != 0 { t.Errorf("extra observation queries: %d; want 0", counter.observations) } - if counter.queries != 100 { - t.Errorf("got %d queries; want one packet query and 99 unchanged path-resolution queries", counter.queries) + if counter.queries != 99 { + t.Errorf("got %d queries; want one packet query and 98 batched path-resolution queries", counter.queries) } counter.queries, counter.observations = 0, 0 missing, err := store.GetTraceByTag(ctx, "00000000") diff --git a/db/traces_test.go b/db/traces_test.go index 0e3d4b4..ef0a441 100644 --- a/db/traces_test.go +++ b/db/traces_test.go @@ -99,7 +99,7 @@ func TestGetTraceByTag_WithPacket(t *testing.T) { firstHeard := pgtype.Timestamptz{Time: time.UnixMilli(1700000000000), Valid: true} lastHeard := pgtype.Timestamptz{Time: time.UnixMilli(1700000001000), Valid: true} // aabbccdd is valid hex for the packet hash - parsedPayload := []byte(`{"pathHashes":["aabb"],"snrValues":[15.0],"flags":0}`) + parsedPayload := []byte(`{"pathHashes":["aabb"],"snrValues":[15.0],"flags":1}`) scopeName := "default" mock.EXPECT(). @@ -107,7 +107,8 @@ func TestGetTraceByTag_WithPacket(t *testing.T) { Return([]sqlc.GetPacketsByTraceTagRow{ { PacketHashHex: "aabbccdd", - RouteType: 1, + RouteType: 2, + RawPayload: []byte{0, 0, 0, 0, 0, 0, 0, 0, 1, 0xaa, 0xbb}, ScopeName: &scopeName, FirstHeardAt: firstHeard, LastHeardAt: lastHeard, @@ -117,11 +118,11 @@ func TestGetTraceByTag_WithPacket(t *testing.T) { }, nil) mock.EXPECT(). - ResolvePathHashesP2(gomock.Any(), sqlc.ResolvePathHashesP2Params{ - Iata: "YVR", - Column2: [][]byte{{0xaa, 0xbb}}, + ResolveTraceHashes(gomock.Any(), sqlc.ResolveTraceHashesParams{ + Iatas: []string{"YVR"}, + Hashes: [][]byte{{0xaa, 0xbb}}, }). - Return([]sqlc.ResolvePathHashesP2Row{}, nil) + Return([]sqlc.ResolveTraceHashesRow{}, nil) store := &Store{q: mock} detail, err := store.GetTraceByTag(context.Background(), "trace-001") diff --git a/docs/docs.go b/docs/docs.go index 93eaf49..9665b7c 100644 --- a/docs/docs.go +++ b/docs/docs.go @@ -4583,6 +4583,9 @@ const docTemplate = `{ "description": "matched transport scope name e.g. \"#bc\"", "type": "string" }, + "traceQuality": { + "$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_tracequality.Quality" + }, "transportCodes": { "$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.PacketTransportCodes" } @@ -4714,6 +4717,13 @@ const docTemplate = `{ }, "sourceBroker": { "type": "string" + }, + "traceQuality": { + "$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_tracequality.Quality" + }, + "wirePathBytes": { + "description": "Exact on-air TRACE SNR bytes, before the display path substitution.", + "type": "string" } } }, @@ -5663,6 +5673,9 @@ const docTemplate = `{ "description": "hex-encoded packet hash", "type": "string" }, + "quality": { + "$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_tracequality.Quality" + }, "rawPath": { "description": "hops as received in the packet", "type": "array", @@ -5717,6 +5730,9 @@ const docTemplate = `{ "type": "string" } }, + "quality": { + "$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_tracequality.Quality" + }, "snrValues": { "description": "SNR per hop from the most complete observation", "type": "array", @@ -5828,6 +5844,21 @@ const docTemplate = `{ } } }, + "github_com_MeshCore-Beacon_beacon-server_internal_tracequality.Quality": { + "type": "object", + "properties": { + "reasons": { + "type": "array", + "items": { + "type": "string" + } + }, + "status": { + "description": "supported, suspect, or ambiguous; supported is not authenticated", + "type": "string" + } + } + }, "internal_api_handlers.APIError": { "type": "object", "properties": { diff --git a/docs/swagger.json b/docs/swagger.json index 8d6c035..1a204f3 100644 --- a/docs/swagger.json +++ b/docs/swagger.json @@ -4581,6 +4581,9 @@ "description": "matched transport scope name e.g. \"#bc\"", "type": "string" }, + "traceQuality": { + "$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_tracequality.Quality" + }, "transportCodes": { "$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.PacketTransportCodes" } @@ -4712,6 +4715,13 @@ }, "sourceBroker": { "type": "string" + }, + "traceQuality": { + "$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_tracequality.Quality" + }, + "wirePathBytes": { + "description": "Exact on-air TRACE SNR bytes, before the display path substitution.", + "type": "string" } } }, @@ -5661,6 +5671,9 @@ "description": "hex-encoded packet hash", "type": "string" }, + "quality": { + "$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_tracequality.Quality" + }, "rawPath": { "description": "hops as received in the packet", "type": "array", @@ -5715,6 +5728,9 @@ "type": "string" } }, + "quality": { + "$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_tracequality.Quality" + }, "snrValues": { "description": "SNR per hop from the most complete observation", "type": "array", @@ -5826,6 +5842,21 @@ } } }, + "github_com_MeshCore-Beacon_beacon-server_internal_tracequality.Quality": { + "type": "object", + "properties": { + "reasons": { + "type": "array", + "items": { + "type": "string" + } + }, + "status": { + "description": "supported, suspect, or ambiguous; supported is not authenticated", + "type": "string" + } + } + }, "internal_api_handlers.APIError": { "type": "object", "properties": { diff --git a/docs/swagger.yaml b/docs/swagger.yaml index 726cdce..8fda64e 100644 --- a/docs/swagger.yaml +++ b/docs/swagger.yaml @@ -788,6 +788,8 @@ definitions: scope: description: matched transport scope name e.g. "#bc" type: string + traceQuality: + $ref: '#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_tracequality.Quality' transportCodes: $ref: '#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.PacketTransportCodes' type: object @@ -887,6 +889,11 @@ definitions: type: number sourceBroker: type: string + traceQuality: + $ref: '#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_tracequality.Quality' + wirePathBytes: + description: Exact on-air TRACE SNR bytes, before the display path substitution. + type: string type: object github_com_MeshCore-Beacon_beacon-server_internal_api.PacketObservationSummary: properties: @@ -1530,6 +1537,8 @@ definitions: packetHash: description: hex-encoded packet hash type: string + quality: + $ref: '#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_tracequality.Quality' rawPath: description: hops as received in the packet items: @@ -1569,6 +1578,8 @@ definitions: items: type: string type: array + quality: + $ref: '#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_tracequality.Quality' snrValues: description: SNR per hop from the most complete observation items: @@ -1642,6 +1653,16 @@ definitions: url: type: string type: object + github_com_MeshCore-Beacon_beacon-server_internal_tracequality.Quality: + properties: + reasons: + items: + type: string + type: array + status: + description: supported, suspect, or ambiguous; supported is not authenticated + type: string + type: object internal_api_handlers.APIError: properties: code: diff --git a/internal/api/packets.go b/internal/api/packets.go index e5f47fe..f2fccb7 100644 --- a/internal/api/packets.go +++ b/internal/api/packets.go @@ -3,6 +3,8 @@ package api +import "github.com/MeshCore-Beacon/beacon-server/internal/tracequality" + import ( "encoding/hex" "encoding/json" @@ -53,19 +55,21 @@ type PacketPathLength struct { // PacketObservationDetail is a full observation including radio settings and resolved path. type PacketObservationDetail struct { - ID int64 `json:"id"` - ObserverID uuid.UUID `json:"observerId"` - ObserverName *string `json:"observerName,omitempty"` - IATA string `json:"iata"` - HeardAt int64 `json:"heardAt"` // epoch ms - PathLength PacketPathLength `json:"pathLength"` - PathBytes *string `json:"pathBytes,omitempty"` // hex-encoded accumulated path hashes - RSSI *int16 `json:"rssi,omitempty"` - SNR *float32 `json:"snr,omitempty"` - PropagationTimeMs *int32 `json:"propagationTimeMs"` // ms since first observation; 0 for first - Radio *PacketRadio `json:"radio,omitempty"` - SourceBroker string `json:"sourceBroker"` - ResolvedPath []ResolvedHop `json:"resolvedPath"` // per-observation resolved path hashes + WirePathBytes *string `json:"wirePathBytes,omitempty"` // Exact on-air TRACE SNR bytes, before the display path substitution. + TraceQuality *tracequality.Quality `json:"traceQuality,omitempty"` + ID int64 `json:"id"` + ObserverID uuid.UUID `json:"observerId"` + ObserverName *string `json:"observerName,omitempty"` + IATA string `json:"iata"` + HeardAt int64 `json:"heardAt"` // epoch ms + PathLength PacketPathLength `json:"pathLength"` + PathBytes *string `json:"pathBytes,omitempty"` // hex-encoded accumulated path hashes + RSSI *int16 `json:"rssi,omitempty"` + SNR *float32 `json:"snr,omitempty"` + PropagationTimeMs *int32 `json:"propagationTimeMs"` // ms since first observation; 0 for first + Radio *PacketRadio `json:"radio,omitempty"` + SourceBroker string `json:"sourceBroker"` + ResolvedPath []ResolvedHop `json:"resolvedPath"` // per-observation resolved path hashes // ResolvedSource/ResolvedDestination are the packet's endpoints, when the payload type // carries one. Prefer the snapshot captured at ingest; legacy observations without a // snapshot use the current node registry. Endpoint matching itself is unchanged: @@ -134,6 +138,7 @@ type PacketTransportCodes struct { // Packet is the full packet representation including all observations and resolved paths. type Packet struct { + TraceQuality *tracequality.Quality `json:"traceQuality,omitempty"` PacketHash string `json:"packetHash"` Header PacketHeader `json:"header"` TransportCodes *PacketTransportCodes `json:"transportCodes,omitempty"` diff --git a/internal/api/traces.go b/internal/api/traces.go index b08fd7c..f4a70d0 100644 --- a/internal/api/traces.go +++ b/internal/api/traces.go @@ -3,29 +3,33 @@ package api +import "github.com/MeshCore-Beacon/beacon-server/internal/tracequality" + // TraceTagSummary is a single trace tag with aggregate metadata. type TraceTagSummary struct { - TraceTag string `json:"traceTag"` // hex-encoded 4-byte tag - FirstHeardAt int64 `json:"firstHeardAt"` // epoch ms - LastHeardAt int64 `json:"lastHeardAt"` // epoch ms - PacketCount int64 `json:"packetCount"` // number of packets with this trace tag - IATACount int64 `json:"iataCount"` // number of distinct IATAs where heard - TraceType string `json:"traceType"` // TRACE or PING - PathHashes []string `json:"pathHashes"` // hops from the most complete observation - SNRValues []float32 `json:"snrValues"` // SNR per hop from the most complete observation + Quality tracequality.Quality `json:"quality"` + TraceTag string `json:"traceTag"` // hex-encoded 4-byte tag + FirstHeardAt int64 `json:"firstHeardAt"` // epoch ms + LastHeardAt int64 `json:"lastHeardAt"` // epoch ms + PacketCount int64 `json:"packetCount"` // number of packets with this trace tag + IATACount int64 `json:"iataCount"` // number of distinct IATAs where heard + TraceType string `json:"traceType"` // TRACE or PING + PathHashes []string `json:"pathHashes"` // hops from the most complete observation + SNRValues []float32 `json:"snrValues"` // SNR per hop from the most complete observation } // TracePacket is a single packet within a trace series, including its // resolved route derived from the trace path hashes. type TracePacket struct { - PacketHash string `json:"packetHash"` // hex-encoded packet hash - RouteType int16 `json:"routeType"` // numeric route type - RouteTypeName string `json:"routeTypeName"` // human-readable route type - Scope *string `json:"scope,omitempty"` // transport scope name, if known - FirstHeardAt int64 `json:"firstHeardAt"` // epoch ms - LastHeardAt int64 `json:"lastHeardAt"` // epoch ms - RawPath []RawHop `json:"rawPath"` // hops as received in the packet - ResolvedRoute []ResolvedHop `json:"resolvedRoute"` // hops resolved to known nodes + Quality tracequality.Quality `json:"quality"` + PacketHash string `json:"packetHash"` // hex-encoded packet hash + RouteType int16 `json:"routeType"` // numeric route type + RouteTypeName string `json:"routeTypeName"` // human-readable route type + Scope *string `json:"scope,omitempty"` // transport scope name, if known + FirstHeardAt int64 `json:"firstHeardAt"` // epoch ms + LastHeardAt int64 `json:"lastHeardAt"` // epoch ms + RawPath []RawHop `json:"rawPath"` // hops as received in the packet + ResolvedRoute []ResolvedHop `json:"resolvedRoute"` // hops resolved to known nodes } // RawHop is a single hop in a trace path as received in the packet. diff --git a/internal/ingest/ingest.go b/internal/ingest/ingest.go index 6c3abb7..46d7f83 100644 --- a/internal/ingest/ingest.go +++ b/internal/ingest/ingest.go @@ -158,6 +158,7 @@ type DB interface { // ResolvePathHashes returns a list of node UUIDs for the given path hash prefixes and IATA. ResolvePathHashes(ctx context.Context, iata string, hashes [][]byte) (map[string][]api.ResolvedPathEntry, error) + ResolveTracePathHashes(ctx context.Context, iata string, hashes [][]byte) (map[string][]api.ResolvedPathEntry, error) // ResolveEndpointHashes matches one-byte logical endpoints, including companions. ResolveEndpointHashes(ctx context.Context, iata string, hashes [][]byte) (map[string][]api.ResolvedPathEntry, error) diff --git a/internal/ingest/ingest_test.go b/internal/ingest/ingest_test.go index cffa465..0c45b48 100644 --- a/internal/ingest/ingest_test.go +++ b/internal/ingest/ingest_test.go @@ -266,6 +266,9 @@ func (s *stubDB) GetObserverScopes(_ context.Context, _ uuid.UUID) ([]string, er func (s *stubDB) ResolvePathHashes(_ context.Context, _ string, _ [][]byte) (map[string][]api.ResolvedPathEntry, error) { return nil, nil } +func (s *stubDB) ResolveTracePathHashes(_ context.Context, _ string, _ [][]byte) (map[string][]api.ResolvedPathEntry, error) { + return nil, nil +} func (s *stubDB) ResolveEndpointHashes(_ context.Context, _ string, _ [][]byte) (map[string][]api.ResolvedPathEntry, error) { return nil, nil diff --git a/internal/ingest/packet.go b/internal/ingest/packet.go index e875492..2a4b84e 100644 --- a/internal/ingest/packet.go +++ b/internal/ingest/packet.go @@ -18,6 +18,7 @@ import ( "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/MeshCore-Beacon/beacon-server/internal/tracequality" "github.com/google/uuid" "github.com/meshcore-go/meshcore-go" ) @@ -202,13 +203,14 @@ type parsedGroupEnvelope struct { // is the accumulated per-hop SNR sequence carried in the packet header's // path field (a protocol-level repurposing distinct from the payload). type parsedTrace struct { - Raw string `json:"raw"` - Type string `json:"type"` - TraceTag string `json:"traceTag"` - AuthCode uint32 `json:"authCode"` - Flags byte `json:"flags"` - PathHashes []string `json:"pathHashes"` - SNRValues []float32 `json:"snrValues"` + Quality tracequality.Quality `json:"quality"` + Raw string `json:"raw"` + Type string `json:"type"` + TraceTag string `json:"traceTag"` + AuthCode uint32 `json:"authCode"` + Flags byte `json:"flags"` + PathHashes []string `json:"pathHashes"` + SNRValues []float32 `json:"snrValues"` } // parsedAck is the parsed form of an ACK payload (ยง2.12). Checksum is the @@ -557,7 +559,10 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [ hashes = append(hashes, hex.EncodeToString(h)) rawHashes = append(rawHashes, h) } - traceRawHashes = rawHashes + quality := tracequality.Assess(packet.Payload, int(packet.PayloadVer()), int(packet.RouteType()), int(packet.PathLength)) + if quality.Status == "supported" { + traceRawHashes = rawHashes + } // SNR values are in packet.Path, one signed int8 per consumed hop snrValues := make([]float32, 0, len(packet.Path)) for _, b := range packet.Path { @@ -568,6 +573,7 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [ traceType = "PING" } pt := parsedTrace{ + Quality: quality, Raw: hex.EncodeToString(packet.Payload), Type: traceType, TraceTag: hex.EncodeToString(uint32ToBytes(trace.Tag)), @@ -585,8 +591,8 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [ // originator isn't in PathHashes) and is skipped. Either side // resolving ambiguously (>1 candidate) or not at all (0 // candidates) skips that specific pair. - if len(rawHashes) >= 2 { - resolved, rErr := w.db.ResolvePathHashes(ctx, iata, rawHashes) + if quality.Status == "supported" && len(rawHashes) >= 2 { + resolved, rErr := w.db.ResolveTracePathHashes(ctx, iata, rawHashes) if rErr == nil { for i := 1; i < len(rawHashes); i++ { if i >= len(snrValues) { @@ -822,12 +828,20 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [ if packet.PayloadType() == meshcore.PayloadTypeTrace { hashes = traceRawHashes } - resolved, err := w.db.ResolvePathHashes(ctx, iata, hashes) + var resolved map[string][]api.ResolvedPathEntry + if packet.PayloadType() == meshcore.PayloadTypeTrace { + resolved, err = w.db.ResolveTracePathHashes(ctx, iata, hashes) + } else { + resolved, err = w.db.ResolvePathHashes(ctx, iata, hashes) + } if err != nil { w.log.Error("path resolution failed", "error", err) } var resolvedIDs []uuid.UUID for _, entries := range resolved { + if len(entries) != 1 { + continue + } for _, e := range entries { resolvedIDs = append(resolvedIDs, e.NodeID) } @@ -846,13 +860,31 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [ nodeIDs = append(nodeIDs, entries[0].NodeID) hashPrefixes = append(hashPrefixes, hash) } - if allHigh && len(nodeIDs) > 1 { + if allHigh && len(nodeIDs) > 1 && packet.PayloadType() != meshcore.PayloadTypeTrace { if err := w.db.UpsertKnownRoute(ctx, nodeIDs, hashPrefixes, iata, int32(len(nodeIDs))); err != nil { w.log.Error("failed to upsert known route", "error", err) } } } - w.runCapabilityDetection(ctx, packet.PayloadType(), packet.PathHashSize(), resolvedIDs) + hashSize := packet.PathHashSize() + if packet.PayloadType() == meshcore.PayloadTypeTrace { + hashSize = 1 + if len(traceRawHashes) > 0 { + hashSize = uint8(len(traceRawHashes[0])) + } + // Only consumed, uniquely resolved hops demonstrate trace capability. + resolvedIDs = nil + for i, hash := range traceRawHashes { + if i >= len(packet.Path) { + break + } + entries := resolved[hex.EncodeToString(hash)] + if len(entries) == 1 { + resolvedIDs = append(resolvedIDs, entries[0].NodeID) + } + } + } + w.runCapabilityDetection(ctx, packet.PayloadType(), hashSize, resolvedIDs) if inserted || repeat { if inserted { diff --git a/internal/ingest/trace_quality_test.go b/internal/ingest/trace_quality_test.go new file mode 100644 index 0000000..89c36b1 --- /dev/null +++ b/internal/ingest/trace_quality_test.go @@ -0,0 +1,75 @@ +package ingest + +import ( + "context" + "encoding/hex" + "encoding/json" + "testing" + + "github.com/MeshCore-Beacon/beacon-server/internal/api" + "github.com/google/uuid" + "github.com/meshcore-go/meshcore-go" +) + +type traceSafetyDB struct { + stubDB + saved UpsertPacketParams + resolutions, neighbors, routes int +} + +func (s *traceSafetyDB) UpsertPacket(_ context.Context, p UpsertPacketParams) (bool, error) { + s.saved = p + return true, nil +} +func (s *traceSafetyDB) ResolveTracePathHashes(_ context.Context, _ string, hashes [][]byte) (map[string][]api.ResolvedPathEntry, error) { + if len(hashes) > 0 { + s.resolutions++ + } + out := map[string][]api.ResolvedPathEntry{} + for i, h := range hashes { + id := uuid.UUID{} + id[15] = byte(i + 1) + out[hex.EncodeToString(h)] = []api.ResolvedPathEntry{{NodeID: id}} + } + return out, nil +} +func (s *traceSafetyDB) UpsertNodeNeighbor(_ context.Context, _, _ uuid.UUID, _ string, _ *float32, _ *string) error { + s.neighbors++ + return nil +} +func (s *traceSafetyDB) UpsertKnownRoute(_ context.Context, _ []uuid.UUID, _ [][]byte, _ string, _ int32) error { + s.routes++ + return nil +} + +func TestTraceQualityDoesNotPromoteMalformedOrUnvisitedHops(t *testing.T) { + for _, bad := range []bool{false, true} { + w, _ := newTestWorker() + db := &traceSafetyDB{} + w.db = db + payload, _ := (&meshcore.Trace{Tag: 7, Flags: 1, PathHashes: []byte{0xaa, 0xbb, 0xcc, 0xdd, 0xee, 0xff}}).ToBytes() + version := byte(0) + if bad { + version = 1 + } + pkt := &meshcore.Packet{Header: meshcore.MakeHeader(meshcore.RouteTypeDirect, meshcore.PayloadTypeTrace, version), PathLength: 2, Path: []byte{0, 252}, Payload: payload} + w.handlePacket(context.Background(), "YKF", "0102", packetEnvelope(t, pkt)) + if len(db.saved.RawPayload) == 0 || len(db.traceHearings) != 1 { + t.Fatal("raw evidence lost") + } + var parsed parsedTrace + if err := json.Unmarshal(db.saved.ParsedPayload, &parsed); err != nil { + t.Fatal(err) + } + if bad { + if parsed.Quality.Status != "suspect" || db.neighbors != 0 || db.resolutions != 0 || len(db.setCapabilityCalls) != 0 { + t.Fatalf("malformed trace promoted: %+v", db) + } + } else if parsed.Quality.Status != "supported" || db.neighbors != 1 || len(db.setCapabilityCalls) != 2 { + t.Fatalf("valid consumed links/capabilities missing: %+v", db) + } + if db.routes != 0 { + t.Fatal("planned route promoted as observed") + } + } +} diff --git a/internal/tracequality/quality.go b/internal/tracequality/quality.go new file mode 100644 index 0000000..e885753 --- /dev/null +++ b/internal/tracequality/quality.go @@ -0,0 +1,96 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later + +// Package tracequality checks TRACE structure, not authenticity or RF plausibility. +package tracequality + +import "encoding/hex" + +type Quality struct { + Status string `json:"status"` // supported, suspect, or ambiguous; supported is not authenticated + Reasons []string `json:"reasons"` +} + +func (q *Quality) Add(reason string) { + for _, r := range q.Reasons { + if r == reason { + return + } + } + q.Reasons = append(q.Reasons, reason) + if reason == "ambiguous_prefix" && q.Status != "suspect" { + q.Status = "ambiguous" + } else if reason != "ambiguous_prefix" { + q.Status = "suspect" + } +} + +// Assess uses TRACE's power-of-two payload hash widths, NOT the ordinary +// packet path's 1/2/3-byte widths. The path field contains consumed-hop SNRs. +// -1 means a header/path field is unavailable in a retained summary. +// Mesh.cpp onRecvPacket/sendDirect: a trace uses direct routing, flags bits +// 0..1, and fewer than 64 SNR bytes. Partial probes and return loops are valid. +func Assess(raw []byte, version, route, pathLength int) Quality { + q := Quality{Status: "supported", Reasons: []string{}} + if version > 0 { + q.Add("unsupported_version") + } + if route >= 0 && route != 2 && route != 3 { + q.Add("non_direct_trace") + } + if len(raw) < 10 { + q.Add("incomplete_payload") + return q + } + if len(raw) > 184 { + q.Add("payload_too_large") + } + if raw[8]&0xfc != 0 { + q.Add("unsupported_flags") + } + size := 1 << (raw[8] & 3) + if (len(raw)-9)%size != 0 { + q.Add("misaligned_hashes") + } + count := (len(raw) - 9) / size + if count > 63 { + q.Add("too_many_hops") + } + if pathLength > 63 || pathLength > count { + q.Add("invalid_snr_path") + } + return q +} + +// Stored also checks old parsed rows, without rewriting or deleting the evidence. +func Stored(raw string, flags byte, hashes []string, snrs []float32) Quality { + b, err := hex.DecodeString(raw) + if err != nil { + return Quality{Status: "suspect", Reasons: []string{"incomplete_payload"}} + } + if raw == "" { // Older exports may contain only decoded fields. + b = make([]byte, 9) + b[8] = flags + for _, h := range hashes { + v, e := hex.DecodeString(h) + if e != nil { + return Quality{Status: "suspect", Reasons: []string{"misaligned_hashes"}} + } + b = append(b, v...) + } + } + q := Assess(b, -1, -1, len(snrs)) + size := 1 << (flags & 3) + if len(b) >= 9 && b[8] != flags { + q.Add("misaligned_hashes") + } + for _, h := range hashes { + if len(h) != 2*size { + q.Add("misaligned_hashes") + } + } + if len(b) >= 9 && len(hashes) != (len(b)-9)/size { + q.Add("misaligned_hashes") + } + return q +} diff --git a/internal/tracequality/quality_test.go b/internal/tracequality/quality_test.go new file mode 100644 index 0000000..f811ae3 --- /dev/null +++ b/internal/tracequality/quality_test.go @@ -0,0 +1,51 @@ +package tracequality + +import ( + "bytes" + "slices" + "testing" +) + +func TestTraceFormat(t *testing.T) { + for flags := byte(0); flags < 4; flags++ { + raw := make([]byte, 9) + raw[8] = flags + raw = append(raw, bytes.Repeat([]byte{0xaa}, 3*(1<