path: root/server/apps/target.py

# 目标状态感知
# 时延测试接口
import ipaddress
import json
import random
import threading
import asyncio
from operator import or_
from concurrent.futures import ThreadPoolExecutor
import time
from flask import current_app
import requests
from apiflask import APIBlueprint, Schema
from apiflask.fields import Integer, String, List, Nested, IP, DateTime, Dict, Float
from apiflask.validators import OneOf
from requests.exceptions import Timeout
from sqlalchemy import and_, distinct, func, case

from apps.util import debug, is_ipaddress
from exts import db, scheduler
from model import Policy, Target, Task, Agent, TargetStatus, TaskPolicy, Target2

from sklearn.cluster import KMeans, MiniBatchKMeans
import numpy as np

bp = APIBlueprint("目标信息及状态接口集合", __name__, url_prefix="/target")

icmp_delaytable = {}
tcp_delaytable = {}
dns_delaytable = {}

target_map = {
    "ADDRv4": "ipv4",
    "ADDRv6": "ipv6",
    "COU": "cou",
    "ISP": "isp",
    "LAT": "lat",
    "LNG": "lng",
    "UPDATED_TIME": "time"
}


class TargetSchema(Schema):
    ipv4 = String()
    ipv6 = String()
    protocol = List(String())
    protect = String()
    cou = String()
    isp = String()
    lat = Float()
    lng = Float()
    time = DateTime()


class TestNode(Schema):
    Id = String()
    Name = String()
    Ip = String()
    Lat = Float()
    Lng = Float()
    Loc = String()
    # Port = Integer()


class Delay(Schema):
    Id = String()
    CurrDelay = String()
    # MeanDelay=Integer()
    # MaxDelay=Integer()
    Type = String()


class DelayOut(Schema):
    code = Integer()
    delay_data = List(Nested(Delay()))

from marshmallow import Schema, fields

class ProtectSchema(Schema):
    dnssec_enabled = fields.Boolean(description="是否启用 DNSSEC（DNS 安全扩展）")
    _0x20_encoding = fields.Boolean(description="是否使用 0x20 编码（域名是否随机化大小写）")
    merge_dup = fields.Boolean(description="是否支持合并重复的 DNS 记录")
    max_ns_cnt = fields.Integer(description="查询中允许的最大名称服务器（NS）记录数量")
    max_cname_cnt = fields.Integer(description="查询中允许的最大 CNAME 链深度")
    retry_limit = fields.Integer(description="在失败后重试 DNS 查询的最大次数")
    fetch_limit = fields.Integer(description="在一次查询中可以获取的最大资源记录数")
    timeout = fields.Integer(description="DNS 查询的超时时间，单位是毫秒")
    random_port = fields.Boolean(description="是否使用随机端口进行 DNS 查询")
    random_id = fields.Boolean(description="是否使用随机 ID 进行 DNS 查询")
    version = fields.String(description="DNS 软件版本")
    bailiwick = fields.Boolean(description="是否遵循 Bailiwick 规则")
    max_ttl = fields.Integer(description="最大 TTL 值，单位为秒")
    min_ttl = fields.Integer(description="最小 TTL 值，单位为秒")
    edns_support = fields.Boolean(description="是否支持 EDNS")
    rrl_qps = fields.Integer(description="每秒允许的资源记录锁查询次数")

class TestNode(Schema):
    Id = fields.String()
    Name = fields.String()
    Ip = fields.String()
    Lat = fields.Float()
    Lng = fields.Float()
    Loc = fields.String()
    addrv4 = fields.String()
    addrv6 = fields.Boolean()
    ipv6 = fields.Boolean()
    dnssec = fields.Boolean()
    dot = fields.Boolean()
    doh = fields.Boolean()
    cou = fields.String()
    isp = fields.String()
    lat = fields.Float()
    lng = fields.Float()
    protect = fields.Nested(ProtectSchema)



# 地图统计信息返回结构体
class CouInfo(Schema):
    # 国家名称
    name = String()
    # 不同协议的解析器详细数量说明
    title = String()
    # 总数量
    value = Integer()
# from apiflask.fields import Integer, String, List, Nested, Float

class ClusterInfo(Schema):
    lat = Float(required=True)
    lng = Float(required=True)
    count = Integer(required=True)

class TargetInfo(Schema):
    ipv4 = String(allow_none=True)
    ipv6 = String(allow_none=True)
    cou = String()
    isp = String()
    lat = Float()
    lng = Float()
    time = String()
    protocol = List(String())
    protect = String()


def average_without_max(results):
    if results:
        results.remove(max(results))
        return sum(results) / len(results) if results else 0
    return 0



def task_monitoring(task):
    with scheduler.app.app_context():
        # target_status = task.task_policies[-1].target_status
        task = db.session.query(Task).filter_by(task_id = task.task_id).first()
        addr = task.target.addrv4 if task.target.addrv4 else task.target.addrv6

        nodes_info = json.loads(task.SCAN_AGENT_ID_LIST)
        
        # 初始化延迟结果列表
        icmp_results = []
        tcp_results = []
        dns_results = []

        # 执行 ICMP 查询并收集结果
        with ThreadPoolExecutor() as executor:
            for id, ip_port in nodes_info.items():
                future = executor.submit(icmp_delay_query, addr, f"{ip_port}")
                icmp_results.append(float(future.result()))

        # 执行 TCP 查询并收集结果
        with ThreadPoolExecutor() as executor:
            for id, ip_port in nodes_info.items():
                future = executor.submit(tcp_delay_query, addr, f"{ip_port}")
                tcp_results.append(float(future.result()))

        # 执行 DNS 查询并收集结果
        with ThreadPoolExecutor() as executor:
            for id, ip_port in nodes_info.items():
                future = executor.submit(dns_delay_query, addr, f"{ip_port}")
                dns_results.append(float(future.result()))

        # 计算平均值
        icmp_avg = average_without_max(icmp_results)
        tcp_avg = average_without_max(tcp_results)
        dns_avg = average_without_max(dns_results)
        # id, ip_port = nodes_info.items()[0]
        query_data = {
            'rev' : task.target.addrv4 if task.target.addrv4 else task.target.addrv6,
            'domain' : task.target_domain,
            'qtype' :'A' if task.target.addrv4 else "AAAA"
        }
        target_domain = get_record(query_data)
        try:
            ip_list = str([entry["rrset"] for entry in target_domain])
        except:
            ip_list = ""
        target_status = TargetStatus(
            tp_id = task.task_policies[-1].tp_id,
            icmp = icmp_avg,
            tcp = tcp_avg,
            dns = dns_avg,
            recorde = ip_list,
        )

        db.session.add(target_status)
        db.session.commit()
        task = db.session.query(Task).get(task.task_id)
        last_task_policy = task.task_policies[-1]

        # 现在可以安全地访问 target_status
        target_status = last_task_policy.target_status

def dida_task(task, ):
    from .task import effective_detection, finish_task
    from .task import adjust_task
    with scheduler.app.app_context():
        task = db.session.query(Task).filter_by(task_id = task.task_id).first()
        task_policy=db.session.query(TaskPolicy).get(task.task_policies[-1].tp_id)
        task_monitoring(task)
        # 如果任务没有成功
        if not effective_detection(task_policy=task_policy):
            debug(task.status)
            # 识别后自动切换
            if task.status == "stopped":
                adjust_task(task=task)
        else:
            finish_task(task)
            pass
    debug("didadida")

def start_task_monitoring(task):
    
    with scheduler.app.app_context():
        scheduler.add_job(
            func=dida_task,  # 要执行的函数
            trigger='interval',  # 触发器类型为间隔
            args = (task, ),  # 传递给函数的参数
            id = task.task_id,  # 任务的唯一标识符
            seconds = 30,  # 触发器的参数，表示每 5 秒执行一次
            max_instances = 100
        )

def stop_task_monitoring(task):
    scheduler.remove_job(task.task_id)


@bp.get("/nodes")
@bp.doc("参与探测的节点信息获取接口", "传入任务编号taskid")
@bp.input({
    "taskid": String(required=True)
}, location="query")
@bp.output({
    "code": Integer(),
    "nodes": List(Nested(TestNode()))
})
def get_nodes(query_data):
    node_list = []
    query_session = db.session
    # 查询负责该任务的所有状态感知节点简要信息
    node_data = query_session.query(Task.SCAN_AGENT_ID_LIST).filter(Task.task_id == query_data['taskid']).all()

    # 存在数据
    if len(node_data) > 0:
        # 负责该任务目标的状态感知的节点
        nodes_info = json.loads(node_data[0].SCAN_AGENT_ID_LIST)
        # 查询每一个节点的信息
        nodes = query_session.query(Agent).filter(Agent.agent_id.in_(nodes_info.keys())).all()
        for node in nodes:
            node_list.append({
                "Id": node.agent_id,
                "Name": "编号" + str(node.agent_id),
                "Ip": nodes_info[node.agent_id],
                # TODO:根据经纬度调整location的值
                "Loc": node.con,
                "Lat": float(node.lat),
                "Lng": float(node.lng)
            })
        query_session.close()
    else:
        query_session.close()
        return {"code": 500, "nodes": []}

    return {"code": 200, "nodes": node_list}


@bp.get("/delay")
@bp.doc("获取一个目标每个对应探测节点的时延数据", "type参数为{icmp,dns,tcp}中的一个")
@bp.input({
    "target": IP(required=True),
    "taskid": String(required=True),
    "type": String(required=True, validate=OneOf(['icmp', 'dns', 'tcp']))
    }, location="query")
@bp.output(DelayOut)
def get_pernode_delay(query_data):
    query_session = db.session
    # TODO:DoH处理
    # 探测地址
    addr = query_data['target']
    taskid = query_data['taskid']
    scan_type = query_data['type']

    # 响应数据
    ans = []
    # 线程池
    threads = []

    # 检索探测节点信息
    # 查询负责该任务的所有状态感知节点简要信息
    node_data = query_session.query(Task.SCAN_AGENT_ID_LIST).filter(Task.task_id == taskid).all()

    # 存在数据
    if len(node_data) > 0:
        # 负责该任务目标的状态感知的节点ID和IP地址+端口号组成的列表
        nodes_info = json.loads(node_data[0].SCAN_AGENT_ID_LIST)
        # 向每个节点查询时延信息
        for id, ip_port in nodes_info.items():
            mythread = threading.Thread(target=task, args=[ans, addr, {'id': id, 'ip_port': ip_port}, scan_type])
            mythread.start()
            threads.append(mythread)
        for t in threads:
            t.join()
        query_session.close()
    else:
        query_session.close()
        return {"code": 500, "delay_data": []}

    return {"code": 200, 'delay_data': ans}


threadLock = threading.Lock()


def task(ans, addr, agent, scan_type):
    res = 0
    if scan_type == "icmp":
        res = icmp_delay_query(addr, agent['ip_port'])
    if scan_type == "tcp":
        res = tcp_delay_query(addr, agent['ip_port'])
    if scan_type == "dns":
        res = dns_delay_query(addr, agent['ip_port'])
    threadLock.acquire()
    ans.append({
        'Id': agent['id'],
        'CurrDelay': res,
        'Type': scan_type})
    threadLock.release()


def icmp_delay_query(target, addr_port):
    try:
        res = requests.get(url="http://" + addr_port + "/delay/icmp", params={'ip': target}, timeout=5)
        debug("icmp ok:" + addr_port + "-------" + res.text + "-------" + str(res.elapsed.total_seconds()))
        icmp_delaytable[str(addr_port) + str(target)] = res.text
        return res.text
    except Exception:
        return 5000+random.randint(-300,300)


def tcp_delay_query(target, addr_port):
    try:
        res = requests.get(url="http://" + addr_port + "/delay/tcp", params={'ip': target, 'port': 53}, timeout=5)
        debug("tcp ok:" + addr_port + "-------" + res.text)
        tcp_delaytable[str(addr_port) + str(target)] = res.text
        return res.text
    except Exception:
        return 5000+random.randint(-300,300)


def dns_delay_query(target, addr_port):
    try:
        res = requests.get(url="http://" + addr_port + "/delay/dns", params={'ip': target}, timeout=5)
        # debug(res.text)
        debug("dns ok:" + addr_port + "-------" + str(float(res.text)))
        # debug("dns ok:" + addr_port + "-------" + res.text)
        dns_delaytable[str(addr_port) + str(target)] = res.text
        return dns_delaytable[str(addr_port) + str(target)]
    except Exception as e:
        return 5000+random.randint(-300,300)

# 状态感知——DNS记录测试接口
import dns.nameserver
from apiflask.fields import String
from apiflask.validators import OneOf, ContainsOnly
from dns import resolver

def get_record(query_data):
 # 特殊协议头
    protols = ["https", "tls"]
    ans = []
    # 参数读取
    rev = query_data['rev']
    domain = query_data['domain']
    qtype = query_data['qtype']

    # 解析器配置
    myResolver = resolver.Resolver()
    # 根据rev参数配置解析器
    is_Do53 = any(proto if proto in rev else False for proto in protols)
    if not is_Do53:
        # # 存在端口指定
        # if ":" in rev:
        #     res = rev.split(":")
        #     myResolver.nameservers = [dns.nameserver.Do53Nameserver(res[0], port=res[1])]
        # else:
        myResolver.nameservers = [rev]
    else:
        #  doh服务
        if "https" in rev:
            myResolver.nameservers = [dns.nameserver.DoHNameserver(url=rev)]
        # dot服务
        if "tls" in rev:
            res = rev.split("//")
            myResolver.nameservers = [dns.nameserver.DoTNameserver(address=res[1])]
    myAnswers = myResolver.resolve(domain, qtype)
    if myAnswers.rrset is not None:
        for r in myAnswers.rrset:
            ans.append({"rrset": str(r)})
    return ans

@bp.get("/target_check")
@bp.doc("展示数据库中的目标解析结果", description="参数说明:</br>" +
                                                "rev:解析器的IP地址</br>"
                                                + "domain:查询的目标域名</br>"
                                                + "qtype:查询的记录类型")
@bp.input({
    "taskid": String(required=True),
}, location='query')
@bp.output({
    "code": Integer(),
    "ans": List(Dict(String(validate=ContainsOnly(["rrset"])), String()))
})

def get_target_record(query_data):
    task_id = query_data["taskid"]
    
    base_policy = db.session.query(TaskPolicy).filter(TaskPolicy.for_task==task_id).order_by(TaskPolicy.tp_time.desc()).first()
    now = db.session.query(TargetStatus).filter_by(tp_id=base_policy.tp_id).order_by(TargetStatus.time.desc()).first()
    res = {
        "ans": [
            {
                "rrset": now.recorde
            }
        ]
    }
    return {"code": 200, 'ans': res["ans"]}

@bp.get("/check")
@bp.doc("通过指定的解析器获取指定域名的A/AAAA记录", description="参数说明:</br>" +
                                                                "rev:解析器的IP地址</br>"
                                                                + "domain:查询的目标域名</br>"
                                                                + "qtype:查询的记录类型")
@bp.input({
    'rev': String(required=True),
    'domain': String(required=True),
    'qtype': String(required=True, validate=OneOf(['A', 'AAAA', "CNAME", "NS"]))
}, location='query')
@bp.output({
    "code": Integer(),
    "ans": List(Dict(String(validate=ContainsOnly(["rrset"])), String()))
})
def record(query_data):
   ans = get_record(query_data)
   return {"code": 200, 'ans': ans}


@bp.get("/")
@bp.doc("(表格)目标信息获取接口", "返回目标信息")
@bp.input({
    "protocol": String(load_default=None, validate=OneOf(['IPv6', 'DNSSEC', "DoH", "DoT"])),
    "cou": String(load_default=None),
    "isp": String(load_default=None),
    "per_page": Integer(load_default=10),
    "page": Integer(load_default=1),
    "ip": IP(load_default=None)
}, location='query')
@bp.output({
    "code": Integer(),
    "data": List(Nested(TargetSchema())),
    "total": Integer()
})
def target_info(query_data):
    per_page = query_data["per_page"]
    page = query_data["page"]
    proto = query_data["protocol"]
    cou = query_data["cou"]
    isp = query_data["isp"]
    ip = query_data["ip"]

    # 目标信息列表
    target_list = []
    query = db.session.query(Target)
    # 普通检索，默认所有条件为单选
    if ip is None:
        # 分页查询
        res = query.filter(
            or_(Target.ipv6 == True, proto != 'IPv6'),
            or_(Target.dnssec == True, proto != 'DNSSEC'),
            or_(Target.doh == True, proto != 'DoH'),
            or_(Target.dot == True, proto != 'DoT'),
            or_(Target.cou == cou, cou == None),
            or_(Target.isp == isp, isp == None)).paginate(page=page, per_page=per_page)
        # 查询总数
        res_count = res.total
    else:
        # 查询目标，根据v4、v6地址分类
        res = query.filter(
            or_(Target.addrv6 == ip, ipaddress.ip_address(ip).version != 6),
            or_(Target.addrv4 == ip, ipaddress.ip_address(ip).version != 4)).all()
        # 查询总数
        res_count = len(res)
    # 结果转换
    for r in res:
        target = {
            "ipv4": r.addrv4,
            "ipv6": r.addrv6,
            "cou": r.cou,
            "isp": r.isp,
            "lat": r.lat,
            "lng": r.lng,
            "time": r.updated_time
        }

        # 支持的协议特殊处理
        protocol = []
        # 目标各协议支持情况
        proto_state = {"IPv6": r.ipv6, "DNSSEC": r.dnssec, "DoH": r.doh, "DoT": r.dot}
        for k, v in proto_state.items():
            if bool(v) is True:
                protocol.append(k)
        target["protocol"] = protocol

        # 防护措施特殊处理
        target["protect"] = json.dumps(r.protect)
        target_list.append(target)
    return {"code": 200, "data": target_list, "total": res_count}


@bp.get("/filter")
@bp.doc("目标可筛选信息获取接口")
@bp.output({
    "code": Integer(),
    "proto": List(String()),
    "isp": List(String()),
    "cou": List(String())
})
def filter_info():
    proto = ["IPv6", "DNSSEC", "DoH", "DoT"]

    query_session = db.session
    # 查询所有的isp
    isp_data = query_session.query(distinct(Target.isp)).all()
    # 执行查询
    isp = [i[0] for i in isp_data]

    # 查询所有的国家
    cou_data = query_session.query(distinct(Target.cou)).all()
    # 执行查询
    cou = [i[0] for i in cou_data]

    query_session.close()
    return {"code": 200, "proto": proto, "isp": isp, "cou": cou}


@bp.get("/map")
@bp.doc("地图信息获取接口")
@bp.input({
    "zoom_level": Integer(),
    "min_lat": Float(required=False),
    "max_lat": Float(required=False),
    "min_lng": Float(required=False),
    "max_lng": Float(required=False)
}, location="query")
# @bp.output({ 
#     "code": Integer(),
#     "data": List(Nested(ClusterInfo() if query_data.get("zoom_level") < 10 else TargetInfo())),
#     "total": Integer(),
# })

def map_info(query_data):
    zoom_level = query_data.get("zoom_level")  # 从前端获取缩放级别
    min_lat = query_data.get("min_lat")  # 获取最小纬度
    max_lat = query_data.get("max_lat")  # 获取最大纬度
    min_lng = query_data.get("min_lng")  # 获取最小经度
    max_lng = query_data.get("max_lng")  # 获取最大经度
    if zoom_level == 0:
        res = db.session.query(Target.target_id, Target.lat, Target.lng).all()
        res_dict = [{"target_id": row.target_id, "lat": row.lat, "lng": row.lng} for row in res]
        return {"code": 200, "data": res_dict, "total": len(res_dict)}
    if zoom_level <= 10:

        # 查询目标数据
        query = db.session.query(Target.lat, Target.lng).filter(
            and_(
                Target.lat >= min_lat,
                Target.lat <= max_lat,
                Target.lng >= min_lng,
                Target.lng <= max_lng
            )
        )

        # 查询符合条件的数据
        res = query.all()

        # 结果数量
        res_count = len(res)

        # 如果没有结果，返回空数据
        if res_count == 0:
            return {"code": 200, "data": [], "total": 0}

        # 处理数据，提取经纬度信息
        lat_lng = np.array([[float(r.lat), float(r.lng)] for r in res])
        # 聚类操作，当 zoom_level 小于 10 时进行聚类
        num_clusters = max(1, int(res_count / (30000 / zoom_level)))  # 根据缩放级别动态调整聚类数量
        kmeans = MiniBatchKMeans(n_clusters=num_clusters, n_init=5, batch_size=1000)
        kmeans.fit(lat_lng)
        centers = kmeans.cluster_centers_

        # 聚类结果转换
        clustered_data = []
        for i, center in enumerate(centers):
            clustered_data.append({
                "lat": center[0],
                "lng": center[1],
                "count": int(sum(kmeans.labels_ == i))  # 统计每个聚类的节点数量
            })

        return {"code": 200, "data": clustered_data}

    else:
        # 当 zoom_level 大于等于 10 时，返回详细信息
        r = db.session.query(Target).filter(
            and_(
                Target.lat >= min_lat,
                Target.lat <= max_lat,
                Target.lng >= min_lng,
                Target.lng <= max_lng
            )
        )
        target_list = []
        for r_item in r:
            target = {
                "ipv4": r_item.addrv4,
                "ipv6": r_item.addrv6,
                "cou": r_item.cou,
                "isp": r_item.isp,
                "lat": r_item.lat,
                "lng": r_item.lng,
                "time": r_item.updated_time
            }

            protocol = []
            proto_state = {"IPv6": r_item.ipv6, "DNSSEC": r_item.dnssec, "DoH": r_item.doh, "DoT": r_item.dot}
            for k, v in proto_state.items():
                if bool(v):
                    protocol.append(k)
            target["protocol"] = protocol
            target["protect"] = json.dumps(r_item.protect)

            target_list.append(target)

        return {"code": 200, "data": target_list}

    

@bp.get("/gz")
@bp.doc("目标感知")
@bp.input({
    "ip": String(required=True)
}, location="query")
# @bp.output({
#     "code": Integer(),
#     "dataObject": Nested(TargetSchema())
# })
def target_GZ_API(query_data):
    ip = query_data["ip"]
    target = target_GZ(ip)
    target_dict = {
            "addrv4": target.IP,
            "addrv6": target.IPv6,
            "protocol":{
                "Do53": target.Do53,
                "DoH": target.DoH,
                "DoT": target.DoT
            },
            "lat": target.latwgs,
            "lng": target.lngwgs,
            "parameter":{
                "tls_jarm_fingerprint": target.tls_jarm_fingerprint,  # TLS指纹
                "certificate_serial": target.certificate_serial,  # 证书序列号
                "certificate_issuer": target.certificate_issuer,  # 证书颁发者
                "certificate_cn": target.certificate_cn,  # 证书CN/SAN
                "certificate_san": target.certificate_san,
                "certificate_validity": target.certificate_validity,  # 证书有效性
                "http_max_version": target.http_max_version,  # HTTP最大版本
                "service_path": target.service_path,  # 服务路径
                "request_method": target.request_method,  # 请求方法
                "http_status_code": target.http_status_code,  # 状态码
                "http_version": target.http_version,  # HTTP版本
                "http_headers": target.http_headers,  # HTTP头信息
                "nsec3_support": target.nsec3_support,  # NSEC3支持
                "encryption_algorithms": target.encryption_algorithms,  # 加密算法
                "downgrade_attack": target.downgrade_attack,  # 可降级攻击
                "dns_response_packet": target.dns_response_packet,
                "edns": target.edns,  # EDNS支持
                "merge_dup": target.merge_dup,  # 合并重复请求
                "dnssec": target.dnssec,  # DNSSEC支持
                "0x20": target.proto0x20,  # 0x20编码
                "any_response": target.any_response,  # 最大并发响应
                "max_ns_cnt": target.max_ns_cnt,  # 最大委派上限
                "max_cname_cnt": target.max_cname_cnt,  # 最大重定向上限
                "rrl": target.rrl,  # 响应速率限制
                "bailiwick": target.bailiwick,  # Bailiwick验证
                "random_id": target.random_id,  # 随机ID
                "random_port": target.random_port,  # 随机端口
                "version": target.version,  # 版本
                "max_ttl": target.max_ttl,  # 最大TTL
                "min_ttl": target.min_ttl,  # 最小TTL
                "retry_limit": target.retry_limit,  # 重试次数
                "timeout": target.timeout,  # 超时时间
                "fetch_limit": target.fetch_limit  # 最大查询限制
            }

        }
    return {
        'code': 200,
        'dataObject': target_dict
    }

def target_GZ(IP_addr):

    existing_obj = db.session.query(Target2).filter(
        (Target2.IP == IP_addr) | (Target2.IPv6 == IP_addr)
    ).first()
    if existing_obj:
        return existing_obj

    ipv6 = None
    ipv4 = None

    # 判断 IP 地址的版本并存储
    if 6 == is_ipaddress(IP_addr):
        ipv6 = IP_addr
    elif 4 == is_ipaddress(IP_addr):
        ipv4 = IP_addr

    # 获取随机的 agent
    csgz = db.session.query(Agent).filter(Agent.agent_type=='gjst').filter(Agent.agent_id=="inrz674e").all()
    csgz = random.choice(csgz)

    # 根据 IP 地址类型构建 URL
    if ipv6:
        # IPv6 地址需要加上中括号
        url = f"http://{csgz.ipaddr}:{csgz.port}/target_gz/[{ipv6}]"
    elif ipv4:
        url = f"http://{csgz.ipaddr}:{csgz.port}/target_gz/{ipv4}"
    else:
        raise ValueError("Invalid IP address")

    # 发送请求
    i = 0
    while i < 30:
        data = requests.get(url, timeout = 180)
        status_code = data.status_code
        debug(f"目标感知：重试{i}次")
        i+=1
        if status_code == 200:
            break
        else:
            time.sleep(0.5)
    data = data.json()
    # 存数据库
    target = Target2(
        IP=data['IP'],
        Do53=data['Do53'],
        DoH=data['DoH'],
        DoT=data['DoT'],
        dnssec=data['dnssec'],
        merge_dup=data['merge_dup'],
        max_ns_cnt=data['max_ns_cnt'],
        max_cname_cnt=data['max_cname_cnt'],
        retry_limit=data['retry_limit'],
        fetch_limit=data['fetch_limit'],
        timeout=data['timeout'],
        random_port=bool(data['random_port']),
        random_id=bool(data['random_id']),
        version=data['version'],
        bailiwick=data['bailiwick'],
        max_ttl=data['max_ttl'],
        min_ttl=data['min_ttl'],
        edns=data['edns'],
        any_response=data['any_response'],
        rrl=data['rrl'],
        service_path=data['service_path'],
        http_max_version=data['http_max_version'],
        request_method=data['request_method'],
        http_status_code=data['http_status_code'],
        http_version=data['http_version'],
        http_headers=data['http_headers'],
        web_component=data['web_component'],
        tls_jarm_fingerprint=data['tls_jarm_fingerprint'],
        certificate_serial=data['certificate_serial'],
        certificate_issuer=data['certificate_issuer'],
        certificate_cn=data['certificate_cn'],
        certificate_san=data['certificate_san'],
        certificate_validity=data['certificate_validity']
    )
    db.session.add(target)
    db.session.commit()
    return target