diff options
Diffstat (limited to 'server')
| -rw-r--r-- | server/README.md | 24 | ||||
| -rw-r--r-- | server/app.py | 11 | ||||
| -rw-r--r-- | server/apps/agentcomm.py | 232 | ||||
| -rw-r--r-- | server/apps/model.py | 60 | ||||
| -rw-r--r-- | server/apps/sysinfo.py | 68 | ||||
| -rw-r--r-- | server/apps/sysmange.py | 7 | ||||
| -rw-r--r-- | server/apps/target.py | 229 | ||||
| -rw-r--r-- | server/apps/task.py | 53 | ||||
| -rw-r--r-- | server/apps/util.py | 244 | ||||
| -rw-r--r-- | server/requirements.txt | 16 | ||||
| -rw-r--r-- | server/settings.py | 21 |
11 files changed, 826 insertions, 139 deletions
diff --git a/server/README.md b/server/README.md index 4b8b3e7..f81b547 100644 --- a/server/README.md +++ b/server/README.md @@ -2,7 +2,11 @@ ## 简介 -主控端代码，运行app.py将启动一个flask服务，端口为2525。 +主控端代码，运行app.py将启动一个flask服务，默认端口为12525。 +包含三个分支： +main分支，稳定版本 +dev分支，用于根据fake分支完善的接口定义开发实际业务逻辑 +fake分支，用于快速定义新的数据接口与前端对接 访问/docs可以看到swagger-api文档 @@ -16,3 +20,21 @@ | target.py | 目标感知 | | task.py | 任务管理 | +model.py文件包含了一些参数与数据库表字段名或类型之间的映射关系 + +util.py 提供了访问数据库的工具类和日志输出函数 + +## 部署过程 + +1. 首先根据script.sql创建符合要求的数据库表，并在settings.py中配置对应的数据库连接选项 +2. 接下来使用pip 根据requirements.txt安装第三方环境依赖 + +```shell +pip install -r requirements.txt +``` + +3. 启动主控端服务 + +```shell +python app.py +```
\ No newline at end of file diff --git a/server/app.py b/server/app.py index 3c53056..b1ed6a5 100644 --- a/server/app.py +++ b/server/app.py @@ -1,7 +1,7 @@ -import argparse - from apiflask import APIFlask +from flask_cors import CORS +import settings from apps.agentcomm import bp as agentbp from apps.sysinfo import bp as sysbp from apps.sysmange import bp as mangbp @@ -10,6 +10,8 @@ from apps.task import bp as taskbp # 注册蓝图 app = APIFlask(__name__, template_folder='./static/templates') +# 跨域支持 +CORS(app) # 目标状态获取接口 app.register_blueprint(targetbp) # 代理接口 @@ -29,7 +31,4 @@ def hello(): if __name__ == '__main__': - parser = argparse.ArgumentParser() - parser.add_argument("-p", "--port", type=int, default=12525, help="主控服务的开放通信端口") - args = parser.parse_args() - app.run(host="0.0.0.0", debug=True, port=args.port) + app.run(host="0.0.0.0", debug=settings.DEBUG, port=settings.SERVER_PORT) diff --git a/server/apps/agentcomm.py b/server/apps/agentcomm.py index ab50797..4a224ac 100644 --- a/server/apps/agentcomm.py +++ b/server/apps/agentcomm.py @@ -2,18 +2,46 @@ import random from apiflask import APIBlueprint, Schema -from apiflask.fields import String, Integer, List, Nested, Boolean, DateTime +from apiflask.fields import String, Integer, List, Nested, Boolean, DateTime, Float from apiflask.validators import OneOf +from flask import request -from apps.util import fake +from settings import * +from .util import da, error, string_to_mysql bp = APIBlueprint("代理管理接口集合", __name__, url_prefix="/agent") +# 代理类型 +agent_type = ["攻击渗透", "状态感知", "参数感知"] + +# 代理选项参数和代理类型的映射关系 +agent_key_map = {"gjst": agent_type[0], "ztgz": agent_type[1], "csgz": agent_type[2]} + +# 状态选项参数和值的映射关系 +status_map = {1: True, 0: False} +idle_map = {1: True, 0: False} + +# 数据库列与返回值的键对应关系 +agent_response_map = { + "AGENT_ID": "id", + "IPADDR": "ipaddr", + "START_TIME": "start_time", + "LAT": "lat", + "LNG": "lng", + "AGENT_TYPE": "atype", + "SYS": "sys", + "PORT": "port", + "CPU_NUM": "cpu_num", + "STATUS": "status", + "IDLE": "idle", + "MEM": "mem" +} + class agent(Schema): id = String() ipaddr = List(String()) - atype = String(validate=OneOf(["渗透攻击", "状态感知", "参数感知"])) + atype = String(validate=OneOf(agent_type)) status = Boolean() idle = Boolean() port = Integer() @@ -24,36 +52,112 @@ class agent(Schema): # 代理注册接口 [email protected]("/register", ) [email protected]("代理注册接口", "返回分配给代理的编号值", hide=True) [email protected]("/register") [email protected]("代理注册接口", "返回分配给代理的编号值，用于代理和主控端通信注册，前端界面无需调用") @bp.input({ - # 代理类型 - "type": String(required=True, validate=OneOf(["gjst", "mbgz", "ztgz"])), - # 代理的通信端口 - "port": Integer(required=True), - # 代理所在的操作系统 - "sys": String() -}) -def register_agent(query_data): - print(query_data) + "atype": String(), + 'v6addr': List(String()), + 'lat': String(), + 'lng': String(), + 'cpu_num': Integer(), + 'ram_size': Integer(), + 'ram_per': Float(), + 'sys': String(), + "port": Integer() +}, example={'atype': "gjst", 'v6addr': ["2001::1"], 'lat': "-1.111", 'lng': "1.1111", 'cpu_num': 4, + 'ram_size': 16, 'ram_per': 0.55, + 'sys': "linux", "port": 2525}) +# 参数说明 +# 代理类型 +# "type": String(required=True, validate=OneOf(["gjst", "mbgz", "ztgz"])), +# 代理的通信端口 +# "port": Integer(required=True), +# # 代理所在的操作系统 +# "sys": String(), +# # IPv6地址 +# # "v6addr": List(String()), +# # 经度 +# "lat": String(), +# # 纬度 +# "lng": String(), +# # cpu核心数 +# "cpu_num": Integer(), +# # 内存大小 +# "ram_size": Integer(), +# # 已用内存比例 +# "ram_per": Float() +def register_agent(json_data): + data = dict(json_data) + id = "".join(random.sample('1234567890zyxwvutsrqponmlkjihgfedcba', 8)) + # 代理所有参数 + param = {"id": id, + "ipaddr": "|".join([request.remote_addr] + [i for i in data["v6addr"]]), + # "start_time",在数据库中实现 + "lat": data["lat"], + "lng": data["lng"], + "cpunum": data["cpu_num"], + "mem": str(data["ram_size"]) + "GB" + "(" + str( + '{:.2f}'.format(100 * (1 - float(data["ram_per"])))) + "%" + "可用)", + "sys": data["sys"], + "status": True, + "idle": True, + "port": data["port"], + "atype": data["atype"], + } + # 对字符串值进行处理 + param = string_to_mysql(param) + # 数据库表名 + param["tab"] = MYSQL_TAB_AGENT + + # 插入记录 + err = insert_agent(param) + if not err: + return {"id": id} + else: + return {"code": 500, "err": err} # 代理任务下发 -def deliver_task(): +# TODO:具体实现 +def deliver_task(agent_id, policy, policy_param): + # 查询agent_id对应代理的ip和端口信息 + # 将policy和policy_param作为参数传递到对应接口 pass -# 代理输出接收 @bp.post("/res") @bp.doc("代理输出消息接收接口", hide=True) -# TODO:代理输出接收接口实现 -def task_ret(): - return "ok" +def task_ret(json_data): + """ + task_ret 代理输出信息接收 + + Arguments: + json_data -- 请求中的json内容，必须包含"id"、"taskpolicy"、"level"和"info"两个keyword + + Returns: + "ok" -- 成功执行 + """ + sql = """INSERT INTO %s(CREATED_BY_AGENT,TLOG_LEVEL,TLOG_INFO,TLOG_TP) + VALUES(%s,%s,%s,%s) + """ % (MYSQL_TAB_TASK_LOG, json_data["id"], json_data["level"], json_data["info"], + json_data["taskpolicy"]) + try: + da.cursor.execute(sql) + da.conn.commit() + return "ok" + except Exception as e: + print(e) @bp.get("/") [email protected]("代理信息获取接口") [email protected]("代理信息获取接口", description="输入参数说明：</br>" + + "atype:能力类型，可选参数范围及对应含义为：all-全部（默认），gjst-攻击渗透，ztgz-状态感知，csgz-参数感知 </br>" + + "status:当前状态，可选参数范围及对应含义为：2-全部(默认)，1-在线，0-离线 </br>" + + "idle:是否空闲，可选参数范围及对应含义为：2-全部(默认)，1-空闲，0-执行中 </br>") @bp.input({ + "atype": String(load_default="all", validate=OneOf(["all", "gjst", "ztgz", "csgz"])), + "status": Integer(load_default=2, validate=OneOf([0, 1, 2])), + "idle": Integer(load_default=2, validate=OneOf([0, 1, 2])), "page": Integer(load_default=1), "per_page": Integer(load_default=10) }, location="query") @@ -65,20 +169,76 @@ def task_ret(): def agent_info(query_data): per_page = query_data["per_page"] page = query_data["page"] + agent_type = query_data["atype"] + status = query_data["status"] + idle = query_data["idle"] + + # 代理信息列表 agent_list = [] - # TODO:接口数据库贯通实现 - for i in range(per_page): - agent_list.append({ - "id": fake.unique.random_int(), - "ipaddr": [fake.ipv4_public(), fake.ipv6()], - "atype": fake.word(ext_word_list=["参数感知", "状态感知", "攻击渗透"]), - "status": random.choice([True, False]), - "idle": random.choice([True, False]), - "port": random.randint(1000, 65534), - "sys": random.choice(["Linux", "Windows", "Macos"]), - "cpu_num": 4, - "mem": random.choice(["4", "8", "16", "32"]) + "GB(" + str( - '{:.2f}'.format(100 * random.random())) + "%可用)", - "start_time": fake.date_time_between(start_date="-1y")}) - - return {"code": 200, "agent_data": agent_list, "total": 10 * per_page} + res = da.get_data(data_type="agent", search={"atype": agent_type, "status": status, "idle": idle}, + offset=(page - 1) * per_page, limit=per_page) + res_count = da.count_data(data_type="agent", search={"atype": agent_type, "status": status, "idle": idle}) + for r in res: + agent = {} + for key, value in r.items(): + agent[agent_response_map[key]] = value if key != "IPADDR" else str(value).split("|") + agent_list.append(agent) + return {"code": 200, "agent_data": agent_list, "total": res_count} + + [email protected]("代理删除接口", "输入参数说明：</br>" + + "id: 代理编号") [email protected]("/del") + "id": String(required=True) +}) + "code": Integer(), + "msg": String() +}) +def del_agent(json_data): + print(json_data) + sql = """DELETE FROM %s WHERE AGENT_ID = '%s' + """ % (MYSQL_TAB_AGENT, json_data["id"]) + try: + da.cursor.execute(sql) + da.conn.commit() + except Exception as e: + error(message=str(e)) + return {"code": 500, "msg": str(e)} + return {"code": 200, "msg": "ok"} + + +# 代理信息存储到数据库 +def insert_agent(param: dict): + sql = """REPLACE INTO %(tab)s( + AGENT_ID, + IPADDR, + LAT, + LNG , + AGENT_TYPE, + SYS , + PORT , + CPU_NUM , + STATUS , + MEM, + IDLE) + VALUES( + %(id)s, + %(ipaddr)s, + %(lat)s, + %(lng)s, + %(atype)s, + %(sys)s, + %(port)s, + %(cpunum)s, + %(status)s, + %(mem)s, + %(idle)s);""" % param + try: + da.cursor.execute(sql) + da.conn.commit() + except Exception as e: + error(message=str(e)) + return e + return None diff --git a/server/apps/model.py b/server/apps/model.py new file mode 100644 index 0000000..0748f70 --- /dev/null +++ b/server/apps/model.py @@ -0,0 +1,60 @@ +# 记录参数值和数据库表字段名之间的双向映射 +from bidict import bidict + +from settings import * + +# 参数与数据表映射关系 +tabmapping = bidict({ + "agent": MYSQL_TAB_AGENT, + "task": MYSQL_TAB_TASK, + "user": MYSQL_TAB_USER, + "target": MYSQL_TAB_TARGETDATA, + "syslog": MYSQL_TAB_SYSLOG, }) + +# 参数与数据表中时间字段的映射关系 +timemapping = { + "agent": "START_TIME", + "task": "CREATED_TIME", + "user": "CREATED_TIME", + "target": "UPDATED_TIME", +} + +# 不同接口的参数（或值）和对应数据表字段的映射关系 +keymapping = { + "agent": bidict({ + "atype": "AGENT_TYPE", + "status": "STATUS", + "idle": "IDLE" + }), + "target": bidict({ + "cou": "COU", + "isp": "ISP", + "IPv6": "IPv6", + "DNSSEC": "DNSSEC", + "DoH": "DOH", + "DoT": "DOT", + "ADDRv4": "ADDRv4", + "ADDRv6": "ADDRv6" + })} + +# 所有参数在数据库中对应的数据类型，用于拼接sql语句时特殊处理 +typemapping = { + "atype": "str", + "status": "int", + "idle": "int", + "cou": "str", + "isp": "str", + "ip": "str", + "IPv6": "int", + "DNSSEC": "int", + "DoH": "int", + "DoT": "int", + "ADDRv4": "str", + "ADDRv6": "str" +} + +# 默认参数 +default_data = { + "agent": {"atype": "all", "status": 2, "idle": 2}, + "target": {"proto": None, "isp": None, "cou": None} +} diff --git a/server/apps/sysinfo.py b/server/apps/sysinfo.py index 49a6031..04e5cbf 100644 --- a/server/apps/sysinfo.py +++ b/server/apps/sysinfo.py @@ -1,15 +1,22 @@ # 系统信息接口 -import datetime from apiflask import APIBlueprint, Schema from apiflask.fields import Date, Integer, Nested, Dict, DateTime, String, List from apiflask.validators import OneOf # 测试用 -from apps.util import fake +from settings import * +from .util import da bp = APIBlueprint("仪表盘接口", __name__, url_prefix="/sys") +log_map = { + "LOG_LEVEL": "level", + "LOG_INFO": "info", + "LOG_TIME": "time", + "S_IP": "ip", +} + class TargetOut(Schema): v6dns = Integer() @@ -37,18 +44,17 @@ class LogOut(Schema): "target": Nested(TargetOut()) }) def systate(): - # TODO:从数据库读取信息并统计 # 已部署代理节点数量 - agent_num = fake.random.randint(10, 100) + agent_num = da.count_data("agent") # 已执行任务数量（包括执行完毕和正在执行） - task_num = fake.random.randint(10, 100) + task_num = da.count_data("task") # 系统已运行天数 - workday = fake.random.randint(10, 100) + workday = datetime.date.today() - START_DAY # 已探测目标统计 - v6dns_num = fake.random.randint(10, 100) - dnssec_num = fake.random.randint(10, 100) - doh_num = fake.random.randint(10, 100) - dot_num = fake.random.randint(10, 100) + v6dns_num = da.count_data("target", {"IPv6": True}) + dnssec_num = da.count_data("target", {"DNSSEC": True}) + doh_num = da.count_data("target", {"DOH": True}) + dot_num = da.count_data("target", {"DOT": True}) # 返回结果 return { "code": 200, @@ -76,19 +82,23 @@ def target_date(): day = 1 while day < 8: d = today - datetime.timedelta(day) - # TODO:从数据库读取并统计 dates[d] = { - "v6dns": fake.random.randint(10, 20) * day, - "dnssec": fake.random.randint(10, 20) * day, - "doh": fake.random.randint(10, 20) * day, - "dot": fake.random.randint(10, 20) * day + "v6dns": da.count_data_by_time("target", time=d, search={"IPv6": True}), + "dnssec": da.count_data_by_time("target", time=d, search={"DNSSEC": True}), + "doh": da.count_data_by_time("target", time=d, search={"DOH": True}), + "dot": da.count_data_by_time("target", time=d, search={"DOT": True}) } day += 1 return {"code": 200, "date_data": dates} +# 系统操作日志获取接口 @bp.get("/log") @bp.input({ + "begin": Date(), + "end": Date(), + "level": String(validate=OneOf(["INFO", "WARNING", "ERROR"])), + "user": String(), "per_page": Integer(load_default=10), "page": Integer(load_default=1) }, location="query") @@ -98,19 +108,25 @@ def target_date(): "log_data": List(Nested(LogOut())), "total": Integer() }) +# TODO:完善 def sys_log(query_data): # 每页显示多少项 per_page = query_data["per_page"] # 页数 page = query_data["page"] - log_list = [] - # TODO:具体实现 - for i in range(per_page): - log_list.append({ - "time": fake.date_time_between(start_date="-1y"), - "level": fake.word(ext_word_list=["INFO", "WARNING", "ERROR"]), - "info": fake.text(max_nb_chars=20, ext_word_list=None), - "user": fake.name(), - "ip": fake.ipv4() - }) - return {"code": 200, "log_data": log_list, "total": 10 * per_page} + tabs = { + "syslog": MYSQL_TAB_SYSLOG, + "user": MYSQL_TAB_USER, + "start": (page - 1) * per_page, + "stop": per_page + } + sql = """SELECT + s.LOG_LEVEL as level, + s.LOG_INFO as info, + s.LOG_TIME as time, + s.S_IP as ip, + CONCAT(u.USER_NAME,"(ID:",u.USER_ID,")") as user + from %(syslog)s s,%(user)s u WHERE s.USER_ID=u.USER_ID LIMIT %(start)s,%(stop)s """ % tabs + da.cursor.execute(sql) + log_list = da.cursor.fetchall() + return {"log_data": log_list} diff --git a/server/apps/sysmange.py b/server/apps/sysmange.py index 7921077..4143d5c 100644 --- a/server/apps/sysmange.py +++ b/server/apps/sysmange.py @@ -3,7 +3,7 @@ from apiflask import APIBlueprint, Schema from apiflask.fields import String, Integer, DateTime, List, Nested -from apps.util import fake +from .util import fake bp = APIBlueprint("系统管理接口集合", __name__, url_prefix="/user") @@ -24,9 +24,11 @@ class User(Schema): }, location='query') @bp.output({ "code": Integer(), - "data": List(Nested(User())) + "data": List(Nested(User())), + "total": Integer() }) def list_user(query_data): + # TODO：具体实现 if query_data['page'] == 1: user_list = [ {"account": "Root", "username": "管理员", "time": fake.date_time_between(start_date='-10y', end_date='-9y'), @@ -57,5 +59,6 @@ def list_user(query_data): "msg": String() }) def create_user(json_data): + # TODO:具体实现 print(json_data) return {"code": 200, "msg": "ok"} diff --git a/server/apps/target.py b/server/apps/target.py index d11d479..5a1875e 100644 --- a/server/apps/target.py +++ b/server/apps/target.py @@ -1,5 +1,4 @@ # 目标状态感知 -import random # 时延测试接口 import threading @@ -9,7 +8,8 @@ from apiflask import APIBlueprint, Schema from apiflask.fields import Integer, String, List, Nested, IP, DateTime, Dict from requests.exceptions import Timeout -from apps.util import fake +from settings import * +from .util import da, debug bp = APIBlueprint("目标信息及状态接口集合", __name__, url_prefix="/target") @@ -17,6 +17,16 @@ icmp_delaytable = {} tcp_delaytable = {} dns_delaytable = {} +target_map = { + "ADDRv4": "ipv4", + "ADDRv6": "ipv6", + "COU": "cou", + "ISP": "isp", + "LAT": "lat", + "LNG": "lng", + "UPDATED_TIME": "time" +} + class Target(Schema): ipv4 = String() @@ -51,11 +61,22 @@ class DelayOut(Schema): delay_data = List(Nested(Delay())) +# 地图统计信息返回结构体 +class CouInfo(Schema): + # 国家名称 + name = String() + # 不同协议的解析器详细数量说明 + title = String() + # 总数量 + value = Integer() + + @bp.get("/delay/<string:type>") @bp.doc("获取每个节点的时延数据", "type参数为{icmp,dns,tcp}中的一个", hide=True) @bp.input({"ip": IP(required=False)}, location="query") @bp.output(DelayOut) def get_pernode_delay(query_data, type): + # TODO:节点选择 addr = "" if 'ip' in query_data.keys(): addr = query_data['ip'] @@ -154,11 +175,9 @@ def dns_delay_query(target, addr): # 状态感知——DNS记录测试接口 import dns.nameserver from apiflask.fields import String -from apiflask.validators import OneOf +from apiflask.validators import OneOf, ContainsOnly from dns import resolver -rrest_key = "rrset" - @bp.get("/check") @bp.doc("通过指定的解析器获取指定域名的A/AAAA记录", hide=True) @@ -169,7 +188,7 @@ rrest_key = "rrset" }, location='query') @bp.output({ "code": Integer(), - "ans": List(Dict(String(validate=OneOf([rrest_key])), String())) + "ans": List(Dict(String(validate=ContainsOnly(["rrset"])), String())) }) def record(query_data): # 特殊协议头 @@ -202,52 +221,180 @@ def record(query_data): myAnswers = myResolver.resolve(domain, qtype) if myAnswers.rrset is not None: for r in myAnswers.rrset: - ans.append({rrest_key: str(r)}) + ans.append({"rrset": str(r)}) return {"code": 200, 'ans': ans} @bp.get("/") [email protected]("目标信息获取接口", "返回目标信息") [email protected]("(表格)目标信息获取接口", "返回目标信息") @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(Target())) + "data": List(Nested(Target())), + "total": Integer() }) def target_info(query_data): - # 普通检索 - if query_data["ip"] is None: - per_page = query_data["per_page"] - target_list = [] - - for _ in range(per_page): - lat, lng, _, cou, _ = fake.location_on_land() - target_list.append({ - "ipv4": fake.ipv4_public(), - "ipv6": fake.ipv6(), - "protocol": random.choices(["IPv6", "DNSSEC", "DoH", "DoT"], k=random.randint(1, 4)), - "protect": random.choices(["0x20", "端口随机化", "请求最小化", "DNSSEC验证"], k=random.randint(1, 4)), - "cou": cou, - "lat": lat, - "lng": lng, - "isp": fake.company(), - "time": fake.date_time_between('-2y'), - }) - return {"code": 200, "data": target_list} - # 查询目标 + 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 = [] + + # 普通检索，默认所有条件为单选 + if ip is None: + # 无协议筛选 + if proto == None: + res = da.get_data(data_type="target", search={"proto": proto, "cou": cou, "isp": isp}, + offset=(page - 1) * per_page, limit=per_page) + res_count = da.count_data(data_type="target", search={"proto": proto, "cou": cou, "isp": isp}) + # 协议筛选，填入筛选的协议 + else: + res = da.get_data(data_type="target", search={proto: True, "cou": cou, "isp": isp}, + offset=(page - 1) * per_page, limit=per_page) + res_count = da.count_data(data_type="target", search={proto: True, "cou": cou, "isp": isp}) + else: + # 查询目标，根据v4、v6地址分类 + res = da.get_data(data_type="target", + search={"ADDRv4": ip} if "." in str(ip) else {"ADDRv6": ip}) + res_count = 1 + # 结果转换 + for r in res: + target = {} + # 支持的协议特殊处理 + protocol = [] + for k, v in r.items(): + debug(k + ":" + str(v)) + if k in ["IPv6", "DNSSEC", "DOH", "DOT"] and bool(v) is True: + protocol.append(k) + target["protocol"] = protocol + # 防护措施特殊处理 + protect = str(r["PROTECT"]).split("|") + target["protect"] = protect + # 其他统一处理 + for k in target_map.keys(): + # 原数据的值赋给经过映射之后的键 + target[target_map[k]] = r[k] + target_list.append(target) + return {"code": 200, "data": target_list, "total": res_count} + + [email protected]("/filter") [email protected]("目标可筛选信息获取接口") + "code": Integer(), + "proto": List(String()), + "isp": List(String()), + "cou": List(String()) +}) +def filter_info(): + proto = ["IPv6", "DNSSEC", "DoH", "DoT"] + # 查询所有的isp + isp_sql = """SELECT DISTINCT ISP from %s """ % MYSQL_TAB_TARGETDATA + # 执行查询 + da.cursor.execute(isp_sql) + isp_data = da.cursor.fetchall() + isp = [i.popitem()[1] for i in isp_data] + + # 查询所有的国家 + cou_sql = """SELECT DISTINCT COU from %s """ % MYSQL_TAB_TARGETDATA + # 执行查询 + da.cursor.execute(cou_sql) + cou_data = da.cursor.fetchall() + cou = [i.popitem()[1] for i in cou_data] + return {"code": 200, "proto": proto, "isp": isp, "cou": cou} + + [email protected]("/map") [email protected]("地图信息获取接口") + "protocol": String(load_default=None, validate=OneOf(['IPv6', 'DNSSEC', "DoH", "DoT"])), + "cou": String(load_default=None), + "isp": String(load_default=None), + "ip": IP(load_default=None) +}, location="query") + "code": Integer(), + "dataObject": Dict(String(load_default="earthAddTitle"), List(Nested(CouInfo()))), +}) +def map_info(query_data): + ip = query_data["ip"] + proto = query_data["protocol"] + cou = query_data["cou"] + isp = query_data["isp"] + + if ip is None: + cou_list = [] + # 比较和默认参数值不同的参数 + differ = set({"proto": proto, "cou": cou, "isp": isp}.items()).difference( + {"proto": None, "cou": None, "isp": None}.items()) + if differ == set(): + # 在数据表中进行分组查询,无额外筛选 + sql = """ + SELECT T.COU,sum(T.c) as R,sum(T.dnssec) as DNSSEC,sum(T.v6) as IPv6,sum(T.dh) as DoH,sum(T.dt) as DoT + FROM + (SELECT COU,count(*) as c,count(DNSSEC=1 or null) as dnssec,count(IPv6=1 or null) as v6,count(DOH=1 or null) as dh,count(DOT=1 or null) as dt + FROM %s + GROUP BY COU,DNSSEC,IPv6,DOH,DOT) as T GROUP BY T.COU;""" % (MYSQL_TAB_TARGETDATA) + else: + l = len(differ) + # 条件字典 + condition = {} + for _ in range(l): + key, val = differ.pop() + # 协议参数在数据表中对应的字段名 + if key == "proto": + condition[{'IPv6': "IPv6", "DNSSEC": "DNSSEC", "DoH": "DOH", "DoT": "DOT"}[val]] = str(True) + else: + # 国家和isp属性的键只需调整字母为大写 + key = key.upper() + condition[key] = "\"".join(["", str(val), ""]) + # 在数据表中进行分组查询,无额外筛选 + sql = """ + SELECT T.COU,sum(T.c) as R,sum(T.dnssec) as DNSSEC,sum(T.v6) as IPv6,sum(T.dh) as DoH,sum(T.dt) as DoT + FROM + (SELECT COU,count(*) as c,count(DNSSEC=1 or null) as dnssec,count(IPv6=1 or null) as v6,count(DOH=1 or null) as dh,count(DOT=1 or null) as dt + FROM %s + WHERE %s + GROUP BY COU,DNSSEC,IPv6,DOH,DOT) as T GROUP BY T.COU;""" % ( + MYSQL_TAB_TARGETDATA, " AND ".join(["=".join(condition.popitem()) for _ in range(l)])) + + # 执行查询 + da.cursor.execute(sql) + data = da.cursor.fetchall() + for d in data: + # 单一国家的数据 + value = [d['IPv6'], d['DNSSEC'], d['DoH'], d['DoT']] + cou_data = { + "name": d['COU'], + 'title': "支持各类协议的解析器数量为：{IPv6：" + str(d['IPv6']) + "，DNSSEC：" + str( + d['DNSSEC']) + "，DoH：" + str(d['DoH']) + "，DoT：" + str(d['DoT']) + "}", + "value": d['R'] + } + cou_list.append(cou_data) + + return {"code": 200, "dataObject": {"earthAddTitle": cou_list}} + # 查询目标 else: - lat, lng, _, cou, _ = fake.location_on_land() - target = [{"ipv4": query_data["ip"], - "ipv6": fake.ipv6(), - "protocol": random.choices(["IPv6", "DNSSEC", "DoH", "DoT"], k=random.randint(1, 4)), - "protect": random.choices(["0x20", "端口随机化", "请求最小化", "DNSSEC验证"], - k=random.randint(1, 4)), - "cou": cou, - "lat": lat, - "lng": lng, - "isp": fake.company(), - "time": fake.date_time_between('-2y'), }] - return {"code": 200, "data": target} + # 查询目标，根据v4、v6地址分类 + res = da.get_data(data_type="target", + search={"ADDRv4": ip} if "." in str(ip) else {"ADDRv6": ip}) + # 支持协议 + proto = [] + for p in ["IPv6", "DNSSEC", "DOH", "DOT"]: + if res[0][p]: + proto.append(p) + target = [{"name": res[0]["COU"], + "title": "该解析器支持：" + '、'.join(proto) + " 协议", + "value": 1, }] + return {"code": 200, "dataObject": {"earthAddTitle": target}} diff --git a/server/apps/task.py b/server/apps/task.py index 006cd69..751a4a1 100644 --- a/server/apps/task.py +++ b/server/apps/task.py @@ -6,10 +6,23 @@ from apiflask import APIBlueprint, Schema from apiflask.fields import String, Integer, IP, DateTime, List, Nested from apiflask.validators import OneOf -from apps.util import fake +from .util import fake, da bp = APIBlueprint("任务管理接口集合", __name__, url_prefix="/task") +# 数据库列与返回值的键对应关系 +task_response_map = { + "TASK_ID": "id", + "TARGET_IP": "target", + "TASK_NAME": "name", + "AGENT_ID": "agent", + "TARGET_DOMAIN": "target_domain", + "TARGET_RR": "target_rr", + "POLICY": "policy", + "CREATE_TIME": "create_time", + "STATUS": "status", +} + class Task(Schema): id = Integer() @@ -53,8 +66,9 @@ class TaskState(Schema): # 创建任务接口 @bp.post("/create") [email protected]("任务创建接口", - description="部分字段值的映射关系：</br> policy 期望策略，可选参数范围及对应含义为：，auto-自动，ddos-拒绝服务，sjqp-数据欺骗</br> scan 状态感知方式，可选参数范围及对应含义为：auto-自动，icmp-icmp/v6时延，tcp-tcp时延，dns-dns时延，record-记录正确性验证") [email protected]("任务创建接口", "部分字段值的映射关系：</br>" + + "policy 期望策略，可选参数范围及对应含义为：，auto-自动，ddos-拒绝服务，sjqp-数据欺骗</br>" + + "scan 状态感知方式，可选参数范围及对应含义为：auto-自动，icmp-icmp/v6时延，tcp-tcp时延，dns-dns时延，record-记录正确性验证") @bp.input({ # 任务名称 "name": String(), @@ -79,7 +93,7 @@ class TaskState(Schema): }) @bp.output({ "code": Integer(), - "msg": String(), + "msg": String() }) # TODO：创建任务接口具体实现 def make_task(json_data): @@ -122,20 +136,19 @@ def ops_task(json_data): # TODO：查询任务状态接口具体实现 def tasks_state(query_data): per_page = query_data["per_page"] + page = query_data["page"] + + # 任务列表 task_list = [] - for _ in range(per_page): - task_list.append({ - "id": fake.random.randint(1, 10000), - "target": fake.ipv4(), - "name": "示例任务", - "agent": fake.random.randint(1, 10000), - "target_domain": fake.domain_name(), - "target_rr": "NS attack.com", - "policy": random.choice(["auto", "ddos", "sjqp"]), - "create_time": fake.date_time_between(start_date="-1y"), - "status": random.choice(["working", "stop", "finish"]) - }) - return {"code": 200, "data": task_list, "total": 10 * per_page} + res = da.get_data(data_type="task", + offset=(page - 1) * per_page, limit=per_page) + res_count = da.count_data(data_type="task") + for r in res: + task = {} + for key, value in r.items(): + task[task_response_map[key]] = value + task_list.append(task) + return {"code": 200, "data": task_list, "total": res_count} # 任务详情接口 @@ -170,7 +183,7 @@ def task_info(query_data): "policy_id": str(fake.random.randint(1, 10000)), "policy_status": "评估中" }) - return {"data": task_state_list} + return {"code": 200, "data": task_state_list} @bp.get("/tp") @@ -179,7 +192,7 @@ def task_info(query_data): "id": String(required=True), "per_page": Integer(load_default=10), "page": Integer(load_default=1) -}) +}, location="query") @bp.output({ "code": Integer(), "data": List(Nested(TaskLog())), @@ -194,6 +207,6 @@ def taskpolicy_log(query_data): "targetip": "2406:1234:1234:1234:1234:1234:1234:1234", "level": random.choice(["INFO", "WARNING", "ERROR"]), "info": fake.text(max_nb_chars=20, ext_word_list=None) - } for _ in range(20)] + } for _ in range(per_page)] return {"code": 200, "data": policy_output, "total": 10 * per_page} diff --git a/server/apps/util.py b/server/apps/util.py index 9dff80f..7a7932f 100644 --- a/server/apps/util.py +++ b/server/apps/util.py @@ -1,8 +1,250 @@ # 工具包 +import sys + +import pymysql from faker import Faker -from faker.providers import date_time from faker.providers import company +from faker.providers import date_time +from loguru import logger +from pymysql.cursors import DictCursor + +import apps.model as model +from settings import * fake = Faker("zh_CN") fake.add_provider(date_time) fake.add_provider(company) + + +# 数据库访问类 +class DataHandler: + + # 数据库链接及数据库初始化 + def __init__(self): + + # mysql连接,采用字典游标，返回一系列字典值 + try: + self.conn = pymysql.connect(cursorclass=DictCursor, host=MYSQL_HOST, user='root', + password=MYSQL_PAWD, port=MYSQL_PORT) + except Exception as e: + error(str(e)) + # 数据库连接失败，停止后续操作 + sys.exit(1) + self.cursor = self.conn.cursor() + + # 初始化sql语句 + # 创建数据库 + dbsql = "CREATE DATABASE IF NOT EXISTS %s" % MYSQL_DATADB + + agentsql = """CREATE TABLE IF NOT EXISTS %s ( + `AGENT_ID` varchar(255) NOT NULL PRIMARY KEY, + `IPADDR` varchar(255), + `START_TIME` datetime DEFAULT CURRENT_TIMESTAMP, + `LAT` varchar(255), + `LNG` varchar(255), + `AGENT_TYPE` varchar(255), + `SYS` varchar(255), + `PORT` int, + `CPU_NUM` int, + `STATUS` bool, + `MEM` varchar(255), + `IDLE` bool)ENGINE=innodb DEFAULT CHARSET=utf8; """ % MYSQL_TAB_AGENT + + policysql = """CREATE TABLE IF NOT EXISTS %s( + `P_ID` varchar(255) NOT NULL PRIMARY KEY, + `P_EXE` varchar(255), + `P_TYPE` varchar(255), + `P_DESC` varchar(255), + `P_PAYLOAD` varchar(255), + `P_NAME` varchar(255), + `P_PROTO` varchar(255))ENGINE=innodb DEFAULT CHARSET=utf8; """ % MYSQL_TAB_POLICY + + syslogsql = """CREATE TABLE IF NOT EXISTS %s( + `LOG_ID` int NOT NULL AUTO_INCREMENT PRIMARY KEY, + `LOG_LEVEL` varchar(255), + `LOG_INFO` varchar(255), + `LOG_TIME` datetime DEFAULT CURRENT_TIMESTAMP, + `S_IP` varchar(255), + `USER_ID` varchar(255), + FOREIGN KEY(`USER_ID`) REFERENCES %s(`USER_ID`) + )ENGINE=innodb DEFAULT CHARSET=utf8; """ % (MYSQL_TAB_SYSLOG, MYSQL_TAB_USER) + + targetsql = """CREATE TABLE IF NOT EXISTS %s( + `TARGET_ID` int NOT NULL AUTO_INCREMENT PRIMARY KEY, + `ADDRv4` varchar(255), + `ADDRv6` varchar(255), + `IPv6` bool, + `DNSSEC` bool, + `DOT` bool, + `DOH` bool, + `COU` varchar(255), + `ISP` varchar(255), + `LAT` varchar(255), + `LNG` varchar(255), + `UPDATED_TIME` datetime DEFAULT CURRENT_TIMESTAMP, + `PROTECT` varchar(255), + `DOH_DOMAIN` varchar(255))ENGINE=innodb DEFAULT CHARSET=utf8; """ % MYSQL_TAB_TARGETDATA + + tasksql = """CREATE TABLE IF NOT EXISTS %s( + `TASK_ID` varchar(255) NOT NULL PRIMARY KEY, + `TASK_NAME` varchar(255), + `AGENT_ID` varchar(255), + `CREATED_BY` varchar(255), + `TARGET_IP` varchar(255), + `CREATED_TIME` datetime DEFAULT CURRENT_TIMESTAMP, + `POLICY` int, + `STATUS` varchar(255), + `POLICY_DELAY` varchar(255), + `TASK_DELAY` varchar(255), + `TARGET_SCAN` varchar(255), + `TARGET_DOMAIN` varchar(255), + `TARGET_RTYPE` varchar(255), + `TARGET_RR` varchar(255), + FOREIGN KEY(`AGENT_ID`) REFERENCES %s(`AGENT_ID`), + FOREIGN KEY(`CREATED_BY`) REFERENCES %s(`USER_ID`) + )ENGINE=innodb DEFAULT CHARSET=utf8; """ % ( + MYSQL_TAB_TASK, MYSQL_TAB_AGENT, MYSQL_TAB_USER) + + tasklogsql = """CREATE TABLE IF NOT EXISTS %s( + `TLOG_ID` int NOT NULL AUTO_INCREMENT PRIMARY KEY, + `CREATED_BY_AGENT` varchar(255), + `CREATED_TIME` datetime DEFAULT CURRENT_TIMESTAMP, + `TLOG_LEVEL` varchar(255), + `TLOG_INFO` varchar(255), + `TLOG_TP` int, + FOREIGN KEY(`CREATED_BY_AGENT`) REFERENCES %s(`AGENT_ID`), + FOREIGN KEY(`TLOG_TP`) REFERENCES %s(`TP_ID`) + )ENGINE=innodb DEFAULT CHARSET=utf8;""" % (MYSQL_TAB_TASK_LOG, MYSQL_TAB_AGENT, MYSQL_TAB_TASKPOLICY) + + taskpolicysql = """CREATE TABLE IF NOT EXISTS %s( + `TP_ID` int NOT NULL AUTO_INCREMENT PRIMARY KEY, + `TP_TIME` datetime DEFAULT CURRENT_TIMESTAMP, + `POLICY` varchar(255), + `POLICY_PARAM` varchar(255), + `FOR_TASK` varchar(255), + FOREIGN KEY(`POLICY`) REFERENCES %s(`P_ID`), + FOREIGN KEY(`FOR_TASK`) REFERENCES %s(`TASK_ID`) + ) ENGINE=innodb DEFAULT CHARSET=utf8;""" % (MYSQL_TAB_TASKPOLICY, MYSQL_TAB_POLICY, MYSQL_TAB_TASK) + + usersql = """CREATE TABLE IF NOT EXISTS %s( + `USER_ID` varchar(255) NOT NULL PRIMARY KEY, + `USER_NAME` varchar(255), + `CREATED_BY` varchar(255), + `CREATED_TIME` datetime DEFAULT CURRENT_TIMESTAMP, + `USER_PWD_HASH` varchar(255), + `USER_GROUP` varchar(255)) ENGINE=innodb DEFAULT CHARSET=utf8;""" % MYSQL_TAB_USER + + # 和task_policysql存在双向外键引用的问题，需要额外定义 + fkey = """ + ALTER TABLE %s ADD CONSTRAINT `fk_TASK_TASK_POLICY_1` + FOREIGN KEY(`POLICY`) REFERENCES %s(`TP_ID`)""" % ( + MYSQL_TAB_TASK, MYSQL_TAB_TASKPOLICY) + + # 执行sql语句 + try: + # 创建数据库 + self.cursor.execute(dbsql) + self.conn.commit() + self.conn.select_db(MYSQL_DATADB) + + # 创建表格 + self.cursor.execute(agentsql) + self.cursor.execute(usersql) + self.cursor.execute(syslogsql) + self.cursor.execute(targetsql) + self.cursor.execute(tasksql) + self.cursor.execute(policysql) + self.cursor.execute(taskpolicysql) + self.cursor.execute(tasklogsql) + self.conn.commit() + + # 创建外键 + self.cursor.execute(fkey) + except Exception as e: + # 重复外键 + if str(e).find("(1826"): + pass + + # 获取信息(代理、任务) + # data_type可选范围参照DataHandler.tabmapping的键 + # 若需要按条件检索，则以将检索维度与检索值以字典形式传入search + def get_data(self, search: dict = None, data_type="agent", offset=0, limit=10, count=False): + # 参数映射到表名 + tabname = model.tabmapping[data_type] + # 比较输入参数和默认值的差异 + if search == None: + differ = set() + else: + differ = set(search.items()).difference(model.default_data[data_type].items()) + # 完全一致 + if len(differ) == 0: + if not count: + sql = """SELECT * FROM %s LIMIT %s, %s""" % (tabname, offset, offset + limit) + self.cursor.execute(sql) + return self.cursor.fetchall() + else: + sql = """SELECT count(*) FROM %s""" % (tabname) + self.cursor.execute(sql) + return dict(self.cursor.fetchall()[0]).popitem()[1] + else: + l = len(differ) + # 条件字典 + condition = {} + for _ in range(l): + key, val = differ.pop() + # 参数在数据表中对应的字段名 + tab_key = model.keymapping[data_type][key] + if model.typemapping[key] == "str": + condition[tab_key] = "\"".join(["", str(val), ""]) + else: + condition[tab_key] = str(val) + if not count: + sql = """SELECT * FROM %s WHERE %s LIMIT %s, %s""" % ( + tabname, " AND ".join(["=".join(condition.popitem()) for _ in range(l)]), offset, offset + limit) + print(sql) + self.cursor.execute(sql) + return self.cursor.fetchall() + else: + sql = """SELECT count(*) FROM %s WHERE %s LIMIT %s, %s""" % ( + tabname, " AND ".join(["=".join(condition.popitem()) for _ in range(l)]), offset, limit) + self.cursor.execute(sql) + return dict(self.cursor.fetchall()[0]).popitem()[1] + + # 统计符合条件的信息数量 + def count_data(self, search: dict = None, data_type="agent"): + return self.get_data(search=search, data_type=data_type, count=True) + + def count_data_by_time(self, data_type="agent", time=None, search=None): + tabname = model.tabmapping[data_type] + timename = model.timemapping[data_type] + key, value = search.popitem() + sql = """SELECT COUNT(*) FROM %s WHERE + DATE_FORMAT(%s,'%Y-%m-%d')=%s AND %s=%s""" % (tabname, timename, time, key, value) + self.cursor.execute(sql) + return dict(self.cursor.fetchall()[0]).popitem()[1] + + +da = DataHandler() + + +# 将插入mysql中的字符串处理为满足mysql形式的单引号字符串，输入为字典形式 +def string_to_mysql(data: dict): + res = {} + for key, val in data.items(): + if type(val) == type("text"): + res[key] = "\'" + val + "\'" + else: + res[key] = val + return res + + +def debug(message, *args, **kwargs): + logger.debug(message, *args, **kwargs) + + +def info(message, *args, **kwargs): + logger.info(message, *args, **kwargs) + + +def error(message, *args, **kwargs): + logger.error(message, *args, **kwargs) diff --git a/server/requirements.txt b/server/requirements.txt index 5494f55..c239b02 100644 --- a/server/requirements.txt +++ b/server/requirements.txt @@ -1,9 +1,13 @@ +requests~=2.31.0 +Flask~=3.0.0 +APIFlask~=2.1.0 click~=8.1.7 numpy~=1.25.0 six~=1.16.0 -requests~=2.31.0 -pandas~=2.1.1 -APIFlask~=2.0.2 -flask~=3.0.0 -dnspython~=2.4.2 -icmplib~=3.0.4
\ No newline at end of file +pandas~=2.1.4 +faker~=18.9.0 +dnspython~=2.6.1 +pymysql~=1.1.0 +loguru~=0.5.3 +flask-cors +bidict~=0.23.1
\ No newline at end of file diff --git a/server/settings.py b/server/settings.py new file mode 100644 index 0000000..a67923f --- /dev/null +++ b/server/settings.py @@ -0,0 +1,21 @@ +import datetime + +DEBUG = True +SERVER_PORT = 12526 +START_DAY = datetime.date.today() + +# 数据库配置 +MYSQL_HOST = '127.0.0.1' +MYSQL_PORT = 13306 +MYSQL_PAWD = 'yydns' + +MYSQL_DATADB = "yydnsdata" + +MYSQL_TAB_SYSLOG = "SYSLOG" +MYSQL_TAB_USER = "USER" +MYSQL_TAB_AGENT = "AGENT" +MYSQL_TAB_TARGETDATA = "TARGETDATA" +MYSQL_TAB_TASK = "TASK" +MYSQL_TAB_POLICY = "POLICY" +MYSQL_TAB_TASKPOLICY = "TASK_POLICY" +MYSQL_TAB_TASK_LOG = "TASK_LOG" |