Add response-header overrides to BucketObject.generate_url

generate_url gains optional content_disposition/content_type parameters
that ask the backing store to serve the object with those response
headers on GET — needed by consumers (e.g. Galaxy) that hand out
presigned download URLs with a forced filename and MIME type.

The signature is uniform across all four providers:
- AWS: ResponseContentDisposition/ResponseContentType presigned params
- Azure: content_disposition/content_type on the blob SAS
- GCP: response-content-disposition/response-content-type signed query
  parameters
- OpenStack Swift: the filename portion of the disposition is honored
  via the tempurl filename/inline query parameters (appended after
  signing — the tempurl signature covers only method, path and expiry);
  the content type cannot be overridden

Both parameters are ignored for writable URLs.

CHANGELOG.rst

@@ -8,6 +8,12 @@
   3.10–3.13, and CI runs the mock-provider suite on both the lowest and highest
   supported versions. ``mypy`` type-checks against 3.10 so newer-stdlib usage
   is caught at lint time.
+* **Response-header overrides on signed URLs.** ``BucketObject.generate_url`` now
+  accepts optional ``content_disposition`` and ``content_type`` parameters that ask
+  the backing store to serve the object with those response headers on GET. Honored
+  fully by AWS, Azure and GCP; OpenStack Swift honors the filename portion of the
+  disposition via its tempurl ``filename`` parameter (the content type cannot be
+  overridden). Both are ignored for writable URLs.
 
 4.2.0 - July 9, 2026 (sha 60dbe305f0af46a7ce3eadd093756d31b8131b2e)
 -------------------------------------------------------------------

cloudbridge/interfaces/resources.py

@@ -2435,7 +2435,9 @@ class BucketObject(CloudResource):
         pass
 
     @abstractmethod
-    def generate_url(self, expires_in: int, writable: bool = False) -> str:
+    def generate_url(self, expires_in: int, writable: bool = False,
+                     content_disposition: str | None = None,
+                     content_type: str | None = None) -> str:
         """
         Generate a signed URL to this object.
 
@@ -2449,6 +2451,20 @@ class BucketObject(CloudResource):
         :param writable: Write permission for this signed URL. Users with the URL
             will be able to upload to this object, but they will NOT be able to
             read from it.
+        :type content_disposition: ``str``
+        :param content_disposition: When set, ask the backing store to serve
+            the object with this ``Content-Disposition`` response header on
+            GET (e.g. ``attachment; filename="data.txt"``). This is a
+            response-serving hint: every provider accepts it and honors it
+            where the backing store supports response-header overrides (AWS,
+            Azure and GCP fully; OpenStack Swift honors the filename portion
+            via its tempurl ``filename`` parameter). Ignored when
+            ``writable`` is ``True``.
+        :type content_type: ``str``
+        :param content_type: When set, ask the backing store to serve the
+            object with this ``Content-Type`` response header on GET. Honored
+            by AWS, Azure and GCP; OpenStack Swift cannot override the
+            content type. Ignored when ``writable`` is ``True``.
 
         :rtype: ``str``
         :return: A URL to access the object.

cloudbridge/providers/aws/resources.py

@@ -947,11 +947,19 @@ class AWSBucketObject(BaseBucketObject):
     def delete(self) -> None:
         self._obj.delete()
 
-    def generate_url(self, expires_in: int, writable: bool = False) -> str:
+    def generate_url(self, expires_in: int, writable: bool = False,
+                     content_disposition: str | None = None,
+                     content_type: str | None = None) -> str:
+        params = {'Bucket': self._obj.bucket_name, 'Key': self.id}
+        if not writable:
+            if content_disposition:
+                params['ResponseContentDisposition'] = content_disposition
+            if content_type:
+                params['ResponseContentType'] = content_type
         return cast("AWSCloudProvider", self._provider).s3_conn.meta.client \
             .generate_presigned_url(
                 'put_object' if writable else 'get_object',
-                Params={'Bucket': self._obj.bucket_name, 'Key': self.id},
+                Params=params,
                 ExpiresIn=expires_in)
 
     def refresh(self) -> None:

cloudbridge/providers/azure/azure_client.py

@@ -508,7 +508,9 @@ class AzureClient(object):
         blob_client.delete_blob(delete_snapshots)
 
     def get_blob_url(self, container_name: Any, blob_name: str,
-                     expiry_time: int, writable: bool) -> str:
+                     expiry_time: int, writable: bool,
+                     content_disposition: str | None = None,
+                     content_type: str | None = None) -> str:
         now = datetime.datetime.utcnow()
         expiry = now + datetime.timedelta(
             seconds=expiry_time)
@@ -520,7 +522,9 @@ class AzureClient(object):
         sas = generate_blob_sas(
             self.storage_account, container_name, blob_name,
             permission=BlobSasPermissions(read=True, write=writable), expiry=expiry,
-            user_delegation_key=delegation_key
+            user_delegation_key=delegation_key,
+            content_disposition=content_disposition,
+            content_type=content_type
         )
         url = f"https://{self.storage_account}.blob.core.windows.net/{container_name}/{blob_name}?{sas}"
         return url

cloudbridge/providers/azure/resources.py

@@ -311,13 +311,17 @@ class AzureBucketObject(BaseBucketObject):
         """
         self._blob_client.delete_blob()
 
-    def generate_url(self, expires_in: int, writable: bool = False) -> str:
+    def generate_url(self, expires_in: int, writable: bool = False,
+                     content_disposition: str | None = None,
+                     content_type: str | None = None) -> str:
         """
         Generate a URL to this object.
         """
         return cast(
             "AzureCloudProvider", self._provider).azure_client.get_blob_url(
-            self._container, self.name, expires_in, writable)
+            self._container, self.name, expires_in, writable,
+            content_disposition=None if writable else content_disposition,
+            content_type=None if writable else content_type)
 
     def refresh(self) -> None:
         pass

cloudbridge/providers/gcp/resources.py

@@ -2198,7 +2198,9 @@ class GCPBucketObject(BaseBucketObject):
          .delete(bucket=self._obj['bucket'], object=self.name)
          .execute())
 
-    def generate_url(self, expires_in: int, writable: bool = False) -> str:
+    def generate_url(self, expires_in: int, writable: bool = False,
+                     content_disposition: str | None = None,
+                     content_type: str | None = None) -> str:
         """
         Generates a signed URL accessible to everyone.
 
@@ -2210,10 +2212,19 @@ class GCPBucketObject(BaseBucketObject):
         provider = cast("GCPCloudProvider", self._provider)
         http_method = "PUT" if writable else "GET"
 
+        query_parameters: dict[str, Any] = {}
+        if not writable:
+            if content_disposition:
+                query_parameters[
+                    'response-content-disposition'] = content_disposition
+            if content_type:
+                query_parameters['response-content-type'] = content_type
+
         # pylint:disable=protected-access
         return helpers.generate_signed_url(
             provider._credentials, self._obj['bucket'], self.name,
-            expiration=expires_in, http_method=http_method)
+            expiration=expires_in, http_method=http_method,
+            query_parameters=query_parameters or None)
 
     def refresh(self) -> None:
         # pylint:disable=protected-access

cloudbridge/providers/openstack/resources.py

@@ -14,6 +14,7 @@ from typing import IO
 from typing import Iterable
 from typing import TYPE_CHECKING
 from typing import cast
+from urllib.parse import quote
 from urllib.parse import urljoin
 from urllib.parse import urlparse
 
@@ -1474,7 +1475,9 @@ class OpenStackBucketObject(BaseBucketObject):
             for del_res in swift.delete(self.cbcontainer.name, [self.name, ]):
                 del_res['success']
 
-    def generate_url(self, expires_in: int, writable: bool = False) -> str:
+    def generate_url(self, expires_in: int, writable: bool = False,
+                     content_disposition: str | None = None,
+                     content_type: str | None = None) -> str:
         http_method = "PUT" if writable else "GET"
         # Set a temp url key on the object (http://bit.ly/2NBiXGD)
         temp_url_key = "cloudbridge-tmp-url-key"
@@ -1484,8 +1487,23 @@ class OpenStackBucketObject(BaseBucketObject):
         base_url = urlparse(swift.get_service_auth()[0])
         access_point = "{0}://{1}".format(base_url.scheme, base_url.netloc)
         url_path = "/".join([base_url.path, self.cbcontainer.name, self.name])
-        return urljoin(access_point, generate_temp_url(url_path, expires_in,
-                                                       temp_url_key, http_method))
+        url = urljoin(access_point, generate_temp_url(url_path, expires_in,
+                                                      temp_url_key,
+                                                      http_method))
+        if not writable and content_disposition:
+            # Swift's tempurl middleware serves Content-Disposition via the
+            # unsigned `filename`/`inline` query parameters (the signature
+            # covers only method, path and expiry); Content-Type cannot be
+            # overridden.
+            match = re.search(r'filename\*?=(?:"([^"]+)"|([^;]+))',
+                              content_disposition)
+            filename = (match.group(1) or match.group(2)).strip() \
+                if match else None
+            if content_disposition.strip().lower().startswith('inline'):
+                url += '&inline'
+            if filename:
+                url += '&filename=' + quote(filename)
+        return url
 
     def refresh(self) -> None:
         self._obj = self.cbcontainer.objects.get(self.id)._obj

tests/test_object_store_service.py

@@ -229,6 +229,69 @@ class CloudObjectStoreServiceTestCase(ProviderTestBase):
                 obj.save_content(target_stream)
                 self.assertEqual(target_stream.getvalue(), content)
 
+    @helpers.skipIfNoService(['storage.buckets'])
+    def test_generate_url_with_response_headers(self):
+        name = "cbtestbucketobjs-{0}".format(helpers.get_uuid())
+        test_bucket = self.provider.storage.buckets.create(name)
+
+        with cb_helpers.cleanup_action(lambda: test_bucket.delete()):
+            obj_name = "hello_response_headers.txt"
+            obj = test_bucket.objects.create(obj_name)
+
+            with cb_helpers.cleanup_action(lambda: obj.delete()):
+                content = b"Hello World. Serve me with response headers."
+                obj.upload(content)
+
+                disposition = 'attachment; filename="hello.txt"'
+                content_type = "application/octet-stream"
+                url = obj.generate_url(100,
+                                       content_disposition=disposition,
+                                       content_type=content_type)
+                if isinstance(self.provider, TestMockHelperMixin):
+                    # Presigned URLs are constructed client-side, so the
+                    # response-header overrides can be asserted on the query
+                    # string without a network round trip.
+                    self.assertIn('response-content-disposition', url)
+                    self.assertIn('response-content-type', url)
+                    raise self.skipTest(
+                        "Skipping rest of test - mock providers can't"
+                        " access generated url")
+                response = requests.get(url)
+                self.assertEqual(response.content, content)
+                got_disposition = response.headers.get(
+                    'Content-Disposition', '')
+                self.assertTrue(got_disposition.startswith('attachment'),
+                                "Expected attachment disposition, got: "
+                                f"{got_disposition}")
+                self.assertIn('hello.txt', got_disposition)
+                if self.provider.PROVIDER_ID != 'openstack':
+                    # Swift's tempurl middleware cannot override Content-Type
+                    # (it only honors the filename portion of the disposition
+                    # via its `filename` query parameter).
+                    self.assertEqual(response.headers.get('Content-Type'),
+                                     content_type)
+
+    @helpers.skipIfNoService(['storage.buckets'])
+    def test_generate_url_writable_ignores_response_headers(self):
+        name = "cbtestbucketobjs-{0}".format(helpers.get_uuid())
+        test_bucket = self.provider.storage.buckets.create(name)
+
+        with cb_helpers.cleanup_action(lambda: test_bucket.delete()):
+            obj_name = "hello_response_headers.txt"
+            obj = test_bucket.objects.create(obj_name)
+
+            with cb_helpers.cleanup_action(lambda: obj.delete()):
+                url = obj.generate_url(
+                    100, writable=True,
+                    content_disposition='attachment; filename="hello.txt"',
+                    content_type="application/octet-stream")
+                # Response-header overrides only make sense for reads; a
+                # writable URL must not carry them (across all providers:
+                # AWS/GCP query params, Azure SAS rscd/rsct, Swift filename).
+                self.assertNotIn('response-content-disposition', url)
+                self.assertNotIn('rscd=', url)
+                self.assertNotIn('filename=', url)
+
     @helpers.skipIfNoService(['storage.buckets'])
     def test_upload_download_bucket_content_from_file(self):
         name = "cbtestbucketobjs-{0}".format(helpers.get_uuid())