diff --git a/backend/ee/onyx/configs/app_configs.py b/backend/ee/onyx/configs/app_configs.py index 98416f08b5..2edfbe074e 100644 --- a/backend/ee/onyx/configs/app_configs.py +++ b/backend/ee/onyx/configs/app_configs.py @@ -156,6 +156,12 @@ ONEDRIVE_PERMISSION_GROUP_SYNC_FREQUENCY_S = int( ZOOM_PERMISSION_GROUP_SYNC_FREQUENCY = int( os.environ.get("ZOOM_PERMISSION_GROUP_SYNC_FREQUENCY") or 5 * 60 ) +# In seconds, default is 6 hours: a sync lists every host's whole recording +# history, about 170 calls per host, and reads the share settings of every +# recording, one call each. +ZOOM_PERMISSION_DOC_SYNC_FREQUENCY = int( + os.environ.get("ZOOM_PERMISSION_DOC_SYNC_FREQUENCY") or 6 * 60 * 60 +) #### diff --git a/backend/ee/onyx/external_permissions/sync_params.py b/backend/ee/onyx/external_permissions/sync_params.py index eabdd1a407..fa8031bc60 100644 --- a/backend/ee/onyx/external_permissions/sync_params.py +++ b/backend/ee/onyx/external_permissions/sync_params.py @@ -24,6 +24,7 @@ from ee.onyx.configs.app_configs import ( SLACK_PERMISSION_DOC_SYNC_FREQUENCY, TEAMS_PERMISSION_DOC_SYNC_FREQUENCY, TEAMS_PERMISSION_GROUP_SYNC_FREQUENCY, + ZOOM_PERMISSION_DOC_SYNC_FREQUENCY, ZOOM_PERMISSION_GROUP_SYNC_FREQUENCY, ) from ee.onyx.external_permissions.perm_sync_types import ( @@ -208,6 +209,12 @@ def _load_teams_group_sync() -> GroupSyncFuncType: return teams_group_sync +def _load_zoom_doc_sync() -> DocSyncFuncType: + from ee.onyx.external_permissions.zoom.doc_sync import zoom_doc_sync + + return zoom_doc_sync + + def _load_zoom_group_sync() -> GroupSyncFuncType: from ee.onyx.external_permissions.zoom.group_sync import zoom_group_sync @@ -239,7 +246,7 @@ class SyncConfig(BaseModel): censoring_config: CensoringConfig | None = None -# No-op doc sync: these sources set permissions while indexing instead. +# No-op doc sync: the mock connector sets permissions while indexing instead. def mock_doc_sync( cc_pair: "ConnectorCredentialPair", # noqa: ARG001 fetch_all_docs_fn: FetchAllDocumentsFunction, # noqa: ARG001 @@ -348,8 +355,8 @@ _SOURCE_TO_SYNC_CONFIG: dict[DocumentSource, SyncConfig] = { # Domain sign-in rules: groups filled from the account's user roster. DocumentSource.ZOOM: SyncConfig( doc_sync_config=DocSyncConfig( - doc_sync_frequency=DEFAULT_PERMISSION_DOC_SYNC_FREQUENCY, - doc_sync_func=mock_doc_sync, + doc_sync_frequency=ZOOM_PERMISSION_DOC_SYNC_FREQUENCY, + doc_sync_func=_lazy_doc_sync(_load_zoom_doc_sync), initial_index_should_sync=True, ), group_sync_config=GroupSyncConfig( diff --git a/backend/ee/onyx/external_permissions/zoom/doc_sync.py b/backend/ee/onyx/external_permissions/zoom/doc_sync.py new file mode 100644 index 0000000000..7296695ad0 --- /dev/null +++ b/backend/ee/onyx/external_permissions/zoom/doc_sync.py @@ -0,0 +1,38 @@ +from collections.abc import Generator + +from ee.onyx.external_permissions.perm_sync_types import ( + FetchAllDocumentsFunction, + FetchAllDocumentsIdsFunction, +) +from ee.onyx.external_permissions.utils import credential_json, generic_doc_sync +from onyx.access.models import ElementExternalAccess +from onyx.configs.constants import DocumentSource +from onyx.connectors.factory import build_connector_kwargs +from onyx.connectors.zoom.connector import ZoomConnector +from onyx.db.models import ConnectorCredentialPair +from onyx.indexing.indexing_heartbeat import IndexingHeartbeatInterface + +ZOOM_DOC_SYNC_LABEL = "zoom_doc_sync" + + +def zoom_doc_sync( + cc_pair: ConnectorCredentialPair, + fetch_all_existing_docs_fn: FetchAllDocumentsFunction, # noqa: ARG001 + fetch_all_existing_docs_ids_fn: FetchAllDocumentsIdsFunction, + callback: IndexingHeartbeatInterface | None, +) -> Generator[ElementExternalAccess, None, None]: + connector = ZoomConnector( + **build_connector_kwargs( + cc_pair.connector.source, cc_pair.connector.connector_specific_config + ) + ) + connector.load_credentials(credential_json(cc_pair)) + + yield from generic_doc_sync( + cc_pair=cc_pair, + fetch_all_existing_docs_ids_fn=fetch_all_existing_docs_ids_fn, + callback=callback, + doc_source=DocumentSource.ZOOM, + slim_connector=connector, + label=ZOOM_DOC_SYNC_LABEL, + ) diff --git a/backend/ee/onyx/external_permissions/zoom/group_sync.py b/backend/ee/onyx/external_permissions/zoom/group_sync.py index ff57a90705..0958febb2e 100644 --- a/backend/ee/onyx/external_permissions/zoom/group_sync.py +++ b/backend/ee/onyx/external_permissions/zoom/group_sync.py @@ -13,6 +13,7 @@ from fnmatch import fnmatchcase from ee.onyx.db.external_perm import ExternalUserGroup from ee.onyx.external_permissions.utils import credential_json from onyx.access.utils import build_domain_group_id +from onyx.connectors.factory import build_connector_kwargs from onyx.connectors.models import ConnectorMissingCredentialError from onyx.connectors.zoom.client import MAX_LISTING_PAGES, ZoomClient from onyx.connectors.zoom.connector import ZoomConnector @@ -28,7 +29,11 @@ def zoom_group_sync( tenant_id: str, # noqa: ARG001 cc_pair: ConnectorCredentialPair, ) -> Generator[ExternalUserGroup, None, None]: - connector = ZoomConnector(**cc_pair.connector.connector_specific_config) + connector = ZoomConnector( + **build_connector_kwargs( + cc_pair.connector.source, cc_pair.connector.connector_specific_config + ) + ) connector.load_credentials(credential_json(cc_pair)) if connector.client is None: raise ConnectorMissingCredentialError("Zoom") diff --git a/backend/onyx/connectors/zoom/connector.py b/backend/onyx/connectors/zoom/connector.py index 1385f25920..76bdce3b91 100644 --- a/backend/onyx/connectors/zoom/connector.py +++ b/backend/onyx/connectors/zoom/connector.py @@ -9,6 +9,7 @@ its document id, with no discovery and no checkpoint. """ import copy +import logging from collections.abc import Generator from functools import partial from typing import Any @@ -24,6 +25,7 @@ from onyx.connectors.interfaces import ( Resolver, SecondsSinceUnixEpoch, SlimConnector, + SlimConnectorWithPermSync, ) from onyx.connectors.models import ( ConnectorCheckpoint, @@ -59,7 +61,10 @@ from onyx.connectors.zoom.recordings.processing import ( process_occurrence, ) from onyx.connectors.zoom.recordings.recording_access import ( + AccessResolver, RuleGrant, + ZoomAccessListUnavailable, + ZoomRecordingGone, load_rule_grants, look_up_owner_email, resolve_recording_access, @@ -165,7 +170,10 @@ class ZoomConnectorCheckpoint(ConnectorCheckpoint): class ZoomConnector( - CheckpointedConnectorWithPermSync[ZoomConnectorCheckpoint], SlimConnector, Resolver + CheckpointedConnectorWithPermSync[ZoomConnectorCheckpoint], + SlimConnector, + SlimConnectorWithPermSync, + Resolver, ): def __init__( self, @@ -199,6 +207,9 @@ class ZoomConnector( # permission-sync probe asks about the same things instead of sampling # again. None means it has not run. self._probe_sample: ProbeSample | None = None + # Recordings the running doc sync made private, by reason. + self._unnamed = 0 + self._gone = 0 def load_credentials(self, credentials: dict[str, Any]) -> dict[str, Any] | None: account_id = credentials.get("zoom_account_id") @@ -285,7 +296,9 @@ class ZoomConnector( raise ConnectorMissingCredentialError("Zoom") host_id = recording.host_id if host_id not in self._owner_emails: - self._owner_emails[host_id] = look_up_owner_email(client, host_id) + self._owner_emails[host_id] = ( + look_up_owner_email(client, host_id) if host_id else None + ) return resolve_recording_access( client, recording, @@ -299,6 +312,28 @@ class ZoomConnector( self._rule_grants = load_rule_grants(client) return self._rule_grants.get(rule_id) + def _synced_access(self, recording: ZoomRecordingEntry) -> ExternalAccess: + """Resolve one recording's access for the periodic doc sync. + + Indexing can fail one document and carry on, but the doc sync cannot. + So when nobody can be named to read the recording, or Zoom has deleted + it since it was listed, this answers with an empty access list: nobody + may read it. Any other error is raised unchanged, which fails the + attempt. The shared sync writes each access list as it is yielded, so + the documents reached before the error keep the access this attempt + found and the rest keep what the last attempt wrote, until an attempt + gets through. The client has already retried anything transient. + """ + try: + return self._resolve_access(recording) + except ZoomAccessListUnavailable as e: + logger.warning("%s", e) + self._unnamed += 1 + except ZoomRecordingGone as e: + logger.info("%s", e) + self._gone += 1 + return ExternalAccess.empty() + def validate_checkpoint_json(self, checkpoint_json: str) -> ZoomConnectorCheckpoint: return ZoomConnectorCheckpoint.model_validate_json(checkpoint_json) @@ -329,18 +364,41 @@ class ZoomConnector( ) -> GenerateSlimDocumentOutput: """Pruning deletes every indexed document this does not list, so the poll window is ignored and the callback is not needed: the caller drives its - own heartbeat off the batches. + own heartbeat off the batches.""" + return self._slim_documents(resolve_access=None) - Not SlimConnectorWithPermSync. Zoom builds its access lists while - indexing, so pruning wants ids and nothing else. - """ + def retrieve_all_slim_docs_perm_sync( + self, + start: SecondsSinceUnixEpoch | None = None, # noqa: ARG002 + end: SecondsSinceUnixEpoch | None = None, # noqa: ARG002 + callback: IndexingHeartbeatInterface | None = None, # noqa: ARG002 + ) -> GenerateSlimDocumentOutput: + """The doc sync makes every indexed document this does not list private, + so like pruning this ignores the poll window.""" + self._unnamed = self._gone = 0 + yield from self._slim_documents(resolve_access=self._synced_access) + if self._unnamed or self._gone: + # One line per walk, so a mass revocation is not lost among the + # per-recording lines. Deletions alone are routine. + logger.log( + logging.WARNING if self._unnamed else logging.INFO, + "Zoom permission sync made %s recording(s) private: %s that " + "nobody could be named for and %s that Zoom deleted during the walk", + self._unnamed + self._gone, + self._unnamed, + self._gone, + ) + + def _slim_documents( + self, resolve_access: AccessResolver | None + ) -> GenerateSlimDocumentOutput: if self.client is None: raise ConnectorMissingCredentialError("Zoom") # Checked again here because instantiate_connector skips # validate_connector_settings, so a connector saved with a blank form # would reach this, list nothing, and delete everything it indexed. self._raise_if_nothing_is_in_scope() - return zoom_slim_documents(self.client, self._sources) + return zoom_slim_documents(self.client, self._sources, resolve_access) def reindex( self, diff --git a/backend/onyx/connectors/zoom/recordings/discovery.py b/backend/onyx/connectors/zoom/recordings/discovery.py index 0c91578f83..69bb9f4b57 100644 --- a/backend/onyx/connectors/zoom/recordings/discovery.py +++ b/backend/onyx/connectors/zoom/recordings/discovery.py @@ -31,6 +31,7 @@ from onyx.connectors.zoom.recordings.models import ( Host, HostScope, OccurrenceWork, + ProvenRecording, ZoomListingIncomplete, ZoomSessionType, fails_the_whole_run, @@ -179,10 +180,10 @@ class DiscoveryStepResult(BaseModel): class InventoryScope(BaseModel): hosts: list[HostScope] = Field(default_factory=list) - # Occurrences Zoom answered for directly, kept whatever a host listing goes + # Recordings Zoom answered for directly, kept whatever a host listing goes # on to say later. A listing reaches neither a recording made on-premise nor # one whose owner nobody can resolve. - proven: list[OccurrenceWork] = Field(default_factory=list) + proven: list[ProvenRecording] = Field(default_factory=list) # Hosts and sessions Zoom has no record of. Tracked so the walk can tell # one that was deleted from a credential pointed at the wrong Zoom account, # which makes every one of them look deleted. @@ -339,16 +340,10 @@ class IdAllowlistSource(DiscoverySource): handler = get_session_type_handler(session_type) found = handler.find_host(client, session_id) - proven: list[OccurrenceWork] = [] + proven: list[ProvenRecording] = [] if found.anchor is not None: proven.append( - OccurrenceWork( - session_type=session_type, - session_id=session_id, - occurrence_uuid=found.anchor.uuid, - start_time=found.anchor.start_time, - topic=found.anchor.topic, - ) + ProvenRecording(session_type=session_type, recording=found.anchor) ) if found.host_id is not None and found.host_id not in listable: diff --git a/backend/onyx/connectors/zoom/recordings/inventory.py b/backend/onyx/connectors/zoom/recordings/inventory.py index 28f60738c6..425e78255e 100644 --- a/backend/onyx/connectors/zoom/recordings/inventory.py +++ b/backend/onyx/connectors/zoom/recordings/inventory.py @@ -1,14 +1,11 @@ -"""Every Zoom document that still exists, for pruning. - -A document still exists if its recording still appears in its host's recordings -listing, which leaves deleted and trashed recordings out and has no age limit. - -This is the mirror image of discovery.py. Discovery skips what it cannot read -and reports a failure, but anything skipped here is deleted instead, so this -raises rather than skipping, and it walks from Zoom's launch rather than the -poll window. The one thing it does skip is a host Zoom says it has no record -of, whose documents are then meant to go. The caller renews its lock only when -it receives a batch, so batches go out even when they are empty. +"""Every Zoom document that still exists, for pruning and the permission doc +sync: one whose recording its host's listing still shows, however old, which +leaves deleted and trashed recordings out. Anything left out here is deleted by +pruning and made private by the doc sync, so unlike discovery this raises rather +than skips, and it walks from Zoom's launch rather than the poll window. The one +skip is a host Zoom says it has no record of, whose documents are meant to go. +The caller renews its lock only when it receives a batch, so batches go out +even when they are empty. """ from collections.abc import Iterator @@ -26,8 +23,9 @@ from onyx.connectors.zoom.recordings.discovery import ( list_every_recording, listing_windows, ) -from onyx.connectors.zoom.recordings.models import Host, HostScope, OccurrenceWork +from onyx.connectors.zoom.recordings.models import Host, HostScope, ZoomSessionType from onyx.connectors.zoom.recordings.processing import zoom_document_id +from onyx.connectors.zoom.recordings.recording_access import AccessResolver from onyx.connectors.zoom.recordings.session_types import ( is_portal_upload, session_type_for_recording, @@ -37,6 +35,9 @@ from onyx.utils.logger import setup_logger logger = setup_logger() _MAX_DOCUMENTS_PER_BATCH = 500 +# Bounds the ids remembered to skip a recording listed twice, at about 30 MB. +# Past it a repeat listing costs one more access call, and nothing is lost. +_MAX_REMEMBERED_IDS = 250_000 # listing_windows runs a day past the end it is given, so 28 is the widest # trailing pass that still fits in one Zoom call per host. _TRAILING_WINDOW_DAYS = 28 @@ -46,27 +47,38 @@ _SCOPES_PER_HEARTBEAT = 25 def zoom_slim_documents( - client: ZoomClient, sources: list[DiscoverySource] + client: ZoomClient, + sources: list[DiscoverySource], + resolve_access: AccessResolver | None, ) -> GenerateSlimDocumentOutput: """Rebuilds each batch on the way out. Onyx types a slim batch as a list that may hold a HierarchyNode, and a list is invariant, so this copy is what lets everything below say what it really produces.""" - for batch in _slim_batches(client, sources): + for batch in _slim_batches(client, sources, resolve_access): yield [*batch] def _slim_batches( - client: ZoomClient, sources: list[DiscoverySource] + client: ZoomClient, + sources: list[DiscoverySource], + resolve_access: AccessResolver | None, ) -> Iterator[list[SlimDocument]]: scopes: list[HostScope] = [] proven: list[SlimDocument] = [] + emitted: set[str] = set() unrecognised = 0 anchors = 0 resolved = 0 for source in sources: for scope in source.inventory_scopes(client): scopes.extend(scope.hosts) - proven.extend(_slim_document(work) for work in scope.proven) + proven.extend( + document + for p in scope.proven + for document in _documents( + {p.session_type}, p.recording, resolve_access, emitted + ) + ) unrecognised += len(scope.unrecognised) anchors += len(scope.proven) resolved += 1 @@ -77,6 +89,12 @@ def _slim_batches( yield proven hosts = _merged(scopes) + # The same walk serves pruning and the permission sync. + job, consequence = ( + ("pruning", "delete every document it has indexed") + if resolve_access is None + else ("permission sync", "make every document it has indexed private") + ) if unrecognised and not hosts and not anchors: # Zoom answers the same not-found for a user or session that was deleted # and for one in another account, so a credential pointed at the wrong @@ -84,15 +102,16 @@ def _slim_batches( # One recording Zoom did answer for proves the account is right. raise ConnectorValidationError( "Zoom recognised none of the hosts or sessions this connector names, " - "so pruning stopped rather than delete every document it has " - "indexed. Either they were all deleted in Zoom, or the credentials " - "now point at a different account. To prune them anyway, replace " - "the entries Zoom no longer has or delete the connector" + f"so {job} stopped rather than {consequence}. Either they were all " + "deleted in Zoom, or the credentials now point at a different " + "account. To go ahead anyway, replace the entries Zoom no longer has " + "or delete the connector" ) today = datetime.now(timezone.utc).date() windows = listing_windows(EARLIEST_RECORDING_DATE, today) logger.info( - "Zoom pruning is listing %s host(s) over %s windows, about %s calls", + "Zoom %s is listing %s host(s) over %s windows, about %s calls", + job, len(hosts), len(windows), len(hosts) * (len(windows) + 1), @@ -100,7 +119,9 @@ def _slim_batches( unrecognised_types: set[str] = set() for scope in hosts: - yield from _host_documents(client, scope, windows, unrecognised_types) + yield from _host_documents( + client, scope, windows, unrecognised_types, resolve_access, emitted + ) # A recording that finishes while this walk runs lands in the newest window, # so every host is asked for that window again once the walk is over. This @@ -112,7 +133,9 @@ def _slim_batches( now = datetime.now(timezone.utc).date() trailing = listing_windows(now - timedelta(days=_TRAILING_WINDOW_DAYS), now) for scope in hosts: - yield from _host_documents(client, scope, trailing, unrecognised_types) + yield from _host_documents( + client, scope, trailing, unrecognised_types, resolve_access, emitted + ) def _merged(scopes: list[HostScope]) -> list[HostScope]: @@ -147,6 +170,8 @@ def _host_documents( scope: HostScope, windows: list[tuple[date, date]], unrecognised_types: set[str], + resolve_access: AccessResolver | None, + emitted: set[str], ) -> Iterator[list[SlimDocument]]: """Do not reach for `_UserRecordingsSource._recordings` instead. It trims to the poll window, so it would drop every meeting that ran while this walk was @@ -155,7 +180,11 @@ def _host_documents( for from_date, to_date in windows: for recording in list_every_recording(client, scope.host, from_date, to_date): - batch.extend(_slim_documents_for(recording, scope, unrecognised_types)) + batch.extend( + _slim_documents_for( + recording, scope, unrecognised_types, resolve_access, emitted + ) + ) if len(batch) >= _MAX_DOCUMENTS_PER_BATCH: yield batch batch = [] @@ -164,7 +193,11 @@ def _host_documents( def _slim_documents_for( - recording: ZoomRecordingEntry, scope: HostScope, unrecognised_types: set[str] + recording: ZoomRecordingEntry, + scope: HostScope, + unrecognised_types: set[str], + resolve_access: AccessResolver | None, + emitted: set[str], ) -> list[SlimDocument]: if recording.type is not None and is_portal_upload(recording.type): return [] @@ -186,25 +219,35 @@ def _slim_documents_for( recording.uuid, ) + document_types = scope.emitted_types(recording.session_id, derived) + return _documents(document_types, recording, resolve_access, emitted) + + +def _documents( + document_types: set[ZoomSessionType], + recording: ZoomRecordingEntry, + resolve_access: AccessResolver | None, + emitted: set[str], +) -> list[SlimDocument]: + """One document per type indexing may have written this recording under, + normally one. A recording the trailing pass or a walked host lists again + is skipped, so it costs no second access call.""" + new_ids = [ + document_id + for document_type in document_types + if (document_id := zoom_document_id(document_type, recording.uuid)) + not in emitted + ] + if not new_ids: + return [] + if len(emitted) < _MAX_REMEMBERED_IDS: + emitted.update(new_ids) + access = resolve_access(recording) if resolve_access is not None else None + created_at = _created_at(recording.start_time) return [ - _slim_document( - OccurrenceWork( - session_type=session_type, - session_id=recording.session_id, - occurrence_uuid=recording.uuid, - start_time=recording.start_time, - topic=recording.topic, - ) - ) - for session_type in scope.emitted_types(recording.session_id, derived) + SlimDocument(id=document_id, doc_created_at=created_at, external_access=access) + for document_id in new_ids ] - - -def _slim_document(work: OccurrenceWork) -> SlimDocument: - return SlimDocument( - id=zoom_document_id(work.session_type, work.occurrence_uuid), - doc_created_at=_created_at(work.start_time), - ) def _created_at(start_time: str | None) -> datetime | None: diff --git a/backend/onyx/connectors/zoom/recordings/models.py b/backend/onyx/connectors/zoom/recordings/models.py index 284bd014be..c852c9ee1e 100644 --- a/backend/onyx/connectors/zoom/recordings/models.py +++ b/backend/onyx/connectors/zoom/recordings/models.py @@ -44,15 +44,21 @@ class OccurrenceWork(BaseModel): class HostScope(BaseModel): - """The host and Group mechanisms scope by `session_types`, keeping every - recording of those types. The ID allowlist scopes by `sessions`, keeping - those numbers whatever else the host recorded.""" + """Which of one host's recordings to keep, and as which document type.""" host: Host + # Set when the admin listed this host by email or through a Zoom Group. + # Holds the ticked checkboxes, meetings and/or webinars: keep every + # recording of those kinds. session_types: frozenset[ZoomSessionType] = frozenset() + # Set when the admin typed a meeting or webinar number this host owns. + # Each entry is (the field it was typed into, the number), for example + # (MEETING, "111"): keep that number's recordings as meetings. sessions: frozenset[tuple[ZoomSessionType, str]] = frozenset() def merged_with(self, other: "HostScope") -> "HostScope": + """Join the scopes of a host that was both listed by email or Group and + named by a number, so the host is walked once.""" return HostScope( host=Host( user_id=self.host.user_id, @@ -62,25 +68,30 @@ class HostScope(BaseModel): sessions=self.sessions | other.sessions, ) + # TODO(subash): reject a number typed into the wrong field at validation and + # collapse this to one type. Zoom lists a webinar's occurrences on + # /past_meetings/{id}/instances, so a webinar number in the meeting field is + # indexed under the wrong type today; /meetings/{id} answers such a number + # with 400 code 3000, which is the check. Checked live on 2026-09-23. def emitted_types( self, session_id: str, derived: ZoomSessionType | None ) -> set[ZoomSessionType]: - """Every session type any scope claims for this recording. + """Which document types to make for one recording. - Generous on purpose, because an id the index never held costs nothing - while a missing one deletes a document. That covers a number configured - as both a meeting and a webinar, a number typed into the wrong field, - and a type code Zoom adds after this was written.""" - claimed = { + The recording is kept as a meeting if the admin typed its number into + the meeting field, and likewise for webinars. It is also kept as its + own kind, `derived`, when that checkbox is ticked. When Zoom's type + code is one this connector cannot read, `derived` is None and the + recording is kept as every ticked kind: an extra id costs nothing, + while a missing one deletes a document.""" + typed_as = { session_type - for session_type, claimed_id in self.sessions - if claimed_id == session_id + for session_type, number in self.sessions + if number == session_id } if derived is None: - return claimed | set(self.session_types) - if derived in self.session_types: - claimed.add(derived) - return claimed + return typed_as | set(self.session_types) + return typed_as | ({derived} & self.session_types) class SessionHost(BaseModel): @@ -94,6 +105,14 @@ class SessionHost(BaseModel): anchor: ZoomRecordingEntry | None = None +class ProvenRecording(BaseModel): + """A recording Zoom answered for by its session number, under the type the + admin typed the number into.""" + + session_type: ZoomSessionType + recording: ZoomRecordingEntry + + class RecordingsState(BaseModel): source_index: int = 0 # Only the active source may read what is inside this. The connector diff --git a/backend/onyx/connectors/zoom/recordings/processing.py b/backend/onyx/connectors/zoom/recordings/processing.py index c3eef5ab87..b8839316c9 100644 --- a/backend/onyx/connectors/zoom/recordings/processing.py +++ b/backend/onyx/connectors/zoom/recordings/processing.py @@ -3,9 +3,6 @@ Anything that differs between meetings and webinars belongs on the SessionTypeHandler, not in a branch here. """ -from collections.abc import Callable - -from onyx.access.models import ExternalAccess from onyx.configs.constants import DocumentSource from onyx.connectors.cross_connector_utils.miscellaneous_utils import time_str_to_utc from onyx.connectors.models import ( @@ -15,14 +12,16 @@ from onyx.connectors.models import ( TextSection, ) from onyx.connectors.zoom.client import ZoomClient -from onyx.connectors.zoom.models import ZoomRecordingEntry from onyx.connectors.zoom.recordings.models import ( OccurrenceWork, ZoomSessionType, fails_the_whole_run, has_no_transcript, ) -from onyx.connectors.zoom.recordings.recording_access import ZoomAccessListUnavailable +from onyx.connectors.zoom.recordings.recording_access import ( + AccessResolver, + ZoomAccessListUnavailable, +) from onyx.connectors.zoom.recordings.session_types import get_session_type_handler from onyx.file_processing.webvtt import parse_vtt_transcript from onyx.utils.logger import setup_logger @@ -63,7 +62,7 @@ def process_occurrence( client: ZoomClient, work: OccurrenceWork, *, - resolve_access: Callable[[ZoomRecordingEntry], ExternalAccess] | None, + resolve_access: AccessResolver | None, ) -> Document | ConnectorFailure | None: """One occurrence is at most one transcript, so this answers with the document, the failure that replaces it, or nothing when the occurrence has diff --git a/backend/onyx/connectors/zoom/recordings/recording_access.py b/backend/onyx/connectors/zoom/recordings/recording_access.py index 3ad5975898..4cfd02f228 100644 --- a/backend/onyx/connectors/zoom/recordings/recording_access.py +++ b/backend/onyx/connectors/zoom/recordings/recording_access.py @@ -13,6 +13,7 @@ catalogue is only asked for when a recording names a rule. from collections.abc import Callable, Sequence from typing import NamedTuple +import requests from pydantic import ValidationError from onyx.access.models import ExternalAccess @@ -27,11 +28,16 @@ from onyx.connectors.zoom.models import ( ZoomRecordingSettings, ZoomShareRecording, ) -from onyx.connectors.zoom.recordings.models import user_does_not_exist +from onyx.connectors.zoom.recordings.models import ( + definitely_absent, + user_does_not_exist, +) from onyx.utils.logger import setup_logger logger = setup_logger() +AccessResolver = Callable[[ZoomRecordingEntry], ExternalAccess] + # Zoom's `authentication_option` for "Only people with access". It never appears # in the rule catalogue, and it arrives with `share_recording` still "publicly", # so it is checked before that field is read. @@ -99,6 +105,12 @@ class ZoomAccessListUnavailable(Exception): the whole attempt.""" +class ZoomRecordingGone(Exception): + """The recording's own settings answered not found, so it was deleted after + it was listed. A not-found from any other call says nothing about the + recording, so it is not turned into this.""" + + def approved_registrant_emails( registrants: Sequence[ZoomRecordingRegistrant], ) -> list[str]: @@ -147,7 +159,7 @@ def resolve_recording_access( """Raises ZoomAccessListUnavailable rather than answering with an empty list, which would read as nobody having access.""" try: - settings = client.get_recording_settings(recording.uuid) + settings = _recording_settings(client, recording.uuid) grant = _link_access( settings, recording.uuid, treat_link_access_as_public, rule_grant ) @@ -205,6 +217,17 @@ def resolve_recording_access( return access +def _recording_settings(client: ZoomClient, uuid: str) -> ZoomRecordingSettings: + try: + return client.get_recording_settings(uuid) + except requests.HTTPError as e: + if definitely_absent(e): + raise ZoomRecordingGone( + f"Zoom recording {uuid} is gone since it was listed" + ) from e + raise + + def _link_access( settings: ZoomRecordingSettings, uuid: str, diff --git a/backend/tests/unit/ee/onyx/external_permissions/zoom/test_zoom_doc_sync.py b/backend/tests/unit/ee/onyx/external_permissions/zoom/test_zoom_doc_sync.py new file mode 100644 index 0000000000..4049536323 --- /dev/null +++ b/backend/tests/unit/ee/onyx/external_permissions/zoom/test_zoom_doc_sync.py @@ -0,0 +1,85 @@ +"""The Zoom doc sync hands the connector's permission-aware walk to the shared +sync, which turns it into access rows and hides what the walk no longer lists.""" + +from unittest.mock import MagicMock, patch + +from ee.onyx.configs.app_configs import ZOOM_PERMISSION_DOC_SYNC_FREQUENCY +from ee.onyx.external_permissions.sync_params import get_source_perm_sync_config +from ee.onyx.external_permissions.zoom.doc_sync import zoom_doc_sync +from onyx.access.models import DocExternalAccess, ExternalAccess +from onyx.configs.constants import DocumentSource +from onyx.connectors.models import SlimDocument + +MODULE = "ee.onyx.external_permissions.zoom.doc_sync" + +OWNER = ExternalAccess( + external_user_emails={"owner@example.com"}, + external_user_group_ids=set(), + is_public=False, +) + + +def _cc_pair() -> MagicMock: + cc_pair = MagicMock() + cc_pair.id = 7 + cc_pair.connector.connector_specific_config = {"host_emails": ["owner@example.com"]} + cc_pair.connector.indexing_start = None + cc_pair.credential.credential_json.get_value.return_value = { + "zoom_account_id": "acct" + } + return cc_pair + + +def test_the_walk_becomes_access_rows_and_unlisted_documents_go_private() -> None: + connector = MagicMock() + connector.retrieve_all_slim_docs_perm_sync.return_value = iter( + [ + [ + SlimDocument(id="ZOOM_MEETING_uuid-1", external_access=OWNER), + SlimDocument( + id="ZOOM_MEETING_uuid-2", external_access=ExternalAccess.empty() + ), + ] + ] + ) + + with patch(f"{MODULE}.ZoomConnector", return_value=connector) as factory: + rows = list( + zoom_doc_sync( + _cc_pair(), + MagicMock(), + lambda: [ + "ZOOM_MEETING_uuid-1", + "ZOOM_MEETING_uuid-2", + "ZOOM_MEETING_gone", + ], + None, + ) + ) + + factory.assert_called_once_with(host_emails=["owner@example.com"]) + connector.load_credentials.assert_called_once_with({"zoom_account_id": "acct"}) + assert rows == [ + DocExternalAccess(doc_id="ZOOM_MEETING_uuid-1", external_access=OWNER), + DocExternalAccess( + doc_id="ZOOM_MEETING_uuid-2", external_access=ExternalAccess.empty() + ), + DocExternalAccess( + doc_id="ZOOM_MEETING_gone", external_access=ExternalAccess.empty() + ), + ] + + +def test_zoom_is_registered_with_its_own_doc_sync() -> None: + config = get_source_perm_sync_config(DocumentSource.ZOOM) + assert config is not None and config.doc_sync_config is not None + cc_pair, fetch_docs, fetch_ids = MagicMock(), MagicMock(), MagicMock() + + # The registry holds a lazy wrapper, so calling it is what shows where it leads. + with patch(f"{MODULE}.zoom_doc_sync") as doc_sync: + config.doc_sync_config.doc_sync_func(cc_pair, fetch_docs, fetch_ids, None) + + doc_sync.assert_called_once_with(cc_pair, fetch_docs, fetch_ids, None) + assert ( + config.doc_sync_config.doc_sync_frequency == ZOOM_PERMISSION_DOC_SYNC_FREQUENCY + ) diff --git a/backend/tests/unit/onyx/connectors/zoom/test_zoom_slim_retrieval.py b/backend/tests/unit/onyx/connectors/zoom/test_zoom_slim_retrieval.py index 9b8c67f77f..ab760e3d07 100644 --- a/backend/tests/unit/onyx/connectors/zoom/test_zoom_slim_retrieval.py +++ b/backend/tests/unit/onyx/connectors/zoom/test_zoom_slim_retrieval.py @@ -6,6 +6,7 @@ does it raise rather than answer short. """ import itertools +from collections.abc import Callable from datetime import date, datetime, timezone from typing import Any from unittest.mock import MagicMock, patch @@ -13,11 +14,13 @@ from unittest.mock import MagicMock, patch import pytest import requests +from onyx.access.models import ExternalAccess from onyx.background.celery.celery_utils import extract_ids_from_runnable_connector from onyx.connectors.exceptions import ( ConnectorValidationError, InsufficientPermissionsError, ) +from onyx.connectors.interfaces import GenerateSlimDocumentOutput from onyx.connectors.models import ConnectorMissingCredentialError, SlimDocument from onyx.connectors.zoom.connector import ZoomConnector from onyx.connectors.zoom.models import ( @@ -28,6 +31,7 @@ from onyx.connectors.zoom.models import ( ZoomUser, ZoomUserPage, ) +from onyx.connectors.zoom.recordings import inventory from onyx.connectors.zoom.recordings.discovery import ( EARLIEST_RECORDING_DATE, listing_windows, @@ -36,11 +40,16 @@ from onyx.connectors.zoom.recordings.inventory import ( _MAX_DOCUMENTS_PER_BATCH, _SCOPES_PER_HEARTBEAT, ) -from tests.unit.onyx.connectors.zoom.helpers import http_error, mock_zoom_client +from tests.unit.onyx.connectors.zoom.helpers import ( + http_error, + mock_zoom_client, + with_recording_access, +) from tests.unit.onyx.connectors.zoom.zoom_api_shapes import ( meeting_details, past_meeting_details, recording_entry, + recording_settings, user, webinar_details, ) @@ -54,6 +63,14 @@ _ZOOM_CREDS = { # The id every recording builder defaults to. _NUMBER = "6840331990" +# What the access stub's default share settings resolve to: the owner, and +# public because the recording is shared with the account and the box is on. +_ACCOUNT_SHARE = ExternalAccess( + external_user_emails={"owner@example.com"}, + external_user_group_ids=set(), + is_public=True, +) + def _windows_per_host() -> int: """What one host costs the walk. Zoom's launch is the listing floor and @@ -101,19 +118,27 @@ def _listing(client: MagicMock, *recordings: ZoomRecordingEntry) -> None: client.list_user_recordings.return_value = _recording_page(*recordings) -def _documents(connector: ZoomConnector) -> list[SlimDocument]: +def _items(batches: GenerateSlimDocumentOutput) -> list[SlimDocument]: return [ - item - for batch in connector.retrieve_all_slim_docs() - for item in batch - if isinstance(item, SlimDocument) + item for batch in batches for item in batch if isinstance(item, SlimDocument) ] +def _documents(connector: ZoomConnector) -> list[SlimDocument]: + return _items(connector.retrieve_all_slim_docs()) + + def _ids(connector: ZoomConnector) -> set[str]: return {document.id for document in _documents(connector)} +def _synced(connector: ZoomConnector) -> dict[str, ExternalAccess | None]: + return { + item.id: item.external_access + for item in _items(connector.retrieve_all_slim_docs_perm_sync()) + } + + def _windows(client: MagicMock) -> list[dict[str, Any]]: """The listing calls that were windows of the walk. The one-day probe that checks a host exists before the walk is not one.""" @@ -264,14 +289,24 @@ class TestSlimFailuresNeverDeleteAnything: assert asked.count("gone@example.com") == 1 assert "gone@example.com" not in _walked(client) - def test_a_credential_that_recognises_nobody_stops_the_prune(self) -> None: + @pytest.mark.parametrize( + ("walk", "consequence"), + [ + (_documents, "pruning stopped rather than delete"), + (_synced, "permission sync stopped rather than make"), + ], + ids=["pruning", "doc-sync"], + ) + def test_a_credential_that_recognises_nobody_stops_the_walk( + self, walk: Callable[[ZoomConnector], object], consequence: str + ) -> None: # Zoom answers the same 1001 for a deleted user and for one in another # account, so a credential pointed elsewhere would delete everything. connector, client = _connector(host_emails=["jill@example.com"]) client.list_user_recordings.side_effect = _listing_by_host({}) - with pytest.raises(ConnectorValidationError, match="recognised none"): - list(connector.retrieve_all_slim_docs()) + with pytest.raises(ConnectorValidationError, match=consequence): + walk(connector) @pytest.mark.parametrize( "error", @@ -571,3 +606,168 @@ class TestBatchingKeepsThePruneAlive: assert len(batches) > 1 assert all(len(batch) <= _MAX_DOCUMENTS_PER_BATCH for batch in batches[:-1]) + + +class TestThePermSyncWalk: + """The doc sync makes every indexed document the walk leaves out private, + so it lists exactly what pruning lists, with each recording's access.""" + + def test_every_document_carries_the_recordings_access(self) -> None: + connector, client = _connector(host_emails=["jill@example.com"]) + with_recording_access(client) + _listing(client, _recording("uuid-1"), _recording("uuid-2")) + + synced = _synced(connector) + + assert synced == { + "ZOOM_MEETING_uuid-1": _ACCOUNT_SHARE, + "ZOOM_MEETING_uuid-2": _ACCOUNT_SHARE, + } + assert set(synced) == _ids(connector) + + def test_pruning_asks_for_no_access(self) -> None: + connector, client = _connector(host_emails=["jill@example.com"]) + with_recording_access(client) + _listing(client, _recording("uuid-1")) + + documents = _documents(connector) + + assert [d.external_access for d in documents] == [None] + client.get_recording_settings.assert_not_called() + + def test_a_recording_listed_again_costs_no_second_access_call(self) -> None: + # The stub answers every window and the trailing pass with the same + # recording, so the walk lists it many times over. + connector, client = _connector(host_emails=["jill@example.com"]) + with_recording_access(client) + _listing(client, _recording("uuid-1")) + + _synced(connector) + + client.get_recording_settings.assert_called_once() + client.get_user.assert_called_once() + + def test_past_the_remembered_ids_cap_nothing_is_lost(self) -> None: + connector, client = _connector(host_emails=["jill@example.com"]) + with_recording_access(client) + _listing(client, _recording("uuid-1"), _recording("uuid-2")) + + with patch.object(inventory, "_MAX_REMEMBERED_IDS", 1): + synced = _synced(connector) + + assert set(synced) == {"ZOOM_MEETING_uuid-1", "ZOOM_MEETING_uuid-2"} + # The second recording was not remembered, so each listing resolved it. + assert client.get_recording_settings.call_count > 2 + + def test_a_recording_nobody_can_be_named_for_is_private(self) -> None: + connector, client = _connector(host_emails=["jill@example.com"]) + with_recording_access( + client, settings=recording_settings(share_recording="none") + ) + client.get_user.side_effect = http_error(404, 1001) + _listing(client, _recording("uuid-1")) + + assert _synced(connector) == {"ZOOM_MEETING_uuid-1": ExternalAccess.empty()} + + def test_the_walk_ends_with_a_count_of_what_went_private( + self, caplog: pytest.LogCaptureFixture + ) -> None: + connector, client = _connector(host_emails=["jill@example.com"]) + with_recording_access( + client, settings=recording_settings(share_recording="none") + ) + client.get_user.side_effect = http_error(404, 1001) + _listing(client, _recording("uuid-1"), _recording("uuid-2")) + + with caplog.at_level("WARNING"): + _synced(connector) + + said = [r for r in caplog.records if "made 2 recording(s)" in r.getMessage()] + assert len(said) == 1 + assert "2 that nobody could be named for" in said[0].getMessage() + + def test_a_recording_deleted_mid_walk_is_private(self) -> None: + connector, client = _connector(host_emails=["jill@example.com"]) + with_recording_access(client) + client.get_recording_settings.side_effect = http_error(404) + _listing(client, _recording("uuid-1")) + + assert _synced(connector) == {"ZOOM_MEETING_uuid-1": ExternalAccess.empty()} + + def test_any_other_settings_error_fails_the_sync(self) -> None: + connector, client = _connector(host_emails=["jill@example.com"]) + with_recording_access(client) + client.get_recording_settings.side_effect = http_error(400, 1234) + _listing(client, _recording("uuid-1")) + + with pytest.raises(requests.HTTPError): + _synced(connector) + + @pytest.mark.parametrize( + "refuse", + [ + lambda client: client.get_recording_authentication_rules, + lambda client: client.get_user, + ], + ids=["rule-catalogue", "owner-lookup"], + ) + def test_a_404_from_any_other_call_fails_the_sync( + self, refuse: Callable[[MagicMock], MagicMock] + ) -> None: + # Only the recording's own settings answering 404 means it was deleted. + connector, client = _connector(host_emails=["jill@example.com"]) + with_recording_access(client) + refuse(client).side_effect = http_error(404) + _listing(client, _recording("uuid-1"), _recording("uuid-2")) + + with pytest.raises(requests.HTTPError): + _synced(connector) + + def test_a_proven_recording_is_resolved_from_its_anchor(self) -> None: + connector, client = _connector(meeting_ids=[_NUMBER]) + with_recording_access(client) + client.get_recording.return_value = _recording("uuid-proven") + client.list_user_recordings.side_effect = itertools.repeat( + http_error(404, 1001) + ) + + assert _synced(connector) == {"ZOOM_MEETING_uuid-proven": _ACCOUNT_SHARE} + + def test_a_proven_recording_its_host_also_lists_is_resolved_once(self) -> None: + connector, client = _connector(meeting_ids=[_NUMBER]) + with_recording_access(client) + recording = _recording("uuid-proven") + client.get_recording.return_value = recording + _listing(client, recording) + + assert _synced(connector) == {"ZOOM_MEETING_uuid-proven": _ACCOUNT_SHARE} + client.get_recording_settings.assert_called_once() + + def test_a_recording_with_no_host_is_resolved_without_an_owner(self) -> None: + connector, client = _connector(host_emails=["jill@example.com"]) + with_recording_access(client) + _listing(client, _recording("uuid-1", host_id="")) + + synced = _synced(connector) + + assert synced == { + "ZOOM_MEETING_uuid-1": ExternalAccess( + external_user_emails=set(), + external_user_group_ids=set(), + is_public=True, + ) + } + client.get_user.assert_not_called() + + def test_a_recording_of_unreadable_type_carries_one_access_on_every_id( + self, + ) -> None: + connector, client = _connector(host_emails=["jill@example.com"]) + with_recording_access(client) + _listing(client, _recording("uuid-1", type="4242")) + + assert _synced(connector) == { + "ZOOM_MEETING_uuid-1": _ACCOUNT_SHARE, + "ZOOM_WEBINAR_uuid-1": _ACCOUNT_SHARE, + } + client.get_recording_settings.assert_called_once() diff --git a/cli/internal/deploy/deployfiles/embedded/docker_compose/env.template b/cli/internal/deploy/deployfiles/embedded/docker_compose/env.template index 1ccd2d8c7d..95877cc568 100644 --- a/cli/internal/deploy/deployfiles/embedded/docker_compose/env.template +++ b/cli/internal/deploy/deployfiles/embedded/docker_compose/env.template @@ -412,6 +412,7 @@ LOG_ONYX_MODEL_INTERACTIONS=False # GONG_CONNECTOR_START_TIME= # ZOOM_TRANSCRIPT_LAG_BUFFER_HOURS= # ZOOM_PERMISSION_GROUP_SYNC_FREQUENCY= +# ZOOM_PERMISSION_DOC_SYNC_FREQUENCY= # NOTION_CONNECTOR_DISABLE_RECURSIVE_PAGE_LOOKUP= # GITHUB_CONNECTOR_BASE_URL= # MAX_DOCUMENT_CHARS= diff --git a/deployment/docker_compose/env.template b/deployment/docker_compose/env.template index 1ccd2d8c7d..95877cc568 100644 --- a/deployment/docker_compose/env.template +++ b/deployment/docker_compose/env.template @@ -412,6 +412,7 @@ LOG_ONYX_MODEL_INTERACTIONS=False # GONG_CONNECTOR_START_TIME= # ZOOM_TRANSCRIPT_LAG_BUFFER_HOURS= # ZOOM_PERMISSION_GROUP_SYNC_FREQUENCY= +# ZOOM_PERMISSION_DOC_SYNC_FREQUENCY= # NOTION_CONNECTOR_DISABLE_RECURSIVE_PAGE_LOOKUP= # GITHUB_CONNECTOR_BASE_URL= # MAX_DOCUMENT_CHARS= diff --git a/deployment/helm/charts/onyx/Chart.yaml b/deployment/helm/charts/onyx/Chart.yaml index 632e6e1f3f..7cec2198b8 100644 --- a/deployment/helm/charts/onyx/Chart.yaml +++ b/deployment/helm/charts/onyx/Chart.yaml @@ -5,7 +5,7 @@ home: https://www.onyx.app/ sources: - "https://github.com/onyx-dot-app/onyx" type: application -version: 0.9.7 +version: 0.9.8 appVersion: latest annotations: category: Productivity diff --git a/deployment/helm/charts/onyx/values.yaml b/deployment/helm/charts/onyx/values.yaml index df9547332d..53cc5da87b 100644 --- a/deployment/helm/charts/onyx/values.yaml +++ b/deployment/helm/charts/onyx/values.yaml @@ -1781,6 +1781,7 @@ configMap: GONG_CONNECTOR_START_TIME: "" ZOOM_TRANSCRIPT_LAG_BUFFER_HOURS: "" ZOOM_PERMISSION_GROUP_SYNC_FREQUENCY: "" + ZOOM_PERMISSION_DOC_SYNC_FREQUENCY: "" NOTION_CONNECTOR_DISABLE_RECURSIVE_PAGE_LOOKUP: "" # Push every successfully indexed public-connector document to an external # HTTP endpoint (failures never fail indexing). Intended for non-EE