Spaces:

Leon4gr45
/

powerbi

Paused

powerbi / tests /integration_tests /security /api_tests.py

Upload folder using huggingface_hub (part 16)

263637f verified 2 months ago

24.1 kB

# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.
# isort:skip_file
"""Tests for security api methods"""

import jwt
import pytest

from flask.ctx import AppContext
from flask_wtf.csrf import generate_csrf
from superset import db, security_manager
from superset.daos.dashboard import EmbeddedDashboardDAO
from superset.models.dashboard import Dashboard
from superset.utils.urls import get_url_host
from superset.utils import json
from tests.conftest import with_config
from tests.integration_tests.base_tests import SupersetTestCase
from tests.integration_tests.constants import ADMIN_USERNAME, GAMMA_USERNAME
from tests.integration_tests.test_app import app
from tests.integration_tests.fixtures.birth_names_dashboard import (
    load_birth_names_dashboard_with_slices,  # noqa: F401
    load_birth_names_data,  # noqa: F401
)


@pytest.fixture
def create_test_roles_with_users(app_context: AppContext):
    """
    Fixture that creates two test roles with specific users, permissions, and groups.
    """
    user1, user2, user3 = [
        security_manager.add_user(
            username=f"test_user_{i}",
            first_name="Test",
            last_name=f"User{i}",
            email=f"test_user_{i}@test.com",
            role=[],
            password="password",  # noqa: S106
        )
        for i in range(3)
    ]

    test_group = security_manager.add_group(
        name="test_group_1",
        label="Test Group 1",
        description="Test group for role testing",
        roles=[],
    )

    pvm1 = security_manager.add_permission_view_menu("can_read", "Dashboard")
    pvm2 = security_manager.add_permission_view_menu("can_write", "Dashboard")
    pvm3 = security_manager.add_permission_view_menu("can_read", "Chart")

    test_role_1 = security_manager.add_role("test_role_1", [pvm1, pvm2])
    test_role_1.user.append(user1)
    test_role_1.user.append(user2)
    test_role_1.groups.append(test_group)

    test_role_2 = security_manager.add_role("test_role_2", [pvm3])
    test_role_2.user.append(user3)

    db.session.commit()

    test_data = {
        "test_role_1": {
            "role": test_role_1,
            "user_ids": sorted([user1.id, user2.id]),
            "permission_ids": sorted([pvm1.id, pvm2.id]),
            "group_ids": [test_group.id],
        },
        "test_role_2": {
            "role": test_role_2,
            "user_ids": [user3.id],
            "permission_ids": [pvm3.id],
            "group_ids": [],
        },
    }

    yield test_data

    # Cleanup
    db.session.delete(test_role_1)
    db.session.delete(test_role_2)
    db.session.delete(user1)
    db.session.delete(user2)
    db.session.delete(user3)
    db.session.delete(test_group)
    db.session.commit()


@pytest.fixture
def inject_test_roles_data(request, create_test_roles_with_users):
    request.instance.test_roles_data = create_test_roles_with_users


class TestSecurityCsrfApi(SupersetTestCase):
    resource_name = "security"

    def _assert_get_csrf_token(self):
        uri = f"api/v1/{self.resource_name}/csrf_token/"
        response = self.client.get(uri)
        self.assert200(response)
        data = json.loads(response.data.decode("utf-8"))
        assert generate_csrf() == data["result"]

    def test_get_csrf_token(self):
        """
        Security API: Test get CSRF token
        """
        self.login(ADMIN_USERNAME)
        self._assert_get_csrf_token()

    def test_get_csrf_token_gamma(self):
        """
        Security API: Test get CSRF token by gamma
        """
        self.login(GAMMA_USERNAME)
        self._assert_get_csrf_token()

    def test_get_csrf_unauthorized(self):
        """
        Security API: Test get CSRF no login
        """
        uri = f"api/v1/{self.resource_name}/csrf_token/"
        response = self.client.get(uri)
        self.assert401(response)

    def test_login(self):
        """
        Security API: Test get login
        """
        uri = f"api/v1/{self.resource_name}/login"
        response = self.client.post(
            uri,
            json={"username": ADMIN_USERNAME, "password": "general", "provider": "db"},
        )
        assert response.status_code == 200
        assert "access_token" in response.json


class TestSecurityGuestTokenApi(SupersetTestCase):
    uri = "api/v1/security/guest_token/"

    def test_post_guest_token_unauthenticated(self):
        """
        Security API: Cannot create a guest token without authentication
        """
        response = self.client.post(self.uri)
        self.assert401(response)

    def test_post_guest_token_unauthorized(self):
        """
        Security API: Cannot create a guest token without authorization
        """
        self.login(GAMMA_USERNAME)
        response = self.client.post(self.uri)
        self.assert403(response)

    @pytest.mark.usefixtures("load_birth_names_dashboard_with_slices")
    def test_post_guest_token_authorized(self):
        self.dash = db.session.query(Dashboard).filter_by(slug="births").first()
        self.embedded = EmbeddedDashboardDAO.upsert(self.dash, [])
        self.login(ADMIN_USERNAME)
        user = {"username": "bob", "first_name": "Bob", "last_name": "Also Bob"}
        resource = {"type": "dashboard", "id": str(self.embedded.uuid)}
        rls_rule = {"dataset": 1, "clause": "1=1"}
        params = {"user": user, "resources": [resource], "rls": [rls_rule]}

        response = self.client.post(
            self.uri, data=json.dumps(params), content_type="application/json"
        )

        self.assert200(response)
        token = json.loads(response.data)["token"]
        decoded_token = jwt.decode(
            token,
            self.app.config["GUEST_TOKEN_JWT_SECRET"],
            audience=get_url_host(),
            algorithms=["HS256"],
        )
        assert user == decoded_token["user"]
        assert resource == decoded_token["resources"][0]

    @pytest.mark.usefixtures("load_birth_names_dashboard_with_slices")
    def test_post_guest_token_with_attributes(self) -> None:
        """
        Security API: Create a guest token with user attributes
        """
        self.dash = db.session.query(Dashboard).filter_by(slug="births").first()
        self.embedded = EmbeddedDashboardDAO.upsert(self.dash, [])
        self.login(ADMIN_USERNAME)

        user = {
            "username": "bob_with_attrs",
            "first_name": "Bob",
            "last_name": "Also Bob",
            "attributes": {
                "department": "Engineering",
                "region": "US",
                "role": "developer",
                "team": "data-platform",
                "clearance_level": "standard",
                "projects": ["analytics", "ml-platform"],
                "team_lead": True,
            },
        }
        resource = {"type": "dashboard", "id": str(self.embedded.uuid)}
        rls_rule = {"dataset": 1, "clause": "1=1"}
        params = {"user": user, "resources": [resource], "rls": [rls_rule]}

        response = self.client.post(
            self.uri, data=json.dumps(params), content_type="application/json"
        )

        self.assert200(response)
        token = json.loads(response.data)["token"]
        decoded_token = jwt.decode(
            token,
            self.app.config["GUEST_TOKEN_JWT_SECRET"],
            audience=get_url_host(),
            algorithms=["HS256"],
        )

        # Verify user attributes are preserved in the token
        assert user == decoded_token["user"]
        assert "attributes" in decoded_token["user"]
        assert decoded_token["user"]["attributes"]["department"] == "Engineering"
        assert decoded_token["user"]["attributes"]["region"] == "US"
        assert decoded_token["user"]["attributes"]["role"] == "developer"
        assert decoded_token["user"]["attributes"]["team"] == "data-platform"
        assert decoded_token["user"]["attributes"]["clearance_level"] == "standard"
        assert decoded_token["user"]["attributes"]["projects"] == [
            "analytics",
            "ml-platform",
        ]
        assert decoded_token["user"]["attributes"]["team_lead"] is True
        assert resource == decoded_token["resources"][0]

    @pytest.mark.usefixtures("load_birth_names_dashboard_with_slices")
    def test_post_guest_token_with_empty_attributes(self) -> None:
        """
        Security API: Create a guest token with empty user attributes
        """
        self.dash = db.session.query(Dashboard).filter_by(slug="births").first()
        self.embedded = EmbeddedDashboardDAO.upsert(self.dash, [])
        self.login(ADMIN_USERNAME)

        user = {
            "username": "bob_empty_attrs",
            "first_name": "Bob",
            "last_name": "Also Bob",
            "attributes": {},
        }
        resource = {"type": "dashboard", "id": str(self.embedded.uuid)}
        rls_rule = {"dataset": 1, "clause": "1=1"}
        params = {"user": user, "resources": [resource], "rls": [rls_rule]}

        response = self.client.post(
            self.uri, data=json.dumps(params), content_type="application/json"
        )

        self.assert200(response)
        token = json.loads(response.data)["token"]
        decoded_token = jwt.decode(
            token,
            self.app.config["GUEST_TOKEN_JWT_SECRET"],
            audience=get_url_host(),
            algorithms=["HS256"],
        )

        # Verify empty attributes are preserved in the token
        assert user == decoded_token["user"]
        assert "attributes" in decoded_token["user"]
        assert decoded_token["user"]["attributes"] == {}
        assert resource == decoded_token["resources"][0]

    @pytest.mark.usefixtures("load_birth_names_dashboard_with_slices")
    def test_post_guest_token_with_null_attributes(self) -> None:
        """
        Security API: Create a guest token with null user attributes
        """
        self.dash = db.session.query(Dashboard).filter_by(slug="births").first()
        self.embedded = EmbeddedDashboardDAO.upsert(self.dash, [])
        self.login(ADMIN_USERNAME)

        user = {
            "username": "bob_null_attrs",
            "first_name": "Bob",
            "last_name": "Also Bob",
            "attributes": None,
        }
        resource = {"type": "dashboard", "id": str(self.embedded.uuid)}
        rls_rule = {"dataset": 1, "clause": "1=1"}
        params = {"user": user, "resources": [resource], "rls": [rls_rule]}

        response = self.client.post(
            self.uri, data=json.dumps(params), content_type="application/json"
        )

        self.assert200(response)
        token = json.loads(response.data)["token"]
        decoded_token = jwt.decode(
            token,
            self.app.config["GUEST_TOKEN_JWT_SECRET"],
            audience=get_url_host(),
            algorithms=["HS256"],
        )

        # Verify null attributes are preserved in the token
        assert user == decoded_token["user"]
        assert "attributes" in decoded_token["user"]
        assert decoded_token["user"]["attributes"] is None
        assert resource == decoded_token["resources"][0]

    @pytest.mark.usefixtures("load_birth_names_dashboard_with_slices")
    def test_post_guest_token_without_attributes_backward_compatibility(self) -> None:
        """
        Security API: Create a guest token without attributes (backward compatibility)
        """
        self.dash = db.session.query(Dashboard).filter_by(slug="births").first()
        self.embedded = EmbeddedDashboardDAO.upsert(self.dash, [])
        self.login(ADMIN_USERNAME)

        user = {
            "username": "bob_no_attrs",
            "first_name": "Bob",
            "last_name": "Also Bob",
            # Note: no attributes field
        }
        resource = {"type": "dashboard", "id": str(self.embedded.uuid)}
        rls_rule = {"dataset": 1, "clause": "1=1"}
        params = {"user": user, "resources": [resource], "rls": [rls_rule]}

        response = self.client.post(
            self.uri, data=json.dumps(params), content_type="application/json"
        )

        self.assert200(response)
        token = json.loads(response.data)["token"]
        decoded_token = jwt.decode(
            token,
            self.app.config["GUEST_TOKEN_JWT_SECRET"],
            audience=get_url_host(),
            algorithms=["HS256"],
        )

        # Verify user without attributes works and no attributes field is present
        assert user == decoded_token["user"]
        assert "attributes" not in decoded_token["user"]
        assert resource == decoded_token["resources"][0]

    @pytest.mark.usefixtures("load_birth_names_dashboard_with_slices")
    def test_post_guest_token_bad_resources(self):
        self.login(ADMIN_USERNAME)
        user = {"username": "bob", "first_name": "Bob", "last_name": "Also Bob"}
        resource = {"type": "dashboard", "id": "bad-id"}
        rls_rule = {"dataset": 1, "clause": "1=1"}
        params = {"user": user, "resources": [resource], "rls": [rls_rule]}

        response = self.client.post(
            self.uri, data=json.dumps(params), content_type="application/json"
        )

        self.assert400(response)


@pytest.mark.usefixtures("load_birth_names_dashboard_with_slices", scope="class")
class TestSecurityGuestTokenApiTokenValidator(SupersetTestCase):
    uri = "api/v1/security/guest_token/"

    def _get_guest_token_with_rls(self, rls_rule):
        dash = db.session.query(Dashboard).filter_by(slug="births").first()
        self.embedded = EmbeddedDashboardDAO.upsert(dash, [])
        self.login(ADMIN_USERNAME)
        user = {"username": "bob", "first_name": "Bob", "last_name": "Also Bob"}
        resource = {"type": "dashboard", "id": str(self.embedded.uuid)}
        params = {"user": user, "resources": [resource], "rls": [rls_rule]}
        return self.client.post(
            self.uri, data=json.dumps(params), content_type="application/json"
        )

    @with_config({"GUEST_TOKEN_VALIDATOR_HOOK": lambda x: False})
    def test_guest_token_validator_hook_denied(self):
        """
        Security API: Test False case from validator - should be 400
        """
        rls_rule = {"dataset": 1, "clause": "tenant_id=123"}
        self.assert400(self._get_guest_token_with_rls(rls_rule))

    @with_config({"GUEST_TOKEN_VALIDATOR_HOOK": lambda x: True})
    def test_guest_token_validator_hook_denied_allowed(self):
        """
        Security API: Test True case from validator - should be 200
        """
        rls_rule = {"dataset": 1, "clause": "tenant_id=123"}
        self.assert200(self._get_guest_token_with_rls(rls_rule))

    @with_config({"GUEST_TOKEN_VALIDATOR_HOOK": 123})
    def test_guest_validator_hook_not_callable(self):
        """
        Security API: Test validator throws exception when isn't callable
         should be 500
        """
        rls_rule = {"dataset": 1, "clause": "tenant_id=123"}
        self.assert500(self._get_guest_token_with_rls(rls_rule))

    @with_config({"GUEST_TOKEN_VALIDATOR_HOOK": lambda x: [][0]})
    def test_guest_validator_hook_throws_exception(self):
        """
        Security API: Test validator throws exception - should be 500
        """
        rls_rule = {"dataset": 1, "clause": "tenant_id=123"}
        self.assert500(self._get_guest_token_with_rls(rls_rule))

    @with_config(
        {
            "GUEST_TOKEN_VALIDATOR_HOOK": lambda x: len(x["rls"]) == 1
            and "tenant_id=" in x["rls"][0]["clause"]
        }
    )
    def test_guest_validator_hook_real_world_example_positive(self):
        """
        Security API: Test validator real world example, check tenant_id is in clause
            Positive case
        """
        # Test validator real world example, check tenant_id is in clause
        # Should be 200.
        rls_rule = {"dataset": 1, "clause": "tenant_id=123"}
        self.assert200(self._get_guest_token_with_rls(rls_rule))

    @with_config(
        {
            "GUEST_TOKEN_VALIDATOR_HOOK": lambda x: len(x["rls"]) == 1
            and "tenant_id=" in x["rls"][0]["clause"]
        }
    )
    def test_guest_validator_hook_real_world_example_negative(self):
        """
        Security API: Test validator real world example, check tenant_id is in clause
            Negative case
        """
        rls_rule = {}
        self.assert400(self._get_guest_token_with_rls(rls_rule))


class TestSecurityRolesApi(SupersetTestCase):
    uri = "api/v1/security/roles/"  # noqa: F541
    show_uri = "api/v1/security/roles/search/"

    @with_config({"FAB_ADD_SECURITY_API": True})
    def test_get_security_roles_admin(self):
        """
        Security API: Admin should be able to get roles
        """
        self.login(ADMIN_USERNAME)
        response = self.client.get(self.uri)
        self.assert200(response)

    @with_config({"FAB_ADD_SECURITY_API": True})
    def test_get_security_roles_gamma(self):
        """
        Security API: Gamma should not be able to get roles
        """
        self.login(GAMMA_USERNAME)
        response = self.client.get(self.uri)
        self.assert403(response)

    @with_config({"FAB_ADD_SECURITY_API": True})
    def test_post_security_roles_gamma(self):
        """
        Security API: Gamma should not be able to create roles
        """
        self.login(GAMMA_USERNAME)
        response = self.client.post(
            self.uri,
            data=json.dumps({"name": "new_role"}),
            content_type="application/json",
        )
        self.assert403(response)

    @with_config({"FAB_ADD_SECURITY_API": True})
    def test_put_security_roles_gamma(self):
        """
        Security API: Gamma shouldnt be able to update roles
        """
        self.login(GAMMA_USERNAME)
        response = self.client.put(
            f"{self.uri}1",
            data=json.dumps({"name": "new_role"}),
            content_type="application/json",
        )
        self.assert403(response)

    @with_config({"FAB_ADD_SECURITY_API": True})
    def test_delete_security_roles_gamma(self):
        """
        Security API: Gamma shouldnt be able to delete roles
        """
        self.login(GAMMA_USERNAME)
        response = self.client.delete(
            f"{self.uri}1",
            data=json.dumps({"name": "new_role"}),
            content_type="application/json",
        )
        self.assert403(response)

    def test_show_roles_unexpected_error_returns_generic_message(self):
        """
        Security API: an unexpected error in role listing returns a generic 500
        body (no raw exception text) and is logged server-side.
        """
        from unittest.mock import patch

        self.login(ADMIN_USERNAME)
        error_detail = "raw-driver-detail-should-not-leak"
        # Patch a symbol used only inside get_list's query construction so the
        # failure happens within the handler's try/except, not in the @protect()
        # auth check (which also touches db.session.query).
        with (
            patch(
                "superset.security.api.selectinload",
                side_effect=Exception(error_detail),
            ),
            patch("superset.security.api.logger") as mock_logger,
        ):
            response = self.client.get(self.show_uri)

        assert response.status_code == 500
        body = response.data.decode("utf-8")
        assert error_detail not in body
        assert "An unexpected error occurred" in body
        mock_logger.exception.assert_called_once()

    def test_show_roles_admin(self):
        """
        Security API: Admin should be able to show roles with permissions and users
        """
        self.login(ADMIN_USERNAME)
        response = self.client.get(self.show_uri)
        self.assert200(response)

    def test_show_roles_gamma(self):
        """
        Security API: Gamma should not be able to show roles
        """
        self.login(GAMMA_USERNAME)
        response = self.client.get(self.show_uri)
        self.assert403(response)

    @pytest.mark.usefixtures("inject_test_roles_data")
    def test_get_roles_with_specific_test_data(self):
        """
        Security API: Test roles endpoint with specific test data
        """
        self.login(ADMIN_USERNAME)
        response = self.client.get(f"{self.show_uri}?q=(page_size:100)")
        self.assert200(response)

        data = json.loads(response.data.decode("utf-8"))

        # Create a mapping of role names to API response
        api_roles_by_name = {role["name"]: role for role in data["result"]}

        # Verify test_role_1
        assert "test_role_1" in api_roles_by_name, (
            f"test_role_1 not found in API response. "
            f"Available roles: {list(api_roles_by_name.keys())}"
        )
        role1_api = api_roles_by_name["test_role_1"]
        role1_expected = self.test_roles_data["test_role_1"]

        assert sorted(role1_api["user_ids"]) == role1_expected["user_ids"]
        assert sorted(role1_api["permission_ids"]) == role1_expected["permission_ids"]
        assert sorted(role1_api["group_ids"]) == role1_expected["group_ids"]

        # Verify test_role_2
        assert "test_role_2" in api_roles_by_name, (
            f"test_role_2 not found in API response. "
            f"Available roles: {list(api_roles_by_name.keys())}"
        )
        role2_api = api_roles_by_name["test_role_2"]
        role2_expected = self.test_roles_data["test_role_2"]

        assert sorted(role2_api["user_ids"]) == role2_expected["user_ids"]
        assert sorted(role2_api["permission_ids"]) == role2_expected["permission_ids"]
        assert role2_api["group_ids"] == role2_expected["group_ids"]


class TestLogoutSessionInvalidation(SupersetTestCase):
    """Regression for #24713: a session cookie captured pre-logout must not grant
    access after the user logs out. The original report describes copying the
    session cookie out, calling /logout/, and successfully reusing the cookie in
    a second browser to bypass authentication."""

    def test_session_cookie_invalidated_after_logout(self):
        self.login(ADMIN_USERNAME)

        resp_authed = self.client.get("api/v1/dashboard/", follow_redirects=False)
        assert resp_authed.status_code == 200, (
            f"Login did not yield an authenticated session "
            f"(got {resp_authed.status_code})"
        )

        # Werkzeug 2.3+ exposes the test client's cookies on `_cookies` as a
        # mapping keyed by (domain, path, key). Snapshot the session cookie
        # value — this is what a malicious actor would copy out of a browser.
        captured = None
        for cookie in self.client._cookies.values():
            if cookie.key == "session":
                captured = cookie.value
                break
        assert captured, "expected a session cookie after login"

        self.client.get("/logout/", follow_redirects=True)

        # Replay the captured cookie in a fresh client (simulates importing
        # the cookie into a second browser).
        replay_client = app.test_client()
        replay_client.set_cookie("session", captured, domain="localhost")

        resp_replay = replay_client.get("api/v1/dashboard/", follow_redirects=False)
        assert resp_replay.status_code != 200, (
            f"Captured session cookie was still accepted after logout "
            f"(status={resp_replay.status_code}); see issue #24713"
        )

	# Licensed to the Apache Software Foundation (ASF) under one
	# or more contributor license agreements. See the NOTICE file
	# distributed with this work for additional information
	# regarding copyright ownership. The ASF licenses this file
	# to you under the Apache License, Version 2.0 (the
	# "License"); you may not use this file except in compliance
	# with the License. You may obtain a copy of the License at
	#
	# http://www.apache.org/licenses/LICENSE-2.0
	#
	# Unless required by applicable law or agreed to in writing,
	# software distributed under the License is distributed on an
	# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
	# KIND, either express or implied. See the License for the
	# specific language governing permissions and limitations
	# under the License.
	# isort:skip_file
	"""Tests for security api methods"""

	import jwt
	import pytest

	from flask.ctx import AppContext
	from flask_wtf.csrf import generate_csrf
	from superset import db, security_manager
	from superset.daos.dashboard import EmbeddedDashboardDAO
	from superset.models.dashboard import Dashboard
	from superset.utils.urls import get_url_host
	from superset.utils import json
	from tests.conftest import with_config
	from tests.integration_tests.base_tests import SupersetTestCase
	from tests.integration_tests.constants import ADMIN_USERNAME, GAMMA_USERNAME
	from tests.integration_tests.test_app import app
	from tests.integration_tests.fixtures.birth_names_dashboard import (
	load_birth_names_dashboard_with_slices, # noqa: F401
	load_birth_names_data, # noqa: F401
	)


	@pytest.fixture
	def create_test_roles_with_users(app_context: AppContext):
	"""
	Fixture that creates two test roles with specific users, permissions, and groups.
	"""
	user1, user2, user3 = [
	security_manager.add_user(
	username=f"test_user_{i}",
	first_name="Test",
	last_name=f"User{i}",
	email=f"test_user_{i}@test.com",
	role=[],
	password="password", # noqa: S106
	)
	for i in range(3)
	]

	test_group = security_manager.add_group(
	name="test_group_1",
	label="Test Group 1",
	description="Test group for role testing",
	roles=[],
	)

	pvm1 = security_manager.add_permission_view_menu("can_read", "Dashboard")
	pvm2 = security_manager.add_permission_view_menu("can_write", "Dashboard")
	pvm3 = security_manager.add_permission_view_menu("can_read", "Chart")

	test_role_1 = security_manager.add_role("test_role_1", [pvm1, pvm2])
	test_role_1.user.append(user1)
	test_role_1.user.append(user2)
	test_role_1.groups.append(test_group)

	test_role_2 = security_manager.add_role("test_role_2", [pvm3])
	test_role_2.user.append(user3)

	db.session.commit()

	test_data = {
	"test_role_1": {
	"role": test_role_1,
	"user_ids": sorted([user1.id, user2.id]),
	"permission_ids": sorted([pvm1.id, pvm2.id]),
	"group_ids": [test_group.id],
	},
	"test_role_2": {
	"role": test_role_2,
	"user_ids": [user3.id],
	"permission_ids": [pvm3.id],
	"group_ids": [],
	},
	}

	yield test_data

	# Cleanup
	db.session.delete(test_role_1)
	db.session.delete(test_role_2)
	db.session.delete(user1)
	db.session.delete(user2)
	db.session.delete(user3)
	db.session.delete(test_group)
	db.session.commit()


	@pytest.fixture
	def inject_test_roles_data(request, create_test_roles_with_users):
	request.instance.test_roles_data = create_test_roles_with_users


	class TestSecurityCsrfApi(SupersetTestCase):
	resource_name = "security"

	def _assert_get_csrf_token(self):
	uri = f"api/v1/{self.resource_name}/csrf_token/"
	response = self.client.get(uri)
	self.assert200(response)
	data = json.loads(response.data.decode("utf-8"))
	assert generate_csrf() == data["result"]

	def test_get_csrf_token(self):
	"""
	Security API: Test get CSRF token
	"""
	self.login(ADMIN_USERNAME)
	self._assert_get_csrf_token()

	def test_get_csrf_token_gamma(self):
	"""
	Security API: Test get CSRF token by gamma
	"""
	self.login(GAMMA_USERNAME)
	self._assert_get_csrf_token()

	def test_get_csrf_unauthorized(self):
	"""
	Security API: Test get CSRF no login
	"""
	uri = f"api/v1/{self.resource_name}/csrf_token/"
	response = self.client.get(uri)
	self.assert401(response)

	def test_login(self):
	"""
	Security API: Test get login
	"""
	uri = f"api/v1/{self.resource_name}/login"
	response = self.client.post(
	uri,
	json={"username": ADMIN_USERNAME, "password": "general", "provider": "db"},
	)
	assert response.status_code == 200
	assert "access_token" in response.json


	class TestSecurityGuestTokenApi(SupersetTestCase):
	uri = "api/v1/security/guest_token/"

	def test_post_guest_token_unauthenticated(self):
	"""
	Security API: Cannot create a guest token without authentication
	"""
	response = self.client.post(self.uri)
	self.assert401(response)

	def test_post_guest_token_unauthorized(self):
	"""
	Security API: Cannot create a guest token without authorization
	"""
	self.login(GAMMA_USERNAME)
	response = self.client.post(self.uri)
	self.assert403(response)

	@pytest.mark.usefixtures("load_birth_names_dashboard_with_slices")
	def test_post_guest_token_authorized(self):
	self.dash = db.session.query(Dashboard).filter_by(slug="births").first()
	self.embedded = EmbeddedDashboardDAO.upsert(self.dash, [])
	self.login(ADMIN_USERNAME)
	user = {"username": "bob", "first_name": "Bob", "last_name": "Also Bob"}
	resource = {"type": "dashboard", "id": str(self.embedded.uuid)}
	rls_rule = {"dataset": 1, "clause": "1=1"}
	params = {"user": user, "resources": [resource], "rls": [rls_rule]}

	response = self.client.post(
	self.uri, data=json.dumps(params), content_type="application/json"
	)

	self.assert200(response)
	token = json.loads(response.data)["token"]
	decoded_token = jwt.decode(
	token,
	self.app.config["GUEST_TOKEN_JWT_SECRET"],
	audience=get_url_host(),
	algorithms=["HS256"],
	)
	assert user == decoded_token["user"]
	assert resource == decoded_token["resources"][0]

	@pytest.mark.usefixtures("load_birth_names_dashboard_with_slices")
	def test_post_guest_token_with_attributes(self) -> None:
	"""
	Security API: Create a guest token with user attributes
	"""
	self.dash = db.session.query(Dashboard).filter_by(slug="births").first()
	self.embedded = EmbeddedDashboardDAO.upsert(self.dash, [])
	self.login(ADMIN_USERNAME)

	user = {
	"username": "bob_with_attrs",
	"first_name": "Bob",
	"last_name": "Also Bob",
	"attributes": {
	"department": "Engineering",
	"region": "US",
	"role": "developer",
	"team": "data-platform",
	"clearance_level": "standard",
	"projects": ["analytics", "ml-platform"],
	"team_lead": True,
	},
	}
	resource = {"type": "dashboard", "id": str(self.embedded.uuid)}
	rls_rule = {"dataset": 1, "clause": "1=1"}
	params = {"user": user, "resources": [resource], "rls": [rls_rule]}

	response = self.client.post(
	self.uri, data=json.dumps(params), content_type="application/json"
	)

	self.assert200(response)
	token = json.loads(response.data)["token"]
	decoded_token = jwt.decode(
	token,
	self.app.config["GUEST_TOKEN_JWT_SECRET"],
	audience=get_url_host(),
	algorithms=["HS256"],
	)

	# Verify user attributes are preserved in the token
	assert user == decoded_token["user"]
	assert "attributes" in decoded_token["user"]
	assert decoded_token["user"]["attributes"]["department"] == "Engineering"
	assert decoded_token["user"]["attributes"]["region"] == "US"
	assert decoded_token["user"]["attributes"]["role"] == "developer"
	assert decoded_token["user"]["attributes"]["team"] == "data-platform"
	assert decoded_token["user"]["attributes"]["clearance_level"] == "standard"
	assert decoded_token["user"]["attributes"]["projects"] == [
	"analytics",
	"ml-platform",
	]
	assert decoded_token["user"]["attributes"]["team_lead"] is True
	assert resource == decoded_token["resources"][0]

	@pytest.mark.usefixtures("load_birth_names_dashboard_with_slices")
	def test_post_guest_token_with_empty_attributes(self) -> None:
	"""
	Security API: Create a guest token with empty user attributes
	"""
	self.dash = db.session.query(Dashboard).filter_by(slug="births").first()
	self.embedded = EmbeddedDashboardDAO.upsert(self.dash, [])
	self.login(ADMIN_USERNAME)

	user = {
	"username": "bob_empty_attrs",
	"first_name": "Bob",
	"last_name": "Also Bob",
	"attributes": {},
	}
	resource = {"type": "dashboard", "id": str(self.embedded.uuid)}
	rls_rule = {"dataset": 1, "clause": "1=1"}
	params = {"user": user, "resources": [resource], "rls": [rls_rule]}

	response = self.client.post(
	self.uri, data=json.dumps(params), content_type="application/json"
	)

	self.assert200(response)
	token = json.loads(response.data)["token"]
	decoded_token = jwt.decode(
	token,
	self.app.config["GUEST_TOKEN_JWT_SECRET"],
	audience=get_url_host(),
	algorithms=["HS256"],
	)

	# Verify empty attributes are preserved in the token
	assert user == decoded_token["user"]
	assert "attributes" in decoded_token["user"]
	assert decoded_token["user"]["attributes"] == {}
	assert resource == decoded_token["resources"][0]

	@pytest.mark.usefixtures("load_birth_names_dashboard_with_slices")
	def test_post_guest_token_with_null_attributes(self) -> None:
	"""
	Security API: Create a guest token with null user attributes
	"""
	self.dash = db.session.query(Dashboard).filter_by(slug="births").first()
	self.embedded = EmbeddedDashboardDAO.upsert(self.dash, [])
	self.login(ADMIN_USERNAME)

	user = {
	"username": "bob_null_attrs",
	"first_name": "Bob",
	"last_name": "Also Bob",
	"attributes": None,
	}
	resource = {"type": "dashboard", "id": str(self.embedded.uuid)}
	rls_rule = {"dataset": 1, "clause": "1=1"}
	params = {"user": user, "resources": [resource], "rls": [rls_rule]}

	response = self.client.post(
	self.uri, data=json.dumps(params), content_type="application/json"
	)

	self.assert200(response)
	token = json.loads(response.data)["token"]
	decoded_token = jwt.decode(
	token,
	self.app.config["GUEST_TOKEN_JWT_SECRET"],
	audience=get_url_host(),
	algorithms=["HS256"],
	)

	# Verify null attributes are preserved in the token
	assert user == decoded_token["user"]
	assert "attributes" in decoded_token["user"]
	assert decoded_token["user"]["attributes"] is None
	assert resource == decoded_token["resources"][0]

	@pytest.mark.usefixtures("load_birth_names_dashboard_with_slices")
	def test_post_guest_token_without_attributes_backward_compatibility(self) -> None:
	"""
	Security API: Create a guest token without attributes (backward compatibility)
	"""
	self.dash = db.session.query(Dashboard).filter_by(slug="births").first()
	self.embedded = EmbeddedDashboardDAO.upsert(self.dash, [])
	self.login(ADMIN_USERNAME)

	user = {
	"username": "bob_no_attrs",
	"first_name": "Bob",
	"last_name": "Also Bob",
	# Note: no attributes field
	}
	resource = {"type": "dashboard", "id": str(self.embedded.uuid)}
	rls_rule = {"dataset": 1, "clause": "1=1"}
	params = {"user": user, "resources": [resource], "rls": [rls_rule]}

	response = self.client.post(
	self.uri, data=json.dumps(params), content_type="application/json"
	)

	self.assert200(response)
	token = json.loads(response.data)["token"]
	decoded_token = jwt.decode(
	token,
	self.app.config["GUEST_TOKEN_JWT_SECRET"],
	audience=get_url_host(),
	algorithms=["HS256"],
	)

	# Verify user without attributes works and no attributes field is present
	assert user == decoded_token["user"]
	assert "attributes" not in decoded_token["user"]
	assert resource == decoded_token["resources"][0]

	@pytest.mark.usefixtures("load_birth_names_dashboard_with_slices")
	def test_post_guest_token_bad_resources(self):
	self.login(ADMIN_USERNAME)
	user = {"username": "bob", "first_name": "Bob", "last_name": "Also Bob"}
	resource = {"type": "dashboard", "id": "bad-id"}
	rls_rule = {"dataset": 1, "clause": "1=1"}
	params = {"user": user, "resources": [resource], "rls": [rls_rule]}

	response = self.client.post(
	self.uri, data=json.dumps(params), content_type="application/json"
	)

	self.assert400(response)


	@pytest.mark.usefixtures("load_birth_names_dashboard_with_slices", scope="class")
	class TestSecurityGuestTokenApiTokenValidator(SupersetTestCase):
	uri = "api/v1/security/guest_token/"

	def _get_guest_token_with_rls(self, rls_rule):
	dash = db.session.query(Dashboard).filter_by(slug="births").first()
	self.embedded = EmbeddedDashboardDAO.upsert(dash, [])
	self.login(ADMIN_USERNAME)
	user = {"username": "bob", "first_name": "Bob", "last_name": "Also Bob"}
	resource = {"type": "dashboard", "id": str(self.embedded.uuid)}
	params = {"user": user, "resources": [resource], "rls": [rls_rule]}
	return self.client.post(
	self.uri, data=json.dumps(params), content_type="application/json"
	)

	@with_config({"GUEST_TOKEN_VALIDATOR_HOOK": lambda x: False})
	def test_guest_token_validator_hook_denied(self):
	"""
	Security API: Test False case from validator - should be 400
	"""
	rls_rule = {"dataset": 1, "clause": "tenant_id=123"}
	self.assert400(self._get_guest_token_with_rls(rls_rule))

	@with_config({"GUEST_TOKEN_VALIDATOR_HOOK": lambda x: True})
	def test_guest_token_validator_hook_denied_allowed(self):
	"""
	Security API: Test True case from validator - should be 200
	"""
	rls_rule = {"dataset": 1, "clause": "tenant_id=123"}
	self.assert200(self._get_guest_token_with_rls(rls_rule))

	@with_config({"GUEST_TOKEN_VALIDATOR_HOOK": 123})
	def test_guest_validator_hook_not_callable(self):
	"""
	Security API: Test validator throws exception when isn't callable
	should be 500
	"""
	rls_rule = {"dataset": 1, "clause": "tenant_id=123"}
	self.assert500(self._get_guest_token_with_rls(rls_rule))

	@with_config({"GUEST_TOKEN_VALIDATOR_HOOK": lambda x: [][0]})
	def test_guest_validator_hook_throws_exception(self):
	"""
	Security API: Test validator throws exception - should be 500
	"""
	rls_rule = {"dataset": 1, "clause": "tenant_id=123"}
	self.assert500(self._get_guest_token_with_rls(rls_rule))

	@with_config(
	{
	"GUEST_TOKEN_VALIDATOR_HOOK": lambda x: len(x["rls"]) == 1
	and "tenant_id=" in x["rls"][0]["clause"]
	}
	)
	def test_guest_validator_hook_real_world_example_positive(self):
	"""
	Security API: Test validator real world example, check tenant_id is in clause
	Positive case
	"""
	# Test validator real world example, check tenant_id is in clause
	# Should be 200.
	rls_rule = {"dataset": 1, "clause": "tenant_id=123"}
	self.assert200(self._get_guest_token_with_rls(rls_rule))

	@with_config(
	{
	"GUEST_TOKEN_VALIDATOR_HOOK": lambda x: len(x["rls"]) == 1
	and "tenant_id=" in x["rls"][0]["clause"]
	}
	)
	def test_guest_validator_hook_real_world_example_negative(self):
	"""
	Security API: Test validator real world example, check tenant_id is in clause
	Negative case
	"""
	rls_rule = {}
	self.assert400(self._get_guest_token_with_rls(rls_rule))


	class TestSecurityRolesApi(SupersetTestCase):
	uri = "api/v1/security/roles/" # noqa: F541
	show_uri = "api/v1/security/roles/search/"

	@with_config({"FAB_ADD_SECURITY_API": True})
	def test_get_security_roles_admin(self):
	"""
	Security API: Admin should be able to get roles
	"""
	self.login(ADMIN_USERNAME)
	response = self.client.get(self.uri)
	self.assert200(response)

	@with_config({"FAB_ADD_SECURITY_API": True})
	def test_get_security_roles_gamma(self):
	"""
	Security API: Gamma should not be able to get roles
	"""
	self.login(GAMMA_USERNAME)
	response = self.client.get(self.uri)
	self.assert403(response)

	@with_config({"FAB_ADD_SECURITY_API": True})
	def test_post_security_roles_gamma(self):
	"""
	Security API: Gamma should not be able to create roles
	"""
	self.login(GAMMA_USERNAME)
	response = self.client.post(
	self.uri,
	data=json.dumps({"name": "new_role"}),
	content_type="application/json",
	)
	self.assert403(response)

	@with_config({"FAB_ADD_SECURITY_API": True})
	def test_put_security_roles_gamma(self):
	"""
	Security API: Gamma shouldnt be able to update roles
	"""
	self.login(GAMMA_USERNAME)
	response = self.client.put(
	f"{self.uri}1",
	data=json.dumps({"name": "new_role"}),
	content_type="application/json",
	)
	self.assert403(response)

	@with_config({"FAB_ADD_SECURITY_API": True})
	def test_delete_security_roles_gamma(self):
	"""
	Security API: Gamma shouldnt be able to delete roles
	"""
	self.login(GAMMA_USERNAME)
	response = self.client.delete(
	f"{self.uri}1",
	data=json.dumps({"name": "new_role"}),
	content_type="application/json",
	)
	self.assert403(response)

	def test_show_roles_unexpected_error_returns_generic_message(self):
	"""
	Security API: an unexpected error in role listing returns a generic 500
	body (no raw exception text) and is logged server-side.
	"""
	from unittest.mock import patch

	self.login(ADMIN_USERNAME)
	error_detail = "raw-driver-detail-should-not-leak"
	# Patch a symbol used only inside get_list's query construction so the
	# failure happens within the handler's try/except, not in the @protect()
	# auth check (which also touches db.session.query).
	with (
	patch(
	"superset.security.api.selectinload",
	side_effect=Exception(error_detail),
	),
	patch("superset.security.api.logger") as mock_logger,
	):
	response = self.client.get(self.show_uri)

	assert response.status_code == 500
	body = response.data.decode("utf-8")
	assert error_detail not in body
	assert "An unexpected error occurred" in body
	mock_logger.exception.assert_called_once()

	def test_show_roles_admin(self):
	"""
	Security API: Admin should be able to show roles with permissions and users
	"""
	self.login(ADMIN_USERNAME)
	response = self.client.get(self.show_uri)
	self.assert200(response)

	def test_show_roles_gamma(self):
	"""
	Security API: Gamma should not be able to show roles
	"""
	self.login(GAMMA_USERNAME)
	response = self.client.get(self.show_uri)
	self.assert403(response)

	@pytest.mark.usefixtures("inject_test_roles_data")
	def test_get_roles_with_specific_test_data(self):
	"""
	Security API: Test roles endpoint with specific test data
	"""
	self.login(ADMIN_USERNAME)
	response = self.client.get(f"{self.show_uri}?q=(page_size:100)")
	self.assert200(response)

	data = json.loads(response.data.decode("utf-8"))

	# Create a mapping of role names to API response
	api_roles_by_name = {role["name"]: role for role in data["result"]}

	# Verify test_role_1
	assert "test_role_1" in api_roles_by_name, (
	f"test_role_1 not found in API response. "
	f"Available roles: {list(api_roles_by_name.keys())}"
	)
	role1_api = api_roles_by_name["test_role_1"]
	role1_expected = self.test_roles_data["test_role_1"]

	assert sorted(role1_api["user_ids"]) == role1_expected["user_ids"]
	assert sorted(role1_api["permission_ids"]) == role1_expected["permission_ids"]
	assert sorted(role1_api["group_ids"]) == role1_expected["group_ids"]

	# Verify test_role_2
	assert "test_role_2" in api_roles_by_name, (
	f"test_role_2 not found in API response. "
	f"Available roles: {list(api_roles_by_name.keys())}"
	)
	role2_api = api_roles_by_name["test_role_2"]
	role2_expected = self.test_roles_data["test_role_2"]

	assert sorted(role2_api["user_ids"]) == role2_expected["user_ids"]
	assert sorted(role2_api["permission_ids"]) == role2_expected["permission_ids"]
	assert role2_api["group_ids"] == role2_expected["group_ids"]


	class TestLogoutSessionInvalidation(SupersetTestCase):
	"""Regression for #24713: a session cookie captured pre-logout must not grant
	access after the user logs out. The original report describes copying the
	session cookie out, calling /logout/, and successfully reusing the cookie in
	a second browser to bypass authentication."""

	def test_session_cookie_invalidated_after_logout(self):
	self.login(ADMIN_USERNAME)

	resp_authed = self.client.get("api/v1/dashboard/", follow_redirects=False)
	assert resp_authed.status_code == 200, (
	f"Login did not yield an authenticated session "
	f"(got {resp_authed.status_code})"
	)

	# Werkzeug 2.3+ exposes the test client's cookies on `_cookies` as a
	# mapping keyed by (domain, path, key). Snapshot the session cookie
	# value — this is what a malicious actor would copy out of a browser.
	captured = None
	for cookie in self.client._cookies.values():
	if cookie.key == "session":
	captured = cookie.value
	break
	assert captured, "expected a session cookie after login"

	self.client.get("/logout/", follow_redirects=True)

	# Replay the captured cookie in a fresh client (simulates importing
	# the cookie into a second browser).
	replay_client = app.test_client()
	replay_client.set_cookie("session", captured, domain="localhost")

	resp_replay = replay_client.get("api/v1/dashboard/", follow_redirects=False)
	assert resp_replay.status_code != 200, (
	f"Captured session cookie was still accepted after logout "
	f"(status={resp_replay.status_code}); see issue #24713"
	)