diff --git a/cmd/masque-authority/main.go b/cmd/masque-authority/main.go new file mode 100644 index 00000000..870543ac --- /dev/null +++ b/cmd/masque-authority/main.go @@ -0,0 +1,354 @@ +// Copyright 2026 Datum Cloud, Inc. +// +// SPDX-License-Identifier: AGPL-3.0-or-later + +// Command masque-authority is the MASQUE grant authority. "serve" answers +// the MASQUE provider pods' Authorize, Refresh and Retire calls over HTTPS +// and sweeps revoked authorizations. The "ledger" and "grant" commands are +// the operator's internal provisioning path: they create allocation ledgers +// and connector grants, print a new credential exactly once, and rotate, +// disable or retire connectors. +package main + +import ( + "context" + "crypto/tls" + "encoding/json" + "errors" + "fmt" + "log/slog" + "net/http" + "net/netip" + "os" + "os/signal" + "strings" + "sync" + "syscall" + "time" + + "github.com/spf13/cobra" + "k8s.io/apimachinery/pkg/runtime" + utilruntime "k8s.io/apimachinery/pkg/util/runtime" + clientgoscheme "k8s.io/client-go/kubernetes/scheme" + ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/client" + + cloudv1alpha1 "go.datum.net/galactic/api/cloud/v1alpha1" + galacticv1alpha1 "go.datum.net/galactic/api/v1alpha1" + "go.datum.net/galactic/internal/edge/admission" + "go.datum.net/galactic/internal/edge/masque/authority" + "go.datum.net/galactic/internal/logging" + "go.datum.net/galactic/internal/metadata" +) + +func main() { + if err := newRootCommand().Execute(); err != nil { + os.Exit(1) + } +} + +func newRootCommand() *cobra.Command { + var logOpts logging.Options + var namespace string + cmd := &cobra.Command{ + Use: "masque-authority", + Short: "Authorize MASQUE connectors and provision their grants", + Version: metadata.Version, + SilenceUsage: true, + PersistentPreRunE: func(*cobra.Command, []string) error { + return logOpts.Setup() + }, + } + logOpts.BindFlags(cmd.PersistentFlags()) + cmd.PersistentFlags().StringVar(&namespace, "namespace", os.Getenv("POD_NAMESPACE"), + "Namespace holding the grant records and ledgers (env POD_NAMESPACE)") + cmd.AddCommand(newServeCommand(&namespace), newLedgerCommand(&namespace), newGrantCommand(&namespace)) + return cmd +} + +// newClient returns an uncached API client. +func newClient() (client.Client, error) { + scheme := runtime.NewScheme() + utilruntime.Must(clientgoscheme.AddToScheme(scheme)) + utilruntime.Must(galacticv1alpha1.AddToScheme(scheme)) + utilruntime.Must(cloudv1alpha1.AddToScheme(scheme)) + cfg, err := ctrl.GetConfig() + if err != nil { + return nil, err + } + // Every call gives up well before its two-second budget would let a + // stale answer through. + cfg.Timeout = 2 * time.Second + return client.New(cfg, client.Options{Scheme: scheme}) +} + +func requireNamespace(ns string) error { + if ns == "" { + return errors.New("--namespace or POD_NAMESPACE is required") + } + return nil +} + +type serveOptions struct { + provider, providerSA, cell, listen, certFile, keyFile, audience string + sweep time.Duration +} + +func newServeCommand(namespace *string) *cobra.Command { + o := &serveOptions{} + cmd := &cobra.Command{ + Use: "serve", + Short: "Serve the provider pods' Authorize, Refresh and Retire calls", + Args: cobra.NoArgs, + RunE: func(cmd *cobra.Command, _ []string) error { return serve(cmd.Context(), *namespace, o) }, + } + f := cmd.Flags() + f.StringVar(&o.provider, "provider", "", "EdgeProvider the issued EdgeServices grant (required)") + f.StringVar(&o.providerSA, "provider-service-account", "", + "namespace/name of the ServiceAccount the provider pods run as; no other caller is served (required)") + f.StringVar(&o.cell, "cell", "", "This cell's name; only its grant records are honored (required)") + f.StringVar(&o.listen, "listen", ":8443", "HTTPS listen address") + f.StringVar(&o.certFile, "tls-cert", "/etc/masque-authority/tls/tls.crt", "Server certificate, reread on change") + f.StringVar(&o.keyFile, "tls-key", "/etc/masque-authority/tls/tls.key", "Server key, reread on change") + f.StringVar(&o.audience, "audience", authority.DefaultAudience, "Audience of the provider pods' tokens") + f.DurationVar(&o.sweep, "sweep-interval", time.Second, "How often revoked authorizations are swept") + return cmd +} + +func serve(parent context.Context, namespace string, o *serveOptions) error { + if err := requireNamespace(namespace); err != nil { + return err + } + saNS, saName, ok := strings.Cut(o.providerSA, "/") + if o.provider == "" || o.cell == "" || !ok || saNS == "" || saName == "" { + return errors.New("--provider, --cell and --provider-service-account namespace/name are required") + } + if parent == nil { + parent = context.Background() + } + ctx, stop := signal.NotifyContext(parent, syscall.SIGINT, syscall.SIGTERM) + defer stop() + c, err := newClient() + if err != nil { + return err + } + certs := &certReloader{certFile: o.certFile, keyFile: o.keyFile} + if _, err := certs.get(nil); err != nil { + return err + } + a := &authority.Authority{ + Client: c, Namespace: namespace, Provider: o.provider, Cell: o.cell, + ProviderServiceAccount: admission.ServiceAccount{Namespace: saNS, Name: saName}, Log: slog.Default(), + } + reviewer := authority.KubeTokenReviewer{Client: c, Audience: o.audience} + srv := &http.Server{ + Addr: o.listen, + Handler: (&authority.Server{Authority: a, Reviewer: reviewer}).Handler(), + TLSConfig: &tls.Config{MinVersion: tls.VersionTLS13, GetCertificate: certs.get}, + ReadHeaderTimeout: 2 * time.Second, + ReadTimeout: 5 * time.Second, + WriteTimeout: 5 * time.Second, + IdleTimeout: 60 * time.Second, + } + go a.RunSweeps(ctx, o.sweep) + errc := make(chan error, 1) + go func() { errc <- srv.ListenAndServeTLS("", "") }() + slog.Info("serving", "listen", o.listen, "provider", o.provider, "cell", o.cell) + select { + case err := <-errc: + return err + case <-ctx.Done(): + } + shutdown, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + return srv.Shutdown(shutdown) +} + +// certReloader serves the certificate in certFile and keyFile, reloading it +// when either file changes. +type certReloader struct { + certFile, keyFile string + mu sync.Mutex + stamp string + cert *tls.Certificate +} + +func (r *certReloader) get(*tls.ClientHelloInfo) (*tls.Certificate, error) { + var stamp strings.Builder + for _, p := range []string{r.certFile, r.keyFile} { + fi, err := os.Stat(p) + if err != nil { + return nil, err + } + fmt.Fprintf(&stamp, "%d/%d;", fi.ModTime().UnixNano(), fi.Size()) + } + r.mu.Lock() + defer r.mu.Unlock() + if r.cert != nil && r.stamp == stamp.String() { + return r.cert, nil + } + cert, err := tls.LoadX509KeyPair(r.certFile, r.keyFile) + if err != nil { + return nil, fmt.Errorf("load the server certificate: %w", err) + } + r.cert, r.stamp = &cert, stamp.String() + return r.cert, nil +} + +func newLedgerCommand(namespace *string) *cobra.Command { + var vpc, cell, connectors, pools string + cmd := &cobra.Command{Use: "ledger", Short: "Manage connector allocation ledgers"} + initCmd := &cobra.Command{ + Use: "init", + Short: "Create the allocation ledger of a VPC in a cell", + Args: cobra.NoArgs, + RunE: func(cmd *cobra.Command, _ []string) error { + if err := requireNamespace(*namespace); err != nil { + return err + } + ref, err := parseVPC(vpc) + if err != nil { + return err + } + cp, err := netip.ParsePrefix(connectors) + if err != nil { + return fmt.Errorf("--connectors: %w", err) + } + pp, err := netip.ParsePrefix(pools) + if err != nil { + return fmt.Errorf("--pools: %w", err) + } + c, err := newClient() + if err != nil { + return err + } + l := authority.Ledgers{Client: c, Namespace: *namespace, Now: time.Now} + return l.Init(cmd.Context(), ref, cell, cp, pp) + }, + } + f := initCmd.Flags() + f.StringVar(&vpc, "vpc", "", "VPC as namespace/name (required)") + f.StringVar(&cell, "cell", "", "Cell name (required)") + f.StringVar(&connectors, "connectors", "", "Reserved IPv6 /80 the connector /96s come from (required)") + f.StringVar(&pools, "pools", "", "Reserved IPv6 /80 the attachment /120 pools come from (required)") + cmd.AddCommand(initCmd) + return cmd +} + +func newGrantCommand(namespace *string) *cobra.Command { + cmd := &cobra.Command{Use: "grant", Short: "Provision, rotate, disable and retire connector grants"} + grants := func() (authority.Grants, error) { + if err := requireNamespace(*namespace); err != nil { + return authority.Grants{}, err + } + c, err := newClient() + return authority.Grants{Client: c, Namespace: *namespace, Now: time.Now}, err + } + + var req authority.CreateRequest + var vpc, ingress, egress string + var routes []string + create := &cobra.Command{ + Use: "create", + Short: "Allocate a connector and print its new credential once", + Args: cobra.NoArgs, + RunE: func(cmd *cobra.Command, _ []string) error { + g, err := grants() + if err != nil { + return err + } + if req.VPC, err = parseVPC(vpc); err != nil { + return err + } + for _, r := range routes { + p, err := netip.ParsePrefix(r) + if err != nil { + return fmt.Errorf("--route: %w", err) + } + req.Routes = append(req.Routes, p) + } + if ingress != "" || egress != "" { + req.TrafficPolicy = &galacticv1alpha1.TrafficPolicyBindingSpec{} + if ingress != "" { + req.TrafficPolicy.IngressPolicyRef = &galacticv1alpha1.LocalPolicyReference{Name: ingress} + } + if egress != "" { + req.TrafficPolicy.EgressPolicyRef = &galacticv1alpha1.LocalPolicyReference{Name: egress} + } + } + issued, err := g.Create(cmd.Context(), req) + if err != nil { + return err + } + return printJSON(cmd, issued) + }, + } + f := create.Flags() + f.StringVar(&vpc, "vpc", "", "VPC as namespace/name (required)") + f.StringVar(&req.Cell, "cell", "", "Cell name (required)") + f.StringVar(&req.Connector, "connector", "", "Connector UUID; empty generates one") + f.DurationVar(&req.Lifetime, "lifetime", authority.MaxLifetime, "Credential lifetime, at most 24h") + f.StringSliceVar(&routes, "route", nil, "gVPC destination prefix the connector may reach; repeatable") + f.StringVar(&ingress, "ingress-policy", "", "TrafficPolicy for traffic delivered to the connector") + f.StringVar(&egress, "egress-policy", "", "TrafficPolicy for traffic the connector sends") + + var lifetime time.Duration + rotate := &cobra.Command{ + Use: "rotate CONNECTOR", + Short: "Replace a connector's credential and print the new one once", + Args: cobra.ExactArgs(1), + RunE: func(cmd *cobra.Command, args []string) error { + g, err := grants() + if err != nil { + return err + } + issued, err := g.Rotate(cmd.Context(), args[0], lifetime) + if err != nil { + return err + } + return printJSON(cmd, issued) + }, + } + rotate.Flags().DurationVar(&lifetime, "lifetime", authority.MaxLifetime, "Credential lifetime, at most 24h") + + disable := &cobra.Command{ + Use: "disable CONNECTOR", + Short: "Revoke a connector; its allocation is kept", + Args: cobra.ExactArgs(1), + RunE: func(cmd *cobra.Command, args []string) error { + g, err := grants() + if err != nil { + return err + } + return g.Disable(cmd.Context(), args[0]) + }, + } + retire := &cobra.Command{ + Use: "retire CONNECTOR", + Short: "Revoke a connector and tombstone its allocation for good", + Args: cobra.ExactArgs(1), + RunE: func(cmd *cobra.Command, args []string) error { + g, err := grants() + if err != nil { + return err + } + return g.Retire(cmd.Context(), args[0]) + }, + } + cmd.AddCommand(create, rotate, disable, retire) + return cmd +} + +func parseVPC(s string) (authority.VPCRef, error) { + ns, name, ok := strings.Cut(s, "/") + if !ok || ns == "" || name == "" { + return authority.VPCRef{}, fmt.Errorf("--vpc %q is not namespace/name", s) + } + return authority.VPCRef{Namespace: ns, Name: name}, nil +} + +func printJSON(cmd *cobra.Command, v any) error { + enc := json.NewEncoder(cmd.OutOrStdout()) + enc.SetIndent("", " ") + return enc.Encode(v) +} diff --git a/docs/designs/masque/acceptance.md b/docs/designs/masque/acceptance.md index 7a3a9c30..6dbc0ce2 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, M07, and the #989 findings A06, A14 and A16, plus NAT egress from the attachment node, two connectors sharing one attachment while one request ends, router garbage collection while connectors forward, a provider process killed and restarted in place, pod recreation, a partly set-up instance killed without shutdown, and asserted teardown. A14's namespace-only impostor is a real-kernel test of `internal/edge/hostattach`. The provider is the test-only fixture in `tests/edgeconformance/masqueprovider`. It runs the [MASQUE provider adapter](../../edge/masque-provider.md) with `internal/edge/connectip`, and admits static test credentials in place of the [#977](https://github.com/datum-cloud/galactic/issues/977) authority. A lab result does not satisfy a row that requires the target platform. [#985](https://github.com/datum-cloud/galactic/issues/985) owns running these checks on the target platform. +`task verify:masque-conformance`, run from `deploy/containerlab`, runs the [#974](https://github.com/datum-cloud/galactic/issues/974) packet-access checks on the lab's dfw cluster. It covers M16, M07, and the #989 findings A06, A14 and A16, plus NAT egress from the attachment node, two connectors sharing one attachment while one request ends, router garbage collection while connectors forward, a provider process killed and restarted in place, pod recreation, a partly set-up instance killed without shutdown, and asserted teardown. A14's namespace-only impostor is a real-kernel test of `internal/edge/hostattach`. The provider is the test-only fixture in `tests/edgeconformance/masqueprovider`. It runs the [MASQUE provider adapter](../../edge/masque-provider.md) with `internal/edge/connectip`, and admits static test credentials, or authorizes through the [#977](https://github.com/datum-cloud/galactic/issues/977) [grant authority](../../edge/masque-authority.md) when configured. A lab result does not satisfy a row that requires the target platform. [#985](https://github.com/datum-cloud/galactic/issues/985) owns running these checks on the target platform. ## Epic and implementation checks diff --git a/docs/designs/masque/contract.md b/docs/designs/masque/contract.md index 75046ecb..b10fb1c7 100644 --- a/docs/designs/masque/contract.md +++ b/docs/designs/masque/contract.md @@ -38,6 +38,8 @@ The operator owns one internal grant record per connector: immutable connector U The provider sends credentials to this authority over an internal authenticated TLS channel. This is a private MASQUE interface delivered by [#977](https://github.com/datum-cloud/galactic/issues/977): `Authorize` takes the credential, requested VPC selector, authenticated pod UID and request nonce; `Refresh` takes the resulting authorization ID, credential revision and service UID; `Retire` conditionally ends that authorization ID. A successful response returns connector/allocation IDs, VPC UID, service namespace/name/UID, /96, allowed destinations, credential revision and absolute deadline. Invalid credentials return unauthenticated, wrong VPC or retired revision returns forbidden, and unavailable/stale authority returns retryable unavailable without a grant. Refresh never accepts a new credential revision implicitly. The authority validates every call against current records and binds authorization IDs to the calling pod; retries with the same nonce cannot create a second incarnation. Revocation notifications carry the authorization ID and revision and may accelerate closure, but delivery is not the safety backstop. The authority verifies the provider's pod-bound identity and uses uncached API reads of the grant and VPC; a caller-supplied VPC is only a selector that must match. Trust roots and server identity are operator-mounted, rotated with a bounded overlap of 24 hours; unknown or expired roots fail closed. TLS session resumption does not bypass a fresh admission decision. Neither the provider nor the client chooses its prefix, service, attachment identity or allowed destinations. +The [grant authority guide](../../edge/masque-authority.md) describes what #977 built. It differs from this section in two ways. An `Authorize` call also carries the time it was first sent, and a retired incarnation is kept for a 10 s retry window, so a late retry can never create a second incarnation. Revocation notifications are not built; the provider learns of a revocation from its next `Refresh`. + The allocator holds one ConfigMap ledger per VPC/cell, at most 256 entries and 256 KiB. Entries contain connector UUID, allocation UUID, /96, distinct attachment /120 pool, creation time and retired state. Every insert or retirement uses the ledger UID and resourceVersion as preconditions. Concurrent allocations conflict and retry from a fresh read; a lost reply is resolved by reading the connector's allocation UUID. The ledger is authoritative even when no EdgeRoute exists. It is writable only by the authority; providers receive immutable allocation snapshots. A missing/recreated ledger is an operator recovery event: refuse allocation and restore from backup, never infer free space from absent routes. Reaching the ledger limit refuses new connectors; sharding is future design work. Retain an allocation until explicit operator retirement, across disconnects, credential rotation, expiration, pod restart and HTTP/2 fallback. Retirement revokes the grant and waits for the route to be released and its old gates fenced. For this milestone, retired /96s and attachment pools remain tombstoned and are **never reassigned** within that VPC/cell. This avoids an arbitrary time-based claim that delayed traffic has disappeared. Restore the ledger and grants together; rollback must not reuse allocation identities or credential revisions. diff --git a/docs/edge/README.md b/docs/edge/README.md index 94827f6f..0841ec3e 100644 --- a/docs/edge/README.md +++ b/docs/edge/README.md @@ -5,6 +5,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. +- [MASQUE grant authority](masque-authority.md): connector credentials, allocation ledgers, EdgeService issue, refresh and revocation, and the operator commands. - [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-authority.md b/docs/edge/masque-authority.md new file mode 100644 index 00000000..c161d52c --- /dev/null +++ b/docs/edge/masque-authority.md @@ -0,0 +1,115 @@ +# MASQUE grant authority + +`internal/edge/masque/authority` and `cmd/masque-authority` authenticate MASQUE connector credentials and issue the finite `EdgeService` of each connector authorization. The [MASQUE provider adapter](masque-provider.md) consumes them through `authority.Client`, its `masque.Authorizer`. The design is in the [MASQUE connector contract](../designs/masque/contract.md#authority-and-durable-allocations). This page covers what is built and how to operate it. No image, manifest or published release ships it yet. + +The authority runs under its own ServiceAccount. It holds no provider token and no host access. The provider can't write grant records, ledgers or EdgeServices; it receives each grant from the authority. + +## State + +All state is in the API. The authority keeps nothing in memory between calls, so a restarted or second instance rereads everything before it issues anything. + +| Object | Where | Written by | Holds | +| --------------------------------------- | ------------------------------ | ---------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| Grant record `masque-grant-` | Secret, authority namespace | `grant` commands | Connector UUID, cell, VPC namespace/name/UID, credential revision, SHA-256 digest of the credential, expiry (at most 24 h), enabled flag, at most 64 allowed destinations, traffic policy | +| Ledger `masque-ledger--` | ConfigMap, authority namespace | `ledger init`, `grant` | The VPC's connector /80 and pool /80, and one entry per connector: allocation UUID, /96, /120 pool, creation time, retired flag | +| `EdgeService masque-` | The VPC's namespace | `serve` | One authorization incarnation: the /96 as its only prefix, the /120 as `clientPool`, `validUntil`, and annotations naming the authorization, pod UID, revision, nonce and cell | + +The credential itself is never stored, logged or returned after the command that created it. A ledger holds at most 256 entries. Every ledger and record write carries the read's UID and resourceVersion, so a concurrent writer or a recreated object makes the write fail rather than overwrite. A retired ledger entry is a tombstone: its /96 and pool are never allocated again. A missing ledger refuses every allocation until the operator restores it. + +## Calls + +Provider pods call `serve` over HTTPS (TLS 1.3) at `POST /v1/authorize`, `/v1/refresh` and `/v1/retire`. Each call presents the pod's projected ServiceAccount token for audience `masque-authority`. The authority reviews it with a `TokenReview` and accepts only a token bound to one pod of `--provider-service-account`. Each call has a 2 s budget, and every read it makes is uncached. + +| Call | Succeeds when | Effect | +| ----------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| `authorize` | The credential's digest matches one record (compared in constant time against every record), the credential is unexpired, the record is enabled and in this cell, the VPC selector `namespace/name` matches, the VPC has the record's UID, and the ledger entry is live | Returns the pod's current incarnation, renewed, if it has one under this revision. Otherwise retires every other incarnation of the connector and issues a new one once core has released them | +| `refresh` | The incarnation is the calling pod's, unretired and unexpired, issued under the record's current revision, and the record, VPC and ledger entry still pass | Moves `validUntil` forward. A refusal also retires the incarnation | +| `retire` | The incarnation is the calling pod's | Marks it retired and sets `validUntil` in the past. Retiring one already deleted succeeds | + +An issued or renewed `validUntil` is the credential's expiry or the grant read's start plus 28 s (30 s less a 2 s issuer allowance), whichever is earlier, rounded down to the second. Core and galactic-edge stop the connector 2 s (MaxSkew) before it. The adapter refreshes every 5 s, so a renewal comes well before that cutoff. Renewing in place is a new EdgeService generation, and core keeps the attachment, route and epoch across it (`TestAgainstCore/renewing_the_service_keeps_forwarding`). + +Answers map to HTTP status as follows. 401 is unauthenticated: an unknown, malformed or expired credential, or a caller that is not a provider pod. 403 is forbidden: a disabled connector, another VPC or cell, a recreated VPC, a retired allocation or authorization, or a stale revision. 400 is a malformed call. 503 means the authority couldn't decide now and issued nothing. + +### Retries and replacement + +Each `authorize` call carries a random nonce and the time it was first sent. A retry resends both. The authority refuses a call sent more than 10 s ago (`RetryWindow`) or more than 2 s in the future. A retired incarnation is deleted only after `RetryWindow` has passed and core has released every client-pool address it held (`status.allocations` empty). A retry therefore either finds the incarnation it created, or is refused as stale. It never creates a second one. + +A connector's pool is the same in every incarnation, and core refuses the attachments of the later of two services with overlapping pools. Replacement is therefore sequential: + +1. Retire the old incarnation, which ends its forwarding at core and galactic-edge. +2. Wait for core to release its attachments and for `RetryWindow` to pass. +3. Delete it. +4. Issue the successor. + +Until then, `authorize` answers 503 and the client retries. An incarnation is named for its connector, and the authority creates one only after every other incarnation of the connector is gone. Pods racing to issue therefore contend for one name: the API server accepts one create, and the others get 503. The UID tells successive incarnations of one name apart. After each create or renewal, the authority rereads the grant record. If the record was rotated, disabled or deleted in the meantime, it retires the incarnation and answers 403. + +### Revocation + +`serve` sweeps every second (`--sweep-interval`), and only touches incarnations its own cell issued. An unretired incarnation is retired when: + +- its record is deleted, disabled or expired; +- its record is at another revision; or +- its `validUntil` has passed without renewal. + +Retired incarnations that core has released are deleted after `RetryWindow`. The provider learns of a revocation from its next `refresh` (403), at most 5 s later. The service's `validUntil` is the backstop if the provider, the authority or the API stalls: no packets 31 s after a committed revocation, or 1 s after credential expiry. Revocation push notifications are not built. Revoking one connector never touches another connector's incarnation, and the shared instance Lease keeps renewing. + +## Operator tasks + +All commands take `--namespace` (or `POD_NAMESPACE`) for the authority namespace and use the caller's kubeconfig. + +```sh +# Once per VPC and cell: reserve a connector /80 and a pool /80 inside the VPC. +masque-authority ledger init --vpc tenant/vpc --cell cell-1 \ + --connectors fd20:a1b:2c3d:10:1::/80 --pools fd20:a1b:2c3d:10:2::/80 + +# Create a test connector. The credential is printed once, as JSON. +masque-authority grant create --vpc tenant/vpc --cell cell-1 --lifetime 12h \ + --route fd20:a1b:2c3d::/48 --ingress-policy allow-in --egress-policy allow-out + +masque-authority grant rotate # new credential, next revision +masque-authority grant disable # revoke; the allocation is kept +masque-authority grant retire # revoke and tombstone the allocation +``` + +`ledger init` refuses ranges that aren't masked IPv6 /80s inside one of the VPC's networks, or that overlap each other. It also refuses ranges that overlap those of another cell's ledger for the same VPC, so two cells are never handed overlapping /80s. And it refuses while any EdgeService the authority issued for the VPC in that cell still exists, so a recreated ledger can't hand out a /96 or pool that one still holds. Issued services of other cells don't block it. Choose ranges that overlap no workload subnet; core refuses a delegated route that does. + +Run the server with: + +```sh +masque-authority serve --provider masque --provider-service-account edge/masque --cell cell-1 \ + --tls-cert /etc/masque-authority/tls/tls.crt --tls-key /etc/masque-authority/tls/tls.key +``` + +It rereads the certificate and key when either file changes. `authority.Client` rereads its CA bundle when it changes, so roots can rotate with an overlap and no restart. + +The authority's ServiceAccount needs these permissions: + +- get, list, create and update on Secrets and ConfigMaps in its namespace; +- get, list, create, update and delete on `edgeservices`, and get on `vpcs`, in each tenant namespace it serves; +- list on `edgeservices` across the cluster, for the sweep; +- create on `tokenreviews`. + +The provider's ServiceAccount needs none of these. + +The conformance fixture uses the authority when its configuration sets `grantAuthority` (`url`, `caFile`, `tokenFile`) in place of `services` and `connectors`. + +## Verification + +- `go test ./internal/edge/masque/authority/` runs the ledger unit tests. With `KUBEBUILDER_ASSETS` set, `TestAuthority` runs against a real API server. It covers: + - admission and every refusal; + - retries and the retry window; + - takeover after release, and racing pods; + - rotation, credential expiry, revocation and the sweep; + - an unavailable store; + - recreated VPCs and ledgers; + - credentials staying out of logs. + + `TestOverHTTPS` runs `authority.Client` against `Server` on TLS 1.3. +- `go test ./internal/edge/masque/ -run TestAgainstCore` includes the in-place renewal check against core. + +Not yet covered: + +- the HTTP/2 transport ([#981](https://github.com/datum-cloud/galactic/issues/981)); +- refusing tenant packets and MTU probes before admission ([#980](https://github.com/datum-cloud/galactic/issues/980), [#981](https://github.com/datum-cloud/galactic/issues/981)); +- deployment manifests and an image ([#980](https://github.com/datum-cloud/galactic/issues/980)); +- lab or target-platform measurements of the revocation and expiry bounds ([#985](https://github.com/datum-cloud/galactic/issues/985)). diff --git a/docs/edge/masque-provider.md b/docs/edge/masque-provider.md index a5e39c7b..662ca33c 100644 --- a/docs/edge/masque-provider.md +++ b/docs/edge/masque-provider.md @@ -1,6 +1,6 @@ # 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. +`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 is a separate deliverable of the [MASQUE connector design](../designs/masque/contract.md). The [grant authority](masque-authority.md) issues its grants, and `authority.Client` is its production `Authorizer`. The conformance fixture in `tests/edgeconformance/masqueprovider` runs it with static test credentials, or with the grant authority when configured. 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. @@ -9,7 +9,7 @@ The adapter never touches host networking. galactic-edge programs and releases e 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. +2. Grant each connector an `EdgeService` in its VPC's namespace. The [grant authority](masque-authority.md) issues one per connector authorization. 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. The `Authorizer` also refreshes an open request's grant (`Refresh`) and retires a grant (`Retire`); see [Grant refresh and revocation](#grant-refresh-and-revocation). @@ -45,7 +45,7 @@ A request forwards a packet, in either direction, only while its instance Lease ## Grant refresh and revocation -Every 5 seconds (`Config.Refresh`) the adapter calls `Authorizer.Refresh` for each open request, with a 2 second deadline. A refreshed grant with the same connector, VPC, service, /96 and routes moves the request's deadline. Any other answer has one of these results: +Every 5 seconds (`Config.Refresh`) the adapter calls `Authorizer.Refresh` for each open request, with a 2 second deadline. A refreshed grant with the same connector, VPC, service, /96, routes, authorization and credential revision moves the request's deadline. Any other answer has one of these results: - Unauthenticated or forbidden: the connector's grant was revoked, retired or rotated, and only that connector's request ends. - A changed grant: the request ends. diff --git a/internal/edge/masque/adapter_envtest_test.go b/internal/edge/masque/adapter_envtest_test.go index f4bfbdb4..a3e860ba 100644 --- a/internal/edge/masque/adapter_envtest_test.go +++ b/internal/edge/masque/adapter_envtest_test.go @@ -591,6 +591,45 @@ func TestAgainstCore(t *testing.T) { t.Fatal("one connector's service expiring changed the shared instance's lease") } }) + t.Run("renewing the service keeps forwarding", func(t *testing.T) { + // The grant authority renews a connector's service by moving its + // validUntil forward in place. Each renewal is a new generation; + // the connector keeps its attachment, route and epoch past the + // first deadline. + svc := f.service("renewed", "fd00:10:0:e::600/120", Capabilities, "fd00:10:0:1:8::/96") + renew := func(ahead time.Duration) { + t.Helper() + if err := f.admin.Get(context.Background(), client.ObjectKeyFromObject(svc), svc); err != nil { + t.Fatal(err) + } + svc.Spec.ValidUntil = &metav1.Time{Time: time.Now().Add(shortParams.MaxSkew + ahead)} + if err := f.admin.Update(context.Background(), svc); err != nil { + t.Fatal(err) + } + } + // Admission takes a few seconds; validUntil carries whole seconds. + renew(6 * time.Second) + first, deadline := svc.Generation, svc.Spec.ValidUntil.Time + f.connector("c8", svc, "fd00:10:0:1:8::/96") + l := f.admit(a, "c8") + defer func() { _ = l.Close() }() + attUID := f.attachment(l.Info().Attachment).UID + epoch := f.route(l.Info().Claim).Spec.Claim.Epoch + for time.Now().Before(deadline.Add(2 * time.Second)) { + time.Sleep(time.Second) + renew(4 * time.Second) + } + expectLive(t, l, "renewing the connector's service") + att := f.attachment(l.Info().Attachment) + rt := f.route(l.Info().Claim) + if svc.Generation <= first+2 || att == nil || att.UID != attUID || + !meta.IsStatusConditionTrue(att.Status.Conditions, galacticv1alpha1.EdgeAttachmentAdmitted) || + rt.Spec.Claim.Released || rt.Spec.Claim.Epoch != epoch || + !acceptedFor(rt, l.Info().Attachment, attUID, mustLease(t, a)) { + t.Fatalf("after renewals to generation %d: attachment %+v, claim %+v", svc.Generation, att, rt.Spec.Claim) + } + expectLive(t, l2, "renewing another connector's service") + }) t.Run("replacement keeps the claim", func(t *testing.T) { svc := f.service("handoff", "fd00:10:0:e::500/120", Capabilities, "fd00:10:0:1:7::/96") f.connector("c7", svc, "fd00:10:0:1:7::/96") diff --git a/internal/edge/masque/authority/authority.go b/internal/edge/masque/authority/authority.go new file mode 100644 index 00000000..d358a372 --- /dev/null +++ b/internal/edge/masque/authority/authority.go @@ -0,0 +1,677 @@ +// Copyright 2026 Datum Cloud, Inc. +// +// SPDX-License-Identifier: AGPL-3.0-or-later + +package authority + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "errors" + "fmt" + "log/slog" + "net/netip" + "regexp" + "strconv" + "time" + + corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "sigs.k8s.io/controller-runtime/pkg/client" + + cloudv1alpha1 "go.datum.net/galactic/api/cloud/v1alpha1" + galacticv1alpha1 "go.datum.net/galactic/api/v1alpha1" + "go.datum.net/galactic/internal/edge/admission" + "go.datum.net/galactic/internal/edge/connectip" +) + +// Errors the authority's decisions wrap. connectip.ErrUnauthenticated and +// connectip.ErrForbidden mark refusals; ErrUnavailable marks a decision the +// authority could not make now, which issues nothing. +var ( + ErrUnavailable = errors.New("authority unavailable") + ErrInvalid = errors.New("invalid request") +) + +const ( + // LabelManagedBy marks the EdgeServices the authority issued. + LabelManagedBy = "app.kubernetes.io/managed-by" + // ManagedBy is LabelManagedBy's value. + ManagedBy = "masque-authority" + + annotationPrefix = "masque.galactic.datumapis.com/" + // AnnotationAuthorization is an issued EdgeService's authorization ID. + AnnotationAuthorization = annotationPrefix + "authorization" + // AnnotationPod is the UID of the provider pod the authorization is + // bound to. + AnnotationPod = annotationPrefix + "pod-uid" + // AnnotationRevision is the credential revision it was issued under. + AnnotationRevision = annotationPrefix + "revision" + // AnnotationNonce is the nonce of the Authorize call that issued it. + AnnotationNonce = annotationPrefix + "nonce" + // AnnotationAllocation is the ledger allocation it grants. + AnnotationAllocation = annotationPrefix + "allocation" + // AnnotationRetired marks a retired authorization, which is never + // renewed again. + AnnotationRetired = annotationPrefix + "retired" + // AnnotationRetiredAt is when the authorization was retired. + AnnotationRetiredAt = annotationPrefix + "retired-at" + // AnnotationCell is the cell of the authority that issued it. + AnnotationCell = annotationPrefix + "cell" + + // packetAccess and prefixReachability are the capabilities an issued + // EdgeService grants. + packetAccess = "PacketAccess" + prefixReachability = "PrefixReachability" + + // DefaultTTL is how long after a grant read's start an issued + // validUntil may be, before IssuerSkew is subtracted. + DefaultTTL = 30 * time.Second + // DefaultIssuerSkew is the issuer clock allowance subtracted from TTL. + DefaultIssuerSkew = 2 * time.Second + // RetryWindow is how long after it was issued an Authorize call is + // accepted. A retired authorization is kept at least this long, so a + // retry of the call that issued it finds it and is refused. + RetryWindow = 10 * time.Second + // maxCallerSkew is how far ahead of the authority's clock a call's + // issue time may be. + maxCallerSkew = 2 * time.Second +) + +// nonceRE is the form of an Authorize nonce. +var nonceRE = regexp.MustCompile(`^[A-Za-z0-9]{16,64}$`) + +// ServiceRef identifies an issued EdgeService. +type ServiceRef struct { + Namespace string `json:"namespace"` + Name string `json:"name"` + UID string `json:"uid"` +} + +// Grant is an issued authorization, as the provider receives it. +type Grant struct { + Authorization string `json:"authorization"` + Connector string `json:"connector"` + Allocation string `json:"allocation"` + VPC string `json:"vpc"` + VPCUID string `json:"vpcUID"` + VPCID string `json:"vpcID"` + Service ServiceRef `json:"service"` + Prefix netip.Prefix `json:"prefix"` + Routes []netip.Prefix `json:"routes"` + Revision int64 `json:"revision"` + NotAfter time.Time `json:"notAfter"` +} + +// Authority decides Authorize, Refresh and Retire calls of the provider pods +// of one EdgeProvider in one cell, and sweeps revoked authorizations. +type Authority struct { + // Client reads without a cache: every decision rests on fresh reads. + Client client.Client + // Namespace holds the grant record Secrets and ledger ConfigMaps. + Namespace string + // Provider is the EdgeProvider the issued services grant. + Provider string + // ProviderServiceAccount is the only ServiceAccount whose pods may + // call. + ProviderServiceAccount admission.ServiceAccount + // Cell is this authority's cell. + Cell string + // TTL and IssuerSkew bound an issued validUntil; zero means the + // defaults. + TTL, IssuerSkew time.Duration + Now func() time.Time + Log *slog.Logger +} + +func (a *Authority) now() time.Time { + if a.Now != nil { + return a.Now() + } + return time.Now() +} + +func (a *Authority) log() *slog.Logger { + if a.Log != nil { + return a.Log + } + return slog.Default() +} + +func (a *Authority) ledgers() Ledgers { + return Ledgers{Client: a.Client, Namespace: a.Namespace, Now: a.now} +} + +// deadline is the validUntil of a grant read that started at start: the +// credential's expiry or start plus TTL less IssuerSkew, whichever is +// earlier, rounded down to the second an EdgeService can carry. +func (a *Authority) deadline(start time.Time, r *Record) time.Time { + ttl, skew := a.TTL, a.IssuerSkew + if ttl == 0 { + ttl = DefaultTTL + } + if skew == 0 { + skew = DefaultIssuerSkew + } + until := start.Add(ttl - skew) + if r.Expires.Before(until) { + until = r.Expires + } + return until.UTC().Truncate(time.Second) +} + +// caller refuses an identity that is not a pod of the provider's +// ServiceAccount. +func (a *Authority) caller(id admission.Identity) error { + if id.PodUID == "" || id.ServiceAccount != a.ProviderServiceAccount { + return fmt.Errorf("%w: %s is not the provider's ServiceAccount", connectip.ErrUnauthenticated, + id.ServiceAccount) + } + return nil +} + +// Authorize authenticates req's credential, authorizes it for req's VPC +// selector and returns the connector's grant, bound to the calling pod. A +// call issued more than RetryWindow ago is refused. A retry with the same +// nonce returns the grant the first call issued, with its original +// deadline, and never a new one. A pod that already holds a current grant +// of the same credential revision gets that grant renewed. Any other grant +// of the connector is retired first, and a new one is issued only once core +// released its attachments. +// +// It returns an error wrapping connectip.ErrUnauthenticated for an unknown +// or expired credential or a caller that is not a provider pod, +// connectip.ErrForbidden for a disabled connector, another VPC or cell, a +// recreated VPC or a retired allocation or authorization, ErrInvalid for a +// malformed request, and ErrUnavailable when it cannot decide now. +func (a *Authority) Authorize(ctx context.Context, id admission.Identity, req AuthorizeRequest) (Grant, error) { + if err := a.caller(id); err != nil { + return Grant{}, err + } + vpc, nonce := req.VPC, req.Nonce + if !nonceRE.MatchString(nonce) { + return Grant{}, fmt.Errorf("%w: malformed nonce", ErrInvalid) + } + start := a.now() + if req.Issued.Before(start.Add(-RetryWindow)) || req.Issued.After(start.Add(maxCallerSkew)) { + return Grant{}, fmt.Errorf("%w: call issued at %v is outside the retry window", ErrInvalid, req.Issued) + } + r, err := a.match(ctx, req.Credential) + if err != nil { + return Grant{}, err + } + if err := a.current(r, start); err != nil { + return Grant{}, err + } + if vpc != r.VPC.Selector() { + return Grant{}, fmt.Errorf("%w: connector %s may not use VPC %q", connectip.ErrForbidden, r.Connector, vpc) + } + vpcID, err := a.vpc(ctx, r.VPC) + if err != nil { + return Grant{}, err + } + e, err := a.allocation(ctx, r) + if err != nil { + return Grant{}, err + } + var list galacticv1alpha1.EdgeServiceList + if err := a.Client.List(ctx, &list, client.InNamespace(r.VPC.Namespace), + client.MatchingLabels{LabelConnector: r.Connector, LabelManagedBy: ManagedBy}); err != nil { + return Grant{}, fmt.Errorf("%w: list services of connector %s: %w", ErrUnavailable, r.Connector, err) + } + for i := range list.Items { + s := &list.Items[i] + if s.Annotations[AnnotationNonce] == nonce && s.Annotations[AnnotationPod] == id.PodUID { + if retired(s) || expired(s, a.now()) || + s.Annotations[AnnotationRevision] != strconv.FormatInt(r.Revision, 10) || + s.Annotations[AnnotationAllocation] != e.Allocation { + return Grant{}, fmt.Errorf("%w: authorization %s is retired, expired or stale", connectip.ErrForbidden, + s.Annotations[AnnotationAuthorization]) + } + return a.grant(s, r, e, vpcID) + } + } + for i := range list.Items { + s := &list.Items[i] + if !retired(s) && s.Annotations[AnnotationPod] == id.PodUID && + s.Annotations[AnnotationRevision] == strconv.FormatInt(r.Revision, 10) && !expired(s, a.now()) { + if err := a.renew(ctx, s, a.deadline(start, r)); err != nil { + return Grant{}, err + } + return a.grant(s, r, e, vpcID) + } + } + // Every other incarnation of the connector ends before a new one + // exists: its pool is the new one's, and pools of one VPC may not + // overlap. + draining := 0 + for i := range list.Items { + s := &list.Items[i] + if !retired(s) { + if err := a.retire(ctx, s, "superseded"); err != nil { + return Grant{}, err + } + } + if released, err := a.reap(ctx, s); err != nil || !released { + draining++ + } + } + if draining > 0 { + return Grant{}, fmt.Errorf("%w: connector %s's previous authorization is still being released", + ErrUnavailable, r.Connector) + } + return a.issue(ctx, id, r, e, vpcID, nonce, a.deadline(start, r)) +} + +// match returns the record whose digest is credential's, comparing every +// record in constant time. +func (a *Authority) match(ctx context.Context, credential string) (*Record, error) { + if !wellFormed(credential) { + return nil, fmt.Errorf("%w: malformed credential", connectip.ErrUnauthenticated) + } + digest := Digest(credential) + var list corev1.SecretList + if err := a.Client.List(ctx, &list, client.InNamespace(a.Namespace), + client.MatchingLabels{LabelGrant: yes}); err != nil { + return nil, fmt.Errorf("%w: list grant records: %w", ErrUnavailable, err) + } + var found *Record + for i := range list.Items { + r, err := decodeRecord(&list.Items[i]) + if err != nil { + a.log().Warn("skipping malformed grant record", "secret", list.Items[i].Name, "error", err) + continue + } + if sameDigest(r.Digest, digest) { + found = r + } + } + if found == nil { + return nil, fmt.Errorf("%w: unknown credential", connectip.ErrUnauthenticated) + } + return found, nil +} + +// current refuses a record that cannot authorize anything at now. +func (a *Authority) current(r *Record, now time.Time) error { + switch { + case !now.Before(r.Expires): + return fmt.Errorf("%w: connector %s's credential expired", connectip.ErrUnauthenticated, r.Connector) + case !r.Enabled: + return fmt.Errorf("%w: connector %s is disabled", connectip.ErrForbidden, r.Connector) + case r.Cell != a.Cell: + return fmt.Errorf("%w: connector %s belongs to cell %s", connectip.ErrForbidden, r.Connector, r.Cell) + } + return nil +} + +// vpc reads the record's VPC and returns its base62 identifier. It refuses +// a VPC deleted or recreated since the record was written. +func (a *Authority) vpc(ctx context.Context, ref VPCRef) (string, error) { + var v cloudv1alpha1.VPC + switch err := a.Client.Get(ctx, client.ObjectKey{Namespace: ref.Namespace, Name: ref.Name}, &v); { + case apierrors.IsNotFound(err): + return "", fmt.Errorf("%w: VPC %s no longer exists", connectip.ErrForbidden, ref.Selector()) + case err != nil: + return "", fmt.Errorf("%w: read VPC %s: %w", ErrUnavailable, ref.Selector(), err) + case string(v.UID) != ref.UID || !v.DeletionTimestamp.IsZero(): + return "", fmt.Errorf("%w: VPC %s was replaced or is being deleted", connectip.ErrForbidden, ref.Selector()) + case v.Status.VPC == "": + return "", fmt.Errorf("%w: VPC %s has no fabric identifier yet", ErrUnavailable, ref.Selector()) + } + return v.Status.VPC, nil +} + +// allocation returns the record's ledger entry, refusing a retired or +// missing one. +func (a *Authority) allocation(ctx context.Context, r *Record) (Entry, error) { + l, _, err := a.ledgers().Read(ctx, r.VPC.UID, r.Cell) + if err != nil { + return Entry{}, fmt.Errorf("%w: %w", ErrUnavailable, err) + } + e, ok := l.Lookup(r.Connector) + switch { + case !ok: + return Entry{}, fmt.Errorf("%w: connector %s has no allocation", connectip.ErrForbidden, r.Connector) + case e.Retired: + return Entry{}, fmt.Errorf("%w: connector %s's allocation is retired", connectip.ErrForbidden, r.Connector) + } + return e, nil +} + +// authorizationID derives the authorization ID of an Authorize call, so a +// retry of it names the same EdgeService. +func authorizationID(connector string, revision int64, pod, nonce string) string { + sum := sha256.Sum256([]byte(connector + "\x00" + strconv.FormatInt(revision, 10) + "\x00" + pod + "\x00" + nonce)) + return hex.EncodeToString(sum[:12]) +} + +// ServiceName is the name of connector's EdgeService. A connector has at +// most one incarnation at a time, so concurrent issuers contend for one name +// and only one create succeeds; the UID tells successive incarnations apart. +func ServiceName(connector string) string { return "masque-" + connector } + +// issue creates the EdgeService of a new authorization incarnation. +func (a *Authority) issue(ctx context.Context, id admission.Identity, r *Record, e Entry, vpcID, nonce string, + until time.Time) (Grant, error) { + auth := authorizationID(r.Connector, r.Revision, id.PodUID, nonce) + s := &galacticv1alpha1.EdgeService{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: r.VPC.Namespace, Name: ServiceName(r.Connector), + Labels: map[string]string{LabelConnector: r.Connector, LabelManagedBy: ManagedBy}, + Annotations: map[string]string{ + AnnotationAuthorization: auth, AnnotationPod: id.PodUID, AnnotationNonce: nonce, + AnnotationRevision: strconv.FormatInt(r.Revision, 10), AnnotationAllocation: e.Allocation, + AnnotationCell: a.Cell, + }, + }, + Spec: galacticv1alpha1.EdgeServiceSpec{ + ProviderRef: galacticv1alpha1.EdgeProviderReference{Name: a.Provider}, + VPCRef: galacticv1alpha1.EdgeVPCReference{Name: r.VPC.Name, UID: r.VPC.UID}, + Capabilities: []string{packetAccess, prefixReachability}, + Prefixes: []string{e.Prefix.String()}, + ClientPool: e.Pool.String(), + ValidUntil: &metav1.Time{Time: until}, + TrafficPolicy: r.TrafficPolicy, + }, + } + if err := a.Client.Create(ctx, s); err != nil { + return Grant{}, fmt.Errorf("%w: create EdgeService %s/%s: %w", ErrUnavailable, s.Namespace, s.Name, err) + } + if err := a.unchanged(ctx, s, r); err != nil { + return Grant{}, err + } + a.log().Info("authorization issued", "authorization", auth, "connector", r.Connector, "revision", r.Revision, + "podRef", id.PodUID, "serviceName", s.Namespace+"/"+s.Name, "serviceUID", s.UID, "validUntil", until) + return a.grant(s, r, e, vpcID) +} + +// grant is the Grant of issued service s. +func (a *Authority) grant(s *galacticv1alpha1.EdgeService, r *Record, e Entry, vpcID string) (Grant, error) { + if s.Spec.ValidUntil == nil { + return Grant{}, fmt.Errorf("%w: EdgeService %s/%s has no validUntil", ErrUnavailable, s.Namespace, s.Name) + } + return Grant{ + Authorization: s.Annotations[AnnotationAuthorization], Connector: r.Connector, Allocation: e.Allocation, + VPC: r.VPC.Selector(), VPCUID: r.VPC.UID, VPCID: vpcID, + Service: ServiceRef{Namespace: s.Namespace, Name: s.Name, UID: string(s.UID)}, + Prefix: e.Prefix, Routes: r.Routes, Revision: r.Revision, NotAfter: s.Spec.ValidUntil.Time, + }, nil +} + +// Refresh revalidates the authorization auth of the calling pod, issued +// under credential revision revision as EdgeService ref, and renews its +// validUntil from this call's grant read. It never accepts a new credential +// revision and never renews an authorization that expired or was retired. +// A refusal retires the authorization. It returns the errors Authorize +// does. +func (a *Authority) Refresh(ctx context.Context, id admission.Identity, auth string, revision int64, + ref ServiceRef) (Grant, error) { + if err := a.caller(id); err != nil { + return Grant{}, err + } + start := a.now() + s, err := a.service(ctx, id, auth, ref) + if err != nil { + return Grant{}, err + } + if s.Annotations[AnnotationRevision] != strconv.FormatInt(revision, 10) { + return Grant{}, fmt.Errorf("%w: authorization %s was not issued under revision %d", connectip.ErrForbidden, + auth, revision) + } + if retired(s) || expired(s, start) { + if err := a.retire(ctx, s, "expired"); err != nil { + a.log().Warn("could not retire an expired authorization", "authorization", auth, "error", err) + } + return Grant{}, fmt.Errorf("%w: authorization %s is retired or expired", connectip.ErrForbidden, auth) + } + g, err := a.revalidate(ctx, s, revision, start) + if err != nil { + if !errors.Is(err, ErrUnavailable) { + if rerr := a.retire(ctx, s, err.Error()); rerr != nil { + a.log().Warn("could not retire a refused authorization", "authorization", auth, "error", rerr) + } + } + return Grant{}, err + } + return g, nil +} + +// revalidate rereads the record, VPC and allocation of issued service s and +// renews its validUntil. +func (a *Authority) revalidate(ctx context.Context, s *galacticv1alpha1.EdgeService, revision int64, + start time.Time) (Grant, error) { + connector := s.Labels[LabelConnector] + r, _, err := readRecord(ctx, a.Client, a.Namespace, connector) + switch { + case apierrors.IsNotFound(err): + return Grant{}, fmt.Errorf("%w: connector %s has no grant record", connectip.ErrForbidden, connector) + case err != nil: + return Grant{}, fmt.Errorf("%w: read grant record of %s: %w", ErrUnavailable, connector, err) + } + if err := a.current(r, start); err != nil { + return Grant{}, err + } + if r.Revision != revision || r.VPC.UID != s.Spec.VPCRef.UID { + return Grant{}, fmt.Errorf("%w: connector %s's grant changed", connectip.ErrForbidden, connector) + } + vpcID, err := a.vpc(ctx, r.VPC) + if err != nil { + return Grant{}, err + } + e, err := a.allocation(ctx, r) + if err != nil { + return Grant{}, err + } + if e.Allocation != s.Annotations[AnnotationAllocation] || len(s.Spec.Prefixes) != 1 || + s.Spec.Prefixes[0] != e.Prefix.String() || s.Spec.ClientPool != e.Pool.String() { + return Grant{}, fmt.Errorf("%w: connector %s's allocation changed", connectip.ErrForbidden, connector) + } + if err := a.renew(ctx, s, a.deadline(start, r)); err != nil { + return Grant{}, err + } + if err := a.unchanged(ctx, s, r); err != nil { + return Grant{}, err + } + return a.grant(s, r, e, vpcID) +} + +// unchanged rereads the record s was just issued or renewed under, and +// retires s if the record changed since r was read: a rotation, disable or +// deletion that raced the write never leaves s live. +func (a *Authority) unchanged(ctx context.Context, s *galacticv1alpha1.EdgeService, r *Record) error { + cur, _, err := readRecord(ctx, a.Client, a.Namespace, r.Connector) + if err == nil && cur.Revision == r.Revision && cur.Enabled && sameDigest(cur.Digest, r.Digest) { + return nil + } + if rerr := a.retire(ctx, s, "grant changed during issue"); rerr != nil { + a.log().Warn("could not retire an authorization whose grant changed", "serviceName", s.Name, "error", rerr) + } + if err != nil && !apierrors.IsNotFound(err) { + return fmt.Errorf("%w: reread grant record of %s: %w", ErrUnavailable, r.Connector, err) + } + return fmt.Errorf("%w: connector %s's grant changed", connectip.ErrForbidden, r.Connector) +} + +// service reads the issued EdgeService ref and checks it is authorization +// auth of the calling pod. +func (a *Authority) service(ctx context.Context, id admission.Identity, auth string, + ref ServiceRef) (*galacticv1alpha1.EdgeService, error) { + var s galacticv1alpha1.EdgeService + switch err := a.Client.Get(ctx, client.ObjectKey{Namespace: ref.Namespace, Name: ref.Name}, &s); { + case apierrors.IsNotFound(err): + return nil, fmt.Errorf("%w: authorization %s no longer exists", connectip.ErrForbidden, auth) + case err != nil: + return nil, fmt.Errorf("%w: read EdgeService %s/%s: %w", ErrUnavailable, ref.Namespace, ref.Name, err) + case string(s.UID) != ref.UID || s.Labels[LabelManagedBy] != ManagedBy || + s.Annotations[AnnotationAuthorization] != auth || s.Annotations[AnnotationPod] != id.PodUID || + !s.DeletionTimestamp.IsZero(): + return nil, fmt.Errorf("%w: EdgeService %s/%s is not authorization %s of this pod", connectip.ErrForbidden, + ref.Namespace, ref.Name, auth) + } + return &s, nil +} + +// renew sets s's validUntil to until, conditioned on the resourceVersion +// read, so a concurrent retirement wins. +func (a *Authority) renew(ctx context.Context, s *galacticv1alpha1.EdgeService, until time.Time) error { + if s.Spec.ValidUntil != nil && s.Spec.ValidUntil.Time.Equal(until) { + return nil + } + s.Spec.ValidUntil = &metav1.Time{Time: until} + if err := a.Client.Update(ctx, s); err != nil { + return fmt.Errorf("%w: renew EdgeService %s/%s: %w", ErrUnavailable, s.Namespace, s.Name, err) + } + return nil +} + +// Retire ends the authorization auth of the calling pod, issued as +// EdgeService ref: it is never renewed again, and its validUntil is set in +// the past so core revokes what it admitted. Retiring a retired or deleted +// authorization succeeds. +func (a *Authority) Retire(ctx context.Context, id admission.Identity, auth string, ref ServiceRef) error { + if err := a.caller(id); err != nil { + return err + } + var cur galacticv1alpha1.EdgeService + err := a.Client.Get(ctx, client.ObjectKey{Namespace: ref.Namespace, Name: ref.Name}, &cur) + if apierrors.IsNotFound(err) || (err == nil && string(cur.UID) != ref.UID) { + // The incarnation is already gone. + return nil + } + s, err := a.service(ctx, id, auth, ref) + if err != nil { + return err + } + return a.retire(ctx, s, "retired by the provider") +} + +// retire marks s retired and moves its validUntil into the past, retrying +// from a fresh read after a conflict. +func (a *Authority) retire(ctx context.Context, s *galacticv1alpha1.EdgeService, reason string) error { + for range conflictRetries { + if retired(s) { + return nil + } + if s.Annotations == nil { + s.Annotations = map[string]string{} + } + s.Annotations[AnnotationRetired] = yes + s.Annotations[AnnotationRetiredAt] = a.now().UTC().Format(time.RFC3339Nano) + past := a.now().Add(-time.Minute).UTC().Truncate(time.Second) + if s.Spec.ValidUntil == nil || s.Spec.ValidUntil.After(past) { + s.Spec.ValidUntil = &metav1.Time{Time: past} + } + err := a.Client.Update(ctx, s) + if err == nil { + a.log().Info("authorization retired", "authorization", s.Annotations[AnnotationAuthorization], + "connector", s.Labels[LabelConnector], "serviceName", s.Namespace+"/"+s.Name, "reason", reason) + return nil + } + if !apierrors.IsConflict(err) { + return fmt.Errorf("%w: retire EdgeService %s/%s: %w", ErrUnavailable, s.Namespace, s.Name, err) + } + uid := s.UID + if err := a.Client.Get(ctx, client.ObjectKeyFromObject(s), s); err != nil || s.UID != uid { + return fmt.Errorf("%w: reread EdgeService %s/%s: %w", ErrUnavailable, s.Namespace, s.Name, err) + } + } + return fmt.Errorf("%w: EdgeService %s/%s kept changing", ErrUnavailable, s.Namespace, s.Name) +} + +// reap deletes retired service s once it was retired for RetryWindow and +// core holds no client pool address for it, conditioned on its UID. It +// reports whether s is gone. +func (a *Authority) reap(ctx context.Context, s *galacticv1alpha1.EdgeService) (bool, error) { + if !retired(s) || len(s.Status.Allocations) > 0 { + return false, nil + } + at, err := time.Parse(time.RFC3339Nano, s.Annotations[AnnotationRetiredAt]) + if err == nil && a.now().Before(at.Add(RetryWindow)) { + return false, nil + } + err = a.Client.Delete(ctx, s, client.Preconditions{UID: &s.UID}) + if err != nil && !apierrors.IsNotFound(err) { + return false, err + } + return true, nil +} + +// Sweep retires every issued authorization whose grant record is gone, +// disabled, expired or at another revision, or whose validUntil passed +// unrenewed, and deletes retired ones core has released. It returns the +// errors it met. +func (a *Authority) Sweep(ctx context.Context) error { + var records corev1.SecretList + if err := a.Client.List(ctx, &records, client.InNamespace(a.Namespace), + client.MatchingLabels{LabelGrant: yes}); err != nil { + return fmt.Errorf("list grant records: %w", err) + } + byConnector := map[string]*Record{} + for i := range records.Items { + if r, err := decodeRecord(&records.Items[i]); err == nil { + byConnector[r.Connector] = r + } + } + var services galacticv1alpha1.EdgeServiceList + if err := a.Client.List(ctx, &services, client.MatchingLabels{LabelManagedBy: ManagedBy}); err != nil { + return fmt.Errorf("list issued EdgeServices: %w", err) + } + now := a.now() + var errs []error + for i := range services.Items { + s := &services.Items[i] + if s.Spec.ProviderRef.Name != a.Provider || s.Annotations[AnnotationCell] != a.Cell { + continue + } + if !retired(s) { + r := byConnector[s.Labels[LabelConnector]] + reason := "" + switch { + case r == nil: + reason = "grant record deleted" + case a.current(r, now) != nil: + reason = "grant disabled or expired" + case strconv.FormatInt(r.Revision, 10) != s.Annotations[AnnotationRevision]: + reason = "credential rotated" + case expired(s, now): + reason = "not renewed" + } + if reason == "" { + continue + } + if err := a.retire(ctx, s, reason); err != nil { + errs = append(errs, err) + continue + } + } + if _, err := a.reap(ctx, s); err != nil { + errs = append(errs, fmt.Errorf("delete EdgeService %s/%s: %w", s.Namespace, s.Name, err)) + } + } + return errors.Join(errs...) +} + +// RunSweeps runs Sweep every interval until ctx ends. +func (a *Authority) RunSweeps(ctx context.Context, interval time.Duration) { + t := time.NewTicker(interval) + defer t.Stop() + for { + if err := a.Sweep(ctx); err != nil && ctx.Err() == nil { + a.log().Warn("sweep failed", "error", err) + } + select { + case <-ctx.Done(): + return + case <-t.C: + } + } +} + +func retired(s *galacticv1alpha1.EdgeService) bool { return s.Annotations[AnnotationRetired] == yes } + +func expired(s *galacticv1alpha1.EdgeService, now time.Time) bool { + return s.Spec.ValidUntil == nil || !now.Before(s.Spec.ValidUntil.Time) +} diff --git a/internal/edge/masque/authority/authority_envtest_test.go b/internal/edge/masque/authority/authority_envtest_test.go new file mode 100644 index 00000000..51c62bbb --- /dev/null +++ b/internal/edge/masque/authority/authority_envtest_test.go @@ -0,0 +1,749 @@ +// Copyright 2026 Datum Cloud, Inc. +// +// SPDX-License-Identifier: AGPL-3.0-or-later + +package authority + +import ( + "bytes" + "context" + "crypto/tls" + "encoding/pem" + "errors" + "fmt" + "log/slog" + "net/http/httptest" + "net/netip" + "os" + "path/filepath" + "strings" + "sync" + "testing" + "time" + + authenticationv1 "k8s.io/api/authentication/v1" + corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/envtest" + + cloudv1alpha1 "go.datum.net/galactic/api/cloud/v1alpha1" + galacticv1alpha1 "go.datum.net/galactic/api/v1alpha1" + "go.datum.net/galactic/internal/edge/admission" + "go.datum.net/galactic/internal/edge/connectip" + "go.datum.net/galactic/internal/edge/masque" +) + +const ( + testPodA = "pod-a" + testProvider = "masque" + testVPCID = "000000abc" + vpcSelector = tenantNS + "/" + vpcName +) + +var providerSA = admission.ServiceAccount{Namespace: "edge", Name: "masque"} + +func podID(uid string) admission.Identity { + return admission.Identity{ServiceAccount: providerSA, PodUID: uid} +} + +// fixture is a real API server with the CRDs, a VPC with a ledger, and an +// authority on a clock the test moves. +type fixture struct { + t *testing.T + admin client.Client + vpc *cloudv1alpha1.VPC + + mu sync.Mutex + now time.Time + + grants Grants + auth *Authority + logs *syncBuffer + // issued is when each nonce was first sent; a retry resends it. + issued map[string]time.Time +} + +func (f *fixture) clock() time.Time { + f.mu.Lock() + defer f.mu.Unlock() + return f.now +} + +func (f *fixture) advance(d time.Duration) { + f.mu.Lock() + defer f.mu.Unlock() + f.now = f.now.Add(d) +} + +// syncBuffer is a bytes.Buffer safe for concurrent writers. +type syncBuffer struct { + mu sync.Mutex + b bytes.Buffer +} + +func (s *syncBuffer) Write(p []byte) (int, error) { + s.mu.Lock() + defer s.mu.Unlock() + return s.b.Write(p) +} + +func (s *syncBuffer) String() string { + s.mu.Lock() + defer s.mu.Unlock() + return s.b.String() +} + +func startFixture(t *testing.T) *fixture { + t.Helper() + if os.Getenv("KUBEBUILDER_ASSETS") == "" { + if os.Getenv("CI") != "" { + t.Fatal("KUBEBUILDER_ASSETS not set in CI") + } + t.Skip("KUBEBUILDER_ASSETS not set") + } + te := &envtest.Environment{ + CRDDirectoryPaths: []string{filepath.Join("..", "..", "..", "..", "config", "crd")}, + ErrorIfCRDPathMissing: true, + } + cfg, err := te.Start() + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = te.Stop() }) + admin, err := client.New(cfg, client.Options{Scheme: testScheme()}) + if err != nil { + t.Fatal(err) + } + f := &fixture{t: t, admin: admin, now: time.Date(2026, 10, 27, 12, 0, 0, 0, time.UTC), logs: &syncBuffer{}, + issued: map[string]time.Time{}} + ctx := context.Background() + for _, ns := range []string{testNS, tenantNS} { + f.create(&corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: ns}}) + } + f.vpc = f.newVPC() + log := slog.New(slog.NewJSONHandler(f.logs, &slog.HandlerOptions{Level: slog.LevelDebug})) + f.grants = Grants{Client: admin, Namespace: testNS, Now: f.clock} + f.auth = &Authority{Client: admin, Namespace: testNS, Provider: testProvider, ProviderServiceAccount: providerSA, + Cell: testCell, Now: f.clock, Log: log} + ref := VPCRef{Namespace: tenantNS, Name: vpcName, UID: string(f.vpc.UID)} + if err := f.auth.ledgers().Init(ctx, ref, testCell, testConnectors, testPools); err != nil { + t.Fatal(err) + } + return f +} + +func (f *fixture) newVPC() *cloudv1alpha1.VPC { + f.t.Helper() + v := &cloudv1alpha1.VPC{ + ObjectMeta: metav1.ObjectMeta{Namespace: tenantNS, Name: vpcName}, + Spec: cloudv1alpha1.VPCSpec{Networks: []cloudv1alpha1.Network{"fd20:a1b:2c3d::/48"}}, + } + f.create(v) + v.Status.VPC = testVPCID + if err := f.admin.Status().Update(context.Background(), v); err != nil { + f.t.Fatal(err) + } + return v +} + +func (f *fixture) create(obj client.Object) { + f.t.Helper() + if err := f.admin.Create(context.Background(), obj); err != nil { + f.t.Fatalf("create %T %s: %v", obj, obj.GetName(), err) + } +} + +func (f *fixture) connector() Issued { + f.t.Helper() + is, err := f.grants.Create(context.Background(), CreateRequest{ + VPC: VPCRef{Namespace: tenantNS, Name: vpcName}, Cell: testCell, Lifetime: MaxLifetime, + Routes: []netip.Prefix{netip.MustParsePrefix("fd20:a1b:2c3d::/48")}, + }) + if err != nil { + f.t.Fatal(err) + } + return is +} + +// req is an Authorize call with nonce, issued when nonce was first used. +func (f *fixture) req(credential, vpc, nonce string) AuthorizeRequest { + at, ok := f.issued[nonce] + if !ok { + at = f.clock() + f.issued[nonce] = at + } + return AuthorizeRequest{Credential: credential, VPC: vpc, Nonce: nonce, Issued: at} +} + +func (f *fixture) tryAuthorize(id admission.Identity, credential, nonce string) (Grant, error) { + return f.auth.Authorize(context.Background(), id, f.req(credential, vpcSelector, nonce)) +} + +func (f *fixture) authorize(id admission.Identity, credential, nonce string) Grant { + f.t.Helper() + g, err := f.tryAuthorize(id, credential, nonce) + if err != nil { + f.t.Fatalf("Authorize: %v", err) + } + return g +} + +// service returns g's incarnation, or nil once it is gone, even when a +// successor took its name. +func (f *fixture) service(g Grant) *galacticv1alpha1.EdgeService { + f.t.Helper() + var s galacticv1alpha1.EdgeService + err := f.admin.Get(context.Background(), client.ObjectKey{Namespace: g.Service.Namespace, Name: g.Service.Name}, &s) + if apierrors.IsNotFound(err) || (err == nil && string(s.UID) != g.Service.UID) { + return nil + } + if err != nil { + f.t.Fatal(err) + } + return &s +} + +// deadline is the validUntil a grant read starting now gets without a +// credential expiry. +func (f *fixture) deadline() time.Time { + return f.clock().Add(DefaultTTL - DefaultIssuerSkew).Truncate(time.Second) +} + +func expectErr(t *testing.T, what string, err, want error) { + t.Helper() + if !errors.Is(err, want) { + t.Fatalf("%s = %v, want %v", what, err, want) + } +} + +func TestAuthority(t *testing.T) { //nolint:gocyclo // one ordered scenario over one API server + f := startFixture(t) + ctx := context.Background() + podA, podB := podID(testPodA), podID("pod-b") + c1 := f.connector() + g := f.authorize(podA, c1.Credential, "nonceA0000000001") + + t.Run("admits", func(t *testing.T) { + s := f.service(g) + if s == nil { + t.Fatal("no EdgeService issued") + } + want := galacticv1alpha1.EdgeServiceSpec{ + ProviderRef: galacticv1alpha1.EdgeProviderReference{Name: testProvider}, + VPCRef: galacticv1alpha1.EdgeVPCReference{Name: vpcName, UID: string(f.vpc.UID)}, + Capabilities: []string{packetAccess, prefixReachability}, + Prefixes: []string{c1.Prefix.String()}, ClientPool: "fd20:a1b:2c3d:10:2::100/120", + } + if s.Spec.ProviderRef != want.ProviderRef || s.Spec.VPCRef != want.VPCRef || + strings.Join(s.Spec.Capabilities, ",") != strings.Join(want.Capabilities, ",") || + strings.Join(s.Spec.Prefixes, ",") != c1.Prefix.String() || s.Spec.ClientPool != want.ClientPool { + t.Fatalf("issued spec %+v, want %+v", s.Spec, want) + } + if !s.Spec.ValidUntil.Time.Equal(f.deadline()) || !g.NotAfter.Equal(f.deadline()) { + t.Fatalf("validUntil %v, grant %v, want %v", s.Spec.ValidUntil, g.NotAfter, f.deadline()) + } + if g.Connector != c1.Connector || g.Prefix != c1.Prefix || g.Revision != 1 || g.VPCID != testVPCID || + g.VPCUID != string(f.vpc.UID) || g.VPC != vpcSelector || g.Service.UID != string(s.UID) || + s.Annotations[AnnotationPod] != testPodA || g.Authorization == "" { + t.Fatalf("grant %+v of service %+v", g, s.ObjectMeta) + } + }) + t.Run("retry keeps the original deadline", func(t *testing.T) { + f.advance(3 * time.Second) + r := f.authorize(podA, c1.Credential, "nonceA0000000001") + if r.Service != g.Service || !r.NotAfter.Equal(g.NotAfter) { + t.Fatalf("retry = %+v, want %+v unchanged", r, g) + } + }) + t.Run("same pod renews its current authorization", func(t *testing.T) { + r := f.authorize(podA, c1.Credential, "nonceA0000000002") + if r.Service != g.Service || !r.NotAfter.Equal(f.deadline()) { + t.Fatalf("second request = %+v, want %+v renewed to %v", r, g.Service, f.deadline()) + } + }) + t.Run("refresh renews", func(t *testing.T) { + f.advance(5 * time.Second) + r, err := f.auth.Refresh(ctx, podA, g.Authorization, g.Revision, g.Service) + if err != nil || !r.NotAfter.Equal(f.deadline()) || r.Service != g.Service { + t.Fatalf("Refresh = %+v, %v; want deadline %v", r, err, f.deadline()) + } + }) + t.Run("refusals", func(t *testing.T) { + otherSA := admission.Identity{ServiceAccount: admission.ServiceAccount{Namespace: "edge", Name: "x"}, + PodUID: testPodA} + unbound := admission.Identity{ServiceAccount: providerSA} + for name, tc := range map[string]struct { + id admission.Identity + cred, vpc, nonce string + want error + }{ + "unknown credential": {podA, NewCredential(), vpcSelector, "nonceR0000000001", connectip.ErrUnauthenticated}, + "malformed": {podA, "secret", vpcSelector, "nonceR0000000002", connectip.ErrUnauthenticated}, + "another VPC": {podA, c1.Credential, tenantNS + "/other", "nonceR0000000003", connectip.ErrForbidden}, + "another VPC's name": {podA, c1.Credential, "other/vpc", "nonceR0000000004", connectip.ErrForbidden}, + "not the provider SA": {otherSA, c1.Credential, vpcSelector, "nonceR0000000005", connectip.ErrUnauthenticated}, + "no pod": {unbound, c1.Credential, vpcSelector, "nonceR0000000006", connectip.ErrUnauthenticated}, + "bad nonce": {podA, c1.Credential, vpcSelector, "short", ErrInvalid}, + } { + _, err := f.auth.Authorize(ctx, tc.id, f.req(tc.cred, tc.vpc, tc.nonce)) + expectErr(t, name, err, tc.want) + } + for name, at := range map[string]time.Time{ + "stale call": f.clock().Add(-RetryWindow - time.Second), "future call": f.clock().Add(time.Minute), + "no issue time": {}, + } { + _, err := f.auth.Authorize(ctx, podA, AuthorizeRequest{Credential: c1.Credential, VPC: vpcSelector, + Nonce: "nonceS0000000001", Issued: at}) + expectErr(t, name, err, ErrInvalid) + } + for name, tc := range map[string]struct { + id admission.Identity + revision int64 + ref ServiceRef + }{ + "another pod": {podB, g.Revision, g.Service}, + "another revision": {podA, g.Revision + 1, g.Service}, + "another UID": {podA, g.Revision, ServiceRef{g.Service.Namespace, g.Service.Name, "x"}}, + "unknown service": {podA, g.Revision, ServiceRef{g.Service.Namespace, "masque-none", "x"}}, + } { + _, err := f.auth.Refresh(ctx, tc.id, g.Authorization, tc.revision, tc.ref) + expectErr(t, "Refresh from "+name, err, connectip.ErrForbidden) + } + if s := f.service(g); s == nil || retired(s) { + t.Fatal("a refused stale Refresh retired the current authorization") + } + }) + t.Run("store unavailable issues nothing", func(t *testing.T) { + broken := *f.auth + broken.Client = failingClient{f.admin} + _, err := broken.Authorize(ctx, podA, f.req(c1.Credential, vpcSelector, "nonceU0000000001")) + expectErr(t, "Authorize", err, ErrUnavailable) + before := f.service(g).Spec.ValidUntil + _, err = broken.Refresh(ctx, podA, g.Authorization, g.Revision, g.Service) + expectErr(t, "Refresh", err, ErrUnavailable) + if s := f.service(g); !s.Spec.ValidUntil.Equal(before) || retired(s) { + t.Fatalf("an unavailable store changed the authorization: %v", s.Spec.ValidUntil) + } + }) + + var gb Grant + t.Run("another pod takes over only after release", func(t *testing.T) { + // Core still holds an address of pod A's service. + s := f.service(g) + s.Status.Allocations = []galacticv1alpha1.EdgeAllocation{{Address: "fd20:a1b:2c3d:10:2::101", + Attachment: "edge/a", AttachmentUID: "u"}} + if err := f.admin.Status().Update(ctx, s); err != nil { + t.Fatal(err) + } + _, err := f.tryAuthorize(podB, c1.Credential, "nonceB0000000001") + expectErr(t, "takeover while released is pending", err, ErrUnavailable) + s = f.service(g) + if !retired(s) || s.Spec.ValidUntil.After(f.clock()) { + t.Fatalf("predecessor %v not retired and expired", s.Annotations) + } + _, err = f.auth.Refresh(ctx, podA, g.Authorization, g.Revision, g.Service) + expectErr(t, "predecessor's Refresh", err, connectip.ErrForbidden) + s.Status.Allocations = nil + if err := f.admin.Status().Update(ctx, s); err != nil { + t.Fatal(err) + } + // Within the retry window, a retry of the predecessor's call + // finds its retired authorization; the takeover waits. + _, err = f.tryAuthorize(podA, c1.Credential, "nonceA0000000001") + expectErr(t, "the predecessor's retry within the window", err, connectip.ErrForbidden) + _, err = f.tryAuthorize(podB, c1.Credential, "nonceB0000000001") + expectErr(t, "takeover within the retry window", err, ErrUnavailable) + f.advance(RetryWindow) + gb = f.authorize(podB, c1.Credential, "nonceB0000000002") + if f.service(g) != nil || gb.Service == g.Service || gb.Prefix != g.Prefix { + t.Fatalf("takeover = %+v, want a new service for the same /96 with the old one deleted", gb) + } + _, err = f.tryAuthorize(podA, c1.Credential, "nonceA0000000001") + expectErr(t, "the predecessor's retry after the window", err, ErrInvalid) + if s := f.service(gb); s == nil || retired(s) { + t.Fatal("a stale retry disturbed the successor") + } + if err := f.auth.Retire(ctx, podA, g.Authorization, g.Service); err != nil { + t.Fatalf("retiring a deleted incarnation = %v, want success", err) + } + if s := f.service(gb); s == nil || retired(s) { + t.Fatal("retiring a deleted incarnation touched its successor of the same name") + } + }) + t.Run("retired authorization never renews", func(t *testing.T) { + if err := f.auth.Retire(ctx, podB, gb.Authorization, gb.Service); err != nil { + t.Fatal(err) + } + _, err := f.auth.Refresh(ctx, podB, gb.Authorization, gb.Revision, gb.Service) + expectErr(t, "Refresh", err, connectip.ErrForbidden) + _, err = f.tryAuthorize(podB, c1.Credential, "nonceB0000000002") + expectErr(t, "retried Authorize", err, connectip.ErrForbidden) + if s := f.service(gb); s == nil || !retired(s) || s.Spec.ValidUntil.After(f.clock()) { + t.Fatal("retired service not expired") + } + }) + + var gr Grant + var c1r Issued + t.Run("rotation", func(t *testing.T) { + f.advance(RetryWindow) + g2 := f.authorize(podA, c1.Credential, "nonceA0000000003") + var err error + if c1r, err = f.grants.Rotate(ctx, c1.Connector, MaxLifetime); err != nil { + t.Fatal(err) + } + if c1r.Revision != 2 || c1r.Prefix != c1.Prefix || c1r.Credential == c1.Credential { + t.Fatalf("rotated %+v", c1r) + } + _, err = f.auth.Refresh(ctx, podA, g2.Authorization, g2.Revision, g2.Service) + expectErr(t, "Refresh after rotation", err, connectip.ErrForbidden) + if s := f.service(g2); s == nil || !retired(s) { + t.Fatal("Refresh refused on rotation did not retire the authorization") + } + _, err = f.tryAuthorize(podA, c1.Credential, "nonceA0000000004") + expectErr(t, "old credential", err, connectip.ErrUnauthenticated) + _, err = f.tryAuthorize(podA, c1r.Credential, "nonceA0000000005") + expectErr(t, "new revision while the old one is kept", err, ErrUnavailable) + f.advance(RetryWindow) + gr = f.authorize(podA, c1r.Credential, "nonceA0000000006") + if gr.Revision != 2 || gr.Service == g2.Service || gr.Prefix != c1.Prefix { + t.Fatalf("new revision's grant %+v", gr) + } + }) + t.Run("credential expiry caps the deadline", func(t *testing.T) { + short, err := f.grants.Rotate(ctx, c1.Connector, 30*time.Second) + if err != nil { + t.Fatal(err) + } + if err := f.auth.Sweep(ctx); err != nil { + t.Fatal(err) + } + if s := f.service(gr); s == nil || !retired(s) { + t.Fatal("sweep left a rotated revision's authorization live") + } + f.advance(RetryWindow) + gs := f.authorize(podA, short.Credential, "nonceA0000000007") + if !gs.NotAfter.Equal(short.Expires) || !gs.NotAfter.Before(f.deadline()) { + t.Fatalf("deadline %v, want the credential expiry %v", gs.NotAfter, short.Expires) + } + f.advance(21 * time.Second) + _, err = f.auth.Refresh(ctx, podA, gs.Authorization, gs.Revision, gs.Service) + expectErr(t, "Refresh past expiry", err, connectip.ErrForbidden) + _, err = f.tryAuthorize(podA, short.Credential, "nonceA0000000008") + expectErr(t, "expired credential", err, connectip.ErrUnauthenticated) + }) + + t.Run("revoking one connector leaves another", func(t *testing.T) { + c2 := f.connector() + g2 := f.authorize(podA, c2.Credential, "nonceC0000000001") + cur, err := f.grants.Rotate(ctx, c1.Connector, MaxLifetime) + if err != nil { + t.Fatal(err) + } + f.advance(RetryWindow) + g1 := f.authorize(podA, cur.Credential, "nonceA0000000009") + if err := f.grants.Disable(ctx, c1.Connector); err != nil { + t.Fatal(err) + } + if err := f.auth.Sweep(ctx); err != nil { + t.Fatal(err) + } + if s := f.service(g1); s == nil || !retired(s) || s.Spec.ValidUntil.After(f.clock()) { + t.Fatal("the sweep left a disabled connector's authorization live") + } + _, err = f.tryAuthorize(podA, cur.Credential, "nonceA0000000010") + expectErr(t, "disabled connector", err, connectip.ErrForbidden) + s := f.service(g2) + if s == nil || retired(s) { + t.Fatal("revoking one connector retired another's authorization") + } + if _, err := f.auth.Refresh(ctx, podA, g2.Authorization, g2.Revision, g2.Service); err != nil { + t.Fatalf("the other connector's Refresh: %v", err) + } + + t.Run("unrenewed authorization is swept", func(t *testing.T) { + f.advance(DefaultTTL) + if err := f.auth.Sweep(ctx); err != nil { + t.Fatal(err) + } + if s := f.service(g2); s == nil || !retired(s) { + t.Fatal("the sweep left an authorization past its validUntil unretired") + } + f.advance(RetryWindow) + if err := f.auth.Sweep(ctx); err != nil { + t.Fatal(err) + } + if f.service(g2) != nil || f.service(g1) != nil { + t.Fatal("the sweep kept released authorizations past the retry window") + } + }) + t.Run("retired connector", func(t *testing.T) { + if err := f.grants.Retire(ctx, c2.Connector); err != nil { + t.Fatal(err) + } + l, _, err := f.auth.ledgers().Read(ctx, string(f.vpc.UID), testCell) + if err != nil { + t.Fatal(err) + } + if e, _ := l.Lookup(c2.Connector); !e.Retired { + t.Fatalf("retired connector's entry %+v", e) + } + c3 := f.connector() + if c3.Prefix == c1.Prefix || c3.Prefix == c2.Prefix { + t.Fatalf("a new connector reused /96 %s", c3.Prefix) + } + }) + }) + + t.Run("concurrent pods leave one successor", func(t *testing.T) { + c5 := f.connector() + for round := range 5 { + reqs := map[string]AuthorizeRequest{} + for _, pod := range []string{"pod-x", "pod-y", "pod-z"} { + reqs[pod] = f.req(c5.Credential, vpcSelector, fmt.Sprintf("nonce%s%011d", pod[4:], round)) + } + var wg sync.WaitGroup + for pod, req := range reqs { + wg.Go(func() { _, _ = f.auth.Authorize(ctx, podID(pod), req) }) + } + wg.Wait() + var list galacticv1alpha1.EdgeServiceList + if err := f.admin.List(ctx, &list, client.MatchingLabels{LabelConnector: c5.Connector}); err != nil { + t.Fatal(err) + } + live := 0 + for i := range list.Items { + if !retired(&list.Items[i]) { + live++ + } + } + if live > 1 { + t.Fatalf("round %d: %d live incarnations of one connector", round, live) + } + f.advance(RetryWindow) + if err := f.auth.Sweep(ctx); err != nil { + t.Fatal(err) + } + } + }) + + t.Run("rotation racing an issue leaves nothing live", func(t *testing.T) { + c6 := f.connector() + racing := *f.auth + racing.Client = hookClient{Client: f.admin, beforeCreate: func() { + if _, err := f.grants.Rotate(ctx, c6.Connector, MaxLifetime); err != nil { + t.Error(err) + } + }} + _, err := racing.Authorize(ctx, podA, f.req(c6.Credential, vpcSelector, "nonceE0000000001")) + expectErr(t, "Authorize racing a rotation", err, connectip.ErrForbidden) + var s galacticv1alpha1.EdgeService + if err := f.admin.Get(ctx, client.ObjectKey{Namespace: tenantNS, Name: ServiceName(c6.Connector)}, &s); err != nil || + !retired(&s) || s.Spec.ValidUntil.After(f.clock()) { + t.Fatalf("incarnation issued across a rotation = %v, %v; want it retired", s.Annotations, err) + } + }) + t.Run("foreign service with the connector's labels", func(t *testing.T) { + c7 := f.connector() + foreign := &galacticv1alpha1.EdgeService{ + ObjectMeta: metav1.ObjectMeta{Namespace: tenantNS, Name: ServiceName(c7.Connector), + Labels: map[string]string{LabelConnector: c7.Connector, LabelManagedBy: ManagedBy}}, + Spec: galacticv1alpha1.EdgeServiceSpec{ + ProviderRef: galacticv1alpha1.EdgeProviderReference{Name: testProvider}, + VPCRef: galacticv1alpha1.EdgeVPCReference{Name: vpcName, UID: string(f.vpc.UID)}, + Capabilities: []string{packetAccess}, + }, + } + f.create(foreign) + if err := f.auth.Sweep(ctx); err != nil { + t.Fatal(err) + } + _, err := f.tryAuthorize(podA, c7.Credential, "nonceF0000000001") + expectErr(t, "Authorize beside a foreign service", err, ErrUnavailable) + f.advance(RetryWindow) + if g7 := f.authorize(podA, c7.Credential, "nonceF0000000002"); g7.Service.UID == string(foreign.UID) { + t.Fatal("the foreign service was adopted") + } + }) + + t.Run("recreated ledger refuses a stale write", func(t *testing.T) { + l := f.auth.ledgers() + ref := VPCRef{Namespace: tenantNS, Name: vpcName, UID: string(f.vpc.UID)} + if err := l.Init(ctx, ref, "cell-9", testConnectors, testPools); err == nil { + t.Fatal("Init succeeded with ranges another cell's ledger holds") + } + other := netip.MustParsePrefix("fd20:a1b:2c3d:20:1::/80") + otherPools := netip.MustParsePrefix("fd20:a1b:2c3d:20:2::/80") + if err := l.Init(ctx, ref, "cell-9", other, otherPools); err != nil { + t.Fatalf("Init of another cell with disjoint ranges: %v", err) + } + _, cm, err := l.Read(ctx, string(f.vpc.UID), testCell) + if err != nil { + t.Fatal(err) + } + stale := cm.DeepCopy() + if err := f.admin.Delete(ctx, cm); err != nil { + t.Fatal(err) + } + if err := l.Init(ctx, ref, testCell, testConnectors, testPools); err == nil { + t.Fatal("Init succeeded while this cell's issued services hold allocations") + } + if err := f.admin.DeleteAllOf(ctx, &galacticv1alpha1.EdgeService{}, client.InNamespace(tenantNS), + client.MatchingLabels{LabelManagedBy: ManagedBy}); err != nil { + t.Fatal(err) + } + if err := l.Init(ctx, ref, testCell, testConnectors, testPools); err != nil { + t.Fatal(err) + } + stale.Data = map[string]string{ledgerKey: "{}"} + if err := f.admin.Update(ctx, stale); err == nil { + t.Fatal("a write conditioned on the deleted ledger succeeded") + } + if _, _, err := l.Read(ctx, string(f.vpc.UID), testCell); err != nil { + t.Fatalf("recreated ledger: %v", err) + } + }) + + t.Run("recreated VPC", func(t *testing.T) { + c4 := f.connector() + g4 := f.authorize(podA, c4.Credential, "nonceD0000000001") + if err := f.admin.Delete(ctx, f.vpc); err != nil { + t.Fatal(err) + } + f.newVPC() + _, err := f.tryAuthorize(podA, c4.Credential, "nonceD0000000002") + expectErr(t, "Authorize", err, connectip.ErrForbidden) + _, err = f.auth.Refresh(ctx, podA, g4.Authorization, g4.Revision, g4.Service) + expectErr(t, "Refresh", err, connectip.ErrForbidden) + }) + + t.Run("credentials stay out of logs", func(t *testing.T) { + logs := f.logs.String() + for _, c := range []string{c1.Credential, c1r.Credential} { + if strings.Contains(logs, c) { + t.Fatal("a credential appears in the authority's logs") + } + } + }) +} + +// hookClient runs beforeCreate ahead of every create. +type hookClient struct { + client.Client + beforeCreate func() +} + +func (h hookClient) Create(ctx context.Context, obj client.Object, opts ...client.CreateOption) error { + h.beforeCreate() + return h.Client.Create(ctx, obj, opts...) +} + +// failingClient fails every read, as an unavailable API server does. +type failingClient struct{ client.Client } + +func (failingClient) Get(context.Context, client.ObjectKey, client.Object, ...client.GetOption) error { + return errors.New("API server unavailable") +} + +func (failingClient) List(context.Context, client.ObjectList, ...client.ListOption) error { + return errors.New("API server unavailable") +} + +// TestOverHTTPS runs a Client against a Server on TLS 1.3 with a token +// reviewer standing in for the API server's TokenReview. +func TestOverHTTPS(t *testing.T) { + f := startFixture(t) + c1 := f.connector() + logs := &syncBuffer{} + srvLog := slog.New(slog.NewJSONHandler(logs, &slog.HandlerOptions{Level: slog.LevelDebug})) + f.auth.Log = srvLog + reviewer := tokenReviewer{ + "pod-a-token": {Username: "system:serviceaccount:edge:masque", + Extra: map[string]authenticationv1.ExtraValue{admission.PodUIDExtra: {testPodA}}}, + "unbound-token": {Username: "system:serviceaccount:edge:masque"}, + } + ts := httptest.NewUnstartedServer((&Server{Authority: f.auth, Reviewer: reviewer, Log: srvLog}).Handler()) + ts.TLS = &tls.Config{MinVersion: tls.VersionTLS13} + ts.StartTLS() + defer ts.Close() + dir := t.TempDir() + caFile, tokenFile := filepath.Join(dir, "ca.crt"), filepath.Join(dir, "token") + ca := pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: ts.Certificate().Raw}) + if err := os.WriteFile(caFile, ca, 0o600); err != nil { + t.Fatal(err) + } + writeToken := func(tok string) { + if err := os.WriteFile(tokenFile, []byte(tok+"\n"), 0o600); err != nil { + t.Fatal(err) + } + } + writeToken("pod-a-token") + c := &Client{URL: ts.URL, TokenFile: tokenFile, CAFile: caFile, Now: f.clock} + ctx := context.Background() + + g, err := c.Authorize(ctx, c1.Credential, vpcSelector) + if err != nil { + t.Fatal(err) + } + if g.Connector != c1.Connector || g.Principal != c1.Connector || g.Prefix != c1.Prefix || g.VPC != vpcSelector || + g.VPCID != testVPCID || g.Revision != 1 || g.Authorization == "" || g.Service.UID == "" || + !g.NotAfter.Equal(f.deadline()) || len(g.Routes) != 1 { + t.Fatalf("grant %+v", g) + } + f.advance(5 * time.Second) + r, err := c.Refresh(ctx, g) + if err != nil || !r.NotAfter.Equal(f.deadline()) { + t.Fatalf("Refresh = %+v, %v", r, err) + } + if _, err := c.Authorize(ctx, NewCredential(), vpcSelector); !errors.Is(err, connectip.ErrUnauthenticated) { + t.Fatalf("unknown credential = %v", err) + } + if _, err := c.Authorize(ctx, c1.Credential, "tenant/other"); !errors.Is(err, connectip.ErrForbidden) { + t.Fatalf("another VPC = %v", err) + } + stale := g + stale.Revision++ + if _, err := c.Refresh(ctx, stale); !errors.Is(err, connectip.ErrForbidden) { + t.Fatalf("stale revision = %v", err) + } + for _, tok := range []string{"unbound-token", "unknown-token"} { + writeToken(tok) + if _, err := c.Refresh(ctx, g); !errors.Is(err, connectip.ErrUnauthenticated) { + t.Fatalf("Refresh with %s = %v", tok, err) + } + } + writeToken("pod-a-token") + if err := c.Retire(ctx, g); err != nil { + t.Fatal(err) + } + if _, err := c.Refresh(ctx, g); !errors.Is(err, connectip.ErrForbidden) { + t.Fatalf("Refresh after Retire = %v", err) + } + + t.Run("untrusted server", func(t *testing.T) { + other := httptest.NewTLSServer(nil) + defer other.Close() + u := &Client{URL: other.URL, TokenFile: tokenFile, CAFile: caFile, Now: f.clock} + _, err := u.Authorize(ctx, c1.Credential, vpcSelector) + if err == nil || errors.Is(err, connectip.ErrForbidden) || errors.Is(err, connectip.ErrUnauthenticated) { + t.Fatalf("Authorize against an untrusted server = %v, want a retryable TLS error", err) + } + }) + if strings.Contains(logs.String(), c1.Credential) || strings.Contains(logs.String(), "pod-a-token") { + t.Fatal("a credential or token appears in the server's logs") + } + var _ masque.Authorizer = c +} + +// tokenReviewer authenticates the tokens it maps. +type tokenReviewer map[string]authenticationv1.UserInfo + +func (r tokenReviewer) Review(_ context.Context, token string) (authenticationv1.UserInfo, error) { + u, ok := r[token] + if !ok { + return authenticationv1.UserInfo{}, connectip.ErrUnauthenticated + } + return u, nil +} diff --git a/internal/edge/masque/authority/client.go b/internal/edge/masque/authority/client.go new file mode 100644 index 00000000..418d022d --- /dev/null +++ b/internal/edge/masque/authority/client.go @@ -0,0 +1,197 @@ +// Copyright 2026 Datum Cloud, Inc. +// +// SPDX-License-Identifier: AGPL-3.0-or-later + +package authority + +import ( + "bytes" + "context" + "crypto/rand" + "crypto/tls" + "crypto/x509" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "os" + "strings" + "sync" + "time" + + "go.datum.net/galactic/internal/edge/connectip" + "go.datum.net/galactic/internal/edge/masque" +) + +// authorizeAttempts bounds how often Authorize sends one call, with one +// nonce, after a transport failure. +const authorizeAttempts = 2 + +// maxResponse bounds an answer the client reads: a grant with MaxRoutes +// routes fits well within it. +const maxResponse = 64 << 10 + +// Client is the provider side of the authority: a masque.Authorizer that +// calls a Server over HTTPS as the pod its ServiceAccount token is bound to. +type Client struct { + // URL is the authority's base URL, https://host:port. + URL string + // TokenFile holds the pod's projected ServiceAccount token for the + // authority's audience. It is reread on every call, so a rotated token + // is used once the kubelet writes it. + TokenFile string + // CAFile holds the PEM roots the authority's certificate must chain to. + // It is reread when it changes, so roots rotate without a restart. + CAFile string + // Timeout bounds one call; zero means two seconds. + Timeout time.Duration + // Now stamps Authorize calls; nil means time.Now. + Now func() time.Time + + mu sync.Mutex + transport *http.Transport + caStamp string +} + +var _ masque.Authorizer = (*Client)(nil) + +// Authorize implements masque.Authorizer. +func (c *Client) Authorize(ctx context.Context, credential, vpc string) (masque.Grant, error) { + now := time.Now + if c.Now != nil { + now = c.Now + } + req := AuthorizeRequest{Credential: credential, VPC: vpc, Nonce: rand.Text(), Issued: now()} + var g Grant + var err error + for range authorizeAttempts { + var sent bool + sent, err = c.call(ctx, PathAuthorize, req, &g) + if err == nil { + return toGrant(&g), nil + } + if sent || ctx.Err() != nil { + break + } + } + return masque.Grant{}, err +} + +// Refresh implements masque.Authorizer. +func (c *Client) Refresh(ctx context.Context, g masque.Grant) (masque.Grant, error) { + var out Grant + _, err := c.call(ctx, PathRefresh, RefreshRequest{ + Authorization: g.Authorization, Revision: g.Revision, Service: fromServiceRef(g.Service), + }, &out) + if err != nil { + return masque.Grant{}, err + } + return toGrant(&out), nil +} + +// Retire implements masque.Authorizer. +func (c *Client) Retire(ctx context.Context, g masque.Grant) error { + _, err := c.call(ctx, PathRetire, RetireRequest{ + Authorization: g.Authorization, Service: fromServiceRef(g.Service), + }, &struct{}{}) + return err +} + +// call posts body to path and decodes a successful answer into out. It +// reports whether the server answered. A 401 wraps +// connectip.ErrUnauthenticated and a 403 connectip.ErrForbidden; any other +// failure is retryable. +func (c *Client) call(ctx context.Context, path string, body, out any) (bool, error) { + timeout := c.Timeout + if timeout == 0 { + timeout = requestTimeout + } + ctx, cancel := context.WithTimeout(ctx, timeout) + defer cancel() + token, err := os.ReadFile(c.TokenFile) + if err != nil { + return false, fmt.Errorf("read the authority token: %w", err) + } + transport, err := c.currentTransport() + if err != nil { + return false, err + } + b, err := json.Marshal(body) + if err != nil { + return false, err + } + req, err := http.NewRequestWithContext(ctx, http.MethodPost, strings.TrimSuffix(c.URL, "/")+path, + bytes.NewReader(b)) + if err != nil { + return false, err + } + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Authorization", "Bearer "+strings.TrimSpace(string(token))) + resp, err := (&http.Client{Transport: transport}).Do(req) + if err != nil { + return false, fmt.Errorf("call the authority: %w", err) + } + defer func() { _ = resp.Body.Close() }() + r := io.LimitReader(resp.Body, maxResponse) + switch resp.StatusCode { + case http.StatusOK: + if err := json.NewDecoder(r).Decode(out); err != nil { + return true, fmt.Errorf("decode the authority's answer: %w", err) + } + return true, nil + case http.StatusUnauthorized: + return true, fmt.Errorf("authority: %w", connectip.ErrUnauthenticated) + case http.StatusForbidden: + return true, fmt.Errorf("authority: %w", connectip.ErrForbidden) + default: + return true, fmt.Errorf("authority answered %s", resp.Status) + } +} + +// currentTransport returns a TLS 1.3 transport trusting the roots in CAFile, +// rebuilt when the file changed. +func (c *Client) currentTransport() (*http.Transport, error) { + fi, err := os.Stat(c.CAFile) + if err != nil { + return nil, fmt.Errorf("stat the authority CA: %w", err) + } + stamp := fmt.Sprintf("%d/%d", fi.ModTime().UnixNano(), fi.Size()) + c.mu.Lock() + defer c.mu.Unlock() + if c.transport != nil && stamp == c.caStamp { + return c.transport, nil + } + pem, err := os.ReadFile(c.CAFile) + if err != nil { + return nil, fmt.Errorf("read the authority CA: %w", err) + } + pool := x509.NewCertPool() + if !pool.AppendCertsFromPEM(pem) { + return nil, errors.New("the authority CA file holds no certificate") + } + if c.transport != nil { + c.transport.CloseIdleConnections() + } + c.transport = &http.Transport{ + TLSClientConfig: &tls.Config{RootCAs: pool, MinVersion: tls.VersionTLS13}, + ForceAttemptHTTP2: true, + MaxIdleConnsPerHost: 4, + IdleConnTimeout: 30 * time.Second, + } + c.caStamp = stamp + return c.transport, nil +} + +func toGrant(g *Grant) masque.Grant { + return masque.Grant{ + Connector: g.Connector, Principal: g.Connector, VPC: g.VPC, VPCUID: g.VPCUID, VPCID: g.VPCID, + Service: masque.ServiceRef{Namespace: g.Service.Namespace, Name: g.Service.Name, UID: g.Service.UID}, + Prefix: g.Prefix, Routes: g.Routes, NotAfter: g.NotAfter, Authorization: g.Authorization, + Revision: g.Revision, + } +} + +func fromServiceRef(s masque.ServiceRef) ServiceRef { + return ServiceRef{Namespace: s.Namespace, Name: s.Name, UID: s.UID} +} diff --git a/internal/edge/masque/authority/credential.go b/internal/edge/masque/authority/credential.go new file mode 100644 index 00000000..3dc4c895 --- /dev/null +++ b/internal/edge/masque/authority/credential.go @@ -0,0 +1,42 @@ +// Copyright 2026 Datum Cloud, Inc. +// +// SPDX-License-Identifier: AGPL-3.0-or-later + +package authority + +import ( + "crypto/rand" + "crypto/sha256" + "crypto/subtle" + "encoding/base64" + "encoding/hex" +) + +// credentialBytes is the number of random bytes in a bearer credential. +const credentialBytes = 32 + +// NewCredential returns a new bearer credential: 32 random bytes, base64url +// encoded without padding. +func NewCredential() string { + b := make([]byte, credentialBytes) + _, _ = rand.Read(b) // crypto/rand.Read never returns an error. + return base64.RawURLEncoding.EncodeToString(b) +} + +// Digest returns the hex SHA-256 digest of credential, the only form in +// which the authority stores it. +func Digest(credential string) string { + sum := sha256.Sum256([]byte(credential)) + return hex.EncodeToString(sum[:]) +} + +// wellFormed reports whether credential could be one NewCredential issued. +func wellFormed(credential string) bool { + b, err := base64.RawURLEncoding.DecodeString(credential) + return err == nil && len(b) == credentialBytes +} + +// sameDigest compares two digests in constant time. +func sameDigest(a, b string) bool { + return subtle.ConstantTimeCompare([]byte(a), []byte(b)) == 1 +} diff --git a/internal/edge/masque/authority/doc.go b/internal/edge/masque/authority/doc.go new file mode 100644 index 00000000..93f9bfcf --- /dev/null +++ b/internal/edge/masque/authority/doc.go @@ -0,0 +1,31 @@ +// Copyright 2026 Datum Cloud, Inc. +// +// SPDX-License-Identifier: AGPL-3.0-or-later + +// Package authority is the MASQUE grant authority: it authenticates +// connector credentials, keeps the durable connector allocation ledger and +// issues the finite EdgeService of each connector authorization, which the +// MASQUE provider's adapter consumes through masque.Authorizer. +// +// The operator keeps one grant record per connector in a Secret of the +// authority's namespace. A record stores the SHA-256 digest of the +// connector's bearer credential, never the credential, with its revision, +// expiry, enabled flag, VPC and allowed destinations. One ledger ConfigMap +// per VPC and cell holds every connector's /96 and attachment pool; retired +// entries stay as tombstones, so no /96 or pool is ever reassigned. +// +// Authorize matches a credential against every record in constant time, +// rereads the record, the VPC and the ledger without a cache, and issues one +// EdgeService per authorization incarnation, bound to the calling provider +// pod. Its validUntil is the credential expiry or the read start plus TTL +// less IssuerSkew, whichever is earlier. Refresh repeats those reads and +// renews validUntil only while the credential revision is current and +// enabled. Retire, revocation and rotation set validUntil in the past and +// mark the incarnation retired, which is never renewed again. The authority +// keeps no state of its own beyond the API objects, so a restarted or new +// instance rereads everything before issuing anything. +// +// Server serves Authorize, Refresh and Retire over HTTPS to provider pods +// that present a pod-bound ServiceAccount token, and Client is the provider +// side's masque.Authorizer. +package authority diff --git a/internal/edge/masque/authority/grants.go b/internal/edge/masque/authority/grants.go new file mode 100644 index 00000000..4b8f8d39 --- /dev/null +++ b/internal/edge/masque/authority/grants.go @@ -0,0 +1,185 @@ +// Copyright 2026 Datum Cloud, Inc. +// +// SPDX-License-Identifier: AGPL-3.0-or-later + +package authority + +import ( + "context" + "errors" + "fmt" + "net/netip" + "time" + + "github.com/google/uuid" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "sigs.k8s.io/controller-runtime/pkg/client" + + cloudv1alpha1 "go.datum.net/galactic/api/cloud/v1alpha1" + galacticv1alpha1 "go.datum.net/galactic/api/v1alpha1" +) + +// Grants provisions connector grant records in Namespace for an operator. +// Client must read without a cache. +type Grants struct { + Client client.Client + Namespace string + Now func() time.Time +} + +// Issued is a credential as issued to an operator: the only time the +// credential itself is returned. +type Issued struct { + Connector string `json:"connector"` + Credential string `json:"credential"` + Revision int64 `json:"revision"` + VPC string `json:"vpc"` + Prefix netip.Prefix `json:"prefix"` + Expires time.Time `json:"expires"` +} + +// CreateRequest describes a new connector. +type CreateRequest struct { + // VPC is the VPC's namespace and name; its UID is read from the API. + VPC VPCRef + Cell string + // Connector is the connector's UUID; empty means a new random one. + Connector string + // Lifetime is how long the credential is valid, at most MaxLifetime. + Lifetime time.Duration + Routes []netip.Prefix + TrafficPolicy *galacticv1alpha1.TrafficPolicyBindingSpec +} + +func (g Grants) ledgers() Ledgers { + return Ledgers(g) +} + +// expiry returns now plus lifetime, refusing a lifetime outside +// (0, MaxLifetime]. +func (g Grants) expiry(lifetime time.Duration) (time.Time, error) { + if lifetime <= 0 || lifetime > MaxLifetime { + return time.Time{}, fmt.Errorf("lifetime %v is not within (0, %v]", lifetime, MaxLifetime) + } + return g.Now().Add(lifetime).UTC().Truncate(time.Second), nil +} + +// Create allocates the connector's /96 in the ledger of its VPC and cell, +// writes its grant record with a new credential and returns that +// credential. It refuses a connector that already has a record. +func (g Grants) Create(ctx context.Context, req CreateRequest) (Issued, error) { + expires, err := g.expiry(req.Lifetime) + if err != nil { + return Issued{}, err + } + if req.Connector == "" { + req.Connector = uuid.NewString() + } + var vpc cloudv1alpha1.VPC + if err := g.Client.Get(ctx, client.ObjectKey{Namespace: req.VPC.Namespace, Name: req.VPC.Name}, &vpc); err != nil { + return Issued{}, fmt.Errorf("read VPC %s: %w", req.VPC.Selector(), err) + } + if req.VPC.UID != "" && req.VPC.UID != string(vpc.UID) { + return Issued{}, fmt.Errorf("VPC %s has UID %s, not %s", req.VPC.Selector(), vpc.UID, req.VPC.UID) + } + req.VPC.UID = string(vpc.UID) + cred := NewCredential() + r := &Record{ + Connector: req.Connector, Cell: req.Cell, VPC: req.VPC, Revision: 1, Digest: Digest(cred), + Expires: expires, Enabled: true, Routes: req.Routes, TrafficPolicy: req.TrafficPolicy, + } + s := &corev1.Secret{ObjectMeta: metav1.ObjectMeta{Namespace: g.Namespace}} + if err := encodeRecord(s, r); err != nil { + return Issued{}, err + } + e, err := g.ledgers().Allocate(ctx, r.VPC.UID, r.Cell, r.Connector) + if err != nil { + return Issued{}, fmt.Errorf("allocate connector %s: %w", r.Connector, err) + } + if err := g.Client.Create(ctx, s); err != nil { + return Issued{}, fmt.Errorf("create grant record %s: %w", s.Name, err) + } + return Issued{Connector: r.Connector, Credential: cred, Revision: r.Revision, VPC: r.VPC.Selector(), + Prefix: e.Prefix, Expires: expires}, nil +} + +// Rotate replaces the connector's credential with a new one valid for +// lifetime, under the next revision, and returns it. Authorizations issued +// under the old revision are no longer renewed, and the old credential +// admits nothing. +func (g Grants) Rotate(ctx context.Context, connector string, lifetime time.Duration) (Issued, error) { + expires, err := g.expiry(lifetime) + if err != nil { + return Issued{}, err + } + cred := NewCredential() + var out Issued + var vpcUID, cell string + err = g.update(ctx, connector, func(r *Record) error { + if !r.Enabled { + return errors.New("connector is disabled") + } + r.Revision++ + r.Digest, r.Expires = Digest(cred), expires + out = Issued{Connector: r.Connector, Credential: cred, Revision: r.Revision, VPC: r.VPC.Selector(), + Expires: expires} + vpcUID, cell = r.VPC.UID, r.Cell + return nil + }) + if err != nil { + return Issued{}, err + } + // The new credential is committed: return it even when the prefix, + // which is informational, cannot be read. + if l, _, err := g.ledgers().Read(ctx, vpcUID, cell); err == nil { + if e, ok := l.Lookup(connector); ok { + out.Prefix = e.Prefix + } + } + return out, nil +} + +// Disable revokes the connector under the next revision: its credential +// admits nothing and its authorizations are no longer renewed. +func (g Grants) Disable(ctx context.Context, connector string) error { + return g.update(ctx, connector, func(r *Record) error { + if r.Enabled { + r.Enabled = false + r.Revision++ + } + return nil + }) +} + +// Retire disables the connector and tombstones its allocation, so its /96 +// and attachment pool are never allocated again. +func (g Grants) Retire(ctx context.Context, connector string) error { + if err := g.Disable(ctx, connector); err != nil { + return err + } + r, _, err := readRecord(ctx, g.Client, g.Namespace, connector) + if err != nil { + return err + } + return g.ledgers().Retire(ctx, r.VPC.UID, r.Cell, connector) +} + +// update applies change to a fresh read of the connector's record and +// writes it back conditioned on the read's UID and resourceVersion. +func (g Grants) update(ctx context.Context, connector string, change func(*Record) error) error { + r, s, err := readRecord(ctx, g.Client, g.Namespace, connector) + if err != nil { + return fmt.Errorf("read grant record of %s: %w", connector, err) + } + if err := change(r); err != nil { + return fmt.Errorf("connector %s: %w", connector, err) + } + if err := encodeRecord(s, r); err != nil { + return err + } + if err := g.Client.Update(ctx, s); err != nil { + return fmt.Errorf("update grant record of %s: %w", connector, err) + } + return nil +} diff --git a/internal/edge/masque/authority/ledger.go b/internal/edge/masque/authority/ledger.go new file mode 100644 index 00000000..4356360e --- /dev/null +++ b/internal/edge/masque/authority/ledger.go @@ -0,0 +1,303 @@ +// Copyright 2026 Datum Cloud, Inc. +// +// SPDX-License-Identifier: AGPL-3.0-or-later + +package authority + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "net/netip" + "time" + + "github.com/google/uuid" + corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/util/validation" + "sigs.k8s.io/controller-runtime/pkg/client" + + cloudv1alpha1 "go.datum.net/galactic/api/cloud/v1alpha1" + galacticv1alpha1 "go.datum.net/galactic/api/v1alpha1" +) + +const ( + // LabelLedger marks a ConfigMap of the authority's namespace as an + // allocation ledger. + LabelLedger = "masque.galactic.datumapis.com/ledger" + // ledgerKey is the ConfigMap data key holding a ledger. + ledgerKey = "ledger.json" + // MaxEntries is the most entries, retired ones included, one ledger + // holds. + MaxEntries = 256 + // maxLedgerBytes bounds a ledger's encoded size. + maxLedgerBytes = 256 << 10 + // rangeBits is the size of the connector and pool ranges a ledger + // allocates from. + rangeBits = 80 + // conflictRetries bounds how often a ledger write is retried from a + // fresh read after a conflict. + conflictRetries = 5 +) + +// ErrLedgerMissing is returned when a VPC and cell have no ledger. A missing +// ledger is an operator recovery event: nothing is allocated until it is +// restored. +var ErrLedgerMissing = errors.New("allocation ledger does not exist") + +// Ledger is the durable record of every connector allocation of one VPC in +// one cell. +type Ledger struct { + VPC VPCRef `json:"vpc"` + Cell string `json:"cell"` + // Connectors is the /80 connector /96s are allocated from. + Connectors netip.Prefix `json:"connectors"` + // Pools is the /80 attachment /120 pools are allocated from. + Pools netip.Prefix `json:"pools"` + // Entries are in allocation order. An entry is never removed. + Entries []Entry `json:"entries"` +} + +// Entry is one connector's allocation. +type Entry struct { + Connector string `json:"connector"` + Allocation string `json:"allocation"` + Prefix netip.Prefix `json:"prefix"` + Pool netip.Prefix `json:"pool"` + Created time.Time `json:"created"` + // Retired entries are tombstones: their prefix and pool are never + // allocated again. + Retired bool `json:"retired,omitempty"` +} + +// Lookup returns connector's entry. +func (l *Ledger) Lookup(connector string) (Entry, bool) { + for _, e := range l.Entries { + if e.Connector == connector { + return e, true + } + } + return Entry{}, false +} + +// LedgerName is the name of the ledger ConfigMap of vpcUID in cell. +func LedgerName(vpcUID, cell string) string { return "masque-ledger-" + vpcUID + "-" + cell } + +// Ledgers reads and writes allocation ledgers in Namespace. Client must read +// without a cache. +type Ledgers struct { + Client client.Client + Namespace string + Now func() time.Time +} + +// Init creates the ledger of vpc in cell, allocating connector /96s from +// connectors and attachment pools from pools. Both must be masked, +// disjoint IPv6 /80s inside one of the VPC's networks, and vpc's UID must be +// the live VPC's. It refuses to replace an existing ledger, to create one +// while an EdgeService the authority issued for the VPC in cell exists, and +// to create one whose ranges overlap those of any other ledger of the VPC. +func (s Ledgers) Init(ctx context.Context, vpc VPCRef, cell string, connectors, pools netip.Prefix) error { + if errs := validation.IsDNS1123Label(cell); len(errs) > 0 { + return fmt.Errorf("cell %q: %v", cell, errs) + } + var obj cloudv1alpha1.VPC + if err := s.Client.Get(ctx, client.ObjectKey{Namespace: vpc.Namespace, Name: vpc.Name}, &obj); err != nil { + return fmt.Errorf("read VPC %s: %w", vpc.Selector(), err) + } + if vpc.UID == "" { + vpc.UID = string(obj.UID) + } + if string(obj.UID) != vpc.UID { + return fmt.Errorf("VPC %s has UID %s, not %s", vpc.Selector(), obj.UID, vpc.UID) + } + for _, p := range []netip.Prefix{connectors, pools} { + if !p.IsValid() || !p.Addr().Is6() || p.Addr().Is4In6() || p.Bits() != rangeBits || p != p.Masked() { + return fmt.Errorf("range %s is not a masked IPv6 /%d", p, rangeBits) + } + if !inNetworks(obj.Spec.Networks, p) { + return fmt.Errorf("range %s is outside VPC %s's networks", p, vpc.Selector()) + } + } + if connectors.Overlaps(pools) { + return fmt.Errorf("connector range %s overlaps pool range %s", connectors, pools) + } + // A recreated ledger restarts its allocations; it must not hand out a + // /96 or pool an issued service still holds. + var issued galacticv1alpha1.EdgeServiceList + if err := s.Client.List(ctx, &issued, client.InNamespace(vpc.Namespace), + client.MatchingLabels{LabelManagedBy: ManagedBy}); err != nil { + return fmt.Errorf("list issued EdgeServices: %w", err) + } + for _, svc := range issued.Items { + if svc.Spec.VPCRef.UID == vpc.UID && svc.Annotations[AnnotationCell] == cell { + return fmt.Errorf("EdgeService %s/%s still holds an allocation of VPC %s; delete it first", + svc.Namespace, svc.Name, vpc.Selector()) + } + } + // Cells of one VPC must not be handed overlapping ranges. + var ledgers corev1.ConfigMapList + if err := s.Client.List(ctx, &ledgers, client.InNamespace(s.Namespace), + client.MatchingLabels{LabelLedger: yes}); err != nil { + return fmt.Errorf("list ledgers: %w", err) + } + for _, other := range ledgers.Items { + var ol Ledger + if err := json.Unmarshal([]byte(other.Data[ledgerKey]), &ol); err != nil { + return fmt.Errorf("decode ledger %s: %w", other.Name, err) + } + if ol.VPC.UID != vpc.UID || ol.Cell == cell { + continue + } + for _, p := range []netip.Prefix{connectors, pools} { + if p.Overlaps(ol.Connectors) || p.Overlaps(ol.Pools) { + return fmt.Errorf("range %s overlaps a range of the ledger of VPC %s in cell %s", + p, vpc.Selector(), ol.Cell) + } + } + } + l := &Ledger{VPC: vpc, Cell: cell, Connectors: connectors, Pools: pools, Entries: []Entry{}} + cm := &corev1.ConfigMap{ObjectMeta: metav1.ObjectMeta{ + Namespace: s.Namespace, Name: LedgerName(vpc.UID, cell), + Labels: map[string]string{LabelLedger: yes}, + }} + if err := encodeLedger(cm, l); err != nil { + return err + } + if err := s.Client.Create(ctx, cm); err != nil { + return fmt.Errorf("create ledger %s: %w", cm.Name, err) + } + return nil +} + +// inNetworks reports whether p lies inside one of networks. +func inNetworks(networks []cloudv1alpha1.Network, p netip.Prefix) bool { + for _, n := range networks { + np, err := netip.ParsePrefix(string(n)) + if err == nil && np.Addr().Is6() && np.Bits() <= p.Bits() && np.Contains(p.Addr()) { + return true + } + } + return false +} + +// Read returns the ledger of vpcUID in cell and its ConfigMap. It returns +// ErrLedgerMissing when there is none. +func (s Ledgers) Read(ctx context.Context, vpcUID, cell string) (*Ledger, *corev1.ConfigMap, error) { + var cm corev1.ConfigMap + err := s.Client.Get(ctx, client.ObjectKey{Namespace: s.Namespace, Name: LedgerName(vpcUID, cell)}, &cm) + if apierrors.IsNotFound(err) { + return nil, nil, fmt.Errorf("%w: VPC %s in cell %s", ErrLedgerMissing, vpcUID, cell) + } + if err != nil { + return nil, nil, err + } + var l Ledger + if err := json.Unmarshal([]byte(cm.Data[ledgerKey]), &l); err != nil { + return nil, nil, fmt.Errorf("decode ledger %s: %w", cm.Name, err) + } + if l.VPC.UID != vpcUID || l.Cell != cell { + return nil, nil, fmt.Errorf("ledger %s is for VPC %s in cell %s", cm.Name, l.VPC.UID, l.Cell) + } + return &l, &cm, nil +} + +// Allocate returns connector's allocation in the ledger of vpcUID in cell, +// adding one when the connector has none. A new allocation takes the next +// /96 and /120 no entry ever held. It refuses a retired connector and a +// full ledger. +func (s Ledgers) Allocate(ctx context.Context, vpcUID, cell, connector string) (Entry, error) { + var out Entry + err := s.update(ctx, vpcUID, cell, func(l *Ledger) (bool, error) { + if e, ok := l.Lookup(connector); ok { + if e.Retired { + return false, fmt.Errorf("connector %s is retired", connector) + } + out = e + return false, nil + } + if len(l.Entries) >= MaxEntries { + return false, fmt.Errorf("ledger is full at %d entries", MaxEntries) + } + n := uint64(len(l.Entries) + 1) + out = Entry{ + Connector: connector, Allocation: uuid.NewString(), + Prefix: netip.PrefixFrom(withIndex(l.Connectors.Addr(), n<<32), 96), + Pool: netip.PrefixFrom(withIndex(l.Pools.Addr(), n<<8), 120), + Created: s.Now().UTC().Truncate(time.Second), + } + l.Entries = append(l.Entries, out) + return true, nil + }) + return out, err +} + +// Retire marks connector's allocation retired. Its /96 and pool stay +// tombstoned. Retiring a retired or unknown connector does nothing. +func (s Ledgers) Retire(ctx context.Context, vpcUID, cell, connector string) error { + return s.update(ctx, vpcUID, cell, func(l *Ledger) (bool, error) { + for i := range l.Entries { + if l.Entries[i].Connector == connector && !l.Entries[i].Retired { + l.Entries[i].Retired = true + return true, nil + } + } + return false, nil + }) +} + +// update applies change to a fresh read of the ledger and writes it back +// conditioned on the read's UID and resourceVersion, retrying from a fresh +// read after a conflict. change reports whether it modified the ledger. +func (s Ledgers) update(ctx context.Context, vpcUID, cell string, change func(*Ledger) (bool, error)) error { + for range conflictRetries { + l, cm, err := s.Read(ctx, vpcUID, cell) + if err != nil { + return err + } + changed, err := change(l) + if err != nil || !changed { + return err + } + if err := encodeLedger(cm, l); err != nil { + return err + } + // Update carries the read's UID and resourceVersion, so a ledger + // changed or recreated since conflicts. + err = s.Client.Update(ctx, cm) + if !apierrors.IsConflict(err) { + return err + } + } + return fmt.Errorf("ledger %s kept changing", LedgerName(vpcUID, cell)) +} + +// withIndex returns base with n added to its low 64 bits. +func withIndex(base netip.Addr, n uint64) netip.Addr { + b := base.As16() + lo := uint64(0) + for i := 8; i < 16; i++ { + lo = lo<<8 | uint64(b[i]) + } + lo += n + for i := 15; i >= 8; i-- { + b[i] = byte(lo) + lo >>= 8 + } + return netip.AddrFrom16(b) +} + +// encodeLedger stores l in cm. +func encodeLedger(cm *corev1.ConfigMap, l *Ledger) error { + b, err := json.Marshal(l) + if err != nil { + return err + } + if len(b) > maxLedgerBytes { + return fmt.Errorf("ledger would be %d bytes, over the %d limit", len(b), maxLedgerBytes) + } + cm.Data = map[string]string{ledgerKey: string(b)} + return nil +} diff --git a/internal/edge/masque/authority/ledger_test.go b/internal/edge/masque/authority/ledger_test.go new file mode 100644 index 00000000..7d59c6c2 --- /dev/null +++ b/internal/edge/masque/authority/ledger_test.go @@ -0,0 +1,208 @@ +// Copyright 2026 Datum Cloud, Inc. +// +// SPDX-License-Identifier: AGPL-3.0-or-later + +package authority + +import ( + "context" + "errors" + "fmt" + "net/netip" + "strconv" + "sync" + "testing" + "time" + + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + utilruntime "k8s.io/apimachinery/pkg/util/runtime" + clientgoscheme "k8s.io/client-go/kubernetes/scheme" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + + cloudv1alpha1 "go.datum.net/galactic/api/cloud/v1alpha1" + galacticv1alpha1 "go.datum.net/galactic/api/v1alpha1" +) + +const ( + testNS = "masque-system" + tenantNS = "tenant" + testCell = "cell-1" + vpcName = "vpc" + cell2 = "cell-2" +) + +var ( + testConnectors = netip.MustParsePrefix("fd20:a1b:2c3d:10:1::/80") + testPools = netip.MustParsePrefix("fd20:a1b:2c3d:10:2::/80") +) + +func testScheme() *runtime.Scheme { + s := runtime.NewScheme() + utilruntime.Must(clientgoscheme.AddToScheme(s)) + utilruntime.Must(galacticv1alpha1.AddToScheme(s)) + utilruntime.Must(cloudv1alpha1.AddToScheme(s)) + return s +} + +func fakeLedgers(t *testing.T) (Ledgers, VPCRef) { + t.Helper() + vpc := &cloudv1alpha1.VPC{ + ObjectMeta: metav1.ObjectMeta{Namespace: tenantNS, Name: vpcName, UID: "11111111-1111-1111-1111-111111111111"}, + Spec: cloudv1alpha1.VPCSpec{Networks: []cloudv1alpha1.Network{"fd20:a1b:2c3d::/48"}}, + } + c := fake.NewClientBuilder().WithScheme(testScheme()).WithObjects(vpc).Build() + l := Ledgers{Client: c, Namespace: testNS, Now: time.Now} + ref := VPCRef{Namespace: tenantNS, Name: vpcName, UID: string(vpc.UID)} + if err := l.Init(context.Background(), ref, testCell, testConnectors, testPools); err != nil { + t.Fatal(err) + } + return l, ref +} + +func TestLedgerInitValidates(t *testing.T) { + l, ref := fakeLedgers(t) + ctx := context.Background() + other := VPCRef{Namespace: tenantNS, Name: vpcName, UID: "22222222-2222-2222-2222-222222222222"} + for name, tc := range map[string]struct { + vpc VPCRef + cell string + connectors, pools netip.Prefix + }{ + "exists": {ref, testCell, testConnectors, testPools}, + "wrong UID": {other, cell2, testConnectors, testPools}, + "not a /80": {ref, cell2, netip.MustParsePrefix("fd20:a1b:2c3d:10:1::/96"), testPools}, + "unmasked": {ref, cell2, netip.MustParsePrefix("fd20:a1b:2c3d:10:1:1::/80"), testPools}, + "outside VPC": {ref, cell2, netip.MustParsePrefix("fd99::/80"), testPools}, + "overlapping": {ref, cell2, testConnectors, testConnectors}, + "bad cell": {ref, "Cell_2", testConnectors, testPools}, + "IPv4": {ref, cell2, netip.MustParsePrefix("10.0.0.0/8"), testPools}, + } { + if err := l.Init(ctx, tc.vpc, tc.cell, tc.connectors, tc.pools); err == nil { + t.Errorf("%s: Init succeeded, want an error", name) + } + } +} + +func TestLedgerInitAcrossCells(t *testing.T) { + l, ref := fakeLedgers(t) + ctx := context.Background() + c2 := netip.MustParsePrefix("fd20:a1b:2c3d:20:1::/80") + p2 := netip.MustParsePrefix("fd20:a1b:2c3d:20:2::/80") + if err := l.Init(ctx, ref, cell2, testConnectors, p2); err == nil { + t.Error("a second cell sharing a connector range was accepted") + } + if err := l.Init(ctx, ref, cell2, c2, testPools); err == nil { + t.Error("a second cell sharing a pool range was accepted") + } + if err := l.Init(ctx, ref, cell2, p2, c2); err != nil { + t.Fatalf("a second cell with disjoint ranges: %v", err) + } + if err := l.Init(ctx, ref, "cell-3", c2, p2); err == nil { + t.Error("a third cell reusing the second cell's ranges was accepted") + } +} + +func TestLedgerAllocates(t *testing.T) { + l, ref := fakeLedgers(t) + ctx := context.Background() + a, err := l.Allocate(ctx, ref.UID, testCell, "a") + if err != nil { + t.Fatal(err) + } + if a.Prefix != netip.MustParsePrefix("fd20:a1b:2c3d:10:1:1::/96") || + a.Pool != netip.MustParsePrefix("fd20:a1b:2c3d:10:2::100/120") || a.Allocation == "" { + t.Fatalf("first allocation %+v", a) + } + again, err := l.Allocate(ctx, ref.UID, testCell, "a") + if err != nil || again != a { + t.Fatalf("allocating again = %+v, %v; want the same %+v", again, err, a) + } + if err := l.Retire(ctx, ref.UID, testCell, "a"); err != nil { + t.Fatal(err) + } + if _, err := l.Allocate(ctx, ref.UID, testCell, "a"); err == nil { + t.Fatal("allocating a retired connector succeeded") + } + b, err := l.Allocate(ctx, ref.UID, testCell, "b") + if err != nil { + t.Fatal(err) + } + if b.Prefix == a.Prefix || b.Pool == a.Pool || b.Prefix != netip.MustParsePrefix("fd20:a1b:2c3d:10:1:2::/96") { + t.Fatalf("allocation after a tombstone %+v reuses or skips %+v", b, a) + } + if _, err := l.Allocate(ctx, ref.UID, "cell-9", "c"); !errors.Is(err, ErrLedgerMissing) { + t.Fatalf("allocating without a ledger = %v, want %v", err, ErrLedgerMissing) + } +} + +func TestLedgerFull(t *testing.T) { + l, ref := fakeLedgers(t) + ctx := context.Background() + for i := range MaxEntries { + if _, err := l.Allocate(ctx, ref.UID, testCell, strconv.Itoa(i)); err != nil { + t.Fatalf("allocation %d: %v", i, err) + } + } + if _, err := l.Allocate(ctx, ref.UID, testCell, "one-more"); err == nil { + t.Fatal("allocating past MaxEntries succeeded") + } +} + +// TestLedgerConcurrentAllocations: concurrent writers conflict on the +// resourceVersion and retry from a fresh read, so no two connectors share a +// /96 or pool. +func TestLedgerConcurrentAllocations(t *testing.T) { + l, ref := fakeLedgers(t) + ctx := context.Background() + const n = 8 + var wg sync.WaitGroup + got := make([]Entry, n) + errs := make([]error, n) + for i := range n { + wg.Go(func() { got[i], errs[i] = l.Allocate(ctx, ref.UID, testCell, fmt.Sprint("c", i)) }) + } + wg.Wait() + prefixes, pools := map[netip.Prefix]bool{}, map[netip.Prefix]bool{} + for i := range n { + if errs[i] != nil { + // A writer can lose every retry; it must not have + // allocated. + continue + } + if prefixes[got[i].Prefix] || pools[got[i].Pool] { + t.Fatalf("allocation %+v duplicates another", got[i]) + } + prefixes[got[i].Prefix], pools[got[i].Pool] = true, true + } + lg, _, err := l.Read(ctx, ref.UID, testCell) + if err != nil { + t.Fatal(err) + } + if len(lg.Entries) != len(prefixes) { + t.Fatalf("ledger holds %d entries, %d allocations succeeded", len(lg.Entries), len(prefixes)) + } +} + +func TestWithIndex(t *testing.T) { + base := netip.MustParseAddr("fd00::ffff:ffff:ff00") + if got := withIndex(base, 0x100); got != netip.MustParseAddr("fd00::1:0:0:0") { + t.Fatalf("withIndex carries to %s", got) + } +} + +func TestRecordRoutesBounded(t *testing.T) { + r := Record{Connector: "6f1c0a3e-8f43-4f6e-9a55-1f6f1f0b5c2d", Cell: testCell, Revision: 1, + VPC: VPCRef{Namespace: tenantNS, Name: vpcName, UID: "u"}, Digest: Digest(NewCredential()), + Expires: time.Now().Add(time.Hour)} + for i := range MaxRoutes + 1 { + r.Routes = append(r.Routes, netip.PrefixFrom(withIndex(netip.MustParseAddr("fd00::"), uint64(i)<<16), 112)) + } + if err := r.validate(); err == nil { + t.Fatalf("a record with %d routes validated", len(r.Routes)) + } + r.Routes = r.Routes[:MaxRoutes] + if err := r.validate(); err != nil { + t.Fatalf("a record with %d routes: %v", MaxRoutes, err) + } +} diff --git a/internal/edge/masque/authority/record.go b/internal/edge/masque/authority/record.go new file mode 100644 index 00000000..c2697342 --- /dev/null +++ b/internal/edge/masque/authority/record.go @@ -0,0 +1,149 @@ +// Copyright 2026 Datum Cloud, Inc. +// +// SPDX-License-Identifier: AGPL-3.0-or-later + +package authority + +import ( + "context" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "net/netip" + "time" + + "github.com/google/uuid" + corev1 "k8s.io/api/core/v1" + "sigs.k8s.io/controller-runtime/pkg/client" + + galacticv1alpha1 "go.datum.net/galactic/api/v1alpha1" +) + +const ( + // LabelGrant marks a Secret of the authority's namespace as a grant + // record. + LabelGrant = "masque.galactic.datumapis.com/grant" + // LabelConnector names the connector of a grant record or of an + // EdgeService the authority issued. + LabelConnector = "masque.galactic.datumapis.com/connector" + // recordKey is the Secret data key holding a record. + recordKey = "grant.json" + // MaxLifetime is the longest a credential may stay valid. + MaxLifetime = 24 * time.Hour + // MaxRoutes bounds a record's allowed destinations. + MaxRoutes = 64 + // yes is the value of a set marker label or annotation. + yes = "true" +) + +// VPCRef identifies a VPC: its namespace and name, and the UID that pins +// the object. +type VPCRef struct { + Namespace string `json:"namespace"` + Name string `json:"name"` + UID string `json:"uid"` +} + +// Selector is the value a CONNECT-IP request names the VPC by: +// namespace/name. +func (v VPCRef) Selector() string { return v.Namespace + "/" + v.Name } + +// Record is one connector's grant, as the operator approved it. +type Record struct { + // Connector is the connector's immutable UUID. + Connector string `json:"connector"` + // Cell is the only cell whose authority honors the record. + Cell string `json:"cell"` + // VPC is the only VPC the connector may use. + VPC VPCRef `json:"vpc"` + // Revision is the credential revision. Rotating or disabling the + // credential increments it. + Revision int64 `json:"revision"` + // Digest is the hex SHA-256 digest of the current credential. + Digest string `json:"digest"` + // Expires is when the current credential stops being valid. + Expires time.Time `json:"expires"` + // Enabled is false once the operator revoked the connector. + Enabled bool `json:"enabled"` + // Routes are the gVPC destinations the connector may reach, advertised + // to the client. + Routes []netip.Prefix `json:"routes,omitempty"` + // TrafficPolicy is copied into every EdgeService the record yields. + TrafficPolicy *galacticv1alpha1.TrafficPolicyBindingSpec `json:"trafficPolicy,omitempty"` +} + +// RecordName is the name of connector's grant record Secret. +func RecordName(connector string) string { return "masque-grant-" + connector } + +// validate reports the first malformed field of r. +func (r *Record) validate() error { + if _, err := uuid.Parse(r.Connector); err != nil || r.Connector != uuid.MustParse(r.Connector).String() { + return fmt.Errorf("connector %q is not a lowercase UUID", r.Connector) + } + if r.Cell == "" || r.VPC.Namespace == "" || r.VPC.Name == "" || r.VPC.UID == "" { + return errors.New("cell and VPC namespace, name and UID are required") + } + if b, err := hex.DecodeString(r.Digest); err != nil || len(b) != 32 { + return errors.New("digest is not a hex SHA-256 digest") + } + if r.Expires.IsZero() || r.Revision < 1 { + return errors.New("expiry and a positive revision are required") + } + if len(r.Routes) > MaxRoutes { + return fmt.Errorf("%d routes exceed the limit of %d", len(r.Routes), MaxRoutes) + } + for _, p := range r.Routes { + if !p.IsValid() || !p.Addr().Is6() || p.Addr().Is4In6() || p != p.Masked() { + return fmt.Errorf("route %s is not a masked IPv6 prefix", p) + } + } + return nil +} + +// decodeRecord reads the record of a grant Secret. +func decodeRecord(s *corev1.Secret) (*Record, error) { + var r Record + if err := json.Unmarshal(s.Data[recordKey], &r); err != nil { + return nil, fmt.Errorf("decode grant record %s: %w", s.Name, err) + } + if err := r.validate(); err != nil { + return nil, fmt.Errorf("grant record %s: %w", s.Name, err) + } + if s.Name != RecordName(r.Connector) || s.Labels[LabelConnector] != r.Connector { + return nil, fmt.Errorf("grant record %s is not named for connector %s", s.Name, r.Connector) + } + return &r, nil +} + +// encodeRecord stores r in s. +func encodeRecord(s *corev1.Secret, r *Record) error { + if err := r.validate(); err != nil { + return err + } + b, err := json.Marshal(r) + if err != nil { + return err + } + s.Name = RecordName(r.Connector) + if s.Labels == nil { + s.Labels = map[string]string{} + } + s.Labels[LabelGrant] = yes + s.Labels[LabelConnector] = r.Connector + s.Type = corev1.SecretTypeOpaque + s.Data = map[string][]byte{recordKey: b} + return nil +} + +// readRecord reads connector's record from namespace through reader. It +// returns the Secret too, for a conditional update. +func readRecord(ctx context.Context, reader client.Reader, namespace, connector string) (*Record, *corev1.Secret, + error) { + var s corev1.Secret + if err := reader.Get(ctx, client.ObjectKey{Namespace: namespace, Name: RecordName(connector)}, &s); err != nil { + return nil, nil, err + } + r, err := decodeRecord(&s) + return r, &s, err +} diff --git a/internal/edge/masque/authority/server.go b/internal/edge/masque/authority/server.go new file mode 100644 index 00000000..44fe8bd6 --- /dev/null +++ b/internal/edge/masque/authority/server.go @@ -0,0 +1,199 @@ +// Copyright 2026 Datum Cloud, Inc. +// +// SPDX-License-Identifier: AGPL-3.0-or-later + +package authority + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "log/slog" + "net/http" + "slices" + "strings" + "time" + + authenticationv1 "k8s.io/api/authentication/v1" + "sigs.k8s.io/controller-runtime/pkg/client" + + "go.datum.net/galactic/internal/edge/admission" + "go.datum.net/galactic/internal/edge/connectip" +) + +const ( + // DefaultAudience is the token audience provider pods present. + DefaultAudience = "masque-authority" + // requestTimeout bounds one call, API reads and writes included. + requestTimeout = 2 * time.Second + // maxBody bounds a request body. + maxBody = 8 << 10 +) + +// API paths. +const ( + PathAuthorize = "/v1/authorize" + PathRefresh = "/v1/refresh" + PathRetire = "/v1/retire" +) + +// AuthorizeRequest is the body of an Authorize call. +type AuthorizeRequest struct { + Credential string `json:"credential"` + VPC string `json:"vpc"` + // Nonce identifies the call: its retries send the same nonce and + // issue time. + Nonce string `json:"nonce"` + // Issued is when the caller first sent the call. + Issued time.Time `json:"issued"` +} + +// RefreshRequest is the body of a Refresh call. +type RefreshRequest struct { + Authorization string `json:"authorization"` + Revision int64 `json:"revision"` + Service ServiceRef `json:"service"` +} + +// RetireRequest is the body of a Retire call. +type RetireRequest struct { + Authorization string `json:"authorization"` + Service ServiceRef `json:"service"` +} + +// errorBody is the body of a refused or failed call. +type errorBody struct { + Error string `json:"error"` +} + +// TokenReviewer authenticates a bearer token. +type TokenReviewer interface { + // Review returns the user the token authenticates, or an error. + Review(ctx context.Context, token string) (authenticationv1.UserInfo, error) +} + +// KubeTokenReviewer reviews tokens with the API server's TokenReview for +// Audience. +type KubeTokenReviewer struct { + Client client.Client + Audience string +} + +// Review implements TokenReviewer. A token not issued for Audience is +// refused. +func (k KubeTokenReviewer) Review(ctx context.Context, token string) (authenticationv1.UserInfo, error) { + tr := &authenticationv1.TokenReview{Spec: authenticationv1.TokenReviewSpec{ + Token: token, Audiences: []string{k.Audience}, + }} + if err := k.Client.Create(ctx, tr); err != nil { + return authenticationv1.UserInfo{}, fmt.Errorf("%w: review token: %w", ErrUnavailable, err) + } + if !tr.Status.Authenticated || !slices.Contains(tr.Status.Audiences, k.Audience) { + return authenticationv1.UserInfo{}, fmt.Errorf("%w: token not authenticated for %s", + connectip.ErrUnauthenticated, k.Audience) + } + return tr.Status.User, nil +} + +// Server serves an Authority over HTTP. Run it behind TLS. +type Server struct { + Authority *Authority + Reviewer TokenReviewer + Log *slog.Logger +} + +// Handler returns the server's routes. +func (s *Server) Handler() http.Handler { + mux := http.NewServeMux() + mux.HandleFunc("POST "+PathAuthorize, s.handle(func(ctx context.Context, id admission.Identity, + dec *json.Decoder) (any, error) { + var req AuthorizeRequest + if err := dec.Decode(&req); err != nil { + return nil, fmt.Errorf("%w: %w", ErrInvalid, err) + } + return s.Authority.Authorize(ctx, id, req) + })) + mux.HandleFunc("POST "+PathRefresh, s.handle(func(ctx context.Context, id admission.Identity, + dec *json.Decoder) (any, error) { + var req RefreshRequest + if err := dec.Decode(&req); err != nil { + return nil, fmt.Errorf("%w: %w", ErrInvalid, err) + } + return s.Authority.Refresh(ctx, id, req.Authorization, req.Revision, req.Service) + })) + mux.HandleFunc("POST "+PathRetire, s.handle(func(ctx context.Context, id admission.Identity, + dec *json.Decoder) (any, error) { + var req RetireRequest + if err := dec.Decode(&req); err != nil { + return nil, fmt.Errorf("%w: %w", ErrInvalid, err) + } + return struct{}{}, s.Authority.Retire(ctx, id, req.Authorization, req.Service) + })) + return mux +} + +// handle authenticates the calling pod, bounds the call and writes call's +// result or error. Request bodies, which may carry a credential, are never +// logged. +func (s *Server) handle(call func(context.Context, admission.Identity, *json.Decoder) (any, error)) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + ctx, cancel := context.WithTimeout(r.Context(), requestTimeout) + defer cancel() + res, err := s.call(ctx, r, call) + if err != nil { + code := status(err) + s.log().Info("call refused", "path", r.URL.Path, "status", code, "error", err) + writeJSON(w, code, errorBody{Error: http.StatusText(code)}) + return + } + writeJSON(w, http.StatusOK, res) + } +} + +func (s *Server) call(ctx context.Context, r *http.Request, + call func(context.Context, admission.Identity, *json.Decoder) (any, error)) (any, error) { + token, ok := strings.CutPrefix(r.Header.Get("Authorization"), "Bearer ") + if !ok || token == "" { + return nil, fmt.Errorf("%w: no bearer token", connectip.ErrUnauthenticated) + } + user, err := s.Reviewer.Review(ctx, token) + if err != nil { + return nil, err + } + id, err := admission.IdentityFromUserInfo(user) + if err != nil { + return nil, fmt.Errorf("%w: %w", connectip.ErrUnauthenticated, err) + } + dec := json.NewDecoder(http.MaxBytesReader(nil, r.Body, maxBody)) + dec.DisallowUnknownFields() + return call(ctx, id, dec) +} + +func (s *Server) log() *slog.Logger { + if s.Log != nil { + return s.Log + } + return slog.Default() +} + +// status is the HTTP status of a call's error. +func status(err error) int { + switch { + case errors.Is(err, connectip.ErrUnauthenticated): + return http.StatusUnauthorized + case errors.Is(err, connectip.ErrForbidden): + return http.StatusForbidden + case errors.Is(err, ErrInvalid): + return http.StatusBadRequest + default: + return http.StatusServiceUnavailable + } +} + +func writeJSON(w http.ResponseWriter, code int, v any) { + w.Header().Set("Content-Type", "application/json") + w.Header().Set("Cache-Control", "no-store") + w.WriteHeader(code) + _ = json.NewEncoder(w).Encode(v) +} diff --git a/internal/edge/masque/grant.go b/internal/edge/masque/grant.go index 154fb617..19d3d572 100644 --- a/internal/edge/masque/grant.go +++ b/internal/edge/masque/grant.go @@ -49,6 +49,11 @@ type Grant struct { Routes []netip.Prefix // NotAfter is when the grant stops authorizing forwarding. NotAfter time.Time + // Authorization identifies the authorization incarnation the authority + // issued; Refresh and Retire name it. + Authorization string + // Revision is the credential revision the grant was issued under. + Revision int64 } // Authorizer authenticates a CONNECT-IP request's credential and authorizes @@ -73,11 +78,13 @@ type Authorizer interface { } // sameAuthority reports whether a refreshed grant r still authorizes what g -// did: the same connector, principal, VPC, service, /96 and routes. Only -// the deadline may change. +// did: the same connector, principal, VPC, service, /96, routes, +// authorization incarnation and credential revision. Only the deadline may +// change. func (g *Grant) sameAuthority(r *Grant) bool { return g.Connector == r.Connector && g.Principal == r.Principal && g.VPC == r.VPC && g.VPCUID == r.VPCUID && - g.VPCID == r.VPCID && g.Service == r.Service && g.Prefix == r.Prefix && slices.Equal(g.Routes, r.Routes) + g.VPCID == r.VPCID && g.Service == r.Service && g.Prefix == r.Prefix && slices.Equal(g.Routes, r.Routes) && + g.Authorization == r.Authorization && g.Revision == r.Revision } // validate reports why g cannot admit a request for vpc at now. diff --git a/internal/logging/logging.go b/internal/logging/logging.go index ac7f0738..2bbf516c 100644 --- a/internal/logging/logging.go +++ b/internal/logging/logging.go @@ -60,6 +60,7 @@ var Binaries = []string{ "galactic-nat", "galactic-router", "galactic-vrf", + "masque-authority", } // Options holds the --log-level value and controller-runtime's standard zap diff --git a/tests/edgeconformance/masqueprovider/config.go b/tests/edgeconformance/masqueprovider/config.go index 94a592be..c508c34a 100644 --- a/tests/edgeconformance/masqueprovider/config.go +++ b/tests/edgeconformance/masqueprovider/config.go @@ -13,6 +13,7 @@ import ( "fmt" "net/netip" "os" + "strings" "sync" "time" @@ -20,6 +21,7 @@ import ( "go.datum.net/galactic/internal/edge/connectip" "go.datum.net/galactic/internal/edge/masque" + "go.datum.net/galactic/internal/edge/masque/authority" ) // Config is the fixture's configuration file. @@ -35,6 +37,10 @@ type Config struct { Services []Service `json:"services"` // Connectors are the CONNECT-IP clients the fixture admits. Connectors []Connector `json:"connectors"` + // GrantAuthority, when set, authorizes requests through a MASQUE grant + // authority instead of Services and Connectors, which must then be + // empty. + GrantAuthority *GrantAuthority `json:"grantAuthority,omitempty"` mu sync.Mutex // retired holds the IDs of connectors whose grant was retired; the @@ -72,6 +78,17 @@ type Connector struct { Routes []netip.Prefix `json:"routes"` } +// GrantAuthority locates a MASQUE grant authority. +type GrantAuthority struct { + // URL is the authority's https:// base URL. + URL string `json:"url"` + // CAFile holds the roots the authority's certificate chains to. + CAFile string `json:"caFile"` + // TokenFile holds the pod's projected ServiceAccount token for the + // authority's audience. + TokenFile string `json:"tokenFile"` +} + // Key is how a connector names the service: its namespace and name. func (s Service) Key() string { return s.Namespace + "/" + s.Name } @@ -93,6 +110,9 @@ func (c *Config) Validate() error { if c.Provider == "" || c.Authority == "" || c.Listen == "" { return errors.New("provider, authority and listen are required") } + if c.GrantAuthority != nil { + return c.validateGrantAuthority() + } services := map[string]bool{} for _, s := range c.Services { if s.Namespace == "" || s.Name == "" || s.UID == "" || s.VPCUID == "" || s.VPCID == "" || s.VPC == "" { @@ -130,6 +150,28 @@ func (c *Config) Validate() error { return nil } +// Authorizer is the fixture's masque.Authorizer: the grant authority when +// one is configured, and the static connectors otherwise. +func (c *Config) Authorizer() masque.Authorizer { + if g := c.GrantAuthority; g != nil { + return &authority.Client{URL: g.URL, CAFile: g.CAFile, TokenFile: g.TokenFile} + } + return c +} + +// validateGrantAuthority reports why the grant authority configuration is +// unusable. +func (c *Config) validateGrantAuthority() error { + g := c.GrantAuthority + switch { + case !strings.HasPrefix(g.URL, "https://") || g.CAFile == "" || g.TokenFile == "": + return errors.New("grantAuthority needs an https:// url, a caFile and a tokenFile") + case len(c.Services) > 0 || len(c.Connectors) > 0: + return errors.New("services and connectors come from the grant authority when it is set") + } + 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. diff --git a/tests/edgeconformance/masqueprovider/main.go b/tests/edgeconformance/masqueprovider/main.go index c663741c..4873292d 100644 --- a/tests/edgeconformance/masqueprovider/main.go +++ b/tests/edgeconformance/masqueprovider/main.go @@ -13,8 +13,10 @@ // packets on the packet device galactic-edge creates in the pod. It needs // no host namespace, hostPath, NET_ADMIN or BPF. // -// 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. +// By default 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. With grantAuthority set, the fixture authorizes through a MASQUE +// grant authority (cmd/masque-authority) instead. // // SIGTERM stops the server, which ends every session, then the adapter // releases the routes, deletes the attachments and releases the lease. @@ -98,7 +100,7 @@ func run() error { 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, + Client: c, Provider: cfg.Provider, Pod: pod, Authorizer: cfg.Authorizer(), Claims: claims, OpenDevice: masque.OpenTUN, Log: log.With("podName", pod.Name), }) if err != nil { diff --git a/tests/edgeconformance/masqueprovider/provider_test.go b/tests/edgeconformance/masqueprovider/provider_test.go index 0999f5e7..bce13314 100644 --- a/tests/edgeconformance/masqueprovider/provider_test.go +++ b/tests/edgeconformance/masqueprovider/provider_test.go @@ -16,6 +16,7 @@ import ( "go.datum.net/galactic/internal/edge/connectip" "go.datum.net/galactic/internal/edge/masque" + "go.datum.net/galactic/internal/edge/masque/authority" ) const ( @@ -38,10 +39,22 @@ func validConfig() *Config { } } +// tokenPath is a token file path in configurations under test. +const tokenPath = "token" + func TestConfigValidate(t *testing.T) { if err := validConfig().Validate(); err != nil { t.Fatalf("identical /96s in different services: %v", err) } + ga := validConfig() + ga.Services, ga.Connectors = nil, nil + ga.GrantAuthority = &GrantAuthority{URL: "https://authority", CAFile: "ca", TokenFile: tokenPath} + if err := ga.Validate(); err != nil { + t.Fatalf("grant authority configuration: %v", err) + } + if _, ok := ga.Authorizer().(*authority.Client); !ok { + t.Fatalf("Authorizer() = %T with a grant authority, want *authority.Client", ga.Authorizer()) + } for _, tc := range []struct { name string mutate func(*Config) @@ -60,6 +73,13 @@ func TestConfigValidate(t *testing.T) { {"unmasked route", func(c *Config) { c.Connectors[0].Routes = []netip.Prefix{netip.MustParsePrefix("fd20::1/32")} }, "route"}, + {"grant authority beside static connectors", func(c *Config) { + c.GrantAuthority = &GrantAuthority{URL: "https://authority", CAFile: "ca", TokenFile: tokenPath} + }, "come from the grant authority"}, + {"grant authority over plain HTTP", func(c *Config) { + c.Services, c.Connectors = nil, nil + c.GrantAuthority = &GrantAuthority{URL: "http://authority", CAFile: "ca", TokenFile: tokenPath} + }, "https://"}, } { t.Run(tc.name, func(t *testing.T) { c := validConfig()