package mqtt import ( "net" "strings" "testing" ) func TestPlaintextToAPublicHostIsRefused(t *testing.T) { // The payloads carry customer visit records and the connection carries the // tenant's broker password. A tcp:// URL to a public host is not a config // choice, it is a mistake — and one that WORKS, which is exactly why it // has to fail here rather than be noticed after a year of traffic. for _, url := range []string{ "tcp://broker.example.com:1883", "mqtt://66.116.226.234:1883", "tcp://10.0.0.5:1883", "tcp://[2001:db8::1]:1883", } { if _, err := NewClient(ClientOptions{BrokerURL: url}); err == nil || !strings.Contains(err.Error(), "refusing plaintext") { t.Errorf("%s was not refused (err=%v)", url, err) } } } func TestPlaintextToLocalhostIsAllowed(t *testing.T) { // Local testing against a Mosquitto on the same box crosses no network. // Checked at the transport gate rather than through NewClient: dialling a // port nothing is listening on burns the full 20s connect timeout, and a // slow test is a test people start skipping. for _, url := range []string{"tcp://127.0.0.1:1883", "tcp://localhost:1883", "mqtt://[::1]:1883"} { if err := checkTransport(url); err != nil { t.Errorf("loopback %s was refused: %v", url, err) } } } func TestPlaintextEscapeHatchIsExplicit(t *testing.T) { // An override must exist for a lab, but it has to be a deliberate act, // not a config field someone leaves set. t.Setenv("BEHAVISION_ALLOW_PLAINTEXT_MQTT", "1") if err := checkTransport("tcp://broker.example.com:1883"); err != nil { t.Fatalf("escape hatch did not apply: %v", err) } } func TestTLSUrlsSkipTheTransportCheck(t *testing.T) { for _, url := range []string{"tls://b:8883", "ssl://b:8883", "wss://b:443"} { if err := checkTransport(url); err != nil { t.Errorf("%s rejected: %v", url, err) } } } func TestAnEmptyBrokerUrlIsAnError(t *testing.T) { if _, err := NewClient(ClientOptions{}); err == nil { t.Fatal("empty broker url accepted") } } func TestTLSConfigRejectsAnUnreadableCA(t *testing.T) { // Silently falling back to system roots when a pinned CA is missing would // quietly undo the pinning. if _, err := tlsConfig(ClientOptions{CAFile: "/nonexistent/ca.pem"}); err == nil { t.Fatal("missing CA file accepted") } } func TestTLSConfigRejectsAFileWithNoCertificates(t *testing.T) { f := t.TempDir() + "/not-a-cert.pem" if err := writeFile(f, "hello"); err != nil { t.Fatal(err) } if _, err := tlsConfig(ClientOptions{CAFile: f}); err == nil { t.Fatal("a file with no PEM certificates was accepted as a CA") } } func TestTLSFloorIsTLS12(t *testing.T) { cfg, err := tlsConfig(ClientOptions{}) if err != nil { t.Fatal(err) } if cfg.MinVersion < 0x0303 { t.Fatalf("MinVersion %#x allows TLS below 1.2", cfg.MinVersion) } } func TestPublishOnADeadClientErrorsRatherThanPanics(t *testing.T) { // The pump calls this on every tick; a nil-client panic would take the // whole agent down instead of backing off. c := &Client{} // The nil context is the POINT: the pump must not panic on a client that // never connected. //nolint is golangci-lint's directive and staticcheck // ignores it, which is why this kept being reported. //lint:ignore SA1012 passing nil is what is under test if err := c.Publish(nil, "t", []byte("{}")); err == nil { t.Fatal("publish on an unconnected client reported success") } if c.Connected() { t.Fatal("an unconnected client reported Connected") } } func writeFile(path, content string) error { return osWriteFile(path, []byte(content), 0o600) } // "The broker refused this PC" and "the broker is not there" need opposite // actions - re-link this PC, or go and look at the network - and paho's // connect-retry makes both arrive as a timeout. Measured on a real broker: // mosquitto logged "not authorised" while the agent logged a timeout, which // sent the diagnosis to the wrong place. func TestARefusedBrokerIsNotDescribedAsUnreachable(t *testing.T) { // A listener that accepts TCP and then says nothing is exactly what a // broker rejecting a client looks like from out here. ln, err := net.Listen("tcp", "127.0.0.1:0") if err != nil { t.Fatal(err) } defer ln.Close() go func() { for { c, err := ln.Accept() if err != nil { return } _ = c } }() got := describeStall("tcp://" + ln.Addr().String()) if !strings.Contains(got, "reachable but did not accept") { t.Fatalf("a reachable broker was described as unreachable: %s", got) } if !strings.Contains(got, "claim") { t.Errorf("the message does not say what to do about it: %s", got) } } func TestAnAbsentBrokerIsDescribedAsUnreachable(t *testing.T) { // Bound and immediately closed, so the port is certainly nobody's. ln, err := net.Listen("tcp", "127.0.0.1:0") if err != nil { t.Fatal(err) } addr := ln.Addr().String() ln.Close() got := describeStall("tcp://" + addr) if !strings.Contains(got, "cannot reach the broker") { t.Fatalf("an absent broker was not described as unreachable: %s", got) } } // An IPv6 literal is bracketed and full of colons, so scanning for the first // one gives "[". The same bug this package already fixed once for broker URLs. func TestTheProbeAddressHandlesIPv6AndDefaultPorts(t *testing.T) { for _, tc := range []struct{ in, want string }{ {"tcp://127.0.0.1:51883", "127.0.0.1:51883"}, {"tcp://[::1]:1883", "[::1]:1883"}, {"tcp://broker.example", "broker.example:1883"}, {"tls://broker.example", "broker.example:8883"}, {"tls://[2001:db8::1]:8884", "[2001:db8::1]:8884"}, } { if got := brokerHostPort(tc.in); got != tc.want { t.Errorf("brokerHostPort(%q) = %q, want %q", tc.in, got, tc.want) } } }