支持创建多AZ桶, 修复可签入签名的headers列表, 配置自定义域名证书

qcloud_cos/cos_auth.py

@@ -33,14 +33,7 @@ def filter_headers(data):
         "if-unmodified-since",
         "origin",
         "range",
-        "response-cache-control",
-        "response-content-disposition",
-        "response-content-encoding",
-        "response-content-language",
-        "response-content-type",
-        "response-expires",
         "transfer-encoding",
-        "versionid",
     ]
     headers = {}
     for i in data:

qcloud_cos/cos_client.py

@@ -1273,23 +1273,35 @@ def select_object_content(self, Bucket, Key, Expression, ExpressionType, InputSe
         return data
 
     # s3 bucket interface begin
-    def create_bucket(self, Bucket, **kwargs):
+    def create_bucket(self, Bucket, BucketAZConfig=None, **kwargs):
         """创建一个bucket
 
         :param Bucket(string): 存储桶名称.
+        :param BucketAZConfig(string): 存储桶的多AZ配置
         :param kwargs(dict): 设置请求headers.
         :return: None.
 
         .. code-block:: python
 
             config = CosConfig(Region=region, SecretId=secret_id, SecretKey=secret_key, Token=token)  # 获取配置对象
             client = CosS3Client(config)
-            # 创建bucket
+            # 创建单AZ bucket
             response = client.create_bucket(
                 Bucket='bucket'
             )
+            # 创建多AZ bucket
+            response = client.create_bucket(
+                Bucket='bucket',
+                BucketAZConfig='MAZ'
+            )
         """
         headers = mapped(kwargs)
+        xml_config = None
+        if BucketAZConfig == 'MAZ':
+            bucket_config = {'BucketAZConfig': 'MAZ'}
+            xml_config = format_xml(data=bucket_config, root='CreateBucketConfiguration')
+            headers['Content-MD5'] = get_md5(xml_config)
+            headers['Content-Type'] = 'application/xml'
         url = self._conf.uri(bucket=Bucket)
         logger.info("create bucket, url=:{url} ,headers=:{headers}".format(
             url=url,
@@ -1298,6 +1310,7 @@ def create_bucket(self, Bucket, **kwargs):
             method='PUT',
             url=url,
             bucket=Bucket,
+            data=xml_config,
             auth=CosS3Auth(self._conf),
             headers=headers)
         return None
@@ -2569,6 +2582,119 @@ def delete_bucket_domain(self, Bucket, **kwargs):
             params=params)
         return None
 
+    def put_bucket_domain_certificate(self, Bucket, DomainCertificateConfiguration, **kwargs):
+        """设置bucket的自定义域名证书配置规则
+
+        :param Bucket(string): 存储桶名称.
+        :param DomainCertificateConfiguration(dict): 设置Bucket的自定义域名证书配置规则.
+        :param kwargs(dict): 设置请求headers.
+        :return: None
+
+        .. code-block:: python
+            config = CosConfig(Region=region, SecretId=secret_id, SecretKey=secret_key, Token=token)  # 获取配置对象
+            client = CosS3Client(config)
+            # 设置bucket自定义域名证书配置规则
+            domain_cert_config = {}
+            response = client.put_bucket_domain_certificate(
+                Bucket='bucket',
+                DomainCertificateConfiguration=domain_cert_config
+            )
+        """
+        lst = ['<DomainList>', '</DomainList>']  # 类型为list的标签
+        xml_config = format_xml(data=DomainCertificateConfiguration, root='DomainCertificate', lst=lst)
+        headers = mapped(kwargs)
+        headers['Content-MD5'] = get_md5(xml_config)
+        headers['Content-Type'] = 'application/xml'
+        # params = {'domaincertificate': ''}
+        # 目前 Domain Certificate API 不能使用 params 传递 query_string '?domaincertificate=',
+        # 只能将'?domaincertificate'拼接到url
+        url = self._conf.uri(bucket=Bucket)
+        url += '?domaincertificate'
+        logger.info("put bucket domain certificate, url=:{url} ,headers=:{headers}".format(
+            url=url,
+            headers=headers))
+        rt = self.send_request(
+            method='PUT',
+            url=url,
+            bucket=Bucket,
+            data=xml_config,
+            auth=CosS3Auth(self._conf),
+            headers=headers)
+        return None
+
+    def get_bucket_domain_certificate(self, Bucket, DomainName, **kwargs):
+        """获取bucket的自定义域名证书配置规则
+
+        :param Bucket(string): 存储桶名称.
+        :param DomainName(string): Bucket的自定义域名.
+        :param kwargs(dict): 设置请求headers.
+        :return(dict): Bucket的自定义域名证书配置规则.
+
+        .. code-block:: python
+            config = CosConfig(Region=region, SecretId=secret_id, SecretKey=secret_key, Token=token)  # 获取配置对象
+            client = CosS3Client(config)
+            # 获取bucket自定义域名证书配置规则
+            response = client.get_bucket_domain_certificate(
+                Bucket='bucket',
+                DomainName='domain-name'
+            )
+        """
+        headers = mapped(kwargs)
+        # 目前 Domain Certificate API 不能使用 params 传递 query_string '?domaincertificate=',
+        # 只能将'?domaincertificate'拼接到url
+        # params = {'domaincertificate': '', 'domainname': DomainName}
+        params = {'domainname': DomainName}
+        url = self._conf.uri(bucket=Bucket)
+        url += '?domaincertificate'
+        logger.info("get bucket domain certificate, url=:{url} ,headers=:{headers}".format(
+            url=url,
+            headers=headers))
+        rt = self.send_request(
+            method='GET',
+            url=url,
+            bucket=Bucket,
+            auth=CosS3Auth(self._conf, params=params),
+            headers=headers,
+            params=params)
+        data = xml_to_dict(rt.content)
+        return data
+
+    def delete_bucket_domain_certificate(self, Bucket, DomainName, **kwargs):
+        """删除bucket的自定义域名证书配置规则
+
+        :param Bucket(string): 存储桶名称.
+        :param DomainName(string): Bucket的自定义域名.
+        :param kwargs(dict): 设置请求headers.
+        :return: None
+
+        .. code-block:: python
+            config = CosConfig(Region=region, SecretId=secret_id, SecretKey=secret_key, Token=token)  # 获取配置对象
+            client = CosS3Client(config)
+            # 删除bucket自定义域名证书配置规则
+            response = client.delete_bucket_domain_certificate(
+                Bucket='bucket',
+                DomainName='domain-name'
+            )
+        """
+        headers = mapped(kwargs)
+        # 目前 Domain Certificate API 不能使用 params 传递 query_string '?domaincertificate=',
+        # 只能将'?domaincertificate'拼接到url
+        # params = {'domaincertificate': '', 'domainname': DomainName}
+        params = {'domainname': DomainName}
+        url = self._conf.uri(bucket=Bucket)
+        url += "?domaincertificate"
+        logger.info("delete bucket domain certificate, url=:{url} ,headers=:{headers}".format(
+            url=url,
+            headers=headers))
+        rt = self.send_request(
+            method='DELETE',
+            url=url,
+            bucket=Bucket,
+            auth=CosS3Auth(self._conf, params=params),
+            headers=headers,
+            params=params)
+        return None
+
     def put_bucket_origin(self, Bucket, OriginConfiguration={}, **kwargs):
         """设置bucket的回源规则
 

ut/test.py

@@ -406,6 +406,21 @@ def test_create_head_delete_bucket():
         Bucket=bucket_name
     )
 
+def test_create_head_delete_maz_bucket():
+    """创建一个多AZ bucket,head它是否存在,最后删除一个空bucket"""
+    bucket_id = str(random.randint(0, 1000)) + str(random.randint(0, 1000))
+    bucket_name = 'buckettest-maz' + bucket_id + '-' + APPID
+    response = client.create_bucket(
+        Bucket=bucket_name,
+        BucketAZConfig='MAZ',
+        ACL='public-read'
+    )
+    response = client.head_bucket(
+        Bucket=bucket_name
+    )
+    response = client.delete_bucket(
+        Bucket=bucket_name
+    )
 
 def test_put_bucket_acl_illegal():
     """设置非法的ACL"""
@@ -1034,6 +1049,91 @@ def test_put_get_delete_bucket_domain():
         Bucket=test_bucket
     )
 
+def test_put_get_delete_bucket_domain_certificate():
+    """测试设置获取删除bucket自定义域名证书"""
+
+    """
+    存储桶 bj-1259654469 专门用于测试自定义域名证书
+    """
+
+    temp_bucket = 'bj-1259654469'
+    temp_conf = CosConfig(
+        Region='ap-beijing',
+        SecretId=SECRET_ID,
+        SecretKey=SECRET_KEY
+    )
+    temp_client = CosS3Client(
+        conf=temp_conf,
+        retry=3
+    )
+
+    domain = 'testcertificate.coshelper.com'
+    domain_config = {
+        'DomainRule': [
+            {
+                'Name': domain,
+                'Type': 'REST',
+                'Status': 'ENABLED',
+            },
+        ]
+    }
+
+    # put domain
+    response = temp_client.put_bucket_domain(
+        Bucket=temp_bucket,
+        DomainConfiguration=domain_config
+    ) 
+
+    with open('./testcertificate.coshelper.com.key', 'rb') as f:
+        key = f.read().decode('utf-8')
+    with open('./testcertificate.coshelper.com.pem', 'rb') as f:
+        cert = f.read().decode('utf-8')
+
+    domain_cert_config = {
+        'CertificateInfo': {
+            'CertType': 'CustomCert',
+            'CustomCert': {
+                'Cert': cert,
+                'PrivateKey': key,
+            },
+        },
+        'DomainList': [
+            {
+                'DomainName': domain
+            },
+        ],
+    }
+
+    # put domain certificate
+    response = temp_client.delete_bucket_domain_certificate(
+        Bucket=temp_bucket,
+        DomainName=domain
+    )
+
+    time.sleep(2)
+    response = temp_client.put_bucket_domain_certificate(
+        Bucket=temp_bucket,
+        DomainCertificateConfiguration=domain_cert_config
+    )
+    # wait for sync
+    # get domain certificate
+    time.sleep(4)
+    response = temp_client.get_bucket_domain_certificate(
+        Bucket=temp_bucket,
+        DomainName=domain
+    )
+    assert response["Status"] == "Enabled"
+
+    # delete domain certificate
+    response = temp_client.delete_bucket_domain_certificate(
+        Bucket=temp_bucket,
+        DomainName=domain
+    )
+
+    # delete domain
+    response = temp_client.delete_bucket_domain(
+        Bucket=temp_bucket,
+    )
 
 def test_put_get_delete_bucket_inventory():
     """测试设置获取删除bucket清单"""