"""Feishu (Lark) OAuth 2.0 login client. Reuses the ``chat_approval`` integration row for ``app_id`` / ``app_secret`` — operators manage a single Feishu app for approval, notifications, and SSO. Storing a separate copy here would double the secret-rotation surface for no benefit. Flow: 1. Frontend POSTs ``/auth/feishu/authorize-url`` with the page's own redirect_uri (``window.location.origin + "/auth/feishu/callback"``). 2. Backend generates a 256-bit ``state`` token, persists it in Redis with a 10-minute TTL bound to that redirect_uri, returns the full Feishu authorize URL. 3. Frontend redirects browser to the authorize URL. User logs in on Feishu, gets redirected back to the callback URL with ``?code=&state=``. 4. Frontend POSTs ``/auth/feishu/verify`` with ``{code, state, redirect_uri}``. 5. Backend single-use consumes the state (atomic GETDEL), verifies it matches the same redirect_uri, exchanges code → user_access_token → user_info via Feishu OAuth v2, then matches by enterprise_email → ``identity_persons.work_email`` and issues the same JWT pair Google SSO does. Refresh token rides in the existing httpOnly cookie. Defences applied (mirror Google SEC-001): - ``trust_env=False`` on every outbound httpx call — a rogue ``HTTPS_PROXY`` env var must not be able to MitM identity bootstrap. - ``follow_redirects=False`` — both Feishu endpoints answer 200/4xx; an unexpected 30x is a sign of compromise. - redirect_uri must match one of ``settings.CORS_ORIGINS`` + the callback path — defence in depth alongside the Feishu open-platform whitelist. - State is single-use (atomic GETDEL) and bound to redirect_uri. """ from __future__ import annotations import logging import secrets from typing import Any from urllib.parse import urlencode import httpx from sqlalchemy.ext.asyncio import AsyncSession from app.core.config import settings from app.core.exceptions import AppException from app.core.redis_client import get_redis from app.integrations.compat import get_feishu_config logger = logging.getLogger(__name__) # State TTL — short enough that a stalled browser tab can't be revived # weeks later for replay; long enough to cover a slow Feishu login # (MFA, password reset, etc.). _STATE_TTL_SECONDS = 600 _STATE_REDIS_PREFIX = "feishu_sso:state:" # Feishu OAuth 2.0 endpoints. The v2 token exchange takes # client_id / client_secret directly (OAuth 2.0 standard), unlike the # older v1 flow that required a separate app_access_token round-trip. # /authen/v1/user_info is stable across both — no v2 user-info endpoint # exists yet. _AUTHORIZE_URL = "https://accounts.feishu.cn/open-apis/authen/v1/authorize" _TOKEN_URL = "https://open.feishu.cn/open-apis/authen/v2/oauth/token" _USER_INFO_URL = "https://open.feishu.cn/open-apis/authen/v1/user_info" _HTTP_TIMEOUT_SECONDS = 10.0 def _callback_path() -> str: return "/auth/feishu/callback" def allowed_redirect_uris() -> set[str]: """Set of exact-match redirect_uris this deploy accepts. Built from ``settings.CORS_ORIGINS`` — the same allowlist already enforced at the browser layer for cross-origin requests. The Feishu open-platform's own redirect-URI whitelist is the authoritative gate; this set is defence-in-depth so a future Feishu whitelist misconfiguration cannot turn Helix into an open redirect oracle. """ return { f"{origin.rstrip('/')}{_callback_path()}" for origin in settings.CORS_ORIGINS if origin and origin != "*" } def validate_redirect_uri(redirect_uri: str) -> None: """Reject redirect URIs not in the allowed origins set. Raises :class:`ValueError` so the caller wraps it in the standard login-failure audit + 401 response. """ if redirect_uri not in allowed_redirect_uris(): raise ValueError( "redirect_uri not allowed for this deployment", ) async def _store_state(state: str, redirect_uri: str) -> None: redis = get_redis() await redis.setex( f"{_STATE_REDIS_PREFIX}{state}", _STATE_TTL_SECONDS, redirect_uri, ) async def _consume_state(state: str) -> str | None: """Single-use consume of a state token. Returns the stored redirect_uri or None if the state never existed / is expired / has already been consumed. Uses Redis ``GETDEL`` so concurrent verify calls racing on the same code+state (e.g. a refresh on the callback page) can't both succeed. """ redis = get_redis() key = f"{_STATE_REDIS_PREFIX}{state}" value = await redis.getdel(key) if value is None: return None # redis_client.get_redis() uses decode_responses=True so values # come back as str. Guard against a future config flip anyway. if isinstance(value, bytes): return value.decode("utf-8") return value async def _get_feishu_credentials(db: AsyncSession) -> tuple[str, str]: """Pull app_id / app_secret from the chat_approval integration. Threads the request's DB session through so the lookup honours the test-suite's session override (``compat.get_feishu_config`` falls back to ``async_session_factory()`` when no session is passed, which doesn't respect the ``conftest`` monkey-patch — direct ``from app.core.database import async_session_factory`` captures the original reference at import time). Raises ``AppException(503, FEISHU_NOT_CONFIGURED)`` if no Feishu integration is configured yet. """ cfg = await get_feishu_config(session=db) app_id = cfg.get("app_id") app_secret = cfg.get("app_secret") if not app_id or not app_secret: raise AppException( code="FEISHU_NOT_CONFIGURED", message=( "Feishu integration is not configured. Configure the " "Feishu app in Admin > Integrations before using " "Feishu SSO." ), status_code=503, ) return app_id, app_secret async def build_authorize_url(db: AsyncSession, redirect_uri: str) -> str: """Validate redirect_uri, persist a fresh state, return the Feishu authorize URL the frontend should navigate to. """ validate_redirect_uri(redirect_uri) app_id, _ = await _get_feishu_credentials(db) state = secrets.token_urlsafe(32) await _store_state(state, redirect_uri) params = { "app_id": app_id, "redirect_uri": redirect_uri, "state": state, "response_type": "code", } return f"{_AUTHORIZE_URL}?{urlencode(params)}" async def _exchange_code( db: AsyncSession, code: str, redirect_uri: str | None, ) -> dict[str, Any]: """Inner: call Feishu's v2 OAuth token endpoint + v1 user_info. ``redirect_uri=None`` is used by the H5 SDK flow which never went through an OAuth authorize redirect. Per OAuth 2.0 spec, when the authorize step did not pass ``redirect_uri`` (or there was no authorize step) the token request MUST NOT include it either. """ app_id, app_secret = await _get_feishu_credentials(db) token_body: dict[str, str] = { "grant_type": "authorization_code", "client_id": app_id, "client_secret": app_secret, "code": code, } if redirect_uri is not None: token_body["redirect_uri"] = redirect_uri async with httpx.AsyncClient( timeout=_HTTP_TIMEOUT_SECONDS, trust_env=False, follow_redirects=False, ) as client: try: token_resp = await client.post( _TOKEN_URL, json=token_body, headers={"Content-Type": "application/json; charset=utf-8"}, ) except httpx.HTTPError as exc: raise ValueError( f"Feishu token exchange transport error: {type(exc).__name__}" ) from exc if token_resp.status_code != 200: raise ValueError( f"Feishu token exchange failed: HTTP {token_resp.status_code}" ) try: token_payload = token_resp.json() except ValueError as exc: raise ValueError( "Feishu token response was not valid JSON" ) from exc # Feishu surfaces some OAuth errors as 200 with an ``error`` # field; v2 also returns ``code`` (0 = success). Treat anything # non-zero / non-null as failure but DON'T echo the raw body # back to the caller — it can contain the app_secret in debug # builds. if token_payload.get("error") or token_payload.get("code") not in (None, 0): logger.warning( "Feishu token exchange rejected: code=%s error=%s", token_payload.get("code"), token_payload.get("error"), ) raise ValueError("Feishu token exchange rejected") access_token = token_payload.get("access_token") if not access_token or not isinstance(access_token, str): raise ValueError("Feishu token response missing access_token") try: info_resp = await client.get( _USER_INFO_URL, headers={"Authorization": f"Bearer {access_token}"}, ) except httpx.HTTPError as exc: raise ValueError( f"Feishu user_info transport error: {type(exc).__name__}" ) from exc if info_resp.status_code != 200: raise ValueError( f"Feishu user_info fetch failed: HTTP {info_resp.status_code}" ) try: info_payload = info_resp.json() except ValueError as exc: raise ValueError("Feishu user_info response was not valid JSON") from exc if info_payload.get("code") not in (None, 0): logger.warning( "Feishu user_info rejected: code=%s msg=%s", info_payload.get("code"), info_payload.get("msg"), ) raise ValueError("Feishu user_info rejected") data = info_payload.get("data") if not isinstance(data, dict): raise ValueError("Feishu user_info returned non-dict data") return data async def exchange_code_for_user_info( db: AsyncSession, code: str, redirect_uri: str, ) -> dict[str, Any]: """Exchange an authorization code for the user's profile via Feishu OAuth v2 (redirect-based web flow). Returns the ``data`` block of the user_info response which contains ``open_id``, ``union_id``, ``name``, ``en_name``, ``avatar_url``, ``email``, ``enterprise_email``, ``mobile``, ``tenant_key`` (and ``user_id`` when the app holds that scope). Raises :class:`ValueError` on any failure so the caller can wrap it into the standard login-failure audit + 401 response. """ return await _exchange_code(db, code, redirect_uri) async def exchange_h5_code_for_user_info( db: AsyncSession, code: str, ) -> dict[str, Any]: """Exchange an H5-SDK ``tt.requestAccess`` code for the user's profile. Used by the in-Feishu webview login flow (``/auth/feishu/h5-login``). The JS SDK returns a code WITHOUT performing an OAuth authorize redirect — so no ``state`` token and no ``redirect_uri`` exist to validate. CSRF is structurally impossible here: ``tt.requestAccess`` only succeeds when the page is loaded inside the Feishu app's own webview, which carries the user's authenticated session, and the code is bound to (app_id, user_id) on the Feishu side. A page rendered in a normal browser cannot obtain such a code. Returns the same ``data`` block shape as :func:`exchange_code_for_user_info` so the downstream ``AuthService.login_via_feishu`` path is identical. """ return await _exchange_code(db, code, None) async def verify_state(state: str, redirect_uri: str) -> None: """Consume a state token and confirm it was issued for the same redirect_uri. Raises :class:`ValueError` if the state is invalid, expired, replayed, or paired with a different URI. """ stored_uri = await _consume_state(state) if stored_uri is None: raise ValueError("Invalid or expired state") if stored_uri != redirect_uri: raise ValueError("State / redirect_uri mismatch") def pick_match_emails(user_info: dict[str, Any]) -> tuple[str | None, str | None]: """Return normalised ``(enterprise_email, email)`` candidates. Both fields are returned independently so the caller can apply the admin-bound-vs-self-editable trust distinction (see :func:`AuthService.login_via_feishu` for the local-part anti-takeover guard applied when falling back from ``enterprise_email`` to ``email``). Pre-2026-05-29 the single-string ``pick_match_email`` returned ``enterprise_email or email`` — that collapsed both candidates into one, so when the Feishu tenant admin rotated ``enterprise_email`` to a different domain (``@pingcap.com`` → ``@pingcap.com.cn``) the code never even *tried* the still-correct ``email`` field and locked already-bound users out. Returning both lets the caller try each candidate with appropriate trust gates. """ def _norm(raw: Any) -> str | None: if not isinstance(raw, str): return None normalised = raw.strip().lower() return normalised or None return _norm(user_info.get("enterprise_email")), _norm(user_info.get("email"))