Merge pull request #228 from l-iberty/dev/aksk-security

AKSK密钥轮换安全方案适配

Author: l-iberty

qcloud_cos/cos_auth.py

@@ -45,8 +45,10 @@ def filter_headers(data):
 class CosS3Auth(AuthBase):
 
     def __init__(self, conf, key=None, params={}, expire=10000, sign_host=None):
-        self._secret_id = conf._secret_id
-        self._secret_key = conf._secret_key
+        self._secret_id = conf._secret_id if conf._secret_id else \
+            (conf._credential_inst.secret_id if conf._credential_inst else None)
+        self._secret_key = conf._secret_key if conf._secret_key else \
+            (conf._credential_inst.secret_key if conf._credential_inst else None)
         self._anonymous = conf._anonymous
         self._expire = expire
         self._params = params

qcloud_cos/cos_client.py

@@ -40,7 +40,7 @@
 class CosConfig(object):
     """config类，保存用户相关信息"""
 
-    def __init__(self, Appid=None, Region=None, SecretId=None, SecretKey=None, Token=None, Scheme=None, Timeout=None,
+    def __init__(self, Appid=None, Region=None, SecretId=None, SecretKey=None, Token=None, CredentialInstance=None, Scheme=None, Timeout=None,
                  Access_id=None, Access_key=None, Secret_id=None, Secret_key=None, Endpoint=None, IP=None, Port=None,
                  Anonymous=None, UA=None, Proxies=None, Domain=None, ServiceDomain=None, PoolConnections=10,
                  PoolMaxSize=10, AllowRedirects=False, SignHost=True, EndpointCi=None, EndpointPic=None, EnableOldDomain=True, EnableInternalDomain=True):
@@ -119,9 +119,14 @@ def __init__(self, Appid=None, Region=None, SecretId=None, SecretKey=None, Token
         elif (Access_id and Access_key):
             self._secret_id = self.convert_secret_value(Access_id)
             self._secret_key = self.convert_secret_value(Access_key)
+        elif (CredentialInstance and hasattr(CredentialInstance, "secret_id") and hasattr(CredentialInstance, "secret_key") and hasattr(CredentialInstance, "token")):
+            self._secret_id = None
+            self._secret_key = None
+            self._credential_inst = CredentialInstance
         elif self._anonymous:
             self._secret_id = None
             self._secret_key = None
+            self._credential_inst = None
         else:
             raise CosClientError('SecretId and SecretKey is Required!')
 

ut/test.py

@@ -23,16 +23,39 @@
 REGION = os.environ["REGION"]
 APPID = '1251668577'
 TEST_CI = os.environ["TEST_CI"]
+USE_CREDENTIAL_INST = os.environ["USE_CREDENTIAL_INST"]
 test_bucket = 'cos-python-v5-test-' + str(sys.version_info[0]) + '-' + str(
     sys.version_info[1]) + '-' + REGION + '-' + APPID
 copy_test_bucket = 'copy-' + test_bucket
 test_object = "test.txt"
 special_file_name = "中文" + "→↓←→↖↗↙↘! \"#$%&'()*+,-./0123456789:;<=>@ABCDEFGHIJKLMNOPQRSTUVWXYZ[\\]^_`abcdefghijklmnopqrstuvwxyz{|}~"
-conf = CosConfig(
-    Region=REGION,
-    SecretId=SECRET_ID,
-    SecretKey=SECRET_KEY,
-)
+
+""" CredentialDemo """
+class CredentialDemo:
+    @property
+    def secret_id(self):
+        return SECRET_ID
+    
+    @property
+    def secret_key(self):
+        return SECRET_KEY
+    
+    @property
+    def token(self):
+        return ''
+
+if USE_CREDENTIAL_INST == 'true':
+    conf = CosConfig(
+        Region=REGION,
+        CredentialInstance=CredentialDemo()
+    )
+else:
+    conf = CosConfig(
+        Region=REGION,
+        SecretId=SECRET_ID,
+        SecretKey=SECRET_KEY,
+    )
+
 client = CosS3Client(conf, retry=3)
 rsa_provider = RSAProvider()
 client_for_rsa = CosEncryptionClient(conf, rsa_provider)