dandri/beacon-server
1
0 Fork 0
mirror of https://github.com/MeshCore-Beacon/beacon-server.git synced 2026-10-22 14:11:27 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(traces): validate trace frames and preserve ambiguous identities

Browse Source
This commit is contained in:
n30nex committed 2026-10-03 23:58:00 -04:00
1 parent 8c7fbb8423
commit c554238485
20 files changed
+763 -126
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

db/packets.go
+20 -3
View File
@@ -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
db/queries/queries.sql
+20 -1
View File
@@ -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;
db/sqlc/mock/querier.go
+16 -1
View File
@@ -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()
db/sqlc/querier.go
+3
View File
@@ -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
db/sqlc/queries.sql.go
+93 -16
View File
@@ -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,
db/trace_quality_integration_test.go
+77
View File
@@ -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")
}
}
}
db/traces.go
+127 -52
View File
@@ -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
db/traces_integration_test.go
+8 -5
View File
@@ -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")
db/traces_test.go
+7 -6
View File
@@ -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")
docs/docs.go
+31
View File
@@ -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": {
docs/swagger.json
+31
View File
@@ -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": {
docs/swagger.yaml
+21
View File
@@ -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:
internal/api/packets.go
+18 -13
View File
@@ -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"`
internal/api/traces.go
+20 -16
View File
@@ -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.
internal/ingest/ingest.go
+1
View File
@@ -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)
internal/ingest/ingest_test.go
+3
View File
@@ -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
internal/ingest/packet.go
+45 -13
View File
@@ -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 {
internal/ingest/trace_quality_test.go
+75
View File
@@ -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")
}
}
}
internal/tracequality/quality.go
+96
View File
@@ -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
}
internal/tracequality/quality_test.go
+51
View File
@@ -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<<flags))...)
for _, consumed := range []int{0, 1, 3} {
if q := Assess(raw, 0, 2, consumed); q.Status != "supported" {
t.Fatalf("valid width %d, partial/loop trace: %+v", 1<<flags, q)
}
}
}
cases := []struct {
name string
version, route, path, flags, count int
reason string
}{
{"version", 1, 2, 0, 0, 3, "unsupported_version"},
{"flood", 0, 1, 0, 0, 3, "non_direct_trace"},
{"flags", 0, 2, 0, 133, 74, "unsupported_flags"},
{"101 hops", 0, 2, 55, 0, 101, "too_many_hops"},
{"145 hops", 0, 2, 10, 0, 145, "too_many_hops"},
{"ordinary multibyte metadata is not trace SNR count", 0, 2, 65, 1, 3, "invalid_snr_path"},
{"extra SNR", 0, 2, 4, 0, 3, "invalid_snr_path"},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
raw := make([]byte, 9)
raw[8] = byte(tc.flags)
raw = append(raw, bytes.Repeat([]byte{0xaa}, tc.count*(1<<(tc.flags&3)))...)
q := Assess(raw, tc.version, tc.route, tc.path)
if q.Status != "suspect" || !slices.Contains(q.Reasons, tc.reason) {
t.Fatalf("%+v", q)
}
})
}
if q := Assess([]byte{0, 0, 0, 0, 0, 0, 0, 0, 2, 0, 1, 2}, 0, 2, 0); !slices.Contains(q.Reasons, "misaligned_hashes") {
t.Fatal(q)
}
// Legitimate missing reports, negative SNR, and a returning loop are not corruption.
if q := Stored("", 0, []string{"aa", "bb", "aa"}, []float32{-12.25, 0}); q.Status != "supported" {
t.Fatal(q)
}
}