v0_3_1_auth_events.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/registry/migrations/v0_3_1_auth_events.py
 
# OCT-LINT: disable=tests_exist reason="Covered by oc_chat/tests/registry/test_auth_signing.py"
"""
Runtime migration for auth_events v0.3.1.
 
Purpose
-------
Rebuild ``auth_events`` so unknown-key failures can be recorded with a NULL
``hmac_key_id`` while successful auth rows still require a credential key.
 
Responsibilities
----------------
- Preserve existing auth event rows.
- Keep the FK to ``agent_credentials`` for non-NULL key ids.
- Bump SQLite ``user_version`` to 31.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - migration lifecycle
    L3 - schema version decisions
    L4 - table definition checks
 
Contracts
---------
- Callers own transaction boundaries.
- The migration is idempotent.
 
Dependencies
------------
- sqlite3 (migration connection)
- oc_diagnostics (structured debug logging via _dbg)
"""
from __future__ import annotations
 
import sqlite3
 
from oc_diagnostics import _dbg
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
AUTH_EVENTS_SCHEMA_VERSION = 31
AUTH_EVENTS_SCHEMA_DESCRIPTION = "auth_events REGISTRY_SCHEMA_V0_3_1"
 
AUTH_EVENTS_CHECK = "auth_ok = 0 OR hmac_key_id IS NOT NULL"
 
 
def current_user_version(conn: sqlite3.Connection) -> int:
    """Return SQLite's integer schema user version."""
 
    return int(conn.execute("PRAGMA user_version").fetchone()[0])
 
 
def apply(conn: sqlite3.Connection) -> None:
    """Apply the auth_events v0.3.1 rebuild if needed."""
 
    if is_applied(conn):
        conn.execute(f"PRAGMA user_version = {AUTH_EVENTS_SCHEMA_VERSION}")
        return
    _rebuild_auth_events(conn)
    conn.execute(f"PRAGMA user_version = {AUTH_EVENTS_SCHEMA_VERSION}")
 
 
def rollback(conn: sqlite3.Connection) -> None:
    """Reset user_version for tests; table shape remains backward-compatible."""
 
    conn.execute("PRAGMA user_version = 3")
    conn.execute(
        "DELETE FROM schema_migrations WHERE version = ?",
        (AUTH_EVENTS_SCHEMA_VERSION,),
    )
 
 
def is_applied(conn: sqlite3.Connection) -> bool:
    """Return True when the auth_events CHECK is present."""
 
    row = conn.execute(
        # OCT-LINT: disable=no_hardcoded_secrets reason="Schema object names are migration constants, not secrets"
        "SELECT sql FROM sqlite_master WHERE type = 'table' AND name = 'auth_events'"
    ).fetchone()
    if row is None:
        return False
    try:
        sql = row["sql"]
    except (TypeError, IndexError):
        sql = row[0]
    return AUTH_EVENTS_CHECK in str(sql)
 
 
def migration_sql() -> str:
    """Return SQL used when building a fresh schema in one executescript call."""
 
    return _REBUILD_SQL + f"\nPRAGMA user_version = {AUTH_EVENTS_SCHEMA_VERSION};\n"
 
 
def _rebuild_auth_events(conn: sqlite3.Connection) -> None:
    conn.executescript(_REBUILD_SQL)
 
 
# OCT-LINT: disable=no_hardcoded_secrets reason="Schema object names are migration constants, not secrets"
_REBUILD_SQL = """
ALTER TABLE auth_events RENAME TO auth_events_old;
 
CREATE TABLE auth_events (
  id INTEGER PRIMARY KEY AUTOINCREMENT,
  ts TEXT NOT NULL,
  attempted_slot TEXT,
  hmac_key_id TEXT REFERENCES agent_credentials(hmac_key_id),
  command TEXT NOT NULL,
  payload_sha256 TEXT,
  auth_ok INTEGER NOT NULL CHECK(auth_ok IN (0,1)),
  failure_reason TEXT
    CHECK(failure_reason IN
      ('bad_signature','expired_ts','unknown_slot','no_credential',
       'revoked_key','malformed_payload','clock_skew','missing_secrets_file')),
  remote_hint TEXT CHECK(remote_hint IS NULL OR length(remote_hint) <= 255),
  CHECK (
    (auth_ok = 1 AND failure_reason IS NULL) OR
    (auth_ok = 0 AND failure_reason IS NOT NULL)
  ),
  CHECK (auth_ok = 0 OR hmac_key_id IS NOT NULL)
);
 
INSERT INTO auth_events(
  id, ts, attempted_slot, hmac_key_id, command, payload_sha256,
  auth_ok, failure_reason, remote_hint
)
SELECT
  id, ts, attempted_slot, hmac_key_id, command, payload_sha256,
  auth_ok, failure_reason, remote_hint
FROM auth_events_old;
 
DROP TABLE auth_events_old;
 
CREATE INDEX idx_auth_events_ts ON auth_events(ts);
CREATE INDEX idx_auth_events_slot ON auth_events(attempted_slot, ts);
CREATE INDEX idx_auth_events_slot_failure
  ON auth_events(attempted_slot, failure_reason, ts)
  WHERE auth_ok = 0;
"""