Source code for loom.core.sql.roles

"""Identity-bound role resolution of the SQL pillar.

The effective roles of a query are derived from the VERIFIED identity the
authentication middleware published, never from the request body.  The body may
only narrow the resulting set.  Resolution is fail-closed: an anonymous caller
or one holding no allowlisted role is denied, and ``default_role`` is never a
fallback once a connection binds roles to identity.

This module knows nothing about tokens or claims: whichever mechanism
authenticated the caller, it sees the same
:class:`~loom.core.identity.identity.Identity`.

Public API of the SQL pillar: any layer that executes caller-scoped queries
(the REST endpoint, the agent compiler's SQL capability at run time) resolves
the effective roles through this single function.
"""

from __future__ import annotations

import logging
from collections.abc import Sequence
from typing import NoReturn

from loom.core.identity import Identity
from loom.core.sql.abc import RoleNotAllowedError, RolesNotBoundError

_logger = logging.getLogger(__name__)




def resolve_query_roles(
    identity: Identity,
    *,
    connection: str,
    roles_bound: bool,
    allowed_roles: frozenset[str],
    requested_roles: Sequence[str] | None,
) -> tuple[str, ...]:
    """Resolve the roles one query may use.

    Args:
        identity: Verified caller published by the authentication middleware.
        connection: Name of the SQL connection being queried.
        roles_bound: Whether the configured authentication mechanism binds
            roles to the identity.  ``False`` means the connection declares no
            binding and is single-role by config.
        allowed_roles: Connection allowlist, the ceiling of the intersection.
        requested_roles: Roles asked for in the body; they may only narrow.

    Returns:
        The effective roles for this single request.

    Raises:
        RolesNotBoundError: When no allowed role can be derived from the
            verified identity.
        RoleNotAllowedError: When the body asks for a role the identity does
            not hold.
    """
    if not roles_bound:
        # No binding for this connection: its allowlist is empty — the binder
        # guard refuses to mount a non-empty one without a binding — so the
        # service rejects every caller-supplied role and applies default_role.
        return tuple(requested_roles or ())

    authorized = _authorized_roles(identity, connection=connection, allowed_roles=allowed_roles)
    return _narrow(authorized, requested_roles, connection)



def _authorized_roles(
    identity: Identity,
    *,
    connection: str,
    allowed_roles: frozenset[str],
) -> tuple[str, ...]:
    """Intersect the roles the identity holds with the allowlist, fail-closed."""
    if not identity.is_authenticated:
        _deny(connection, identity, "the request carries no verified identity")
    if not identity.roles:
        _deny(connection, identity, "the verified identity carries no role")
    authorized = tuple(role for role in identity.roles if role in allowed_roles)
    if not authorized:
        _deny(connection, identity, "no role held by the identity is allowlisted")
    return authorized


def _narrow(
    authorized: tuple[str, ...],
    requested: Sequence[str] | None,
    connection: str,
) -> tuple[str, ...]:
    """Apply the body narrowing: a subset of *authorized*, never a widening."""
    if not requested:
        return authorized
    for role in requested:
        if role not in authorized:
            raise RoleNotAllowedError(role, connection=connection)
    return tuple(dict.fromkeys(requested))


def _deny(connection: str, identity: Identity, reason: str) -> NoReturn:
    """Log the audit trail of a denial and refuse the request.

    The response message stays generic on purpose (no oracle about which part
    of the credentials failed); the precise reason is only recorded server-side.
    """
    _logger.warning(
        "SQL role authorization denied: connection=%s subject=%s mechanism=%s reason=%s",
        connection,
        identity.subject,
        identity.mechanism,
        reason,
    )
    raise RolesNotBoundError(connection)