feat(connectors): restricted sync access type (#15295)

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

backend/ee/onyx/background/celery/tasks/external_group_syncing/tasks.py

@@ -195,7 +195,7 @@ def check_for_external_group_sync(self: Task, *, tenant_id: str) -> bool | None:
                 cc_pairs_to_dedupe = get_cc_pairs_by_source(
                     db_session,
                     source,
-                    access_type=AccessType.SYNC,
+                    access_types=AccessType.perm_synced_types(),
                     status=ConnectorCredentialPairStatus.ACTIVE,
                 )
                 # dedupe cc_pairs to only keep the first one

backend/ee/onyx/db/cc_pair_data_access.py

@@ -1,9 +1,10 @@
 """Data-access groups of cc-pairs (UserGroup__CCPairDataAccess).
 
 For a PRIVATE pair, members of its data-access groups may read all its
-documents. The rows are read at query time (the cc-pair access filter) and at
-index time (the group: ACL entries). A write marks the pair's documents for
-metadata sync, so the group: entries stay current.
+documents. For a SYNC_RESTRICTED pair, they may read the documents their source
+ACL allows. The rows are read at query time (the cc-pair access filter) and,
+for PRIVATE pairs, at index time (the group: ACL entries). A write marks the
+pair's documents for metadata sync, so the group: entries stay current.
 """
 
 from collections.abc import Collection
@@ -67,6 +68,70 @@ def fetch_private_cc_pair_ids(
     )
 
 
+def fetch_cc_pair_ids_with_data_access(
+    db_session: Session, cc_pair_ids: Collection[int]
+) -> set[int]:
+    """The pairs of cc_pair_ids whose data-access groups decide who may read
+    them (PRIVATE and SYNC_RESTRICTED)."""
+    if not cc_pair_ids:
+        return set()
+    return set(
+        db_session.scalars(
+            select(ConnectorCredentialPair.id).where(
+                ConnectorCredentialPair.id.in_(cc_pair_ids),
+                ConnectorCredentialPair.access_type.in_(AccessType.data_access_types()),
+            )
+        )
+    )
+
+
+def lock_cc_pairs_for_data_access__no_commit(
+    db_session: Session, cc_pair_ids: Collection[int]
+) -> None:
+    """Row-locks the pairs in id order. A writer that can remove a data-access
+    group takes this lock before it writes, so that
+    assert_restricted_cc_pairs_keep_a_group sees the removals of concurrent
+    writers. NO KEY UPDATE does not block inserts that reference the pairs."""
+    if not cc_pair_ids:
+        return
+    db_session.execute(
+        select(ConnectorCredentialPair.id)
+        .where(ConnectorCredentialPair.id.in_(cc_pair_ids))
+        .order_by(ConnectorCredentialPair.id)
+        .with_for_update(key_share=True)
+    )
+
+
+def assert_restricted_cc_pairs_keep_a_group(
+    db_session: Session, cc_pair_ids: Collection[int]
+) -> None:
+    """A SYNC_RESTRICTED pair with no data-access group is visible to nobody,
+    so a write must not remove its last group. The caller locks the pairs with
+    lock_cc_pairs_for_data_access__no_commit before it removes groups."""
+    if not cc_pair_ids:
+        return
+    has_group = (
+        select(UserGroup__CCPairDataAccess.cc_pair_id)
+        .where(UserGroup__CCPairDataAccess.cc_pair_id == ConnectorCredentialPair.id)
+        .exists()
+    )
+    ungrouped_ids = sorted(
+        db_session.scalars(
+            select(ConnectorCredentialPair.id).where(
+                ConnectorCredentialPair.id.in_(cc_pair_ids),
+                ConnectorCredentialPair.access_type == AccessType.SYNC_RESTRICTED,
+                ~has_group,
+            )
+        )
+    )
+    if ungrouped_ids:
+        raise OnyxError(
+            OnyxErrorCode.INVALID_INPUT,
+            "A restricted connector needs at least one data-access group: "
+            f"{ungrouped_ids}",
+        )
+
+
 def add_cc_pair_data_access__no_commit(
     db_session: Session,
     *,

backend/ee/onyx/db/connector_credential_pair.py

@@ -65,11 +65,11 @@ def _delete_connector_credential_pair_user_groups_relationship__no_commit(
 def get_cc_pairs_by_source(
     db_session: Session,
     source_type: DocumentSource,
-    access_type: AccessType | None = None,
+    access_types: list[AccessType] | None = None,
     status: ConnectorCredentialPairStatus | None = None,
 ) -> list[ConnectorCredentialPair]:
     """
-    Get all cc_pairs for a given source type with optional filtering by access_type and status
+    Get all cc_pairs for a given source type with optional filtering by access_types and status
     result is sorted by cc_pair id
     """
     query = (
@@ -79,8 +79,8 @@ def get_cc_pairs_by_source(
         .order_by(ConnectorCredentialPair.id)
     )
 
-    if access_type is not None:
-        query = query.filter(ConnectorCredentialPair.access_type == access_type)
+    if access_types is not None:
+        query = query.filter(ConnectorCredentialPair.access_type.in_(access_types))
 
     if status is not None:
         query = query.filter(ConnectorCredentialPair.status == status)
@@ -95,7 +95,7 @@ def get_all_auto_sync_cc_pairs(
     return (
         db_session.query(ConnectorCredentialPair)
         .where(
-            ConnectorCredentialPair.access_type == AccessType.SYNC,
+            ConnectorCredentialPair.access_type.in_(AccessType.perm_synced_types()),
         )
         .all()
     )

backend/ee/onyx/db/hierarchy.py

@@ -8,7 +8,11 @@ from sqlalchemy.orm import Session
 from sqlalchemy.sql.elements import ColumnElement
 
 from onyx.configs.constants import DocumentSource
-from onyx.db.connector_credential_pair import build_user_cc_pair_access_filter
+from onyx.db.connector_credential_pair import (
+    build_restricted_acl_guard,
+    build_user_cc_pair_access_filter,
+    has_sync_restricted_cc_pairs,
+)
 from onyx.db.enums import (
     ConnectorCredentialPairStatus,
     HierarchyNodeType,
@@ -46,28 +50,54 @@ def _build_connector_access_filter(user_id: UUID) -> ColumnElement[bool]:
 
 
 def _build_hierarchy_access_filter(
+    db_session: Session,
     user_email: str,
     external_group_ids: list[str],
     user_id: UUID | None = None,
 ) -> ColumnElement[bool]:
     """Grant access through the node ACL or an associated connector."""
-    access_filters: list[ColumnElement[bool]] = [
-        HierarchyNode.node_type == HierarchyNodeType.SOURCE,
-        HierarchyNode.is_public.is_(True),
-    ]
+    acl_filters: list[ColumnElement[bool]] = [HierarchyNode.is_public.is_(True)]
     if user_email:
-        access_filters.append(any_(HierarchyNode.external_user_emails) == user_email)
+        acl_filters.append(any_(HierarchyNode.external_user_emails) == user_email)
     if external_group_ids:
-        access_filters.append(
+        acl_filters.append(
             HierarchyNode.external_user_group_ids.overlap(
                 cast(postgresql.array(external_group_ids), postgresql.ARRAY(String))
             )
         )
+    acl_match = or_(*acl_filters)
+    # The guard's per-row EXISTS checks can only matter once a restricted pair exists.
+    if has_sync_restricted_cc_pairs(db_session):
+        acl_match = and_(
+            acl_match, build_restricted_acl_guard(user_id, _node_has_cc_pair)
+        )
+    access_filters: list[ColumnElement[bool]] = [
+        HierarchyNode.node_type == HierarchyNodeType.SOURCE,
+        acl_match,
+    ]
     if user_id:
         access_filters.append(_build_connector_access_filter(user_id))
     return or_(*access_filters)
 
 
+def _node_has_cc_pair(clause: ColumnElement[bool]) -> ColumnElement[bool]:
+    node_cc_pair = HierarchyNodeByConnectorCredentialPair
+    return (
+        select(1)
+        .select_from(node_cc_pair)
+        .join(
+            ConnectorCredentialPair,
+            and_(
+                ConnectorCredentialPair.connector_id == node_cc_pair.connector_id,
+                ConnectorCredentialPair.credential_id == node_cc_pair.credential_id,
+            ),
+        )
+        .where(node_cc_pair.hierarchy_node_id == HierarchyNode.id, clause)
+        .correlate(HierarchyNode)
+        .exists()
+    )
+
+
 def _get_accessible_hierarchy_nodes_for_source(
     db_session: Session,
     source: DocumentSource,
@@ -81,7 +111,9 @@ def _get_accessible_hierarchy_nodes_for_source(
         HierarchyNode.node_type != HierarchyNodeType.STUB,
     )
     stmt = stmt.where(
-        _build_hierarchy_access_filter(user_email, external_group_ids, user_id)
+        _build_hierarchy_access_filter(
+            db_session, user_email, external_group_ids, user_id
+        )
     )
     stmt = stmt.order_by(HierarchyNode.display_name)
     return list(db_session.execute(stmt).scalars().all())
@@ -105,7 +137,9 @@ def _search_accessible_hierarchy_nodes(
                 [HierarchyNodeType.STUB, HierarchyNodeType.SOURCE]
             ),
             HierarchyNode.display_name.ilike(pattern, escape="\\"),
-            _build_hierarchy_access_filter(user_email, external_group_ids, user_id),
+            _build_hierarchy_access_filter(
+                db_session, user_email, external_group_ids, user_id
+            ),
         )
         .order_by(HierarchyNode.display_name)
         .limit(limit)
@@ -125,6 +159,8 @@ def _filter_accessible_hierarchy_node_ids(
     """EE version: keep only the node ids the user can access."""
     stmt = select(HierarchyNode.id).where(HierarchyNode.id.in_(node_ids))
     stmt = stmt.where(
-        _build_hierarchy_access_filter(user_email, external_group_ids, user_id)
+        _build_hierarchy_access_filter(
+            db_session, user_email, external_group_ids, user_id
+        )
     )
     return set(db_session.execute(stmt).scalars().all())

backend/ee/onyx/db/user_group.py

@@ -11,8 +11,10 @@ from sqlalchemy.orm import Session, selectinload
 from ee.onyx.db.cc_pair_data_access import (
     add_cc_pair_data_access__no_commit,
     apply_group_cc_pair_change_to_data_access__no_commit,
+    assert_restricted_cc_pairs_keep_a_group,
+    fetch_cc_pair_ids_with_data_access,
     fetch_data_access_cc_pair_ids_for_user_group,
-    fetch_private_cc_pair_ids,
+    lock_cc_pairs_for_data_access__no_commit,
     remove_cc_pair_data_access__no_commit,
 )
 from ee.onyx.server.user_group.models import (
@@ -1024,9 +1026,9 @@ def set_user_group_data_access_cc_pairs(
     user_group_id: int,
     cc_pair_ids: set[int],
 ) -> set[int]:
-    """Sets the PRIVATE pairs whose documents the group's members may read.
-    Every added pair must be one the user may edit. Current pairs the user
-    cannot edit stay. Returns the group's data-access pair ids."""
+    """Sets the PRIVATE and SYNC_RESTRICTED pairs whose documents the group's
+    members may read. Every added pair must be one the user may edit. Current
+    pairs the user cannot edit stay. Returns the group's data-access pair ids."""
     assert_manages_group(user, db_session, group_id=user_group_id)
 
     db_user_group = fetch_user_group(db_session, user_group_id)
@@ -1064,20 +1066,23 @@ def set_user_group_data_access_cc_pairs(
             OnyxErrorCode.INSUFFICIENT_PERMISSIONS,
             "You can only change data access of connectors you can edit.",
         )
-    if non_private_ids := added_cc_pair_ids - fetch_private_cc_pair_ids(
+    if invalid_ids := added_cc_pair_ids - fetch_cc_pair_ids_with_data_access(
         db_session, added_cc_pair_ids
     ):
         raise OnyxError(
             OnyxErrorCode.INVALID_INPUT,
-            f"Data access can only be set on private connectors: {sorted(non_private_ids)}",
+            "Data access can only be set on private or restricted connectors: "
+            f"{sorted(invalid_ids)}",
         )
 
+    lock_cc_pairs_for_data_access__no_commit(db_session, removed_cc_pair_ids)
     add_cc_pair_data_access__no_commit(
         db_session, cc_pair_ids=added_cc_pair_ids, user_group_ids=[user_group_id]
     )
     remove_cc_pair_data_access__no_commit(
         db_session, cc_pair_ids=removed_cc_pair_ids, user_group_ids=[user_group_id]
     )
+    assert_restricted_cc_pairs_keep_a_group(db_session, removed_cc_pair_ids)
     mark_cc_pair_documents_for_sync__no_commit(db_session, changed_cc_pair_ids)
     db_session.commit()
     return final_cc_pair_ids

backend/ee/onyx/server/documents/cc_pair.py

@@ -10,7 +10,9 @@ from ee.onyx.background.celery.tasks.external_group_syncing.tasks import (
     try_creating_external_group_sync_task,
 )
 from ee.onyx.db.cc_pair_data_access import (
+    assert_restricted_cc_pairs_keep_a_group,
     fetch_data_access_groups_for_cc_pair,
+    lock_cc_pairs_for_data_access__no_commit,
     set_cc_pair_data_access_groups__no_commit,
 )
 from ee.onyx.db.connector_manage_access import (
@@ -278,12 +280,13 @@ def set_cc_pair_data_access(
             OnyxErrorCode.INSUFFICIENT_PERMISSIONS,
             "Connection not found for current user's permissions",
         )
-    if cc_pair.access_type != AccessType.PRIVATE:
+    if cc_pair.access_type not in AccessType.data_access_types():
         raise OnyxError(
             OnyxErrorCode.INVALID_INPUT,
-            "Data-access groups can only be set on private connectors.",
+            "Data-access groups can only be set on private or restricted connectors.",
         )
 
+    lock_cc_pairs_for_data_access__no_commit(db_session, [cc_pair_id])
     visible_group_ids = get_visible_user_group_ids(user, db_session)
     set_cc_pair_data_access_groups__no_commit(
         db_session,
@@ -291,6 +294,7 @@ def set_cc_pair_data_access(
         requested_group_ids=set(request.group_ids),
         visible_group_ids=visible_group_ids,
     )
+    assert_restricted_cc_pairs_keep_a_group(db_session, [cc_pair_id])
     db_session.commit()
     return _to_cc_pair_data_access(db_session, cc_pair_id, visible_group_ids)
 

backend/ee/onyx/server/user_group/api.py

@@ -415,7 +415,8 @@ def set_user_group_data_access_cc_pairs_endpoint(
     ),
     db_session: Session = Depends(get_session),
 ) -> UserGroupDataAccessCCPairs:
-    """Sets the private connectors whose documents the group's members may read."""
+    """Sets the private and restricted connectors whose documents the group's
+    members may read."""
     try:
         cc_pair_ids = set_user_group_data_access_cc_pairs(
             db_session=db_session,

backend/onyx/access/access.py

@@ -21,7 +21,10 @@ from onyx.configs.constants import (
     FileOrigin,
 )
 from onyx.context.search.models import CCPairAccessMode
-from onyx.db.connector_credential_pair import get_cc_pair_access_sets_for_user
+from onyx.db.connector_credential_pair import (
+    get_cc_pair_access_sets_for_user,
+    has_sync_restricted_cc_pairs,
+)
 from onyx.db.document import (
     get_access_info_for_document,
     get_access_info_for_documents,
@@ -396,9 +399,27 @@ def _user_can_access_connector_file(
     user_acl = get_acl_for_user(user, db_session)
     doc_access = get_access_for_documents(document_ids, db_session)
     if get_cc_pair_access_mode(db_session) != CCPairAccessMode.ENFORCE:
-        return any(
-            not user_acl.isdisjoint(access.to_acl()) for access in doc_access.values()
+        if not has_sync_restricted_cc_pairs(db_session):
+            return any(
+                not user_acl.isdisjoint(access.to_acl())
+                for access in doc_access.values()
+            )
+        # The old ACL rule, without SYNC_RESTRICTED pairs that grant the user
+        # nothing (see the OpenSearch restricted cc-pair guard).
+        access_sets = get_cc_pair_access_sets_for_user(db_session, user)
+        granting_cc_pair_ids = (
+            access_sets.open_cc_pair_ids | access_sets.acl_cc_pair_ids
         )
+        doc_id_to_cc_pair_ids = get_cc_pair_ids_for_documents(db_session, document_ids)
+        for document_id, access in doc_access.items():
+            cc_pair_ids = set(doc_id_to_cc_pair_ids.get(document_id, []))
+            if not access_sets.hidden_restricted_cc_pair_ids.isdisjoint(
+                cc_pair_ids
+            ) and granting_cc_pair_ids.isdisjoint(cc_pair_ids):
+                continue
+            if not user_acl.isdisjoint(access.to_acl()):
+                return True
+        return False
 
     # The query-time cc-pair rule of the OpenSearch filter, applied in Python.
     access_sets = get_cc_pair_access_sets_for_user(db_session, user)

backend/onyx/context/search/models.py

@@ -151,6 +151,9 @@ class AssistantKnowledgeFilters(BaseModel):
 
 
 class CCPairAccessMode(str, Enum):
+    # The cc-pair filter is off. It is built only to hide SYNC_RESTRICTED
+    # pairs from the old ACL filter.
+    OFF = "off"
     # Results use the old ACL filter; the cc-pair filter is only compared.
     SHADOW = "shadow"
     # Results use the cc-pair filter.
@@ -168,6 +171,10 @@ class CCPairAccessFilter(BaseModel):
     # The user's user_email: and external_group: entries. No group: entries,
     # since group access comes from open_cc_pair_ids.
     user_acl: list[str]
+    # SYNC_RESTRICTED pairs that grant the user nothing. The old ACL filter
+    # cannot express the restriction, so it hides their chunks unless an open
+    # or ACL pair of the chunk grants access.
+    hidden_restricted_cc_pair_ids: list[int] = []
 
 
 class UserAccessFilters(BaseModel):

backend/onyx/context/search/preprocessing/access_filters.py

@@ -3,8 +3,15 @@ from sqlalchemy.orm import Session
 from onyx.access.access import get_acl_for_user
 from onyx.access.cc_pair_access import get_cc_pair_access_mode
 from onyx.access.utils import EXTERNAL_GROUP_ACL_PREFIX, USER_EMAIL_ACL_PREFIX
-from onyx.context.search.models import CCPairAccessFilter, UserAccessFilters
-from onyx.db.connector_credential_pair import get_cc_pair_access_sets_for_user
+from onyx.context.search.models import (
+    CCPairAccessFilter,
+    CCPairAccessMode,
+    UserAccessFilters,
+)
+from onyx.db.connector_credential_pair import (
+    get_cc_pair_access_sets_for_user,
+    has_sync_restricted_cc_pairs,
+)
 from onyx.db.models import User
 
 
@@ -21,8 +28,12 @@ def _build_cc_pair_access_filter(
 ) -> CCPairAccessFilter | None:
     mode = get_cc_pair_access_mode(db_session)
     if mode is None:
-        return None
+        if not has_sync_restricted_cc_pairs(db_session):
+            return None
+        mode = CCPairAccessMode.OFF
     access_sets = get_cc_pair_access_sets_for_user(db_session, user)
+    if mode == CCPairAccessMode.OFF and not access_sets.hidden_restricted_cc_pair_ids:
+        return None
     return CCPairAccessFilter(
         mode=mode,
         open_cc_pair_ids=sorted(access_sets.open_cc_pair_ids),
@@ -32,4 +43,5 @@ def _build_cc_pair_access_filter(
             for entry in user_acl
             if entry.startswith((USER_EMAIL_ACL_PREFIX, EXTERNAL_GROUP_ACL_PREFIX))
         ),
+        hidden_restricted_cc_pair_ids=sorted(access_sets.hidden_restricted_cc_pair_ids),
     )

backend/onyx/db/connector_credential_pair.py

@@ -1,3 +1,4 @@
+from collections.abc import Callable
 from datetime import datetime
 from enum import Enum
 from typing import TypeVarTuple
@@ -94,11 +95,67 @@ def build_user_cc_pair_access_filter(user_id: UUID) -> ColumnElement[bool]:
     )
 
 
+def build_user_acl_cc_pair_filter(user_id: UUID | None) -> ColumnElement[bool]:
+    """Pairs whose documents the user may see on a document ACL match ("ACL"
+    pairs): SYNC pairs, and SYNC_RESTRICTED pairs where the user is in a
+    data-access group of the pair. Anonymous users (None) get only SYNC pairs.
+    Does not exclude DELETING pairs."""
+    is_sync = ConnectorCredentialPair.access_type == AccessType.SYNC
+    if user_id is None:
+        return is_sync
+    group_access_clause_fn = fetch_versioned_implementation(
+        "onyx.db.connector_credential_pair", "_build_user_group_cc_pair_access_clause"
+    )
+    return or_(
+        is_sync,
+        and_(
+            ConnectorCredentialPair.access_type == AccessType.SYNC_RESTRICTED,
+            group_access_clause_fn(user_id),
+        ),
+    )
+
+
+def build_restricted_acl_guard(
+    user_id: UUID | None,
+    has_cc_pair: Callable[[ColumnElement[bool]], ColumnElement[bool]],
+) -> ColumnElement[bool]:
+    """For readers that match a document ACL without knowing which pair grants
+    it. The match must not count for an item of a SYNC_RESTRICTED pair the user
+    can't see, unless another pair of the item grants ACL access.
+
+    has_cc_pair(clause) is an EXISTS over the item's pairs that match clause."""
+    visible_acl_pair = and_(
+        ConnectorCredentialPair.status != ConnectorCredentialPairStatus.DELETING,
+        build_user_acl_cc_pair_filter(user_id),
+    )
+    hidden_restricted_pair = and_(
+        ConnectorCredentialPair.access_type == AccessType.SYNC_RESTRICTED,
+        ~visible_acl_pair,
+    )
+    return or_(~has_cc_pair(hidden_restricted_pair), has_cc_pair(visible_acl_pair))
+
+
+def has_sync_restricted_cc_pairs(db_session: Session) -> bool:
+    return bool(
+        db_session.scalar(
+            select(
+                select(ConnectorCredentialPair.id)
+                .where(
+                    ConnectorCredentialPair.access_type == AccessType.SYNC_RESTRICTED
+                )
+                .exists()
+            )
+        )
+    )
+
+
 class CCPairAccessSets(BaseModel):
     # Documents of these pairs are visible with no document ACL match.
     open_cc_pair_ids: set[int]
     # Documents of these pairs are visible if public or matching the user's ACL.
     acl_cc_pair_ids: set[int]
+    # SYNC_RESTRICTED pairs that grant the user nothing, DELETING ones included.
+    hidden_restricted_cc_pair_ids: set[int]
 
 
 def get_cc_pair_access_sets_for_user(
@@ -106,30 +163,36 @@ def get_cc_pair_access_sets_for_user(
 ) -> CCPairAccessSets:
     """The query-time access rule for a user, as sets of cc-pair ids.
 
-    Open pairs follow build_user_cc_pair_access_filter; anonymous users get
-    only PUBLIC pairs. Every user gets all perm-synced (SYNC, SYNC_RESTRICTED)
-    pairs as ACL pairs, matching the chunk ACLs written today. DELETING pairs
-    are left out of both sets."""
+    Open pairs follow build_user_cc_pair_access_filter and ACL pairs follow
+    build_user_acl_cc_pair_filter; anonymous users get only PUBLIC open pairs
+    and SYNC ACL pairs. DELETING pairs are in neither set."""
     open_clause = (
         ConnectorCredentialPair.access_type == AccessType.PUBLIC
         if user.is_anonymous
         else build_user_cc_pair_access_filter(user.id)
     )
-    is_acl_pair = ConnectorCredentialPair.access_type.in_(
-        AccessType.perm_synced_types()
-    )
+    acl_clause = build_user_acl_cc_pair_filter(None if user.is_anonymous else user.id)
+    is_live = ConnectorCredentialPair.status != ConnectorCredentialPairStatus.DELETING
+    is_restricted = ConnectorCredentialPair.access_type == AccessType.SYNC_RESTRICTED
     rows = db_session.execute(
-        select(ConnectorCredentialPair.id, is_acl_pair).where(
-            ConnectorCredentialPair.status != ConnectorCredentialPairStatus.DELETING,
-            or_(open_clause, is_acl_pair),
-        )
+        select(
+            ConnectorCredentialPair.id,
+            and_(is_live, acl_clause).label("is_acl"),
+            and_(is_live, open_clause).label("is_open"),
+        ).where(or_(and_(is_live, or_(open_clause, acl_clause)), is_restricted))
     ).tuples()
-    access_sets = CCPairAccessSets(open_cc_pair_ids=set(), acl_cc_pair_ids=set())
-    for cc_pair_id, is_acl in rows:
+    access_sets = CCPairAccessSets(
+        open_cc_pair_ids=set(),
+        acl_cc_pair_ids=set(),
+        hidden_restricted_cc_pair_ids=set(),
+    )
+    for cc_pair_id, is_acl, is_open in rows:
         if is_acl:
             access_sets.acl_cc_pair_ids.add(cc_pair_id)
-        else:
+        elif is_open:
             access_sets.open_cc_pair_ids.add(cc_pair_id)
+        else:
+            access_sets.hidden_restricted_cc_pair_ids.add(cc_pair_id)
     return access_sets
 
 
@@ -313,7 +376,7 @@ def _add_user_filters(
 
     where_clause = User__UG.user_id == user.id
     where_clause |= ConnectorCredentialPair.access_type == AccessType.PUBLIC
-    where_clause |= ConnectorCredentialPair.access_type == AccessType.SYNC
+    where_clause |= build_user_acl_cc_pair_filter(user.id)
 
     return stmt.where(where_clause)
 
@@ -917,7 +980,7 @@ def add_credential_to_connector(
     if connector is None:
         raise HTTPException(status_code=404, detail="Connector does not exist")
 
-    if access_type == AccessType.SYNC:
+    if access_type.is_perm_synced():
         fetch_ee_implementation_or_noop(
             "onyx.utils.tier",
             "require_business_tier_for_sync_access",
@@ -928,9 +991,9 @@ def add_credential_to_connector(
             "check_if_valid_sync_source",
             noop_return_value=True,
         )(connector.source):
-            raise HTTPException(
-                status_code=400,
-                detail=f"Connector of type {connector.source} does not support SYNC access type",
+            raise OnyxError(
+                OnyxErrorCode.INVALID_INPUT,
+                f"Connector of type {connector.source} does not support permission sync",
             )
 
     if credential is None:
@@ -990,6 +1053,18 @@ def add_credential_to_connector(
                 else list(manage_access)
             ),
         )
+    elif access_type == AccessType.SYNC_RESTRICTED:
+        # With no data-access group the pair would be visible to nobody.
+        if not data_access_group_ids:
+            raise OnyxError(
+                OnyxErrorCode.INVALID_INPUT,
+                "A restricted connector needs at least one data-access group.",
+            )
+        _relate_data_access_groups_to_cc_pair__no_commit(
+            db_session=db_session,
+            cc_pair_id=association.id,
+            user_group_ids=data_access_group_ids,
+        )
 
     db_session.commit()
 

backend/onyx/db/document.py

@@ -642,7 +642,7 @@ def get_accessible_documents_for_hierarchy_node_paginated(
         DbDocument.parent_hierarchy_node_id == parent_hierarchy_node_id
     )
     stmt = apply_document_access_filter(
-        stmt, user_email, external_group_ids, user_id=user_id
+        db_session, stmt, user_email, external_group_ids, user_id=user_id
     )
 
     # Apply cursor filter based on sort type and direction
@@ -951,7 +951,8 @@ def get_cc_pair_ids_for_documents(
     Uses the same DocumentByConnectorCredentialPair rows that document access is
     built from (get_access_info_for_documents, fetch_user_groups_for_documents):
     rows with has_been_indexed=False count, rows whose cc-pair is DELETING do
-    not. Documents with no such row are left out.
+    not. DELETING SYNC_RESTRICTED pairs stay, so their documents stay hidden
+    until they are deleted. Documents with no such row are left out.
     """
     stmt = (
         select(DocumentByConnectorCredentialPair.id, ConnectorCredentialPair.id)
@@ -965,7 +966,13 @@ def get_cc_pair_ids_for_documents(
             ),
         )
         .where(DocumentByConnectorCredentialPair.id.in_(document_ids))
-        .where(ConnectorCredentialPair.status != ConnectorCredentialPairStatus.DELETING)
+        .where(
+            or_(
+                ConnectorCredentialPair.status
+                != ConnectorCredentialPairStatus.DELETING,
+                ConnectorCredentialPair.access_type == AccessType.SYNC_RESTRICTED,
+            )
+        )
     )
     doc_id_to_cc_pair_ids: dict[str, list[int]] = defaultdict(list)
     for document_id, cc_pair_id in db_session.execute(stmt):

backend/onyx/db/document_access.py

@@ -4,10 +4,14 @@ from uuid import UUID
 
 from sqlalchemy import Select, String, and_, any_, cast, or_, select
 from sqlalchemy.dialects import postgresql
-from sqlalchemy.orm import Session
+from sqlalchemy.orm import Session, aliased
 from sqlalchemy.sql.elements import ColumnElement
 
-from onyx.db.connector_credential_pair import build_user_cc_pair_access_filter
+from onyx.db.connector_credential_pair import (
+    build_restricted_acl_guard,
+    build_user_cc_pair_access_filter,
+    has_sync_restricted_cc_pairs,
+)
 from onyx.db.enums import AccessType, ConnectorCredentialPairStatus
 from onyx.db.models import (
     ConnectorCredentialPair,
@@ -17,6 +21,7 @@ from onyx.db.models import (
 
 
 def apply_document_access_filter(
+    db_session: Session,
     stmt: Select,
     user_email: str | None,
     external_group_ids: list[str],
@@ -41,30 +46,55 @@ def apply_document_access_filter(
         ConnectorCredentialPair.status != ConnectorCredentialPairStatus.DELETING
     )
 
-    access_filters: list[ColumnElement[bool]] = [
-        ConnectorCredentialPair.access_type == AccessType.PUBLIC,
-        Document.is_public.is_(True),
-    ]
+    acl_filters: list[ColumnElement[bool]] = [Document.is_public.is_(True)]
     if user_email:
-        access_filters.append(any_(Document.external_user_emails) == user_email)
+        acl_filters.append(any_(Document.external_user_emails) == user_email)
     if prior_emails:
-        access_filters.append(
+        acl_filters.append(
             Document.external_user_emails.overlap(
                 cast(postgresql.array(prior_emails), postgresql.ARRAY(String))
             )
         )
     if external_group_ids:
-        access_filters.append(
+        acl_filters.append(
             Document.external_user_group_ids.overlap(
                 cast(postgresql.array(external_group_ids), postgresql.ARRAY(String))
             )
         )
+    acl_match = or_(*acl_filters)
+    # The guard's per-row EXISTS checks can only matter once a restricted pair exists.
+    if has_sync_restricted_cc_pairs(db_session):
+        acl_match = and_(
+            acl_match, build_restricted_acl_guard(user_id, _document_has_cc_pair)
+        )
+    access_filters: list[ColumnElement[bool]] = [
+        ConnectorCredentialPair.access_type == AccessType.PUBLIC,
+        acl_match,
+    ]
     if user_id:
         access_filters.append(build_user_cc_pair_access_filter(user_id))
 
     return stmt.where(or_(*access_filters))
 
 
+def _document_has_cc_pair(clause: ColumnElement[bool]) -> ColumnElement[bool]:
+    doc_cc_pair = aliased(DocumentByConnectorCredentialPair)
+    return (
+        select(1)
+        .select_from(doc_cc_pair)
+        .join(
+            ConnectorCredentialPair,
+            and_(
+                doc_cc_pair.connector_id == ConnectorCredentialPair.connector_id,
+                doc_cc_pair.credential_id == ConnectorCredentialPair.credential_id,
+            ),
+        )
+        .where(doc_cc_pair.id == Document.id, clause)
+        .correlate(Document)
+        .exists()
+    )
+
+
 def get_accessible_documents_by_ids(
     db_session: Session,
     document_ids: list[str],
@@ -78,7 +108,7 @@ def get_accessible_documents_by_ids(
 
     stmt = select(Document).where(Document.id.in_(document_ids))
     stmt = apply_document_access_filter(
-        stmt, user_email, external_group_ids, user_id=user_id
+        db_session, stmt, user_email, external_group_ids, user_id=user_id
     )
     stmt = stmt.distinct()
     return list(db_session.execute(stmt).scalars().all())

backend/onyx/db/enums.py

@@ -308,6 +308,11 @@ class AccessType(str, PyEnum):
     def perm_synced_types(cls) -> list["AccessType"]:
         return [cls.SYNC, cls.SYNC_RESTRICTED]
 
+    @classmethod
+    def data_access_types(cls) -> list["AccessType"]:
+        """Types whose data-access groups decide who may read the documents."""
+        return [cls.PRIVATE, cls.SYNC_RESTRICTED]
+
 
 class EmbeddingPrecision(str, PyEnum):
     """Unused. Kept only because old Alembic migrations import it."""

backend/onyx/db/tag.py

@@ -182,6 +182,7 @@ def find_tags(
         .correlate(Tag)
     )
     accessible_documents = apply_document_access_filter(
+        db_session,
         accessible_documents,
         user_email,
         external_group_ids,

backend/onyx/document_index/opensearch/search.py

@@ -238,6 +238,7 @@ class DocumentQuery:
             include_hidden=include_hidden,
             access_control_list=index_filters.access_control_list,
             cc_pair_access=_get_enforced_cc_pair_access(index_filters),
+            restricted_cc_pair_guard=index_filters.cc_pair_access,
             source_types=index_filters.source_type or [],
             tags=index_filters.tags or [],
             document_sets=index_filters.document_set or [],
@@ -411,6 +412,7 @@ class DocumentQuery:
                 include_hidden=False,
                 access_control_list=index_filters.access_control_list,
                 cc_pair_access=filter_cc_pair_access,
+                restricted_cc_pair_guard=cc_pair_access,
                 source_types=index_filters.source_type or [],
                 tags=index_filters.tags or [],
                 document_sets=index_filters.document_set or [],
@@ -509,6 +511,7 @@ class DocumentQuery:
             # redundant filters in queries that may affect performance.
             access_control_list=index_filters.access_control_list,
             cc_pair_access=_get_enforced_cc_pair_access(index_filters),
+            restricted_cc_pair_guard=index_filters.cc_pair_access,
             source_types=index_filters.source_type or [],
             tags=index_filters.tags or [],
             document_sets=index_filters.document_set or [],
@@ -607,6 +610,7 @@ class DocumentQuery:
             # redundant filters in queries that may affect performance.
             access_control_list=index_filters.access_control_list,
             cc_pair_access=_get_enforced_cc_pair_access(index_filters),
+            restricted_cc_pair_guard=index_filters.cc_pair_access,
             source_types=index_filters.source_type or [],
             tags=index_filters.tags or [],
             document_sets=index_filters.document_set or [],
@@ -695,6 +699,7 @@ class DocumentQuery:
             # redundant filters in queries that may affect performance.
             access_control_list=index_filters.access_control_list,
             cc_pair_access=_get_enforced_cc_pair_access(index_filters),
+            restricted_cc_pair_guard=index_filters.cc_pair_access,
             source_types=index_filters.source_type or [],
             tags=index_filters.tags or [],
             document_sets=index_filters.document_set or [],
@@ -760,6 +765,7 @@ class DocumentQuery:
             include_hidden=False,
             access_control_list=index_filters.access_control_list,
             cc_pair_access=_get_enforced_cc_pair_access(index_filters),
+            restricted_cc_pair_guard=index_filters.cc_pair_access,
             source_types=index_filters.source_type or [],
             tags=index_filters.tags or [],
             document_sets=index_filters.document_set or [],
@@ -1034,6 +1040,7 @@ class DocumentQuery:
         hierarchy_node_ids: list[int] | None = None,
         # Operator-forced document-set scope (NAMES), applied as a standalone AND clause.
         forced_document_sets: list[str] | None = None,
+        restricted_cc_pair_guard: CCPairAccessFilter | None = None,
     ) -> list[dict[str, Any]]:
         """Returns filters to be passed into the "filter" key of a search query.
 
@@ -1060,6 +1067,9 @@ class DocumentQuery:
             cc_pair_access: If not None (and access_control_list is not None),
                 replaces the access control list filter above with the
                 cc-pair access filter.
+            restricted_cc_pair_guard: Read only when the access control list
+                filter is used. Its hidden SYNC_RESTRICTED pairs are removed
+                from that filter; see _get_restricted_cc_pair_guard.
             source_types: If supplied, only documents of one of these source
                 types will be retrieved.
             tags: If supplied, only documents with an entry in their metadata
@@ -1219,6 +1229,49 @@ class DocumentQuery:
                 )
             return {"bool": {"should": visibility_clauses, "minimum_should_match": 1}}
 
+        def _get_restricted_cc_pair_guard(
+            guard: CCPairAccessFilter,
+        ) -> dict[str, Any]:
+            """The access control list filter cannot require a data-access
+            group, so it would show a SYNC_RESTRICTED pair's chunks to anyone
+            matching the source ACL. This clause hides chunks of the pairs
+            that grant the user nothing, unless an open or ACL pair of the
+            chunk grants access.
+
+            Raises:
+                ValueError: A term list is longer than
+                    MAX_NUM_TERMS_ALLOWED_IN_TERMS_QUERY.
+            """
+            granting_cc_pair_ids = sorted(
+                {*guard.open_cc_pair_ids, *guard.acl_cc_pair_ids}
+            )
+            for name, terms in (
+                ("hidden restricted cc-pair ids", guard.hidden_restricted_cc_pair_ids),
+                ("granting cc-pair ids", granting_cc_pair_ids),
+            ):
+                if len(terms) > MAX_NUM_TERMS_ALLOWED_IN_TERMS_QUERY:
+                    raise ValueError(
+                        f"Too many {name}: {len(terms)}. Max allowed: {MAX_NUM_TERMS_ALLOWED_IN_TERMS_QUERY}."
+                    )
+            guard_clauses: list[dict[str, Any]] = [
+                {
+                    "bool": {
+                        "must_not": [
+                            {
+                                "terms": {
+                                    CC_PAIR_IDS_FIELD_NAME: guard.hidden_restricted_cc_pair_ids
+                                }
+                            }
+                        ]
+                    }
+                }
+            ]
+            if granting_cc_pair_ids:
+                guard_clauses.append(
+                    {"terms": {CC_PAIR_IDS_FIELD_NAME: granting_cc_pair_ids}}
+                )
+            return {"bool": {"should": guard_clauses, "minimum_should_match": 1}}
+
         def _get_source_type_filter(
             source_types: list[DocumentSource],
         ) -> TermsQuery[str]:
@@ -1493,13 +1546,21 @@ class DocumentQuery:
             # one acl provided here is present in the document's acl list. If
             # there is explicitly no list provided, we make no restrictions on
             # the documents that can be retrieved.
-            filter_clauses.append(
-                _get_acl_visibility_filter(access_control_list)
-                if cc_pair_access is None
-                else _get_cc_pair_access_visibility_filter(
-                    cc_pair_access, access_control_list
+            if cc_pair_access is not None:
+                filter_clauses.append(
+                    _get_cc_pair_access_visibility_filter(
+                        cc_pair_access, access_control_list
+                    )
                 )
-            )
+            else:
+                filter_clauses.append(_get_acl_visibility_filter(access_control_list))
+                if (
+                    restricted_cc_pair_guard is not None
+                    and restricted_cc_pair_guard.hidden_restricted_cc_pair_ids
+                ):
+                    filter_clauses.append(
+                        _get_restricted_cc_pair_guard(restricted_cc_pair_guard)
+                    )
 
         if forced_document_sets:
             # Its own top-level AND clause (not merged into the OR-based

backend/onyx/server/documents/cc_pair.py

@@ -812,6 +812,15 @@ def get_cc_pair_indexing_errors(
     )
 
 
+def _assert_sync_restricted_allowed() -> None:
+    if not get_security_settings().allow_connector_group_restrictions:
+        raise OnyxError(
+            OnyxErrorCode.FEATURE_NOT_AVAILABLE,
+            "Group restrictions on permission-synced connectors are turned off "
+            "for this workspace.",
+        )
+
+
 @router.put(
     "/connector/{connector_id}/credential/{credential_id}", tags=PUBLIC_API_TAGS
 )
@@ -832,27 +841,19 @@ def associate_credential_to_connector(
     """
 
     if metadata.access_type == AccessType.SYNC_RESTRICTED:
-        # Becomes creatable in the same change that enforces its data-access
-        # groups at query time, so no restricted pair exists without them.
-        # TODO(evan, ENG-4342): remove this rejection in the enforcement change,
-        # together with:
-        # - the allowed-connector query filter and the /chat/file check
-        # - the creation path: restriction_group_ids, validation, persistence
-        #   (branch jtahara/connector-group-restrictions-creation-path)
-        # - SYNC-only checks in connector_credential_pair.py: listing
-        #   visibility, tier/source validation, get_all_auto_sync_cc_pairs,
-        #   get_cc_pairs_by_source
-        # - creating the pair and its data-access rows in one transaction
-        raise OnyxError(
-            OnyxErrorCode.FEATURE_NOT_AVAILABLE,
-            "Restricted perm-synced connectors are not available yet.",
-        )
-
-    if metadata.data_access:
-        if metadata.access_type != AccessType.PRIVATE:
+        _assert_sync_restricted_allowed()
+        if not metadata.data_access:
             raise OnyxError(
                 OnyxErrorCode.INVALID_INPUT,
-                "Data-access groups can only be set on private connectors.",
+                "A restricted connector needs at least one data-access group.",
+            )
+
+    if metadata.data_access:
+        if metadata.access_type not in AccessType.data_access_types():
+            raise OnyxError(
+                OnyxErrorCode.INVALID_INPUT,
+                "Data-access groups can only be set on private or restricted "
+                "connectors.",
             )
         visible_group_ids = get_visible_user_group_ids(user, db_session)
         if visible_group_ids is not None and not visible_group_ids.issuperset(

backend/tests/external_dependency_unit/document_index/test_cc_pair_access_filter.py

@@ -36,6 +36,7 @@ from onyx.db.document import (
     get_cc_pair_ids_for_documents,
     upsert_document_by_connector_credential_pair,
 )
+from onyx.db.document_access import get_accessible_documents_by_ids
 from onyx.db.enums import (
     AccessType,
     ConnectorCredentialPairStatus,
@@ -385,6 +386,18 @@ def _visible_docs(
     user: User,
     mode: CCPairAccessMode | None,
     docs: _Docs,
+) -> set[str]:
+    return _visible_doc_ids(
+        db_session, index, user, mode, set(docs.model_dump().values())
+    )
+
+
+def _visible_doc_ids(
+    db_session: Session,
+    index: OpenSearchDocumentIndex,
+    user: User,
+    mode: CCPairAccessMode | None,
+    doc_ids: set[str],
 ) -> set[str]:
     with patch(f"{_ACCESS_FILTERS_MODULE}.get_cc_pair_access_mode", return_value=mode):
         access_filters = build_access_filters_for_user(user, db_session)
@@ -396,7 +409,7 @@ def _visible_docs(
         ),
         num_to_retrieve=100,
     )
-    return {chunk.document_id for chunk in chunks} & set(docs.model_dump().values())
+    return {chunk.document_id for chunk in chunks} & doc_ids
 
 
 @pytest.mark.usefixtures("ee")
@@ -618,3 +631,185 @@ def test_enforce_gate_waits_for_pending_cc_pairs(db_session: Session) -> None:
     # Every cc-pair done: the enforce flag decides, and can be turned back off.
     assert _mode([], enforce=True) == CCPairAccessMode.ENFORCE
     assert _mode([], enforce=False) == CCPairAccessMode.SHADOW
+
+
+# --- SYNC_RESTRICTED --------------------------------------------------------
+
+
+class _Restricted(BaseModel):
+    model_config = ConfigDict(arbitrary_types_allowed=True)
+
+    acl_member: User  # in the data-access group and the source ACL
+    member: User  # in the data-access group only
+    acl_outsider: User  # in the source ACL only
+    admin: User
+    group: UserGroup
+    pair: ConnectorCredentialPair
+    groupless_pair: ConnectorCredentialPair
+    doc_ids: dict[str, str]
+
+
+@pytest.fixture
+def restricted(
+    db_session: Session,
+    world: _World,
+    opensearch_index: OpenSearchDocumentIndex,
+    test_index_name: str,
+) -> Generator[_Restricted, None, None]:
+    acl_member = create_test_user(db_session, "cc_restricted_acl_member")
+    member = create_test_user(db_session, "cc_restricted_member")
+    acl_outsider = create_test_user(db_session, "cc_restricted_acl_outsider")
+    admin = create_test_user(db_session, "cc_restricted_admin", is_admin=True)
+    group = create_test_user_group(db_session, members=[acl_member, member])
+    pair = _make_pair(db_session, AccessType.SYNC_RESTRICTED)
+    groupless_pair = _make_pair(db_session, AccessType.SYNC_RESTRICTED)
+    db_session.add(
+        UserGroup__CCPairDataAccess(user_group_id=group.id, cc_pair_id=pair.id)
+    )
+    db_session.commit()
+
+    suffix = uuid4().hex[:8]
+    doc_ids = {
+        name: f"acc-restricted-{name}-{suffix}"
+        for name in ("acl", "public", "groupless", "with_sync", "with_private")
+    }
+    _add_document(
+        db_session,
+        [pair],
+        doc_ids["acl"],
+        external_user_emails=[acl_member.email, acl_outsider.email],
+    )
+    _add_document(db_session, [pair], doc_ids["public"], is_public=True)
+    _add_document(
+        db_session,
+        [groupless_pair],
+        doc_ids["groupless"],
+        is_public=True,
+        external_user_emails=[acl_member.email],
+    )
+    _add_document(
+        db_session,
+        [pair, world.sync_pair],
+        doc_ids["with_sync"],
+        external_user_emails=[acl_outsider.email],
+    )
+    _add_document(
+        db_session,
+        [pair, world.private_pair],
+        doc_ids["with_private"],
+        external_user_emails=[acl_outsider.email],
+    )
+    for doc_id in doc_ids.values():
+        _write_chunk(db_session, opensearch_index, doc_id)
+    OpenSearchIndexClient(index_name=test_index_name).refresh_index()
+
+    try:
+        yield _Restricted(
+            acl_member=acl_member,
+            member=member,
+            acl_outsider=acl_outsider,
+            admin=admin,
+            group=group,
+            pair=pair,
+            groupless_pair=groupless_pair,
+            doc_ids=doc_ids,
+        )
+    finally:
+        db_session.rollback()
+        for doc_id in doc_ids.values():
+            opensearch_index.delete(doc_id)
+        for restricted_pair in (pair, groupless_pair):
+            cleanup_cc_pair(db_session, restricted_pair)
+        delete_test_user(db_session, acl_member, member, acl_outsider, admin)
+        db_session.execute(delete(UserGroup).where(UserGroup.id == group.id))
+        db_session.commit()
+
+
+@pytest.mark.usefixtures("ee")
+def test_restricted_pair_needs_data_access_group_and_acl_match(
+    db_session: Session,
+    world: _World,
+    restricted: _Restricted,
+    opensearch_index: OpenSearchDocumentIndex,
+) -> None:
+    doc_ids = restricted.doc_ids
+    # A pair grants only "data-access group AND (public OR ACL match)"; a pair
+    # with no group grants nobody; a shared document gets the union of its pairs.
+    expected: list[tuple[User, set[str]]] = [
+        (restricted.acl_member, {doc_ids["acl"], doc_ids["public"]}),
+        (restricted.member, {doc_ids["public"]}),
+        (restricted.acl_outsider, {doc_ids["with_sync"]}),
+        (world.member, {doc_ids["with_private"]}),
+        (restricted.admin, set()),
+        (world.outsider, set()),
+        (world.anonymous, set()),
+    ]
+    for user, expected_doc_ids in expected:
+        # The old ACL filter hides the pairs that grant the user nothing, so
+        # it gives the same result while enforcement is off.
+        for mode in (CCPairAccessMode.ENFORCE, None):
+            visible = _visible_doc_ids(
+                db_session, opensearch_index, user, mode, set(doc_ids.values())
+            )
+            assert visible == expected_doc_ids, (user.email, mode)
+
+        if user.is_anonymous:
+            continue
+        # The Postgres form of the rule (assistant attach, tags, hierarchy).
+        accessible = get_accessible_documents_by_ids(
+            db_session,
+            list(doc_ids.values()),
+            user_email=user.email,
+            external_group_ids=[],
+            user_id=user.id,
+        )
+        assert {doc.id for doc in accessible} == expected_doc_ids, user.email
+
+
+@pytest.mark.usefixtures("ee")
+def test_restricted_pair_access_sets_and_chat_file(
+    db_session: Session, restricted: _Restricted
+) -> None:
+    member_sets = get_cc_pair_access_sets_for_user(db_session, restricted.member)
+    assert restricted.pair.id in member_sets.acl_cc_pair_ids
+    assert restricted.groupless_pair.id in member_sets.hidden_restricted_cc_pair_ids
+    admin_sets = get_cc_pair_access_sets_for_user(db_session, restricted.admin)
+    assert {restricted.pair.id, restricted.groupless_pair.id} <= (
+        admin_sets.hidden_restricted_cc_pair_ids
+    )
+
+    file_id = f"file-{restricted.doc_ids['acl']}"
+    doc = db_session.get(DbDocument, restricted.doc_ids["acl"])
+    assert doc is not None
+    doc.file_id = file_id
+    db_session.commit()
+    for mode in (None, CCPairAccessMode.ENFORCE):
+        with patch("onyx.access.access.get_cc_pair_access_mode", return_value=mode):
+            assert user_can_access_chat_file(file_id, restricted.acl_member, db_session)
+            assert not user_can_access_chat_file(
+                file_id, restricted.acl_outsider, db_session
+            )
+
+
+@pytest.mark.usefixtures("ce")
+def test_restricted_pair_grants_nothing_in_ce(
+    db_session: Session, restricted: _Restricted
+) -> None:
+    access_sets = get_cc_pair_access_sets_for_user(db_session, restricted.acl_member)
+    assert restricted.pair.id not in access_sets.acl_cc_pair_ids
+    assert restricted.pair.id in access_sets.hidden_restricted_cc_pair_ids
+
+
+@pytest.mark.usefixtures("ee")
+def test_deleting_restricted_pair_stays_on_chunks(
+    db_session: Session, restricted: _Restricted
+) -> None:
+    restricted.pair.status = ConnectorCredentialPairStatus.DELETING
+    db_session.commit()
+    doc_id = restricted.doc_ids["acl"]
+    # Without the pair id, a rewritten chunk would fall back to the old ACL filter.
+    assert get_cc_pair_ids_for_documents(db_session, [doc_id]) == {
+        doc_id: [restricted.pair.id]
+    }
+    access_sets = get_cc_pair_access_sets_for_user(db_session, restricted.acl_member)
+    assert restricted.pair.id in access_sets.hidden_restricted_cc_pair_ids

backend/tests/external_dependency_unit/server/documents/test_restricted_data_access_lock.py

@@ -0,0 +1,96 @@
+"""Concurrent data-access writes cannot remove the last group of a restricted pair."""
+
+import contextvars
+import threading
+import time
+from uuid import uuid4
+
+import pytest
+from sqlalchemy.orm import Session
+
+from ee.onyx.db.cc_pair_data_access import (
+    assert_restricted_cc_pairs_keep_a_group,
+    fetch_data_access_groups_for_cc_pair,
+    lock_cc_pairs_for_data_access__no_commit,
+    remove_cc_pair_data_access__no_commit,
+)
+from ee.onyx.server.documents.cc_pair import set_cc_pair_data_access
+from ee.onyx.server.documents.models import CCPairDataAccessUpdateRequest
+from onyx.db.engine.sql_engine import get_session_with_current_tenant
+from onyx.db.enums import AccessType
+from onyx.db.models import UserGroup, UserGroup__CCPairDataAccess
+from tests.external_dependency_unit.conftest import create_test_user
+from tests.external_dependency_unit.indexing_helpers import (
+    cleanup_cc_pair,
+    make_cc_pair,
+)
+
+pytestmark = pytest.mark.usefixtures("tenant_context")
+
+
+def _group(db_session: Session) -> UserGroup:
+    group = UserGroup(name=f"rda-{uuid4().hex[:12]}")
+    db_session.add(group)
+    db_session.flush()
+    return group
+
+
+def test_concurrent_removals_keep_one_group_on_a_restricted_pair(
+    db_session: Session,
+) -> None:
+    """One writer removes the first group while a PUT sets only the first
+    group. Without the lock, the PUT would remove the second group and both
+    would commit."""
+    admin = create_test_user(db_session, "rda-admin", is_admin=True)
+    first = _group(db_session)
+    second = _group(db_session)
+    cc_pair = make_cc_pair(db_session)
+    cc_pair.access_type = AccessType.SYNC_RESTRICTED
+    for group in (first, second):
+        db_session.add(
+            UserGroup__CCPairDataAccess(cc_pair_id=cc_pair.id, user_group_id=group.id)
+        )
+    db_session.commit()
+
+    errors: list[Exception] = []
+
+    def _remove_second() -> None:
+        with get_session_with_current_tenant() as other_session:
+            try:
+                set_cc_pair_data_access(
+                    cc_pair.id,
+                    CCPairDataAccessUpdateRequest(group_ids=[first.id]),
+                    admin,
+                    other_session,
+                )
+            except Exception as e:
+                errors.append(e)
+
+    try:
+        with get_session_with_current_tenant() as holder:
+            lock_cc_pairs_for_data_access__no_commit(holder, [cc_pair.id])
+            remove_cc_pair_data_access__no_commit(
+                holder, cc_pair_ids=[cc_pair.id], user_group_ids=[first.id]
+            )
+            assert_restricted_cc_pairs_keep_a_group(holder, [cc_pair.id])
+
+            context = contextvars.copy_context()
+            other = threading.Thread(target=context.run, args=(_remove_second,))
+            other.start()
+            time.sleep(1)
+            # The other PUT waits for this transaction.
+            assert other.is_alive()
+            holder.commit()
+        other.join(timeout=30)
+        assert not other.is_alive()
+
+        # The other PUT read the groups after this commit, so it kept one.
+        assert errors == []
+        db_session.expire_all()
+        assert [
+            group.id
+            for group in fetch_data_access_groups_for_cc_pair(db_session, cc_pair.id)
+        ] == [first.id]
+    finally:
+        db_session.rollback()
+        cleanup_cc_pair(db_session, cc_pair)

backend/tests/integration/tests/permissions_resources/test_cc_pair_data_access.py

@@ -7,11 +7,20 @@ because these tests cannot turn on the enforce flag.
 """
 
 import os
+from collections.abc import Generator
 
 import httpx
 import pytest
 
+from onyx.configs.constants import DocumentSource
+from onyx.connectors.models import InputType
 from onyx.db.enums import AccessType, ConnectorManageRole
+from tests.integration.common_utils.constants import (
+    API_SERVER_URL,
+    MOCK_CONNECTOR_SERVER_HOST,
+    MOCK_CONNECTOR_SERVER_PORT,
+)
+from tests.integration.common_utils.http_client import client
 from tests.integration.common_utils.managers.cc_pair import CCPairManager
 from tests.integration.common_utils.managers.user_group import UserGroupManager
 from tests.integration.common_utils.test_models import (
@@ -28,6 +37,7 @@ pytestmark = pytest.mark.skipif(
 _FORBIDDEN = 403
 _BAD_REQUEST = 400
 _MISSING_GROUP_ID = 999_999_999
+_FEATURE_NOT_AVAILABLE = 402
 
 
 def _private_pair(
@@ -226,3 +236,105 @@ def test_legacy_group_patch_writes_manage_and_data_access(
     group.cc_pair_ids = []
     UserGroupManager.edit(group, user_performing_action=admin)
     assert CCPairManager.get_data_access_group_ids(pair.id, admin) == set()
+
+
+# --- SYNC_RESTRICTED --------------------------------------------------------
+
+
+def _set_group_restrictions(enabled: bool | None, admin: DATestUser) -> None:
+    client.put(
+        f"{API_SERVER_URL}/admin/security",
+        json={"allow_connector_group_restrictions": enabled},
+        headers=admin.headers,
+    ).raise_for_status()
+
+
+@pytest.fixture
+def group_restrictions_on(
+    permission_admin_user: DATestUser,
+) -> Generator[None, None, None]:
+    _set_group_restrictions(True, permission_admin_user)
+    yield
+    _set_group_restrictions(None, permission_admin_user)
+
+
+def _restricted_pair(admin: DATestUser, data_access: list[int]) -> DATestCCPair:
+    return CCPairManager.create_from_scratch(
+        source=DocumentSource.MOCK_CONNECTOR,
+        input_type=InputType.POLL,
+        connector_specific_config={
+            "mock_server_host": MOCK_CONNECTOR_SERVER_HOST,
+            "mock_server_port": MOCK_CONNECTOR_SERVER_PORT,
+        },
+        access_type=AccessType.SYNC_RESTRICTED,
+        data_access=data_access,
+        user_performing_action=admin,
+    )
+
+
+def test_restricted_create_needs_workspace_toggle_and_a_group(
+    permission_admin_user: DATestUser,
+    scoped_managed_group: DATestUserGroup,
+) -> None:
+    admin = permission_admin_user
+    with pytest.raises(httpx.HTTPStatusError) as error:
+        _restricted_pair(admin, data_access=[scoped_managed_group.id])
+    assert error.value.response.status_code == _FEATURE_NOT_AVAILABLE
+
+    _set_group_restrictions(True, admin)
+    try:
+        # With no data-access group the connector would be visible to nobody.
+        with pytest.raises(httpx.HTTPStatusError) as error:
+            _restricted_pair(admin, data_access=[])
+        assert error.value.response.status_code == _BAD_REQUEST
+
+        pair = _restricted_pair(admin, data_access=[scoped_managed_group.id])
+        assert CCPairManager.get_data_access_group_ids(pair.id, admin) == {
+            scoped_managed_group.id
+        }
+    finally:
+        _set_group_restrictions(None, admin)
+
+
+@pytest.mark.usefixtures("group_restrictions_on")
+def test_restricted_data_access_keeps_a_group(
+    permission_admin_user: DATestUser,
+    scoped_managed_group: DATestUserGroup,
+    scoped_other_group: DATestUserGroup,
+) -> None:
+    admin = permission_admin_user
+    # A syncing group refuses group-side writes.
+    UserGroupManager.wait_for_sync(
+        user_performing_action=admin,
+        user_groups_to_check=[scoped_managed_group, scoped_other_group],
+    )
+    pair = _restricted_pair(admin, data_access=[scoped_managed_group.id])
+
+    CCPairManager.set_data_access(
+        pair.id, [scoped_managed_group.id, scoped_other_group.id], admin
+    ).raise_for_status()
+    CCPairManager.set_data_access(
+        pair.id, [scoped_other_group.id], admin
+    ).raise_for_status()
+
+    response = CCPairManager.set_data_access(pair.id, [], admin)
+    assert response.status_code == _BAD_REQUEST
+    response = UserGroupManager.set_data_access_cc_pairs(scoped_other_group, [], admin)
+    assert response.status_code == _BAD_REQUEST
+    assert CCPairManager.get_data_access_group_ids(pair.id, admin) == {
+        scoped_other_group.id
+    }
+
+    # The group side can add a restricted pair too. A fresh group has no other
+    # pairs, so the PUT's full list is just this one.
+    new_group = UserGroupManager.create(admin, name="restricted-group-side")
+    UserGroupManager.wait_for_sync(
+        user_performing_action=admin, user_groups_to_check=[new_group]
+    )
+    UserGroupManager.set_data_access_cc_pairs(
+        new_group, [pair.id], admin
+    ).raise_for_status()
+    assert CCPairManager.get_data_access_group_ids(pair.id, admin) == {
+        scoped_other_group.id,
+        new_group.id,
+    }

web/src/lib/connectors/accessType.ts

@@ -1,7 +1,7 @@
 import type { AccessType } from "@/lib/types";
 
-// Perm sync plus a data-access group restriction (ENG-4342). The wire name is
-// provisional until the connector editing initiative confirms it.
+// Perm sync, narrowed to members of the connector's data-access groups. The
+// groups are sent as the create request's `data_access`.
 export const SYNC_RESTRICTED_ACCESS_TYPE =
   "sync_restricted" satisfies AccessType;
 

web/src/lib/credentials/svc.ts

@@ -92,7 +92,8 @@ export function linkCredential(
   accessType?: AccessType,
   groups?: number[],
   autoSyncOptions?: Record<string, any>,
-  processingMode?: ProcessingMode
+  processingMode?: ProcessingMode,
+  dataAccess?: number[]
 ) {
   return fetch(
     `/api/manage/connector/${connectorId}/credential/${credentialId}`,
@@ -107,6 +108,8 @@ export function linkCredential(
         groups: groups || null,
         auto_sync_options: autoSyncOptions || null,
         processing_mode: processingMode || "REGULAR",
+        // Left out, a private connector gives data access to its groups.
+        ...(dataAccess !== undefined && { data_access: dataAccess }),
       }),
     }
   );

web/src/views/admin/connectors/AddConnectorPage/AddConnector.tsx

@@ -61,7 +61,10 @@ import { escapeMarkdown, markdown } from "@opal/utils";
 import { deleteConnector } from "@/lib/connector";
 import { SvgArrowExchange } from "@opal/icons";
 import { useTranslations } from "next-intl";
-import { toWireAccess } from "@/lib/connectors/accessType";
+import {
+  SYNC_RESTRICTED_ACCESS_TYPE,
+  toWireAccess,
+} from "@/lib/connectors/accessType";
 
 export interface AdvancedConfig {
   refreshFreq: number;
@@ -391,15 +394,17 @@ export default function AddConnector({
                 currentCredential ||
                 liveGDriveCredential ||
                 liveGmailCredential;
-              // TODO(evan, ENG-4342): send wireAccess.restriction_group_ids
-              // once the backend accepts them; this call creates the cc-pair.
               const linkCredentialResponse = await linkCredential(
                 response.id,
                 credential!.id,
                 name,
                 access_type,
                 groups,
-                auto_sync_options
+                auto_sync_options,
+                undefined,
+                access_type === SYNC_RESTRICTED_ACCESS_TYPE
+                  ? wireAccess.restriction_group_ids
+                  : undefined
               );
               if (linkCredentialResponse.ok) {
                 onSuccess();