feat: public access prevention

google/cloud/storage/bucket.py

@@ -50,6 +50,7 @@
 from google.cloud.storage.constants import MULTI_REGIONAL_LEGACY_STORAGE_CLASS
 from google.cloud.storage.constants import MULTI_REGION_LOCATION_TYPE
 from google.cloud.storage.constants import NEARLINE_STORAGE_CLASS
+from google.cloud.storage.constants import PUBLIC_ACCESS_PREVENTION_UNSPECIFIED
 from google.cloud.storage.constants import REGIONAL_LEGACY_STORAGE_CLASS
 from google.cloud.storage.constants import REGION_LOCATION_TYPE
 from google.cloud.storage.constants import STANDARD_STORAGE_CLASS
@@ -383,6 +384,12 @@ class IAMConfiguration(dict):
  :type bucket: :class:`Bucket`
  :params bucket: Bucket for which this instance is the policy.
 
+ :type public_access_prevention: str
+ :params public_access_prevention:
+ (Optional) Whether the public access prevention policy is 'unspecified' (default) or 'enforced'
+ See: https://cloud.google.com/storage/docs/public-access-prevention
+ See: https://cloud.google.com/storage/docs/public-access-prevention
+
  :type uniform_bucket_level_access_enabled: bool
  :params bucket_policy_only_enabled:
  (Optional) Whether the IAM-only policy is enabled for the bucket.
@@ -404,6 +411,7 @@ class IAMConfiguration(dict):
  def __init__(
  self,
  bucket,
+ public_access_prevention=_default,
  uniform_bucket_level_access_enabled=_default,
  uniform_bucket_level_access_locked_time=_default,
  bucket_policy_only_enabled=_default,
@@ -428,8 +436,14 @@ def __init__(
  if uniform_bucket_level_access_enabled is _default:
  uniform_bucket_level_access_enabled = False
 
+ if public_access_prevention is _default:
+ public_access_prevention = PUBLIC_ACCESS_PREVENTION_UNSPECIFIED
+
  data = {
- "uniformBucketLevelAccess": {"enabled": uniform_bucket_level_access_enabled}
+ "uniformBucketLevelAccess": {
+ "enabled": uniform_bucket_level_access_enabled
+ },
+ "publicAccessPrevention": public_access_prevention,
  }
  if uniform_bucket_level_access_locked_time is not _default:
  data["uniformBucketLevelAccess"]["lockedTime"] = _datetime_to_rfc3339(
@@ -464,6 +478,21 @@ def bucket(self):
  """
  return self._bucket
 
+ @property
+ def public_access_prevention(self):
+ """Setting for public access prevention policy. Options are 'unspecified' (default) or 'enforced'.
+ More information can be found at https://cloud.google.com/storage/docs/public-access-prevention
+
+ :rtype: string
+ :returns: the public access prevention status, either 'enforced' or 'unspecified'.
+ """
+ return self["publicAccessPrevention"]
+
+ @public_access_prevention.setter
+ def public_access_prevention(self, value):
+ self["publicAccessPrevention"] = value
+ self.bucket._patch_property("iamConfiguration", self)
+
  @property
  def uniform_bucket_level_access_enabled(self):
  """If set, access checks only use bucket-level IAM policies or above.

google/cloud/storage/constants.py

@@ -96,3 +96,16 @@
 _DEFAULT_TIMEOUT = 60 # in seconds
 """The default request timeout in seconds if a timeout is not explicitly given.
 """
+
+# Public Access Prevention
+PUBLIC_ACCESS_PREVENTION_ENFORCED = "enforced"
+"""Enforced public access prevention value.
+
+See: https://cloud.google.com/storage/docs/public-access-prevention
+"""
+
+PUBLIC_ACCESS_PREVENTION_UNSPECIFIED = "unspecified"
+"""Unspecified public access prevention value.
+
+See: https://cloud.google.com/storage/docs/public-access-prevention
+"""

tests/system/test_bucket.py

@@ -766,3 +766,82 @@ def test_ubla_set_unset_preserves_acls(
 
  assert bucket_acl_before == bucket_acl_after
  assert blob_acl_before == blob_acl_after
+
+
+def test_new_bucket_created_w_unspecified_pap(
+ storage_client, buckets_to_delete, blobs_to_delete,
+):
+ from google.cloud.storage import constants
+
+ bucket_name = _helpers.unique_name("new-w-pap-unspecified")
+ bucket = storage_client.bucket(bucket_name)
+ bucket.iam_configuration.uniform_bucket_level_access_enabled = True
+ bucket.create()
+ buckets_to_delete.append(bucket)
+
+ assert (
+ bucket.iam_configuration.public_access_prevention
+ == constants.PUBLIC_ACCESS_PREVENTION_UNSPECIFIED
+ )
+
+ bucket.iam_configuration.public_access_prevention = (
+ constants.PUBLIC_ACCESS_PREVENTION_ENFORCED
+ )
+ bucket.patch()
+ assert (
+ bucket.iam_configuration.public_access_prevention
+ == constants.PUBLIC_ACCESS_PREVENTION_ENFORCED
+ )
+ assert bucket.iam_configuration.uniform_bucket_level_access_enabled
+
+ bucket.iam_configuration.uniform_bucket_level_access_enabled = False
+ bucket.patch()
+ assert (
+ bucket.iam_configuration.public_access_prevention
+ == constants.PUBLIC_ACCESS_PREVENTION_ENFORCED
+ )
+
+ with pytest.raises(exceptions.BadRequest):
+ bucket.iam_configuration.public_access_prevention = "unexpected value"
+ bucket.patch()
+
+ with pytest.raises(exceptions.PreconditionFailed):
+ bucket.make_public()
+
+ blob_name = "my-blob.txt"
+ blob = bucket.blob(blob_name)
+ payload = b"DEADBEEF"
+ blob.upload_from_string(payload)
+
+ with pytest.raises(exceptions.PreconditionFailed):
+ blob.make_public()
+
+
+def test_new_bucket_created_w_enforced_pap(
+ storage_client, buckets_to_delete, blobs_to_delete,
+):
+ from google.cloud.storage import constants
+
+ bucket_name = _helpers.unique_name("new-w-pap-enforced")
+ bucket = storage_client.bucket(bucket_name)
+ bucket.iam_configuration.public_access_prevention = (
+ constants.PUBLIC_ACCESS_PREVENTION_ENFORCED
+ )
+ bucket.create()
+ buckets_to_delete.append(bucket)
+
+ assert (
+ bucket.iam_configuration.public_access_prevention
+ == constants.PUBLIC_ACCESS_PREVENTION_ENFORCED
+ )
+
+ bucket.iam_configuration.public_access_prevention = (
+ constants.PUBLIC_ACCESS_PREVENTION_UNSPECIFIED
+ )
+ bucket.patch()
+
+ assert (
+ bucket.iam_configuration.public_access_prevention
+ == constants.PUBLIC_ACCESS_PREVENTION_UNSPECIFIED
+ )
+ assert not bucket.iam_configuration.uniform_bucket_level_access_enabled

tests/unit/test_bucket.py

@@ -22,6 +22,8 @@
 from google.cloud.storage.retry import DEFAULT_RETRY_IF_ETAG_IN_JSON
 from google.cloud.storage.retry import DEFAULT_RETRY_IF_GENERATION_SPECIFIED
 from google.cloud.storage.retry import DEFAULT_RETRY_IF_METAGENERATION_SPECIFIED
+from google.cloud.storage.constants import PUBLIC_ACCESS_PREVENTION_ENFORCED
+from google.cloud.storage.constants import PUBLIC_ACCESS_PREVENTION_UNSPECIFIED
 
 
 def _create_signing_credentials():
@@ -356,6 +358,9 @@ def test_ctor_defaults(self):
  self.assertIs(config.bucket, bucket)
  self.assertFalse(config.uniform_bucket_level_access_enabled)
  self.assertIsNone(config.uniform_bucket_level_access_locked_time)
+ self.assertEqual(
+ config.public_access_prevention, PUBLIC_ACCESS_PREVENTION_UNSPECIFIED
+ )
  self.assertFalse(config.bucket_policy_only_enabled)
  self.assertIsNone(config.bucket_policy_only_locked_time)
 
@@ -378,6 +383,24 @@ def test_ctor_explicit_ubla(self):
  self.assertTrue(config.bucket_policy_only_enabled)
  self.assertEqual(config.bucket_policy_only_locked_time, now)
 
+ def test_ctor_explicit_pap(self):
+ bucket = self._make_bucket()
+
+ config = self._make_one(
+ bucket, public_access_prevention=PUBLIC_ACCESS_PREVENTION_ENFORCED,
+ )
+
+ self.assertIs(config.bucket, bucket)
+ self.assertFalse(config.uniform_bucket_level_access_enabled)
+ self.assertEqual(
+ config.public_access_prevention, PUBLIC_ACCESS_PREVENTION_ENFORCED
+ )
+
+ config.public_access_prevention = PUBLIC_ACCESS_PREVENTION_UNSPECIFIED
+ self.assertEqual(
+ config.public_access_prevention, PUBLIC_ACCESS_PREVENTION_UNSPECIFIED
+ )
+
  def test_ctor_explicit_bpo(self):
  import datetime
  import pytz