From acc01c3303d9352459ae4f292de90117ea310ccb Mon Sep 17 00:00:00 2001 From: kelvinBen Date: Mon, 9 Aug 2021 19:51:45 +0800 Subject: [PATCH] =?UTF-8?q?-=20=E4=BC=98=E5=8C=96excle=E6=96=87=E4=BB=B6?= =?UTF-8?q?=E8=BE=93=E5=87=BA=20-=20=E6=9B=B4=E6=96=B0=E9=A1=B9=E7=9B=AE?= =?UTF-8?q?=E5=A4=84=E7=90=86=E6=B5=81=E7=A8=8B=20-=20=E6=B7=BB=E5=8A=A0?= =?UTF-8?q?=E6=89=B9=E9=87=8F=E5=A4=84=E7=90=86=E6=93=8D=E4=BD=9C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .gitignore | 4 + app.py | 47 ++++--- libs/core/__init__.py | 242 +++++++++++++++++++++++-------------- libs/core/net.py | 22 ++-- libs/core/parses.py | 2 +- libs/task/android_task.py | 20 +-- libs/task/base_task.py | 157 ++++++++++++++---------- libs/task/download_task.py | 74 ++++++++---- libs/task/ios_task.py | 14 +-- libs/task/net_task.py | 34 ++---- requirements.txt | 4 +- update.md | 11 +- 12 files changed, 379 insertions(+), 252 deletions(-) diff --git a/.gitignore b/.gitignore index b23dff5..526c5fc 100644 --- a/.gitignore +++ b/.gitignore @@ -114,3 +114,7 @@ venv.bak/ # add .idea/ + +# +1.py +.vscode/ \ No newline at end of file diff --git a/app.py b/app.py index a7b6a46..cb02e10 100644 --- a/app.py +++ b/app.py @@ -11,7 +11,13 @@ from libs.task.base_task import BaseTask @click.group(help="Python script for automatically retrieving key information in app.") def cli(): - pass + try: + LOG_FORMAT = "%(message)s" # 日志格式化输出 + fp = logging.FileHandler('info.log', mode='w',encoding='utf-8') + fs = logging.StreamHandler() + logging.basicConfig(level=logging.INFO, format=LOG_FORMAT, handlers=[fp, fs]) # 调用 + except Exception as e: + logging.error("{}".format(e)) # 创建Android任务 @cli.command(help="Get the key information of Android system.") @@ -24,13 +30,13 @@ def cli(): @click.option("-o", '--output',required=False, type=str,default=None,help="Specify the result set output directory.") @click.option("-p", '--package',required=False,type=str,default="",help="Specifies the package name information that needs to be scanned.") def android(inputs: str, rules: str, sniffer: bool, no_resource:bool, all:bool, threads:int, output, package:str) -> None: - try: - bootstrapper = Bootstrapper(__file__, output, all, no_resource) - bootstrapper.init() + + + bootstrapper = Bootstrapper(rules, sniffer, threads, all ,no_resource) + bootstrapper.init_dir(__file__, output) + + BaseTask().start("Android", inputs, package) - BaseTask("Android", inputs, rules, sniffer, threads, package).start() - except Exception as e: - raise e @cli.command(help="Get the key information of iOS system.") @click.option("-i", "--inputs", required=True, type=str, help="Please enter IPA file or ELF file to scan or corresponding IPA download address. App store is not supported at present.") @@ -41,13 +47,12 @@ def android(inputs: str, rules: str, sniffer: bool, no_resource:bool, all:bool, @click.option("-t", '--threads',required=False, type=int,default=10,help="Set the number of concurrency. The larger the concurrency, the faster the speed. The default value is 10.") @click.option("-o", '--output',required=False, type=str,default=None,help="Specify the result set output directory.") def ios(inputs: str, rules: str, sniffer: bool, no_resource:bool, all:bool, threads:int, output:str) -> None: - try: - bootstrapper = Bootstrapper(__file__, output, all, no_resource) - bootstrapper.init() + - BaseTask("iOS", inputs, rules, sniffer, threads).start() - except Exception as e: - raise e + + bootstrapper = Bootstrapper(rules, sniffer, threads, all ,no_resource) + bootstrapper.init_dir(__file__, output) + BaseTask().start("iOS", inputs) @cli.command(help="Get the key information of Web system.") @click.option("-i", "--inputs", required=True, type=str, help="Please enter the site directory or site file to scan or the corresponding site download address.") @@ -58,20 +63,14 @@ def ios(inputs: str, rules: str, sniffer: bool, no_resource:bool, all:bool, thre @click.option("-t", '--threads',required=False, type=int,default=10,help="Set the number of concurrency. The larger the concurrency, the faster the speed. The default value is 10.") @click.option("-o", '--output',required=False, type=str,default=None,help="Specify the result set output directory.") def web(inputs: str, rules: str, sniffer: bool, no_resource:bool, all:bool, threads:int, output:str) -> None: - try: - bootstrapper = Bootstrapper(__file__, output, all, no_resource) - bootstrapper.init() + - BaseTask("Web", inputs, rules, sniffer, threads).start() - except Exception as e: - raise e + bootstrapper = Bootstrapper(rules, sniffer, threads, all ,no_resource) + bootstrapper.init_dir(__file__, output) + BaseTask().start("Web", inputs) 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 a5fc85c..733bccb 100644 --- a/libs/core/__init__.py +++ b/libs/core/__init__.py @@ -2,119 +2,179 @@ # -*- coding: utf-8 -*- # Author: kelvinBen # Github: https://github.com/kelvinBen/AppInfoScanner + import os -import time import shutil import platform import logging -# smali 所在路径 -smali_path = "" - -# backsmli 所在路径 -backsmali_path = "" - -# apktool 所在路径 -apktool_path = "" +backsmali_file = None +apktool_file = None +strings_file = None +app_history_file= None +domain_history_file = None +result_dir = None +download_dir = None +decode_dir = None +user_add_rules = None -# 系统类型 -os_type = "" - -# 输出路径 -output_path = "" - -# 下载完成标记 download_flag = False +net_sniffer_flag = False +all_string_out = False +no_resource_flag = False + +excel_row = 1 +threads_num = 10 -# excel 起始行号 -excel_row = 0 class Bootstrapper(object): - def __init__(self, path, out_path, all=False, no_resource= False): - global smali_path - global backsmali_path - global apktool_path - global os_type - global output_path - global script_root_dir - global txt_result_path - global xls_result_path - global strings_path - global history_path - global app_history_path - global domain_history_path - global excel_row - global download_path + def __init__(self, rules, threads, sniffer, all, no_resource): + # backsmali 加载路径 + global backsmali_file + # apktool 加载路径 + global apktool_file + # string 加载路径 + global strings_file + # App 扫描历史文件 + global app_history_file + # 域名 扫描历史文件 + global domain_history_file + # 结果输出目录 + global result_dir + # 临时文件下载目录 + global download_dir + # 临时反编译目录 + global decode_dir + # 下载完成标记 global download_flag - global out_dir - global all_flag - global resource_flag - - all_flag = not all - resource_flag = no_resource + # excel 行号 + global excel_row + # 用户自定义规则 + global user_add_rules + # 用户指定线程数 + global threads_num + # 网络嗅探标记 + global net_sniffer_flag + # 输出所有字符传 + global all_string_out + # 忽略资源标记 + global no_resource_flag + + user_add_rules = rules + threads_num = threads + net_sniffer_flag = sniffer + all_string_out = all + no_resource_flag = no_resource + + # 需要创建的目录列表 + self.__create_dir_list__= [] + # 需要删除目录的列表 + self.__remove_dir_list__= [] + + logging.info("[*] System env: {}".format(platform.system())) + + def init_dir(self, app_input_path, user_out_path): + logging.info("[*] init dir...") + + # 脚本执行目录 + script_root_dir = os.path.dirname(os.path.abspath(app_input_path)) - create_time = time.strftime("%Y%m%d%H%M%S", time.localtime()) - script_root_dir = os.path.dirname(os.path.abspath(path)) - if out_path: - out_dir = out_path - else: - out_dir = script_root_dir + # 加载集成的工具 + self.__tools_loading__(script_root_dir) + + # 构建持久化目录 + self.__build_persistent_path__(script_root_dir) + + # 构建结果输出目录 + self.__build_result_out__path__(script_root_dir,user_out_path) + + # 统一目录构建中心 + self.__build_dir__() + + # 加载集成的工具 + def __tools_loading__(self,script_root_dir): tools_dir = os.path.join(script_root_dir,"tools") - output_path = os.path.join(out_dir,"out") - history_path = os.path.join(script_root_dir,"history") + + backsmali_file = os.path.join(tools_dir,"baksmali.jar") + logging.info("[*] Backsmali Path: {}".format(backsmali_file)) + + apktool_file = os.path.join(tools_dir, "apktool.jar") + logging.info("[*] Apktool Path: {}".format(apktool_file)) if platform.system() == "Windows": machine2bits = {'AMD64':64, 'x86_64': 64, 'i386': 32, 'x86': 32} machine2bits.get(platform.machine()) if platform.machine() == 'i386' or platform.machine() == 'x86': - strings_path = os.path.join(tools_dir,"strings.exe") + strings_file = os.path.join(tools_dir,"strings.exe") else: - strings_path = os.path.join(tools_dir,"strings64.exe") + strings_file = os.path.join(tools_dir,"strings64.exe") + else: + strings_file ="strings" + + logging.info("[*] Strings Path: {}".format(strings_file)) + + # 构建持久化目录 + def __build_persistent_path__(self,script_root_dir): + # 当前用户文档目录 + doc_path = os.path.join(os.path.expanduser("~"), 'Documents') + if os.path.exists(doc_path): + app_dir = os.path.join(doc_path,"AppInfoScanner") else: - strings_path ="strings" + app_dir = os.path.join(script_root_dir,"AppInfoScanner") + + # 历史任务加载目录 + history_path = os.path.join(app_dir,"history") + + app_history_file = os.path.join(history_path,"app_history.txt") + domain_history_file = os.path.join(history_path,"domain_history.txt") - backsmali_path = os.path.join(tools_dir,"baksmali.jar") - apktool_path = os.path.join(tools_dir, "apktool.jar") - download_path = os.path.join(out_dir,"download") - txt_result_path = os.path.join(out_dir,"result_"+str(create_time)+".txt") - xls_result_path = os.path.join(out_dir,"result_"+str(create_time)+".xls") - app_history_path = os.path.join(history_path,"app_history.txt") - domain_history_path = os.path.join(history_path,"domain_history.txt") - - def init(self): - if not os.path.exists(out_dir): - os.makedirs(out_dir) - print("[*] Create directory {}".format(out_dir)) - - if os.path.exists(output_path): - try: - shutil.rmtree(output_path) - except Exception as e: - # 解决windows超长文件名删除问题 - if not (platform.system() == "Windows"): - raise e - self.__removed_dirs_cmd__(output_path) - - os.makedirs(output_path) - logging.info("[*] Create directory {}".format(output_path)) - - if not os.path.exists(download_path): - # shutil.rmtree(download_path) - os.makedirs(download_path) - logging.info("[*] Create directory {}".format(download_path)) - - if not os.path.exists(history_path): - os.makedirs(history_path) - logging.info("[*] Create directory {}".format(history_path)) + self.__create_dir_list__.append(app_dir) + self.__create_dir_list__.append(history_path) + + # 构建结果输出目录 + def __build_result_out__path__(self,script_root_dir,user_out_path): + result_out_dir = script_root_dir + + # 用户指定输出目录结果则为输出到指定目录 + if user_out_path: + result_out_dir = user_out_path - if os.path.exists(txt_result_path): - os.remove(txt_result_path) + # 统一输出目录 + out_dir = os.path.join(result_out_dir,"out") + # 临时结果输出目录 + decode_dir = os.path.join(out_dir,"decode") + # 临时文件下载目录 + download_dir = os.path.join(out_dir,"download") + # 最终结果输出目录 + result_dir = os.path.join(out_dir,"result") + + self.__create_dir_list__.append(out_dir) + self.__create_dir_list__.append(decode_dir) + self.__create_dir_list__.append(download_dir) + self.__create_dir_list__.append(result_dir) + self.__remove_dir_list__.append(decode_dir) + + # 统一目录构建中心 + def __build_dir__(self): + for dir_path in self.__create_dir_list__: + if (os.path.exists(dir_path)) and (dir_path in self.__remove_dir_list__): + # 删除目录 + try: + shutil.rmtree(dir_path) + logging.info("[-] Remove Dir: {}".format(dir_path)) + except Exception as e: + # 解决windows超长文件名删除问题 + if not (platform.system() == "Windows"): + raise e + self.__removed_dirs_cmd__(dir_path) + + # 创建目录 + if not os.path.exists(dir_path): + os.makedirs(dir_path) + logging.info("[+] Create Dir: {}".format(dir_path)) - if os.path.exists(xls_result_path): - os.remove(xls_result_path) - def __removed_dirs_cmd__(self,output_path): files = os.listdir(output_path) for file in files: @@ -122,8 +182,12 @@ class Bootstrapper(object): old_dir = os.path.join(output_path,file) if not os.path.exists(new_dir): os.makedirs(new_dir) + logging.info("[+] Create Dir: {}".format(new_dir)) os.chdir(output_path) cmd = ("robocopy %s %s /purge") % (new_dir, old_dir) + logging.debug("[*] cmd : {}".format(cmd)) os.system(cmd) os.removedirs(new_dir) - os.removedirs(old_dir) \ No newline at end of file + os.removedirs(old_dir) + logging.info("[-] Remove Dir: {}".format(new_dir)) + logging.info("[-] Remove Dir: {}".format(old_dir)) \ No newline at end of file diff --git a/libs/core/net.py b/libs/core/net.py index 636e420..68cea85 100644 --- a/libs/core/net.py +++ b/libs/core/net.py @@ -28,20 +28,22 @@ class NetThreads(threading.Thread): url_ip = domains["url_ip"] time.sleep(2) result = self.__get_request_result__(url_ip) - logging.info("[+] Processing URL address："+url_ip) + logging.info("[+] " + url_ip) if result != "error": if self.lock.acquire(True): cores.excel_row = cores.excel_row + 1 - self.worksheet.write(cores.excel_row, 0, label = cores.excel_row) - self.worksheet.write(cores.excel_row, 1, label = url_ip) - self.worksheet.write(cores.excel_row, 2, label = domain) + self.worksheet.cell(row=cores.excel_row, column=1).value = cores.excel_row + self.worksheet.cell(row=cores.excel_row, column=2).value = url_ip + self.worksheet.cell(row=cores.excel_row, column=3).value = domain + if result != "timeout": - self.worksheet.write(cores.excel_row, 3, label = result["status"]) - self.worksheet.write(cores.excel_row, 4, label = result["des_ip"]) - self.worksheet.write(cores.excel_row, 5, label = result["server"]) - self.worksheet.write(cores.excel_row, 6, label = result["title"]) - self.worksheet.write(cores.excel_row, 7, label = result["cdn"]) - # self.worksheet.write(cores.excel_row, 8, label = "") + self.worksheet.cell(row=cores.excel_row, column=4).value = result["status"] + self.worksheet.cell(row=cores.excel_row, column=5).value = result["des_ip"] + self.worksheet.cell(row=cores.excel_row, column=6).value = result["server"] + self.worksheet.cell(row=cores.excel_row, column=7).value = result["title"] + self.worksheet.cell(row=cores.excel_row, column=8).value = result["cdn"] + self.worksheet.cell(row=cores.excel_row, column=9).value = "" + self.lock.release() def __get_request_result__(self,url): diff --git a/libs/core/parses.py b/libs/core/parses.py index 48b5ed6..61f6cce 100644 --- a/libs/core/parses.py +++ b/libs/core/parses.py @@ -81,7 +81,7 @@ class ParsesThreads(threading.Thread): self.threadLock.acquire() if cores.all_flag: - logging.info("[+] The string searched for matching rule is: %s" % (resl_str)) + logging.info("[+] String : {}".format(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 334efb0..6193b0f 100644 --- a/libs/task/android_task.py +++ b/libs/task/android_task.py @@ -38,8 +38,6 @@ class AndroidTask(object): return {"comp_list":self.comp_list,"shell_flag":self.shell_flag,"file_queue":self.file_queue,"packagename":self.packagename,"file_identifier":self.file_identifier} def __decode_file__(self,file_path): - apktool_path = str(cores.apktool_path) - backsmali_path = str(cores.backsmali_path) base_out_path = str(cores.output_path) filename = os.path.basename(file_path) suffix_name = filename.split(".")[-1] @@ -47,7 +45,7 @@ class AndroidTask(object): if suffix_name == "apk": name = filename.split(".")[0] output_path = os.path.join(base_out_path,name) - self.__decode_apk__(file_path,apktool_path,output_path) + self.__decode_apk__(file_path,output_path) elif suffix_name == "dex": f = open(file_path,'rb') md5_obj = hashlib.md5() @@ -61,7 +59,7 @@ class AndroidTask(object): output_path = os.path.join(base_out_path,dex_md5) if not os.path.exists(output_path): os.makedirs(output_path) - self.__decode_dex__(file_path,backsmali_path,output_path) + self.__decode_dex__(file_path,output_path) else: return "error" @@ -76,23 +74,25 @@ class AndroidTask(object): continue # 分解apk - def __decode_apk__(self,file_path,apktool_path,output_path): - cmd_str = ('java -jar "%s" d -f "%s" -o "%s" --only-main-classe') % (str(apktool_path),str(file_path),str(output_path)) + def __decode_apk__(self,file_path,output_path): + cmd_str = ('java -jar "%s" d -f "%s" -o "%s" --only-main-classe') % (cores.apktool_file,str(file_path),str(output_path)) + logging.debug("[*] cmd {}".format(cmd_str)) if os.system(cmd_str) == 0: self.__shell_test__(output_path) self.__scanner_file_by_apktool__(output_path) else: - logging.info("[-] Decompilation failed, please submit error information at https://github.com/kelvinBen/AppInfoScanner/issues") + logging.error("[x] Decompilation failed, please submit error information at https://github.com/kelvinBen/AppInfoScanner/issues") raise Exception(file_path + ", Decompilation failed.") # 分解dex - def __decode_dex__(self,file_path,backsmali_path,output_path): - cmd_str = ('java -jar "%s" d "%s"') % (str(backsmali_path),str(file_path)) + def __decode_dex__(self,file_path,output_path): + cmd_str = ('java -jar "%s" d "%s"') % (cores.backsmali_file,str(file_path)) + logging.debug("[*] cmd {}".format(cmd_str)) if os.system(cmd_str) == 0: self.__get_scanner_file__(output_path) else: - logging.info("[-] Decompilation failed, please submit error information at https://github.com/kelvinBen/AppInfoScanner/issues") + logging.error("[x] 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 02e97de..e90386c 100644 --- a/libs/task/base_task.py +++ b/libs/task/base_task.py @@ -17,68 +17,95 @@ from libs.task.android_task import AndroidTask from libs.task.download_task import DownloadTask class BaseTask(object): - thread_list =[] - result_dict = {} - app_history_list=[] - domain_history_list=[] - # 统一初始化入口 - def __init__(self, types="Android", inputs="", rules="", sniffer=True, threads=10, package=""): - self.types = types - self.path = inputs - if rules: - config.filter_strs.append(r'.*'+str(rules)+'.*') - self.sniffer = not sniffer - self.threads = threads - self.package = package + + def __init__(self): + if cores.user_add_rules: + config.filter_strs.append(r'.*'+str(cores.user_add_rules)+'.*') self.file_queue = Queue() + self.file_path_list = [] + self.thread_list = [] + self.app_history_list= [] + self.domain_history_list = [] + self.result_dict = {} - - # 统一调度平台 - def start(self): - - logging.info("[*] AI is analyzing filtering rules......") + # 统一启动 + def start(self, types="Android", user_input_path="", package=""): + # 如果输入路径为目录，则自动检索DEX、IPA、APK等文件 + if not(types == "Web") and os.path.isdir(user_input_path): + self.__scanner_specified_file__(self.file_path_list, user_input_path) + # 如果输入的路径为txt, 则加载txt中的内容 + elif user_input_path.endswith("txt"): + with open(user_input_path) as f: + lines = f.readlines() + for line in lines: + if line.startswith("http://") or line.startswith("https://") or line.endswith("apk") or line.endswith(".dex") or line.endswith("ipa"): + self.file_path_list.append(line) + f.close() + else: + # 如果是文件则追加到文件列表中 + self.file_path_list.append(user_input_path) + + # 长度小于1需重新选择目录 + if len(self.file_path_list) < 1: + raise Exception('[x] The specified DEX, IPA and APK files are not found. Please re-enter the directory to be scanned!') + + # 遍历目录 + for file_path in self.file_path_list: + self.__control_center__(file_path,types) - # 获取历史记录 + # 控制中心 + def __control_center__(self,file_path,types): + logging.info("[*] Processing {}".format(file_path)) + logging.info("[*] AI is analyzing filtering rules......") + + # 处理历史记录 self.__history_handle__() + logging.info("[*] The filtering rules obtained by AI are as follows: {}".format(set(config.filter_no))) - logging.info("[*] The filtering rules obtained by AI are as follows: %s" % (set(config.filter_no)) ) + # AI 修正扫描类型 + cache_info = DownloadTask().start(file_path, types) + cacar_path = cache_info["path"] + types = cache_info["type"] # 任务控制中心 - task_info = self.__tast_control__() + task_info = self.__tast_control__(file_path) if len(task_info) < 1: - return - - file_queue = task_info["file_queue"] - shell_flag = task_info["shell_flag"] - comp_list = task_info["comp_list"] - packagename = task_info["packagename"] - file_identifier = task_info["file_identifier"] - - if shell_flag: - 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 + return + + # 文件队列 + file_queue = task_info["file_queue"] + # 是否存在壳 + shell_flag = task_info["shell_flag"] + # 组件列表(仅适用于Android) + comp_list = task_info["comp_list"] + # 报名信息(仅适用于Android) + packagename = task_info["packagename"] + # 文件标识符 + file_identifier = task_info["file_identifier"] + + if shell_flag: + logging.error('[x] This application has shell, the retrieval results may not be accurate, Please remove the shell and try again!') + continue - # 线程控制中心 - logging.info("[*] ========= Searching for strings that match the rules ===============") - self.__threads_control__(file_queue) + # 线程控制中心 + logging.info("[*] ========= Searching for strings that match the rules ===============") + self.__threads_control__(file_queue) - # 等待线程结束 - for thread in self.thread_list: - thread.join() - - # 结果输出中心 - self.__print_control__(packagename,comp_list,file_identifier) + # 等待线程结束 + for thread in self.thread_list: + thread.join() + + # 结果输出中心 + self.__print_control__(packagename,comp_list,file_identifier) - def __tast_control__(self): + # 任务控制中心 + def __tast_control__(self,user_input_path): task_info = {} - # 自动根据文件后缀名称进行修正 - cache_info = DownloadTask().start(self.path,self.types) - cacar_path = cache_info["path"] - types = cache_info["type"] + if (not os.path.exists(cacar_path) and cores.download_flag): - logging.info("[-] File download failed! Please download the file manually and try again.") + logging.error("[x] File download failed! Please download the file manually and try again.") return task_info # 调用Android 相关处理逻辑 @@ -92,21 +119,19 @@ class BaseTask(object): task_info = WebTask(cacar_path).start() return task_info + # 线程控制中心 def __threads_control__(self,file_queue): - for threadID in range(1,self.threads): + for threadID in range(1, cores.threads_num): name = "Thread - " + str(int(threadID)) thread = ParsesThreads(threadID,name,file_queue,self.result_dict,self.types) thread.start() self.thread_list.append(thread) - - def __print_control__(self,packagename,comp_list,file_identifier): - txt_result_path = cores.txt_result_path - xls_result_path = cores.xls_result_path - all_flag = cores.all_flag - - if self.sniffer: + + # 信息输出中心 + def __print_control__(self,packagename,comp_list,file_identifier): + if cores.net_sniffer_flag: 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() + NetTask(self.result_dict,self.app_history_list,self.domain_history_list,file_identifier).start() if packagename: logging.info("[*] ========= The package name of this APP is: ===============") @@ -117,7 +142,7 @@ class BaseTask(object): for json in comp_list: logging.info(json) - if all_flag: + if cores.all_flag: value_list = [] with open(txt_result_path,"a+",encoding='utf-8',errors='ignore') as f: for key,value in self.result_dict.items(): @@ -128,11 +153,12 @@ class BaseTask(object): value_list.append(result) f.write("\t"+result+"\r") f.close() - logging.info("[*] 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: {}".format(cores.txt_result_path)) - if self.sniffer: - logging.info("[*] For more information about the search, see XLS file result: %s" %(xls_result_path)) + if cores.net_sniffer_flag: + logging.info("[>] For more information about the search, see XLS file result: {}".format(cores.xls_result_path)) + # 获取历史记录 def __history_handle__(self): domain_history_path = cores.domain_history_path app_history_path = cores.app_history_path @@ -159,5 +185,14 @@ class BaseTask(object): if domain_count >= cout: config.filter_no.append(".*" + domain) f.close() - + # 扫描指定后缀文件 + def __scanner_specified_file__(self, file_list, root_dir, file_suffix=['dex','ipa','apk']): + dir_or_files = os.listdir(root_dir) + for dir_or_file in dir_or_files: + dir_or_file_path = os.path.join(root_dir,dir_or_file) + if os.path.isdir(dir_or_file_path): + self.__scanner_specified_file__(file_list,dir_or_file_path,file_suffix) + else: + if dir_or_file_path.split(".")[-1] in file_suffix: + file_list.append(dir_or_file_path) \ No newline at end of file diff --git a/libs/task/download_task.py b/libs/task/download_task.py index 7c311a6..ebba650 100644 --- a/libs/task/download_task.py +++ b/libs/task/download_task.py @@ -7,40 +7,70 @@ import re import time import config import hashlib +import logging from queue import Queue import libs.core as cores from libs.core.download import DownloadThreads class DownloadTask(object): + def __init__(self): + self.download_file_queue = Queue() + self.thread_list = [] - def start(self,path,types): + def start(self, path, types): + self.__local_or_remote__(path, types) + + for threadID in range(1, cores.threads_num): + name = "Thread - " + str(int(threadID)) + thread = DownloadThreads(threadID,name,self.download_file_queue) + thread.start() + thread.join() + + # 判断文件是本地加载还是远程加载 + def __local_or_remote__(self,path,types): + # 处理本地文件 + if not(path.startswith("http://") or path.startswith("https://")): + if not os.path.isdir(path): # 不是目录 + return {"path":path,"type":types} + else: # 目录处理 + return {"path":path,"type":types} + else: + self.__net_header__(path,types) + # self.download_file_queue.put(path) + + # 处理网络请求 + def __net_header__(self, path, types): create_time = time.strftime("%Y%m%d%H%M%S", time.localtime()) - if path.endswith("apk"): + if path.endswith("apk") or types == "Android": types = "Android" file_name = create_time+ ".apk" - elif path.endswith("ipa"): + elif path.endswith("ipa") or types == "iOS": types = "iOS" file_name = create_time + ".ipa" else: - if types == "Android": - types = "Android" + types = "WEB" + file_name = create_time + ".html" + + logging.info("[*] Detected that the task is not local, preparing to download file......") + cache_path = os.path.join(cores.download_dir, file_name) + self.download_file_queue.put({"path":path, "cache_path":cache_path, "types":types}) + # thread = DownloadThreads(path,file_name,cache_path,types) + # thread.start() + # thread.join() + return {"path":cache_path,"type":types} + + def __update_type__(self, path, types, file_name=None): + create_time = time.strftime("%Y%m%d%H%M%S", time.localtime()) + if path.endswith("apk") or types == "Android": + types = "Android" + if not file_name: file_name = create_time+ ".apk" - elif types == "iOS": - types = "iOS" + elif path.endswith("ipa") or types == "iOS": + types = "iOS" + if not file_name: file_name = create_time + ".ipa" - else: - types = "WEB" - file_name = create_time + ".html" - if not(path.startswith("http://") or path.startswith("https://")): - if not os.path.isdir(path): # 不是目录 - return {"path":path,"type":types} - else: # 目录处理 - return {"path":path,"type":types} else: - 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() - logging.info() - return {"path":cache_path,"type":types} \ No newline at end of file + types = "WEB" + if not file_name: + file_name = create_time + ".html" + return types,file_name \ No newline at end of file diff --git a/libs/task/ios_task.py b/libs/task/ios_task.py index 7a640ff..6c64a5d 100644 --- a/libs/task/ios_task.py +++ b/libs/task/ios_task.py @@ -3,8 +3,6 @@ # Author: kelvinBen # Github: https://github.com/kelvinBen/AppInfoScanner import os -import re -import shutil import zipfile import binascii import platform @@ -23,13 +21,12 @@ class iOSTask(object): file_path = self.path if file_path.split(".")[-1] == 'ipa': self.__decode_ipa__(cores.output_path) - self.__scanner_file_by_ipa__(cores.output_path) elif self.__get_file_header__(file_path): self.file_queue.put(file_path) else: raise Exception("Retrieval of this file type is not supported. Select IPA file or Mach-o file.") 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): crypt_load_command_hex = "2C000000" macho_name = os.path.split(file_path)[-1] @@ -50,11 +47,6 @@ class iOSTask(object): macho_file.close() return False - def __scanner_file_by_ipa__(self,output): - scanner_file_suffix = ["plist","js","xml","html"] - scanner_dir = os.path.join(output,"Payload") - self.__get_scanner_file__(scanner_dir,scanner_file_suffix) - def __get_scanner_file__(self,scanner_dir,file_suffix): dir_or_files = os.listdir(scanner_dir) for dir_file in dir_or_files: @@ -76,6 +68,9 @@ class iOSTask(object): self.file_queue.put(dir_file_path) def __decode_ipa__(self,output_path): + scanner_file_suffix = ["plist","js","xml","html"] + scanner_dir = os.path.join(output_path,"Payload") + with zipfile.ZipFile(self.path,"r") as zip_files: zip_file_names = zip_files.namelist() for zip_file_name in zip_file_names: @@ -90,6 +85,7 @@ class iOSTask(object): 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) + self.__get_scanner_file__(scanner_dir,scanner_file_suffix) def __get_parse_dir__(self,output_path,file_path): start = file_path.index("Payload/") diff --git a/libs/task/net_task.py b/libs/task/net_task.py index a10e4ba..9ddcb19 100644 --- a/libs/task/net_task.py +++ b/libs/task/net_task.py @@ -2,32 +2,29 @@ # -*- coding: utf-8 -*- # Author: kelvinBen # Github: https://github.com/kelvinBen/AppInfoScanner - -import re -import xlwt -import socket import config from queue import Queue import libs.core as cores +from openpyxl import Workbook from libs.core.net import NetThreads -import requests class NetTask(object): value_list = [] domain_list=[] - def __init__(self,result_dict,app_history_list,domain_history_list,file_identifier,threads): + def __init__(self,result_dict,app_history_list,domain_history_list,file_identifier): self.result_dict = result_dict self.app_history_list = app_history_list + self.domain_history_list = domain_history_list self.file_identifier = file_identifier + self.domain_queue = Queue() - self.threads = int(threads) self.thread_list = [] - self.domain_history_list = domain_history_list + def start(self): xls_result_path = cores.xls_result_path - workbook = xlwt.Workbook(encoding = 'utf-8') + workbook = Workbook() worksheet = self.__creating_excel_header__(workbook) self.__write_result_to_txt__() @@ -40,17 +37,12 @@ class NetTask(object): workbook.save(xls_result_path) def __creating_excel_header__(self,workbook): - worksheet = workbook.add_sheet("Result",cell_overwrite_ok=True) - worksheet.write(0,0, label = "Number") - worksheet.write(0,1, label = "IP/URL") - worksheet.write(0,2, label = "Domain") - worksheet.write(0,3, label = "Status") - worksheet.write(0,4, label = "IP") - worksheet.write(0,5, label = "Server") - worksheet.write(0,6, label = "Title") - worksheet.write(0,7, label = "CDN") - # worksheet.write(0,8, label = "Finger") - return worksheet + worksheet = workbook.create_sheet("Result",0) + excel_headers = ["Number","IP/URL","Domain","Status","IP","Server","Title","CDN","Finger"] + for head_cell in excel_headers: + column = excel_headers.index(head_cell) + 1 + worksheet.cell(row=1, column=column).value = head_cell + return worksheet def __write_result_to_txt__(self): txt_result_path = cores.txt_result_path @@ -97,7 +89,7 @@ class NetTask(object): append_file_flag = False def __start_threads__(self,worksheet): - for threadID in range(0,self.threads) : + for threadID in range(0, cores.threads_num): name = "Thread - " + str(threadID) thread = NetThreads(threadID,name,self.domain_queue,worksheet) thread.start() diff --git a/requirements.txt b/requirements.txt index e0753ef..e05d17f 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,3 +1,5 @@ requests click -xlwt \ No newline at end of file +xlwt +pillow +openpyxl \ No newline at end of file diff --git a/update.md b/update.md index b0de0e7..5411d1c 100644 --- a/update.md +++ b/update.md @@ -1,12 +1,15 @@ ### V1.0.9 +- 新增IPA批量操作 - 优化iOS壳检测速度 -- 优化iPA文件效率 +- 优化iPA文件解压效率 - 优化日志输出 +- 修复内容过长无法输出到excle + ### V1.0.8 -- 添加AK和SK的检测 -- 添加检测规则提交入口 -- 添加.gitignore文件 +- 新增AK和SK的检测 +- 新增检测规则提交入口 +- 新增.gitignore文件 - 优化txt结果集输出方式 - 修复目录中包含空格无法解析的问题 - 修复WEB页面或者目录扫描的问题