diff --git a/src/medcheck/providers/easyradiology.py b/src/medcheck/providers/easyradiology.py index 567059b..00b2595 100644 --- a/src/medcheck/providers/easyradiology.py +++ b/src/medcheck/providers/easyradiology.py @@ -110,6 +110,13 @@ def _decrypt_aes_cbc(ciphertext_b64: str, password: str, salt_hex: str, iv_b64: salt = bytes.fromhex(salt_hex) iv = b64decode(iv_b64) ct = b64decode(ciphertext_b64) + # Validate up front: pycryptodome raises an opaque block-size ValueError on + # a non-multiple-of-16 body, and empty input would crash the padding check + # below with an IndexError. AES-CBC here is unauthenticated (the protocol + # is server-dictated, no MAC), so the manual PKCS7 check is the only + # integrity signal and must be robust against malformed input. + if not ct or len(ct) % 16 != 0: + raise ValueError(f"AES-CBC decryption failed: ciphertext length {len(ct)} is not a positive multiple of 16") key = _scrypt_derive(password, salt) cipher = AES.new(key, AES.MODE_CBC, iv) dec = cipher.decrypt(ct) @@ -119,6 +126,13 @@ def _decrypt_aes_cbc(ciphertext_b64: str, password: str, salt_hex: str, iv_b64: raise ValueError("AES-CBC decryption failed: invalid PKCS7 padding") +def _unexpected_response(step: str, detail: str) -> RuntimeError: + """Provider-level error for a portal response that doesn't match the protocol.""" + return RuntimeError( + f"Unexpected easyRadiology portal response during {step}: {detail}. The portal API may have changed." + ) + + class EasyRadiologyProvider(DataProvider): name = "easyradiology" url_patterns: ClassVar[list[str]] = ["easyradiology.net", "easyradiology.de"] @@ -132,11 +146,15 @@ def authenticate(self, credentials: dict[str, str]) -> bool: return bool(credentials.get("code")) def fetch(self, target: str, credentials: dict[str, str]) -> list[DicomSeries]: - code = credentials["code"] + code = credentials.get("code") + if not code: + raise ValueError("An access code is required to fetch from the easyRadiology portal") exam_hash = self._extract_exam_hash(target) access_key = self._check_view_code(code, exam_hash) viewer_model = self._get_viewer_model(exam_hash) - zip_url = viewer_model["linkToERI"] + zip_url = viewer_model.get("linkToERI") + if not isinstance(zip_url, str) or not zip_url: + raise _unexpected_response("GetViewerModel", "missing download link (linkToERI)") return self._download_and_decrypt(zip_url, access_key, exam_hash) def _extract_exam_hash(self, url: str) -> str: @@ -153,22 +171,47 @@ def _check_view_code(self, code: str, exam_hash: str) -> str: json={"ViewCodeName": view_code, "KeyVerification": key_verification, "WithDeferred": True}, ) resp.raise_for_status() - data = resp.json() - if not data.get("exams"): + try: + data = resp.json() + except ValueError as exc: + raise _unexpected_response("CheckViewCode", "response body is not JSON") from exc + if not isinstance(data, dict) or not data.get("exams"): raise ValueError("No exams found for the provided access code") - exam = data["exams"][0] - enc = json.loads(exam["encryptedAccessKey"]) - access_key_bytes = _decrypt_aes_cbc(enc["CipherOutputText"], code, enc["Salt"], enc["AesRijndaelIv"]) - return access_key_bytes.decode("utf-8") + try: + enc = json.loads(data["exams"][0]["encryptedAccessKey"]) + ciphertext = enc["CipherOutputText"] + salt_hex = enc["Salt"] + iv_b64 = enc["AesRijndaelIv"] + except (KeyError, IndexError, TypeError, ValueError) as exc: + raise _unexpected_response("CheckViewCode", f"malformed exam entry ({exc.__class__.__name__})") from exc + try: + access_key_bytes = _decrypt_aes_cbc(ciphertext, code, salt_hex, iv_b64) + return access_key_bytes.decode("utf-8") + except (ValueError, TypeError) as exc: + # Padding/decode failures usually mean a wrong access code, not a + # protocol change — keep this distinct from _unexpected_response. + raise ValueError( + "Could not decrypt the exam access key — the access code may be wrong or the link expired" + ) from exc def _get_viewer_model(self, exam_hash: str) -> dict[str, Any]: with httpx.Client(timeout=30.0) as client: resp = client.post(f"{self.BASE_URL}/api/viewexam/GetViewerModel", json=[{"ExamHash": exam_hash}]) resp.raise_for_status() - data = resp.json() + try: + data = resp.json() + except ValueError as exc: + raise _unexpected_response("GetViewerModel", "response body is not JSON") from exc + if not isinstance(data, dict): + raise _unexpected_response("GetViewerModel", f"expected a JSON object, got {type(data).__name__}") if data.get("hasError"): raise ValueError(f"Viewer model error: {data.get('errorMessage', '')}") - result: dict[str, Any] = data["exams"][0] + try: + result = data["exams"][0] + except (KeyError, IndexError, TypeError) as exc: + raise _unexpected_response("GetViewerModel", f"missing exam entry ({exc.__class__.__name__})") from exc + if not isinstance(result, dict): + raise _unexpected_response("GetViewerModel", f"exam entry is {type(result).__name__}, expected object") return result def _download_and_decrypt(self, zip_url: str, access_key: str, exam_hash: str) -> list[DicomSeries]: diff --git a/tests/unit/test_providers/test_easyradiology.py b/tests/unit/test_providers/test_easyradiology.py index fa8f242..89f9c52 100644 --- a/tests/unit/test_providers/test_easyradiology.py +++ b/tests/unit/test_providers/test_easyradiology.py @@ -207,3 +207,120 @@ def test_validate_download_url_accepts_trusted_https(url): def test_validate_download_url_rejects_untrusted(url): with pytest.raises(ValueError): _validate_download_url(url) + + +# --------------------------------------------------------------------------- +# Portal orchestration (mocked httpx) — #139 +# --------------------------------------------------------------------------- + +VALID_CODE = "dz8d-zkg9-4hzc-aosn" + + +def _mock_httpx_client_post(json_data=None, json_exc=None) -> MagicMock: + """Build a MagicMock standing in for httpx.Client whose post() returns *json_data*.""" + resp = MagicMock() + resp.raise_for_status.return_value = None + if json_exc is not None: + resp.json.side_effect = json_exc + else: + resp.json.return_value = json_data + client = MagicMock() + client.post.return_value = resp + client_cm = MagicMock() + client_cm.__enter__ = MagicMock(return_value=client) + client_cm.__exit__ = MagicMock(return_value=False) + return client_cm + + +def _check_view_code_with(json_data=None, json_exc=None) -> str: + provider = EasyRadiologyProvider() + client_cm = _mock_httpx_client_post(json_data=json_data, json_exc=json_exc) + with patch("medcheck.providers.easyradiology.httpx.Client", return_value=client_cm): + return provider._check_view_code(VALID_CODE, "examhash") + + +def test_check_view_code_non_json_body(): + with pytest.raises(RuntimeError, match="not JSON"): + _check_view_code_with(json_exc=ValueError("Expecting value")) + + +def test_check_view_code_empty_exams(): + with pytest.raises(ValueError, match="No exams found"): + _check_view_code_with(json_data={"exams": []}) + + +def test_check_view_code_non_dict_payload(): + with pytest.raises(ValueError, match="No exams found"): + _check_view_code_with(json_data=[1, 2, 3]) + + +def test_check_view_code_missing_encrypted_key(): + with pytest.raises(RuntimeError, match="portal API may have changed"): + _check_view_code_with(json_data={"exams": [{"somethingElse": True}]}) + + +def test_check_view_code_undecryptable_key_hints_wrong_code(): + import json as jsonlib + + enc = jsonlib.dumps( + { + "CipherOutputText": b64encode(b"x" * 16).decode(), # garbage ciphertext + "Salt": "00" * 16, + "AesRijndaelIv": b64encode(b"\x00" * 16).decode(), + } + ) + with pytest.raises(ValueError, match="access code may be wrong"): + _check_view_code_with(json_data={"exams": [{"encryptedAccessKey": enc}]}) + + +def _get_viewer_model_with(json_data=None, json_exc=None): + provider = EasyRadiologyProvider() + client_cm = _mock_httpx_client_post(json_data=json_data, json_exc=json_exc) + with patch("medcheck.providers.easyradiology.httpx.Client", return_value=client_cm): + return provider._get_viewer_model("examhash") + + +def test_get_viewer_model_non_json_body(): + with pytest.raises(RuntimeError, match="not JSON"): + _get_viewer_model_with(json_exc=ValueError("Expecting value")) + + +def test_get_viewer_model_error_flag(): + with pytest.raises(ValueError, match="Viewer model error"): + _get_viewer_model_with(json_data={"hasError": True, "errorMessage": "expired"}) + + +def test_get_viewer_model_missing_exams(): + with pytest.raises(RuntimeError, match="missing exam entry"): + _get_viewer_model_with(json_data={"hasError": False}) + + +def test_get_viewer_model_non_object_exam(): + with pytest.raises(RuntimeError, match="expected object"): + _get_viewer_model_with(json_data={"exams": ["not-a-dict"]}) + + +def test_fetch_requires_access_code(): + provider = EasyRadiologyProvider() + with pytest.raises(ValueError, match="access code is required"): + provider.fetch("https://portal.easyradiology.net/View/abc", {}) + + +def test_fetch_missing_download_link(): + provider = EasyRadiologyProvider() + with ( + patch.object(provider, "_check_view_code", return_value="access-key"), + patch.object(provider, "_get_viewer_model", return_value={"noLink": True}), + ): + with pytest.raises(RuntimeError, match="linkToERI"): + provider.fetch("https://portal.easyradiology.net/View/abc", {"code": VALID_CODE}) + + +def test_decrypt_aes_cbc_empty_ciphertext(): + with pytest.raises(ValueError, match="positive multiple of 16"): + _decrypt_aes_cbc(b64encode(b"").decode(), "pw", "00" * 16, b64encode(b"\x00" * 16).decode()) + + +def test_decrypt_aes_cbc_partial_block(): + with pytest.raises(ValueError, match="positive multiple of 16"): + _decrypt_aes_cbc(b64encode(b"short").decode(), "pw", "00" * 16, b64encode(b"\x00" * 16).decode())