Refactor XML handling of {set,get,delete}_bucket_encryption APIs

docs/API.md

@@ -53,7 +53,7 @@ s3Client = Minio(
 | [`set_bucket_policy`](#set_bucket_policy)                   |                                                   |                                                   |
 | [`delete_bucket_policy`](#delete_bucket_policy)             |                                                   |                                                   |
 | [`get_bucket_encryption`](#get_bucket_encryption)           |                                                   |                                                   |
-| [`put_bucket_encryption`](#put_bucket_encryption)           |                                                   |                                                   |
+| [`set_bucket_encryption`](#set_bucket_encryption)           |                                                   |                                                   |
 | [`delete_bucket_encryption`](#delete_bucket_encryption)     |                                                   |                                                   |
 
 ## 1. Constructor
@@ -447,42 +447,35 @@ __Parameters__
 
 __Return Value__
 
-| Param                               |
-|:------------------------------------|
-| Encryption configuration as _dict_. |
+| Param               |
+|:--------------------|
+| _SSEConfig_ object. |
 
 __Example__
 
 ```py
 config = minio.get_bucket_encryption("my-bucketname")
 ```
 
-<a name="put_bucket_encryption"></a>
+<a name="set_bucket_encryption"></a>
 
-### put_bucket_encryption(bucket_name, encryption_configuration)
+### set_bucket_encryption(bucket_name, config)
 
 Set encryption configuration of a bucket.
 
 __Parameters__
 
-| Param           | Type   | Description                                       |
-|:----------------|:-------|:--------------------------------------------------|
-| ``bucket_name`` | _str_  | Name of the bucket.                               |
-| ``enc_config``  | _dict_ | Encryption configuration as dictionary to be set. |
+| Param           | Type        | Description                           |
+|:----------------|:------------|:--------------------------------------|
+| ``bucket_name`` | _str_       | Name of the bucket.                   |
+| ``config``      | _SSEConfig_ | Server-side encryption configuration. |
 
 __Example__
 
 ```py
-# Sample default encryption configuration
-config = {
-    'ServerSideEncryptionConfiguration':{
-        'Rule': [
-            {'ApplyServerSideEncryptionByDefault': {'SSEAlgorithm': 'AES256'}}
-        ]
-    }
-}
-
-minio.put_bucket_encryption("my-bucketname", config)
+minio.set_bucket_encryption(
+    "my-bucketname", SSEConfig(Rule.new_sse_s3_rule()),
+)
 ```
 
 <a name="delete_bucket_encryption"></a>

examples/remove_bucket_encryption.py → examples/delete_bucket_encryption.py

@@ -18,15 +18,11 @@
 # dummy values, please replace them with original values.
 
 from minio import Minio
-from minio.error import ResponseError
 
-client = Minio('s3.amazonaws.com',
-               access_key='YOUR-ACCESSKEYID',
-               secret_key='YOUR-SECRETACCESSKEY',
-               secure=True)
+client = Minio(
+    "play.min.io",
+    access_key="Q3AM3UQ867SPQQA43P2F",
+    secret_key="zuf+tfteSlswRu7BJ86wekitnifILbZam1KYY3TG",
+)
 
-try:
-    # Delete default encryption configuration on bucket 'my-bucketname'.
-    client.delete_bucket_encryption('my-bucketname')
-except ResponseError as err:
-    print(err)
+client.delete_bucket_encryption("my-bucketname")

examples/get_bucket_encryption.py

@@ -18,15 +18,11 @@
 # dummy values, please replace them with original values.
 
 from minio import Minio
-from minio.error import ResponseError
 
-client = Minio('s3.amazonaws.com',
-               access_key='YOUR-ACCESSKEYID',
-               secret_key='YOUR-SECRETACCESSKEY',
-               secure=True)
+client = Minio(
+    "play.min.io",
+    access_key="Q3AM3UQ867SPQQA43P2F",
+    secret_key="zuf+tfteSlswRu7BJ86wekitnifILbZam1KYY3TG",
+)
 
-try:
-    # Get current policy of bucket 'my-bucketname'.
-    print(client.get_bucket_encryption('my-bucketname'))
-except ResponseError as err:
-    print(err)
+config = client.get_bucket_encryption("my-bucketname")

examples/set_bucket_encryption.py

@@ -18,27 +18,14 @@
 # dummy values, please replace them with original values.
 
 from minio import Minio
-from minio.error import ResponseError
+from minio.sseconfig import Rule, SSEConfig
 
-client = Minio('s3.amazonaws.com',
-               access_key='YOUR-ACCESSKEYID',
-               secret_key='YOUR-SECRETACCESSKEY',
-               secure=True)
+client = Minio(
+    "play.min.io",
+    access_key="Q3AM3UQ867SPQQA43P2F",
+    secret_key="zuf+tfteSlswRu7BJ86wekitnifILbZam1KYY3TG",
+)
 
-try:
-    # Set default encryption configuration for bucket 'my-bucketname'
-    ENC_CONFIG = {
-        'ServerSideEncryptionConfiguration': {
-            'Rule': [
-                {
-                    'ApplyServerSideEncryptionByDefault': {
-                        'SSEAlgorithm': 'AES256'
-                    }
-                }
-            ]
-        }
-    }
-
-    client.put_bucket_encryption('my-bucketname', ENC_CONFIG)
-except ResponseError as err:
-    print(err)
+client.set_bucket_encryption(
+    "my-bucketname", SSEConfig(Rule.new_sse_s3_rule()),
+)

minio/api.py

@@ -65,13 +65,12 @@
 from .signer import (AMZ_DATE_FORMAT, SIGN_V4_ALGORITHM, get_credential_string,
                      post_presign_v4, presign_v4, sign_v4_s3)
 from .sse import SseCustomerKey
+from .sseconfig import SSEConfig
 from .thread_pool import ThreadPool
 from .versioningconfig import VersioningConfig
 from .xml import Element, SubElement, marshal, unmarshal
 from .xml_marshal import (marshal_complete_multipart,
-                          xml_marshal_bucket_encryption,
-                          xml_marshal_delete_objects, xml_marshal_select,
-                          xml_to_dict)
+                          xml_marshal_delete_objects, xml_marshal_select)
 
 try:
     from json.decoder import JSONDecodeError
@@ -760,22 +759,22 @@ def delete_bucket_notification(self, bucket_name):
         """
         self.set_bucket_notification(bucket_name, NotificationConfig())
 
-    def put_bucket_encryption(self, bucket_name, enc_config):
+    def set_bucket_encryption(self, bucket_name, config):
         """
         Set encryption configuration of a bucket.
 
         :param bucket_name: Name of the bucket.
-        :param enc_config: Encryption configuration as dictionary to be set.
+        :param config: :class:`SSEConfig <SSEConfig>` object.
 
         Example::
-            minio.put_bucket_encryption("my-bucketname", config)
+            minio.set_bucket_encryption(
+                "my-bucketname", SSEConfig(Rule.new_sse_s3_rule()),
+            )
         """
         check_bucket_name(bucket_name)
-
-        # 'Rule' is a list, so we need to go through each one of
-        # its key/value pair and collect the encryption values.
-        rules = enc_config['ServerSideEncryptionConfiguration']['Rule']
-        body = xml_marshal_bucket_encryption(rules)
+        if not isinstance(config, SSEConfig):
+            raise ValueError("config must be SSEConfig type")
+        body = marshal(config)
         self._execute(
             "PUT",
             bucket_name,
@@ -789,18 +788,23 @@ def get_bucket_encryption(self, bucket_name):
         Get encryption configuration of a bucket.
 
         :param bucket_name: Name of the bucket.
-        :return: Encryption configuration.
+        :return: :class:`SSEConfig <SSEConfig>` object.
 
         Example::
             config = minio.get_bucket_encryption("my-bucketname")
         """
         check_bucket_name(bucket_name)
-        response = self._execute(
-            "GET",
-            bucket_name,
-            query_params={"encryption": ""},
-        )
-        return xml_to_dict(response.data.decode())
+        try:
+            response = self._execute(
+                "GET",
+                bucket_name,
+                query_params={"encryption": ""},
+            )
+            return unmarshal(SSEConfig, response.data.decode())
+        except S3Error as exc:
+            if exc.code != "ServerSideEncryptionConfigurationNotFoundError":
+                raise
+        return None
 
     def delete_bucket_encryption(self, bucket_name):
         """
@@ -812,11 +816,15 @@ def delete_bucket_encryption(self, bucket_name):
             minio.delete_bucket_encryption("my-bucketname")
         """
         check_bucket_name(bucket_name)
-        self._execute(
-            "DELETE",
-            bucket_name,
-            query_params={"encryption": ""},
-        )
+        try:
+            self._execute(
+                "DELETE",
+                bucket_name,
+                query_params={"encryption": ""},
+            )
+        except S3Error as exc:
+            if exc.code != "ServerSideEncryptionConfigurationNotFoundError":
+                raise
 
     def listen_bucket_notification(self, bucket_name, prefix='', suffix='',
                                    events=('s3:ObjectCreated:*',

minio/sseconfig.py

@@ -0,0 +1,98 @@
+# -*- coding: utf-8 -*-
+# MinIO Python Library for Amazon S3 Compatible Cloud Storage, (C)
+# 2020 MinIO, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Request/response of PutBucketEncryption and GetBucketEncryption APIs."""
+
+from __future__ import absolute_import
+
+from abc import ABCMeta
+
+from .xml import Element, SubElement, find, findtext
+
+AES256 = "AES256"
+AWS_KMS = "aws:kms"
+
+
+class Rule:
+    """Server-side encryption rule. """
+    __metaclass__ = ABCMeta
+
+    def __init__(self, sse_algorithm, kms_master_key_id=None):
+        self._sse_algorithm = sse_algorithm
+        self._kms_master_key_id = kms_master_key_id
+
+    @property
+    def sse_algorithm(self):
+        """Get SSE algorithm."""
+        return self._sse_algorithm
+
+    @property
+    def kms_master_key_id(self):
+        """Get KMS master key ID."""
+        return self._kms_master_key_id
+
+    @classmethod
+    def new_sse_s3_rule(cls):
+        """Create SSE-S3 rule."""
+        return cls(AES256)
+
+    @classmethod
+    def new_sse_kms_rule(cls, kms_master_key_id=None):
+        """Create new SSE-KMS rule."""
+        return cls(AWS_KMS, kms_master_key_id)
+
+    @classmethod
+    def fromxml(cls, element):
+        """Create new object with values from XML element."""
+        element = find(element, "ApplyServerSideEncryptionByDefault")
+        sse_algorithm = findtext(element, "SSEAlgorithm", True)
+        kms_master_key_id = findtext(element, "KMSMasterKeyID")
+        return cls(sse_algorithm, kms_master_key_id)
+
+    def toxml(self, element):
+        """Convert to XML."""
+        element = SubElement(element, "Rule")
+        tag = SubElement(element, "ApplyServerSideEncryptionByDefault")
+        SubElement(tag, "SSEAlgorithm", self._sse_algorithm)
+        if self._kms_master_key_id is not None:
+            SubElement(tag, "KMSMasterKeyID", self._kms_master_key_id)
+        return element
+
+
+class SSEConfig:
+    """server-side encyption configuration."""
+
+    def __init__(self, rule):
+        if not rule:
+            raise ValueError("rule must be provided")
+        self._rule = rule
+
+    @property
+    def rule(self):
+        """Get rule."""
+        return self._rule
+
+    @classmethod
+    def fromxml(cls, element):
+        """Create new object with values from XML element."""
+        element = find(element, "Rule")
+        return cls(Rule.fromxml(element))
+
+    def toxml(self, element):
+        """Convert to XML."""
+        element = Element("ServerSideEncryptionConfiguration")
+        self._rule.toxml(element)
+        return element

minio/xml_marshal.py

@@ -90,27 +90,6 @@ def xml_to_dict(in_xml):
     return _etree_to_dict(elem)
 
 
-def xml_marshal_bucket_encryption(rules):
-    """Encode bucket encryption to XML."""
-
-    root = Element('ServerSideEncryptionConfiguration')
-
-    if rules:
-        # As server supports only one rule, the first rule is taken due to
-        # no validation is done at server side.
-        apply_element = SubElement(SubElement(root, 'Rule'),
-                                   'ApplyServerSideEncryptionByDefault')
-        SubElement(apply_element, 'SSEAlgorithm',
-                   rules[0]['ApplyServerSideEncryptionByDefault'].get(
-                       'SSEAlgorithm', 'AES256'))
-        kms_text = rules[0]['ApplyServerSideEncryptionByDefault'].get(
-            'KMSMasterKeyID')
-        if kms_text:
-            SubElement(apply_element, 'KMSMasterKeyID', kms_text)
-
-    return _get_xml_data(root)
-
-
 def xml_marshal_select(req):
     """Encode select request to XML."""