Jak zaimplementować MFA (Multi-Factor Authentication) w aplikacji FastAPI

Multi-Factor Authentication (MFA) to proces, który zwiększa poziom bezpieczeństwa aplikacji, wymagając od użytkowników dostarczenia dwóch lub więcej czynników weryfikacyjnych, aby uzyskać dostęp do zasobów. Dodanie MFA do aplikacji FastAPI, w której użytkownik musi przedstawić zarówno coś, co zna (np. hasło), jak i coś, co posiada (np. urządzenie), stanowi istotny krok w kierunku zapewnienia wyższego poziomu ochrony.

W tej części przedstawimy, jak zaimplementować MFA w aplikacji FastAPI, korzystając z metody opierającej się na jednorazowych hasłach opóźnionych w czasie (TOTP). System ten generuje 6-8 cyfrowy kod, który jest ważny przez krótki okres, zazwyczaj 30 sekund.

Aby rozpocząć, należy upewnić się, że mamy zainstalowane odpowiednie pakiety. Zainstalujemy bibliotekę pyotp, która implementuje algorytmy generowania jednorazowych haseł, w tym TOTP. W tym celu użyjemy poniższego polecenia:

bash
Copy code
pip install pyotp

Następnie musimy dostosować naszą tabelę użytkowników w bazie danych, aby przechowywała sekret TOTP, który będzie używany do walidacji generowanego kodu. W tym celu dodajemy nowe pole totp_secret do modelu użytkownika:

python
Copy code
class User(Base):
    # istniejące pola
    totp_secret: Mapped[str] = mapped_column(
        nullable=True
    )

Po dodaniu odpowiednich zmian w bazie danych możemy przystąpić do implementacji MFA. Zacznijmy od stworzenia dwóch pomocniczych funkcji, które będą generować sekret TOTP oraz URI TOTP do użycia przez aplikację uwierzytelniającą.

Pierwsza funkcja generuje sekret TOTP:

python
Copy code
import pyotp

def generate_totp_secret():
    return pyotp.random_base32()

Druga funkcja generuje URI TOTP, które może być użyte do stworzenia kodu QR w aplikacji uwierzytelniającej:

python
Copy code
def generate_totp_uri(secret, user_email):
    return pyotp.totp.TOTP(secret).provisioning_uri(
        name=user_email, issuer_name="YourAppName"
    )

Następnie możemy zaimplementować punkt końcowy, który umożliwi użytkownikowi aktywację MFA. Punkt ten zwróci URI TOTP, które będzie mogło zostać użyte do wygenerowania kodu QR w aplikacji mobilnej użytkownika:

python
Copy code
from fastapi import APIRouter, Depends, HTTPException, status

from sqlalchemy.orm import Session
from db_connection import get_session
from operations import get_user
from rbac import get_current_user
from responses import UserCreateResponse
router = APIRouter()
@router.post("/user/enable-mfa")
def enable_mfa(
    user: UserCreateResponse = Depends(get_current_user),
    db_session: Session = Depends(get_session),
):
    secret = generate_totp_secret()
    db_user = get_user(db_session, user.username)
    db_user.totp_secret = secret
    db_session.add(db_user)
    db_session.commit()
    totp_uri = generate_totp_uri(secret, user.email)
    return {
        "totp_uri": totp_uri,
        "secret_numbers": pyotp.TOTP(secret).now(),
    }

Po aktywacji MFA, użytkownik może przejść do następnego kroku, czyli weryfikacji kodu TOTP, który zostanie wygenerowany przez aplikację uwierzytelniającą. Implementacja punktu weryfikacji wygląda następująco:

python
Copy code
@app.post("/verify-totp")
def verify_totp(
    code: str, username: str, session: Session = Depends(get_session)
):
    user = get_user(session, username)
    if not user.totp_secret:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST, detail="MFA not activated"
        )
    totp = pyotp.TOTP(user.totp_secret)
    if not totp.verify(code):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid TOTP token"
        )
    return {
        "message": "TOTP token verified successfully"
    }

Zanim przejdziemy do końca, upewnijmy się, że odpowiedni router jest dodany do głównej aplikacji FastAPI, a aplikacja jest uruchomiona za pomocą komendy:

bash
Copy code
uvicorn main:app

Warto zwrócić uwagę na kilka istotnych kwestii, które warto rozważyć przy implementacji MFA:

1. Bezpieczeństwo transmisji: Zawsze należy używać HTTPS, aby zapobiec przechwyceniu kodów TOTP i innych wrażliwych informacji podczas transmisji.

2. Bezpieczne przechowywanie sekretów: Sekrety TOTP nie powinny być przechowywane w sposób niezabezpieczony. W produkcyjnych aplikacjach warto zastosować rozwiązania typu HSM (Hardware Security Module) lub dedykowane usługi do przechowywania kluczy.

3. Regularna rotacja sekretów: Regularna zmiana sekretów TOTP to dobry sposób na zminimalizowanie ryzyka ich kompromitacji. Należy rozważyć implementację mechanizmów automatycznej zmiany sekretu.

4. Używanie aplikacji uwierzytelniających: Zamiast polegać na SMS-ach lub e-mailach do weryfikacji, lepiej jest korzystać z aplikacji uwierzytelniających, takich jak Google Authenticator czy Authy, które zapewniają wyższy poziom bezpieczeństwa.

MFA stanowi solidną barierę ochrony przed nieautoryzowanym dostępem, nawet w przypadku wycieku hasła użytkownika. Poprawna implementacja takich rozwiązań w aplikacji to inwestycja w bezpieczeństwo, która pozwoli na minimalizację ryzyka ataków.

Jak przeprowadzać testy wydajności aplikacji w Pythonie z użyciem Locust?

Testowanie wydajności aplikacji jest kluczowe, by upewnić się, że nasza aplikacja jest w stanie obsłużyć rzeczywiste scenariusze użytkowania, szczególnie w warunkach dużego obciążenia. Dzięki systematycznemu wdrażaniu testów wydajnościowych, analizowaniu wyników i optymalizacji na podstawie uzyskanych danych, możemy znacząco poprawić responsywność, stabilność oraz skalowalność aplikacji. W tym rozdziale przedstawimy podstawy przeprowadzania testów wydajnościowych za pomocą frameworka Locust, który jest oparty na składni Pythona.

Aby przeprowadzić testy wydajności, niezbędna jest działająca aplikacja oraz odpowiedni framework do testowania. W tym przypadku użyjemy Locust, frameworka, który bazuje na Pythonie. Szczegółowe informacje na temat Locust można znaleźć w dokumentacji oficjalnej pod adresem: https://docs.locust.io/en/stable/. Przed rozpoczęciem upewnij się, że masz zainstalowany Locust w swoim środowisku wirtualnym, wykonując polecenie:

ruby
Copy code
$ pip install locust

Po zainstalowaniu Locust możemy przejść do konfiguracji i uruchomienia instancji Locust w celu przeprowadzenia testu wydajności. W tym celu stworzymy plik locustfile.py w głównym katalogu projektu. Ten plik zdefiniuje zachowanie użytkowników w interakcji z naszą aplikacją, którą będziemy testować.

Minimalny przykład pliku locustfile.py wygląda następująco:

python
Copy code
from locust import HttpUser, task

class ProtoappUser(HttpUser):
    host = "http://localhost:8000"
    @task
    def hello_world(self):
        self.client.get("/home")

Konfiguracja określa klasę użytkownika, który będzie korzystał z naszego serwisu, a także adres hosta oraz endpoint, który ma zostać przetestowany. Następnie, uruchamiamy nasz serwer FastAPI za pomocą polecenia:

ruby
Copy code
$ uvicorn protoapp.main:app

Po uruchomieniu serwera, otwieramy nową instancję terminala i uruchamiamy Locust, wykonując polecenie:

ruby
Copy code
$ locust

Przechodzimy do przeglądarki i wchodzimy na adres: http://localhost:8089, aby uzyskać dostęp do interfejsu webowego Locust. Interfejs ten jest intuicyjny, dzięki czemu łatwo jest ustawić parametry testu:

1. Liczba równoczesnych użytkowników – określa maksymalną liczbę użytkowników, którzy będą korzystać z aplikacji w tym samym czasie.

2. Wskaźnik wzrostu użytkowników – ustala, jak szybko nowych użytkowników będzie dodawanych w trakcie testu, aby symulować rosnący ruch.

scss
Copy code
$ locust --headless --users 10 --spawn-rate 1

To polecenie uruchamia Locust w trybie headless, symulując:

• 10 użytkowników równocześnie korzystających z aplikacji,

• Wzrost liczby użytkowników o 1 osobę na sekundę.

Oprócz powyższych kroków, warto zwrócić uwagę na szereg istotnych kwestii związanych z testowaniem aplikacji pod kątem dużego obciążenia. Warto pamiętać, że testy wydajnościowe są najbardziej efektywne, gdy przeprowadzane są w realistycznych warunkach – na przykład przy symulacji różnorodnych scenariuszy użytkowania, takich jak zwiększony ruch, czy wielokrotne obciążenie serwera. Testy powinny być regularnie powtarzane, by mieć pewność, że aplikacja utrzymuje odpowiednią wydajność również po wprowadzeniu nowych funkcjonalności lub zmian w kodzie.

Ponadto warto również rozważyć zastosowanie różnych narzędzi do monitorowania wydajności, takich jak Prometheus, Grafana czy ELK stack (Elasticsearch, Logstash, Kibana), które pozwolą na dokładniejszą analizę i wizualizację wyników testów wydajnościowych. Testowanie wydajności powinno być integralną częścią procesu tworzenia oprogramowania, ponieważ pozwala na wczesne wykrycie problemów z wydajnością, zanim staną się one krytyczne.

Jak skonfigurować połączenie z bazą danych w FastAPI przy użyciu SQLAlchemy i asyncio?

Współczesne aplikacje webowe często wymagają efektywnego zarządzania bazami danych, zwłaszcza w kontekście używania asynchronicznych metod operacji, które zapewniają lepszą wydajność w przypadku dużych obciążeń. Jednym z najpopularniejszych frameworków do pracy z bazami danych w Pythonie jest SQLAlchemy, który wspiera zarówno tradycyjne, jak i asynchroniczne podejścia do operacji na bazach danych. Integracja FastAPI z SQLAlchemy i asyncio umożliwia tworzenie wydajnych aplikacji webowych. Oto jak skonfigurować takie połączenie krok po kroku.

Proces integracji FastAPI z bazą danych przy użyciu SQLAlchemy, a także asyncio, składa się z kilku etapów. Pierwszym z nich jest stworzenie odpowiednich klas mapujących obiekty na tabele bazy danych. W tym celu tworzymy plik database.py, w którym zdefiniujemy nasze modele bazodanowe. Przykład klasy Ticket, która będzie reprezentować tabelę w bazie danych, może wyglądać następująco:

python
Copy code
from sqlalchemy import Column, Float, ForeignKey, Table

from sqlalchemy.orm import (DeclarativeBase, Mapped, mapped_column)
class Base(DeclarativeBase):
    pass
class Ticket(Base):
    __tablename__ = "tickets"
    id: Mapped[int] = mapped_column(primary_key=True)
    price: Mapped[float] = mapped_column(nullable=True)
    show: Mapped[str | None]
    user: Mapped[str | None]

W tym przykładzie klasa Ticket odpowiada tabeli tickets, której kolumny odpowiadają atrybutom tej klasy. Kolumna id jest kluczem podstawowym, a price, show i user to dane związane z biletami.

Po utworzeniu klas mapujących, kolejnym krokiem jest stworzenie warstwy abstrakcji, która umożliwi komunikację z bazą danych za pomocą SQLAlchemy. Będzie to obejmować utworzenie silnika (engine) oraz sesji (session), które będą służyły do wykonywania operacji na danych. W tym celu tworzymy plik db_connection.py, w którym zdefiniujemy funkcję do uzyskania silnika oraz sesji asynchronicznej:

python
Copy code
from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy.orm import sessionmaker
SQLALCHEMY_DATABASE_URL = "sqlite+aiosqlite:///.database.db"
def get_engine():
    return create_async_engine(SQLALCHEMY_DATABASE_URL, echo=True)
from sqlalchemy.ext.asyncio import AsyncSession
AsyncSessionLocal = sessionmaker(
    autocommit=False,
    autoflush=False,
    bind=get_engine(),
    class_=AsyncSession,
)
async def get_db_session():
    async with AsyncSessionLocal() as session:
        yield session

W tym przypadku używamy bazy danych SQLite i obsługi asynchronicznej poprzez bibliotekę aiosqlite. Dzięki temu, nasze operacje na bazie danych będą wykonywane w sposób asynchroniczny, co pozwala na lepszą obsługę równoczesnych zapytań.

python
Copy code
from contextlib import asynccontextmanager
from fastapi import FastAPI
from app.database import Base
from app.db_connection import get_db_session
@asynccontextmanager
async def lifespan(app: FastAPI):
    engine = get_engine()
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield
    await engine.dispose()
app = FastAPI(lifespan=lifespan)

W powyższym kodzie zapewniamy, że przy starcie aplikacji wszystkie brakujące tabele w bazie danych zostaną utworzone, a po zakończeniu działania aplikacji połączenie z bazą zostanie zamknięte.

Po skonfigurowaniu połączenia z bazą danych, możemy przejść do implementacji operacji CRUD (Create, Read, Update, Delete). Operacje te są podstawą interakcji z bazą danych, a ich implementacja w FastAPI jest łatwa i intuicyjna.

Przykładowo, aby dodać nowy bilet do bazy danych, tworzymy odpowiednią funkcję:

python
Copy code
from sqlalchemy.ext.asyncio import AsyncSession

from sqlalchemy.future import select
from app.database import Ticket
async def create_ticket(
    db_session: AsyncSession,
    show_name: str,
    user: str = None,
    price: float = None,
) -> int:
    ticket = Ticket(show=show_name, user=user, price=price)
    async with db_session.begin():
        db_session.add(ticket)
        await db_session.flush()
        ticket_id = ticket.id
        await db_session.commit()
    return ticket_id

Funkcja ta przyjmuje dane nowego biletu, dodaje je do bazy danych i zwraca ID dodanego biletu. Podobnie, możemy zrealizować operacje do pobierania i aktualizowania danych:

python
Copy code
async def get_ticket(db_session: AsyncSession, ticket_id: int) -> Ticket | None:
    query = select(Ticket).where(Ticket.id == ticket_id)
    async with db_session as session:
        tickets = await session.execute(query)
        return tickets.scalars().first()
async def update_ticket_price(
    db_session: AsyncSession,
    ticket_id: int,
    new_price: float,
) -> bool:
    query = update(Ticket).where(Ticket.id == ticket_id).values(price=new_price)
    async with db_session as session:
        ticket_updated = await session.execute(query)
        await session.commit()
        if ticket_updated.rowcount == 0:
            return False
    return True

Podobne operacje można zaimplementować dla innych operacji CRUD, takich jak usuwanie rekordów czy aktualizowanie innych pól.

Warto dodać, że oprócz używania SQLite, SQLAlchemy wspiera wiele innych baz danych, takich jak MySQL, PostgreSQL czy MariaDB. Wystarczy zmienić odpowiedni ciąg połączenia (SQLALCHEMY_DATABASE_URL) na właściwy dla wybranej bazy danych, np. dla MySQL:

python
Copy code
mysql+aiomysql://user:password@host:port/dbname

Powyższa zmiana wymaga instalacji odpowiedniego sterownika (aiomysql) w środowisku.

Oczywiście, ważnym aspektem jest również obsługa błędów i zapewnienie, aby operacje na bazie danych były transakcyjne, co pozwala na zapewnienie integralności danych, nawet w przypadku nieoczekiwanych błędów aplikacji. Warto również pamiętać o optymalizacji zapytań, szczególnie w przypadku bardziej skomplikowanych operacji na dużych zbiorach danych.

Jak zintegrować FastAPI z bazą danych MongoDB: Przewodnik po ustawieniu połączenia i operacjach CRUD

Aby aplikacja FastAPI mogła komunikować się z MongoDB, musisz zainstalować pakiet motor, który jest asynchronicznym sterownikiem do MongoDB stworzonym przez MongoDB Inc. Instalację możesz przeprowadzić za pomocą polecenia:
pip install motor.

Po zainstalowaniu wymaganych pakietów, możemy przejść do konfiguracji połączenia z MongoDB w naszej aplikacji FastAPI. Pierwszym krokiem jest stworzenie folderu streaming_platform, a w nim podfolderu app, w którym umieścimy plik db_connection.py. Plik ten będzie odpowiedzialny za konfigurację połączenia z MongoDB. W tym pliku definiujemy klienta MongoDB za pomocą AsyncIOMotorClient:

python
Copy code
from motor.motor_asyncio import AsyncIOMotorClient
mongo_client = AsyncIOMotorClient("mongodb://localhost:27017")

Teraz możemy stworzyć funkcję, która będzie sprawdzać, czy MongoDB jest dostępne. W tym celu używamy logera uvicorn.error, który pozwoli nam śledzić stan połączenia:

python
Copy code
import logging

logger = logging.getLogger("uvicorn.error")
async def ping_mongo_db_server():
    try:
        await mongo_client.admin.command("ping")
        logger.info("Connected to MongoDB")
    except Exception as e:
        logger.error(f"Error connecting to MongoDB: {e}")
        raise e

Funkcja ping_mongo_db_server sprawdza, czy serwer MongoDB odpowiada na zapytanie. Jeśli serwer nie jest dostępny, wyrzuca błąd, który przerywa dalsze działanie aplikacji.

Kolejnym krokiem jest utworzenie pliku main.py, w którym za pomocą menedżera kontekstu lifespan uruchomimy funkcję ping_mongo_db_server podczas startu aplikacji FastAPI. Wartością lifespan jest zapewnienie, że sprawdzenie połączenia z bazą danych odbędzie się przed pełnym uruchomieniem serwera:

Jak definiować i używać modeli zapytań i odpowiedzi w FastAPI?

FastAPI, dzięki swojej prostocie i elastyczności, oferuje programistom bardzo wygodne narzędzia do definiowania i walidowania danych w aplikacjach webowych. Kluczowym elementem w tym procesie są modele Pydantic, które pozwalają na dokładne określenie struktury danych, zarówno dla zapytań, jak i odpowiedzi. W tej sekcji omówimy, jak tworzyć modele, jak zarządzać danymi w aplikacji oraz jak zapewnić ich poprawność i zgodność ze zdefiniowanymi schematami.

python
Copy code
from pydantic import BaseModel
class Book(BaseModel):
    title: str
    author: str
    year: int

W tym przypadku Book jest klasą pochodzącą od BaseModel, która definiuje trzy atrybuty: tytuł, autora i rok wydania, przy czym każdy z tych atrybutów ma określony typ danych.

W FastAPI modele Pydantic pełnią dwie główne funkcje. Po pierwsze, służą do walidacji danych wejściowych, na przykład gdy użytkownik przesyła dane za pomocą zapytania POST. Po drugie, umożliwiają generowanie odpowiedzi, które są zgodne z określonym schematem. Na przykład, jeśli mamy endpoint do dodawania książek, możemy zdefiniować endpoint, który przyjmie dane w formacie JSON, a FastAPI automatycznie sprawdzi, czy te dane są zgodne z naszym modelem:

python
Copy code
@app.post("/book")

async def create_book(book: Book):
    return book

W tym przypadku, jeśli użytkownik wyśle zapytanie POST z danymi w formacie JSON, które nie będą spełniały wymagań modelu (np. brakujący tytuł lub nieprawidłowy rok wydania), FastAPI automatycznie zwróci odpowiedź o błędzie. Ta walidacja odbywa się dzięki Pydantic.

Jednak Pydantic oferuje znacznie więcej funkcji walidacji, które umożliwiają dokładniejsze doprecyzowanie, jakie dane mogą zostać przekazane do aplikacji. Przykładem może być dodanie reguł dotyczących długości pól tekstowych czy zakresu liczb:

python
Copy code
from pydantic import BaseModel, Field

class Book(BaseModel):
    title: str = Field(..., min_length=1, max_length=100)
    author: str = Field(..., min_length=1, max_length=50)
    year: int = Field(..., gt=1900, lt=2100)

W tym przykładzie używamy funkcji Field, aby dodać ograniczenia: minimalną i maksymalną długość tytułu i autora oraz zakres lat dla roku wydania książki. Dzięki temu aplikacja jest w stanie odrzucić dane, które nie spełniają tych warunków.

Oczywiście w wielu przypadkach konieczne jest dostosowanie odpowiedzi API do potrzeb klienta. Często chcemy zwrócić tylko część danych lub zmienić format odpowiedzi. FastAPI umożliwia stworzenie modeli odpowiedzi, które precyzyjnie definiują, jakie dane mają zostać zwrócone. Przykład:

python
Copy code
class BookResponse(BaseModel):

    title: str
    author: str
@app.get("/allbooks", response_model=List[BookResponse])
async def read_all_books():
    return [
        {"title": "1984", "author": "George Orwell"},

        {"title": "The Great Gatsby", "author": "F. Scott Fitzgerald"}

    ]

Tutaj definiujemy model BookResponse, który zawiera tylko tytuł i autora książki, a w odpowiedzi na zapytanie GET zwracamy listę książek w tym formacie. Dzięki temu klient otrzymuje dane w odpowiedniej postaci, bez zbędnych informacji, jak np. rok wydania.

Modele odpowiedzi są szczególnie przydatne, gdy chcemy w sposób bardziej kontrolowany zarządzać danymi w odpowiedziach API, na przykład ukrywając informacje, które mogą być wrażliwe, lub dostosowując odpowiedź do konkretnego formatu oczekiwanego przez użytkownika.

FastAPI nie ogranicza się jednak tylko do prostych modeli. Można tworzyć bardziej zaawansowane struktury danych, które zawierają zagnieżdżone modele lub nawet listy innych modeli. Pydantic pozwala na tworzenie złożonych schematów danych, które są zarówno łatwe do zdefiniowania, jak i bardzo efektywne pod względem wydajności. Oczywiście, zależnie od wymagań aplikacji, można również definiować domyślne wartości, sprawdzać zakresy, stosować wyrażenia regularne i inne mechanizmy walidacji, które pomagają w jeszcze dokładniejszym zarządzaniu danymi.

Ważnym aspektem w pracy z FastAPI i Pydantic jest również możliwość testowania naszych endpointów. Można to robić za pomocą Swagger UI lub Postman, które automatycznie generują interfejsy pozwalające na łatwe testowanie endpointów. Dzięki nim, wystarczy wprowadzić odpowiednie parametry (np. author_id lub year w zapytaniach GET), aby sprawdzić, czy aplikacja zwraca poprawne odpowiedzi. Testowanie za pomocą tych narzędzi jest szybkie, intuicyjne i pozwala na natychmiastowe wykrywanie błędów w implementacji.

Praca z FastAPI i Pydantic to ogromna zaleta, jeśli chodzi o tworzenie bezpiecznych, szybkich i skalowalnych aplikacji. Dzięki prostocie składni i potężnym funkcjom walidacji oraz zarządzania danymi, programiści mogą szybko budować aplikacje, które nie tylko działają, ale również są zgodne z określonymi normami i zasadami. Modelowanie danych w FastAPI pozwala na utrzymanie porządku i spójności w aplikacji, jednocześnie minimalizując ryzyko wystąpienia błędów związanych z nieprawidłowymi danymi.