feat(edge): Add the MASQUE grant authority

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

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
 

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.

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.