Merge pull request #6568 from jmchilton/dataset-permissions-api-2

Unified dataset permission API.

test/api/test_dataset_collections.py

@@ -160,7 +160,7 @@ class DatasetCollectionApiTestCase(api.ApiTestCase):
 
     def test_hda_security(self):
         element_identifiers = self.dataset_collection_populator.pair_identifiers(self.history_id)
-
+        self.dataset_populator.make_private(self.history_id, element_identifiers[0]["id"])
         with self._different_user():
             history_id = self.dataset_populator.new_history()
             payload = dict(
@@ -169,11 +169,8 @@ class DatasetCollectionApiTestCase(api.ApiTestCase):
                 element_identifiers=json.dumps(element_identifiers),
                 collection_type="paired",
             )
-
-            self._post("dataset_collections", payload)
-            # TODO: re-enable once there is a way to restrict access
-            # to this dataset via the API.
-            # self._assert_status_code_is( create_response, 403 )
+            create_response = self._post("dataset_collections", payload)
+            self._assert_status_code_is(create_response, 403)
 
     def test_enforces_unique_names(self):
         element_identifiers = self.dataset_collection_populator.list_identifiers(self.history_id)

test/api/test_history_contents.py

@@ -30,6 +30,114 @@ class HistoryContentsApiTestCase(api.ApiTestCase, TestsDatasets):
         hda_summary = self.__check_for_hda(contents_response, hda1)
         assert "display_types" not in hda_summary  # Quick summary, not full details
 
+    def test_make_private_and_public(self):
+        hda1 = self._wait_for_new_hda()
+        update_url = "histories/%s/contents/%s/permissions" % (self.history_id, hda1["id"])
+
+        role_id = self.dataset_populator.user_private_role_id()
+        # Give manage permission to the user.
+        payload = {
+            "access": [],
+            "manage": [role_id],
+        }
+        update_response = self._update_permissions(update_url, payload, admin=True)
+        self._assert_status_code_is(update_response, 200)
+        self._assert_other_user_can_access(hda1["id"])
+        # Then we restrict access.
+        payload = {
+            "action": "make_private",
+        }
+        update_response = self._update_permissions(update_url, payload)
+        self._assert_status_code_is(update_response, 200)
+        self._assert_other_user_cannot_access(hda1["id"])
+
+        # Then we restrict access.
+        payload = {
+            "action": "remove_restrictions",
+        }
+        update_response = self._update_permissions(update_url, payload)
+        self._assert_status_code_is(update_response, 200)
+        self._assert_other_user_can_access(hda1["id"])
+
+    def test_set_permissions_add_admin_history_contents(self):
+        self._verify_dataset_permissions("history_contents")
+
+    def test_set_permissions_add_admin_datasets(self):
+        self._verify_dataset_permissions("dataset")
+
+    def _verify_dataset_permissions(self, api_endpoint):
+        hda1 = self._wait_for_new_hda()
+        hda_id = hda1["id"]
+        if api_endpoint == "history_contents":
+            update_url = "histories/%s/contents/%s/permissions" % (self.history_id, hda_id)
+        else:
+            update_url = "datasets/%s/permissions" % hda_id
+
+        role_id = self.dataset_populator.user_private_role_id()
+
+        payload = {
+            "access": [role_id],
+            "manage": [role_id],
+        }
+
+        # Other users cannot modify permissions.
+        with self._different_user():
+            update_response = self._update_permissions(update_url, payload)
+            self._assert_status_code_is(update_response, 403)
+
+        # First the details render for another user.
+        self._assert_other_user_can_access(hda_id)
+
+        # Then we restrict access.
+        update_response = self._update_permissions(update_url, payload, admin=True)
+        self._assert_status_code_is(update_response, 200)
+
+        # Finally the details don't render.
+        self._assert_other_user_cannot_access(hda_id)
+
+        # But they do for the original user.
+        contents_response = self._get("histories/%s/contents/%s" % (self.history_id, hda_id)).json()
+        assert "name" in contents_response
+
+        update_response = self._update_permissions(update_url, payload)
+        self._assert_status_code_is(update_response, 200)
+
+        payload = {
+            "access": [role_id],
+            "manage": [role_id],
+        }
+        update_response = self._update_permissions(update_url, payload)
+        self._assert_status_code_is(update_response, 200)
+        self._assert_other_user_cannot_access(hda_id)
+
+        user_id = self.dataset_populator.user_id()
+        with self._different_user():
+            different_user_id = self.dataset_populator.user_id()
+        combined_user_role = self.dataset_populator.create_role([user_id, different_user_id], description="role for testing permissions")
+
+        payload = {
+            "access": [combined_user_role["id"]],
+            "manage": [role_id],
+        }
+        update_response = self._update_permissions(update_url, payload)
+        self._assert_status_code_is(update_response, 200)
+        # Now other user can see dataset again with access permission.
+        self._assert_other_user_can_access(hda_id)
+        # access doesn't imply management though...
+        with self._different_user():
+            update_response = self._update_permissions(update_url, payload)
+            self._assert_status_code_is(update_response, 403)
+
+    def _assert_other_user_cannot_access(self, history_content_id):
+        with self._different_user():
+            contents_response = self._get("histories/%s/contents/%s" % (self.history_id, history_content_id)).json()
+            assert "name" not in contents_response
+
+    def _assert_other_user_can_access(self, history_content_id):
+        with self._different_user():
+            contents_response = self._get("histories/%s/contents/%s" % (self.history_id, history_content_id)).json()
+            assert "name" in contents_response
+
     def test_index_hda_all_details(self):
         hda1 = self._new_dataset(self.history_id)
         contents_response = self._get("histories/%s/contents?details=all" % self.history_id)
@@ -105,8 +213,15 @@ class HistoryContentsApiTestCase(api.ApiTestCase, TestsDatasets):
         self._wait_for_history(self.history_id)
         return hda1
 
-    def _raw_update(self, item_id, data):
-        update_url = self._api_url("histories/%s/contents/%s" % (self.history_id, item_id), use_key=True)
+    def _raw_update(self, item_id, data, admin=False):
+        key_param = "use_admin_key" if admin else "use_key"
+        update_url = self._api_url("histories/%s/contents/%s" % (self.history_id, item_id), **{key_param: True})
+        update_response = put(update_url, json=data)
+        return update_response
+
+    def _update_permissions(self, url, data, admin=False):
+        key_param = "use_admin_key" if admin else "use_key"
+        update_url = self._api_url(url, **{key_param: True})
         update_response = put(update_url, json=data)
         return update_response
 

test/api/test_roles.py

@@ -0,0 +1,104 @@
+import json
+
+from base import api
+from base.api_asserts import (
+    assert_has_keys,
+    assert_status_code_is,
+)
+from base.populators import (  # noqa: I100
+    DatasetPopulator,
+)
+
+
+class RolesApiTestCase(api.ApiTestCase):
+
+    def setUp(self):
+        super(RolesApiTestCase, self).setUp()
+        self.dataset_populator = DatasetPopulator(self.galaxy_interactor)
+
+    def test_list_and_show(self):
+
+        def check_roles_response(response):
+            assert response.status_code == 200
+            as_list = response.json()
+            assert isinstance(as_list, list)
+            assert len(as_list) > 0
+            for role in as_list:
+                RolesApiTestCase.check_role_dict(role)
+
+        user_role_id = self.dataset_populator.user_private_role_id()
+        with self._different_user():
+            different_user_role_id = self.dataset_populator.user_private_role_id()
+
+        admin_roles_response = self._get("roles", admin=True)
+        user_roles_response = self._get("roles")
+
+        check_roles_response(admin_roles_response)
+        check_roles_response(user_roles_response)
+
+        admin_roles_response_ids = [r["id"] for r in admin_roles_response.json()]
+        user_roles_response_ids = [r["id"] for r in user_roles_response.json()]
+
+        # User can see their own private role not the other users, admin can see both.
+        assert user_role_id in user_roles_response_ids
+        assert different_user_role_id not in user_roles_response_ids
+
+        assert user_role_id in admin_roles_response_ids
+        assert different_user_role_id in admin_roles_response_ids
+
+        # Check showing a valid, role.
+        role_response = self._get("roles/%s" % user_role_id)
+        assert role_response.status_code == 200
+        role = role_response.json()
+        RolesApiTestCase.check_role_dict(role, assert_id=user_role_id)
+
+    def test_create_valid(self):
+        name = self.dataset_populator.get_random_name()
+        description = "A test role."
+        payload = {
+            "name": name,
+            "description": description,
+            "user_ids": json.dumps([self.dataset_populator.user_id()]),
+        }
+        response = self._post("roles", payload, admin=True)
+        assert_status_code_is(response, 200)
+        # TODO: Why does this return a singleton list - that is bad - should be deprecated
+        # and return a single role.
+        role = response.json()[0]
+        RolesApiTestCase.check_role_dict(role)
+
+        assert role["name"] == name
+        assert role["description"] == description
+
+        user_roles_response = self._get("roles")
+        with self._different_user():
+            different_user_roles_response = self._get("roles")
+
+        user_roles_response_ids = [r["id"] for r in user_roles_response.json()]
+        different_user_roles_response_ids = [r["id"] for r in different_user_roles_response.json()]
+
+        # This new role is public, all users see it.
+        assert role["id"] in user_roles_response_ids
+        assert role["id"] in different_user_roles_response_ids
+
+    def test_show_error_codes(self):
+        # Bad role ids are 400.
+        response = self._get("roles/badroleid")
+        assert_status_code_is(response, 400)
+
+        # Trying to access roles are errors - should probably be 403 not 400 though?
+        with self._different_user():
+            different_user_role_id = self.dataset_populator.user_private_role_id()
+        response = self._get("roles/%s" % different_user_role_id)
+        assert_status_code_is(response, 400)
+
+    def test_create_only_admin(self):
+        response = self._post("roles")
+        assert_status_code_is(response, 403)
+
+    @staticmethod
+    def check_role_dict(role_dict, assert_id=None):
+        assert_has_keys(role_dict, "id", "name", "model_class", "url")
+        assert role_dict["model_class"] == "Role"
+        if assert_id is not None:
+            assert role_dict["id"] == assert_id

test/api/test_tools.py

@@ -1,4 +1,5 @@
 # Test tools API.
+import contextlib
 import json
 import os
 
@@ -251,6 +252,37 @@ class ToolsTestCase(api.ApiTestCase):
             zipped_hdca = self.dataset_populator.get_history_collection_details(history_id, hid=output_collections[0]["hid"])
             assert zipped_hdca["collection_type"] == "paired"
 
+    @skip_without_tool("__ZIP_COLLECTION__")
+    @uses_test_history(require_new=False)
+    def test_collection_operation_dataset_input_permissions(self, history_id):
+        hda1 = dataset_to_param(self.dataset_populator.new_dataset(history_id, content='1\t2\t3'))
+        self.dataset_populator.wait_for_history(history_id, assert_ok=True)
+        self.dataset_populator.make_private(history_id, hda1["id"])
+        inputs = {
+            "input_forward": hda1,
+            "input_reverse": hda1,
+        }
+        with self._different_user_and_history() as other_history_id:
+            response = self._run("__ZIP_COLLECTION__", other_history_id, inputs, assert_ok=False)
+            self._assert_dataset_permission_denied_response(response)
+
+    @skip_without_tool("__UNZIP_COLLECTION__")
+    @uses_test_history(require_new=False)
+    def test_collection_operation_collection_input_permissions(self, history_id):
+        create_response = self.dataset_collection_populator.create_pair_in_history(history_id, direct_upload=True)
+        self._assert_status_code_is(create_response, 200)
+        collection = create_response.json()["outputs"][0]
+        self.dataset_populator.wait_for_history(history_id, assert_ok=True)
+        collection = self.dataset_populator.get_history_collection_details(history_id, hid=collection["hid"])
+        element_id = collection["elements"][0]["object"]["id"]
+        self.dataset_populator.make_private(history_id, element_id)
+        inputs = {
+            "input": {"src": "hdca", "id": collection["id"]},
+        }
+        with self._different_user_and_history() as other_history_id:
+            response = self._run("__UNZIP_COLLECTION__", other_history_id, inputs, assert_ok=False)
+            self._assert_dataset_permission_denied_response(response)
+
     def test_zip_list_inputs(self):
         with self.dataset_populator.test_history() as history_id:
             hdca1_id = self.dataset_collection_populator.create_list_in_history(history_id, contents=["a\nb\nc\nd", "e\nf\ng\nh"]).json()["id"]
@@ -496,6 +528,79 @@ class ToolsTestCase(api.ApiTestCase):
         output1_content = self.dataset_populator.get_history_dataset_content(history_id, dataset=output1)
         self.assertEqual(output1_content.strip(), "123")
 
+    @skip_without_tool("cat1")
+    @uses_test_history(require_new=False)
+    def test_guess_derived_permissions(self, history_id):
+
+        def assert_inputs(inputs, can_be_used=True):
+            # Until we make the dataset private, _different_user() can use it:
+            with self._different_user_and_history() as other_history_id:
+                response = self._run("cat1", other_history_id, inputs)
+                if can_be_used:
+                    assert response.status_code == 200
+                else:
+                    self._assert_dataset_permission_denied_response(response)
+
+        new_dataset = self.dataset_populator.new_dataset(history_id, content='Cat1Test')
+        inputs = dict(
+            input1=dataset_to_param(new_dataset),
+        )
+        # Until we make the dataset private, _different_user() can use it:
+        assert_inputs(inputs, can_be_used=True)
+        self.dataset_populator.make_private(history_id, new_dataset["id"])
+        # _different_user can no longer use the input dataset.
+        assert_inputs(inputs, can_be_used=False)
+
+        outputs = self._cat1_outputs(history_id, inputs=inputs)
+        self.assertEquals(len(outputs), 1)
+        output1 = outputs[0]
+
+        inputs_2 = dict(
+            input1=dataset_to_param(output1),
+        )
+        # _different_user cannot use datasets derived from the private input.
+        assert_inputs(inputs_2, can_be_used=False)
+
+    @skip_without_tool("collection_creates_list")
+    @uses_test_history(require_new=False)
+    def test_guess_derived_permissions_collections(self, history_id):
+        def first_element_dataset_id(hdca):
+            # Fetch full and updated details for HDCA
+            print(hdca)
+            full_hdca = self.dataset_populator.get_history_collection_details(history_id, hid=hdca["hid"])
+            elements = full_hdca["elements"]
+            element0 = elements[0]["object"]
+            return element0["id"]
+
+        response = self.dataset_collection_populator.create_list_in_history(history_id, contents=["a\nb\nc\nd", "e\nf\ng\nh"], direct_upload=True)
+        self._assert_status_code_is(response, 200)
+        hdca = response.json()["output_collections"][0]
+        self.dataset_populator.wait_for_history(history_id, assert_ok=True)
+        inputs = {
+            "input1": {"src": "hdca", "id": hdca["id"]},
+        }
+        public_output_response = self._run("collection_creates_list", history_id, inputs)
+        self._assert_status_code_is(public_output_response, 200)
+        self.dataset_populator.wait_for_history(history_id, assert_ok=True)
+
+        input_element_id = first_element_dataset_id(hdca)
+        self.dataset_populator.make_private(history_id, input_element_id)
+
+        private_output_response = self._run("collection_creates_list", history_id, inputs)
+        self._assert_status_code_is(private_output_response, 200)
+        self.dataset_populator.wait_for_history(history_id, assert_ok=True)
+
+        public_element_id = first_element_dataset_id(public_output_response.json()["output_collections"][0])
+        private_element_id = first_element_dataset_id(private_output_response.json()["output_collections"][0])
+
+        def _dataset_accessible(dataset_id):
+            contents_response = self._get("histories/%s/contents/%s" % (history_id, dataset_id)).json()
+            return "name" in contents_response
+
+        with self._different_user():
+            assert _dataset_accessible(public_element_id)
+            assert not _dataset_accessible(private_element_id)
+
     @skip_without_tool("validation_default")
     @uses_test_history(require_new=False)
     def test_validation(self, history_id):
@@ -1808,6 +1913,19 @@ class ToolsTestCase(api.ApiTestCase):
         hdca_id = create_response.json()["outputs"][0]["id"]
         return hdca_id
 
+    def _assert_dataset_permission_denied_response(self, response):
+        # TODO: This should be 403, should just need to throw more specific exception in the
+        # Galaxy code.
+        assert response.status_code != 200
+        # err_message = response.json()["err_msg"]
+        # assert "User does not have permission to use a dataset" in err_message, err_message
+
+    @contextlib.contextmanager
+    def _different_user_and_history(self):
+        with self._different_user():
+            with self.dataset_populator.test_history() as other_history_id:
+                yield other_history_id
+
 
 def dataset_to_param(dataset):
     return dict(