Skip to content

symfonic.agent.fastapi.authgate

authgate

May this deployment serve at all — the production auth gate (Sprint B, v5.6.1).

Split out of :mod:.dependencies when that module passed its budget, and the seam is the honest one: "may this deployment serve?" is a different question from "what scope is this request?", asked at a different moment, with a different answer type.

The rule the whole module exists to hold: the object that serves requests is the object that must satisfy the check. It has been broken twice, both times by validating something adjacent to what dispatch uses —

  • the gate read the module global while get_tenant_scope preferred an injected resolver, so a stale global excused an untrusted resolver;
  • the gate read the app passed to create_agent_router while dispatch reads request.app, so an app validated at construction excused a different app mounted afterwards.

Both were reachable, and neither logged anything. That is why enforcement now happens at both moments against the same function: construction keeps the fail-fast, and the request-time call is what closes the gap between the app that was checked and the app that answers.

enforce_production_auth

enforce_production_auth(resolver: Any, legacy_verifier: Any) -> None

Raise RuntimeError if production is detected and auth is not wired.

Parameters:

Name Type Description Default
resolver Any

the scope resolver belonging to the app in question, or None when it has none (or when no app is known).

required
legacy_verifier Any

the module-global verifier, or None.

required

The precedence mirrors get_tenant_scope: an injected resolver wins, so a resolver that is present but does not attest is a refusal even when a global is registered — no request would reach that global.

Source code in src/symfonic/agent/fastapi/authgate.py
def enforce_production_auth(resolver: Any, legacy_verifier: Any) -> None:
    """Raise ``RuntimeError`` if production is detected and auth is not wired.

    Args:
        resolver: the scope resolver belonging to the app in question, or
            ``None`` when it has none (or when no app is known).
        legacy_verifier: the module-global verifier, or ``None``.

    The precedence mirrors ``get_tenant_scope``: an injected resolver wins, so
    a resolver that is present but does not attest is a refusal *even when a
    global is registered* — no request would reach that global.
    """
    if not is_production_env():
        return
    if os.environ.get("ALLOW_INSECURE_PROD", "").strip().lower() == "true":
        logger.critical(
            "ALLOW_INSECURE_PROD=true detected — tenant auth disabled "
            "in production. Data WILL leak across tenants. Set this to "
            "false and wire auth immediately: install_scope_resolver(app, "
            "HeaderScopeResolver(verifier=...)) with create_agent_router("
            "agent, app=app), or the legacy set_tenant_auth_verifier().",
        )
        return
    if resolver is not None:
        if resolver_satisfies_gate(resolver):
            return
        raise RuntimeError(
            "Production environment detected and this app's injected scope "
            "resolver does not attest to a verifier. It -- not any global -- "
            "is what serves requests, because get_tenant_scope prefers an "
            "injected resolver. Construct it with "
            "HeaderScopeResolver(verifier=...), or implement posture() "
            "returning verifier_registered=True on a custom resolver. A "
            "registered set_tenant_auth_verifier() does NOT excuse this: no "
            "request would reach it. Or set ALLOW_INSECURE_PROD=true to bypass "
            "(NOT RECOMMENDED).",
        )
    if legacy_verifier is not None:
        return
    raise RuntimeError(
        "Production environment detected but no tenant auth is wired. "
        "Either inject a resolver -- symfonic.agent.fastapi.dependencies."
        "install_scope_resolver(app, HeaderScopeResolver(verifier=...)) and "
        "pass create_agent_router(agent, app=app) so this check can see it -- "
        "or register the legacy global with set_tenant_auth_verifier(verifier) "
        "before create_agent_router(). Or set ALLOW_INSECURE_PROD=true to "
        "bypass (NOT RECOMMENDED).",
    )

is_production_env

is_production_env() -> bool

Check common env-var conventions for production.

Any of SYMFONIC_ENV, APP_ENV, ENVIRONMENT, NODE_ENV with the value production (or prod, case-insensitive) flips the bit.

Source code in src/symfonic/agent/fastapi/authgate.py
def is_production_env() -> bool:
    """Check common env-var conventions for production.

    Any of ``SYMFONIC_ENV``, ``APP_ENV``, ``ENVIRONMENT``, ``NODE_ENV`` with
    the value ``production`` (or ``prod``, case-insensitive) flips the bit.
    """
    return any(
        os.environ.get(var, "").strip().lower() in _PROD_VALUES
        for var in _PROD_ENV_VARS
    )

resolver_satisfies_gate

resolver_satisfies_gate(resolver: Any) -> bool

Whether an injected resolver counts as authentication being wired.

Presence is not enough, and the difference is the whole point. HeaderScopeResolver(verifier=None) accepts X-Tenant-ID on trust and reports exactly that in posture().verifier_registered; if merely having a resolver satisfied the gate, migrating off the global would be a way to turn tenant isolation off while startup reported success.

Attestation is required, not assumed. A resolver with no posture(), one whose posture() raises, and one whose reading says nothing about a verifier are all refusals. An earlier version accepted them, reasoning that refusing what we cannot introspect would discourage custom resolvers; in production that is backwards. "I could not verify this" is not evidence of authentication, and a gate that accepts an unreadable answer is not fail-closed whatever its docstring says.

The cost is one method on a custom resolver. The alternative is a gate that approves anything with the right shape.

Source code in src/symfonic/agent/fastapi/authgate.py
def resolver_satisfies_gate(resolver: Any) -> bool:
    """Whether an injected resolver counts as authentication being wired.

    Presence is not enough, and the difference is the whole point.
    ``HeaderScopeResolver(verifier=None)`` accepts ``X-Tenant-ID`` on trust and
    reports exactly that in ``posture().verifier_registered``; if merely having
    a resolver satisfied the gate, migrating off the global would be a way to
    turn tenant isolation off while startup reported success.

    **Attestation is required, not assumed.** A resolver with no ``posture()``,
    one whose ``posture()`` raises, and one whose reading says nothing about a
    verifier are all refusals. An earlier version accepted them, reasoning that
    refusing what we cannot introspect would discourage custom resolvers; in
    production that is backwards. "I could not verify this" is not evidence of
    authentication, and a gate that accepts an unreadable answer is not
    fail-closed whatever its docstring says.

    The cost is one method on a custom resolver. The alternative is a gate that
    approves anything with the right shape.
    """
    posture = getattr(resolver, "posture", None)
    if not callable(posture):
        return False
    try:
        reading = posture()
    except Exception:  # noqa: BLE001 - an unreadable posture attests nothing
        return False
    return bool(getattr(reading, "verifier_registered", False))