audit.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oct/git/audit.py
 
"""
Purpose
-------
Audit JSONL logger and AI-Trace decorator for Phase 4B ``oct git``
commands. Writes one JSON object per line to
``<project>/logs/git-audit-YYYYMMDD-HHMMSS.jsonl`` with size + count
rotation, and provides the :func:`audited` decorator that every
``oct git`` Click command is wrapped in so that entry/exit trace events
and a persistent audit row are recorded even if the command crashes.
 
Responsibilities
----------------
- Define :class:`AuditRecord` — the exact on-disk schema (blueprint
  §12.2, verbatim).
- Implement :class:`AuditLogger` with size-based rotation (5 MB) and
  count-based pruning (keep the 20 newest files).
- Redact credentials from any string value written to the log by
  deferring to :func:`oct.core.git._redact_git_url`.
- Emit AI-Trace shadow events via ``_dbg("GIT", msg, 4)`` — file-only,
  off by default; see blueprint §12.3 and Phase 4B design decision D7.
 
Diagnostics
-----------
Domain: GIT
Levels:
    L1 — errors: rotation IO failures, directory-create failures
    L2 — lifecycle: file rotated, files pruned
    L4 — deep trace: every ``write()`` call and every command start/end
 
Contracts
---------
- This module does not import ``click``, does not shell out, and does
  not touch any git state. It only reads config-less arguments passed
  in from the caller.
- :meth:`AuditLogger.write` must never raise. On any IO failure it
  logs an error via ``_dbg`` and returns silently — the audit log is
  *observability*, not a correctness path.
- Filenames follow ``git-audit-YYYYMMDD-HHMMSS.jsonl`` strictly so that
  lexical ordering matches chronological ordering for pruning.
- Every ``AuditRecord`` field is JSON-serialisable. ``args`` is always
  passed through credential redaction before serialisation.
"""
 
from __future__ import annotations
 
import functools
import json
import time
from dataclasses import asdict, dataclass, field
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Callable
 
from oct.core.diagnostics import _dbg
from oct.core.git import _redact_git_url
from oct.core.option_c_dir import resolve_logs_dir
 
 
# ---------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------
 
#: Maximum number of ``git-audit-*.jsonl`` files kept on disk. When the
#: 21st file is created by a rotation event, the oldest is deleted.
AUDIT_MAX_FILES: int = 20
 
#: Rotation threshold. When the current audit file reaches this size in
#: bytes, the next :meth:`AuditLogger.write` call closes it and opens a
#: new timestamped file.
AUDIT_MAX_SIZE_BYTES: int = 5 * 1024 * 1024  # 5 MB
 
#: ``strftime`` format for the timestamp embedded in each rotated
#: filename. Lexical sort order is identical to chronological order.
_AUDIT_TS_FORMAT: str = "%Y%m%d-%H%M%S"
 
#: Filename prefix and suffix used for glob matching during pruning.
_AUDIT_FILENAME_PREFIX: str = "git-audit-"
_AUDIT_FILENAME_SUFFIX: str = ".jsonl"
_AUDIT_GLOB: str = f"{_AUDIT_FILENAME_PREFIX}*{_AUDIT_FILENAME_SUFFIX}"
 
#: Directory (relative to project root) where audit logs are written —
#: legacy fallback only. FS-539 resolves the actual path dynamically via
#: :func:`oct.core.option_c_dir.resolve_logs_dir`, which prefers
#: ``.option_c/logs/`` on migrated projects and falls back to this
#: subdir on legacy ones. Kept as a module constant for any external
#: test that imports it.
_AUDIT_SUBDIR: str = "logs"
 
 
# ---------------------------------------------------------------------
# Data model (blueprint §12.2)
# ---------------------------------------------------------------------
 
 
@dataclass
class AuditRecord:
    """Schema for one audit row. Serialised via :func:`dataclasses.asdict`.
 
    Every field is either JSON-primitive or a list/dict of JSON-primitives
    so that ``json.dumps(asdict(record))`` is guaranteed to succeed.
    ``args`` is credential-redacted by the caller before being assigned
    (see :func:`_redact_args`).
 
    Fields whose value is unknown at record-construction time may be
    left at their default; missing fields are emitted as ``null`` in
    the JSON output.
    """
 
    timestamp: str = ""                       # ISO 8601 UTC
    command: str = ""                         # e.g. "oct git check"
    args: list[str] = field(default_factory=list)
    branch: str | None = None
    staged_files: int = 0
    checks_passed: bool = False
    secrets_detected: bool = False
    lint_violations: int = 0
    format_violations: int = 0
    exit_code: int | None = None
    duration_ms: int = 0
    #: F1.3: per-subproject quality-gate results when oct git commit
    #: ran a workspace fan-out. Empty list when fan-out was not used.
    #: Each entry: {name, profile, files, lint_violations,
    #: format_violations, secrets}.
    per_subproject_results: list[dict] = field(default_factory=list)
 
 
def _utc_now_iso() -> str:
    """Return the current UTC time as an ISO 8601 string.
 
    Uses ``datetime.now(timezone.utc)`` rather than ``datetime.utcnow``
    (deprecated in 3.12+) and appends a literal ``Z`` so readers can
    round-trip the value without timezone ambiguity.
    """
    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
 
 
def _redact_args(argv: list[str] | tuple[str, ...] | None) -> list[str]:
    """Run each argv element through :func:`_redact_git_url`.
 
    This is defensive: an ``oct git`` command rarely receives a remote
    URL as a positional arg, but a user *could* pass one via future
    subcommands (e.g. ``oct git init --remote``). Applying the redactor
    unconditionally costs a regex match per arg and closes the hole.
    """
    if not argv:
        return []
    out: list[str] = []
    for item in argv:
        if isinstance(item, str):
            out.append(_redact_git_url(item))
        else:
            out.append(str(item))
    return out
 
 
# ---------------------------------------------------------------------
# Rotating JSONL logger
# ---------------------------------------------------------------------
 
 
class AuditLogger:
    """Append-only rotating JSONL logger for ``oct git`` commands.
 
    One instance corresponds to one project root. The constructor does
    not open a file — the first :meth:`write` call lazily materialises
    the ``logs/`` directory and the current audit file. This lets a
    command that never reaches the write stage (e.g. a CLI parse error)
    leave no side-effects on disk.
 
    Rotation strategy
    -----------------
    - **Size rotation:** before each write, if the current file is at or
      above :data:`AUDIT_MAX_SIZE_BYTES`, close it and open a new
      timestamped file. The rotating write itself goes into the new
      file, not the old one.
    - **Count pruning:** after any rotation, if the number of
      ``git-audit-*.jsonl`` files in the logs directory exceeds
      :data:`AUDIT_MAX_FILES`, delete the oldest (lexical order) until
      the count is within the limit.
 
    Error handling
    --------------
    Any IO error during :meth:`write` — creating the directory, opening
    the file, serialising the record, writing the line, or rotating —
    is caught and logged via ``_dbg("GIT", ..., 1)``. The method never
    raises; audit logging is best-effort observability and must not
    mask command failures.
    """
 
    def __init__(self, project_root: Path) -> None:
        func = "AuditLogger.__init__"
        self._project_root: Path = Path(project_root)
        # FS-539: .option_c/logs/ wins when present, else root/logs/ (legacy).
        # On a fresh project the resolver returns the modern path so the
        # first write lands under .option_c/.
        self._logs_dir: Path = resolve_logs_dir(self._project_root)
        self._current_path: Path | None = None
        _dbg("GIT", func, f"project_root={self._project_root}", 4)
 
    # -- internal helpers ------------------------------------------------
 
    def _ensure_logs_dir(self) -> None:
        """Create ``<project>/logs/`` on demand. Idempotent."""
        if not self._logs_dir.exists():
            self._logs_dir.mkdir(parents=True, exist_ok=True)
 
    def _new_filename(self) -> Path:
        """Return a fresh ``git-audit-YYYYMMDD-HHMMSS.jsonl`` path.
 
        Uses the current UTC timestamp. If the resulting filename
        already exists (unlikely but possible at sub-second rotation or
        in a fast loop), disambiguate by appending a monotonic suffix
        ``-<counter>`` so no two rotations collide.
        """
        ts = datetime.now(timezone.utc).strftime(_AUDIT_TS_FORMAT)
        base = f"{_AUDIT_FILENAME_PREFIX}{ts}{_AUDIT_FILENAME_SUFFIX}"
        candidate = self._logs_dir / base
        counter = 1
        while candidate.exists():
            candidate = (
                self._logs_dir
                / f"{_AUDIT_FILENAME_PREFIX}{ts}-{counter}{_AUDIT_FILENAME_SUFFIX}"
            )
            counter += 1
        return candidate
 
    def _current_file_size(self) -> int:
        """Return the current audit file's size in bytes, 0 if missing."""
        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 fresh file if the current one is missing or full.
 
        Called at the top of :meth:`write`. After opening a new file,
        delegates to :meth:`_prune_old_files` so count-based rotation
        runs at the same moment size-based rotation does (cheapest
        common path).
        """
        func = "AuditLogger._rotate_if_needed"
        if self._current_path is None:
            self._current_path = self._new_filename()
            self._current_path.touch()  # create on disk so prune counts it
            _dbg("GIT", func, f"opened initial file={self._current_path.name}", 2)
            self._prune_old_files()
            return
        if self._current_file_size() >= AUDIT_MAX_SIZE_BYTES:
            _dbg("GIT", 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_existing_files(self) -> list[Path]:
        """Return all ``git-audit-*.jsonl`` files in sorted order.
 
        Sort is lexical, which matches chronological thanks to the
        fixed-width ``YYYYMMDD-HHMMSS`` timestamp format.
        """
        if not self._logs_dir.exists():
            return []
        return sorted(self._logs_dir.glob(_AUDIT_GLOB))
 
    def _prune_old_files(self) -> None:
        """Delete the oldest files until count <= :data:`AUDIT_MAX_FILES`.
 
        Deletion is best-effort: an ``OSError`` from a concurrent
        reader holding a handle on Windows is logged but not re-raised.
        """
        func = "AuditLogger._prune_old_files"
        files = self._list_existing_files()
        excess = len(files) - AUDIT_MAX_FILES
        if excess <= 0:
            return
        victims = files[:excess]
        for path in victims:
            try:
                path.unlink()
                _dbg("GIT", func, f"pruned {path.name}", 2)
            except OSError as exc:  # pragma: no cover - best-effort
                _dbg("GIT", func, f"prune failed {path.name}: {exc}", 1)
 
    # -- public API ------------------------------------------------------
 
    def write(self, record: AuditRecord) -> None:
        """Serialise ``record`` and append it as one JSONL line.
 
        Never raises. On IO failure, logs an error via ``_dbg`` at
        level 1 and returns. The caller is never blocked by audit log
        misbehaviour.
        """
        func = "AuditLogger.write"
        try:
            self._ensure_logs_dir()
            self._rotate_if_needed()
            assert self._current_path is not None  # set by _rotate_if_needed
            # Defensive redaction at the boundary. Even though callers
            # should already have redacted, applying again is idempotent.
            record.args = _redact_args(record.args)
            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("GIT", func, f"wrote record exit_code={record.exit_code}", 4)
        except OSError as exc:
            _dbg("GIT", func, f"audit write failed: {exc}", 1)
        except (TypeError, ValueError) as exc:
            _dbg("GIT", func, f"audit serialize failed: {exc}", 1)
 
    @property
    def current_path(self) -> Path | None:
        """Current audit filename (``None`` before first write)."""
        return self._current_path
 
 
# ---------------------------------------------------------------------
# @audited decorator (AI-Trace + finally-write audit row)
# ---------------------------------------------------------------------
 
 
def audited(cmd_name: str) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
    """Decorator that wraps an ``oct git`` Click command.
 
    Responsibilities:
 
    1. Emit ``_dbg("GIT", "command_start ...", 4)`` before the command.
    2. Run the wrapped function.
    3. In a ``finally`` block, emit ``_dbg("GIT", "command_end ...", 4)``
       and write an :class:`AuditRecord` into the project's audit log.
 
    The wrapped command is responsible for populating
    ``ctx.obj["_audit_record"]`` during its run — the decorator reads
    that slot at the end and writes whatever is there, falling back to
    a minimal record (with ``exit_code=None`` and ``checks_passed=False``)
    if the command crashed before populating it. This guarantees every
    command invocation leaves exactly one audit row.
 
    Parameters
    ----------
    cmd_name
        Human-readable command identifier for the trace and audit row,
        e.g. ``"oct git check"``.
    """
 
    def decorator(fn: Callable[..., Any]) -> Callable[..., Any]:
        @functools.wraps(fn)
        def wrapper(ctx, *args, **kwargs):
            import click  # lazy import so this module stays Click-free at top level
 
            func = f"audited::{cmd_name}"
            start = time.monotonic()
            _dbg("GIT", func, f"command_start cmd={cmd_name}", 4)
 
            # Resolve project root from ctx; fall back to cwd.
            project_root = _resolve_project_root_from_ctx(ctx)
 
            # Make sure ctx.obj exists so the command can stash its record.
            if not isinstance(getattr(ctx, "obj", None), dict):
                ctx.obj = {}
            ctx.obj.setdefault("_audit_record", AuditRecord(
                timestamp=_utc_now_iso(),
                command=cmd_name,
            ))
 
            logger = AuditLogger(project_root)
            try:
                return fn(ctx, *args, **kwargs)
            finally:
                duration_ms = int((time.monotonic() - start) * 1000)
                record: AuditRecord = ctx.obj.get("_audit_record") or AuditRecord(
                    timestamp=_utc_now_iso(),
                    command=cmd_name,
                )
                if not record.timestamp:
                    record.timestamp = _utc_now_iso()
                if not record.command:
                    record.command = cmd_name
                record.duration_ms = duration_ms
                # F1.3: pull per-subproject results from ctx.obj if the
                # command stashed them there. The command code populates
                # ctx.obj["_per_subproject_results"]; the audit record
                # owns the canonical list.
                psr = ctx.obj.get("_per_subproject_results")
                if isinstance(psr, list):
                    record.per_subproject_results = psr
                _dbg(
                    "GIT", func,
                    f"command_end cmd={cmd_name} exit_code={record.exit_code} "
                    f"duration_ms={duration_ms}",
                    4,
                )
                logger.write(record)
 
        return wrapper
 
    return decorator
 
 
def _resolve_project_root_from_ctx(ctx: Any) -> Path:
    """Best-effort extraction of the project root from a Click context.
 
    Tries, in order:
 
    1. ``ctx.obj["project_root"]`` — the canonical slot set by the
       top-level ``cli`` group (see [oct/oct/cli.py](../cli.py)).
    2. ``ctx.obj["root"]`` — older convention used by a couple of
       legacy commands.
    3. :func:`oct.core.project_root.get_project_root` — the formal
       resolver, which consults ``--root-dir`` and walks upward.
    4. ``Path.cwd()`` as the absolute last resort.
 
    Any exception is swallowed and falls through to the next step so
    the decorator never blocks command execution on a resolution error.
    """
    obj = getattr(ctx, "obj", None)
    if isinstance(obj, dict):
        for key in ("project_root", "root"):
            candidate = obj.get(key)
            if candidate:
                try:
                    return Path(candidate)
                except (TypeError, ValueError):
                    pass
    try:
        from oct.core.project_root import get_project_root
        return Path(get_project_root(ctx))
    except Exception:  # pragma: no cover - fall through to cwd
        return Path.cwd()