v0.0.1

2026-08-26 12:57:50 +08:00 · 2019-08-02 00:35:53 +08:00
parent 2eda73ed34
commit 9d581e5134
82 changed files with 129461 additions and 751 deletions
@@ -0,0 +1,98 @@
+# coding=utf-8
+"""
+查询域名的NS记录（域名服务器记录，记录该域名由哪台域名服务器解析），
+检查查出的域名服务器是否开启DNS域传送，如果开启且没做访问控制和身份验证便加以利用获取域名的所有记录
+
+DNS域传送（DNS zone transfer）指的是一台备用域名服务器使用来自主域名服务器的数据刷新自己的域数据库，
+目的是为了做冗余备份，防止主域名服务器出现故障时 dns 解析不可用。
+当主服务器开启DNS域传送同时又对来请求的备用服务器未作访问控制和身份验证便可以利用此漏洞获取某个域的所有记录。
+"""
+import time
+import queue
+import dns.resolver
+import dns.zone
+from config import logger
+from common import utils, resolve
+from common.module import Module
+
+
+class CheckAXFR(Module):
+    """
+    DNS域传送漏洞检查类
+    """
+    def __init__(self, domain: str):
+        Module.__init__(self)
+        self.domain = self.register(domain)
+        self.module = 'Check'
+        self.source = 'AXFRCheck'
+        self.nsservers = []
+        self.results = []
+
+    def check(self):
+        """
+        正则匹配响应头中的内容安全策略字段以发现子域名
+        :return: None
+        """
+        resolver = resolve.dns_resolver()
+        try:
+            answers = resolver.query(self.domain, "NS")
+        except Exception as e:
+            logger.log('ERROR', e)
+            return
+        self.nsservers = [str(answer) for answer in answers]
+        if not len(self.nsservers):
+            logger.log('ALERT', f'没有找到{self.domain}的NS域名服务器记录')
+            return
+        for nsserver in self.nsservers:
+            logger.log('DEBUG', f'正在尝试对{self.domain}的域名服务器{nsserver}进行域传送')
+            try:
+                xfr = dns.query.xfr(nsserver, self.domain)
+                zone = dns.zone.from_xfr(xfr)
+            except Exception as e:
+                logger.log('DEBUG', str(e))
+                logger.log('INFOR', f'对{self.domain}的域名服务器{nsserver}进行域传送失败')
+                continue
+            else:
+                names = zone.nodes.keys()
+                for name in names:
+                    subdomain = utils.match_subdomain(self.domain, str(name)+'.'+self.domain)
+                    self.subdomains = self.subdomains.union(subdomain)
+                    record = zone[name].to_text(name)
+                    self.results.append(record)
+            if self.results:
+                logger.log('INFOR', f'发现{self.domain}在{nsserver}上的域传送记录')
+                logger.log('DEBUG', '\n'.join(self.results))
+                self.results = []
+
+    def run(self, rx_queue):
+        """
+        类执行入口
+        """
+        logger.log('DEBUG', f'开始执行{self.source}检查{self.domain}的域传送漏洞')
+        start = time.time()
+        self.check()
+        end = time.time()
+        self.elapsed = round(end - start, 1)
+        logger.log('DEBUG', f'结束执行{self.source}检查{self.domain}的域传送漏洞')
+        self.save_json()
+        self.gen_result()
+        self.save_db()
+        rx_queue.put(self.results)
+
+
+def do(domain, rx_queue):  # 统一入口名字 方便多线程调用
+    """
+    类统一调用入口
+    :param str domain: 域名
+    :param rx_queue: 结果集队列
+    """
+    check = CheckAXFR(domain)
+    check.run(rx_queue)
+    logger.log('INFOR', f'{check.source}模块耗时{check.elapsed}秒发现子域{len(check.subdomains)}个')
+    logger.log('DEBUG', f'{check.source}模块发现的子域 {check.subdomains}')
+
+
+if __name__ == '__main__':
+    # do('ZoneTransfer.me')
+    result_queue = queue.Queue()
+    do('owasp.org', result_queue)
@@ -0,0 +1,64 @@
+# coding=utf-8
+"""
+检查crossdomain.xml文件收集子域名
+"""
+import time
+import queue
+from config import logger
+from common.utils import match_subdomain
+from common.module import Module
+
+
+class CheckCDX(Module):
+    """
+    检查crossdomain.xml文件收集子域名
+    """
+    def __init__(self, domain: str):
+        Module.__init__(self)
+        self.domain = self.register(domain)
+        self.module = 'Check'
+        self.source = "CrossDomainXml"
+
+    def check(self):
+        """
+        检查crossdomain.xml收集子域名
+        :return:
+        """
+        url = f'http://{self.domain}/crossdomain.xml'
+        self.header = self.get_header()
+        self.proxy = self.get_proxy(self.source)
+        resp = self.get(url)
+        if not resp:
+            return
+        self.subdomains = match_subdomain(self.domain, resp.text)
+
+    def run(self, rx_queue):
+        """
+        类执行入口
+        """
+        logger.log('DEBUG', f'开始执行{self.source}检查{self.domain}域的crossdomain.xml')
+        start = time.time()
+        self.check()
+        end = time.time()
+        self.elapsed = round(end - start, 1)
+        self.save_json()
+        self.gen_result()
+        self.save_db()
+        rx_queue.put(self.results)
+        logger.log('DEBUG', f'结束执行{self.source}检查{self.domain}域的crossdomain.xml')
+
+
+def do(domain, rx_queue):  # 统一入口名字 方便多线程调用
+    """
+    类统一调用入口
+    :param domain: 域名
+    :param rx_queue: 结果集队列
+    """
+    check = CheckCDX(domain)
+    check.run(rx_queue)
+    logger.log('INFOR', f'{check.source}模块耗时{check.elapsed}秒发现子域{len(check.subdomains)}个')
+    logger.log('DEBUG', f'{check.source}模块发现的子域 {check.subdomains}')
+
+
+if __name__ == '__main__':
+    do('163.com')
@@ -0,0 +1,70 @@
+#!/usr/bin/env python3
+# coding=utf-8
+
+"""
+检查域名证书收集子域名
+"""
+import ssl
+import time
+import queue
+import socket
+from config import logger
+from common import utils
+from common.module import Module
+
+
+class CheckCert(Module):
+    def __init__(self, domain):
+        Module.__init__(self)
+        self.domain = self.register(domain)
+        self.port = 443  # ssl port
+        self.module = 'Check'
+        self.source = 'CertInfo'
+
+    def check(self):
+        """
+        获取域名证书并匹配证书中的子域名
+        """
+        ctx = ssl.create_default_context()
+        sock = ctx.wrap_socket(socket.socket(), server_hostname=self.domain)
+        try:
+            sock.connect((self.domain, self.port))
+            cert_dict = sock.getpeercert()
+        except Exception as e:
+            logger.log('ERROR', e)
+            return
+        subdomains_find = utils.match_subdomain(self.domain, str(cert_dict))
+        self.subdomains = self.subdomains.union(subdomains_find)
+
+    def run(self, rx_queue):
+        """
+        类执行入口
+        """
+        logger.log('DEBUG', f'开始执行{self.source}检查{self.domain}域的证书中的子域')
+        start = time.time()
+        self.check()
+        end = time.time()
+        self.elapsed = round(end - start, 1)
+        logger.log('DEBUG', f'结束执行{self.source}检查{self.domain}域的证书中的子域')
+        self.save_json()
+        self.gen_result()
+        self.save_db()
+        rx_queue.put(self.results)
+
+
+def do(domain, rx_queue):  # 统一入口名字 方便多线程调用
+    """
+    类统一调用入口
+
+    :param str domain: 域名
+    :param rx_queue: 结果集队列
+    """
+    check = CheckCert(domain)
+    check.run(rx_queue)
+    logger.log('INFOR', f'{check.source}模块耗时{check.elapsed}秒发现子域{len(check.subdomains)}个')
+    logger.log('DEBUG', f'{check.source}模块发现的子域 {check.subdomains}')
+
+
+if __name__ == '__main__':
+    result_queue = queue.Queue()
+    do('owasp.org', result_queue)
@@ -0,0 +1,75 @@
+# coding=utf-8
+"""
+检查内容安全策略收集子域名收集子域名
+"""
+import time
+import queue
+from config import logger
+from common import utils
+from common.module import Module
+
+
+class CheckCSP(Module):
+    """
+    检查内容安全策略收集子域名
+    """
+    def __init__(self, domain, header):
+        Module.__init__(self)
+        self.domain = self.register(domain)
+        self.module = 'Check'
+        self.source = 'Content-Security-Policy'
+        self.header = header
+
+    def check(self):
+        """
+        正则匹配响应头中的内容安全策略字段以发现子域名
+        """
+        if not self.header:
+            url = f'http://www.{self.domain}'
+            resp = self.get(url)
+            if not resp:
+                return
+            self.header = resp.headers
+        csp = self.header.get('Content-Security-Policy')
+        if not csp:
+            logger.log('DEBUG', f'{self.domain}域的响应头不存在内容安全策略字段')
+            return
+        logger.log('DEBUG', f'{self.domain}域的响应头存在内容安全策略字段')
+        self.subdomains = utils.match_subdomain(self.domain, csp)
+
+    def run(self, rx_queue):
+        """
+        类执行入口
+        """
+        logger.log('DEBUG', f'开始执行{self.source}检查{self.domain}域响应头中的内容安全策略字段')
+        start = time.time()
+        self.check()
+        end = time.time()
+        self.elapsed = round(end - start, 1)
+        logger.log('DEBUG', f'结束执行{self.source}检查{self.domain}域响应头中的内容安全策略字段')
+        self.save_json()
+        self.gen_result()
+        self.save_db()
+        rx_queue.put(self.results)
+
+
+def do(domain, rx_queue, header=None):  # 统一入口名字 方便多线程调用
+    """
+    类统一调用入口
+
+    :param str domain: 域名
+    :param rx_queue: 结果集队列
+    :param dict or None header: 响应头
+    """
+    check = CheckCSP(domain, header)
+    check.run(rx_queue)
+    logger.log('INFOR', f'{check.source}模块耗时{check.elapsed}秒发现子域{len(check.subdomains)}个')
+    logger.log('DEBUG', f'{check.source}模块发现的子域 {check.subdomains}')
+
+
+if __name__ == '__main__':
+    import requests
+    # resp = requests.get('https://content-security-policy.com/')
+    result_queue = queue.Queue()
+    resp = requests.get('https://www.baidu.com/')
+    do('google-analytics.com', result_queue, resp.headers)