Behavision/agent/pkg/mqtt/client_test.go

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{}
	if err := c.Publish(nil, "t", []byte("{}")); err == nil { //nolint:staticcheck
		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)
		}
	}
}