|
| 1 | +import uuid |
| 2 | +from contextvars import ContextVar, Token |
| 3 | +from typing import Any |
| 4 | +from uuid import UUID |
| 5 | + |
| 6 | +import httpx |
| 7 | +from fastapi import HTTPException, Request, status |
| 8 | +from jose import JWTError, jwk, jwt |
| 9 | +from sqlmodel.ext.asyncio.session import AsyncSession |
| 10 | +from src.backend.base.langflow.logging.logger import logger |
| 11 | + |
| 12 | +from langflow.services.database.models.user import User, UserCreate |
| 13 | +from langflow.services.database.models.user.crud import get_user_by_id |
| 14 | +from langflow.services.deps import get_settings_service |
| 15 | + |
| 16 | +# Context variable to store decoded clerk claims per request |
| 17 | +auth_header_ctx: ContextVar[dict | None] = ContextVar("auth_header_ctx", default=None) |
| 18 | + |
| 19 | +_jwks_cache: dict[str, dict[str, Any]] = {} |
| 20 | + |
| 21 | +# APIs that require Clerk token decoding in middleware |
| 22 | +PROTECTED_PATHS = ["/api/v1/users/"] |
| 23 | + |
| 24 | + |
| 25 | +async def _get_jwks(issuer: str) -> dict[str, Any]: |
| 26 | + """Retrieve and cache JWKS for a Clerk issuer.""" |
| 27 | + issuer = issuer.rstrip("/") |
| 28 | + if issuer not in _jwks_cache: |
| 29 | + url = f"{issuer}/.well-known/jwks.json" |
| 30 | + async with httpx.AsyncClient() as client: |
| 31 | + response = await client.get(url) |
| 32 | + response.raise_for_status() |
| 33 | + data = response.json() |
| 34 | + _jwks_cache[issuer] = {k["kid"]: k for k in data.get("keys", [])} |
| 35 | + return _jwks_cache[issuer] |
| 36 | + |
| 37 | + |
| 38 | +async def verify_clerk_token(token: str) -> dict[str, Any]: |
| 39 | + """Verify a Clerk token, add a UUID derived from the Clerk ID, and return the payload.""" |
| 40 | + try: |
| 41 | + unverified_header = jwt.get_unverified_header(token) |
| 42 | + unverified_claims = jwt.get_unverified_claims(token) |
| 43 | + issuer: str | None = unverified_claims.get("iss") |
| 44 | + kid: str | None = unverified_header.get("kid") |
| 45 | + if not issuer or not kid: |
| 46 | + msg = "Missing issuer or kid" |
| 47 | + raise JWTError(msg) |
| 48 | + jwks = await _get_jwks(issuer) |
| 49 | + key = jwks.get(kid) |
| 50 | + if not key: |
| 51 | + _jwks_cache.pop(issuer, None) # force refresh |
| 52 | + jwks = await _get_jwks(issuer) |
| 53 | + key = jwks.get(kid) |
| 54 | + if not key: |
| 55 | + msg = "Public key not found" |
| 56 | + raise JWTError(msg) |
| 57 | + |
| 58 | + public_key = jwk.construct(key, unverified_header.get("alg", "RS256")) |
| 59 | + payload = jwt.decode( |
| 60 | + token, |
| 61 | + public_key, |
| 62 | + algorithms=[unverified_header.get("alg", "RS256")], |
| 63 | + audience=unverified_claims.get("aud"), |
| 64 | + issuer=issuer, |
| 65 | + ) |
| 66 | + # ✅ Add deterministic UUID to the payload |
| 67 | + clerk_id = payload.get("sub") |
| 68 | + if not clerk_id: |
| 69 | + msg = "Missing 'sub' (Clerk ID) in token payload" |
| 70 | + raise JWTError(msg) |
| 71 | + payload["uuid"] = str(uuid.uuid5(uuid.NAMESPACE_DNS, str(clerk_id))) |
| 72 | + |
| 73 | + except JWTError as exc: |
| 74 | + msg = "Invalid token" |
| 75 | + raise ValueError(msg) from exc |
| 76 | + return payload |
| 77 | + |
| 78 | + |
| 79 | +async def process_new_user_with_clerk(_user: UserCreate, new_user: User): |
| 80 | + settings = get_settings_service().auth_settings |
| 81 | + # ✅ If Clerk is enabled, pull UUID from enriched auth_header_ctx payload |
| 82 | + if settings.CLERK_AUTH_ENABLED: |
| 83 | + payload = auth_header_ctx.get() |
| 84 | + if not payload: |
| 85 | + raise HTTPException(status_code=401, detail="Missing Clerk payload") |
| 86 | + clerk_uuid = payload.get("uuid") |
| 87 | + if not clerk_uuid: |
| 88 | + raise HTTPException(status_code=401, detail="Missing Clerk UUID") |
| 89 | + new_user.id = UUID(clerk_uuid) |
| 90 | + logger.info(f"[process_new_user_with_clerk] Assigned Clerk UUID {new_user.id} to new user object") |
| 91 | + |
| 92 | +async def get_user_from_clerk_payload(token: str, db: AsyncSession) -> User: |
| 93 | + """Retrieve the current user using the payload from ``verify_clerk_token``.""" |
| 94 | + try: |
| 95 | + payload = await verify_clerk_token(token) |
| 96 | + except Exception as exc: |
| 97 | + raise HTTPException( |
| 98 | + status_code=status.HTTP_401_UNAUTHORIZED, |
| 99 | + detail="Authentication failed", |
| 100 | + headers={"WWW-Authenticate": "Bearer"}, |
| 101 | + ) from exc |
| 102 | + |
| 103 | + uuid_str = payload.get("uuid") |
| 104 | + logger.info(f"uuid_str: {uuid_str}") |
| 105 | + if not uuid_str: |
| 106 | + raise HTTPException( |
| 107 | + status_code=status.HTTP_401_UNAUTHORIZED, |
| 108 | + detail="Missing Clerk UUID", |
| 109 | + headers={"WWW-Authenticate": "Bearer"}, |
| 110 | + ) |
| 111 | + |
| 112 | + try: |
| 113 | + user_id = UUID(uuid_str) |
| 114 | + except ValueError as err: |
| 115 | + raise HTTPException( |
| 116 | + status_code=status.HTTP_401_UNAUTHORIZED, |
| 117 | + detail="Invalid Clerk UUID format", |
| 118 | + headers={"WWW-Authenticate": "Bearer"}, |
| 119 | + ) from err |
| 120 | + |
| 121 | + user = await get_user_by_id(db, user_id) |
| 122 | + logger.info(f"Retrieved user: {user}") |
| 123 | + if user is None: |
| 124 | + logger.info("User not found.") |
| 125 | + raise HTTPException( |
| 126 | + status_code=status.HTTP_401_UNAUTHORIZED, |
| 127 | + detail="User not found.", |
| 128 | + headers={"WWW-Authenticate": "Bearer"}, |
| 129 | + ) |
| 130 | + |
| 131 | + if not user.is_active: |
| 132 | + logger.info(f"User {user.id} is inactive.") |
| 133 | + raise HTTPException( |
| 134 | + status_code=status.HTTP_401_UNAUTHORIZED, |
| 135 | + detail="User is inactive.", |
| 136 | + headers={"WWW-Authenticate": "Bearer"}, |
| 137 | + ) |
| 138 | + |
| 139 | + return user |
| 140 | + |
| 141 | + |
| 142 | +async def clerk_token_middleware(request: Request, call_next): |
| 143 | + """Middleware to decode Clerk token for specific paths.""" |
| 144 | + settings = get_settings_service() |
| 145 | + |
| 146 | + ctx_token: Token | None = None |
| 147 | + if settings.auth_settings.CLERK_AUTH_ENABLED and request.url.path in PROTECTED_PATHS: |
| 148 | + auth_header = request.headers.get("Authorization") |
| 149 | + if auth_header and auth_header.startswith("Bearer "): |
| 150 | + token = auth_header[len("Bearer ") :] |
| 151 | + try: |
| 152 | + payload = await verify_clerk_token(token) |
| 153 | + ctx_token = auth_header_ctx.set(payload) |
| 154 | + except Exception as exc: # noqa: BLE001 |
| 155 | + logger.warning(f"Failed to verify Clerk token: {exc}") |
| 156 | + |
| 157 | + try: |
| 158 | + return await call_next(request) |
| 159 | + finally: |
| 160 | + if ctx_token is not None: |
| 161 | + auth_header_ctx.reset(ctx_token) |
| 162 | + else: |
| 163 | + auth_header_ctx.set(None) |
0 commit comments