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

重构项目目录结构

Browse Source
This commit is contained in:
Jing Ling
2020-05-08 18:37:07 +08:00
parent e09c8b9dfc
commit 5d64f69348
115 changed files with 5443 additions and 5349 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/certificates/censys_api.py
+72
View File
@@ -0,0 +1,72 @@
from config import api
from common.query import Query
from config.log import logger
class CensysAPI(Query):
def __init__(self, domain):
Query.__init__(self)
self.domain = self.register(domain)
self.module = 'Certificate'
self.source = "CensysAPIQuery"
self.addr = 'https://www.censys.io/api/v1/search/certificates'
self.id = api.censys_api_id
self.secret = api.censys_api_secret
self.delay = 3.0 # Censys 接口查询速率限制 最快2.5秒查1次
def query(self):
"""
向接口查询子域并做子域匹配
"""
self.header = self.get_header()
self.proxy = self.get_proxy(self.source)
data = {
'query': f'parsed.names: {self.domain}',
'page': 1,
'fields': ['parsed.subject_dn', 'parsed.names'],
'flatten': True}
resp = self.post(self.addr, json=data, auth=(self.id, self.secret))
if not resp:
return
json = resp.json()
status = json.get('status')
if status != 'ok':
logger.log('ALERT', status)
return
subdomains = self.match(self.domain, str(json))
self.subdomains = self.subdomains.union(subdomains)
pages = json.get('metadata').get('pages')
for page in range(2, pages + 1):
data['page'] = page
resp = self.post(self.addr, json=data, auth=(self.id, self.secret))
if not resp:
return
subdomains = self.match(self.domain, str(resp.json()))
self.subdomains = self.subdomains.union(subdomains)
def run(self):
"""
类执行入口
"""
if not self.check(self.id, self.secret):
return
self.begin()
self.query()
self.finish()
self.save_json()
self.gen_result()
self.save_db()
def do(domain): # 统一入口名字 方便多线程调用
"""
类统一调用入口
:param str domain: 域名
"""
query = CensysAPI(domain)
query.run()
if __name__ == '__main__':
do('example.com')
modules/certificates/certspotter.py
+53
View File
@@ -0,0 +1,53 @@
from common import utils
from common.query import Query
class CertSpotter(Query):
def __init__(self, domain):
Query.__init__(self)
self.domain = domain
self.module = 'Certificate'
self.source = 'CertSpotterQuery'
self.addr = 'https://api.certspotter.com/v1/issuances'
def query(self):
"""
向接口查询子域并做子域匹配
"""
self.header = self.get_header()
self.proxy = self.get_proxy(self.source)
params = {'domain': self.domain,
'include_subdomains': 'true',
'expand': 'dns_names'}
resp = self.get(self.addr, params)
if not resp:
return
subdomains = utils.match_subdomain(self.domain, str(resp.json()))
# 合并搜索子域名搜索结果
self.subdomains = self.subdomains.union(subdomains)
def run(self):
"""
类执行入口
"""
self.begin()
self.query()
self.finish()
self.save_json()
self.gen_result()
self.save_db()
def do(domain): # 统一入口名字 方便多线程调用
"""
类统一调用入口
:param str domain: 域名
"""
query = CertSpotter(domain)
query.run()
if __name__ == '__main__':
do('example.com')
modules/certificates/crtsh.py
+49
View File
@@ -0,0 +1,49 @@
from common import utils
from common.query import Query
class Crtsh(Query):
def __init__(self, domain):
Query.__init__(self)
self.domain = self.register(domain)
self.module = 'Certificate'
self.source = 'CrtshQuery'
self.addr = 'https://crt.sh/'
def query(self):
"""
向接口查询子域并做子域匹配
"""
self.header = self.get_header()
self.proxy = self.get_proxy(self.source)
params = {'q': f'%.{self.domain}', 'output': 'json'}
resp = self.get(self.addr, params)
if not resp:
return
subdomains = utils.match_subdomain(self.domain, str(resp.json()))
self.subdomains = self.subdomains.union(subdomains)
def run(self):
"""
类执行入口
"""
self.begin()
self.query()
self.finish()
self.save_json()
self.gen_result()
self.save_db()
def do(domain): # 统一入口名字 方便多线程调用
"""
类统一调用入口
:param str domain: 域名
"""
query = Crtsh(domain)
query.run()
if __name__ == '__main__':
do('example.com')
modules/certificates/entrust.py
+51
View File
@@ -0,0 +1,51 @@
from common import utils
from common.query import Query
class Entrust(Query):
def __init__(self, domain):
Query.__init__(self)
self.domain = self.register(domain)
self.module = 'Certificate'
self.source = 'EntrustQuery'
self.addr = 'https://ctsearch.entrust.com/api/v1/certificates'
def query(self):
"""
向接口查询子域并做子域匹配
"""
self.header = self.get_header()
self.proxy = self.get_proxy(self.source)
params = {'fields': 'subjectDN',
'domain': self.domain,
'includeExpired': 'true'}
resp = self.get(self.addr, params)
if not resp:
return
subdomains = utils.match_subdomain(self.domain, str(resp.json()))
self.subdomains = self.subdomains.union(subdomains)
def run(self):
"""
类执行入口
"""
self.begin()
self.query()
self.finish()
self.save_json()
self.gen_result()
self.save_db()
def do(domain): # 统一入口名字 方便多线程调用
"""
类统一调用入口
:param str domain: 域名
"""
query = Entrust(domain)
query.run()
if __name__ == '__main__':
do('example.com')
modules/certificates/google.py
+53
View File
@@ -0,0 +1,53 @@
from common import utils
from common.query import Query
class Google(Query):
def __init__(self, domain):
Query.__init__(self)
self.domain = self.register(domain)
self.module = 'Certificate'
self.source = 'GoogleQuery'
self.addr = 'https://transparencyreport.google.com/' \
'transparencyreport/api/v3/httpsreport/ct/certsearch'
def query(self):
"""
向接口查询子域并做子域匹配
"""
self.header = self.get_header()
self.proxy = self.get_proxy(self.source)
params = {'include_expired': 'true',
'include_subdomains': 'true',
'domain': self.domain}
resp = self.get(self.addr, params)
if not resp:
return
subdomains = utils.match_subdomain(self.domain, resp.text)
# 合并搜索子域名搜索结果
self.subdomains = self.subdomains.union(subdomains)
def run(self):
"""
类执行入口
"""
self.begin()
self.query()
self.finish()
self.save_json()
self.gen_result()
self.save_db()
def do(domain): # 统一入口名字 方便多线程调用
"""
类统一调用入口
:param str domain: 域名
"""
query = Google(domain)
query.run()
if __name__ == '__main__':
do('example.com')
modules/certificates/spyse_api.py
+65
View File
@@ -0,0 +1,65 @@
from config import api
from common import utils
from common.query import Query
class SpyseAPI(Query):
def __init__(self, domain):
Query.__init__(self)
self.domain = domain
self.module = 'Certificate'
self.source = 'CertDBAPIQuery'
self.addr = 'https://api.spyse.com/v1/subdomains'
self.token = api.spyse_api_token
def query(self):
"""
向接口查询子域并做子域匹配
"""
page_num = 1
while True:
self.header = self.get_header()
self.proxy = self.get_proxy(self.source)
params = {'domain': self.domain,
'api_token': self.token,
'page': page_num}
resp = self.get(self.addr, params)
if not resp:
return
json = resp.json()
subdomains = utils.match_subdomain(self.domain, str(json))
if not subdomains: # 搜索没有发现子域名则停止搜索
break
# 合并搜索子域名搜索结果
self.subdomains = self.subdomains.union(subdomains)
page_num += 1
# 默认每次查询最多返回30条 当前条数小于30条说明已经查完
if json.get('count') < 30:
break
def run(self):
"""
类执行入口
"""
if not self.check(self.token):
return
self.begin()
self.query()
self.finish()
self.save_json()
self.gen_result()
self.save_db()
def do(domain): # 统一入口名字 方便多线程调用
"""
类统一调用入口
:param str domain: 域名
"""
query = SpyseAPI(domain)
query.run()
if __name__ == '__main__':
do('example.com')