test(eks): EKS control-plane E2E suite (Sprint 6g) (#304)

* 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://<mgmtIP>: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.
This commit is contained in:
Josh Siviour
2026-06-07 20:42:25 +10:00
committed by GitHub
parent 609471c448
commit 496424c41c
31 changed files with 1981 additions and 589 deletions
+80 -4
View File
@@ -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: |
+134
View File
@@ -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(&region, "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)
}
+74 -110
View File
@@ -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 {
+66 -66
View File
@@ -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) {
+1 -1
View File
@@ -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")
+103
View File
@@ -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
}
+101 -216
View File
@@ -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 <manifest.conf> [--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 <manifest.conf> [--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 \\"
@@ -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
}
+24 -20
View File
@@ -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 <kind-suffix>: 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
+9 -7
View File
@@ -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.
@@ -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: <unix-seconds> }
#
# 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
}
+9 -21
View File
@@ -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 \
+40 -54
View File
@@ -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)
}
+16
View File
@@ -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",
+91
View File
@@ -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
}
@@ -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)
}
+54
View File
@@ -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
}
+86
View File
@@ -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)
}
+7 -2
View File
@@ -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,
+17 -16
View File
@@ -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)
+32 -19
View File
@@ -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
+14 -10
View File
@@ -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")
+15 -9
View File
@@ -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)
+92
View File
@@ -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
}
+156
View File
@@ -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)
}
+4 -3
View File
@@ -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/<suite>.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
+404
View File
@@ -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))
}
+4
View File
@@ -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
+73
View File
@@ -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
}
+98
View File
@@ -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
}
+67
View File
@@ -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 <kc> <args...>` 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" }