""" Emisión y verificación de JWT. Endurecimiento aplicado: - Sin fallback de secreto débil: si la configuración no es válida, falla al arranque. - `iss` y `aud` obligatorios. - `nbf` (not-before) y `iat` validados. - `typ=access` para evitar mezclar tipos de token. - Algoritmo fijo (no se acepta "none" ni cambios por payload). - Errores opacos: nunca se expone el motivo del fallo de verificación al cliente. """ from __future__ import annotations from datetime import datetime, timedelta, timezone from typing import Any from uuid import uuid4 from fastapi import Depends, HTTPException, Request, status from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer from jose import JWTError, jwt from sqlalchemy.orm import Session from app.core.config import settings from app.db.models import Researcher from app.db.session import get_db from app.utils.orcid_validator import is_valid_orcid _bearer = HTTPBearer(auto_error=False) def create_access_token(*, subject: str, extra: dict[str, Any] | None = None) -> str: """ Emite un access token firmado con HS256 (configurable). `subject` debe ser el ORCID iD verificado del investigador. """ if not is_valid_orcid(subject): raise ValueError("subject must be a valid ORCID iD") now = datetime.now(timezone.utc) payload: dict[str, Any] = { "iss": settings.JWT_ISSUER, "aud": settings.JWT_AUDIENCE, "sub": subject, "iat": int(now.timestamp()), "nbf": int(now.timestamp()), "exp": int((now + timedelta(minutes=settings.JWT_EXPIRES_MINUTES)).timestamp()), "jti": uuid4().hex, "typ": "access", } if extra: for reserved in ("iss", "aud", "sub", "iat", "nbf", "exp", "jti", "typ"): extra.pop(reserved, None) payload.update(extra) return jwt.encode(payload, settings.JWT_SECRET, algorithm=settings.JWT_ALGORITHM) def _decode_token(token: str) -> dict[str, Any]: try: return jwt.decode( token, settings.JWT_SECRET, algorithms=[settings.JWT_ALGORITHM], audience=settings.JWT_AUDIENCE, issuer=settings.JWT_ISSUER, options={ "require_iat": True, "require_nbf": True, "require_exp": True, "require_aud": True, "require_iss": True, }, ) except JWTError as exc: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid or expired token", headers={"WWW-Authenticate": "Bearer"}, ) from exc def get_current_researcher( request: Request, creds: HTTPAuthorizationCredentials | None = Depends(_bearer), db: Session = Depends(get_db), ) -> Researcher: if not creds or not creds.credentials: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Missing bearer token", headers={"WWW-Authenticate": "Bearer"}, ) payload = _decode_token(creds.credentials) if payload.get("typ") != "access": raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token type", headers={"WWW-Authenticate": "Bearer"}, ) orcid_id = payload.get("sub") if not isinstance(orcid_id, str) or not is_valid_orcid(orcid_id): raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token subject", headers={"WWW-Authenticate": "Bearer"}, ) researcher = db.query(Researcher).filter(Researcher.orcid_id == orcid_id).first() if not researcher or not researcher.authenticated: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Researcher not authenticated", headers={"WWW-Authenticate": "Bearer"}, ) request.state.researcher = researcher return researcher def get_optional_current_researcher( request: Request, creds: HTTPAuthorizationCredentials | None = Depends(_bearer), db: Session = Depends(get_db), ) -> Researcher | None: """ Devuelve el investigador autenticado si hay Bearer válido. Si no hay Bearer, devuelve None. Si hay Bearer inválido, lanza 401 (no se acepta como anónimo). """ if not creds or not creds.credentials: return None return get_current_researcher(request=request, creds=creds, db=db)