control-plane-api, agent: capability-token validity is configurable

crates/agent/src/integration_tests/harness.rs

@@ -1790,6 +1790,8 @@ impl TestHarness {
             self.publisher.clone(),
             snapshot_watch.clone(),
             None,
+            // The production default of the --capability-token-validity setting.
+            std::time::Duration::from_secs(3600),
         ));
 
         self.control_plane_app = Some(app);

crates/agent/src/main.rs

@@ -95,6 +95,17 @@ struct Args {
     )]
     #[arg(value_parser = humantime::parse_duration)]
     heartbeat_timeout: std::time::Duration,
+    /// Validity window of access tokens minted by the `capability_token`
+    /// grant. The default matches the one-hour access tokens of the SQL
+    /// `generate_access_token` mint; raising it widens the exposure of a
+    /// leaked masked token, so overrides warrant care.
+    #[clap(
+        long = "capability-token-validity",
+        env = "CAPABILITY_TOKEN_VALIDITY",
+        default_value = "1h"
+    )]
+    #[arg(value_parser = humantime::parse_duration)]
+    capability_token_validity: std::time::Duration,
 
     #[clap(long = "log-format", env = "LOG_FORMAT", default_value = "json")]
     log_format: LogFormat,
@@ -392,6 +403,7 @@ async fn async_main(args: Args) -> Result<(), anyhow::Error> {
         publisher.clone(),
         snapshot_watch.clone(),
         args.stripe_webhook_secret,
+        args.capability_token_validity,
     ));
     let api_router = control_plane_api::build_router(
         api_app.clone(),

crates/control-plane-api/src/authority.rs

@@ -58,6 +58,29 @@ impl Requirement for NoRequirement {
 /// `internal.user_roles` rather than the snapshot walk — and for operations
 /// which would let a masked bearer escape its mask, such as minting a
 /// full-authority refresh credential.
+///
+/// The audited inventory of unmasked-only surfaces (#3376):
+/// - The `capability_token` grant of `POST /api/v1/auth/token`: a reduced
+///   token must not widen or re-mint itself.
+/// - GraphQL `createRefreshToken`: a refresh token exchanges for a
+///   full-authority access token, so a masked bearer minting one escapes
+///   its mask. The resolver enforces this through
+///   [`Forbidden::require_unmasked`], since an axum extractor cannot reach
+///   a resolver.
+/// - `/admin/create-data-plane` and `/admin/update-l2-reporting`: their
+///   SQL authorization cannot bind the mask, so they fail closed instead.
+///
+/// Everything else needs no unmasked guard. Identity-gated revocation
+/// (`revokeRefreshToken`) deliberately stays open to masked bearers,
+/// because revocation never widens the bearer's authority.
+/// Capability-gated operations — `createApiKey`, `createServiceAccount`,
+/// `revokeApiKey`, and kin — authorize through the grant walk, which the
+/// mask already filters. Invite redemption widens the *user's* grants
+/// while the bearer still exercises them only through its mask. SQL
+/// functions reachable through PostgREST (`public.create_refresh_token`,
+/// `public.gateway_auth_token`) are outside this crate's enforcement
+/// entirely — they are part of the documented PostgREST mask bypass whose
+/// resolution is the #2877 migration, tracked under #3376.
 pub struct RequireUnmasked;
 
 impl Requirement for RequireUnmasked {
@@ -133,6 +156,20 @@ impl Forbidden {
         }
     }
 
+    /// The masked-bearer refusal shared by every surface which demands a
+    /// full-authority credential: a definitive denial keyed on the
+    /// *presence* of the `capability_mask` claim, never its value — a mask
+    /// which happens to enable everything is still a deliberately-reduced
+    /// credential. Requirement evaluation consumes this during extraction;
+    /// GraphQL resolvers consume it directly, where an axum extractor
+    /// cannot reach.
+    pub fn require_unmasked(claims: &crate::ControlClaims) -> Result<(), Self> {
+        if claims.capability_mask.is_some() {
+            return Err(Self::unmasked_token_required());
+        }
+        Ok(())
+    }
+
     pub fn unmasked_token_required() -> Self {
         Self {
             error: "unmasked_token_required",
@@ -304,8 +341,8 @@ fn evaluate_requirement<R: Requirement>(
     };
     let mask = CapabilityMask::from_claim(claims.capability_mask.as_deref());
 
-    if R::REQUIRE_UNMASKED && claims.capability_mask.is_some() {
-        return Err(Rejection::Forbidden(Forbidden::unmasked_token_required()));
+    if R::REQUIRE_UNMASKED {
+        Forbidden::require_unmasked(claims).map_err(Rejection::Forbidden)?;
     }
 
     let required = R::required();

crates/control-plane-api/src/server/create_data_plane.rs

@@ -53,14 +53,14 @@ pub struct Request {
 pub struct Response {}
 
 /// Authorization is SQL `internal.user_roles` rather than the snapshot walk,
-/// so the bearer's capability mask has no effect — a masked bearer is
-/// treated as unmasked — until this endpoint's authorization is refactored
-/// onto the snapshot. See #3376.
+/// so the capability ceiling cannot bind here; `RequireUnmasked` fail-closes
+/// masked bearers instead, until the wider refactor retires this endpoint's
+/// SQL authorization (#3376, decision 12).
 #[axum::debug_handler(state=std::sync::Arc<crate::App>)]
 #[tracing::instrument(skip(app, env), ret, err(Debug, level = tracing::Level::WARN))]
 pub async fn create_data_plane(
     axum::extract::State(app): axum::extract::State<std::sync::Arc<crate::App>>,
-    crate::Authority { envelope: env, .. }: crate::Authority,
+    crate::Authority { envelope: env, .. }: crate::Authority<crate::RequireUnmasked>,
     super::Request(Request {
         name,
         private,
@@ -330,3 +330,62 @@ impl Validate for Category {
         }
     }
 }
+
+#[cfg(test)]
+mod test {
+    use crate::test_server;
+
+    const ALICE: uuid::Uuid = uuid::Uuid::from_bytes([0x11; 16]);
+
+    /// A masked bearer is refused at extraction with the structured `403`,
+    /// while an unmasked bearer reaches the handler's own `ops/` admin gate
+    /// unchanged.
+    #[sqlx::test(
+        migrations = "../../supabase/migrations",
+        fixtures(path = "../fixtures", scripts("data_planes", "alice"))
+    )]
+    async fn test_create_data_plane_requires_unmasked(pool: sqlx::PgPool) {
+        let _guard = test_server::init();
+
+        let server = test_server::TestServer::start(
+            pool.clone(),
+            test_server::snapshot(pool.clone(), true).await,
+        )
+        .await;
+
+        let body = serde_json::json!({"name": "test-plane", "category": "managed"});
+
+        // The mask enables Admin, and is still refused: masked-ness is the
+        // claim's presence, never its value.
+        let masked_token =
+            server.make_masked_access_token(ALICE, Some("alice@example.com"), Some(vec!["Admin"]));
+        let response = server
+            .rest_client()
+            .post("/admin/create-data-plane", &body, Some(&masked_token))
+            .send()
+            .await
+            .unwrap();
+        let (status, text) = (response.status(), response.text().await.unwrap());
+        assert_eq!(status, reqwest::StatusCode::FORBIDDEN, "{text}");
+        insta::assert_snapshot!(
+            text,
+            @r###"{"error":"unmasked_token_required","message":"this operation requires a full-authority token, but the bearer token carries a capability mask","missing_capabilities":[]}"###
+        );
+
+        // An unmasked bearer passes extraction and lands on the handler's
+        // SQL authorization, which refuses alice: she is no `ops/` admin.
+        let unmasked_token = server.make_access_token(ALICE, Some("alice@example.com"));
+        let response = server
+            .rest_client()
+            .post("/admin/create-data-plane", &body, Some(&unmasked_token))
+            .send()
+            .await
+            .unwrap();
+        let (status, text) = (response.status(), response.text().await.unwrap());
+        assert_eq!(status, reqwest::StatusCode::FORBIDDEN, "{text}");
+        assert!(
+            text.contains("not an admin of the 'ops/' tenant"),
+            "the unmasked refusal is the handler's own gate: {text}"
+        );
+    }
+}

crates/control-plane-api/src/server/mod.rs

@@ -35,6 +35,11 @@ pub enum Rejection {
 pub struct App {
     pub _id_generator: std::sync::Mutex<models::IdGenerator>,
     pub billing_provider: Option<Arc<dyn crate::billing::BillingProvider>>,
+    /// Validity window of access tokens minted by the `capability_token`
+    /// grant. The agent binary configures this via
+    /// `--capability-token-validity` / `CAPABILITY_TOKEN_VALIDITY`, whose
+    /// default of one hour matches the SQL `generate_access_token` mint.
+    pub capability_token_validity: std::time::Duration,
     pub control_plane_jwt_decode_keys: Vec<tokens::jwt::DecodingKey>,
     pub control_plane_jwt_encode_key: tokens::jwt::EncodingKey,
     pub pg_pool: sqlx::PgPool,
@@ -55,10 +60,12 @@ impl App {
         publisher: crate::publications::Publisher,
         snapshot: Arc<dyn tokens::Watch<Snapshot>>,
         stripe_webhook_secret: Option<String>,
+        capability_token_validity: std::time::Duration,
     ) -> Self {
         Self {
             _id_generator: std::sync::Mutex::new(id_generator),
             billing_provider,
+            capability_token_validity,
             control_plane_jwt_decode_keys: vec![tokens::jwt::DecodingKey::from_secret(jwt_secret)],
             control_plane_jwt_encode_key: tokens::jwt::EncodingKey::from_secret(jwt_secret),
             pg_pool,

crates/control-plane-api/src/server/public/graphql/refresh_tokens.rs

@@ -117,6 +117,8 @@ pub struct RefreshTokensMutation;
 impl RefreshTokensMutation {
     /// Create a refresh token for the authenticated user.
     ///
+    /// Capability-masked callers are rejected: a refresh token exchanges for
+    /// a full-authority access token, which would escape the bearer's mask.
     /// Service-account callers are rejected: their API keys are administered
     /// via createApiKey and revokeApiKey.
     async fn create_refresh_token(
@@ -133,6 +135,11 @@ impl RefreshTokensMutation {
         let env = ctx.data::<crate::Envelope>()?;
         let claims = env.claims()?;
 
+        // The mask refusal precedes the service-account lookup: it's a pure
+        // function of the verified claims, so a masked caller is refused
+        // without a database round-trip, in the same order as the
+        // capability_token mint.
+        crate::Forbidden::require_unmasked(claims).map_err(crate::ApiError::Forbidden)?;
         super::service_accounts::verify_not_service_account(&env.pg_pool, claims.sub).await?;
 
         // ISO 8601 durations begin with 'P'; considering this cheap and good enough validation for now.
@@ -483,4 +490,81 @@ mod test {
             .await;
         assert!(revoke_again["errors"].is_array());
     }
+
+    /// A masked bearer cannot mint a refresh credential: a refresh token
+    /// exchanges for a full-authority access token, which would escape the
+    /// mask. Revocation never widens authority, so it deliberately stays
+    /// open to masked bearers.
+    #[sqlx::test(
+        migrations = "../../supabase/migrations",
+        fixtures(path = "../../../fixtures", scripts("data_planes", "alice"))
+    )]
+    async fn test_create_refresh_token_requires_unmasked(pool: sqlx::PgPool) {
+        let _guard = test_server::init();
+
+        let server = test_server::TestServer::start(
+            pool.clone(),
+            test_server::snapshot(pool.clone(), true).await,
+        )
+        .await;
+
+        let alice = uuid::Uuid::from_bytes([0x11; 16]);
+        let unmasked_token = server.make_access_token(alice, Some("alice@example.com"));
+        // The mask enables Admin, and is still refused: masked-ness is the
+        // claim's presence, never its value.
+        let masked_token =
+            server.make_masked_access_token(alice, Some("alice@example.com"), Some(vec!["Admin"]));
+
+        let refused: serde_json::Value = server
+            .graphql(
+                &serde_json::json!({
+                    "query": r#"mutation { createRefreshToken(validFor: "P30D") { id } }"#
+                }),
+                Some(&masked_token),
+            )
+            .await;
+        assert_eq!(
+            refused["errors"][0]["extensions"]["error"], "unmasked_token_required",
+            "a masked caller is refused with the structured code: {refused}"
+        );
+        assert_eq!(
+            refused["errors"][0]["extensions"]["missing_capabilities"],
+            serde_json::json!([]),
+            "no re-mint can remedy this refusal: {refused}"
+        );
+
+        // An unmasked caller mints one...
+        let create: serde_json::Value = server
+            .graphql(
+                &serde_json::json!({
+                    "query": r#"mutation { createRefreshToken(validFor: "P30D") { id } }"#
+                }),
+                Some(&unmasked_token),
+            )
+            .await;
+        assert!(
+            create["errors"].is_null(),
+            "create should succeed: {create}"
+        );
+        let token_id = create["data"]["createRefreshToken"]["id"]
+            .as_str()
+            .unwrap()
+            .to_string();
+
+        // ...and a masked bearer may revoke it.
+        let revoke: serde_json::Value = server
+            .graphql(
+                &serde_json::json!({
+                    "query": r#"mutation($id: Id!) { revokeRefreshToken(id: $id) }"#,
+                    "variables": { "id": token_id }
+                }),
+                Some(&masked_token),
+            )
+            .await;
+        assert!(
+            revoke["errors"].is_null(),
+            "masked revocation should succeed: {revoke}"
+        );
+        assert_eq!(revoke["data"]["revokeRefreshToken"], true);
+    }
 }