This is an automated email from the ASF dual-hosted git repository.
FreeOnePlus pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris-mcp-server.git
The following commit(s) were added to refs/heads/master by this push:
new 44e910a fix: make OAuth authorization mapping configurable (#157)
44e910a is described below
commit 44e910aa00ac486b3af80d41cf5b239ef6a154f8
Author: Yijia Su <[email protected]>
AuthorDate: Thu Jul 30 18:48:10 2026 +0800
fix: make OAuth authorization mapping configurable (#157)
---
.env.example | 11 ++
README.md | 52 +++--
doris_mcp_server/auth/oauth_provider.py | 92 +++++----
doris_mcp_server/utils/config.py | 281 ++++++++++++++++++++++++++++
test/auth/test_external_oauth_config.py | 63 +++++++
test/auth/test_external_oauth_validation.py | 61 ++++++
6 files changed, 502 insertions(+), 58 deletions(-)
diff --git a/.env.example b/.env.example
index 676e8b9..1b9ff10 100644
--- a/.env.example
+++ b/.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
diff --git a/README.md b/README.md
index 5bcd34a..6f735a7 100644
--- a/README.md
+++ b/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:
diff --git a/doris_mcp_server/auth/oauth_provider.py
b/doris_mcp_server/auth/oauth_provider.py
index 24c649b..ae5e0a8 100644
--- a/doris_mcp_server/auth/oauth_provider.py
+++ b/doris_mcp_server/auth/oauth_provider.py
@@ -36,6 +36,13 @@ from .oauth_types import OAuthTokens, OAuthUserInfo
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 @@ class OAuthAuthenticationProvider:
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 @@ class OAuthAuthenticationProvider:
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
diff --git a/doris_mcp_server/utils/config.py b/doris_mcp_server/utils/config.py
index dcbd7c5..6102c81 100644
--- a/doris_mcp_server/utils/config.py
+++ b/doris_mcp_server/utils/config.py
@@ -66,6 +66,36 @@ DORIS_OAUTH_METADATA_TOOL_ALLOWLIST_DEFAULT = list(
DORIS_OAUTH_METADATA_TOOL_NAMES
)
+EXTERNAL_OAUTH_SECURITY_LEVELS = frozenset(
+ {"public", "internal", "confidential", "secret"}
+)
+DEFAULT_EXTERNAL_OAUTH_ROLE_SECURITY_LEVELS = {
+ "admin": "secret",
+ "administrator": "secret",
+ "data_admin": "secret",
+ "super_admin": "secret",
+ "data_analyst": "confidential",
+ "developer": "confidential",
+ "manager": "confidential",
+}
+DEFAULT_EXTERNAL_OAUTH_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"],
+}
+
@dataclass(frozen=True)
class ConfigValue:
@@ -146,6 +176,52 @@ def _env_csv(name: str, default: list[str]) -> list[str]:
return [part.strip() for part in value.split(",") if part.strip()]
+def _env_json_string_map(
+ name: str,
+ default: dict[str, str],
+) -> dict[str, str]:
+ value = os.getenv(name)
+ if value is None:
+ return default
+ try:
+ parsed = json.loads(value)
+ except json.JSONDecodeError as exc:
+ raise AuthConfigError(f"{name} must be a valid JSON object") from exc
+ if not isinstance(parsed, dict) or any(
+ not isinstance(key, str) or not isinstance(item, str)
+ for key, item in parsed.items()
+ ):
+ raise AuthConfigError(f"{name} must map strings to strings")
+ return {
+ key.strip(): item.strip()
+ for key, item in parsed.items()
+ }
+
+
+def _env_json_string_list_map(
+ name: str,
+ default: dict[str, list[str]],
+) -> dict[str, list[str]]:
+ value = os.getenv(name)
+ if value is None:
+ return default
+ try:
+ parsed = json.loads(value)
+ except json.JSONDecodeError as exc:
+ raise AuthConfigError(f"{name} must be a valid JSON object") from exc
+ if not isinstance(parsed, dict) or any(
+ not isinstance(key, str)
+ or not isinstance(items, list)
+ or any(not isinstance(item, str) for item in items)
+ for key, items in parsed.items()
+ ):
+ raise AuthConfigError(f"{name} must map strings to arrays of strings")
+ return {
+ key.strip(): [item.strip() for item in items]
+ for key, items in parsed.items()
+ }
+
+
def _coerce_csv_config(value: Any) -> list[str]:
if value is None:
return []
@@ -482,6 +558,21 @@ class SecurityConfig:
oauth_name_claim: str = "name"
oauth_roles_claim: str = "roles" # Custom claim for roles
oauth_default_roles: list[str] = field(default_factory=lambda:
["oauth_user"])
+ oauth_default_security_level: str = "internal"
+ oauth_trusted_domains: list[str] = field(default_factory=list)
+ oauth_trusted_domain_security_level: str = "confidential"
+ oauth_role_security_levels: dict[str, str] = field(
+ default_factory=lambda:
dict(DEFAULT_EXTERNAL_OAUTH_ROLE_SECURITY_LEVELS)
+ )
+ oauth_role_permissions: dict[str, list[str]] = field(
+ default_factory=lambda: {
+ role: list(permissions)
+ for role, permissions in
DEFAULT_EXTERNAL_OAUTH_ROLE_PERMISSIONS.items()
+ }
+ )
+ oauth_default_permissions: list[str] = field(
+ default_factory=lambda: ["read_data"]
+ )
def __post_init__(self) -> None:
"""Initialize default OAuth scopes based on provider"""
@@ -496,6 +587,142 @@ class SecurityConfig:
self.oauth_scopes = ["openid", "email", "profile"]
+def _normalize_external_oauth_authorization_config(
+ security: SecurityConfig,
+) -> None:
+ """Normalize and validate external OAuth authorization mappings."""
+
+ def normalize_security_level(value: Any, setting: str) -> str:
+ normalized = str(value or "").strip().lower()
+ if normalized not in EXTERNAL_OAUTH_SECURITY_LEVELS:
+ allowed = ", ".join(sorted(EXTERNAL_OAUTH_SECURITY_LEVELS))
+ raise AuthConfigError(f"{setting} must be one of {allowed}")
+ return normalized
+
+ security.oauth_default_security_level = normalize_security_level(
+ security.oauth_default_security_level,
+ "OAUTH_DEFAULT_SECURITY_LEVEL",
+ )
+ security.oauth_trusted_domain_security_level = normalize_security_level(
+ security.oauth_trusted_domain_security_level,
+ "OAUTH_TRUSTED_DOMAIN_SECURITY_LEVEL",
+ )
+
+ if not isinstance(security.oauth_trusted_domains, list) or any(
+ not isinstance(domain, str)
+ for domain in security.oauth_trusted_domains
+ ):
+ raise AuthConfigError(
+ "OAUTH_TRUSTED_DOMAINS must be a comma-separated list of domains"
+ )
+ normalized_domains = []
+ for configured_domain in security.oauth_trusted_domains:
+ domain = str(configured_domain).strip().lower().removeprefix("@")
+ if (
+ not domain
+ or "@" in domain
+ or any(character.isspace() for character in domain)
+ ):
+ raise AuthConfigError(
+ "OAUTH_TRUSTED_DOMAINS must contain exact email domains"
+ )
+ if domain not in normalized_domains:
+ normalized_domains.append(domain)
+ security.oauth_trusted_domains = normalized_domains
+
+ normalized_role_levels: dict[str, str] = {}
+ if not isinstance(security.oauth_role_security_levels, dict):
+ raise AuthConfigError(
+ "OAUTH_ROLE_SECURITY_LEVELS_JSON must be a JSON object"
+ )
+ for configured_role, configured_level in (
+ security.oauth_role_security_levels.items()
+ ):
+ if not isinstance(configured_role, str) or not isinstance(
+ configured_level,
+ str,
+ ):
+ raise AuthConfigError(
+ "OAUTH_ROLE_SECURITY_LEVELS_JSON must map strings to strings"
+ )
+ role = configured_role.strip().lower()
+ if not role:
+ raise AuthConfigError(
+ "OAUTH_ROLE_SECURITY_LEVELS_JSON contains an empty role"
+ )
+ normalized_role_levels[role] = normalize_security_level(
+ configured_level,
+ f"OAUTH_ROLE_SECURITY_LEVELS_JSON[{role}]",
+ )
+ security.oauth_role_security_levels = normalized_role_levels
+
+ normalized_role_permissions: dict[str, list[str]] = {}
+ if not isinstance(security.oauth_role_permissions, dict):
+ raise AuthConfigError(
+ "OAUTH_ROLE_PERMISSIONS_JSON must be a JSON object"
+ )
+ for configured_role, configured_permissions in (
+ security.oauth_role_permissions.items()
+ ):
+ if not isinstance(configured_role, str):
+ raise AuthConfigError(
+ "OAUTH_ROLE_PERMISSIONS_JSON role names must be strings"
+ )
+ role = configured_role.strip().lower()
+ if not role:
+ raise AuthConfigError(
+ "OAUTH_ROLE_PERMISSIONS_JSON contains an empty role"
+ )
+ if not isinstance(configured_permissions, list) or any(
+ not isinstance(permission, str)
+ for permission in configured_permissions
+ ):
+ raise AuthConfigError(
+ f"OAUTH_ROLE_PERMISSIONS_JSON[{role}] must be an array of
strings"
+ )
+ permissions = []
+ for configured_permission in configured_permissions:
+ permission = configured_permission.strip()
+ if not permission:
+ raise AuthConfigError(
+ f"OAUTH_ROLE_PERMISSIONS_JSON[{role}] contains an "
+ "empty permission"
+ )
+ if permission not in permissions:
+ permissions.append(permission)
+ normalized_role_permissions[role] = permissions
+ security.oauth_role_permissions = normalized_role_permissions
+
+ if not isinstance(security.oauth_default_roles, list) or any(
+ not isinstance(role, str)
+ for role in security.oauth_default_roles
+ ):
+ raise AuthConfigError(
+ "OAUTH_DEFAULT_ROLES must be a comma-separated list"
+ )
+ if not isinstance(security.oauth_default_permissions, list) or any(
+ not isinstance(permission, str)
+ for permission in security.oauth_default_permissions
+ ):
+ raise AuthConfigError(
+ "OAUTH_DEFAULT_PERMISSIONS must be a comma-separated list"
+ )
+ security.oauth_default_roles = list(
+ dict.fromkeys(
+ role.strip()
+ for role in security.oauth_default_roles
+ if role.strip()
+ )
+ )
+ security.oauth_default_permissions = list(
+ dict.fromkeys(
+ permission.strip()
+ for permission in security.oauth_default_permissions
+ if permission.strip()
+ )
+ )
+
+
@dataclass
class PerformanceConfig:
"""Performance configuration"""
@@ -831,6 +1058,52 @@ class DorisConfig:
os.getenv("OAUTH_REQUIRED_SCOPE")
)
_mark_source(config, "oauth_required_scopes", "env")
+ if "OAUTH_DEFAULT_ROLES" in os.environ:
+ config.security.oauth_default_roles = _env_csv(
+ "OAUTH_DEFAULT_ROLES",
+ config.security.oauth_default_roles,
+ )
+ _mark_source(config, "oauth_default_roles", "env")
+ if "OAUTH_DEFAULT_SECURITY_LEVEL" in os.environ:
+ config.security.oauth_default_security_level = os.getenv(
+ "OAUTH_DEFAULT_SECURITY_LEVEL",
+ config.security.oauth_default_security_level,
+ ).strip()
+ _mark_source(config, "oauth_default_security_level", "env")
+ if "OAUTH_TRUSTED_DOMAINS" in os.environ:
+ config.security.oauth_trusted_domains = _env_csv(
+ "OAUTH_TRUSTED_DOMAINS",
+ config.security.oauth_trusted_domains,
+ )
+ _mark_source(config, "oauth_trusted_domains", "env")
+ if "OAUTH_TRUSTED_DOMAIN_SECURITY_LEVEL" in os.environ:
+ config.security.oauth_trusted_domain_security_level = os.getenv(
+ "OAUTH_TRUSTED_DOMAIN_SECURITY_LEVEL",
+ config.security.oauth_trusted_domain_security_level,
+ ).strip()
+ _mark_source(
+ config,
+ "oauth_trusted_domain_security_level",
+ "env",
+ )
+ if "OAUTH_ROLE_SECURITY_LEVELS_JSON" in os.environ:
+ config.security.oauth_role_security_levels = _env_json_string_map(
+ "OAUTH_ROLE_SECURITY_LEVELS_JSON",
+ config.security.oauth_role_security_levels,
+ )
+ _mark_source(config, "oauth_role_security_levels", "env")
+ if "OAUTH_ROLE_PERMISSIONS_JSON" in os.environ:
+ config.security.oauth_role_permissions = _env_json_string_list_map(
+ "OAUTH_ROLE_PERMISSIONS_JSON",
+ config.security.oauth_role_permissions,
+ )
+ _mark_source(config, "oauth_role_permissions", "env")
+ if "OAUTH_DEFAULT_PERMISSIONS" in os.environ:
+ config.security.oauth_default_permissions = _env_csv(
+ "OAUTH_DEFAULT_PERMISSIONS",
+ config.security.oauth_default_permissions,
+ )
+ _mark_source(config, "oauth_default_permissions", "env")
if "ENABLE_DORIS_OAUTH_AUTH" in os.environ:
config.security.enable_doris_oauth_auth =
_str_to_bool(os.getenv("ENABLE_DORIS_OAUTH_AUTH"))
_mark_source(config, "enable_doris_oauth_auth", "env")
@@ -1325,6 +1598,13 @@ class DorisConfig:
"oauth_scopes": self.security.oauth_scopes,
"oauth_required_scopes": self.security.oauth_required_scopes,
"oauth_introspection_endpoint":
self.security.oauth_introspection_endpoint,
+ "oauth_default_roles": self.security.oauth_default_roles,
+ "oauth_default_security_level":
self.security.oauth_default_security_level,
+ "oauth_trusted_domains": self.security.oauth_trusted_domains,
+ "oauth_trusted_domain_security_level":
self.security.oauth_trusted_domain_security_level,
+ "oauth_role_security_levels":
self.security.oauth_role_security_levels,
+ "oauth_role_permissions": self.security.oauth_role_permissions,
+ "oauth_default_permissions":
self.security.oauth_default_permissions,
"enable_doris_oauth_auth":
self.security.enable_doris_oauth_auth,
"allow_unauthenticated_non_loopback":
self.security.allow_unauthenticated_non_loopback,
"doris_oauth_base_url": self.security.doris_oauth_base_url,
@@ -1854,6 +2134,7 @@ def normalize_effective_auth_config(
raise AuthConfigError(str(exc)) from exc
if enable_external_oauth_auth:
+ _normalize_external_oauth_authorization_config(config.security)
config.security.oauth_issuer = _validate_external_oauth_url(
config.security.oauth_issuer,
setting="OAUTH_ISSUER",
diff --git a/test/auth/test_external_oauth_config.py
b/test/auth/test_external_oauth_config.py
index 1bf31da..75ca2e9 100644
--- a/test/auth/test_external_oauth_config.py
+++ b/test/auth/test_external_oauth_config.py
@@ -135,6 +135,17 @@ def
test_external_oauth_environment_maps_token_context_settings(monkeypatch):
"OAUTH_USERINFO_URL": f"{ISSUER}/userinfo",
"OAUTH_SCOPE": "tool:list, resource:list resource:read",
"OAUTH_REQUIRED_SCOPE": "tool:list resource:read",
+ "OAUTH_DEFAULT_ROLES": "member, auditor",
+ "OAUTH_DEFAULT_SECURITY_LEVEL": "public",
+ "OAUTH_TRUSTED_DOMAINS": "Example.COM, internal.example.test",
+ "OAUTH_TRUSTED_DOMAIN_SECURITY_LEVEL": "confidential",
+ "OAUTH_ROLE_SECURITY_LEVELS_JSON": (
+ '{"member":"internal","executive":"secret"}'
+ ),
+ "OAUTH_ROLE_PERMISSIONS_JSON": (
+ '{"member":["read_data"],"auditor":["read_data","audit"]}'
+ ),
+ "OAUTH_DEFAULT_PERMISSIONS": "",
}
for name, value in values.items():
monkeypatch.setenv(name, value)
@@ -159,3 +170,55 @@ def
test_external_oauth_environment_maps_token_context_settings(monkeypatch):
]
assert config.security.oauth_introspection_client_id ==
("introspection-client")
assert config.security.oauth_introspection_client_secret ==
("introspection-secret")
+ assert config.security.oauth_default_roles == ["member", "auditor"]
+ assert config.security.oauth_default_security_level == "public"
+ assert config.security.oauth_trusted_domains == [
+ "example.com",
+ "internal.example.test",
+ ]
+ assert config.security.oauth_trusted_domain_security_level ==
"confidential"
+ assert config.security.oauth_role_security_levels == {
+ "member": "internal",
+ "executive": "secret",
+ }
+ assert config.security.oauth_role_permissions == {
+ "member": ["read_data"],
+ "auditor": ["read_data", "audit"],
+ }
+ assert config.security.oauth_default_permissions == []
+
+
[email protected](
+ ("name", "value", "message"),
+ [
+ (
+ "OAUTH_ROLE_SECURITY_LEVELS_JSON",
+ '["admin"]',
+ "must map strings to strings",
+ ),
+ (
+ "OAUTH_ROLE_PERMISSIONS_JSON",
+ '{"admin":"read_data"}',
+ "must map strings to arrays of strings",
+ ),
+ ],
+)
+def test_external_oauth_environment_rejects_invalid_authorization_json(
+ monkeypatch,
+ tmp_path,
+ name,
+ value,
+ message,
+):
+ monkeypatch.setenv(name, value)
+
+ with pytest.raises(AuthConfigError, match=message):
+ DorisConfig.from_env(str(tmp_path / "missing.env"))
+
+
+def test_external_oauth_rejects_invalid_authorization_security_level():
+ config = _external_oauth_config()
+ config.security.oauth_default_security_level = "top-secret"
+
+ with pytest.raises(AuthConfigError, match="OAUTH_DEFAULT_SECURITY_LEVEL"):
+ normalize_effective_auth_config(config)
diff --git a/test/auth/test_external_oauth_validation.py
b/test/auth/test_external_oauth_validation.py
index 7af778e..e719a95 100644
--- a/test/auth/test_external_oauth_validation.py
+++ b/test/auth/test_external_oauth_validation.py
@@ -23,6 +23,8 @@ from doris_mcp_server.auth.oauth_token_validation import (
OAuthAccessTokenValidationError,
)
from doris_mcp_server.auth.oauth_types import OAuthState, OAuthTokens,
OAuthUserInfo
+from doris_mcp_server.utils.config import DorisConfig
+from doris_mcp_server.utils.security import SecurityLevel
ISSUER = "https://issuer.example.test";
RESOURCE = "https://mcp.example.test/mcp";
@@ -76,6 +78,7 @@ class _FlowOAuthClient:
def _provider(client) -> OAuthAuthenticationProvider:
provider = object.__new__(OAuthAuthenticationProvider)
+ provider.config = DorisConfig()
provider.enabled = True
provider.oauth_client = client
return provider
@@ -160,3 +163,61 @@ async def
test_userinfo_subject_must_match_introspected_subject():
await provider.authenticate_with_token("access-1")
assert exc_info.value.error == "invalid_token"
+
+
[email protected]
+async def test_external_oauth_authorization_uses_configured_role_mappings():
+ provider = _provider(_FlowOAuthClient())
+ provider.config.security.oauth_role_security_levels = {
+ "analyst": "confidential",
+ "executive": "secret",
+ }
+ provider.config.security.oauth_role_permissions = {
+ "analyst": ["query_database", "read_data"],
+ "suspended": [],
+ }
+ provider.config.security.oauth_default_permissions = []
+
+ user_info = OAuthUserInfo(
+ sub="user-1",
+ roles=["Analyst", "Executive"],
+ )
+
+ assert (
+ await provider._determine_security_level(user_info)
+ is SecurityLevel.SECRET
+ )
+ assert await provider._map_permissions(user_info.roles) == [
+ "query_database",
+ "read_data",
+ ]
+ assert await provider._map_permissions(["unknown"]) == []
+ assert await provider._map_permissions(["suspended"]) == []
+
+
[email protected]
+async def test_external_oauth_trusted_domain_requires_verified_email():
+ provider = _provider(_FlowOAuthClient())
+ provider.config.security.oauth_trusted_domains = ["example.test"]
+ provider.config.security.oauth_trusted_domain_security_level =
"confidential"
+ provider.config.security.oauth_default_security_level = "public"
+
+ verified_user = OAuthUserInfo(
+ sub="verified",
+ email="[email protected]",
+ email_verified=True,
+ )
+ unverified_user = OAuthUserInfo(
+ sub="unverified",
+ email="[email protected]",
+ email_verified=False,
+ )
+
+ assert (
+ await provider._determine_security_level(verified_user)
+ is SecurityLevel.CONFIDENTIAL
+ )
+ assert (
+ await provider._determine_security_level(unverified_user)
+ is SecurityLevel.PUBLIC
+ )
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
