"""Casdoor JWT authentication — the v2.6 identity source. - Verifies the Bearer token against Casdoor's JWKS (/.well-known/jwks). The expected iss claim is CASDOOR_ISSUER (defaults to CASDOOR_ENDPOINT), so the endpoint may be the in-network service URL while the issuer is the browser-facing origin Casdoor advertises. - Auto-provisions the `users` row from the token claims on first login (role stays NULL until assigned by an admin / seed). - Resolves the caller's role (users.role) and, for PMs, their single project (projects.pm_user_id). - 401 for missing/invalid/expired tokens, 403 when the user has no role yet, 503 when Casdoor itself is unreachable. - Dev fallback: when no Authorization header is present, endpoints that use `get_optional_user` degrade gracefully; the inventory router keeps its header-based placeholder mode for local testing. Dependencies (pyjwt[crypto]): pip install pyjwt[crypto] """ import logging from dataclasses import dataclass import jwt from fastapi import Depends, Request from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.config import settings from app.db.engine import get_db from app.models.procurement import Project, User from app.utils.errors import ( ForbiddenError, ServiceUnavailableError, UnauthorizedError, ) logger = logging.getLogger("procurement.auth") _JWKS_CACHE: dict | None = None def _jwks_url() -> str: return settings.CASDOOR_JWKS_URL or f"{settings.CASDOOR_ENDPOINT}/.well-known/jwks" async def _get_jwks() -> dict: """Fetch (and cache) Casdoor's public JWKS.""" global _JWKS_CACHE if _JWKS_CACHE is not None: return _JWKS_CACHE import httpx try: async with httpx.AsyncClient(timeout=10) as client: resp = await client.get(_jwks_url()) resp.raise_for_status() _JWKS_CACHE = resp.json() except httpx.HTTPError as exc: logger.error("JWKS fetch failed (%s): %s", _jwks_url(), exc) raise ServiceUnavailableError("سرویس احراز هویت (Casdoor) در دسترس نیست") return _JWKS_CACHE def _key_from_jwk(jwk: dict): """Build a crypto key object from a JWKS entry (pyjwt 2.13 no longer accepts raw JWK dicts in jwt.decode).""" kty = jwk.get("kty") if kty == "RSA": return jwt.algorithms.RSAAlgorithm.from_jwk(jwk) if kty == "EC": return jwt.algorithms.ECAlgorithm.from_jwk(jwk) return jwk async def _verify_token(token: str) -> dict: """Validate the Casdoor JWT and return its claims (401 on any failure).""" try: kid = jwt.get_unverified_header(token).get("kid") except jwt.PyJWTError as exc: logger.warning("JWT unreadable: %s", exc) raise UnauthorizedError("توکن نامعتبر است", field="authorization") jwks = await _get_jwks() entry = next((k for k in jwks.get("keys", []) if k.get("kid") == kid), None) if entry is None: logger.warning("JWT kid %r not found in JWKS", kid) raise UnauthorizedError("امضای توکن نامعتبر است", field="authorization") key = _key_from_jwk(entry) issuer = settings.CASDOOR_ISSUER or settings.CASDOOR_ENDPOINT audience = settings.CASDOOR_AUDIENCE or settings.CASDOOR_CLIENT_ID try: return jwt.decode( token, key, algorithms=["RS256", "RS384", "RS512", "ES256", "ES384", "ES512"], issuer=issuer, audience=audience, ) except jwt.PyJWTError as exc: logger.warning("JWT verification failed: %s", exc) raise UnauthorizedError("توکن نامعتبر یا منقضی شده است", field="authorization") @dataclass class UserContext: user: User role: str | None project_id: object | None # uuid.UUID | None display_name: str async def context_from_claims(claims: dict, db: AsyncSession) -> UserContext: """Auto-provision the `users` row from token claims; resolve role + PM project. First login: creates the row with role NULL (an admin assigns it later). PMs: resolves their single project via projects.pm_user_id. """ casdoor_id = str(claims.get("sub") or claims.get("id") or "") if not casdoor_id: raise UnauthorizedError("توکن فاقد شناسه کاربر است", field="authorization") user = await db.scalar(select(User).where(User.casdoor_id == casdoor_id)) if user is None: user = User( casdoor_id=casdoor_id, first_name=str( claims.get("firstName") or claims.get("firstname") or claims.get("displayName") or claims.get("name") or "کاربر" ), last_name=str(claims.get("lastName") or claims.get("lastname") or ""), phone=str(claims.get("phone") or "") or None, ) db.add(user) await db.flush() project_id = None if user.role == "pm": project = await db.scalar(select(Project).where(Project.pm_user_id == user.id)) project_id = project.id if project else None display_name = f"{user.first_name} {user.last_name}".strip() or user.casdoor_id return UserContext(user=user, role=user.role, project_id=project_id, display_name=display_name) async def get_optional_user(request: Request, db: AsyncSession) -> UserContext | None: """Resolve the caller from the Casdoor Bearer token, or None when absent. Raises UnauthorizedError for a present-but-invalid token; callers decide their own fallback only for the no-token case (e.g. dev headers). """ auth = request.headers.get("authorization", "") if not auth.lower().startswith("bearer "): return None token = auth.split(" ", 1)[1].strip() if not token: return None claims = await _verify_token(token) return await context_from_claims(claims, db) async def get_current_user(request: Request, db: AsyncSession = Depends(get_db)) -> UserContext: """Require a valid Casdoor token (401 missing/invalid; 403 unprovisioned).""" ctx = await get_optional_user(request, db) if ctx is None: raise UnauthorizedError("احراز هویت لازم است — توکن Casdoor ارسال کنید") if ctx.role is None: raise ForbiddenError("نقش شما هنوز در سیستم تعیین نشده است") return ctx def require_role(*roles: str): """Dependency factory: require the caller to hold one of the given roles.""" async def _dep(ctx: UserContext = Depends(get_current_user)) -> UserContext: if ctx.role not in roles: raise ForbiddenError("دسترسی به این بخش برای نقش شما مجاز نیست") return ctx return _dep