diff --git a/cmd/beacon/main.go b/cmd/beacon/main.go index bdad2fd..588bc28 100644 --- a/cmd/beacon/main.go +++ b/cmd/beacon/main.go @@ -197,6 +197,17 @@ func main() { keys := keystore.NewMapKeyStore(entries) + // ── Backfill channel messages ──────────────────────────────────────────── + // Packets whose channel key wasn't yet configured at ingest time were stored as + // hash-only channels and never decrypted. Retry them now against the keystore we just + // built, so adding a channel key to the config surfaces its history on the next boot + // instead of leaving it stranded in the DB indefinitely. + if n, err := ingest.BackfillChannelMessages(ctx, store, keys); err != nil { + log.Printf("config: channel message backfill failed: %v", err) + } else if n > 0 { + log.Printf("config: backfilled %d previously-undecrypted channel message(s)", n) + } + // ── Build geographic ingest filter ─────────────────────────────────────────────────────────── allowedIATAs := iatadb.BuildAllowedSet(cfg.Ingest.AllowCountries, cfg.Ingest.AllowContinents) if allowedIATAs != nil { diff --git a/db/channels.go b/db/channels.go index 9c652fc..dbdbddc 100644 --- a/db/channels.go +++ b/db/channels.go @@ -47,6 +47,24 @@ func (s *Store) UpsertChannelHashOnly(ctx context.Context, channelHash []byte) ( return int(rowID), nil } +// ListUndecryptedGroupTextPackets returns GRP_TXT packets never successfully decrypted -- +// see internal/ingest.BackfillChannelMessages, which retries these against the current +// keystore at boot. +func (s *Store) ListUndecryptedGroupTextPackets(ctx context.Context) ([]ingest.UndecryptedPacket, error) { + rows, err := s.q.ListUndecryptedGroupTextPackets(ctx) + if err != nil { + return nil, err + } + packets := make([]ingest.UndecryptedPacket, 0, len(rows)) + for _, v := range rows { + packets = append(packets, ingest.UndecryptedPacket{ + PacketHash: v.PacketHash, + RawPayload: v.RawPayload, + }) + } + return packets, nil +} + func (s *Store) UpsertChannelIATA(ctx context.Context, channelHash []byte, iata string, heardAt time.Time) error { return s.q.UpsertChannelIATA(ctx, sqlc.UpsertChannelIATAParams{ ChannelHash: channelHash, diff --git a/db/queries/queries.sql b/db/queries/queries.sql index 2cef212..b2253b3 100644 --- a/db/queries/queries.sql +++ b/db/queries/queries.sql @@ -703,6 +703,14 @@ ON CONFLICT (channel_hash) WHERE key_fingerprint IS NULL DO UPDATE SET last_seen = NOW() RETURNING id; +-- name: ListUndecryptedGroupTextPackets :many +-- Returns GRP_TXT packets (payload_type=5) never successfully decrypted. Used at boot to +-- retry decryption against the current keystore for packets whose channel key was only added +-- to the config after they'd already been ingested -- see +-- internal/ingest.BackfillChannelMessages. +SELECT packet_hash, raw_payload FROM packets +WHERE payload_type = 5 AND decrypted IS NOT TRUE; + -- name: UpsertChannelIATA :exec -- Refreshes at most hourly so repeat hears don't churn the row. INSERT INTO channel_iatas (channel_hash, iata, last_heard) diff --git a/db/sqlc/mock/querier.go b/db/sqlc/mock/querier.go index 1cb3b7a..91f7e0f 100644 --- a/db/sqlc/mock/querier.go +++ b/db/sqlc/mock/querier.go @@ -982,6 +982,21 @@ func (mr *MockQuerierMockRecorder) ListTraceTags(ctx, arg any) *gomock.Call { return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ListTraceTags", reflect.TypeOf((*MockQuerier)(nil).ListTraceTags), ctx, arg) } +// ListUndecryptedGroupTextPackets mocks base method. +func (m *MockQuerier) ListUndecryptedGroupTextPackets(ctx context.Context) ([]db.ListUndecryptedGroupTextPacketsRow, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "ListUndecryptedGroupTextPackets", ctx) + ret0, _ := ret[0].([]db.ListUndecryptedGroupTextPacketsRow) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// ListUndecryptedGroupTextPackets indicates an expected call of ListUndecryptedGroupTextPackets. +func (mr *MockQuerierMockRecorder) ListUndecryptedGroupTextPackets(ctx any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ListUndecryptedGroupTextPackets", reflect.TypeOf((*MockQuerier)(nil).ListUndecryptedGroupTextPackets), ctx) +} + // ReconfirmNeighbors mocks base method. func (m *MockQuerier) ReconfirmNeighbors(ctx context.Context) error { m.ctrl.T.Helper() diff --git a/db/sqlc/querier.go b/db/sqlc/querier.go index a56029f..03a6a84 100644 --- a/db/sqlc/querier.go +++ b/db/sqlc/querier.go @@ -156,6 +156,11 @@ type Querier interface { // IATA membership comes from trace_iatas (joining observations here spilled the // hash join). Per-tag details filled in only for the returned page. ListTraceTags(ctx context.Context, arg ListTraceTagsParams) ([]ListTraceTagsRow, error) + // Returns GRP_TXT packets (payload_type=5) never successfully decrypted. Used at boot to + // retry decryption against the current keystore for packets whose channel key was only added + // to the config after they'd already been ingested -- see + // internal/ingest.BackfillChannelMessages. + ListUndecryptedGroupTextPackets(ctx context.Context) ([]ListUndecryptedGroupTextPacketsRow, error) // Delete node_neighbors where the neighbor has departed from node_short_ids // for that IATA, or where its prefix_4 is now ambiguous. ReconfirmNeighbors(ctx context.Context) error diff --git a/db/sqlc/queries.sql.go b/db/sqlc/queries.sql.go index 75c87ee..4156701 100644 --- a/db/sqlc/queries.sql.go +++ b/db/sqlc/queries.sql.go @@ -3029,6 +3029,40 @@ func (q *Queries) ListTraceTags(ctx context.Context, arg ListTraceTagsParams) ([ return items, nil } +const listUndecryptedGroupTextPackets = `-- name: ListUndecryptedGroupTextPackets :many +SELECT packet_hash, raw_payload FROM packets +WHERE payload_type = 5 AND decrypted IS NOT TRUE +` + +type ListUndecryptedGroupTextPacketsRow struct { + PacketHash []byte `json:"packet_hash"` + RawPayload []byte `json:"raw_payload"` +} + +// Returns GRP_TXT packets (payload_type=5) never successfully decrypted. Used at boot to +// retry decryption against the current keystore for packets whose channel key was only added +// to the config after they'd already been ingested -- see +// internal/ingest.BackfillChannelMessages. +func (q *Queries) ListUndecryptedGroupTextPackets(ctx context.Context) ([]ListUndecryptedGroupTextPacketsRow, error) { + rows, err := q.db.Query(ctx, listUndecryptedGroupTextPackets) + if err != nil { + return nil, err + } + defer rows.Close() + items := []ListUndecryptedGroupTextPacketsRow{} + for rows.Next() { + var i ListUndecryptedGroupTextPacketsRow + if err := rows.Scan(&i.PacketHash, &i.RawPayload); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + const reconfirmNeighbors = `-- name: ReconfirmNeighbors :exec DELETE FROM node_neighbors nn WHERE NOT EXISTS ( diff --git a/internal/ingest/backfill.go b/internal/ingest/backfill.go new file mode 100644 index 0000000..c046912 --- /dev/null +++ b/internal/ingest/backfill.go @@ -0,0 +1,123 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later + +package ingest + +import ( + "context" + "encoding/hex" + "fmt" + "log" + "strings" + "time" + + "github.com/MeshCore-Beacon/beacon-server/internal/keystore" + "github.com/meshcore-go/meshcore-go" +) + +// UndecryptedPacket is a stored GRP_TXT packet that was never successfully decrypted -- +// either because no key was known for its channel hash at ingest time, or because the keys +// that were known didn't match. See ListUndecryptedGroupTextPackets. +type UndecryptedPacket struct { + PacketHash []byte + RawPayload []byte +} + +// DecryptGroupTextResult is the outcome of a successful DecryptGroupText call. +type DecryptGroupTextResult struct { + Payload *meshcore.GroupTextPayload + ChannelID int + ChannelHash []byte + Entry keystore.Entry + // NewMessage is false if InsertChannelMessage found the message already existed + // (packet_hash is unique per message) -- e.g. re-running the backfill twice. + NewMessage bool +} + +// DecryptGroupText attempts to decrypt a GRP_TXT payload against keys and, on success, +// upserts the channel and inserts the decrypted message. Shared by the live per-packet +// ingest path (side_effects.go) and BackfillChannelMessages below, so the two can't drift. +// +// A nil result with a nil error means no known key decrypted this packet -- the caller +// decides what to do with that. The live path falls back to UpsertChannelHashOnly itself; +// BackfillChannelMessages, which only scans packets already recorded that way, just leaves +// them as they are and tries again next boot. +func DecryptGroupText(ctx context.Context, db DB, keys ChannelKeyStore, packetHash, rawPayload []byte) (*DecryptGroupTextResult, error) { + grpTxt, err := meshcore.GroupTextFromBytes(rawPayload) + if err != nil { + return nil, err + } + channelHashBytes := []byte{grpTxt.ChannelHash} + + var payload *meshcore.GroupTextPayload + var usedEntry keystore.Entry + for _, entry := range keys.GetKey(channelHashBytes) { + if p, err := grpTxt.DecryptStruct(entry.Key); err == nil { + payload = p + usedEntry = entry + break + } + } + if payload == nil { + return nil, nil + } + + channelID, err := db.UpsertChannel(ctx, channelHashBytes, usedEntry.Fingerprint, usedEntry.Name, usedEntry.Hashtag) + if err != nil { + return nil, fmt.Errorf("upsert channel: %w", err) + } + params := InsertChannelMessageParams{ + ChannelID: channelID, + PacketHash: packetHash, + SenderName: strings.ReplaceAll(strings.ToValidUTF8(payload.Sender, "\uFFFD"), "\x00", ""), + SentAt: time.Unix(int64(payload.Timestamp), 0), + Content: strings.ReplaceAll(strings.ToValidUTF8(payload.Text, "\uFFFD"), "\x00", ""), + } + newMsg, err := db.InsertChannelMessage(ctx, params) + if err != nil { + return nil, fmt.Errorf("insert channel message: %w", err) + } + if newMsg { + // Non-fatal: the message is stored either way, so just log and continue -- matches + // the live ingest path's existing behavior of not treating this as a hard failure. + if err := db.SetPacketDecrypted(ctx, packetHash); err != nil { + log.Printf("ingest: failed to set packet decrypted for %s: %v", hex.EncodeToString(packetHash), err) + } + } + return &DecryptGroupTextResult{ + Payload: payload, + ChannelID: channelID, + ChannelHash: channelHashBytes, + Entry: usedEntry, + NewMessage: newMsg, + }, nil +} + +// BackfillChannelMessages scans packets already stored but never successfully decrypted and +// retries decryption against the current keystore. Meant to run once at boot, after the +// keystore is built from config: without this, a channel added to the config after its +// packets already arrived would sit undecrypted in the database forever, since nothing else +// ever re-processes old packets when a new key shows up. +// +// Returns the number of packets newly decrypted. Deliberately does not broadcast live WS +// channelMessage events for backfilled messages -- those are historical, not new activity, +// and broadcasting a burst of them on every boot would look like a flood of new messages to +// anyone connected at startup. +func BackfillChannelMessages(ctx context.Context, db DB, keys ChannelKeyStore) (int, error) { + packets, err := db.ListUndecryptedGroupTextPackets(ctx) + if err != nil { + return 0, err + } + decrypted := 0 + for _, p := range packets { + result, err := DecryptGroupText(ctx, db, keys, p.PacketHash, p.RawPayload) + if err != nil { + log.Printf("ingest: backfill: decrypt failed for packet %s: %v", hex.EncodeToString(p.PacketHash), err) + continue + } + if result != nil && result.NewMessage { + decrypted++ + } + } + return decrypted, nil +} diff --git a/internal/ingest/backfill_test.go b/internal/ingest/backfill_test.go new file mode 100644 index 0000000..748332e --- /dev/null +++ b/internal/ingest/backfill_test.go @@ -0,0 +1,148 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later + +package ingest + +import ( + "context" + "testing" + + "github.com/MeshCore-Beacon/beacon-server/internal/keystore" + "github.com/meshcore-go/meshcore-go" +) + +// encryptedGroupText returns the raw GRP_TXT payload bytes for a message encrypted under psk +// -- a real round trip through meshcore-go's own Encrypt, not a hand-rolled fixture. +func encryptedGroupText(t *testing.T, channelHash byte, psk []byte, sender, text string) []byte { + t.Helper() + grpTxt, err := (&meshcore.GroupTextPayload{ + Timestamp: 1000, + Sender: sender, + Text: text, + }).Encrypt(channelHash, psk) + if err != nil { + t.Fatalf("encrypt group text: %v", err) + } + raw, err := grpTxt.ToBytes() + if err != nil { + t.Fatalf("group text to bytes: %v", err) + } + return raw +} + +func TestDecryptGroupText_Success(t *testing.T) { + db := &stubDB{insertChannelMessageResult: true} + psk := make([]byte, 16) + channelHash := byte(0x11) + keys := &mapKeys{entries: map[byte][]keystore.Entry{ + channelHash: {{Key: psk, Fingerprint: []byte{0xAA}, Name: "Public", Hashtag: "public"}}, + }} + raw := encryptedGroupText(t, channelHash, psk, "ded", "hello") + + result, err := DecryptGroupText(context.Background(), db, keys, []byte{0x01}, raw) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if result == nil { + t.Fatal("expected a non-nil result for a decryptable packet") + } + if result.Payload.Sender != "ded" || result.Payload.Text != "hello" { + t.Errorf("expected decrypted sender=ded text=hello, got sender=%s text=%s", result.Payload.Sender, result.Payload.Text) + } + if !result.NewMessage { + t.Error("expected NewMessage true when InsertChannelMessage reports a new insert") + } + if db.upsertChannelCalls != 1 { + t.Errorf("expected UpsertChannel to be called once, got %d", db.upsertChannelCalls) + } +} + +func TestDecryptGroupText_NoMatchingKey(t *testing.T) { + db := &stubDB{} + channelHash := byte(0x22) + // Encrypted under a key the store doesn't have -- keys is otherwise empty for this hash. + raw := encryptedGroupText(t, channelHash, make([]byte, 16), "ded", "hello") + keys := &mapKeys{entries: map[byte][]keystore.Entry{}} + + result, err := DecryptGroupText(context.Background(), db, keys, []byte{0x02}, raw) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if result != nil { + t.Errorf("expected a nil result when no known key decrypts the packet, got %+v", result) + } + if db.upsertChannelCalls != 0 { + t.Errorf("expected UpsertChannel NOT to be called, got %d calls", db.upsertChannelCalls) + } +} + +func TestDecryptGroupText_WrongKeyForHash(t *testing.T) { + db := &stubDB{} + channelHash := byte(0x33) + raw := encryptedGroupText(t, channelHash, make([]byte, 16), "ded", "hello") + // A key IS registered for this hash, but it's the wrong one (1-byte hash collisions are + // expected; DecryptStruct's MAC check is what actually distinguishes them). + wrongKey := make([]byte, 16) + wrongKey[0] = 0xFF + keys := &mapKeys{entries: map[byte][]keystore.Entry{ + channelHash: {{Key: wrongKey, Fingerprint: []byte{0xBB}}}, + }} + + result, err := DecryptGroupText(context.Background(), db, keys, []byte{0x03}, raw) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if result != nil { + t.Errorf("expected a nil result for a MAC mismatch, got %+v", result) + } +} + +func TestDecryptGroupText_MalformedPayload(t *testing.T) { + db := &stubDB{} + keys := &mapKeys{} + _, err := DecryptGroupText(context.Background(), db, keys, []byte{0x04}, []byte{}) + if err == nil { + t.Fatal("expected an error for an empty/malformed payload") + } +} + +func TestBackfillChannelMessages_DecryptsAndCounts(t *testing.T) { + psk := make([]byte, 16) + channelHash := byte(0x44) + decryptable := encryptedGroupText(t, channelHash, psk, "ded", "hi") + stillUnknown := encryptedGroupText(t, byte(0x55), make([]byte, 16), "someone", "bye") + + db := &stubDB{ + insertChannelMessageResult: true, + undecryptedPackets: []UndecryptedPacket{ + {PacketHash: []byte{0x01}, RawPayload: decryptable}, + {PacketHash: []byte{0x02}, RawPayload: stillUnknown}, + }, + } + keys := &mapKeys{entries: map[byte][]keystore.Entry{ + channelHash: {{Key: psk, Fingerprint: []byte{0xAA}, Name: "Public"}}, + }} + + n, err := BackfillChannelMessages(context.Background(), db, keys) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if n != 1 { + t.Errorf("expected 1 packet newly decrypted (the other has no known key yet), got %d", n) + } + if db.upsertChannelCalls != 1 { + t.Errorf("expected UpsertChannel called once (only for the decryptable packet), got %d", db.upsertChannelCalls) + } +} + +func TestBackfillChannelMessages_NoUndecryptedPackets(t *testing.T) { + db := &stubDB{} + keys := &mapKeys{} + n, err := BackfillChannelMessages(context.Background(), db, keys) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if n != 0 { + t.Errorf("expected 0, got %d", n) + } +} diff --git a/internal/ingest/ingest.go b/internal/ingest/ingest.go index 0fd5215..ff907a3 100644 --- a/internal/ingest/ingest.go +++ b/internal/ingest/ingest.go @@ -151,6 +151,10 @@ type DB interface { // but can be safely ignored since unknown-key channels have no messages. UpsertChannelHashOnly(ctx context.Context, channelHash []byte) (int, error) + // ListUndecryptedGroupTextPackets returns GRP_TXT packets never successfully decrypted -- + // used by BackfillChannelMessages to retry them against the current keystore at boot. + ListUndecryptedGroupTextPackets(ctx context.Context) ([]UndecryptedPacket, error) + // UpsertChannelIATA upserts a channel_iatas row. UpsertChannelIATA(ctx context.Context, channelHash []byte, iata string, heardAt time.Time) error diff --git a/internal/ingest/ingest_test.go b/internal/ingest/ingest_test.go index fbde760..b01b9cc 100644 --- a/internal/ingest/ingest_test.go +++ b/internal/ingest/ingest_test.go @@ -180,6 +180,8 @@ type stubDB struct { upsertChannelIATACalls int upsertTraceIATACalls int observationInserted bool + insertChannelMessageResult bool // configurable return for InsertChannelMessage; default false + undecryptedPackets []UndecryptedPacket } type setCapabilityCall struct { @@ -225,7 +227,7 @@ func (s *stubDB) GetNodesByIDs(_ context.Context, _ []uuid.UUID) (map[uuid.UUID] } func (s *stubDB) InsertChannelMessage(_ context.Context, _ InsertChannelMessageParams) (bool, error) { - return false, nil + return s.insertChannelMessageResult, nil } func (s *stubDB) UpdateObserverStatus(_ context.Context, _ UpdateObserverStatusParams) (uuid.UUID, error) { @@ -260,6 +262,10 @@ func (s *stubDB) UpsertChannelHashOnly(_ context.Context, _ []byte) (int, error) return 0, nil } +func (s *stubDB) ListUndecryptedGroupTextPackets(_ context.Context) ([]UndecryptedPacket, error) { + return s.undecryptedPackets, nil +} + func (s *stubDB) UpsertChannelIATA(_ context.Context, _ []byte, _ string, _ time.Time) error { s.upsertChannelIATACalls++ return nil diff --git a/internal/ingest/side_effects.go b/internal/ingest/side_effects.go index eff7359..9a78ab1 100644 --- a/internal/ingest/side_effects.go +++ b/internal/ingest/side_effects.go @@ -13,7 +13,6 @@ import ( "github.com/MeshCore-Beacon/beacon-server/internal/api" "github.com/MeshCore-Beacon/beacon-server/internal/hub" - "github.com/MeshCore-Beacon/beacon-server/internal/keystore" "github.com/meshcore-go/meshcore-go" ) @@ -185,60 +184,28 @@ func (w *Worker) handlePayloadTypeSideEffects(ctx context.Context, packet *meshc } channelHashBytes := []byte{grpTxt.ChannelHash} - // Try each known key entry for this hash. - entries := w.keys.GetKey(channelHashBytes) - if len(entries) == 0 { - // channel key unknown; record a hash-only row and store the - // message as an encrypted blob only. - _, _ = w.db.UpsertChannelHashOnly(ctx, channelHashBytes) + result, err := DecryptGroupText(ctx, w.db, w.keys, packetHash, packet.Payload) + if err != nil { + log.Printf("ingest[%s]: decrypt group text failed: %v", w.cfg.BrokerName, err) return } - var payload *meshcore.GroupTextPayload - var usedEntry keystore.Entry - for _, entry := range entries { - if p, err := grpTxt.DecryptStruct(entry.Key); err == nil { - payload = p - usedEntry = entry - break - } - } - if payload == nil { - // none of the known keys worked for this hash; record a - // hash-only row so the channel is still visible. + if result == nil { + // none of the known keys worked for this hash (or none are known at all); + // record a hash-only row so the channel is still visible. If a matching key + // gets added to the config later, BackfillChannelMessages retries this packet + // at the next boot. _, _ = w.db.UpsertChannelHashOnly(ctx, channelHashBytes) return } - // Upsert the keyed channel row — messages are associated with this row. - channelID, err := w.db.UpsertChannel(ctx, channelHashBytes, usedEntry.Fingerprint, usedEntry.Name, usedEntry.Hashtag) - if err != nil { - log.Printf("ingest[%s]: db: upsert keyed channel failed: %v", w.cfg.BrokerName, err) - return - } - params := InsertChannelMessageParams{ - ChannelID: channelID, - PacketHash: packetHash[:], - SenderName: strings.ReplaceAll(strings.ToValidUTF8(payload.Sender, "\uFFFD"), "\x00", ""), - SentAt: time.Unix(int64(payload.Timestamp), 0), - Content: strings.ReplaceAll(strings.ToValidUTF8(payload.Text, "\uFFFD"), "\x00", ""), - } - newMsg, err := w.db.InsertChannelMessage(ctx, params) - if err != nil { - log.Printf("ingest[%s]: db: insert channel message failed: %v", w.cfg.BrokerName, err) - return - } - - if newMsg { - if err := w.db.SetPacketDecrypted(ctx, packetHash[:]); err != nil { - log.Printf("ingest[%s]: failed to set packet decrypted: %v", w.cfg.BrokerName, err) - } + if result.NewMessage { evt := channelMessageEvent{ - ChannelID: channelID, + ChannelID: result.ChannelID, ChannelHash: hex.EncodeToString(channelHashBytes), PacketHash: hex.EncodeToString(packetHash), - SenderName: strings.ReplaceAll(strings.ToValidUTF8(payload.Sender, "\uFFFD"), "\x00", ""), - Content: strings.ReplaceAll(strings.ToValidUTF8(payload.Text, "\uFFFD"), "\x00", ""), - SentAt: time.Unix(int64(payload.Timestamp), 0).UnixMilli(), + SenderName: strings.ReplaceAll(strings.ToValidUTF8(result.Payload.Sender, "\uFFFD"), "\x00", ""), + Content: strings.ReplaceAll(strings.ToValidUTF8(result.Payload.Text, "\uFFFD"), "\x00", ""), + SentAt: time.Unix(int64(result.Payload.Timestamp), 0).UnixMilli(), } w.broadcast(hub.EventChannelMessage, iata, 0, fmt.Sprintf("%02x", grpTxt.ChannelHash), evt) }