diff NewTests/Authorization/test_authorization.py @ 577:0649a19df194

new tests for auth-service
author Alain Mazy <am@osimis.io>
date Fri, 08 Sep 2023 12:03:50 +0200
parents 80ba6f1d521c
children c474f0f815b6
line wrap: on
line diff
--- a/NewTests/Authorization/test_authorization.py	Wed Sep 06 17:04:36 2023 +0200
+++ b/NewTests/Authorization/test_authorization.py	Fri Sep 08 12:03:50 2023 +0200
@@ -21,7 +21,7 @@
     auth_service_process = None
 
     @classmethod
-    def _terminate(cls):
+    def terminate(cls):
         cls.auth_service_process.terminate()
 
     @classmethod
@@ -33,16 +33,21 @@
 
         cls.clear_storage(storage_name=storage_name)
 
+        auth_service_hostname = "localhost"
+        if Helpers.is_docker():
+            auth_service_hostname = "auth-service"
+            cls.create_docker_network("auth-test-network")
+
         config = {
                 "AuthenticationEnabled": False,
                 "Authorization": {
-                    "WebServiceRootUrl": "http://localhost:8020/",
+                    "WebServiceRootUrl": f"http://{auth_service_hostname}:8020/",
                     "StandardConfigurations": [
                         "orthanc-explorer-2",
                         "stone-webviewer"
                     ],
                     "CheckedLevel": "studies",
-                    "TokenHttpHeaders": ["user-token-key"],
+                    "TokenHttpHeaders": ["user-token-key", "resource-token-key"],
                     "TokenGetArguments": ["resource-token-key"]
                 }
             }
@@ -54,9 +59,16 @@
             plugins=Helpers.plugins
         )
 
-        # Start the auth-service application as a subprocess and wait for it to start
-        cls.auth_service_process = subprocess.Popen(["uvicorn", "auth_service:app", "--host", "0.0.0.0", "--port", "8020"], cwd=here)
-        time.sleep(2)
+        if Helpers.is_exe():
+            # Start the auth-service application as a subprocess and wait for it to start
+            cls.auth_service_process = subprocess.Popen(["uvicorn", "auth_service:app", "--host", "0.0.0.0", "--port", "8020"], cwd=here)
+            time.sleep(2)
+        else:
+            # first build the docker image for the auth-service
+            subprocess.run(["docker", "build", "-t", "auth-service", "."], cwd=here)
+            cls.auth_service_process = subprocess.Popen(["docker", "run", "-p", "8020:8020", "--network", "auth-test-network", "--name", "auth-service", "auth-service"])
+            pass
+
 
         if Helpers.break_before_preparation:
             print(f"++++ It is now time to start your Orthanc under tests with configuration file '{config_path}' +++++")
@@ -66,7 +78,8 @@
                 config_name=f"{test_name}",
                 storage_name=storage_name,
                 config=config,
-                plugins=Helpers.plugins
+                plugins=Helpers.plugins,
+                docker_network="auth-test-network"
             )
 
         uploader = OrthancApiClient(cls.o._root_url, headers={"user-token-key": "token-uploader"})
@@ -86,6 +99,12 @@
         cls.no_label_study_id = uploader.instances.get_parent_study_id(instances_ids[0])
 
 
+    def assert_is_forbidden(self, api_call):
+        with self.assertRaises(orthanc_exceptions.HttpError) as ctx:
+            api_call()
+        self.assertEqual(403, ctx.exception.http_status_code)
+
+
     def test_admin_user(self):
         
         o = OrthancApiClient(self.o._root_url, headers={"user-token-key": "token-admin"})
@@ -131,13 +150,8 @@
         self.assertEqual("label_a", all_labels[0])
 
         # make sure we can access only the label_a studies
-        with self.assertRaises(orthanc_exceptions.HttpError) as ctx:
-            o.studies.get_tags(self.label_b_study_id)
-        self.assertEqual(403, ctx.exception.http_status_code)
-
-        with self.assertRaises(orthanc_exceptions.HttpError) as ctx:
-            o.studies.get_tags(self.no_label_study_id)
-        self.assertEqual(403, ctx.exception.http_status_code)
+        self.assert_is_forbidden(lambda: o.studies.get_tags(self.label_b_study_id))
+        self.assert_is_forbidden(lambda: o.studies.get_tags(self.no_label_study_id))
 
         # should not raise
         o.studies.get_tags(self.label_a_study_id)
@@ -148,9 +162,7 @@
         o.instances.get_tags(instances_ids[0])
 
         # make sure we can not access series and instances of the label_b studies
-        with self.assertRaises(orthanc_exceptions.HttpError) as ctx:
-            series_ids = o.studies.get_series_ids(self.label_b_study_id)
-        self.assertEqual(403, ctx.exception.http_status_code)
+        self.assert_is_forbidden(lambda: o.studies.get_series_ids(self.label_b_study_id))
 
         # make sure tools/find only returns the label_a studies
         studies = o.studies.find(query={},
@@ -167,25 +179,19 @@
         self.assertEqual(self.label_a_study_id, studies[0].orthanc_id)
 
         # if searching Any of label_b, expect a Forbidden access
-        with self.assertRaises(orthanc_exceptions.HttpError) as ctx:
-            studies = o.studies.find(query={},
-                                     labels=['label_b'],
-                                     labels_constraint='Any')
-        self.assertEqual(403, ctx.exception.http_status_code)
+        self.assert_is_forbidden(lambda: o.studies.find(query={},
+                                                        labels=['label_b'],
+                                                        labels_constraint='Any'))
 
         # if searching None of label_b, expect a Forbidden access because we are not able to compute this filter
-        with self.assertRaises(orthanc_exceptions.HttpError) as ctx:
-            studies = o.studies.find(query={},
-                                     labels=['label_b'],
-                                     labels_constraint='None')
-        self.assertEqual(403, ctx.exception.http_status_code)
+        self.assert_is_forbidden(lambda: o.studies.find(query={},
+                                                        labels=['label_b'],
+                                                        labels_constraint='None'))
 
         # if searching All of label_b, expect a Forbidden access because we are not able to compute this filter
-        with self.assertRaises(orthanc_exceptions.HttpError) as ctx:
-            studies = o.studies.find(query={},
-                                     labels=['label_b'],
-                                     labels_constraint='All')
-        self.assertEqual(403, ctx.exception.http_status_code)
+        self.assert_is_forbidden(lambda: o.studies.find(query={},
+                                                        labels=['label_b'],
+                                                        labels_constraint='All'))
 
         studies = o.studies.find(query={"PatientName": "KNIX"},  # KNIX is label_a
                                  labels=[],
@@ -197,8 +203,48 @@
                                  labels_constraint='Any')
         self.assertEqual(1, len(studies))
 
-        with self.assertRaises(orthanc_exceptions.HttpError) as ctx:
-            studies = o.studies.find(query={"PatientName": "KNIX"},  # KNIX is label_a
-                                     labels=['label_b'],
-                                     labels_constraint='Any')
-        self.assertEqual(403, ctx.exception.http_status_code)
\ No newline at end of file
+        self.assert_is_forbidden(lambda: o.studies.find(query={"PatientName": "KNIX"},  # KNIX is label_a
+                                                        labels=['label_b'],
+                                                        labels_constraint='Any'))
+
+        # make sure some generic routes are not accessible
+        self.assert_is_forbidden(lambda: o.get_json('patients?expand'))
+        self.assert_is_forbidden(lambda: o.get_json('studies?expand'))
+        self.assert_is_forbidden(lambda: o.get_json('series?expand'))
+        self.assert_is_forbidden(lambda: o.get_json('instances?expand'))
+        self.assert_is_forbidden(lambda: o.get_json('studies'))
+        self.assert_is_forbidden(lambda: o.get_json('studies/'))
+
+
+
+    def test_resource_token(self):
+
+        o = OrthancApiClient(self.o._root_url, headers={"resource-token-key": "token-knix-study"})
+
+        # with a resource token, we can access only the given resource, not generic resources or resources from other studies
+
+        # generic resources are forbidden
+        self.assert_is_forbidden(lambda: o.studies.find(query={"PatientName": "KNIX"},  # KNIX is label_a
+                                                        labels=['label_b'],
+                                                        labels_constraint='Any'))
+        self.assert_is_forbidden(lambda: o.get_all_labels())
+        self.assert_is_forbidden(lambda: o.studies.get_all_ids())
+        self.assert_is_forbidden(lambda: o.patients.get_all_ids())
+        self.assert_is_forbidden(lambda: o.series.get_all_ids())
+        self.assert_is_forbidden(lambda: o.instances.get_all_ids())
+        self.assert_is_forbidden(lambda: o.get_json('patients?expand'))
+        self.assert_is_forbidden(lambda: o.get_json('studies?expand'))
+        self.assert_is_forbidden(lambda: o.get_json('series?expand'))
+        self.assert_is_forbidden(lambda: o.get_json('instances?expand'))
+        
+        # some resources are still accessible to the 'anonymous' user  -> does not throw
+        o.get_system()
+        o.lookup("1.2.3")   # this route is still explicitely authorized because it is used by Stone
+
+        # other studies are forbidden
+        self.assert_is_forbidden(lambda: o.studies.get_series_ids(self.label_b_study_id))
+
+        # the label_a study is allowed
+        o.studies.get_series_ids(self.label_a_study_id)
+
+        # TODO: test with DicomWEB routes + sub-routes
\ No newline at end of file