Storage: add systests for V4 signed URLs with CSEK headers #7626

Activity

1. tseaver

   added

   api: storageIssues related to the Cloud Storage API.Issues related to the Cloud Storage API.

   testing

   type: processA process-related concern. May include testing, release, or the like.A process-related concern. May include testing, release, or the like.

   backend

   on Apr 1, 2019
2. tseaver commented on May 24, 2019

   

   tseaver

   on May 24, 2019

   ContributorAuthor

   More actions

   @frankyn, @jkwlui I'm don't know the status of the back-end bug which blocks this issue.
3. frankyn commented on Jun 18, 2019

   

   frankyn

   on Jun 18, 2019

   Contributor

   More actions

   @tseaver the fix is being rolled out to production right now.
4. frankyn commented on Oct 9, 2019

   

   frankyn

   on Oct 9, 2019

   Contributor

   More actions

   This is fixed in production.
5. tseaver

   self-assigned this

   on Oct 9, 2019
6. tseaver commented on Oct 10, 2019

   

   tseaver

   on Oct 10, 2019

   ContributorAuthor

   More actions

   @frankyn I was trying to use the same pattern for testing both V2 and V4, but the V2 test fails: do you expect signing + CSEK to work with V2? E.g., I'm tweaking Blob.generate_signed_url so:

   --- a/storage/google/cloud/storage/blob.py
   +++ b/storage/google/cloud/storage/blob.py
   @@ -468,6 +468,11 @@ class Blob(_PropertyMixin):
            else:
                helper = generate_signed_url_v4
    
   +        if self._encryption_key is not None:
   +            if headers is None:
   +                headers = {}
   +            headers.update(_get_encryption_headers(self._encryption_key))
   +
            return helper(
                credentials,
                resource=resource,

   and the tests so:

   --- a/storage/tests/system.py
   +++ b/storage/tests/system.py
   @@ -12,7 +12,9 @@
    # See the License for the specific language governing permissions and
    # limitations under the License.
    
   +import base64
    import datetime
   +import hashlib
    import os
    import re
    import tempfile
   @@ -860,11 +862,12 @@ class TestStorageSignURLs(unittest.TestCase):
            version="v2",
            payload=None,
            expiration=None,
   +        encryption_key=None,
        ):
            expiration = self._morph_expiration(version, expiration)
    
            if payload is not None:
   -            blob = self.bucket.blob(blob_name)
   +            blob = self.bucket.blob(blob_name, encryption_key=encryption_key)
                blob.upload_from_string(payload)
            else:
                blob = self.blob
   @@ -873,7 +876,17 @@ class TestStorageSignURLs(unittest.TestCase):
                expiration=expiration, method=method, client=Config.CLIENT, version
   =version
            )
    
   -        response = requests.get(signed_url)
   +        headers = {}
   +
   +        if encryption_key is not None:
   +            headers['x-goog-encryption-algorithm'] = "AES256"
   +            encoded_key = base64.b64encode(encryption_key).decode('utf-8')
   +            headers['x-goog-encryption-key'] = encoded_key
   +            key_hash = hashlib.sha256(encryption_key).digest()
   +            key_hash = base64.b64encode(key_hash).decode('utf-8')
   +            headers['x-goog-encryption-key-sha256'] = key_hash
   +
   +        response = requests.get(signed_url, headers=headers)
            self.assertEqual(response.status_code, 200)
            if payload is not None:
                self.assertEqual(response.content, payload)
   @@ -916,6 +929,23 @@ class TestStorageSignURLs(unittest.TestCase):
                version="v4",
            )
    
   +    def test_create_signed_read_url_v2_w_csek(self):
   +        encryption_key = os.urandom(32)
   +        self._create_signed_read_url_helper(
   +            blob_name="v2-w-csek.txt",
   +            payload=b"Test signed URL for blob w/ CSEK",
   +            encryption_key=encryption_key,
   +        )
   +
   +    def test_create_signed_read_url_v4_w_csek(self):
   +        encryption_key = os.urandom(32)
   +        self._create_signed_read_url_helper(
   +            blob_name="v2-w-csek.txt",
   +            payload=b"Test signed URL for blob w/ CSEK",
   +            encryption_key=encryption_key,
   +            version="v4",
   +        )
   +
        def _create_signed_delete_url_helper(self, version="v2", expiration=None):
            expiration = self._morph_expiration(version, expiration)
    
7. frankyn commented on Oct 10, 2019

   

   frankyn

   on Oct 10, 2019

   Contributor

   More actions

   This is only for V4, for V2 CSEK headers aren't allowed.
8. tseaver commented on Oct 10, 2019

   

   tseaver

   on Oct 10, 2019

   ContributorAuthor

   More actions

   This is only for V4, for V2 CSEK headers aren't allowed.

   OK, I will remove the V2 systest and unit tests for that feature.
9. tseaver commented on Oct 10, 2019

   

   tseaver

   on Oct 10, 2019

   ContributorAuthor

   More actions

   Resolved in chat, per https://cloud.google.com/storage/docs/access-control/signed-urls-v2#about-canonical-extension-headers
10. tseaver

    closed this as completedin #9450on Oct 10, 2019
11. tseaver

    added a commit that references this issue on Oct 10, 2019

    fix(storage): enable CSEK w/ V4 signed URLs (#9450)

    ...

    This issue will close once commit 42ce2ef is merged into the 'main' branch.

    42ce2ef
12. parthea

    added a commit that references this issue on Mar 9, 2026

    fix(storage): enable CSEK w/ V4 signed URLs (#9450)

    ...

    This issue will close once commit a3d940d is merged into the 'main' branch.

    a3d940d