Merge pull request #417 from qiniu/features/add-x-qiniu-date

add X-Qiniu-Date

Author: Mei-Zhao

.github/workflows/ci-test.yml

@@ -21,7 +21,7 @@ jobs:
       - name: Install dependencies
         run: |
           python -m pip install --upgrade pip
-          pip install flake8 pytest pytest-cov requests scrutinizer-ocular codecov
+          pip install flake8 pytest pytest-cov freezegun requests scrutinizer-ocular codecov
       - name: Run cases
         env:
           QINIU_ACCESS_KEY: ${{ secrets.QINIU_ACCESS_KEY }}

qiniu/auth.py

@@ -49,11 +49,12 @@ class Auth(object):
         __secret_key: 账号密钥对重的secretKey，详见 https://portal.qiniu.com/user/key
     """
 
-    def __init__(self, access_key, secret_key):
+    def __init__(self, access_key, secret_key, disable_qiniu_timestamp_signature=None):
         """初始化Auth类"""
         self.__checkKey(access_key, secret_key)
         self.__access_key = access_key
         self.__secret_key = b(secret_key)
+        self.disable_qiniu_timestamp_signature = disable_qiniu_timestamp_signature
 
     def get_access_key(self):
         return self.__access_key
@@ -229,11 +230,12 @@ class QiniuMacAuth(object):
     http://kirk-docs.qiniu.com/apidocs/#TOC_325b437b89e8465e62e958cccc25c63f
     """
 
-    def __init__(self, access_key, secret_key):
+    def __init__(self, access_key, secret_key, disable_qiniu_timestamp_signature=None):
         self.qiniu_header_prefix = "X-Qiniu-"
         self.__checkKey(access_key, secret_key)
         self.__access_key = access_key
         self.__secret_key = b(secret_key)
+        self.disable_qiniu_timestamp_signature = disable_qiniu_timestamp_signature
 
     def __token(self, data):
         data = b(data)

qiniu/http.py

@@ -1,6 +1,8 @@
 # -*- coding: utf-8 -*-
 import logging
+import os
 import platform
+from datetime import datetime
 
 import requests
 from requests.auth import AuthBase
@@ -20,6 +22,19 @@
 _headers = {'User-Agent': USER_AGENT}
 
 
+def __add_auth_headers(headers, auth):
+    x_qiniu_date = datetime.utcnow().strftime('%Y%m%dT%H%M%SZ')
+    if auth.disable_qiniu_timestamp_signature is not None:
+        if not auth.disable_qiniu_timestamp_signature:
+            headers['X-Qiniu-Date'] = x_qiniu_date
+    elif os.getenv('DISABLE_QINIU_TIMESTAMP_SIGNATURE'):
+        if os.getenv('DISABLE_QINIU_TIMESTAMP_SIGNATURE').lower() != 'true':
+            headers['X-Qiniu-Date'] = x_qiniu_date
+    else:
+        headers['X-Qiniu-Date'] = x_qiniu_date
+    return headers
+
+
 def __return_wrapper(resp):
     if resp.status_code != 200 or resp.headers.get('X-Reqid') is None:
         return None, ResponseInfo(resp)
@@ -155,14 +170,18 @@ def _post_with_qiniu_mac(url, data, auth):
     qn_auth = qiniu.auth.QiniuMacRequestsAuth(
         auth
     ) if auth is not None else None
-    return _post(url, data, None, qn_auth)
+    headers = __add_auth_headers({}, auth)
+
+    return _post(url, data, None, qn_auth, headers=headers)
 
 
 def _get_with_qiniu_mac(url, params, auth):
     qn_auth = qiniu.auth.QiniuMacRequestsAuth(
         auth
     ) if auth is not None else None
-    return _get(url, params, qn_auth)
+    headers = __add_auth_headers({}, auth)
+
+    return _get(url, params, qn_auth, headers=headers)
 
 
 def _get_with_qiniu_mac_and_headers(url, params, auth, headers):

qiniu/services/storage/bucket.py

@@ -17,7 +17,10 @@ class BucketManager(object):
 
     def __init__(self, auth, zone=None):
         self.auth = auth
-        self.mac_auth = QiniuMacAuth(auth.get_access_key(), auth.get_secret_key())
+        self.mac_auth = QiniuMacAuth(
+            auth.get_access_key(),
+            auth.get_secret_key(),
+            auth.disable_qiniu_timestamp_signature)
         if (zone is None):
             self.zone = config.get_default('default_zone')
         else:

test_qiniu.py

@@ -10,6 +10,7 @@
 
 import unittest
 import pytest
+from freezegun import freeze_time
 
 from qiniu import Auth, set_default, etag, PersistentFop, build_op, op_save, Zone, QiniuMacAuth
 from qiniu import put_data, put_file, put_stream
@@ -494,6 +495,36 @@ def test_set_object_lifecycle_with_cond(self):
         )
         assert info.status_code == 200
 
+    @freeze_time("1970-01-01")
+    def test_invalid_x_qiniu_date(self):
+        ret, info = self.bucket.stat(bucket_name, 'python-sdk.html')
+        assert ret is None
+        assert info.status_code == 403
+
+    @freeze_time("1970-01-01")
+    def test_invalid_x_qiniu_date_with_disable_date_sign(self):
+        q = Auth(access_key, secret_key, disable_qiniu_timestamp_signature=True)
+        bucket = BucketManager(q)
+        ret, info = bucket.stat(bucket_name, 'python-sdk.html')
+        assert 'hash' in ret
+
+    @freeze_time("1970-01-01")
+    def test_invalid_x_qiniu_date_env(self):
+        os.environ['DISABLE_QINIU_TIMESTAMP_SIGNATURE'] = 'True'
+        ret, info = self.bucket.stat(bucket_name, 'python-sdk.html')
+        os.unsetenv('DISABLE_QINIU_TIMESTAMP_SIGNATURE')
+        assert 'hash' in ret
+
+    @freeze_time("1970-01-01")
+    def test_invalid_x_qiniu_date_env_be_ignored(self):
+        os.environ['DISABLE_QINIU_TIMESTAMP_SIGNATURE'] = 'True'
+        q = Auth(access_key, secret_key, disable_qiniu_timestamp_signature=False)
+        bucket = BucketManager(q)
+        ret, info = bucket.stat(bucket_name, 'python-sdk.html')
+        os.unsetenv('DISABLE_QINIU_TIMESTAMP_SIGNATURE')
+        assert ret is None
+        assert info.status_code == 403
+
 
 class UploaderTestCase(unittest.TestCase):
     mime_type = "text/plain"