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

优化代码结构 删除无用队列结果集参数

Browse Source
This commit is contained in:
shmilylty
2019-08-05 18:00:01 +08:00
parent 53213304e2
commit 11642edf6e
50 changed files with 332 additions and 502 deletions
Download Patch File Download Diff File Expand all files Collapse all files
oneforall/modules/certificates/censys_api.py
+8 -11
View File
@@ -1,6 +1,6 @@
# coding=utf-8
import time
import queue
import config
from common.query import Query
from config import logger
@@ -48,34 +48,31 @@ class CensysAPI(Query):
subdomains_find = self.match(self.domain, str(resp.json()))
self.subdomains = self.subdomains.union(subdomains_find)
def run(self, rx_queue):
def run(self):
"""
类执行入口
"""
if not (self.id and self.secret):
logger.log('ERROR', f'{self.source}模块API配置错误')
logger.log('ALERT', f'不执行{self.source}模块')
if not self.check(self.id, self.secret):
return
self.begin()
self.query()
self.save_json()
self.gen_result()
self.save_db()
rx_queue.put(self.results)
self.finish()
def do(domain, rx_queue): # 统一入口名字 方便多线程调用
def do(domain): # 统一入口名字 方便多线程调用
"""
类统一调用入口
:param str domain: 域名
:param rx_queue: 结果集队列
"""
query = CensysAPI(domain)
query.run(rx_queue)
query.run()
if __name__ == '__main__':
result_queue = queue.Queue()
do('example.com', result_queue)
do('example.com')
oneforall/modules/certificates/certdb_api.py
+7 -13
View File
@@ -1,10 +1,9 @@
# coding=utf-8
import time
import queue
import config
from common.query import Query
from common import utils
from config import logger
from common.query import Query
class CertDBAPI(Query):
@@ -36,34 +35,29 @@ class CertDBAPI(Query):
if resp_json.get('count') < 30: # 默认每次查询最多返回30条 当前条数小于30条说明已经查完
break
def run(self, rx_queue):
def run(self):
"""
类执行入口
"""
if not self.token:
logger.log('ERROR', f'{self.source}模块API配置错误')
logger.log('ALERT', f'不执行{self.source}模块')
if not self.check(self.token):
return
self.begin()
self.query()
self.save_json()
self.gen_result()
self.save_db()
rx_queue.put(self.results)
self.finish()
def do(domain, rx_queue): # 统一入口名字 方便多线程调用
def do(domain): # 统一入口名字 方便多线程调用
"""
类统一调用入口
:param str domain: 域名
:param rx_queue: 结果集队列
"""
query = CertDBAPI(domain)
query.run(rx_queue)
query.run()
if __name__ == '__main__':
result_queue = queue.Queue()
do('example.com', result_queue)
do('example.com')
oneforall/modules/certificates/certspotter.py
+6 -8
View File
@@ -1,5 +1,4 @@
# coding=utf-8
import queue
import time
from common import utils
@@ -28,7 +27,7 @@ class CertSpotter(Query):
subdomains_find = utils.match_subdomain(self.domain, str(resp.json()))
self.subdomains = self.subdomains.union(subdomains_find) # 合并搜索子域名搜索结果
def run(self, rx_queue):
def run(self):
"""
类执行入口
"""
@@ -37,21 +36,20 @@ class CertSpotter(Query):
self.save_json()
self.gen_result()
self.save_db()
rx_queue.put(self.results)
self.finish()
def do(domain, rx_queue): # 统一入口名字 方便多线程调用
def do(domain): # 统一入口名字 方便多线程调用
"""
类统一调用入口
:param str domain: 域名
:param rx_queue: 结果集队列
"""
query = CertSpotter(domain)
query.run(rx_queue)
query.run()
if __name__ == '__main__':
result_queue = queue.Queue()
do('example.com', result_queue)
do('example.com')
oneforall/modules/certificates/crtsh.py
+4 -8
View File
@@ -1,5 +1,4 @@
# coding=utf-8
import queue
import time
from common import utils
@@ -28,7 +27,7 @@ class Crtsh(Query):
subdomains_find = utils.match_subdomain(self.domain, str(resp.json()))
self.subdomains = self.subdomains.union(subdomains_find)
def run(self, rx_queue):
def run(self):
"""
类执行入口
"""
@@ -37,21 +36,18 @@ class Crtsh(Query):
self.save_json()
self.gen_result()
self.save_db()
rx_queue.put(self.results)
self.finish()
def do(domain, rx_queue): # 统一入口名字 方便多线程调用
def do(domain): # 统一入口名字 方便多线程调用
"""
类统一调用入口
:param str domain: 域名
:param rx_queue: 结果集队列
"""
query = Crtsh(domain)
query.run(rx_queue)
query.run()
if __name__ == '__main__':
result_queue = queue.Queue()
do('example.com', result_queue)
do('example.com')
oneforall/modules/certificates/entrust.py
+4 -8
View File
@@ -1,5 +1,4 @@
# coding=utf-8
import queue
import time
from common import utils
@@ -28,7 +27,7 @@ class Entrust(Query):
subdomains_find = utils.match_subdomain(self.domain, str(resp.json()))
self.subdomains = self.subdomains.union(subdomains_find)
def run(self, rx_queue):
def run(self):
"""
类执行入口
"""
@@ -37,21 +36,18 @@ class Entrust(Query):
self.save_json()
self.gen_result()
self.save_db()
rx_queue.put(self.results)
self.finish()
def do(domain, rx_queue): # 统一入口名字 方便多线程调用
def do(domain): # 统一入口名字 方便多线程调用
"""
类统一调用入口
:param str domain: 域名
:param rx_queue: 结果集队列
"""
query = Entrust(domain)
query.run(rx_queue)
query.run()
if __name__ == '__main__':
result_queue = queue.Queue()
do('example.com', result_queue)
do('example.com')
oneforall/modules/certificates/google.py
+4 -8
View File
@@ -1,5 +1,4 @@
# coding=utf-8
import queue
import time
from common import utils
@@ -28,7 +27,7 @@ class Google(Query):
subdomains_find = utils.match_subdomain(self.domain, resp.text)
self.subdomains = self.subdomains.union(subdomains_find) # 合并搜索子域名搜索结果
def run(self, rx_queue):
def run(self):
"""
类执行入口
"""
@@ -37,21 +36,18 @@ class Google(Query):
self.save_json()
self.gen_result()
self.save_db()
rx_queue.put(self.results)
self.finish()
def do(domain, rx_queue): # 统一入口名字 方便多线程调用
def do(domain): # 统一入口名字 方便多线程调用
"""
类统一调用入口
:param str domain: 域名
:param rx_queue: 结果集队列
"""
query = Google(domain)
query.run(rx_queue)
query.run()
if __name__ == '__main__':
result_queue = queue.Queue()
do('example.com', result_queue)
do('example.com')