|
| 1 | +package main |
| 2 | + |
| 3 | +import ( |
| 4 | + "crypto/rand" |
| 5 | + "encoding/binary" |
| 6 | + "encoding/hex" |
| 7 | + "io" |
| 8 | + "net/url" |
| 9 | + "sort" |
| 10 | + "strconv" |
| 11 | + "strings" |
| 12 | + "sync/atomic" |
| 13 | + "time" |
| 14 | + |
| 15 | + "github.com/meshcore-analyzer/brokerurl" |
| 16 | +) |
| 17 | + |
| 18 | +// MQTT client IDs (#118). |
| 19 | +// |
| 20 | +// Without SetClientID paho connects with a zero-length client ID and |
| 21 | +// CleanSession=true; whether the broker then assigns one, rejects the client |
| 22 | +// or lets two ingestors take over each other's session is broker-dependent. |
| 23 | +// Every source therefore gets an explicit ID: |
| 24 | +// |
| 25 | +// - mqttSources[].clientId, used verbatim. It must be unique among all |
| 26 | +// clients connected to that broker at the same time. |
| 27 | +// - otherwise corescope-<name>-<8 hex>, generated once per client |
| 28 | +// construction (buildMQTTOpts). <name> is the sanitized source name, else |
| 29 | +// the broker host, else omitted; the suffix is 32 random bits from |
| 30 | +// crypto/rand. paho reuses the options for its own reconnects and the |
| 31 | +// watchdog's force-reconnect reuses the same client, so the ID is stable |
| 32 | +// for the process lifetime but differs between sources and processes. |
| 33 | +// |
| 34 | +// MQTT 3.1.1 (what paho tries first) only guarantees IDs of 1–23 characters |
| 35 | +// from [0-9A-Za-z], although brokers such as Mosquitto and EMQX accept longer |
| 36 | +// ones. paho falls back to MQTT 3.1 after a failed first handshake, and a |
| 37 | +// strict 3.1 broker rejects IDs over 23 characters. The default is |
| 38 | +// 19 characters plus the name part, so for such a broker configure a short |
| 39 | +// clientId. (The previous empty ID was invalid under 3.1 as well.) |
| 40 | + |
| 41 | +const mqttClientIDMaxBase = 32 |
| 42 | + |
| 43 | +// clientIDRandom is swapped in tests to simulate an entropy failure. |
| 44 | +var clientIDRandom io.Reader = rand.Reader |
| 45 | + |
| 46 | +var clientIDFallbackSeq atomic.Uint64 |
| 47 | + |
| 48 | +// clientIDNow feeds the fallback suffix; swapped in tests to model a coarse |
| 49 | +// clock that returns the same reading twice. |
| 50 | +var clientIDNow = time.Now |
| 51 | + |
| 52 | +// mqttClientID returns the client ID for a new client of this source. |
| 53 | +func mqttClientID(source MQTTSource) string { |
| 54 | + if strings.TrimSpace(source.ClientID) != "" { |
| 55 | + return source.ClientID |
| 56 | + } |
| 57 | + base := sanitizeClientIDPart(source.Name) |
| 58 | + if base == "" { |
| 59 | + // the masked broker: url.Parse alone may take a user name for |
| 60 | + // the host (see brokerForLog) |
| 61 | + if u, err := url.Parse(brokerForLog(source.Broker)); err == nil { |
| 62 | + base = sanitizeClientIDPart(u.Hostname()) |
| 63 | + } |
| 64 | + } |
| 65 | + id := "corescope-" |
| 66 | + if base != "" { |
| 67 | + id += base + "-" |
| 68 | + } |
| 69 | + return id + clientIDSuffix() |
| 70 | +} |
| 71 | + |
| 72 | +// sanitizeClientIDPart lowercases s, keeps [a-z0-9], turns every other run |
| 73 | +// into one '-', trims '-' and caps the result at mqttClientIDMaxBase. |
| 74 | +func sanitizeClientIDPart(s string) string { |
| 75 | + var b strings.Builder |
| 76 | + dash := false |
| 77 | + for _, r := range strings.ToLower(s) { |
| 78 | + if (r >= 'a' && r <= 'z') || (r >= '0' && r <= '9') { |
| 79 | + if dash && b.Len() > 0 { |
| 80 | + b.WriteByte('-') |
| 81 | + } |
| 82 | + dash = false |
| 83 | + b.WriteRune(r) |
| 84 | + continue |
| 85 | + } |
| 86 | + dash = true |
| 87 | + } |
| 88 | + out := b.String() |
| 89 | + if len(out) > mqttClientIDMaxBase { |
| 90 | + out = strings.TrimRight(out[:mqttClientIDMaxBase], "-") |
| 91 | + } |
| 92 | + return out |
| 93 | +} |
| 94 | + |
| 95 | +// clientIDSuffix is 8 hex chars from crypto/rand. Should that ever fail it |
| 96 | +// mixes the clock and a process-wide counter instead, so two constructions |
| 97 | +// still never share a suffix within a process. |
| 98 | +func clientIDSuffix() string { |
| 99 | + var b [4]byte |
| 100 | + if _, err := io.ReadFull(clientIDRandom, b[:]); err != nil { |
| 101 | + v := uint64(clientIDNow().UnixNano()) ^ (clientIDFallbackSeq.Add(1) * 0x9E3779B97F4A7C15) |
| 102 | + binary.BigEndian.PutUint32(b[:], uint32(v>>32)^uint32(v)) |
| 103 | + } |
| 104 | + return hex.EncodeToString(b[:]) |
| 105 | +} |
| 106 | + |
| 107 | +// mqttConnectedLogLine is the "connected" log line. The broker URL is logged |
| 108 | +// with its user-info masked and without query or fragment (brokerForLog), so |
| 109 | +// credentials or device tokens embedded in it never reach the log. |
| 110 | +func mqttConnectedLogLine(tag, broker, clientID string) string { |
| 111 | + return "MQTT [" + tag + "] connected to " + brokerForLog(broker) + " as client " + clientID |
| 112 | +} |
| 113 | + |
| 114 | +// mqttSourceTag is the base of a source's tag in logs and the |
| 115 | +// liveness/status registries: its name or, for an unnamed source, the |
| 116 | +// broker with its credentials masked (brokerForLog). mqttSourceTags makes the tags |
| 117 | +// of a whole configuration unique. |
| 118 | +func mqttSourceTag(source MQTTSource) string { |
| 119 | + if source.Name != "" { |
| 120 | + return source.Name |
| 121 | + } |
| 122 | + return brokerForLog(source.Broker) |
| 123 | +} |
| 124 | + |
| 125 | +// mqttSourceTags returns the tag of every source. Unnamed sources on the |
| 126 | +// same broker (say, with different credentials) would share one tag: the |
| 127 | +// second would lose watchdog tracking and the two would share status |
| 128 | +// counters. So an unnamed source whose tag is taken, by any named source or |
| 129 | +// an earlier unnamed one, gets " (2)", " (3)", … A duplicate Name is left as |
| 130 | +// it is: that is a configuration error, reported by registerLivenessOrSkip. |
| 131 | +func mqttSourceTags(sources []MQTTSource) []string { |
| 132 | + used := make(map[string]bool, len(sources)) |
| 133 | + for _, s := range sources { |
| 134 | + if s.Name != "" { |
| 135 | + used[s.Name] = true |
| 136 | + } |
| 137 | + } |
| 138 | + tags := make([]string, len(sources)) |
| 139 | + for i, s := range sources { |
| 140 | + tag := mqttSourceTag(s) |
| 141 | + if s.Name == "" { |
| 142 | + base := tag |
| 143 | + for n := 2; used[tag]; n++ { |
| 144 | + tag = base + " (" + strconv.Itoa(n) + ")" |
| 145 | + } |
| 146 | + used[tag] = true |
| 147 | + } |
| 148 | + tags[i] = tag |
| 149 | + } |
| 150 | + return tags |
| 151 | +} |
| 152 | + |
| 153 | +// brokerForLog returns broker with its user-info replaced by "****" and |
| 154 | +// without query or fragment (brokerurl.Mask), so credentials or tokens |
| 155 | +// embedded in it never reach a log, the stats file or a client ID, while the |
| 156 | +// "****@" shows that the URL carries credentials and that the host may be |
| 157 | +// cut short. A broker without a scheme is read as tcp://, as paho's |
| 158 | +// AddBroker does. |
| 159 | +func brokerForLog(broker string) string { |
| 160 | + if !strings.Contains(broker, "://") { |
| 161 | + broker = "tcp://" + broker |
| 162 | + } |
| 163 | + return brokerurl.Mask(broker) |
| 164 | +} |
| 165 | + |
| 166 | +// mqttSourceSecrets returns the non-empty secrets of a source: its |
| 167 | +// password and user name, and the user-info, query and fragment of its |
| 168 | +// broker URL as configured (brokerurl.Secrets). |
| 169 | +func mqttSourceSecrets(source MQTTSource) []string { |
| 170 | + var out []string |
| 171 | + for _, v := range append([]string{source.Password, source.Username}, brokerurl.Secrets(source.Broker)...) { |
| 172 | + if v != "" { |
| 173 | + out = append(out, v) |
| 174 | + } |
| 175 | + } |
| 176 | + return out |
| 177 | +} |
| 178 | + |
| 179 | +// errForLog is err's text with each of secrets (mqttSourceSecrets) replaced |
| 180 | +// by "****", longest first, and then any broker URL or user-info it quotes |
| 181 | +// masked (brokerurl.MaskText): MaskText only spots URL-shaped text, so a |
| 182 | +// query token or a password quoted on its own would pass it. paho's errors |
| 183 | +// normally quote none, but they reach the log and the stats file. |
| 184 | +func errForLog(err error, secrets ...string) string { |
| 185 | + if err == nil { |
| 186 | + return "<nil>" |
| 187 | + } |
| 188 | + s := err.Error() |
| 189 | + sorted := append([]string(nil), secrets...) |
| 190 | + sort.Slice(sorted, func(i, j int) bool { return len(sorted[i]) > len(sorted[j]) }) |
| 191 | + for _, v := range sorted { |
| 192 | + if v != "" { |
| 193 | + s = strings.ReplaceAll(s, v, brokerurl.Marker) |
| 194 | + } |
| 195 | + } |
| 196 | + return brokerurl.MaskText(s) |
| 197 | +} |
0 commit comments