Guards¶

def has_all_permissions[PermissionNameT: str](*permissions: PermissionNameT) -> Guard:
    """Return a guard that requires every configured permission.

    Explicit-name alias of :func:`has_permission` that reads symmetrically next to
    :func:`has_any_permission`; both apply identical conjunctive semantics.
    """
    return _build_permission_guard(
        required_permissions=_normalize_required_permissions(permissions),
        require_all=True,
        guard_name="has_all_permissions",
    )

def has_all_roles[RoleNameT: str](*roles: RoleNameT) -> Guard:
    """Return a guard that requires all configured normalized roles.

    The returned guard first enforces the same authenticated-active user contract as
    :func:`is_active`, then compares the normalized flat role membership exposed by
    the authenticated user against the configured roles. Both sides use the same
    trim/lowercase/deduplicate/sort semantics as the model and manager layers.

    Args:
        *roles: Required role names. At least one role must be provided. Each value is
            normalized with trim/lowercase/deduplicate/sort semantics, and empty or
            whitespace-only role names are rejected at guard-build time.

    Returns:
        Litestar-compatible guard callable for route ``guards=[...]`` lists.
    """
    return _build_role_guard(
        required_roles=_normalize_required_roles(roles),
        require_all=True,
        guard_name="has_all_roles",
    )

def has_any_permission[PermissionNameT: str](*permissions: PermissionNameT) -> Guard:
    """Return a guard that requires at least one configured permission."""
    return _build_permission_guard(
        required_permissions=_normalize_required_permissions(permissions),
        require_all=False,
        guard_name="has_any_permission",
    )

def has_any_role[RoleNameT: str](*roles: RoleNameT) -> Guard:
    """Return a guard that requires any configured normalized role.

    The returned guard first enforces the same authenticated-active user contract as
    :func:`is_active`, then compares the normalized flat role membership exposed by
    the authenticated user against the configured roles. Both sides use the same
    trim/lowercase/deduplicate/sort semantics as the model and manager layers.

    Args:
        *roles: Required role names. At least one role must be provided. Each value is
            normalized with trim/lowercase/deduplicate/sort semantics, and empty or
            whitespace-only role names are rejected at guard-build time.

    Returns:
        Litestar-compatible guard callable for route ``guards=[...]`` lists.
    """
    return _build_role_guard(
        required_roles=_normalize_required_roles(roles),
        require_all=False,
        guard_name="has_any_role",
    )

def has_any_scope[ScopeNameT: str](*scopes: ScopeNameT) -> Guard:
    """Return a guard requiring an API key to have any configured scope."""
    return _build_scope_guard(required_scopes=_normalize_required_scopes(scopes), require_all=False)

def has_organization_permission[OrganizationPermissionNameT: str](
    *permissions: OrganizationPermissionNameT,
) -> Guard:
    """Return a guard that requires organization-scoped effective permissions."""
    required_permissions = _normalize_required_permissions(permissions)
    required_permission_set = frozenset(required_permissions)

    def _guard(
        connection: ASGIConnection[Any, Any, Any, Any],
        _handler: BaseRouteHandler,
    ) -> None:
        guarded = _require_active_guarded_user(connection, guard_name="has_organization_permission")
        _require_role_capable_user(guarded, guard_name="has_organization_permission")
        if resolve_current_organization_roles(connection) is None:
            raise InsufficientOrganizationPermissionsError(
                required_permissions=required_permission_set,
                granted_permissions=frozenset(),
                require_all=True,
            )

        granted_permissions = resolve_connection_permissions(connection)
        if _permissions_include_all(granted_permissions, api_key_delegation_scopes(connection), required_permissions):
            return
        raise InsufficientOrganizationPermissionsError(
            required_permissions=required_permission_set,
            granted_permissions=granted_permissions,
            require_all=True,
        )

    return _guard

def has_organization_role[OrganizationRoleNameT: str](*roles: OrganizationRoleNameT) -> Guard:
    """Return a guard that requires organization-scoped membership roles."""
    required_roles = _normalize_required_roles(roles)
    required_role_set = frozenset(required_roles)

    def _guard(
        connection: ASGIConnection[Any, Any, Any, Any],
        _handler: BaseRouteHandler,
    ) -> None:
        _require_active_guarded_user(connection, guard_name="has_organization_role")
        organization_roles = resolve_current_organization_roles(connection)
        if organization_roles is not None and _roles_include_all_fixed_work(organization_roles, required_roles):
            return
        raise InsufficientOrganizationRolesError(
            required_roles=required_role_set,
            user_roles=organization_roles or frozenset(),
            require_all=True,
        )

    return _guard

def has_permission[PermissionNameT: str](*permissions: PermissionNameT) -> Guard:
    """Return a guard that requires every configured permission.

    Conjunctive permission check: the request is allowed only when the user's
    effective permissions satisfy every argument. :func:`has_all_permissions` is
    the explicit-name alias provided for symmetry with :func:`has_any_permission`.
    """
    return _build_permission_guard(
        required_permissions=_normalize_required_permissions(permissions),
        require_all=True,
        guard_name="has_permission",
    )

def has_scope[ScopeNameT: str](*scopes: ScopeNameT) -> Guard:
    """Return a guard requiring an API key to have every configured scope."""
    return _build_scope_guard(required_scopes=_normalize_required_scopes(scopes), require_all=True)

def is_active(
    connection: ASGIConnection[Any, Any, Any, Any],
    _handler: BaseRouteHandler,
) -> None:
    """Ensure the authenticated user is active.

    Args:
        connection: Incoming ASGI connection (Litestar request scope).
        _handler: Route handler being guarded; unused but required by Litestar guard signature.

    Note:
        Delegates to :func:`_require_active_guarded_user` for exceptions.
    """
    _require_active_guarded_user(connection, guard_name="is_active")

def is_authenticated(
    connection: ASGIConnection[Any, Any, Any, Any],
    _handler: BaseRouteHandler,
) -> None:
    """Ensure the request has an authenticated user.

    Args:
        connection: Incoming ASGI connection (Litestar request scope).
        _handler: Route handler being guarded; unused but required by Litestar guard signature.

    Raises:
        NotAuthorizedException: Raised when no authenticated user is attached to the connection.
    """
    if connection.user is not None:
        return

    msg = "Authentication credentials were not provided."
    raise NotAuthorizedException(detail=msg)

def is_superuser(
    connection: ASGIConnection[Any, Any, Any, Any],
    _handler: BaseRouteHandler,
) -> None:
    """Ensure the authenticated user has superuser privileges.

    Args:
        connection: Incoming ASGI connection (Litestar request scope).
        _handler: Route handler being guarded; unused but required by Litestar guard signature.

    Raises:
        PermissionDeniedException: Raised when the user lacks superuser privileges.

    Note:
        :func:`_require_active_guarded_user` may raise :exc:`NotAuthorizedException` when no user is
        present, or :exc:`PermissionDeniedException` for missing account state or inactive users.
    """
    guarded = _require_active_guarded_user(connection, guard_name="is_superuser")
    user_roles = _normalized_user_roles(guarded, guard_name="is_superuser")
    superuser_role_name = read_scope_superuser_role_name(connection)
    if _roles_intersect_fixed_work(user_roles, (superuser_role_name,)):
        return

    msg = "The authenticated user does not have sufficient privileges."
    raise PermissionDeniedException(detail=msg)

def is_verified(
    connection: ASGIConnection[Any, Any, Any, Any],
    _handler: BaseRouteHandler,
) -> None:
    """Ensure the authenticated user has a verified account.

    Args:
        connection: Incoming ASGI connection (Litestar request scope).
        _handler: Route handler being guarded; unused but required by Litestar guard signature.

    Raises:
        PermissionDeniedException: Raised when the user is unverified.

    Note:
        :func:`_require_active_guarded_user` may raise :exc:`NotAuthorizedException` when no user is
        present, or :exc:`PermissionDeniedException` for missing account state or inactive users.
    """
    guarded = _require_active_guarded_user(connection, guard_name="is_verified")
    if guarded.is_verified:
        return

    msg = "The authenticated user is not verified."
    raise PermissionDeniedException(detail=msg)

def requires_api_key(
    connection: ASGIConnection[Any, Any, Any, Any],
    _handler: BaseRouteHandler,
) -> None:
    """Require authentication through the API-key backend."""
    if _api_key_context(connection) is not None:
        return
    _raise_authorization_denied(_API_KEY_REQUIRED_DETAIL)

def requires_organization_membership(
    connection: ASGIConnection[Any, Any, Any, Any],
    _handler: BaseRouteHandler,
) -> None:
    """Require the authenticated active user to have a verified current organization."""
    _require_active_guarded_user(connection, guard_name="requires_organization_membership")
    if read_scope_current_organization_context(connection) is not None:
        return
    _raise_organization_membership_denied()

def requires_password_session(
    connection: ASGIConnection[Any, Any, Any, Any],
    _handler: BaseRouteHandler,
) -> None:
    """Reject API-key-authenticated callers from password-session-only routes."""
    if _api_key_context(connection) is None:
        return
    _raise_authorization_denied(_SESSION_AUTH_REQUIRED_DETAIL)