roles.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/registry/auth/roles.py
 
# OCT-LINT: disable=tests_exist reason="Covered by oc_chat/tests/registry/test_auth_roles.py"
"""
Runtime role guard for registry operations.
 
Purpose
-------
Map registry operation namespaces to the actor slot kinds that may author
them.
 
Responsibilities
----------------
- Keep the allow table in one auditable location.
- Raise typed role errors before any operation-log mutation starts.
- Provide an explicit system-actor bypass for enumerated internal rows.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - lifecycle events
    L3 - semantic details
    L4 - deep tracing
 
Contracts
---------
- A role mismatch is not an auth failure and does not write ``auth_events``.
- Unknown test/private operation namespaces remain compatibility-allowed.
 
Dependencies
------------
- oc_chat.registry.errors (ActorRoleMismatchError raised on a role mismatch)
- oc_diagnostics (structured debug logging via _dbg)
"""
from __future__ import annotations
 
from oc_diagnostics import _dbg
 
from oc_chat.registry.errors import ActorRoleMismatchError
 
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
SlotKind = str
 
ALL_SLOT_KINDS = frozenset({"operator", "orchestrator", "worker"})
SYSTEM_OPERATION_PREFIXES = ("migrate.", "heartbeat.")
SYSTEM_OPERATION_TYPES = frozenset({"doctor.diagnose"})
 
ROLE_ALLOWSETS: dict[str, frozenset[SlotKind]] = {
    "auth.register_key": frozenset({"operator"}),
    "auth.rotate_start": frozenset({"operator"}),
    "auth.rotate_finalize": frozenset({"operator"}),
    "auth.reset": frozenset({"operator"}),
    "audit.override": frozenset({"operator", "orchestrator"}),
    "audit.verify": frozenset({"operator", "orchestrator"}),
    "db.archive_log.start": frozenset({"operator"}),
    "db.archive_log.complete": frozenset({"operator"}),
    "doctor.repair": frozenset({"operator"}),
    "handover.accept": frozenset({"orchestrator", "operator"}),
    "handover.initiate": frozenset({"orchestrator", "operator"}),
    "handover.timeout": frozenset({"operator"}),
    "role.handover.complete": frozenset({"orchestrator", "operator"}),
    "role.handover.start": frozenset({"orchestrator", "operator"}),
}
 
PREFIX_ALLOWSETS: tuple[tuple[str, frozenset[SlotKind]], ...] = (
    ("plan.", frozenset({"operator", "orchestrator", "worker"})),
    ("queue.", frozenset({"operator", "orchestrator"})),
    ("roadmap_", frozenset({"operator", "orchestrator"})),
    ("roadmap.", frozenset({"operator", "orchestrator"})),
    ("round.", frozenset({"orchestrator"})),
    ("migrate.", frozenset({"operator"})),
    ("db.", frozenset({"operator"})),
    ("doctor.", frozenset({"operator"})),
    ("_test.", ALL_SLOT_KINDS),
)
 
 
class RoleGuard:
    """Table-driven operation role checker."""
 
    def check(
        self,
        operation_type: str,
        actor_slot_kind: str,
        *,
        actor_slot: str,
        system_actor: bool = False,
    ) -> None:
        """Raise if ``actor_slot_kind`` cannot author ``operation_type``."""
 
        if system_actor and _is_system_operation(operation_type):
            return
        allowed = self.allowed_slot_kinds(operation_type)
        if actor_slot_kind not in allowed:
            raise ActorRoleMismatchError(
                operation_type=operation_type,
                actor_slot=actor_slot,
                slot_kind=actor_slot_kind,
                allowed_slot_kinds=tuple(sorted(allowed)),
            )
 
    def allowed_slot_kinds(self, operation_type: str) -> frozenset[SlotKind]:
        """Return allowed slot kinds for an operation type."""
 
        exact = ROLE_ALLOWSETS.get(operation_type)
        if exact is not None:
            return exact
        for prefix, allowed in PREFIX_ALLOWSETS:
            if operation_type.startswith(prefix):
                return allowed
        return ALL_SLOT_KINDS
 
 
def _is_system_operation(operation_type: str) -> bool:
    return operation_type in SYSTEM_OPERATION_TYPES or operation_type.startswith(
        SYSTEM_OPERATION_PREFIXES
    )