FastAPI Authentication — JWT and OAuth2 with Password Flow

import os
from datetime import datetime, timedelta, timezone
from typing import Annotated
from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from passlib.context import CryptContext

# Never hardcode this value — load from secrets manager at runtime
# Generate a safe key with: python3 -c "import secrets; print(secrets.token_hex(32))"
SECRET_KEY = os.getenv("SECRET_KEY", "")
if not SECRET_KEY:
    raise RuntimeError("SECRET_KEY environment variable is not set — refusing to start")

# HS256 for single-service; RS256 for distributed systems where multiple services verify tokens
ALGORITHM = os.getenv("JWT_ALGORITHM", "HS256")

# Keep access tokens short — 15-30 minutes limits the damage window if a token is stolen
ACCESS_TOKEN_EXPIRE_MINUTES = int(os.getenv("ACCESS_TOKEN_EXPIRE_MINUTES", "30"))

# bcrypt with deprecated=auto handles algorithm migrations gracefully
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

# tokenUrl tells Swagger UI where to send login requests for the Authorize popup
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

app = FastAPI()

from datetime import datetime, timedelta, timezone
from typing import Annotated
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from jose import jwt

def create_access_token(data: dict, expires_delta: timedelta | None = None) -> str:
    """
    Create a signed JWT access token.
    Always pass expires_delta explicitly in production — the 15-minute
    fallback exists for defensive safety, not as a default you should rely on.
    """
    to_encode = data.copy()
    expire = datetime.now(timezone.utc) + (expires_delta or timedelta(minutes=15))
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)


@app.post("/token")
async def login_for_access_token(
    form_data: Annotated[OAuth2PasswordRequestForm, Depends()]
):
    """
    Exchange credentials for a JWT access token.
    In production, replace the hardcoded check with a database lookup
    and use pwd_context.verify() to compare the submitted password
    against the stored bcrypt hash — never compare plaintext passwords.
    """
    # Replace this with: user = await get_user_from_db(form_data.username)
    # Then: if not user or not pwd_context.verify(form_data.password, user.hashed_password)
    if form_data.username != "forge_dev" or form_data.password != "secret_pass":
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )

    access_token = create_access_token(
        data={"sub": form_data.username},
        expires_delta=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES),
    )
    return {"access_token": access_token, "token_type": "bearer"}

from typing import Annotated
from fastapi import Depends, HTTPException, status
from jose import JWTError, jwt

async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]):
    """
    Core authentication dependency. Inject this into any route that requires
    a valid, non-expired JWT.

    Returns the user dict on success. Raises 401 on any validation failure.
    The error message is intentionally generic — specific failure reasons
    help attackers probe your validation logic.
    """
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str | None = payload.get("sub")
        if username is None:
            raise credentials_exception
    except JWTError:
        # Catches ExpiredSignatureError, DecodeError, InvalidSignatureError
        # All map to the same 401 response — do not differentiate for clients
        raise credentials_exception

    # In production, fetch the full user from the database here
    # user = await db.users.find_one({"username": username})
    # if user is None: raise credentials_exception
    return {"username": username, "active": True}


async def require_admin(
    current_user: Annotated[dict, Depends(get_current_user)]
):
    """
    Role-based access dependency — chains on get_current_user.
    Use this for admin-only endpoints instead of duplicating the role check.
    """
    if current_user.get("role") != "admin":
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Insufficient privileges",
        )
    return current_user


@app.get("/users/me")
async def read_users_me(
    current_user: Annotated[dict, Depends(get_current_user)]
):
    return current_user


@app.get("/admin/dashboard")
async def admin_dashboard(
    current_user: Annotated[dict, Depends(require_admin)]
):
    return {"message": "Welcome to the admin dashboard", "user": current_user}

import uuid
from datetime import datetime, timedelta, timezone
from typing import Annotated
from fastapi import Depends, HTTPException, status
from jose import JWTError, jwt
import redis.asyncio as redis

redis_client = redis.from_url(
    os.getenv("REDIS_URL", "redis://localhost:6379"),
    decode_responses=True
)


def create_access_token_with_jti(
    data: dict,
    expires_delta: timedelta | None = None
) -> tuple[str, str]:
    """
    Issues a JWT with a unique JTI claim for individual revocation.
    Returns (encoded_token, jti) — store the JTI if you need to revoke later.
    """
    to_encode = data.copy()
    jti = str(uuid.uuid4())
    expire = datetime.now(timezone.utc) + (expires_delta or timedelta(minutes=30))
    to_encode.update({"jti": jti, "exp": expire})
    token = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return token, jti


async def revoke_token(jti: str, ttl_seconds: int) -> None:
    """
    Add a token's JTI to the Redis blacklist.
    TTL should match the token's remaining lifetime — not the full token duration.
    This keeps the blacklist bounded without a separate cleanup job.
    """
    await redis_client.setex(f"blacklist:{jti}", ttl_seconds, "revoked")


async def is_token_revoked(jti: str) -> bool:
    """Returns True if the JTI is in the blacklist."""
    return await redis_client.exists(f"blacklist:{jti}") == 1


async def get_current_user_with_blacklist(
    token: Annotated[str, Depends(oauth2_scheme)]
):
    """
    Authentication dependency with JTI blacklist check.
    Drop-in replacement for get_current_user when revocation is required.
    Adds ~1ms Redis GET per request — acceptable cost for immediate revocability.
    """
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        jti: str | None = payload.get("jti")
        username: str | None = payload.get("sub")
        if username is None or jti is None:
            raise credentials_exception
        if await is_token_revoked(jti):
            raise credentials_exception
    except JWTError:
        raise credentials_exception
    return {"username": username, "jti": jti}

import secrets
from datetime import datetime, timedelta, timezone
from fastapi import HTTPException, status
from jose import jwt

REFRESH_TOKEN_EXPIRE_DAYS = int(os.getenv("REFRESH_TOKEN_EXPIRE_DAYS", "7"))

# In production, replace with async database operations
# Store hashed refresh tokens: hashlib.sha256(token.encode()).hexdigest()
# Never store the raw token value — treat it like a password
refresh_token_store: dict[str, dict] = {}


def create_refresh_token(user_id: str) -> str:
    """
    Generate a cryptographically secure opaque refresh token.
    secrets.token_urlsafe(64) gives 384 bits of entropy — sufficient for
    any brute-force resistance requirement.
    """
    token = secrets.token_urlsafe(64)
    refresh_token_store[token] = {
        "user_id": user_id,
        "expires_at": datetime.now(timezone.utc) + timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS),
        "used": False,
    }
    return token


async def refresh_access_token(refresh_token: str) -> dict:
    """
    Exchange a valid refresh token for a new access + refresh token pair.
    Rotation: the submitted refresh token is consumed and a new one issued.
    Reuse detection: if a consumed token appears again, revoke everything
    for that user — this means a stolen token was used after legitimate rotation.
    """
    record = refresh_token_store.get(refresh_token)

    if not record:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid refresh token",
        )

    if record["used"]:
        # A consumed token reappeared — this is a theft indicator.
        # Revoke all sessions for this user and force re-authentication.
        # In production: invalidate all refresh tokens for record["user_id"]
        # and increment the user's token_version in the database.
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Refresh token reuse detected — all sessions have been revoked",
        )

    if datetime.now(timezone.utc) > record["expires_at"]:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Refresh token expired",
        )

    # Mark as consumed before issuing the new pair
    # In production, this must be atomic — use a database transaction
    # or Redis SET NX to prevent race conditions on concurrent refresh requests
    record["used"] = True

    new_access_token = create_access_token(
        data={"sub": record["user_id"]},
        expires_delta=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES),
    )
    new_refresh_token = create_refresh_token(record["user_id"])

    return {
        "access_token": new_access_token,
        "refresh_token": new_refresh_token,
        "token_type": "bearer",
    }