"""Auto-create a Google Workspace account for an active employee. Single entry point: :func:`create_account_for(db, person_id)`. Idempotent — safe to re-run for the same person; returns the existing external identity if Google already has them. What this does (in order): 1. Read ``platform_settings.google.account_provisioning`` to know FTE / contractor domains + OUs + the master gating switch. 2. Skip if the gating switch is off, the person is non-active, the person already carries a ``google`` IdentityExternalIdentity, OR no Google integration is configured. 3. Compute the proposed local part. PREFER the HRIS-provided corporate email (``work_email``, set by the SuccessFactors sync — HR pre-assigns the joiner's address): we honour its local part verbatim. Fall back to the pinyin ``given.family`` derivation when work_email is missing or the synthetic ``@successfactors.local`` placeholder, and to ``employee_no`` when even the name yields nothing usable. 4. Pick the domain and OU based on whether the person's ``company_code`` is in ``platform_settings.google.contractor_company_codes``. 5. Generate a 16-char strong password. 6. Try ``driver.create_user`` with the proposed email. On 409 (email collision), append a zero-padded counter (``zhang.wei01``, ``zhang.wei02``, … ``zhang.wei09``) up to 9 attempts, then give up with a clear error. 7. On success: write IdentityExternalIdentity(provider='google'), write a SoftwareSeatAllocation against the seeded "Google Workspace" plan (so the existing offboarding scan picks it up automatically), send a welcome-with-password email to the person's ``personal_email`` (NOT work_email — the work mailbox we just created has nothing in it yet), and report the final chosen email back to HRIS via ``notify_hris_email_assigned`` (a logged no-op until the HR callback URL is configured). 8. Audit every step with the same trace_id so the timeline view reads cleanly. Welcome-email failure self-heal (``pending_welcome``): if the account is minted but the welcome email (carrying the first-login password) fails, we SUSPEND the mailbox (security control — never leave an active mailbox with an undelivered password) AND flip the identity to ``identity_status='pending_welcome'`` instead of ``active``. That state is a first-class marker, not a dead end: * ``create_account_for`` re-runs (hire_day cron / manual) short-circuit into :func:`_reprovision_pending_welcome`, which regenerates the password, re-sends the welcome, and unsuspends ONLY on email success — preserving the create-time invariant. A leaver never reaches this path (the ``employment_status == 'active'`` gate skips them first). * ``google.seat_sync`` drift-cleanup leaves a ``pending_welcome`` seat alone, so an email outage no longer strips the base seat after 2h. Rehire heal (``inactive`` → unsuspend, 2026-07-10 P01485): when a leaver whose mailbox offboarding auto-suspended is rehired (``leaver_pending``/ ``left`` → ``active``), the onboard ``reactivate`` dispatch re-enters ``create_account_for``; the ``inactive`` login binding routes into :func:`_reprovision_rehired_leaver`, which reverses the offboarding suspend through the same pending_welcome heal (fresh password → welcome → unsuspend on email success) or, if the mailbox is already live, just restores the SSO binding + seat. Shared / service / retained mailboxes (external-account registry) and the deliberate admin freeze (``identity_status='suspended'``) are never auto-touched. The function NEVER raises — every failure path returns a structured result for the caller to log. HRIS sync is a batch loop and one person's Google failure must not abort the rest. """ from __future__ import annotations import re from dataclasses import dataclass from datetime import UTC, datetime from typing import Any from uuid import UUID from pypinyin import Style, lazy_pinyin from sqlalchemy import func, or_, select from sqlalchemy.ext.asyncio import AsyncSession from app.audit.service import AuditService from app.core.corporate_domains import ( email_is_corporate, get_corporate_email_domains, ) from app.core.locale import pick_localized, resolve_lang from app.core.phone import to_e164 from app.core.safe_logging import get_safe_logger from app.core.txn import best_effort from app.domains.asset.models import ( AssetSoftwareCatalog, SoftwareLicensePlan, SoftwareSeatAllocation, ) from app.identity.hris_email_writeback import notify_hris_email_assigned from app.identity.person_models import ( IdentityExternalIdentity, IdentityPerson, ) from app.integrations.drivers.google_workspace import ( GoogleWorkspaceDriver, WorkspaceOpResult, ) from app.integrations.models import Integration from app.integrations.service import _resolve, get_platform_setting from app.notifications.templates.email_group import ( TEMPLATES as _EMAIL_GROUP_TEMPLATES, ) logger = get_safe_logger(__name__) # Copy for the first-login credential mail. Resolved at import so a missing / # renamed key fails the smoke-import in CI rather than at 09:00 on someone's # first day. See ``_send_password_email`` for why this bypasses the # ``platform_settings`` override layer. # Typed ``dict[str, object]`` (not ``dict[str, str]``) purely so it satisfies # ``pick_localized``'s parameter — ``dict`` is invariant in its value type. _PASSWORD_MAIL_TEMPLATE: dict[str, object] = dict( _EMAIL_GROUP_TEMPLATES["google_account_password"] ) _GOOGLE_PROVIDER = "google" _GOOGLE_CATALOG_NAME = "Google Workspace" _PROVISIONING_SETTING = "google.account_provisioning" _CONTRACTOR_CODES_SETTING = "google.contractor_company_codes" # Self-heal marker for a mailbox that was MINTED but left SUSPENDED because # the welcome email (carrying the first-login password) failed to send. The # identity row is NOT ``active`` (auth blocks login on any non-active status, # and the onboarding cockpit's active-login readiness excludes it — both # correct, the account is suspended and has no delivered credential), but it # is also NOT ``suspended``/``inactive`` (which route to manual review). This # distinct value tells ``create_account_for`` to RE-PROVISION on the next run # (regenerate password → re-send welcome → unsuspend ONLY on email success) # and tells ``google.seat_sync`` to leave the base seat alone while pending. # 15 chars — fits ``identity_external_identities.identity_status`` String(16). _STATUS_PENDING_WELCOME = "pending_welcome" # 9 zero-padded suffix attempts (``01`` … ``09``) cover any realistic # collision — a 10th zhang.wei would mean nine real Zhang Weis already # work here. _MAX_COLLISION_ATTEMPTS = 9 _PLACEHOLDER_LOCAL_PARTS = frozenset({ "unknown", "null", "na", "none", "nil", "placeholder", "dummy", }) def _is_placeholder_local_part(value: str | None) -> bool: local = (value or "").strip().lower().split("@", 1)[0] normalized = re.sub(r"[^a-z0-9]", "", local) return normalized in _PLACEHOLDER_LOCAL_PARTS @dataclass(slots=True) class ProvisioningOutcome: """Result of a single create_account_for call. ``status`` values: * ``created`` — new Google account minted, password emailed. * ``linked`` — Google already had this email; we just attached the IdentityExternalIdentity row + the seat allocation. * ``skipped`` — gating off, person ineligible, integration missing, etc. ``reason`` carries the human-readable cause. * ``failed`` — the Google API call (or the dependent email send) failed in a way the operator must investigate. ``reason`` carries the upstream error. """ status: str reason: str | None = None primary_email: str | None = None google_user_id: str | None = None seat_id: UUID | None = None # Outcome of the HRIS email write-back (``sent`` / ``skipped`` / # ``failed`` / ``error`` / ``None`` when no account was created). hris_notify_status: str | None = None def _local_part_from_name( *, display_name: str | None, name_pinyin: str | None, employee_no: str | None, max_length: int, first_name: str | None = None, last_name: str | None = None, ) -> str: """Build the email local part. **Convention** (PingCAP, 2026-06-02, post-PR #677 override): emails are ``given.family`` (名.姓). Examples — ``陈伟`` → ``wei.chen``, ``张三丰`` → ``sanfeng.zhang``, ``欧阳娜娜`` → ``nana.ouyang``, ``Tom Smith`` → ``tom.smith``. The 3-syllable case (1-char family + 2-char given) concatenates the given syllables into one token so a new hire gets ``sanfeng.zhang`` rather than ``san.feng.zhang`` — matches the email-system rule "exactly one dot, family on the right". Preference order: 1. ``first_name`` + ``last_name`` populated by HRIS sync (the SF ``User.firstName`` / ``User.lastName`` columns, migration 243). AUTHORITATIVE path — given/family is unambiguous so the output is canonical ``given.family`` regardless of source script. Pinyinises each side independently into ONE token, so a compound family name like ``欧阳`` becomes ``ouyang`` (one token) and the email is the correct ``nana.ouyang`` — the only path that gets compound family names right. 2. ``name_pinyin`` already populated by HRIS sync (legacy — used on older rows that pre-date migration 243 or whose SF firstName / lastName are blank). 3. Pinyin computed on the fly from ``display_name``. 4. ``employee_no`` lowercased. Paths 2–3 can't tell which character is the family for a compound surname; they apply the "first token is family" heuristic which is correct ~99% of the time (single-char family names are the overwhelming majority). The SF firstName/lastName path is the only correct answer for ``欧阳`` / ``司马`` / ``上官`` employees. Output is capped at ``max_length``; non-letter / non-digit chars stripped from each token. """ auth_tokens = _name_tokens_from_first_last( first_name=first_name, last_name=last_name, ) if auth_tokens: result = _cap_local_part(auth_tokens, max_length=max_length) return "" if _is_placeholder_local_part(result) else result tokens = _name_tokens(display_name=display_name, name_pinyin=name_pinyin) if tokens: result = _cap_local_part(tokens, max_length=max_length) # A present placeholder must stop account creation; falling through # to employee_no would still mint an account from an SF shell record. return "" if _is_placeholder_local_part(result) else result if employee_no: return _sanitize_token(employee_no)[:max_length] return "" def _local_part_from_hris_email( work_email: str | None, *, max_length: int, corporate_domains: frozenset[str], ) -> str: """Extract the local part of the HRIS-provided corporate email. HR pre-assigns the joiner's corporate 邮箱名 in SuccessFactors; it flows into ``IdentityPerson.work_email`` via the HRIS sync. This is the authoritative "邮件名" the business wants Helix to mint on Google, so we honour its local part verbatim (sanitised, dots preserved). Two shapes arrive here (HRIS field update, 2026-06-11): a full address (``san.zhang@pingcap.com``) or a **bare local part** (``san.zhang``) — the onboarding push now carries just the 邮箱名 because the final address depends on Helix's collision dedup. A bare value is used as the local part directly. Returns ``""`` — so the caller falls back to the pinyin derivation — when work_email is missing, is the synthetic SF placeholder ``@successfactors.local`` (minted when SF has no real email for the person yet), or is a **full address on a non-corporate domain** (a personal/contact address such as ``...@qq.com`` that leaked into work_email — NOT the corporate 邮箱名; defence-in-depth on top of the sync-side guard, and the path that self-heals rows written before that guard existed). A **bare local part** (no ``@``) has no domain to vet and is always honoured. **Whitespace is a name separator, not noise** (P01574, 2026-09-07). HR sometimes types the person's NAME into the 邮箱名 field — prod had ``"Koji Betsuyaku"`` against a live ``koji.betsuyaku@`` mailbox. The old code stripped the space along with every other non-alnum char and read that as ``kojibetsuyaku`` — a value nobody ever wanted, that disagreed with the real mailbox forever and paged IT nightly to "rename" a correct address. A two-token value IS ``given family``, and the house convention (PR #677) spells that ``given.family``, so the separator becomes a dot. Underscore is folded the same way. Three or more tokens (``"Nicholas Dakota Muniz"``) is a display name, not an email name: there is no safe way to guess which token is the family, so it returns ``""`` and the caller falls through to the authoritative ``firstName``/``lastName`` derivation, which gets it right. Minting ``nicholasdakotamuniz@`` — the old behaviour — was never the answer. """ email = (work_email or "").strip().lower() local, sep, domain = email.partition("@") if not local or domain.endswith("successfactors.local"): return "" if sep and not email_is_corporate(email, corporate_domains): return "" # Fold whitespace / underscore into the dot structure BEFORE the dot # split below, so ``koji betsuyaku`` and ``koji.betsuyaku`` are the # same address rather than two different ones. groups = [g for g in re.split(r"[\s_]+", local) if g] if len(groups) > 2: return "" local = ".".join(groups) # Preserve the dot structure (``wei.chen``) — sanitise each piece and # drop empties so a stray leading/trailing dot can't sneak through. pieces = [_sanitize_token(p) for p in local.split(".")] pieces = [p for p in pieces if p] if not pieces: return "" result = _cap_local_part(pieces, max_length=max_length) return "" if _is_placeholder_local_part(result) else result def _cap_local_part(tokens: list[str], *, max_length: int) -> str: joined = ".".join(tokens) if len(joined) <= max_length: return joined # Truncate the LAST token to fit. Tokens are ``[given..., family]`` # post the CJK reverse, so the LAST token is the family — the # most-likely-unique part of the name. head = ".".join(tokens[:-1]) if not head: # ONE over-long token: there is no head to keep it apart from, so # just truncate it. Falling through to the ``f"{head}.{...}"`` # return below prepended a **leading dot** (a 50-char HR value # minted ``.aaaa…@``) — the old length-only test could not see it. return tokens[-1][:max_length] budget = max_length - len(head) - 1 # room for the dot if budget < 1: return head[:max_length] return f"{head}.{tokens[-1][:budget]}" def _name_tokens_from_first_last( *, first_name: str | None, last_name: str | None, ) -> list[str]: """Build the unambiguous ``[given, family]`` token list from the SF firstName / lastName columns. Each side is pinyinised independently then sanitised — so: * ``("伟", "陈")`` → ``["wei", "chen"]`` (→ ``wei.chen``) * ``("三丰", "张")`` → ``["sanfeng", "zhang"]`` (→ ``sanfeng.zhang``) * ``("娜娜", "欧阳")`` → ``["nana", "ouyang"]`` (→ ``nana.ouyang``) * ``("Tom", "Smith")`` → ``["tom", "smith"]`` (→ ``tom.smith``) Returns ``[]`` when either side is empty / whitespace-only — the caller falls through to the legacy display_name pinyin path so we never write a half-name email (e.g. ``wei.`` with a missing family). """ first_clean = (first_name or "").strip() last_clean = (last_name or "").strip() if not first_clean or not last_clean: return [] first_token = _pinyinise_single_side(first_clean) last_token = _pinyinise_single_side(last_clean) if not first_token or not last_token: return [] return [first_token, last_token] def _pinyinise_single_side(value: str) -> str: """Pinyinise + sanitise one side of a name into a single token. Chinese chars get romanised through ``lazy_pinyin`` (per-char output, joined without separator) so a compound 2-char family like ``欧阳`` becomes ``ouyang``, not ``ou.yang`` — it's ONE family name. Same for a 2-char given ``三丰`` → ``sanfeng``. Latin chars pass through as-is. Parens and anything else are stripped. """ cleaned = re.sub(r"[(（].*?[)）]", "", value).strip() if not cleaned: return "" pieces = lazy_pinyin(cleaned, style=Style.NORMAL, errors="default") return _sanitize_token("".join(pieces)) def _name_tokens( *, display_name: str | None, name_pinyin: str | None, ) -> list[str]: """Legacy fallback used when SF ``firstName`` / ``lastName`` are not populated on the row. Normalises a name into ``[given, family]`` ASCII lowercase tokens following the 名.姓 convention. CJK names: ``lazy_pinyin`` (or HRIS-supplied ``name_pinyin``) emits tokens in DISPLAY order = ``family given...``. The legacy heuristic is "first token is family, rest is given"; given parts get concatenated into ONE token so the output is always exactly ``[given_concat, family]`` — two-token form, one dot in the final email: * 陈伟 (2 tokens) → ``["wei", "chen"]`` → ``wei.chen`` * 张三丰 (3 tokens) → ``["sanfeng", "zhang"]`` → ``sanfeng.zhang`` * 欧阳娜娜 (4 tokens) → ``["yangnana", "ou"]`` → approximate (compound family ``欧阳`` mis-identified as ``欧``; SF firstName/lastName path is the correct fix for those). Latin names are written in Western order — ``given [middle…] family`` — so the family is the LAST token. Any middle/given tokens before it are concatenated into ONE given token so the output keeps PingCAP's "exactly one dot, family on the right" invariant (see ``_latin_given_family_concat``): * ``Tom Smith`` (2 tokens) → ``["tom", "smith"]`` → ``tom.smith`` * ``Chi Yeung Yip`` (3 tokens) → ``["chiyeung", "yip"]`` → ``chiyeung.yip`` (romanised HK/overseas name: given=``Chi Yeung``, family=``Yip`` — indistinguishable by script from a Western middle name, but the SF instance leaves firstName/lastName blank for these, so this is the only line of defence against a multi-dot mailbox.) * ``Tom Robert Smith`` (3 tokens)→ ``["tomrobert", "smith"]`` → ``tomrobert.smith`` (a genuine Western middle name collapses too; the one-dot invariant wins — these are rare and the address stays valid + on-convention). Mixed ``Andy 陈`` style: Latin given tokens come first, then a concatenation of all CJK chars treated as ONE family. The Latin given side is likewise collapsed to one token: ``Andy 陈`` → ``["andy", "chen"]``; ``Andy Robert 陈`` → ``["andyrobert", "chen"]``. """ has_cjk_display = bool(display_name and _contains_cjk(display_name)) # Path 1 — pinyin string already has whitespace, trust it. The # CJK-vs-Latin distinction comes from ``display_name`` (HRIS only # populates name_pinyin for Chinese-name employees, so when # name_pinyin has spaces it's almost certainly CJK, but defer to # the display_name signal when present). if name_pinyin and any(c.isspace() for c in name_pinyin): tokens = [_sanitize_token(t) for t in name_pinyin.split()] tokens = [t for t in tokens if t] if not tokens: pass # fall through to display_name elif has_cjk_display or display_name is None: # Treat as CJK: reverse + concat given parts into one. return _cjk_reverse_concat(tokens) else: # display_name is Latin (Western order, family last) — collapse # any given/middle tokens so the email keeps exactly one dot. return _latin_given_family_concat(tokens) # Path 2 — display_name is the source of truth. if display_name: cleaned = re.sub(r"[(（].*?[)）]", " ", display_name) latin_tokens: list[str] = [] cjk_segment_tokens: list[str] = [] for segment in cleaned.split(): if _contains_cjk(segment): raw_tokens = lazy_pinyin( segment, style=Style.NORMAL, errors="default", ) for t in raw_tokens: sanitized = _sanitize_token(t) if sanitized: cjk_segment_tokens.append(sanitized) else: raw_tokens = lazy_pinyin( segment, style=Style.NORMAL, errors="default", ) for t in raw_tokens: sanitized = _sanitize_token(t) if sanitized: latin_tokens.append(sanitized) if latin_tokens and cjk_segment_tokens: # Mixed shape like ``Andy 陈`` — the CJK part is purely the # family (one or more chars concatenated). Latin tokens are # the given (possibly with middle); collapse them so the # given side stays one token → exactly one dot. family = "".join(cjk_segment_tokens) return _latin_given_family_concat([*latin_tokens, family]) if cjk_segment_tokens: return _cjk_reverse_concat(cjk_segment_tokens) if latin_tokens: return _latin_given_family_concat(latin_tokens) # Path 3 — fall back to ``name_pinyin`` even if it has no spaces # (single-token name like "chenwei" or romanised "tomsmith"). One # token is fine for an email local part. We CAN'T separate # family/given here without losing information, so accept the # single-token form; the SF firstName/lastName path is the only # way to get correct order for these legacy rows. if name_pinyin: token = _sanitize_token(name_pinyin) if token: return [token] return [] def _cjk_reverse_concat(tokens: list[str]) -> list[str]: """Apply the CJK ``[family, given...]`` → ``[given_concat, family]`` rule. Single-token list is returned as-is. * 2-token [chen, wei] → [wei, chen] * 3-token [zhang, san, feng] → [sanfeng, zhang] * 4-token [ou, yang, na, na] → [yangnana, ou] (approximate for compound family names — SF firstName/lastName path is the authoritative fix for those). """ if len(tokens) < 2: return tokens family = tokens[0] given = "".join(tokens[1:]) return [given, family] def _latin_given_family_concat(tokens: list[str]) -> list[str]: """Romanised (non-CJK) counterpart of :func:`_cjk_reverse_concat`. Latin/romanised display names are written in Western order — ``given [middle…] family`` — so the family is the LAST token. Collapse everything before it into ONE given token so the email obeys PingCAP's "exactly one dot, family on the right" invariant: * 2-token [tom, smith] → [tom, smith] (already 1 dot) * 3-token [chi, yeung, yip] → [chiyeung, yip] → chiyeung.yip * 3-token [tom, robert, smith] → [tomrobert, smith] → tomrobert.smith Fewer than 3 tokens already satisfy the one-dot rule and pass through unchanged. This is the last line of defence for romanised HK / overseas names whose SF instance leaves firstName/lastName blank — without it the tokens stay verbatim and the joiner gets a multi-dot mailbox (``chi.yeung.yip@``). Where SF *does* give firstName/lastName the authoritative :func:`_name_tokens_from_first_last` path runs first and this never fires. """ if len(tokens) < 3: return tokens return ["".join(tokens[:-1]), tokens[-1]] _CJK_RE = re.compile(r"[一-鿿]") def _contains_cjk(value: str) -> bool: return bool(_CJK_RE.search(value)) def _sanitize_token(token: str) -> str: """Lowercase and drop everything that's not [a-z0-9].""" return re.sub(r"[^a-z0-9]", "", token.lower()) def _resolve_email_overrides( person: IdentityPerson, contractor_codes: list[str], ) -> dict[str, str]: """Decide whether this person is contractor or FTE. Match key is ``person.company_code`` (SF EmpJob.company). An empty contractor list means everyone is FTE — the safe default that matches what ops asked for. """ is_contractor = bool( person.company_code and person.company_code in set(contractor_codes) ) return {"role": "contractor" if is_contractor else "fte"} def _candidate_local_part( local_part: str, attempt: int, *, max_local: int, ) -> str: """Build the email local part for collision ``attempt`` (0 = bare). Attempts 1-9 append a zero-padded counter ``01`` … ``09``. The base is capped to ``max_local - 2`` BEFORE the suffix is appended so the counter survives the length cap (result length == ``max_local`` for any realistic ``max_local >= 3``; the default is 30, Google's ceiling 64). Capping ``{local_part}{NN}`` to ``max_local`` AFTER the fact (the old behaviour) chopped the suffix back off when ``local_part`` was already at the cap — making every retry collide on the identical address and burning the whole suffix budget for nothing. """ if attempt == 0: return local_part[:max_local] base = local_part[: max(1, max_local - 2)] return f"{base}{attempt:02d}" def _filter_cc_emails( cc_emails_raw: list[Any], normalised_suffixes: list[str], ) -> tuple[list[str], list[str]]: """Allowlist + cap the welcome-email CC list. Returns ``(kept, dropped)``. The welcome email carries the first-login plaintext password, so a CC is kept ONLY when an allowlist is configured AND the address suffix matches it. Fail CLOSED: an empty/unset allowlist drops EVERY CC — we cannot prove any address is an internal, company-mailable domain, and silently waving them all through (the prior ``if normalised_suffixes and …`` form) defeated the guard on the default-empty setting. Kept list is de-duplicated (order-preserving) and hard-capped to 5. """ kept: list[str] = [] seen: set[str] = set() dropped: list[str] = [] for raw in cc_emails_raw: if not isinstance(raw, str): dropped.append(str(raw)) continue candidate = raw.strip().lower() if "@" not in candidate: dropped.append(raw) continue suffix = candidate.split("@", 1)[1] if not normalised_suffixes or not any( suffix == s or suffix.endswith("." + s) for s in normalised_suffixes ): dropped.append(raw) continue if candidate in seen: continue seen.add(candidate) kept.append(candidate) return kept[:5], dropped async def _resolve_welcome_cc_additions( db: AsyncSession, person: IdentityPerson, ) -> list[str]: """Always-on CC recipients for the new-hire welcome email. The welcome email announces the live mailbox (and carries the first-login plaintext password), so beyond the setting-driven ``notify_cc_emails`` list it is CC'd to the people who need to know the account was provisioned: * the hire's **HRBP** and **HR Ops** contacts — resolved from the HRIS-sourced relationship fields; and * every active asset administrator of the hire's local office, resolved with the canonical asset-office location rule; and * the **IT account-provisioning contact(s)** — 孟凡生 (``fansheng.meng@pingcap.com``) by default, configurable via the ``google.account_provisioning.created_notify_emails`` setting (the SAME setting the account-outcome notifier reads, so the contact is configured in one place). These are internally-resolved, trusted recipients, so they are gated on :func:`email_is_corporate` (which fails *open* to the built-in company domains — ``pingcap.com`` &c.) rather than on the setting-driven ``notify_cc_emails`` suffix allowlist (which fails *closed* to a possibly-empty setting). That keeps the "never leak the password to an external mailbox" invariant while guaranteeing the HRBP + IT contact are actually CC'd even when the CC allowlist setting is unconfigured. Returns lower-cased, de-duplicated addresses. """ raw: list[str] = [] from app.core.coordinators.new_hire_recipient_resolver import ( resolve_new_hire_operational_recipients, ) for recipient in await resolve_new_hire_operational_recipients(db, person): if recipient.work_email: raw.append(recipient.work_email) it_contacts = await get_platform_setting( "google.account_provisioning.created_notify_emails", default=["fansheng.meng@pingcap.com"], session=db, ) if isinstance(it_contacts, list): raw.extend(str(e) for e in it_contacts if str(e).strip()) corporate_domains = await get_corporate_email_domains(db) kept: list[str] = [] seen: set[str] = set() for candidate in raw: addr = candidate.strip().lower() if addr in seen or not email_is_corporate(addr, corporate_domains): continue seen.add(addr) kept.append(addr) return kept async def _get_google_integration( db: AsyncSession, ) -> Integration | None: """Resolve the primary enabled Google Workspace integration row. Returns None when either the row doesn't exist or is disabled — callers treat both as "skip silently, this is a config-not-yet" state, NOT as an error. """ row = (await db.execute( select(Integration) .where( Integration.kind == "software_provisioning", Integration.provider == "google_workspace", Integration.enabled.is_(True), ) .order_by(Integration.is_primary.desc(), Integration.created_at) )).scalars().first() return row async def _get_existing_google_identity( db: AsyncSession, person_id: UUID, ) -> IdentityExternalIdentity | None: """Return the active (non-tombstoned) Google external identity, if any. Used by the ``create_account_for`` short-circuits, which intentionally ignore tombstones: a soft-deleted row must NOT be treated as "already provisioned" (that would wrongly divert a rehire to manual review) — it is instead revived in place by :func:`_revive_or_insert_google_identity` further down the reconcile/mint path. """ return (await db.execute( select(IdentityExternalIdentity).where( IdentityExternalIdentity.person_id == person_id, IdentityExternalIdentity.provider == _GOOGLE_PROVIDER, IdentityExternalIdentity.deleted_at.is_(None), ), )).scalar_one_or_none() async def _revive_or_insert_google_identity( db: AsyncSession, person_id: UUID, *, google_user_id: str | None, primary_email: str, ) -> None: """Attach the ``google`` identity for ``person_id`` — reviving a soft-deleted tombstone in place instead of inserting a colliding row. ``uq_identity_external_person_provider`` (person_id, provider) does NOT consider ``deleted_at``. On a rehire whose prior google identity was tombstoned (offboarding / identity_sync soft-delete), a blind ``db.add(IdentityExternalIdentity(...))`` collides → ``IntegrityError`` (1062) → the shared async session is poisoned (``PendingRollbackError``), which both breaks ``create_account_for``'s documented "NEVER raises" contract AND corrupts every subsequent person in the same HR-sync batch. So we look up the google row WITHOUT the ``deleted_at`` filter (the UNIQUE constraint guarantees at most one ever exists) and revive it, mirroring the revive pattern in ``successfactors_sync`` / ``hr.service``. Caller is responsible for the ``flush()``. """ existing = (await db.execute( select(IdentityExternalIdentity).where( IdentityExternalIdentity.person_id == person_id, IdentityExternalIdentity.provider == _GOOGLE_PROVIDER, ), )).scalar_one_or_none() if existing is None: db.add(IdentityExternalIdentity( person_id=person_id, provider=_GOOGLE_PROVIDER, provider_subject=google_user_id or primary_email, provider_email=primary_email, provider_username=primary_email, is_login_identity=True, identity_status="active", )) return existing.identity_status = "active" existing.deleted_at = None existing.is_login_identity = True if google_user_id: existing.provider_subject = google_user_id existing.provider_email = primary_email existing.provider_username = primary_email async def _get_or_create_google_seat( db: AsyncSession, *, person_id: UUID, reason: str, ) -> UUID | None: """Ensure the person has an active SoftwareSeatAllocation against the seeded "Google Workspace" license plan. Returns the seat id, or None if the catalog/plan rows don't exist (migration 103 not applied — non-fatal, the auto-revoke chain just won't pick this person up). """ catalog = (await db.execute( select(AssetSoftwareCatalog).where( AssetSoftwareCatalog.name == _GOOGLE_CATALOG_NAME, ).limit(1), )).scalar_one_or_none() if catalog is None: return None plan = (await db.execute( select(SoftwareLicensePlan).where( SoftwareLicensePlan.software_id == catalog.id, SoftwareLicensePlan.status == "active", ) .order_by(SoftwareLicensePlan.created_at.asc()) .limit(1), )).scalar_one_or_none() if plan is None: return None # Round-49: include ``pending`` in the existing-seat lookup. # Migration 119's partial-unique-index ``uq_software_seat_active_pending`` # covers BOTH active AND pending; a stuck-pending row would # otherwise slip past this check and trip an IntegrityError on # ``db.flush()`` below — and the caller chain at # ``hr_sync_coordinator`` catches but doesn't rollback, poisoning # the next person's flush in the same HR-sync batch. existing = (await db.execute( select(SoftwareSeatAllocation).where( SoftwareSeatAllocation.plan_id == plan.id, SoftwareSeatAllocation.user_id == person_id, SoftwareSeatAllocation.status.in_(("active", "pending")), ), )).scalar_one_or_none() if existing is not None: return existing.id seat = SoftwareSeatAllocation( plan_id=plan.id, user_id=person_id, status="active", allocated_at=datetime.now(UTC), reason=reason, ) db.add(seat) await db.flush() return seat.id async def _link_google_identity( db: AsyncSession, person: IdentityPerson, *, google_user_id: str | None, primary_email: str, reason: str, ) -> UUID | None: """Attach (or re-activate) the ``google`` IdentityExternalIdentity for a person whose Google mailbox ALREADY exists, and ensure the seat row. This is the "reconcile / link" counterpart to minting a new account — used by both :func:`create_account_for` (reconcile-first) and the ``google.identity_reconcile`` cron. Idempotent: a re-run on an already linked person just re-stamps the row. ``work_email`` is realigned to the live Google primary so downstream SSO/notification routing is correct. Returns the seat id (or None if the catalog/plan rows are absent). """ primary_email = primary_email.strip().lower() await _revive_or_insert_google_identity( db, person.id, google_user_id=google_user_id, primary_email=primary_email, ) if (person.work_email or "").lower() != primary_email: person.work_email = primary_email await db.flush() return await _get_or_create_google_seat(db, person_id=person.id, reason=reason) _PERSONAL_EMAIL_SHAPE_RE = re.compile( r"^[A-Za-z0-9._%+\-]+@[A-Za-z0-9.\-]+\.[A-Za-z]{2,}$" ) # Corporate domains the welcome password MUST NOT be sent to — the # work mailbox doesn't exist yet (we just created it) and a corporate # domain on file as ``personal_email`` is by definition a data-entry # error. Sending the credential there ships it to a coworker or to # an internal mailing list. Defence-in-depth on top of the explicit # `personal_email != work_email` check below. _CORPORATE_PERSONAL_DENYLIST = frozenset({ "pingcap.com", "pingcap.cn", "pingcap.net", }) def _validate_personal_email_for_welcome( person: IdentityPerson, work_email: str, ) -> tuple[bool, str | None]: """Refuse to ship the first-login password to a malformed / duplicate / corporate ``personal_email``. Failure modes guarded: * Empty / whitespace — caller has nothing to send to. * Shape regex — typos like ``foo@bar`` or ``foo@bar.``. * Equals ``work_email`` — sending to the freshly-minted mailbox that doesn't have anyone reading it yet. * Domain on ``_CORPORATE_PERSONAL_DENYLIST`` — a coworker's mailbox or a mailing list at our own domain. Returns ``(ok, reason_if_not_ok)`` where the reason is a short machine-friendly string suitable for the audit row. Never echoes the work_email back to the caller in the reason (the audit log has a separate field for that). """ addr = (person.personal_email or "").strip().lower() if not addr: return False, "personal_email empty" if not _PERSONAL_EMAIL_SHAPE_RE.match(addr): return False, "personal_email shape invalid" if addr == (work_email or "").strip().lower(): return False, "personal_email equals work_email" domain = addr.rsplit("@", 1)[1] if "@" in addr else "" if domain in _CORPORATE_PERSONAL_DENYLIST: return False, "personal_email is on the corporate denylist" return True, None def _scrub_password_from_text(text: str, password: str) -> str: """Strip the plaintext password out of any error text before it leaves this module. Some email drivers echo the full request body (including the rendered welcome message) on a 5xx — without scrubbing, an ``error_message`` returned upstream would carry the password into ``audit_logs.detail.password_email_error`` permanently. """ if not text or not password: return text return text.replace(password, "***REDACTED***") async def _send_password_email( db: AsyncSession, *, person: IdentityPerson, work_email: str, initial_password: str, cc_emails: list[str], ) -> tuple[bool, str | None]: """Send the welcome-with-password message to the person's ``personal_email``. Falls back to logging a warning when no personal email is on file — IT will need to deliver out-of-band. We deliberately bypass ``NotificationService.dispatch`` because that helper hardcodes ``recipient.work_email`` as the destination, which is exactly the new mailbox that doesn't exist yet from the user's perspective. The returned ``reason`` is guaranteed to be password-free — every upstream error message is run through ``_scrub_password_from_text`` before it leaves the function so the password can never be persisted to ``audit_logs.detail`` or worker logs. """ ok, reason = _validate_personal_email_for_welcome(person, work_email) if not ok: return False, reason from app.integrations.driver_base import EmailDriver, EmailMessage from app.integrations.service import build_primary_driver try: driver = await build_primary_driver("email", session=db) except Exception as exc: # noqa: BLE001 — surface as soft failure return False, _scrub_password_from_text( f"email integration not configured: {exc}", initial_password, ) if not isinstance(driver, EmailDriver): return False, ( f"primary email integration is not an EmailDriver " f"({getattr(driver, 'provider', '?')})" ) # Copy comes from the ``google_account_password`` notification template # (``app/notifications/templates/email_group.py``) rather than the two # inline branches this used to carry. Two reasons: # # * The inline pair only spoke Chinese and English, so a Japanese hire got # the English credential mail. The template ships all three and # ``pick_localized`` walks ja → en → zh. # * The copy is now reviewable next to every other notification instead of # hiding inside the provisioner. # # Language selection stays on the shared resolver (``app.core.locale``): # explicit preference → country → labour-contract entity → baseline. # ``resolve_lang`` supersedes the ``is_en`` boolean this used to call — same # module, same normalisation of production's free-text SuccessFactors values # ("chinese" / "中文"), but it can also say "ja". Do NOT regress this to a # ``startswith("zh")`` test: that read those free-text values as *not* # Chinese and sent the one email carrying a temporary password in English # to people who had explicitly asked for Chinese. # # This send bypasses ``NotificationService.dispatch`` (see the docstring), # so the template is resolved and formatted here. Builtin-only: the # ``platform_settings`` override layer is deliberately not consulted, so an # admin edit can never break the one email that must render correctly. lang = resolve_lang(person) fmt = { "person_name": person.display_name or work_email, "work_email": work_email, "temp_password": initial_password, } try: subject = ( pick_localized(_PASSWORD_MAIL_TEMPLATE, "subject", lang) or "" ).format(**fmt) body_text = ( pick_localized(_PASSWORD_MAIL_TEMPLATE, "body_text", lang) or "" ).format(**fmt) except (KeyError, IndexError, ValueError) as exc: # A malformed template must not ship a half-substituted credential # mail (or worse, one with a literal ``{temp_password}``). Fail the # send as a soft failure — the caller leaves the mailbox suspended and # IT delivers out of band, which is the same path a bounce takes. return False, _scrub_password_from_text( f"password mail template is malformed: {type(exc).__name__}: {exc}", initial_password, ) msg = EmailMessage( to=[person.personal_email], cc=list(cc_emails or []), subject=subject, body_text=body_text, ) send_ok = False send_err: str | None = None try: result = await driver.send(msg) send_ok = bool(result.ok) if not result.ok: send_err = result.error_message or "email driver returned ok=False" except Exception as exc: # noqa: BLE001 send_err = f"email send raised: {type(exc).__name__}: {exc}" # Record the send (WITH the temporary password in ``body_text``) as a # terminal outbox row so it surfaces in the notification log — admins # need the password to hand-deliver when the personal-email send bounces # (兜底, 2026-07-02). ``last_error`` is scrubbed; the body deliberately # keeps the password (temp, forces change at first login). Best-effort: # a recording hiccup must not fail the provisioning outcome. # SAVEPOINT, not a bare ``except``: ``record_external_email`` inserts a # terminal outbox row and flushes, and it is NOT one of the # self-isolating NotificationService entry points. A swallowed flush # failure would deactivate the transaction, so the provisioning outcome # this "must not fail the provisioning outcome" comment protects would # be exactly what dies — on the caller's next statement. async with best_effort( db, label=f"record password mail for person {person.id}", log=logger, ) as rec_step: from app.notifications.service import NotificationService await NotificationService(db).record_external_email( recipient=person, recipient_email=person.personal_email, subject=subject, body_text=body_text, ok=send_ok, template_key="google_account_password", error=( _scrub_password_from_text(send_err, initial_password) if send_err else None ), related_resource_type="identity_person", related_resource_id=person.id, ) if not rec_step.ok: logger.warning( "record_external_email failed for password mail person=%s: %s", person.id, _scrub_password_from_text(rec_step.reason() or "", initial_password), ) if not send_ok: return False, _scrub_password_from_text( send_err or "email send failed", initial_password, ) return True, None async def _reprovision_pending_welcome( db: AsyncSession, person: IdentityPerson, *, existing_ident: IdentityExternalIdentity, integration: Integration, cc_emails: list[str], trace_id: str, actor_user_id: UUID | None, audit: AuditService, ) -> ProvisioningOutcome: """Re-run welcome-credential delivery for a ``pending_welcome`` account. Reached from the ``create_account_for`` short-circuit when the person's Google identity is in the ``pending_welcome`` self-heal state: the mailbox was minted earlier but left SUSPENDED because the welcome email (carrying the first-login password) failed to send. Also the terminal leg of the rehire heal (:func:`_reprovision_rehired_leaver`), which flips an offboarding-suspended binding to ``pending_welcome`` before delegating here — same suspended-mailbox-without-delivered-credential shape. Heal, preserving the create-time security invariant — *never leave an active mailbox with an undelivered password*: 1. regenerate a fresh temporary password (the earlier one was never delivered and is unrecoverable — that's fine, we mint a new one); 2. reset it in Google (force change-at-next-login, matching create); 3. re-send the welcome to the personal email; 4. on email success → unsuspend the mailbox and flip the identity back to ``active``; 5. on email failure → RE-suspend (belt-and-suspenders: the original safety-suspend may itself have failed) and KEEP ``pending_welcome`` so the next run re-attempts. The caller's top-of-function eligibility gates (``employment_status == 'active'``, not a future hire, ``auto_create_on_hire``) have already passed, so this only ever runs for a genuine active hire — a leaver (``employment_status != 'active'``) is skipped long before here and can NEVER be re-provisioned/unsuspended by this path. Never raises — returns a structured :class:`ProvisioningOutcome` like the rest of the module. """ gemail = ( existing_ident.provider_email or person.work_email or existing_ident.provider_subject or "" ).strip().lower() if not gemail: return ProvisioningOutcome( status="failed", reason="pending_welcome identity has no resolvable email", ) resolved = await _resolve(db, integration) driver = GoogleWorkspaceDriver( integration_id=str(integration.id), config=resolved.config, ) # Confirm the mailbox still exists (and read its current suspend state). # A lookup error must NOT fall through to "email delivered" — surface it. try: profile = await driver.get_user_profile(gemail) except Exception as exc: # noqa: BLE001 await audit.log( trace_id=trace_id, actor_type="user" if actor_user_id else "system", actor_user_id=actor_user_id, action="google_workspace.reprovision_lookup_failed", status="failed", resource_type="identity_person", resource_id=str(person.id), detail={"primary_email": gemail, "error": str(exc)}, ) return ProvisioningOutcome( status="failed", primary_email=gemail, reason=f"google directory lookup failed for {gemail}: {exc}", ) if profile is None: # The suspended mailbox we expected to heal is gone (deleted in the # Google console out-of-band). Do NOT silently re-mint here — surface # for manual review so an admin decides (recreate vs. investigate). return ProvisioningOutcome( status="skipped", primary_email=gemail, reason=( f"pending_welcome mailbox {gemail} no longer exists in Google " "(deleted out-of-band); manual review required" ), ) google_user_id = profile.get("id") or existing_ident.provider_subject was_suspended = bool(profile.get("suspended")) # Fresh temporary password — force rotation at first login. new_password = GoogleWorkspaceDriver._generate_strong_password() reset_res = await driver.reset_user_password( user_email=gemail, new_password=new_password, change_password_at_next_login=True, ) if not reset_res.ok: reset_err = _scrub_password_from_text( reset_res.error_message or "password reset returned ok=False", new_password, ) await audit.log( trace_id=trace_id, actor_type="user" if actor_user_id else "system", actor_user_id=actor_user_id, action="google_workspace.reprovision_reset_failed", status="failed", resource_type="identity_person", resource_id=str(person.id), detail={"primary_email": gemail, "error": reset_err}, ) return ProvisioningOutcome( status="failed", primary_email=gemail, google_user_id=google_user_id, reason=f"password reset failed for {gemail}: {reset_err}", ) # Re-send the welcome with the fresh password. email_ok, email_err = await _send_password_email( db, person=person, work_email=gemail, initial_password=new_password, cc_emails=cc_emails, ) from app.integrations.driver_base import ProvisioningContext if not email_ok: # Still undeliverable — RE-suspend (in case the original safety-suspend # failed, leaving the mailbox live) and keep pending_welcome. The next # run re-attempts. Best-effort suspend; never raise. resuspend_ok: bool | None = None resuspend_err: str | None = None resuspend_raw: dict[str, Any] | None = None try: # must_exist: ``get_user_profile`` above already proved this # mailbox exists (a None profile returned long before here), # so a 404 now is not "already gone" — it must not be audited # as a successful re-suspend. sres = await driver._set_suspended( ProvisioningContext( user_email=gemail, user_display_name=person.display_name, ), suspended=True, must_exist=True, ) resuspend_raw = sres.raw_response resuspend_ok = bool(sres.ok) if not resuspend_ok: resuspend_err = _scrub_password_from_text( sres.error_message or "suspend returned ok=False", new_password, ) except Exception as exc: # noqa: BLE001 resuspend_ok = False resuspend_err = _scrub_password_from_text( f"{type(exc).__name__}: {exc}", new_password, ) # Leave identity_status = pending_welcome unchanged. await audit.log( trace_id=trace_id, actor_type="user" if actor_user_id else "system", actor_user_id=actor_user_id, action="google_workspace.reprovision_email_failed", status="partial", resource_type="identity_person", resource_id=str(person.id), detail={ "primary_email": gemail, "password_email_error": email_err, "resuspend_ok": resuspend_ok, "resuspend_error": resuspend_err, # See failure_rollback_suspend_result: keep the driver's # verdict payload so the branches stay distinguishable. "resuspend_result": resuspend_raw, }, ) return ProvisioningOutcome( status="failed", primary_email=gemail, google_user_id=google_user_id, reason=( "re-provision welcome email failed again; mailbox kept " f"suspended (pending_welcome): {email_err}" ), ) # Email delivered → unsuspend (if needed) then promote to active. If the # unsuspend itself fails the credential is out but the mailbox stays # locked: keep pending_welcome and report failed so the next run retries # (a healthy Google will accept the unsuspend then). unsuspend_ok: bool | None = None if was_suspended: try: ures = await driver._set_suspended( ProvisioningContext( user_email=gemail, user_display_name=person.display_name, ), suspended=False, ) unsuspend_ok = bool(ures.ok) except Exception as exc: # noqa: BLE001 unsuspend_ok = False logger.warning( "google.account_provisioner: unsuspend failed during " "re-provision for %s: %s", gemail, exc, ) if not unsuspend_ok: await audit.log( trace_id=trace_id, actor_type="user" if actor_user_id else "system", actor_user_id=actor_user_id, action="google_workspace.reprovision_unsuspend_failed", status="partial", resource_type="identity_person", resource_id=str(person.id), detail={"primary_email": gemail}, ) return ProvisioningOutcome( status="failed", primary_email=gemail, google_user_id=google_user_id, reason=( "re-provision delivered the welcome email but UNSUSPEND " f"FAILED for {gemail} — mailbox still suspended; re-run " "once Google is healthy or unsuspend in the admin console." ), ) existing_ident.identity_status = "active" existing_ident.is_login_identity = True if google_user_id: existing_ident.provider_subject = google_user_id existing_ident.provider_email = gemail if (person.work_email or "").lower() != gemail: person.work_email = gemail await db.flush() seat_id = await _get_or_create_google_seat( db, person_id=person.id, reason=f"re-provisioned pending_welcome Google account ({gemail})", ) # Report the (now live) address back to HRIS, mirroring the create path. hris_notify = await notify_hris_email_assigned( db, person=person, work_email=gemail, trace_id=trace_id, ) await audit.log( trace_id=trace_id, actor_type="user" if actor_user_id else "system", actor_user_id=actor_user_id, action="google_workspace.account_reprovisioned", status="success", resource_type="identity_person", resource_id=str(person.id), detail={ "primary_email": gemail, "google_user_id": google_user_id, "was_suspended": was_suspended, "unsuspend_ok": unsuspend_ok, "seat_id": str(seat_id) if seat_id else None, "hris_notify": hris_notify, }, ) return ProvisioningOutcome( status="created", primary_email=gemail, google_user_id=google_user_id, seat_id=seat_id, hris_notify_status=hris_notify.get("status"), reason="re-provisioned after earlier welcome-email failure", ) async def _reprovision_rehired_leaver( db: AsyncSession, person: IdentityPerson, *, existing_ident: IdentityExternalIdentity, integration: Integration, cc_emails: list[str], trace_id: str, actor_user_id: UUID | None, audit: AuditService, ) -> ProvisioningOutcome | None: """Heal a rehired employee's Google account after an offboarding suspend. Reached from the ``create_account_for`` short-circuit when the person is active again (rehire: ``leaver_pending``/``left`` → ``active`` — the P01485 jiong.li 2026-07-08 gap) but their ``google`` external identity is still ``inactive`` — the state leaver hygiene (``identity_reconcile.deactivate_login_identities``) left it in when offboarding auto-suspended the mailbox. Pre-fix this fell into the blanket "manual review required" skip, so a rehire kept a suspended mailbox and a dead SSO binding forever with no automated recovery. ``inactive`` ONLY: ``suspended`` is the deliberate per-person admin freeze (``identity/profile_service``) and must never be undone by a lifecycle event; any other status stays at manual review too. Dispositions by live Google state: * mailbox SUSPENDED → flip the identity to ``pending_welcome`` and route through :func:`_reprovision_pending_welcome` — fresh temporary password, welcome re-send, unsuspend ONLY on email success. A failed heal therefore lands in exactly the welcome-failed state the existing self-heal machinery owns (seat protected by ``google.seat_sync``, next ``create_account_for`` run retries). * mailbox LIVE (admin already unsuspended by hand) → just restore the Helix-side binding (``active`` + ``is_login_identity``) so Google SSO works again, and re-ensure the base seat. No password rotation. * mailbox GONE (delete-sweep ran during the gap) → skip to manual review; never silently re-mint over a rehire. Returns ``None`` to decline — no resolvable email, or the address is registered as a shared / service / retained account in the external-account registry (the same classification set offboarding's suspend exempts): a mailbox the person doesn't personally own must never be unsuspended or have its password rotated by a rehire event. The caller falls through to the manual-review skip. Never raises — same contract as the rest of the module. """ gemail = ( existing_ident.provider_email or person.work_email or existing_ident.provider_subject or "" ).strip().lower() if not gemail or "@" not in gemail: return None # Ownership guard: shared / service / retained mailboxes are classified # in ExternalAccountRegistry — offboarding never auto-suspends those # (``_is_google_account_retained``), and the symmetric rule holds here. from app.domains.asset.models import ( # noqa: PLC0415 EXTERNAL_ACCOUNT_KNOWN_CLASSIFICATIONS, ExternalAccountRegistry, ) registry_row = (await db.execute( select(ExternalAccountRegistry).where( func.lower(ExternalAccountRegistry.email) == gemail, ), )).scalars().first() if ( registry_row is not None and registry_row.classification in EXTERNAL_ACCOUNT_KNOWN_CLASSIFICATIONS ): return ProvisioningOutcome( status="skipped", primary_email=gemail, reason=( f"rehire: {gemail} is registered as " f"{registry_row.classification!r} in the external-account " "registry (shared/service/retained) — not auto-touching; " "manual review required" ), ) resolved = await _resolve(db, integration) driver = GoogleWorkspaceDriver( integration_id=str(integration.id), config=resolved.config, ) try: profile = await driver.get_user_profile(gemail) except Exception as exc: # noqa: BLE001 await audit.log( trace_id=trace_id, actor_type="user" if actor_user_id else "system", actor_user_id=actor_user_id, action="google_workspace.rehire_reprovision_lookup_failed", status="failed", resource_type="identity_person", resource_id=str(person.id), detail={"primary_email": gemail, "error": str(exc)}, ) return ProvisioningOutcome( status="failed", primary_email=gemail, reason=f"rehire: google directory lookup failed for {gemail}: {exc}", ) if profile is None: # The prior mailbox is gone (offboarding delete-sweep / console # delete during the employment gap). Do NOT silently re-mint over # the existing identity row — surface for manual review so an admin # decides (recreate vs. restore-from-trash vs. investigate). return ProvisioningOutcome( status="skipped", primary_email=gemail, reason=( f"rehire: prior mailbox {gemail} no longer exists in Google; " "manual review required" ), ) google_user_id = profile.get("id") or existing_ident.provider_subject if not profile.get("suspended"): # Mailbox already live — someone unsuspended it by hand, or the # offboarding suspend never landed. Restore only the Helix-side # binding (auth hard-rejects any non-active login binding, so # without this the rehired employee cannot Google-SSO into Helix) # and re-ensure the base seat. No credential rotation needed. existing_ident.identity_status = "active" existing_ident.is_login_identity = True if google_user_id: existing_ident.provider_subject = google_user_id existing_ident.provider_email = gemail await db.flush() seat_id = await _get_or_create_google_seat( db, person_id=person.id, reason=f"rehire relink of live Google account ({gemail})", ) await audit.log( trace_id=trace_id, actor_type="user" if actor_user_id else "system", actor_user_id=actor_user_id, action="google_workspace.rehire_identity_relinked", status="success", resource_type="identity_person", resource_id=str(person.id), detail={ "primary_email": gemail, "google_user_id": google_user_id, "seat_id": str(seat_id) if seat_id else None, }, ) return ProvisioningOutcome( status="linked", primary_email=gemail, google_user_id=google_user_id, seat_id=seat_id, reason="rehire: relinked existing live Google account", ) # Suspended by offboarding → hand over to the pending_welcome heal. # Flip the marker FIRST so a mid-heal crash / email failure leaves the # account in the canonical welcome-failed state (suspended mailbox + # undelivered credential) instead of a state no automation owns. existing_ident.identity_status = _STATUS_PENDING_WELCOME await db.flush() await audit.log( trace_id=trace_id, actor_type="user" if actor_user_id else "system", actor_user_id=actor_user_id, action="google_workspace.rehire_unsuspend_started", status="success", resource_type="identity_person", resource_id=str(person.id), detail={ "primary_email": gemail, "google_user_id": google_user_id, "prior_identity_status": "inactive", }, ) return await _reprovision_pending_welcome( db, person, existing_ident=existing_ident, integration=integration, cc_emails=cc_emails, trace_id=trace_id, actor_user_id=actor_user_id, audit=audit, ) async def _known_taken_emails( db: AsyncSession, domain: str, *, exclude_person_id: UUID, ) -> set[str]: """Lowercased set of corporate addresses at ``domain`` already taken in Helix — every person's ``work_email`` plus every Google identity's ``provider_email``. Used for best-effort collision avoidance when RESERVING a planned address ahead of actual Google creation. Other pending reservations are included automatically because they were written into ``work_email`` too, so two same-named future hires get distinct planned addresses. Google remains authoritative at creation.""" suffix = f"@{domain}" taken: set[str] = set() rows = await db.execute( select(IdentityPerson.work_email).where( IdentityPerson.work_email.is_not(None), IdentityPerson.id != exclude_person_id, ) ) for (email,) in rows: if email and email.lower().endswith(suffix): taken.add(email.lower()) rows2 = await db.execute( select(IdentityExternalIdentity.provider_email).where( IdentityExternalIdentity.provider == _GOOGLE_PROVIDER, IdentityExternalIdentity.provider_email.is_not(None), ) ) for (email,) in rows2: if email and email.lower().endswith(suffix): taken.add(email.lower()) return taken async def reserve_work_email_for( db: AsyncSession, person_id: UUID, *, trace_id: str, email_name: str | None = None, ) -> ProvisioningOutcome: """Compute the PLANNED corporate email for a (possibly future-dated) hire WITHOUT creating the live Google account, persist it to ``person.work_email``, and write it back to SuccessFactors PerEmail. Decouples "tell HRIS the address" (fire the moment the onboard event arrives, even weeks before the start date) from "activate the mailbox" (still deferred to ``hire_date`` by :func:`create_account_for`). So HR sees the corporate email before day 1 without us minting a live account / consuming a Google license / leaving an account behind for a no-show. The write-back here is ADDRESS-ONLY (``is_primary_override=False``): SF records the corporate address but keeps the personal email as primary until the mailbox actually goes live. ``create_account_for`` promotes it to primary (and demotes the personal email) at hire-date creation — see the two-step note in ``hris_email_writeback``. The address is collision-checked against Helix's known addresses (best-effort). Google is authoritative at actual creation: if a real collision forces a different address on ``hire_date``, ``create_account_for`` re-writes the corrected value to SF (its writeback is unconditional). Gated on ``auto_create_on_hire`` so we never pre-announce an address for an org that provisions manually. Never raises (mirrors ``create_account_for``).""" person = await db.get(IdentityPerson, person_id) if person is None or person.deleted_at is not None: return ProvisioningOutcome(status="skipped", reason="person not found") # HR test-record band (P99XX): never reserve — and write back to SF! — # an address for one of HR's own test records (confirmed 2026-07-13). # Same gate as ``_create_account_for_locked``. from app.core.person_filters import is_test_employee_no # noqa: PLC0415 if is_test_employee_no(person.employee_no) or is_test_employee_no( person.hris_person_id ): return ProvisioningOutcome( status="skipped", reason=( f"employee_no={person.employee_no!r} is in HR's reserved " "P99XX test band — not reserving an address" ), ) if person.employment_status != "active": return ProvisioningOutcome( status="skipped", reason=f"employment_status={person.employment_status}", ) # Mirror of the purged-tombstone gate in ``_create_account_for_locked`` # — reserving (and writing back to SF!) an address for a record HRIS # already deleted is pure noise. if person.hris_record_state == "purged": return ProvisioningOutcome( status="skipped", reason=( "hris_record_state='purged' — the HRIS record was deleted " "at the source; not reserving an address" ), ) # Same duplicate-employment gate as ``_create_account_for_locked``. # This path is irreversible in its own right: it writes the reserved # address back into SF ``PerEmail`` — and it resolves the target by # ``personIdExternal``, so for a 兼岗 it would stamp the address onto # the CANONICAL person's HR record. That is exactly the corruption # #3434 produced (战捷's business email replaced by a dead # colleague's address), so the refusal has to sit here too, not only # in front of the mailbox. from app.identity.sf_assignment import ( # noqa: PLC0415 duplicate_employment_for, ) _dup = await duplicate_employment_for(db, person) if _dup is not None: logger.warning( "google.account_provisioner: refusing to reserve/write back an " "address for person=%s hris=%s — %s (signal=%s)", person.id, person.hris_person_id, _dup.detail, _dup.signal, ) return ProvisioningOutcome( status="skipped", reason=f"duplicate employment ({_dup.signal}): {_dup.detail}", ) settings = await get_platform_setting( _PROVISIONING_SETTING, default={}, session=db, ) or {} if not isinstance(settings, dict): settings = {} if not settings.get("auto_create_on_hire", False): return ProvisioningOutcome( status="skipped", reason="auto_create_on_hire is False", ) # Already has a live Google account → address is final; nothing to plan. if await _get_existing_google_identity(db, person.id) is not None: return ProvisioningOutcome( status="skipped", reason="google identity already exists", ) contractor_codes = await get_platform_setting( _CONTRACTOR_CODES_SETTING, default=[], session=db, ) or [] if not isinstance(contractor_codes, list): contractor_codes = [] role = _resolve_email_overrides(person, contractor_codes)["role"] role_settings = settings.get(role) or {} domain = (role_settings.get("domain") or "").strip().lower() if not domain: return ProvisioningOutcome( status="failed", reason=f"no domain configured for role={role}", ) max_local = int(settings.get("local_part_max_length") or 30) corporate_domains = await get_corporate_email_domains(db) # Derivation precedence (highest first): # 0. ``email_name`` from the inbound event — the HRIS side's # explicitly-requested mailbox name (bare local part OR full # address; routed through the same extractor so both work). # 0b. ``person.hris_email_name`` — the same value as persisted on the # golden record by a previous sync tick. Lets a re-reserve (the # ``hr.email_name_reconcile`` cron, or any caller that has a person # but no event in hand) reach HR's choice without re-reading SF. # 1. HRIS-provided corporate email (SF ``User.email`` → work_email). # 2. given.family pinyin. 3. employee_no. local_part = "" if email_name: local_part = _local_part_from_hris_email( email_name, max_length=max_local, corporate_domains=corporate_domains, ) if not local_part: local_part = _local_part_from_hris_email( person.hris_email_name, max_length=max_local, corporate_domains=corporate_domains, ) if not local_part: local_part = _local_part_from_hris_email( person.work_email, max_length=max_local, corporate_domains=corporate_domains, ) if not local_part: local_part = _local_part_from_name( display_name=person.display_name, name_pinyin=person.name_pinyin, employee_no=person.employee_no, max_length=max_local, first_name=person.first_name, last_name=person.last_name, ) if not local_part: return ProvisioningOutcome( status="failed", reason="could not derive local part", ) taken = await _known_taken_emails(db, domain, exclude_person_id=person.id) planned: str | None = None for attempt in range(_MAX_COLLISION_ATTEMPTS + 1): cand = f"{_candidate_local_part(local_part, attempt, max_local=max_local)}@{domain}".lower() if cand not in taken: planned = cand break if planned is None: return ProvisioningOutcome( status="failed", reason="no free planned address (collision budget exhausted)", ) if (person.work_email or "").lower() != planned: person.work_email = planned await db.flush() # Address-only write: REGISTER the planned corporate email on SF now, # but do NOT promote it to primary (nor demote the personal email) — # the mailbox doesn't exist until hire-date creation. ``create_account_for`` # flips the same PerEmail row to primary once the mailbox is live. notify = await notify_hris_email_assigned( db, person=person, work_email=planned, trace_id=trace_id, is_primary_override=False, ) return ProvisioningOutcome( status="ok", reason=f"reserved {planned}; sf_writeback={notify.get('status')}", ) def _is_phone_error(message: str | None) -> bool: """True when a Google ``users.insert`` rejection is about the recovery / home phone number. ``to_e164`` produces a syntactically-valid ``+`` string, but Google additionally rejects numbers it can't treat as a real recovery phone (e.g. a Malaysia national number that is E.164-shaped but not a valid mobile — 2026-07-02 "Google API error 400: Invalid recovery phone."). The recovery phone is an OPTIONAL convenience field, so a bad one must never block mailbox creation: detect the phone rejection and retry the same address without it. """ if not message: return False return "phone" in message.lower() async def create_account_for( db: AsyncSession, person_id: UUID, *, trace_id: str, actor_user_id: UUID | None = None, ) -> ProvisioningOutcome: """The single entry-point. See module docstring for the full flow. #1795: serialized per-person with a Redis advisory lock. Without it, the HRIS confirm-batch hook (``hr_sync_coordinator.provision_google_ account``) and the hourly ``hr_lifecycle.hire_day`` cron (or a manual trigger) can run for the SAME person concurrently — both pass the ``_get_existing_google_identity`` short-circuit (no identity row yet), both call ``create_user``, the loser hits a 409 → the collision loop mints a SECOND real mailbox (``zhang.wei01@``), then its ``UNIQUE(person_id, provider)`` flush fails and orphans that live Google account (the module's own comment at ~1408 documents this). Mirrors the canonical per-requester lock pattern (``domain_registration.submit`` / ``apikey.submit``). """ from app.core.distributed_lock import ( # noqa: PLC0415 LockBusy, distributed_lock, ) try: # auto_renew: the critical section spans several Google Directory API # calls plus a Gmail send. Without renewal a section exceeding the # 180s TTL lets the Redis key lapse mid-work, so a concurrent # coordinator+cron run could enter and double-mint (the exact orphan # this lock exists to prevent; reconcile-first mitigates but only when # the lock hasn't already lapsed). auto_renew keeps the single-flight # guarantee for the whole section — same mechanism the identity-sync # sweep uses. (Pairs with the renew-deadline hardening in #1896.) async with distributed_lock( f"google_account_provision:{person_id}", ttl=180, auto_renew=True, ): return await _create_account_for_locked( db, person_id, trace_id=trace_id, actor_user_id=actor_user_id, ) except LockBusy: # Another provisioning run for this person is mid-flight. Skip # benignly — the winner creates the account, and the next re-fire of # this path (idempotent) finds it via the existence short-circuit. return ProvisioningOutcome( status="skipped", reason=( "another Google account provisioning run for this person is " "in progress (per-person lock held) — will reconcile next tick" ), ) async def _create_account_for_locked( db: AsyncSession, person_id: UUID, *, trace_id: str, actor_user_id: UUID | None = None, ) -> ProvisioningOutcome: """Body of ``create_account_for``, run under the per-person lock.""" audit = AuditService(db) person = await db.get(IdentityPerson, person_id) if person is None or person.deleted_at is not None: return ProvisioningOutcome(status="skipped", reason="person not found") # ── Eligibility gates ────────────────────────────────────── # HR test-record band (P99XX): HR reserves P9900-P9999 for their own # SF test records (confirmed 2026-07-13). Never mint a real mailbox # for one — defence-in-depth alongside the ingest-time skip # (``HrSyncService.should_skip_user``), covering any test row that # predates the ingest gate or arrives via a non-SF path. from app.core.person_filters import is_test_employee_no # noqa: PLC0415 if is_test_employee_no(person.employee_no) or is_test_employee_no( person.hris_person_id ): return ProvisioningOutcome( status="skipped", reason=( f"employee_no={person.employee_no!r} is in HR's reserved " "P99XX test band — refusing to create a Google account" ), ) if person.employment_status != "active": return ProvisioningOutcome( status="skipped", reason=f"employment_status={person.employment_status}", ) # Purged-tombstone gate (2026-07-11 roger.zhou02 incident): the # ``hris.absent_person_check`` cron stamps ``hris_record_state= # 'purged'`` when the HRIS row vanishes from the source feed (HR # deleted a botched entry) but deliberately leaves employment_status # 'active'. The status gate above therefore does NOT catch it, and # the hire-day cron happily minted a live mailbox — with the deleted # record's cloned junk address — for a person HRIS no longer knows. # Every identity-sync target already skips purged rows; the account # minter must too. Re-entry in HRIS flips the state back to # 'present' and provisioning resumes without manual action. if person.hris_record_state == "purged": return ProvisioningOutcome( status="skipped", reason=( "hris_record_state='purged' — the HRIS record was deleted " "at the source; refusing to create a Google account for a " "tombstoned person" ), ) # ``hire_date`` gate (mirror of offboarding's leaver_date gate # added in PR for the 2026-05-28 incident). HR enters new hires # into SuccessFactors weeks ahead of their actual start date — # the HRIS sync flips employment_status to 'active' on confirm, # but the hire_date can still be in the future. Without this # gate, the welcome email (which carries a plaintext first-login # password) goes out immediately; the Google account sits live # for weeks before the person walks in the door. Defer via the # SHARED ``is_future_hire`` gate (employee-local TZ + the # HR-configurable ``onboarding.provision_lead_hours`` lead), so this # internal check agrees exactly with the coordinator that called us # — otherwise the coordinator could pass its gate while we still # skip, or vice-versa. The hourly ``hr_lifecycle.hire_day`` cron # re-fires this path so a deferred hire is created within the lead # window before their local start date. if person.hire_date is not None: from app.core.working_hours import get_provision_lead_hours from app.core.working_hours import is_future_hire as _is_future_hire _lead_hours = await get_provision_lead_hours(db) if _is_future_hire(person, lead_hours=_lead_hours): return ProvisioningOutcome( status="skipped", reason=( f"hire_date {person.hire_date.isoformat()} is more than " f"{_lead_hours}h in the future (employee-local TZ). " f"Account creation will be re-attempted within the " f"{_lead_hours}h lead window by the " "hr_lifecycle.hire_day cron." ), ) # ── Duplicate-employment gate (2026-09-04, #3228 + #3434) ── # A 兼岗 has now been minted into a real mailbox TWICE: Hiroki # Shinagawa's became ``hiroki.shinagawa01@pingcap.com``, and 战捷's # landed on a purged leaver's row and was provisioned at that dead # colleague's address (which Helix then wrote back into SF as 战捷's # primary business email). Both root causes were real and both fixes # were ingest-side — and an ingest guard only ever covers the paths # it is wired into. What was missing is a check at the point of no # return: nothing here ever asked "is this person a second # employment of somebody we already know?" before minting. # # Asked here it is path-independent, and the cost is irrelevant — # account creation runs a handful of times a day, and this sits # AFTER every cheap local gate so a skipped hire pays nothing. # Fails OPEN on any SF failure: a broken lookup must never block a # genuine new hire (#191 ate two real ones that way); only an actual # answer from SF stops us. from app.identity.sf_assignment import ( # noqa: PLC0415 duplicate_employment_for, ) _dup = await duplicate_employment_for(db, person) if _dup is not None: await audit.log( trace_id=trace_id, actor_type="user" if actor_user_id else "system", actor_user_id=actor_user_id, action="google_workspace.account_refused_duplicate_employment", status="success", resource_type="identity_person", resource_id=str(person.id), detail={ "signal": _dup.signal, "reason": _dup.detail, "canonical_person_id": ( str(_dup.owner_id) if _dup.owner_id else None ), "canonical_hris_person_id": _dup.owner_hris_person_id, "person_id_external": _dup.person_id_external, "hris_person_id": person.hris_person_id, }, ) logger.warning( "google.account_provisioner: refusing to mint a mailbox for " "person=%s hris=%s — %s (signal=%s)", person.id, person.hris_person_id, _dup.detail, _dup.signal, ) return ProvisioningOutcome( status="skipped", reason=f"duplicate employment ({_dup.signal}): {_dup.detail}", ) settings = await get_platform_setting( _PROVISIONING_SETTING, default={}, session=db, ) or {} if not isinstance(settings, dict): settings = {} if not settings.get("auto_create_on_hire", False): return ProvisioningOutcome( status="skipped", reason="auto_create_on_hire is False", ) contractor_codes = await get_platform_setting( _CONTRACTOR_CODES_SETTING, default=[], session=db, ) or [] if not isinstance(contractor_codes, list): contractor_codes = [] role = _resolve_email_overrides(person, contractor_codes)["role"] role_settings = settings.get(role) or {} domain = (role_settings.get("domain") or "").strip().lower() if not domain: return ProvisioningOutcome( status="failed", reason=f"no domain configured for role={role}", ) org_unit_path = (role_settings.get("org_unit_path") or "/").strip() or "/" max_local = int(settings.get("local_part_max_length") or 30) cc_emails_raw = settings.get("notify_cc_emails") or [] if not isinstance(cc_emails_raw, list): cc_emails_raw = [] # Allowlist + cap the CC list — the welcome email carries the # first-login plaintext password, so a malformed setting value # ("hr@partner.example.com" pasted into the CC field) would # broadcast the new-hire password to an external address. Suffix # allowlist mirrors ``email_group.suffix_whitelist`` (the same # company-mailable-domains rule), and we hard-cap to 5 CCs. allowed_suffixes = await get_platform_setting( "google.mail_group.suffix_whitelist", default=[], session=db, ) or [] if not isinstance(allowed_suffixes, list): allowed_suffixes = [] normalised_suffixes = [ str(s).strip().lower() for s in allowed_suffixes if str(s).strip() ] cc_emails, dropped_cc = _filter_cc_emails(cc_emails_raw, normalised_suffixes) if dropped_cc: logger.warning( "google.account_provisioner: dropping %d non-allowlisted " "cc_emails (suffix not in google.mail_group.suffix_whitelist; " "empty allowlist fails closed)", len(dropped_cc), ) # Always CC the hire's HRBP, HR Ops, local-office asset admins, and IT # account-provisioning contact (孟凡生 by default) so they know the mailbox # is live. These are # internally-resolved and gated on the fail-OPEN corporate-domain # guard (see ``_resolve_welcome_cc_additions``), so they survive even # when the ``notify_cc_emails`` suffix allowlist above is unconfigured. # INTERNAL-FIRST ordering (2026-07-16): the guarantee used to be # defeated by the cap — the internal always-CCs were appended AFTER # the setting-driven list and then the union truncated to 5, so configured # extras could silently evict operational contacts. Build the # list internal-first and cap only the setting-driven extras (which # ``_filter_cc_emails`` already bounds to 5), so the total stays # bounded at internal + 5 and the internal recipients ALWAYS fit. internal_cc = await _resolve_welcome_cc_additions(db, person) extras = [addr for addr in cc_emails if addr not in internal_cc] cc_emails = internal_cc + extras[:5] # Already linked? Treat as "skipped" with a useful reason; the # offboarding flow will still find them via the existing identity. # # ``_get_existing_google_identity`` returns ANY non-deleted google # identity, and the DB enforces ``UNIQUE(person_id, provider)`` # (uq_identity_external_person_provider). So we must short-circuit for # EVERY non-deleted identity, not only ``active`` — otherwise a # non-active-but-undeleted google row (a state auth/service.py already # recognises) would fall through, mint a SECOND live mailbox, then # crash on ``db.flush()`` with an IntegrityError — orphaning the # freshly-created Google account (the module contract is "NEVER raises"). existing_ident = await _get_existing_google_identity(db, person.id) if existing_ident is not None: existing_email = ( existing_ident.provider_email or existing_ident.provider_subject ) if existing_ident.identity_status == "active": seat_id = await _get_or_create_google_seat( db, person_id=person.id, reason=( f"auto-linked seat for existing Google identity " f"{existing_email}" ), ) return ProvisioningOutcome( status="skipped", reason="person already has an active Google identity", primary_email=existing_email, seat_id=seat_id, ) if existing_ident.identity_status == _STATUS_PENDING_WELCOME: # Self-heal: the mailbox was minted but left SUSPENDED because the # first welcome email failed. The eligibility gates above already # confirmed this is an active (non-leaver), non-future hire, so # RE-PROVISION — regenerate the password, re-send the welcome, and # unsuspend ONLY on email success (preserving the create-time # invariant). ``cc_emails`` was computed above; ``integration`` is # fetched here (the create path fetches it below the short-circuit). reprov_integration = await _get_google_integration(db) if reprov_integration is None: return ProvisioningOutcome( status="skipped", reason="no enabled google_workspace integration", primary_email=existing_email, ) return await _reprovision_pending_welcome( db, person, existing_ident=existing_ident, integration=reprov_integration, cc_emails=cc_emails, trace_id=trace_id, actor_user_id=actor_user_id, audit=audit, ) if existing_ident.identity_status == "inactive": # Rehire heal (P01485 jiong.li 2026-07-08): ``inactive`` is the # state leaver hygiene leaves the binding in when offboarding # auto-suspends the mailbox. The eligibility gates above already # proved this person is active again (a rehire) — reverse the # offboarding suspend instead of dead-ending at manual review. # (``suspended`` = deliberate admin freeze; stays manual below.) rehire_integration = await _get_google_integration(db) if rehire_integration is not None: rehire_outcome = await _reprovision_rehired_leaver( db, person, existing_ident=existing_ident, integration=rehire_integration, cc_emails=cc_emails, trace_id=trace_id, actor_user_id=actor_user_id, audit=audit, ) if rehire_outcome is not None: return rehire_outcome # Non-active but undeleted: a previously-provisioned mailbox that # was suspended/disabled (not soft-deleted). Do NOT mint a new # account — surface for manual review so an admin reactivates the # existing identity instead of creating a duplicate. return ProvisioningOutcome( status="skipped", reason=( f"person has a non-active Google identity " f"(status={existing_ident.identity_status!r}); " f"manual review required" ), primary_email=existing_email, ) integration = await _get_google_integration(db) if integration is None: return ProvisioningOutcome( status="skipped", reason="no enabled google_workspace integration", ) # Derivation precedence at MINT time — highest first: # 0. ``person.hris_email_name`` — the 邮箱名 HR assigned in SF # (PerPersonal.customString4), kept on the golden record by # ``HrSyncService._upsert_person`` and refreshed on every sync tick. # 1. ``person.work_email`` — the address reserved when the onboard # event first arrived. # 2. given.family pinyin. # # Why 0 must beat 1 (P10026 Nicholas Muniz, 2026-08-31 — the whole point # of this ordering): the reservation in (1) is computed ONCE, the first # time Helix sees the person, from whatever HR had filled in by then. # HR filled ``customString4`` two days later. Every 4-hourly sync in the # nine days between that and the hire date # that carried ``nick.muniz``, but the reserve coordinator only fires on # ``action in {create, reactivate}`` so the reservation stayed frozen at # the ``given.family`` fallback — and this function, reading only # ``work_email``, minted ``nicholas.muniz@`` on day-1 with the correct # name sitting one column away. Reading the golden record HERE makes the # mailbox a function of the LATEST authoritative input at the moment it # is actually created, which is the only moment that is hard to undo: # after this call a fix costs a Directory API rename, an alias, and a # correction email to a brand-new hire. # # A disagreement between (0) and (1) self-heals downstream: this # function's SF write-back is unconditional, so PerEmail gets the # corrected address in the same run. corporate_domains = await get_corporate_email_domains(db) local_part = _local_part_from_hris_email( person.hris_email_name, max_length=max_local, corporate_domains=corporate_domains, ) local_part_source = "hris_email_name" if not local_part: local_part = _local_part_from_hris_email( person.work_email, max_length=max_local, corporate_domains=corporate_domains, ) local_part_source = "hris_email" if not local_part: local_part = _local_part_from_name( display_name=person.display_name, name_pinyin=person.name_pinyin, employee_no=person.employee_no, max_length=max_local, first_name=person.first_name, last_name=person.last_name, ) local_part_source = "derived_name" if not local_part: return ProvisioningOutcome( status="failed", reason="could not derive local part from HRIS email / name / employee_no", ) given_name, family_name = _split_name_for_google( display_name=person.display_name, local_part=local_part, first_name=person.first_name, last_name=person.last_name, name_pinyin=person.name_pinyin, ) resolved = await _resolve(db, integration) driver = GoogleWorkspaceDriver( integration_id=str(integration.id), config=resolved.config, ) # ── Reconcile-first (2026-06-26) ─────────────────────────── # The company's PRIMARY Google account provisioning happens OUTSIDE # Helix (HRIS-driven at hire), so by the time this fires the mailbox # almost always ALREADY exists. The old code went straight to # ``create_user`` and, on the inevitable 409, appended ``01`` — minting # a DUPLICATE mailbox for someone who already had one. ALWAYS look up # the person's AUTHORITATIVE work_email in Google first: # * found → link it (and unsuspend a rehire's mailbox) # * lookup errored → fail loudly; NEVER fall through to create, # or we recreate the very duplicate we guard. # * genuinely absent → fall through to the create loop below. # We match ONLY on HRIS-assigned addresses (never a name-derived # guess) so we can't adopt a different person's mailbox. # # Two HRIS-assigned candidates, most-authoritative first (2026-08-31): # 1. ``hris_email_name@domain`` — the 邮箱名 HR typed into SF. Probed # first because ``work_email`` is NOT always HRIS-assigned: when HR # had not filled the field at create time, ``reserve_work_email_for`` # put a Helix-DERIVED ``given.family`` guess there. Skipping this # probe would look up the guess, miss a mailbox that already sits at # HR's address, fall into the create loop, take a 409 and mint # ``nick.muniz01@`` — a duplicate at the very address we were trying # to converge on. # 2. ``person.work_email`` — the historical behaviour, unchanged. # Both are HR-sourced, so neither widens the "don't adopt a stranger's # mailbox" contract; the shared/service-account registry guard below # runs on whichever one hits. # # BOTH are probed, in order, and the FIRST hit wins — probing only the # new candidate would be a regression in the mirror case (HR fills the # name late but a mailbox already exists at the old derived address: # missing it would mint a SECOND mailbox for the same person). reconcile_candidates: list[str] = [] if local_part_source == "hris_email_name" and domain: reconcile_candidates.append(f"{local_part}@{domain}".lower()) _stored = (person.work_email or "").strip().lower() if _stored and _stored not in reconcile_candidates: reconcile_candidates.append(_stored) reconcile_candidates = [ c for c in reconcile_candidates if c and not c.endswith("successfactors.local") ] existing_profile = None reconcile_email = "" for candidate in reconcile_candidates: reconcile_email = candidate try: existing_profile = await driver.get_user_profile(candidate) except Exception as exc: # noqa: BLE001 await audit.log( trace_id=trace_id, actor_type="user" if actor_user_id else "system", actor_user_id=actor_user_id, action="google_workspace.reconcile_lookup_failed", status="failed", resource_type="identity_person", resource_id=str(person.id), detail={"reconcile_email": candidate, "error": str(exc)}, ) # Fail closed on ANY lookup error, including on the first of two # candidates: falling through to the second (or to create) after # a 503 is exactly how the duplicate-mailbox bug this block # guards against gets reintroduced. return ProvisioningOutcome( status="failed", reason=f"google directory lookup failed for {candidate}: {exc}", ) if existing_profile is not None: break if existing_profile is not None: gid = existing_profile.get("id") gemail = (existing_profile.get("primaryEmail") or reconcile_email).lower() # Shared/service/retained mailbox guard — mirror the rehire path # (_reprovision_rehired_leaver). ``work_email`` is NOT always # HR-assigned: reserve_work_email_for derives a pinyin name-based # address whose only collision check hits Helix's own DB, never # Google. So a namesake Google mailbox that is a shared/functional # account (or a hard-deleted former employee's) and untracked in # Helix could otherwise be adopted here — linking the new hire to # it and even unsuspending it. Classify against the registry and # skip for manual review rather than adopt someone else's mailbox. from app.domains.asset.models import ( # noqa: PLC0415 EXTERNAL_ACCOUNT_KNOWN_CLASSIFICATIONS, ExternalAccountRegistry, ) reconcile_registry_row = (await db.execute( select(ExternalAccountRegistry).where( func.lower(ExternalAccountRegistry.email) == gemail, ), )).scalars().first() if ( reconcile_registry_row is not None and reconcile_registry_row.classification in EXTERNAL_ACCOUNT_KNOWN_CLASSIFICATIONS ): await audit.log( trace_id=trace_id, actor_type="user" if actor_user_id else "system", actor_user_id=actor_user_id, action="google_workspace.reconcile_skipped_classified", status="success", resource_type="identity_person", resource_id=str(person.id), detail={ "reconcile_email": gemail, "classification": reconcile_registry_row.classification, }, ) return ProvisioningOutcome( status="skipped", primary_email=gemail, reason=( f"reconcile: {gemail} is registered as " f"{reconcile_registry_row.classification!r} " "(shared/service/retained) — not auto-linking; " "manual review required" ), ) # Cross-person ownership guard (2026-07-13 review). The # registry check above only covers accounts CLASSIFIED as # shared/service/retained. But ``work_email`` here is often a # pinyin name-derived address whose collision check only hit # Helix's own DB — so a namesake mailbox could still be another # HUMAN's live account. If the found Google account (subject or # email) is ALREADY the active login identity of a DIFFERENT # Helix person, adopting it would bind two people to one mailbox. # Refuse and route to manual review. Does NOT false-positive on # a legitimate rehire: a rehire's prior identity is the SAME # person, so it never matches ``person_id != person.id``. clash = await db.scalar( select(IdentityExternalIdentity).where( IdentityExternalIdentity.provider == _GOOGLE_PROVIDER, IdentityExternalIdentity.deleted_at.is_(None), IdentityExternalIdentity.is_login_identity.is_(True), IdentityExternalIdentity.person_id != person.id, or_( IdentityExternalIdentity.provider_subject == str(gid), func.lower(IdentityExternalIdentity.provider_email) == gemail, ), ) ) if clash is not None: await audit.log( trace_id=trace_id, actor_type="user" if actor_user_id else "system", actor_user_id=actor_user_id, action="google_workspace.reconcile_skipped_namesake", status="success", resource_type="identity_person", resource_id=str(person.id), detail={ "reconcile_email": gemail, "owned_by_person_id": str(clash.person_id), }, ) return ProvisioningOutcome( status="skipped", primary_email=gemail, reason=( f"reconcile: {gemail} is already the active Google " f"login identity of another person ({clash.person_id}) " "— not adopting a namesake's mailbox; manual review " "required" ), ) was_suspended = bool(existing_profile.get("suspended")) unsuspend_ok: bool | None = None if was_suspended: # Returning rehire whose mailbox was suspended on leave — # bring it back rather than minting a fresh one. from app.integrations.driver_base import ProvisioningContext try: sres = await driver._set_suspended( ProvisioningContext( user_email=gemail, user_display_name=person.display_name, ), suspended=False, ) unsuspend_ok = bool(sres.ok) except Exception as exc: # noqa: BLE001 unsuspend_ok = False logger.warning( "google.account_provisioner: unsuspend failed for " "%s during reconcile: %s", gemail, exc, ) seat_id = await _link_google_identity( db, person, google_user_id=gid, primary_email=gemail, reason=f"reconcile-link existing Google account {gemail}", ) await audit.log( trace_id=trace_id, actor_type="user" if actor_user_id else "system", actor_user_id=actor_user_id, action="google_workspace.account_linked", status="success", resource_type="identity_person", resource_id=str(person.id), detail={ "primary_email": gemail, "google_user_id": gid, "reactivated": was_suspended, "unsuspend_ok": unsuspend_ok, "seat_id": str(seat_id) if seat_id else None, }, ) if was_suspended and unsuspend_ok is False: # The identity link is created (correct), but the mailbox is # STILL suspended — the rehire cannot log in. Reporting # ``status="linked"`` (a success outcome the coordinator does # NOT warn on) would silently hide the lockout. Surface it as # ``failed`` so it reaches an admin; a later retry re-enters # this same reconcile path, finds the profile still suspended, # and re-attempts the unsuspend (self-heals once Google # accepts it). return ProvisioningOutcome( status="failed", primary_email=gemail, google_user_id=gid, seat_id=seat_id, reason=( "linked existing Google account but UNSUSPEND FAILED — " "mailbox is still suspended and the user cannot log in; " "unsuspend it in the Google Admin console or re-run " "provisioning once Google is healthy." ), ) return ProvisioningOutcome( status="linked", primary_email=gemail, google_user_id=gid, seat_id=seat_id, reason="reactivated suspended mailbox" if was_suspended else None, ) initial_password = GoogleWorkspaceDriver._generate_strong_password() # HRIS personal contact → Google's home + recovery slots so the new # mailbox isn't a bare shell: "Email (Home)" / "Phone (Home)" on the # user card and the Security → Recovery panel. Guard the personal email # through the same validator the welcome uses (present, well-formed, not # corporate, not the work address); the recovery phone is formatted to # strict E.164 (Google rejects anything else) and skipped when it can't # be — a wrong recovery phone is worse than an empty one. _pe_ok, _ = _validate_personal_email_for_welcome(person, person.work_email or "") home_email = person.personal_email if _pe_ok else None home_phone = (person.mobile_phone or "").strip() or None recovery_phone = to_e164(person.mobile_phone, person.country) # ── Conflict-aware create loop ───────────────────────────── last_error: str | None = None chosen_email: str | None = None google_user_id: str | None = None # One-shot fallback: if Google rejects the (optional) recovery/home phone # we drop it and retry the SAME address once, then keep it dropped for any # later collision-suffix attempts too. phones_dropped = False for attempt in range(_MAX_COLLISION_ATTEMPTS + 1): # attempt 0 → bare local part; collisions append a zero-padded # counter ``01`` … ``09`` (wei.chen → wei.chen01 → … → wei.chen09). candidate_local = _candidate_local_part( local_part, attempt, max_local=max_local, ) candidate = f"{candidate_local}@{domain}" result: WorkspaceOpResult = await driver.create_user( primary_email=candidate, given_name=given_name, family_name=family_name, password=initial_password, org_unit_path=org_unit_path, change_password_at_next_login=True, recovery_email=home_email, recovery_phone=recovery_phone, home_email=home_email, home_phone=home_phone, ) # Recovery/home phone rejected → drop it and retry this same address # once. A recovery phone is optional and must never block creation. if ( not result.ok and not result.already_exists and not phones_dropped and (recovery_phone or home_phone) and _is_phone_error(result.error_message) ): logger.warning( "google.account_provisioner: Google rejected recovery/home " "phone for %s (%s) — retrying without phone", candidate, result.error_message, ) recovery_phone = None home_phone = None phones_dropped = True result = await driver.create_user( primary_email=candidate, given_name=given_name, family_name=family_name, password=initial_password, org_unit_path=org_unit_path, change_password_at_next_login=True, recovery_email=home_email, recovery_phone=None, home_email=home_email, home_phone=None, ) if result.ok and not result.already_exists: chosen_email = result.data.get("primary_email") or candidate google_user_id = result.data.get("id") break if result.already_exists: # 409 — try the next suffix. We never link a colliding # account: the person could legitimately want a brand-new # mailbox (rare but possible — a Zhang Wei coincidentally # sharing a name with a left contractor whose mailbox is # suspended). If we exhaust the suffix budget we fail loudly # for an admin to resolve, rather than silently adopting # someone else's mailbox. last_error = f"409 collision on {candidate}" continue # Hard error from Google — give up immediately. last_error = result.error_message break if chosen_email is None: # Either Google rejected with an error, or we exhausted the # collision-suffix budget without finding a free address. Either # way we did NOT create or link an account — fail with the last # error so an admin can resolve (free up a suffix / pick a name). await audit.log( trace_id=trace_id, actor_type="user" if actor_user_id else "system", actor_user_id=actor_user_id, action="google_workspace.account_create_failed", status="failed", resource_type="identity_person", resource_id=str(person.id), detail={ "reason": last_error, "local_part_attempted": local_part, "domain": domain, }, ) # Note: the IT-admin push on a ``failed`` outcome is fired by the # caller (hr_sync_coordinator._notify_admins_google_account_outcome, # #1428) — so we don't double-notify here. return ProvisioningOutcome( status="failed", reason=last_error or "create_user returned no email", ) # ── Persist linkage + seat allocation ────────────────────── # work_email becomes the new Google primary so future Helix flows # (notifications, SSO match) hit the right mailbox. The previous # value (often the SF placeholder ``@successfactors.local``) # is overwritten — we keep no audit copy of it because it was # never a real address. # # Lowercase at every write site (matches HrSyncService and # successfactors_sync) so the case-insensitive UNIQUE index on # work_email never collides with a same-address mixed-case row. chosen_email = chosen_email.lower() person.work_email = chosen_email # Revive-or-insert (never a blind add): a rehire may still carry a # tombstoned google identity that the deleted_at-filtered short-circuit # above did not see. ``uq_identity_external_person_provider`` ignores # ``deleted_at``, so a blind insert here would 1062 AFTER the live Google # mailbox was already minted — orphaning it and poisoning the HR-sync # session. See :func:`_revive_or_insert_google_identity`. await _revive_or_insert_google_identity( db, person.id, google_user_id=google_user_id, primary_email=chosen_email, ) await db.flush() seat_id = await _get_or_create_google_seat( db, person_id=person.id, reason=( f"auto-allocated on Google account creation " f"({chosen_email}, role={role})" ), ) # ── Notify the new hire on their personal email ──────────── email_ok, email_err = await _send_password_email( db, person=person, work_email=chosen_email, initial_password=initial_password, cc_emails=cc_emails, ) # Failure-rollback (P0): if Google account creation succeeded but # the welcome email failed, the new hire is locked out — the # plaintext password was generated, never persisted, and never # delivered. Two moves, in order: # 1. SUSPEND the Google account so the credential window closes # immediately (deliberate security control — never leave an # active mailbox with an undeliverable password live in memory # for the SMTP-retry / broken-driver window), and # 2. flip the identity to ``pending_welcome`` so this is NOT a dead # end. On the next run (hire_day cron / manual) the short-circuit # re-enters ``_reprovision_pending_welcome`` — regenerate password # → re-send welcome → unsuspend ONLY on email success — and # ``google.seat_sync`` leaves the base seat alone while pending. # Without the flip the row stayed ``active`` → every re-run # short-circuited to ``skipped`` forever AND the suspended account # lost its seat after 2h = permanent lockout on any email outage. rollback_suspend_ok: bool | None = None rollback_suspend_error: str | None = None rollback_suspend_raw: dict[str, Any] | None = None if not email_ok and google_user_id: try: from app.integrations.driver_base import ProvisioningContext if isinstance(driver, GoogleWorkspaceDriver): # Best-effort suspend; never raise back into the # outcome path so the admin always sees the # partial-success audit row. rollback_ctx = ProvisioningContext( user_email=chosen_email, user_display_name=person.display_name, ) # must_exist: ``create_user`` just handed us this # account's google_user_id, so a 404 from Google can only # be its post-insert settle window — never "the mailbox is # gone". Without this the driver reported the settle-window # 404 as idempotent success and the rollback silently never # fired (prod 2026-08-25, hiroki.shinagawa01@pingcap.com). suspend_result = await driver._set_suspended( rollback_ctx, suspended=True, must_exist=True, ) rollback_suspend_raw = suspend_result.raw_response rollback_suspend_ok = bool(suspend_result.ok) if not rollback_suspend_ok: rollback_suspend_error = _scrub_password_from_text( suspend_result.error_message or "suspend returned ok=False", initial_password, ) except Exception as exc: # noqa: BLE001 rollback_suspend_ok = False rollback_suspend_error = _scrub_password_from_text( f"{type(exc).__name__}: {exc}", initial_password, ) logger.warning( "google.account_provisioner: failure-rollback suspend " "FAILED for primary_email=%s — account is active with " "undeliverable password. Admin must manually reset + " "suspend via Google console. Error: %s", chosen_email, rollback_suspend_error, ) # Mark the self-heal state. Flip regardless of whether the safety-suspend # succeeded: pending_welcome routes the next run through re-provision, # which re-attempts the suspend before promoting back to active — so even # a failed suspend here is recovered rather than left as a live mailbox. if not email_ok: pending_ident = await _get_existing_google_identity(db, person.id) if pending_ident is not None: pending_ident.identity_status = _STATUS_PENDING_WELCOME await db.flush() # ── Report the final chosen email back to HRIS ───────────── # Tell SuccessFactors which corporate email Helix actually minted # (the HRIS-suggested local part, or a collision-resolved …01-…09 # variant). Logged no-op until the HR callback URL is configured — # see app/identity/hris_email_writeback.py. Only fire on full # success: a partial-success account is suspended pending admin # fix, so its address isn't live yet. Best-effort — never blocks. if email_ok: hris_notify = await notify_hris_email_assigned( db, person=person, work_email=chosen_email, trace_id=trace_id, ) else: hris_notify = { "status": "skipped", "reason": "welcome email failed; account suspended pending admin", } hris_notify_status = hris_notify.get("status") await audit.log( trace_id=trace_id, actor_type="user" if actor_user_id else "system", actor_user_id=actor_user_id, action="google_workspace.account_created", status="success" if email_ok else "partial", resource_type="identity_person", resource_id=str(person.id), detail={ "primary_email": chosen_email, "google_user_id": google_user_id, "role": role, "local_part_source": local_part_source, "org_unit_path": org_unit_path, "seat_id": str(seat_id) if seat_id else None, "password_email_ok": email_ok, "password_email_error": email_err, "personal_email_used": person.personal_email, "failure_rollback_suspend_ok": rollback_suspend_ok, "failure_rollback_suspend_error": rollback_suspend_error, # Keep the driver's verdict payload, not just the boolean. The # boolean alone made ``already_absent`` (no update ever sent), # ``already_suspended`` (pre-read short-circuit) and a real # suspend indistinguishable in prod forensics — all three # audited as ``ok: true``. That cost a full investigation # against a mailbox that had already been hard-deleted. "failure_rollback_suspend_result": rollback_suspend_raw, "identity_status_after": ( "active" if email_ok else _STATUS_PENDING_WELCOME ), "hris_notify": hris_notify, }, ) return ProvisioningOutcome( status="created", primary_email=chosen_email, google_user_id=google_user_id, seat_id=seat_id, hris_notify_status=hris_notify_status, reason=None if email_ok else f"account created; email failed: {email_err}", ) def _split_name_for_google( *, display_name: str | None, local_part: str, first_name: str | None = None, last_name: str | None = None, name_pinyin: str | None = None, ) -> tuple[str, str]: """Pick (givenName, familyName) for the Google user record. Google computes the visible name as ``givenName + " " + familyName`` and hard-rejects an empty half, so we must always return two non-empty strings. The old fallback set BOTH halves to the full ``display_name`` whenever the email local-part had no dot — Google then rendered the whole name TWICE (prod bug: a hire created with an employee-number mailbox showed "黄浚源 Junyuan Huang 黄浚源 Junyuan Huang"). Derive a real given/family split, most-reliable source first, and only duplicate a *single* romanised token as the last resort — never the whole "CJK + pinyin" display string. """ # 1. SF structured firstName / lastName — authoritative when present. tokens = _name_tokens_from_first_last( first_name=first_name, last_name=last_name, ) if len(tokens) >= 2: return tokens[0].title(), tokens[-1].title() # 2. Email local-part with a dot ("given.family") — PingCAP convention. if "." in local_part: given, _, family = local_part.partition(".") if given.strip() and family.strip(): return given.capitalize(), family.split(".")[-1].capitalize() # 3. name_pinyin carrying a space ("Junyuan Huang") — already in # given-family Latin order, so keep it as-is (do NOT route through # the CJK reverse heuristic, which would flip the halves). pinyin = (name_pinyin or "").strip() if " " in pinyin: given, _, family = pinyin.rpartition(" ") if given.strip() and family.strip(): return given.strip().title(), family.strip().title() # 4. display_name with a space — drop any leading CJK tokens so # "黄浚源 Junyuan Huang" yields ("Junyuan", "Huang") instead of a # duplicated full string. display = (display_name or "").strip() if " " in display: latin = [t for t in display.split() if not _contains_cjk(t)] if len(latin) >= 2: return latin[0].title(), " ".join(latin[1:]).title() # 5. General romanised derivation (pure-CJK blob, concatenated # pinyin). Returns ``[given, family]`` tokens following the email # naming rules; ``"黄浚源"`` → ``["junyuan", "huang"]``. derived = _name_tokens(display_name=display_name, name_pinyin=name_pinyin) if len(derived) >= 2: return derived[0].title(), derived[-1].title() if len(derived) == 1: # Single unsplittable token — reuse it for both halves. This # duplicates one romanised word at worst, never the whole name. return derived[0].title(), derived[0].title() # 6. Nothing usable — fall back to the local part for both halves. # The caller already guarantees a non-empty local part; the final # ``"User"`` guard only fires for a degenerate direct call and # keeps us from ever handing Google an empty (rejected) half. token = ( (_sanitize_token(local_part) or local_part).title() or local_part or "User" ) return token, token __all__ = ["create_account_for", "ProvisioningOutcome"]