JWT، کاربر جاری و نقش‌ها

۱. JWT: کارت شناسایی امضاشده‌ای که کلاینت نگه می‌دارد

یک JWT سه بخش base64 است که با نقطه از هم جدا شده‌اند:

header . payload . signature
{"alg":"HS256"} . {"sub":"7","role":"student","exp":...} . <HMAC of both>

• payload فقط کدگذاری شده است، رمزگذاری نشده: هر کسی می‌تواند آن را بخواند. هرگز رمز یا راز در آن نگذارید.
• امضا با کلید مخفی سرور ساخته می‌شود. یک حرف payload را عوض کنید و امضا دیگر جور نیست؛ سرور توکن را رد می‌کند.
• پس سرور لازم نیست توکن‌ها را ذخیره کند: امضا را بررسی می‌کند و به ادعاهای درون توکن اعتماد می‌کند.

def main() -> None:
    token = jwt.encode({'sub': '7', 'role': 'student'}, SECRET_KEY, algorithm=ALGORITHM)
    print(token)

    header, payload, signature = token.split('.')
    padded = payload + '=' * (-len(payload) % 4)
    print('anyone can read the payload:', json.loads(base64.urlsafe_b64decode(padded)))

    print('decoded by the server:', jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]))

    # A student makes themself admin by editing the payload...
    fake_payload = base64.urlsafe_b64encode(b'{"sub":"7","role":"admin"}').decode().rstrip('=')
    forged = f'{header}.{fake_payload}.{signature}'
    try:
        jwt.decode(forged, SECRET_KEY, algorithms=[ALGORITHM])
    except jwt.InvalidSignatureError as error:
        print('forged token rejected:', error)

    try:
        jwt.decode(token, 'another-server-with-another-secret-key-0000000', algorithms=[ALGORITHM])
    except jwt.InvalidSignatureError as error:
        print('wrong key rejected   :', error)

    # `algorithms=[...]` is a white list: never let the token choose ("alg": "none" attack)
    unsigned = jwt.encode({'sub': '1', 'role': 'admin'}, key=None, algorithm='none')
    try:
        jwt.decode(unsigned, SECRET_KEY, algorithms=[ALGORITHM])
    except jwt.InvalidAlgorithmError as error:
        print('alg=none rejected    :', error)

خواندن، جعل و رد شدن: نمونه 01_jwt_encode_decode/app.py

anyone can read the payload: {'sub': '7', 'role': 'student'}
forged token rejected: Signature verification failed
wrong key rejected   : Signature verification failed
alg=none rejected    : The specified alg value is not allowed

۲. ادعاها و انقضا

ادعاهای استاندارد: sub یعنی چه کسی (یک رشته، طبق قاعده JWT)، exp زمان انقضا و iat زمان صدور. به‌علاوه ادعای خود ما: role.

def create_token(user_id: int, minutes: float) -> str:
    now = datetime.now(UTC)
    claims = {'sub': str(user_id), 'iat': now, 'exp': now + timedelta(minutes=minutes)}
    return jwt.encode(claims, SECRET_KEY, algorithm='HS256')


def check(token: str, leeway: int = 0) -> str:
    try:
        claims = jwt.decode(
            token,
            SECRET_KEY,
            algorithms=['HS256'],
            options={'require': ['exp', 'iat', 'sub']},
            leeway=leeway,
        )
    except jwt.ExpiredSignatureError:
        return 'EXPIRED -> 401, please log in again'
    except jwt.MissingRequiredClaimError as error:
        return f'INVALID -> {error}'
    expires = datetime.fromtimestamp(claims['exp'], UTC)
    return f'valid for user {claims["sub"]}, expires {expires:%H:%M:%S}'

انقضا و ادعاهای الزامی: نمونه 02_token_expiry/app.py

توکن دزدیده‌شده تا زمان انقضا کار می‌کند؛ سرور فهرستی ندارد که آن را از رویش پاک کند. پس توکن‌های دسترسی عمر کوتاهی دارند، معمولاً پانزده تا شصت دقیقه.

۳. وابستگی کاربر جاری

@app.post('/auth/token')
def login(form: Annotated[OAuth2PasswordRequestForm, Depends()]) -> dict[str, str]:
    user = next((u for u in USERS.values() if u.email == form.username.lower()), None)
    if user is None or not password_hash.verify(form.password, user.hashed_password):
        raise CREDENTIALS_ERROR
    expires = datetime.now(UTC) + timedelta(minutes=30)
    token = jwt.encode(
        {'sub': str(user.id), 'role': user.role, 'exp': expires}, SECRET_KEY, ALGORITHM
    )
    return {'access_token': token, 'token_type': 'bearer'}


def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]) -> User:
    try:
        claims = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
    except jwt.InvalidTokenError:  # bad signature, expired, malformed...
        raise CREDENTIALS_ERROR from None
    user = USERS.get(int(claims['sub']))
    if user is None:
        raise CREDENTIALS_ERROR
    if not user.is_active:
        raise HTTPException(status.HTTP_403_FORBIDDEN, detail='Account disabled')
    return user


CurrentUser = Annotated[User, Depends(get_current_user)]


@app.get('/users/me')
def read_me(user: CurrentUser) -> dict:
    return {'id': user.id, 'email': user.email, 'role': user.role}

از توکن تا کاربر: نمونه 03_current_user/app.py

get_current_user قلب ماجراست:

1. OAuth2PasswordBearer هدر را می‌خواند (اگر نباشد ۴۰۱).
2. JWT رمزگشایی و امضا و انقضایش بررسی می‌شود (اگر نامعتبر باشد ۴۰۱).
3. کاربر از «پایگاه داده» بارگذاری می‌شود (اگر نباشد ۴۰۱ و اگر غیرفعال باشد ۴۰۳).

هر endpoint محافظت‌شده فقط CurrentUser را می‌خواهد. همین.

۴. نقش‌ها: کارخانه وابستگی

def require_roles(*roles: str) -> Callable[[User], User]:
    def checker(user: CurrentUser) -> User:
        if user.role not in roles:
            raise HTTPException(
                status.HTTP_403_FORBIDDEN, detail=f'Requires one of the roles: {", ".join(roles)}'
            )
        return user

    return checker


AdminUser = Annotated[User, Depends(require_roles('admin'))]
StaffUser = Annotated[User, Depends(require_roles('admin', 'instructor'))]
StudentUser = Annotated[User, Depends(require_roles('student'))]


@app.get('/users/me')
def me(user: CurrentUser) -> User:
    return user


@app.post('/courses')
def create_course(user: StaffUser) -> dict[str, str]:
    return {'created by': user.email}


@app.delete('/courses/{course_id}')
def delete_course(course_id: int, admin: AdminUser) -> dict[str, str | int]:
    return {'deleted': course_id, 'by': admin.email}


@app.post('/courses/{course_id}/enrollments')
def enroll(course_id: int, student: StudentUser) -> dict[str, str | int]:
    return {'enrolled': student.email, 'course': course_id}

require_roles و نام‌های مستعار: نمونه 04_role_dependency/app.py

این همان تابع مرتبه بالای فصل اول است: require_roles یک بستار برمی‌گرداند که roles را به خاطر دارد. endpointها مثل یک جمله خوانده می‌شوند: def delete_course(course_id: int, admin: AdminUser).

۵. فراتر از نقش: قاعده مالکیت

نقش می‌گوید کاربر از چه نوعی است. اما «مدرس دوره‌های خودش را تغییر دهد» را نقش به‌تنهایی نمی‌تواند تعیین کند؛ باید دوره را بارگذاری کنیم و course.instructor_id را با user.id مقایسه کنیم. این قاعده‌ها در سرویس، درست بعد از بارگذاری شیء، و به شکل توابع ساده بولی نوشته می‌شوند:

"""Authorization rules that depend on the data, not only on the role."""

from app.exceptions import PermissionDeniedError
from app.models import Course, User


def can_manage_course(course: Course, user: User) -> bool:
    """Admins manage every course; instructors only their own."""
    if user.role == 'admin':
        return True
    return user.role == 'instructor' and course.instructor_id == user.id


def ensure_can_manage_course(course: Course, user: User) -> None:
    if not can_manage_course(course, user):
        raise PermissionDeniedError('You can only manage your own courses')

قاعده‌های وابسته به داده: 07_project/app/permissions.py

۶. OAuth2 scopes، برای شناختن

نقش‌ها درشت‌اند. scopeها توانایی‌های تکی را نام می‌برند: courses:read، courses:write. توکن فهرست scopeهای اعطاشده را دارد و هر endpoint با Security(...) به جای Depends(...) می‌گوید چه چیزی لازم دارد:

def get_current_user(
    security_scopes: SecurityScopes, token: Annotated[str, Depends(oauth2_scheme)]
) -> dict:
    try:
        claims = jwt.decode(token, SECRET_KEY, algorithms=['HS256'])
    except jwt.InvalidTokenError:
        raise HTTPException(401, detail='Invalid token') from None
    missing = [s for s in security_scopes.scopes if s not in claims['scopes']]
    if missing:
        raise HTTPException(
            status.HTTP_403_FORBIDDEN,
            detail=f'Missing scopes: {missing}',
            headers={'WWW-Authenticate': f'Bearer scope="{security_scopes.scope_str}"'},
        )
    return claims


@app.get('/courses')
def list_courses(
    user: Annotated[dict, Security(get_current_user, scopes=['courses:read'])],
) -> dict:
    return {'courses': ['FastAPI'], 'your scopes': user['scopes']}


@app.post('/courses')
def create_course(
    user: Annotated[dict, Security(get_current_user, scopes=['courses:write'])],
) -> dict:
    return {'created by': user['sub']}

Security و SecurityScopes: نمونه 06_scopes/app.py

کاربرد رایج: یک برنامه شخص ثالث توکنی می‌گیرد که فقط courses:read دارد، حتی اگر کسی که به آن اجازه داده مدیر باشد. در این دوره مجوز را با نقش‌ها پیاده می‌کنیم؛ scopeها را معرفی می‌کنیم تا در پروژه‌های دیگر آنها را بشناسید.

۷. پروژه: JWT و نقش‌ها

def create_access_token(user_id: int, role: str, settings: Settings) -> str:
    now = datetime.now(UTC)
    claims = {
        'sub': str(user_id),
        'role': role,
        'iat': now,
        'exp': now + timedelta(minutes=settings.access_token_expire_minutes),
    }
    return jwt.encode(claims, settings.jwt_secret_key, algorithm=settings.jwt_algorithm)


def decode_access_token(token: str, settings: Settings) -> TokenPayload:
    """Check signature and expiry; every problem becomes AuthenticationError (401)."""
    try:
        claims = jwt.decode(
            token,
            settings.jwt_secret_key,
            algorithms=[settings.jwt_algorithm],  # a white list, never from the token
            options={'require': ['sub', 'exp', 'iat']},
        )
        return TokenPayload.model_validate(claims)
    except jwt.ExpiredSignatureError:
        raise AuthenticationError('Token has expired') from None
    except (jwt.InvalidTokenError, ValidationError):
        raise AuthenticationError('Could not validate credentials') from None

ساختن و بررسی توکن: 07_project/app/security.py

SettingsDep = Annotated[Settings, Depends(get_settings)]
DbSession = Annotated[Session, Depends(get_db)]

# Reads "Authorization: Bearer <token>" (401 if missing) and adds the
# Authorize button to /docs, which posts the login form to tokenUrl.
oauth2_scheme = OAuth2PasswordBearer(tokenUrl='/auth/token')
TokenDep = Annotated[str, Depends(oauth2_scheme)]


def get_current_user(token: TokenDep, db: DbSession, settings: SettingsDep) -> User:
    payload = decode_access_token(token, settings)
    user = db.get(User, int(payload.sub))
    if user is None:
        raise AuthenticationError('The user of this token no longer exists')
    if not user.is_active:
        raise PermissionDeniedError('This account is disabled')
    return user


CurrentUser = Annotated[User, Depends(get_current_user)]


def require_roles(*roles: str) -> Callable[[User], User]:
    """A dependency factory: `Depends(require_roles('admin', 'instructor'))`."""

    def checker(user: CurrentUser) -> User:
        if user.role not in roles:
            raise PermissionDeniedError(f'This action requires the role: {" or ".join(roles)}')
        return user

    return checker


AdminUser = Annotated[User, Depends(require_roles('admin'))]
StaffUser = Annotated[User, Depends(require_roles('admin', 'instructor'))]
StudentUser = Annotated[User, Depends(require_roles('student'))]

کاربر جاری و نقش‌ها: 07_project/app/deps.py

در مجموعه Bruno این فصل، چهار درخواست ورود اول توکن‌ها را در متغیرهای adminToken، teacherToken، teacher2Token و studentToken ذخیره می‌کنند و بقیه درخواست‌ها از آنها استفاده می‌کنند. بیشتر درخواست‌ها عمداً باید ۴۰۱ یا ۴۰۳ بگیرند.

۸. خلاصه فصل