server端仪表盘界面接口定义完毕

1 files changed, 185 insertions, 0 deletions

diff --git a/server/apps/target.py b/server/apps/target.py
new file mode 100644
index 0000000..50e4908
--- /dev/null
+++ b/server/apps/target.py
@@ -0,0 +1,185 @@
+# 目标状态感知
+
+# 时延测试接口
+import threading
+
+import pandas as pd
+import requests
+from apiflask import APIBlueprint, Schema
+from apiflask.fields import Integer, String, List, Nested, IP
+from requests.exceptions import Timeout
+
+bp = APIBlueprint("target", __name__, url_prefix="/target")
+
+icmp_delaytable = {}
+tcp_delaytable = {}
+dns_delaytable = {}
+
+
+class TestNode(Schema):
+    Id = Integer()
+    Name = String()
+    Ip = String()
+    Loc = String()
+    Port = Integer()
+
+
+class Delay(Schema):
+    Id = Integer()
+    CurrDelay = String()
+    # MeanDelay=Integer()
+    # MaxDelay=Integer()
+    Type = String()
+
+
+class DelayOut(Schema):
+    delay_data = List(Nested(Delay))
+
+
[email protected]("/delay/<string:type>")
[email protected]("获取每个节点的时延数据", "type参数为{icmp,dns,tcp}中的一个")
[email protected]({"ip": IP(required=False)}, location="query")
[email protected](DelayOut)
+def get_pernode_delay(query_data, type):
+    addr = ""
+    if 'ip' in query_data.keys():
+        addr = query_data['ip']
+    ans = []
+    threads = []
+    df = pd.read_csv("./server.csv", encoding="utf-8")
+    for index, row in df.iterrows():
+        mythread = threading.Thread(target=task, args=[ans, addr, row, type])
+        mythread.start()
+        threads.append(mythread)
+    for t in threads:
+        t.join()
+    return {'delay_data': ans}
+
+
+threadLock = threading.Lock()
+
+
+def task(ans, addr, row, type):
+    if type == "icmp":
+        res = icmp_delay_query(addr, row['ip'])
+        threadLock.acquire()
+        ans.append({
+            'Id': row['id'],
+            'CurrDelay': res,
+            'Type': type})
+        threadLock.release()
+        return
+    if type == "tcp":
+        res = tcp_delay_query(addr, row['ip'])
+        threadLock.acquire()
+        ans.append({
+            'Id': row['id'],
+            'CurrDelay': res,
+            'Type': type})
+        threadLock.release()
+        return
+    if type == "dns":
+        res = dns_delay_query(addr, row['ip'])
+        threadLock.acquire()
+        ans.append({
+            'Id': row['id'],
+            'CurrDelay': res,
+            'Type': type})
+        threadLock.release()
+
+
+def icmp_delay_query(target, addr):
+    try:
+        res = requests.get(url="http://" + addr + ":2525/script/icmpdelay", params={'ip': target}, timeout=5)
+        print("icmp ok:" + addr + "-------" + res.text + "-------" + str(res.elapsed.total_seconds()))
+        icmp_delaytable[str(addr) + str(target)] = res.text
+        return res.text
+    except Timeout:
+        # 如果存在旧数据
+        if str(addr) + str(target) in icmp_delaytable.keys():
+            pass
+        # 不存在则设0
+        else:
+            icmp_delaytable[str(addr) + str(target)] = 0
+        return icmp_delaytable[str(addr) + str(target)]
+
+
+def tcp_delay_query(target, addr):
+    try:
+        res = requests.get(url="http://" + addr + ":2525/script/tcpdelay", params={'ip': target, 'port': 53}, timeout=5)
+        print("tcp ok:" + addr + "-------" + res.text)
+        tcp_delaytable[str(addr) + str(target)] = res.text
+        return res.text
+    except Timeout:
+        # 如果存在旧数据
+        if str(addr) + str(target) in tcp_delaytable.keys():
+            pass
+        # 不存在则设0
+        else:
+            tcp_delaytable[str(addr) + str(target)] = 0
+        return tcp_delaytable[str(addr) + str(target)]
+
+
+def dns_delay_query(target, addr):
+    try:
+        res = requests.get(url="http://" + addr + ":2525/script/dnsdelay", params={'ip': target}, timeout=5)
+        print("dns ok:" + addr + "-------" + res.text)
+        dns_delaytable[str(addr) + str(target)] = res.text
+        return dns_delaytable[str(addr) + str(target)]
+    except Timeout:
+        # 如果存在旧数据
+        if str(addr) + str(target) in dns_delaytable.keys():
+            pass
+        # 不存在则设0
+        else:
+            dns_delaytable[str(addr) + str(target)] = 0
+        return dns_delaytable[str(addr) + str(target)]
+
+
+# 状态感知——DNS记录测试接口
+import dns.nameserver
+from apiflask.fields import String
+from apiflask.validators import OneOf
+from dns import resolver
+
+
[email protected]("/check")
[email protected]("通过指定的解析器获取指定域名的A/AAAA记录")
[email protected]({
+    'rev': String(required=True),
+    'domain': String(required=True),
+    'qtype': String(required=True, validate=OneOf(['A', 'AAAA', "CNAME", "NS"]))
+}, location='query')
+def 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': ans}