Authentication: JWT, Password, Role

uv add "pwdlib[argon2]" pyjwt

# core/security.py
from pwdlib import PasswordHash

password_hasher = PasswordHash.recommended()     # Argon2


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


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

>>> hash_password("rahim123")
'$argon2id$v=19$m=65536,t=3,p=4$Xc8q...$9vG1...'

// payload (base64 খুললেই পড়া যায়!)
{
  "sub": "42",              // user id
  "role": "user",
  "type": "access",
  "exp": 1793214000,        // মেয়াদ শেষের সময়
  "iat": 1793213100
}

# core/security.py (চলবে)
from datetime import datetime, timedelta, timezone
import jwt
from app.core.config import get_settings

ALGORITHM = "HS256"


def create_token(user_id: int, token_type: str, minutes: int, extra: dict | None = None) -> str:
    now = datetime.now(timezone.utc)
    payload = {
        "sub": str(user_id),
        "type": token_type,
        "iat": now,
        "exp": now + timedelta(minutes=minutes),
        **(extra or {}),
    }
    return jwt.encode(payload, get_settings().jwt_secret, algorithm=ALGORITHM)


def decode_token(token: str, expected_type: str) -> dict:
    payload = jwt.decode(token, get_settings().jwt_secret, algorithms=[ALGORITHM])  # মেয়াদ নিজে check করে
    if payload.get("type") != expected_type:
        raise jwt.InvalidTokenError("ভুল ধরনের token")
    return payload

# api/v1/auth.py
from fastapi import APIRouter
from pydantic import BaseModel, EmailStr

router = APIRouter(prefix="/auth", tags=["auth"])


class LoginIn(BaseModel):
    email: EmailStr
    password: str


class TokenPair(BaseModel):
    access_token: str
    refresh_token: str
    token_type: str = "bearer"


@router.post("/login", response_model=TokenPair)
async def login(data: LoginIn, db: DbDep):
    user = await db.scalar(select(User).where(User.email == data.email.lower()))
    # user না থাকলেও একই message আর প্রায় একই সময়: কোন email এ account আছে সেটা ফাঁস না করা
    if user is None or not verify_password(data.password, user.password_hash) or not user.is_active:
        raise UnauthorizedError("Email বা password ভুল")
    s = get_settings()
    return TokenPair(
        access_token=create_token(user.id, "access", s.jwt_access_minutes, {"role": user.role}),
        refresh_token=create_token(user.id, "refresh", s.jwt_refresh_days * 24 * 60),
    )

# api/deps.py
from typing import Annotated
from fastapi import Depends
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
import jwt

bearer = HTTPBearer(auto_error=False)


async def get_current_user(
    db: DbDep,
    creds: Annotated[HTTPAuthorizationCredentials | None, Depends(bearer)],
) -> User:
    if creds is None:
        raise UnauthorizedError("Login করতে হবে")
    try:
        payload = decode_token(creds.credentials, expected_type="access")
    except jwt.ExpiredSignatureError:
        raise UnauthorizedError("Token এর মেয়াদ শেষ", code="TOKEN_EXPIRED")
    except jwt.InvalidTokenError:
        raise UnauthorizedError("Token সঠিক না")
    user = await db.get(User, int(payload["sub"]))
    if user is None or not user.is_active:
        raise UnauthorizedError("Account পাওয়া যায়নি বা বন্ধ")
    return user


CurrentUser = Annotated[User, Depends(get_current_user)]

@router.post("/refresh", response_model=TokenPair)
async def refresh(data: RefreshIn, svc: AuthServiceDep):
    return await svc.rotate_refresh_token(data.refresh_token)

from enum import StrEnum

class Role(StrEnum):
    USER = "user"
    STAFF = "staff"
    ADMIN = "admin"


def require_role(*allowed: Role):
    async def checker(user: CurrentUser) -> User:
        if user.role not in allowed:
            raise PermissionDeniedError()
        return user
    return checker


@router.get("/admin/users", dependencies=[Depends(require_role(Role.ADMIN))])
async def list_all_users(...): ...

# ❌ ভয়ানক: login আছে কিনা দেখলো, কিন্তু note কার সেটা দেখলো না
@router.get("/notes/{note_id}")
async def get_note(note_id: int, user: CurrentUser, db: DbDep):
    return await db.get(Note, note_id)      # যে কেউ ১, ২, ৩... দিয়ে সবার note পড়তে পারবে

# ✓ service এ মালিকানা যাচাই (আগের অধ্যায়ের get_owned)