From 5b177d6fc9752da9e925a6f5c19684d06d25cd7f Mon Sep 17 00:00:00 2001 From: MrAlders0n Date: Wed, 30 Sep 2026 18:29:22 -0400 Subject: [PATCH] fix(ingest): strip NUL bytes from observer status An observer sending NUL-padded status strings had every status update rejected by Postgres, freezing its status and telemetry. Strip them from the text fields and the stored JSON, and warn with the fields affected. Also drop the timestamp-clamp, empty-timestamp and malformed-IATA logs to debug; they're expected and outnumbered inserted packets several times over. --- internal/ingest/ingest.go | 2 +- internal/ingest/logging_test.go | 13 +++---- internal/ingest/packet.go | 4 +-- internal/ingest/status.go | 62 +++++++++++++++++++++++++++++++++ internal/ingest/status_test.go | 39 +++++++++++++++++++++ 5 files changed, 111 insertions(+), 9 deletions(-) diff --git a/internal/ingest/ingest.go b/internal/ingest/ingest.go index 523a1fb..59b9b4e 100644 --- a/internal/ingest/ingest.go +++ b/internal/ingest/ingest.go @@ -374,7 +374,7 @@ func (w *Worker) handleMessageContext(parent context.Context, msg mqtt.Message) // iata_codes.iata is CHAR(3); anything else would fail the DB insert // downstream, so reject malformed topic segments here instead. if !isValidIATA(iata) { - w.log.Warn("dropped packet with malformed IATA", "iata", iata, "topic", msg.Topic()) + w.log.Debug("dropped packet with malformed IATA", "iata", iata, "topic", msg.Topic()) return } diff --git a/internal/ingest/logging_test.go b/internal/ingest/logging_test.go index fa8bcc4..3f9d343 100644 --- a/internal/ingest/logging_test.go +++ b/internal/ingest/logging_test.go @@ -22,14 +22,15 @@ func (m malformedTopicMessage) MessageID() uint16 { return 0 } func (m malformedTopicMessage) Payload() []byte { return nil } func (m malformedTopicMessage) Ack() {} -func TestMalformedIATAWarnNamesSource(t *testing.T) { +func TestMalformedIATALogNamesSource(t *testing.T) { for _, format := range []string{"json", "text"} { for _, iata := range []string{"XX", "XX\r\nforged=true"} { t.Run(format+"/"+strconv.Quote(iata), func(t *testing.T) { var output bytes.Buffer - var handler slog.Handler = slog.NewJSONHandler(&output, nil) + opts := &slog.HandlerOptions{Level: slog.LevelDebug} + var handler slog.Handler = slog.NewJSONHandler(&output, opts) if format == "text" { - handler = slog.NewTextHandler(&output, nil) + handler = slog.NewTextHandler(&output, opts) } previous := slog.Default() slog.SetDefault(slog.New(handler)) @@ -38,7 +39,7 @@ func TestMalformedIATAWarnNamesSource(t *testing.T) { topic := "meshcore/" + iata + "/0011223344556677/packets" worker.handleMessage(malformedTopicMessage{topic: topic}) if bytes.Count(output.Bytes(), []byte("\n")) != 1 || bytes.ContainsRune(output.Bytes(), '\r') { - t.Fatalf("warning did not remain one escaped record: %q", output.String()) + t.Fatalf("log did not remain one escaped record: %q", output.String()) } if format == "json" { var record map[string]any @@ -46,7 +47,7 @@ func TestMalformedIATAWarnNamesSource(t *testing.T) { t.Fatal(err) } if record["iata"] != iata || record["topic"] != topic || record["msg"] != "dropped packet with malformed IATA" { - t.Fatalf("warning lost its source: %v", record) + t.Fatalf("log lost its source: %v", record) } } else { expectedIATA, expectedTopic := iata, topic @@ -54,7 +55,7 @@ func TestMalformedIATAWarnNamesSource(t *testing.T) { expectedIATA, expectedTopic = strconv.Quote(iata), strconv.Quote(topic) } if !strings.Contains(output.String(), "iata="+expectedIATA) || !strings.Contains(output.String(), "topic="+expectedTopic) { - t.Fatalf("warning lost its source: %q", output.String()) + t.Fatalf("log lost its source: %q", output.String()) } } }) diff --git a/internal/ingest/packet.go b/internal/ingest/packet.go index ad5336e..7bc4ccf 100644 --- a/internal/ingest/packet.go +++ b/internal/ingest/packet.go @@ -732,14 +732,14 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [ } heardAt, err := parseHeardAt(envelope.Timestamp) if err != nil { - w.log.Warn(fmt.Sprintf("failed to parse timestamp %q", envelope.Timestamp), "error", err) + w.log.Debug(fmt.Sprintf("failed to parse timestamp %q", envelope.Timestamp), "error", err) heardAt = time.Now().UTC() } else { // clamp to server time if offset is suspicious (> 30 min drift) now := time.Now().UTC() diff := heardAt.UTC().Sub(now) if diff > 30*time.Minute || diff < -30*time.Minute { - w.log.Warn(fmt.Sprintf("clamping suspicious timestamp %s (diff %v) for pubkey %s", envelope.Timestamp, diff, pubkeyHex[:8])) + w.log.Debug(fmt.Sprintf("clamping suspicious timestamp %s (diff %v) for pubkey %s", envelope.Timestamp, diff, pubkeyHex[:8])) heardAt = now } } diff --git a/internal/ingest/status.go b/internal/ingest/status.go index 3e2ded6..9fe6a82 100644 --- a/internal/ingest/status.go +++ b/internal/ingest/status.go @@ -4,10 +4,12 @@ package ingest import ( + "bytes" "context" "encoding/hex" "encoding/json" "fmt" + "slices" "strconv" "strings" "time" @@ -70,6 +72,57 @@ func (s statusStats) usable() bool { return s.NoiseFloor != 0 || s.TxAirSecs != 0 || s.RxAirSecs != 0 } +// stripNULs removes NUL bytes, which Postgres text columns reject, and returns the +// names of the fields that had any. +func stripNULs(fields map[string]*string) []string { + var dirty []string + for name, s := range fields { + if strings.Contains(*s, "\x00") { + *s = strings.ReplaceAll(*s, "\x00", "") + dirty = append(dirty, name) + } + } + slices.Sort(dirty) + return dirty +} + +// stripJSONNULs drops \u0000 from a JSON document's strings, which jsonb rejects. +// Payloads without one are returned untouched. +func stripJSONNULs(raw []byte) ([]byte, bool) { + if !bytes.Contains(raw, []byte(`\u0000`)) { + return raw, false + } + dec := json.NewDecoder(bytes.NewReader(raw)) + dec.UseNumber() + var v any + if err := dec.Decode(&v); err != nil { + return raw, false + } + out, err := json.Marshal(dropNULs(v)) + if err != nil { + return raw, false + } + return out, true +} + +func dropNULs(v any) any { + switch v := v.(type) { + case string: + return strings.ReplaceAll(v, "\x00", "") + case []any: + for i := range v { + v[i] = dropNULs(v[i]) + } + case map[string]any: + out := make(map[string]any, len(v)) + for k, e := range v { + out[strings.ReplaceAll(k, "\x00", "")] = dropNULs(e) + } + return out + } + return v +} + // handleStatus processes a /status message and fans out an observerStatus event. func (w *Worker) handleStatus(ctx context.Context, pubkeyHex string, raw []byte) { var envelope struct { @@ -90,6 +143,15 @@ func (w *Worker) handleStatus(ctx context.Context, pubkeyHex string, raw []byte) w.log.Warn(fmt.Sprintf("invalid pubkey hex in status from %s", pubkeyHex), "error", err) return } + dirty := stripNULs(map[string]*string{ + "source": &envelope.ObserverType, "client_version": &envelope.SoftwareVersion, + "model": &envelope.HardwareModel, "firmware_version": &envelope.FirmwareVersion, + "origin": &envelope.DisplayName, "radio": &envelope.RadioString, + }) + raw, rawDirty := stripJSONNULs(raw) + if len(dirty) > 0 || rawDirty { + w.log.Warn(fmt.Sprintf("stripped NUL bytes from status from %s", pubkeyHex), "fields", dirty, "metadata", rawDirty) + } id, _, err := w.db.UpsertObserver(ctx, pubkey) if err != nil { w.log.Error(fmt.Sprintf("db: upsert observer failed in status from %s", pubkeyHex), "error", err) diff --git a/internal/ingest/status_test.go b/internal/ingest/status_test.go index d61713b..6d8876c 100644 --- a/internal/ingest/status_test.go +++ b/internal/ingest/status_test.go @@ -33,3 +33,42 @@ func TestStatusStatsUsable(t *testing.T) { }) } } + +func TestStripNULs(t *testing.T) { + fw := "v1.17.1-d929643\x00\x00\x00" + name := "Canadaverse\x00 RemoteTerm" + model := "Heltec Wireless Paper" + got := stripNULs(map[string]*string{"firmware_version": &fw, "origin": &name, "model": &model}) + if fw != "v1.17.1-d929643" || name != "Canadaverse RemoteTerm" || model != "Heltec Wireless Paper" { + t.Errorf("stripped to %q %q %q", fw, name, model) + } + if len(got) != 2 || got[0] != "firmware_version" || got[1] != "origin" { + t.Errorf("reported fields %v, want [firmware_version origin]", got) + } +} + +func TestStripJSONNULs(t *testing.T) { + clean := []byte(`{"origin":"a","stats":{"uptime_secs":12}}`) + if out, changed := stripJSONNULs(clean); changed || string(out) != string(clean) { + t.Errorf("clean payload rewritten: %s", out) + } + + dirty := []byte(`{"firmware_version":"v1.17\u0000\u0000","extra":{"list":["x\u0000y"],"n":1.50}}`) + out, changed := stripJSONNULs(dirty) + if !changed { + t.Fatal("expected payload to be cleaned") + } + var v struct { + FirmwareVersion string `json:"firmware_version"` + Extra struct { + List []string `json:"list"` + N json.Number `json:"n"` + } `json:"extra"` + } + if err := json.Unmarshal(out, &v); err != nil { + t.Fatal(err) + } + if v.FirmwareVersion != "v1.17" || v.Extra.List[0] != "xy" || v.Extra.N != "1.50" { + t.Errorf("cleaned to %s", out) + } +}