Sadmin
a1de5794e2
Грабля: created_at в SQLite точен до секунды — 7 быстрых смен дают равные метки, а secondary-сортировка по UUID-id случайна → «последние 5» выбираются произвольно (флакущий тест + неверно хранимая история). id теперь autoincrement (обрезка order_by id desc), UUID остаётся как public_id. Миграция 010 обновлена (таблица только на CT108, пересоздаётся миграцией заново — см. деплой).
129 lines
6.7 KiB
Python
129 lines
6.7 KiB
Python
"""B23: исполняемая политика паролей (security_settings).
|
|
|
|
До сих пор security_settings были декларативными: срок действия и история
|
|
пароля показывались в настройках, но код проверял только минимальную длину.
|
|
Этот модуль — единственная точка валидации и записи пароля.
|
|
|
|
Правила (ключи security_settings):
|
|
password_min_length 8
|
|
password_require_complexity true (буквы + цифры)
|
|
password_expiry_days 90 (0 = не проверять срок)
|
|
password_history_count 5 (0 = не проверять переиспользование)
|
|
|
|
Важно: проверка срока действия выполняется при КАЖДОМ защищённом запросе
|
|
(get_current_user) — истёкший пароль переводит пользователя в
|
|
must_change_password (запросы к /auth/* работают, прочее API → 403 с
|
|
подсказкой сменить пароль). Так сессии не ломаются внезапно, но политика
|
|
исполняется.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from backend.logging_setup import get_logger
|
|
from backend.models import PasswordHistoryEntry, SecuritySetting, User
|
|
|
|
log = get_logger('auth')
|
|
|
|
_COMPLEXITY_DIGIT_RE = re.compile(r'[0-9]')
|
|
_COMPLEXITY_LETTER_RE = re.compile(r'[^\W\d_]', re.UNICODE)
|
|
|
|
|
|
def get_setting(db: Session, key: str, default: str) -> str:
|
|
row = db.query(SecuritySetting).filter(SecuritySetting.key == key).first()
|
|
return row.value if row else default
|
|
|
|
|
|
def _aware(dt: datetime) -> datetime:
|
|
if dt.tzinfo is None:
|
|
return dt.replace(tzinfo=timezone.utc)
|
|
return dt
|
|
|
|
|
|
def validate_new_password(db: Session, user: User, new_password: str,
|
|
*, current_hash: str | None = None) -> None:
|
|
"""Бросает ValueError с текстом на русском, если пароль нарушает политику.
|
|
|
|
Проверяет: длину, сложность (буквы+цифры), совпадение с текущим паролем
|
|
и переиспользование (последние password_history_count записей истории).
|
|
Вызывать ДО записи нового хэша.
|
|
"""
|
|
min_len = int(get_setting(db, 'password_min_length', '8'))
|
|
need_complexity = get_setting(db, 'password_require_complexity', 'true').strip().lower() in {'1', 'true', 'yes', 'on'}
|
|
history_count = int(get_setting(db, 'password_history_count', '5'))
|
|
|
|
if len(new_password) < min_len:
|
|
raise ValueError(f'Минимальная длина пароля: {min_len} символов')
|
|
if need_complexity:
|
|
if not _COMPLEXITY_DIGIT_RE.search(new_password) or not _COMPLEXITY_LETTER_RE.search(new_password):
|
|
raise ValueError('Пароль должен содержать буквы и цифры')
|
|
|
|
# Текущий пароль считается частью истории (нельзя «сменить на тот же»)
|
|
candidates: list[str] = []
|
|
if current_hash:
|
|
candidates.append(current_hash)
|
|
if history_count > 0:
|
|
rows = (db.query(PasswordHistoryEntry)
|
|
.filter(PasswordHistoryEntry.user_id == user.id)
|
|
.order_by(PasswordHistoryEntry.created_at.desc())
|
|
.limit(history_count).all())
|
|
candidates.extend(r.hashed_password for r in rows)
|
|
from backend.routers.auth import verify_password
|
|
for old_hash in candidates:
|
|
if old_hash and verify_password(new_password, old_hash):
|
|
raise ValueError('Новый пароль совпадает с ранее использованным')
|
|
|
|
|
|
def record_password_change(db: Session, user: User, new_hash: str,
|
|
*, previous_hash: str | None = None) -> None:
|
|
"""Записать смену пароля: история + password_changed_at + обрезка истории.
|
|
|
|
В историю кладётся ПРЕДЫДУЩИЙ хэш (previous_hash — хэш, который заменяют),
|
|
чтобы его нельзя было вернуть, пока он не выпал из password_history_count
|
|
последних записей. Обрезка хранит ровно password_history_count записей;
|
|
порядок — по monotonic id (created_at в SQLite точен до секунды).
|
|
"""
|
|
history_count = int(get_setting(db, 'password_history_count', '5'))
|
|
old_hash = previous_hash if previous_hash is not None else user.hashed_password
|
|
if old_hash:
|
|
db.add(PasswordHistoryEntry(user_id=user.id, hashed_password=old_hash))
|
|
user.password_changed_at = datetime.now(timezone.utc)
|
|
if history_count > 0:
|
|
keep = (db.query(PasswordHistoryEntry.id)
|
|
.filter(PasswordHistoryEntry.user_id == user.id)
|
|
.order_by(PasswordHistoryEntry.id.desc())
|
|
.limit(history_count).all())
|
|
keep_ids = [row[0] for row in keep]
|
|
if keep_ids:
|
|
(db.query(PasswordHistoryEntry)
|
|
.filter(PasswordHistoryEntry.user_id == user.id,
|
|
PasswordHistoryEntry.id.notin_(keep_ids))
|
|
.delete(synchronize_session=False))
|
|
|
|
|
|
def password_expired(db: Session, user: User) -> bool:
|
|
"""True, если срок действия пароля истёк (password_expiry_days > 0)."""
|
|
expiry_days = int(get_setting(db, 'password_expiry_days', '90'))
|
|
if expiry_days <= 0:
|
|
return False
|
|
changed_at = getattr(user, 'password_changed_at', None)
|
|
if changed_at is None:
|
|
# Миграция поставила NOW() — но если записи нет (создан до B23),
|
|
# считаем от updated_at, чтобы не выгонять всех разом.
|
|
changed_at = getattr(user, 'updated_at', None)
|
|
if changed_at is None:
|
|
return False
|
|
return _aware(changed_at) < datetime.now(timezone.utc) - timedelta(days=expiry_days)
|
|
|
|
|
|
def enforce_expiry(db: Session, user: User) -> None:
|
|
"""Отметить must_change_password, если срок истёк (вызывается на каждом запросе)."""
|
|
if not getattr(user, 'must_change_password', False) and password_expired(db, user):
|
|
user.must_change_password = True
|
|
log.warning('password_expired user=%s — пароль старше %s дн.',
|
|
getattr(user, 'username', '—'),
|
|
get_setting(db, 'password_expiry_days', '90'))
|
|
db.commit() |