| # 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_wtf.csrf import generate_csrf |
| from superset import db |
| 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.fixtures.birth_names_dashboard import ( |
| load_birth_names_dashboard_with_slices, # noqa: F401 |
| load_birth_names_data, # noqa: F401 |
| ) |
| |
| |
| 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_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_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) |
