fix(providers): harden easyRadiology portal flow against malformed responses

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]:

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())