fix(auth): Casdoor JWT verification and 401 semantics; add /auth/login password-grant endpoint

- verify kid from the JWT header (pyjwt decode() returns claims, not the
  header — the old lookup was always None and rejected every token)
- convert JWK entries to crypto keys explicitly (pyjwt 2.13 dropped dict
  keys) and fetch the JWKS from /.well-known/jwks
- 401 for missing/invalid/expired/wrong-issuer/wrong-audience tokens
  (was 403/400); 403 stays for valid-but-roleless users; 503 when Casdoor
  is unreachable; new CASDOOR_ISSUER setting separates the in-network
  endpoint from the browser-facing issuer
- dev-header fallback in inventory/pm scopes now requires an explicit
  X-User-Role header (no auth info at all -> 401)
- new POST /api/v1/auth/login: password grant against Casdoor, local JWT
  verification, auto-provisioning, returns token + profile with the role
  resolved from the app users table

backend/app/auth.py

@@ -1,10 +1,15 @@
 """Casdoor JWT authentication — the v2.6 identity source.
 
-- Verifies the Bearer token against Casdoor's JWKS (issuer = CASDOOR_ENDPOINT).
+- 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.
@@ -23,7 +28,11 @@ 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, ValidationError
+from app.utils.errors import (
+    ForbiddenError,
+    ServiceUnavailableError,
+    UnauthorizedError,
+)
 
 logger = logging.getLogger("procurement.auth")
 
@@ -31,7 +40,7 @@ _JWKS_CACHE: dict | None = None
 
 
 def _jwks_url() -> str:
-    return settings.CASDOOR_JWKS_URL or f"{settings.CASDOOR_ENDPOINT}/api/public_jwks"
+    return settings.CASDOOR_JWKS_URL or f"{settings.CASDOOR_ENDPOINT}/.well-known/jwks"
 
 
 async def _get_jwks() -> dict:
@@ -40,34 +49,54 @@ async def _get_jwks() -> dict:
     if _JWKS_CACHE is not None:
         return _JWKS_CACHE
     import httpx
-    async with httpx.AsyncClient(timeout=10) as client:
+    try:
-        resp = await client.get(_jwks_url())
+        async with httpx.AsyncClient(timeout=10) as client:
-        resp.raise_for_status()
+            resp = await client.get(_jwks_url())
-        _JWKS_CACHE = resp.json()
+            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."""
+    """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()
-    unverified = jwt.decode(token, options={"verify_signature": False})
+    entry = next((k for k in jwks.get("keys", []) if k.get("kid") == kid), None)
-    kid = unverified.get("kid")
+    if entry is None:
-    key = next((k for k in jwks.get("keys", []) if k.get("kid") == kid), None)
+        logger.warning("JWT kid %r not found in JWKS", kid)
-    if key is None:
+        raise UnauthorizedError("امضای توکن نامعتبر است", field="authorization")
-        raise ValidationError("امضای توکن نامعتبر است", field="authorization")
+    key = _key_from_jwk(entry)
-    issuer = settings.CASDOOR_ENDPOINT
+    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"],
+            algorithms=["RS256", "RS384", "RS512", "ES256", "ES384", "ES512"],
             issuer=issuer,
             audience=audience,
         )
     except jwt.PyJWTError as exc:
         logger.warning("JWT verification failed: %s", exc)
-        raise ValidationError("توکن نامعتبر یا منقضی شده است", field="authorization")
+        raise UnauthorizedError("توکن نامعتبر یا منقضی شده است", field="authorization")
 
 
 @dataclass
@@ -78,31 +107,28 @@ class UserContext:
     display_name: str
 
 
-async def get_optional_user(request: Request, db: AsyncSession) -> UserContext | None:
+async def context_from_claims(claims: dict, db: AsyncSession) -> UserContext:
-    """Resolve the caller from the Casdoor Bearer token, or None when absent/invalid.
+    """Auto-provision the `users` row from token claims; resolve role + PM project.
 
-    Never raises on a missing token — callers decide their own fallback
+    First login: creates the row with role NULL (an admin assigns it later).
-    (e.g. the inventory router's dev headers).
+    PMs: resolves their single project via projects.pm_user_id.
     """
-    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)
-
     casdoor_id = str(claims.get("sub") or claims.get("id") or "")
     if not casdoor_id:
-        raise ValidationError("توکن فاقد شناسه کاربر است", field="authorization")
+        raise UnauthorizedError("توکن فاقد شناسه کاربر است", field="authorization")
 
-    # Auto-provision on first login (role stays NULL until assigned).
     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("name") or "کاربر"),
+            first_name=str(
-            last_name=str(claims.get("lastname") or ""),
+                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)
@@ -117,11 +143,27 @@ async def get_optional_user(request: Request, db: AsyncSession) -> UserContext |
     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 (403 when missing or unprovisioned)."""
+    """Require a valid Casdoor token (401 missing/invalid; 403 unprovisioned)."""
     ctx = await get_optional_user(request, db)
     if ctx is None:
-        raise ForbiddenError("احراز هویت لازم است — توکن Casdoor ارسال کنید")
+        raise UnauthorizedError("احراز هویت لازم است — توکن Casdoor ارسال کنید")
     if ctx.role is None:
         raise ForbiddenError("نقش شما هنوز در سیستم تعیین نشده است")
     return ctx

backend/app/config.py

@@ -41,6 +41,7 @@ class Settings(BaseSettings):
 
     # Casdoor (OIDC) — JWT verification against the issuer's JWKS.
     CASDOOR_ENDPOINT: str = "http://localhost:8080"
+    CASDOOR_ISSUER: str = ""  # expected token iss claim (default: CASDOOR_ENDPOINT)
     CASDOOR_CLIENT_ID: str = "inventory-app-client"
     CASDOOR_CLIENT_SECRET: str = "inventory-app-secret-2026"
     CASDOOR_ORGANIZATION: str = "inventory"

backend/app/routers/__init__.py

@@ -1,6 +1,7 @@
 """All API routers, re-exported for main.py assembly."""
 from app.routers import (
     attribute_definitions,
+    auth,
     brand_materials,
     brands,
     category_attributes,
@@ -29,6 +30,7 @@ from app.routers import (
 )
 
 ROUTERS = [
+    auth.router,
     units.router,
     brands.router,
     material_categories.router,

backend/app/routers/auth.py

@@ -0,0 +1,86 @@
+"""Auth router — in-page login against Casdoor (password grant).
+
+POST /auth/login: the frontend sends username/password; this router exchanges
+them at Casdoor's OAuth token endpoint (grant_type=password), verifies the
+issued JWT locally (JWKS), auto-provisions the `users` row and returns the
+token + profile (role resolved from users.role). No browser redirect.
+
+Error mapping (documented in docs/casdoor-auth.md):
+- wrong credentials           -> 401 UNAUTHORIZED
+- Casdoor unreachable         -> 503 UPSTREAM_UNAVAILABLE
+- Casdoor misconfigured       -> 503 UPSTREAM_UNAVAILABLE (details in api logs)
+"""
+import logging
+
+import httpx
+from fastapi import APIRouter, Depends
+from pydantic import BaseModel
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.auth import _verify_token, context_from_claims
+from app.config import settings
+from app.db.engine import get_db
+from app.routers.users import _user_project, user_to_dict
+from app.utils.errors import ServiceUnavailableError, UnauthorizedError
+
+logger = logging.getLogger("procurement.auth")
+
+router = APIRouter(prefix="/auth", tags=["auth"])
+
+
+class LoginRequest(BaseModel):
+    username: str
+    password: str
+    organization: str | None = None  # optional override; defaults to CASDOOR_ORGANIZATION
+
+
+@router.post("/login")
+async def login(payload: LoginRequest, db: AsyncSession = Depends(get_db)):
+    """Exchange username/password for a Casdoor JWT (OAuth password grant)."""
+    url = f"{settings.CASDOOR_ENDPOINT}/api/login/oauth/access_token"
+    form = {
+        "grant_type": "password",
+        "client_id": settings.CASDOOR_CLIENT_ID,
+        "client_secret": settings.CASDOOR_CLIENT_SECRET,
+        "username": payload.username.strip(),
+        "password": payload.password,
+        "scope": "openid profile email",
+    }
+    try:
+        async with httpx.AsyncClient(timeout=10) as client:
+            resp = await client.post(url, data=form)
+    except httpx.HTTPError as exc:
+        logger.error("Casdoor token endpoint unreachable (%s): %s", url, exc)
+        raise ServiceUnavailableError("سرویس احراز هویت (Casdoor) در دسترس نیست") from exc
+
+    try:
+        body = resp.json()
+    except ValueError as exc:
+        logger.error("Casdoor token endpoint returned non-JSON (status %s)", resp.status_code)
+        raise ServiceUnavailableError("پاسخ نامعتبر از سرویس احراز هویت") from exc
+
+    access_token = body.get("access_token")
+    if not access_token:
+        err = body.get("error") or "unknown_error"
+        if err in ("invalid_username_or_password", "invalid_grant"):
+            raise UnauthorizedError("نام کاربری یا رمز عبور نادرست است", field="password")
+        logger.error(
+            "Casdoor rejected password grant: error=%s description=%s",
+            err,
+            body.get("error_description", ""),
+        )
+        raise ServiceUnavailableError("سرویس احراز هویت به درستی پیکربندی نشده است")
+
+    # Verify the freshly issued token with our own JWKS checks, then resolve
+    # the app-side user (auto-provision on first login).
+    claims = await _verify_token(access_token)
+    ctx = await context_from_claims(claims, db)
+    project = await _user_project(db, ctx.user) if ctx.role == "pm" else None
+    return {
+        "data": {
+            "access_token": access_token,
+            "token_type": "bearer",
+            "expires_in": body.get("expires_in"),
+            "user": user_to_dict(ctx.user, project),
+        }
+    }

backend/app/routers/inventory.py

@@ -36,7 +36,12 @@ from app.services.inventory_service import (
     apply_transaction,
     get_warehouse_or_404,
 )
-from app.utils.errors import ForbiddenError, NotFoundError, ValidationError
+from app.utils.errors import (
+    ForbiddenError,
+    NotFoundError,
+    UnauthorizedError,
+    ValidationError,
+)
 from app.utils.pagination import paginate, sort_query
 from app.utils.serialization import to_iso
 
@@ -59,7 +64,9 @@ async def inventory_scope(request: Request, db: AsyncSession = Depends(get_db))
     # projects.pm_user_id (single PM per project).
     if request.headers.get("authorization"):
         user: UserContext = await get_optional_user(request, db)
-        if user is not None and user.role is not None:
+        if user is not None:
+            if user.role is None:
+                raise ForbiddenError("نقش شما هنوز در سیستم تعیین نشده است")
             # v2.6: مدیر تدارکات = ادمین دفتر — دسترسی فقط‌خواندنی به پنل انبار.
             if user.role == "procurement_manager":
                 return Scope(role="admin", project_id=None, actor=user.display_name)
@@ -73,7 +80,10 @@ async def inventory_scope(request: Request, db: AsyncSession = Depends(get_db))
                 return Scope(role="pm", project_id=user.project_id, actor=user.display_name)
             return Scope(role="admin", project_id=None, actor=user.display_name)
 
-    # Dev fallback: header-based identity (fork's placeholder mode).
+    # Dev fallback: header-based identity (fork's placeholder mode) — only
+    # when an explicit X-User-Role header is sent. No auth info at all → 401.
+    if not request.headers.get("x-user-role"):
+        raise UnauthorizedError("احراز هویت لازم است — توکن Casdoor یا حالت توسعه را ارسال کنید")
     role = (request.headers.get("x-user-role") or settings.INVENTORY_DEFAULT_ROLE or ROLE_PM).strip().lower()
     if role == "procurement_manager":
         role = ROLE_ADMIN  # مدیر تدارکات = ادمین (فقط‌خواندنی)

backend/app/routers/pm.py

@@ -47,7 +47,12 @@ from app.models.procurement import (
 )
 from app.schemas.procurement import PMReceiptCreate
 from app.services.inventory_service import canonical_specs_hash
-from app.utils.errors import ForbiddenError, NotFoundError, ValidationError
+from app.utils.errors import (
+    ForbiddenError,
+    NotFoundError,
+    UnauthorizedError,
+    ValidationError,
+)
 from app.utils.serialization import to_iso
 
 router = APIRouter(prefix="/pm", tags=["pm"])
@@ -71,6 +76,8 @@ async def arrivals_scope(request: Request, db: AsyncSession = Depends(get_db)) -
     if request.headers.get("authorization"):
         ctx: UserContext | None = await get_optional_user(request, db)
         if ctx is not None:
+            if ctx.role is None:
+                raise ForbiddenError("نقش شما هنوز در سیستم تعیین نشده است")
             if ctx.role == "pm":
                 if ctx.project_id is None:
                     raise ForbiddenError("شما به هیچ پروژه‌ای متصل نیستید (projects.pm_user_id)")
@@ -79,6 +86,8 @@ async def arrivals_scope(request: Request, db: AsyncSession = Depends(get_db)) -
                 return ArrivalScope(role="admin", user_id=ctx.user.id, project_id=None, display_name=ctx.display_name)
             raise ForbiddenError("این بخش فقط برای مدیر پروژه یا مدیر تدارکات است")
     # Dev fallback: X-User-Role: pm | admin (procurement_manager maps to admin)
+    if not request.headers.get("x-user-role"):
+        raise UnauthorizedError("احراز هویت لازم است — توکن Casdoor یا حالت توسعه را ارسال کنید")
     role = (request.headers.get("x-user-role") or "").strip().lower()
     if role == "procurement_manager":
         role = "admin"

backend/app/utils/errors.py

@@ -49,3 +49,17 @@ class ForbiddenError(AppError):
 
     def __init__(self, message: str, field: str | None = None):
         super().__init__(403, "FORBIDDEN", message, field)
+
+
+class UnauthorizedError(AppError):
+    """401 UNAUTHORIZED — missing, invalid, expired or untrusted credentials."""
+
+    def __init__(self, message: str, field: str | None = None):
+        super().__init__(401, "UNAUTHORIZED", message, field)
+
+
+class ServiceUnavailableError(AppError):
+    """503 UPSTREAM_UNAVAILABLE — a dependent service (Casdoor/MinIO) is down."""
+
+    def __init__(self, message: str, field: str | None = None):
+        super().__init__(503, "UPSTREAM_UNAVAILABLE", message, field)