From 9db1a590b6abdade8e3f396ee79a4a5ea793ec05 Mon Sep 17 00:00:00 2001 From: Scot Wells Date: Thu, 8 Oct 2026 14:18:23 -0500 Subject: [PATCH 1/3] test: Add a private service validation environment --- tests/psc/README.md | 119 ++++++++ tests/psc/example.yaml | 83 ++++++ tests/psc/fixture/constants_linux.go | 30 ++ tests/psc/fixture/main_linux.go | 398 +++++++++++++++++++++++++++ tests/psc/fixture/network_linux.go | 307 +++++++++++++++++++++ tests/psc/fixture/query.py | 47 ++++ tests/psc/fixture/relay.py | 111 ++++++++ tests/psc/fixture/relay_test.py | 45 +++ tests/psc/fixture/remote_linux.go | 240 ++++++++++++++++ 9 files changed, 1380 insertions(+) create mode 100644 tests/psc/README.md create mode 100644 tests/psc/example.yaml create mode 100644 tests/psc/fixture/constants_linux.go create mode 100644 tests/psc/fixture/main_linux.go create mode 100644 tests/psc/fixture/network_linux.go create mode 100644 tests/psc/fixture/query.py create mode 100644 tests/psc/fixture/relay.py create mode 100644 tests/psc/fixture/relay_test.py create mode 100644 tests/psc/fixture/remote_linux.go diff --git a/tests/psc/README.md b/tests/psc/README.md new file mode 100644 index 00000000..52b11dd7 --- /dev/null +++ b/tests/psc/README.md @@ -0,0 +1,119 @@ +# Private Service Connect qualification + +This Linux fixture runs the production service route controller and TC dataplane +against a real edge API. Two consumer VPCs use the same client address, source +port, and service frontend. A shared producer returns its destination identity +and the request payload over UDP and TCP. No DNS service is required. + +Build after generating the eBPF objects: + +```sh +task build:ebpf +go build -o bin/psc-fixture ./tests/psc/fixture +python3 -m unittest discover -s tests/psc/fixture -p '*_test.py' +``` + +Run in a disposable privileged Linux environment with IPv6, `ip`, Python 3, +bpffs, and an edge kubeconfig. Supply two existing consumer VPCs and attachments, +a shared producer attachment, and the service endpoints and policies from the +[example manifest](example.yaml). Apply the Cloud and Network CRDs first. +Fill the policy variables before applying the example: + +```sh +export CONSUMER_A_UID=$(kubectl get vpc consumer-a -n psc-test -o jsonpath='{.metadata.uid}') +export CONSUMER_B_UID=$(kubectl get vpc consumer-b -n psc-test -o jsonpath='{.metadata.uid}') +export AUTHORIZATION_VALID_UNTIL=$(date -u -d '+90 seconds' +%Y-%m-%dT%H:%M:%SZ) +envsubst < tests/psc/example.yaml | kubectl apply -f - +``` + +The kubeconfig must allow the fixture to read those objects, update attachment +status and policy status, and watch routing dependencies. + +## Local delivery + +Save a configuration file: + +```json +{ + "role": "local", + "kubeconfig": "/runtime/edge.kubeconfig", + "namespace": "psc-test", + "nodeName": "node-a", + "state": "/runtime/psc-state.json", + "cases": [ + { + "project": "a", + "vpcName": "consumer-a", + "vpcIdentity": "consumer-a", + "consumerAttachment": "client-a", + "producerAttachment": "producer", + "destination": "fd70:100::10" + }, + { + "project": "b", + "vpcName": "consumer-b", + "vpcIdentity": "consumer-b", + "consumerAttachment": "client-b", + "producerAttachment": "producer", + "destination": "fd70:100::11" + } + ] +} +``` + +Start `bin/psc-fixture --config /runtime/psc.json`. The state file identifies +consumer namespaces. Query each namespace using its expected producer address: + +```sh +ip netns exec CONSUMER_A python3 /runtime/psc-query.py udp fd70:100::10 request-a +ip netns exec CONSUMER_B python3 /runtime/psc-query.py tcp fd70:100::11 request-b +``` + +A successful result identifies the expected producer and the frontend +`fd70:ffff::10`. A denied or expired request exits with status 3. Repeat both +transports from both VPCs using the same source port to check isolation. + +## Remote delivery + +Run a consumer and producer fixture in separate containers on one IPv6 network. +Use role `consumer` on node A and role `producer` on node B. Add `nodeID`, +`locator`, `routerName`, `peerAddress`, and `peerMAC` to each configuration: + +```json +{ + "role": "consumer", + "nodeName": "node-a", + "nodeID": 256, + "locator": "fd00:123:456::/48", + "routerName": "psc-node-a", + "peerAddress": "PEER_IPV6_ADDRESS", + "peerMAC": "PEER_ETHERNET_MAC" +} +``` + +Use node ID 257 for the producer and reverse the peer details. Give each container +its own state file. The fixture creates routing inputs and maps for the real +SRv6 request and return paths. Its kubeconfig also needs permission to manage +BGPRouter and BGPVRFInstance objects. Inspect `/status` on each node to compare +kernel map IDs and rows with the node programming reports in the API. + +## Failure checks + +The local HTTP endpoint at `127.0.0.1:18081` supports `GET /status` and +`POST /pause`, `/resume`, `/restart`, `/reconcile`, and `/delay`. + +- Pause the manager, wait beyond the policy authorization deadline, and verify + that new UDP/TCP requests fail while producer listeners remain available. +- Send a delayed request, let authorization expire, and verify that its reply + does not return to the consumer. `/delay` accepts `{"seconds":10}` and + `/status` includes timestamps for request receipt and the reply attempt. +- Renew authorization and verify recovery without restarting the controller. +- Create conflicting frontend policies in either order, remove one, and verify + that the remaining policy recovers. +- Delete and recreate an endpoint or consumer VPC and verify that stale grants + and node reports do not authorize the replacement. + +The fixture assists attachment status, Linux interfaces, VRFs, and routing maps. +It does not qualify normal CNI lifecycle, BGP convergence, federation, or regional +deployment. Unit and kernel regression tests accompany the implementation PRs; +this fixture provides an explicit environment for combined qualification. diff --git a/tests/psc/example.yaml b/tests/psc/example.yaml new file mode 100644 index 00000000..abc918f5 --- /dev/null +++ b/tests/psc/example.yaml @@ -0,0 +1,83 @@ +# Apply in a disposable edge API with existing VPCs and attachments. +# Consumer attachments need access: a or access: b; the shared producer +# attachment needs role: producer. Use envsubst to fill the variables. +apiVersion: network.datumapis.com/v1alpha1 +kind: ServiceEndpoint +metadata: + name: service-a + namespace: psc-test +spec: + # Each service destination identifies one isolated producer context. + address: fd70:100::10 + protocol: udp + port: 8443 + # Prefer a local producer; allow the remote fixture to serve it otherwise. + deliveryMode: PreferNodeLocal + attachmentSelector: + matchLabels: + role: producer +--- +apiVersion: network.datumapis.com/v1alpha1 +kind: ServiceEndpoint +metadata: + name: service-b + namespace: psc-test +spec: + address: fd70:100::11 + protocol: udp + port: 8443 + deliveryMode: PreferNodeLocal + attachmentSelector: + matchLabels: + role: producer +--- +apiVersion: network.datumapis.com/v1alpha1 +kind: ServiceRoutePolicy +metadata: + name: access-a + namespace: psc-test +spec: + serviceRef: + name: service-a + # Pin access to the current consumer VPC lifetime. + consumerVPCRef: + name: consumer-a + uid: ${CONSUMER_A_UID} + # Both VPCs intentionally use the same frontend address and port. + frontend: + address: fd70:ffff::10 + # Set a deadline within two minutes; renew explicitly during recovery tests. + authorization: + validUntil: ${AUTHORIZATION_VALID_UNTIL} + attachmentSelector: + matchLabels: + access: a + protocolPorts: + - protocol: udp + port: 8443 + - protocol: tcp + port: 8443 +--- +apiVersion: network.datumapis.com/v1alpha1 +kind: ServiceRoutePolicy +metadata: + name: access-b + namespace: psc-test +spec: + serviceRef: + name: service-b + consumerVPCRef: + name: consumer-b + uid: ${CONSUMER_B_UID} + frontend: + address: fd70:ffff::10 + authorization: + validUntil: ${AUTHORIZATION_VALID_UNTIL} + attachmentSelector: + matchLabels: + access: b + protocolPorts: + - protocol: udp + port: 8443 + - protocol: tcp + port: 8443 diff --git a/tests/psc/fixture/constants_linux.go b/tests/psc/fixture/constants_linux.go new file mode 100644 index 00000000..f595755b --- /dev/null +++ b/tests/psc/fixture/constants_linux.go @@ -0,0 +1,30 @@ +//go:build linux + +// Copyright 2026 Datum Cloud, Inc. +// SPDX-License-Identifier: AGPL-3.0-or-later + +package main + +const ( + ipAdd = "add" + ipRoute = "route" + ipTable = "table" + producerGateway = "fd00:da:1::1" + ipLink = "link" + ipSet = "set" + ipVia = "via" + consumerCIDR = "fd00:c::2/128" + ipNetNS = "netns" + ipAddr = "addr" + ipExec = "exec" + ipDevice = "dev" + ipName = "name" + ipType = "type" + ipVeth = "veth" + ipPeer = "peer" + ipNoDAD = "nodad" + ipDefault = "default" + ipReplace = "replace" + servicePort = "8443" + relayTable = "100" +) diff --git a/tests/psc/fixture/main_linux.go b/tests/psc/fixture/main_linux.go new file mode 100644 index 00000000..953bb503 --- /dev/null +++ b/tests/psc/fixture/main_linux.go @@ -0,0 +1,398 @@ +// Copyright 2026 Datum Cloud, Inc. +// +// SPDX-License-Identifier: AGPL-3.0-or-later + +// Command psc-fixture runs the production service route +// reconciler against a real edge API and an isolated Linux packet fixture. +package main + +import ( + "context" + "embed" + "encoding/json" + "errors" + "flag" + "log" + "net/http" + "os" + "os/signal" + "path/filepath" + "strconv" + "sync" + "syscall" + "time" + + "github.com/cilium/ebpf" + corev1 "k8s.io/api/core/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/client-go/tools/clientcmd" + ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/client" + controllerconfig "sigs.k8s.io/controller-runtime/pkg/config" + "sigs.k8s.io/controller-runtime/pkg/metrics/server" + + cloudapi "go.datum.net/cloud/api/v1alpha1" + "go.datum.net/galactic/internal/controller" + "go.datum.net/galactic/internal/plumbing/ebpf/prog" + "go.datum.net/galactic/internal/serviceroute" + networkapi "go.datum.net/network/api/v1alpha1" +) + +//go:embed relay.py query.py +var scripts embed.FS + +type fixtureConfig struct { + Role string `json:"role"` + NodeID int32 `json:"nodeID"` + Locator string `json:"locator"` + RouterName string `json:"routerName"` + PeerAddress string `json:"peerAddress"` + PeerMAC string `json:"peerMAC"` + + Kubeconfig string `json:"kubeconfig"` + Namespace string `json:"namespace"` + NodeName string `json:"nodeName"` + State string `json:"state"` + Cases []fixtureCase `json:"cases"` +} + +type fixtureCase struct { + Project string `json:"project"` + VPCName string `json:"vpcName"` + VPCUID string `json:"vpcUID"` + VPCIdentity string `json:"vpcIdentity"` + Destination string `json:"destination"` + ConsumerAttachment string `json:"consumerAttachment"` + ProducerAttachment string `json:"producerAttachment"` +} + +type fixture struct { + objects *prog.UsidObjects + block uint64 + + cfg fixtureConfig + api client.Client + scheme *runtime.Scheme + labs []labState + pinDir string + mu sync.Mutex + cancel context.CancelFunc + done chan error + reconciler *controller.ServiceRoutePolicyReconciler + running bool + revision int + cleanup []func() + relayMu sync.Mutex + destinations map[string]bool + rootSource, rootDevice string + producerNS, producerHost string +} + +func main() { + configPath := flag.String("config", "/runtime/psc-fixture.json", "fixture configuration JSON") + flag.Parse() + if err := run(*configPath); err != nil { + log.Fatal(err) + } +} + +func run(path string) error { + raw, err := os.ReadFile(path) + if err != nil { + return err + } + f := &fixture{destinations: make(map[string]bool)} + if err := json.Unmarshal(raw, &f.cfg); err != nil { + return err + } + if len(f.cfg.Cases) != 2 || f.cfg.NodeName == "" || f.cfg.Namespace == "" { + return errors.New("two cases, namespace, and nodeName required") + } + if f.cfg.State == "" { + f.cfg.State = "/runtime/psc-fixture-state.json" + } + if err := os.MkdirAll(filepath.Dir(f.cfg.State), 0755); err != nil { + return err + } + for _, name := range []string{"relay.py", "query.py"} { + data, err := scripts.ReadFile(name) + if err != nil { + return err + } + if err := os.WriteFile(filepath.Join(filepath.Dir(f.cfg.State), "psc-"+name), data, 0600); err != nil { + return err + } + } + f.scheme = runtime.NewScheme() + for _, add := range []func(*runtime.Scheme) error{corev1.AddToScheme, cloudapi.AddToScheme, networkapi.AddToScheme} { + if err := add(f.scheme); err != nil { + return err + } + } + cfg, err := clientcmd.BuildConfigFromFlags("", f.cfg.Kubeconfig) + if err != nil { + return err + } + cfg.QPS, cfg.Burst = 50, 100 + f.api, err = client.New(cfg, client.Options{Scheme: f.scheme}) + if err != nil { + return err + } + ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) + defer func() { + cancel() + f.close() + }() + if err := f.setupNetwork(ctx); err != nil { + return err + } + if err := f.startController(ctx); err != nil { + return err + } + // Keep retired destinations listening while testing authorization withdrawal. + go f.followDestinations(ctx) + mux := http.NewServeMux() + mux.HandleFunc("/status", f.handleStatus) + mux.HandleFunc("/delay", f.handleDelay) + for _, action := range []string{"pause", "resume", "restart", "reconcile"} { + mux.HandleFunc("/"+action, f.handleControl(ctx, action)) + } + httpServer := &http.Server{Addr: "127.0.0.1:18081", Handler: mux, ReadHeaderTimeout: 5 * time.Second} + go func() { <-ctx.Done(); _ = httpServer.Shutdown(context.Background()) }() + log.Printf("fixture ready: state=%s control=%s", f.cfg.State, httpServer.Addr) + if err := httpServer.ListenAndServe(); !errors.Is(err, http.ErrServerClosed) { + return err + } + return nil +} + +func (f *fixture) startController(ctx context.Context) error { + f.mu.Lock() + defer f.mu.Unlock() + if f.running { + return nil + } + cfg, err := clientcmd.BuildConfigFromFlags("", f.cfg.Kubeconfig) + if err != nil { + return err + } + cfg.QPS, cfg.Burst = 50, 100 + // The fixture fully stops each manager before replacing it in this process. + skipNameValidation := true + manager, err := ctrl.NewManager(cfg, ctrl.Options{ + Scheme: f.scheme, Metrics: server.Options{BindAddress: "0"}, HealthProbeBindAddress: "0", + Controller: controllerconfig.Controller{SkipNameValidation: &skipNameValidation}, + }) + if err != nil { + return err + } + programmer := &serviceroute.EBPFRouteProgrammer{PinDir: f.pinDir} + if err := programmer.Initialize(); err != nil { + return err + } + r := &controller.ServiceRoutePolicyReconciler{Client: manager.GetClient(), Scheme: f.scheme, + NodeName: f.cfg.NodeName, BGPNamespace: f.cfg.Namespace, Programmer: programmer, FrontendEnabled: true} + if err := r.SetupWithManager(manager); err != nil { + return err + } + child, cancel := context.WithCancel(ctx) + done := make(chan error, 1) + go func() { done <- manager.Start(child) }() + syncCtx, syncCancel := context.WithTimeout(child, 30*time.Second) + defer syncCancel() + if !manager.GetCache().WaitForCacheSync(syncCtx) { + cancel() + return errors.New("edge API cache failed to synchronize") + } + f.cancel, f.done, f.reconciler, f.running = cancel, done, r, true + f.revision++ + return f.writeStateLocked() +} + +func (f *fixture) pauseController() error { + f.mu.Lock() + defer f.mu.Unlock() + if !f.running { + return nil + } + f.cancel() + select { + case err := <-f.done: + if err != nil { + return err + } + case <-time.After(15 * time.Second): + return errors.New("controller failed to stop") + } + f.running = false + f.reconciler = nil + return f.writeStateLocked() +} + +func (f *fixture) reconcile(ctx context.Context) error { + f.mu.Lock() + defer f.mu.Unlock() + if !f.running { + return errors.New("controller is paused") + } + policies := &networkapi.ServiceRoutePolicyList{} + if err := f.api.List(ctx, policies, client.InNamespace(f.cfg.Namespace)); err != nil { + return err + } + for _, policy := range policies.Items { + request := ctrl.Request{NamespacedName: client.ObjectKeyFromObject(&policy)} + if _, err := f.reconciler.Reconcile(ctx, request); err != nil { + return err + } + } + return nil +} + +func (f *fixture) handleControl(ctx context.Context, action string) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + http.Error(w, "POST required", http.StatusMethodNotAllowed) + return + } + var err error + switch action { + case "pause": + err = f.pauseController() + case "resume": + err = f.startController(ctx) + case "restart": + err = f.pauseController() + if err == nil { + err = f.startController(ctx) + } + case "reconcile": + err = f.reconcile(r.Context()) + } + if err != nil { + log.Printf("controller control %s: %v", action, err) + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + f.handleStatus(w, r) + } +} + +func (f *fixture) stateLocked() map[string]any { + state := map[string]any{"nodeName": f.cfg.NodeName, "pinDir": f.pinDir, "running": f.running, + "revision": f.revision, "pid": os.Getpid(), "labs": f.labs, "cases": f.labs, "producerNS": f.producerNS, + "role": f.cfg.Role, + "fixtureAssistance": "Attachment lifecycle, VRF/CNI/SRv6 maps, BGP desired inputs and one-hop underlay", + + "readinessSource": "production ServiceRoutePolicyReconciler after kernel programming"} + if f.objects != nil { + maps := map[string]*ebpf.Map{"routes": f.objects.ServiceRouteTable, "remoteGrants": f.objects.ServiceRemoteGrantTable, + "reverse": f.objects.ServiceReverseTable, "identities": f.objects.AttachmentIdentityTable} + ids := map[string]uint32{} + counts := map[string]int{} + for name, kernelMap := range maps { + info, err := kernelMap.Info() + if err == nil { + if id, ok := info.ID(); ok { + ids[name] = uint32(id) + } + } + key, value := make([]byte, kernelMap.KeySize()), make([]byte, kernelMap.ValueSize()) + iterator := kernelMap.Iterate() + for iterator.Next(&key, &value) { + counts[name]++ + } + if err := iterator.Err(); err != nil { + log.Printf("read fixture map %s: %v", name, err) + } + } + state["mapIDs"], state["mapRows"] = ids, counts + } + raw, err := os.ReadFile(filepath.Join(filepath.Dir(f.cfg.State), "psc-delay-trace.json")) + if err == nil { + var trace map[string]any + if err := json.Unmarshal(raw, &trace); err == nil { + state["delayTrace"] = trace + } + } + return state +} + +func (f *fixture) writeStateLocked() error { + raw, err := json.MarshalIndent(f.stateLocked(), "", " ") + if err != nil { + return err + } + temp := f.cfg.State + ".tmp" + if err := os.WriteFile(temp, raw, 0600); err != nil { + return err + } + return os.Rename(temp, f.cfg.State) +} + +func (f *fixture) handleStatus(w http.ResponseWriter, _ *http.Request) { + f.mu.Lock() + defer f.mu.Unlock() + w.Header().Set("Content-Type", "application/json") + if err := json.NewEncoder(w).Encode(f.stateLocked()); err != nil { + log.Printf("write fixture state: %v", err) + } +} + +func (f *fixture) close() { + if err := f.pauseController(); err != nil { + log.Printf("stop controller: %v", err) + } + f.relayMu.Lock() + defer f.relayMu.Unlock() + for i := len(f.cleanup) - 1; i >= 0; i-- { + f.cleanup[i]() + } +} + +func (f *fixture) handleDelay(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + http.Error(w, "POST required", http.StatusMethodNotAllowed) + return + } + var delay struct { + Seconds float64 `json:"seconds"` + Token string `json:"token"` + } + if err := json.NewDecoder(r.Body).Decode(&delay); err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + if delay.Seconds <= 0 || delay.Seconds > 30 { + http.Error(w, "seconds must be in (0,30]", http.StatusBadRequest) + return + } + delay.Token = strconv.FormatInt(time.Now().UnixNano(), 10) + raw, err := json.Marshal(delay) + if err == nil { + path := filepath.Join(filepath.Dir(f.cfg.State), "psc-delay.json") + tracePath := filepath.Join(filepath.Dir(f.cfg.State), "psc-delay-trace.json") + if removeErr := os.Remove(tracePath); removeErr != nil && !errors.Is(removeErr, os.ErrNotExist) { + err = removeErr + } + if err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + err = os.WriteFile(path+".tmp", raw, 0600) + if err == nil { + err = os.Rename(path+".tmp", path) + } + } + if err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + f.mu.Lock() + defer f.mu.Unlock() + state := f.stateLocked() + state["delayToken"] = delay.Token + w.Header().Set("Content-Type", "application/json") + if err := json.NewEncoder(w).Encode(state); err != nil { + log.Printf("write delay state: %v", err) + } +} diff --git a/tests/psc/fixture/network_linux.go b/tests/psc/fixture/network_linux.go new file mode 100644 index 00000000..a5b3755a --- /dev/null +++ b/tests/psc/fixture/network_linux.go @@ -0,0 +1,307 @@ +// Copyright 2026 Datum Cloud, Inc. +// +// SPDX-License-Identifier: AGPL-3.0-or-later + +package main + +import ( + "bufio" + "context" + "errors" + "fmt" + "log" + "net" + "os" + "os/exec" + "path/filepath" + "strconv" + "strings" + "time" + + meta "k8s.io/apimachinery/pkg/api/meta" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "sigs.k8s.io/controller-runtime/pkg/client" + + cloudapi "go.datum.net/cloud/api/v1alpha1" + "go.datum.net/galactic/internal/plumbing/ebpf/attach" + networkapi "go.datum.net/network/api/v1alpha1" +) + +type labState struct { + Project string `json:"project"` + ConsumerNS string `json:"consumerNS"` + ConsumerHost string `json:"consumerHost"` + ProducerNS string `json:"producerNS"` + ProducerHost string `json:"producerHost"` + Destination string `json:"destination"` +} + +func ip(args ...string) error { + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + output, err := exec.CommandContext(ctx, "ip", args...).CombinedOutput() + if err != nil { + return fmt.Errorf("ip %s: %w: %s", strings.Join(args, " "), err, output) + } + return nil +} + +func (f *fixture) assistAttachment(ctx context.Context, name, host, identity string) error { + attachment := &cloudapi.VPCAttachment{} + if err := f.api.Get(ctx, client.ObjectKey{Namespace: f.cfg.Namespace, Name: name}, attachment); err != nil { + return err + } + before := attachment.DeepCopy() + attachment.Status.Node = f.cfg.NodeName + attachment.Status.HostInterface = host + if identity != "" { + attachment.Status.VPC = identity + } + if attachment.Status.VPC == "" { + attachment.Status.VPC = "fixtureprod" + } + if attachment.Status.VPCAttachment == "" { + attachment.Status.VPCAttachment = host + } + attachment.Status.ObservedGeneration = attachment.Generation + for _, condition := range []string{cloudapi.ConditionTypeReady, cloudapi.ConditionTypeProgrammed} { + meta.SetStatusCondition(&attachment.Status.Conditions, metav1.Condition{Type: condition, Status: metav1.ConditionTrue, + ObservedGeneration: attachment.Generation, Reason: "QualificationFixture", + Message: "Isolated Linux fixture created interface and TC chain"}) + } + return f.api.Status().Patch(ctx, attachment, client.MergeFrom(before)) +} + +func (f *fixture) addDestination(address string) error { + f.relayMu.Lock() + defer f.relayMu.Unlock() + if f.destinations[address] { + return nil + } + parsed := net.ParseIP(address) + if parsed == nil || parsed.To4() != nil { + return fmt.Errorf("fixture destination %q must be IPv6", address) + } + for _, args := range [][]string{ + {ipNetNS, ipExec, f.producerNS, "ip", ipAddr, ipAdd, address + "/128", ipDevice, "p0", ipNoDAD}, + {"-6", ipRoute, ipReplace, address + "/128", ipDevice, f.producerHost}, + {"-6", ipRoute, ipReplace, ipTable, relayTable, address + "/128", ipDevice, f.rootDevice}, + } { + if err := ip(args...); err != nil { + return err + } + } + if f.cfg.Role == roleProducer { + if err := ip("-6", ipRoute, ipReplace, ipTable, "801", address+"/128", ipDevice, f.producerHost); err != nil { + return err + } + } + port := strconv.Itoa(12000 + len(f.destinations)) + script := filepath.Join(filepath.Dir(f.cfg.State), "psc-relay.py") + for _, transport := range []string{"udp", "tcp"} { + for _, mode := range []string{"root", roleProducer} { + args := []string{script, mode, transport, producerGateway, port, address, servicePort, f.rootSource} + command := "python3" + if mode == roleProducer { + args = []string{ipNetNS, ipExec, f.producerNS, "python3", script, mode, transport, + address, servicePort, producerGateway, port, "-"} + command = "ip" + } + process := exec.CommandContext(context.Background(), command, args...) + if err := f.startRelay(process); err != nil { + return err + } + } + } + f.destinations[address] = true + log.Printf("shared producer serves %s via root relay port %s", address, port) + return nil +} + +func (f *fixture) followDestinations(ctx context.Context) { + if f.cfg.Role == roleConsumer { + <-ctx.Done() + return + } + + ticker := time.NewTicker(time.Second) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + endpoints := &networkapi.ServiceEndpointList{} + if err := f.api.List(ctx, endpoints, client.InNamespace(f.cfg.Namespace)); err != nil { + log.Printf("list fixture destinations: %v", err) + continue + } + for _, endpoint := range endpoints.Items { + if endpoint.Spec.Port == 8443 { + if err := f.addDestination(endpoint.Spec.Address); err != nil { + log.Printf("add fixture destination: %v", err) + } + } + } + } + } +} + +func globalIPv6() (string, string, error) { + interfaces, err := net.Interfaces() + if err != nil { + return "", "", err + } + for _, iface := range interfaces { + addresses, err := iface.Addrs() + if err != nil { + return "", "", err + } + for _, address := range addresses { + parsed, _, err := net.ParseCIDR(address.String()) + if err == nil && parsed.To4() == nil && parsed.IsGlobalUnicast() && !parsed.IsLinkLocalUnicast() { + return parsed.String(), iface.Name, nil + } + } + } + return "", "", errors.New("fixture container requires global IPv6 on the fixture underlay") +} + +func (f *fixture) startRelay(process *exec.Cmd) error { + output, err := process.StdoutPipe() + if err != nil { + return err + } + process.Stderr = os.Stderr + if err := process.Start(); err != nil { + return err + } + f.cleanup = append(f.cleanup, func() { _ = process.Process.Kill(); _ = process.Wait() }) + ready := make(chan struct{}) + go func() { + scanner := bufio.NewScanner(output) + marked := false + for scanner.Scan() { + text := scanner.Text() + log.Print(text) + if !marked && strings.HasPrefix(text, "READY ") { + marked = true + close(ready) + } + } + if err := scanner.Err(); err != nil { + log.Printf("read fixture relay: %v", err) + } + }() + select { + case <-ready: + return nil + case <-time.After(5 * time.Second): + return errors.New("fixture relay did not become ready") + } +} + +func (f *fixture) setupNetwork(ctx context.Context) error { + if f.cfg.Role == "" || f.cfg.Role == "local" { + return f.setupLocalNetwork(ctx) + } + return f.setupRemoteNetwork(ctx) +} + +func (f *fixture) setupLocalNetwork(ctx context.Context) error { + source, device, err := globalIPv6() + if err != nil { + return err + } + f.rootSource, f.rootDevice = source, device + suffix := strconv.Itoa(os.Getpid()) + f.producerNS, f.producerHost = "psc-p-"+suffix, "pscp"+suffix + f.pinDir = filepath.Join("/sys/fs/bpf", "galactic-psc-qualification-"+suffix) + if err := os.MkdirAll(f.pinDir, 0755); err != nil { + return err + } + objects, err := attach.Load(f.pinDir) + if err != nil { + return fmt.Errorf("load production TC programs: %w", err) + } + f.objects = objects + f.cleanup = append(f.cleanup, func() { _ = objects.Close(); _ = os.RemoveAll(f.pinDir) }) + if err := ip(ipNetNS, ipAdd, f.producerNS); err != nil { + return err + } + f.cleanup = append(f.cleanup, func() { _ = ip(ipNetNS, "del", f.producerNS) }) + for _, args := range [][]string{ + {ipLink, ipAdd, f.producerHost, ipType, ipVeth, ipPeer, ipName, "p0"}, + {ipLink, ipSet, "p0", ipNetNS, f.producerNS}, + {ipLink, ipSet, f.producerHost, "up"}, + {ipAddr, ipAdd, "fd00:da:1::1/64", ipDevice, f.producerHost, ipNoDAD}, + {ipNetNS, ipExec, f.producerNS, "ip", ipLink, ipSet, "lo", "up"}, + {ipNetNS, ipExec, f.producerNS, "ip", ipAddr, ipAdd, "fd00:da:1::2/64", ipDevice, "p0", ipNoDAD}, + {ipNetNS, ipExec, f.producerNS, "ip", ipLink, ipSet, "p0", "up"}, + {ipNetNS, ipExec, f.producerNS, "ip", ipRoute, ipAdd, ipDefault, ipVia, producerGateway}, + {"-6", "rule", ipAdd, "priority", relayTable, "from", source + "/128", ipTable, relayTable}, + } { + if err := ip(args...); err != nil { + return err + } + } + f.cleanup = append(f.cleanup, func() { _ = ip("-6", "rule", "del", "priority", relayTable) }) + producerLink, err := net.InterfaceByName(f.producerHost) + if err != nil { + return err + } + if err := attach.AttachEgressWithIdentity(objects.UsidServiceEgress, objects.UsidEgress, + objects.AttachmentIdentityTable, uint32(producerLink.Index), f.producerHost); err != nil { + return err + } + producerName := f.cfg.Cases[0].ProducerAttachment + if producerName == "" || producerName != f.cfg.Cases[1].ProducerAttachment { + return errors.New("cases require one shared producer attachment") + } + if err := f.assistAttachment(ctx, producerName, f.producerHost, ""); err != nil { + return err + } + for index, entry := range f.cfg.Cases { + letter := string(rune('a' + index)) + lab := labState{Project: entry.Project, ConsumerNS: "psc-c-" + letter + suffix, + ConsumerHost: "pscc" + letter + suffix, + ProducerNS: f.producerNS, ProducerHost: f.producerHost, Destination: entry.Destination} + gateway := "fd00:ca:" + strconv.Itoa(index+1) + "::1" + if err := ip(ipNetNS, ipAdd, lab.ConsumerNS); err != nil { + return err + } + ns := lab.ConsumerNS + f.cleanup = append(f.cleanup, func() { _ = ip(ipNetNS, "del", ns) }) + for _, args := range [][]string{ + {ipLink, ipAdd, lab.ConsumerHost, ipType, ipVeth, ipPeer, ipName, "c0"}, + {ipLink, ipSet, "c0", ipNetNS, lab.ConsumerNS}, + {ipLink, ipSet, lab.ConsumerHost, "up"}, + {ipAddr, ipAdd, gateway + "/64", ipDevice, lab.ConsumerHost, ipNoDAD}, + {ipNetNS, ipExec, lab.ConsumerNS, "ip", ipLink, ipSet, "lo", "up"}, + {ipNetNS, ipExec, lab.ConsumerNS, "ip", ipAddr, ipAdd, consumerCIDR, ipDevice, "c0", ipNoDAD}, + {ipNetNS, ipExec, lab.ConsumerNS, "ip", ipLink, ipSet, "c0", "up"}, + {ipNetNS, ipExec, lab.ConsumerNS, "ip", ipRoute, ipAdd, gateway + "/128", ipDevice, "c0"}, + {ipNetNS, ipExec, lab.ConsumerNS, "ip", ipRoute, ipAdd, "fd70:ffff::10/128", ipVia, gateway}, + } { + if err := ip(args...); err != nil { + return err + } + } + consumerLink, err := net.InterfaceByName(lab.ConsumerHost) + if err != nil { + return err + } + if err := attach.AttachEgressWithIdentity(objects.UsidServiceEgress, objects.UsidEgress, + objects.AttachmentIdentityTable, uint32(consumerLink.Index), lab.ConsumerHost); err != nil { + return err + } + if err := f.assistAttachment(ctx, entry.ConsumerAttachment, lab.ConsumerHost, entry.VPCIdentity); err != nil { + return err + } + if err := f.addDestination(entry.Destination); err != nil { + return err + } + f.labs = append(f.labs, lab) + } + return nil +} diff --git a/tests/psc/fixture/query.py b/tests/psc/fixture/query.py new file mode 100644 index 00000000..2d75dda3 --- /dev/null +++ b/tests/psc/fixture/query.py @@ -0,0 +1,47 @@ +#!/usr/bin/env python3 +"""Check the same consumer address and source tuple in different VPCs.""" +import json +import os +import socket +import sys + +transport, expected, payload = sys.argv[1:4] +timeout = float(sys.argv[4]) if len(sys.argv) > 4 else 2 +frontend = os.environ.get("PSC_FRONTEND", "fd70:ffff::10") +port = int(os.environ.get("PSC_PORT", "8443")) + + +def receive_exact(conn, length): + data = b"" + while len(data) < length: + chunk = conn.recv(length - len(data)) + if not chunk: + raise OSError("service frame truncated") + data += chunk + return data + + +kind = socket.SOCK_DGRAM if transport == "udp" else socket.SOCK_STREAM +with socket.socket(socket.AF_INET6, kind) as client: + client.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + client.settimeout(timeout) + client.bind(("fd00:c::2", 40530)) + try: + request = payload.encode() + if transport == "tcp": + client.connect((frontend, port)) + client.sendall(len(request).to_bytes(2, "big") + request) + response = receive_exact(client, int.from_bytes(receive_exact(client, 2), "big")) + peer = client.getpeername()[0] + else: + client.sendto(request, (frontend, port)) + response, address = client.recvfrom(65535) + peer = address[0] + except OSError as error: + print(json.dumps({"transport": transport, "timeout": True, "error": str(error)})) + sys.exit(3) +result = json.loads(response) +result.update({"transport": transport, "peer": peer, "timeout": False}) +print(json.dumps(result)) +if peer != frontend or result["destination"] != expected or result["payload"] != payload: + raise RuntimeError("response reached the wrong consumer or producer") diff --git a/tests/psc/fixture/relay.py b/tests/psc/fixture/relay.py new file mode 100644 index 00000000..3a1fa1ac --- /dev/null +++ b/tests/psc/fixture/relay.py @@ -0,0 +1,111 @@ +#!/usr/bin/env python3 +"""Persistent fixture relay; network authorization stays in Galactic's TC path.""" +import json +import os +import time +import socket +import sys +import threading + +mode, transport, listen_address, listen_port, target, target_port, source = sys.argv[1:] +socket_type = socket.SOCK_DGRAM if transport == "udp" else socket.SOCK_STREAM + + +def receive_frame(conn): + header = receive_exact(conn, 2) + return receive_exact(conn, int.from_bytes(header, "big")) + + +def receive_exact(conn, length): + data = b"" + while len(data) < length: + chunk = conn.recv(length - len(data)) + if not chunk: + raise OSError("connection closed before service frame completed") + data += chunk + return data + + +def exchange(query): + if mode == "root": + return json.dumps({"destination": target, "payload": query.decode()}).encode() + with socket.socket(socket.AF_INET6, socket_type) as upstream: + upstream.settimeout(3) + if transport == "tcp": + upstream.connect((target, int(target_port))) + upstream.sendall(len(query).to_bytes(2, "big") + query) + return receive_frame(upstream) + upstream.sendto(query, (target, int(target_port))) + response, _ = upstream.recvfrom(65535) + return response + + +def save_trace(trace): + path = os.path.join(os.path.dirname(__file__), "psc-delay-trace.json") + temp = path + f".{os.getpid()}.{threading.get_ident()}" + with open(temp, "w", encoding="utf-8") as trace_file: + json.dump(trace, trace_file) + os.replace(temp, path) + + +def delay_response(received_at): + if mode != "producer": + return + directive = os.path.join(os.path.dirname(__file__), "psc-delay.json") + claim = directive + f".{os.getpid()}.{threading.get_ident()}" + try: + os.rename(directive, claim) + except FileNotFoundError: + return + try: + with open(claim, encoding="utf-8") as source_file: + directive = json.load(source_file) + seconds = directive["seconds"] + finally: + os.unlink(claim) + trace = {"token": directive["token"], "destination": listen_address, "transport": transport, + "requestReceivedAt": received_at, "upstreamResponseAt": time.time(), "seconds": seconds} + save_trace(trace) + print(f"DELAY producer {transport} {seconds}s after upstream response token={trace['token']}", flush=True) + time.sleep(seconds) + trace["replyAttemptAt"] = time.time() + save_trace(trace) + print(f"DELAY_REPLY producer {transport} token={trace['token']} time={trace['replyAttemptAt']}", flush=True) + + +def serve_tcp(conn): + with conn: + conn.settimeout(4) + try: + while True: + query = receive_frame(conn) + received_at = time.time() + response = exchange(query) + delay_response(received_at) + conn.sendall(len(response).to_bytes(2, "big") + response) + except (OSError, socket.timeout) as error: + print(f"{mode} TCP: {error}", flush=True) + + +def serve_udp(listener, query, peer, received_at): + try: + response = exchange(query) + delay_response(received_at) + listener.sendto(response, peer) + except (OSError, socket.timeout) as error: + print(f"{mode} UDP: {error}", flush=True) + + +with socket.socket(socket.AF_INET6, socket_type) as listener: + listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + listener.bind((listen_address, int(listen_port))) + if transport == "tcp": + listener.listen(32) + print(f"READY {mode} {transport} {listen_address}:{listen_port}", flush=True) + while True: + if transport == "tcp": + conn, _ = listener.accept() + threading.Thread(target=serve_tcp, args=(conn,), daemon=True).start() + else: + query, peer = listener.recvfrom(65535) + threading.Thread(target=serve_udp, args=(listener, query, peer, time.time()), daemon=True).start() diff --git a/tests/psc/fixture/relay_test.py b/tests/psc/fixture/relay_test.py new file mode 100644 index 00000000..df711155 --- /dev/null +++ b/tests/psc/fixture/relay_test.py @@ -0,0 +1,45 @@ +"""Check the generic producer protocol without a DNS service.""" +import json +import pathlib +import socket +import subprocess +import unittest + + +class ProducerProtocolTest(unittest.TestCase): + def test_udp_and_tcp_destinations(self): + for transport in ("udp", "tcp"): + with self.subTest(transport=transport): + kind = socket.SOCK_DGRAM if transport == "udp" else socket.SOCK_STREAM + with socket.socket(socket.AF_INET6, kind) as reservation: + reservation.bind(("::1", 0)) + port = reservation.getsockname()[1] + process = subprocess.Popen( + ["python3", str(pathlib.Path(__file__).with_name("relay.py")), + "root", transport, "::1", str(port), "fd70:100::10", "8443", "-"], + stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True) + try: + self.assertTrue(process.stdout.readline().startswith("READY")) + payload = b"consumer-a request" + with socket.socket(socket.AF_INET6, kind) as client: + client.settimeout(3) + if transport == "tcp": + client.connect(("::1", port)) + client.sendall(len(payload).to_bytes(2, "big") + payload) + length = int.from_bytes(client.recv(2), "big") + response = b"" + while len(response) < length: + response += client.recv(length - len(response)) + else: + client.sendto(payload, ("::1", port)) + response, _ = client.recvfrom(65535) + self.assertEqual(json.loads(response), { + "destination": "fd70:100::10", "payload": payload.decode()}) + finally: + process.terminate() + process.wait(timeout=3) + process.stdout.close() + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/psc/fixture/remote_linux.go b/tests/psc/fixture/remote_linux.go new file mode 100644 index 00000000..4612eaf6 --- /dev/null +++ b/tests/psc/fixture/remote_linux.go @@ -0,0 +1,240 @@ +// Copyright 2026 Datum Cloud, Inc. +// +// SPDX-License-Identifier: AGPL-3.0-or-later + +package main + +import ( + "context" + "errors" + "fmt" + "net" + "os" + "path/filepath" + "strconv" + + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/controller/controllerutil" + + cloudapi "go.datum.net/cloud/api/v1alpha1" + "go.datum.net/galactic/internal/crdnames" + "go.datum.net/galactic/internal/plumbing/ebpf/attach" + "go.datum.net/galactic/internal/plumbing/ebpf/prog" + "go.datum.net/galactic/internal/plumbing/ebpf/uformat" + "go.datum.net/galactic/internal/plumbing/ebpf/usidmap" + "go.datum.net/galactic/internal/plumbing/srv6" + networkapi "go.datum.net/network/api/v1alpha1" +) + +const ( + roleConsumer = "consumer" + roleProducer = "producer" +) + +func (f *fixture) setupRemoteNetwork(ctx context.Context) error { + if f.cfg.Role != roleConsumer && f.cfg.Role != roleProducer { + return errors.New("role must be consumer or producer") + } + if f.cfg.RouterName == "" || f.cfg.PeerMAC == "" || f.cfg.NodeID == 0 || f.cfg.Locator == "" { + return errors.New("routerName, nodeID, locator and peerMAC required") + } + source, device, err := globalIPv6() + if err != nil { + return err + } + f.rootSource, f.rootDevice = source, device + suffix := strconv.Itoa(os.Getpid()) + f.pinDir = filepath.Join("/sys/fs/bpf", "psc-remote-"+f.cfg.NodeName+"-"+suffix) + f.objects, err = attach.Load(f.pinDir) + if err != nil { + return err + } + f.cleanup = append(f.cleanup, func() { _ = f.objects.Close(); _ = os.RemoveAll(f.pinDir) }) + if err := f.configureNode(ctx); err != nil { + return err + } + if f.cfg.Role == roleProducer { + return f.setupProducer(ctx, suffix) + } + for index, entry := range f.cfg.Cases { + if err := f.setupConsumer(ctx, entry, index, suffix); err != nil { + return err + } + } + return nil +} + +func (f *fixture) configureNode(ctx context.Context) error { + router := &networkapi.BGPRouter{ObjectMeta: metav1.ObjectMeta{Name: f.cfg.RouterName, Namespace: f.cfg.Namespace}} + if _, err := controllerutil.CreateOrUpdate(ctx, f.api, router, func() error { + router.Spec = networkapi.BGPRouterSpec{TargetRef: networkapi.TargetRef{Kind: "Node", Name: f.cfg.NodeName}, + LocalASN: 65000, RouterID: fmt.Sprintf("192.0.2.%d", f.cfg.NodeID%250+1), + AddressFamilies: []networkapi.AddressFamily{{AFI: "ipv6", SAFI: "unicast"}}, + SRv6Locator: f.cfg.Locator, NodeID: f.cfg.NodeID} + return nil + }); err != nil { + return err + } + if err := f.api.Get(ctx, client.ObjectKeyFromObject(router), router); err != nil { + return err + } + base, err := srv6.NodeSIDBase(router.Spec.SRv6Locator, router.Spec.NodeID) + if err != nil { + return err + } + block, err := uformat.Block(base) + if err != nil { + return err + } + f.block = block + registry := usidmap.NewRegistryFromObjects(f.objects) + if err := registry.Locator.Register(block, uint16(router.Spec.NodeID)); err != nil { + return err + } + if err := registry.Function.Register(block, uformat.FunctionEndDT46); err != nil { + return err + } + if err := f.objects.NodeSrcAddrTable.Put(uint32(0), base.As16()); err != nil { + return err + } + iface, err := net.InterfaceByName(f.rootDevice) + if err != nil { + return err + } + peerMAC, err := net.ParseMAC(f.cfg.PeerMAC) + if err != nil { + return err + } + uplink := prog.UsidPublicUplinkValue{LinkIfindex: uint32(iface.Index)} + copy(uplink.Dmac[:], peerMAC) + copy(uplink.Smac[:], iface.HardwareAddr) + if err := f.objects.PublicUplinkTable.Put(uint32(0), uplink); err != nil { + return err + } + if err := attach.Attach(f.objects.UsidIngress, []string{f.rootDevice}); err != nil { + return err + } + for _, scope := range []string{"all", f.rootDevice} { + if err := os.WriteFile("/proc/sys/net/ipv6/conf/"+scope+"/forwarding", []byte("1"), 0600); err != nil { + return err + } + } + return nil +} + +func (f *fixture) configureAttachment(ctx context.Context, identity, host string, argument uint16) error { + routerRef := &networkapi.RouterRef{Name: f.cfg.RouterName} + object := &networkapi.BGPVRFInstance{ObjectMeta: metav1.ObjectMeta{ + Name: crdnames.BGPVRFInstanceName(identity, f.cfg.NodeName), Namespace: f.cfg.Namespace}} + if _, err := controllerutil.CreateOrUpdate(ctx, f.api, object, func() error { + rt := networkapi.RouteTarget{Value: fmt.Sprintf("65000:%d", argument)} + object.Spec = networkapi.BGPVRFInstanceSpec{RouterTarget: networkapi.RouterTarget{RouterRef: routerRef}, + VRFID: int32(argument), ImportRouteTargets: []networkapi.RouteTarget{rt}, + ExportRouteTargets: []networkapi.RouteTarget{rt}} + return nil + }); err != nil { + return err + } + if err := f.api.Get(ctx, client.ObjectKeyFromObject(object), object); err != nil { + return err + } + actualArgument := uint16(object.Spec.VRFID) + tableID := uint32(500) + uint32(actualArgument) + registry := usidmap.NewRegistryFromObjects(f.objects) + if err := registry.VRF.Register(f.block, actualArgument, tableID, usidmap.EgressKindVeth); err != nil { + return err + } + iface, err := net.InterfaceByName(host) + if err != nil { + return err + } + if err := f.objects.IfindexVrfTable.Put(uint32(iface.Index), prog.UsidIfindexVrfValue{ + Block: f.block, Argument: actualArgument}); err != nil { + return err + } + if err := f.objects.IfindexEgressKindTable.Put(uint32(iface.Index), uint32(0)); err != nil { + return err + } + return attach.AttachEgressWithIdentity(f.objects.UsidServiceEgress, f.objects.UsidEgress, + f.objects.AttachmentIdentityTable, uint32(iface.Index), host) +} + +func (f *fixture) setupConsumer(ctx context.Context, entry fixtureCase, index int, suffix string) error { + letter := string(rune('a' + index)) + lab := labState{Project: entry.Project, ConsumerNS: "pscr-c-" + letter + suffix, + ConsumerHost: "pscrc" + letter + suffix, + Destination: entry.Destination} + gateway := "fd00:ca:" + strconv.Itoa(index+1) + "::1" + if err := ip(ipNetNS, ipAdd, lab.ConsumerNS); err != nil { + return err + } + f.cleanup = append(f.cleanup, func() { _ = ip(ipNetNS, "del", lab.ConsumerNS) }) + for _, args := range [][]string{ + {ipLink, ipAdd, lab.ConsumerHost, ipType, ipVeth, ipPeer, ipName, "c0"}, + {ipLink, ipSet, "c0", ipNetNS, lab.ConsumerNS}, {ipLink, ipSet, lab.ConsumerHost, "up"}, + {ipAddr, ipAdd, gateway + "/64", ipDevice, lab.ConsumerHost, ipNoDAD}, + {ipNetNS, ipExec, lab.ConsumerNS, "ip", ipLink, ipSet, "lo", "up"}, + {ipNetNS, ipExec, lab.ConsumerNS, "ip", ipAddr, ipAdd, consumerCIDR, ipDevice, "c0", ipNoDAD}, + {ipNetNS, ipExec, lab.ConsumerNS, "ip", ipLink, ipSet, "c0", "up"}, + {ipNetNS, ipExec, lab.ConsumerNS, "ip", ipRoute, ipAdd, gateway + "/128", ipDevice, "c0"}, + {ipNetNS, ipExec, lab.ConsumerNS, "ip", ipRoute, ipAdd, "fd70:ffff::10/128", ipVia, gateway}, + {"-6", ipRoute, ipAdd, ipTable, strconv.Itoa(601 + index), consumerCIDR, ipDevice, lab.ConsumerHost}, + } { + if err := ip(args...); err != nil { + return err + } + } + if err := f.assistAttachment(ctx, entry.ConsumerAttachment, lab.ConsumerHost, entry.VPCIdentity); err != nil { + return err + } + if err := f.configureAttachment(ctx, entry.VPCIdentity, lab.ConsumerHost, uint16(101+index)); err != nil { + return err + } + f.labs = append(f.labs, lab) + return nil +} + +func (f *fixture) setupProducer(ctx context.Context, suffix string) error { + producerName := f.cfg.Cases[0].ProducerAttachment + if producerName == "" || producerName != f.cfg.Cases[1].ProducerAttachment { + return errors.New("one shared producer required") + } + f.producerNS, f.producerHost = "pscr-p-"+suffix, "pscrp"+suffix + if err := ip(ipNetNS, ipAdd, f.producerNS); err != nil { + return err + } + f.cleanup = append(f.cleanup, func() { _ = ip(ipNetNS, "del", f.producerNS) }) + for _, args := range [][]string{ + {ipLink, ipAdd, f.producerHost, ipType, ipVeth, ipPeer, ipName, "p0"}, + {ipLink, ipSet, "p0", ipNetNS, f.producerNS}, {ipLink, ipSet, f.producerHost, "up"}, + {ipAddr, ipAdd, "fd00:da:1::1/64", ipDevice, f.producerHost, ipNoDAD}, + {ipNetNS, ipExec, f.producerNS, "ip", ipLink, ipSet, "lo", "up"}, + {ipNetNS, ipExec, f.producerNS, "ip", ipAddr, ipAdd, "fd00:da:1::2/64", ipDevice, "p0", ipNoDAD}, + {ipNetNS, ipExec, f.producerNS, "ip", ipLink, ipSet, "p0", "up"}, + {ipNetNS, ipExec, f.producerNS, "ip", ipRoute, ipAdd, ipDefault, ipVia, producerGateway}, + {"-6", "rule", ipAdd, "priority", relayTable, "from", f.rootSource + "/128", ipTable, relayTable}, + } { + if err := ip(args...); err != nil { + return err + } + } + if err := f.assistAttachment(ctx, producerName, f.producerHost, ""); err != nil { + return err + } + producer := &cloudapi.VPCAttachment{} + if err := f.api.Get(ctx, client.ObjectKey{Namespace: f.cfg.Namespace, Name: producerName}, producer); err != nil { + return err + } + if err := f.configureAttachment(ctx, producer.Status.VPC, f.producerHost, 301); err != nil { + return err + } + for _, entry := range f.cfg.Cases { + if err := f.addDestination(entry.Destination); err != nil { + return err + } + f.labs = append(f.labs, labState{Project: entry.Project, ProducerNS: f.producerNS, ProducerHost: f.producerHost, + Destination: entry.Destination}) + } + return nil +} From 2eea1f5cadb137b0468d969fd7bb81d24f1c98f1 Mon Sep 17 00:00:00 2001 From: Scot Wells Date: Thu, 8 Oct 2026 14:19:19 -0500 Subject: [PATCH 2/3] test: Ignore generated fixture Python caches --- tests/psc/.gitignore | 1 + 1 file changed, 1 insertion(+) create mode 100644 tests/psc/.gitignore diff --git a/tests/psc/.gitignore b/tests/psc/.gitignore new file mode 100644 index 00000000..c18dd8d8 --- /dev/null +++ b/tests/psc/.gitignore @@ -0,0 +1 @@ +__pycache__/ From efaa90839c40364eddf83eb1a4051157ae205f77 Mon Sep 17 00:00:00 2001 From: Scot Wells Date: Thu, 8 Oct 2026 14:31:31 -0500 Subject: [PATCH 3/3] refactor: Replace private service Python helpers with Go --- tests/psc/.gitignore | 1 - tests/psc/README.md | 13 +- tests/psc/fixture/main_linux.go | 24 ++- tests/psc/fixture/network_linux.go | 17 +- tests/psc/fixture/query.py | 47 ----- tests/psc/fixture/relay.py | 111 ----------- tests/psc/fixture/relay_test.py | 45 ----- tests/psc/internal/traffic/command.go | 99 ++++++++++ tests/psc/internal/traffic/relay.go | 211 +++++++++++++++++++++ tests/psc/internal/traffic/socket.go | 16 ++ tests/psc/internal/traffic/traffic.go | 116 +++++++++++ tests/psc/internal/traffic/traffic_test.go | 181 ++++++++++++++++++ 12 files changed, 655 insertions(+), 226 deletions(-) delete mode 100644 tests/psc/.gitignore delete mode 100644 tests/psc/fixture/query.py delete mode 100644 tests/psc/fixture/relay.py delete mode 100644 tests/psc/fixture/relay_test.py create mode 100644 tests/psc/internal/traffic/command.go create mode 100644 tests/psc/internal/traffic/relay.go create mode 100644 tests/psc/internal/traffic/socket.go create mode 100644 tests/psc/internal/traffic/traffic.go create mode 100644 tests/psc/internal/traffic/traffic_test.go diff --git a/tests/psc/.gitignore b/tests/psc/.gitignore deleted file mode 100644 index c18dd8d8..00000000 --- a/tests/psc/.gitignore +++ /dev/null @@ -1 +0,0 @@ -__pycache__/ diff --git a/tests/psc/README.md b/tests/psc/README.md index 52b11dd7..9e8c4473 100644 --- a/tests/psc/README.md +++ b/tests/psc/README.md @@ -10,10 +10,10 @@ Build after generating the eBPF objects: ```sh task build:ebpf go build -o bin/psc-fixture ./tests/psc/fixture -python3 -m unittest discover -s tests/psc/fixture -p '*_test.py' +go test -race ./tests/psc/internal/traffic ``` -Run in a disposable privileged Linux environment with IPv6, `ip`, Python 3, +Run in a disposable privileged Linux environment with IPv6, `ip`, bpffs, and an edge kubeconfig. Supply two existing consumer VPCs and attachments, a shared producer attachment, and the service endpoints and policies from the [example manifest](example.yaml). Apply the Cloud and Network CRDs first. @@ -65,10 +65,15 @@ Start `bin/psc-fixture --config /runtime/psc.json`. The state file identifies consumer namespaces. Query each namespace using its expected producer address: ```sh -ip netns exec CONSUMER_A python3 /runtime/psc-query.py udp fd70:100::10 request-a -ip netns exec CONSUMER_B python3 /runtime/psc-query.py tcp fd70:100::11 request-b +ip netns exec CONSUMER_A "$(pwd)/bin/psc-fixture" query \ + --transport udp --expected fd70:100::10 --payload request-a +ip netns exec CONSUMER_B "$(pwd)/bin/psc-fixture" query \ + --transport tcp --expected fd70:100::11 --payload request-b ``` +The same binary supplies the service relays and consumer queries. +Use `--timeout 15s` for delayed reply checks. + A successful result identifies the expected producer and the frontend `fd70:ffff::10`. A denied or expired request exits with status 3. Repeat both transports from both VPCs using the same source port to check isolation. diff --git a/tests/psc/fixture/main_linux.go b/tests/psc/fixture/main_linux.go index 953bb503..4074ca20 100644 --- a/tests/psc/fixture/main_linux.go +++ b/tests/psc/fixture/main_linux.go @@ -8,7 +8,6 @@ package main import ( "context" - "embed" "encoding/json" "errors" "flag" @@ -35,12 +34,10 @@ import ( "go.datum.net/galactic/internal/controller" "go.datum.net/galactic/internal/plumbing/ebpf/prog" "go.datum.net/galactic/internal/serviceroute" + "go.datum.net/galactic/tests/psc/internal/traffic" networkapi "go.datum.net/network/api/v1alpha1" ) -//go:embed relay.py query.py -var scripts embed.FS - type fixtureConfig struct { Role string `json:"role"` NodeID int32 `json:"nodeID"` @@ -89,6 +86,16 @@ type fixture struct { } func main() { + if len(os.Args) > 1 && (os.Args[1] == "query" || os.Args[1] == "relay") { + ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) + defer cancel() + code, err := traffic.Run(ctx, os.Args[1:], os.Stdout) + if err != nil { + log.Print(err) + } + os.Exit(code) + } + configPath := flag.String("config", "/runtime/psc-fixture.json", "fixture configuration JSON") flag.Parse() if err := run(*configPath); err != nil { @@ -114,15 +121,6 @@ func run(path string) error { if err := os.MkdirAll(filepath.Dir(f.cfg.State), 0755); err != nil { return err } - for _, name := range []string{"relay.py", "query.py"} { - data, err := scripts.ReadFile(name) - if err != nil { - return err - } - if err := os.WriteFile(filepath.Join(filepath.Dir(f.cfg.State), "psc-"+name), data, 0600); err != nil { - return err - } - } f.scheme = runtime.NewScheme() for _, add := range []func(*runtime.Scheme) error{corev1.AddToScheme, cloudapi.AddToScheme, networkapi.AddToScheme} { if err := add(f.scheme); err != nil { diff --git a/tests/psc/fixture/network_linux.go b/tests/psc/fixture/network_linux.go index a5b3755a..3804a23d 100644 --- a/tests/psc/fixture/network_linux.go +++ b/tests/psc/fixture/network_linux.go @@ -97,14 +97,21 @@ func (f *fixture) addDestination(address string) error { } } port := strconv.Itoa(12000 + len(f.destinations)) - script := filepath.Join(filepath.Dir(f.cfg.State), "psc-relay.py") + binary, err := os.Executable() + if err != nil { + return err + } for _, transport := range []string{"udp", "tcp"} { for _, mode := range []string{"root", roleProducer} { - args := []string{script, mode, transport, producerGateway, port, address, servicePort, f.rootSource} - command := "python3" + listen, target := net.JoinHostPort(producerGateway, port), net.JoinHostPort(address, servicePort) + if mode == roleProducer { + listen, target = target, listen + } + args := []string{"relay", "--mode", mode, "--transport", transport, + "--listen", listen, "--target", target, "--state-dir", filepath.Dir(f.cfg.State)} + command := binary if mode == roleProducer { - args = []string{ipNetNS, ipExec, f.producerNS, "python3", script, mode, transport, - address, servicePort, producerGateway, port, "-"} + args = append([]string{ipNetNS, ipExec, f.producerNS, binary}, args...) command = "ip" } process := exec.CommandContext(context.Background(), command, args...) diff --git a/tests/psc/fixture/query.py b/tests/psc/fixture/query.py deleted file mode 100644 index 2d75dda3..00000000 --- a/tests/psc/fixture/query.py +++ /dev/null @@ -1,47 +0,0 @@ -#!/usr/bin/env python3 -"""Check the same consumer address and source tuple in different VPCs.""" -import json -import os -import socket -import sys - -transport, expected, payload = sys.argv[1:4] -timeout = float(sys.argv[4]) if len(sys.argv) > 4 else 2 -frontend = os.environ.get("PSC_FRONTEND", "fd70:ffff::10") -port = int(os.environ.get("PSC_PORT", "8443")) - - -def receive_exact(conn, length): - data = b"" - while len(data) < length: - chunk = conn.recv(length - len(data)) - if not chunk: - raise OSError("service frame truncated") - data += chunk - return data - - -kind = socket.SOCK_DGRAM if transport == "udp" else socket.SOCK_STREAM -with socket.socket(socket.AF_INET6, kind) as client: - client.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) - client.settimeout(timeout) - client.bind(("fd00:c::2", 40530)) - try: - request = payload.encode() - if transport == "tcp": - client.connect((frontend, port)) - client.sendall(len(request).to_bytes(2, "big") + request) - response = receive_exact(client, int.from_bytes(receive_exact(client, 2), "big")) - peer = client.getpeername()[0] - else: - client.sendto(request, (frontend, port)) - response, address = client.recvfrom(65535) - peer = address[0] - except OSError as error: - print(json.dumps({"transport": transport, "timeout": True, "error": str(error)})) - sys.exit(3) -result = json.loads(response) -result.update({"transport": transport, "peer": peer, "timeout": False}) -print(json.dumps(result)) -if peer != frontend or result["destination"] != expected or result["payload"] != payload: - raise RuntimeError("response reached the wrong consumer or producer") diff --git a/tests/psc/fixture/relay.py b/tests/psc/fixture/relay.py deleted file mode 100644 index 3a1fa1ac..00000000 --- a/tests/psc/fixture/relay.py +++ /dev/null @@ -1,111 +0,0 @@ -#!/usr/bin/env python3 -"""Persistent fixture relay; network authorization stays in Galactic's TC path.""" -import json -import os -import time -import socket -import sys -import threading - -mode, transport, listen_address, listen_port, target, target_port, source = sys.argv[1:] -socket_type = socket.SOCK_DGRAM if transport == "udp" else socket.SOCK_STREAM - - -def receive_frame(conn): - header = receive_exact(conn, 2) - return receive_exact(conn, int.from_bytes(header, "big")) - - -def receive_exact(conn, length): - data = b"" - while len(data) < length: - chunk = conn.recv(length - len(data)) - if not chunk: - raise OSError("connection closed before service frame completed") - data += chunk - return data - - -def exchange(query): - if mode == "root": - return json.dumps({"destination": target, "payload": query.decode()}).encode() - with socket.socket(socket.AF_INET6, socket_type) as upstream: - upstream.settimeout(3) - if transport == "tcp": - upstream.connect((target, int(target_port))) - upstream.sendall(len(query).to_bytes(2, "big") + query) - return receive_frame(upstream) - upstream.sendto(query, (target, int(target_port))) - response, _ = upstream.recvfrom(65535) - return response - - -def save_trace(trace): - path = os.path.join(os.path.dirname(__file__), "psc-delay-trace.json") - temp = path + f".{os.getpid()}.{threading.get_ident()}" - with open(temp, "w", encoding="utf-8") as trace_file: - json.dump(trace, trace_file) - os.replace(temp, path) - - -def delay_response(received_at): - if mode != "producer": - return - directive = os.path.join(os.path.dirname(__file__), "psc-delay.json") - claim = directive + f".{os.getpid()}.{threading.get_ident()}" - try: - os.rename(directive, claim) - except FileNotFoundError: - return - try: - with open(claim, encoding="utf-8") as source_file: - directive = json.load(source_file) - seconds = directive["seconds"] - finally: - os.unlink(claim) - trace = {"token": directive["token"], "destination": listen_address, "transport": transport, - "requestReceivedAt": received_at, "upstreamResponseAt": time.time(), "seconds": seconds} - save_trace(trace) - print(f"DELAY producer {transport} {seconds}s after upstream response token={trace['token']}", flush=True) - time.sleep(seconds) - trace["replyAttemptAt"] = time.time() - save_trace(trace) - print(f"DELAY_REPLY producer {transport} token={trace['token']} time={trace['replyAttemptAt']}", flush=True) - - -def serve_tcp(conn): - with conn: - conn.settimeout(4) - try: - while True: - query = receive_frame(conn) - received_at = time.time() - response = exchange(query) - delay_response(received_at) - conn.sendall(len(response).to_bytes(2, "big") + response) - except (OSError, socket.timeout) as error: - print(f"{mode} TCP: {error}", flush=True) - - -def serve_udp(listener, query, peer, received_at): - try: - response = exchange(query) - delay_response(received_at) - listener.sendto(response, peer) - except (OSError, socket.timeout) as error: - print(f"{mode} UDP: {error}", flush=True) - - -with socket.socket(socket.AF_INET6, socket_type) as listener: - listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) - listener.bind((listen_address, int(listen_port))) - if transport == "tcp": - listener.listen(32) - print(f"READY {mode} {transport} {listen_address}:{listen_port}", flush=True) - while True: - if transport == "tcp": - conn, _ = listener.accept() - threading.Thread(target=serve_tcp, args=(conn,), daemon=True).start() - else: - query, peer = listener.recvfrom(65535) - threading.Thread(target=serve_udp, args=(listener, query, peer, time.time()), daemon=True).start() diff --git a/tests/psc/fixture/relay_test.py b/tests/psc/fixture/relay_test.py deleted file mode 100644 index df711155..00000000 --- a/tests/psc/fixture/relay_test.py +++ /dev/null @@ -1,45 +0,0 @@ -"""Check the generic producer protocol without a DNS service.""" -import json -import pathlib -import socket -import subprocess -import unittest - - -class ProducerProtocolTest(unittest.TestCase): - def test_udp_and_tcp_destinations(self): - for transport in ("udp", "tcp"): - with self.subTest(transport=transport): - kind = socket.SOCK_DGRAM if transport == "udp" else socket.SOCK_STREAM - with socket.socket(socket.AF_INET6, kind) as reservation: - reservation.bind(("::1", 0)) - port = reservation.getsockname()[1] - process = subprocess.Popen( - ["python3", str(pathlib.Path(__file__).with_name("relay.py")), - "root", transport, "::1", str(port), "fd70:100::10", "8443", "-"], - stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True) - try: - self.assertTrue(process.stdout.readline().startswith("READY")) - payload = b"consumer-a request" - with socket.socket(socket.AF_INET6, kind) as client: - client.settimeout(3) - if transport == "tcp": - client.connect(("::1", port)) - client.sendall(len(payload).to_bytes(2, "big") + payload) - length = int.from_bytes(client.recv(2), "big") - response = b"" - while len(response) < length: - response += client.recv(length - len(response)) - else: - client.sendto(payload, ("::1", port)) - response, _ = client.recvfrom(65535) - self.assertEqual(json.loads(response), { - "destination": "fd70:100::10", "payload": payload.decode()}) - finally: - process.terminate() - process.wait(timeout=3) - process.stdout.close() - - -if __name__ == "__main__": - unittest.main() diff --git a/tests/psc/internal/traffic/command.go b/tests/psc/internal/traffic/command.go new file mode 100644 index 00000000..a50c1bd8 --- /dev/null +++ b/tests/psc/internal/traffic/command.go @@ -0,0 +1,99 @@ +// Copyright 2026 Datum Cloud, Inc. +// SPDX-License-Identifier: AGPL-3.0-or-later + +package traffic + +import ( + "context" + "encoding/json" + "errors" + "flag" + "io" + "net" + "time" +) + +const ( + commandQuery = "query" + transportTCP = "tcp" + transportUDP = "udp" + modeRoot = "root" + modeProducer = "producer" +) + +// Run executes a query or relay subcommand and returns its process exit status. +func Run(ctx context.Context, args []string, output io.Writer) (int, error) { + if len(args) == 0 { + return 2, errors.New("expected query or relay command") + } + if args[0] == "relay" { + return runRelay(ctx, args[1:], output) + } + if args[0] != commandQuery { + return 2, errors.New("expected query or relay command") + } + flags := flag.NewFlagSet(commandQuery, flag.ContinueOnError) + flags.SetOutput(io.Discard) + cfg := QueryConfig{} + flags.StringVar(&cfg.Transport, "transport", transportUDP, "udp or tcp") + flags.StringVar(&cfg.Frontend, "frontend", "[fd70:ffff::10]:8443", "consumer frontend address") + flags.StringVar(&cfg.Source, "source", "[fd00:c::2]:40530", "consumer source address") + flags.StringVar(&cfg.Expected, "expected", "", "expected producer IP") + flags.StringVar(&cfg.Payload, "payload", "", "request payload") + flags.DurationVar(&cfg.Timeout, "timeout", 2*time.Second, "request timeout") + if err := flags.Parse(args[1:]); err != nil { + return 2, err + } + if flags.NArg() != 0 || !validTransport(cfg.Transport) || cfg.Expected == "" || cfg.Timeout <= 0 { + return 2, errors.New("query requires udp or tcp, expected producer IP, and a positive timeout") + } + reply, peer, queryErr := Query(ctx, cfg) + result := struct { + Reply + Transport string `json:"transport"` + Peer string `json:"peer"` + Timeout bool `json:"timeout"` + Error string `json:"error,omitempty"` + }{Reply: reply, Transport: cfg.Transport, Peer: peer} + var networkErr net.Error + if queryErr != nil { + result.Error = queryErr.Error() + result.Timeout = errors.As(queryErr, &networkErr) + } + if err := json.NewEncoder(output).Encode(result); err != nil { + return 1, err + } + if result.Timeout { + return 3, queryErr + } + if queryErr != nil { + return 1, queryErr + } + return 0, nil +} + +func runRelay(ctx context.Context, args []string, output io.Writer) (int, error) { + flags := flag.NewFlagSet("relay", flag.ContinueOnError) + flags.SetOutput(io.Discard) + cfg := RelayConfig{} + flags.StringVar(&cfg.Mode, "mode", modeRoot, "root responder or producer relay") + flags.StringVar(&cfg.Transport, "transport", transportUDP, "udp or tcp") + flags.StringVar(&cfg.Listen, "listen", "", "listener address") + flags.StringVar(&cfg.Target, "target", "", "producer identity or upstream address") + flags.StringVar(&cfg.StateDir, "state-dir", "", "response delay state directory") + if err := flags.Parse(args); err != nil { + return 2, err + } + if flags.NArg() != 0 || !validTransport(cfg.Transport) || (cfg.Mode != modeRoot && cfg.Mode != modeProducer) || + cfg.Listen == "" || cfg.Target == "" || (cfg.Mode == modeProducer && cfg.StateDir == "") { + return 2, errors.New("relay requires a valid mode, transport, listen address, target, and producer state directory") + } + if err := Serve(ctx, cfg, output); err != nil { + return 1, err + } + return 0, nil +} + +func validTransport(transport string) bool { + return transport == transportTCP || transport == transportUDP +} diff --git a/tests/psc/internal/traffic/relay.go b/tests/psc/internal/traffic/relay.go new file mode 100644 index 00000000..f6c17946 --- /dev/null +++ b/tests/psc/internal/traffic/relay.go @@ -0,0 +1,211 @@ +// Copyright 2026 Datum Cloud, Inc. +// SPDX-License-Identifier: AGPL-3.0-or-later + +package traffic + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "io" + "log" + "net" + "os" + "path/filepath" + "sync" + "time" +) + +// RelayConfig selects a producer relay or the shared responding service. +type RelayConfig struct { + Mode, Transport, Listen, Target, StateDir string +} + +// Serve handles UDP datagrams or framed TCP requests until ctx is canceled. +func Serve(ctx context.Context, cfg RelayConfig, output io.Writer) error { + var ready sync.WaitGroup + defer ready.Wait() + listen := net.ListenConfig{Control: reuseAddress} + if cfg.Transport == transportTCP { + listener, err := listen.Listen(ctx, "tcp6", cfg.Listen) + if err != nil { + return err + } + defer func() { _ = listener.Close() }() + stop := context.AfterFunc(ctx, func() { _ = listener.Close() }) + defer stop() + if _, err := fmt.Fprintf(output, "READY %s %s %s\n", cfg.Mode, cfg.Transport, listener.Addr()); err != nil { + return err + } + for { + conn, err := listener.Accept() + if err != nil { + if ctx.Err() != nil { + return nil + } + return err + } + ready.Add(1) + go func() { defer ready.Done(); serveTCP(ctx, cfg, conn) }() + } + } + listener, err := listen.ListenPacket(ctx, "udp6", cfg.Listen) + if err != nil { + return err + } + defer func() { _ = listener.Close() }() + stop := context.AfterFunc(ctx, func() { _ = listener.Close() }) + defer stop() + if _, err := fmt.Fprintf(output, "READY %s %s %s\n", cfg.Mode, cfg.Transport, listener.LocalAddr()); err != nil { + return err + } + for { + payload := make([]byte, maxFrameSize) + n, peer, err := listener.ReadFrom(payload) + if err != nil { + if ctx.Err() != nil { + return nil + } + return err + } + received := time.Now() + ready.Add(1) + go func() { + defer ready.Done() + response, err := relayResponse(ctx, cfg, payload[:n], received) + if err == nil { + _, err = listener.WriteTo(response, peer) + } + if err != nil && ctx.Err() == nil { + log.Printf("relay UDP: %v", err) + } + }() + } +} + +func serveTCP(ctx context.Context, cfg RelayConfig, conn net.Conn) { + defer func() { _ = conn.Close() }() + stop := context.AfterFunc(ctx, func() { _ = conn.Close() }) + defer stop() + for { + if err := conn.SetReadDeadline(time.Now().Add(4 * time.Second)); err != nil { + return + } + payload, err := readFrame(conn) + if err != nil { + return + } + response, err := relayResponse(ctx, cfg, payload, time.Now()) + if err != nil { + log.Printf("relay TCP: %v", err) + return + } + // Start the write deadline after any intentional response delay. + if err := conn.SetWriteDeadline(time.Now().Add(4 * time.Second)); err != nil { + return + } + if err := writeFrame(conn, response); err != nil { + return + } + } +} + +func relayResponse(ctx context.Context, cfg RelayConfig, payload []byte, received time.Time) ([]byte, error) { + if cfg.Mode == modeRoot { + host, _, err := net.SplitHostPort(cfg.Target) + if err != nil { + return nil, err + } + return json.Marshal(Reply{Destination: host, Payload: string(payload)}) + } + response, _, err := exchange(ctx, cfg.Transport, cfg.Target, payload, &net.Dialer{Timeout: 3 * time.Second}, + 3*time.Second) + if err != nil { + return nil, err + } + if err := delayResponse(ctx, cfg, received); err != nil { + return nil, err + } + return response, nil +} + +type delayDirective struct { + Seconds float64 `json:"seconds"` + Token string `json:"token"` +} + +type delayTrace struct { + Token string `json:"token"` + Destination string `json:"destination"` + Transport string `json:"transport"` + RequestReceivedAt float64 `json:"requestReceivedAt"` + UpstreamResponseAt float64 `json:"upstreamResponseAt"` + ReplyAttemptAt float64 `json:"replyAttemptAt,omitempty"` + Seconds float64 `json:"seconds"` +} + +func unixSeconds(t time.Time) float64 { return float64(t.UnixNano()) / float64(time.Second) } + +func delayResponse(ctx context.Context, cfg RelayConfig, received time.Time) error { + claim, err := os.CreateTemp(cfg.StateDir, "psc-delay-claim-*") + if err != nil { + return err + } + name := claim.Name() + defer func() { _ = os.Remove(name) }() + if err := claim.Close(); err != nil { + return err + } + if err := os.Rename(filepath.Join(cfg.StateDir, "psc-delay.json"), name); err != nil { + if errors.Is(err, os.ErrNotExist) { + return nil + } + return err + } + raw, err := os.ReadFile(name) + if err != nil { + return err + } + var directive delayDirective + if err := json.Unmarshal(raw, &directive); err != nil { + return err + } + if directive.Seconds <= 0 || directive.Seconds > 30 { + return errors.New("delay seconds must be in (0,30]") + } + destination, _, err := net.SplitHostPort(cfg.Listen) + if err != nil { + return err + } + trace := delayTrace{Token: directive.Token, Destination: destination, Transport: cfg.Transport, + RequestReceivedAt: unixSeconds(received), UpstreamResponseAt: unixSeconds(time.Now()), Seconds: directive.Seconds} + if err := saveTrace(cfg.StateDir, trace); err != nil { + return err + } + timer := time.NewTimer(time.Duration(directive.Seconds * float64(time.Second))) + defer timer.Stop() + select { + case <-ctx.Done(): + return ctx.Err() + case <-timer.C: + } + trace.ReplyAttemptAt = unixSeconds(time.Now()) + return saveTrace(cfg.StateDir, trace) +} + +func saveTrace(dir string, trace delayTrace) error { + file, err := os.CreateTemp(dir, "psc-delay-trace-*") + if err != nil { + return err + } + defer func() { _ = os.Remove(file.Name()) }() + if err := json.NewEncoder(file).Encode(trace); err != nil { + _ = file.Close() + return err + } + if err := file.Close(); err != nil { + return err + } + return os.Rename(file.Name(), filepath.Join(dir, "psc-delay-trace.json")) +} diff --git a/tests/psc/internal/traffic/socket.go b/tests/psc/internal/traffic/socket.go new file mode 100644 index 00000000..d4da262c --- /dev/null +++ b/tests/psc/internal/traffic/socket.go @@ -0,0 +1,16 @@ +// Copyright 2026 Datum Cloud, Inc. +// SPDX-License-Identifier: AGPL-3.0-or-later + +package traffic + +import "syscall" + +func reuseAddress(_, _ string, raw syscall.RawConn) error { + var socketErr error + if err := raw.Control(func(fd uintptr) { + socketErr = syscall.SetsockoptInt(int(fd), syscall.SOL_SOCKET, syscall.SO_REUSEADDR, 1) + }); err != nil { + return err + } + return socketErr +} diff --git a/tests/psc/internal/traffic/traffic.go b/tests/psc/internal/traffic/traffic.go new file mode 100644 index 00000000..e07564cb --- /dev/null +++ b/tests/psc/internal/traffic/traffic.go @@ -0,0 +1,116 @@ +// Copyright 2026 Datum Cloud, Inc. +// SPDX-License-Identifier: AGPL-3.0-or-later + +// Package traffic supplies the private service fixture's query and relay tools. +package traffic + +import ( + "bytes" + "context" + "encoding/binary" + "encoding/json" + "errors" + "fmt" + "io" + "net" + "time" +) + +const maxFrameSize = 65535 + +// Reply identifies the producer destination and the original request payload. +type Reply struct { + Destination string `json:"destination"` + Payload string `json:"payload"` +} + +func readFrame(r io.Reader) ([]byte, error) { + var header [2]byte + if _, err := io.ReadFull(r, header[:]); err != nil { + return nil, err + } + payload := make([]byte, binary.BigEndian.Uint16(header[:])) + _, err := io.ReadFull(r, payload) + return payload, err +} + +func writeFrame(w io.Writer, payload []byte) error { + if len(payload) > maxFrameSize { + return errors.New("service frame exceeds 65535 bytes") + } + var header [2]byte + binary.BigEndian.PutUint16(header[:], uint16(len(payload))) + _, err := io.Copy(w, io.MultiReader(bytes.NewReader(header[:]), bytes.NewReader(payload))) + return err +} + +func exchange(ctx context.Context, transport, address string, payload []byte, dialer *net.Dialer, + timeout time.Duration, +) ([]byte, string, error) { + conn, err := dialer.DialContext(ctx, transport+"6", address) + if err != nil { + return nil, "", err + } + defer func() { _ = conn.Close() }() + if err := conn.SetDeadline(time.Now().Add(timeout)); err != nil { + return nil, "", err + } + var response []byte + if transport == transportTCP { + if err := writeFrame(conn, payload); err != nil { + return nil, "", err + } + response, err = readFrame(conn) + } else { + if _, err := conn.Write(payload); err != nil { + return nil, "", err + } + response = make([]byte, maxFrameSize) + var n int + n, err = conn.Read(response) + response = response[:n] + } + host, _, splitErr := net.SplitHostPort(conn.RemoteAddr().String()) + if splitErr != nil { + return nil, "", splitErr + } + return response, host, err +} + +// QueryConfig defines one consumer request with a fixed source and frontend. +type QueryConfig struct { + Transport, Frontend, Source, Expected, Payload string + Timeout time.Duration +} + +// Query checks that the selected producer replies through the consumer frontend. +func Query(ctx context.Context, cfg QueryConfig) (Reply, string, error) { + var local net.Addr + var err error + if cfg.Transport == transportTCP { + local, err = net.ResolveTCPAddr("tcp6", cfg.Source) + } else { + local, err = net.ResolveUDPAddr("udp6", cfg.Source) + } + if err != nil { + return Reply{}, "", err + } + dialer := &net.Dialer{LocalAddr: local, Timeout: cfg.Timeout, Control: reuseAddress} + raw, peer, err := exchange(ctx, cfg.Transport, cfg.Frontend, []byte(cfg.Payload), dialer, cfg.Timeout) + if err != nil { + return Reply{}, peer, err + } + var reply Reply + if err := json.Unmarshal(raw, &reply); err != nil { + return reply, peer, err + } + frontend, _, err := net.SplitHostPort(cfg.Frontend) + if err != nil { + return reply, peer, err + } + if !net.ParseIP(peer).Equal(net.ParseIP(frontend)) || + reply.Destination != cfg.Expected || reply.Payload != cfg.Payload { + return reply, peer, fmt.Errorf("unexpected producer reply: %+v from %s", reply, peer) + } + return reply, peer, nil +} diff --git a/tests/psc/internal/traffic/traffic_test.go b/tests/psc/internal/traffic/traffic_test.go new file mode 100644 index 00000000..41d0192a --- /dev/null +++ b/tests/psc/internal/traffic/traffic_test.go @@ -0,0 +1,181 @@ +// Copyright 2026 Datum Cloud, Inc. +// SPDX-License-Identifier: AGPL-3.0-or-later + +package traffic + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "io" + "net" + "os" + "path/filepath" + "strings" + "testing" + "time" +) + +const ( + testDestination = "fd70:100::10" + testListen = "[::1]:0" +) + +type readyOutput struct{ address chan string } + +func (r readyOutput) Write(data []byte) (int, error) { + fields := strings.Fields(string(data)) + r.address <- fields[len(fields)-1] + return len(data), nil +} + +func startService(t *testing.T, cfg RelayConfig) string { + t.Helper() + ctx, cancel := context.WithCancel(context.Background()) + ready := readyOutput{address: make(chan string, 1)} + done := make(chan error, 1) + go func() { done <- Serve(ctx, cfg, ready) }() + t.Cleanup(func() { + cancel() + select { + case err := <-done: + if err != nil { + t.Errorf("stop relay: %v", err) + } + case <-time.After(5 * time.Second): + t.Error("relay did not stop") + } + }) + select { + case address := <-ready.address: + return address + case <-time.After(5 * time.Second): + t.Fatal("relay did not become ready") + return "" + } +} + +func TestProducerRelayAndQuery(t *testing.T) { + for _, transport := range []string{transportUDP, transportTCP} { + t.Run(transport, func(t *testing.T) { + root := startService(t, RelayConfig{Mode: modeRoot, Transport: transport, + Listen: testListen, Target: net.JoinHostPort(testDestination, "8443")}) + stateDir := t.TempDir() + producer := startService(t, RelayConfig{Mode: modeProducer, Transport: transport, + Listen: testListen, Target: root, StateDir: stateDir}) + cfg := QueryConfig{Transport: transport, Frontend: producer, Source: testListen, + Expected: testDestination, Payload: "consumer-a request", Timeout: time.Second} + if _, _, err := Query(context.Background(), cfg); err != nil { + t.Fatal(err) + } + directive := []byte(`{"seconds":0.05,"token":"delayed-request"}`) + if err := os.WriteFile(filepath.Join(stateDir, "psc-delay.json"), directive, 0600); err != nil { + t.Fatal(err) + } + if _, _, err := Query(context.Background(), cfg); err != nil { + t.Fatalf("delayed reply: %v", err) + } + raw, err := os.ReadFile(filepath.Join(stateDir, "psc-delay-trace.json")) + if err != nil { + t.Fatal(err) + } + var trace delayTrace + if err := json.Unmarshal(raw, &trace); err != nil { + t.Fatal(err) + } + if trace.Token != "delayed-request" || trace.Transport != transport || + trace.UpstreamResponseAt < trace.RequestReceivedAt || trace.ReplyAttemptAt < trace.UpstreamResponseAt+0.04 { + t.Fatalf("incorrect reply trace: %+v", trace) + } + if _, err := os.Stat(filepath.Join(stateDir, "psc-delay.json")); !os.IsNotExist(err) { + t.Fatalf("delay directive not consumed: %v", err) + } + if _, _, err := Query(context.Background(), cfg); err != nil { + t.Fatalf("next request: %v", err) + } + cfg.Expected = "fd70:100::11" + if _, _, err := Query(context.Background(), cfg); err == nil { + t.Fatal("accepted the wrong producer identity") + } + }) + } +} + +func TestFramedTCPRequests(t *testing.T) { + address := startService(t, RelayConfig{Mode: modeRoot, Transport: transportTCP, + Listen: testListen, Target: net.JoinHostPort(testDestination, "8443")}) + conn, err := (&net.Dialer{}).DialContext(context.Background(), "tcp6", address) + if err != nil { + t.Fatal(err) + } + defer func() { _ = conn.Close() }() + if err := conn.SetDeadline(time.Now().Add(time.Second)); err != nil { + t.Fatal(err) + } + for _, payload := range []string{"first request", "second request"} { + var frame bytes.Buffer + if err := writeFrame(&frame, []byte(payload)); err != nil { + t.Fatal(err) + } + // Deliver the frame in separate bytes to exercise partial socket reads. + for _, value := range frame.Bytes() { + if _, err := conn.Write([]byte{value}); err != nil { + t.Fatal(err) + } + } + raw, err := readFrame(conn) + if err != nil { + t.Fatal(err) + } + var reply Reply + if err := json.Unmarshal(raw, &reply); err != nil { + t.Fatal(err) + } + if reply.Payload != payload || reply.Destination != testDestination { + t.Fatalf("unexpected reply: %+v", reply) + } + } + if _, err := readFrame(bytes.NewReader([]byte{0, 2, 1})); !errors.Is(err, io.ErrUnexpectedEOF) { + t.Fatalf("truncated frame: %v", err) + } + if err := writeFrame(io.Discard, make([]byte, maxFrameSize+1)); err == nil { + t.Fatal("accepted oversized frame") + } +} + +func TestDelayedTCPReply(t *testing.T) { + root := startService(t, RelayConfig{Mode: modeRoot, Transport: transportTCP, + Listen: testListen, Target: net.JoinHostPort(testDestination, "8443")}) + stateDir := t.TempDir() + producer := startService(t, RelayConfig{Mode: modeProducer, Transport: transportTCP, + Listen: testListen, Target: root, StateDir: stateDir}) + directive := []byte(`{"seconds":4.1,"token":"past-read-deadline"}`) + if err := os.WriteFile(filepath.Join(stateDir, "psc-delay.json"), directive, 0600); err != nil { + t.Fatal(err) + } + cfg := QueryConfig{Transport: transportTCP, Frontend: producer, Source: testListen, + Expected: testDestination, Payload: "delayed TCP request", Timeout: 6 * time.Second} + if _, _, err := Query(context.Background(), cfg); err != nil { + t.Fatalf("intentional delay must not expire the write deadline: %v", err) + } +} + +func TestQueryCommandExitStatus(t *testing.T) { + listener, err := (&net.ListenConfig{}).ListenPacket(context.Background(), "udp6", testListen) + if err != nil { + t.Fatal(err) + } + defer func() { _ = listener.Close() }() + args := []string{commandQuery, "--transport", transportUDP, "--frontend", listener.LocalAddr().String(), + "--source", testListen, "--expected", testDestination, "--payload", "request", "--timeout", "20ms"} + var output bytes.Buffer + code, err := Run(context.Background(), args, &output) + if code != 3 || err == nil || !strings.Contains(output.String(), `"timeout":true`) { + t.Fatalf("blocked query: status=%d err=%v output=%s", code, err, output.String()) + } + code, err = Run(context.Background(), []string{commandQuery, "--transport", "invalid"}, io.Discard) + if code != 2 || err == nil { + t.Fatalf("invalid command: status=%d err=%v", code, err) + } +}