You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
217 lines
8.8 KiB
217 lines
8.8 KiB
#! /usr/bin/python3
|
|
# -*- coding: utf-8 -*-
|
|
# Author: kelvinBen
|
|
# Github: https://github.com/kelvinBen/AppInfoScanner
|
|
import os
|
|
import re
|
|
import config
|
|
import logging
|
|
import threading
|
|
from queue import Queue
|
|
import libs.core as cores
|
|
from libs.task.ios_task import iOSTask
|
|
from libs.task.web_task import WebTask
|
|
from libs.task.net_task import NetTask
|
|
from libs.core.parses import ParsesThreads
|
|
from libs.task.android_task import AndroidTask
|
|
from libs.core.download import DownloadThreads
|
|
|
|
class BaseTask(object):
|
|
|
|
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):
|
|
threadName = "Thread - " + str(int(threadID))
|
|
thread = DownloadThreads(threadID, threadName, self.download_file_queue, self.download_file_list, types)
|
|
thread.start()
|
|
thread.join()
|
|
|
|
# 控制中心
|
|
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)))
|
|
|
|
# 任务控制中心
|
|
task_info = self.__tast_control__(file_path, types)
|
|
if len(task_info) < 1:
|
|
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!')
|
|
return
|
|
|
|
# 线程控制中心
|
|
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)
|
|
|
|
|
|
# 任务控制中心
|
|
def __tast_control__(self, file_path, types):
|
|
task_info = {}
|
|
|
|
# 通过网络下载的文件如果不存在就直接返回任务控制中心
|
|
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
|
|
|
|
# 调用Android 相关处理逻辑
|
|
if types == "Android":
|
|
task_info = AndroidTask(file_path, self.package).start()
|
|
# 调用iOS 相关处理逻辑
|
|
elif types == "iOS":
|
|
task_info = iOSTask(file_path).start()
|
|
# 调用Web 相关处理逻辑
|
|
else:
|
|
task_info = WebTask(file_path).start()
|
|
return task_info
|
|
|
|
# 线程控制中心
|
|
def __threads_control__(self,file_queue):
|
|
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):
|
|
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).start()
|
|
|
|
if packagename:
|
|
logging.info("[*] ========= The package name of this APP is: ===============")
|
|
logging.info(packagename)
|
|
|
|
if len(comp_list) != 0:
|
|
logging.info("[*] ========= Component information is as follows :===============")
|
|
for json in comp_list:
|
|
logging.info(json)
|
|
|
|
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():
|
|
f.write(key+"\r")
|
|
for result in value:
|
|
if result in value_list:
|
|
continue
|
|
value_list.append(result)
|
|
f.write("\t"+result+"\r")
|
|
f.close()
|
|
logging.info("[>] For more information about the search, see TXT file result: {}".format(cores.txt_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
|
|
if os.path.exists(domain_history_path):
|
|
domain_counts = {}
|
|
app_size = 0
|
|
with open(app_history_path,"r",encoding='utf-8',errors='ignore') as f:
|
|
lines = f.readlines()
|
|
app_size = len(lines)
|
|
for line in lines:
|
|
self.app_history_list.append(line.replace("\r","").replace("\n",""))
|
|
|
|
f.close()
|
|
|
|
with open(domain_history_path,"r",encoding='utf-8',errors='ignore') as f:
|
|
lines = f.readlines()
|
|
cout = 3
|
|
if (app_size>3) and (app_size%3==0):
|
|
cout = cout + 1
|
|
for line in lines:
|
|
domain = line.replace("\r","").replace("\n","")
|
|
self.domain_history_list.append(domain)
|
|
domain_count = lines.count(line)
|
|
if domain_count >= cout:
|
|
config.filter_no.append(".*" + domain)
|
|
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) |