Lab, spotkanie 2: Walidacja, logowanie, role i ochrona

🛠️ Laboratorium — Spotkanie 2 (3h)

Cztery części: dopinamy walidację danych, dodajemy prawdziwych użytkowników z hasłami, uwierzytelnianie przez JWT z rolami, i na koniec łatamy dziury bezpieczeństwa, które w tym momencie ma każdy niedbale napisany serwer.

Część I — Walidacja danych i obsługa błędów

🎯 Cel

Wzmocnić walidację ponad to, co Pydantic robi domyślnie (typy) — dodać reguły biznesowe (np. data_od musi być przed data_do) i ujednolicić sposób zgłaszania błędów w całym API.

1️⃣ Walidacja na poziomie pola — Field

from pydantic import BaseModel, Field

class PoligonCreate(BaseModel):
    nazwa: str = Field(min_length=2, max_length=100)
    lokalizacja: str = Field(min_length=2)
    typ: str
    dostepny: bool = True

Field pozwala dodać ograniczenia (długość, zakres liczbowy) bez pisania własnego kodu sprawdzającego — Pydantic zwróci 422 z czytelnym opisem, zanim Twoja funkcja w ogóle zobaczy dane.

2️⃣ Walidacja reguł biznesowych — model_validator

Reguła “data_od musi być przed data_do” dotyczy relacji między dwoma polami, nie pojedynczego pola — potrzebuje własnego walidatora:

from pydantic import model_validator
from datetime import datetime

class RezerwacjaCreate(BaseModel):
    poligon_id: int
    jednostka_id: int
    data_od: str
    data_do: str

    @model_validator(mode="after")
    def sprawdz_zakres_dat(self):
        if datetime.fromisoformat(self.data_od) >= datetime.fromisoformat(self.data_do):
            raise ValueError("data_od musi byc wczesniejsza niz data_do")
        return self

Taki błąd Pydantic zamienia automatycznie na 422 — spójnie z resztą walidacji.

3️⃣ Spójne kody błędów w całym API

Ustal (i trzymaj się) konwencji:

Sytuacja Kod
Dane niepoprawnego typu/kształtu 422 (automatycznie od Pydantic)
Zasób nie istnieje 404
Poprawne dane, ale konflikt ze stanem (duplikat, nakładający się termin) 409
Brak uprawnień 403
Brak uwierzytelnienia 401

✅ Mini-zadanie części I

Dodaj model_validator sprawdzający, że data_od rezerwacji nie jest w przeszłości (porównaj z datetime.now()). Sprawdź, że próba rezerwacji poligonu na wczoraj zwraca 422 z czytelnym komunikatem.

Część II — Rejestracja i logowanie

🎯 Cel

Dodać prawdziwych użytkowników: tabelę w bazie, bezpieczne hashowanie haseł, i pierwsze dwa endpointy uwierzytelniania: /register i /login.

1️⃣ Tabela użytkowników

CREATE TABLE IF NOT EXISTS uzytkownicy (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    login TEXT NOT NULL UNIQUE,
    hash_hasla TEXT NOT NULL,
    rola TEXT NOT NULL DEFAULT 'dowodca',
    jednostka_id INTEGER REFERENCES jednostki(id)
);

Konto dowódcy otrzymuje przypisanie do jednostki od prowadzącego/administratora. Nowe konto ma jednostka_id = NULL i do czasu przypisania nie może zarządzać rezerwacjami. Formularz rejestracji nie przyjmuje roli ani identyfikatora jednostki. Planista może obsługiwać wszystkie jednostki i nie wymaga własnego przypisania.

Jeśli tabela już istnieje, CREATE TABLE IF NOT EXISTS jej nie zmieni. Sprawdź PRAGMA table_info(uzytkownicy) i tylko gdy kolumny brakuje, wykonaj jednorazowo:

ALTER TABLE uzytkownicy ADD COLUMN jednostka_id INTEGER REFERENCES jednostki(id);

W get_db() ze spotkania 1 włącz conn.execute("PRAGMA foreign_keys = ON") zaraz po otwarciu każdego połączenia, przed rozpoczęciem transakcji. Przypisanie wykonuje prowadzący na bazie ćwiczenia, używając rzeczywistych identyfikatorów:

with get_db() as conn:
    if conn.execute("SELECT id FROM jednostki WHERE id = ?", (unit_id,)).fetchone() is None:
        raise ValueError("Unknown unit")
    cursor = conn.execute(
        "UPDATE uzytkownicy SET jednostka_id = ? WHERE login = ?",
        (unit_id, account_login),
    )
    if cursor.rowcount != 1:
        raise ValueError("Unknown account")
    conn.commit()

unit_id i account_login to wartości wybrane przez prowadzącego dla istniejącej jednostki i konta. Przygotuj osobno konta Alfa i Bravo. Nie udostępniaj tego kodu jako publicznego endpointu samodzielnego przypisywania jednostki.

2️⃣ Hashowanie haseł — nigdy jawnym tekstem

pip install bcrypt
import bcrypt

def zahashuj_haslo(haslo: str) -> str:
    return bcrypt.hashpw(haslo.encode(), bcrypt.gensalt()).decode()

def sprawdz_haslo(haslo: str, hash_z_bazy: str) -> bool:
    return bcrypt.checkpw(haslo.encode(), hash_z_bazy.encode())

3️⃣ Endpoint /register

class RejestracjaRequest(BaseModel):
    login: str
    haslo: str = Field(min_length=8)
    # UWAGA: celowo BRAK pola "rola" tutaj — patrz callout ponizej

@app.post("/register", status_code=201)
def rejestracja(dane: RejestracjaRequest):
    with get_db() as conn:
        istniejacy = conn.execute("SELECT id FROM uzytkownicy WHERE login = ?", (dane.login,)).fetchone()
        if istniejacy:
            raise HTTPException(status_code=409, detail="Login juz zajety")
        cursor = conn.execute(
            "INSERT INTO uzytkownicy (login, hash_hasla, rola) VALUES (?, ?, 'dowodca')",
            (dane.login, zahashuj_haslo(dane.haslo)),
        )
        conn.commit()
        return {"id": cursor.lastrowid, "login": dane.login, "rola": "dowodca"}
ImportantRola nie może pochodzić od klienta

Wersja tego endpointu, która przyjmuje rola: str wprost z żądania (dane.rola), pozwala dowolnemu nieuwierzytelnionemu użytkownikowi zarejestrować się od razu jako "planista" — czyli samodzielnie nadać sobie uprawnienia, które dopiero za chwilę (część IV) będziemy starannie chronić 403-ką. To nie teoretyczna dziura: to dokładnie ten sam błąd klasy mass assignment, co pozwolenie klientowi wysłać is_admin: true w formularzu rejestracji.

/register zawsze tworzy rolę dowodca, z góry, bez pytania klienta. Nadawanie roli planista to osobna operacja — patrz część III, gdzie budujemy do niej chroniony endpoint (wymagający, żeby robił to już istniejący planista).

🔹 Skąd bierze się PIERWSZY planista?

Skoro awans na planistę wymaga (część III) bycia już zalogowanym jako planista, ktoś musi być tym pierwszym — jajko i kura. W praktyce rozwiązuje się to jednym z dwóch sposobów:

  1. Seed przy starcie projektu — skrypt seed.py (macie go już z laboratorium o bazie danych) wstawia jedno konto z rola = 'planista' bezpośrednio przez SQL, z pominięciem API.
  2. Ręczna zmiana w bazie — prowadzący/administrator loguje się do poligony.db i robi UPDATE uzytkownicy SET rola = 'planista' WHERE login = ... dla konkretnego konta.

Żadna z tych dróg nie prowadzi przez publiczne, nieuwierzytelnione API — i tak właśnie powinno być.

4️⃣ Endpoint /login (na razie bez tokenu — dopiero część III)

class LoginRequest(BaseModel):
    login: str
    haslo: str

@app.post("/login")
def login(dane: LoginRequest):
    with get_db() as conn:
        uzytkownik = conn.execute("SELECT * FROM uzytkownicy WHERE login = ?", (dane.login,)).fetchone()
        if not uzytkownik or not sprawdz_haslo(dane.haslo, uzytkownik["hash_hasla"]):
            raise HTTPException(status_code=401, detail="Nieprawidlowy login lub haslo")
        return {"komunikat": "Zalogowano", "rola": uzytkownik["rola"]}

Zwróć uwagę: komunikat błędu jest celowo taki sam niezależnie od tego, czy zły jest login, czy hasło (“Nieprawidlowy login lub haslo”) — gdyby rozróżniać (“login nie istnieje” vs “złe hasło”), atakujący mógłby w ten sposób sprawdzać, które loginy istnieją w systemie.

✅ Mini-zadanie części II

Dodaj walidację, że login ma co najmniej 3 znaki i składa się tylko z liter/cyfr (np. przez Field(pattern=...) albo model_validator).

Część III — JWT i autoryzacja ról

🎯 Cel

Zamienić /login z części II na wydawanie prawdziwego tokenu JWT, i zabezpieczyć wybrane endpointy tak, żeby wymagały konkretnej roli (np. tylko planista może zatwierdzać rezerwacje).

1️⃣ Generowanie tokenu

pip install python-jose[cryptography]
from jose import jwt
from jose.exceptions import JWTError
from datetime import datetime, timedelta

SECRET_KEY = "zmien-mnie-w-prawdziwym-projekcie"  # docelowo: zmienna srodowiskowa!
ALGORITHM = "HS256"

def utworz_token(login: str, rola: str) -> str:
    dane = {
        "sub": login,
        "rola": rola,
        "exp": datetime.utcnow() + timedelta(hours=8),
    }
    return jwt.encode(dane, SECRET_KEY, algorithm=ALGORITHM)

Zaktualizuj /login, żeby zwracał token zamiast samego komunikatu:

@app.post("/login")
def login(dane: LoginRequest):
    # ... weryfikacja jak w czesci II ...
    token = utworz_token(uzytkownik["login"], uzytkownik["rola"])
    return {"access_token": token, "token_type": "bearer"}

2️⃣ Weryfikacja tokenu — Depends

FastAPI ma wbudowany mechanizm wstrzykiwania zależności (Depends) — idealny do “za każdym razem zweryfikuj token, zanim wykonasz właściwą logikę endpointu”.

NoteDlaczego nie OAuth2PasswordBearer?

FastAPI ma gotowy schemat OAuth2PasswordBearer, ale zakłada, że logowanie przyjmuje dane w formacie OAuth2PasswordRequestForm — pola username/password wysłane jako formularz (application/x-www-form-urlencoded), nie nasz JSON z polami login/haslo. Użycie go “na skróty” (jak w niejednej gotowej dokumentacji w internecie) sprawiłoby, że przycisk “Authorize” w /docs wysyłałby dane w formacie, którego nasz /login nie rozumie. Zamiast dopasowywać cały nasz projekt do cudzej konwencji, czytamy nagłówek Authorization ręcznie — mniej “magii”, w pełni spójne z tym, co już mamy.

from fastapi import Depends, Header

def pobierz_biezacego_uzytkownika(authorization: str | None = Header(default=None)) -> dict:
    if not authorization or not authorization.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="Brak lub nieprawidlowy naglowek Authorization")
    token = authorization.removeprefix("Bearer ")
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        account_login = payload.get("sub")
        if not isinstance(account_login, str):
            raise HTTPException(status_code=401, detail="Invalid token subject")
    except JWTError:
        raise HTTPException(status_code=401, detail="Nieprawidlowy lub wygasly token")
    with get_db() as conn:
        account = conn.execute(
            "SELECT id, login, rola, jednostka_id FROM uzytkownicy WHERE login = ?",
            (account_login,),
        ).fetchone()
    if account is None:
        raise HTTPException(status_code=401, detail="Account no longer exists")
    return dict(account)

Nagłówek jest opcjonalny na poziomie walidacji parametrów, aby nasz kod zwracał 401 również wtedy, gdy go brakuje. Wymagany Header(...) spowodowałby wcześniej błąd walidacji 422. Token potwierdza login, natomiast aktualną rolę i przypisanie do jednostki odczytujemy z bazy; nie ufamy starej roli zapisanej w tokenie.

Dowolny endpoint dostaje uwierzytelnionego użytkownika, dopisując jeden parametr:

@app.get("/rezerwacje")
def lista_rezerwacji(uzytkownik: dict = Depends(pobierz_biezacego_uzytkownika)):
    ...

3️⃣ Autoryzacja — wymagaj konkretnej roli

def wymagaj_roli(wymagana_rola: str):
    def sprawdz(uzytkownik: dict = Depends(pobierz_biezacego_uzytkownika)) -> dict:
        if uzytkownik["rola"] != wymagana_rola:
            raise HTTPException(status_code=403, detail="Brak uprawnien do tej operacji")
        return uzytkownik
    return sprawdz


@app.post("/rezerwacje/{rez_id}/zatwierdz")
def zatwierdz_rezerwacje(rez_id: int, uzytkownik: dict = Depends(wymagaj_roli("planista"))):
    ...

wymagaj_roli("planista") to fabryka zależności — funkcja zwracająca funkcję, dopasowaną do konkretnej roli. Dzięki temu jeden mechanizm obsługuje dowolną liczbę ról bez powtarzania kodu.

Kontrola jednostki: rola nie wystarcza

Dodaj funkcję używaną przy każdej operacji na konkretnej jednostce:

def require_unit_access(account: dict, unit_id: int) -> None:
    if account["rola"] == "planista":
        return
    if account["jednostka_id"] is None or account["jednostka_id"] != unit_id:
        raise HTTPException(status_code=403, detail="Unit access denied")


@app.delete("/rezerwacje/{reservation_id}")
def delete_reservation(
    reservation_id: int,
    account: dict = Depends(pobierz_biezacego_uzytkownika),
):
    with get_db() as conn:
        conn.execute("BEGIN IMMEDIATE")
        reservation = conn.execute(
            "SELECT jednostka_id FROM rezerwacje WHERE id = ?", (reservation_id,),
        ).fetchone()
        if reservation is None:
            raise HTTPException(status_code=404, detail="Reservation not found")
        require_unit_access(account, reservation["jednostka_id"])
        conn.execute("DELETE FROM rezerwacje WHERE id = ?", (reservation_id,))
        conn.commit()
    return {"deleted": reservation_id}

Zamknięcie połączenia przez get_db() cofa niezatwierdzoną transakcję w ścieżce odmowy. W przykładzie tworzenia rezerwacji ze spotkania 1 dodaj parametr account: dict = Depends(pobierz_biezacego_uzytkownika) oraz wywołaj require_unit_access(account, rez.jednostka_id) przed zapisem, zachowując atomowe sprawdzenie konfliktu. Przy zmianie rezerwacji sprawdź zarówno jej obecną, jak i proponowaną jednostkę — nie wystarczy sprawdzić danych przesłanych przez klienta.

Dla GET /rezerwacje planista odczytuje wszystkie wiersze. Dowódca bez przypisania otrzymuje 403, a przypisany dowódca odczytuje wyłącznie WHERE jednostka_id = ? z wartością pobraną z konta. Filtr przekazany w URL nie może poszerzyć tego zakresu. Zastąp wcześniejsze niezabezpieczone endpointy, zamiast rejestrować dwie funkcje dla tej samej metody i ścieżki.

Próba odbiorcza na świeżych danych: Alfa usuwa swoją rezerwację; Alfa nie może usunąć rezerwacji Bravo (403 i rekord nadal istnieje); planista może ją usunąć. Brak nagłówka daje 401. Konto bez jednostki dostaje 403 przy próbie rezerwacji. Przed każdą próbą odtwórz potrzebny rekord, aby wcześniejsze usunięcie nie zamieniło próby uprawnień w zwykłą odpowiedź 404.

4️⃣ Awans na planistę — dopiero teraz to możliwe

Mamy już wymagaj_roli, więc możemy dotrzymać obietnicy z części II: nadawanie roli planista to osobny, chroniony endpoint, nie pole w /register.

class AwansRequest(BaseModel):
    login: str

@app.post("/uzytkownicy/awansuj")
def awansuj_na_planiste(
    dane: AwansRequest,
    uzytkownik: dict = Depends(wymagaj_roli("planista")),  # tylko obecny planista moze awansowac
):
    with get_db() as conn:
        cursor = conn.execute(
            "UPDATE uzytkownicy SET rola = 'planista' WHERE login = ?", (dane.login,)
        )
        conn.commit()
        if cursor.rowcount == 0:
            raise HTTPException(status_code=404, detail="Uzytkownik nie znaleziony")
        return {"komunikat": f"{dane.login} jest teraz planista"}

Zamknięty krąg: żeby kogoś awansować, sam musisz już być planistą — a pierwszy planista powstaje przez seed.py albo ręczny UPDATE, jak opisano w części II. Nikt nie zostaje planistą przez samo wypełnienie formularza rejestracji.

✅ Mini-zadanie części III

  1. Zabezpiecz DELETE /poligony/{id} tak, żeby tylko planista mógł usuwać poligony — dowodca próbujący to zrobić powinien dostać 403.
  2. Sprawdź ręcznie (przez seed/UPDATE) nadanie sobie roli planisty, zaloguj się, i przetestuj /uzytkownicy/awansuj na drugim, świeżo zarejestrowanym koncie.

Część IV — Ochrona przed atakami

🎯 Cel

Zamknąć trzy luki bezpieczeństwa, które są łatwe do przeoczenia w kodzie napisanym “na szybko”: brak walidacji nazw kolumn w sortowaniu, brak limitu żądań, i brak kontroli CORS.

1️⃣ SQL injection przez nazwę kolumny — luka, której nie łata ?

# PODATNE — mimo ze reszta zapytania jest parametryzowana!
@app.get("/rezerwacje")
def lista(sortuj: str = "data_od"):
    return conn.execute(f"SELECT * FROM rezerwacje ORDER BY {sortuj}").fetchall()

Placeholdery (?) działają tylko dla wartości, nie dla nazw kolumn/tabel — sortuj wstawiony przez f-string pozwala atakującemu wpisać dowolny SQL. Obrona: biała lista dozwolonych wartości.

DOZWOLONE_KOLUMNY = {"data_od", "data_do", "status"}

@app.get("/rezerwacje")
def lista(sortuj: str = "data_od"):
    if sortuj not in DOZWOLONE_KOLUMNY:
        raise HTTPException(status_code=400, detail="Nieprawidlowa kolumna sortowania")
    return conn.execute(f"SELECT * FROM rezerwacje ORDER BY {sortuj}").fetchall()

2️⃣ Rate limiting — ogranicz liczbę żądań

Bez limitu, ktoś może próbować zalogować się tysiące razy na sekundę (atak brute-force na hasło) albo po prostu zasypać serwer żądaniami.

pip install slowapi
from slowapi import Limiter
from slowapi.util import get_remote_address

limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter

@app.post("/login")
@limiter.limit("5/minute")
def login(request: Request, dane: LoginRequest):
    ...

3️⃣ CORS — kto może wywoływać Twoje API z przeglądarki

Domyślnie przeglądarki blokują żądania JavaScript do innej domeny niż ta, z której strona została załadowana (Same-Origin Policy). Jeśli frontend (część II następnego spotkania) ma żyć na innej domenie/porcie niż backend, musisz jawnie zezwolić:

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000"],  # tylko Twoj frontend, NIE "*" na produkcji
    allow_methods=["*"],
    allow_headers=["*"],
)
ImportantNigdy allow_origins=[“*”] razem z uwierzytelnianiem

"*" (dowolna domena) w połączeniu z endpointami wymagającymi logowania otwiera drzwi dla ataków CSRF-podobnych z dowolnej złośliwej strony. Zawsze wymieniaj konkretne, zaufane domeny.

✅ Mini-zadanie części IV

  1. Dodaj białą listę dozwolonych kolumn do wszystkich endpointów z parametrem sortuj.
  2. Dodaj rate limiting do /login (max 5 prób/minutę z jednego adresu IP).
  3. Skonfiguruj CORS dla http://localhost:3000 (przyda się w laboratorium o frontendzie).

✅ Zadanie na koniec spotkania

Zademonstruj prowadzącemu: rejestrację nowego użytkownika, logowanie zwracające token JWT, endpoint chroniony rolą (403 dla złej roli, sukces dla właściwej), oraz że próba SQL injection przez ?sortuj= jest odrzucana.

Do oddania: link do repozytorium / commit hash.