vector/backend/tests/test_b23_policy.py

"""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()