diff --git a/deploy/containerlab/Taskfile.yaml b/deploy/containerlab/Taskfile.yaml index fe8073d7..8d673328 100644 --- a/deploy/containerlab/Taskfile.yaml +++ b/deploy/containerlab/Taskfile.yaml @@ -26,6 +26,7 @@ tasks: - "build:galactic-controller" - "build:galactic-edge" - "build:edge-provider" + - "build:masque-provider" - "build:fabric-router-controller" - "build:remote-host" @@ -81,6 +82,11 @@ tasks: cmds: - docker build --network=host -t edge-provider:latest -f ../../examples/edge-provider/Dockerfile ../.. + "build:masque-provider": + desc: Build the conformance suite's MASQUE fixture provider image (tests/edgeconformance/masqueprovider) + cmds: + - docker build --network=host -t masque-provider:latest -f ../../tests/edgeconformance/masqueprovider/Dockerfile ../.. + "build:fabric-router-controller": desc: Build the fabric-router-controller image (typed fabric routing) cmds: @@ -294,6 +300,7 @@ tasks: load fabric-router-controller:latest {{.FABRIC_NODES}} load galactic-edge:latest {{.EDGE_NODES}} load edge-provider:latest {{.EDGE_NODES}} + load masque-provider:latest {{.EDGE_NODES}} "deploy:system": desc: Install CRDs, namespace, and RBAC @@ -1238,6 +1245,15 @@ tasks: cmds: - EDGE_CONFORMANCE_KUBECONFIG={{.ROOT_DIR}}/dfw.kubeconfig go test -count=1 -v -timeout 90m -run TestConformance ./tests/edgeconformance/ + # Not part of verify: it takes about 15 minutes and deletes and replaces + # its own provider pods on dfw-worker2. Results land beside the + # conformance suite's in bin/edge-conformance/. + "verify:masque-conformance": + desc: Run the MASQUE packet-access rows against dfw with the MASQUE fixture provider + dir: ../.. + cmds: + - EDGE_CONFORMANCE_KUBECONFIG={{.ROOT_DIR}}/dfw.kubeconfig go test -count=1 -v -timeout 40m -run TestMASQUE ./tests/edgeconformance/ + "verify:ns10": desc: Verify ns10 ping (IPv6-only, 3-site mesh) cmds: diff --git a/docs/designs/masque/acceptance.md b/docs/designs/masque/acceptance.md index 034eed08..99e8594a 100644 --- a/docs/designs/masque/acceptance.md +++ b/docs/designs/masque/acceptance.md @@ -10,6 +10,8 @@ Packet cases use sequence-numbered traffic in both directions, capture at sender [#985](https://github.com/datum-cloud/galactic/issues/985) packages setup, fault injection, collection and cleanup commands in a clean-checkout fixture. It asserts removal of test-created links, routes, maps, namespaces and API resources and preservation of unrelated resources. The ledger's retained/tombstoned allocations are expected state, identified explicitly rather than called a leak. Cleanup errors fail the run. Tests never deploy to production as part of this milestone. +`task verify:masque-conformance`, run from `deploy/containerlab`, runs the [#974](https://github.com/datum-cloud/galactic/issues/974) packet-access checks on the lab's dfw cluster. It covers M16 and M07, NAT egress from the attachment node, pod recreation, a partly set-up instance killed without shutdown, and asserted teardown. The provider is the test-only fixture in `tests/edgeconformance/masqueprovider`. It holds an instance lease, `PacketAccess` attachments and per-/96 routes through the provider contract, carries packets with `internal/edge/connectip`, and admits static test credentials in place of the [#977](https://github.com/datum-cloud/galactic/issues/977) authority. A lab result does not satisfy a row that requires the target platform. + ## Epic and implementation checks Run packet, authentication, isolation and failure cases on both HTTP/3 and HTTP/2 unless the row explicitly tests one transport. For HTTP/2, record negotiated h2/TLS/TCP and block UDP/443; a QUIC handshake does not count. diff --git a/tests/edgeconformance/edgeprobe/connectip.go b/tests/edgeconformance/edgeprobe/connectip.go new file mode 100644 index 00000000..2a889f2c --- /dev/null +++ b/tests/edgeconformance/edgeprobe/connectip.go @@ -0,0 +1,216 @@ +// Copyright 2026 Datum Cloud, Inc. +// +// SPDX-License-Identifier: AGPL-3.0-or-later + +package main + +import ( + "context" + "crypto/tls" + "crypto/x509" + "encoding/binary" + "errors" + "flag" + "fmt" + "net" + "net/netip" + "os" + "strings" + "time" + + connectip "github.com/quic-go/connect-ip-go" + "github.com/quic-go/quic-go" + "github.com/quic-go/quic-go/http3" + "github.com/yosida95/uritemplate/v3" + "golang.org/x/net/icmp" + "golang.org/x/net/ipv6" +) + +// connectIP opens one CONNECT-IP request over HTTP/3 and sends ICMPv6 echo +// requests through it from an address in its assigned /96, one to each +// destination of -to on every tick. It prints the +// response status {"t","status"}, the assignment {"t","assigned"}, each echo +// reply {"t","seq","from","dst","reply":"echo"}, every other packet the +// tunnel delivers {"t","from","dst","reply":"other"}, the end of a tunnel +// the server closed {"t","end","error"} and the count {"t","sent"}. A +// refused request exits 0 after printing its status. +func connectIP(args []string) error { + fs := flag.NewFlagSet("connectip", flag.ExitOnError) + server := fs.String("server", "", "server [address]:port") + authority := fs.String("authority", "", "host:port of the URI template; the host is the TLS server name") + caFile := fs.String("ca", "", "PEM file of the CA that signed the server's certificate") + token := fs.String("token", "", "bearer credential") + vpc := fs.String("vpc", "", "Galactic-VPC header value") + to := fs.String("to", "", "comma-separated echo destinations") + srcFlag := fs.String("src", "", "source address; default the first address after the assigned prefix's") + id := fs.Int("id", os.Getpid()&0xffff, "ICMPv6 echo identifier") + size := fs.Int("size", 64, "IPv6 packet size in bytes") + interval := fs.Duration("interval", 50*time.Millisecond, "time between echo requests") + duration := fs.Duration("duration", 5*time.Second, "how long to send") + _ = fs.Parse(args) + var dsts []netip.Addr + for _, a := range strings.Split(*to, ",") { + dst, err := netip.ParseAddr(a) + if err != nil { + return fmt.Errorf("-to: %w", err) + } + dsts = append(dsts, dst) + } + host, _, err := net.SplitHostPort(*authority) + if err != nil { + return fmt.Errorf("-authority: %w", err) + } + pemBytes, err := os.ReadFile(*caFile) + if err != nil { + return err + } + roots := x509.NewCertPool() + if !roots.AppendCertsFromPEM(pemBytes) { + return errors.New("-ca holds no certificate") + } + tmpl, err := uritemplate.New("https://" + *authority + "/.well-known/masque/ip/*/*/") + if err != nil { + return err + } + + ctx, cancel := context.WithTimeout(context.Background(), *duration+30*time.Second) + defer cancel() + qconn, err := quic.DialAddr(ctx, *server, &tls.Config{ServerName: host, RootCAs: roots, + NextProtos: []string{http3.NextProtoH3}, MinVersion: tls.VersionTLS13}, + &quic.Config{EnableDatagrams: true, InitialPacketSize: 1350}) + if err != nil { + return fmt.Errorf("dial %s: %w", *server, err) + } + defer func() { _ = qconn.CloseWithError(0, "") }() + cc, err := (&connectip.Transport{}).NewClientConn(qconn) + if err != nil { + return err + } + req, err := connectip.NewRequest(ctx, tmpl) + if err != nil { + return err + } + req.Header().Set("Authorization", "Bearer "+*token) + req.Header().Set("Galactic-VPC", *vpc) + conn, rsp, err := cc.Dial(req) + if rsp != nil { + _ = out.Encode(cipEvent{T: boottime(), Status: rsp.StatusCode}) + } + if err != nil { + if rsp != nil { + return nil + } + return err + } + defer func() { _ = conn.Close() }() + assigned, err := conn.ReceiveAddressAssignment(ctx) + if err != nil { + return err + } + if len(assigned) != 1 { + return fmt.Errorf("assigned %v, want one prefix", assigned) + } + prefix := assigned[0].IPPrefix + _ = out.Encode(cipEvent{T: boottime(), Assigned: prefix.String()}) + src := prefix.Addr().Next() + if *srcFlag != "" { + if src, err = netip.ParseAddr(*srcFlag); err != nil { + return fmt.Errorf("-src: %w", err) + } + } + + done := make(chan struct{}) + go func() { + defer close(done) + buf := make([]byte, 65535) + for { + n, err := conn.ReadPacket(buf) + if err != nil { + _ = out.Encode(cipEvent{T: boottime(), End: true, Error: err.Error()}) + return + } + report(buf[:n], *id) + } + }() + tick := time.NewTicker(*interval) + defer tick.Stop() + deadline := time.Now().Add(*duration) + sent := 0 +loop: + for time.Now().Before(deadline) { + for _, dst := range dsts { + pkt, err := echoRequest(src, dst, *id, sent, *size) + if err != nil { + return err + } + if _, err := conn.WritePacket(pkt); err != nil { + break loop + } + sent++ + } + select { + case <-tick.C: + case <-done: + break loop + } + } + select { + case <-done: + case <-time.After(500 * time.Millisecond): + } + _ = conn.Close() + <-done + return out.Encode(cipEvent{T: boottime(), Sent: sent}) +} + +// cipEvent is one line connectIP prints. +type cipEvent struct { + T int64 `json:"t"` + Status int `json:"status,omitempty"` + Assigned string `json:"assigned,omitempty"` + Seq *int `json:"seq,omitempty"` + From string `json:"from,omitempty"` + Dst string `json:"dst,omitempty"` + Reply string `json:"reply,omitempty"` + End bool `json:"end,omitempty"` + Error string `json:"error,omitempty"` + Sent int `json:"sent,omitempty"` +} + +// report prints one packet the tunnel delivered. +func report(p []byte, id int) { + t := boottime() + if len(p) < ipv6.HeaderLen || p[0]>>4 != 6 { + _ = out.Encode(cipEvent{T: t, Reply: "other"}) + return + } + src, _ := netip.AddrFromSlice(p[8:24]) + dst, _ := netip.AddrFromSlice(p[24:40]) + ev := cipEvent{T: t, From: src.String(), Dst: dst.String(), Reply: "other"} + if p[6] == 58 { + if m, err := icmp.ParseMessage(58, p[ipv6.HeaderLen:]); err == nil && m.Type == ipv6.ICMPTypeEchoReply { + if e, ok := m.Body.(*icmp.Echo); ok && e.ID == id { + ev.Reply, ev.Seq = "echo", &e.Seq + } + } + } + _ = out.Encode(ev) +} + +// echoRequest builds an IPv6 ICMPv6 echo request of size bytes. +func echoRequest(src, dst netip.Addr, id, seq, size int) ([]byte, error) { + data := make([]byte, max(size-ipv6.HeaderLen-8, 0)) + m := icmp.Message{Type: ipv6.ICMPTypeEchoRequest, Body: &icmp.Echo{ID: id, Seq: seq & 0xffff, Data: data}} + body, err := m.Marshal(icmp.IPv6PseudoHeader(net.IP(src.AsSlice()), net.IP(dst.AsSlice()))) + if err != nil { + return nil, err + } + pkt := make([]byte, ipv6.HeaderLen, ipv6.HeaderLen+len(body)) + pkt[0] = 6 << 4 + binary.BigEndian.PutUint16(pkt[4:6], uint16(len(body))) + pkt[6], pkt[7] = 58, 64 + s, d := src.As16(), dst.As16() + copy(pkt[8:24], s[:]) + copy(pkt[24:40], d[:]) + return append(pkt, body...), nil +} diff --git a/tests/edgeconformance/edgeprobe/main.go b/tests/edgeconformance/edgeprobe/main.go index 1cf4ce40..493824e0 100644 --- a/tests/edgeconformance/edgeprobe/main.go +++ b/tests/edgeconformance/edgeprobe/main.go @@ -13,6 +13,11 @@ // edgeprobe sniff -iface link -port 7777 -duration 30s // prints each IPv6 UDP datagram to port on link: {"t","src","dst"}, // then {"t","end"} when the capture ends +// edgeprobe sniff -iface link -icmp -duration 30s +// prints each ICMPv6 echo request on link instead: {"t","src","dst","id"} +// edgeprobe connectip -server [addr]:port -authority host:port -ca file \ +// -token T -vpc V -to addr -duration 5s +// pings addr through one CONNECT-IP request (see connectIP) package main import ( @@ -33,7 +38,7 @@ import ( func main() { if len(os.Args) < 2 { - fmt.Fprintln(os.Stderr, "usage: edgeprobe send|sniff [flags]") + fmt.Fprintln(os.Stderr, "usage: edgeprobe send|sniff|connectip [flags]") os.Exit(2) } var err error @@ -42,6 +47,8 @@ func main() { err = send(os.Args[2:]) case "sniff": err = sniff(os.Args[2:]) + case "connectip": + err = connectIP(os.Args[2:]) default: err = fmt.Errorf("unknown command %q", os.Args[1]) } @@ -114,6 +121,7 @@ func sniff(args []string) error { fs := flag.NewFlagSet("sniff", flag.ExitOnError) iface := fs.String("iface", "", "link to capture on") port := fs.Uint("port", 7777, "UDP destination port") + echo := fs.Bool("icmp", false, "capture ICMPv6 echo requests instead of UDP") duration := fs.Duration("duration", 10*time.Second, "how long to capture") _ = fs.Parse(args) ifi, err := net.InterfaceByName(*iface) @@ -156,13 +164,18 @@ func sniff(args []string) error { } t := boottime() p := buf[:n] - if len(p) < 48 || p[0]>>4 != 6 || p[6] != unix.IPPROTO_UDP || - uint(binary.BigEndian.Uint16(p[42:44])) != *port { + if len(p) < 48 || p[0]>>4 != 6 { continue } src, _ := netip.AddrFromSlice(p[8:24]) dst, _ := netip.AddrFromSlice(p[24:40]) - _ = out.Encode(map[string]any{"t": t, "src": src.String(), "dst": dst.String()}) + switch { + case *echo && p[6] == unix.IPPROTO_ICMPV6 && p[40] == 128: + _ = out.Encode(map[string]any{"t": t, "src": src.String(), "dst": dst.String(), + "id": binary.BigEndian.Uint16(p[44:46])}) + case !*echo && p[6] == unix.IPPROTO_UDP && uint(binary.BigEndian.Uint16(p[42:44])) == *port: + _ = out.Encode(map[string]any{"t": t, "src": src.String(), "dst": dst.String()}) + } } return out.Encode(map[string]any{"t": boottime(), "end": true}) } diff --git a/tests/edgeconformance/fixtures_test.go b/tests/edgeconformance/fixtures_test.go index d54892df..d9083c49 100644 --- a/tests/edgeconformance/fixtures_test.go +++ b/tests/edgeconformance/fixtures_test.go @@ -104,12 +104,19 @@ patches: // applyOverlay renders a kustomization over one of // examples/edge-provider/deploy's directories and applies it. func (l *lab) applyOverlay(t *testing.T, name, base, extra string, files map[string]string) { + t.Helper() + l.applyOverlayOf(t, filepath.Join(l.root, "examples", "edge-provider", "deploy", base), name, extra, files) +} + +// applyOverlayOf renders a kustomization named name over the directory base +// and applies it. +func (l *lab) applyOverlayOf(t *testing.T, base, name, extra string, files map[string]string) { t.Helper() dir := filepath.Join(l.artifacts, "overlays", name) if err := os.MkdirAll(dir, 0o755); err != nil { t.Fatal(err) } - rel, err := filepath.Rel(dir, filepath.Join(l.root, "examples", "edge-provider", "deploy", base)) + rel, err := filepath.Rel(dir, base) if err != nil { t.Fatal(err) } diff --git a/tests/edgeconformance/harness_test.go b/tests/edgeconformance/harness_test.go index f8e67601..52c21cf3 100644 --- a/tests/edgeconformance/harness_test.go +++ b/tests/edgeconformance/harness_test.go @@ -239,6 +239,12 @@ type event struct { Dst string `json:"dst"` Sent int `json:"sent"` End bool `json:"end"` + // ID is a captured ICMPv6 echo request's identifier. + ID int `json:"id"` + // Status and Assigned are a CONNECT-IP request's response status and + // assigned prefix. + Status int `json:"status"` + Assigned string `json:"assigned"` } // last reports whether e marks the end of a probe's window: a send probe's diff --git a/tests/edgeconformance/masque_test.go b/tests/edgeconformance/masque_test.go new file mode 100644 index 00000000..74991c60 --- /dev/null +++ b/tests/edgeconformance/masque_test.go @@ -0,0 +1,993 @@ +// Copyright 2026 Datum Cloud, Inc. +// +// SPDX-License-Identifier: AGPL-3.0-or-later + +package edgeconformance_test + +import ( + "context" + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" + "crypto/x509" + "crypto/x509/pkix" + "encoding/json" + "encoding/pem" + "fmt" + "math/big" + "net/netip" + "os" + "os/exec" + "path/filepath" + "slices" + "strconv" + "strings" + "testing" + "time" + + coordinationv1 "k8s.io/api/coordination/v1" + corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + "k8s.io/apimachinery/pkg/api/meta" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" + "sigs.k8s.io/controller-runtime/pkg/client" + + networkv1alpha1 "go.datum.net/galactic/api/network/v1alpha1" + galacticv1alpha1 "go.datum.net/galactic/api/v1alpha1" +) + +// The MASQUE fixture's provider, grants and connectors. Both tenants' +// connectors hold the same /96, so their packets are told apart only by the +// VPC they enter; each tenant's client uses its own ICMPv6 echo identifier +// so a packet seen anywhere names the client that sent it. +const ( + masqueNS = "edge-masque" + masqueProvider = "masque-fixture" + masqueService = "svc-masque" + masqueGrant = "fd00:ed9e:300::/48" + masquePool = "fd00:ed9e:0:30::/64" + masquePrefix = "fd00:ed9e:300:1::/96" + masqueOutside = "fd00:ed9e:400:1::/96" + // masquePartial is the partly set-up instance's /96, also outside the + // grant and claimed by no other instance. + masquePartial = "fd00:ed9e:400:2::/96" + masqueAuthority = "masque.test:8443" + masquePort = 8443 + // remoteHost is the lab's off-fabric host, reached through the NAT + // shards. + remoteHost = "2001:db8:1:40::2" + // masqueCA is where the suite copies the fixture's CA onto each node. + masqueCA = "/usr/local/share/masque-ca.pem" + systemNS = "galactic-system" +) + +// masqueTenant is one VPC's connector. +type masqueTenant struct { + tn tenant + svc *galacticv1alpha1.EdgeService + vpcUID string + token string + id int +} + +// masque is the MASQUE fixture: one provider pod serving both tenants. +type masque struct { + l *lab + a, b *masqueTenant + // refused is the credential of a connector whose route core refuses. + refused string + certPEM []byte + keyPEM []byte + // pod is the running pod of the fixture's main Deployment. + pod *corev1.Pod + // links maps every attachment the run created to its host link, for + // teardown's check that none remains. + links map[string]string + // unrelated holds the UIDs of the tenants' edge-client pods, which the + // run must leave in place. + unrelated map[string]types.UID +} + +// TestMASQUE proves CONNECT-IP packet access on the Galactic Edge foundation +// with the MASQUE fixture provider (tests/edgeconformance/masqueprovider): +// the deployed agent creates the fixture's PacketAccess devices and the +// unprivileged fixture opens them, two VPCs with identical connector /96s +// stay isolated in both directions, NAT egress works from the attachment +// node, and recreation, partial setup and teardown clean up through core +// while leaving unrelated and replacement resources alone. Run it with +// task verify:masque-conformance. +func TestMASQUE(t *testing.T) { + l := getLab(t) + t.Cleanup(func() { l.writeSummary(t) }) + m := newMasque(t, l) + if os.Getenv("EDGE_CONFORMANCE_KEEP") == "" { + t.Cleanup(func() { m.teardown(t) }) + } + if !t.Run("setup", m.setup) { + t.FailNow() + } + t.Run("M16-unprivileged-provider", m.rowUnprivileged) + t.Run("M07-isolation", func(t *testing.T) { m.isolation(t, "cold") }) + t.Run("nat-egress", m.rowNATEgress) + t.Run("partial-setup", m.rowPartial) + t.Run("recreation", m.rowRecreation) +} + +func newMasque(t *testing.T, l *lab) *masque { + t.Helper() + certPEM, keyPEM, caPEM := masqueCerts(t) + ca := filepath.Join(l.artifacts, "masque-ca.pem") + if err := os.WriteFile(ca, caPEM, 0o644); err != nil { + t.Fatal(err) + } + for _, n := range []string{computeNode, edgeNode1, edgeNode2} { + cp := exec.CommandContext(context.Background(), "docker", "cp", ca, n+":"+masqueCA) + if out, err := cp.CombinedOutput(); err != nil { + t.Fatalf("copy CA to %s: %v: %s", n, err, out) + } + } + return &masque{ + l: l, + a: &masqueTenant{tn: tenant{vpcNamespace: "ns10", vpc: "10", provider: masqueProvider}, + token: rand.Text(), id: 0x0a10}, + b: &masqueTenant{tn: tenant{vpcNamespace: "ns20", vpc: "20", provider: masqueProvider}, + token: rand.Text(), id: 0x0b20}, + refused: rand.Text(), certPEM: certPEM, keyPEM: keyPEM, + links: map[string]string{}, unrelated: map[string]types.UID{}, + } +} + +// masqueCerts returns a server certificate for the authority's host, its +// key, and the CA that signed it. +func masqueCerts(t *testing.T) (certPEM, keyPEM, caPEM []byte) { + t.Helper() + caKey, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + if err != nil { + t.Fatal(err) + } + caTmpl := &x509.Certificate{SerialNumber: big.NewInt(1), Subject: pkix.Name{CommonName: "masque conformance CA"}, + NotBefore: time.Now().Add(-time.Hour), NotAfter: time.Now().Add(24 * time.Hour), IsCA: true, + BasicConstraintsValid: true, KeyUsage: x509.KeyUsageCertSign} + caDER, err := x509.CreateCertificate(rand.Reader, caTmpl, caTmpl, &caKey.PublicKey, caKey) + if err != nil { + t.Fatal(err) + } + caCert, err := x509.ParseCertificate(caDER) + if err != nil { + t.Fatal(err) + } + key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + if err != nil { + t.Fatal(err) + } + host := strings.Split(masqueAuthority, ":")[0] + tmpl := &x509.Certificate{SerialNumber: big.NewInt(2), Subject: pkix.Name{CommonName: host}, + DNSNames: []string{host}, NotBefore: time.Now().Add(-time.Hour), NotAfter: time.Now().Add(24 * time.Hour), + KeyUsage: x509.KeyUsageDigitalSignature, ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth}} + der, err := x509.CreateCertificate(rand.Reader, tmpl, caCert, &key.PublicKey, caKey) + if err != nil { + t.Fatal(err) + } + keyDER, err := x509.MarshalPKCS8PrivateKey(key) + if err != nil { + t.Fatal(err) + } + return pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: der}), + pem.EncodeToMemory(&pem.Block{Type: "PRIVATE KEY", Bytes: keyDER}), + pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: caDER}) +} + +func (m *masque) setup(t *testing.T) { + l := m.l + for _, mt := range []*masqueTenant{m.a, m.b} { + for i := 1; i <= 2; i++ { + p := m.client(t, mt.tn, i) + m.unrelated[p.Namespace+"/"+p.Name] = p.UID + } + } + l.ensureNamespace(t, masqueNS) + l.applyOverlayOf(t, filepath.Join(l.root, "tests", "edgeconformance", "masqueprovider", "deploy", "registration"), + "masque-registration", "", nil) + if _, err := l.kubectl("patch", "edgeprovider", masqueProvider, "--subresource=status", "--type=merge", + "-p", `{"status":{"approved":true}}`); err != nil { + t.Fatal(err) + } + for _, mt := range []*masqueTenant{m.a, m.b} { + mt.vpcUID = l.vpcUID(t, mt.tn) + mt.svc = l.grantService(t, mt.tn, masqueService, func(s *galacticv1alpha1.EdgeService) { + s.Spec.Capabilities = []string{"PacketAccess", "PrefixReachability"} + s.Spec.Prefixes = []string{masqueGrant} + s.Spec.ClientPool = masquePool + }) + l.waitServiceAdmitted(t, mt.svc, true) + } + m.pod = m.deploy(t, "m1", edgeNode1, m.config(true)) + m.waitServing(t, m.pod) + m.waitConnected(t, m.a) + m.waitConnected(t, m.b) +} + +// fixtureConfig, fixtureService and fixtureConnector are the fixture's +// config.yaml (tests/edgeconformance/masqueprovider/config.go). +type fixtureConfig struct { + Provider string `json:"provider"` + Authority string `json:"authority"` + Listen string `json:"listen"` + Services []fixtureService `json:"services"` + Connectors []fixtureConnector `json:"connectors"` +} + +type fixtureService struct { + Namespace string `json:"namespace"` + Name string `json:"name"` + UID string `json:"uid"` + VPCUID string `json:"vpcUID"` + VPC string `json:"vpc"` +} + +type fixtureConnector struct { + ID string `json:"id"` + Credential string `json:"credential"` + Service string `json:"service"` + Prefix string `json:"prefix"` + Routes []string `json:"routes,omitempty"` +} + +// config is the fixture's config.yaml: both tenants' services and +// connectors, plus, when withRefused is set, a VPC 10 connector whose /96 +// lies outside the grant, so core refuses its route. +func (m *masque) config(withRefused bool) string { + svcs := make([]fixtureService, 0, 2) + conns := make([]fixtureConnector, 0, 3) + for _, mt := range []*masqueTenant{m.a, m.b} { + svcs = append(svcs, mt.serviceConfig()) + conns = append(conns, fixtureConnector{ID: "connector-" + mt.tn.vpc, Credential: mt.token, + Service: mt.serviceKey(), Prefix: masquePrefix, + Routes: []string{"fd20:" + mt.tn.vpc + "::/32", remoteHost + "/128"}}) + } + if withRefused { + conns = append(conns, fixtureConnector{ID: "connector-outside", Credential: m.refused, + Service: m.a.serviceKey(), Prefix: masqueOutside}) + } + return marshalConfig(svcs, conns) +} + +// serviceConfig is mt's grant as the fixture's config.yaml lists it. +func (mt *masqueTenant) serviceConfig() fixtureService { + return fixtureService{Namespace: mt.svc.Namespace, Name: mt.svc.Name, UID: string(mt.svc.UID), + VPCUID: mt.vpcUID, VPC: mt.tn.vpc} +} + +// serviceKey is how a fixture connector names mt's service. +func (mt *masqueTenant) serviceKey() string { return mt.svc.Namespace + "/" + mt.svc.Name } + +func marshalConfig(svcs []fixtureService, conns []fixtureConnector) string { + b, _ := json.Marshal(fixtureConfig{Provider: masqueProvider, Authority: masqueAuthority, + Listen: ":" + strconv.Itoa(masquePort), Services: svcs, Connectors: conns}) + return string(b) +} + +// deploy runs the fixture's Deployment name on node with cfg and returns its +// running pod. +func (m *masque) deploy(t *testing.T, name, node, cfg string) *corev1.Pod { + t.Helper() + l := m.l + secret := &corev1.Secret{ + ObjectMeta: metav1.ObjectMeta{Namespace: masqueNS, Name: "masque-config-" + name}, + Data: map[string][]byte{"config.yaml": []byte(cfg), "tls.crt": m.certPEM, "tls.key": m.keyPEM}, + } + l.deleteAndWait(t, &corev1.Secret{ObjectMeta: secret.ObjectMeta}) + if err := l.c.Create(context.Background(), secret); err != nil { + t.Fatal(err) + } + l.applyOverlayOf(t, filepath.Join(l.root, "tests", "edgeconformance", "masqueprovider", "deploy", "provider"), + "masque-"+name, fmt.Sprintf(` +namespace: %s +patches: + - target: + kind: Deployment + options: + allowNameChange: true + patch: |- + - op: replace + path: /metadata/name + value: masque-provider-%s + - op: replace + path: /spec/template/spec/volumes/0/secret/secretName + value: masque-config-%s + - op: add + path: /spec/template/spec/nodeSelector + value: {kubernetes.io/hostname: %s} + - op: add + path: /spec/template/spec/tolerations + value: [{key: galactic.datumapis.com/nat, operator: Exists, effect: NoSchedule}] + - op: add + path: /metadata/labels/edge-conformance~1instance + value: %s + - op: add + path: /spec/template/metadata/labels/edge-conformance~1instance + value: %s + - op: add + path: /spec/selector/matchLabels/edge-conformance~1instance + value: %s +`, masqueNS, name, name, node, name, name, name), nil) + return l.waitProviderPod(t, masqueNS, name, node) +} + +// waitServing waits until pod's attachments to both services are ready and +// both tenants' routes are accepted for them, and records their host links. +func (m *masque) waitServing(t *testing.T, pod *corev1.Pod) { + t.Helper() + for _, mt := range []*masqueTenant{m.a, m.b} { + att := m.l.waitServing(t, pod, mt.svc) + m.links[att.Name] = att.Status.HostLink + m.l.waitAccepted(t, masqueNS, mt.vpcUID, netip.MustParsePrefix(masquePrefix), att.Name, 3*time.Minute) + } +} + +// waitConnected waits until mt's connector is admitted and its tenant's +// workload on the provider's node answers through it. +func (m *masque) waitConnected(t *testing.T, mt *masqueTenant) { + t.Helper() + dst := m.workload(t, mt.tn, 2).addr + eventually(t, 2*time.Minute, 2*time.Second, "connector of VPC "+mt.tn.vpc+" answered", func() (bool, error) { + evs, err := m.connectOnce(mt.token, mt.tn.vpc, mt.id, time.Second, dst) + if err != nil { + return false, err + } + return len(echoes(evs, dst)) > 0, nil + }) +} + +// connect runs one CONNECT-IP session to the fixture's pod from its node's +// host network, pinging dsts. The client runs on the pod's node because the +// lab's pod network does not carry UDP from a node to a pod on another node; +// where the client runs does not change the gVPC path behind the fixture. +func (m *masque) connect(t *testing.T, name, token, vpc string, id int, d time.Duration, dsts ...string) *probeRun { + t.Helper() + p, err := startProbe(m.pod.Spec.NodeName, 1, name, m.connectArgs(token, vpc, id, d, dsts)...) + if err != nil { + t.Fatal(err) + } + return p +} + +func (m *masque) connectArgs(token, vpc string, id int, d time.Duration, dsts []string) []string { + return []string{"connectip", "-server", "[" + m.pod.Status.PodIP + "]:" + strconv.Itoa(masquePort), + "-authority", masqueAuthority, "-ca", masqueCA, "-token", token, "-vpc", vpc, "-id", strconv.Itoa(id), + "-to", strings.Join(dsts, ","), "-interval", probeInterval.String(), "-duration", d.String()} +} + +// connectOnce runs one short session and returns its events. +func (m *masque) connectOnce(token, vpc string, id int, d time.Duration, dsts ...string) ([]event, error) { + p, err := startProbe(m.pod.Spec.NodeName, 1, "once", m.connectArgs(token, vpc, id, d, dsts)...) + if err != nil { + return nil, err + } + if err := p.cmd.Wait(); err != nil { + return nil, fmt.Errorf("connectip: %w: %s", err, p.out.String()) + } + var evs []event + for _, line := range strings.Split(strings.TrimSpace(p.out.String()), "\n") { + var e event + if line == "" { + continue + } + if err := json.Unmarshal([]byte(line), &e); err != nil { + return nil, err + } + evs = append(evs, e) + } + return evs, nil +} + +// echoes returns the echo replies from dst. +func echoes(evs []event, dst string) []event { + var out []event + for _, e := range evs { + if e.Reply == "echo" && e.From == dst { + out = append(out, e) + } + } + return out +} + +// others counts the packets the tunnel delivered that were not echo replies +// to this client. +func others(evs []event) int { + n := 0 + for _, e := range evs { + if e.Reply == "other" { + n++ + } + } + return n +} + +// requireSession fails unless evs is a request that was accepted with the +// connector /96 and sent through its whole window of d to n destinations. A +// refused request, or a tunnel the server closed early, sends fewer than 90% +// of the probes the window holds, so its reply ratio alone proves nothing. +func requireSession(t *testing.T, what string, evs []event, d time.Duration, n int) { + t.Helper() + if s := status(evs); s != 200 { + t.Errorf("%s: status %d, want 200", what, s) + } + assigned := "" + for _, e := range evs { + if e.Assigned != "" { + assigned = e.Assigned + } + } + if assigned != masquePrefix { + t.Errorf("%s: assigned %q, want %s", what, assigned, masquePrefix) + } + if want, got := int(d/probeInterval)*n, sentCount(evs); got < want*9/10 { + t.Errorf("%s: sent %d of the window's %d probes", what, got, want) + } +} + +func status(evs []event) int { + for _, e := range evs { + if e.Status != 0 { + return e.Status + } + } + return 0 +} + +// workloadRef is a tenant workload: its node, sandbox, link and address. +type workloadRef struct { + node string + pid int + iface string + addr string +} + +// workload returns tenant tn's edge client i: 1 on the compute node, 2 on +// the fixture's node. +func (m *masque) workload(t *testing.T, tn tenant, i int) workloadRef { + t.Helper() + node := map[int]string{1: computeNode, 2: edgeNode1}[i] + pid := m.l.clientPID(t, tn, node) + out, err := nodeExec(node, "nsenter", "-t", strconv.Itoa(pid), "-n", "ip", "-6", "-o", "addr", "show", + "scope", "global") + if err != nil { + t.Fatal(err) + } + for _, line := range strings.Split(out, "\n") { + f := strings.Fields(line) + if len(f) >= 4 && strings.HasPrefix(f[3], "fd20:"+tn.vpc+":") { + return workloadRef{node: node, pid: pid, iface: f[1], addr: strings.Split(f[3], "/")[0]} + } + } + t.Fatalf("no VPC %s address on edge-client-%d: %s", tn.vpc, i, out) + return workloadRef{} +} + +// client returns tenant tn's running edge-client-i pod. +func (m *masque) client(t *testing.T, tn tenant, i int) *corev1.Pod { + t.Helper() + var pods corev1.PodList + if err := m.l.c.List(context.Background(), &pods, client.InNamespace(tn.vpcNamespace), + client.MatchingLabels{"app": "edge-client-" + strconv.Itoa(i)}); err != nil { + t.Fatal(err) + } + for i := range pods.Items { + if p := &pods.Items[i]; p.Status.Phase == corev1.PodRunning && p.DeletionTimestamp == nil { + return p + } + } + t.Fatalf("no running edge-client-%d in %s", i, tn.vpcNamespace) + return nil +} + +func (m *masque) rowUnprivileged(t *testing.T) { + l, pod := m.l, m.pod + spec := pod.Spec + if spec.HostNetwork || spec.HostPID || spec.HostIPC { + t.Errorf("fixture pod shares a host namespace: network=%t pid=%t ipc=%t", spec.HostNetwork, spec.HostPID, + spec.HostIPC) + } + for _, v := range spec.Volumes { + if v.HostPath != nil { + t.Errorf("fixture pod mounts hostPath %s", v.HostPath.Path) + } + } + for _, c := range spec.Containers { + sc := c.SecurityContext + if sc == nil || sc.Capabilities == nil || !slices.Contains(sc.Capabilities.Drop, "ALL") || + len(sc.Capabilities.Add) > 0 || (sc.Privileged != nil && *sc.Privileged) { + t.Errorf("container %s is not unprivileged: %+v", c.Name, sc) + } + } + pid, err := containerPID(edgeNode1, pod.Namespace, pod.Name, "masque-provider") + if err != nil { + t.Fatal(err) + } + st, err := nodeExec(edgeNode1, "grep", "-E", "^(Cap(Eff|Prm|Bnd)|Uid|Gid):", "/proc/"+strconv.Itoa(pid)+"/status") + if err != nil { + t.Fatal(err) + } + l.save(t, "process-status.txt", []byte(st)) + for _, line := range strings.Split(st, "\n") { + if f := strings.Fields(line); len(f) == 2 && (f[0] == "CapEff:" || f[0] == "CapPrm:") && + f[1] != "0000000000000000" { + t.Errorf("fixture process holds capabilities: %s", line) + } + } + l.record(t, "fixture process", strings.ReplaceAll(st, "\n", "; ")) + + m.checkPacketDevices(t) + m.checkAdmission(t) + m.recordVersions(t) +} + +// checkPacketDevices checks that galactic-edge created each packet device +// in the fixture's pod, owned by the fixture's UID, and that the fixture +// opened it and reported Ready. +func (m *masque) checkPacketDevices(t *testing.T) { + t.Helper() + l, pod := m.l, m.pod + sandbox := podPID(t, edgeNode1, pod.Namespace, pod.Name) + for _, mt := range []*masqueTenant{m.a, m.b} { + att := l.attachmentOf(t, pod, mt.svc) + dev, err := nodeExec(edgeNode1, "nsenter", "-t", strconv.Itoa(sandbox), "-n", "ip", "-d", "link", "show", + "dev", att.Status.PacketDevice) + if err != nil { + t.Fatal(err) + } + l.record(t, "VPC "+mt.tn.vpc+" packet device", strings.Join(strings.Fields(dev), " ")) + if !strings.Contains(dev, "tun type tun") || !strings.Contains(dev, "user 65532 group 65532") { + t.Errorf("VPC %s packet device %s is not a TUN device owned by 65532: %s", mt.tn.vpc, + att.Status.PacketDevice, dev) + } + c := meta.FindStatusCondition(att.Status.Conditions, galacticv1alpha1.EdgeAttachmentReady) + if c == nil || c.Status != metav1.ConditionTrue || c.Reason != "Serving" { + t.Errorf("VPC %s attachment not served by the fixture: %+v", mt.tn.vpc, c) + } + } + +} + +// checkAdmission checks that admission refuses an unknown credential, a +// credential presented for the other VPC, and a connector whose route core +// refused. +func (m *masque) checkAdmission(t *testing.T) { + t.Helper() + l := m.l + dst := m.workload(t, m.a.tn, 2).addr + att := l.attachmentOf(t, m.pod, m.a.svc) + l.record(t, "outside-grant route", m.waitRefused(t, m.a.vpcUID, netip.MustParsePrefix(masqueOutside), att.Name)) + for _, c := range []struct { + what, token, vpc string + want int + }{ + {"unknown credential", rand.Text(), "10", 401}, + {"VPC 10 credential for VPC 20", m.a.token, "20", 403}, + {"connector with a refused route", m.refused, "10", 503}, + } { + evs, err := m.connectOnce(c.token, c.vpc, 1, time.Second, dst) + if err != nil { + t.Fatal(err) + } + l.record(t, c.what, status(evs)) + if got := status(evs); got != c.want || len(echoes(evs, dst)) > 0 { + t.Errorf("%s: status %d with %d replies, want %d and none", c.what, got, len(echoes(evs, dst)), c.want) + } + } +} + +// recordVersions saves what the run executed against: Kubernetes, the +// nodes' kernels and runtimes, and the image of every component the rows +// exercise. +func (m *masque) recordVersions(t *testing.T) { + t.Helper() + l := m.l + v := map[string]string{} + if out, err := l.kubectl("version", "-o", "json"); err == nil { + v["kubernetes"] = strings.Join(strings.Fields(out), " ") + } + var nodes corev1.NodeList + if err := l.c.List(context.Background(), &nodes); err != nil { + t.Fatal(err) + } + for _, n := range nodes.Items { + v["node "+n.Name] = n.Status.NodeInfo.KernelVersion + ", " + n.Status.NodeInfo.ContainerRuntimeVersion + + ", " + n.Status.NodeInfo.KubeletVersion + } + var pods corev1.PodList + for _, ns := range []string{systemNS, "kube-system", masqueNS} { + if err := l.c.List(context.Background(), &pods, client.InNamespace(ns)); err != nil { + t.Fatal(err) + } + for _, p := range pods.Items { + if p.Spec.NodeName != edgeNode1 && p.Spec.NodeName != computeNode { + continue + } + for _, cs := range p.Status.ContainerStatuses { + v["image "+cs.Image] = cs.ImageID + } + } + } + b, _ := json.MarshalIndent(v, "", " ") + l.save(t, "versions.json", b) + for k, s := range v { + l.record(t, k, s) + } +} + +// isolation sends from each tenant's connector, at once, to its own +// workloads on the compute node and on the fixture's node and to the other +// tenant's, with neighbor caches flushed first. Each tenant's own workloads +// answer; nothing reaches the other tenant's workloads, whose links are +// captured, and no client receives a packet that is not its own reply. +func (m *masque) isolation(t *testing.T, label string) { + l := m.l + wl := map[*masqueTenant][]workloadRef{} + for _, mt := range []*masqueTenant{m.a, m.b} { + wl[mt] = []workloadRef{m.workload(t, mt.tn, 1), m.workload(t, mt.tn, 2)} + for _, w := range wl[mt] { + if _, err := nodeExec(w.node, "nsenter", "-t", strconv.Itoa(w.pid), "-n", "ip", "-6", "neigh", "flush", + "all"); err != nil { + t.Fatal(err) + } + } + } + for _, link := range m.currentLinks(t) { + if _, err := nodeExec(edgeNode1, "ip", "-6", "neigh", "flush", "dev", link); err != nil { + t.Fatal(err) + } + } + type sniffer struct { + w workloadRef + mt *masqueTenant + p *probeRun + } + var sniffs []sniffer + for _, mt := range []*masqueTenant{m.a, m.b} { + for _, w := range wl[mt] { + p, err := startProbe(w.node, w.pid, label+"-sniff-"+mt.tn.vpc+"-"+w.node, "sniff", "-iface", w.iface, + "-icmp", "-duration", "9s") + if err != nil { + t.Fatal(err) + } + sniffs = append(sniffs, sniffer{w, mt, p}) + } + } + time.Sleep(time.Second) + runs := map[*masqueTenant]*probeRun{} + for _, mt := range []*masqueTenant{m.a, m.b} { + other := m.b + if mt == m.b { + other = m.a + } + dsts := []string{wl[mt][0].addr, wl[mt][1].addr, wl[other][0].addr, wl[other][1].addr} + runs[mt] = m.connect(t, label+"-vpc-"+mt.tn.vpc, mt.token, mt.tn.vpc, mt.id, 5*time.Second, dsts...) + } + for _, mt := range []*masqueTenant{m.a, m.b} { + other := m.b + if mt == m.b { + other = m.a + } + evs := runs[mt].wait(l, t) + requireSession(t, "VPC "+mt.tn.vpc+"'s request", evs, 5*time.Second, 4) + perDst := sentCount(evs) / 4 + for _, w := range wl[mt] { + n := len(echoes(evs, w.addr)) + l.record(t, fmt.Sprintf("VPC %s to its workload on %s", mt.tn.vpc, w.node), fmt.Sprintf("%d/%d", n, perDst)) + if perDst == 0 || n < perDst*9/10 { + t.Errorf("VPC %s to its workload on %s: %d of %d answered", mt.tn.vpc, w.node, n, perDst) + } + } + for _, w := range wl[other] { + n := len(echoes(evs, w.addr)) + l.record(t, fmt.Sprintf("VPC %s to VPC %s's workload on %s", mt.tn.vpc, other.tn.vpc, w.node), n) + if n > 0 { + t.Errorf("VPC %s reached VPC %s's workload on %s: %d replies", mt.tn.vpc, other.tn.vpc, w.node, n) + } + } + if n := others(evs); n > 0 { + t.Errorf("VPC %s's client received %d packets that were not its replies", mt.tn.vpc, n) + } + } + for _, s := range sniffs { + evs := s.p.wait(l, t) + own, foreign := 0, 0 + for _, e := range evs { + switch e.ID { + case s.mt.id: + own++ + case m.a.id, m.b.id: + foreign++ + } + } + l.record(t, fmt.Sprintf("echo requests on VPC %s's workload on %s", s.mt.tn.vpc, s.w.node), + fmt.Sprintf("%d own, %d from the other VPC's client", own, foreign)) + if foreign > 0 { + t.Errorf("VPC %s's workload on %s received %d of the other VPC's echo requests", s.mt.tn.vpc, s.w.node, + foreign) + } + if own < int(5*time.Second/probeInterval)/2 { + t.Errorf("capture on VPC %s's workload on %s saw only %d of its own tenant's requests", s.mt.tn.vpc, + s.w.node, own) + } + } +} + +func (m *masque) rowNATEgress(t *testing.T) { + runs := map[*masqueTenant]*probeRun{} + for _, mt := range []*masqueTenant{m.a, m.b} { + runs[mt] = m.connect(t, "nat-vpc-"+mt.tn.vpc, mt.token, mt.tn.vpc, mt.id, 5*time.Second, remoteHost) + } + for mt, p := range runs { + evs := p.wait(m.l, t) + requireSession(t, "VPC "+mt.tn.vpc+"'s request", evs, 5*time.Second, 1) + n := len(echoes(evs, remoteHost)) + m.l.record(t, "VPC "+mt.tn.vpc+" to the off-fabric host", fmt.Sprintf("%d/%d", n, sentCount(evs))) + if n < sentCount(evs)*9/10 || others(evs) > 0 { + t.Errorf("VPC %s to the off-fabric host: %d of %d answered, %d other packets", mt.tn.vpc, n, + sentCount(evs), others(evs)) + } + } +} + +// rowPartial starts a second instance whose setup cannot complete, its only +// connector's route refused, kills it without shutdown, and checks that +// core reaps everything it left while the first instance keeps forwarding. +func (m *masque) rowPartial(t *testing.T) { + l := m.l + dst := m.workload(t, m.a.tn, 2).addr + b := marshalConfig([]fixtureService{m.a.serviceConfig()}, []fixtureConnector{{ID: "connector-partial", + Credential: rand.Text(), Service: m.a.serviceKey(), Prefix: masquePartial}}) + pod := m.deploy(t, "partial", edgeNode1, b) + att := l.waitServing(t, pod, m.a.svc) + m.links[att.Name] = att.Status.HostLink + route := m.waitRefused(t, m.a.vpcUID, netip.MustParsePrefix(masquePartial), att.Name) + l.record(t, "partial instance's route", route) + + p := m.connect(t, "partial-continuity", m.a.token, "10", m.a.id, 20*time.Second, dst) + time.Sleep(3 * time.Second) + // Orphaning the Deployment and its ReplicaSet leaves nothing to replace + // the pod, which is then killed without running its shutdown. + for _, kind := range []string{"deployment", "replicaset"} { + if _, err := l.kubectl("-n", masqueNS, "delete", kind, "-l", "edge-conformance/instance=partial", + "--cascade=orphan"); err != nil { + t.Fatal(err) + } + } + if _, err := l.kubectl("-n", masqueNS, "delete", "pod", pod.Name, "--grace-period=0", "--force"); err != nil { + t.Fatal(err) + } + killed := time.Now() + evs := p.wait(l, t) + requireSession(t, "first instance's request", evs, 20*time.Second, 1) + rs := echoes(evs, dst) + l.record(t, "first instance across the kill", fmt.Sprintf("%d/%d, largest gap %v", len(rs), sentCount(evs), + largestGap(rs))) + if len(rs) < sentCount(evs)*9/10 || largestGap(rs) > time.Second { + t.Errorf("first instance answered %d of %d with a %v gap across the kill", len(rs), sentCount(evs), + largestGap(rs)) + } + m.waitReaped(t, pod.UID, []*galacticv1alpha1.EdgeAttachment{att}, killed) + m.waitNoRoute(t, m.a.vpcUID, netip.MustParsePrefix(masquePartial)) + m.waitAcceptedFor(t, 20*time.Second) +} + +// waitRefused waits for core to refuse the claim on p and returns the +// refusal's reason. +func (m *masque) waitRefused(t *testing.T, vpcUID string, p netip.Prefix, attachment string) string { + t.Helper() + var reason string + eventually(t, time.Minute, time.Second, "refusal of "+p.String(), func() (bool, error) { + r := m.l.route(t, masqueNS, vpcUID, p) + if r == nil || r.Spec.Claim == nil || r.Spec.Claim.Attachment != attachment { + return false, nil + } + c := meta.FindStatusCondition(r.Status.Conditions, galacticv1alpha1.EdgeRouteClaimed) + if c == nil || c.Status != metav1.ConditionFalse || c.Reason == "WaitingForFencing" { + return false, nil + } + reason = c.Reason + return true, nil + }) + return reason +} + +// waitReaped waits until core has deleted atts and the instance and lease of +// podUID, a pod that no longer exists, and the node has removed each +// attachment's host link. It records each time from since. +func (m *masque) waitReaped(t *testing.T, podUID types.UID, atts []*galacticv1alpha1.EdgeAttachment, + since time.Time) { + t.Helper() + l, ctx := m.l, context.Background() + gone := func(obj client.Object) { + eventually(t, 3*time.Minute, time.Second, fmt.Sprintf("%T %s reaped", obj, obj.GetName()), func() (bool, error) { + err := l.c.Get(ctx, client.ObjectKeyFromObject(obj), obj) + return apierrors.IsNotFound(err), client.IgnoreNotFound(err) + }) + l.record(t, fmt.Sprintf("%T %s reaped after", obj, obj.GetName()), time.Since(since).Round(time.Second)) + } + for _, att := range atts { + gone(&galacticv1alpha1.EdgeAttachment{ObjectMeta: metav1.ObjectMeta{Namespace: masqueNS, Name: att.Name}}) + deleted := time.Now() + eventually(t, time.Minute, time.Second, "host link "+att.Status.HostLink+" removed", func() (bool, error) { + present, err := linkPresent(edgeNode1, att.Status.HostLink) + return err == nil && !present, err + }) + l.record(t, "host link "+att.Status.HostLink+" removed after its attachment", + time.Since(deleted).Round(time.Second)) + } + inst := "ei-" + string(podUID) + gone(&galacticv1alpha1.EdgeInstance{ObjectMeta: metav1.ObjectMeta{Namespace: masqueNS, Name: inst}}) + gone(&coordinationv1.Lease{ObjectMeta: metav1.ObjectMeta{Namespace: masqueNS, Name: inst}}) +} + +// waitNoRoute waits until core has reaped the route of p. +func (m *masque) waitNoRoute(t *testing.T, vpcUID string, p netip.Prefix) { + t.Helper() + eventually(t, 3*time.Minute, time.Second, "route "+p.String()+" reaped", func() (bool, error) { + return m.l.route(t, masqueNS, vpcUID, p) == nil, nil + }) +} + +// waitAcceptedFor checks that both tenants' routes stay accepted for the +// current pod's attachments for d. +func (m *masque) waitAcceptedFor(t *testing.T, d time.Duration) { + t.Helper() + deadline := time.Now().Add(d) + for time.Now().Before(deadline) { + for _, mt := range []*masqueTenant{m.a, m.b} { + att := m.l.attachmentOf(t, m.pod, mt.svc) + r := m.l.route(t, masqueNS, mt.vpcUID, netip.MustParsePrefix(masquePrefix)) + if att == nil || r == nil || r.Status.Accepted == nil || r.Status.Accepted.Attachment != att.Name || + !meta.IsStatusConditionTrue(r.Status.Conditions, galacticv1alpha1.EdgeRouteAccepted) { + t.Fatalf("VPC %s's route is no longer accepted for the current attachment", mt.tn.vpc) + } + } + time.Sleep(2 * time.Second) + } +} + +// rowRecreation deletes the fixture's pod, lets its Deployment replace it, +// and checks the replacement serves both tenants in isolation, the old +// pod's attachments, instance, lease and host links are reaped, and the +// old pod's shutdown leaves the replacement's routes alone. +func (m *masque) rowRecreation(t *testing.T) { + l := m.l + old := m.pod + oldAtts := make([]*galacticv1alpha1.EdgeAttachment, 0, 2) + for _, mt := range []*masqueTenant{m.a, m.b} { + oldAtts = append(oldAtts, l.attachmentOf(t, old, mt.svc)) + } + if _, err := l.kubectl("-n", masqueNS, "delete", "pod", old.Name, "--wait=false"); err != nil { + t.Fatal(err) + } + deleted := time.Now() + eventually(t, 3*time.Minute, time.Second, "replacement pod", func() (bool, error) { + p := l.providerPod(t, masqueNS, "m1") + if p == nil || p.UID == old.UID { + return false, nil + } + m.pod = p + return true, nil + }) + m.waitServing(t, m.pod) + m.waitConnected(t, m.a) + m.waitConnected(t, m.b) + l.record(t, "replacement serving after", time.Since(deleted).Round(time.Second)) + m.waitReaped(t, old.UID, oldAtts, deleted) + m.waitAcceptedFor(t, 20*time.Second) + m.isolation(t, "after-recreation") +} + +// currentLinks returns the host links of the current pod's attachments. +func (m *masque) currentLinks(t *testing.T) []string { + t.Helper() + var out []string + for _, mt := range []*masqueTenant{m.a, m.b} { + if att := m.l.attachmentOf(t, m.pod, mt.svc); att != nil && att.Status.HostLink != "" { + m.links[att.Name] = att.Status.HostLink + out = append(out, att.Status.HostLink) + } + } + return out +} + +// linkPresent reports whether node has a link named name. +func linkPresent(node, name string) (bool, error) { + out, err := nodeExec(node, "ip", "-o", "link", "show") + if err != nil { + return false, err + } + for _, line := range strings.Split(out, "\n") { + if f := strings.Fields(line); len(f) > 1 && strings.Split(strings.TrimSuffix(f[1], ":"), "@")[0] == name { + return true, nil + } + } + return false, nil +} + +// waitNamespaceEmpty waits until no EdgeAttachment, EdgeInstance, Lease or +// EdgeRoute remains in the fixture's namespace. +func (m *masque) waitNamespaceEmpty(t *testing.T) { + t.Helper() + ctx := context.Background() + eventually(t, 3*time.Minute, 2*time.Second, "fixture objects removed", func() (bool, error) { + left := 0 + for _, list := range []client.ObjectList{&galacticv1alpha1.EdgeAttachmentList{}, + &galacticv1alpha1.EdgeInstanceList{}, &coordinationv1.LeaseList{}, &galacticv1alpha1.EdgeRouteList{}} { + if err := m.l.c.List(ctx, list, client.InNamespace(masqueNS)); err != nil { + return false, err + } + left += meta.LenList(list) + } + return left == 0, fmt.Errorf("%d objects remain", left) + }) +} + +// teardown removes the fixture and fails the run unless everything it +// created is gone and everything it did not create is untouched. +func (m *masque) teardown(t *testing.T) { + l, ctx := m.l, context.Background() + if m.pod != nil { + m.currentLinks(t) + } + if _, err := l.kubectl("-n", masqueNS, "delete", "deployment,replicaset", "--all", "--wait=true", + "--timeout=120s"); err != nil { + t.Error(err) + } + eventually(t, 2*time.Minute, time.Second, "fixture pods gone", func() (bool, error) { + var pods corev1.PodList + err := l.c.List(ctx, &pods, client.InNamespace(masqueNS)) + return err == nil && len(pods.Items) == 0, err + }) + for _, mt := range []*masqueTenant{m.a, m.b} { + l.deleteAndWait(t, &galacticv1alpha1.EdgeService{ObjectMeta: metav1.ObjectMeta{Namespace: mt.tn.vpcNamespace, + Name: masqueService}}) + } + // The provider and core must remove what the run created before the + // namespace's deletion would remove it for them. + m.waitNamespaceEmpty(t) + if _, err := l.kubectl("delete", "namespace", masqueNS, "--wait=true", "--timeout=180s"); err != nil { + t.Error(err) + } + l.deleteAndWait(t, &galacticv1alpha1.EdgeProvider{ObjectMeta: metav1.ObjectMeta{Name: masqueProvider}}) + + var advs networkv1alpha1.BGPAdvertisementList + if err := l.c.List(ctx, &advs, client.InNamespace(systemNS)); err != nil { + t.Error(err) + } + for _, a := range advs.Items { + if strings.HasPrefix(a.Annotations["galactic.datumapis.com/edge-route"], masqueNS+"/") { + t.Errorf("BGPAdvertisement %s of the fixture's route %s remains", a.Name, + a.Annotations["galactic.datumapis.com/edge-route"]) + } + } + for att, link := range m.links { + for _, n := range []string{edgeNode1, edgeNode2} { + if present, err := linkPresent(n, link); err != nil || present { + t.Errorf("host link %s of attachment %s on %s: present=%t, %v", link, att, n, present, err) + } + } + } + for _, n := range []string{computeNode, edgeNode1, edgeNode2} { + if _, err := nodeExec(n, "rm", "-f", masqueCA); err != nil { + t.Error(err) + } + } + for _, mt := range []*masqueTenant{m.a, m.b} { + for i := 1; i <= 2; i++ { + p := m.client(t, mt.tn, i) + if want := m.unrelated[p.Namespace+"/"+p.Name]; want != p.UID { + t.Errorf("unrelated pod %s/%s was replaced", p.Namespace, p.Name) + } + } + c1, w2 := m.client(t, mt.tn, 1), m.workload(t, mt.tn, 2) + if out, err := l.kubectl("-n", c1.Namespace, "exec", c1.Name, "--", "ping", "-6", "-c", "3", "-W", "2", + w2.addr); err != nil { + t.Errorf("VPC %s's workloads no longer reach each other after teardown: %v: %s", mt.tn.vpc, err, out) + } + } +} diff --git a/tests/edgeconformance/masqueprovider/Dockerfile b/tests/edgeconformance/masqueprovider/Dockerfile new file mode 100644 index 00000000..565f4fb7 --- /dev/null +++ b/tests/edgeconformance/masqueprovider/Dockerfile @@ -0,0 +1,21 @@ +# Build the conformance suite's MASQUE fixture provider. It is a test +# fixture, not a product image: the lab loads it into its edge nodes. +FROM --platform=$BUILDPLATFORM golang:1.26 AS builder +ARG TARGETOS +ARG TARGETARCH + +WORKDIR /workspace +COPY go.mod go.sum ./ +RUN go mod download +COPY api/ api/ +COPY internal/ internal/ +COPY tests/edgeconformance/masqueprovider/ tests/edgeconformance/masqueprovider/ +RUN CGO_ENABLED=0 GOOS=${TARGETOS:-linux} GOARCH=${TARGETARCH} go build -ldflags "-s -w" \ + -o /masqueprovider ./tests/edgeconformance/masqueprovider + +# The fixture needs no shell or host tool. It runs unprivileged and opens +# its packet devices as the owner the attachment names, 65532:65532. +FROM scratch +COPY --from=builder /masqueprovider /masqueprovider +USER 65532:65532 +ENTRYPOINT ["/masqueprovider"] diff --git a/tests/edgeconformance/masqueprovider/config.go b/tests/edgeconformance/masqueprovider/config.go new file mode 100644 index 00000000..d539a566 --- /dev/null +++ b/tests/edgeconformance/masqueprovider/config.go @@ -0,0 +1,115 @@ +// Copyright 2026 Datum Cloud, Inc. +// +// SPDX-License-Identifier: AGPL-3.0-or-later + +package main + +import ( + "errors" + "fmt" + "net/netip" + "os" + + "sigs.k8s.io/yaml" +) + +// Config is the fixture's configuration file. +type Config struct { + // Provider is the EdgeProvider this pod runs as. + Provider string `json:"provider"` + // Authority is the host[:port] clients send as :authority. + Authority string `json:"authority"` + // Listen is the UDP address the HTTP/3 server listens on. + Listen string `json:"listen"` + // Services are the EdgeServices the pod attaches to, one PacketAccess + // attachment each. + Services []Service `json:"services"` + // Connectors are the CONNECT-IP clients the fixture admits. + Connectors []Connector `json:"connectors"` +} + +// Service is one EdgeService grant. A provider cannot read an EdgeService, +// so the deployer copies its UID and its VPC's UID from the grant. +type Service struct { + Namespace string `json:"namespace"` + Name string `json:"name"` + UID string `json:"uid"` + VPCUID string `json:"vpcUID"` + // VPC is the Galactic-VPC header value that selects this service. + VPC string `json:"vpc"` +} + +// Connector is one test client: a static bearer credential bound to one +// service and one /96. +type Connector struct { + // ID is the connector's immutable identity. + ID string `json:"id"` + // Credential is the bearer token the client presents. It exists only + // for this fixture's test connectors. + Credential string `json:"credential"` + // Service is the namespace/name of the entry of Services the + // connector belongs to. + Service string `json:"service"` + // Prefix is the connector's /96, inside its service's grant. + Prefix netip.Prefix `json:"prefix"` + // Routes are the destinations advertised to the client. + Routes []netip.Prefix `json:"routes"` +} + +// Key is how a connector names the service: its namespace and name. +func (s Service) Key() string { return s.Namespace + "/" + s.Name } + +// LoadConfig reads and validates the configuration file at path. +func LoadConfig(path string) (*Config, error) { + b, err := os.ReadFile(path) + if err != nil { + return nil, err + } + var c Config + if err := yaml.UnmarshalStrict(b, &c); err != nil { + return nil, fmt.Errorf("parse %s: %w", path, err) + } + return &c, c.Validate() +} + +// Validate reports the first invalid field. +func (c *Config) Validate() error { + if c.Provider == "" || c.Authority == "" || c.Listen == "" { + return errors.New("provider, authority and listen are required") + } + services := map[string]bool{} + for _, s := range c.Services { + if s.Namespace == "" || s.Name == "" || s.UID == "" || s.VPCUID == "" || s.VPC == "" { + return fmt.Errorf("service %q: namespace, name, uid, vpcUID and vpc are required", s.Name) + } + if services[s.Key()] { + return fmt.Errorf("service %q is listed twice", s.Key()) + } + services[s.Key()] = true + } + ids, creds := map[string]bool{}, map[string]bool{} + prefixes := map[string]bool{} + for _, cn := range c.Connectors { + switch { + case cn.ID == "" || cn.Credential == "": + return errors.New("every connector needs an id and a credential") + case ids[cn.ID] || creds[cn.Credential]: + return fmt.Errorf("connector %q: id or credential is not unique", cn.ID) + case !services[cn.Service]: + return fmt.Errorf("connector %q names unknown service %q", cn.ID, cn.Service) + case !cn.Prefix.Addr().Is6() || cn.Prefix.Addr().Is4In6() || cn.Prefix.Bits() != 96 || + cn.Prefix != cn.Prefix.Masked(): + return fmt.Errorf("connector %q: prefix %s is not a masked IPv6 /96", cn.ID, cn.Prefix) + case prefixes[cn.Service+"/"+cn.Prefix.String()]: + return fmt.Errorf("connector %q: prefix %s is already used in service %q", cn.ID, cn.Prefix, cn.Service) + } + for _, r := range cn.Routes { + if !r.IsValid() || r != r.Masked() { + return fmt.Errorf("connector %q: route %s is not a masked prefix", cn.ID, r) + } + } + ids[cn.ID], creds[cn.Credential] = true, true + prefixes[cn.Service+"/"+cn.Prefix.String()] = true + } + return nil +} diff --git a/tests/edgeconformance/masqueprovider/deploy/provider/deployment.yaml b/tests/edgeconformance/masqueprovider/deploy/provider/deployment.yaml new file mode 100644 index 00000000..c6e556ef --- /dev/null +++ b/tests/edgeconformance/masqueprovider/deploy/provider/deployment.yaml @@ -0,0 +1,95 @@ +# One fixture instance per pod. The pod needs nothing from its node: no host +# namespace, hostPath, NET_ADMIN or BPF. galactic-edge creates its packet +# devices in the pod's own network namespace; the lab's device plugin gives +# it /dev/net/tun. +apiVersion: apps/v1 +kind: Deployment +metadata: + name: masque-provider + labels: + app.kubernetes.io/name: masque-provider +spec: + replicas: 1 + # Two live instances of one pod template would claim the same routes in + # turn, so the old pod stops before its replacement starts. + strategy: + type: Recreate + selector: + matchLabels: + app.kubernetes.io/name: masque-provider + template: + metadata: + labels: + app.kubernetes.io/name: masque-provider + spec: + serviceAccountName: edge-provider + # SIGTERM ends the sessions, releases the routes and deletes the + # attachments; the fixture gives that 20 seconds. + terminationGracePeriodSeconds: 30 + affinity: + nodeAffinity: + requiredDuringSchedulingIgnoredDuringExecution: + nodeSelectorTerms: + - matchExpressions: + - key: galactic.datumapis.com/node + operator: In + values: + - edge + - key: galactic.datumapis.com/edge + operator: In + values: + - enabled + securityContext: + runAsNonRoot: true + runAsUser: 65532 + runAsGroup: 65532 + seccompProfile: + type: RuntimeDefault + containers: + - name: masque-provider + image: masque-provider:latest + imagePullPolicy: Never + ports: + - name: connect-ip + containerPort: 8443 + protocol: UDP + env: + - name: POD_NAMESPACE + valueFrom: + fieldRef: + fieldPath: metadata.namespace + - name: POD_NAME + valueFrom: + fieldRef: + fieldPath: metadata.name + - name: POD_UID + valueFrom: + fieldRef: + fieldPath: metadata.uid + - name: NODE_NAME + valueFrom: + fieldRef: + fieldPath: spec.nodeName + resources: + limits: + devic.es/tun: "1" + securityContext: + allowPrivilegeEscalation: false + readOnlyRootFilesystem: true + capabilities: + drop: ["ALL"] + volumeMounts: + - name: config + mountPath: /etc/masque + readOnly: true + volumes: + - name: config + secret: + secretName: masque-config + items: + - key: config.yaml + path: config.yaml + - key: tls.crt + path: tls/tls.crt + - key: tls.key + path: tls/tls.key diff --git a/tests/edgeconformance/masqueprovider/deploy/provider/kustomization.yaml b/tests/edgeconformance/masqueprovider/deploy/provider/kustomization.yaml new file mode 100644 index 00000000..dc025148 --- /dev/null +++ b/tests/edgeconformance/masqueprovider/deploy/provider/kustomization.yaml @@ -0,0 +1,6 @@ +# The fixture's ServiceAccount and Deployment. The suite's overlay sets the +# namespace, the node, and the masque-config Secret (config.yaml, tls.crt, +# tls.key) the pod mounts. +resources: + - serviceaccount.yaml + - deployment.yaml diff --git a/tests/edgeconformance/masqueprovider/deploy/provider/serviceaccount.yaml b/tests/edgeconformance/masqueprovider/deploy/provider/serviceaccount.yaml new file mode 100644 index 00000000..c311fff5 --- /dev/null +++ b/tests/edgeconformance/masqueprovider/deploy/provider/serviceaccount.yaml @@ -0,0 +1,4 @@ +apiVersion: v1 +kind: ServiceAccount +metadata: + name: edge-provider diff --git a/tests/edgeconformance/masqueprovider/deploy/registration/edgeprovider.yaml b/tests/edgeconformance/masqueprovider/deploy/registration/edgeprovider.yaml new file mode 100644 index 00000000..b91e3925 --- /dev/null +++ b/tests/edgeconformance/masqueprovider/deploy/registration/edgeprovider.yaml @@ -0,0 +1,20 @@ +apiVersion: galactic.datumapis.com/v1alpha1 +kind: EdgeProvider +metadata: + name: masque-fixture +spec: + contractVersions: + - v1alpha1 + capabilities: + - name: PacketAccess + version: v1 + - name: PrefixReachability + version: v1 + serviceAccounts: + - namespace: edge-masque + name: edge-provider + # The fixture takes its connectors from its own configuration file, not + # from an API kind; this names that format for admission's check. + configuration: + group: conformance.galactic.datumapis.com + kind: MASQUEFixtureConfig diff --git a/tests/edgeconformance/masqueprovider/deploy/registration/kustomization.yaml b/tests/edgeconformance/masqueprovider/deploy/registration/kustomization.yaml new file mode 100644 index 00000000..be5e7ef0 --- /dev/null +++ b/tests/edgeconformance/masqueprovider/deploy/registration/kustomization.yaml @@ -0,0 +1,2 @@ +resources: + - edgeprovider.yaml diff --git a/tests/edgeconformance/masqueprovider/main.go b/tests/edgeconformance/masqueprovider/main.go new file mode 100644 index 00000000..1067f69c --- /dev/null +++ b/tests/edgeconformance/masqueprovider/main.go @@ -0,0 +1,186 @@ +// Copyright 2026 Datum Cloud, Inc. +// +// SPDX-License-Identifier: AGPL-3.0-or-later + +//go:build linux + +// Command masqueprovider is the Galactic Edge conformance suite's MASQUE +// fixture: a CONNECT-IP (RFC 9484) server over HTTP/3 that runs as an +// unprivileged Galactic Edge provider. It uses only the provider contract +// and its helpers in internal/edge/provider: it holds its EdgeInstance's +// lease, requests one PacketAccess attachment per configured EdgeService, +// claims one EdgeRoute per connector /96, and carries each admitted +// connector's packets through internal/edge/connectip's TUN backend on the +// packet device galactic-edge creates in its pod. It needs no host +// namespace, hostPath, NET_ADMIN or BPF. +// +// Admission is a test fixture, not a product authority: each connector is a +// static bearer credential bound to one service's VPC and one /96, admitted +// only while the instance holds its lease and core has accepted the +// connector's route on a programmed attachment. Expiry or loss of any of +// them ends the connector's session. +// +// SIGTERM ends every session, then releases the routes, deletes the +// attachments and releases the lease. +package main + +import ( + "context" + "crypto/tls" + "errors" + "flag" + "fmt" + "log/slog" + "net" + "os" + "os/signal" + "strings" + "syscall" + "time" + + "golang.org/x/sys/unix" + coordinationv1 "k8s.io/api/coordination/v1" + "k8s.io/apimachinery/pkg/runtime" + ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/client" + + galacticv1alpha1 "go.datum.net/galactic/api/v1alpha1" + "go.datum.net/galactic/internal/edge/claim" + "go.datum.net/galactic/internal/edge/connectip" + "go.datum.net/galactic/internal/edge/provider" + "go.datum.net/galactic/internal/edge/timing" +) + +func main() { + if err := run(); err != nil { + fmt.Fprintln(os.Stderr, "masqueprovider:", err) + os.Exit(1) + } +} + +func run() error { + configPath := flag.String("config", "/etc/masque/config.yaml", "configuration file") + certPath := flag.String("tls-cert", "/etc/masque/tls/tls.crt", "server certificate") + keyPath := flag.String("tls-key", "/etc/masque/tls/tls.key", "server key") + flag.Parse() + log := slog.New(slog.NewJSONHandler(os.Stderr, nil)) + + cfg, err := LoadConfig(*configPath) + if err != nil { + return err + } + cert, err := tls.LoadX509KeyPair(*certPath, *keyPath) + if err != nil { + return err + } + pod, err := podFromEnv() + if err != nil { + return err + } + scheme := runtime.NewScheme() + for _, add := range []func(*runtime.Scheme) error{galacticv1alpha1.AddToScheme, coordinationv1.AddToScheme} { + if err := add(scheme); err != nil { + return err + } + } + rest, err := ctrl.GetConfig() + if err != nil { + return err + } + // Every request gives up well before the next renewal is due. + rest.Timeout = 3 * time.Second + c, err := client.New(rest, client.Options{Scheme: scheme}) + if err != nil { + return err + } + + p := &Provider{ + Client: c, Log: log.With("pod", pod.Name), Config: cfg, Pod: pod, Clock: timing.BootClock, + Renew: timing.RenewInterval, OpenTUN: openTUN, + Claims: &claim.Instance{ + ID: newHolderID(), Clock: timing.BootClock, Params: timing.Params, + Store: provider.InstanceStore{Client: c, Namespace: pod.Namespace, Duration: timing.Params.Duration}, + }, + } + ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGTERM, os.Interrupt) + defer stop() + if err := p.Start(ctx); err != nil { + return err + } + + srv, err := connectip.NewServer(connectip.Config{ + Authority: cfg.Authority, + TLSConfig: &tls.Config{Certificates: []tls.Certificate{cert}, MinVersion: tls.VersionTLS13}, + Admission: p, Backend: p, Logger: log, + }) + if err != nil { + return err + } + pc, err := (&net.ListenConfig{}).ListenPacket(ctx, "udp", cfg.Listen) + if err != nil { + return err + } + serveErr := make(chan error, 1) + go func() { serveErr <- srv.Serve(pc) }() + + t := time.NewTicker(time.Second) + defer t.Stop() + for { + if err := p.Reconcile(ctx); err != nil { + _ = srv.Close() + return err + } + select { + case <-ctx.Done(): + // Sessions end before the routes are released. + _ = srv.Close() + sctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) + defer cancel() + return p.Shutdown(sctx) + case err := <-serveErr: + sctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) + defer cancel() + return errors.Join(fmt.Errorf("HTTP/3 server stopped: %w", err), p.Shutdown(sctx)) + case <-t.C: + } + } +} + +// podFromEnv reads the pod's identity from the downward API variables the +// Deployment sets, and the node's boot ID. +func podFromEnv() (Pod, error) { + p := Pod{ + Namespace: os.Getenv("POD_NAMESPACE"), Name: os.Getenv("POD_NAME"), UID: os.Getenv("POD_UID"), + Node: os.Getenv("NODE_NAME"), RunAsUser: int64(os.Getuid()), RunAsGroup: int64(os.Getgid()), + } + if p.Namespace == "" || p.Name == "" || p.UID == "" || p.Node == "" { + return p, errors.New("POD_NAMESPACE, POD_NAME, POD_UID and NODE_NAME must be set") + } + b, err := os.ReadFile("/proc/sys/kernel/random/boot_id") + if err != nil { + return p, fmt.Errorf("read boot ID: %w", err) + } + p.BootID = strings.TrimSpace(string(b)) + return p, nil +} + +// openTUN attaches to the existing TUN device name as its owner, which +// needs no capability, and returns a nonblocking descriptor without +// packet-information headers. +func openTUN(name string) (*os.File, error) { + fd, err := unix.Open("/dev/net/tun", unix.O_RDWR|unix.O_NONBLOCK|unix.O_CLOEXEC, 0) + if err != nil { + return nil, fmt.Errorf("open /dev/net/tun: %w", err) + } + ifr, err := unix.NewIfreq(name) + if err != nil { + _ = unix.Close(fd) + return nil, err + } + ifr.SetUint16(unix.IFF_TUN | unix.IFF_NO_PI) + if err := unix.IoctlIfreq(fd, unix.TUNSETIFF, ifr); err != nil { + _ = unix.Close(fd) + return nil, fmt.Errorf("attach to TUN %s: %w", name, err) + } + return os.NewFile(uintptr(fd), name), nil +} diff --git a/tests/edgeconformance/masqueprovider/provider.go b/tests/edgeconformance/masqueprovider/provider.go new file mode 100644 index 00000000..bdba1604 --- /dev/null +++ b/tests/edgeconformance/masqueprovider/provider.go @@ -0,0 +1,524 @@ +// Copyright 2026 Datum Cloud, Inc. +// +// SPDX-License-Identifier: AGPL-3.0-or-later + +//go:build linux + +package main + +import ( + "context" + "crypto/rand" + "crypto/subtle" + "errors" + "fmt" + "log/slog" + "net/netip" + "os" + "sync" + "sync/atomic" + "time" + + apierrors "k8s.io/apimachinery/pkg/api/errors" + "k8s.io/apimachinery/pkg/api/meta" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "sigs.k8s.io/controller-runtime/pkg/client" + + galacticv1alpha1 "go.datum.net/galactic/api/v1alpha1" + "go.datum.net/galactic/internal/edge/claim" + "go.datum.net/galactic/internal/edge/connectip" + "go.datum.net/galactic/internal/edge/provider" +) + +// Pod identifies the pod this process runs in. +type Pod struct { + Namespace, Name, UID, Node, BootID string + // RunAsUser and RunAsGroup own the packet devices this process opens. + RunAsUser, RunAsGroup int64 +} + +// Provider is one fixture instance: it holds its EdgeInstance's lease, one +// PacketAccess attachment per service and one EdgeRoute per connector /96, +// and admits a connector's CONNECT-IP request only while all three are +// current. Its Client must read from the API server. +type Provider struct { + Client client.Client + Log *slog.Logger + Config *Config + Pod Pod + Claims *claim.Instance + Clock claim.Clock + // Renew is how often the lease is renewed. + Renew time.Duration + // OpenTUN opens an attachment's packet device as a nonblocking + // descriptor. + OpenTUN func(name string) (*os.File, error) + + instance *galacticv1alpha1.EdgeInstance + + mu sync.Mutex + lease *claim.Lease + services map[string]*service + connectors []*connector + // deadline is the held lease's deadline on Clock, or zero; every packet + // reads it, so forwarding stops at the deadline even while the + // reconcile loop waits on the API. + deadline atomic.Int64 + sessionID atomic.Uint64 +} + +type service struct { + cfg Service + attachment *galacticv1alpha1.EdgeAttachment + // programmed identifies the host state the backend was opened on. + programmed string + backend *connectip.TUNBackend +} + +type connector struct { + cfg Connector + svc *service + route string + epoch int64 + active bool + grants map[*grant]struct{} + logged string +} + +// forwarding reports whether the held lease allows packets now. +func (p *Provider) forwarding() bool { return int64(p.Clock()) < p.deadline.Load() } + +func (p *Provider) setLease(l *claim.Lease) { + p.lease = l + if l == nil { + p.deadline.Store(0) + return + } + p.deadline.Store(int64(l.Deadline)) +} + +// Start registers the pod's EdgeInstance and sets up the service and +// connector state. +func (p *Provider) Start(ctx context.Context) error { + inst, err := provider.CreateInstance(ctx, p.Client, provider.NewInstance(p.Pod.Namespace, p.Config.Provider, + p.Pod.Name, p.Pod.UID, p.Pod.Node, p.Pod.BootID)) + if err != nil { + return fmt.Errorf("register EdgeInstance: %w", err) + } + p.instance = inst + p.services = map[string]*service{} + for _, s := range p.Config.Services { + p.services[s.Key()] = &service{cfg: s} + } + for _, c := range p.Config.Connectors { + svc := p.services[c.Service] + p.connectors = append(p.connectors, &connector{cfg: c, svc: svc, + route: provider.RouteName(svc.cfg.VPCUID, c.Prefix), grants: map[*grant]struct{}{}}) + } + p.Log.Info("instance registered", "instance", inst.Name) + return nil +} + +// Reconcile makes one pass: lease, then each attachment, then each route. +// It returns an error only when retrying cannot succeed. +func (p *Provider) Reconcile(ctx context.Context) error { + if err := p.reconcileLease(ctx); err != nil { + return err + } + for _, s := range p.Config.Services { + if err := p.reconcileAttachment(ctx, p.services[s.Key()]); err != nil { + if errors.Is(err, provider.ErrNameTaken) { + return err + } + p.Log.Warn("attachment reconcile failed", "service", s.Name, "error", err) + } + } + for _, c := range p.connectors { + p.reconcileRoute(ctx, c) + } + return nil +} + +func (p *Provider) reconcileLease(ctx context.Context) error { + p.mu.Lock() + l := p.lease + p.mu.Unlock() + now := p.Clock() + if l != nil && !l.Valid(now) { + p.Log.Warn("lease expired", "epoch", l.Epoch) + p.loseLease() + l = nil + } + if l == nil { + got, err := p.Claims.Acquire(ctx, p.instance.Name) + switch { + case errors.Is(err, provider.ErrNameTaken): + return err + case err != nil: + p.Log.Warn("lease not yet acquired", "reason", err) + return nil + } + p.mu.Lock() + p.setLease(&got) + p.mu.Unlock() + p.Log.Info("lease acquired", "epoch", got.Epoch) + return nil + } + // A lease's deadline is its renewal's start plus the claim duration. + if now < l.Deadline-p.Claims.Params.Duration+p.Renew { + return nil + } + got, err := p.Claims.Renew(ctx, *l) + switch { + case errors.Is(err, claim.ErrLost): + p.Log.Warn("lease lost", "epoch", l.Epoch, "reason", err) + p.loseLease() + case err != nil: + p.Log.Warn("lease renewal failed; the lease holds until its deadline", "error", err) + default: + p.mu.Lock() + p.setLease(&got) + p.mu.Unlock() + } + return nil +} + +// loseLease ends every session: every route claim was made under the lost +// epoch. +func (p *Provider) loseLease() { + p.mu.Lock() + defer p.mu.Unlock() + p.setLease(nil) + for _, c := range p.connectors { + p.deactivate(c, "lease lost") + } +} + +func (p *Provider) reconcileAttachment(ctx context.Context, s *service) error { + name := provider.AttachmentName(s.cfg.UID, p.Pod.UID) + var att galacticv1alpha1.EdgeAttachment + err := p.Client.Get(ctx, client.ObjectKey{Namespace: p.Pod.Namespace, Name: name}, &att) + switch { + case apierrors.IsNotFound(err): + p.stopServing(s, "attachment missing") + created, err := provider.CreateAttachment(ctx, p.Client, p.attachmentRequest(s, name)) + if err != nil { + return fmt.Errorf("request attachment %s: %w", name, err) + } + p.Log.Info("attachment requested", "attachment", created.Name, "service", s.cfg.Name) + return nil + case err != nil: + return err + } + if att.Spec.PodUID != p.Pod.UID || att.Spec.ServiceRef.UID != s.cfg.UID { + return fmt.Errorf("%w: EdgeAttachment %s", provider.ErrNameTaken, name) + } + // Admit reads the stored copy while setReady writes att. + p.mu.Lock() + s.attachment = att.DeepCopy() + p.mu.Unlock() + if !meta.IsStatusConditionTrue(att.Status.Conditions, galacticv1alpha1.EdgeAttachmentProgrammed) || + att.DeletionTimestamp != nil { + p.stopServing(s, "attachment not programmed") + return p.setReady(ctx, &att, false, "NotProgrammed", "waiting for galactic-edge to program the attachment") + } + key := att.Status.HostLink + "/" + att.Status.PacketDevice + "/" + att.Status.Address + p.mu.Lock() + open := s.backend != nil && s.programmed == key + p.mu.Unlock() + if !open { + p.stopServing(s, "attachment reprogrammed") + if err := p.openBackend(s, att.Status.PacketDevice, key); err != nil { + return p.setReady(ctx, &att, false, "DataPlaneFailed", err.Error()) + } + p.Log.Info("serving attachment", "attachment", name, "device", att.Status.PacketDevice) + } + return p.setReady(ctx, &att, true, "Serving", "the provider is serving the attachment") +} + +func (p *Provider) attachmentRequest(s *service, name string) *galacticv1alpha1.EdgeAttachment { + return &galacticv1alpha1.EdgeAttachment{ + ObjectMeta: metav1.ObjectMeta{Namespace: p.Pod.Namespace, Name: name}, + Spec: galacticv1alpha1.EdgeAttachmentSpec{ + ProviderRef: galacticv1alpha1.EdgeProviderReference{Name: p.Config.Provider}, + ServiceRef: galacticv1alpha1.EdgeServiceReference{Namespace: s.cfg.Namespace, Name: s.cfg.Name, + UID: s.cfg.UID}, + InstanceRef: galacticv1alpha1.EdgeInstanceReference{Name: p.instance.Name}, + PodUID: p.Pod.UID, + ContractVersion: "v1alpha1", + Capabilities: []galacticv1alpha1.EdgeCapability{{Name: "PacketAccess", Version: "v1"}}, + PacketDevice: &galacticv1alpha1.EdgePacketDevice{Owner: p.Pod.RunAsUser, Group: p.Pod.RunAsGroup}, + }, + } +} + +func (p *Provider) openBackend(s *service, device, key string) error { + if device == "" { + return errors.New("the attachment has no packet device") + } + f, err := p.OpenTUN(device) + if err != nil { + return err + } + b, err := connectip.NewTUNBackend(f, s.cfg.VPC) + if err != nil { + _ = f.Close() + return err + } + p.mu.Lock() + s.backend, s.programmed = b, key + p.mu.Unlock() + return nil +} + +// stopServing ends s's sessions and closes its backend. +func (p *Provider) stopServing(s *service, reason string) { + p.mu.Lock() + b := s.backend + s.backend, s.programmed = nil, "" + for _, c := range p.connectors { + if c.svc == s { + p.deactivate(c, reason) + } + } + p.mu.Unlock() + if b != nil { + _ = b.Close() + } +} + +// setReady writes the attachment's Ready condition, the only one a +// provider sets, when it differs. +func (p *Provider) setReady(ctx context.Context, att *galacticv1alpha1.EdgeAttachment, ready bool, reason, + msg string) error { + status := metav1.ConditionFalse + if ready { + status = metav1.ConditionTrue + } + if c := meta.FindStatusCondition(att.Status.Conditions, galacticv1alpha1.EdgeAttachmentReady); c != nil && + c.Status == status && c.Reason == reason && c.ObservedGeneration == att.Generation { + return nil + } + meta.SetStatusCondition(&att.Status.Conditions, metav1.Condition{Type: galacticv1alpha1.EdgeAttachmentReady, + Status: status, Reason: reason, Message: msg, ObservedGeneration: att.Generation}) + return p.Client.Status().Update(ctx, att) +} + +// reconcileRoute claims c's /96 while the lease is held and c's backend is +// open, and activates c once core accepts the claim. +func (p *Provider) reconcileRoute(ctx context.Context, c *connector) { + p.mu.Lock() + l, att, open := p.lease, c.svc.attachment, c.svc.backend != nil + p.mu.Unlock() + if l == nil || att == nil || !open { + return + } + var r galacticv1alpha1.EdgeRoute + err := p.Client.Get(ctx, client.ObjectKey{Namespace: p.Pod.Namespace, Name: c.route}, &r) + if apierrors.IsNotFound(err) { + r = *provider.NewRoute(p.Pod.Namespace, p.Config.Provider, c.svc.cfg.VPCUID, c.cfg.Prefix) + r.Spec.Claim = &galacticv1alpha1.EdgeRouteClaim{Attachment: att.Name, HolderPodUID: p.Pod.UID, + HolderInstance: l.Key, InstanceEpoch: int64(l.Epoch), Epoch: 1} + err = p.Client.Create(ctx, &r) + } else if err == nil { + err = provider.Claim(ctx, p.Client, &r, att.Name, p.Pod.UID, *l) + } + if err != nil { + if !apierrors.IsConflict(err) && !apierrors.IsAlreadyExists(err) { + p.Log.Warn("route claim failed", "connector", c.cfg.ID, "error", err) + } + return + } + accepted := acceptedFor(&r, att.Name, l) + p.mu.Lock() + defer p.mu.Unlock() + if p.lease == nil || p.lease.Epoch != l.Epoch || c.svc.attachment == nil || c.svc.attachment.Name != att.Name { + return + } + switch { + case accepted && (!c.active || c.epoch != r.Spec.Claim.Epoch): + p.deactivate(c, "claim epoch changed") + c.active, c.epoch = true, r.Spec.Claim.Epoch + case !accepted: + p.deactivate(c, "route not accepted") + } + if state := fmt.Sprintf("accepted=%t epoch=%d", accepted, r.Spec.Claim.Epoch); state != c.logged { + c.logged = state + p.Log.Info("route", "connector", c.cfg.ID, "route", r.Name, "prefix", c.cfg.Prefix, "state", state) + } +} + +// acceptedFor reports whether core accepted the route's current claim for +// attachment under l and has not revoked or closed it. +func acceptedFor(r *galacticv1alpha1.EdgeRoute, attachment string, l *claim.Lease) bool { + a, cl := r.Status.Accepted, r.Spec.Claim + return a != nil && cl != nil && !cl.Released && a.Epoch == cl.Epoch && a.Attachment == attachment && + a.Instance == l.Key && a.InstanceEpoch == int64(l.Epoch) && r.Status.ClosedEpoch < cl.Epoch && + meta.IsStatusConditionTrue(r.Status.Conditions, galacticv1alpha1.EdgeRouteAccepted) +} + +// deactivate ends every session of c. The caller holds p.mu. +func (p *Provider) deactivate(c *connector, reason string) { + if c.active || len(c.grants) > 0 { + p.Log.Info("connector inactive", "connector", c.cfg.ID, "reason", reason) + } + c.active, c.epoch = false, 0 + for g := range c.grants { + g.end() + delete(c.grants, g) + } +} + +// Shutdown ends every session, then releases the routes, deletes the +// attachments and releases the lease, in that order, so a replacement takes +// nothing over while this instance still forwards. +func (p *Provider) Shutdown(ctx context.Context) error { + p.Log.Info("shutting down") + p.mu.Lock() + l := p.lease + p.setLease(nil) + p.mu.Unlock() + for _, s := range p.services { + p.stopServing(s, "shutting down") + } + var errs []error + if l != nil { + for _, c := range p.connectors { + if err := p.releaseRoute(ctx, c, *l); err != nil { + errs = append(errs, fmt.Errorf("release route of %s: %w", c.cfg.ID, err)) + } + } + } + for _, s := range p.services { + if s.attachment == nil { + continue + } + if err := p.Client.Delete(ctx, s.attachment); client.IgnoreNotFound(err) != nil { + errs = append(errs, fmt.Errorf("delete attachment %s: %w", s.attachment.Name, err)) + } + } + if l != nil { + if err := p.Claims.Release(ctx, *l); err != nil { + errs = append(errs, fmt.Errorf("release lease: %w", err)) + } + } + return errors.Join(errs...) +} + +func (p *Provider) releaseRoute(ctx context.Context, c *connector, l claim.Lease) error { + for range 5 { + var r galacticv1alpha1.EdgeRoute + err := p.Client.Get(ctx, client.ObjectKey{Namespace: p.Pod.Namespace, Name: c.route}, &r) + if err != nil { + return client.IgnoreNotFound(err) + } + err = provider.Release(ctx, p.Client, &r, l) + if errors.Is(err, provider.ErrNotHolder) { + return nil + } + if !apierrors.IsConflict(err) { + return err + } + } + return errors.New("route kept changing") +} + +// Admit implements connectip.Admission for the configured connectors. +func (p *Provider) Admit(_ context.Context, credential, vpc string) (connectip.Lease, error) { + var c *connector + for _, cand := range p.connectors { + if subtle.ConstantTimeCompare([]byte(cand.cfg.Credential), []byte(credential)) == 1 { + c = cand + } + } + if c == nil { + return nil, connectip.ErrUnauthenticated + } + if c.svc.cfg.VPC != vpc { + return nil, connectip.ErrForbidden + } + p.mu.Lock() + defer p.mu.Unlock() + if !c.active || p.lease == nil || c.svc.backend == nil || c.svc.attachment == nil || !p.forwarding() { + return nil, fmt.Errorf("connector %s has no accepted route on a programmed attachment", c.cfg.ID) + } + g := &grant{ + p: p, c: c, done: make(chan struct{}), + info: connectip.SessionInfo{ + ID: p.sessionID.Add(1), Principal: c.cfg.ID, VPC: vpc, Prefix: c.cfg.Prefix, + Connector: c.cfg.ID, Attachment: c.svc.attachment.Name, Claim: c.route, Epoch: uint64(c.epoch), + }, + } + c.grants[g] = struct{}{} + return g, nil +} + +// grant is one admitted request's lease. +type grant struct { + p *Provider + c *connector + info connectip.SessionInfo + done chan struct{} + endOnce sync.Once +} + +func (g *grant) Info() connectip.SessionInfo { return g.info } +func (g *grant) Routes() []netip.Prefix { return g.c.cfg.Routes } +func (g *grant) Done() <-chan struct{} { return g.done } +func (g *grant) end() { g.endOnce.Do(func() { close(g.done) }) } + +func (g *grant) Close() error { + g.end() + g.p.mu.Lock() + delete(g.c.grants, g) + g.p.mu.Unlock() + return nil +} + +// Open implements connectip.PacketBackend: it opens the session on the +// backend of the attachment admission named, and drops its packets once +// the lease's deadline passes. +func (p *Provider) Open(ctx context.Context, info connectip.SessionInfo) (connectip.BackendSession, error) { + p.mu.Lock() + var b *connectip.TUNBackend + for _, s := range p.services { + if s.attachment != nil && s.attachment.Name == info.Attachment { + b = s.backend + } + } + p.mu.Unlock() + if b == nil { + return nil, fmt.Errorf("attachment %s is not being served", info.Attachment) + } + s, err := b.Open(ctx, info) + if err != nil { + return nil, err + } + return &gatedSession{BackendSession: s, allow: p.forwarding}, nil +} + +var errExpired = errors.New("the instance's lease has expired") + +// gatedSession forwards packets only while allow reports true. +type gatedSession struct { + connectip.BackendSession + allow func() bool +} + +func (s *gatedSession) WritePacket(pkt []byte) error { + if !s.allow() { + return errExpired + } + return s.BackendSession.WritePacket(pkt) +} + +func (s *gatedSession) ReadPacket(ctx context.Context, buf []byte) (int, error) { + n, err := s.BackendSession.ReadPacket(ctx, buf) + if err == nil && !s.allow() { + return 0, errExpired + } + return n, err +} + +// newHolderID returns a unique holder identity for this process. +func newHolderID() string { return rand.Text() } diff --git a/tests/edgeconformance/masqueprovider/provider_test.go b/tests/edgeconformance/masqueprovider/provider_test.go new file mode 100644 index 00000000..61dd44b0 --- /dev/null +++ b/tests/edgeconformance/masqueprovider/provider_test.go @@ -0,0 +1,196 @@ +// Copyright 2026 Datum Cloud, Inc. +// +// SPDX-License-Identifier: AGPL-3.0-or-later + +//go:build linux + +package main + +import ( + "context" + "errors" + "io" + "log/slog" + "net/netip" + "strings" + "testing" + "time" + + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + + galacticv1alpha1 "go.datum.net/galactic/api/v1alpha1" + "go.datum.net/galactic/internal/edge/claim" + "go.datum.net/galactic/internal/edge/connectip" +) + +const notA96 = "/96" + +func validConfig() *Config { + return &Config{ + Provider: "masque-fixture", Authority: "masque.test:8443", Listen: ":8443", + Services: []Service{ + {Namespace: "ns10", Name: "svc", UID: "s10", VPCUID: "v10", VPC: "10"}, + {Namespace: "ns20", Name: "svc", UID: "s20", VPCUID: "v20", VPC: "20"}, + }, + Connectors: []Connector{ + {ID: "c10", Credential: "t10", Service: "ns10/svc", Prefix: netip.MustParsePrefix("fd00:1::/96")}, + {ID: "c20", Credential: "t20", Service: "ns20/svc", Prefix: netip.MustParsePrefix("fd00:1::/96")}, + }, + } +} + +func TestConfigValidate(t *testing.T) { + if err := validConfig().Validate(); err != nil { + t.Fatalf("identical /96s in different services: %v", err) + } + for _, tc := range []struct { + name string + mutate func(*Config) + want string + }{ + {"no authority", func(c *Config) { c.Authority = "" }, "required"}, + {"service without VPC", func(c *Config) { c.Services[0].VPC = "" }, "required"}, + {"duplicate service", func(c *Config) { c.Services[1].Namespace = "ns10" }, "twice"}, + {"shared credential", func(c *Config) { c.Connectors[1].Credential = "t10" }, "not unique"}, + {"unknown service", func(c *Config) { c.Connectors[0].Service = "ns30/svc" }, "unknown service"}, + {"not a /96", func(c *Config) { c.Connectors[0].Prefix = netip.MustParsePrefix("fd00:1::/64") }, notA96}, + {"unmasked", func(c *Config) { c.Connectors[0].Prefix = netip.MustParsePrefix("fd00:1::1/96") }, notA96}, + {"IPv4-mapped", func(c *Config) { c.Connectors[0].Prefix = netip.MustParsePrefix("::ffff:0.0.0.0/96") }, notA96}, + {"same /96 twice in one service", func(c *Config) { c.Connectors[1].Service = "ns10/svc" }, "already used"}, + {"unmasked route", func(c *Config) { + c.Connectors[0].Routes = []netip.Prefix{netip.MustParsePrefix("fd20::1/32")} + }, "route"}, + } { + t.Run(tc.name, func(t *testing.T) { + c := validConfig() + tc.mutate(c) + if err := c.Validate(); err == nil || !strings.Contains(err.Error(), tc.want) { + t.Fatalf("Validate() = %v, want an error containing %q", err, tc.want) + } + }) + } +} + +// fixture returns a provider with both connectors' services served and the +// lease held until deadline on a clock that reads now. +func fixture(t *testing.T, now *time.Duration, deadline time.Duration) *Provider { + t.Helper() + p := &Provider{Log: slog.New(slog.NewTextHandler(io.Discard, nil)), Config: validConfig(), + Clock: func() time.Duration { return *now }} + p.services = map[string]*service{} + for _, s := range p.Config.Services { + p.services[s.Key()] = &service{cfg: s, + attachment: &galacticv1alpha1.EdgeAttachment{ObjectMeta: metav1.ObjectMeta{Name: "ea-" + s.Namespace}}} + } + for _, c := range p.Config.Connectors { + p.connectors = append(p.connectors, &connector{cfg: c, svc: p.services[c.Service], route: "er-" + c.ID, + grants: map[*grant]struct{}{}}) + } + p.setLease(&claim.Lease{Key: "ei-pod", Epoch: 1, Deadline: deadline}) + return p +} + +// serve marks every service's backend open and every connector's route +// accepted at epoch 3. +func serve(p *Provider) { + for _, s := range p.services { + s.backend = &connectip.TUNBackend{} + } + for _, c := range p.connectors { + c.active, c.epoch = true, 3 + } +} + +func TestAdmit(t *testing.T) { + now := time.Second + p := fixture(t, &now, time.Minute) + ctx := context.Background() + + if _, err := p.Admit(ctx, "unknown", "10"); !errors.Is(err, connectip.ErrUnauthenticated) { + t.Errorf("unknown credential: %v, want ErrUnauthenticated", err) + } + if _, err := p.Admit(ctx, "t10", "20"); !errors.Is(err, connectip.ErrForbidden) { + t.Errorf("VPC 10's credential for VPC 20: %v, want ErrForbidden", err) + } + if _, err := p.Admit(ctx, "t10", "10"); err == nil || errors.Is(err, connectip.ErrForbidden) || + errors.Is(err, connectip.ErrUnauthenticated) { + t.Errorf("connector without an accepted route: %v, want an unavailable error", err) + } + + serve(p) + l, err := p.Admit(ctx, "t20", "20") + if err != nil { + t.Fatal(err) + } + want := connectip.SessionInfo{ID: 1, Principal: "c20", VPC: "20", Prefix: netip.MustParsePrefix("fd00:1::/96"), + Connector: p.connectors[1].cfg.ID, Attachment: "ea-ns20", Claim: "er-c20", Epoch: 3} + if got := l.Info(); got != want { + t.Errorf("Info() = %+v, want %+v", got, want) + } + + // Losing the route ends that connector's session and no other. + other, err := p.Admit(ctx, "t10", "10") + if err != nil { + t.Fatal(err) + } + p.mu.Lock() + p.deactivate(p.connectors[1], "test") + p.mu.Unlock() + select { + case <-l.Done(): + default: + t.Error("VPC 20's session survived its route") + } + select { + case <-other.Done(): + t.Error("VPC 10's session ended with VPC 20's route") + default: + } + if _, err := p.Admit(ctx, "t20", "20"); err == nil { + t.Error("admitted a connector whose route is no longer accepted") + } + _ = other.Close() + if len(p.connectors[0].grants) != 0 { + t.Error("Close left the grant registered") + } + + // Past the lease's deadline nothing is admitted. + serve(p) + now = 2 * time.Minute + if _, err := p.Admit(ctx, "t10", "10"); err == nil { + t.Error("admitted past the lease's deadline") + } +} + +type fakeSession struct{ writes, reads int } + +func (s *fakeSession) WritePacket([]byte) error { s.writes++; return nil } +func (s *fakeSession) ReadPacket(context.Context, []byte) (int, error) { + s.reads++ + return 40, nil +} +func (s *fakeSession) Close() error { return nil } + +func TestGatedSessionStopsAtDeadline(t *testing.T) { + now := time.Second + p := fixture(t, &now, 10*time.Second) + inner := &fakeSession{} + s := &gatedSession{BackendSession: inner, allow: p.forwarding} + buf := make([]byte, 64) + if err := s.WritePacket(buf); err != nil { + t.Fatal(err) + } + if n, err := s.ReadPacket(context.Background(), buf); err != nil || n != 40 { + t.Fatalf("ReadPacket = %d, %v", n, err) + } + now = 10 * time.Second + if err := s.WritePacket(buf); !errors.Is(err, errExpired) { + t.Errorf("write at the deadline: %v, want errExpired", err) + } + if _, err := s.ReadPacket(context.Background(), buf); !errors.Is(err, errExpired) { + t.Errorf("read at the deadline: %v, want errExpired", err) + } + if inner.writes != 1 { + t.Errorf("%d writes reached the device, want 1", inner.writes) + } +}