events_writer.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/registry/auth/events_writer.py
 
# OCT-LINT: disable=tests_exist reason="Covered by oc_chat/tests/registry/test_auth_signing.py"
"""
Dedicated auth-event writer.
 
Purpose
-------
Record failed HMAC verification attempts in a short transaction that is
separate from the main registry mutation.
 
Responsibilities
----------------
- Insert one ``auth_events`` row for each failed verification.
- Keep failure writes out of ``operation_log``.
- Avoid exposing secret material in diagnostics.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - lifecycle events
    L3 - semantic details
    L4 - deep tracing
 
Contracts
---------
- The caller has not opened the mutating registry transaction yet.
- ``failure_reason`` is one of the schema-enumerated values.
 
Dependencies
------------
- oc_chat.registry.backends.sqlite (SQLite backend whose dedicated transaction records the failure row)
- oc_chat.registry.auth.signing (payload digest stored with each failure)
- oc_diagnostics (structured debug logging via _dbg)
"""
from __future__ import annotations
 
from typing import Mapping
 
from oc_diagnostics import _dbg
 
from oc_chat.registry.auth.signing import payload_sha256
from oc_chat.registry.backends.sqlite import SQLiteBackend
 
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
 
class AuthEventsWriter:
    """Write failed-auth rows through a dedicated transaction."""
 
    def __init__(self, backend: SQLiteBackend) -> None:
        self.backend = backend
 
    def write_failure(
        self,
        *,
        attempted_slot: str | None,
        hmac_key_id: str | None,
        command: str,
        payload: Mapping[str, object],
        failure_reason: str,
        remote_hint: str | None = None,
    ) -> int:
        """Insert one failed-auth row and return its row id."""
 
        digest = payload_sha256(payload)
        with self.backend.transaction("immediate") as conn:
            cursor = conn.execute(
                # OCT-LINT: disable=no_hardcoded_secrets reason="Auth event field names are schema identifiers"
                "INSERT INTO auth_events("
                "ts, attempted_slot, hmac_key_id, command, payload_sha256, "
                "auth_ok, failure_reason, remote_hint"
                ") VALUES (?, ?, ?, ?, ?, 0, ?, ?)",
                (
                    self.backend.clock.now_wall(),
                    attempted_slot,
                    hmac_key_id,
                    command,
                    digest,
                    failure_reason,
                    remote_hint,
                ),
            )
            return int(cursor.lastrowid)