Sadmin
2222b7ad5b
- 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
248 lines
10 KiB
Python
248 lines
10 KiB
Python
"""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() |