Специалист по Code Review

Проводит профессиональный код-ревью с оценкой читаемости, SOLID, безопасности и приоритизированными рекомендациями по рефакторингу.

// промпт
Ты — старший инженер-ревьюер с многолетним опытом в {{iazyk_programmirovaniia}} и большим стажем code review в продуктовых командах. Проведи строгий, но конструктивный разбор кода ниже, сохранив его исходную функциональность. ## Контекст - **Язык / стек:** {{iazyk_programmirovaniia}} - **Назначение кода:** {{cto_delaet_kod}} - **Окружение / ограничения:** {{kontekst_i_ograniceniia}} - **Фокус ревью:** {{prioritet_reviu}} ## Что проанализировать 1. **Читаемость и поддерживаемость** — нейминг, структура, связность, дублирование, качество комментариев. 2. **Принципы SOLID и архитектура** — единственная ответственность, открытость/закрытость, подстановка Лисков, разделение интерфейсов, инверсия зависимостей. 3. **Корректность** — граничные случаи, обработка ошибок, гонки, потенциальные баги. 4. **Производительность** — алгоритмическая сложность, лишние аллокации, узкие места. 5. **Безопасность** — валидация ввода, утечки данных, типовые уязвимости (инъекции, небезопасные зависимости). ## Формат ответа **Общая оценка:** балл от 1 до 10 с краткой мотивировкой. ### Сильные стороны Маркированный список того, что сделано хорошо. ### Найденные проблемы Для каждой проблемы: - **Категория** (читаемость / SOLID / корректность / производительность / безопасность) - **Серьёзность:** Критичная / Высокая / Средняя / Низкая - **Описание** и почему это важно - **Фрагмент** с проблемным местом и предлагаемый исправленный вариант кода Отсортируй проблемы по убыванию серьёзности. ### Рекомендации по рефакторингу Конкретные, применимые шаги: что переименовать, что выделить, какие паттерны или тесты добавить. ## Правила - Опирайся только на предоставленный код; если контекста не хватает — явно перечисли допущения. - Давай конкретные примеры вместо общих фраз. Тон — уважительный и нацеленный на улучшение. ### Код для ревью ``` {{kod_dlia_reviu}} ```
Заполните переменные

Пример запуска

Сгенерировано Claude Opus 5.5, 24 сентября 2026 г.

Значения переменных

Язык программирования
Python 3.11 (FastAPI)
Что делает код
Это модуль регистрации и аутентификации пользователей в бэкенде сервиса заказа корпоративных обедов «LunchHub». Класс UserService инкапсулирует создание нового пользователя, проверку логина/пароля при входе и обновление профиля. Данные о пользователях хранятся в PostgreSQL, доступ к БД идёт через синхронный psycopg2-коннектор, который прокидывается в сервис через конструктор. Модуль вызывается из роутов FastAPI (/auth/register, /auth/login, /users/{id}) и используется примерно 300 корпоративных клиентов, каждый со своим доменом почты. Пароли раньше хранились в открытом виде в тестовой БД, сейчас идёт миграция на хэширование — этот код является частью PR, где как раз добавляется хэширование пароля и минимальная валидация email.
Контекст и ограничения
Сервис крутится в Docker-контейнере на Kubernetes (2 пода, без sticky session), под нагрузкой около 50 RPS в пиковые обеденные часы (11:30–13:00 по будням). БД — управляемый PostgreSQL 14 в Yandex Cloud, connection pool ограничен 20 соединениями на под. Команда — 4 бэкенд-разработчика, релизы раз в неделю по вторникам. Обязательные ограничения: код должен работать под Python 3.11 без сторонних тяжёлых зависимостей (уже используется passlib и pydantic, добавлять новые библиотеки нежелательно без согласования с тимлидом). Нужно сохранить обратную совместимость со старыми хэшами паролей (bcrypt), сгенерированными предыдущей версией сервиса. Дедлайн — до конца следующей недели, так как аудит безопасности от клиента-партнёра назначен на начало октября.
Приоритет ревью
Безопасность и корректность обработки паролей/входных данных
Код для ревью
import hashlib import psycopg2 class UserService: def __init__(self, db_conn): self.conn = db_conn def register_user(self, email, password, name): cursor = self.conn.cursor() query = "SELECT * FROM users WHERE email = '%s'" % email cursor.execute(query) existing = cursor.fetchone() if existing: return {"error": "User already exists"} password_hash = hashlib.md5(password.encode()).hexdigest() insert_query = "INSERT INTO users (email, password_hash, name) VALUES ('%s', '%s', '%s')" % ( email, password_hash, name ) cursor.execute(insert_query) self.conn.commit() return {"status": "ok"} def login(self, email, password): cursor = self.conn.cursor() query = "SELECT id, password_hash FROM users WHERE email = '%s'" % email cursor.execute(query) row = cursor.fetchone() if row is None: return None user_id, stored_hash = row password_hash = hashlib.md5(password.encode()).hexdigest() if password_hash == stored_hash: return {"user_id": user_id, "token": str(user_id) + "_" + password_hash} else: return None def update_profile(self, user_id, new_name): cursor = self.conn.cursor() query = "UPDATE users SET name = '%s' WHERE id = %s" % (new_name, user_id) cursor.execute(query) self.conn.commit() return {"status": "updated"} def get_user_by_email(self, email): cursor = self.conn.cursor() query = "SELECT id, email, name FROM users WHERE email = '%s'" % email cursor.execute(query) row = cursor.fetchone() if row: return {"id": row[0], "email": row[1], "name": row[2]} return None

Общая оценка: 2/10. Код компактный и понятный по структуре, но в текущем виде его нельзя мёржить. В нём есть SQL-инъекции во всех четырёх методах. Пароли хэшируются через MD5 без соли. Токен сессии строится из хэша пароля. Кроме того, PR ломает вход всем пользователям со старыми bcrypt-хэшами, а обещанной валидации email в коде нет. Перед аудитом каждая из этих проблем по отдельности даёт критичную находку.


Допущения

  1. Проверка токена выполняется в другом модуле (скорее всего, dependency в FastAPI), и её код здесь не показан.
  2. Авторизацию в /users/{id} (право текущего пользователя менять профиль id) делает роут. Если это не так, проблема №8 становится критичной.
  3. Колонка users.email не имеет уникального индекса, иначе не было бы ручной проверки SELECT ... WHERE email.
  4. Роуты вызывают сервис синхронно или через run_in_threadpool. Соединение берётся из пула на время запроса.
  5. Под «старыми хэшами bcrypt» понимаются строки вида $2b$12$... в users.password_hash в проде.

Сильные стороны

  • Внедрение зависимости через конструктор. Соединение передаётся в UserService, а не создаётся внутри, поэтому сервис легко тестировать.
  • Узкие выборки. В login и get_user_by_email выбираются только нужные колонки (кроме register_user с SELECT *).
  • Одинаковый ответ при входе. login возвращает None и при неизвестном email, и при неверном пароле, то есть по ответу нельзя понять, существует ли пользователь. Ниже показано, что по времени ответа это всё же различимо.
  • get_user_by_email не отдаёт password_hash. Это правильный инстинкт.
  • Явный commit() после записи. Намерение управлять транзакцией есть, осталось сделать его надёжным.

Найденные проблемы

1. SQL-инъекция во всех запросах
  • Категория: безопасность
  • Серьёзность: Критичная

Все запросы собираются через %-форматирование строк. Примеры атак:

  • email ' OR '1'='1' -- в get_user_by_email вернёт первого пользователя из таблицы;
  • x' UNION SELECT id, password_hash FROM users LIMIT 1 -- в login позволяет подсунуть известный хэш;
  • в update_profile имя x', password_hash = '<md5 известного пароля> даёт захват аккаунта;
  • user_id подставляется вообще без кавычек, поэтому в WHERE id = 1 OR 1=1 меняются имена всех пользователей.

Кавычки вокруг '%s' не защищают. psycopg2 умеет безопасно передавать параметры, и этим нужно пользоваться.

# Было
query = "SELECT id, password_hash FROM users WHERE email = '%s'" % email
cursor.execute(query)

# Стало: параметр передаётся драйверу, кавычки в SQL НЕ ставятся
cursor.execute(
    "SELECT id, password_hash FROM users WHERE email = %s",
    (email,),
)

Важно: "... %s" % email и execute("... %s", (email,)) выглядят похоже, но делают принципиально разное. Стоит добавить в CI линтер, который ловит это различие: bandit (правило B608) или ruff с правилом S608. Ruff, скорее всего, уже есть в проекте.


2. MD5 без соли вместо bcrypt
  • Категория: безопасность

  • Серьёзность: Критичная

  • MD5 считается на GPU со скоростью порядка десятков миллиардов хэшей в секунду. 8-символьный пароль перебирается за минуты.

  • Соли нет. Одинаковые пароли дают одинаковые хэши, а по готовым rainbow-таблицам популярные пароли вскрываются мгновенно.

  • passlib уже есть в зависимостях, поэтому переход на bcrypt не требует согласования новых библиотек.

# Было
password_hash = hashlib.md5(password.encode()).hexdigest()

# Стало
from passlib.context import CryptContext

pwd_context = CryptContext(
    schemes=["bcrypt"],
    deprecated="auto",
    bcrypt__rounds=12,
)

password_hash = pwd_context.hash(password)   # "$2b$12$<соль><хэш>"

3. PR ломает вход всем существующим пользователям с bcrypt-хэшами
  • Категория: корректность
  • Серьёзность: Критичная

В проде хранятся хэши вида $2b$12$.... Новый login сравнивает с ними md5(password), и это сравнение никогда не пройдёт. После релиза во вторник ни один старый пользователь не сможет войти. Это прямо нарушает требование обратной совместимости.

CryptContext.verify_and_update решает сразу три задачи:

  • проверяет старые bcrypt-хэши;
  • даёт единый путь для будущей смены алгоритма;
  • возвращает новый хэш, если старый устарел (например, был посчитан с меньшим rounds).
# Было
password_hash = hashlib.md5(password.encode()).hexdigest()
if password_hash == stored_hash:
    ...

# Стало
try:
    is_valid, upgraded_hash = pwd_context.verify_and_update(password, stored_hash)
except ValueError:
    # Некорректный формат хэша (например, остаток открытого пароля из тестовой БД)
    is_valid, upgraded_hash = False, None

if is_valid and upgraded_hash:
    repo.update_password_hash(user_id, upgraded_hash)   # прозрачный rehash при входе

Про открытые пароли из тестовой БД: не добавляйте схему plaintext в CryptContext ради совместимости. Такие записи нужно один раз перехэшировать миграционным скриптом или принудительно сбросить.


4. Токен сессии предсказуем и фактически является хэшем пароля
  • Категория: безопасность
  • Серьёзность: Критичная
return {"user_id": user_id, "token": str(user_id) + "_" + password_hash}

Проблемы этого решения:

  • Утечка хэша. Токен содержит md5(password), а он лежит в логах, в localStorage и в заголовках. Из токена восстанавливается пароль (см. №2), который пользователи часто повторяют на других сервисах.
  • Токен не отзывается и не истекает. Он действует бессрочно, пока пользователь не сменит пароль. Выйти из аккаунта нельзя.
  • Токен можно подделать. Злоумышленник, у которого есть дамп БД или результат инъекции из №1, строит "<id>_<hash>" для любого пользователя, не зная паролей. Это классический pass-the-hash.

Решение без новых зависимостей: непрозрачный случайный токен. В таблице сессий хранится его SHA-256, срок жизни и отметка об отзыве. Два пода без sticky session здесь не мешают, потому что состояние хранится в общей БД.

import hashlib
import secrets
from datetime import datetime, timedelta, timezone

SESSION_TTL = timedelta(hours=12)

def issue_session_token(cur, user_id: int) -> str:
    token = secrets.token_urlsafe(32)                       # 256 бит энтропии
    token_hash = hashlib.sha256(token.encode()).hexdigest() # в БД — только хэш
    cur.execute(
        """
        INSERT INTO user_sessions (token_hash, user_id, expires_at)
        VALUES (%s, %s, %s)
        """,
        (token_hash, user_id, datetime.now(timezone.utc) + SESSION_TTL),
    )
    return token

Здесь SHA-256 уместен: токен случайный и высокоэнтропийный, перебирать его бессмысленно. Если команда предпочитает JWT, понадобится PyJWT, а это новая зависимость и её нужно согласовать с тимлидом. Кроме того, отзыв JWT-токенов всё равно требует хранилища.


5. Timing-атаки: сравнение хэшей и перечисление пользователей
  • Категория: безопасность

  • Серьёзность: Высокая

  • Оператор == на строках сравнивает символы до первого расхождения, поэтому время ответа зависит от длины совпадающего префикса.

  • Если email не найден, login отвечает сразу. Если найден, считается хэш. С bcrypt разница составит 100–250 мс, и по ней легко узнать, какие адреса зарегистрированы. Для B2B-сервиса это полноценная утечка списка сотрудников клиентов.

# Модульная константа: хэш случайной строки, вычисляется один раз
_DUMMY_HASH = pwd_context.hash(secrets.token_urlsafe(16))

row = repo.get_credentials(email)
stored_hash = row.password_hash if row else _DUMMY_HASH
is_valid, upgraded = pwd_context.verify_and_update(password, stored_hash)  # постоянное время
if row is None or not is_valid:
    raise InvalidCredentials()

6. Гонка при регистрации и дубли из-за регистра email
  • Категория: корректность

  • Серьёзность: Высокая

  • Схема «сначала SELECT, потом INSERT» не атомарна. Два пода или два потока могут одновременно не найти пользователя и оба выполнить вставку. Двойной клик по кнопке «Зарегистрироваться» воспроизводит это без всякой злонамеренности.

  • Адреса Ivan@Corp.ru и ivan@corp.ru становятся двумя разными аккаунтами.

Надёжная защита делается на уровне БД. Код должен ловить нарушение уникальности, а не проверять его заранее:

-- миграция; перед ней проверить существующие дубли:
-- SELECT lower(email), count(*) FROM users GROUP BY 1 HAVING count(*) > 1;
CREATE UNIQUE INDEX CONCURRENTLY users_email_lower_uq ON users (lower(email));
from psycopg2 import errors

try:
    with self._conn, self._conn.cursor() as cur:
        cur.execute(
            "INSERT INTO users (email, password_hash, name) VALUES (%s, %s, %s) RETURNING id",
            (email, password_hash, name),
        )
        user_id = cur.fetchone()[0]
except errors.UniqueViolation:
    raise UserAlreadyExists() from None

Email нужно нормализовать через strip().lower() при регистрации, при входе и при поиске.


7. Нет обработки ошибок транзакций, курсоры не закрываются
  • Категория: корректность
  • Серьёзность: Высокая

Если execute выбросит исключение (нарушение ограничения, таймаут, инъекция, сломавшая синтаксис), rollback() не вызывается. Соединение остаётся в состоянии InFailedSqlTransaction, возвращается в пул, и все следующие запросы на нём падают с ошибкой current transaction is aborted. На пуле из 20 соединений под обеденной нагрузкой это быстро превращается в каскадные 500-е.

Кроме того, login и get_user_by_email открывают транзакцию, но никогда её не закрывают. Соединения висят в состоянии idle in transaction.

# Было
cursor = self.conn.cursor()
cursor.execute(...)
self.conn.commit()

# Стало: `with conn` — commit при успехе / rollback при исключении;
#         `with cursor` — гарантированное закрытие курсора
with self._conn, self._conn.cursor() as cur:
    cur.execute(...)

with conn в psycopg2 не закрывает соединение, он управляет только транзакцией. Для соединений из пула это именно то, что нужно.


8. update_profile: нет валидации, нет проверки результата, риск IDOR
  • Категория: корректность / безопасность

  • Серьёзность: Высокая (Критичная, если роут не проверяет владельца)

  • Метод возвращает "updated", даже если пользователя с таким id нет (rowcount == 0).

  • Имя можно передать пустым, длиной в мегабайт или содержащим управляющие символы.

  • Сервис никак не связывает user_id с вызывающим пользователем. Если /users/{id} проверяет только наличие токена, любой может переименовать любого (IDOR).

def update_profile(self, user_id: int, new_name: str) -> None:
    name = validate_name(new_name)
    with self._conn, self._conn.cursor() as cur:
        cur.execute("UPDATE users SET name = %s WHERE id = %s", (name, user_id))
        if cur.rowcount == 0:
            raise UserNotFound(user_id)

В роуте:

@router.patch("/users/{user_id}")
def update_user(user_id: int, body: ProfileUpdate, current_user: CurrentUser = Depends(get_current_user)):
    if current_user.id != user_id:
        raise HTTPException(status.HTTP_403_FORBIDDEN)
    ...

Аннотация user_id: int в роуте дополнительно отсекает нечисловые значения.


9. Валидации email и пароля нет, хотя PR её заявляет
  • Категория: безопасность / корректность

  • Серьёзность: Высокая

  • Email. Проверки нет совсем. Pydantic-тип EmailStr требует пакета email-validator, то есть это новая зависимость, её нужно согласовать. До согласования хватит минимальной проверки без зависимостей, указанной ниже.

  • Пароль. Нет минимальной длины: пустой пароль "" сейчас регистрируется.

  • Ограничение bcrypt в 72 байта. Bcrypt учитывает только первые 72 байта пароля. Кириллический символ в UTF-8 занимает 2 байта, так что это примерно 36 русских букв. Новые версии пакета bcrypt (5.x) на длинном пароле выбрасывают ValueError, а это необработанное исключение и 500 при регистрации. С passlib 1.7.4 связка bcrypt>=4.1 выдаёт предупреждение, а 5.x может сломать инициализацию backend. Проверьте, какая версия закреплена в requirements/poetry.lock, и зафиксируйте bcrypt<5 явно.

import re

_EMAIL_RE = re.compile(r"[A-Za-z0-9._%+\-]+@[A-Za-z0-9\-]+(?:\.[A-Za-z0-9\-]+)+")
MAX_EMAIL_LEN = 254
MIN_PASSWORD_LEN = 10
BCRYPT_MAX_BYTES = 72

def normalize_email(raw: str) -> str:
    email = raw.strip().lower()
    if len(email) > MAX_EMAIL_LEN or not _EMAIL_RE.fullmatch(email):
        raise ValidationError("Некорректный email")
    return email

def validate_password(password: str) -> None:
    if len(password) < MIN_PASSWORD_LEN:
        raise ValidationError(f"Пароль должен быть не короче {MIN_PASSWORD_LEN} символов")
    if len(password.encode("utf-8")) > BCRYPT_MAX_BYTES:
        raise ValidationError("Пароль слишком длинный")

Для 300 корпоративных клиентов, у каждого из которых свой домен, стоит решить, нужно ли проверять домен email по списку разрешённых доменов клиента. Иначе на /auth/register может зарегистрироваться кто угодно с адресом @gmail.com. Это вопрос к продукту, а не к коду, но аудитор его, скорее всего, задаст.


10. Синхронный драйвер и bcrypt под нагрузкой
  • Категория: производительность

  • Серьёзность: Средняя

  • Если роуты объявлены как async def, синхронный psycopg2 и особенно bcrypt (около 200 мс CPU на rounds=12) блокируют event loop, и весь под перестаёт отвечать. Роуты, которые вызывают UserService, должны быть обычными def: тогда FastAPI выполняет их в threadpool.

  • Threadpool по умолчанию содержит 40 потоков, а пул БД 20 соединений. При всплеске запросов потоки ждут соединения. Это не баг, но стоит выставить разумный таймаут ожидания соединения в пуле.

  • Расчёт ёмкости. Если хотя бы 10 из 50 RPS приходятся на /auth/login, получается около 2 CPU-секунд в секунду только на bcrypt, то есть примерно одно ядро на под. Нужно проверить resources.limits.cpu в манифесте Kubernetes. Если лимит меньше, в 11:30 логины начнут тормозить. Замерьте время pwd_context.hash() внутри контейнера и подберите rounds так, чтобы укладываться примерно в 100–250 мс.

  • Rate limiting на /auth/login. Без него bcrypt превращается в дешёвый DoS-вектор, а брутфорс ничем не ограничен. Проще всего включить лимит на ingress (например, аннотации nginx.ingress.kubernetes.io/limit-rpm), это не требует новых Python-зависимостей.


11. Непоследовательный контракт методов
  • Категория: читаемость / SOLID
  • Серьёзность: Средняя

Методы возвращают то {"error": ...}, то None, то {"status": "ok"}. Роутам приходится угадывать, что считать ошибкой. Ошибку в словаре легко пропустить, и клиент получит 200 OK с "error" в теле. К тому же register_user не возвращает id созданного пользователя.

Лучше использовать доменные исключения и типизированные результаты. Преобразование в HTTP-коды должно жить в одном месте, в exception handler.

class AuthError(Exception): ...
class UserAlreadyExists(AuthError): ...
class InvalidCredentials(AuthError): ...
class UserNotFound(AuthError): ...
class ValidationError(AuthError): ...

@dataclass(frozen=True)
class LoginResult:
    user_id: int
    token: str
@app.exception_handler(InvalidCredentials)
def _(request, exc):
    return JSONResponse({"detail": "Неверный email или пароль"}, status_code=401)

Отдельный вопрос: сообщение "User already exists" тоже раскрывает, зарегистрирован ли email. Для B2B это часто приемлемо, но решение стоит принять осознанно и зафиксировать.


12. Смешение ответственностей (SRP / DIP)
  • Категория: SOLID
  • Серьёзность: Средняя

UserService одновременно пишет SQL, выбирает алгоритм хэширования, выпускает токены и отвечает за бизнес-правила. Из-за этого:

  • смена алгоритма хэширования требует правки сервиса, что нарушает OCP;
  • юнит-тест логики входа требует настоящей БД;
  • формат токена зашит в бизнес-логику.

Предлагаемое разделение: UserRepository отвечает за SQL, PasswordHasher за обёртку над CryptContext, SessionService за токены, а UserService оркеструет их вызовы. Код показан ниже.


13. Мелочи
  • Категория: читаемость

  • Серьёзность: Низкая

  • import psycopg2 не используется, а hashlib после исправлений нужен только для токенов.

  • Нет аннотаций типов и docstring'ов.

  • SELECT * в register_user: для проверки существования достаточно SELECT 1, а после исправления №6 запрос вообще не нужен.

  • self.conn лучше сделать приватным: self._conn.

  • Нет логирования. Неудачные попытки входа нужно логировать для аудита, но без пароля и полного email: достаточно user_id или хэша email.


Рекомендации по рефакторингу

Шаг 1. Итоговый вариант модуля (функциональность сохранена, новых зависимостей нет):

from __future__ import annotations

import hashlib
import logging
import re
import secrets
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone

from passlib.context import CryptContext
from psycopg2 import errors
from psycopg2.extensions import connection as PgConnection

logger = logging.getLogger(__name__)

# ---------- Доменные ошибки ----------
class AuthError(Exception): ...
class ValidationError(AuthError): ...
class UserAlreadyExists(AuthError): ...
class InvalidCredentials(AuthError): ...
class UserNotFound(AuthError): ...

# ---------- Валидация ----------
_EMAIL_RE = re.compile(r"[A-Za-z0-9._%+\-]+@[A-Za-z0-9\-]+(?:\.[A-Za-z0-9\-]+)+")
MAX_EMAIL_LEN = 254
MIN_PASSWORD_LEN = 10
BCRYPT_MAX_BYTES = 72
MAX_NAME_LEN = 100

def normalize_email(raw: str) -> str:
    email = raw.strip().lower()
    if len(email) > MAX_EMAIL_LEN or not _EMAIL_RE.fullmatch(email):
        raise ValidationError("Некорректный email")
    return email

def validate_password(password: str) -> None:
    if len(password) < MIN_PASSWORD_LEN:
        raise ValidationError(f"Пароль должен быть не короче {MIN_PASSWORD_LEN} символов")
    if len(password.encode("utf-8")) > BCRYPT_MAX_BYTES:
        raise ValidationError("Пароль слишком длинный")

def validate_name(raw: str) -> str:
    name = raw.strip()
    if not name or len(name) > MAX_NAME_LEN or not name.isprintable():
        raise ValidationError("Некорректное имя")
    return name

# ---------- Хэширование паролей ----------
class PasswordHasher:
    """Обёртка над passlib; совместима со старыми bcrypt-хэшами."""

    def __init__(self, rounds: int = 12) -> None:
        self._ctx = CryptContext(schemes=["bcrypt"], deprecated="auto", bcrypt__rounds=rounds)
        self._dummy_hash = self._ctx.hash(secrets.token_urlsafe(16))

    def hash(self, password: str) -> str:
        return self._ctx.hash(password)

    def verify(self, password: str, stored_hash: str | None) -> tuple[bool, str | None]:
        """Постоянное время независимо от существования пользователя."""
        try:
            return self._ctx.verify_and_update(password, stored_hash or self._dummy_hash)
        except ValueError:  # битый/нераспознанный хэш или слишком длинный пароль
            return False, None

# ---------- Доступ к данным ----------
@dataclass(frozen=True)
class UserCredentials:
    id: int
    password_hash: str

@dataclass(frozen=True)
class UserProfile:
    id: int
    email: str
    name: str

class UserRepository:
    def __init__(self, conn: PgConnection) -> None:
        self._conn = conn

    def create(self, email: str, password_hash: str, name: str) -> int:
        try:
            with self._conn, self._conn.cursor() as cur:
                cur.execute(
                    "INSERT INTO users (email, password_hash, name) "
                    "VALUES (%s, %s, %s) RETURNING id",
                    (email, password_hash, name),
                )
                return cur.fetchone()[0]
        except errors.UniqueViolation:
            raise UserAlreadyExists() from None

    def get_credentials(self, email: str) -> UserCredentials | None:
        with self._conn, self._conn.cursor() as cur:
            cur.execute(
                "SELECT id, password_hash FROM users WHERE lower(email) = %s",
                (email,),
            )
            row = cur.fetchone()
        return UserCredentials(*row) if row else None

    def update_password_hash(self, user_id: int, password_hash: str) -> None:
        with self._conn, self._conn.cursor() as cur:
            cur.execute("UPDATE users SET password_hash = %s WHERE id = %s", (password_hash, user_id))

    def update_name(self, user_id: int, name: str) -> bool:
        with self._conn, self._conn.cursor() as cur:
            cur.execute("UPDATE users SET name = %s WHERE id = %s", (name, user_id))
            return cur.rowcount > 0

    def get_by_email(self, email: str) -> UserProfile | None:
        with self._conn, self._conn.cursor() as cur:
            cur.execute("SELECT id, email, name FROM users WHERE lower(email) = %s", (email,))
            row = cur.fetchone()
        return UserProfile(*row) if row else None

# ---------- Сессии ----------
class SessionService:
    def __init__(self, conn: PgConnection, ttl: timedelta = timedelta(hours=12)) -> None:
        self._conn = conn
        self._ttl = ttl

    def issue(self, user_id: int) -> str:
        token = secrets.token_urlsafe(32)
        token_hash = hashlib.sha256(token.encode()).hexdigest()
        with self._conn, self._conn.cursor() as cur:
            cur.execute(
                "INSERT INTO user_sessions (token_hash, user_id, expires_at) VALUES (%s, %s, %s)",
                (token_hash, user_id, datetime.now(timezone.utc) + self._ttl),
            )
        return token

# ---------- Бизнес-логика ----------
@dataclass(frozen=True)
class LoginResult:
    user_id: int
    token: str

class UserService:
    def __init__(self, users: UserRepository, hasher: PasswordHasher, sessions: SessionService) -> None:
        self._users = users
        self._hasher = hasher
        self._sessions = sessions

    def register_user(self, email: str, password: str, name: str) -> int:
        email = normalize_email(email)
        validate_password(password)
        name = validate_name(name)
        return self._users.create(email, self._hasher.hash(password), name)

    def login(self, email: str, password: str) -> LoginResult:
        try:
            email = normalize_email(email)
        except ValidationError:
            raise InvalidCredentials() from None

        creds = self._users.get_credentials(email)
        is_valid, upgraded_hash = self._hasher.verify(password, creds.password_hash if creds else None)
        if creds is None or not is_valid:
            logger.info("login_failed", extra={"user_id": creds.id if creds else None})
            raise InvalidCredentials()

        if upgraded_hash:
            self._users.update_password_hash(creds.id, upgraded_hash)
        return LoginResult(user_id=creds.id, token=self._sessions.issue(creds.id))

    def update_profile(self, user_id: int, new_name: str) -> None:
        if not self._users.update_name(user_id, validate_name(new_name)):
            raise UserNotFound(user_id)

    def get_user_by_email(self, email: str) -> UserProfile | None:
        try:
            return self._users.get_by_email(normalize_email(email))
        except ValidationError:
            return None

PasswordHasher нужно создавать один раз на процесс, например через lru_cache-dependency. Иначе dummy-хэш будет пересчитываться на каждый запрос.

Шаг 2. Миграции БД (до релиза):

  1. Проверить дубли lower(email) и разобрать их вручную.
  2. CREATE UNIQUE INDEX CONCURRENTLY users_email_lower_uq ON users (lower(email));
  3. Убедиться, что password_hash имеет тип varchar(255) или text. Bcrypt-хэш занимает 60 символов, и если колонку уже сузили под MD5 до 32, вставка упадёт.
  4. Создать таблицу user_sessions (token_hash char(64) PK, user_id FK, expires_at timestamptz, revoked_at timestamptz NULL) с индексом по user_id.
  5. Одноразовым скриптом перехэшировать или сбросить открытые пароли из тестовых данных.

Шаг 3. Тесты (минимум для аудита):

  • Параметризованный тест с инъекционными строками (' OR '1'='1' --, '; DROP TABLE users; --) для каждого метода: данные должны остаться целыми.
  • Вход со старым bcrypt-хэшем, взятым из фикстуры в формате прод-БД, проходит успешно.
  • Вход со старым хэшем с rounds=10 проходит и приводит к rehash до 12.
  • Параллельная регистрация одного email (два потока) даёт ровно одну запись и одно исключение UserAlreadyExists.
  • Ivan@Corp.RU при регистрации и ivan@corp.ru при входе работают как один пользователь.
  • Пустой пароль, пароль из 9 символов и пароль из 73 байт (кириллица) дают ValidationError, а не 500.
  • Токен не содержит user_id и хэша пароля, два входа подряд дают разные токены.
  • update_profile несуществующего id выбрасывает UserNotFound.
  • После исключения внутри транзакции следующий запрос на том же соединении проходит (регрессия по №7).
  • Время login для несуществующего и существующего email различается не более чем на ~20%. Это грубая проверка, её лучше держать в отдельной группе тестов.

Шаг 4. Инфраструктура и процесс:

  • ruff с правилами S (flake8-bandit) в CI. Правило S608 поймает регрессии SQL-форматирования.
  • Закрепить совместимые версии passlib и bcrypt в lock-файле и проверить в контейнере, что CryptContext инициализируется без ошибок.
  • Rate limit на /auth/login и /auth/register на уровне ingress.
  • Проверить, что роуты с UserService объявлены через def, а не async def.
  • Проверить CPU-лимиты подов с учётом стоимости bcrypt.

Приоритизация под дедлайн. Пункты 1–7 и 9 закрыть обязательно до релиза, иначе аудит не будет пройден. Пункт 8 закрыть, если авторизация в роуте не подтверждена. Пункты 10–13 можно вынести в следующий спринт, но рефакторинг из шага 1 почти бесплатно закрывает большинство из них, поэтому разумнее сделать всё одним PR.

Похожие промпты

Программирование и Разработка

Проектировщик REST API

Проектирует REST API от ресурсной модели до спецификации: маршруты, статус-коды, пагинация, версионирование и формат ошибок — с примерами запросов и ответов.

Программирование и Разработка

Ревью Next.js-кода: App Router и Server Components

Проводит ревью Next.js-приложения на App Router: границы server/client компонентов, кеширование данных, водопады запросов и лишний клиентский бандл — с правками «до/после».

Программирование и Разработка

SEO-аудит Next.js-приложения

Проверяет Next.js-приложение по SEO-чеклисту: метаданные и canonical, рендеринг для ботов, sitemap/robots, Core Web Vitals — и выдаёт план правок по приоритету.

Программирование и Разработка

Ревью backend-кода: архитектура и безопасность

Проводит ревью серверного кода: слои и границы модулей, работа с БД и транзакциями, обработка ошибок, инъекции и утечки данных — замечания ранжированы, правки «до/после».