orchestrata_ultimate.gateway.config

Reference › orchestrata_ultimate › gateway › config

config defines GatewayConfig, the single frozen dataclass from which every gateway component takes its settings. Almost every module of the package imports it, including server, auth, ring_buffer, session and audit. See gateway for the package context.

Every field has a default_factory that reads the environment, so constructing the class reads the process environment. Passing an environ mapping makes __post_init__ override every field from that mapping through _override_from_environ, which tests use for hermetic configuration. The fields fall into groups:

• Server and HTTP surface: GATEWAY_HOST, GATEWAY_PORT (8765), GATEWAY_HEALTH_ENABLED, GATEWAY_HTTP_PORT (8766) and GATEWAY_HTTP_HOST; http_bind_host returns the HTTP host or falls back to the WebSocket host.
• Auth: the JWT secret from _resolve_jwt_secret, auth_mode from GATEWAY_AUTH_MODE, the static token, HMAC secrets, drift window and client scopes, the token lifetime setting TOKEN_EXPIRY_SECONDS, and GATEWAY_ADMIN_TOKEN for admin-gated volk calls.
• Redis and graph: Redis host, port and database come from the core.redis_factory resolvers, which raise when unset; ring_buffer_max_len; the graph URI comes from KG_ROUTING_URI or KG_URI through _default_kg_uri, which returns an empty string when neither is set.
• Kernel link: the gRPC readiness timeout, the bounded connect attempts and backoff, and the background re-probe interval used by kernel_reprobe, plus the kernel reply timeout and backoff base.
• Heartbeat and sessions: the ping interval, timeout and maximum misses, and SESSION_ROOT.

__post_init__ rejects an auth_mode outside the valid set and a negative re-probe interval with ValueError. Helpers _env_int, _parse_csv and _parse_scope_map parse values. An insecure development JWT secret raises InsecureJWTSecretError unless the explicit dev opt-in is set.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

GatewayConfig (dataclass)

Decorators: dataclass(frozen=True) · Line 140

WebSocket gateway configuration.

Functions

Source

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

"""Gateway configuration from environment variables."""

import logging
import os
from collections.abc import Mapping
from dataclasses import dataclass, field

from orchestrata_ultimate.core.redis_factory import (
    resolve_redis_host,
    resolve_redis_port,
)
from orchestrata_ultimate.gateway.exceptions import InsecureJWTSecretError

logger = logging.getLogger(__name__)


def _env_int(key: str, default: str, environ: Mapping[str, str] | None = None) -> int:
    """Read an integer from an environment variable."""
    env = os.environ if environ is None else environ
    return int(env.get(key, default))


def _parse_csv(raw: str) -> tuple[str, ...]:
    """Split a comma-separated env value into a tuple, dropping blanks.

    Used by ``hmac_secrets`` (SPEC-014 FR-007, REQ-014-099: multiple shared
    secrets for key rotation) -- a plain env-var convenience parser, not a
    validated config file.
    """
    return tuple(item.strip() for item in raw.split(",") if item.strip())


def _parse_scope_map(raw: str) -> dict[str, str]:
    """Parse ``"client_id:scope,client_id:scope"`` into a dict.

    Used by ``hmac_client_scopes`` (SPEC-014 FR-007, REQ-014-100: HMAC auth
    returns scope from a client_id -> scope mapping). Malformed entries
    (missing ``:``, blank client_id/scope) are skipped rather than raising.
    """
    result: dict[str, str] = {}
    for pair in raw.split(","):
        pair = pair.strip()
        if not pair or ":" not in pair:
            continue
        client_id, _, scope = pair.partition(":")
        client_id = client_id.strip()
        scope = scope.strip()
        if client_id and scope:
            result[client_id] = scope
    return result


# Not a credential: the WELL-KNOWN dev default that _resolve_jwt_secret
# compares against to FAIL CLOSED (SPEC-099 P0, InsecureJWTSecretError).
# It must stay a source literal for that comparison to work -- encrypting
# it would disable the very check it exists to perform. Nothing signs with
# it outside a recognized dev/test context; the value is public by design.
# gate-ok: public fail-closed sentinel, not a secret (SPEC-099 P0)
_DEV_JWT_SECRET = "dev-secret"

# SPEC-014 FR-005: the three shipped auth strategies GatewayConfig.auth_mode
# selects between (default "jwt" -- every existing deployment/test keeps its
# current behaviour unless it opts in to "static" or "hmac").
_VALID_AUTH_MODES: frozenset[str] = frozenset({"jwt", "static", "hmac"})


def _dev_secret_allowed(environ: Mapping[str, str] | None = None) -> bool:
    """Whether the insecure default JWT secret may be used unmodified.

    Two explicit dev/test signals are recognized:
      - ``ALLOW_DEV_JWT_SECRET`` set truthy, or
      - running under pytest (``PYTEST_CURRENT_TEST``, set automatically by
        pytest for the duration of each test) -- the project's only
        established dev/test execution context.

    Resolved from *environ* exactly like the other helpers in this module:
    ``environ=None`` reads the real process environment; an explicit
    mapping is treated as the complete, hermetic environment, so tests can
    simulate a bare production environment with neither signal present.
    """
    env = os.environ if environ is None else environ
    if "PYTEST_CURRENT_TEST" in env:
        return True
    return env.get("ALLOW_DEV_JWT_SECRET", "").strip().lower() in ("1", "true", "yes")


def _resolve_jwt_secret(environ: Mapping[str, str] | None = None) -> str:
    """Resolve the JWT signing secret, fail-closed against the dev default.

    SPEC-099 (P0): a deploy that forgets to set ``JWT_SECRET`` must not
    silently accept the well-known ``"dev-secret"`` default -- anyone who
    knows that default could then forge valid tokens. The default is
    permitted only in a recognized dev/test context (see
    :func:`_dev_secret_allowed`); anywhere else, a missing/default secret
    is a startup error naming the fix.

    Raises:
        InsecureJWTSecretError: ``JWT_SECRET`` is unset (or explicitly
            ``"dev-secret"``) outside a recognized dev/test context.
    """
    env = os.environ if environ is None else environ
    secret = env.get("JWT_SECRET", _DEV_JWT_SECRET)
    if secret == _DEV_JWT_SECRET and not _dev_secret_allowed(environ):
        logger.error(
            "Refusing to start with the insecure default JWT secret outside "
            "a dev/test context."
        )
        raise InsecureJWTSecretError(
            "JWT_SECRET is not set (or is explicitly the insecure default "
            "'dev-secret'). Set the JWT_SECRET environment variable to a "
            "strong, unique secret before starting the gateway. For local "
            "development/testing only, set ALLOW_DEV_JWT_SECRET=1 to "
            "explicitly opt into the insecure default."
        )
    return secret


def _default_kg_uri(environ: Mapping[str, str] | None = None) -> str:
    """Resolve the operational-graph URI from the environment.

    Reads the canonical KG_ROUTING_URI / KG_URI seam. The NEO4J_URI /
    NEO4J_HOST / NEO4J_PORT keys this used to read were retired with the Neo4j
    containers on 2026-07-27 (migration plan Step 5.1), along with the
    `bolt://localhost:7687` assembled default — nothing in this infrastructure
    has ever listened on localhost (every graph is an Incus container on
    10.53.1.x), so that default could only ever turn a config gap into a
    connection-refused that reads like an outage.

    Returns the empty string when unset rather than raising: GatewayConfig is a
    frozen dataclass whose fields are built by default_factory, so raising here
    would make merely CONSTRUCTING a config fail. Callers connect through
    core.graph_factory.resolve_uri(), which raises with the key name at the
    point of use.
    """
    env = os.environ if environ is None else environ
    return env.get("KG_ROUTING_URI") or env.get("KG_URI") or ""


@dataclass(frozen=True)
class GatewayConfig:
    """WebSocket gateway configuration.

    All connection parameters read from environment
    variables with sensible localhost defaults.
    """

    # Optional explicit environment mapping for testing.
    # When provided, __post_init__ overrides all fields from this mapping.
    environ: Mapping[str, str] | None = None

    # Server
    host: str = field(
        default_factory=lambda: os.environ.get("GATEWAY_HOST", "localhost")
    )
    port: int = field(default_factory=lambda: _env_int("GATEWAY_PORT", "8765"))

    # HTTP surface (health/readiness/metrics) — SPEC-014 FR-021
    health_enabled: bool = field(
        default_factory=lambda: (
            os.environ.get("GATEWAY_HEALTH_ENABLED", "true").lower()
            not in ("0", "false", "no")
        )
    )
    http_port: int = field(
        default_factory=lambda: _env_int("GATEWAY_HTTP_PORT", "8766")
    )
    # Separate bind address for the HTTP surface, because /metrics has a
    # different reachability requirement from the WS API. In prod the gateway
    # binds 127.0.0.1, which is correct for the WS control plane and fatal for
    # monitoring: a Prometheus container on incusbr0 cannot scrape loopback.
    # Empty string (the default) means "inherit ``host``", so behaviour is
    # UNCHANGED unless this is set explicitly -- setting it exposes health/
    # readiness/metrics only, never the WS API.
    http_host: str = field(
        default_factory=lambda: os.environ.get("GATEWAY_HTTP_HOST", "")
    )

    @property
    def http_bind_host(self) -> str:
        """Address the HTTP surface binds to (``http_host`` or ``host``)."""
        return self.http_host or self.host

    # Auth
    jwt_secret: str = field(default_factory=lambda: _resolve_jwt_secret())
    jwt_algorithm: str = "HS256"
    token_expiry_seconds: int = field(
        default_factory=lambda: _env_int("TOKEN_EXPIRY_SECONDS", "3600")
    )

    # SPEC-014 FR-005: which shipped auth strategy the gateway validates
    # connections with -- "jwt" (default, unchanged behaviour), "static"
    # (FR-006 pre-shared token), or "hmac" (FR-007 HMAC-SHA256 request
    # signing). Validated in __post_init__.
    auth_mode: str = field(
        default_factory=lambda: os.environ.get("GATEWAY_AUTH_MODE", "jwt")
    )

    # FR-006 Static Token Auth: the config/env-seeded pre-shared token.
    # StaticTokenAuthenticator treats this as the INITIAL value only -- the
    # live value is Redis-backed and rotatable without a gateway restart
    # (REQ-014-094); this field just seeds it the first time.
    static_token: str | None = field(
        default_factory=lambda: os.environ.get("GATEWAY_TOKEN")
    )

    # FR-007 HMAC Auth: multiple shared secrets so an old and new secret can
    # both be live during rotation (REQ-014-099) -- comma-separated.
    hmac_secrets: tuple[str, ...] = field(
        default_factory=lambda: _parse_csv(os.environ.get("GATEWAY_HMAC_SECRETS", ""))
    )
    # Timestamp-drift window in seconds (REQ-014-097, default 60) -- also
    # doubles as the Redis TTL for nonce-replay tracking (REQ-014-098):
    # once a timestamp falls outside this window it is rejected regardless
    # of nonce state, so a nonce only needs to be remembered for exactly
    # this long.
    hmac_drift_seconds: int = field(
        default_factory=lambda: _env_int("GATEWAY_HMAC_DRIFT_SECONDS", "60")
    )
    # client_id -> scope mapping (REQ-014-100), e.g.
    # "client-a:admin,client-b:read_only". A client_id with no entry here
    # falls back to HmacAuthenticator.DEFAULT_SCOPE.
    hmac_client_scopes: Mapping[str, str] = field(
        default_factory=lambda: _parse_scope_map(
            os.environ.get("GATEWAY_HMAC_CLIENT_SCOPES", "")
        )
    )

    # Redis — no literal fallback (RUL-003): resolve_redis_host/port RAISE
    # when REDIS_HOST/REDIS_PORT is unset rather than silently defaulting to
    # localhost:6379.
    redis_host: str = field(default_factory=resolve_redis_host)
    redis_port: int = field(default_factory=resolve_redis_port)
    redis_db: int = field(default_factory=lambda: _env_int("REDIS_DB", "0"))
    ring_buffer_max_len: int = field(
        default_factory=lambda: _env_int("RING_BUFFER_MAX_LEN", "10000")
    )

    # Operational knowledge graph (orchgraph cluster).
    graph_uri: str = field(default_factory=lambda: _default_kg_uri())
    # No credentials: every remaining graph endpoint negotiates no auth.
    graph_user: str = ""
    graph_password: str = ""

    # Anonymous access
    allow_anonymous: bool = field(
        default_factory=lambda: (
            os.getenv("GATEWAY_ALLOW_ANONYMOUS", "false").lower() == "true"
        )
    )

    # Admin gating for volk.register/deregister.
    admin_token: str | None = field(
        default_factory=lambda: os.environ.get("GATEWAY_ADMIN_TOKEN")
    )

    # CXP gRPC kernel readiness probe timeout.
    grpc_ready_timeout_seconds: float = field(
        default_factory=lambda: float(
            os.environ.get("GATEWAY_GRPC_READY_TIMEOUT_S", "3.0")
        )
    )

    # Bounded retry for the gRPC kernel connect during _init_kernel: the
    # deployed kernel unit is Type=simple, so systemd's After= only orders
    # the gateway to start AFTER the kernel process is *spawned*, not after
    # it actually binds CXP_GRPC_PORT (measured ~2s spawn-to-bind gap,
    # 2026-08-12). A single probe loses that race and falls back to the
    # in-process PipelineKernelClient permanently, even though the kernel
    # comes up moments later. Retrying a bounded number of times with a
    # short backoff closes the race without risking an unbounded boot hang:
    # each attempt still honours grpc_ready_timeout_seconds (unchanged,
    # 2026-08-05 wedged-kernel guard) for its connect()/health_check()
    # asyncio.wait_for calls -- attempts/backoff only widen the window
    # across retries, they never loosen the per-attempt bound.
    grpc_connect_attempts: int = field(
        default_factory=lambda: _env_int("GATEWAY_GRPC_CONNECT_ATTEMPTS", "3")
    )
    grpc_retry_backoff_seconds: float = field(
        default_factory=lambda: float(
            os.environ.get("GATEWAY_GRPC_RETRY_BACKOFF_S", "0.5")
        )
    )

    # Background re-probe after the boot probe above gave up (2026-10-09:
    # all 44 prod gateway starts in 7 days fell back to the in-process
    # pipeline because the kernel bound its port after the ~1-2 s boot
    # window, and nothing ever retried). One bounded attempt per interval
    # until the kernel answers, then the gateway switches to grpc. 0
    # disables the re-probe; negative is rejected in __post_init__.
    grpc_reprobe_interval_seconds: float = field(
        default_factory=lambda: float(
            os.environ.get("GATEWAY_GRPC_REPROBE_INTERVAL_S", "15")
        )
    )

    # _consume_kernel_replies' streaming-pump window: the CXP
    # CXPReceiveRequest.timeout_seconds passed to _pump_kernel_replies_once.
    # int32 on the wire (see cxp.proto), so this stays a whole-second
    # granularity even for tests that shrink it.
    kernel_reply_timeout_s: int = field(
        default_factory=lambda: _env_int("KERNEL_REPLY_TIMEOUT_S", "5")
    )
    # _consume_kernel_replies' exponential-backoff base (backoff =
    # min(base * 2**consecutive_failures, 60)) -- pure in-process
    # asyncio.sleep, not wire-bound, so tests can shrink it arbitrarily.
    kernel_reply_backoff_base_s: float = field(
        default_factory=lambda: float(
            os.environ.get("KERNEL_REPLY_BACKOFF_BASE_S", "1.0")
        )
    )

    # Heartbeat (SPEC-014 FR-014) — server-initiated ping loop. Typed
    # ``int | float`` (not just ``int``): the env-parsed path
    # (``HEARTBEAT_INTERVAL``, via ``_env_int``) always yields a whole
    # number of seconds, but direct construction (tests exercising the ping
    # loop without an env round trip) needs sub-second intervals to stay
    # fast without literally sleeping the production 30s default.
    heartbeat_interval: int | float = field(
        default_factory=lambda: _env_int("HEARTBEAT_INTERVAL", "30")
    )
    heartbeat_timeout: int = field(
        default_factory=lambda: _env_int("HEARTBEAT_TIMEOUT", "90")
    )
    # REQ-014-108: consecutive missed pongs tolerated before the server
    # force-disconnects a connection. Distinct from heartbeat_timeout above
    # (an older, currently-unused field) -- the ping loop tracks misses as a
    # COUNT across heartbeat_interval-spaced ticks, not a single elapsed-time
    # budget, so it needs its own knob.
    heartbeat_max_misses: int = field(
        default_factory=lambda: _env_int("HEARTBEAT_MAX_MISSES", "3")
    )

    # Session-persistence root (fix/session-root, RUL-011 ports doctrine).
    # SessionHandler refuses to guess this from the process cwd or its own
    # package location -- it must be configured explicitly, here or via a
    # directly-injected constructor argument. ``None`` means "not
    # configured", never "use the checkout gateway happens to run from".
    session_root: str | None = field(
        default_factory=lambda: os.environ.get("SESSION_ROOT")
    )

    def __post_init__(self) -> None:
        if self.environ is not None:
            self._override_from_environ(self.environ)
        if self.auth_mode not in _VALID_AUTH_MODES:
            raise ValueError(
                f"GATEWAY_AUTH_MODE must be one of {sorted(_VALID_AUTH_MODES)!r}, "
                f"got {self.auth_mode!r}"
            )
        if self.grpc_reprobe_interval_seconds < 0:
            raise ValueError(
                "GATEWAY_GRPC_REPROBE_INTERVAL_S must be >= 0 (0 disables), "
                f"got {self.grpc_reprobe_interval_seconds!r}"
            )

    def _override_from_environ(self, env: Mapping[str, str]) -> None:
        """Override every field from an explicit environment mapping.

        Split out of ``__post_init__`` so auth_mode validation (below) runs
        unconditionally -- including for the ``environ=None`` (real-process-
        env) construction path -- rather than only when an explicit environ
        override is supplied.
        """
        object.__setattr__(self, "host", env.get("GATEWAY_HOST", "localhost"))
        object.__setattr__(self, "port", _env_int("GATEWAY_PORT", "8765", environ=env))
        object.__setattr__(
            self,
            "health_enabled",
            env.get("GATEWAY_HEALTH_ENABLED", "true").lower()
            not in ("0", "false", "no"),
        )
        object.__setattr__(
            self, "http_port", _env_int("GATEWAY_HTTP_PORT", "8766", environ=env)
        )
        object.__setattr__(self, "http_host", env.get("GATEWAY_HTTP_HOST", ""))
        object.__setattr__(self, "jwt_secret", _resolve_jwt_secret(environ=env))
        object.__setattr__(
            self,
            "token_expiry_seconds",
            _env_int("TOKEN_EXPIRY_SECONDS", "3600", environ=env),
        )
        object.__setattr__(self, "auth_mode", env.get("GATEWAY_AUTH_MODE", "jwt"))
        object.__setattr__(self, "static_token", env.get("GATEWAY_TOKEN"))
        object.__setattr__(
            self, "hmac_secrets", _parse_csv(env.get("GATEWAY_HMAC_SECRETS", ""))
        )
        object.__setattr__(
            self,
            "hmac_drift_seconds",
            _env_int("GATEWAY_HMAC_DRIFT_SECONDS", "60", environ=env),
        )
        object.__setattr__(
            self,
            "hmac_client_scopes",
            _parse_scope_map(env.get("GATEWAY_HMAC_CLIENT_SCOPES", "")),
        )
        object.__setattr__(self, "redis_host", resolve_redis_host(env))
        object.__setattr__(self, "redis_port", resolve_redis_port(env))
        object.__setattr__(self, "redis_db", _env_int("REDIS_DB", "0", environ=env))
        object.__setattr__(
            self,
            "ring_buffer_max_len",
            _env_int("RING_BUFFER_MAX_LEN", "10000", environ=env),
        )
        object.__setattr__(self, "graph_uri", _default_kg_uri(environ=env))
        object.__setattr__(
            self,
            "allow_anonymous",
            env.get("GATEWAY_ALLOW_ANONYMOUS", "false").lower() == "true",
        )
        object.__setattr__(self, "admin_token", env.get("GATEWAY_ADMIN_TOKEN"))
        object.__setattr__(
            self,
            "grpc_ready_timeout_seconds",
            float(env.get("GATEWAY_GRPC_READY_TIMEOUT_S", "3.0")),
        )
        object.__setattr__(
            self,
            "grpc_connect_attempts",
            _env_int("GATEWAY_GRPC_CONNECT_ATTEMPTS", "3", environ=env),
        )
        object.__setattr__(
            self,
            "grpc_retry_backoff_seconds",
            float(env.get("GATEWAY_GRPC_RETRY_BACKOFF_S", "0.5")),
        )
        object.__setattr__(
            self,
            "grpc_reprobe_interval_seconds",
            float(env.get("GATEWAY_GRPC_REPROBE_INTERVAL_S", "15")),
        )
        object.__setattr__(
            self,
            "kernel_reply_timeout_s",
            _env_int("KERNEL_REPLY_TIMEOUT_S", "5", environ=env),
        )
        object.__setattr__(
            self,
            "kernel_reply_backoff_base_s",
            float(env.get("KERNEL_REPLY_BACKOFF_BASE_S", "1.0")),
        )
        object.__setattr__(
            self,
            "heartbeat_interval",
            _env_int("HEARTBEAT_INTERVAL", "30", environ=env),
        )
        object.__setattr__(
            self, "heartbeat_timeout", _env_int("HEARTBEAT_TIMEOUT", "90", environ=env)
        )
        object.__setattr__(
            self,
            "heartbeat_max_misses",
            _env_int("HEARTBEAT_MAX_MISSES", "3", environ=env),
        )
        object.__setattr__(self, "session_root", env.get("SESSION_ROOT"))