feat(connectors): named permission-sync and group-sync checks for Google Drive (#15805)

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

.agents/feature-map/components/permission-sync.md

@@ -219,8 +219,8 @@ in every connector SDK.
 
 Perm-sync capability checks (`validate_perm_sync` probes and named checks) are
 dispatched from `ee/onyx/connectors/capability_checks.py` and `perm_sync_valid.py`. The
-check implementations live in the connector modules. Confluence, OneDrive, Outlook, and
-Slack have named perm-sync checks. See [[connectors]] §4.8.
+check implementations live in the connector modules. Confluence, Google Drive, OneDrive,
+Outlook, and Slack have named perm-sync checks. See [[connectors]] §4.8.
 
 ### 4.2 Doc sync flow
 

backend/ee/onyx/connectors/capability_checks.py

@@ -20,6 +20,10 @@ from onyx.connectors.confluence.capability_checks import (
     build_confluence_doc_permission_sync_checks,
     build_confluence_group_sync_checks,
 )
+from onyx.connectors.google_drive.capability_checks import (
+    build_google_drive_doc_permission_sync_checks,
+    build_google_drive_group_sync_checks,
+)
 from onyx.connectors.jira.capability_checks import (
     build_jira_doc_permission_sync_checks,
     build_jira_group_sync_checks,
@@ -44,6 +48,7 @@ _DOC_PERMISSION_SYNC_CHECKS_BY_SOURCE: dict[DocumentSource, list[CapabilityCheck
     DocumentSource.OUTLOOK: build_outlook_doc_permission_sync_checks(),
     DocumentSource.ONEDRIVE: build_onedrive_doc_permission_sync_checks(),
     DocumentSource.JIRA: build_jira_doc_permission_sync_checks(),
+    DocumentSource.GOOGLE_DRIVE: build_google_drive_doc_permission_sync_checks(),
 }
 
 # Slack registers nothing here by design: it has no group sync (channel access
@@ -53,6 +58,7 @@ _EXTERNAL_GROUP_SYNC_CHECKS_BY_SOURCE: dict[DocumentSource, list[CapabilityCheck
     DocumentSource.ONEDRIVE: build_onedrive_group_sync_checks(),
     DocumentSource.CONFLUENCE: build_confluence_group_sync_checks(),
     DocumentSource.JIRA: build_jira_group_sync_checks(),
+    DocumentSource.GOOGLE_DRIVE: build_google_drive_group_sync_checks(),
 }
 
 

backend/onyx/connectors/google_drive/capability_checks.py

@@ -23,7 +23,7 @@ without it.
 from collections.abc import Iterator
 from contextlib import contextmanager
 from itertools import islice
-from typing import NoReturn
+from typing import Any, NoReturn
 
 from onyx.configs.constants import DocumentSource
 from onyx.connectors.capability_checks.form_state import FormState
@@ -59,6 +59,7 @@ from onyx.connectors.google_drive.drive_access import (
 )
 from onyx.connectors.google_drive.models import (
     GDriveMimeType,
+    GoogleDirectoryUser,
     GoogleDriveAuth,
     GoogleDriveFileType,
 )
@@ -1026,3 +1027,314 @@ def build_google_drive_indexing_checks() -> list[CapabilityCheck]:
         _ContentReadCheck(),
         _DocsApiCheck(),
     ]
+
+
+# Permission sync. The EE registry registers these checks. Doc sync re-lists
+# the files with their permissions and reads the permissions Drive gives only
+# by id (shared drive files, folders). Group sync reads the Workspace
+# directory (users, groups and their members), the members of each shared
+# drive, and the permissions of each folder. Both kinds of credential need a
+# primary admin who is a Workspace admin.
+
+_PERMISSION_SAMPLE_FIELDS = (
+    "nextPageToken, files(id, name, permissionIds, "
+    "permissions(id, emailAddress, type, domain))"
+)
+_PERMISSION_FIELDS = "permissions(id, emailAddress, type),nextPageToken"
+# Permission reads the file check makes before it gives up.
+_MAX_PERMISSION_READS = 3
+_DIRECTORY_HINT = (
+    "Make the primary admin a Google Workspace admin. Service account: give it "
+    "the admin.directory.user.readonly and admin.directory.group.readonly "
+    "scopes in domain-wide delegation. Turn on the Admin SDK API."
+)
+
+
+def _is_item_denial(error: GoogleDriveHttpError) -> bool:
+    """A 403 or 404 on one file or folder, not a disabled API or a missing
+    scope."""
+    return (
+        error.status_code in (403, 404)
+        and error.access_denied
+        and not _DISABLED_API_REASONS.intersection(error.reasons)
+        and not _SCOPE_REASONS.intersection(error.reasons)
+    )
+
+
+def _read_permissions(
+    context: CapabilityCheckContext,
+    ops: GoogleDriveSourceOperations,
+    user_email: str,
+    item: GoogleDriveFileType,
+) -> bool:
+    """Reads the permissions of a file or folder that lists them only by id,
+    as the sync does. False when Google denies them for this item or returns
+    none."""
+    try:
+        permissions: list[dict[str, Any]] = list(
+            ops.list_file_permissions(
+                user_email=user_email, file_id=item["id"], fields=_PERMISSION_FIELDS
+            )
+        )
+    except GoogleDriveHttpError as e:
+        if _is_item_denial(e):
+            return False
+        _raise_google_error(
+            context, e, api=_DRIVE_API, denied="The credential cannot read permissions"
+        )
+    except GoogleDriveRefreshError as e:
+        _raise_google_error(context, e, api=_DRIVE_API, denied="")
+    return bool(permissions)
+
+
+class _FilePermissionsCheck(_GoogleDriveCheck):
+    """Lists files with their permissions as the primary admin, and reads the
+    permissions of files that have them only by id. Google denies the
+    permissions of some files (for example files of another organization); the
+    sync gives those files to the user who found them, so one denial is not a
+    failure."""
+
+    def __init__(self) -> None:
+        super().__init__(
+            capability=CredentialCapability.DOC_PERMISSION_SYNC,
+            check_id="google_drive_file_permissions",
+            display_name="File permissions are readable",
+            remediation=(
+                "Give the service account the drive.readonly and "
+                "drive.metadata.readonly scopes (OAuth: connect again and accept "
+                "every permission)."
+            ),
+        )
+
+    def run(self, context: CapabilityCheckContext) -> None:
+        ops: GoogleDriveSourceOperations = _gateway(context)
+        with _google_errors(
+            context,
+            api=_DRIVE_API,
+            denied="The credential cannot list files with their permissions",
+        ):
+            auth: GoogleDriveAuth = ops.authenticate()
+            files: list[GoogleDriveFileType] = _files(
+                ops.list_files(
+                    variant=DriveCorpus.ALL_DRIVES,
+                    user_email=auth.primary_admin_email,
+                    query="trashed = false",
+                    fields=_PERMISSION_SAMPLE_FIELDS,
+                    page_size=_SAMPLE_FILES,
+                    max_num_pages=1,
+                ),
+                _SAMPLE_FILES,
+            )
+        by_id_only: list[GoogleDriveFileType] = [
+            file
+            for file in files
+            if file.get("permissionIds") and not file.get("permissions")
+        ][:_MAX_PERMISSION_READS]
+        for file in by_id_only:
+            if _read_permissions(context, ops, auth.primary_admin_email, file):
+                return
+        if any(file.get("permissions") for file in files):
+            return
+        if by_id_only:
+            raise UnexpectedValidationError(
+                "Google Drive denied the permissions of every sampled file, so "
+                "Onyx cannot verify the permission sync. The sync gives such "
+                "files only to the user who found them."
+            )
+        if files:
+            raise InsufficientPermissionsError(
+                "Google Drive listed files without their permissions, so Onyx "
+                "cannot tell who can open them. Give the credential the "
+                "drive.metadata.readonly scope."
+            )
+
+
+class _DirectoryAdminCheck(_GoogleDriveCheck):
+    """Reads the primary admin's directory entry, as group sync does first."""
+
+    def __init__(self) -> None:
+        super().__init__(
+            capability=CredentialCapability.EXTERNAL_GROUP_SYNC,
+            check_id="google_drive_directory_admin",
+            display_name="Primary admin can use the Workspace directory",
+            remediation=_DIRECTORY_HINT,
+        )
+
+    def run(self, context: CapabilityCheckContext) -> None:
+        ops: GoogleDriveSourceOperations = _gateway(context)
+        admin: str | None = context.credential_json.get(
+            DB_CREDENTIALS_PRIMARY_ADMIN_KEY
+        )
+        with _google_errors(
+            context,
+            api=_ADMIN_API,
+            denied=(
+                f"Primary admin {admin or ''} is not authorized on the Google "
+                "Workspace directory API. Reconnect the connector with an "
+                "account that has admin directory access"
+            ),
+        ):
+            ops.authenticate()
+            ops.get_admin_user()
+
+
+class _DirectoryUsersCheck(_GoogleDriveCheck):
+    """Group sync puts every listed user in the group of the Workspace domain,
+    which "anyone at the domain" shares grant."""
+
+    def __init__(self) -> None:
+        super().__init__(
+            capability=CredentialCapability.EXTERNAL_GROUP_SYNC,
+            check_id="google_drive_directory_users",
+            display_name="Workspace users can be listed",
+            remediation=_DIRECTORY_HINT,
+        )
+
+    def run(self, context: CapabilityCheckContext) -> None:
+        ops: GoogleDriveSourceOperations = _gateway(context)
+        with _google_errors(
+            context,
+            api=_ADMIN_API,
+            denied="The primary admin cannot list the Workspace users",
+        ):
+            ops.authenticate()
+            first_user: str | None = next(ops.list_user_emails(), None)
+        if first_user is None:
+            raise UnexpectedValidationError(
+                "The Workspace directory listed no user, so Onyx cannot verify "
+                "the user listing."
+            )
+
+
+class _GroupsCheck(_GoogleDriveCheck):
+    """Lists the groups and reads the members of the first. Passes when the
+    domain has no group."""
+
+    def __init__(self) -> None:
+        super().__init__(
+            capability=CredentialCapability.EXTERNAL_GROUP_SYNC,
+            check_id="google_drive_groups",
+            display_name="Groups and their members are readable",
+            remediation=_DIRECTORY_HINT,
+        )
+
+    def run(self, context: CapabilityCheckContext) -> None:
+        ops: GoogleDriveSourceOperations = _gateway(context)
+        with _google_errors(
+            context,
+            api=_ADMIN_API,
+            denied="The primary admin cannot read the Workspace groups",
+        ):
+            ops.authenticate()
+            first_group: str | None = next(ops.list_groups(), None)
+            if first_group is None:
+                return
+            next(ops.list_group_members(group_email=first_group), None)
+
+
+class _DriveMembershipCheck(_GoogleDriveCheck):
+    """Reads the members of the first shared drive, as group sync reads every
+    drive. A primary admin who is not a Workspace admin reads only the drives
+    it is a member of."""
+
+    def __init__(self) -> None:
+        super().__init__(
+            capability=CredentialCapability.EXTERNAL_GROUP_SYNC,
+            check_id="google_drive_shared_drive_members",
+            display_name="Shared drive members are readable",
+            required=False,
+            remediation=(
+                "Make the primary admin a Workspace admin who can manage shared "
+                "drives. Without it, users get access to shared drive files only "
+                "through the files' own permissions."
+            ),
+        )
+
+    def run(self, context: CapabilityCheckContext) -> None:
+        ops: GoogleDriveSourceOperations = _gateway(context)
+        with _google_errors(
+            context,
+            api=_DRIVE_API,
+            denied="The primary admin cannot read the members of shared drives",
+        ):
+            auth: GoogleDriveAuth = ops.authenticate()
+            admin_user: GoogleDirectoryUser = ops.get_admin_user()
+            is_admin: bool = admin_user.is_admin or admin_user.is_delegated_admin
+            drive_id: str | None = next(
+                ops.list_drives(
+                    user_email=auth.primary_admin_email,
+                    use_domain_admin_access=is_admin,
+                ),
+                None,
+            )
+            if drive_id is not None:
+                next(
+                    ops.list_drive_members(
+                        drive_id=drive_id, use_domain_admin_access=is_admin
+                    ),
+                    None,
+                )
+        if not is_admin:
+            raise InsufficientPermissionsError(
+                f"{auth.primary_admin_email} is not a Workspace admin, so group "
+                "sync reads only the shared drives it is a member of."
+            )
+
+
+class _FolderPermissionsCheck(_GoogleDriveCheck):
+    """Lists folders with their permissions as the primary admin, as group
+    sync does for each user, and reads the permissions of a folder that has
+    them only by id."""
+
+    def __init__(self) -> None:
+        super().__init__(
+            capability=CredentialCapability.EXTERNAL_GROUP_SYNC,
+            check_id="google_drive_folder_permissions",
+            display_name="Folder permissions are readable",
+            required=False,
+            remediation=(
+                "Give the service account the drive.metadata.readonly scope "
+                "(OAuth: connect again and accept every permission). Without it, "
+                "users get access only through each file's own permissions."
+            ),
+        )
+
+    def run(self, context: CapabilityCheckContext) -> None:
+        ops: GoogleDriveSourceOperations = _gateway(context)
+        with _google_errors(
+            context,
+            api=_DRIVE_API,
+            denied="The credential cannot list folders with their permissions",
+        ):
+            auth: GoogleDriveAuth = ops.authenticate()
+            # The sync ends the listing quietly on a denial; the check reports it.
+            folder: GoogleDriveFileType | None = next(
+                ops.list_folders_with_permissions(
+                    user_email=auth.primary_admin_email, continue_on_404_or_403=False
+                ),
+                None,
+            )
+        if folder is None or folder.get("permissions"):
+            return
+        if folder.get("permissionIds") and not _read_permissions(
+            context, ops, auth.primary_admin_email, folder
+        ):
+            name: str = folder.get("name") or folder["id"]
+            raise InsufficientPermissionsError(
+                f"The credential cannot read who can open the folder '{name}', "
+                "so its members do not get access through it."
+            )
+
+
+def build_google_drive_doc_permission_sync_checks() -> list[CapabilityCheck]:
+    return [_FilePermissionsCheck()]
+
+
+def build_google_drive_group_sync_checks() -> list[CapabilityCheck]:
+    return [
+        _DirectoryAdminCheck(),
+        _DirectoryUsersCheck(),
+        _GroupsCheck(),
+        _DriveMembershipCheck(),
+        _FolderPermissionsCheck(),
+    ]

backend/onyx/connectors/google_drive/source_operations.py

@@ -256,9 +256,6 @@ _PERM_SYNC_CAPABILITIES = frozenset(
         CredentialCapability.EXTERNAL_GROUP_SYNC,
     }
 )
-_PERM_SYNC_CHECKS_PENDING = (
-    "Not yet tested: the permission-sync checks land in the next PR."
-)
 _DEGRADES_ON_DENIAL = (
     "Hierarchy and shortcut reads treat a denial as a missing item, so a probe "
     "cannot fail on permissions."
@@ -342,7 +339,6 @@ class GoogleDriveSourceOperations(SourceOperations):
     @source_operation(
         capabilities=_ALL_CAPABILITIES,
         consumes=OperationConsumes.CREDENTIAL,
-        untested=_PERM_SYNC_CHECKS_PENDING,
     )
     @_translated
     def authenticate(self) -> GoogleDriveAuth:
@@ -540,7 +536,6 @@ class GoogleDriveSourceOperations(SourceOperations):
             CredentialCapability.EXTERNAL_GROUP_SYNC,
         },
         consumes=OperationConsumes.CREDENTIAL,
-        untested=_PERM_SYNC_CHECKS_PENDING,
     )
     @_translated
     def list_drives(
@@ -626,7 +621,6 @@ class GoogleDriveSourceOperations(SourceOperations):
             CredentialCapability.EXTERNAL_GROUP_SYNC,
         },
         consumes=OperationConsumes.CREDENTIAL,
-        untested=_PERM_SYNC_CHECKS_PENDING,
     )
     @_translated
     def list_drive_members(
@@ -649,7 +643,6 @@ class GoogleDriveSourceOperations(SourceOperations):
     @source_operation(
         capabilities=_PERM_SYNC_CAPABILITIES,
         consumes=OperationConsumes.CREDENTIAL,
-        untested=_PERM_SYNC_CHECKS_PENDING,
     )
     @_translated
     def list_file_permissions(
@@ -745,7 +738,6 @@ class GoogleDriveSourceOperations(SourceOperations):
             CredentialCapability.EXTERNAL_GROUP_SYNC,
         },
         consumes=OperationConsumes.CREDENTIAL,
-        untested=_PERM_SYNC_CHECKS_PENDING,
     )
     @_translated
     def list_user_emails(self, *, query: str | None = None) -> Iterator[str]:
@@ -765,7 +757,6 @@ class GoogleDriveSourceOperations(SourceOperations):
     @source_operation(
         capabilities={CredentialCapability.EXTERNAL_GROUP_SYNC},
         consumes=OperationConsumes.CREDENTIAL,
-        untested=_PERM_SYNC_CHECKS_PENDING,
     )
     @_translated
     def get_admin_user(self) -> GoogleDirectoryUser:
@@ -785,7 +776,6 @@ class GoogleDriveSourceOperations(SourceOperations):
     @source_operation(
         capabilities={CredentialCapability.EXTERNAL_GROUP_SYNC},
         consumes=OperationConsumes.CREDENTIAL,
-        untested=_PERM_SYNC_CHECKS_PENDING,
     )
     @_translated
     def list_groups(self) -> Iterator[str]:
@@ -804,7 +794,6 @@ class GoogleDriveSourceOperations(SourceOperations):
             CredentialCapability.EXTERNAL_GROUP_SYNC,
         },
         consumes=OperationConsumes.CREDENTIAL,
-        untested=_PERM_SYNC_CHECKS_PENDING,
     )
     @_translated
     def list_group_members(self, *, group_email: str) -> Iterator[GoogleGroupMember]:
@@ -820,17 +809,17 @@ class GoogleDriveSourceOperations(SourceOperations):
     @source_operation(
         capabilities={CredentialCapability.EXTERNAL_GROUP_SYNC},
         consumes=OperationConsumes.CREDENTIAL,
-        untested=_PERM_SYNC_CHECKS_PENDING,
     )
     @_translated
     def list_folders_with_permissions(
-        self, *, user_email: str
+        self, *, user_email: str, continue_on_404_or_403: bool = True
     ) -> Iterator[GoogleDriveFileType]:
-        """Every folder the user can see, with its published permissions."""
+        """Every folder the user can see, with its published permissions. By
+        default a 403 or 404 ends the listing without an error."""
         yield from execute_paginated_retrieval(
             retrieval_function=self._drive(user_email).files().list,
             list_key="files",
-            continue_on_404_or_403=True,
+            continue_on_404_or_403=continue_on_404_or_403,
             corpora=DriveCorpus.ALL_DRIVES.value,
             supportsAllDrives=True,
             includeItemsFromAllDrives=True,

backend/tests/daily/connectors/google_drive/test_google_drive_capability_checks_live.py

@@ -1,5 +1,5 @@
-"""Runs the Google Drive indexing checks against the live test Workspace with
-the service account. Read only."""
+"""Runs the Google Drive indexing, permission-sync and group-sync checks
+against the live test Workspace with the service account. Read only."""
 
 from typing import Any
 
@@ -14,11 +14,14 @@ from onyx.connectors.capability_checks.models import (
 from onyx.connectors.capability_checks.runner import run_capability_checks
 from onyx.connectors.credentials_provider import OnyxStaticCredentialsProvider
 from onyx.connectors.google_drive.capability_checks import (
+    build_google_drive_doc_permission_sync_checks,
+    build_google_drive_group_sync_checks,
     build_google_drive_indexing_checks,
 )
 from onyx.connectors.google_drive.source_operations import (
     GoogleDriveSourceOperations,
 )
+from onyx.db.enums import AccessType
 from tests.daily.connectors.google_drive.conftest import build_credentials
 from tests.daily.connectors.google_drive.consts_and_utils import (
     ADMIN_EMAIL,
@@ -43,6 +46,7 @@ def _run(
         source=DocumentSource.GOOGLE_DRIVE,
         credential_json=credential_json,
         connector_specific_config=config,
+        access_type=AccessType.SYNC,
         source_operations=GoogleDriveSourceOperations(
             credentials_provider=OnyxStaticCredentialsProvider(
                 None, DocumentSource.GOOGLE_DRIVE.value, credential_json
@@ -50,7 +54,12 @@ def _run(
             connector_specific_config=config,
         ),
     )
-    results = run_capability_checks(build_google_drive_indexing_checks(), context)
+    checks = (
+        build_google_drive_indexing_checks()
+        + build_google_drive_doc_permission_sync_checks()
+        + build_google_drive_group_sync_checks()
+    )
+    results = run_capability_checks(checks, context)
     return {result.check_id: result for result in results}
 
 

backend/tests/unit/ee/onyx/connectors/test_ee_capability_checks.py

@@ -94,14 +94,14 @@ def test_fallback_synthesis_respects_applicability(
     )
 
     # Under test.
-    checks = get_perm_sync_capability_checks(DocumentSource.GOOGLE_DRIVE)
+    checks = get_perm_sync_capability_checks(DocumentSource.BOX)
 
     # Postcondition.
     assert [check.capability for check in checks] == [
         CredentialCapability.DOC_PERMISSION_SYNC
     ]
     assert checks[0].is_fallback is True
-    assert checks[0].check_id == "google_drive_perm_sync"
+    assert checks[0].check_id == "box_perm_sync"
 
 
 def test_named_checks_ignore_the_probe_allowlist(
@@ -139,7 +139,7 @@ def test_unregistered_sync_source_gets_shared_fallback_checks() -> None:
     Verifies fallback synthesis for a sync-capable source with no named checks.
     """
     # Under test.
-    checks = get_perm_sync_capability_checks(DocumentSource.GOOGLE_DRIVE)
+    checks = get_perm_sync_capability_checks(DocumentSource.BOX)
 
     # Postcondition.
     # One fallback per applicable capability, sharing a single check_id (and
@@ -150,7 +150,7 @@ def test_unregistered_sync_source_gets_shared_fallback_checks() -> None:
     }
     assert all(check.is_fallback for check in checks)
     assert len({type(check) for check in checks}) == 1
-    assert {check.check_id for check in checks} == {"google_drive_perm_sync"}
+    assert {check.check_id for check in checks} == {"box_perm_sync"}
 
 
 def test_registered_checks_clobber_only_their_capability(
@@ -165,23 +165,23 @@ def test_registered_checks_clobber_only_their_capability(
     # requires the gateway).
     monkeypatch.setattr(source_operations_module, "_SOURCE_OPERATIONS_BY_SOURCE", {})
 
-    class _GoogleDriveOperations(SourceOperations):
-        source = DocumentSource.GOOGLE_DRIVE
+    class _BoxOperations(SourceOperations):
+        source = DocumentSource.BOX
         sdk_modules = ()
 
     named_check = _NamedCheck(
         capability=CredentialCapability.DOC_PERMISSION_SYNC,
-        check_id="google_drive_named_check",
+        check_id="box_named_check",
         display_name="Named check",
     )
     monkeypatch.setitem(
         ee_capability_checks._DOC_PERMISSION_SYNC_CHECKS_BY_SOURCE,
-        DocumentSource.GOOGLE_DRIVE,
+        DocumentSource.BOX,
         [named_check],
     )
 
     # Under test.
-    checks = get_perm_sync_capability_checks(DocumentSource.GOOGLE_DRIVE)
+    checks = get_perm_sync_capability_checks(DocumentSource.BOX)
 
     # Postcondition.
     # Doc sync uses the named check; group sync, applicable but still
@@ -212,11 +212,11 @@ def test_fallback_check_calls_validate_perm_sync() -> None:
     # Precondition.
     connector = MagicMock(spec=BaseConnector)
     context = CapabilityCheckContext(
-        source=DocumentSource.GOOGLE_DRIVE,
+        source=DocumentSource.BOX,
         credential_json={},
         connector=connector,
     )
-    check = get_perm_sync_capability_checks(DocumentSource.GOOGLE_DRIVE)[0]
+    check = get_perm_sync_capability_checks(DocumentSource.BOX)[0]
 
     # Under test.
     check.run(context)

backend/tests/unit/onyx/connectors/google_drive/test_google_drive_perm_sync_checks.py

@@ -0,0 +1,223 @@
+"""The Google Drive permission-sync and group-sync checks, against a fake
+gateway."""
+
+from collections.abc import Iterator
+from typing import Any
+from unittest.mock import MagicMock
+
+import pytest
+
+from onyx.configs.constants import DocumentSource
+from onyx.connectors.capability_checks.models import (
+    CapabilityCheckContext,
+    CapabilityCheckResult,
+    CapabilityCheckStatus,
+)
+from onyx.connectors.capability_checks.runner import run_capability_checks
+from onyx.connectors.google_drive.capability_checks import (
+    build_google_drive_doc_permission_sync_checks,
+    build_google_drive_group_sync_checks,
+)
+from onyx.connectors.google_drive.models import GoogleDirectoryUser, GoogleDriveAuth
+from onyx.connectors.google_drive.source_operations import (
+    GoogleDriveHttpError,
+    GoogleDriveSourceOperations,
+)
+from onyx.connectors.google_utils.shared_constants import (
+    DB_CREDENTIALS_PRIMARY_ADMIN_KEY,
+    GoogleCredentialKind,
+)
+from onyx.db.enums import AccessType
+
+_ADMIN = "admin@co.com"
+
+
+def _http_error(status: int, *reasons: str) -> GoogleDriveHttpError:
+    return GoogleDriveHttpError(
+        status_code=status,
+        reasons=reasons,
+        access_denied=status in (403, 404),
+        message=f"HTTP {status}",
+    )
+
+
+def _listing(*items: Any) -> Any:
+    def listing(**_kwargs: Any) -> Iterator[Any]:
+        return iter(items)
+
+    return listing
+
+
+def _ops() -> MagicMock:
+    ops = MagicMock(spec=GoogleDriveSourceOperations)
+    ops.authenticate.return_value = GoogleDriveAuth(
+        kind=GoogleCredentialKind.SERVICE_ACCOUNT, primary_admin_email=_ADMIN
+    )
+    ops.get_admin_user.return_value = GoogleDirectoryUser(
+        is_admin=True, is_delegated_admin=False
+    )
+    ops.list_files.side_effect = _listing(
+        {"id": "f1", "permissions": [{"id": "p", "type": "user"}]}
+    )
+    ops.list_user_emails.side_effect = _listing(_ADMIN)
+    ops.list_groups.side_effect = _listing("team@co.com")
+    ops.list_group_members.side_effect = _listing()
+    ops.list_drives.side_effect = _listing("d1")
+    ops.list_drive_members.side_effect = _listing()
+    ops.list_folders_with_permissions.side_effect = _listing()
+    ops.list_file_permissions.side_effect = _listing({"id": "p"})
+    return ops
+
+
+def _run(check_id: str, ops: MagicMock) -> CapabilityCheckResult:
+    (check,) = [
+        check
+        for check in build_google_drive_doc_permission_sync_checks()
+        + build_google_drive_group_sync_checks()
+        if check.check_id == check_id
+    ]
+    context = CapabilityCheckContext(
+        source=DocumentSource.GOOGLE_DRIVE,
+        credential_json={DB_CREDENTIALS_PRIMARY_ADMIN_KEY: _ADMIN},
+        access_type=AccessType.SYNC,
+        source_operations=ops,
+    )
+    (result,) = run_capability_checks([check], context)
+    return result
+
+
+@pytest.mark.parametrize(
+    "check_id",
+    [
+        "google_drive_file_permissions",
+        "google_drive_directory_admin",
+        "google_drive_directory_users",
+        "google_drive_groups",
+        "google_drive_shared_drive_members",
+        "google_drive_folder_permissions",
+    ],
+)
+def test_checks_pass_for_a_workspace_admin(check_id: str) -> None:
+    assert _run(check_id, _ops()).status == CapabilityCheckStatus.PASSED
+
+
+def test_checks_apply_only_to_permission_synced_connectors() -> None:
+    check = build_google_drive_group_sync_checks()[0]
+
+    assert check.access_types is not None
+    assert AccessType.PUBLIC not in check.access_types
+
+
+def test_file_permissions_listed_by_id_are_read() -> None:
+    ops = _ops()
+    ops.list_files.side_effect = _listing({"id": "f1", "permissionIds": ["p"]})
+
+    assert _run("google_drive_file_permissions", ops).status == (
+        CapabilityCheckStatus.PASSED
+    )
+    assert ops.list_file_permissions.call_args.kwargs["file_id"] == "f1"
+
+
+def test_a_file_denied_by_id_is_skipped_for_the_next() -> None:
+    ops = _ops()
+    ops.list_files.side_effect = _listing(
+        {"id": "external", "permissionIds": ["p"]},
+        {"id": "f1", "permissionIds": ["p"]},
+    )
+
+    def permissions(*, file_id: str, **_kwargs: Any) -> Iterator[dict[str, Any]]:
+        if file_id == "external":
+            raise _http_error(403, "insufficientFilePermissions")
+        return iter([{"id": "p"}])
+
+    ops.list_file_permissions.side_effect = permissions
+
+    assert _run("google_drive_file_permissions", ops).status == (
+        CapabilityCheckStatus.PASSED
+    )
+
+
+def test_every_sampled_file_denied_is_indeterminate() -> None:
+    ops = _ops()
+    ops.list_files.side_effect = _listing({"id": "f1", "permissionIds": ["p"]})
+    ops.list_file_permissions.side_effect = _http_error(403)
+
+    result = _run("google_drive_file_permissions", ops)
+
+    assert result.status == CapabilityCheckStatus.INDETERMINATE
+
+
+def test_a_missing_scope_on_the_permission_read_fails() -> None:
+    ops = _ops()
+    ops.list_files.side_effect = _listing({"id": "f1", "permissionIds": ["p"]})
+    ops.list_file_permissions.side_effect = _http_error(
+        403, "ACCESS_TOKEN_SCOPE_INSUFFICIENT"
+    )
+
+    result = _run("google_drive_file_permissions", ops)
+
+    assert result.status == CapabilityCheckStatus.FAILED
+    assert "missing a scope" in result.message
+
+
+def test_directory_admin_keeps_the_legacy_message() -> None:
+    ops = _ops()
+    ops.get_admin_user.side_effect = _http_error(403)
+
+    result = _run("google_drive_directory_admin", ops)
+
+    assert result.status == CapabilityCheckStatus.FAILED
+    assert result.error_type == "InsufficientPermissionsError"
+    assert f"Primary admin {_ADMIN} is not authorized" in result.message
+
+
+def test_admin_sdk_turned_off_is_reported() -> None:
+    ops = _ops()
+    ops.list_groups.side_effect = _http_error(403, "accessNotConfigured")
+
+    result = _run("google_drive_groups", ops)
+
+    assert result.status == CapabilityCheckStatus.FAILED
+    assert "Admin SDK API is turned off" in result.message
+
+
+def test_non_admin_reads_only_its_own_shared_drives() -> None:
+    ops = _ops()
+    ops.get_admin_user.return_value = GoogleDirectoryUser(
+        is_admin=False, is_delegated_admin=False
+    )
+
+    result = _run("google_drive_shared_drive_members", ops)
+
+    assert result.status == CapabilityCheckStatus.FAILED
+    assert result.required is False
+    ops.list_drives.assert_called_once_with(
+        user_email=_ADMIN, use_domain_admin_access=False
+    )
+
+
+def test_folder_permissions_listed_by_id_are_read() -> None:
+    ops = _ops()
+    ops.list_folders_with_permissions.side_effect = _listing(
+        {"id": "folder", "permissionIds": ["p"]}
+    )
+    ops.list_file_permissions.side_effect = _listing()
+
+    result = _run("google_drive_folder_permissions", ops)
+
+    assert result.status == CapabilityCheckStatus.FAILED
+    assert "who can open the folder 'folder'" in result.message
+
+
+def test_a_denied_folder_listing_is_reported() -> None:
+    ops = _ops()
+    ops.list_folders_with_permissions.side_effect = _http_error(403)
+
+    result = _run("google_drive_folder_permissions", ops)
+
+    assert result.status == CapabilityCheckStatus.FAILED
+    assert "cannot list folders" in result.message
+    assert (
+        ops.list_folders_with_permissions.call_args.kwargs["continue_on_404_or_403"]
+        is False
+    )

backend/tests/unit/onyx/connectors/test_source_operation_coverage.py

@@ -54,11 +54,13 @@ def _configure_confluence_spy(spy: MagicMock) -> None:
 
 def _configure_google_drive_spy(spy: MagicMock) -> None:
     """A service account whose primary admin sees a Google Doc, a small
-    upload and one shared drive, organized through a group."""
+    upload, a shared drive file and a folder whose permissions come only by
+    id, and one shared drive, organized through a group."""
     admin = "admin@example.com"
     sample = [
         {"id": "doc", "name": "doc", "mimeType": GDriveMimeType.DOC.value},
         {"id": "pdf", "name": "pdf", "mimeType": "application/pdf", "size": "10"},
+        {"id": "shared", "name": "shared", "permissionIds": ["permission"]},
     ]
     spy.authenticate.return_value = GoogleDriveAuth(
         kind=GoogleCredentialKind.SERVICE_ACCOUNT, primary_admin_email=admin
@@ -73,6 +75,12 @@ def _configure_google_drive_spy(spy: MagicMock) -> None:
     )
     spy.list_user_emails.side_effect = lambda **_kwargs: iter([admin])
     spy.can_list_drive.return_value = True
+    spy.list_folders_with_permissions.side_effect = lambda **_kwargs: iter(
+        [{"id": "folder", "name": "folder", "permissionIds": ["permission"]}]
+    )
+    spy.list_file_permissions.side_effect = lambda **_kwargs: iter(
+        [{"id": "permission"}]
+    )
 
 
 def _configure_jira_site(spy: MagicMock) -> None: