audit

audit

Audit logger — persist security events to SQLite with Merkle hash chain.

Classes

AuditLogger

AuditLogger(
    db_path: Union[str, Path] = DEFAULT_CONFIG_DIR
    / "audit.db",
    bus: Optional[EventBus] = None,
)

Append-only SQLite audit log for security events.

PARAMETER	DESCRIPTION
db_path	Path to the SQLite database file. TYPE: Union[str, Path] DEFAULT: DEFAULT_CONFIG_DIR / 'audit.db'
bus	Optional event bus — if provided, subscribes to security events (SECURITY_SCAN, SECURITY_ALERT, SECURITY_BLOCK). TYPE: Optional[EventBus] DEFAULT: None

Source code in src/diapason/security/audit.py

def __init__(
    self,
    db_path: Union[str, Path] = DEFAULT_CONFIG_DIR / "audit.db",
    bus: Optional[EventBus] = None,
) -> None:
    # `Path()` accepte tout objet exposant `__fspath__` — un `MagicMock`
    # en fait partie, et le sien rend « MagicMock/<nom>/<id> ». Un test
    # qui patchait `load_config` sans configurer `security.audit_log_path`
    # faisait donc créer, en silence, un vrai répertoire et une vraie base
    # SQLite à la racine du dépôt : 42 fichiers y ont dormi jusqu'au
    # 25 août 2026. La signature promettait `str | Path` ; elle le vérifie
    # désormais, et l'échec est bruyant plutôt qu'écrit sur le disque.
    if not isinstance(db_path, (str, Path)):
        raise TypeError(
            f"db_path doit être un str ou un Path, pas un {type(db_path).__name__}."
        )
    self._db_path = Path(db_path)
    from diapason.security.file_utils import secure_create

    secure_create(self._db_path)
    self._conn = sqlite3.connect(str(self._db_path), check_same_thread=False)
    self._lock = threading.RLock()
    self._conn.execute(
        """
        CREATE TABLE IF NOT EXISTS security_events (
            id          INTEGER PRIMARY KEY,
            timestamp   REAL,
            event_type  TEXT,
            findings_json TEXT,
            content_preview TEXT,
            action_taken TEXT,
            row_hash    TEXT DEFAULT '',
            prev_hash   TEXT DEFAULT ''
        )
        """
    )
    self._conn.commit()
    self._migrate_schema()

    if bus is not None:
        bus.subscribe(EventType.SECURITY_SCAN, self._on_event)
        bus.subscribe(EventType.SECURITY_ALERT, self._on_event)
        bus.subscribe(EventType.SECURITY_BLOCK, self._on_event)

Methods:

log

log(event: SecurityEvent) -> None

Insert a security event into the audit log with Merkle hash chain.

Source code in src/diapason/security/audit.py

def log(self, event: SecurityEvent) -> None:
    """Insert a security event into the audit log with Merkle hash chain."""
    findings_json = json.dumps(
        [
            {
                "pattern_name": f.pattern_name,
                # Audit metadata must never become a second secret store.
                "matched_text": "",
                "matched_sha256": hashlib.sha256(
                    f.matched_text.encode()
                ).hexdigest()
                if f.matched_text
                else "",
                "threat_level": f.threat_level.value,
                "start": f.start,
                "end": f.end,
                "description": f.description,
            }
            for f in event.findings
        ]
    )
    content_marker = ""
    if event.content_preview:
        content_marker = (
            "sha256:"
            f"{hashlib.sha256(event.content_preview.encode()).hexdigest()}"
            f";len:{len(event.content_preview)}"
        )

    with self._lock:
        # Compute and insert the next link atomically across worker threads.
        prev_hash = self.tail_hash()
        hash_input = (
            f"{prev_hash}|{event.timestamp}|{event.event_type.value}"
            f"|{findings_json}|{content_marker}|{event.action_taken}"
        )
        row_hash = hashlib.sha256(hash_input.encode()).hexdigest()

        self._conn.execute(
            """
            INSERT INTO security_events
                (timestamp, event_type, findings_json, content_preview,
                 action_taken, row_hash, prev_hash)
            VALUES (?, ?, ?, ?, ?, ?, ?)
            """,
            (
                event.timestamp,
                event.event_type.value,
                findings_json,
                content_marker,
                event.action_taken,
                row_hash,
                prev_hash,
            ),
        )
        self._conn.commit()

query

query(
    *,
    event_type: Optional[str] = None,
    since: Optional[float] = None,
    limit: int = 100,
) -> List[SecurityEvent]

Query logged security events with optional filters.

Source code in src/diapason/security/audit.py

def query(
    self,
    *,
    event_type: Optional[str] = None,
    since: Optional[float] = None,
    limit: int = 100,
) -> List[SecurityEvent]:
    """Query logged security events with optional filters."""
    sql = (
        "SELECT timestamp, event_type, findings_json,"
        " content_preview, action_taken"
        " FROM security_events WHERE 1=1"
    )
    params: list = []

    if event_type is not None:
        sql += " AND event_type = ?"
        params.append(event_type)
    if since is not None:
        sql += " AND timestamp >= ?"
        params.append(since)

    sql += " ORDER BY timestamp DESC LIMIT ?"
    params.append(limit)

    with self._lock:
        rows = self._conn.execute(sql, params).fetchall()
    events: List[SecurityEvent] = []
    for row in rows:
        ts, etype, findings_json, preview, action = row
        findings_raw = json.loads(findings_json) if findings_json else []
        findings = [
            ScanFinding(
                pattern_name=f["pattern_name"],
                matched_text=f["matched_text"],
                threat_level=ThreatLevel(f["threat_level"]),
                start=f["start"],
                end=f["end"],
                description=f.get("description", ""),
            )
            for f in findings_raw
        ]
        events.append(
            SecurityEvent(
                event_type=SecurityEventType(etype),
                timestamp=ts,
                findings=findings,
                content_preview=preview or "",
                action_taken=action or "",
            )
        )
    return events

tail_hash

tail_hash() -> str

Return the hash of the last row in the chain, or empty string.

Source code in src/diapason/security/audit.py

def tail_hash(self) -> str:
    """Return the hash of the last row in the chain, or empty string."""
    with self._lock:
        row = self._conn.execute(
            "SELECT row_hash FROM security_events ORDER BY id DESC LIMIT 1"
        ).fetchone()
    return row[0] if row and row[0] else ""

verify_chain

verify_chain() -> Tuple[bool, Optional[int]]

Verify the Merkle hash chain integrity.

RETURNS	DESCRIPTION
tuple	(True, None) if the chain is valid, or (False, row_id) where row_id is the first broken link.

Source code in src/diapason/security/audit.py

def verify_chain(self) -> Tuple[bool, Optional[int]]:
    """Verify the Merkle hash chain integrity.

    Returns
    -------
    tuple
        ``(True, None)`` if the chain is valid, or
        ``(False, row_id)`` where *row_id* is the first broken link.
    """
    with self._lock:
        rows = self._conn.execute(
            "SELECT id, timestamp, event_type, findings_json,"
            " content_preview, action_taken, row_hash, prev_hash"
            " FROM security_events ORDER BY id"
        ).fetchall()

    expected_prev = ""
    for row in rows:
        rid, ts, etype, fj, preview, action, stored_hash, stored_prev = row
        # Skip rows that predate the Merkle upgrade
        if not stored_hash:
            continue
        # Verify prev_hash link
        if stored_prev != expected_prev:
            return False, rid
        # Verify row_hash
        hash_input = f"{stored_prev}|{ts}|{etype}|{fj}|{preview}|{action}"
        computed = hashlib.sha256(hash_input.encode()).hexdigest()
        if computed != stored_hash:
            return False, rid
        expected_prev = stored_hash

    return True, None

count

count() -> int

Return the total number of logged security events.

Source code in src/diapason/security/audit.py

def count(self) -> int:
    """Return the total number of logged security events."""
    with self._lock:
        row = self._conn.execute("SELECT COUNT(*) FROM security_events").fetchone()
    return row[0] if row else 0

close

close() -> None

Close the SQLite connection.

Source code in src/diapason/security/audit.py

def close(self) -> None:
    """Close the SQLite connection."""
    with self._lock:
        self._conn.close()