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) + } +}