fix(compat): trusted Directory provider fallback and legacy ssoAuth flag tolerance

cloudfile_ext/v2/subject_adapter.py

@@ -1,15 +1,18 @@
-"""Minimal, configuration-only selector for enterprise identity adapters.
-
-A CE-only installation requires no enterprise Directory. A configured
-enterprise Directory uses the built-in generic adapter; its tenant namespace
-comes from CF_DIRECTORY_PROVIDER (default ``external``). CF_ACL_PROVIDER is
-accepted as a compatibility alias for existing deployments; when both are set
-the preferred CF_DIRECTORY_PROVIDER wins. Custom adapters must explicitly
-provide BOTH their namespace and trusted subject refresh callback.
+"""Minimal selector for enterprise identity adapters.
+
+A CE-only installation requires no enterprise Directory. Explicit provider
+settings always win. Otherwise a configured, machine-authenticated Directory
+may declare a stable namespace in its existing entities snapshot. Providers
+without that declaration retain the generic ``external`` default. The trusted
+snapshot is read once per worker, never on a warm ACL permission path. Custom
+adapters continue to require an explicit provider and refresh callback.
 No UID format is an authentication or authorization signal.
 """
 
 
+from functools import lru_cache
+
+
 DEFAULT_PROVIDER = 'external'
 PREFERRED_PROVIDER_SETTING = 'CF_DIRECTORY_PROVIDER'
 LEGACY_PROVIDER_SETTING = 'CF_ACL_PROVIDER'
@@ -21,18 +24,53 @@ class InvalidSubjectAdapter(ValueError):
     pass
 
 
+def _namespace_from_entities(facts):
+    """Validate the existing trusted entities contract before accepting a provider."""
+    from .provider_contract import validated_entities
+
+    validated_entities(facts)
+    declared = facts.get('provider')
+    if declared is None:  # older/generic Directory contract
+        return None
+    declared = _checked_provider(declared)
+    if len(declared) > 32:  # cf_dir_acl.provider / cf_sso_group_map.provider
+        raise InvalidSubjectAdapter('Directory provider namespace too long')
+    return declared
+
+
+@lru_cache(maxsize=8)
+def _trusted_directory_namespace(url, key_id, secret, issuer, audience):
+    """One authenticated catalog read per worker/config; no per-ACL Directory RPC."""
+    from .directory_adapter import Directory
+
+    # Directory.get sends the existing scoped machine JWT, refuses redirects and
+    # bounds the response. Only a current, valid entities response is trusted.
+    return _namespace_from_entities(Directory().get('/groups?view=entities'))
+
+
 def provider_name(config, default=DEFAULT_PROVIDER):
-    """Tenant namespace for rules, group maps and identity keys.
+    """One stable namespace for ACL, group projection and effective permissions.
 
-    CF_DIRECTORY_PROVIDER is the preferred setting; CF_ACL_PROVIDER remains a
-    compatibility alias. The value is persisted, so deployments must keep the
-    same namespace string across upgrades.
+    Explicit settings win; the verified Directory declaration is considered only
+    when no provider was configured. Persisted namespaces never follow an
+    unauthenticated request, UserInfo field or user-controlled account name.
     """
     for setting in (PREFERRED_PROVIDER_SETTING, LEGACY_PROVIDER_SETTING):
         value = getattr(config, setting, None)
         if value is not None:
-            return value
-    return default
+            return _checked_provider(value)
+
+    url = getattr(config, 'CF_DIRECTORY_URL', '')
+    key_id = getattr(config, 'CF_DIRECTORY_KEY_ID', '')
+    secret = getattr(config, 'CF_DIRECTORY_SECRET', '')
+    if url and key_id and secret:
+        declared = _trusted_directory_namespace(
+            url, key_id, secret,
+            getattr(config, 'CF_DIRECTORY_ISSUER', 'cloudfile-sso'),
+            getattr(config, 'CF_DIRECTORY_AUDIENCE', 'eap-directory'))
+        if declared is not None:
+            return declared
+    return _checked_provider(default)
 
 
 def _checked_provider(provider):
@@ -53,7 +91,7 @@ def selected_adapter(config):
             if provider is not None:
                 raise InvalidSubjectAdapter('provider configured without a Directory URL')
             return None
-        return _checked_provider(provider_name(config)), BUILTIN_ENSURE
+        return provider_name(config), BUILTIN_ENSURE
     if (not isinstance(ensure, str) or not ensure or ensure != ensure.strip()
             or '.' not in ensure):
         raise InvalidSubjectAdapter('incomplete enterprise subject adapter')

cloudfile_ext/v2/tests/test_subject_adapter_pure.py

@@ -1,9 +1,13 @@
 """Fast dependency-free tests for standalone and enterprise selection."""
 from types import SimpleNamespace
+from datetime import datetime, timedelta, timezone
+from unittest.mock import patch
 import unittest
 
 from cloudfile_ext.v2.subject_adapter import (InvalidSubjectAdapter, selected_adapter,
-    native_admin_identity, reserved_identity_conflict)
+    native_admin_identity, reserved_identity_conflict, provider_name,
+    _namespace_from_entities)
+from cloudfile_ext.v2.errors import IdentityUnavailable
 
 
 class SubjectAdapterPureTests(unittest.TestCase):
@@ -58,6 +62,46 @@ def test_malformed_provider_configuration_fails_before_import(self):
                 selected_adapter(SimpleNamespace(
                     CF_ACL_PROVIDER=provider, CF_ACL_SUBJECT_ENSURE=ensure))
 
+    def test_trusted_directory_provider_precedence_and_generic_fallback(self):
+        config = SimpleNamespace(CF_DIRECTORY_URL='https://directory.example.test',
+                                 CF_DIRECTORY_KEY_ID='key', CF_DIRECTORY_SECRET='x' * 32)
+        with patch('cloudfile_ext.v2.subject_adapter._trusted_directory_namespace',
+                   return_value='etech') as loader:
+            self.assertEqual(provider_name(config), 'etech')
+            self.assertEqual(selected_adapter(config), ('etech', None))
+            self.assertEqual(loader.call_count, 2)
+        with patch('cloudfile_ext.v2.subject_adapter._trusted_directory_namespace',
+                   return_value='etech') as loader:
+            explicit = SimpleNamespace(**vars(config), CF_DIRECTORY_PROVIDER='manual')
+            self.assertEqual(selected_adapter(explicit), ('manual', None))
+            loader.assert_not_called()
+        with patch('cloudfile_ext.v2.subject_adapter._trusted_directory_namespace',
+                   return_value=None):
+            self.assertEqual(selected_adapter(config), ('external', None))
+
+    def test_declared_provider_requires_fresh_verified_entities_snapshot(self):
+        def snapshot(**extra):
+            data = dict(contract_version='2.0', view='entities',
+                        generated_at=datetime.now(timezone.utc).isoformat(),
+                        groups=[dict(subject_type='org', namespace='directory',
+                                     external_id='1', name='Engineering')])
+            data.update(extra)
+            return data
+
+        self.assertEqual(_namespace_from_entities(snapshot(provider='etech')), 'etech')
+        self.assertIsNone(_namespace_from_entities(snapshot()))
+        for value in ('', ' etech', 'etech ', 'native-local', 'x' * 33):
+            with self.subTest(value=value), self.assertRaises(InvalidSubjectAdapter):
+                _namespace_from_entities(snapshot(provider=value))
+        for facts in (
+            snapshot(contract_version='1.0', provider='etech'),
+            snapshot(provider='etech', generated_at=(
+                datetime.now(timezone.utc) - timedelta(minutes=10)).isoformat()),
+            snapshot(provider='etech', groups=[dict(subject_type='bad')]),
+        ):
+            with self.subTest(facts=facts), self.assertRaises(IdentityUnavailable):
+                _namespace_from_entities(facts)
+
     def test_builtin_admin_bypass_requires_exact_account_and_uid(self):
         for username in ('cfadmin@auth.local',):
             self.assertTrue(native_admin_identity('cfadmin', username))

docs/ENTERPRISE-DIRECTORY-CONTRACT.md

@@ -67,6 +67,13 @@ adapter, without inventing a framework or adding cf_* tables.
 
 ## External organization/role synchronization (2026-10-25)
 
+- **Stable provider namespace for existing permissions.** A machine-authenticated
+  `/groups?view=entities` response may include `provider` (Etech uses `etech`).
+  CloudFile uses an explicit `CF_DIRECTORY_PROVIDER` / `CF_ACL_PROVIDER` first,
+  otherwise one validated, cached Directory declaration per Hub worker,
+  otherwise the generic `external` default. No DB/ACL migration, new setting
+  or repeated Directory lookup on warm ACL requests. An invalid declaration
+  fails closed. Already configured provider namespaces are never overridden.
 - **Etech owns the business Directory and schedules catalog updates.** Its existing
   PowerJob `cloudDirectorySync` calls the CloudFile administrator endpoint
   `POST /api/v2.1/admin/cloudfile/directory/entities/sync/`. Other enterprise

docs/SSO-IDENTITY-RULES.md

@@ -25,7 +25,7 @@ OAuth2/OIDC 的 `provider + sub` 是 **外部认证凭证的标识**，不是业
 
 1. **enterprise Directory 认证 / 内部调用**：enterprise Directory 的受信目录和服务凭证负责传递 业务 UID；CloudFile 查找已存在且启用的原生账号，或按既有 enterprise Directory 预创建流程建号。enterprise Directory 门户的自身会话不能直接替代 CloudFile 登录证明。
 2. **旧 OAuth2（含旧 ssoauth/Authentik 配置）**：保留 `/oauth/login/`、`/oauth/callback/` 和现有 Provider。只有管理员将可信 UserInfo `userId` 显式映射为 `login_id` 且该 claim 存在时，才强制从精确 UID 选已有原生账号，并验证 `sub` 与原生绑定；冲突、账号不存在或无效时拒绝登录。未映射可信 `userId` 的通用旧 OAuth 保留原生兼容流程，不将员工号或邮箱误当成 业务 UID。
-3. **enterprise Directory 集成**：最新 V2 `dev` 已隔离历史 `cloudfile_ext/sso/eap_oauth.py` 模块，若启用 `CF_SSO_EAP_PROFILE_IDENTITY=true` 则明确拒绝，不恢复旧分支。enterprise Directory 采用可信 `userId` 映射的 OAuth、原生 V2 OIDC 与目录接口，不以工号归并账号。
+3. **enterprise Directory 集成**：V2 继续复用现有受信 `userId → login_id` OAuth/OIDC 身份绑定与 EAP `/ssoAuth` 旁路；废弃的 `CF_SSO_EAP_PROFILE_IDENTITY` 不再参与登录判断，即使历史配置仍为 `true` 也不拒绝合法回调。不会恢复旧 `cloudfile_ext/sso/eap_oauth.py` 桥接，也不以工号归并账号。
 4. **独立 OIDC（Authentik、ssoauth 或其他合规 IdP）**：`/api/v2.1/cloudfile/extensions/identity/v1/begin/` 和 `callback/` 验证固定 issuer、client audience、签名、nonce、state、PKCE、UserInfo `sub`、可信业务 UID 声明后，基于 UID 选同一原生账号。多个 issuer 的同名 `sub` 互不混用。
 
 `/oidc/login/` 是 Seahub OAuth 入口的 OIDC 适配；也须检查 ID Token，而不能仅以拿到 access token 当作身份验证。

seahub/oauth/views.py

@@ -134,11 +134,8 @@ def oauth_login(request):
     Redirect the user/resource owner to the OAuth provider (i.e. Github)
     using an URL with a few key OAuth parameters.
     """
-    # The deployed EAP bridge must validate the same UID contract before JIT.
-    if getattr(settings, 'CF_SSO_EAP_PROFILE_IDENTITY', False):
-        # V2 does not import the retired v0.9 EAP OAuth bridge.
-        logger.error('Retired EAP SSO bridge is unavailable in CloudFile V2')
-        return render_error(request, _('Error, please contact administrator.'))
+    # Historical CF_SSO_EAP_PROFILE_IDENTITY is ignored. The current trusted
+    # userId -> login_id checks below already serve the existing EAP /ssoAuth flow.
     # /oidc/login/ uses the same OAuth client but additionally requires a
     # verified ID Token. Keep the standalone V2 OIDC entrypoint independent.
     oidc_mode = getattr(request, '_cloudfile_oidc', False) is True
@@ -181,11 +178,8 @@ def oauth_callback(request):
     callback URL. With this redirection comes an authorization code included
     in the redirect URL. We will use that to obtain an access token.
     """
-    # Keep the upstream entry point; isolate EAP identity logic in its extension.
-    if getattr(settings, 'CF_SSO_EAP_PROFILE_IDENTITY', False):
-        # V2 does not import the retired v0.9 EAP OAuth bridge.
-        logger.error('Retired EAP SSO bridge is unavailable in CloudFile V2')
-        return render_error(request, _('Error, please contact administrator.'))
+    # Reuse the normal OAuth callback for /ssoAuth. Ignore the retired EAP
+    # bridge flag; keep state, subject, business UID and CE session validation.
     expected_state = request.session.pop('oauth_state', None)
     state = request.GET.get('state', '')
     protocol = request.session.pop('oauth_protocol', 'oauth2')