Python安全工具源碼分析：wydomain
摘要：腳本直接運行 > parser_args命令行參數獲取用戶輸入 > 傳入主函數run > 根據傳入的參數，生成DomainFuzzer類的實例 > 調用DomainFuzzer類實例的run方法 > 獲取返回的數據寫入到空列表subdomains裏 > 調用save_result函數將列表中的內容寫入到json文件。腳本直接運行 > parser_args命令行參數獲取用戶輸入 > 傳入主函數run > 根據傳入的參數，生成DomainFuzzer類的實例 > 調用DomainFuzzer類實例的run方法 > 獲取返回的數據寫入到空列表subdomains裏 > 調用save_result函數將列表中的內容寫入到json文件。
工具介紹
wydomain是豬豬俠開發的一款子域名信息蒐集工具，因其枚舉速度快，結果準確，成爲不少白帽居家旅行的必備神器。工具主要分爲兩個模塊，dnsburte模塊和wydomain模塊，dnsburte模塊通過用戶自定義字典發送dns查詢，最後篩選出符合條件的域名。而wydomain模塊則是通過調用多個第三方網站的公開api獲取子域名數據。
工具目錄：
. # 目錄由tree生成 ├── captcha.py # 驗證碼識別 ├── common.py # 發送http請求 ├── config.py # 主配置文件 ├── default.csv # 默認字典 ├── dnsburte.py # dnsburte模塊通過用戶自定義字典發送dns查詢，最後篩選出符合條件的域名。 ├── dnspod.csv # dnspod子域名字典 ├── domains.log ├── ph_cookie.js ├── README.md ├── requirements.txt # 依賴庫 ├── result # 子域名枚舉結果文件夾 │ ├── aliyun.com │ │ ├── alexa.json │ │ ├── chaxunla.json │ │ ├── dnsburte.json │ │ ├── googlect_dnsnames.json │ │ ├── googlect_subject.json │ │ ├── ilinks.json │ │ ├── netcraft.json │ │ ├── sitedossier.json │ │ ├── threatcrowd.json │ │ └── threatminer.json │ └── weibo.com │ ├── alexa.json │ ├── chaxunla.json │ ├── dnsburte.json │ ├── googlect_dnsnames.json │ ├── googlect_subject.json │ ├── ilinks.json │ ├── netcraft.json │ ├── sitedossier.json │ ├── threatcrowd.json │ └── threatminer.json ├── tools │ ├── __init__.py │ └── skynet.py ├── upload.py ├── utils # 第三方網站查詢模塊 │ ├── alexa.py │ ├── chaxunla.py │ ├── fileutils.py │ ├── googlect.py │ ├── ilinks.py │ ├── __init__.py │ ├── netcraft.py │ ├── passivetotal.py │ ├── sitedossier.py │ ├── threatcrowd.py │ └── threatminer.py ├── weibo_domains.log ├── wydomain.csv # wydomain 字典 └── wydomain.py # dnsburte模塊通過用戶自定義字典發送dns查詢，最後篩選出符合條件的域名。 5 directories, 47 files
項目地址： https://github.com/ring04h/wydomain
dnsburte模塊
主運行流程
腳本直接運行 > parser_args命令行參數獲取用戶輸入 > 傳入主函數run > 根據傳入的參數，生成DomainFuzzer類的實例 > 調用DomainFuzzer類實例的run方法 > 獲取返回的數據寫入到空列表subdomains裏 > 調用save_result函數將列表中的內容寫入到json文件。
在逐個拆解代碼塊分析之前，最好能夠大致瀏覽一下腳本里都定義了哪些函數，哪些類，調用了哪些python標準模塊，第三方模塊，以及作者是否自己編寫了模塊。通過函數名稱，類名稱及作者的註釋可以快速理解部分代碼塊的主要作用。
python標準模塊：
import time import re import os import sys import json import Queue import random import logging import argparse import threading
python第三方模塊：
import dns.query import dns.resolver import dns.rdatatype
作者自定義模塊：
from common import save_result from utils.fileutils import FileUtils
自定義函數：
run()
自定義類：
Domain()
DomainFuzzer()
bruteWorker(threading.Thread)
全局變量：
logging.basicConfig( # logging模塊基礎配置文件
level=logging.INFO,
format='%(asctime)s [%(levelname)s] %(message)s',
)
nameservers = [ # 公共dns列表
'119.29.29.29', '182.254.116.116',
'8.8.8.8', '8.8.4.4',
'180.76.76.76',
'1.2.4.8', '210.2.4.8',
'101.226.4.6', '218.30.118.6',
'8.26.56.26', '8.20.247.20'
]
代碼段分析
當dnsburte模塊被用戶直接運行時，作者通過使用argparse模塊從用戶終端獲取命令行參數，最後將獲取到的參數傳入到主函數run()裏。
if __name__ == '__main__':
parser = argparse.ArgumentParser(description="wydomian v 2.0 to bruteforce subdomains of your target domain.")
parser.add_argument("-t","--thread",metavar="",default=16, # 線程數
help="thread count")
parser.add_argument("-time","--timeout",metavar="",default=5, # 超時時間
help="query timeout")
parser.add_argument("-d","--domain",metavar="", # 目標域名
help="domain name")
parser.add_argument("-f","--file",metavar="",default="wydomain.csv", # 字典
help="subdomains dict file name")
parser.add_argument("-o","--out",metavar="",default="bruteforce.log", # 結果
help="result out file")
args = parser.parse_args() # 將命令行參數傳遞到run函數。
try:
run(args)
except KeyboardInterrupt:
logging.info("Ctrl C - Stopping Client")
sys.exit(1)
調用流程：parser.parse_args() > run()
主函數run()獲取到用戶終端傳入的命令行參數，首先判斷了一下用戶是否已經輸入域名，如果沒有輸入域名則打印出幫助信息，程序退出。
def run(args):
domain = args.domain # 域名
thread_cnt = int(args.thread) # 線程數
timeout = int(args.timeout) # 超時時間
dict_file = args.file # 字典
outfile = args.out # 枚舉結果
if not domain:
print('usage: dnsburte.py -d aliyun.com') # 判斷域名是否已經輸入
sys.exit(1)
該工具正常運行時，子域名枚舉的結果存放在result/domain.com文件夾下。腳本首先獲取當前文件運行的絕對路徑，然後通過域名名稱將文件夾創建到result目錄下，如果文件夾不存在則直接新建一個文件夾。
# 初始化緩存路徑
script_path = os.path.dirname(os.path.abspath(__file__)) # 當前文件運行的絕對路徑
_cache_path = os.path.join(script_path, 'result/{0}'.format(domain)) # 緩存的路徑
if not os.path.exists(_cache_path): # 新建文件夾
os.makedirs(_cache_path, 0777) # 權限
初始化緩存路徑以後，作者又新建一個空列表用於存放枚舉出的子域名，然後將域名，字典傳入到DomainFuzzer()類中，生成一個類的實例。調用實例的run()方法獲取子域名列表，通過循環讀取出子域名，最後存放到列表裏。
調用流程：DomainFuzzer() > 類實例dnsfuzz > dnsfuzz.run() > subdomains_list
subdomains = [] # 空列表
dnsfuzz = DomainFuzzer(target=domain, dict_file=dict_file, timeout=timeout) # 生成DomainFuzzer的實例
logging.info("starting bruteforce threading({0}) : {1}".format(thread_cnt, domain))
for subname in dnsfuzz.run(thread_cnt=thread_cnt): # 調用實例的run方法
subdomains.append(subname) # 將循環讀取出的子域存放到空列表裏
_cache_path是域名枚舉結果的文件夾，在上上個代碼段，作者只是新建了文件夾用於存放結果，並沒有將域名枚舉的結果存放到文件夾裏，os.path.join()將存放的文件夾與dnsfuzz.run() 獲取到域名枚舉的結果拼接成一個新的文件路徑，最後調用作者自定義的函數save_result()將結果寫入到指定文件。而outfile_file()則是用戶從命令行傳遞過來指定保存結果的路徑。
_cache_file = os.path.join(_cache_path, 'dnsburte.json' ) # _cache_path是域名的文件夾，dnsburte.json是寫入文件的名稱
save_result(_cache_file, subdomains)
script_path = os.path.dirname(os.path.abspath(__file__)) # 當前文件運行的絕對路徑
_outfile_file = os.path.join(script_path, outfile) # 拼接最後生成的文件路徑
save_result(_outfile_file, subdomains)
logging.info("dns bruteforce subdomains({0}) successfully...".format(len(subdomains)))
logging.info("result save in : {0}".format(_outfile_file))
主函數分析完畢以後，將目光聚焦到dnsfuzz.run() ，這是DomainFuzzer()類實例調用的方法，作者通過這個方法獲取到了子域名枚舉的結果，我們現在跟進這個類，找到這個類方法進行分析就能知道作者是通過什麼樣的方式枚舉子域名。
DomainFuzzer類的初始化方法裏定義了根域名，子域名爆破所用到的字典，及Domain()，Domain()類可以先不用着急去跟進分析，先通過變量名簡單推斷一下這個類究竟是用來幹什麼的。domain肯定是和域名處理有關，resolver是分解器，組合到一起就是域名分解器……(要不要這麼機智 = = )。接下來繼續看run()方法，run()有一個默認參數線程數量爲16，類方法裏定一了兩個兄寶胎隊列，長得一摸一樣，不過兩兄弟一個隊列是用於存放讀取字典的隊列。而另一個隊列則是域名爆破結束以後寫入結果的隊列。光有空隊列不行哇，所以得通過for循環從字典中將子域名字符串提取出來，並且使用join()方法將子域名拼接完成，最後調用隊列的put()方法將域名直接put到隊列裏。
class DomainFuzzer(object):
"""docstring for DomainFuzzer with brute force"""
def __init__(self, target, dict_file='domain.csv', timeout=5):
self.target = target # 根域名
self.dict = FileUtils.getLines(dict_file) # 子域名爆破字典
self.resolver = Domain(timeout=timeout) # 將超時時間傳入到domain類中
def run(self, thread_cnt=16):
iqueue, oqueue = Queue.Queue(), Queue.Queue() # 兩個隊列iqueue是讀取字典的隊列，oqueue是寫入結果的隊列
for line in self.dict:
iqueue.put('.'.join([str(line),str(self.target)])) # 從字典裏提取出子域拼接域名。
# 將拼接好的域名直接put到隊列裏
到了這裏，作者從字典讀取子域名並拼接成功最後put到隊列裏，隊列裏有數據了，在後續執行dns查詢的時候，只需要從隊列裏get數據就可以發送請求了哇~
作者又定義了一個空列表用於存放線程，而Domain.extensive()傳入了目標的域名，並且返回到變量extensive，從函數的字面意思來看，這裏是和泛解析有關的處理。
在run()函數的初始化中，作者定義了線程的數量爲16，有線程數，但是並沒有生成相應的線程。於是使用for循環生成相應的線程數並添加到線程空列表裏，現在唯一值得困惑的就是extensive，不清楚extensive變量的具體類型是什麼，是用於幹什麼的。現在不用過於糾結。也不用跟進分析。知道有這些疑問，自己做個簡單的筆記就繼續往下看。bruteWorker()，它是一個類。從類的名稱來看，這個類的主要作用是用於枚舉工作的類。bruteWorker()裏傳入進去作者定義的兩個隊列。
現在，threads線程列表裏有相應的線程。直接使用for循環依次啓動線程。最後兩行代碼作者判斷了一下oqueue寫入結果的隊列是否不爲空，如果不爲空則直接get獲取隊列裏的數據
extensive, threads = self.resolver.extensive(self.target), [] # 定義了兩個變量一個extensive 用於存放泛解析,另一個變量則是線程空列表
for i in xrange(thread_cnt):
'''根據線程數生成相應的線程並添加到線程列表裏'''
threads.append(self.bruteWorker(self, iqueue, oqueue, extensive)) # 將bruteWorker傳入隊列
for t in threads: t.start() # 啓動線程
for t in threads: t.join()
while not oqueue.empty():# 判斷隊列是否不爲空
yield oqueue.get()
調用流程：dnsfuzz.run() > DomainFuzzer().run() > bruteWorker()
bruteWorker()不是一個單獨的類，而是在DomainFuzzer()類裏的嵌套類。
該類繼承於threading.Thread多線程類，類的初始化方法中定義了DomainFuzzer()類的實例及讀取字典與寫入結果的隊列，extensive的默認參數爲空列表。
class bruteWorker(threading.Thread):
'''多線程模塊'''
"""
domain name brute force threading worker class
@param dfuzzer DomainFuzzer base class
@param iqueue Subdomain dict Queue()
@param oqueue Brutefoce result Queue()
@param extensive Doman extensive record sets
"""
def __init__(self, dfuzzer, iqueue, oqueue, extensive=[]):
threading.Thread.__init__(self)
self.queue = iqueue # 讀取隊列
self.output = oqueue # 寫入隊列
self.dfuzzer = dfuzzer
self.extensive = extensive
run()方法首先判斷讀取字典的隊列是否不爲空，如果不爲空則從隊列裏獲取數據，其次判斷extensive的結果是否存在，如果不存在則直接調用DomainFuzzer()類實例的brute()方法。如果存在仍然調用DomainFuzzer()類實例的brute()方法，只是傳入了一個ret的參數爲True。從這裏可以分析出DomainFuzzer()類實例的brute方法是根據extensive的狀態來進行爆破。並且會將爆破的結果返回到rrset參數中。如果rrset參數的值不爲空，則通過for循環將rrset中的數據遍歷出來，最後將結果put到寫入結果的隊列。
def run(self):
try:
while not self.queue.empty():
sub = self.queue.get_nowait() # 從隊列裏獲取數據
if len(self.extensive) == 0: # not_extensive
if self.dfuzzer.resolver.brute(sub):
self.output.put(sub)
else:
rrset = self.dfuzzer.resolver.brute(sub, ret=True)
if rrset is not None:
for answer in rrset['A']:
if answer not in self.extensive:
self.output.put(sub) # 寫入結果
except Exception, e:
pass
調用流程：bruteWorker() > bruteWorker().run() > Domain.brute()
到目前爲止，只剩下Domain.extensive()，和Domain.brute()方法還沒有跟進分析。不過沒關係，想要的結果都會在Domain()類中找到答案。
Domain()類的初始化方法中定義了一個recursion空字典及dns.resolver.Resolver()dns查詢類。
class Domain(object):
"""docstring for Domain base class"""
def __init__(self, timeout=5):
self.recursion = {}
self.resolver = dns.resolver.Resolver()
if timeout:
self.resolver.timeout = timeout
self.resolver.lifetime = timeout
Domain()類的brute方法根據ret的狀態執行不同的查詢，ret條件爲True時，使用原始dns函數resolver.query()進行查詢，並且函數直接返回True，爲False時則使用Domain.query()進行查詢，最後將dns查詢以後的結果返回。作者在這裏使用異常處理，用於解決在執行dns查詢中會遇到的一些error，不過大多都是捕獲以後直接返回False，說句實話我現在也讓這兩個query()函數搞暈了，不過先靜下心慢慢分析，把存有疑問的地方記錄到紙上，繼續往下分析代碼。
def brute(self, target, ret=False):
"""
domain brute force fuzz
@param target burte force, domain name arg
@param ret return result flag
"""
try:
if not ret: # return_flag set false, using dns original query func
if self.resolver.query(target, 'A'):
return True
else: # return_flag set true
return self.query(target, 'A')
except dns.resolver.NoAnswer:
return False # catch the except, nothing to do
except dns.resolver.NXDOMAIN:
return False # catch the except, nothing to do
except dns.resolver.Timeout:
return self.burte(target) # timeout retry
except Exception, e:
logging.info(str(e))
return False
要解決掉上一個代碼段所存在的疑問，所以現在跟進Domain()類 > query()方法進行分析。
query()方法定義了兩個變量，一個是目標地址，另一個則是rdtype，rdtype一開始不理解具體是用來幹什麼的可以不用過於糾結，只需要知道傳入的是一個類型的變量。當函數獲取到用戶傳遞過來兩個參數，調用random隨機取公共dns列表裏的ip地址對target進行查詢。作者將執行dns查詢的以後的answer對象傳遞給了Domain()類 > parser()方法。
def query(self, target, rdtype):
try:
self.resolver.nameservers = [random.choice(nameservers),random.choice(nameservers)]
answer = self.resolver.query(target, rdtype)
return self.parser(answer)
except dns.resolver.NoAnswer:
return None # catch the except, nothing to do
except dns.resolver.NXDOMAIN:
return None # catch the except, nothing to do
except dns.resolver.Timeout:
# timeout retry
print(target, rdtype, '<timeout>')
except Exception, e:
logging.info(str(e))
緊接着跟進parse()方法進行分析，這個函數有點東西哇。函數內定義了一個空字典，最後return的也是該變量，５個if條件都是和域名解析的類型有關，A類型，CNAME類型等等。這是粗略看了一下函數內部的部分代碼塊得到的信息，接下來詳細開始分析。
兩個for循環用於從dns查詢的結果裏獲取解析的類型，ip地址等信息，item.rdtype變量裏存放的是執行查詢以後的子域名解析類型。變量的類型則是數字。比如id 1所對應的解析類型是A，id 5所對應的解析類型是CNAME。函數通過dns查詢獲取到子域名所對應的解析id，而接着的一個if條件是與get_type_id(‘A’)做的判斷，由此可以推斷出腳本的get_type_name()方法是通過id反查解析類型，而get_type_id()方法則是通過解析類型反查id。
推斷結果正不正確不要急，反正也是推斷，到最後我們在跟進函數進行分析，驗證一下猜想。
根據作者的註釋可以得知result字典裏只存放了兩種格式，域名和ip地址。字典的鍵爲解析的類型，值爲域名或者ip地址組成的列表。
def parser(self, answer):
"""
result relationship only two format
@domain domain name
@address ip address
"""
result = {}
for rrsets in answer.response.answer:
for item in rrsets.items:
rdtype = self.get_type_name(item.rdtype) # 獲取子域名解析的類型
if item.rdtype == self.get_type_id('A'):
'''判斷解析的類型'''
if result.has_key(rdtype):
result[rdtype].append(item.address)
else:
result[rdtype] = [item.address]
if item.rdtype == self.get_type_id('CNAME'):
'''判斷解析的類型'''
if result.has_key(rdtype):
result[rdtype].append(item.target.to_text())
else:
result[rdtype] = [item.target.to_text()]
if item.rdtype == self.get_type_id('MX'):
'''判斷解析的類型'''
if result.has_key(rdtype):
result[rdtype].append(item.exchange.to_text())
else:
result[rdtype] = [item.exchange.to_text()]
if item.rdtype == self.get_type_id('NS'):
'''判斷解析的類型'''
if result.has_key(rdtype):
result[rdtype].append(item.target.to_text())
else:
result[rdtype] = [item.target.to_text()]
return result
前幾個解析類型都大同小異，但txt解析類型首先獲取了item.to_text()的值，然後判斷include字符串是否在item.to_text()結果裏，如果存在使用正則提取字符串，然後判斷該字符串是否是域名，如果是域名則繼續執行query()查詢域名解析。如果include字符串不在結果裏，則正則匹配有關ip的字符串。
if item.rdtype == self.get_type_id('TXT'):
rd_result = item.to_text()
if 'include' in rd_result:
regex = re.compile(r'(?<=include:).*?(?= )')
for record in regex.findall(rd_result):
if self.is_domain(record):
self.query(record, rdtype)
else:
regex = re.compile(r'(?<=ip4:).*?(?= |/)')
qname = rrsets.name.to_text()
self.recursion[qname] = []
for record in regex.findall(rd_result):
self.recursion[qname].append(record)
result[rdtype] = self.recursion
調用流程：Domain() > Domain.brute() > Domain.query() > Domain.parser()
get_type_name()是通過dns查詢結果中的id反查解析類型，而get_type_id()則是通過解析類型反查id。
is_domian()是一個類靜態方法，類和實例均可以調用。根據正則匹配的結果返回True或者Ｆalse。
is_ipv4()則是匹配ip地址。根據正則的匹配結果返回True或者False。
在這裏也收穫到兩條正則，記錄下來，以後遇到匹配域名或匹配ip地址的地方就可以直接借鑑了哇～
def get_type_name(self, typeid):
return dns.rdatatype.to_text(typeid) # 根據類型的id返回解析文字
def get_type_id(self, name):
return dns.rdatatype.from_text(name) # 根據解析的文字類型返回id
@staticmethod # 靜態方法，類和實例均可調用
def is_domain(self, domain):
'''正則匹配域名'''
domain_regex = re.compile(
r'(?:[A-Z0-9_](?:[A-Z0-9-_]{0,247}[A-Z0-9])?\.)+(?:[A-Z]{2,6}|[A-Z0-9-]{2,}(?<!-))\Z',
re.IGNORECASE)
return True if domain_regex.match(domain) else False
def is_ipv4(self, address):
'''正則匹配ipv4地址'''
ipv4_regex = re.compile(
r'(?:25[0-5]|2[0-4]\d|[0-1]?\d?\d)(?:\.(?:25[0-5]|2[0-4]\d|[0-1]?\d?\d)){3}',
re.IGNORECASE)
return True if ipv4_regex.match(address) else False
調用流程：Domain.parser() > get_type_name() & get_type_id() & is_domain()
Domain類的最後，則是一個處理泛解析的函數。
泛域名解析是指，利用通配符* （星號）來做二級域名以實現所有的二級域名均指向同一IP地址。在域名前添加任何子域名，均可訪問到所指向的web地址。
在這個函數中，作者定義了兩個列表，第一個列表爲空，第二個列表則是生成指定的三個子域名。然後循環進行dns查詢，如果查詢的結果不爲空，則將esets列表中添加新的記錄以後return。
def extensive(self, target):
'''處理泛解析'''
(ehost, esets) = ['wyspider{0}.{1}'.format(i, target) for i in range(3)], []
for host in ehost:
try:
record = self.query(host, 'A')
if record is not None:
esets.extend(record['A'])
except Exception:
pass
return esets
調用流程：Domain() > extensive()
總結：
腳本直接運行 > parser_args命令行參數獲取用戶輸入 > 傳入主函數run > 根據傳入的參數，生成DomainFuzzer類的實例 > 調用DomainFuzzer類實例的run方法 > 獲取返回的數據寫入到空列表subdomains裏 > 調用save_result函數將列表中的內容寫入到json文件。
wydomain模塊
wydomain模塊通過調用多個，第三方威脅情報蒐集網站的公開api獲取子域名數據。
主運行流程
腳本被直接運行時 > parser.parse_args()獲取用戶命令行輸入 > 命令行獲取的參數傳入主函數run() > 根據用戶傳入的域名調用指定類的run()方法獲取結果 > 最後調用save_result()將獲取到的結果寫入到指定文件。
python標準模塊：
import os import sys import json import argparse import logging
作者自定義模塊：
from common import save_result, read_json from utils.fileutils import FileUtils from utils.alexa import Alexa from utils.threatminer import Threatminer from utils.threatcrowd import Threatcrowd from utils.sitedossier import Sitedossier from utils.netcraft import Netcraft from utils.ilinks import Ilinks from utils.chaxunla import Chaxunla from utils.googlect import TransparencyReport
自定義函數：
run()
全局變量：
logging.basicConfig(
level=logging.INFO, # filename='/tmp/wyproxy.log',
format='%(asctime)s [%(levelname)s] %(message)s',
)
代碼塊分析
作者使用args = parser.parse_args()獲取用戶從命令行傳遞過來的域名及寫入子域名結果文件的路徑。最後將兩個參數傳入到run()函數中。
if __name__ == '__main__':
parser = argparse.ArgumentParser(description="wydomain v 2.0 to discover subdomains of your target domain.")
parser.add_argument("-d","--domain",metavar="",
help="domain name") # 域名
parser.add_argument("-o","--out",metavar="",default="domains.log",
help="result out file") # 結果輸出文件路徑
args = parser.parse_args()
try:
run(args)
except KeyboardInterrupt:
logging.info("Ctrl C - Stopping Client") # 異常記錄到日誌
sys.exit(1)
調用流程：parser.parse_args() > run()
run()函數獲取域名及寫入子域名結果文件的路徑。
首先判斷用戶是否已經輸入域名，如果沒有輸入域名則腳本打印幫助信息，程序退出。
def run(args):
domain = args.domain
outfile = args.out
if not domain:
print('usage: wydomain.py -d aliyun.com')
sys.exit(1)
初始化緩存文件路徑。
該工具正常運行時，api獲取的結果存放在result/domain.com文件夾下。腳本首先獲取當前文件運行的絕對路徑，然後通過域名名稱將文件夾創建到result目錄下，如果文件夾不存在則直接新建一個文件夾。
# init _cache_path
script_path = os.path.dirname(os.path.abspath(__file__))
_cache_path = os.path.join(script_path, 'result/{0}'.format(domain))
if not os.path.exists(_cache_path):
os.makedirs(_cache_path, 0777)
logging.info()在這裏的作用其實就是相當於print打印了一條提示信息。有關logging的代碼塊可以直接忽略掉哇～
os.path.join()則是將緩存的域名路徑與第三方網站獲取到子域名的信息文件組合爲一個完整的路徑。方便作者後續寫入文件的操作。在這個函數中，作者調用了多個類的run()方法，Alexa.run()，Threatminer.run()，Threatcrowd.run()，Sitedossier.run()，Netcraft.run()，Ilinks.run()，Chaxunla().run()，TransparencyReport().run()這些類分別對應了需要進行查詢的網站，當執行run方法以後，函數通過發送請求獲取數據進行處理，返回指定網站查詢的數據存入到變量，最後通過save_result()方法將結果寫入到用戶所指定的路徑。
# alexa result json file
logging.info("starting alexa fetcher...")
_cache_file = os.path.join(_cache_path, 'alexa.json')
result = Alexa(domain=domain).run()
save_result(_cache_file, result)
logging.info("alexa fetcher subdomains({0}) successfully...".format(len(result)))
# threatminer result json file
logging.info("starting threatminer fetcher...")
_cache_file = os.path.join(_cache_path, 'threatminer.json')
result = Threatminer(domain=domain).run()
save_result(_cache_file, result)
logging.info("threatminer fetcher subdomains({0}) successfully...".format(len(result)))
# threatcrowd result json file
logging.info("starting threatcrowd fetcher...")
_cache_file = os.path.join(_cache_path, 'threatcrowd.json')
result = Threatcrowd(domain=domain).run()
save_result(_cache_file, result)
logging.info("threatcrowd fetcher subdomains({0}) successfully...".format(len(result)))
# sitedossier result json file
logging.info("starting sitedossier fetcher...")
_cache_file = os.path.join(_cache_path, 'sitedossier.json')
result = Sitedossier(domain=domain).run()
save_result(_cache_file, result)
logging.info("sitedossier fetcher subdomains({0}) successfully...".format(len(result)))
# netcraft result json file
logging.info("starting netcraft fetcher...")
_cache_file = os.path.join(_cache_path, 'netcraft.json')
result = Netcraft(domain=domain).run()
save_result(_cache_file, result)
logging.info("netcraft fetcher subdomains({0}) successfully...".format(len(result)))
# ilinks result json file
logging.info("starting ilinks fetcher...")
_cache_file = os.path.join(_cache_path, 'ilinks.json')
result = Ilinks(domain=domain).run()
save_result(_cache_file, result)
logging.info("ilinks fetcher subdomains({0}) successfully...".format(len(result)))
# chaxunla result json file
logging.info("starting chaxunla fetcher...")
_cache_file = os.path.join(_cache_path, 'chaxunla.json')
result = Chaxunla(domain=domain).run()
save_result(_cache_file, result)
logging.info("chaxunla fetcher subdomains({0}) successfully...".format(len(result)))
# google TransparencyReport result json file
logging.info("starting google TransparencyReport fetcher...")
result = TransparencyReport(domain=domain).run()
_cache_file = os.path.join(_cache_path, 'googlect_subject.json')
save_result(_cache_file, result.get('subjects'))
_cache_file = os.path.join(_cache_path, 'googlect_dnsnames.json')
save_result(_cache_file, result.get('dns_names'))
logging.info("google TransparencyReport fetcher subdomains({0}) successfully...".format(len(result.get('dns_names'))))
調用流程：run() > Alexa.run() &Threatminer.run() & Threatcrowd.run() & Sitedossier.run() & Netcraft.run() & Ilinks.run() > Chaxunla.run() > TransparencyReport.run() > save_result() 寫入返回的結果
作者代碼寫的很清晰。不過在這裏也可以借鑑sublist3r工具作者的實現思路。
以下代碼塊節選自國外子域名爆破工具sublist3r.py，該工具的作者首先定義了一個supported_engines字典用於存放類，字典的鍵爲搜索引擎的名稱，值則爲搜索引擎所對應的類名。chosenEnums空字典則用於存放用戶以選擇的搜索引擎。如果用戶沒有指定搜索引擎則默認使用所有的搜索引擎，否則使用split函數進行分割分別獲取用戶指定的搜索引擎，然後調用for循環將用戶以選擇的搜索引擎類添加到列表。最後通過列表解析從選擇好的搜索引擎列表中傳入指定的域名進行子域名枚舉。
supported_engines = {'baidu': BaiduEnum,
'yahoo': YahooEnum,
'google': GoogleEnum,
'bing': BingEnum,
'ask': AskEnum,
'netcraft': NetcraftEnum,
'dnsdumpster': DNSdumpster,
'virustotal': Virustotal,
'threatcrowd': ThreatCrowd,
'ssl': CrtSearch,
'passivedns': PassiveDNS
} # 存放搜索引擎類
chosenEnums = []
if engines is None:
chosenEnums = [
BaiduEnum, YahooEnum, GoogleEnum, BingEnum, AskEnum,
NetcraftEnum, DNSdumpster, Virustotal, ThreatCrowd,
CrtSearch, PassiveDNS
] # 默認使用所有的搜索引擎。
else:
engines = engines.split(',')
for engine in engines:
if engine.lower() in supported_engines:
chosenEnums.append(supported_engines[engine.lower()])
# Start the engines enumeration
enums = [enum(domain, [], q=subdomains_queue, silent=silent, verbose=verbose) for enum in chosenEnums]
for enum in enums:
enum.start()
for enum in enums:
enum.join()
繼續回到run()函數，作者通過調用不同類的run()方法獲取個個網站的api查詢結果，最後將他們分別保存到了不同的文件。subdomains空列表裏用於存放所有獲取到的子域名地址，而run函數剩下的這一部分代碼塊則是作者將所有的查詢結果文件進行處理，最後進行整合。
read_json()，save_result()這些函數都是作者自定義的函數，主要的作用是對於文件的讀取，寫入，刪除等等操作。
# Collection API Subdomains
sub_files = [
'alexa.json',
'chaxunla.json',
'ilinks.json',
'netcraft.json',
'sitedossier.json',
'threatcrowd.json',
'threatminer.json']
# process all cache files
subdomains = []
for file in sub_files:
_cache_file = os.path.join(_cache_path, file)
json_data = read_json(_cache_file)
if json_data:
subdomains.extend(json_data)
# process openssl x509 dns_names
_cache_file = os.path.join(_cache_path, 'googlect_dnsnames.json')
json_data = read_json(_cache_file)
for sub in json_data:
if sub.endswith(domain):
subdomains.append(sub)
# collection burte force subdomains
_burte_file = os.path.join(_cache_path, 'dnsburte.json')
if FileUtils.exists(_burte_file):
json_data = read_json(_burte_file)
if json_data:
subdomains.extend(json_data)
# save all subdomains to outfile
subdomains = list(set(subdomains))
_result_file = os.path.join(script_path, outfile)
save_result(_result_file, subdomains)
logging.info("{0} {1} subdomains save to {2}".format(
domain, len(subdomains), _result_file))
調用流程：subdomains for循環讀取> sub_files內容
只要和網站查詢有關，就一定會涉及到發送請求及獲取數據，最後根據需求篩選數據。接下來逐個跟進這些類進行分析，Alexa()，Threatminer()，Threatcrowd()，Sitedossier()，Netcraft()，Ilinks()，Chaxunla()，TransparencyReport()。
1.alexa模塊
數據源： http://www.alexa.cn/
python標準模塊：
import re import time import logging
作者自定義模塊：
from common import http_request_get, http_request_post, is_domain
自定義類：
Alexa(object)
自定義類方法：
run() fetch_chinaz() fetch_alexa_cn() get_sign_alexa_cn()
Alexa類的初始化方法中，定義了兩個變量，一個是目標域名，而另一個則是空列表
class Alexa(object):
"""docstring for Alexa"""
def __init__(self, domain):
super(Alexa, self).__init__()
self.domain = domain
self.subset = []
run方法爲Alexa類的主要運行函數。函數內部調用了fetch_chinaz()與fetch_alexa_cn()方法。從這兩個方法的命名來看，通過調用這兩個方法可直接取指定網站的返回結果，函數最後返回的則是一個去重以後的列表。
def run(self):
try:
self.fetch_chinaz()
self.fetch_alexa_cn()
return list(set(self.subset))
except Exception as e:
logging.info(str(e))
return self.subset
現在跟進fetch_chinaz()，與fetch_alexa_cn()方法。作者在註釋裏也提示這兩個方法，可以直接獲取指定網站的子域名查詢結果。既然要進行查詢，肯定會發送http請求，作者使用http_request_get()發送api接口的查詢請求。http的請求過程則讓作者定義到函數里了，在發送請求的時候直接傳入域名就可以發送請求了。發送請求以後，腳本獲取到了http的返回信息，然後實用正則將html中的子域名提取出來，最後存放到subset空列表裏。
fetch_alexa_cn()獲取數據的方法與fetch_chinaz()大同小異，只是在發送查詢請求之前調用了get_sign_alexa_cn()，從方法的命名來看，這裏獲取了簽名的狀態，當簽名的狀態不爲空時，則獲取簽名中domain，sig，keyt的參數，沒有這三個參數是無法發送http請求進行查詢。根據獲取到簽名的這三個參數。作者構造post請求進行發送，最後通過各種字符串的處理提取出子域名，並將結果添加到subset列表裏。
def fetch_chinaz(self):
"""get subdomains from alexa.chinaz.com"""
url = 'http://alexa.chinaz.com/?domain={0}'.format(self.domain)
r = http_request_get(url).content
subs = re.compile(r'(?<="\>\r\n<li>).*?(?=</li>)') # 正則匹配<li>標籤內的域名
result = subs.findall(r)
for sub in result:
if is_domain(sub):
self.subset.append(sub)
def fetch_alexa_cn(self):
"""get subdomains from alexa.cn"""
sign = self.get_sign_alexa_cn()
if sign is None:
raise Exception("sign_fetch_is_failed")
else:
(domain,sig,keyt) = sign
pre_domain = self.domain.split('.')[0] # baidu.com > baidu == pre_domain
url = 'http://www.alexa.cn/api_150710.php'
payload = {
'url': domain,
'sig': sig,
'keyt': keyt,
}
r = http_request_post(url, payload=payload).text
for sub in r.split('*')[-1:][0].split('__'):
if sub.split(':')[0:1][0] == 'OTHER':
break
else:
sub_name = sub.split(':')[0:1][0]
sub_name = ''.join((sub_name.split(pre_domain)[0], domain))
if is_domain(sub_name):
self.subset.append(sub_name)
調用流程：Alexa() > run() > fetch_chinaz() & fetch_alexa_cn()
2.threatminer模塊
數據源： https://www.threatminer.org
python標準模塊：
import re import logging
作者自定義模塊：
from common import curl_get_content, http_request_get, is_domain
自定義類：
Threatminer()
自定義類方法：
run()
Threatminer類初始化方法中定義了三個參數，第一個參數爲用戶指定的根域名domain，第二個參數爲subset空列表用於存放已經獲取到的子域名結果，第三個參數website則是數據源地址。
run()方法中根據用戶指定的域名拼接api接口請求，然後獲取返回的html信息，使用正則進行篩選子域名。最後將獲取到的結果添加到subset列表。
class Threatminer(object):
"""docstring for Threatminer"""
def __init__(self, domain):
super(Threatminer, self).__init__()
self.domain = domain
self.subset = []
self.website = "https://www.threatminer.org"
def run(self):
try:
url = "{0}/getData.php?e=subdomains_container&q={1}&t=0&rt=10&p=1".format(self.website, self.domain)
# content = curl_get_content(url).get('resp')
content = http_request_get(url).content
_regex = re.compile(r'(?<=<a href\="domain.php\?q=).*?(?=">)')
for sub in _regex.findall(content):
if is_domain(sub):
self.subset.append(sub)
return list(set(self.subset))
except Exception as e:
logging.info(str(e))
return self.subset
調用流程：Threatminer() > run()
3.threatcrowd模塊
數據源：
https://www.threatcrowd.org
python標準模塊
import re import json import logging
作者自定義模塊：
from common import curl_get_content, http_request_get, is_domain
自定義類：
Threatcrowd()
自定義類方法：
run()
Threatcrowd()類初始化方法中定義了三個參數，第一個參數爲用戶指定的根域名domain，第二個參數爲subset空列表用於存放已經獲取到的子域名結果，第三個參數website則是數據源地址。
run()方法中根據用戶指定的域名拼接api接口請求，然後獲取返回的json信息，使用json.loads()進行加載篩選子域名。最後將獲取到的結果添加到subset列表。
class Threatcrowd(object):
"""docstring for Threatcrowd"""
def __init__(self, domain):
super(Threatcrowd, self).__init__()
self.domain = domain
self.subset = []
self.website = "https://www.threatcrowd.org"
def run(self):
try:
url = "{0}/searchApi/v2/domain/report/?domain={1}".format(self.website, self.domain)
# content = curl_get_content(url).get('resp')
content = http_request_get(url).content
for sub in json.loads(content).get('subdomains'):
if is_domain(sub):
self.subset.append(sub)
return list(set(self.subset))
except Exception as e:
logging.info(str(e))
return self.subset
調用流程：Threatcrowd() > run()
4.sitedossier模塊
數據源： http://www.sitedossier.com
python標準模塊：
import re import time import logging
作者自定義模塊：
from captcha import Captcha from common import http_request_get, http_request_post, is_domain
自定義類：
Sitedossier()
自定類方法：
run() get_content() parser() get_subdomain() human_act() audit() get_audit_img() __str__()
類初始化方法中定義了域名參數domain，Captcha()驗證碼類，subset空列表用於存放獲取到的子域名結果。
class Sitedossier(object):
"""docstring for Sitedossier"""
def __init__(self, domain):
super(Sitedossier, self).__init__()
self.domain = domain
self.captcha = Captcha()
self.subset = []
run()方法根據用戶傳遞的domain參數調用get_content()方法進行api查詢，獲取查詢的結果。parser()方法用於從返回的結果裏篩選特定的數據。set函數用於列表去重。
def run(self):
try:
url = 'http://www.sitedossier.com/parentdomain/{0}'.format(self.domain)
r = self.get_content(url)
self.parser(r)
return list(set(self.subset))
except Exception, e:
logging.info(str(e))
return self.subset
run() > get_content() 方法中調用http_request_get()發送http請求，發送請求以後首先判斷網站是否存在驗證碼驗證，如果存在，則調用human_act進行驗證碼處理。否則調用get_content()獲取網頁數據。
def get_content(self, url):
logging.info('request: {0}'.format(url))
r = http_request_get(url).text
if self.human_act(r) is True:
return r
else:
self.get_content(url)
跟進get_content() > human_act()方法進行分析，human_act()首先獲取發送請求以後的數據，if條件判斷auditimage與blacklisted是否在返回的頁面裏，如果存在，則調用get_audit_img()獲取圖片驗證碼，如果圖片的地址不爲空，則調用captcha.verification()進行驗證碼驗證。再次進行判斷驗證碼驗證以後的結果中是否存在Result鍵，如果存在則直接調用audit()方法。
def human_act(self, response):
if 'auditimage' in response or 'blacklisted' in response:
imgurl = self.get_audit_img(response)
if imgurl is not None:
ret = self.captcha.verification(imgurl)
if ret.has_key('Result'):
self.audit(ret['Result'])
return True
else:
raise Exception("captcha_verification_is_empty")
else:
raise Exception("audit_img_is_empty")
else:
return True
跟進human_act() > get_audit_img()方法，作者使用正則表達式篩選出圖片驗證碼的url地址，最後拼接完整的url地址進行返回。
def get_audit_img(self, response):
auditimg = re.compile(r'(?<=<img src\=\"/auditimage/).*?(?=\?" alt="Please)')
imgurl = auditimg.findall(response)[0:]
if len(imgurl) >= 1:
imgurl = 'http://www.sitedossier.com/auditimage/{0}'.format(imgurl[0])
return imgurl
else:
return None
跟進human_act() > audit()方法，方法內部根據用戶傳入的code參數發送http請求。
def audit(self, code):
payload = {'w':code}
url = 'http://www.sitedossier.com/audit'
r = http_request_post(url, payload=payload)
human_act() > captcha.verification方法暫時不跟進，只需要知道這裏作者調用了第三方的驗證碼識別接口，通過獲取到驗證碼圖片的地址以後，傳遞給該函數，該函數將識別出的驗證碼結果返回。後續詳細分析Captcha()類。
接下來跟進run() > parser()函數進行分析。作者通過調用正則表達式從response中篩選頁面的page，根據page調用get_subdomain()方法獲取子域名信息。最後將結果添加到subset列表。
def parser(self, response):
npage = re.search('<a href="/parentdomain/(.*?)"><b>Show', response)
if npage:
for sub in self.get_subdomain(response):
self.subset.append(sub)
nurl = 'http://www.sitedossier.com/parentdomain/{0}'.format(npage.group(1))
response = self.get_content(nurl)
self.parser(response)
else:
for sub in self.get_subdomain(response):
self.subset.append(sub)
跟進parser() > get_subdomain()方法，函數內部通過正則表達式提取出域名。
def get_subdomain(self, response):
domain = re.compile(r'(?<=<a href\=\"/site/).*?(?=\">)')
for sub in domain.findall(response):
yield sub
跟進最後一個類方法Sitedossier() > str
def __str__(self):
handler = lambda e: str(e)
return json.dumps(self, indent=2, default=handler)
調用方法：Sitedossier() > run() > get_content() & parser()
get_content() > human_act() > get_audit_img() & audit()
parser() > get_subdomain()
5.netcraft模塊
數據源： http://searchdns.netcraft.com
python標準模塊：
import re import time import logging import subprocess
作者自定義模塊：
from common import http_request_get, http_request_post, is_domain
自定義類：
Netcraft()
自定義類函數類：
run() parser() get_subdomains() get_cookie()
Netcraft類初始化方法中定義了cookie，subset空列表，用戶指定的域名地址，及數據源地址site參數。
class Netcraft(object):
"""docstring for Netcraft"""
def __init__(self, domain):
super(Netcraft, self).__init__()
self.cookie = ''
self.subset = []
self.domain = domain
self.site = 'http://searchdns.netcraft.com'
主運行方法run()，首先通過get_cookie()方法獲取到用戶cookie信息，調用http_request_get()發送http請求之前會將cookie信息傳遞到請求體裏，最後通過parser()方法處理http的返回信息，並且將獲取到的子域名信息添加到subset列表。
def run(self):
try:
self.cookie = self.get_cookie().get('cookie')
url = '{0}/?restriction=site+contains&position=limited&host=.{1}'.format(
self.site, self.domain)
r = http_request_get(url, custom_cookie=self.cookie)
self.parser(r.text)
return list(set(self.subset))
except Exception, e:
logging.info(str(e))
return self.subset
跟進run() > get_cookie()與run() > parser()方法進行詳細分析。
get_cookie() 方法裏通過subprocess.Popen()調用操作系統命令執行phantomjs ph_cookie.js，將運行的結果返回到stdoutput與erroutput參數中，最後組成cookie字典return。如果cmdline字符串用戶可以從外部控制，則這裏會存在遠程命令漏洞。
def get_cookie(self):
try:
cmdline = 'phantomjs ph_cookie.js'
run_proc = subprocess.Popen(cmdline,shell=True,stdout=subprocess.PIPE,stderr=subprocess.PIPE)
(stdoutput,erroutput) = run_proc.communicate()
response = {
'cookie': stdoutput.rstrip(),
'error': erroutput.rstrip(),
}
return response
except Exception, e:
logging.info(str(e))
return {'cookie':'', 'error': str(e)}
parser()方法使用正則表達式提取出頁面的page id，然後通過parser() > get_subdomains()獲取子域名信息。 parser() > is_domain()方法判斷獲取到的數據是否爲域名格式。最後將獲取到的數據添加到subset列表。
def parser(self, response):
npage = re.search('<A href="(.*?)"><b>Next page</b></a>', response)
if npage:
for item in self.get_subdomains(response):
if is_domain(item):
self.subset.append(item)
nurl = '{0}{1}'.format(self.site, npage.group(1))
r = http_request_get(nurl, custom_cookie=self.cookie)
time.sleep(3)
self.parser(r.text)
else:
for item in self.get_subdomains(response):
if is_domain(item):
self.subset.append(item)
跟進最後一個方法parser() > get_subdomains()，作者使用正則匹配a標籤內的域名地址。
def get_subdomains(self, response):
_regex = re.compile(r'(?<=<a href\="http://).*?(?=/" rel\="nofollow")')
domains = _regex.findall(response)
for sub in domains:
yield sub
調用流程：Netcraft() > run() > get_cookie() & http_request_get() & parser()
parser() > get_subdomains()
6.Ilinks模塊：
數據源： http://i.links.cn/subdomain/
python標準模塊：
import re
作者自定義模塊：
from common import http_request_get, http_request_post, is_domain
自定義類：
Ilinks()
自定義類方法：
run()
Ilinks()類初始化方法中定義了，用戶傳遞的域名domain參數，及數據源的api地址，subset列表用於存放執行查詢以後的子域名結果。
run()方法根據用戶指定的domain地址，傳遞到post請求體內，發送請求獲取返回的數據，最後使用正則表達式將查詢的結果篩選出來存放到列表 subset。
class Ilinks(object):
"""docstring for Ilinks"""
def __init__(self, domain):
super(Ilinks, self).__init__()
self.domain = domain
self.url = 'http://i.links.cn/subdomain/'
self.subset = []
def run(self):
try:
payload = {
'b2': 1,
'b3': 1,
'b4': 1,
'domain': self.domain
}
r = http_request_post(self.url,payload=payload).text
subs = re.compile(r'(?<=value\=\"http://).*?(?=\"><input)')
for item in subs.findall(r):
if is_domain(item):
self.subset.append(item)
return list(set(self.subset))
except Exception as e:
logging.info(str(e))
return self.subset
調用流程：Ilinks() > run()
7.chaxunla模塊：
數據源： http://api.chaxun.la/toolsAPI/getDomain
python標準模塊：
import re import time import json import logging
python第三方模塊：
import requests
作者自定義模塊：
from captcha import Captcha from common import is_domain
自定義類：
Chaxunla()
自定義類方法：
run() download() verify_code()
全局變量：
req = requests.Session()
Chaxunla的初始化方法中定義了用戶傳遞的domain參數，數據源的api地址url參數，子域名枚舉的結果列表subset，還有一個用於驗證的空字符串。
class Chaxunla(object):
"""docstring for Chaxunla"""
def __init__(self, domain):
super(Chaxunla, self).__init__()
self.domain = domain
self.url = 'http://api.chaxun.la/toolsAPI/getDomain/'
self.subset = []
self.verify = ""
run()方法中首先拼接url地址發送查詢的請求，通過if條件判斷返回的字符串status是否等於1，如果等於1則執行正常數據查詢，最後將獲取到的數據添加到subset列表中。
def run(self):
try:
timestemp = time.time()
url = "{0}?0.{1}&callback=&k={2}&page=1&order=default&sort=desc&action=moreson&_={3}&verify={4}".format(
self.url, timestemp, self.domain, timestemp, self.verify)
result = json.loads(req.get(url).content) # 讀取json格式數據
if result.get('status') == '1':
for item in result.get('data'):
if is_domain(item.get('domain')):
self.subset.append(item.get('domain'))
elif result.get('status') == 3:
logging.info("chaxun.la api block you ip...")
logging.info("input you verify_code in http://subdomain.chaxun.la/wuyun.org/")
# print('get verify_code():', self.verify)
# self.verify_code()
# self.run()
return list(set(self.subset))
except Exception as e:
logging.info(str(e))
return self.subset
跟進verify_code()方法，首先生成了時間戳，然後通過時間戳構造圖片url地址，調用download()方法下載圖片，最後調用captcha.verification()方法識別驗證碼，返回識別以後的結果。
def verify_code(self):
timestemp = time.time() # 時間戳
imgurl = 'http://api.chaxun.la/api/seccode/?0.{0}'.format(timestemp)
if self.download(imgurl):
captcha = Captcha()
code_result = captcha.verification(filename='captcha.gif')
self.verify = code_result.get('Result')
跟進download方法，根據url發送請求獲取圖片信息，然後保存到本地。
def download(self, url):
try:
r = req.get(url)
with open("captcha.gif", "wb") as image:
image.write(r.content)
return True
except Exception, e:
return False
調用流程：Chaxunla() > run() > verify_code() & download()
8.googlect模塊
數據源： https://www.google.com/transparencyreport/jsonp/ct
python標準模塊：
import time import json import logging from random import Random,uniform from urllib import quote
作者自定義模塊：
from common import http_request_get
自定義函數：
random_sleep() random_str()
自定義類：
TransparencyReport()
TransparencyReport()類初始化方法中定義了用戶指定的域名domain參數，token空字符串，dns_names空列表用於存放dns，subjects空列表，hashs空列表，默認num_result參數，website參數爲api接口地址。
class TransparencyReport(object):
"""docstring for TransparencyReport"""
def __init__(self, domain):
self.domain = domain
self.token = ""
self.dns_names = []
self.subjects = []
self.hashs = []
self.num_result = 0
self.website = 'https://www.google.com/transparencyreport/jsonp/ct'
run()方法中主要調用了parser_subject()，與parser_dnsname()這這兩個方法，run()方法最後的返回值是一個字典，字典的值爲空列表。所以推斷通過運行上面的兩個方法獲取某些數據，添加到已經定義好的空列表裏，最後return組成好的字典變量。
def run(self):
self.parser_subject()
self.hashs = list(set(self.hashs)) # unique sort hash
self.parser_dnsname()
self.dns_names = list(set(self.dns_names))
self.subjects = list(set(self.subjects))
return {'subjects': self.subjects, 'dns_names': self.dns_names}
主要跟進run() > parser_subject()與run() > parser_dnsname()這兩個方法進行分析。
run() > parser_subject() > 調用random_str()獲取執行以後的結果。獲取callback參數然後拼接url地址，通過http_request_get()方法發送http請求並獲取返回的content，json.loads讀取並處理返回的json格式數據。result通過get()函數獲取讀取下一頁數據的token，最後 for循環根據條件讀取出域名數據，將其添加到dns_names列表中。
def parser_subject(self):
try:
callback = random_str()
url = '{0}/search?domain={1}&incl_exp=true&incl_sub=true&token={2}&c={3}'.format(self.website, self.domain, quote(self.token), callback)
content = http_request_get(url).content
result = json.loads(content[27:-3])
self.token = result.get('nextPageToken')
for subject in result.get('results'):
if subject.get('subject'):
self.dns_names.append(subject.get('subject'))
if subject.get('hash'):
self.hashs.append(subject.get('hash'))
except Exception as e:
logging.info(str(e))
if self.token:
self.parser_subject()
run > parser_dnsname()方法作者主要用於處理調用谷歌搜索的部分加密數據。如果沒有特定的hash字符串api的請求是無法完成的。
def parser_dnsname(self):
for hashstr in self.hashs:
try:
callback = random_str()
url = '{0}/cert?hash={1}&c={2}'.format(
self.website, quote(hashstr), callback)
content = http_request_get(url).content
result = json.loads(content[27:-3])
if result.get('result').get('subject'):
self.subjects.append(result.get('result').get('subject'))
if result.get('result').get('dnsNames'):
self.dns_names.extend(result.get('result').get('dnsNames'))
except Exception as e:
logging.info(str(e))
random_sleep()
很多搜索引擎的查詢結果都使用自身的加密算法對搜索的結果進行加密，比如雅虎搜索引擎，所以在編寫這種類似爬蟲腳本的時候，一定要根據網站執行查詢請求時發送的數據與返回的數據進行相應的處理。
以下代碼塊節選自EngineCrawler項目 lib/engines/yahoo.py 文件
def _extract_url(self, resp):
'''
雅虎搜索返回的url結果：
yahoo_encryption_url = '
http://r.search.yahoo.com/_ylt=AwrgDaPBKfdaMjAAwOBXNyoA;_ylu=X3oDMTByb2lvbXVuBGNvbG8DZ3ExBHBvcwMxBHZ0aWQDBHNlYwNzcg--/RV=2/RE=1526176322/RO=10/RU=
http://berkeleyrecycling.org/page.php?id=1
/RK=2/RS=6rTzLqNgZrFS8Kb4ivPrFbBBuFs-'
'''
soup = BeautifulSoup(resp.text, "lxml")
try:
a_tags = soup.find_all("a", " ac-algo fz-l ac-21th lh-24")
for a_tag in a_tags:
yahoo_encryption_url = a_tag['href']
yahoo_decrypt_url = unquote(yahoo_encryption_url) # 解碼
split_url = yahoo_decrypt_url.split('http://')
if len(split_url) == 1:
result_https_url = 'https://' + split_url[0].split('https://')[1].split('/R')[0] # 獲取返回https協議的url
self.urls.append(result_https_url)
print '[-]Yahoo: ' + result_https_url
else:
result_http_url = 'http://' + split_url[2].split('/R')[0] # 獲取返回http協議的url
print '[-]Yahoo: ' + result_http_url
self.urls.append(result_http_url)
except Exception:
pass
調用流程：
TransparencyReport() > run() > parser_subject() & parser_dnsname() & parser_dnsname()
9.passivetotal模塊
數據源地址： https://www.passivetotal.org
python標準模塊：
import re import json import logging
python第三方模塊：
import requests
作者自定義模塊：
from common import curl_get_content, http_request_get, is_domain,http_request_post
自定義類：
PassiveTotal()
自定義類方法：
run()
PassiveTotal()類初始化方法中定義了用戶指定的domain參數，用於存放子域名的subset空列表，website則是數據源地址。
run()方法中首先根據賬號，key進行認證，然後構造http請求將需要查詢的網站地址傳遞到post請求體裏，根據返回的信息使用json.loads進行讀取處理。最後使用for循環將子域名結果添加到列表內。
class PassiveTotal(object):
"""docstring for PassiveTotal"""
def __init__(self, domain):
super(PassiveTotal, self).__init__()
self.domain = domain
self.subset = []
self.website = "https://www.passivetotal.org"
def run(self):
try:
# 此auth 需要自行申請，每個auth 每天有查詢次數限制
auth=("[email protected]","d160262241ccf53222d42edc6883c129")
payload={"query":"*.%s" % self.domain}
url = "https://api.passivetotal.org/v2/enrichment/subdomains"
response = requests.get(url,auth=auth,params=payload)
for sub in json.loads(response.content)['subdomains']:
sub="%s.%s" %(sub,self.domain)
if is_domain(sub):
self.subset.append(sub)
return list(set(self.subset))
except Exception as e:
logging.info(str(e))
return self.subset
調用流程：PassiveTotal() > run()
*本文作者：c01d，轉載請註明來自FreeBuf.COM
