Sourcery Starbot ⭐ refactored Explorare/Cloud-Bucket-Leak-Detection-Tools

core/aliyunOss.py

@@ -101,8 +101,7 @@ def Aliyun_Oss_GetBucketPolicy(self):
         """
         try:
             result = self.bucket.get_bucket_policy()
-            policy_json = json.loads(result.policy)
-            return policy_json
+            return json.loads(result.policy)
         except oss2.exceptions.AccessDenied:
             return False
         except oss2.exceptions.NoSuchBucketPolicy:

core/main.py

@@ -36,14 +36,14 @@ def aliyun(target):
     aliyun_print_table_header = pt.PrettyTable(
         ['Bucket', 'BucketHijack', 'GetBucketObjectList', 'PutBucketObject', 'GetBucketAcl', 'PutBucketAcl',
          'GetBucketPolicy'])
-    aliyun_scan_results = {}
     get_domain = urllib.parse.urlparse(target).netloc
     if get_domain == "":
         get_target_list = target.split('.')
         aliyunOss_Check_init = aliyunOss.Aliyun_Oss_Bucket_Check(target=get_target_list[0],
                                                                  location=get_target_list[1])
         aliyunOss_Exploit_init = aliyunOss.Aliyun_Oss_Bucket_Exploit(target=get_target_list[0],
                                                                      location=get_target_list[1])
+        aliyun_scan_results = {}
         if aliyunOss_Check_init.Aliyun_Oss_BucketDoesBucketExist():
             logger.log("INFOR", f"{target}> 当前存储桶不存在, 尝试劫持存储桶")
             if aliyunOss_Exploit_init.Aliyun_Oss_CreateBucket_Exp():
@@ -53,45 +53,46 @@ def aliyun(target):
                 aliyunOss_Exploit_init.Aliyun_Oss_PutBucketPolicy_Exp()
                 aliyunOss_Exploit_init.Aliyun_Oss_GetBucketPolicy_Exp()
                 aliyunOss_Exploit_init.Aliyun_Oss_PutBucketAcl_Exp()
-                aliyun_scan_results.update({"BucketDoesBucketExist": "true"})
+                aliyun_scan_results["BucketDoesBucketExist"] = "true"
         else:
-            aliyun_scan_results.update({"BucketDoesBucketExist": "false"})
+            aliyun_scan_results["BucketDoesBucketExist"] = "false"
             if aliyunOss_Check_init.Aliyun_Oss_GetBucketObject_List():
                 logger.log("INFOR", f"{target}> 存储桶对象可遍历")
-                aliyun_scan_results.update({"GetBucketObject": "true"})
+                aliyun_scan_results["GetBucketObject"] = "true"
             else:
                 logger.log("ALERT", f"{target}> 存储桶对象不可遍历")
-                aliyun_scan_results.update({"GetBucketObject": "false"})
+                aliyun_scan_results["GetBucketObject"] = "false"
 
             if aliyunOss_Check_init.Aliyun_Oss_PutBucketObject():
                 logger.log("INFOR", f"{target}> 可未授权上传对象至存储桶（可导致覆盖已有对象）")
-                aliyun_scan_results.update({"PutBucketObject": "true"})
+                aliyun_scan_results["PutBucketObject"] = "true"
             else:
                 logger.log("ALERT", f"{target}> 不可未授权上传对象至存储桶")
-                aliyun_scan_results.update({"PutBucketObject": "false"})
+                aliyun_scan_results["PutBucketObject"] = "false"
 
             if aliyunOss_Check_init.Aliyun_Oss_GetBucketAcl():
                 logger.log("INFOR", f"{target}> 可公开访问存储桶ACL策略")
-                aliyun_scan_results.update({"GetBucketAcl": "true"})
+                aliyun_scan_results["GetBucketAcl"] = "true"
             else:
                 logger.log("ALERT", f"{target}> 不可公开访问存储桶ACL策略")
-                aliyun_scan_results.update({"GetBucketAcl": "false"})
+                aliyun_scan_results["GetBucketAcl"] = "false"
 
             if aliyunOss_Check_init.Aliyun_Oss_PutBucketAcl():
                 logger.log("INFOR", f"{target}> 可上传覆盖存储桶ACL策略")
-                aliyun_scan_results.update({"PutBucketAcl": "true"})
+                aliyun_scan_results["PutBucketAcl"] = "true"
             else:
                 logger.log("ALERT", f"{target}> 不可上传覆盖存储桶ACL策略")
-                aliyun_scan_results.update({"PutBucketAcl": "false"})
+                aliyun_scan_results["PutBucketAcl"] = "false"
 
-            results_policy = aliyunOss_Check_init.Aliyun_Oss_GetBucketPolicy()
-            if results_policy:
+            if (
+                results_policy := aliyunOss_Check_init.Aliyun_Oss_GetBucketPolicy()
+            ):
                 logger.log("INFOR", f"{target}> 可公开获取存储桶Policy策略组")
                 logger.log("INFOR", f"{target}Policy> {results_policy}")
-                aliyun_scan_results.update({"GetBucketPolicy": "true"})
+                aliyun_scan_results["GetBucketPolicy"] = "true"
             else:
                 logger.log("ALERT", f"{target}> 不可公开获取存储桶Policy策略")
-                aliyun_scan_results.update({"GetBucketPolicy": "false"})
+                aliyun_scan_results["GetBucketPolicy"] = "false"
 
             aliyun_print_table_header.add_row([target,
                                                aliyun_scan_results['BucketDoesBucketExist'],

plugins/results.py

@@ -13,12 +13,12 @@
 
 
 def aliyun_save_file(target, BucketHijack, GetBucketObjectList, PutBucketObject, GetBucketAcl, PutBucketAcl, GetBucketPolicy):
-    headers = ['Bucket', 'BucketHijack', 'GetBucketObjectList', 'PutBucketObject', 'GetBucketAcl', 'PutBucketAcl', 'GetBucketPolicy']
     filepath = f'{os.getcwd()}/results/{NowTime}.csv'
     rows = [
         [f"{target}", BucketHijack, GetBucketObjectList, PutBucketObject, GetBucketAcl, PutBucketAcl, GetBucketPolicy]
     ]
     if not os.path.isfile(filepath):
+        headers = ['Bucket', 'BucketHijack', 'GetBucketObjectList', 'PutBucketObject', 'GetBucketAcl', 'PutBucketAcl', 'GetBucketPolicy']
         with open(filepath, 'a+', newline='') as f:
             f = csv.writer(f)
             f.writerow(headers)