diff --git a/src/dicomweb_client/uri.py b/src/dicomweb_client/uri.py index 1d19600..39cb9e3 100644 --- a/src/dicomweb_client/uri.py +++ b/src/dicomweb_client/uri.py @@ -629,13 +629,16 @@ def _validate_resource_identifiers_and_suffix( def _validate_uid(uid: str, permissive: bool) -> None: """Validates a DICOM UID.""" - if len(uid) > _MAX_UID_LENGTH: - raise ValueError('UID cannot have more than 64 chars. ' - f'Actual count in {uid!r}: {len(uid)}') - if not permissive and _REGEX_UID.fullmatch(uid) is None: - raise ValueError(f'UID {uid!r} must match regex {_REGEX_UID!r} in ' - 'conformance with the DICOM Standard.') - elif permissive and _REGEX_PERMISSIVE_UID.fullmatch(uid) is None: + if not isinstance(uid, str): + raise TypeError('DICOM UID must be a string.') + if not permissive: + if len(uid) > _MAX_UID_LENGTH: + raise ValueError('UID cannot have more than 64 chars. ' + f'Actual count in {uid!r}: {len(uid)}') + if _REGEX_UID.fullmatch(uid) is None: + raise ValueError(f'UID {uid!r} must match regex {_REGEX_UID!r} in ' + 'conformance with the DICOM Standard.') + elif _REGEX_PERMISSIVE_UID.fullmatch(uid) is None: raise ValueError(f'Permissive mode is enabled. UID {uid!r} must match ' f'regex {_REGEX_PERMISSIVE_UID!r}.') diff --git a/src/dicomweb_client/web.py b/src/dicomweb_client/web.py index 71ef626..f8e0d0e 100644 --- a/src/dicomweb_client/web.py +++ b/src/dicomweb_client/web.py @@ -32,7 +32,11 @@ import requests import retrying -from dicomweb_client.uri import build_query_string, parse_query_parameters +from dicomweb_client.uri import ( + build_query_string, + parse_query_parameters, + _validate_uid +) logger = logging.getLogger(__name__) @@ -200,7 +204,8 @@ def __init__( proxies: Optional[Dict[str, str]] = None, headers: Optional[Dict[str, str]] = None, callback: Optional[Callable] = None, - chunk_size: int = 10**6 + chunk_size: int = 10**6, + permissive_uid: bool = False ) -> None: """Instatiate client. @@ -235,6 +240,14 @@ def __init__( when streaming data from the server using chunked transfer encoding (used by ``iter_*()`` methods as well as the ``store_instances()`` method); defaults to ``10**6`` bytes (1MB) + permissive_uid: bool, optional + If ``True``, relaxes the DICOM Standard validation for UIDs (see + main docstring for details). This option is made available since + users may be occasionally forced to work with DICOMs or services + that may be in violation of the standard. Unless required, use of + this flag is **not** recommended, since non-conformant UIDs may + lead to unexpected errors downstream, e.g., rejection by a DICOMweb + server, etc. Warning ------- @@ -314,6 +327,7 @@ def __init__( if callback is not None: self._session.hooks = {'response': [callback, ]} self._chunk_size = chunk_size + self._permissive_uid = permissive_uid self.set_http_retry_params() def _get_transaction_url(self, transaction: _Transaction) -> str: @@ -2163,29 +2177,6 @@ def delete_study( url += f'?{additional_params_query_string}' self._http_delete(url) - def _assert_uid_format(self, uid: str) -> None: - """Check whether a DICOM UID has the correct format. - - Parameters - ---------- - uid: str - DICOM UID - - Raises - ------ - TypeError - When `uid` is not a string - ValueError - When `uid` doesn't match the regular expression pattern - ``"^[.0-9]+$"`` - - """ - if not isinstance(uid, str): - raise TypeError('DICOM UID must be a string.') - pattern = re.compile('^[.0-9]+$') - if not pattern.search(uid): - raise ValueError('DICOM UID has invalid format.') - def search_for_series( self, study_instance_uid: Optional[str] = None, @@ -2238,7 +2229,7 @@ def search_for_series( """ # noqa: E501 if study_instance_uid is not None: - self._assert_uid_format(study_instance_uid) + _validate_uid(study_instance_uid, self._permissive_uid) logger.info(f'search for series of study "{study_instance_uid}"') else: logger.info('search for series') @@ -2297,12 +2288,12 @@ def _get_series( f'retrieve series "{series_instance_uid}" ' f'of study "{study_instance_uid}"' ) - self._assert_uid_format(study_instance_uid) + _validate_uid(study_instance_uid, self._permissive_uid) if series_instance_uid is None: raise ValueError( 'Series Instance UID is required for retrieval of series.' ) - self._assert_uid_format(series_instance_uid) + _validate_uid(series_instance_uid, self._permissive_uid) url = self._get_series_url( _Transaction.RETRIEVE, study_instance_uid, @@ -2457,7 +2448,7 @@ def retrieve_series_metadata( 'Study Instance UID is required for retrieval of ' 'series metadata.' ) - self._assert_uid_format(study_instance_uid) + _validate_uid(study_instance_uid, self._permissive_uid) if series_instance_uid is None: raise ValueError( 'Series Instance UID is required for retrieval of ' @@ -2467,7 +2458,7 @@ def retrieve_series_metadata( f'retrieve metadata of series "{series_instance_uid}" ' f'of study "{study_instance_uid}"' ) - self._assert_uid_format(series_instance_uid) + _validate_uid(series_instance_uid, self._permissive_uid) url = self._get_series_url( _Transaction.RETRIEVE, study_instance_uid, @@ -2660,7 +2651,7 @@ def search_for_instances( if series_instance_uid is not None: message += f' of series "{series_instance_uid}"' if study_instance_uid is not None: - self._assert_uid_format(study_instance_uid) + _validate_uid(study_instance_uid, self._permissive_uid) message += f' of study "{study_instance_uid}"' logger.info(message) url = self._get_instances_url( @@ -2727,12 +2718,12 @@ def retrieve_instance( raise ValueError( 'Study Instance UID is required for retrieval of instance.' ) - self._assert_uid_format(study_instance_uid) + _validate_uid(study_instance_uid, self._permissive_uid) if series_instance_uid is None: raise ValueError( 'Series Instance UID is required for retrieval of instance.' ) - self._assert_uid_format(series_instance_uid) + _validate_uid(series_instance_uid, self._permissive_uid) if sop_instance_uid is None: raise ValueError( 'SOP Instance UID is required for retrieval of instance.' @@ -2742,7 +2733,7 @@ def retrieve_instance( f'of series "{series_instance_uid}" ' f'of study "{study_instance_uid}"' ) - self._assert_uid_format(sop_instance_uid) + _validate_uid(sop_instance_uid, self._permissive_uid) url = self._get_instances_url( _Transaction.RETRIEVE, study_instance_uid, diff --git a/tests/test_uri.py b/tests/test_uri.py index c568c6d..c432e7d 100644 --- a/tests/test_uri.py +++ b/tests/test_uri.py @@ -70,6 +70,21 @@ def test_uid_permissive_invalid(uid): URI(_BASE_URL, uid, permissive=True) +def test_uid_length_permissive(): + """Tests that UIDs longer than 64 chars are allowed in permissive mode.""" + # Create a UID with 65 characters + uid_65 = '1' * 65 + + # Should fail in strict mode + with pytest.raises(ValueError, match='UID cannot have more than 64 chars'): + URI(_BASE_URL, uid_65, permissive=False) + + # Should succeed in permissive mode + URI(_BASE_URL, uid_65, permissive=True) + URI(_BASE_URL, '1.2.3', uid_65, permissive=True) + URI(_BASE_URL, '1.2.3', '4.5.6', uid_65, permissive=True) + + def test_uid_missing_error(): """Checks *ValueError* is raised when an expected UID is missing.""" with pytest.raises(ValueError, match='`study_instance_uid` missing with'): diff --git a/tests/test_web.py b/tests/test_web.py index 2de4893..ca91e63 100644 --- a/tests/test_web.py +++ b/tests/test_web.py @@ -1608,3 +1608,35 @@ def test_load_xml_response(httpserver, client, cache_dir): dataset = _load_xml_dataset(tree) assert dataset.RetrieveURL.startswith('https://wadors.hospital.com') assert len(dataset.ReferencedSOPSequence) == 2 + + +@pytest.mark.parametrize('invalid_uid', [ + '1.2/3.4', '1.2@3.4', '1.2.a.4', '1.2.A.4', '.23.4', '1.2..4', '1.2.', '.' +]) +def test_uid_strict_validation_fail(httpserver, invalid_uid): + client = DICOMwebClient(httpserver.url, permissive_uid=False) + with pytest.raises(ValueError, + match=f'UID {invalid_uid!r} must match regex'): + client.search_for_series(study_instance_uid=invalid_uid) + + +@pytest.mark.parametrize('uid', ['13-abc', 'hello', 'is it', "you're me"]) +def test_uid_permissive_validation_pass(httpserver, uid): + client_strict = DICOMwebClient(httpserver.url, permissive_uid=False) + client_permissive = DICOMwebClient(httpserver.url, permissive_uid=True) + + # Should fail in strict mode + with pytest.raises(ValueError, match=f'UID {uid!r} must match regex'): + client_strict.search_for_series(study_instance_uid=uid) + + # Should not raise exception in permissive mode + resp = client_permissive.search_for_series(study_instance_uid=uid) + assert isinstance(resp, list) + + +@pytest.mark.parametrize('uid', ['1.23.5@', '1/23.4']) +def test_uid_permissive_validation_fail(httpserver, uid): + client = DICOMwebClient(httpserver.url, permissive_uid=True) + with pytest.raises(ValueError, + match=f'UID {uid!r} must match regex'): + client.search_for_series(study_instance_uid=uid)