fix(control-plane): delegation mints per-request JWT for caller's tenant

The initial delegation layer used Control Plane's long-lived admin
service JWT to authenticate forwarded requests. That's wrong for
reads: Core's query_events (and Query Service's auth pipeline)
resolves tenant from the JWT claims, so every forwarded query was
scoped to tenant="system" (CP's admin identity) and our injected
tenant_id query param was ignored. Users got cross-tenant reads —
including other users' events and the "default" tenant's seed data.

Live repro against the initial deploy:
  POST /api/v1/events → event lands at tenant="default" (Core
    v0.18.x hardcoded; since-fix landed via b08aa1a + redeploy)
  GET  /api/v1/events/query → returns 100 events from
    tenant="default", not the caller's tenant

Fix: AuthClient.SignDelegationJWT(userID, tenantID, role) mints a
short-lived (60s) HS256 JWT that represents the authenticated
caller. delegationClient stores the signer as a func value — not a
single static token — and each forward call takes a bearer param.

ProxyIngestSingle / ProxyIngestBatch / ProxyEventsQuery now:
  1. Pull user_id + tenant_id + role from AuthContext (set by
     AuthMiddleware after JWT or ask_ validation).
  2. Inject tenant into body (writes) or query param (reads).
  3. Mint a per-request JWT with the caller's identity.
  4. Forward with Authorization: Bearer <per-request JWT>.

Backends (Core, Query Service) now see the real caller's tenant and
enforce isolation as they always would. No backend changes needed.

Post-redeploy smoke test (allsource-control-plane.fly.dev):
  - register → ask_ key with tenant=final-{ts}-at-test-local
  - POST /events with body.tenant_id="SPOOFED" → returns event_id
  - GET /events/query?event_type=final.{ts} → count=1, event has
    tenant_id="final-{ts}-at-test-local" (spoofed value overwritten,
    cross-tenant reads impossible)

Also requires allsource-core redeploy to pick up the tenant_id-from-
body fix that landed in main (commit b08aa1a) but shipped with
v0.18.0 on Fly. Done in parallel with this commit.

Tests: delegation tests updated to assert the per-request JWT shape
(stub signer echoes userID/tenantID/role into the forwarded
Authorization header for easy assertion). Full CP suite green.

Author: decebal

apps/control-plane/auth.go

@@ -101,6 +101,31 @@ func NewAuthClient(jwtSecret, coreURL string) *AuthClient {
 	}
 }
 
+// SignDelegationJWT mints a short-lived JWT scoped to a single data-plane
+// request. The delegation layer uses this so backends (Core, Query Service)
+// see the authenticated caller's tenant + role, not Control Plane's admin
+// service identity — tenant isolation works end-to-end without every
+// backend needing to validate ask_ keys directly.
+//
+// 60 second TTL is comfortably longer than any single forwarded request but
+// short enough that a leaked token is near-useless.
+func (a *AuthClient) SignDelegationJWT(userID, tenantID string, role entities.Role) (string, error) {
+	now := time.Now()
+	claims := &Claims{
+		UserID:   userID,
+		TenantID: tenantID,
+		Role:     role,
+		IsAPIKey: true,
+		StandardClaims: jwt.StandardClaims{
+			ExpiresAt: now.Add(60 * time.Second).Unix(),
+			IssuedAt:  now.Unix(),
+			Issuer:    "allsource",
+			Subject:   userID,
+		},
+	}
+	return jwt.NewWithClaims(jwt.SigningMethodHS256, claims).SignedString([]byte(a.jwtSecret))
+}
+
 // SignAPIKey generates a long-lived JWT API key for a given tenant and role.
 // Used by agent registration, onboarding, and demo flows.
 func (a *AuthClient) SignAPIKey(tenantID, username string, role entities.Role) (string, error) {

apps/control-plane/delegation.go

@@ -19,21 +19,24 @@ import (
 	"strings"
 
 	"github.com/gin-gonic/gin"
+
+	"github.com/allsource/control-plane/internal/domain/entities"
 )
 
 // delegationClient holds the resolved backend URLs and the shared HTTP
-// client used to forward requests. serviceToken is Control Plane's long-lived
-// admin JWT — backends trust it and authorize the request on CP's behalf.
+// client used to forward requests. signToken mints a per-request JWT for
+// the authenticated caller — NOT a Control Plane admin token — so backends
+// see the real tenant and role and enforce isolation accordingly.
 type delegationClient struct {
 	core         *url.URL
 	queryService *url.URL
-	serviceToken string
+	signToken    func(userID, tenantID string, role entities.Role) (string, error)
 	http         *http.Client
 }
 
-func newDelegationClient(coreURL, queryURL, serviceToken string, httpClient *http.Client) (*delegationClient, error) {
-	if serviceToken == "" {
-		return nil, errors.New("delegation requires serviceToken")
+func newDelegationClient(coreURL, queryURL string, signToken func(userID, tenantID string, role entities.Role) (string, error), httpClient *http.Client) (*delegationClient, error) {
+	if signToken == nil {
+		return nil, errors.New("delegation requires signToken")
 	}
 	core, err := url.Parse(strings.TrimRight(coreURL, "/"))
 	if err != nil {
@@ -49,7 +52,7 @@ func newDelegationClient(coreURL, queryURL, serviceToken string, httpClient *htt
 	return &delegationClient{
 		core:         core,
 		queryService: qs,
-		serviceToken: serviceToken,
+		signToken:    signToken,
 		http:         httpClient,
 	}, nil
 }
@@ -70,6 +73,26 @@ func authTenantFromContext(c *gin.Context) (string, bool) {
 	return s, true
 }
 
+// authIdentityFromContext returns the three identity fields delegation needs
+// to mint a per-request JWT: user ID, tenant ID, and role. AuthMiddleware
+// sets these after validating either a JWT or an ask_ API key, so a
+// successful context lookup means the caller is authenticated.
+func authIdentityFromContext(c *gin.Context) (userID, tenantID string, role entities.Role, ok bool) {
+	tenantID, ok = authTenantFromContext(c)
+	if !ok {
+		return
+	}
+	uid, uidOk := c.Get("auth_user_id")
+	r, rOk := c.Get("auth_role")
+	if !uidOk || !rOk {
+		ok = false
+		return
+	}
+	uidStr, _ := uid.(string)
+	roleVal, _ := r.(entities.Role)
+	return uidStr, tenantID, roleVal, true
+}
+
 // injectTenantIntoObject overwrites the top-level tenant_id field in a JSON
 // object. Used for single-event ingest where the Core handler reads
 // req.tenant_id.
@@ -109,10 +132,11 @@ func injectTenantIntoBatch(body []byte, tenantID string) ([]byte, error) {
 }
 
 // forwardRequest builds a new outgoing request against backend, copies the
-// body, attaches the service JWT, and copies the backend's response to the
-// client. It does NOT forward any inbound auth headers — the backend trusts
-// Control Plane, not the caller.
-func (d *delegationClient) forwardRequest(c *gin.Context, method string, backend *url.URL, path string, query url.Values, body []byte) {
+// body, attaches the per-request bearer JWT (minted from the authenticated
+// caller's identity), and copies the backend's response to the client. It
+// does NOT forward the inbound Authorization header — the backend trusts
+// the JWT we signed, not any ask_ key the client presented.
+func (d *delegationClient) forwardRequest(c *gin.Context, method string, backend *url.URL, path string, query url.Values, body []byte, bearerToken string) {
 	target := *backend
 	target.Path = strings.TrimRight(backend.Path, "/") + "/" + strings.TrimLeft(path, "/")
 	if query != nil {
@@ -129,7 +153,7 @@ func (d *delegationClient) forwardRequest(c *gin.Context, method string, backend
 		c.JSON(http.StatusInternalServerError, gin.H{"error": "delegation build request", "message": err.Error()})
 		return
 	}
-	req.Header.Set("Authorization", "Bearer "+d.serviceToken)
+	req.Header.Set("Authorization", "Bearer "+bearerToken)
 	if body != nil {
 		req.Header.Set("Content-Type", "application/json")
 	}
@@ -153,12 +177,30 @@ func (d *delegationClient) forwardRequest(c *gin.Context, method string, backend
 	}
 }
 
+// mintBearer builds a per-request JWT that represents the authenticated
+// caller. Backends validate this JWT against the shared JWT_SECRET and get
+// the right tenant + role — which is how tenant isolation survives the hop
+// through Control Plane.
+func (cp *ControlPlane) mintBearer(c *gin.Context) (string, bool) {
+	userID, tenantID, role, ok := authIdentityFromContext(c)
+	if !ok {
+		c.JSON(http.StatusUnauthorized, gin.H{"error": "unauthorized", "message": "no auth context"})
+		return "", false
+	}
+	token, err := cp.delegation.signToken(userID, tenantID, role)
+	if err != nil {
+		c.JSON(http.StatusInternalServerError, gin.H{"error": "delegation sign", "message": err.Error()})
+		return "", false
+	}
+	return token, true
+}
+
 // ProxyIngestSingle forwards POST /api/v1/events to Core with tenant_id
 // injected into the JSON body from the authenticated caller.
 func (cp *ControlPlane) ProxyIngestSingle(c *gin.Context) {
-	tenantID, ok := authTenantFromContext(c)
+	_, tenantID, _, ok := authIdentityFromContext(c)
 	if !ok {
-		c.JSON(http.StatusUnauthorized, gin.H{"error": "unauthorized", "message": "no tenant context"})
+		c.JSON(http.StatusUnauthorized, gin.H{"error": "unauthorized", "message": "no auth context"})
 		return
 	}
 	body, err := io.ReadAll(c.Request.Body)
@@ -171,15 +213,19 @@ func (cp *ControlPlane) ProxyIngestSingle(c *gin.Context) {
 		c.JSON(http.StatusBadRequest, gin.H{"error": "bad request", "message": err.Error()})
 		return
 	}
-	cp.delegation.forwardRequest(c, http.MethodPost, cp.delegation.core, "/api/v1/events", nil, rewritten)
+	bearer, ok := cp.mintBearer(c)
+	if !ok {
+		return
+	}
+	cp.delegation.forwardRequest(c, http.MethodPost, cp.delegation.core, "/api/v1/events", nil, rewritten, bearer)
 }
 
 // ProxyIngestBatch forwards POST /api/v1/events/batch to Core with tenant_id
 // injected onto every event in the batch.
 func (cp *ControlPlane) ProxyIngestBatch(c *gin.Context) {
-	tenantID, ok := authTenantFromContext(c)
+	_, tenantID, _, ok := authIdentityFromContext(c)
 	if !ok {
-		c.JSON(http.StatusUnauthorized, gin.H{"error": "unauthorized", "message": "no tenant context"})
+		c.JSON(http.StatusUnauthorized, gin.H{"error": "unauthorized", "message": "no auth context"})
 		return
 	}
 	body, err := io.ReadAll(c.Request.Body)
@@ -192,19 +238,30 @@ func (cp *ControlPlane) ProxyIngestBatch(c *gin.Context) {
 		c.JSON(http.StatusBadRequest, gin.H{"error": "bad request", "message": err.Error()})
 		return
 	}
-	cp.delegation.forwardRequest(c, http.MethodPost, cp.delegation.core, "/api/v1/events/batch", nil, rewritten)
+	bearer, ok := cp.mintBearer(c)
+	if !ok {
+		return
+	}
+	cp.delegation.forwardRequest(c, http.MethodPost, cp.delegation.core, "/api/v1/events/batch", nil, rewritten, bearer)
 }
 
-// ProxyEventsQuery forwards GET /api/v1/events/query to Query Service with
-// tenant_id injected as a query param. Any client-supplied tenant_id is
-// overwritten — the authenticated caller's tenant is authoritative.
+// ProxyEventsQuery forwards GET /api/v1/events/query to Core with tenant_id
+// injected as a query param AND a per-request JWT scoped to the caller's
+// tenant + role. Core's query_events enforces tenant from auth, which with
+// the user-scoped JWT matches what we're asking for. Any client-supplied
+// tenant_id in the URL is overwritten — the authenticated caller's tenant
+// is authoritative.
 func (cp *ControlPlane) ProxyEventsQuery(c *gin.Context) {
-	tenantID, ok := authTenantFromContext(c)
+	_, tenantID, _, ok := authIdentityFromContext(c)
 	if !ok {
-		c.JSON(http.StatusUnauthorized, gin.H{"error": "unauthorized", "message": "no tenant context"})
+		c.JSON(http.StatusUnauthorized, gin.H{"error": "unauthorized", "message": "no auth context"})
 		return
 	}
 	q := c.Request.URL.Query()
 	q.Set("tenant_id", tenantID)
-	cp.delegation.forwardRequest(c, http.MethodGet, cp.delegation.queryService, "/api/v1/events/query", q, nil)
+	bearer, ok := cp.mintBearer(c)
+	if !ok {
+		return
+	}
+	cp.delegation.forwardRequest(c, http.MethodGet, cp.delegation.core, "/api/v1/events/query", q, nil, bearer)
 }

apps/control-plane/delegation_test.go

@@ -48,9 +48,16 @@ func newFakeBackend(respBy func(w http.ResponseWriter, r *http.Request)) (*fakeB
 // newTestCP builds a minimal ControlPlane with only the pieces the delegation
 // handlers use. It skips AuthMiddleware; tests call handlers directly with a
 // pre-populated gin.Context to isolate delegation behaviour.
+//
+// The token signer is deterministic for test assertions — it echoes the
+// inputs so tests can read tenant/user/role out of the forwarded Authorization
+// header without JWT parsing.
 func newTestCP(t *testing.T, coreURL, qsURL string) *ControlPlane {
 	t.Helper()
-	d, err := newDelegationClient(coreURL, qsURL, "test-service-token", http.DefaultClient)
+	signer := func(userID, tenantID string, role entities.Role) (string, error) {
+		return "test-token:" + userID + ":" + tenantID + ":" + string(role), nil
+	}
+	d, err := newDelegationClient(coreURL, qsURL, signer, http.DefaultClient)
 	if err != nil {
 		t.Fatalf("newDelegationClient: %v", err)
 	}
@@ -63,6 +70,8 @@ func callHandler(h gin.HandlerFunc, req *http.Request, tenantID string) *httptes
 	c.Request = req
 	if tenantID != "" {
 		c.Set("auth_tenant_id", tenantID)
+		c.Set("auth_user_id", "user-"+tenantID)
+		c.Set("auth_role", entities.RoleDeveloper)
 	}
 	h(c)
 	return w
@@ -87,8 +96,8 @@ func TestProxyIngestSingle_InjectsTenantAndForwardsToCore(t *testing.T) {
 	if core.path != "/api/v1/events" {
 		t.Errorf("upstream path: got %q, want /api/v1/events", core.path)
 	}
-	if core.auth != "Bearer test-service-token" {
-		t.Errorf("upstream auth: got %q, want service JWT", core.auth)
+	if core.auth != "Bearer test-token:user-tenant-real:tenant-real:developer" {
+		t.Errorf("upstream auth: got %q, want per-request JWT scoped to caller", core.auth)
 	}
 
 	var forwarded map[string]any
@@ -178,8 +187,8 @@ func TestProxyEventsQuery_InjectsTenantAsQueryParam(t *testing.T) {
 	if got := qs.query.Get("since"); got != "2026-04-17T00:00:00Z" {
 		t.Errorf("upstream since: got %q, want original value", got)
 	}
-	if qs.auth != "Bearer test-service-token" {
-		t.Errorf("upstream auth: got %q, want service JWT", qs.auth)
+	if qs.auth != "Bearer test-token:user-tenant-real:tenant-real:developer" {
+		t.Errorf("upstream auth: got %q, want per-request JWT scoped to caller", qs.auth)
 	}
 }
 

apps/control-plane/main.go

@@ -261,7 +261,7 @@ func NewControlPlane(ctx context.Context) (*ControlPlane, error) {
 	if queryURL == "" {
 		queryURL = "http://allsource-query.internal:3902"
 	}
-	delegation, err := newDelegationClient(coreURL, queryURL, serviceToken, NewPooledHTTPClient())
+	delegation, err := newDelegationClient(coreURL, queryURL, authClient.SignDelegationJWT, NewPooledHTTPClient())
 	if err != nil {
 		return nil, fmt.Errorf("init delegation client: %w", err)
 	}