[SSO] Add foundation for JWT group-to-role sync

doc/developer/generated/adapter-types/dyncfgs.md

@@ -1,10 +1,10 @@
 ---
 source: src/adapter-types/src/dyncfgs.rs
-revision: 1505e0c3f0
+revision: 0c832dee79
 ---
 
 # mz-adapter-types::dyncfgs
 
 Declares all dynamic configuration flags owned by the adapter layer as `mz_dyncfg::Config` constants.
-Covers session gating (`ALLOW_USER_SESSIONS`), zero-downtime deployment parameters (`WITH_0DT_*`, `ENABLE_0DT_*`), feature flags for expression caching, multi-replica sources, statement lifecycle logging, introspection subscribes, plan insights optimization thresholds, continual task builtins, password authentication, OIDC settings (`OIDC_ISSUER`, `OIDC_AUDIENCE`, `OIDC_AUTHENTICATION_CLAIM`), console OIDC configuration (`CONSOLE_OIDC_CLIENT_ID`, `CONSOLE_OIDC_SCOPES`), MCP endpoint toggles (`ENABLE_MCP_AGENT`, `ENABLE_MCP_AGENT_QUERY_TOOL`, `ENABLE_MCP_DEVELOPER`) and response size limit (`MCP_MAX_RESPONSE_SIZE`), persist fast-path ordering, S3 Tables region checks, and the user ID pool batch size.
+Covers session gating (`ALLOW_USER_SESSIONS`), zero-downtime deployment parameters (`WITH_0DT_*`, `ENABLE_0DT_*`), feature flags for expression caching, multi-replica sources, statement lifecycle logging, introspection subscribes, plan insights optimization thresholds, continual task builtins, password authentication, OIDC settings (`OIDC_ISSUER`, `OIDC_AUDIENCE`, `OIDC_AUTHENTICATION_CLAIM`), OIDC JWT group-to-role sync (`OIDC_GROUP_ROLE_SYNC_ENABLED`, `OIDC_GROUP_CLAIM`, `OIDC_GROUP_ROLE_SYNC_STRICT`), console OIDC configuration (`CONSOLE_OIDC_CLIENT_ID`, `CONSOLE_OIDC_SCOPES`), MCP endpoint toggles (`ENABLE_MCP_AGENT`, `ENABLE_MCP_AGENT_QUERY_TOOL`, `ENABLE_MCP_DEVELOPER`) and response size limit (`MCP_MAX_RESPONSE_SIZE`), persist fast-path ordering, S3 Tables region checks, and the user ID pool batch size.
 `all_dyncfgs` registers every config constant into a `ConfigSet` for use during bootstrap.

misc/python/materialize/mzcompose/__init__.py

@@ -646,6 +646,9 @@ def get_default_system_parameters(
     "oidc_issuer",
     "oidc_audience",
     "oidc_authentication_claim",
+    "oidc_group_role_sync_enabled",
+    "oidc_group_claim",
+    "oidc_group_role_sync_strict",
     "console_oidc_client_id",
     "console_oidc_scopes",
     "enable_mcp_agent",

misc/python/materialize/parallel_workload/action.py

@@ -1794,6 +1794,9 @@ def __init__(
             "oidc_issuer",
             "oidc_audience",
             "oidc_authentication_claim",
+            "oidc_group_role_sync_enabled",
+            "oidc_group_claim",
+            "oidc_group_role_sync_strict",
             "console_oidc_client_id",
             "console_oidc_scopes",
         ]

src/adapter-types/src/dyncfgs.rs

@@ -146,6 +146,29 @@ pub const OIDC_AUTHENTICATION_CLAIM: Config<&'static str> = Config::new(
     "OIDC authentication claim to use as username.",
 );
 
+/// Whether OIDC group-to-role sync is enabled.
+/// When true, JWT group claims are used to sync role memberships on login.
+pub const OIDC_GROUP_ROLE_SYNC_ENABLED: Config<bool> = Config::new(
+    "oidc_group_role_sync_enabled",
+    false,
+    "Enable OIDC JWT group-to-role membership sync on login.",
+);
+
+/// The JWT claim name that contains group memberships.
+pub const OIDC_GROUP_CLAIM: Config<&'static str> = Config::new(
+    "oidc_group_claim",
+    "groups",
+    "JWT claim name containing group memberships for role sync.",
+);
+
+/// Whether to reject login when group sync fails (strict/fail-closed mode).
+/// When false (default), sync failures are logged but login proceeds (fail-open).
+pub const OIDC_GROUP_ROLE_SYNC_STRICT: Config<bool> = Config::new(
+    "oidc_group_role_sync_strict",
+    false,
+    "When true, reject login if OIDC group-to-role sync fails (fail-closed).",
+);
+
 pub const PERSIST_FAST_PATH_ORDER: Config<bool> = Config::new(
     "persist_fast_path_order",
     false,
@@ -236,6 +259,9 @@ pub fn all_dyncfgs(configs: ConfigSet) -> ConfigSet {
         .add(&OIDC_ISSUER)
         .add(&OIDC_AUDIENCE)
         .add(&OIDC_AUTHENTICATION_CLAIM)
+        .add(&OIDC_GROUP_ROLE_SYNC_ENABLED)
+        .add(&OIDC_GROUP_CLAIM)
+        .add(&OIDC_GROUP_ROLE_SYNC_STRICT)
         .add(&PERSIST_FAST_PATH_ORDER)
         .add(&ENABLE_S3_TABLES_REGION_CHECK)
         .add(&ENABLE_MCP_AGENT)

src/authenticator/src/oidc.rs

@@ -12,7 +12,7 @@
 //! This module provides JWT-based authentication using OpenID Connect (OIDC).
 //! JWTs are validated locally using JWKS fetched from the configured provider.
 
-use std::collections::BTreeMap;
+use std::collections::{BTreeMap, BTreeSet};
 use std::sync::{Arc, Mutex};
 use std::time::Duration;
 
@@ -244,6 +244,50 @@ impl OidcClaims {
             .get(authentication_claim)
             .and_then(|value| value.as_str())
     }
+
+    /// Extracts group names from the specified JWT claim for group-to-role sync.
+    ///
+    /// Returns `None` if the claim is absent (skip sync, preserve current state),
+    /// `Some(vec![])` if the claim is present but empty (revoke all sync-granted
+    /// roles), or `Some(vec![...])` with normalized (lowercased, deduplicated,
+    /// sorted) group names.
+    ///
+    /// Accepts arrays of strings, single strings, or mixed arrays (non-string
+    /// elements are filtered out). Other JSON types are treated as absent.
+    pub fn groups(&self, claim_name: &str) -> Option<Vec<String>> {
+        let value = self.unknown_claims.get(claim_name)?;
+
+        let raw_groups: Vec<String> = match value {
+            serde_json::Value::Array(arr) => arr
+                .iter()
+                .filter_map(|v| v.as_str().map(String::from))
+                .collect(),
+            serde_json::Value::String(s) => {
+                if s.is_empty() {
+                    vec![]
+                } else {
+                    vec![s.clone()]
+                }
+            }
+            _ => {
+                warn!(
+                    claim_name,
+                    "OIDC group claim has unexpected type; skipping group sync"
+                );
+                return None;
+            }
+        };
+
+        let normalized: Vec<String> = raw_groups
+            .into_iter()
+            .map(|g| g.trim().to_lowercase())
+            .filter(|g| !g.is_empty())
+            .collect::<BTreeSet<_>>()
+            .into_iter()
+            .collect();
+
+        Some(normalized)
+    }
 }
 
 #[derive(Zeroize, ZeroizeOnDrop)]
@@ -662,4 +706,251 @@ mod tests {
         assert!(claims.aud.is_empty());
         assert!(claims.unknown_claims.is_empty());
     }
+
+    #[mz_ore::test]
+    fn test_groups_array() {
+        let json = r#"{"sub":"user","iss":"issuer","exp":1234,"aud":"app","groups":["analytics","platform_eng"]}"#;
+        let claims: OidcClaims = serde_json::from_str(json).unwrap();
+        assert_eq!(
+            claims.groups("groups"),
+            Some(vec!["analytics".to_string(), "platform_eng".to_string()])
+        );
+    }
+
+    #[mz_ore::test]
+    fn test_groups_single_string() {
+        let json = r#"{"sub":"user","iss":"issuer","exp":1234,"aud":"app","groups":"analytics"}"#;
+        let claims: OidcClaims = serde_json::from_str(json).unwrap();
+        assert_eq!(claims.groups("groups"), Some(vec!["analytics".to_string()]));
+    }
+
+    #[mz_ore::test]
+    fn test_groups_missing() {
+        let json = r#"{"sub":"user","iss":"issuer","exp":1234,"aud":"app"}"#;
+        let claims: OidcClaims = serde_json::from_str(json).unwrap();
+        assert_eq!(claims.groups("groups"), None);
+    }
+
+    #[mz_ore::test]
+    fn test_groups_empty_array() {
+        let json = r#"{"sub":"user","iss":"issuer","exp":1234,"aud":"app","groups":[]}"#;
+        let claims: OidcClaims = serde_json::from_str(json).unwrap();
+        assert_eq!(claims.groups("groups"), Some(vec![]));
+    }
+
+    #[mz_ore::test]
+    fn test_groups_mixed_case() {
+        let json = r#"{"sub":"user","iss":"issuer","exp":1234,"aud":"app","groups":["Analytics","PLATFORM_ENG","analytics"]}"#;
+        let claims: OidcClaims = serde_json::from_str(json).unwrap();
+        assert_eq!(
+            claims.groups("groups"),
+            Some(vec!["analytics".to_string(), "platform_eng".to_string()])
+        );
+    }
+
+    #[mz_ore::test]
+    fn test_groups_custom_claim_name() {
+        let json =
+            r#"{"sub":"user","iss":"issuer","exp":1234,"aud":"app","roles":["admin","viewer"]}"#;
+        let claims: OidcClaims = serde_json::from_str(json).unwrap();
+        assert_eq!(
+            claims.groups("roles"),
+            Some(vec!["admin".to_string(), "viewer".to_string()])
+        );
+        assert_eq!(claims.groups("groups"), None);
+    }
+
+    #[mz_ore::test]
+    fn test_groups_non_string_values_in_array() {
+        let json = r#"{"sub":"user","iss":"issuer","exp":1234,"aud":"app","groups":["valid",123,true,"also_valid"]}"#;
+        let claims: OidcClaims = serde_json::from_str(json).unwrap();
+        assert_eq!(
+            claims.groups("groups"),
+            Some(vec!["also_valid".to_string(), "valid".to_string()])
+        );
+    }
+
+    #[mz_ore::test]
+    fn test_groups_non_array_non_string() {
+        let json = r#"{"sub":"user","iss":"issuer","exp":1234,"aud":"app","groups":42}"#;
+        let claims: OidcClaims = serde_json::from_str(json).unwrap();
+        assert_eq!(claims.groups("groups"), None);
+    }
+
+    #[mz_ore::test]
+    fn test_groups_empty_string() {
+        let json = r#"{"sub":"user","iss":"issuer","exp":1234,"aud":"app","groups":""}"#;
+        let claims: OidcClaims = serde_json::from_str(json).unwrap();
+        assert_eq!(claims.groups("groups"), Some(vec![]));
+    }
+
+    #[mz_ore::test]
+    fn test_groups_null_claim() {
+        // Explicit null → treated as absent (not a valid group representation)
+        let json = r#"{"sub":"user","iss":"issuer","exp":1234,"aud":"app","groups":null}"#;
+        let claims: OidcClaims = serde_json::from_str(json).unwrap();
+        assert_eq!(claims.groups("groups"), None);
+    }
+
+    #[mz_ore::test]
+    fn test_groups_boolean_claim() {
+        // Boolean value → not array or string, treated as absent
+        let json = r#"{"sub":"user","iss":"issuer","exp":1234,"aud":"app","groups":true}"#;
+        let claims: OidcClaims = serde_json::from_str(json).unwrap();
+        assert_eq!(claims.groups("groups"), None);
+    }
+
+    #[mz_ore::test]
+    fn test_groups_object_claim() {
+        // JSON object → not array or string, treated as absent
+        let json =
+            r#"{"sub":"user","iss":"issuer","exp":1234,"aud":"app","groups":{"team":"eng"}}"#;
+        let claims: OidcClaims = serde_json::from_str(json).unwrap();
+        assert_eq!(claims.groups("groups"), None);
+    }
+
+    #[mz_ore::test]
+    fn test_groups_array_all_non_strings() {
+        // Array with zero valid string elements → Some(vec![]), not None
+        // (the claim *is* present, it just has no usable group names)
+        let json =
+            r#"{"sub":"user","iss":"issuer","exp":1234,"aud":"app","groups":[1,2,true,null]}"#;
+        let claims: OidcClaims = serde_json::from_str(json).unwrap();
+        assert_eq!(claims.groups("groups"), Some(vec![]));
+    }
+
+    #[mz_ore::test]
+    fn test_groups_array_with_nested_arrays() {
+        // Nested arrays are not strings, so they're filtered out
+        let json =
+            r#"{"sub":"user","iss":"issuer","exp":1234,"aud":"app","groups":[["nested"],"valid"]}"#;
+        let claims: OidcClaims = serde_json::from_str(json).unwrap();
+        assert_eq!(claims.groups("groups"), Some(vec!["valid".to_string()]));
+    }
+
+    #[mz_ore::test]
+    fn test_groups_array_with_empty_strings() {
+        // Empty strings are not valid role names and are filtered out,
+        // consistent with the single-string case where "" → Some(vec![]).
+        let json = r#"{"sub":"user","iss":"issuer","exp":1234,"aud":"app","groups":["","eng",""]}"#;
+        let claims: OidcClaims = serde_json::from_str(json).unwrap();
+        assert_eq!(claims.groups("groups"), Some(vec!["eng".to_string()]));
+    }
+
+    #[mz_ore::test]
+    fn test_groups_whitespace_only_single_string() {
+        // Whitespace-only string trims to empty and is filtered out.
+        let json = r#"{"sub":"user","iss":"issuer","exp":1234,"aud":"app","groups":"  "}"#;
+        let claims: OidcClaims = serde_json::from_str(json).unwrap();
+        assert_eq!(claims.groups("groups"), Some(vec![]));
+    }
+
+    #[mz_ore::test]
+    fn test_groups_whitespace_names() {
+        // Leading/trailing whitespace is trimmed from group names.
+        let json =
+            r#"{"sub":"user","iss":"issuer","exp":1234,"aud":"app","groups":["  spaces  ","eng"]}"#;
+        let claims: OidcClaims = serde_json::from_str(json).unwrap();
+        assert_eq!(
+            claims.groups("groups"),
+            Some(vec!["eng".to_string(), "spaces".to_string()])
+        );
+    }
+
+    #[mz_ore::test]
+    fn test_groups_unicode_names() {
+        // Unicode group names should be lowercased correctly
+        let json = r#"{"sub":"user","iss":"issuer","exp":1234,"aud":"app","groups":["Développeurs","INGÉNIEURS"]}"#;
+        let claims: OidcClaims = serde_json::from_str(json).unwrap();
+        assert_eq!(
+            claims.groups("groups"),
+            Some(vec!["développeurs".to_string(), "ingénieurs".to_string()])
+        );
+    }
+
+    #[mz_ore::test]
+    fn test_groups_special_characters() {
+        // Group names with special characters (hyphens, underscores, dots)
+        // are common in enterprise IdPs like Azure AD / Okta
+        let json = r#"{"sub":"user","iss":"issuer","exp":1234,"aud":"app","groups":["team-platform.eng","org_data-science","role/admin"]}"#;
+        let claims: OidcClaims = serde_json::from_str(json).unwrap();
+        assert_eq!(
+            claims.groups("groups"),
+            Some(vec![
+                "org_data-science".to_string(),
+                "role/admin".to_string(),
+                "team-platform.eng".to_string(),
+            ])
+        );
+    }
+
+    #[mz_ore::test]
+    fn test_groups_case_insensitive_dedup() {
+        // "Eng" and "eng" and "ENG" should all collapse to one "eng"
+        let json = r#"{"sub":"user","iss":"issuer","exp":1234,"aud":"app","groups":["Eng","eng","ENG","eNg"]}"#;
+        let claims: OidcClaims = serde_json::from_str(json).unwrap();
+        assert_eq!(claims.groups("groups"), Some(vec!["eng".to_string()]));
+    }
+
+    #[mz_ore::test]
+    fn test_groups_large_array() {
+        // Verify we handle a reasonably large group list without issues
+        let groups: Vec<String> = (0..100).map(|i| format!("\"group_{}\"", i)).collect();
+        let json = format!(
+            r#"{{"sub":"user","iss":"issuer","exp":1234,"aud":"app","groups":[{}]}}"#,
+            groups.join(",")
+        );
+        let claims: OidcClaims = serde_json::from_str(&json).unwrap();
+        let result = claims.groups("groups").unwrap();
+        assert_eq!(result.len(), 100);
+        // BTreeSet ensures sorted order
+        assert_eq!(result[0], "group_0");
+        assert_eq!(result[99], "group_99");
+    }
+
+    #[mz_ore::test]
+    fn test_groups_float_claim() {
+        // Float value → not array or string, treated as absent
+        let json = r#"{"sub":"user","iss":"issuer","exp":1234,"aud":"app","groups":3.14}"#;
+        let claims: OidcClaims = serde_json::from_str(json).unwrap();
+        assert_eq!(claims.groups("groups"), None);
+    }
+
+    #[mz_ore::test]
+    fn test_groups_array_with_null_elements() {
+        // Null elements in array are not strings, filtered out
+        let json = r#"{"sub":"user","iss":"issuer","exp":1234,"aud":"app","groups":["eng",null,"ops",null]}"#;
+        let claims: OidcClaims = serde_json::from_str(json).unwrap();
+        assert_eq!(
+            claims.groups("groups"),
+            Some(vec!["eng".to_string(), "ops".to_string()])
+        );
+    }
+
+    #[mz_ore::test]
+    fn test_groups_array_with_object_elements() {
+        // Object elements in array are not strings, filtered out
+        let json = r#"{"sub":"user","iss":"issuer","exp":1234,"aud":"app","groups":["eng",{"name":"ops"},"analytics"]}"#;
+        let claims: OidcClaims = serde_json::from_str(json).unwrap();
+        assert_eq!(
+            claims.groups("groups"),
+            Some(vec!["analytics".to_string(), "eng".to_string()])
+        );
+    }
+
+    #[mz_ore::test]
+    fn test_groups_sorted_output() {
+        // Verify output is sorted alphabetically regardless of input order
+        let json = r#"{"sub":"user","iss":"issuer","exp":1234,"aud":"app","groups":["zebra","alpha","mango","beta"]}"#;
+        let claims: OidcClaims = serde_json::from_str(json).unwrap();
+        assert_eq!(
+            claims.groups("groups"),
+            Some(vec![
+                "alpha".to_string(),
+                "beta".to_string(),
+                "mango".to_string(),
+                "zebra".to_string(),
+            ])
+        );
+    }
 }