Mercurial > hg > orthanc-tests

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

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
new tests for auth-service
author Alain Mazy <am@osimis.io>
date Fri, 08 Sep 2023 12:03:50 +0200
parents 80ba6f1d521c
children c474f0f815b6
comparison
equal deleted inserted replaced
576:80ba6f1d521c 577:0649a19df194
19 label_b_study_id = None 19 label_b_study_id = None
20 no_label_study_id = None 20 no_label_study_id = None
21 auth_service_process = None 21 auth_service_process = None
22 22
23 @classmethod 23 @classmethod
24 def _terminate(cls): 24 def terminate(cls):
25 cls.auth_service_process.terminate() 25 cls.auth_service_process.terminate()
26 26
27 @classmethod 27 @classmethod
28 def prepare(cls): 28 def prepare(cls):
29 test_name = "Authorization" 29 test_name = "Authorization"
31 31
32 print(f'-------------- preparing {test_name} tests') 32 print(f'-------------- preparing {test_name} tests')
33 33
34 cls.clear_storage(storage_name=storage_name) 34 cls.clear_storage(storage_name=storage_name)
35 35
36 auth_service_hostname = "localhost"
37 if Helpers.is_docker():
38 auth_service_hostname = "auth-service"
39 cls.create_docker_network("auth-test-network")
40
36 config = { 41 config = {
37 "AuthenticationEnabled": False, 42 "AuthenticationEnabled": False,
38 "Authorization": { 43 "Authorization": {
39 "WebServiceRootUrl": "http://localhost:8020/", 44 "WebServiceRootUrl": f"http://{auth_service_hostname}:8020/",
40 "StandardConfigurations": [ 45 "StandardConfigurations": [
41 "orthanc-explorer-2", 46 "orthanc-explorer-2",
42 "stone-webviewer" 47 "stone-webviewer"
43 ], 48 ],
44 "CheckedLevel": "studies", 49 "CheckedLevel": "studies",
45 "TokenHttpHeaders": ["user-token-key"], 50 "TokenHttpHeaders": ["user-token-key", "resource-token-key"],
46 "TokenGetArguments": ["resource-token-key"] 51 "TokenGetArguments": ["resource-token-key"]
47 } 52 }
48 } 53 }
49 54
50 config_path = cls.generate_configuration( 55 config_path = cls.generate_configuration(
52 storage_name=storage_name, 57 storage_name=storage_name,
53 config=config, 58 config=config,
54 plugins=Helpers.plugins 59 plugins=Helpers.plugins
55 ) 60 )
56 61
57 # Start the auth-service application as a subprocess and wait for it to start 62 if Helpers.is_exe():
58 cls.auth_service_process = subprocess.Popen(["uvicorn", "auth_service:app", "--host", "0.0.0.0", "--port", "8020"], cwd=here) 63 # Start the auth-service application as a subprocess and wait for it to start
59 time.sleep(2) 64 cls.auth_service_process = subprocess.Popen(["uvicorn", "auth_service:app", "--host", "0.0.0.0", "--port", "8020"], cwd=here)
65 time.sleep(2)
66 else:
67 # first build the docker image for the auth-service
68 subprocess.run(["docker", "build", "-t", "auth-service", "."], cwd=here)
69 cls.auth_service_process = subprocess.Popen(["docker", "run", "-p", "8020:8020", "--network", "auth-test-network", "--name", "auth-service", "auth-service"])
70 pass
71
60 72
61 if Helpers.break_before_preparation: 73 if Helpers.break_before_preparation:
62 print(f"++++ It is now time to start your Orthanc under tests with configuration file '{config_path}' +++++") 74 print(f"++++ It is now time to start your Orthanc under tests with configuration file '{config_path}' +++++")
63 input("Press Enter to continue") 75 input("Press Enter to continue")
64 else: 76 else:
65 cls.launch_orthanc_under_tests( 77 cls.launch_orthanc_under_tests(
66 config_name=f"{test_name}", 78 config_name=f"{test_name}",
67 storage_name=storage_name, 79 storage_name=storage_name,
68 config=config, 80 config=config,
69 plugins=Helpers.plugins 81 plugins=Helpers.plugins,
82 docker_network="auth-test-network"
70 ) 83 )
71 84
72 uploader = OrthancApiClient(cls.o._root_url, headers={"user-token-key": "token-uploader"}) 85 uploader = OrthancApiClient(cls.o._root_url, headers={"user-token-key": "token-uploader"})
73 86
74 uploader.delete_all_content() 87 uploader.delete_all_content()
84 97
85 instances_ids = uploader.upload_file(here / "../../Database/Comunix/Pet/IM-0001-0001.dcm") 98 instances_ids = uploader.upload_file(here / "../../Database/Comunix/Pet/IM-0001-0001.dcm")
86 cls.no_label_study_id = uploader.instances.get_parent_study_id(instances_ids[0]) 99 cls.no_label_study_id = uploader.instances.get_parent_study_id(instances_ids[0])
87 100
88 101
102 def assert_is_forbidden(self, api_call):
103 with self.assertRaises(orthanc_exceptions.HttpError) as ctx:
104 api_call()
105 self.assertEqual(403, ctx.exception.http_status_code)
106
107
89 def test_admin_user(self): 108 def test_admin_user(self):
90 109
91 o = OrthancApiClient(self.o._root_url, headers={"user-token-key": "token-admin"}) 110 o = OrthancApiClient(self.o._root_url, headers={"user-token-key": "token-admin"})
92 111
93 # make sure we can access all these urls (they would throw if not) 112 # make sure we can access all these urls (they would throw if not)
129 all_labels = o.get_all_labels() 148 all_labels = o.get_all_labels()
130 self.assertEqual(1, len(all_labels)) 149 self.assertEqual(1, len(all_labels))
131 self.assertEqual("label_a", all_labels[0]) 150 self.assertEqual("label_a", all_labels[0])
132 151
133 # make sure we can access only the label_a studies 152 # make sure we can access only the label_a studies
134 with self.assertRaises(orthanc_exceptions.HttpError) as ctx: 153 self.assert_is_forbidden(lambda: o.studies.get_tags(self.label_b_study_id))
135 o.studies.get_tags(self.label_b_study_id) 154 self.assert_is_forbidden(lambda: o.studies.get_tags(self.no_label_study_id))
136 self.assertEqual(403, ctx.exception.http_status_code)
137
138 with self.assertRaises(orthanc_exceptions.HttpError) as ctx:
139 o.studies.get_tags(self.no_label_study_id)
140 self.assertEqual(403, ctx.exception.http_status_code)
141 155
142 # should not raise 156 # should not raise
143 o.studies.get_tags(self.label_a_study_id) 157 o.studies.get_tags(self.label_a_study_id)
144 158
145 # make sure we can access series and instances of the label_a studies 159 # make sure we can access series and instances of the label_a studies
146 series_ids = o.studies.get_series_ids(self.label_a_study_id) 160 series_ids = o.studies.get_series_ids(self.label_a_study_id)
147 instances_ids = o.series.get_instances_ids(series_ids[0]) 161 instances_ids = o.series.get_instances_ids(series_ids[0])
148 o.instances.get_tags(instances_ids[0]) 162 o.instances.get_tags(instances_ids[0])
149 163
150 # make sure we can not access series and instances of the label_b studies 164 # make sure we can not access series and instances of the label_b studies
151 with self.assertRaises(orthanc_exceptions.HttpError) as ctx: 165 self.assert_is_forbidden(lambda: o.studies.get_series_ids(self.label_b_study_id))
152 series_ids = o.studies.get_series_ids(self.label_b_study_id)
153 self.assertEqual(403, ctx.exception.http_status_code)
154 166
155 # make sure tools/find only returns the label_a studies 167 # make sure tools/find only returns the label_a studies
156 studies = o.studies.find(query={}, 168 studies = o.studies.find(query={},
157 labels=[], 169 labels=[],
158 labels_constraint='Any') 170 labels_constraint='Any')
165 labels_constraint='Any') 177 labels_constraint='Any')
166 self.assertEqual(1, len(studies)) 178 self.assertEqual(1, len(studies))
167 self.assertEqual(self.label_a_study_id, studies[0].orthanc_id) 179 self.assertEqual(self.label_a_study_id, studies[0].orthanc_id)
168 180
169 # if searching Any of label_b, expect a Forbidden access 181 # if searching Any of label_b, expect a Forbidden access
170 with self.assertRaises(orthanc_exceptions.HttpError) as ctx: 182 self.assert_is_forbidden(lambda: o.studies.find(query={},
171 studies = o.studies.find(query={}, 183 labels=['label_b'],
172 labels=['label_b'], 184 labels_constraint='Any'))
173 labels_constraint='Any')
174 self.assertEqual(403, ctx.exception.http_status_code)
175 185
176 # if searching None of label_b, expect a Forbidden access because we are not able to compute this filter 186 # if searching None of label_b, expect a Forbidden access because we are not able to compute this filter
177 with self.assertRaises(orthanc_exceptions.HttpError) as ctx: 187 self.assert_is_forbidden(lambda: o.studies.find(query={},
178 studies = o.studies.find(query={}, 188 labels=['label_b'],
179 labels=['label_b'], 189 labels_constraint='None'))
180 labels_constraint='None')
181 self.assertEqual(403, ctx.exception.http_status_code)
182 190
183 # if searching All of label_b, expect a Forbidden access because we are not able to compute this filter 191 # if searching All of label_b, expect a Forbidden access because we are not able to compute this filter
184 with self.assertRaises(orthanc_exceptions.HttpError) as ctx: 192 self.assert_is_forbidden(lambda: o.studies.find(query={},
185 studies = o.studies.find(query={}, 193 labels=['label_b'],
186 labels=['label_b'], 194 labels_constraint='All'))
187 labels_constraint='All')
188 self.assertEqual(403, ctx.exception.http_status_code)
189 195
190 studies = o.studies.find(query={"PatientName": "KNIX"}, # KNIX is label_a 196 studies = o.studies.find(query={"PatientName": "KNIX"}, # KNIX is label_a
191 labels=[], 197 labels=[],
192 labels_constraint='Any') 198 labels_constraint='Any')
193 self.assertEqual(1, len(studies)) 199 self.assertEqual(1, len(studies))
195 studies = o.studies.find(query={"PatientName": "KNIX"}, # KNIX is label_a 201 studies = o.studies.find(query={"PatientName": "KNIX"}, # KNIX is label_a
196 labels=['label_a'], 202 labels=['label_a'],
197 labels_constraint='Any') 203 labels_constraint='Any')
198 self.assertEqual(1, len(studies)) 204 self.assertEqual(1, len(studies))
199 205
200 with self.assertRaises(orthanc_exceptions.HttpError) as ctx: 206 self.assert_is_forbidden(lambda: o.studies.find(query={"PatientName": "KNIX"}, # KNIX is label_a
201 studies = o.studies.find(query={"PatientName": "KNIX"}, # KNIX is label_a 207 labels=['label_b'],
202 labels=['label_b'], 208 labels_constraint='Any'))
203 labels_constraint='Any') 209
204 self.assertEqual(403, ctx.exception.http_status_code) 210 # make sure some generic routes are not accessible
211 self.assert_is_forbidden(lambda: o.get_json('patients?expand'))
212 self.assert_is_forbidden(lambda: o.get_json('studies?expand'))
213 self.assert_is_forbidden(lambda: o.get_json('series?expand'))
214 self.assert_is_forbidden(lambda: o.get_json('instances?expand'))
215 self.assert_is_forbidden(lambda: o.get_json('studies'))
216 self.assert_is_forbidden(lambda: o.get_json('studies/'))
217
218
219
220 def test_resource_token(self):
221
222 o = OrthancApiClient(self.o._root_url, headers={"resource-token-key": "token-knix-study"})
223
224 # with a resource token, we can access only the given resource, not generic resources or resources from other studies
225
226 # generic resources are forbidden
227 self.assert_is_forbidden(lambda: o.studies.find(query={"PatientName": "KNIX"}, # KNIX is label_a
228 labels=['label_b'],
229 labels_constraint='Any'))
230 self.assert_is_forbidden(lambda: o.get_all_labels())
231 self.assert_is_forbidden(lambda: o.studies.get_all_ids())
232 self.assert_is_forbidden(lambda: o.patients.get_all_ids())
233 self.assert_is_forbidden(lambda: o.series.get_all_ids())
234 self.assert_is_forbidden(lambda: o.instances.get_all_ids())
235 self.assert_is_forbidden(lambda: o.get_json('patients?expand'))
236 self.assert_is_forbidden(lambda: o.get_json('studies?expand'))
237 self.assert_is_forbidden(lambda: o.get_json('series?expand'))
238 self.assert_is_forbidden(lambda: o.get_json('instances?expand'))
239
240 # some resources are still accessible to the 'anonymous' user -> does not throw
241 o.get_system()
242 o.lookup("1.2.3") # this route is still explicitely authorized because it is used by Stone
243
244 # other studies are forbidden
245 self.assert_is_forbidden(lambda: o.studies.get_series_ids(self.label_b_study_id))
246
247 # the label_a study is allowed
248 o.studies.get_series_ids(self.label_a_study_id)
249
250 # TODO: test with DicomWEB routes + sub-routes