- 优化iOS壳检测速度

- 优化iPA文件解压效率
- 优化日志输出
v1.0.9
kelvinBen 4 years ago
parent 72cd09be93
commit 1f0fa65606
  1. 6
      app.py
  2. 7
      libs/core/__init__.py
  3. 7
      libs/core/download.py
  4. 3
      libs/core/net.py
  5. 5
      libs/core/parses.py
  6. 5
      libs/task/android_task.py
  7. 25
      libs/task/base_task.py
  8. 4
      libs/task/download_task.py
  9. 63
      libs/task/ios_task.py
  10. 3
      libs/task/web_task.py
  11. 5
      update.md

@ -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__":

@ -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)

@ -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()

@ -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

@ -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

@ -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.")

@ -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

@ -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}

@ -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/")

@ -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)

@ -1,3 +1,8 @@
### V1.0.9
- 优化iOS壳检测速度
- 优化iPA文件效率
- 优化日志输出
### V1.0.8
- 添加AK和SK的检测
- 添加检测规则提交入口

Loading…
Cancel
Save