diff --git a/docs/designs/masque/acceptance.md b/docs/designs/masque/acceptance.md index 692a96e4..7a3a9c30 100644 --- a/docs/designs/masque/acceptance.md +++ b/docs/designs/masque/acceptance.md @@ -10,7 +10,7 @@ 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, two connectors sharing one attachment while one request ends, router garbage collection while connectors forward, a provider process killed and restarted in place, 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 runs the [MASQUE provider adapter](../../edge/masque-provider.md) 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. +`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, M07, and the #989 findings A06, A14 and A16, plus NAT egress from the attachment node, two connectors sharing one attachment while one request ends, router garbage collection while connectors forward, a provider process killed and restarted in place, pod recreation, a partly set-up instance killed without shutdown, and asserted teardown. A14's namespace-only impostor is a real-kernel test of `internal/edge/hostattach`. The provider is the test-only fixture in `tests/edgeconformance/masqueprovider`. It runs the [MASQUE provider adapter](../../edge/masque-provider.md) 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. [#985](https://github.com/datum-cloud/galactic/issues/985) owns running these checks on the target platform. ## Epic and implementation checks diff --git a/docs/edge/provider-contract.md b/docs/edge/provider-contract.md index 5c376257..2ce39759 100644 --- a/docs/edge/provider-contract.md +++ b/docs/edge/provider-contract.md @@ -180,6 +180,8 @@ stateDiagram-v2 A failed step leaves the attachment in its last state with a condition naming the reason. Partial host state carries the attachment ID in the host link's alias, so the node finds and finishes it on restart (`internal/edge/hostattach/inventory.go:33`). +The node provisions every attachment again at least once a minute (`agent.Resync`), which repairs host state that drifted under it. It removes every IPv6 unicast route through the host link, the pod link or the packet device in the table it owns, other than the ones it installs, matching destination, gateway and metric. The only kernel routes it leaves are the connected routes of the addresses it assigns itself, and it removes any other global address from the pod link and the packet device, so an address added in the pod brings no route of its own. A route or address that appears after that pass, or that the node fails to remove, fails `Readback`, and the attachment leaves `Programmed`. IPv4 routes, other tables and policy rules are outside this repair. + ### Forwarding claim Each `EdgeRoute` is the claim for one prefix in one gVPC. An instance renews one lease, not one per route, so the claim has two levels: diff --git a/internal/edge/hostattach/closed_gate_test.go b/internal/edge/hostattach/closed_gate_test.go index eb6b77c0..4d236140 100644 --- a/internal/edge/hostattach/closed_gate_test.go +++ b/internal/edge/hostattach/closed_gate_test.go @@ -99,7 +99,7 @@ func TestClosedGateNeverReachesShard(t *testing.T) { return err } defer func() { _ = h.Close() }() - return pruneRoutes(h, link, table, []netip.Prefix{own}) + return pruneRoutes(h, link, table, []installed{{dst: own}}, connected()) }) echoes := f.tunA.echoes.Load() for range 3 { diff --git a/internal/edge/hostattach/hostattach.go b/internal/edge/hostattach/hostattach.go index d3fe244f..c2b29e83 100644 --- a/internal/edge/hostattach/hostattach.go +++ b/internal/edge/hostattach/hostattach.go @@ -518,8 +518,8 @@ func configureHostLink( LinkIndex: idx, Table: int(table), Dst: prefixNet(p), Gw: a.Address.AsSlice(), }) } - if err := pruneRoutes(h, link.Attrs().Index, table, hostPrefixes(a)); err != nil { - return err + if err := pruneRoutes(h, link.Attrs().Index, table, hostRoutes(a), connected()); err != nil { + return fmt.Errorf("hostattach: host link %q: %w", a.HostLink, err) } for _, r := range routes { if err := checkRouteOwner(h, r); err != nil { @@ -578,6 +578,16 @@ func configurePodLink(h *netlink.Handle, link netlink.Link, a Attachment, hostMA }); err != nil { return fmt.Errorf("hostattach: neighbor %s on %q: %w", hostLinkLocal, a.PodLink, err) } + // The pod link carries only the node's addresses and the default route + // toward the host. Any other address brings a connected route, and any + // other route sends pod VRF traffic past the uSID datapath's view of the + // attachment. + if err := pruneAddrs(h, link, []netip.Addr{a.Address}); err != nil { + return fmt.Errorf("hostattach: pod link %q: %w", a.PodLink, err) + } + if err := pruneRoutes(h, idx, a.PodTable, podRoutes, connected(a.Address)); err != nil { + return fmt.Errorf("hostattach: pod link %q: %w", a.PodLink, err) + } def := &netlink.Route{ LinkIndex: idx, Table: int(a.PodTable), Dst: &net.IPNet{IP: net.IPv6zero, Mask: net.CIDRMask(0, 128)}, Gw: hostLinkLocal.AsSlice(), @@ -629,8 +639,11 @@ func ensureTUN(h *netlink.Handle, pod *Namespace, a Attachment) error { if err := h.LinkSetUp(tun); err != nil { return fmt.Errorf("hostattach: set TUN %q up: %w", a.TUN.Name, err) } - if err := pruneRoutes(h, tun.Attrs().Index, a.PodTable, a.Prefixes); err != nil { - return err + if err := pruneAddrs(h, tun, nil); err != nil { + return fmt.Errorf("hostattach: TUN %q: %w", a.TUN.Name, err) + } + if err := pruneRoutes(h, tun.Attrs().Index, a.PodTable, tunRoutes(a), connected()); err != nil { + return fmt.Errorf("hostattach: TUN %q: %w", a.TUN.Name, err) } for _, p := range a.Prefixes { r := &netlink.Route{LinkIndex: tun.Attrs().Index, Table: int(a.PodTable), Dst: prefixNet(p)} diff --git a/internal/edge/hostattach/inventory.go b/internal/edge/hostattach/inventory.go index c5f6ae35..4656fec4 100644 --- a/internal/edge/hostattach/inventory.go +++ b/internal/edge/hostattach/inventory.go @@ -7,6 +7,7 @@ package hostattach import ( "fmt" "net/netip" + "slices" "strings" "github.com/vishvananda/netlink" @@ -87,16 +88,82 @@ func listOwned(h *netlink.Handle) ([]ObservedAttachment, error) { return result, nil } -func hostPrefixes(a Attachment) []netip.Prefix { - prefixes := make([]netip.Prefix, 0, len(a.Prefixes)+1) - prefixes = append(prefixes, netip.PrefixFrom(a.Address, 128)) - return append(prefixes, a.Prefixes...) +// podDefault is the destination of the only route an attachment's pod link +// carries in the pod VRF. +var podDefault = netip.PrefixFrom(netip.IPv6Unspecified(), 0) + +// installed is a route the node installs through an attachment's link: its +// destination and, for a route with one, its gateway. +type installed struct { + dst netip.Prefix + gw netip.Addr } -// extraRoutes lists static unicast destinations on a dedicated owned link that -// no longer appear in desired state. Kernel-derived local/link-local routes -// and routes using other links are never candidates for deletion. -func extraRoutes(h *netlink.Handle, index int, table uint32, want []netip.Prefix) ([]netlink.Route, error) { +// hostRoutes are the routes the node installs through the host link: the +// attachment's address, and each prefix via that address. +func hostRoutes(a Attachment) []installed { + routes := make([]installed, 0, len(a.Prefixes)+1) + routes = append(routes, installed{dst: netip.PrefixFrom(a.Address, 128)}) + for _, p := range a.Prefixes { + routes = append(routes, installed{dst: p, gw: a.Address}) + } + return routes +} + +// podRoutes is the route the node installs through the pod link: the pod +// VRF's default, via the host link's link-local address. +var podRoutes = []installed{{dst: podDefault, gw: hostLinkLocal}} + +// tunRoutes are the routes the node installs through the packet device: one +// per prefix, without a gateway. +func tunRoutes(a Attachment) []installed { + routes := make([]installed, 0, len(a.Prefixes)) + for _, p := range a.Prefixes { + routes = append(routes, installed{dst: p}) + } + return routes +} + +// matches reports whether r is the installed route: the same destination +// and gateway, at the default metric the node's writes leave. +func (want installed) matches(r netlink.Route) bool { + if !sameDst(r.Dst, want.dst) || (r.Priority != 0 && r.Priority != defaultRouteMetric) { + return false + } + gw, _ := netip.AddrFromSlice(r.Gw) + return gw.Unmap() == want.gw +} + +// defaultRouteMetric is the metric the kernel gives an IPv6 route added +// without one. +const defaultRouteMetric = 1024 + +// linkLocalPrefix is the kernel's connected route for any link-local +// address, which every link carries. +var linkLocalPrefix = netip.MustParsePrefix("fe80::/64") + +// connected lists the kernel's connected routes the node expects through +// one of an attachment's links: the link-local prefix, and the prefixes of +// the addresses the node itself assigns to the link. +func connected(own ...netip.Addr) []netip.Prefix { + prefixes := make([]netip.Prefix, 0, len(own)+1) + prefixes = append(prefixes, linkLocalPrefix) + for _, a := range own { + prefixes = append(prefixes, netip.PrefixFrom(a, a.BitLen())) + } + return prefixes +} + +// extraRoutes lists the unicast routes through a dedicated owned link that +// the node does not install: any destination, gateway or metric other than +// want's. A kernel route counts as the kernel's own only when it is one of +// the connected routes the node expects (see connected), never because the +// link happens to carry an address now, so neither a route written with +// "proto kernel" nor the connected route of an address someone else added +// is exempt. Local and multicast routes, and routes through other links, are +// never candidates. +func extraRoutes(h *netlink.Handle, index int, table uint32, want []installed, kernel []netip.Prefix, +) ([]netlink.Route, error) { routes, err := nldump.On(h).RouteListFiltered(unix.AF_INET6, &netlink.Route{Table: int(table), LinkIndex: index}, netlink.RT_FILTER_TABLE|netlink.RT_FILTER_OIF) if err != nil { @@ -104,25 +171,55 @@ func extraRoutes(h *netlink.Handle, index int, table uint32, want []netip.Prefix } var extras []netlink.Route for _, r := range routes { - if r.Type != unix.RTN_UNICAST || r.Protocol == unix.RTPROT_KERNEL { + if r.Type != unix.RTN_UNICAST { continue } - found := false - for _, p := range want { - if sameDst(r.Dst, p) { - found = true - break - } + if r.Protocol == unix.RTPROT_KERNEL && slices.ContainsFunc(kernel, func(p netip.Prefix) bool { + return sameDst(r.Dst, p) + }) { + continue } - if !found { + if !slices.ContainsFunc(want, func(w installed) bool { return w.matches(r) }) { extras = append(extras, r) } } return extras, nil } -func pruneRoutes(h *netlink.Handle, index int, table uint32, want []netip.Prefix) error { - extras, err := extraRoutes(h, index, table, want) +// extraAddrs lists the IPv6 addresses on link other than link-local ones and +// own, the addresses the node assigns to it. +func extraAddrs(h *netlink.Handle, link netlink.Link, own []netip.Addr) ([]netlink.Addr, error) { + addrs, err := nldump.On(h).AddrList(link, unix.AF_INET6) + if err != nil { + return nil, fmt.Errorf("hostattach: list addresses on %q: %w", link.Attrs().Name, err) + } + var extras []netlink.Addr + for _, a := range addrs { + addr, ok := netip.AddrFromSlice(a.IP) + if !ok || addr.Unmap().IsLinkLocalUnicast() || slices.Contains(own, addr.Unmap()) { + continue + } + extras = append(extras, a) + } + return extras, nil +} + +// pruneAddrs removes the addresses extraAddrs finds on link. +func pruneAddrs(h *netlink.Handle, link netlink.Link, own []netip.Addr) error { + extras, err := extraAddrs(h, link, own) + if err != nil { + return err + } + for _, a := range extras { + if err := h.AddrDel(link, &a); err != nil { + return fmt.Errorf("hostattach: remove stray address %s from %q: %w", a.IPNet, link.Attrs().Name, err) + } + } + return nil +} + +func pruneRoutes(h *netlink.Handle, index int, table uint32, want []installed, kernel []netip.Prefix) error { + extras, err := extraRoutes(h, index, table, want, kernel) if err != nil { return err } diff --git a/internal/edge/hostattach/readback.go b/internal/edge/hostattach/readback.go index 441ba061..d4d3265a 100644 --- a/internal/edge/hostattach/readback.go +++ b/internal/edge/hostattach/readback.go @@ -76,12 +76,14 @@ func Readback(host, pod *Namespace, a Attachment) error { checkNeigh(podH, podLink, hostLinkLocal, hostLink.Attrs().HardwareAddr, add) hostIdx := hostLink.Attrs().Index - checkExtraRoutes(hostH, hostIdx, hostTable, hostPrefixes(a), add) + checkExtraRoutes(hostH, hostIdx, hostTable, hostRoutes(a), connected(), add) checkRoute(hostH, hostTable, netip.PrefixFrom(a.Address, 128), hostIdx, netip.Addr{}, add) for _, p := range a.Prefixes { checkRoute(hostH, hostTable, p, hostIdx, a.Address, add) } - checkRoute(podH, a.PodTable, netip.PrefixFrom(netip.IPv6Unspecified(), 0), podLink.Attrs().Index, hostLinkLocal, add) + checkExtraAddrs(podH, podLink, []netip.Addr{a.Address}, add) + checkExtraRoutes(podH, podLink.Attrs().Index, a.PodTable, podRoutes, connected(a.Address), add) + checkRoute(podH, a.PodTable, podDefault, podLink.Attrs().Index, hostLinkLocal, add) checkUnreachableDefault(podH, a.PodTable, add) checkLocalRoute(podH, a.PodTable, a.Address, add) @@ -91,7 +93,8 @@ func Readback(host, pod *Namespace, a Attachment) error { if want := a.TUN.effectiveMTU(a.MTU); tun.Attrs().MTU != want { add("TUN %q MTU %d, want %d", a.TUN.Name, tun.Attrs().MTU, want) } - checkExtraRoutes(podH, tun.Attrs().Index, a.PodTable, a.Prefixes, add) + checkExtraAddrs(podH, tun, nil, add) + checkExtraRoutes(podH, tun.Attrs().Index, a.PodTable, tunRoutes(a), connected(), add) if tun.Attrs().Alias != ownerAlias(a.ID) { add("TUN ownership does not match attachment") } @@ -244,13 +247,26 @@ func sameDst(got *net.IPNet, want netip.Prefix) bool { return netip.PrefixFrom(addr.Unmap(), ones) == want } -func checkExtraRoutes(h *netlink.Handle, index int, table uint32, want []netip.Prefix, add func(string, ...any)) { - extras, err := extraRoutes(h, index, table, want) +func checkExtraAddrs(h *netlink.Handle, link netlink.Link, own []netip.Addr, add func(string, ...any)) { + extras, err := extraAddrs(h, link, own) + if err != nil { + add("%v", err) + return + } + for _, a := range extras { + add("unexpected address %s on %q", a.IPNet, link.Attrs().Name) + } +} + +func checkExtraRoutes(h *netlink.Handle, index int, table uint32, want []installed, kernel []netip.Prefix, + add func(string, ...any)) { + extras, err := extraRoutes(h, index, table, want, kernel) if err != nil { add("%v", err) return } for _, r := range extras { - add("unexpected route %s in table %d through ifindex %d", r.Dst, table, index) + add("unexpected route %s via %s metric %d in table %d through ifindex %d", r.Dst, r.Gw, r.Priority, table, + index) } } diff --git a/internal/edge/hostattach/recovery_test.go b/internal/edge/hostattach/recovery_test.go index f3c4cef5..abc59614 100644 --- a/internal/edge/hostattach/recovery_test.go +++ b/internal/edge/hostattach/recovery_test.go @@ -10,6 +10,7 @@ import ( "net/netip" "os" "runtime" + "strings" "testing" "github.com/vishvananda/netlink" @@ -73,6 +74,167 @@ func TestPeerNamespaceOwnership(t *testing.T) { } } +// TestPeerNamespaceOnlyImpostor offers the host link's peer from another +// namespace: a pod link with the same name and index as the real peer, whose +// own peer has the host link's ifindex in a third namespace. Every index +// matches, so only the peer namespace tells it apart. +func TestPeerNamespaceOnlyImpostor(t *testing.T) { + requireRoot(t) + host := newHost(t, vpcA) + hostH := host.open(t) + victim, impostor, decoy := newTestNS(t), newTestNS(t), newTestNS(t) + victimH, impostorH := victim.open(t), impostor.open(t) + a := recoveryAttachment() + const hostIndex = 4000 + for _, pair := range []struct { + from testNS + name string + peer testNS + }{{host, a.HostLink, victim}, {decoy, "decoy-h", impostor}} { + pair.from.do(t, func() error { + return netlink.LinkAdd(&netlink.Veth{ + LinkAttrs: netlink.LinkAttrs{Name: pair.name, Index: hostIndex}, + PeerName: a.PodLink, PeerNamespace: netlink.NsFd(int(pair.peer.Fd())), + }) + }) + } + podLink := func(n testNS) netlink.LinkAttrs { + var attrs netlink.LinkAttrs + n.do(t, func() error { + l, err := netlink.LinkByName(a.PodLink) + if err == nil { + attrs = *l.Attrs() + } + return err + }) + return attrs + } + realLink, fake := podLink(victim), podLink(impostor) + if realLink.Index != fake.Index || realLink.ParentIndex != hostIndex || fake.ParentIndex != hostIndex { + t.Fatalf("precondition: real pod link %d->%d, impostor %d->%d, want equal indexes peering with %d", + realLink.Index, realLink.ParentIndex, fake.Index, fake.ParentIndex, hostIndex) + } + a.PodIndex = realLink.Index + + if _, err := Provision(hostH, impostorH, a); !errors.Is(err, ErrNotOwned) { + t.Fatalf("adoption through the impostor = %v", err) + } + if err := Release(hostH, impostorH, a); !errors.Is(err, ErrNotOwned) { + t.Fatalf("release through the impostor = %v", err) + } + host.do(t, func() error { + link, err := netlink.LinkByName(a.HostLink) + if err != nil { + return err + } + if link.Attrs().Alias != "" || link.Attrs().MasterIndex != 0 { + t.Error("the impostor's offer changed the host link") + } + return nil + }) + if got := podLink(impostor); got.MasterIndex != 0 || got.Alias != "" || got.Flags&net.FlagUp != 0 { + t.Errorf("the impostor's pod link was changed: master %d, alias %q, flags %v", got.MasterIndex, got.Alias, + got.Flags) + } + if _, err := Provision(hostH, victimH, a); err != nil { + t.Fatalf("adoption through the real peer = %v", err) + } +} + +// TestPodLinkRouteDrift adds routes through the pod link to the pod VRF +// beside the default route toward the host: a stray prefix, a second default +// at a better metric, and a stray prefix written as a kernel route. Readback +// reports each, Provision removes it, and the node's own default stays. +func TestPodLinkRouteDrift(t *testing.T) { + requireRoot(t) + host := newHost(t, vpcA) + hostH := host.open(t) + pod := newTestNS(t) + podH := pod.open(t) + a := recoveryAttachment() + if _, err := Provision(hostH, podH, a); err != nil { + t.Fatal(err) + } + for _, tc := range []struct { + name string + route func(index int) *netlink.Route + }{ + {"stray prefix", func(index int) *netlink.Route { + return &netlink.Route{LinkIndex: index, Table: int(a.PodTable), Dst: ip6Net("fd00:dead::/64")} + }}, + {"second default at a better metric", func(index int) *netlink.Route { + return &netlink.Route{LinkIndex: index, Table: int(a.PodTable), Dst: ip6Net("::/0"), Priority: 1} + }}, + {"stray prefix written as a kernel route", func(index int) *netlink.Route { + return &netlink.Route{LinkIndex: index, Table: int(a.PodTable), Dst: ip6Net("fd00:dead::/64"), + Protocol: unix.RTPROT_KERNEL} + }}, + } { + t.Run(tc.name, func(t *testing.T) { + pod.do(t, func() error { + link, err := netlink.LinkByName(a.PodLink) + if err != nil { + return err + } + return netlink.RouteAdd(tc.route(link.Attrs().Index)) + }) + if err := Readback(hostH, podH, a); !errors.Is(err, ErrDrift) || + !strings.Contains(err.Error(), "unexpected route") { + t.Fatalf("readback = %v, want drift reporting the route", err) + } + if _, err := Provision(hostH, podH, a); err != nil { + t.Fatal(err) + } + if err := Readback(hostH, podH, a); err != nil { + t.Fatalf("readback after repair = %v", err) + } + }) + } +} + +// TestAddressDrift adds a global address to the pod link and to the packet +// device. The kernel's connected route for each would steer pod VRF traffic +// past the attachment's routes, and is "proto kernel", so the node must not +// treat it as its own: Readback reports the addresses, and Provision removes +// them and their routes. +func TestAddressDrift(t *testing.T) { + requireRoot(t) + host := newHost(t, vpcA) + hostH := host.open(t) + pod := newTestNS(t) + podH := pod.open(t) + a := recoveryAttachment() + a.TUN = &TUN{Name: "rtun", Owner: uint32(os.Getuid()), Group: uint32(os.Getgid())} + a.Prefixes = []netip.Prefix{netip.MustParsePrefix("fd00:5::/96")} + if _, err := Provision(hostH, podH, a); err != nil { + t.Fatal(err) + } + for _, tc := range []struct{ link, addr string }{ + {a.PodLink, "fd00:dead::1/64"}, + {a.TUN.Name, "fd00:5::1:0/112"}, + } { + t.Run(tc.link, func(t *testing.T) { + pod.do(t, func() error { + link, err := netlink.LinkByName(tc.link) + if err != nil { + return err + } + return netlink.AddrAdd(link, &netlink.Addr{IPNet: ip6Net(tc.addr), Flags: unix.IFA_F_NODAD}) + }) + if err := Readback(hostH, podH, a); !errors.Is(err, ErrDrift) || + !strings.Contains(err.Error(), "unexpected address") { + t.Fatalf("readback = %v, want drift reporting the address", err) + } + if _, err := Provision(hostH, podH, a); err != nil { + t.Fatal(err) + } + if err := Readback(hostH, podH, a); err != nil { + t.Fatalf("readback after repair = %v", err) + } + }) + } +} + func TestPrefixRemovalAndPartialRecovery(t *testing.T) { requireRoot(t) host := newHost(t, vpcA) diff --git a/tests/edgeconformance/edgeprobe/egresskind.go b/tests/edgeconformance/edgeprobe/egresskind.go new file mode 100644 index 00000000..5e254f3d --- /dev/null +++ b/tests/edgeconformance/edgeprobe/egresskind.go @@ -0,0 +1,45 @@ +// Copyright 2026 Datum Cloud, Inc. +// +// SPDX-License-Identifier: AGPL-3.0-or-later + +package main + +import ( + "errors" + "flag" + "fmt" + "net" + "path/filepath" + + "github.com/cilium/ebpf" +) + +// egressKind prints how the uSID datapath delivers to a host link: the row +// of ifindex_egress_kind_table for the link's ifindex, as +// {"t","ifindex","kind","present"}. kind 0 is a veth, delivered with +// bpf_redirect_peer into its peer's namespace, and 1 is a tap, delivered with +// bpf_redirect. present is false when the link has no row. +func egressKind(args []string) error { + fs := flag.NewFlagSet("egresskind", flag.ExitOnError) + iface := fs.String("iface", "", "host link") + pinDir := fs.String("pin-dir", "/sys/fs/bpf/galactic", "the uSID datapath's pin directory") + _ = fs.Parse(args) + ifi, err := net.InterfaceByName(*iface) + if err != nil { + return err + } + m, err := ebpf.LoadPinnedMap(filepath.Join(*pinDir, "ifindex_egress_kind_table"), + &ebpf.LoadPinOptions{ReadOnly: true}) + if err != nil { + return fmt.Errorf("open ifindex_egress_kind_table: %w", err) + } + defer func() { _ = m.Close() }() + var kind uint32 + present := true + if err := m.Lookup(uint32(ifi.Index), &kind); errors.Is(err, ebpf.ErrKeyNotExist) { //nolint:gosec // ifindex + present = false + } else if err != nil { + return err + } + return out.Encode(map[string]any{"t": boottime(), "ifindex": ifi.Index, "kind": kind, "present": present}) +} diff --git a/tests/edgeconformance/edgeprobe/main.go b/tests/edgeconformance/edgeprobe/main.go index 493824e0..acea89f0 100644 --- a/tests/edgeconformance/edgeprobe/main.go +++ b/tests/edgeconformance/edgeprobe/main.go @@ -18,6 +18,8 @@ // 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) +// edgeprobe egresskind -iface link +// prints the uSID datapath's delivery kind for a host link (see egressKind) package main import ( @@ -38,7 +40,7 @@ import ( func main() { if len(os.Args) < 2 { - fmt.Fprintln(os.Stderr, "usage: edgeprobe send|sniff|connectip [flags]") + fmt.Fprintln(os.Stderr, "usage: edgeprobe send|sniff|connectip|egresskind [flags]") os.Exit(2) } var err error @@ -49,6 +51,8 @@ func main() { err = sniff(os.Args[2:]) case "connectip": err = connectIP(os.Args[2:]) + case "egresskind": + err = egressKind(os.Args[2:]) default: err = fmt.Errorf("unknown command %q", os.Args[1]) } diff --git a/tests/edgeconformance/fixtures_test.go b/tests/edgeconformance/fixtures_test.go index 9aa2744c..d70b9add 100644 --- a/tests/edgeconformance/fixtures_test.go +++ b/tests/edgeconformance/fixtures_test.go @@ -41,6 +41,9 @@ const ( outside = "fd00:ed9e:200:1::/64" capSocket = "SocketAccess" + capPacket = "PacketAccess" + // providerSA is the ServiceAccount every provider deployment runs as. + providerSA = "edge-provider" // fixtureTag fills fields of objects the suite writes as a pod that // runs no provider. fixtureTag = "conformance" diff --git a/tests/edgeconformance/harness_test.go b/tests/edgeconformance/harness_test.go index 52c21cf3..2fb87925 100644 --- a/tests/edgeconformance/harness_test.go +++ b/tests/edgeconformance/harness_test.go @@ -38,6 +38,7 @@ import ( authv1 "k8s.io/api/authentication/v1" coordinationv1 "k8s.io/api/coordination/v1" corev1 "k8s.io/api/core/v1" + rbacv1 "k8s.io/api/rbac/v1" apierrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" @@ -98,7 +99,8 @@ func newLab(kubeconfig string) (*lab, error) { return nil, err } s := runtime.NewScheme() - for _, add := range []func(*runtime.Scheme) error{corev1.AddToScheme, appsv1.AddToScheme, + for _, add := range []func(*runtime.Scheme) error{corev1.AddToScheme, appsv1.AddToScheme, authv1.AddToScheme, + rbacv1.AddToScheme, coordinationv1.AddToScheme, galacticv1alpha1.AddToScheme, cloudv1alpha1.AddToScheme, networkv1alpha1.AddToScheme} { if err := add(s); err != nil { @@ -259,10 +261,13 @@ type probeRun struct { events []event } +// nsenter enters a pod's namespaces from its node. +const nsenter = "nsenter" + // startProbe runs edgeprobe with args in the network namespace of the pod // whose sandbox has pid on node. func startProbe(node string, pid int, name string, args ...string) (*probeRun, error) { - full := append([]string{"exec", node, "nsenter", "-t", strconv.Itoa(pid), "-n", "/usr/local/bin/edgeprobe"}, + full := append([]string{"exec", node, nsenter, "-t", strconv.Itoa(pid), "-n", "/usr/local/bin/edgeprobe"}, args...) p := &probeRun{cmd: exec.CommandContext(context.Background(), "docker", full...), name: name} p.cmd.Stdout = &p.out diff --git a/tests/edgeconformance/masque_foundation_test.go b/tests/edgeconformance/masque_foundation_test.go new file mode 100644 index 00000000..ce033822 --- /dev/null +++ b/tests/edgeconformance/masque_foundation_test.go @@ -0,0 +1,301 @@ +// Copyright 2026 Datum Cloud, Inc. +// +// SPDX-License-Identifier: AGPL-3.0-or-later + +package edgeconformance_test + +import ( + "context" + "crypto/rand" + "encoding/json" + "fmt" + "net/netip" + "slices" + "strconv" + "strings" + "testing" + "time" + + authv1 "k8s.io/api/authentication/v1" + corev1 "k8s.io/api/core/v1" + rbacv1 "k8s.io/api/rbac/v1" + "k8s.io/apimachinery/pkg/api/meta" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" + "sigs.k8s.io/controller-runtime/pkg/client" + + networkv1alpha1 "go.datum.net/galactic/api/network/v1alpha1" + galacticv1alpha1 "go.datum.net/galactic/api/v1alpha1" + "go.datum.net/galactic/internal/edge/admission" +) + +// packetRequest is a PacketAccess EdgeAttachment pod writes for svc. +func packetRequest(pod *corev1.Pod, svc *galacticv1alpha1.EdgeService, contract, capVersion string, +) *galacticv1alpha1.EdgeAttachment { + att := attachmentRequest(pod, masqueProvider, svc, contract, capVersion) + att.Spec.Capabilities = []galacticv1alpha1.EdgeCapability{{Name: capPacket, Version: capVersion}} + att.Spec.PacketDevice = &galacticv1alpha1.EdgePacketDevice{Owner: 65532, Group: 65532} + return att +} + +// rowRefusedRequests checks what the fixture's ServiceAccount cannot do. A +// pod under it may not write for another pod, is refused a service that +// does not exist and versions core or the provider does not implement, may +// not ask for packet access without a device owner, and is denied +// EdgeService, BGPAdvertisement, route index and status writes. Its token +// carries the pod-bound extras the admission policies read, and its only +// binding is the one core keeps for the approved provider. +func (m *masque) rowRefusedRequests(t *testing.T) { + m.inUse(t) + l, ctx := m.l, context.Background() + tp := l.tokenPod(t, masqueNS, "token-masque", edgeNode1) + defer l.deleteAndWait(t, tp) + c := l.asPod(t, tp) + instanceFor(t, c, tp, masqueProvider) + + forged := packetRequest(m.pod, m.a.svc, "v1alpha1", "v1") + forged.Name += "x" + requireRefused(t, "attachment naming another pod", c.Create(ctx, forged)) + + missing := packetRequest(tp, m.a.svc, "v1alpha1", "v1") + missing.Spec.PacketDevice = nil + requireRefused(t, "PacketAccess without a device owner", c.Create(ctx, missing)) + + unknown := m.a.svc.DeepCopy() + unknown.UID = types.UID(strings.ToLower(rand.Text())) + for _, tc := range []struct { + name string + att *galacticv1alpha1.EdgeAttachment + want string + }{ + {"service that does not exist", packetRequest(tp, unknown, "v1alpha1", "v1"), ""}, + {"contract version the provider does not declare", packetRequest(tp, m.a.svc, "v1alpha9", "v1"), + string(admission.UnsupportedContractVersion)}, + {"capability version the provider does not declare", packetRequest(tp, m.a.svc, "v1alpha1", "v2"), + string(admission.UnsupportedCapability)}, + } { + if err := c.Create(ctx, tc.att); err != nil { + t.Fatalf("%s: %v", tc.name, err) + } + reason := l.waitRefused(t, tc.att) + l.record(t, tc.name, reason) + if reason == "admitted" || (tc.want != "" && reason != tc.want) { + t.Errorf("%s: %s, want refused%s", tc.name, reason, map[bool]string{true: " as " + tc.want}[tc.want != ""]) + } + if l.allocated(t, m.a.svc, tc.att.Name) { + t.Errorf("%s: the refused attachment holds an address", tc.name) + } + l.deleteAndWait(t, tc.att) + } + + // The fixture's own pod, under its own token. + own := l.asPod(t, m.pod) + svc := &galacticv1alpha1.EdgeService{ObjectMeta: metav1.ObjectMeta{Namespace: m.a.tn.vpcNamespace, Name: "forged"}, + Spec: *m.a.svc.Spec.DeepCopy()} + requireRefused(t, "EdgeService write", own.Create(ctx, svc)) + adv := &networkv1alpha1.BGPAdvertisement{ObjectMeta: metav1.ObjectMeta{Namespace: systemNS, Name: "forged-masque"}} + requireRefused(t, "BGPAdvertisement write", own.Create(ctx, adv)) + var idx galacticv1alpha1.EdgeRouteIndex + if err := l.c.Get(ctx, client.ObjectKey{Namespace: m.a.tn.vpcNamespace, Name: m.a.vpcUID}, &idx); err != nil { + t.Fatalf("VPC 10's route index: %v", err) + } + idx.Spec.Rebuilt = !idx.Spec.Rebuilt + requireRefused(t, "route index write", own.Update(ctx, &idx)) + r := l.route(t, masqueNS, m.a.vpcUID, netip.MustParsePrefix(masquePrefix)) + r.Status.ClosedEpoch = r.Spec.Claim.Epoch + 1 + requireRefused(t, "route status write", own.Status().Update(ctx, r)) + att := l.attachmentOf(t, m.pod, m.a.svc) + att.Status.HostLink = "forged" + requireRefused(t, "attachment host link write", own.Status().Update(ctx, att)) + + review := &authv1.SelfSubjectReview{} + if err := own.Create(ctx, review); err != nil { + t.Fatalf("self subject review: %v", err) + } + extra := review.Status.UserInfo.Extra + l.record(t, "fixture token user", review.Status.UserInfo.Username) + l.record(t, "fixture token pod extras", fmt.Sprintf("pod-uid=%v node-name=%v", + extra["authentication.kubernetes.io/pod-uid"], extra["authentication.kubernetes.io/node-name"])) + if !slices.Equal(extra["authentication.kubernetes.io/pod-uid"], []string{string(m.pod.UID)}) || + !slices.Equal(extra["authentication.kubernetes.io/node-name"], []string{m.pod.Spec.NodeName}) { + t.Errorf("the fixture's token does not carry its pod UID and node: %v", extra) + } + + var bindings []string + var rbs rbacv1.RoleBindingList + if err := l.c.List(ctx, &rbs); err != nil { + t.Fatal(err) + } + for _, rb := range rbs.Items { + if subjectIsFixture(rb.Subjects) { + bindings = append(bindings, rb.Namespace+"/"+rb.Name+" -> "+rb.RoleRef.Kind+"/"+rb.RoleRef.Name) + } + } + var crbs rbacv1.ClusterRoleBindingList + if err := l.c.List(ctx, &crbs); err != nil { + t.Fatal(err) + } + for _, crb := range crbs.Items { + if subjectIsFixture(crb.Subjects) { + t.Errorf("ClusterRoleBinding %s binds the fixture's ServiceAccount cluster-wide", crb.Name) + } + } + l.record(t, "fixture RoleBindings", strings.Join(bindings, "; ")) + if len(bindings) != 1 || !strings.HasSuffix(bindings[0], " -> ClusterRole/galactic-edge-provider") { + t.Errorf("fixture bindings %v, want only core's galactic-edge-provider binding", bindings) + } +} + +func subjectIsFixture(subjects []rbacv1.Subject) bool { + return slices.ContainsFunc(subjects, func(s rbacv1.Subject) bool { + return s.Kind == rbacv1.ServiceAccountKind && s.Namespace == masqueNS && s.Name == providerSA + }) +} + +// rowMissingDevice runs a fixture pod without /dev/net/tun. A connector's +// request gets its attachment programmed, but the fixture cannot open the +// packet device, so it reports the attachment not ready, refuses the +// request and claims no route: nothing forwards. +func (m *masque) rowMissingDevice(t *testing.T) { + l := m.l + token := rand.Text() + b := marshalConfig([]fixtureService{m.a.serviceConfig()}, []fixtureConnector{{ID: "connector-nodev", + Credential: token, Service: m.a.serviceKey(), Prefix: masquePartial}}) + pod := m.deploy(t, "nodev", edgeNode1, b, ` - target: + kind: Deployment + patch: |- + - op: remove + path: /spec/template/spec/containers/0/resources +`) + defer func() { + if _, err := l.kubectl("-n", masqueNS, "delete", "deployment", "masque-provider-nodev", "--wait=true", + "--timeout=120s"); err != nil { + t.Error(err) + } + }() + dst := m.workload(t, m.a.tn, 2).addr + var reason string + eventually(t, 2*time.Minute, 2*time.Second, "the device-less pod's attachment reported", func() (bool, error) { + evs, err := connectOnceTo(pod, token, "10", 1, time.Second, dst) + if err != nil { + return false, err + } + if s := status(evs); s == 200 || len(echoes(evs, dst)) > 0 { + return false, fmt.Errorf("a pod with no packet device accepted a request: status %d", s) + } + att := l.attachmentOf(t, pod, m.a.svc) + if att == nil || !meta.IsStatusConditionTrue(att.Status.Conditions, galacticv1alpha1.EdgeAttachmentProgrammed) { + return false, nil + } + c := meta.FindStatusCondition(att.Status.Conditions, galacticv1alpha1.EdgeAttachmentReady) + if c == nil || c.Status != metav1.ConditionFalse { + return false, nil + } + reason = c.Reason + ": " + c.Message + return true, nil + }) + l.record(t, "attachment of a pod without /dev/net/tun", reason) + if !strings.HasPrefix(reason, "DataPlaneFailed") { + t.Errorf("attachment of a pod without /dev/net/tun: %s, want DataPlaneFailed", reason) + } + if r := l.route(t, masqueNS, m.a.vpcUID, netip.MustParsePrefix(masquePartial)); r != nil { + t.Errorf("a pod with no packet device claimed route %s", r.Name) + } +} + +// rowRegistration records how the uSID datapath delivers to each fixture +// attachment. galactic-edge registers the attachment's host link, which is +// the host end of a veth whose peer is in the provider pod; the packet +// device sits behind that peer in the pod's VRF. So the host link is +// registered as a veth and delivered to with bpf_redirect_peer, never as a +// tap: the TUN device is not on the host. M07 proves both directions +// through it. +func (m *masque) rowRegistration(t *testing.T) { + m.inUse(t) + for _, mt := range []*masqueTenant{m.a, m.b} { + att := m.l.attachmentOf(t, m.pod, mt.svc) + link, err := nodeExec(edgeNode1, "ip", "-d", "-o", "link", "show", "dev", att.Status.HostLink) + if err != nil { + t.Fatal(err) + } + if !strings.Contains(link, " veth ") { + t.Errorf("VPC %s host link %s is not a veth: %s", mt.tn.vpc, att.Status.HostLink, link) + } + out, err := nodeExec(edgeNode1, "/usr/local/bin/edgeprobe", "egresskind", "-iface", att.Status.HostLink) + if err != nil { + t.Fatal(err) + } + var row struct { + Kind uint32 `json:"kind"` + Present bool `json:"present"` + } + if err := json.Unmarshal([]byte(out), &row); err != nil { + t.Fatalf("egresskind: %v: %s", err, out) + } + m.l.record(t, "VPC "+mt.tn.vpc+" host link delivery", out) + if !row.Present || row.Kind != 0 { + t.Errorf("VPC %s host link %s: delivery row %+v, want a registered veth (kind 0)", mt.tn.vpc, + att.Status.HostLink, row) + } + } +} + +// rowRouteDrift adds a route through the fixture's pod link to its pod VRF, +// as a process with NET_ADMIN in the pod could, while a client is +// connected. galactic-edge removes it at its next pass, without taking the +// attachment out of service. +func (m *masque) rowRouteDrift(t *testing.T) { + m.inUse(t) + l := m.l + att := l.attachmentOf(t, m.pod, m.a.svc) + sandbox := strconv.Itoa(podPID(t, edgeNode1, m.pod.Namespace, m.pod.Name)) + inPod := func(args ...string) (string, error) { + return nodeExec(edgeNode1, append([]string{nsenter, "-t", sandbox, "-n"}, args...)...) + } + links, err := inPod("ip", "-o", "link", "show", "master", att.Status.PodVRF) + if err != nil { + t.Fatal(err) + } + var podLink string + for _, line := range strings.Split(links, "\n") { + if f := strings.Fields(line); len(f) > 1 { + if name := strings.Split(strings.TrimSuffix(f[1], ":"), "@")[0]; name != att.Status.PacketDevice { + podLink = name + } + } + } + if podLink == "" { + t.Fatalf("no pod link in %s: %s", att.Status.PodVRF, links) + } + const stray = "fd00:dead:974::/64" + dst := m.workload(t, m.a.tn, 2).addr + p := m.connect(t, "drift-continuity", m.a.token, "10", m.a.id, 90*time.Second, dst) + time.Sleep(3 * time.Second) + if _, err := inPod("ip", "-6", "route", "add", stray, "dev", podLink, "vrf", att.Status.PodVRF); err != nil { + t.Fatal(err) + } + added := time.Now() + var repaired time.Duration + eventually(t, 150*time.Second, time.Second, "stray route removed", func() (bool, error) { + out, err := inPod("ip", "-6", "route", "show", "vrf", att.Status.PodVRF, stray) + if err != nil { + return false, err + } + cur := l.attachmentOf(t, m.pod, m.a.svc) + if !meta.IsStatusConditionTrue(cur.Status.Conditions, galacticv1alpha1.EdgeAttachmentProgrammed) { + t.Errorf("the attachment left Programmed while the stray route was present") + } + repaired = time.Since(added) + return strings.TrimSpace(out) == "", nil + }) + l.record(t, "stray pod link route removed after", repaired.Round(time.Second)) + evs := p.wait(l, t) + requireSession(t, "the connected client's request", evs, 90*time.Second, 1) + rs := echoes(evs, dst) + l.record(t, "connected client across the repair", fmt.Sprintf("%d/%d, largest gap %v", len(rs), + sentCount(evs), largestGap(rs))) + if len(rs) < sentCount(evs)*9/10 || largestGap(rs) > time.Second { + t.Errorf("connected client answered %d of %d with a %v gap across the repair", len(rs), sentCount(evs), + largestGap(rs)) + } +} diff --git a/tests/edgeconformance/masque_test.go b/tests/edgeconformance/masque_test.go index f7929125..f3987cd0 100644 --- a/tests/edgeconformance/masque_test.go +++ b/tests/edgeconformance/masque_test.go @@ -122,8 +122,12 @@ func TestMASQUE(t *testing.T) { t.FailNow() } t.Run("M16-unprivileged-provider", m.rowUnprivileged) + t.Run("M16-refused-requests", m.rowRefusedRequests) + t.Run("M16-missing-device", m.rowMissingDevice) + t.Run("A06-packet-access-registration", m.rowRegistration) t.Run("M07-isolation", func(t *testing.T) { m.isolation(t, "cold") }) t.Run("nat-egress", m.rowNATEgress) + t.Run("A16-pod-link-route-drift", m.rowRouteDrift) t.Run("shared-attachment-cancel", m.rowSharedCancel) t.Run("maintenance-during-use", m.rowMaintenance) t.Run("process-restart", m.rowProcessRestart) @@ -218,7 +222,7 @@ func (m *masque) setup(t *testing.T) { for _, mt := range []*masqueTenant{m.a, m.b} { mt.vpcUID, mt.vpcID = l.vpcUID(t, mt.tn), l.vpcID(t, mt.tn) mt.svc = l.grantService(t, mt.tn, masqueService, func(s *galacticv1alpha1.EdgeService) { - s.Spec.Capabilities = []string{"PacketAccess", "PrefixReachability"} + s.Spec.Capabilities = []string{capPacket, "PrefixReachability"} s.Spec.Prefixes = []string{masqueGrant} s.Spec.ClientPool = masquePool }) @@ -298,7 +302,7 @@ func marshalConfig(svcs []fixtureService, conns []fixtureConnector) string { // deploy runs the fixture's Deployment name on node with cfg and returns its // running pod. -func (m *masque) deploy(t *testing.T, name, node, cfg string) *corev1.Pod { +func (m *masque) deploy(t *testing.T, name, node, cfg string, patches ...string) *corev1.Pod { t.Helper() l := m.l secret := &corev1.Secret{ @@ -339,7 +343,7 @@ patches: - op: add path: /spec/selector/matchLabels/edge-conformance~1instance value: %s -`, masqueNS, name, name, node, name, name, name), nil) +`, masqueNS, name, name, node, name, name, name)+strings.Join(patches, ""), nil) return l.waitProviderPod(t, masqueNS, name, node) } @@ -353,6 +357,16 @@ func (m *masque) waitServing(t *testing.T, pod *corev1.Pod) { } } +// inUse connects both tenants and waits until the fixture serves their +// attachments. The adapter deletes an attachment that has had no request +// for its idle period, so a row that reads the attachments calls this first. +func (m *masque) inUse(t *testing.T) { + t.Helper() + m.waitConnected(t, m.a) + m.waitConnected(t, m.b) + m.waitServing(t, m.pod) +} + // waitConnected waits until mt's connector is admitted and its tenant's // workload on the provider's node answers through it. func (m *masque) waitConnected(t *testing.T, mt *masqueTenant) { @@ -883,6 +897,7 @@ func (m *masque) rowSharedCancel(t *testing.T) { // after collection is set up and answered. func (m *masque) rowMaintenance(t *testing.T) { l, ctx := m.l, context.Background() + m.inUse(t) atts := map[*masqueTenant]*galacticv1alpha1.EdgeAttachment{} for _, mt := range []*masqueTenant{m.a, m.b} { atts[mt] = l.attachmentOf(t, m.pod, mt.svc) @@ -990,6 +1005,7 @@ func (m *masque) rowMaintenance(t *testing.T) { // attachment objects, under a later lease epoch. func (m *masque) rowProcessRestart(t *testing.T) { l, ctx := m.l, context.Background() + m.inUse(t) atts := map[*masqueTenant]types.UID{} for _, mt := range []*masqueTenant{m.a, m.b} { atts[mt] = l.attachmentOf(t, m.pod, mt.svc).UID @@ -1223,7 +1239,7 @@ func (m *masque) teardown(t *testing.T) { if m.pod != nil { m.currentLinks(t) } - if _, err := l.kubectl("-n", masqueNS, "delete", "deployment,replicaset", "--all", "--wait=true", + if _, err := l.kubectl("-n", masqueNS, "delete", "deployment,replicaset,pod", "--all", "--wait=true", "--timeout=120s"); err != nil { t.Error(err) }