"""DualSignOverrideService — IT admin + secondary signer co-sign an override of the leaver-block invariant. Workflow (matches the docstring on ``app.domains.leaver_blocked.models.LeaverDualSignOverride``): 1. ``initiate(person_id, blocker_ids, primary_reason)`` — IT admin calls this. Validates blockers exist + are pending. Writes a pending_secondary row + audit log + WorkflowRun + an inbox notification to the chosen secondary signer. 2. ``secondary_approve(override_id, secondary_reason)`` — the director (typically the leaver's manager-of-manager or department head) calls this. Validates the caller is allowed to be the secondary signer (manager chain check). Writes the approval; status -> 'approved'. 3. ``secondary_reject(override_id, secondary_reason)`` — same prereqs as approve, but writes a rejection. Status -> 'rejected'. 4. ``execute(override_id)`` — IT admin returns and triggers the actual flip-to-left under the override flag. Calls ``LeaverBlockGuard.flip_to_left(..., override_id=...)``. The ``cancel`` op (initiator-only, while still pending) is implemented for operator escapes. """ from __future__ import annotations import logging import uuid from datetime import UTC, datetime, timedelta from typing import Any from uuid import UUID from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession from app.audit.service import AuditService from app.core.exceptions import ( ConflictError, ForbiddenError, NotFoundError, ValidationError, ) from app.core.rbac import ( can_manage_leaver_blocked, can_secondary_sign_leaver_override, ) from app.core.workflow_journal import ( RUN_STATUS_CANCELLED, RUN_STATUS_FAILED, RUN_STATUS_SUCCEEDED, WORKFLOW_TYPE_LEAVER_DUAL_SIGN, WorkflowJournal, ) from app.core.workflow_journal.constants import ( STEP_STATUS_SUCCEEDED, TRIGGER_USER, ) from app.domains.leaver_blocked.guard import ( LEAVER_VIA_DUAL_SIGN_OVERRIDE, LeaverBlockGuard, ) from app.domains.leaver_blocked.models import ( BLOCKING_DUAL_SIGN_STATUSES, DUAL_SIGN_APPROVED, DUAL_SIGN_CANCELLED, DUAL_SIGN_EXECUTED, DUAL_SIGN_PENDING_SECONDARY, DUAL_SIGN_REJECTED, MIN_REASON_CHARS, OVERRIDE_TTL_DAYS, LeaverDualSignOverride, ) from app.domains.offboarding.models import ( RESO_PENDING, OffboardingBlocker, OffboardingProcess, ) from app.identity.person_models import IdentityPerson from app.notifications.service import NotificationEnvelope, NotificationService from app.notifications.text_safety import safe_envelope_text logger = logging.getLogger(__name__) class DualSignOverrideService: def __init__(self, db: AsyncSession) -> None: self.db = db self.audit = AuditService(db) self.notify = NotificationService(db) self.journal = WorkflowJournal(db) self.guard = LeaverBlockGuard(db) # ── Initiate ─────────────────────────────────────────────── async def initiate( self, *, person_id: UUID, blocker_ids: list[UUID], primary_reason: str, secondary_signer_user_id: UUID, primary_signer: IdentityPerson, trace_id: UUID, client_ip: str | None = None, ) -> LeaverDualSignOverride: if not can_manage_leaver_blocked(primary_signer.role_names): raise ForbiddenError( "Only global_admin or hr role can initiate dual-sign overrides" ) _ensure_reason_long_enough(primary_reason, "primary_reason") person = await self.db.get(IdentityPerson, person_id) if person is None: raise NotFoundError("identity_person", str(person_id)) # ``with_for_update`` serialises concurrent initiate calls for the # same person on the process row, so a second initiate observes the # first's in-flight override below instead of both passing the # duplicate-prevention check (the execute / _load_pending paths # already lock; this matches them). The per-execution # ``_validate_override`` coverage re-check is the ultimate backstop. proc = await self.db.scalar( select(OffboardingProcess).where( OffboardingProcess.person_id == person_id, ).with_for_update() ) if proc is None: raise ConflictError( "No OffboardingProcess for this person — start the " "offboarding workflow before requesting an override" ) # Forbid creating a NEW override when one is already in flight. existing = await self.db.scalar( select(LeaverDualSignOverride).where( LeaverDualSignOverride.process_id == proc.id, LeaverDualSignOverride.status.in_(BLOCKING_DUAL_SIGN_STATUSES), ) ) if existing is not None: raise ConflictError( f"A dual-sign override is already in flight for this person " f"(status={existing.status}, id={existing.id})" ) # Snapshot the named blockers to make sure they (a) exist and # (b) target this person's process and (c) are still pending. if not blocker_ids: raise ValidationError( "Must list at least one blocker_id to override" ) rows = list((await self.db.execute( select(OffboardingBlocker).where( OffboardingBlocker.id.in_(list(blocker_ids)), ) )).scalars().all()) if len(rows) != len(blocker_ids): missing = set(str(b) for b in blocker_ids) - {str(r.id) for r in rows} raise NotFoundError("offboarding_blocker", ", ".join(missing)) for row in rows: if row.process_id != proc.id: raise ValidationError( f"Blocker {row.id} belongs to a different process" ) if row.resolution != RESO_PENDING: raise ValidationError( f"Blocker {row.id} is not pending (current " f"resolution={row.resolution!r}); cannot override" ) snapshot = [ { "id": str(b.id), "resource_type": b.resource_type, "resource_id": str(b.resource_id) if b.resource_id else None, "label": b.label, "software_id": str(b.software_id) if b.software_id else None, } for b in rows ] # Validate the secondary signer choice secondary = await self.db.get(IdentityPerson, secondary_signer_user_id) if secondary is None: raise NotFoundError("identity_person", str(secondary_signer_user_id)) if secondary.id == primary_signer.id: raise ValidationError( "Primary and secondary signer must be different people" ) # Pre-launch P1 fix (2026-05-16): the secondary signer MUST be a # currently-active employee. The dual-sign contract is "the IT # admin's force-override is co-signed by a still-present human # who answers for the consequences." A soft-deleted or leaver/ # left secondary defeats that — and a leaver-pending secondary # creates the absurd shape of an offboarding person authorizing # someone else's offboarding override. if secondary.deleted_at is not None: raise ValidationError( "Secondary signer is soft-deleted; pick an active " "employee." ) if secondary.employment_status != "active": raise ValidationError( f"Secondary signer is {secondary.employment_status!r}; " "pick an active employee." ) if not can_secondary_sign_leaver_override(secondary.role_names): # Allow non-role secondary if they are the leaver's manager # chain (director-by-org), otherwise reject. The chain check # is "manager_of_manager or higher". if not await _is_in_manager_chain(self.db, secondary.id, person.id, depth=4): raise ValidationError( "Secondary signer must be a global_admin/hr OR " "in the leaver's manager chain (≤ 4 levels up)" ) now = datetime.now(UTC) override = LeaverDualSignOverride( id=uuid.uuid4(), process_id=proc.id, person_id=person.id, status=DUAL_SIGN_PENDING_SECONDARY, primary_signer_user_id=primary_signer.id, primary_signed_at=now, primary_reason=primary_reason.strip(), secondary_signer_user_id=secondary.id, overridden_blocker_ids=[str(b.id) for b in rows], overridden_blockers_snapshot=snapshot, expires_at=now + timedelta(days=OVERRIDE_TTL_DAYS), trace_id=trace_id, ) self.db.add(override) await self.db.flush() # Open a WorkflowRun for the override lifecycle so it shows # up in the leaver's timeline. run = await self.journal.start_run( workflow_type=WORKFLOW_TYPE_LEAVER_DUAL_SIGN, person_id=person.id, related_resource_type="leaver_dual_sign_override", related_resource_id=override.id, trace_id=trace_id, trigger_source=TRIGGER_USER, triggered_by_user_id=primary_signer.id, context={ "primary_signer": str(primary_signer.id), "secondary_signer": str(secondary.id), "blocker_count": len(rows), }, ) await self.journal.record_step_blocked( run, "await_secondary_signature", reason="waiting for secondary signer review", details={ "secondary_signer": str(secondary.id), "expires_at": override.expires_at.isoformat() if override.expires_at else None, }, ) await self.audit.log( trace_id=str(trace_id), actor_type="user", actor_user_id=primary_signer.id, action="leaver_blocked.override.initiate", resource_type="leaver_dual_sign_override", resource_id=str(override.id), status="success", detail={ "person_id": str(person.id), "blocker_count": len(rows), "secondary_signer": str(secondary.id), "primary_reason_preview": primary_reason[:120], }, ip_address=client_ip, ) # Notify the secondary signer await self._notify_secondary( override=override, person=person, primary_signer=primary_signer, secondary=secondary, blockers_snapshot=snapshot, ) return override # ── Secondary approve / reject ───────────────────────────── async def secondary_approve( self, *, override_id: UUID, secondary_reason: str, secondary_signer: IdentityPerson, trace_id: UUID, client_ip: str | None = None, ) -> LeaverDualSignOverride: _ensure_reason_long_enough(secondary_reason, "secondary_reason") override = await self._load_pending(override_id) if override.secondary_signer_user_id != secondary_signer.id: raise ForbiddenError( f"Override {override_id} expects secondary signer " f"{override.secondary_signer_user_id}; you are not them" ) if override.primary_signer_user_id == secondary_signer.id: raise ForbiddenError( "You signed this override as the primary; you cannot " "also approve it as the secondary" ) # Pre-launch P1 fix (2026-05-16): re-check active status at # approve time — the secondary may have been deleted or flipped # to leaver_pending/left between initiate and approve, in which # case the override invariant ("still-present human co-signs") # is violated and the approval must be refused. The router gate # already requires the caller to be authenticated, but the auth # gate doesn't read employment_status / deleted_at. if secondary_signer.deleted_at is not None: raise ForbiddenError( "Secondary signer account is soft-deleted; approval " "refused. Re-initiate the override with a different " "secondary signer." ) if secondary_signer.employment_status != "active": raise ForbiddenError( f"Secondary signer is {secondary_signer.employment_status!r}; " "approval refused. Re-initiate the override with a " "different secondary signer." ) override.status = DUAL_SIGN_APPROVED override.secondary_signed_at = datetime.now(UTC) override.secondary_reason = secondary_reason.strip() await self.db.flush() run = await self.journal.get_run_by_resource( "leaver_dual_sign_override", override.id, ) if run is not None: await self.journal.record_signal( run, "secondary_signature", signal_kind="approve", signaller_user_id=secondary_signer.id, payload={"secondary_reason_preview": secondary_reason[:200]}, ) await self.audit.log( trace_id=str(trace_id), actor_type="user", actor_user_id=secondary_signer.id, action="leaver_blocked.override.approve", resource_type="leaver_dual_sign_override", resource_id=str(override.id), status="success", detail={ "person_id": str(override.person_id), "secondary_reason_preview": secondary_reason[:120], }, ip_address=client_ip, ) # Notify the primary signer that they can execute await self._notify_primary_can_execute(override) return override async def secondary_reject( self, *, override_id: UUID, secondary_reason: str, secondary_signer: IdentityPerson, trace_id: UUID, client_ip: str | None = None, ) -> LeaverDualSignOverride: _ensure_reason_long_enough(secondary_reason, "secondary_reason") override = await self._load_pending(override_id) if override.secondary_signer_user_id != secondary_signer.id: raise ForbiddenError( f"Override {override_id} expects secondary signer " f"{override.secondary_signer_user_id}; you are not them" ) # Symmetry with ``secondary_approve``: the "still-present human # co-signs" invariant applies to any decision the secondary makes. # A soft-deleted / departed secondary should re-initiate rather # than act on a stale assignment. if secondary_signer.deleted_at is not None: raise ForbiddenError( "Secondary signer account is soft-deleted; cannot act on " "this override. Re-initiate with a different secondary." ) if secondary_signer.employment_status != "active": raise ForbiddenError( f"Secondary signer is {secondary_signer.employment_status!r}; " "cannot act on this override. Re-initiate with a different " "secondary." ) override.status = DUAL_SIGN_REJECTED override.rejected_at = datetime.now(UTC) override.rejected_reason = secondary_reason.strip() await self.db.flush() run = await self.journal.get_run_by_resource( "leaver_dual_sign_override", override.id, ) if run is not None: await self.journal.record_signal( run, "secondary_signature", signal_kind="reject", signaller_user_id=secondary_signer.id, payload={"reason_preview": secondary_reason[:200]}, ) await self.journal.finish_run( run, status=RUN_STATUS_FAILED, error_message=f"rejected by secondary: {secondary_reason[:200]}", ) await self.audit.log( trace_id=str(trace_id), actor_type="user", actor_user_id=secondary_signer.id, action="leaver_blocked.override.reject", resource_type="leaver_dual_sign_override", resource_id=str(override.id), status="success", detail={ "person_id": str(override.person_id), "secondary_reason_preview": secondary_reason[:120], }, ip_address=client_ip, ) # Pre-launch P1 fix: notify the primary signer of the # rejection so they don't learn about it by polling the # dashboard. Mirrors ``_notify_primary_can_execute`` on the # approve path. await self._notify_primary_rejected( override, secondary_signer=secondary_signer, reason=secondary_reason, ) return override # ── Cancel (initiator) ───────────────────────────────────── async def cancel( self, *, override_id: UUID, actor: IdentityPerson, reason: str, trace_id: UUID, client_ip: str | None = None, ) -> LeaverDualSignOverride: override = await self.db.get(LeaverDualSignOverride, override_id) if override is None: raise NotFoundError("leaver_dual_sign_override", str(override_id)) if override.status != DUAL_SIGN_PENDING_SECONDARY: raise ConflictError( f"Cannot cancel override in status {override.status!r}; " f"only pending_secondary is cancellable" ) if ( override.primary_signer_user_id != actor.id and not can_manage_leaver_blocked(actor.role_names) ): raise ForbiddenError( "Only the primary signer or a global_admin can cancel " "an override" ) override.status = DUAL_SIGN_CANCELLED await self.db.flush() run = await self.journal.get_run_by_resource( "leaver_dual_sign_override", override.id, ) if run is not None: await self.journal.finish_run( run, status=RUN_STATUS_CANCELLED, error_message=f"cancelled by {actor.work_email}: {reason[:200]}", ) await self.audit.log( trace_id=str(trace_id), actor_type="user", actor_user_id=actor.id, action="leaver_blocked.override.cancel", resource_type="leaver_dual_sign_override", resource_id=str(override.id), status="success", detail={ "person_id": str(override.person_id), "reason_preview": reason[:120], }, ip_address=client_ip, ) return override # ── Execute ──────────────────────────────────────────────── async def execute( self, *, override_id: UUID, actor: IdentityPerson, trace_id: UUID, client_ip: str | None = None, user_agent: str | None = None, ) -> LeaverDualSignOverride: # Pre-launch P2 fix (Round 1, CR-018): SELECT ... FOR UPDATE so # two concurrent admin "Execute" clicks serialize at the row # level. Without the lock, both observers would see status = # DUAL_SIGN_APPROVED, both call ``guard.flip_to_left`` (the # second is a no-op vs ``'left'``), both write audit rows + a # WorkflowJournal entry. End state was correct but with # duplicate audit/journal rows muddying the trail. override = ( await self.db.execute( select(LeaverDualSignOverride) .where(LeaverDualSignOverride.id == override_id) .with_for_update() ) ).scalar_one_or_none() if override is None: raise NotFoundError("leaver_dual_sign_override", str(override_id)) if override.status == DUAL_SIGN_EXECUTED: # Idempotent re-call — second click of "Execute" gets the # already-executed row back, not a duplicate flip. return override if override.status != DUAL_SIGN_APPROVED: raise ConflictError( f"Cannot execute override in status {override.status!r}; " f"only approved overrides can be executed" ) # Pre-launch P1 fix: approved-but-unexecuted overrides must # honor ``expires_at``. Without this, an admin can let an # approved override sit for months, then execute it long # after the cited blockers / business context drifted. # ``_expire_stale_overrides`` (sla_service) only sweeps # ``pending_secondary``; this gate plus the matching sweep # extension covers the approved branch. if override.expires_at is not None: now = datetime.now(UTC) expires_at = override.expires_at if expires_at.tzinfo is None: expires_at = expires_at.replace(tzinfo=UTC) if expires_at < now: raise ConflictError( f"Override expired at {expires_at.isoformat()}; " f"approval is no longer valid. Re-initiate the " f"override if the blockers / business context " f"still warrant a flip." ) if ( override.primary_signer_user_id != actor.id and not can_manage_leaver_blocked(actor.role_names) ): raise ForbiddenError( "Only the primary signer or a global_admin can execute " "the override" ) # Pre-launch P1 fix (2026-05-16): re-check actor is still an # active employee. The approved row can sit for up to ``expires_at`` # (days) before someone clicks Execute; in that window the actor # could have been offboarded themselves and an attacker reusing # a not-yet-revoked session could trigger the flip. Belt-and- # braces alongside the dependency chain: ``get_current_user`` # already rejects ``is_effectively_departed`` callers, but the # employment_status / deleted_at recheck is cheap and keeps the # gate honest even if a future refactor relaxes that dependency. if actor.deleted_at is not None: raise ForbiddenError( "Executing actor account is soft-deleted; refusing." ) if actor.employment_status != "active": raise ForbiddenError( f"Executing actor is {actor.employment_status!r}; only " "an active admin / primary signer can execute the override." ) person = await self.db.get(IdentityPerson, override.person_id) if person is None: raise NotFoundError("identity_person", str(override.person_id)) # The guard does the real work — flip + audit + listener # contextvar bypass + override status update. await self.guard.flip_to_left( person, via=LEAVER_VIA_DUAL_SIGN_OVERRIDE, trace_id=trace_id, actor_user_id=actor.id, client_ip=client_ip, user_agent=user_agent, override_id=override.id, ) run = await self.journal.get_run_by_resource( "leaver_dual_sign_override", override.id, ) if run is not None: await self.journal.record_step( run, "execute_flip", STEP_STATUS_SUCCEEDED, actor_user_id=actor.id, actor_kind="user", payload={"override_id": str(override.id)}, ) await self.journal.finish_run( run, status=RUN_STATUS_SUCCEEDED, ) await self.audit.log( trace_id=str(trace_id), actor_type="user", actor_user_id=actor.id, action="leaver_blocked.override.execute", resource_type="leaver_dual_sign_override", resource_id=str(override.id), status="success", detail={ "person_id": str(override.person_id), "blocker_count": len(override.overridden_blocker_ids or []), }, ) return override # ── Listing ──────────────────────────────────────────────── async def list_overrides( self, *, statuses: list[str] | None = None, person_id: UUID | None = None, secondary_signer_user_id: UUID | None = None, limit: int = 100, ) -> list[LeaverDualSignOverride]: query = select(LeaverDualSignOverride) if statuses: query = query.where(LeaverDualSignOverride.status.in_(statuses)) if person_id is not None: query = query.where(LeaverDualSignOverride.person_id == person_id) if secondary_signer_user_id is not None: query = query.where( LeaverDualSignOverride.secondary_signer_user_id == secondary_signer_user_id, ) query = query.order_by( LeaverDualSignOverride.created_at.desc(), ).limit(limit) result = await self.db.execute(query) return list(result.scalars().all()) async def count_overrides( self, *, statuses: list[str] | None = None, person_id: UUID | None = None, secondary_signer_user_id: UUID | None = None, ) -> int: """Full filtered count for ``list_overrides`` pagination — the list caps at ``limit`` so ``len(rows)`` undercounts the true total whenever the result is full.""" query = select(func.count()).select_from(LeaverDualSignOverride) if statuses: query = query.where(LeaverDualSignOverride.status.in_(statuses)) if person_id is not None: query = query.where(LeaverDualSignOverride.person_id == person_id) if secondary_signer_user_id is not None: query = query.where( LeaverDualSignOverride.secondary_signer_user_id == secondary_signer_user_id, ) return (await self.db.execute(query)).scalar_one() # ── Notifications ────────────────────────────────────────── async def _notify_secondary( self, *, override: LeaverDualSignOverride, person: IdentityPerson, primary_signer: IdentityPerson, secondary: IdentityPerson, blockers_snapshot: list[dict[str, Any]], ) -> None: try: await self.notify.dispatch( NotificationEnvelope( recipient=secondary, title=( f"[副签请求] {primary_signer.display_name or primary_signer.work_email} " f"请你共同签名跳过 {person.display_name or person.work_email} " f"的 {len(blockers_snapshot)} 项离职 blocker" ), body_text=( f"IT 同事 {primary_signer.display_name or primary_signer.work_email} " f"申请绕开下列 blocker 让 " f"{person.display_name or person.work_email} 完成离职:\n\n" + "\n".join( f" · [{b['resource_type']}] {b['label']}" for b in blockers_snapshot ) + f"\n\nIT 给出的理由(节选):{safe_envelope_text(override.primary_reason, limit=300)}\n\n" + f"请到 Helix 审阅:\n /admin/leaver-blocked/override/{override.id}\n\n" + f"过期时间:{override.expires_at.isoformat() if override.expires_at else 'N/A'}\n\n" + "副签前请确认: 业务上确实可以放行,且你愿意为此承担合规责任。" ), channels=["feishu", "email"], template_key="leaver_blocked.dual_sign.secondary_request", url=f"/admin/leaver-blocked/override/{override.id}", context={ "override_id": str(override.id), "primary_name": primary_signer.display_name or primary_signer.work_email, "leaver_name": person.display_name or person.work_email, "blocker_count": len(blockers_snapshot), # Rendered blocker list + the primary's reason — the # 三语 template needs both as text (a list of dicts # interpolates as its Python repr, and the reason was # only ever in the hardcoded body). "blocker_list": "\n".join( f" · [{b['resource_type']}] {b['label']}" for b in blockers_snapshot ), "reason": safe_envelope_text( override.primary_reason, limit=300, ), # "N/A" rather than None: {expires_at} would otherwise # render the literal string "None" to the co-signer. "expires_at": ( override.expires_at.isoformat() if override.expires_at else "N/A" ), # ``portal_url`` intentionally absent — pinning the # relative path here beat the renderer's absolute-URL # injection and made the email CTA unclickable; ``url`` # above already carries the same target. }, related_resource_type="leaver_dual_sign_override", related_resource_id=override.id, ), trace_id=str(override.trace_id), ) except Exception: # noqa: BLE001 logger.exception( "leaver_blocked override secondary notify failed for %s", override.id, ) async def _notify_primary_can_execute( self, override: LeaverDualSignOverride, ) -> None: primary = await self.db.get( IdentityPerson, override.primary_signer_user_id, ) if primary is None or primary.deleted_at is not None: return # Pre-launch P1 fix: skip when the primary signer themselves # has been moved to leaver_pending / left between initiating # and the secondary approving. They couldn't execute even if # notified (mutate gate at dependencies.py blocks them). if primary.employment_status != "active": logger.warning( "leaver_blocked override %s primary signer %s is " "%s — skipping execute-ready notify; another admin " "must execute via global_admin path", override.id, primary.work_email, primary.employment_status, ) return person = await self.db.get(IdentityPerson, override.person_id) if person is None: return try: await self.notify.dispatch( NotificationEnvelope( recipient=primary, title=( f"[副签已通过] 你可以执行 {person.display_name or person.work_email} " f"的离职 override" ), body_text=( "副签人已批准。你现在可以到 Helix 执行 override:\n\n" f" /admin/leaver-blocked/override/{override.id}\n\n" "执行后系统将忽略指定 blocker,把人员置 left。" ), channels=["feishu", "email"], template_key="leaver_blocked.dual_sign.execute_ready", url=f"/admin/leaver-blocked/override/{override.id}", context={ "override_id": str(override.id), "leaver_name": person.display_name or person.work_email, # See ``_notify_secondary``: a relative ``portal_url`` # in context overrides the renderer's absolute URL and # yields an unclickable email CTA. ``url`` above is the # single source of the link. }, related_resource_type="leaver_dual_sign_override", related_resource_id=override.id, ), trace_id=str(override.trace_id), ) except Exception: # noqa: BLE001 logger.exception( "leaver_blocked override primary-can-execute notify failed for %s", override.id, ) async def _notify_primary_rejected( self, override: LeaverDualSignOverride, *, secondary_signer: IdentityPerson, reason: str, ) -> None: """Tell the primary signer their override was rejected by the secondary. Pre-launch P1 fix — without this, the IT admin who initiated learns about the rejection only by polling the dashboard. Same pattern as _notify_primary_can_execute.""" primary = await self.db.get( IdentityPerson, override.primary_signer_user_id, ) if primary is None or primary.deleted_at is not None: return if primary.employment_status != "active": return person = await self.db.get(IdentityPerson, override.person_id) if person is None: return try: await self.notify.dispatch( NotificationEnvelope( recipient=primary, title=( f"[副签已驳回] {person.display_name or person.work_email} " f"的离职 override 未通过" ), body_text=( f"副签人 {secondary_signer.display_name or secondary_signer.work_email} " f"驳回了 override。\n\n" f"驳回理由:\n{safe_envelope_text(reason, limit=500)}\n\n" "如仍需 override，请重新发起并附补充材料。" ), channels=["feishu", "email"], template_key="leaver_blocked.dual_sign.rejected", url=f"/admin/leaver-blocked/override/{override.id}", context={ "override_id": str(override.id), "leaver_name": person.display_name or person.work_email, "secondary_signer_name": ( secondary_signer.display_name or secondary_signer.work_email ), "reason": safe_envelope_text(reason, limit=500), # See ``_notify_secondary`` — no relative portal_url. }, related_resource_type="leaver_dual_sign_override", related_resource_id=override.id, ), trace_id=str(override.trace_id), ) except Exception: # noqa: BLE001 logger.exception( "leaver_blocked override rejected notify failed for %s", override.id, ) # ── helpers ──────────────────────────────────────────────── async def _load_pending( self, override_id: UUID, ) -> LeaverDualSignOverride: # Use FOR UPDATE to serialise concurrent approve + reject calls # that both observe status == DUAL_SIGN_PENDING_SECONDARY. # Without the lock both paths pass the status check and both # apply their respective state transition (one writes APPROVED, # one writes REJECTED), producing an audit trail where a forensic # override was simultaneously approved and rejected. override = ( await self.db.execute( select(LeaverDualSignOverride) .where(LeaverDualSignOverride.id == override_id) .with_for_update() ) ).scalar_one_or_none() if override is None: raise NotFoundError("leaver_dual_sign_override", str(override_id)) if override.status != DUAL_SIGN_PENDING_SECONDARY: raise ConflictError( f"Override {override_id} is in status {override.status!r}; " f"only pending_secondary overrides accept signatures" ) # Reject an already-expired override before any signature write. # ``_expire_stale_overrides`` sweeps these on a timer, but in the # window between TTL elapse and the next sweep the secondary signer # could otherwise still approve — minting a DUAL_SIGN_APPROVED audit # stamp + "ready to execute" DM for an override execute()/the guard # will then refuse on expiry. Refuse here too so the 4-eyes audit # trail can't record an approval past the TTL. Mirrors the naive→UTC # coercion execute() uses (asyncmy returns naive DATETIME on load). if override.expires_at is not None: expires_at = override.expires_at if expires_at.tzinfo is None: expires_at = expires_at.replace(tzinfo=UTC) if expires_at < datetime.now(UTC): raise ConflictError( f"Override {override_id} expired at " f"{expires_at.isoformat()}; it can no longer be signed. " f"Re-initiate the override if the blockers still warrant " f"a flip." ) return override def _ensure_reason_long_enough(reason: str, field_name: str) -> None: if reason is None or len(reason.strip()) < MIN_REASON_CHARS: raise ValidationError( f"{field_name} must be at least {MIN_REASON_CHARS} chars; " f"got {len(reason.strip()) if reason else 0}" ) async def _is_in_manager_chain( db: AsyncSession, candidate_id: UUID, person_id: UUID, *, depth: int = 4, ) -> bool: """Walks up to ``depth`` levels of the leaver's manager chain looking for ``candidate_id``. Pre-launch P1 fix (2026-05-16): a soft-deleted intermediate manager stops the walk (returns False) and a soft-deleted candidate match also returns False. The chain represents the org-shaped chain of accountability for the dual-sign override; a deleted ancestor's presence in the path means the link is historic and should not be used to authorize a forensic-grade decision. """ cur_id: UUID | None = person_id for _ in range(depth): cur = await db.get(IdentityPerson, cur_id) if cur_id else None if cur is None or cur.manager_person_id is None: return False if cur.deleted_at is not None: return False if cur.manager_person_id == candidate_id: mgr = await db.get(IdentityPerson, cur.manager_person_id) if mgr is None or mgr.deleted_at is not None: return False if mgr.employment_status != "active": return False return True cur_id = cur.manager_person_id return False __all__ = ["DualSignOverrideService"]