From 59fce18c4dca4312daf0f11168873cc0231b78c9 Mon Sep 17 00:00:00 2001 From: rookiestar28 Date: Thu, 16 Apr 2026 16:06:03 +0800 Subject: [PATCH] refactor(audit): add retained-chain verifier --- scripts/verify_audit_chain.py | 56 ++++ services/audit.py | 133 ++++---- services/audit_pipeline.py | 306 +++++++++++++++++++ services/security_doctor_connector_checks.py | 6 +- services/security_doctor_endpoint_checks.py | 6 +- services/security_doctor_runtime_checks.py | 6 +- tests/security/test_audit.py | 8 + tests/security/test_s78_redaction.py | 8 + tests/test_r159_audit_pipeline.py | 165 ++++++++++ 9 files changed, 628 insertions(+), 66 deletions(-) create mode 100644 scripts/verify_audit_chain.py create mode 100644 services/audit_pipeline.py create mode 100644 tests/test_r159_audit_pipeline.py diff --git a/scripts/verify_audit_chain.py b/scripts/verify_audit_chain.py new file mode 100644 index 0000000..81ad7ec --- /dev/null +++ b/scripts/verify_audit_chain.py @@ -0,0 +1,56 @@ +"""Verify the retained OpenClaw audit hash chain across active and rotated files.""" + +from __future__ import annotations + +import argparse +import json +from pathlib import Path + + +def _repo_root() -> Path: + return Path(__file__).resolve().parents[1] + + +def main() -> int: + parser = argparse.ArgumentParser( + description="Verify the retained OpenClaw audit hash chain." + ) + parser.add_argument( + "--path", + default="", + help="Explicit audit log path. Defaults to the configured OpenClaw audit path.", + ) + parser.add_argument( + "--json", + action="store_true", + help="Emit machine-readable JSON instead of a human summary.", + ) + args = parser.parse_args() + + import sys + + sys.path.insert(0, str(_repo_root())) + from services.audit import verify_audit_chain + + result = verify_audit_chain(args.path or None) + if args.json: + print(json.dumps(result.to_dict(), indent=2)) + else: + status = "PASS" if result.ok else "FAIL" + print(f"Audit Chain Verification: {status}") + print(f"Files checked: {len(result.files_checked)}") + print(f"Entries checked: {result.entries_checked}") + print(f"Window start prev_hash: {result.window_start_prev_hash}") + print(f"Terminal hash: {result.terminal_hash}") + print(f"Window truncated: {result.window_truncated}") + if result.issues: + print("Issues:") + for issue in result.issues: + print( + f"- {issue.code} at {issue.file_path}:{issue.line_number} -> {issue.message}" + ) + return 0 if result.ok else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/services/audit.py b/services/audit.py index 9249861..d5d41c4 100644 --- a/services/audit.py +++ b/services/audit.py @@ -13,6 +13,12 @@ import time import uuid from typing import Any, Dict, Optional, Tuple +from .audit_pipeline import ( + AuditVerificationResult, + LocalFileAuditSink, + read_last_entry_hash_from_chain, +) +from .audit_pipeline import verify_audit_chain as verify_audit_chain_impl from .redaction import redact_json, stable_redaction_tag logger = logging.getLogger("ComfyUI-OpenClaw.services.audit") @@ -44,6 +50,15 @@ AUDIT_LOG_PATH = ( ) +def _default_audit_chain_key_path() -> str: + override = os.environ.get("OPENCLAW_AUDIT_CHAIN_KEY_PATH") or os.environ.get( + "MOLTBOT_AUDIT_CHAIN_KEY_PATH" + ) + if override: + return override + return f"{AUDIT_LOG_PATH}.key" + + def _env_int(primary: str, legacy: str, default: int) -> int: raw = os.environ.get(primary) or os.environ.get(legacy) if raw is None: @@ -109,7 +124,41 @@ def _get_audit_chain_key() -> bytes: raw = os.environ.get("OPENCLAW_AUDIT_CHAIN_KEY") or os.environ.get( "MOLTBOT_AUDIT_CHAIN_KEY" ) - _AUDIT_CHAIN_KEY = raw.encode("utf-8") if raw else secrets.token_bytes(32) + if raw: + _AUDIT_CHAIN_KEY = raw.encode("utf-8") + return _AUDIT_CHAIN_KEY + + key_path = _default_audit_chain_key_path() + try: + with open(key_path, "r", encoding="utf-8") as handle: + persisted = handle.read().strip() + if persisted: + _AUDIT_CHAIN_KEY = bytes.fromhex(persisted) + return _AUDIT_CHAIN_KEY + except Exception: + pass + + # CRITICAL: the audit chain key must stay stable across restarts or + # rotated-chain verification becomes impossible after the first reboot. + generated = secrets.token_bytes(32) + try: + os.makedirs(os.path.dirname(os.path.abspath(key_path)), exist_ok=True) + except Exception: + pass + try: + with open(key_path, "x", encoding="utf-8") as handle: + handle.write(generated.hex()) + try: + os.chmod(key_path, 0o600) + except Exception: + pass + _AUDIT_CHAIN_KEY = generated + except FileExistsError: + with open(key_path, "r", encoding="utf-8") as handle: + persisted = handle.read().strip() + _AUDIT_CHAIN_KEY = bytes.fromhex(persisted) if persisted else generated + except Exception: + _AUDIT_CHAIN_KEY = generated return _AUDIT_CHAIN_KEY @@ -126,54 +175,25 @@ def _chain_hash(prev_hash: str, entry: Dict[str, Any]) -> str: ).hex() -def _rotate_if_needed(path: str) -> None: +def _build_audit_sink(path: str) -> LocalFileAuditSink: max_bytes, backups = _audit_limits() - if max_bytes <= 0 or backups < 0: - return - if not os.path.exists(path): - return + return LocalFileAuditSink( + path=path, + max_bytes=max_bytes, + backups=backups, + chain_hash=_chain_hash, + ) + + +def _rotate_if_needed(path: str) -> None: try: - if os.path.getsize(path) < max_bytes: - return - if backups == 0: - os.remove(path) - return - for idx in range(backups, 0, -1): - src = f"{path}.{idx}" - dst = f"{path}.{idx + 1}" - if os.path.exists(src): - if idx == backups: - os.remove(src) - else: - os.replace(src, dst) - os.replace(path, f"{path}.1") + _build_audit_sink(path).rotate_if_needed() except Exception as exc: logger.error("Audit rotation failed: %s", exc) def _read_last_entry_hash(path: str) -> str: - # Best-effort bootstrap. Fall back to genesis if file is empty/unreadable. - if not os.path.exists(path): - return "GENESIS" - try: - with open(path, "rb") as f: - f.seek(0, os.SEEK_END) - size = f.tell() - if size <= 0: - return "GENESIS" - step = min(size, 8192) - f.seek(size - step) - data = f.read().decode("utf-8", errors="ignore") - lines = [line.strip() for line in data.splitlines() if line.strip()] - if not lines: - return "GENESIS" - last = json.loads(lines[-1]) - last_hash = last.get("entry_hash") - if isinstance(last_hash, str) and last_hash: - return last_hash - except Exception: - pass - return "GENESIS" + return read_last_entry_hash_from_chain(path) _LAST_HASH: Optional[str] = None @@ -182,29 +202,12 @@ _AUDIT_WRITE_LOCK = threading.Lock() def _write_audit_entry(entry: Dict[str, Any]) -> None: global _LAST_HASH - try: - os.makedirs(os.path.dirname(os.path.abspath(AUDIT_LOG_PATH)), exist_ok=True) - except Exception: - # Path may be relative to CWD with no parent folder. - pass - - # CRITICAL: keep read->chain->append->state update atomic to avoid hash-chain forks. + sink = _build_audit_sink(AUDIT_LOG_PATH) + # CRITICAL: keep tail-resolution -> rotate -> append -> cache update atomic so + # restarts and file rotations cannot fork the retained audit chain. with _AUDIT_WRITE_LOCK: - _rotate_if_needed(AUDIT_LOG_PATH) - if _LAST_HASH is None: - _LAST_HASH = _read_last_entry_hash(AUDIT_LOG_PATH) - - prev_hash = _LAST_HASH or "GENESIS" - event_hash = _chain_hash(prev_hash, entry) - wrapped = dict(entry) - wrapped["prev_hash"] = prev_hash - wrapped["entry_hash"] = event_hash - - line = json.dumps(wrapped, sort_keys=True, ensure_ascii=True) + "\n" try: - with open(AUDIT_LOG_PATH, "a", encoding="utf-8") as f: - f.write(line) - _LAST_HASH = event_hash + _LAST_HASH = sink.append_entry(entry, last_hash=_LAST_HASH) except Exception as exc: logger.error("Failed to write audit entry: %s", exc) @@ -395,3 +398,7 @@ def audit_llm_test(actor_ip: str, ok: bool, error: Optional[str] = None) -> None else {"actor_ip_tag": stable_redaction_tag(actor_ip, label="ip")} ), ) + + +def verify_audit_chain(path: Optional[str] = None) -> AuditVerificationResult: + return verify_audit_chain_impl(path or AUDIT_LOG_PATH, chain_hash=_chain_hash) diff --git a/services/audit_pipeline.py b/services/audit_pipeline.py new file mode 100644 index 0000000..3ccf747 --- /dev/null +++ b/services/audit_pipeline.py @@ -0,0 +1,306 @@ +"""Audit persistence sinks and chain verification helpers.""" + +from __future__ import annotations + +import json +import os +import re +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, Callable, Dict, List, Optional, Protocol + + +class AuditSink(Protocol): + def append_entry(self, entry: Dict[str, Any], *, last_hash: Optional[str]) -> str: ... + + +@dataclass +class AuditVerificationIssue: + code: str + file_path: str + line_number: int + message: str + + def to_dict(self) -> Dict[str, Any]: + return { + "code": self.code, + "file_path": self.file_path, + "line_number": self.line_number, + "message": self.message, + } + + +@dataclass +class AuditVerificationResult: + ok: bool + files_checked: List[str] + entries_checked: int + window_start_prev_hash: str + terminal_hash: str + window_truncated: bool + issues: List[AuditVerificationIssue] = field(default_factory=list) + + def to_dict(self) -> Dict[str, Any]: + return { + "ok": self.ok, + "files_checked": list(self.files_checked), + "entries_checked": self.entries_checked, + "window_start_prev_hash": self.window_start_prev_hash, + "terminal_hash": self.terminal_hash, + "window_truncated": self.window_truncated, + "issues": [issue.to_dict() for issue in self.issues], + } + + +def _rotated_chain_paths(base_path: str) -> List[str]: + base = Path(base_path) + parent = base.parent if str(base.parent) else Path(".") + if not parent.exists(): + return [] + pattern = re.compile(rf"^{re.escape(base.name)}\.(\d+)$") + matches = [] + for child in parent.iterdir(): + if not child.is_file(): + continue + match = pattern.match(child.name) + if not match: + continue + matches.append((int(match.group(1)), str(child))) + matches.sort(reverse=True) + return [path for _, path in matches] + + +def iter_audit_chain_paths(base_path: str) -> List[str]: + paths = _rotated_chain_paths(base_path) + if os.path.exists(base_path): + paths.append(base_path) + return paths + + +def _tail_hash_from_file(path: str) -> Optional[str]: + if not os.path.exists(path): + return None + try: + with open(path, "rb") as handle: + handle.seek(0, os.SEEK_END) + size = handle.tell() + if size <= 0: + return None + step = min(size, 8192) + handle.seek(size - step) + data = handle.read().decode("utf-8", errors="ignore") + except Exception: + return None + + lines = [line.strip() for line in data.splitlines() if line.strip()] + if not lines: + return None + try: + payload = json.loads(lines[-1]) + except Exception: + return None + value = payload.get("entry_hash") + return value if isinstance(value, str) and value else None + + +def read_last_entry_hash_from_chain(base_path: str) -> str: + newest_first = [base_path] if os.path.exists(base_path) else [] + newest_first.extend(reversed(_rotated_chain_paths(base_path))) + for path in newest_first: + value = _tail_hash_from_file(path) + if value: + return value + return "GENESIS" + + +class LocalFileAuditSink: + def __init__( + self, + *, + path: str, + max_bytes: int, + backups: int, + chain_hash: Callable[[str, Dict[str, Any]], str], + ) -> None: + self.path = path + self.max_bytes = max_bytes + self.backups = backups + self._chain_hash = chain_hash + + def _ensure_parent_dir(self) -> None: + try: + os.makedirs(os.path.dirname(os.path.abspath(self.path)), exist_ok=True) + except Exception: + pass + + def rotate_if_needed(self) -> None: + if self.max_bytes <= 0 or self.backups < 0: + return + if not os.path.exists(self.path): + return + if os.path.getsize(self.path) < self.max_bytes: + return + if self.backups == 0: + os.remove(self.path) + return + for idx in range(self.backups, 0, -1): + src = f"{self.path}.{idx}" + dst = f"{self.path}.{idx + 1}" + if os.path.exists(src): + if idx == self.backups: + os.remove(src) + else: + os.replace(src, dst) + os.replace(self.path, f"{self.path}.1") + + def append_entry(self, entry: Dict[str, Any], *, last_hash: Optional[str]) -> str: + self._ensure_parent_dir() + self.rotate_if_needed() + prev_hash = last_hash or read_last_entry_hash_from_chain(self.path) + event_hash = self._chain_hash(prev_hash, entry) + wrapped = dict(entry) + wrapped["prev_hash"] = prev_hash + wrapped["entry_hash"] = event_hash + line = json.dumps(wrapped, sort_keys=True, ensure_ascii=True) + "\n" + with open(self.path, "a", encoding="utf-8") as handle: + handle.write(line) + return event_hash + + +def verify_audit_chain( + base_path: str, + *, + chain_hash: Callable[[str, Dict[str, Any]], str], +) -> AuditVerificationResult: + files = iter_audit_chain_paths(base_path) + issues: List[AuditVerificationIssue] = [] + if not files: + issues.append( + AuditVerificationIssue( + code="missing_chain", + file_path=base_path, + line_number=0, + message="No audit log files found for verification.", + ) + ) + return AuditVerificationResult( + ok=False, + files_checked=[], + entries_checked=0, + window_start_prev_hash="GENESIS", + terminal_hash="GENESIS", + window_truncated=False, + issues=issues, + ) + + previous_entry_hash: Optional[str] = None + window_start_prev_hash = "GENESIS" + entries_checked = 0 + + for file_path in files: + with open(file_path, "r", encoding="utf-8") as handle: + for line_number, raw_line in enumerate(handle, start=1): + line = raw_line.strip() + if not line: + continue + try: + wrapped = json.loads(line) + except Exception as exc: + issues.append( + AuditVerificationIssue( + code="invalid_json", + file_path=file_path, + line_number=line_number, + message=f"Invalid JSON entry: {exc}", + ) + ) + return AuditVerificationResult( + ok=False, + files_checked=files, + entries_checked=entries_checked, + window_start_prev_hash=window_start_prev_hash, + terminal_hash=previous_entry_hash or "GENESIS", + window_truncated=window_start_prev_hash != "GENESIS", + issues=issues, + ) + + prev_hash = wrapped.get("prev_hash") + entry_hash = wrapped.get("entry_hash") + if not isinstance(prev_hash, str) or not isinstance(entry_hash, str): + issues.append( + AuditVerificationIssue( + code="missing_hash_fields", + file_path=file_path, + line_number=line_number, + message="Audit entry is missing string prev_hash/entry_hash fields.", + ) + ) + return AuditVerificationResult( + ok=False, + files_checked=files, + entries_checked=entries_checked, + window_start_prev_hash=window_start_prev_hash, + terminal_hash=previous_entry_hash or "GENESIS", + window_truncated=window_start_prev_hash != "GENESIS", + issues=issues, + ) + + if previous_entry_hash is None: + window_start_prev_hash = prev_hash + elif prev_hash != previous_entry_hash: + issues.append( + AuditVerificationIssue( + code="prev_hash_mismatch", + file_path=file_path, + line_number=line_number, + message=( + "Audit chain continuity failed: prev_hash does not match " + "the preceding entry_hash." + ), + ) + ) + return AuditVerificationResult( + ok=False, + files_checked=files, + entries_checked=entries_checked, + window_start_prev_hash=window_start_prev_hash, + terminal_hash=previous_entry_hash or "GENESIS", + window_truncated=window_start_prev_hash != "GENESIS", + issues=issues, + ) + + payload = dict(wrapped) + payload.pop("prev_hash", None) + payload.pop("entry_hash", None) + expected_hash = chain_hash(prev_hash, payload) + if entry_hash != expected_hash: + issues.append( + AuditVerificationIssue( + code="entry_hash_mismatch", + file_path=file_path, + line_number=line_number, + message="Audit entry_hash does not match the persisted payload.", + ) + ) + return AuditVerificationResult( + ok=False, + files_checked=files, + entries_checked=entries_checked, + window_start_prev_hash=window_start_prev_hash, + terminal_hash=previous_entry_hash or "GENESIS", + window_truncated=window_start_prev_hash != "GENESIS", + issues=issues, + ) + + previous_entry_hash = entry_hash + entries_checked += 1 + + return AuditVerificationResult( + ok=True, + files_checked=files, + entries_checked=entries_checked, + window_start_prev_hash=window_start_prev_hash, + terminal_hash=previous_entry_hash or "GENESIS", + window_truncated=window_start_prev_hash != "GENESIS", + issues=[], + ) diff --git a/services/security_doctor_connector_checks.py b/services/security_doctor_connector_checks.py index 0dbffa2..fb83e8a 100644 --- a/services/security_doctor_connector_checks.py +++ b/services/security_doctor_connector_checks.py @@ -4,7 +4,11 @@ from __future__ import annotations import os -from .security_doctor_report import SecurityCheckResult, SecurityReport, SecuritySeverity +from .security_doctor_report import ( + SecurityCheckResult, + SecurityReport, + SecuritySeverity, +) try: from .connector_allowlist_posture import ( diff --git a/services/security_doctor_endpoint_checks.py b/services/security_doctor_endpoint_checks.py index f591647..1576f46 100644 --- a/services/security_doctor_endpoint_checks.py +++ b/services/security_doctor_endpoint_checks.py @@ -7,7 +7,11 @@ import json import os from typing import Dict -from .security_doctor_report import SecurityCheckResult, SecurityReport, SecuritySeverity +from .security_doctor_report import ( + SecurityCheckResult, + SecurityReport, + SecuritySeverity, +) try: from ..config import PACK_VERSION diff --git a/services/security_doctor_runtime_checks.py b/services/security_doctor_runtime_checks.py index d647fa7..24da8e6 100644 --- a/services/security_doctor_runtime_checks.py +++ b/services/security_doctor_runtime_checks.py @@ -8,7 +8,11 @@ import stat import sys from pathlib import Path -from .security_doctor_report import SecurityCheckResult, SecurityReport, SecuritySeverity +from .security_doctor_report import ( + SecurityCheckResult, + SecurityReport, + SecuritySeverity, +) def check_state_dir_permissions(report: SecurityReport) -> None: diff --git a/tests/security/test_audit.py b/tests/security/test_audit.py index de76588..c8235f2 100644 --- a/tests/security/test_audit.py +++ b/tests/security/test_audit.py @@ -17,23 +17,31 @@ from services.audit import ( class TestAudit(unittest.TestCase): def setUp(self): self.test_log = "test_audit.log" + self.test_key = f"{self.test_log}.key" self.path_patcher = patch("services.audit.AUDIT_LOG_PATH", self.test_log) self.path_patcher.start() self.hash_patcher = patch("services.audit._LAST_HASH", None) self.hash_patcher.start() + self.chain_key_patcher = patch("services.audit._AUDIT_CHAIN_KEY", None) + self.chain_key_patcher.start() self.tag_key_patcher = patch.object( redaction_module, "_REDACTION_TAG_KEY", b"audit-test-redaction-key" ) self.tag_key_patcher.start() if os.path.exists(self.test_log): os.remove(self.test_log) + if os.path.exists(self.test_key): + os.remove(self.test_key) def tearDown(self): self.path_patcher.stop() self.hash_patcher.stop() + self.chain_key_patcher.stop() self.tag_key_patcher.stop() if os.path.exists(self.test_log): os.remove(self.test_log) + if os.path.exists(self.test_key): + os.remove(self.test_key) def _read_entries(self): with open(self.test_log, "r", encoding="utf-8") as f: diff --git a/tests/security/test_s78_redaction.py b/tests/security/test_s78_redaction.py index d20b8a4..5fab2e9 100644 --- a/tests/security/test_s78_redaction.py +++ b/tests/security/test_s78_redaction.py @@ -153,23 +153,31 @@ class TestS78BridgeWorkerRedaction(unittest.TestCase): class TestS78AuditRedaction(unittest.TestCase): def setUp(self): self.test_log = "test_s78_audit.log" + self.test_key = f"{self.test_log}.key" self.path_patcher = patch("services.audit.AUDIT_LOG_PATH", self.test_log) self.hash_patcher = patch("services.audit._LAST_HASH", None) + self.chain_key_patcher = patch("services.audit._AUDIT_CHAIN_KEY", None) self.tag_key_patcher = patch.object( redaction_module, "_REDACTION_TAG_KEY", b"s78-test-redaction-key" ) self.path_patcher.start() self.hash_patcher.start() + self.chain_key_patcher.start() self.tag_key_patcher.start() if os.path.exists(self.test_log): os.remove(self.test_log) + if os.path.exists(self.test_key): + os.remove(self.test_key) def tearDown(self): self.path_patcher.stop() self.hash_patcher.stop() + self.chain_key_patcher.stop() self.tag_key_patcher.stop() if os.path.exists(self.test_log): os.remove(self.test_log) + if os.path.exists(self.test_key): + os.remove(self.test_key) def _read_entries(self): with open(self.test_log, "r", encoding="utf-8") as handle: diff --git a/tests/test_r159_audit_pipeline.py b/tests/test_r159_audit_pipeline.py new file mode 100644 index 0000000..60f31cf --- /dev/null +++ b/tests/test_r159_audit_pipeline.py @@ -0,0 +1,165 @@ +import importlib.util +import json +import os +import sys +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +import services.audit as audit_module + + +def _load_verify_script_module(): + root = Path(__file__).resolve().parents[1] + module_path = root / "scripts" / "verify_audit_chain.py" + spec = importlib.util.spec_from_file_location("verify_audit_chain_script", module_path) + if spec is None or spec.loader is None: + raise RuntimeError("Failed to load verify_audit_chain.py") + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +class TestR159AuditPipeline(unittest.TestCase): + def setUp(self): + self.temp_dir = tempfile.TemporaryDirectory() + self.addCleanup(self.temp_dir.cleanup) + self.audit_log = os.path.join(self.temp_dir.name, "audit.log") + self.audit_key = f"{self.audit_log}.key" + self.env_patcher = patch.dict( + os.environ, + { + "OPENCLAW_AUDIT_CHAIN_KEY": "", + "MOLTBOT_AUDIT_CHAIN_KEY": "", + "OPENCLAW_AUDIT_CHAIN_KEY_PATH": "", + "MOLTBOT_AUDIT_CHAIN_KEY_PATH": "", + }, + clear=False, + ) + self.env_patcher.start() + self.addCleanup(self.env_patcher.stop) + self.path_patcher = patch("services.audit.AUDIT_LOG_PATH", self.audit_log) + self.hash_patcher = patch("services.audit._LAST_HASH", None) + self.chain_key_patcher = patch("services.audit._AUDIT_CHAIN_KEY", None) + self.path_patcher.start() + self.hash_patcher.start() + self.chain_key_patcher.start() + self.addCleanup(self.path_patcher.stop) + self.addCleanup(self.hash_patcher.stop) + self.addCleanup(self.chain_key_patcher.stop) + + def _read_chain_entries(self): + payloads = [] + for path in audit_module.verify_audit_chain().files_checked: + with open(path, "r", encoding="utf-8") as handle: + for line in handle: + if line.strip(): + payloads.append(json.loads(line)) + return payloads + + def test_restart_after_rotation_preserves_chain_continuity(self): + with patch("services.audit._audit_limits", return_value=(1, 3)): + audit_module.emit_audit_event( + action="config.update", + target="settings.json", + outcome="allow", + status_code=200, + details={"seq": 1}, + ) + audit_module.emit_audit_event( + action="config.update", + target="settings.json", + outcome="allow", + status_code=200, + details={"seq": 2}, + ) + audit_module._LAST_HASH = None + audit_module._AUDIT_CHAIN_KEY = None + audit_module.emit_audit_event( + action="config.update", + target="settings.json", + outcome="allow", + status_code=200, + details={"seq": 3}, + ) + + result = audit_module.verify_audit_chain() + self.assertTrue(result.ok, result.to_dict()) + self.assertEqual(result.entries_checked, 3) + self.assertEqual(len(result.files_checked), 3) + self.assertEqual(result.window_start_prev_hash, "GENESIS") + entries = self._read_chain_entries() + self.assertEqual(entries[1]["prev_hash"], entries[0]["entry_hash"]) + self.assertEqual(entries[2]["prev_hash"], entries[1]["entry_hash"]) + self.assertTrue(os.path.exists(self.audit_key)) + + def test_verify_detects_tampered_entry_hash(self): + audit_module.emit_audit_event( + action="config.update", + target="settings.json", + outcome="allow", + status_code=200, + details={"seq": 1}, + ) + with open(self.audit_log, "r", encoding="utf-8") as handle: + entry = json.loads(handle.readline()) + entry["action"] = "config.tampered" + with open(self.audit_log, "w", encoding="utf-8") as handle: + handle.write(json.dumps(entry, sort_keys=True, ensure_ascii=True) + "\n") + + result = audit_module.verify_audit_chain() + self.assertFalse(result.ok) + self.assertEqual(result.issues[0].code, "entry_hash_mismatch") + + def test_emit_surfaces_sink_failure_but_does_not_raise(self): + with patch( + "services.audit.LocalFileAuditSink.append_entry", + side_effect=OSError("disk full"), + ): + with self.assertLogs("ComfyUI-OpenClaw.services.audit", level="ERROR") as logs: + audit_module.emit_audit_event( + action="config.update", + target="settings.json", + outcome="allow", + status_code=200, + details={"seq": 1}, + ) + + output = "\n".join(logs.output) + self.assertIn("Failed to write audit entry", output) + self.assertFalse(os.path.exists(self.audit_log)) + + def test_verify_script_reports_pass_and_fail(self): + script = _load_verify_script_module() + audit_module.emit_audit_event( + action="config.update", + target="settings.json", + outcome="allow", + status_code=200, + details={"seq": 1}, + ) + + with patch.object( + sys, + "argv", + ["verify_audit_chain.py", "--path", self.audit_log, "--json"], + ): + self.assertEqual(script.main(), 0) + + with open(self.audit_log, "r", encoding="utf-8") as handle: + entry = json.loads(handle.readline()) + entry["entry_hash"] = "deadbeef" + with open(self.audit_log, "w", encoding="utf-8") as handle: + handle.write(json.dumps(entry, sort_keys=True, ensure_ascii=True) + "\n") + + with patch.object( + sys, + "argv", + ["verify_audit_chain.py", "--path", self.audit_log], + ): + self.assertEqual(script.main(), 1) + + +if __name__ == "__main__": + unittest.main()