Praktik: Otorisasi, Access Control & Rate Limiting

API kita sekarang tahu SIAPA yang login (Bagian 3). Tapi tahu identitas saja belum cukup — kita belum pernah mengecek APA yang boleh diakses identitas itu. Di artikel ini kita tutup dua ancaman sekaligus dari Bagian 2: Broken Object Level Authorization (BOLA) dan Unrestricted Resource Consumption.

Dependency get_current_user — Mengambil Identitas dari Token

Setiap endpoint yang butuh proteksi memakai dependency ini untuk mengetahui siapa yang sedang request:

from fastapi.security import OAuth2PasswordBearer

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login")


def get_current_user(
    token: str = Depends(oauth2_scheme),
    session: Session = Depends(get_session),
) -> User:
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )

    username = decode_access_token(token)
    if username is None:
        raise credentials_exception

    user = session.exec(select(User).where(User.username == username)).first()
    if user is None:
        raise credentials_exception

    return user

OAuth2PasswordBearer otomatis membaca header Authorization: Bearer <token> dari request. Kalau tidak ada header itu sama sekali, FastAPI otomatis menolak dengan 401 SEBELUM fungsi kita sempat jalan. Kalau ada tapi token-nya tidak valid (kedaluwarsa, signature salah, atau user-nya sudah tidak ada), kita tolak manual lewat credentials_exception.

Endpoint yang butuh proteksi tinggal menambahkan satu parameter:

@router.get("/api/tasks")
def list_tasks(current_user: User = Depends(get_current_user)):
    ...

Object Level Authorization — Cek Kepemilikan di SETIAP Endpoint

Ini bagian intinya. Model Task sekarang punya owner_id:

class Task(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    title: str
    ...
    owner_id: int = Field(foreign_key="user.id", index=True)

Dan SETIAP endpoint yang mengakses task spesifik WAJIB memverifikasi kepemilikannya:

@router.get("/{task_id}", response_model=TaskRead)
def get_task(
    task_id: int,
    session: Session = Depends(get_session),
    current_user: User = Depends(get_current_user),
):
    task = session.get(Task, task_id)
    if not task or task.owner_id != current_user.id:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Task not found")
    return task

Untuk list_tasks, kita bahkan tidak perlu cek manual — filter-nya sudah membatasi query sejak awal:

@router.get("", response_model=List[TaskRead])
def list_tasks(current_user: User = Depends(get_current_user), ...):
    query = select(Task).where(Task.owner_id == current_user.id)
    ...

Detail keamanan yang gampang terlewat: perhatikan kita mengembalikan 404 Not Found, BUKAN 403 Forbidden, untuk task yang ada tapi bukan milik user tersebut. Kenapa? Kalau kita balas 403, itu secara tidak langsung mengonfirmasi "task dengan ID ini memang ADA, cuma kamu tidak boleh akses" — informasi yang tidak seharusnya diketahui pihak yang tidak berhak. Dengan 404, dari sudut pandang penyerang, tidak ada beda antara "task tidak ada" dan "task ada tapi bukan milikmu" — keduanya terlihat identik.

Membuktikan dengan Dua User Sungguhan

# Alice bikin task
curl -X POST localhost:8000/api/tasks -H "Authorization: Bearer $ALICE_TOKEN" \
  -H "Content-Type: application/json" -d '{"title":"Task rahasia alice"}'
# {"id":1, "title":"Task rahasia alice", ...}

# Bob coba akses task #1 (milik Alice)
curl localhost:8000/api/tasks/1 -H "Authorization: Bearer $BOB_TOKEN"
# {"detail":"Task not found"}  ← 404, bukan bocor data Alice

# Bob list task-nya sendiri
curl localhost:8000/api/tasks -H "Authorization: Bearer $BOB_TOKEN"
# []  ← kosong, task Alice tidak ikut muncul

Persis seperti dirancang: Bob tidak bisa melihat, apalagi mengubah atau menghapus, task milik Alice — walaupun dia tahu persis ID-nya.

Rate Limiting — Mencegah Brute Force di Endpoint Login

Sekarang bagian kedua: mencegah seseorang mencoba ribuan kombinasi password di endpoint login. Kita pakai Redis (sudah kita kenal dari seri Scalability) sebagai penghitung, dengan pola fixed window:

def rate_limit_login(request: Request, cache: redis.Redis = Depends(get_redis)) -> None:
    identifier = request.client.host if request.client else "unknown"
    key = f"rate-limit:login:{identifier}"

    current_attempts = cache.incr(key)
    if current_attempts == 1:
        cache.expire(key, settings.login_rate_limit_window_seconds)

    if current_attempts > settings.login_rate_limit_max_attempts:
        raise HTTPException(
            status_code=status.HTTP_429_TOO_MANY_REQUESTS,
            detail="Terlalu banyak percobaan login. Coba lagi nanti.",
        )

Cara kerjanya: INCR di Redis menaikkan counter dan mengembalikan nilai barunya secara atomik (aman dari race condition walau banyak request bersamaan). Percobaan PERTAMA dalam window itu yang men-set expire — jadi counter otomatis reset ke 0 setelah login_rate_limit_window_seconds detik berlalu tanpa kita perlu job pembersihan terpisah.

Dipasang sebagai dependency khusus di endpoint login:

@router.post("/login", response_model=Token, dependencies=[Depends(rate_limit_login)])
def login(...):
    ...

Membuktikan dengan percobaan login gagal beruntun (default: maksimum 5 percobaan per 60 detik):

for i in 1 2 3 4 5 6 7; do
  curl -s -o /dev/null -w "percobaan $i: %{http_code}\n" \
    -X POST localhost:8000/auth/login -d "username=alice&password=passwordsalah"
done

percobaan 1: 401
percobaan 2: 401
percobaan 3: 401
percobaan 4: 401
percobaan 5: 401
percobaan 6: 429
percobaan 7: 429

Lima percobaan pertama tetap 401 (password memang salah), tapi percobaan ke-6 dan seterusnya langsung ditolak 429 Too Many Requests — SEBELUM sempat mengecek password sama sekali. Penyerang yang mencoba brute force sekarang dibatasi kecepatannya secara drastis, sementara user sah yang cuma salah ketik sesekali tidak terganggu.

Catatan penting: rate limiter ini menghitung SEMUA percobaan dari satu sumber (di sini, berdasarkan alamat IP), baik yang berhasil maupun gagal — bukan cuma yang gagal. Untuk sistem production yang lebih matang, pertimbangkan rate limit terpisah untuk percobaan gagal saja (supaya user yang berhasil login berkali-kali dalam waktu singkat, misalnya dari banyak tab, tidak ikut kena batas).

Rangkuman