modules/wildcard.py
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
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')
except Timeout as e:
logger.log('ALERT', f'Query timeout, retrying')
logger.log('DEBUG', e.args)
return None, None
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()
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')
# 查询出错退出循环
break
# 每5次连续查询后检查结果列表
ips_check.append(ip)
# 如果出现50个以上的泛解析则结束查询
if len(ips) >= 50:
break
# 如果连续5次查询都没结果则结束查询
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 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.7:
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