import asyncio
import logging
from typing import Any, NoReturn
from uuid import UUID

from fastapi import Cookie, Depends, Request
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from jwt import InvalidTokenError
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload

from app.auth.models import AuthUser, AuthUserRole
from app.auth.service import is_effectively_departed
from app.core.config import settings
from app.core.database import get_db
from app.core.exceptions import ForbiddenError, UnauthorizedError
from app.core.rbac import (
    ASSET_ADMIN_ROLE,
    ASSET_GLOBAL_ADMIN_ROLE,
    AUDITOR_ROLE,
    CLOUD_ADMIN_ROLE,
    FINANCE_ROLE,
    GLOBAL_ADMIN_ROLE,
    HR_ROLE,
    MANAGEMENT_ROLE,
    SECURITY_OFFICER_ROLE,
    apply_view_role_narrow,
    canonicalize_role_name,
)
from app.core.redis_client import get_redis
from app.core.security import decode_token

_log = logging.getLogger(__name__)

bearer_scheme = HTTPBearer(auto_error=False)


def get_real_ip(request: Request) -> str:
    """Return the real client IP, honouring X-Real-IP / X-Forwarded-For set by nginx.

    Only trusts proxy headers when the direct TCP peer is in TRUSTED_PROXY_IPS;
    otherwise the peer address is returned as-is to prevent IP spoofing.

    Pre-launch P1 fix (Round 45, CR-105): when falling back to
    ``X-Forwarded-For``, take the RIGHTMOST element — that's the one
    the trusted nginx hop appended (real client IP), not the LEFTMOST
    which is whatever the client put in the header.  Without this,
    per-IP rate limiters (login, refresh, MCP, PAT) could be bypassed
    by sending ``X-Forwarded-For: <random>`` per attempt — nginx
    appends the real IP, but we were reading the spoofed leftmost
    element, giving the attacker an unbounded per-attempt key.
    """
    from app.core.net import is_trusted_proxy

    peer = request.client.host if request.client else "unknown"
    if not is_trusted_proxy(peer):
        return peer
    x_real_ip = request.headers.get("X-Real-IP")
    if x_real_ip:
        return x_real_ip.strip()
    x_forwarded_for = request.headers.get("X-Forwarded-For")
    if x_forwarded_for:
        # Rightmost element = the one nginx appended (= the real peer
        # IP it observed). Leftmost is the original client header, which
        # an attacker can fully control.
        parts = [p.strip() for p in x_forwarded_for.split(",") if p.strip()]
        if parts:
            return parts[-1]
    return peer


# Backward-compat alias for existing callers that imported the
# private name. New code should use ``get_real_ip``.
_get_real_ip = get_real_ip


async def check_password_lockout(email: str) -> None:
    """Pre-launch hardening 2026-05-04: per-account lockout layered
    on top of per-IP login rate limit.

    Per-IP only is bypassable: an IP-spoofed login farm can hammer
    one account from many addresses and never trip the per-IP cap.
    This counter is keyed by lowercased email so each account has
    its own bucket. Default 10 attempts / 15 min — tight enough to
    block credential stuffing, loose enough that a real user
    fat-fingering 4-5 times still gets in.

    Fail-open on Redis unavailability, mirroring
    ``check_login_rate_limit`` — Redis outage doesn't lock everyone
    out of the platform.

    Round 2b audit fix-up 2026-05-04: raises ``AppException`` (which
    flows through the standard envelope handler with ``errors[0]``)
    instead of bare ``HTTPException`` (which produces ``{detail: ...}``).
    The frontend's login page reads ``errors[0].message`` and was
    silently dropping the lockout message before this change.
    """
    from app.core.exceptions import AppException

    if not email:
        return
    key = f"password_lockout:{email.strip().lower()}"
    try:
        redis = get_redis()
        async with redis.pipeline(transaction=False) as pipe:
            pipe.incr(key)
            pipe.expire(key, settings.LOGIN_RATE_LIMIT_WINDOW_SECONDS, nx=True)
            results = await pipe.execute()
        count = int(results[0]) if results else 0
        # 10 per window per account.
        if count > 10:
            ttl = await redis.ttl(key)
            if ttl < 0:
                ttl = settings.LOGIN_RATE_LIMIT_WINDOW_SECONDS
            raise AppException(
                code="LOGIN_LOCKED",
                message=(
                    f"Too many failed sign-in attempts for this account. "
                    f"Try again in {ttl}s, or use Google sign-in."
                ),
                status_code=429,
                # Round 4 audit fix-up 2026-05-04: AppException now
                # carries optional headers; restore the Retry-After
                # hint that was on the prior HTTPException so
                # standards-aware clients (browsers, Axios retry
                # adapters, password managers) can back off correctly.
                headers={"Retry-After": str(ttl)},
            )
    except AppException:
        raise
    except Exception as exc:
        _log.warning(
            "password lockout Redis unavailable: %s",
            type(exc).__name__,
        )


async def reset_password_lockout(email: str) -> None:
    """Clear the per-account lockout counter on successful login."""
    if not email:
        return
    key = f"password_lockout:{email.strip().lower()}"
    try:
        redis = get_redis()
        await redis.delete(key)
    except Exception:
        pass


async def check_login_rate_limit(request: Request) -> None:
    """Fixed-window rate limiter for login endpoints, keyed by source IP.

    Uses a Redis pipeline (INCR + EXPIRE) executed atomically so the key
    can never be left without a TTL if the EXPIRE call were to fail mid-way.
    If Redis is unavailable the check is skipped (fail-open) so a Redis
    outage never blocks legitimate logins.

    Raises ``AppException(429)`` (envelope-shaped) when the caller exceeds
    LOGIN_RATE_LIMIT_MAX attempts within LOGIN_RATE_LIMIT_WINDOW_SECONDS.
    Bare ``HTTPException`` short-circuited the standard error envelope —
    the frontend reads ``errors[0].message`` and saw nothing.
    """
    from app.core.exceptions import AppException as _AppExc

    ip = get_real_ip(request)
    key = f"login_ratelimit:{ip}"
    try:
        redis = get_redis()
        # Pipeline both commands so the key always receives a TTL atomically.
        # EXPIRE is idempotent on existing keys; setting it every request
        # is a fixed-window (not sliding) but it bounds worst-case at 2×window.
        async with redis.pipeline(transaction=False) as pipe:
            pipe.incr(key)
            pipe.expire(key, settings.LOGIN_RATE_LIMIT_WINDOW_SECONDS, nx=True)
            results = await pipe.execute()
        count = int(results[0]) if results else 0
        if count > settings.LOGIN_RATE_LIMIT_MAX:
            ttl = await redis.ttl(key)
            if ttl < 0:
                ttl = settings.LOGIN_RATE_LIMIT_WINDOW_SECONDS
            raise _AppExc(
                code="LOGIN_RATE_LIMITED",
                message=f"Too many login attempts. Try again in {ttl}s.",
                status_code=429,
                headers={"Retry-After": str(ttl)},
            )
    except _AppExc:
        raise
    except Exception as exc:
        _log.warning("login rate limiter Redis unavailable, applying brute-force delay: %s", exc)
        await asyncio.sleep(0.5)


async def check_refresh_rate_limit(request: Request) -> None:
    """Per-IP rate limiter for ``/auth/refresh``, kept in a Redis bucket
    separate from :func:`check_login_rate_limit`.

    ``/auth/refresh`` and ``/auth/google/verify`` have very different
    legit-traffic shapes. Login traffic for a real user is bursty +
    bounded (a handful of attempts, then a long pause). Refresh traffic
    is steady + can be dense — every page reload that hits the dashboard
    fans out ~5 widgets that 401 in parallel, an idle multi-tab browser
    refreshes silently, and a corporate NAT can fan in many distinct
    users behind one IP. Sharing the bucket means a refresh storm burns
    the ``google/verify`` budget and the user gets a misleading 429 on
    re-login. Use a wider, dedicated bucket here so a refresh burst
    never costs a user their ability to sign back in. 2026-05-19 prod
    incident: a user reloaded enough times to spend 20+ refresh
    attempts in <15min, then could not log in for the rest of the
    window.

    Fail-open on Redis unavailability, mirroring the login limiter.
    """
    from app.core.exceptions import AppException as _AppExc

    ip = get_real_ip(request)
    key = f"refresh_ratelimit:{ip}"
    try:
        redis = get_redis()
        async with redis.pipeline(transaction=False) as pipe:
            pipe.incr(key)
            pipe.expire(key, settings.REFRESH_RATE_LIMIT_WINDOW_SECONDS, nx=True)
            results = await pipe.execute()
        count = int(results[0]) if results else 0
        if count > settings.REFRESH_RATE_LIMIT_MAX:
            ttl = await redis.ttl(key)
            if ttl < 0:
                ttl = settings.REFRESH_RATE_LIMIT_WINDOW_SECONDS
            raise _AppExc(
                code="REFRESH_RATE_LIMITED",
                message=f"Too many token-refresh attempts. Try again in {ttl}s.",
                status_code=429,
                headers={"Retry-After": str(ttl)},
            )
    except _AppExc:
        raise
    except Exception as exc:
        _log.warning("refresh rate limiter Redis unavailable: %s", exc)
        await asyncio.sleep(0.2)


# A single shared egress IP fans in the ENTIRE org's inbound webhook traffic
# (Feishu approval callbacks for ~500 employees + Feishu's aggressive delivery
# retries), so a webhook limiter must be sized for burst+retry volume, NOT the
# bursty-but-bounded per-user login shape. The login ceiling (20/900s) would
# trip almost immediately and feed a 429→retry→429 loop. This ceiling only
# exists to bound a malformed-push flood that slips past the 1MB body cap;
# real callbacks are HMAC/token-verified + replay-guarded downstream.
_WEBHOOK_RATE_LIMIT_MAX = 1200
_WEBHOOK_RATE_LIMIT_WINDOW_SECONDS = 60


async def check_webhook_rate_limit(request: Request) -> None:
    """Per-IP rate limiter for inbound third-party webhook endpoints
    (e.g. the Feishu approval callback), kept in a Redis bucket separate
    from the login/refresh limiters and sized for org-wide burst+retry
    traffic from a shared provider egress IP. Fail-open on Redis
    unavailability, mirroring the login limiter.
    """
    from app.core.exceptions import AppException as _AppExc

    ip = get_real_ip(request)
    key = f"webhook_ratelimit:{ip}"
    try:
        redis = get_redis()
        async with redis.pipeline(transaction=False) as pipe:
            pipe.incr(key)
            pipe.expire(key, _WEBHOOK_RATE_LIMIT_WINDOW_SECONDS, nx=True)
            results = await pipe.execute()
        count = int(results[0]) if results else 0
        if count > _WEBHOOK_RATE_LIMIT_MAX:
            ttl = await redis.ttl(key)
            if ttl < 0:
                ttl = _WEBHOOK_RATE_LIMIT_WINDOW_SECONDS
            raise _AppExc(
                code="WEBHOOK_RATE_LIMITED",
                message=f"Too many webhook requests. Try again in {ttl}s.",
                status_code=429,
                headers={"Retry-After": str(ttl)},
            )
    except _AppExc:
        raise
    except Exception as exc:
        _log.warning("webhook rate limiter Redis unavailable: %s", exc)


async def _load_user_with_roles(db: AsyncSession, user_uuid: UUID) -> AuthUser | None:
    """Load an AuthUser by UUID with the same eager-loading shape that
    every authenticated request uses. Shared by the JWT path and the
    personal-access-token path.
    """
    result = await db.execute(
        select(AuthUser)
        .options(
            selectinload(AuthUser.user_roles).selectinload(AuthUserRole.role),
            selectinload(AuthUser.subordinates),
        )
        .where(AuthUser.id == user_uuid, AuthUser.deleted_at.is_(None))
    )
    return result.scalar_one_or_none()


async def get_current_user(
    request: Request,
    credentials: HTTPAuthorizationCredentials | None = Depends(bearer_scheme),
    db: AsyncSession = Depends(get_db),
) -> AuthUser:
    """Resolve the request's bearer credential to an :class:`AuthUser`.

    Two credential kinds are accepted, in order of precedence:

    1. **Personal access token** — the credential starts with
       ``helix_pat_``. The hash is looked up in
       ``auth_personal_access_tokens``; on success ``last_used_*``
       fields on the token row are bumped, and ``request.state.auth_via``
       is set to ``"pat"`` plus ``request.state.pat_token_id``. Audit
       logs that include this metadata can therefore distinguish a
       UI-initiated action from one performed by an MCP client.
    2. **JWT access token** — the existing Google-SSO / password-login
       path. ``request.state.auth_via`` is set to ``"jwt"``.

    Both paths converge on :func:`_load_user_with_roles`, so downstream
    RBAC sees an identical user object regardless of credential type.
    Departed / deleted users are rejected uniformly.
    """
    # Per-request memoization. Router-level gates — notably
    # ``require_active_employee_for_mutations`` attached on the asset /
    # cmdb / apikey / … routers — resolve the bearer via *this* function
    # imperatively BEFORE the path operation's own
    # ``Depends(get_current_user)`` runs. Without memoization every
    # mutating request authenticates the bearer twice: two
    # ``_load_user_with_roles`` round-trips and, for a PAT, two
    # ``last_used`` bookkeeping writes from independent sessions. Those
    # two writes contend on the same token row — under the test harness
    # (one shared request session held open for the whole request) the
    # second write lock-waits the full ``innodb_lock_wait_timeout`` (50s)
    # before failing open, which is what made the PAT-gate cmdb-lifecycle
    # cases each take ~50s in CI. ``auth_via`` is set only after the user
    # is fully resolved in both branches, so its presence is a safe
    # "already authenticated this request" sentinel; reuse the cached
    # user (auth_via / view-role header / contextvars were set by the
    # first call). One request carries one bearer, so reuse is correct.
    cached_user = getattr(request.state, "current_user", None)
    if cached_user is not None and getattr(request.state, "auth_via", None):
        return cached_user

    # Lazy import — pulling tokens.service at module scope creates an
    # import cycle with app.core.database via Base.
    from app.auth.tokens.service import PAT_PREFIX, PersonalAccessTokenService

    if not credentials:
        raise UnauthorizedError()

    raw = credentials.credentials

    # ── PAT path ──
    if raw.startswith(PAT_PREFIX):
        from app.auth.auth_context import current_auth_method, current_pat_token_id

        service = PersonalAccessTokenService(db)
        pat_row = await service.authenticate(
            raw,
            client_ip=get_real_ip(request),
            user_agent=request.headers.get("User-Agent"),
        )
        if pat_row is None:
            raise UnauthorizedError("Invalid or expired token")
        user = await _load_user_with_roles(db, pat_row.user_id)
        if not user or is_effectively_departed(user):
            raise UnauthorizedError("User not found or deactivated")
        # Pre-launch P2 fix (Round 1, SEC-003): a PAT created before
        # ``force_logout_at`` is rejected. An admin offboarding a person
        # who holds an MCP PAT therefore severs MCP access at the same
        # moment as their browser session. Use the PAT row's
        # ``created_at`` as the credential's "issued at" timestamp.
        _enforce_force_logout(user, pat_row.created_at)
        # Mark PAT auth BEFORE applying the view-role header — the header
        # helper reads ``request.state.auth_via`` to short-circuit PATs
        # (PAT clients have no UI role switcher; the header on a PAT
        # request is suspicious and must not narrow). Setting auth_via
        # afterward would let the short-circuit silently miss and the
        # PAT's role set would be narrowed against the documented
        # contract — caught in second-pass review.
        request.state.current_user = user
        request.state.auth_via = "pat"
        _apply_view_role_header(request, user)
        request.state.pat_token_id = pat_row.id
        request.state.pat_token_prefix = pat_row.token_prefix
        current_auth_method.set("pat")
        current_pat_token_id.set(pat_row.id)
        # Pre-launch P1 fix: stamp the actor contextvar so out-of-
        # guard ORM writes (e.g. an admin PATCH that flips employment
        # status outside LeaverBlockGuard) can be attributed in the
        # before_flush listener's trap audit row.
        from app.auth.auth_context import current_actor_user_id  # noqa: PLC0415
        current_actor_user_id.set(user.id)
        return user

    # ── JWT path ──
    try:
        payload = decode_token(raw)
    except InvalidTokenError:
        raise UnauthorizedError("Invalid or expired token")

    if payload.get("type") != "access":
        raise UnauthorizedError("Invalid token type")

    user_id = payload.get("sub")
    if not user_id:
        raise UnauthorizedError()
    try:
        user_uuid = UUID(str(user_id))
    except (TypeError, ValueError):
        raise UnauthorizedError("Invalid token subject")

    user = await _load_user_with_roles(db, user_uuid)
    if not user or is_effectively_departed(user):
        raise UnauthorizedError("User not found or deactivated")

    # Pre-launch P2 fix (Round 1, SEC-003): when offboarding marks the
    # leaver and pushes Google ``users.signOut``, ``force_logout_at`` is
    # set on the person. A pre-existing JWT issued before that moment
    # carries an ``iat`` claim that pre-dates the force-logout — refuse
    # such tokens so the leaver's existing browser tabs lose access
    # immediately rather than coast to natural JWT TTL expiry. Bounded
    # but real read-access window otherwise. PAT path enforces the
    # same rule below.
    _enforce_force_logout(user, payload.get("iat"))

    from app.auth.auth_context import (
        current_actor_user_id,
        current_auth_method,
    )
    request.state.current_user = user
    # Set auth_via BEFORE narrowing — the view-role helper reads it to
    # short-circuit PATs. Symmetric to the PAT branch above so future
    # refactors don't reintroduce the order-of-operations bug caught
    # in second-pass review.
    request.state.auth_via = "jwt"
    _apply_view_role_header(request, user)
    current_auth_method.set("jwt")
    current_actor_user_id.set(user.id)
    return user


def _enforce_force_logout(user: AuthUser, token_iat: Any) -> None:
    """If ``user.force_logout_at`` is set, refuse credentials issued
    before that timestamp.

    JWTs carry ``iat`` (seconds since epoch). PAT rows carry
    ``last_used_at`` (DateTime) — for those we use the row's
    ``created_at`` as the lower bound (a PAT minted before
    ``force_logout_at`` is honored only as long as the PAT itself is
    still active; rotating force-logout therefore demands rotating
    the PAT).
    """
    # Round 3 P2 fix (R3-002): direct attribute access. Memory rule
    # ``feedback_no_getattr_default_for_model_columns.md`` — the
    # ``getattr(..., None)`` pattern silently masks rename/typo
    # bugs, which is unsafe on the security-critical hot path that
    # gates EVERY authenticated request. ``force_logout_at`` is a
    # mapped ORM column on ``IdentityPerson`` (the alias for
    # ``AuthUser``); a refactor that drops the column should fail
    # loudly, not silently let every leaver back in.
    fl = user.force_logout_at
    if fl is None:
        return
    if token_iat is None:
        return
    # ``iat`` may be an int (JWT seconds), a datetime (PAT created_at).
    if isinstance(token_iat, (int, float)):
        from datetime import UTC as _UTC  # noqa: PLC0415
        from datetime import datetime as _dt  # noqa: PLC0415
        token_dt = _dt.fromtimestamp(int(token_iat), tz=_UTC)
    else:
        token_dt = token_iat
    # Tolerate naive datetimes (TiDB DATETIME without TZ); treat as UTC.
    if getattr(token_dt, "tzinfo", None) is None:
        from datetime import UTC as _UTC  # noqa: PLC0415
        token_dt = token_dt.replace(tzinfo=_UTC)
    fl_aware = fl
    if getattr(fl_aware, "tzinfo", None) is None:
        from datetime import UTC as _UTC  # noqa: PLC0415
        fl_aware = fl_aware.replace(tzinfo=_UTC)
    if token_dt < fl_aware:
        raise UnauthorizedError("Session was terminated")


async def enforce_action_token_subject(
    db: AsyncSession, payload: dict[str, Any],
) -> AuthUser:
    """Reject a signed action-token whose subject has since departed or been
    force-logged-out, and return the loaded subject on success.

    Action tokens (``core.security.create_action_token``) are 14-day signed
    one-click email/Feishu deep-links carrying only ``sub``/``res``/``purpose``.
    ``decode_action_token`` validates signature/exp/type/purpose but nothing
    binds the token to the subject's CURRENT lifecycle state — unlike every
    other credential surface (JWT/PAT/MCP-OAuth all re-resolve the person and
    reject departed / force-logged-out users per request). Without this an
    offboarded employee's un-clicked action link keeps authorizing its action
    for up to 14 days, contradicting the "force_logout severs every credential
    surface" invariant.

    Raises ``ValueError`` (NOT ``UnauthorizedError``) so the no-login HTML
    consumers' existing ``except (jwt.InvalidTokenError, ValueError, KeyError)``
    renders the friendly "link invalid or expired" page (privacy-preserving —
    no "you have departed" disclosure), and the JSON consumers' broad
    ``except Exception -> ForbiddenError`` maps correctly too.

    NOTE: ``create_action_token`` already mints ``iat`` — no minting change is
    needed for the force-logout comparison.
    """
    sub = UUID(str(payload["sub"]))
    person = await db.get(AuthUser, sub)
    if person is None or person.deleted_at is not None:
        raise ValueError("action token subject not found")
    if is_effectively_departed(person):
        raise ValueError("action token subject has departed")
    try:
        _enforce_force_logout(person, payload.get("iat"))
    except UnauthorizedError as exc:
        raise ValueError("action token subject session terminated") from exc
    return person


# Header values longer than this are dropped before they reach the
# RBAC validator. The longest legitimate role name is ``security_auditor``
# at 16 chars; a 64-char cap leaves headroom for future roles while
# refusing pathological header bodies (CRLF injection / log poisoning).
_VIEW_ROLE_HEADER_MAX = 64


def _apply_view_role_header(request: Request, user: AuthUser) -> None:
    """Apply the request's ``X-View-Role`` header to the user instance.

    Defense-in-depth: validates the header before applying it. The
    validator (``apply_view_role_narrow``) silently no-ops on invalid
    or escalating values; this wrapper additionally caps header length
    and skips PATs entirely (PAT-driven automation has no UI role
    switcher, so a PAT-bearing client setting this header is suspicious
    — refuse the narrowing rather than honour it).
    """
    auth_via = getattr(request.state, "auth_via", None)
    if auth_via == "pat":
        # PAT clients are non-interactive — there is no role switcher in
        # the MCP / API surface. A PAT sending this header is either a
        # confused dev or a hostile client; either way, do not narrow.
        # The PAT inherits its issuing user's roles unchanged.
        return
    raw = request.headers.get("X-View-Role")
    if not raw:
        return
    if len(raw) > _VIEW_ROLE_HEADER_MAX:
        return
    apply_view_role_narrow(user, raw)


async def apply_sensitive_rate_limit(user_id) -> None:
    """Per-user rate limiter for sensitive data actions (PII reveal,
    key reveal, export). Shared by HTTP and MCP entry points so a PAT
    holder can't bypass the cap by switching transports.

    Default: 10 calls per hour per user. Fails open on Redis errors but
    adds a small delay to resist abuse. Raises ``AppException(429)``
    on cap breach.
    """
    from app.core.exceptions import AppException

    key = f"sensitive_ratelimit:{user_id}"
    max_calls = settings.SENSITIVE_RATE_LIMIT_MAX
    window = settings.SENSITIVE_RATE_LIMIT_WINDOW_SECONDS
    try:
        redis = get_redis()
        async with redis.pipeline(transaction=False) as pipe:
            pipe.incr(key)
            pipe.expire(key, window, nx=True)
            results = await pipe.execute()
        count = int(results[0]) if results else 0
        if count > max_calls:
            ttl = await redis.ttl(key)
            if ttl < 0:
                ttl = window
            raise AppException(
                code="RATE_LIMITED",
                message=(
                    f"Too many requests to this endpoint. "
                    f"Try again in {ttl}s."
                ),
                status_code=429,
                # Round 8 audit fix-up 2026-05-04: surface Retry-After
                # so client retry adapters can back off correctly.
                headers={"Retry-After": str(ttl)},
            )
    except AppException:
        raise
    except Exception as exc:
        _log.warning("sensitive rate limiter Redis error (fail-open with delay): %s", exc)
        await asyncio.sleep(0.5)


async def check_sensitive_endpoint_rate_limit(
    request: Request,
    current_user: AuthUser = Depends(get_current_user),
) -> None:
    """FastAPI dependency wrapper around :func:`apply_sensitive_rate_limit`.

    NOTE 2026-05-02: this function MUST stay below ``get_current_user``
    in the file — it takes ``Depends(get_current_user)`` as a default
    value, which Python evaluates at function-definition time. The
    earlier placement (commit 814d680) raised ``NameError`` on import.
    """
    from fastapi import HTTPException

    from app.core.exceptions import AppException

    try:
        await apply_sensitive_rate_limit(current_user.id)
    except AppException as exc:
        if exc.status_code == 429:
            raise HTTPException(
                status_code=429,
                detail=exc.message,
                headers={
                    "Retry-After": str(
                        settings.SENSITIVE_RATE_LIMIT_WINDOW_SECONDS,
                    ),
                },
            ) from exc
        raise


async def check_copilot_rate_limit(
    request: Request,
    current_user: AuthUser = Depends(get_current_user),
) -> None:
    """Per-user rate limiter for the AI copilot chat endpoint.

    Each turn fans out to several LLM round-trips against the org's
    DashScope/Bailian budget; without a cap an authenticated employee
    could loop the endpoint and drive unbounded spend. Uses a dedicated,
    more generous budget than the sensitive limiter (10/h would break a
    real interactive chat). Fails open on Redis errors with a small delay.
    """
    from app.core.exceptions import AppException

    key = f"copilot_ratelimit:{current_user.id}"
    max_calls = settings.COPILOT_RATE_LIMIT_MAX
    window = settings.COPILOT_RATE_LIMIT_WINDOW_SECONDS
    try:
        redis = get_redis()
        async with redis.pipeline(transaction=False) as pipe:
            pipe.incr(key)
            pipe.expire(key, window, nx=True)
            results = await pipe.execute()
        count = int(results[0]) if results else 0
        if count > max_calls:
            ttl = await redis.ttl(key)
            if ttl < 0:
                ttl = window
            raise AppException(
                code="RATE_LIMITED",
                message=(
                    f"AI 助手调用过于频繁，请 {ttl}s 后再试。"
                ),
                status_code=429,
                headers={"Retry-After": str(ttl)},
            )
    except AppException:
        raise
    except Exception as exc:
        _log.warning(
            "copilot rate limiter Redis error (fail-open with delay): %s", exc,
        )
        await asyncio.sleep(0.5)


async def check_security_faq_ask_rate_limit(
    current_user: AuthUser = Depends(get_current_user),
) -> None:
    """Per-user rate limiter for the 安全 FAQ ask endpoint.

    A query that keyword search cannot place is handed to the platform
    LLM, so an employee holding down the ask button spends the org's
    DashScope budget. Its own key and budget rather than the copilot's:
    somebody who has been chatting with the assistant all afternoon must
    still be able to look up a security answer.

    Fails open on Redis errors with a small delay, matching the other
    per-user limiters in this module.
    """
    from app.core.exceptions import AppException

    key = f"security_faq_ask_ratelimit:{current_user.id}"
    max_calls = settings.SECURITY_FAQ_ASK_RATE_LIMIT_MAX
    window = settings.SECURITY_FAQ_ASK_RATE_LIMIT_WINDOW_SECONDS
    try:
        redis = get_redis()
        async with redis.pipeline(transaction=False) as pipe:
            pipe.incr(key)
            pipe.expire(key, window, nx=True)
            results = await pipe.execute()
        count = int(results[0]) if results else 0
        if count > max_calls:
            ttl = await redis.ttl(key)
            if ttl < 0:
                ttl = window
            raise AppException(
                code="RATE_LIMITED",
                message=(
                    f"提问过于频繁，请 {ttl}s 后再试。期间仍可直接搜索安全 FAQ。"
                ),
                status_code=429,
                headers={"Retry-After": str(ttl)},
            )
    except AppException:
        raise
    except Exception as exc:
        _log.warning(
            "security FAQ ask rate limiter Redis error "
            "(fail-open with delay): %s", exc,
        )
        await asyncio.sleep(0.5)


async def _ai_labs_limit(
    user_id, *, bucket: str, max_calls: int, window: int, what: str,
) -> None:
    """Shared body for the two AI Labs limiters.

    One function rather than two copies because the only differences are
    the Redis key, the budget and the noun in the message — and a copy is
    how the copilot's and the FAQ's limiters drifted apart on everything
    except the part that matters.

    Fails open with a small delay on Redis errors, matching every other
    per-user limiter in this module: a Redis outage must not take a
    working feature offline, and the delay keeps a hot loop from becoming
    unbounded while it is down.
    """
    from app.core.exceptions import AppException

    key = f"ai_labs_{bucket}_ratelimit:{user_id}"
    try:
        redis = get_redis()
        async with redis.pipeline(transaction=False) as pipe:
            pipe.incr(key)
            pipe.expire(key, window, nx=True)
            results = await pipe.execute()
        count = int(results[0]) if results else 0
        if count > max_calls:
            ttl = await redis.ttl(key)
            if ttl < 0:
                ttl = window
            raise AppException(
                code="RATE_LIMITED",
                message=(
                    f"AI Labs {what} requests are coming in too fast. "
                    f"Please try again in {ttl}s."
                ),
                status_code=429,
                headers={"Retry-After": str(ttl)},
            )
    except AppException:
        raise
    except Exception as exc:
        _log.warning(
            "ai_labs %s rate limiter Redis error (fail-open with delay): %s",
            bucket, exc,
        )
        await asyncio.sleep(0.5)


async def check_ai_labs_rate_limit(
    current_user: AuthUser = Depends(get_current_user),
) -> None:
    """Per-user limiter for AI Labs chat turns and translations."""
    await _ai_labs_limit(
        current_user.id,
        bucket="turn",
        max_calls=settings.AI_LABS_RATE_LIMIT_MAX,
        window=settings.AI_LABS_RATE_LIMIT_WINDOW_SECONDS,
        what="chat",
    )


async def check_ai_labs_generation_rate_limit(
    current_user: AuthUser = Depends(get_current_user),
) -> None:
    """Per-user limiter for AI Labs media generation jobs.

    Tighter than the chat limiter: one video costs the vendor minutes of
    GPU time and writes tens of megabytes to S3, so the shape of abuse
    here is much more expensive than a fast typist.
    """
    await _ai_labs_limit(
        current_user.id,
        bucket="generation",
        max_calls=settings.AI_LABS_GENERATION_RATE_LIMIT_MAX,
        window=settings.AI_LABS_GENERATION_RATE_LIMIT_WINDOW_SECONDS,
        what="generation",
    )


async def apply_directory_read_rate_limit(user_id) -> None:
    """Per-user rate limiter for directory reads (people picker, etc.).

    Wider cap than :func:`apply_sensitive_rate_limit` — 600 calls/hour
    — since these endpoints serve typical interactive UI (employee
    opens multiple Tab switches across SoftwareCatalogPage /
    RelationshipGraphPage / EmailGroupDirectoryPage / picker drawers
    in one session) but still need protection against script-based
    enumeration of the org-chart `(id, display_name)` pairs. The
    600/h cap blocks scripted enumeration (a typical org of ~5000
    people would need to span 8+ hours to harvest the full set under
    cap) while never affecting normal interactive flow.
    Fails open with a small delay on Redis error.
    """
    from app.core.exceptions import AppException

    key = f"directory_ratelimit:{user_id}"
    max_calls = 600
    window = 3600
    try:
        redis = get_redis()
        async with redis.pipeline(transaction=False) as pipe:
            pipe.incr(key)
            pipe.expire(key, window, nx=True)
            results = await pipe.execute()
        count = int(results[0]) if results else 0
        if count > max_calls:
            ttl = await redis.ttl(key)
            if ttl < 0:
                ttl = window
            raise AppException(
                code="RATE_LIMITED",
                message=(
                    f"Too many directory reads. Try again in {ttl}s."
                ),
                status_code=429,
                headers={"Retry-After": str(ttl)},
            )
    except AppException:
        raise
    except Exception as exc:
        _log.warning(
            "directory rate limiter Redis error (fail-open): %s", exc,
        )
        await asyncio.sleep(0.1)


async def apply_approval_submit_rate_limit(user_id) -> None:
    """Per-user rate limiter for approval-submit endpoints (DNS request,
    domain registration). Default: 30 submits per hour per user.

    Round-30 security audit: an authenticated employee could otherwise
    POST 10k DNS-request rows in seconds, flooding the Feishu approver's
    inbox + bloating ``cmdb_dns_record_requests``. Cap blocks scripted
    spam while leaving plenty of headroom for genuine interactive use
    (a typical employee submits <5/day). Fails open with a delay on
    Redis error (better to slow than to drop legitimate submits).
    """
    from app.core.exceptions import AppException

    key = f"approval_submit_ratelimit:{user_id}"
    max_calls = 30
    window = 3600
    try:
        redis = get_redis()
        async with redis.pipeline(transaction=False) as pipe:
            pipe.incr(key)
            pipe.expire(key, window, nx=True)
            results = await pipe.execute()
        count = int(results[0]) if results else 0
        if count > max_calls:
            ttl = await redis.ttl(key)
            if ttl < 0:
                ttl = window
            raise AppException(
                code="RATE_LIMITED",
                message=(
                    f"Too many approval submissions. "
                    f"Try again in {ttl}s."
                ),
                status_code=429,
                headers={"Retry-After": str(ttl)},
            )
    except AppException:
        raise
    except Exception as exc:
        _log.warning(
            "approval-submit rate limiter Redis error "
            "(fail-open with delay): %s",
            exc,
        )
        await asyncio.sleep(0.3)


async def check_approval_submit_rate_limit(
    request: Request,
    current_user: AuthUser = Depends(get_current_user),
) -> None:
    """FastAPI dep wrapper around :func:`apply_approval_submit_rate_limit`.

    Use on POST submit endpoints that fan out to Feishu approval cards
    (DNS request, domain registration). Caps each user to 30 submits/h.
    """
    from fastapi import HTTPException

    from app.core.exceptions import AppException

    try:
        await apply_approval_submit_rate_limit(current_user.id)
    except AppException as exc:
        if exc.status_code == 429:
            raise HTTPException(
                status_code=429,
                detail=exc.message,
                headers={"Retry-After": str(3600)},
            ) from exc
        raise


async def check_directory_read_rate_limit(
    request: Request,
    current_user: AuthUser = Depends(get_current_user),
) -> None:
    """FastAPI dep wrapper around :func:`apply_directory_read_rate_limit`.

    Use on read-only directory endpoints that serve `(id, display_name)`
    style listings — wider cap than ``check_sensitive_endpoint_rate_limit``
    so legitimate multi-tab interactive use doesn't 429.
    """
    from fastapi import HTTPException

    from app.core.exceptions import AppException

    try:
        await apply_directory_read_rate_limit(current_user.id)
    except AppException as exc:
        if exc.status_code == 429:
            raise HTTPException(
                status_code=429,
                detail=exc.message,
                headers={"Retry-After": str(3600)},
            ) from exc
        raise


def require_browser_session(request: Request) -> None:
    """Reject PAT-authenticated requests on sensitive admin endpoints.

    PATs intentionally inherit the issuing user's roles (Web UI parity is
    the whole point — Claude Desktop should be able to do what its owner
    does). But for a small set of catastrophic-if-stolen actions we
    layer step-up auth on top: the call must come from a real Google-SSO
    browser session, never from a PAT.

    What we gate (defense-in-depth, not exploit-prevention):

    * **PII reveal** of *another* user's mobile / personal email —
      a stolen IT-Admin PAT could otherwise dump the entire roster.
    * **Role assignment / removal** — a stolen IT-Admin PAT could
      grant itself permanent admin via role-add (its own revocation
      is also blocked, so the attacker's hold is durable).
    * **API-key plaintext reveal** — admin-only secret view.
    * **PAT mint / revoke** — already gated; preserved here.

    Default fail-closed: when ``request.state.auth_via`` is unset
    (e.g. an internal refactor reordered middleware, an exception
    in ``get_current_user`` left the request progressing without
    auth tagging, or a future endpoint built its own auth path) we
    REJECT the request — only an explicit ``"jwt"`` value set by
    the JWT auth dependency permits sensitive actions.

    Pre-launch hardening 2026-05-06: previously this used
    ``getattr(..., "jwt")`` as the default which made the gate
    *fail-open* — exactly the opposite of the docstring's claim.
    Any caller that bypassed ``get_current_user`` (and therefore
    didn't set ``auth_via``) silently sailed through this gate.

    Use as a FastAPI dependency: ``Depends(require_browser_session)``.
    Or call positionally inside a handler that already has ``request``
    in scope.
    """
    auth_via = getattr(request.state, "auth_via", None)
    if auth_via != "jwt":
        raise ForbiddenError(
            "This action requires a Google SSO browser session. "
            "Personal access tokens cannot perform sensitive admin "
            "actions (PII reveal, role mutation, key reveal).",
        )


def require_roles(*allowed_roles: str):
    """Dependency that checks the current user has at least one of the allowed roles.

    Uses IdentityPerson.role_names property (which reads from eagerly-loaded
    user_roles relationship) so it works regardless of how the relationship
    was loaded.
    """

    async def checker(user: AuthUser = Depends(get_current_user)) -> AuthUser:
        normalized_allowed_roles = {canonicalize_role_name(role) for role in allowed_roles}
        if not set(user.role_names).intersection(normalized_allowed_roles):
            raise ForbiddenError()
        return user

    return checker


def require_admin():
    return require_roles(GLOBAL_ADMIN_ROLE)


def require_admin_read():
    """READ-only gate that mirrors ``require_admin`` but admits
    ``auditor`` and ``management``.

    Use on GET endpoints whose previous gate was ``require_admin()``
    and whose payload is governance-relevant (offboarding processes,
    provisioning task lists, scheduled-job catalog, notification
    outbox, integration metadata, SSO grant lists, identity-sync
    target catalog, …). The corresponding mutation endpoints stay on
    ``require_admin()`` so neither auditor nor management can change
    state — only inspect.

    2026-05-20: introduced as part of the management-role rollout to
    close 403 holes on the leadership read surface without granting
    write authority.
    """
    return require_roles(
        GLOBAL_ADMIN_ROLE, AUDITOR_ROLE, MANAGEMENT_ROLE,
    )


def require_person_filter_lookup():
    """READ gate for the person-population *lookup* endpoints — the
    department tree, the country list, the legal-entity list and the
    person search that back ``PersonFilterEditor``.

    Same audience as :func:`require_admin_read` **plus**
    ``security_officer``. The security desk composes awareness campaigns
    against the identical ``app.core.person_filters`` shape, in the
    identical shared editor; without this the pickers come back empty
    for them — and ``PersonFilterEditor`` swallows lookup failures
    (``.catch(() => {})``), so it would fail *silently* into a form that
    looks functional and selects nobody.

    Widening ``require_admin_read`` itself would have been wrong: it also
    gates offboarding processes, the provisioning queue, the notification
    outbox and integration metadata, none of which the security desk
    needs. This is the narrow gate for the four lookups, whose payload
    (department names, ISO country codes, entity codes, employee name +
    work email) is strictly less sensitive than the vulnerability
    reports and access certifications ``can_view_security`` already
    admits this role to.
    """
    return require_roles(
        GLOBAL_ADMIN_ROLE, AUDITOR_ROLE, MANAGEMENT_ROLE,
        SECURITY_OFFICER_ROLE,
    )


def require_audit_access():
    """Audit / governance READ access.

    Includes ``management`` (read-only leadership view) alongside
    ``global_admin`` and ``auditor``. Auditor mutations (decide review
    items, ack findings) still gate separately via
    ``can_manage_security`` — management cannot mutate audit state.
    """
    return require_roles(
        GLOBAL_ADMIN_ROLE, AUDITOR_ROLE, MANAGEMENT_ROLE,
    )


def require_asset_admin():
    return require_roles(GLOBAL_ADMIN_ROLE, ASSET_ADMIN_ROLE)


def require_global_asset_admin():
    """Gate for the asset-domain configuration that defines authority.

    Narrower than :func:`require_asset_admin` on purpose. The managed
    location master is where asset authority is assigned — a location's
    administrator list *is* the ``asset_admin`` grant set, and the
    location rows themselves are the region vocabulary those grants are
    expressed in. Leaving that surface open to every regional
    ``asset_admin`` would be a self-escalation path: a Beijing admin
    could add themselves to Tokyo, or create a location and grant it to
    anyone. So it takes the global tier — ``global_admin`` or
    ``asset_global_admin`` — which is exactly the set
    :func:`app.core.rbac.has_global_asset_scope` already treats as
    having authority everywhere in the domain.

    Regional admins keep full read access; they simply cannot change
    who is an administrator or what the regions are.
    """
    return require_roles(GLOBAL_ADMIN_ROLE, ASSET_GLOBAL_ADMIN_ROLE)


async def assert_can_manage_asset(
    db: AsyncSession, current_user: AuthUser, asset: Any,
) -> None:
    """Enforce region-scoped write authority on a specific asset.

    Used by every asset write endpoint AFTER the asset has been
    loaded. The route-level ``require_asset_admin`` dep has already
    confirmed the caller holds an admin role; this layer narrows
    it further so an ``asset_admin`` can only mutate assets at
    locations they administer (per ``asset_location_admins``).

    ``global_admin`` always passes (full authority — IT admins
    handle cross-region transfers, recovery from orphan assets,
    etc.). ``asset_admin`` passes iff the asset's resolved location
    (see ``resolve_asset_location_ids_bulk``) is in their managed set.
    Anyone else (Auditor / Management holders who slipped past the
    route dep) is rejected.

    Raises ``ForbiddenError`` when the asset is out of region.
    """
    from app.domains.asset.location_admin_sync import person_can_manage_asset

    role_names = [r.role.name for r in (current_user.user_roles or [])]
    permitted = await person_can_manage_asset(
        db, current_user.id, asset, role_names=role_names,
    )
    if not permitted:
        raise ForbiddenError(
            "asset_admin only authorized to manage assets within own region",
        )


def require_hr_admin():
    return require_roles(GLOBAL_ADMIN_ROLE, HR_ROLE)


# ── Read-only gates that admit ``management`` (and auditor where
# governance-relevant). Mirror the ``can_view_*`` helpers in rbac.py.
# Use these on GET endpoints whose previous gate was a write-side
# ``require_*_admin``; the write-side gates remain untouched so
# management never gains mutate authority.
#
# 2026-05-20: introduced when the management role rollout discovered
# that the org-tree / departments / asset workbench / cloud overview
# read surfaces all 403'd management because their dependencies
# pointed at write gates (``require_hr_admin`` etc.). The split keeps
# the write surface narrow while the read surface tracks
# ``can_view_*``.


def require_hr_read():
    """READ-only gate on HR governance data (org tree, departments,
    position chart, BU member counts, lifecycle digest).

    Mirrors ``can_view_hr_data``: global_admin + hr + auditor +
    management. Mutations still go through ``require_hr_admin``.
    """
    return require_roles(
        GLOBAL_ADMIN_ROLE, HR_ROLE, AUDITOR_ROLE, MANAGEMENT_ROLE,
    )


def require_lifecycle_read():
    """READ-only gate on the onboarding / offboarding / provisioning
    admin surface (process lists + detail, cockpit queues, policies,
    analytics, provisioning task lists).

    Same role set as :func:`require_hr_read` (global_admin + hr +
    auditor + management) — the join/move/leave lifecycle is HR-adjacent
    operational data, so HR reads it alongside global_admin, and
    auditor / management keep the governance read view they already had
    under ``require_admin_read``. Named distinctly from
    ``require_hr_read`` so the intent is legible at the onboarding /
    offboarding call sites, which are NOT in the HR domain module.

    2026-07-08: introduced to open the 入离职管理 pages to HR (read-only).
    Every mutation on these routers stays on ``require_admin`` so HR can
    inspect the lifecycle but never advance / cancel / revoke.
    """
    return require_roles(
        GLOBAL_ADMIN_ROLE, HR_ROLE, AUDITOR_ROLE, MANAGEMENT_ROLE,
    )


def require_directory_read():
    """READ-only gate on the lightweight directory (people picker +
    department dropdown).

    Mirrors :func:`app.core.rbac.can_view_directory` — admits every
    admin-tier role (global_admin, asset_admin, hr, finance,
    cloud_admin, auditor, management). Anchors the picker endpoints
    that every admin view uses:

      * ``GET /api/v1/directory/people`` — minimal-field person search
      * ``GET /api/v1/auth/departments`` — department-name list

    The full-fat HR roster (``GET /api/v1/users``, with manager
    hierarchy + last_login_at + hris_record_state + roles[]) stays on
    :func:`require_hr_read`. This split was introduced PR 2026-06-01
    after regional asset admins reported that the assignee /
    department filters on the asset management pages were silently
    empty — backend was returning 403 on /users + /auth/departments
    and the frontend try/catch swallowed it.
    """
    return require_roles(
        GLOBAL_ADMIN_ROLE,
        HR_ROLE,
        ASSET_ADMIN_ROLE,
        FINANCE_ROLE,
        CLOUD_ADMIN_ROLE,
        AUDITOR_ROLE,
        MANAGEMENT_ROLE,
    )


def require_asset_read():
    """READ-only gate on the asset domain admin surface (workbench,
    inventory list, history, KPI cards, exports).

    Mirrors ``can_read_all_assets``: global_admin + asset_admin +
    auditor + management. Mutations (claim/return/transfer/dispose)
    stay on ``require_asset_admin``.
    """
    return require_roles(
        GLOBAL_ADMIN_ROLE, ASSET_ADMIN_ROLE,
        AUDITOR_ROLE, MANAGEMENT_ROLE,
    )


def require_foundation_read():
    """READ-only gate on foundation org-config (office addresses,
    legal entities, supplier catalogues, business payment registers).

    Mirrors ``require_foundation_admin`` (global_admin + asset_admin +
    hr) plus auditor + management for governance visibility. Mutations
    stay on ``require_foundation_admin``.
    """
    return require_roles(
        GLOBAL_ADMIN_ROLE, ASSET_ADMIN_ROLE, HR_ROLE,
        AUDITOR_ROLE, MANAGEMENT_ROLE,
    )


def require_software_read():
    """READ-only gate on the software / procurement / supplier /
    business-payment / billing surface.

    Mirrors the /help permission matrix row 3 (软件 / 许可 / 采购 /
    供应商): global_admin + asset_admin + finance + auditor +
    management. Hardware-only asset reads (stats / history /
    locations / brands / categories) stay on the narrower
    ``require_asset_read`` because finance has no business with
    laptop intake KPIs per the same matrix (row 2: 财务=—).
    """
    return require_roles(
        GLOBAL_ADMIN_ROLE, ASSET_ADMIN_ROLE, FINANCE_ROLE,
        AUDITOR_ROLE, MANAGEMENT_ROLE,
    )


def require_config_read():
    """READ-only gate on the platform configuration surface
    (integrations catalogue, platform settings, security-tips
    admin content).

    Mirrors the /help permission matrix row 9 (配置中心): admin +
    management only. Auditor is intentionally EXCLUDED — the
    matrix's audit lane is `/admin/audit` + the security-auditor
    tooling, not the integration/settings panel. Secrets reveal
    stays separately gated on ``require_admin + require_browser_session``.
    """
    return require_roles(GLOBAL_ADMIN_ROLE, MANAGEMENT_ROLE)


def require_foundation_admin():
    """Gate for org-config CRUD: office addresses + legal entities.

    HR is included by explicit product decision (2026-05-14): HR owns
    the onboarding / offboarding pipelines that reference both
    ``asset_office_addresses`` (employee's working office) and
    ``asset_legal_entities`` (contracting entity). Without write access
    HR has to ping IT for routine org-config tweaks that arrive via
    SuccessFactors HRIS sync. Asset_admin keeps access because office
    addresses anchor asset locations and legal entities anchor
    depreciation policy.

    Mutations still write ``audit_logs`` rows naming the HR actor so a
    future compliance review can trace every legal-entity change back
    to a specific HR user.
    """
    return require_roles(GLOBAL_ADMIN_ROLE, ASSET_ADMIN_ROLE, HR_ROLE)


_LEAVER_LOCKED_MESSAGE = (
    "Your account is in offboarding state. Mutating actions are "
    "disabled while IT completes the handover. Read access "
    "remains available."
)
_LEAVER_LOCKED_CODE = "LEAVER_RUNTIME_LOCKED"


def raise_leaver_runtime_locked() -> NoReturn:
    """Raise the canonical Leaver Blocked 403.

    Use this from domain services that need to enforce the invariant at
    a finer grain than the router-level dependency can — e.g. a single
    endpoint that must admit *some* payloads from a leaver_pending user
    and reject others (see
    ``ApprovalService.create_asset_approval_request``, which admits
    asset_return / offboarding_buyback and rejects the rest). Sharing
    the code + message keeps every surface's rejection identical, so a
    frontend that maps ``LEAVER_RUNTIME_LOCKED`` to a friendly banner
    only has to map it once.
    """
    from app.core.exceptions import AppException
    raise AppException(
        code=_LEAVER_LOCKED_CODE,
        message=_LEAVER_LOCKED_MESSAGE,
        status_code=403,
    )


# Endpoints that must remain mutating-but-allowed even while the user
# is in leaver_pending. These are paths a departing user must still
# access to wind down: refreshing their JWT, logging out, persisting a
# language preference (so the "your account is in offboarding"
# banner shows in their preferred locale).
_LEAVER_MUTATE_ALLOWLIST: tuple[str, ...] = (
    "/api/v1/auth/refresh",
    "/api/v1/auth/logout",
    "/api/v1/auth/me/profile",
    # Round-51 regression-review R51-C: ``/api/v1/notifications`` got a
    # router-level mutate gate to harden against leaver-admin retry
    # abuse, but mark-read is a strictly self-scoped read-state flip
    # (``WHERE recipient_person_id = me``). Blocking it for a
    # leaver_pending user means their unread badge stays stuck while
    # they legitimately read offboarding / handover Feishu cards.
    # Allowlist this single mutate so leavers can still ack their own
    # inbox without unlocking the dangerous mutate endpoints
    # (outbox/cancel + outbox/retry stay gated).
    "/api/v1/notifications/inbox/mark-read",
)


# EXACT-path allowlist (no prefix expansion). Separate tuple because
# these paths have admin-only *sub*-paths that must stay gated —
# ``/approval/asset/requests/on-behalf`` is role-gated on
# ``can_manage_assets``, which a leaver_pending asset_admin would still
# satisfy, so prefix-matching the parent would hand a departing admin
# the requester-rewrite path. Match the path and nothing under it.
#
# Why these two are exempt (2026-08-13): 归还 (asset_return) and 回购
# (offboarding_buyback) are the two self-service flows a departing
# employee MUST complete — returning the laptop and buying it out are
# offboarding obligations, not new privileges. Before this carve-out
# the router gate 403'd them at submit time, so the only way to close
# an offboarding was IT re-typing the form through
# ``/asset/requests/on-behalf``.
#
# The *real* control is per-template and lives in
# ``ApprovalService.create_asset_approval_request`` — passing this
# allowlist only gets a leaver past the router; the service still
# rejects every other template_key (laptop_replacement,
# computer_borrow, software_*) with the same LEAVER_RUNTIME_LOCKED.
# That placement is deliberate: the MCP twin
# (``submit_approval_request``) shares the service, so both surfaces
# answer the same state identically.
#
# ``/asset/uploads`` carries no template_key, so it cannot be gated
# per-flow. It is allowlisted wholesale because it is inert — it
# stages bytes to Feishu (MIME + size validated) and returns a token;
# nothing in Helix changes and the token is worthless without a
# submit, which is gated. The 恢复出厂设置证明 upload is a required
# widget on the return form, so leaving it blocked would make the
# exemption above useless.
_LEAVER_MUTATE_EXACT_ALLOWLIST: tuple[str, ...] = (
    "/api/v1/approval/asset/requests",
    "/api/v1/approval/asset/uploads",
)


async def require_active_employee(
    user: AuthUser = Depends(get_current_user),
) -> AuthUser:
    """Reject non-active employees from mutating endpoints.

    Pair this with a per-endpoint dependency on every POST / PATCH /
    DELETE that represents a self-service action that the user
    initiates while inside Helix: 申请, 归还, 转让, 上传 etc. Read
    endpoints intentionally do NOT use this gate — leaver_pending
    users still need to see what they own while they hand things over.

    For sweeping coverage at the router level, prefer
    ``require_active_employee_for_mutations`` which checks
    ``request.method`` so a single ``dependencies=[...]`` on the
    router covers GETs (allowed) and POST/PATCH/PUT/DELETE
    (gated) without needing a per-endpoint flag.

    The check itself is in ``app.core.rbac.is_active_employee``
    (employment_status == 'active' && deleted_at IS NULL). When
    rejected we return a 403 with a specific error code so the
    frontend can show "your account is in offboarding — read-only"
    rather than the generic permission denied banner.
    """
    from app.core.rbac import is_active_employee  # local import — keeps the import surface small

    if not is_active_employee(user):
        from app.core.exceptions import AppException
        raise AppException(
            code=_LEAVER_LOCKED_CODE,
            message=_LEAVER_LOCKED_MESSAGE,
            status_code=403,
        )
    return user


async def require_active_employee_for_mutations(
    request: Request,
    credentials: HTTPAuthorizationCredentials | None = Depends(bearer_scheme),
    db: AsyncSession = Depends(get_db),
) -> AuthUser | None:
    """Router-level dependency: gates mutate methods of authenticated
    users only.

    Attach via ``APIRouter(dependencies=[Depends(...)])`` on routers
    that contain a mix of read + write endpoints. Behaviour:

      * GET / OPTIONS / HEAD — pass through.
      * POST / PATCH / PUT / DELETE on a path in the allowlist
        (``/auth/refresh`` etc.) — pass through.
      * Mutate methods, no bearer credential at all — pass through.
        This matters for public webhook endpoints (Feishu approval
        callback signs its own requests; it does not carry our JWT).
        The endpoint's own auth dependencies (or absence thereof)
        still decide whether the request is authorised; we only
        gate active-employee status when there IS a user.
      * Mutate methods with a bearer credential — resolve the user
        and reject if they are not in employment_status='active'.

    The check itself uses the same auth resolution path as
    :func:`get_current_user` (JWT or PAT) so JWT short-TTL on
    leaver_pending users + this gate compose correctly.
    """
    from app.core.rbac import is_active_employee

    method = request.method.upper()
    if method in ("GET", "OPTIONS", "HEAD"):
        return None

    path = request.url.path
    for allow in _LEAVER_MUTATE_ALLOWLIST:
        if path == allow or path.startswith(allow + "/"):
            return None
    if path in _LEAVER_MUTATE_EXACT_ALLOWLIST:
        return None

    # No credentials — let the endpoint's own auth deps decide.
    # Public webhooks (Feishu callback, etc.) reach here.
    if credentials is None:
        return None

    # Resolve the user via the same path get_current_user uses, but
    # we tolerate failures (an invalid token must propagate as 401
    # via the endpoint's own get_current_user dep — not via this gate).
    try:
        user = await get_current_user(request, credentials=credentials, db=db)
    except Exception:
        return None

    if not is_active_employee(user):
        from app.core.exceptions import AppException
        raise AppException(
            code=_LEAVER_LOCKED_CODE,
            message=_LEAVER_LOCKED_MESSAGE,
            status_code=403,
        )
    return user


async def get_refresh_token_payload(
    refresh_token: str | None = Cookie(default=None),
) -> dict:
    if not refresh_token:
        raise UnauthorizedError("No refresh token")
    try:
        payload = decode_token(refresh_token)
    except InvalidTokenError:
        raise UnauthorizedError("Invalid refresh token")
    if payload.get("type") != "refresh":
        raise UnauthorizedError("Invalid token type")
    return payload