"""Google Workspace software-provisioning driver. One integration row = one Google Workspace tenant (one Service Account JSON + one impersonation subject). Per-software bindings carry the operation-specific parameters (which group to add the user to, whether to suspend the account entirely) in their ``binding_config`` JSON — NOT in the integration config. That keeps credentials DRY across all softwares that use Google. The binding_config shape is: {"operation": "group_member", "group_email": "eng@example.com"} {"operation": "user_suspend"} Passed through to the driver via :class:`ProvisioningContext.extra`. """ from __future__ import annotations import asyncio import json import secrets import string from dataclasses import dataclass, field from datetime import UTC, datetime from typing import Any from google.oauth2 import service_account from googleapiclient.discovery import build from googleapiclient.errors import HttpError from app.core.safe_logging import get_safe_logger from app.integrations.driver_base import ( ProvisioningContext, ProvisioningResult, SoftwareProvisioningDriver, ) from app.integrations.registry import register_driver from app.integrations.schemas import ( IntegrationFieldInfo, IntegrationTestResult, ) logger = get_safe_logger(__name__) # ── Error tags ─────────────────────────────────────────────────── # # Mirrors the Zoom + Feishu tag convention: every admin-actionable # failure carries an ``error_tag`` in ``raw_response`` AND a # ``[TAG] …`` prefix on ``error_message`` so the failure-DM # dispatcher (``_dispatch_revoke_failure_dms``) can append a # Chinese remediation hint via ``_extract_remediation_hint``. Pairs # with new entries in ``_PROVISIONING_REMEDIATION_HINTS``. GOOGLE_TAG_SUCCESSOR_NOT_IN_DIRECTORY = "GOOGLE_SUCCESSOR_NOT_IN_DIRECTORY" GOOGLE_TAG_TRANSFER_PERMISSION_DENIED = "GOOGLE_TRANSFER_PERMISSION_DENIED" GOOGLE_TAG_TRANSFER_TIMEOUT = "GOOGLE_TRANSFER_TIMEOUT" GOOGLE_TAG_SCOPE_MISSING = "GOOGLE_SCOPE_MISSING" # How long to wait between transient-error polls in # ``_poll_data_transfer`` when Google returns 429 / 5xx. Bounded by # the surrounding ``budget_seconds`` budget so a wedged Google # datacenter doesn't hang the worker beyond the configured envelope. _GOOGLE_POLL_TRANSIENT_BACKOFF = 5.0 # Status codes that Google returns when the request is well-formed # but the server-side is temporarily unable to serve it (overload # protection, lame-duck pool, ALB hiccup, edge-LB request timeout). # Retrying with backoff is the documented behaviour — these are NOT # permanent failures. 408 (Request Timeout) is per HTTP RFC + Google's # Cloud documented retry guidance; included for completeness even # though Google's own surfaces emit it rarely. _GOOGLE_TRANSIENT_STATUSES: frozenset[int] = frozenset( {408, 429, 500, 502, 503, 504}, ) # Exponential-backoff schedule for the WRITE-path retry wrapper # (``_call_with_transient_retry``). 3 attempts total, sleeps in # between → max wall-clock ~26s before final failure. Same shape as # the poll-loop backoff but front-loaded with one extra short attempt # so transient hiccups (the kind that caused PR #fix's Google 502 # spam) don't even reach the failure-DM dispatcher. _GOOGLE_RETRY_BACKOFF_SECONDS: tuple[float, ...] = (2.0, 6.0, 18.0) _GROUP_SCOPE = "https://www.googleapis.com/auth/admin.directory.group.member" _GROUP_ADMIN_SCOPE = "https://www.googleapis.com/auth/admin.directory.group" _USER_SCOPE = "https://www.googleapis.com/auth/admin.directory.user" # Session-security scope — REQUIRED by ``users.signOut`` (force-logout). # The broad ``admin.directory.user`` scope grants user CRUD (create / # update / suspend / password) but NOT the session-level security # operation ``signOut``, which Google gates behind this narrower scope. # Production 403 post-mortem (2026-07-04, P0051 offboarding): the DWD # service account IS already authorised for this scope (verified via a # read-only signOut probe against a non-existent userKey → 404, i.e. # scope OK, user absent — whereas ``admin.directory.user`` alone → 403 # "insufficient authentication scopes"). The bug was purely that # ``force_logout_user`` requested the wrong scope. Kept separate from # ``_USER_SCOPE`` so a tenant that only delegated the CRUD scope still # gets every other feature; only force-logout degrades if this scope is # missing from domain-wide delegation. _USER_SECURITY_SCOPE = ( "https://www.googleapis.com/auth/admin.directory.user.security" ) _USER_ALIAS_SCOPE = "https://www.googleapis.com/auth/admin.directory.user.alias" _DATA_TRANSFER_SCOPE = "https://www.googleapis.com/auth/admin.datatransfer" # Admin SDK Reports API (read-only audit) — powers # ``list_saml_login_events``, the evidence source for "did this person keep # logging into the SAML app after Helix revoked the grant" (SSO removal # verification, migration 603). Kept separate from every other scope list: # it is a DIFFERENT API (``admin/reports_v1``) and a tenant that delegated # only the Directory scopes must keep working — the SSO verifier records # ``google_reports_scope_missing`` on 403 instead of failing. _REPORTS_AUDIT_SCOPE = "https://www.googleapis.com/auth/admin.reports.audit.readonly" # Cloud Identity (NOT Admin Directory) read-only scope — powers # ``list_dynamic_group_emails``. Distinct API surface # (cloudidentity.googleapis.com), so it needs its own service build and # its own domain-wide-delegation entry. Kept separate from the Directory # scopes: if an admin hasn't added this scope yet, only dynamic-group # detection degrades (group sync + every other feature keeps working). _CLOUD_IDENTITY_GROUPS_RO_SCOPE = ( "https://www.googleapis.com/auth/cloud-identity.groups.readonly" ) # Cloud Identity read-WRITE scope — powers dynamic-group create/patch # (membership query, display name, description). A superset of the RO # scope, so a single DWD entry with this value is enough to both read # and write dynamic groups. Kept separate so a tenant that only wants # the read-only「我的邮件组」detection can delegate the narrower RO # scope and never grant write. One-time DWD setup: add this value to # the same Service Account's domain-wide-delegation allow-list. _CLOUD_IDENTITY_GROUPS_SCOPE = ( "https://www.googleapis.com/auth/cloud-identity.groups" ) # Cloud Identity label that marks a group as a **dynamic** group. Used # in the ``groups.search`` query filter — the ONLY reliable way to # enumerate dynamic groups (see ``list_dynamic_groups`` for why # ``groups.list`` is not). _CLOUD_IDENTITY_DYNAMIC_LABEL = "cloudidentity.googleapis.com/groups.dynamic" # Required label every Cloud Identity group must carry (Google rejects # ``groups.create`` without it). _CLOUD_IDENTITY_DISCUSSION_LABEL = ( "cloudidentity.googleapis.com/groups.discussion_forum" ) # Note (2026-05-07): the previous PR added Cloud Identity # ``inboundSamlSsoProfiles`` scopes here for an "SSO Apps catalog # sync" feature. That API turned out to return INBOUND SSO profiles # (third-party IdP signing INTO Google), not the OUTBOUND "Apps → # Web and mobile apps" list (Google as IdP for external SaaS apps). # Google has no public API for the latter, so the SSO catalog is # admin-managed in Helix instead. The scope constants + driver # methods that called Cloud Identity have been removed; see memory # `feedback_verify_third_party_api_direction.md` for the lesson. # Hard cap on pagination iterations across every paginated Google # API call below. 500 pages × 500/page = 250k entries — well above # any realistic 500-employee tenant. Same rationale as slack/zoom/ # jira/feishu drivers (P1 from CR2). _MAX_PAGINATOR_PAGES = 500 # Google Data Transfer application ids (stable, documented constants). # Values come from /admin/datatransfer/v1/applications; hard-coded here # to avoid an extra lookup call on every revoke. _DATA_TRANSFER_APP_IDS: dict[str, str] = { "drive": "55656082996", "calendar": "435070579839", } # Scopes for the elevated "workspace ops" methods added in PR-D # (group create / delete / alias / password / owner transfer). Kept # separate from the seat-grant scopes so a Service Account Key # provisioned only for membership still works for existing features. _WORKSPACE_OPS_SCOPES = [ _GROUP_ADMIN_SCOPE, _GROUP_SCOPE, _USER_SCOPE, _USER_ALIAS_SCOPE, ] # Read-only group scope, for methods that only ever GET. Kept separate # from ``_WORKSPACE_OPS_SCOPES`` for the same reason that list already # hard-codes it: a tenant may delegate the readonly scope and NOT the # write ones, and asking for an undelegated scope fails at token # exchange (``unauthorized_client``) — so a read that requests write # scopes it does not need becomes permanently indeterminate on a # readonly-delegated tenant. _GROUP_READ_SCOPES = [ "https://www.googleapis.com/auth/admin.directory.group.readonly", ] class GoogleApiError(RuntimeError): """Raised when a Google Admin SDK call fails in a way the caller needs to surface verbatim — distinct from the swallow-and-log pattern used for best-effort list operations. The original ``HttpError.resp.status`` + Google's ``reason`` land in ``status`` and ``reason`` so handlers can map 403 → "service account missing scope X" vs 404 → "tenant domain misconfigured" without parsing strings. ``operation`` identifies which driver method raised it so the scheduled-jobs summary reads cleanly ("list_all_groups: 403 forbidden"). """ def __init__( self, operation: str, *, status: int | None = None, reason: str | None = None, hint: str | None = None, ) -> None: parts = [f"{operation} failed"] if status: parts.append(f"HTTP {status}") if reason: parts.append(reason) if hint: parts.append(f"({hint})") super().__init__(": ".join(parts[:1]) + " — " + " | ".join(parts[1:])) self.operation = operation self.status = status self.reason = reason self.hint = hint @dataclass(slots=True) class WorkspaceOpResult: """Result shape for the Google Workspace "ops" methods. ``ok=True`` → ``data`` carries the driver's understanding of the remote object (group resource id, new owner email, new password, etc.). ``ok=False`` → ``error_message`` is human-readable, ``status`` is the HTTP status when known. ``already_exists`` lets the caller treat 409/alreadyExists as success-idempotent vs. failure. """ ok: bool data: dict[str, Any] = field(default_factory=dict) error_message: str | None = None status: int | None = None already_exists: bool = False # 404-idempotent twin of ``already_exists``: the target is already # gone, so a delete/revoke/signOut is a no-op success. Referenced by # ``force_logout_user`` + the account-lifecycle coordinator before the # field existed → latent TypeError on the signOut 404 path; added # 2026-07-04. already_absent: bool = False def _build_directory_service(sa_json: str, subject: str, scopes: list[str]): info = json.loads(sa_json) creds = service_account.Credentials.from_service_account_info( info, scopes=scopes, subject=subject, ) return build("admin", "directory_v1", credentials=creds, cache_discovery=False) def _build_datatransfer_service(sa_json: str, subject: str): info = json.loads(sa_json) creds = service_account.Credentials.from_service_account_info( info, scopes=[_DATA_TRANSFER_SCOPE], subject=subject, ) return build( "admin", "datatransfer_v1", credentials=creds, cache_discovery=False, ) def _build_cloudidentity_service(sa_json: str, subject: str, scopes: list[str]): info = json.loads(sa_json) creds = service_account.Credentials.from_service_account_info( info, scopes=scopes, subject=subject, ) return build( "cloudidentity", "v1", credentials=creds, cache_discovery=False, ) def _build_reports_service(sa_json: str, subject: str): info = json.loads(sa_json) creds = service_account.Credentials.from_service_account_info( info, scopes=[_REPORTS_AUDIT_SCOPE], subject=subject, ) return build("admin", "reports_v1", credentials=creds, cache_discovery=False) def _rfc3339(dt: datetime) -> str: """Reports API wants RFC 3339 with an explicit offset; naive → UTC.""" if dt.tzinfo is None: dt = dt.replace(tzinfo=UTC) text: str = dt.astimezone(UTC).isoformat(timespec="seconds") return text.replace("+00:00", "Z") def _parse_rfc3339(raw: str | None) -> datetime | None: if not raw: return None text = raw.strip() if text.endswith("Z"): text = text[:-1] + "+00:00" try: parsed = datetime.fromisoformat(text) except ValueError: return None return parsed if parsed.tzinfo is not None else parsed.replace(tzinfo=UTC) async def _call_with_transient_retry( fn, *, label: str, backoff_schedule: tuple[float, ...] = _GOOGLE_RETRY_BACKOFF_SECONDS, extra_retry_statuses: frozenset[int] = frozenset(), ): """Run a sync Google API callable via ``asyncio.to_thread`` and retry on transient HTTP statuses (429 / 5xx) with the provided backoff schedule. Why this exists: the in-flight production failure ("Google Data Transfer init failed 502: Bad Gateway") was caused by Google's Admin Data Transfer API returning a single 502 — a textbook transient. Without retry, that 502 short-circuits the whole revoke task and triggers an admin failure DM, then the seat gets rolled back to ``active``. With the schedule below the next two attempts ~26s later almost always succeed, the task completes, and the admin sees nothing. Non-transient errors (4xx other than 429, including 401/403/404/409) are NOT retried — those are real "you need to fix something" cases and a retry loop only delays the actionable signal. The caller is responsible for distinguishing already-done conditions (404 source user gone, 409 already member) and returning idempotent success. ``extra_retry_statuses`` lets ONE call site opt a normally-permanent status into the retry schedule without widening ``_GOOGLE_TRANSIENT_STATUSES`` for every Google call in the driver. Used by ``_set_suspended`` for 412 ``User creation is not complete.`` — see the settle-window comment there. Returns the callable's result on success. Raises the FINAL ``HttpError`` (or non-HttpError exception) on exhaustion, preserving the original traceback so the caller's existing error-mapping branches still match. ``label`` is used purely for logging. """ last_exc: Exception | None = None attempts = len(backoff_schedule) + 1 for attempt in range(attempts): try: return await asyncio.to_thread(fn) except HttpError as e: status = e.resp.status if getattr(e, "resp", None) else None if status not in (_GOOGLE_TRANSIENT_STATUSES | extra_retry_statuses): # Permanent error — surface immediately so the caller's # existing 4xx mapping (404, 403, 409) runs without # retry-induced delay. raise last_exc = e if attempt < len(backoff_schedule): backoff = backoff_schedule[attempt] logger.warning( "google_workspace.%s transient HTTP %s " "(attempt %d/%d); backing off %.1fs", label, status, attempt + 1, attempts, backoff, ) await asyncio.sleep(backoff) continue # Exhausted: re-raise the last transient with all retry # context captured in logs. logger.warning( "google_workspace.%s exhausted %d transient retries " "on HTTP %s; surfacing failure", label, attempts, status, ) raise # Unreachable — loop either returns or raises. Defensive guard # so the function never returns ``None`` if the schedule is empty. if last_exc is not None: raise last_exc raise RuntimeError( f"_call_with_transient_retry({label}) exited without result" ) @register_driver class GoogleWorkspaceDriver(SoftwareProvisioningDriver): provider = "google_workspace" display_name = "Google Workspace" description = ( "Google Admin SDK(Directory v1)。支持向指定 Group 添加/移除成员," "以及停用/恢复整个用户账号。需要 Service Account JSON key 并启用 " "domain-wide delegation。" ) supports_seat = True supports_permission = False supports_multi_instance = True requires_successor_on_revoke = True @classmethod def revoke_requires_successor(cls, binding_config: dict[str, Any] | None) -> bool: """Only ``user_suspend`` — where the whole account goes dark — requires a successor to receive Drive/Calendar ownership. Group-member removals don't touch user-owned data. """ op = str((binding_config or {}).get("operation") or "") return op == "user_suspend" config_schema = [ IntegrationFieldInfo( key="service_account_json", label="Service Account JSON", is_secret=True, required=True, is_textarea=True, placeholder='{"type":"service_account", ...}', hint="Google Admin Console 创建的 Service Account Key(JSON)。", ), IntegrationFieldInfo( key="impersonate_subject_email", label="Impersonate Subject Email", is_secret=False, required=True, placeholder="admin@example.com", hint="被 Service Account 代入的 super-admin 邮箱。", ), ] # Per-software binding: admin picks an operation for each software # that uses Google Workspace. The driver dispatches in grant_seat on # ctx.extra["operation"]. binding_schema = [ IntegrationFieldInfo( key="operation", label="操作类型", is_secret=False, required=True, default="group_member", options=[ {"value": "group_member", "label": "Group 成员管理(加入/移除一个 Google Group)"}, {"value": "user_suspend", "label": "账号停用/恢复(整个 Google 账号)"}, ], hint="Group 成员模式最常用;账号停用适合 offboarding 一键切断访问。", ), IntegrationFieldInfo( key="group_email", label="Group Email", is_secret=False, required=True, visible_when={"operation": "group_member"}, placeholder="eng@example.com", hint="软件访问所依赖的 Google Group。授权 = 加入该组。", ), ] def _sa_json(self) -> str: v = self.config.get("service_account_json") if not v: raise ValueError("service_account_json is not configured") return str(v) def _subject(self) -> str: v = self.config.get("impersonate_subject_email") if not v: raise ValueError("impersonate_subject_email is not configured") return str(v) async def _resolve_customer_id(self) -> str: """Return the tenant's real ``C0xxxxxxx`` customer id. Cloud Identity ``groups.search`` requires a concrete customer id in its ``parent`` filter — the Directory-API ``my_customer`` alias is NOT accepted there (it 400s "invalid argument"). We resolve it once from the impersonated super-admin's user record (``directory.users().get(...)["customerId"]``) and cache it on the instance for the life of the driver. """ cached = getattr(self, "_customer_id_cache", None) if cached: return cached def _call() -> dict[str, Any]: svc = _build_directory_service( self._sa_json(), self._subject(), [_USER_SCOPE], ) return svc.users().get(userKey=self._subject()).execute() resp = await asyncio.to_thread(_call) cid = (resp.get("customerId") or "").strip() if not cid: raise GoogleApiError( "resolve_customer_id", reason="impersonated user has no customerId", ) self._customer_id_cache = cid return cid async def test_connection(self) -> IntegrationTestResult: def _call() -> dict[str, Any]: svc = _build_directory_service(self._sa_json(), self._subject(), [_USER_SCOPE]) return svc.users().get(userKey=self._subject()).execute() try: resp = await asyncio.to_thread(_call) except HttpError as e: return IntegrationTestResult( ok=False, message=f"Google API error {e.resp.status}: {e.reason}", detail={"status": e.resp.status}, ) except Exception as e: # noqa: BLE001 return IntegrationTestResult( ok=False, message=f"{type(e).__name__}: {e}", detail={"type": type(e).__name__}, ) return IntegrationTestResult( ok=True, message=f"OK · impersonated user {resp.get('primaryEmail')}", detail={"primary_email": resp.get("primaryEmail")}, ) # ── Dispatch based on binding's operation config ─ async def grant_seat(self, ctx: ProvisioningContext) -> ProvisioningResult: op = str((ctx.extra or {}).get("operation") or "") if op == "group_member": return await self._group_add(ctx) if op == "user_suspend": return await self._set_suspended(ctx, suspended=False) return ProvisioningResult( ok=False, error_message=( "Google Workspace binding missing or unknown " f"'operation': {op!r} (expected group_member|user_suspend)" ), ) async def revoke_seat(self, ctx: ProvisioningContext) -> ProvisioningResult: op = str((ctx.extra or {}).get("operation") or "") if op == "group_member": return await self._group_remove(ctx) if op == "user_suspend": if not ctx.successor_email: return ProvisioningResult( ok=False, error_message=( "Google Workspace user_suspend requires a successor " "to receive Drive/Calendar ownership. Retry the " "revoke with a successor selected." ), ) transfer = await self._initiate_data_transfer( old_email=ctx.user_email, new_email=ctx.successor_email, ) if not transfer.ok: return transfer # Idempotent short-circuit: ``_initiate_data_transfer`` now # detects "source user already gone from Google" (HRIS-late # leaver whose Google account was manually cleaned earlier) # and returns ok=True with ``raw_response["skipped"]`` set # but ``external_ref=""``. Skip both poll and suspend in # that case — there's nothing to transfer, and the # downstream ``_set_suspended`` would also return its own # 404-idempotent success. Returning early avoids a confusing # "empty transfer_id; cannot poll" failure on the next # statement. if ( (transfer.raw_response or {}).get("skipped") == "source_user_already_gone" ): return ProvisioningResult( ok=True, external_ref=f"gw-user:{ctx.user_email}", raw_response={ "skipped": "source_user_already_gone", "source_email": ctx.user_email, }, ) # Pre-launch P1 fix: poll the data-transfer status until # it reaches a terminal outcome (or our budget expires) # BEFORE suspending the source account. Without the poll # the suspend fires while the async transfer is still in # flight; if Google then fails the transfer (corrupt # Drive ACL, rate limit, recipient inactive), the source # account is suspended with content stranded under it # and no automatic rollback path. Poll budget caps the # wait at ~5 minutes; on timeout we DO NOT suspend, so # the operator can retry and Drive content stays accessible. polled = await self._poll_data_transfer( transfer_id=transfer.external_ref or "", budget_seconds=300, interval_seconds=10, ) if not polled.ok: return ProvisioningResult( ok=False, error_message=( "Google data-transfer did not complete within " "5 min — refusing to suspend the source account " "to avoid stranding Drive/Calendar content. " f"Detail: {polled.error_message}. Retry the " "revoke; the in-flight transfer will be reused." ), external_ref=transfer.external_ref, raw_response={ "data_transfer_id": transfer.external_ref, "data_transfer_status": ( (polled.raw_response or {}).get("status") ), }, ) result = await self._set_suspended(ctx, suspended=True) if result.ok and result.raw_response is not None: result.raw_response["data_transfer_id"] = ( transfer.external_ref ) result.raw_response["data_transfer_status"] = ( (polled.raw_response or {}).get("status") ) return result return ProvisioningResult( ok=False, error_message=( "Google Workspace binding missing or unknown " f"'operation': {op!r} (expected group_member|user_suspend)" ), ) async def _initiate_data_transfer( self, *, old_email: str, new_email: str, ) -> ProvisioningResult: """Kick off an async Drive + Calendar ownership transfer via the Admin Data Transfer API. Google queues the transfer and completes it server-side within minutes (Drive) to hours (large mailboxes). We only need a 200 here — the actual move is outside our process. We lookup the old and new user ids first because the Data Transfer API wants numeric user ids, not email addresses. """ # Round-1 audit CRITICAL #2 + HIGH #5: pre-look up the recipient # (successor) BEFORE constructing the transfer body. A 404 on # the recipient lookup is a permanent failure (the successor # doesn't exist in Google's directory yet) — tag it distinctly # so the failure-DM tells admin to provision the successor's # Google account first, instead of looping on a transfer that # will never succeed. We do the recipient lookup in a separate # try/except so the error attribution is correct. # # All three sub-calls (recipient lookup, source lookup, transfer # insert) are wrapped in ``_call_with_transient_retry`` so a # transient 502/503/429 from Google's API gateway no longer # short-circuits the revoke into a failure DM. The original # prod incident (2026-06-01) was a single 502 on insert that # caused this branch to return ok=False, the seat rolled back # to active, and the admin received a "needs successor" DM # from the duplicate orphan-sweep task on top of the real 502 # — both noise. def _lookup_recipient() -> dict[str, Any]: dir_svc = _build_directory_service( self._sa_json(), self._subject(), [_USER_SCOPE], ) return dir_svc.users().get(userKey=new_email).execute() try: recipient_user = await _call_with_transient_retry( _lookup_recipient, label="recipient_lookup", ) except HttpError as e: if e.resp.status == 404: return ProvisioningResult( ok=False, error_message=( f"[{GOOGLE_TAG_SUCCESSOR_NOT_IN_DIRECTORY}] " f"Successor {new_email!r} not found in Google " "Workspace directory. Provision the successor's " "Google account first, then retry the revoke." ), raw_response={ "error_tag": GOOGLE_TAG_SUCCESSOR_NOT_IN_DIRECTORY, "missing_recipient": new_email, }, ) if e.resp.status == 403: return ProvisioningResult( ok=False, error_message=( f"[{GOOGLE_TAG_SCOPE_MISSING}] Service Account " f"cannot read user {new_email!r} " f"(directory.user scope missing or recipient " f"domain not delegated): {e.reason}" ), raw_response={ "error_tag": GOOGLE_TAG_SCOPE_MISSING, "scope": "admin.directory.user", "subject_email": new_email, }, ) return ProvisioningResult( ok=False, error_message=( f"Google recipient lookup failed " f"{e.resp.status}: {e.reason}" ), ) except Exception as e: # noqa: BLE001 return ProvisioningResult( ok=False, error_message=( f"Google recipient lookup: {type(e).__name__}: {e}" ), ) # Look up the SOURCE user separately so a 404 on the leaver's # own Google account is recognised as "already gone" → idempotent # success. Before this branch, the source-user 404 was buried # inside the transfer-insert ``_call`` closure and surfaced as # a generic "Google Data Transfer init failed 404: Resource Not # Found: userKey" failure — which both fired the failure DM AND # blocked the offboarding blocker from resolving. Production # data showed this on two of the five 2026-06-01 duplicate # pairs (seats 6b060c1d and f86f29b0): the leaver was HRIS- # late-detected with their Google account already removed by # the previous month's manual cleanup, but Helix kept paging # IT about a "failed" revoke. With the idempotent-success # branch, that case now reads as "target state already met". def _lookup_old_user() -> dict[str, Any]: dir_svc = _build_directory_service( self._sa_json(), self._subject(), [_USER_SCOPE], ) return dir_svc.users().get(userKey=old_email).execute() try: old_user = await _call_with_transient_retry( _lookup_old_user, label="source_lookup", ) except HttpError as e: if e.resp.status == 404: # Source account already gone in Google — target state # (account dark + content nowhere to hand off) is # already met. Skip the transfer entirely and let # ``revoke_seat``'s subsequent ``_set_suspended`` also # short-circuit on its own 404 idempotent branch. logger.info( "google_workspace: source user %r already absent " "from directory — skipping data transfer (idempotent)", old_email, ) return ProvisioningResult( ok=True, external_ref="", raw_response={ "skipped": "source_user_already_gone", "source_email": old_email, }, ) return ProvisioningResult( ok=False, error_message=( f"Google source-user lookup failed " f"{e.resp.status}: {e.reason}" ), ) except Exception as e: # noqa: BLE001 return ProvisioningResult( ok=False, error_message=( f"Google source-user lookup: {type(e).__name__}: {e}" ), ) def _call() -> dict[str, Any]: # Recipient + source already validated above; build the # body inline and call only the transfer endpoint. new_user = recipient_user dt_svc = _build_datatransfer_service(self._sa_json(), self._subject()) body = { "oldOwnerUserId": old_user.get("id"), "newOwnerUserId": new_user.get("id"), "applicationDataTransfers": [ { "applicationId": _DATA_TRANSFER_APP_IDS["drive"], "applicationTransferParams": [ {"key": "PRIVACY_LEVEL", "value": ["PRIVATE", "SHARED"]}, {"key": "RELEASE_RESOURCES", "value": ["TRUE"]}, ], }, { "applicationId": _DATA_TRANSFER_APP_IDS["calendar"], "applicationTransferParams": [ {"key": "RELEASE_RESOURCES", "value": ["TRUE"]}, ], }, ], } return dt_svc.transfers().insert(body=body).execute() try: resp = await _call_with_transient_retry( _call, label="transfer_insert", ) except HttpError as e: # Round-1 audit HIGH #5: distinguish "Data Transfer scope # not delegated to this Service Account" (403) from generic # init failures. Hint pinpoints the recipient domain since # that's the usual culprit when only one of source/recipient # is locked. if e.resp.status == 403: return ProvisioningResult( ok=False, error_message=( f"[{GOOGLE_TAG_TRANSFER_PERMISSION_DENIED}] " f"Google Data Transfer init refused 403: {e.reason}. " f"Hint: the recipient domain ({new_email}) may " "lack data-transfer authorization, or the " "Service Account is missing " "admin.datatransfer scope." ), raw_response={ "error_tag": GOOGLE_TAG_TRANSFER_PERMISSION_DENIED, "subject_email": new_email, }, ) if e.resp.status == 409: # Idempotent adoption of an in-flight transfer. Google # allows only ONE concurrent data transfer per source # user and returns 409 "Data transfer already in progress # for the user" on a second insert. This fires on RETRY: # a first attempt kicked off the transfer, our 5-min poll # budget (revoke_seat) expired before Google finished the # move, the seat task failed with the "retry — the # in-flight transfer will be reused" hint, and the retry # re-enters here. Historically the 409 fell through to the # generic failure below and the retry could NEVER succeed # — the promised "reuse" was never implemented (prod: # sunyishen→cwen, task 52d1eba8, stuck 2 attempts). Locate # the existing transfer and hand its id back so revoke_seat # polls the SAME transfer to completion. adopted = await self._find_existing_transfer( old_owner_id=str(old_user.get("id") or ""), new_owner_id=str(recipient_user.get("id") or ""), ) if adopted is not None: logger.info( "google_workspace: adopting in-flight data " "transfer %r for source %r (409 already in " "progress) instead of failing the retry", adopted.get("id"), old_email, ) return ProvisioningResult( ok=True, external_ref=str(adopted.get("id") or ""), raw_response={ "transfer_id": adopted.get("id"), "overall_transfer_status": adopted.get( "overallTransferStatusCode" ), "transferred_to": new_email, "adopted_in_progress": True, }, ) # 409 but no adoptable (non-failed) transfer found — a # prior transfer may have FAILED server-side, or the list # is scoped differently than expected. Surface an # actionable message rather than the opaque 409 so the # operator checks the Admin Console. return ProvisioningResult( ok=False, error_message=( "Google Data Transfer init returned 409 " "(already in progress) but no adoptable in-flight " f"transfer was found for {old_email}. A prior " "transfer may have failed server-side; check " "Google Admin Console → Data → Data Migration " f"for {old_email} before retrying." ), raw_response={"source_email": old_email}, ) return ProvisioningResult( ok=False, error_message=( "Google Data Transfer init failed " f"{e.resp.status}: {e.reason}" ), ) except Exception as e: # noqa: BLE001 return ProvisioningResult( ok=False, error_message=f"Google Data Transfer init: {type(e).__name__}: {e}", ) return ProvisioningResult( ok=True, external_ref=str(resp.get("id") or ""), raw_response={ "transfer_id": resp.get("id"), "overall_transfer_status": resp.get("overallTransferStatusCode"), "transferred_to": new_email, }, ) async def _find_existing_transfer( self, *, old_owner_id: str, new_owner_id: str, ) -> dict[str, Any] | None: """Locate an existing (non-failed) data transfer for ``old_owner_id``. Used to make a 409 "already in progress" from ``transfers().insert()`` idempotent: return the in-flight (or already-completed) transfer so the caller can poll the SAME id to completion instead of failing the retry forever. Filters to transfers OUT of the leaver (``oldOwnerUserId``) and, when a recipient id is known, prefers the one whose ``newOwnerUserId`` matches so we adopt the transfer we ourselves kicked off rather than an unrelated one. A ``failed`` transfer is never adopted — that's a real failure the operator must see. Returns ``None`` when nothing adoptable is found (caller surfaces an actionable error). """ if not old_owner_id: return None def _list() -> dict[str, Any]: dt_svc = _build_datatransfer_service( self._sa_json(), self._subject(), ) return dt_svc.transfers().list( oldOwnerUserId=old_owner_id, maxResults=100, ).execute() try: resp = await _call_with_transient_retry( _list, label="transfer_list", ) except Exception as e: # noqa: BLE001 — best-effort adoption logger.warning( "google_workspace: transfers.list for in-flight " "adoption failed (source_id=%r): %s", old_owner_id, e, ) return None transfers = resp.get("dataTransfers") or [] # Prefer an in-flight/pending transfer to the intended recipient; # fall back to any non-failed transfer to that recipient; last # resort, any non-failed transfer out of the source user. def _status(t: dict[str, Any]) -> str: return str(t.get("overallTransferStatusCode") or "").lower() # ONLY adopt a transfer whose recipient matches the intended # successor. The previous ``or [any non-failed]`` fallback would, # on a 409 for a DIFFERENT successor (ops re-triggered revoke of the # same leaver picking successor B while the to-A transfer is still # in flight), silently adopt the to-A transfer, report # ``transferred_to: B``, suspend the leaver, and eventually # hard-delete them — with Drive/Calendar handed to the WRONG owner. # No match → return None so the caller surfaces the actionable 409 # instead of adopting an unrelated transfer. to_recipient = [ t for t in transfers if str(t.get("newOwnerUserId") or "") == new_owner_id and _status(t) != "failed" ] if not to_recipient: return None # Prefer still-running transfers over ones already completed. in_flight = [ t for t in to_recipient if _status(t) in ("inprogress", "pending") ] return (in_flight or to_recipient)[0] async def _poll_data_transfer( self, *, transfer_id: str, budget_seconds: int, interval_seconds: int, ) -> ProvisioningResult: """Poll the Admin Data Transfer API until ``overallTransferStatusCode`` reaches a terminal value (``completed``/``failed``) or the ``budget_seconds`` budget elapses. Returns ok=True on completed, ok=False on failed, ok=False with a timeout-flavoured ``error_message`` on budget expiry. Pre- launch P1 fix; the prior implementation only awaited the ``insert()`` call, so the suspend fired while the transfer was still in flight. """ if not transfer_id: return ProvisioningResult( ok=False, error_message="empty transfer_id; cannot poll", ) def _fetch() -> dict[str, Any]: dt_svc = _build_datatransfer_service(self._sa_json(), self._subject()) return dt_svc.transfers().get(dataTransferId=transfer_id).execute() deadline = asyncio.get_running_loop().time() + budget_seconds last_status: str | None = None transient_attempts = 0 while asyncio.get_running_loop().time() < deadline: try: resp = await asyncio.to_thread(_fetch) except HttpError as e: # Round-1 audit CRITICAL #1: distinguish permanent # (403 — scope/permission permanently denied) from # transient (429/5xx — retry within budget). Without # this, a single 429 burst from Google's overload # protection would fail the whole revoke even though # the transfer is succeeding server-side. if e.resp.status == 403: return ProvisioningResult( ok=False, error_message=( f"[{GOOGLE_TAG_TRANSFER_PERMISSION_DENIED}] " f"transfers.get refused 403: {e.reason}. " "Hint: Service Account may have lost " "admin.datatransfer scope or recipient " "domain delegation. Transfer " f"{transfer_id} may still complete " "server-side; check Google Admin Console " "→ Data → Data Migration." ), raw_response={ "status": last_status, "error_tag": GOOGLE_TAG_TRANSFER_PERMISSION_DENIED, "transfer_id": transfer_id, }, ) if e.resp.status in (429, 500, 502, 503, 504): transient_attempts += 1 # Cap transient retries so a wedged Google # datacenter doesn't exhaust the budget with # nothing-to-show. remaining = ( deadline - asyncio.get_running_loop().time() ) if remaining <= 0: break logger.warning( "Google transfers.get transient %s on %s " "(attempt %d); backing off %.1fs", e.resp.status, transfer_id, transient_attempts, _GOOGLE_POLL_TRANSIENT_BACKOFF, ) await asyncio.sleep( min(_GOOGLE_POLL_TRANSIENT_BACKOFF, remaining), ) continue return ProvisioningResult( ok=False, error_message=( f"transfers.get HTTP {e.resp.status}: {e.reason}" ), raw_response={"status": last_status}, ) except Exception as e: # noqa: BLE001 # Round-1 audit HIGH #3: transient exceptions (DNS # blip, Google overload, network reset) used to fail # the revoke immediately, truncating the budget. Treat # exceptions during polling as transient and retry # within budget — only fail fast on HttpError 4xx # above. transient_attempts += 1 remaining = deadline - asyncio.get_running_loop().time() if remaining <= 0 or transient_attempts >= 6: return ProvisioningResult( ok=False, error_message=( f"transfers.get {type(e).__name__}: {e} " f"(after {transient_attempts} transient " "attempts)" ), raw_response={"status": last_status}, ) logger.warning( "Google transfers.get exception %s on %s " "(attempt %d); backing off", type(e).__name__, transfer_id, transient_attempts, ) await asyncio.sleep( min(_GOOGLE_POLL_TRANSIENT_BACKOFF, remaining), ) continue status = str(resp.get("overallTransferStatusCode") or "").lower() last_status = status if status == "completed": return ProvisioningResult( ok=True, external_ref=transfer_id, raw_response={"status": "completed"}, ) if status == "failed": return ProvisioningResult( ok=False, error_message=( f"Google reported transfer {transfer_id} as failed" ), raw_response={"status": "failed"}, ) await asyncio.sleep(interval_seconds) # Round-1 audit MEDIUM #7: timeout now carries actionable # hint — the transfer may still complete server-side; re- # polling with the same id is the right next step. return ProvisioningResult( ok=False, error_message=( f"[{GOOGLE_TAG_TRANSFER_TIMEOUT}] transfer " f"{transfer_id} did not reach a terminal status " f"within {budget_seconds}s (last_status={last_status!r}). " "Hint: polling again will query the same transfer id; " "if it hangs >10min, check Google Admin Console → " "Data → Data Migration for the transfer state." ), raw_response={ "status": last_status, "error_tag": GOOGLE_TAG_TRANSFER_TIMEOUT, "transfer_id": transfer_id, }, ) async def _group_add(self, ctx: ProvisioningContext) -> ProvisioningResult: group = str((ctx.extra or {}).get("group_email") or "") if not group: return ProvisioningResult( ok=False, error_message="binding missing 'group_email'", ) def _call() -> dict[str, Any]: svc = _build_directory_service(self._sa_json(), self._subject(), [_GROUP_SCOPE]) return svc.members().insert( groupKey=group, body={"email": ctx.user_email, "role": "MEMBER"}, ).execute() try: # ``group_member`` is the most common binding op — wrap it in the # same transient-retry helper as every other write so a one-off # Google 429/5xx doesn't fail the grant + fire a false admin DM # (the noise the helper's docstring exists to kill). insert is # idempotent here — 409 already-member is handled below. resp = await _call_with_transient_retry(_call, label="group_add") except HttpError as e: if e.resp.status == 409: return ProvisioningResult( ok=True, external_ref=f"{group}:{ctx.user_email}", raw_response={"already_member": True}, ) # Round-1 audit MEDIUM #6: 403 = Service Account lacks the # admin.directory.group.member scope. Tag so admin gets # the scope-fix hint, not a raw "403: Forbidden". if e.resp.status == 403: return ProvisioningResult( ok=False, error_message=( f"[{GOOGLE_TAG_SCOPE_MISSING}] members.insert " f"refused 403: {e.reason}. Hint: Service " "Account may be missing " "admin.directory.group.member scope." ), raw_response={ "error_tag": GOOGLE_TAG_SCOPE_MISSING, "scope": "admin.directory.group.member", "group_email": group, }, ) return ProvisioningResult( ok=False, error_message=f"{e.resp.status}: {e.reason}", ) except Exception as e: # noqa: BLE001 return ProvisioningResult( ok=False, error_message=f"{type(e).__name__}: {e}", ) return ProvisioningResult( ok=True, external_ref=str(resp.get("id") or f"{group}:{ctx.user_email}"), raw_response=resp, ) async def _group_remove(self, ctx: ProvisioningContext) -> ProvisioningResult: group = str((ctx.extra or {}).get("group_email") or "") if not group: return ProvisioningResult( ok=False, error_message="binding missing 'group_email'", ) def _call() -> None: svc = _build_directory_service(self._sa_json(), self._subject(), [_GROUP_SCOPE]) svc.members().delete(groupKey=group, memberKey=ctx.user_email).execute() try: # Same transient-retry as _group_add — delete is idempotent here # (404 not-member is handled below), so a one-off 429/5xx must # not fail the revoke and page IT. await _call_with_transient_retry(_call, label="group_remove") except HttpError as e: if e.resp.status == 404: return ProvisioningResult( ok=True, external_ref=f"{group}:{ctx.user_email}", raw_response={"already_not_member": True}, ) # Round-1 audit MEDIUM #6: tag 403 scope errors symmetrically # with _group_add so admin sees the same actionable hint # regardless of whether onboarding or offboarding tripped it. if e.resp.status == 403: return ProvisioningResult( ok=False, error_message=( f"[{GOOGLE_TAG_SCOPE_MISSING}] members.delete " f"refused 403: {e.reason}. Hint: Service " "Account may be missing " "admin.directory.group.member scope." ), raw_response={ "error_tag": GOOGLE_TAG_SCOPE_MISSING, "scope": "admin.directory.group.member", "group_email": group, }, ) return ProvisioningResult( ok=False, error_message=f"{e.resp.status}: {e.reason}", ) except Exception as e: # noqa: BLE001 return ProvisioningResult( ok=False, error_message=f"{type(e).__name__}: {e}", ) return ProvisioningResult(ok=True, external_ref=f"{group}:{ctx.user_email}") async def _set_suspended( self, ctx: ProvisioningContext, *, suspended: bool, must_exist: bool = False, ) -> ProvisioningResult: """Flip a Google user's ``suspended`` flag. ``must_exist=True`` means the CALLER just created (or just read back) this account and is holding its identifier — so "not found" can only be Google's post-``users.insert`` settle window, never "the account is gone". Under that flag a 404 must NOT be reported as idempotent success. Why the flag exists (prod incident 2026-08-25, ``hiroki.shinagawa01@pingcap.com``): the create → welcome-email → safety-suspend rollback fires sub-second after ``users.insert``, because the welcome-email pre-flight (``personal_email empty``) fails locally with no network round-trip. Measured against the real tenant, the freshly-inserted address is not writable yet: ~0.7s ``users.get`` → 404 ~1.5s ``users.get`` → 200 but no ``suspended`` key at all 2–5s ``users.update`` → 412 ``User creation is not complete.`` ≥~6s ``users.update`` → 200, actually applies The pre-read below hit the 404 and returned ``already_absent`` = ``ok=True`` without ever sending the update, so the audit row recorded ``failure_rollback_suspend_ok: true`` while the mailbox stayed ACTIVE with a password nobody had received. That branch is still correct for the offboarding callers (the account really was hard-deleted out-of-band), which is exactly why this is a per-call-site flag and not a change of the shared 404 contract. """ # P1-H (2026-06-02 audit): normalize the email key at the # boundary so legacy ``identity_external_identities`` rows with # mixed-case ``provider_email`` (from pre-2026-04 manual SQL # imports / migrations) don't 404 on suspend. Google's # Directory API ``userKey`` accepts mixed case in 2024+ but # historically rejected case-flips and there's no reason to # depend on that. Lowercasing here covers EVERY downstream # caller (provision rollback, leaver suspend, revoke seat, # rename, group ops) in one place. if ctx.user_email and ctx.user_email != ctx.user_email.lower(): import dataclasses as _dc ctx = _dc.replace(ctx, user_email=ctx.user_email.lower()) # Pre-check current suspended state on the revoke direction # so an already-suspended account short-circuits to idempotent # success without firing the destructive ``update`` call. This # is a strict-superset of the post-update 404 "already_absent" # branch below: it also covers the case where the account # exists but is already locked. The check is best-effort — # any failure during pre-check falls through to the actual # update so we don't lose write capability if Google's read # endpoint is what's flaky. if suspended: def _read() -> dict[str, Any]: svc = _build_directory_service( self._sa_json(), self._subject(), [_USER_SCOPE], ) return svc.users().get(userKey=ctx.user_email).execute() try: cur = await _call_with_transient_retry( _read, label="suspend_pre_read", ) if cur.get("suspended") is True: logger.info( "google_workspace: user %s already suspended — " "skipping redundant update (idempotent)", ctx.user_email, ) return ProvisioningResult( ok=True, external_ref=f"gw-user:{ctx.user_email}", raw_response={ "suspended": True, "already_suspended": True, }, ) except HttpError as e: # 404 on the pre-read normally means the target state is # already met (account gone), so don't fall through to an # update that'll just 404 too. But when the caller KNOWS # the account exists (``must_exist``), a 404 here is the # post-insert settle window instead — fall through to the # update, which retries 412 until Google finishes making # the record writable. Reporting already_absent here is # what silently left an active mailbox in prod. if e.resp.status == 404 and not must_exist: return ProvisioningResult( ok=True, external_ref=f"gw-user:{ctx.user_email}", raw_response={"already_absent": True}, ) # 403 / transient-exhausted / 5xx-after-retries / other # 4xx — fall through to the actual update so a flaky # read endpoint doesn't block the write path. The write # path has its own transient-retry + error mapping; if # it ALSO fails we'll get the correct error attribution # there. ``_call_with_transient_retry`` re-raises the # final HttpError after exhausting its schedule, so a # persistent 502 from Google's API gateway also lands # here and falls through cleanly. logger.warning( "google_workspace: suspend pre-read for %s " "returned %s; falling through to update%s", ctx.user_email, e.resp.status, ( " (404 with must_exist — the caller holds this " "account's identifier, so this is Google's " "post-insert settle window, NOT a missing account)" if e.resp.status == 404 else "" ), ) except Exception as e: # noqa: BLE001 # Non-HttpError exceptions can come from: asyncio.to_thread # thread-pool exhaustion, JSON decode failures on a # malformed Google body, or the defensive RuntimeError # ``_call_with_transient_retry`` raises when its # schedule is empty. None are fatal for the SUSPEND- # direction goal — fall through to the actual write so # a degraded read endpoint can't strand the offboarding # pipeline. logger.warning( "google_workspace: suspend pre-read for %s " "raised %s; falling through to update", ctx.user_email, type(e).__name__, ) def _call() -> dict[str, Any]: svc = _build_directory_service(self._sa_json(), self._subject(), [_USER_SCOPE]) return svc.users().update( userKey=ctx.user_email, body={"suspended": suspended}, ).execute() try: resp = await _call_with_transient_retry( _call, label=( "user_suspend_update" if suspended else "user_unsuspend_update" ), # 412 ``User creation is not complete.`` is what Google # returns for the first ~5s after ``users.insert``. It is # transient by definition (the account is mid-creation), # but it is NOT in _GOOGLE_TRANSIENT_STATUSES, so without # this it failed the whole call on the first attempt. # Opted in per-call-site rather than globally: a 412 from # any OTHER Google surface is a real precondition failure # that should still surface immediately. extra_retry_statuses=frozenset({412}), ) except HttpError as e: # Idempotency on the revoke direction only: if the Google # account was deleted out-of-band (admin hard-deleted in the # Google Admin console between data-transfer poll and # suspend), the desired end-state ("account dark") is # already met. Treat 404 as already_absent so the # offboarding blocker closes — mirrors `_group_remove`'s # 404 contract above. We deliberately do NOT swallow 404 on # the unsuspend direction (suspended=False) because that # would silently flip a grant to "ok" when the target user # doesn't exist, masking a real provisioning bug. if suspended and e.resp.status == 404 and not must_exist: return ProvisioningResult( ok=True, external_ref=f"gw-user:{ctx.user_email}", raw_response={"already_absent": True}, ) if suspended and e.resp.status == 404: # ``must_exist``: we hold this account's identifier, so a # 404 is Google still settling the insert (or a wrong key) # — never "already gone". Fail loudly; the caller's audit # row must not claim the mailbox was locked. return ProvisioningResult( ok=False, error_message=( f"users.update 404 for {ctx.user_email} but the " "account is known to exist — Google has not " "finished creating it, or the userKey is wrong. " "Mailbox is NOT suspended." ), raw_response={ "not_found_but_must_exist": True, "target_user": ctx.user_email, }, ) # Round-1 audit HIGH #4: 403 means the Service Account # lost ``admin.directory.user`` scope (org policy change, # domain-wide delegation revoked). Tag it so admin gets a # specific "fix the Service Account delegation" hint # instead of a raw 403. if e.resp.status == 403: return ProvisioningResult( ok=False, error_message=( f"[{GOOGLE_TAG_SCOPE_MISSING}] users.update " f"refused 403: {e.reason}. Hint: Service " "Account may have lost the " "admin.directory.user scope or " "domain-wide delegation was revoked. Check " "Google Admin Console → Security → API " "controls → Domain-wide delegation." ), raw_response={ "error_tag": GOOGLE_TAG_SCOPE_MISSING, "scope": "admin.directory.user", "target_user": ctx.user_email, }, ) return ProvisioningResult( ok=False, error_message=f"{e.resp.status}: {e.reason}", ) except Exception as e: # noqa: BLE001 return ProvisioningResult( ok=False, error_message=f"{type(e).__name__}: {e}", ) # A 200 means "request accepted", NOT "state changed" — Google # echoes the resulting user record, and during the post-insert # settle window that record can come back with ``suspended`` # absent or still False. Reporting ok=True off the bare absence # of an exception is how a no-op got audited as a successful # safety-suspend. Gate ok on what Google actually echoed. observed = bool(resp.get("suspended")) if observed != suspended: logger.warning( "google_workspace: users.update returned 200 for %s but the " "echoed record says suspended=%r (wanted %r) — reporting " "FAILURE rather than a phantom success", ctx.user_email, resp.get("suspended"), suspended, ) return ProvisioningResult( ok=False, error_message=( f"users.update accepted (HTTP 200) but the returned " f"record still has suspended={resp.get('suspended')!r} " f"for {ctx.user_email} (wanted {suspended}) — the change " "did not take effect." ), raw_response={ "suspended": resp.get("suspended"), "requested_suspended": suspended, "state_not_applied": True, }, ) # Round-1 audit LOW #9: log destructive op for forensic trail. if suspended: logger.info( "google_workspace: suspended user %s", ctx.user_email, ) return ProvisioningResult( ok=True, external_ref=f"gw-user:{ctx.user_email}", raw_response={"suspended": resp.get("suspended")}, ) async def _delete_user( self, ctx: ProvisioningContext, ) -> ProvisioningResult: """Hard-DELETE a Google Workspace user — the terminal offboarding disposition (2026-07-08 ADR「离职号最终删除」). Fires only from the offboarding delete-sweep, ``delete_lead_days`` after leaver_date, for accounts we already suspended and that are NOT retention-exempt. Google keeps a deleted user recoverable for ~20 days (``users.undelete``), which — together with the sweep's 2-day suspend→ delete gap — is the safety net that lets this be automatic rather than a manual step. It does **no** data transfer (no successor): the account was dark since suspend; the offboarding blocker cascade already handled every owned resource / seat / group before ``completed``. Idempotent on the delete direction: a 404 (already deleted / never existed) is success (``already_absent=True``). A 403 is tagged ``GOOGLE_TAG_SCOPE_MISSING`` (the SA lost ``admin.directory.user`` or DWD was revoked) so IT gets an actionable hint instead of a raw 403. Never raises — mirrors ``_set_suspended`` so the sweep records a structured verdict instead of aborting the tick. """ # Normalise the email key at the boundary (see _set_suspended P1-H): # legacy mixed-case provider_email rows must not spuriously 404. if ctx.user_email and ctx.user_email != ctx.user_email.lower(): import dataclasses as _dc ctx = _dc.replace(ctx, user_email=ctx.user_email.lower()) def _call() -> dict[str, Any]: svc = _build_directory_service( self._sa_json(), self._subject(), [_USER_SCOPE], ) # users().delete() returns an empty body on success (204). return svc.users().delete(userKey=ctx.user_email).execute() or {} try: await _call_with_transient_retry(_call, label="user_delete") except HttpError as e: if e.resp.status == 404: # Already gone (deleted out-of-band or never existed) — the # desired end-state is met. Idempotent success. return ProvisioningResult( ok=True, external_ref=f"gw-user:{ctx.user_email}", raw_response={"already_absent": True}, ) if e.resp.status == 403: return ProvisioningResult( ok=False, error_message=( f"[{GOOGLE_TAG_SCOPE_MISSING}] users.delete " f"refused 403: {e.reason}. Hint: Service " "Account may have lost the " "admin.directory.user scope or " "domain-wide delegation was revoked. Check " "Google Admin Console → Security → API " "controls → Domain-wide delegation." ), raw_response={ "error_tag": GOOGLE_TAG_SCOPE_MISSING, "scope": "admin.directory.user", "target_user": ctx.user_email, }, ) return ProvisioningResult( ok=False, error_message=f"{e.resp.status}: {e.reason}", ) except Exception as e: # noqa: BLE001 return ProvisioningResult( ok=False, error_message=f"{type(e).__name__}: {e}", ) # Destructive op — forensic trail (mirrors the suspend log). logger.info("google_workspace: DELETED user %s", ctx.user_email) return ProvisioningResult( ok=True, external_ref=f"gw-user:{ctx.user_email}", raw_response={"deleted": True}, ) # ── Workspace ops (PR-D): mail group, alias, password ───────── # # These methods are NOT part of the SoftwareProvisioningDriver # contract — they're Google-specific self-service flows invoked # from the ``email_group`` domain. They all return # :class:`WorkspaceOpResult` (never raise), so the service layer # can surface a clean error to the user without try/except noise. async def create_group( self, *, email: str, name: str, description: str | None, ) -> WorkspaceOpResult: """Create a Google Group. Idempotent on 409 (already exists). The caller is expected to pre-check uniqueness against its own mirror table — this method's 409 path is a last-line defence for race conditions between two parallel creates. """ body = { "email": email, "name": name, "description": description or "", } def _call() -> dict[str, Any]: svc = _build_directory_service( self._sa_json(), self._subject(), _WORKSPACE_OPS_SCOPES, ) return svc.groups().insert(body=body).execute() try: resp = await asyncio.to_thread(_call) except HttpError as e: if e.resp.status == 409: return WorkspaceOpResult( ok=True, data={"email": email}, status=409, already_exists=True, ) return WorkspaceOpResult( ok=False, error_message=f"Google API error {e.resp.status}: {e.reason}", status=e.resp.status, ) except Exception as e: # noqa: BLE001 return WorkspaceOpResult( ok=False, error_message=f"{type(e).__name__}: {e}", ) return WorkspaceOpResult( ok=True, data={"email": resp.get("email", email), "id": resp.get("id")}, ) async def set_group_owner( self, *, group_email: str, member_email: str, ) -> WorkspaceOpResult: """Ensure ``member_email`` is an OWNER of ``group_email``. If already a MEMBER, promotes to OWNER. If not present, inserts as OWNER. Idempotent: calling with an existing owner is a no-op. """ def _call() -> dict[str, Any]: svc = _build_directory_service( self._sa_json(), self._subject(), _WORKSPACE_OPS_SCOPES, ) try: return svc.members().insert( groupKey=group_email, body={"email": member_email, "role": "OWNER"}, ).execute() except HttpError as e: if e.resp.status != 409: raise # Already a member — patch role to OWNER return svc.members().patch( groupKey=group_email, memberKey=member_email, body={"role": "OWNER"}, ).execute() try: resp = await asyncio.to_thread(_call) except HttpError as e: return WorkspaceOpResult( ok=False, error_message=f"Google API error {e.resp.status}: {e.reason}", status=e.resp.status, ) except Exception as e: # noqa: BLE001 return WorkspaceOpResult( ok=False, error_message=f"{type(e).__name__}: {e}", ) return WorkspaceOpResult( ok=True, data={"member_id": resp.get("id"), "role": resp.get("role")}, ) async def add_group_member( self, *, group_email: str, member_email: str, ) -> WorkspaceOpResult: """Ensure ``member_email`` is a MEMBER of ``group_email``. Inserts with role=MEMBER. Idempotent: an ``already a member`` 409 (any role — MEMBER/MANAGER/OWNER) is mapped to ``ok=True`` with ``already_member`` so a re-run (Feishu redelivery, retried approval) never errors and never DOWNGRADES an existing owner to member. Mirrors :meth:`set_group_owner` minus the role patch. """ def _call() -> dict[str, Any]: svc = _build_directory_service( self._sa_json(), self._subject(), _WORKSPACE_OPS_SCOPES, ) return svc.members().insert( groupKey=group_email, body={"email": member_email, "role": "MEMBER"}, ).execute() try: resp = await asyncio.to_thread(_call) except HttpError as e: if e.resp.status == 409: return WorkspaceOpResult( ok=True, data={"already_member": True}, status=409, ) return WorkspaceOpResult( ok=False, error_message=f"Google API error {e.resp.status}: {e.reason}", status=e.resp.status, ) except Exception as e: # noqa: BLE001 return WorkspaceOpResult( ok=False, error_message=f"{type(e).__name__}: {e}", ) return WorkspaceOpResult( ok=True, data={"member_id": resp.get("id"), "role": resp.get("role")}, ) async def set_group_member_role( self, *, group_email: str, member_email: str, role: str, ) -> WorkspaceOpResult: """Set an EXISTING member's role to ``MEMBER`` / ``MANAGER`` / ``OWNER`` via ``members.patch``. Distinct from :meth:`set_group_owner`, which inserts-or-promotes: this one only re-roles someone already in the group and 404s otherwise. That is the right shape for a demotion (owner → member), where an insert-on-missing fallback would silently ADD somebody the caller believed they were demoting. ``already_absent`` on 404 so the caller can report "they are not in this group" rather than a bare Google error. """ def _call() -> dict[str, Any]: svc = _build_directory_service( self._sa_json(), self._subject(), _WORKSPACE_OPS_SCOPES, ) out: dict[str, Any] = svc.members().patch( groupKey=group_email, memberKey=member_email, body={"role": role}, ).execute() return out try: resp = await asyncio.to_thread(_call) except HttpError as e: if e.resp.status == 404: return WorkspaceOpResult( ok=True, data={"already_absent": True}, status=404, ) return WorkspaceOpResult( ok=False, error_message=f"Google API error {e.resp.status}: {e.reason}", status=e.resp.status, ) except Exception as e: # noqa: BLE001 return WorkspaceOpResult( ok=False, error_message=f"{type(e).__name__}: {e}", ) return WorkspaceOpResult(ok=True, data={"role": resp.get("role")}) async def patch_group_directory_display( self, *, group_email: str, name: str | None = None, description: str | None = None, ) -> WorkspaceOpResult: """Update a group's name / description via the **Directory API**. Deliberately NOT :meth:`patch_group_display`, which goes through Cloud Identity and therefore needs the separately-delegated ``cloud-identity.groups`` WRITE scope. Directory ``groups.update`` runs on ``admin.directory.group`` — already in ``_WORKSPACE_OPS_SCOPES`` and already delegated, since group create / delete / alias all use it. So editing a static group's display metadata carries no new ops prerequisite. Cloud Identity remains the right API for *dynamic* groups (its patch is what carries ``dynamicGroupMetadata``); this method is for the hand-maintained ones. """ body: dict[str, Any] = {} if name is not None: body["name"] = name if description is not None: # Google stores "no description" as an empty string, so an # explicit "" is a meaningful clear rather than a no-op. body["description"] = description if not body: return WorkspaceOpResult(ok=True, data={"noop": True}) def _call() -> dict[str, Any]: svc = _build_directory_service( self._sa_json(), self._subject(), _WORKSPACE_OPS_SCOPES, ) out: dict[str, Any] = svc.groups().patch( groupKey=group_email, body=body, ).execute() return out try: resp = await asyncio.to_thread(_call) except HttpError as e: return WorkspaceOpResult( ok=False, error_message=f"Google API error {e.resp.status}: {e.reason}", status=e.resp.status, ) except Exception as e: # noqa: BLE001 return WorkspaceOpResult( ok=False, error_message=f"{type(e).__name__}: {e}", ) return WorkspaceOpResult( ok=True, data={"name": resp.get("name"), "description": resp.get("description")}, ) async def remove_group_owner( self, *, group_email: str, member_email: str, ) -> WorkspaceOpResult: """Remove a member (typically an owner being replaced) from a group. Idempotent on 404. NOTE: Google always requires at least one owner per group, so the caller must ensure a new owner is in place before calling this for the LAST owner. """ def _call() -> None: svc = _build_directory_service( self._sa_json(), self._subject(), _WORKSPACE_OPS_SCOPES, ) svc.members().delete( groupKey=group_email, memberKey=member_email, ).execute() try: await asyncio.to_thread(_call) except HttpError as e: if e.resp.status == 404: return WorkspaceOpResult( ok=True, data={"already_absent": True}, status=404, ) return WorkspaceOpResult( ok=False, error_message=f"Google API error {e.resp.status}: {e.reason}", status=e.resp.status, ) except Exception as e: # noqa: BLE001 return WorkspaceOpResult( ok=False, error_message=f"{type(e).__name__}: {e}", ) return WorkspaceOpResult(ok=True) async def group_exists(self, *, email: str) -> bool | None: """Does this group exist on Google right now? ``True`` / ``False`` are determinate; ``None`` means the question could not be answered (403, 5xx, transport) and the caller must NOT treat that as "gone". Exists so the group sync can confirm an absence before tombstoning. ``list_all_groups`` completes before the DB snapshot is taken, so a group created in that window looks absent — and a tombstone plus an audit row asserting "absent from Google" would both be false. A direct ``groups.get`` is the only answer that does not depend on clock granularity or on comparing a naive DB timestamp with an aware one. Runs on ``_GROUP_READ_SCOPES``, not the ops scopes. This is a pure GET called from the SYNC path, which is exactly the path a readonly-delegated tenant runs: requesting ``admin.directory.group`` (write) there fails at token exchange, this method would answer ``None`` forever, and the tombstone it gates would silently never happen while the cron kept reporting ``ok``. ``group.readonly`` is sufficient for ``groups.get``. """ def _call() -> dict[str, Any]: svc = _build_directory_service( self._sa_json(), self._subject(), _GROUP_READ_SCOPES, ) out: dict[str, Any] = svc.groups().get(groupKey=email).execute() return out try: await asyncio.to_thread(_call) except HttpError as e: if e.resp.status == 404: return False logger.warning( "group_exists(%s) indeterminate: HTTP %s %s", email, e.resp.status, e.reason, ) return None except Exception: logger.exception("group_exists(%s) indeterminate", email) return None return True async def delete_group(self, *, email: str) -> WorkspaceOpResult: """Delete a Google Group. Idempotent on 404.""" def _call() -> None: svc = _build_directory_service( self._sa_json(), self._subject(), _WORKSPACE_OPS_SCOPES, ) svc.groups().delete(groupKey=email).execute() try: await asyncio.to_thread(_call) except HttpError as e: if e.resp.status == 404: return WorkspaceOpResult( ok=True, data={"already_absent": True}, status=404, ) return WorkspaceOpResult( ok=False, error_message=f"Google API error {e.resp.status}: {e.reason}", status=e.resp.status, ) except Exception as e: # noqa: BLE001 return WorkspaceOpResult( ok=False, error_message=f"{type(e).__name__}: {e}", ) return WorkspaceOpResult(ok=True) async def add_alias( self, *, target_email: str, alias_email: str, target_kind: str, ) -> WorkspaceOpResult: """Add an alias to a user mailbox (``target_kind='user'``) or a Google Group (``target_kind='group'``). Google surfaces the operation on different resources depending on the target: users().aliases().insert() vs. groups().aliases().insert(). Both return the alias body on success and raise 409 on duplicate. """ if target_kind not in ("user", "group"): return WorkspaceOpResult( ok=False, error_message=f"unknown target_kind {target_kind!r} (expected user|group)", ) body = {"alias": alias_email} def _call() -> dict[str, Any]: svc = _build_directory_service( self._sa_json(), self._subject(), _WORKSPACE_OPS_SCOPES, ) if target_kind == "user": return svc.users().aliases().insert( userKey=target_email, body=body, ).execute() return svc.groups().aliases().insert( groupKey=target_email, body=body, ).execute() try: resp = await asyncio.to_thread(_call) except HttpError as e: if e.resp.status == 409: return WorkspaceOpResult( ok=True, data={"alias": alias_email}, status=409, already_exists=True, ) return WorkspaceOpResult( ok=False, error_message=f"Google API error {e.resp.status}: {e.reason}", status=e.resp.status, ) except Exception as e: # noqa: BLE001 return WorkspaceOpResult( ok=False, error_message=f"{type(e).__name__}: {e}", ) return WorkspaceOpResult( ok=True, data={"alias": resp.get("alias", alias_email)}, ) async def remove_alias( self, *, target_email: str, alias_email: str, target_kind: str, ) -> WorkspaceOpResult: """Remove an alias. Idempotent on 404.""" if target_kind not in ("user", "group"): return WorkspaceOpResult( ok=False, error_message=f"unknown target_kind {target_kind!r} (expected user|group)", ) def _call() -> None: svc = _build_directory_service( self._sa_json(), self._subject(), _WORKSPACE_OPS_SCOPES, ) if target_kind == "user": svc.users().aliases().delete( userKey=target_email, alias=alias_email, ).execute() else: svc.groups().aliases().delete( groupKey=target_email, alias=alias_email, ).execute() try: await asyncio.to_thread(_call) except HttpError as e: if e.resp.status == 404: return WorkspaceOpResult( ok=True, data={"already_absent": True}, status=404, ) return WorkspaceOpResult( ok=False, error_message=f"Google API error {e.resp.status}: {e.reason}", status=e.resp.status, ) except Exception as e: # noqa: BLE001 return WorkspaceOpResult( ok=False, error_message=f"{type(e).__name__}: {e}", ) return WorkspaceOpResult(ok=True) async def reset_user_password( self, *, user_email: str, new_password: str, change_password_at_next_login: bool = False, ) -> WorkspaceOpResult: """Reset a user's password to the caller-supplied ``new_password``. ``new_password`` is REQUIRED. The driver deliberately strips the plaintext from its return payload (so it never lands in a persisted ``provisioning_tasks.raw_response``), which means the CALLER must already hold the value to deliver it out-of-band. The old "generate internally if None" branch produced a silent account lockout — it minted a password, pushed it to Google, then discarded it with nobody able to sign in. Every caller already generates + passes the password (account_provisioner / email_group), so require it and fail closed on a missing value rather than lock the user out. Google's ``users.update`` endpoint accepts a plaintext ``password`` field (Google hashes it server-side). ``changePasswordAtNextLogin`` defaults to UNSET — the Feishu leaver-reset flow does NOT force next-login rotation. Pass ``change_password_at_next_login=True`` for the new-hire re-provision path (temporary welcome password). """ if not new_password: return WorkspaceOpResult( ok=False, error_message=( "reset_user_password requires new_password — the caller " "must generate it and deliver it out-of-band (the driver " "never returns the plaintext). Refusing to set a password " "nobody can retrieve." ), ) def _call() -> dict[str, Any]: svc = _build_directory_service( self._sa_json(), self._subject(), [_USER_SCOPE], ) body: dict[str, Any] = {"password": new_password} if change_password_at_next_login: body["changePasswordAtNextLogin"] = True return svc.users().update( userKey=user_email, body=body, ).execute() try: resp = await asyncio.to_thread(_call) except HttpError as e: return WorkspaceOpResult( ok=False, error_message=f"Google API error {e.resp.status}: {e.reason}", status=e.resp.status, ) except Exception as e: # noqa: BLE001 return WorkspaceOpResult( ok=False, error_message=f"{type(e).__name__}: {e}", ) # Do NOT include new_password in the data dict — if callers # persist data (e.g. provisioning_tasks.raw_response) the plaintext # password would be stored in the DB. The caller already received # the password via the return value of generate_secure_password() or # via a separate channel; this result is only used to confirm success. return WorkspaceOpResult( ok=True, data={ "password_reset": True, "primary_email": resp.get("primaryEmail", user_email), }, ) async def list_saml_login_events( self, *, user_email: str, start_time: datetime, end_time: datetime | None = None, max_results: int = 500, ) -> list[dict[str, Any]]: """SAML ``login_success`` events for one user from Admin Reports. ``activities.list(userKey=, applicationName='saml', eventName='login_success', startTime=…)``. Each returned dict carries ``time`` (aware datetime), ``application_name`` (the SAML app's name as configured in Google admin) and ``raw`` (the event's parameter list). Pages through ``nextPageToken`` up to ``max_results`` — a single user's SAML logins over a few months are hundreds at most. Error mapping (callers branch on ``GoogleApiError.reason``): * 403 → ``reason='google_reports_scope_missing'`` — the DwD service account has not been granted ``admin.reports.audit.readonly``. Not transient; the caller records it and moves on. * 400 / 404 whose body names ``userKey`` / ``Invalid Input`` / ``not found`` → ``reason='user_not_found'`` — the account no longer exists at Google (deleted leaver). That is *evidence* for the SSO verifier (an absent account cannot log in anywhere), so it is a distinct reason rather than a failure. * 429 / 5xx → retried via ``_call_with_transient_retry``; on exhaustion ``reason='transient'``. """ params: dict[str, Any] = { "userKey": user_email, "applicationName": "saml", "eventName": "login_success", "startTime": _rfc3339(start_time), "maxResults": min(max(int(max_results), 1), 1000), } if end_time is not None: params["endTime"] = _rfc3339(end_time) def _page(page_token: str | None) -> dict[str, Any]: svc = _build_reports_service(self._sa_json(), self._subject()) req_params = dict(params) if page_token: req_params["pageToken"] = page_token resp: dict[str, Any] = svc.activities().list(**req_params).execute() return resp events: list[dict[str, Any]] = [] token: str | None = None try: while True: resp = await _call_with_transient_retry( lambda tok=token: _page(tok), label="list_saml_login_events", ) for item in resp.get("items") or []: when = _parse_rfc3339((item.get("id") or {}).get("time")) for ev in item.get("events") or []: app_name: str | None = None for p in ev.get("parameters") or []: if p.get("name") == "application_name": app_name = p.get("value") break events.append({ "time": when, "application_name": app_name, "event_name": ev.get("name"), "raw": ev.get("parameters") or [], }) if len(events) >= max_results: return events token = resp.get("nextPageToken") if not token: return events except HttpError as e: status = e.resp.status if getattr(e, "resp", None) else None body = "" try: body = (e.content or b"").decode("utf-8", "replace") except Exception: # noqa: BLE001 body = "" lowered = body.lower() if status == 403: raise GoogleApiError( "list_saml_login_events", status=403, reason="google_reports_scope_missing", hint=( "grant the DwD service account " "https://www.googleapis.com/auth/admin.reports.audit.readonly" ), ) from e if status in (400, 404) and ( "userkey" in lowered or "invalid input" in lowered or "not found" in lowered ): raise GoogleApiError( "list_saml_login_events", status=status, reason="user_not_found", ) from e if status in _GOOGLE_TRANSIENT_STATUSES: raise GoogleApiError( "list_saml_login_events", status=status, reason="transient", ) from e # ``HttpError.reason`` is httplib2's HTTP reason phrase ("Ok" on a # synthetic response) — not a diagnosis. Callers branch on the # status-shaped token; the body is in the exception chain. raise GoogleApiError( "list_saml_login_events", status=status, reason=f"http_{status}", ) from e async def force_logout_user(self, *, user_email: str) -> WorkspaceOpResult: """Sign the user out of every active Google Workspace session. Wraps ``users.signOut`` (Directory API). Used by the Leaver Blocked runtime layer the moment HR confirms an offboarding — existing browser sessions stop authenticating within the Google session-cookie refresh interval (~ minutes). Pairs with the JWT-TTL cap on Helix-side tokens. Idempotent: calling on an already-logged-out user returns ok=True. 404 (user not in workspace) is surfaced as a soft ``already_absent=True`` so offboarding callers can treat it the same as success. """ def _call() -> dict[str, Any]: # ``signOut`` needs the session-security scope, NOT the broad # user-CRUD scope — see ``_USER_SECURITY_SCOPE``. Request both # so the build is robust regardless of which the tenant maps. svc = _build_directory_service( self._sa_json(), self._subject(), [_USER_SCOPE, _USER_SECURITY_SCOPE], ) # signOut returns no body on success svc.users().signOut(userKey=user_email).execute() return {} try: await asyncio.to_thread(_call) except HttpError as e: if e.resp.status == 404: return WorkspaceOpResult( ok=True, already_absent=True, status=404, error_message=( f"Google reports user {user_email} not found " f"(treated as already-logged-out)" ), ) return WorkspaceOpResult( ok=False, error_message=f"Google API error {e.resp.status}: {e.reason}", status=e.resp.status, ) except Exception as e: # noqa: BLE001 return WorkspaceOpResult( ok=False, error_message=f"{type(e).__name__}: {e}", ) return WorkspaceOpResult(ok=True, data={"primary_email": user_email}) # ─── Department sync (PR 2026-04-22) ────────────────────────── # # Google Workspace has a per-tenant custom-schema slot for # department hierarchy. Other internal tools already populate it, # so when Helix syncs dept names into Google we must read the # *existing* schema (don't invent a new one, don't pick a wrong # case). ``list_custom_schemas`` enumerates them; the caller # stores the chosen schema in ``platform_settings`` and the # scheduled job reads it back. async def list_custom_schemas(self) -> list[dict[str, Any]]: """List custom user schemas for the tenant. Returns the raw ``fields`` array so the admin UI can show schema_id + field keys and pick the right one without guesswork. Needs admin.directory.userschema (read-only) but userschema falls under the user scope in most deployments. """ def _call() -> dict[str, Any]: svc = _build_directory_service( self._sa_json(), self._subject(), ["https://www.googleapis.com/auth/admin.directory.userschema.readonly"], ) return svc.schemas().list(customerId="my_customer").execute() # Previously swallowed every error and returned ``[]``. That made # "tenant has no schemas" indistinguishable from "403 — SA missing # admin.directory.userschema.readonly". The dept_sync handler then # emitted a misleading "schema not found (available: [])" which # sent ops chasing the wrong config. Raise ``GoogleApiError`` so # the caller can surface the real reason. try: resp = await asyncio.to_thread(_call) return list(resp.get("schemas") or []) except HttpError as e: logger.warning( "list_custom_schemas HttpError %s: %s", e.resp.status, e.reason, ) hint = None if e.resp.status == 403: hint = ( "Service Account likely missing " "admin.directory.userschema.readonly in the " "domain-wide-delegation allow-list" ) raise GoogleApiError( "list_custom_schemas", status=e.resp.status, reason=e.reason, hint=hint, ) from e except Exception as e: logger.exception("list_custom_schemas failed") raise GoogleApiError( "list_custom_schemas", reason=str(e), ) from e async def get_user_profile(self, user_email: str) -> dict[str, Any] | None: """Fetch a user with ``projection=full`` so any populated customSchemas fields come back. Returns None if the user doesn't exist in Google. Used by the dept-sync job to skip users who aren't yet in Google (hire not synced, or contractor with no account) and to diff the current dept payload so we don't PATCH no-ops. """ def _call() -> dict[str, Any]: svc = _build_directory_service(self._sa_json(), self._subject(), [_USER_SCOPE]) return svc.users().get(userKey=user_email, projection="full").execute() try: return await asyncio.to_thread(_call) except HttpError as e: if e.resp.status == 404: return None # Non-404 errors (403 forbidden, 500 server error, etc.) are # re-raised so callers see the real failure rather than getting # a silent None that looks identical to "user doesn't exist". raise except Exception: logger.exception("get_user_profile %s failed", user_email) raise async def patch_user_department( self, *, user_email: str, schema_id: str, fields: dict[str, str | None], ) -> WorkspaceOpResult: """PATCH ``customSchemas.`` for one user. Fields with ``None`` value are cleared (we send them as JSON ``null`` which Google interprets as "delete"); fields with a string are upserted. The caller decides which keys to send — empty L3/L4 for a 2-level dept, populated for 3+. Uses ``users.patch`` (not update) so only the customSchemas slot is touched — never the rest of the user record. """ body = {"customSchemas": {schema_id: fields}} def _call() -> dict[str, Any]: svc = _build_directory_service(self._sa_json(), self._subject(), [_USER_SCOPE]) return svc.users().patch(userKey=user_email, body=body).execute() try: resp = await asyncio.to_thread(_call) except HttpError as e: return WorkspaceOpResult( ok=False, status=e.resp.status, error_message=f"Google API error {e.resp.status}: {e.reason}", ) except Exception as e: # noqa: BLE001 return WorkspaceOpResult( ok=False, error_message=f"{type(e).__name__}: {e}", ) return WorkspaceOpResult( ok=True, data={"primary_email": resp.get("primaryEmail", user_email)}, ) # ─── Groups enumeration (PR 2026-04-22) ─────────────────────── # # Used by the mail-group sync job to mirror every Google Group # into ``email_groups``. Pages through all groups in the domain # at 200-per-page; caller handles retry/backoff. async def list_all_groups(self, domain: str | None = None) -> list[dict[str, Any]]: """Page through every group in the Workspace tenant. ``domain`` optional — omit to let Google return all domains the caller is allowed to see. Returns raw resource dicts (email, id, name, description, directMembersCount, …). """ groups: list[dict[str, Any]] = [] page_token: str | None = None def _call(token: str | None) -> dict[str, Any]: svc = _build_directory_service( self._sa_json(), self._subject(), _GROUP_READ_SCOPES, ) kwargs: dict[str, Any] = {"maxResults": 200} if domain: kwargs["domain"] = domain else: kwargs["customer"] = "my_customer" if token: kwargs["pageToken"] = token return svc.groups().list(**kwargs).execute() # Fail-stop on ANY page failure (not just page 1) — symmetric # with ``list_all_users`` / ``list_group_members``. The original # implementation broke on a mid-stream error and returned the # truncated list. ``run_google_group_sync`` then iterated # ``existing_by_email`` and soft-deleted every group whose # email wasn't in ``seen_emails`` — so a transient 503 on # page 2 silently wiped every group past the failure point. # Unlike the user-soft-delete path, the group-level soft-delete # has no ratio-floor guard. Make truncation impossible at the # source. for _page in range(_MAX_PAGINATOR_PAGES): try: resp = await asyncio.to_thread(_call, page_token) except HttpError as e: logger.warning( "list_all_groups HttpError %s: %s (collected so far: %d)", e.resp.status, e.reason, len(groups), ) hint = None if e.resp.status == 403: hint = ( "Service Account likely missing " "admin.directory.group.readonly in the " "domain-wide-delegation allow-list" ) elif groups: hint = ( f"Mid-stream failure after {len(groups)} groups " "— treating as transient and aborting the call " "so the group-sync handler does not interpret a " "truncated list as 'these groups were deleted'." ) raise GoogleApiError( "list_all_groups", status=e.resp.status, reason=e.reason, hint=hint, ) from e except Exception as e: logger.exception("list_all_groups failed") raise GoogleApiError( "list_all_groups", reason=str(e), ) from e batch = resp.get("groups") or [] groups.extend(batch) page_token = resp.get("nextPageToken") if not page_token: break else: raise GoogleApiError( "list_all_groups", reason=( f"paginator did not terminate after " f"{_MAX_PAGINATOR_PAGES} pages" ), ) return groups async def list_groups_for_user( self, user_email: str, *, raise_on_missing: bool = False, ) -> list[dict[str, Any]]: """Page through every group ``user_email`` is a member of. Uses ``groups.list(userKey=…)`` — Google returns the group resources the user belongs to (membership), NOT their per-group role. The caller derives owner/member from the locally-synced ``email_group_owners`` table rather than an extra ``members.get`` per group. ``raise_on_missing=True`` turns the 404 back into a :class:`GoogleApiError`. The default swallow-to-``[]`` is right for the "我的邮件组" tab (a contractor with no Workspace account is in zero groups) but WRONG wherever ``[]`` will be read as a determinate "not a member of anything" — the cleanup workbench uses this with a GROUP address, where a 404 means the address is stale, not that the group is nested nowhere, and the difference decides an irreversible hard delete. Returns ``[]`` when the user isn't in Google's directory (404) so the "我的邮件组" tab renders an empty list rather than erroring for contractors with no Workspace account. Any other failure (403 scope, 5xx) raises :class:`GoogleApiError` so the caller surfaces the real problem instead of a misleading "you're in zero groups". """ groups: list[dict[str, Any]] = [] page_token: str | None = None def _call(token: str | None) -> dict[str, Any]: svc = _build_directory_service( self._sa_json(), self._subject(), _GROUP_READ_SCOPES, ) kwargs: dict[str, Any] = { "userKey": user_email, "maxResults": 200, } if token: kwargs["pageToken"] = token return svc.groups().list(**kwargs).execute() for _page in range(_MAX_PAGINATOR_PAGES): try: resp = await asyncio.to_thread(_call, page_token) except HttpError as e: if e.resp.status == 404 and not raise_on_missing: # User not in the directory — no memberships. return [] hint = None if e.resp.status == 404: hint = ( "the address itself is not in the directory — " "this is NOT the same as 'belongs to no group'" ) if e.resp.status == 403: hint = ( "Service Account likely missing " "admin.directory.group.readonly in the " "domain-wide-delegation allow-list" ) raise GoogleApiError( "list_groups_for_user", status=e.resp.status, reason=e.reason, hint=hint, ) from e except Exception as e: logger.exception("list_groups_for_user %s failed", user_email) raise GoogleApiError( "list_groups_for_user", reason=str(e), ) from e groups.extend(resp.get("groups") or []) page_token = resp.get("nextPageToken") if not page_token: break else: raise GoogleApiError( "list_groups_for_user", reason=( f"paginator did not terminate after " f"{_MAX_PAGINATOR_PAGES} pages for user {user_email!r}" ), ) return groups async def list_dynamic_groups(self) -> list[dict[str, Any]] | None: """Enumerate every Cloud Identity **dynamic group** with its membership query, display name and description, or ``None`` if the lookup couldn't be performed. WHY ``groups.search`` and not ``groups.list``: Cloud Identity ``groups.list`` does **not** populate ``dynamicGroupMetadata`` — not even with ``view=FULL`` (verified against prod 2026-07-22: list returned all 765 groups but every ``dynamicGroupMetadata`` was empty). The only reliable enumeration is ``groups.search`` filtered on the ``groups.dynamic`` label, which returns the resource names; we then ``groups.get`` each one to read its query (search itself returns BASIC view, so the metadata still needs the per-group get). The ``parent`` filter requires the concrete ``C0xxxx`` customer id — the ``my_customer`` alias is rejected here — so we resolve it first via :meth:`_resolve_customer_id`. Return contract mirrors :meth:`list_group_owners`: an empty list legitimately means "no dynamic groups", ``None`` means "couldn't read" (SA missing ``cloud-identity.groups.readonly`` in DWD, or Cloud Identity not enabled). Callers MUST treat ``None`` as indeterminate. Each returned dict: ``{email, name, display_name, description, query, resource_name, labels}``. """ try: customer_id = await self._resolve_customer_id() except Exception: logger.exception("list_dynamic_groups: customer id resolve failed") return None # ── Phase 1: search by the dynamic label → resource names ── found: list[dict[str, str]] = [] # {name, email} page_token: str | None = None def _search(token: str | None) -> dict[str, Any]: svc = _build_cloudidentity_service( self._sa_json(), self._subject(), [_CLOUD_IDENTITY_GROUPS_RO_SCOPE], ) kwargs: dict[str, Any] = { "query": ( f"parent=='customers/{customer_id}' && " f"'{_CLOUD_IDENTITY_DYNAMIC_LABEL}' in labels" ), "pageSize": 100, } if token: kwargs["pageToken"] = token return svc.groups().search(**kwargs).execute() for _page in range(_MAX_PAGINATOR_PAGES): try: resp = await asyncio.to_thread(_search, page_token) except HttpError as e: logger.warning( "list_dynamic_groups search HTTP %s: %s — returning " "None (check cloud-identity.groups.readonly DWD scope)", e.resp.status, e.reason, ) return None except Exception: logger.exception("list_dynamic_groups search failed") return None for grp in resp.get("groups") or []: name = grp.get("name") email = ((grp.get("groupKey") or {}).get("id") or "").strip().lower() if name and email: found.append({"name": name, "email": email}) page_token = resp.get("nextPageToken") if not page_token: break else: logger.warning("list_dynamic_groups: search paginator overran") return None # ── Phase 2: per-group get → membership query + metadata ── def _get(res_name: str) -> dict[str, Any]: svc = _build_cloudidentity_service( self._sa_json(), self._subject(), [_CLOUD_IDENTITY_GROUPS_RO_SCOPE], ) return svc.groups().get(name=res_name).execute() out: list[dict[str, Any]] = [] for item in found: try: g = await asyncio.to_thread(_get, item["name"]) except Exception: logger.exception( "list_dynamic_groups get failed for %s", item["email"], ) return None meta = g.get("dynamicGroupMetadata") or {} queries = [ (q.get("query") or "").strip() for q in (meta.get("queries") or []) if (q.get("query") or "").strip() ] out.append( { "email": item["email"], "resource_name": item["name"], "name": g.get("displayName") or "", "display_name": g.get("displayName") or "", "description": g.get("description") or "", "query": queries[0] if queries else "", "queries": queries, "labels": sorted((g.get("labels") or {}).keys()), } ) return out async def list_dynamic_group_emails(self) -> set[str] | None: """Return just the set of dynamic-group emails (or ``None`` if indeterminate). Thin wrapper over :meth:`list_dynamic_groups` preserved for ``google.group_sync``'s ``is_dynamic`` stamping. """ groups = await self.list_dynamic_groups() if groups is None: return None return {g["email"] for g in groups if g.get("email")} async def get_dynamic_group( self, *, group_email: str, ) -> dict[str, Any] | None: """Fetch a single dynamic group's metadata by email, or ``None`` if it isn't a dynamic group / couldn't be read. Resolves the resource name via ``groups.lookup`` then ``groups.get``. """ def _call() -> dict[str, Any]: svc = _build_cloudidentity_service( self._sa_json(), self._subject(), [_CLOUD_IDENTITY_GROUPS_RO_SCOPE], ) looked = svc.groups().lookup( groupKey_id=group_email, ).execute() res_name = looked.get("name") if not res_name: return {} return svc.groups().get(name=res_name).execute() try: g = await asyncio.to_thread(_call) except HttpError as e: if e.resp.status == 404: return None logger.warning( "get_dynamic_group %s HTTP %s: %s", group_email, e.resp.status, e.reason, ) return None except Exception: logger.exception("get_dynamic_group %s failed", group_email) return None if not g: return None meta = g.get("dynamicGroupMetadata") or {} queries = [ (q.get("query") or "").strip() for q in (meta.get("queries") or []) if (q.get("query") or "").strip() ] return { "email": group_email, "resource_name": g.get("name"), "display_name": g.get("displayName") or "", "description": g.get("description") or "", "query": queries[0] if queries else "", "queries": queries, "labels": sorted((g.get("labels") or {}).keys()), "is_dynamic": bool(meta), } async def create_dynamic_group( self, *, email: str, name: str, description: str | None, query: str, ) -> WorkspaceOpResult: """Create a Cloud Identity **dynamic group** whose membership is the CEL ``query`` (resourceType USER). Idempotent on 409. Requires the ``cloud-identity.groups`` write scope in DWD. The ``initialGroupConfig=EMPTY`` query param is required by Google for programmatic group creation. """ customer_id = await self._resolve_customer_id() body = { "parent": f"customers/{customer_id}", "groupKey": {"id": email}, "displayName": name, "description": description or "", "labels": {_CLOUD_IDENTITY_DISCUSSION_LABEL: ""}, "dynamicGroupMetadata": { "queries": [{"resourceType": "USER", "query": query}], }, } def _call() -> dict[str, Any]: svc = _build_cloudidentity_service( self._sa_json(), self._subject(), [_CLOUD_IDENTITY_GROUPS_SCOPE], ) return svc.groups().create( body=body, initialGroupConfig="EMPTY", ).execute() try: resp = await asyncio.to_thread(_call) except HttpError as e: if e.resp.status == 409: return WorkspaceOpResult( ok=True, data={"email": email}, status=409, already_exists=True, ) return WorkspaceOpResult( ok=False, error_message=f"Google API error {e.resp.status}: {e.reason}", status=e.resp.status, ) except Exception as e: # noqa: BLE001 return WorkspaceOpResult( ok=False, error_message=f"{type(e).__name__}: {e}", ) # create returns a long-running Operation; the group name is in # ``response.name`` once done, but Cloud Identity group creation # completes synchronously in practice — surface whatever came back. return WorkspaceOpResult( ok=True, data={"email": email, "operation": resp.get("name")}, ) async def patch_dynamic_group_query( self, *, group_email: str, query: str, ) -> WorkspaceOpResult: """Replace the membership query of an existing dynamic group. Uses Cloud Identity ``groups.patch`` with ``updateMask=dynamicGroupMetadata`` — this rewrites the CEL rule; Google recomputes membership asynchronously. (The public REST reference page understates the allowed updateMask, but ``dynamicGroupMetadata`` patch is supported and used by GAM and the google-api-python-client — verified 2026-07-22.) """ def _call() -> dict[str, Any]: svc = _build_cloudidentity_service( self._sa_json(), self._subject(), [_CLOUD_IDENTITY_GROUPS_SCOPE], ) looked = svc.groups().lookup(groupKey_id=group_email).execute() res_name = looked.get("name") if not res_name: raise GoogleApiError( "patch_dynamic_group_query", reason=f"group {group_email} not found", ) return svc.groups().patch( name=res_name, updateMask="dynamicGroupMetadata", body={ "dynamicGroupMetadata": { "queries": [ {"resourceType": "USER", "query": query}, ], }, }, ).execute() try: resp = await asyncio.to_thread(_call) except HttpError as e: return WorkspaceOpResult( ok=False, error_message=f"Google API error {e.resp.status}: {e.reason}", status=e.resp.status, ) except Exception as e: # noqa: BLE001 return WorkspaceOpResult( ok=False, error_message=f"{type(e).__name__}: {e}", ) return WorkspaceOpResult(ok=True, data={"operation": resp.get("name")}) async def patch_group_display( self, *, group_email: str, display_name: str | None = None, description: str | None = None, ) -> WorkspaceOpResult: """Update a Cloud Identity group's ``display_name`` / ``description`` via ``groups.patch``. Only the provided fields are sent in the updateMask. Applies to dynamic and static groups alike. """ body: dict[str, Any] = {} mask_parts: list[str] = [] if display_name is not None: body["displayName"] = display_name mask_parts.append("display_name") if description is not None: body["description"] = description mask_parts.append("description") if not mask_parts: return WorkspaceOpResult(ok=True, data={"noop": True}) def _call() -> dict[str, Any]: svc = _build_cloudidentity_service( self._sa_json(), self._subject(), [_CLOUD_IDENTITY_GROUPS_SCOPE], ) looked = svc.groups().lookup(groupKey_id=group_email).execute() res_name = looked.get("name") if not res_name: raise GoogleApiError( "patch_group_display", reason=f"group {group_email} not found", ) return svc.groups().patch( name=res_name, updateMask=",".join(mask_parts), body=body, ).execute() try: resp = await asyncio.to_thread(_call) except HttpError as e: return WorkspaceOpResult( ok=False, error_message=f"Google API error {e.resp.status}: {e.reason}", status=e.resp.status, ) except Exception as e: # noqa: BLE001 return WorkspaceOpResult( ok=False, error_message=f"{type(e).__name__}: {e}", ) return WorkspaceOpResult(ok=True, data={"operation": resp.get("name")}) async def scan_user_attributes(self) -> list[dict[str, Any]] | None: """Scan every **active** (non-suspended) user's custom-schema attributes so callers can (a) offer HR the real distinct values per field with live counts, and (b) locally reproduce a dynamic group's membership for the expected-vs-actual reconcile. Returns a list of ``{email, suspended, fields}`` where ``fields`` maps ``"." -> value`` for every populated string custom-schema field. Returns ``None`` on read failure so callers treat it as indeterminate rather than "everyone has no attributes" (which would make every rule look empty). """ try: customer_id = await self._resolve_customer_id() except Exception: logger.exception("scan_user_attributes: customer id resolve failed") return None out: list[dict[str, Any]] = [] page_token: str | None = None def _call(token: str | None) -> dict[str, Any]: svc = _build_directory_service( self._sa_json(), self._subject(), [_USER_SCOPE], ) kwargs: dict[str, Any] = { "customer": customer_id, "projection": "full", "maxResults": 500, "query": "isSuspended=false", "fields": "nextPageToken,users(primaryEmail,customSchemas)", } if token: kwargs["pageToken"] = token return svc.users().list(**kwargs).execute() for _page in range(_MAX_PAGINATOR_PAGES): try: resp = await asyncio.to_thread(_call, page_token) except HttpError as e: logger.warning( "scan_user_attributes HTTP %s: %s — returning None", e.resp.status, e.reason, ) return None except Exception: logger.exception("scan_user_attributes failed") return None for u in resp.get("users") or []: email = (u.get("primaryEmail") or "").strip().lower() if not email: continue flat: dict[str, str] = {} for schema, fields in (u.get("customSchemas") or {}).items(): if not isinstance(fields, dict): continue for fk, fv in fields.items(): if isinstance(fv, str) and fv.strip(): flat[f"{schema}.{fk}"] = fv.strip() out.append( {"email": email, "suspended": False, "fields": flat}, ) page_token = resp.get("nextPageToken") if not page_token: break else: logger.warning("scan_user_attributes: paginator overran") return None return out async def remove_group_member( self, *, group_email: str, member_email: str, ) -> WorkspaceOpResult: """Remove ``member_email`` from ``group_email``. Idempotent on 404. Semantically distinct from :meth:`remove_group_owner` (which is the owner-transfer path): this is the self-service "leave a group" operation. Google's ``members.delete`` removes the principal regardless of role, so the caller is responsible for the sole-owner guard before invoking this. """ def _call() -> None: svc = _build_directory_service( self._sa_json(), self._subject(), [_GROUP_SCOPE], ) svc.members().delete( groupKey=group_email, memberKey=member_email, ).execute() try: await asyncio.to_thread(_call) except HttpError as e: if e.resp.status == 404: return WorkspaceOpResult( ok=True, data={"already_absent": True}, status=404, ) if e.resp.status == 403: return WorkspaceOpResult( ok=False, error_message=( f"[{GOOGLE_TAG_SCOPE_MISSING}] members.delete " f"refused 403: {e.reason}. Hint: Service " "Account may be missing " "admin.directory.group.member scope." ), status=e.resp.status, ) return WorkspaceOpResult( ok=False, error_message=f"Google API error {e.resp.status}: {e.reason}", status=e.resp.status, ) except Exception as e: # noqa: BLE001 return WorkspaceOpResult( ok=False, error_message=f"{type(e).__name__}: {e}", ) return WorkspaceOpResult(ok=True) async def list_group_owners(self, group_email: str) -> list[dict[str, Any]] | None: """Return just the OWNER role members for one group, or ``None`` if the fetch failed for a permission-ish reason. The return-type distinction matters for the group-sync handler: an empty list legitimately means "no owners" (the handler will then drop any stale owner rows), while None means "we couldn't read this group, DON'T touch the local owner rows or we'd silently wipe everyone". The earlier implementation returned [] for both cases, which meant a missing ``group.member.readonly`` scope on the service account would blank every email_group_owners row on the first sync after deploy. """ owners: list[dict[str, Any]] = [] page_token: str | None = None def _call(token: str | None) -> dict[str, Any]: svc = _build_directory_service( self._sa_json(), self._subject(), ["https://www.googleapis.com/auth/admin.directory.group.member.readonly"], ) kwargs: dict[str, Any] = { "groupKey": group_email, "roles": "OWNER", "maxResults": 200, } if token: kwargs["pageToken"] = token return svc.members().list(**kwargs).execute() for _page in range(_MAX_PAGINATOR_PAGES): try: resp = await asyncio.to_thread(_call, page_token) except HttpError as e: # 403 = scope / per-group ACL problem — DO NOT treat # as "no owners"; signal indeterminate to caller. # 404 = group deleted between our list + our members # call, legitimately has no owners. if e.resp.status == 403: logger.warning( "list_group_owners %s: 403 (scope issue) — returning None", group_email, ) return None if e.resp.status == 404: return [] logger.warning("list_group_owners %s: %s", group_email, e.reason) return None except Exception: logger.exception("list_group_owners %s failed", group_email) return None owners.extend(resp.get("members") or []) page_token = resp.get("nextPageToken") if not page_token: break else: # Paginator overrun is indeterminate — return None so the # caller (group sync) treats it as "couldn't read this # group" rather than blanking out owners. logger.warning( "list_group_owners %s: paginator did not terminate " "after %d pages — returning None (indeterminate)", group_email, _MAX_PAGINATOR_PAGES, ) return None return owners # ─── User provisioning (PR 2026-05-02) ──────────────────────── # # ``create_user`` is the missing third leg of the joiner pipeline: # the tenant ALWAYS has the user list as the canonical seat record, # so creating the Google account up-front means every downstream # group_member binding finds the principal it's about to add. # ``list_all_users`` powers ``google.seat_sync`` (tenant-wide # mirror + orphan discovery) without per-group binding configuration. async def create_user( self, *, primary_email: str, given_name: str, family_name: str, password: str, org_unit_path: str | None = None, change_password_at_next_login: bool = True, recovery_email: str | None = None, recovery_phone: str | None = None, home_email: str | None = None, home_phone: str | None = None, ) -> WorkspaceOpResult: """Create a Google Workspace user. Idempotent on 409. On 409 (the email already exists somewhere in the tenant — suspended, archived, or active) the result carries ``already_exists=True`` so the caller can decide whether to unsuspend / link instead of treating this as a hard failure. ``org_unit_path`` defaults to ``/`` when not supplied (Google's own default). Caller is expected to pass the FTE / contractor OU path that was resolved from platform settings. ``change_password_at_next_login`` is True by default — the admin-generated password is meant for the very first sign-in only; the user rotates it themselves. """ body: dict[str, Any] = { "primaryEmail": primary_email, "name": {"givenName": given_name, "familyName": family_name}, "password": password, "changePasswordAtNextLogin": change_password_at_next_login, } if org_unit_path: body["orgUnitPath"] = org_unit_path # Contact + recovery info from HRIS (personal email / mobile). The # ``home``-typed contact slots populate the console's "Email (Home)" # / "Phone (Home)"; ``recoveryEmail`` / ``recoveryPhone`` populate # the Security → Recovery panel. ``recoveryPhone`` MUST be strict # E.164 or Google 400s the whole insert — the caller passes a # pre-formatted value (or None) so we set it only when present. if home_email: body["emails"] = [{"address": home_email, "type": "home"}] if home_phone: body["phones"] = [{"value": home_phone, "type": "home"}] if recovery_email: body["recoveryEmail"] = recovery_email if recovery_phone: body["recoveryPhone"] = recovery_phone def _call(insert_body: dict[str, Any]) -> dict[str, Any]: svc = _build_directory_service( self._sa_json(), self._subject(), [_USER_SCOPE], ) return svc.users().insert(body=insert_body).execute() # ``recoveryPhone`` / ``recoveryEmail`` are the only body fields Google # format-validates and 400s the WHOLE insert on. Account creation is # mandatory; recovery fields are best-effort — so on a 400 with either # present, retry ONCE without them rather than fail onboarding over a # fat-fingered HRIS phone (e.g. a Brazil mobile missing its 9-prefix). # home phone/email are free-form display slots, never a 400 cause, so # they stay. If the retry still fails, the 400 was something else and # we surface it. recovery_keys = ("recoveryPhone", "recoveryEmail") try: resp = await asyncio.to_thread(_call, body) except HttpError as e: if e.resp.status == 409: return WorkspaceOpResult( ok=True, data={"primary_email": primary_email}, status=409, already_exists=True, ) dropped = [k for k in recovery_keys if k in body] if e.resp.status == 400 and dropped: logger.warning( "google create_user %s: 400 on first insert; retrying " "without %s (recovery fields are best-effort): %s", primary_email, dropped, e.reason, ) retry_body = { k: v for k, v in body.items() if k not in recovery_keys } try: resp = await asyncio.to_thread(_call, retry_body) except HttpError as e2: if e2.resp.status == 409: return WorkspaceOpResult( ok=True, data={"primary_email": primary_email}, status=409, already_exists=True, ) return WorkspaceOpResult( ok=False, error_message=( f"Google API error {e2.resp.status}: {e2.reason}" ), status=e2.resp.status, ) except Exception as e2: # noqa: BLE001 return WorkspaceOpResult( ok=False, error_message=f"{type(e2).__name__}: {e2}", ) else: return WorkspaceOpResult( ok=False, error_message=( f"Google API error {e.resp.status}: {e.reason}" ), status=e.resp.status, ) except Exception as e: # noqa: BLE001 return WorkspaceOpResult( ok=False, error_message=f"{type(e).__name__}: {e}", ) return WorkspaceOpResult( ok=True, data={ "primary_email": resp.get("primaryEmail", primary_email), "id": resp.get("id"), "org_unit_path": resp.get("orgUnitPath"), }, ) async def list_all_users( self, *, include_suspended: bool = True, projection: str = "basic", ) -> list[dict[str, Any]]: """Page through every user in the Workspace tenant. Returns raw resource dicts (``primaryEmail``, ``id``, ``name``, ``suspended``, ``orgUnitPath``, ``aliases``, ``isAdmin``, ``creationTime``, ``lastLoginTime``, ...). Suspended users are included by default — ``google.seat_sync`` needs them so it can revoke the corresponding active SoftwareSeatAllocation rows when an account was suspended out-of-band. Raises :class:`GoogleApiError` when the FIRST page fails so a permission misconfig doesn't get silently masked as "tenant has zero users". """ users: list[dict[str, Any]] = [] page_token: str | None = None def _call(token: str | None) -> dict[str, Any]: svc = _build_directory_service( self._sa_json(), self._subject(), ["https://www.googleapis.com/auth/admin.directory.user.readonly"], ) kwargs: dict[str, Any] = { "customer": "my_customer", "maxResults": 500, # ``full`` adds customSchemas — the identity-sync # vendor snapshot needs it to show what the # google_schema projection actually wrote. "projection": projection, } if not include_suspended: kwargs["query"] = "isSuspended=false" if token: kwargs["pageToken"] = token return svc.users().list(**kwargs).execute() for _page in range(_MAX_PAGINATOR_PAGES): try: resp = await asyncio.to_thread(_call, page_token) except HttpError as e: logger.warning( "list_all_users HttpError %s: %s (collected so far: %d)", e.resp.status, e.reason, len(users), ) # Fail-stop on ANY page failure (not just page 1). The # original implementation would `break` on a mid-stream # error and return the truncated list. The caller then # used that as the authoritative tenant roster, mass- # revoking every Helix-side seat that happened to be on # a later page. With ~500 employees this means a single # transient 503 on page 2 nukes hundreds of paid seats. hint = None if e.resp.status == 403: hint = ( "Service Account likely missing " "admin.directory.user.readonly in the " "domain-wide-delegation allow-list" ) elif users: hint = ( f"Mid-stream failure after {len(users)} users — " "treating as transient and aborting the call so " "the seat-sync handler does not interpret a " "truncated list as 'these users were removed'." ) raise GoogleApiError( "list_all_users", status=e.resp.status, reason=e.reason, hint=hint, ) from e except Exception as e: logger.exception("list_all_users failed") raise GoogleApiError( "list_all_users", reason=str(e), ) from e batch = resp.get("users") or [] users.extend(batch) page_token = resp.get("nextPageToken") if not page_token: break else: raise GoogleApiError( "list_all_users", reason=( f"paginator did not terminate after " f"{_MAX_PAGINATOR_PAGES} pages (collected so far: " f"{len(users)})" ), ) return users async def list_group_members( self, group_email: str, ) -> list[dict[str, Any]]: """Return every member of ``group_email``. Symmetric with the Jira driver's ``list_group_members`` so ``google.seat_sync`` can mirror the per-plan reconciliation loop verbatim. Returns raw member dicts (``email``, ``role``, ``type``, ``status``, ``id``). Raises :class:`GoogleApiError` on the first page so a missing scope / wrong group_email is surfaced (rather than treated as an empty group, which would auto-revoke every active seat). """ out: list[dict[str, Any]] = [] page_token: str | None = None def _call(token: str | None) -> dict[str, Any]: svc = _build_directory_service( self._sa_json(), self._subject(), ["https://www.googleapis.com/auth/admin.directory.group.member.readonly"], ) kwargs: dict[str, Any] = { "groupKey": group_email, "maxResults": 200, } if token: kwargs["pageToken"] = token return svc.members().list(**kwargs).execute() for _page in range(_MAX_PAGINATOR_PAGES): try: resp = await asyncio.to_thread(_call, page_token) except HttpError as e: # Fail-stop on ANY page failure. Same reasoning as # ``list_all_users`` above — a mid-stream truncation # interpreted as authoritative would mass-revoke every # active seat for members on the missed page(s). hint = None if e.resp.status == 403: hint = ( "Service Account likely missing " "admin.directory.group.member.readonly in the " "domain-wide-delegation allow-list" ) elif e.resp.status == 404: hint = ( "Group not found — check binding's " "group_email or create the group in Google " "Workspace admin" ) elif out: hint = ( f"Mid-stream failure after {len(out)} members — " "aborting the call so seat-sync does not " "interpret a truncated list as removal." ) raise GoogleApiError( "list_group_members", status=e.resp.status, reason=e.reason, hint=hint, ) from e except Exception as e: logger.exception("list_group_members %s failed", group_email) raise GoogleApiError( "list_group_members", reason=str(e), ) from e out.extend(resp.get("members") or []) page_token = resp.get("nextPageToken") if not page_token: break else: raise GoogleApiError( "list_group_members", reason=( f"paginator did not terminate after " f"{_MAX_PAGINATOR_PAGES} pages for group " f"{group_email!r} (collected so far: {len(out)})" ), ) return out # NOTE (2026-05-07): an earlier PR added # ``list_inbound_saml_sso_profiles`` / ``list_idp_credentials`` / # ``create_inbound_saml_sso_profile`` here for an "SSO Apps # catalog sync" feature. They've been removed because Cloud # Identity's ``inboundSamlSsoProfiles`` API turned out to be the # wrong direction (third-party IdP → Google, not Google → SAS # apps). The SSO Apps catalog is admin-managed in Helix instead; # see [`server/app/domains/sso/app_catalog_service.py`] + # memory `feedback_verify_third_party_api_direction.md`. @staticmethod def _generate_strong_password(length: int = 16) -> str: """Generate a strong password using :mod:`secrets`. Composition: ≥1 lowercase, ≥1 uppercase, ≥1 digit, ≥1 symbol. Uses ``secrets`` (CSPRNG), never ``random``. The password is returned plaintext exactly ONCE to the caller — never logged, never persisted. """ if length < 12: length = 12 alphabet_lower = string.ascii_lowercase alphabet_upper = string.ascii_uppercase digits = string.digits # Avoid ambiguous / shell-hostile symbols like `$`,`"`,`'`,`\`. symbols = "!@#%^&*()-_=+[]{};:,.?" all_chars = alphabet_lower + alphabet_upper + digits + symbols required = [ secrets.choice(alphabet_lower), secrets.choice(alphabet_upper), secrets.choice(digits), secrets.choice(symbols), ] remaining = [secrets.choice(all_chars) for _ in range(length - 4)] chars = required + remaining # secrets.SystemRandom().shuffle to avoid positional bias secrets.SystemRandom().shuffle(chars) return "".join(chars)