audit.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oct/mcp/audit.py
 
"""
Purpose
-------
JSONL audit logger for the ``oct-mcp`` server. Writes one JSON object per
line to ``~/.oct-mcp/audit.log`` with size-based rotation (5 MB per file)
and count-based pruning (keep the 20 newest files).
 
Responsibilities
----------------
- Define :class:`McpAuditRecord` - the on-disk schema (blueprint §5.7).
- Implement :class:`McpAuditLogger` with the same rotation/pruning logic
  as :class:`oct.git.audit.AuditLogger`, adapted for the global
  ``~/.oct-mcp/`` path and the richer MCP schema.
- Never log raw ``args`` when ``args_redacted`` is True - callers pass
  the redacted dict; logger serialises whatever it receives.
- Emit diagnostic events via ``_dbg("MCP", ..., level)``.
 
Diagnostics
-----------
Domain: MCP
Levels:
    L1 - errors: IO failures, serialisation failures
    L2 - lifecycle: file opened/rotated, files pruned
    L3 - (none): lifecycle is reported at L2, per-write detail at L4
    L4 - deep trace: every write() call
 
Contracts
---------
- This module does not import ``click``, the MCP SDK, or any other
  ``oct.mcp.*`` module (no cycles).
- :meth:`McpAuditLogger.write` must never raise. IO failures are logged
  at level 1 and swallowed - audit logging is observability, not
  correctness.
- Filenames follow ``mcp-audit-YYYYMMDD-HHMMSS.jsonl`` strictly so
  lexical ordering matches chronological ordering for pruning.
- Every :class:`McpAuditRecord` field is JSON-serialisable.
 
Dependencies
------------
- oct.core.diagnostics (_dbg structured logger)
"""
 
from __future__ import annotations
 
import hashlib
import json
from dataclasses import asdict, dataclass, field
from datetime import datetime, timezone
from pathlib import Path
 
from oct.core.diagnostics import _dbg
 
 
# ---------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------
 
#: Maximum number of ``mcp-audit-*.jsonl`` files kept on disk.
MCP_AUDIT_MAX_FILES: int = 20
 
#: Rotation threshold per audit file.
MCP_AUDIT_MAX_SIZE_BYTES: int = 5 * 1024 * 1024  # 5 MB
 
#: ``strftime`` format embedded in each rotated filename.
_AUDIT_TS_FORMAT: str = "%Y%m%d-%H%M%S"
 
#: Filename prefix / suffix / glob for MCP audit files.
_AUDIT_FILENAME_PREFIX: str = "mcp-audit-"
_AUDIT_FILENAME_SUFFIX: str = ".jsonl"
_AUDIT_GLOB: str = f"{_AUDIT_FILENAME_PREFIX}*{_AUDIT_FILENAME_SUFFIX}"
 
#: Default directory for user-level MCP audit logs.
_DEFAULT_AUDIT_DIR: Path = Path.home() / ".oct-mcp"
 
#: Default audit log filename (no rotation suffix - rotated files carry
#: their own timestamp suffix).
_DEFAULT_AUDIT_FILENAME: str = "audit.log"
 
 
# ---------------------------------------------------------------------
# Data model (blueprint §5.7)
# ---------------------------------------------------------------------
 
 
@dataclass
class McpAuditRecord:
    """Schema for one MCP audit row. Serialised via :func:`dataclasses.asdict`.
 
    All fields are JSON-primitive or containers of JSON-primitives.
    ``args`` should already be credential-redacted by the caller before
    being assigned; when ``args_redacted`` is ``True`` the logger does
    not apply further sanitisation (it trusts the caller). When
    ``args_redacted`` is ``False``, args contain the raw validated dict -
    callers MUST set this consciously.
 
    ``project_root_hash`` is the first 8 hex characters of
    ``sha256(str(project_root))``, giving per-project correlation
    without leaking the absolute path.
    """
 
    timestamp: str = ""
    """ISO 8601 UTC timestamp of the request."""
 
    session_id: str = ""
    """Opaque identifier for the MCP session (set at server startup)."""
 
    request_id: str = ""
    """Unique identifier for this tool invocation."""
 
    tool: str = ""
    """MCP tool name, e.g. ``"oct_lint"``."""
 
    args: dict = field(default_factory=dict)
    """Validated (and optionally redacted) args dict."""
 
    args_redacted: bool = True
    """True → ``args`` has had sensitive values replaced."""
 
    decision: str = ""
    """Policy decision: ``"allowed"``, ``"denied"``, or ``"requires_confirmation"``."""
 
    policy_rule: str = ""
    """Name of the policy rule that determined the decision."""
 
    exit_code: int | None = None
    """Exit code from the sandboxed subprocess (``None`` if not executed)."""
 
    duration_ms: int = 0
    """Wall-clock time from request receipt to response dispatch, in ms."""
 
    output_size_bytes: int = 0
    """Byte length of the (post-redaction) tool output."""
 
    output_truncated: bool = False
    """True if the output was truncated to ``max_output_bytes``."""
 
    output_hash_sha256: str = ""
    """SHA-256 hex digest of the raw (pre-redaction) output for integrity."""
 
    redactions_applied: int = 0
    """Count of redaction substitutions applied by the Output Redactor."""
 
    user: str = ""
    """OS username of the process running the MCP server."""
 
    project_root_hash: str = ""
    """First 8 hex chars of sha256(str(project_root)) - correlation, not path."""
 
 
def _utc_now_iso() -> str:
    """Return the current UTC time as an ISO 8601 string ending in ``Z``."""
    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
 
 
def _hash_project_root(project_root: Path) -> str:
    """Return the first 8 hex chars of sha256(str(project_root))."""
    digest = hashlib.sha256(str(project_root).encode()).hexdigest()
    return digest[:8]
 
 
# ---------------------------------------------------------------------
# Rotating JSONL logger
# ---------------------------------------------------------------------
 
 
class McpAuditLogger:
    """Append-only rotating JSONL logger for ``oct-mcp`` tool calls.
 
    One instance is created per MCP server session (in ``server.py``).
    The constructor does not open a file - the first :meth:`write` call
    lazily creates the ``~/.oct-mcp/`` directory and the first audit
    file. This keeps the constructor side-effect-free so the server can
    instantiate it during startup without touching the filesystem.
 
    Storage location
    ----------------
    The audit log lives at ``audit_log_path`` (from :class:`McpConfig`,
    default ``~/.oct-mcp/audit.log``). Rotated files are placed in the
    same parent directory with a ``mcp-audit-YYYYMMDD-HHMMSS.jsonl``
    naming pattern. The *initial* write goes to the ``audit_log_path``
    itself; once rotation is triggered, subsequent writes go to
    timestamped files. Pruning removes the oldest timestamped files
    when the count exceeds ``max_files``.
 
    Rotation strategy
    -----------------
    - **Size rotation:** before each write, if the current file is at or
      above ``max_size_bytes``, close it and open a new timestamped file.
    - **Count pruning:** after rotation, if the count of
      ``mcp-audit-*.jsonl`` files in the directory exceeds ``max_files``,
      delete the oldest until the count is within the limit.
 
    Error handling
    --------------
    Every IO path in :meth:`write` is wrapped in a try/except. Failures
    are logged via ``_dbg`` at level 1 and the method returns silently.
    The audit log must never block or crash tool execution.
    """
 
    def __init__(
        self,
        audit_log_path: str | Path | None = None,
        max_files: int = MCP_AUDIT_MAX_FILES,
        max_size_bytes: int = MCP_AUDIT_MAX_SIZE_BYTES,
    ) -> None:
        func = "McpAuditLogger.__init__"
        raw_path = Path(audit_log_path) if audit_log_path else (
            _DEFAULT_AUDIT_DIR / _DEFAULT_AUDIT_FILENAME
        )
        # Expand ~ so path works correctly across platforms.
        self._base_path: Path = raw_path.expanduser()
        self._audit_dir: Path = self._base_path.parent
        self._current_path: Path | None = None
        self._max_files: int = max(1, max_files)
        self._max_size_bytes: int = max(1, max_size_bytes)
        # Monotonic counter ensures new filenames are never reused within
        # a session, even after pruning removes old files with low counters.
        self._rotation_counter: int = 0
        _dbg("MCP", func, f"audit_dir={self._audit_dir}", 4)
 
    # -- internal helpers ------------------------------------------------
 
    def _ensure_audit_dir(self) -> None:
        """Create ``~/.oct-mcp/`` on demand. Idempotent."""
        if not self._audit_dir.exists():
            self._audit_dir.mkdir(parents=True, exist_ok=True)
 
    def _new_filename(self) -> Path:
        """Return a fresh ``mcp-audit-YYYYMMDD-HHMMSS-NNN.jsonl`` path.
 
        Uses a monotonically increasing instance-level counter (``_rotation_counter``)
        so that names are never reused within a session, even after old files
        are pruned. Zero-padded to 3 digits so lexical sort order is identical
        to chronological creation order within a session.
        """
        ts = datetime.now(timezone.utc).strftime(_AUDIT_TS_FORMAT)
        self._rotation_counter += 1
        candidate = (
            self._audit_dir
            / f"{_AUDIT_FILENAME_PREFIX}{ts}-{self._rotation_counter:03d}{_AUDIT_FILENAME_SUFFIX}"
        )
        # Collision guard: if a file with this name already exists (e.g.
        # from a previous process run in the same second), keep incrementing.
        while candidate.exists():
            self._rotation_counter += 1
            candidate = (
                self._audit_dir
                / f"{_AUDIT_FILENAME_PREFIX}{ts}-{self._rotation_counter:03d}{_AUDIT_FILENAME_SUFFIX}"
            )
        return candidate
 
    def _current_file_size(self) -> int:
        """Return the current audit file's on-disk size, 0 if unavailable."""
        if self._current_path is None:
            return 0
        try:
            return self._current_path.stat().st_size
        except OSError:
            return 0
 
    def _rotate_if_needed(self) -> None:
        """Open a new file if the current one is missing or full.
 
        On first call: uses the canonical ``_base_path`` (audit.log) so
        the primary log is always predictably named. On subsequent
        rotations: switches to timestamped ``mcp-audit-*.jsonl`` files.
        """
        func = "McpAuditLogger._rotate_if_needed"
        if self._current_path is None:
            # First write: use the base path (audit.log).
            self._current_path = self._base_path
            if not self._current_path.exists():
                self._current_path.touch()
            _dbg("MCP", func, f"opened initial file={self._current_path.name}", 2)
            return
        if self._current_file_size() >= self._max_size_bytes:
            _dbg("MCP", func, f"size rotation from={self._current_path.name}", 2)
            self._current_path = self._new_filename()
            self._current_path.touch()  # create on disk so prune counts it
            self._prune_old_files()
 
    def _list_rotated_files(self) -> list[Path]:
        """Return all ``mcp-audit-*.jsonl`` files sorted lexically (= chronologically)."""
        if not self._audit_dir.exists():
            return []
        return sorted(self._audit_dir.glob(_AUDIT_GLOB))
 
    def _prune_old_files(self) -> None:
        """Delete the oldest rotated files until count <= ``max_files``.
 
        The current file (just created by :meth:`_rotate_if_needed`) is
        excluded from the candidate list so it is never accidentally pruned
        immediately after creation. ``max_files`` counts the current file,
        so we keep at most ``max_files - 1`` older files.
        """
        func = "McpAuditLogger._prune_old_files"
        files = self._list_rotated_files()
        # Exclude the current file - it was just created and must not be pruned.
        candidates = [f for f in files if f != self._current_path]
        excess = len(candidates) - (self._max_files - 1)
        if excess <= 0:
            return
        for path in candidates[:excess]:
            try:
                path.unlink()
                _dbg("MCP", func, f"pruned {path.name}", 2)
            except OSError as exc:  # pragma: no cover - best-effort
                _dbg("MCP", func, f"prune failed {path.name}: {exc}", 1)
 
    # -- public API ------------------------------------------------------
 
    def write(self, record: McpAuditRecord) -> None:
        """Serialise *record* and append it as one JSONL line.
 
        Never raises. Any failure is caught, logged at level 1, and
        swallowed so callers are never blocked by audit misbehaviour.
        """
        func = "McpAuditLogger.write"
        try:
            self._ensure_audit_dir()
            self._rotate_if_needed()
            assert self._current_path is not None
            if not record.timestamp:
                record.timestamp = _utc_now_iso()
            line = json.dumps(asdict(record), ensure_ascii=False)
            with self._current_path.open("a", encoding="utf-8") as fh:
                fh.write(line)
                fh.write("\n")
            _dbg(
                "MCP", func,
                f"wrote record tool={record.tool} decision={record.decision} "
                f"exit_code={record.exit_code}",
                4,
            )
        except OSError as exc:
            _dbg("MCP", func, f"audit write failed: {exc}", 1)
        except (TypeError, ValueError) as exc:
            _dbg("MCP", func, f"audit serialize failed: {exc}", 1)
 
    @property
    def current_path(self) -> Path | None:
        """Current audit file path (``None`` before first write)."""
        return self._current_path
 
    @property
    def audit_dir(self) -> Path:
        """Directory where audit files are stored."""
        return self._audit_dir