orchestrata_ultimate.gateway.handlers.config

Reference › orchestrata_ultimate › gateway › handlers › config

ConfigHandler serves the runtime-configuration RPCs config.get, config.watch and config.set (SPEC-014 US-014-005). It sits on top of the Redis-backed RuntimeConfigStore and SecretRedactor from runtime_config, the pubsub manager and the scope guard; rpc wires it, and set_store attaches the store once it exists.

• handle_get requires read_only scope and returns the values, optionally narrowed by a dotted path, with secret-named keys masked by the redactor and a last_updated time. With no store it returns an empty result.
• handle_watch requires read_only scope and subscribes the connection to the config.changed topic, with optional filters. Delivery is done by the gateway's config-event forwarder in server.
• handle_set always returns a structured config_read_only denial. It is registered so the denial is testable and shows in the audit log, rather than appearing as a method-not-found error. Writes come only from in-process producers.

See gateway.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

ConfigHandler

Line 32

config.get/watch/set -- read-only runtime settings surface.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/gateway/handlers/config.py

"""Runtime configuration RPC handlers -- get/watch/set.

Implements SPEC-014 US-014-005 (REQ-014-081..085). ``handle_get``/
``handle_watch`` require ``read_only`` scope (any authenticated caller);
``handle_set`` NEVER mutates -- it always returns a structured
``config_read_only`` denial (REQ-014-085, D6). Every outbound value passes
through :class:`SecretRedactor` first (REQ-014-083).
"""

import logging
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.tracing.tracer import Tracer
from orchestrata_ultimate.gateway.handlers._scope_guard import ScopeGuard
from orchestrata_ultimate.gateway.pubsub import PubSubManager
from orchestrata_ultimate.gateway.runtime_config import (
    RuntimeConfigStore,
    SecretRedactor,
)

logger = logging.getLogger(__name__)

_READ_ONLY_DENIAL: dict[str, Any] = {
    "error": {
        "code": "config_read_only",
        "message": "Config changes not permitted via gateway (read-only for safety)",
    }
}


class ConfigHandler:
    """config.get/watch/set -- read-only runtime settings surface."""

    def __init__(
        self,
        redactor: SecretRedactor,
        scope_guard: ScopeGuard,
        pubsub: PubSubManager,
        store: RuntimeConfigStore | None = None,
        tracer: Tracer | None = None,
    ) -> None:
        """Wire the handler to its redaction/scope/pubsub strategies and
        (optionally) an already-connected store.

        Args:
            redactor: Secret-value redaction (REQ-014-083).
            scope_guard: Per-connection scope enforcement.
            pubsub: Pub/sub subscription manager (``config.watch``
                registers the ``config.changed`` subscription here;
                delivery happens via ``MessageProcessor.broadcast``, D7).
            store: The runtime-config store, or None until
                :meth:`set_store` is called (e.g. at gateway boot, once
                ``_init_config_service`` connects Redis).
            tracer: Injected debug tracer; defaults to the process Singleton.
        """
        self._redactor = redactor
        self._scope_guard = scope_guard
        self._pubsub = pubsub
        self._store = store
        self._tracer = tracer or Tracer.default()

    def set_store(self, store: RuntimeConfigStore) -> None:
        """Wire the live runtime-config store.

        Called from ``RPCHandler.set_runtime_config_store`` (invoked by
        ``WebSocketGateway._init_config_service``).

        Args:
            store: A connected ``RuntimeConfigStore``.
        """
        self._store = store

    @traces(
        spec="014",
        story="US-014-005",
        reqs=["REQ-014-081", "REQ-014-082", "REQ-014-083", "REQ-014-084"],
    )
    async def handle_get(
        self, connection_id: str, params: dict[str, Any]
    ) -> dict[str, Any]:
        """config.get -- read (optionally path-scoped) runtime settings.

        Args:
            connection_id: Caller connection ID.
            params: Optional ``path`` dotted-path selector (REQ-014-082).

        Returns:
            ``{"values": <redacted>, "path": ..., "last_updated": ...}``.
        """
        await self._scope_guard.require(connection_id, "read_only")
        path = params.get("path")
        if self._store is None:
            return {"values": {}, "path": path, "last_updated": None}
        snapshot = await self._store.get(path)
        return {
            "values": self._redactor.redact(snapshot.values),
            "path": snapshot.path,
            "last_updated": snapshot.last_updated,
        }

    @traces(spec="014", story="US-014-005", reqs=["REQ-014-081"])
    async def handle_watch(
        self, connection_id: str, params: dict[str, Any]
    ) -> dict[str, Any]:
        """config.watch -- subscribe to ``config.changed`` (REQ-014-081).

        Delivery happens via the gateway's ``_consume_config_events``
        forwarder pushing through ``MessageProcessor.broadcast`` -- never
        ``PubSubManager.publish``, which is only a registry counter (D7).

        Args:
            connection_id: Caller connection ID.
            params: Optional ``filters`` for the subscription.

        Returns:
            ``{"subscription_id": ..., "topic": "config.changed"}``.
        """
        await self._scope_guard.require(connection_id, "read_only")
        sub_id = await self._pubsub.subscribe(
            connection_id, "config.changed", params.get("filters", {})
        )
        return {"subscription_id": sub_id, "topic": "config.changed"}

    @traces(spec="014", story="US-014-005", reqs=["REQ-014-085"])
    async def handle_set(
        self, connection_id: str, params: dict[str, Any]
    ) -> dict[str, Any]:
        """config.set -- ALWAYS a structured denial (REQ-014-085, D6).

        Registered (rather than left unbound) so the denial is testable
        and auditable via the RPC dispatcher's existing ``rpc.*`` audit
        log, instead of a generic ``-32601`` that proves nothing.

        Args:
            connection_id: Caller connection ID (unused -- denial applies
                uniformly).
            params: RPC params (unused -- denial applies uniformly).

        Returns:
            The structured ``config_read_only`` denial.
        """
        return dict(_READ_ONLY_DENIAL)