feat: public access prevention

google/cloud/storage/bucket.py

@@ -52,6 +52,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
@@ -385,6 +386,11 @@ 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
+
     :type uniform_bucket_level_access_enabled: bool
     :params bucket_policy_only_enabled:
         (Optional) Whether the IAM-only policy is enabled for the bucket.
@@ -406,6 +412,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,
@@ -430,8 +437,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(
@@ -466,6 +479,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_system.py

@@ -33,6 +33,8 @@
 from google.cloud.storage._helpers import _base64_md5hash
 from google.cloud.storage.bucket import LifecycleRuleDelete
 from google.cloud.storage.bucket import LifecycleRuleSetStorageClass
+from google.cloud.storage.constants import PUBLIC_ACCESS_PREVENTION_ENFORCED
+from google.cloud.storage.constants import PUBLIC_ACCESS_PREVENTION_UNSPECIFIED
 from google.cloud import kms
 from google import resumable_media
 import google.auth
@@ -2464,6 +2466,79 @@ def test_ubla_set_unset_preserves_acls(self):
         self.assertEqual(bucket_acl_before, bucket_acl_after)
         self.assertEqual(blob_acl_before, blob_acl_after)
 
+    def test_new_bucket_created_w_unspecified_pap(self):
+        new_bucket_name = "new-w-pap-unspecified" + unique_resource_id("-")
+        self.assertRaises(
+            exceptions.NotFound, Config.CLIENT.get_bucket, new_bucket_name
+        )
+        bucket = Config.CLIENT.bucket(new_bucket_name)
+        bucket.iam_configuration.uniform_bucket_level_access_enabled = True
+        bucket.create()
+        self.case_buckets_to_delete.append(new_bucket_name)
+
+        self.assertEqual(
+            bucket.iam_configuration.public_access_prevention,
+            PUBLIC_ACCESS_PREVENTION_UNSPECIFIED,
+        )
+
+        bucket.iam_configuration.public_access_prevention = (
+            PUBLIC_ACCESS_PREVENTION_ENFORCED
+        )
+        bucket.patch()
+        self.assertEqual(
+            bucket.iam_configuration.public_access_prevention,
+            PUBLIC_ACCESS_PREVENTION_ENFORCED,
+        )
+        self.assertTrue(bucket.iam_configuration.uniform_bucket_level_access_enabled)
+
+        bucket.iam_configuration.uniform_bucket_level_access_enabled = False
+        bucket.patch()
+        self.assertEqual(
+            bucket.iam_configuration.public_access_prevention,
+            PUBLIC_ACCESS_PREVENTION_ENFORCED,
+        )
+
+        with self.assertRaises(exceptions.BadRequest):
+            bucket.iam_configuration.public_access_prevention = "unexpected value"
+            bucket.patch()
+
+        with self.assertRaises(exceptions.PreconditionFailed):
+            bucket.make_public()
+
+        blob_name = "my-blob.txt"
+        blob = bucket.blob(blob_name)
+        payload = b"DEADBEEF"
+        blob.upload_from_string(payload)
+        with self.assertRaises(exceptions.PreconditionFailed):
+            blob.make_public()
+
+    def test_new_bucket_created_w_enforced_pap(self):
+        new_bucket_name = "new-w-pap-enforced" + unique_resource_id("-")
+        self.assertRaises(
+            exceptions.NotFound, Config.CLIENT.get_bucket, new_bucket_name
+        )
+        bucket = Config.CLIENT.bucket(new_bucket_name)
+        bucket.iam_configuration.public_access_prevention = (
+            PUBLIC_ACCESS_PREVENTION_ENFORCED
+        )
+        bucket.create()
+        self.case_buckets_to_delete.append(new_bucket_name)
+
+        self.assertEqual(
+            bucket.iam_configuration.public_access_prevention,
+            PUBLIC_ACCESS_PREVENTION_ENFORCED,
+        )
+
+        bucket.iam_configuration.public_access_prevention = (
+            PUBLIC_ACCESS_PREVENTION_UNSPECIFIED
+        )
+        bucket.patch()
+        self.assertEqual(
+            bucket.iam_configuration.public_access_prevention,
+            PUBLIC_ACCESS_PREVENTION_UNSPECIFIED,
+        )
+        self.assertFalse(bucket.iam_configuration.uniform_bucket_level_access_enabled)
+
 
 class TestV4POSTPolicies(unittest.TestCase):
     def setUp(self):

tests/unit/test_bucket.py

@@ -21,6 +21,8 @@
 from google.cloud.storage.retry import DEFAULT_RETRY
 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 _make_connection(*responses):
@@ -291,6 +293,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)
 
@@ -313,6 +318,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