import json from collections import defaultdict from datetime import UTC, datetime from uuid import UUID from sqlalchemy import and_, distinct, func, select from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import selectinload from app.auth.models import AuthRole, AuthUser, AuthUserRole from app.core.exceptions import NotFoundError from app.core.rbac import ( ASSET_GLOBAL_ADMIN_ROLE, CANONICAL_ROLE_NAMES, EMPLOYEE_ROLE, LEAVER_REVOCABLE_ROLES, canonicalize_role_name, db_role_names_for, ) from app.core.redis_client import get_redis from app.core.safe_logging import get_safe_logger from app.core.security import ( create_access_token, create_refresh_token, verify_password_constant_time, ) from app.core.txn import best_effort from app.core.working_hours import resolve_timezone from app.identity.person_models import ( IdentityExternalIdentity, IdentityOrgMembership, IdentityOrgNode, ) logger = get_safe_logger(__name__) _REDIS_JTI_PREFIX = "refresh:jti:" # Rotation-grace cache. When the legitimate user re-presents an # already-consumed JTI within this window — StrictMode double-effect on # every page reload (dev), multi-tab refresh fan-out, network-retry # storms — we replay the SAME rotated tokens that the first call # minted instead of treating it as token theft. Real attacker reuse # typically lands minutes to hours after the legitimate consume, well # past the grace, so RFC 6819 / OAuth 2.1 BCP defense still fires for # actual replay. # # 2026-06-02: added after a prod report ("刷新页面就退出登录") root-caused # to StrictMode double-firing AuthRoute's ``initialize()`` → # concurrent ``/auth/refresh`` calls → second one tripped Round-49's # ``force_logout_at`` flip → ALL access tokens issued by the first # call also failed ``_enforce_force_logout`` (iat < force_logout_at) # → user kicked out on every single refresh. The grace breaks that # loop without weakening reuse-detect for real theft. _REDIS_REFRESH_GRACE_PREFIX = "refresh:grace:" _REFRESH_GRACE_TTL_SECONDS = 15 # Atomic consume-and-grace. In ONE serialized Redis execution this: # * reads the JTI's owner (the user id stored by ``_store_refresh_jti``), # * deletes the JTI (enforces one-time use), and # * stamps a rotation-grace marker keyed by the SAME jti, holding the # owner id, with a short TTL. # # Because Redis runs Lua atomically and single-threaded, the instant ANY # other caller observes the JTI as already-consumed, the grace marker is # ALREADY present — the winner wrote it in the SAME atomic step. This # replaces the previous "delete now, write the marker only after minting # the new pair" sequence, which left a window where a genuinely # concurrent refresh (React StrictMode double-effect, multi-tab fan-out, # the apiClient-401-retry racing AuthRoute.initialize) read the marker # before it existed, fell through to reuse-detect, and flipped # ``force_logout_at`` — permanently bricking the account ("刷新页面就退出 # 登录": login still worked, but every subsequent /auth/refresh saw the # sticky force_logout_at and 401'd). With the atomic stamp a benign race # can NEVER trip reuse-detect — no poll, no TOCTOU window. Only a reuse # landing AFTER the marker's TTL (a real replay, typically seconds-to- # hours later) still flips force_logout_at (RFC 6819 / OAuth 2.1 BCP). # # Returns the owner user id on a winning consume, "" otherwise. _CONSUME_AND_GRACE_SCRIPT = """ local owner = redis.call('GET', KEYS[1]) if owner == false then return '' end redis.call('DEL', KEYS[1]) redis.call('SET', KEYS[2], owner, 'EX', tonumber(ARGV[1])) return owner """ # Opaque SSO failure message — every login-failure path that runs AFTER the # provider has already authenticated the user (e.g. binding suspended, # disabled, subject collision, post-takeover-guard) collapses to this one # string so a probing attacker cannot use the response body to enumerate # account state. Specific reasons are still captured server-side in the # audit_logs ``detail.reason`` field (router-side ``_audit_login_failure``). _LOGIN_FAILED_OPAQUE = "Login failed. Contact your administrator if this persists." # Server-side column allowlist for update_user(). Only these ORM fields may be # written via the generic admin PATCH path. Adding a column to UserUpdateRequest # without adding it here causes it to be silently dropped — safer than writing # to an unexpected column (e.g. force_logout_at, hris_person_id, deleted_at). # "department" is handled via _update_primary_department and is intentionally absent. _UPDATE_USER_WRITABLE_COLUMNS: frozenset[str] = frozenset({ "display_name", "work_email", "employment_status", }) async def _store_refresh_jti(token: str, user_id: str) -> None: """Decode refresh token, extract jti + exp, store in Redis with TTL. Fails closed: if Redis is unavailable, raises so that login fails cleanly rather than issuing an untracked refresh token whose JTI is not stored. An untracked JTI bypasses reuse-detection (the consume path always returns False → forced logout on first use), but more importantly the revocation guarantee breaks on Redis restart if we ever silently let logins succeed without writing the JTI. Redis is a declared hard dependency for auth (consume path already fails closed with 503). Startup health check enforces connectivity. """ # Reuse the central decode_token so the algorithm allowlist # stays in one place — never decode with settings.JWT_ALGORITHM # directly. from app.core.security import decode_token as _decode_token payload = _decode_token(token) jti = payload["jti"] exp = payload["exp"] now = int(datetime.now(UTC).timestamp()) ttl = max(exp - now, 1) redis = get_redis() await redis.setex(f"{_REDIS_JTI_PREFIX}{jti}", ttl, user_id) async def _consume_refresh_jti(jti: str) -> str | None: """Atomically consume a refresh JTI and stamp its rotation-grace marker. Returns the owner user id (str) if THIS call won the one-time consume, or ``None`` if the JTI was already consumed/missing — the caller then inspects the grace marker (guaranteed present for a legitimate concurrent rotation, absent for a genuine reuse past the grace TTL) to decide between re-minting and tripping force-logout. The consume + grace stamp happen in a single atomic Redis Lua execution (:data:`_CONSUME_AND_GRACE_SCRIPT`) so there is no window where the JTI is consumed but the marker not yet written. Fails closed: if Redis is unavailable, raises HTTP 503 rather than silently allowing token reuse. """ from fastapi import HTTPException try: redis = get_redis() owner = await redis.eval( _CONSUME_AND_GRACE_SCRIPT, 2, f"{_REDIS_JTI_PREFIX}{jti}", f"{_REDIS_REFRESH_GRACE_PREFIX}{jti}", _REFRESH_GRACE_TTL_SECONDS, ) # decode_responses=True → ``owner`` is a str ("" when the JTI was # already consumed/missing). Decode defensively in case a future # pool is created without decode_responses. if isinstance(owner, bytes): owner = owner.decode("utf-8") return owner or None except HTTPException: raise except Exception as exc: # Same scrub as _store_refresh_jti — log type only. logger.error( "Redis unavailable during token refresh: %s", type(exc).__name__, ) raise HTTPException(status_code=503, detail="Auth service temporarily unavailable") async def _fetch_refresh_grace(old_jti: str) -> str | None: """Return the user id stamped for ``old_jti`` if a recent legitimate rotation is still within grace, else ``None``. The marker is written atomically with the consume by :data:`_CONSUME_AND_GRACE_SCRIPT`, so for a genuinely concurrent rotation this read is guaranteed to find it — no poll needed. It holds ONLY the opaque user id, never the signed tokens (caching live bearer credentials at rest, even for 15s, would turn any Redis snapshot / replication / ``MONITOR`` leak into a credential leak; the racing loser re-mints its own pair server-side instead). Fails closed: a Redis read error returns ``None`` so the caller falls through to reuse-detect (force_logout_at) — never re-mint for a marker we can't verify. """ try: redis = get_redis() raw = await redis.get(f"{_REDIS_REFRESH_GRACE_PREFIX}{old_jti}") if raw is None: return None if isinstance(raw, bytes): raw = raw.decode("utf-8") # Back-compat: an older build briefly stored a JSON blob carrying # the tokens; tolerate any still-live keys mid-deploy. if raw.startswith("{"): try: return json.loads(raw).get("sub") except Exception: # noqa: BLE001 return None return raw except Exception as exc: # noqa: BLE001 logger.warning( "Failed to read refresh-grace key — falling through to reuse-detect: %s", type(exc).__name__, ) return None async def _invalidate_refresh_jti(jti: str) -> None: """Remove a JTI from Redis unconditionally (used on logout). Fails open: if Redis is unavailable, logs a warning and returns so that logout still succeeds from the user's perspective. """ try: redis = get_redis() await redis.delete(f"{_REDIS_JTI_PREFIX}{jti}") except Exception as exc: logger.warning( "Failed to invalidate refresh JTI in Redis (degraded mode): %s", type(exc).__name__, ) # One force-logout flip per distinct reused JTI. See _claim_reuse_flip. _REDIS_REUSE_SEEN_PREFIX = "refresh:reuse_seen:" async def _claim_reuse_flip(jti: str, ttl_seconds: int) -> bool: """First-writer claim for the reuse-detect force-logout flip. Returns ``True`` only for the FIRST presentation of a given reused ``jti``; every later replay of the SAME dead JTI returns ``False``. Backing store is an atomic Redis ``SET NX`` so concurrent replays of one token still elect exactly one winner. Why this gate exists (2026-08-04 incident): the reuse-detect branch used to flip ``force_logout_at = now()`` on EVERY presentation of a dead JTI. A script re-presenting ONE leaked/rotated refresh token in a tight loop (observed: ``python-requests`` hitting ``/auth/refresh`` every ~12s) therefore kept pushing the force-logout cutoff forward forever. Because a successful interactive login CLEARS ``force_logout_at`` (:func:`_clear_stale_force_logout`), the victim could re-login — but the next replay a few seconds later re-bumped the cutoff past their fresh token's ``iat`` and kicked them again: a single leaked refresh token became a perpetual account-lockout DoS. Flipping (and auditing) once per distinct reused JTI keeps RFC 6819 / OAuth 2.1 reuse-detect intact for genuinely-distinct stolen tokens while denying the amplification. Fails TOWARD detection: on a Redis error we return ``True`` so a real reuse is never silently swallowed — a redundant flip is strictly safer than a missed theft signal. """ try: redis = get_redis() got = await redis.set( f"{_REDIS_REUSE_SEEN_PREFIX}{jti}", "1", nx=True, ex=max(int(ttl_seconds), 1), ) return bool(got) except Exception as exc: # noqa: BLE001 — fail toward detection logger.warning( "reuse-seen claim failed (%s) — flipping force_logout anyway", type(exc).__name__, ) return True def _pick_provider_ext( external_identities, provider: str, ) -> "IdentityExternalIdentity | None": """Choose the most-current ext_identity row for a given provider. Prefers a LIVE row (``deleted_at IS NULL``) over a tombstoned one when both exist — necessary because ``user.external_identities`` iteration order is undefined and a careless ``next(... provider ...)`` could pick the tombstone, trigger the resurrection branch, and leave the live row co-existing with the revived tombstone → unique key violation on flush. Returns ``None`` when no row of this provider exists. """ live: IdentityExternalIdentity | None = None tombstone: IdentityExternalIdentity | None = None for ext in external_identities: if ext.provider != provider: continue if ext.deleted_at is None: live = ext break if tombstone is None: tombstone = ext return live if live is not None else tombstone def is_effectively_departed(user) -> bool: """Check if a user should be treated as departed. Returns True if: - ``hris_record_state == 'purged'`` — the upstream HRIS no longer carries this person at all (a full-roster reconciliation found the row missing and tombstoned it, see migration 222). Orthogonal to ``employment_status`` (an orphaned ``active`` row can be purged), so it must be checked independently: without this gate a purged person whose ``employment_status`` was never flipped to ``left`` could still obtain / refresh a Helix session and pass every protected route. The data-quality scanner's ``stale_tombstone`` check flags exactly this state as high-severity; this gate is what actually closes the login surface it warns about. - ``employment_status == 'left'``, OR - ``employment_status == 'leaver_pending'`` AND ``user`` is past their ``termination_date`` *in the user's own local timezone* (resolved from ``country`` via :func:`app.core.working_hours.resolve_timezone`). The local-timezone gate (2026-05-21) prevents a US employee from being locked out of Google SSO / Feishu / API at 2026-05-21 00:00 UTC — which is 2026-05-20 17:00 PDT, *before* their last working day even starts in LA. Without it the leaver loses their mailbox hours before they can send a farewell mail. The cron (``hr.overdue_leavers_check``) applies the same gate so this helper and the eventual ``employment_status='left'`` flip stay in lock-step. """ if user.hris_record_state == "purged": return True if user.employment_status == "left": return True if ( user.employment_status == "leaver_pending" and user.termination_date is not None ): tz = resolve_timezone(user) local_today = datetime.now(UTC).astimezone(tz).date() if local_today > user.termination_date: return True return False def _clear_stale_force_logout(user) -> None: """Clear a sticky ``force_logout_at`` on a successful interactive login. ``force_logout_at`` is flipped in exactly two places: the offboarding leaver-shutdown (legitimate) and the refresh-token reuse-detect path (:meth:`AuthService.refresh_tokens`). The reuse-detect flip CANNOT distinguish a genuine stolen-token replay from a benign "Redis lost the JTI" event — a Redis restart without persistence, a ``FLUSHDB``, a key eviction, or (in local dev) the documented ``docker compose down -v`` reset all drop the JTI, after which the legitimate user's next ``/auth/refresh`` looks identical to a replay and trips the flag. Once set, NO login path cleared it (only offboarding-reopen did), so every subsequent page refresh re-minted a token at login but was then kicked by the unconditional ``force_logout_at is not None`` guard in ``refresh_tokens`` — "刷新页面就退出登录", permanently, even across re-login. #681/#684 only closed the concurrent-rotation race, not this Redis-data-loss class. A successful interactive login (Google / Feishu / password) proves possession of the user's real SSO identity or password — strictly stronger evidence than a replayed refresh cookie — so it is the correct moment to release the brick. This is SAFE with respect to offboarding: every caller runs :func:`is_effectively_departed` and rejects departed users BEFORE reaching here, so a leaver can never self-clear an offboarding shutdown. The RFC 6819 / OAuth 2.1 BCP reuse-detect still fires on real theft — the attacker simply cannot perform an interactive login to undo it. """ if user.force_logout_at is not None: user.force_logout_at = None class AuthService: def __init__(self, db: AsyncSession): self.db = db async def find_user_by_email(self, email: str) -> AuthUser | None: # Case-insensitive comparison (pre-launch audit Round-29 fix). # Email local-parts are spec-case-sensitive but every mainstream # provider treats them case-insensitively in practice. A case- # sensitive lookup here would let a user with HRIS email # ``alice@example.com`` log in via Google with id_token email # ``Alice@example.com`` and trigger "no user found" — or worse, # if HRIS later wrote a mixed-case row, allow account collision. # Normalise the input and use ``func.lower`` on the column so # the comparison goes through MySQL's collation-agnostic path. normalised = (email or "").strip().lower() if not normalised: return None result = await self.db.execute( select(AuthUser) .options( selectinload(AuthUser.user_roles).selectinload(AuthUserRole.role), selectinload(AuthUser.subordinates), ) .where( func.lower(AuthUser.work_email) == normalised, AuthUser.deleted_at.is_(None), ) ) return result.scalar_one_or_none() @staticmethod def _has_active_login_identity(user: AuthUser) -> bool: """True iff ``user`` already has a live (non-tombstoned) external identity flagged as a login identity, for ANY provider. Anti-takeover gate for first-time Feishu binding through the self-editable ``email`` field (see :meth:`login_via_feishu`): an account already claimed by a real SSO login must never be re-resolved via an unverified, user-controlled attribute. ``external_identities`` is eager-loaded (``lazy='selectin'``) so this touches no extra query. """ return any( ext.deleted_at is None and ext.is_login_identity for ext in (user.external_identities or []) ) @staticmethod def _local_parts_match(addr_a: str, addr_b: str) -> bool: """True iff both addresses contain ``@``, have a non-empty local-part, and share the same local-part. Defence-in-depth helper for the anti-takeover guard in :func:`login_via_feishu`: * an address without ``@`` is not a valid RFC-5322 mailbox — must not pass via ``"foo".split("@", 1)[0] == "foo"``. * an address whose local-part is empty (``"@a.com"``, ``"@b.com"``) must not match itself to bypass the guard — both split to ``""`` and would otherwise compare equal. Both inputs are expected already normalised (lowercased, stripped). """ if "@" not in addr_a or "@" not in addr_b: return False local_a = addr_a.split("@", 1)[0] local_b = addr_b.split("@", 1)[0] if not local_a or not local_b: return False return local_a == local_b @staticmethod def _mobile_corroborates( feishu_mobile: str | None, person_mobile: str | None, ) -> bool: """True iff the Feishu account's admin-managed ``mobile`` matches the Helix person's stored ``mobile_phone``. Used as a second factor in :meth:`login_via_feishu` when the Feishu tenant sets no admin-bound ``enterprise_email`` (e.g. the PingCAP JP tenant) so the only email signal is the self-editable ``email`` field. An impersonator can set their self-editable ``email`` to a victim's work_email, but cannot make their own Feishu account's mobile equal the victim's real phone number (the mobile is the account's admin/HR-managed login number, not free text) — so a mobile match corroborates that the authenticating Feishu account genuinely belongs to the resolved Helix person. Feishu returns an international number (``+``); Helix stores the national number, often with a trunk ``0`` (JP ``09064764314``, US/CN without). Both are reduced to digits, an international ``+``/``00`` prefix and a single national trunk ``0`` are dropped, then the Feishu digits must END WITH the person's national digits — Feishu still carries the country code the stored national value omits. A minimum of 8 significant digits guards against trivial suffix collisions (the email already matched exactly; this is corroboration, not the sole signal). """ def _digits(raw: str | None, *, strip_trunk: bool) -> str | None: if not isinstance(raw, str): return None d = "".join(ch for ch in raw if ch.isdigit()) # A ``+`` prefix was dropped by the digit filter; a ``00`` # international prefix carries the same meaning — drop it so # both shapes align on ````. if d.startswith("00"): d = d[2:] if strip_trunk and d.startswith("0"): d = d[1:] return d or None fe = _digits(feishu_mobile, strip_trunk=False) pe = _digits(person_mobile, strip_trunk=True) if not fe or not pe or len(pe) < 8: return False return fe == pe or fe.endswith(pe) async def _safe_swap_subject( self, ext: IdentityExternalIdentity, new_subject: str, ) -> None: """Rotate ``ext.provider_subject`` to ``new_subject`` with a pre-flight collision check. The ``(provider, provider_subject)`` UNIQUE constraint on ``identity_external_identities`` is **unfiltered** by ``deleted_at`` — both live and tombstoned rows occupy the same unique slot at the DB level. The collision SELECT below therefore spans the whole table on purpose: a tombstoned row elsewhere holding ``new_subject`` would still raise ``IntegrityError(1062)`` on flush if we just assigned. By pre-checking and raising a clean ``ValueError``, the caller surfaces a 401 ``LOGIN_FAILED`` instead of 500. Why pre-check rather than post-assign + catch IntegrityError: SQLAlchemy holds the pending mutation in the session even after the exception, so a later flush from another code path could replay it. Pre-checking keeps the session clean on the rejection path. """ if ext.provider_subject == new_subject: return # Collision check spans the WHOLE table — tombstoned rows still # occupy the unique slot at the DB level. A future PR may want # to undelete + repoint instead of raising; for now, fail safe. existing = await self.db.scalar( select(IdentityExternalIdentity.id).where( IdentityExternalIdentity.provider == ext.provider, IdentityExternalIdentity.provider_subject == new_subject, IdentityExternalIdentity.id != ext.id, ) ) if existing is not None: raise ValueError(_LOGIN_FAILED_OPAQUE) ext.provider_subject = new_subject @staticmethod def _should_replace_subject( stored: str | None, preferred: str | None, weak: str | None, *, prefer_prefix: str | None = None, weak_prefix: str | None = None, ) -> str | None: """Decide whether to overwrite ``stored`` with a newer subject. Returns the new value to write, or ``None`` to leave ``stored`` unchanged. Rules: * If ``preferred`` is present (e.g. Feishu ``union_id``) and differs from ``stored``, return ``preferred`` — upgrade / rotate to the more durable id. * If only ``weak`` (e.g. ``open_id``) is available AND ``stored`` is empty OR ``stored`` is shaped like a weak id (``weak_prefix``), return ``weak``. Never downgrade a stored preferred-shape (``prefer_prefix``) or any non-weak-looking legacy value to a transient ``weak`` just because the current login lost the strong scope — the previous behaviour (prefix-mismatch-as-weak heuristic) silently downgraded legacy non-prefixed values like ``'uuid-format-subject'`` to a freshly-rotated ``ou_…`` open_id. * Otherwise return ``None`` (no change). """ if preferred and stored != preferred: return preferred if weak and stored != weak: if not stored: return weak # Only swap into a weak subject when the stored value is # ALSO weak-shaped (e.g. ``ou_…``). A non-weak-prefixed # legacy value is treated as opaque / immutable from this # path — protects against downgrading durable identifiers # that happen to lack the modern prefix. if weak_prefix is not None and stored.startswith(weak_prefix): return weak if prefer_prefix is not None and stored.startswith(prefer_prefix): return None # stored is preferred-shape — never downgrade # Stored doesn't match either shape — opaque legacy value, # leave alone. return None return None async def find_user_by_external_subject( self, provider: str, subjects: list[str], ) -> AuthUser | None: """Resolve a person via an existing ``identity_external_identities`` row. Tries each subject in **list order** as a separate query — preserves the caller's priority signal (e.g. Feishu passes ``[union_id, open_id]`` so a row keyed on the more durable ``union_id`` wins over an older ``open_id``-only row even when both happen to exist for different persons due to migration artefacts). A single ``IN(...)+ LIMIT 1`` query would be at the mercy of MySQL's storage order and could non-deterministically return the wrong row in such cases. Filters: * ``deleted_at IS NULL`` — tombstoned bindings (HR cleanup leftovers) cannot resolve a user. * ``identity_status = 'active'`` — explicitly-suspended rows cannot resolve a user. * ``is_login_identity = True`` — admin-disabled bindings cannot resolve a user (set ``is_login_identity = False`` on a row to revoke its login capability without deleting it). * Person ``deleted_at IS NULL`` — hard-deleted persons cannot authenticate. Subjects are deduplicated before the loop because some Feishu app configurations return identical ``union_id`` and ``open_id`` values — running the same SELECT twice gains nothing. Returns ``None`` when none of the candidate subjects matches a live, enabled binding — the caller then falls through to the email-match path (first-time binding) or hard-fails. """ seen: set[str] = set() for raw in subjects: if not isinstance(raw, str) or not raw or raw in seen: continue seen.add(raw) result = await self.db.execute( select(AuthUser) .join( IdentityExternalIdentity, IdentityExternalIdentity.person_id == AuthUser.id, ) .options( selectinload(AuthUser.user_roles).selectinload(AuthUserRole.role), selectinload(AuthUser.subordinates), ) .where( IdentityExternalIdentity.provider == provider, IdentityExternalIdentity.provider_subject == raw, IdentityExternalIdentity.deleted_at.is_(None), IdentityExternalIdentity.identity_status == "active", IdentityExternalIdentity.is_login_identity.is_(True), AuthUser.deleted_at.is_(None), ) .limit(1) ) user = result.scalars().first() if user is not None: return user return None async def find_user_by_id(self, user_id: UUID) -> AuthUser | None: result = await self.db.execute( select(AuthUser) .options( selectinload(AuthUser.user_roles).selectinload(AuthUserRole.role), selectinload(AuthUser.subordinates), ) .where(AuthUser.id == user_id, AuthUser.deleted_at.is_(None)) ) return result.scalar_one_or_none() async def login_via_google(self, google_payload: dict) -> tuple[AuthUser, str, str]: """Match Google user to identity_persons, issue tokens. Returns (user, access_token, refresh_token). Resolution order: 1. **Subject-first**: an existing ``(google, sub)`` external-identity row pinpoints the Person, regardless of whether the id_token's ``email`` field still matches the HRIS work_email (HRIS rename, Google Workspace primary-alias swap, etc.). 2. **Email** (first-time binding only): falls back to case-insensitive ``identity_persons.work_email`` lookup with the id_token's verified email. The ``email_verified`` claim is already enforced upstream in :func:`verify_google_id_token`, so an unverified address never reaches the email path. """ # Normalise email at the SSO boundary so external-identity row # and audit logs all carry a consistent lowercase form even if # Google ever changes its case-preservation behaviour. Round-29 # pre-launch audit fix paired with the case-insensitive lookup. email_raw = google_payload.get("email") or "" email = email_raw.strip().lower() if isinstance(email_raw, str) else "" google_sub = google_payload.get("sub") if not isinstance(google_sub, str) or not google_sub: raise ValueError("Google id_token missing sub") # Step 1: subject-first. Survives email rotations. user = await self.find_user_by_external_subject("google", [google_sub]) # Step 2: email fallback for first-time binding. if not user: if not email: raise ValueError( "Google id_token has no email and no prior subject binding" ) user = await self.find_user_by_email(email) if not user: raise ValueError( f"No user found for email {email}. User must be synced from HRIS first." ) if is_effectively_departed(user): raise ValueError("User account has been deactivated") # Upsert the Google external identity. ``user.external_identities`` # is eager-loaded (``lazy='selectin'``) and includes BOTH live # and soft-deleted rows — pick up any tombstoned row here so we # can revive it instead of attempting an INSERT that would # collide on ``uq_identity_external_person_provider`` # (constraint covers (person_id, provider) regardless of # ``deleted_at``). This mirrors the HR-side fix landed on # 2026-05-26 for the analogous SF-sync collision. # # PREFER a live row over a tombstone — the iteration order of # ``user.external_identities`` is undefined, and a careless # ``next(... google ...)`` could pick the tombstone even when # a live row exists, then re-revive it and leave the live row # alone → flush hits the unique key 1062. google_ext = _pick_provider_ext(user.external_identities, "google") if google_ext is not None and google_ext.deleted_at is None: # Live row exists — enforce admin-side disable signals before # the subject-first lookup that bypassed it (if any) decides # to log the user in. The lookup ALREADY filters these flags, # so reaching this branch with a non-active / disabled row # means the email-fallback path resolved the user. Refuse to # silently bind through a row whose admin explicitly turned # off — surface as a clean LOGIN_FAILED 401, not a partial # write. if google_ext.identity_status != "active": raise ValueError( "Google login binding is suspended — contact your " "administrator to re-activate." ) if not google_ext.is_login_identity: raise ValueError( "Google login is disabled for this account — contact " "your administrator." ) if not google_ext: # First-time bind: pre-flight collision check so a tombstoned # row on ANOTHER person holding this sub doesn't crash the # flush with IntegrityError → 500. colliding = await self.db.scalar( select(IdentityExternalIdentity.id).where( IdentityExternalIdentity.provider == "google", IdentityExternalIdentity.provider_subject == google_sub, ) ) if colliding is not None: raise ValueError( "google subject already bound to another identity row" ) google_ext = IdentityExternalIdentity( person_id=user.id, provider="google", provider_subject=google_sub, provider_email=email or None, is_login_identity=True, ) self.db.add(google_ext) else: # Validate disable signals BEFORE mutating anything — so a # rejected login leaves the SQLAlchemy session unchanged. if google_ext.deleted_at is None: if not google_ext.is_login_identity: raise ValueError( "google login disabled for this account" ) if google_ext.identity_status != "active": raise ValueError( f"google login binding has identity_status=" f"{google_ext.identity_status!r}" ) else: # Tombstoned row. Reviving must respect ALL admin # kill-switches: if the admin both soft-deleted AND # disabled / suspended, the revival must not silently # log the user in on the very same session that # triggered the revival. Validate BEFORE clearing # ``deleted_at`` so a rejection leaves the session # pristine. if not google_ext.is_login_identity: raise ValueError( "google login disabled (tombstoned + disabled)" ) if google_ext.identity_status != "active": raise ValueError( f"google login binding tombstoned + " f"identity_status={google_ext.identity_status!r}" ) # Pre-flight subject swap collision check BEFORE any field # mutation. If it raises, the session is unchanged. await self._safe_swap_subject(google_ext, google_sub) # Field updates now safe to apply. if google_ext.deleted_at is not None: google_ext.deleted_at = None # revive # Keep ``provider_email`` synced with the latest verified # value from Google so the row reflects what was actually # used at last login (useful for audit / forensics). if email and google_ext.provider_email != email: google_ext.provider_email = email # Deliberately DO NOT re-write ``is_login_identity`` on an # already-live row — preserve an admin's explicit disable # (False) until they revert it. user.last_login_at = datetime.now(UTC) _clear_stale_force_logout(user) picture = google_payload.get("picture") if isinstance(picture, str) and picture: user.avatar_url = picture await self.db.flush() access_token = create_access_token( str(user.id), user.role_names, employment_status=user.employment_status, ) refresh_token = create_refresh_token(str(user.id)) await _store_refresh_jti(refresh_token, str(user.id)) return user, access_token, refresh_token async def login_via_feishu( self, feishu_user_info: dict, ) -> tuple[AuthUser, str, str]: """Match Feishu user to identity_persons, issue tokens. Returns (user, access_token, refresh_token). Resolution order: 1. **Subject-first** by ``(feishu, union_id)`` or ``(feishu, open_id)``. If a prior login bound either subject for the user, that row pinpoints the Person — even if the Feishu admin has since changed the ``enterprise_email`` field (the 2026-05-29 prod incident: admin rotated PingCAP ``enterprise_email`` from ``@pingcap.com`` → ``@pingcap.com.cn``, locking every already-bound user out under the old email-only flow). 2. **Email** (first-time binding only): tries ``enterprise_email`` (admin-bound, harder to forge), then ``email`` (user-self-editable). The ``email`` fallback is gated by a **local-part match** against ``enterprise_email`` to defeat pre-takeover: a Feishu user with ``enterprise_email=a@...`` cannot bind to a Helix person whose ``work_email=b@...`` by setting their self-editable ``email`` field to ``b@...``. When the tenant sets NO ``enterprise_email`` at all (so ``email`` is the only signal) and the resolved Person already holds an active login identity, binding additionally requires the Feishu account's admin-managed ``mobile`` to corroborate the Person's stored ``mobile_phone`` (see :meth:`_mobile_corroborates`) — an impersonator cannot forge the victim's real phone. ``union_id`` is preferred over ``open_id`` for new bindings — union_id is stable across multiple apps from the same developer, so a future split into a dedicated SSO app would not orphan existing external-identity rows. Both are accepted for *lookup* because historic rows may hold either depending on the scope granted when the binding was first written. """ from app.auth.feishu_sso import pick_match_emails raw_union_id = feishu_user_info.get("union_id") raw_open_id = feishu_user_info.get("open_id") union_id = raw_union_id if isinstance(raw_union_id, str) and raw_union_id else None open_id = raw_open_id if isinstance(raw_open_id, str) and raw_open_id else None if not union_id and not open_id: raise ValueError("Feishu user_info missing union_id/open_id") primary_subject = union_id or open_id # union_id preferred when both present enterprise_email, email_field = pick_match_emails(feishu_user_info) # Step 1: subject-first. A prior binding short-circuits the # email check, so admin-side enterprise_email churn does not # lock the user out. subjects = [s for s in (union_id, open_id) if s] user = await self.find_user_by_external_subject("feishu", subjects) # ``matched_email`` is the value we'll persist into # ``provider_email`` for the binding row. Subject-path uses # whichever is present (admin-bound preferred); email-path uses # whichever value resolved the Person. matched_email: str | None = enterprise_email or email_field if not user: # Step 2: first-time binding via email. if not enterprise_email and not email_field: raise ValueError( "Feishu account has no enterprise_email or email — " "cannot match to a Helix person" ) # Try admin-bound enterprise_email first. Its trust level is # higher (only a Feishu tenant admin can set it), so a hit # binds the Helix Person unconditionally. if enterprise_email: user = await self.find_user_by_email(enterprise_email) if user: matched_email = enterprise_email # Fall back to the self-editable ``email`` field. Gated by # local-part equality with ``enterprise_email`` when the # latter is present and didn't resolve — see docstring for # the pre-takeover rationale. ``_local_parts_match`` requires # ``@`` in both addresses so a pathological payload with no # ``@`` can't pass via ``"foo".split("@",1)[0] == "foo"``. if not user and email_field and email_field != enterprise_email: candidate = await self.find_user_by_email(email_field) if candidate: if enterprise_email: if self._local_parts_match(enterprise_email, email_field): user = candidate matched_email = email_field # else: silently reject — surfaces as the # generic "no user found" 401 below, so an # attacker probing for pre-takeover gets the # same response shape as a legitimate miss. else: # enterprise_email empty → the self-editable # ``email`` field is the ONLY signal, and it is # fully user-controlled in Feishu. Auto-linking an # already-claimed account through an unverified # attribute is the classic account-linking / # pre-hijacking takeover: a Feishu tenant guest with # no admin-bound enterprise_email sets their # self-editable ``email`` to a victim's work_email # and logs in as them (Google's path is safe because # it enforces ``email_verified``; Feishu's ``email`` # carries no such trust). Only bind when the resolved # Person has NEVER been claimed by an active login # identity — a genuine first-time external guest can # still self-serve, but an employee who has ever # authenticated (Google or Feishu) can no longer be # impersonated this way. (Security review 2026-07-22.) # # 2026-08-03: the "never claimed" gate alone locked # out every employee whose Feishu tenant sets no # enterprise_email (the whole PingCAP JP tenant) once # onboarding auto-provisioned them a Google login # identity — a first-ever Feishu login then fell into # this branch, saw the (unused) Google binding, and # was silently rejected. Add a mobile second factor: # bind when the Feishu account's admin-managed mobile # ALSO matches the person's stored mobile_phone. An # impersonator cannot forge the victim's real phone, # so this admits the genuine person without weakening # the takeover guard (email-only, mobile-mismatch → # still rejected). if ( not self._has_active_login_identity(candidate) or self._mobile_corroborates( feishu_user_info.get("mobile"), candidate.mobile_phone, ) ): user = candidate matched_email = email_field # else: silently reject → falls through to the # generic "no user found" 401 below, the same # response shape as a legitimate miss (no # takeover-probe oracle). if not user: # Use whichever address the caller supplied as the audit # label — ``email_field`` first because that's what most # legitimate users hit when the admin-bound # ``enterprise_email`` rotates and ``email`` survives, # falling back to enterprise_email so the 401 message # always names *something* identifiable rather than a # ```` placeholder. If both are absent the path # above already raised with a clearer message. label = email_field or enterprise_email raise ValueError( f"No user found for email {label}. User must be synced " f"from HRIS first." ) if is_effectively_departed(user): raise ValueError("User account has been deactivated") # Upsert the Feishu external-identity row. ``external_identities`` # is eager-loaded (``lazy='selectin'``) including soft-deleted # rows — pick up any tombstone here and revive it instead of # blindly INSERTing into a ``(person_id, provider)`` slot # already taken by the soft-deleted row (which would raise # ``uq_identity_external_person_provider`` 1062 — same shape # as the HR-side prod incident on 2026-05-26). # Live row preferred over tombstone (see helper). feishu_ext = _pick_provider_ext(user.external_identities, "feishu") if not feishu_ext: # First-time bind: pre-flight collision check on the # primary_subject so a tombstoned row on another person # doesn't crash the flush. colliding = await self.db.scalar( select(IdentityExternalIdentity.id).where( IdentityExternalIdentity.provider == "feishu", IdentityExternalIdentity.provider_subject == primary_subject, ) ) if colliding is not None: raise ValueError( "feishu subject already bound to another identity row" ) feishu_ext = IdentityExternalIdentity( person_id=user.id, provider="feishu", provider_subject=primary_subject, provider_email=matched_email, is_login_identity=True, ) self.db.add(feishu_ext) else: # Validate disable signals BEFORE mutating anything — a # rejected login leaves the session unchanged. if feishu_ext.deleted_at is None: if feishu_ext.identity_status != "active": raise ValueError( f"feishu login binding has identity_status=" f"{feishu_ext.identity_status!r}" ) if not feishu_ext.is_login_identity: raise ValueError( "feishu login disabled for this account" ) else: # Tombstone — every admin kill-switch must survive the # revival. Validate BEFORE clearing ``deleted_at`` so a # rejection leaves the session pristine. if not feishu_ext.is_login_identity: raise ValueError( "feishu login disabled (tombstoned + disabled)" ) if feishu_ext.identity_status != "active": raise ValueError( f"feishu login binding tombstoned + " f"identity_status={feishu_ext.identity_status!r}" ) # Subject swap rules (see ``_should_replace_subject``): # - ``union_id`` present → always upgrade / rotate to it. # - Only ``open_id`` available → swap ONLY if the stored # value is itself an ``ou_`` open_id; never downgrade # a stored ``on_`` union_id just because this particular # login lost the union_id scope. new_subject = self._should_replace_subject( feishu_ext.provider_subject, preferred=union_id, weak=open_id, prefer_prefix="on_", # Feishu union_id is ``on_…`` weak_prefix="ou_", # Feishu open_id is ``ou_…`` ) if new_subject is not None: # Pre-flight subject collision BEFORE clearing deleted_at # — keeps session pristine if it raises. await self._safe_swap_subject(feishu_ext, new_subject) if feishu_ext.deleted_at is not None: feishu_ext.deleted_at = None # revive after validation if matched_email and feishu_ext.provider_email != matched_email: feishu_ext.provider_email = matched_email # Deliberately DO NOT re-write ``is_login_identity`` on an # already-live row — preserve an admin's explicit disable # until they revert it. Reviving a tombstone above is the # only case where the flag is read (to gate the login). user.last_login_at = datetime.now(UTC) _clear_stale_force_logout(user) avatar = feishu_user_info.get("avatar_url") if isinstance(avatar, str) and avatar: user.avatar_url = avatar await self.db.flush() access_token = create_access_token( str(user.id), user.role_names, employment_status=user.employment_status, ) refresh_token = create_refresh_token(str(user.id)) await _store_refresh_jti(refresh_token, str(user.id)) return user, access_token, refresh_token async def login_via_password(self, email: str, password: str) -> tuple[AuthUser, str, str]: """Authenticate with email + password, issue tokens.""" user = await self.find_user_by_email(email) # Always run a (dummy) bcrypt check so the response time does not # reveal whether the account exists — mitigates email enumeration. hashed = user.hashed_password if user else None if not verify_password_constant_time(password, hashed) or not user: raise ValueError("Invalid email or password") if is_effectively_departed(user): raise ValueError("User account has been deactivated") user.last_login_at = datetime.now(UTC) _clear_stale_force_logout(user) await self.db.flush() access_token = create_access_token( str(user.id), user.role_names, employment_status=user.employment_status, ) refresh_token = create_refresh_token(str(user.id)) await _store_refresh_jti(refresh_token, str(user.id)) return user, access_token, refresh_token async def refresh_tokens(self, payload: dict) -> tuple[AuthUser, str, str]: """Rotate refresh token with JTI validation. Validates the old JTI exists in Redis (one-time use), consumes it, then issues a new access + refresh token pair. Raises ValueError on invalid/already-used JTI (possible token theft). """ jti = payload.get("jti") user_id = payload.get("sub") if not jti or not user_id: raise ValueError("Invalid refresh token payload") # Atomically consume the JTI and stamp its rotation-grace marker # (see _CONSUME_AND_GRACE_SCRIPT). ``None`` means we did NOT win # the one-time consume — another in-flight refresh got there first. owner_uid = await _consume_refresh_jti(jti) if owner_uid is None: # Legitimate-race grace (2026-06-02, hardened 2026-06-02): the # SAME JTI was just rotated. React StrictMode double-effects, # multi-tab refresh fan-out, and the apiClient-401-retry racing # AuthRoute.initialize all hit this path on a healthy session. # The winner stamped the grace marker ATOMICALLY with its # consume, so for a genuine concurrent rotation the marker is # already present here — no poll, no TOCTOU window. A marker # that is absent can only mean the JTI's grace TTL already # elapsed: a real replay (seconds-to-hours after the # legitimate rotation) → token theft → trips the # ``force_logout_at`` defense below. So a benign browser race # can NEVER force-log the user out, while real reuse still does. graced_sub = await _fetch_refresh_grace(jti) if graced_sub == user_id: user = await self.find_user_by_id(UUID(user_id)) if user and not is_effectively_departed(user) and user.force_logout_at is None: # Legitimate concurrent refresh: mint THIS caller its # own fresh pair (independent session) rather than # replaying the winner's tokens. Avoids two clients # sharing one JTI (which would re-trip reuse-detect # once they diverge past the grace window) and avoids # ever caching live tokens at rest. access_token = create_access_token( str(user.id), user.role_names, employment_status=user.employment_status, ) refresh_token = create_refresh_token(str(user.id)) await _store_refresh_jti(refresh_token, str(user.id)) return user, access_token, refresh_token # Elect a SINGLE force-logout flip per distinct reused JTI. # Without this, a script replaying one dead refresh token in a # loop re-bumps ``force_logout_at`` to now() on every hit; since # a successful login clears the flag, the victim gets kicked a # few seconds after each re-login — one leaked token = perpetual # account lockout (2026-08-04 incident). The marker's TTL is the # token's own remaining lifetime: a JTI past ``exp`` can't be # presented again (``decode_token`` rejects it), so re-flagging # after that is moot. reused_exp = payload.get("exp") now_ts = int(datetime.now(UTC).timestamp()) seen_ttl = max( (int(reused_exp) - now_ts) if reused_exp else _REFRESH_GRACE_TTL_SECONDS, 1, ) first_flip = await _claim_reuse_flip(jti, seen_ttl) if not first_flip: # Repeat replay of an already-flagged JTI: the family was # already severed on the first detection. Re-flipping here # is what enabled the lockout amplification, so make it a # no-op (still 401 the caller below). logger.info( "Refresh token reuse: replay of already-flagged JTI — " "skipping redundant force_logout flip. user_id=%s jti=%s", user_id, jti, ) raise ValueError("Refresh token already used or invalid") logger.warning( "Refresh token reuse detected or JTI missing — possible token theft. " "user_id=%s jti=%s", user_id, jti, ) # Round-49: RFC 6819 / OAuth 2.1 BCP — on detected reuse, # invalidate the entire refresh-token family for this user. # The previous behaviour only refused the replayed token; # the attacker's freshly-issued chain (from their race-win) # stayed valid for the full 7-day TTL. Flipping # ``force_logout_at`` severs every JWT + PAT the # ``_enforce_force_logout`` middleware sees, plus an audit # row written in an independent session so the forensic # trail survives the upcoming ValueError + caller rollback. try: # Flip force_logout_at in an independent session so # the side effect survives even if the caller's # context rolls back on the upcoming ``raise``. from app.core.database import async_session_factory async with async_session_factory() as new_session: try: user = await new_session.get(AuthUser, UUID(user_id)) if user is not None: user.force_logout_at = datetime.now(UTC) await new_session.commit() except Exception: await new_session.rollback() raise from app.audit.service import log_denied_action await log_denied_action( trace_id=jti, actor_type="system", actor_user_id=UUID(user_id), action="auth.refresh_token_reuse_detected", resource_type="auth_user", resource_id=user_id, detail={ "jti": jti, "force_logout_set": True, "rfc": "RFC 6819 / OAuth 2.1 BCP", }, ) except Exception: # noqa: BLE001 — best-effort logger.exception( "auth.refresh_token_reuse_detected: failed to set " "force_logout_at for user_id=%s — re-raising the " "ValueError anyway so the attacker's call still fails", user_id, ) raise ValueError("Refresh token already used or invalid") user = await self.find_user_by_id(UUID(user_id)) if not user or is_effectively_departed(user): raise ValueError("User not found or deactivated") if user.force_logout_at is not None: raise ValueError("Account access has been revoked") access_token = create_access_token( str(user.id), user.role_names, employment_status=user.employment_status, ) refresh_token = create_refresh_token(str(user.id)) await _store_refresh_jti(refresh_token, str(user.id)) # NOTE: the rotation-grace marker for ``jti`` was already stamped # ATOMICALLY with the consume above (_CONSUME_AND_GRACE_SCRIPT), so # a concurrent legitimate re-present of the SAME old JTI is # recognised as a race — no separate write needed here. return user, access_token, refresh_token async def logout(self, payload: dict) -> None: """Invalidate the refresh token JTI on logout.""" jti = payload.get("jti") if jti: await _invalidate_refresh_jti(jti) async def list_users( self, page: int = 1, page_size: int = 20, search: str | None = None ) -> tuple[list[AuthUser], int]: query = ( select(AuthUser) .options( selectinload(AuthUser.user_roles).selectinload(AuthUserRole.role), selectinload(AuthUser.subordinates), ) .where(AuthUser.deleted_at.is_(None)) ) count_query = select(func.count()).select_from(AuthUser).where(AuthUser.deleted_at.is_(None)) if search: like_pattern = f"%{search}%" filter_clause = ( AuthUser.work_email.ilike(like_pattern) | AuthUser.display_name.ilike(like_pattern) | AuthUser.name_pinyin.ilike(like_pattern) | AuthUser.name_pinyin_abbr.ilike(like_pattern) | AuthUser.employee_no.ilike(like_pattern) ) query = query.where(filter_clause) count_query = count_query.where(filter_clause) total_result = await self.db.execute(count_query) total = total_result.scalar() or 0 query = query.order_by(AuthUser.created_at.desc()).offset((page - 1) * page_size).limit(page_size) result = await self.db.execute(query) users = list(result.scalars().all()) return users, total async def update_user(self, user_id: UUID, data: dict) -> AuthUser: user = await self.find_user_by_id(user_id) if not user: raise NotFoundError("User", str(user_id)) # Map backward-compatible field names to new model fields field_mapping = { "status": "employment_status", "email": "work_email", } # Fields that are now read-only properties (no longer stored on person) readonly_properties = {"position"} # Pre-launch P0 fix (Round 45 follow-up, CR-312): refuse to # resurrect a ``left`` user via the generic admin PATCH path. # The schema's ``AdminAdjustableEmploymentStatus`` Literal # didn't include ``"left"`` (so an admin can't put someone # there) but DID allow ``"active"`` regardless of the current # state — letting an admin flip ``left → active`` here would # silently bypass ``cancel_process`` (no SSO un-suspend, no # blocker re-grant, no audit chain of resurrection events). # The legitimate resurrection path is # ``OffboardingService.cancel_process`` only. incoming_status = data.get("status") or data.get("employment_status") if ( incoming_status and incoming_status != user.employment_status and (user.employment_status or "").lower() == "left" ): from app.core.exceptions import AppException # noqa: PLC0415 raise AppException( code="RESURRECT_VIA_OFFBOARDING_CANCEL_ONLY", message=( f"Refusing to flip {user.work_email!r} from 'left' " f"to {incoming_status!r} via the generic user PATCH. " "Resurrection requires the offboarding cancel flow " "so SSO suspend / seat revokes / blocker rows are " "rolled back consistently." ), status_code=409, ) for key, value in data.items(): if value is not None: actual_key = field_mapping.get(key, key) # Handle department update via org_membership if key == "department": await self._update_primary_department(user.id, value) elif key in readonly_properties: # Position is no longer stored on person; skip silently pass elif actual_key in _UPDATE_USER_WRITABLE_COLUMNS and hasattr(user, actual_key): setattr(user, actual_key, value) # Unrecognised keys are silently dropped — schema already # validates field names; this is a defence-in-depth guard. await self.db.flush() return user async def _update_primary_department(self, person_id: UUID, department_name: str) -> None: """Update or create the primary org membership for a person.""" result = await self.db.execute( select(IdentityOrgMembership).where( IdentityOrgMembership.person_id == person_id, IdentityOrgMembership.is_primary.is_(True), IdentityOrgMembership.deleted_at.is_(None), ) ) membership = result.scalar_one_or_none() if membership: membership.org_node_name = department_name membership.org_node_source_id = department_name else: membership = IdentityOrgMembership( person_id=person_id, org_node_source_id=department_name, org_node_name=department_name, is_primary=True, ) self.db.add(membership) await self.db.flush() async def ensure_default_role(self, user_id: UUID) -> None: """Ensure a user has the default employee role.""" role_result = await self.db.execute( select(AuthRole).where(AuthRole.name == EMPLOYEE_ROLE) ) role = role_result.scalar_one_or_none() if not role: return existing = await self.db.execute( select(AuthUserRole).where( AuthUserRole.user_id == user_id, AuthUserRole.role_id == role.id ) ) if existing.scalar_one_or_none(): return user_role = AuthUserRole(user_id=user_id, role_id=role.id) self.db.add(user_role) await self.db.flush() async def assign_role(self, user_id: UUID, role_name: str, granted_by: UUID) -> AuthUserRole: user = await self.find_user_by_id(user_id) if not user: raise NotFoundError("User", str(user_id)) canonical_role_name = canonicalize_role_name(role_name) role_result = await self.db.execute( select(AuthRole).where(AuthRole.name == canonical_role_name) ) role = role_result.scalar_one_or_none() if not role: raise NotFoundError("Role", canonical_role_name) # Check if already assigned existing = await self.db.execute( select(AuthUserRole) .join(AuthRole, AuthRole.id == AuthUserRole.role_id) .where( AuthUserRole.user_id == user_id, AuthRole.name.in_(db_role_names_for(canonical_role_name)), ) ) if existing.scalar_one_or_none(): raise ValueError("Role already assigned") user_role = AuthUserRole(user_id=user_id, role_id=role.id, granted_by=granted_by) self.db.add(user_role) await self.db.flush() await self._sync_asset_admin_co_derivation(user_id, canonical_role_name) return user_role async def _sync_asset_admin_co_derivation( self, user_id: UUID, canonical_role_name: str, ) -> None: """``asset_global_admin`` co-derives ``asset_admin`` (the location-less global grant source — see ``location_admin_sync._person_should_hold_role``). Sync immediately on grant/revoke so the holder gains/loses the asset surfaces on their next request instead of waiting for the daily reconciler. Lives in the service (not the routers) so every future caller of assign/remove gets it for free. """ if canonical_role_name != ASSET_GLOBAL_ADMIN_ROLE: return from app.domains.asset.location_admin_sync import ( # noqa: PLC0415 sync_asset_admin_role_for_person, ) await sync_asset_admin_role_for_person(self.db, user_id) async def _assert_not_last_global_admin( self, user_id: UUID, role_being_removed: AuthRole, ) -> None: """Refuse to remove the last ``global_admin`` from the system. Pre-launch audit Round-29 fix. Without this guard a misclick or a buggy admin script could drop the only global_admin role, leaving Helix with no one able to grant roles, manage cloud integrations, run HRIS sync, or recover the state — every admin-gated endpoint would 403 and the system would have to be unwedged by a DB-level INSERT. The guard is purely additive: it only fires when the role being removed is a global_admin equivalent AND no other user holds one. """ # Only enforce for global_admin (and its legacy ``it_admin`` # equivalent). Other roles can be removed freely — losing the # last cloud_admin / auditor / finance role is recoverable # because a global_admin can re-grant them. from app.core.rbac import GLOBAL_ADMIN_ROLE, canonicalize_role_name if canonicalize_role_name(role_being_removed.name) != GLOBAL_ADMIN_ROLE: return # Count distinct users still holding any global_admin-equivalent # role AFTER excluding this user_id. Use the role name set so # both ``global_admin`` and the legacy ``it_admin`` row match. admin_role_names = db_role_names_for(GLOBAL_ADMIN_ROLE) remaining = await self.db.scalar( select(func.count(func.distinct(AuthUserRole.user_id))) .join(AuthRole, AuthRole.id == AuthUserRole.role_id) .where( AuthRole.name.in_(admin_role_names), AuthUserRole.user_id != user_id, ) ) or 0 if remaining < 1: from app.core.exceptions import AppException raise AppException( code="LAST_GLOBAL_ADMIN", message=( "Refusing to remove the last global_admin from the " "system. Grant another user global_admin first, " "then retry the removal." ), status_code=409, ) async def remove_role(self, user_id: UUID, role_id: UUID) -> None: result = await self.db.execute( select(AuthUserRole) .options(selectinload(AuthUserRole.role)) .where( AuthUserRole.user_id == user_id, AuthUserRole.role_id == role_id ) ) user_role = result.scalar_one_or_none() if not user_role: raise NotFoundError("UserRole", f"{user_id}/{role_id}") await self._assert_not_last_global_admin(user_id, user_role.role) removed_role_name = canonicalize_role_name(user_role.role.name) await self.db.delete(user_role) await self.db.flush() await self._sync_asset_admin_co_derivation(user_id, removed_role_name) async def remove_role_by_name(self, user_id: UUID, role_name: str) -> None: """Remove a role from a user by role name.""" canonical_role_name = canonicalize_role_name(role_name) roles_result = await self.db.execute( select(AuthRole).where(AuthRole.name.in_(db_role_names_for(canonical_role_name))) ) roles = list(roles_result.scalars().all()) if not roles: raise NotFoundError("Role", canonical_role_name) result = await self.db.execute( select(AuthUserRole) .options(selectinload(AuthUserRole.role)) .where( AuthUserRole.user_id == user_id, AuthUserRole.role_id.in_([role.id for role in roles]), ) ) # A user can legitimately hold MORE THAN ONE row that canonicalises # to the same role (e.g. both the legacy ``it_admin`` and the # canonical ``global_admin``). ``scalar_one_or_none()`` would raise # ``MultipleResultsFound`` there — not an ``AppException`` — which # propagates past the offboarding LAST_GLOBAL_ADMIN guard and wedges # ``complete_process``. Delete EVERY equivalent row (idempotent # across aliases) instead. user_roles = list(result.scalars().all()) if not user_roles: raise NotFoundError("UserRole", f"{user_id}/{canonical_role_name}") # Role-identity guard; asserting on any one equivalent row is # sufficient (they all canonicalise to the same role). await self._assert_not_last_global_admin(user_id, user_roles[0].role) for user_role in user_roles: await self.db.delete(user_role) await self.db.flush() await self._sync_asset_admin_co_derivation(user_id, canonical_role_name) async def revoke_platform_roles_for_leaver( self, user_id: UUID, *, trace_id: str, actor_user_id: UUID | None = None, ) -> list[str]: """Revoke every privileged platform role a leaver still holds. Closes the offboarding gap where a departed account (``employment_status='left'``) kept manually-granted elevated grants (global_admin / auditor / cloud_admin / hr / finance / management / asset_global_admin) indefinitely. ``employee`` and the reconciler-derived ``asset_admin`` are intentionally left alone — see ``LEAVER_REVOCABLE_ROLES``. Idempotent: only roles actually present are removed, so a second call (or a re-run after the role is already gone) returns ``[]``. Each removal reuses ``remove_role_by_name`` so the ``asset_global_admin`` → ``asset_admin`` co-derivation and the last-global-admin guard both still apply. ``LAST_GLOBAL_ADMIN`` is caught and logged, never raised — a leaver who happens to be the only remaining global_admin keeps the role (an admin must hand it over first) rather than wedging the offboarding-completion / reconciler caller. Returns the canonical role names actually revoked. """ from app.audit.service import AuditService # noqa: PLC0415 from app.core.exceptions import AppException # noqa: PLC0415 held = ( await self.db.execute( select(AuthRole.name) .join(AuthUserRole, AuthUserRole.role_id == AuthRole.id) .where(AuthUserRole.user_id == user_id) ) ).scalars().all() # Canonicalize so a legacy ``it_admin`` / ``security_auditor`` row # collapses onto its canonical name before the revocable check. revocable = sorted({ canonicalize_role_name(name) for name in held if canonicalize_role_name(name) in LEAVER_REVOCABLE_ROLES }) if not revocable: return [] audit = AuditService(self.db) revoked: list[str] = [] for role_name in revocable: try: await self.remove_role_by_name(user_id, role_name) except AppException as exc: if exc.code == "LAST_GLOBAL_ADMIN": logger.warning( "leaver %s holds the last global_admin; skipping " "role revoke (grant another user global_admin " "first, then re-run the cleanup)", user_id, ) continue raise revoked.append(role_name) await audit.log( trace_id=trace_id, actor_type="user" if actor_user_id is not None else "system", actor_user_id=actor_user_id, action="user.role.remove", status="success", resource_type="user_role", resource_id=str(user_id), detail={ "role": role_name, "reason": "offboarding_leaver_cleanup", }, ) return revoked async def list_roles(self) -> list[AuthRole]: result = await self.db.execute( select(AuthRole) .where(AuthRole.name.in_(CANONICAL_ROLE_NAMES)) .order_by(AuthRole.name) ) return list(result.scalars().all()) async def get_subordinate_ids(self, manager_id: UUID) -> list[UUID]: """Get all direct subordinate user IDs.""" result = await self.db.execute( select(AuthUser.id).where( AuthUser.manager_person_id == manager_id, AuthUser.deleted_at.is_(None) ) ) return list(result.scalars().all()) async def has_subordinates(self, user_id: UUID) -> bool: result = await self.db.execute( select(func.count()) .select_from(AuthUser) .where(AuthUser.manager_person_id == user_id, AuthUser.deleted_at.is_(None)) ) return (result.scalar() or 0) > 0 async def is_attendance_subject(self, user_id: UUID) -> bool: """True when this person has any intern-attendance report — as the intern (``person_id``) or as the reviewing leader (``leader_person_id``). Drives whether the optional 「我的考勤」 self-service menu shows: it stays hidden for the majority of staff who are neither interns nor intern-leaders, so the entry doesn't look odd to everyone. Raw EXISTS (both columns indexed) to avoid an auth→attendance domain import; returns False if the table isn't present yet so /auth/me never 500s on a partially-migrated DB.""" from sqlalchemy import text async with best_effort( self.db, label="service best-effort step", log=logger, ) as be_1677_step: row = await self.db.execute( text( "SELECT 1 FROM attendance_reports " "WHERE person_id = :pid OR leader_person_id = :pid LIMIT 1" ), {"pid": str(user_id)}, ) return row.first() is not None if not be_1677_step.ok: return False async def list_users_filtered( self, page: int = 1, page_size: int = 20, search: str | None = None, department: str | None = None, status: str | None = None, role: str | None = None, include_purged: bool = False, employee_type: str | None = None, ) -> tuple[list[AuthUser], int]: """List users with filtering by department, status, and role. ``include_purged`` toggles whether rows marked ``hris_record_state ='purged'`` (the HRIS-tombstone state from migration 222) are included. Default ``False`` keeps the list view clean — purged rows still exist for audit / FK preservation but represent people the upstream HRIS no longer carries. ``employee_type`` is a comma-separated list of SF picklist codes (``app.identity.employee_types``) — same comma shape as ``status``, so the frontend's multi-select serialises the same way for both. Codes are matched exactly; the value is never interpolated into SQL. """ purged_filter = AuthUser.hris_record_state == "present" query = ( select(AuthUser) .options( selectinload(AuthUser.user_roles).selectinload(AuthUserRole.role), selectinload(AuthUser.manager), selectinload(AuthUser.subordinates), ) .where(AuthUser.deleted_at.is_(None)) ) # COUNT(DISTINCT AuthUser.id): the ``role`` filter below joins # user_roles→role (one-to-many — a user can carry the same canonical # role across alias rows), so a plain COUNT(*) would tally a user # once per matching role row while the data query dedupes via # ``.scalars().unique()`` and shows them once. DISTINCT keeps # ``total`` aligned with the page contents. count_query = ( select(func.count(distinct(AuthUser.id))) .select_from(AuthUser) .where(AuthUser.deleted_at.is_(None)) ) if not include_purged: query = query.where(purged_filter) count_query = count_query.where(purged_filter) if search: like_pattern = f"%{search}%" filter_clause = ( AuthUser.work_email.ilike(like_pattern) | AuthUser.display_name.ilike(like_pattern) | AuthUser.name_pinyin.ilike(like_pattern) | AuthUser.name_pinyin_abbr.ilike(like_pattern) | AuthUser.employee_no.ilike(like_pattern) ) query = query.where(filter_clause) count_query = count_query.where(filter_clause) if department: # Join to org_memberships and roll the chosen department's whole # subtree up (pick a parent node in DepartmentTreeSelect → also # match every descendant). Shared matcher keeps this identical to # the cloud-overview / software-cost department filters. from app.core.org_scope import department_membership_subtree_clause subtree = department_membership_subtree_clause(department) query = query.join( IdentityOrgMembership, (IdentityOrgMembership.person_id == AuthUser.id) & (IdentityOrgMembership.is_primary.is_(True)) & (IdentityOrgMembership.deleted_at.is_(None)), ).where(subtree) count_query = count_query.join( IdentityOrgMembership, (IdentityOrgMembership.person_id == AuthUser.id) & (IdentityOrgMembership.is_primary.is_(True)) & (IdentityOrgMembership.deleted_at.is_(None)), ).where(subtree) if status: statuses = [s.strip() for s in status.split(",") if s.strip()] if len(statuses) == 1: query = query.where(AuthUser.employment_status == statuses[0]) count_query = count_query.where(AuthUser.employment_status == statuses[0]) else: query = query.where(AuthUser.employment_status.in_(statuses)) count_query = count_query.where(AuthUser.employment_status.in_(statuses)) if employee_type: codes = [c.strip() for c in employee_type.split(",") if c.strip()] if codes: type_clause = AuthUser.employee_type_code.in_(codes) query = query.where(type_clause) count_query = count_query.where(type_clause) if role: query = query.join(AuthUser.user_roles).join(AuthUserRole.role).where( AuthRole.name.in_(db_role_names_for(role)) ) count_query = count_query.join(AuthUser.user_roles).join(AuthUserRole.role).where( AuthRole.name.in_(db_role_names_for(role)) ) total_result = await self.db.execute(count_query) total = total_result.scalar() or 0 query = query.order_by(AuthUser.created_at.desc()).offset((page - 1) * page_size).limit(page_size) result = await self.db.execute(query) users = list(result.scalars().unique().all()) return users, total async def get_user_detail(self, user_id: UUID) -> AuthUser | None: """Get user with manager and subordinates loaded.""" result = await self.db.execute( select(AuthUser) .options( selectinload(AuthUser.user_roles).selectinload(AuthUserRole.role), selectinload(AuthUser.manager).selectinload(AuthUser.user_roles).selectinload(AuthUserRole.role), selectinload(AuthUser.subordinates).selectinload(AuthUser.user_roles).selectinload(AuthUserRole.role), ) .where(AuthUser.id == user_id, AuthUser.deleted_at.is_(None)) ) return result.scalar_one_or_none() async def get_user_stats(self) -> dict: """Get user statistics.""" base = AuthUser.deleted_at.is_(None) # Total users total_result = await self.db.execute( select(func.count()).select_from(AuthUser).where(base) ) total_users = total_result.scalar() or 0 # Active users active_result = await self.db.execute( select(func.count()).select_from(AuthUser).where(base, AuthUser.employment_status == "active") ) active_users = active_result.scalar() or 0 # Total department nodes dept_result = await self.db.execute( select(func.count()).select_from(IdentityOrgNode) .where( IdentityOrgNode.org_node_type == "department", IdentityOrgNode.deleted_at.is_(None), ) ) departments_count = dept_result.scalar() or 0 # Users with manager mgr_result = await self.db.execute( select(func.count()).select_from(AuthUser).where(base, AuthUser.manager_person_id.isnot(None)) ) with_manager = mgr_result.scalar() or 0 # By department breakdown (via org_memberships) dept_breakdown_result = await self.db.execute( select(IdentityOrgMembership.org_node_name, func.count().label("count")) .join(AuthUser, AuthUser.id == IdentityOrgMembership.person_id) .where( base, IdentityOrgMembership.is_primary.is_(True), IdentityOrgMembership.deleted_at.is_(None), ) .group_by(IdentityOrgMembership.org_node_name) .order_by(func.count().desc()) .limit(10) ) by_department = [ {"department": row.org_node_name, "count": row.count} for row in dept_breakdown_result.all() ] # By role breakdown role_breakdown_result = await self.db.execute( select(AuthRole.name, func.count(AuthUserRole.user_id).label("count")) .join(AuthUserRole, AuthRole.id == AuthUserRole.role_id) .join(AuthUser, AuthUser.id == AuthUserRole.user_id) .where(base) .group_by(AuthRole.name) .order_by(func.count(AuthUserRole.user_id).desc()) ) by_role = [ {"role": row.name, "count": row.count} for row in role_breakdown_result.all() ] # By status breakdown status_breakdown_result = await self.db.execute( select(AuthUser.employment_status, func.count().label("count")) .where(base) .group_by(AuthUser.employment_status) .order_by(func.count().desc()) ) by_status = [ {"status": row.employment_status, "count": row.count} for row in status_breakdown_result.all() ] return { "total_users": total_users, "active_users": active_users, "departments_count": departments_count, "with_manager": with_manager, "by_department": by_department, "by_role": by_role, "by_status": by_status, } async def get_org_tree(self) -> list[dict]: """Build department tree from org nodes, with users as leaf nodes.""" org_nodes_result = await self.db.execute( select(IdentityOrgNode) .where( IdentityOrgNode.org_node_type == "department", IdentityOrgNode.deleted_at.is_(None), ) .order_by( IdentityOrgNode.source_system, IdentityOrgNode.org_level, IdentityOrgNode.org_node_name, ) ) org_nodes = list(org_nodes_result.scalars().all()) people_result = await self.db.execute( select(AuthUser, IdentityOrgMembership) .outerjoin( IdentityOrgMembership, (IdentityOrgMembership.person_id == AuthUser.id) & (IdentityOrgMembership.is_primary.is_(True)) & (IdentityOrgMembership.deleted_at.is_(None)), ) .where( AuthUser.deleted_at.is_(None), # Mirror ``get_position_org_chart``: the org tree is a # live directory of who's here, not a forensic record. # Departed people (``employment_status='left'``) must not # appear — they were leaking in because this query, unlike # the position chart, had no status filter (fixed # 2026-06-17). AuthUser.employment_status.in_(["active", "leaver_pending"]), ) .order_by(AuthUser.display_name, AuthUser.work_email) ) people_rows = people_result.all() def node_key(source_system: str, org_node_type: str, source_id: str) -> str: return f"{source_system}:{org_node_type}:{source_id}" tree_nodes: dict[str, dict] = {} roots: list[dict] = [] for org_node in org_nodes: key = node_key( org_node.source_system, org_node.org_node_type, org_node.org_node_source_id, ) tree_nodes[key] = { "id": key, "node_type": "department", "name": org_node.org_node_name, "email": None, "department": org_node.org_path, "position": None, "status": org_node.node_status, "member_count": 0, "children": [], "_sort_name": org_node.org_node_name.lower(), } for org_node in org_nodes: key = node_key( org_node.source_system, org_node.org_node_type, org_node.org_node_source_id, ) parent_key = ( node_key(org_node.source_system, org_node.org_node_type, org_node.parent_org_node_source_id) if org_node.parent_org_node_source_id else None ) node = tree_nodes[key] if parent_key and parent_key in tree_nodes: tree_nodes[parent_key]["children"].append(node) else: roots.append(node) unassigned_people: list[dict] = [] for person, membership in people_rows: person_node = { "id": str(person.id), "node_type": "person", "name": person.display_name or person.work_email, "email": person.work_email, "department": membership.org_node_name if membership else None, "position": person.position, "status": person.employment_status, "member_count": None, "children": [], "_sort_name": (person.display_name or person.work_email).lower(), } if membership: key = node_key( membership.source_system, membership.org_node_type, membership.org_node_source_id, ) if key in tree_nodes: tree_nodes[key]["children"].append(person_node) continue unassigned_people.append(person_node) if unassigned_people: roots.append( { "id": "virtual:unassigned", "node_type": "department", "name": "Unassigned", "email": None, "department": None, "position": None, "status": "active", "member_count": 0, "children": unassigned_people, "_sort_name": "zzzz-unassigned", } ) def finalize(nodes: list[dict]) -> list[dict]: def count_people(node: dict) -> int: if node["node_type"] == "person": node.pop("_sort_name", None) return 1 node["children"] = sorted( node["children"], key=lambda item: ( 0 if item["node_type"] == "department" else 1, item["_sort_name"], ), ) total = 0 for child in node["children"]: total += count_people(child) node["member_count"] = total node.pop("_sort_name", None) return total ordered = sorted(nodes, key=lambda item: item["_sort_name"]) for node in ordered: count_people(node) return ordered return finalize(roots) async def get_position_org_chart(self) -> list[dict]: """Build department-based org chart from local DB data. Uses identity_org_nodes as tree structure and identity_persons (via org_memberships) as incumbents under each department node. """ # 1. Load all active department nodes org_nodes_result = await self.db.execute( select(IdentityOrgNode).where( IdentityOrgNode.org_node_type == "department", IdentityOrgNode.deleted_at.is_(None), ) ) org_nodes = list(org_nodes_result.scalars().all()) if not org_nodes: return [] # 2. Load all active people with their primary org membership people_result = await self.db.execute( select(AuthUser, IdentityOrgMembership) .outerjoin( IdentityOrgMembership, (IdentityOrgMembership.person_id == AuthUser.id) & (IdentityOrgMembership.is_primary.is_(True)) & (IdentityOrgMembership.deleted_at.is_(None)), ) .where( AuthUser.deleted_at.is_(None), AuthUser.employment_status.in_(["active", "leaver_pending"]), ) .order_by(AuthUser.display_name) ) people_rows = people_result.all() # 3. Group people by org_node_source_id people_by_dept: dict[str, list[dict]] = defaultdict(list) unassigned_people: list[dict] = [] for person, membership in people_rows: person_dict = { "user_id": person.hris_person_id or str(person.id), "helix_person_id": str(person.id), "display_name": person.display_name or person.work_email, "email": person.work_email, } if membership and membership.org_node_source_id: people_by_dept[membership.org_node_source_id].append(person_dict) else: unassigned_people.append(person_dict) # 4. Build tree nodes nodes: dict[str, dict] = {} for org_node in org_nodes: source_id = org_node.org_node_source_id incumbents = people_by_dept.get(source_id, []) nodes[source_id] = { "code": source_id, "name": org_node.org_node_name, "parent_code": org_node.parent_org_node_source_id, "department": org_node.org_path or org_node.org_node_name, "company": None, "location": None, "status": org_node.node_status or "active", "target_fte": float(len(incumbents)) if incumbents else None, "current_fte": len(incumbents), "direct_reports_count": 0, "incumbents": incumbents, "children": [], "_sort_name": org_node.org_node_name.lower(), } # 5. Link parent-child roots: list[dict] = [] for node in nodes.values(): parent_code = node["parent_code"] if parent_code and parent_code in nodes: nodes[parent_code]["children"].append(node) else: roots.append(node) # 6. Add unassigned bucket if needed if unassigned_people: roots.append({ "code": "unassigned", "name": "Unassigned", "parent_code": None, "department": None, "company": None, "location": None, "status": "active", "target_fte": None, "current_fte": len(unassigned_people), "direct_reports_count": 0, "incumbents": unassigned_people, "children": [], "_sort_name": "zzzz-unassigned", }) # 7. Finalize: sort and count def finalize(node: dict) -> dict: raw_children = sorted( node["children"], key=lambda child: child["_sort_name"], ) node["children"] = [finalize(child) for child in raw_children] node["direct_reports_count"] = len(node["children"]) node.pop("_sort_name", None) return node return [finalize(node) for node in sorted(roots, key=lambda item: item["_sort_name"])] async def get_departments(self) -> list[dict]: """Get departments with user counts, deduped by name. Same name can appear under multiple org_nodes (e.g. ``Accounting`` under different company subtrees). ``usage_department`` columns downstream store the bare name string, so returning duplicates is noise — group by name and sum member counts across all matching nodes. """ base = AuthUser.deleted_at.is_(None) result = await self.db.execute( select( IdentityOrgNode.org_node_name.label("department"), func.count(AuthUser.id).label("count"), ) .select_from(IdentityOrgNode) .outerjoin( IdentityOrgMembership, and_( IdentityOrgMembership.org_node_source_id == IdentityOrgNode.org_node_source_id, IdentityOrgMembership.org_node_type == IdentityOrgNode.org_node_type, IdentityOrgMembership.is_primary.is_(True), IdentityOrgMembership.deleted_at.is_(None), ), ) .outerjoin( AuthUser, and_( AuthUser.id == IdentityOrgMembership.person_id, base, ), ) .where( IdentityOrgNode.org_node_type == "department", IdentityOrgNode.deleted_at.is_(None), ) .group_by(IdentityOrgNode.org_node_name) .order_by(IdentityOrgNode.org_node_name.asc()) ) return [ {"department": row.department, "count": row.count} for row in result.all() ] async def get_department_tree(self) -> list[dict]: """Flat list of department org-nodes for the hierarchical picker. Returns one row per ``identity_org_nodes`` department node with its parent pointer + breadcrumb so the frontend can rebuild the tree (``DepartmentTreeSelect``). ``count`` is the node's *direct* primary members (not rolled up) — the picker shows it as a hint; subtree roll-up on selection happens server-side via :func:`app.core.org_scope.department_membership_subtree_clause`. Same role gate as :meth:`get_departments` (``require_directory_read``). """ base = AuthUser.deleted_at.is_(None) result = await self.db.execute( select( IdentityOrgNode.org_node_source_id.label("source_id"), IdentityOrgNode.org_node_name.label("name"), IdentityOrgNode.parent_org_node_source_id.label("parent_source_id"), IdentityOrgNode.org_path.label("org_path"), IdentityOrgNode.org_level.label("level"), func.count(AuthUser.id).label("count"), ) .select_from(IdentityOrgNode) .outerjoin( IdentityOrgMembership, and_( IdentityOrgMembership.org_node_source_id == IdentityOrgNode.org_node_source_id, IdentityOrgMembership.org_node_type == IdentityOrgNode.org_node_type, IdentityOrgMembership.is_primary.is_(True), IdentityOrgMembership.deleted_at.is_(None), ), ) .outerjoin( AuthUser, and_( AuthUser.id == IdentityOrgMembership.person_id, base, ), ) .where( IdentityOrgNode.org_node_type == "department", IdentityOrgNode.deleted_at.is_(None), ) .group_by( IdentityOrgNode.org_node_source_id, IdentityOrgNode.org_node_name, IdentityOrgNode.parent_org_node_source_id, IdentityOrgNode.org_path, IdentityOrgNode.org_level, ) .order_by(IdentityOrgNode.org_path.asc()) ) return [ { "source_id": row.source_id, "name": row.name, "parent_source_id": row.parent_source_id, "org_path": row.org_path, "level": row.level, "count": row.count, } for row in result.all() ]