Аутентификация в Flask с JWT

Ноутбук на сером столе с редактором кода, два столбца кода на экране.

Зачем использовать токены и JWT

Token-based authentication (аутентификация на основе токенов) — это способ подтверждения прав доступа, при котором сервер выдает клиенту краткоживущий или долгоживущий зашифрованный маркер (токен). Клиент прикрепляет этот токен к последующим запросам. JWT (JSON Web Token) — компактный, самодостаточный формат токена, содержащий полезную нагрузку (claims), заголовок и цифровую подпись.

Коротко — что такое JWT (одна строка): JWT — это строка с тремя частями: header, payload и signature, разделёнными точками; подпись защищает от подделки.

Пример закодированного JWT слева и декодированного с выделением заголовка, полезной нагрузки и подписи справа.

Основные преимущества JWT:

Самодостаточность: токен сам содержит нужную информацию (user_id, роли и т. п.).
Масштабируемость: можно избегать серверных сессий при правильной архитектуре.
Кросс-платформенность: стандарт JSON, легко использовать в web, mobile, microservices.

Когда JWT не подходит:

Невозможность эффективной немедленной отзыва токена без вспомогательных механизмов.
Хранение большого объёма сессионных данных в токене — неэффективно.

Структура проекта и подготовка окружения

Создайте директорию проекта и виртуальное окружение.

mkdir flask-project
cd flask-project
python -m venv venv

Активируйте окружение:

# Unix / macOS
source venv/bin/activate

# Windows (PowerShell)
.\venv\Scripts\Activate.ps1

Создайте файл requirements.txt и добавьте зависимости:

flask
pyjwt
python-dotenv
pymongo
bcrypt

Установите пакеты:

pip install -r requirements.txt

Создайте MongoDB — локально или в MongoDB Atlas — и сохраните URI в .env:

MONGO_URI=""
SECRET_KEY="<ваш-секретный-ключ-длиной-не-меньше-32-байт>"
JWT_EXP_DELTA_SECONDS=3600

Примечание: храните SECRET_KEY в защищённом хранилище окружения и не генерируйте его динамически при каждом запуске (os.urandom при запуске приведёт к недействительности ранее выпущенных токенов).

Подключение к MongoDB

Создайте файл utils/db.py:

from pymongo import MongoClient
import os


def connect_to_mongodb(mongo_uri: str):
    client = MongoClient(mongo_uri)
    db = client.get_database()  # использует базу из URI или по умолчанию
    return db

Обратите внимание: get_database() вернёт базу, указанную в URI. Можно также явно выбрать db = client[‘users_db’].

Создание сервера Flask

app.py:

from flask import Flask
from routes.user_auth import register_routes
from utils.db import connect_to_mongodb
from dotenv import load_dotenv
import os

load_dotenv()
app = Flask(__name__)

app.config['MONGO_URI'] = os.getenv('MONGO_URI')
app.config['SECRET_KEY'] = os.getenv('SECRET_KEY')
app.config['JWT_EXP_DELTA_SECONDS'] = int(os.getenv('JWT_EXP_DELTA_SECONDS', 3600))

db = connect_to_mongodb(app.config['MONGO_URI'])
register_routes(app, db)

if __name__ == '__main__':
    app.run(host='0.0.0.0', debug=True)

Важно: debug=True не используйте в production.

Модель пользователя

Создайте файл models/user_model.py:

from pymongo.collection import Collection
from bson.objectid import ObjectId


class User:
    def __init__(self, collection: Collection, username: str, password: str):
        self.collection = collection
        self.username = username
        self.password = password

    def save(self):
        user_data = {
            'username': self.username,
            'password': self.password
        }
        result = self.collection.insert_one(user_data)
        return str(result.inserted_id)

    @staticmethod
    def find_by_id(collection: Collection, user_id: str):
        return collection.find_one({'_id': ObjectId(user_id)})

    @staticmethod
    def find_by_username(collection: Collection, username: str):
        return collection.find_one({'username': username})

Кратко: модель инкапсулирует доступ к коллекции пользователей.

Маршруты аутентификации (routes/user_auth.py)

Ниже — полноценный и безопаснее оформленный пример маршрутов регистрации, входа, декоратора проверки токена и защищённого маршрута.

import jwt
from functools import wraps
from flask import jsonify, request, make_response, current_app
from models.user_model import User
import bcrypt
import datetime


def register_routes(app, db):
    collection = db.users

    def token_required(f):
        @wraps(f)
        def decorated(*args, **kwargs):
            token = request.cookies.get('access_token') or request.headers.get('Authorization')
            if token and token.startswith('Bearer '):
                token = token.split(' ', 1)[1]

            if not token:
                return jsonify({'message': 'Token is missing!'}), 401

            try:
                payload = jwt.decode(token, app.config['SECRET_KEY'], algorithms=['HS256'])
                current_user = User.find_by_id(collection, payload['user_id'])
            except jwt.ExpiredSignatureError:
                return jsonify({'message': 'Token has expired!'}), 401
            except jwt.InvalidTokenError:
                return jsonify({'message': 'Invalid token!'}), 401

            return f(current_user, *args, **kwargs)

        return decorated

    @app.route('/api/register', methods=['POST'])
    def register():
        data = request.get_json() or {}
        username = data.get('username')
        password = data.get('password')

        if not username or not password:
            return jsonify({'message': 'Username and password are required.'}), 400

        if User.find_by_username(collection, username):
            return jsonify({'message': 'Username already exists!'}), 409

        # Политика паролей: минимум 8 символов (пример)
        if len(password) < 8:
            return jsonify({'message': 'Password must be at least 8 characters.'}), 400

        hashed_password = bcrypt.hashpw(password.encode('utf-8'), bcrypt.gensalt())
        new_user = User(collection, username, hashed_password.decode('utf-8'))
        user_id = new_user.save()

        return jsonify({'message': 'User registered successfully!', 'user_id': user_id}), 201

    @app.route('/api/login', methods=['POST'])
    def login():
        data = request.get_json() or {}
        username = data.get('username')
        password = data.get('password')

        if not username or not password:
            return jsonify({'message': 'Username and password are required.'}), 400

        user = User.find_by_username(collection, username)
        if user and bcrypt.checkpw(password.encode('utf-8'), user['password'].encode('utf-8')):
            payload = {
                'user_id': str(user['_id']),
                'exp': datetime.datetime.utcnow() + datetime.timedelta(seconds=app.config['JWT_EXP_DELTA_SECONDS'])
            }
            token = jwt.encode(payload, app.config['SECRET_KEY'], algorithm='HS256')

            # Безопасная установка cookie
            response = make_response(jsonify({'message': 'Login successful!'}))
            response.set_cookie('access_token', token, httponly=True, secure=True, samesite='Lax')
            return response

        return jsonify({'message': 'Invalid username or password'}), 401

    @app.route('/api/users', methods=['GET'])
    @token_required
    def get_users(current_user):
        users = list(collection.find({}, {'password': 0}))
        for u in users:
            u['_id'] = str(u['_id'])
        return jsonify(users), 200

Ключевые улучшения по сравнению с базовой реализацией:

Используется SECRET_KEY из окружения.
Токен имеет время жизни exp.
Токен можно передавать либо в заголовке Authorization: Bearer , либо в защищённой cookie.
Cookie устанавливается с флагами httponly и secure, samesite.
Пароли хранятся в bcrypt.
Ответы имеют соответствующие коды состояния HTTP.

Безопасность: рекомендации и hardening

Хранение секретов

SECRET_KEY храните в защищённом хранилище окружения (Vault, AWS Secrets Manager, Azure Key Vault).
Не генерируйте SECRET_KEY на лету в рабочем окружении.

HTTPS

Всегда используйте HTTPS в production. Флаги secure у cookies требуют HTTPS.

CSRF

Если вы храните JWT в cookie, защититесь от CSRF: используйте SameSite=strict/lax и CSRF-токен для state-changing запросов.
Альтернатива: храните access token в памяти клиента и передавайте в заголовке Authorization.

Refresh tokens

Реализуйте refresh token с длительным сроком жизни, храня его в безопасном месте (HttpOnly cookie) и имея endpoint для его обмена на новый access token.
Храните refresh token в базе для возможности отзыва.

Механизм отзыва

Для немедленного аннулирования токенов используйте blacklist/whitelist в базе (например, коллекция revoked_tokens с ttl-index).
Альтернатива — хранить версию токена в профиле пользователя (token_version) и проверять её в payload.

Минимизация полезной нагрузки

Не кладите в JWT чувствительные данные (пароли, PII). Лучше — идентификатор и минимальные разрешения.

Ограничение длительности

Access token — короткоживущий (например, 15–60 минут).
Refresh token — дольше (дни/недели), но с возможностью отзыва.

Логи и мониторинг

Логируйте неуспешные попытки входа и подозрительные активности.
Настройте оповещения при аномалиях (резкий рост неудачных логинов).

Политики паролей и MFA

Внедрите политику сложности паролей и многофакторную аутентификацию (MFA) для чувствительных операций.

Различные подходы и когда использовать

Сеансовая аутентификация (server-side sessions): простая, хороша для традиционных web-приложений, где много серверного состояния.
JWT: отлично подходит для API, SPA, мобильных приложений и распределённых систем.
OAuth2 / OpenID Connect: используйте, когда нужна авторизация через третьи стороны (Google, GitHub) или централизованная IAM.

Выбор зависит от требований к отзывам токенов, масштабируемости и наличию централизованной авторизации.

Тесты и критерии приёмки

Критерии приёмки

Регистрация создает пользователя и возвращает 201.
Логин с правильными данными возвращает cookie с access_token и 200.
Доступ к /api/users без токена — 401.
Доступ с корректным токеном — 200 и список пользователей без полей password.

Тест-кейсы

Успешная регистрация: POST /api/register с валидными username/password — ожидается 201 и id.
Повторная регистрация same username — ожидается 409.
Логин с неверным паролем — 401.
Попытка доступа к защищённому ресурсу без token — 401.
Попытка доступа с просроченным token — 401.
Проверка флагов cookie (httponly, secure, samesite) — для браузерного запроса.

Автоматизация: напишите модульные тесты для логики проверки токена и интеграционные тесты для endpoints с использованием pytest + requests.

Роли и чек-листы (для команды)

Разработчик

Использовать SECRET_KEY из окружения.
Добавить exp в payload токена.
Не записывать пароли в лог.

Security Engineer

Провести ревью кода на предмет уязвимостей.
Настроить хранение секретов и HTTPS.

Написать тесты на все кейсы регистрации/логина/доступа.
Проверить обработку ошибок и статусы HTTP.

DevOps

Настроить переменные окружения и CI/CD для секретов.
Обеспечить мониторинг и алерты на аномалии логов.

Модель зрелости и шаги внедрения

Минимально жизнеспособный продукт (MVP)

Регистрация и вход, JWT с exp, хранение пароля в bcrypt.

Производство (production-ready)

SECRET_KEY в секретном хранилище, HTTPS, cookie flags, короткоживущие access tokens.

Полный уровень защиты

Refresh tokens, механизм отзыва/blacklist, MFA, аудит логов и автоматизированное тестирование.

Примеры обходных ситуаций и нюансы

Если токен хранится в localStorage, XSS уязвимость может привести к утечке. Предпочтительнее HttpOnly cookie + CSRF защита.
При горизонтальном масштабировании убедитесь, что проверка токена не зависит от состояния одного инстанса (если используется blacklist, он доступен в общей БД).

Принцип минимальных привилегий

Давайте всегда выдавливаем минимально необходимую информацию в токен: user_id и роль/scope. Для дополнительных данных делайте запросы к сервису с проверкой прав.

Краткая инструкция по разворачиванию в локальной среде

Клонируйте репозиторий проекта.
Создайте venv и установите зависимости.
Создайте .env с MONGO_URI и SECRET_KEY.
Запустите MongoDB или подключитесь к Atlas.
Запустите приложение: flask run или python app.py.
Используйте Postman/HTTPie/cURL для тестирования endpoint-ов.

Чек-лист безопасности перед релизом

SECRET_KEY в безопасном хранилище
HTTPS включён
Access token короткоживущий
Refresh token реализован и можно отзывать
Логи и алерты настроены
Автотесты покрывают ключевые сценарии

Короткая методология внедрения (5 шагов)

Реализовать базовый flow: register/login, JWT с exp.
Добавить хранение секретов и HTTPS.
Внедрить refresh tokens и механизм отзыва.
Сформировать политики паролей и MFA.
Тестирование, мониторинг, ревью и выпуск.

Глоссарий (1 строка на термин)

JWT: формат токена с подписью (header.payload.signature).
Access token: токен короткой жизни для доступа к API.
Refresh token: токен для получения новых access token’ов.
SECRET_KEY: секрет для подписи JWT.

Заключение

JWT — удобный и распространённый способ аутентификации для API и SPA, но сам по себе не решает всех задач безопасности. Правильная архитектура включает: безопасное хранение секретов, короткоживущие access token’ы, refresh token’ы с поддержкой отзыва, HTTPS, CSRF-защиту, аудит и мониторинг.

Важно: реализуйте и тестируйте сценарии отзыва, протоколируйте неуспешные попытки и применяйте принцип минимальных привилегий.

Ниже — краткое напоминание основных команд для локальной проверки:

# Создать виртуальное окружение
python -m venv venv
source venv/bin/activate
pip install -r requirements.txt
flask run

Внимание: для production настройте Gunicorn/uWSGI и обратный прокси (Nginx) с SSL.

Аутентификация в Flask с JWT: практическое руководство