Authorization

class Action(str, Enum):
    """The built-in actions SQLAdmin asks about.

    Members are plain strings as well, so ``Action.EDIT == "edit"`` and a grant
    stored as ``("user", "edit")`` matches ``("user", Action.EDIT)`` -- use
    whichever reads better.

    ???+ usage
        ```python
        from sqladmin.authorization import Action

        grants = {("user", Action.LIST), ("user", Action.EDIT)}
        ```

    Custom actions declared with [`@action`][sqladmin.application.action] are
    asked about as ``"action:<slug>"`` -- see
    [`custom_action`][sqladmin.authorization.custom_action].
    """

    LIST = "list"
    DETAILS = "details"
    CREATE = "create"
    EDIT = "edit"
    DELETE = "delete"
    EXPORT = "export"
    IMPORT = "import"

    # ``enum.StrEnum`` needs Python 3.11. Without this, Python 3.12+ formats a
    # ``(str, Enum)`` member as ``"Action.EDIT"`` in f-strings.
    __str__ = str.__str__

class AuthorizationBackend(ABC):
    """Base class for deciding *what* the current user may do.

    Where [`AuthenticationBackend`][sqladmin.authentication.AuthenticationBackend]
    answers "who is this request", this answers "may they do this". Subclass
    it and implement
    [`has_permission`][sqladmin.authorization.AuthorizationBackend.has_permission],
    then pass an instance as ``Admin(authorization_backend=...)``.

    ???+ usage
        ```python
        class RoleAuthorization(AuthorizationBackend):
            def has_permission(self, request, identity, action, obj=None):
                role = request.session.get("role")
                return role == "admin" or action in ("list", "details")


        admin = Admin(app, engine, authorization_backend=RoleAuthorization())
        ```

    Without one, `Admin` uses
    [`AllowAllAuthorizationBackend`][sqladmin.authorization.AllowAllAuthorizationBackend].
    """

    def setup(self, admin: BaseAdmin) -> None:
        """Check that ``admin`` is configured the way this backend needs.

        Called once, when the `Admin` is created. Raise
        [`ImproperlyConfigured`][sqladmin.exceptions.ImproperlyConfigured] to
        report a mistake at startup rather than on the first request.

        Does nothing by default.
        """

    async def load(self, request: Request) -> None:
        """Prepare per-request authorization state.

        Called once per request, right after authentication succeeds and before
        any permission is checked. This is where I/O belongs -- query the
        database, call an external service -- storing the result on
        ``request.state`` for `has_permission` to read.

        Does nothing by default.
        """

    @abstractmethod
    def has_permission(
        self,
        request: Request,
        identity: str,
        action: str,
        obj: Any | None = None,
    ) -> bool:
        """Return whether the current user may perform ``action`` on ``identity``.

        Args:
            request: The current request.
            identity: The `ModelView.identity` (or `BaseView.identity`) being
                acted on.
            action: One of [`Action`][sqladmin.authorization.Action], or
                ``"action:<slug>"`` for a custom `@action` endpoint.
            obj: The specific object being acted on, when there is one. Passed
                for ``details``, ``edit`` and ``delete``; ``None`` everywhere
                else, including the row-less buttons that lead to those pages.

        This method is called many times while rendering a single page -- once
        per row on the list page -- so it must be cheap and must not perform
        I/O. Do the lookups in
        [`load`][sqladmin.authorization.AuthorizationBackend.load].
        """

class AllowAllAuthorizationBackend(AuthorizationBackend):
    """The backend `Admin` uses when none is configured: it allows everything.

    With it, only the ``can_*`` flags and your own ``is_accessible`` /
    ``check_can_*`` overrides decide what users may do.
    """

    def has_permission(
        self,
        request: Request,
        identity: str,
        action: str,
        obj: Any | None = None,
    ) -> bool:
        return True

class GrantsAuthorizationBackend(AuthorizationBackend):
    """An `AuthorizationBackend` backed by a set of ``(identity, action)`` grants.

    Subclasses implement
    [`get_grants`][sqladmin.authorization.GrantsAuthorizationBackend.get_grants];
    this class loads them once per request and matches them, wildcards
    included. A superuser is simply granted ``("*", "*")``.

    ???+ usage
        ```python
        class SessionAuthorization(GrantsAuthorizationBackend):
            async def get_grants(self, request):
                if request.session.get("is_admin"):
                    return {("*", "*")}
                # "user:action:deactivate" -> ("user", "action:deactivate")
                return {tuple(g.split(":", 1)) for g in request.session["grants"]}
        ```
    """

    @abstractmethod
    async def get_grants(self, request: Request) -> set[tuple[str, str]]:
        """Return the ``(identity, action)`` pairs granted to the current user.

        Called once per request from `load`. Either element of a pair may be
        [`WILDCARD`][sqladmin.authorization.WILDCARD]; return
        ``{("*", "*")}`` for a user who may do everything.
        """

    async def load(self, request: Request) -> None:
        setattr(request.state, _GRANTS_STATE_ATTR, await self.get_grants(request))

    def has_permission(
        self,
        request: Request,
        identity: str,
        action: str,
        obj: Any | None = None,
    ) -> bool:
        # ``None`` means ``load()`` never ran -- deny rather than fall open.
        grants = getattr(request.state, _GRANTS_STATE_ATTR, None)
        return grants is not None and matches_grant(grants, identity, action)

class DBAuthorizationBackend(GrantsAuthorizationBackend):
    """Resolve a user's grants from the group tables, once per request.

    Args:
        session_maker: A sync or async sessionmaker.
        user_model: Your user model. Required, keyword-only.
        groups_attr: Relationship on the user model pointing at groups.
        accesses_attr: Relationship on the group model pointing at grants.
        superuser_attr: Boolean attribute on the user model that bypasses all
            grant checks. Missing attributes are treated as `False`, so a user
            model without one simply has no superusers.

    ???+ usage
        ```python
        backend = DBAuthorizationBackend(session_maker, user_model=User)
        admin = Admin(app, engine, authorization_backend=backend)
        ```

    The user id comes from
    [`AuthenticationBackend.get_user_id`][sqladmin.authentication.AuthenticationBackend.get_user_id].
    Superusers are granted ``("*", "*")``. Without an
    ``Admin(authentication_backend=...)`` there is no user id to look up, so
    creating the `Admin` fails with
    [`ImproperlyConfigured`][sqladmin.exceptions.ImproperlyConfigured].
    """

    def __init__(
        self,
        session_maker: Any,
        *,
        user_model: Any,
        groups_attr: str = "groups",
        accesses_attr: str = "accesses",
        superuser_attr: str = "is_superuser",
    ) -> None:
        self.session_maker = session_maker
        self.is_async = is_async_session_maker(session_maker)

        pks = get_primary_keys(user_model)
        if len(pks) != 1:
            raise ImproperlyConfigured(
                f"{type(self).__name__} needs {user_model.__name__} to have a "
                "single-column primary key."
            )

        mapper = sa_inspect(user_model).mapper
        group_model = _related_model(
            user_model,
            groups_attr,
            "Pass groups_attr= with its relationship to the group model.",
        )
        access_model = _related_model(
            group_model,
            accesses_attr,
            "Pass accesses_attr= with its relationship to the grant model.",
        )

        # A missing superuser column means there are no superusers.
        superuser = (
            getattr(user_model, superuser_attr)
            if superuser_attr in mapper.column_attrs
            else false()
        )

        # One round trip: the superuser flag plus every grant of every group,
        # each once even when several groups hold it. A user without groups
        # yields one row of NULL grants; an unknown user yields no rows.
        self._grants_query = (
            select(superuser, access_model.identity, access_model.action)
            .select_from(user_model)
            .outerjoin(getattr(user_model, groups_attr))
            .outerjoin(getattr(group_model, accesses_attr))
            .distinct()
        )
        self._user_pk = pks[0]

    def _query_for(self, user_id: Any) -> Select:
        return self._grants_query.where(self._user_pk == user_id)

    @staticmethod
    def _grants_from_rows(rows: Any) -> set[tuple[str, str]]:
        grants = set()
        for is_superuser, identity, action in rows:
            if is_superuser:
                return {(WILDCARD, WILDCARD)}
            if identity is not None:
                grants.add((identity, action))
        return grants

    def _load_grants_sync(self, user_id: Any) -> set[tuple[str, str]]:
        with self.session_maker() as session:
            return self._grants_from_rows(session.execute(self._query_for(user_id)))

    def setup(self, admin: BaseAdmin) -> None:
        if admin.authentication_backend is None:
            raise ImproperlyConfigured(
                f"{type(self).__name__} looks users up by the id from the "
                "authentication backend, but none is configured. Pass "
                "authentication_backend= to Admin."
            )

    async def get_grants(self, request: Request) -> set[tuple[str, str]]:
        user_id = get_current_user_id(request)
        if user_id is None:
            return set()
        if not self.is_async:
            return await anyio.to_thread.run_sync(self._load_grants_sync, user_id)
        async with self.session_maker() as session:
            result = await session.execute(self._query_for(user_id))
            return self._grants_from_rows(result)

class GroupMixin:
    """Declarative mixin for the group table.

    Mix into your own `Base` and set ``__tablename__``. If you name the access
    model something other than ``GroupAccess``, set ``__access_model__`` to
    match.
    """

    __access_model__: ClassVar[str] = DEFAULT_ACCESS_MODEL

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    name: Mapped[str] = mapped_column(String(255), unique=True, nullable=False)

    @declared_attr
    @classmethod
    def accesses(cls) -> Mapped[list[Any]]:
        return relationship(
            cls.__access_model__,
            back_populates="group",
            cascade="all, delete-orphan",
        )

    def __str__(self) -> str:
        return self.name

class GroupAccessMixin:
    """Declarative mixin for one ``(identity, action)`` grant on a group.

    Mix into your own `Base` and set ``__tablename__``. Set ``__group_table__``
    and ``__group_model__`` if your group table or class is not named
    ``admin_groups`` / ``Group``.
    """

    __group_table__: ClassVar[str] = DEFAULT_GROUP_TABLE
    __group_model__: ClassVar[str] = DEFAULT_GROUP_MODEL

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    identity: Mapped[str] = mapped_column(String(255), nullable=False)
    action: Mapped[str] = mapped_column(String(64), nullable=False)

    @declared_attr
    @classmethod
    def group_id(cls) -> Mapped[int]:
        return mapped_column(
            ForeignKey(f"{cls.__group_table__}.id", ondelete="CASCADE"),
            nullable=False,
            index=True,
        )

    @declared_attr
    @classmethod
    def group(cls) -> Mapped[Any]:
        return relationship(cls.__group_model__, back_populates="accesses")

    @declared_attr.directive
    @classmethod
    def __table_args__(cls) -> tuple:
        return (UniqueConstraint("group_id", "identity", "action"),)

    def __str__(self) -> str:
        return f"{self.identity}:{self.action}"

class GroupAdmin(ModelView):
    """Admin view for editing a group and its permissions.

    Subclass it with your group model::

        class MyGroupAdmin(GroupAdmin, model=Group):
            pass

    The permission matrix replaces the raw list of grant rows: one row per
    registered view, one checkbox per action.

    Saving only touches the grants the matrix offers. Anything else stored on
    the group -- wildcard grants such as ``("*", "*")``, or grants for an action
    a view has since disabled -- is left alone.

    By default an editor may only *grant* permissions they hold themselves
    (superusers hold all of them), so edit access to this view cannot be
    turned into more access than the editor already has. Ticking any other box
    fails with a form error and nothing is saved. Override
    [`can_grant`][sqladmin.contrib.rbac.GroupAdmin.can_grant] to change that
    rule. Removing grants is not restricted beyond access to this view:
    who may edit or delete groups is decided by the ``group`` permissions.

    The grants are written in the same transaction as the group, and the
    resulting grants are recorded in the audit entry under the permissions
    field. If you override `on_model_change` or `after_model_change`, call
    ``super()``.
    """

    accesses_attr: ClassVar[str] = "accesses"
    """Relationship on the group model holding its ``(identity, action)`` rows."""

    icon: ClassVar[str] = "fa-solid fa-users"

    column_list: ClassVar[Any] = ["id", "name"]
    form_columns: ClassVar[Any] = ["name"]

    def __init__(self) -> None:
        super().__init__()
        # Checked here so a misconfigured view fails when it is added.
        self._access_model = _related_model(
            self.model,
            self.accesses_attr,
            f"Set {type(self).__name__}.accesses_attr to its relationship to grants.",
        )
        # The edit form reads the grants and saving changes them, so load them
        # with the group there -- and only there.
        self._form_relations = [
            *self._form_relations,
            getattr(self.model, self.accesses_attr),
        ]

    async def scaffold_form(self, rules: list[str] | None = None) -> type[Form]:
        base_form = await super().scaffold_form(rules)
        return type(
            "GroupPermissionForm",
            (base_form,),
            {
                _PERMISSIONS_FIELD: PermissionMatrixField(
                    label=lazy_gettext("Permissions"),
                    rows=build_permission_rows(self._admin_ref),
                    validate_choice=True,
                )
            },
        )

    async def get_form_data_for_edit(self, obj: Any) -> dict[str, Any]:
        data = await super().get_form_data_for_edit(obj)
        # ``form_edit_query`` loads the grants unless an override dropped them.
        if self.accesses_attr in sa_inspect(obj).unloaded:
            relation = getattr(self.model, self.accesses_attr)
            accesses = await self._run_query(
                select(self._access_model).where(with_parent(obj, relation))
            )
        else:
            accesses = getattr(obj, self.accesses_attr)
        data[_PERMISSIONS_FIELD] = [
            _grant_value(access.identity, access.action) for access in accesses
        ]
        return data

    # Persistence ---------------------------------------------------------

    async def on_model_change(
        self, data: dict, model: Any, is_created: bool, request: Request
    ) -> None:
        await super().on_model_change(data, model, is_created, request)

        if _PERMISSIONS_FIELD not in data:
            return
        # Popped so the base class does not try to set it as a model attribute.
        # No box ticked arrives as ``None``, not ``[]``.
        submitted = data.pop(_PERMISSIONS_FIELD) or []

        offered = {
            choice.value: f"{row.label}: {choice.label}"
            for row in build_permission_rows(self._admin_ref)
            for choice in row.choices
        }
        accesses = getattr(model, self.accesses_attr)
        existing = {_grant_value(a.identity, a.action): a for a in accesses}

        desired = set(submitted) & offered.keys()
        changed = (desired ^ set(existing)) & offered.keys()

        added = changed & desired
        forbidden = sorted(
            offered[value]
            for value in added
            if not self.can_grant(request, *_parse_grant(value))
        )
        if forbidden:
            raise PermissionEscalationError(
                gettext("You are not allowed to grant: %(permissions)s")
                % {"permissions": ", ".join(forbidden)}
            )

        for value in changed - desired:
            accesses.remove(existing[value])

        for value in sorted(added):
            identity, action = _parse_grant(value)
            accesses.append(self._access_model(identity=identity, action=action))

        # The field was popped above so it is not set on the model; hand the
        # outcome to ``after_model_change`` so the audit entry includes it.
        setattr(
            request.state,
            _SAVED_PERMISSIONS_ATTR,
            sorted((set(existing) - changed) | added),
        )

    async def after_model_change(
        self, data: dict, model: Any, is_created: bool, request: Request
    ) -> Response | None:
        saved = getattr(request.state, _SAVED_PERMISSIONS_ATTR, None)
        if saved is not None:
            delattr(request.state, _SAVED_PERMISSIONS_ATTR)
            # ``data`` is the dict the audit entry is built from.
            data[_PERMISSIONS_FIELD] = saved
        return await super().after_model_change(data, model, is_created, request)

    def can_grant(self, request: Request, identity: str, action: str) -> bool:
        """Return whether the current user may give ``(identity, action)`` to a group.

        Called for every newly ticked box when a group is saved. The default
        allows only permissions the user holds themselves, so nobody can use
        this screen to gain access they do not already have.

        ???+ usage
            ```python
            class MyGroupAdmin(GroupAdmin, model=Group):
                def can_grant(self, request, identity, action):
                    # HR assigns billing access without using it themselves.
                    if identity == "billing":
                        return self.has_permission(request, "edit")
                    return super().can_grant(request, identity, action)
            ```
        """

        return self._authorization_backend().has_permission(request, identity, action)

class PermissionEscalationError(SQLAdminException, PermissionError):
    """A group editor tried to grant a permission they may not grant.

    See `GroupAdmin.can_grant`.
    """

class InvalidRelationshipError(ImproperlyConfigured):
    """A model has no relationship under the name SQLAdmin was told to use.

    Raised at startup when an RBAC model or view is misconfigured, e.g. when
    `DBAuthorizationBackend(groups_attr=...)` names a missing relationship.
    """

class ImproperlyConfigured(SQLAdminException, ValueError):
    """SQLAdmin was set up in a way that cannot work.

    Raised as soon as the mistake can be detected: usually when the view or
    backend is created, or on the first admin request when it depends on how
    the `Admin` was configured. Also a `ValueError`, so existing
    ``except ValueError`` handlers still work.
    """