petrolyst-infra/petrolyst-insights/petrolyst_reader.py

"""Readers: ATOMA's users, who only read the ATOMA dashboard.

A reader is a signed-in user whose roles are "ATOMA Reader" and nothing more ("Public" aside,
which carries no permissions here). A user who also has Admin, Alpha or Gamma is not a reader.
The role itself (roles/atoma_reader.py) grants no menus and nothing to write; this module adds
what a role cannot express. Checked against apache/superset:6.1.0.

- Superset lets an owner past the role check, and anyone onto a dashboard that has neither
  roles nor datasets. ReaderAccessMixin lets readers reach a dashboard only through its roles
  (DASHBOARD_RBAC), and the dashboard list shows them only those.
- Chart-data responses carry each query's SQL for everyone but embedded guests. Readers get none:
  ReaderAccessMixin.redacts_query, read by patches/fix_redact_query_for_readers.py.
- Readers land on the ATOMA dashboard instead of Superset's home page, and any other page they
  open (a chart or dashboard list, SQL Lab, Explore, user info) leads there too. Every page
  loads reader.css for them (templates/head_custom_extra.html). It hides what Superset shows
  whatever the permissions: the dashboard's title row, the theme switcher, Info and About in
  Settings; and the filter column on ATOMA's Fleet tab. Its address names its content, since
  browsers keep Superset's static files for a year.

superset_config.py sets FLASK_APP_MUTATOR = register; both security managers (Keycloak in
petrolyst_security.py, database login on the local stack) take ReaderAccessMixin.
"""

from __future__ import annotations

import hashlib
from pathlib import Path
from typing import Any

from flask import Flask, g, has_request_context, redirect, request
from flask_login import current_user
from jinja2 import ChoiceLoader, FileSystemLoader
from werkzeug.wrappers import Response as WerkzeugResponse

READER_ROLE = "ATOMA Reader"
# Roles a reader may have besides READER_ROLE without stopping being one.
NEUTRAL_ROLES = {"Public"}
LANDING = "/superset/dashboard/atoma/"
# The only pages a reader opens; any other page leads to LANDING. Requests that are not page
# loads (the API, static files) keep their own permission checks.
READER_PAGES = ("/superset/dashboard/", "/login/", "/logout/")
TEMPLATES = Path(__file__).with_name("templates")


def _signed_in_user() -> Any:
    user = getattr(g, "user", None)
    if user is None and has_request_context():
        user = current_user
    return user


def is_reader(user: Any = None) -> bool:
    """True for a signed-in user whose roles are READER_ROLE and nothing that grants more."""
    user = user if user is not None else _signed_in_user()
    if user is None or not getattr(user, "is_authenticated", False):
        return False
    names = {role.name for role in getattr(user, "roles", None) or []}
    return READER_ROLE in names and names <= {READER_ROLE, *NEUTRAL_ROLES}


class ReaderAccessMixin:
    """Reader rules for a SupersetSecurityManager subclass; list it before that base."""

    def raise_for_ownership(self, resource: Any) -> None:
        # is_owner() goes through here: a reader owns nothing, so no owner path to a
        # dashboard, chart or dataset remains (also for a user who owned things before).
        if is_reader():
            from superset.errors import ErrorLevel, SupersetError, SupersetErrorType
            from superset.exceptions import SupersetSecurityException

            raise SupersetSecurityException(
                SupersetError(
                    error_type=SupersetErrorType.MISSING_OWNERSHIP_ERROR,
                    message=f"A reader does not own {resource}",
                    level=ErrorLevel.ERROR,
                )
            )
        super().raise_for_ownership(resource)  # type: ignore[misc]

    def raise_for_access(self, *args: Any, **kwargs: Any) -> None:
        # The dashboard page, its API, permalinks and chart data (via can_access_dashboard)
        # all check a dashboard here.
        dashboard = kwargs.get("dashboard", args[0] if args else None)
        if dashboard is not None and is_reader() and not self._reader_may_open(dashboard):
            from superset.exceptions import SupersetSecurityException

            raise SupersetSecurityException(
                self.get_dashboard_access_error_object(dashboard)  # type: ignore[attr-defined]
            )
        super().raise_for_access(*args, **kwargs)  # type: ignore[misc]

    def _reader_may_open(self, dashboard: Any) -> bool:
        """Only DASHBOARD_RBAC: published, and one of its roles is the reader's."""
        mine = {role.id for role in self.get_user_roles()}  # type: ignore[attr-defined]
        return bool(dashboard.published) and bool(mine & {role.id for role in dashboard.roles})

    def redacts_query(self) -> bool:
        """Whether chart-data responses leave out the SQL (embedded guests aside)."""
        return is_reader()


def _land_readers() -> WerkzeugResponse | None:
    if request.method != "GET" or request.path.startswith(READER_PAGES) or not is_reader():
        return None
    # A page load in a browser, or Superset's home page from anything.
    if request.headers.get("Sec-Fetch-Dest") == "document" or request.path in (
        "/",
        "/superset/welcome/",
    ):
        return redirect(LANDING)
    return None


def _limit_dashboard_list() -> None:
    """The dashboard list shows readers only published dashboards that carry their roles."""
    from superset import security_manager
    from superset.dashboards.filters import DashboardAccessFilter
    from superset.models.dashboard import Dashboard

    original = DashboardAccessFilter.apply
    if getattr(original, "petrolyst_readers", False):
        return

    def apply(self: DashboardAccessFilter, query: Any, value: Any) -> Any:
        query = original(self, query, value)
        if is_reader():
            role_ids = [role.id for role in security_manager.get_user_roles()]
            query = query.filter(
                Dashboard.published.is_(True),
                Dashboard.roles.any(security_manager.role_model.id.in_(role_ids)),
            )
        return query

    apply.petrolyst_readers = True  # type: ignore[attr-defined]
    DashboardAccessFilter.apply = apply  # type: ignore[method-assign]


def register(app: Flask) -> None:
    """FLASK_APP_MUTATOR: runs once Superset is initialized."""
    # Our templates first: head_custom_extra.html, which Superset's spa.html includes on every
    # page, is the only file there.
    app.jinja_env.loader = ChoiceLoader([FileSystemLoader(str(TEMPLATES)), app.jinja_env.loader])
    app.jinja_env.globals["petrolyst_is_reader"] = is_reader
    # Static files are cached for a year (Cache-Control max-age), so the address carries the
    # file's hash: a changed reader.css reaches readers on their next page load.
    css = Path(app.static_folder or "") / "assets" / "branding" / "reader.css"
    version = hashlib.sha256(css.read_bytes()).hexdigest()[:12]
    app.jinja_env.globals["petrolyst_reader_css_version"] = version
    app.before_request(_land_readers)
    _limit_dashboard_list()