"""EmailGroupCleanupService — the empty-static-group cleanup workbench. **The pool** is `status='active' AND is_dynamic=0 AND direct_members_count=0`. On production that is 133 groups out of 720 static ones (measured 2026-09-03). Because Google counts OWNERS inside ``directMembersCount``, every one of those 133 is also ownerless — the two facts are one fact, not two findings. **Why this is not a one-click sweep.** "Zero members" does NOT mean "abandoned". A group with no members can still be a live inbound address: an externally published contact (``careers@``, ``jobs@``), a robot / alert destination (``tidbcloud-bot@``, ``security-log@``), a nested delivery hop, or an IAM principal. Deleting it is a **Google hard delete** — the archive goes with it and mail to the address bounces afterwards, with no undelete. So the workbench's job is to *inform a per-row human decision*, not to automate the decision away: * every candidate carries **signals**, each with an explicit level; * an ``indeterminate`` signal is never rendered as "safe" — the whole point is that a probe we could not run is different from a probe that came back clean; * the executor **re-verifies live membership per group** immediately before deleting, because the mirror can be up to a sync interval stale and a group that gained a member since the list was drawn must not be swept; * one failed group never aborts the batch, and every deletion writes its own audit row keyed by a shared ``batch_id`` so the whole sweep is reconstructable afterwards. **Cloud IAM: half of it is queryable, and that half is queried.** Helix's own data-access domain compiles BigQuery grants whose principal IS an ``email_groups`` row — ``data_access_bindings`` and ``data_access_compiled_effects`` both carry an indexed ``principal_email_group_id`` FK and a ``provider_member`` of ``group:``. Deleting such a group destroys live dataset access AND orphans the binding that a reconcile tick will keep trying to compile, so it is a BLOCKING signal and this service reads it. What remains genuinely invisible is IAM granted **outside** Helix — a ``group:`` member added by hand in the GCP console on a project or org. GCP exposes those inline and this platform's ``list_scope_grants`` filters to ``user:`` only, so no Helix table carries them. Rather than emit a "no cloud reference" verdict that would be a confident lie for ``gcp-organization-admins@``, the service flags infrastructure-shaped names as an explicitly-labelled *heuristic*, names the gap in ``known_blind_spots``, and the surface says that check is manual. """ from __future__ import annotations import asyncio import logging import re import uuid from datetime import UTC, datetime from typing import Any from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.audit.service import AuditService from app.core.exceptions import AppException from app.domains.email_group.manage_service import ( EmailGroupManageService, real_aliases, ) from app.domains.email_group.models import EmailGroup from app.identity.person_models import IdentityPerson logger = logging.getLogger(__name__) # One execute call deletes at most this many groups. Each is a Google # ``groups.delete`` plus a live ``members.list`` re-verify, so the bound # keeps a single request inside the platform HTTP timeout — and it makes # the operator confirm a 133-group sweep in seven deliberate batches # rather than one click. MAX_CLEANUP_BATCH = 20 # A batch hard-deletes real addresses, so the reason bar is higher than # the single-group delete's 4 characters: this one has to read like a # sentence somebody wrote on purpose. MIN_CLEANUP_REASON = 10 # Rows the candidate list will return in one page. # # The page size is NOT the probe fan-out any more — that coupling is # what made this constant dangerous. Probing is capped independently at # ``_MAX_PROBED_ROWS`` and runs at ``_NESTING_PROBE_CONCURRENCY``, so # raising this number adds DB rows, not Google calls; rows past the # probe budget come back ``nesting_not_probed`` (indeterminate, never # clean). An earlier version of this comment cited "~25 sequential # rounds", arithmetic for a 20-way concurrency that was never # deliverable — see ``_NESTING_PROBE_CONCURRENCY``. # # Sized to hold the whole pool in one page ON PURPOSE. Prod's pool is # 133, and the workbench filters by RISK — which is derived from the # signals, so a server-paginated pool with a client-side risk filter # makes every number on screen describe a different set (page-scoped # counts under a pool-wide total, a pager offering pages the filter # cannot fill). On the surface whose premise is that the operator counts # what they are hard-deleting, one honest page beats three lying ones. MAX_CANDIDATE_PAGE = 500 # Concurrency for the nesting probe, and a hard cap on how many rows # get probed at all. # # Two corrections to an earlier version of this comment, which claimed # 20-way concurrency turned a 500-row page into "25 rounds (~8s)": # # 1. **20 was never deliverable.** ``list_groups_for_user`` runs each # call through ``asyncio.to_thread``, i.e. the event loop's DEFAULT # executor — ``min(32, cpu_count + 4)``, which is 6 on a 2-vCPU EC2 # box, and nothing in ``app/`` calls ``set_default_executor``. A # semaphore wider than the thread pool does not add parallelism, it # just queues; and every queued item sits in the pool that every # other Google/AWS driver call in the process shares. 6 matches the # real width instead of pretending. # 2. **The page size must not BE the budget.** ``_build_directory_service`` # rebuilds ``service_account.Credentials`` per call, so each probed # row costs a token exchange plus the API call — and the impersonated # subject's Directory quota (2400 q/min/user) is shared with # ``google.group_sync``, ``google.seat_sync`` and Day-1 account # provisioning. An unbounded 500-row page could degrade onboarding. # So the probe stops at ``_MAX_PROBED_ROWS`` and the remaining rows # get ``nesting_unknown`` — which keeps them at ``risk: review`` and # out of ``low``, i.e. it degrades in the SAFE direction using the # indeterminate machinery that already exists for a failed probe. _NESTING_PROBE_CONCURRENCY = 6 _MAX_PROBED_ROWS = 200 # Name markers operators already use to retire a group. Weak on their # own, useful for triage: matching means "somebody already said this is # dead", not "safe to delete". _DEPRECATED_MARKERS = ("弃用", "停用", "deprecated", "obsolete", "retired") _TEST_NAME = re.compile(r"(?:^|[^a-z])test\d*(?:$|[^a-z])|^tt$|demo", re.I) # Address shapes that commonly ARE cloud / infrastructure principals. # Deliberately a HEURISTIC, surfaced as one: see the module docstring on # why a definitive cloud signal is not obtainable from Helix data. _INFRA_NAME = re.compile( r"(?:^|[-_.])(?:gcp|aws|aliyun|tencent|azure|k8s|iam|sre|infra|devops|" r"admins?|security|audit|billing|root|bot|alert|noreply|no-reply)" r"(?:$|[-_.])", re.I, ) # Signal levels, most severe first. ``blocking`` never actually blocks # the API — a human may still decide to delete — but it is the level the # UI must make impossible to miss. LEVEL_BLOCKING = "blocking" LEVEL_WARNING = "warning" LEVEL_INDETERMINATE = "indeterminate" LEVEL_INFO = "info" _LEVEL_RANK = { LEVEL_BLOCKING: 3, LEVEL_INDETERMINATE: 2, LEVEL_WARNING: 1, LEVEL_INFO: 0, } def _signal(key: str, level: str, detail: dict[str, Any] | None = None) -> dict: return {"key": key, "level": level, "detail": detail or {}} class EmailGroupCleanupService: def __init__(self, db: AsyncSession) -> None: self.db = db self.audit = AuditService(db) self._manage = EmailGroupManageService(db) async def _get_driver(self): from app.domains.email_group.service import EmailGroupService return await EmailGroupService(self.db)._get_driver() def _require_manage(self, actor_roles: list[str]) -> None: self._manage.require_manage(actor_roles) # ── candidate pool ──────────────────────────────────────────── @staticmethod def _pool_conditions() -> list[Any]: """The pool, spelled so no NULL can leak in. ``is_dynamic == False`` (not ``!= True``): a NULL means Cloud Identity detection never ran determinately, and an undetermined group might be rule-driven — it must never enter a static-group cleanup pool. ``direct_members_count == 0`` (not ``COALESCE(...,0) == 0``): a NULL means never synced, and a group nobody has ever synced is not a group we know is empty. """ return [ EmailGroup.status == "active", EmailGroup.is_dynamic.is_(False), EmailGroup.direct_members_count == 0, ] async def blocking_signals_for( self, group: EmailGroup, ) -> tuple[list[dict], str]: """The same signals the workbench computes, for ONE group. Exists so the single-group delete cannot be a weaker door to the same irreversible operation. The workbench builds four independent "deleting this breaks something else" checks — nested-in-another-group, a real second address, a live BigQuery dataset grant, a Helix config reference — and the per-row 删除 button on the SAME screen had none of them: an operator could hard-delete, one group at a time, a group the workbench marks ``blocking``. Deliberately NOT including the pool's emptiness rule. The workbench refuses a group that has members because its pool is "empty static groups"; disbanding a group that still has members is a legitimate, intended action for its owner. What generalises is "something else depends on this address", not "it is empty". Costs one Google call (the nesting probe) plus a handful of indexed reads. Degrades the same way the list does: a probe that cannot run yields ``nesting_unknown`` (indeterminate), which is NOT blocking — an unavailable probe must not make deletion impossible, only unverified. """ internal = await self._internal_reference_index([group.email]) data_access = await self._data_access_index([group.id]) nested, probe_error, _skipped = await self._probe_nesting([group.email]) signals = self._signals_for( group, internal_refs=internal.get(group.email.lower(), []), nested_in=(nested or {}).get(group.email), nesting_probed=nested is not None, nesting_attempted=(nested is None or group.email in nested), probe_error=probe_error, data_access=data_access.get(group.id), ) return signals, self._risk_of(signals) async def list_candidates( self, *, actor_roles: list[str], probe_nesting: bool = True, page: int = 1, page_size: int = 50, ) -> dict[str, Any]: """Paginated candidates, each with its signals. The nesting probe is one Google call per returned row, so it is scoped to the page rather than the whole pool: an operator triages page by page, and a 133-call fan-out on every page load would make the surface unusable. """ self._require_manage(actor_roles) page = max(page, 1) page_size = max(min(page_size, MAX_CANDIDATE_PAGE), 1) from sqlalchemy import func conds = self._pool_conditions() total = int((await self.db.execute( select(func.count(EmailGroup.id)).where(*conds) )).scalar() or 0) rows = list((await self.db.execute( select(EmailGroup) .where(*conds) .order_by(EmailGroup.email) .offset((page - 1) * page_size) .limit(page_size) )).scalars().all()) # Helix-internal references, scoped to the rows being rendered. internal_refs = await self._internal_reference_index( [r.email for r in rows] ) data_access = await self._data_access_index([r.id for r in rows]) # dict[addr, list | None]: a None VALUE means probed and # failed for that address, an ABSENT KEY means never attempted # (past the budget). The two render as different signals. nested: dict[str, list[str] | None] | None = None probe_error: str | None = None probe_skipped = 0 if probe_nesting and rows: nested, probe_error, probe_skipped = await self._probe_nesting( [r.email for r in rows], ) items = [] for row in rows: signals = self._signals_for( row, # Index keys are lowercased; nothing enforces # lowercase on the column, and a raw-value lookup # would drop the signal for a mixed-case row — # failing in the UNSAFE direction. internal_refs=internal_refs.get(row.email.lower(), []), nested_in=(nested or {}).get(row.email), nesting_probed=nested is not None, # Absent key ⇒ past the budget, never attempted. nesting_attempted=( nested is None or row.email in nested ), probe_error=probe_error, data_access=data_access.get(row.id), ) items.append({ "id": str(row.id), "email": row.email, "name": row.name, "description": row.description, "aliases": real_aliases(row.aliases), "created_at": row.created_at, "last_synced_at": row.last_synced_at, "signals": signals, "risk": self._risk_of(signals), }) return { "total": total, "page": page, "page_size": page_size, "items": items, "nesting_probed": nested is not None, "nesting_probe_error": probe_error, # How many rows were never probed because the per-request # budget ran out. A DIFFERENT fact from a probe failure, and # the only one the operator can act on (ask for fewer rows), # so it gets its own field and its own banner. "nesting_probe_budget": _MAX_PROBED_ROWS, "nesting_rows_not_probed": probe_skipped, # Stated on every response so no caller can present this # surface as an exhaustive safety check. # Named precisely. Helix's OWN group→BigQuery grants ARE # checked (``data_access_principal``); what is invisible is # IAM a human added by hand in the GCP console, and mail # actually flowing through the address. "known_blind_spots": ["cloud_iam_outside_helix", "mail_traffic"], } async def _internal_reference_index( self, addresses: list[str], ) -> dict[str, list[str]]: """``{group email: [where Helix itself references it]}``. Catches the case that matters most among the machine-shaped candidates: Helix's own configuration points at the address, so deleting it breaks a Helix flow rather than somebody else's. Both source tables are tiny (41 integrations + 131 platform settings on prod), so this is one scan of them, not N LIKEs. Scoped to the addresses actually being rendered. Indexing the whole estate meant 760 × 172 substring passes per page load and threw 93% of the result away. Matching is BOUNDARY-ANCHORED ON BOTH SIDES. An unanchored ``addr in text`` makes any address that is a suffix of a longer one inherit the longer one's references — ``bot@pingcap.com`` matching ``tidbcloud-bot@pingcap.com`` — which produced a BLOCKING signal whose evidence was a fabrication, on a surface whose whole premise is that each signal is trustworthy. The right side needs the same treatment and for the same reason: this tenant's Google test-domain mirrors are literally ``.test-google-a.com``, so a left-only anchor makes ``ops@pingcap.com`` match ``ops@pingcap.com.test-google-a.com`` and fabricate a reference in the other direction. """ import re as _re from app.integrations.models import Integration, PlatformSetting wanted = {a.lower() for a in addresses if a} if not wanted: return {} # A local part may not be preceded by another local-part # character, and the domain may not be followed by another # domain character — ``.test-google-a.com`` mirrors make the # right-hand anchor load-bearing, not decorative. patterns = { addr: _re.compile( r"(? None: if blob is None: return text = str(blob).lower() for addr, pat in patterns.items(): if pat.search(text): index.setdefault(addr, []).append(label) # The block below is two SELECTs plus pure Python string work — # no flush, so nothing in it can deactivate the transaction # (only a failed flush does). The catch exists so a renamed or # missing config table cannot take down the whole candidate # list; the signal then degrades to absent, which is safe # because the response's ``known_blind_spots`` already tells the # caller this surface is not exhaustive. # savepoint-exempt: read-only (two SELECTs), no flush to poison. try: integrations = (await self.db.execute( select(Integration.provider, Integration.kind, Integration.config_public) )).all() for provider, kind, config in integrations: _scan(config, f"integration:{kind}/{provider}") settings = (await self.db.execute( select(PlatformSetting.key, PlatformSetting.value) )).all() for key, value in settings: _scan(value, f"platform_setting:{key}") except Exception: # A missing/renamed config table must not take down the # whole candidate list; the signal degrades to absent and # the response's ``known_blind_spots`` already tells the # caller this surface is not exhaustive. logger.exception("cleanup: internal-reference scan failed") return {} return index async def _data_access_index( self, group_ids: list[uuid.UUID], ) -> dict[uuid.UUID, dict[str, Any]]: """``{group id: BigQuery access this group is the principal of}``. The module docstring used to claim no Helix query could produce a cloud-IAM signal. That was false for the case with the worst blast radius: Helix's OWN data-access domain grants BigQuery dataset access to ``email_groups`` rows, keyed by an indexed ``principal_email_group_id`` FK, and compiles it into real GCP IAM as ``group:``. Hard-deleting such a group revokes live analyst access to a dataset and leaves a binding whose reconcile tick can never converge again. Two indexed ``IN`` queries on the ids already on the page — not a scan, and not per-row. Split by liveness on purpose. ``status IN ('active', 'revoked_pending')`` is the domain's own definition of a live binding (it is the CASE guarding ``active_semantic_key``); a fully revoked binding is history, which is worth SAYING but must not block a delete forever. """ from sqlalchemy import func from app.domains.data_access.models import ( DataAccessBinding, DataAccessCompiledEffect, ) if not group_ids: return {} out: dict[uuid.UUID, dict[str, Any]] = {} # savepoint-exempt: read-only (two SELECTs), no flush to poison. rows = (await self.db.execute( select( DataAccessBinding.principal_email_group_id, DataAccessBinding.status, func.count(DataAccessBinding.id), ) .where(DataAccessBinding.principal_email_group_id.in_(group_ids)) .group_by( DataAccessBinding.principal_email_group_id, DataAccessBinding.status, ) )).all() for gid, status, count in rows: if gid is None: continue entry = out.setdefault(gid, {"live": 0, "historical": 0, "datasets": set()}) if status in ("active", "revoked_pending"): entry["live"] += int(count or 0) else: entry["historical"] += int(count or 0) # The VERDICT comes from an aggregate, so it cannot be # truncated. An earlier version read the effect rows with a bare # ``.limit(500)`` and no ORDER BY and derived # ``has_present_effect`` from them — so on a large page the # later group ids, in an arbitrary result order, silently lost # both their dataset evidence AND (in the binding-revoked / # effect-still-present drift case this flag exists for) the # blocking verdict itself. A row rendered as BLOCKING with an # empty tooltip is the dead end the signal-evidence branch was # added to prevent; a row that quietly stops being blocking is # worse. present_counts = (await self.db.execute( select( DataAccessCompiledEffect.principal_email_group_id, func.count(DataAccessCompiledEffect.id), ) .where( DataAccessCompiledEffect.principal_email_group_id.in_( group_ids ), DataAccessCompiledEffect.desired_state == "present", ) .group_by(DataAccessCompiledEffect.principal_email_group_id) )).all() for gid, count in present_counts: if gid is None or not count: continue entry = out.setdefault(gid, {"live": 0, "historical": 0, "datasets": set()}) entry["has_present_effect"] = True entry["present_effects"] = int(count) # Dataset names are EVIDENCE, and evidence may legitimately be a # bounded sample — but a deterministic one, so two loads of the # same page show the same tooltip. effects = (await self.db.execute( select( DataAccessCompiledEffect.principal_email_group_id, DataAccessCompiledEffect.target_project_id, DataAccessCompiledEffect.target_dataset_id, ) .where( DataAccessCompiledEffect.principal_email_group_id.in_( group_ids ), DataAccessCompiledEffect.desired_state == "present", ) .order_by( DataAccessCompiledEffect.principal_email_group_id, DataAccessCompiledEffect.target_project_id, DataAccessCompiledEffect.target_dataset_id, ) .limit(2000) )).all() for gid, project, dataset in effects: if gid is None: continue entry = out.setdefault(gid, {"live": 0, "historical": 0, "datasets": set()}) entry["datasets"].add(f"{project}.{dataset}") return { gid: { "live": e["live"], "historical": e["historical"], "datasets": sorted(e["datasets"]), "has_present_effect": bool(e.get("has_present_effect")), "present_effects": int(e.get("present_effects") or 0), } for gid, e in out.items() } async def _probe_nesting( self, emails: list[str], ) -> tuple[dict[str, list[str] | None] | None, str | None, int]: """For each address, which OTHER groups list it as a member. ``groups.list(userKey=)`` answers this directly. It is the strongest available "still wired into something" signal: a group with no members of its own can still be a delivery hop inside another group, and deleting it silently breaks that path. Called with ``raise_on_missing=True``. By default the driver swallows Google's 404 into ``[]`` — correct for its original caller (a contractor with no Workspace account really is in zero groups), catastrophic here: for a GROUP address a 404 means the address is stale (renamed in the console since the last sync), and ``[]`` would be read as the determinate answer "not nested anywhere", clearing the row to ``risk: low`` and making it freely selectable for an irreversible hard delete. Returns ``(None, error)`` when the probe could not run at all — which the caller must render as *unknown*, never as "not nested". A per-address failure is likewise recorded as unknown for that address only. """ try: driver = await self._get_driver() except Exception as exc: # noqa: BLE001 return None, f"{type(exc).__name__}: {exc}", 0 # One Google round trip per address, and the page size IS the # fan-out (up to ``MAX_CANDIDATE_PAGE`` = 500). Serially that is # 500 sequential round trips per page load — minutes of pure I/O # wait on a triage screen. Bounded fan-out is the codebase's # established shape for exactly this (see # ``approval_registry/discovery._bounded_gather`` and the ~20 # ``asyncio.Semaphore`` probe sites under ``cmdb/providers``). sem = asyncio.Semaphore(_NESTING_PROBE_CONCURRENCY) # Budget applies to the ADDRESSES, in the order the page renders # them, so the rows an operator looks at first are the probed # ones. Everything past the budget is reported as unknown by # simply not appearing in the result dict. probed = emails[:_MAX_PROBED_ROWS] skipped = len(emails) - len(probed) async def _one(raw_addr: str) -> tuple[str, list[str] | None]: # Both sides of the self-comparison below must be lowercased. # ``emails`` comes from ``EmailGroup.email`` as stored, so a # mixed-case row made the group its own "parent" — an # instant BLOCKING ``nested_member_of`` on a self-reference. addr = raw_addr.lower() async with sem: try: parents = await driver.list_groups_for_user( addr, raise_on_missing=True, ) except Exception: # noqa: BLE001 # ``None`` ⇒ unknown for THIS address. Returning an # empty list would claim "not nested". # # Keyed by ``raw_addr``, like the success path: the # caller looks results up by ``row.email`` as # stored, so keying a failure by the LOWERCASED # address made a mixed-case row miss the lookup — # and, now that failures are kept in the dict rather # than skipped, miss straight into the "never # attempted" branch. A real permissions or # stale-address failure would then be reported as a # Helix-imposed probe cap, the exact inversion that # branch was added to prevent. return raw_addr, None return raw_addr, sorted({ (g.get("email") or "").lower() for g in parents if g.get("email") and (g.get("email") or "").lower() != addr }) # ``None`` value ⇒ probed and FAILED for that address. # Absent key ⇒ never attempted (past the budget). Both render as # ``nesting_unknown``, but with different reasons: telling an # operator their probe "failed" for a cap Helix imposed on # itself sends them off to investigate Google permissions that # are fine. out: dict[str, list[str] | None] = {} failures = 0 for addr, parents_or_none in await asyncio.gather( *(_one(e) for e in probed) ): if parents_or_none is None: failures += 1 out[addr] = parents_or_none if failures and failures == len(probed): return ( None, f"every nesting probe failed ({failures} addresses)", skipped, ) # The budget count travels SEPARATELY from the error string. # Folding it in made the only banner that renders that string — # headed 「部分嵌套探测失败」 — announce probe failures when zero # probes had failed, reasserting at page level the exact lie the # per-row ``nesting_not_probed`` branch removed. err = ( f"{failures} of {len(probed)} nesting probes failed" if failures else None ) return out, err, skipped def _signals_for( self, row: EmailGroup, *, internal_refs: list[str], nested_in: list[str] | None, nesting_probed: bool, nesting_attempted: bool = True, probe_error: str | None, data_access: dict[str, Any] | None = None, ) -> list[dict]: signals: list[dict] = [] # ── nesting ── if not nesting_probed: signals.append(_signal( "nesting_unknown", LEVEL_INDETERMINATE, {"reason": probe_error or "probe not run"}, )) elif not nesting_attempted: # Past the per-request probe budget. Nothing was tried, so # calling it a failure would be a lie AND would point the # operator at Google permissions instead of at the page # size. Still INDETERMINATE — an unprobed row must never # read as clean. signals.append(_signal( "nesting_not_probed", LEVEL_INDETERMINATE, # ``reason`` (not a bare ``budget`` number): the # frontend's evidence chain reads ``reason``, and a # detail key nothing reads renders a tag with NO # tooltip — the same declared-and-unread defect that # left ``live_bindings`` invisible one round earlier. { "reason": ( f"本次请求最多探测 {_MAX_PROBED_ROWS} 行,这一行" "超出了预算,没有探测过(不是探测失败)。" "缩小每页行数即可探测到它。" ), "budget": _MAX_PROBED_ROWS, }, )) elif nested_in is None: signals.append(_signal( "nesting_unknown", LEVEL_INDETERMINATE, {"reason": "probe failed for this address"}, )) elif nested_in: signals.append(_signal( "nested_member_of", LEVEL_BLOCKING, {"parents": nested_in}, )) # ── Helix's own BigQuery access, where this group IS the # principal (see ``_data_access_index``) ── if data_access: live = int(data_access.get("live") or 0) if live or data_access.get("has_present_effect"): signals.append(_signal( "data_access_principal", LEVEL_BLOCKING, { "live_bindings": live, # From the aggregate, so it is the true number # even when the dataset sample below is capped. "present_effects": int( data_access.get("present_effects") or 0 ), "datasets": data_access.get("datasets") or [], }, )) elif data_access.get("historical"): signals.append(_signal( "data_access_principal_historical", LEVEL_WARNING, { "revoked_bindings": int( data_access.get("historical") or 0 ), "datasets": data_access.get("datasets") or [], }, )) # ── real second address ── # ``aliases`` is TRI-STATE in the column: NULL = never synced, # ``[]`` = synced and genuinely none. ``real_aliases`` collapses # both to ``[]``, so testing only its truthiness makes a # never-synced group indistinguishable from a confirmed-no-alias # one — and lets it reach ``risk: low``, which is precisely the # bucket that says "we checked". A group with an unknown alias # list can be carrying a live second address whose deletion # bounces mail nobody knew was flowing, so the honest answer is # INDETERMINATE, the same treatment the nesting probe gets when # it cannot run. aliases = real_aliases(row.aliases) if aliases: signals.append(_signal( "has_real_alias", LEVEL_BLOCKING, {"aliases": aliases}, )) elif row.aliases is None: signals.append(_signal( "alias_unknown", LEVEL_INDETERMINATE, {"reason": "该组从未同步过别名列表"}, )) # ── Helix's own config points at it ── if internal_refs: signals.append(_signal( "referenced_in_helix_config", LEVEL_BLOCKING, {"references": sorted(set(internal_refs))}, )) # ── infrastructure-shaped name (heuristic, not a verdict) ── if _INFRA_NAME.search(row.email.split("@")[0]): signals.append(_signal( "possible_cloud_principal", LEVEL_WARNING, {"note": "name shape only — Helix cannot see group IAM bindings"}, )) # ── operator already marked it dead ── haystack = f"{row.name} {row.description or ''}".lower() markers = [m for m in _DEPRECATED_MARKERS if m in haystack] if markers: signals.append(_signal( "marked_deprecated", LEVEL_INFO, {"markers": markers}, )) local = row.email.split("@")[0] if _TEST_NAME.search(local) or _TEST_NAME.search(row.name): signals.append(_signal("looks_like_test", LEVEL_INFO, {})) signals.sort(key=lambda s: -_LEVEL_RANK.get(s["level"], 0)) return signals @staticmethod def _risk_of(signals: list[dict]) -> str: """Row-level triage bucket. ``low`` requires that nothing blocking OR indeterminate was found — a probe that could not run leaves the row in ``review``, because "we don't know" and "we checked, it's clean" must not collapse into the same colour. """ levels = {s["level"] for s in signals} if LEVEL_BLOCKING in levels: return "blocking" if LEVEL_INDETERMINATE in levels or LEVEL_WARNING in levels: return "review" return "low" # ── execute ─────────────────────────────────────────────────── async def _log_refusal( self, *, trace_id: str, actor_id: uuid.UUID, batch_id: str, group: EmailGroup, sequence: int, reason: str, refusal: str, extra: dict[str, Any], ) -> None: """One audit row per refused group. Best-effort. A refusal is a real, reviewable event — somebody asked for an irreversible delete and the system said no, with a reason — and it used to exist only as a line inside the batch summary's ``detail``, which the 16KB cap can discard in one piece. This row also lands in the GROUP's own timeline, which is where anyone investigating that address would look. Wrapped in ``best_effort`` so a failure to record a REFUSAL cannot abort a batch that still has real deletions to do — and cannot poison the session those deletions need. """ from app.core.txn import best_effort async with best_effort( self.db, label="email_group.cleanup_refused", ) as step: await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor_id, action="email_group.cleanup_refused", status="failure", resource_type="email_group", resource_id=str(group.id), detail={ "group_email": group.email, "batch_id": batch_id, "sequence": sequence, "reason": reason, "refusal": refusal, **extra, }, ) if not step.ok: logger.warning( "cleanup: could not record the refusal of %s (batch %s)", group.email, batch_id, ) async def _log_batch_summary( self, *, trace_id: str, actor_id: uuid.UUID, batch_id: str, status: str, detail: dict[str, Any], ) -> None: """The one batch-level audit row, plus its commit. Extracted so the caller's ``try`` wraps BOTH the write and the commit. Leaving the commit outside would defeat the guard: the ``audit.log`` would be caught while the failing ``commit`` two lines later still raised. """ await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor_id, action="email_group.cleanup_batch", status=status, resource_type="email_group", # The batch is not about one group; its id IS the resource. resource_id=batch_id, detail=detail, ) await self.db.commit() async def cleanup( self, *, group_ids: list[uuid.UUID], reason: str, actor: IdentityPerson, actor_roles: list[str], trace_id: str, ) -> dict[str, Any]: """Hard-delete the named groups on Google; soft-delete locally. Per group, in order: 1. Load it and re-check the pool predicate against the DB. 2. **Re-read live membership from Google.** The mirror is up to one sync interval old; a group that gained a member since the candidate list was drawn must be refused, not swept. This is the only guard standing between "stale list" and "deleted a live group". 3. Google ``groups.delete`` → local ``status='deleted'``. 4. One audit row for this group, carrying the batch id, its position in the batch, the reason, and a snapshot of the address / name / aliases so the row stays meaningful after the group is gone. A refusal or failure on one group never stops the others: a 20-group batch that dies on its 3rd entry would leave the operator unable to tell which ones went through. """ self._require_manage(actor_roles) reason_clean = (reason or "").strip() if len(reason_clean) < MIN_CLEANUP_REASON: raise AppException( code="REASON_TOO_SHORT", message=( f"批量清理是不可恢复操作,请填写至少 " f"{MIN_CLEANUP_REASON} 个字的原因。" ), status_code=400, ) # De-dupe while preserving the caller's order, so the audit # ``sequence`` matches what the operator saw on screen. unique_ids: list[uuid.UUID] = [] seen: set[uuid.UUID] = set() for gid in group_ids: if gid not in seen: seen.add(gid) unique_ids.append(gid) if not unique_ids: raise AppException( code="NO_GROUPS_GIVEN", message="请至少选择一个邮件组。", status_code=400, ) if len(unique_ids) > MAX_CLEANUP_BATCH: raise AppException( code="BATCH_TOO_LARGE", message=( f"一次最多清理 {MAX_CLEANUP_BATCH} 个邮件组" f"(本次 {len(unique_ids)} 个)。请分批执行。" ), status_code=400, ) batch_id = str(uuid.uuid4()) driver = await self._get_driver() results: list[dict[str, Any]] = [] # Snapshot into a plain local. The loop below can ``rollback()`` # (see the bookkeeping guard), and rollback EXPIRES every ORM # instance in the session — so a later ``actor.id`` would emit a # refresh SELECT from inside an argument list, which is the # MissingGreenlet trap. Same reasoning as # ``idempotent_route``'s reserved-row snapshot. actor_id = actor.id for index, gid in enumerate(unique_ids): row = (await self.db.execute( select(EmailGroup).where(EmailGroup.id == gid) )).scalar_one_or_none() if row is None: results.append({ "group_id": str(gid), "email": None, "ok": False, "outcome": "not_found", "reason": "该邮件组在 Helix 里不存在。", }) continue refusal = self._pool_refusal(row) if refusal is not None: results.append({ "group_id": str(gid), "email": row.email, "ok": False, "outcome": "refused", "reason": refusal, }) await self._log_refusal( trace_id=trace_id, actor_id=actor_id, batch_id=batch_id, group=row, sequence=index, reason=reason_clean, refusal=refusal, extra={}, ) continue # Live re-verify — the whole reason a stale list is safe. try: live = await driver.list_group_members(row.email) except Exception as exc: # noqa: BLE001 results.append({ "group_id": str(gid), "email": row.email, "ok": False, "outcome": "refused", "reason": ( "无法向 Google 确认当前成员数,拒绝删除:" f"{type(exc).__name__}: {exc}" ), }) continue # Count the RAW entries. Google returns whole-domain # members as ``type: "CUSTOMER"`` with an ``id`` and NO # ``email``, so an email-keyed reduction answers ``[]`` for a # group that has a member — and ``[]`` here is the last gate # before an irreversible hard delete. Any entry at all, with # or without an address, means "not empty". live_entries = list(live) if live_entries: with_email = sorted({ (m.get("email") or "").lower() for m in live_entries if m.get("email") }) results.append({ "group_id": str(gid), "email": row.email, "ok": False, "outcome": "refused", "reason": ( f"Google 上这个组现在有 {len(live_entries)} 名成员" "(候选清单生成后有人加入),已跳过。" ), "live_member_count": len(live_entries), "live_members": with_email[:20], }) # A refusal gets its OWN audit row. # # "An operator attempted a hard delete on careers@ and # the system refused because it had just gained # members" is exactly the event this desk exists to # record — and it used to live only inside # ``cleanup_batch``'s ``detail``, where 20 refusals × 20 # addresses serialises past ``AuditService``'s 16KB cap # and the WHOLE manifest (batch reason included) is # replaced by a ``_truncated`` marker. Per-group rows # cannot be lost that way, and they show up in the # group's own timeline where somebody would look. await self._log_refusal( trace_id=trace_id, actor_id=actor_id, batch_id=batch_id, group=row, sequence=index, reason=reason_clean, refusal=( f"Google 上这个组现在有 {len(live_entries)} 名成员" ), extra={ "live_member_count": len(live_entries), "live_members": with_email[:20], }, ) continue resp = await driver.delete_group(email=row.email) if not resp.ok: results.append({ "group_id": str(gid), "email": row.email, "ok": False, "outcome": "failed", "reason": resp.error_message, }) continue already_absent = bool((resp.data or {}).get("already_absent")) snapshot = { "group_email": row.email, "name": row.name, "description": row.description, "aliases": real_aliases(row.aliases), "created_at": ( row.created_at.isoformat() if row.created_at else None ), } # From here on the group is ALREADY GONE from Google, so the # local bookkeeping must not be allowed to abort the batch: # a DataError or deadlock on group 7 of 20 would propagate, # skip groups 8-20, never write the batch summary, and leave # the operator with a 500 and no idea which addresses were # hard-deleted — for an operation with no undelete. try: row.status = "deleted" row.updated_at = datetime.now(UTC) await self.db.flush() await self.audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor_id, action="email_group.cleanup_deleted", status="success", resource_type="email_group", resource_id=str(row.id), detail={ **snapshot, "reason": reason_clean, "batch_id": batch_id, # ``audit_logs.created_at`` is a whole-second # DATETIME, so a 20-group batch lands inside one # or two seconds and cannot be ordered by time. # The sequence is what makes the batch replayable # in the order the operator saw. "sequence": index, "already_absent_on_google": already_absent, }, ) # Commit per group: a mid-batch crash must leave the # groups already deleted on Google recorded as deleted in # Helix. One trailing commit would turn any late failure # into silent drift between Google and the mirror. await self.db.commit() except Exception as exc: # noqa: BLE001 # The Google delete SUCCEEDED; only the local record # failed. Say exactly that — reporting it as "failed" # would read as "the group is still there". await self.db.rollback() logger.exception( "cleanup: %s deleted on Google but local bookkeeping " "failed (batch %s)", snapshot["group_email"], batch_id, ) results.append({ "group_id": str(gid), "email": snapshot["group_email"], "ok": False, "outcome": "deleted_record_failed", "reason": ( "Google 已删除该组,但 Helix 本地记录写入失败:" f"{type(exc).__name__}: {exc}。" "下一次邮件组同步会把本地状态改为已删除。" ), }) continue results.append({ "group_id": str(gid), "email": row.email, "ok": True, "outcome": "already_absent" if already_absent else "deleted", }) deleted = sum(1 for r in results if r["ok"]) record_failed = sum( 1 for r in results if r["outcome"] == "deleted_record_failed" ) failed = sum(1 for r in results if r["outcome"] == "failed") # ``status`` describes whether the BATCH did what it was asked, # not whether anything was deleted. A sweep whose every row was # correctly refused (someone joined the group in the meantime) # did its job exactly right; labelling that ``failure`` teaches # the auditor to ignore the field. Only a Google delete error or # a lost local record is a failure. batch_status = "failure" if (failed or record_failed) else "success" # Guarded for the same reason as the per-group block above: by # this point real addresses are ALREADY GONE from Google, and # every one of them is already committed with its own audit row. # Letting a DataError on the summary raise would turn a # successful irreversible sweep into a 500 with no response # body — the operator would not learn WHICH addresses were # deleted, on the one operation that has no undo. try: await self._log_batch_summary( trace_id=trace_id, actor_id=actor_id, batch_id=batch_id, status=batch_status, detail={ "batch_id": batch_id, "reason": reason_clean, "requested": len(unique_ids), "deleted": deleted, "refused": sum( 1 for r in results if r["outcome"] == "refused" ), "failed": failed, "not_found": sum( 1 for r in results if r["outcome"] == "not_found" ), "deleted_record_failed": record_failed, # Trimmed on purpose. ``AuditService`` replaces the # WHOLE detail with a ``_truncated`` marker past # 16KB, and 20 refusals each carrying 20 live # addresses measured ~19.7KB — so the fat version # cost the batch reason and every counter as well. # The full per-address evidence lives in the # per-group rows (``cleanup_deleted`` / # ``cleanup_refused``), which cannot be discarded # together; this row is the index into them. "results": [ { k: v for k, v in r.items() if k != "live_members" } for r in results ], }, ) except Exception: await self.db.rollback() logger.exception( "cleanup: batch summary audit failed for batch %s " "(per-group rows are already committed)", batch_id, ) return { "batch_id": batch_id, "requested": len(unique_ids), "deleted": deleted, "refused": sum(1 for r in results if r["outcome"] == "refused"), "failed": sum(1 for r in results if r["outcome"] == "failed"), "not_found": sum(1 for r in results if r["outcome"] == "not_found"), "deleted_record_failed": record_failed, "results": results, } @staticmethod def _pool_refusal(row: EmailGroup) -> str | None: """Why this row may NOT be cleaned up, or ``None`` if it may. Re-checked at execute time rather than trusted from the request: the client's candidate list can be minutes old, and a group that has since been made dynamic, gained members, or been deleted must not be swept because it was on screen when the operator started. """ if row.status != "active": return "该邮件组已经是删除状态。" if row.is_dynamic is True: return "这是动态组,成员由规则计算,不在静态组清理范围内。" if row.is_dynamic is None: return ( "该组的静态/动态类型尚未判定,不能当作静态空组清理。" "请先运行一次邮件组同步。" ) if row.direct_members_count is None: return "该组从未同步过成员数,不能当作空组清理。" if row.direct_members_count != 0: return ( f"Helix 记录该组有 {row.direct_members_count} 名成员," "不在空组清理范围内。" ) return None