From 496424c41c7dda52251a96444bc8a767e129d76a Mon Sep 17 00:00:00 2001 From: Josh Siviour Date: Sun, 7 Jun 2026 20:42:25 +1000 Subject: [PATCH] test(eks): EKS control-plane E2E suite (Sprint 6g) (#304) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * test(eks): EKS control-plane e2e suite (Phase 1, API-level) Add tests/e2e/eks/eks_test.go driving the EKS lifecycle against awsgw: CreateCluster -> ACTIVE (endpoint + CA asserted), kubeconfig artifact assembly, AccessEntry CRUD, get-token presign decode, DeleteCluster -> gone. Harness: EKS SDK client on AWSClient + WaitForEKSCluster{Active,Deleted} pollers. Wire e2e-eks into the Makefile and the single-node CI suite list. * test(eks): Phase 2 kubectl get-nodes data-plane gate Add the reachability shim (harness/eks_mgmt.go) that reads the control-plane VM's br-mgmt IP from the cluster KV meta over NATS, plus a kubectl wrapper (harness/kubectl.go) that skips when kubectl is absent. KubectlGetNodes points kubectl at https://:6443 (insecure-skip-tls-verify; cert SANs the NLB DNS, not the mgmt IP) with an aws eks get-token exec block and polls until the control-plane node is Ready — the get-token webhook auth regression gate. IRSA pod + the real NLB/Route53 passthrough are deferred to a follow-up. * test(eks): tolerate doubled NotFound code + dial loopback NATS DeleteCluster detection matched the SDK constant ResourceNotFoundException, but awsgw emits ResourceNotFoundExceptionException on the wire — match the NotFound code by substring so deletion is detected immediately instead of timing out. The mgmt-IP shim dialed NATS.Host verbatim (0.0.0.0:4222), which the serving cert does not SAN; rewrite the wildcard bind to loopback. Validated live: all 6 subtests green, kubectl get nodes shows the control-plane node Ready via the get-token webhook auth path. * feat: build system images with libguestfs (no host block devices) Replace the qemu-nbd + sudo mount + chroot + /dev-bind flow in build-system-image.sh with libguestfs (virt-customize / virt-resize / guestfish). Every operation runs in an isolated userspace appliance VM with its own kernel, so the build touches no host block device, no host mount, and needs no sudo — removing the failure class that wiped the host's /dev on shared CI runners (a leaked /dev bind traversed by rm -rf). Applies to all distros: alpine resizes via guestfish e2fsck+resize2fs, ubuntu via virt-resize; customization (packages, binaries, files, setup script, service enable, cache cleanup) is one ordered virt-customize invocation. LIBGUESTFS_BACKEND=direct (no libvirt on runners). Wire the eks-node AMI into the single-node e2e job: the runner builds the device-free raw, ships it to the VM, and the VM imports it (predastore/NATS are VM-only) so the eks suite's CreateCluster has an AMI to boot. * feat(eks): bootstrap + state-report via AWS gateway broker Converge the EKS control-plane → host channel onto the ELBv2 lb-agent model: the K3s server VM publishes its bootstrap envelopes and periodic state reports via SigV4-signed HTTPS POST to the AWS gateway, instead of dialing core NATS directly. A new gateway action (PublishInternal, POST /clusters/{name}/internal-publish) relays the payload verbatim onto the existing eks.bus.* / eks.state.* subjects, so the daemon-side subscriber, KV persistence/encryption, and reconciler are unchanged. This removes the VM's need to reach NATS at all, so NATS can stay cluster-internal behind its bind/firewall/mTLS controls — the prior direct path broke when NATS was not VM-reachable (e2e CreateCluster timeout) and was DDIL-fragile (fire-and-forget nats pub, no ack/retry). - gateway/eks: PublishInternal broker + route; closed-set channel/kind validation - cmd/eks-gateway-publish: one-shot SigV4 publisher (retry-until-ack) baked into the eks-node AMI; replaces the nats-cli publisher in k3s-first-boot.sh and mulga-eks-state-report.sh - handlers/eks + daemon: hand the K3s server VM the mgmt-reachable gateway URL + system (Predastore) SigV4 creds + gateway CA instead of NATS URL/token/CA; new resolveSystemGatewayBaseURL (mgmt-bridge-first) - eks-node image: build/install eks-gateway-publish, drop nats-cli download * feat: migrate eks-token-webhook to the AWS gateway broker The TokenReview webhook dialed core NATS directly (STS verify + AccessEntry KV read). On a mgmt-only control-plane VM with NATS bound off-subnet that fails, so the webhook exited at startup, the apiserver token-webhook kubeconfig was never written, and the apiserver crash-looped — the cluster never reached ACTIVE. Converge it onto the same gateway-broker model as bootstrap/state-report and ELBv2's lb-agent: the webhook SigV4-signs an HTTPS POST {accountId,token} to /clusters/{name}/token-review, and the gateway runs the verify + lookup host-side and returns the resolved identity. The VM now speaks no core NATS. - internal/eksgw: shared on-VM SigV4 client (also adopted by eks-gateway-publish). - handlers/eks/token_review.go: Authenticate decision core (moved from the webhook binary) + ResolveTokenReview host-side wiring. - gateway/eks token-review route + WebhookTokenReview handler. - webhook: drop NATS, relay via eksgw.Client; broker fault -> 503 (retryable) not a silent deny. - initd: export EKS_GATEWAY_* / EKS_ACCESS_* creds, not SPINIFEX_NATS_*. * fix: build eks-node helpers with GOFIPS140; cover token-review broker eks-gateway-publish blank-imports internal/fipsboot, whose init() panics unless the binary is built with GOFIPS140=v1.0.0. The eks-node AMI build omitted the flag, so the binary paniced at first boot right after k3s came up — k3s-first-boot (set -eu) died before publishing the bootstrap token and the cluster never reached ACTIVE. Build both eks-node helpers with GOFIPS140=v1.0.0, matching cmd/lb-agent. Add NATS-backed unit tests for ResolveTokenReview and the WebhookTokenReview gateway handler (verify responder + account KV) to cover the new broker path. * fix: export first-boot env so eks-gateway-publish child sees it k3s-first-boot.sh and mulga-eks-state-report.sh sourced /etc/spinifex-eks/ first-boot.env but never exported the vars, so the eks-gateway-publish child process inherited an empty environment and exited '--account-id is required'. The control-plane VM came up healthy but never published its bootstrap envelopes, so the cluster never reached ACTIVE. Wrap the source in set -a / set +a so the KEY=value lines export to child processes. * fix: make awsgw start --host override actually bind The 'awsgw start --host' override printed 'Overwriting awsgw host' but the assignment was commented out, so AWSGW always bound the spinifex.toml host (a specific WAN IP, e.g. 192.168.0.10). A system instance (EKS K3s control plane, internal LB agent) has a mgmt-only NIC on-link to the br-mgmt IP with no route to the WAN IP, so its SigV4 POSTs to the gateway were refused and the cluster never bootstrapped. Honour --host (e2e passes 0.0.0.0:9999) so AWSGW listens on all interfaces incl. br-mgmt; the serving cert already SANs every local interface IP, and WAN clients still reach it. * test(eks): install kubectl on e2e VM so KubectlGetNodes runs The eks data-plane subtest (TestEKS/KubectlGetNodes) drives the full aws eks get-token -> kubectl -> apiserver -> token-webhook -> gateway broker -> STS verify path — the only end-to-end exercise of the token-review broker. The e2e binaries run on the VM, where kubectl was absent, so harness.NewKubectl t.Skip'd it and the path went unvalidated. Install a checksum-verified kubectl (pinned to the k3s server version) on the VM, gated on e2e-eks. --- .github/workflows/e2e.yml | 84 +++- cmd/eks-gateway-publish/main.go | 134 ++++++ cmd/eks-token-webhook/main.go | 184 ++++---- cmd/eks-token-webhook/main_test.go | 132 +++--- cmd/spinifex/cmd/service.go | 2 +- internal/eksgw/client.go | 103 +++++ scripts/build-system-image.sh | 317 +++++--------- .../images/eks-node/eks-token-webhook.initd | 18 +- scripts/images/eks-node/k3s-first-boot.sh | 44 +- scripts/images/eks-node/manifest.conf | 16 +- .../images/eks-node/mulga-eks-state-report.sh | 43 +- scripts/images/eks-node/setup.sh | 30 +- spinifex/daemon/eks_deps.go | 94 ++-- spinifex/gateway/eks.go | 16 + spinifex/gateway/eks/internal_publish.go | 91 ++++ spinifex/gateway/eks/internal_publish_test.go | 80 ++++ spinifex/gateway/eks/token_review.go | 54 +++ spinifex/gateway/eks/token_review_test.go | 86 ++++ spinifex/gateway/eks_test.go | 9 +- spinifex/handlers/eks/delete_cluster_test.go | 33 +- spinifex/handlers/eks/k3s_server_vm.go | 51 ++- spinifex/handlers/eks/k3s_server_vm_test.go | 24 +- spinifex/handlers/eks/service_impl.go | 24 +- spinifex/handlers/eks/token_review.go | 92 ++++ spinifex/handlers/eks/token_review_test.go | 156 +++++++ tests/e2e/Makefile | 7 +- tests/e2e/eks/eks_test.go | 404 ++++++++++++++++++ tests/e2e/harness/aws.go | 4 + tests/e2e/harness/eks.go | 73 ++++ tests/e2e/harness/eks_mgmt.go | 98 +++++ tests/e2e/harness/kubectl.go | 67 +++ 31 files changed, 1981 insertions(+), 589 deletions(-) create mode 100644 cmd/eks-gateway-publish/main.go create mode 100644 internal/eksgw/client.go create mode 100644 spinifex/gateway/eks/internal_publish.go create mode 100644 spinifex/gateway/eks/internal_publish_test.go create mode 100644 spinifex/gateway/eks/token_review.go create mode 100644 spinifex/gateway/eks/token_review_test.go create mode 100644 spinifex/handlers/eks/token_review.go create mode 100644 spinifex/handlers/eks/token_review_test.go create mode 100644 tests/e2e/eks/eks_test.go create mode 100644 tests/e2e/harness/eks.go create mode 100644 tests/e2e/harness/eks_mgmt.go create mode 100644 tests/e2e/harness/kubectl.go diff --git a/.github/workflows/e2e.yml b/.github/workflows/e2e.yml index 813cbb24d..9a9aa78ec 100644 --- a/.github/workflows/e2e.yml +++ b/.github/workflows/e2e.yml @@ -18,9 +18,9 @@ on: description: "PR head SHA to report status on" required: false suites: - description: "Space-separated Make targets to run (e.g. 'e2e-single e2e-multinode e2e-cert e2e-lb')" + description: "Space-separated Make targets to run (e.g. 'e2e-single e2e-multinode e2e-cert e2e-lb e2e-eks')" required: false - default: "e2e-single e2e-multinode e2e-cert e2e-lb" + default: "e2e-single e2e-multinode e2e-cert e2e-lb e2e-eks" test_run: description: "Optional regex for -test.run (e.g. '^TestMultinodeIPSec$'). Empty = run all tests in suite." required: false @@ -32,7 +32,7 @@ permissions: env: GIT_BRANCH: ${{ github.ref_name }} - SUITES: ${{ inputs.suites || 'e2e-single e2e-multinode e2e-cert e2e-lb' }} + SUITES: ${{ inputs.suites || 'e2e-single e2e-multinode e2e-cert e2e-lb e2e-eks' }} TEST_RUN: ${{ inputs.test_run || '' }} jobs: @@ -42,7 +42,7 @@ jobs: # suites='e2e-reboot' only). Push triggers and the default SUITES list # both include single/cert/lb so the inputs.suites null-check keeps the # job running on plain PR pushes. - if: github.event_name != 'workflow_dispatch' || contains(github.event.inputs.suites, 'e2e-single') || contains(github.event.inputs.suites, 'e2e-cert') || contains(github.event.inputs.suites, 'e2e-lb') + if: github.event_name != 'workflow_dispatch' || contains(github.event.inputs.suites, 'e2e-single') || contains(github.event.inputs.suites, 'e2e-cert') || contains(github.event.inputs.suites, 'e2e-lb') || contains(github.event.inputs.suites, 'e2e-eks') runs-on: [ self-hosted, ci-single ] timeout-minutes: 45 env: @@ -154,6 +154,82 @@ jobs: echo "wan_ip=${WAN_IP}" >> "$GITHUB_OUTPUT" echo "ssh_key=${SSH_KEY}" >> "$GITHUB_OUTPUT" + # The eks suite's CreateCluster boots a control-plane VM from the + # eks-node AMI, so that AMI must exist in the VM's image store before the + # suite runs. build-system-image.sh customizes the image with libguestfs + # (an isolated userspace appliance) — it touches no host block device, so + # it is safe on the shared runner. The build needs the spinifex source + + # network (runner only); the import needs predastore/NATS (VM only), so + # build on the runner then ship the raw and import on the VM. + - name: Build eks-node AMI (libguestfs, runner) + if: contains(env.SUITES, 'e2e-eks') + working-directory: spinifex + run: | + set -euo pipefail + source "${GITHUB_WORKSPACE}/spinifex/.github/scripts/ci-fmt.sh" + banner "build eks-node AMI (libguestfs — no host devices)" + sudo apt-get update -qq + sudo apt-get install -y --no-install-recommends libguestfs-tools qemu-utils + # Debian/Ubuntu ship /boot/vmlinuz-* mode 0600; the libguestfs + # appliance needs to read a host kernel. direct backend = no libvirt. + export LIBGUESTFS_BACKEND=direct + sudo chmod 0644 /boot/vmlinuz-* || true + group_open "make build-eks-node-image" + make build-eks-node-image + group_close + ok "raw: $(du -h /tmp/eks-node-image-build/eks-node-alpine.raw | cut -f1)" + + - name: Import eks-node AMI on VM + if: contains(env.SUITES, 'e2e-eks') + run: | + set -euo pipefail + source "${GITHUB_WORKSPACE}/spinifex/.github/scripts/ci-fmt.sh" + # Pull import args straight from the manifest so they never drift. + eval "$(grep -E '^(ALPINE_VERSION|AMI_NAME|SYSTEM_TAG)=' \ + spinifex/scripts/images/eks-node/manifest.conf)" + SSH_OPTS=(-o StrictHostKeyChecking=no -i "${{ steps.vm.outputs.ssh_key }}") + VM=tf-user@${{ steps.vm.outputs.wan_ip }} + banner "import eks-node AMI to VM image store" + scp "${SSH_OPTS[@]}" \ + /tmp/eks-node-image-build/eks-node-alpine.raw \ + "$VM":/tmp/eks-node-alpine.raw + ssh "${SSH_OPTS[@]}" "$VM" \ + "spx admin images import --file /tmp/eks-node-alpine.raw \ + --distro alpine --version ${ALPINE_VERSION} --arch x86_64 \ + --boot-mode bios --ami-name ${AMI_NAME} --tag ${SYSTEM_TAG}" + ok "eks-node AMI imported" + + # The eks data-plane subtest (TestEKS/KubectlGetNodes) drives the full + # `aws eks get-token` → kubectl → apiserver → token-webhook → gateway + # broker → STS verify path. The e2e binaries run ON the VM, so kubectl + # must be on the VM's PATH or harness.NewKubectl t.Skips the subtest. + # Pin to the k3s server version (setup.sh K3S_VERSION) — kubectl is + # ±1-minor skew tolerant; matching is simplest. Checksum-verified. + - name: Install kubectl on VM + if: contains(env.SUITES, 'e2e-eks') + run: | + set -euo pipefail + source "${GITHUB_WORKSPACE}/spinifex/.github/scripts/ci-fmt.sh" + SSH_OPTS=(-o StrictHostKeyChecking=no -i "${{ steps.vm.outputs.ssh_key }}") + VM=tf-user@${{ steps.vm.outputs.wan_ip }} + banner "install kubectl on VM (eks data-plane subtest)" + ssh "${SSH_OPTS[@]}" "$VM" 'bash -s' <<'EOF' + set -euo pipefail + KVER=v1.32.5 + case "$(uname -m)" in + x86_64) ARCH=amd64 ;; + aarch64) ARCH=arm64 ;; + *) echo "unsupported arch $(uname -m)" >&2; exit 1 ;; + esac + cd /tmp + curl -fsSLO "https://dl.k8s.io/release/${KVER}/bin/linux/${ARCH}/kubectl" + curl -fsSLO "https://dl.k8s.io/release/${KVER}/bin/linux/${ARCH}/kubectl.sha256" + echo "$(cat kubectl.sha256) kubectl" | sha256sum -c - + sudo install -m 0755 kubectl /usr/local/bin/kubectl + kubectl version --client + EOF + ok "kubectl installed on VM" + - name: Build E2E test binaries on runner working-directory: spinifex/tests/e2e run: | diff --git a/cmd/eks-gateway-publish/main.go b/cmd/eks-gateway-publish/main.go new file mode 100644 index 000000000..5cd33a7b2 --- /dev/null +++ b/cmd/eks-gateway-publish/main.go @@ -0,0 +1,134 @@ +// eks-gateway-publish runs inside the EKS K3s control-plane VM and relays a +// single publication to the host through the AWS gateway, instead of dialing +// core NATS directly. +// +// It reads a JSON payload on stdin, wraps it as +// {accountId, channel, kind, payload}, SigV4-signs (service "eks") an HTTPS +// POST to {gateway}/clusters/{cluster}/internal-publish, and retries with +// backoff until the gateway returns 2xx or the attempt budget is exhausted — +// so a degraded link surfaces as a non-zero exit rather than a silently +// dropped message (the failure mode of fire-and-forget `nats pub`). +// +// Usage: +// +// echo '{"token":"..."}' | eks-gateway-publish -channel bootstrap -kind k3s-bootstrap-token +// kubectl ... | eks-gateway-publish -channel state +// +// Flags default to environment variables seeded by cloud-init: +// EKS_GATEWAY_URL, EKS_GATEWAY_CA, EKS_ACCESS_KEY, EKS_SECRET_KEY, EKS_REGION, +// EKS_ACCOUNT_ID, EKS_CLUSTER_NAME. +package main + +import ( + "bytes" + "encoding/json" + "flag" + "fmt" + "io" + "log/slog" + "os" + "time" + + "github.com/mulgadc/spinifex/internal/eksgw" + + _ "github.com/mulgadc/spinifex/internal/fipsboot" +) + +const ( + // Retry budget: the control plane reaches readiness before this runs, so a + // failing POST means a degraded link, not a cold start. Bounded so a stuck + // boot still terminates the OpenRC service. + maxAttempts = 30 + retryDelay = 5 * time.Second +) + +type publishBody struct { + AccountID string `json:"accountId"` + Channel string `json:"channel"` + Kind string `json:"kind"` + Payload json.RawMessage `json:"payload"` +} + +func main() { + var ( + gatewayURL string + gatewayCA string + accessKey string + secretKey string + region string + accountID string + clusterName string + channel string + kind string + ) + + flag.StringVar(&gatewayURL, "gateway", os.Getenv("EKS_GATEWAY_URL"), "Gateway URL (e.g. https://10.15.8.1:9999)") + flag.StringVar(&gatewayCA, "gateway-ca", os.Getenv("EKS_GATEWAY_CA"), "Path to gateway TLS CA PEM (optional; falls back to system trust)") + flag.StringVar(&accessKey, "access-key", os.Getenv("EKS_ACCESS_KEY"), "AWS access key ID") + flag.StringVar(&secretKey, "secret-key", os.Getenv("EKS_SECRET_KEY"), "AWS secret access key") + flag.StringVar(®ion, "region", os.Getenv("EKS_REGION"), "AWS region for SigV4 signing") + flag.StringVar(&accountID, "account-id", os.Getenv("EKS_ACCOUNT_ID"), "Cluster account ID") + flag.StringVar(&clusterName, "cluster", os.Getenv("EKS_CLUSTER_NAME"), "Cluster name") + flag.StringVar(&channel, "channel", "", "Publish channel: bootstrap|state") + flag.StringVar(&kind, "kind", "", "Bootstrap subject kind (bootstrap channel only)") + flag.Parse() + + switch { + case accountID == "": + fatal("--account-id is required (or set EKS_ACCOUNT_ID)") + case clusterName == "": + fatal("--cluster is required (or set EKS_CLUSTER_NAME)") + case channel != "bootstrap" && channel != "state": + fatal("--channel must be bootstrap or state") + case channel == "bootstrap" && kind == "": + fatal("--kind is required for the bootstrap channel") + } + + payload, err := io.ReadAll(os.Stdin) + if err != nil { + fatal(fmt.Sprintf("read stdin payload: %v", err)) + } + if len(bytes.TrimSpace(payload)) == 0 { + fatal("empty stdin payload") + } + if !json.Valid(payload) { + fatal("stdin payload is not valid JSON") + } + + body, err := json.Marshal(publishBody{ + AccountID: accountID, + Channel: channel, + Kind: kind, + Payload: json.RawMessage(payload), + }) + if err != nil { + fatal(fmt.Sprintf("marshal request body: %v", err)) + } + + client, err := eksgw.New(gatewayURL, gatewayCA, accessKey, secretKey, region) + if err != nil { + fatal(fmt.Sprintf("build gateway client: %v", err)) + } + path := "/clusters/" + clusterName + "/internal-publish" + + var lastErr error + for attempt := 1; attempt <= maxAttempts; attempt++ { + if _, err := client.Post(path, body); err != nil { + lastErr = err + slog.Warn("eks-gateway-publish: attempt failed", + "channel", channel, "kind", kind, "attempt", attempt, "err", err) + if attempt < maxAttempts { + time.Sleep(retryDelay) + } + continue + } + slog.Info("eks-gateway-publish: published", "channel", channel, "kind", kind) + return + } + fatal(fmt.Sprintf("publish failed after %d attempts: %v", maxAttempts, lastErr)) +} + +func fatal(msg string) { + slog.Error("eks-gateway-publish: " + msg) + os.Exit(1) +} diff --git a/cmd/eks-token-webhook/main.go b/cmd/eks-token-webhook/main.go index b06a396e9..a314a2051 100644 --- a/cmd/eks-token-webhook/main.go +++ b/cmd/eks-token-webhook/main.go @@ -1,8 +1,10 @@ // 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`, 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. +// the eks-server AMI. Per kubectl call the kube-apiserver POSTs the bearer +// token (produced by `aws eks get-token`); the webhook relays it through the +// AWS gateway broker (SigV4 HTTPS POST to /clusters/{name}/token-review), which +// host-side runs the STS verify (cross-cluster x-k8s-aws-id pin) + AccessEntry +// KV lookup and returns the resolved identity. The webhook never speaks core +// NATS — same gateway-broker model as ELBv2's lb-agent and eks-gateway-publish. // // 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 @@ -20,46 +22,48 @@ import ( "os" "time" + "github.com/mulgadc/spinifex/internal/eksgw" 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 + gatewayURL string + gatewayCA string + accessKey string + secretKey string + region 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"), + gatewayURL: os.Getenv("EKS_GATEWAY_URL"), + gatewayCA: os.Getenv("EKS_GATEWAY_CA"), + accessKey: os.Getenv("EKS_ACCESS_KEY"), + secretKey: os.Getenv("EKS_SECRET_KEY"), + region: os.Getenv("EKS_REGION"), 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 == "" { + switch { + case c.gatewayURL == "": + return c, errors.New("EKS_GATEWAY_URL not set") + case c.accessKey == "" || c.secretKey == "": + return c, errors.New("EKS_ACCESS_KEY / EKS_SECRET_KEY not set") + case c.accountID == "": return c, errors.New("EKS_ACCOUNT_ID not set") - } - if c.clusterName == "" { + case c.clusterName == "": return c, errors.New("EKS_CLUSTER_NAME not set") } return c, nil @@ -103,22 +107,15 @@ func run(cfg config) error { return fmt.Errorf("write apiserver kubeconfig: %w", err) } - nc, err := utils.ConnectNATSWithRetry(cfg.natsURL, cfg.natsToken, cfg.natsCA) + client, err := eksgw.New(cfg.gatewayURL, cfg.gatewayCA, cfg.accessKey, cfg.secretKey, cfg.region) 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) + return fmt.Errorf("build gateway client: %w", err) } authr := &authenticator{ - nc: nc, - kv: kv, + accountID: cfg.accountID, clusterName: cfg.clusterName, - verifyTO: cfg.verifyTO, + review: gatewayReviewer(client, cfg.clusterName, cfg.accountID), } mux := http.NewServeMux() @@ -136,50 +133,42 @@ func run(cfg config) error { 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 +// tokenReviewRequest is the body POSTed to /clusters/{name}/token-review. The +// webhook holds system SigV4 creds (system account, not the cluster account), +// so accountId names the cluster account explicitly — same as PublishInternal. +type tokenReviewRequest struct { + AccountID string `json:"accountId"` + Token string `json:"token"` +} + +// gatewayReviewer returns the per-call relay: SigV4-POST {accountId,token} to +// the gateway and decode the resolved identity. A non-2xx / transport error is +// surfaced (handle() turns it into a 5xx so the apiserver retries) rather than a +// silent deny, distinguishing "webhook broker down" from "token rejected". +func gatewayReviewer(client *eksgw.Client, clusterName, accountID string) func(string) (handlers_eks.WebhookTokenReviewResult, error) { + path := "/clusters/" + clusterName + "/token-review" + return func(token string) (handlers_eks.WebhookTokenReviewResult, error) { + body, err := json.Marshal(tokenReviewRequest{AccountID: accountID, Token: token}) + if err != nil { + return handlers_eks.WebhookTokenReviewResult{}, fmt.Errorf("marshal request: %w", err) } - lastErr = kvErr - time.Sleep(2 * time.Second) + respBody, err := client.Post(path, body) + if err != nil { + return handlers_eks.WebhookTokenReviewResult{}, err + } + var res handlers_eks.WebhookTokenReviewResult + if err := json.Unmarshal(respBody, &res); err != nil { + return handlers_eks.WebhookTokenReviewResult{}, fmt.Errorf("decode response: %w", err) + } + return res, nil } - return nil, fmt.Errorf("kv bucket %s not available: %w", bucket, lastErr) } -// authenticator resolves a TokenReview against STS (over NATS) + AccessEntry KV. +// authenticator resolves a TokenReview by relaying to the gateway broker. type authenticator struct { - nc *nats.Conn - kv nats.KeyValue + accountID string 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) + review func(token string) (handlers_eks.WebhookTokenReviewResult, error) } func (a *authenticator) handle(w http.ResponseWriter, r *http.Request) { @@ -193,8 +182,22 @@ func (a *authenticator) handle(w http.ResponseWriter, r *http.Request) { return } - status := authenticate(review.Spec.Token, a.verify, a.lookup) - if status.Authenticated { + res, err := a.review(review.Spec.Token) + if err != nil { + // Broker fault (gateway down, transport error): a 5xx tells the apiserver + // to retry rather than treating a transient outage as a hard deny. + slog.Error("TokenReview relay failed", "cluster", a.clusterName, "err", err) + http.Error(w, "token review unavailable", http.StatusServiceUnavailable) + return + } + + status := tokenReviewStatus{Authenticated: res.Authenticated} + if res.Authenticated { + status.User = userInfo{ + Username: res.Username, + UID: res.UID, + Groups: res.Groups, + } // Confirm the token for the audiences the apiserver asked about; an empty // status.audiences is rejected when --api-audiences is configured. status.Audiences = review.Spec.Audiences @@ -212,45 +215,6 @@ func (a *authenticator) handle(w http.ResponseWriter, r *http.Request) { } } -// 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 { diff --git a/cmd/eks-token-webhook/main_test.go b/cmd/eks-token-webhook/main_test.go index 8b35e5e64..889ae0832 100644 --- a/cmd/eks-token-webhook/main_test.go +++ b/cmd/eks-token-webhook/main_test.go @@ -2,9 +2,13 @@ package main import ( "encoding/base64" + "encoding/json" "errors" + "net/http" + "net/http/httptest" "os" "path/filepath" + "strings" "testing" handlers_eks "github.com/mulgadc/spinifex/spinifex/handlers/eks" @@ -12,83 +16,79 @@ import ( "github.com/stretchr/testify/require" ) -func validToken(url string) string { - return "k8s-aws-v1." + base64.RawURLEncoding.EncodeToString([]byte(url)) +// postReview drives authenticator.handle with a TokenReview carrying token and +// audiences, returning the decoded response status and HTTP code. +func postReview(t *testing.T, a *authenticator, token string, audiences []string) (tokenReviewStatus, int) { + t.Helper() + body, err := json.Marshal(tokenReview{ + APIVersion: "authentication.k8s.io/v1", + Kind: "TokenReview", + Spec: tokenReviewSpec{Token: token, Audiences: audiences}, + }) + require.NoError(t, err) + + req := httptest.NewRequest(http.MethodPost, "/authenticate", strings.NewReader(string(body))) + w := httptest.NewRecorder() + a.handle(w, req) + + if w.Code != http.StatusOK { + return tokenReviewStatus{}, w.Code + } + var resp tokenReview + require.NoError(t, json.Unmarshal(w.Body.Bytes(), &resp)) + return resp.Status, w.Code } -const testARN = "arn:aws:iam::111122223333:role/admin" +func TestHandle_GrantsAndStampsAudiences(t *testing.T) { + a := &authenticator{ + accountID: "111122223333", + clusterName: "alpha", + review: func(string) (handlers_eks.WebhookTokenReviewResult, error) { + return handlers_eks.WebhookTokenReviewResult{ + Authenticated: true, + Username: "arn:aws:iam::111122223333:role/admin", + UID: "AROAEXAMPLE:session", + Groups: []string{"system:masters"}, + }, nil + }, + } -func okVerify(_ string) (*handlers_eks.TokenVerifyResponse, error) { - return &handlers_eks.TokenVerifyResponse{ - AccountID: "111122223333", - ARN: testARN, - UserID: "AROAEXAMPLE:session", - PrincipalType: "AssumedRole", - }, nil + status, code := postReview(t, a, "tok", []string{"https://kubernetes.default.svc"}) + require.Equal(t, http.StatusOK, code) + require.True(t, status.Authenticated) + assert.Equal(t, "arn:aws:iam::111122223333:role/admin", status.User.Username) + assert.Equal(t, "AROAEXAMPLE:session", status.User.UID) + assert.Equal(t, []string{"system:masters"}, status.User.Groups) + assert.Equal(t, []string{"https://kubernetes.default.svc"}, status.Audiences) } -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 +func TestHandle_DeniesWithoutStampingAudiences(t *testing.T) { + a := &authenticator{ + clusterName: "alpha", + review: func(string) (handlers_eks.WebhookTokenReviewResult, error) { + return handlers_eks.WebhookTokenReviewResult{Authenticated: false}, 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) + status, code := postReview(t, a, "tok", []string{"https://kubernetes.default.svc"}) + require.Equal(t, http.StatusOK, code) + assert.False(t, status.Authenticated) + assert.Empty(t, status.User.Username) + assert.Empty(t, status.Audiences, "unauthenticated review must not stamp audiences") } -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 +// A broker fault is a 5xx (retryable), not a silent deny — a transient gateway +// outage must not look identical to a rejected token. +func TestHandle_BrokerFaultReturns5xx(t *testing.T) { + a := &authenticator{ + clusterName: "alpha", + review: func(string) (handlers_eks.WebhookTokenReviewResult, error) { + return handlers_eks.WebhookTokenReviewResult{}, errors.New("gateway returned 503") + }, } - 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) + _, code := postReview(t, a, "tok", nil) + assert.Equal(t, http.StatusServiceUnavailable, code) } func TestEnsureServingCert_PersistsAndReuses(t *testing.T) { diff --git a/cmd/spinifex/cmd/service.go b/cmd/spinifex/cmd/service.go index 8cdb0d879..10724e8f9 100644 --- a/cmd/spinifex/cmd/service.go +++ b/cmd/spinifex/cmd/service.go @@ -538,7 +538,7 @@ var awsgwStartCmd = &cobra.Command{ awsgwHost := viper.GetString("host") if awsgwHost != "" { fmt.Println("Overwriting awsgw host to:", awsgwHost) - //nodeConfig.AWSGW.Host = awsgwHost + nodeConfig.AWSGW.Host = awsgwHost } awsgwTlsCert := viper.GetString("tls-cert") diff --git a/internal/eksgw/client.go b/internal/eksgw/client.go new file mode 100644 index 000000000..dd7441067 --- /dev/null +++ b/internal/eksgw/client.go @@ -0,0 +1,103 @@ +// Package eksgw is the on-VM SigV4 client EKS control-plane helpers use to +// reach the AWS gateway, instead of dialing core NATS. Both eks-gateway-publish +// (bootstrap + state relay) and eks-token-webhook (TokenReview relay) sign with +// the system (Predastore) credentials for service "eks" and POST over HTTPS; +// the gateway authorizes via IAM and relays to the host-side NATS/STS/KV. +package eksgw + +import ( + "bytes" + "crypto/sha256" + "crypto/tls" + "crypto/x509" + "encoding/hex" + "fmt" + "io" + "net/http" + "os" + "strings" + "time" + + "github.com/mulgadc/predastore/auth" + "github.com/mulgadc/spinifex/internal/tlsconfig" +) + +// Client posts SigV4-signed requests to the AWS gateway for the "eks" service. +type Client struct { + baseURL string + accessKey string + secretKey string + region string + httpClient *http.Client +} + +// New builds a client. caPath optionally pins the gateway TLS CA; when empty the +// client relies on the system trust store. region defaults to us-east-1 when +// empty (SigV4 requires a non-empty region). +func New(baseURL, caPath, accessKey, secretKey, region string) (*Client, error) { + if baseURL == "" { + return nil, fmt.Errorf("eksgw: baseURL is required") + } + if accessKey == "" || secretKey == "" { + return nil, fmt.Errorf("eksgw: access key and secret key are required") + } + if region == "" { + region = "us-east-1" + } + + tlsCfg := &tls.Config{ + MinVersion: tls.VersionTLS13, + CurvePreferences: tlsconfig.Curves, + } + if caPath != "" { + pem, err := os.ReadFile(caPath) + if err != nil { + return nil, fmt.Errorf("eksgw: read gateway CA %q: %w", caPath, err) + } + pool := x509.NewCertPool() + if !pool.AppendCertsFromPEM(pem) { + return nil, fmt.Errorf("eksgw: gateway CA %q has no usable certificates", caPath) + } + tlsCfg.RootCAs = pool + } + + return &Client{ + baseURL: strings.TrimRight(baseURL, "/"), + accessKey: accessKey, + secretKey: secretKey, + region: region, + httpClient: &http.Client{ + Timeout: 10 * time.Second, + Transport: &http.Transport{TLSClientConfig: tlsCfg, MaxIdleConns: 2, IdleConnTimeout: 30 * time.Second}, + }, + }, nil +} + +// Post SigV4-signs and sends a single POST of body to path (e.g. +// "/clusters/alpha/token-review"). It returns the response body on a 2xx, or an +// error (including the gateway status + body) otherwise. No retry — callers that +// need it wrap Post in their own loop. +func (c *Client) Post(path string, body []byte) ([]byte, error) { + req, err := http.NewRequest(http.MethodPost, c.baseURL+path, bytes.NewReader(body)) + if err != nil { + return nil, fmt.Errorf("create request: %w", err) + } + req.Header.Set("Content-Type", "application/x-amz-json-1.1") + + sum := sha256.Sum256(body) + if err := auth.SignReq(req, c.accessKey, c.secretKey, hex.EncodeToString(sum[:]), "eks", c.region); err != nil { + return nil, fmt.Errorf("sign request: %w", err) + } + + resp, err := c.httpClient.Do(req) + if err != nil { + return nil, fmt.Errorf("send request: %w", err) + } + defer resp.Body.Close() + + respBody, _ := io.ReadAll(resp.Body) + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return nil, fmt.Errorf("gateway returned %d: %s", resp.StatusCode, strings.TrimSpace(string(respBody))) + } + return respBody, nil +} diff --git a/scripts/build-system-image.sh b/scripts/build-system-image.sh index e8c56d6fc..6da9c732d 100755 --- a/scripts/build-system-image.sh +++ b/scripts/build-system-image.sh @@ -3,17 +3,28 @@ set -euo pipefail # build-system-image.sh — Build a minimal system AMI from a manifest # -# Supports Alpine Linux and Ubuntu 24.04 as base distros. -# Creates a pre-baked image with custom packages, binaries, and setup scripts -# installed, ready for import as a Spinifex AMI. +# Supports Alpine Linux and Ubuntu 24.04+ as base distros. Creates a pre-baked +# image with custom packages, binaries, and setup scripts installed, ready for +# import as a Spinifex AMI. # -# Requirements: qemu-nbd, qemu-img, sudo (for mount/chroot), curl +# Customization runs entirely inside the libguestfs appliance (its own kernel + +# userspace), so the build touches NO host block device, NO host mount, and +# needs NO sudo. This deliberately replaces the previous qemu-nbd + `sudo mount` +# + chroot + /dev bind-mount flow, whose leaked binds could be traversed by a +# later `rm -rf` and wipe the host's /dev (it wedged CI runners). libguestfs +# isolates all of that in a throwaway VM. +# +# Requirements: virt-customize, guestfish, virt-resize (libguestfs-tools), +# qemu-img, curl # Usage: ./scripts/build-system-image.sh [--import] # --import Also import the image as an AMI via spx admin images import SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)" PROJECT_DIR="$(cd "$SCRIPT_DIR/.." && pwd)" +# No libvirt on CI runners — drive qemu directly. Harmless when libvirt exists. +export LIBGUESTFS_BACKEND="${LIBGUESTFS_BACKEND:-direct}" + usage() { echo "Usage: $0 [--import]" echo "" @@ -60,8 +71,6 @@ done # Derived paths BUILD_DIR="/tmp/${IMAGE_NAME}-image-build" -NBD_DEV="/dev/nbd0" -MOUNT_DIR="${BUILD_DIR}/mnt" case "$DISTRO" in alpine) @@ -74,7 +83,6 @@ case "$DISTRO" in OUTPUT_IMAGE="${BUILD_DIR}/${IMAGE_NAME}-alpine.qcow2" OUTPUT_RAW="${BUILD_DIR}/${IMAGE_NAME}-alpine.raw" DISTRO_VERSION="${ALPINE_VERSION}" - ROOT_PART="${NBD_DEV}" BOOT_MODE="bios" ;; ubuntu) @@ -87,7 +95,6 @@ case "$DISTRO" in OUTPUT_IMAGE="${BUILD_DIR}/${IMAGE_NAME}-ubuntu.qcow2" OUTPUT_RAW="${BUILD_DIR}/${IMAGE_NAME}-ubuntu.raw" DISTRO_VERSION="${UBUNTU_VERSION}" - ROOT_PART="${NBD_DEV}p1" BOOT_MODE="uefi" ;; *) @@ -96,20 +103,20 @@ case "$DISTRO" in ;; esac -cleanup() { - echo "Cleaning up..." - # Recursive lazy unmount detaches the chroot binds (/dev, /dev/pts, /sys, - # /proc) and the root mount from the path hierarchy at once, even if a - # straggler still holds one busy. This MUST succeed: a leaked /dev bind left - # under the build dir would be traversed by a later `rm -rf` and wipe the - # host's /dev (deletes /dev/null, wedges the box). -R recurses children, - # -l detaches busy mounts immediately so the path no longer crosses /dev. - sudo umount -R -l "${MOUNT_DIR}" 2>/dev/null || true - sudo qemu-nbd --disconnect "${NBD_DEV}" 2>/dev/null || true - exec 9>&- 2>/dev/null || true # release nbd lock if held - echo "Done." +# import_ami registers the built raw image as a Spinifex AMI. Shared by the +# fresh-build and cached-build paths. +import_ami() { + echo "Importing as AMI..." + rm -f "$OUTPUT_IMAGE" + local args=(--file "$OUTPUT_RAW" --distro "${DISTRO}" --version "${DISTRO_VERSION}" --arch x86_64 --boot-mode "${BOOT_MODE}") + if [[ -n "${AMI_NAME:-}" ]]; then + args+=(--ami-name "$AMI_NAME") + fi + if [[ -n "${SYSTEM_TAG:-}" ]]; then + args+=(--tag "$SYSTEM_TAG") + fi + spx admin images import "${args[@]}" } -trap cleanup EXIT # In quiet mode, redirect build output to /dev/null (import output still shown) if [[ "$QUIET" == true ]]; then @@ -117,7 +124,7 @@ if [[ "$QUIET" == true ]]; then exec 1>/dev/null # suppress build output fi -echo "=== System Image Builder ===" +echo "=== System Image Builder (libguestfs) ===" echo "Image: ${IMAGE_NAME} — ${IMAGE_DESCRIPTION:-}" echo "Distro: ${DISTRO} ${DISTRO_VERSION}" echo "Size: ${IMAGE_SIZE}" @@ -125,18 +132,18 @@ echo "Build: ${BUILD_DIR}" echo "" # Step 0: Check prerequisites -if ! command -v qemu-nbd &>/dev/null; then - echo "ERROR: qemu-nbd not found. Install qemu-utils." +for tool in virt-customize qemu-img; do + if ! command -v "$tool" &>/dev/null; then + echo "ERROR: $tool not found. Install libguestfs-tools + qemu-utils." + exit 1 + fi +done +if [[ "$DISTRO" == "alpine" ]] && ! command -v guestfish &>/dev/null; then + echo "ERROR: guestfish not found. Install libguestfs-tools." exit 1 fi - -if ! command -v qemu-img &>/dev/null; then - echo "ERROR: qemu-img not found. Install qemu-utils." - exit 1 -fi - -if [[ "$DISTRO" == "ubuntu" ]] && ! command -v parted &>/dev/null && ! [[ -x /usr/sbin/parted ]]; then - echo "ERROR: parted not found. Install parted." +if [[ "$DISTRO" == "ubuntu" ]] && ! command -v virt-resize &>/dev/null; then + echo "ERROR: virt-resize not found. Install libguestfs-tools." exit 1 fi @@ -167,11 +174,11 @@ if [[ -n "${INSTALL_BINARIES:-}" ]]; then done fi -# Serialize the entire image build with flock — concurrent builds on the same -# host (e.g. CI single-node + multi-node jobs) share /dev/nbd0 and BUILD_DIR. -NBD_LOCK="/tmp/build-system-image.lock" +# Serialize the build with flock — concurrent builds on the same host (e.g. CI +# single-node + multi-node jobs) share BUILD_DIR + the cached raw image. +BUILD_LOCK="/tmp/build-system-image.lock" echo "Acquiring build lock..." -exec 9>"$NBD_LOCK" +exec 9>"$BUILD_LOCK" flock 9 echo "Lock acquired" @@ -190,21 +197,12 @@ if [[ -f "$OUTPUT_RAW" ]] && [[ $(( $(date +%s) - $(stat -c %Y "$OUTPUT_RAW") )) echo " raw: $OUTPUT_RAW ($(du -h "$OUTPUT_RAW" | cut -f1))" if [[ "$DO_IMPORT" == true ]]; then - echo "Importing as AMI..." - rm -f "$OUTPUT_IMAGE" - IMPORT_ARGS=(--file "$OUTPUT_RAW" --distro "${DISTRO}" --version "${DISTRO_VERSION}" --arch x86_64 --boot-mode "${BOOT_MODE}") - if [[ -n "${AMI_NAME:-}" ]]; then - IMPORT_ARGS+=(--ami-name "$AMI_NAME") - fi - if [[ -n "${SYSTEM_TAG:-}" ]]; then - IMPORT_ARGS+=(--tag "$SYSTEM_TAG") - fi - spx admin images import "${IMPORT_ARGS[@]}" + import_ami fi exit 0 fi -mkdir -p "$BUILD_DIR" "$MOUNT_DIR" +mkdir -p "$BUILD_DIR" # Step 1: Download base cloud image if [[ -f "${BUILD_DIR}/${SOURCE_IMAGE}" ]]; then @@ -229,130 +227,66 @@ rm -f "$OUTPUT_IMAGE" echo "Copying image for customization..." cp "${BUILD_DIR}/${SOURCE_IMAGE}" "$OUTPUT_IMAGE" -# Resize the image to provide room for packages -qemu-img resize "$OUTPUT_IMAGE" "$IMAGE_SIZE" - -# Step 3: Connect via qemu-nbd -echo "Connecting image via qemu-nbd..." -sudo modprobe nbd max_part=4 2>/dev/null || true -if [[ ! -e "${NBD_DEV}" ]]; then - echo "ERROR: ${NBD_DEV} does not exist. Is the nbd kernel module loaded? Try: sudo modprobe nbd" - exit 1 +# Step 3: Resize to provide room for packages. Both paths grow the root +# filesystem entirely inside the libguestfs appliance — no host loop/nbd device. +echo "Resizing image to ${IMAGE_SIZE}..." +if [[ "$DISTRO" == "alpine" ]]; then + # Alpine cloud images have ext4 directly on the whole disk (no partition + # table); grow the disk file then expand the fs from inside the appliance. + qemu-img resize "$OUTPUT_IMAGE" "$IMAGE_SIZE" + guestfish --rw -a "$OUTPUT_IMAGE" run \ + : e2fsck /dev/sda forceall:true \ + : resize2fs /dev/sda +else + # Ubuntu images use a GPT partition table with the root fs on /dev/sda1; + # virt-resize expands the partition + fs into a fresh IMAGE_SIZE disk. + RESIZED="${OUTPUT_IMAGE}.resized" + rm -f "$RESIZED" + qemu-img create -f qcow2 "$RESIZED" "$IMAGE_SIZE" + virt-resize --expand /dev/sda1 "$OUTPUT_IMAGE" "$RESIZED" + mv -f "$RESIZED" "$OUTPUT_IMAGE" fi -sudo qemu-nbd --disconnect "${NBD_DEV}" 2>/dev/null || true -sudo qemu-nbd --connect="${NBD_DEV}" "$OUTPUT_IMAGE" -# Wait for the nbd device to be ready -for i in $(seq 1 10); do - if sudo blockdev --getsize64 "${NBD_DEV}" &>/dev/null; then - break - fi - sleep 1 -done -sleep 1 - -# Ubuntu images have a partition table; wait for partition devices then resize -# the partition to fill the expanded image before resizing the filesystem. +# Step 4: Assemble the virt-customize operation list. virt-customize runs the +# operations in the order they appear on the command line, so this mirrors the +# old chroot sequence: packages → binaries → files → setup → services → clean. +CUST=(virt-customize -a "$OUTPUT_IMAGE" --network) if [[ "$DISTRO" == "ubuntu" ]]; then - for i in $(seq 1 10); do - if [[ -e "${ROOT_PART}" ]]; then break; fi - sleep 1 - done - echo "Resizing partition..." - # After qemu-img resize the GPT backup header sits at the old end of disk. - # Relocate it to the new end (sfdisk --relocate is util-linux built-in, - # no extra packages needed), then re-read the partition table before parted - # runs — without this, parted sees stale GPT geometry and fails with - # "Unable to satisfy all constraints on the partition." - sudo sgdisk --move-second-header "${NBD_DEV}" - sudo partprobe "${NBD_DEV}" 2>/dev/null || true - sleep 1 - sudo parted --script "${NBD_DEV}" resizepart 1 100% + # DKMS driver bakes (GPU images) need real resources in the appliance. + CUST+=(--memsize 6144 --smp 4) +else + CUST+=(--memsize 2048) fi -# Alpine cloud images have ext4 directly on the block device (no partition table). -echo "Checking filesystem..." -sudo e2fsck -f -y "${ROOT_PART}" || { - ec=$? - if [[ $ec -gt 1 ]]; then - echo "ERROR: e2fsck failed with exit code $ec on ${ROOT_PART}" - exit 1 - fi -} - -echo "Resizing filesystem..." -if ! sudo resize2fs "${ROOT_PART}"; then - echo "ERROR: resize2fs failed on ${ROOT_PART}" - exit 1 -fi - -# Step 4: Mount and customize -echo "Mounting root filesystem..." -sudo mount "${ROOT_PART}" "$MOUNT_DIR" -# Mark the build-dir subtree private so the chroot's /dev bind below cannot -# propagate into the host mount namespace (a propagated bind is what turned a -# leaked mount into a host /dev wipe on the CI runners). -sudo mount --make-rprivate "$MOUNT_DIR" 2>/dev/null || true - -# Set up resolv.conf for DNS inside chroot. -# Ubuntu cloud images symlink /etc/resolv.conf → /run/systemd/resolve/stub-resolv.conf -# which doesn't exist yet. Remove the symlink and write a real file. -sudo rm -f "${MOUNT_DIR}/etc/resolv.conf" -sudo cp /etc/resolv.conf "${MOUNT_DIR}/etc/resolv.conf" - -# Ubuntu chroot requires /proc /sys /dev bind mounts for systemd and DKMS -if [[ "$DISTRO" == "ubuntu" ]]; then - sudo mount --bind /proc "${MOUNT_DIR}/proc" - sudo mount --bind /sys "${MOUNT_DIR}/sys" - sudo mount --bind /dev "${MOUNT_DIR}/dev" - sudo mount --bind /dev/pts "${MOUNT_DIR}/dev/pts" -fi - -# Install packages +# Packages if [[ "$DISTRO" == "alpine" ]] && [[ -n "${APK_PACKAGES:-}" ]]; then - echo "Installing packages: ${APK_PACKAGES}..." - sudo chroot "$MOUNT_DIR" /bin/sh -c " -set -e -# Enable community repo -sed -i 's|^#\(.*community\)|\1|' /etc/apk/repositories 2>/dev/null || true - -# Ensure community repo is present -if ! grep -q community /etc/apk/repositories; then - MIRROR=\$(grep main /etc/apk/repositories | head -1 | sed 's|/main|/community|') - echo \"\$MIRROR\" >> /etc/apk/repositories -fi - -apk update -apk add ${APK_PACKAGES} -" + echo "Will install packages: ${APK_PACKAGES}" + CUST+=(--run-command 'sed -i "s|^#\(.*community\)|\1|" /etc/apk/repositories 2>/dev/null || true') + CUST+=(--run-command 'grep -q community /etc/apk/repositories || grep main /etc/apk/repositories | head -1 | sed "s|/main|/community|" >> /etc/apk/repositories') + CUST+=(--run-command 'apk update') + CUST+=(--run-command "apk add ${APK_PACKAGES}") elif [[ "$DISTRO" == "ubuntu" ]] && [[ -n "${APT_PACKAGES:-}" ]]; then - echo "Installing packages: ${APT_PACKAGES}..." - sudo chroot "$MOUNT_DIR" /bin/bash -c " -export DEBIAN_FRONTEND=noninteractive -apt-get update -apt-get install -y --no-install-recommends ${APT_PACKAGES} -" + echo "Will install packages: ${APT_PACKAGES}" + CUST+=(--run-command 'export DEBIAN_FRONTEND=noninteractive; apt-get update') + CUST+=(--run-command "export DEBIAN_FRONTEND=noninteractive; apt-get install -y --no-install-recommends ${APT_PACKAGES}") fi -# Step 5: Copy binaries into the image (before setup script, which may reference them) +# Binaries (mode 0755). --mkdir is idempotent (mkdir -p semantics). if [[ -n "${INSTALL_BINARIES:-}" ]]; then - echo "Installing binaries..." IFS=' ' read -ra BINARY_PAIRS <<< "$INSTALL_BINARIES" for pair in "${BINARY_PAIRS[@]}"; do src="${pair%%:*}" dst="${pair#*:}" src_path="${PROJECT_DIR}/${src}" - echo " ${src} -> ${dst}" - sudo cp "$src_path" "${MOUNT_DIR}${dst}" - sudo chmod 755 "${MOUNT_DIR}${dst}" + CUST+=(--mkdir "$(dirname "$dst")") + CUST+=(--upload "${src_path}:${dst}") + CUST+=(--chmod "0755:${dst}") done fi -# Copy auxiliary files (systemd units, OpenRC initd scripts, cron entries, configs). -# Same src:dst pair grammar as INSTALL_BINARIES; mode 0644, parent dirs auto-created. -# Optional — manifests without INSTALL_FILES (e.g. existing GPU images) skip this step. +# Auxiliary files (systemd units, OpenRC initd scripts, cron entries, configs). +# Mode 0644; setup scripts may chmod role-specific files to 0755 afterwards. if [[ -n "${INSTALL_FILES:-}" ]]; then - echo "Installing files..." IFS=' ' read -ra FILE_PAIRS <<< "$INSTALL_FILES" for pair in "${FILE_PAIRS[@]}"; do src="${pair%%:*}" @@ -362,85 +296,45 @@ if [[ -n "${INSTALL_FILES:-}" ]]; then echo "ERROR: INSTALL_FILES source not found: $src_path" exit 1 fi - echo " ${src} -> ${dst}" - sudo mkdir -p "${MOUNT_DIR}$(dirname "$dst")" - sudo cp "$src_path" "${MOUNT_DIR}${dst}" - sudo chmod 0644 "${MOUNT_DIR}${dst}" + CUST+=(--mkdir "$(dirname "$dst")") + CUST+=(--upload "${src_path}:${dst}") + CUST+=(--chmod "0644:${dst}") done fi -# Run custom setup script inside chroot (after binaries are installed) +# Custom setup script (uploaded + executed + removed by --run, inside the guest). if [[ -n "${SETUP_SCRIPT:-}" ]]; then setup_path="${PROJECT_DIR}/${SETUP_SCRIPT}" if [[ ! -f "$setup_path" ]]; then echo "ERROR: Setup script not found: $setup_path" exit 1 fi - echo "Running setup script: ${SETUP_SCRIPT}..." - sudo cp "$setup_path" "${MOUNT_DIR}/tmp/setup.sh" - sudo chmod 755 "${MOUNT_DIR}/tmp/setup.sh" - if [[ "$DISTRO" == "ubuntu" ]]; then - if [[ ! -x "${MOUNT_DIR}/bin/bash" ]]; then - echo "ERROR: /bin/bash not found or not executable in chroot" - exit 1 - fi - sudo chroot "$MOUNT_DIR" /bin/bash -x /tmp/setup.sh - else - sudo chroot "$MOUNT_DIR" /tmp/setup.sh - fi - sudo rm -f "${MOUNT_DIR}/tmp/setup.sh" + CUST+=(--run "$setup_path") fi -# Enable services (runs after INSTALL_FILES + SETUP_SCRIPT so init scripts the -# manifest references are already on disk). +# Enable services (after files + setup so referenced init scripts exist). if [[ -n "${ENABLE_SERVICES:-}" ]]; then - echo "Enabling services: ${ENABLE_SERVICES}..." IFS=' ' read -ra SERVICES <<< "$ENABLE_SERVICES" for svc in "${SERVICES[@]}"; do if [[ "$DISTRO" == "alpine" ]]; then - if ! sudo chroot "$MOUNT_DIR" /bin/sh -c "rc-update add ${svc} default"; then - echo "ERROR: Failed to enable service '${svc}' — does it exist in the image?" - exit 1 - fi + CUST+=(--run-command "rc-update add ${svc} default") else - if ! sudo chroot "$MOUNT_DIR" /bin/bash -c "systemctl enable ${svc}"; then - echo "ERROR: Failed to enable service '${svc}'" - exit 1 - fi + CUST+=(--run-command "systemctl enable ${svc}") fi done fi -# Step 6: Clean up and unmount -echo "Cleaning up image..." +# Final cleanup of package caches + tmp. if [[ "$DISTRO" == "alpine" ]]; then - sudo chroot "$MOUNT_DIR" /bin/sh -c ' -apk cache clean 2>/dev/null || true -rm -rf /var/cache/apk/* /tmp/* -' + CUST+=(--run-command 'apk cache clean 2>/dev/null || true; rm -rf /var/cache/apk/* /tmp/* 2>/dev/null || true') else - sudo chroot "$MOUNT_DIR" /bin/bash -c ' -export DEBIAN_FRONTEND=noninteractive -apt-get clean -rm -rf /var/lib/apt/lists/* /tmp/* -' + CUST+=(--run-command 'export DEBIAN_FRONTEND=noninteractive; apt-get clean; rm -rf /var/lib/apt/lists/* /tmp/* 2>/dev/null || true') fi -# Restore the systemd-resolved symlink (Ubuntu default); cloud-init sets DNS on boot. -sudo rm -f "${MOUNT_DIR}/etc/resolv.conf" -sudo ln -sf /run/systemd/resolve/stub-resolv.conf "${MOUNT_DIR}/etc/resolv.conf" +echo "Customizing image (libguestfs appliance)..." +"${CUST[@]}" -echo "Unmounting..." -if [[ "$DISTRO" == "ubuntu" ]]; then - sudo umount "${MOUNT_DIR}/dev/pts" - sudo umount "${MOUNT_DIR}/dev" - sudo umount "${MOUNT_DIR}/sys" - sudo umount "${MOUNT_DIR}/proc" -fi -sudo umount "$MOUNT_DIR" -sudo qemu-nbd --disconnect "${NBD_DEV}" - -# Step 7: Convert to raw for import +# Step 5: Convert to raw for import echo "Converting to raw format..." qemu-img convert -f qcow2 -O raw "$OUTPUT_IMAGE" "$OUTPUT_RAW" @@ -457,16 +351,7 @@ echo " raw: $OUTPUT_RAW ($(du -h "$OUTPUT_RAW" | cut -f1))" echo "" if [[ "$DO_IMPORT" == true ]]; then - echo "Importing as AMI..." - rm -f "$OUTPUT_IMAGE" - IMPORT_ARGS=(--file "$OUTPUT_RAW" --distro "${DISTRO}" --version "${DISTRO_VERSION}" --arch x86_64 --boot-mode "${BOOT_MODE}") - if [[ -n "${AMI_NAME:-}" ]]; then - IMPORT_ARGS+=(--ami-name "$AMI_NAME") - fi - if [[ -n "${SYSTEM_TAG:-}" ]]; then - IMPORT_ARGS+=(--tag "$SYSTEM_TAG") - fi - spx admin images import "${IMPORT_ARGS[@]}" + import_ami else echo "To import as AMI, run:" echo " spx admin images import \\" diff --git a/scripts/images/eks-node/eks-token-webhook.initd b/scripts/images/eks-node/eks-token-webhook.initd index 290825805..6b8260b28 100755 --- a/scripts/images/eks-node/eks-token-webhook.initd +++ b/scripts/images/eks-node/eks-token-webhook.initd @@ -1,9 +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). -# 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 +# Relays `aws eks get-token` bearer tokens through the AWS gateway broker (SigV4 +# HTTPS), which host-side verifies via Mulga STS + resolves the principal to a +# TokenReview via the cluster AccessEntry KV — the VM never speaks core NATS. +# Ordered `before k3s` so its apiserver kubeconfig exists before k3s reads # --authentication-token-webhook-config-file. description="EKS Token Review webhook (Mulga)" @@ -17,13 +18,14 @@ pidfile="/run/eks-token-webhook.pid" output_log="/dev/console" error_log="/dev/console" -# 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. +# Gateway endpoint + SigV4 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 EKS_GATEWAY_* / EKS_ACCESS_KEY / EKS_SECRET_KEY +# / EKS_REGION / 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 + export EKS_GATEWAY_URL EKS_GATEWAY_CA EKS_ACCESS_KEY EKS_SECRET_KEY EKS_REGION EKS_ACCOUNT_ID EKS_CLUSTER_NAME fi depend() { @@ -33,7 +35,7 @@ depend() { 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" + eerror "/etc/spinifex-eks/first-boot.env not present — cloud-init did not seed gateway creds" return 1 fi } diff --git a/scripts/images/eks-node/k3s-first-boot.sh b/scripts/images/eks-node/k3s-first-boot.sh index 991ac5060..72166df9f 100755 --- a/scripts/images/eks-node/k3s-first-boot.sh +++ b/scripts/images/eks-node/k3s-first-boot.sh @@ -5,13 +5,17 @@ set -eu # Reads the bootstrap node-token and admin kubeconfig that K3s writes at # server startup, rewrites the kubeconfig server address to the cluster's # NLB endpoint (so workers and external kubectl can use it), and publishes -# both as one-shot NATS messages for the spinifex cluster reconciler to -# consume into KV. +# the bootstrap artifacts to the host via the AWS gateway (SigV4-signed HTTPS +# POST — the eks-gateway-publish helper) for the spinifex cluster reconciler +# to consume into KV. The gateway relays onto the eks.bus.* NATS subjects; the +# VM never speaks NATS directly. # # Required env (from cloud-init user-data /etc/spinifex-eks/first-boot.env): -# SPINIFEX_NATS_URL nats://... -# SPINIFEX_NATS_TOKEN shared NATS auth token -# SPINIFEX_NATS_CA /etc/spinifex-eks/nats-ca.pem (TLS CA bundle) +# EKS_GATEWAY_URL https://{mgmt-host}:9999 (AWS gateway) +# EKS_GATEWAY_CA /etc/spinifex-eks/gateway-ca.pem (TLS CA bundle) +# EKS_ACCESS_KEY system SigV4 access key id +# EKS_SECRET_KEY system SigV4 secret access key +# EKS_REGION SigV4 signing region # EKS_ACCOUNT_ID # EKS_CLUSTER_NAME # EKS_NLB_ENDPOINT https://{cluster}.{accountID}.eks.{region}.{suffix} @@ -36,10 +40,15 @@ if [ ! -f "${ENVFILE}" ]; then die "${ENVFILE} not found — cloud-init did not seed first-boot env" fi +# set -a so the sourced KEY=value lines are exported: the eks-gateway-publish +# child reads its config from the environment (EKS_ACCOUNT_ID etc.). A bare +# source leaves them unexported and the helper exits "--account-id is required". +set -a # shellcheck disable=SC1090 . "${ENVFILE}" +set +a -for v in SPINIFEX_NATS_URL EKS_ACCOUNT_ID EKS_CLUSTER_NAME EKS_NLB_ENDPOINT; do +for v in EKS_GATEWAY_URL EKS_ACCESS_KEY EKS_SECRET_KEY EKS_ACCOUNT_ID EKS_CLUSTER_NAME EKS_NLB_ENDPOINT; do eval "val=\${$v:-}" [ -n "${val}" ] || die "env ${v} not set" done @@ -105,23 +114,18 @@ CA_B64=$(awk '/certificate-authority-data:/ {print $2; exit}' "${KUBECONFIG_FILE JWKS=$(kubectl --kubeconfig "${KUBECONFIG_FILE}" get --raw='/openid/v1/jwks' 2>/dev/null) [ -n "${JWKS}" ] || die "apiserver returned empty /openid/v1/jwks" -# 3. Publish the four one-shot bootstrap messages. Each is a BootstrapEnvelope -# JSON document (handlers/eks/nats_bootstrap.go); jq encodes the values so -# embedded newlines/quotes in the kubeconfig and JWKS stay valid JSON. -NATS_ARGS="-s ${SPINIFEX_NATS_URL}" -if [ -n "${SPINIFEX_NATS_TOKEN:-}" ]; then - NATS_ARGS="${NATS_ARGS} --token ${SPINIFEX_NATS_TOKEN}" -fi -if [ -n "${SPINIFEX_NATS_CA:-}" ] && [ -f "${SPINIFEX_NATS_CA}" ]; then - NATS_ARGS="${NATS_ARGS} --tlsca ${SPINIFEX_NATS_CA}" -fi +# 3. Publish the four bootstrap messages through the AWS gateway. Each is a +# BootstrapEnvelope JSON document (handlers/eks/nats_bootstrap.go); jq encodes +# the values so embedded newlines/quotes in the kubeconfig and JWKS stay valid +# JSON. eks-gateway-publish SigV4-signs the POST and retries until the gateway +# acks, then the gateway relays onto eks.bus.{account}.{cluster}.{kind}. It +# reads EKS_GATEWAY_URL/EKS_GATEWAY_CA/EKS_ACCESS_KEY/EKS_SECRET_KEY/EKS_REGION/ +# EKS_ACCOUNT_ID/EKS_CLUSTER_NAME from the env file already sourced above. # publish_envelope : reads the envelope JSON from stdin. publish_envelope() { - subj="eks.bus.${EKS_ACCOUNT_ID}.${EKS_CLUSTER_NAME}.$1" - log "publishing $1 to ${subj}" - # shellcheck disable=SC2086 - nats ${NATS_ARGS} pub "${subj}" --force-stdin + log "publishing $1 via gateway" + eks-gateway-publish -channel bootstrap -kind "$1" } jq -n --arg t "${NODE_TOKEN}" '{token: $t}' | publish_envelope k3s-bootstrap-token diff --git a/scripts/images/eks-node/manifest.conf b/scripts/images/eks-node/manifest.conf index 10fc2713e..317270b3c 100644 --- a/scripts/images/eks-node/manifest.conf +++ b/scripts/images/eks-node/manifest.conf @@ -9,16 +9,18 @@ ALPINE_VERSION="3.21.7" IMAGE_SIZE="4G" # Server superset of both package sets (etcd-ctl is ~5MB, not worth branching). -# K3s pulls cgroups + iptables; nats-cli used by k3s-first-boot publisher + -# state-report; jq builds the bootstrap-message JSON envelopes; etcd-client -# used by the snapshot script (server role only). +# K3s pulls cgroups + iptables; jq builds the bootstrap-message JSON envelopes; +# etcd-client used by the snapshot script (server role only). The control plane +# reaches the host via the AWS gateway (eks-gateway-publish), not a NATS CLI. APK_PACKAGES="ca-certificates curl jq iptables ip6tables conntrack-tools cgroup-tools etcd-ctl openrc" -# eks-token-webhook binary always installed (agents never start it). The k3s -# binary is downloaded + SHA256-verified by setup.sh (network in chroot). -BUILD_COMMANDS="CGO_ENABLED=0 go build -o bin/eks-token-webhook ./cmd/eks-token-webhook" +# eks-token-webhook binary always installed (agents never start it). +# eks-gateway-publish relays the bootstrap envelopes + state reports to the AWS +# gateway over SigV4 HTTPS (server role only). The k3s binary is downloaded + +# SHA256-verified by setup.sh (network in chroot). +BUILD_COMMANDS="CGO_ENABLED=0 GOFIPS140=v1.0.0 go build -o bin/eks-token-webhook ./cmd/eks-token-webhook && CGO_ENABLED=0 GOFIPS140=v1.0.0 go build -o bin/eks-gateway-publish ./cmd/eks-gateway-publish" -INSTALL_BINARIES="bin/eks-token-webhook:/usr/local/bin/eks-token-webhook" +INSTALL_BINARIES="bin/eks-token-webhook:/usr/local/bin/eks-token-webhook bin/eks-gateway-publish:/usr/local/bin/eks-gateway-publish" # All role services ship; none but the selector are enabled at bake time. # eks-node-role picks server vs agent at first boot and enables the rest. diff --git a/scripts/images/eks-node/mulga-eks-state-report.sh b/scripts/images/eks-node/mulga-eks-state-report.sh index 5339aa738..f7464d3f9 100755 --- a/scripts/images/eks-node/mulga-eks-state-report.sh +++ b/scripts/images/eks-node/mulga-eks-state-report.sh @@ -1,42 +1,40 @@ #!/bin/sh set -eu -# mulga-eks-state-report — NATS publish of K3s server health for the spinifex -# cluster reconciler. The apiserver binds the VPC node-ip, unreachable from the -# host-side daemon, so the daemon cannot probe /healthz directly — this report -# carries {healthz, node_count} over the management NATS bus the daemon shares. +# mulga-eks-state-report — publishes K3s server health to the spinifex cluster +# reconciler through the AWS gateway. The apiserver binds the VPC node-ip, +# unreachable from the host-side daemon, so the daemon cannot probe /healthz +# directly — this report carries {healthz, node_count}. eks-gateway-publish +# SigV4-signs the POST to the gateway, which relays it onto the management NATS +# bus the daemon shares; the VM never speaks NATS directly. # -# Subject: eks.state.{accountID}.{clusterName}.server +# Subject (gateway-side): eks.state.{accountID}.{clusterName}.server # Payload: JSON { healthz: "ok"|"fail", node_count: N, ts: } # -# Reads NATS creds + cluster identity from the cloud-init-seeded first-boot env -# (same vars k3s-first-boot.sh publishes its bootstrap artifacts with). Run as a -# loop by the mulga-eks-state-report OpenRC service when STATE_REPORT_INTERVAL is -# set; one-shot otherwise (e.g. a manual invocation). +# Reads gateway creds + cluster identity from the cloud-init-seeded first-boot +# env (same vars k3s-first-boot.sh publishes its bootstrap artifacts with). Run +# as a loop by the mulga-eks-state-report OpenRC service when +# STATE_REPORT_INTERVAL is set; one-shot otherwise (e.g. a manual invocation). ENVFILE=/etc/spinifex-eks/first-boot.env [ -f "${ENVFILE}" ] || { logger -t mulga-eks-state-report "${ENVFILE} missing — exiting"; exit 0; } +# set -a so the sourced KEY=value lines are exported to the eks-gateway-publish +# child (it reads EKS_ACCOUNT_ID etc. from its environment); a bare source +# leaves them unexported and the helper exits "--account-id is required". +set -a # shellcheck disable=SC1090 . "${ENVFILE}" +set +a -: "${SPINIFEX_NATS_URL:?}" +: "${EKS_GATEWAY_URL:?}" +: "${EKS_ACCESS_KEY:?}" +: "${EKS_SECRET_KEY:?}" : "${EKS_ACCOUNT_ID:?}" : "${EKS_CLUSTER_NAME:?}" KUBECONFIG=${KUBECONFIG:-/etc/rancher/k3s/k3s.yaml} export KUBECONFIG -SUBJ="eks.state.${EKS_ACCOUNT_ID}.${EKS_CLUSTER_NAME}.server" - -# NATS args mirror k3s-first-boot.sh (token + TLS CA) — the proven publish path. -NATS_ARGS="-s ${SPINIFEX_NATS_URL}" -if [ -n "${SPINIFEX_NATS_TOKEN:-}" ]; then - NATS_ARGS="${NATS_ARGS} --token ${SPINIFEX_NATS_TOKEN}" -fi -if [ -n "${SPINIFEX_NATS_CA:-}" ] && [ -f "${SPINIFEX_NATS_CA}" ]; then - NATS_ARGS="${NATS_ARGS} --tlsca ${SPINIFEX_NATS_CA}" -fi - publish_report() { if curl -sk --max-time 3 https://127.0.0.1:6443/healthz | grep -q '^ok$'; then health=ok @@ -48,8 +46,7 @@ publish_report() { node_count=0 fi payload=$(printf '{"healthz":"%s","node_count":%s,"ts":%s}' "${health}" "${node_count}" "$(date +%s)") - # shellcheck disable=SC2086 - printf '%s' "${payload}" | nats ${NATS_ARGS} pub "${SUBJ}" --force-stdin 2>&1 \ + printf '%s' "${payload}" | eks-gateway-publish -channel state 2>&1 \ | logger -t mulga-eks-state-report } diff --git a/scripts/images/eks-node/setup.sh b/scripts/images/eks-node/setup.sh index a81db515c..2e0896acb 100755 --- a/scripts/images/eks-node/setup.sh +++ b/scripts/images/eks-node/setup.sh @@ -1,16 +1,17 @@ #!/bin/sh set -eu -# setup.sh — chroot customisation for the unified eks-node AMI (server + agent). +# setup.sh — guest customisation for the unified eks-node AMI (server + agent). # -# Runs inside an Alpine 3.21 chroot under build-system-image.sh after packages -# and binaries are installed. Downloads the pinned K3s binary, verifies its -# SHA256, drops it into /usr/local/bin, sets executable bits on all role init -# scripts + cron entries, and writes the K3s server config skeleton. The role -# (server vs agent) is selected per-instance at first boot by eks-node-role. +# Runs inside the libguestfs appliance (via virt-customize --run) under +# build-system-image.sh after packages and binaries are installed. Downloads the +# pinned K3s binary, verifies its SHA256, drops it into /usr/local/bin, sets +# executable bits on all role init scripts + cron entries, and writes the K3s +# server config skeleton. The role (server vs agent) is selected per-instance at +# first boot by eks-node-role. # -# Network access inside the chroot is provided by the host's /etc/resolv.conf -# (build-system-image.sh copies it in). curl is in APK_PACKAGES. +# Network access inside the appliance is provided by virt-customize --network. +# curl is in APK_PACKAGES. K3S_VERSION="v1.32.5+k3s1" K3S_URL_BASE="https://github.com/k3s-io/k3s/releases/download/${K3S_VERSION}" @@ -39,19 +40,6 @@ ln -sf /usr/local/bin/k3s /usr/local/bin/kubectl ln -sf /usr/local/bin/k3s /usr/local/bin/crictl ln -sf /usr/local/bin/k3s /usr/local/bin/ctr -# nats-cli is not in Alpine main/community — pull the upstream release binary. -# Used by k3s-first-boot.sh (one-shot publishes) and mulga-eks-state-report.sh -# (periodic publishes). -NATS_CLI_VERSION="0.4.0" -NATS_CLI_URL="https://github.com/nats-io/natscli/releases/download/v${NATS_CLI_VERSION}/nats-${NATS_CLI_VERSION}-linux-amd64.zip" -echo "[eks-node-setup] downloading nats-cli ${NATS_CLI_VERSION}" -curl -fsSL -o /tmp/nats-cli.zip "${NATS_CLI_URL}" -apk add --no-cache unzip -unzip -q -d /tmp/nats-cli /tmp/nats-cli.zip -install -m 0755 /tmp/nats-cli/nats-${NATS_CLI_VERSION}-linux-amd64/nats /usr/local/bin/nats -rm -rf /tmp/nats-cli /tmp/nats-cli.zip -apk del --no-cache unzip - # Init scripts ship as 0644 from INSTALL_FILES; OpenRC requires 0755. Every # role's services are baked; the selector enables the right ones at first boot. chmod 0755 /etc/init.d/eks-node-role /etc/init.d/k3s /etc/init.d/k3s-agent \ diff --git a/spinifex/daemon/eks_deps.go b/spinifex/daemon/eks_deps.go index 8a3f51f7a..6284e3150 100644 --- a/spinifex/daemon/eks_deps.go +++ b/spinifex/daemon/eks_deps.go @@ -29,37 +29,39 @@ func (d *Daemon) buildEKSServiceDeps() handlers_eks.EKSServiceDeps { masterKey = nil } - // CA PEM the K3s server VM uses to verify the daemon's NATS TLS. Same CA - // that signs the AWSGW server cert; read here so the VM can publish its - // one-shot bootstrap messages over the token+TLS NATS endpoint. - natsCA := "" + // CA PEM the K3s server VM uses to verify the AWS gateway's HTTPS cert when + // the eks-gateway-publish helper POSTs its bootstrap envelopes + state + // reports. The same CA signs the AWSGW server cert and the NATS server cert, + // so it is read from the configured NATS CA path. + gatewayCA := "" if d.config.NATS.CACert != "" { if caBytes, readErr := os.ReadFile(d.config.NATS.CACert); readErr == nil { - natsCA = string(caBytes) + gatewayCA = string(caBytes) } else { - slog.Warn("EKS: read NATS CACert failed; K3s server VM will not reach NATS over TLS", + slog.Warn("EKS: read gateway CACert failed; K3s server VM will not verify the gateway over TLS", "path", d.config.NATS.CACert, "err", readErr) } } return handlers_eks.EKSServiceDeps{ - Config: d.config, - NATSConn: d.natsConn, - MasterKey: masterKey, - GatewayBaseURL: d.resolveGatewayBaseURL(), - Region: d.config.Region, - HolderID: d.node, - NATSURL: d.resolveSystemNATSURL(), - NATSToken: d.config.NATS.ACL.Token, - NATSCACert: natsCA, - VPCSG: d.vpcService, - VPCK3s: d.vpcService, - VPCSubnet: &daemonEKSSubnetResolver{d: d}, - NLB: d.elbv2Service, - Instance: d, - Image: d.imageService, - EIP: d.eipService, - Worker: d, + Config: d.config, + NATSConn: d.natsConn, + MasterKey: masterKey, + GatewayBaseURL: d.resolveGatewayBaseURL(), + Region: d.config.Region, + HolderID: d.node, + SystemGatewayURL: d.resolveSystemGatewayBaseURL(), + SystemAccessKey: d.config.Predastore.AccessKey, + SystemSecretKey: d.config.Predastore.SecretKey, + GatewayCACert: gatewayCA, + VPCSG: d.vpcService, + VPCK3s: d.vpcService, + VPCSubnet: &daemonEKSSubnetResolver{d: d}, + NLB: d.elbv2Service, + Instance: d, + Image: d.imageService, + EIP: d.eipService, + Worker: d, } } @@ -121,40 +123,24 @@ func (d *Daemon) resolveGatewayBaseURL() string { return "https://" + net.JoinHostPort(host, gatewayPort) } -// resolveSystemNATSURL derives the NATS URL a system instance (the K3s server -// VM) dials to publish its one-shot bootstrap messages. Unlike a customer VM, +// resolveSystemGatewayBaseURL derives the AWS gateway URL a system instance +// (the K3s server VM) POSTs its bootstrap envelopes + state reports to. Unlike +// a customer LB VM — which reaches the gateway via VPC→external (AdvertiseIP) — // a system instance reaches the daemon only over the mgmt bridge: its cloud-init -// mgmt0 NIC is on-link to mgmtBridgeIP with no off-link route, so NATS must be -// dialed at the bridge address directly. NATS listens on 0.0.0.0:4222, so the -// bridge IP is reachable. Falls back to resolveNATSURL() (gateway-host -// precedence) when no mgmt bridge exists — e.g. dev-shim networking. -func (d *Daemon) resolveSystemNATSURL() string { +// mgmt0 NIC is on-link to mgmtBridgeIP with no off-link route, so the gateway +// must be dialed at the bridge address directly. AWSGW binds 0.0.0.0 (or the +// mgmt IP), and the server cert SANs include the bridge IP, so HTTPS validates. +// Falls back to resolveGatewayBaseURL() (gateway-host precedence) when no mgmt +// bridge exists — e.g. dev-shim networking. +func (d *Daemon) resolveSystemGatewayBaseURL() string { if d.mgmtBridgeIP == "" { - return d.resolveNATSURL() + return d.resolveGatewayBaseURL() } - natsPort := "4222" - if d.config.NATS.Host != "" { - if _, port, splitErr := net.SplitHostPort(d.config.NATS.Host); splitErr == nil && port != "" { - natsPort = port + gatewayPort := "9999" + if d.config.AWSGW.Host != "" { + if _, port, splitErr := net.SplitHostPort(d.config.AWSGW.Host); splitErr == nil && port != "" { + gatewayPort = port } } - return "nats://" + net.JoinHostPort(d.mgmtBridgeIP, natsPort) -} - -// resolveNATSURL derives the NATS URL the K3s server VM dials to publish its -// one-shot bootstrap messages. Host reuses the gateway host resolution; port -// comes from the configured NATS listen address (default 4222). Returns "" if -// no reachable host can be derived. -func (d *Daemon) resolveNATSURL() string { - host := d.resolveGatewayHost() - if host == "" { - return "" - } - natsPort := "4222" - if d.config.NATS.Host != "" { - if _, port, splitErr := net.SplitHostPort(d.config.NATS.Host); splitErr == nil && port != "" { - natsPort = port - } - } - return "nats://" + net.JoinHostPort(host, natsPort) + return "https://" + net.JoinHostPort(d.mgmtBridgeIP, gatewayPort) } diff --git a/spinifex/gateway/eks.go b/spinifex/gateway/eks.go index ce6ab2d9f..95c1a0c17 100644 --- a/spinifex/gateway/eks.go +++ b/spinifex/gateway/eks.go @@ -60,6 +60,22 @@ var eksRoutes = []eksRoute{ func(gw *GatewayConfig, acct, callerARN string, p []string, b []byte) (any, error) { return gateway_eks.UpdateClusterVersion(gw.NATSConn, acct, p[0], b) }}, + // Control-plane VM broker: the K3s server VM POSTs bootstrap envelopes and + // state reports here (system SigV4 creds) and the gateway relays them onto + // the existing eks.bus.*/eks.state.* NATS subjects. acct (system account) + // and callerARN are ignored — the cluster account comes from the body. + {"POST", regexp.MustCompile(`^/clusters/([^/]+)/internal-publish$`), "PublishInternal", + func(gw *GatewayConfig, acct, callerARN string, p []string, b []byte) (any, error) { + return gateway_eks.PublishInternal(gw.NATSConn, p[0], b) + }}, + // Control-plane VM TokenReview broker: the eks-token-webhook POSTs the + // `aws eks get-token` bearer token here (system SigV4 creds) and the gateway + // resolves it host-side (STS verify + AccessEntry lookup), so the webhook + // never speaks NATS. Cluster account comes from the body. + {"POST", regexp.MustCompile(`^/clusters/([^/]+)/token-review$`), "WebhookTokenReview", + func(gw *GatewayConfig, acct, callerARN string, p []string, b []byte) (any, error) { + return gateway_eks.WebhookTokenReview(gw.NATSConn, p[0], b) + }}, // Nodegroup {"POST", regexp.MustCompile(`^/clusters/([^/]+)/node-groups$`), "CreateNodegroup", diff --git a/spinifex/gateway/eks/internal_publish.go b/spinifex/gateway/eks/internal_publish.go new file mode 100644 index 000000000..e66b8a4bf --- /dev/null +++ b/spinifex/gateway/eks/internal_publish.go @@ -0,0 +1,91 @@ +package gateway_eks + +import ( + "encoding/json" + "errors" + "log/slog" + + "github.com/mulgadc/spinifex/spinifex/awserrors" + handlers_eks "github.com/mulgadc/spinifex/spinifex/handlers/eks" + "github.com/nats-io/nats.go" +) + +// internalPublishChannels are the two VM→host channels the broker relays. +const ( + internalChannelBootstrap = "bootstrap" + internalChannelState = "state" +) + +// internalPublishRequest is the wire shape the on-VM eks-gateway-publish helper +// POSTs to /clusters/{name}/internal-publish. The control-plane VM holds system +// SigV4 creds (system account, not the cluster account), so AccountID names the +// customer cluster account explicitly rather than being taken from the SigV4 +// auth context. Payload is the already-encoded subject body (a BootstrapEnvelope +// for the bootstrap channel, a ServerStateReport for the state channel) — the +// broker relays it verbatim onto the existing NATS subject, so the daemon-side +// subscriber, KV persistence, and reconciler are unchanged. +type internalPublishRequest struct { + AccountID string `json:"accountId"` + Channel string `json:"channel"` + Kind string `json:"kind"` + Payload json.RawMessage `json:"payload"` +} + +// publishInternalOutput is the empty success body (`{}`). +type publishInternalOutput struct{} + +// validBootstrapKinds is the closed set of bootstrap subject suffixes the broker +// will relay — a malformed/hostile kind cannot be turned into an arbitrary +// subject suffix. +var validBootstrapKinds = map[string]struct{}{ + handlers_eks.BootstrapSubjectToken: {}, + handlers_eks.BootstrapSubjectKubeconfig: {}, + handlers_eks.BootstrapSubjectJWKS: {}, + handlers_eks.BootstrapSubjectCA: {}, +} + +// PublishInternal — POST /clusters/{name}/internal-publish. Relays a +// control-plane VM publication onto the existing bootstrap/state NATS subjects. +// This is the gateway-broker replacement for the VM dialing core NATS directly: +// the VM reaches the daemon over the mgmt-reachable AWSGW with SigV4 + retry +// (DDIL-correct), and NATS stays cluster-internal. +func PublishInternal(natsConn *nats.Conn, clusterName string, body []byte) (*publishInternalOutput, error) { + if natsConn == nil { + return nil, errors.New(awserrors.ErrorServerInternal) + } + if clusterName == "" { + return nil, errors.New(awserrors.ErrorInvalidParameterValue) + } + + var req internalPublishRequest + if err := json.Unmarshal(body, &req); err != nil { + slog.Debug("PublishInternal: bad body", "cluster", clusterName, "err", err) + return nil, errors.New(awserrors.ErrorInvalidParameterValue) + } + if req.AccountID == "" || len(req.Payload) == 0 { + return nil, errors.New(awserrors.ErrorInvalidParameterValue) + } + + var subject string + switch req.Channel { + case internalChannelBootstrap: + if _, ok := validBootstrapKinds[req.Kind]; !ok { + slog.Debug("PublishInternal: unknown bootstrap kind", "cluster", clusterName, "kind", req.Kind) + return nil, errors.New(awserrors.ErrorInvalidParameterValue) + } + subject = handlers_eks.BootstrapSubject(req.AccountID, clusterName, req.Kind) + case internalChannelState: + subject = handlers_eks.StateSubject(req.AccountID, clusterName) + default: + slog.Debug("PublishInternal: unknown channel", "cluster", clusterName, "channel", req.Channel) + return nil, errors.New(awserrors.ErrorInvalidParameterValue) + } + + if err := natsConn.Publish(subject, req.Payload); err != nil { + slog.Error("PublishInternal: NATS publish failed", "subject", subject, "err", err) + return nil, errors.New(awserrors.ErrorServerInternal) + } + slog.Debug("PublishInternal: relayed", "subject", subject, "bytes", len(req.Payload)) + + return &publishInternalOutput{}, nil +} diff --git a/spinifex/gateway/eks/internal_publish_test.go b/spinifex/gateway/eks/internal_publish_test.go new file mode 100644 index 000000000..9a8bf592c --- /dev/null +++ b/spinifex/gateway/eks/internal_publish_test.go @@ -0,0 +1,80 @@ +package gateway_eks + +import ( + "testing" + "time" + + handlers_eks "github.com/mulgadc/spinifex/spinifex/handlers/eks" + "github.com/mulgadc/spinifex/spinifex/testutil" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestPublishInternal_RelaysToSubjects(t *testing.T) { + _, nc := testutil.StartTestNATS(t) + + cases := []struct { + name string + body string + subject string + payload string + }{ + { + name: "bootstrap token", + body: `{"accountId":"111122223333","channel":"bootstrap","kind":"k3s-bootstrap-token","payload":{"token":"k3s-secret"}}`, + subject: handlers_eks.BootstrapSubject("111122223333", "alpha", handlers_eks.BootstrapSubjectToken), + payload: `{"token":"k3s-secret"}`, + }, + { + name: "state report", + body: `{"accountId":"111122223333","channel":"state","payload":{"healthz":"ok","node_count":1,"ts":42}}`, + subject: handlers_eks.StateSubject("111122223333", "alpha"), + payload: `{"healthz":"ok","node_count":1,"ts":42}`, + }, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + sub, err := nc.SubscribeSync(tc.subject) + require.NoError(t, err) + t.Cleanup(func() { _ = sub.Unsubscribe() }) + + out, err := PublishInternal(nc, "alpha", []byte(tc.body)) + require.NoError(t, err) + require.NotNil(t, out) + + msg, err := sub.NextMsg(2 * time.Second) + require.NoError(t, err) + assert.JSONEq(t, tc.payload, string(msg.Data)) + }) + } +} + +func TestPublishInternal_Rejects(t *testing.T) { + _, nc := testutil.StartTestNATS(t) + + cases := []struct { + name string + body string + }{ + {"empty account", `{"accountId":"","channel":"state","payload":{"healthz":"ok"}}`}, + {"empty payload", `{"accountId":"111122223333","channel":"state"}`}, + {"unknown channel", `{"accountId":"111122223333","channel":"wat","payload":{"x":1}}`}, + {"unknown bootstrap kind", `{"accountId":"111122223333","channel":"bootstrap","kind":"evil","payload":{"x":1}}`}, + {"malformed json", `{not-json`}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + _, err := PublishInternal(nc, "alpha", []byte(tc.body)) + require.Error(t, err) + }) + } +} + +func TestPublishInternal_NilConnAndEmptyCluster(t *testing.T) { + _, err := PublishInternal(nil, "alpha", []byte(`{}`)) + require.Error(t, err) + + _, nc := testutil.StartTestNATS(t) + _, err = PublishInternal(nc, "", []byte(`{"accountId":"1","channel":"state","payload":{"x":1}}`)) + require.Error(t, err) +} diff --git a/spinifex/gateway/eks/token_review.go b/spinifex/gateway/eks/token_review.go new file mode 100644 index 000000000..284532595 --- /dev/null +++ b/spinifex/gateway/eks/token_review.go @@ -0,0 +1,54 @@ +package gateway_eks + +import ( + "encoding/json" + "errors" + "log/slog" + "time" + + "github.com/mulgadc/spinifex/spinifex/awserrors" + handlers_eks "github.com/mulgadc/spinifex/spinifex/handlers/eks" + "github.com/nats-io/nats.go" +) + +// tokenReviewVerifyTimeout bounds the host-side STS verify NATS round-trip. The +// apiserver fails the TokenReview on its own timeout, so keep this tight. +const tokenReviewVerifyTimeout = 5 * time.Second + +// webhookTokenReviewRequest is the body the on-VM eks-token-webhook POSTs to +// /clusters/{name}/token-review. The webhook holds system SigV4 creds (system +// account, not the cluster account), so AccountID names the cluster account +// explicitly — same pattern as PublishInternal. +type webhookTokenReviewRequest struct { + AccountID string `json:"accountId"` + Token string `json:"token"` +} + +// WebhookTokenReview — POST /clusters/{name}/token-review. Resolves an +// `aws eks get-token` bearer token to a K8s identity host-side (STS verify + +// AccessEntry lookup) and returns it. This is the gateway-broker replacement +// for the webhook dialing core NATS/STS/KV directly. +func WebhookTokenReview(natsConn *nats.Conn, clusterName string, body []byte) (*handlers_eks.WebhookTokenReviewResult, error) { + if natsConn == nil { + return nil, errors.New(awserrors.ErrorServerInternal) + } + if clusterName == "" { + return nil, errors.New(awserrors.ErrorInvalidParameterValue) + } + + var req webhookTokenReviewRequest + if err := json.Unmarshal(body, &req); err != nil { + slog.Debug("WebhookTokenReview: bad body", "cluster", clusterName, "err", err) + return nil, errors.New(awserrors.ErrorInvalidParameterValue) + } + if req.AccountID == "" || req.Token == "" { + return nil, errors.New(awserrors.ErrorInvalidParameterValue) + } + + res, err := handlers_eks.ResolveTokenReview(natsConn, req.AccountID, clusterName, req.Token, tokenReviewVerifyTimeout) + if err != nil { + slog.Error("WebhookTokenReview: resolve failed", "cluster", clusterName, "err", err) + return nil, errors.New(awserrors.ErrorServerInternal) + } + return &res, nil +} diff --git a/spinifex/gateway/eks/token_review_test.go b/spinifex/gateway/eks/token_review_test.go new file mode 100644 index 000000000..5d62d57f0 --- /dev/null +++ b/spinifex/gateway/eks/token_review_test.go @@ -0,0 +1,86 @@ +package gateway_eks + +import ( + "encoding/base64" + "encoding/json" + "testing" + + "github.com/mulgadc/spinifex/spinifex/awserrors" + handlers_eks "github.com/mulgadc/spinifex/spinifex/handlers/eks" + "github.com/mulgadc/spinifex/spinifex/testutil" + "github.com/nats-io/nats.go" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +const testARN = "arn:aws:iam::111122223333:role/admin" + +// validToken builds a k8s-aws-v1 bearer token (the `aws eks get-token` shape): +// the literal prefix + base64 raw-url of the presigned GetCallerIdentity URL. +func validToken(url string) string { + return "k8s-aws-v1." + base64.RawURLEncoding.EncodeToString([]byte(url)) +} + +func TestWebhookTokenReview_Rejects(t *testing.T) { + _, nc := testutil.StartTestNATS(t) + + // nil conn → ServerInternal + _, err := WebhookTokenReview(nil, "alpha", []byte(`{"accountId":"1","token":"t"}`)) + require.Error(t, err) + assert.Equal(t, awserrors.ErrorServerInternal, err.Error()) + + cases := []struct { + name, cluster, body, want string + }{ + {"empty cluster", "", `{"accountId":"1","token":"t"}`, awserrors.ErrorInvalidParameterValue}, + {"malformed json", "alpha", `{not-json`, awserrors.ErrorInvalidParameterValue}, + {"empty account", "alpha", `{"accountId":"","token":"t"}`, awserrors.ErrorInvalidParameterValue}, + {"empty token", "alpha", `{"accountId":"1","token":""}`, awserrors.ErrorInvalidParameterValue}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + _, err := WebhookTokenReview(nc, tc.cluster, []byte(tc.body)) + require.Error(t, err) + assert.Equal(t, tc.want, err.Error()) + }) + } +} + +// A genuine infra fault (account bucket absent) maps to ServerInternal. +func TestWebhookTokenReview_InfraFaultIsServerInternal(t *testing.T) { + _, nc, _ := testutil.StartTestJetStream(t) + _, err := WebhookTokenReview(nc, "alpha", []byte(`{"accountId":"111122223333","token":"`+validToken("https://sts/?x=1")+`"}`)) + require.Error(t, err) + assert.Equal(t, awserrors.ErrorServerInternal, err.Error()) +} + +func TestWebhookTokenReview_ResolvesIdentity(t *testing.T) { + _, nc, js := testutil.StartTestJetStream(t) + kv := testutil.SeedKV(t, js, handlers_eks.AccountBucketName("111122223333"), nil) + require.NoError(t, handlers_eks.PutAccessEntryRecord(kv, &handlers_eks.AccessEntryRecord{ + ClusterName: "alpha", + PrincipalARN: testARN, + KubernetesUsername: testARN, + KubernetesGroups: []string{"system:masters"}, + Type: handlers_eks.AccessEntryTypeStandard, + })) + + sub, err := nc.Subscribe(handlers_eks.TokenVerifySubject, func(m *nats.Msg) { + resp, _ := json.Marshal(handlers_eks.TokenVerifyResponse{ + AccountID: "111122223333", + ARN: testARN, + UserID: "AROAEXAMPLE:session", + PrincipalType: "AssumedRole", + }) + _ = m.Respond(resp) + }) + require.NoError(t, err) + t.Cleanup(func() { _ = sub.Unsubscribe() }) + + out, err := WebhookTokenReview(nc, "alpha", []byte(`{"accountId":"111122223333","token":"`+validToken("https://sts/?Action=GetCallerIdentity")+`"}`)) + require.NoError(t, err) + require.NotNil(t, out) + assert.True(t, out.Authenticated) + assert.Equal(t, testARN, out.Username) + assert.Equal(t, []string{"system:masters"}, out.Groups) +} diff --git a/spinifex/gateway/eks_test.go b/spinifex/gateway/eks_test.go index f581ae9d6..787e074ad 100644 --- a/spinifex/gateway/eks_test.go +++ b/spinifex/gateway/eks_test.go @@ -97,9 +97,14 @@ func TestLookupEKSAction_EncodedPrincipalARN(t *testing.T) { } } -func TestLookupEKSAction_CoversAll34Actions(t *testing.T) { +func TestLookupEKSAction_CoversAllActions(t *testing.T) { expected := map[string]bool{ - "CreateCluster": false, + "CreateCluster": false, + // PublishInternal + WebhookTokenReview are internal control-plane VM + // broker routes, not AWS-SDK EKS actions — registered alongside the + // public actions. + "PublishInternal": false, + "WebhookTokenReview": false, "DescribeCluster": false, "ListClusters": false, "UpdateClusterConfig": false, diff --git a/spinifex/handlers/eks/delete_cluster_test.go b/spinifex/handlers/eks/delete_cluster_test.go index 6d89e82b2..5808ab174 100644 --- a/spinifex/handlers/eks/delete_cluster_test.go +++ b/spinifex/handlers/eks/delete_cluster_test.go @@ -53,22 +53,23 @@ func newEKSServiceFixture(t *testing.T) *eksServiceFixture { worker := newFakeWorkerLauncher() svc, err := NewEKSServiceImpl(EKSServiceDeps{ - NATSConn: nc, - MasterKey: bootstrapTestMasterKey, - GatewayBaseURL: "https://gw.local:9999", - Region: "us-east-1", - HolderID: "node-1", - NATSURL: "nats://gw.local:4222", - NATSToken: "s3cr3t-token", - NATSCACert: "-----BEGIN CERTIFICATE-----\nFAKECA\n-----END CERTIFICATE-----\n", - VPCSG: sg, - VPCK3s: vpc, - VPCSubnet: fakeSubnetResolver{}, - NLB: nlb, - Instance: inst, - Image: ami, - EIP: eip, - Worker: worker, + NATSConn: nc, + MasterKey: bootstrapTestMasterKey, + GatewayBaseURL: "https://gw.local:9999", + Region: "us-east-1", + HolderID: "node-1", + SystemGatewayURL: "https://gw.local:9999", + SystemAccessKey: "AKIAEXAMPLE", + SystemSecretKey: "s3cr3t-key", + GatewayCACert: "-----BEGIN CERTIFICATE-----\nFAKECA\n-----END CERTIFICATE-----\n", + VPCSG: sg, + VPCK3s: vpc, + VPCSubnet: fakeSubnetResolver{}, + NLB: nlb, + Instance: inst, + Image: ami, + EIP: eip, + Worker: worker, }) require.NoError(t, err) t.Cleanup(svc.Shutdown) diff --git a/spinifex/handlers/eks/k3s_server_vm.go b/spinifex/handlers/eks/k3s_server_vm.go index 6dcb02e50..f366ac850 100644 --- a/spinifex/handlers/eks/k3s_server_vm.go +++ b/spinifex/handlers/eks/k3s_server_vm.go @@ -76,10 +76,11 @@ const ( // in scripts/images/eks-node/eks-node-role.sh and k3s-agent.initd. agentEnvPath = "/etc/spinifex-eks/agent.env" - // k3sNATSCAPath is the on-VM destination for the NATS CA cert PEM. The K3s - // VM uses it to verify the daemon's NATS TLS when publishing bootstrap - // messages. Path matches k3s-first-boot.sh SPINIFEX_NATS_CA. - k3sNATSCAPath = "/etc/spinifex-eks/nats-ca.pem" + // k3sGatewayCAPath is the on-VM destination for the AWS gateway TLS CA cert + // PEM. The K3s VM uses it to verify the gateway's HTTPS cert when the + // eks-gateway-publish helper POSTs bootstrap envelopes + state reports. + // Path matches k3s-first-boot.sh EKS_GATEWAY_CA. + k3sGatewayCAPath = "/etc/spinifex-eks/gateway-ca.pem" // k3sTokenWebhookKubeconfigPath is the apiserver token-webhook config the // eks-token-webhook service (ordered `before k3s`) writes before the @@ -120,10 +121,17 @@ type K3sServerInput struct { OIDCIssuer string OIDCPrivateKeyPEM string OIDCPublicKeyPEM string - NATSURL string - NATSToken string - NATSCACert string - InstanceType string + // Gateway broker config: the control-plane VM publishes its bootstrap + // envelopes + state reports via SigV4-signed HTTPS POST to the AWS gateway + // (the ELBv2 lb-agent model), not by dialing core NATS. GatewayURL is the + // mgmt-reachable AWSGW endpoint; AccessKey/SecretKey are the system + // (Predastore) SigV4 creds; GatewayCACert is the PEM that signs the gateway + // server cert. + GatewayURL string + AccessKey string + SecretKey string + GatewayCACert string + InstanceType string } // K3sServerOutput carries the identifiers the caller needs to persist into @@ -346,12 +354,14 @@ func validateK3sServerInput(in K3sServerInput) error { return errors.New("eks: K3sServerInput empty OIDCPrivateKeyPEM") case strings.TrimSpace(in.OIDCPublicKeyPEM) == "": return errors.New("eks: K3sServerInput empty OIDCPublicKeyPEM") - case in.NATSURL == "": - return errors.New("eks: K3sServerInput empty NATSURL") - case in.NATSToken == "": - return errors.New("eks: K3sServerInput empty NATSToken") - case strings.TrimSpace(in.NATSCACert) == "": - return errors.New("eks: K3sServerInput empty NATSCACert") + case in.GatewayURL == "": + return errors.New("eks: K3sServerInput empty GatewayURL") + case in.AccessKey == "": + return errors.New("eks: K3sServerInput empty AccessKey") + case in.SecretKey == "": + return errors.New("eks: K3sServerInput empty SecretKey") + case strings.TrimSpace(in.GatewayCACert) == "": + return errors.New("eks: K3sServerInput empty GatewayCACert") } return nil } @@ -374,9 +384,11 @@ func buildK3sUserData(in K3sServerInput) string { // control-plane services (k3s server + token webhook + bootstrap // publisher). Nodegroup workers seed "agent" through their own path. "SPINIFEX_K3S_ROLE=server", - "SPINIFEX_NATS_URL=" + in.NATSURL, - "SPINIFEX_NATS_TOKEN=" + in.NATSToken, - "SPINIFEX_NATS_CA=" + k3sNATSCAPath, + "EKS_GATEWAY_URL=" + in.GatewayURL, + "EKS_GATEWAY_CA=" + k3sGatewayCAPath, + "EKS_ACCESS_KEY=" + in.AccessKey, + "EKS_SECRET_KEY=" + in.SecretKey, + "EKS_REGION=" + in.Region, "EKS_ACCOUNT_ID=" + in.AccountID, "EKS_CLUSTER_NAME=" + in.ClusterName, "EKS_NLB_ENDPOINT=" + nlbEndpoint, @@ -412,12 +424,13 @@ func buildK3sUserData(in K3sServerInput) string { }, "\n") files := []userDataFile{ - // first-boot.env carries the NATS token, so keep it root-only (0600). + // first-boot.env carries the system SigV4 secret key, so keep it + // root-only (0600). {Path: k3sFirstBootEnvPath, Perms: "0600", Body: envBody}, {Path: k3sOIDCSigningKeyPath, Perms: "0600", Body: strings.TrimRight(in.OIDCPrivateKeyPEM, "\n")}, {Path: k3sOIDCPublicKeyPath, Perms: "0644", Body: strings.TrimRight(in.OIDCPublicKeyPEM, "\n")}, {Path: k3sConfigPath, Perms: "0644", Body: k3sConfig}, - {Path: k3sNATSCAPath, Perms: "0644", Body: strings.TrimRight(in.NATSCACert, "\n")}, + {Path: k3sGatewayCAPath, Perms: "0644", Body: strings.TrimRight(in.GatewayCACert, "\n")}, // IMDS on-link route. Alpine's cloud-init eni renderer crashes on a // gateway-less route in network-config, so it's delivered out-of-band: // a persistent local.d script (re-applied every boot by the OpenRC diff --git a/spinifex/handlers/eks/k3s_server_vm_test.go b/spinifex/handlers/eks/k3s_server_vm_test.go index e05113c34..851831763 100644 --- a/spinifex/handlers/eks/k3s_server_vm_test.go +++ b/spinifex/handlers/eks/k3s_server_vm_test.go @@ -118,9 +118,10 @@ func validK3sInput() K3sServerInput { OIDCIssuer: "https://oidc.spinifex.local/clusters/111122223333/alpha", OIDCPrivateKeyPEM: "-----BEGIN PRIVATE KEY-----\nFAKEKEY\n-----END PRIVATE KEY-----\n", OIDCPublicKeyPEM: "-----BEGIN PUBLIC KEY-----\nFAKEPUB\n-----END PUBLIC KEY-----\n", - NATSURL: "nats://localhost:4222", - NATSToken: "s3cr3t-token", - NATSCACert: "-----BEGIN CERTIFICATE-----\nFAKECA\n-----END CERTIFICATE-----\n", + GatewayURL: "https://10.15.8.1:9999", + AccessKey: "AKIAEXAMPLE", + SecretKey: "s3cr3t-key", + GatewayCACert: "-----BEGIN CERTIFICATE-----\nFAKECA\n-----END CERTIFICATE-----\n", } } @@ -142,9 +143,10 @@ func TestLaunchK3sServerVM_EmptyInputsRejected(t *testing.T) { {"empty OIDCIssuer", mk(func(in *K3sServerInput) { in.OIDCIssuer = "" })}, {"empty OIDCPrivateKeyPEM", mk(func(in *K3sServerInput) { in.OIDCPrivateKeyPEM = " \n " })}, {"empty OIDCPublicKeyPEM", mk(func(in *K3sServerInput) { in.OIDCPublicKeyPEM = " \n " })}, - {"empty NATSURL", mk(func(in *K3sServerInput) { in.NATSURL = "" })}, - {"empty NATSToken", mk(func(in *K3sServerInput) { in.NATSToken = "" })}, - {"empty NATSCACert", mk(func(in *K3sServerInput) { in.NATSCACert = " \n" })}, + {"empty GatewayURL", mk(func(in *K3sServerInput) { in.GatewayURL = "" })}, + {"empty AccessKey", mk(func(in *K3sServerInput) { in.AccessKey = "" })}, + {"empty SecretKey", mk(func(in *K3sServerInput) { in.SecretKey = "" })}, + {"empty GatewayCACert", mk(func(in *K3sServerInput) { in.GatewayCACert = " \n" })}, } for _, tc := range cases { t.Run(tc.name, func(t *testing.T) { @@ -269,9 +271,11 @@ func TestLaunchK3sServerVM_UserDataContainsAllArtifacts(t *testing.T) { // The eks-node-role first-boot selector keys off this to start control-plane // services; without it the unified AMI boots into no role. assert.Contains(t, udata, "SPINIFEX_K3S_ROLE=server") - assert.Contains(t, udata, "SPINIFEX_NATS_URL=nats://localhost:4222") - assert.Contains(t, udata, "SPINIFEX_NATS_TOKEN=s3cr3t-token") - assert.Contains(t, udata, "SPINIFEX_NATS_CA="+k3sNATSCAPath) + assert.Contains(t, udata, "EKS_GATEWAY_URL=https://10.15.8.1:9999") + assert.Contains(t, udata, "EKS_GATEWAY_CA="+k3sGatewayCAPath) + assert.Contains(t, udata, "EKS_ACCESS_KEY=AKIAEXAMPLE") + assert.Contains(t, udata, "EKS_SECRET_KEY=s3cr3t-key") + assert.Contains(t, udata, "EKS_REGION=us-east-1") assert.Contains(t, udata, "EKS_ACCOUNT_ID=111122223333") assert.Contains(t, udata, "EKS_CLUSTER_NAME=alpha") assert.Contains(t, udata, "EKS_NLB_ENDPOINT=https://eks-alpha-lb-001.us-east-1.elb.spinifex.local:443") @@ -302,7 +306,7 @@ func TestLaunchK3sServerVM_UserDataContainsAllArtifacts(t *testing.T) { // webhook never invoked. assert.Contains(t, udata, "authentication-token-webhook-config-file="+k3sTokenWebhookKubeconfigPath) - assert.Contains(t, udata, "path: "+k3sNATSCAPath) + assert.Contains(t, udata, "path: "+k3sGatewayCAPath) assert.Contains(t, udata, "-----BEGIN CERTIFICATE-----") assert.Contains(t, udata, "FAKECA") diff --git a/spinifex/handlers/eks/service_impl.go b/spinifex/handlers/eks/service_impl.go index 10d740dbc..f891dce2a 100644 --- a/spinifex/handlers/eks/service_impl.go +++ b/spinifex/handlers/eks/service_impl.go @@ -41,12 +41,17 @@ type EKSServiceDeps struct { Region string HolderID string - // NATS auth handed to the K3s server VM so it can publish its one-shot - // bootstrap messages. Shared token + CA (PEM content) + a VM-reachable - // URL — spinifex has no per-principal nkeys hierarchy to scope against. - NATSURL string - NATSToken string - NATSCACert string + // Gateway broker config handed to the K3s server VM so it can publish its + // bootstrap envelopes + state reports via SigV4-signed HTTPS POST to the AWS + // gateway (the ELBv2 lb-agent model) instead of dialing core NATS. + // SystemGatewayURL is the mgmt-reachable AWSGW endpoint (distinct from + // GatewayBaseURL, which is the OIDC issuer host); SystemAccessKey/SecretKey + // are the system (Predastore) SigV4 creds; GatewayCACert is the PEM that + // signs the gateway server cert. + SystemGatewayURL string + SystemAccessKey string + SystemSecretKey string + GatewayCACert string VPCSG sgProvisioner VPCK3s k3sVPCProvisioner @@ -392,9 +397,10 @@ func (s *EKSServiceImpl) CreateCluster(input *eks.CreateClusterInput, accountID, OIDCIssuer: oidcIssuer, OIDCPrivateKeyPEM: privPEM, OIDCPublicKeyPEM: pubPEM, - NATSURL: s.deps.NATSURL, - NATSToken: s.deps.NATSToken, - NATSCACert: s.deps.NATSCACert, + GatewayURL: s.deps.SystemGatewayURL, + AccessKey: s.deps.SystemAccessKey, + SecretKey: s.deps.SystemSecretKey, + GatewayCACert: s.deps.GatewayCACert, }) if err != nil { s.markFailed(acctKV, name) diff --git a/spinifex/handlers/eks/token_review.go b/spinifex/handlers/eks/token_review.go new file mode 100644 index 000000000..748b76a3b --- /dev/null +++ b/spinifex/handlers/eks/token_review.go @@ -0,0 +1,92 @@ +package handlers_eks + +import ( + "fmt" + "log/slog" + "time" + + "github.com/mulgadc/spinifex/spinifex/utils" + "github.com/nats-io/nats.go" +) + +// WebhookTokenReviewResult is the gateway's reply to the eks-token-webhook's +// TokenReview relay: the resolved K8s identity for an `aws eks get-token` +// bearer token, or Authenticated=false. The webhook maps it into the +// authentication.k8s.io/v1 TokenReview the apiserver expects. The locationName +// tags let the gateway's restjson marshaler (jsonutil) emit the same keys the +// json tags name, so the webhook decodes the body with encoding/json. +type WebhookTokenReviewResult struct { + Authenticated bool `json:"authenticated" locationName:"authenticated"` + Username string `json:"username,omitempty" locationName:"username"` + UID string `json:"uid,omitempty" locationName:"uid"` + Groups []string `json:"groups,omitempty" locationName:"groups"` +} + +// Authenticate is the pure TokenReview decision core: decode token → STS verify +// → AccessEntry lookup → identity. Any failure yields Authenticated=false (never +// an error) 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 without NATS. +func Authenticate( + token string, + verify func(presignedURL string) (*TokenVerifyResponse, error), + lookup func(principalARN string) (*AccessEntryRecord, error), +) WebhookTokenReviewResult { + presignedURL, err := DecodeGetToken(token) + if err != nil { + return WebhookTokenReviewResult{Authenticated: false} + } + ident, err := verify(presignedURL) + if err != nil { + slog.Debug("token verify rejected", "err", err) + return WebhookTokenReviewResult{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 WebhookTokenReviewResult{Authenticated: false} + } + uid := ident.UserID + if uid == "" { + uid = ident.ARN + } + return WebhookTokenReviewResult{ + Authenticated: true, + Username: rec.KubernetesUsername, + UID: uid, + Groups: rec.KubernetesGroups, + } +} + +// ResolveTokenReview runs the TokenReview decision host-side, wiring the real +// STS verify (NATS request to the awsgw-hosted verify subject) and AccessEntry +// KV read into Authenticate. The eks-token-webhook reaches this through the +// gateway broker (POST /clusters/{name}/token-review) so the control-plane VM +// never speaks NATS. Returns an error only for a genuine infrastructure fault +// (no JetStream / missing account bucket); a forged token resolves to an +// Authenticated=false result, not an error. +func ResolveTokenReview(nc *nats.Conn, accountID, clusterName, token string, verifyTimeout time.Duration) (WebhookTokenReviewResult, error) { + if nc == nil { + return WebhookTokenReviewResult{}, fmt.Errorf("eks: ResolveTokenReview nil nats conn") + } + js, err := nc.JetStream() + if err != nil { + return WebhookTokenReviewResult{}, fmt.Errorf("eks: jetstream: %w", err) + } + kv, err := js.KeyValue(AccountBucketName(accountID)) + if err != nil { + return WebhookTokenReviewResult{}, fmt.Errorf("eks: open account KV %s: %w", accountID, err) + } + + verify := func(presignedURL string) (*TokenVerifyResponse, error) { + return utils.NATSRequest[TokenVerifyResponse]( + nc, TokenVerifySubject, + TokenVerifyRequest{PresignedURL: presignedURL, ClusterName: clusterName}, + verifyTimeout, "") + } + lookup := func(principalARN string) (*AccessEntryRecord, error) { + return GetAccessEntryRecord(kv, clusterName, principalARN) + } + return Authenticate(token, verify, lookup), nil +} diff --git a/spinifex/handlers/eks/token_review_test.go b/spinifex/handlers/eks/token_review_test.go new file mode 100644 index 000000000..4d29d0e02 --- /dev/null +++ b/spinifex/handlers/eks/token_review_test.go @@ -0,0 +1,156 @@ +package handlers_eks + +import ( + "encoding/base64" + "encoding/json" + "errors" + "testing" + "time" + + "github.com/mulgadc/spinifex/spinifex/testutil" + "github.com/nats-io/nats.go" + "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) (*TokenVerifyResponse, error) { + return &TokenVerifyResponse{ + AccountID: "111122223333", + ARN: testARN, + UserID: "AROAEXAMPLE:session", + PrincipalType: "AssumedRole", + }, nil +} + +func TestAuthenticate_GrantsWhenEntryExists(t *testing.T) { + lookup := func(arn string) (*AccessEntryRecord, error) { + assert.Equal(t, testARN, arn) + return &AccessEntryRecord{ + KubernetesUsername: testARN, + KubernetesGroups: []string{"system:masters"}, + }, nil + } + + res := Authenticate(validToken("https://sts.amazonaws.com/?Action=GetCallerIdentity"), okVerify, lookup) + + require.True(t, res.Authenticated) + assert.Equal(t, testARN, res.Username) + assert.Equal(t, "AROAEXAMPLE:session", res.UID) + assert.Equal(t, []string{"system:masters"}, res.Groups) +} + +func TestAuthenticate_DeniesMalformedToken(t *testing.T) { + called := false + verify := func(string) (*TokenVerifyResponse, error) { called = true; return nil, nil } + lookup := func(string) (*AccessEntryRecord, error) { return nil, nil } + + res := Authenticate("not-a-k8s-aws-token", verify, lookup) + + assert.False(t, res.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) (*TokenVerifyResponse, error) { + return nil, errors.New("signature mismatch") + } + lookup := func(string) (*AccessEntryRecord, error) { + t.Fatal("lookup must not run when verify fails") + return nil, nil + } + + res := Authenticate(validToken("https://sts/?x=1"), verify, lookup) + assert.False(t, res.Authenticated) +} + +func TestAuthenticate_DeniesWhenNoAccessEntry(t *testing.T) { + lookup := func(string) (*AccessEntryRecord, error) { + return nil, ErrAccessEntryNotFound + } + + res := Authenticate(validToken("https://sts/?x=1"), okVerify, lookup) + assert.False(t, res.Authenticated) + assert.Empty(t, res.Username) +} + +func TestAuthenticate_FallsBackUIDToARN(t *testing.T) { + verify := func(string) (*TokenVerifyResponse, error) { + return &TokenVerifyResponse{ARN: testARN}, nil // no UserID + } + lookup := func(string) (*AccessEntryRecord, error) { + return &AccessEntryRecord{KubernetesUsername: testARN, KubernetesGroups: []string{"system:masters"}}, nil + } + + res := Authenticate(validToken("https://sts/?x=1"), verify, lookup) + require.True(t, res.Authenticated) + assert.Equal(t, testARN, res.UID) +} + +func TestResolveTokenReview_NilConn(t *testing.T) { + _, err := ResolveTokenReview(nil, "111122223333", "alpha", "tok", time.Second) + require.Error(t, err) +} + +// A genuine infra fault (account bucket never created) is an error, not a +// silent Authenticated=false — the webhook turns it into a retryable 5xx. +func TestResolveTokenReview_MissingBucketErrors(t *testing.T) { + _, nc, _ := testutil.StartTestJetStream(t) + _, err := ResolveTokenReview(nc, "111122223333", "alpha", validToken("https://sts/?x=1"), time.Second) + require.Error(t, err) +} + +func TestResolveTokenReview_AuthenticatesViaVerifyAndKV(t *testing.T) { + _, nc, js := testutil.StartTestJetStream(t) + kv := testutil.SeedKV(t, js, AccountBucketName("111122223333"), nil) + require.NoError(t, PutAccessEntryRecord(kv, &AccessEntryRecord{ + ClusterName: "alpha", + PrincipalARN: testARN, + KubernetesUsername: testARN, + KubernetesGroups: []string{"system:masters"}, + Type: AccessEntryTypeStandard, + })) + + // Stand in for the awsgw-hosted STS verify responder. + sub, err := nc.Subscribe(TokenVerifySubject, func(m *nats.Msg) { + resp, _ := json.Marshal(TokenVerifyResponse{ + AccountID: "111122223333", + ARN: testARN, + UserID: "AROAEXAMPLE:session", + PrincipalType: "AssumedRole", + }) + _ = m.Respond(resp) + }) + require.NoError(t, err) + t.Cleanup(func() { _ = sub.Unsubscribe() }) + + res, err := ResolveTokenReview(nc, "111122223333", "alpha", validToken("https://sts/?Action=GetCallerIdentity"), 2*time.Second) + require.NoError(t, err) + require.True(t, res.Authenticated) + assert.Equal(t, testARN, res.Username) + assert.Equal(t, "AROAEXAMPLE:session", res.UID) + assert.Equal(t, []string{"system:masters"}, res.Groups) +} + +// A valid IAM principal with no AccessEntry resolves to Authenticated=false +// (a clean 401), not an error. +func TestResolveTokenReview_NoAccessEntryDenies(t *testing.T) { + _, nc, js := testutil.StartTestJetStream(t) + testutil.SeedKV(t, js, AccountBucketName("111122223333"), nil) + + sub, err := nc.Subscribe(TokenVerifySubject, func(m *nats.Msg) { + resp, _ := json.Marshal(TokenVerifyResponse{AccountID: "111122223333", ARN: testARN}) + _ = m.Respond(resp) + }) + require.NoError(t, err) + t.Cleanup(func() { _ = sub.Unsubscribe() }) + + res, err := ResolveTokenReview(nc, "111122223333", "alpha", validToken("https://sts/?x=1"), 2*time.Second) + require.NoError(t, err) + assert.False(t, res.Authenticated) +} diff --git a/tests/e2e/Makefile b/tests/e2e/Makefile index a2b0242f0..301612feb 100644 --- a/tests/e2e/Makefile +++ b/tests/e2e/Makefile @@ -1,4 +1,4 @@ -PKG ?= ./cert/... ./lb/... ./multinode/... ./baremetal/... ./reboot/... ./tofu/... +PKG ?= ./cert/... ./lb/... ./eks/... ./multinode/... ./baremetal/... ./reboot/... ./tofu/... ARTIFACT_DIR ?= /tmp/spinifex-e2e-artifacts TIMEOUT ?= 30m MODE ?= single @@ -11,9 +11,9 @@ SUITE_TAG ?= $(notdir $(patsubst %/...,%,$(firstword $(PKG)))) # named suite into BIN_DIR/.test so CI can ship binaries (not source + # go toolchain) to the remote VM. BIN_DIR ?= $(CURDIR)/_bin -SUITES_BUILD ?= cert lb single gpu +SUITES_BUILD ?= cert lb eks single gpu -.PHONY: e2e e2e-cert e2e-lb e2e-single e2e-multinode e2e-bm e2e-reboot e2e-tofu e2e-gpu e2e-all clean e2e-build +.PHONY: e2e e2e-cert e2e-lb e2e-eks e2e-single e2e-multinode e2e-bm e2e-reboot e2e-tofu e2e-gpu e2e-all clean e2e-build ifeq ($(GOTESTSUM),1) E2E_CMD = gotestsum \ @@ -34,6 +34,7 @@ e2e: e2e-cert: ; $(MAKE) e2e PKG=./cert/... e2e-lb: ; $(MAKE) e2e PKG=./lb/... +e2e-eks: ; $(MAKE) e2e PKG=./eks/... e2e-single: ; $(MAKE) e2e PKG=./single/... e2e-multinode: ; $(MAKE) e2e PKG=./multinode/... MODE=multinode e2e-bm: ; $(MAKE) e2e PKG=./baremetal/... MODE=baremetal diff --git a/tests/e2e/eks/eks_test.go b/tests/e2e/eks/eks_test.go new file mode 100644 index 000000000..375d53233 --- /dev/null +++ b/tests/e2e/eks/eks_test.go @@ -0,0 +1,404 @@ +//go:build e2e + +package eks + +import ( + "encoding/base64" + "errors" + "fmt" + "os" + "os/exec" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/aws/awserr" + "github.com/aws/aws-sdk-go/service/ec2" + "github.com/aws/aws-sdk-go/service/eks" + "github.com/aws/aws-sdk-go/service/sts" + "github.com/mulgadc/spinifex/tests/e2e/harness" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +const ( + eksVPCCIDR = "10.210.0.0/16" + eksSubnetCIDR = "10.210.1.0/24" + eksClusterPfx = "eks-e2e" + getTokenTpl = "k8s-aws-v1." +) + +// TestEKS drives the EKS control-plane lifecycle against the local awsgw +// endpoint: a real VPC/subnet, CreateCluster → ACTIVE (which boots the K3s +// server VM + NLB), kubeconfig artifact assembly, AccessEntry CRUD, get-token, +// and DeleteCluster → gone. Phase 1 is API-level only (no kubectl); the +// kubectl/IRSA data-plane subtests land in Phase 2 once apiserver reachability +// from the runner is wired. +// +// One cluster fixture is shared across the subtests (create once, delete in +// Cleanup) — control-plane boot is the slowest step, so re-creating per subtest +// would blow the suite timeout on dev nodes. +func TestEKS(t *testing.T) { + env := harness.LoadEnv(t) + artifacts := harness.ArtifactDir(t, env) + c := harness.NewAWSClient(t, env) + + fx := setupClusterFixture(t, c, env, artifacts) + + t.Run("CreateCluster", func(t *testing.T) { + cl := harness.WaitForEKSClusterActive(t, c, fx.ClusterName) + assert.Equal(t, eks.ClusterStatusActive, aws.StringValue(cl.Status)) + assert.NotEmpty(t, aws.StringValue(cl.Endpoint), "ACTIVE cluster must expose an endpoint") + require.NotNil(t, cl.CertificateAuthority, "ACTIVE cluster must expose certificateAuthority") + assert.NotEmpty(t, aws.StringValue(cl.CertificateAuthority.Data), "certificateAuthority.data must be populated") + ca, err := base64.StdEncoding.DecodeString(aws.StringValue(cl.CertificateAuthority.Data)) + require.NoError(t, err, "certificateAuthority.data must be base64") + assert.Contains(t, string(ca), "BEGIN CERTIFICATE", "CA data must be a PEM cert") + fx.Cluster = cl + }) + + t.Run("DescribeKubeconfigArtifacts", func(t *testing.T) { + requireClusterReady(t, fx) + path := writeKubeconfig(t, artifacts, fx.Cluster) + raw, err := os.ReadFile(path) //nolint:gosec // artifact path built by the test + require.NoError(t, err) + kc := string(raw) + assert.Contains(t, kc, "server: "+aws.StringValue(fx.Cluster.Endpoint), "kubeconfig server = cluster endpoint") + assert.Contains(t, kc, "certificate-authority-data:", "kubeconfig embeds the CA") + assert.Contains(t, kc, "command: aws", "kubeconfig exec block shells aws") + assert.Contains(t, kc, "get-token", "kubeconfig exec block calls eks get-token") + assert.Contains(t, kc, fx.ClusterName, "kubeconfig exec block targets this cluster") + }) + + t.Run("AccessEntry", func(t *testing.T) { + requireClusterReady(t, fx) + // CreateCluster already seeds a system:masters AccessEntry for the caller + // principal (BootstrapClusterCreatorAdminPermissions), so exercise the API + // against a *distinct* principal to avoid ResourceInUse on the bootstrap one. + principal := fmt.Sprintf("arn:aws:iam::%s:role/%s-extra", fx.AccountID, fx.ClusterName) + _, err := c.EKS.CreateAccessEntry(&eks.CreateAccessEntryInput{ + ClusterName: aws.String(fx.ClusterName), + PrincipalArn: aws.String(principal), + KubernetesGroups: aws.StringSlice([]string{"system:masters"}), + Username: aws.String("e2e-extra-admin"), + }) + require.NoError(t, err, "create-access-entry") + t.Cleanup(func() { + _, _ = c.EKS.DeleteAccessEntry(&eks.DeleteAccessEntryInput{ + ClusterName: aws.String(fx.ClusterName), + PrincipalArn: aws.String(principal), + }) + }) + + list, err := c.EKS.ListAccessEntries(&eks.ListAccessEntriesInput{ClusterName: aws.String(fx.ClusterName)}) + require.NoError(t, err, "list-access-entries") + assert.Contains(t, aws.StringValueSlice(list.AccessEntries), principal, "new entry must appear in list") + + desc, err := c.EKS.DescribeAccessEntry(&eks.DescribeAccessEntryInput{ + ClusterName: aws.String(fx.ClusterName), + PrincipalArn: aws.String(principal), + }) + require.NoError(t, err, "describe-access-entry") + require.NotNil(t, desc.AccessEntry) + assert.Equal(t, principal, aws.StringValue(desc.AccessEntry.PrincipalArn)) + assert.Equal(t, "e2e-extra-admin", aws.StringValue(desc.AccessEntry.Username)) + assert.Contains(t, aws.StringValueSlice(desc.AccessEntry.KubernetesGroups), "system:masters") + }) + + t.Run("GetToken", func(t *testing.T) { + requireClusterReady(t, fx) + token := awsEKSGetToken(t, env, fx.ClusterName) + require.True(t, strings.HasPrefix(token, getTokenTpl), "token must carry the %q prefix, got %.16q", getTokenTpl, token) + + raw, err := base64.RawURLEncoding.DecodeString(strings.TrimPrefix(token, getTokenTpl)) + require.NoError(t, err, "token body must be base64url(presigned URL)") + url := string(raw) + assert.Contains(t, url, "Action=GetCallerIdentity", "presigned URL must call GetCallerIdentity") + assert.Contains(t, url, "X-Amz-Signature=", "presigned URL must be signed") + assert.Contains(t, strings.ToLower(url), "x-k8s-aws-id", "presigned URL must pin the cluster via x-k8s-aws-id") + }) + + t.Run("KubectlGetNodes", func(t *testing.T) { + requireClusterReady(t, fx) + // Reachability shim: the kubeconfig endpoint (NLB DNS) is unresolvable + // from here, so point kubectl at the control-plane VM's mgmt IP:6443 + // directly. The serving cert SANs the NLB DNS + node IPs, not the mgmt + // IP, so skip TLS verification — auth still flows through the get-token + // webhook, which is what this asserts. + mgmtIP := harness.ControlPlaneMgmtIP(t, env, fx.AccountID, fx.ClusterName) + kcPath := writeKubectlKubeconfig(t, artifacts, fx.Cluster, mgmtIP) + kc := harness.NewKubectl(t, kcPath, getTokenEnv(t, env)) + + // Auth + reachability: poll until the apiserver answers and reports a + // Ready node. A 401 here means the get-token webhook chain regressed. + harness.EventuallyErr(t, func() error { + out, err := kc.Run(30*time.Second, "get", "nodes", + "-o", `jsonpath={range .items[*]}{.metadata.name}{"="}{.status.conditions[?(@.type=="Ready")].status}{"\n"}{end}`) + if err != nil { + return fmt.Errorf("kubectl get nodes: %v\n%s", err, out) + } + if !strings.Contains(out, "=True") { + return fmt.Errorf("no Ready node yet:\n%s", out) + } + return nil + }, 3*time.Minute, 5*time.Second) + + out, _ := kc.Run(30*time.Second, "get", "nodes", "-o", "wide") + t.Logf("kubectl get nodes:\n%s", out) + }) + + t.Run("DeleteCluster", func(t *testing.T) { + requireClusterReady(t, fx) + _, err := c.EKS.DeleteCluster(&eks.DeleteClusterInput{Name: aws.String(fx.ClusterName)}) + require.NoError(t, err, "delete-cluster") + harness.WaitForEKSClusterDeleted(t, c, fx.ClusterName) + fx.Deleted = true + }) +} + +// --- Fixture -------------------------------------------------------------- + +type clusterFixture struct { + ClusterName string + AccountID string + VPCID string + SubnetID string + Cluster *eks.Cluster + Deleted bool +} + +func setupClusterFixture(t *testing.T, c *harness.AWSClient, env *harness.Env, artifacts string) *clusterFixture { + t.Helper() + fx := &clusterFixture{ClusterName: fmt.Sprintf("%s-%d", eksClusterPfx, time.Now().Unix())} + + harness.Phase(t, "Resolving caller account") + ident, err := c.STS.GetCallerIdentity(&sts.GetCallerIdentityInput{}) + require.NoError(t, err, "sts get-caller-identity") + fx.AccountID = aws.StringValue(ident.Account) + t.Logf("account: %s", fx.AccountID) + + harness.Phase(t, "Creating VPC topology (%s)", eksVPCCIDR) + createVPC(t, c, fx) + t.Cleanup(func() { deleteVPC(t, c, fx) }) + createSubnet(t, c, fx) + t.Cleanup(func() { deleteSubnet(t, c, fx) }) + + harness.Phase(t, "Creating cluster %q", fx.ClusterName) + roleArn := fmt.Sprintf("arn:aws:iam::%s:role/%s-role", fx.AccountID, fx.ClusterName) + _, err = c.EKS.CreateCluster(&eks.CreateClusterInput{ + Name: aws.String(fx.ClusterName), + RoleArn: aws.String(roleArn), + ResourcesVpcConfig: &eks.VpcConfigRequest{ + SubnetIds: aws.StringSlice([]string{fx.SubnetID}), + }, + }) + require.NoError(t, err, "create-cluster") + t.Cleanup(func() { deleteClusterBestEffort(t, c, fx) }) + + harness.OnFailure(t, func() { dumpCluster(t, c, artifacts, fx.ClusterName) }) + return fx +} + +func requireClusterReady(t *testing.T, fx *clusterFixture) { + t.Helper() + if fx.Cluster == nil { + t.Skip("cluster never reached ACTIVE (CreateCluster subtest failed)") + } +} + +// --- VPC / Subnet --------------------------------------------------------- + +func createVPC(t *testing.T, c *harness.AWSClient, fx *clusterFixture) { + t.Helper() + out, err := c.EC2.CreateVpc(&ec2.CreateVpcInput{CidrBlock: aws.String(eksVPCCIDR)}) + require.NoError(t, err, "create-vpc") + fx.VPCID = aws.StringValue(out.Vpc.VpcId) + t.Logf("VPC: %s", fx.VPCID) +} + +func deleteVPC(t *testing.T, c *harness.AWSClient, fx *clusterFixture) { + if fx.VPCID == "" { + return + } + if _, err := c.EC2.DeleteVpc(&ec2.DeleteVpcInput{VpcId: aws.String(fx.VPCID)}); err != nil { + t.Logf("delete VPC %s: %v", fx.VPCID, err) + } +} + +func createSubnet(t *testing.T, c *harness.AWSClient, fx *clusterFixture) { + t.Helper() + out, err := c.EC2.CreateSubnet(&ec2.CreateSubnetInput{ + VpcId: aws.String(fx.VPCID), + CidrBlock: aws.String(eksSubnetCIDR), + }) + require.NoError(t, err, "create-subnet") + fx.SubnetID = aws.StringValue(out.Subnet.SubnetId) + t.Logf("subnet: %s", fx.SubnetID) +} + +func deleteSubnet(t *testing.T, c *harness.AWSClient, fx *clusterFixture) { + if fx.SubnetID == "" { + return + } + if _, err := c.EC2.DeleteSubnet(&ec2.DeleteSubnetInput{SubnetId: aws.String(fx.SubnetID)}); err != nil { + t.Logf("delete subnet %s: %v", fx.SubnetID, err) + } +} + +// deleteClusterBestEffort tears the cluster down if a subtest left it standing +// (the DeleteCluster subtest already removed it on the happy path). VPC/subnet +// cleanup must wait for the cluster's NLB + VM to release them, so this runs +// before the VPC/subnet Cleanups (Cleanups run LIFO; this is registered last). +func deleteClusterBestEffort(t *testing.T, c *harness.AWSClient, fx *clusterFixture) { + if fx.Deleted { + return + } + if _, err := c.EKS.DeleteCluster(&eks.DeleteClusterInput{Name: aws.String(fx.ClusterName)}); err != nil { + var aerr awserr.Error + if errors.As(err, &aerr) && strings.Contains(aerr.Code(), "ResourceNotFound") { + return + } + t.Logf("cleanup delete-cluster %s: %v", fx.ClusterName, err) + return + } + harness.WaitForEKSClusterDeleted(t, c, fx.ClusterName) +} + +// --- kubeconfig artifact -------------------------------------------------- + +// writeKubeconfig assembles the kubeconfig `aws eks update-kubeconfig` would +// produce from DescribeCluster output and writes it to the artifact dir. The +// suite asserts the structure (server, CA, exec block) rather than shelling out +// to update-kubeconfig so the assertion is hermetic. +func writeKubeconfig(t *testing.T, artifacts string, cl *eks.Cluster) string { + t.Helper() + name := aws.StringValue(cl.Name) + kc := fmt.Sprintf(`apiVersion: v1 +kind: Config +clusters: +- name: %[1]s + cluster: + server: %[2]s + certificate-authority-data: %[3]s +contexts: +- name: %[1]s + context: + cluster: %[1]s + user: %[1]s +current-context: %[1]s +users: +- name: %[1]s + user: + exec: + apiVersion: client.authentication.k8s.io/v1beta1 + command: aws + args: + - eks + - get-token + - --cluster-name + - %[1]s +`, name, aws.StringValue(cl.Endpoint), aws.StringValue(cl.CertificateAuthority.Data)) + + path := filepath.Join(artifacts, "kubeconfig-"+name+".yaml") + require.NoError(t, os.WriteFile(path, []byte(kc), 0o600), "write kubeconfig") + t.Logf("kubeconfig: %s", path) + return path +} + +// writeKubectlKubeconfig assembles a kubeconfig pointed at the control-plane +// VM's reachable mgmt IP (Phase-2 shim) with TLS verification disabled, and an +// exec block that mints a bearer token via `aws eks get-token`. Distinct from +// writeKubeconfig (the Phase-1 artifact built from the real NLB endpoint). +func writeKubectlKubeconfig(t *testing.T, artifacts string, cl *eks.Cluster, mgmtIP string) string { + t.Helper() + name := aws.StringValue(cl.Name) + server := fmt.Sprintf("https://%s:6443", mgmtIP) + kc := fmt.Sprintf(`apiVersion: v1 +kind: Config +clusters: +- name: %[1]s + cluster: + server: %[2]s + insecure-skip-tls-verify: true +contexts: +- name: %[1]s + context: + cluster: %[1]s + user: %[1]s +current-context: %[1]s +users: +- name: %[1]s + user: + exec: + apiVersion: client.authentication.k8s.io/v1beta1 + command: aws + args: + - eks + - get-token + - --cluster-name + - %[1]s +`, name, server) + + path := filepath.Join(artifacts, "kubectl-"+name+".yaml") + require.NoError(t, os.WriteFile(path, []byte(kc), 0o600), "write kubectl kubeconfig") + t.Logf("kubectl kubeconfig: %s (server=%s)", path, server) + return path +} + +// --- get-token ------------------------------------------------------------ + +// awsEKSGetToken shells the stock AWS CLI `aws eks get-token`, which presigns an +// STS GetCallerIdentity URL client-side. Inherits the caller env (AWS_PROFILE, +// endpoint, CA bundle) so it signs against the same awsgw the SDK clients use. +func awsEKSGetToken(t *testing.T, env *harness.Env, cluster string) string { + t.Helper() + cmd := exec.Command("aws", "eks", "get-token", "--cluster-name", cluster, "--output", "text") + cmd.Env = getTokenEnv(t, env) + out, err := cmd.CombinedOutput() + require.NoErrorf(t, err, "aws eks get-token:\n%s", out) + // `--output text` prints the token in the last whitespace-separated field. + fields := strings.Fields(string(out)) + require.NotEmpty(t, fields, "get-token produced no output") + return fields[len(fields)-1] +} + +func getTokenEnv(t *testing.T, env *harness.Env) []string { + t.Helper() + e := os.Environ() + if os.Getenv("AWS_PROFILE") == "" { + e = append(e, "AWS_PROFILE="+envOr("SPINIFEX_AWS_PROFILE", "spinifex")) + } + if os.Getenv("AWS_REGION") == "" { + e = append(e, "AWS_REGION="+envOr("SPINIFEX_AWS_REGION", "ap-southeast-2")) + } + // Trust the spinifex CA for the presign STS call unless the profile already + // carries a ca_bundle / the run opted into insecure mode. + if os.Getenv("AWS_CA_BUNDLE") == "" && os.Getenv("SPINIFEX_AWS_INSECURE") != "1" { + if ca, err := harness.ResolveCACert(env); err == nil { + e = append(e, "AWS_CA_BUNDLE="+ca) + } + } + return e +} + +func envOr(key, def string) string { + if v := os.Getenv(key); v != "" { + return v + } + return def +} + +// --- diagnostics ---------------------------------------------------------- + +func dumpCluster(t *testing.T, c *harness.AWSClient, artifacts, name string) { + out, err := c.EKS.DescribeCluster(&eks.DescribeClusterInput{Name: aws.String(name)}) + if err != nil { + t.Logf("dumpCluster describe %s: %v", name, err) + return + } + path := filepath.Join(artifacts, "cluster-"+name+".txt") + _ = os.WriteFile(path, []byte(out.Cluster.String()), 0o600) + t.Logf("cluster dump: %s (status=%s)", path, aws.StringValue(out.Cluster.Status)) +} diff --git a/tests/e2e/harness/aws.go b/tests/e2e/harness/aws.go index 76d3ba85f..531c94864 100644 --- a/tests/e2e/harness/aws.go +++ b/tests/e2e/harness/aws.go @@ -14,6 +14,7 @@ import ( "github.com/aws/aws-sdk-go/aws/credentials" "github.com/aws/aws-sdk-go/aws/session" "github.com/aws/aws-sdk-go/service/ec2" + "github.com/aws/aws-sdk-go/service/eks" "github.com/aws/aws-sdk-go/service/elbv2" "github.com/aws/aws-sdk-go/service/iam" "github.com/aws/aws-sdk-go/service/sts" @@ -25,6 +26,7 @@ import ( type AWSClient struct { EC2 *ec2.EC2 ELBv2 *elbv2.ELBV2 + EKS *eks.EKS IAM *iam.IAM STS *sts.STS } @@ -125,6 +127,7 @@ func newAWSClient(t *testing.T, env *Env, accessKey, secretKey, sessionToken str return &AWSClient{ EC2: ec2.New(sess), ELBv2: elbv2.New(sess), + EKS: eks.New(sess), IAM: iam.New(sess), STS: sts.New(sess), } @@ -137,6 +140,7 @@ func (c *AWSClient) IgnoreCertErrors() { hc := &http.Client{Transport: tr} c.EC2.Config.HTTPClient = hc c.ELBv2.Config.HTTPClient = hc + c.EKS.Config.HTTPClient = hc c.IAM.Config.HTTPClient = hc c.STS.Config.HTTPClient = hc _ = (*x509.CertPool)(nil) // silence unused-import on hardened paths diff --git a/tests/e2e/harness/eks.go b/tests/e2e/harness/eks.go new file mode 100644 index 000000000..2e8105264 --- /dev/null +++ b/tests/e2e/harness/eks.go @@ -0,0 +1,73 @@ +//go:build e2e + +package harness + +import ( + "errors" + "fmt" + "strings" + "testing" + "time" + + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/aws/awserr" + "github.com/aws/aws-sdk-go/service/eks" +) + +// WaitForEKSClusterActive polls DescribeCluster until status == ACTIVE. Control +// plane provisioning boots a K3s server VM + NLB, so default timeout is 10min. +func WaitForEKSClusterActive(t *testing.T, c *AWSClient, name string, opts ...PollOpt) *eks.Cluster { + t.Helper() + cfg := applyOpts(pollCfg{timeout: 10 * time.Minute, interval: 5 * time.Second}, opts...) + var last *eks.Cluster + EventuallyErr(t, func() error { + out, err := c.EKS.DescribeCluster(&eks.DescribeClusterInput{Name: aws.String(name)}) + if err != nil { + return fmt.Errorf("describe-cluster %s: %w", name, err) + } + last = out.Cluster + state := aws.StringValue(last.Status) + switch state { + case eks.ClusterStatusActive: + return nil + case eks.ClusterStatusFailed: + return fmt.Errorf("%s entered FAILED state", name) + default: + return fmt.Errorf("%s status=%s want=ACTIVE", name, state) + } + }, cfg.timeout, cfg.interval) + t.Logf("cluster %s reached status ACTIVE", name) + return last +} + +// WaitForEKSClusterDeleted polls DescribeCluster until it returns +// ResourceNotFoundException (cluster fully gone). +func WaitForEKSClusterDeleted(t *testing.T, c *AWSClient, name string, opts ...PollOpt) { + t.Helper() + cfg := applyOpts(pollCfg{timeout: 10 * time.Minute, interval: 5 * time.Second}, opts...) + EventuallyErr(t, func() error { + _, err := c.EKS.DescribeCluster(&eks.DescribeClusterInput{Name: aws.String(name)}) + if err == nil { + return fmt.Errorf("%s still exists", name) + } + if isClusterNotFound(err) { + return nil + } + return fmt.Errorf("describe-cluster %s: %w", name, err) + }, cfg.timeout, cfg.interval) + t.Logf("cluster %s deleted", name) +} + +// isClusterNotFound reports whether err is the EKS "cluster gone" error. The +// SDK constant is "ResourceNotFoundException" but awsgw emits the doubled +// "ResourceNotFoundExceptionException" on the wire, so match by substring. +func isClusterNotFound(err error) bool { + if err == nil { + return false + } + var aerr awserr.Error + if errors.As(err, &aerr) { + return strings.Contains(aerr.Code(), "ResourceNotFound") + } + return false +} diff --git a/tests/e2e/harness/eks_mgmt.go b/tests/e2e/harness/eks_mgmt.go new file mode 100644 index 000000000..fc031da87 --- /dev/null +++ b/tests/e2e/harness/eks_mgmt.go @@ -0,0 +1,98 @@ +//go:build e2e + +package harness + +import ( + "os" + "path/filepath" + "strings" + "testing" + + "github.com/mulgadc/spinifex/spinifex/config" + handlers_eks "github.com/mulgadc/spinifex/spinifex/handlers/eks" + "github.com/mulgadc/spinifex/spinifex/utils" +) + +// ControlPlaneMgmtIP resolves the EKS control-plane VM's host-bridge (br-mgmt) +// address — the only address reachable from the host/runner. The cluster +// endpoint in DescribeCluster is the internal NLB DNS, which is NXDOMAIN here +// and whose :443->:6443 passthrough is not wired; the VPC NIC (172.31.x) is +// likewise unreachable. The mgmt IP is persisted unencrypted on the cluster's +// KV meta record (ClusterMeta.ControlPlaneMgmtIP), so read it directly over +// NATS. This is the Phase-2 reachability shim — see the eks-6g-e2e plan; the +// real fix (NLB/Route53 passthrough) is tracked separately. +func ControlPlaneMgmtIP(t *testing.T, env *Env, accountID, clusterName string) string { + t.Helper() + host, token, ca := natsConn(t, env) + nc, err := utils.ConnectNATS(host, token, ca) + if err != nil { + t.Fatalf("connect NATS %s: %v", host, err) + } + defer nc.Close() + + js, err := nc.JetStream() + if err != nil { + t.Fatalf("jetstream: %v", err) + } + kv, err := js.KeyValue(handlers_eks.AccountBucketName(accountID)) + if err != nil { + t.Fatalf("open EKS account KV for %s: %v", accountID, err) + } + meta, err := handlers_eks.GetClusterMeta(kv, clusterName) + if err != nil { + t.Fatalf("get cluster meta %s: %v", clusterName, err) + } + if meta.ControlPlaneMgmtIP == "" { + t.Fatalf("cluster %s has no ControlPlaneMgmtIP recorded", clusterName) + } + t.Logf("control-plane mgmt IP: %s", meta.ControlPlaneMgmtIP) + return meta.ControlPlaneMgmtIP +} + +// natsConn resolves the NATS host/token/CA for the harness's own JetStream +// reads. SPINIFEX_NATS_* env overrides win (CI / runner-resident runs); else it +// reads the local node's stanza from spinifex.toml. +func natsConn(t *testing.T, env *Env) (host, token, ca string) { + t.Helper() + host = os.Getenv("SPINIFEX_NATS_URL") + token = os.Getenv("SPINIFEX_NATS_TOKEN") + ca = os.Getenv("SPINIFEX_NATS_CA") + if host != "" { + return dialableNATSHost(host), token, ca + } + + cfgPath := filepath.Join(env.ConfigDir, "spinifex.toml") + cc, err := config.LoadConfig(cfgPath) + if err != nil { + t.Fatalf("load %s for NATS settings: %v", cfgPath, err) + } + node := nodeConfig(cc) + if node == nil { + t.Fatalf("no node stanza with a NATS host in %s", cfgPath) + } + return dialableNATSHost(node.NATS.Host), node.NATS.ACL.Token, node.NATS.CACert +} + +// dialableNATSHost rewrites a wildcard bind address to loopback. NATS.Host is a +// bind address (commonly 0.0.0.0:4222); the TLS serving cert SANs the loopback +// and real node IPs but not 0.0.0.0, so dial 127.0.0.1 instead. +func dialableNATSHost(host string) string { + return strings.Replace(host, "0.0.0.0", "127.0.0.1", 1) +} + +// nodeConfig returns the local node's Config (cc.Node), falling back to the +// first node carrying a NATS host. +func nodeConfig(cc *config.ClusterConfig) *config.Config { + if cc == nil { + return nil + } + if n, ok := cc.Nodes[cc.Node]; ok && n.NATS.Host != "" { + return &n + } + for _, n := range cc.Nodes { + if n.NATS.Host != "" { + return &n + } + } + return nil +} diff --git a/tests/e2e/harness/kubectl.go b/tests/e2e/harness/kubectl.go new file mode 100644 index 000000000..63941a11f --- /dev/null +++ b/tests/e2e/harness/kubectl.go @@ -0,0 +1,67 @@ +//go:build e2e + +package harness + +import ( + "bytes" + "os" + "os/exec" + "testing" + "time" +) + +// Kubectl wraps `kubectl` invocations against a specific kubeconfig for the EKS +// data-plane scenarios. AWSEnv carries the credential-plugin environment +// (AWS_PROFILE / endpoint / CA) that the kubeconfig's `aws eks get-token` exec +// block needs — kubectl spawns it with this process's environment. +type Kubectl struct { + Bin string + Kubeconfig string + AWSEnv []string +} + +// NewKubectl locates kubectl on PATH (KUBECTL_BIN overrides) and SKIPS the +// calling test if it is absent — the Phase-2 data-plane subtests are optional +// where kubectl is not installed. +func NewKubectl(t *testing.T, kubeconfig string, awsEnv []string) *Kubectl { + t.Helper() + bin := os.Getenv("KUBECTL_BIN") + if bin == "" { + bin = "kubectl" + } + resolved, err := exec.LookPath(bin) + if err != nil { + t.Skipf("kubectl not found on PATH (%v); skipping data-plane subtest", err) + } + return &Kubectl{Bin: resolved, Kubeconfig: kubeconfig, AWSEnv: awsEnv} +} + +// Run executes `kubectl --kubeconfig ` and returns combined +// output. Errors are returned (not fatal) so callers can poll readiness. +func (k *Kubectl) Run(timeout time.Duration, args ...string) (string, error) { + full := append([]string{"--kubeconfig", k.Kubeconfig}, args...) + cmd := exec.Command(k.Bin, full...) //nolint:gosec // bin is LookPath-resolved, args test-controlled + env := os.Environ() + env = append(env, k.AWSEnv...) + cmd.Env = env + var buf bytes.Buffer + cmd.Stdout = &buf + cmd.Stderr = &buf + done := make(chan error, 1) + if err := cmd.Start(); err != nil { + return "", err + } + go func() { done <- cmd.Wait() }() + select { + case err := <-done: + return buf.String(), err + case <-time.After(timeout): + _ = cmd.Process.Kill() + <-done + return buf.String(), &timeoutError{} + } +} + +type timeoutError struct{} + +func (*timeoutError) Error() string { return "kubectl timed out" }