// Copyright 2026 Beacon Contributors // SPDX-License-Identifier: AGPL-3.0-or-later package ingest import ( "context" "crypto/ed25519" "encoding/binary" "encoding/hex" "encoding/json" "strings" "testing" "time" "github.com/MeshCore-Beacon/beacon-server/internal/hub" "github.com/MeshCore-Beacon/beacon-server/internal/keystore" "github.com/meshcore-go/meshcore-go" ) // mapKeys is a ChannelKeyStore stub that returns a fixed set of entries for // any hash, letting tests control whether a channel's key is "known". type mapKeys struct { entries map[byte][]keystore.Entry } func (k *mapKeys) GetKey(hash []byte) []keystore.Entry { if len(hash) == 0 { return nil } return k.entries[hash[0]] } // buildAdvertPacket signs (or, if tamper is true, signs then mutates) an // advert payload and wraps it in a minimal Packet with no path (zero-hop). func buildAdvertPacket(t *testing.T, tamper bool) *meshcore.Packet { return buildAdvertPacketWithData(t, []byte{meshcore.AdvertTypeRepeater}, tamper) } func buildAdvertPacketWithData(t *testing.T, data []byte, tamper bool) *meshcore.Packet { t.Helper() pub, priv, err := ed25519.GenerateKey(nil) if err != nil { t.Fatalf("generate key: %v", err) } id, err := meshcore.NewIdentityFromBytes(pub) if err != nil { t.Fatalf("new identity: %v", err) } advert := &meshcore.Advert{ PublicKey: id, Timestamp: 12345, RawAppData: data, } advert.Sign(priv) if tamper { // Flip the device-role bits after signing, as if a relay (or an // attacker) altered the payload in transit. advert.RawAppData = []byte{meshcore.AdvertTypeRoom} } payload, err := advert.ToBytes() if err != nil { t.Fatalf("advert to bytes: %v", err) } return &meshcore.Packet{ Header: meshcore.MakeHeader(meshcore.RouteTypeFlood, meshcore.PayloadTypeAdvert, 0), Payload: payload, } } func TestAdvertLocationPresence(t *testing.T) { for _, tc := range []struct { name string present, tamper, clear bool lat, lon int32 }{ {"ordinary location", true, false, false, 45000000, -75000000}, {"explicit reset", true, false, true, 0, 0}, {"zero latitude", true, false, false, 0, -75000000}, {"zero longitude", true, false, false, 45000000, 0}, {"no location", false, false, false, 0, 0}, {"tampered reset", true, true, false, 0, 0}, } { t.Run(tc.name, func(t *testing.T) { data := []byte{meshcore.AdvertTypeRepeater} if tc.present { // Preserve the wire presence bit even for zero coordinates; the // library's AppData encoder omits location when both values are zero. data[0] |= meshcore.AdvertLatLonMask data = binary.LittleEndian.AppendUint32(data, uint32(tc.lat)) data = binary.LittleEndian.AppendUint32(data, uint32(tc.lon)) } worker, store := newTestWorker() packet := buildAdvertPacketWithData(t, data, tc.tamper) worker.handlePayloadTypeSideEffects(context.Background(), packet, "YOW", []byte{1}, RadioSettings{}, nil, nil, nil, 0) if tc.tamper { if store.upsertNodeCalls != 0 { t.Fatal("invalid signature updated the node") } return } if store.upsertNodeCalls != 1 { t.Fatal("signed advert did not update the node") } got := store.upsertNodeParams if got.ClearLocation != tc.clear { t.Fatalf("ClearLocation = %v, want %v", got.ClearLocation, tc.clear) } if !tc.present || tc.clear { if got.Latitude != nil || got.Longitude != nil { t.Fatal("absent location became an update") } return } if got.Latitude == nil || got.Longitude == nil || *got.Latitude != float64(tc.lat)/1e6 || *got.Longitude != float64(tc.lon)/1e6 { t.Fatal("advertised coordinates did not reach the node update") } }) } } func TestAdvertClearEmitsNullLocation(t *testing.T) { w, _ := newTestWorker() go w.hub.Run() client := w.hub.NewClient() defer w.hub.Remove(client) w.hub.AddScope(client, "clear", hub.Scope{Events: []hub.EventType{hub.EventNodeUpdate, hub.EventObserverStatus}}) ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() waitForSummarySubscriber(t, ctx, w.hub, client) for _, tc := range []struct { name string present, want bool lat, lon int32 wantLat string wantLng string }{ {"explicit reset sends null", true, true, 0, 0, "null", "null"}, {"omission sends nothing", false, false, 0, 0, "", ""}, {"real location sends coordinates", true, true, 45000000, -75000000, "45", "-75"}, } { t.Run(tc.name, func(t *testing.T) { data := []byte{meshcore.AdvertTypeRepeater} if tc.present { data[0] |= meshcore.AdvertLatLonMask data = binary.LittleEndian.AppendUint32(data, uint32(tc.lat)) data = binary.LittleEndian.AppendUint32(data, uint32(tc.lon)) } w.handlePayloadTypeSideEffects(ctx, buildAdvertPacketWithData(t, data, false), "AAA", []byte{1}, RadioSettings{}, nil, nil, nil, 0) for { select { case event := <-client.Send: if event.Type != hub.EventNodeUpdate { continue } payload := string(event.Payload) for field, want := range map[string]string{`"lat"`: tc.wantLat, `"lng"`: tc.wantLng} { if tc.want && !strings.Contains(payload, field+":"+want+",") { t.Fatalf("%s != %s: %s", field, want, payload) } if !tc.want && strings.Contains(payload, field) { t.Fatalf("%s present on omission: %s", field, payload) } } return case <-ctx.Done(): t.Fatal("node update not received") } } }) } } func TestHandlePayloadTypeSideEffects_Advert_ValidSignature_UpsertsNode(t *testing.T) { w, db := newTestWorker() packet := buildAdvertPacket(t, false) w.handlePayloadTypeSideEffects(context.Background(), packet, "TEST", []byte{0x01}, RadioSettings{}, nil, nil, nil, 0) if db.upsertNodeCalls != 1 { t.Errorf("expected UpsertNode to be called once for a validly-signed advert, got %d", db.upsertNodeCalls) } } func TestHandlePayloadTypeSideEffects_Advert_InvalidSignature_SkipsUpsert(t *testing.T) { w, db := newTestWorker() packet := buildAdvertPacket(t, true) w.handlePayloadTypeSideEffects(context.Background(), packet, "TEST", []byte{0x01}, RadioSettings{}, nil, nil, nil, 0) if db.upsertNodeCalls != 0 { t.Errorf("expected UpsertNode NOT to be called for a tampered advert, got %d calls", db.upsertNodeCalls) } } func buildGrpTxtPacket(t *testing.T, channelHash byte, psk []byte) *meshcore.Packet { t.Helper() grpTxt, err := (&meshcore.GroupTextPayload{ Timestamp: 1000, Sender: "ded", Text: "hello", }).Encrypt(channelHash, psk) if err != nil { t.Fatalf("encrypt group text: %v", err) } payload, err := grpTxt.ToBytes() if err != nil { t.Fatalf("group text to bytes: %v", err) } return &meshcore.Packet{ Header: meshcore.MakeHeader(meshcore.RouteTypeFlood, meshcore.PayloadTypeGrpTxt, 0), Payload: payload, } } func TestHandlePayloadTypeSideEffects_GrpTxt_KnownKey_OnlyUpsertsKeyedChannel(t *testing.T) { w, db := newTestWorker() psk := make([]byte, 16) channelHash := byte(0x42) w.keys = &mapKeys{entries: map[byte][]keystore.Entry{ channelHash: {{Key: psk, Fingerprint: []byte{0xAA}, Name: "Public", Hashtag: "public"}}, }} packet := buildGrpTxtPacket(t, channelHash, psk) w.handlePayloadTypeSideEffects(context.Background(), packet, "TEST", []byte{0x02}, RadioSettings{}, nil, nil, nil, 0) if db.upsertChannelCalls != 1 { t.Errorf("expected UpsertChannel to be called once, got %d", db.upsertChannelCalls) } if db.upsertChannelHashOnlyCalls != 0 { t.Errorf("expected UpsertChannelHashOnly NOT to be called when the key is known, got %d calls", db.upsertChannelHashOnlyCalls) } } func TestHandlePayloadTypeSideEffects_GrpTxt_UnknownKey_OnlyUpsertsHashOnlyChannel(t *testing.T) { w, db := newTestWorker() // default stubKeys returns no entries for any hash channelHash := byte(0x99) packet := buildGrpTxtPacket(t, channelHash, make([]byte, 16)) w.handlePayloadTypeSideEffects(context.Background(), packet, "TEST", []byte{0x03}, RadioSettings{}, nil, nil, nil, 0) if db.upsertChannelHashOnlyCalls != 1 { t.Errorf("expected UpsertChannelHashOnly to be called once for an unknown-key channel, got %d", db.upsertChannelHashOnlyCalls) } if db.upsertChannelCalls != 0 { t.Errorf("expected UpsertChannel NOT to be called when the key is unknown, got %d calls", db.upsertChannelCalls) } } // packetEnvelope wraps a packet in the minimal broker JSON that handlePacket expects. func packetEnvelope(t *testing.T, packet *meshcore.Packet) []byte { t.Helper() raw, err := packet.ToBytes() if err != nil { t.Fatalf("packet to bytes: %v", err) } env, err := json.Marshal(map[string]string{ "raw": hex.EncodeToString(raw), "timestamp": time.Now().UTC().Format("2006-01-02T15:04:05.000000"), }) if err != nil { t.Fatalf("marshal envelope: %v", err) } return env } func TestHandlePacket_GrpTxt_UpsertsChannelIATA(t *testing.T) { w, db := newTestWorker() db.observationInserted = true envelope := packetEnvelope(t, buildGrpTxtPacket(t, 0x1a, make([]byte, 16))) w.handlePacket(context.Background(), "YOW", "0102", envelope) if db.upsertChannelIATACalls != 1 { t.Errorf("expected UpsertChannelIATA to be called once for a stored group text, got %d", db.upsertChannelIATACalls) } } func TestHandlePacket_GrpTxt_DedupObservation_StillUpsertsChannelIATA(t *testing.T) { w, db := newTestWorker() // stub reports the observation as a duplicate envelope := packetEnvelope(t, buildGrpTxtPacket(t, 0x1a, make([]byte, 16))) w.handlePacket(context.Background(), "YOW", "0102", envelope) if db.upsertChannelIATACalls != 1 { t.Errorf("expected UpsertChannelIATA to run for a duplicate observation too, got %d calls", db.upsertChannelIATACalls) } } func TestHandlePacket_Advert_SkipsChannelIATA(t *testing.T) { w, db := newTestWorker() db.observationInserted = true envelope := packetEnvelope(t, buildAdvertPacket(t, false)) w.handlePacket(context.Background(), "YOW", "0102", envelope) if db.upsertChannelIATACalls != 0 { t.Errorf("expected UpsertChannelIATA NOT to be called for a non-channel packet, got %d calls", db.upsertChannelIATACalls) } if len(db.traceHearings) != 0 { t.Errorf("expected RecordTrace NOT to be called for a non-trace packet, got %d calls", len(db.traceHearings)) } } func buildTracePacket(t *testing.T) *meshcore.Packet { t.Helper() payload, err := (&meshcore.Trace{Tag: 0xdeadbeef, AuthCode: 1, PathHashes: []byte{0xaa, 0xbb}}).ToBytes() if err != nil { t.Fatalf("trace to bytes: %v", err) } return &meshcore.Packet{ Header: meshcore.MakeHeader(meshcore.RouteTypeFlood, meshcore.PayloadTypeTrace, 0), Payload: payload, } } func TestHandlePacket_Trace_RecordsTrace(t *testing.T) { w, db := newTestWorker() db.observationInserted = true db.packetIsNew = true envelope := packetEnvelope(t, buildTracePacket(t)) w.handlePacket(context.Background(), "YOW", "0102", envelope) db.packetIsNew, db.observationInserted = false, false w.handlePacket(context.Background(), "YOW", "0102", envelope) if len(db.traceHearings) != 2 { t.Fatalf("expected RecordTrace for every hearing, got %d", len(db.traceHearings)) } for _, h := range db.traceHearings { if h.IATA != "YOW" || h.HeardAt.IsZero() || hex.EncodeToString(h.TraceTag) != "efbeadde" { t.Errorf("unexpected hearing %+v", h) } } }