symfonic.agent.authority

authority

Authenticated principal authority binding shared by native and legacy hosts.

AdminAuthority dataclass

AdminAuthority(principal_id: str, is_admin: bool = False)

An admin claim a host derived, and the principal it derived it for.

The adapter value for a deployment whose authentication is the legacy request-stashing verifier rather than a platform :class:~symfonic.platform.ports.ScopeResolver. A resolver's :class:~symfonic.platform.values.AuthenticatedPrincipal already satisfies what :func:bind_admin_authority reads, so it is bound directly and this class is not in its way.

principal_id is required and not defaulted. That is the type doing the contract's work: the legacy keyword's defect was that True said nothing about who, and a claim value with an optional identity would have reintroduced the same anonymous boolean at a longer address.

UnauthenticatedAdminClaimError

Bases: CutoverPathError

An admin claim was bound without naming the principal it belongs to.

Subclasses :class:~symfonic.services.shadow.errors.CutoverPathError, and through it ConfigurationError and SymfonicError, for the reason every other error in this package gives for its ancestry: an adopter's existing except around the agent API keeps catching this. A widening, never a rename.

Named separately from :class:SelfAssertedAuthorityError because the two are different incidents. That one is "you asked for a keyword this line retired"; this one is "you used the replacement and it is not carrying an identity", which is a host wiring its own authentication wrongly and is the failure that would otherwise let the replacement decay back into the thing it replaced.

admin_authority

admin_authority() -> bool

The bound principal's admin bit, or False when nothing is bound.

The single place the engine asks "is this turn an administrator's?", so run, stream and stream_typed cannot answer it differently. It is read after :func:refuse_self_asserted_admin has run, so the keyword's value can never contribute: the two are one decision expressed as a refusal and a derivation rather than as a precedence rule, because a precedence rule between an authenticated claim and a self-asserted one is a rule that can be got backwards.

Source code in src/symfonic/agent/authority.py

def admin_authority() -> bool:
    """The bound principal's admin bit, or ``False`` when nothing is bound.

    The single place the engine asks "is this turn an administrator's?", so
    ``run``, ``stream`` and ``stream_typed`` cannot answer it differently. It is
    read *after* :func:`refuse_self_asserted_admin` has run, so the keyword's
    value can never contribute: the two are one decision expressed as a refusal
    and a derivation rather than as a precedence rule, because a precedence rule
    between an authenticated claim and a self-asserted one is a rule that can be
    got backwards.
    """
    return _ADMIN_AUTHORITY.get()

bind_admin_authority

bind_admin_authority(claim: Any) -> Iterator[bool]

Bind an authenticated principal's admin bit for the turns inside.

claim is anything carrying principal_id and is_admin -- :class:~symfonic.platform.values.AuthenticatedPrincipal structurally, or :class:AdminAuthority for a host whose verifier stashes facts on the request. None is not an error -- a host that authenticated nobody says so by having nothing to bind, and forcing it to construct an anonymous claim would be forcing it to write the sentence this contract exists to make unsayable -- and it binds False rather than binding nothing. That difference only shows inside an outer binding, and it is the point: None is what InvocationRequest.admin_claim returns for a non-admin request, so a host reusing that spelling to scope a nested turn down would otherwise keep the outer administrator's authority while this function handed it False. The value yielded and the value :func:admin_authority reads are one value in every case, which is the only version of this contract that cannot be read backwards.

Refuses an administrator claim with no principal_id. A non-admin claim with no id is merely uninformative and binds False, which is what it would have meant anyway; an admin claim with no id is the anonymous boolean wearing the replacement's clothes, and it is the one shape this function exists to reject.

A context manager rather than a parameter, because the value has to survive into an async generator the host returns and the caller drives later -- stream and stream_typed are consumed after their handler has returned. The reset tolerates a token minted in another context rather than raising: a generator driven from one task and closed from another would otherwise turn a tidy-up into the turn's exception, and that closing context never held this binding at all -- see the comment on the branch.

Source code in src/symfonic/agent/authority.py

@contextmanager
def bind_admin_authority(claim: Any) -> Iterator[bool]:
    """Bind an authenticated principal's admin bit for the turns inside.

    ``claim`` is anything carrying ``principal_id`` and ``is_admin`` --
    :class:`~symfonic.platform.values.AuthenticatedPrincipal` structurally, or
    :class:`AdminAuthority` for a host whose verifier stashes facts on the
    request. ``None`` is not an error -- a host that authenticated nobody says
    so by having nothing to bind, and forcing it to construct an anonymous
    claim would be forcing it to write the sentence this contract exists to
    make unsayable -- and it binds ``False`` rather than binding nothing. That
    difference only shows inside an outer binding, and it is the point:
    ``None`` is what ``InvocationRequest.admin_claim`` returns for a non-admin
    request, so a host reusing that spelling to scope a nested turn *down*
    would otherwise keep the outer administrator's authority while this
    function handed it ``False``. The value yielded and the value
    :func:`admin_authority` reads are one value in every case, which is the
    only version of this contract that cannot be read backwards.

    Refuses an *administrator* claim with no ``principal_id``. A non-admin claim
    with no id is merely uninformative and binds ``False``, which is what it
    would have meant anyway; an admin claim with no id is the anonymous boolean
    wearing the replacement's clothes, and it is the one shape this function
    exists to reject.

    A context manager rather than a parameter, because the value has to survive
    into an async generator the host returns and the caller drives later --
    ``stream`` and ``stream_typed`` are consumed after their handler has
    returned. The reset tolerates a token minted in another context rather than
    raising: a generator driven from one task and closed from another would
    otherwise turn a tidy-up into the turn's exception, and that closing
    context never held this binding at all -- see the comment on the branch.
    """
    if claim is None:
        is_admin = False
        principal_id = ""
    else:
        is_admin = bool(getattr(claim, "is_admin", False))
        principal_id = str(getattr(claim, "principal_id", "") or "").strip()
    if is_admin and not principal_id:
        raise UnauthenticatedAdminClaimError(
            "an admin claim must name the principal it was derived for: "
            f"{type(claim).__name__} carries is_admin=True and no "
            "principal_id. On the "
            f"{LEVER_RETIREMENT_LINE} line the admin bypass is a fact about an "
            "authenticated principal, so an anonymous claim is refused rather "
            "than honoured -- it is the retired is_admin=True keyword under a "
            "new name, and honouring it here would undo the replacement it is "
            "part of."
        )
    token = _ADMIN_AUTHORITY.set(is_admin)
    try:
        yield is_admin
    finally:
        # ``ValueError`` means the token was minted in another context -- the
        # shape ``_bound`` takes when sse-starlette drives the frames in one
        # task and closes the generator from another. Suppressed rather than
        # compensated for: the ``set`` landed in the *driving* task's context
        # and this one was never touched, so writing anything here (``False``,
        # or the value seen where the bind was entered, which in this shape is
        # the driving task's and not this context's) cannot reach the context
        # holding the binding and WOULD overwrite an outer binding held by the
        # closing task -- a silent de-escalation of somebody else's turn,
        # dressed as tidy-up. The driving task's copy dies with that task.
        with suppress(ValueError):
            _ADMIN_AUTHORITY.reset(token)