Authentication

# pip install python-jose[cryptography] passlib[bcrypt]
from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt
from datetime import datetime, timedelta, timezone

SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

app = FastAPI()
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/token")

def create_access_token(data: dict, expires_delta: timedelta | None = None) -> str:
    to_encode = data.copy()
    expire = datetime.now(timezone.utc) + (expires_delta or timedelta(minutes=15))
    to_encode["exp"] = expire
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

def get_current_user(token: str = Depends(oauth2_scheme)):
    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 = payload.get("sub")
        if username is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception
    user = db.get_user(username)
    if user is None:
        raise credentials_exception
    return user

@app.get("/me")
def me(current_user = Depends(get_current_user)):
    return current_user

from fastapi.security import OAuth2PasswordRequestForm
from passlib.context import CryptContext
from pydantic import BaseModel

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

class Token(BaseModel):
    access_token: str
    token_type: str

def verify_password(plain: str, hashed: str) -> bool:
    return pwd_context.verify(plain, hashed)

def hash_password(plain: str) -> str:
    return pwd_context.hash(plain)

@app.post("/auth/token", response_model=Token)
def login(form: OAuth2PasswordRequestForm = Depends()):
    user = db.get_user(form.username)
    if not user or not verify_password(form.password, user.hashed_password):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    token = create_access_token(
        data={"sub": user.username},
        expires_delta=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES),
    )
    return Token(access_token=token, token_type="bearer")

from fastapi.security import HTTPBasic, HTTPBasicCredentials
import secrets

security = HTTPBasic()

def verify_basic(credentials: HTTPBasicCredentials = Depends(security)):
    correct_user = secrets.compare_digest(credentials.username, "admin")
    correct_pass = secrets.compare_digest(credentials.password, "secret")
    if not (correct_user and correct_pass):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect credentials",
            headers={"WWW-Authenticate": "Basic"},
        )
    return credentials.username

@app.get("/admin")
def admin_panel(user: str = Depends(verify_basic)):
    return {"user": user}

from fastapi.security import APIKeyHeader

api_key_header = APIKeyHeader(name="X-API-Key", auto_error=True)

def require_api_key(api_key: str = Depends(api_key_header)):
    if api_key != settings.api_key:
        raise HTTPException(status_code=403, detail="Invalid API key")
    return api_key

@app.get("/data", dependencies=[Depends(require_api_key)])
def data():
    return []

from fastapi.security import APIKeyQuery

api_key_query = APIKeyQuery(name="api_key", auto_error=False)

def require_api_key(api_key: str | None = Depends(api_key_query)):
    if api_key != settings.api_key:
        raise HTTPException(status_code=403, detail="Invalid API key")
    return api_key

from fastapi.security import APIKeyCookie

api_key_cookie = APIKeyCookie(name="session", auto_error=True)

def require_session(session: str = Depends(api_key_cookie)):
    user = db.get_user_by_session(session)
    if not user:
        raise HTTPException(status_code=401, detail="Not authenticated")
    return user

from enum import Enum

class Role(str, Enum):
    user = "user"
    admin = "admin"

def require_role(*roles: Role):
    def dependency(current_user = Depends(get_current_user)):
        if current_user.role not in roles:
            raise HTTPException(status_code=403, detail="Not enough permissions")
        return current_user
    return dependency

@app.delete("/users/{id}", dependencies=[Depends(require_role(Role.admin))])
def delete_user(id: int):
    db.delete(id)

REFRESH_TOKEN_EXPIRE_DAYS = 7

def create_refresh_token(user_id: int) -> str:
    return create_access_token(
        data={"sub": str(user_id), "type": "refresh"},
        expires_delta=timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS),
    )

@app.post("/auth/refresh", response_model=Token)
def refresh(refresh_token: str = Body(...)):
    try:
        payload = jwt.decode(refresh_token, SECRET_KEY, algorithms=[ALGORITHM])
        if payload.get("type") != "refresh":
            raise HTTPException(401, "Invalid token type")
        user_id = int(payload["sub"])
    except JWTError:
        raise HTTPException(401, "Invalid refresh token")
    user = db.get(user_id)
    return Token(
        access_token=create_access_token({"sub": str(user.id)}),
        token_type="bearer",
    )

from fastapi import Request, Response, Cookie

@app.post("/auth/cookie-login")
def cookie_login(response: Response, form: OAuth2PasswordRequestForm = Depends()):
    user = authenticate(form.username, form.password)
    token = create_access_token({"sub": user.username})
    response.set_cookie(
        key="access_token",
        value=f"Bearer {token}",
        httponly=True,
        secure=True,
        samesite="lax",
        max_age=1800,
    )
    return {"status": "logged in"}

def get_current_user_cookie(access_token: str = Cookie(default=None)):
    if not access_token or not access_token.startswith("Bearer "):
        raise HTTPException(401, "Not authenticated")
    token = access_token.removeprefix("Bearer ")
    return decode_token(token)

# Option 1: return value used
@app.get("/profile")
def profile(user = Depends(get_current_user)):
    return user

# Option 2: only the side effect matters (auth check)
@app.delete("/items/{id}", dependencies=[Depends(get_current_user)])
def delete_item(id: int):
    db.delete(id)

# Option 3: entire router protected
router = APIRouter(dependencies=[Depends(get_current_user)])

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/token", auto_error=False)

def get_optional_user(token: str | None = Depends(oauth2_scheme)):
    if token is None:
        return None          # anonymous
    return decode_token(token)

@app.get("/content")
def content(user = Depends(get_optional_user)):
    if user:
        return {"full": True}
    return {"preview": True}