river_tao/OneForAll-mirror
1
0
Fork 0
mirror of https://github.com/shmilylty/OneForAll.git synced 2026-10-25 00:56:14 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/master'

Browse Source
This commit is contained in:
Jing Ling
2021-11-08 01:32:29 +08:00
parent 94d0c13134 a5dfba2554
commit d811d754f0
55 changed files with 1492 additions and 1485 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/altdns.py
+27 -12
View File
@@ -7,6 +7,7 @@ import itertools
from config import settings
from modules import wildcard
from common import utils
from common import resolve
from common import request
@@ -75,14 +76,16 @@ class Altdns(Module):
# test1.example.com -> test2.example.com, test3.example.com, ...
# test01.example.com -> test02.example.com, test03.example.com, ...
count = 0
digits = re.findall(r'\d{1,3}', subname)
for d in digits:
for m in range(self.num_count):
replacement = str(int(d) + 1 + m).zfill(len(d))
tmp_domain = subname.replace(d, replacement)
new_domain = f'{tmp_domain}.{self.domain}'
self.new_subdomains.add(new_domain)
count += 1
logger.log('DEBUG', f'The increase_num generated {count} subdomains')
def decrease_num(self, subname):
"""
@@ -94,8 +97,8 @@ class Altdns(Module):
# test4.example.com -> test3.example.com, test2.example.com, ...
# test04.example.com -> test03.example.com, test02.example.com, ...
count = 0
digits = re.findall(r'\d{1,3}', subname)
for d in digits:
for m in range(self.num_count):
new_digit = (int(d) - 1 - m)
@@ -106,6 +109,8 @@ class Altdns(Module):
tmp_domain = subname.replace(d, replacement)
new_domain = f'{tmp_domain}.{self.domain}'
self.new_subdomains.add(new_domain)
count += 1
logger.log('DEBUG', f'The decrease_num generated {count} subdomains')
def insert_word(self, parts):
"""
@@ -118,12 +123,15 @@ class Altdns(Module):
# test.1.foo.WORD.example.com,
# ...
count = 0
for word in self.words:
for index in range(len(parts)):
tmp_parts = parts.copy()
tmp_parts.insert(index, word)
new_domain = '.'.join(tmp_parts)
self.new_subdomains.add(new_domain)
count += 1
logger.log('DEBUG', f'The insert_word generated {count} subdomains')
def add_word(self, subnames):
"""
@@ -131,6 +139,7 @@ class Altdns(Module):
append existing content with `-WORD`
"""
count = 0
for word in self.words:
for index, name in enumerate(subnames):
# Prepend with `-`
@@ -146,6 +155,8 @@ class Altdns(Module):
tmp_subnames[index] = f'{name}-{word}'
new_subname = '.'.join(tmp_subnames + [self.domain])
self.new_subdomains.add(new_subname)
count += 1
logger.log('DEBUG', f'The add_word generated {count} subdomains')
def replace_word(self, subname):
"""
@@ -158,6 +169,7 @@ class Altdns(Module):
# WORD4.1.foo.example.com,
# ..
count = 0
for word in self.words:
if word not in subname:
continue
@@ -167,16 +179,22 @@ class Altdns(Module):
new_subname = subname.replace(word, word_alt)
new_subdomain = f'{new_subname}.{self.domain}'
self.new_subdomains.add(new_subdomain)
count += 1
logger.log('DEBUG', f'The replace_word generated {count} subdomains')
def gen_new_subdomains(self):
for subdomain in self.now_subdomains:
subname, parts = split_domain(subdomain)
subnames = subname.split('.')
self.increase_num(subname)
self.decrease_num(subname)
self.replace_word(subname)
if not settings.enable_fast_alt:
if settings.altdns_increase_num:
self.increase_num(subname)
if settings.altdns_decrease_num:
self.decrease_num(subname)
if settings.altdns_replace_word:
self.replace_word(subname)
if settings.altdns_insert_word:
self.insert_word(parts)
if settings.altdns_add_word:
self.add_word(subnames)
count = len(self.new_subdomains)
logger.log('DEBUG', f'The altdns module generated {count} subdomains')
@@ -193,9 +211,6 @@ class Altdns(Module):
self.end = time.time()
self.elapse = round(self.end - self.start, 1)
self.gen_result()
temp_data = resolve.run_resolve(self.domain, self.results)
fina_data = request.run_request(self.domain, temp_data, port)
data = data + fina_data
logger.log('INFOR', f'Saving altdns results')
utils.save_db(self.domain, data, 'altdns')
return data
resolved_data = resolve.run_resolve(self.domain, self.results)
valid_data = wildcard.deal_wildcard(resolved_data) # 强制开启泛解析处理
request.run_request(self.domain, valid_data, port)
modules/certificates/censys_api.py
+1 -1
View File
@@ -9,7 +9,7 @@ class CensysAPI(Query):
self.domain = domain
self.module = 'Certificate'
self.source = "CensysAPIQuery"
self.addr = 'https://www.censys.io/api/v1/search/certificates'
self.addr = 'https://search.censys.io/api/v1/search/certificates'
self.id = settings.censys_api_id
self.secret = settings.censys_api_secret
self.delay = 3.0 # Censys 接口查询速率限制 最快2.5秒查1次
modules/check/nsec.py
+3
View File
@@ -26,6 +26,9 @@ class NSEC(Check):
self.subdomains.update(subdomains)
if subdomain == self.domain: # 当查出子域为主域 说明完成了一个循环 不再继续查询
break
if domain != self.domain: # 防止出现wwdmas.cn 000.000.wwdmas.cn 000.000.000.wwdmas.cn情况
if domain.split('.')[0] == subdomain.split('.')[0]:
break
domain = subdomain
return self.subdomains
modules/collect.py
+3 -2
View File
@@ -23,7 +23,7 @@ class Collect(object):
module_path = settings.module_dir.joinpath(module)
for path in module_path.rglob('*.py'):
import_module = f'modules.{module}.{path.stem}'
self.modules.append([import_module, path.stem])
self.modules.append(import_module)
else:
self.modules = settings.enable_partial_module
@@ -31,7 +31,8 @@ class Collect(object):
"""
Import do function
"""
for module, name in self.modules:
for module in self.modules:
name = module.split('.')[-1]
import_object = importlib.import_module(module)
func = getattr(import_object, 'run')
self.collect_funcs.append([func, name])
modules/datasets/dnsdumpster.py
+6 -6
View File
@@ -1,12 +1,12 @@
from common.query import Query
class DNSdumpster(Query):
class DNSDumpster(Query):
def __init__(self, domain):
Query.__init__(self)
self.domain = domain
self.module = 'Dataset'
self.source = "DNSdumpsterQuery"
self.source = "DNSDumpsterQuery"
self.addr = 'https://dnsdumpster.com/'
def query(self):
@@ -21,7 +21,8 @@ class DNSdumpster(Query):
return
self.cookie = resp.cookies
data = {'csrfmiddlewaretoken': self.cookie.get('csrftoken'),
'targetip': self.domain}
'targetip': self.domain,
'user':'free'}
resp = self.post(self.addr, data)
self.subdomains = self.collect_subdomains(resp)
@@ -43,10 +44,9 @@ def run(domain):
:param str domain: 域名
"""
query = DNSdumpster(domain)
query = DNSDumpster(domain)
query.run()
if __name__ == '__main__':
run('example.com')
run('mi.com')
modules/datasets/netcraft.py
+4 -22
View File
@@ -12,39 +12,21 @@ class NetCraft(Query):
self.domain = domain
self.module = 'Dataset'
self.source = 'NetCraftQuery'
self.init = 'https://searchdns.netcraft.com/'
self.addr = 'https://searchdns.netcraft.com/?restriction=site+contains'
self.addr = 'https://searchdns.netcraft.com/?restriction=site+contains&position=limited'
self.page_num = 1
self.per_page_num = 20
def bypass_verification(self):
"""
绕过NetCraft的JS验证
"""
self.header = self.get_header() # Netcraft会检查User-Agent
resp = self.get(self.init)
if not resp:
return False
self.cookie = resp.cookies
cookie_value = self.cookie['netcraft_js_verification_challenge']
cookie_encode = parse.unquote(cookie_value).encode('utf-8')
verify_taken = hashlib.sha1(cookie_encode).hexdigest()
self.cookie['netcraft_js_verification_response'] = verify_taken
return True
def query(self):
"""
向接口查询子域并做子域匹配
"""
if not self.bypass_verification():
return
self.header = self.get_header() # NetCraft会检查User-Agent
self.proxy = self.get_proxy(self.source)
last = ''
while True:
time.sleep(self.delay)
self.header = self.get_header()
self.proxy = self.get_proxy(self.source)
params = {'restriction': 'site ends with',
'host': '.' + self.domain,
params = {'host': '*.' + self.domain,
'from': self.page_num}
resp = self.get(self.addr + last, params)
subdomains = self.match_subdomains(resp)
modules/datasets/phonebook.py
-61
View File
@@ -1,61 +0,0 @@
from config.log import logger
from common.query import Query
class PhoneBook(Query):
def __init__(self, domain):
Query.__init__(self)
self.domain = domain
self.module = 'Dataset'
self.source = 'PhoneBookQuery'
def query(self):
"""
向接口查询子域并做子域匹配
"""
self.header = self.get_header()
self.proxy = self.get_proxy(self.source)
self.header.update({'Referer': 'https://phonebook.cz/',
'Origin': 'https://phonebook.cz'})
addr = 'https://public.intelx.io/phonebook/search'
key = 'd7d1ed06-f0c5-49d4-a9ca-a167e6d2ffab'
url = f'{addr}?k={key}'
data = {"term": self.domain, "maxresults": 10000,
"media": 0, "target": 1,
"terminate": [], "timeout": 20}
resp = self.post(url, json=data)
if not resp:
return
json = resp.json()
ids = json.get('id')
if not ids:
logger.log('ALERT', f'Get PhoneBook id fail')
return
url = f'{addr}/result?k={key}&id={ids}&limit=10000'
resp = self.get(url)
self.subdomains = self.collect_subdomains(resp)
def run(self):
"""
类执行入口
"""
self.begin()
self.query()
self.finish()
self.save_json()
self.gen_result()
self.save_db()
def run(domain):
"""
类统一调用入口
:param str domain: 域名
"""
query = PhoneBook(domain)
query.run()
if __name__ == '__main__':
run('freebuf.com')
modules/datasets/robtex.py
+1 -1
View File
@@ -17,7 +17,7 @@ class Robtex(Query):
"""
self.header = self.get_header()
self.proxy = self.get_proxy(self.source)
base_addr = 'https://freeapi.robtex.com/pdns/'
base_addr = 'https://freeapi.robtex.com/pdns'
url = f'{base_addr}/forward/{self.domain}'
resp = self.get(url)
if not resp:
modules/datasets/wzpc.py
-68
View File
@@ -1,68 +0,0 @@
import time
from config.log import logger
from common.query import Query
class WZPCQuery(Query):
def __init__(self, domain):
Query.__init__(self)
self.domain = domain
self.module = 'Dataset'
self.source = 'WZPCQuery'
def query(self):
"""
向接口查询子域并做子域匹配
"""
base_addr = 'http://114.55.181.28/check_web/' \
'databaseInfo_mainSearch.action'
page_num = 1
while True:
time.sleep(self.delay)
self.header = self.get_header()
self.proxy = self.get_proxy(self.source)
params = {'isSearch': 'true', 'searchType': 'url',
'term': self.domain, 'pageNo': page_num}
try:
resp = self.get(base_addr, params)
except Exception as e:
logger.log('ERROR', e.args)
break
if not resp:
break
subdomains = self.match_subdomains(resp.text)
if not subdomains: # 没有发现子域名则停止查询
break
self.subdomains.update(subdomains)
if not subdomains:
break
if page_num > 10:
break
page_num += 1
def run(self):
"""
类执行入口
"""
self.begin()
self.query()
self.finish()
self.save_json()
self.gen_result()
self.save_db()
def run(domain):
"""
类统一调用入口
:param str domain: 域名
"""
query = WZPCQuery(domain)
query.run()
if __name__ == '__main__':
run('sc.gov.cn')
run('bkzy.org')
modules/enrich.py
+72
View File
@@ -0,0 +1,72 @@
from modules import iscdn
from common import utils
from common.database import Database
from common.ipasn import IPAsnInfo
from common.ipreg import IpRegData
def get_ips(info):
ip = info.get('ip')
if not ip:
return None
ips = ip.split(',')
return ips
def enrich_info(data):
ip_asn = IPAsnInfo()
ip_reg = IpRegData()
for index, info in enumerate(data):
ips = get_ips(info)
if not ips:
continue
public = list()
cidr = list()
asn = list()
org = list()
addr = list()
isp = list()
for ip in ips:
public.append(str(utils.ip_is_public(ip)))
asn_info = ip_asn.find(ip)
cidr.append(asn_info.get('cidr'))
asn.append(asn_info.get('asn'))
org.append(asn_info.get('org'))
ip_info = ip_reg.query(ip)
addr.append(ip_info.get('addr'))
isp.append(ip_info.get('isp'))
data[index]['public'] = ','.join(public)
data[index]['cidr'] = ','.join(cidr)
data[index]['asn'] = ','.join(asn)
data[index]['org'] = ','.join(org)
data[index]['addr'] = ','.join(addr)
data[index]['isp'] = ','.join(isp)
return data
class Enrich(object):
def __init__(self, domain):
self.domain = domain
def get_data(self):
db = Database()
fields = ['url', 'cname', 'ip', 'public', 'cdn', 'header',
'cidr', 'asn', 'org', 'addr', 'isp']
results = db.get_data_by_fields(self.domain, fields)
return results.as_dict()
def save_db(self, data):
db = Database()
for info in data:
url = info.pop('url')
info.pop('cname')
info.pop('ip')
info.pop('header')
db.update_data_by_url(self.domain, info, url)
db.close()
def run(self):
data = self.get_data()
data = enrich_info(data)
data = iscdn.do_check(data)
self.save_db(data)
modules/finder.py
+15 -10
View File
@@ -7,6 +7,7 @@ from common import utils
from common import resolve
from common import request
from common.module import Module
from common.database import Database
from config import settings
from config.log import logger
@@ -25,16 +26,11 @@ class Finder(Module):
new_subdomains = found_subdomains - existing_subdomains
if not len(new_subdomains):
self.finish() # 未发现新的子域就直接返回
return data
self.subdomains = new_subdomains
self.finish()
self.gen_result()
temp_data = resolve.run_resolve(domain, self.results)
fina_data = request.run_request(domain, temp_data, port)
data = data + fina_data
logger.log('INFOR', f'Saving finder results')
utils.save_db(domain, data, 'finder')
return data
resolved_data = resolve.run_resolve(domain, self.results)
request.run_request(domain, resolved_data, port)
file_path = settings.data_storage_dir.joinpath('common_js_library.json')
@@ -177,22 +173,31 @@ def find_js_urls(domain, req_url, rsp_html):
return js_urls
def convert_to_dict(url_list):
url_dict = []
for url in url_list:
url_dict.append({'url': url})
return url_dict
def find_subdomains(domain, data):
subdomains = set()
js_urls = set()
db = Database()
for infos in data:
jump_history = infos.get('history')
req_url = infos.get('url')
subdomains.update(find_in_history(domain, req_url, jump_history))
rsp_html = infos.get('response')
rsp_html = db.get_resp_by_url(domain, req_url)
if not rsp_html:
logger.log('DEBUG', f'an abnormal response occurred in the request {req_url}')
continue
subdomains.update(find_in_resp(domain, req_url, rsp_html))
js_urls.update(find_js_urls(domain, req_url, rsp_html))
resp_data = request.bulk_request(js_urls)
for _, resp in resp_data:
req_data = convert_to_dict(js_urls)
resp_data = request.bulk_request(domain, req_data, ret=True)
while not resp_data.empty():
_, resp = resp_data.get()
if not isinstance(resp, Response):
continue
text = utils.decode_resp_text(resp)
modules/intelligence/threatminer.py
+2 -3
View File
@@ -7,7 +7,7 @@ class ThreatMiner(Query):
self.domain = domain
self.module = 'Intelligence'
self.source = 'ThreatMinerQuery'
self.addr = 'https://www.threatminer.org/getData.php'
self.addr = 'https://api.threatminer.org/v2/domain.php'
def query(self):
"""
@@ -15,8 +15,7 @@ class ThreatMiner(Query):
"""
self.header = self.get_header()
self.proxy = self.get_proxy(self.source)
params = {'e': 'subdomains_container',
'q': self.domain, 't': 0, 'rt': 10}
params = {'q': self.domain, 'rt': 5}
resp = self.get(self.addr, params)
self.subdomains = self.collect_subdomains(resp)
modules/iscdn.py
+35 -33
View File
@@ -3,6 +3,7 @@ import ipaddress
from config import settings
from common import utils
from common.database import Database
from config.log import logger
data_dir = settings.data_storage_dir
@@ -18,6 +19,8 @@ cdn_header_key = utils.load_json(data_dir.joinpath('cdn_header_keys.json'))
def check_cname_keyword(cname):
if not cname:
return False
names = cname.lower().split(',')
for name in names:
for keyword in cdn_cname_keyword.keys():
@@ -26,14 +29,22 @@ def check_cname_keyword(cname):
def check_header_key(header):
header = set(map(lambda x: x.lower(), header.keys()))
for key in cdn_header_key:
if key in header:
return True
if isinstance(header, str):
header = json.loads(header)
if isinstance(header, dict):
header = set(map(lambda x: x.lower(), header.keys()))
for key in cdn_header_key:
if key in header:
return True
else:
return False
def check_cdn_cidr(content):
ips = set(content.split(','))
def check_cdn_cidr(ips):
if isinstance(ips, str):
ips = set(ips.split(','))
else:
return False
for ip in ips:
try:
ip = ipaddress.ip_address(ip)
@@ -46,39 +57,30 @@ def check_cdn_cidr(content):
def check_cdn_asn(asn):
if str(asn) in cdn_asn_list:
return True
if isinstance(asn, str):
if asn in cdn_asn_list:
return True
return False
def check_cdn(data):
logger.log('DEBUG', f'Start cdn check module')
def do_check(data):
logger.log('DEBUG', f'Checking cdn')
for index, item in enumerate(data):
cname = item.get('cname')
if cname:
if check_cname_keyword(cname):
data[index]['cdn'] = 1
continue
if check_cname_keyword(cname):
data[index]['cdn'] = 1
continue
header = item.get('header')
if header:
header = json.loads(header)
if check_header_key(header):
data[index]['cdn'] = 1
continue
if check_header_key(header):
data[index]['cdn'] = 1
continue
ip = item.get('ip')
if ip:
if check_cdn_cidr(ip):
data[index]['cdn'] = 1
continue
if check_cdn_cidr(ip):
data[index]['cdn'] = 1
continue
asn = item.get('asn')
if asn:
asn = asn[2:] # 去除AS
if check_cdn_asn(asn):
data[index]['cdn'] = 1
continue
if check_cdn_asn(asn):
data[index]['cdn'] = 1
continue
data[index]['cdn'] = 0
return data
def save_db(name, data):
logger.log('DEBUG', f'Saving cdn check results')
utils.save_db(name, data, 'cdn')
modules/search/baidu.py
+1 -1
View File
@@ -57,7 +57,7 @@ class Baidu(Search):
self.subdomains.update(subdomains)
self.page_num += self.per_page_num
# 搜索页面没有出现下一页时停止搜索
if '&pn={next_pn}&'.format(next_pn=self.page_num) not in resp.text:
if f'&pn={self.page_num}&' not in resp.text:
break
if self.page_num >= self.limit_num: # 搜索条数限制
break
modules/search/bing.py
+1 -1
View File
@@ -19,9 +19,9 @@ class Bing(Search):
:param str domain: 域名
:param str filtered_subdomain: 过滤的子域
"""
self.page_num = 0 # 二次搜索重新置0
self.header = self.get_header()
self.proxy = self.get_proxy(self.source)
self.page_num = 0 # 二次搜索重新置0
resp = self.get(self.init)
if not resp:
return
modules/search/fofa_api.py
+3 -3
View File
@@ -21,7 +21,7 @@ class FoFa(Search):
发送搜索请求并做子域匹配
"""
self.page_num = 1
subdomain_encode = f'domain={self.domain} || cert={self.domain}'.encode('utf-8')
subdomain_encode = f'domain="{self.domain}"'.encode('utf-8')
query_data = base64.b64encode(subdomain_encode)
while True:
time.sleep(self.delay)
@@ -32,7 +32,7 @@ class FoFa(Search):
'qbase64': query_data,
'page': self.page_num,
'full': 'true',
'size': 5000}
'size': 1000}
resp = self.get(self.addr, query)
if not resp:
return
@@ -42,7 +42,7 @@ class FoFa(Search):
break
self.subdomains.update(subdomains)
size = resp_json.get('size')
if size < 5000:
if size < 1000:
break
self.page_num += 1
modules/search/gitee.py
+2 -1
View File
@@ -11,7 +11,6 @@ class Gitee(Search):
self.module = 'Search'
self.addr = 'https://search.gitee.com/'
self.domain = domain
self.header = self.get_header()
def search(self):
"""
@@ -20,6 +19,8 @@ class Gitee(Search):
page_num = 1
while True:
time.sleep(self.delay)
self.header = self.get_header()
self.proxy = self.get_proxy(self.source)
params = {'pageno': page_num, 'q': self.domain, 'type': 'code'}
try:
resp = self.get(self.addr, params=params)
modules/search/github_api.py
+12 -34
View File
@@ -1,4 +1,5 @@
import requests
import time
from config import settings
from common.search import Search
from config.log import logger
@@ -11,53 +12,30 @@ class GithubAPI(Search):
self.module = 'Search'
self.addr = 'https://api.github.com/search/code'
self.domain = domain
self.session = requests.Session()
self.session.trust_env = False
self.auth_url = 'https://api.github.com'
self.delay = 5
self.token = settings.github_api_token
def auth_github(self):
"""
github api 认证
:return: 认证失败返回False 成功返回True
"""
self.session.headers.update({'Authorization': 'token ' + self.token})
try:
resp = self.session.get(self.auth_url)
except Exception as e:
logger.log('ERROR', e.args)
return False
if resp.status_code != 200:
resp_json = resp.json()
msg = resp_json.get('message')
logger.log('ERROR', msg)
return False
return True
def search(self):
"""
向接口查询子域并做子域匹配
"""
self.session.headers = self.get_header()
self.session.proxies = self.get_proxy(self.source)
self.session.verify = self.verify
self.session.headers.update(
self.header = self.get_header()
self.proxy = self.get_proxy(self.source)
self.header.update(
{'Accept': 'application/vnd.github.v3.text-match+json'})
if not self.auth_github():
logger.log('ERROR', f'{self.source} module login failed')
return
page = 1
while True:
time.sleep(self.delay)
params = {'q': self.domain, 'per_page': 100,
'page': page, 'sort': 'indexed'}
'page': page, 'sort': 'indexed',
'access_token': self.token}
try:
resp = self.session.get(self.addr, params=params)
resp = self.get(self.addr, params=params)
except Exception as e:
logger.log('ERROR', e.args)
break
if resp.status_code != 200:
if not resp or resp.status_code != 200:
logger.log('ERROR', f'{self.source} module query failed')
break
subdomains = self.match_subdomains(resp)
@@ -103,4 +81,4 @@ def run(domain):
if __name__ == '__main__':
run('example.com')
run('freebuf.com')
modules/search/yahoo.py
+2 -1
View File
@@ -21,13 +21,14 @@ class Yahoo(Search):
:param str domain: 域名
:param str filtered_subdomain: 过滤的子域
"""
self.header = self.get_header()
self.proxy = self.get_proxy(self.source)
resp = self.get(self.init)
if not resp:
return
self.cookie = resp.cookies # 获取cookie Yahoo在搜索时需要带上cookie
while True:
time.sleep(self.delay)
self.header = self.get_header()
self.proxy = self.get_proxy(self.source)
query = 'site:.' + domain + filtered_subdomain
params = {'p': query, 'b': self.page_num, 'pz': self.per_page_num}
modules/search/yandex.py
+2 -1
View File
@@ -20,6 +20,8 @@ class Yandex(Search):
:param str domain: 域名
:param str filtered_subdomain: 过滤的子域
"""
self.header = self.get_header()
self.proxy = self.get_proxy(self.source)
self.page_num = 0 # 二次搜索重新置0
resp = self.get(self.init)
if not resp:
@@ -27,7 +29,6 @@ class Yandex(Search):
self.cookie = resp.cookies # 获取cookie
while True:
time.sleep(self.delay)
self.header = self.get_header()
self.proxy = self.get_proxy(self.source)
query = 'site:.' + domain + filtered_subdomain
params = {'text': query, 'p': self.page_num,
modules/wildcard.py
+322
View File
@@ -0,0 +1,322 @@
import secrets
import tenacity
from dns.exception import Timeout
from dns.resolver import NXDOMAIN, YXDOMAIN, NoAnswer, NoNameservers
from common import utils
from config import settings
from common import similarity
from config.log import logger
def gen_random_subdomains(domain, count):
"""
生成指定数量的随机子域域名列表
:param domain: 主域
:param count: 数量
"""
subdomains = set()
if count < 1:
return subdomains
for _ in range(count):
token = secrets.token_hex(4)
subdomains.add(f'{token}.{domain}')
return subdomains
def query_a_record(subdomain, resolver):
"""
查询子域A记录
:param subdomain: 子域
:param resolver: DNS解析器
"""
try:
answer = resolver.query(subdomain, 'A')
except Exception as e:
logger.log('DEBUG', f'Query {subdomain} wildcard dns record error')
logger.log('DEBUG', e.args)
return False
if answer.rrset is None:
return False
ttl = answer.ttl
name = answer.name
ips = {item.address for item in answer}
logger.log('ALERT', f'{subdomain} resolve to: {name} '
f'IP: {ips} TTL: {ttl}')
return True
def all_resolve_success(subdomains):
"""
判断是否所有子域都解析成功
:param subdomains: 子域列表
"""
resolver = utils.dns_resolver()
resolver.cache = None # 不使用DNS缓存
status = set()
for subdomain in subdomains:
status.add(query_a_record(subdomain, resolver))
return all(status)
def all_request_success(subdomains):
"""
判断是否所有子域都请求成功
:param subdomains: 子域列表
"""
result = list()
for subdomain in subdomains:
url = f'http://{subdomain}'
resp = utils.get_url_resp(url)
if resp:
logger.log('ALERT', f'Request: {url} Status: {resp.status_code} '
f'Size: {len(resp.content)}')
result.append(resp.text)
else:
result.append(resp)
return all(result), result
def any_similar_html(resp_list):
"""
判断是否有一组HTML页面结构相似
:param resp_list: 响应HTML页面
"""
html_doc1, html_doc2, html_doc3 = resp_list
if similarity.is_similar(html_doc1, html_doc2):
return True
if similarity.is_similar(html_doc1, html_doc3):
return True
if similarity.is_similar(html_doc2, html_doc3):
return True
return False
def to_detect_wildcard(domain):
"""
Detect use wildcard dns record or not
:param str domain: domain
:return bool use wildcard dns record or not
"""
logger.log('INFOR', f'Detecting {domain} use wildcard dns record or not')
random_subdomains = gen_random_subdomains(domain, 3)
if not all_resolve_success(random_subdomains):
return False
is_all_success, all_request_resp = all_request_success(random_subdomains)
if not is_all_success:
return True
return any_similar_html(all_request_resp)
def detect_wildcard(domain):
is_enable = to_detect_wildcard(domain)
if is_enable:
logger.log('ALERT', f'The domain {domain} enables wildcard')
else:
logger.log('ALERT', f'The domain {domain} disables wildcard')
return is_enable
@tenacity.retry(stop=tenacity.stop_after_attempt(2))
def get_wildcard_record(domain, resolver):
logger.log('INFOR', f"Query {domain} 's wildcard dns record "
f"in authoritative name server")
try:
answer = resolver.query(domain, 'A')
# 如果查询随机域名A记录时抛出Timeout异常则重新查询
except Timeout as e:
logger.log('ALERT', f'Query timeout, retrying')
logger.log('DEBUG', e.args)
raise e
except (NXDOMAIN, YXDOMAIN, NoAnswer, NoNameservers) as e:
logger.log('DEBUG', e.args)
logger.log('DEBUG', f'{domain} dont have A record on authoritative name server')
return None, None
except Exception as e:
logger.log('ERROR', e.args)
logger.log('ERROR', f'Query {domain} wildcard dns record in '
f'authoritative name server error')
exit(1)
else:
if answer.rrset is None:
logger.log('DEBUG', f'No record of query result')
return None, None
name = answer.name
ip = {item.address for item in answer}
ttl = answer.ttl
logger.log('INFOR', f'{domain} results on authoritative name server: {name} '
f'IP: {ip} TTL: {ttl}')
return ip, ttl
def collect_wildcard_record(domain, authoritative_ns):
logger.log('INFOR', f'Collecting wildcard dns record for {domain}')
if not authoritative_ns:
return list(), int()
resolver = utils.dns_resolver()
resolver.nameservers = authoritative_ns # 使用权威名称服务器
resolver.rotate = True # 随机使用NS
resolver.cache = None # 不使用DNS缓存
ips = set()
ttl = int()
ttls_check = list()
ips_stat = dict()
ips_check = list()
while True:
token = secrets.token_hex(4)
random_subdomain = f'{token}.{domain}'
try:
ip, ttl = get_wildcard_record(random_subdomain, resolver)
except Exception as e:
logger.log('DEBUG', e.args)
logger.log('ALERT', f'Multiple query errors,'
f'try to query a new random subdomain')
continue
# 每5次查询检查结果列表 如果都没结果则结束查询
ips_check.append(ip)
ttls_check.append(ttl)
if len(ips_check) == 5:
if not any(ips_check):
logger.log('ALERT', 'The query ends because there are '
'no results for 5 consecutive queries.')
break
ips_check = list()
if len(ttls_check) == 5 and len(set(ttls_check)) == 5:
logger.log('ALERT', 'The query ends because there are '
'5 different TTL results for 5 consecutive queries.')
ips, ttl = set(), int()
break
if ip is None:
continue
ips.update(ip)
# 统计每个泛解析IP出现次数
for addr in ip:
count = ips_stat.setdefault(addr, 0)
ips_stat[addr] = count + 1
# 筛选出出现次数2次以上的IP地址
addrs = list()
for addr, times in ips_stat.items():
if times >= 2:
addrs.append(addr)
# 大部分的IP地址出现次数大于2次停止收集泛解析IP记录
if len(addrs) / len(ips) >= 0.8:
break
logger.log('DEBUG', f'Collected the wildcard dns record of {domain}\n{ips}\n{ttl}')
return ips, ttl
def check_by_compare(ip, ttl, wc_ips, wc_ttl):
"""
Use TTL comparison to detect wildcard dns record
:param set ip: A record IP address set
:param int ttl: A record TTL value
:param set wc_ips: wildcard dns record IP address set
:param int wc_ttl: wildcard dns record TTL value
:return bool: result
"""
# Reference：http://sh3ll.me/archives/201704041222.txt
if ip not in wc_ips:
return False # 子域IP不在泛解析IP集合则不是泛解析
if ttl != wc_ttl and ttl % 60 == 0 and wc_ttl % 60 == 0:
return False
return True
def check_ip_times(times):
"""
Use IP address times to determine wildcard or not
:param times: IP address times
:return bool: result
"""
if times > settings.ip_appear_maximum:
return True
return False
def check_cname_times(times):
"""
Use cname times to determine wildcard or not
:param times: cname times
:return bool: result
"""
if times > settings.cname_appear_maximum:
return True
return False
def is_valid_subdomain(ip=None, ip_num=None, cname=None, cname_num=None,
ttl=None, wc_ttl=None, wc_ips=None):
ip_blacklist = settings.brute_ip_blacklist
cname_blacklist = settings.brute_cname_blacklist
if cname and cname in cname_blacklist:
return 0, 'cname blacklist' # 有些泛解析会统一解析到一个cname上
if ip and ip in ip_blacklist: # 解析ip在黑名单ip则为非法子域
return 0, 'IP blacklist'
if all([wc_ips, wc_ttl]): # 有泛解析记录才进行对比
if check_by_compare(ip, ttl, wc_ips, wc_ttl):
return 0, 'IP wildcard'
if ip_num and check_ip_times(ip_num):
return 0, 'IP exceeded'
if cname_num and check_cname_times(cname_num):
return 0, 'cname exceeded'
return 1, 'OK'
def stat_times(data):
times = dict()
for info in data:
ip_str = info.get('ip')
if isinstance(ip_str, str):
ips = ip_str.split(',')
for ip in ips:
value_one = times.setdefault(ip, 0)
times[ip] = value_one + 1
cname_str = info.get('cname')
if isinstance(cname_str, str):
cnames = cname_str.split(',')
for cname in cnames:
value_two = times.setdefault(cname, 0)
times[cname] = value_two + 1
return times
def check_valid_subdomain(appear_times, info):
ip_str = info.get('ip')
if ip_str:
ips = ip_str.split(',')
for ip in ips:
ip_num = appear_times.get(ip)
isvalid, reason = is_valid_subdomain(ip=ip, ip_num=ip_num)
if not isvalid:
return False, reason
cname_str = info.get('cname')
if cname_str:
cnames = cname_str.split(',')
for cname in cnames:
cname_num = appear_times.get(cname)
isvalid, reason = is_valid_subdomain(cname=cname, cname_num=cname_num)
if not isvalid:
return False, reason
return True, 'OK'
def deal_wildcard(data):
new_data = list()
appear_times = stat_times(data)
for info in data:
subdomain = info.get('subdomain')
isvalid, reason = check_valid_subdomain(appear_times, info)
logger.log('DEBUG', f'{subdomain} is {isvalid} subdomain reason because {reason}')
if isvalid:
new_data.append(info)
return new_data