diff --git a/tests/mesh/test_audit_read_walker_defenses.py b/tests/mesh/test_audit_read_walker_defenses.py new file mode 100644 index 000000000..c00076da2 --- /dev/null +++ b/tests/mesh/test_audit_read_walker_defenses.py @@ -0,0 +1,130 @@ +"""Forensic-walker defenses for :mod:`strands_robots.mesh.audit`. + +``read_audit_log`` is both the operator-facing forensic reader and the +seed source for ``_load_seq_counters`` when the seq sidecar is corrupt. +That dual role makes its file-handling discipline security-relevant: + +* A rotated log file swapped for a SYMLINK must be refused, not followed + (mirrors the O_NOFOLLOW discipline applied to every other open in the + module). Following it would let an attacker redirect the forensic read + to ``/dev/null`` (fail-open seq reset) or to forged content. +* ``_audit_log_files_in_order`` must only treat files whose suffix after + the active-log name is purely numeric as rotated copies; an unrelated + sibling such as ``mesh_audit.jsonl.bak`` must be ignored so its bytes + never enter the forensic stream. +* The ``since=`` cutoff must drop records strictly older than the + timestamp (and records with a non-numeric ``ts``) while keeping newer + ones, so a time-bounded forensic query returns only the window asked + for. + +These behaviors had no direct regression coverage; this file pins them. +""" + +from __future__ import annotations + +import os + +import pytest + +from strands_robots.mesh import audit + + +@pytest.fixture(autouse=True) +def _isolated_audit(monkeypatch, tmp_path): + """Fresh audit dir + reset module state for each test.""" + monkeypatch.setenv("STRANDS_MESH_AUDIT_DIR", str(tmp_path)) + monkeypatch.delenv("STRANDS_MESH_AUDIT_PSK", raising=False) + audit._SEQ_COUNTERS.clear() + audit._AUDIT_STATE.seq_loaded = False + audit._AUDIT_STATE.audit_log_seeded = False + audit._AUDIT_STATE.psk_fingerprint = None + yield + audit._SEQ_COUNTERS.clear() + audit._AUDIT_STATE.seq_loaded = False + audit._AUDIT_STATE.audit_log_seeded = False + audit._AUDIT_STATE.psk_fingerprint = None + + +@pytest.mark.skipif(not hasattr(os, "symlink"), reason="platform without symlink support") +def test_read_audit_log_refuses_symlinked_rotated_file(tmp_path, monkeypatch, caplog): + """A rotated log file that is a SYMLINK must be skipped, not followed. + + The active log is read normally; the symlinked rotated copy's target + bytes must never appear in the returned records. + """ + # Write a legitimate record to the active log. + audit.log_safety_event("real_event", "peer-a", {"index": 1}) + active = audit.audit_log_path() + assert active.exists() + + # Plant attacker-controlled bytes outside the audit dir, then point a + # rotated-log path at them via a symlink. + secret = tmp_path / "attacker_payload.jsonl" + secret.write_text('{"event": "forged", "payload": {"index": 999}}\n', encoding="utf-8") + rotated = active.with_suffix(active.suffix + ".1") + os.symlink(secret, rotated) + assert rotated.is_symlink() + + with caplog.at_level("WARNING"): + records = audit.read_audit_log() + + events = [r.get("event") for r in records] + assert "real_event" in events, "active log record should still be read" + assert "forged" not in events, "symlinked rotated log must not be followed" + assert any("refusing to read" in m and "SYMLINK" in m for m in caplog.messages), ( + f"expected a symlink-refusal warning, got {caplog.messages}" + ) + + +def test_files_in_order_ignores_non_numeric_suffix_siblings(tmp_path, monkeypatch): + """Only ``.`` siblings count as rotated copies. + + A sibling like ``mesh_audit.jsonl.bak`` must be excluded so its bytes + never enter the forensic read stream. + """ + audit.log_safety_event("real_event", "peer-a", {"index": 1}) + active = audit.audit_log_path() + + # A genuine rotated copy (numeric suffix) and a decoy (non-numeric). + rotated = active.with_suffix(active.suffix + ".1") + rotated.write_text('{"event": "rotated_event", "payload": {"index": 0}}\n', encoding="utf-8") + decoy = active.with_suffix(active.suffix + ".bak") + decoy.write_text('{"event": "decoy_event", "payload": {"index": -1}}\n', encoding="utf-8") + + ordered = audit._audit_log_files_in_order() + names = [p.name for p in ordered] + assert rotated.name in names, "numeric-suffix rotated copy must be included" + assert decoy.name not in names, "non-numeric-suffix sibling must be excluded" + # Chronological order: oldest rotated copy first, active log last. + assert names[-1] == active.name + + events = [r.get("event") for r in audit.read_audit_log()] + assert "rotated_event" in events + assert "real_event" in events + assert "decoy_event" not in events + + +def test_read_audit_log_since_filters_old_and_non_numeric_ts(tmp_path, monkeypatch): + """``since=`` keeps records at/after the cutoff and drops older ones. + + A record whose ``ts`` is missing or non-numeric is also dropped, since + it cannot be placed in the requested time window. + """ + active = audit.audit_log_path() + active.parent.mkdir(parents=True, exist_ok=True) + active.write_text( + "\n".join( + [ + '{"event": "old", "ts": 100.0, "payload": {}}', + '{"event": "new", "ts": 200.0, "payload": {}}', + '{"event": "no_ts", "payload": {}}', + '{"event": "bad_ts", "ts": "not-a-number", "payload": {}}', + ] + ) + + "\n", + encoding="utf-8", + ) + + records = audit.read_audit_log(since=150.0) + events = [r.get("event") for r in records] + assert events == ["new"], f"since= should keep only records at/after the cutoff, got {events}"