fix(eks): bind internal report publishing to the caller's cluster

PublishInternal was authorised by IAM policy alone, with its resource
built from the body accountId and the URL cluster. The control-plane
role grants it on "*", so any control-plane VM session, any tenant
holding eks:*, or static system credentials could publish cluster
state, bootstrap and add-on reports for another cluster or account.

PublishInternal now passes the existing internal-route gate: the caller
must be a control-plane instance-role session that is a member of the
URL cluster in the body account. The gate and the handler decode the
body through one shared decoder, so the account checked is the account
the subject is built from. Every unauthorised target returns the same
AccessDenied, and invalid account or cluster names are treated as not
found. Valid publishes keep their subjects and payload bytes.

spinifex/gateway/eks.go

@@ -34,8 +34,9 @@ var eksRoutes = []eksRoute{
 		Handler: func(ctx context.Context, gw *GatewayConfig, acct, callerARN string, p []string, b []byte) (any, error) {
 			return gateway_eks.ListClusters(ctx, gw.NATSConn, acct)
 		}},
-	// Control-plane VM broker: relays bootstrap/state POSTs onto eks.bus.*/eks.state.* NATS subjects.
-	// acct and callerARN are ignored; cluster account comes from the body.
+	// Control-plane VM broker: relays bootstrap/state/addon POSTs onto eks.bus/state/addon subjects.
+	// acct and callerARN are ignored; the cluster account comes from the body, and
+	// AuthorizeInternal binds that account and cluster to the caller's own.
 	{Method: "POST", Pattern: "/clusters/{clusterName}/internal-publish", Action: "PublishInternal",
 		Handler: func(ctx context.Context, gw *GatewayConfig, acct, callerARN string, p []string, b []byte) (any, error) {
 			return gateway_eks.PublishInternal(ctx, gw.NATSConn, p[0], b)
@@ -210,15 +211,6 @@ func (gw *GatewayConfig) EKS_Request(w http.ResponseWriter, r *http.Request) err
 		return errors.New(awserrors.ErrorInternalError)
 	}
 
-	// Ahead of the policy check: the internal routes name the target account in
-	// the path, so an eks:* grant evaluates as permitted and only the principal
-	// class plus the caller's own instance say whether that account is its own.
-	if gateway_eks.IsInternalAction(action) {
-		if err := gateway_eks.AuthorizeInternal(r.Context(), gw.NATSConn, action, eksCaller(r), params); err != nil {
-			return err
-		}
-	}
-
 	body, err := readBoundedBody(r)
 	if err != nil {
 		slog.ErrorContext(r.Context(), "EKS_Request: failed to read body", "err", err)
@@ -239,6 +231,15 @@ func (gw *GatewayConfig) EKS_Request(w http.ResponseWriter, r *http.Request) err
 		}
 	}
 
+	// Ahead of the policy check: the internal routes name the target account in
+	// the path or body, so an eks:* grant evaluates as permitted and only the
+	// principal class plus the caller's own instance say whether it is its own.
+	if gateway_eks.IsInternalAction(action) {
+		if err := gateway_eks.AuthorizeInternal(r.Context(), gw.NATSConn, action, eksCaller(r), params, body); err != nil {
+			return err
+		}
+	}
+
 	// After the body read: CreateCluster and the other create actions name their
 	// resource there rather than in the path.
 	resources, err := gateway_eks.ResourceARNs(action, gw.Region, accountID, params, body)

spinifex/gateway/eks/authz.go

@@ -29,6 +29,7 @@ const (
 	sourceCluster
 	sourceClusterFromBody
 	sourceInternalCluster
+	sourceInternalBodyCluster
 	sourceBodyAccountCluster
 	sourceNodegroup
 	sourceNodegroupFromBody
@@ -56,7 +57,7 @@ var eksScopes = map[string][]resourceSource{
 	"GetRecoveryDirective": {sourceInternalCluster},
 	// The owning account arrives in the body on these two, and the ARN names
 	// that account for the same reason: it is the account the handler acts in.
-	"PublishInternal":    {sourceBodyAccountCluster},
+	"PublishInternal":    {sourceInternalBodyCluster},
 	"WebhookTokenReview": {sourceBodyAccountCluster},
 
 	// Nodegroups. Create evaluates the cluster and the nodegroup it is about
@@ -152,7 +153,7 @@ func resolve(source resourceSource, action, region, accountID string, params []s
 	case sourceInternalCluster:
 		return clusterARN(region, param(params, 1), param(params, 0)), nil
 
-	case sourceBodyAccountCluster:
+	case sourceInternalBodyCluster, sourceBodyAccountCluster:
 		scope, err := bodyscope.Parse(action, body)
 		if err != nil {
 			return "", errors.New(awserrors.ErrorInvalidParameterValue)

spinifex/gateway/eks/internal_authz.go

@@ -31,17 +31,20 @@ type Caller struct {
 }
 
 // IsInternalAction reports whether action is one of the internal CP-VM routes:
-// the ones naming the customer account in the path, which no customer grant may
-// reach. Derived from eksScopes so a new such route cannot ship ungated — that
-// table is exhaustive by contract against the dispatch table.
+// the ones naming the customer account in the path or body, which no customer
+// grant may reach. Derived from eksScopes so a new such route cannot ship
+// ungated — that table is exhaustive by contract against the dispatch table.
 func IsInternalAction(action string) bool {
-	return slices.Contains(eksScopes[action], sourceInternalCluster)
+	return slices.ContainsFunc(eksScopes[action], func(source resourceSource) bool {
+		return source == sourceInternalCluster || source == sourceInternalBodyCluster
+	})
 }
 
 // AuthorizeInternal is the principal gate for the internal CP-VM routes, run
 // ahead of the policy check. params are the route captures: cluster name,
-// account ID, and for GetRecoveryDirective the member's instance ID.
-func AuthorizeInternal(ctx context.Context, natsConn *nats.Conn, action string, caller Caller, params []string) error {
+// account ID, and for GetRecoveryDirective the member's instance ID. body is
+// the request body the handler will read; PublishInternal names its account there.
+func AuthorizeInternal(ctx context.Context, natsConn *nats.Conn, action string, caller Caller, params []string, body []byte) error {
 	if !IsInternalAction(action) {
 		return nil
 	}
@@ -49,9 +52,9 @@ func AuthorizeInternal(ctx context.Context, natsConn *nats.Conn, action string,
 		return err
 	}
 
-	clusterName, accountID := param(params, 0), param(params, 1)
-	if clusterName == "" || accountID == "" {
-		return errors.New(awserrors.ErrorInvalidParameterValue)
+	clusterName, accountID, err := internalTarget(action, params, body)
+	if err != nil {
+		return err
 	}
 
 	// A member reads its own directive and no other's: the path segment is only
@@ -79,6 +82,24 @@ func AuthorizeInternal(ctx context.Context, natsConn *nats.Conn, action string,
 	return nil
 }
 
+// internalTarget reads the cluster and owning account an internal route acts
+// on, from the same place its handler reads them, so the binding checks the
+// account the handler then uses.
+func internalTarget(action string, params []string, body []byte) (clusterName, accountID string, err error) {
+	clusterName, accountID = param(params, 0), param(params, 1)
+	if slices.Contains(eksScopes[action], sourceInternalBodyCluster) {
+		req, decodeErr := decodeInternalPublish(body)
+		if decodeErr != nil {
+			return "", "", errors.New(awserrors.ErrorInvalidParameterValue)
+		}
+		accountID = req.AccountID
+	}
+	if clusterName == "" || accountID == "" {
+		return "", "", errors.New(awserrors.ErrorInvalidParameterValue)
+	}
+	return clusterName, accountID, nil
+}
+
 // The class check on its own: a session assumed from the CP VM's instance role
 // in the system account. It says the caller is a control-plane VM, not which
 // one — the binding to a cluster is AuthorizeInternal's, and needs NATS.
@@ -97,8 +118,9 @@ func requireCPAgent(ctx context.Context, action string, caller Caller) error {
 
 // Reads JetStream directly, matching the OIDC discovery path, rather than a
 // service round trip through the handler. js.KeyValue still costs a STREAM.INFO
-// for the bucket handle. A nil meta is returned for an account with no clusters
-// and for a cluster that is absent; both are denials rather than failures.
+// for the bucket handle. A nil meta is returned for an account with no clusters,
+// for a cluster that is absent, and for a name no bucket or key can hold; all
+// are denials rather than failures.
 func lookupClusterMeta(ctx context.Context, natsConn *nats.Conn, accountID, clusterName string) (*handlers_eks.ClusterMeta, error) {
 	if natsConn == nil {
 		return nil, errors.New("gateway NATS connection not initialised")
@@ -109,13 +131,13 @@ func lookupClusterMeta(ctx context.Context, natsConn *nats.Conn, accountID, clus
 	}
 	kv, err := js.KeyValue(ctx, handlers_eks.AccountBucketName(accountID))
 	if err != nil {
-		if errors.Is(err, jetstream.ErrBucketNotFound) {
+		if errors.Is(err, jetstream.ErrBucketNotFound) || errors.Is(err, jetstream.ErrInvalidBucketName) {
 			return nil, nil
 		}
 		return nil, err
 	}
 	entry, err := kv.Get(ctx, handlers_eks.ClusterMetaKey(clusterName))
-	if errors.Is(err, jetstream.ErrKeyNotFound) {
+	if errors.Is(err, jetstream.ErrKeyNotFound) || errors.Is(err, jetstream.ErrInvalidKey) {
 		return nil, nil
 	}
 	if err != nil {

spinifex/gateway/eks/internal_authz_test.go

@@ -47,6 +47,10 @@ func seedCluster(t *testing.T, nc *nats.Conn, accountID, cluster string, meta *h
 	require.NoError(t, handlers_eks.PutClusterMeta(t.Context(), kv, meta))
 }
 
+func publishBody(accountID string) []byte {
+	return []byte(`{"accountId":"` + accountID + `","channel":"state","payload":{"healthz":"ok"}}`)
+}
+
 func assertDenied(t *testing.T, err error) {
 	t.Helper()
 	require.Error(t, err)
@@ -88,9 +92,11 @@ func TestAuthorizeInternal_RejectsNonCPPrincipals(t *testing.T) {
 	for _, tc := range cases {
 		t.Run(tc.name, func(t *testing.T) {
 			assertDenied(t, gateway_eks.AuthorizeInternal(t.Context(), nil, "ListInternalAddons",
-				tc.caller, []string{"alpha", tenantAccount}))
+				tc.caller, []string{"alpha", tenantAccount}, nil))
 			assertDenied(t, gateway_eks.AuthorizeInternal(t.Context(), nil, "GetRecoveryDirective",
-				tc.caller, []string{"alpha", tenantAccount, cpInstanceID}))
+				tc.caller, []string{"alpha", tenantAccount, cpInstanceID}, nil))
+			assertDenied(t, gateway_eks.AuthorizeInternal(t.Context(), nil, "PublishInternal",
+				tc.caller, []string{"alpha"}, publishBody(tenantAccount)))
 		})
 	}
 }
@@ -99,7 +105,7 @@ func TestAuthorizeInternal_RejectsNonCPPrincipals(t *testing.T) {
 func TestAuthorizeInternal_IgnoresOtherActions(t *testing.T) {
 	assert.False(t, gateway_eks.IsInternalAction("DescribeCluster"))
 	require.NoError(t, gateway_eks.AuthorizeInternal(t.Context(), nil, "DescribeCluster",
-		gateway_eks.Caller{AccountID: tenantAccount, PrincipalType: "user"}, []string{"alpha"}))
+		gateway_eks.Caller{AccountID: tenantAccount, PrincipalType: "user"}, []string{"alpha"}, nil))
 }
 
 func TestAuthorizeInternal_AllowsClusterMember(t *testing.T) {
@@ -111,9 +117,11 @@ func TestAuthorizeInternal_AllowsClusterMember(t *testing.T) {
 	})
 
 	require.NoError(t, gateway_eks.AuthorizeInternal(t.Context(), nc, "ListInternalAddons",
-		cpAgent(cpInstanceID), []string{"alpha", tenantAccount}))
+		cpAgent(cpInstanceID), []string{"alpha", tenantAccount}, nil))
 	require.NoError(t, gateway_eks.AuthorizeInternal(t.Context(), nc, "GetRecoveryDirective",
-		cpAgent(cpInstanceID), []string{"alpha", tenantAccount, cpInstanceID}))
+		cpAgent(cpInstanceID), []string{"alpha", tenantAccount, cpInstanceID}, nil))
+	require.NoError(t, gateway_eks.AuthorizeInternal(t.Context(), nc, "PublishInternal",
+		cpAgent(cpInstanceID), []string{"alpha"}, publishBody(tenantAccount)))
 }
 
 // Clusters persisted before ControlPlaneNodes existed carry the member in the
@@ -125,7 +133,7 @@ func TestAuthorizeInternal_AllowsLegacyScalarMember(t *testing.T) {
 	})
 
 	require.NoError(t, gateway_eks.AuthorizeInternal(t.Context(), nc, "ListInternalAddons",
-		cpAgent(cpInstanceID), []string{"alpha", tenantAccount}))
+		cpAgent(cpInstanceID), []string{"alpha", tenantAccount}, nil))
 }
 
 // A CP VM that is a genuine agent still cannot name an account or a cluster it
@@ -148,11 +156,15 @@ func TestAuthorizeInternal_RejectsClusterTheCallerDoesNotServe(t *testing.T) {
 		{"an account with no clusters", "999988887777", "alpha"},
 		{"a cluster that does not exist", tenantAccount, "gamma"},
 		{"its own account, another cluster", tenantAccount, "beta"},
+		{"an account no bucket can be named for", tenantAccount + ".x", "alpha"},
+		{"a cluster no key can be named for", tenantAccount, "al*pha"},
 	}
 	for _, tc := range cases {
 		t.Run(tc.name, func(t *testing.T) {
 			assertDenied(t, gateway_eks.AuthorizeInternal(t.Context(), nc, "ListInternalAddons",
-				cpAgent(cpInstanceID), []string{tc.cluster, tc.account}))
+				cpAgent(cpInstanceID), []string{tc.cluster, tc.account}, nil))
+			assertDenied(t, gateway_eks.AuthorizeInternal(t.Context(), nc, "PublishInternal",
+				cpAgent(cpInstanceID), []string{tc.cluster}, publishBody(tc.account)))
 		})
 	}
 }
@@ -167,12 +179,23 @@ func TestAuthorizeInternal_RecoveryIsSelfOnly(t *testing.T) {
 	})
 
 	assertDenied(t, gateway_eks.AuthorizeInternal(t.Context(), nc, "GetRecoveryDirective",
-		cpAgent(cpInstanceID), []string{"alpha", tenantAccount, otherInstance}))
+		cpAgent(cpInstanceID), []string{"alpha", tenantAccount, otherInstance}, nil))
+}
+
+// PublishInternal names its account in the body; one the handler could not read
+// is the caller's validation fault, as an empty path segment is.
+func TestAuthorizeInternal_RejectsUnreadablePublishBody(t *testing.T) {
+	for _, body := range []string{``, `{not-json`, `{"channel":"state"}`, `{"accountId":7}`} {
+		err := gateway_eks.AuthorizeInternal(t.Context(), nil, "PublishInternal", cpAgent(cpInstanceID),
+			[]string{"alpha"}, []byte(body))
+		require.Error(t, err)
+		assert.Equal(t, awserrors.ErrorInvalidParameterValue, awserrors.ValidErrorCodeFromError(err), "body %q", body)
+	}
 }
 
 func TestAuthorizeInternal_RejectsEmptyPathSegments(t *testing.T) {
 	for _, params := range [][]string{{"", tenantAccount}, {"alpha", ""}, {"alpha"}} {
-		err := gateway_eks.AuthorizeInternal(t.Context(), nil, "ListInternalAddons", cpAgent(cpInstanceID), params)
+		err := gateway_eks.AuthorizeInternal(t.Context(), nil, "ListInternalAddons", cpAgent(cpInstanceID), params, nil)
 		require.Error(t, err)
 		assert.Equal(t, awserrors.ErrorInvalidParameterValue, err.Error())
 	}
@@ -183,7 +206,12 @@ func TestAuthorizeInternal_RejectsEmptyPathSegments(t *testing.T) {
 // chasing a permission it already has.
 func TestAuthorizeInternal_NoNATSIsServerInternal(t *testing.T) {
 	err := gateway_eks.AuthorizeInternal(t.Context(), nil, "ListInternalAddons",
-		cpAgent(cpInstanceID), []string{"alpha", tenantAccount})
+		cpAgent(cpInstanceID), []string{"alpha", tenantAccount}, nil)
+	require.Error(t, err)
+	assert.Equal(t, awserrors.ErrorServerInternal, err.Error())
+
+	err = gateway_eks.AuthorizeInternal(t.Context(), nil, "PublishInternal",
+		cpAgent(cpInstanceID), []string{"alpha"}, publishBody(tenantAccount))
 	require.Error(t, err)
 	assert.Equal(t, awserrors.ErrorServerInternal, err.Error())
 }

spinifex/gateway/eks/internal_publish.go

@@ -42,9 +42,23 @@ var validBootstrapKinds = map[string]struct{}{
 	handlers_eks.BootstrapSubjectCA:         {},
 }
 
+// decodeInternalPublish is the one reading of the body that both AuthorizeInternal
+// and PublishInternal use, so the account the gate binds is the account the
+// published subject names.
+func decodeInternalPublish(body []byte) (internalPublishRequest, error) {
+	var req internalPublishRequest
+	if err := json.Unmarshal(body, &req); err != nil {
+		return internalPublishRequest{}, err
+	}
+	if req.AccountID == "" {
+		return internalPublishRequest{}, errors.New("accountId is required")
+	}
+	return req, nil
+}
+
 // PublishInternal — POST /clusters/{name}/internal-publish. Relays a VM
-// publication onto the bootstrap/state NATS subjects via the AWSGW, keeping
-// NATS cluster-internal.
+// publication onto the bootstrap/state/addon NATS subjects via the AWSGW,
+// keeping NATS cluster-internal. AuthorizeInternal has bound the caller first.
 func PublishInternal(ctx context.Context, natsConn *nats.Conn, clusterName string, body []byte) (*publishInternalOutput, error) {
 	if natsConn == nil {
 		return nil, errors.New(awserrors.ErrorServerInternal)
@@ -53,12 +67,12 @@ func PublishInternal(ctx context.Context, natsConn *nats.Conn, clusterName strin
 		return nil, errors.New(awserrors.ErrorInvalidParameterValue)
 	}
 
-	var req internalPublishRequest
-	if err := json.Unmarshal(body, &req); err != nil {
+	req, err := decodeInternalPublish(body)
+	if err != nil {
 		slog.DebugContext(ctx, "PublishInternal: bad body", "cluster", clusterName, "err", err)
 		return nil, errors.New(awserrors.ErrorInvalidParameterValue)
 	}
-	if req.AccountID == "" || len(req.Payload) == 0 {
+	if len(req.Payload) == 0 {
 		return nil, errors.New(awserrors.ErrorInvalidParameterValue)
 	}
 

spinifex/gateway/eks_authz_test.go

@@ -5,12 +5,16 @@ package gateway
 
 import (
 	"context"
+	"errors"
 	awsidentifiers "github.com/mulgadc/spinifex/spinifex/foundation/aws/identifiers"
 	"net/http"
 	"net/http/httptest"
 	"net/url"
 	"strings"
 	"testing"
+	"time"
+
+	eksv1 "github.com/mulgadc/spinifex/contracts/eks/v1"
 
 	"github.com/mulgadc/spinifex/internal/testkit"
 	"github.com/mulgadc/spinifex/spinifex/domains/eks/access"
@@ -18,6 +22,7 @@ import (
 	"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
 	gateway_eks "github.com/mulgadc/spinifex/spinifex/gateway/eks"
 	handlers_eks "github.com/mulgadc/spinifex/spinifex/handlers/eks"
+	"github.com/nats-io/nats.go"
 	"github.com/nats-io/nats.go/jetstream"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -233,3 +238,217 @@ func TestEKSRequest_MissingAccountIDReturnsInternalError(t *testing.T) {
 	require.Error(t, err)
 	assert.Equal(t, awserrors.ErrorInternalError, err.Error())
 }
+
+// publishFixture seeds two clusters in the caller's account and one in another,
+// each served by its own control-plane VM, and subscribes to every EKS subject.
+type publishFixture struct {
+	gw  *GatewayConfig
+	nc  *nats.Conn
+	sub *nats.Subscription
+}
+
+const (
+	publishCPAlpha    = "i-cp00000000000a1"
+	publishCPBeta     = "i-cp00000000000b1"
+	publishCPGamma    = "i-cp00000000000c1"
+	publishOtherAcct  = "444455556666"
+	publishNoClusters = "999988887777"
+)
+
+func newPublishFixture(t *testing.T) *publishFixture {
+	t.Helper()
+	_, nc, js := testutil.StartTestJetStream(t)
+	for _, c := range []struct{ acct, cluster, member string }{
+		{authzAccountID, "alpha", publishCPAlpha},
+		{authzAccountID, "beta", publishCPBeta},
+		{publishOtherAcct, "gamma", publishCPGamma},
+	} {
+		kv, err := handlers_eks.GetOrCreateAccountBucket(t.Context(), js, c.acct)
+		require.NoError(t, err)
+		require.NoError(t, handlers_eks.PutClusterMeta(t.Context(), kv, &handlers_eks.ClusterMeta{
+			Name:              c.cluster,
+			ControlPlaneNodes: []handlers_eks.ControlPlaneNode{{InstanceID: c.member}},
+		}))
+	}
+	sub, err := nc.SubscribeSync("eks.>")
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = sub.Unsubscribe() })
+
+	gw := scopedPolicyGateway(statement("Allow", "eks:*", "*"))
+	gw.NATSConn = nc
+	return &publishFixture{gw: gw, nc: nc, sub: sub}
+}
+
+// publish POSTs body to cluster's internal-publish route under ctx's identity.
+func (f *publishFixture) publish(t *testing.T, ctx context.Context, cluster, body string) (*httptest.ResponseRecorder, error) {
+	t.Helper()
+	req := httptest.NewRequest(http.MethodPost, "/clusters/"+cluster+"/internal-publish", strings.NewReader(body))
+	rec := httptest.NewRecorder()
+	return rec, f.gw.EKS_Request(rec, req.WithContext(ctx))
+}
+
+// requireNothingPublished proves a denial stopped the relay, not just the reply.
+func (f *publishFixture) requireNothingPublished(t *testing.T) {
+	t.Helper()
+	require.NoError(t, f.nc.Flush())
+	msg, err := f.sub.NextMsg(100 * time.Millisecond)
+	require.ErrorIs(t, err, nats.ErrTimeout, "relayed onto %v", msg)
+}
+
+// cpAgentContext is the auth context a CP VM's IMDS credentials produce.
+func cpAgentContext(instanceID string) context.Context {
+	roleName := handlers_eks.CPInstanceRoleName
+	ctx := context.WithValue(context.Background(), ctxService, "eks")
+	ctx = context.WithValue(ctx, ctxAccountID, awsidentifiers.GlobalAccountID)
+	ctx = context.WithValue(ctx, ctxIdentity, instanceID)
+	ctx = context.WithValue(ctx, ctxPrincipalType, principalTypeAssumedRole)
+	ctx = context.WithValue(ctx, ctxUnderlyingRoleARN,
+		"arn:aws:iam::"+awsidentifiers.GlobalAccountID+":role/"+roleName)
+	return context.WithValue(ctx, ctxAssumedRoleARN,
+		"arn:aws:sts::"+awsidentifiers.GlobalAccountID+":assumed-role/"+roleName+"/"+instanceID)
+}
+
+func userContext(accountID, name string) context.Context {
+	ctx := context.WithValue(context.Background(), ctxService, "eks")
+	ctx = context.WithValue(ctx, ctxAccountID, accountID)
+	ctx = context.WithValue(ctx, ctxIdentity, name)
+	return context.WithValue(ctx, ctxPrincipalType, principalTypeUser)
+}
+
+// publishChannel is one body shape PublishInternal relays, and where it lands.
+type publishChannel struct {
+	name    string
+	fields  string
+	subject func(accountID, cluster string) string
+}
+
+func publishChannels() []publishChannel {
+	channels := []publishChannel{
+		{"state", `"channel":"state"`, handlers_eks.StateSubject},
+		{"addon", `"channel":"addon"`, eksv1.AddonStatusSubject},
+	}
+	for _, kind := range []string{
+		handlers_eks.BootstrapSubjectToken, handlers_eks.BootstrapSubjectKubeconfig,
+		handlers_eks.BootstrapSubjectJWKS, handlers_eks.BootstrapSubjectCA,
+	} {
+		channels = append(channels, publishChannel{
+			name:   "bootstrap " + kind,
+			fields: `"channel":"bootstrap","kind":"` + kind + `"`,
+			subject: func(accountID, cluster string) string {
+				return handlers_eks.BootstrapSubject(accountID, cluster, kind)
+			},
+		})
+	}
+	return channels
+}
+
+const publishPayload = `{"healthz":"ok", "node_count":1,"ts":42}`
+
+func publishBody(accountID string, ch publishChannel) string {
+	return `{"accountId":"` + accountID + `",` + ch.fields + `,"payload":` + publishPayload + `}`
+}
+
+// A CP VM reporting for the cluster it serves still reaches its subject, with
+// the payload bytes relayed verbatim, on every channel.
+func TestEKSRequest_PublishInternalRelaysForTheCallersOwnCluster(t *testing.T) {
+	f := newPublishFixture(t)
+	for _, ch := range publishChannels() {
+		t.Run(ch.name, func(t *testing.T) {
+			rec, err := f.publish(t, cpAgentContext(publishCPAlpha), "alpha", publishBody(authzAccountID, ch))
+			require.NoError(t, err)
+			assert.Equal(t, http.StatusOK, rec.Code)
+			assert.JSONEq(t, `{}`, rec.Body.String())
+
+			msg, err := f.sub.NextMsg(2 * time.Second)
+			require.NoError(t, err)
+			assert.Equal(t, ch.subject(authzAccountID, "alpha"), msg.Subject)
+			assert.Equal(t, publishPayload, string(msg.Data))
+			f.requireNothingPublished(t)
+		})
+	}
+}
+
+// The bypass this closes: eks:PublishInternal on "*" let a CP VM report for any
+// cluster in any account. Every target it does not serve is the same denial, so
+// the reply cannot tell an existing cluster from an absent one.
+func TestEKSRequest_PublishInternalDeniesAClusterTheCallerDoesNotServe(t *testing.T) {
+	f := newPublishFixture(t)
+	targets := []struct {
+		name, account, cluster string
+	}{
+		{"its own account, another cluster", authzAccountID, "beta"},
+		{"another account's cluster", publishOtherAcct, "gamma"},
+		{"its own cluster's name in another account", publishOtherAcct, "alpha"},
+		{"a cluster that does not exist", authzAccountID, "delta"},
+		{"an account with no clusters", publishNoClusters, "alpha"},
+		{"an account no bucket can be named for", publishOtherAcct + ".gamma", "alpha"},
+		{"a cluster no key can be named for", authzAccountID, "*"},
+	}
+	for _, ch := range publishChannels() {
+		for _, target := range targets {
+			t.Run(ch.name+"/"+target.name, func(t *testing.T) {
+				_, err := f.publish(t, cpAgentContext(publishCPAlpha), url.PathEscape(target.cluster), publishBody(target.account, ch))
+				require.Error(t, err)
+				assert.Equal(t, awserrors.ErrorAccessDenied, awserrors.ValidErrorCodeFromError(err))
+				assert.Equal(t, errors.New(awserrors.ErrorAccessDenied), err)
+				f.requireNothingPublished(t)
+			})
+		}
+	}
+}
+
+// The account the gate binds is the one the subject is built from: a body that
+// spells the account under another key, or adds a cluster field, moves nothing.
+func TestEKSRequest_PublishInternalRedirectionIsDenied(t *testing.T) {
+	f := newPublishFixture(t)
+	state := publishChannels()[0]
+
+	// URL names the caller's cluster, body names another account.
+	_, err := f.publish(t, cpAgentContext(publishCPAlpha), "alpha", publishBody(publishOtherAcct, state))
+	assertDenied(t, err)
+	f.requireNothingPublished(t)
+
+	// encoding/json folds case, so a re-cased key is the account the handler reads.
+	_, err = f.publish(t, cpAgentContext(publishCPAlpha), "gamma",
+		`{"AccountID":"`+publishOtherAcct+`","channel":"state","payload":{"x":1}}`)
+	assertDenied(t, err)
+	f.requireNothingPublished(t)
+
+	// A repeated key decodes to its last value, which is what the gate binds.
+	_, err = f.publish(t, cpAgentContext(publishCPAlpha), "alpha",
+		`{"accountId":"`+authzAccountID+`","accountId":"`+publishOtherAcct+`","channel":"state","payload":{"x":1}}`)
+	assertDenied(t, err)
+	f.requireNothingPublished(t)
+
+	// The body has no cluster field; the route's cluster is the only one used.
+	_, err = f.publish(t, cpAgentContext(publishCPAlpha), "alpha",
+		`{"accountId":"`+authzAccountID+`","clusterName":"beta","channel":"state","payload":{"x":1}}`)
+	require.NoError(t, err)
+	msg, err := f.sub.NextMsg(2 * time.Second)
+	require.NoError(t, err)
+	assert.Equal(t, handlers_eks.StateSubject(authzAccountID, "alpha"), msg.Subject)
+	f.requireNothingPublished(t)
+}
+
+// Only a CP VM's instance-role session may publish: a tenant holding eks:*, and
+// baked static system credentials with no instance identity, are both refused.
+func TestEKSRequest_PublishInternalRejectsNonCPCallers(t *testing.T) {
+	f := newPublishFixture(t)
+	callers := []struct {
+		name string
+		ctx  context.Context
+	}{
+		{"tenant user, own account", userContext(authzAccountID, "alice")},
+		{"tenant user, another account", userContext(publishOtherAcct, "mallory")},
+		{"static system credentials", userContext(awsidentifiers.GlobalAccountID, "admin")},
+	}
+	for _, ch := range publishChannels() {
+		for _, caller := range callers {
+			t.Run(ch.name+"/"+caller.name, func(t *testing.T) {
+				_, err := f.publish(t, caller.ctx, "alpha", publishBody(authzAccountID, ch))
+				assertDenied(t, err)
+				f.requireNothingPublished(t)
+			})
+		}
+	}
+}

spinifex/handlers/eks/k3s_server_role.go

@@ -35,7 +35,7 @@ func (s *EKSServiceImpl) ensureCPInstanceProfile(accountID string) string {
 	iam := s.iamEnsurer()
 	if iam == nil {
 		slog.Warn("EKS: IAM service unwired; CP VM falls back to baked static gateway creds and "+
-			"will be denied eks:ListInternalAddons and eks:GetRecoveryDirective (no instance identity to bind)",
+			"will be denied eks:PublishInternal, eks:ListInternalAddons and eks:GetRecoveryDirective (no instance identity to bind)",
 			"accountID", accountID)
 		return ""
 	}
@@ -43,7 +43,7 @@ func (s *EKSServiceImpl) ensureCPInstanceProfile(accountID string) string {
 		CPInstanceRoleName, eksServerInlinePolicyName, eksServerInlinePolicy)
 	if err != nil {
 		slog.Error("EKS: ensure CP instance profile failed; CP VM falls back to baked static gateway creds and "+
-			"will be denied eks:ListInternalAddons and eks:GetRecoveryDirective (no instance identity to bind)",
+			"will be denied eks:PublishInternal, eks:ListInternalAddons and eks:GetRecoveryDirective (no instance identity to bind)",
 			"accountID", accountID, "role", CPInstanceRoleName, "err", err)
 		return ""
 	}

tests/integration/cross_tenant_test.go

@@ -427,13 +427,19 @@ func eksUncased() map[string]string {
 		"TagResource":   sameHelper("ListTagsForResource"),
 		"UntagResource": sameHelper("ListTagsForResource"),
 	}
+	// These name the cluster's account in the path or body. AuthorizeInternal
+	// refuses every tenant principal outright and binds a control-plane VM to
+	// the cluster it serves, so no tenant ARN reaches the evaluator.
+	for _, action := range []string{"GetRecoveryDirective", "ListInternalAddons", "PublishInternal"} {
+		uncased[action] = "internal control-plane route: refused to tenants, bound to the caller's own cluster"
+	}
 	for _, action := range []string{
 		"AssociateAccessPolicy", "CreateAccessEntry", "CreateAddon", "CreateCluster",
 		"CreateNodegroup", "DeleteAccessEntry", "DeleteAddon", "DeleteCluster", "DeleteNodegroup",
 		"DescribeAccessEntry", "DescribeAddon", "DescribeAddonVersions", "DescribeCluster",
-		"DescribeNodegroup", "DisassociateAccessPolicy", "GetRecoveryDirective",
+		"DescribeNodegroup", "DisassociateAccessPolicy",
 		"ListAccessEntries", "ListAccessPolicies", "ListAddons", "ListAssociatedAccessPolicies",
-		"ListClusters", "ListInternalAddons", "ListNodegroups", "PublishInternal",
+		"ListClusters", "ListNodegroups",
 		"UpdateAccessEntry", "UpdateAddon", "UpdateNodegroupConfig", "WebhookTokenReview",
 	} {
 		uncased[action] = byName