排除CDN的小技巧 附实现代码

工具介绍

工具来源：ybdt师傅的分享

实现比较简单，附件是自动化工具的一部分，摘选出来，可能代码不是很完整，期待师傅们更好的思路~

基本思路

ping子域名，处理ping后返回的数据，返回的域名不变，则暂定为未使用CDN，返回的域名变了，则往往使用CDN或云WAF，由于每个ping比较慢，最慢也得1秒，所以需要采用队列+多线程

核心代码

cdn-exclude.py

# 基于ping判断是否为cdn节点或云waf节点
def work(q, cdn_list, non_cdn_list, non_parse_list, non_cdn_ip_list):
    while True:
        if q.empty():
            return
        else:
            domain = q.get()
            # print(domain)
            p = subprocess.Popen(["ping.exe", "-n", "1", "-w", "1000", domain], shell=True, stdout=subprocess.PIPE,
                                 stderr=subprocess.PIPE)
            ping_str = p.stdout.read()
            # print(ping_str)
            ping_list = ping_str.decode("GBK").split("\r\n")
            # print(ping_list)
            # 奇怪的是，在pycharm下返回中文，在ConEmu下返回英文，故需要兼容
            if ping_list[0].startswith("Ping request could not find host") or ping_list[0].startswith("Ping 请求找不到主机"):
                # print("Ping 请求找不到主机 " + domain)
                non_parse_list.append(domain + "__" + "Ping 请求找不到主机")
                continue
            else:
                ping_return_domain = ping_list[1].split(" ")[1]
                ping_return_ip = ping_list[1].split(" ")[2].strip("[").strip("]")
                # print(ping_return_domain)
                if ping_return_domain == domain:
                    non_cdn_list.append(domain + "__" + ping_return_ip)
                    non_cdn_ip_list.append(ping_return_ip)
                else:
                    cdn_list.append(domain + "__" + ping_return_domain)

def main():
    # 创建队列并填充消息
        domain_list = []
        with open("domain.txt", "r", encoding="UTF-8") as fr:
            for line in fr.readlines():
                domain_list.append(line.strip("\n"))
        q = queue.Queue()
        for domain in domain_list:
            q.put(domain)

        # 多线程执行队列中的消息
        cdn_list = []
        non_cdn_list = []
        non_cdn_ip_list = []
        non_parse_list = []
        thread_num = 10
        threads = []
        for i in range(thread_num):
            t = threading.Thread(target=work, args=(q, cdn_list, non_cdn_list, non_parse_list, non_cdn_ip_list))
            threads.append(t)
            t.start()
        for t in threads:
            t.join()

main()

 

标签：工具分享, 资产测绘