from uuid import UUID from fastapi import APIRouter, Depends, HTTPException, Query, Request, Response from sqlalchemy.ext.asyncio import AsyncSession from app.audit.service import AuditService from app.auth.feishu_sso import ( build_authorize_url as build_feishu_authorize_url, ) from app.auth.feishu_sso import ( exchange_code_for_user_info as exchange_feishu_code, ) from app.auth.feishu_sso import ( exchange_h5_code_for_user_info as exchange_feishu_h5_code, ) from app.auth.feishu_sso import ( validate_redirect_uri as validate_feishu_redirect_uri, ) from app.auth.feishu_sso import ( verify_state as verify_feishu_state, ) from app.auth.google_sso import verify_google_id_token from app.auth.models import AuthUser from app.auth.schemas import ( DirectoryPersonResponse, FeishuAuthorizeUrlRequest, FeishuAuthorizeUrlResponse, FeishuH5LoginRequest, FeishuVerifyRequest, GoogleVerifyRequest, ManagerBrief, MeResponse, MyProfilePatchRequest, PasswordLoginRequest, RevealContactResponse, RoleAssignRequest, RoleBatchRequest, RoleResponse, SubordinateBrief, UserDetailResponse, UserListItemResponse, UserUpdateRequest, ) from app.auth.service import _LOGIN_FAILED_OPAQUE, AuthService from app.core.config import settings from app.core.database import get_db from app.core.dependencies import ( check_directory_read_rate_limit, check_login_rate_limit, check_password_lockout, check_refresh_rate_limit, check_sensitive_endpoint_rate_limit, get_current_user, get_real_ip, get_refresh_token_payload, require_admin, require_browser_session, require_directory_read, require_hr_admin, require_hr_read, reset_password_lockout, ) from app.core.exceptions import ( AppException, UnauthorizedError, is_transaction_aborting_db_error, ) from app.core.idempotent_route import idempotent from app.core.masking import mask_phone from app.core.response import paginated_response, success_response from app.core.safe_logging import get_safe_logger from app.core.txn import best_effort from app.identity.employee_types import employee_type_slug logger = get_safe_logger(__name__) _COOKIE_SECURE = settings.APP_ENV != "development" def _should_secure_cookie(request: Request) -> bool: """Decide whether to set Secure on a refresh-token cookie. Pre-launch hardening 2026-05-04: previously hard-tied to APP_ENV, so a staging deploy over HTTP leaked the refresh token. Now: True when the actual incoming request is HTTPS, so a misconfigured staging-over-HTTP refuses to plant a Secure cookie (browser drops it) instead of silently sending one over plaintext. Production is always HTTPS and ``_COOKIE_SECURE`` stays True there. """ if request.url.scheme == "https": return True return _COOKIE_SECURE router = APIRouter(prefix="/api/v1", tags=["auth"]) def _mask_phone(raw: str | None) -> str | None: """Masked mobile number — delegates to the shared ``app.core.masking`` helper so the country-code split can't drift from every other surface. A previous hand-rolled copy here greedily grabbed up to 3 digits as the country code, mangling every ``+CC`` number with a 1-2 digit code (``+86138…`` → ``+861 38…``). The shared helper splits on a real CC boundary. Falsy input returns ``None`` to preserve this endpoint's existing contract. """ if not raw: return None return mask_phone(raw) def _mask_email(raw: str | None) -> str | None: """Mask local part of an email, preserving the domain. ``alice.example@gmail.com`` → ``a***e@gmail.com`` ``a@x.io`` → ``*@x.io`` """ if not raw or "@" not in raw: return raw local, _, domain = raw.partition("@") if len(local) <= 2: return f"{'*' * len(local)}@{domain}" return f"{local[0]}{'*' * (len(local) - 2)}{local[-1]}@{domain}" def _user_to_response( user: AuthUser, is_manager: bool = False, is_attendance_subject: bool = False, ) -> dict: # ``actual_role_names`` (not ``role_names``) — the frontend's # ``viewRole`` switcher menu enumerates the user's real role set; # if we returned the narrowed view-role-applied set here, a # global_admin who switched to "员工 view" would lose the ability # to switch back because /auth/me would now report ``roles = # ['employee']``. See ``apply_view_role_narrow`` in app.core.rbac. return { "id": user.id, "email": user.work_email, "display_name": user.display_name, "avatar_url": user.avatar_url, "department": user.department, "position": user.position, "status": user.employment_status, "roles": user.actual_role_names, "is_manager": is_manager, "is_attendance_subject": is_attendance_subject, "last_login_at": user.last_login_at, "created_at": user.created_at, } @router.post("/auth/google/verify") async def google_verify( body: GoogleVerifyRequest, request: Request, response: Response, db: AsyncSession = Depends(get_db), _rate: None = Depends(check_login_rate_limit), ): # Failed-attempt audit (cycle 2 audit fix): without this row, the # SOC dashboard sees only successful logins. Per-account lockout # only fires after 10 attempts, so the first 9 brute-force tries # were entirely invisible. Write a ``status="failure"`` row # BEFORE raising — independent session so the request rollback # doesn't take it out. async def _audit_login_failure(reason: str, action: str = "auth.login") -> None: from app.core.database import async_session_factory # noqa: PLC0415 try: async with async_session_factory() as audit_db: await AuditService(audit_db).log( trace_id=request.state.trace_id, actor_user_id=None, actor_type="anonymous", action=action, resource_type="auth_user", resource_id="-", status="failure", ip_address=get_real_ip(request), user_agent=request.headers.get("user-agent"), detail={"reason": reason[:500]}, ) await audit_db.commit() except Exception: # noqa: BLE001 logger.exception("failed to record auth-failure audit row") try: google_payload = await verify_google_id_token(body.id_token) except ValueError as e: await _audit_login_failure(f"google verify: {e}") raise AppException(code="GOOGLE_AUTH_FAILED", message=str(e), status_code=401) service = AuthService(db) try: user, access_token, refresh_token = await service.login_via_google(google_payload) except ValueError as e: # Detailed reason → audit only. Client-facing message is the # opaque ``LOGIN_FAILED`` constant: distinct strings would let # an unauthenticated probe enumerate account state (which # emails exist, which bindings are suspended / disabled / # rotation-blocked). SOC/SIEM still sees the detail via the # audit row's ``detail.reason``. await _audit_login_failure(f"login via google: {e}") raise AppException( code="LOGIN_FAILED", message=_LOGIN_FAILED_OPAQUE, status_code=401, ) is_manager = await service.has_subordinates(user.id) is_attendance_subject = await service.is_attendance_subject(user.id) # Set refresh_token as httpOnly cookie response.set_cookie( key="refresh_token", value=refresh_token, httponly=True, secure=_should_secure_cookie(request), samesite="strict", max_age=7 * 24 * 3600, path="/api/v1/auth", ) # Audit log audit = AuditService(db) await audit.log( trace_id=request.state.trace_id, actor_user_id=user.id, actor_type="user", action="auth.login", resource_type="auth_user", resource_id=str(user.id), status="success", ip_address=get_real_ip(request), user_agent=request.headers.get("user-agent"), ) return success_response( data={ "access_token": access_token, "token_type": "bearer", "user": _user_to_response(user, is_manager, is_attendance_subject), }, request=request, ) @router.post("/auth/feishu/authorize-url") async def feishu_authorize_url( body: FeishuAuthorizeUrlRequest, request: Request, db: AsyncSession = Depends(get_db), # Rate-limit the URL-issuing endpoint same as login. A bot # spraying state tokens would otherwise fill Redis with junk # entries (each consumes ~64 bytes for 10 minutes). _rate: None = Depends(check_login_rate_limit), ): """Issue a Feishu OAuth authorize URL with a one-time-use state. Defence-in-depth: ``redirect_uri`` must match one of ``CORS_ORIGINS`` + ``/auth/feishu/callback``. Feishu's own redirect whitelist is the authoritative gate, but we refuse early so a misconfigured Feishu whitelist can't turn Helix into a redirect oracle. """ try: url = await build_feishu_authorize_url(db, body.redirect_uri) except ValueError as e: raise AppException( code="FEISHU_AUTHORIZE_URL_FAILED", message=str(e), status_code=400, ) return success_response( data=FeishuAuthorizeUrlResponse(authorize_url=url).model_dump(), request=request, ) @router.get("/auth/feishu/app-id") async def feishu_app_id( request: Request, db: AsyncSession = Depends(get_db), ): """Expose the configured Feishu app_id so the H5 JS SDK can call ``tt.requestAccess({appID})``. Not a new disclosure — the same app_id is embedded in the URL returned by ``/auth/feishu/authorize-url``. Returning it from a dedicated endpoint avoids the H5 page having to issue a no-op authorize-url POST just to scrape the app_id. 503 ``FEISHU_NOT_CONFIGURED`` when no Feishu integration row exists yet — same as the SSO endpoints. """ from app.integrations.compat import get_feishu_config # noqa: PLC0415 cfg = await get_feishu_config(session=db) app_id = cfg.get("app_id") if not app_id: raise AppException( code="FEISHU_NOT_CONFIGURED", message=( "Feishu integration is not configured. Configure the " "Feishu app in Admin > Integrations before using " "Feishu H5 login." ), status_code=503, ) return success_response(data={"app_id": app_id}, request=request) @router.post("/auth/feishu/verify") async def feishu_verify( body: FeishuVerifyRequest, request: Request, response: Response, db: AsyncSession = Depends(get_db), _rate: None = Depends(check_login_rate_limit), ): """Exchange a Feishu OAuth ``code`` for a Helix JWT. Mirrors :func:`google_verify` line-for-line on auditing, cookie settings, and 401-on-failure behaviour. Differences are confined to the upstream call: state validation + code-for-token exchange instead of id_token verification. """ async def _audit_login_failure(reason: str) -> None: from app.core.database import async_session_factory # noqa: PLC0415 try: async with async_session_factory() as audit_db: await AuditService(audit_db).log( trace_id=request.state.trace_id, actor_user_id=None, actor_type="anonymous", action="auth.feishu_login", resource_type="auth_user", resource_id="-", status="failure", ip_address=get_real_ip(request), user_agent=request.headers.get("user-agent"), detail={"reason": reason[:500]}, ) await audit_db.commit() except Exception: # noqa: BLE001 logger.exception( "failed to record feishu-auth-failure audit row", ) # Reject bad redirect_uri BEFORE we even peek at the state — a # caller scanning for valid state values shouldn't get a # different error message from a redirect-uri probe. try: validate_feishu_redirect_uri(body.redirect_uri) except ValueError as e: await _audit_login_failure(f"feishu redirect_uri: {e}") raise AppException( code="FEISHU_AUTH_FAILED", message=str(e), status_code=400, ) try: await verify_feishu_state(body.state, body.redirect_uri) except ValueError as e: await _audit_login_failure(f"feishu state: {e}") raise AppException( code="FEISHU_AUTH_FAILED", message=str(e), status_code=401, ) try: feishu_user_info = await exchange_feishu_code( db, body.code, body.redirect_uri, ) except ValueError as e: await _audit_login_failure(f"feishu code exchange: {e}") raise AppException( code="FEISHU_AUTH_FAILED", message=str(e), status_code=401, ) service = AuthService(db) try: user, access_token, refresh_token = await service.login_via_feishu( feishu_user_info, ) except ValueError as e: # Audit gets the detailed reason; client sees the opaque # message so a probe cannot enumerate account state. await _audit_login_failure(f"login via feishu: {e}") raise AppException( code="LOGIN_FAILED", message=_LOGIN_FAILED_OPAQUE, status_code=401, ) is_manager = await service.has_subordinates(user.id) is_attendance_subject = await service.is_attendance_subject(user.id) response.set_cookie( key="refresh_token", value=refresh_token, httponly=True, secure=_should_secure_cookie(request), samesite="strict", max_age=7 * 24 * 3600, path="/api/v1/auth", ) audit = AuditService(db) await audit.log( trace_id=request.state.trace_id, actor_user_id=user.id, actor_type="user", action="auth.feishu_login", resource_type="auth_user", resource_id=str(user.id), status="success", ip_address=get_real_ip(request), user_agent=request.headers.get("user-agent"), ) return success_response( data={ "access_token": access_token, "token_type": "bearer", "user": _user_to_response(user, is_manager, is_attendance_subject), }, request=request, ) @router.post("/auth/feishu/h5-login") async def feishu_h5_login( body: FeishuH5LoginRequest, request: Request, response: Response, db: AsyncSession = Depends(get_db), _rate: None = Depends(check_login_rate_limit), ): """Exchange a Feishu H5-SDK ``tt.requestAccess`` code for a Helix JWT. **DEPRECATED — believed to have no caller (2026-07-28).** The mobile shell dropped the H5-SDK path on 2026-05-22 (the SDK step failed on the first attempt often enough, and its native black toast could not be suppressed, so boot goes straight to the OAuth redirect flow). A repo sweep finds no frontend call site; the only remaining reference is an axios interceptor prefix list. It is NOT deleted yet on purpose: this is an authentication endpoint, and "no caller in this repo" is weaker evidence than "no caller in the world" — a stale Feishu app configuration or a pinned client would lose login with no signal. Instead every hit now logs a WARNING with the caller's IP, so a deploy cycle of silence is the evidence that makes removal safe. If this line never appears, delete the endpoint, its schema, its `feishu_sso.login_via_feishu_h5` service function, and the `/auth/feishu/h5-login` entry in `web/src/api/client.ts`'s `AUTH_NO_REFRESH_PREFIXES`. Distinct from :func:`feishu_verify` (OAuth redirect flow): * No ``state`` token — there is no authorize step to bind one to. * No ``redirect_uri`` validation — the SDK does not perform a browser redirect. Both omissions are safe because ``tt.requestAccess`` only yields a code when the page is loaded inside the Feishu app's own webview, holding the user's authenticated session, and the code is bound to ``(app_id, user_id)`` on the Feishu side. A page in a normal browser cannot synthesise such a code. Downstream (``login_via_feishu``, JWT/cookie, audit fan-out) is identical to the redirect-flow endpoint — only the upstream call differs. """ logger.warning( "DEPRECATED endpoint /auth/feishu/h5-login was called from ip=%s " "ua=%s — it is believed to have no caller; investigate before the " "planned removal.", get_real_ip(request), request.headers.get("User-Agent", "-"), ) async def _audit_login_failure(reason: str) -> None: from app.core.database import async_session_factory # noqa: PLC0415 try: async with async_session_factory() as audit_db: await AuditService(audit_db).log( trace_id=request.state.trace_id, actor_user_id=None, actor_type="anonymous", action="auth.feishu_h5_login", resource_type="auth_user", resource_id="-", status="failure", ip_address=get_real_ip(request), user_agent=request.headers.get("user-agent"), detail={"reason": reason[:500]}, ) await audit_db.commit() except Exception: # noqa: BLE001 logger.exception( "failed to record feishu-h5-auth-failure audit row", ) try: feishu_user_info = await exchange_feishu_h5_code(db, body.code) except ValueError as e: await _audit_login_failure(f"feishu h5 code exchange: {e}") raise AppException( code="FEISHU_AUTH_FAILED", message=str(e), status_code=401, ) service = AuthService(db) try: user, access_token, refresh_token = await service.login_via_feishu( feishu_user_info, ) except ValueError as e: # Audit gets the detailed reason; client sees opaque message. await _audit_login_failure(f"login via feishu h5: {e}") raise AppException( code="LOGIN_FAILED", message=_LOGIN_FAILED_OPAQUE, status_code=401, ) is_manager = await service.has_subordinates(user.id) is_attendance_subject = await service.is_attendance_subject(user.id) response.set_cookie( key="refresh_token", value=refresh_token, httponly=True, secure=_should_secure_cookie(request), samesite="strict", max_age=7 * 24 * 3600, path="/api/v1/auth", ) audit = AuditService(db) await audit.log( trace_id=request.state.trace_id, actor_user_id=user.id, actor_type="user", action="auth.feishu_h5_login", resource_type="auth_user", resource_id=str(user.id), status="success", ip_address=get_real_ip(request), user_agent=request.headers.get("user-agent"), ) return success_response( data={ "access_token": access_token, "token_type": "bearer", "user": _user_to_response(user, is_manager, is_attendance_subject), }, request=request, ) @router.post("/auth/login", include_in_schema=settings.ALLOW_PASSWORD_LOGIN) async def password_login( body: PasswordLoginRequest, request: Request, response: Response, db: AsyncSession = Depends(get_db), _rate: None = Depends(check_login_rate_limit), ): if not settings.ALLOW_PASSWORD_LOGIN: raise HTTPException(status_code=404, detail="Not found") # Pre-launch hardening 2026-05-04: per-account lockout layered on # the per-IP rate limit so a credential-stuffing farm hitting one # account from many IPs still gets stopped. await check_password_lockout(body.email) service = AuthService(db) try: user, access_token, refresh_token = await service.login_via_password(body.email, body.password) except ValueError as e: # Failed-attempt audit (cycle 2 fix). Independent session so # the request rollback doesn't take it out. from app.core.database import async_session_factory # noqa: PLC0415 try: async with async_session_factory() as audit_db: await AuditService(audit_db).log( trace_id=request.state.trace_id, actor_user_id=None, actor_type="anonymous", action="auth.login_password", resource_type="auth_user", resource_id="-", status="failure", ip_address=get_real_ip(request), user_agent=request.headers.get("user-agent"), # ``email_attempted`` is masked by truncating to # local-part length — full email could be PII for # a typo-attack against a real account. detail={ "reason": str(e)[:500], "email_attempted": _mask_email(body.email), }, ) await audit_db.commit() except Exception: # noqa: BLE001 logger.exception("failed to record password-login failure audit") raise AppException(code="LOGIN_FAILED", message=str(e), status_code=401) # Successful login → reset the per-account counter so the user # doesn't carry a near-lockout state into next session. await reset_password_lockout(body.email) is_manager = await service.has_subordinates(user.id) is_attendance_subject = await service.is_attendance_subject(user.id) response.set_cookie( key="refresh_token", value=refresh_token, httponly=True, # Pre-launch hardening 2026-05-04: tie ``secure`` to the actual # request scheme so a misconfigured staging deploy over HTTP # doesn't leak the refresh token. Production runs HTTPS only; # dev runs HTTP localhost where Secure-cookies are dropped. secure=_should_secure_cookie(request), samesite="strict", max_age=7 * 24 * 3600, path="/api/v1/auth", ) audit = AuditService(db) await audit.log( trace_id=request.state.trace_id, actor_user_id=user.id, actor_type="user", action="auth.login_password", resource_type="auth_user", resource_id=str(user.id), status="success", ip_address=get_real_ip(request), user_agent=request.headers.get("user-agent"), ) return success_response( data={ "access_token": access_token, "token_type": "bearer", "user": _user_to_response(user, is_manager, is_attendance_subject), }, request=request, ) @router.post("/auth/refresh") async def refresh_token( request: Request, response: Response, payload: dict = Depends(get_refresh_token_payload), db: AsyncSession = Depends(get_db), # Refresh tokens are long-lived (7 days) and a leaked cookie could # otherwise be exchanged for fresh access tokens at unlimited rate, # so keep a per-IP cap — but in its OWN bucket (wider window), # not the login bucket. Sharing with ``/auth/google/verify`` meant a # refresh-traffic burst (page reloads, multi-tab fan-out, corporate # NAT) would lock the same IP out of login for 15 min — see # 2026-05-19 prod incident in ``check_refresh_rate_limit``. _rate: None = Depends(check_refresh_rate_limit), ): service = AuthService(db) try: user, access_token, new_refresh_token = await service.refresh_tokens(payload) except ValueError as e: # Round-49: write an audit row on every refresh failure # (deactivated user / leaver / force_logout / reuse-detected). # The previous code raised UnauthorizedError with no trail; a # leaver's laptop continuing to poll ``/auth/refresh`` after # offboarding was invisible to the SOC. Independent session # so it survives the caller's rollback. import uuid as _uuid from app.audit.service import log_denied_action try: actor_uuid = ( UUID(str(payload.get("sub"))) if payload.get("sub") else None ) except (ValueError, TypeError): actor_uuid = None await log_denied_action( trace_id=getattr(request.state, "trace_id", None) or str(_uuid.uuid4()), actor_type="user", actor_user_id=actor_uuid, action="auth.refresh_failure", resource_type="auth_user", resource_id=str(payload.get("sub") or ""), detail={ "reason": str(e), "jti": payload.get("jti"), }, ip_address=get_real_ip(request), user_agent=request.headers.get("User-Agent"), ) raise UnauthorizedError(str(e)) response.set_cookie( key="refresh_token", value=new_refresh_token, httponly=True, secure=_should_secure_cookie(request), samesite="strict", max_age=7 * 24 * 3600, path="/api/v1/auth", ) return success_response( data={"access_token": access_token, "token_type": "bearer"}, request=request, ) @router.post("/auth/logout") async def logout( request: Request, response: Response, user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): # Best-effort: try to read and invalidate the refresh token JTI from cookie. # Guard: only invalidate the JTI when the token's sub matches the calling # user — prevents a user from invalidating a different user's refresh token # by crafting a cookie with another user's JTI. # SAVEPOINT, not a bare ``except``: ``AuthService.logout`` invalidates the # refresh JTI through this session, and the very next statement is an # ``AuditService.log`` — which is a bare add()+flush() and does not # self-isolate. A swallowed flush failure here therefore deactivates the # transaction and the audit write raises, turning "logout succeeded but # the token was already expired" into a 500 on the logout endpoint. async with best_effort(db, label="invalidate refresh token on logout") as step: from app.core.security import decode_token as _decode_token raw_rt = request.cookies.get("refresh_token") if raw_rt: rt_payload = _decode_token(raw_rt) if rt_payload.get("sub") == str(user.id): service = AuthService(db) await service.logout(rt_payload) if not step.ok: # Don't fail logout if the refresh token is already expired/missing. logger.info("auth.logout: refresh-token invalidation skipped: %s", step.reason()) response.delete_cookie(key="refresh_token", path="/api/v1/auth") audit = AuditService(db) await audit.log( trace_id=request.state.trace_id, actor_user_id=user.id, actor_type="user", action="auth.logout", resource_type="auth_user", resource_id=str(user.id), status="success", ip_address=get_real_ip(request), ) return success_response(data={"message": "Logged out"}, request=request) @router.get("/auth/me") async def get_me( request: Request, user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): service = AuthService(db) is_manager = await service.has_subordinates(user.id) is_attendance_subject = await service.is_attendance_subject(user.id) # global_admin has authority over every location unconditionally; # the frontend treats ``managed_location_ids=[]`` together with # ``roles ∋ global_admin`` as the "no restriction" sentinel. Even # if a global_admin happens to be listed in # ``asset_location_admins`` (dual-role setup), we still return [] # so callers don't have to compute the union themselves. # # ``asset_global_admin`` (全局资产管理员) also has authority over # every location, but the frontend's no-restriction sentinel is # keyed on the global_admin role, so [] would read as "manages # nothing" for them (AssetDetailPage hides the write toolbar when # the set is empty). Materialize the full location-id list instead # — same wire shape as a regional admin, no frontend special case. role_names = set(user.role_names or []) if "global_admin" in role_names or "it_admin" in role_names: managed_location_ids: list = [] elif "asset_global_admin" in role_names: from sqlalchemy import select as _select from app.domains.asset.models import AssetLocation # No ``is_active`` filter — mirrors # ``get_location_ids_managed_by`` (regional admins keep # authority over a deactivated location's leftover assets, so # the global tier must too). rows = await db.execute(_select(AssetLocation.id)) managed_location_ids = sorted(row[0] for row in rows.all()) else: from app.domains.asset.location_admin_sync import ( get_location_ids_managed_by, ) managed_location_ids = sorted( await get_location_ids_managed_by(db, user.id) ) return success_response( data=MeResponse( id=user.id, email=user.work_email, display_name=user.display_name, avatar_url=user.avatar_url, department=user.department, position=user.position, status=user.employment_status, # See ``_user_to_response`` — /auth/me must report the REAL # role set so the frontend role-switcher menu stays # navigable across narrowed sessions. roles=user.actual_role_names, is_manager=is_manager, is_attendance_subject=is_attendance_subject, preferred_language=user.preferred_language, managed_location_ids=managed_location_ids, ).model_dump(), request=request, ) @router.get("/auth/me/profile") async def get_my_profile( request: Request, user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Return full profile for the calling user. Same shape as ``GET /users/{id}`` but scoped to ``self`` and not gated behind ``require_admin``. PII (mobile_phone / personal_email) is returned masked; the caller may unmask its own values via ``POST /auth/me/reveal-contact`` (audit-logged like the admin path). """ service = AuthService(db) target = await service.get_user_detail(user.id) if not target: from app.core.exceptions import NotFoundError raise NotFoundError("User", str(user.id)) mobile_raw = target.mobile_phone email_raw = target.personal_email data = UserDetailResponse( id=target.id, email=target.work_email, display_name=target.display_name, avatar_url=target.avatar_url, employee_id=target.employee_no, department=target.department, position=target.position, status=target.employment_status, # ``actual_role_names`` — this is the SELF profile endpoint. If the # session identity-map returned the same ORM instance the request's # ``X-View-Role`` narrowed, ``target.role_names`` would carry the # narrowed view and any UI driving "show admin tab" off this payload # would silently hide it. Stay symmetric with /auth/me so both # self-shape endpoints report the real role set. roles=target.actual_role_names, manager=ManagerBrief( id=target.manager.id, email=target.manager.work_email, display_name=target.manager.display_name, department=target.manager.department, ) if target.manager else None, subordinates=[ SubordinateBrief( id=s.id, email=s.work_email, display_name=s.display_name, department=s.department, position=s.position, status=s.employment_status, ) for s in target.subordinates # Completed leavers keep their manager pointer + non-deleted # row, so without this they'd linger in "my reports" forever # (same invariant as dashboard ``_subordinate_ids`` / team # footprint). ``leaver_pending`` stays — still employed. if s.deleted_at is None and s.employment_status != "left" ], last_login_at=target.last_login_at, created_at=target.created_at, country=target.country, city=target.city, preferred_language=target.preferred_language, employee_type_code=target.employee_type_code, employee_type=employee_type_slug(target.employee_type_code), github_id=target.github_id, mobile_phone_masked=_mask_phone(mobile_raw), mobile_country_code=target.mobile_country_code, personal_email_masked=_mask_email(email_raw), has_mobile_phone=bool(mobile_raw), has_personal_email=bool(email_raw), # International assignment (外派) — supplementary HRIS context. assignment_home_country=target.assignment_home_country, assignment_host_country=target.assignment_host_country, assignment_start_date=target.assignment_start_date, assignment_end_date=target.assignment_end_date, is_on_international_assignment=target.is_on_international_assignment, hire_date=target.hire_date, termination_date=target.termination_date, work_start_date=target.work_start_date, annual_leave_carry_legal_days=target.annual_leave_carry_legal_days, annual_leave_carry_welfare_days=target.annual_leave_carry_welfare_days, annual_leave_used_legal_days=target.annual_leave_used_legal_days, annual_leave_used_welfare_days=target.annual_leave_used_welfare_days, hris_record_state=target.hris_record_state, hris_purged_at=target.hris_purged_at, hris_purged_reason=target.hris_purged_reason, ).model_dump(mode="json") return success_response(data=data, request=request) @router.post("/auth/me/reveal-contact") async def reveal_my_contact( request: Request, user: AuthUser = Depends(get_current_user), # Pre-launch P1 fix (Round 45 follow-up, CR-317): browser-session # required. Mirrors the admin twin # ``/users/{user_id}/reveal-contact`` which already enforces it. # Without this, a stolen PAT can pull the owner's mobile_phone + # personal_email — the very fields the dual-tier mask exists to # gate. Same posture as ``patch_my_profile``. Declared AFTER # ``get_current_user`` so ``request.state.auth_via`` is set # before the gate reads it. _gate: None = Depends(require_browser_session), _rate: None = Depends(check_sensitive_endpoint_rate_limit), db: AsyncSession = Depends(get_db), ): """Self-service unmask of the caller's own mobile + personal email. Audit-logged exactly like the admin reveal endpoint so a forensic trail survives even when the user reveals their own PII. """ from app.identity.person_models import IdentityPerson target = await db.get(IdentityPerson, user.id) if target is None or target.deleted_at is not None: from app.core.exceptions import NotFoundError raise NotFoundError("User", str(user.id)) audit = AuditService(db) await audit.log( trace_id=getattr(request.state, "trace_id", ""), actor_user_id=user.id, actor_type="user", action="identity.person.contact_reveal_self", resource_type="identity_person", resource_id=str(user.id), status="success", detail={ "revealed_mobile": bool(target.mobile_phone), "revealed_email": bool(target.personal_email), }, ip_address=get_real_ip(request), user_agent=request.headers.get("user-agent"), ) await db.commit() payload = RevealContactResponse( mobile_phone=target.mobile_phone, personal_email=target.personal_email, ).model_dump() return success_response(data=payload, request=request) # Allowed values mirror the locale bundles shipped in # ``web/src/i18n/index.ts``. Validate at the API boundary so a typo # can't silently land the user on the wrong template branch. # zh_CN → Chinese templates; en_US / ja_JP → English templates. _ALLOWED_PREFERRED_LANGUAGES = frozenset({"zh_CN", "en_US", "ja_JP"}) @router.patch("/auth/me/profile") async def patch_my_profile( body: MyProfilePatchRequest, request: Request, user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Update the calling user's own preferred_language. Self-service path — no admin role required, but the PAT-management pattern applies: requests authenticated via PAT are NOT allowed to change a user's locale (a stolen PAT shouldn't be able to mess with the human's day-to-day workspace). JWT-authenticated requests (the web UI's normal flow) work as expected. The chosen language flows two places: * Frontend reads ``preferred_language`` from ``GET /auth/me`` on boot and sets ``i18n.language`` accordingly. * Backend ``NotificationService`` reads it on every send and picks the matching template variant for emails + Feishu cards. Empty string clears the value back to ``NULL``, at which point the platform-wide ``app.default_locale`` (seeded as ``en_US`` per ADR "默认 English") takes over. """ # Refuse PAT-bearer self-mutations of profile preferences for the # same forensic reason ``/me/tokens`` does: a stolen PAT must not # be able to flip the human's preferences (e.g. silently switch # them to a language they don't read so they miss notifications). # # Pre-launch P2 fix (Round 1, SEC-006): fail-closed when # ``auth_via`` is missing — symmetric defense with # ``require_browser_session`` (which was already switched to # fail-closed in 2026-05-06). The default-"jwt" fallback was # benign in practice because ``get_current_user`` always sets # ``auth_via`` before this dependency runs, but a future refactor # that reorders dependencies must not silently re-enable PAT # PATCH on a path that explicitly forbids it. auth_via = getattr(request.state, "auth_via", None) if auth_via is None or auth_via == "pat": raise AppException( code="PAT_CANNOT_PATCH_PROFILE", message=( "Personal access tokens cannot modify profile preferences. " "Sign in via Google SSO to change your language." ), status_code=403, ) from app.identity.person_models import IdentityPerson target = await db.get(IdentityPerson, user.id) if target is None or target.deleted_at is not None: from app.core.exceptions import NotFoundError raise NotFoundError("User", str(user.id)) audit = AuditService(db) changed: dict[str, dict[str, str | None]] = {} # ``None`` = field omitted from the request body (Pydantic default # for an unset field). We intentionally treat empty string as # "clear back to NULL so the platform default takes over" — this # matches the behaviour the personal-settings page advertises. if body.preferred_language is not None: new_value = body.preferred_language.strip() or None if new_value is not None and new_value not in _ALLOWED_PREFERRED_LANGUAGES: raise AppException( code="PREFERRED_LANGUAGE_INVALID", message=( f"preferred_language must be one of " f"{sorted(_ALLOWED_PREFERRED_LANGUAGES)} or empty, " f"got {body.preferred_language!r}" ), status_code=422, ) previous = target.preferred_language if new_value != previous: target.preferred_language = new_value changed["preferred_language"] = { "from": previous, "to": new_value, } # Capture the post-PATCH value before commit/rollback so we can # render it after the session state shifts. SQLAlchemy can detach # the instance after rollback in async mode, causing a refresh-on- # access that explodes in the no-op path. final_value = target.preferred_language if changed: await audit.log( trace_id=getattr(request.state, "trace_id", ""), actor_user_id=user.id, actor_type="user", action="identity.person.profile_patch_self", resource_type="identity_person", resource_id=str(user.id), status="success", detail=changed, ip_address=get_real_ip(request), user_agent=request.headers.get("user-agent"), ) await db.commit() else: # No-op: keep the response shape consistent rather than 304ing # so the frontend can re-fetch /auth/me on every save without # branching on the HTTP status. await db.rollback() # Return the freshly persisted value so the UI doesn't need a # second round-trip to /auth/me to confirm. return success_response( data={"preferred_language": final_value}, request=request, ) # ── User management (IT Admin only) ── @router.get("/auth/users/stats") async def get_user_stats( request: Request, user: AuthUser = Depends(require_hr_read()), db: AsyncSession = Depends(get_db), ): service = AuthService(db) stats = await service.get_user_stats() return success_response(data=stats, request=request) @router.get("/auth/org-tree") async def get_org_tree( request: Request, user: AuthUser = Depends(require_hr_read()), db: AsyncSession = Depends(get_db), ): service = AuthService(db) tree = await service.get_org_tree() return success_response(data=tree, request=request) @router.get("/auth/position-org-chart") async def get_position_org_chart( request: Request, user: AuthUser = Depends(require_hr_read()), db: AsyncSession = Depends(get_db), ): service = AuthService(db) chart = await service.get_position_org_chart() return success_response(data=chart, request=request) @router.get("/auth/departments") async def list_departments( request: Request, user: AuthUser = Depends(require_directory_read()), db: AsyncSession = Depends(get_db), ): """List the active departments for picker dropdowns. Gated by :func:`require_directory_read` (all admin-tier roles) — every admin view (asset / finance / cloud / HR) needs to filter people / assets / payments by department. Department names are public org-chart structure with no PII; widening from ``require_hr_read`` to ``require_directory_read`` lets the asset list / asset intake / payments / cloud filters populate for asset_admin / finance / cloud_admin holders instead of silently returning an empty dropdown. """ service = AuthService(db) departments = await service.get_departments() return success_response(data=departments, request=request) @router.get("/auth/departments/tree") async def list_departments_tree( request: Request, user: AuthUser = Depends(require_directory_read()), db: AsyncSession = Depends(get_db), ): """Hierarchical department nodes for the shared ``DepartmentTreeSelect``. Returns the flat org-node list (``source_id`` / ``name`` / ``parent_source_id`` / ``org_path`` / ``level`` / ``count``); the frontend rebuilds the tree client-side. Same ``require_directory_read`` gate as ``/auth/departments`` — department structure is public org-chart with no PII, so every admin-tier filter (users / cloud / asset / finance) can populate it. Picking a node sends the node *name*; the backend rolls its whole subtree up via :mod:`app.core.org_scope`. """ service = AuthService(db) tree = await service.get_department_tree() return success_response(data=tree, request=request) @router.get("/auth/business-units") async def list_business_units( request: Request, current_user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """List active BUs (top-level org nodes). Available to every logged-in user — the result feeds the BU filter on the software-cost and AI-spending pages, both of which already enforce per-row scoping server-side. ``member_count`` is only returned for admins / auditors; for everyone else the field is redacted to avoid leaking headcount per BU to the entire workforce. """ from app.core.rbac import ( AUDITOR_ROLE, GLOBAL_ADMIN_ROLE, MANAGEMENT_ROLE, has_role, ) from app.identity.bu_helpers import list_bus data = await list_bus(db) # Round-27 pre-launch (2026-05-25): use rbac constants instead of # string literals — a typo (`"sec_auditor"` / `"managment"`) would # silently 403 the wrong role and the positive-path tests wouldn't # catch it. ``has_role`` canonicalises via ``canonicalize_role_name`` # so the constants are still routed through the same compatibility # layer (it_admin → global_admin) just as a string would. can_see_counts = has_role( current_user.role_names, GLOBAL_ADMIN_ROLE, AUDITOR_ROLE, MANAGEMENT_ROLE, ) if not can_see_counts: data = [{"name": row["name"]} for row in data] return success_response(data=data, request=request) @router.get("/directory/people") async def list_directory_people( request: Request, page: int = 1, page_size: int = Query(50, ge=1, le=200), search: str | None = None, department: str | None = None, status: str | None = None, role: str | None = None, user: AuthUser = Depends(require_directory_read()), _rate_limit: None = Depends(check_directory_read_rate_limit), db: AsyncSession = Depends(get_db), ): """Lightweight people picker for admin-tier dropdowns. **What this is** — the search-style person lookup powering every admin view's UserSelect / MultiUserSelect: * asset_admin: 资产清单 → 持有人筛选, 入库 → 指派对象, 变更持有人 / 转交 / 借出 modals, 盘点 cycle scope filter * finance: 付款 / 预算 / 业务付款审批人 / 采购联系人 pickers * cloud_admin: 云资源 owner 转让 / 告警抄送 / scope 添加成员 * hr / global_admin / management / auditor: 通用员工查找 **Why it's separate from /users** — :func:`list_users` returns full HR governance fields (``roles[]``, ``last_login_at``, ``hris_record_state``, manager hierarchy) and stays gated to :func:`require_hr_read`. The picker only needs ``{id, display_name, work_email, employee_no, department, employment_status}`` and that minimal slice is safe for the broader admin tier — exposing display_name + work_email + department to an asset_admin is operationally necessary (they already see assigned_to person names in the asset list) and matches the directory tab semantics every HR org already exposes internally. **Why not just widen ``/users``** — the HR-governance fields ARE sensitive (a non-HR asset_admin should not see who logged in last week or who's been HRIS-purged). Keeping the surface split preserves the principle of least privilege at the field level while unblocking the picker use case at the endpoint level. **Rate limit** — same bucket as the cloud-scope pickable-people endpoint (``check_directory_read_rate_limit``), wider than ``check_sensitive_endpoint_rate_limit`` to accommodate legitimate multi-tab interactive use without 429s. Background: PR 2026-06-01 — regional asset admins reported the asset-management filter dropdowns and 持有人 picker were consistently empty. Backend was returning 403 on the HR-only ``/users``; the frontend ``UserSelect`` swallowed the error via ``catch { /* silent */ }``. Confirmed in prod nginx logs: ``GET /api/v1/users?page=1&page_size=50&status=...`` → 403 from every asset-management page view. """ # ``role`` validation — reject unknown values at the boundary # rather than silently returning an empty list. The unchanged # ``/users`` endpoint has the same silent-empty behaviour, but # introducing the same pattern on a freshly-shipped endpoint # would bake in a quiet-error contract; better to fail loud. # Legacy aliases (``it_admin`` → ``global_admin``, # ``security_auditor`` → ``auditor``) canonicalise transparently, # so callers using either form still work. if role is not None: from app.core.exceptions import ValidationError from app.core.rbac import CANONICAL_ROLE_NAMES, canonicalize_role_name canonical_role = canonicalize_role_name(role) if canonical_role not in CANONICAL_ROLE_NAMES: raise ValidationError( f"unknown role filter: {role!r}. " f"Allowed: {sorted(CANONICAL_ROLE_NAMES)}", ) service = AuthService(db) # ``include_purged=False`` is hard-coded — purged HRIS rows are # not pickable for any operational admin (the operations that # need them — full directory views, lifecycle audits — go through # the HR-gated ``/users`` endpoint). # # ``role`` filter is honoured (e.g. asset-return modal narrows to # ``global_admin`` holders) — the role identity is already a # public org signal in this platform (every employee sees IT # admin names on their asset assignment + approval surfaces), # so exposing "who holds role X" to the admin tier doesn't add a # disclosure step beyond what the broader UI already does. users, total = await service.list_users_filtered( page, page_size, search, department, status, role, include_purged=False, ) data = [ DirectoryPersonResponse( id=u.id, display_name=u.display_name, work_email=u.work_email, employee_no=u.employee_no, department=u.department, employment_status=u.employment_status, ).model_dump() for u in users ] return paginated_response( data=data, total=total, page=page, page_size=page_size, request=request, ) @router.get("/directory/people/lite") async def list_directory_people_lite( request: Request, q: str = Query(default="", max_length=200), limit: int = Query(default=20, ge=1, le=30), db: AsyncSession = Depends(get_db), current_user: AuthUser = Depends(get_current_user), _rate_limit: None = Depends(check_directory_read_rate_limit), ): """Search-only people picker for non-admin self-service flows. Distinct from :func:`list_directory_people` (admin-tier surface with employee_no + role-filter + pagination): this endpoint is callable by any authenticated employee and is the picker used by every self-service modal where one colleague hands a resource off to another — DNS / cloud-server owner transfer, API-key transfer, asset-inventory "I gave it to X", SSO-system owner contact. Why it had to exist: those modals all live on ``/me/...`` / ``/my/...`` routes mounted outside any admin ``RoleRoute``. They were silently broken because ``UserSelect`` called the admin-only ``/directory/people`` endpoint, which 403s for non-admin callers; the frontend ``try/catch`` swallowed the error and the user just saw an empty dropdown (the exact bug the PR 2026-06-01 split was meant to prevent, except it only covered the admin-tier slice). Disclosure surface — equivalent to ``/feed/mention-search``: the fields returned (id / display_name / work_email / department / employment_status) are already visible to every employee through Feishu's internal directory and through the company-wide celebration / feed widgets. ``employee_no`` is deliberately OMITTED — it's an HR-only internal signal and isn't needed for picker UX. Anti-scrape safeguards: * ``q`` is MANDATORY (empty → empty list, no full-roster dump). * ``limit`` capped at 30 — same as ``/feed/mention-search``. * No pagination — a caller can't traverse the directory by bumping page. * Status hard-filtered to ``active + leaver_pending`` and ``hris_record_state = 'present'`` — purged and fully-departed rows never reach the picker. * Rate-limited (same bucket as the admin picker). Search matches across display_name / work_email / name_pinyin / name_pinyin_abbr so a user typing "孟" / "meng" / "MQ" / partial email all resolve. ``employee_no`` is intentionally NOT a search surface here — only HR uses employee numbers as an identifier and HR holds the broader gated endpoint. """ from sqlalchemy import or_, select from sqlalchemy.orm import selectinload keyword = (q or "").strip() if not keyword: # No blank scrape. The picker MUST receive a search term; # without one we return empty so a caller can't paginate the # full roster one keystroke at a time. return success_response(data=[], request=request) like = f"%{keyword}%" stmt = ( select(AuthUser) .options(selectinload(AuthUser.org_memberships)) .where( AuthUser.deleted_at.is_(None), AuthUser.hris_record_state == "present", AuthUser.employment_status.in_(["active", "leaver_pending"]), or_( AuthUser.display_name.ilike(like), AuthUser.work_email.ilike(like), AuthUser.name_pinyin.ilike(like), AuthUser.name_pinyin_abbr.ilike(like), ), ) .order_by(AuthUser.display_name) .limit(limit) ) rows = (await db.execute(stmt)).scalars().all() data = [ { "id": str(p.id), "display_name": p.display_name, "work_email": p.work_email, "department": p.primary_department, "employment_status": p.employment_status, } for p in rows ] return success_response(data=data, request=request) @router.get("/users") async def list_users( request: Request, page: int = 1, page_size: int = Query(20, ge=1, le=200), search: str | None = None, department: str | None = None, status: str | None = None, role: str | None = None, include_purged: bool = False, employee_type: str | None = Query( None, description=( "Comma-separated 用工类型 codes (SF EmpJob.customString25): " "1070 regular / 1071 EOR / 1072 intern / 1073 consultant / " "1074 outsourced / 1075 dispatched." ), ), user: AuthUser = Depends(require_hr_read()), db: AsyncSession = Depends(get_db), ): service = AuthService(db) users, total = await service.list_users_filtered( page, page_size, search, department, status, role, include_purged=include_purged, employee_type=employee_type, ) data = [ UserListItemResponse( id=u.id, email=u.work_email, display_name=u.display_name, avatar_url=u.avatar_url, employee_id=u.employee_no, department=u.department, position=u.position, status=u.employment_status, # ``actual_role_names`` (not ``role_names``): this admin workbench # must show each person's REAL granted roles. The requester's own # row is the SAME ORM instance ``get_current_user`` loaded (shared # session identity-map), so it still carries the # ``_role_names_override`` set from the X-View-Role header — # ``role_names`` would collapse the requester's own row to just # their active view chip (e.g. only ``global_admin``), hiding every # other grant (finance, …) and its remove button while the header # switcher — which reads ``actual_role_names`` — still offered it. # Same self-shape symmetry as /auth/me and /auth/me/profile. roles=u.actual_role_names, manager_name=u.manager.display_name if u.manager else None, manager_email=u.manager.work_email if u.manager else None, manager_employee_no=u.manager.employee_no if u.manager else None, manager_employment_status=u.manager.employment_status if u.manager else None, employee_type_code=u.employee_type_code, employee_type=employee_type_slug(u.employee_type_code), last_login_at=u.last_login_at, created_at=u.created_at, hris_record_state=u.hris_record_state, hris_purged_at=u.hris_purged_at, ).model_dump() for u in users ] return paginated_response(data=data, total=total, page=page, page_size=page_size, request=request) @router.get("/users/{user_id}") async def get_user( user_id: UUID, request: Request, user: AuthUser = Depends(require_hr_read()), db: AsyncSession = Depends(get_db), ): service = AuthService(db) target = await service.get_user_detail(user_id) if not target: from app.core.exceptions import NotFoundError raise NotFoundError("User", str(user_id)) # HR work-relationships (SF EmpJobRelationships). # # HRBP is resolved through the dispatch-authoritative resolver # (``resolve_hrbp_for_person``: SF-authoritative column → Feishu-cached # column → live Feishu walk, each hop requiring a non-deleted person # with ``employment_status == "active"``) so the card shows exactly # who leaver-SLA / attendance notices would target. A plain # ``hris_hrbp_person_id or hrbp_person_id`` short-circuit diverged # from dispatch: a departed-but-not-soft-deleted SF HRBP rendered as # the person's HRBP (violating the leaver-display invariant) while # dispatch had already fallen back to the cached / live-resolved one. # One extra resolve call is fine — this is the single-user detail # endpoint, not a list. # # hr_ops / comp_owner have no fallback resolver — resolve them in one # batched SELECT mirroring the resolver's liveness filter (non-deleted # AND active) so a departed person renders as absent rather than stale. from sqlalchemy import select from app.identity.person_models import IdentityPerson from app.integrations.feishu_hrbp import resolve_hrbp_for_person hr_ops_id = target.hris_hrops_person_id comp_owner_id = target.hris_budget_owner_person_id _rel_ids = {i for i in (hr_ops_id, comp_owner_id) if i is not None} _rel_map: dict[UUID, IdentityPerson] = {} if _rel_ids: _rows = await db.execute( select(IdentityPerson).where( IdentityPerson.id.in_(_rel_ids), IdentityPerson.deleted_at.is_(None), IdentityPerson.employment_status == "active", ) ) _rel_map = {p.id: p for p in _rows.scalars().all()} def _rel_brief(person_id: UUID | None) -> ManagerBrief | None: p = _rel_map.get(person_id) if person_id else None if p is None: return None return ManagerBrief( id=p.id, email=p.work_email, display_name=p.display_name, department=p.department, ) hrbp_brief: ManagerBrief | None = None _hrbp_res = await resolve_hrbp_for_person(db, target) if _hrbp_res.person_id is not None: # Resolver guarantees the row exists, is non-deleted and active. # Re-load via SELECT (not ``db.get``) so the ``selectin``-eager # ``org_memberships`` collection is populated — ``.department`` # is a property over it and a bare identity-map hit would trip # MissingGreenlet on the sync lazy load. _hrbp_row = (await db.execute( select(IdentityPerson).where( IdentityPerson.id == _hrbp_res.person_id, ) )).scalar_one_or_none() if _hrbp_row is not None: hrbp_brief = ManagerBrief( id=_hrbp_row.id, email=_hrbp_row.work_email, display_name=_hrbp_row.display_name, department=_hrbp_row.department, ) mobile_raw = target.mobile_phone email_raw = target.personal_email data = UserDetailResponse( id=target.id, email=target.work_email, display_name=target.display_name, avatar_url=target.avatar_url, employee_id=target.employee_no, department=target.department, position=target.position, status=target.employment_status, # Real grants, not the requester's X-View-Role narrowing — same # identity-map reasoning as the list endpoint and /auth/me/profile. roles=target.actual_role_names, manager=ManagerBrief( id=target.manager.id, email=target.manager.work_email, display_name=target.manager.display_name, department=target.manager.department, ) if target.manager else None, hrbp=hrbp_brief, hr_ops=_rel_brief(hr_ops_id), comp_owner=_rel_brief(comp_owner_id), subordinates=[ SubordinateBrief( id=s.id, email=s.work_email, display_name=s.display_name, department=s.department, position=s.position, status=s.employment_status, ) for s in target.subordinates # Completed leavers keep their manager pointer + non-deleted # row, so without this they'd linger in "my reports" forever # (same invariant as dashboard ``_subordinate_ids`` / team # footprint). ``leaver_pending`` stays — still employed. if s.deleted_at is None and s.employment_status != "left" ], last_login_at=target.last_login_at, created_at=target.created_at, # HRIS-sourced fields (migration 048). PII values masked by # default; reveal via POST /users/{id}/reveal-contact. country=target.country, city=target.city, preferred_language=target.preferred_language, employee_type_code=target.employee_type_code, employee_type=employee_type_slug(target.employee_type_code), github_id=target.github_id, mobile_phone_masked=_mask_phone(mobile_raw), mobile_country_code=target.mobile_country_code, personal_email_masked=_mask_email(email_raw), has_mobile_phone=bool(mobile_raw), has_personal_email=bool(email_raw), # International assignment (外派) — supplementary HRIS context. assignment_home_country=target.assignment_home_country, assignment_host_country=target.assignment_host_country, assignment_start_date=target.assignment_start_date, assignment_end_date=target.assignment_end_date, is_on_international_assignment=target.is_on_international_assignment, hire_date=target.hire_date, termination_date=target.termination_date, work_start_date=target.work_start_date, annual_leave_carry_legal_days=target.annual_leave_carry_legal_days, annual_leave_carry_welfare_days=target.annual_leave_carry_welfare_days, annual_leave_used_legal_days=target.annual_leave_used_legal_days, annual_leave_used_welfare_days=target.annual_leave_used_welfare_days, hris_record_state=target.hris_record_state, hris_purged_at=target.hris_purged_at, hris_purged_reason=target.hris_purged_reason, ).model_dump(mode="json") return success_response(data=data, request=request) @router.post("/users/{user_id}/reveal-contact") async def reveal_user_contact( user_id: UUID, request: Request, current_user: AuthUser = Depends(require_hr_admin()), db: AsyncSession = Depends(get_db), _rate: None = Depends(check_sensitive_endpoint_rate_limit), _gate: None = Depends(require_browser_session), ): """Return unmasked ``mobile_phone`` + ``personal_email`` for one user. Every call writes an ``audit_logs`` row naming the admin who revealed the PII and the target user. The frontend shows masked values by default; clicking the "显示" eye icon triggers this endpoint, logs the access, then unmasks just for that session. """ from app.identity.person_models import IdentityPerson target = await db.get(IdentityPerson, user_id) if target is None or target.deleted_at is not None: from app.core.exceptions import NotFoundError raise NotFoundError("User", str(user_id)) audit = AuditService(db) await audit.log( trace_id=getattr(request.state, "trace_id", ""), actor_user_id=current_user.id, actor_type="user", action="identity.person.contact_reveal", resource_type="identity_person", resource_id=str(user_id), status="success", detail={ "revealed_mobile": bool(target.mobile_phone), "revealed_email": bool(target.personal_email), "target_display_name": target.display_name, "target_work_email": target.work_email, }, ip_address=get_real_ip(request), user_agent=request.headers.get("user-agent"), ) await db.commit() payload = RevealContactResponse( mobile_phone=target.mobile_phone, personal_email=target.personal_email, ).model_dump() return success_response(data=payload, request=request) @router.patch("/users/{user_id}") async def update_user( user_id: UUID, body: UserUpdateRequest, request: Request, user: AuthUser = Depends(require_admin()), db: AsyncSession = Depends(get_db), _gate: None = Depends(require_browser_session), ): service = AuthService(db) updated = await service.update_user(user_id, body.model_dump(exclude_unset=True)) is_manager = await service.has_subordinates(updated.id) audit = AuditService(db) await audit.log( trace_id=request.state.trace_id, actor_user_id=user.id, actor_type="user", action="user.update", resource_type="auth_user", resource_id=str(user_id), status="success", detail=body.model_dump(mode="json", exclude_unset=True), ip_address=get_real_ip(request), ) return success_response(data=_user_to_response(updated, is_manager), request=request) @router.get("/users/{user_id}/roles") async def get_user_roles( user_id: UUID, request: Request, user: AuthUser = Depends(require_hr_read()), db: AsyncSession = Depends(get_db), ): """List the canonical roles assigned to a specific user.""" service = AuthService(db) target = await service.find_user_by_id(user_id) if not target: from app.core.exceptions import NotFoundError raise NotFoundError("User", str(user_id)) available_roles = {role.name: role for role in await service.list_roles()} data = [ RoleResponse( id=available_roles[role_name].id, name=available_roles[role_name].name, description=available_roles[role_name].description, is_system=available_roles[role_name].is_system, ).model_dump() for role_name in target.role_names if role_name in available_roles ] return success_response(data=data, request=request) @router.post("/users/{user_id}/roles") # RBAC grant: a replayed assign would write a second audit row and, for # roles with derived side effects, re-run them. @idempotent("auth.assign_role", key_params=("user_id", "user.id")) async def assign_role( user_id: UUID, body: RoleAssignRequest, request: Request, user: AuthUser = Depends(require_admin()), db: AsyncSession = Depends(get_db), _gate: None = Depends(require_browser_session), ): # ``asset_admin`` is fully auto-derived from the office admin # list (PR 2026-05-29). Manual grants would silently be undone # by the next reconciler run — reject them up front so the API # behaviour matches the UI and the audit log doesn't accumulate # phantom grant/revoke pairs. from app.core.rbac import ASSET_ADMIN_ROLE, canonicalize_role_name if canonicalize_role_name(body.role_name) == ASSET_ADMIN_ROLE: raise AppException( code="ROLE_AUTO_DERIVED", message=( "asset_admin is auto-derived from the office admin list; " "manage membership at /admin/office-addresses." ), status_code=400, ) service = AuthService(db) try: # NB: granting ``asset_global_admin`` co-derives ``asset_admin`` # inside the service (``_sync_asset_admin_co_derivation``). await service.assign_role(user_id, body.role_name, user.id) except ValueError as e: raise AppException(code="ROLE_ASSIGN_FAILED", message=str(e)) audit = AuditService(db) await audit.log( trace_id=request.state.trace_id, actor_user_id=user.id, actor_type="user", action="user.role.assign", resource_type="auth_user_role", resource_id=f"{user_id}/{body.role_name}", status="success", ip_address=get_real_ip(request), ) return success_response(data={"message": "Role assigned"}, request=request) @router.delete("/users/{user_id}/roles/{role_id}") async def remove_role( user_id: UUID, role_id: UUID, request: Request, user: AuthUser = Depends(require_admin()), db: AsyncSession = Depends(get_db), _gate: None = Depends(require_browser_session), ): # Refuse manual removal of asset_admin — auto-derived. from app.auth.models import AuthRole from app.core.rbac import ASSET_ADMIN_ROLE role = await db.get(AuthRole, role_id) if role is not None and role.name == ASSET_ADMIN_ROLE: raise AppException( code="ROLE_AUTO_DERIVED", message=( "asset_admin is auto-derived from the office admin list; " "manage membership at /admin/office-addresses." ), status_code=400, ) service = AuthService(db) # NB: removing ``asset_global_admin`` revokes the co-derived # ``asset_admin`` inside the service. await service.remove_role(user_id, role_id) audit = AuditService(db) await audit.log( trace_id=request.state.trace_id, actor_user_id=user.id, actor_type="user", action="user.role.remove", resource_type="auth_user_role", resource_id=f"{user_id}/{role_id}", status="success", ip_address=get_real_ip(request), ) return success_response(data={"message": "Role removed"}, request=request) @router.post("/auth/roles/batch-assign") async def batch_assign_roles( body: RoleBatchRequest, request: Request, user: AuthUser = Depends(require_admin()), db: AsyncSession = Depends(get_db), _gate: None = Depends(require_browser_session), ): """Assign one role to many users in a single request. Per-row SAVEPOINT isolation; users already holding the role → ``skipped``. Same RBAC + browser-session gate as the single-row endpoint, and the same ``asset_admin`` auto-derived rejection. """ from app.core.exceptions import NotFoundError from app.core.rbac import ASSET_ADMIN_ROLE, canonicalize_role_name canonical_role = canonicalize_role_name(body.role_name) if canonical_role == ASSET_ADMIN_ROLE: raise AppException( code="ROLE_AUTO_DERIVED", message=( "asset_admin is auto-derived from the office admin list; " "manage membership at /admin/office-addresses." ), status_code=400, ) service = AuthService(db) known_roles = {role.name for role in await service.list_roles()} if canonical_role not in known_roles: raise NotFoundError("Role", canonical_role) results: list[dict] = [] succeeded = skipped = failed = 0 for user_id in body.user_ids: try: async with db.begin_nested(): await service.assign_role(user_id, canonical_role, user.id) succeeded += 1 results.append({"id": str(user_id), "status": "succeeded"}) except ValueError: # ``Role already assigned`` — idempotent retry. skipped += 1 results.append({"id": str(user_id), "status": "skipped"}) except AppException as e: failed += 1 results.append({"id": str(user_id), "status": "failed", "error": e.code}) except Exception as e: # noqa: BLE001 — single-row failure must not sink the batch # A lock-wait/deadlock (1205/1213/1305) rolls back the WHOLE # transaction server-side — the savepoint is already gone, so # swallowing it turns a transient, retryable deadlock into a # "SAVEPOINT does not exist" cascade that marks every remaining # row failed and commits a partial batch. Re-raise so the # caller restarts on a fresh tx (mirrors permission.batch_revoke). if is_transaction_aborting_db_error(e): raise failed += 1 logger.error( "auth.roles.batch_assign row failed user_id=%s trace_id=%s " "error_type=%s", user_id, request.state.trace_id, type(e).__name__, ) results.append({ "id": str(user_id), "status": "failed", "error": "INTERNAL_ERROR", }) audit = AuditService(db) await audit.log( trace_id=request.state.trace_id, actor_user_id=user.id, actor_type="user", action="auth.roles.batch_assign", resource_type="auth_user_role", resource_id=canonical_role, status="success", detail={ "role_name": canonical_role, "total": len(body.user_ids), "succeeded": succeeded, "skipped": skipped, "failed": failed, "ids": [str(uid) for uid in body.user_ids[:50]], }, ip_address=get_real_ip(request), ) await db.commit() return success_response( data={ "succeeded": succeeded, "skipped": skipped, "failed": failed, "results": results, }, request=request, ) @router.post("/auth/roles/batch-revoke") async def batch_revoke_roles( body: RoleBatchRequest, request: Request, user: AuthUser = Depends(require_admin()), db: AsyncSession = Depends(get_db), _gate: None = Depends(require_browser_session), ): """Remove one role from many users in a single request. Users who don't hold the role → ``skipped``. Self-lockout guard: the caller cannot batch-remove ``global_admin`` from themselves — that row fails with ``cannot_remove_own_admin`` (the single-row last-global-admin guard still applies per row on top). """ from app.core.exceptions import NotFoundError from app.core.rbac import ( ASSET_ADMIN_ROLE, GLOBAL_ADMIN_ROLE, canonicalize_role_name, ) canonical_role = canonicalize_role_name(body.role_name) if canonical_role == ASSET_ADMIN_ROLE: raise AppException( code="ROLE_AUTO_DERIVED", message=( "asset_admin is auto-derived from the office admin list; " "manage membership at /admin/office-addresses." ), status_code=400, ) service = AuthService(db) known_roles = {role.name for role in await service.list_roles()} if canonical_role not in known_roles: raise NotFoundError("Role", canonical_role) results: list[dict] = [] succeeded = skipped = failed = 0 for user_id in body.user_ids: if canonical_role == GLOBAL_ADMIN_ROLE and user_id == user.id: # Self-lockout is a real incident surface: an admin # batch-selecting "everyone" must not strip their own # global_admin in passing. failed += 1 results.append({ "id": str(user_id), "status": "failed", "error": "cannot_remove_own_admin", }) continue try: async with db.begin_nested(): await service.remove_role_by_name(user_id, canonical_role) succeeded += 1 results.append({"id": str(user_id), "status": "succeeded"}) except NotFoundError: # User doesn't hold the role — already in target state. skipped += 1 results.append({"id": str(user_id), "status": "skipped"}) except AppException as e: failed += 1 results.append({"id": str(user_id), "status": "failed", "error": e.code}) except Exception as e: # noqa: BLE001 — single-row failure must not sink the batch # See batch_assign_roles: re-raise transaction-aborting DB # errors so a deadlock doesn't cascade into a silent partial # batch via "SAVEPOINT does not exist". if is_transaction_aborting_db_error(e): raise failed += 1 logger.error( "auth.roles.batch_revoke row failed user_id=%s trace_id=%s " "error_type=%s", user_id, request.state.trace_id, type(e).__name__, ) results.append({ "id": str(user_id), "status": "failed", "error": "INTERNAL_ERROR", }) audit = AuditService(db) await audit.log( trace_id=request.state.trace_id, actor_user_id=user.id, actor_type="user", action="auth.roles.batch_revoke", resource_type="auth_user_role", resource_id=canonical_role, status="success", detail={ "role_name": canonical_role, "total": len(body.user_ids), "succeeded": succeeded, "skipped": skipped, "failed": failed, "ids": [str(uid) for uid in body.user_ids[:50]], }, ip_address=get_real_ip(request), ) await db.commit() return success_response( data={ "succeeded": succeeded, "skipped": skipped, "failed": failed, "results": results, }, request=request, ) @router.delete("/users/{user_id}/roles/by-name/{role_name}") async def remove_role_by_name( user_id: UUID, role_name: str, request: Request, user: AuthUser = Depends(require_admin()), db: AsyncSession = Depends(get_db), _gate: None = Depends(require_browser_session), ): """Remove a role from a user by canonical role name.""" from app.core.rbac import ASSET_ADMIN_ROLE, canonicalize_role_name if canonicalize_role_name(role_name) == ASSET_ADMIN_ROLE: raise AppException( code="ROLE_AUTO_DERIVED", message=( "asset_admin is auto-derived from the office admin list; " "manage membership at /admin/office-addresses." ), status_code=400, ) service = AuthService(db) # NB: removing ``asset_global_admin`` revokes the co-derived # ``asset_admin`` inside the service. await service.remove_role_by_name(user_id, role_name) audit = AuditService(db) await audit.log( trace_id=request.state.trace_id, actor_user_id=user.id, actor_type="user", action="user.role.remove", resource_type="auth_user_role", resource_id=f"{user_id}/{role_name}", status="success", ip_address=get_real_ip(request), ) return success_response(data={"message": "Role removed"}, request=request) @router.get("/roles") async def list_roles( request: Request, user: AuthUser = Depends(require_hr_read()), db: AsyncSession = Depends(get_db), ): service = AuthService(db) roles = await service.list_roles() data = [RoleResponse.model_validate(r).model_dump() for r in roles] return success_response(data=data, request=request)