Autoryzacja w FastAPI: użytkownik widzi tylko swoje dane

Logowanie (uwierzytelnianie) odpowiada na pytanie „kim jesteś?". Autoryzacja odpowiada na kolejne: „do czego masz dostęp?". Brak tej drugiej warstwy, czyli izolacji danych między użytkownikami, to jedna z najczęstszych luk w aplikacjach, bo kod działa idealnie dla jednego użytkownika, a problem wychodzi dopiero, gdy pojawia się drugi. W tym artykule domykasz cykl o bazach danych: dodajesz autoryzację do historii rozmów tak, żeby każdy widział tylko swoją, i piszesz test, który to potwierdza. Wszystko uruchomiłem lokalnie (Python 3.13, SQLAlchemy 2.1, FastAPI 0.159, pytest 9.1).

Ten temat omawiamy na kursie Python online, w module o bazach danych i sesjach. Zakładam, że masz modele i sesję z artykułu o SQLAlchemy 2 oraz działające logowanie z tokenem JWT, które opisuję w artykule o uwierzytelnianiu w FastAPI.

Czym różni się uwierzytelnianie od autoryzacji w FastAPI?

Drugi błąd bywa podstępny, bo token jest prawidłowy i wszystkie zabezpieczenia logowania działają. Użytkownik po prostu prosi o rozmowę o numerze, który nie jest jego, a serwer ją oddaje. Ten rodzaj luki nazywa się IDOR (Insecure Direct Object Reference), czyli niezabezpieczone bezpośrednie odwołanie do obiektu.

Jak wygląda błąd, przez który użytkownik widzi cudze rozmowy?

Poniższy endpoint wygląda niewinnie i przechodzi każdy test napisany dla jednego użytkownika:

@app.get("/conversations/{conversation_id}")
async def get_conversation(
    conversation_id: int,
    user: Annotated[User, Depends(get_current_user)],
    session: Annotated[AsyncSession, Depends(get_session)],
):
    conversation = await session.get(Conversation, conversation_id)
    if conversation is None:
        raise HTTPException(404, "Nie ma takiej rozmowy")
    return {"id": conversation.id, "title": conversation.title}

Endpoint sprawdza, czy ktoś jest zalogowany (parametr user), ale nigdzie nie sprawdza, czyja jest rozmowa. Odtworzyłem to: Anna ma rozmowę „Prywatna Anny", a zalogowany Bartek wysyła GET /conversations/1 i dostaje:

200 {'id': 1, 'title': 'Prywatna Anny'}

Numery rozmów są kolejnymi liczbami, więc Bartek mógłby przejrzeć wszystkie po kolei. Wystarczy jedno zapomniane sprawdzenie w jednym endpoincie.

Jak zrobić autoryzację właściciela w jednym miejscu?

Zamiast pamiętać o warunku w każdym endpoincie, umieść go w jednej zależności. Każda trasa, która dotyka rozmowy, używa tej zależności i nie ma możliwości pobrać rozmowy inaczej.

from typing import Annotated

from fastapi import Depends, HTTPException
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from auth import get_current_user
from db import get_session
from models import Conversation, User

SessionDep = Annotated[AsyncSession, Depends(get_session)]
CurrentUser = Annotated[User, Depends(get_current_user)]


async def get_own_conversation(
    conversation_id: int, user: CurrentUser, session: SessionDep
) -> Conversation:
    stmt = select(Conversation).where(
        Conversation.id == conversation_id,
        Conversation.user_id == user.id,
    )
    conversation = await session.scalar(stmt)
    if conversation is None:
        raise HTTPException(404, "Nie ma takiej rozmowy")
    return conversation


OwnConversation = Annotated[Conversation, Depends(get_own_conversation)]

Kluczowa jest druga linia warunku: Conversation.user_id == user.id. Baza zwróci rozmowę tylko wtedy, gdy zgadza się zarówno numer, jak i właściciel. Typ OwnConversation możesz od teraz wpisać jako parametr dowolnego endpointu. FastAPI pobierze rozmowę, sprawdzi właściciela i dopiero wtedy wywoła Twoją funkcję. Jak działa Depends, wyjaśniam w artykule o uwierzytelnianiu w FastAPI.

Dlaczego cudza rozmowa ma dawać 404, a nie 403?

Kod 403 oznacza „ta rozmowa istnieje, ale nie wolno Ci jej zobaczyć". To już jest informacja: atakujący, który zgaduje numery, od razu wie, które są zajęte. Kod 404 zachowuje się tak samo dla rozmowy nieistniejącej i dla cudzej, więc nic nie zdradza.

Wyjątek stanowią sytuacje, w których użytkownik ma uzasadnione prawo wiedzieć, że zasób istnieje (np. udostępnione dokumenty z różnymi uprawnieniami). Przy prywatnych rozmowach 404 jest bezpieczniejszym wyborem.

Jak napisać endpointy tworzące i listujące rozmowy?

Właściciela ustawiasz zawsze po stronie serwera, z tokenu, a nie z danych przesłanych przez klienta. Klient nie może więc założyć rozmowy „w imieniu" kogoś innego.

@app.post("/conversations", status_code=201)
async def create_conversation(title: str, user: CurrentUser, session: SessionDep):
    conversation = Conversation(title=title, user_id=user.id)
    session.add(conversation)
    await session.commit()
    return {"id": conversation.id, "title": conversation.title}


@app.get("/conversations")
async def list_conversations(user: CurrentUser, session: SessionDep):
    stmt = (
        select(Conversation)
        .where(Conversation.user_id == user.id)
        .order_by(Conversation.id.desc())
    )
    rows = await session.scalars(stmt)
    return [{"id": c.id, "title": c.title} for c in rows]


@app.get("/conversations/{conversation_id}")
async def get_conversation(conversation: OwnConversation):
    return {"id": conversation.id, "title": conversation.title}


@app.delete("/conversations/{conversation_id}", status_code=204)
async def delete_conversation(conversation: OwnConversation, session: SessionDep):
    await session.delete(conversation)
    await session.commit()

Endpoint GET /conversations/{id} nie potrzebuje już własnego warunku, bo robi to zależność. Lista musi mieć filtr własny, bo nie ma konkretnego numeru, który zależność mogłaby sprawdzić. To miejsce, o którym najłatwiej zapomnieć.

Jak zarządzać historią rozmowy?

Historia może mieć setki wiadomości, a klientowi zwykle wystarczą ostatnie. Zamiast oddawać wszystko, stosujesz paginację: pobierasz najnowsze wiadomości, a starsze dopiero na żądanie.

from fastapi import Query
from pydantic import BaseModel


class MessageIn(BaseModel):
    role: str
    content: str


class MessageOut(MessageIn):
    id: int


@app.post("/conversations/{conversation_id}/messages", status_code=201)
async def add_message(conversation: OwnConversation, body: MessageIn, session: SessionDep):
    session.add(Message(conversation_id=conversation.id, **body.model_dump()))
    await session.commit()
    return {"ok": True}


@app.get("/conversations/{conversation_id}/messages", response_model=list[MessageOut])
async def get_history(
    conversation: OwnConversation,
    session: SessionDep,
    limit: Annotated[int, Query(ge=1, le=100)] = 20,
    before_id: int | None = None,
):
    stmt = select(Message).where(Message.conversation_id == conversation.id)
    if before_id is not None:
        stmt = stmt.where(Message.id < before_id)
    stmt = stmt.order_by(Message.id.desc()).limit(limit)
    rows = (await session.scalars(stmt)).all()
    return list(reversed(rows))

Zasada działania: limit mówi, ile wiadomości zwrócić, a before_id oznacza „daj mi te starsze niż ta". Pobierasz je od najnowszej (order_by(Message.id.desc())), a przed zwróceniem odwracasz kolejność, żeby klient dostał je w porządku chronologicznym. Przy pięciu wiadomościach (m0…m4) pierwsze wywołanie z limit=2 zwróciło m3, m4, a kolejne z before_id pierwszej z nich zwróciło m1, m2.

Dwie uwagi:

• Bez order_by baza nie gwarantuje żadnej kolejności wierszy. Zawsze określaj ją jawnie.
• Paginacja po id (a nie po numerze strony) nie gubi i nie dubluje wiadomości, gdy w trakcie czytania pojawiają się nowe.

Ta sama historia posłuży Ci przy aplikacjach z modelem językowym: ostatnie limit wiadomości to kontekst, który wysyłasz do modelu. Dlaczego okno kontekstu ma limit, opisuję w artykule o API OpenAI w Pythonie.

Co się dzieje z historią po usunięciu rozmowy?

W modelu Conversation ustawiliśmy cascade="all, delete-orphan", więc usunięcie rozmowy usuwa jej wiadomości. W teście po DELETE kolejne zapytanie o historię tej rozmowy zwraca 404, bo rozmowy już nie ma. Jeśli system ma pozwalać na przywracanie, zamiast kasować możesz dodać kolumnę archived i migrację w Alembicie, jak w artykule o migracjach w SQLAlchemy.

Jak przetestować autoryzację, czyli że dane są odizolowane?

Test dla jednego użytkownika nigdy nie wykryje przecieku. Potrzebujesz dwóch: jeden tworzy dane, drugi próbuje je odczytać, zmienić i usunąć. Fixture z bazą w pamięci i podmianą get_session pochodzą z artykułu o testowaniu bazy w pytest. Do tego dochodzą dwie fixture z klientami:

@pytest.fixture
async def users(session):
    anna, bartek = User(email="anna@example.com"), User(email="bartek@example.com")
    session.add_all([anna, bartek])
    await session.commit()
    return anna, bartek


@pytest.fixture
async def anna_client(make_client, users):
    async with await make_client(users[0]) as c:
        yield c


@pytest.fixture
async def bartek_client(make_client, users):
    async with await make_client(users[1]) as c:
        yield c

make_client tworzy klienta httpx z nagłówkiem Authorization: Bearer <token> wystawionym dla danego użytkownika:

@pytest.fixture
def make_client(engine):
    async def build(user=None):
        maker = async_sessionmaker(engine, expire_on_commit=False)

        async def override():
            async with maker() as s:
                yield s

        app.dependency_overrides[get_session] = override
        headers = {"Authorization": f"Bearer {create_token(user.id)}"} if user else {}
        return AsyncClient(
            transport=ASGITransport(app=app), base_url="http://test", headers=headers
        )

    yield build
    app.dependency_overrides.clear()

Sam test wygląda tak:

async def test_bez_tokenu_401(make_client):
    async with await make_client() as c:
        assert (await c.get("/conversations")).status_code == 401


async def test_lista_zawiera_tylko_moje(anna_client, bartek_client):
    await anna_client.post("/conversations", params={"title": "Anny"})
    await bartek_client.post("/conversations", params={"title": "Bartka"})
    assert [c["title"] for c in (await anna_client.get("/conversations")).json()] == ["Anny"]
    assert [c["title"] for c in (await bartek_client.get("/conversations")).json()] == ["Bartka"]


async def test_cudza_rozmowa_daje_404(anna_client, bartek_client):
    cid = (await anna_client.post("/conversations", params={"title": "Prywatna"})).json()["id"]
    assert (await bartek_client.get(f"/conversations/{cid}")).status_code == 404


async def test_nie_dopiszesz_do_cudzej(anna_client, bartek_client):
    cid = (await anna_client.post("/conversations", params={"title": "P"})).json()["id"]
    r = await bartek_client.post(
        f"/conversations/{cid}/messages", json={"role": "user", "content": "x"}
    )
    assert r.status_code == 404
    assert (await anna_client.get(f"/conversations/{cid}/messages")).json() == []


async def test_nie_usuniesz_cudzej(anna_client, bartek_client):
    cid = (await anna_client.post("/conversations", params={"title": "P"})).json()["id"]
    assert (await bartek_client.delete(f"/conversations/{cid}")).status_code == 404
    assert (await anna_client.get(f"/conversations/{cid}")).status_code == 200

Zwróć uwagę na ostatnią linię dwóch ostatnich testów. Nie wystarczy sprawdzić, że Bartek dostał 404, trzeba jeszcze zajrzeć do danych Anny i upewnić się, że nic się nie zmieniło. Odmowa, która mimo wszystko zapisała dane, byłaby błędem.

Wynik: 8 testów zaliczonych (te powyżej, plus odczyt cudzej historii, paginacja i usunięcie z historią). Test cudzej rozmowy uruchomiłem też na naiwnym endpoincie z początku artykułu, żeby zobaczyć, że potrafi się wywalić. Dostałem assert 200 == 404, czyli test wykrył przeciek.

Jakie błędy mentor poprawia najczęściej?

Najczęstszy błąd to sprawdzanie, czy ktoś jest zalogowany, ale nie czy dane należą do niego. Aplikacja z jednym użytkownikiem działa bez zarzutu, a przeciek wychodzi dopiero, gdy pojawi się drugi. Zanim uznasz endpoint za gotowy, napisz test z dwoma kontami i spróbuj odczytać cudzy zasób.

Czego jeszcze brakuje w tej aplikacji?

Ten kod nie jest gotowy na produkcję. Jest dobrym szkieletem, ale przed wdrożeniem dopisz:

• indeks na user_id w tabeli rozmów, bo przy tysiącach rozmów filtr po właścicielu zacznie spowalniać zapytania (nie mierzyłem tego tutaj),
• limit liczby rozmów i wiadomości na użytkownika, żeby nikt nie zapełnił Ci bazy,
• role (np. administrator, który może zobaczyć cudze dane), jeśli ich potrzebujesz,
• ochronę klucza JWT i hasła, o której piszę w artykule o uwierzytelnianiu.

Co dalej?

Masz za sobą cały cykl: modele i sesję, migracje w Alembicie, testy bazy i autoryzację dostępu do danych. To komplet potrzebny, żeby zbudować trwały zapis rozmów i dokumentów z własnym API. Jeśli w pracy trafisz na dane wielu klientów w jednej bazie, ta sama zasada „filtr po właścicielu w jednym miejscu i test z dwoma użytkownikami" nadal obowiązuje.

Wszystkie te zagadnienia omówimy na przyszłym kursie AI developer, który niedługo ogłosimy. Śledź listę naszych ścieżek, żeby nie przegapić startu.