models: capability-mask vocabulary and lenient claim

crates/agent/src/integration_tests/harness.rs

@@ -1683,6 +1683,7 @@ impl TestHarness {
             exp: (req_start + chrono::Duration::hours(1)).timestamp() as u64,
             role: "authenticated".to_string(),
             email: Some("user@example.com".to_string()),
+            capability_mask: None,
         };
 
         let token = tokens::jwt::sign(&claims, &app.control_plane_jwt_encode_key)

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

@@ -94,6 +94,9 @@ pub async fn authorize_dekaf(
         sub: uuid::Uuid::nil(),
         role: DEKAF_ROLE.to_string(),
         email: None,
+        // This token authorizes the `dekaf` role rather than a user (`sub`
+        // is nil), so there is no user authority for a mask to attenuate.
+        capability_mask: None,
     };
 
     // Only return a token if we are not redirecting

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

@@ -371,6 +371,7 @@ mod tests {
             sub: user_id,
             role: "authenticated".to_string(),
             email,
+            capability_mask: None,
         };
 
         match evaluate_authorization(&snapshot, &claims, &collection, capability) {

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

@@ -551,6 +551,7 @@ mod tests {
             sub: user_id,
             role: "authenticated".to_string(),
             email,
+            capability_mask: None,
         };
 
         match evaluate_authorization(&snapshot, &claims, &prefix, &data_plane, capability) {

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

@@ -498,6 +498,7 @@ mod tests {
             sub: user_id,
             role: "authenticated".to_string(),
             email,
+            capability_mask: None,
         };
 
         match evaluate_authorization(&snapshot, &claims, &task, capability) {

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

@@ -170,6 +170,7 @@ impl TestServer {
             role: "authenticated".to_string(),
             aud: "authenticated".to_string(),
             email: email.map(String::from),
+            capability_mask: None,
         };
 
         jsonwebtoken::encode(

crates/models/src/authorizations.rs

@@ -20,6 +20,29 @@ pub struct ControlClaims {
     // Authorized user email, if known.
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub email: Option<String>,
+    // Capability-bundle names to which this token's authority is masked.
+    //
+    // `None` is an unmasked token — today that's every token we mint, and a
+    // full-authority credential stays unmasked once masking exists. `Some`
+    // is a masked token whose authority is its user's live grants
+    // intersected with the capability bits of the recognized bundle names
+    // herein, and an empty list is valid: it mints an identity-only token.
+    // Every individual capability is itself a same-named bundle, so the
+    // vocabulary spans coarse bundles and single capability bits alike.
+    //
+    // Deliberately an opaque list of strings rather than
+    // `authz::CapabilitySet`, so that this shared claim doesn't structurally
+    // depend on the newest capability variant. A name an instance doesn't
+    // recognize must parse and then be inert — it can never widen authority —
+    // which is what makes mixed-version fleets and future capability names
+    // safe by construction. See `authz::CapabilityMask::from_claim`.
+    //
+    // Leniency is over names only. The claim's shape stays strict — an array,
+    // absent, or null — because the control plane is its sole minter, so a
+    // differently-shaped claim is a corrupt or forged token and failing
+    // verification is the fail-safe outcome.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub capability_mask: Option<Vec<String>>,
 }
 
 impl ControlClaims {
@@ -250,3 +273,241 @@ pub struct DekafAuthResponse {
 const fn capability_read() -> crate::Capability {
     crate::Capability::Read
 }
+
+#[cfg(test)]
+mod test {
+    use super::ControlClaims;
+    use crate::authz::CapabilityMask;
+
+    #[test]
+    fn test_capability_mask_claim_forms() {
+        // Every form of the claim parses. A token which fails to parse is a
+        // token which fails to authenticate, so an unrecognized name must
+        // never be a deserialization error: it's carried through and is then
+        // inert when the mask is built.
+        let base = serde_json::json!({
+            "aud": "authenticated",
+            "iat": 1000,
+            "exp": 2000,
+            "sub": "11111111-1111-1111-1111-111111111111",
+            "role": "authenticated",
+        });
+        // `None` leaves the field absent; `Some` sets it, including to an
+        // explicit JSON null. Absent and null both mean "no mask", while an
+        // empty array is a mask enabling nothing — and the difference
+        // is load-bearing: `[]` attenuates authority to nothing, absence
+        // doesn't attenuate at all.
+        let cases = [
+            // Absent: every token minted before capability masks existed.
+            None,
+            Some(serde_json::Value::Null),
+            Some(serde_json::json!([
+                "CatalogRead",
+                "JournalRead",
+                "Delegate"
+            ])),
+            Some(serde_json::json!(["Viewer"])),
+            Some(serde_json::json!([])),
+            Some(serde_json::json!(["SpecEdit", "FutureCapability"])),
+            Some(serde_json::json!(["FutureCapability"])),
+        ];
+
+        let outcomes: Vec<(Option<Vec<String>>, CapabilityMask)> = cases
+            .into_iter()
+            .map(|mask| {
+                let mut claims = base.clone();
+                if let Some(mask) = mask {
+                    claims["capability_mask"] = mask;
+                }
+                let claims: ControlClaims = serde_json::from_value(claims).unwrap();
+                let mask = CapabilityMask::from_claim(claims.capability_mask.as_deref());
+                (claims.capability_mask, mask)
+            })
+            .collect();
+
+        // Claim-less forms map to the unmasked (full) set. Asserted rather
+        // than snapshotted so this test doesn't churn when a capability
+        // variant is added.
+        for (claim, mask) in &outcomes {
+            if claim.is_none() {
+                assert_eq!(*mask, CapabilityMask::UNMASKED);
+            }
+        }
+        let bounded: Vec<_> = outcomes
+            .into_iter()
+            .filter(|(claim, _)| claim.is_some())
+            .collect();
+
+        insta::assert_debug_snapshot!(bounded, @r#"
+        [
+            (
+                Some(
+                    [
+                        "CatalogRead",
+                        "JournalRead",
+                        "Delegate",
+                    ],
+                ),
+                CapabilityMask(
+                    EnumSet(CatalogRead | JournalRead | Delegate),
+                ),
+            ),
+            (
+                Some(
+                    [
+                        "Viewer",
+                    ],
+                ),
+                CapabilityMask(
+                    EnumSet(CatalogRead | JournalRead | ViewDataPlanePrivateNetworking),
+                ),
+            ),
+            (
+                Some(
+                    [],
+                ),
+                CapabilityMask(
+                    EnumSet(),
+                ),
+            ),
+            (
+                Some(
+                    [
+                        "SpecEdit",
+                        "FutureCapability",
+                    ],
+                ),
+                CapabilityMask(
+                    EnumSet(SpecEdit),
+                ),
+            ),
+            (
+                Some(
+                    [
+                        "FutureCapability",
+                    ],
+                ),
+                CapabilityMask(
+                    EnumSet(),
+                ),
+            ),
+        ]
+        "#);
+    }
+
+    #[test]
+    fn test_capability_mask_claim_rejects_malformed_shapes() {
+        // Leniency is over names only. The claim must be an array of
+        // strings, absent, or null; any other shape fails deserialization,
+        // and a token whose claims fail to parse fails to authenticate.
+        // The control plane is the claim's sole minter, so a
+        // differently-shaped claim is corrupt or forged, and refusing the
+        // token is the fail-safe outcome.
+        let base = serde_json::json!({
+            "aud": "authenticated",
+            "iat": 1000,
+            "exp": 2000,
+            "sub": "11111111-1111-1111-1111-111111111111",
+            "role": "authenticated",
+        });
+        let errors: Vec<String> = [
+            serde_json::json!("CatalogRead"),
+            serde_json::json!(42),
+            serde_json::json!(true),
+            serde_json::json!({"names": ["CatalogRead"]}),
+            serde_json::json!([["CatalogRead"]]),
+            serde_json::json!(["CatalogRead", 42]),
+            serde_json::json!([null]),
+        ]
+        .into_iter()
+        .map(|mask| {
+            let mut claims = base.clone();
+            claims["capability_mask"] = mask;
+            serde_json::from_value::<ControlClaims>(claims)
+                .unwrap_err()
+                .to_string()
+        })
+        .collect();
+
+        insta::assert_debug_snapshot!(errors, @r#"
+        [
+            "invalid type: string \"CatalogRead\", expected a sequence",
+            "invalid type: integer `42`, expected a sequence",
+            "invalid type: boolean `true`, expected a sequence",
+            "invalid type: map, expected a sequence",
+            "invalid type: sequence, expected a string",
+            "invalid type: integer `42`, expected a string",
+            "invalid type: null, expected a string",
+        ]
+        "#);
+    }
+
+    #[test]
+    fn test_capability_mask_claim_round_trip() {
+        let claims = ControlClaims {
+            aud: "authenticated".to_string(),
+            iat: 1000,
+            exp: 2000,
+            sub: uuid::Uuid::nil(),
+            role: "authenticated".to_string(),
+            email: None,
+            capability_mask: None,
+        };
+
+        // An unmasked token doesn't carry the claim at all, so tokens we
+        // mint today are unchanged on the wire.
+        insta::assert_json_snapshot!(claims, @r#"
+        {
+          "aud": "authenticated",
+          "iat": 1000,
+          "exp": 2000,
+          "sub": "00000000-0000-0000-0000-000000000000",
+          "role": "authenticated"
+        }
+        "#);
+
+        // An empty mask is distinct from an absent one on the wire, and
+        // survives a round trip as such.
+        let masked = ControlClaims {
+            capability_mask: Some(Vec::new()),
+            ..claims
+        };
+        insta::assert_json_snapshot!(masked, @r#"
+        {
+          "aud": "authenticated",
+          "iat": 1000,
+          "exp": 2000,
+          "sub": "00000000-0000-0000-0000-000000000000",
+          "role": "authenticated",
+          "capability_mask": []
+        }
+        "#);
+
+        // A populated mask serializes its names verbatim — including names
+        // this binary doesn't recognize — and they survive a round trip
+        // intact. Carry-through is load-bearing: an upgrade token's
+        // unrecognized names must re-mint unchanged rather than being
+        // dropped by whichever instance happens to re-sign it.
+        let masked = ControlClaims {
+            capability_mask: Some(vec!["SpecEdit".to_string(), "FutureCapability".to_string()]),
+            ..masked
+        };
+        let round_tripped: ControlClaims =
+            serde_json::from_value(serde_json::to_value(&masked).unwrap()).unwrap();
+        assert_eq!(round_tripped.capability_mask, masked.capability_mask);
+
+        insta::assert_json_snapshot!(masked, @r#"
+        {
+          "aud": "authenticated",
+          "iat": 1000,
+          "exp": 2000,
+          "sub": "00000000-0000-0000-0000-000000000000",
+          "role": "authenticated",
+          "capability_mask": [
+            "SpecEdit",
+            "FutureCapability"
+          ]
+        }
+        "#);
+    }
+}