@ -5,6 +5,7 @@
import os import os
import re import re
import config import config
import logging
import threading import threading
from queue import Queue from queue import Queue
import libs . core as cores import libs . core as cores
@ -13,52 +14,100 @@ from libs.task.web_task import WebTask
from libs . task . net_task import NetTask from libs . task . net_task import NetTask
from libs . core . parses import ParsesThreads from libs . core . parses import ParsesThreads
from libs . task . android_task import AndroidTask from libs . task . android_task import AndroidTask
from libs . task . download_task import DownloadTask from libs . core . download import DownloadThre ad s
class BaseTask ( object ) : 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
self . file_queue = Queue ( )
def __init__ ( self ) :
if cores . user_add_rules :
config . filter_strs . append ( r ' .* ' + str ( cores . user_add_rules ) + ' .* ' )
self . file_queue = Queue ( )
# 文件下载队列
self . download_file_queue = Queue ( )
# 文件下载列表
self . download_file_list = [ ]
self . thread_list = [ ]
self . app_history_list = [ ]
self . domain_history_list = [ ]
self . result_dict = { }
# 统一启动
def start ( self , types = " Android " , user_input_path = " " , package = " " ) :
# 如果输入路径为目录,且类型非web,则自动检索DEX、IPA、APK等文件
if not ( types == " Web " ) and os . path . isdir ( user_input_path ) :
self . __scanner_specified_file__ ( 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 :
# http:// 或者 https:// 开头 或者 apk/dex/ipa结尾的文件且文件存在
if ( line . startswith ( " http:// " ) or line . startswith ( " https:// " ) ) or ( ( line . endswith ( " apk " ) or line . endswith ( " .dex " ) or line . endswith ( " ipa " ) ) and os . path . exists ( line ) ) :
self . download_file_queue . put ( line )
f . close ( )
else :
# 如果是文件或者类型为web的目录
self . download_file_queue . put ( user_input_path )
# 长度小于1需重新选择目录
if self . download_file_queue . qsize ( ) < 1 :
raise Exception ( ' [x] The specified DEX, IPA and APK files are not found. Please re-enter the directory to be scanned! ' )
# 统一文件下载中心
self . __download_file_center__ ( types )
for download_file in self . download_file_list :
file_path = download_file [ " path " ]
types = download_file [ " type " ]
# 控制中心
self . __control_center__ ( file_path , types )
# 统一文件下载中心
def __download_file_center__ ( self , types ) :
# 杜绝资源浪费
if self . download_file_queue . qsize ( ) < cores . threads_num :
threads_num = self . download_file_queue . qsize ( )
else :
threads_num = cores . threads_num
# 统一调度平台 for threadID in range ( 1 , threads_num ) :
def start ( self ) : threadName = " Thread - " + str ( int ( threadID ) )
thread = DownloadThreads ( threadID , threadName , self . download_file_queue , self . download_file_list , types )
thread . start ( )
thread . join ( )
print ( " [*] AI is analyzing filtering rules...... " ) # 控制中心
def __control_center__ ( self , file_path , types ) :
logging . info ( " [*] Processing {} " . format ( file_path ) )
logging . info ( " [*] AI is analyzing filtering rules...... " )
# 获取历史记录 # 处理 历史记录
self . __history_handle__ ( ) self . __history_handle__ ( )
logging . info ( " [*] The filtering rules obtained by AI are as follows: {} " . format ( set ( config . filter_no ) ) )
print ( " [*] The filtering rules obtained by AI are as follows: %s " % ( set ( config . filter_no ) ) )
# 任务控制中心 # 任务控制中心
task_info = self . __tast_control__ ( ) task_info = self . __tast_control__ ( file_path , types )
if len ( task_info ) < 1 : if len ( task_info ) < 1 :
return return
# 文件队列
file_queue = task_info [ " file_queue " ] file_queue = task_info [ " file_queue " ]
# 是否存在壳
shell_flag = task_info [ " shell_flag " ] shell_flag = task_info [ " shell_flag " ]
# 组件列表(仅适用于Android)
comp_list = task_info [ " comp_list " ] comp_list = task_info [ " comp_list " ]
# 报名信息(仅适用于Android)
packagename = task_info [ " packagename " ] packagename = task_info [ " packagename " ]
# 文件标识符
file_identifier = task_info [ " file_identifier " ] file_identifier = task_info [ " file_identifier " ]
if shell_flag : 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 . error ( ' [x] This application has shell, the retrieval results may not be accurate, Please remove the shell and try again! ' )
return return
# 线程控制中心 # 线程控制中心
print ( " [*] ========= Searching for strings that match the rules =============== " ) logging . info ( " [*] ========= Searching for strings that match the rules =============== " )
self . __threads_control__ ( file_queue ) self . __threads_control__ ( file_queue )
# 等待线程结束 # 等待线程结束
@ -69,54 +118,50 @@ class BaseTask(object):
self . __print_control__ ( packagename , comp_list , file_identifier ) self . __print_control__ ( packagename , comp_list , file_identifier )
def __tast_control__ ( self ) : # 任务控制中心
def __tast_control__ ( self , file_path , types ) :
task_info = { } 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 ) : # 通过网络下载的文件如果不存在就直接返回任务控制中心
print ( " [-] File download failed! Please download the file manually and try again. " ) if ( not os . path . exists ( file_path ) and cores . download_flag ) :
logging . error ( " [x] {} download failed! Please download the file manually and try again. " . format ( file_path ) )
return task_info return task_info
# 调用Android 相关处理逻辑 # 调用Android 相关处理逻辑
if types == " Android " : if types == " Android " :
task_info = AndroidTask ( cacar_path , self . package ) . start ( ) task_info = AndroidTask ( file_path , self . package ) . start ( )
# 调用iOS 相关处理逻辑 # 调用iOS 相关处理逻辑
elif types == " iOS " : elif types == " iOS " :
task_info = iOSTask ( cacar _path) . start ( ) task_info = iOSTask ( file _path) . start ( )
# 调用Web 相关处理逻辑 # 调用Web 相关处理逻辑
else : else :
task_info = WebTask ( cacar _path) . start ( ) task_info = WebTask ( file _path) . start ( )
return task_info return task_info
# 线程控制中心
def __threads_control__ ( self , file_queue ) : 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 ) ) name = " Thread - " + str ( int ( threadID ) )
thread = ParsesThreads ( threadID , name , file_queue , self . result_dict , self . types ) thread = ParsesThreads ( threadID , name , file_queue , self . result_dict , self . types )
thread . start ( ) thread . start ( )
self . thread_list . append ( thread ) self . thread_list . append ( thread )
# 信息输出中心
def __print_control__ ( self , packagename , comp_list , file_identifier ) : def __print_control__ ( self , packagename , comp_list , file_identifier ) :
txt_result_path = cores . txt_result_path if cores . net_sniffer_flag :
xls_result_path = cores . xls_result_path logging . info ( " [*] ========= Sniffing the URL address of the search =============== " )
all_flag = cores . all_flag NetTask ( self . result_dict , self . app_history_list , self . domain_history_list , file_identifier ) . start ( )
if self . sniffer :
print ( " [*] ========= 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 : if packagename :
print ( " [*] ========= The package name of this APP is: =============== " ) logging . info ( " [*] ========= The package name of this APP is: =============== " )
print ( packagename ) logging . info ( packagename )
if len ( comp_list ) != 0 : if len ( comp_list ) != 0 :
print ( " [*] ========= Component information is as follows :=============== " ) logging . info ( " [*] ========= Component information is as follows :=============== " )
for json in comp_list : for json in comp_list :
print ( json ) logging . info ( json )
if all_flag : if cores . all_flag :
value_list = [ ] value_list = [ ]
with open ( txt_result_path , " a+ " , encoding = ' utf-8 ' , errors = ' ignore ' ) as f : with open ( txt_result_path , " a+ " , encoding = ' utf-8 ' , errors = ' ignore ' ) as f :
for key , value in self . result_dict . items ( ) : for key , value in self . result_dict . items ( ) :
@ -127,11 +172,12 @@ class BaseTask(object):
value_list . append ( result ) value_list . append ( result )
f . write ( " \t " + result + " \r " ) f . write ( " \t " + result + " \r " )
f . close ( ) 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: {} " . format ( cores . txt_result_path ) )
if self . sniffer : if cores . net_sniffer_flag :
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: {} " . format ( cores . xls_result_path ) )
# 获取历史记录
def __history_handle__ ( self ) : def __history_handle__ ( self ) :
domain_history_path = cores . domain_history_path domain_history_path = cores . domain_history_path
app_history_path = cores . app_history_path app_history_path = cores . app_history_path
@ -159,4 +205,13 @@ class BaseTask(object):
config . filter_no . append ( " .* " + domain ) config . filter_no . append ( " .* " + domain )
f . close ( ) f . close ( )
# 扫描指定后缀文件
def __scanner_specified_file__ ( self , base_dir , file_suffix = [ ' dex ' , ' ipa ' , ' apk ' ] ) :
files = os . listdir ( base_dir )
for file in files :
dir_or_file_path = os . path . join ( base_dir , file )
if os . path . isdir ( dir_or_file_path ) :
self . __scanner_specified_file__ ( dir_or_file_path , file_suffix )
else :
if dir_or_file_path . split ( " . " ) [ - 1 ] in file_suffix :
self . download_file_queue . put ( dir_or_file_path )
kelvinBen
kelvinBen