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

重构网络判断和地址判断

Browse Source
This commit is contained in:
Jing Ling
2020-11-09 18:43:16 +08:00
parent 0f2b365a25
commit fec72d2082
11 changed files with 84 additions and 164 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
-2
View File
@@ -97,8 +97,6 @@ docker run -it --rm -v ~/results:/OneForAll/results -v ~/.config:/OneForAll/conf
<details>
<summary><b>✨使用演示</b></summary>
如果你的主机不在中国，请把 [setting](https://github.com/shmilylty/OneForAll/blob/master/config/setting.py#L46) 中`brute_nameservers_path`选项的`cn_nameservers.txt`修改为`nameservers.txt`。
如果你是通过pip3安装的依赖则使用以下命令运行示例：
```bash
python3 oneforall.py --target example.com run
brute.py
+4 -16
View File
@@ -349,18 +349,6 @@ def collect_wildcard_record(domain, authoritative_ns):
return ips, ttl
def get_nameservers_path(enable_wildcard, ns_ip_list):
path = settings.brute_nameservers_path
if not enable_wildcard:
return path
if not ns_ip_list:
return path
path = settings.authoritative_dns_path
ns_data = '\n'.join(ns_ip_list)
utils.save_data(path, ns_data)
return path
def check_dict():
if not settings.enable_check_dict:
return
@@ -607,8 +595,8 @@ class Brute(Module):
self.domain = str() # 当前正在进行爆破的域名
self.ips_times = dict() # IP集合出现次数
self.enable_wildcard = False # 当前域名是否使用泛解析
self.check_env = True
self.quite = False
self.in_china = None
def gen_brute_dict(self, domain):
logger.log('INFOR', f'Generating dictionary for {domain}')
@@ -684,7 +672,7 @@ class Brute(Module):
if self.enable_wildcard:
wildcard_ips, wildcard_ttl = collect_wildcard_record(domain,
ns_ip_list)
ns_path = get_nameservers_path(self.enable_wildcard, ns_ip_list)
ns_path = utils.get_ns_path(self.in_china, self.enable_wildcard, ns_ip_list)
dict_set = self.gen_brute_dict(domain)
@@ -727,8 +715,8 @@ class Brute(Module):
def run(self):
logger.log('INFOR', f'Start running {self.source} module')
if self.check_env:
utils.check_env()
if self.in_china is None:
_, self.in_china = utils.get_net_env()
self.domains = utils.get_domains(self.target, self.targets)
for self.domain in self.domains:
self.results = list() # 置空
common/resolve.py
+1 -2
View File
@@ -155,8 +155,7 @@ def run_resolve(domain, data):
output_name = f'resolved_result_{domain}_{timestring}.json'
output_path = temp_dir.joinpath(output_name)
log_path = result_dir.joinpath('massdns.log')
ns_path = settings.brute_nameservers_path
ns_path = utils.get_ns_path()
logger.log('INFOR', f'Running massdns to resolve subdomains')
utils.call_massdns(massdns_path, save_path, ns_path,
common/utils.py
+36 -98
View File
@@ -518,15 +518,16 @@ def delete_file(*paths):
@tenacity.retry(stop=tenacity.stop_after_attempt(3))
def check_net():
logger.log('INFOR', 'Checking Internet environment')
urls = ['http://www.baidu.com', 'http://www.bing.com',
'http://www.apple.com', 'http://www.microsoft.com']
urls = ['http://ipinfo.io/json', 'http://ipconfig.io/json']
url = random.choice(urls)
logger.log('INFOR', f'Trying to access {url}')
header = {'User_Agent': 'curl'}
timeout = settings.request_timeout_second
verify = settings.request_ssl_verify
logger.log('DEBUG', f'Trying to access {url}')
session = requests.Session()
session.trust_env = False
try:
rsp = session.get(url, proxies=get_proxy())
rsp = session.get(url, headers=header, timeout=timeout, verify=verify)
except Exception as e:
logger.log('ERROR', e.args)
logger.log('ALERT', 'Can not access Internet, retrying')
@@ -536,10 +537,16 @@ def check_net():
f'{rsp.status_code} {rsp.reason}')
logger.log('ALERT', 'Can not access Internet normally, retrying')
raise tenacity.TryAgain
logger.log('INFOR', 'Access to Internet OK')
logger.log('DEBUG', 'Access to Internet OK')
country = rsp.json().get('country').lower()
if country in ['cn', 'china']:
logger.log('DEBUG', f'The host in china')
return True, True
else:
return True, False
def check_pre():
def check_dep():
logger.log('INFOR', 'Checking dependent environment')
implementation = platform.python_implementation()
version = platform.python_version()
@@ -551,15 +558,15 @@ def check_pre():
exit(1)
def check_env():
logger.log('INFOR', 'Checking the environment')
def get_net_env():
logger.log('INFOR', 'Checking network environment')
try:
check_net()
result = check_net()
except Exception as e:
logger.log('DEBUG', e.args)
logger.log('FATAL', 'Can not access Internet')
exit(1)
check_pre()
logger.log('ALERT', 'Can not access Internet')
return False, None
return result
def check_version(local):
@@ -577,7 +584,7 @@ def check_version(local):
resp_json = resp.json()
latest = resp_json['tag_name']
except Exception as e:
logger.log('ERROR', 'An error occurred while checking the latest version')
logger.log('ALERT', 'An error occurred while checking the latest version')
logger.log('DEBUG', e.args)
return
if latest > local:
@@ -743,90 +750,6 @@ def sort_by_subdomain(data):
return sorted(data, key=lambda item: item.get('subdomain'))
def ping(host, path):
param = '-n' if platform.system().lower() == 'windows' else '-c'
command = ['ping', param, '5', host]
with open(path, "w") as f:
return subprocess.call(command, stdout=f, stderr=f)
def ping_avg_time(nameserver):
check_dir(settings.temp_save_dir)
temp_path = settings.temp_save_dir.joinpath('ping')
ping(nameserver, path=temp_path)
with open(temp_path, 'r') as f:
text = f.read()
if '100.0% packet loss' in text or '100% packet loss' in text or '100% 丢失' in text:
logger.log('ALERT', f'100.0% packet loss, ping {nameserver} failed.')
return None
elif platform.system() in ('Darwin', 'Linux'):
try:
avg_time = re.findall(r'(?:min/avg/max/.+ )(?:\d+\.\d+)/(\d+\.\d+)/', text)[0]
logger.log('INFOR', f'ping {nameserver} average time {avg_time} ms.')
except IndexError:
return None
return avg_time
elif platform.system() == 'Windows':
try:
avg_time = re.findall(r'(?:Average|平均).+(\d.?)ms', text)[0]
logger.log('INFOR', f'ping {nameserver} average time {avg_time} ms.')
except IndexError:
return None
return avg_time
else:
logger.log('ALERT', f'{text}')
return None
def auto_select_nameserver():
logger.log('INFOR', f'Ping test start, to select nameservers.')
avg_time1 = ping_avg_time('114.114.114.114')
avg_time2 = ping_avg_time('8.8.8.8')
if avg_time1 and avg_time2:
if avg_time1 < avg_time2:
change_nameservers_file('cn')
logger.log('INFOR', f'Ping test finished, use cn nameservers.')
else:
change_nameservers_file('common')
logger.log('INFOR', f'Ping test finished, use common nameservers.')
elif avg_time1 and not avg_time2:
change_nameservers_file('cn')
logger.log('INFOR', f'Ping test finished, use cn nameservers.')
elif not avg_time1 and avg_time1:
change_nameservers_file('common')
logger.log('INFOR', f'Ping test finished, use common nameservers.')
elif not avg_time1 and not avg_time1:
change_nameservers_file('default')
logger.log('INFOR', f'Ping test finished, use default nameservers.')
return
def change_nameservers_file(option):
text = ''
if option == 'cn':
with open(settings.data_storage_dir.joinpath('cn_nameservers.txt'), 'r') as f:
text = f.read()
elif option == 'common':
with open(settings.data_storage_dir.joinpath('common_nameservers.txt'), 'r') as f:
text = f.read()
elif option == 'default':
for n in default_nameserver():
text = '\n'.join(n)
with open(settings.data_storage_dir.joinpath('nameservers.txt'), 'w') as f:
f.write(text)
return
def default_nameserver():
try:
resolver = dns.resolver.Resolver()
return resolver.nameservers
except dns.resolver.NoResolverConfiguration:
logger.log('ERROR', 'Resolver configuration could not be read '
'or specified no nameservers.')
exit(1)
def looks_like_ip(maybe_ip):
"""Does the given str look like an IP address?"""
if not maybe_ip[0].isdigit():
@@ -860,3 +783,18 @@ def clear_data(domain):
db = Database()
db.drop_table(domain)
db.close()
def get_ns_path(in_china=None, enable_wildcard=None, ns_ip_list=None):
data_dir = settings.data_storage_dir
path = data_dir.joinpath('nameservers.txt')
if in_china:
path = data_dir.joinpath('nameservers_cn.txt')
if not enable_wildcard:
return path
if not ns_ip_list:
return path
path = settings.authoritative_dns_path
ns_data = '\n'.join(ns_ip_list)
save_data(path, ns_data)
return path
config/default.py
-3
View File
@@ -58,9 +58,6 @@ brute_socket_num = 1 # 爆破时每个进程下的socket数量
brute_resolve_num = 15 # 解析失败时尝试换名称服务器重查次数
# 爆破所使用的字典路径 默认data/subdomains.txt
brute_wordlist_path = data_storage_dir.joinpath('subnames.txt')
# 爆破所使用的字典路径 默认data/cn_nameservers.txt
# 如果你不在中国请改为nameservers.txt
brute_nameservers_path = data_storage_dir.joinpath('cn_nameservers.txt')
# 域名的权威DNS名称服务器的保存路径 当域名开启了泛解析时会使用该名称服务器来进行A记录查询
authoritative_dns_path = data_storage_dir.joinpath('authoritative_dns.txt')
enable_recursive_brute = False # 是否使用递归爆破(默认False)
config/setting.py
-4
View File
@@ -38,10 +38,6 @@ enable_partial_module = [] # 启用部分收集模块 必须禁用enable_all_mo
brute_concurrent_num = 2000 # 爆破时并发查询数量(默认2000，最大推荐10000)
# 爆破所使用的字典路径 默认data/subdomains.txt
brute_wordlist_path = data_storage_dir.joinpath('subnames.txt')
# 爆破所使用的DNS服务器路径 默认data/cn_nameservers.txt 如果你不在中国请改为nameservers.txt
# DNS resolve server file path default data/nameservers.txt
# If your computer's location are not in China, change `cn_nameservers.txt` to `nameservers.txt` plz.
brute_nameservers_path = data_storage_dir.joinpath('cn_nameservers.txt')
# 域名的权威DNS名称服务器的保存路径 当域名开启了泛解析时会使用该名称服务器来进行A记录查询
authoritative_dns_path = data_storage_dir.joinpath('authoritative_dns.txt')
enable_recursive_brute = False # 是否使用递归爆破(默认False)
data/common_nameservers.txt
-30
View File
@@ -1,30 +0,0 @@
8.8.8.8
8.8.4.4
9.9.9.9
9.9.9.10
149.112.112.112
4.2.2.1
4.2.2.2
4.2.2.3
4.2.2.4
4.2.2.5
4.2.2.6
1.1.1.1
1.0.0.1
1.0.0.2
1.0.0.3
1.0.0.19
208.67.222.222
208.67.220.220
8.26.56.26
8.20.247.20
84.200.69.80
84.200.70.40
185.228.168.9
185.228.169.9
64.6.64.6
64.6.65.6
198.101.242.72
23.253.163.53
176.103.130.130
176.103.130.131
data/nameservers.txt
+30
View File
@@ -0,0 +1,30 @@
8.8.8.8
8.8.4.4
9.9.9.9
9.9.9.10
149.112.112.112
4.2.2.1
4.2.2.2
4.2.2.3
4.2.2.4
4.2.2.5
4.2.2.6
1.1.1.1
1.0.0.1
1.0.0.2
1.0.0.3
1.0.0.19
208.67.222.222
208.67.220.220
8.26.56.26
8.20.247.20
84.200.69.80
84.200.70.40
185.228.168.9
185.228.169.9
64.6.64.6
64.6.65.6
198.101.242.72
23.253.163.53
176.103.130.130
176.103.130.131
data/cn_nameservers.txt → data/nameservers_cn.txt
View File
docs/en-us/README.md
-2
View File
@@ -89,8 +89,6 @@ Result will be saved in `~/results`.
<details>
<summary><b>✨OneForAll usage</b></summary>
If your computer are not in China, change [setting](https://github.com/shmilylty/OneForAll/blob/master/config/setting.py#L46) `brute_nameservers_path` param `cn_nameservers.txt` to `nameservers.txt` plz.
If you are use pip3, run the following command:
```bash
oneforall.py
+13 -7
View File
@@ -97,6 +97,8 @@ class OneForAll(object):
self.domains = set() # All domains that are to be collected
self.data = list() # The subdomain results of the current domain
self.datas = list() # All subdomain results of the domain
self.in_china = None
self.access_internet = False
def config_param(self):
"""
@@ -144,8 +146,12 @@ class OneForAll(object):
:return: subdomain results
:rtype: list
"""
collect = Collect(self.domain)
collect.run()
if not self.access_internet:
logger.log('ALERT', 'Because it cannot access the Internet, '
'OneForAll will not execute the subdomain collection module!')
if self.access_internet:
collect = Collect(self.domain)
collect.run()
srv = BruteSRV(self.domain)
srv.run()
@@ -154,7 +160,7 @@ class OneForAll(object):
# Due to there will be a large number of dns resolution requests,
# may cause other network tasks to be error
brute = Brute(self.domain, word=True, export=False)
brute.check_env = False
brute.in_china = self.in_china
brute.quite = True
brute.run()
@@ -214,12 +220,12 @@ class OneForAll(object):
print(oneforall_banner)
dt = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
print(f'[*] Starting OneForAll @ {dt}\n')
utils.check_env()
utils.auto_select_nameserver()
if settings.enable_check_version:
utils.check_version(version)
logger.log('DEBUG', 'Python ' + utils.python_version())
logger.log('DEBUG', 'OneForAll ' + version)
utils.check_dep()
self.access_internet, self.in_china = utils.get_net_env()
if self.access_internet and settings.enable_check_version:
utils.check_version(version)
logger.log('INFOR', 'Start running OneForAll')
self.config_param()
self.check_param()