diff --git a/cmd/galactic-router/root.go b/cmd/galactic-router/root.go index 5d2876b0..d2b02409 100644 --- a/cmd/galactic-router/root.go +++ b/cmd/galactic-router/root.go @@ -281,7 +281,8 @@ func runCmd(cfg *config.RouterConfig) error { // controller process before informer reconciliation starts. A missing map is // tolerated because the datapath loader may start after galactic-router; // the first Apply retries initialization. - if err := setupServiceRouteController(mgr, nodeName, cfg.GCNamespace, serviceRouteMetrics); err != nil { + if err := setupServiceRouteController(mgr, nodeName, cfg.GCNamespace, cfg.ServiceFrontendEnabled, + serviceRouteMetrics); err != nil { return fmt.Errorf("setup ServiceRoutePolicy controller: %w", err) } @@ -351,7 +352,7 @@ func runCmd(cfg *config.RouterConfig) error { func setupServiceRouteController( mgr ctrl.Manager, - nodeName, bgpNamespace string, + nodeName, bgpNamespace string, frontendEnabled bool, serviceRouteMetrics *serviceroute.Metrics, ) error { programmer := &serviceroute.EBPFRouteProgrammer{} @@ -360,13 +361,14 @@ func setupServiceRouteController( } return (&controller.ServiceRoutePolicyReconciler{ - Client: mgr.GetClient(), - Scheme: mgr.GetScheme(), - NodeName: nodeName, - BGPNamespace: bgpNamespace, - Programmer: programmer, - Metrics: serviceRouteMetrics, - Applied: make(map[types.NamespacedName]map[types.NamespacedName]serviceroute.RouteIntent), + Client: mgr.GetClient(), + Scheme: mgr.GetScheme(), + NodeName: nodeName, + BGPNamespace: bgpNamespace, + Programmer: programmer, + Metrics: serviceRouteMetrics, + FrontendEnabled: frontendEnabled, + Applied: make(map[types.NamespacedName]map[types.NamespacedName]serviceroute.RouteIntent), }).SetupWithManager(mgr) } @@ -415,6 +417,8 @@ func newRootCommand() *cobra.Command { cmd.Flags().DurationP("gc-interval", "", config.DefaultRouterGCInterval, "Cleanup interval") + cmd.Flags().Bool("service-frontend-enabled", false, + "Enable consumer-facing private-service address translation") cmd.Flags().Bool("webhook-enabled", false, "Enable the NetworkRule admission webhook (requires TLS cert material; see config/webhook/)") cmd.Flags().IntP("webhook-port", "", diff --git a/config/galactic-router/rbac.yaml b/config/galactic-router/rbac.yaml index a0286d84..656429ac 100644 --- a/config/galactic-router/rbac.yaml +++ b/config/galactic-router/rbac.yaml @@ -50,6 +50,7 @@ rules: verbs: ["get", "update", "patch"] - apiGroups: ["cloud.datumapis.com"] resources: + - vpcs - vpcattachments verbs: ["get", "list", "watch"] - apiGroups: [""] diff --git a/docs/router/private-service-routes.md b/docs/router/private-service-routes.md index 09949f8a..914714d0 100644 --- a/docs/router/private-service-routes.md +++ b/docs/router/private-service-routes.md @@ -1,11 +1,12 @@ # Private service routes -`ServiceEndpoint` uses a direct-endpoint contract. Its declared +`ServiceEndpoint` declares the producer destination. Its declared `spec.address`, `spec.protocol`, and `spec.port` are the exact destination tuple that the selected producer attachment must accept. Galactic does not interpret the address as a Kubernetes Service frontend: it does not discover -EndpointSlices, select a backend address, or translate the destination before -delivery. +EndpointSlices or select a Kubernetes backend. A translated policy uses +`spec.frontend.address` inside the consumer VPC and translates it to the +endpoint address before delivery. The controller that publishes the `ServiceEndpoint` owns the service implementation and health. It must ensure every attachment selected by @@ -30,15 +31,12 @@ carries the original packet through the SRv6 service tunnel. In both cases the producer receives the declared destination and the original consumer source unchanged. -The [Private Service Connect proposal](../enhancements/networking/private-service-connect/README.md) -describes the product capability and proposed frontend integration. - ## No consumer-visible route Galactic does not add the service address or a service-specific route to a consumer, and does not inject a route into a consumer guest. Consumers send the packet through their ordinary default gateway. The attachment's TC-eBPF -classifier matches the exact service tuple and redirects it to the selected +classifier matches the exact service or frontend tuple and redirects it to the selected producer; unauthorized ports are dropped. Galactic also does not install a host route for the service address. The producer service may still need to configure or bind the address locally so its own network stack or listener @@ -63,6 +61,22 @@ does not reference a `ServiceVIPBinding`, and the private service-route dataplane does not consult that translation to discover a backend. Reusing the same address in both APIs does not connect the two mechanisms. -If a service needs a distinct frontend and backend address or port, its API -must identify that backend explicitly and define its selection and health -semantics. `ServiceEndpoint` intentionally carries no such fields. +## Consumer service frontends + +Frontend translation is disabled by default. Enable it on participating routers +with `--service-frontend-enabled` or `GALACTIC_ROUTER_SERVICE_FRONTEND_ENABLED=true`. +The publishing controller supplies the producer address in `ServiceEndpoint`; +`ServiceRoutePolicy.spec.frontend.address` supplies the consumer address. +Both addresses must use the same IP family. Ports are not translated. + +A translated policy must pin the consumer VPC by name and UID and include an +`authorization.validUntil` deadline within the next two minutes. Selected +attachments must match that VPC's allocated network identity. Kernel checks +stop new requests and established replies when the authorization expires, even +if the controller or its API is unavailable. Renew authorization explicitly to +restore access. + +Policies claiming the same frontend address, protocol, and port on one consumer +attachment are rejected together. Separate VPC attachments can reuse that tuple. +The publisher owns authorization renewal and service health; Galactic enforces +network access and forwards packets. It does not interpret DNS records or zones. diff --git a/go.mod b/go.mod index 4d1ed154..d9c618d3 100644 --- a/go.mod +++ b/go.mod @@ -22,7 +22,7 @@ require ( github.com/spf13/viper v1.21.0 github.com/vishvananda/netlink v1.3.2-0.20261002201547-1d1c62507e3c go.datum.net/cloud v0.1.1-0.20260918213150-260f797bbc5c - go.datum.net/network v0.3.1-0.20261006185716-0f099eb3b0ea + go.datum.net/network v0.3.1-0.20261008185758-e391e93cf3de go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.69.0 go.opentelemetry.io/otel v1.44.0 go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.44.0 diff --git a/go.sum b/go.sum index 40853519..158f1bf8 100644 --- a/go.sum +++ b/go.sum @@ -206,6 +206,8 @@ go.datum.net/cloud v0.1.1-0.20260918213150-260f797bbc5c h1:k0P5980+kS90zCa0M4Moh go.datum.net/cloud v0.1.1-0.20260918213150-260f797bbc5c/go.mod h1:HTeaepXL0EqiqNsQ13K4/kGEwv+VLJ//BGrUfVIFfEA= go.datum.net/network v0.3.1-0.20261006185716-0f099eb3b0ea h1:q/8ZPV2fjLwFTpTepNn/9HigADL9c9nQHWKKGWMRGHQ= go.datum.net/network v0.3.1-0.20261006185716-0f099eb3b0ea/go.mod h1:dqzM8WZczbiZ9bCvsxjkoI10GJqQ24NVWnc9boXgOkE= +go.datum.net/network v0.3.1-0.20261008185758-e391e93cf3de h1:vGtnfrEwHrnH2YipcqgobhfPRFaA6OEtDFlvKYC+z9E= +go.datum.net/network v0.3.1-0.20261008185758-e391e93cf3de/go.mod h1:LawI34k+ddtv3kRbmjiss0Lb6QKpSr60YP3fX1IG5RU= go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64= go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y= go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.69.0 h1:2yEATaop1/a1I4psnSLgWVPLWwCzkqWakgJy7xTDVy0= diff --git a/internal/config/router.go b/internal/config/router.go index 10f30caf..d5f28958 100644 --- a/internal/config/router.go +++ b/internal/config/router.go @@ -51,14 +51,15 @@ var RouterBMPPolicies = []string{"pre-policy", "post-policy", "local-rib", "all" // --- Router environment variable keys -------------------------------------- const ( - EnvRouterNodeName = "GALACTIC_ROUTER_NODE_NAME" - EnvRouterReflector = "GALACTIC_ROUTER_REFLECTOR" - EnvRouterBGPListenPort = "GALACTIC_ROUTER_BGP_LISTEN_PORT" - EnvRouterBGPLocalAddr = "GALACTIC_ROUTER_BGP_LOCAL_ADDRESS" - EnvRouterMetricsPort = "GALACTIC_ROUTER_METRICS_PORT" - EnvRouterGRPCHealthPort = "GALACTIC_ROUTER_GRPC_HEALTH_PORT" - EnvRouterGCNamespace = "GALACTIC_ROUTER_GC_NAMESPACE" - EnvRouterGCInterval = "GALACTIC_ROUTER_GC_INTERVAL" + EnvRouterNodeName = "GALACTIC_ROUTER_NODE_NAME" + EnvRouterReflector = "GALACTIC_ROUTER_REFLECTOR" + EnvRouterBGPListenPort = "GALACTIC_ROUTER_BGP_LISTEN_PORT" + EnvRouterBGPLocalAddr = "GALACTIC_ROUTER_BGP_LOCAL_ADDRESS" + EnvRouterMetricsPort = "GALACTIC_ROUTER_METRICS_PORT" + EnvRouterGRPCHealthPort = "GALACTIC_ROUTER_GRPC_HEALTH_PORT" + EnvRouterGCNamespace = "GALACTIC_ROUTER_GC_NAMESPACE" + EnvRouterGCInterval = "GALACTIC_ROUTER_GC_INTERVAL" + EnvRouterServiceFrontendEnabled = "GALACTIC_ROUTER_SERVICE_FRONTEND_ENABLED" // EnvRouterWebhookEnabled gates the NetworkRule admission webhook. It // defaults to false: enabling it requires TLS cert material this repo does @@ -93,13 +94,14 @@ type RouterConfig struct { // mesh. A distinct signal from the listen port: whether a node accepts // inbound connections is not the same property as whether it is the // fabric's route reflector. - Reflector bool - BGPListenPort int - BGPLocalAddr string - MetricsPort int - GRPCHealthPort int - GCNamespace string - GCInterval time.Duration + Reflector bool + BGPListenPort int + BGPLocalAddr string + MetricsPort int + GRPCHealthPort int + GCNamespace string + GCInterval time.Duration + ServiceFrontendEnabled bool // WebhookEnabled, WebhookPort, and WebhookCertDir configure the NetworkRule // admission webhook. Disabled by default. @@ -131,6 +133,7 @@ func NewRouterConfig() *RouterConfig { v.SetDefault(KeyGRPCHealthPort, DefaultRouterGRPCHealthPort) v.SetDefault("gc_namespace", DefaultRouterGCNamespace) v.SetDefault("gc_interval", DefaultRouterGCInterval.String()) + v.SetDefault("service_frontend_enabled", false) v.SetDefault("webhook_enabled", false) v.SetDefault("webhook_port", DefaultRouterWebhookPort) v.SetDefault("webhook_cert_dir", "") @@ -161,6 +164,7 @@ func (c *RouterConfig) BindFlags(flags *pflag.FlagSet) { {FlagGRPCHealthPort, KeyGRPCHealthPort}, {"gc-namespace", "gc_namespace"}, {"gc-interval", "gc_interval"}, + {"service-frontend-enabled", "service_frontend_enabled"}, {"webhook-enabled", "webhook_enabled"}, {"webhook-port", "webhook_port"}, {"webhook-cert-dir", "webhook_cert_dir"}, @@ -189,6 +193,7 @@ func (c *RouterConfig) readFields() { c.GRPCHealthPort = c.v.GetInt(KeyGRPCHealthPort) c.GCNamespace = c.v.GetString("gc_namespace") c.GCInterval = c.v.GetDuration("gc_interval") + c.ServiceFrontendEnabled = c.v.GetBool("service_frontend_enabled") c.WebhookEnabled = c.v.GetBool("webhook_enabled") c.WebhookPort = c.v.GetInt("webhook_port") c.WebhookCertDir = c.v.GetString("webhook_cert_dir") diff --git a/internal/config/router_test.go b/internal/config/router_test.go index cfec6def..eff4774c 100644 --- a/internal/config/router_test.go +++ b/internal/config/router_test.go @@ -43,6 +43,9 @@ func TestRouterConfigDefaults(t *testing.T) { if cfg.Reflector { t.Error("Reflector = true, want false") } + if cfg.ServiceFrontendEnabled { + t.Error("service frontends must be disabled by default") + } if cfg.WebhookEnabled { t.Error("WebhookEnabled = true, want false (disabled by default)") } @@ -309,3 +312,10 @@ func TestParseBMPStation(t *testing.T) { }) } } + +func TestRouterConfigServiceFrontendEnv(t *testing.T) { + t.Setenv(EnvRouterServiceFrontendEnabled, testBoolTrue) + if !NewRouterConfig().ServiceFrontendEnabled { + t.Fatal("explicit service frontend enablement was ignored") + } +} diff --git a/internal/controller/serviceroute_access.go b/internal/controller/serviceroute_access.go new file mode 100644 index 00000000..a9d73621 --- /dev/null +++ b/internal/controller/serviceroute_access.go @@ -0,0 +1,84 @@ +// Copyright 2026 Datum Cloud, Inc. +// SPDX-License-Identifier: AGPL-3.0-or-later + +package controller + +import ( + "context" + "errors" + "fmt" + "net" + "time" + + "k8s.io/apimachinery/pkg/types" + + cloudv1alpha1 "go.datum.net/cloud/api/v1alpha1" + "go.datum.net/galactic/internal/serviceroute" + networkv1alpha1 "go.datum.net/network/api/v1alpha1" +) + +const maxServiceAuthorizationDuration = 2 * time.Minute + +func (r *ServiceRoutePolicyReconciler) now() time.Time { + if r.Now != nil { + return r.Now() + } + return time.Now() +} + +func (r *ServiceRoutePolicyReconciler) validateAuthorization(policy *networkv1alpha1.ServiceRoutePolicy) error { + if policy.Spec.Frontend == nil { + return nil + } + if policy.Spec.Authorization == nil { + return errors.New("frontend translation requires authorization.validUntil") + } + deadline := policy.Spec.Authorization.ValidUntil.Time + now := r.now() + if !deadline.After(now) || deadline.After(now.Add(maxServiceAuthorizationDuration)) { + return errors.New("network authorization deadline must be within the next two minutes") + } + return nil +} + +func (r *ServiceRoutePolicyReconciler) conflictingPolicies( + ctx context.Context, attachments []*cloudv1alpha1.VPCAttachment, +) (map[types.NamespacedName]bool, error) { + policies := &networkv1alpha1.ServiceRoutePolicyList{} + if err := r.List(ctx, policies); err != nil { + return nil, fmt.Errorf("list service policy conflicts: %w", err) + } + desired := map[types.NamespacedName][]serviceroute.RouteIntent{} + translated := map[types.NamespacedName]bool{} + nodes := map[string]bool{} + for _, a := range attachments { + nodes[a.Status.Node] = true + } + for i := range policies.Items { + p := &policies.Items[i] + if p.DeletionTimestamp != nil || r.validateAuthorization(p) != nil || (p.Spec.Frontend != nil && !r.FrontendEnabled) { + continue + } + endpoint := &networkv1alpha1.ServiceEndpoint{} + if err := r.Get(ctx, types.NamespacedName{ + Namespace: p.Namespace, + Name: p.Spec.ServiceRef.Name, + }, endpoint); err != nil { + continue + } + identity, err := r.consumerVPCIdentity(ctx, p) + if err != nil { + continue + } + key := types.NamespacedName{Namespace: p.Namespace, Name: p.Name} + translated[key] = p.Spec.Frontend != nil + for node := range nodes { + intents, err := serviceroute.CompileForConsumerVPC(p, endpoint, attachments, node, + func(*cloudv1alpha1.VPCAttachment) (net.IP, error) { return net.ParseIP("fd00::1"), nil }, identity) + if err == nil { + desired[key] = append(desired[key], intents...) + } + } + } + return serviceroute.ConflictingPolicies(desired, translated), nil +} diff --git a/internal/controller/serviceroute_access_test.go b/internal/controller/serviceroute_access_test.go new file mode 100644 index 00000000..031e2e90 --- /dev/null +++ b/internal/controller/serviceroute_access_test.go @@ -0,0 +1,323 @@ +// Copyright 2026 Datum Cloud, Inc. +// SPDX-License-Identifier: AGPL-3.0-or-later + +package controller + +import ( + "context" + "testing" + "time" + + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/types" + ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + + cloudv1alpha1 "go.datum.net/cloud/api/v1alpha1" + networkv1alpha1 "go.datum.net/network/api/v1alpha1" +) + +const ( + testPSCVPCName = "vpc" + testPSCEligibleLabel = "yes" + testPSCProducerLabel = "producer" + testPSCCompetingName = "psc-other" +) + +func TestTranslatedPolicyRequiresLiveConsumerPin(t *testing.T) { + scheme := runtime.NewScheme() + _ = cloudv1alpha1.AddToScheme(scheme) + vpc := &cloudv1alpha1.VPC{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: testServiceRouteConsumerName, + Name: testPSCVPCName, + UID: "live", + }, + Status: cloudv1alpha1.VPCStatus{ + VPC: "identity", + }, + } + r := &ServiceRoutePolicyReconciler{Client: fake.NewClientBuilder().WithScheme(scheme).WithObjects(vpc).Build()} + for _, tt := range []struct { + name string + ref *networkv1alpha1.ServiceRouteVPCReference + frontend bool + wantError bool + }{ + { + "LegacyDirect", + nil, + false, + false, + }, { + "Missing", + nil, + true, + true, + }, { + "Incomplete", + &networkv1alpha1.ServiceRouteVPCReference{ + Name: testPSCVPCName, + }, + true, + true, + }, + { + "WrongUID", + &networkv1alpha1.ServiceRouteVPCReference{ + Name: testPSCVPCName, + UID: "old", + }, + true, + true, + }, { + "Current", + &networkv1alpha1.ServiceRouteVPCReference{ + Name: testPSCVPCName, + UID: "live", + }, + true, + false, + }, + } { + t.Run(tt.name, func(t *testing.T) { + p := &networkv1alpha1.ServiceRoutePolicy{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: testServiceRouteConsumerName, + }, + Spec: networkv1alpha1.ServiceRoutePolicySpec{ + ConsumerVPCRef: tt.ref, + }, + } + if tt.frontend { + p.Spec.Frontend = &networkv1alpha1.ServiceRouteFrontend{Address: "fd53::53"} + } + _, err := r.consumerVPCIdentity(context.Background(), p) + if (err != nil) != tt.wantError { + t.Fatalf("error=%v", err) + } + }) + } +} + +func TestTranslatedAuthorizationBound(t *testing.T) { + now := time.Unix(1000, 0) + r := &ServiceRoutePolicyReconciler{Now: func() time.Time { return now }} + for _, tt := range []struct { + name string + offset time.Duration + wantError bool + }{ + { + "Expired", + -time.Second, + true, + }, + { + "AtDeadline", + 0, + true, + }, + { + "Current", + 90 * time.Second, + false, + }, + { + "Unbounded", + 121 * time.Second, + true, + }, + } { + t.Run(tt.name, func(t *testing.T) { + p := &networkv1alpha1.ServiceRoutePolicy{ + Spec: networkv1alpha1.ServiceRoutePolicySpec{ + Frontend: &networkv1alpha1.ServiceRouteFrontend{}, + Authorization: &networkv1alpha1.ServiceRouteAuthorization{ + ValidUntil: metav1.NewTime(now.Add(tt.offset)), + }, + }, + } + if err := r.validateAuthorization(p); (err != nil) != tt.wantError { + t.Fatalf("error=%v", err) + } + }) + } +} + +func pscFixture(t *testing.T) ( + *ServiceRoutePolicyReconciler, *networkv1alpha1.ServiceRoutePolicy, *networkv1alpha1.ServiceEndpoint, +) { + t.Helper() + scheme := runtime.NewScheme() + _ = cloudv1alpha1.AddToScheme(scheme) + _ = networkv1alpha1.AddToScheme(scheme) + now := time.Now() + p := &networkv1alpha1.ServiceRoutePolicy{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: testServiceRouteConsumerName, + Name: testServiceRouteName, + UID: "policy", + Generation: 1, + }, + Spec: networkv1alpha1.ServiceRoutePolicySpec{ + ServiceRef: networkv1alpha1.ServiceEndpointReference{ + Name: testServiceRouteName, + }, + ConsumerVPCRef: &networkv1alpha1.ServiceRouteVPCReference{ + Name: testPSCVPCName, + UID: testPSCVPCName, + }, + Frontend: &networkv1alpha1.ServiceRouteFrontend{ + Address: "fd53::53", + }, + Authorization: &networkv1alpha1.ServiceRouteAuthorization{ + ValidUntil: metav1.NewTime(now.Add(90 * time.Second)), + }, + AttachmentSelector: metav1.LabelSelector{ + MatchLabels: map[string]string{ + testServiceRouteConsumerName: testPSCEligibleLabel, + }, + }, + }, + } + e := &networkv1alpha1.ServiceEndpoint{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: testServiceRouteConsumerName, + Name: testServiceRouteName, + UID: "endpoint", + Generation: 1, + }, + Spec: networkv1alpha1.ServiceEndpointSpec{ + Address: "fd70::10", + Protocol: networkv1alpha1.NetworkRuleProtocolUDP, + Port: 53, + DeliveryMode: networkv1alpha1.ServiceEndpointDeliveryModeNodeLocal, + AttachmentSelector: &metav1.LabelSelector{ + MatchLabels: map[string]string{ + testPSCProducerLabel: testPSCEligibleLabel, + }, + }, + }, + } + v := &cloudv1alpha1.VPC{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: testServiceRouteConsumerName, + Name: testPSCVPCName, + UID: testPSCVPCName, + }, + Status: cloudv1alpha1.VPCStatus{ + VPC: "vpc-identity", + }, + } + attachment := func(ns, name, role, vpc string) *cloudv1alpha1.VPCAttachment { + return &cloudv1alpha1.VPCAttachment{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: ns, + Name: name, + UID: types.UID(name), + Generation: 1, + Labels: map[string]string{ + role: testPSCEligibleLabel, + }, + }, + Spec: cloudv1alpha1.VPCAttachmentSpec{ + VPC: cloudv1alpha1.VPCRef{ + Name: testPSCVPCName, + }, + }, + Status: cloudv1alpha1.VPCAttachmentStatus{ + ObservedGeneration: 1, + Node: "node-a", + VPC: vpc, + VPCAttachment: name, + HostInterface: name, + Conditions: []metav1.Condition{ + { + Type: "Ready", + Status: metav1.ConditionTrue, + }, + { + Type: "Programmed", + Status: metav1.ConditionTrue, + }, + }, + }, + } + } + c := attachment(testServiceRouteConsumerName, "client", testServiceRouteConsumerName, "vpc-identity") + s := attachment("service", "server", testPSCProducerLabel, "service-vpc") + r := &ServiceRoutePolicyReconciler{ + Client: fake.NewClientBuilder().WithScheme(scheme). + WithStatusSubresource(&networkv1alpha1.ServiceRoutePolicy{}).WithObjects(p, e, v, c, s).Build(), + NodeName: "node-a", + FrontendEnabled: true, + Programmer: &retryRouteProgrammer{}, + Now: func() time.Time { return now }, + } + return r, p, e +} + +func TestConflictingPoliciesFailClosedAndRecover(t *testing.T) { + for _, reverse := range []bool{false, true} { + t.Run(map[bool]string{false: "Forward", true: "Reverse"}[reverse], func(t *testing.T) { + r, p, e := pscFixture(t) + ctx := context.Background() + other := p.DeepCopy() + other.Name = testPSCCompetingName + other.UID = testPSCCompetingName + other.ResourceVersion = "" + e2 := e.DeepCopy() + e2.Name = testPSCCompetingName + e2.UID = "other-endpoint" + e2.ResourceVersion = "" + e2.Spec.Address = "fd70::20" + other.Spec.ServiceRef.Name = e2.Name + if err := r.Create(ctx, e2); err != nil { + t.Fatal(err) + } + if err := r.Create(ctx, other); err != nil { + t.Fatal(err) + } + policies := []*networkv1alpha1.ServiceRoutePolicy{p, other} + if reverse { + policies[0], policies[1] = policies[1], policies[0] + } + for _, policy := range policies { + key := types.NamespacedName{Namespace: policy.Namespace, Name: policy.Name} + if _, err := r.Reconcile(ctx, ctrl.Request{NamespacedName: key}); err != nil { + t.Fatal(err) + } + got := &networkv1alpha1.ServiceRoutePolicy{} + _ = r.Get(ctx, key, got) + if len(got.Status.Conditions) == 0 || got.Status.Conditions[0].Reason != "ConflictingPolicies" { + t.Fatalf("status=%+v", got.Status) + } + } + if len(r.Applied) != 0 { + t.Fatal("conflicting winner retained") + } + if err := r.syncAllPolicies(ctx); err != nil { + t.Fatal(err) + } + if len(r.Applied) != 0 { + t.Fatal("startup adopted conflicting policies") + } + if err := r.Delete(ctx, other); err != nil { + t.Fatal(err) + } + if _, err := r.Reconcile(ctx, ctrl.Request{ + NamespacedName: types.NamespacedName{ + Namespace: p.Namespace, + Name: p.Name, + }, + }); err != nil { + t.Fatal(err) + } + if len(r.Applied) != 1 { + t.Fatal("surviving policy did not recover") + } + }) + } +} diff --git a/internal/controller/serviceroute_controller.go b/internal/controller/serviceroute_controller.go index 86d5d802..8b4331e1 100644 --- a/internal/controller/serviceroute_controller.go +++ b/internal/controller/serviceroute_controller.go @@ -20,10 +20,12 @@ import ( "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/types" ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/builder" "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/handler" "sigs.k8s.io/controller-runtime/pkg/log" "sigs.k8s.io/controller-runtime/pkg/manager" + "sigs.k8s.io/controller-runtime/pkg/predicate" cloudv1alpha1 "go.datum.net/cloud/api/v1alpha1" "go.datum.net/galactic/internal/config" @@ -41,11 +43,13 @@ const reconcileResultError = "error" // resources or Linux routes. type ServiceRoutePolicyReconciler struct { client.Client - Scheme *runtime.Scheme - NodeName string - BGPNamespace string - Programmer serviceroute.RouteProgrammer - Metrics *serviceroute.Metrics + Scheme *runtime.Scheme + NodeName string + BGPNamespace string + Programmer serviceroute.RouteProgrammer + Metrics *serviceroute.Metrics + FrontendEnabled bool + Now func() time.Time mu sync.Mutex Applied map[types.NamespacedName]map[types.NamespacedName]serviceroute.RouteIntent @@ -108,7 +112,15 @@ func (r *ServiceRoutePolicyReconciler) Reconcile(ctx context.Context, req ctrl.R logger.Error(err, "get ServiceRoutePolicy") return ctrl.Result{}, err } + if policy.DeletionTimestamp != nil { + return ctrl.Result{}, r.removePolicyLocked(req.NamespacedName) + } endpoint := &networkv1alpha1.ServiceEndpoint{} + if policy.Spec.Frontend != nil && !r.FrontendEnabled { + joined := errors.Join(r.replacePolicyLocked(req.NamespacedName, nil), + r.setAccepted(ctx, policy, metav1.ConditionFalse, "FeatureDisabled", "service frontend translation is disabled")) + return ctrl.Result{}, joined + } endpointKey := types.NamespacedName{Namespace: policy.Namespace, Name: policy.Spec.ServiceRef.Name} if err := r.Get(ctx, endpointKey, endpoint); err != nil { if apierrors.IsNotFound(err) { @@ -126,6 +138,21 @@ func (r *ServiceRoutePolicyReconciler) Reconcile(ctx context.Context, req ctrl.R logger.Error(err, "get ServiceEndpoint", "endpoint", policy.Spec.ServiceRef.Name) return ctrl.Result{}, fmt.Errorf("get ServiceEndpoint %s/%s: %w", policy.Namespace, policy.Spec.ServiceRef.Name, err) } + if endpoint.DeletionTimestamp != nil { + return ctrl.Result{}, errors.Join(r.removePolicyLocked(req.NamespacedName), r.setAccepted(ctx, policy, + metav1.ConditionFalse, "EndpointDeleting", "service endpoint is deleting")) + } + vpcIdentity, err := r.consumerVPCIdentity(ctx, policy) + if err != nil { + joined := errors.Join(r.replacePolicyLocked(req.NamespacedName, nil), + r.setAccepted(ctx, policy, metav1.ConditionFalse, "InvalidConsumerVPC", err.Error())) + return ctrl.Result{}, joined + } + if err := r.validateAuthorization(policy); err != nil { + return ctrl.Result{RequeueAfter: serviceRouteMapResyncInterval}, + errors.Join(r.removePolicyLocked(req.NamespacedName), r.setAccepted(ctx, policy, metav1.ConditionFalse, + "InvalidAuthorization", err.Error())) + } attachments := &cloudv1alpha1.VPCAttachmentList{} if err := r.List(ctx, attachments); err != nil { resultLabel = reconcileResultError @@ -136,7 +163,20 @@ func (r *ServiceRoutePolicyReconciler) Reconcile(ctx context.Context, req ctrl.R for i := range attachments.Items { all = append(all, &attachments.Items[i]) } - intents, err := serviceroute.Compile(policy, endpoint, all, r.NodeName, r.sidResolver(ctx)) + conflicts, conflictErr := r.conflictingPolicies(ctx, all) + if conflictErr != nil { + return ctrl.Result{}, errors.Join(conflictErr, r.removePolicyLocked(req.NamespacedName)) + } + if conflicts[req.NamespacedName] { + var errs []error + for key := range conflicts { + errs = append(errs, r.removePolicyLocked(key)) + } + errs = append(errs, r.setAccepted(ctx, policy, metav1.ConditionFalse, "ConflictingPolicies", + "multiple policies claim a translated consumer tuple")) + return ctrl.Result{RequeueAfter: serviceRouteMapResyncInterval}, errors.Join(errs...) + } + intents, err := serviceroute.CompileForConsumerVPC(policy, endpoint, all, r.NodeName, r.sidResolver(ctx), vpcIdentity) if err != nil { var dependencyErr *serviceroute.DependencyNotReadyError if errors.As(err, &dependencyErr) { @@ -186,7 +226,7 @@ func (r *ServiceRoutePolicyReconciler) Reconcile(ctx context.Context, req ctrl.R return ctrl.Result{}, err } requeue := ctrl.Result{} - if len(intents) != 0 { + if len(intents) != 0 || policy.Spec.Frontend != nil { requeue.RequeueAfter = serviceRouteMapResyncInterval } return requeue, nil @@ -229,9 +269,12 @@ func (r *ServiceRoutePolicyReconciler) SetupWithManager(mgr ctrl.Manager) error return fmt.Errorf("add initial service route policy map sync: %w", err) } return ctrl.NewControllerManagedBy(mgr). - For(&networkv1alpha1.ServiceRoutePolicy{}). + For(&networkv1alpha1.ServiceRoutePolicy{}, builder.WithPredicates(predicate.GenerationChangedPredicate{})). + Watches(&networkv1alpha1.ServiceRoutePolicy{}, handler.EnqueueRequestsFromMapFunc(r.attachmentPolicies), + builder.WithPredicates(predicate.GenerationChangedPredicate{})). Watches(&networkv1alpha1.ServiceEndpoint{}, handler.EnqueueRequestsFromMapFunc(r.endpointPolicies)). Watches(&cloudv1alpha1.VPCAttachment{}, handler.EnqueueRequestsFromMapFunc(r.attachmentPolicies)). + Watches(&cloudv1alpha1.VPC{}, handler.EnqueueRequestsFromMapFunc(r.attachmentPolicies)). Watches(&networkv1alpha1.BGPRouter{}, handler.EnqueueRequestsFromMapFunc(r.routingDependencyPolicies)). Watches(&networkv1alpha1.BGPVRFInstance{}, handler.EnqueueRequestsFromMapFunc(r.routingDependencyPolicies)). Complete(r) @@ -282,6 +325,10 @@ func (r *ServiceRoutePolicyReconciler) syncAllPolicies(ctx context.Context) erro allAttachments = append(allAttachments, &attachments.Items[index]) } + conflicts, err := r.conflictingPolicies(ctx, allAttachments) + if err != nil { + return startupSyncError(err, true) + } type compiledPolicy struct { key types.NamespacedName intents []serviceroute.RouteIntent @@ -292,6 +339,10 @@ func (r *ServiceRoutePolicyReconciler) syncAllPolicies(ctx context.Context) erro policy := &policies.Items[index] key := types.NamespacedName{Namespace: policy.Namespace, Name: policy.Name} currentPolicies[key] = struct{}{} + if (policy.Spec.Frontend != nil && !r.FrontendEnabled) || r.validateAuthorization(policy) != nil || conflicts[key] { + compiled = append(compiled, compiledPolicy{key: key}) + continue + } endpoint := &networkv1alpha1.ServiceEndpoint{} endpointKey := types.NamespacedName{Namespace: policy.Namespace, Name: policy.Spec.ServiceRef.Name} if err := r.Get(ctx, endpointKey, endpoint); err != nil { @@ -301,7 +352,13 @@ func (r *ServiceRoutePolicyReconciler) syncAllPolicies(ctx context.Context) erro } return startupSyncError(fmt.Errorf("get ServiceEndpoint %s for startup sync: %w", endpointKey, err), true) } - intents, err := serviceroute.Compile(policy, endpoint, allAttachments, r.NodeName, r.sidResolver(ctx)) + vpcIdentity, err := r.consumerVPCIdentity(ctx, policy) + if err != nil { + compiled = append(compiled, compiledPolicy{key: key}) + continue + } + intents, err := serviceroute.CompileForConsumerVPC( + policy, endpoint, allAttachments, r.NodeName, r.sidResolver(ctx), vpcIdentity) if err != nil { // Invalid policies and node-local dependencies that are not ready // both fail closed in the startup snapshot. Ordinary reconciliation @@ -363,8 +420,38 @@ func (r *ServiceRoutePolicyReconciler) syncAllPolicies(ctx context.Context) erro return startupSyncError(errors.Join(syncErrs...), !finalized) } +func (r *ServiceRoutePolicyReconciler) consumerVPCIdentity( + ctx context.Context, policy *networkv1alpha1.ServiceRoutePolicy, +) (string, error) { + if policy.Spec.ConsumerVPCRef == nil { + if policy.Spec.Frontend != nil { + return "", errors.New("frontend translation requires consumerVPCRef") + } + return "", nil + } + if policy.Spec.ConsumerVPCRef.Name == "" || policy.Spec.ConsumerVPCRef.UID == "" { + return "", errors.New("consumerVPCRef requires name and UID") + } + ref := policy.Spec.ConsumerVPCRef + vpc := &cloudv1alpha1.VPC{} + key := types.NamespacedName{Namespace: policy.Namespace, Name: ref.Name} + if err := r.Get(ctx, key, vpc); err != nil { + return "", fmt.Errorf("resolve consumer VPC %s: %w", key, err) + } + if vpc.DeletionTimestamp != nil { + return "", errors.New("consumer VPC is deleting") + } + if string(vpc.UID) != ref.UID { + return "", fmt.Errorf("consumer VPC %s UID is %q, policy pins %q", key, vpc.UID, ref.UID) + } + if vpc.Status.VPC == "" { + return "", fmt.Errorf("consumer VPC %s has no allocated status identity", key) + } + return vpc.Status.VPC, nil +} + func (r *ServiceRoutePolicyReconciler) endpointPolicies(ctx context.Context, obj client.Object) []ctrl.Request { - return r.policiesForEndpoint(ctx, obj.GetNamespace(), obj.GetName()) + return r.allPolicies(ctx) } func (r *ServiceRoutePolicyReconciler) attachmentPolicies(ctx context.Context, _ client.Object) []ctrl.Request { @@ -422,23 +509,6 @@ func (r *ServiceRoutePolicyReconciler) sidResolver(ctx context.Context) servicer } } -func (r *ServiceRoutePolicyReconciler) policiesForEndpoint(ctx context.Context, namespace, name string) []ctrl.Request { - list := &networkv1alpha1.ServiceRoutePolicyList{} - if err := r.List(ctx, list, client.InNamespace(namespace)); err != nil { - return nil - } - requests := make([]ctrl.Request, 0) - for _, policy := range list.Items { - if policy.Spec.ServiceRef.Name == name { - requests = append(requests, ctrl.Request{NamespacedName: types.NamespacedName{ - Namespace: policy.Namespace, - Name: policy.Name, - }}) - } - } - return requests -} - func (r *ServiceRoutePolicyReconciler) allPolicies(ctx context.Context) []ctrl.Request { list := &networkv1alpha1.ServiceRoutePolicyList{} if err := r.List(ctx, list); err != nil { diff --git a/internal/serviceroute/conflicts.go b/internal/serviceroute/conflicts.go new file mode 100644 index 00000000..5cee82b1 --- /dev/null +++ b/internal/serviceroute/conflicts.go @@ -0,0 +1,49 @@ +// Copyright 2026 Datum Cloud, Inc. +// SPDX-License-Identifier: AGPL-3.0-or-later + +package serviceroute + +import "k8s.io/apimachinery/pkg/types" + +// ConflictingPolicies returns all policies whose consumer attachment and +// frontend tuple overlap a translated policy. Ambiguity rejects every claimant. +func ConflictingPolicies(intents map[types.NamespacedName][]RouteIntent, + translated map[types.NamespacedName]bool, +) map[types.NamespacedName]bool { + type tuple struct { + attachment types.NamespacedName + address, protocol string + port int32 + } + owners := map[tuple][]types.NamespacedName{} + for key, list := range intents { + for _, in := range list { + if in.Kind == RouteIntentRemoteProducer { + continue + } + for _, p := range in.Ports { + address := in.Service.String() + if in.Frontend != nil { + address = in.Frontend.String() + } + t := tuple{in.Attachment, address, string(p.Protocol), p.Port} + owners[t] = append(owners[t], key) + } + } + } + conflicts := map[types.NamespacedName]bool{} + for _, keys := range owners { + distinct := map[types.NamespacedName]bool{} + anyTranslated := false + for _, k := range keys { + distinct[k] = true + anyTranslated = anyTranslated || translated[k] + } + if len(distinct) > 1 && anyTranslated { + for k := range distinct { + conflicts[k] = true + } + } + } + return conflicts +} diff --git a/internal/serviceroute/conflicts_test.go b/internal/serviceroute/conflicts_test.go new file mode 100644 index 00000000..d473a606 --- /dev/null +++ b/internal/serviceroute/conflicts_test.go @@ -0,0 +1,34 @@ +// Copyright 2026 Datum Cloud, Inc. +// SPDX-License-Identifier: AGPL-3.0-or-later + +package serviceroute + +import ( + "net" + "testing" + + "k8s.io/apimachinery/pkg/types" + + networkv1alpha1 "go.datum.net/network/api/v1alpha1" +) + +func TestTranslatedConflictScopePreservesDistinctAttachments(t *testing.T) { + _, service, _ := net.ParseCIDR("fd70::10/128") + _, frontend, _ := net.ParseCIDR("fd53::53/128") + first := types.NamespacedName{Namespace: "a", Name: "psc-conflict"} + second := types.NamespacedName{Namespace: "b", Name: "psc-conflict"} + route := RouteIntent{Attachment: types.NamespacedName{Namespace: "a", Name: "client"}, Kind: RouteIntentLocal, + Service: service, Frontend: frontend, Ports: []networkv1alpha1.ServiceRouteProtocolPort{{Protocol: "udp", Port: 53}}} + other := route + other.Attachment.Namespace = "b" + intents := map[types.NamespacedName][]RouteIntent{first: {route}, second: {other}} + translated := map[types.NamespacedName]bool{first: true, second: true} + if conflicts := ConflictingPolicies(intents, translated); len(conflicts) != 0 { + t.Fatalf("distinct VPC attachments conflict=%v", conflicts) + } + other.Attachment = route.Attachment + intents[second] = []RouteIntent{other} + if conflicts := ConflictingPolicies(intents, translated); len(conflicts) != 2 { + t.Fatalf("duplicate claim did not fail closed=%v", conflicts) + } +} diff --git a/internal/serviceroute/planner.go b/internal/serviceroute/planner.go index b3274b58..4cbe21f0 100644 --- a/internal/serviceroute/planner.go +++ b/internal/serviceroute/planner.go @@ -71,6 +71,20 @@ func (e *DependencyNotReadyError) Unwrap() error { return e.Err } // consumer and emits only intent halves owned by localNode. func Compile(policy *networkv1alpha1.ServiceRoutePolicy, endpoint *networkv1alpha1.ServiceEndpoint, attachments []*cloudv1alpha1.VPCAttachment, localNode string, resolveSID SIDResolver, +) ([]RouteIntent, error) { + return compile(policy, endpoint, attachments, localNode, resolveSID, "") +} + +// CompileForConsumerVPC compiles a policy while requiring selected consumer +// attachments to carry the live provider VPC's allocated status identity. +func CompileForConsumerVPC(policy *networkv1alpha1.ServiceRoutePolicy, endpoint *networkv1alpha1.ServiceEndpoint, + attachments []*cloudv1alpha1.VPCAttachment, localNode string, resolveSID SIDResolver, consumerVPCIdentity string, +) ([]RouteIntent, error) { + return compile(policy, endpoint, attachments, localNode, resolveSID, consumerVPCIdentity) +} + +func compile(policy *networkv1alpha1.ServiceRoutePolicy, endpoint *networkv1alpha1.ServiceEndpoint, + attachments []*cloudv1alpha1.VPCAttachment, localNode string, resolveSID SIDResolver, consumerVPCIdentity string, ) ([]RouteIntent, error) { if policy == nil || endpoint == nil { return nil, errors.New("service route policy and endpoint are required") @@ -86,6 +100,16 @@ func Compile(policy *networkv1alpha1.ServiceRoutePolicy, endpoint *networkv1alph if err != nil { return nil, fmt.Errorf("parse service endpoint address: %w", err) } + frontend := cloneIPNet(service) + if policy.Spec.Frontend != nil { + frontend, err = hostPrefix(policy.Spec.Frontend.Address) + if err != nil { + return nil, fmt.Errorf("parse service frontend address: %w", err) + } + if (frontend.IP.To4() == nil) != (service.IP.To4() == nil) { + return nil, errors.New("service frontend and endpoint address families must match") + } + } ports, err := servicePorts(policy, endpoint) if err != nil { return nil, err @@ -104,6 +128,9 @@ func Compile(policy *networkv1alpha1.ServiceRoutePolicy, endpoint *networkv1alph if consumer.Status.HostInterface == "" || !consumerSelector.Matches(labels.Set(consumer.Labels)) { continue } + if !consumerMatchesVPC(policy, consumer, consumerVPCIdentity) { + continue + } producer := chooseProducer(endpoint.Spec.DeliveryMode, consumer.Status.Node, producers) if producer == nil { continue @@ -114,7 +141,7 @@ func Compile(policy *networkv1alpha1.ServiceRoutePolicy, endpoint *networkv1alph if consumer.Status.Node == localNode { intents = append(intents, RouteIntent{ Attachment: consumerKey, ProducerAttachment: producerKey, Kind: RouteIntentLocal, - Service: cloneIPNet(service), ConsumerDevice: consumer.Status.HostInterface, + Service: cloneIPNet(service), Frontend: cloneIPNet(frontend), ConsumerDevice: consumer.Status.HostInterface, ServiceDevice: producer.Status.HostInterface, Ports: clonePorts(ports), }) } @@ -135,7 +162,7 @@ func Compile(policy *networkv1alpha1.ServiceRoutePolicy, endpoint *networkv1alph } intents = append(intents, RouteIntent{ Attachment: consumerKey, ProducerAttachment: producerKey, Kind: RouteIntentRemoteConsumer, - Service: cloneIPNet(service), ConsumerDevice: consumer.Status.HostInterface, + Service: cloneIPNet(service), Frontend: cloneIPNet(frontend), ConsumerDevice: consumer.Status.HostInterface, ServiceSID: cloneIP(serviceSID), GrantID: grantID, Ports: clonePorts(ports), }) } @@ -147,11 +174,12 @@ func Compile(policy *networkv1alpha1.ServiceRoutePolicy, endpoint *networkv1alph } intents = append(intents, RouteIntent{ Attachment: consumerKey, ProducerAttachment: producerKey, Kind: RouteIntentRemoteProducer, - Service: cloneIPNet(service), ServiceDevice: producer.Status.HostInterface, + Service: cloneIPNet(service), Frontend: cloneIPNet(frontend), ServiceDevice: producer.Status.HostInterface, ConsumerSID: cloneIP(consumerSID), GrantID: grantID, Ports: clonePorts(ports), }) } } + setIntentAuthorization(policy, intents) sort.Slice(intents, func(i, j int) bool { if intents[i].Attachment != intents[j].Attachment { return intents[i].Attachment.String() < intents[j].Attachment.String() @@ -221,7 +249,8 @@ func producerCandidates(endpoint *networkv1alpha1.ServiceEndpoint, func readyAttachments(attachments []*cloudv1alpha1.VPCAttachment) []*cloudv1alpha1.VPCAttachment { ready := make([]*cloudv1alpha1.VPCAttachment, 0, len(attachments)) for _, attachment := range attachments { - if attachment == nil || attachment.Status.ObservedGeneration != attachment.Generation || + if attachment == nil || attachment.DeletionTimestamp != nil || + attachment.Status.ObservedGeneration != attachment.Generation || attachment.Status.Node == "" || attachment.Status.VPC == "" || attachment.Status.VPCAttachment == "" || !apimeta.IsStatusConditionTrue(attachment.Status.Conditions, cloudv1alpha1.ConditionTypeReady) || !apimeta.IsStatusConditionTrue(attachment.Status.Conditions, cloudv1alpha1.ConditionTypeProgrammed) { @@ -286,6 +315,19 @@ func serviceGrantID(policy *networkv1alpha1.ServiceRoutePolicy, endpoint *networ _, _ = h.Write(length[:]) _, _ = h.Write([]byte(uid)) } + // Preserve direct-policy grant identifiers across mixed-version router fleets. + if policy.Spec.Frontend != nil || policy.Spec.ConsumerVPCRef != nil || policy.Spec.Authorization != nil { + _, _ = fmt.Fprintf(h, "|%d|%d|%d|%d|", + policy.Generation, endpoint.Generation, consumer.Generation, producer.Generation) + _, _ = h.Write([]byte(endpoint.Spec.Address)) + if policy.Spec.Frontend != nil { + _, _ = h.Write([]byte(policy.Spec.Frontend.Address)) + } + if policy.Spec.ConsumerVPCRef != nil { + _, _ = h.Write([]byte(policy.Spec.ConsumerVPCRef.Name)) + _, _ = h.Write([]byte(policy.Spec.ConsumerVPCRef.UID)) + } + } for _, port := range ports { _, _ = h.Write([]byte(port.Protocol)) var encoded [4]byte @@ -317,3 +359,22 @@ func cloneIP(in net.IP) net.IP { return append(net.IP(nil), in...) } func clonePorts(in []networkv1alpha1.ServiceRouteProtocolPort) []networkv1alpha1.ServiceRouteProtocolPort { return append([]networkv1alpha1.ServiceRouteProtocolPort(nil), in...) } + +func consumerMatchesVPC(policy *networkv1alpha1.ServiceRoutePolicy, + consumer *cloudv1alpha1.VPCAttachment, identity string, +) bool { + if ref := policy.Spec.ConsumerVPCRef; ref != nil && + (consumer.Namespace != policy.Namespace || consumer.Spec.VPC.Name != ref.Name) { + return false + } + return identity == "" || consumer.Status.VPC == identity +} + +func setIntentAuthorization(policy *networkv1alpha1.ServiceRoutePolicy, intents []RouteIntent) { + if policy.Spec.Frontend == nil || policy.Spec.Authorization == nil { + return + } + for i := range intents { + intents[i].AuthorizationValidUntil = policy.Spec.Authorization.ValidUntil.Time + } +} diff --git a/internal/serviceroute/planner_test.go b/internal/serviceroute/planner_test.go index 6ca1e331..46611e38 100644 --- a/internal/serviceroute/planner_test.go +++ b/internal/serviceroute/planner_test.go @@ -216,6 +216,34 @@ func TestCompileUsesObservedNodeAndSelector(t *testing.T) { } } +func TestCompileTranslatesFrontendAndPinsConsumerVPC(t *testing.T) { + policy, endpoint := testPolicyAndEndpoint(networkv1alpha1.ServiceEndpointDeliveryModeNodeLocal) + policy.Spec.Frontend = &networkv1alpha1.ServiceRouteFrontend{Address: "fd53::53"} + policy.Spec.ConsumerVPCRef = &networkv1alpha1.ServiceRouteVPCReference{Name: "consumer-vpc", UID: "vpc-uid"} + consumer := readyAttachment("platform", "selected", "consumer-uid", testNodeName, "consumer0", + testLabels(testAccessLabel, "yes")) + consumer.Spec.VPC.Name = "consumer-vpc" + forged := readyAttachment("platform", "forged", "forged-uid", testNodeName, "forged0", + testLabels(testAccessLabel, "yes")) + forged.Spec.VPC.Name = "other-vpc" + producer := readyAttachment("service", "producer", "producer-uid", testNodeName, "producer0", + testLabels(testProducerLabel, "yes")) + + consumer.Status.VPC = "live-provider-identity" + forged.Status.VPC = "retired-provider-identity" + got, err := CompileForConsumerVPC(policy, endpoint, + []*cloudv1alpha1.VPCAttachment{consumer, forged, producer}, testNodeName, nil, "live-provider-identity") + if err != nil { + t.Fatalf("Compile: %v", err) + } + if len(got) != 1 || got[0].Attachment.Name != consumer.Name { + t.Fatalf("intents = %+v, want only live VPC member", got) + } + if got[0].Frontend.String() != "fd53::53/128" || got[0].Service.String() != "fd00::53/128" { + t.Fatalf("translation = %s -> %s", got[0].Frontend, got[0].Service) + } +} + func TestCompileUsesEndpointPortByDefault(t *testing.T) { policy, endpoint := testPolicyAndEndpoint(networkv1alpha1.ServiceEndpointDeliveryModeNodeLocal) policy.Spec.ProtocolPorts = nil @@ -404,3 +432,29 @@ func TestServiceGrantIDCanonicalizesPortOrder(t *testing.T) { t.Fatalf("canonical grant IDs differ: %x / %x", idA, idB) } } + +func TestServiceGrantGenerationCompatibility(t *testing.T) { + for _, translated := range []bool{false, true} { + t.Run(map[bool]string{false: "Direct", true: "Translated"}[translated], func(t *testing.T) { + policy, endpoint := testPolicyAndEndpoint(networkv1alpha1.ServiceEndpointDeliveryModePreferNodeLocal) + if translated { + policy.Spec.Frontend = &networkv1alpha1.ServiceRouteFrontend{Address: "fd70::1"} + } + consumer := readyAttachment("tenant", "client", "client", testNodeName, "client0", nil) + producer := readyAttachment("service", "server", "server", testRemoteNodeName, "server0", nil) + ports := []networkv1alpha1.ServiceRouteProtocolPort{{Protocol: "udp", Port: 53}} + before, err := serviceGrantID(policy, endpoint, consumer, producer, ports) + if err != nil { + t.Fatal(err) + } + policy.Generation++ + after, err := serviceGrantID(policy, endpoint, consumer, producer, ports) + if err != nil { + t.Fatal(err) + } + if (before != after) != translated { + t.Fatal("legacy grants must remain compatible; translated grants must fence changed authorization") + } + }) + } +}