From c75f13ef9d202379ed3569aa68efd04923811933 Mon Sep 17 00:00:00 2001 From: R_N_A <33271130+RNA4219@users.noreply.github.com> Date: Sat, 1 Nov 2025 11:05:55 +0900 Subject: [PATCH] Remove legacy audit utilities --- tools/audit/purge_logs.py | 60 ---------------------- tools/audit/verify_log_chain.py | 91 --------------------------------- 2 files changed, 151 deletions(-) delete mode 100644 tools/audit/purge_logs.py delete mode 100644 tools/audit/verify_log_chain.py diff --git a/tools/audit/purge_logs.py b/tools/audit/purge_logs.py deleted file mode 100644 index f0c1ff2..0000000 --- a/tools/audit/purge_logs.py +++ /dev/null @@ -1,60 +0,0 @@ -# SPDX-License-Identifier: Apache-2.0 -"""Purge audit logs that exceeded the configured retention period.""" - -from __future__ import annotations - -import argparse -from datetime import datetime, timezone -from pathlib import Path -from typing import Iterable, Sequence - - -def _resolve_targets(target: Path) -> Iterable[Path]: - if target.is_file(): - yield target - return - if not target.exists(): - return - for child in target.iterdir(): - if child.is_file(): - yield child - - -def purge_expired_logs(directory: Path, older_than_days: int, now: datetime | None = None) -> list[Path]: - if older_than_days <= 0: - raise ValueError("older_than_days must be a positive integer") - reference = now or datetime.now(timezone.utc) - cutoff_timestamp = reference.timestamp() - older_than_days * 24 * 60 * 60 - removed: list[Path] = [] - for candidate in _resolve_targets(directory): - if candidate.stat().st_mtime < cutoff_timestamp: - candidate.unlink() - removed.append(candidate) - removed.sort() - return removed - - -def build_parser() -> argparse.ArgumentParser: - parser = argparse.ArgumentParser(description="Purge audit log files older than the retention window.") - parser.add_argument("path", type=Path, help="Directory (or single file) containing audit logs.") - parser.add_argument("--older-than", type=int, required=True, help="Delete files strictly older than this many days.") - return parser - - -def main(argv: Sequence[str] | None = None) -> int: - parser = build_parser() - args = parser.parse_args(argv) - removed: list[Path] = [] - try: - removed = purge_expired_logs(args.path, args.older_than) - except ValueError as exc: - parser.error(str(exc)) - if not removed: - return 0 - for path in removed: - print(path) - return 0 - - -if __name__ == "__main__": # pragma: no cover - CLI entry point - raise SystemExit(main()) diff --git a/tools/audit/verify_log_chain.py b/tools/audit/verify_log_chain.py deleted file mode 100644 index 0ae995b..0000000 --- a/tools/audit/verify_log_chain.py +++ /dev/null @@ -1,91 +0,0 @@ -# SPDX-License-Identifier: Apache-2.0 -"""Verify audit log chains generated with HMAC signatures.""" - -from __future__ import annotations - -import argparse -import hmac -import json -import sys -from dataclasses import dataclass -from hashlib import sha256 -from pathlib import Path -from typing import Iterable, Mapping, Sequence - - -class VerificationFailure(RuntimeError): - """Raised when audit log chain verification fails.""" - - -@dataclass(frozen=True) -class LogEntry: - payload: str - previous_signature: str - signature: str - - -def parse_log_entries(log_file: Path) -> Iterable[LogEntry]: - with log_file.open("r", encoding="utf-8") as handle: - for index, raw_line in enumerate(handle, start=1): - line = raw_line.strip() - if not line: - continue - try: - payload = json.loads(line) - except json.JSONDecodeError as exc: # pragma: no cover - validated in tests via failure path - raise VerificationFailure(f"Line {index}: invalid JSON: {exc}") from exc - if not isinstance(payload, Mapping): - raise VerificationFailure(f"Line {index}: expected object, received {type(payload).__name__}") - try: - yield LogEntry( - payload=str(payload["payload"]), - previous_signature=str(payload["previous_signature"]), - signature=str(payload["signature"]), - ) - except KeyError as exc: - raise VerificationFailure(f"Line {index}: missing field {exc.args[0]}") from exc - - -def compute_expected_signature(secret: bytes, previous_signature: str, payload: str) -> str: - return hmac.new(secret, f"{previous_signature}:{payload}".encode("utf-8"), sha256).hexdigest() - - -def verify_log_entries(log_file: Path, secret: str, initial_signature: str = "") -> None: - secret_bytes = secret.encode("utf-8") - expected_previous = initial_signature - for entry in parse_log_entries(log_file): - if entry.previous_signature != expected_previous: - raise VerificationFailure( - "previous signature mismatch: expected " - f"{expected_previous or ''}, received {entry.previous_signature}" - ) - expected_signature = compute_expected_signature(secret_bytes, entry.previous_signature, entry.payload) - if entry.signature != expected_signature: - raise VerificationFailure( - "signature mismatch: expected " - f"{expected_signature}, received {entry.signature}" - ) - expected_previous = entry.signature - - -def build_parser() -> argparse.ArgumentParser: - parser = argparse.ArgumentParser(description="Verify JSONL audit log chains with HMAC signatures.") - parser.add_argument("log_file", type=Path, help="Path to the JSONL audit log.") - parser.add_argument("--secret", required=True, help="Shared secret used for HMAC verification.") - parser.add_argument("--initial-signature", default="", help="Initial signature seed for the first entry.") - return parser - - -def main(argv: Sequence[str] | None = None) -> int: - parser = build_parser() - args = parser.parse_args(argv) - try: - verify_log_entries(args.log_file, args.secret, args.initial_signature) - except VerificationFailure as exc: - print(f"Verification failed: {exc}", file=sys.stderr) - return 1 - return 0 - - -if __name__ == "__main__": # pragma: no cover - CLI entry point - raise SystemExit(main())