552 lines
21 KiB
Python
552 lines
21 KiB
Python
|
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()
|