Audit log retention: a way to purge, off by default
The audit log is append-only and nothing ever removed a row, so it grows for as long as a site runs. This site reached tens of thousands of pages, which is past the point of being useful: nobody pages through that to find anything. Retention is OPT-IN and the default keeps everything. An upgrade that silently started deleting a site's audit history would be indefensible, so audit_retentiondays defaults to 0 and nothing is removed until somebody sets a number. The Site & Facility card shows the log's size and exactly how many entries a purge would delete BEFORE anything is pressed. Deleting happens in committed batches. That is not an optimisation: one DELETE over a million rows builds a single enormous transaction and holds locks across the table while it runs, which on this server means blocking the collectors posting inventory. The API caps its batch count so the request cannot outlive a proxy timeout, and a run that stops early reports how much is left rather than claiming it finished - a purge reporting success over 900k remaining rows would be the same silent half-failure this codebase keeps getting bitten by. `flask audit stats` and `flask audit purge` for the server, the latter reporting by default and deleting only with --apply, with no batch cap because it is not behind a proxy and a purge that stops halfway leaves the operator guessing. audit.purge is a separate permission from audit.view. Reading the log and destroying it are different acts, and plenty of people need only the first. Chunking selects a page of ids and deletes those rather than using DELETE ... LIMIT, which is MySQL-only while the suite runs on SQLite.
This commit is contained in:
135
tests/test_core/test_audit_retention.py
Normal file
135
tests/test_core/test_audit_retention.py
Normal file
@@ -0,0 +1,135 @@
|
||||
"""Audit retention: the purge, its default, and its refusal to overreach.
|
||||
|
||||
The audit log is the one table in this app where deleting is the FEATURE, which
|
||||
makes every guard here load-bearing. A retention bug does not corrupt data, it
|
||||
destroys history nobody can get back.
|
||||
"""
|
||||
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
import pytest
|
||||
|
||||
from shopdb.core.models import AuditLog
|
||||
from shopdb.core.services import auditretention
|
||||
|
||||
|
||||
def _utcnow():
|
||||
return datetime.now(timezone.utc).replace(tzinfo=None)
|
||||
|
||||
|
||||
def _entry(db, ageindays, action='updated'):
|
||||
row = AuditLog(action=action, entitytype='Machine', entityname='M-1',
|
||||
timestamp=_utcnow() - timedelta(days=ageindays))
|
||||
db.session.add(row)
|
||||
db.session.commit()
|
||||
return row
|
||||
|
||||
|
||||
def test_nothing_is_deleted_when_no_retention_is_configured(db):
|
||||
"""The default must keep everything.
|
||||
|
||||
An upgrade that started silently deleting a site's audit history because a
|
||||
setting defaulted to 90 days would be indefensible.
|
||||
"""
|
||||
_entry(db, 4000)
|
||||
assert auditretention.configureddays() == 0
|
||||
with pytest.raises(ValueError):
|
||||
auditretention.purge(0)
|
||||
assert AuditLog.query.count() == 1
|
||||
|
||||
|
||||
def test_a_purge_deletes_only_what_is_older_than_the_window(db):
|
||||
_entry(db, 400)
|
||||
_entry(db, 100)
|
||||
keeper = _entry(db, 5)
|
||||
|
||||
result = auditretention.purge(90)
|
||||
|
||||
assert result['deleted'] == 2
|
||||
assert result['complete'] is True
|
||||
assert [row.auditlogid for row in AuditLog.query.all()] == [keeper.auditlogid]
|
||||
|
||||
|
||||
def test_a_dry_run_reports_without_deleting(db):
|
||||
_entry(db, 400)
|
||||
_entry(db, 400)
|
||||
|
||||
result = auditretention.purge(90, dryrun=True)
|
||||
|
||||
assert result['matched'] == 2
|
||||
assert result['deleted'] == 0
|
||||
assert AuditLog.query.count() == 2, 'a dry run that deleted would be a trap'
|
||||
|
||||
|
||||
def test_a_capped_purge_says_it_did_not_finish(db):
|
||||
"""A half-done purge must not report success.
|
||||
|
||||
The API caps batches so the request cannot outlive a proxy timeout. If that
|
||||
path reported a finished job, an admin would believe a log was trimmed while
|
||||
most of it was still there.
|
||||
"""
|
||||
for _ in range(5):
|
||||
_entry(db, 400)
|
||||
|
||||
result = auditretention.purge(90, maxbatches=2, chunksize=1)
|
||||
|
||||
assert result['deleted'] == 2
|
||||
assert result['remaining'] == 3
|
||||
assert result['complete'] is False
|
||||
|
||||
|
||||
def test_chunking_deletes_everything_it_matched(db):
|
||||
"""Batching is an implementation detail; the outcome must not depend on it."""
|
||||
for _ in range(7):
|
||||
_entry(db, 400)
|
||||
|
||||
result = auditretention.purge(90, chunksize=2)
|
||||
|
||||
assert result['deleted'] == 7
|
||||
assert result['complete'] is True
|
||||
assert AuditLog.query.count() == 0
|
||||
|
||||
|
||||
def test_stats_report_the_backlog_a_purge_would_remove(db):
|
||||
_entry(db, 400)
|
||||
_entry(db, 5)
|
||||
|
||||
info = auditretention.stats(90)
|
||||
|
||||
assert info['total'] == 2
|
||||
assert info['expired'] == 1
|
||||
assert info['oldest'] is not None
|
||||
|
||||
|
||||
def test_purging_requires_its_own_permission(client, db, auth_headers):
|
||||
"""audit.view is not permission to destroy the log."""
|
||||
from werkzeug.security import generate_password_hash
|
||||
from shopdb.core.models import User, Role, Permission
|
||||
|
||||
perm = Permission(name='audit.view', description='View audit logs',
|
||||
category='admin')
|
||||
role = Role(rolename='auditreader', description='Reads the audit log')
|
||||
role.permissions.append(perm)
|
||||
db.session.add_all([perm, role])
|
||||
db.session.flush()
|
||||
user = User(username='auditreader', email='ar@test.local',
|
||||
passwordhash=generate_password_hash('testpass'))
|
||||
user.roles.append(role)
|
||||
db.session.add(user)
|
||||
db.session.commit()
|
||||
|
||||
login = client.post('/api/auth/login',
|
||||
json={'username': 'auditreader', 'password': 'testpass'})
|
||||
headers = {'Authorization':
|
||||
'Bearer ' + login.get_json()['data']['access_token']}
|
||||
|
||||
assert client.get('/api/auditlogs/retention',
|
||||
headers=headers).status_code == 200
|
||||
assert client.post('/api/auditlogs/purge', json={'days': 90},
|
||||
headers=headers).status_code == 403
|
||||
|
||||
|
||||
def test_the_purge_route_refuses_when_no_retention_is_set(client, db, auth_headers):
|
||||
response = client.post('/api/auditlogs/purge', json={}, headers=auth_headers)
|
||||
assert response.status_code == 400
|
||||
assert 'retention' in response.get_json()['data']['error']['message'].lower()
|
||||
Reference in New Issue
Block a user