validate pagination link host in container registry client

Author: alhudz

sdk/containerregistry/azure-containerregistry/azure/containerregistry/_container_registry_client.py

@@ -41,6 +41,7 @@
     _compute_digest,
     _is_tag,
     _parse_next_link,
+    _validate_next_link,
     _validate_digest,
     _get_blob_size,
     _get_manifest_size,
@@ -221,6 +222,7 @@ def prepare_request(next_link=None):
                     url, query_parameters, header_parameters
                 )
             else:
+                _validate_next_link(next_link, self._endpoint)
                 url = next_link
                 query_parameters: Dict[str, Any] = {}
                 path_format_arguments = {
@@ -369,6 +371,7 @@ def prepare_request(next_link=None):
                     url, query_parameters, header_parameters
                 )
             else:
+                _validate_next_link(next_link, self._endpoint)
                 url = next_link
                 query_parameters: Dict[str, Any] = {}
                 path_format_arguments = {
@@ -598,6 +601,7 @@ def prepare_request(next_link=None):
                     url, query_parameters, header_parameters
                 )
             else:
+                _validate_next_link(next_link, self._endpoint)
                 url = next_link
                 query_parameters: Dict[str, Any] = {}
                 path_format_arguments = {

sdk/containerregistry/azure-containerregistry/azure/containerregistry/_helpers.py

@@ -104,6 +104,25 @@ def _parse_next_link(link_string: str) -> Optional[str]:
     return link_string[1 : link_string.find(">")]
 
 
+def _validate_next_link(next_link: str, endpoint: str) -> None:
+    """Ensure a pagination continuation link targets the configured registry host.
+
+    The continuation link comes from a service response (the Link header or the
+    response body) and is therefore untrusted. A relative link, or an absolute
+    link whose host matches the endpoint, is accepted. An absolute link to any
+    other host is rejected so request credentials are never sent to a host the
+    caller did not configure.
+
+    :param str next_link: The continuation link returned by the service.
+    :param str endpoint: The configured registry endpoint.
+    :return: None
+    :raises ValueError: If next_link points at a different host than endpoint.
+    """
+    host = urlparse(next_link).netloc
+    if host and host.lower() != urlparse(endpoint).netloc.lower():
+        raise ValueError("The continuation link host does not match the registry endpoint.")
+
+
 def _enforce_https(request: PipelineRequest) -> None:
     """Raise ServiceRequestError if the request URL is non-HTTPS and the sender did not specify enforce_https=False
 

sdk/containerregistry/azure-containerregistry/azure/containerregistry/aio/_async_container_registry_client.py

@@ -50,6 +50,7 @@
     _compute_digest,
     _is_tag,
     _parse_next_link,
+    _validate_next_link,
     _validate_digest,
     _get_blob_size,
     _get_manifest_size,
@@ -223,6 +224,7 @@ def prepare_request(next_link=None):
 
             else:
                 # make call to next link with the client's api-version
+                _validate_next_link(next_link, self._endpoint)
                 _parsed_next_link = urllib.parse.urlparse(next_link)
                 _next_request_params = case_insensitive_dict(
                     {
@@ -365,6 +367,7 @@ def prepare_request(next_link=None):
                     _request.url, **path_format_arguments
                 )  # pylint: disable=protected-access
             else:
+                _validate_next_link(next_link, self._endpoint)
                 _parsed_next_link = urllib.parse.urlparse(next_link)
                 _next_request_params = case_insensitive_dict(
                     {
@@ -608,6 +611,7 @@ def prepare_request(next_link=None):
                     url, query_parameters, header_parameters
                 )
             else:
+                _validate_next_link(next_link, self._endpoint)
                 url = next_link
                 query_parameters: Dict[str, Any] = {}
                 path_format_arguments = {