Extended dicom

README.md

@@ -253,6 +253,20 @@ qsshserver.kill_server()
     - Port: 11112/tcp
     - Lib: pynetdicom
     - Logs: ip, port, events
+    - Custom configuration (through `config.json`):
+      - `"store_images"` (`bool`, default: `False`): start a storage SCP that is able to receive image files
+      - `"storage_dir"` (`str`, default: `"/tmp/dicom_storage"`): the storage directory where received image files are stored
+      - example:
+        ```json
+        {
+          "honeypots": {
+            "dicom": {
+              "store_images": true,
+              "storage_dir": "/tmp/dicom_storage"
+            }
+          }
+        }
+        ```
 - QFTPServer
     - Server: FTP 
     - Port: 21/tcp
@@ -262,7 +276,7 @@ qsshserver.kill_server()
 - HL7Server
     - Server: HL7 
     - Port: 2575/tcp
-    - Lib: socketserver
+    - Lib: hl7apy
     - Logs: ip, port and data
 - QHTTPProxyServer
     - Server: HTTP Proxy

honeypots/__main__.py

@@ -163,9 +163,15 @@ def __init__(self, options: Namespace, server_args: dict[str, str | int]):
         self.options = options
         self.server_args = server_args
         self.config_data: dict = load_config(self.options.config) if self.options.config else {}
+        self._validate_config()
         self.auto = options.auto if geteuid() != 0 else False
         self.honeypots: list[tuple[Any, str, bool]] = []
 
+    def _validate_config(self):
+        for server in self.config_data.get("honeypots", []):
+            if server not in all_servers:
+                logger.warning(f'Config file contains unknown server: "{server}" (ignored)')
+
     def main(self):
         logger.info("For updates, check https://github.com/qeeqbox/honeypots")
         if not is_privileged():

honeypots/dicom_server.py

@@ -12,9 +12,11 @@
 
 from contextlib import suppress
 from enum import Enum
+from pathlib import Path
 from typing import TYPE_CHECKING
 from unittest.mock import patch
 
+from pydicom.filewriter import write_file_meta_info
 from pynetdicom import (
     ae,
     ALL_TRANSFER_SYNTAXES,
@@ -50,13 +52,23 @@ class UserIdType(Enum):
 
 
 SUCCESS = 0x0000
+CANNOT_UNDERSTAND = 0xC000
 
 
 class QDicomServer(BaseServer):
     NAME = "dicom_server"
     DEFAULT_PORT = 11112
 
-    def server_main(self):  # noqa: C901
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
+        self.store_images = bool(getattr(self, "store_images", False))
+        if hasattr(self, "storage_dir") and isinstance(self.storage_dir, str):
+            self.storage_dir = Path(self.storage_dir)
+        else:
+            self.storage_dir = Path("/tmp/dicom_storage")
+        self.storage_dir.mkdir(parents=True, exist_ok=True)
+
+    def server_main(self):  # noqa: C901,PLR0915
         _q_s = self
 
         class CustomDUL(DULServiceProvider):
@@ -98,7 +110,28 @@ def process_request(
                 )
                 super().process_request(request, client_address)
 
-        def handle_login(event: evt.Event, *_) -> tuple[bool, bytes | None]:
+        def handle_store(event: evt.Event) -> int:
+            handle_event(event)  # for logging
+            try:
+                output_file = _q_s.storage_dir / event.request.AffectedSOPInstanceUID
+                with output_file.open("wb") as fp:
+                    preamble = b"\x00" * 128
+                    prefix = b"DICM"
+                    fp.write(preamble + prefix)
+                    write_file_meta_info(fp, event.file_meta)
+                    fp.write(event.request.DataSet.getvalue())
+                _q_s.log(
+                    {
+                        "action": "store_image",
+                        "data": {"path": str(output_file), "size": output_file.stat().st_size},
+                    }
+                )
+                return SUCCESS
+            except Exception as error:
+                _q_s.logger.critical(f"Exception occurred during store event: {error}")
+                return CANNOT_UNDERSTAND
+
+        def handle_login(event: evt.Event) -> tuple[bool, bytes | None]:
             # EVT_USER_ID event
             # see https://pydicom.github.io/pynetdicom/stable/reference/generated/pynetdicom._handlers.doc_handle_userid.html
             user_id_type = UserIdType(event.user_id_type)
@@ -165,11 +198,14 @@ def handle_event(event: evt.Event, *_):
                     }
                 )
             except Exception as error:
-                _q_s.logger.critical(f"exception during event logging: {error}")
+                _q_s.logger.debug(f"exception during event logging: {error}")
             return SUCCESS
 
+        special_handlers = {"EVT_USER_ID": handle_login}
+        if _q_s.store_images:
+            special_handlers["EVT_C_STORE"] = handle_store
         handlers = [
-            (event_, handle_event) if event_.name != "EVT_USER_ID" else (event_, handle_login)
+            (event_, special_handlers.get(event_.name, handle_event))
             for event_ in evt._INTERVENTION_EVENTS
             if event_.name not in UNINTERESTING_EVENTS
         ]

honeypots/helper.py

@@ -201,6 +201,7 @@ def get_free_port() -> int:
         tcp.bind(("", 0))
         _, port = tcp.getsockname()
         tcp.close()
+        return port
     except OSError:
         logger.error("Could not get a free port")
         return 0

tests/test_dicom_server.py

@@ -1,18 +1,22 @@
 from __future__ import annotations
 
+from pathlib import Path
 from shlex import split
 from subprocess import run
 from time import sleep
 
 import pytest
+from pydicom import dcmread, Dataset
+from pydicom.uid import CTImageStorage
 from pynetdicom.sop_class import PatientRootQueryRetrieveInformationModelFind
 from pynetdicom.pdu_primitives import UserIdentityNegotiation
-from pynetdicom import AE
+from pynetdicom import AE, tests
 
 from honeypots import QDicomServer
 from .utils import (
     assert_connect_is_logged,
     assert_login_is_logged,
+    IP,
     load_logs_from_file,
     PASSWORD,
     USERNAME,
@@ -31,7 +35,7 @@
 )
 def test_dicom_echo(server_logs):
     with wait_for_server(PORT):
-        proc = run(split(f"python -m pynetdicom echoscu 127.0.0.1 {PORT}"), check=False)
+        proc = run(split(f"python -m pynetdicom echoscu {IP} {PORT}"), check=False)
 
     assert proc.returncode == 0
 
@@ -62,15 +66,8 @@ def test_login(server_logs):
     user_identity.secondary_field = PASSWORD.encode()
 
     with wait_for_server(PORT + 1):
-        for _ in range(RETRIES):
-            # this is somehow a bit flaky so we retry here
-            assoc = ae.associate("127.0.0.1", PORT + 1, ext_neg=[user_identity])
-            if assoc.is_established:
-                assoc.release()
-                break
-            sleep(1)
-        else:
-            pytest.fail("could not establish connection")
+        association = _retry_association(ae, PORT + 1, ext_neg=[user_identity])
+        association.release()
 
     logs = load_logs_from_file(server_logs)
 
@@ -81,3 +78,48 @@ def test_login(server_logs):
 
     assert associate["action"] == "A-ASSOCIATE-RQ"
     assert release["action"] == "A-RELEASE-RQ"
+
+
+def _retry_association(ae: AE, port: int, ext_neg=None):
+    for _ in range(RETRIES):
+        # this is somehow a bit flaky so we retry here
+        association = ae.associate(IP, port, ext_neg=ext_neg)
+        if association.is_established:
+            return association
+        sleep(1)
+    pytest.fail("could not establish connection")
+
+
+SERVER_CONFIG = {
+    "honeypots": {
+        "dicom": {
+            "store_images": True,
+        },
+    }
+}
+
+
+@pytest.mark.parametrize(
+    "server_logs",
+    [{"server": QDicomServer, "port": PORT + 2, "custom_config": SERVER_CONFIG}],
+    indirect=True,
+)
+def test_store(server_logs):
+    dataset_path = Path(tests.__file__).parent / "dicom_files" / "CTImageStorage.dcm"
+    dataset = dcmread(dataset_path)
+
+    ae = AE()
+    ae.add_requested_context(CTImageStorage)
+    with wait_for_server(PORT + 2):
+        association = _retry_association(ae, PORT + 2)
+        response = association.send_c_store(dataset)
+        association.release()
+
+    assert isinstance(response, Dataset)
+    assert response.Status == 0x0000  # success
+
+    logs = load_logs_from_file(server_logs)
+    assert len(logs) > 1
+    logs_by_action = {entry["action"]: entry for entry in logs}
+    assert "store_image" in logs_by_action
+    assert logs_by_action["store_image"]["data"]["size"] == "39102"