Storage: add 'bucket_policy_only' IAM property

storage/google/cloud/storage/blob.py

@@ -1533,7 +1533,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,82 @@ 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 enabled: bool
+    :params enabled: (optional) whether the IAM-only policy is enabled for the bucket.
+
+    :type locked_time: :class:`datetime.datetime`
+    :params 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, enabled=False, locked_time=None):
+        data = {"bucketPolicyOnly": {"enabled": enabled}}
+        if locked_time is not None:
+            data["bucketPolicyOnly"]["lockedTime"] = _datetime_to_rfc3339(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(self):
+        """Is the bucket configured to allow only IAM policy?
+
+        :rtype: bool
+        :returns: whether the bucket is configured to allow only IAM.
+        """
+        bpo = self.get("bucketPolicyOnly", {})
+        return bpo.get("enabled", False)
+
+    @bucket_policy_only.setter
+    def bucket_policy_only(self, value):
+        bpo = self.setdefault("bucketPolicyOnly", {})
+        bpo["enabled"] = bool(value)
+        self.bucket._patch_property("iamConfiguration", self)
+
+    @property
+    def locked_time(self):
+        """When was the bucket configured to allow only IAM policy?
+
+        :rtype: Union[:class:`datetime.datetime`, None]
+        :returns:  (readonly) the time the bucket's IAM-only policy was set.
+        """
+        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 +1210,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,88 @@ 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 = 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)
+        blob_policy = blob.get_iam_policy()
+
+        # 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 = 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 = 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)

storage/tests/unit/test_bucket.py

@@ -175,6 +175,108 @@ def test_from_api_repr(self):
         self.assertEqual(dict(rule), resource)
 
 
+class Test_IAMConfiguration(unittest.TestCase):
+    @staticmethod
+    def _get_target_class():
+        from google.cloud.storage.bucket import IAMConfiguration
+
+        return IAMConfiguration
+
+    def _make_one(self, bucket, **kw):
+        return self._get_target_class()(bucket, **kw)
+
+    @staticmethod
+    def _make_bucket():
+        from google.cloud.storage.bucket import Bucket
+
+        return mock.create_autospec(Bucket, instance=True)
+
+    def test_ctor_defaults(self):
+        bucket = self._make_bucket()
+
+        config = self._make_one(bucket)
+
+        self.assertIs(config.bucket, bucket)
+        self.assertFalse(config.bucket_policy_only)
+        self.assertIsNone(config.locked_time)
+
+    def test_ctor_explicit(self):
+        import datetime
+        import pytz
+
+        bucket = self._make_bucket()
+        now = datetime.datetime.utcnow().replace(tzinfo=pytz.UTC)
+
+        config = self._make_one(bucket, enabled=True, locked_time=now)
+
+        self.assertIs(config.bucket, bucket)
+        self.assertTrue(config.bucket_policy_only)
+        self.assertEqual(config.locked_time, now)
+
+    def test_from_api_repr_w_empty_resource(self):
+        klass = self._get_target_class()
+        bucket = self._make_bucket()
+        resource = {}
+
+        config = klass.from_api_repr(resource, bucket)
+
+        self.assertIs(config.bucket, bucket)
+        self.assertFalse(config.bucket_policy_only)
+        self.assertIsNone(config.locked_time)
+
+    def test_from_api_repr_w_empty_bpo(self):
+        klass = self._get_target_class()
+        bucket = self._make_bucket()
+        resource = {"bucketPolicyOnly": {}}
+
+        config = klass.from_api_repr(resource, bucket)
+
+        self.assertIs(config.bucket, bucket)
+        self.assertFalse(config.bucket_policy_only)
+        self.assertIsNone(config.locked_time)
+
+    def test_from_api_repr_w_disabled(self):
+        klass = self._get_target_class()
+        bucket = self._make_bucket()
+        resource = {"bucketPolicyOnly": {"enabled": False}}
+
+        config = klass.from_api_repr(resource, bucket)
+
+        self.assertIs(config.bucket, bucket)
+        self.assertFalse(config.bucket_policy_only)
+        self.assertIsNone(config.locked_time)
+
+    def test_from_api_repr_w_enabled(self):
+        import datetime
+        import pytz
+        from google.cloud._helpers import _datetime_to_rfc3339
+
+        klass = self._get_target_class()
+        bucket = self._make_bucket()
+        now = datetime.datetime.utcnow().replace(tzinfo=pytz.UTC)
+        resource = {
+            "bucketPolicyOnly": {
+                "enabled": True,
+                "lockedTime": _datetime_to_rfc3339(now),
+            }
+        }
+
+        config = klass.from_api_repr(resource, bucket)
+
+        self.assertIs(config.bucket, bucket)
+        self.assertTrue(config.bucket_policy_only)
+        self.assertEqual(config.locked_time, now)
+
+    def test_bucket_policy_only_setter(self):
+        bucket = self._make_bucket()
+        config = self._make_one(bucket)
+
+        config.bucket_policy_only = True
+
+        self.assertTrue(config["bucketPolicyOnly"]["enabled"])
+        bucket._patch_property.assert_called_once_with("iamConfiguration", config)
+
+
 class Test_Bucket(unittest.TestCase):
     @staticmethod
     def _get_target_class():
@@ -1092,6 +1194,44 @@ def test_location_setter(self, mock_warn):
             bucket_module._LOCATION_SETTER_MESSAGE, DeprecationWarning, stacklevel=2
         )
 
+    def test_iam_configuration_policy_missing(self):
+        from google.cloud.storage.bucket import IAMConfiguration
+
+        NAME = "name"
+        bucket = self._make_one(name=NAME)
+
+        config = bucket.iam_configuration
+
+        self.assertIsInstance(config, IAMConfiguration)
+        self.assertIs(config.bucket, bucket)
+        self.assertFalse(config.bucket_policy_only)
+        self.assertIsNone(config.locked_time)
+
+    def test_iam_configuration_policy_w_entry(self):
+        import datetime
+        import pytz
+        from google.cloud._helpers import _datetime_to_rfc3339
+        from google.cloud.storage.bucket import IAMConfiguration
+
+        now = datetime.datetime.utcnow().replace(tzinfo=pytz.UTC)
+        NAME = "name"
+        properties = {
+            "iamConfiguration": {
+                "bucketPolicyOnly": {
+                    "enabled": True,
+                    "lockedTime": _datetime_to_rfc3339(now),
+                }
+            }
+        }
+        bucket = self._make_one(name=NAME, properties=properties)
+
+        config = bucket.iam_configuration
+
+        self.assertIsInstance(config, IAMConfiguration)
+        self.assertIs(config.bucket, bucket)
+        self.assertTrue(config.bucket_policy_only)
+        self.assertEqual(config.locked_time, now)
+
     def test_lifecycle_rules_getter_unknown_action_type(self):
         NAME = "name"
         BOGUS_RULE = {"action": {"type": "Bogus"}, "condition": {"age": 42}}