rotation.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/registry/auth/rotation.py
 
# OCT-LINT: disable=tests_exist reason="Covered by oc_chat/tests/registry/test_auth_rotation.py"
"""
Actor HMAC credential rotation.
 
Purpose
-------
Implement the two-phase credential rotation used by M11.5 actor auth.
 
Responsibilities
----------------
- Create a rotating-new credential and expire the previous active key.
- Finalize a rotation by revoking the old key and promoting the new key.
- Keep the local secret file in sync with generated key material.
- Emit audited ``auth.rotate_*`` rows through the registry seam.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - lifecycle events
    L3 - semantic details
    L4 - deep tracing
 
Contracts
---------
- Raw secrets are written only to ``agent_secrets.json``.
- Credential-table writes happen before their audit operation is emitted.
- Nested rotations for one slot are rejected.
 
Dependencies
------------
- pathlib (secrets file paths)
- uuid (key ids for new credentials)
- oc_chat.registry.auth.secrets_store (generate new secrets and keep the local secrets file in sync)
- oc_chat.registry.clock (grace-period expiry timestamp)
- oc_chat.registry.errors (RotationInFlightError for a nested rotation)
- oc_chat.registry.api (Registry that emits the audited rotation operations)
- oc_chat.registry.backends.base (OperationLogRow carried in the rotation result)
- oc_diagnostics (structured debug logging via _dbg)
"""
from __future__ import annotations
 
from dataclasses import dataclass
from pathlib import Path
import uuid
 
from oc_diagnostics import _dbg
 
from oc_chat.registry.api import Registry
from oc_chat.registry.auth.secrets_store import (
    SecretBlock,
    block_from_secret,
    generate_secret,
    load_secrets,
    write_secrets,
)
from oc_chat.registry.backends.base import OperationLogRow
from oc_chat.registry.clock import wall_plus_seconds
from oc_chat.registry.errors import RotationInFlightError
 
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
 
@dataclass(frozen=True)
class RotationResult:
    """Result for one rotation start/finalize call."""
 
    slot: str
    old_key_id: str
    new_key_id: str
    operation: OperationLogRow
 
 
def rotate_secret(
    registry: Registry,
    *,
    slot: str,
    actor_slot: str,
    actor_type: str = "human_operator",
    grace_seconds: int | None = None,
    secrets_path: str | Path | None = None,
    trigger: str = "manual",
) -> RotationResult:
    """Start a two-phase credential rotation for ``slot``."""
 
    backend = registry.backend
    connect_func = getattr(backend, "connect", None)
    transaction_func = getattr(backend, "transaction", None)
    if connect_func is None or transaction_func is None:
        raise TypeError("rotation requires a SQLite-compatible registry backend")
    now = registry.clock.now_wall()
    effective_grace = _effective_grace_seconds(backend, slot, grace_seconds)
    expires_at = wall_plus_seconds(now, effective_grace)
    raw_secret = generate_secret()
    new_key_id = str(uuid.uuid4())
    new_block = block_from_secret(
        slot=slot,
        phase="rotating_new",
        hmac_key_id=new_key_id,
        raw_secret=raw_secret,
        created_at=now,
        expires_at=None,
    )
    with transaction_func("immediate") as conn:
        in_flight = conn.execute(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Credential field names are schema identifiers"
            "SELECT 1 FROM agent_credentials WHERE slot = ? "
            "AND rotation_phase IN ('rotating_old', 'rotating_new') "
            "AND revoked_at IS NULL",
            (slot,),
        ).fetchone()
        if in_flight is not None:
            raise RotationInFlightError(slot)
        active = conn.execute(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Credential field names are schema identifiers"
            "SELECT hmac_key_id, secret_fingerprint FROM agent_credentials "
            "WHERE slot = ? AND rotation_phase = 'active' AND revoked_at IS NULL",
            (slot,),
        ).fetchone()
        if active is None:
            raise ValueError(f"no active credential for slot {slot!r}")
        old_key_id = str(active["hmac_key_id"])
        old_fingerprint = str(active["secret_fingerprint"])
        conn.execute(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Credential field names are schema identifiers"
            "UPDATE agent_credentials SET rotation_phase = 'rotating_old', "
            "expires_at = ? WHERE hmac_key_id = ?",
            (expires_at, old_key_id),
        )
        conn.execute(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Credential field names are schema identifiers"
            "INSERT INTO agent_credentials("
            "hmac_key_id, slot, secret_fingerprint, created_at, expires_at, "
            "rotation_phase"
            ") VALUES (?, ?, ?, ?, NULL, 'rotating_new')",
            (new_key_id, slot, new_block.secret_fingerprint, now),
        )
    _write_rotation_secret(
        backend,
        slot=slot,
        new_block=new_block,
        secrets_path=secrets_path,
        expires_at=expires_at,
    )
    operation = registry.execute_domain_command(
        "auth.rotate_start",
        actor_slot,
        {
            "version": 1,
            "slot": slot,
            "old_key_id": old_key_id,
            "new_key_id": new_key_id,
            "old_key_fingerprint": old_fingerprint,
            "new_key_fingerprint": new_block.secret_fingerprint,
            "effective_grace_seconds": effective_grace,
            "grace_source": "cli_override"
            if grace_seconds is not None
            else "global_default",
            "expires_at": expires_at,
            "trigger": trigger,
        },
        actor_type=actor_type,
    )
    return RotationResult(slot, old_key_id, new_key_id, operation)
 
 
def finalize_rotation(
    registry: Registry,
    *,
    slot: str,
    actor_slot: str,
    actor_type: str = "human_operator",
    secrets_path: str | Path | None = None,
    trigger: str = "manual_finalize",
) -> RotationResult:
    """Finalize a pending rotation for ``slot``."""
 
    backend = registry.backend
    transaction_func = getattr(backend, "transaction", None)
    if transaction_func is None:
        raise TypeError("rotation requires a SQLite-compatible registry backend")
    now = registry.clock.now_wall()
    with transaction_func("immediate") as conn:
        old = conn.execute(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Credential field names are schema identifiers"
            "SELECT hmac_key_id, created_at FROM agent_credentials "
            "WHERE slot = ? AND rotation_phase = 'rotating_old' "
            "AND revoked_at IS NULL",
            (slot,),
        ).fetchone()
        new = conn.execute(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Credential field names are schema identifiers"
            "SELECT hmac_key_id FROM agent_credentials "
            "WHERE slot = ? AND rotation_phase = 'rotating_new' "
            "AND revoked_at IS NULL",
            (slot,),
        ).fetchone()
        if old is None or new is None:
            raise ValueError(f"no rotation in flight for slot {slot!r}")
        old_key_id = str(old["hmac_key_id"])
        new_key_id = str(new["hmac_key_id"])
        conn.execute(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Credential field names are schema identifiers"
            "UPDATE agent_credentials SET revoked_at = ?, revoked_by = ? "
            "WHERE hmac_key_id = ?",
            (now, actor_slot, old_key_id),
        )
        conn.execute(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Credential field names are schema identifiers"
            "UPDATE agent_credentials SET rotation_phase = 'active' "
            "WHERE hmac_key_id = ?",
            (new_key_id,),
        )
    _promote_secret_block(backend, slot=slot, secrets_path=secrets_path)
    start_row = _latest_rotation_start_id(backend, slot, old_key_id, new_key_id)
    operation = registry.execute_domain_command(
        "auth.rotate_finalize",
        actor_slot,
        {
            "version": 1,
            "slot": slot,
            "old_key_id": old_key_id,
            "new_key_id": new_key_id,
            "trigger": trigger,
            "started_at": str(old["created_at"]),
            "completed_at": now,
            "parent_operation_id": start_row,
        },
        actor_type=actor_type,
        parent_operation_id=start_row,
    )
    return RotationResult(slot, old_key_id, new_key_id, operation)
 
 
def _effective_grace_seconds(
    backend: object,
    slot: str,
    grace_seconds: int | None,
) -> int:
    if grace_seconds is not None:
        return grace_seconds
    connect_func = getattr(backend, "connect")
    with connect_func() as conn:
        row = conn.execute(
            "SELECT value FROM settings WHERE key = ?",
            (f"hmac_rotation_grace_seconds.{slot}",),
        ).fetchone()
        if row is None:
            row = conn.execute(
                # OCT-LINT: disable=no_hardcoded_secrets reason="Credential field names are schema identifiers"
                "SELECT value FROM settings WHERE key = 'hmac_rotation_grace_seconds'"
            ).fetchone()
    return 86400 if row is None else int(row["value"])
 
 
def _write_rotation_secret(
    backend: object,
    *,
    slot: str,
    new_block: SecretBlock,
    secrets_path: str | Path | None,
    expires_at: str,
) -> None:
    path = _secrets_path(backend, secrets_path)
    blocks = load_secrets(path)
    slot_blocks = dict(blocks.get(slot, {}))
    active = slot_blocks.get("active")
    if active is not None:
        slot_blocks["rotating_old"] = SecretBlock(
            slot=active.slot,
            phase="rotating_old",
            hmac_key_id=active.hmac_key_id,
            secret_b64=active.secret_b64,
            secret_fingerprint=active.secret_fingerprint,
            created_at=active.created_at,
            expires_at=expires_at,
        )
        slot_blocks.pop("active", None)
    slot_blocks["rotating_new"] = new_block
    blocks[slot] = slot_blocks
    write_secrets(blocks, path)
 
 
def _promote_secret_block(
    backend: object,
    *,
    slot: str,
    secrets_path: str | Path | None,
) -> None:
    path = _secrets_path(backend, secrets_path)
    blocks = load_secrets(path)
    slot_blocks = dict(blocks.get(slot, {}))
    new = slot_blocks.get("rotating_new")
    if new is None:
        return
    slot_blocks["active"] = SecretBlock(
        slot=new.slot,
        phase="active",
        hmac_key_id=new.hmac_key_id,
        secret_b64=new.secret_b64,
        secret_fingerprint=new.secret_fingerprint,
        created_at=new.created_at,
        expires_at=None,
    )
    slot_blocks.pop("rotating_new", None)
    slot_blocks.pop("rotating_old", None)
    blocks[slot] = slot_blocks
    write_secrets(blocks, path)
 
 
def _latest_rotation_start_id(
    backend: object,
    slot: str,
    old_key_id: str,
    new_key_id: str,
) -> int | None:
    connect_func = getattr(backend, "connect")
    with connect_func() as conn:
        row = conn.execute(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Credential field names are schema identifiers"
            "SELECT id FROM operation_log WHERE operation_type = "
            "'auth.rotate_start' AND args_json LIKE ? AND args_json LIKE ? "
            "AND args_json LIKE ? ORDER BY id DESC LIMIT 1",
            (f"%{slot}%", f"%{old_key_id}%", f"%{new_key_id}%"),
        ).fetchone()
    return None if row is None else int(row["id"])
 
 
def _secrets_path(backend: object, override: str | Path | None) -> Path:
    if override is not None:
        return Path(override)
    backend_path = getattr(backend, "path", None)
    if isinstance(backend_path, Path) and backend_path.name != ":memory:":
        if (
            backend_path.parent.name == "orchestration"
            and backend_path.parent.parent.name == ".option_c"
        ):
            return backend_path.parent.parent / "agent_secrets.json"
        return backend_path.parent / "agent_secrets.json"
    return Path(".option_c") / "agent_secrets.json"