mesh-host/internal/bootstrap/ports_test.go

package bootstrap

import (
	"context"
	"crypto/ecdh"
	"crypto/rand"
	"encoding/base64"
	"errors"
	"os"
	"path/filepath"
	"strings"
	"testing"
	"time"

	"git.novox.be/novox/mesh-sdk/go/declaration"
	"github.com/novox/mesh-host/internal/reachable"
	"github.com/novox/mesh-host/internal/store"
	"github.com/novox/mesh-host/internal/tunnel"
)

// Defends novox/hq ADR 0100: the foundation's ports are the node's — inputs to genesis, checked free,
// rewritten into the bundle, and handed to the controller as the node's settings.

func producedBundle(t *testing.T) Rewritten {
	t.Helper()
	template, err := os.ReadFile("../../examples/foundation-first-node.lock")
	if err != nil {
		t.Skip("no example bundle beside this checkout")
	}
	r, err := Rewrite(template, "sha256:"+strings.Repeat("ab", 32))
	if err != nil {
		t.Fatal(err)
	}
	if _, err := RewriteRoot(&r, testCredentials(t, "s", "b")); err != nil {
		t.Fatal(err)
	}
	return r
}

func containerNamed(d *declaration.Declaration, name string) *declaration.Container {
	for _, r := range d.Resources {
		if c, ok := r.(*declaration.Container); ok && c.Name == name {
			return c
		}
	}
	return nil
}

func TestTheDefaultPortsLeaveTheBundleAsItWas(t *testing.T) {
	r := producedBundle(t)
	before := string(r.Bundle)
	got, err := RewritePorts(&r, DefaultPorts(), DefaultOverlayRange)
	if err != nil {
		t.Fatal(err)
	}
	if got.Places != 0 || string(r.Bundle) != before {
		t.Errorf("the default ports rewrote %d place(s)", got.Places)
	}
}

func TestGivenPortsMoveOnlyTheMachinesSide(t *testing.T) {
	r := producedBundle(t)
	p := FoundationPorts{Store: 5433, Bus: 5771, Registry: 5100}
	got, err := RewritePorts(&r, p, "10.77.0.0/16")
	if err != nil {
		t.Fatal(err)
	}
	if got.Places == 0 {
		t.Fatal("nothing was rewritten")
	}
	storeC := containerNamed(r.Declaration, "postgres")
	if len(storeC.Ports) != 1 || storeC.Ports[0] != "5433:5432" {
		t.Errorf("the store publishes %v", storeC.Ports)
	}
	bus := containerNamed(r.Declaration, "mesh-broker-nats")
	if strings.Join(bus.Ports, " ") != "5771:4222 127.0.0.1:8222:8222" {
		t.Errorf("the bus publishes %v", bus.Ports)
	}
	control, err := controlPlaneIn(r.Declaration)
	if err != nil {
		t.Fatal(err)
	}
	for key, value := range control.Env {
		if strings.HasPrefix(key, "MESH_STORE_") && !strings.Contains(value, "@127.0.0.1:5433/") {
			t.Errorf("%s still dials %s", key, value)
		}
	}
	if !strings.HasSuffix(control.Env[BusVar], "@127.0.0.1:5771") {
		t.Errorf("the control plane dials the bus at %s", control.Env[BusVar])
	}
	if control.Env["MESH_BROKER_ADDRESS"] != "192.0.2.10:5771" || r.BrokerAddress != "192.0.2.10:5771" {
		t.Errorf("nodes are told to dial %s (%s)", control.Env["MESH_BROKER_ADDRESS"], r.BrokerAddress)
	}
	if control.Env["MESH_OVERLAY_CIDR"] != "10.77.0.0/16" {
		t.Errorf("the control plane is told the range %q", control.Env["MESH_OVERLAY_CIDR"])
	}
	// The schema step reaches the store inside its own network, on the container's port.
	text := string(r.Bundle)
	if !strings.Contains(text, `MESH_STORE_INVENTORY=postgres://postgres:s@127.0.0.1:5432/inventory`) {
		t.Error("the schema step's connection, inside the store's network, was moved off the container's port")
	}
	for _, want := range []string{"tcp dport 5771 accept", "ct original proto-dst 5771 accept",
		"tcp dport 5100 accept", "ct original proto-dst 5100 accept"} {
		if !strings.Contains(text, want) {
			t.Errorf("the base filter does not say %q", want)
		}
	}
	if strings.Contains(text, "dport 5671 accept") || strings.Contains(text, "dport 5000 accept") {
		t.Error("the base filter still admits a default port")
	}
}

func TestATemplateThatDoesNotSayItsPortsAsExpectedIsRefused(t *testing.T) {
	r := producedBundle(t)
	r.Bundle = []byte(strings.Replace(string(r.Bundle), `"ports": ["5432:5432"]`, `"ports": [ "5432:5432" ]`, 1))
	if _, err := RewritePorts(&r, FoundationPorts{Store: 5433}, ""); err == nil {
		t.Error("a store port the installer could not find was silently left")
	}
}

func TestTwoThingsOnOnePortAreRefused(t *testing.T) {
	p := DefaultPorts()
	p.Registry = p.Store
	if err := p.Check(); err == nil {
		t.Error("the registry and the store were both given one port")
	}
	p = DefaultPorts()
	p.Hub = 5432 // udp, beside the store's tcp: two different ports
	if err := p.Check(); err != nil {
		t.Errorf("a udp port beside a tcp one of the same number was refused: %v", err)
	}
}

// machineRunner answers ss, docker ps, docker inspect and ip from fixtures.
type machineRunner struct {
	ss, ps, addrs, routes, wg string
	unlabelled                map[string]bool
	labelled                  map[string]bool
}

func (m machineRunner) run(_ context.Context, name string, args ...string) (string, error) {
	switch {
	case name == "ss":
		return m.ss, nil
	case name == "docker" && args[0] == "ps":
		return m.ps, nil
	case name == "docker" && args[0] == "container":
		n := args[len(args)-1]
		if m.labelled[n] {
			return "abc\n", nil
		}
		if m.unlabelled[n] {
			return "\n", nil
		}
		return "", errors.New("no such container")
	case name == "ip" && args[1] == "addr":
		return m.addrs, nil
	case name == "ip" && args[1] == "route":
		return m.routes, nil
	case name == "wg":
		return m.wg, nil
	}
	return "", nil
}

func noneOurs(reachable.Reach) bool { return false }

func TestABusyPortIsRefusedNamingItsHolder(t *testing.T) {
	m := machineRunner{
		ss: "tcp LISTEN 0 4096 0.0.0.0:5000 0.0.0.0:* users:((\"docker-proxy\",pid=1,fd=7))\n" +
			"tcp LISTEN 0 4096 0.0.0.0:5671 0.0.0.0:* users:((\"beam.smp\",pid=2,fd=7))\n",
		ps: "predecessor-registry\t0.0.0.0:5000->5000/tcp\n",
	}
	err := PortsFree(context.Background(), m.run, DefaultPorts(), noneOurs)
	if err == nil {
		t.Fatal("held ports were not refused")
	}
	for _, want := range []string{"predecessor-registry", "beam.smp", "tcp/5000", "tcp/5671", "--registry-port"} {
		if !strings.Contains(err.Error(), want) {
			t.Errorf("the refusal does not say %q: %v", want, err)
		}
	}
	p := DefaultPorts()
	p.Registry, p.Bus = 5100, 5771
	if err := PortsFree(context.Background(), m.run, p, noneOurs); err != nil {
		t.Errorf("other ports given and still refused: %v", err)
	}
}

func TestTheFoundationsOwnContainersAreNotCountedOnARerun(t *testing.T) {
	m := machineRunner{
		ss: "tcp LISTEN 0 4096 0.0.0.0:5432 0.0.0.0:* users:((\"docker-proxy\",pid=1,fd=7))\n",
		ps: "mesh-store\t0.0.0.0:5432->5432/tcp\n",
	}
	ours := func(r reachable.Reach) bool { return r.By == "mesh-store" }
	if err := PortsFree(context.Background(), m.run, DefaultPorts(), ours); err != nil {
		t.Errorf("the foundation's own store was counted as holding its port: %v", err)
	}
}

func TestAnOverlappingTunnelIsRefusedNamingItsInterface(t *testing.T) {
	m := machineRunner{
		addrs:  "1: lo    inet 127.0.0.1/8 scope host lo\n5: wg0    inet 10.42.3.1/24 scope global wg0\n7: mesh0    inet 10.42.0.1/16 scope global mesh0\n",
		routes: "default via 192.0.2.1 dev eth0\n10.42.3.0/24 dev wg0 proto kernel scope link src 10.42.3.1\n",
	}
	err := OverlayClear(context.Background(), m.run, "")
	if err == nil || !strings.Contains(err.Error(), "wg0") || strings.Contains(err.Error(), "mesh0") {
		t.Fatalf("the overlap was not named by its interface alone: %v", err)
	}
	if err := OverlayClear(context.Background(), m.run, "10.77.0.0/16"); err != nil {
		t.Errorf("a clear range was refused: %v", err)
	}
}

func TestAPredecessorsContainerUnderTheMeshsNameIsRefused(t *testing.T) {
	m := machineRunner{unlabelled: map[string]bool{"mesh-registry": true}, labelled: map[string]bool{"mesh-store": true}}
	err := NamesFree(context.Background(), m.run, []string{"mesh-store", "mesh-registry", "mesh-broker"}, store.State{})
	if err == nil || !strings.Contains(err.Error(), "mesh-registry") || strings.Contains(err.Error(), "mesh-store") {
		t.Fatalf("names: %v", err)
	}
	known := store.State{Resources: []store.Applied{{ID: "x", Type: "container", Target: "mesh-registry"}}}
	if err := NamesFree(context.Background(), m.run, []string{"mesh-registry"}, known); err != nil {
		t.Errorf("a container this node has a record of was refused: %v", err)
	}
}

// controlRecorder is a control plane that answers everything and writes down what it was told,
// with the content of every file carried to it, by the name it was carried under.
type controlRecorder struct {
	told     []string
	settings map[string]string
}

func (c *controlRecorder) run(_ context.Context, name string, args ...string) (string, error) {
	if name == "docker" && args[0] == "cp" {
		raw, _ := os.ReadFile(args[1])
		c.settings[filepath.Base(args[2])] = string(raw)
		return "", nil
	}
	if name == "docker" && args[0] == "exec" {
		c.told = append(c.told, strings.Join(args[3:], " "))
		if strings.Join(args[3:], " ") == "broker accounts" {
			return "accounts {}\n", nil // the composed user list, placed after every push (issue 608)
		}
	}
	if name == "sh" {
		c.told = append(c.told, "placed the bus's users")
	}
	return "", nil
}

func (c *controlRecorder) index(prefix string) int {
	for i, t := range c.told {
		if strings.HasPrefix(t, prefix) {
			return i
		}
	}
	return -1
}

func TestTheNodesPortsAreSetBeforeTheModuleIsPushed(t *testing.T) {
	t.Setenv("TMPDIR", t.TempDir())
	c := &controlRecorder{settings: map[string]string{}}
	control := controlPlane{container: "temp-mesh-controller", run: c.run, timeout: time.Second}
	o := Options{Node: "anchor", Ports: FoundationPorts{Registry: 5100}, Wait: time.Second}
	if _, err := installModule(context.Background(), o, control, RegistryModule, []byte(`{}`), quietly); err != nil {
		t.Fatal(err)
	}
	set, push := c.index("settings set distribution"), c.index("push anchor")
	if set < 0 || push < 0 || set > push {
		t.Fatalf("settings were not set before the push: %v", c.told)
	}
	if add := c.index("module add"); add > set {
		t.Errorf("settings were set before the module existed: %v", c.told)
	}
	if !strings.Contains(c.told[set], "--node anchor") {
		t.Errorf("the settings are not the node's: %s", c.told[set])
	}
	if got := c.settings["distribution-settings.json"]; got != `{"ports":{"5000":5100}}` {
		t.Errorf("the registry was told %s", got)
	}
}

func TestAGenesisOnTheDefaultsSetsNoSettings(t *testing.T) {
	t.Setenv("TMPDIR", t.TempDir())
	c := &controlRecorder{settings: map[string]string{}}
	control := controlPlane{container: "temp-mesh-controller", run: c.run, timeout: time.Second}
	o := Options{Node: "anchor", Wait: time.Second}
	if _, err := installModule(context.Background(), o, control, RegistryModule, []byte(`{}`), quietly); err != nil {
		t.Fatal(err)
	}
	if c.index("settings") >= 0 {
		t.Errorf("a converged genesis on the default ports set settings: %v", c.told)
	}
}

func TestARerunOfGenesisFindsItsOwnPackageRegistry(t *testing.T) {
	// Raised by genesis itself with no label and no record, so a re-run finds it unlabelled.
	m := machineRunner{unlabelled: map[string]bool{giteaBootstrap: true}}
	rerun := store.State{Resources: []store.Applied{{ID: "store", Type: "container", Target: "mesh-store"}}}
	if err := NamesFree(context.Background(), m.run, []string{giteaBootstrap, "mesh-store"}, rerun); err != nil {
		t.Errorf("a re-run refused the package registry genesis raised: %v", err)
	}
	// On a machine genesis never ran on, a container under that name is a predecessor's.
	if err := NamesFree(context.Background(), m.run, []string{giteaBootstrap}, store.State{}); err == nil {
		t.Error("a container under the package registry's name on a fresh machine was not refused")
	}
}

// novox/hq ADR 0105: an adopted genesis takes over the tunnel it finds — its port is the hub's,
// its range the mesh's, and neither is refused for being held by it.
func TestAnAdoptedGenesisSettlesOnTheTunnelItFinds(t *testing.T) {
	private, err := aFoundKey()
	if err != nil {
		t.Fatal(err)
	}
	conf := "[Interface]\nPrivateKey = " + private + "\nListenPort = 51900\nAddress = 192.0.2.1/24\n" +
		"[Peer]\nPublicKey = PEER=\nAllowedIPs = 192.0.2.2/32\n"
	tunnel.ReadFile = func(path string) ([]byte, error) {
		if path == tunnel.ConfigDir+"/wg0.conf" {
			return []byte(conf), nil
		}
		return nil, errors.New("no such file")
	}
	t.Cleanup(func() { tunnel.ReadFile = os.ReadFile })
	m := machineRunner{
		wg:     "wg0\n",
		ss:     "udp UNCONN 0 0 0.0.0.0:51900 0.0.0.0:*\n",
		addrs:  "5: wg0    inet 192.0.2.1/24 scope global wg0\n",
		routes: "192.0.2.0/24 dev wg0 proto kernel scope link src 192.0.2.1\n",
	}

	o := Options{Adopted: true, Ports: DefaultPorts(), OverlayRange: DefaultOverlayRange, State: filepath.Join(t.TempDir(), "state.json")}
	found, err := TakeTheTunnel(&o, m.run)
	if err != nil || found == nil {
		t.Fatalf("the tunnel was not found and taken: %+v %v", found, err)
	}
	if o.Tunnel != "wg0" || o.Ports.Hub != 51900 || o.OverlayRange != "192.0.2.0/24" {
		t.Fatalf("genesis did not settle on the tunnel's port and range: %+v", o)
	}
	// Its port is held by the tunnel and its range overlaps the tunnel's — by design, not refused.
	if err := CheckTheMachine(context.Background(), o, m.run, producedBundle(t).Declaration, func(string) {}); err != nil {
		t.Fatalf("the machine was refused for the tunnel it takes over: %v", err)
	}
	// Whereas the same machine not taking it over is refused on both counts (ADR 0100).
	plain := o
	plain.Tunnel = ""
	if err := CheckTheMachine(context.Background(), plain, m.run, producedBundle(t).Declaration, func(string) {}); err == nil ||
		!strings.Contains(err.Error(), "51900") {
		t.Fatalf("a tunnel not taken over stopped being refused for holding the hub's port: %v", err)
	}
	plain.Ports.Hub = 51821
	if err := CheckTheMachine(context.Background(), plain, m.run, producedBundle(t).Declaration, func(string) {}); err == nil ||
		!strings.Contains(err.Error(), "wg0") {
		t.Fatalf("a tunnel not taken over stopped being refused for overlapping the range: %v", err)
	}

	// Numbers that disagree with the tunnel are refused, naming the tunnel's.
	for name, given := range map[string]Options{
		"--hub-port":      {Adopted: true, Ports: FoundationPorts{Hub: 51821}, OverlayRange: DefaultOverlayRange},
		"--overlay-range": {Adopted: true, Ports: DefaultPorts(), OverlayRange: "10.77.0.0/16"},
	} {
		if _, err := TakeTheTunnel(&given, m.run); err == nil || !strings.Contains(err.Error(), name) {
			t.Errorf("a %s disagreeing with the tunnel was accepted: %v", name, err)
		}
	}
	// And no tunnel up is an ordinary machine.
	m.wg = "mesh0\n"
	none := Options{Adopted: true, Ports: DefaultPorts(), OverlayRange: DefaultOverlayRange}
	if found, err := TakeTheTunnel(&none, m.run); err != nil || found != nil || none.Ports.Hub != DefaultPorts().Hub {
		t.Errorf("a machine with no tunnel was not left as it was: %+v %v", found, err)
	}
}

func aFoundKey() (string, error) {
	k, err := ecdh.X25519().GenerateKey(rand.Reader)
	if err != nil {
		return "", err
	}
	return base64.StdEncoding.EncodeToString(k.Bytes()), nil
}