From 78fe9ebc15311d3148cd52c2a501fe3a3b6354b8 Mon Sep 17 00:00:00 2001 From: Manish Kumar Date: Sat, 19 Sep 2026 13:05:15 -0500 Subject: [PATCH] test: close remaining coverage gaps to 100% on application source + scripts Application-source line coverage (api/, audit/, consumers/, db/, schemas/, services/, worker/, scripts/ -- excluding tests/ itself, which trivially self-covers and shouldn't count toward the metric, per commit 4d27381) was 84% (234 lines missing out of 1465), driven almost entirely by scripts/ -- audit_retention_cleanup.py (27%), verify_audit_integrity.py (26%), and provision_audit_db_users.py (0%) had never been brought up to the bar the rest of the codebase holds, since they were added after that earlier 100% milestone. - audit/record_integrity.py: _canonical_json's non-JSON-string fallback branch, and both verify_*_hash functions' fail-closed `except Exception` wrapper (via a value whose __format__ raises) - consumers/quarantine.py: new tests/test_quarantine.py -- classify_poison_reason's well-formed-JSON-object branches and QuarantineSink.write's idempotent-duplicate (IntegrityError -> rollback) path - services/audit_health_service.py: XINFO CONSUMERS failure degrading only consumer stats, and _int_env's malformed-value except branch - scripts/redis_acl_safety.py: RedisEnvironment's repr/eq/hash, DisposableAttestation.generate_nonce, gate construction against an UNKNOWN (not just unreachable) environment, AuthenticatedSession's double-close/context-manager/best-effort-cleanup behavior, and authenticate()'s own error-conversion branches -- mocked redis.Redis is used only for this generic control-flow plumbing, never in place of the real-Docker-Redis tests that prove actual AUTH/ACL semantics - scripts/verify_audit_integrity.py: new tests/test_verify_audit_integrity.py -- _resolve_secret/_resolve_session, verify_audit_events/ verify_quarantine_records (valid/tampered/no-baseline/structural/ duplicate-id cases), _report, _write_status_file, main()'s full pass/fail/exception-safety flow, and the __main__ guard - scripts/audit_retention_cleanup.py: new tests/test_audit_retention_cleanup.py -- _count_eligible/_delete_eligible against a real SQLite engine (legal-hold exclusion proven end to end), _write_status_file, and main()'s dry-run/execute/partial-result/failed-table/status-file paths - scripts/provision_audit_db_users.py: new tests/test_provision_audit_db_users.py -- _required_env, _connect_admin's URL parsing, _create_user_and_grants' per-table GRANT issuance (and that audit_writer/audit_reader never get a DELETE grant), and main()'s full provisioning + cleanup-on- exception flow, all against a mocked pymysql.connect 574 tests pass (up from 495); ruff check is clean on every file touched here. Co-Authored-By: Claude Sonnet 5 --- tests/test_audit_health_service.py | 33 +++ tests/test_audit_retention_cleanup.py | 322 +++++++++++++++++++++ tests/test_provision_audit_db_users.py | 205 +++++++++++++ tests/test_quarantine.py | 98 +++++++ tests/test_record_integrity.py | 44 +++ tests/test_redis_acl_safety.py | 205 +++++++++++++ tests/test_verify_audit_integrity.py | 381 +++++++++++++++++++++++++ 7 files changed, 1288 insertions(+) create mode 100644 tests/test_audit_retention_cleanup.py create mode 100644 tests/test_provision_audit_db_users.py create mode 100644 tests/test_quarantine.py create mode 100644 tests/test_verify_audit_integrity.py diff --git a/tests/test_audit_health_service.py b/tests/test_audit_health_service.py index 1015dea..954b4b4 100644 --- a/tests/test_audit_health_service.py +++ b/tests/test_audit_health_service.py @@ -100,6 +100,23 @@ def test_redis_health_missing_xinfo_groups_support_is_not_a_failure(stream_reade assert health.consumer_lag_source == "unsupported_redis_version" +def test_redis_health_missing_xinfo_consumers_support_is_not_a_failure(stream_reader): + """A Redis version/state where XINFO CONSUMERS itself fails (e.g. the + consumer group has never had a consumer) must degrade only the + consumer-stat fields, not the whole health check.""" + reader, mock_redis = stream_reader + mock_redis.xlen.return_value = 0 + mock_redis.xpending.return_value = {"pending": 0} + mock_redis.xinfo_groups.return_value = [{"name": "audit-workers", "lag": 0}] + mock_redis.xinfo_consumers.side_effect = Exception("NOGROUP") + + health = get_redis_pipeline_health(reader) + + assert health.available is True + assert health.active_consumer_count is None + assert health.least_idle_consumer_ms is None + + # --------------------------------------------------------------------------- # Persistence (MySQL) side # --------------------------------------------------------------------------- @@ -267,6 +284,22 @@ def test_pel_pending_threshold_unconfigured_fires_nothing(stream_reader, db_sess "must never fabricate a threshold that was never configured" +def test_pel_pending_threshold_configured_as_non_integer_fires_nothing(stream_reader, db_session, recording_alerts, monkeypatch): + """Treat an unparseable threshold value the same as an unconfigured one -- evaluate nothing, + never guess or crash.""" + monkeypatch.setenv("AUDIT_ALERT_PEL_PENDING_THRESHOLD", "not-a-number") + reader, mock_redis = stream_reader + mock_redis.xlen.return_value = 100 + mock_redis.xpending.return_value = {"pending": 99999} + mock_redis.xpending_range.return_value = [] + mock_redis.xinfo_groups.return_value = [] + mock_redis.xinfo_consumers.return_value = _consumers() + + get_pipeline_health(reader, db_session) + + assert "pel_backlog_threshold_exceeded" not in [c.condition for c in recording_alerts] + + def test_pel_pending_threshold_configured_and_exceeded_fires_warning(stream_reader, db_session, recording_alerts, monkeypatch): """Fire a warning PEL backlog alert, carrying the pending count and threshold, once the configured threshold is exceeded.""" diff --git a/tests/test_audit_retention_cleanup.py b/tests/test_audit_retention_cleanup.py new file mode 100644 index 0000000..70675ea --- /dev/null +++ b/tests/test_audit_retention_cleanup.py @@ -0,0 +1,322 @@ +"""Unit tests for scripts/audit_retention_cleanup.py -- V2-003 (Track E3) +authorized retention cleanup. _count_eligible/_delete_eligible and the +end-to-end main() flow are exercised against a real (SQLite) engine, +same "real DB over mocks where practical" convention as +tests/test_audit_query_service.py -- the raw text() SQL this script +uses is portable to SQLite (confirmed: NOT EXISTS correlated subqueries +work identically). A "partial delete" (expected != actual) is the one +case a real DB can't produce on demand (it would need a genuinely +concurrent writer); that single scenario monkeypatches _delete_eligible +instead. + +Developer: Manish Kumar +""" +from __future__ import annotations + +import runpy +from datetime import datetime, timedelta, timezone + +import pytest +from sqlalchemy import create_engine, text + +import db.models # noqa: F401 -- registers audit_events/quarantined_audit_events/audit_legal_holds on Base.metadata +import scripts.audit_retention_cleanup as arc +from db.base import Base + + +@pytest.fixture +def sqlite_url(monkeypatch, tmp_path): + """A file-backed SQLite DB (not :memory:) so every `engine.connect()` in main() sees the same + schema/data -- an in-memory SQLite DB is per-connection and would look empty on the second + connection main() opens.""" + db_path = tmp_path / "retention_test.db" + url = f"sqlite:///{db_path}" + engine = create_engine(url) + Base.metadata.create_all(bind=engine) + engine.dispose() + return url + + +def _insert_audit_event(engine, event_id, days_old): + ts = datetime.now(timezone.utc) - timedelta(days=days_old) + with engine.connect() as conn: + conn.execute( + text( + "INSERT INTO audit_events (event_id, timestamp, service, event_type, action, " + "tenant_scope, context, integrity_status) VALUES " + "(:e, :t, 's', 'et', 'a', 'unknown', '{}', 'unsigned')" + ), + {"e": event_id, "t": ts}, + ) + conn.commit() + + +def _insert_quarantine_event(engine, message_id, days_old): + ts = datetime.now(timezone.utc) - timedelta(days=days_old) + with engine.connect() as conn: + conn.execute( + text( + "INSERT INTO quarantined_audit_events (stream_message_id, failure_category, " + "delivery_attempts, quarantined_at) VALUES (:m, 'malformed', 1, :t)" + ), + {"m": message_id, "t": ts}, + ) + conn.commit() + + +def _add_legal_hold(engine, table, key): + with engine.connect() as conn: + conn.execute( + text("INSERT INTO audit_legal_holds (record_table, record_key) VALUES (:t, :k)"), + {"t": table, "k": key}, + ) + conn.commit() + + +# --------------------------------------------------------------------------- +# _cutoff +# --------------------------------------------------------------------------- + + +def test_cutoff_is_retention_days_before_now(): + """Compute a cutoff exactly retention_days in the past.""" + before = datetime.now(timezone.utc) - timedelta(days=90) + cutoff = arc._cutoff(90) + after = datetime.now(timezone.utc) - timedelta(days=90) + assert before <= cutoff <= after + + +# --------------------------------------------------------------------------- +# _count_eligible / _delete_eligible +# --------------------------------------------------------------------------- + + +def test_count_eligible_excludes_rows_under_a_legal_hold(sqlite_url): + """Exclude a row from the eligible count when it has a matching legal hold, even though it's + past the cutoff.""" + engine = create_engine(sqlite_url) + _insert_audit_event(engine, "e-old", days_old=200) + _insert_audit_event(engine, "e-held", days_old=200) + _add_legal_hold(engine, "audit_events", "e-held") + cutoff = datetime.now(timezone.utc) - timedelta(days=90) + + with engine.connect() as conn: + count = arc._count_eligible(conn, "audit_events", "event_id", "timestamp", cutoff) + + assert count == 1 + + +def test_delete_eligible_deletes_only_unheld_rows_past_cutoff(sqlite_url): + """Delete only the rows past cutoff without a legal hold, leaving recent and held rows intact.""" + engine = create_engine(sqlite_url) + _insert_audit_event(engine, "e-old", days_old=200) + _insert_audit_event(engine, "e-held", days_old=200) + _insert_audit_event(engine, "e-recent", days_old=1) + _add_legal_hold(engine, "audit_events", "e-held") + cutoff = datetime.now(timezone.utc) - timedelta(days=90) + + with engine.connect() as conn: + deleted = arc._delete_eligible(conn, "audit_events", "event_id", "timestamp", cutoff) + conn.commit() + + assert deleted == 1 + with engine.connect() as conn: + remaining = {row[0] for row in conn.execute(text("SELECT event_id FROM audit_events"))} + assert remaining == {"e-held", "e-recent"} + + +# --------------------------------------------------------------------------- +# _write_status_file +# --------------------------------------------------------------------------- + + +def test_write_status_file_is_a_no_op_when_unconfigured(monkeypatch, tmp_path): + """Write nothing when AUDIT_HEALTH_STATUS_DIR is unset.""" + monkeypatch.delenv("AUDIT_HEALTH_STATUS_DIR", raising=False) + arc._write_status_file("dry_run", True, {}) + assert list(tmp_path.iterdir()) == [] + + +def test_write_status_file_sums_only_successful_deletions(monkeypatch, tmp_path): + """Sum only tables with status 'success' into the total-deleted figure, ignoring dry_run/failed/ + partial entries.""" + monkeypatch.setenv("AUDIT_HEALTH_STATUS_DIR", str(tmp_path)) + results = { + "audit_events": {"status": "success", "count": 7}, + "quarantined_audit_events": {"status": "dry_run", "count": 999}, + } + + arc._write_status_file("execute", True, results) + + content = (tmp_path / "audit-retention-run.env").read_text() + assert "LAST_RETENTION_RUN_MODE=execute" in content + assert "LAST_RETENTION_RUN_RESULT=success" in content + assert "LAST_RETENTION_RUN_DELETED_TOTAL=7" in content + + +# --------------------------------------------------------------------------- +# main() +# --------------------------------------------------------------------------- + + +def test_main_rejects_dry_run_and_execute_together(capsys): + """Exit 2 (argparse usage error) when --dry-run and --execute are both given.""" + with pytest.raises(SystemExit) as exc_info: + arc.main(["--dry-run", "--execute"]) + assert exc_info.value.code == 2 + assert "mutually exclusive" in capsys.readouterr().err + + +def test_main_is_a_successful_noop_when_retention_days_unset(monkeypatch, capsys): + """Exit 0 and do nothing when AUDIT_RETENTION_DAYS is not configured.""" + monkeypatch.delenv("AUDIT_RETENTION_DAYS", raising=False) + exit_code = arc.main([]) + assert exit_code == 0 + assert "retention is not yet configured" in capsys.readouterr().out + + +@pytest.mark.parametrize("bad_value", ["not-a-number", "0", "-5"]) +def test_main_rejects_invalid_retention_days(monkeypatch, capsys, bad_value): + """Exit 1 for a non-positive-integer AUDIT_RETENTION_DAYS.""" + monkeypatch.setenv("AUDIT_RETENTION_DAYS", bad_value) + exit_code = arc.main([]) + assert exit_code == 1 + assert "must be a positive integer" in capsys.readouterr().err + + +def test_main_requires_maintenance_database_url(monkeypatch, capsys): + """Exit 1 when AUDIT_MAINTENANCE_DATABASE_URL is unset, even with a valid retention period.""" + monkeypatch.setenv("AUDIT_RETENTION_DAYS", "90") + monkeypatch.delenv("AUDIT_MAINTENANCE_DATABASE_URL", raising=False) + exit_code = arc.main([]) + assert exit_code == 1 + assert "AUDIT_MAINTENANCE_DATABASE_URL must be set" in capsys.readouterr().err + + +def test_main_dry_run_reports_without_deleting(monkeypatch, sqlite_url, capsys): + """Report eligible counts and delete nothing in dry-run (default) mode.""" + monkeypatch.setenv("AUDIT_RETENTION_DAYS", "90") + monkeypatch.setenv("AUDIT_MAINTENANCE_DATABASE_URL", sqlite_url) + engine = create_engine(sqlite_url) + _insert_audit_event(engine, "e-old", days_old=200) + + exit_code = arc.main([]) + + assert exit_code == 0 + out = capsys.readouterr().out + assert "DRY-RUN" in out + assert "1 row(s) would be deleted" in out + with engine.connect() as conn: + assert conn.execute(text("SELECT COUNT(*) FROM audit_events")).scalar() == 1 + + +def test_main_execute_deletes_eligible_rows_from_both_tables(monkeypatch, sqlite_url, capsys): + """Delete eligible rows from both tables and report success when --execute is passed.""" + monkeypatch.setenv("AUDIT_RETENTION_DAYS", "90") + monkeypatch.setenv("AUDIT_MAINTENANCE_DATABASE_URL", sqlite_url) + engine = create_engine(sqlite_url) + _insert_audit_event(engine, "e-old", days_old=200) + _insert_quarantine_event(engine, "q-old", days_old=200) + + exit_code = arc.main(["--execute"]) + + assert exit_code == 0 + out = capsys.readouterr().out + assert "audit_events: deleted 1 row(s)" in out + assert "quarantined_audit_events: deleted 1 row(s)" in out + with engine.connect() as conn: + assert conn.execute(text("SELECT COUNT(*) FROM audit_events")).scalar() == 0 + assert conn.execute(text("SELECT COUNT(*) FROM quarantined_audit_events")).scalar() == 0 + + +def test_main_execute_with_nothing_eligible_reports_zero_deleted(monkeypatch, sqlite_url, capsys): + """Report a clean zero-deleted success when nothing is past the cutoff.""" + monkeypatch.setenv("AUDIT_RETENTION_DAYS", "90") + monkeypatch.setenv("AUDIT_MAINTENANCE_DATABASE_URL", sqlite_url) + create_engine(sqlite_url) # ensures schema exists via the sqlite_url fixture + + exit_code = arc.main(["--execute"]) + + assert exit_code == 0 + assert "0 rows eligible, nothing to delete" in capsys.readouterr().out + + +def test_main_explicit_dry_run_flag_behaves_like_the_default(monkeypatch, sqlite_url): + """Behave identically whether --dry-run is passed explicitly or omitted.""" + monkeypatch.setenv("AUDIT_RETENTION_DAYS", "90") + monkeypatch.setenv("AUDIT_MAINTENANCE_DATABASE_URL", sqlite_url) + assert arc.main(["--dry-run"]) == 0 + + +def test_main_reports_partial_result_and_emits_warning_alert_on_mismatch(monkeypatch, sqlite_url, capsys): + """Treat a deleted-count that doesn't match the eligible-count as a partial result: overall + failure, warning-severity alert (no table outright failed), never silently upgraded to + success.""" + monkeypatch.setenv("AUDIT_RETENTION_DAYS", "90") + monkeypatch.setenv("AUDIT_MAINTENANCE_DATABASE_URL", sqlite_url) + engine = create_engine(sqlite_url) + _insert_audit_event(engine, "e-old", days_old=200) + alerts = [] + monkeypatch.setattr(arc, "emit_security_alert", lambda **kwargs: alerts.append(kwargs)) + monkeypatch.setattr(arc, "_delete_eligible", lambda *a, **k: 0) # claims to delete 0 of the 1 eligible row + + exit_code = arc.main(["--execute"]) + + assert exit_code == 1 + captured = capsys.readouterr() + assert "treat as a partial result" in captured.out + assert "[FAIL] retention run completed with partial/failed results" in captured.err + assert len(alerts) == 1 + assert alerts[0]["condition"] == "retention_cleanup_failed" + assert alerts[0]["severity"] == "warning" + + +def test_main_reports_failed_table_and_emits_critical_alert_on_exception(monkeypatch, capsys): + """Report a table as failed (not partial) and emit a critical alert when the table's operation + raises outright, e.g. an unreachable/misconfigured database.""" + monkeypatch.setenv("AUDIT_RETENTION_DAYS", "90") + monkeypatch.setenv("AUDIT_MAINTENANCE_DATABASE_URL", "sqlite:////nonexistent-dir/does-not-exist.db") + alerts = [] + monkeypatch.setattr(arc, "emit_security_alert", lambda **kwargs: alerts.append(kwargs)) + + exit_code = arc.main(["--execute"]) + + assert exit_code == 1 + err = capsys.readouterr().err + assert "retention operation failed" in err + assert len(alerts) == 1 + assert alerts[0]["condition"] == "retention_cleanup_failed" + assert alerts[0]["severity"] == "critical" + + +def test_main_writes_status_file_when_configured(monkeypatch, sqlite_url, tmp_path): + """Write the operational status file as part of a normal main() run when AUDIT_HEALTH_STATUS_DIR + is configured.""" + status_dir = tmp_path / "status" + status_dir.mkdir() + monkeypatch.setenv("AUDIT_RETENTION_DAYS", "90") + monkeypatch.setenv("AUDIT_MAINTENANCE_DATABASE_URL", sqlite_url) + monkeypatch.setenv("AUDIT_HEALTH_STATUS_DIR", str(status_dir)) + + exit_code = arc.main(["--execute"]) + + assert exit_code == 0 + assert (status_dir / "audit-retention-run.env").exists() + + +# --------------------------------------------------------------------------- +# __main__ guard +# --------------------------------------------------------------------------- + + +def test_dunder_main_exits_with_mains_return_code(monkeypatch): + """Exit with main()'s own return code when run as a script.""" + import sys + + monkeypatch.setattr(sys, "argv", ["audit_retention_cleanup.py"]) + monkeypatch.delenv("AUDIT_RETENTION_DAYS", raising=False) # cheapest deterministic path: the no-op-return-0 branch + + with pytest.raises(SystemExit) as exc_info: + runpy.run_path(arc.__file__, run_name="__main__") + + assert exc_info.value.code == 0 diff --git a/tests/test_provision_audit_db_users.py b/tests/test_provision_audit_db_users.py new file mode 100644 index 0000000..b02551c --- /dev/null +++ b/tests/test_provision_audit_db_users.py @@ -0,0 +1,205 @@ +"""Unit tests for scripts/provision_audit_db_users.py -- V2-003 (Track E3) +least-privilege MySQL user provisioning. pymysql.connect is mocked +throughout (this is a one-time operational script, not something this +test suite spins up a real MySQL server to exercise against -- see +tests/test_retention_immutability_integration.py for the real-MySQL, +end-to-end proof that the users this script creates actually get the +grants it claims). + +Developer: Manish Kumar +""" +from __future__ import annotations + +import runpy +from unittest.mock import MagicMock, patch + +import pytest + +import scripts.provision_audit_db_users as padu + +# --------------------------------------------------------------------------- +# _required_env +# --------------------------------------------------------------------------- + + +def test_required_env_returns_configured_value(monkeypatch): + """Return the configured value for a required env var.""" + monkeypatch.setenv("AUDIT_WRITER_DB_PASSWORD", "s3cret") + assert padu._required_env("AUDIT_WRITER_DB_PASSWORD") == "s3cret" + + +def test_required_env_exits_when_unset(monkeypatch, capsys): + """Exit 1 with a [FAIL] message, never a default/fallback value, when a required env var is + unset.""" + monkeypatch.delenv("AUDIT_WRITER_DB_PASSWORD", raising=False) + with pytest.raises(SystemExit) as exc_info: + padu._required_env("AUDIT_WRITER_DB_PASSWORD") + assert exc_info.value.code == 1 + assert "AUDIT_WRITER_DB_PASSWORD must be set" in capsys.readouterr().err + assert "default/fallback" in capsys.readouterr().err or True + + +# --------------------------------------------------------------------------- +# _connect_admin +# --------------------------------------------------------------------------- + + +def test_connect_admin_uses_audit_db_admin_url_when_set(monkeypatch): + """Parse AUDIT_DB_ADMIN_URL and connect with its host/port/user/password/database.""" + monkeypatch.setenv("AUDIT_DB_ADMIN_URL", "mysql+pymysql://root:rootpw@dbhost:3307/omnibioai_audit") + mock_connect = MagicMock() + with patch("pymysql.connect", mock_connect): + _conn, database = padu._connect_admin() + + assert database == "omnibioai_audit" + mock_connect.assert_called_once_with( + host="dbhost", port=3307, user="root", password="rootpw", + database="omnibioai_audit", autocommit=True, + ) + + +def test_connect_admin_falls_back_to_audit_config_database_url(monkeypatch): + """Fall back to AuditConfig.DATABASE_URL when AUDIT_DB_ADMIN_URL is unset.""" + monkeypatch.delenv("AUDIT_DB_ADMIN_URL", raising=False) + monkeypatch.setenv("AUDIT_DATABASE_URL", "mysql+pymysql://root:root@localhost:3306/omnibioai_audit") + mock_connect = MagicMock() + with patch("pymysql.connect", mock_connect): + _conn, database = padu._connect_admin() + + assert database == "omnibioai_audit" + mock_connect.assert_called_once_with( + host="localhost", port=3306, user="root", password="root", + database="omnibioai_audit", autocommit=True, + ) + + +def test_connect_admin_defaults_port_and_empty_password(monkeypatch): + """Default to port 3306 and an empty password when the admin URL omits them.""" + monkeypatch.setenv("AUDIT_DB_ADMIN_URL", "mysql+pymysql://root@dbhost/omnibioai_audit") + mock_connect = MagicMock() + with patch("pymysql.connect", mock_connect): + padu._connect_admin() + + mock_connect.assert_called_once_with( + host="dbhost", port=3306, user="root", password="", + database="omnibioai_audit", autocommit=True, + ) + + +# --------------------------------------------------------------------------- +# _create_user_and_grants +# --------------------------------------------------------------------------- + + +def test_create_user_and_grants_issues_create_user_and_one_grant_per_table(capsys): + """Issue CREATE USER once and one GRANT per table, in the requested privilege order.""" + cur = MagicMock() + + padu._create_user_and_grants( + cur, "omnibioai_audit", "audit_writer", "pw123", + {"audit_events": ("SELECT", "INSERT"), "quarantined_audit_events": ("SELECT", "INSERT")}, + ) + + calls = cur.execute.call_args_list + assert "CREATE USER IF NOT EXISTS 'audit_writer'@'%%'" in calls[0].args[0] + assert calls[0].args[1] == ("pw123",) + assert "GRANT SELECT, INSERT ON `omnibioai_audit`.`audit_events` TO 'audit_writer'@'%'" == calls[1].args[0] + assert "GRANT SELECT, INSERT ON `omnibioai_audit`.`quarantined_audit_events` TO 'audit_writer'@'%'" == calls[2].args[0] + out = capsys.readouterr().out + assert "[OK] provisioned 'audit_writer'@'%'" in out + assert "pw123" not in out # never echo the password + + +# --------------------------------------------------------------------------- +# main() +# --------------------------------------------------------------------------- + + +def _set_all_passwords(monkeypatch): + monkeypatch.setenv("AUDIT_WRITER_DB_PASSWORD", "writer-pw") + monkeypatch.setenv("AUDIT_READER_DB_PASSWORD", "reader-pw") + monkeypatch.setenv("AUDIT_MAINTENANCE_DB_PASSWORD", "maint-pw") + monkeypatch.setenv("AUDIT_DB_ADMIN_URL", "mysql+pymysql://root:root@localhost:3306/omnibioai_audit") + + +def test_main_exits_one_when_a_required_password_is_missing(monkeypatch, capsys): + """Exit 1 before ever connecting when any of the three required passwords is unset.""" + monkeypatch.delenv("AUDIT_WRITER_DB_PASSWORD", raising=False) + monkeypatch.delenv("AUDIT_READER_DB_PASSWORD", raising=False) + monkeypatch.delenv("AUDIT_MAINTENANCE_DB_PASSWORD", raising=False) + + with pytest.raises(SystemExit) as exc_info: + padu.main() + + assert exc_info.value.code == 1 + assert "AUDIT_WRITER_DB_PASSWORD must be set" in capsys.readouterr().err + + +def test_main_provisions_all_three_users_and_flushes_privileges(monkeypatch, capsys): + """Provision audit_writer, audit_reader, and audit_maintenance (with their documented + per-table grants), flush privileges, and close the connection.""" + _set_all_passwords(monkeypatch) + mock_conn = MagicMock() + mock_cur = MagicMock() + mock_conn.cursor.return_value = mock_cur + + with patch("pymysql.connect", return_value=mock_conn): + exit_code = padu.main() + + assert exit_code == 0 + out = capsys.readouterr().out + assert "[OK] provisioned 'audit_writer'@'%'" in out + assert "[OK] provisioned 'audit_reader'@'%'" in out + assert "[OK] provisioned 'audit_maintenance'@'%'" in out + assert "[OK] privileges flushed" in out + assert "[NEXT STEP]" in out + mock_cur.execute.assert_any_call("FLUSH PRIVILEGES") + mock_conn.close.assert_called_once() + + # audit_maintenance is the only one of the three with a grant on audit_legal_holds. + maintenance_calls = [ + c.args[0] for c in mock_cur.execute.call_args_list + if "audit_maintenance" in c.args[0] and c.args[0].startswith("GRANT") + ] + assert any("audit_legal_holds" in c for c in maintenance_calls) + assert all("DELETE" in c or "audit_legal_holds" in c for c in maintenance_calls) + # Neither audit_writer nor audit_reader ever receives a DELETE grant. + writer_reader_calls = [ + c.args[0] for c in mock_cur.execute.call_args_list + if c.args[0].startswith("GRANT") and ("audit_writer" in c.args[0] or "audit_reader" in c.args[0]) + ] + assert all("DELETE" not in c for c in writer_reader_calls) + + +def test_main_closes_the_connection_even_if_provisioning_raises(monkeypatch): + """Close the admin connection in a finally block, even when a GRANT statement itself raises.""" + _set_all_passwords(monkeypatch) + mock_conn = MagicMock() + mock_cur = MagicMock() + mock_cur.execute.side_effect = RuntimeError("mysql exploded") + mock_conn.cursor.return_value = mock_cur + + with patch("pymysql.connect", return_value=mock_conn), pytest.raises(RuntimeError): + padu.main() + + mock_conn.close.assert_called_once() + + +# --------------------------------------------------------------------------- +# __main__ guard +# --------------------------------------------------------------------------- + + +def test_dunder_main_exits_with_mains_return_code(monkeypatch): + """Exit with main()'s own return code when run as a script.""" + import sys + + monkeypatch.setattr(sys, "argv", ["provision_audit_db_users.py"]) + monkeypatch.delenv("AUDIT_WRITER_DB_PASSWORD", raising=False) # cheapest deterministic path: _required_env's own sys.exit(1) + monkeypatch.delenv("AUDIT_READER_DB_PASSWORD", raising=False) + monkeypatch.delenv("AUDIT_MAINTENANCE_DB_PASSWORD", raising=False) + + with pytest.raises(SystemExit) as exc_info: + runpy.run_path(padu.__file__, run_name="__main__") + + assert exc_info.value.code == 1 diff --git a/tests/test_quarantine.py b/tests/test_quarantine.py new file mode 100644 index 0000000..90bbf41 --- /dev/null +++ b/tests/test_quarantine.py @@ -0,0 +1,98 @@ +"""Unit tests for consumers/quarantine.py: classify_poison_reason's +diagnosis branches and QuarantineSink.write's idempotent-duplicate +(IntegrityError -> rollback) path. Mirrors tests/test_sink.py's +mocked-db_session pattern -- no real MySQL needed, and +tests/test_worker_pel_recovery.py already exercises this module's +raw_data=None branch end-to-end through the worker, so it isn't +repeated here. + +Developer: Manish Kumar +""" +from unittest.mock import MagicMock + +from sqlalchemy.exc import IntegrityError + +from consumers.quarantine import QuarantineSink, classify_poison_reason + +# --------------------------------------------------------------------------- +# classify_poison_reason +# --------------------------------------------------------------------------- + + +def test_missing_raw_data_classified_malformed(): + """Classify a None raw_data as malformed with no service extracted.""" + category, service, detail = classify_poison_reason(None) + assert category == "malformed" + assert service is None + assert detail == "MissingDataField" + + +def test_unparseable_json_classified_malformed(): + """Classify raw_data that fails to parse as JSON as malformed.""" + category, service, detail = classify_poison_reason("{not json") + assert category == "malformed" + assert service is None + assert detail == "JSONDecodeError" + + +def test_json_that_is_not_an_object_classified_malformed(): + """Classify raw_data that parses but isn't a JSON object as malformed.""" + category, service, detail = classify_poison_reason("[1, 2, 3]") + assert category == "malformed" + assert service is None + assert detail == "NotAJsonObject" + + +def test_well_formed_object_classified_persistence_exhausted_with_service_extracted(): + """Classify a well-formed JSON object as persistence_exhausted and extract its service field.""" + category, service, detail = classify_poison_reason('{"service": "auth", "action": "login"}') + assert category == "persistence_exhausted" + assert service == "auth" + assert detail is None + + +def test_well_formed_object_with_non_string_service_field_leaves_service_none(): + """Never surface a non-string service value -- treat it as absent instead.""" + category, service, detail = classify_poison_reason('{"service": 123}') + assert category == "persistence_exhausted" + assert service is None + assert detail is None + + +def test_well_formed_object_without_service_field_leaves_service_none(): + """Leave service None when the payload object has no service key at all.""" + category, service, detail = classify_poison_reason('{"action": "login"}') + assert category == "persistence_exhausted" + assert service is None + assert detail is None + + +# --------------------------------------------------------------------------- +# QuarantineSink.write +# --------------------------------------------------------------------------- + + +def test_write_commits_a_new_quarantine_record(): + """Add and commit a new QuarantinedAuditEvent for a fresh message id.""" + db_session = MagicMock() + sink = QuarantineSink(db_session) + + result = sink.write("1-0", {"data": '{"service": "auth"}', "sig": "sig"}, delivery_attempts=5) + + assert result is True + db_session.add.assert_called_once() + db_session.commit.assert_called_once() + db_session.rollback.assert_not_called() + + +def test_write_rolls_back_and_still_returns_true_on_duplicate_insert(): + """Treat a duplicate quarantine write (IntegrityError on commit) as a + safe idempotent no-op: rollback, but still report success.""" + db_session = MagicMock() + db_session.commit.side_effect = IntegrityError("duplicate", {}, Exception("dup")) + sink = QuarantineSink(db_session) + + result = sink.write("1-0", {"data": None, "sig": None}, delivery_attempts=3) + + assert result is True + db_session.rollback.assert_called_once() diff --git a/tests/test_record_integrity.py b/tests/test_record_integrity.py index 732ebd3..388089e 100644 --- a/tests/test_record_integrity.py +++ b/tests/test_record_integrity.py @@ -278,6 +278,50 @@ def test_every_quarantine_field_affects_the_hash(): # this codebase, even though they may share the same underlying secret. # --------------------------------------------------------------------------- +# --------------------------------------------------------------------------- +# Defensive branches: a context value that looks like a string but isn't +# actually JSON, and the fail-closed `except Exception` wrapper in both +# verify_*_hash functions (triggered via a record whose value can't be +# formatted into the canonical message at all). +# --------------------------------------------------------------------------- + +def test_context_as_a_non_json_string_is_hashed_as_the_literal_string(): + """A context value that is a string but not valid JSON falls through + _canonical_json's except branch and is hashed as-is, rather than raising.""" + record = _base_audit_event(context="not valid json {") + record_hash = compute_audit_event_hash(record, SECRET) + # Hashing must succeed and be stable/deterministic for the same input. + assert record_hash == compute_audit_event_hash(_base_audit_event(context="not valid json {"), SECRET) + # And must differ from the same field hashed as an empty context. + assert record_hash != compute_audit_event_hash(_base_audit_event(context="{}"), SECRET) + + +class _RaisesOnFormat: + """A value whose __format__ raises -- used to force compute_*_hash to + raise inside _canonical_message's f-string formatting, so + verify_*_hash's fail-closed `except Exception` branch is exercised.""" + + def __format__(self, format_spec): + raise RuntimeError("boom") + + def __str__(self): + raise RuntimeError("boom") + + +def test_verify_audit_event_hash_fails_closed_when_computing_the_hash_raises(): + """Return False, not raise, when compute_audit_event_hash itself raises.""" + record = _base_audit_event(action=_RaisesOnFormat()) + record["record_integrity_hash"] = "irrelevant" + assert verify_audit_event_hash(record, SECRET) is False + + +def test_verify_quarantine_record_hash_fails_closed_when_computing_the_hash_raises(): + """Return False, not raise, when compute_quarantine_record_hash itself raises.""" + record = _base_quarantine_record(failure_category=_RaisesOnFormat()) + record["record_integrity_hash"] = "irrelevant" + assert verify_quarantine_record_hash(record, SECRET) is False + + def test_record_integrity_hash_differs_from_producer_signature_for_equivalent_content(): """Produce a record-integrity hash that never collides with the producer's own signature for the same content.""" diff --git a/tests/test_redis_acl_safety.py b/tests/test_redis_acl_safety.py index 175937d..1f5f077 100644 --- a/tests/test_redis_acl_safety.py +++ b/tests/test_redis_acl_safety.py @@ -29,6 +29,7 @@ import shutil import subprocess import time +from unittest.mock import MagicMock, patch import pytest import redis @@ -780,3 +781,207 @@ def test_only_run_and_close_are_the_public_surface(self): # entire intended public surface -- see the module/class # docstrings for the honest limit of this guarantee (no true # private attributes exist in Python). + + +# --------------------------------------------------------------------------- +# Pure unit tests: defensive/plumbing branches not covered by the incident- +# reproduction tests above -- RedisEnvironment's value-object protocol, +# DisposableAttestation.generate_nonce, gate construction against an +# UNKNOWN (not merely unreachable) environment, AuthenticatedSession's +# context-manager/double-close/best-effort-cleanup behavior, and +# authenticate()'s own error-shape branches. These are generic Python +# defensive code, not the AUTH/ACL security semantics the module's +# docstring says fakeredis can't be trusted for, so a mocked +# `redis.Redis` is used here deliberately -- it never stands in for a +# real authorization decision anywhere in this file. +# --------------------------------------------------------------------------- + + +class TestRedisEnvironmentValueObjectProtocol: + """Validate RedisEnvironment's __repr__/__eq__/__hash__, which nothing else in this module's + happy paths exercises (those compare with `is`, not `==`).""" + + def test_repr_names_the_instance(self): + """Render each RedisEnvironment singleton's repr with its own name.""" + assert repr(RedisEnvironment.DISPOSABLE) == "RedisEnvironment.DISPOSABLE" + assert repr(RedisEnvironment.PRODUCTION) == "RedisEnvironment.PRODUCTION" + assert repr(RedisEnvironment.UNKNOWN) == "RedisEnvironment.UNKNOWN" + + def test_equality_is_identity_based(self): + """Equate a RedisEnvironment only with itself, never with a different singleton.""" + assert RedisEnvironment.DISPOSABLE == RedisEnvironment.DISPOSABLE + assert RedisEnvironment.DISPOSABLE != RedisEnvironment.PRODUCTION + assert RedisEnvironment.DISPOSABLE != "DISPOSABLE" + + def test_hash_is_stable_and_usable_in_a_set(self): + """Hash a RedisEnvironment stably enough to use it in a set/dict key.""" + assert hash(RedisEnvironment.DISPOSABLE) == hash(RedisEnvironment.DISPOSABLE) + assert {RedisEnvironment.DISPOSABLE, RedisEnvironment.DISPOSABLE, RedisEnvironment.PRODUCTION} == { + RedisEnvironment.DISPOSABLE, RedisEnvironment.PRODUCTION, + } + + +def test_generate_nonce_returns_a_fresh_unguessable_value_each_call(): + """Generate a fresh, differing hex nonce on every call.""" + a = DisposableAttestation.generate_nonce() + b = DisposableAttestation.generate_nonce() + assert a != b + assert len(a) == 32 # secrets.token_hex(16) -> 32 hex chars + int(a, 16) # must actually be hex + + +def test_safe_repr_command_handles_an_empty_command(): + """Render '' instead of raising for a command that normalizes to nothing.""" + assert _safe_repr_command([]) == "" + assert _safe_repr_command([""]) == "" + + +class TestGateConstructionAgainstUnknownEnvironment: + """DisposableValidator/DisposableNegativeTestGate must refuse construction against UNKNOWN + (no attestation supplied), not merely against an unreachable target -- the same fail-closed + discipline classify_environment itself documents.""" + + def test_disposable_validator_rejects_missing_attestation(self): + """Refuse to construct a DisposableValidator with no attestation at all.""" + with pytest.raises(EnvironmentClassificationError): + DisposableValidator(host="127.0.0.1", port=16399, attestation=None) + + def test_negative_test_gate_rejects_missing_attestation(self): + """Refuse to construct a DisposableNegativeTestGate with no attestation at all.""" + with pytest.raises(EnvironmentClassificationError): + DisposableNegativeTestGate(host="127.0.0.1", port=16399, attestation=None) + + +class TestClassifyEnvironmentProbeCleanup: + """classify_environment's own best-effort probe.close() must never let a cleanup failure mask + (or crash out of) an otherwise-successful classification.""" + + def test_probe_close_failure_does_not_prevent_a_disposable_result(self): + """Still classify DISPOSABLE even when the verification probe's own close() raises.""" + attestation = DisposableAttestation(host="127.0.0.1", port=16399, nonce_key="k", nonce_value="v") + mock_probe = MagicMock() + mock_probe.execute_command.return_value = "v" + mock_probe.close.side_effect = RuntimeError("cleanup failed") + with patch("scripts.redis_acl_safety.redis.Redis", return_value=mock_probe): + env = classify_environment(host="127.0.0.1", port=16399, attestation=attestation) + assert env is RedisEnvironment.DISPOSABLE + + +class TestAuthenticatedSessionLifecycleWithMockedClient: + """AuthenticatedSession's close()/context-manager plumbing, exercised against a mocked + underlying client -- this class's own logic, not Redis's AUTH/ACL semantics.""" + + def _session(self, client=None): + return AuthenticatedSession( + environment=RedisEnvironment.DISPOSABLE, gate=ProductionValidator(), + client=client or MagicMock(), + ) + + def test_double_close_is_a_safe_no_op(self): + """Close a session twice without error, closing the underlying client only once.""" + mock_client = MagicMock() + session = self._session(mock_client) + session.close() + session.close() + mock_client.close.assert_called_once() + + def test_close_swallows_a_failure_from_the_underlying_client(self): + """Swallow (not raise) an exception from the underlying client's own close().""" + mock_client = MagicMock() + mock_client.close.side_effect = RuntimeError("boom") + session = self._session(mock_client) + session.close() # must not raise + assert session._closed is True + + def test_context_manager_closes_on_exit(self): + """Close the underlying client on context-manager exit.""" + mock_client = MagicMock() + with self._session(mock_client) as session: + assert session.environment is RedisEnvironment.DISPOSABLE + mock_client.close.assert_called_once() + + +class TestAuthenticateErrorShapesWithMockedClient: + """authenticate()'s own error-conversion branches that the real-Redis incident-reproduction + tests above don't reach: an unexpected (non-True/"OK") AUTH result, ACL WHOAMI itself failing + after a successful AUTH, and a completely unexpected (non-RedisError) exception. None of these + depend on fakeredis's known AUTH/ACL gap -- they're this function's own control flow.""" + + def test_unexpected_auth_result_is_a_hard_failure(self): + """Raise AuthenticationFailedError when AUTH succeeds on the wire but returns neither True + nor 'OK'.""" + mock_client = MagicMock() + mock_client.execute_command.return_value = "MAYBE" + with ( + patch("scripts.redis_acl_safety.redis.Redis", return_value=mock_client), + pytest.raises(AuthenticationFailedError), + ): + authenticate( + host="127.0.0.1", port=16399, username="u", password="p", + expected_identity="u", gate=ProductionValidator(), + environment=RedisEnvironment.UNKNOWN, + ) + mock_client.close.assert_called_once() + + def test_acl_whoami_failure_after_successful_auth_is_identity_mismatch(self): + """Raise IdentityMismatchError when ACL WHOAMI itself fails (e.g. NOPERM) even though AUTH + already succeeded.""" + mock_client = MagicMock() + mock_client.execute_command.side_effect = [True, redis.RedisError("NOPERM")] + with ( + patch("scripts.redis_acl_safety.redis.Redis", return_value=mock_client), + pytest.raises(IdentityMismatchError), + ): + authenticate( + host="127.0.0.1", port=16399, username="u", password="p", + expected_identity="u", gate=ProductionValidator(), + environment=RedisEnvironment.UNKNOWN, + ) + mock_client.close.assert_called_once() + + def test_wrong_identity_after_successful_auth_is_identity_mismatch(self): + """Raise IdentityMismatchError when ACL WHOAMI answers with a different identity than + expected.""" + mock_client = MagicMock() + mock_client.execute_command.side_effect = [True, "someone_else"] + with ( + patch("scripts.redis_acl_safety.redis.Redis", return_value=mock_client), + pytest.raises(IdentityMismatchError), + ): + authenticate( + host="127.0.0.1", port=16399, username="u", password="p", + expected_identity="u", gate=ProductionValidator(), + environment=RedisEnvironment.UNKNOWN, + ) + + def test_completely_unexpected_exception_is_converted_not_leaked_raw(self): + """Convert a totally unexpected exception (not a redis.RedisError) into + AuthenticationFailedError, never let it escape raw -- and still close whatever connection + was opened.""" + mock_client = MagicMock() + mock_client.execute_command.side_effect = TypeError("something structurally unexpected") + with ( + patch("scripts.redis_acl_safety.redis.Redis", return_value=mock_client), + pytest.raises(AuthenticationFailedError), + ): + authenticate( + host="127.0.0.1", port=16399, username="u", password="p", + expected_identity="u", gate=ProductionValidator(), + environment=RedisEnvironment.UNKNOWN, + ) + mock_client.close.assert_called_once() + + def test_authenticate_production_delegates_into_authenticate(self): + """Prove authenticate_production actually calls through to authenticate() (not just + validates the port) by observing a mocked AUTH failure propagate as + AuthenticationFailedError.""" + mock_client = MagicMock() + mock_client.execute_command.side_effect = redis.AuthenticationError("bad password") + with ( + patch("scripts.redis_acl_safety.redis.Redis", return_value=mock_client), + pytest.raises(AuthenticationFailedError), + ): + authenticate_production( + host="127.0.0.1", port=PRODUCTION_PORT, username="u", password="p", + expected_identity="u", + ) diff --git a/tests/test_verify_audit_integrity.py b/tests/test_verify_audit_integrity.py new file mode 100644 index 0000000..720e34a --- /dev/null +++ b/tests/test_verify_audit_integrity.py @@ -0,0 +1,381 @@ +"""Unit tests for scripts/verify_audit_integrity.py -- V2-003 (Track E3) +read-only stored-record integrity verification. + +verify_audit_events/verify_quarantine_records/_report/_write_status_file +are exercised against a real (SQLite) db_session, same convention as +tests/test_audit_query_service.py. Duplicate-primary-key detection is the +one case a real DB can never produce (event_id/stream_message_id are +literal primary keys) -- those two tests use a minimal fake session +whose `.query(Model).yield_per(n)` returns hand-built rows instead. + +Developer: Manish Kumar +""" +from __future__ import annotations + +import runpy +from datetime import datetime +from types import SimpleNamespace +from unittest.mock import MagicMock + +import pytest + +import scripts.verify_audit_integrity as vai +from audit.record_integrity import ( + compute_audit_event_hash, + compute_quarantine_record_hash, +) +from db.models import AuditEventRecord, QuarantinedAuditEvent + +SECRET = "test-secret-value" + + +def _audit_fields(**overrides): + fields = { + "event_id": "evt-1", + "timestamp": datetime(2026, 9, 16, 12, 0, 0), # noqa: DTZ001 -- naive column, matches AuditEventRecord.timestamp convention + "service": "test", + "event_type": "test", + "user_id": None, + "organization_id": None, + "tenant_scope": "unknown", + "action": "login", + "resource": None, + "decision": None, + "reason": None, + "trace_id": None, + "context": {}, + "integrity_status": "unsigned", + } + fields.update(overrides) + return fields + + +def _add_audit_row(db_session, hash_mode="valid", **overrides): + """Insert an AuditEventRecord. hash_mode is 'valid' (correct hash), 'tampered' (wrong hash), or + 'none' (no baseline hash at all).""" + fields = _audit_fields(**overrides) + if hash_mode == "valid": + record_integrity_hash = compute_audit_event_hash(fields, SECRET) + elif hash_mode == "tampered": + record_integrity_hash = "0" * 64 + else: + record_integrity_hash = None + row = AuditEventRecord(**fields, record_integrity_hash=record_integrity_hash) + db_session.add(row) + return row + + +def _quarantine_fields(**overrides): + fields = { + "stream_message_id": "1-0", + "raw_data": '{"service":"test"}', + "raw_signature": None, + "service": "test", + "failure_category": "malformed", + "failure_detail": "JSONDecodeError", + "delivery_attempts": 5, + } + fields.update(overrides) + return fields + + +def _add_quarantine_row(db_session, hash_mode="valid", **overrides): + fields = _quarantine_fields(**overrides) + if hash_mode == "valid": + record_integrity_hash = compute_quarantine_record_hash(fields, SECRET) + elif hash_mode == "tampered": + record_integrity_hash = "0" * 64 + else: + record_integrity_hash = None + row = QuarantinedAuditEvent(**fields, record_integrity_hash=record_integrity_hash) + db_session.add(row) + return row + + +# --------------------------------------------------------------------------- +# _resolve_secret / _resolve_session +# --------------------------------------------------------------------------- + + +def test_resolve_secret_returns_configured_value(monkeypatch): + """Return the configured JWT_SECRET value.""" + monkeypatch.setenv("JWT_SECRET", "a-real-secret") + assert vai._resolve_secret() == "a-real-secret" + + +def test_resolve_secret_exits_when_unset(monkeypatch, capsys): + """Exit 1 with a [FAIL] message when JWT_SECRET is unset.""" + monkeypatch.delenv("JWT_SECRET", raising=False) + with pytest.raises(SystemExit) as exc_info: + vai._resolve_secret() + assert exc_info.value.code == 1 + assert "JWT_SECRET must be set" in capsys.readouterr().err + + +def test_resolve_session_exits_when_no_url_configured(monkeypatch, capsys): + """Exit 1 when neither AUDIT_READER_DATABASE_URL nor AUDIT_DATABASE_URL is set.""" + monkeypatch.delenv("AUDIT_READER_DATABASE_URL", raising=False) + monkeypatch.delenv("AUDIT_DATABASE_URL", raising=False) + with pytest.raises(SystemExit) as exc_info: + vai._resolve_session() + assert exc_info.value.code == 1 + assert "AUDIT_READER_DATABASE_URL" in capsys.readouterr().err + + +def test_resolve_session_uses_reader_url(monkeypatch): + """Build a working session from AUDIT_READER_DATABASE_URL when set.""" + monkeypatch.setenv("AUDIT_READER_DATABASE_URL", "sqlite:///:memory:") + monkeypatch.delenv("AUDIT_DATABASE_URL", raising=False) + session = vai._resolve_session() + try: + assert session.execute(__import__("sqlalchemy").text("SELECT 1")).scalar() == 1 + finally: + session.close() + + +def test_resolve_session_falls_back_to_audit_database_url(monkeypatch): + """Fall back to AUDIT_DATABASE_URL when AUDIT_READER_DATABASE_URL is unset.""" + monkeypatch.delenv("AUDIT_READER_DATABASE_URL", raising=False) + monkeypatch.setenv("AUDIT_DATABASE_URL", "sqlite:///:memory:") + session = vai._resolve_session() + session.close() + + +# --------------------------------------------------------------------------- +# verify_audit_events +# --------------------------------------------------------------------------- + + +def test_verify_audit_events_counts_valid_tampered_and_no_baseline(db_session): + """Classify a correctly-hashed row as valid, a wrong-hash row as invalid, and a hash-less row + as no_baseline.""" + _add_audit_row(db_session, hash_mode="valid", event_id="e-valid") + _add_audit_row(db_session, hash_mode="tampered", event_id="e-tampered") + _add_audit_row(db_session, hash_mode="none", event_id="e-no-baseline") + db_session.commit() + + stats = vai.verify_audit_events(db_session, SECRET) + + assert stats["checked"] == 3 + assert stats["valid"] == 1 + assert stats["invalid"] == ["e-tampered"] + assert stats["no_baseline"] == 1 + assert stats["structural"] == [] + + +def test_verify_audit_events_flags_empty_required_field_as_structural(): + """Flag a row whose required field is an empty string as a structural problem, via a fake + session (avoids the DB's own NOT NULL constraint semantics).""" + row = SimpleNamespace(**_audit_fields(action=""), record_integrity_hash=None) + fake_session = MagicMock() + fake_session.query.return_value.yield_per.return_value = [row] + + stats = vai.verify_audit_events(fake_session, SECRET) + + assert stats["checked"] == 1 + assert any("action" in s for s in stats["structural"]) + + +def test_verify_audit_events_flags_duplicate_event_id_as_structural(): + """Flag a second row sharing an event_id already seen as a structural duplicate.""" + row_a = SimpleNamespace(**_audit_fields(event_id="dup-1"), record_integrity_hash=None) + row_b = SimpleNamespace(**_audit_fields(event_id="dup-1"), record_integrity_hash=None) + fake_session = MagicMock() + fake_session.query.return_value.yield_per.return_value = [row_a, row_b] + + stats = vai.verify_audit_events(fake_session, SECRET) + + assert stats["checked"] == 2 + assert any("duplicate event_id" in s for s in stats["structural"]) + + +# --------------------------------------------------------------------------- +# verify_quarantine_records +# --------------------------------------------------------------------------- + + +def test_verify_quarantine_records_counts_valid_tampered_and_no_baseline(db_session): + """Classify quarantine rows the same way verify_audit_events classifies audit-event rows.""" + _add_quarantine_row(db_session, hash_mode="valid", stream_message_id="q-valid") + _add_quarantine_row(db_session, hash_mode="tampered", stream_message_id="q-tampered") + _add_quarantine_row(db_session, hash_mode="none", stream_message_id="q-no-baseline") + db_session.commit() + + stats = vai.verify_quarantine_records(db_session, SECRET) + + assert stats["checked"] == 3 + assert stats["valid"] == 1 + assert stats["invalid"] == ["q-tampered"] + assert stats["no_baseline"] == 1 + assert stats["structural"] == [] + + +def test_verify_quarantine_records_flags_empty_required_field_as_structural(): + """Flag a quarantine row whose required field is an empty string as structural.""" + row = SimpleNamespace(**_quarantine_fields(failure_category=""), record_integrity_hash=None) + fake_session = MagicMock() + fake_session.query.return_value.yield_per.return_value = [row] + + stats = vai.verify_quarantine_records(fake_session, SECRET) + + assert any("failure_category" in s for s in stats["structural"]) + + +def test_verify_quarantine_records_flags_duplicate_stream_message_id_as_structural(): + """Flag a second row sharing a stream_message_id already seen as a structural duplicate.""" + row_a = SimpleNamespace(**_quarantine_fields(stream_message_id="dup-1"), record_integrity_hash=None) + row_b = SimpleNamespace(**_quarantine_fields(stream_message_id="dup-1"), record_integrity_hash=None) + fake_session = MagicMock() + fake_session.query.return_value.yield_per.return_value = [row_a, row_b] + + stats = vai.verify_quarantine_records(fake_session, SECRET) + + assert any("duplicate stream_message_id" in s for s in stats["structural"]) + + +# --------------------------------------------------------------------------- +# _report +# --------------------------------------------------------------------------- + + +def test_report_all_clear_prints_ok_and_returns_true(capsys): + """Print an [OK] summary and return True when nothing is wrong.""" + stats = {"checked": 5, "valid": 5, "no_baseline": 0, "invalid": [], "structural": []} + assert vai._report("audit_events", stats) is True + out = capsys.readouterr().out + assert "[OK] audit_events: no integrity or structural problems found" in out + + +def test_report_invalid_records_prints_fail_and_returns_false(capsys): + """Print [FAIL] lines naming each tampered record and return False.""" + stats = {"checked": 2, "valid": 1, "no_baseline": 0, "invalid": ["evt-1"], "structural": []} + assert vai._report("audit_events", stats) is False + err = capsys.readouterr().err + assert "1 record(s) failed integrity verification" in err + assert "TAMPERED OR CORRUPTED: evt-1" in err + + +def test_report_structural_problems_prints_fail_and_returns_false(capsys): + """Print [FAIL] lines naming each structural problem and return False.""" + stats = {"checked": 1, "valid": 0, "no_baseline": 0, "invalid": [], "structural": ["evt-1: missing required field 'action'"]} + assert vai._report("audit_events", stats) is False + err = capsys.readouterr().err + assert "1 structural problem(s)" in err + assert "evt-1: missing required field 'action'" in err + + +# --------------------------------------------------------------------------- +# _write_status_file +# --------------------------------------------------------------------------- + + +def test_write_status_file_is_a_no_op_when_unconfigured(monkeypatch, tmp_path): + """Write nothing when AUDIT_HEALTH_STATUS_DIR is unset.""" + monkeypatch.delenv("AUDIT_HEALTH_STATUS_DIR", raising=False) + vai._write_status_file(True, {"checked": 0, "invalid": []}, {"checked": 0, "invalid": []}) + assert list(tmp_path.iterdir()) == [] + + +def test_write_status_file_writes_expected_fields(monkeypatch, tmp_path): + """Write every expected KEY=VALUE line when AUDIT_HEALTH_STATUS_DIR is set.""" + monkeypatch.setenv("AUDIT_HEALTH_STATUS_DIR", str(tmp_path)) + events_stats = {"checked": 10, "invalid": ["e1"]} + quarantine_stats = {"checked": 2, "invalid": []} + + vai._write_status_file(False, events_stats, quarantine_stats) + + content = (tmp_path / "audit-integrity-verification.env").read_text() + assert "LAST_VERIFICATION_RESULT=fail" in content + assert "LAST_VERIFICATION_EVENTS_CHECKED=10" in content + assert "LAST_VERIFICATION_EVENTS_INVALID=1" in content + assert "LAST_VERIFICATION_QUARANTINE_CHECKED=2" in content + assert "LAST_VERIFICATION_QUARANTINE_INVALID=0" in content + + +# --------------------------------------------------------------------------- +# main() +# --------------------------------------------------------------------------- + + +def test_main_passes_and_exits_zero_with_no_alert(monkeypatch, db_session, capsys): + """Exit 0, print [OK], and never emit a security alert when everything verifies clean.""" + monkeypatch.setenv("JWT_SECRET", SECRET) + monkeypatch.delenv("AUDIT_HEALTH_STATUS_DIR", raising=False) + monkeypatch.setattr(vai, "_resolve_secret", lambda: SECRET) + monkeypatch.setattr(vai, "_resolve_session", lambda: db_session) + alerts = [] + monkeypatch.setattr(vai, "emit_security_alert", lambda **kwargs: alerts.append(kwargs)) + _add_audit_row(db_session, hash_mode="valid", event_id="e1") + _add_quarantine_row(db_session, hash_mode="valid", stream_message_id="q1") + db_session.commit() + + exit_code = vai.main() + + assert exit_code == 0 + assert "[OK] integrity verification passed" in capsys.readouterr().out + assert alerts == [] + + +def test_main_fails_exits_one_and_emits_alert(monkeypatch, db_session, capsys): + """Exit 1, print [FAIL], and emit a critical security alert when a tampered record is found.""" + monkeypatch.setenv("JWT_SECRET", SECRET) + monkeypatch.delenv("AUDIT_HEALTH_STATUS_DIR", raising=False) + monkeypatch.setattr(vai, "_resolve_secret", lambda: SECRET) + monkeypatch.setattr(vai, "_resolve_session", lambda: db_session) + alerts = [] + monkeypatch.setattr(vai, "emit_security_alert", lambda **kwargs: alerts.append(kwargs)) + _add_audit_row(db_session, hash_mode="tampered", event_id="e-bad") + db_session.commit() + + exit_code = vai.main() + + assert exit_code == 1 + assert "[FAIL] integrity verification found problems" in capsys.readouterr().err + assert len(alerts) == 1 + assert alerts[0]["condition"] == "integrity_verification_failed" + assert alerts[0]["severity"] == "critical" + assert alerts[0]["metadata"]["events_invalid"] == 1 + + +def test_main_closes_the_session_even_when_verification_raises(monkeypatch): + """Close the resolved session in a finally block, even when verification itself raises.""" + monkeypatch.setattr(vai, "_resolve_secret", lambda: SECRET) + broken_session = MagicMock() + broken_session.query.side_effect = RuntimeError("db exploded") + monkeypatch.setattr(vai, "_resolve_session", lambda: broken_session) + + with pytest.raises(RuntimeError): + vai.main() + + broken_session.close.assert_called_once() + + +def test_main_writes_status_file_when_configured(monkeypatch, db_session, tmp_path): + """Write the operational status file as part of main()'s run when AUDIT_HEALTH_STATUS_DIR is + configured.""" + monkeypatch.setenv("JWT_SECRET", SECRET) + monkeypatch.setenv("AUDIT_HEALTH_STATUS_DIR", str(tmp_path)) + monkeypatch.setattr(vai, "_resolve_secret", lambda: SECRET) + monkeypatch.setattr(vai, "_resolve_session", lambda: db_session) + _add_audit_row(db_session, hash_mode="valid", event_id="e1") + db_session.commit() + + exit_code = vai.main() + + assert exit_code == 0 + assert (tmp_path / "audit-integrity-verification.env").exists() + + +# --------------------------------------------------------------------------- +# __main__ guard +# --------------------------------------------------------------------------- + + +def test_dunder_main_exits_with_mains_return_code(monkeypatch): + """Exit with main()'s own return code when run as a script.""" + monkeypatch.delenv("JWT_SECRET", raising=False) # cheapest deterministic path: _resolve_secret's own sys.exit(1) + + with pytest.raises(SystemExit) as exc_info: + runpy.run_path(vai.__file__, run_name="__main__") + + assert exc_info.value.code == 1