Assign HQ-defined roles to CCA user

hq_superset/const.py

@@ -2,3 +2,67 @@
 HQ_DATABASE_NAME = "HQ Data"
 
 OAUTH2_DATABASE_NAME = "oauth2-server-data"
+
+HQ_USER_ROLE_NAME = "hq_user"
+
+HQ_ROLE_NAME_MAPPING = {
+    "gamma": "Gamma",
+    "dataset_editor": "dataset_editor",
+    "sql_lab": "sql_lab",
+}
+
+# Permissions
+SCHEMA_ACCESS_PERMISSION = "schema_access"
+MENU_ACCESS_PERMISSION = "menu_access"
+
+CAN_SHOW_PERMISSION = "can_show"
+CAN_LIST_PERMISSION = "can_list"
+CAN_GET_PERMISSION = "can_get"
+CAN_EXTERNAL_METADATA_PERMISSION = "can_external_metadata"
+CAN_EXTERNAL_METADATA_BY_NAME_PERMISSION = "can_external_metadata_by_name"
+CAN_READ_PERMISSION = "can_read"
+
+READ_ONLY_PERMISSIONS = [
+    CAN_SHOW_PERMISSION,
+    CAN_LIST_PERMISSION,
+    CAN_GET_PERMISSION,
+    CAN_EXTERNAL_METADATA_PERMISSION,
+    CAN_EXTERNAL_METADATA_BY_NAME_PERMISSION,
+    CAN_READ_PERMISSION,
+]
+
+CAN_WRITE_PERMISSION = "can_write"
+CAN_EDIT_PERMISSION = "can_edit"
+CAN_ADD_PERMISSION = "can_add"
+CAN_DELETE_PERMISSIONS = "can_delete"
+
+WRITE_PERMISSIONS = [
+    CAN_WRITE_PERMISSION,
+    CAN_EDIT_PERMISSION,
+    CAN_ADD_PERMISSION,
+    CAN_DELETE_PERMISSIONS,
+]
+
+READ_ONLY_MENU_PERMISSIONS = {
+    "Chart": READ_ONLY_PERMISSIONS,
+    "Dataset": READ_ONLY_PERMISSIONS,
+    "Dashboard": READ_ONLY_PERMISSIONS,
+    "Datasource": READ_ONLY_PERMISSIONS,
+    "OpenApi": READ_ONLY_PERMISSIONS,
+    "Explore": READ_ONLY_PERMISSIONS,
+    "Select a Domain": [MENU_ACCESS_PERMISSION],
+    "Home": [MENU_ACCESS_PERMISSION],
+    "Data": [MENU_ACCESS_PERMISSION],
+    "Dashboards": [MENU_ACCESS_PERMISSION],
+    "Charts": [MENU_ACCESS_PERMISSION],
+    "Datasets": [MENU_ACCESS_PERMISSION],
+    "ExploreFormDataRestApi": [CAN_READ_PERMISSION]
+}
+
+WRITE_MENU_PERMISSIONS = {
+    "Chart": WRITE_PERMISSIONS,
+    "Dataset": WRITE_PERMISSIONS,
+    "Dashboard": WRITE_PERMISSIONS,
+    "Datasource": WRITE_PERMISSIONS,
+    "ExploreFormDataRestApi": [CAN_WRITE_PERMISSION],
+}

hq_superset/hq_url.py

@@ -30,3 +30,7 @@ def datasource_unsubscribe(domain, datasource_id):
         f"a/{domain}/configurable_reports/data_sources/unsubscribe/"
         f"{datasource_id}/"
     )
+
+
+def user_domain_roles(domain):
+    return f"a/{domain}/api/v0.5/cca-roles/"

hq_superset/oauth.py

@@ -56,6 +56,15 @@ def set_oauth_session(self, provider, oauth_response):
         # }
         session[SESSION_OAUTH_RESPONSE_KEY] = oauth_response
 
+    def set_role_permissions(self, role, permissions):
+        """
+        This method sets the permissions on a role by overwriting the existing
+        permissions
+        """
+        role.permissions = []
+        for permission in permissions:
+            self.add_permission_role(role, permission)
+
 
 def get_valid_cchq_oauth_token():
     # Returns a valid working oauth access_token and also saves it on session

hq_superset/tests/config_for_tests.py

@@ -25,11 +25,6 @@
     '0fXurIGyQM4HQYoe7feuwV8c1Kz_88BdmCNutLKiO38=',  # Don't reuse this!
 ]
 
-# Any other additional roles to be assigned to the user on top of the base role
-# Note: by design we cannot use AUTH_USER_REGISTRATION_ROLE to
-# specify more than one role
-AUTH_USER_ADDITIONAL_ROLES = ["sql_lab"]
-
 HQ_DATABASE_URI = "postgresql://commcarehq:commcarehq@localhost:5432/test_superset_hq"
 
 AUTH_TYPE = AUTH_OAUTH

hq_superset/tests/test_oauth.py

@@ -72,6 +72,7 @@ def test_oauth_user_info(self):
             oauth_mock.domain_json["objects"]
         )
 
+
 class TestGetOAuthTokenGetter(SupersetTestCase):
 
     def tearDown(self):
@@ -110,3 +111,27 @@ def test_tries_token_refresh_if_expired(self):
             set_mock.assert_called_once_with(
                 "commcare", {"access_token": "new key"}
             )
+
+
+class TestSetRolePermissions(SupersetTestCase):
+
+    def tearDown(self):
+        session.clear()
+
+    def test_set_role_permissions(self):
+        appbuilder = self.app.appbuilder
+        role_name = "test_role"
+
+        role = appbuilder.sm.add_role(role_name)
+        permissions = [
+            appbuilder.sm.add_permission_view_menu("can_edit", "Chart"),
+            appbuilder.sm.add_permission_view_menu("can_edit", "Dashboard"),
+        ]
+
+        appbuilder.sm.set_role_permissions(role, permissions)
+        role = appbuilder.sm.find_role(role_name)
+        assert role.permissions == permissions
+
+        appbuilder.sm.set_role_permissions(role, [])
+        role = appbuilder.sm.find_role(role_name)
+        assert role.permissions == []

hq_superset/tests/test_utils.py

@@ -1,7 +1,9 @@
 import doctest
+from unittest.mock import patch
 
-from hq_superset.utils import get_column_dtypes
-
+from hq_superset.utils import get_column_dtypes, DomainSyncUtil
+from hq_superset.const import READ_ONLY_MENU_PERMISSIONS, WRITE_MENU_PERMISSIONS
+from .base_test import SupersetTestCase
 from .const import TEST_DATASOURCE
 
 
@@ -24,3 +26,143 @@ def test_doctests():
     import hq_superset.utils
     results = doctest.testmod(hq_superset.utils)
     assert results.failed == 0
+
+
+class TestDomainSyncUtil(SupersetTestCase):
+    PLATFORM_ROLE_NAMES = ["Gamma", "sql_lab", "dataset_editor"]
+
+    @patch.object(DomainSyncUtil, "_domain_user_role_name")
+    def test_get_user_domain_role_for_permissions_read_only(self, domain_user_role_name_mock):
+        role_name = "test-domain_user_1"
+        domain_user_role_name_mock.return_value = role_name
+        security_manager = self.app.appbuilder.sm
+
+        role = DomainSyncUtil(security_manager)._get_user_domain_role_for_permissions(
+            "test-domain", {"can_read": True, "can_write": False}
+        )
+        assert role.name == role_name
+
+        expected_permissions_count = 0
+        for view_menu_name, permissions_names in READ_ONLY_MENU_PERMISSIONS.items():
+            for permission_name in permissions_names:
+                expected_permissions_count += 1
+                pv = security_manager.find_permission_view_menu(permission_name, view_menu_name)
+                assert pv in role.permissions
+
+        assert len(role.permissions) == expected_permissions_count
+
+    @patch.object(DomainSyncUtil, "_domain_user_role_name")
+    def test_get_user_domain_role_for_permissions_write_only(self, domain_user_role_name_mock):
+        role_name = "test-domain_user_1"
+        domain_user_role_name_mock.return_value = role_name
+        security_manager = self.app.appbuilder.sm
+
+        role = DomainSyncUtil(security_manager)._get_user_domain_role_for_permissions(
+            "test-domain", {"can_read": False, "can_write": True}
+        )
+        assert role.name == role_name
+
+        expected_permissions_count = 0
+        for view_menu_name, permissions_names in WRITE_MENU_PERMISSIONS.items():
+            for permission_name in permissions_names:
+                expected_permissions_count += 1
+                pv = security_manager.find_permission_view_menu(permission_name, view_menu_name)
+                assert pv in role.permissions
+
+        assert len(role.permissions) == expected_permissions_count
+
+    @patch.object(DomainSyncUtil, "_domain_user_role_name")
+    @patch.object(DomainSyncUtil, "_get_domain_access")
+    def test_admin_additional_domain_roles(self, get_domain_access_mock, domain_user_role_name_mock):
+        domain_user_role_name = "test-domain_user_1"
+        domain_user_role_name_mock.return_value = domain_user_role_name
+
+        security_manager = self.app.appbuilder.sm
+        self._ensure_platform_roles_exist(security_manager)
+
+        get_domain_access_mock.return_value = self._to_permissions_response(
+            can_write=True,
+            can_read=True,
+            roles=["Gamma", "sql_lab", "dataset_editor"],
+        )
+        domain_user_role, platform_roles = DomainSyncUtil(security_manager)._get_additional_user_roles("test-domain")
+
+        assert domain_user_role.name == "test-domain_user_1"
+        assert sorted([role.name for role in platform_roles]) == sorted(self.PLATFORM_ROLE_NAMES)
+
+    @patch.object(DomainSyncUtil, "_domain_user_role_name")
+    @patch.object(DomainSyncUtil, "_get_domain_access")
+    def test_limited_additional_domain_roles(self, get_domain_access_mock, domain_user_role_name_mock):
+        domain_user_role_name = "test-domain_user_1"
+        domain_user_role_name_mock.return_value = domain_user_role_name
+
+        security_manager = self.app.appbuilder.sm
+        self._ensure_platform_roles_exist(security_manager)
+
+        get_domain_access_mock.return_value = self._to_permissions_response(
+            can_write=False,
+            can_read=True,
+            roles=["sql_lab"],
+        )
+        _, platform_roles = DomainSyncUtil(security_manager)._get_additional_user_roles("test-domain")
+
+        assert len(platform_roles) == 1
+        assert platform_roles[0].name == "sql_lab"
+
+    @patch.object(DomainSyncUtil, "_domain_user_role_name")
+    @patch.object(DomainSyncUtil, "_get_domain_access")
+    def test_no_access_domain_roles(self, get_domain_access_mock, domain_user_role_name_mock):
+        domain_user_role_name = "test-domain_user_1"
+        domain_user_role_name_mock.return_value = domain_user_role_name
+
+        security_manager = self.app.appbuilder.sm
+        self._ensure_platform_roles_exist(security_manager)
+
+        get_domain_access_mock.return_value = self._to_permissions_response(
+            can_write=False,
+            can_read=False,
+            roles=[],
+        )
+        domain_user_role, platform_roles = DomainSyncUtil(security_manager)._get_additional_user_roles("test-domain")
+        assert domain_user_role is None
+        assert platform_roles == []
+
+    @patch.object(DomainSyncUtil, "_domain_user_role_name")
+    @patch.object(DomainSyncUtil, "_get_domain_access")
+    def test_permissions_change_updates_user_role(self, get_domain_access_mock, domain_user_role_name_mock):
+        domain_user_role_name = "test-domain_user_1"
+        domain_user_role_name_mock.return_value = domain_user_role_name
+
+        security_manager = self.app.appbuilder.sm
+        self._ensure_platform_roles_exist(security_manager)
+
+        # user has maximum access
+        get_domain_access_mock.return_value = self._to_permissions_response(
+            can_write=False,
+            can_read=True,
+            roles=["sql_lab"],
+        )
+        _, platform_roles = DomainSyncUtil(security_manager)._get_additional_user_roles("test-domain")
+
+        assert platform_roles[0].name == "sql_lab"
+
+        # user has no access
+        get_domain_access_mock.return_value = self._to_permissions_response(
+            can_write=False,
+            can_read=False,
+            roles=[],
+        )
+        domain_user_role, platform_roles = DomainSyncUtil(security_manager)._get_additional_user_roles("test-domain")
+        assert domain_user_role is None
+        assert platform_roles == []
+
+    def _ensure_platform_roles_exist(self, sm):
+        for role_name in self.PLATFORM_ROLE_NAMES:
+            sm.add_role(role_name)
+
+    @staticmethod
+    def _to_permissions_response(can_write, can_read, roles):
+        return {
+            "can_write": can_write,
+            "can_read": can_read,
+        }, roles

hq_superset/tests/test_views.py

@@ -12,6 +12,7 @@
 from hq_superset.utils import (
     SESSION_USER_DOMAINS_KEY,
     get_schema_name_for_domain,
+    DomainSyncUtil,
 )
 
 from .base_test import HQDBTestCase
@@ -80,6 +81,10 @@ def __init__(self):
             ]
         }
         self.api_base_url = "https://cchq.org/"
+        self.user_domain_roles = {
+            "permissions": {"can_view": True, "can_edit": True},
+            "roles": ["Gamma", "sql_lab"],
+        }
 
     def authorize_access_token(self):
         return {"access_token": "some-key"}
@@ -92,6 +97,8 @@ def get(self, url, token):
             'a/test2/api/v0.5/ucr_data_source/': MockResponse(self.test2_datasources, 200),
             'a/test1/api/v0.5/ucr_data_source/test1_ucr1/': MockResponse(TEST_DATASOURCE, 200),
             'a/test1/configurable_reports/data_sources/export/test1_ucr1/?format=csv': MockResponse(TEST_UCR_CSV_V1, 200),
+            'a/test1/api/v0.5/cca-roles/': MockResponse(self.user_domain_roles, 200),
+            'a/test2/api/v0.5/cca-roles/': MockResponse(self.user_domain_roles, 200),
         }[url]
 
 
@@ -171,7 +178,8 @@ def test_redirects_to_domain_select_after_login(self):
             )
             self.logout(client)
 
-    def test_domain_select_works(self):
+    @patch.object(DomainSyncUtil, "_get_domain_access", return_value=({"can_write": True, "can_read": True}, []))
+    def test_domain_select_works(self, *args):
         client = self.app.test_client()
         self.login(client)
 
@@ -201,6 +209,7 @@ def test_non_user_domain_cant_be_selected(self):
         self.logout(client)
 
     @patch('hq_superset.hq_requests.get_valid_cchq_oauth_token', return_value={})
+    @patch.object(DomainSyncUtil, "sync_domain_role", return_value=True)
     def test_datasource_list(self, *args):
         def _do_assert(datasources):
             self.assert_template_used("hq_datasource_list.html")
@@ -218,6 +227,7 @@ def _do_assert(datasources):
         client.get('/hq_datasource/list/', follow_redirects=True)
         _do_assert(self.oauth_mock.test2_datasources)
 
+    @patch.object(DomainSyncUtil, "sync_domain_role", return_value=True)
     def test_datasource_upload(self, *args):
         client = self.app.test_client()
         self.login(client)
@@ -232,7 +242,8 @@ def test_datasource_upload(self, *args):
                 'ds1'
             )
 
-    def test_trigger_datasource_refresh_with_api_exception(self):
+    @patch.object(DomainSyncUtil, "sync_domain_role", return_value=True)
+    def test_trigger_datasource_refresh_with_api_exception(self, *args):
         with patch("hq_superset.views.download_and_subscribe_to_datasource", side_effect=HQAPIException('mocked error')):
             client = self.app.test_client()
             self.login(client)