Mercurial > hg > orthanc-tests
comparison 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 |
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 |