Add black referer (#334)

Author: zhuxiaolong37

oss2/models.py

@@ -733,16 +733,19 @@ class BucketReferer(object):
 
     :param bool allow_empty_referer: 是否允许空的Referer。
     :param referers: Referer列表，每个元素是一个str。
+    :param black_referers: BlackReferer列表，每个元素是一个str。
     """
-    def __init__(self, allow_empty_referer, referers):
+    def __init__(self, allow_empty_referer, referers, allow_truncate_query_string=None, black_referers=None):
         self.allow_empty_referer = allow_empty_referer
         self.referers = referers
+        self.allow_truncate_query_string = allow_truncate_query_string
+        self.black_referers = black_referers
 
 
 class GetBucketRefererResult(RequestResult, BucketReferer):
     def __init__(self, resp):
         RequestResult.__init__(self, resp)
-        BucketReferer.__init__(self, False, [])
+        BucketReferer.__init__(self, False, [], None, [])
 
 class Condition(object):
     """ 匹配规则

oss2/xml_utils.py

@@ -418,6 +418,11 @@ def parse_get_bucket_referer(result, body):
     result.allow_empty_referer = _find_bool(root, 'AllowEmptyReferer')
     result.referers = _find_all_tags(root, 'RefererList/Referer')
 
+    if root.find("AllowTruncateQueryString") is not None:
+        result.allow_truncate_query_string = _find_bool(root, 'AllowTruncateQueryString')
+    if root.find("RefererBlacklist/Referer") is not None:
+        result.black_referers = _find_all_tags(root, 'RefererBlacklist/Referer')
+
     return result
 
 def parse_condition_include_header(include_header_node):
@@ -869,6 +874,15 @@ def to_put_bucket_referer(bucket_referer):
     for r in bucket_referer.referers:
         _add_text_child(list_node, 'Referer', r)
 
+    if bucket_referer.allow_truncate_query_string is not None:
+        _add_text_child(root, 'AllowTruncateQueryString', str(bucket_referer.allow_truncate_query_string).lower())
+
+    if bucket_referer.black_referers:
+        black_referer_node = ElementTree.SubElement(root, 'RefererBlacklist')
+
+        for r in bucket_referer.black_referers:
+            _add_text_child(black_referer_node, 'Referer', r)
+
     return _node_to_string(root)
 
 

tests/test_bucket.py

@@ -667,6 +667,57 @@ def test_referer(self):
         self.assertTrue(result.allow_empty_referer)
         self.assertEqual(sorted(to_string(r) for r in referers), sorted(to_string(r) for r in result.referers))
 
+    def test_black_referer(self):
+        referers = ['http://hello.com', 'mibrowser:home', '中文+referer', u'中文+referer']
+        black_referers = ['http://hello2.com', 'mibrowser2:home', '中文2+referer', u'中文2+referer']
+        allow_empty_referer = True
+        allow_truncate_query_string = True
+
+        # black referer 1
+        config = oss2.models.BucketReferer(allow_empty_referer, referers, allow_truncate_query_string)
+        self.bucket.put_bucket_referer(config)
+        wait_meta_sync()
+
+        result = self.bucket.get_bucket_referer()
+        self.assertTrue(result.allow_empty_referer)
+        self.assertTrue(result.allow_truncate_query_string)
+        self.assertEqual(sorted(to_string(r) for r in referers), sorted(to_string(r) for r in result.referers))
+
+        # black referer 2
+        allow_empty_referer = False
+        allow_truncate_query_string = False
+        config = oss2.models.BucketReferer(allow_empty_referer, referers, allow_truncate_query_string, black_referers)
+        self.bucket.put_bucket_referer(config)
+        wait_meta_sync()
+
+        result = self.bucket.get_bucket_referer()
+        self.assertFalse(result.allow_empty_referer)
+        self.assertFalse(result.allow_truncate_query_string)
+        self.assertEqual(sorted(to_string(r) for r in referers), sorted(to_string(r) for r in result.referers))
+        self.assertEqual(sorted(to_string(r) for r in black_referers), sorted(to_string(r) for r in result.black_referers))
+
+        # black referer 3
+        config = oss2.models.BucketReferer(allow_empty_referer, referers, black_referers=None)
+        self.bucket.put_bucket_referer(config)
+        wait_meta_sync()
+
+        result = self.bucket.get_bucket_referer()
+        self.assertFalse(result.allow_empty_referer)
+        self.assertEqual(True, result.allow_truncate_query_string)
+        self.assertEqual(sorted(to_string(r) for r in referers), sorted(to_string(r) for r in result.referers))
+        self.assertEqual([], result.black_referers)
+
+        # black referer 4
+        config = oss2.models.BucketReferer(allow_empty_referer, referers, black_referers=[])
+        self.bucket.put_bucket_referer(config)
+        wait_meta_sync()
+
+        result = self.bucket.get_bucket_referer()
+        self.assertFalse(result.allow_empty_referer)
+        self.assertEqual(True, result.allow_truncate_query_string)
+        self.assertEqual(sorted(to_string(r) for r in referers), sorted(to_string(r) for r in result.referers))
+        self.assertEqual([], result.black_referers)
+
     def test_location(self):
         result = self.bucket.get_bucket_location()
         self.assertTrue(result.location)

unittests/test_bucket.py

@@ -2271,5 +2271,135 @@ def test_delete_bucket_style(self, do_request):
 
         self.assertRequest(req_info, request_text)
 
+
+    @patch('oss2.Session.do_request')
+    def test_put_black_referer(self, do_request):
+        from oss2.models import BucketReferer
+
+        request_text = '''PUT /?referer= 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=
+
+<RefererConfiguration>
+<AllowEmptyReferer>true</AllowEmptyReferer>
+<RefererList>
+<Referer>http://hello.com</Referer>
+<Referer>mibrowser:home</Referer>
+<Referer>{0}</Referer>
+</RefererList>
+<AllowTruncateQueryString>true</AllowTruncateQueryString>
+<RefererBlacklist>
+<Referer>http://hello2.com</Referer>
+<Referer>mibrowser2:home</Referer>
+<Referer>{1}</Referer>
+</RefererBlacklist>
+</RefererConfiguration>'''
+
+        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_referer(BucketReferer(True, ['http://hello.com', 'mibrowser:home', '阿里巴巴'], True, ['http://hello2.com', 'mibrowser2:home', '阿里巴巴2']))
+
+        self.assertRequest(req_info, request_text.format('阿里巴巴', '阿里巴巴2'))
+
+
+    @patch('oss2.Session.do_request')
+    def test_get_black_referer(self, do_request):
+        request_text = '''GET /?referer= 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"?>
+<RefererConfiguration>
+  <AllowEmptyReferer>true</AllowEmptyReferer>
+  <AllowTruncateQueryString>false</AllowTruncateQueryString>
+  <RefererList>
+    <Referer>http://hello.com</Referer>
+    <Referer>mibrowser:home</Referer>
+    <Referer>{0}</Referer>
+  </RefererList>
+  <RefererBlacklist>
+    <Referer>http://www.aliyun.com</Referer>
+    <Referer>mibrowser:home.com</Referer>
+  </RefererBlacklist>
+</RefererConfiguration>'''.format('阿里巴巴')
+
+        req_info = mock_response(do_request, response_text)
+
+        result = bucket().get_bucket_referer()
+
+        self.assertRequest(req_info, request_text)
+
+        self.assertEqual(result.allow_empty_referer, True)
+        self.assertEqual(result.allow_truncate_query_string, False)
+        self.assertSortedListEqual(result.referers, ['http://hello.com', 'mibrowser:home', '阿里巴巴'])
+        self.assertSortedListEqual(result.black_referers, ['http://www.aliyun.com', 'mibrowser:home.com'])
+
+
+    @patch('oss2.Session.do_request')
+    def test_get_black_referer2(self, do_request):
+        request_text = '''GET /?referer= 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"?>
+<RefererConfiguration>
+  <AllowEmptyReferer>true</AllowEmptyReferer>
+  <RefererList>
+    <Referer>http://hello.com</Referer>
+    <Referer>mibrowser:home</Referer>
+    <Referer>{0}</Referer>
+  </RefererList>
+
+</RefererConfiguration>'''.format('阿里巴巴')
+
+        req_info = mock_response(do_request, response_text)
+
+        result = bucket().get_bucket_referer()
+
+        self.assertRequest(req_info, request_text)
+        self.assertEqual(result.allow_empty_referer, True)
+        self.assertSortedListEqual(result.referers, ['http://hello.com', 'mibrowser:home', '阿里巴巴'])
+
+
+
 if __name__ == '__main__':
     unittest.main()