From 7f0c56ea285c6e09ba944fe5d59014e0a16e5c72 Mon Sep 17 00:00:00 2001 From: Peter Sprygada Date: Tue, 27 Oct 2026 12:50:21 -0400 Subject: [PATCH 1/3] test(edge): Add a MASQUE fixture provider The MASQUE packet-access checks need a CONNECT-IP server that runs as an ordinary Galactic Edge provider. internal/edge/connectip has the protocol component and a TUN packet backend, but nothing connected them to EdgeInstance, EdgeAttachment and EdgeRoute. tests/edgeconformance/masqueprovider is that fixture. It registers its pod's EdgeInstance and holds the instance lease through provider.InstanceStore and claim.Instance with the cell's timing. It requests one PacketAccess attachment per configured EdgeService and claims one EdgeRoute per connector /96. Once core accepts a route, it carries that connector's packets through connectip.TUNBackend on the device galactic-edge created in the pod. It uses 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. A connector is admitted only while the lease is held and its route is accepted on a programmed attachment, and losing any of those ends that connector's sessions without touching others. Packets stop at the lease deadline on CLOCK_BOOTTIME even while the reconcile loop waits on the API. On SIGTERM the fixture ends sessions before it releases routes, deletes attachments and releases the lease. The Dockerfile builds a scratch image that runs as 65532, and deploy/ holds its registration and an unprivileged Deployment. --- .../edgeconformance/masqueprovider/Dockerfile | 21 + .../edgeconformance/masqueprovider/config.go | 115 ++++ .../deploy/provider/deployment.yaml | 95 ++++ .../deploy/provider/kustomization.yaml | 6 + .../deploy/provider/serviceaccount.yaml | 4 + .../deploy/registration/edgeprovider.yaml | 20 + .../deploy/registration/kustomization.yaml | 2 + tests/edgeconformance/masqueprovider/main.go | 186 +++++++ .../masqueprovider/provider.go | 524 ++++++++++++++++++ .../masqueprovider/provider_test.go | 196 +++++++ 10 files changed, 1169 insertions(+) create mode 100644 tests/edgeconformance/masqueprovider/Dockerfile create mode 100644 tests/edgeconformance/masqueprovider/config.go create mode 100644 tests/edgeconformance/masqueprovider/deploy/provider/deployment.yaml create mode 100644 tests/edgeconformance/masqueprovider/deploy/provider/kustomization.yaml create mode 100644 tests/edgeconformance/masqueprovider/deploy/provider/serviceaccount.yaml create mode 100644 tests/edgeconformance/masqueprovider/deploy/registration/edgeprovider.yaml create mode 100644 tests/edgeconformance/masqueprovider/deploy/registration/kustomization.yaml create mode 100644 tests/edgeconformance/masqueprovider/main.go create mode 100644 tests/edgeconformance/masqueprovider/provider.go create mode 100644 tests/edgeconformance/masqueprovider/provider_test.go 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) + } +} From 8e84b0674940b57337a7a6623346e3da18ced7f4 Mon Sep 17 00:00:00 2001 From: Peter Sprygada Date: Tue, 27 Oct 2026 12:50:31 -0400 Subject: [PATCH 2/3] test(edge): Prove MASQUE packet access in the containerlab lab The MASQUE fixture provider has no proof yet that the foundation carries its packets correctly. TestMASQUE runs it on the lab's dfw cluster and checks what the packet-access milestone requires. - M16: the fixture pod shares no host namespace, mounts no hostPath and adds no capability, and its process holds none. galactic-edge creates each TUN device in the pod, owned by 65532, and the fixture opens it. An unknown credential gets 401, a credential presented for the other VPC gets 403, and a connector whose route core refused gets 503. - M07: VPCs 10 and 20 hold the same connector /96 in one provider process. Neighbour caches are flushed first. Each tenant pings its own workloads on the compute node and on the fixture's node, and the other tenant's. Each client uses its own ICMPv6 echo identifier, so captures on every workload link show which client a packet came from. The own-tenant capture is the positive control. - NAT egress: both tenants reach the off-fabric host through the NAT shards from the attachment node. - Partial setup: a second instance whose only route is refused is killed without shutdown. The first instance keeps answering, and core reaps the attachment, instance, lease and route, and the node removes the host link. - Recreation: the fixture's pod is deleted, and the replacement serves both tenants in isolation. The old pod's objects and host links are reaped, and the replacement's routes stay accepted. - Teardown fails the run unless the provider and core have removed every attachment, instance, lease and route before the namespace is deleted, no fixture advertisement or host link remains, and the tenants' edge-client pods are unchanged and still reach each other. Every packet row also requires an accepted request with the configured /96 that sent through its whole window, so a refused or cut-short request cannot pass on its reply ratio. edgeprobe gains a connectip client and an ICMPv6 echo capture. The client runs on the fixture'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. task verify:masque-conformance runs the rows, and build:masque-provider builds and loads the image. --- deploy/containerlab/Taskfile.yaml | 16 + docs/designs/masque/acceptance.md | 2 + tests/edgeconformance/edgeprobe/connectip.go | 216 ++++ tests/edgeconformance/edgeprobe/main.go | 21 +- tests/edgeconformance/fixtures_test.go | 9 +- tests/edgeconformance/harness_test.go | 6 + tests/edgeconformance/masque_test.go | 988 +++++++++++++++++++ 7 files changed, 1253 insertions(+), 5 deletions(-) create mode 100644 tests/edgeconformance/edgeprobe/connectip.go create mode 100644 tests/edgeconformance/masque_test.go 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..8802e582 --- /dev/null +++ b/tests/edgeconformance/masque_test.go @@ -0,0 +1,988 @@ +// 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 _, 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) + } + } +} From 98f59d6e4deb8359b61f6c99e42ea863b6d2ee9b Mon Sep 17 00:00:00 2001 From: Peter Sprygada Date: Tue, 27 Oct 2026 13:14:19 -0400 Subject: [PATCH 3/3] test(edge): Remove the MASQUE CA from the lab nodes at teardown TestMASQUE copies its fixture CA onto the three dfw nodes so the CONNECT-IP client can verify the fixture's certificate. Teardown now deletes it, so the run leaves nothing of its own behind on the nodes. --- tests/edgeconformance/masque_test.go | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/tests/edgeconformance/masque_test.go b/tests/edgeconformance/masque_test.go index 8802e582..74991c60 100644 --- a/tests/edgeconformance/masque_test.go +++ b/tests/edgeconformance/masque_test.go @@ -972,6 +972,11 @@ func (m *masque) teardown(t *testing.T) { } } } + 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)