signing.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/registry/auth/signing.py
 
# OCT-LINT: disable=tests_exist reason="Covered by oc_chat/tests/registry/test_auth_signing.py"
"""
Actor HMAC signing and verification.
 
Purpose
-------
Build and verify canonical HMAC-SHA256 proofs for mutating registry
commands.
 
Responsibilities
----------------
- Canonicalize the signed command payload.
- Produce proof mappings for CLI/API callers.
- Verify signatures with constant-time comparison and skew checks.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - lifecycle events
    L3 - semantic details
    L4 - deep tracing
 
Contracts
---------
- The signed payload includes slot, command, args, timestamp, and key id.
- Verification returns a structured failure reason for ``auth_events``.
- This module performs no registry writes and reads no secret files.
 
Dependencies
------------
- dataclasses (proof, credential and result records)
- hashlib (SHA-256 for signatures and payload digests)
- hmac (HMAC signing and constant-time comparison)
- oc_chat.registry.canonical (canonical JSON of the signed payload)
- oc_chat.registry.clock (timestamps for signing, skew and expiry checks)
- oc_diagnostics (structured debug logging via _dbg)
"""
from __future__ import annotations
 
from dataclasses import dataclass
import hashlib
import hmac
from typing import Mapping
 
from oc_diagnostics import _dbg
 
from oc_chat.registry.canonical import canonical_json
from oc_chat.registry.clock import Clock, RealClock
 
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
SIGNATURE_ALGORITHM = "HMAC-SHA256"
 
 
@dataclass(frozen=True)
class HmacProof:
    """Client-supplied proof fields for one registry mutation."""
 
    hmac_key_id: str
    ts_ms: int
    signature_hex: str
    algorithm: str = SIGNATURE_ALGORITHM
 
    def to_mapping(self) -> dict[str, object]:
        """Return a JSON-safe proof mapping."""
 
        return {
            "algorithm": self.algorithm,
            "hmac_key_id": self.hmac_key_id,
            "ts_ms": self.ts_ms,
            "signature_hex": self.signature_hex,
        }
 
    @classmethod
    def from_mapping(cls, value: Mapping[str, object]) -> "HmacProof":
        """Parse a proof mapping."""
 
        key_id = value.get("hmac_key_id")
        signature = value.get("signature_hex")
        ts_ms = value.get("ts_ms")
        algorithm = value.get("algorithm", SIGNATURE_ALGORITHM)
        if not isinstance(key_id, str) or not key_id:
            raise ValueError("hmac_key_id is required")
        if not isinstance(signature, str) or not signature:
            raise ValueError("signature_hex is required")
        if not isinstance(ts_ms, int):
            raise ValueError("ts_ms is required")
        if algorithm != SIGNATURE_ALGORITHM:
            raise ValueError("unsupported HMAC algorithm")
        return cls(hmac_key_id=key_id, ts_ms=ts_ms, signature_hex=signature)
 
 
@dataclass(frozen=True)
class CredentialMaterial:
    """Credential fields needed by HMAC verification."""
 
    hmac_key_id: str
    slot: str
    raw_secret: bytes
    rotation_phase: str
    revoked_at: str | None = None
    expires_at: str | None = None
 
 
@dataclass(frozen=True)
class VerifyResult:
    """Outcome returned by ``verify_hmac``."""
 
    ok: bool
    hmac_key_id: str | None
    failure_reason: str | None = None
 
 
def sign_command(
    *,
    slot: str,
    command: str,
    args: Mapping[str, object],
    raw_secret: bytes,
    hmac_key_id: str,
    clock: Clock | None = None,
    ts_ms: int | None = None,
) -> HmacProof:
    """Return an HMAC proof for one command payload."""
 
    effective_ts = _now_ms(clock) if ts_ms is None else ts_ms
    payload = canonical_payload(
        slot=slot,
        command=command,
        args=args,
        ts_ms=effective_ts,
        hmac_key_id=hmac_key_id,
    )
    return HmacProof(
        hmac_key_id=hmac_key_id,
        ts_ms=effective_ts,
        signature_hex=_sign(payload, raw_secret),
    )
 
 
def verify_hmac(
    *,
    slot: str,
    command: str,
    args: Mapping[str, object],
    proof: HmacProof,
    credential: CredentialMaterial | None,
    clock: Clock | None = None,
    max_skew_seconds: int = 300,
) -> VerifyResult:
    """Verify one HMAC proof against resolved credential material."""
 
    if credential is None:
        return VerifyResult(False, proof.hmac_key_id, "no_credential")
    if credential.slot != slot:
        return VerifyResult(False, proof.hmac_key_id, "bad_signature")
    if credential.revoked_at is not None:
        return VerifyResult(False, proof.hmac_key_id, "revoked_key")
    now_wall_func = getattr(clock or RealClock(), "now_wall")
    now_wall = now_wall_func()
    if credential.expires_at is not None and credential.expires_at < now_wall:
        return VerifyResult(False, proof.hmac_key_id, "revoked_key")
    if max_skew_seconds >= 0:
        skew_ms = abs(_now_ms(clock) - proof.ts_ms)
        if skew_ms > max_skew_seconds * 1000:
            return VerifyResult(False, proof.hmac_key_id, "clock_skew")
    expected = _sign(
        canonical_payload(
            slot=slot,
            command=command,
            args=args,
            ts_ms=proof.ts_ms,
            hmac_key_id=proof.hmac_key_id,
        ),
        credential.raw_secret,
    )
    if not hmac.compare_digest(expected, proof.signature_hex):
        return VerifyResult(False, proof.hmac_key_id, "bad_signature")
    return VerifyResult(True, credential.hmac_key_id)
 
 
def canonical_payload(
    *,
    slot: str,
    command: str,
    args: Mapping[str, object],
    ts_ms: int,
    hmac_key_id: str,
) -> str:
    """Return canonical JSON text for the signed command."""
 
    return canonical_json(
        {
            "version": 1,
            "slot": slot,
            "command": command,
            "args": args,
            "ts_ms": ts_ms,
            "hmac_key_id": hmac_key_id,
        }
    )
 
 
def payload_sha256(args: Mapping[str, object]) -> str:
    """Return the canonical payload digest used in ``auth_events``."""
 
    return hashlib.sha256(canonical_json(args).encode("utf-8")).hexdigest()
 
 
def _sign(payload: str, raw_secret: bytes) -> str:
    return hmac.new(
        raw_secret,
        payload.encode("utf-8"),
        hashlib.sha256,
    ).hexdigest()
 
 
def _now_ms(clock: Clock | None) -> int:
    effective_clock = RealClock() if clock is None else clock
    return effective_clock.now_monotonic_ns() // 1_000_000