1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
|
from datetime import datetime
from nonebot.params import ArgPlainText
from nonebot.adapters.onebot.v11 import PrivateMessageEvent, GroupMessageEvent
from ATRI import conf
from ATRI.log import log
from ATRI.service import Service
from ATRI.permission import MASTER
from ATRI.message import MessageBuilder
from ATRI.utils.apscheduler import scheduler
from .data_source import AuthDealer, get_host_ip
plugin = (
Service("控制台")
.document("前端管理页面")
.only_admin(True)
.permission(MASTER)
.main_cmd("/con")
)
async def __del_auth_key():
await AuthDealer.clear()
log.warning("控制台验证密钥已过期")
gen_console_key = plugin.cmd_as_group("auth", "获取进入网页后台的凭证")
@gen_console_key.got("is_pub_n", "咱的运行环境是否有公网(y/n)")
async def _(event: PrivateMessageEvent, is_pub_n: str = ArgPlainText("is_pub_n")):
is_access_key = conf.BotConfig.access_token
if not is_access_key:
await gen_console_key.finish(
MessageBuilder("缺少设置: access_token")
.text("请先填写该内容, 以保证ATRI与协议端链接的安全性")
.text("填写并重启, 方可启用控制台")
.text("Tip: 该内容请尽可能地复杂, 请勿使用中文")
)
auth_info = AuthDealer.get()
if auth_info:
now_time = datetime.now().timestamp()
if now_time < auth_info.dead_time:
raw_last_time = auth_info.dead_time - now_time
last_time = datetime.fromtimestamp(raw_last_time).minute
await gen_console_key.finish(
MessageBuilder("之前生成的密钥还在有效时间内奥")
.text(f"Token: {auth_info.token}")
.text(f"剩余有效时间: {last_time} min")
)
await AuthDealer.clear()
if is_pub_n != "y":
host = str(await get_host_ip(False))
await gen_console_key.send("没有公网吗...嗯知道了")
else:
host = str(await get_host_ip(True))
port = conf.BotConfig.port
auth = AuthDealer()
data = await auth.store()
msg = (
MessageBuilder("控制台信息已生成!")
.text(f"请访问: {host}:{port}")
.text(f"Token: {auth.get_token()}")
.text("该 token 有效时间为 15min")
)
scheduler.add_job(
__del_auth_key,
name="清除后台验证凭证",
next_run_time=datetime.fromtimestamp(data.dead_time),
misfire_grace_time=15,
)
await gen_console_key.finish(msg)
@gen_console_key.handle()
async def _(event: GroupMessageEvent):
await gen_console_key.finish("请私戳咱获取（")
del_console_key = plugin.cmd_as_group("del", "销毁进入网页后台的凭证")
@del_console_key.got("is_sure_d", "...你确定吗(y/n)")
async def _(is_sure: str = ArgPlainText("is_sure_d")):
if is_sure != "y":
await del_console_key.finish("反悔了呢...")
data = AuthDealer.get()
if data is None:
await del_console_key.finish("你还没向咱索取凭证呢...私戳咱键入 /con.auth 以获取")
await AuthDealer.clear()
await del_console_key.finish("销毁成功！如需再次获取: /con.auth")
from ATRI import driver as dr
from .data_source import init_resource
from .driver import init_driver
dr().on_startup(init_resource)
dr().on_startup(init_driver)
|
