fix(brokerurl): mask user, password, query values and decoded forms without residue (#159)

cmd/ingestor/main.go

@@ -138,7 +138,8 @@ func main() {
 	tags := mqttSourceTags(sources)
 	for i, source := range sources {
 		tag := tags[i]
-		opts, status, liveness := prepareMQTTSource(source, tag)
+		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)
 
@@ -161,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)
+		// #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.
@@ -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(), 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.
@@ -584,7 +585,11 @@ func buildMQTTOpts(source MQTTSource) *mqtt.ClientOptions {
 // left set by a Disconnect(), so there it is a strong hint, not a guarantee.
 // The info line therefore keeps paho's error and claims no more than paho
 // reports. With IsConnected() false the error is logged as a failure.
-func buildForceReconnectFn(client mqtt.Client, tag string) func() {
+//
+// Either line logs the error with secrets (mqttSourceSecrets) masked
+// (errForLog): paho's errors may quote the broker URL. The classification
+// reads the raw error, whose text it compares.
+func buildForceReconnectFn(client mqtt.Client, tag string, secrets ...string) func() {
 	return func() {
 		if client.IsConnectionOpen() {
 			client.Disconnect(250)
@@ -598,9 +603,9 @@ func buildForceReconnectFn(client mqtt.Client, tag string) func() {
 		switch {
 		case err == nil:
 		case connectRetryInProgress(client, err):
-			log.Printf("MQTT [%s] WATCHDOG force-reconnect: Connect() returned %v; paho reports a retry pending (IsConnected=true), not starting a new attempt", tag, err)
+			log.Printf("MQTT [%s] WATCHDOG force-reconnect: Connect() returned %s; paho reports a retry pending (IsConnected=true), not starting a new attempt", tag, errForLog(err, secrets...))
 		default:
-			log.Printf("MQTT [%s] WATCHDOG force-reconnect Connect() failed: %v", tag, err)
+			log.Printf("MQTT [%s] WATCHDOG force-reconnect Connect() failed: %s", tag, errForLog(err, secrets...))
 		}
 	}
 }

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 "<nil>"
 	}
-	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...))
 }

cmd/ingestor/mqtt_credentials_159_test.go

@@ -0,0 +1,129 @@
+package main
+
+import (
+	"errors"
+	"os"
+	"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"},
+		// 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"},
+	} {
+		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)
+		}
+	}
+}
+
+// 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)
+		}
+	}
+}
+
+// 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")
+	}
+}
+
+// #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)
+	}
+}

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 {

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

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