diff --git a/oss2/api.py b/oss2/api.py index 2b85fb8a..6a6cea41 100644 --- a/oss2/api.py +++ b/oss2/api.py @@ -203,7 +203,7 @@ def progress_callback(bytes_consumed, total_bytes): class _Base(object): def __init__(self, auth, endpoint, is_cname, session, connect_timeout, - app_name='', enable_crc=True, proxies=None, region=None, cloudbox_id= None, is_path_style=False): + app_name='', enable_crc=True, proxies=None, region=None, cloudbox_id= None, is_path_style=False, is_verify_object_strict=True): self.auth = auth self.endpoint = _normalize_endpoint(endpoint.strip()) if utils.is_valid_endpoint(self.endpoint) is not True: @@ -219,6 +219,7 @@ def __init__(self, auth, endpoint, is_cname, session, connect_timeout, if self.cloudbox_id is not None: self.product = 'oss-cloudbox' self._make_url = _UrlMaker(self.endpoint, is_cname, is_path_style) + self.is_verify_object_strict = is_verify_object_strict def _do(self, method, bucket_name, key, **kwargs): @@ -418,6 +419,8 @@ class Bucket(_Base): :param str app_name: 应用名。该参数不为空,则在User Agent中加入其值。 注意到,最终这个字符串是要作为HTTP Header的值传输的,所以必须要遵循HTTP标准。 + + :param bool is_verify_object_strict: 严格验证对象名称的标志。默认为True。 """ ACL = 'acl' @@ -477,12 +480,13 @@ def __init__(self, auth, endpoint, bucket_name, proxies=None, region=None, cloudbox_id=None, - is_path_style=False): + is_path_style=False, + is_verify_object_strict=True): logger.debug("Init Bucket: {0}, endpoint: {1}, isCname: {2}, connect_timeout: {3}, app_name: {4}, enabled_crc: {5}, region: {6}" ", proxies: {6}".format(bucket_name, endpoint, is_cname, connect_timeout, app_name, enable_crc, proxies, region)) super(Bucket, self).__init__(auth, endpoint, is_cname, session, connect_timeout, app_name=app_name, enable_crc=enable_crc, proxies=proxies, - region=region, cloudbox_id=cloudbox_id, is_path_style=is_path_style) + region=region, cloudbox_id=cloudbox_id, is_path_style=is_path_style, is_verify_object_strict=is_verify_object_strict) self.bucket_name = bucket_name.strip() if utils.is_valid_bucket_name(self.bucket_name) is not True: @@ -515,6 +519,10 @@ def sign_url(self, method, key, expires, headers=None, params=None, slash_safe=F if key is None or len(key.strip()) <= 0: raise ClientError("The key is invalid, please check it.") key = to_string(key) + + if self.is_verify_object_strict and key.startswith('?'): + raise ClientError("The key cannot start with `?`, please check it.") + logger.debug( "Start to sign_url, method: {0}, bucket: {1}, key: {2}, expires: {3}, headers: {4}, params: {5}, slash_safe: {6}".format( method, self.bucket_name, to_string(key), expires, headers, params, slash_safe)) diff --git a/tests/test_sign.py b/tests/test_sign.py index 907aeda4..a84e1a9d 100644 --- a/tests/test_sign.py +++ b/tests/test_sign.py @@ -113,5 +113,57 @@ def test_sign_key_not_empty(self): bucket.sign_url('PUT', key, 1650801600, headers=headers) except oss2.exceptions.ClientError as e: self.assertEqual(e.body, 'ClientError: The key is invalid, please check it.') + + def test_sign_key_is_key_strictly(self): + auth = oss2.Auth(OSS_ID, OSS_SECRET) + bucket_name = self.OSS_BUCKET + "-sign-v1-is-key-strictly-default" + bucket = oss2.Bucket(auth, OSS_ENDPOINT, bucket_name) + bucket.create_bucket() + key = '123.txt' + headers = dict() + content = 'test example' + url = bucket.sign_url('PUT', key, 1650801600, headers=headers) + print(url) + + put_result = bucket.put_object(key, content) + self.assertEqual(200, put_result.status) + + get_result = bucket.get_object(key) + self.assertEqual(200, get_result.status) + + del_result = bucket.delete_object(key) + self.assertEqual(204, del_result.status) + + key = '?123.txt' + try: + bucket.sign_url('PUT', key, 1650801600, headers=headers) + except oss2.exceptions.ClientError as e: + self.assertEqual(e.body, 'ClientError: The key cannot start with `?`, please check it.') + + key = '?' + try: + bucket.sign_url('PUT', key, 1650801600, headers=headers) + except oss2.exceptions.ClientError as e: + self.assertEqual(e.body, 'ClientError: The key cannot start with `?`, please check it.') + + bucket.delete_bucket() + + bucket_name = self.OSS_BUCKET + "-sign-v1-is-key-strictly" + bucket2 = oss2.Bucket(auth, OSS_ENDPOINT, bucket_name, is_verify_object_strict=False) + bucket2.create_bucket() + key = '?123.txt' + url2 = bucket2.sign_url('PUT', key, 1650801600, headers=headers) + print(url2) + + put_result2 = bucket2.put_object(key, content) + self.assertEqual(200, put_result2.status) + + get_result2 = bucket2.get_object(key) + self.assertEqual(200, get_result2.status) + + del_result2 = bucket2.delete_object(key) + self.assertEqual(204, del_result2.status) + bucket2.delete_bucket() + if __name__ == '__main__': unittest.main()