From bfacca7789353b56df4d8cdb48a450e4236fea50 Mon Sep 17 00:00:00 2001 From: Josh Siviour Date: Thu, 4 Jun 2026 11:25:56 +1000 Subject: [PATCH] =?UTF-8?q?feat(eks):=206c=20token=20webhook=20+=20AccessE?= =?UTF-8?q?ntry=20(IAM=20auth=20=E2=86=92=20kubectl)=20(#284)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat(eks): 6c AccessEntry CRUD + access-policy association Replace the nine notImpl() AccessEntry/AccessPolicy stubs with real KV-backed handlers (Q9, API auth mode): - AccessEntry CRUD persisted at clusters/{name}/access-entries/{hash}; the principal ARN is hashed into the key (ARNs carry ':', illegal in NATS KV keys) while the record keeps the plaintext ARN. - AssociateAccessPolicy/Disassociate/ListAssociated against the four supported AWS-managed policies; unsupported policy ARN or bad access scope -> InvalidParameterValue. - ListAccessPolicies returns the static supported catalogue. - Cluster-existence gate -> ResourceNotFoundException; duplicate create -> ResourceInUseException; node-type entries rejected until 6d. Updates the daemon routing test for the now-implemented handlers. * feat(eks): plumb caller principal ARN + bootstrap-creator-admin Thread the caller's resolved IAM principal ARN from the gateway to the EKS control plane via a new X-Principal-ARN NATS header, so CreateCluster can mint the bootstrapClusterCreatorAdminPermissions AccessEntry (Q9): - utils.NATSRequest takes optional extra headers; PrincipalARNHeader + PrincipalARNFromMsg added. - Gateway resolves the caller ARN best-effort (buildCallerARN) and passes it down the EKS route table; only CreateCluster consumes it. - CreateCluster signature gains callerPrincipalARN; default-true bootstrap flag mints a system:masters entry keyed by the caller's exact ARN (the token webhook resolves the same ARN from STS). '' degrades gracefully. - Dedicated daemon handler reads both headers. * feat(eks): IAM token webhook + STS verify-over-NATS + apiserver wiring The eks-token-webhook (baked into the eks-server AMI) authenticates kubectl requests that present an `aws eks get-token` bearer token: decode the presigned STS GetCallerIdentity URL, verify it, resolve the principal ARN to its cluster AccessEntry, and answer the apiserver TokenReview with the mapped username + groups. STS is hosted in-process in the awsgw service and is not otherwise on the bus, so awsgw now serves an `eks.VerifyToken` NATS subject that wraps VerifyPresignedGetCallerIdentity (the cluster name is forwarded so STS enforces the x-k8s-aws-id anti-replay pin). The webhook reads the AccessEntry KV directly over its own NATS connection. Apiserver wiring: the k3s config skeleton sets authentication-token-webhook-config-file; the webhook writes its own self-signed serving cert + the apiserver kubeconfig at startup (ordered `before k3s`, with a bounded wait in the k3s start_pre so the apiserver never loads a missing file). * fix(eks): omit empty spec from token webhook TokenReview response * fix(sts): presigned GetCallerIdentity must hash empty body, not UNSIGNED-PAYLOAD VerifyPresignedGetCallerIdentity built the SigV4 canonical request with "UNSIGNED-PAYLOAD" as the payload hash. That is an S3-only convention; botocore and the aws-sdk presign non-S3 requests (sts:GetCallerIdentity, the shape `aws eks get-token` produces) with the SHA256 of the empty string. Every real get-token therefore failed signature verification with InvalidIdentityToken, breaking the entire EKS IAM token-auth flow end to end. Use the empty-string SHA256. The existing hand-rolled presign tests masked the bug by signing with the same wrong constant; they now sign correctly, and a new test presigns with the real aws-sdk-go v4 signer (the same canonicalisation the CLI uses) so this can't regress silently. Verified live: a real `aws eks get-token --cluster-name dev1` now resolves to the caller principal over the awsgw eks.VerifyToken surface; an otherwise-valid token presented for a different cluster is still rejected (anti-replay intact). * fix(eks): match access-entry routes on escaped path so slashed principal ARNs resolve IAM principal ARNs (user/admin, role/foo) embed a '/' that the AWS CLI percent-encodes in the request path. The dispatcher matched r.URL.Path (decoded), so %2F became a literal '/' and split the ARN across the ([^/]+) capture, dropping every describe/update/delete-access-entry and associate/disassociate/list-access-policy call to InvalidActionException. Match r.URL.EscapedPath() so the encoded ARN stays one segment, then url.PathUnescape each captured param before it reaches the handler. * test(eks): unit-cover access-entry scope validation, AWS converters, store guards --- cmd/eks-token-webhook/cert.go | 137 ++++++++ cmd/eks-token-webhook/main.go | 253 +++++++++++++- cmd/eks-token-webhook/main_test.go | 123 +++++++ cmd/eks-token-webhook/tokenreview.go | 29 ++ .../images/eks-server/eks-token-webhook.initd | 21 +- scripts/images/eks-server/k3s.initd | 13 + scripts/images/eks-server/setup.sh | 15 +- spinifex/daemon/daemon_handlers.go | 22 ++ spinifex/daemon/daemon_handlers_eks.go | 2 +- spinifex/daemon/daemon_handlers_eks_test.go | 96 +++--- spinifex/gateway/eks.go | 123 ++++--- spinifex/gateway/eks/cluster.go | 8 +- spinifex/gateway/eks/wrappers_test.go | 6 +- spinifex/gateway/eks_test.go | 31 ++ spinifex/handlers/eks/access_entry.go | 263 +++++++++++++++ spinifex/handlers/eks/access_entry_test.go | 95 ++++++ spinifex/handlers/eks/create_cluster_test.go | 49 ++- spinifex/handlers/eks/service.go | 6 +- spinifex/handlers/eks/service_impl.go | 319 +++++++++++++++++- spinifex/handlers/eks/service_impl_test.go | 174 ++++++++-- spinifex/handlers/eks/service_nats.go | 5 +- spinifex/handlers/eks/store.go | 18 +- spinifex/handlers/eks/store_test.go | 2 +- spinifex/handlers/eks/token_webhook.go | 58 ++++ spinifex/handlers/eks/token_webhook_test.go | 44 +++ spinifex/handlers/eks/types.go | 33 +- spinifex/handlers/sts/presigned_url_verify.go | 10 +- .../handlers/sts/presigned_url_verify_test.go | 42 ++- spinifex/services/awsgw/awsgw.go | 13 + spinifex/services/awsgw/eks_token_verify.go | 76 +++++ .../services/awsgw/eks_token_verify_test.go | 79 +++++ spinifex/utils/nats.go | 31 +- 32 files changed, 2019 insertions(+), 177 deletions(-) create mode 100644 cmd/eks-token-webhook/cert.go create mode 100644 cmd/eks-token-webhook/main_test.go create mode 100644 cmd/eks-token-webhook/tokenreview.go create mode 100644 spinifex/handlers/eks/access_entry.go create mode 100644 spinifex/handlers/eks/access_entry_test.go create mode 100644 spinifex/handlers/eks/token_webhook.go create mode 100644 spinifex/handlers/eks/token_webhook_test.go create mode 100644 spinifex/services/awsgw/eks_token_verify.go create mode 100644 spinifex/services/awsgw/eks_token_verify_test.go diff --git a/cmd/eks-token-webhook/cert.go b/cmd/eks-token-webhook/cert.go new file mode 100644 index 000000000..3a4763197 --- /dev/null +++ b/cmd/eks-token-webhook/cert.go @@ -0,0 +1,137 @@ +package main + +import ( + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" + "crypto/tls" + "crypto/x509" + "crypto/x509/pkix" + "encoding/base64" + "encoding/pem" + "fmt" + "math/big" + "net" + "os" + "time" + + "github.com/mulgadc/spinifex/internal/tlsconfig" +) + +// tlsCertificate is the webhook's loopback serving certificate. +type tlsCertificate struct { + cert tls.Certificate +} + +func (c tlsCertificate) serverTLSConfig() *tls.Config { + return &tls.Config{ + Certificates: []tls.Certificate{c.cert}, + MinVersion: tls.VersionTLS13, + CurvePreferences: tlsconfig.Curves, + } +} + +// ensureServingCert loads the persisted self-signed serving cert/key, or mints +// a fresh ECDSA-P256 pair (SAN 127.0.0.1 / localhost) and persists it. Reusing +// across restarts keeps the CA the apiserver loaded into its webhook kubeconfig +// valid even if the webhook process is restarted. Returns the parsed cert and +// the certificate PEM (for embedding as the kubeconfig CA bundle). +func ensureServingCert(certPath, keyPath string) (tlsCertificate, []byte, error) { + if certPEM, keyPEM, ok := readCertPair(certPath, keyPath); ok { + cert, err := tls.X509KeyPair(certPEM, keyPEM) + if err == nil { + return tlsCertificate{cert: cert}, certPEM, nil + } + // Fall through to regenerate on a corrupt/incompatible persisted pair. + } + + certPEM, keyPEM, err := generateSelfSigned() + if err != nil { + return tlsCertificate{}, nil, err + } + if err := os.WriteFile(certPath, certPEM, 0o644); err != nil { + return tlsCertificate{}, nil, fmt.Errorf("write cert %s: %w", certPath, err) + } + if err := os.WriteFile(keyPath, keyPEM, 0o600); err != nil { + return tlsCertificate{}, nil, fmt.Errorf("write key %s: %w", keyPath, err) + } + cert, err := tls.X509KeyPair(certPEM, keyPEM) + if err != nil { + return tlsCertificate{}, nil, fmt.Errorf("parse minted keypair: %w", err) + } + return tlsCertificate{cert: cert}, certPEM, nil +} + +func readCertPair(certPath, keyPath string) (certPEM, keyPEM []byte, ok bool) { + c, err := os.ReadFile(certPath) + if err != nil { + return nil, nil, false + } + k, err := os.ReadFile(keyPath) + if err != nil { + return nil, nil, false + } + return c, k, true +} + +func generateSelfSigned() (certPEM, keyPEM []byte, err error) { + priv, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + if err != nil { + return nil, nil, fmt.Errorf("generate key: %w", err) + } + serial, err := rand.Int(rand.Reader, new(big.Int).Lsh(big.NewInt(1), 128)) + if err != nil { + return nil, nil, fmt.Errorf("serial: %w", err) + } + now := time.Now().UTC() + tmpl := x509.Certificate{ + SerialNumber: serial, + Subject: pkix.Name{CommonName: "eks-token-webhook"}, + NotBefore: now.Add(-time.Hour), + NotAfter: now.AddDate(10, 0, 0), + KeyUsage: x509.KeyUsageDigitalSignature | x509.KeyUsageKeyEncipherment | x509.KeyUsageCertSign, + ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth}, + BasicConstraintsValid: true, + IsCA: true, + IPAddresses: []net.IP{net.IPv4(127, 0, 0, 1), net.IPv6loopback}, + DNSNames: []string{"localhost"}, + } + der, err := x509.CreateCertificate(rand.Reader, &tmpl, &tmpl, &priv.PublicKey, priv) + if err != nil { + return nil, nil, fmt.Errorf("create cert: %w", err) + } + keyDER, err := x509.MarshalECPrivateKey(priv) + if err != nil { + return nil, nil, fmt.Errorf("marshal key: %w", err) + } + certPEM = pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: der}) + keyPEM = pem.EncodeToMemory(&pem.Block{Type: "EC PRIVATE KEY", Bytes: keyDER}) + return certPEM, keyPEM, nil +} + +// writeAPIServerKubeconfig writes the kubeconfig kube-apiserver loads via +// --authentication-token-webhook-config-file. It points the apiserver at the +// loopback webhook and trusts its self-signed serving cert as the CA bundle. +func writeAPIServerKubeconfig(path, addr string, certPEM []byte) error { + caB64 := base64.StdEncoding.EncodeToString(certPEM) + kubeconfig := fmt.Sprintf(`apiVersion: v1 +kind: Config +clusters: +- name: mulga-eks-token-webhook + cluster: + certificate-authority-data: %s + server: https://%s/authenticate +users: +- name: kube-apiserver +contexts: +- name: webhook + context: + cluster: mulga-eks-token-webhook + user: kube-apiserver +current-context: webhook +`, caB64, addr) + if err := os.WriteFile(path, []byte(kubeconfig), 0o600); err != nil { + return fmt.Errorf("write %s: %w", path, err) + } + return nil +} diff --git a/cmd/eks-token-webhook/main.go b/cmd/eks-token-webhook/main.go index b2b28ba55..5457d8c9c 100644 --- a/cmd/eks-token-webhook/main.go +++ b/cmd/eks-token-webhook/main.go @@ -1,38 +1,257 @@ // eks-token-webhook is the K8s TokenReview webhook authenticator baked into // the eks-server AMI. It decodes the SigV4-presigned GetCallerIdentity URL -// produced by `aws eks get-token`, enforces the X-K8s-Aws-Id cross-cluster -// pin, calls Mulga STS over NATS, and resolves the principal to a TokenReview -// response via the cluster's AccessEntry KV. +// produced by `aws eks get-token`, calls Mulga STS over NATS (which enforces +// the x-k8s-aws-id cross-cluster pin against this cluster's name), and resolves +// the principal ARN to a TokenReview response via the cluster's AccessEntry KV. // -// This is a placeholder build that returns 503 Service Unavailable on every -// path. It exists so the eks-server AMI build succeeds (INSTALL_BINARIES is -// satisfied) and the K3s server VM boots cleanly. Real implementation lands -// with bead mulga-cs-eks-6c. +// It binds 127.0.0.1 only; the kube-apiserver is the sole client (loopback). +// On startup it writes the apiserver webhook kubeconfig (with its self-signed +// serving CA) so k3s can be pointed at it via +// --authentication-token-webhook-config-file. package main import ( + "encoding/json" + "errors" "flag" + "fmt" "log/slog" "net/http" "os" + "time" + + handlers_eks "github.com/mulgadc/spinifex/spinifex/handlers/eks" + "github.com/mulgadc/spinifex/spinifex/utils" + "github.com/nats-io/nats.go" ) +// config is the webhook's runtime configuration, sourced from the first-boot +// env (/etc/spinifex-eks/first-boot.env) cloud-init seeds onto the VM. +type config struct { + addr string + natsURL string + natsToken string + natsCA string + accountID string + clusterName string + certPath string + keyPath string + kubeconfig string + verifyTO time.Duration +} + +func loadConfig() (config, error) { + c := config{ + addr: flagAddr, + natsURL: os.Getenv("SPINIFEX_NATS_URL"), + natsToken: os.Getenv("SPINIFEX_NATS_TOKEN"), + natsCA: os.Getenv("SPINIFEX_NATS_CA"), + accountID: os.Getenv("EKS_ACCOUNT_ID"), + clusterName: os.Getenv("EKS_CLUSTER_NAME"), + certPath: envOr("EKS_WEBHOOK_CERT", "/etc/spinifex-eks/token-webhook.crt"), + keyPath: envOr("EKS_WEBHOOK_KEY", "/etc/spinifex-eks/token-webhook.key"), + kubeconfig: envOr("EKS_WEBHOOK_KUBECONFIG", "/etc/spinifex-eks/token-webhook.kubeconfig"), + verifyTO: 5 * time.Second, + } + if c.natsURL == "" { + return c, errors.New("SPINIFEX_NATS_URL not set") + } + if c.accountID == "" { + return c, errors.New("EKS_ACCOUNT_ID not set") + } + if c.clusterName == "" { + return c, errors.New("EKS_CLUSTER_NAME not set") + } + return c, nil +} + +func envOr(key, def string) string { + if v := os.Getenv(key); v != "" { + return v + } + return def +} + +var flagAddr string + func main() { - addr := flag.String("addr", "127.0.0.1:8443", "listen address") + flag.StringVar(&flagAddr, "addr", "127.0.0.1:8443", "listen address (loopback only)") flag.Parse() slog.SetDefault(slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelInfo}))) - mux := http.NewServeMux() - mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) { - w.Header().Set("Retry-After", "30") - w.Header().Set("Content-Type", "application/json") - http.Error(w, `{"error":"eks-token-webhook stub: real implementation not yet deployed"}`, http.StatusServiceUnavailable) - }) + cfg, err := loadConfig() + if err != nil { + slog.Error("eks-token-webhook config error", "err", err) + os.Exit(1) + } - slog.Info("eks-token-webhook stub starting", "addr", *addr) - if err := http.ListenAndServe(*addr, mux); err != nil { - slog.Error("listen failed", "err", err) + if err := run(cfg); err != nil { + slog.Error("eks-token-webhook fatal", "err", err) os.Exit(1) } } + +func run(cfg config) error { + // Serving cert + apiserver kubeconfig. Persisted and reused across restarts + // so a webhook crash does not invalidate the CA the apiserver already loaded. + tlsCert, certPEM, err := ensureServingCert(cfg.certPath, cfg.keyPath) + if err != nil { + return fmt.Errorf("serving cert: %w", err) + } + if err := writeAPIServerKubeconfig(cfg.kubeconfig, cfg.addr, certPEM); err != nil { + return fmt.Errorf("write apiserver kubeconfig: %w", err) + } + + nc, err := utils.ConnectNATSWithRetry(cfg.natsURL, cfg.natsToken, cfg.natsCA) + if err != nil { + return fmt.Errorf("connect NATS: %w", err) + } + defer nc.Close() + + kv, err := openAccountKV(nc, cfg.accountID) + if err != nil { + return fmt.Errorf("open account KV: %w", err) + } + + authr := &authenticator{ + nc: nc, + kv: kv, + clusterName: cfg.clusterName, + verifyTO: cfg.verifyTO, + } + + mux := http.NewServeMux() + mux.HandleFunc("/healthz", func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte("ok")) + }) + mux.HandleFunc("/authenticate", authr.handle) + + server := newServer(cfg.addr, mux, tlsCert) + slog.Info("eks-token-webhook listening", "addr", cfg.addr, "cluster", cfg.clusterName) + if err := server.ListenAndServeTLS("", ""); err != nil && !errors.Is(err, http.ErrServerClosed) { + return fmt.Errorf("serve: %w", err) + } + return nil +} + +// openAccountKV attaches to the per-account EKS KV bucket. The bucket already +// exists by the time a control-plane VM is running (CreateCluster created it), +// but JetStream may briefly lag at boot, so retry a few times. +func openAccountKV(nc *nats.Conn, accountID string) (nats.KeyValue, error) { + js, err := nc.JetStream() + if err != nil { + return nil, fmt.Errorf("jetstream: %w", err) + } + bucket := handlers_eks.AccountBucketName(accountID) + var lastErr error + for range 30 { + kv, kvErr := js.KeyValue(bucket) + if kvErr == nil { + return kv, nil + } + lastErr = kvErr + time.Sleep(2 * time.Second) + } + return nil, fmt.Errorf("kv bucket %s not available: %w", bucket, lastErr) +} + +// authenticator resolves a TokenReview against STS (over NATS) + AccessEntry KV. +type authenticator struct { + nc *nats.Conn + kv nats.KeyValue + clusterName string + verifyTO time.Duration +} + +// verify resolves a presigned get-token URL into the caller principal via the +// awsgw-hosted STS verify subject, binding the signature to this cluster name. +func (a *authenticator) verify(presignedURL string) (*handlers_eks.TokenVerifyResponse, error) { + req := handlers_eks.TokenVerifyRequest{ + PresignedURL: presignedURL, + ClusterName: a.clusterName, + } + // accountID header is unused by the verify responder; pass empty. + return utils.NATSRequest[handlers_eks.TokenVerifyResponse]( + a.nc, handlers_eks.TokenVerifySubject, req, a.verifyTO, "") +} + +// lookup reads the AccessEntry for a principal ARN from the cluster KV. +func (a *authenticator) lookup(principalARN string) (*handlers_eks.AccessEntryRecord, error) { + return handlers_eks.GetAccessEntryRecord(a.kv, a.clusterName, principalARN) +} + +func (a *authenticator) handle(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + return + } + var review tokenReview + if err := json.NewDecoder(r.Body).Decode(&review); err != nil { + http.Error(w, "bad request", http.StatusBadRequest) + return + } + + status := authenticate(review.Spec.Token, a.verify, a.lookup) + + resp := tokenReview{ + APIVersion: "authentication.k8s.io/v1", + Kind: "TokenReview", + Status: status, + } + w.Header().Set("Content-Type", "application/json") + if err := json.NewEncoder(w).Encode(resp); err != nil { + slog.Error("encode TokenReview response", "err", err) + } +} + +// authenticate is the pure decision core: decode token → STS verify → KV +// AccessEntry lookup → TokenReview status. Any failure yields an unauthenticated +// status (never a 5xx) so a forged/unknown token is a clean K8s 401, not a +// webhook outage. verify and lookup are injected so the logic is unit-testable. +func authenticate( + token string, + verify func(presignedURL string) (*handlers_eks.TokenVerifyResponse, error), + lookup func(principalARN string) (*handlers_eks.AccessEntryRecord, error), +) tokenReviewStatus { + presignedURL, err := handlers_eks.DecodeGetToken(token) + if err != nil { + return tokenReviewStatus{Authenticated: false} + } + ident, err := verify(presignedURL) + if err != nil { + slog.Debug("token verify rejected", "err", err) + return tokenReviewStatus{Authenticated: false} + } + rec, err := lookup(ident.ARN) + if err != nil { + // No AccessEntry for an otherwise-valid IAM principal: authenticated to + // AWS, but not granted any K8s identity on this cluster. + slog.Debug("no access entry for principal", "arn", ident.ARN, "err", err) + return tokenReviewStatus{Authenticated: false} + } + uid := ident.UserID + if uid == "" { + uid = ident.ARN + } + return tokenReviewStatus{ + Authenticated: true, + User: userInfo{ + Username: rec.KubernetesUsername, + UID: uid, + Groups: rec.KubernetesGroups, + }, + } +} + +// newServer builds the loopback TLS server. Split out so tests can construct it +// without binding a port. +func newServer(addr string, handler http.Handler, cert tlsCertificate) *http.Server { + return &http.Server{ + Addr: addr, + Handler: handler, + ReadHeaderTimeout: 5 * time.Second, + TLSConfig: cert.serverTLSConfig(), + } +} diff --git a/cmd/eks-token-webhook/main_test.go b/cmd/eks-token-webhook/main_test.go new file mode 100644 index 000000000..8b35e5e64 --- /dev/null +++ b/cmd/eks-token-webhook/main_test.go @@ -0,0 +1,123 @@ +package main + +import ( + "encoding/base64" + "errors" + "os" + "path/filepath" + "testing" + + handlers_eks "github.com/mulgadc/spinifex/spinifex/handlers/eks" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func validToken(url string) string { + return "k8s-aws-v1." + base64.RawURLEncoding.EncodeToString([]byte(url)) +} + +const testARN = "arn:aws:iam::111122223333:role/admin" + +func okVerify(_ string) (*handlers_eks.TokenVerifyResponse, error) { + return &handlers_eks.TokenVerifyResponse{ + AccountID: "111122223333", + ARN: testARN, + UserID: "AROAEXAMPLE:session", + PrincipalType: "AssumedRole", + }, nil +} + +func TestAuthenticate_GrantsWhenEntryExists(t *testing.T) { + lookup := func(arn string) (*handlers_eks.AccessEntryRecord, error) { + assert.Equal(t, testARN, arn) + return &handlers_eks.AccessEntryRecord{ + KubernetesUsername: testARN, + KubernetesGroups: []string{"system:masters"}, + }, nil + } + + st := authenticate(validToken("https://sts.amazonaws.com/?Action=GetCallerIdentity"), okVerify, lookup) + + require.True(t, st.Authenticated) + assert.Equal(t, testARN, st.User.Username) + assert.Equal(t, "AROAEXAMPLE:session", st.User.UID) + assert.Equal(t, []string{"system:masters"}, st.User.Groups) +} + +func TestAuthenticate_DeniesMalformedToken(t *testing.T) { + called := false + verify := func(string) (*handlers_eks.TokenVerifyResponse, error) { called = true; return nil, nil } + lookup := func(string) (*handlers_eks.AccessEntryRecord, error) { return nil, nil } + + st := authenticate("not-a-k8s-aws-token", verify, lookup) + + assert.False(t, st.Authenticated) + assert.False(t, called, "must not call STS for a token that fails to decode") +} + +func TestAuthenticate_DeniesWhenVerifyFails(t *testing.T) { + verify := func(string) (*handlers_eks.TokenVerifyResponse, error) { + return nil, errors.New("signature mismatch") + } + lookup := func(string) (*handlers_eks.AccessEntryRecord, error) { + t.Fatal("lookup must not run when verify fails") + return nil, nil + } + + st := authenticate(validToken("https://sts/?x=1"), verify, lookup) + assert.False(t, st.Authenticated) +} + +func TestAuthenticate_DeniesWhenNoAccessEntry(t *testing.T) { + lookup := func(string) (*handlers_eks.AccessEntryRecord, error) { + return nil, handlers_eks.ErrAccessEntryNotFound + } + + st := authenticate(validToken("https://sts/?x=1"), okVerify, lookup) + assert.False(t, st.Authenticated) + assert.Empty(t, st.User.Username) +} + +func TestAuthenticate_FallsBackUIDToARN(t *testing.T) { + verify := func(string) (*handlers_eks.TokenVerifyResponse, error) { + return &handlers_eks.TokenVerifyResponse{ARN: testARN}, nil // no UserID + } + lookup := func(string) (*handlers_eks.AccessEntryRecord, error) { + return &handlers_eks.AccessEntryRecord{KubernetesUsername: testARN, KubernetesGroups: []string{"system:masters"}}, nil + } + + st := authenticate(validToken("https://sts/?x=1"), verify, lookup) + require.True(t, st.Authenticated) + assert.Equal(t, testARN, st.User.UID) +} + +func TestEnsureServingCert_PersistsAndReuses(t *testing.T) { + dir := t.TempDir() + certPath := filepath.Join(dir, "wh.crt") + keyPath := filepath.Join(dir, "wh.key") + + c1, pem1, err := ensureServingCert(certPath, keyPath) + require.NoError(t, err) + require.NotEmpty(t, pem1) + require.NotNil(t, c1.serverTLSConfig()) + + // Second call reuses the persisted pair (same cert bytes). + _, pem2, err := ensureServingCert(certPath, keyPath) + require.NoError(t, err) + assert.Equal(t, pem1, pem2) +} + +func TestWriteAPIServerKubeconfig(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "wh.kubeconfig") + certPEM := []byte("-----BEGIN CERTIFICATE-----\nMIIB\n-----END CERTIFICATE-----\n") + + require.NoError(t, writeAPIServerKubeconfig(path, "127.0.0.1:8443", certPEM)) + + data, err := os.ReadFile(path) + require.NoError(t, err) + body := string(data) + assert.Contains(t, body, "server: https://127.0.0.1:8443/authenticate") + assert.Contains(t, body, base64.StdEncoding.EncodeToString(certPEM)) + assert.Contains(t, body, "current-context: webhook") +} diff --git a/cmd/eks-token-webhook/tokenreview.go b/cmd/eks-token-webhook/tokenreview.go new file mode 100644 index 000000000..a3610f778 --- /dev/null +++ b/cmd/eks-token-webhook/tokenreview.go @@ -0,0 +1,29 @@ +package main + +// Minimal authentication.k8s.io/v1 TokenReview wire types. Hand-rolled to avoid +// pulling k8s.io/api into the spinifex module for these few fields; the shapes +// match what kube-apiserver sends and expects on the webhook contract. + +type tokenReview struct { + APIVersion string `json:"apiVersion"` + Kind string `json:"kind"` + Spec tokenReviewSpec `json:"spec,omitzero"` + Status tokenReviewStatus `json:"status"` +} + +type tokenReviewSpec struct { + Token string `json:"token"` + Audiences []string `json:"audiences,omitempty"` +} + +type tokenReviewStatus struct { + Authenticated bool `json:"authenticated"` + User userInfo `json:"user,omitzero"` + Error string `json:"error,omitempty"` +} + +type userInfo struct { + Username string `json:"username,omitempty"` + UID string `json:"uid,omitempty"` + Groups []string `json:"groups,omitempty"` +} diff --git a/scripts/images/eks-server/eks-token-webhook.initd b/scripts/images/eks-server/eks-token-webhook.initd index 9f18d2147..c6a752fd9 100644 --- a/scripts/images/eks-server/eks-token-webhook.initd +++ b/scripts/images/eks-server/eks-token-webhook.initd @@ -1,7 +1,10 @@ #!/sbin/openrc-run # eks-token-webhook — K8s TokenReview authenticator for IAM-backed kubectl. # Binds 127.0.0.1:8443 only; kube-apiserver is the sole client (loopback path). -# Stub binary returns 503 on every request until cs-eks-6c lands. +# Decodes `aws eks get-token` presigned URLs, verifies them via Mulga STS over +# NATS, and resolves the principal to a TokenReview via the cluster AccessEntry +# KV. Ordered `before k3s` so its apiserver kubeconfig exists before k3s reads +# --authentication-token-webhook-config-file. description="EKS Token Review webhook (Mulga)" command="/usr/local/bin/eks-token-webhook" @@ -11,7 +14,23 @@ pidfile="/run/eks-token-webhook.pid" output_log="/var/log/eks-token-webhook.log" error_log="/var/log/eks-token-webhook.log" +# NATS creds + cluster identity come from the cloud-init-seeded first-boot env. +# Sourced here and exported so start-stop-daemon hands them to the binary, which +# reads SPINIFEX_NATS_* / EKS_ACCOUNT_ID / EKS_CLUSTER_NAME from its environment. +if [ -f /etc/spinifex-eks/first-boot.env ]; then + # shellcheck disable=SC1091 + . /etc/spinifex-eks/first-boot.env + export SPINIFEX_NATS_URL SPINIFEX_NATS_TOKEN SPINIFEX_NATS_CA EKS_ACCOUNT_ID EKS_CLUSTER_NAME +fi + depend() { need net before k3s } + +start_pre() { + if [ ! -f /etc/spinifex-eks/first-boot.env ]; then + eerror "/etc/spinifex-eks/first-boot.env not present — cloud-init did not seed NATS creds" + return 1 + fi +} diff --git a/scripts/images/eks-server/k3s.initd b/scripts/images/eks-server/k3s.initd index 04ed37d3b..b48ea792e 100644 --- a/scripts/images/eks-server/k3s.initd +++ b/scripts/images/eks-server/k3s.initd @@ -33,4 +33,17 @@ start_pre() { return 1 fi fi + + # The apiserver is configured with --authentication-token-webhook-config-file + # pointing at the kubeconfig eks-token-webhook writes at startup. `before k3s` + # orders the webhook's start but it backgrounds, so wait (bounded) for the + # kubeconfig to land before letting the apiserver load a missing file. + i=0 + while [ ! -f /etc/spinifex-eks/token-webhook.kubeconfig ] && [ "${i}" -lt 30 ]; do + sleep 1 + i=$((i + 1)) + done + if [ ! -f /etc/spinifex-eks/token-webhook.kubeconfig ]; then + ewarn "eks-token-webhook kubeconfig absent after ${i}s; IAM token auth will be unavailable until the webhook starts" + fi } diff --git a/scripts/images/eks-server/setup.sh b/scripts/images/eks-server/setup.sh index c2a1a6be5..f47d4e02b 100644 --- a/scripts/images/eks-server/setup.sh +++ b/scripts/images/eks-server/setup.sh @@ -57,13 +57,13 @@ chmod 0755 /usr/local/sbin/k3s-first-boot chmod 0755 /etc/periodic/15min/mulga-eks-state-report chmod 0755 /etc/periodic/daily/mulga-eks-etcd-snapshot -# K3s server config — empty skeleton; cloud-init / first-boot fills in the +# K3s server config — skeleton; cloud-init / first-boot fills in the # per-cluster fields (cluster-cidr, service-cidr, token-file, etc). -# Webhook-token-auth is NOT wired in the skeleton: enabling it before the -# eks-token-webhook binary is the real implementation (currently a 503 stub, -# replaced by cs-eks-6c) would block /healthz from anonymous callers. The -# Sprint 6c orchestrator drops a webhook-config.yaml + restarts k3s once -# the binary is the real one. +# IAM token-auth is wired here via kube-apiserver-arg: the eks-token-webhook +# service (ordered `before k3s`) writes its kubeconfig to +# /etc/spinifex-eks/token-webhook.kubeconfig before the apiserver reads it. +# This only affects bearer-token requests; anonymous and client-cert paths +# (the first-boot /readyz probe uses the admin kubeconfig) are unaffected. mkdir -p /etc/rancher/k3s cat > /etc/rancher/k3s/config.yaml.skel <<'EOF' # Populated at first boot by cloud-init user-data via k3s-first-boot.sh. @@ -73,6 +73,9 @@ disable: - traefik - servicelb - local-storage +kube-apiserver-arg: + - "authentication-token-webhook-config-file=/etc/spinifex-eks/token-webhook.kubeconfig" + - "authentication-token-webhook-cache-ttl=5m" EOF # Sentinel file marker — k3s-first-boot self-disables after first success by diff --git a/spinifex/daemon/daemon_handlers.go b/spinifex/daemon/daemon_handlers.go index 2dd71a721..62be46a9d 100644 --- a/spinifex/daemon/daemon_handlers.go +++ b/spinifex/daemon/daemon_handlers.go @@ -58,6 +58,28 @@ func handleNATSRequest[I any, O any](msg *nats.Msg, serviceFn func(*I, string) ( respondWithJSON(msg, output) } +// handleNATSRequestWithPrincipal is handleNATSRequest for service methods that +// also need the caller's IAM principal ARN (X-Principal-ARN header) — e.g. EKS +// CreateCluster, which mints the bootstrap-creator-admin AccessEntry for the +// caller. +func handleNATSRequestWithPrincipal[I any, O any](msg *nats.Msg, serviceFn func(*I, string, string) (*O, error)) { + accountID := utils.AccountIDFromMsg(msg) + principalARN := utils.PrincipalARNFromMsg(msg) + input := new(I) + if errResp := utils.UnmarshalJsonPayload(input, msg.Data); errResp != nil { + if err := msg.Respond(errResp); err != nil { + slog.Error("Failed to respond to NATS request", "err", err) + } + return + } + output, err := serviceFn(input, accountID, principalARN) + if err != nil { + respondWithError(msg, awserrors.ValidErrorCode(err.Error())) + return + } + respondWithJSON(msg, output) +} + // handleEC2Events processes incoming EC2 instance events (start, stop, terminate, attach-volume) func (d *Daemon) handleEC2Events(msg *nats.Msg) { var command types.EC2InstanceCommand diff --git a/spinifex/daemon/daemon_handlers_eks.go b/spinifex/daemon/daemon_handlers_eks.go index 12147a168..ada154367 100644 --- a/spinifex/daemon/daemon_handlers_eks.go +++ b/spinifex/daemon/daemon_handlers_eks.go @@ -5,7 +5,7 @@ import "github.com/nats-io/nats.go" // --- Cluster --- func (d *Daemon) handleEKSCreateCluster(msg *nats.Msg) { - handleNATSRequest(msg, d.eksService.CreateCluster) + handleNATSRequestWithPrincipal(msg, d.eksService.CreateCluster) } func (d *Daemon) handleEKSDescribeCluster(msg *nats.Msg) { diff --git a/spinifex/daemon/daemon_handlers_eks_test.go b/spinifex/daemon/daemon_handlers_eks_test.go index 5d446bd2a..9c9623aa3 100644 --- a/spinifex/daemon/daemon_handlers_eks_test.go +++ b/spinifex/daemon/daemon_handlers_eks_test.go @@ -26,10 +26,10 @@ func setupEKSDaemon(t *testing.T) (*Daemon, *nats.Conn) { } // requestEKS publishes a request on the given subject and returns the response -// payload. The non-lifecycle EKS handlers all return NotImplemented so the -// payload always decodes to that error code; CreateCluster / DescribeCluster -// / ListClusters / DeleteCluster have real bodies and are covered by their -// own handlers/eks unit tests. +// payload. Proves a responder is wired for the subject and dispatched to the +// service. The expected Code per case is asserted by the caller; the empty `{}` +// body steers implemented handlers into a deterministic validation/dispatch +// error so the wiring is provable without a full orchestration graph. func requestEKS(t *testing.T, nc *nats.Conn, subject string) []byte { t.Helper() msg := nats.NewMsg(subject) @@ -40,53 +40,62 @@ func requestEKS(t *testing.T, nc *nats.Conn, subject string) []byte { return resp.Data } -func assertNotImpl(t *testing.T, payload []byte) { +// assertErrorCode decodes the error envelope and asserts the AWS error Code. +func assertErrorCode(t *testing.T, payload []byte, want string) { t.Helper() var env struct { Code *string `json:"Code"` } require.NoError(t, json.Unmarshal(payload, &env), "decode payload %q", payload) require.NotNil(t, env.Code, "payload %q has no Code field", payload) - assert.Equal(t, awserrors.ErrorNotImplemented, *env.Code) + assert.Equal(t, want, *env.Code) } func TestDaemonHandleEKS_AllHandlersDispatchToService(t *testing.T) { d, nc := setupEKSDaemon(t) + // wantCode is the error Code an empty-body request must elicit, proving the + // subject dispatched to its service method. Access-entry handlers are + // implemented (Sprint 6c): an empty body fails input/cluster validation with + // InvalidParameterValue. ListAccessPolicies returns the static catalogue with + // no error, so wantCode is "" (success — assert a non-error payload). + notImpl := awserrors.ErrorNotImplemented + invalid := awserrors.ErrorInvalidParameterValue cases := []struct { - subject string - handler nats.MsgHandler + subject string + handler nats.MsgHandler + wantCode string }{ - {"eks.UpdateClusterConfig", d.handleEKSUpdateClusterConfig}, - {"eks.UpdateClusterVersion", d.handleEKSUpdateClusterVersion}, - {"eks.CreateNodegroup", d.handleEKSCreateNodegroup}, - {"eks.DescribeNodegroup", d.handleEKSDescribeNodegroup}, - {"eks.ListNodegroups", d.handleEKSListNodegroups}, - {"eks.UpdateNodegroupConfig", d.handleEKSUpdateNodegroupConfig}, - {"eks.UpdateNodegroupVersion", d.handleEKSUpdateNodegroupVersion}, - {"eks.DeleteNodegroup", d.handleEKSDeleteNodegroup}, - {"eks.CreateAccessEntry", d.handleEKSCreateAccessEntry}, - {"eks.DescribeAccessEntry", d.handleEKSDescribeAccessEntry}, - {"eks.ListAccessEntries", d.handleEKSListAccessEntries}, - {"eks.UpdateAccessEntry", d.handleEKSUpdateAccessEntry}, - {"eks.DeleteAccessEntry", d.handleEKSDeleteAccessEntry}, - {"eks.AssociateAccessPolicy", d.handleEKSAssociateAccessPolicy}, - {"eks.DisassociateAccessPolicy", d.handleEKSDisassociateAccessPolicy}, - {"eks.ListAssociatedAccessPolicies", d.handleEKSListAssociatedAccessPolicies}, - {"eks.ListAccessPolicies", d.handleEKSListAccessPolicies}, - {"eks.ListAddons", d.handleEKSListAddons}, - {"eks.DescribeAddonVersions", d.handleEKSDescribeAddonVersions}, - {"eks.CreateAddon", d.handleEKSCreateAddon}, - {"eks.DeleteAddon", d.handleEKSDeleteAddon}, - {"eks.DescribeAddon", d.handleEKSDescribeAddon}, - {"eks.UpdateAddon", d.handleEKSUpdateAddon}, - {"eks.AssociateIdentityProviderConfig", d.handleEKSAssociateIdentityProviderConfig}, - {"eks.DescribeIdentityProviderConfig", d.handleEKSDescribeIdentityProviderConfig}, - {"eks.ListIdentityProviderConfigs", d.handleEKSListIdentityProviderConfigs}, - {"eks.DisassociateIdentityProviderConfig", d.handleEKSDisassociateIdentityProviderConfig}, - {"eks.TagResource", d.handleEKSTagResource}, - {"eks.UntagResource", d.handleEKSUntagResource}, - {"eks.ListTagsForResource", d.handleEKSListTagsForResource}, + {"eks.UpdateClusterConfig", d.handleEKSUpdateClusterConfig, notImpl}, + {"eks.UpdateClusterVersion", d.handleEKSUpdateClusterVersion, notImpl}, + {"eks.CreateNodegroup", d.handleEKSCreateNodegroup, notImpl}, + {"eks.DescribeNodegroup", d.handleEKSDescribeNodegroup, notImpl}, + {"eks.ListNodegroups", d.handleEKSListNodegroups, notImpl}, + {"eks.UpdateNodegroupConfig", d.handleEKSUpdateNodegroupConfig, notImpl}, + {"eks.UpdateNodegroupVersion", d.handleEKSUpdateNodegroupVersion, notImpl}, + {"eks.DeleteNodegroup", d.handleEKSDeleteNodegroup, notImpl}, + {"eks.CreateAccessEntry", d.handleEKSCreateAccessEntry, invalid}, + {"eks.DescribeAccessEntry", d.handleEKSDescribeAccessEntry, invalid}, + {"eks.ListAccessEntries", d.handleEKSListAccessEntries, invalid}, + {"eks.UpdateAccessEntry", d.handleEKSUpdateAccessEntry, invalid}, + {"eks.DeleteAccessEntry", d.handleEKSDeleteAccessEntry, invalid}, + {"eks.AssociateAccessPolicy", d.handleEKSAssociateAccessPolicy, invalid}, + {"eks.DisassociateAccessPolicy", d.handleEKSDisassociateAccessPolicy, invalid}, + {"eks.ListAssociatedAccessPolicies", d.handleEKSListAssociatedAccessPolicies, invalid}, + {"eks.ListAccessPolicies", d.handleEKSListAccessPolicies, ""}, + {"eks.ListAddons", d.handleEKSListAddons, notImpl}, + {"eks.DescribeAddonVersions", d.handleEKSDescribeAddonVersions, notImpl}, + {"eks.CreateAddon", d.handleEKSCreateAddon, notImpl}, + {"eks.DeleteAddon", d.handleEKSDeleteAddon, notImpl}, + {"eks.DescribeAddon", d.handleEKSDescribeAddon, notImpl}, + {"eks.UpdateAddon", d.handleEKSUpdateAddon, notImpl}, + {"eks.AssociateIdentityProviderConfig", d.handleEKSAssociateIdentityProviderConfig, notImpl}, + {"eks.DescribeIdentityProviderConfig", d.handleEKSDescribeIdentityProviderConfig, notImpl}, + {"eks.ListIdentityProviderConfigs", d.handleEKSListIdentityProviderConfigs, notImpl}, + {"eks.DisassociateIdentityProviderConfig", d.handleEKSDisassociateIdentityProviderConfig, notImpl}, + {"eks.TagResource", d.handleEKSTagResource, notImpl}, + {"eks.UntagResource", d.handleEKSUntagResource, notImpl}, + {"eks.ListTagsForResource", d.handleEKSListTagsForResource, notImpl}, } require.Equal(t, 30, len(cases), "expected exactly one handler per non-lifecycle AWS EKS action") @@ -97,7 +106,16 @@ func TestDaemonHandleEKS_AllHandlersDispatchToService(t *testing.T) { t.Cleanup(func() { _ = sub.Unsubscribe() }) payload := requestEKS(t, nc, c.subject) - assertNotImpl(t, payload) + if c.wantCode == "" { + // Success payload: dispatch is proven by a decodable non-error body. + var env struct { + Code *string `json:"Code"` + } + require.NoError(t, json.Unmarshal(payload, &env), "decode payload %q", payload) + assert.Nil(t, env.Code, "expected success, got error %q", payload) + return + } + assertErrorCode(t, payload, c.wantCode) }) } } diff --git a/spinifex/gateway/eks.go b/spinifex/gateway/eks.go index 69a17751a..ce6ab2d9f 100644 --- a/spinifex/gateway/eks.go +++ b/spinifex/gateway/eks.go @@ -5,6 +5,7 @@ import ( "io" "log/slog" "net/http" + "net/url" "regexp" "github.com/mulgadc/spinifex/spinifex/awserrors" @@ -30,10 +31,12 @@ type eksRoute struct { // eksRouteHandler invokes the per-action gateway function. It receives the // path-capture values in declaration order, the request body, the parent -// gateway config (NATS conn + ServiceURL), and the caller's accountID. It -// returns the response payload object (already a typed *eks.Output) or an -// error whose Error() is an awserrors code. -type eksRouteHandler func(gw *GatewayConfig, accountID string, params []string, body []byte) (any, error) +// gateway config (NATS conn + ServiceURL), the caller's accountID, and the +// caller's resolved IAM principal ARN (used by CreateCluster for the +// bootstrap-creator-admin AccessEntry; ignored by the rest). It returns the +// response payload object (already a typed *eks.Output) or an error whose +// Error() is an awserrors code. +type eksRouteHandler func(gw *GatewayConfig, accountID, callerARN string, params []string, body []byte) (any, error) // eksRoutes is the dispatch table. Order matters: more-specific paths must // come before less-specific ones with the same prefix (e.g. the @@ -42,127 +45,127 @@ type eksRouteHandler func(gw *GatewayConfig, accountID string, params []string, var eksRoutes = []eksRoute{ // Cluster {"POST", regexp.MustCompile(`^/clusters$`), "CreateCluster", - func(gw *GatewayConfig, acct string, p []string, b []byte) (any, error) { - return gateway_eks.CreateCluster(gw.NATSConn, acct, b) + func(gw *GatewayConfig, acct, callerARN string, p []string, b []byte) (any, error) { + return gateway_eks.CreateCluster(gw.NATSConn, acct, callerARN, b) }}, {"GET", regexp.MustCompile(`^/clusters$`), "ListClusters", - func(gw *GatewayConfig, acct string, p []string, b []byte) (any, error) { + func(gw *GatewayConfig, acct, callerARN string, p []string, b []byte) (any, error) { return gateway_eks.ListClusters(gw.NATSConn, acct) }}, {"POST", regexp.MustCompile(`^/clusters/([^/]+)/update-config$`), "UpdateClusterConfig", - func(gw *GatewayConfig, acct string, p []string, b []byte) (any, error) { + func(gw *GatewayConfig, acct, callerARN string, p []string, b []byte) (any, error) { return gateway_eks.UpdateClusterConfig(gw.NATSConn, acct, p[0], b) }}, {"POST", regexp.MustCompile(`^/clusters/([^/]+)/update-version$`), "UpdateClusterVersion", - func(gw *GatewayConfig, acct string, p []string, b []byte) (any, error) { + func(gw *GatewayConfig, acct, callerARN string, p []string, b []byte) (any, error) { return gateway_eks.UpdateClusterVersion(gw.NATSConn, acct, p[0], b) }}, // Nodegroup {"POST", regexp.MustCompile(`^/clusters/([^/]+)/node-groups$`), "CreateNodegroup", - func(gw *GatewayConfig, acct string, p []string, b []byte) (any, error) { + func(gw *GatewayConfig, acct, callerARN string, p []string, b []byte) (any, error) { return gateway_eks.CreateNodegroup(gw.NATSConn, acct, p[0], b) }}, {"GET", regexp.MustCompile(`^/clusters/([^/]+)/node-groups$`), "ListNodegroups", - func(gw *GatewayConfig, acct string, p []string, b []byte) (any, error) { + func(gw *GatewayConfig, acct, callerARN string, p []string, b []byte) (any, error) { return gateway_eks.ListNodegroups(gw.NATSConn, acct, p[0]) }}, {"POST", regexp.MustCompile(`^/clusters/([^/]+)/node-groups/([^/]+)/update-config$`), "UpdateNodegroupConfig", - func(gw *GatewayConfig, acct string, p []string, b []byte) (any, error) { + func(gw *GatewayConfig, acct, callerARN string, p []string, b []byte) (any, error) { return gateway_eks.UpdateNodegroupConfig(gw.NATSConn, acct, p[0], p[1], b) }}, {"POST", regexp.MustCompile(`^/clusters/([^/]+)/node-groups/([^/]+)/update-version$`), "UpdateNodegroupVersion", - func(gw *GatewayConfig, acct string, p []string, b []byte) (any, error) { + func(gw *GatewayConfig, acct, callerARN string, p []string, b []byte) (any, error) { return gateway_eks.UpdateNodegroupVersion(gw.NATSConn, acct, p[0], p[1], b) }}, {"GET", regexp.MustCompile(`^/clusters/([^/]+)/node-groups/([^/]+)$`), "DescribeNodegroup", - func(gw *GatewayConfig, acct string, p []string, b []byte) (any, error) { + func(gw *GatewayConfig, acct, callerARN string, p []string, b []byte) (any, error) { return gateway_eks.DescribeNodegroup(gw.NATSConn, acct, p[0], p[1]) }}, {"DELETE", regexp.MustCompile(`^/clusters/([^/]+)/node-groups/([^/]+)$`), "DeleteNodegroup", - func(gw *GatewayConfig, acct string, p []string, b []byte) (any, error) { + func(gw *GatewayConfig, acct, callerARN string, p []string, b []byte) (any, error) { return gateway_eks.DeleteNodegroup(gw.NATSConn, acct, p[0], p[1]) }}, // AccessEntry / AccessPolicy {"POST", regexp.MustCompile(`^/clusters/([^/]+)/access-entries$`), "CreateAccessEntry", - func(gw *GatewayConfig, acct string, p []string, b []byte) (any, error) { + func(gw *GatewayConfig, acct, callerARN string, p []string, b []byte) (any, error) { return gateway_eks.CreateAccessEntry(gw.NATSConn, acct, p[0], b) }}, {"GET", regexp.MustCompile(`^/clusters/([^/]+)/access-entries$`), "ListAccessEntries", - func(gw *GatewayConfig, acct string, p []string, b []byte) (any, error) { + func(gw *GatewayConfig, acct, callerARN string, p []string, b []byte) (any, error) { return gateway_eks.ListAccessEntries(gw.NATSConn, acct, p[0]) }}, {"POST", regexp.MustCompile(`^/clusters/([^/]+)/access-entries/([^/]+)/access-policies$`), "AssociateAccessPolicy", - func(gw *GatewayConfig, acct string, p []string, b []byte) (any, error) { + func(gw *GatewayConfig, acct, callerARN string, p []string, b []byte) (any, error) { return gateway_eks.AssociateAccessPolicy(gw.NATSConn, acct, p[0], p[1], b) }}, {"DELETE", regexp.MustCompile(`^/clusters/([^/]+)/access-entries/([^/]+)/access-policies$`), "DisassociateAccessPolicy", - func(gw *GatewayConfig, acct string, p []string, b []byte) (any, error) { + func(gw *GatewayConfig, acct, callerARN string, p []string, b []byte) (any, error) { return gateway_eks.DisassociateAccessPolicy(gw.NATSConn, acct, p[0], p[1], b) }}, {"GET", regexp.MustCompile(`^/clusters/([^/]+)/access-entries/([^/]+)/access-policies$`), "ListAssociatedAccessPolicies", - func(gw *GatewayConfig, acct string, p []string, b []byte) (any, error) { + func(gw *GatewayConfig, acct, callerARN string, p []string, b []byte) (any, error) { return gateway_eks.ListAssociatedAccessPolicies(gw.NATSConn, acct, p[0], p[1]) }}, {"GET", regexp.MustCompile(`^/clusters/([^/]+)/access-entries/([^/]+)$`), "DescribeAccessEntry", - func(gw *GatewayConfig, acct string, p []string, b []byte) (any, error) { + func(gw *GatewayConfig, acct, callerARN string, p []string, b []byte) (any, error) { return gateway_eks.DescribeAccessEntry(gw.NATSConn, acct, p[0], p[1]) }}, {"POST", regexp.MustCompile(`^/clusters/([^/]+)/access-entries/([^/]+)$`), "UpdateAccessEntry", - func(gw *GatewayConfig, acct string, p []string, b []byte) (any, error) { + func(gw *GatewayConfig, acct, callerARN string, p []string, b []byte) (any, error) { return gateway_eks.UpdateAccessEntry(gw.NATSConn, acct, p[0], p[1], b) }}, {"DELETE", regexp.MustCompile(`^/clusters/([^/]+)/access-entries/([^/]+)$`), "DeleteAccessEntry", - func(gw *GatewayConfig, acct string, p []string, b []byte) (any, error) { + func(gw *GatewayConfig, acct, callerARN string, p []string, b []byte) (any, error) { return gateway_eks.DeleteAccessEntry(gw.NATSConn, acct, p[0], p[1]) }}, {"GET", regexp.MustCompile(`^/access-policies$`), "ListAccessPolicies", - func(gw *GatewayConfig, acct string, p []string, b []byte) (any, error) { + func(gw *GatewayConfig, acct, callerARN string, p []string, b []byte) (any, error) { return gateway_eks.ListAccessPolicies(gw.NATSConn, acct) }}, // Addons {"GET", regexp.MustCompile(`^/addons/supported-versions$`), "DescribeAddonVersions", - func(gw *GatewayConfig, acct string, p []string, b []byte) (any, error) { + func(gw *GatewayConfig, acct, callerARN string, p []string, b []byte) (any, error) { return gateway_eks.DescribeAddonVersions(gw.NATSConn, acct) }}, {"GET", regexp.MustCompile(`^/addons$`), "ListAddons", - func(gw *GatewayConfig, acct string, p []string, b []byte) (any, error) { + func(gw *GatewayConfig, acct, callerARN string, p []string, b []byte) (any, error) { return gateway_eks.ListAddons(gw.NATSConn, acct) }}, {"POST", regexp.MustCompile(`^/clusters/([^/]+)/addons$`), "CreateAddon", - func(gw *GatewayConfig, acct string, p []string, b []byte) (any, error) { + func(gw *GatewayConfig, acct, callerARN string, p []string, b []byte) (any, error) { return gateway_eks.CreateAddon(gw.NATSConn, acct, p[0], b) }}, {"POST", regexp.MustCompile(`^/clusters/([^/]+)/addons/([^/]+)/update$`), "UpdateAddon", - func(gw *GatewayConfig, acct string, p []string, b []byte) (any, error) { + func(gw *GatewayConfig, acct, callerARN string, p []string, b []byte) (any, error) { return gateway_eks.UpdateAddon(gw.NATSConn, acct, p[0], p[1], b) }}, {"GET", regexp.MustCompile(`^/clusters/([^/]+)/addons/([^/]+)$`), "DescribeAddon", - func(gw *GatewayConfig, acct string, p []string, b []byte) (any, error) { + func(gw *GatewayConfig, acct, callerARN string, p []string, b []byte) (any, error) { return gateway_eks.DescribeAddon(gw.NATSConn, acct, p[0], p[1]) }}, {"DELETE", regexp.MustCompile(`^/clusters/([^/]+)/addons/([^/]+)$`), "DeleteAddon", - func(gw *GatewayConfig, acct string, p []string, b []byte) (any, error) { + func(gw *GatewayConfig, acct, callerARN string, p []string, b []byte) (any, error) { return gateway_eks.DeleteAddon(gw.NATSConn, acct, p[0], p[1]) }}, // OIDC identity-provider configs {"POST", regexp.MustCompile(`^/clusters/([^/]+)/identity-provider-configs/associate$`), "AssociateIdentityProviderConfig", - func(gw *GatewayConfig, acct string, p []string, b []byte) (any, error) { + func(gw *GatewayConfig, acct, callerARN string, p []string, b []byte) (any, error) { return gateway_eks.AssociateIdentityProviderConfig(gw.NATSConn, acct, p[0], b) }}, {"POST", regexp.MustCompile(`^/clusters/([^/]+)/identity-provider-configs/describe$`), "DescribeIdentityProviderConfig", - func(gw *GatewayConfig, acct string, p []string, b []byte) (any, error) { + func(gw *GatewayConfig, acct, callerARN string, p []string, b []byte) (any, error) { return gateway_eks.DescribeIdentityProviderConfig(gw.NATSConn, acct, p[0], b) }}, {"POST", regexp.MustCompile(`^/clusters/([^/]+)/identity-provider-configs/disassociate$`), "DisassociateIdentityProviderConfig", - func(gw *GatewayConfig, acct string, p []string, b []byte) (any, error) { + func(gw *GatewayConfig, acct, callerARN string, p []string, b []byte) (any, error) { return gateway_eks.DisassociateIdentityProviderConfig(gw.NATSConn, acct, p[0], b) }}, {"GET", regexp.MustCompile(`^/clusters/([^/]+)/identity-provider-configs$`), "ListIdentityProviderConfigs", - func(gw *GatewayConfig, acct string, p []string, b []byte) (any, error) { + func(gw *GatewayConfig, acct, callerARN string, p []string, b []byte) (any, error) { return gateway_eks.ListIdentityProviderConfigs(gw.NATSConn, acct, p[0]) }}, @@ -170,25 +173,25 @@ var eksRoutes = []eksRoute{ // /clusters/{name}/... routes so the regexp matcher prefers the deeper // matches first when iterating in declaration order. {"GET", regexp.MustCompile(`^/clusters/([^/]+)$`), "DescribeCluster", - func(gw *GatewayConfig, acct string, p []string, b []byte) (any, error) { + func(gw *GatewayConfig, acct, callerARN string, p []string, b []byte) (any, error) { return gateway_eks.DescribeCluster(gw.NATSConn, acct, p[0]) }}, {"DELETE", regexp.MustCompile(`^/clusters/([^/]+)$`), "DeleteCluster", - func(gw *GatewayConfig, acct string, p []string, b []byte) (any, error) { + func(gw *GatewayConfig, acct, callerARN string, p []string, b []byte) (any, error) { return gateway_eks.DeleteCluster(gw.NATSConn, acct, p[0]) }}, // Tags {"POST", regexp.MustCompile(`^/tags/(.+)$`), "TagResource", - func(gw *GatewayConfig, acct string, p []string, b []byte) (any, error) { + func(gw *GatewayConfig, acct, callerARN string, p []string, b []byte) (any, error) { return gateway_eks.TagResource(gw.NATSConn, acct, p[0], b) }}, {"DELETE", regexp.MustCompile(`^/tags/(.+)$`), "UntagResource", - func(gw *GatewayConfig, acct string, p []string, b []byte) (any, error) { + func(gw *GatewayConfig, acct, callerARN string, p []string, b []byte) (any, error) { return gateway_eks.UntagResource(gw.NATSConn, acct, p[0], b) }}, {"GET", regexp.MustCompile(`^/tags/(.+)$`), "ListTagsForResource", - func(gw *GatewayConfig, acct string, p []string, b []byte) (any, error) { + func(gw *GatewayConfig, acct, callerARN string, p []string, b []byte) (any, error) { return gateway_eks.ListTagsForResource(gw.NATSConn, acct, p[0]) }}, } @@ -196,6 +199,12 @@ var eksRoutes = []eksRoute{ // lookupEKSAction walks eksRoutes in declaration order, returning the matched // route's action name + path params, or ("", nil, false) when nothing matches. // Used by EKS_Request and unit tests to verify routing. +// +// path must be the *escaped* request path (r.URL.EscapedPath()): IAM principal +// ARNs are passed as a path segment and the CLI percent-encodes the embedded +// `/` (`user%2Fadmin`). Matching the decoded path would split that ARN across +// the `([^/]+)` capture and no route would match. Captured params are +// PathUnescape'd here so handlers receive the real value. func lookupEKSAction(method, path string) (string, []string, eksRouteHandler, bool) { for _, route := range eksRoutes { if route.method != method { @@ -207,7 +216,15 @@ func lookupEKSAction(method, path string) (string, []string, eksRouteHandler, bo } var params []string if len(m) > 1 { - params = m[1:] + params = make([]string, 0, len(m)-1) + for _, raw := range m[1:] { + decoded, err := url.PathUnescape(raw) + if err != nil { + slog.Debug("EKS: bad percent-encoding in path param", "param", raw, "err", err) + decoded = raw + } + params = append(params, decoded) + } } return route.action, params, route.handler, true } @@ -220,7 +237,7 @@ func lookupEKSAction(method, path string) (string, []string, eksRouteHandler, bo // are returned as plain awserrors codes; the caller (gateway.Request) routes // them through the shared ErrorHandler, which now emits EKS REST-JSON. func (gw *GatewayConfig) EKS_Request(w http.ResponseWriter, r *http.Request) error { - action, params, handler, ok := lookupEKSAction(r.Method, r.URL.Path) + action, params, handler, ok := lookupEKSAction(r.Method, r.URL.EscapedPath()) if !ok { slog.Debug("EKS: no route for request", "method", r.Method, "path", r.URL.Path) return errors.New(awserrors.ErrorInvalidAction) @@ -246,7 +263,11 @@ func (gw *GatewayConfig) EKS_Request(w http.ResponseWriter, r *http.Request) err return errors.New(awserrors.ErrorInvalidParameterValue) } - output, err := handler(gw, accountID, params, body) + // Resolve the caller's IAM principal ARN best-effort; only CreateCluster + // consumes it (bootstrap-creator-admin) and it degrades gracefully to "". + callerARN := eksCallerPrincipalARN(r) + + output, err := handler(gw, accountID, callerARN, params, body) if err != nil { return err } @@ -254,3 +275,21 @@ func (gw *GatewayConfig) EKS_Request(w http.ResponseWriter, r *http.Request) err gateway_eks.WriteJSONResponse(w, output) return nil } + +// eksCallerPrincipalARN resolves the caller's IAM principal ARN from the SigV4 +// auth context, best-effort. Returns "" when the context is incomplete or the +// ARN can't be composed — CreateCluster then skips the creator-admin entry +// rather than failing the request. +func eksCallerPrincipalARN(r *http.Request) string { + ctx := r.Context() + accountID, _ := ctx.Value(ctxAccountID).(string) + identity, _ := ctx.Value(ctxIdentity).(string) + principalType, _ := ctx.Value(ctxPrincipalType).(string) + assumedRoleARN, _ := ctx.Value(ctxAssumedRoleARN).(string) + arn, err := buildCallerARN(accountID, identity, principalType, assumedRoleARN) + if err != nil { + slog.Debug("EKS_Request: could not resolve caller principal ARN", "err", err) + return "" + } + return arn +} diff --git a/spinifex/gateway/eks/cluster.go b/spinifex/gateway/eks/cluster.go index c4168a573..22ed19dcb 100644 --- a/spinifex/gateway/eks/cluster.go +++ b/spinifex/gateway/eks/cluster.go @@ -9,13 +9,15 @@ import ( "github.com/nats-io/nats.go" ) -// CreateCluster — POST /clusters -func CreateCluster(natsConn *nats.Conn, accountID string, body []byte) (*eks.CreateClusterOutput, error) { +// CreateCluster — POST /clusters. callerPrincipalARN is the resolved caller IAM +// principal ARN, plumbed through so the control plane can mint the +// bootstrap-creator-admin AccessEntry for the caller. +func CreateCluster(natsConn *nats.Conn, accountID, callerPrincipalARN string, body []byte) (*eks.CreateClusterOutput, error) { input := new(eks.CreateClusterInput) if err := unmarshalIfBody(body, input); err != nil { return nil, err } - return handlers_eks.NewNATSEKSService(natsConn).CreateCluster(input, accountID) + return handlers_eks.NewNATSEKSService(natsConn).CreateCluster(input, accountID, callerPrincipalARN) } // DescribeCluster — GET /clusters/{name} diff --git a/spinifex/gateway/eks/wrappers_test.go b/spinifex/gateway/eks/wrappers_test.go index a97405dc0..a012fd0b7 100644 --- a/spinifex/gateway/eks/wrappers_test.go +++ b/spinifex/gateway/eks/wrappers_test.go @@ -30,11 +30,11 @@ func TestGatewayWrappers_Cluster(t *testing.T) { _, nc := testutil.StartTestNATS(t) stubEKSResponder(t, nc) - out1, err := CreateCluster(nc, acct, []byte(`{"name":"alpha"}`)) + out1, err := CreateCluster(nc, acct, "arn:aws:iam::111122223333:role/dev", []byte(`{"name":"alpha"}`)) require.NoError(t, err) assert.NotNil(t, out1) - out2, err := CreateCluster(nc, acct, nil) + out2, err := CreateCluster(nc, acct, "", nil) require.NoError(t, err) assert.NotNil(t, out2) @@ -71,7 +71,7 @@ func TestGatewayWrappers_Cluster_BadJSON(t *testing.T) { _, nc := testutil.StartTestNATS(t) stubEKSResponder(t, nc) - _, err := CreateCluster(nc, acct, []byte(`{not-json`)) + _, err := CreateCluster(nc, acct, "", []byte(`{not-json`)) require.Error(t, err) _, err = UpdateClusterConfig(nc, acct, "alpha", []byte(`{not-json`)) require.Error(t, err) diff --git a/spinifex/gateway/eks_test.go b/spinifex/gateway/eks_test.go index 3398f218b..f581ae9d6 100644 --- a/spinifex/gateway/eks_test.go +++ b/spinifex/gateway/eks_test.go @@ -66,6 +66,37 @@ func TestLookupEKSAction_ResolvesKnownRoutes(t *testing.T) { } } +// IAM principal ARNs contain a `/` (`user/admin`) and a `:` which the AWS CLI +// percent-encodes in the request path. lookupEKSAction is fed r.URL.EscapedPath() +// so `%2F` stays a single `([^/]+)` segment; the captured param is then +// PathUnescape'd back to the real ARN before reaching the handler. +func TestLookupEKSAction_EncodedPrincipalARN(t *testing.T) { + const ( + arn = "arn:aws:iam::000000000001:user/admin" + escaped = "arn%3Aaws%3Aiam%3A%3A000000000001%3Auser%2Fadmin" + ) + cases := []struct { + method, path string + wantAction string + }{ + {"GET", "/clusters/alpha/access-entries/" + escaped, "DescribeAccessEntry"}, + {"POST", "/clusters/alpha/access-entries/" + escaped, "UpdateAccessEntry"}, + {"DELETE", "/clusters/alpha/access-entries/" + escaped, "DeleteAccessEntry"}, + {"POST", "/clusters/alpha/access-entries/" + escaped + "/access-policies", "AssociateAccessPolicy"}, + {"DELETE", "/clusters/alpha/access-entries/" + escaped + "/access-policies", "DisassociateAccessPolicy"}, + {"GET", "/clusters/alpha/access-entries/" + escaped + "/access-policies", "ListAssociatedAccessPolicies"}, + } + for _, tc := range cases { + t.Run(tc.method+"_"+tc.wantAction, func(t *testing.T) { + action, params, handler, ok := lookupEKSAction(tc.method, tc.path) + require.True(t, ok, "encoded ARN path should match: %s %s", tc.method, tc.path) + require.NotNil(t, handler) + assert.Equal(t, tc.wantAction, action) + assert.Equal(t, []string{"alpha", arn}, params) + }) + } +} + func TestLookupEKSAction_CoversAll34Actions(t *testing.T) { expected := map[string]bool{ "CreateCluster": false, diff --git a/spinifex/handlers/eks/access_entry.go b/spinifex/handlers/eks/access_entry.go new file mode 100644 index 000000000..736014afc --- /dev/null +++ b/spinifex/handlers/eks/access_entry.go @@ -0,0 +1,263 @@ +package handlers_eks + +import ( + "encoding/json" + "errors" + "fmt" + "strings" + "time" + + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/service/eks" + "github.com/nats-io/nats.go" +) + +// Access-entry types (Q9). v1 supports STANDARD principals only; the node +// types (EC2_LINUX/EC2_WINDOWS/FARGATE_LINUX) are accepted by AWS but the +// nodegroup join path (Sprint 6d) handles node identity, so they are rejected +// here until that lands. +const ( + AccessEntryTypeStandard = "STANDARD" + + accessScopeCluster = "cluster" + accessScopeNamespace = "namespace" +) + +// Supported AWS-managed access policies (Q9). The map value is the K8s +// ClusterRole the policy maps onto; the token webhook + RBAC bindings consume +// it. Any policy ARN outside this set is rejected with InvalidParameterValue. +var supportedAccessPolicies = map[string]string{ + "arn:aws:eks::aws:cluster-access-policy/AmazonEKSClusterAdminPolicy": "cluster-admin", + "arn:aws:eks::aws:cluster-access-policy/AmazonEKSAdminPolicy": "admin", + "arn:aws:eks::aws:cluster-access-policy/AmazonEKSEditPolicy": "edit", + "arn:aws:eks::aws:cluster-access-policy/AmazonEKSViewPolicy": "view", +} + +// ErrAccessEntryNotFound is returned by GetAccessEntryRecord / the CAS helper +// when no entry exists for the principal. Callers translate to the AWS shape +// (ResourceNotFoundException) at the service boundary. +var ErrAccessEntryNotFound = errors.New("eks: access entry not found") + +// AccessEntryARN composes the access-entry ARN. AWS uses a per-entry UUID; this +// build keys the discriminator off the principal-ARN hash instead so the ARN is +// deterministic (no RNG) and one-per-principal — clients address access entries +// by principalArn, not by this ARN, so the divergence is informational only. +func AccessEntryARN(region, accountID, cluster, principalARN string) string { + return fmt.Sprintf("arn:aws:eks:%s:%s:access-entry/%s/%s", + region, accountID, cluster, PrincipalARNHash(principalARN)) +} + +// PutAccessEntryRecord writes the entry unconditionally. +func PutAccessEntryRecord(kv nats.KeyValue, rec *AccessEntryRecord) error { + if rec == nil { + return errors.New("eks: PutAccessEntryRecord nil record") + } + if rec.ClusterName == "" || rec.PrincipalARN == "" { + return errors.New("eks: PutAccessEntryRecord missing cluster or principal ARN") + } + data, err := json.Marshal(rec) + if err != nil { + return fmt.Errorf("marshal access entry %s: %w", rec.PrincipalARN, err) + } + key := AccessEntryKey(rec.ClusterName, rec.PrincipalARN) + if _, err := kv.Put(key, data); err != nil { + return fmt.Errorf("kv put %s: %w", key, err) + } + return nil +} + +// GetAccessEntryRecord reads one entry. Returns ErrAccessEntryNotFound if absent. +func GetAccessEntryRecord(kv nats.KeyValue, cluster, principalARN string) (*AccessEntryRecord, error) { + if cluster == "" || principalARN == "" { + return nil, errors.New("eks: GetAccessEntryRecord empty cluster or principal ARN") + } + entry, err := kv.Get(AccessEntryKey(cluster, principalARN)) + if err != nil { + if errors.Is(err, nats.ErrKeyNotFound) { + return nil, ErrAccessEntryNotFound + } + return nil, fmt.Errorf("kv get access entry: %w", err) + } + var rec AccessEntryRecord + if err := json.Unmarshal(entry.Value(), &rec); err != nil { + return nil, fmt.Errorf("unmarshal access entry %s: %w", principalARN, err) + } + return &rec, nil +} + +// ListAccessEntryRecords returns every access entry under a cluster, sorted by +// principal ARN for stable output. +func ListAccessEntryRecords(kv nats.KeyValue, cluster string) ([]*AccessEntryRecord, error) { + if cluster == "" { + return nil, errors.New("eks: ListAccessEntryRecords empty cluster") + } + keys, err := kv.Keys() + if err != nil { + if errors.Is(err, nats.ErrNoKeysFound) { + return nil, nil + } + return nil, fmt.Errorf("kv keys: %w", err) + } + prefix := AccessEntriesPrefix(cluster) + out := make([]*AccessEntryRecord, 0) + for _, k := range keys { + if !strings.HasPrefix(k, prefix) { + continue + } + entry, err := kv.Get(k) + if err != nil { + if errors.Is(err, nats.ErrKeyNotFound) { + continue + } + return nil, fmt.Errorf("kv get %s: %w", k, err) + } + var rec AccessEntryRecord + if err := json.Unmarshal(entry.Value(), &rec); err != nil { + return nil, fmt.Errorf("unmarshal access entry %s: %w", k, err) + } + out = append(out, &rec) + } + sortAccessEntries(out) + return out, nil +} + +// DeleteAccessEntryRecord removes one entry. Returns ErrAccessEntryNotFound if +// it did not exist so DeleteAccessEntry can surface ResourceNotFoundException. +func DeleteAccessEntryRecord(kv nats.KeyValue, cluster, principalARN string) error { + key := AccessEntryKey(cluster, principalARN) + if _, err := kv.Get(key); err != nil { + if errors.Is(err, nats.ErrKeyNotFound) { + return ErrAccessEntryNotFound + } + return fmt.Errorf("kv get %s: %w", key, err) + } + if err := kv.Delete(key); err != nil { + return fmt.Errorf("kv delete %s: %w", key, err) + } + return nil +} + +// casUpdateAccessEntry does a revision-checked read-modify-write. mutate returns +// true if it changed a field. Returns ErrAccessEntryNotFound if absent. +func casUpdateAccessEntry(kv nats.KeyValue, cluster, principalARN string, mutate func(*AccessEntryRecord) bool) (*AccessEntryRecord, error) { + key := AccessEntryKey(cluster, principalARN) + for range maxClusterStateCASRetries { + entry, err := kv.Get(key) + if err != nil { + if errors.Is(err, nats.ErrKeyNotFound) { + return nil, ErrAccessEntryNotFound + } + return nil, fmt.Errorf("kv get %s: %w", key, err) + } + var rec AccessEntryRecord + if err := json.Unmarshal(entry.Value(), &rec); err != nil { + return nil, fmt.Errorf("unmarshal access entry %s: %w", principalARN, err) + } + if !mutate(&rec) { + return &rec, nil + } + data, err := json.Marshal(&rec) + if err != nil { + return nil, fmt.Errorf("marshal access entry %s: %w", principalARN, err) + } + _, err = kv.Update(key, data, entry.Revision()) + if err == nil { + return &rec, nil + } + if errors.Is(err, nats.ErrKeyExists) { + continue + } + return nil, fmt.Errorf("kv update %s: %w", key, err) + } + return nil, fmt.Errorf("eks: casUpdateAccessEntry %s exhausted CAS retries", principalARN) +} + +func sortAccessEntries(recs []*AccessEntryRecord) { + for i := 1; i < len(recs); i++ { + for j := i; j > 0 && recs[j-1].PrincipalARN > recs[j].PrincipalARN; j-- { + recs[j-1], recs[j] = recs[j], recs[j-1] + } + } +} + +// validateAccessScope checks an AccessScope per AWS rules: type ∈ +// {cluster, namespace}; namespace scope requires at least one namespace and +// cluster scope must not carry any. +func validateAccessScope(scope *eks.AccessScope) (AccessScope, error) { + if scope == nil || aws.StringValue(scope.Type) == "" { + return AccessScope{}, errors.New("eks: accessScope.type is required") + } + t := strings.ToLower(aws.StringValue(scope.Type)) + ns := aws.StringValueSlice(scope.Namespaces) + switch t { + case accessScopeCluster: + if len(ns) > 0 { + return AccessScope{}, errors.New("eks: cluster-scoped policy must not list namespaces") + } + return AccessScope{Type: accessScopeCluster}, nil + case accessScopeNamespace: + if len(ns) == 0 { + return AccessScope{}, errors.New("eks: namespace-scoped policy requires at least one namespace") + } + return AccessScope{Type: accessScopeNamespace, Namespaces: ns}, nil + default: + return AccessScope{}, fmt.Errorf("eks: unsupported accessScope.type %q", t) + } +} + +// accessEntryRecordToAWS converts the persisted record to the SDK shape. +func accessEntryRecordToAWS(rec *AccessEntryRecord) *eks.AccessEntry { + out := &eks.AccessEntry{ + AccessEntryArn: aws.String(rec.ARN), + ClusterName: aws.String(rec.ClusterName), + PrincipalArn: aws.String(rec.PrincipalARN), + Username: aws.String(rec.KubernetesUsername), + KubernetesGroups: aws.StringSlice(rec.KubernetesGroups), + Type: aws.String(rec.Type), + CreatedAt: aws.Time(rec.CreatedAt), + ModifiedAt: aws.Time(rec.ModifiedAt), + } + if len(rec.Tags) > 0 { + out.Tags = aws.StringMap(rec.Tags) + } + return out +} + +// associatedPolicyToAWS converts a persisted associated policy to the SDK shape. +func associatedPolicyToAWS(p AssociatedAccessPolicy) *eks.AssociatedAccessPolicy { + scope := &eks.AccessScope{Type: aws.String(p.AccessScope.Type)} + if len(p.AccessScope.Namespaces) > 0 { + scope.Namespaces = aws.StringSlice(p.AccessScope.Namespaces) + } + return &eks.AssociatedAccessPolicy{ + PolicyArn: aws.String(p.PolicyARN), + AccessScope: scope, + AssociatedAt: aws.Time(p.AssociatedAt), + ModifiedAt: aws.Time(p.ModifiedAt), + } +} + +// defaultKubernetesUsername mirrors AWS: when the caller omits username, EKS +// derives it from the principal ARN. +func defaultKubernetesUsername(principalARN string) string { + return principalARN +} + +// newAccessEntryRecord builds a record, defaulting the username to the +// principal ARN when unset (AWS behaviour). now stamps Created/Modified. +func newAccessEntryRecord(region, accountID, cluster, principalARN, username string, groups []string, entryType string, tags map[string]string, now time.Time) *AccessEntryRecord { + if username == "" { + username = defaultKubernetesUsername(principalARN) + } + return &AccessEntryRecord{ + ARN: AccessEntryARN(region, accountID, cluster, principalARN), + ClusterName: cluster, + PrincipalARN: principalARN, + KubernetesUsername: username, + KubernetesGroups: groups, + Type: entryType, + Tags: tags, + CreatedAt: now, + ModifiedAt: now, + } +} diff --git a/spinifex/handlers/eks/access_entry_test.go b/spinifex/handlers/eks/access_entry_test.go new file mode 100644 index 000000000..882609c28 --- /dev/null +++ b/spinifex/handlers/eks/access_entry_test.go @@ -0,0 +1,95 @@ +package handlers_eks + +import ( + "testing" + "time" + + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/service/eks" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestValidateAccessScope(t *testing.T) { + cases := []struct { + name string + scope *eks.AccessScope + want AccessScope + wantErr bool + }{ + {"nil scope", nil, AccessScope{}, true}, + {"empty type", &eks.AccessScope{}, AccessScope{}, true}, + {"cluster ok", &eks.AccessScope{Type: aws.String("cluster")}, AccessScope{Type: "cluster"}, false}, + {"cluster mixed case", &eks.AccessScope{Type: aws.String("Cluster")}, AccessScope{Type: "cluster"}, false}, + {"cluster with namespaces", &eks.AccessScope{Type: aws.String("cluster"), Namespaces: aws.StringSlice([]string{"kube-system"})}, AccessScope{}, true}, + {"namespace ok", &eks.AccessScope{Type: aws.String("namespace"), Namespaces: aws.StringSlice([]string{"team-a", "team-b"})}, AccessScope{Type: "namespace", Namespaces: []string{"team-a", "team-b"}}, false}, + {"namespace without namespaces", &eks.AccessScope{Type: aws.String("namespace")}, AccessScope{}, true}, + {"unsupported type", &eks.AccessScope{Type: aws.String("galaxy")}, AccessScope{}, true}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + got, err := validateAccessScope(tc.scope) + if tc.wantErr { + require.Error(t, err) + return + } + require.NoError(t, err) + assert.Equal(t, tc.want, got) + }) + } +} + +func TestAccessEntryRecordToAWS_IncludesTags(t *testing.T) { + now := time.Unix(1700000000, 0).UTC() + rec := &AccessEntryRecord{ + ARN: "arn:aws:eks:ap-southeast-2:000000000001:access-entry/dev1/abc", + ClusterName: "dev1", + PrincipalARN: "arn:aws:iam::000000000001:user/admin", + KubernetesUsername: "arn:aws:iam::000000000001:user/admin", + KubernetesGroups: []string{"system:masters"}, + Type: AccessEntryTypeStandard, + Tags: map[string]string{"team": "platform"}, + CreatedAt: now, + ModifiedAt: now, + } + out := accessEntryRecordToAWS(rec) + assert.Equal(t, "dev1", aws.StringValue(out.ClusterName)) + assert.Equal(t, rec.PrincipalARN, aws.StringValue(out.PrincipalArn)) + assert.Equal(t, []string{"system:masters"}, aws.StringValueSlice(out.KubernetesGroups)) + require.NotNil(t, out.Tags) + assert.Equal(t, "platform", aws.StringValue(out.Tags["team"])) +} + +func TestAssociatedPolicyToAWS_ScopeNamespaces(t *testing.T) { + now := time.Unix(1700000000, 0).UTC() + p := AssociatedAccessPolicy{ + PolicyARN: "arn:aws:eks::aws:cluster-access-policy/AmazonEKSViewPolicy", + AccessScope: AccessScope{Type: "namespace", Namespaces: []string{"team-a"}}, + AssociatedAt: now, + ModifiedAt: now, + } + out := associatedPolicyToAWS(p) + assert.Equal(t, p.PolicyARN, aws.StringValue(out.PolicyArn)) + assert.Equal(t, "namespace", aws.StringValue(out.AccessScope.Type)) + assert.Equal(t, []string{"team-a"}, aws.StringValueSlice(out.AccessScope.Namespaces)) + + cluster := associatedPolicyToAWS(AssociatedAccessPolicy{ + PolicyARN: "arn:aws:eks::aws:cluster-access-policy/AmazonEKSClusterAdminPolicy", + AccessScope: AccessScope{Type: "cluster"}, + }) + assert.Equal(t, "cluster", aws.StringValue(cluster.AccessScope.Type)) + assert.Empty(t, cluster.AccessScope.Namespaces) +} + +// The record-store guard clauses reject malformed input before touching the KV, +// so a nil handle is enough to exercise them. +func TestAccessEntryRecordGuards(t *testing.T) { + require.Error(t, PutAccessEntryRecord(nil, nil)) + require.Error(t, PutAccessEntryRecord(nil, &AccessEntryRecord{PrincipalARN: "arn:aws:iam::000000000001:user/admin"})) + + _, err := GetAccessEntryRecord(nil, "", "arn:aws:iam::000000000001:user/admin") + require.Error(t, err) + + _, err = ListAccessEntryRecords(nil, "") + require.Error(t, err) +} diff --git a/spinifex/handlers/eks/create_cluster_test.go b/spinifex/handlers/eks/create_cluster_test.go index 084a2ec50..f283bb465 100644 --- a/spinifex/handlers/eks/create_cluster_test.go +++ b/spinifex/handlers/eks/create_cluster_test.go @@ -34,7 +34,7 @@ func TestCreateCluster_NLBArnPersistedBeforeLaterFailure(t *testing.T) { // NLB-arn persist checkpoint, but before the final PutClusterMeta. f.inst.launchErr = errors.New("no capacity") - _, err := f.svc.CreateCluster(createInput("alpha"), testAccountID) + _, err := f.svc.CreateCluster(createInput("alpha"), testAccountID, "") require.Error(t, err) meta, getErr := GetClusterMeta(f.kv, "alpha") @@ -50,7 +50,7 @@ func TestCreateCluster_FailedCreateThenDeleteReclaimsNLB(t *testing.T) { f := newEKSServiceFixture(t) f.inst.launchErr = errors.New("no capacity") - _, err := f.svc.CreateCluster(createInput("alpha"), testAccountID) + _, err := f.svc.CreateCluster(createInput("alpha"), testAccountID, "") require.Error(t, err) // The NLB the partial create provisioned exists in the fake. @@ -73,7 +73,7 @@ func TestCreateCluster_ExistingClusterReturnsResourceInUse(t *testing.T) { meta.Status = ClusterStatusActive require.NoError(t, PutClusterMeta(f.kv, meta)) - _, err := f.svc.CreateCluster(createInput("alpha"), testAccountID) + _, err := f.svc.CreateCluster(createInput("alpha"), testAccountID, "") require.EqualError(t, err, awserrors.ErrorEKSResourceInUse) } @@ -84,7 +84,7 @@ func TestCreateCluster_FailedClusterIsReclaimedOnRetry(t *testing.T) { // First attempt fails at VM launch, leaving a FAILED meta with NLB ARNs. f.inst.launchErr = errors.New("no capacity") - _, err := f.svc.CreateCluster(createInput("alpha"), testAccountID) + _, err := f.svc.CreateCluster(createInput("alpha"), testAccountID, "") require.Error(t, err) failed, err := GetClusterMeta(f.kv, "alpha") require.NoError(t, err) @@ -93,7 +93,7 @@ func TestCreateCluster_FailedClusterIsReclaimedOnRetry(t *testing.T) { // Retry now succeeds: the reclaim path purges the failed attempt's NLB and // the create starts clean. f.inst.launchErr = nil - _, err = f.svc.CreateCluster(createInput("alpha"), testAccountID) + _, err = f.svc.CreateCluster(createInput("alpha"), testAccountID, "") require.NoError(t, err) got, err := GetClusterMeta(f.kv, "alpha") @@ -161,7 +161,7 @@ func TestCreateCluster_MissingAMIReturnsServiceUnavailable(t *testing.T) { f := newEKSServiceFixture(t) f.ami.describeOut = &ec2.DescribeImagesOutput{} // no images tagged managed-by=eks - _, err := f.svc.CreateCluster(createInput("alpha"), testAccountID) + _, err := f.svc.CreateCluster(createInput("alpha"), testAccountID, "") require.EqualError(t, err, awserrors.ErrorServiceUnavailable) meta, getErr := GetClusterMeta(f.kv, "alpha") @@ -172,7 +172,7 @@ func TestCreateCluster_MissingAMIReturnsServiceUnavailable(t *testing.T) { func TestCreateCluster_HappyPathPersistsActiveCreatingMeta(t *testing.T) { f := newEKSServiceFixture(t) - out, err := f.svc.CreateCluster(createInput("alpha"), testAccountID) + out, err := f.svc.CreateCluster(createInput("alpha"), testAccountID, "") require.NoError(t, err) require.NotNil(t, out) @@ -184,10 +184,43 @@ func TestCreateCluster_HappyPathPersistsActiveCreatingMeta(t *testing.T) { assert.NotEmpty(t, meta.ControlPlaneENIID) } +// bootstrapClusterCreatorAdminPermissions defaults true: a successful create +// with a known caller principal ARN mints a system:masters AccessEntry for the +// caller, keyed by their exact ARN (the token webhook looks it up by the same). +func TestCreateCluster_SeedsCreatorAdminAccessEntry(t *testing.T) { + f := newEKSServiceFixture(t) + const caller = "arn:aws:iam::111122223333:role/admin" + + _, err := f.svc.CreateCluster(createInput("alpha"), testAccountID, caller) + require.NoError(t, err) + + rec, err := GetAccessEntryRecord(f.kv, "alpha", caller) + require.NoError(t, err) + assert.Equal(t, []string{"system:masters"}, rec.KubernetesGroups) + assert.Equal(t, caller, rec.KubernetesUsername) + assert.Equal(t, AccessEntryTypeStandard, rec.Type) +} + +// With the bootstrap flag explicitly false, no creator-admin entry is minted. +func TestCreateCluster_SkipsCreatorAdminWhenDisabled(t *testing.T) { + f := newEKSServiceFixture(t) + const caller = "arn:aws:iam::111122223333:role/admin" + in := createInput("alpha") + in.AccessConfig = &eks.CreateAccessConfigRequest{ + BootstrapClusterCreatorAdminPermissions: aws.Bool(false), + } + + _, err := f.svc.CreateCluster(in, testAccountID, caller) + require.NoError(t, err) + + _, err = GetAccessEntryRecord(f.kv, "alpha", caller) + assert.ErrorIs(t, err, ErrAccessEntryNotFound) +} + func TestCreateCluster_AllocatesHiddenEgressEIP(t *testing.T) { f := newEKSServiceFixture(t) - _, err := f.svc.CreateCluster(createInput("egress"), testAccountID) + _, err := f.svc.CreateCluster(createInput("egress"), testAccountID, "") require.NoError(t, err) meta, err := GetClusterMeta(f.kv, "egress") diff --git a/spinifex/handlers/eks/service.go b/spinifex/handlers/eks/service.go index 96f793e39..aaf17ede8 100644 --- a/spinifex/handlers/eks/service.go +++ b/spinifex/handlers/eks/service.go @@ -11,8 +11,10 @@ import "github.com/aws/aws-sdk-go/service/eks" // (PascalCase). Signatures use the aws-sdk-go input/output types verbatim so // downstream callers can pass straight through from generated SDK code. type EKSService interface { - // Cluster - CreateCluster(input *eks.CreateClusterInput, accountID string) (*eks.CreateClusterOutput, error) + // Cluster. CreateCluster also takes the caller's resolved IAM principal ARN + // (from the X-Principal-ARN header) so it can mint the bootstrap + // cluster-creator-admin AccessEntry; "" skips that step. + CreateCluster(input *eks.CreateClusterInput, accountID, callerPrincipalARN string) (*eks.CreateClusterOutput, error) DescribeCluster(input *eks.DescribeClusterInput, accountID string) (*eks.DescribeClusterOutput, error) ListClusters(input *eks.ListClustersInput, accountID string) (*eks.ListClustersOutput, error) UpdateClusterConfig(input *eks.UpdateClusterConfigInput, accountID string) (*eks.UpdateClusterConfigOutput, error) diff --git a/spinifex/handlers/eks/service_impl.go b/spinifex/handlers/eks/service_impl.go index d6e457c45..e490e5bfc 100644 --- a/spinifex/handlers/eks/service_impl.go +++ b/spinifex/handlers/eks/service_impl.go @@ -250,7 +250,7 @@ func (s *EKSServiceImpl) missingOrchestrationDeps() []string { // --- Cluster --- -func (s *EKSServiceImpl) CreateCluster(input *eks.CreateClusterInput, accountID string) (*eks.CreateClusterOutput, error) { +func (s *EKSServiceImpl) CreateCluster(input *eks.CreateClusterInput, accountID, callerPrincipalARN string) (*eks.CreateClusterOutput, error) { if err := s.requireOrchestrationDeps("CreateCluster"); err != nil { return nil, err } @@ -421,12 +421,37 @@ func (s *EKSServiceImpl) CreateCluster(input *eks.CreateClusterInput, accountID return nil, err } + // bootstrapClusterCreatorAdminPermissions (default true): grant the caller + // system:masters via an AccessEntry so the cluster creator can immediately + // authenticate through the token webhook (Q9). Keyed by the caller's exact + // principal ARN — the webhook looks the entry up by the same ARN STS returns. + if bootstrapCreatorAdmin(input) && callerPrincipalARN != "" { + rec := newAccessEntryRecord(region, accountID, name, callerPrincipalARN, "", + []string{"system:masters"}, AccessEntryTypeStandard, nil, time.Now().UTC()) + if err := PutAccessEntryRecord(acctKV, rec); err != nil { + s.markFailed(acctKV, name) + return nil, fmt.Errorf("seed cluster-creator admin access entry: %w", err) + } + } else if bootstrapCreatorAdmin(input) { + slog.Warn("CreateCluster: bootstrapClusterCreatorAdminPermissions set but caller principal ARN unknown; skipping creator-admin AccessEntry", + "cluster", name, "accountID", accountID) + } + s.spawnBootstrap(accountID, name, acctKV, nil) s.spawnReconciler(accountID, name, meta) return &eks.CreateClusterOutput{Cluster: clusterMetaToAWS(meta)}, nil } +// bootstrapCreatorAdmin reports whether the cluster-creator-admin AccessEntry +// should be minted. AWS defaults this to true when unspecified. +func bootstrapCreatorAdmin(input *eks.CreateClusterInput) bool { + if input.AccessConfig == nil || input.AccessConfig.BootstrapClusterCreatorAdminPermissions == nil { + return true + } + return *input.AccessConfig.BootstrapClusterCreatorAdminPermissions +} + // systemEgressEvent is the wire shape for vpc.add-system-egress / // vpc.delete-system-egress (mirrors network/subscribers.SystemEgressEvent; // kept local to avoid importing the network layer from a handler). @@ -674,40 +699,298 @@ func (s *EKSServiceImpl) DeleteNodegroup(_ *eks.DeleteNodegroupInput, _ string) // --- AccessEntry + AccessPolicy --- -func (s *EKSServiceImpl) CreateAccessEntry(_ *eks.CreateAccessEntryInput, _ string) (*eks.CreateAccessEntryOutput, error) { - return nil, notImpl() +// acctKVForCluster opens the per-account bucket and verifies the cluster exists, +// translating absence to the AWS ResourceNotFoundException shape. The +// AccessEntry handlers all need both. +func (s *EKSServiceImpl) acctKVForCluster(accountID, cluster string) (nats.KeyValue, error) { + if cluster == "" { + return nil, errors.New(awserrors.ErrorInvalidParameterValue) + } + js, err := s.deps.NATSConn.JetStream() + if err != nil { + return nil, fmt.Errorf("jetstream: %w", err) + } + acctKV, err := GetOrCreateAccountBucket(js, accountID) + if err != nil { + return nil, fmt.Errorf("get account bucket: %w", err) + } + if _, err := GetClusterMeta(acctKV, cluster); err != nil { + if errors.Is(err, ErrClusterNotFound) { + return nil, errors.New(awserrors.ErrorEKSResourceNotFound) + } + return nil, err + } + return acctKV, nil } -func (s *EKSServiceImpl) DescribeAccessEntry(_ *eks.DescribeAccessEntryInput, _ string) (*eks.DescribeAccessEntryOutput, error) { - return nil, notImpl() +func (s *EKSServiceImpl) CreateAccessEntry(input *eks.CreateAccessEntryInput, accountID string) (*eks.CreateAccessEntryOutput, error) { + if input == nil { + return nil, errors.New(awserrors.ErrorInvalidParameterValue) + } + cluster := aws.StringValue(input.ClusterName) + principalARN := aws.StringValue(input.PrincipalArn) + if principalARN == "" { + return nil, errors.New(awserrors.ErrorInvalidParameterValue) + } + entryType := aws.StringValue(input.Type) + if entryType == "" { + entryType = AccessEntryTypeStandard + } + if entryType != AccessEntryTypeStandard { + // Node-type entries (EC2_LINUX/etc.) need the nodegroup join path; reject + // until Sprint 6d wires them. + return nil, errors.New(awserrors.ErrorInvalidParameterValue) + } + acctKV, err := s.acctKVForCluster(accountID, cluster) + if err != nil { + return nil, err + } + if _, err := GetAccessEntryRecord(acctKV, cluster, principalARN); err == nil { + return nil, errors.New(awserrors.ErrorEKSResourceInUse) + } else if !errors.Is(err, ErrAccessEntryNotFound) { + return nil, err + } + rec := newAccessEntryRecord(s.deps.Region, accountID, cluster, principalARN, + aws.StringValue(input.Username), aws.StringValueSlice(input.KubernetesGroups), + entryType, aws.StringValueMap(input.Tags), time.Now().UTC()) + if err := PutAccessEntryRecord(acctKV, rec); err != nil { + return nil, err + } + return &eks.CreateAccessEntryOutput{AccessEntry: accessEntryRecordToAWS(rec)}, nil } -func (s *EKSServiceImpl) ListAccessEntries(_ *eks.ListAccessEntriesInput, _ string) (*eks.ListAccessEntriesOutput, error) { - return nil, notImpl() +func (s *EKSServiceImpl) DescribeAccessEntry(input *eks.DescribeAccessEntryInput, accountID string) (*eks.DescribeAccessEntryOutput, error) { + if input == nil { + return nil, errors.New(awserrors.ErrorInvalidParameterValue) + } + cluster := aws.StringValue(input.ClusterName) + principalARN := aws.StringValue(input.PrincipalArn) + acctKV, err := s.acctKVForCluster(accountID, cluster) + if err != nil { + return nil, err + } + rec, err := GetAccessEntryRecord(acctKV, cluster, principalARN) + if err != nil { + if errors.Is(err, ErrAccessEntryNotFound) { + return nil, errors.New(awserrors.ErrorEKSResourceNotFound) + } + return nil, err + } + return &eks.DescribeAccessEntryOutput{AccessEntry: accessEntryRecordToAWS(rec)}, nil } -func (s *EKSServiceImpl) UpdateAccessEntry(_ *eks.UpdateAccessEntryInput, _ string) (*eks.UpdateAccessEntryOutput, error) { - return nil, notImpl() +func (s *EKSServiceImpl) ListAccessEntries(input *eks.ListAccessEntriesInput, accountID string) (*eks.ListAccessEntriesOutput, error) { + if input == nil { + return nil, errors.New(awserrors.ErrorInvalidParameterValue) + } + cluster := aws.StringValue(input.ClusterName) + acctKV, err := s.acctKVForCluster(accountID, cluster) + if err != nil { + return nil, err + } + recs, err := ListAccessEntryRecords(acctKV, cluster) + if err != nil { + return nil, err + } + filter := aws.StringValue(input.AssociatedPolicyArn) + arns := make([]string, 0, len(recs)) + for _, rec := range recs { + if filter != "" && !hasAssociatedPolicy(rec, filter) { + continue + } + arns = append(arns, rec.PrincipalARN) + } + return &eks.ListAccessEntriesOutput{AccessEntries: aws.StringSlice(arns)}, nil } -func (s *EKSServiceImpl) DeleteAccessEntry(_ *eks.DeleteAccessEntryInput, _ string) (*eks.DeleteAccessEntryOutput, error) { - return nil, notImpl() +func (s *EKSServiceImpl) UpdateAccessEntry(input *eks.UpdateAccessEntryInput, accountID string) (*eks.UpdateAccessEntryOutput, error) { + if input == nil { + return nil, errors.New(awserrors.ErrorInvalidParameterValue) + } + cluster := aws.StringValue(input.ClusterName) + principalARN := aws.StringValue(input.PrincipalArn) + acctKV, err := s.acctKVForCluster(accountID, cluster) + if err != nil { + return nil, err + } + now := time.Now().UTC() + rec, err := casUpdateAccessEntry(acctKV, cluster, principalARN, func(r *AccessEntryRecord) bool { + if input.KubernetesGroups != nil { + r.KubernetesGroups = aws.StringValueSlice(input.KubernetesGroups) + } + if u := aws.StringValue(input.Username); u != "" { + r.KubernetesUsername = u + } + r.ModifiedAt = now + return true + }) + if err != nil { + if errors.Is(err, ErrAccessEntryNotFound) { + return nil, errors.New(awserrors.ErrorEKSResourceNotFound) + } + return nil, err + } + return &eks.UpdateAccessEntryOutput{AccessEntry: accessEntryRecordToAWS(rec)}, nil } -func (s *EKSServiceImpl) AssociateAccessPolicy(_ *eks.AssociateAccessPolicyInput, _ string) (*eks.AssociateAccessPolicyOutput, error) { - return nil, notImpl() +func (s *EKSServiceImpl) DeleteAccessEntry(input *eks.DeleteAccessEntryInput, accountID string) (*eks.DeleteAccessEntryOutput, error) { + if input == nil { + return nil, errors.New(awserrors.ErrorInvalidParameterValue) + } + cluster := aws.StringValue(input.ClusterName) + principalARN := aws.StringValue(input.PrincipalArn) + acctKV, err := s.acctKVForCluster(accountID, cluster) + if err != nil { + return nil, err + } + if err := DeleteAccessEntryRecord(acctKV, cluster, principalARN); err != nil { + if errors.Is(err, ErrAccessEntryNotFound) { + return nil, errors.New(awserrors.ErrorEKSResourceNotFound) + } + return nil, err + } + return &eks.DeleteAccessEntryOutput{}, nil } -func (s *EKSServiceImpl) DisassociateAccessPolicy(_ *eks.DisassociateAccessPolicyInput, _ string) (*eks.DisassociateAccessPolicyOutput, error) { - return nil, notImpl() +func (s *EKSServiceImpl) AssociateAccessPolicy(input *eks.AssociateAccessPolicyInput, accountID string) (*eks.AssociateAccessPolicyOutput, error) { + if input == nil { + return nil, errors.New(awserrors.ErrorInvalidParameterValue) + } + cluster := aws.StringValue(input.ClusterName) + principalARN := aws.StringValue(input.PrincipalArn) + policyARN := aws.StringValue(input.PolicyArn) + if _, ok := supportedAccessPolicies[policyARN]; !ok { + return nil, errors.New(awserrors.ErrorInvalidParameterValue) + } + scope, err := validateAccessScope(input.AccessScope) + if err != nil { + return nil, errors.New(awserrors.ErrorInvalidParameterValue) + } + acctKV, err := s.acctKVForCluster(accountID, cluster) + if err != nil { + return nil, err + } + now := time.Now().UTC() + var assoc AssociatedAccessPolicy + _, err = casUpdateAccessEntry(acctKV, cluster, principalARN, func(r *AccessEntryRecord) bool { + assoc = AssociatedAccessPolicy{PolicyARN: policyARN, AccessScope: scope, AssociatedAt: now, ModifiedAt: now} + for i := range r.AssociatedPolicies { + if r.AssociatedPolicies[i].PolicyARN == policyARN { + assoc.AssociatedAt = r.AssociatedPolicies[i].AssociatedAt + r.AssociatedPolicies[i] = assoc + r.ModifiedAt = now + return true + } + } + r.AssociatedPolicies = append(r.AssociatedPolicies, assoc) + r.ModifiedAt = now + return true + }) + if err != nil { + if errors.Is(err, ErrAccessEntryNotFound) { + return nil, errors.New(awserrors.ErrorEKSResourceNotFound) + } + return nil, err + } + return &eks.AssociateAccessPolicyOutput{ + ClusterName: aws.String(cluster), + PrincipalArn: aws.String(principalARN), + AssociatedAccessPolicy: associatedPolicyToAWS(assoc), + }, nil } -func (s *EKSServiceImpl) ListAssociatedAccessPolicies(_ *eks.ListAssociatedAccessPoliciesInput, _ string) (*eks.ListAssociatedAccessPoliciesOutput, error) { - return nil, notImpl() +func (s *EKSServiceImpl) DisassociateAccessPolicy(input *eks.DisassociateAccessPolicyInput, accountID string) (*eks.DisassociateAccessPolicyOutput, error) { + if input == nil { + return nil, errors.New(awserrors.ErrorInvalidParameterValue) + } + cluster := aws.StringValue(input.ClusterName) + principalARN := aws.StringValue(input.PrincipalArn) + policyARN := aws.StringValue(input.PolicyArn) + acctKV, err := s.acctKVForCluster(accountID, cluster) + if err != nil { + return nil, err + } + now := time.Now().UTC() + _, err = casUpdateAccessEntry(acctKV, cluster, principalARN, func(r *AccessEntryRecord) bool { + for i := range r.AssociatedPolicies { + if r.AssociatedPolicies[i].PolicyARN == policyARN { + r.AssociatedPolicies = append(r.AssociatedPolicies[:i], r.AssociatedPolicies[i+1:]...) + r.ModifiedAt = now + return true + } + } + return false + }) + if err != nil { + if errors.Is(err, ErrAccessEntryNotFound) { + return nil, errors.New(awserrors.ErrorEKSResourceNotFound) + } + return nil, err + } + return &eks.DisassociateAccessPolicyOutput{}, nil +} + +func (s *EKSServiceImpl) ListAssociatedAccessPolicies(input *eks.ListAssociatedAccessPoliciesInput, accountID string) (*eks.ListAssociatedAccessPoliciesOutput, error) { + if input == nil { + return nil, errors.New(awserrors.ErrorInvalidParameterValue) + } + cluster := aws.StringValue(input.ClusterName) + principalARN := aws.StringValue(input.PrincipalArn) + acctKV, err := s.acctKVForCluster(accountID, cluster) + if err != nil { + return nil, err + } + rec, err := GetAccessEntryRecord(acctKV, cluster, principalARN) + if err != nil { + if errors.Is(err, ErrAccessEntryNotFound) { + return nil, errors.New(awserrors.ErrorEKSResourceNotFound) + } + return nil, err + } + policies := make([]*eks.AssociatedAccessPolicy, 0, len(rec.AssociatedPolicies)) + for _, p := range rec.AssociatedPolicies { + policies = append(policies, associatedPolicyToAWS(p)) + } + return &eks.ListAssociatedAccessPoliciesOutput{ + ClusterName: aws.String(cluster), + PrincipalArn: aws.String(principalARN), + AssociatedAccessPolicies: policies, + }, nil } func (s *EKSServiceImpl) ListAccessPolicies(_ *eks.ListAccessPoliciesInput, _ string) (*eks.ListAccessPoliciesOutput, error) { - return nil, notImpl() + arns := make([]string, 0, len(supportedAccessPolicies)) + for arn := range supportedAccessPolicies { + arns = append(arns, arn) + } + sort.Strings(arns) + policies := make([]*eks.AccessPolicy, 0, len(arns)) + for _, arn := range arns { + policies = append(policies, &eks.AccessPolicy{ + Arn: aws.String(arn), + Name: aws.String(accessPolicyName(arn)), + }) + } + return &eks.ListAccessPoliciesOutput{AccessPolicies: policies}, nil +} + +// hasAssociatedPolicy reports whether the entry has the given policy ARN bound. +func hasAssociatedPolicy(rec *AccessEntryRecord, policyARN string) bool { + for _, p := range rec.AssociatedPolicies { + if p.PolicyARN == policyARN { + return true + } + } + return false +} + +// accessPolicyName extracts the policy short name from its ARN +// (…/cluster-access-policy/AmazonEKSViewPolicy → AmazonEKSViewPolicy). +func accessPolicyName(arn string) string { + if i := strings.LastIndex(arn, "/"); i >= 0 { + return arn[i+1:] + } + return arn } // --- Addons --- diff --git a/spinifex/handlers/eks/service_impl_test.go b/spinifex/handlers/eks/service_impl_test.go index 5b293d557..b6dfe9a00 100644 --- a/spinifex/handlers/eks/service_impl_test.go +++ b/spinifex/handlers/eks/service_impl_test.go @@ -32,7 +32,7 @@ func setupTestService(t *testing.T) *EKSServiceImpl { func TestEKSServiceImpl_ClusterLifecycleShimMode(t *testing.T) { svc := setupTestService(t) - _, err := svc.CreateCluster(&eks.CreateClusterInput{Name: aws.String("c1")}, testAccountID) + _, err := svc.CreateCluster(&eks.CreateClusterInput{Name: aws.String("c1")}, testAccountID, "") require.EqualError(t, err, awserrors.ErrorServiceUnavailable) _, err = svc.DescribeCluster(&eks.DescribeClusterInput{Name: aws.String("c1")}, testAccountID) @@ -133,35 +133,169 @@ func TestEKSServiceImpl_NodegroupMethodsReturnNotImplemented(t *testing.T) { require.EqualError(t, err, awserrors.ErrorNotImplemented) } -func TestEKSServiceImpl_AccessMethodsReturnNotImplemented(t *testing.T) { +// seedTestCluster lays down a minimal ACTIVE cluster meta in the per-account +// bucket so the AccessEntry handlers (which gate on cluster existence) can run. +func seedTestCluster(t *testing.T, svc *EKSServiceImpl, cluster string) { + t.Helper() + js, err := svc.deps.NATSConn.JetStream() + require.NoError(t, err) + kv, err := GetOrCreateAccountBucket(js, testAccountID) + require.NoError(t, err) + require.NoError(t, PutClusterMeta(kv, &ClusterMeta{Name: cluster, Status: ClusterStatusActive})) +} + +const testPrincipalARN = "arn:aws:iam::111122223333:role/dev" + +func TestAccessEntry_UnknownClusterIsNotFound(t *testing.T) { svc := setupTestService(t) + _, err := svc.CreateAccessEntry(&eks.CreateAccessEntryInput{ + ClusterName: aws.String("missing"), PrincipalArn: aws.String(testPrincipalARN), + }, testAccountID) + require.EqualError(t, err, awserrors.ErrorEKSResourceNotFound) +} - _, err := svc.CreateAccessEntry(&eks.CreateAccessEntryInput{ClusterName: aws.String("c1"), PrincipalArn: aws.String("arn:aws:iam::111122223333:user/dev")}, testAccountID) - require.EqualError(t, err, awserrors.ErrorNotImplemented) +func TestAccessEntry_CreateDescribeListDelete(t *testing.T) { + svc := setupTestService(t) + seedTestCluster(t, svc, "c1") - _, err = svc.DescribeAccessEntry(&eks.DescribeAccessEntryInput{ClusterName: aws.String("c1")}, testAccountID) - require.EqualError(t, err, awserrors.ErrorNotImplemented) + out, err := svc.CreateAccessEntry(&eks.CreateAccessEntryInput{ + ClusterName: aws.String("c1"), + PrincipalArn: aws.String(testPrincipalARN), + KubernetesGroups: aws.StringSlice([]string{"system:masters"}), + }, testAccountID) + require.NoError(t, err) + require.NotNil(t, out.AccessEntry) + // Username defaults to the principal ARN; type defaults to STANDARD. + assert.Equal(t, testPrincipalARN, aws.StringValue(out.AccessEntry.Username)) + assert.Equal(t, AccessEntryTypeStandard, aws.StringValue(out.AccessEntry.Type)) + assert.Contains(t, aws.StringValue(out.AccessEntry.AccessEntryArn), ":access-entry/c1/") - _, err = svc.ListAccessEntries(&eks.ListAccessEntriesInput{ClusterName: aws.String("c1")}, testAccountID) - require.EqualError(t, err, awserrors.ErrorNotImplemented) + // Duplicate Create → ResourceInUseException. + _, err = svc.CreateAccessEntry(&eks.CreateAccessEntryInput{ + ClusterName: aws.String("c1"), PrincipalArn: aws.String(testPrincipalARN), + }, testAccountID) + require.EqualError(t, err, awserrors.ErrorEKSResourceInUse) - _, err = svc.UpdateAccessEntry(&eks.UpdateAccessEntryInput{ClusterName: aws.String("c1")}, testAccountID) - require.EqualError(t, err, awserrors.ErrorNotImplemented) + desc, err := svc.DescribeAccessEntry(&eks.DescribeAccessEntryInput{ + ClusterName: aws.String("c1"), PrincipalArn: aws.String(testPrincipalARN), + }, testAccountID) + require.NoError(t, err) + assert.Equal(t, []string{"system:masters"}, aws.StringValueSlice(desc.AccessEntry.KubernetesGroups)) - _, err = svc.DeleteAccessEntry(&eks.DeleteAccessEntryInput{ClusterName: aws.String("c1")}, testAccountID) - require.EqualError(t, err, awserrors.ErrorNotImplemented) + list, err := svc.ListAccessEntries(&eks.ListAccessEntriesInput{ClusterName: aws.String("c1")}, testAccountID) + require.NoError(t, err) + assert.Equal(t, []string{testPrincipalARN}, aws.StringValueSlice(list.AccessEntries)) - _, err = svc.AssociateAccessPolicy(&eks.AssociateAccessPolicyInput{ClusterName: aws.String("c1")}, testAccountID) - require.EqualError(t, err, awserrors.ErrorNotImplemented) + _, err = svc.DeleteAccessEntry(&eks.DeleteAccessEntryInput{ + ClusterName: aws.String("c1"), PrincipalArn: aws.String(testPrincipalARN), + }, testAccountID) + require.NoError(t, err) - _, err = svc.DisassociateAccessPolicy(&eks.DisassociateAccessPolicyInput{ClusterName: aws.String("c1")}, testAccountID) - require.EqualError(t, err, awserrors.ErrorNotImplemented) + _, err = svc.DescribeAccessEntry(&eks.DescribeAccessEntryInput{ + ClusterName: aws.String("c1"), PrincipalArn: aws.String(testPrincipalARN), + }, testAccountID) + require.EqualError(t, err, awserrors.ErrorEKSResourceNotFound) +} - _, err = svc.ListAssociatedAccessPolicies(&eks.ListAssociatedAccessPoliciesInput{ClusterName: aws.String("c1")}, testAccountID) - require.EqualError(t, err, awserrors.ErrorNotImplemented) +func TestAccessEntry_RejectsNodeType(t *testing.T) { + svc := setupTestService(t) + seedTestCluster(t, svc, "c1") + _, err := svc.CreateAccessEntry(&eks.CreateAccessEntryInput{ + ClusterName: aws.String("c1"), PrincipalArn: aws.String(testPrincipalARN), Type: aws.String("EC2_LINUX"), + }, testAccountID) + require.EqualError(t, err, awserrors.ErrorInvalidParameterValue) +} - _, err = svc.ListAccessPolicies(&eks.ListAccessPoliciesInput{}, testAccountID) - require.EqualError(t, err, awserrors.ErrorNotImplemented) +func TestAccessEntry_DescribeDeleteMissingIsNotFound(t *testing.T) { + svc := setupTestService(t) + seedTestCluster(t, svc, "c1") + _, err := svc.DescribeAccessEntry(&eks.DescribeAccessEntryInput{ + ClusterName: aws.String("c1"), PrincipalArn: aws.String(testPrincipalARN), + }, testAccountID) + require.EqualError(t, err, awserrors.ErrorEKSResourceNotFound) + _, err = svc.DeleteAccessEntry(&eks.DeleteAccessEntryInput{ + ClusterName: aws.String("c1"), PrincipalArn: aws.String(testPrincipalARN), + }, testAccountID) + require.EqualError(t, err, awserrors.ErrorEKSResourceNotFound) +} + +func TestAccessPolicy_AssociateListDisassociate(t *testing.T) { + svc := setupTestService(t) + seedTestCluster(t, svc, "c1") + _, err := svc.CreateAccessEntry(&eks.CreateAccessEntryInput{ + ClusterName: aws.String("c1"), PrincipalArn: aws.String(testPrincipalARN), + }, testAccountID) + require.NoError(t, err) + + const viewPolicy = "arn:aws:eks::aws:cluster-access-policy/AmazonEKSViewPolicy" + assoc, err := svc.AssociateAccessPolicy(&eks.AssociateAccessPolicyInput{ + ClusterName: aws.String("c1"), + PrincipalArn: aws.String(testPrincipalARN), + PolicyArn: aws.String(viewPolicy), + AccessScope: &eks.AccessScope{Type: aws.String("cluster")}, + }, testAccountID) + require.NoError(t, err) + assert.Equal(t, viewPolicy, aws.StringValue(assoc.AssociatedAccessPolicy.PolicyArn)) + + listed, err := svc.ListAssociatedAccessPolicies(&eks.ListAssociatedAccessPoliciesInput{ + ClusterName: aws.String("c1"), PrincipalArn: aws.String(testPrincipalARN), + }, testAccountID) + require.NoError(t, err) + require.Len(t, listed.AssociatedAccessPolicies, 1) + + // Entry now discoverable by the associated-policy filter. + filtered, err := svc.ListAccessEntries(&eks.ListAccessEntriesInput{ + ClusterName: aws.String("c1"), AssociatedPolicyArn: aws.String(viewPolicy), + }, testAccountID) + require.NoError(t, err) + assert.Equal(t, []string{testPrincipalARN}, aws.StringValueSlice(filtered.AccessEntries)) + + _, err = svc.DisassociateAccessPolicy(&eks.DisassociateAccessPolicyInput{ + ClusterName: aws.String("c1"), PrincipalArn: aws.String(testPrincipalARN), PolicyArn: aws.String(viewPolicy), + }, testAccountID) + require.NoError(t, err) + listed, err = svc.ListAssociatedAccessPolicies(&eks.ListAssociatedAccessPoliciesInput{ + ClusterName: aws.String("c1"), PrincipalArn: aws.String(testPrincipalARN), + }, testAccountID) + require.NoError(t, err) + assert.Empty(t, listed.AssociatedAccessPolicies) +} + +func TestAccessPolicy_AssociateRejectsUnsupportedPolicyAndScope(t *testing.T) { + svc := setupTestService(t) + seedTestCluster(t, svc, "c1") + _, err := svc.CreateAccessEntry(&eks.CreateAccessEntryInput{ + ClusterName: aws.String("c1"), PrincipalArn: aws.String(testPrincipalARN), + }, testAccountID) + require.NoError(t, err) + + // Unsupported policy ARN. + _, err = svc.AssociateAccessPolicy(&eks.AssociateAccessPolicyInput{ + ClusterName: aws.String("c1"), PrincipalArn: aws.String(testPrincipalARN), + PolicyArn: aws.String("arn:aws:eks::aws:cluster-access-policy/MadeUpPolicy"), + AccessScope: &eks.AccessScope{Type: aws.String("cluster")}, + }, testAccountID) + require.EqualError(t, err, awserrors.ErrorInvalidParameterValue) + + // namespace scope without namespaces. + _, err = svc.AssociateAccessPolicy(&eks.AssociateAccessPolicyInput{ + ClusterName: aws.String("c1"), PrincipalArn: aws.String(testPrincipalARN), + PolicyArn: aws.String("arn:aws:eks::aws:cluster-access-policy/AmazonEKSViewPolicy"), + AccessScope: &eks.AccessScope{Type: aws.String("namespace")}, + }, testAccountID) + require.EqualError(t, err, awserrors.ErrorInvalidParameterValue) +} + +func TestListAccessPolicies_ReturnsSupportedCatalogue(t *testing.T) { + svc := setupTestService(t) + out, err := svc.ListAccessPolicies(&eks.ListAccessPoliciesInput{}, testAccountID) + require.NoError(t, err) + require.Len(t, out.AccessPolicies, len(supportedAccessPolicies)) + for _, p := range out.AccessPolicies { + _, ok := supportedAccessPolicies[aws.StringValue(p.Arn)] + assert.True(t, ok, "unexpected policy %s", aws.StringValue(p.Arn)) + assert.NotEmpty(t, aws.StringValue(p.Name)) + } } func TestEKSServiceImpl_AddonsMethodsReturnNotImplemented(t *testing.T) { diff --git a/spinifex/handlers/eks/service_nats.go b/spinifex/handlers/eks/service_nats.go index e82194731..89534411a 100644 --- a/spinifex/handlers/eks/service_nats.go +++ b/spinifex/handlers/eks/service_nats.go @@ -25,8 +25,9 @@ func NewNATSEKSService(conn *nats.Conn) EKSService { // --- Cluster --- -func (s *NATSEKSService) CreateCluster(input *eks.CreateClusterInput, accountID string) (*eks.CreateClusterOutput, error) { - return utils.NATSRequest[eks.CreateClusterOutput](s.natsConn, "eks.CreateCluster", input, defaultTimeout, accountID) +func (s *NATSEKSService) CreateCluster(input *eks.CreateClusterInput, accountID, callerPrincipalARN string) (*eks.CreateClusterOutput, error) { + return utils.NATSRequest[eks.CreateClusterOutput](s.natsConn, "eks.CreateCluster", input, defaultTimeout, accountID, + utils.NATSHeader{Key: utils.PrincipalARNHeader, Value: callerPrincipalARN}) } func (s *NATSEKSService) DescribeCluster(input *eks.DescribeClusterInput, accountID string) (*eks.DescribeClusterOutput, error) { diff --git a/spinifex/handlers/eks/store.go b/spinifex/handlers/eks/store.go index 2e4048e78..1deca7f96 100644 --- a/spinifex/handlers/eks/store.go +++ b/spinifex/handlers/eks/store.go @@ -1,6 +1,8 @@ package handlers_eks import ( + "crypto/sha256" + "encoding/hex" "errors" "fmt" "log/slog" @@ -52,9 +54,23 @@ func NodegroupKey(cluster, ng string) string { return fmt.Sprintf("clusters/%s/nodegroups/%s", cluster, ng) } +// AccessEntriesPrefix returns the KV key prefix under which all of a cluster's +// AccessEntry records live. Used by ListAccessEntries to enumerate. +func AccessEntriesPrefix(cluster string) string { + return fmt.Sprintf("clusters/%s/access-entries/", cluster) +} + // AccessEntryKey returns the KV key for an AccessEntry record under a cluster. +// The principal ARN is hashed because IAM ARNs contain ':' which is not a legal +// NATS JetStream KV key character; the record itself carries the plaintext ARN. func AccessEntryKey(cluster, principalARN string) string { - return fmt.Sprintf("clusters/%s/access-entries/%s", cluster, principalARN) + return AccessEntriesPrefix(cluster) + PrincipalARNHash(principalARN) +} + +// PrincipalARNHash maps an IAM principal ARN to a KV-key-safe token. +func PrincipalARNHash(principalARN string) string { + sum := sha256.Sum256([]byte(principalARN)) + return hex.EncodeToString(sum[:]) } // OIDCProviderKey returns the KV key for a registered OIDC provider config diff --git a/spinifex/handlers/eks/store_test.go b/spinifex/handlers/eks/store_test.go index 64879eb2e..8432a6a88 100644 --- a/spinifex/handlers/eks/store_test.go +++ b/spinifex/handlers/eks/store_test.go @@ -47,7 +47,7 @@ func TestKeyPaths_MatchQ2Spec(t *testing.T) { }{ {"ClusterMetaKey", ClusterMetaKey("alpha"), "clusters/alpha/meta"}, {"NodegroupKey", NodegroupKey("alpha", "ng-1"), "clusters/alpha/nodegroups/ng-1"}, - {"AccessEntryKey", AccessEntryKey("alpha", "arn:aws:iam::111122223333:user/dev"), "clusters/alpha/access-entries/arn:aws:iam::111122223333:user/dev"}, + {"AccessEntryKey", AccessEntryKey("alpha", "arn:aws:iam::111122223333:user/dev"), "clusters/alpha/access-entries/" + PrincipalARNHash("arn:aws:iam::111122223333:user/dev")}, {"OIDCProviderKey", OIDCProviderKey("alpha", "abc123"), "clusters/alpha/oidc-providers/abc123"}, {"OIDCSigningKeyKey", OIDCSigningKeyKey("alpha"), "clusters/alpha/oidc-signing-key.pem.enc"}, {"OIDCJWKSKey", OIDCJWKSKey("alpha"), "clusters/alpha/oidc-jwks.json"}, diff --git a/spinifex/handlers/eks/token_webhook.go b/spinifex/handlers/eks/token_webhook.go new file mode 100644 index 000000000..8f4e483e0 --- /dev/null +++ b/spinifex/handlers/eks/token_webhook.go @@ -0,0 +1,58 @@ +package handlers_eks + +import ( + "encoding/base64" + "errors" + "strings" +) + +// TokenVerifySubject is the NATS request subject the EKS token webhook calls to +// resolve an `aws eks get-token` presigned URL into a caller principal. It is +// served by the awsgw service (which hosts the in-process STS service) because +// STS is gateway-local and not otherwise exposed on NATS. +const TokenVerifySubject = "eks.VerifyToken" //nolint:gosec // NATS subject name, not a credential + +// getTokenV1Prefix is the scheme prefix `aws eks get-token` prepends to the +// base64url-encoded presigned STS GetCallerIdentity URL. +const getTokenV1Prefix = "k8s-aws-v1." //nolint:gosec // token scheme prefix, not a credential + +// ErrMalformedToken is returned by DecodeGetToken when the bearer token is not +// a well-formed `k8s-aws-v1.` value. +var ErrMalformedToken = errors.New("eks: malformed k8s-aws-v1 token") + +// TokenVerifyRequest is the NATS request payload sent to TokenVerifySubject. +// ClusterName is the anti-replay binding (Q10): the webhook passes its own +// cluster name and STS rejects a URL signed for any other x-k8s-aws-id. +type TokenVerifyRequest struct { + PresignedURL string `json:"presignedURL"` + ClusterName string `json:"clusterName"` +} + +// TokenVerifyResponse is the NATS reply: the resolved caller principal. It +// mirrors the fields of handlers_sts.PresignedCallerIdentity the webhook needs +// to key its AccessEntry lookup. +type TokenVerifyResponse struct { + AccountID string `json:"accountID"` + ARN string `json:"arn"` + UserID string `json:"userID"` + PrincipalType string `json:"principalType"` +} + +// DecodeGetToken extracts the presigned STS GetCallerIdentity URL from a +// `k8s-aws-v1.` bearer token. The encoding is base64url without +// padding, matching the aws-iam-authenticator / aws-cli get-token output. +func DecodeGetToken(token string) (string, error) { + token = strings.TrimSpace(token) + if !strings.HasPrefix(token, getTokenV1Prefix) { + return "", ErrMalformedToken + } + encoded := strings.TrimPrefix(token, getTokenV1Prefix) + raw, err := base64.RawURLEncoding.DecodeString(encoded) + if err != nil { + return "", ErrMalformedToken + } + if len(raw) == 0 { + return "", ErrMalformedToken + } + return string(raw), nil +} diff --git a/spinifex/handlers/eks/token_webhook_test.go b/spinifex/handlers/eks/token_webhook_test.go new file mode 100644 index 000000000..69f89c1a8 --- /dev/null +++ b/spinifex/handlers/eks/token_webhook_test.go @@ -0,0 +1,44 @@ +package handlers_eks + +import ( + "encoding/base64" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestDecodeGetToken_RoundTrips(t *testing.T) { + url := "https://sts.amazonaws.com/?Action=GetCallerIdentity&Version=2011-06-15&X-Amz-Algorithm=AWS4-HMAC-SHA256" + token := getTokenV1Prefix + base64.RawURLEncoding.EncodeToString([]byte(url)) + + got, err := DecodeGetToken(token) + require.NoError(t, err) + assert.Equal(t, url, got) +} + +func TestDecodeGetToken_TrimsWhitespace(t *testing.T) { + url := "https://sts.amazonaws.com/?Action=GetCallerIdentity" + token := " " + getTokenV1Prefix + base64.RawURLEncoding.EncodeToString([]byte(url)) + "\n" + + got, err := DecodeGetToken(token) + require.NoError(t, err) + assert.Equal(t, url, got) +} + +func TestDecodeGetToken_RejectsMalformed(t *testing.T) { + cases := map[string]string{ + "missing prefix": base64.RawURLEncoding.EncodeToString([]byte("https://sts")), + "empty": "", + "prefix only": getTokenV1Prefix, + "bad base64": getTokenV1Prefix + "!!!not-base64!!!", + "padded base64": getTokenV1Prefix + base64.URLEncoding.EncodeToString([]byte("https://sts.amazonaws.com/?x=12")), + "wrong scheme": "k8s-aws-v2." + base64.RawURLEncoding.EncodeToString([]byte("https://sts")), + } + for name, tok := range cases { + t.Run(name, func(t *testing.T) { + _, err := DecodeGetToken(tok) + assert.ErrorIs(t, err, ErrMalformedToken) + }) + } +} diff --git a/spinifex/handlers/eks/types.go b/spinifex/handlers/eks/types.go index c42bf0997..3e91d889d 100644 --- a/spinifex/handlers/eks/types.go +++ b/spinifex/handlers/eks/types.go @@ -40,12 +40,33 @@ type NodegroupRecord struct { // AccessEntryRecord is the persisted-state envelope for an EKS API-mode // AccessEntry (Q9). type AccessEntryRecord struct { - ClusterName string `json:"clusterName"` - PrincipalARN string `json:"principalARN"` - KubernetesUsername string `json:"kubernetesUsername"` - KubernetesGroups []string `json:"kubernetesGroups,omitempty"` - Type string `json:"type"` - CreatedAt time.Time `json:"createdAt"` + ARN string `json:"arn"` + ClusterName string `json:"clusterName"` + PrincipalARN string `json:"principalARN"` + KubernetesUsername string `json:"kubernetesUsername"` + KubernetesGroups []string `json:"kubernetesGroups,omitempty"` + Type string `json:"type"` + Tags map[string]string `json:"tags,omitempty"` + AssociatedPolicies []AssociatedAccessPolicy `json:"associatedPolicies,omitempty"` + CreatedAt time.Time `json:"createdAt"` + ModifiedAt time.Time `json:"modifiedAt"` +} + +// AssociatedAccessPolicy is a managed access policy bound to an AccessEntry +// with a cluster- or namespace-scoped grant (Q9). +type AssociatedAccessPolicy struct { + PolicyARN string `json:"policyARN"` + AccessScope AccessScope `json:"accessScope"` + AssociatedAt time.Time `json:"associatedAt"` + ModifiedAt time.Time `json:"modifiedAt"` +} + +// AccessScope restricts an associated access policy to the whole cluster or a +// set of namespaces. Type is "cluster" or "namespace"; Namespaces is required +// (and only meaningful) when Type is "namespace". +type AccessScope struct { + Type string `json:"type"` + Namespaces []string `json:"namespaces,omitempty"` } // OIDCProviderConfigRecord captures the minimum needed to register an OIDC diff --git a/spinifex/handlers/sts/presigned_url_verify.go b/spinifex/handlers/sts/presigned_url_verify.go index 1414149b3..c605c58f4 100644 --- a/spinifex/handlers/sts/presigned_url_verify.go +++ b/spinifex/handlers/sts/presigned_url_verify.go @@ -143,7 +143,11 @@ func (s *STSServiceImpl) VerifyPresignedGetCallerIdentity(presignedURL, expected canonicalQuery, canonicalHeaders, signedHeadersList, - "UNSIGNED-PAYLOAD", + // botocore / aws-sdk presign `sts:GetCallerIdentity` (a non-S3, empty-body + // request) with the SHA256 of the empty string as the payload hash. + // "UNSIGNED-PAYLOAD" is an S3-only convention and would mismatch every + // real `aws eks get-token`. + emptyStringSHA256, }, "\n") stringToSign := strings.Join([]string{ @@ -360,6 +364,10 @@ func hexSHA256(s string) string { return hex.EncodeToString(sum[:]) } +// emptyStringSHA256 is the SigV4 payload hash for an empty body — the value +// botocore/aws-sdk use when presigning non-S3 requests like GetCallerIdentity. +var emptyStringSHA256 = hexSHA256("") + func hmacSHA256(key []byte, data string) []byte { mac := hmac.New(sha256.New, key) mac.Write([]byte(data)) diff --git a/spinifex/handlers/sts/presigned_url_verify_test.go b/spinifex/handlers/sts/presigned_url_verify_test.go index ed0b5af05..e7c05b7dd 100644 --- a/spinifex/handlers/sts/presigned_url_verify_test.go +++ b/spinifex/handlers/sts/presigned_url_verify_test.go @@ -2,6 +2,7 @@ package handlers_sts import ( "encoding/hex" + "net/http" "net/url" "sort" "strconv" @@ -10,6 +11,8 @@ import ( "time" "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/aws/credentials" + v4 "github.com/aws/aws-sdk-go/aws/signer/v4" "github.com/aws/aws-sdk-go/service/iam" "github.com/mulgadc/spinifex/spinifex/awserrors" "github.com/stretchr/testify/assert" @@ -52,7 +55,7 @@ func presignTestURL(t *testing.T, accessKeyID, secret, clusterName string, signe canonHeaders := "host:" + host + "\n" + "x-k8s-aws-id:" + clusterName + "\n" signedHeadersList := "host;x-k8s-aws-id" canonRequest := strings.Join([]string{ - "GET", "/", canonQ, canonHeaders, signedHeadersList, "UNSIGNED-PAYLOAD", + "GET", "/", canonQ, canonHeaders, signedHeadersList, emptyStringSHA256, }, "\n") stringToSign := strings.Join([]string{ @@ -204,7 +207,7 @@ func TestVerifyPresignedGetCallerIdentity_MissingXK8sAwsIdInSignedHeaders(t *tes canonQ := canonicalQueryStringForTest(q) canonHeaders := "host:" + testPresignedHost + "\n" canonReq := strings.Join([]string{ - "GET", "/", canonQ, canonHeaders, "host", "UNSIGNED-PAYLOAD", + "GET", "/", canonQ, canonHeaders, "host", emptyStringSHA256, }, "\n") stringToSign := strings.Join([]string{ "AWS4-HMAC-SHA256", amzDate, credentialScope, hexSHA256(canonReq), @@ -383,3 +386,38 @@ func TestDeriveSigningKey_KnownVector(t *testing.T) { const wantHex = "f4780e2d9f65fa895f9c67b32ce1baf0b0d8a43505a000a1a9e090d414db404d" assert.Equal(t, wantHex, hex.EncodeToString(key)) } + +// TestVerifyPresignedGetCallerIdentity_RealSDKPresign signs the URL with the +// actual aws-sdk-go v4 presigner — the same canonicalisation `aws eks +// get-token` uses — rather than the hand-rolled presignTestURL. This is the +// faithful end-to-end guard: it caught the payload-hash bug (verify used the +// S3-only "UNSIGNED-PAYLOAD" instead of the empty-string SHA256 that botocore +// signs non-S3 requests with), which every hand-rolled test masked by using +// the same wrong constant. +func TestVerifyPresignedGetCallerIdentity_RealSDKPresign(t *testing.T) { + svc, _ := newTestSetup(t) + akid, secret := seedAccessKey(t, svc, testCallerAccountID, "carol") + + signedAt := time.Now().UTC().Truncate(time.Second) + withFrozenTime(t, signedAt) + + const cluster = "prod-cluster" + req, err := http.NewRequest(http.MethodGet, + "https://"+testPresignedHost+"/?Action=GetCallerIdentity&Version=2011-06-15", nil) + require.NoError(t, err) + req.Header.Set("X-K8s-Aws-Id", cluster) + + signer := v4.NewSigner(credentials.NewStaticCredentials(akid, secret, "")) + _, err = signer.Presign(req, nil, "sts", testPresignedRegion, 900*time.Second, signedAt) + require.NoError(t, err) + + got, err := svc.VerifyPresignedGetCallerIdentity(req.URL.String(), cluster) + require.NoError(t, err) + require.NotNil(t, got) + assert.Equal(t, cluster, got.XK8sAwsID) + assert.Contains(t, got.ARN, ":user/carol") + + // Anti-replay: the SDK-signed URL must NOT verify against a different cluster. + _, err = svc.VerifyPresignedGetCallerIdentity(req.URL.String(), "other-cluster") + require.Error(t, err) +} diff --git a/spinifex/services/awsgw/awsgw.go b/spinifex/services/awsgw/awsgw.go index ac7f49c1e..992089724 100644 --- a/spinifex/services/awsgw/awsgw.go +++ b/spinifex/services/awsgw/awsgw.go @@ -146,6 +146,19 @@ func launchService(config *config.ClusterConfig) error { defer cancelJanitor() go stsService.RunJanitor(janitorCtx) + // Expose the in-process STS presigned-URL verify over NATS for the + // in-cluster eks-token-webhook (STS is gateway-local, not otherwise on the + // bus). Bound to the process lifetime via natsConn.Close on return. + tokenVerifySub, err := registerEKSTokenVerify(natsConn, stsService) + if err != nil { + return fmt.Errorf("subscribe EKS token verify: %w", err) + } + defer func() { + if err := tokenVerifySub.Unsubscribe(); err != nil { + slog.Warn("EKS token verify: unsubscribe failed", "err", err) + } + }() + // First boot: consume bootstrap.json → seed IAM users into NATS KV → delete file. // Check data directory first (production: /var/lib/spinifex/awsgw/), then // awsgw subdir (dev: ~/spinifex/awsgw/), then legacy config dir. diff --git a/spinifex/services/awsgw/eks_token_verify.go b/spinifex/services/awsgw/eks_token_verify.go new file mode 100644 index 000000000..730fc8e9c --- /dev/null +++ b/spinifex/services/awsgw/eks_token_verify.go @@ -0,0 +1,76 @@ +package awsgw + +import ( + "encoding/json" + "log/slog" + + "github.com/mulgadc/spinifex/spinifex/awserrors" + handlers_eks "github.com/mulgadc/spinifex/spinifex/handlers/eks" + handlers_sts "github.com/mulgadc/spinifex/spinifex/handlers/sts" + "github.com/mulgadc/spinifex/spinifex/utils" + "github.com/nats-io/nats.go" +) + +// eksTokenVerifyQueue is the NATS queue group for the EKS token-verify +// responders so the request is load-balanced across awsgw nodes. +const eksTokenVerifyQueue = "awsgw-eks-token-verify" //nolint:gosec // queue-group name, not a credential + +// presignedVerifier is the slice of the STS service the token-verify responder +// needs: resolve an `aws eks get-token` presigned URL into a caller principal, +// binding the signature to the expected cluster name (anti-replay, Q10). +type presignedVerifier interface { + VerifyPresignedGetCallerIdentity(presignedURL, expectedClusterName string) (*handlers_sts.PresignedCallerIdentity, error) +} + +var _ presignedVerifier = (*handlers_sts.STSServiceImpl)(nil) + +// registerEKSTokenVerify subscribes the awsgw service to TokenVerifySubject so +// the in-cluster eks-token-webhook can resolve get-token presigned URLs over +// NATS (STS is hosted in-process here and is not otherwise on the bus). +func registerEKSTokenVerify(nc *nats.Conn, verifier presignedVerifier) (*nats.Subscription, error) { + return nc.QueueSubscribe(handlers_eks.TokenVerifySubject, eksTokenVerifyQueue, handleEKSTokenVerify(verifier)) +} + +func handleEKSTokenVerify(verifier presignedVerifier) nats.MsgHandler { + return func(msg *nats.Msg) { + var req handlers_eks.TokenVerifyRequest + if err := json.Unmarshal(msg.Data, &req); err != nil { + respondTokenVerifyErr(msg, awserrors.ErrorInvalidParameterValue) + return + } + if req.PresignedURL == "" || req.ClusterName == "" { + respondTokenVerifyErr(msg, awserrors.ErrorInvalidParameterValue) + return + } + + ident, err := verifier.VerifyPresignedGetCallerIdentity(req.PresignedURL, req.ClusterName) + if err != nil { + // A verify failure is the common, expected case for a forged/replayed + // token — log at debug so a port-scan does not flood the error log. + slog.Debug("EKS token verify rejected", "cluster", req.ClusterName, "err", err) + respondTokenVerifyErr(msg, awserrors.ValidErrorCode(err.Error())) + return + } + + resp := handlers_eks.TokenVerifyResponse{ + AccountID: ident.AccountID, + ARN: ident.ARN, + UserID: ident.UserID, + PrincipalType: ident.PrincipalType, + } + data, err := json.Marshal(resp) + if err != nil { + respondTokenVerifyErr(msg, awserrors.ErrorServerInternal) + return + } + if err := msg.Respond(data); err != nil { + slog.Error("EKS token verify: failed to respond", "err", err) + } + } +} + +func respondTokenVerifyErr(msg *nats.Msg, code string) { + if err := msg.Respond(utils.GenerateErrorPayload(code)); err != nil { + slog.Error("EKS token verify: failed to respond with error", "err", err) + } +} diff --git a/spinifex/services/awsgw/eks_token_verify_test.go b/spinifex/services/awsgw/eks_token_verify_test.go new file mode 100644 index 000000000..069193fed --- /dev/null +++ b/spinifex/services/awsgw/eks_token_verify_test.go @@ -0,0 +1,79 @@ +package awsgw + +import ( + "errors" + "testing" + "time" + + handlers_eks "github.com/mulgadc/spinifex/spinifex/handlers/eks" + handlers_sts "github.com/mulgadc/spinifex/spinifex/handlers/sts" + "github.com/mulgadc/spinifex/spinifex/testutil" + "github.com/mulgadc/spinifex/spinifex/utils" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +type fakeVerifier struct { + ident *handlers_sts.PresignedCallerIdentity + err error + gotURL string + gotName string +} + +func (f *fakeVerifier) VerifyPresignedGetCallerIdentity(presignedURL, expectedClusterName string) (*handlers_sts.PresignedCallerIdentity, error) { + f.gotURL = presignedURL + f.gotName = expectedClusterName + return f.ident, f.err +} + +func TestEKSTokenVerify_ResolvesPrincipal(t *testing.T) { + _, nc := testutil.StartTestNATS(t) + fake := &fakeVerifier{ident: &handlers_sts.PresignedCallerIdentity{ + AccountID: "111122223333", + ARN: "arn:aws:iam::111122223333:role/admin", + UserID: "AROAEXAMPLE:sess", + PrincipalType: "AssumedRole", + }} + sub, err := registerEKSTokenVerify(nc, fake) + require.NoError(t, err) + defer func() { _ = sub.Unsubscribe() }() + + req := handlers_eks.TokenVerifyRequest{PresignedURL: "https://sts/?x=1", ClusterName: "alpha"} + resp, err := utils.NATSRequest[handlers_eks.TokenVerifyResponse]( + nc, handlers_eks.TokenVerifySubject, req, 2*time.Second, "") + require.NoError(t, err) + + assert.Equal(t, "arn:aws:iam::111122223333:role/admin", resp.ARN) + assert.Equal(t, "AROAEXAMPLE:sess", resp.UserID) + assert.Equal(t, "AssumedRole", resp.PrincipalType) + // Cluster name is forwarded for the STS anti-replay pin. + assert.Equal(t, "alpha", fake.gotName) +} + +func TestEKSTokenVerify_RejectedTokenIsError(t *testing.T) { + _, nc := testutil.StartTestNATS(t) + fake := &fakeVerifier{err: errors.New("signature mismatch (cluster name replay)")} + sub, err := registerEKSTokenVerify(nc, fake) + require.NoError(t, err) + defer func() { _ = sub.Unsubscribe() }() + + req := handlers_eks.TokenVerifyRequest{PresignedURL: "https://sts/?x=1", ClusterName: "alpha"} + _, err = utils.NATSRequest[handlers_eks.TokenVerifyResponse]( + nc, handlers_eks.TokenVerifySubject, req, 2*time.Second, "") + require.Error(t, err, "a verify failure must surface as a NATS error payload") +} + +func TestEKSTokenVerify_RejectsEmptyFields(t *testing.T) { + _, nc := testutil.StartTestNATS(t) + fake := &fakeVerifier{ident: &handlers_sts.PresignedCallerIdentity{ARN: "arn:..."}} + sub, err := registerEKSTokenVerify(nc, fake) + require.NoError(t, err) + defer func() { _ = sub.Unsubscribe() }() + + // Missing ClusterName — responder rejects before calling the verifier. + req := handlers_eks.TokenVerifyRequest{PresignedURL: "https://sts/?x=1"} + _, err = utils.NATSRequest[handlers_eks.TokenVerifyResponse]( + nc, handlers_eks.TokenVerifySubject, req, 2*time.Second, "") + require.Error(t, err) + assert.Empty(t, fake.gotURL, "verifier must not run on an invalid request") +} diff --git a/spinifex/utils/nats.go b/spinifex/utils/nats.go index 19426f24b..bda51cbe3 100644 --- a/spinifex/utils/nats.go +++ b/spinifex/utils/nats.go @@ -230,12 +230,30 @@ func ConnectNATSWithRetry(host, token, caCertPath string, opts ...RetryOption) ( // AWS account ID from the gateway to daemon handlers. const AccountIDHeader = "X-Account-ID" +// PrincipalARNHeader carries the caller's resolved IAM principal ARN from the +// gateway to daemon handlers that attribute an action to its caller (e.g. the +// EKS bootstrap-creator-admin AccessEntry). +const PrincipalARNHeader = "X-Principal-ARN" + +// NATSHeader is an extra request header passed to NATSRequest beyond the +// always-set X-Account-ID. +type NATSHeader struct{ Key, Value string } + +// PrincipalARNFromMsg extracts the caller's principal ARN from a NATS message +// header. Returns "" when absent. +func PrincipalARNFromMsg(msg *nats.Msg) string { + if msg == nil || msg.Header == nil { + return "" + } + return msg.Header.Get(PrincipalARNHeader) +} + // NATSRequest performs a NATS request-response with JSON marshaling. // It marshals the input, sends to the given subject with the X-Account-ID -// header, validates the response for error payloads, and unmarshals the -// successful response into Out. Handlers can ignore the account ID if the -// operation is unscoped (e.g. DescribeInstanceTypes). -func NATSRequest[Out any](conn *nats.Conn, subject string, input any, timeout time.Duration, accountID string) (*Out, error) { +// header (plus any extra headers), validates the response for error payloads, +// and unmarshals the successful response into Out. Handlers can ignore the +// account ID if the operation is unscoped (e.g. DescribeInstanceTypes). +func NATSRequest[Out any](conn *nats.Conn, subject string, input any, timeout time.Duration, accountID string, headers ...NATSHeader) (*Out, error) { if conn == nil || !conn.IsConnected() { return nil, ErrClusterUnavailable } @@ -248,6 +266,11 @@ func NATSRequest[Out any](conn *nats.Conn, subject string, input any, timeout ti reqMsg := nats.NewMsg(subject) reqMsg.Data = jsonData reqMsg.Header.Set(AccountIDHeader, accountID) + for _, h := range headers { + if h.Key != "" { + reqMsg.Header.Set(h.Key, h.Value) + } + } msg, err := conn.RequestMsg(reqMsg, timeout) if err != nil {