From 2222b7ad5b6ffdef29a12d03eb9b7b33a4ec2857 Mon Sep 17 00:00:00 2001 From: viktot14-ai Date: Fri, 25 Sep 2026 09:56:26 +0300 Subject: [PATCH] =?UTF-8?q?B23:=20=D0=BF=D0=BE=D0=BB=D0=B8=D1=82=D0=B8?= =?UTF-8?q?=D0=BA=D0=B0=20=D0=BF=D0=B0=D1=80=D0=BE=D0=BB=D0=B5=D0=B9=20?= =?UTF-8?q?=D0=B8=D1=81=D0=BF=D0=BE=D0=BB=D0=BD=D1=8F=D0=B5=D1=82=D1=81?= =?UTF-8?q?=D1=8F=20+=20=D1=80=D0=B5=D1=82=D0=B5=D0=BD=D1=86=D0=B8=D1=8F?= =?UTF-8?q?=20=D0=B6=D1=83=D1=80=D0=BD=D0=B0=D0=BB=D0=BE=D0=B2=20=D0=BF?= =?UTF-8?q?=D0=BE-=D0=BD=D0=B0=D1=81=D1=82=D0=BE=D1=8F=D1=89=D0=B5=D0=BC?= =?UTF-8?q?=D1=83?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - backend/password_policy.py: validate_new_password (длина, сложность буквы+цифры, сверка с текущим и историей), record_password_change (хранит ПРЕДЫДУШИЙ хэш, обрезка до password_history_count), password_expired + enforce_expiry (пометка must_change_password) - users.password_changed_at + таблица password_history (010_b23_policy) - change-password и админ-сброс теперь под полной политикой (сложность и переиспользование вместо голой длины); get_current_user проверяет срок действия каждого пароля при запросе - backend/purge_logs.py: реальный DELETE просроченных audit_events (+changes) и auth_events; POST /admin/users/maintenance/purge-logs (manage_security) + CLI python -m backend.purge_logs для cron - тесты: +19 (258 passed) — валидация/история/срок, purge/идемпотентность/403 --- backend/alembic/versions/010_b23_policy.py | 40 ++++ backend/models.py | 17 ++ backend/password_policy.py | 129 +++++++++++ backend/purge_logs.py | 96 ++++++++ backend/routers/admin_users.py | 45 +++- backend/routers/auth.py | 21 +- backend/tests/test_b23_policy.py | 248 +++++++++++++++++++++ 7 files changed, 585 insertions(+), 11 deletions(-) create mode 100644 backend/alembic/versions/010_b23_policy.py create mode 100644 backend/password_policy.py create mode 100644 backend/purge_logs.py create mode 100644 backend/tests/test_b23_policy.py diff --git a/backend/alembic/versions/010_b23_policy.py b/backend/alembic/versions/010_b23_policy.py new file mode 100644 index 0000000..8a0305e --- /dev/null +++ b/backend/alembic/versions/010_b23_policy.py @@ -0,0 +1,40 @@ +"""B23: исполнение политики паролей + реальная ретенция журналов. + +Revises: 009_e3_operations +""" +from __future__ import annotations + +import sqlalchemy as sa +from alembic import op +from sqlalchemy.dialects import postgresql as pg + +revision = '010_b23_policy' +down_revision = '009_e3_operations' +branch_labels = None +depends_on = None + + +def upgrade() -> None: + # Парольная политика: когда пароль был сменён в последний раз + op.add_column( + 'users', + sa.Column('password_changed_at', sa.DateTime(timezone=True), + server_default=sa.text('NOW()'), nullable=True), + ) + # История паролей (проверка переиспользования, password_history_count) + op.create_table( + 'password_history', + sa.Column('id', pg.UUID(as_uuid=True), primary_key=True, + server_default=sa.text('gen_random_uuid()')), + sa.Column('user_id', pg.UUID(as_uuid=True), + sa.ForeignKey('users.id', ondelete='CASCADE'), nullable=False), + sa.Column('hashed_password', sa.String(255), nullable=False), + sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('NOW()')), + ) + op.create_index('idx_password_history_user', 'password_history', ['user_id', 'created_at']) + + +def downgrade() -> None: + op.drop_index('idx_password_history_user', table_name='password_history') + op.drop_table('password_history') + op.drop_column('users', 'password_changed_at') \ No newline at end of file diff --git a/backend/models.py b/backend/models.py index 573d145..212d327 100644 --- a/backend/models.py +++ b/backend/models.py @@ -33,6 +33,8 @@ class User(Base): must_change_password = Column(Boolean, default=False) created_by = Column(UUID(as_uuid=True), ForeignKey("users.id", ondelete="SET NULL")) updated_at = Column(DateTime, server_default=func.now()) + # B23: политика паролей (expiry/history исполняется в routers/auth.py) + password_changed_at = Column(DateTime(timezone=True)) class Case(Base): @@ -335,6 +337,21 @@ class SecuritySetting(Base): key = Column(String(50), primary_key=True) value = Column(String(255), nullable=False) + +class PasswordHistoryEntry(Base): + """B23: история паролей (bcrypt-хэши). Записывается при каждой смене, + проверяется против password_history_count последних записей.""" + __tablename__ = "password_history" + + id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) + user_id = Column(UUID(as_uuid=True), ForeignKey("users.id", ondelete="CASCADE"), nullable=False) + hashed_password = Column(String(255), nullable=False) + created_at = Column(DateTime(timezone=True), server_default=func.now()) + + __table_args__ = ( + sa.Index("idx_password_history_user", "user_id", "created_at"), + ) + # ==================== B21/E2: аудит действий ==================== class SearchOperation(Base): diff --git a/backend/password_policy.py b/backend/password_policy.py new file mode 100644 index 0000000..8f0faf7 --- /dev/null +++ b/backend/password_policy.py @@ -0,0 +1,129 @@ +"""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() \ No newline at end of file diff --git a/backend/purge_logs.py b/backend/purge_logs.py new file mode 100644 index 0000000..09f0aeb --- /dev/null +++ b/backend/purge_logs.py @@ -0,0 +1,96 @@ +"""B23: ретенция журналов — реальное удаление, а не фильтр при чтении. + +До сих пор «ретенция 90 дней» из security_settings реализовывалась как +фильтр в GET /admin/users/audit — записи копились в БД вечно. Этот модуль +удаляет просроченное по-настоящему (DELETE), охватывая: + + audit_events (+ audit_changes каскадно) — по audit_retention_days + auth_events — по audit_retention_days + +Запуск: POST /api/v1/admin/maintenance/purge-logs (право manage_security) +или вручную: + docker exec vector-backend python -m backend.purge_logs +(режим разового запуска — для cron внутри контейнера) +""" +from __future__ import annotations + +import os +from datetime import datetime, timedelta, timezone + +from sqlalchemy import text +from sqlalchemy.orm import Session + +from backend.logging_setup import get_logger +from backend.models import AuthEvent, SecuritySetting + +log = get_logger('retention') + + +def _retention_days(db: Session, default: int = 90) -> int: + row = db.query(SecuritySetting).filter( + SecuritySetting.key == 'audit_retention_days').first() + try: + return int(row.value) if row else default + except (TypeError, ValueError): + return default + + +def purge_logs(db: Session, retention_days: int | None = None) -> dict[str, int]: + """Удалить записи журналов старше retention_days. Возвращает счётчики. + + audit_changes удаляются каскадом по FK (ondelete=CASCADE); для SQLite-тестов + (без каскада на уровне БД) удаляем явно. + """ + if retention_days is None: + retention_days = _retention_days(db) + cutoff = datetime.now(timezone.utc) - timedelta(days=max(0, retention_days)) + # SQLite отдаёт naive datetimes (DateTime(timezone=True) не respected) — + # сравнение aware-cutoff с naive-колонкой падает TypeError. Убираем tzinfo: + cutoff = cutoff.replace(tzinfo=None) + + audit_deleted = 0 + try: + from backend.models import AuditChange, AuditEvent + change_ids = (db.query(AuditChange.id) + .join(AuditEvent, AuditChange.event_id == AuditEvent.id) + .filter(AuditEvent.created_at < cutoff).all()) + change_ids = [row[0] for row in change_ids] + if change_ids: + (db.query(AuditChange) + .filter(AuditChange.id.in_(change_ids)) + .delete(synchronize_session=False)) # changes считаются вместе с событием + audit_deleted += (db.query(AuditEvent) + .filter(AuditEvent.created_at < cutoff) + .delete(synchronize_session=False)) + except Exception as e: + db.rollback() + log.error('purge audit_events не удался: %s', e) + raise + + auth_deleted = 0 + try: + auth_deleted += (db.query(AuthEvent) + .filter(AuthEvent.created_at < cutoff) + .delete(synchronize_session=False)) + except Exception as e: + log.error('purge auth_events не удался: %s', e) + raise + + db.commit() + if audit_deleted or auth_deleted: + log.info('retention purge: %d аудит-записей, %d auth-записей удалено (cutoff=%s, %d дн.)', + audit_deleted, auth_deleted, cutoff.date(), retention_days) + return {'audit_events': audit_deleted, 'auth_events': auth_deleted} + + +def _cli() -> None: + """Разовый прогон: python -m backend.purge_logs (для cron).""" + from backend.database import SessionLocal + with SessionLocal() as db: + result = purge_logs(db) + print(f'purge: {result}') + raise SystemExit(0) + + +if __name__ == '__main__': + _cli() \ No newline at end of file diff --git a/backend/routers/admin_users.py b/backend/routers/admin_users.py index 40ed90d..c6dcf11 100644 --- a/backend/routers/admin_users.py +++ b/backend/routers/admin_users.py @@ -17,6 +17,9 @@ from sqlalchemy import text from sqlalchemy.orm import Session from backend.audit import audit_log, visible_unit_ids +from backend.logging_setup import get_logger +from backend.password_policy import record_password_change, validate_new_password +from backend.purge_logs import _retention_days, purge_logs from backend.database import get_db from backend.models import MchsUnit, Role, User, UserRole from backend.routers.auth import ( @@ -139,6 +142,13 @@ def create_user( ) -> dict[str, Any]: if len(payload.password) < 8: raise HTTPException(status_code=400, detail='Минимальная длина пароля: 8 символов') + # B23: сложность проверяется и на временный пароль админа (история пуста — + # новый пользователь, проверка переиспользования не применима) + probe = User(username=payload.username, email=payload.email) + try: + validate_new_password(db, probe, payload.password) + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) if db.query(User).filter(User.username == payload.username).first(): raise HTTPException(status_code=400, detail='Пользователь с таким именем уже существует') if db.query(User).filter(User.email == payload.email).first(): @@ -234,10 +244,18 @@ def update_user( changes['is_active'] = (target.is_active, payload.is_active) target.is_active = payload.is_active if payload.reset_password: - if len(payload.reset_password) < 8: - raise HTTPException(status_code=400, detail='Минимальная длина пароля: 8 символов') + # B23: сброс админом тоже под полной политикой + probe = User(username=target.username, email=target.email, id=target.id) + try: + validate_new_password(db, probe, payload.reset_password, + current_hash=target.hashed_password) + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) changes['password'] = ('***', '***') - target.hashed_password = get_password_hash(payload.reset_password) + new_hash = get_password_hash(payload.reset_password) + record_password_change(db, target, new_hash, + previous_hash=target.hashed_password) + target.hashed_password = new_hash target.must_change_password = True if not changes: @@ -285,4 +303,23 @@ def list_audit( 'changes': [{'field': ch.field_name, 'old': ch.old_value, 'new': ch.new_value} for ch in changes], }) - return {'items': items, 'total': q.count(), 'retention_days': retention} \ No newline at end of file + return {'items': items, 'total': q.count(), 'retention_days': retention} + + +@router.post('/maintenance/purge-logs') +def run_purge_logs( + request: Request, + current_user: User = Depends(require_permission('manage_security')), + db: Session = Depends(get_db), +) -> dict[str, Any]: + """B23: реальное удаление журналов старше audit_retention_days. + + Ручной запуск из админки; регулярный — cron `python -m backend.purge_logs`. + Удаляет audit_events (+changes каскадом) и auth_events. + """ + result = purge_logs(db) + audit_log(db, current_user, 'purge_logs', object_type='maintenance', + request=request, details=result) + log = get_logger('admin') + log.info('purge_logs by %s: %s', getattr(current_user, 'username', '—'), result) + return {'purged': result, 'retention_days': _retention_days(db)} \ No newline at end of file diff --git a/backend/routers/auth.py b/backend/routers/auth.py index b126b2f..7ecdebb 100644 --- a/backend/routers/auth.py +++ b/backend/routers/auth.py @@ -17,6 +17,7 @@ from sqlalchemy.orm import Session from backend.database import get_db from backend.logging_setup import get_logger from backend.models import AuthEvent, SecuritySetting, User, UserSession +from backend.password_policy import enforce_expiry, record_password_change, validate_new_password log = get_logger('auth') @@ -308,6 +309,9 @@ async def get_current_user( headers={'WWW-Authenticate': 'Bearer'}, ) + # B23: срок действия пароля (security_settings) — исполняется, а не декларируется + enforce_expiry(db, user) + return user @@ -407,15 +411,18 @@ def change_password( current_user: User = Depends(get_current_user), db: Session = Depends(get_db), ) -> dict[str, str]: - min_len = int(_get_setting(db, 'password_min_length', '8')) if not verify_password(payload.current_password, current_user.hashed_password): raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail='Текущий пароль неверен') - if len(payload.new_password) < min_len: - raise HTTPException( - status_code=status.HTTP_400_BAD_REQUEST, - detail=f'Минимальная длина пароля: {min_len} символов', - ) - current_user.hashed_password = get_password_hash(payload.new_password) + # B23: полная политика (длина, сложность, история) вместо голой длины + try: + validate_new_password(db, current_user, payload.new_password, + current_hash=current_user.hashed_password) + except ValueError as e: + raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e)) + new_hash = get_password_hash(payload.new_password) + record_password_change(db, current_user, new_hash, + previous_hash=current_user.hashed_password) + current_user.hashed_password = new_hash current_user.must_change_password = False current_user.updated_at = datetime.now(timezone.utc) # Отзываем все сессии после смены пароля (кроме текущей — её пересоздадим) diff --git a/backend/tests/test_b23_policy.py b/backend/tests/test_b23_policy.py new file mode 100644 index 0000000..aa0f5a7 --- /dev/null +++ b/backend/tests/test_b23_policy.py @@ -0,0 +1,248 @@ +"""B23: тесты исполняемой политики паролей и реальной ретенции журналов. + +SQLite in-memory (те же компиляторы, что в test_e1_rbac). +""" +from __future__ import annotations + +import sys +from datetime import datetime, timedelta, timezone +from pathlib import Path + +import bcrypt +import pytest + +REPO_ROOT = Path(__file__).resolve().parents[2] +if str(REPO_ROOT) not in sys.path: + sys.path.insert(0, str(REPO_ROOT)) + +import sqlalchemy # noqa: E402 +from sqlalchemy import create_engine # noqa: E402 +from sqlalchemy.dialects.postgresql import JSONB, UUID as PG_UUID # noqa: E402 +from sqlalchemy.ext.compiler import compiles # noqa: E402 +from sqlalchemy.orm import sessionmaker # noqa: E402 +from sqlalchemy.pool import StaticPool # noqa: E402 + + +@compiles(PG_UUID, 'sqlite') +def _uuid_sqlite(type_, compiler, **kw): + return 'CHAR(36)' + + +@compiles(JSONB, 'sqlite') +def _jsonb_sqlite(type_, compiler, **kw): + return 'JSON' + + +@compiles(sqlalchemy.ARRAY, 'sqlite') +def _array_sqlite(type_, compiler, **kw): + return 'TEXT' + + +def _hash(pw: str) -> str: + return bcrypt.hashpw(pw.encode(), bcrypt.gensalt()).decode() + + +@pytest.fixture() +def policy_db(): + from backend import models as m + from backend.database import Base + from backend.models import SecuritySetting + + engine = create_engine( + 'sqlite:///:memory:', + connect_args={'check_same_thread': False}, + poolclass=StaticPool, + ) + Base.metadata.create_all(engine) + TestingSession = sessionmaker(bind=engine) + db = TestingSession() + + user = m.User(username='op1', email='op1@test.by', hashed_password=_hash('oldpass123'), + full_name='Оператор', role='operator', is_active=True) + db.add(user) + db.commit() + + for k, v in (('password_min_length', '8'), + ('password_require_complexity', 'true'), + ('password_expiry_days', '90'), + ('password_history_count', '5'), + ('audit_retention_days', '90')): + db.add(SecuritySetting(key=k, value=v)) + db.commit() + + yield {'db': db, 'user': user} + db.close() + + +def _validate(env, password, current_hash=None): + from backend.password_policy import validate_new_password + validate_new_password(env['db'], env['user'], password, current_hash=current_hash) + + +class TestPasswordValidation: + def test_short_rejected(self, policy_db): + with pytest.raises(ValueError, match='Минимальная длина'): + _validate(policy_db, 'Ab1') + + def test_no_digits_rejected(self, policy_db): + with pytest.raises(ValueError, match='буквы и цифры'): + _validate(policy_db, 'abcdefgh') + + def test_no_letters_rejected(self, policy_db): + with pytest.raises(ValueError, match='буквы и цифры'): + _validate(policy_db, '12345678') + + def test_complex_ok(self, policy_db): + _validate(policy_db, 'newpass456') # буквы+цифры, длина ок — не бросает + + def test_same_as_current_rejected(self, policy_db): + with pytest.raises(ValueError, match='ранее использованным'): + _validate(policy_db, 'oldpass123', current_hash=policy_db['user'].hashed_password) + + +class TestPasswordHistory: + def _change(self, env, new_password: str): + """Как в реальном коде: validate → record (со старым хэшем) → подмена.""" + from backend.password_policy import record_password_change, validate_new_password + db, user = env['db'], env['user'] + validate_new_password(db, user, new_password, current_hash=user.hashed_password) + new_hash = _hash(new_password) + record_password_change(db, user, new_hash, + previous_hash=user.hashed_password) + user.hashed_password = new_hash + db.commit() + + def test_reuse_recent_rejected(self, policy_db): + self._change(policy_db, 'firstpass1') + with pytest.raises(ValueError, match='ранее использованным'): + _validate(policy_db, 'oldpass123') + + def test_old_password_beyond_history_allowed(self, policy_db): + """Пароль, выпавший из истории, можно вернуть. count=5: история хранит + 5 последних СТАРЫХ хэшей, текущий (6-й) отдельно. После 6 смен в + истории s1..s5, oldpass выпал — разрешён.""" + for i in range(1, 7): + self._change(policy_db, f'secretpw{i}') + # oldpass за пределами последних 5 — разрешён + _validate(policy_db, 'oldpass123') + + def test_current_password_protected(self, policy_db): + """Текущий пароль тоже нельзя вернуть (сверка с current_hash).""" + self._change(policy_db, 'firstpass1') + with pytest.raises(ValueError, match='ранее использованным'): + _validate(policy_db, 'oldpass123') + + def test_history_trimmed_to_count(self, policy_db): + from backend.models import PasswordHistoryEntry + db = policy_db['db'] + for i in range(1, 8): + self._change(policy_db, f'secretpw{i}') + count = db.query(PasswordHistoryEntry).count() + assert count == 5 + + def test_same_password_rejected_immediately(self, policy_db): + with pytest.raises(ValueError, match='ранее использованным'): + _validate(policy_db, 'oldpass123', current_hash=policy_db['user'].hashed_password) + + +class TestPasswordExpiry: + def test_fresh_password_not_expired(self, policy_db): + from backend.password_policy import password_expired + assert not password_expired(policy_db['db'], policy_db['user']) + + def test_expired_after_90_days(self, policy_db): + from backend.password_policy import password_expired + db, user = policy_db['db'], policy_db['user'] + user.password_changed_at = datetime.now(timezone.utc) - timedelta(days=91) + db.commit() + assert password_expired(db, user) + + def test_not_expired_within_90_days(self, policy_db): + from backend.password_policy import password_expired + db, user = policy_db['db'], policy_db['user'] + user.password_changed_at = datetime.now(timezone.utc) - timedelta(days=89) + db.commit() + assert not password_expired(db, user) + + def test_expiry_disabled_when_zero(self, policy_db): + from backend.password_policy import password_expired + db, user = policy_db['db'], policy_db['user'] + from backend.models import SecuritySetting + row = db.query(SecuritySetting).filter(SecuritySetting.key == 'password_expiry_days').first() + row.value = '0' + db.commit() + user.password_changed_at = datetime.now(timezone.utc) - timedelta(days=400) + db.commit() + assert not password_expired(db, user) + + def test_naive_datetime_handled(self, policy_db): + from backend.password_policy import password_expired + db, user = policy_db['db'], policy_db['user'] + user.password_changed_at = datetime.now(timezone.utc).replace(tzinfo=None) - timedelta(days=91) + db.commit() + assert password_expired(db, user) + + +class TestRetentionPurge: + @pytest.fixture() + def logs_db(self, policy_db): + from backend.models import AuditChange, AuditEvent, AuthEvent + db = policy_db['db'] + now = datetime.now(timezone.utc) + # 2 старых (200 дн.) + 2 свежих (10 дн.): событие + change + auth-запись на каждое + for days in (200, 10): + for i in (1, 2): + ev = AuditEvent(username='op1', event_type='user_update', created_at=now - timedelta(days=days)) + db.add(ev) + db.flush() + db.add(AuditChange(event_id=ev.id, field_name='f', old_value='a', new_value='b')) + db.add(AuthEvent(username='op1', event_type='login', created_at=now - timedelta(days=days))) + db.commit() + yield {'db': db, 'now': now} + db.close() + + def test_purge_deletes_old(self, logs_db): + from backend.models import AuditChange, AuditEvent, AuthEvent + from backend.purge_logs import purge_logs + db = logs_db['db'] + result = purge_logs(db) + assert result == {'audit_events': 2, 'auth_events': 2} + assert db.query(AuditEvent).count() == 2 + assert db.query(AuthEvent).count() == 2 + # field-level изменения старых событий удалены вместе с событиями + assert db.query(AuditChange).count() == 2 + + def test_purge_respects_setting(self, logs_db): + from backend.models import SecuritySetting + from backend.purge_logs import purge_logs + db = logs_db['db'] + row = db.query(SecuritySetting).filter(SecuritySetting.key == 'audit_retention_days').first() + row.value = '10000' # всё свежее — ничего не удалить + db.commit() + result = purge_logs(db) + assert result == {'audit_events': 0, 'auth_events': 0} + + def test_purge_idempotent(self, logs_db): + from backend.purge_logs import purge_logs + purge_logs(logs_db['db']) + second = purge_logs(logs_db['db']) + assert second == {'audit_events': 0, 'auth_events': 0} + + def test_purge_endpoint_requires_manage_security(self, logs_db): + """Без права manage_security — 403.""" + from backend.main import app + from backend.routers import auth as auth_router + from fastapi.testclient import TestClient + from types import SimpleNamespace + + app.dependency_overrides[auth_router.get_db] = lambda: logs_db['db'] + app.dependency_overrides[auth_router.get_current_user] = lambda: SimpleNamespace( + id='u1', username='op1', role='operator', is_active=True, unit_id=None, + must_change_password=False, status='active', failed_login_count=0, + locked_until=None, email='op1@test.by', full_name='Оператор') + try: + client = TestClient(app) + r = client.post('/api/v1/admin/users/maintenance/purge-logs') + assert r.status_code == 403 + finally: + app.dependency_overrides.clear() \ No newline at end of file