From 310c40d7aa8b6e48617eddcf74cee879ce472e10 Mon Sep 17 00:00:00 2001 From: shmilylty Date: Sun, 18 Aug 2019 19:34:16 +0800 Subject: [PATCH] =?UTF-8?q?=E6=B7=BB=E5=8A=A0=E5=AD=90=E5=9F=9F=E6=8E=A5?= =?UTF-8?q?=E7=AE=A1=E6=A3=80=E6=9F=A5=E5=8A=9F=E8=83=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- oneforall/oneforall.py | 13 +- oneforall/takeover.py | 282 +++++++++++++++++++---------------------- 2 files changed, 140 insertions(+), 155 deletions(-) diff --git a/oneforall/oneforall.py b/oneforall/oneforall.py index 8e57823..2314395 100644 --- a/oneforall/oneforall.py +++ b/oneforall/oneforall.py @@ -19,6 +19,7 @@ from collect import Collect from aiobrute import AIOBrute from common import utils, resolve, request from common.database import Database +from takeover import Takeover yellow = '\033[01;33m' white = '\033[01;37m' @@ -46,12 +47,13 @@ class OneForAll(object): Example: python3 oneforall.py --target example.com run - python3 oneforall.py --target ./domains.txt run + python3 oneforall.py --target ./subdomains.txt run python3 oneforall.py --target example.com --valid None run python3 oneforall.py --target example.com --brute True run python3 oneforall.py --target example.com --port medium run python3 oneforall.py --target example.com --format csv run python3 oneforall.py --target example.com --verify False run + python3 oneforall.py --target example.com --takeover False run python3 oneforall.py --target example.com --show True run Note: @@ -70,7 +72,7 @@ class OneForAll(object): :param bool show: 终端显示导出数据(默认False) """ def __init__(self, target, brute=None, verify=None, port='medium', valid=1, - format='xls', show=False): + format='xls', takeover=True, show=False): self.target = target self.port = port self.domains = set() @@ -78,6 +80,7 @@ class OneForAll(object): self.datas = list() self.brute = brute self.verify = verify + self.takeover = takeover self.valid = valid self.format = format self.show = show @@ -138,6 +141,12 @@ class OneForAll(object): db.drop_table(rename_table) db.rename_table(self.domain, rename_table) db.close() + # 子域接管检查 + + if self.takeover: + subdomains = set(map(lambda x: x.get('subdomain'), self.datas)) + takeover = Takeover(subdomains) + takeover.run() def run(self): print(banner) diff --git a/oneforall/takeover.py b/oneforall/takeover.py index d3ead0a..331ab43 100644 --- a/oneforall/takeover.py +++ b/oneforall/takeover.py @@ -1,167 +1,143 @@ -# encoding: utf-8 +#!/usr/bin/python3 +# coding=utf-8 -import requests -import dns.resolver -import sys,getopt,os,base64,json -import yaml +""" +OneForAll子域接管模块 + +:copyright: Copyright (c) 2019, Jing Ling. All rights reserved. +:license: GNU General Public License v3.0, see LICENSE for more details. +""" +import time +import json +from threading import Thread +from queue import Queue +import fire +from tablib import Dataset import config +from common import resolve, utils +from common.module import Module +from common.domain import Domain +from config import logger -HEADERS = { - "Accept":"application/json, text/javascript, */*; q=0.01", - "Accept-Language":"zh-CN,zh;q=0.9", - "User-Agent":"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_12_5) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/63.0.3239.84 Safari/537.36", -} -# github自动接管 -def github_takeover(url): - # 读取config配置文件 - repo_name = url - print('[*]正在读取配置文件...') - user = config.github_api_user - token = config.github_api_token - CHECK_HEADERS = { - "Authorization": 'token '+ token, - "Accept": "application/vnd.github.switcheroo-preview+json" - } - repos_url = 'https://api.github.com/repos/'+ user +'/' + repo_name - repos_r = requests.get(url=repos_url,headers=CHECK_HEADERS) - # 验证token是否正确 - if 'message' in repos_r.json(): - if repos_r.json()['message'] == 'Bad credentials': - print('[*]请检查Token是否正确') - elif repos_r.json()['message'] == 'Not Found': - print('[*]正在生成接管库...') # 生成接管库 - creat_repo_dict = { - "name": repo_name, - "description": "This is a subdomain takeover Repository", - } - creat_repo_url = 'https://api.github.com/user/repos' - creat_repo_r = requests.post(url=creat_repo_url,headers=CHECK_HEADERS,data=json.dumps(creat_repo_dict)) - creat_repo_status = creat_repo_r.status_code - if creat_repo_status == 201: - print('[*]创建接管库' + repo_name + '成功，正在进行自动接管...' ) - # 接管文件生成 - # index.html文件 - html = b''' - -

Subdomain Takerover Test! - - ''' - html64 = base64.b64encode(html).decode('utf-8') - html_dict = { - "message": "my commit message", - "committer": { - "name": "user", #提交id，非必改项 - "email": "user@163.com" #同上 - }, - "content": html64 - } - # CNAME文件 - cname_url = bytes(url,encoding='utf-8') - cname_url64 = base64.b64encode(cname_url).decode('utf-8') - url_dict = { - "message": "my commit message", - "committer": { - "name": "user", - "email": "user@163.com" - }, - "content": cname_url64 - } - html_url = 'https://api.github.com/repos/' + user + '/' + repo_name + '/contents/index.html' - url_url = 'https://api.github.com/repos/' + user + '/' + repo_name + '/contents/CNAME' - html_r = requests.put(url=html_url,data=json.dumps(html_dict), headers=CHECK_HEADERS) #上传index.html - cname_r = requests.put(url=url_url,data=json.dumps(url_dict), headers=CHECK_HEADERS) # 上传CNAME - rs=cname_r.status_code - if rs==201: - print('[*]生成接管库成功，正在开启Github pages...') - page_url = "https://api.github.com/repos/"+ user +"/"+url+"/pages" - page_dict={ - "source": { - "branch": "master" - } - } - page_r = requests.post(url=page_url,data=json.dumps(page_dict), headers=CHECK_HEADERS) # 开启page - if page_r.status_code == 201: - print('[+]自动接管成功，请稍后访问http://'+str(url)+'查看结果') - else: - print('[+]开启Github pages失败，请检查网络或稍后重试...') - else: - print('[+]生成接管库失败，请检查网络或稍后重试...') - elif url in repos_r.json()['name']: - print('[*]生成接管库失败，请检查https://github.com/'+user+ - '?tab=repositories是否存在同名接管库...') +def get_fingerprint(): + path = config.data_storage_path.joinpath('fingerprints.json') + with open(path) as file: + fingerprints = json.load(file) + return fingerprints -# 发起请求 -def url_get(url): - r = requests.get(url,HEADERS,timeout=5) - status_code = r.status_code - response_text = r.content.decode('utf-8') - return status_code,response_text -# 指纹读取 存储到json_dicts中 -def providers_read(): +def get_cname(subdomain): + resolver = resolve.dns_resolver() try: - with open('./data/providers.json','r') as f: - str_json = f.read() - json_dicts = json.loads(str_json) - return json_dicts - except: - print('[*] Wrong! 请检查是否存在providers.json文件') - -# 获取cname记录 -def cname_get(url): - print('[*]正在获取'+ url + '的CNAME记录') - try: - cn = dns.resolver.query(url,'CNAME') - for rrset in cn.response.answer: - for cname in rrset.items: - return (cname.to_text()) - except: #不存在cname解析，pass - print('[*]' + url + '未找到CNAME记录') + answers = resolver.query(subdomain, 'CNAME') + except Exception as e: + logger.log('DEBUG', e.args) + return None + for answer in answers: + return answer.to_text() # 一个子域只有一个CNAME记录 -# 检查是否存在子域接管漏洞 -def takeover_check(url,cname,fingercname_lists): - check_cname = 'http://' + cname - check_url = 'http://' + url - cnameresponse_text = url_get(check_cname)[1] # 解析cname返回文本 - url_response_text = url_get(check_url)[1] # 解析url返回文本 - # 与指纹对比查看cname对比判断是否存在接管风险，与url对比判断是否已被接管 - for fingerprint in fingercname_lists: - if fingerprint in cnameresponse_text: - print('[*]'+ url + '存在子域接管风险') - print('[*]正在检测当前是否已经被接管...') - if fingerprint in url_response_text: - print('[+]当前未被接管，url：' + url + ',CNMAE：' + cname) - else: - print('[*]当前可能已被接管，url：' + url + ',CNMAE：' + cname) +def get_maindomain(subdomain): + return Domain(subdomain).registered() + + +class Takeover(Module): + """ + OneForAll多线程子域接管风险检查模块 + + Example: + python3 takeover.py --target www.example.com --format csv run + python3 takeover.py --target ./subdomains.txt --thread 10 run + + Note: + 参数format可选格式有'txt', 'rst', 'csv', 'tsv', 'json', 'yaml', 'html', + 'jira', 'xls', 'xlsx', 'dbf', 'latex', 'ods' + 参数dpath为None默认使用OneForAll结果目录 + + :param str target: 单个子域或者每行一个子域的文件路径(必需参数) + :param int thread: 线程数(默认10) + :param str format: 导出格式(默认xls) + :param str dpath: 导出目录(默认None) + """ + def __init__(self, target, thread=10, dpath=None, format='xls'): + Module.__init__(self) + self.subdomains = set() + self.module = 'Check' + self.source = 'Takeover' + self.target = target + self.thread = thread + self.dpath = dpath + self.format = format + self.fingerprints = None + self.domainq = Queue() + self.cnames = list() + self.results = Dataset() + + def save(self): + logger.log('INFOR', '正在保存检查结果') + if self.format == 'txt': + data = str(self.results) else: - pass + data = self.results.export(self.format) + fpath = self.dpath.joinpath(f'takeover.{self.format}') + utils.save_data(fpath, data) -# 自动接管模块 -def auto_take(url): - github_takeover(url) # Github自动接管 + def compare(self, subdomain, cname, responses): + domain_resp = self.get('http://' + subdomain, check=False) + cname_resp = self.get('http://'+cname, check=False) + if domain_resp is None or cname_resp is None: + return + + for resp in responses: + if resp in domain_resp.text and resp in cname_resp.text: + logger.log('ALERT', f'{subdomain}存在子域接管风险') + self.results.append([subdomain, cname]) + break + + def check(self): + while not self.domainq.empty(): # 保证域名队列遍历结束后能退出线程 + subdomain = self.domainq.get() # 从队列中获取域名 + cname = get_cname(subdomain) + maindomain = get_maindomain(cname) + if cname is None: + return + for fingerprint in self.fingerprints: + cnames = fingerprint.get('cname') + if maindomain not in cnames: + continue + responses = fingerprint.get('response') + self.compare(subdomain, cname, responses) + + def run(self): + start = time.time() + logger.log('INFOR', f'正在检查子域接管风险') + self.format = utils.check_format(self.format) + self.dpath = utils.check_dpath(self.dpath) + self.subdomains = utils.get_domains(self.target) + if self.subdomains: + self.fingerprints = get_fingerprint() + self.results.headers = ['subdomain', 'cname'] + for domain in self.subdomains: + self.domainq.put(domain) + threads = [] + for _ in range(self.thread): + thread = Thread(target=self.check, daemon=True) + thread.start() + threads.append(thread) + for thread in threads: + thread.join() + self.save() + else: + logger.log('FATAL', f'获取域名失败') + end = time.time() + elapsed = round(end - start, 1) + logger.log('INFOR', f'{self.source}模块耗时{elapsed}秒' + f'发现{len(self.results)}个子域存在接管风险') -# 主函数 -def main(url): - cname = cname_get(url) - if cname != None: - print('[*]CNAME获取成功，正在验证是否存在于敏感列表中...') - json_dicts = providers_read() # 接收指纹信息 - for json_dict in json_dicts: - fingerprint_lists = json_dict['response'] # 存储指纹信息 - fingercname_lists = json_dict['cname'] # 存储cname信息 - for fingercname in fingercname_lists: - # 查看cname解析值是否在指纹列表中 - if fingercname in cname: - print('[*]存在于指纹列表中，正在检测子域接管风险...') - takeover_check(url,cname,fingerprint_lists) # 检查是否有风险以及是否已经被接管 - auto_take(url) # 自动接管 - else: - pass - else: - print('[*]' + url + '不存在被接管风险') if __name__ == '__main__': - url = 'test.djmag.club' #传入目标 - main(url) + fire.Fire(Takeover) + # takeover = Takeover('./subdomains.txt') + # takeover = Takeover('www.baidu.com') + # takeover.run()