support to verify object name strictly (#369)

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))

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()