mesh-controller/modules/route-proxy/cmd/route-proxy/ledger_test.go

package main

import (
	"context"
	"crypto/ecdsa"
	"crypto/elliptic"
	"crypto/rand"
	"crypto/tls"
	"crypto/x509"
	"crypto/x509/pkix"
	"encoding/json"
	"encoding/pem"
	"errors"
	"fmt"
	"math/big"
	"net"
	"net/http"
	"net/http/httptest"
	"sort"
	"strings"
	"testing"
	"time"

	"golang.org/x/crypto/acme/autocert"

	stdio "git.novox.be/novox/mesh-sdk/go"

	"route-proxy/internal/routestate"
)

// fakeState is a state on the bus, in memory, counting what was written to it.
type fakeState struct {
	values map[string]json.RawMessage
	puts   int
	tries  int   // every Put asked, refused or not
	fail   error // what every Put is refused with, while set
}

func newFakeState() *fakeState { return &fakeState{values: map[string]json.RawMessage{}} }

func (f *fakeState) Put(key string, value any) (uint64, error) {
	f.tries++
	if f.fail != nil {
		return 0, f.fail
	}
	raw, err := json.Marshal(value)
	if err != nil {
		return 0, err
	}
	f.values[key] = raw
	f.puts++
	return uint64(f.puts), nil
}
func (f *fakeState) Delete(key string) error { delete(f.values, key); return nil }
func (f *fakeState) Keys() ([]string, error) {
	out := []string{}
	for k := range f.values {
		out = append(out, k)
	}
	sort.Strings(out)
	return out, nil
}
func (f *fakeState) Get(key string) (*stdio.StateEntry, error) {
	v, ok := f.values[key]
	if !ok {
		return nil, nil
	}
	return &stdio.StateEntry{Key: key, Value: v}, nil
}

func (f *fakeState) route(t *testing.T, key string) *routestate.Route {
	t.Helper()
	raw, ok := f.values[key]
	if !ok {
		return nil
	}
	var r routestate.Route
	if err := json.Unmarshal(raw, &r); err != nil {
		t.Fatal(err)
	}
	return &r
}

// issuedPEM is what autocert stores for a name: the key, then the certificate, signed by an issuer named here.
func issuedPEM(t *testing.T, name string, notAfter time.Time) []byte {
	t.Helper()
	key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
	if err != nil {
		t.Fatal(err)
	}
	template := &x509.Certificate{SerialNumber: big.NewInt(1), DNSNames: []string{name},
		Subject: pkix.Name{CommonName: name}, Issuer: pkix.Name{CommonName: "R11", Organization: []string{"Let's Encrypt"}},
		NotBefore: notAfter.Add(-90 * 24 * time.Hour), NotAfter: notAfter}
	issuer := &x509.Certificate{SerialNumber: big.NewInt(2), Subject: template.Issuer, NotBefore: template.NotBefore,
		NotAfter: notAfter, IsCA: true, BasicConstraintsValid: true}
	der, err := x509.CreateCertificate(rand.Reader, template, issuer, &key.PublicKey, key)
	if err != nil {
		t.Fatal(err)
	}
	keyDER, _ := x509.MarshalECPrivateKey(key)
	out := pem.EncodeToMemory(&pem.Block{Type: "EC PRIVATE KEY", Bytes: keyDER})
	return append(out, pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: der})...)
}

// What the proxy tells the bus (issue 774): each served name with where it goes, its certificate and the last
// upstream answer; each name refused; and only on change. Without the ledger nothing reaches the bus, and the
// operator's question "is grafana reachable" is answered with curl.
func TestTheBusIsToldWhatIsServedAndRefused(t *testing.T) {
	notAfter := time.Date(2027, 1, 28, 12, 0, 0, 0, time.UTC)
	gone := func() string { // an address nothing listens on: the upstream that does not answer
		l, err := net.Listen("tcp", "127.0.0.1:0")
		if err != nil {
			t.Fatal(err)
		}
		defer l.Close()
		return l.Addr().String()
	}()
	_, port, _ := net.SplitHostPort(gone)
	contributions := []contribution{
		{From: "grafana", Node: "anchor", At: "127.0.0.1", Values: map[string]any{"name": "grafana.example", "port": mustFloat(t, port)}},
		{From: "drive", Node: "home-server", At: "10.0.0.2", Values: map[string]any{"internal-name": "drive.internal", "port": 8080.0}},
	}

	for _, row := range []struct {
		name  string
		act   func(t *testing.T, held *table)
		check func(t *testing.T, routes, unrouted *fakeState)
	}{
		{"a served name is written with the module and node it forwards to",
			func(t *testing.T, held *table) {},
			func(t *testing.T, routes, _ *fakeState) {
				r := routes.route(t, "anchor.grafana.example")
				if r == nil || r.Proxy != "anchor" || r.Reach != routestate.ReachPublic || len(r.Targets) != 1 ||
					r.Targets[0].Module != "grafana" || r.Targets[0].Node != "anchor" || r.Targets[0].Target != "http://127.0.0.1:"+port {
					t.Fatalf("grafana's route is not on the bus as served: %+v", r)
				}
				if d := routes.route(t, "anchor.drive.internal"); d == nil || d.Reach != routestate.ReachInternal ||
					d.Certificate.Authority != routestate.ReachInternal {
					t.Fatalf("an internal-only name is not on the bus as internal: %+v", d)
				}
			}},
		{"a withdrawn route is deleted",
			func(t *testing.T, held *table) {
				routes, public := routesOf(contributions[1:])
				held.set(routes, public)
			},
			func(t *testing.T, routes, _ *fakeState) {
				if routes.route(t, "anchor.grafana.example") != nil {
					t.Fatal("a route no longer served is still on the bus")
				}
			}},
		{"an upstream that does not answer is written as a 502 with why, once a minute",
			func(t *testing.T, held *table) {
				for range 3 {
					req := httptest.NewRequest(http.MethodGet, "http://grafana.example/login", nil)
					handler(held).ServeHTTP(httptest.NewRecorder(), req)
				}
			},
			func(t *testing.T, routes, _ *fakeState) {
				r := routes.route(t, "anchor.grafana.example")
				if r == nil || r.LastAnswer == nil || r.LastAnswer.Status != http.StatusBadGateway ||
					r.LastAnswer.Error == "" || r.LastAnswer.Path != "/login" {
					t.Fatalf("the failed answer is not on the bus: %+v", r)
				}
				if routes.puts != 3 { // the two routes, then the first answer; the same answer again is not written
					t.Fatalf("the same answer was written %d times in all, want 3 writes", routes.puts)
				}
			}},
		{"a name under a served domain is written as asked, without who asked, and leaves when it is served",
			func(t *testing.T, held *table) {
				req := httptest.NewRequest(http.MethodGet, "http://wiki.grafana.example/", nil)
				req.RemoteAddr = "192.0.2.7:51000"
				handler(held).ServeHTTP(httptest.NewRecorder(), req)
				held.ledger.flush()
				var a routestate.Unrouted
				raw := held.ledger.unrouted.(*fakeState).values["anchor.wiki.grafana.example"]
				if err := json.Unmarshal(raw, &a); err != nil || a.Times != 1 || a.AskedAt.IsZero() {
					t.Fatalf("the refused name is not on the bus as asked: %s", raw)
				}
				if strings.Contains(string(raw), "192.0.2.7") || strings.Contains(string(raw), `"from"`) {
					t.Fatalf("who asked is on the bus: %s", raw)
				}
				routes, public := routesOf(append(contributions, contribution{From: "grafana", Node: "anchor",
					Values: map[string]any{"name": "wiki.grafana.example", "port": 3000.0}}))
				held.set(routes, public)
			},
			func(t *testing.T, _, unrouted *fakeState) {
				if len(unrouted.values) != 0 {
					t.Fatalf("a name now served is still written as unrouted: %v", unrouted.values)
				}
			}},
		{"a handshake for a name not routed is written as asked",
			func(t *testing.T, held *table) {
				pick := certificateFor(held, func(*tls.ClientHelloInfo) (*tls.Certificate, error) {
					return nil, errors.New("no public route for \"forgotten.grafana.example\" in this mesh, so no certificate is asked for")
				}, nil)
				_, _ = pick(&tls.ClientHelloInfo{ServerName: "forgotten.grafana.example"})
			},
			func(t *testing.T, _, unrouted *fakeState) {
				var a routestate.Unrouted
				if err := json.Unmarshal(unrouted.values["anchor.forgotten.grafana.example"], &a); err != nil || !a.Handshake {
					t.Fatalf("a handshake refused for its name is not on the bus: %v", unrouted.values)
				}
			}},
		{"a flood of names writes at most the limit by name, counts the rest once a minute, and never retries a refused write",
			func(t *testing.T, held *table) {
				clock := time.Date(2026, 10, 30, 21, 0, 0, 0, time.UTC)
				held.ledger.now = func() time.Time { return clock }
				full := held.ledger.unrouted.(*fakeState)
				for i := range 40 {
					held.ledger.refused(fmt.Sprintf("x%d.grafana.example", i), false, false) // under a served domain
					held.ledger.refused(fmt.Sprintf("x%d.scan.test", i), false, true)        // under none
				}
				held.ledger.flush()
				if named := len(full.values); named != newPerMinute {
					t.Fatalf("a flood wrote %d names, want the limit %d: %v", named, newPerMinute, full.values)
				}
				full.fail = errors.New("nats: maximum bytes exceeded")
				clock = clock.Add(time.Minute)
				held.ledger.refused("y.grafana.example", false, false) // a new minute: one more name,
				held.ledger.refused("v.scan.test", false, false)       // and the last minute's count
				held.ledger.flush()
				tries := full.tries
				held.ledger.refused("y.grafana.example", false, false)
				held.ledger.refused("z.scan.test", false, false)
				held.ledger.flush()
				if full.tries != tries {
					t.Fatalf("writes the bucket refused were tried again: %d tries, then %d", tries, full.tries)
				}
				full.fail = nil
				clock = clock.Add(time.Minute)
				held.ledger.refused("w.scan.test", false, false)
			},
			func(t *testing.T, _, unrouted *fakeState) {
				var o routestate.Other
				if err := json.Unmarshal(unrouted.values["anchor._other"], &o); err != nil || o.LastMinute != 2 || o.SinceStart != 78 {
					t.Fatalf("the refusals not kept by name are not counted on the bus: %s", unrouted.values["anchor._other"])
				}
			}},
		{"a certificate an authority issued is written with its issuer, expiry and renewal",
			func(t *testing.T, held *table) {
				cache := recordingCache{Cache: autocert.DirCache(t.TempDir()), held: held}
				if err := cache.Put(context.Background(), "grafana.example", issuedPEM(t, "grafana.example", notAfter)); err != nil {
					t.Fatal(err)
				}
				_ = cache.Put(context.Background(), "acme_account+key", []byte("not a certificate"))
			},
			func(t *testing.T, routes, _ *fakeState) {
				c := routes.route(t, "anchor.grafana.example").Certificate
				if c.Outcome != routestate.Issued || c.Issuer != "Let's Encrypt R11" || c.NotAfter == nil ||
					!c.NotAfter.Equal(notAfter) || c.RenewedAt == nil {
					t.Fatalf("the issued certificate is not on the bus: %+v", c)
				}
			}},
		{"a certificate that cannot be had is written with why",
			func(t *testing.T, held *table) {
				pick := certificateFor(held, func(*tls.ClientHelloInfo) (*tls.Certificate, error) {
					return nil, errors.New("acme: urn:ietf:params:acme:error:rateLimited")
				}, nil)
				_, _ = pick(&tls.ClientHelloInfo{ServerName: "grafana.example"})
			},
			func(t *testing.T, routes, _ *fakeState) {
				c := routes.route(t, "anchor.grafana.example").Certificate
				if c.Outcome != routestate.Failed || c.Error != "acme: urn:ietf:params:acme:error:rateLimited" {
					t.Fatalf("the failed certificate is not on the bus: %+v", c)
				}
			}},
	} {
		t.Run(row.name, func(t *testing.T) {
			held := newTable()
			held.ledger = newLedger()
			routes, unrouted := newFakeState(), newFakeState()
			held.ledger.attach("anchor", routes, unrouted)
			table, public := routesOf(contributions)
			held.set(table, public)
			held.ledger.flush()
			row.act(t, held)
			held.ledger.flush()
			row.check(t, routes, unrouted)
		})
	}
}

func mustFloat(t *testing.T, port string) float64 {
	t.Helper()
	var n float64
	if err := json.Unmarshal([]byte(port), &n); err != nil {
		t.Fatal(err)
	}
	return n
}