secrets_store.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/registry/auth/secrets_store.py
 
# OCT-LINT: disable=tests_exist reason="Covered by oc_chat/tests/registry/test_auth_signing.py and test_auth_rotation.py"
"""
JSON-backed actor credential storage.
 
Purpose
-------
Load and write the M11.5 ``agent_secrets.json`` file that stores local
actor HMAC material outside SQLite.
 
Responsibilities
----------------
- Parse the slot/phase secret-file shape used by doctor and CLI flows.
- Encode and fingerprint raw HMAC secrets.
- Write updates atomically and apply restrictive POSIX file modes.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - lifecycle events
    L3 - semantic details
    L4 - deep tracing
 
Contracts
---------
- Raw secret bytes never enter the registry database.
- Existing unknown JSON fields are ignored on read.
- Writes use replace semantics so readers never observe partial JSON.
 
Dependencies
------------
- base64 (encode and decode the secret material)
- dataclasses (the SecretBlock record)
- hashlib (SHA-256 secret fingerprints)
- json (read and write the secrets file)
- os (fsync and atomic replace on write)
- pathlib (secrets file paths)
- secrets (fresh HMAC secrets and temporary file names)
- oc_diagnostics (structured debug logging via _dbg)
"""
from __future__ import annotations
 
import base64
from dataclasses import dataclass
import hashlib
import json
import os
from pathlib import Path
import secrets
from typing import Mapping
 
from oc_diagnostics import _dbg
 
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
 
SECRET_FILE_MODE = 0o600
DEFAULT_SECRETS_PATH = Path(".option_c") / "agent_secrets.json"
ROTATION_PHASES = {"active", "rotating_old", "rotating_new"}
 
 
@dataclass(frozen=True)
class SecretBlock:
    """One slot credential block from ``agent_secrets.json``."""
 
    slot: str
    phase: str
    hmac_key_id: str
    secret_b64: str
    secret_fingerprint: str
    created_at: str | None = None
    expires_at: str | None = None
 
    @property
    def raw_secret(self) -> bytes:
        """Decode the base64 secret material."""
 
        return base64.b64decode(self.secret_b64.encode("ascii"), validate=True)
 
    def to_json(self) -> dict[str, object]:
        """Return the canonical JSON block shape."""
 
        data: dict[str, object] = {
            "hmac_key_id": self.hmac_key_id,
            "secret_b64": self.secret_b64,
            "secret_fingerprint": self.secret_fingerprint,
        }
        if self.created_at is not None:
            data["created_at"] = self.created_at
        if self.expires_at is not None:
            data["expires_at"] = self.expires_at
        return data
 
 
def generate_secret() -> bytes:
    """Return a fresh 256-bit HMAC secret."""
 
    return secrets.token_bytes(32)
 
 
def encode_secret(raw_secret: bytes) -> str:
    """Return base64 text for raw HMAC secret bytes."""
 
    return base64.b64encode(raw_secret).decode("ascii")
 
 
def fingerprint_secret(raw_secret: bytes) -> str:
    """Return the DB-stored fingerprint for raw HMAC secret bytes."""
 
    return hashlib.sha256(raw_secret).hexdigest()
 
 
def block_from_secret(
    *,
    slot: str,
    phase: str,
    hmac_key_id: str,
    raw_secret: bytes,
    created_at: str | None = None,
    expires_at: str | None = None,
) -> SecretBlock:
    """Build a :class:`SecretBlock` from raw bytes."""
 
    if phase not in ROTATION_PHASES:
        raise ValueError(f"invalid rotation phase: {phase}")
    return SecretBlock(
        slot=slot,
        phase=phase,
        hmac_key_id=hmac_key_id,
        secret_b64=encode_secret(raw_secret),
        secret_fingerprint=fingerprint_secret(raw_secret),
        created_at=created_at,
        expires_at=expires_at,
    )
 
 
def load_secrets(path: str | Path = DEFAULT_SECRETS_PATH) -> dict[str, dict[str, SecretBlock]]:
    """Load secret blocks keyed by ``slot`` then ``rotation_phase``."""
 
    source = Path(path)
    if not source.is_file():
        return {}
    data = json.loads(source.read_text(encoding="utf-8"))
    if not isinstance(data, Mapping):
        raise ValueError("agent_secrets.json root must be an object")
    blocks: dict[str, dict[str, SecretBlock]] = {}
    for slot, slot_value in data.items():
        if not isinstance(slot, str) or not isinstance(slot_value, Mapping):
            continue
        phase_blocks: dict[str, SecretBlock] = {}
        for phase, phase_value in slot_value.items():
            if not isinstance(phase, str) or not isinstance(phase_value, Mapping):
                continue
            if phase not in ROTATION_PHASES:
                continue
            block = _parse_block(slot, phase, phase_value)
            if block is not None:
                phase_blocks[phase] = block
        if phase_blocks:
            blocks[slot] = phase_blocks
    return blocks
 
 
def write_secrets(
    blocks: Mapping[str, Mapping[str, SecretBlock]],
    path: str | Path = DEFAULT_SECRETS_PATH,
) -> Path:
    """Atomically write secret blocks and return the target path."""
 
    target = Path(path)
    target.parent.mkdir(parents=True, exist_ok=True)
    payload = {
        slot: {
            phase: block.to_json()
            for phase, block in sorted(phase_blocks.items())
        }
        for slot, phase_blocks in sorted(blocks.items())
    }
    text = json.dumps(payload, indent=2, sort_keys=True) + "\n"
    tmp = target.with_name(f".tmp-{target.name}-{secrets.token_hex(8)}")
    with tmp.open("w", encoding="utf-8", newline="\n") as handle:
        handle.write(text)
        handle.flush()
        os.fsync(handle.fileno())
    _chmod_secret(tmp)
    os.replace(tmp, target)
    _chmod_secret(target)
    return target
 
 
def select_secret(
    blocks: Mapping[str, Mapping[str, SecretBlock]],
    *,
    slot: str,
    hmac_key_id: str | None = None,
    phase_preference: tuple[str, ...] = ("rotating_new", "active", "rotating_old"),
) -> SecretBlock | None:
    """Return the matching block for a slot/key id."""
 
    slot_blocks = blocks.get(slot)
    if slot_blocks is None:
        return None
    if hmac_key_id is not None:
        for block in slot_blocks.values():
            if block.hmac_key_id == hmac_key_id:
                return block
        return None
    for phase in phase_preference:
        block = slot_blocks.get(phase)
        if block is not None:
            return block
    return None
 
 
def _parse_block(
    slot: str,
    phase: str,
    value: Mapping[str, object],
) -> SecretBlock | None:
    key_id = value.get("hmac_key_id")
    secret_b64 = value.get("secret_b64")
    fingerprint = value.get("secret_fingerprint")
    if not isinstance(key_id, str) or not isinstance(secret_b64, str):
        return None
    if not isinstance(fingerprint, str):
        fingerprint = ""
    created_at = value.get("created_at")
    expires_at = value.get("expires_at")
    return SecretBlock(
        slot=slot,
        phase=phase,
        hmac_key_id=key_id,
        secret_b64=secret_b64,
        secret_fingerprint=fingerprint,
        created_at=created_at if isinstance(created_at, str) else None,
        expires_at=expires_at if isinstance(expires_at, str) else None,
    )
 
 
def _chmod_secret(path: Path) -> None:
    try:
        path.chmod(SECRET_FILE_MODE)
    except OSError:
        pass