Специалист по Code Review
Проводит профессиональный код-ревью с оценкой читаемости, SOLID, безопасности и приоритизированными рекомендациями по рефакторингу.
Пример запуска
Сгенерировано 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 в коде нет. Перед аудитом каждая из этих проблем по отдельности даёт критичную находку.
Допущения
- Проверка токена выполняется в другом модуле (скорее всего, dependency в FastAPI), и её код здесь не показан.
- Авторизацию в
/users/{id}(право текущего пользователя менять профильid) делает роут. Если это не так, проблема №8 становится критичной. - Колонка
users.emailне имеет уникального индекса, иначе не было бы ручной проверкиSELECT ... WHERE email. - Роуты вызывают сервис синхронно или через
run_in_threadpool. Соединение берётся из пула на время запроса. - Под «старыми хэшами 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. Миграции БД (до релиза):
- Проверить дубли
lower(email)и разобрать их вручную. CREATE UNIQUE INDEX CONCURRENTLY users_email_lower_uq ON users (lower(email));- Убедиться, что
password_hashимеет типvarchar(255)илиtext. Bcrypt-хэш занимает 60 символов, и если колонку уже сузили под MD5 до 32, вставка упадёт. - Создать таблицу
user_sessions (token_hash char(64) PK, user_id FK, expires_at timestamptz, revoked_at timestamptz NULL)с индексом поuser_id. - Одноразовым скриптом перехэшировать или сбросить открытые пароли из тестовых данных.
Шаг 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-кода: архитектура и безопасность
Проводит ревью серверного кода: слои и границы модулей, работа с БД и транзакциями, обработка ошибок, инъекции и утечки данных — замечания ранжированы, правки «до/после».