Authentication and authorization

# pip install "pwdlib[argon2]" pyjwt python-multipart
import os
from datetime import datetime, timedelta, timezone
import jwt
from pwdlib import PasswordHash
from fastapi import Depends, FastAPI, HTTPException
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm

SECRET = os.environ["JWT_SECRET"]  # long random secret
ALGORITHM = "HS256"
password_hash = PasswordHash.recommended()
app = FastAPI()
oauth2 = OAuth2PasswordBearer(tokenUrl="token")

# Illustrative store. Persist users and password hashes in a DB.
users = {"ada": {"id": 1, "username": "ada",
                 "password_hash": password_hash.hash("replace-in-real-app"),
                 "is_active": True}}

@app.post("/token")
def login(form: OAuth2PasswordRequestForm = Depends()):
    user = users.get(form.username)
    if not user or not password_hash.verify(
        form.password, user["password_hash"]
    ):
        raise HTTPException(401, "Invalid credentials",
                            headers={"WWW-Authenticate": "Bearer"})
    now = datetime.now(timezone.utc)
    token = jwt.encode(
        {"sub": str(user["id"]), "iat": now,
         "exp": now + timedelta(minutes=15)},
        SECRET, algorithm=ALGORITHM,
    )
    return {"access_token": token, "token_type": "bearer"}

from typing import Annotated
from fastapi import Depends, HTTPException
from jwt import InvalidTokenError

# Continue the same module as the login example.
def current_user(token: Annotated[str, Depends(oauth2)]):
    credentials_error = HTTPException(
        401, "Invalid credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        claims = jwt.decode(token, SECRET, algorithms=[ALGORITHM])
        user_id = int(claims["sub"])
    except (InvalidTokenError, KeyError, ValueError):
        raise credentials_error
    user = next((u for u in users.values() if u["id"] == user_id), None)
    if user is None or not user["is_active"]:
        raise credentials_error
    return user

CurrentUser = Annotated[dict, Depends(current_user)]

@app.get("/me")
def me(user: CurrentUser):
    return {"id": user["id"], "username": user["username"]}

from fastapi import Depends, FastAPI, HTTPException
from typing import Annotated

# current_user comes from the authentication section.
def require_editor(user: CurrentUser):
    if user.get("role") not in {"editor", "admin"}:
        raise HTTPException(403, "Editor role required")
    return user

Editor = Annotated[dict, Depends(require_editor)]

@app.get("/articles/{article_id}")
def article(article_id: int, user: CurrentUser, db: Db):
    item = db.get(Article, article_id)
    if item is None:
        raise HTTPException(404, "Article not found")
    if item.owner_id != user["id"] and user.get("role") != "admin":
        raise HTTPException(403, "Not allowed")
    return {"id": item.id, "title": item.title}

from typing import Annotated
from fastapi import Depends, HTTPException, Security
from fastapi.security import OAuth2PasswordBearer, SecurityScopes

scheme = OAuth2PasswordBearer(
    tokenUrl="token", scopes={"articles:write": "Edit articles"}
)

def user_with_scopes(
    scopes: SecurityScopes,
    token: Annotated[str, Depends(scheme)],
):
    # Decode and validate token as above, then obtain user's
    # effective scopes from trusted DB/session data.
    effective = {"articles:write"}  # replace with actual lookup
    if not set(scopes.scopes).issubset(effective):
        raise HTTPException(403, "Insufficient scope")
    return {"id": 1}

@app.post("/articles", dependencies=[
    Security(user_with_scopes, scopes=["articles:write"])
])
def create_article():
    return {"ok": True}