/surfsense_backend/app/gateway/auth_invariant.py

"""Authorization invariants for external-chat-routed turns."""

from __future__ import annotations

from fastapi import HTTPException
from sqlalchemy.ext.asyncio import AsyncSession

from app.auth.context import AuthContext
from app.db import ExternalChatBinding, Permission, User
from app.gateway.bindings import suspend_binding
from app.observability.metrics import record_gateway_auth_invariant_failure
from app.utils.rbac import check_permission, check_workspace_access


class GatewaySuspendedError(RuntimeError):
    def __init__(self, reason: str) -> None:
        self.reason = reason
        super().__init__(reason)


async def _fail(
    session: AsyncSession,
    binding: ExternalChatBinding,
    reason: str,
) -> None:
    suspend_binding(binding, reason)
    record_gateway_auth_invariant_failure(cause=reason)
    await session.flush()
    raise GatewaySuspendedError(reason)


async def assert_authorization_invariant(
    session: AsyncSession,
    binding: ExternalChatBinding,
) -> User:
    if binding.state != "bound":
        await _fail(session, binding, "binding_not_bound")

    user = await session.get(User, binding.user_id)
    if user is None:
        await _fail(session, binding, "owner_missing")

    auth = AuthContext.system(user, source="gateway")

    try:
        await check_workspace_access(session, auth, binding.workspace_id)
        await check_permission(
            session,
            auth,
            binding.workspace_id,
            Permission.CHATS_CREATE.value,
            "External chat owner no longer has permission to chat in this workspace",
        )
    except HTTPException as exc:
        await _fail(session, binding, f"rbac_{exc.status_code}")

    return user