perf(auth): periodic directory catalog and no sync on permission hot paths

cloudfile_ext/apps.py

@@ -28,6 +28,7 @@ def ready(self):
             module.register(registry)
 
         self._register_native_acl_gate(registry)
+        self._register_directory_catalog(registry)
 
         registry.seal()
 
@@ -49,6 +50,23 @@ def _register_native_acl_gate(registry):
         from cloudfile_ext.v2 import native_gate
         registry.register_permission_check(native_gate.check_permission)
 
+    @staticmethod
+    def _register_directory_catalog(registry):
+        """Sync enterprise org/role catalog on the existing worker, never on ACL reads.
+
+        The login callback projects an authenticated existing user's memberships;
+        a missing group or user in an ordinary file request does not create it.
+        """
+        from django.conf import settings
+
+        if not getattr(settings, 'CF_DIRECTORY_URL', ''):
+            return
+        interval = getattr(settings, 'CF_DIRECTORY_SYNC_INTERVAL_SECONDS', 600)
+        if type(interval) is not int or not 60 <= interval <= 86400:
+            raise ValueError('CF_DIRECTORY_SYNC_INTERVAL_SECONDS must be 60..86400')
+        from cloudfile_ext.v2.directory_adapter import sync_entity_catalog
+        registry.register_periodic_task('directory-entities', interval, sync_entity_catalog)
+
     def _configure_database(self):
         """Add the seafile-db connection and the cf_* router.
 

cloudfile_ext/management/commands/cf_worker.py

@@ -6,15 +6,11 @@
 the cf-worker container runs this command from the same image as the
 application.
 
-Registration happens in ``CloudFileConfig.ready()``. The V2 runtime registers
-no periodic task today: the SSO/search/external-source modules that used to own
-one are retired, and their tables are not part of the V2 schema (SSO directory
-sync is the push trigger at
-``/api/v2.1/admin/cloudfile/directory/entities/sync/``, and the Meilisearch
-indexer is the isolated ``cloudfile_ext.v2.search.worker`` process rather than
-a registry task). With no enabled task the command exits 0 immediately, so a
-deployment must not restart the container on a clean exit -- only a real
-failure is worth retrying.
+Registration happens in ``CloudFileConfig.ready()``. Enterprise Directory
+org/role catalog synchronization uses this worker at a default 600-second
+interval. It never provisions users or GroupUser during permission checks.
+Login remains the existing user membership synchronization point. With no
+enabled periodic task this command exits 0 immediately.
 """
 
 import logging

cloudfile_ext/v2/acl_http.py

@@ -421,11 +421,18 @@ def post(self, request):
             checks = data['checks']
             if not isinstance(checks, list) or not 1 <= len(checks) <= 4096:
                 raise ValueError()
-            authority = Authority(data['username'])
-            permissions = []
+            # One authorization authority per call. Writes must read current
+            # ACL rows, not a cross-request TTL cache; per-request decisions
+            # still reuse the same library/native permission snapshot.
             for check in checks:
-                if set(check) != {'path', 'kind', 'action'} or check['action'] not in ('read', 'write'):
+                if (not isinstance(check, dict) or set(check) != {'path', 'kind', 'action'}
+                        or check['action'] not in ('read', 'write')):
                     raise ValueError()
+            only_write = not partial and all(check['action'] == 'write' for check in checks)
+            authority = Authority(data['username'], cache_rules=not any(
+                check['action'] == 'write' for check in checks))
+            permissions = []
+            for check in checks:
                 decision = authority.effective(data['repo_id'], check['path'], check['kind'])
                 if not (decision.can_read if check['action'] == 'read' else decision.can_write):
                     if not partial:
@@ -438,8 +445,13 @@ def post(self, request):
             # when it has no library grant at all, and it needs to know whether
             # the qualified share root is r or rw: probing again with
             # action=write would double the round trip for every write.
-            return response({'allowed': True, 'contract': 'cf-acl-v1',
-                             'permissions': permissions})
+            # Final-commit callers need an all-or-deny verdict, not 4096
+            # repeated 'rw' strings. Omit levels for write-only batches so the
+            # existing 4 KB native HTTP reply limit never rejects a valid batch.
+            result = {'allowed': True, 'contract': 'cf-acl-v1'}
+            if not only_write:
+                result['permissions'] = permissions
+            return response(result)
         except Exception as error:
             return failure(error)
 

cloudfile_ext/v2/acl_runtime.py

@@ -107,7 +107,10 @@ def current_principal(username):
     # A configured Directory owns *all* other external UIDs, including strings
     # with letters. Missing subjects and expired memberships fail closed; never
     # treat a non-numeric UID as a native user with broader CE access.
-    import_string(ensure)(uid, username)
+    if ensure is not None:
+        # Explicit custom adapters must be read-only; builtin CE projection is
+        # populated on login and never refreshed on an authorization request.
+        import_string(ensure)(uid, username)
     schema = identifier(SeafileDB().db_name)
     accounts = identifier(get_ccnet_db_name())
     with connection.cursor() as cursor:
@@ -121,14 +124,15 @@ def current_principal(username):
 
 
 class Authority:
-    def __init__(self, username, *, principal=None, rules=None, api=None):
+    def __init__(self, username, *, principal=None, rules=None, api=None, cache_rules=True):
         self.username = username
         self.principal = current_principal(username) if principal is None else principal
         self.store = store() if rules is None else rules
         if api is None:
             from seaserv import seafile_api
             api = seafile_api
         self.api = api
+        self.cache_rules = cache_rules
         self._repos = {}
         self._virtual = {}
         self._versions = {}
@@ -139,7 +143,7 @@ def __init__(self, username, *, principal=None, rules=None, api=None):
 
     def _rules(self, repo_id, directory):
         if repo_id not in self._library:
-            rows = library_rules(self.store, repo_id, cache_ttl())
+            rows = library_rules(self.store, repo_id, cache_ttl() if self.cache_rules else 0)
             # One shared library snapshot for both read and write checks.
             # Per-request path buckets avoid rescanning every rule per child
             # of a large directory page; these are NOT another Redis cache.

cloudfile_ext/v2/directory_adapter.py

@@ -229,49 +229,62 @@ def subject_ready(uid, username):
 
 
 def ensure_subject(uid, username):
+    """Login-only refresh for an already authenticated/created native account.
+
+    Never create missing groups as a side effect of user login or ACL lookup.
+    The periodic catalog synchronizer owns organization/role creation; a
+    missing mapped entity refuses membership projection until the next sync.
+    """
     if subject_ready(uid, username):
         return
     with runtime() as (directory, identity, catalog, memberships):
-        # Entity synchronization never expands users. Catalog is consulted only
-        # after the subject TTL expires, including newly assigned departments.
-        def fetch(current_uid):
-            subject = directory.subject(current_uid)
-            _, desired = validate_subject(subject, current_uid)
-            if desired:
-                entities = directory.entities()
-                if not desired <= entities.keys():
-                    raise IdentityUnavailable('Directory membership entity missing')
-                for key in sorted(desired):
-                    catalog.ensure(*key, entities[key])
-            return subject
-        memberships.fetch = fetch
         memberships.ensure(uid, username)
 
 
 def sync_entity_catalog():
-    """PowerJob/manual entity reconciliation, without any user/member writes.
+    """Scheduled/manual org+role reconciliation; no user or GroupUser writes.
 
-    enterprise Directory remains the only organization truth. Preserve deleted/retired CE
-    groups until a separate reviewed cleanup: they may own libraries.
+    Read current native mappings once. Unchanged thousands of groups must not
+    each acquire a provider lock, query CE and call ccnet RPC every ten minutes.
+    Missing/deleted users are deliberately outside this task.
     """
     from seaserv import ccnet_api
     with runtime() as (directory, identity, catalog, memberships):
-        entities = directory.entities()  # Already validated /groups?view=entities.
+        entities = directory.entities()
         if not entities:
             raise IdentityUnavailable('empty entity catalog cannot be treated as a successful sync')
-        renamed = unchanged = 0
+
+        with catalog.connection.cursor() as cursor:
+            cursor.execute(
+                'SELECT m.subject_type,m.namespace,m.external_id,m.group_id,'
+                'g.group_name,g.parent_group_id FROM cf_sso_group_map m '
+                'LEFT JOIN ' + catalog.groups + ' g ON g.group_id=m.group_id '
+                'WHERE m.provider=%s', (catalog.provider,))
+            rows = cursor.fetchall()
+        if len(rows) > 16384:
+            raise IdentityUnavailable('native group catalog too large')
+        mapped = {}
+        for kind, namespace, external_id, gid, label, parent in rows:
+            key = (kind, namespace, external_id)
+            if (kind not in ('org', 'role') or key in mapped or
+                    not isinstance(gid, int) or not isinstance(label, str) or parent != 0):
+                raise IdentityUnavailable('invalid mapped native organization/role')
+            mapped[key] = (gid, label)
+
+        created = renamed = unchanged = 0
         for key, name in sorted(entities.items()):
-            group_id = catalog.ensure(*key, name)
-            group = ccnet_api.get_group(group_id)
-            if group is None:
-                raise IdentityUnavailable('mapped native group missing')
-            if group.group_name != name:
-                ccnet_api.set_group_name(group_id, name)
-                actual = ccnet_api.get_group(group_id)
-                if actual is None or actual.group_name != name:
-                    raise IdentityUnavailable('native group rename not confirmed')
-                renamed += 1
-            else:
+            if key not in mapped:
+                catalog.ensure(*key, name)
+                created += 1
+                continue
+            group_id, current_name = mapped[key]
+            if current_name == name:
                 unchanged += 1
-        return dict(status='OK', entities=len(entities), renamed=renamed,
-                    unchanged=unchanged, membership_writes=0)
+                continue
+            ccnet_api.set_group_name(group_id, name)
+            actual = ccnet_api.get_group(group_id)
+            if actual is None or actual.group_name != name:
+                raise IdentityUnavailable('native group rename not confirmed')
+            renamed += 1
+        return dict(status='OK', entities=len(entities), created=created,
+                    renamed=renamed, unchanged=unchanged, membership_writes=0)

cloudfile_ext/v2/memberships.py

@@ -120,26 +120,55 @@ def _project(self, uid, username, desired):
         try:
             self.identity.active(username, uid, locked=True)
             with self.connection.cursor() as cursor:
-                cursor.execute('SELECT subject_type,namespace,external_id,group_id FROM cf_sso_group_map WHERE provider=%s ORDER BY group_id FOR UPDATE', (self.provider,))
-                rows = cursor.fetchall()
-                if len(rows) > 16384:
-                    raise IdentityUnavailable('group catalog too large')
-                mappings = {(SUBJECT_KIND.get(row[0], row[0]), row[1], row[2]): row[3] for row in rows}
-                owned = set(mappings.values())
-                if len(owned) != len(rows) or not desired <= mappings.keys():
-                    raise IdentityUnavailable('required native group mapping missing')
-                # Flat native groups avoid implicit CE hierarchy grants.
-                # Only scope-approved members are ever projected to GroupUser.
-                for group in sorted(owned):
-                    cursor.execute('SELECT group_id,parent_group_id FROM ' + self.groups + ' WHERE group_id=%s FOR UPDATE', (group,))
-                    if cursor.fetchall() != ((group, 0),):
-                        raise IdentityUnavailable('native mapped group changed')
-                cursor.execute('SELECT group_id,is_staff FROM ' + self.members + ' WHERE user_name=%s ORDER BY group_id FOR UPDATE', (username,))
+                # Read just this user's current groups, not every group owned by
+                # the enterprise Provider. GroupUser rows are the only mutable
+                # membership facts, and the provider advisory lock prevents the
+                # scheduled catalog writer from changing mappings mid-refresh.
+                cursor.execute('SELECT group_id,is_staff FROM ' + self.members +
+                               ' WHERE user_name=%s ORDER BY group_id FOR UPDATE', (username,))
                 members = cursor.fetchall()
+                existing = {group for group, _ in members}
+                if len(existing) != len(members):
+                    raise IdentityUnavailable('duplicate native membership')
+                owned = set()
+                if existing:
+                    sql = ('SELECT group_id FROM cf_sso_group_map WHERE provider=%s '
+                           'AND group_id IN (' + ','.join(['%s'] * len(existing)) + ')')
+                    cursor.execute(sql, [self.provider, *sorted(existing)])
+                    rows = cursor.fetchall()
+                    owned = {row[0] for row in rows}
+                    if len(owned) != len(rows):
+                        raise IdentityUnavailable('duplicate provider group mapping')
+
+                required = set()
+                if desired:
+                    subjects = sorted(desired)
+                    sql = ('SELECT subject_type,namespace,external_id,group_id '
+                           'FROM cf_sso_group_map WHERE provider=%s AND '
+                           '(subject_type,namespace,external_id) IN (' +
+                           ','.join(['(%s,%s,%s)'] * len(subjects)) + ')')
+                    cursor.execute(sql, [self.provider, *(v for key in subjects for v in key)])
+                    mappings = {(SUBJECT_KIND.get(t, t), ns, ext): gid
+                                for t, ns, ext, gid in cursor.fetchall()}
+                    if len(mappings) != len(subjects) or any(key not in mappings for key in subjects):
+                        # Never attempt on-demand catalog sync or create a user
+                        # from an ACL or login membership lookup.
+                        raise IdentityUnavailable('required native group mapping missing')
+                    required = set(mappings.values())
+                    if len(required) != len(subjects):
+                        raise IdentityUnavailable('duplicate native group mapping')
+
+                relevant = owned | required
+                if relevant:
+                    cursor.execute('SELECT group_id,parent_group_id FROM ' + self.groups +
+                                   ' WHERE group_id IN (' + ','.join(['%s'] * len(relevant)) +
+                                   ') FOR UPDATE', sorted(relevant))
+                    group_rows = cursor.fetchall()
+                    if (len(group_rows) != len(relevant) or
+                            any(group not in relevant or parent != 0 for group, parent in group_rows)):
+                        raise IdentityUnavailable('native mapped group changed')
                 if any(group in owned and staff != 0 for group, staff in members):
                     raise IdentityConflict('native group administrator requires review')
-                existing = {group for group, _ in members}
-                required = {mappings[item] for item in desired}
                 for group in sorted((existing & owned) - required):
                     cursor.execute('DELETE FROM ' + self.members + ' WHERE group_id=%s AND user_name=%s AND is_staff=0', (group, username))
                 for group in sorted(required - existing):

cloudfile_ext/v2/subject_adapter.py

@@ -13,7 +13,8 @@
 DEFAULT_PROVIDER = 'external'
 PREFERRED_PROVIDER_SETTING = 'CF_DIRECTORY_PROVIDER'
 LEGACY_PROVIDER_SETTING = 'CF_ACL_PROVIDER'
-BUILTIN_ENSURE = 'cloudfile_ext.v2.directory_adapter.ensure_subject'
+BUILTIN_ENSURE = None  # CE Profile + GroupUser are read-only on permission paths
+LEGACY_ENSURE = 'cloudfile_ext.v2.directory_adapter.ensure_subject'
 
 
 class InvalidSubjectAdapter(ValueError):
@@ -58,7 +59,7 @@ def selected_adapter(config):
         raise InvalidSubjectAdapter('incomplete enterprise subject adapter')
     if provider is None:
         raise InvalidSubjectAdapter('incomplete enterprise subject adapter')
-    return _checked_provider(provider), ensure
+    return _checked_provider(provider), None if ensure == LEGACY_ENSURE else ensure
 
 
 # Only the built-in local CE administrator may bypass the configured enterprise

cloudfile_ext/v2/tests/test_acl_batch_pure.py

@@ -20,7 +20,7 @@ def test_snapshot_order_version_and_provider(self):
 
     def test_same_library_rules_for_reads_and_writes(self):
         source = (Path(__file__).resolve().parents[1] / 'acl_runtime.py').read_text()
-        self.assertIn('rows = library_rules(self.store, repo_id, cache_ttl())', source)
+        self.assertIn('rows = library_rules(self.store, repo_id, cache_ttl() if self.cache_rules else 0)', source)
         self.assertIn('self._rule_paths[repo_id] = buckets', source)
         self.assertIn('return [rule for path in ancestors(directory)', source)
         policy = (Path(__file__).resolve().parents[1] / 'acl_http.py').read_text()

cloudfile_ext/v2/tests/test_acl_http.py

@@ -195,12 +195,23 @@ def test_native_proof_must_be_internal_short_lived_and_current(self):
             authority.return_value.effective.return_value = Decision('rw','/shared',True)
             good = dict(is_internal=True,exp=int(time.time())+60)
             self.assertEqual(self.native(good).status_code,200)
-            authority.assert_called_once_with('native')
+            authority.assert_called_once_with('native', cache_rules=False)
             for bad in (dict(exp=int(time.time())+60),dict(is_internal=True,exp=1),dict(is_internal='true',exp=int(time.time())+60),dict(is_internal=True,exp=int(time.time())+3600)):
                 self.assertEqual(self.native(bad).status_code,403)
             authority.return_value.effective.return_value = Decision('invisible','/shared',True)
             self.assertEqual(self.native(good).status_code,403)
 
+    def test_write_batch_is_one_fresh_authority_and_small_verdict(self):
+        claims = dict(is_internal=True, exp=int(time.time()) + 60)
+        checks = [dict(path='/shared/file', kind='file', action='write')] * 1024
+        with patch('cloudfile_ext.v2.acl_http.Authority') as authority:
+            authority.return_value.effective.return_value = Decision('rw', '/shared', True)
+            result = self.native(claims, dict(username='native', repo_id=RID, checks=checks))
+            self.assertEqual(result.status_code, 200)
+            self.assertTrue(result.data['allowed'])
+            self.assertNotIn('permissions', result.data)
+            authority.assert_called_once_with('native', cache_rules=False)
+
     def test_native_partial_directory_listing_preserves_hidden_children(self):
         claims = dict(is_internal=True, exp=int(time.time()) + 60)
         checks = [