[SSO] Fix OIDC group sync privilege escalation via case folding

src/adapter/src/catalog.rs

@@ -928,14 +928,6 @@ impl Catalog {
         self.state.try_get_role_by_name(role_name)
     }
 
-    pub fn try_get_role_by_name_case_insensitive(&self, role_name: &str) -> Option<&Role> {
-        self.state.try_get_role_by_name_case_insensitive(role_name)
-    }
-
-    pub fn roles_by_lowercase_name(&self) -> BTreeMap<String, &Role> {
-        self.state.roles_by_lowercase_name()
-    }
-
     pub fn try_get_role_auth_by_id(&self, id: &RoleId) -> Option<&RoleAuth> {
         self.state.try_get_role_auth_by_id(id)
     }

src/adapter/src/catalog/state.rs

@@ -914,31 +914,6 @@ impl CatalogState {
             .map(|id| &self.roles_by_id[id])
     }
 
-    /// Case-insensitive role lookup for JWT group-to-role sync.
-    ///
-    /// Groups from JWTs are lowercased during extraction. SQL role names are
-    /// stored as-is (unquoted identifiers are lowercased by the parser, but
-    /// quoted identifiers preserve case). This method iterates all roles to
-    /// find a case-insensitive match.
-    pub fn try_get_role_by_name_case_insensitive(&self, role_name: &str) -> Option<&Role> {
-        let lower = role_name.to_lowercase();
-        self.roles_by_name
-            .iter()
-            .find(|(name, _)| name.to_lowercase() == lower)
-            .map(|(_, id)| &self.roles_by_id[id])
-    }
-
-    /// Returns a map from lowercased role name to `Role` for all roles.
-    ///
-    /// Use this when doing multiple case-insensitive lookups (e.g. iterating
-    /// JWT group claims) to avoid O(n) per lookup.
-    pub fn roles_by_lowercase_name(&self) -> BTreeMap<String, &Role> {
-        self.roles_by_name
-            .iter()
-            .map(|(name, id)| (name.to_lowercase(), &self.roles_by_id[id]))
-            .collect()
-    }
-
     pub(super) fn get_role_auth(&self, id: &RoleId) -> &RoleAuth {
         self.role_auth_by_id
             .get(id)

src/adapter/src/coord/group_sync.rs

@@ -44,7 +44,7 @@ pub struct GroupSyncDiff {
 /// - `current_membership`: The user's current `RoleMembership.map`
 ///   (role_id → grantor_id).
 /// - `target_role_ids`: Role IDs resolved from the JWT group names via
-///   case-insensitive catalog lookup.
+///   exact (case-sensitive) catalog lookup.
 ///
 /// # Semantics
 /// - Only roles granted by the JWT sync sentinel (`MZ_JWT_SYNC_ROLE_ID`)
@@ -149,8 +149,8 @@ impl Coordinator {
 
     /// Syncs the user's role memberships based on JWT group claims.
     ///
-    /// Resolves group names to catalog role IDs (case-insensitive),
-    /// computes the diff against current memberships, and executes
+    /// Resolves group names to catalog role IDs via exact (case-sensitive)
+    /// lookup, computes the diff against current memberships, and executes
     /// grant/revoke operations via `catalog_transact`.
     ///
     /// Groups that map to reserved role names (`mz_`/`pg_` prefixes) are
@@ -162,13 +162,12 @@ impl Coordinator {
         groups: &[String],
         notices: &mut Vec<AdapterNotice>,
     ) -> Result<(), AdapterError> {
-        let role_map = self.catalog().roles_by_lowercase_name();
-
-        // Resolve group names to role IDs (case-insensitive).
+        // Resolve group names to role IDs (exact, case-sensitive match).
         let mut target_role_ids = BTreeSet::new();
         for group in groups {
             // Filter out reserved role names (mz_/pg_ prefixes, PUBLIC).
-            if catalog::is_reserved_role_name(group) {
+            // Check case-insensitively so "MZ_SYSTEM" is also blocked.
+            if catalog::is_reserved_role_name(&group.to_lowercase()) {
                 warn!(
                     group = group.as_str(),
                     "OIDC group maps to reserved role name, skipping"
@@ -179,7 +178,7 @@ impl Coordinator {
                 continue;
             }
 
-            match role_map.get(&group.to_lowercase()).copied() {
+            match self.catalog().try_get_role_by_name(group) {
                 Some(role) => {
                     // Skip if the group resolves to the user's own role. This
                     // happens when an IdP echoes the username/email into the

src/authenticator/src/oidc.rs

@@ -251,8 +251,9 @@ impl OidcClaims {
     ///
     /// 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.
+    /// roles), or `Some(vec![...])` with deduplicated, sorted group names
+    /// (exact case preserved — matching against catalog role names is
+    /// case-sensitive).
     ///
     /// Accepts arrays of strings, single strings, or mixed arrays (non-string
     /// elements are filtered out). Other JSON types are treated as absent.
@@ -280,15 +281,14 @@ impl OidcClaims {
             }
         };
 
-        let normalized: Vec<String> = raw_groups
+        let groups: Vec<String> = raw_groups
             .into_iter()
-            .map(|g| g.trim().to_lowercase())
             .filter(|g| !g.is_empty())
             .collect::<BTreeSet<_>>()
             .into_iter()
             .collect();
 
-        Some(normalized)
+        Some(groups)
     }
 }
 
@@ -752,9 +752,14 @@ mod tests {
     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();
+        // Case is preserved; "Analytics" and "analytics" are distinct groups.
         assert_eq!(
             claims.groups("groups"),
-            Some(vec!["analytics".to_string(), "platform_eng".to_string()])
+            Some(vec![
+                "Analytics".to_string(),
+                "PLATFORM_ENG".to_string(),
+                "analytics".to_string(),
+            ])
         );
     }
 
@@ -849,32 +854,32 @@ mod tests {
 
     #[mz_ore::test]
     fn test_groups_whitespace_only_single_string() {
-        // Whitespace-only string trims to empty and is filtered out.
+        // Whitespace-only string is preserved as-is (exact matching, no trim).
         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![]));
+        assert_eq!(claims.groups("groups"), Some(vec!["  ".to_string()]));
     }
 
     #[mz_ore::test]
     fn test_groups_whitespace_names() {
-        // Leading/trailing whitespace is trimmed from group names.
+        // Leading/trailing whitespace is preserved (exact matching, no trim).
         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()])
+            Some(vec!["  spaces  ".to_string(), "eng".to_string()])
         );
     }
 
     #[mz_ore::test]
     fn test_groups_unicode_names() {
-        // Unicode group names should be lowercased correctly
+        // Unicode group names are preserved as-is (no case folding).
         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()])
+            Some(vec!["Développeurs".to_string(), "INGÉNIEURS".to_string()])
         );
     }
 
@@ -895,11 +900,19 @@ mod tests {
     }
 
     #[mz_ore::test]
-    fn test_groups_case_insensitive_dedup() {
-        // "Eng" and "eng" and "ENG" should all collapse to one "eng"
+    fn test_groups_no_case_folding() {
+        // Case is preserved; "Eng", "eng", "ENG", "eNg" are four distinct groups.
         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()]));
+        assert_eq!(
+            claims.groups("groups"),
+            Some(vec![
+                "ENG".to_string(),
+                "Eng".to_string(),
+                "eNg".to_string(),
+                "eng".to_string(),
+            ])
+        );
     }
 
     #[mz_ore::test]

test/http-auth/listener_config_oidc.json

@@ -0,0 +1,44 @@
+{
+  "sql": {
+    "internal": {
+      "addr": "0.0.0.0:6877",
+      "authenticator_kind": "None",
+      "allowed_roles": "Internal",
+      "enable_tls": false
+    }
+  },
+  "http": {
+    "external": {
+      "addr": "0.0.0.0:6876",
+      "authenticator_kind": "Oidc",
+      "allowed_roles": "NormalAndInternal",
+      "enable_tls": false,
+      "routes": {
+        "base": true,
+        "webhook": false,
+        "internal": false,
+        "metrics": false,
+        "profiling": false,
+        "mcp_agent": false,
+        "mcp_developer": false,
+        "console_config": false
+      }
+    },
+    "internal": {
+      "addr": "0.0.0.0:6878",
+      "authenticator_kind": "None",
+      "allowed_roles": "NormalAndInternal",
+      "enable_tls": false,
+      "routes": {
+        "base": false,
+        "webhook": false,
+        "internal": false,
+        "metrics": true,
+        "profiling": false,
+        "mcp_agent": false,
+        "mcp_developer": false,
+        "console_config": false
+      }
+    }
+  }
+}