vector/backend/password_policy.py

"""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 записей.
    """
    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.created_at.desc(),
                          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()