Storage: add 'bucket_policy_only' IAM property

storage/google/cloud/storage/blob.py

@@ -1537,7 +1537,7 @@ def update_storage_class(self, new_class, client=None):
             raise ValueError("Invalid storage class: %s" % (new_class,))
 
         # Update current blob's storage class prior to rewrite
-        self._patch_property('storageClass', new_class)
+        self._patch_property("storageClass", new_class)
 
         # Execute consecutive rewrite operations until operation is done
         token, _, _ = self.rewrite(self)

storage/google/cloud/storage/bucket.py

@@ -272,6 +272,96 @@ def from_api_repr(cls, resource):
         return instance
 
 
+class IAMConfiguration(dict):
+    """Map a bucket's IAM configuration.
+
+    :type bucket: :class:`Bucket`
+    :params bucket: Bucket for which this instance is the policy.
+
+    :type bucket_policy_only_enabled: bool
+    :params bucket_policy_only_enabled: (optional) whether the IAM-only policy is enabled for the bucket.
+
+    :type bucket_policy_only_locked_time: :class:`datetime.datetime`
+    :params bucket_policy_only_locked_time: (optional) When the bucket's IAM-only policy was ehabled.  This value should normally only be set by the back-end API.
+    """
+
+    def __init__(
+        self,
+        bucket,
+        bucket_policy_only_enabled=False,
+        bucket_policy_only_locked_time=None,
+    ):
+        data = {"bucketPolicyOnly": {"enabled": bucket_policy_only_enabled}}
+        if bucket_policy_only_locked_time is not None:
+            data["bucketPolicyOnly"]["lockedTime"] = _datetime_to_rfc3339(
+                bucket_policy_only_locked_time
+            )
+        super(IAMConfiguration, self).__init__(data)
+        self._bucket = bucket
+
+    @classmethod
+    def from_api_repr(cls, resource, bucket):
+        """Factory:  construct instance from resource.
+
+        :type bucket: :class:`Bucket`
+        :params bucket: Bucket for which this instance is the policy.
+
+        :type resource: dict
+        :param resource: mapping as returned from API call.
+
+        :rtype: :class:`IAMConfiguration`
+        :returns: Instance created from resource.
+        """
+        instance = cls(bucket)
+        instance.update(resource)
+        return instance
+
+    @property
+    def bucket(self):
+        """Bucket for which this instance is the policy.
+
+        :rtype: :class:`Bucket`
+        :returns: the instance's bucket.
+        """
+        return self._bucket
+
+    @property
+    def bucket_policy_only_enabled(self):
+        """If set, access checks only use bucket-level IAM policies or above.
+
+        :rtype: bool
+        :returns: whether the bucket is configured to allow only IAM.
+        """
+        bpo = self.get("bucketPolicyOnly", {})
+        return bpo.get("enabled", False)
+
+    @bucket_policy_only_enabled.setter
+    def bucket_policy_only_enabled(self, value):
+        bpo = self.setdefault("bucketPolicyOnly", {})
+        bpo["enabled"] = bool(value)
+        self.bucket._patch_property("iamConfiguration", self)
+
+    @property
+    def bucket_policy_only_locked_time(self):
+        """Deadline for changing :attr:`bucket_policy_only_enabled` from true to false.
+
+        If the bucket's :attr:`bucket_policy_only_enabled` is true, this property
+        is time time after which that setting becomes immutable.
+
+        If the bucket's :attr:`bucket_policy_only_enabled` is false, this property
+        is ``None``.
+
+        :rtype: Union[:class:`datetime.datetime`, None]
+        :returns:  (readonly) Time after which :attr:`bucket_policy_only_enabled` will
+                   be frozen as true.
+        """
+        bpo = self.get("bucketPolicyOnly", {})
+        stamp = bpo.get("lockedTime")
+        if stamp is not None:
+            stamp = _rfc3339_to_datetime(stamp)
+        return stamp
+
+
 class Bucket(_PropertyMixin):
     """A class representing a Bucket on Cloud Storage.
 
@@ -1134,6 +1224,16 @@ def id(self):
         """
         return self._properties.get("id")
 
+    @property
+    def iam_configuration(self):
+        """Retrieve IAM configuration for this bucket.
+
+        :rtype: :class:`IAMConfiguration`
+        :returns: an instance for managing the bucket's IAM configuration.
+        """
+        info = self._properties.get("iamConfiguration", {})
+        return IAMConfiguration.from_api_repr(info, self)
+
     @property
     def lifecycle_rules(self):
         """Retrieve or set lifecycle rules configured for this bucket.

storage/tests/system.py

@@ -1470,3 +1470,87 @@ def test_bucket_lock_retention_policy(self):
         bucket.retention_period = None
         with self.assertRaises(exceptions.Forbidden):
             bucket.patch()
+
+
+class TestIAMConfiguration(unittest.TestCase):
+    def setUp(self):
+        self.case_buckets_to_delete = []
+
+    def tearDown(self):
+        for bucket_name in self.case_buckets_to_delete:
+            bucket = Config.CLIENT.bucket(bucket_name)
+            retry_429(bucket.delete)(force=True)
+
+    def test_new_bucket_w_bpo(self):
+        new_bucket_name = "new-w-bpo" + unique_resource_id("-")
+        self.assertRaises(
+            exceptions.NotFound, Config.CLIENT.get_bucket, new_bucket_name
+        )
+        bucket = Config.CLIENT.bucket(new_bucket_name)
+        bucket.iam_configuration.bucket_policy_only_enabled = True
+        retry_429(bucket.create)()
+        self.case_buckets_to_delete.append(new_bucket_name)
+
+        bucket_acl = bucket.acl
+        with self.assertRaises(exceptions.BadRequest):
+            bucket_acl.reload()
+
+        bucket_acl.loaded = True  # Fake that we somehow loaded the ACL
+        bucket_acl.all().grant_read()
+        with self.assertRaises(exceptions.BadRequest):
+            bucket_acl.save()
+
+        blob_name = "my-blob.txt"
+        blob = bucket.blob(blob_name)
+        payload = b"DEADBEEF"
+        blob.upload_from_string(payload)
+
+        found = bucket.get_blob(blob_name)
+        self.assertEqual(found.download_as_string(), payload)
+
+        blob_acl = blob.acl
+        with self.assertRaises(exceptions.BadRequest):
+            blob_acl.reload()
+
+        blob_acl.loaded = True  # Fake that we somehow loaded the ACL
+        blob_acl.all().grant_read()
+        with self.assertRaises(exceptions.BadRequest):
+            blob_acl.save()
+
+    def test_bpo_set_unset_preserves_acls(self):
+        new_bucket_name = "bpo-acls" + unique_resource_id("-")
+        self.assertRaises(
+            exceptions.NotFound, Config.CLIENT.get_bucket, new_bucket_name
+        )
+        bucket = retry_429(Config.CLIENT.create_bucket)(new_bucket_name)
+        self.case_buckets_to_delete.append(new_bucket_name)
+
+        blob_name = "my-blob.txt"
+        blob = bucket.blob(blob_name)
+        payload = b"DEADBEEF"
+        blob.upload_from_string(payload)
+
+        # Preserve ACLs before setting BPO
+        bucket_acl_before = list(bucket.acl)
+        blob_acl_before = list(bucket.acl)
+
+        # Set BPO
+        bucket.iam_configuration.bucket_policy_only_enabled = True
+        bucket.patch()
+
+        # While BPO is set, cannot get / set ACLs
+        with self.assertRaises(exceptions.BadRequest):
+            bucket.acl.reload()
+
+        # Clear BPO
+        bucket.iam_configuration.bucket_policy_only_enabled = False
+        bucket.patch()
+
+        # Query ACLs after clearing BPO
+        bucket.acl.reload()
+        bucket_acl_after = list(bucket.acl)
+        blob.acl.reload()
+        blob_acl_after = list(bucket.acl)
+
+        self.assertEqual(bucket_acl_before, bucket_acl_after)
+        self.assertEqual(blob_acl_before, blob_acl_after)

storage/tests/unit/test_blob.py

@@ -2508,13 +2508,13 @@ def test_update_storage_class_large_file(self):
             "objectSize": 84,
             "done": False,
             "rewriteToken": TOKEN,
-            "resource": {"storageClass": STORAGE_CLASS}
+            "resource": {"storageClass": STORAGE_CLASS},
         }
         COMPLETE_RESPONSE = {
             "totalBytesRewritten": 84,
             "objectSize": 84,
             "done": True,
-            "resource": {"storageClass": STORAGE_CLASS}
+            "resource": {"storageClass": STORAGE_CLASS},
         }
         response_1 = ({"status": http_client.OK}, INCOMPLETE_RESPONSE)
         response_2 = ({"status": http_client.OK}, COMPLETE_RESPONSE)
@@ -2534,7 +2534,7 @@ def test_update_storage_class_wo_encryption_key(self):
             "totalBytesRewritten": 42,
             "objectSize": 42,
             "done": True,
-            "resource": {"storageClass": STORAGE_CLASS}
+            "resource": {"storageClass": STORAGE_CLASS},
         }
         response = ({"status": http_client.OK}, RESPONSE)
         connection = _Connection(response)
@@ -2579,7 +2579,7 @@ def test_update_storage_class_w_encryption_key_w_user_project(self):
             "totalBytesRewritten": 42,
             "objectSize": 42,
             "done": True,
-            "resource": {"storageClass": STORAGE_CLASS}
+            "resource": {"storageClass": STORAGE_CLASS},
         }
         response = ({"status": http_client.OK}, RESPONSE)
         connection = _Connection(response)