Navigation Menu

Skip to content

Commit

Permalink
feat: add getters and setters for encryption_key and kms_key_name (#409)
Browse files Browse the repository at this point in the history
* fix: add getters and setters for attributes encryption_key and kms_key_name

* fix docstring formatting

* revise docstring

Co-authored-by: Tres Seaver <tseaver@palladion.com>
  • Loading branch information
cojenco and tseaver committed Apr 20, 2021
1 parent f090548 commit 2adfb59
Show file tree
Hide file tree
Showing 2 changed files with 77 additions and 7 deletions.
48 changes: 41 additions & 7 deletions google/cloud/storage/blob.py
Expand Up @@ -254,6 +254,31 @@ def chunk_size(self, value):
)
self._chunk_size = value

@property
def encryption_key(self):
"""Retrieve the customer-supplied encryption key for the object.
:rtype: bytes or ``NoneType``
:returns:
The encryption key or ``None`` if no customer-supplied encryption key was used,
or the blob's resource has not been loaded from the server.
"""
return self._encryption_key

@encryption_key.setter
def encryption_key(self, value):
"""Set the blob's encryption key.
See https://cloud.google.com/storage/docs/encryption#customer-supplied
To perform a key rotation for an encrypted blob, use :meth:`rewrite`.
See https://cloud.google.com/storage/docs/encryption/using-customer-supplied-keys?hl=ca#rotating
:type value: bytes
:param value: 32 byte encryption key for customer-supplied encryption.
"""
self._encryption_key = value

@staticmethod
def path_helper(bucket_path, blob_name):
"""Relative URL path for a blob.
Expand Down Expand Up @@ -347,25 +372,25 @@ def public_url(self):
def from_string(cls, uri, client=None):
"""Get a constructor for blob object by URI.
:type uri: str
:param uri: The blob uri pass to get blob object.
:type uri: str
:param uri: The blob uri pass to get blob object.
:type client: :class:`~google.cloud.storage.client.Client`
:param client:
(Optional) The client to use. If not passed, falls back to the
``client`` stored on the blob's bucket.
:rtype: :class:`google.cloud.storage.blob.Blob`
:returns: The blob object created.
:rtype: :class:`google.cloud.storage.blob.Blob`
:returns: The blob object created.
Example:
Get a constructor for blob object by URI..
Example:
Get a constructor for blob object by URI.
>>> from google.cloud import storage
>>> from google.cloud.storage.blob import Blob
>>> client = storage.Client()
>>> blob = Blob.from_string("gs://bucket/object")
"""
"""
from google.cloud.storage.bucket import Bucket

scheme, netloc, path, query, frag = urlsplit(uri)
Expand Down Expand Up @@ -3839,6 +3864,15 @@ def kms_key_name(self):
"""
return self._properties.get("kmsKeyName")

@kms_key_name.setter
def kms_key_name(self, value):
"""Set KMS encryption key for object.
:type value: str or ``NoneType``
:param value: new KMS key name (None to clear any existing key).
"""
self._patch_property("kmsKeyName", value)

storage_class = _scalar_property("storageClass")
"""Retrieve the storage class for the object.
Expand Down
36 changes: 36 additions & 0 deletions tests/unit/test_blob.py
Expand Up @@ -262,6 +262,42 @@ def test_acl_property(self):
self.assertIsInstance(acl, ObjectACL)
self.assertIs(acl, blob._acl)

def test_encryption_key_getter(self):
BLOB_NAME = "blob-name"
BUCKET = object()
blob = self._make_one(BLOB_NAME, bucket=BUCKET)
self.assertIsNone(blob.encryption_key)
VALUE = object()
blob._encryption_key = VALUE
self.assertIs(blob.encryption_key, VALUE)

def test_encryption_key_setter(self):
BLOB_NAME = "blob-name"
BUCKET = object()
blob = self._make_one(BLOB_NAME, bucket=BUCKET)
self.assertIsNone(blob._encryption_key)
key = b"12345678901234567890123456789012"
blob.encryption_key = key
self.assertEqual(blob._encryption_key, key)

def test_kms_key_name_getter(self):
BLOB_NAME = "blob-name"
BUCKET = object()
blob = self._make_one(BLOB_NAME, bucket=BUCKET)
self.assertIsNone(blob.kms_key_name)
VALUE = object()
blob._patch_property("kmsKeyName", VALUE)
self.assertIs(blob.kms_key_name, VALUE)

def test_kms_key_name_setter(self):
BLOB_NAME = "blob-name"
BUCKET = object()
blob = self._make_one(BLOB_NAME, bucket=BUCKET)
self.assertIsNone(blob._properties.get("kmsKeyName"))
kms_key_name = "cryptoKeys/test-key"
blob.kms_key_name = kms_key_name
self.assertEqual(blob._properties.get("kmsKeyName"), kms_key_name)

def test_path_bad_bucket(self):
fake_bucket = object()
name = u"blob-name"
Expand Down

0 comments on commit 2adfb59

Please sign in to comment.