from util.util_http import build_dns_query, make_ssl_context, get_domain_from_cert from dns.message import from_wire import dns from dns.name import BadLabelType from dns.exception import FormError import time import configparser import pandas as pd from util.util_kafka import make_kafka_producer, kafka_topic_fail, kafka_topic_cert, kafka_topic_suc, make_seed_consumer from hyper import HTTP20Connection, HTTP11Connection from hyper.http20.exceptions import ConnectionError, StreamResetError import socket, h2.exceptions, ssl from util.concurrent.futures import ThreadPoolExecutor import threading from util.thread_timeout import time_limited import os import json from hyper.http20.response import HTTPHeaderMap from hyper.http11.parser import ParseError import zlib import sys import csv from _csv import writer # 定义except的错误信息，遇到这类错误需要特例处理 NOT_SUPPORT_FOR_HTTP2 = "No suitable protocol found" CERTIFICATE_VERIFY_FAILED = "[SSL: CERTIFICATE_VERIFY_FAILED]" # HTTP连接类型 UN_CONNECTED = 0 CONNECTED_HTTP1_1 = 1 CONNECTED_HTTP2_0 = 2 # 超时基数 TIMEOUT = 2 def write_data(data: dict, kafka_topic: str = ""): lock.acquire() if write_model == "csv": data["status"] = kafka_topic writer.writerow([data.__str__()]) else: try: writer.send(kafka_topic, data) except Exception as e: ... lock.release() def make_doh_request_body(q_name="baidu.com"): dns_query = build_dns_query(q_name) dns_query_wire_format = dns_query.to_wire() return dns_query_wire_format def make_doh_request_header(hostname=None, conn_status=CONNECTED_HTTP2_0): headers = { "content-type": "application/dns-message", "accept": "application/dns-message", "user-agent": 'Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/36.0.1985.125 Safari/537.36 ' } if hostname: if conn_status == CONNECTED_HTTP2_0: headers[":authority"] = hostname elif conn_status == CONNECTED_HTTP1_1: headers["host"] = hostname else: ... return headers def decode_header(header: HTTPHeaderMap): # print(header.__str__()) ret = {} for key, value in header.items(): key = key.decode() try: value = value.decode() except UnicodeDecodeError: key = "UnicodeDecodeError" if key not in ret.keys(): # 解决单key,多value情况 ret[key] = value else: ret[key] += (";" + value) return ret def probe_http2_0(conn: HTTP20Connection, host_group, path_group, methods): stream_ids = {} for hostname in host_group: headers = make_doh_request_header(hostname) for path in path_group: for method in methods: if len(stream_ids.keys()) >= 100: # 单IP、port限制探测100次 continue if method == "GET": full_path = "%s?dns=DUIBAAABAAAAAAAABWJhaWR1A2NvbQAAAQAB" % path try: stream_id = conn.request(method=method, url=full_path, headers=headers) except Exception as e: continue else: try: stream_id = conn.request(method=method, url=path, body=body, headers=headers) except Exception as e: continue stream_ids[stream_id] = (hostname, path, method) for stream_id, conn_info in stream_ids.items(): hostname, path, method = conn_info try: rep = conn.get_response(stream_id) except Exception as e: # produce failInfo to kafka data = { "ip": conn.host, "port": conn.port, "host": hostname, "path": path, "method": method, "connect_type": CONNECTED_HTTP2_0, "info": e.__str__(), } write_data(data, kafka_topic_fail) continue rep_code = rep.status rep_header = decode_header(rep.headers) try: rep_body = rep.read() except Exception as e: # produce failInfo to kafka data = { "ip": conn.host, "port": conn.port, "host": hostname, "path": path, "method": method, "connect_type": CONNECTED_HTTP2_0, "status_code": rep_code, "rep_header": rep_header, "info": e.__str__() } write_data(data, kafka_topic_fail) return # print(rep_body) try: rep_body = from_wire(rep_body) if rep_body: # produce sucInfo to kafka data = { "ip": conn.host, "port": conn.port, "host": hostname, "path": path, "method": method, "connect_type": CONNECTED_HTTP2_0, "status_code": rep_code, "rep_header": rep_header, "rep_body": rep_body.__str__() } # print(rep_body.__str__()) write_data(data, kafka_topic_suc) else: # produce failInfo to kafka data = { "ip": conn.host, "port": conn.port, "host": hostname, "path": path, "method": method, "connect_type": CONNECTED_HTTP2_0, "status_code": rep_code, "rep_header": rep_header, "info": rep_body.__str__() } write_data(data, kafka_topic_fail) except Exception: # produce failInfo to kafka data = { "ip": conn.host, "port": conn.port, "host": hostname, "path": path, "method": method, "connect_type": CONNECTED_HTTP2_0, "status_code": rep_code, "rep_header": rep_header, "info": rep_body.__str__() } write_data(data, kafka_topic_fail) def probe_http1_1(conn: HTTP11Connection, host_group, path_group, methods): # start_time = time.time() for hostname in host_group: headers = make_doh_request_header(hostname, CONNECTED_HTTP1_1) for path in path_group: for method in methods: if method == "GET": full_path = "%s?dns=DUIBAAABAAAAAAAABWJhaWR1A2NvbQAAAQAB" % path try: conn.request(method=method, url=full_path, headers=headers) except Exception as e: # produce failInfo to kafka data = { "ip": conn.host, "port": conn.port, "host": hostname, "path": path, "method": method, "connect_type": CONNECTED_HTTP1_1, "info": e.__str__(), } write_data(data, kafka_topic_fail) # print("GET方法执行失败，错误信息如下：", e.__str__) continue # 此处直接 else: try: conn.request(method=method, url=path, body=body, headers=headers) except Exception as e: # produce failInfo to kafka data = { "ip": conn.host, "port": conn.port, "host": hostname, "path": path, "method": method, "connect_type": CONNECTED_HTTP1_1, "info": e.__str__(), } write_data(data, kafka_topic_fail) # print("POST方法执行失败，错误信息如下：", e.__str__) continue try: rep = conn.get_response() except Exception as e: # produce failInfo to kafka data = { "ip": conn.host, "port": conn.port, "host": hostname, "path": path, "method": method, "connect_type": CONNECTED_HTTP1_1, "info": e.__str__(), } write_data(data, kafka_topic_fail) # print("get_response方法执行失败，错误信息如下：", e) continue rep_code = rep.status rep_header = decode_header(rep.headers) try: rep_body = rep.read() except Exception as e: # produce failInfo to kafka data = { "ip": conn.host, "port": conn.port, "host": hostname, "path": path, "method": method, "connect_type": CONNECTED_HTTP1_1, "status_code": rep_code, "rep_header": rep_header, "rep_body": e.__str__() } write_data(data, kafka_topic_fail) # print("read方法执行失败，错误信息如下：", e) continue # print(rep_body) try: rep_body = from_wire(rep_body) # # print(hostname, path, method) # # print(rep_body) if rep_body: # produce sucInfo to kafka data = { "ip": conn.host, "port": conn.port, "host": hostname, "path": path, "method": method, "connect_type": CONNECTED_HTTP1_1, "status_code": rep_code, "rep_header": rep_header, "rep_body": rep_body.__str__() } # print(rep_body.__str__()) write_data(data, kafka_topic_suc) else: # produce failInfo to kafka data = { "ip": conn.host, "port": conn.port, "host": hostname, "path": path, "method": method, "connect_type": CONNECTED_HTTP1_1, "status_code": rep_code, "rep_header": rep_header, "info": None } write_data(data, kafka_topic_fail) except Exception as e: # produce failInfo to kafka data = { "ip": conn.host, "port": conn.port, "host": hostname, "path": path, "method": method, "connect_type": CONNECTED_HTTP1_1, "status_code": rep_code, "rep_header": rep_header, "info": e.__str__() } write_data(data, kafka_topic_fail) # print("from_wire方法执行失败，错误信息如下：", e.__str__) @time_limited(TIMEOUT * 10) def doh_probe(ip, port, host_group=None, path_group=None): verify_mode = ssl.CERT_OPTIONAL ctx_2_0 = make_ssl_context() ctx_2_0.set_alpn_protocols(['h2']) conn = HTTP20Connection(ip, port=port, ssl_context=ctx_2_0) conn_status = UN_CONNECTED try: conn.connect() conn_status = CONNECTED_HTTP2_0 # print("成功建立HTTP2.0连接！") except Exception as e: if e.__str__().startswith(NOT_SUPPORT_FOR_HTTP2): # 预期内的错误，继续尝试HTTP1.1 # print("建立连接失败！服务端不支持HTTP2.0，继续尝试HTTP1.1") ... elif e.__str__().startswith(CERTIFICATE_VERIFY_FAILED): # 证书导致的错误，将sslContext设置为CERT_NONE # print("证书原因导致连接建立失败，设置sslContext为CERT_NONE") ctx_2_0.verify_mode = ssl.CERT_NONE verify_mode = ssl.CERT_NONE conn = HTTP20Connection(ip, port=port, ssl_context=ctx_2_0) try: conn.connect() conn_status = CONNECTED_HTTP2_0 # print("成功建立HTTP2.0连接！") except Exception as e: # print("建立连接失败！服务端不支持HTTP2.0，继续尝试HTTP1.1") ... else: # print("建立连接失败！错误信息如下：", e) # produce failInfo to kafka data = { "ip": ip, "port": port, "connect_type": UN_CONNECTED, "info": e.__str__(), } write_data(data, kafka_topic_fail) return if conn_status == UN_CONNECTED: ctx_1_1 = make_ssl_context() ctx_1_1.verify_mode = verify_mode conn = HTTP11Connection(ip, port=port, ssl_context=ctx_1_1) try: conn.connect() conn_status = CONNECTED_HTTP1_1 # print("成功建立HTTP1.1连接！") except Exception as e: # 如使用HTTP1.1仍然无法建立连接，则放弃 # print("建立连接失败！错误信息如下：", e) # produce failInfo to kafka data = { "ip": ip, "port": port, "connect_type": UN_CONNECTED, "info": e.__str__(), } write_data(data, kafka_topic_fail) return try: cert = conn._sock.getpeercert() # produce certInfo to kafka data = { "ip": ip, "port": port, "certificate": cert.__str__() } write_data(data, kafka_topic_cert) except Exception: cert = None if host_group: host_group += get_domain_from_cert(cert) else: host_group = get_domain_from_cert(cert) host_group += [None, ip] # host_group中至少包含None和原始IP地址 host_group = list(set(host_group)) # 去重 default_path_group = ["/dns-query", "/", "/resolve", "/doh", "/doh/family-filter", "/doh/secure-filter", "/query", "/ads", "/uncensored", "adblock"] if path_group: path_group += default_path_group path_group = list(set(path_group)) # 去重 else: path_group = default_path_group methods = ["POST", "GET"] if conn_status == CONNECTED_HTTP2_0: probe_http2_0(conn, host_group, path_group, methods) elif conn_status == CONNECTED_HTTP1_1: probe_http1_1(conn, host_group, path_group, methods) else: return try: conn.close() except Exception as e: ... def save_status(count): data = {"count": count} with open(save_status_file, "w") as fp: json.dump(data, fp) def load_status(): if not os.path.exists(save_status_file): return 0 with open(save_status_file, "r") as fp: a = json.load(fp) return a["count"] def run_with_csv_model(): chunks = pd.read_csv(seed_filename, iterator=True) count = cpu_num * thread_num * union_detect_num save_point = load_status() count_scan = 0 while True: print(1) start_time = time.time() data_group = [] try: chunk_data = chunks.get_chunk(count) except StopIteration: # 尾部数据丢失 break for data in chunk_data.itertuples(): if count_scan < save_point: # 表示种子已经被探测过 count_scan += 1 continue ip = data.saddr port = data.sport data_group.append((ip, port)) count_scan += 1 with ThreadPoolExecutor(max_workers=thread_num, thread_name_prefix="") as threadPool: for data in data_group: threadPool.submit(doh_probe, *data) if count_scan > save_point: save_status(count_scan) if len(data_group) != 0: end_time = time.time() print(time.strftime('%y%m%d%H', time.localtime(time.time()))) print("CPU核数：", cpu_num, end=" ") print("单核线程数：", thread_num, end=" ") print("本轮完成的探测量为：", len(data_group), end=" ") print("总耗时：", end_time - start_time, end=" ") print("平均每秒探测量", len(data_group) / (end_time - start_time)) def run_with_kafka_model(): consumer = make_seed_consumer() print_num = 10000 # 扫描print_num个IP后打印输出 max_queue_size = thread_num * 10 # 任务队列的长度是线程队列长度的10倍 with ThreadPoolExecutor(max_workers=thread_num, max_queue_size=max_queue_size, thread_name_prefix="") as threadPool: count = 0 start_time = time.time() for message in consumer: data = message.value ip = data["ip"] port = int(data["port"]) threadPool.submit(doh_probe, ip, port) # print("成功从seed中消费数据：", ip, port) count += 1 if count % print_num == 0: end_time = time.time() print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())), end=" ") print("CPU核数：", cpu_num, end=" ") print("单核线程数：", thread_num, end=" ") print("本轮完成的探测量为：", print_num, end=" ") print("总耗时：", end_time - start_time, end=" ") print("平均每秒探测量", print_num / (end_time - start_time)) if __name__ == '__main__': # 定义根目录，方便程序移植 root_dir = sys.path[0] # 读取配置文件 config = configparser.ConfigParser() config_file = os.path.join(root_dir, "config.ini") config.read(config_file) cpu_num = config.getint('DetectConfigs', 'cpu_num', fallback=1) # 程序对CPU要求不高，暂不考虑多进程 thread_num = config.getint('DetectConfigs', 'thread_num', fallback=32) write_model = config.get('DetectConfigs', 'write_model') if write_model == "csv": save_filename = os.path.join(root_dir, config.get('DetectConfigs', 'save_filename')) writer = csv.writer(open(save_filename, "w", newline="")) elif write_model == "kafka": writer = make_kafka_producer() else: print("error write model!") sys.exit() seed_read_model = config.get('DetectConfigs', 'seed_read_model') if seed_read_model == "csv": union_detect_num = config.getint('DetectConfigs', 'union_detect_num', fallback=100) # 单线程单次探测量 seed_filename = os.path.join(root_dir, config.get('DetectConfigs', 'filename')) elif seed_read_model == "kafka": union_detect_num = None seed_filename = None else: print("error seed_read_model!") sys.exit() # 定义探测状态记录文件，用作记录探测进度，防止程序崩溃导致重复探测 save_status_path = os.path.join(root_dir, "./status") save_status_file = os.path.join(save_status_path, "status.json") if not os.path.exists(save_status_path): os.mkdir(save_status_path) lock = threading.Lock() body = make_doh_request_body() # producers = [None] * thread_num print("开始探测：") print("-----", "进程数：", cpu_num) print("-----", "线程数：", thread_num) if seed_read_model == "csv": print("-----", "单线程探测量：", union_detect_num) run_with_csv_model() elif seed_read_model == "kafka": run_with_kafka_model()