From 1f0fa65606c8279e9e68b6deaff34c76666504ad Mon Sep 17 00:00:00 2001 From: kelvinBen Date: Sun, 8 Aug 2021 18:51:07 +0800 Subject: [PATCH] =?UTF-8?q?-=20=E4=BC=98=E5=8C=96iOS=E5=A3=B3=E6=A3=80?= =?UTF-8?q?=E6=B5=8B=E9=80=9F=E5=BA=A6=20-=20=E4=BC=98=E5=8C=96iPA?= =?UTF-8?q?=E6=96=87=E4=BB=B6=E8=A7=A3=E5=8E=8B=E6=95=88=E7=8E=87=20-=20?= =?UTF-8?q?=E4=BC=98=E5=8C=96=E6=97=A5=E5=BF=97=E8=BE=93=E5=87=BA?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app.py | 6 ++++ libs/core/__init__.py | 7 +++-- libs/core/download.py | 7 ++--- libs/core/net.py | 3 +- libs/core/parses.py | 5 +-- libs/task/android_task.py | 5 +-- libs/task/base_task.py | 25 +++++++-------- libs/task/download_task.py | 4 +-- libs/task/ios_task.py | 63 ++++++++++---------------------------- libs/task/web_task.py | 3 +- update.md | 5 +++ 11 files changed, 60 insertions(+), 73 deletions(-) diff --git a/app.py b/app.py index 5bc7fec..a7b6a46 100644 --- a/app.py +++ b/app.py @@ -4,6 +4,7 @@ # Github: https://github.com/kelvinBen/AppInfoScanner import click +import logging from libs.core import Bootstrapper from libs.task.base_task import BaseTask @@ -66,6 +67,11 @@ def web(inputs: str, rules: str, sniffer: bool, no_resource:bool, all:bool, thre raise e def main(): + LOG_FORMAT = "%(levelname)s - %(message)s" # 日志格式化输出 + DATE_FORMAT = "%m/%d/%Y %H:%M:%S %p" # 日期格式 + fp = logging.FileHandler('info.log', encoding='utf-8') + fs = logging.StreamHandler() + logging.basicConfig(level=logging.INFO, format=LOG_FORMAT, datefmt=DATE_FORMAT, handlers=[fp, fs]) # 调用 cli() if __name__ == "__main__": diff --git a/libs/core/__init__.py b/libs/core/__init__.py index 231ee88..a5fc85c 100644 --- a/libs/core/__init__.py +++ b/libs/core/__init__.py @@ -6,6 +6,7 @@ import os import time import shutil import platform +import logging # smali 所在路径 smali_path = "" @@ -97,16 +98,16 @@ class Bootstrapper(object): self.__removed_dirs_cmd__(output_path) os.makedirs(output_path) - print("[*] Create directory {}".format(output_path)) + logging.info("[*] Create directory {}".format(output_path)) if not os.path.exists(download_path): # shutil.rmtree(download_path) os.makedirs(download_path) - print("[*] Create directory {}".format(download_path)) + logging.info("[*] Create directory {}".format(download_path)) if not os.path.exists(history_path): os.makedirs(history_path) - print("[*] Create directory {}".format(history_path)) + logging.info("[*] Create directory {}".format(history_path)) if os.path.exists(txt_result_path): os.remove(txt_result_path) diff --git a/libs/core/download.py b/libs/core/download.py index 9c61d1d..7691c63 100644 --- a/libs/core/download.py +++ b/libs/core/download.py @@ -7,6 +7,7 @@ import os import sys import time import config +import logging import requests import threading import libs.core as cores @@ -40,7 +41,6 @@ class DownloadThreads(threading.Thread): if self.types == "Android" or self.types == "iOS": count = 0 progress_tmp = 0 - time1 = time.time() length = float(resp.headers['content-length']) with open(self.cache_path, "wb") as f: for chunk in resp.iter_content(chunk_size = 512): @@ -50,8 +50,8 @@ class DownloadThreads(threading.Thread): progress = int(count / length * 100) if progress != progress_tmp: progress_tmp = progress - print("\r", end="") - print("[*] Download progress: {}%: ".format(progress), "▋" * (progress // 2), end="") + logging.info("\r", end="") + logging.info("[*] Download progress: {}%: ".format(progress), "▋" * (progress // 2), end="") sys.stdout.flush() f.close() else: @@ -62,7 +62,6 @@ class DownloadThreads(threading.Thread): cores.download_flag = True except Exception as e: raise Exception(e) - return def run(self): threadLock = threading.Lock() diff --git a/libs/core/net.py b/libs/core/net.py index 434515a..636e420 100644 --- a/libs/core/net.py +++ b/libs/core/net.py @@ -4,6 +4,7 @@ # Github: https://github.com/kelvinBen/AppInfoScanner import re import time +import logging import threading import requests import libs.core as cores @@ -27,7 +28,7 @@ class NetThreads(threading.Thread): url_ip = domains["url_ip"] time.sleep(2) result = self.__get_request_result__(url_ip) - print("[+] Processing URL address:"+url_ip) + logging.info("[+] Processing URL address:"+url_ip) if result != "error": if self.lock.acquire(True): cores.excel_row = cores.excel_row + 1 diff --git a/libs/core/parses.py b/libs/core/parses.py index 8ce73af..48b5ed6 100644 --- a/libs/core/parses.py +++ b/libs/core/parses.py @@ -6,6 +6,7 @@ import re import os import config +import logging import threading import libs.core as cores @@ -59,7 +60,7 @@ class ParsesThreads(threading.Thread): akAndSkList = re.compile(r'.*accessKeyId.*".*"|.*accessKeySecret.*".*"|.*secret.*".*"').findall(file_content) for akAndSk in akAndSkList: self.result_list.append(akAndSk.strip()) - print("[+] AK or SK in:",akAndSk.strip()) + logging.info("[+] AK or SK in:",akAndSk.strip()) # 遍历所有的字符串 for result in set(results): @@ -80,7 +81,7 @@ class ParsesThreads(threading.Thread): self.threadLock.acquire() if cores.all_flag: - print("[+] The string searched for matching rule is: %s" % (resl_str)) + logging.info("[+] The string searched for matching rule is: %s" % (resl_str)) self.result_list.append(resl_str) self.threadLock.release() continue diff --git a/libs/task/android_task.py b/libs/task/android_task.py index b4624d6..334efb0 100644 --- a/libs/task/android_task.py +++ b/libs/task/android_task.py @@ -5,6 +5,7 @@ import os import re import config +import logging import hashlib from queue import Queue import libs.core as cores @@ -81,7 +82,7 @@ class AndroidTask(object): self.__shell_test__(output_path) self.__scanner_file_by_apktool__(output_path) else: - print("[-] Decompilation failed, please submit error information at https://github.com/kelvinBen/AppInfoScanner/issues") + logging.info("[-] Decompilation failed, please submit error information at https://github.com/kelvinBen/AppInfoScanner/issues") raise Exception(file_path + ", Decompilation failed.") @@ -91,7 +92,7 @@ class AndroidTask(object): if os.system(cmd_str) == 0: self.__get_scanner_file__(output_path) else: - print("[-] Decompilation failed, please submit error information at https://github.com/kelvinBen/AppInfoScanner/issues") + logging.info("[-] Decompilation failed, please submit error information at https://github.com/kelvinBen/AppInfoScanner/issues") raise Exception(file_path + ", Decompilation failed.") diff --git a/libs/task/base_task.py b/libs/task/base_task.py index 008ec70..02e97de 100644 --- a/libs/task/base_task.py +++ b/libs/task/base_task.py @@ -5,6 +5,7 @@ import os import re import config +import logging import threading from queue import Queue import libs.core as cores @@ -35,12 +36,12 @@ class BaseTask(object): # 统一调度平台 def start(self): - print("[*] AI is analyzing filtering rules......") + logging.info("[*] AI is analyzing filtering rules......") # 获取历史记录 self.__history_handle__() - print("[*] The filtering rules obtained by AI are as follows: %s" % (set(config.filter_no)) ) + logging.info("[*] The filtering rules obtained by AI are as follows: %s" % (set(config.filter_no)) ) # 任务控制中心 task_info = self.__tast_control__() @@ -54,11 +55,11 @@ class BaseTask(object): file_identifier = task_info["file_identifier"] if shell_flag: - print('[-] \033[3;31m Error: This application has shell, the retrieval results may not be accurate, Please remove the shell and try again!') + logging.info('[-] \033[3;31m Error: This application has shell, the retrieval results may not be accurate, Please remove the shell and try again!') return # 线程控制中心 - print("[*] ========= Searching for strings that match the rules ===============") + logging.info("[*] ========= Searching for strings that match the rules ===============") self.__threads_control__(file_queue) # 等待线程结束 @@ -77,7 +78,7 @@ class BaseTask(object): types = cache_info["type"] if (not os.path.exists(cacar_path) and cores.download_flag): - print("[-] File download failed! Please download the file manually and try again.") + logging.info("[-] File download failed! Please download the file manually and try again.") return task_info # 调用Android 相关处理逻辑 @@ -104,17 +105,17 @@ class BaseTask(object): all_flag = cores.all_flag if self.sniffer: - print("[*] ========= Sniffing the URL address of the search ===============") + logging.info("[*] ========= Sniffing the URL address of the search ===============") NetTask(self.result_dict,self.app_history_list,self.domain_history_list,file_identifier,self.threads).start() if packagename: - print("[*] ========= The package name of this APP is: ===============") - print(packagename) + logging.info("[*] ========= The package name of this APP is: ===============") + logging.info(packagename) if len(comp_list) != 0: - print("[*] ========= Component information is as follows :===============") + logging.info("[*] ========= Component information is as follows :===============") for json in comp_list: - print(json) + logging.info(json) if all_flag: value_list = [] @@ -127,10 +128,10 @@ class BaseTask(object): value_list.append(result) f.write("\t"+result+"\r") f.close() - print("[*] For more information about the search, see TXT file result: %s" %(txt_result_path)) + logging.info("[*] For more information about the search, see TXT file result: %s" %(txt_result_path)) if self.sniffer: - print("[*] For more information about the search, see XLS file result: %s" %(xls_result_path)) + logging.info("[*] For more information about the search, see XLS file result: %s" %(xls_result_path)) def __history_handle__(self): domain_history_path = cores.domain_history_path diff --git a/libs/task/download_task.py b/libs/task/download_task.py index f388a05..7c311a6 100644 --- a/libs/task/download_task.py +++ b/libs/task/download_task.py @@ -37,10 +37,10 @@ class DownloadTask(object): else: # 目录处理 return {"path":path,"type":types} else: - print("[*] Detected that the task is not local, preparing to download file......") + logging.info("[*] Detected that the task is not local, preparing to download file......") cache_path = os.path.join(cores.download_path, file_name) thread = DownloadThreads(path,file_name,cache_path,types) thread.start() thread.join() - print() + logging.info() return {"path":cache_path,"type":types} \ No newline at end of file diff --git a/libs/task/ios_task.py b/libs/task/ios_task.py index 0262239..7a640ff 100644 --- a/libs/task/ios_task.py +++ b/libs/task/ios_task.py @@ -31,32 +31,24 @@ class iOSTask(object): return {"shell_flag":self.shell_flag,"file_queue":self.file_queue,"comp_list":[],"packagename":None,"file_identifier":self.file_identifier} def __get_file_header__(self,file_path): - hex_hand = 0x0 + crypt_load_command_hex = "2C000000" macho_name = os.path.split(file_path)[-1] self.file_identifier.append(macho_name) with open(file_path,"rb") as macho_file: - macho_file.seek(hex_hand,0) + macho_file.seek(0x0,0) magic = binascii.hexlify(macho_file.read(4)).decode().upper() macho_magics = ["CFFAEDFE","CEFAEDFE","BEBAFECA","CAFEBABE"] if magic in macho_magics: - self.__shell_test__(macho_file,hex_hand) + hex_str = binascii.hexlify(macho_file.read()).decode().upper() + if crypt_load_command_hex in hex_str: + macho_file.seek(int(hex_str.index("2C000000")/2)+20,0) + cryptid = binascii.hexlify(macho_file.read(4)).decode() + if cryptid == "01000000": + self.shell_flag = True macho_file.close() return True macho_file.close() return False - - - def __shell_test__(self,macho_file,hex_hand): - while True: - magic = binascii.hexlify(macho_file.read(4)).decode().upper() - if magic == "2C000000": - macho_file.seek(hex_hand,0) - encryption_info_command = binascii.hexlify(macho_file.read(24)).decode() - cryptid = encryption_info_command[-8:len(encryption_info_command)] - if cryptid == "01000000": - self.shell_flag = True - break - hex_hand = hex_hand + 4 def __scanner_file_by_ipa__(self,output): scanner_file_suffix = ["plist","js","xml","html"] @@ -86,39 +78,18 @@ class iOSTask(object): def __decode_ipa__(self,output_path): with zipfile.ZipFile(self.path,"r") as zip_files: zip_file_names = zip_files.namelist() - zip_files.extract(zip_file_names[0],output_path) - try: - new_zip_file = zip_file_names[0].encode('cp437').decode('utf-8') - except UnicodeEncodeError: - new_zip_file = zip_file_names[0].encode('utf-8').decode('utf-8') - - old_zip_dir = self.__get_parse_dir__(output_path,zip_file_names[0]) - new_zip_dir = self.__get_parse_dir__(output_path,new_zip_file) - os.rename(old_zip_dir,new_zip_dir) - - for zip_file in zip_file_names: - old_ext_path = zip_files.extract(zip_file,output_path) - start = str(old_ext_path).index("Payload") - dir_path = old_ext_path[start:len(old_ext_path)] - old_ext_path = os.path.join(output_path,dir_path) + for zip_file_name in zip_file_names: try: - new_zip_file = zip_file.encode('cp437').decode('utf-8') + if platform.system() == "Windows": + new_file_name = zip_file_name.encode('cp437').decode('GBK') + else: + new_file_name = zip_file_name.encode('cp437').decode('utf-8') except UnicodeEncodeError: - new_zip_file = zip_file.encode('utf-8').decode('utf-8') - - new_ext_path = os.path.join(output_path,new_zip_file) - - if platform.system() == "Windows": - new_ext_path = new_ext_path.replace("/","\\") - - if not os.path.exists(new_ext_path): - dir_path = os.path.dirname(new_ext_path) - if not os.path.exists(dir_path): - os.makedirs(dir_path) - shutil.move(old_ext_path, new_ext_path) - if os.path.exists(old_ext_path): - os.remove(old_ext_path) + new_file_name = zip_file_name.encode('utf-8').decode('utf-8') + new_ext_file_path = os.path.join(output_path,new_file_name) + ext_file_path = zip_files.extract(zip_file_name,output_path) + os.rename(ext_file_path,new_ext_file_path) def __get_parse_dir__(self,output_path,file_path): start = file_path.index("Payload/") diff --git a/libs/task/web_task.py b/libs/task/web_task.py index 4bd8018..b028cb6 100644 --- a/libs/task/web_task.py +++ b/libs/task/web_task.py @@ -4,6 +4,7 @@ # Github: https://github.com/kelvinBen/AppInfoScanner import os import config +import hashlib from queue import Queue class WebTask(object): @@ -39,7 +40,7 @@ class WebTask(object): else: if len(dir_file.split("."))>1: if dir_file.split(".")[-1] in file_suffix: - with open(file_path,'rb') as f: + with open(dir_file_path,'rb') as f: dex_md5 = str(hashlib.md5().update(f.read()).hexdigest()).upper() self.file_identifier.append(dex_md5) self.file_queue.put(dir_file_path) \ No newline at end of file diff --git a/update.md b/update.md index b756e7d..b0de0e7 100644 --- a/update.md +++ b/update.md @@ -1,3 +1,8 @@ +### V1.0.9 +- 优化iOS壳检测速度 +- 优化iPA文件效率 +- 优化日志输出 + ### V1.0.8 - 添加AK和SK的检测 - 添加检测规则提交入口