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" "git.novox.be/novox/mesh-controller/modules/route-proxy/contract" ) // 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) *contract.Route { t.Helper() raw, ok := f.values[key] if !ok { return nil } var r contract.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", Route: contract.Request{Name: "grafana.example", Port: int(mustFloat(t, port))}}, {From: "drive", Node: "home-server", At: "10.0.0.2", Route: contract.Request{InternalName: "drive.internal", Port: 8080}}, } 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 != contract.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 != contract.ReachInternal || d.Certificate.Authority != contract.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 contract.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", Route: contract.Request{Name: "wiki.grafana.example", Port: 3000}})) 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 contract.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 contract.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 != contract.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 != contract.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 }