From 51d32c6133860e426c2825c5498755c16f58ccf8 Mon Sep 17 00:00:00 2001 From: Peter Sprygada Date: Tue, 27 Oct 2026 15:00:06 -0400 Subject: [PATCH] test(edge): Run the MASQUE lab fixture on the provider adapter The MASQUE conformance fixture carried its own copy of the provider lifecycle: it set up every attachment and route at start and checked only its instance lease before forwarding. It now runs internal/edge/masque behind a static test Authorizer, so the lab exercises the adapter the CONNECT-IP transports use. The adapter requests an attachment and claims a connector's route when a request first needs them and releases the claim when the connector's last request ends. The lab rows therefore set up through requests, and the checks that routes stay accepted run with requests open. Three rows are new. Two connectors share one attachment while one's request ends and the other keeps forwarding. galactic-router's garbage collection runs while the fixture's attachments are their VPCs' only users on the node. A provider process killed in place adopts its attachments when it restarts. Key changes: - Replace the fixture's provider lifecycle with masque.Adapter and a static Authorizer over its config, which now names each VPC's base62 identifier - Rework setup, admission, partial-setup and recreation rows for on-demand claims, and add the shared-attachment, maintenance and process-restart rows - Add docs/edge/masque-provider.md for setup, readiness, drain and cleanup, and point the MASQUE design and acceptance documents at it --- docs/designs/masque/acceptance.md | 2 +- docs/designs/masque/contract.md | 2 +- docs/edge/README.md | 1 + docs/edge/masque-provider.md | 53 ++ tests/edgeconformance/fixtures_test.go | 19 +- tests/edgeconformance/masque_test.go | 354 ++++++++++-- .../edgeconformance/masqueprovider/config.go | 50 +- tests/edgeconformance/masqueprovider/main.go | 115 ++-- .../masqueprovider/provider.go | 524 ------------------ .../masqueprovider/provider_test.go | 149 +---- 10 files changed, 507 insertions(+), 762 deletions(-) create mode 100644 docs/edge/masque-provider.md delete mode 100644 tests/edgeconformance/masqueprovider/provider.go diff --git a/docs/designs/masque/acceptance.md b/docs/designs/masque/acceptance.md index 99e8594a..692a96e4 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, 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. +`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. ## Epic and implementation checks diff --git a/docs/designs/masque/contract.md b/docs/designs/masque/contract.md index 1f8dcec5..75046ecb 100644 --- a/docs/designs/masque/contract.md +++ b/docs/designs/masque/contract.md @@ -82,7 +82,7 @@ The existing implementation mechanism is `EdgeService.spec.validUntil` and [the ## Interfaces and session transitions -Use the existing [connectip interfaces](../../../internal/edge/connectip/backend.go): `Admission.Admit`, `Lease.Info/Routes/Done/Close`, and `PacketBackend.Open`. Their production adapter is follow-on work. `SessionInfo.ID` is the request generation, `Connector` is the connector UUID, `VPC` carries the immutable VPC identity, and `Epoch` is the Edge route epoch. Do not overload one epoch for both session and foundation authority. The adapter keeps the full attachment/service/route UIDs, instance incarnation/epoch and process boot ID beside this snapshot; string names alone cannot authorize I/O. +Use the existing [connectip interfaces](../../../internal/edge/connectip/backend.go): `Admission.Admit`, `Lease.Info/Routes/Done/Close`, and `PacketBackend.Open`. [`internal/edge/masque`](../../edge/masque-provider.md) implements them over the provider contract. `SessionInfo.ID` is the request generation, `Connector` is the connector UUID, `VPC` carries the immutable VPC identity, and `Epoch` is the Edge route epoch. Do not overload one epoch for both session and foundation authority. The adapter keeps the full attachment/service/route UIDs, instance incarnation/epoch and process boot ID beside this snapshot; string names alone cannot authorize I/O. 1. **Pending:** reserve bounded pre-admission capacity before credential parsing or API work. Authenticate and authorize with the authority. No TUN write or MTU probe yet. 2. **Allocated:** read the retained /96 and finite service grant; reserve one connector session transition. A failed admission releases provisional request resources, not the durable allocation. diff --git a/docs/edge/README.md b/docs/edge/README.md index 502f3633..94827f6f 100644 --- a/docs/edge/README.md +++ b/docs/edge/README.md @@ -4,6 +4,7 @@ Galactic Edge gives providers authorized attachments to tenant VPCs and controls - [Provider contract](provider-contract.md): resources, writers, lifecycle, timing assumptions and required failure evidence. - [Troubleshooting](troubleshooting.md): conditions, Events, metrics and locating the current prefix holder. +- [MASQUE provider adapter](masque-provider.md): setup, readiness, drain and cleanup of the CONNECT-IP provider's attachments and routes. - [CONNECT-IP interoperability](connectip-interop.md): HTTP/3 and HTTP/2 protocol proof with an independent client, selections and remaining HTTP/2 listener work. - [Foundation metrics](../observability/edge.md): lifecycle and forwarding-gate measurements. - [Design record](../designs/galactic-edge.md): motivation and design alternatives. diff --git a/docs/edge/masque-provider.md b/docs/edge/masque-provider.md new file mode 100644 index 00000000..60e4f441 --- /dev/null +++ b/docs/edge/masque-provider.md @@ -0,0 +1,53 @@ +# MASQUE provider adapter + +`internal/edge/masque` connects a CONNECT-IP server (`internal/edge/connectip`) to a tenant VPC through the [provider contract](provider-contract.md). It implements `connectip.Admission` and `connectip.PacketBackend`, so every transport uses the same attachment, route and packet checks. It has no command of its own: the service binary that embeds it, and the grant authority that issues its grants, are separate deliverables of the [MASQUE connector design](../designs/masque/contract.md). The conformance fixture in `tests/edgeconformance/masqueprovider` runs it with static test credentials. + +The adapter never touches host networking. galactic-edge programs and releases each attachment's host state, core accepts or refuses each route, and the adapter only writes what the contract assigns to a provider: its instance and Lease, its attachments and their `Ready` condition, and its route claims. + +## Setup + +The cell must run galactic-controller with `--edge` and the galactic-edge DaemonSet on the provider's node ([deployment](provider-contract.md)). The provider's node also needs a TUN device plugin, and the VPC needs a workload on that node, because an attachment joins the VPC's existing host VRF. + +1. Register the provider: an `EdgeProvider` that lists contract version `v1alpha1`, `PacketAccess/v1` and `PrefixReachability/v1`, and names the provider's ServiceAccount. An operator approves it (`status.approved`). galactic-controller then binds the provider role and admission policy to that ServiceAccount. +2. Grant each connector an `EdgeService` in its VPC's namespace. It grants `PacketAccess` and `PrefixReachability`, lists the connector's /96 in `prefixes`, and sets a `clientPool` outside every connector /96. A service that lacks `PrefixReachability` gets an attachment, but core refuses every route on it. +3. Run the provider pod as that ServiceAccount, with a pod-bound token, the downward-API pod name, namespace, UID and node name, a device-plugin TUN, and no host namespace, hostPath, `NET_ADMIN` or BPF access. `tests/edgeconformance/masqueprovider/deploy/` is a worked example. + +The adapter's `Authorizer` turns a request's credential and VPC selector into a `Grant`: connector ID, VPC UID and base62 identifier, the `EdgeService` reference, the connector /96, the routes to advertise and an absolute deadline. A grant the adapter cannot use (wrong VPC, no service, not a masked IPv6 /96, already expired) is refused. + +## Readiness + +The adapter registers its instance at start and admits nothing until it holds the instance's Lease. It prepares a connector on the connector's first request: + +1. It requests the pod's attachment to the grant's service (`ea--`), requesting `PacketAccess` only, and verifies the stored object names this pod, instance, provider and service. +2. It waits until core admits the attachment for the grant's VPC on this node and galactic-edge reports it `Programmed` with a packet device. It then opens that device and sets `Ready=True` with reason `Serving`. +3. It claims the connector's route (`er-` and a hash of the VPC UID and /96) and waits until core accepts that claim for this attachment's UID and this Lease's incarnation and epoch. + +Only then does the request get its packet path. The request waits up to 30 seconds for these steps (`Config.Prepare`), including any wait behind another request preparing the same attachment or route, and is answered 503 if they do not complete. It fails at once, also with 503, when core refuses the attachment, the attachment is draining, its status names another VPC or node or no packet device, the device cannot be opened, or core refuses the route outright (for example `CapabilityNotGranted`, `PrefixNotGranted` or `PrefixConflict`). On a refusal after the device is known, the attachment reports `Ready=False` with reason `DataPlaneFailed`. + +Connectors of one service share its attachment and device. A connector has at most one open request: another request for it is answered 503 before any preparation, and the open request is left in place. + +## Forwarding + +A request forwards a packet, in either direction, only while its instance Lease is held at the epoch it was admitted under and before the Lease's deadline, before its grant's deadline, and while the request is open. Every second the adapter also checks each attachment and route it uses. It ends the requests on an attachment that is deleted, replaced, draining, no longer programmed or reprogrammed, and the requests on a route whose claim core no longer accepts, whose gate the node closed, or whose claim moved to another attachment. Losing the lease ends every request. A full device drops the packet and counts it instead of ending the request. These checks are in addition to the foundation's: galactic-edge closes a route's gate at the Lease deadline even if the provider stops. + +## Drain and shutdown + +When core drains an attachment, after its service is deleted or its authority ends, the adapter ends that attachment's requests at its next check and serves no new request on it. Core deletes the attachment after the service's `drainSeconds`. + +On shutdown, stop the transports first so every request ends, then call `Shutdown`. It stops admitting, releases each route claim the adapter still holds, deletes its attachments and releases the instance Lease. A replacement can therefore take nothing over while this pod still forwards. Shutdown is bounded by the caller's context; the fixture gives it 20 seconds. + +## Cleanup + +- When a connector's last request ends, the adapter releases that connector's route claim. Core deletes the route once its attachments are gone. +- An attachment with no request for 30 seconds (`Config.Idle`) is deleted. A reconnect within that time, such as an HTTP/3 to HTTP/2 fallback, reuses it. +- Every delete carries the attachment's UID as a precondition, so a late delete never removes a replacement under the same name. +- Core holds a deleted attachment with its finalizer until the node reports `Released`. A new request for the same service waits for that before requesting a new attachment. +- A process that restarts in the same pod adopts the pod's attachments and reports them `Ready=False` until it serves them again on request. It deletes the unused ones once idle. It restores no forwarding permission from the previous process; each route is claimed again under the Lease it acquires. +- When the pod is deleted without shutdown, core reaps its instance, Lease, attachments and routes ([contract](provider-contract.md)). + +`kubectl get edgeattachments,edgeroutes,leases -n ` shows what remains, and [troubleshooting](troubleshooting.md) explains each condition. + +## Verification + +- `go test ./internal/edge/masque/` runs the adapter's unit tests and, with `KUBEBUILDER_ASSETS` set, `TestAgainstCore`. That test runs the adapter as a provider pod's pod-bound identity against core's attachment and route controllers and the shipped RBAC and admission policies. +- `task verify:masque-conformance`, run from `deploy/containerlab`, runs the fixture in the lab: packet access, isolation, NAT egress, shared-attachment cancellation, maintenance, process restart, partial setup, recreation and teardown. A lab result is not target-platform qualification. diff --git a/tests/edgeconformance/fixtures_test.go b/tests/edgeconformance/fixtures_test.go index d9083c49..9aa2744c 100644 --- a/tests/edgeconformance/fixtures_test.go +++ b/tests/edgeconformance/fixtures_test.go @@ -153,6 +153,23 @@ func (l *lab) vpcUID(t *testing.T, tn tenant) string { return string(v.UID) } +// appLabel selects a tenant's edge-client pods. +const appLabel = "app" + +// vpcID returns the base62 identifier of tn's VPC, as core reports it on +// an admitted attachment. +func (l *lab) vpcID(t *testing.T, tn tenant) string { + t.Helper() + var v cloudv1alpha1.VPC + if err := l.c.Get(context.Background(), client.ObjectKey{Namespace: tn.vpcNamespace, Name: tn.vpc}, &v); err != nil { + t.Fatal(err) + } + if v.Status.VPC == "" { + t.Fatalf("VPC %s/%s has no identifier yet", tn.vpcNamespace, tn.vpc) + } + return v.Status.VPC +} + // grantService creates or replaces the platform authority's grant of tn's // VPC to tn's provider, and returns it as stored. func (l *lab) grantService(t *testing.T, tn tenant, name string, mutate func(*galacticv1alpha1.EdgeService), @@ -396,7 +413,7 @@ func (l *lab) clientPID(t *testing.T, tn tenant, node string) int { i := map[string]string{computeNode: "1", edgeNode1: "2", edgeNode2: "3"}[node] var pods corev1.PodList if err := l.c.List(context.Background(), &pods, client.InNamespace(tn.vpcNamespace), - client.MatchingLabels{"app": "edge-client-" + i}); err != nil { + client.MatchingLabels{appLabel: "edge-client-" + i}); err != nil { t.Fatal(err) } for _, p := range pods.Items { diff --git a/tests/edgeconformance/masque_test.go b/tests/edgeconformance/masque_test.go index 74991c60..d78a631c 100644 --- a/tests/edgeconformance/masque_test.go +++ b/tests/edgeconformance/masque_test.go @@ -25,6 +25,7 @@ import ( "testing" "time" + appsv1 "k8s.io/api/apps/v1" coordinationv1 "k8s.io/api/coordination/v1" corev1 "k8s.io/api/core/v1" apierrors "k8s.io/apimachinery/pkg/api/errors" @@ -48,7 +49,10 @@ const ( masqueGrant = "fd00:ed9e:300::/48" masquePool = "fd00:ed9e:0:30::/64" masquePrefix = "fd00:ed9e:300:1::/96" - masqueOutside = "fd00:ed9e:400:1::/96" + // masqueSecond is VPC 10's second connector, in the same service as + // its first, so the two share one attachment. + masqueSecond = "fd00:ed9e:300:2::/96" + masqueOutside = "fd00:ed9e:400:1::/96" // masquePartial is the partly set-up instance's /96, also outside the // grant and claimed by no other instance. masquePartial = "fd00:ed9e:400:2::/96" @@ -67,6 +71,7 @@ type masqueTenant struct { tn tenant svc *galacticv1alpha1.EdgeService vpcUID string + vpcID string token string id int } @@ -77,6 +82,8 @@ type masque struct { a, b *masqueTenant // refused is the credential of a connector whose route core refuses. refused string + // second is the credential of VPC 10's second connector. + second string certPEM []byte keyPEM []byte // pod is the running pod of the fixture's main Deployment. @@ -90,13 +97,16 @@ type masque struct { } // TestMASQUE proves CONNECT-IP packet access on the Galactic Edge foundation -// with the MASQUE fixture provider (tests/edgeconformance/masqueprovider): -// the deployed agent creates the fixture's PacketAccess devices and the +// with the MASQUE fixture provider (tests/edgeconformance/masqueprovider), +// which runs the MASQUE provider adapter (internal/edge/masque): the +// deployed agent creates the fixture's PacketAccess devices and the // unprivileged fixture opens them, two VPCs with identical connector /96s // stay isolated in both directions, NAT egress works from the attachment -// node, and recreation, partial setup and teardown clean up through core -// while leaving unrelated and replacement resources alone. Run it with -// task verify:masque-conformance. +// node, two connectors share one attachment and ending one leaves the +// other forwarding, a restarted provider process adopts its attachments, +// router maintenance keeps an attachment in use, and recreation, partial +// setup and teardown clean up through core while leaving unrelated and +// replacement resources alone. Run it with task verify:masque-conformance. func TestMASQUE(t *testing.T) { l := getLab(t) t.Cleanup(func() { l.writeSummary(t) }) @@ -110,6 +120,9 @@ func TestMASQUE(t *testing.T) { t.Run("M16-unprivileged-provider", m.rowUnprivileged) t.Run("M07-isolation", func(t *testing.T) { m.isolation(t, "cold") }) t.Run("nat-egress", m.rowNATEgress) + t.Run("shared-attachment-cancel", m.rowSharedCancel) + t.Run("maintenance-during-use", m.rowMaintenance) + t.Run("process-restart", m.rowProcessRestart) t.Run("partial-setup", m.rowPartial) t.Run("recreation", m.rowRecreation) } @@ -133,7 +146,7 @@ func newMasque(t *testing.T, l *lab) *masque { token: rand.Text(), id: 0x0a10}, b: &masqueTenant{tn: tenant{vpcNamespace: "ns20", vpc: "20", provider: masqueProvider}, token: rand.Text(), id: 0x0b20}, - refused: rand.Text(), certPEM: certPEM, keyPEM: keyPEM, + refused: rand.Text(), second: rand.Text(), certPEM: certPEM, keyPEM: keyPEM, links: map[string]string{}, unrelated: map[string]types.UID{}, } } @@ -194,7 +207,7 @@ func (m *masque) setup(t *testing.T) { t.Fatal(err) } for _, mt := range []*masqueTenant{m.a, m.b} { - mt.vpcUID = l.vpcUID(t, mt.tn) + 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.Prefixes = []string{masqueGrant} @@ -203,9 +216,11 @@ func (m *masque) setup(t *testing.T) { l.waitServiceAdmitted(t, mt.svc, true) } m.pod = m.deploy(t, "m1", edgeNode1, m.config(true)) - m.waitServing(t, m.pod) + // The adapter requests a connector's attachment and claims its route + // when a request first needs them. m.waitConnected(t, m.a) m.waitConnected(t, m.b) + m.waitServing(t, m.pod) } // fixtureConfig, fixtureService and fixtureConnector are the fixture's @@ -223,6 +238,7 @@ type fixtureService struct { Name string `json:"name"` UID string `json:"uid"` VPCUID string `json:"vpcUID"` + VPCID string `json:"vpcID"` VPC string `json:"vpc"` } @@ -235,17 +251,20 @@ type fixtureConnector struct { } // config is the fixture's config.yaml: both tenants' services and -// connectors, plus, when withRefused is set, a VPC 10 connector whose /96 -// lies outside the grant, so core refuses its route. +// connectors, VPC 10's second connector, plus, when withRefused is set, a +// VPC 10 connector whose /96 lies outside the grant, so core refuses its +// route. func (m *masque) config(withRefused bool) string { svcs := make([]fixtureService, 0, 2) - conns := make([]fixtureConnector, 0, 3) + conns := make([]fixtureConnector, 0, 4) for _, mt := range []*masqueTenant{m.a, m.b} { svcs = append(svcs, mt.serviceConfig()) conns = append(conns, fixtureConnector{ID: "connector-" + mt.tn.vpc, Credential: mt.token, Service: mt.serviceKey(), Prefix: masquePrefix, Routes: []string{"fd20:" + mt.tn.vpc + "::/32", remoteHost + "/128"}}) } + conns = append(conns, fixtureConnector{ID: "connector-10-second", Credential: m.second, + Service: m.a.serviceKey(), Prefix: masqueSecond, Routes: []string{"fd20:10::/32"}}) if withRefused { conns = append(conns, fixtureConnector{ID: "connector-outside", Credential: m.refused, Service: m.a.serviceKey(), Prefix: masqueOutside}) @@ -256,7 +275,7 @@ func (m *masque) config(withRefused bool) string { // serviceConfig is mt's grant as the fixture's config.yaml lists it. func (mt *masqueTenant) serviceConfig() fixtureService { return fixtureService{Namespace: mt.svc.Namespace, Name: mt.svc.Name, UID: string(mt.svc.UID), - VPCUID: mt.vpcUID, VPC: mt.tn.vpc} + VPCUID: mt.vpcUID, VPCID: mt.vpcID, VPC: mt.tn.vpc} } // serviceKey is how a fixture connector names mt's service. @@ -315,14 +334,13 @@ patches: return l.waitProviderPod(t, masqueNS, name, node) } -// waitServing waits until pod's attachments to both services are ready and -// both tenants' routes are accepted for them, and records their host links. +// waitServing waits until pod's attachments to both services are ready, +// and records their host links. func (m *masque) waitServing(t *testing.T, pod *corev1.Pod) { t.Helper() for _, mt := range []*masqueTenant{m.a, m.b} { att := m.l.waitServing(t, pod, mt.svc) m.links[att.Name] = att.Status.HostLink - m.l.waitAccepted(t, masqueNS, mt.vpcUID, netip.MustParsePrefix(masquePrefix), att.Name, 3*time.Minute) } } @@ -346,22 +364,28 @@ func (m *masque) waitConnected(t *testing.T, mt *masqueTenant) { // where the client runs does not change the gVPC path behind the fixture. func (m *masque) connect(t *testing.T, name, token, vpc string, id int, d time.Duration, dsts ...string) *probeRun { t.Helper() - p, err := startProbe(m.pod.Spec.NodeName, 1, name, m.connectArgs(token, vpc, id, d, dsts)...) + p, err := startProbe(m.pod.Spec.NodeName, 1, name, connectArgs(m.pod, token, vpc, id, d, dsts)...) if err != nil { t.Fatal(err) } return p } -func (m *masque) connectArgs(token, vpc string, id int, d time.Duration, dsts []string) []string { - return []string{"connectip", "-server", "[" + m.pod.Status.PodIP + "]:" + strconv.Itoa(masquePort), +func connectArgs(pod *corev1.Pod, token, vpc string, id int, d time.Duration, dsts []string) []string { + return []string{"connectip", "-server", "[" + pod.Status.PodIP + "]:" + strconv.Itoa(masquePort), "-authority", masqueAuthority, "-ca", masqueCA, "-token", token, "-vpc", vpc, "-id", strconv.Itoa(id), "-to", strings.Join(dsts, ","), "-interval", probeInterval.String(), "-duration", d.String()} } -// connectOnce runs one short session and returns its events. +// connectOnce runs one short session to the fixture's pod and returns its +// events. func (m *masque) connectOnce(token, vpc string, id int, d time.Duration, dsts ...string) ([]event, error) { - p, err := startProbe(m.pod.Spec.NodeName, 1, "once", m.connectArgs(token, vpc, id, d, dsts)...) + return connectOnceTo(m.pod, token, vpc, id, d, dsts...) +} + +// connectOnceTo runs one short session to pod and returns its events. +func connectOnceTo(pod *corev1.Pod, token, vpc string, id int, d time.Duration, dsts ...string) ([]event, error) { + p, err := startProbe(pod.Spec.NodeName, 1, "once", connectArgs(pod, token, vpc, id, d, dsts)...) if err != nil { return nil, err } @@ -410,6 +434,12 @@ func others(evs []event) int { // refused request, or a tunnel the server closed early, sends fewer than 90% // of the probes the window holds, so its reply ratio alone proves nothing. func requireSession(t *testing.T, what string, evs []event, d time.Duration, n int) { + t.Helper() + requireSessionFor(t, what, evs, d, n, masquePrefix) +} + +// requireSessionFor is requireSession for a connector assigned prefix. +func requireSessionFor(t *testing.T, what string, evs []event, d time.Duration, n int, prefix string) { t.Helper() if s := status(evs); s != 200 { t.Errorf("%s: status %d, want 200", what, s) @@ -420,14 +450,29 @@ func requireSession(t *testing.T, what string, evs []event, d time.Duration, n i assigned = e.Assigned } } - if assigned != masquePrefix { - t.Errorf("%s: assigned %q, want %s", what, assigned, masquePrefix) + if assigned != prefix { + t.Errorf("%s: assigned %q, want %s", what, assigned, prefix) } if want, got := int(d/probeInterval)*n, sentCount(evs); got < want*9/10 { t.Errorf("%s: sent %d of the window's %d probes", what, got, want) } } +// requireAnswered fails unless evs is a whole session of d to dst in which +// at least 90% of the probes were answered, no two answers were more than a +// second apart, and nothing else arrived. +func (m *masque) requireAnswered(t *testing.T, what string, evs []event, d time.Duration, dst string) { + t.Helper() + requireSession(t, what, evs, d, 1) + rs := echoes(evs, dst) + m.l.record(t, what, fmt.Sprintf("%d/%d, largest gap %v, %d other", len(rs), sentCount(evs), largestGap(rs), + others(evs))) + if len(rs) < sentCount(evs)*9/10 || largestGap(rs) > time.Second || others(evs) > 0 { + t.Errorf("%s: %d of %d answered, largest gap %v, %d other packets", what, len(rs), sentCount(evs), + largestGap(rs), others(evs)) + } +} + func status(evs []event) int { for _, e := range evs { if e.Status != 0 { @@ -471,7 +516,7 @@ func (m *masque) client(t *testing.T, tn tenant, i int) *corev1.Pod { t.Helper() var pods corev1.PodList if err := m.l.c.List(context.Background(), &pods, client.InNamespace(tn.vpcNamespace), - client.MatchingLabels{"app": "edge-client-" + strconv.Itoa(i)}); err != nil { + client.MatchingLabels{appLabel: "edge-client-" + strconv.Itoa(i)}); err != nil { t.Fatal(err) } for i := range pods.Items { @@ -558,8 +603,6 @@ func (m *masque) checkAdmission(t *testing.T) { t.Helper() l := m.l dst := m.workload(t, m.a.tn, 2).addr - att := l.attachmentOf(t, m.pod, m.a.svc) - l.record(t, "outside-grant route", m.waitRefused(t, m.a.vpcUID, netip.MustParsePrefix(masqueOutside), att.Name)) for _, c := range []struct { what, token, vpc string want int @@ -577,6 +620,9 @@ func (m *masque) checkAdmission(t *testing.T) { t.Errorf("%s: status %d with %d replies, want %d and none", c.what, got, len(echoes(evs, dst)), c.want) } } + // The refused connector's request claimed its route, which core refused. + att := l.attachmentOf(t, m.pod, m.a.svc) + l.record(t, "outside-grant route", m.waitRefused(t, m.a.vpcUID, netip.MustParsePrefix(masqueOutside), att.Name)) } // recordVersions saves what the run executed against: Kubernetes, the @@ -739,15 +785,24 @@ func (m *masque) rowNATEgress(t *testing.T) { func (m *masque) rowPartial(t *testing.T) { l := m.l dst := m.workload(t, m.a.tn, 2).addr + token := rand.Text() b := marshalConfig([]fixtureService{m.a.serviceConfig()}, []fixtureConnector{{ID: "connector-partial", - Credential: rand.Text(), Service: m.a.serviceKey(), Prefix: masquePartial}}) + Credential: token, Service: m.a.serviceKey(), Prefix: masquePartial}}) pod := m.deploy(t, "partial", edgeNode1, b) + // Its only connector's request gets the attachment set up, and is + // refused once core refuses the route. + var evs []event + eventually(t, 2*time.Minute, 2*time.Second, "partial instance's request refused", func() (bool, error) { + var err error + evs, err = connectOnceTo(pod, token, "10", 1, time.Second, dst) + return err == nil && status(evs) == 503, err + }) att := l.waitServing(t, pod, m.a.svc) m.links[att.Name] = att.Status.HostLink route := m.waitRefused(t, m.a.vpcUID, netip.MustParsePrefix(masquePartial), att.Name) l.record(t, "partial instance's route", route) - p := m.connect(t, "partial-continuity", m.a.token, "10", m.a.id, 20*time.Second, dst) + p := m.connect(t, "partial-continuity", m.a.token, "10", m.a.id, 40*time.Second, dst) time.Sleep(3 * time.Second) // Orphaning the Deployment and its ReplicaSet leaves nothing to replace // the pod, which is then killed without running its shutdown. @@ -761,8 +816,9 @@ func (m *masque) rowPartial(t *testing.T) { t.Fatal(err) } killed := time.Now() - evs := p.wait(l, t) - requireSession(t, "first instance's request", evs, 20*time.Second, 1) + m.waitAcceptedFor(t, 20*time.Second, m.a) + evs = p.wait(l, t) + requireSession(t, "first instance's request", evs, 40*time.Second, 1) rs := echoes(evs, dst) l.record(t, "first instance across the kill", fmt.Sprintf("%d/%d, largest gap %v", len(rs), sentCount(evs), largestGap(rs))) @@ -772,7 +828,219 @@ func (m *masque) rowPartial(t *testing.T) { } m.waitReaped(t, pod.UID, []*galacticv1alpha1.EdgeAttachment{att}, killed) m.waitNoRoute(t, m.a.vpcUID, netip.MustParsePrefix(masquePartial)) - m.waitAcceptedFor(t, 20*time.Second) +} + +// rowSharedCancel runs VPC 10's two connectors at once. Both are in one +// service, so they share its attachment and packet device. The second's +// request ends while the first's runs: the first keeps forwarding without a +// gap and its route stays accepted, the second's claim is released, and the +// attachment is neither replaced nor taken out of service. +func (m *masque) rowSharedCancel(t *testing.T) { + l := m.l + dst := m.workload(t, m.a.tn, 2).addr + att := l.attachmentOf(t, m.pod, m.a.svc) + first := m.connect(t, "shared-first", m.a.token, "10", m.a.id, 45*time.Second, dst) + time.Sleep(3 * time.Second) + second := m.connect(t, "shared-second", m.second, "10", m.a.id+1, 5*time.Second, dst) + evs := second.wait(l, t) + requireSessionFor(t, "second connector's request", evs, 5*time.Second, 1, masqueSecond) + if n := len(echoes(evs, dst)); n < sentCount(evs)*9/10 || others(evs) > 0 { + t.Errorf("second connector: %d of %d answered, %d other packets", n, sentCount(evs), others(evs)) + } + ended := time.Now() + r := l.route(t, masqueNS, m.a.vpcUID, netip.MustParsePrefix(masqueSecond)) + if r == nil || r.Spec.Claim == nil || r.Spec.Claim.Attachment != att.Name { + t.Fatalf("the second connector's route was not claimed on the shared attachment %s", att.Name) + } + eventually(t, 10*time.Second, 500*time.Millisecond, "second connector's claim released", func() (bool, error) { + r := l.route(t, masqueNS, m.a.vpcUID, netip.MustParsePrefix(masqueSecond)) + return r != nil && r.Spec.Claim != nil && r.Spec.Claim.Released, nil + }) + l.record(t, "second connector's claim released after its request ended", time.Since(ended).Round(time.Second)) + m.waitAcceptedFor(t, 15*time.Second, m.a) + m.requireAnswered(t, "first connector across the second's end", first.wait(l, t), 45*time.Second, dst) + cur := l.attachmentOf(t, m.pod, m.a.svc) + if cur == nil || cur.UID != att.UID || + !meta.IsStatusConditionTrue(cur.Status.Conditions, galacticv1alpha1.EdgeAttachmentReady) { + t.Error("ending one connector's request replaced the shared attachment or took it out of service") + } +} + +// rowMaintenance leaves the fixture's attachments their VPCs' only users on +// the fixture's node, by scaling each tenant's edge client there to zero, +// and lets galactic-router's garbage collection run while a request of each +// tenant reaches its workload on the compute node. The VRFs, the +// attachments and the requests survive it, and a connector's first request +// after collection is set up and answered. +func (m *masque) rowMaintenance(t *testing.T) { + l, ctx := m.l, context.Background() + atts := map[*masqueTenant]*galacticv1alpha1.EdgeAttachment{} + for _, mt := range []*masqueTenant{m.a, m.b} { + atts[mt] = l.attachmentOf(t, m.pod, mt.svc) + old := m.client(t, mt.tn, 2) + key := client.ObjectKey{Namespace: mt.tn.vpcNamespace, Name: "edge-client-2"} + scale := func(n int32) error { + d := &appsv1.Deployment{} + if err := l.c.Get(ctx, key, d); err != nil { + return err + } + d.Spec.Replicas = &n + return l.c.Update(ctx, d) + } + if err := scale(0); err != nil { + t.Fatal(err) + } + // The row replaces this client on purpose; teardown expects its + // replacement. + defer func() { + if err := scale(1); err != nil { + t.Error(err) + return + } + var p *corev1.Pod + eventually(t, 3*time.Minute, 2*time.Second, "edge-client-2 restored in "+mt.tn.vpcNamespace, + func() (bool, error) { + var pods corev1.PodList + err := l.c.List(ctx, &pods, client.InNamespace(mt.tn.vpcNamespace), + client.MatchingLabels{appLabel: "edge-client-2"}) + for i := range pods.Items { + if c := &pods.Items[i]; c.Status.Phase == corev1.PodRunning && c.DeletionTimestamp == nil { + p = c + } + } + return p != nil, err + }) + delete(m.unrelated, old.Namespace+"/"+old.Name) + m.unrelated[p.Namespace+"/"+p.Name] = p.UID + }() + } + // Precondition: no pod of either VPC is left on the fixture's node, so + // the attachment is all that keeps each VPC's VRF there. + for _, mt := range []*masqueTenant{m.a, m.b} { + eventually(t, 3*time.Minute, 2*time.Second, "no VPC "+mt.tn.vpc+" pod on "+edgeNode1, func() (bool, error) { + var pods corev1.PodList + if err := l.c.List(ctx, &pods, client.InNamespace(mt.tn.vpcNamespace)); err != nil { + return false, err + } + for _, p := range pods.Items { + if p.Spec.NodeName == edgeNode1 { + return false, nil + } + } + return true, nil + }) + } + router := l.podOn(t, systemNS, "galactic-router", edgeNode1) + since := time.Now().UTC().Format(time.RFC3339) + interval := l.gcInterval(t) + l.record(t, "galactic-router GC interval", interval) + d := interval + 45*time.Second + runs := map[*masqueTenant]*probeRun{} + for _, mt := range []*masqueTenant{m.a, m.b} { + runs[mt] = m.connect(t, "maintenance-vpc-"+mt.tn.vpc, mt.token, mt.tn.vpc, mt.id, d, + m.workload(t, mt.tn, 1).addr) + } + time.Sleep(interval + 30*time.Second) + logs, err := l.kubectl("-n", systemNS, "logs", router, "--since-time="+since) + if err != nil { + t.Fatal(err) + } + for _, mt := range []*masqueTenant{m.a, m.b} { + vrf := fmt.Sprintf("G%09sV", atts[mt].Status.VPC) + if _, err := nodeExec(edgeNode1, "ip", "link", "show", vrf); err != nil { + t.Errorf("VPC %s's VRF %s is gone with the fixture's attachment still on it: %v", mt.tn.vpc, vrf, err) + } + // A collection pass ran and kept the VRF for the attachment. + if !slices.ContainsFunc(strings.Split(logs, "\n"), func(line string) bool { + return strings.Contains(line, "GC: kept VRF an edge attachment uses") && strings.Contains(line, vrf) + }) { + t.Errorf("no garbage collection pass kept VPC %s's VRF %s during the row", mt.tn.vpc, vrf) + } + } + dst := m.workload(t, m.a.tn, 1).addr + evs, err := m.connectOnce(m.second, "10", m.a.id+1, 3*time.Second, dst) + if err != nil { + t.Fatal(err) + } + requireSessionFor(t, "second connector's first request after collection", evs, 3*time.Second, 1, masqueSecond) + if n := len(echoes(evs, dst)); n < sentCount(evs)*9/10 { + t.Errorf("second connector after collection: %d of %d answered", n, sentCount(evs)) + } + for mt, p := range runs { + m.requireAnswered(t, "VPC "+mt.tn.vpc+" across garbage collection", p.wait(l, t), d, + m.workload(t, mt.tn, 1).addr) + if cur := l.attachmentOf(t, m.pod, mt.svc); cur == nil || cur.UID != atts[mt].UID { + t.Errorf("VPC %s's attachment did not survive garbage collection", mt.tn.vpc) + } + } +} + +// rowProcessRestart kills the fixture's process without shutdown. The +// kubelet restarts its container in the same pod, and the new process +// adopts the pod's attachments: both tenants are served again on the same +// attachment objects, under a later lease epoch. +func (m *masque) rowProcessRestart(t *testing.T) { + l, ctx := m.l, context.Background() + atts := map[*masqueTenant]types.UID{} + for _, mt := range []*masqueTenant{m.a, m.b} { + atts[mt] = l.attachmentOf(t, m.pod, mt.svc).UID + } + leaseKey := client.ObjectKey{Namespace: masqueNS, Name: "ei-" + string(m.pod.UID)} + epoch := func() int32 { + var lease coordinationv1.Lease + if err := l.c.Get(ctx, leaseKey, &lease); err != nil { + t.Fatal(err) + } + if lease.Spec.LeaseTransitions == nil { + return 0 + } + return *lease.Spec.LeaseTransitions + } + before := epoch() + pid, err := containerPID(edgeNode1, m.pod.Namespace, m.pod.Name, "masque-provider") + if err != nil { + t.Fatal(err) + } + if _, err := nodeExec(edgeNode1, "kill", "-KILL", strconv.Itoa(pid)); err != nil { + t.Fatal(err) + } + killed := time.Now() + m.waitConnected(t, m.a) + m.waitConnected(t, m.b) + l.record(t, "serving again after the process was killed", time.Since(killed).Round(time.Second)) + p := l.providerPod(t, masqueNS, "m1") + if p == nil || p.UID != m.pod.UID { + t.Fatal("killing the process replaced the pod") + } + restarts := int32(0) + for _, cs := range p.Status.ContainerStatuses { + restarts += cs.RestartCount + } + l.record(t, "fixture container restarts", restarts) + if restarts == 0 { + t.Error("the fixture's container was not restarted") + } + logs, err := l.kubectl("-n", masqueNS, "logs", m.pod.Name, "-c", "masque-provider") + if err != nil { + t.Fatal(err) + } + for _, mt := range []*masqueTenant{m.a, m.b} { + cur := l.attachmentOf(t, m.pod, mt.svc) + if cur == nil || cur.UID != atts[mt] { + t.Errorf("VPC %s's attachment was replaced instead of adopted", mt.tn.vpc) + continue + } + // The new process found the attachment at start, before any + // request asked for it. + if !slices.ContainsFunc(strings.Split(logs, "\n"), func(line string) bool { + return strings.Contains(line, `"msg":"adopted attachment"`) && strings.Contains(line, cur.Name) + }) { + t.Errorf("the restarted process did not adopt VPC %s's attachment %s", mt.tn.vpc, cur.Name) + } + } + if after := epoch(); after <= before { + t.Errorf("the restarted process forwards under lease epoch %d, not after %d", after, before) + } } // waitRefused waits for core to refuse the claim on p and returns the @@ -832,13 +1100,15 @@ func (m *masque) waitNoRoute(t *testing.T, vpcUID string, p netip.Prefix) { }) } -// waitAcceptedFor checks that both tenants' routes stay accepted for the -// current pod's attachments for d. -func (m *masque) waitAcceptedFor(t *testing.T, d time.Duration) { +// waitAcceptedFor checks that each of tenants' routes stays accepted for +// the current pod's attachment for d. The adapter holds a connector's route +// only while a request of the connector is open, so the caller keeps one +// open throughout. +func (m *masque) waitAcceptedFor(t *testing.T, d time.Duration, tenants ...*masqueTenant) { t.Helper() deadline := time.Now().Add(d) for time.Now().Before(deadline) { - for _, mt := range []*masqueTenant{m.a, m.b} { + for _, mt := range tenants { att := m.l.attachmentOf(t, m.pod, mt.svc) r := m.l.route(t, masqueNS, mt.vpcUID, netip.MustParsePrefix(masquePrefix)) if att == nil || r == nil || r.Status.Accepted == nil || r.Status.Accepted.Attachment != att.Name || @@ -873,12 +1143,22 @@ func (m *masque) rowRecreation(t *testing.T) { m.pod = p return true, nil }) - m.waitServing(t, m.pod) m.waitConnected(t, m.a) m.waitConnected(t, m.b) + m.waitServing(t, m.pod) l.record(t, "replacement serving after", time.Since(deleted).Round(time.Second)) m.waitReaped(t, old.UID, oldAtts, deleted) - m.waitAcceptedFor(t, 20*time.Second) + runs := map[*masqueTenant]*probeRun{} + for _, mt := range []*masqueTenant{m.a, m.b} { + runs[mt] = m.connect(t, "recreated-vpc-"+mt.tn.vpc, mt.token, mt.tn.vpc, mt.id, 30*time.Second, + m.workload(t, mt.tn, 2).addr) + } + time.Sleep(3 * time.Second) + m.waitAcceptedFor(t, 20*time.Second, m.a, m.b) + for mt, p := range runs { + m.requireAnswered(t, "VPC "+mt.tn.vpc+" after the old pod was reaped", p.wait(l, t), 30*time.Second, + m.workload(t, mt.tn, 2).addr) + } m.isolation(t, "after-recreation") } diff --git a/tests/edgeconformance/masqueprovider/config.go b/tests/edgeconformance/masqueprovider/config.go index d539a566..909edd64 100644 --- a/tests/edgeconformance/masqueprovider/config.go +++ b/tests/edgeconformance/masqueprovider/config.go @@ -2,15 +2,23 @@ // // SPDX-License-Identifier: AGPL-3.0-or-later +//go:build linux + package main import ( + "context" + "crypto/subtle" "errors" "fmt" "net/netip" "os" + "time" "sigs.k8s.io/yaml" + + "go.datum.net/galactic/internal/edge/connectip" + "go.datum.net/galactic/internal/edge/masque" ) // Config is the fixture's configuration file. @@ -29,12 +37,14 @@ type Config struct { } // Service is one EdgeService grant. A provider cannot read an EdgeService, -// so the deployer copies its UID and its VPC's UID from the grant. +// so the deployer copies its UID, its VPC's UID and its VPC's base62 +// identifier from the grant. type Service struct { Namespace string `json:"namespace"` Name string `json:"name"` UID string `json:"uid"` VPCUID string `json:"vpcUID"` + VPCID string `json:"vpcID"` // VPC is the Galactic-VPC header value that selects this service. VPC string `json:"vpc"` } @@ -79,8 +89,8 @@ func (c *Config) Validate() error { } services := map[string]bool{} for _, s := range c.Services { - if s.Namespace == "" || s.Name == "" || s.UID == "" || s.VPCUID == "" || s.VPC == "" { - return fmt.Errorf("service %q: namespace, name, uid, vpcUID and vpc are required", s.Name) + if s.Namespace == "" || s.Name == "" || s.UID == "" || s.VPCUID == "" || s.VPCID == "" || s.VPC == "" { + return fmt.Errorf("service %q: namespace, name, uid, vpcUID, vpcID and vpc are required", s.Name) } if services[s.Key()] { return fmt.Errorf("service %q is listed twice", s.Key()) @@ -113,3 +123,37 @@ func (c *Config) Validate() error { } return nil } + +// grantTTL is how long a fixture grant authorizes forwarding after each +// request. The fixture's credentials are test credentials with no expiry of +// their own. +const grantTTL = time.Hour + +// Authorize implements masque.Authorizer over the configured connectors: +// each is a static bearer credential bound to one service's VPC and one +// /96. +func (c *Config) Authorize(_ context.Context, credential, vpc string) (masque.Grant, error) { + var found *Connector + for i := range c.Connectors { + if subtle.ConstantTimeCompare([]byte(c.Connectors[i].Credential), []byte(credential)) == 1 { + found = &c.Connectors[i] + } + } + if found == nil { + return masque.Grant{}, connectip.ErrUnauthenticated + } + var svc Service + for _, s := range c.Services { + if s.Key() == found.Service { + svc = s + } + } + if svc.VPC != vpc { + return masque.Grant{}, connectip.ErrForbidden + } + return masque.Grant{ + Connector: found.ID, Principal: found.ID, VPC: svc.VPC, VPCUID: svc.VPCUID, VPCID: svc.VPCID, + Service: masque.ServiceRef{Namespace: svc.Namespace, Name: svc.Name, UID: svc.UID}, + Prefix: found.Prefix, Routes: found.Routes, NotAfter: time.Now().Add(grantTTL), + }, nil +} diff --git a/tests/edgeconformance/masqueprovider/main.go b/tests/edgeconformance/masqueprovider/main.go index 1067f69c..73d3ff06 100644 --- a/tests/edgeconformance/masqueprovider/main.go +++ b/tests/edgeconformance/masqueprovider/main.go @@ -6,26 +6,23 @@ // Command masqueprovider is the Galactic Edge conformance suite's MASQUE // fixture: a CONNECT-IP (RFC 9484) server over HTTP/3 that runs as an -// unprivileged Galactic Edge provider. It uses only the provider contract -// and its helpers in internal/edge/provider: it holds its EdgeInstance's -// lease, requests one PacketAccess attachment per configured EdgeService, -// claims one EdgeRoute per connector /96, and carries each admitted -// connector's packets through internal/edge/connectip's TUN backend on the -// packet device galactic-edge creates in its pod. It needs no host -// namespace, hostPath, NET_ADMIN or BPF. +// unprivileged Galactic Edge provider through internal/edge/masque, the +// MASQUE provider adapter. The adapter holds the pod's EdgeInstance lease, +// requests the PacketAccess attachment of each admitted connector's +// EdgeService, claims the connector's /96 EdgeRoute, and carries its +// packets on the packet device galactic-edge creates in the pod. It needs +// no host namespace, hostPath, NET_ADMIN or BPF. // -// Admission is a test fixture, not a product authority: each connector is a -// static bearer credential bound to one service's VPC and one /96, admitted -// only while the instance holds its lease and core has accepted the -// connector's route on a programmed attachment. Expiry or loss of any of -// them ends the connector's session. +// Authorization is a test fixture, not a product authority: each connector +// is a static bearer credential bound to one service's VPC and one /96. // -// SIGTERM ends every session, then releases the routes, deletes the -// attachments and releases the lease. +// SIGTERM stops the server, which ends every session, then the adapter +// releases the routes, deletes the attachments and releases the lease. package main import ( "context" + "crypto/rand" "crypto/tls" "errors" "flag" @@ -38,7 +35,6 @@ import ( "syscall" "time" - "golang.org/x/sys/unix" coordinationv1 "k8s.io/api/coordination/v1" "k8s.io/apimachinery/pkg/runtime" ctrl "sigs.k8s.io/controller-runtime" @@ -47,6 +43,7 @@ import ( galacticv1alpha1 "go.datum.net/galactic/api/v1alpha1" "go.datum.net/galactic/internal/edge/claim" "go.datum.net/galactic/internal/edge/connectip" + "go.datum.net/galactic/internal/edge/masque" "go.datum.net/galactic/internal/edge/provider" "go.datum.net/galactic/internal/edge/timing" ) @@ -94,24 +91,27 @@ func run() error { return err } - p := &Provider{ - Client: c, Log: log.With("pod", pod.Name), Config: cfg, Pod: pod, Clock: timing.BootClock, - Renew: timing.RenewInterval, OpenTUN: openTUN, - Claims: &claim.Instance{ - ID: newHolderID(), Clock: timing.BootClock, Params: timing.Params, - Store: provider.InstanceStore{Client: c, Namespace: pod.Namespace, Duration: timing.Params.Duration}, - }, + claims := &claim.Instance{ + ID: rand.Text(), Clock: timing.BootClock, Params: timing.Params, + Store: provider.InstanceStore{Client: c, Namespace: pod.Namespace, Duration: timing.Params.Duration}, + } + a, err := masque.New(masque.Config{ + Client: c, Provider: cfg.Provider, Pod: pod, Authorizer: cfg, Claims: claims, + OpenDevice: masque.OpenTUN, Log: log.With("pod", pod.Name), + }) + if err != nil { + return err } ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGTERM, os.Interrupt) defer stop() - if err := p.Start(ctx); err != nil { + if err := a.Start(ctx); err != nil { return err } srv, err := connectip.NewServer(connectip.Config{ Authority: cfg.Authority, TLSConfig: &tls.Config{Certificates: []tls.Certificate{cert}, MinVersion: tls.VersionTLS13}, - Admission: p, Backend: p, Logger: log, + Admission: a, Backend: a, Logger: log, }) if err != nil { return err @@ -122,34 +122,34 @@ func run() error { } serveErr := make(chan error, 1) go func() { serveErr <- srv.Serve(pc) }() + runCtx, cancelRun := context.WithCancel(ctx) + runErr := make(chan error, 1) + go func() { runErr <- a.Run(runCtx) }() - t := time.NewTicker(time.Second) - defer t.Stop() - for { - if err := p.Reconcile(ctx); err != nil { - _ = srv.Close() - return err - } - select { - case <-ctx.Done(): - // Sessions end before the routes are released. - _ = srv.Close() - sctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) - defer cancel() - return p.Shutdown(sctx) - case err := <-serveErr: - sctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) - defer cancel() - return errors.Join(fmt.Errorf("HTTP/3 server stopped: %w", err), p.Shutdown(sctx)) - case <-t.C: - } - } + var stopped error + select { + case <-ctx.Done(): + case err := <-serveErr: + stopped = fmt.Errorf("HTTP/3 server stopped: %w", err) + case err := <-runErr: + // Run ends early only when the instance's lease can never be held. + runErr <- err + } + // Sessions end before the routes, attachments and lease are released. + _ = srv.Close() + cancelRun() + if err := <-runErr; err != nil { + stopped = errors.Join(stopped, fmt.Errorf("adapter stopped: %w", err)) + } + sctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) + defer cancel() + return errors.Join(stopped, a.Shutdown(sctx)) } // podFromEnv reads the pod's identity from the downward API variables the // Deployment sets, and the node's boot ID. -func podFromEnv() (Pod, error) { - p := Pod{ +func podFromEnv() (masque.Pod, error) { + p := masque.Pod{ Namespace: os.Getenv("POD_NAMESPACE"), Name: os.Getenv("POD_NAME"), UID: os.Getenv("POD_UID"), Node: os.Getenv("NODE_NAME"), RunAsUser: int64(os.Getuid()), RunAsGroup: int64(os.Getgid()), } @@ -163,24 +163,3 @@ func podFromEnv() (Pod, error) { p.BootID = strings.TrimSpace(string(b)) return p, nil } - -// openTUN attaches to the existing TUN device name as its owner, which -// needs no capability, and returns a nonblocking descriptor without -// packet-information headers. -func openTUN(name string) (*os.File, error) { - fd, err := unix.Open("/dev/net/tun", unix.O_RDWR|unix.O_NONBLOCK|unix.O_CLOEXEC, 0) - if err != nil { - return nil, fmt.Errorf("open /dev/net/tun: %w", err) - } - ifr, err := unix.NewIfreq(name) - if err != nil { - _ = unix.Close(fd) - return nil, err - } - ifr.SetUint16(unix.IFF_TUN | unix.IFF_NO_PI) - if err := unix.IoctlIfreq(fd, unix.TUNSETIFF, ifr); err != nil { - _ = unix.Close(fd) - return nil, fmt.Errorf("attach to TUN %s: %w", name, err) - } - return os.NewFile(uintptr(fd), name), nil -} diff --git a/tests/edgeconformance/masqueprovider/provider.go b/tests/edgeconformance/masqueprovider/provider.go deleted file mode 100644 index bdba1604..00000000 --- a/tests/edgeconformance/masqueprovider/provider.go +++ /dev/null @@ -1,524 +0,0 @@ -// Copyright 2026 Datum Cloud, Inc. -// -// SPDX-License-Identifier: AGPL-3.0-or-later - -//go:build linux - -package main - -import ( - "context" - "crypto/rand" - "crypto/subtle" - "errors" - "fmt" - "log/slog" - "net/netip" - "os" - "sync" - "sync/atomic" - "time" - - apierrors "k8s.io/apimachinery/pkg/api/errors" - "k8s.io/apimachinery/pkg/api/meta" - metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" - "sigs.k8s.io/controller-runtime/pkg/client" - - galacticv1alpha1 "go.datum.net/galactic/api/v1alpha1" - "go.datum.net/galactic/internal/edge/claim" - "go.datum.net/galactic/internal/edge/connectip" - "go.datum.net/galactic/internal/edge/provider" -) - -// Pod identifies the pod this process runs in. -type Pod struct { - Namespace, Name, UID, Node, BootID string - // RunAsUser and RunAsGroup own the packet devices this process opens. - RunAsUser, RunAsGroup int64 -} - -// Provider is one fixture instance: it holds its EdgeInstance's lease, one -// PacketAccess attachment per service and one EdgeRoute per connector /96, -// and admits a connector's CONNECT-IP request only while all three are -// current. Its Client must read from the API server. -type Provider struct { - Client client.Client - Log *slog.Logger - Config *Config - Pod Pod - Claims *claim.Instance - Clock claim.Clock - // Renew is how often the lease is renewed. - Renew time.Duration - // OpenTUN opens an attachment's packet device as a nonblocking - // descriptor. - OpenTUN func(name string) (*os.File, error) - - instance *galacticv1alpha1.EdgeInstance - - mu sync.Mutex - lease *claim.Lease - services map[string]*service - connectors []*connector - // deadline is the held lease's deadline on Clock, or zero; every packet - // reads it, so forwarding stops at the deadline even while the - // reconcile loop waits on the API. - deadline atomic.Int64 - sessionID atomic.Uint64 -} - -type service struct { - cfg Service - attachment *galacticv1alpha1.EdgeAttachment - // programmed identifies the host state the backend was opened on. - programmed string - backend *connectip.TUNBackend -} - -type connector struct { - cfg Connector - svc *service - route string - epoch int64 - active bool - grants map[*grant]struct{} - logged string -} - -// forwarding reports whether the held lease allows packets now. -func (p *Provider) forwarding() bool { return int64(p.Clock()) < p.deadline.Load() } - -func (p *Provider) setLease(l *claim.Lease) { - p.lease = l - if l == nil { - p.deadline.Store(0) - return - } - p.deadline.Store(int64(l.Deadline)) -} - -// Start registers the pod's EdgeInstance and sets up the service and -// connector state. -func (p *Provider) Start(ctx context.Context) error { - inst, err := provider.CreateInstance(ctx, p.Client, provider.NewInstance(p.Pod.Namespace, p.Config.Provider, - p.Pod.Name, p.Pod.UID, p.Pod.Node, p.Pod.BootID)) - if err != nil { - return fmt.Errorf("register EdgeInstance: %w", err) - } - p.instance = inst - p.services = map[string]*service{} - for _, s := range p.Config.Services { - p.services[s.Key()] = &service{cfg: s} - } - for _, c := range p.Config.Connectors { - svc := p.services[c.Service] - p.connectors = append(p.connectors, &connector{cfg: c, svc: svc, - route: provider.RouteName(svc.cfg.VPCUID, c.Prefix), grants: map[*grant]struct{}{}}) - } - p.Log.Info("instance registered", "instance", inst.Name) - return nil -} - -// Reconcile makes one pass: lease, then each attachment, then each route. -// It returns an error only when retrying cannot succeed. -func (p *Provider) Reconcile(ctx context.Context) error { - if err := p.reconcileLease(ctx); err != nil { - return err - } - for _, s := range p.Config.Services { - if err := p.reconcileAttachment(ctx, p.services[s.Key()]); err != nil { - if errors.Is(err, provider.ErrNameTaken) { - return err - } - p.Log.Warn("attachment reconcile failed", "service", s.Name, "error", err) - } - } - for _, c := range p.connectors { - p.reconcileRoute(ctx, c) - } - return nil -} - -func (p *Provider) reconcileLease(ctx context.Context) error { - p.mu.Lock() - l := p.lease - p.mu.Unlock() - now := p.Clock() - if l != nil && !l.Valid(now) { - p.Log.Warn("lease expired", "epoch", l.Epoch) - p.loseLease() - l = nil - } - if l == nil { - got, err := p.Claims.Acquire(ctx, p.instance.Name) - switch { - case errors.Is(err, provider.ErrNameTaken): - return err - case err != nil: - p.Log.Warn("lease not yet acquired", "reason", err) - return nil - } - p.mu.Lock() - p.setLease(&got) - p.mu.Unlock() - p.Log.Info("lease acquired", "epoch", got.Epoch) - return nil - } - // A lease's deadline is its renewal's start plus the claim duration. - if now < l.Deadline-p.Claims.Params.Duration+p.Renew { - return nil - } - got, err := p.Claims.Renew(ctx, *l) - switch { - case errors.Is(err, claim.ErrLost): - p.Log.Warn("lease lost", "epoch", l.Epoch, "reason", err) - p.loseLease() - case err != nil: - p.Log.Warn("lease renewal failed; the lease holds until its deadline", "error", err) - default: - p.mu.Lock() - p.setLease(&got) - p.mu.Unlock() - } - return nil -} - -// loseLease ends every session: every route claim was made under the lost -// epoch. -func (p *Provider) loseLease() { - p.mu.Lock() - defer p.mu.Unlock() - p.setLease(nil) - for _, c := range p.connectors { - p.deactivate(c, "lease lost") - } -} - -func (p *Provider) reconcileAttachment(ctx context.Context, s *service) error { - name := provider.AttachmentName(s.cfg.UID, p.Pod.UID) - var att galacticv1alpha1.EdgeAttachment - err := p.Client.Get(ctx, client.ObjectKey{Namespace: p.Pod.Namespace, Name: name}, &att) - switch { - case apierrors.IsNotFound(err): - p.stopServing(s, "attachment missing") - created, err := provider.CreateAttachment(ctx, p.Client, p.attachmentRequest(s, name)) - if err != nil { - return fmt.Errorf("request attachment %s: %w", name, err) - } - p.Log.Info("attachment requested", "attachment", created.Name, "service", s.cfg.Name) - return nil - case err != nil: - return err - } - if att.Spec.PodUID != p.Pod.UID || att.Spec.ServiceRef.UID != s.cfg.UID { - return fmt.Errorf("%w: EdgeAttachment %s", provider.ErrNameTaken, name) - } - // Admit reads the stored copy while setReady writes att. - p.mu.Lock() - s.attachment = att.DeepCopy() - p.mu.Unlock() - if !meta.IsStatusConditionTrue(att.Status.Conditions, galacticv1alpha1.EdgeAttachmentProgrammed) || - att.DeletionTimestamp != nil { - p.stopServing(s, "attachment not programmed") - return p.setReady(ctx, &att, false, "NotProgrammed", "waiting for galactic-edge to program the attachment") - } - key := att.Status.HostLink + "/" + att.Status.PacketDevice + "/" + att.Status.Address - p.mu.Lock() - open := s.backend != nil && s.programmed == key - p.mu.Unlock() - if !open { - p.stopServing(s, "attachment reprogrammed") - if err := p.openBackend(s, att.Status.PacketDevice, key); err != nil { - return p.setReady(ctx, &att, false, "DataPlaneFailed", err.Error()) - } - p.Log.Info("serving attachment", "attachment", name, "device", att.Status.PacketDevice) - } - return p.setReady(ctx, &att, true, "Serving", "the provider is serving the attachment") -} - -func (p *Provider) attachmentRequest(s *service, name string) *galacticv1alpha1.EdgeAttachment { - return &galacticv1alpha1.EdgeAttachment{ - ObjectMeta: metav1.ObjectMeta{Namespace: p.Pod.Namespace, Name: name}, - Spec: galacticv1alpha1.EdgeAttachmentSpec{ - ProviderRef: galacticv1alpha1.EdgeProviderReference{Name: p.Config.Provider}, - ServiceRef: galacticv1alpha1.EdgeServiceReference{Namespace: s.cfg.Namespace, Name: s.cfg.Name, - UID: s.cfg.UID}, - InstanceRef: galacticv1alpha1.EdgeInstanceReference{Name: p.instance.Name}, - PodUID: p.Pod.UID, - ContractVersion: "v1alpha1", - Capabilities: []galacticv1alpha1.EdgeCapability{{Name: "PacketAccess", Version: "v1"}}, - PacketDevice: &galacticv1alpha1.EdgePacketDevice{Owner: p.Pod.RunAsUser, Group: p.Pod.RunAsGroup}, - }, - } -} - -func (p *Provider) openBackend(s *service, device, key string) error { - if device == "" { - return errors.New("the attachment has no packet device") - } - f, err := p.OpenTUN(device) - if err != nil { - return err - } - b, err := connectip.NewTUNBackend(f, s.cfg.VPC) - if err != nil { - _ = f.Close() - return err - } - p.mu.Lock() - s.backend, s.programmed = b, key - p.mu.Unlock() - return nil -} - -// stopServing ends s's sessions and closes its backend. -func (p *Provider) stopServing(s *service, reason string) { - p.mu.Lock() - b := s.backend - s.backend, s.programmed = nil, "" - for _, c := range p.connectors { - if c.svc == s { - p.deactivate(c, reason) - } - } - p.mu.Unlock() - if b != nil { - _ = b.Close() - } -} - -// setReady writes the attachment's Ready condition, the only one a -// provider sets, when it differs. -func (p *Provider) setReady(ctx context.Context, att *galacticv1alpha1.EdgeAttachment, ready bool, reason, - msg string) error { - status := metav1.ConditionFalse - if ready { - status = metav1.ConditionTrue - } - if c := meta.FindStatusCondition(att.Status.Conditions, galacticv1alpha1.EdgeAttachmentReady); c != nil && - c.Status == status && c.Reason == reason && c.ObservedGeneration == att.Generation { - return nil - } - meta.SetStatusCondition(&att.Status.Conditions, metav1.Condition{Type: galacticv1alpha1.EdgeAttachmentReady, - Status: status, Reason: reason, Message: msg, ObservedGeneration: att.Generation}) - return p.Client.Status().Update(ctx, att) -} - -// reconcileRoute claims c's /96 while the lease is held and c's backend is -// open, and activates c once core accepts the claim. -func (p *Provider) reconcileRoute(ctx context.Context, c *connector) { - p.mu.Lock() - l, att, open := p.lease, c.svc.attachment, c.svc.backend != nil - p.mu.Unlock() - if l == nil || att == nil || !open { - return - } - var r galacticv1alpha1.EdgeRoute - err := p.Client.Get(ctx, client.ObjectKey{Namespace: p.Pod.Namespace, Name: c.route}, &r) - if apierrors.IsNotFound(err) { - r = *provider.NewRoute(p.Pod.Namespace, p.Config.Provider, c.svc.cfg.VPCUID, c.cfg.Prefix) - r.Spec.Claim = &galacticv1alpha1.EdgeRouteClaim{Attachment: att.Name, HolderPodUID: p.Pod.UID, - HolderInstance: l.Key, InstanceEpoch: int64(l.Epoch), Epoch: 1} - err = p.Client.Create(ctx, &r) - } else if err == nil { - err = provider.Claim(ctx, p.Client, &r, att.Name, p.Pod.UID, *l) - } - if err != nil { - if !apierrors.IsConflict(err) && !apierrors.IsAlreadyExists(err) { - p.Log.Warn("route claim failed", "connector", c.cfg.ID, "error", err) - } - return - } - accepted := acceptedFor(&r, att.Name, l) - p.mu.Lock() - defer p.mu.Unlock() - if p.lease == nil || p.lease.Epoch != l.Epoch || c.svc.attachment == nil || c.svc.attachment.Name != att.Name { - return - } - switch { - case accepted && (!c.active || c.epoch != r.Spec.Claim.Epoch): - p.deactivate(c, "claim epoch changed") - c.active, c.epoch = true, r.Spec.Claim.Epoch - case !accepted: - p.deactivate(c, "route not accepted") - } - if state := fmt.Sprintf("accepted=%t epoch=%d", accepted, r.Spec.Claim.Epoch); state != c.logged { - c.logged = state - p.Log.Info("route", "connector", c.cfg.ID, "route", r.Name, "prefix", c.cfg.Prefix, "state", state) - } -} - -// acceptedFor reports whether core accepted the route's current claim for -// attachment under l and has not revoked or closed it. -func acceptedFor(r *galacticv1alpha1.EdgeRoute, attachment string, l *claim.Lease) bool { - a, cl := r.Status.Accepted, r.Spec.Claim - return a != nil && cl != nil && !cl.Released && a.Epoch == cl.Epoch && a.Attachment == attachment && - a.Instance == l.Key && a.InstanceEpoch == int64(l.Epoch) && r.Status.ClosedEpoch < cl.Epoch && - meta.IsStatusConditionTrue(r.Status.Conditions, galacticv1alpha1.EdgeRouteAccepted) -} - -// deactivate ends every session of c. The caller holds p.mu. -func (p *Provider) deactivate(c *connector, reason string) { - if c.active || len(c.grants) > 0 { - p.Log.Info("connector inactive", "connector", c.cfg.ID, "reason", reason) - } - c.active, c.epoch = false, 0 - for g := range c.grants { - g.end() - delete(c.grants, g) - } -} - -// Shutdown ends every session, then releases the routes, deletes the -// attachments and releases the lease, in that order, so a replacement takes -// nothing over while this instance still forwards. -func (p *Provider) Shutdown(ctx context.Context) error { - p.Log.Info("shutting down") - p.mu.Lock() - l := p.lease - p.setLease(nil) - p.mu.Unlock() - for _, s := range p.services { - p.stopServing(s, "shutting down") - } - var errs []error - if l != nil { - for _, c := range p.connectors { - if err := p.releaseRoute(ctx, c, *l); err != nil { - errs = append(errs, fmt.Errorf("release route of %s: %w", c.cfg.ID, err)) - } - } - } - for _, s := range p.services { - if s.attachment == nil { - continue - } - if err := p.Client.Delete(ctx, s.attachment); client.IgnoreNotFound(err) != nil { - errs = append(errs, fmt.Errorf("delete attachment %s: %w", s.attachment.Name, err)) - } - } - if l != nil { - if err := p.Claims.Release(ctx, *l); err != nil { - errs = append(errs, fmt.Errorf("release lease: %w", err)) - } - } - return errors.Join(errs...) -} - -func (p *Provider) releaseRoute(ctx context.Context, c *connector, l claim.Lease) error { - for range 5 { - var r galacticv1alpha1.EdgeRoute - err := p.Client.Get(ctx, client.ObjectKey{Namespace: p.Pod.Namespace, Name: c.route}, &r) - if err != nil { - return client.IgnoreNotFound(err) - } - err = provider.Release(ctx, p.Client, &r, l) - if errors.Is(err, provider.ErrNotHolder) { - return nil - } - if !apierrors.IsConflict(err) { - return err - } - } - return errors.New("route kept changing") -} - -// Admit implements connectip.Admission for the configured connectors. -func (p *Provider) Admit(_ context.Context, credential, vpc string) (connectip.Lease, error) { - var c *connector - for _, cand := range p.connectors { - if subtle.ConstantTimeCompare([]byte(cand.cfg.Credential), []byte(credential)) == 1 { - c = cand - } - } - if c == nil { - return nil, connectip.ErrUnauthenticated - } - if c.svc.cfg.VPC != vpc { - return nil, connectip.ErrForbidden - } - p.mu.Lock() - defer p.mu.Unlock() - if !c.active || p.lease == nil || c.svc.backend == nil || c.svc.attachment == nil || !p.forwarding() { - return nil, fmt.Errorf("connector %s has no accepted route on a programmed attachment", c.cfg.ID) - } - g := &grant{ - p: p, c: c, done: make(chan struct{}), - info: connectip.SessionInfo{ - ID: p.sessionID.Add(1), Principal: c.cfg.ID, VPC: vpc, Prefix: c.cfg.Prefix, - Connector: c.cfg.ID, Attachment: c.svc.attachment.Name, Claim: c.route, Epoch: uint64(c.epoch), - }, - } - c.grants[g] = struct{}{} - return g, nil -} - -// grant is one admitted request's lease. -type grant struct { - p *Provider - c *connector - info connectip.SessionInfo - done chan struct{} - endOnce sync.Once -} - -func (g *grant) Info() connectip.SessionInfo { return g.info } -func (g *grant) Routes() []netip.Prefix { return g.c.cfg.Routes } -func (g *grant) Done() <-chan struct{} { return g.done } -func (g *grant) end() { g.endOnce.Do(func() { close(g.done) }) } - -func (g *grant) Close() error { - g.end() - g.p.mu.Lock() - delete(g.c.grants, g) - g.p.mu.Unlock() - return nil -} - -// Open implements connectip.PacketBackend: it opens the session on the -// backend of the attachment admission named, and drops its packets once -// the lease's deadline passes. -func (p *Provider) Open(ctx context.Context, info connectip.SessionInfo) (connectip.BackendSession, error) { - p.mu.Lock() - var b *connectip.TUNBackend - for _, s := range p.services { - if s.attachment != nil && s.attachment.Name == info.Attachment { - b = s.backend - } - } - p.mu.Unlock() - if b == nil { - return nil, fmt.Errorf("attachment %s is not being served", info.Attachment) - } - s, err := b.Open(ctx, info) - if err != nil { - return nil, err - } - return &gatedSession{BackendSession: s, allow: p.forwarding}, nil -} - -var errExpired = errors.New("the instance's lease has expired") - -// gatedSession forwards packets only while allow reports true. -type gatedSession struct { - connectip.BackendSession - allow func() bool -} - -func (s *gatedSession) WritePacket(pkt []byte) error { - if !s.allow() { - return errExpired - } - return s.BackendSession.WritePacket(pkt) -} - -func (s *gatedSession) ReadPacket(ctx context.Context, buf []byte) (int, error) { - n, err := s.BackendSession.ReadPacket(ctx, buf) - if err == nil && !s.allow() { - return 0, errExpired - } - return n, err -} - -// newHolderID returns a unique holder identity for this process. -func newHolderID() string { return rand.Text() } diff --git a/tests/edgeconformance/masqueprovider/provider_test.go b/tests/edgeconformance/masqueprovider/provider_test.go index 61dd44b0..0999f5e7 100644 --- a/tests/edgeconformance/masqueprovider/provider_test.go +++ b/tests/edgeconformance/masqueprovider/provider_test.go @@ -9,28 +9,27 @@ package main import ( "context" "errors" - "io" - "log/slog" "net/netip" "strings" "testing" "time" - metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" - - galacticv1alpha1 "go.datum.net/galactic/api/v1alpha1" - "go.datum.net/galactic/internal/edge/claim" "go.datum.net/galactic/internal/edge/connectip" + "go.datum.net/galactic/internal/edge/masque" ) -const notA96 = "/96" +const ( + notA96 = "/96" + svcName = "svc" + required = "required" +) func validConfig() *Config { return &Config{ Provider: "masque-fixture", Authority: "masque.test:8443", Listen: ":8443", Services: []Service{ - {Namespace: "ns10", Name: "svc", UID: "s10", VPCUID: "v10", VPC: "10"}, - {Namespace: "ns20", Name: "svc", UID: "s20", VPCUID: "v20", VPC: "20"}, + {Namespace: "ns10", Name: svcName, UID: "s10", VPCUID: "v10", VPCID: "000000v10", VPC: "10"}, + {Namespace: "ns20", Name: svcName, UID: "s20", VPCUID: "v20", VPCID: "000000v20", VPC: "20"}, }, Connectors: []Connector{ {ID: "c10", Credential: "t10", Service: "ns10/svc", Prefix: netip.MustParsePrefix("fd00:1::/96")}, @@ -48,8 +47,9 @@ func TestConfigValidate(t *testing.T) { mutate func(*Config) want string }{ - {"no authority", func(c *Config) { c.Authority = "" }, "required"}, - {"service without VPC", func(c *Config) { c.Services[0].VPC = "" }, "required"}, + {"no authority", func(c *Config) { c.Authority = "" }, required}, + {"service without VPC", func(c *Config) { c.Services[0].VPC = "" }, required}, + {"service without VPC ID", func(c *Config) { c.Services[0].VPCID = "" }, required}, {"duplicate service", func(c *Config) { c.Services[1].Namespace = "ns10" }, "twice"}, {"shared credential", func(c *Config) { c.Connectors[1].Credential = "t10" }, "not unique"}, {"unknown service", func(c *Config) { c.Connectors[0].Service = "ns30/svc" }, "unknown service"}, @@ -71,126 +71,21 @@ func TestConfigValidate(t *testing.T) { } } -// fixture returns a provider with both connectors' services served and the -// lease held until deadline on a clock that reads now. -func fixture(t *testing.T, now *time.Duration, deadline time.Duration) *Provider { - t.Helper() - p := &Provider{Log: slog.New(slog.NewTextHandler(io.Discard, nil)), Config: validConfig(), - Clock: func() time.Duration { return *now }} - p.services = map[string]*service{} - for _, s := range p.Config.Services { - p.services[s.Key()] = &service{cfg: s, - attachment: &galacticv1alpha1.EdgeAttachment{ObjectMeta: metav1.ObjectMeta{Name: "ea-" + s.Namespace}}} - } - for _, c := range p.Config.Connectors { - p.connectors = append(p.connectors, &connector{cfg: c, svc: p.services[c.Service], route: "er-" + c.ID, - grants: map[*grant]struct{}{}}) - } - p.setLease(&claim.Lease{Key: "ei-pod", Epoch: 1, Deadline: deadline}) - return p -} - -// serve marks every service's backend open and every connector's route -// accepted at epoch 3. -func serve(p *Provider) { - for _, s := range p.services { - s.backend = &connectip.TUNBackend{} - } - for _, c := range p.connectors { - c.active, c.epoch = true, 3 - } -} - -func TestAdmit(t *testing.T) { - now := time.Second - p := fixture(t, &now, time.Minute) - ctx := context.Background() - - if _, err := p.Admit(ctx, "unknown", "10"); !errors.Is(err, connectip.ErrUnauthenticated) { - t.Errorf("unknown credential: %v, want ErrUnauthenticated", err) - } - if _, err := p.Admit(ctx, "t10", "20"); !errors.Is(err, connectip.ErrForbidden) { - t.Errorf("VPC 10's credential for VPC 20: %v, want ErrForbidden", err) - } - if _, err := p.Admit(ctx, "t10", "10"); err == nil || errors.Is(err, connectip.ErrForbidden) || - errors.Is(err, connectip.ErrUnauthenticated) { - t.Errorf("connector without an accepted route: %v, want an unavailable error", err) - } - - serve(p) - l, err := p.Admit(ctx, "t20", "20") - if err != nil { - t.Fatal(err) - } - want := connectip.SessionInfo{ID: 1, Principal: "c20", VPC: "20", Prefix: netip.MustParsePrefix("fd00:1::/96"), - Connector: p.connectors[1].cfg.ID, Attachment: "ea-ns20", Claim: "er-c20", Epoch: 3} - if got := l.Info(); got != want { - t.Errorf("Info() = %+v, want %+v", got, want) - } - - // Losing the route ends that connector's session and no other. - other, err := p.Admit(ctx, "t10", "10") +func TestAuthorize(t *testing.T) { + c := validConfig() + g, err := c.Authorize(context.Background(), "t20", "20") if err != nil { t.Fatal(err) } - p.mu.Lock() - p.deactivate(p.connectors[1], "test") - p.mu.Unlock() - select { - case <-l.Done(): - default: - t.Error("VPC 20's session survived its route") - } - select { - case <-other.Done(): - t.Error("VPC 10's session ended with VPC 20's route") - default: - } - if _, err := p.Admit(ctx, "t20", "20"); err == nil { - t.Error("admitted a connector whose route is no longer accepted") - } - _ = other.Close() - if len(p.connectors[0].grants) != 0 { - t.Error("Close left the grant registered") - } - - // Past the lease's deadline nothing is admitted. - serve(p) - now = 2 * time.Minute - if _, err := p.Admit(ctx, "t10", "10"); err == nil { - t.Error("admitted past the lease's deadline") - } -} - -type fakeSession struct{ writes, reads int } - -func (s *fakeSession) WritePacket([]byte) error { s.writes++; return nil } -func (s *fakeSession) ReadPacket(context.Context, []byte) (int, error) { - s.reads++ - return 40, nil -} -func (s *fakeSession) Close() error { return nil } - -func TestGatedSessionStopsAtDeadline(t *testing.T) { - now := time.Second - p := fixture(t, &now, 10*time.Second) - inner := &fakeSession{} - s := &gatedSession{BackendSession: inner, allow: p.forwarding} - buf := make([]byte, 64) - if err := s.WritePacket(buf); err != nil { - t.Fatal(err) - } - if n, err := s.ReadPacket(context.Background(), buf); err != nil || n != 40 { - t.Fatalf("ReadPacket = %d, %v", n, err) - } - now = 10 * time.Second - if err := s.WritePacket(buf); !errors.Is(err, errExpired) { - t.Errorf("write at the deadline: %v, want errExpired", err) + if g.Connector != "c20" || g.VPC != "20" || g.VPCUID != "v20" || g.VPCID != "000000v20" || + g.Service != (masque.ServiceRef{Namespace: "ns20", Name: svcName, UID: "s20"}) || + g.Prefix != netip.MustParsePrefix("fd00:1::/96") || !g.NotAfter.After(time.Now()) { + t.Fatalf("grant %+v does not match connector c20", g) } - if _, err := s.ReadPacket(context.Background(), buf); !errors.Is(err, errExpired) { - t.Errorf("read at the deadline: %v, want errExpired", err) + if _, err := c.Authorize(context.Background(), "nope", "10"); !errors.Is(err, connectip.ErrUnauthenticated) { + t.Fatalf("unknown credential: %v", err) } - if inner.writes != 1 { - t.Errorf("%d writes reached the device, want 1", inner.writes) + if _, err := c.Authorize(context.Background(), "t10", "20"); !errors.Is(err, connectip.ErrForbidden) { + t.Fatalf("credential for another VPC: %v", err) } }