fix: make external OAuth authorization configurable

.env.example

@@ -208,6 +208,17 @@ OAUTH_EMAIL_CLAIM=email
 OAUTH_ROLES_CLAIM=roles
 OAUTH_GROUPS_CLAIM=groups
 
+# External OAuth authorization mapping. Role names are matched
+# case-insensitively. Trusted domains require email_verified=true.
+OAUTH_DEFAULT_ROLES=oauth_user
+OAUTH_DEFAULT_SECURITY_LEVEL=internal
+OAUTH_DEFAULT_PERMISSIONS=read_data
+OAUTH_TRUSTED_DOMAINS=
+OAUTH_TRUSTED_DOMAIN_SECURITY_LEVEL=confidential
+# JSON values replace the complete built-in mapping when configured.
+# OAUTH_ROLE_SECURITY_LEVELS_JSON={"analyst":"internal","executive":"secret"}
+# OAUTH_ROLE_PERMISSIONS_JSON={"analyst":["read_data","query_database"],"executive":["read_data"]}
+
 # OAuth session settings
 OAUTH_SESSION_SECRET=your_oauth_session_secret_here
 OAUTH_SESSION_EXPIRY=3600

README.md

@@ -1116,29 +1116,41 @@ The system supports four security levels with hierarchical access control:
 
 #### Role Configuration
 
-Configure user roles and permissions:
+External OAuth role mapping is configured through environment variables:
 
-```python
-# Example role configuration
-role_permissions = {
-    "data_analyst": {
-        "security_level": "internal",
-        "permissions": ["read_data", "execute_query"],
-        "allowed_tables": ["sales", "products", "orders"]
-    },
-    "data_admin": {
-        "security_level": "confidential", 
-        "permissions": ["read_data", "execute_query", "admin"],
-        "allowed_tables": ["*"]
-    },
-    "executive": {
-        "security_level": "secret",
-        "permissions": ["read_data", "execute_query", "admin"],
-        "allowed_tables": ["*"]
-    }
-}
+```env
+# Roles supplied when the provider returns no role claim
+OAUTH_DEFAULT_ROLES=oauth_user
+
+# Fallbacks for users whose roles do not occur in the JSON mappings
+OAUTH_DEFAULT_SECURITY_LEVEL=internal
+OAUTH_DEFAULT_PERMISSIONS=read_data
+
+# Exact domains only. Domain elevation is applied only when the provider
+# returns email_verified=true.
+OAUTH_TRUSTED_DOMAINS=example.com,internal.example.com
+OAUTH_TRUSTED_DOMAIN_SECURITY_LEVEL=confidential
+
+# Each JSON value replaces the complete built-in mapping.
+OAUTH_ROLE_SECURITY_LEVELS_JSON={"analyst":"internal","executive":"secret"}
+OAUTH_ROLE_PERMISSIONS_JSON={"analyst":["read_data","query_database"],"executive":["read_data","query_database","admin"]}
 ```
 
+Role names and trusted domains are matched case-insensitively. Supported
+security levels are `public`, `internal`, `confidential`, and `secret`. An
+explicit empty permission array denies application permissions for that role;
+an empty `OAUTH_DEFAULT_PERMISSIONS` value makes unknown roles fail closed.
+
+The built-in role defaults preserve previous behavior for `admin`,
+`administrator`, `data_admin`, `super_admin`, `data_analyst`, `developer`,
+`manager`, `viewer`, `user`, and `oauth_user`. No email domain is trusted by
+default.
+
+These settings govern MCP application permissions and security classification.
+Database, table, column, and row access must still be enforced with Doris users,
+roles, grants, views, and row policies; OAuth mapping does not bypass Doris
+authorization.
+
 ### SQL Security Validation
 
 The system automatically validates SQL queries for security risks:

doris_mcp_server/auth/oauth_provider.py

@@ -36,6 +36,13 @@
 
 logger = get_logger(__name__)
 
+_SECURITY_LEVEL_RANK = {
+    SecurityLevel.PUBLIC: 0,
+    SecurityLevel.INTERNAL: 1,
+    SecurityLevel.CONFIDENTIAL: 2,
+    SecurityLevel.SECRET: 3,
+}
+
 
 class OAuthAuthenticationProvider:
     """OAuth authentication provider for Doris MCP Server"""
@@ -275,26 +282,42 @@ async def _determine_security_level(self, user_info: OAuthUserInfo) -> SecurityL
         Returns:
             SecurityLevel for the user
         """
-        # Check if user has admin roles
-        admin_roles = {"admin", "administrator", "data_admin", "super_admin"}
-        if any(role.lower() in admin_roles for role in user_info.roles):
-            return SecurityLevel.SECRET
-
-        # Check email domain for internal users
-        if user_info.email:
-            # You can configure trusted domains for internal access
-            trusted_domains = ["yourcompany.com", "internal.org"]  # Configure as needed
-            email_domain = user_info.email.split("@")[-1].lower()
-            if email_domain in trusted_domains:
-                return SecurityLevel.CONFIDENTIAL
-
-        # Check for special roles
-        elevated_roles = {"data_analyst", "developer", "manager"}
-        if any(role.lower() in elevated_roles for role in user_info.roles):
-            return SecurityLevel.CONFIDENTIAL
-
-        # Default to internal level for OAuth users
-        return SecurityLevel.INTERNAL
+        security_config = self.config.security
+        role_levels = {
+            str(role).strip().lower(): SecurityLevel(str(level).strip().lower())
+            for role, level in security_config.oauth_role_security_levels.items()
+        }
+        matched_levels = [
+            role_levels[role]
+            for role in {
+                configured_role.strip().lower()
+                for configured_role in user_info.roles
+            }
+            if role in role_levels
+        ]
+
+        if user_info.email and user_info.email_verified is True:
+            local_part, separator, email_domain = user_info.email.rpartition("@")
+            trusted_domains = {
+                domain.strip().lower().removeprefix("@")
+                for domain in security_config.oauth_trusted_domains
+            }
+            if (
+                separator
+                and local_part
+                and email_domain.lower() in trusted_domains
+            ):
+                matched_levels.append(
+                    SecurityLevel(
+                        security_config.oauth_trusted_domain_security_level.lower()
+                    )
+                )
+
+        if matched_levels:
+            return max(matched_levels, key=_SECURITY_LEVEL_RANK.__getitem__)
+        return SecurityLevel(
+            security_config.oauth_default_security_level.strip().lower()
+        )
 
     async def _map_permissions(self, roles: list[str]) -> list[str]:
         """Map OAuth roles to application permissions
@@ -305,32 +328,25 @@ async def _map_permissions(self, roles: list[str]) -> list[str]:
         Returns:
             List of application permissions
         """
-        permissions = set()
-
-        # Role to permission mapping
+        permissions: set[str] = set()
+        matched_role = False
         role_permissions = {
-            "admin": ["admin", "read_data", "write_data", "manage_users"],
-            "administrator": ["admin", "read_data", "write_data", "manage_users"],
-            "data_admin": ["admin", "read_data", "write_data"],
-            "super_admin": ["admin", "read_data", "write_data", "manage_users", "system_admin"],
-            "data_analyst": ["read_data", "query_database"],
-            "developer": ["read_data", "query_database", "debug"],
-            "viewer": ["read_data"],
-            "user": ["read_data"],
-            "oauth_user": ["read_data"]  # Default OAuth user permission
+            str(role).strip().lower(): configured_permissions
+            for role, configured_permissions in (
+                self.config.security.oauth_role_permissions.items()
+            )
         }
 
-        # Map roles to permissions
         for role in roles:
-            role_lower = role.lower()
+            role_lower = role.strip().lower()
             if role_lower in role_permissions:
+                matched_role = True
                 permissions.update(role_permissions[role_lower])
 
-        # Ensure OAuth users have at least basic permissions
-        if not permissions:
-            permissions.add("read_data")
+        if not matched_role:
+            permissions.update(self.config.security.oauth_default_permissions)
 
-        return list(permissions)
+        return sorted(permissions)
 
     def get_provider_info(self) -> dict[str, Any]:
         """Get OAuth provider information