diff --git a/qcloud_cos/cos_auth.py b/qcloud_cos/cos_auth.py index 4f4c051c..39bca9c2 100644 --- a/qcloud_cos/cos_auth.py +++ b/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: diff --git a/qcloud_cos/cos_client.py b/qcloud_cos/cos_client.py index 3273d0c9..27d82bf8 100644 --- a/qcloud_cos/cos_client.py +++ b/qcloud_cos/cos_client.py @@ -1273,10 +1273,11 @@ 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. @@ -1284,12 +1285,23 @@ def create_bucket(self, Bucket, **kwargs): 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 = ['', ''] # 类型为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的回源规则 diff --git a/ut/test.py b/ut/test.py index a6a1096e..ee5b433b 100644 --- a/ut/test.py +++ b/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清单"""