From 4732887b0ef87bd416be6555f1e5cf59a573fe23 Mon Sep 17 00:00:00 2001 From: dborup Date: Sun, 4 Oct 2026 08:23:57 +0000 Subject: [PATCH 1/6] test(brokerurl): reproduce secret parts, decoded forms, overlap and short values (#159) Table tests for the examples in #159: a user name or password quoted on its own, a decoded password (p%40ss -> p@ss), a query value on its own, overlapping secrets (abcd + cdef on abcdef), and 1-2 character values that must leave unrelated text alone. Plus the watchdog force-reconnect log line from the issue comment, with user:pass@ and ?token= in the error. Red on master (MaskSecrets and the secrets argument of buildForceReconnectFn do not exist yet). Co-Authored-By: Claude Opus 5.5 --- cmd/ingestor/mqtt_credentials_159_test.go | 50 +++++++++++++++ internal/brokerurl/brokerurl_test.go | 76 ++++++++++++++++++++++- 2 files changed, 123 insertions(+), 3 deletions(-) create mode 100644 cmd/ingestor/mqtt_credentials_159_test.go diff --git a/cmd/ingestor/mqtt_credentials_159_test.go b/cmd/ingestor/mqtt_credentials_159_test.go new file mode 100644 index 000000000..6040044df --- /dev/null +++ b/cmd/ingestor/mqtt_credentials_159_test.go @@ -0,0 +1,50 @@ +package main + +import ( + "errors" + "strings" + "testing" +) + +// #159: errForLog masks the parts of the broker URL's credentials, raw and +// decoded, wherever they appear in an error, and overlapping secrets +// leave no residue. +func TestErrForLogMasksSecretParts_159(t *testing.T) { + for _, c := range []struct{ broker, in, want string }{ + {"tcp://dev-user:hunter2@host", "bad password hunter2 for dev-user", "bad password **** for ****"}, + {"tcp://dev-user:p%40ss@host", "auth p@ss rejected", "auth **** rejected"}, + {"wss://host/mqtt?token=abc123", "token abc123 expired", "token **** expired"}, + // short values are left alone in free text; the URL itself is + // still masked by MaskText + {"tcp://u:pw@host", "uptime 1h, pw ok: tcp://u:pw@host refused", "uptime 1h, pw ok: tcp://****@host refused"}, + } { + got := errForLog(errors.New(c.in), mqttSourceSecrets(MQTTSource{Broker: c.broker})...) + if got != c.want { + t.Errorf("errForLog(%q) for %q = %q, want %q", c.in, c.broker, got, c.want) + } + } + // a configured password overlapping the URL's user name + src := MQTTSource{Broker: "tcp://abcd@host", Password: "cdef"} + if got := errForLog(errors.New("auth abcdef"), mqttSourceSecrets(src)...); got != "auth ****" { + t.Errorf("overlap: %q, want %q", got, "auth ****") + } +} + +// #159 (comment): the watchdog's forced reconnect logs Connect()'s error +// with the source's secrets masked, like every other connect path. +func TestBuildForceReconnectFnMasksConnectError_159(t *testing.T) { + src := MQTTSource{Broker: "wss://" + credUser + ":" + credPass + "@host/mqtt?token=abc123"} + c := &fakeClient{connectErr: errors.New(`dial "` + src.Broker + `": refused; token abc123 rejected for ` + credUser)} + buf := captureLog118(t) + buildForceReconnectFn(c, "feed", mqttSourceSecrets(src)...)() + out := buf.String() + const want = `MQTT [feed] WATCHDOG force-reconnect Connect() failed: dial "wss://****@host/mqtt refused; token **** rejected for ****` + if !strings.Contains(out, want) { + t.Errorf("log %q, want it to contain %q", out, want) + } + for _, bad := range []string{credUser, credPass, "abc123", "token="} { + if strings.Contains(out, bad) { + t.Errorf("log leaks %q: %s", bad, out) + } + } +} diff --git a/internal/brokerurl/brokerurl_test.go b/internal/brokerurl/brokerurl_test.go index 3656f3f55..41439946e 100644 --- a/internal/brokerurl/brokerurl_test.go +++ b/internal/brokerurl/brokerurl_test.go @@ -59,9 +59,15 @@ func TestSecrets(t *testing.T) { in string want []string }{ - {"tcp://dev-user:secret@host:1883", []string{"dev-user:secret"}}, - {"wss://host/mqtt?token=abc#frag", []string{"token=abc", "frag"}}, - {"dev-user:1234?abc@host?x=1", []string{"dev-user:1234?abc", "x=1"}}, + {"tcp://dev-user:secret@host:1883", []string{"dev-user:secret", "dev-user", "secret"}}, + {"wss://host/mqtt?token=abc#frag", []string{"token=abc", "abc", "frag"}}, + {"dev-user:1234?abc@host?x=1", []string{"dev-user:1234?abc", "dev-user", "1234?abc", "x=1", "1"}}, + // #159: each part raw and URL-decoded, without duplicates + {"tcp://dev-user:p%40ss@host", []string{"dev-user:p%40ss", "dev-user:p@ss", "dev-user", "p%40ss", "p@ss"}}, + {"wss://host/?a=x+y%21&b&c=", []string{"a=x+y%21&b&c=", "a=x+y!&b&c=", "a=x y!&b&c=", "x+y%21", "x+y!", "x y!", "b"}}, + {"wss://host/#access_token=xyz&state=1", []string{"access_token=xyz&state=1", "xyz", "1"}}, + {"tcp://:pw@host", []string{":pw", "pw"}}, + {"tcp://u%zz@host", []string{"u%zz"}}, // a bad escape has no decoded form {"tcp://@host", nil}, {"wss://host/mqtt?#", nil}, {"mqtt://host:1883", nil}, @@ -108,3 +114,67 @@ func TestMaskIsIdempotent(t *testing.T) { } } } + +// #159: MaskSecrets masks every part of a broker URL's credentials that +// appears in free text, raw or decoded, in one pass over merged match +// intervals, so overlapping secrets leave no residue, and it leaves +// values shorter than MinSecretLen alone. +func TestMaskSecrets_159(t *testing.T) { + for _, c := range []struct { + name, in string + secrets []string + want string + }{ + {"user and password alone", "bad password hunter2 for dev-user", + Secrets("tcp://dev-user:hunter2@host"), "bad password **** for ****"}, + {"decoded password", "auth p@ss rejected", + Secrets("tcp://dev-user:p%40ss@host"), "auth **** rejected"}, + {"raw password", "auth p%40ss rejected", + Secrets("tcp://dev-user:p%40ss@host"), "auth **** rejected"}, + {"query value alone", "token abc123 expired", + Secrets("wss://host/mqtt?token=abc123"), "token **** expired"}, + {"query value decoded", "token a b/c! expired", + Secrets("wss://host/mqtt?token=a+b%2Fc%21"), "token **** expired"}, + {"fragment value", "bad token xyz789", + Secrets("wss://host/#access_token=xyz789"), "bad token ****"}, + {"overlapping secrets", "abcdef", []string{"abcd", "cdef"}, "****"}, + {"overlap inside text", "x abcdef y", []string{"cdef", "abcd"}, "x **** y"}, + {"one secret contains another", "user:pass", []string{"user", "user:pass", "pass"}, "****"}, + {"a secret overlapping itself", "aaaa", []string{"aaa"}, "****"}, + {"adjacent matches", "abcdef", []string{"abc", "def"}, "****"}, + {"separate matches", "abc-def", []string{"abc", "def"}, "****-****"}, + {"short values untouched", "uptime 1h, pw ok, queue u2", + Secrets("tcp://u:pw@host?x=1"), "uptime 1h, pw ok, queue u2"}, + {"short values in a longer secret", "auth u:pw failed", + Secrets("tcp://u:pw@host"), "auth **** failed"}, + {"length counts runes", "pæs æø", []string{"pæs", "æø"}, "**** æø"}, + {"no secrets", "EOF", nil, "EOF"}, + {"empty secret", "EOF", []string{""}, "EOF"}, + {"marker is not re-scanned", "x abc y", []string{"abc", "***"}, "x **** y"}, + } { + got := MaskSecrets(c.in, c.secrets...) + if got != c.want { + t.Errorf("%s: MaskSecrets(%q) = %q, want %q", c.name, c.in, got, c.want) + } + } +} + +// What a caller logs, MaskText(MaskSecrets(...)), keeps nothing of a +// decoded password with an '@' in it: MaskText alone would read "p@ss" as +// user-info and show "ss". +func TestMaskSecretsThenMaskTextLeavesNoResidue_159(t *testing.T) { + got := MaskText(MaskSecrets("auth p@ss rejected", Secrets("tcp://dev-user:p%40ss@host")...)) + if got != "auth **** rejected" { + t.Errorf("got %q, want %q", got, "auth **** rejected") + } +} + +// No change to Mask: what MaskSecrets adds is for free text only, and a +// full URL masked by both is still masked as before. +func TestMaskSecretsKeepsMaskOutput_159(t *testing.T) { + for _, in := range []string{"tcp://dev-user:secret@host:1883", "wss://host/mqtt?token=abc", "tcp://dev-user:p%40ss@host"} { + if got := MaskText(MaskSecrets(Mask(in), Secrets(in)...)); got != Mask(in) { + t.Errorf("%q: %q, want %q", in, got, Mask(in)) + } + } +} From be93b8918c290c5851f9ef41cdf39edf19671fd7 Mon Sep 17 00:00:00 2001 From: dborup Date: Sun, 4 Oct 2026 09:06:17 +0000 Subject: [PATCH 2/6] fix(brokerurl): mask user, password, query values and decoded forms without residue (#159) Secrets now also returns the user name and password, each query and fragment value, each raw and %-decoded (path and query forms), without duplicates or empty values. The new MaskSecrets merges the match intervals of all secrets (overlapping matches of one secret included) and replaces each merged interval once, so overlapping secrets leave no residue, and skips secrets shorter than MinSecretLen (3 runes) so a 1-2 character user name or password no longer masks unrelated text. errForLog is now MaskText(MaskSecrets(...)). The watchdog's forced reconnect logs Connect()'s error through errForLog with the source's secrets, like every other connect path (issue comment). Mask and MaskText are unchanged. Co-Authored-By: Claude Opus 5.5 --- cmd/ingestor/main.go | 12 ++-- cmd/ingestor/mqtt_client_id.go | 26 +++---- internal/brokerurl/brokerurl.go | 117 +++++++++++++++++++++++++++++--- 3 files changed, 127 insertions(+), 28 deletions(-) diff --git a/cmd/ingestor/main.go b/cmd/ingestor/main.go index 9f98b23b0..77a0943b9 100644 --- a/cmd/ingestor/main.go +++ b/cmd/ingestor/main.go @@ -139,6 +139,7 @@ func main() { for i, source := range sources { tag := tags[i] opts, status, liveness := prepareMQTTSource(source, tag) + secrets := mqttSourceSecrets(source) connectTimeout := source.ConnectTimeoutOrDefault() log.Printf("MQTT [%s] connect timeout: %ds", tag, connectTimeout) @@ -169,7 +170,7 @@ func main() { // Captured-by-value `client` is the same pointer used everywhere // else for this source. See buildForceReconnectFn for why this is // NOT simply "Disconnect(250) then Connect()". - liveness.ForceReconnectFn = buildForceReconnectFn(client, tag) + liveness.ForceReconnectFn = buildForceReconnectFn(client, tag, secrets...) // PR #1216 r2 item 3: tag collisions used to log.Fatalf, which // killed the entire ingestor over one config typo and recreated // the #1212 total-ingest-stop class this PR exists to prevent. @@ -188,7 +189,7 @@ func main() { continue } if token.Error() != nil { - log.Printf("MQTT [%s] connection failed (non-fatal): %s", tag, errForLog(token.Error(), mqttSourceSecrets(source)...)) + log.Printf("MQTT [%s] connection failed (non-fatal): %s", tag, errForLog(token.Error(), secrets...)) // BL1 fix: Disconnect to stop Paho's internal retry goroutines. // With ConnectRetry=true, Connect() spawns background goroutines // that leak if the client is simply discarded. @@ -571,7 +572,10 @@ func buildMQTTOpts(source MQTTSource) *mqtt.ClientOptions { // call Disconnect; Connect() alone is a safe no-op per paho when a retry is // already under way, and properly starts a fresh attempt on the rare // occasion status has actually settled to disconnected). -func buildForceReconnectFn(client mqtt.Client, tag string) func() { +// +// Connect()'s error is logged with secrets (mqttSourceSecrets) masked +// (errForLog): paho's errors may quote the broker URL. +func buildForceReconnectFn(client mqtt.Client, tag string, secrets ...string) func() { return func() { if client.IsConnectionOpen() { client.Disconnect(250) @@ -582,7 +586,7 @@ func buildForceReconnectFn(client mqtt.Client, tag string) func() { // fresh connection attempt leaves the token pending in the // background, and we must not block this call on that. if token := client.Connect(); token.Error() != nil { - log.Printf("MQTT [%s] WATCHDOG force-reconnect Connect() failed: %v", tag, token.Error()) + log.Printf("MQTT [%s] WATCHDOG force-reconnect Connect() failed: %s", tag, errForLog(token.Error(), secrets...)) } } } diff --git a/cmd/ingestor/mqtt_client_id.go b/cmd/ingestor/mqtt_client_id.go index 887ed14b6..307909722 100644 --- a/cmd/ingestor/mqtt_client_id.go +++ b/cmd/ingestor/mqtt_client_id.go @@ -6,7 +6,6 @@ import ( "encoding/hex" "io" "net/url" - "sort" "strconv" "strings" "sync/atomic" @@ -165,7 +164,8 @@ func brokerForLog(broker string) string { // mqttSourceSecrets returns the non-empty secrets of a source: its // password and user name, and the user-info, query and fragment of its -// broker URL as configured (brokerurl.Secrets). +// broker URL as configured, with their parts, raw and decoded +// (brokerurl.Secrets). func mqttSourceSecrets(source MQTTSource) []string { var out []string for _, v := range append([]string{source.Password, source.Username}, brokerurl.Secrets(source.Broker)...) { @@ -176,22 +176,16 @@ func mqttSourceSecrets(source MQTTSource) []string { return out } -// errForLog is err's text with each of secrets (mqttSourceSecrets) replaced -// by "****", longest first, and then any broker URL or user-info it quotes -// masked (brokerurl.MaskText): MaskText only spots URL-shaped text, so a -// query token or a password quoted on its own would pass it. paho's errors -// normally quote none, but they reach the log and the stats file. +// errForLog is err's text with each of secrets (mqttSourceSecrets) masked +// by "****" in one pass over merged matches, skipping values too short to +// tell from other text (brokerurl.MaskSecrets), and then any broker URL or +// user-info it quotes masked (brokerurl.MaskText): MaskText only spots +// URL-shaped text, so a query token or a password quoted on its own would +// pass it. paho's errors normally quote none, but they reach the log and +// the stats file. func errForLog(err error, secrets ...string) string { if err == nil { return "" } - s := err.Error() - sorted := append([]string(nil), secrets...) - sort.Slice(sorted, func(i, j int) bool { return len(sorted[i]) > len(sorted[j]) }) - for _, v := range sorted { - if v != "" { - s = strings.ReplaceAll(s, v, brokerurl.Marker) - } - } - return brokerurl.MaskText(s) + return brokerurl.MaskText(brokerurl.MaskSecrets(err.Error(), secrets...)) } diff --git a/internal/brokerurl/brokerurl.go b/internal/brokerurl/brokerurl.go index 29dea72af..4bf46cb6c 100644 --- a/internal/brokerurl/brokerurl.go +++ b/internal/brokerurl/brokerurl.go @@ -13,13 +13,17 @@ package brokerurl import ( + "net/url" "regexp" + "sort" "strings" + "unicode/utf8" ) // Marker replaces removed user-info in Mask and MaskText, so a reader can // tell credentials were present, and that a host shown after it may be -// only what followed an '@' in the path or query. +// only what followed an '@' in the path or query. MaskSecrets replaces +// each masked span with it too. const Marker = "****" // Mask returns s with its user-info replaced by Marker and without query @@ -33,21 +37,118 @@ func Mask(s string) string { return join(p.scheme, rest) } -// Secrets returns what Mask removes from s, so that a caller can mask the -// same values where they appear without a URL around them (say, a query -// token quoted in an error): the user-info, the query and the fragment, -// each only when non-empty. +// Secrets returns what Mask removes from s, and its parts, so that a +// caller can mask them where they appear without a URL around them (say, +// a password or a query token quoted in an error; see MaskSecrets): the +// user-info and, when it has a ':', the user name before it and the +// password after it; the query and each of its values; the fragment and +// each of its values. A value is what follows the first '=' of an +// '&'-separated part, or the whole part when it has no '='. Each comes +// raw and, where it differs, %-decoded as in a path and as in a query +// ('+' as space). Empty values and duplicates are left out. func Secrets(s string) []string { p := split(s) - var out []string - for _, v := range []string{p.userinfo, p.query, p.fragment} { + var raw []string + if p.userinfo != "" { + raw = append(raw, p.userinfo) + if user, pass, ok := strings.Cut(p.userinfo, ":"); ok { + raw = append(raw, user, pass) + } + } + for _, v := range []string{p.query, p.fragment} { if v != "" { - out = append(out, v) + raw = append(raw, v) + raw = append(raw, values(v)...) } } + var out []string + seen := map[string]bool{"": true} + for _, v := range raw { + for _, d := range decoded(v) { + if !seen[d] { + seen[d] = true + out = append(out, d) + } + } + } + return out +} + +// values returns the value of each '&'-separated part of a query or +// fragment. +func values(q string) []string { + var out []string + for _, part := range strings.Split(q, "&") { + if _, v, ok := strings.Cut(part, "="); ok { + part = v + } + out = append(out, part) + } + return out +} + +// decoded returns v and its %-decoded forms; a bad escape has none. +func decoded(v string) []string { + out := []string{v} + if d, err := url.PathUnescape(v); err == nil { + out = append(out, d) + } + if d, err := url.QueryUnescape(v); err == nil { + out = append(out, d) + } return out } +// MinSecretLen is the length in runes below which MaskSecrets leaves a +// secret alone. A one- or two-character user name or password would mask +// that substring all over unrelated text, and the free text is all +// MaskSecrets sees: a URL that holds such a value is still masked whole +// by Mask and MaskText, and a longer secret containing it (such as the +// user-info "u:pw") still is by MaskSecrets. +const MinSecretLen = 3 + +// MaskSecrets replaces every occurrence in s of each of secrets (Secrets, +// or values a caller knows, such as a configured password) by Marker. It +// collects the match intervals of all secrets, including overlapping +// matches of one secret, merges those that overlap or touch, and replaces +// each merged interval once, so overlapping secrets leave no residue +// ("abcd" and "cdef" turn "abcdef" into one Marker) and a Marker is never +// matched again. Secrets shorter than MinSecretLen are skipped. +func MaskSecrets(s string, secrets ...string) string { + type span struct{ start, end int } + var spans []span + for _, v := range secrets { + if utf8.RuneCountInString(v) < MinSecretLen { + continue + } + for i := 0; ; { + j := strings.Index(s[i:], v) + if j < 0 { + break + } + spans = append(spans, span{i + j, i + j + len(v)}) + i += j + 1 + } + } + if len(spans) == 0 { + return s + } + sort.Slice(spans, func(a, b int) bool { return spans[a].start < spans[b].start }) + var b strings.Builder + last := 0 + for k := 0; k < len(spans); { + start, end := spans[k].start, spans[k].end + for k++; k < len(spans) && spans[k].start <= end; k++ { + end = max(end, spans[k].end) + } + b.WriteString(s[last:start]) + b.WriteString(Marker) + last = end + } + b.WriteString(s[last:]) + return b.String() +} + // urlUserinfoRe matches a URL with user-info in free text: from the start of // the token holding the scheme to the last '@' on the line and the rest of // that token, so a password with whitespace is covered whole (and any text From fe40aae5f29c0ca9aa51cd2117697c6804811a42 Mon Sep 17 00:00:00 2001 From: dborup Date: Sun, 4 Oct 2026 10:36:42 +0000 Subject: [PATCH 3/6] test(brokerurl): reproduce an '@' later in the URL, invalid UTF-8 and unwired secrets (#159) Review round 1 of #213: - F1: an '@' in the query hides the password and the query values from Secrets (tcp://dev-user:hunter2@broker.example:1883?mail=a@b with "bad password hunter2", tcp://h?token=abc123&mail=a@b with "token abc123 expired"). - F4: a decoded form that is not valid UTF-8 (%A6abc) matches inside a rune and leaves a broken byte. - F6/F7: the per-source setup main() uses returns the secrets once and wires the watchdog's forced reconnect with them (attachClient). Red at be93b891: assertions fail in internal/brokerurl, and cmd/ingestor does not build (prepareMQTTSource returns no setup yet). Co-Authored-By: Claude Opus 5.5 --- cmd/ingestor/mqtt_credentials_159_test.go | 32 +++++++++++++++++++ cmd/ingestor/mqtt_credentials_r3_118_test.go | 3 +- .../mqtt_status_credentials_118_test.go | 6 ++-- internal/brokerurl/brokerurl_test.go | 18 ++++++++++- 4 files changed, 55 insertions(+), 4 deletions(-) diff --git a/cmd/ingestor/mqtt_credentials_159_test.go b/cmd/ingestor/mqtt_credentials_159_test.go index 6040044df..090e432ce 100644 --- a/cmd/ingestor/mqtt_credentials_159_test.go +++ b/cmd/ingestor/mqtt_credentials_159_test.go @@ -14,6 +14,10 @@ func TestErrForLogMasksSecretParts_159(t *testing.T) { {"tcp://dev-user:hunter2@host", "bad password hunter2 for dev-user", "bad password **** for ****"}, {"tcp://dev-user:p%40ss@host", "auth p@ss rejected", "auth **** rejected"}, {"wss://host/mqtt?token=abc123", "token abc123 expired", "token **** expired"}, + // an '@' later in the URL does not hide the password or the query + // values (review of round 1, F1) + {"tcp://dev-user:hunter2@broker.example:1883?mail=a@b", "bad password hunter2", "bad password ****"}, + {"tcp://h?token=abc123&mail=a@b", "token abc123 expired", "token **** expired"}, // short values are left alone in free text; the URL itself is // still masked by MaskText {"tcp://u:pw@host", "uptime 1h, pw ok: tcp://u:pw@host refused", "uptime 1h, pw ok: tcp://****@host refused"}, @@ -48,3 +52,31 @@ func TestBuildForceReconnectFnMasksConnectError_159(t *testing.T) { } } } + +// The per-source setup main() uses computes the source's secrets once and +// wires the watchdog to the client with them, so a forced reconnect's +// Connect() error is logged masked (review of round 1, F6 and F7). +func TestAttachClientWiresSecretsToForceReconnect_159(t *testing.T) { + src := MQTTSource{Name: "feed", Broker: "wss://" + credUser + ":" + credPass + "@host/mqtt?token=abc123", Password: "cfg-pass"} + setup := prepareMQTTSource(src, "feed") + if got, want := strings.Join(setup.secrets, "|"), strings.Join(mqttSourceSecrets(src), "|"); got != want { + t.Errorf("secrets %q, want %q", got, want) + } + c := &fakeClient{connectErr: errors.New(`dial "` + src.Broker + `": cfg-pass refused; token abc123 rejected for ` + credUser)} + setup.attachClient(c) + if setup.liveness.IsConnectedFn == nil || setup.liveness.ForceReconnectFn == nil { + t.Fatal("attachClient left the watchdog unwired") + } + buf := captureLog118(t) + setup.liveness.ForceReconnectFn() + out := buf.String() + const want = `MQTT [feed] WATCHDOG force-reconnect Connect() failed: dial "wss://****@host/mqtt **** refused; token **** rejected for ****` + if !strings.Contains(out, want) { + t.Errorf("log %q, want it to contain %q", out, want) + } + for _, bad := range []string{credUser, credPass, "abc123", "cfg-pass"} { + if strings.Contains(out, bad) { + t.Errorf("log leaks %q: %s", bad, out) + } + } +} diff --git a/cmd/ingestor/mqtt_credentials_r3_118_test.go b/cmd/ingestor/mqtt_credentials_r3_118_test.go index b47ffa9f7..5ff184598 100644 --- a/cmd/ingestor/mqtt_credentials_r3_118_test.go +++ b/cmd/ingestor/mqtt_credentials_r3_118_test.go @@ -79,7 +79,8 @@ func TestDisconnectErrorMasksKnownSecrets_118(t *testing.T) { t.Cleanup(resetSourceStatusRegistry) buf := captureLog118(t) src := MQTTSource{Name: "feed", Broker: "wss://host/mqtt?token=abc", Password: "cfg-pass"} - opts, status, _ := prepareMQTTSource(src, "feed") + setup := prepareMQTTSource(src, "feed") + opts, status := setup.opts, setup.status opts.OnConnectionLost(nil, errors.New("dial host/mqtt?token=abc: cfg-pass rejected")) const want = "dial host/mqtt?****: **** rejected" if got := status.snapshot(time.Now()).LastError; got != want { diff --git a/cmd/ingestor/mqtt_status_credentials_118_test.go b/cmd/ingestor/mqtt_status_credentials_118_test.go index 27ed4e253..5c94f35f5 100644 --- a/cmd/ingestor/mqtt_status_credentials_118_test.go +++ b/cmd/ingestor/mqtt_status_credentials_118_test.go @@ -135,7 +135,8 @@ func TestMQTTSourceWiringLeaksNoCredentials_118(t *testing.T) { buf := captureLog118(t) src := MQTTSource{Broker: broker, Topics: []string{"meshcore/#"}} tag := mqttSourceTags([]MQTTSource{src})[0] - opts, _, liveness := prepareMQTTSource(src, tag) + setup := prepareMQTTSource(src, tag) + opts, liveness := setup.opts, setup.liveness opts.SetConnectTimeout(time.Second). SetMaxReconnectInterval(100 * time.Millisecond). SetConnectRetryInterval(50 * time.Millisecond) @@ -200,7 +201,8 @@ func TestStatsFileHasNoCredentials_118(t *testing.T) { {Broker: "tcp://tok3n@host:1883"}, } for i, tag := range mqttSourceTags(sources) { - _, status, liveness := prepareMQTTSource(sources[i], tag) + setup := prepareMQTTSource(sources[i], tag) + status, liveness := setup.status, setup.liveness status.MarkDisconnect(time.Now(), errors.New("connect "+sources[i].Broker+" refused")) registerLivenessOrSkip(liveness) } diff --git a/internal/brokerurl/brokerurl_test.go b/internal/brokerurl/brokerurl_test.go index 41439946e..d94622277 100644 --- a/internal/brokerurl/brokerurl_test.go +++ b/internal/brokerurl/brokerurl_test.go @@ -61,7 +61,17 @@ func TestSecrets(t *testing.T) { }{ {"tcp://dev-user:secret@host:1883", []string{"dev-user:secret", "dev-user", "secret"}}, {"wss://host/mqtt?token=abc#frag", []string{"token=abc", "abc", "frag"}}, - {"dev-user:1234?abc@host?x=1", []string{"dev-user:1234?abc", "dev-user", "1234?abc", "x=1", "1"}}, + {"dev-user:1234?abc@host?x=1", []string{"dev-user:1234?abc", "dev-user", "1234?abc", "x=1", "1", "abc@host?x=1"}}, + // an '@' in the query: the parts are also read with the user-info + // ending inside the authority, so the password and the query + // values are listed on their own too + {"tcp://dev-user:hunter2@broker.example:1883?mail=a@b", []string{ + "dev-user:hunter2@broker.example:1883?mail=a", "dev-user", "hunter2@broker.example:1883?mail=a", + "dev-user:hunter2", "hunter2", "mail=a@b", "a@b"}}, + {"tcp://h?token=abc123&mail=a@b", []string{"h?token=abc123&mail=a", "token=abc123&mail=a@b", "abc123", "a@b"}}, + {"wss://h/p@x#k=fragsecret", []string{"h/p", "k=fragsecret", "fragsecret"}}, + // a decoded form that is not valid UTF-8 is left out + {"tcp://u1x:%A6abc@h", []string{"u1x:%A6abc", "u1x", "%A6abc"}}, // #159: each part raw and URL-decoded, without duplicates {"tcp://dev-user:p%40ss@host", []string{"dev-user:p%40ss", "dev-user:p@ss", "dev-user", "p%40ss", "p@ss"}}, {"wss://host/?a=x+y%21&b&c=", []string{"a=x+y%21&b&c=", "a=x+y!&b&c=", "a=x y!&b&c=", "x+y%21", "x+y!", "x y!", "b"}}, @@ -135,6 +145,12 @@ func TestMaskSecrets_159(t *testing.T) { Secrets("wss://host/mqtt?token=abc123"), "token **** expired"}, {"query value decoded", "token a b/c! expired", Secrets("wss://host/mqtt?token=a+b%2Fc%21"), "token **** expired"}, + {"password with an '@' in the query", "bad password hunter2", + Secrets("tcp://dev-user:hunter2@broker.example:1883?mail=a@b"), "bad password ****"}, + {"query value with an '@' later", "token abc123 expired", + Secrets("tcp://h?token=abc123&mail=a@b"), "token **** expired"}, + {"no match inside a rune", "user æabc here", + Secrets("tcp://u1x:%A6abc@h"), "user æabc here"}, {"fragment value", "bad token xyz789", Secrets("wss://host/#access_token=xyz789"), "bad token ****"}, {"overlapping secrets", "abcdef", []string{"abcd", "cdef"}, "****"}, From 3e2ff7379c5f9ec3965c53117ad936452aded840 Mon Sep 17 00:00:00 2001 From: dborup Date: Sun, 4 Oct 2026 10:39:34 +0000 Subject: [PATCH 4/6] test(ingestor): pin main()'s source-secret wiring (#159) main() is not unit-testable, so a source-text guard (as for the route mask backfill) pins its share of the wiring: prepareMQTTSource, then setup.attachClient(client) before the first Connect(), and the initial connect error logged with setup.secrets. Red at fe40aae5. Kills the review's mutant M5 and its variants, and guards the coming merge with #210 in the same loop. Co-Authored-By: Claude Opus 5.5 --- cmd/ingestor/mqtt_credentials_159_test.go | 26 +++++++++++++++++++++++ 1 file changed, 26 insertions(+) diff --git a/cmd/ingestor/mqtt_credentials_159_test.go b/cmd/ingestor/mqtt_credentials_159_test.go index 090e432ce..51df0518a 100644 --- a/cmd/ingestor/mqtt_credentials_159_test.go +++ b/cmd/ingestor/mqtt_credentials_159_test.go @@ -2,6 +2,7 @@ package main import ( "errors" + "os" "strings" "testing" ) @@ -80,3 +81,28 @@ func TestAttachClientWiresSecretsToForceReconnect_159(t *testing.T) { } } } + +// main() is not unit-testable, so this pins its share of the wiring: the +// watchdog is attached through the source's setup (attachClient, which +// carries the secrets) before the first Connect(), and the initial +// connect error is logged with the same secrets. A rewrite or merge of +// the connect loop must keep both (review of round 1, F7). +func TestMainWiresSourceSecrets_159(t *testing.T) { + b, err := os.ReadFile("main.go") + if err != nil { + t.Fatal(err) + } + s := string(b) + prepare := strings.Index(s, "setup := prepareMQTTSource(source, tag)") + attach := strings.Index(s, "setup.attachClient(client)") + connect := strings.Index(s, "token := client.Connect()") + if prepare < 0 || attach < 0 || connect < 0 || !(prepare < attach && attach < connect) { + t.Errorf("main() must prepare the source, attachClient, then Connect(): prepare=%d attach=%d connect=%d", prepare, attach, connect) + } + if !strings.Contains(s, "errForLog(token.Error(), setup.secrets...)") { + t.Error("main() must log the initial connect error with the source's secrets") + } + if strings.Contains(s, "ForceReconnectFn = ") { + t.Error("main() wires ForceReconnectFn itself; use attachClient, which passes the secrets") + } +} From d9b8ebae20a46692f0c0e274759f22ad41bdcd77 Mon Sep 17 00:00:00 2001 From: dborup Date: Sun, 4 Oct 2026 10:39:34 +0000 Subject: [PATCH 5/6] fix(brokerurl): read user-info within the authority too, skip invalid UTF-8 decodes (#159) Review round 1 of #213: - F1: Secrets reads the URL twice, with the user-info ending at the last '@' (as Mask does) and at the last '@' before the first '/', '?' or '#' (RFC 3986), so an '@' in the path, query or fragment no longer hides the password or the query values. Mask is unchanged. - F4: a %-decoded form that is not valid UTF-8 is dropped. - F2: Secrets documents that only one layer is decoded and nothing is re-encoded. - F6/F7: prepareMQTTSource returns an mqttSourceSetup carrying the secrets, computed once; attachClient wires the watchdog's connected check and forced reconnect with them, and main() logs the initial connect error with setup.secrets. Co-Authored-By: Claude Opus 5.5 --- cmd/ingestor/main.go | 24 +++++----- cmd/ingestor/mqtt_source.go | 28 ++++++++---- internal/brokerurl/brokerurl.go | 79 ++++++++++++++++++++++----------- 3 files changed, 84 insertions(+), 47 deletions(-) diff --git a/cmd/ingestor/main.go b/cmd/ingestor/main.go index 77a0943b9..52bccfc72 100644 --- a/cmd/ingestor/main.go +++ b/cmd/ingestor/main.go @@ -138,8 +138,8 @@ func main() { tags := mqttSourceTags(sources) for i, source := range sources { tag := tags[i] - opts, status, liveness := prepareMQTTSource(source, tag) - secrets := mqttSourceSecrets(source) + setup := prepareMQTTSource(source, tag) + opts, status, liveness := setup.opts, setup.status, setup.liveness connectTimeout := source.ConnectTimeoutOrDefault() log.Printf("MQTT [%s] connect timeout: %ds", tag, connectTimeout) @@ -162,15 +162,15 @@ func main() { // Wire IsConnectedFn now that the client exists, then register. // Registration BEFORE Connect so the attempt counter is available // to OnConnectAttempt on the very first dial. - liveness.IsConnectedFn = client.IsConnected - // #1335: wire force-reconnect so the watchdog can drop a - // half-open TCP socket and re-dial when paho.IsConnected==true - // but no messages have flowed past the stall threshold. Throttled - // per source by the watchdog itself (forceReconnectThrottle). - // Captured-by-value `client` is the same pointer used everywhere - // else for this source. See buildForceReconnectFn for why this is - // NOT simply "Disconnect(250) then Connect()". - liveness.ForceReconnectFn = buildForceReconnectFn(client, tag, secrets...) + // #1335: attachClient also wires force-reconnect so the watchdog + // can drop a half-open TCP socket and re-dial when + // paho.IsConnected==true but no messages have flowed past the + // stall threshold. Throttled per source by the watchdog itself + // (forceReconnectThrottle). Captured-by-value `client` is the same + // pointer used everywhere else for this source. See + // buildForceReconnectFn for why this is NOT simply + // "Disconnect(250) then Connect()". + setup.attachClient(client) // PR #1216 r2 item 3: tag collisions used to log.Fatalf, which // killed the entire ingestor over one config typo and recreated // the #1212 total-ingest-stop class this PR exists to prevent. @@ -189,7 +189,7 @@ func main() { continue } if token.Error() != nil { - log.Printf("MQTT [%s] connection failed (non-fatal): %s", tag, errForLog(token.Error(), secrets...)) + log.Printf("MQTT [%s] connection failed (non-fatal): %s", tag, errForLog(token.Error(), setup.secrets...)) // BL1 fix: Disconnect to stop Paho's internal retry goroutines. // With ConnectRetry=true, Connect() spawns background goroutines // that leak if the client is simply discarded. diff --git a/cmd/ingestor/mqtt_source.go b/cmd/ingestor/mqtt_source.go index d0ddbfd27..ace04f0ba 100644 --- a/cmd/ingestor/mqtt_source.go +++ b/cmd/ingestor/mqtt_source.go @@ -7,13 +7,17 @@ import ( mqtt "github.com/eclipse/paho.mqtt.golang" ) -// prepareMQTTSource is main()'s per-source setup up to the client itself: -// the paho options with the connect, connection-lost and reconnecting -// handlers, the source's status-registry entry and its liveness state -// (IsConnectedFn and ForceReconnectFn are wired by the caller once the -// client exists; registration is left to the caller too). The message -// handler needs the store and ingest buffer, so main() sets it. -func prepareMQTTSource(source MQTTSource, tag string) (*mqtt.ClientOptions, *sourceStatusState, *SourceLivenessState) { +// mqttSourceSetup is main()'s per-source setup up to the client itself. +type mqttSourceSetup struct { + opts *mqtt.ClientOptions // with the connect, connection-lost and reconnecting handlers + status *sourceStatusState // the source's status-registry entry + liveness *SourceLivenessState // wired to the client by attachClient; registration is left to the caller + secrets []string // mqttSourceSecrets, for every error the source logs +} + +// prepareMQTTSource builds a source's mqttSourceSetup. The message handler +// needs the store and ingest buffer, so main() sets it. +func prepareMQTTSource(source MQTTSource, tag string) *mqttSourceSetup { logBroker := brokerForLog(source.Broker) secrets := mqttSourceSecrets(source) opts := buildMQTTOpts(source) @@ -64,5 +68,13 @@ func prepareMQTTSource(source MQTTSource, tag string) (*mqtt.ClientOptions, *sou log.Printf("MQTT [%s] reconnecting to %s", tag, logBroker) }) - return opts, status, liveness + return &mqttSourceSetup{opts: opts, status: status, liveness: liveness, secrets: secrets} +} + +// attachClient wires the source's liveness state to its client once that +// exists: the watchdog's connected check and its forced reconnect, which +// logs Connect()'s error with the source's secrets masked. +func (s *mqttSourceSetup) attachClient(client mqtt.Client) { + s.liveness.IsConnectedFn = client.IsConnected + s.liveness.ForceReconnectFn = buildForceReconnectFn(client, s.liveness.Tag, s.secrets...) } diff --git a/internal/brokerurl/brokerurl.go b/internal/brokerurl/brokerurl.go index 4bf46cb6c..d714722a6 100644 --- a/internal/brokerurl/brokerurl.go +++ b/internal/brokerurl/brokerurl.go @@ -29,7 +29,7 @@ const Marker = "****" // Mask returns s with its user-info replaced by Marker and without query // and fragment. A scheme ("tcp://") is kept when it is a valid URL scheme. func Mask(s string) string { - p := split(s) + p := split(s, false) rest := p.rest if p.hasUserinfo { rest = Marker + "@" + rest @@ -43,24 +43,22 @@ func Mask(s string) string { // user-info and, when it has a ':', the user name before it and the // password after it; the query and each of its values; the fragment and // each of its values. A value is what follows the first '=' of an -// '&'-separated part, or the whole part when it has no '='. Each comes -// raw and, where it differs, %-decoded as in a path and as in a query -// ('+' as space). Empty values and duplicates are left out. +// '&'-separated part, or the whole part when it has no '='. +// +// s is read twice: as Mask reads it, with the user-info ending at the +// last '@', and as RFC 3986 does, with the user-info ending at the last +// '@' before the first '/', '?' or '#'. Either can be the one a library +// quoting a part used, and each hides parts from the other: an '@' in the +// query moves Mask's user-info past the password, and an unescaped '/' in +// the password moves RFC 3986's past it. +// +// Each part comes raw and, where it differs and is valid UTF-8, %-decoded +// as in a path and as in a query ('+' as space). Only one layer is +// decoded and nothing is re-encoded, so a value decoded twice, or quoted +// %-encoded when it was configured raw, is not listed; inside a URL, +// MaskText still masks it. Empty values and duplicates are left out. func Secrets(s string) []string { - p := split(s) - var raw []string - if p.userinfo != "" { - raw = append(raw, p.userinfo) - if user, pass, ok := strings.Cut(p.userinfo, ":"); ok { - raw = append(raw, user, pass) - } - } - for _, v := range []string{p.query, p.fragment} { - if v != "" { - raw = append(raw, v) - raw = append(raw, values(v)...) - } - } + raw := append(split(s, false).secretParts(), split(s, true).secretParts()...) var out []string seen := map[string]bool{"": true} for _, v := range raw { @@ -74,6 +72,25 @@ func Secrets(s string) []string { return out } +// secretParts returns p's user-info, user name and password, query, +// fragment and their values, raw. +func (p parts) secretParts() []string { + var out []string + if p.userinfo != "" { + out = append(out, p.userinfo) + if user, pass, ok := strings.Cut(p.userinfo, ":"); ok { + out = append(out, user, pass) + } + } + for _, v := range []string{p.query, p.fragment} { + if v != "" { + out = append(out, v) + out = append(out, values(v)...) + } + } + return out +} + // values returns the value of each '&'-separated part of a query or // fragment. func values(q string) []string { @@ -87,14 +104,15 @@ func values(q string) []string { return out } -// decoded returns v and its %-decoded forms; a bad escape has none. +// decoded returns v and its %-decoded forms. A bad escape has none, and a +// form that is not valid UTF-8 is dropped: "%A6abc" decodes to a stray +// continuation byte that matches inside a rune of other text. func decoded(v string) []string { out := []string{v} - if d, err := url.PathUnescape(v); err == nil { - out = append(out, d) - } - if d, err := url.QueryUnescape(v); err == nil { - out = append(out, d) + for _, unescape := range []func(string) (string, error){url.PathUnescape, url.QueryUnescape} { + if d, err := unescape(v); err == nil && utf8.ValidString(d) { + out = append(out, d) + } } return out } @@ -183,7 +201,7 @@ func maskSpan(m string) string { return Mask(m) } -// parts is a broker URL as Mask reads it. +// parts is a broker URL as split reads it. type parts struct { scheme string userinfo string // everything before the last '@' @@ -193,13 +211,20 @@ type parts struct { fragment string // without '#' } -func split(s string) parts { +// split reads s as Mask does, with the user-info ending at the last '@', +// or, with authority set, as RFC 3986 does, with the user-info ending at +// the last '@' before the first '/', '?' or '#' (see Secrets). +func split(s string, authority bool) parts { var p parts rest := s if i := strings.Index(rest, "://"); i > 0 && validScheme(rest[:i]) { p.scheme, rest = rest[:i], rest[i+len("://"):] } - if i := strings.LastIndex(rest, "@"); i >= 0 { + end := len(rest) + if i := strings.IndexAny(rest, "/?#"); authority && i >= 0 { + end = i + } + if i := strings.LastIndex(rest[:end], "@"); i >= 0 { p.userinfo, p.hasUserinfo, rest = rest[:i], true, rest[i+1:] } if i := strings.IndexByte(rest, '#'); i >= 0 { From 120c1e3192b46c805e4b582ef96b59169005e5bf Mon Sep 17 00:00:00 2001 From: dborup Date: Sun, 4 Oct 2026 11:27:46 +0000 Subject: [PATCH 6/6] test(ingestor): lock masking of the force-reconnect retry-pending line (#159) #210 added a retry-pending info line next to the Connect() failed line. The test drives it with paho's status error and IsConnected()=true, and a configured password that occurs in that text, and asserts the line is masked and still classified as retry-pending (classification reads the raw error). Co-Authored-By: Claude Opus 5.5 --- cmd/ingestor/mqtt_credentials_159_test.go | 21 +++++++++++++++++++++ 1 file changed, 21 insertions(+) diff --git a/cmd/ingestor/mqtt_credentials_159_test.go b/cmd/ingestor/mqtt_credentials_159_test.go index 51df0518a..7d3bcbf22 100644 --- a/cmd/ingestor/mqtt_credentials_159_test.go +++ b/cmd/ingestor/mqtt_credentials_159_test.go @@ -106,3 +106,24 @@ func TestMainWiresSourceSecrets_159(t *testing.T) { t.Error("main() wires ForceReconnectFn itself; use attachClient, which passes the secrets") } } + +// #210 split the force-reconnect log into a retry-pending info line and a +// failure line; both log the error with the source's secrets masked. paho's +// status error never quotes the URL, so a configured password that occurs +// in its text stands in for one that would. The classification reads the +// raw error: masked first, it would no longer match paho's text and the +// line would be logged as a failure. +func TestBuildForceReconnectFnMasksRetryPendingLine_159(t *testing.T) { + src := MQTTSource{Broker: "tcp://host:1883", Password: "transition"} + c := &fakeClient{isConnected: true, connectErr: errors.New(pahoErrStatusMustBeDisconnected)} + logs := captureLog118(t) + buildForceReconnectFn(c, "retry-source", mqttSourceSecrets(src)...)() + out := logs.String() + const want = "MQTT [retry-source] WATCHDOG force-reconnect: Connect() returned status can only **** to connecting from disconnected; paho reports a retry pending" + if !strings.Contains(out, want) { + t.Errorf("log %q, want it to contain %q", out, want) + } + if strings.Contains(out, "transition") || strings.Contains(out, connectFailedLine102) { + t.Errorf("retry-pending line unmasked or logged as a failure:\n%s", out) + } +}