Validate DICOM images when saving to PACS (#15)

carlosmartinez · web-flow · commit d2b755ee4cda · 2026-01-23T21:37:26.000Z
If modalities are sending invalid DICOM to the gateway, or if our processing (compression/resizing) is causing the DICOM to be invalid, we prefer to know about it sooner rather than later. This is a somewhat naive but fast implementation. If we find that we need something more comprehensive, we could use pydicom's [dicom-validator](https://pydicom.github.io/dicom-validator).
diff --git a/src/services/dicom/c_store.py b/src/services/dicom/c_store.py
@@ -10,6 +10,7 @@
 
 from services.dicom import FAILURE, SUCCESS
 from services.dicom.image_compressor import ImageCompressor
+from services.dicom.validator import DicomValidationError, DicomValidator
 from services.storage import InstanceExistsError, PACSStorage
 
 logger = logging.getLogger(__name__)
@@ -21,9 +22,15 @@ class CStore:
         DigitalMammographyXRayImageStorageForProcessing,
     ]
 
-    def __init__(self, storage: PACSStorage, compressor: ImageCompressor | None = None):
+    def __init__(
+        self,
+        storage: PACSStorage,
+        compressor: ImageCompressor | None = None,
+        validator: DicomValidator | None = None,
+    ):
         self.storage = storage
         self.compressor = compressor or ImageCompressor()
+        self.validator = validator or DicomValidator()
 
     def call(self, event: Event) -> int:
         try:
@@ -47,12 +54,28 @@ def call(self, event: Event) -> int:
             accession_number = ds.get("AccessionNumber", "")
             patient_name = str(ds.get("PatientName", ""))
 
+            # Validate dataset before compression
+            try:
+                self.validator.validate_dataset(ds)
+                self.validator.validate_pixel_data(ds)
+            except DicomValidationError as e:
+                logger.error(f"DICOM validation failed: {e}")
+                return FAILURE
+
             # Compress dataset before storing
             compressed_ds = self.compressor.compress(ds)
 
+            # Serialize and validate output
+            dicom_bytes = self.dataset_to_bytes(compressed_ds)
+            try:
+                self.validator.validate_bytes(dicom_bytes)
+            except DicomValidationError as e:
+                logger.error(f"Serialized DICOM invalid: {e}")
+                return FAILURE
+
             self.storage.store_instance(
                 sop_instance_uid,
-                self.dataset_to_bytes(compressed_ds),
+                dicom_bytes,
                 {
                     "accession_number": accession_number,
                     "patient_id": patient_id,
diff --git a/src/services/dicom/validator.py b/src/services/dicom/validator.py
@@ -0,0 +1,46 @@
+"""DICOM Validation utilities."""
+
+import logging
+
+from pydicom import Dataset
+
+logger = logging.getLogger(__name__)
+
+
+class DicomValidationError(Exception):
+    """Raised when DICOM validation fails."""
+
+    pass
+
+
+class DicomValidator:
+    REQUIRED_TAGS = ["SOPInstanceUID", "PatientID", "StudyInstanceUID", "SOPClassUID"]
+    DICOM_PREFIX = b"DICM"
+    PREAMBLE_LENGTH = 128
+
+    def validate_dataset(self, ds: Dataset) -> None:
+        """Validate dataset has required DICOM tags."""
+        for tag in self.REQUIRED_TAGS:
+            value = ds.get(tag)
+            if not value:
+                raise DicomValidationError(f"Missing required tag: {tag}")
+
+    def validate_bytes(self, data: bytes) -> None:
+        """Validate serialized DICOM bytes have valid preamble."""
+        min_size = self.PREAMBLE_LENGTH + len(self.DICOM_PREFIX)
+        if len(data) < min_size:
+            raise DicomValidationError(f"DICOM too small ({len(data)} bytes), missing preamble")
+
+        preamble = data[self.PREAMBLE_LENGTH : self.PREAMBLE_LENGTH + 4]
+        if preamble != self.DICOM_PREFIX:
+            raise DicomValidationError(f"Invalid DICOM prefix: {preamble!r}, expected {self.DICOM_PREFIX!r}")
+
+    def validate_pixel_data(self, ds: Dataset) -> None:
+        """Validate pixel data consistency if present."""
+        if not hasattr(ds, "PixelData") or ds.PixelData is None:
+            return  # No pixel data to validate
+
+        required_image_tags = ["Rows", "Columns", "BitsAllocated"]
+        for tag in required_image_tags:
+            if not hasattr(ds, tag):
+                raise DicomValidationError(f"Image has PixelData but missing {tag}")
diff --git a/tests/integration/test_c_store_saves_metadata.py b/tests/integration/test_c_store_saves_metadata.py
@@ -21,6 +21,8 @@ def mock_event(self):
         dataset.AccessionNumber = "ABC123"
         dataset.PatientID = "9990001112"
         dataset.SOPInstanceUID = "1.2.3.4.5.6"  # gitleaks:allow
+        dataset.StudyInstanceUID = "1.2.3.4.5.6.7"  # gitleaks:allow
+        dataset.SOPClassUID = "1.2.840.10008.5.1.4.1.1.1.2"  # gitleaks:allow
         file_meta = FileMetaDataset()
         file_meta.TransferSyntaxUID = ExplicitVRLittleEndian
         file_meta.MediaStorageSOPClassUID = DigitalMammographyXRayImageStorageForProcessing
@@ -89,6 +91,8 @@ def test_compressed_image_stored_on_filesystem(self, storage, dataset_with_pixel
         dataset_with_pixels.AccessionNumber = "DEF456"
         dataset_with_pixels.PatientID = "9990002223"
         dataset_with_pixels.SOPInstanceUID = "1.2.3.4.5.7"  # gitleaks:allow
+        dataset_with_pixels.StudyInstanceUID = "1.2.3.4.5.7.8"  # gitleaks:allow
+        dataset_with_pixels.SOPClassUID = "1.2.840.10008.5.1.4.1.1.1.2"  # gitleaks:allow
 
         # Wrap in PropertyMock to simulate DICOM event
         event = PropertyMock()
diff --git a/tests/services/dicom/test_c_store.py b/tests/services/dicom/test_c_store.py
@@ -19,6 +19,8 @@ def mock_event(self, dataset_with_pixels):
         dataset_with_pixels.SOPInstanceUID = "1.2.3.4.5.6"  # gitleaks:allow
         dataset_with_pixels.PatientID = "9990001112"
         dataset_with_pixels.PatientName = "JANE^SMITH"
+        dataset_with_pixels.StudyInstanceUID = "1.2.3.4.5.6.7"  # gitleaks:allow
+        dataset_with_pixels.SOPClassUID = "1.2.840.10008.5.1.4.1.1.1.2"  # gitleaks:allow
 
         # Wrap in PropertyMock to simulate DICOM event
         event = PropertyMock()
diff --git a/tests/services/dicom/test_validator.py b/tests/services/dicom/test_validator.py
@@ -0,0 +1,149 @@
+import pytest
+from pydicom import Dataset
+
+from services.dicom.validator import DicomValidationError, DicomValidator
+
+# Test UIDs for DICOM validation tests
+TEST_SOP_INSTANCE_UID = "1.2.3.4.5"  # gitleaks:allow
+TEST_STUDY_INSTANCE_UID = "1.2.3.4.5.6"  # gitleaks:allow
+TEST_SOP_CLASS_UID = "1.2.840.10008.5.1.4.1.1.1.2"  # gitleaks:allow
+TEST_PATIENT_ID = "123456"
+
+
+class TestDicomValidator:
+    @pytest.fixture
+    def valid_dataset(self):
+        ds = Dataset()
+        ds.SOPInstanceUID = TEST_SOP_INSTANCE_UID
+        ds.PatientID = TEST_PATIENT_ID
+        ds.StudyInstanceUID = TEST_STUDY_INSTANCE_UID
+        ds.SOPClassUID = TEST_SOP_CLASS_UID
+        return ds
+
+    @pytest.fixture
+    def valid_image_dataset(self, valid_dataset):
+        valid_dataset.PixelData = b"\x00" * 100
+        valid_dataset.Rows = 10
+        valid_dataset.Columns = 10
+        valid_dataset.BitsAllocated = 8
+        return valid_dataset
+
+    def test_validate_dataset_success(self, valid_dataset):
+        validator = DicomValidator()
+        validator.validate_dataset(valid_dataset)  # Should not raise
+
+    def test_validate_dataset_missing_sop_instance_uid(self):
+        ds = Dataset()
+        ds.PatientID = TEST_PATIENT_ID
+        ds.StudyInstanceUID = TEST_STUDY_INSTANCE_UID
+        ds.SOPClassUID = TEST_SOP_CLASS_UID
+
+        validator = DicomValidator()
+        with pytest.raises(DicomValidationError, match="Missing required tag: SOPInstanceUID"):
+            validator.validate_dataset(ds)
+
+    def test_validate_dataset_missing_patient_id(self):
+        ds = Dataset()
+        ds.SOPInstanceUID = TEST_SOP_INSTANCE_UID
+        ds.StudyInstanceUID = TEST_STUDY_INSTANCE_UID
+        ds.SOPClassUID = TEST_SOP_CLASS_UID
+
+        validator = DicomValidator()
+        with pytest.raises(DicomValidationError, match="Missing required tag: PatientID"):
+            validator.validate_dataset(ds)
+
+    def test_validate_dataset_missing_study_instance_uid(self):
+        ds = Dataset()
+        ds.SOPInstanceUID = TEST_SOP_INSTANCE_UID
+        ds.PatientID = TEST_PATIENT_ID
+        ds.SOPClassUID = TEST_SOP_CLASS_UID
+
+        validator = DicomValidator()
+        with pytest.raises(DicomValidationError, match="Missing required tag: StudyInstanceUID"):
+            validator.validate_dataset(ds)
+
+    def test_validate_dataset_missing_sop_class_uid(self):
+        ds = Dataset()
+        ds.SOPInstanceUID = TEST_SOP_INSTANCE_UID
+        ds.PatientID = TEST_PATIENT_ID
+        ds.StudyInstanceUID = TEST_STUDY_INSTANCE_UID
+
+        validator = DicomValidator()
+        with pytest.raises(DicomValidationError, match="Missing required tag: SOPClassUID"):
+            validator.validate_dataset(ds)
+
+    def test_validate_bytes_valid_preamble(self):
+        # 128 bytes preamble + DICM + minimal content
+        data = b"\x00" * 128 + b"DICM" + b"\x00" * 100
+
+        validator = DicomValidator()
+        validator.validate_bytes(data)  # Should not raise
+
+    def test_validate_bytes_missing_preamble(self):
+        # DICM at wrong position (no 128-byte preamble before it)
+        data = b"DICM" + b"\x00" * 200
+
+        validator = DicomValidator()
+        with pytest.raises(DicomValidationError, match="Invalid DICOM prefix"):
+            validator.validate_bytes(data)
+
+    def test_validate_bytes_too_small(self):
+        data = b"\x00" * 50
+
+        validator = DicomValidator()
+        with pytest.raises(DicomValidationError, match="too small"):
+            validator.validate_bytes(data)
+
+    def test_validate_bytes_wrong_magic(self):
+        data = b"\x00" * 128 + b"XXXX" + b"\x00" * 100
+
+        validator = DicomValidator()
+        with pytest.raises(DicomValidationError, match="Invalid DICOM prefix"):
+            validator.validate_bytes(data)
+
+    def test_validate_pixel_data_valid(self, valid_image_dataset):
+        validator = DicomValidator()
+        validator.validate_pixel_data(valid_image_dataset)  # Should not raise
+
+    def test_validate_pixel_data_missing_rows(self):
+        ds = Dataset()
+        ds.PixelData = b"\x00" * 100
+        ds.Columns = 10
+        ds.BitsAllocated = 8
+
+        validator = DicomValidator()
+        with pytest.raises(DicomValidationError, match="missing Rows"):
+            validator.validate_pixel_data(ds)
+
+    def test_validate_pixel_data_missing_columns(self):
+        ds = Dataset()
+        ds.PixelData = b"\x00" * 100
+        ds.Rows = 10
+        ds.BitsAllocated = 8
+
+        validator = DicomValidator()
+        with pytest.raises(DicomValidationError, match="missing Columns"):
+            validator.validate_pixel_data(ds)
+
+    def test_validate_pixel_data_missing_bits_allocated(self):
+        ds = Dataset()
+        ds.PixelData = b"\x00" * 100
+        ds.Rows = 10
+        ds.Columns = 10
+
+        validator = DicomValidator()
+        with pytest.raises(DicomValidationError, match="missing BitsAllocated"):
+            validator.validate_pixel_data(ds)
+
+    def test_validate_pixel_data_no_pixel_data(self):
+        ds = Dataset()  # No PixelData
+
+        validator = DicomValidator()
+        validator.validate_pixel_data(ds)  # Should not raise
+
+    def test_validate_pixel_data_none_pixel_data(self):
+        ds = Dataset()
+        ds.PixelData = None
+
+        validator = DicomValidator()
+        validator.validate_pixel_data(ds)  # Should not raise
