test(edge): Prove MASQUE packet access in the lab

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:

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.

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
+}

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})
 }

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)
 	}

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