Add the BucketHttpsConfig interface (#383)

examples/bucket_https_config.py

@@ -0,0 +1,35 @@
+import os
+import oss2
+
+# Specify access information, such as AccessKeyId, AccessKeySecret, and Endpoint.
+# You can obtain access information from evironment variables or replace sample values in the code, such as <your AccessKeyId> with actual values.
+#
+# For example, if your bucket is located in the China (Hangzhou) region, you can set Endpoint to one of the following values:
+#   http://oss-cn-hangzhou.aliyuncs.com
+#   https://oss-cn-hangzhou.aliyuncs.com
+from oss2.models import BucketTlsVersion
+
+access_key_id = os.getenv('OSS_TEST_ACCESS_KEY_ID', '<yourAccessKeyId>')
+access_key_secret = os.getenv('OSS_TEST_ACCESS_KEY_SECRET', '<yourAccessKeySecret>')
+bucket_name = os.getenv('OSS_TEST_BUCKET', '<yourBucketName>')
+endpoint = os.getenv('OSS_TEST_ENDPOINT', '<yourEndpoint>')
+
+
+# Make sure that all parameters are correctly configured
+for param in (access_key_id, access_key_secret, bucket_name, endpoint):
+    assert '<' not in param, 'Please set parameters：' + param
+
+
+# Create a bucket. You can use the bucket to call all object-related operations
+bucket = oss2.Bucket(oss2.Auth(access_key_id, access_key_secret), endpoint, bucket_name)
+
+# Configure transfer acceleration for the bucket.
+# If enabled is set to true, transfer acceleration is enabled. If enabled is set to false, transfer acceleration is disabled.
+https_config = BucketTlsVersion(True, ['TLSv1.2', 'TLSv1.3'])
+bucket.put_bucket_https_config(https_config)
+
+# Query the transfer acceleration status of the bucket.
+# If the returned value is true, the transfer acceleration feature is enabled for the bucket. If the returned value is false, the transfer acceleration feature is disabled for the bucket.
+result = bucket.get_bucket_https_config()
+print("Return information on whether to enable TLS version settings: {0}".format(result.tls_enabled))
+print("Return TLS version number: {0}".format(result.tls_version))

oss2/api.py

@@ -470,8 +470,8 @@ class Bucket(_Base):
     STYLE_NAME = 'styleName'
     ASYNC_PROCESS = 'x-oss-async-process'
     CALLBACK = 'callback'
-    ARCHIVE_DIRECT_READ = "bucketArchiveDirectRead";
-
+    ARCHIVE_DIRECT_READ = "bucketArchiveDirectRead"
+    HTTPS_CONFIG = 'httpsConfig'
 
     def __init__(self, auth, endpoint, bucket_name,
                  is_cname=False,
@@ -2894,6 +2894,26 @@ def get_bucket_archive_direct_read(self):
         logger.debug("Get bucket archive direct read done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
         return self._parse_result(resp, xml_utils.parse_get_bucket_archive_direct_read, GetBucketArchiveDirectReadResult)
 
+    def put_bucket_https_config(self, httpsConfig):
+        """Bucket开启或关闭TLS版本设置。
+        :param httpsConfig: TLS版本信息设置
+        """
+        logger.debug("Start to put bucket https config, bucket: {0}, https config: {1}".format(self.bucket_name, httpsConfig))
+        data = xml_utils.to_do_bucket_https_config_request(httpsConfig)
+        resp = self.__do_bucket('PUT', data=data, params={Bucket.HTTPS_CONFIG: ''})
+        logger.debug("Put bucket https config done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
+
+        return RequestResult(resp)
+
+    def get_bucket_https_config(self):
+        """查看Bucket的TLS版本设置。
+        :return: :class:`HttpsConfigResult <oss2.models.HttpsConfigResult>`
+        """
+        logger.debug("Start to get bucket https config, bucket: {0}".format(self.bucket_name))
+        resp = self.__do_bucket('GET', params={Bucket.HTTPS_CONFIG: ''})
+        logger.debug("Get bucket https config done, req_id: {0}, status_code: {1}".format(resp.request_id, resp.status))
+        return self._parse_result(resp, xml_utils.parse_get_bucket_https_config, HttpsConfigResult)
+
     def __do_object(self, method, key, **kwargs):
         if not self.bucket_name:
             raise ClientError("Bucket name should not be null or empty.")

oss2/auth.py

@@ -92,7 +92,8 @@ class ProviderAuth(AuthBase):
          'callback-var', 'worm', 'wormId', 'wormExtend', 'replication', 'replicationLocation',
          'replicationProgress', 'transferAcceleration', 'cname', 'metaQuery',
          'x-oss-ac-source-ip', 'x-oss-ac-subnet-mask', 'x-oss-ac-vpc-id', 'x-oss-ac-forward-allow',
-         'resourceGroup', 'style', 'styleName', 'x-oss-async-process', 'regionList', 'x-oss-write-get-object-response', 'bucketArchiveDirectRead']
+         'resourceGroup', 'style', 'styleName', 'x-oss-async-process', 'regionList', 'x-oss-write-get-object-response',
+         'bucketArchiveDirectRead', 'httpsConfig']
     )
 
     def _sign_request(self, req, bucket_name, key):

oss2/models.py

@@ -2761,7 +2761,6 @@ def __init__(self, resp):
         super(CallbackPolicyResult, self).__init__(resp)
         self.callback_policies = []
 
-
 class GetBucketArchiveDirectReadResult(RequestResult):
     """获取归档直读。
 
@@ -2770,3 +2769,25 @@ class GetBucketArchiveDirectReadResult(RequestResult):
     def __init__(self, resp):
         super(GetBucketArchiveDirectReadResult, self).__init__(resp)
         self.enabled = None
+
+class BucketTlsVersion(object):
+    """BucketTLS版本设置。
+
+    :param bool tls_enabled: 是否为Bucket开启TLS版本设置。
+    :param tls_version: TLS版本。
+    """
+    def __init__(self, tls_enabled=False, tls_version=None):
+        self.tls_enabled = tls_enabled
+        self.tls_version = tls_version
+
+class HttpsConfigResult(RequestResult):
+    """返回Bucket TLS版本信息。
+
+    :param bool tls_enabled: bucket是否开启TLS版本设置。
+    :param tls_version: TLS版本。
+    """
+
+    def __init__(self, resp):
+        super(HttpsConfigResult, self).__init__(resp)
+        self.tls_enabled = None
+        self.tls_version = []

oss2/xml_utils.py

@@ -2097,7 +2097,6 @@ def parse_callback_policy_result(result, body):
 
         result.callback_policies.append(tmp)
 
-
 def to_put_bucket_archive_direct_read(enabled):
     root = ElementTree.Element('ArchiveDirectReadConfiguration')
     _add_text_child(root, 'Enabled', str(enabled).lower())
@@ -2107,3 +2106,25 @@ def parse_get_bucket_archive_direct_read(result, body):
     root = ElementTree.fromstring(body)
     if root.find("Enabled") is not None:
         result.enabled = _find_bool(root, "Enabled")
+
+def to_do_bucket_https_config_request(https_config):
+    root = ElementTree.Element('HttpsConfiguration')
+
+    list_node = ElementTree.SubElement(root, 'TLS')
+
+    _add_text_child(list_node, 'Enable', str(https_config.tls_enabled).lower())
+    if https_config.tls_version:
+        for r in https_config.tls_version:
+            _add_text_child(list_node, 'TLSVersion', r)
+
+    return _node_to_string(root)
+
+
+def parse_get_bucket_https_config(result, body):
+    root = ElementTree.fromstring(body)
+
+    result.tls_enabled = _find_bool(root, 'TLS/Enable')
+    if root.find("TLS/TLSVersion") is not None:
+        result.tls_version = _find_all_tags(root, 'TLS/TLSVersion')
+
+    return result

tests/test_bucket_https_config.py

@@ -0,0 +1,52 @@
+from oss2.models import BucketTlsVersion
+from .common import *
+
+class TestHttpsConfig(OssTestCase):
+    def test_https_config_normal(self):
+        https_config = BucketTlsVersion(True, ['TLSv1.2', 'TLSv1.3'])
+        result = self.bucket.put_bucket_https_config(https_config)
+        self.assertEqual(200, result.status)
+
+        result2 = self.bucket.get_bucket_https_config()
+        self.assertEqual(200, result2.status)
+        self.assertEqual(result2.tls_enabled, True)
+        self.assertListEqual(result2.tls_version, ['TLSv1.2', 'TLSv1.3'])
+
+        https_config2 = BucketTlsVersion()
+        result3 = self.bucket.put_bucket_https_config(https_config2)
+        self.assertEqual(200, result3.status)
+
+        result4 = self.bucket.get_bucket_https_config()
+        self.assertEqual(200, result4.status)
+        self.assertEqual(result4.tls_enabled, False)
+        self.assertListEqual(result4.tls_version, [])
+
+
+    def test_https_config_exception_1(self):
+        try:
+            https_config = BucketTlsVersion(True)
+            self.bucket.put_bucket_https_config(https_config)
+            self.assertTrue(False)
+        except oss2.exceptions.ServerError as e:
+            self.assertEqual(e.code, 'MalformedXML')
+
+    def test_https_config_exception_2(self):
+        try:
+            https_config = BucketTlsVersion(True, ['aaa', 'bbb'])
+            self.bucket.put_bucket_https_config(https_config)
+            self.assertTrue(False)
+        except oss2.exceptions.ServerError as e:
+            self.assertEqual(e.code, 'MalformedXML')
+
+    def test_https_config_exception_3(self):
+        https_config = BucketTlsVersion(True, ['TLSv1.2', 'TLSv1.2'])
+        result = self.bucket.put_bucket_https_config(https_config)
+        self.assertEqual(200, result.status)
+
+        result2 = self.bucket.get_bucket_https_config()
+        self.assertEqual(200, result2.status)
+        self.assertEqual(result2.tls_enabled, True)
+        self.assertListEqual(result2.tls_version, ['TLSv1.2'])
+
+if __name__ == '__main__':
+    unittest.main()

unittests/test_bucket.py

@@ -7,7 +7,7 @@
 
 from oss2 import to_string, iso8601_to_unixtime
 from oss2.headers import OSS_ALLOW_ACTION_OVERLAP
-from oss2.models import AggregationsRequest, MetaQuery, CallbackPolicyInfo
+from oss2.models import AggregationsRequest, MetaQuery, CallbackPolicyInfo, BucketTlsVersion
 from unittests.common import *
 
 
@@ -3252,5 +3252,70 @@ def test_get_bucket_archive_direct_read(self, do_request):
         self.assertEqual(result.status, 200)
         self.assertEqual(result.enabled, False)
 
+    @patch('oss2.Session.do_request')
+    def test_put_bucket_https_config(self, do_request):
+
+        request_text = '''PUT /?httpsConfig= HTTP/1.1
+Host: ming-oss-share.oss-cn-hangzhou.aliyuncs.com
+Accept-Encoding: identity
+Connection: keep-alive
+Content-Length: 249
+date: Sat, 12 Dec 2015 00:35:46 GMT
+User-Agent: aliyun-sdk-python/2.0.2(Windows/7/;3.3.3)
+Accept: */*
+authorization: OSS ZCDmm7TPZKHtx77j:Kq2RS9nmT44C1opXGbcLzNdTt1A=
+
+<HttpsConfiguration><TLS><Enable>true</Enable><TLSVersion>TLSv1.2</TLSVersion><TLSVersion>TLSv1.3</TLSVersion></TLS></HttpsConfiguration>'''
+
+        response_text = '''HTTP/1.1 200 OK
+Server: AliyunOSS
+Date: Sat, 12 Dec 2015 00:35:46 GMT
+Content-Length: 0
+Connection: keep-alive
+x-oss-request-id: 566B6BE244ABFA2608E5A8AD'''
+
+        req_info = mock_response(do_request, response_text)
+
+        bucket().put_bucket_https_config(BucketTlsVersion(True, ['TLSv1.2', 'TLSv1.3']))
+
+        self.assertRequest(req_info, request_text)
+
+    @patch('oss2.Session.do_request')
+    def test_get_bucket_https_config(self, do_request):
+        request_text = '''GET /?httpsConfig= HTTP/1.1
+Host: ming-oss-share.oss-cn-hangzhou.aliyuncs.com
+Accept-Encoding: identity
+Connection: keep-alive
+date: Sat, 12 Dec 2015 00:35:47 GMT
+User-Agent: aliyun-sdk-python/2.0.2(Windows/7/;3.3.3)
+Accept: */*
+authorization: OSS ZCDmm7TPZKHtx77j:nWqS3JExf/lsVxm+Sbxbg2cQyrc='''
+
+        response_text = '''HTTP/1.1 200 OK
+Server: AliyunOSS
+Date: Sat, 12 Dec 2015 00:35:47 GMT
+Content-Type: application/xml
+Content-Length: 319
+Connection: keep-alive
+x-oss-request-id: 566B6BE3BCD1D4FE65D449A2
+
+<?xml version="1.0" encoding="UTF-8"?>
+<HttpsConfiguration>
+  <TLS>
+    <Enable>true</Enable>
+    <TLSVersion>TLSv1.2</TLSVersion>
+    <TLSVersion>TLSv1.3</TLSVersion>
+  </TLS>
+</HttpsConfiguration>'''
+
+        req_info = mock_response(do_request, response_text)
+
+        result = bucket().get_bucket_https_config()
+
+        self.assertRequest(req_info, request_text)
+
+        self.assertEqual(result.tls_enabled, True)
+        self.assertSortedListEqual(result.tls_version, ['TLSv1.2', 'TLSv1.3'])
+
 if __name__ == '__main__':
     unittest.main()