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 @@ 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 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 @@ class AuthConfigError(ValueError): 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 @@ def __post_init__(self) -> None: 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 @@ def from_env(cls, env_file: str | None = None) -> "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 @@ def to_dict(self) -> dict[str, Any]: "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 == [] + + +@pytest.mark.parametrize( + ("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 @@ 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 @@ async def get_user_info(self, tokens): 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" + + +@pytest.mark.asyncio +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"]) == [] + + +@pytest.mark.asyncio +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="user@example.test", + email_verified=True, + ) + unverified_user = OAuthUserInfo( + sub="unverified", + email="user@example.test", + 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 + )