rename

src/plugins/nonebot_bison/admin_page/__init__.py

@@ -0,0 +1,114 @@
+from dataclasses import dataclass
+import importlib
+from pathlib import Path
+from typing import Callable
+
+from fastapi.staticfiles import StaticFiles
+from fastapi.templating import Jinja2Templates
+from nonebot import get_driver, on_command
+import nonebot
+from nonebot.adapters.cqhttp.bot import Bot
+from nonebot.adapters.cqhttp.event import GroupMessageEvent, PrivateMessageEvent
+from nonebot.drivers.fastapi import Driver
+from nonebot.log import logger
+from nonebot.rule import to_me
+from nonebot.typing import T_State
+import socketio
+import functools
+
+from starlette.requests import Request
+
+from .api import test, get_global_conf, auth, get_subs_info, get_target_name, add_group_sub
+from .token_manager import token_manager as tm
+from .jwt import load_jwt
+from ..plugin_config import plugin_config
+
+URL_BASE = '/bison/'
+GLOBAL_CONF_URL = f'{URL_BASE}api/global_conf'
+AUTH_URL = f'{URL_BASE}api/auth'
+SUBSCRIBE_URL = f'{URL_BASE}api/subs'
+GET_TARGET_NAME_URL = f'{URL_BASE}api/target_name'
+TEST_URL = f'{URL_BASE}test'
+
+sio = socketio.AsyncServer(async_mode="asgi", cors_allowed_origins="*")
+socket_app = socketio.ASGIApp(sio, socketio_path="socket")
+
+
+def register_router_fastapi(driver: Driver, socketio):
+    from fastapi.security import OAuth2PasswordBearer
+    from fastapi.param_functions import Depends
+    from fastapi import HTTPException, status
+
+    oath_scheme = OAuth2PasswordBearer(tokenUrl='token')
+
+    async def get_jwt_obj(token: str = Depends(oath_scheme)):
+        obj = load_jwt(token)
+        if not obj:
+            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)
+        return obj
+
+    async def check_group_permission(groupNumber: str, token_obj: dict = Depends(get_jwt_obj)):
+        groups = token_obj['groups']
+        if groupNumber not in groups:
+            raise HTTPException(status_code=status.HTTP_403_FORBIDDEN)
+
+    @dataclass
+    class AddSubscribeReq:
+        platformName: str
+        target: str
+        targetName: str
+        categories: list[str]
+        tags: list[str]
+
+    app = driver.server_app
+    static_path = str((Path(__file__).parent / "dist").resolve())
+    app.get(TEST_URL)(test)
+    app.get(GLOBAL_CONF_URL)(get_global_conf)
+    app.get(AUTH_URL)(auth)
+
+    @app.get(SUBSCRIBE_URL)
+    async def subs(jwt_obj: dict = Depends(get_jwt_obj)):
+        return await get_subs_info(jwt_obj)
+    @app.get(GET_TARGET_NAME_URL)
+    async def _get_target_name(platformName: str, target: str, jwt_obj: dict = Depends(get_jwt_obj)):
+        return await get_target_name(platformName, target, jwt_obj)
+    @app.post(SUBSCRIBE_URL, dependencies=[Depends(check_group_permission)])
+    async def _add_group_subs(groupNumber: str, req: AddSubscribeReq):
+        return await add_group_sub(group_number=groupNumber, platform_name=req.platformName,
+                target=req.target, target_name=req.targetName, cats=req.categories, tags=req.tags)
+    app.mount(URL_BASE, StaticFiles(directory=static_path, html=True), name="bison")
+    templates = Jinja2Templates(directory=static_path)
+
+    @app.get(f'{URL_BASE}{{rest_path:path}}')
+    async def serve_sap(request: Request, rest_path: str):
+        return templates.TemplateResponse("index.html", {"request": request})
+
+
+def init():
+    driver = get_driver()
+    if driver.type == 'fastapi':
+        assert(isinstance(driver, Driver))
+        register_router_fastapi(driver, socket_app)
+    else:
+        logger.warning(f"Driver {driver.type} not supported")
+        return
+    host = str(driver.config.host)
+    port = driver.config.port
+    if host in ["0.0.0.0", "127.0.0.1"]:
+        host = "localhost"
+    logger.opt(colors=True).info(f"Nonebot test frontend will be running at: "
+                                 f"<b><u>http://{host}:{port}{URL_BASE}</u></b>")
+
+init()
+
+get_token = on_command('后台管理', rule=to_me(), priority=5)
+@get_token.handle()
+async def send_token(bot: "Bot", event: PrivateMessageEvent, state: T_State):
+    driver = nonebot.get_driver()
+    superusers = driver.config.superusers
+    if event.get_user_id() not in superusers:
+        await get_token.finish('你不是管理员')
+    else:
+        token = tm.get_user_token((event.get_user_id(), event.sender.nickname))
+        await get_token.finish(f'请访问: {plugin_config.bison_outer_url}auth/{token}')
+

src/plugins/nonebot_bison/admin_page/api.py

@@ -0,0 +1,72 @@
+from ..platform import platform_manager, check_sub_target
+from .token_manager import token_manager
+from .jwt import pack_jwt
+from ..config import Config
+import nonebot
+from nonebot.adapters.cqhttp.bot import Bot
+
+async def test():
+    return {"status": 200, "text": "test"}
+
+async def get_global_conf():
+    res = {}
+    for platform_name, platform in platform_manager.items():
+        res[platform_name] = {
+            'platformName': platform_name,
+            'categories': platform.categories,
+            'enabledTag': platform.enable_tag,
+            'name': platform.name,
+            'hasTarget': getattr(platform, 'has_target')
+            }
+    return { 'platformConf': res }
+
+
+async def auth(token: str):
+    if qq_tuple := token_manager.get_user(token):
+        qq, nickname = qq_tuple
+        bot = nonebot.get_bot()
+        assert(isinstance(bot, Bot))
+        groups = await bot.call_api('get_group_list')
+        if str(qq) in nonebot.get_driver().config.superusers:
+            jwt_obj = {
+                    'id': str(qq),
+                    'groups': list(map(
+                        lambda info: {'id': info['group_id'], 'name': info['group_name']}, 
+                        groups)),
+                    }
+            ret_obj = {
+                    'type': 'admin',
+                    'name': nickname,
+                    'id': str(qq),
+                    'token': pack_jwt(jwt_obj)
+                    }
+            return { 'status': 200, **ret_obj }
+        else:
+            return { 'status': 400, 'type': '', 'name': '', 'id': '', 'token': '' }
+    else:
+        return { 'status': 400, 'type': '', 'name': '', 'id': '', 'token': '' }
+
+async def get_subs_info(jwt_obj: dict):
+    groups = jwt_obj['groups']
+    res = {}
+    for group in groups:
+        group_id = group['id']
+        config = Config()
+        subs = list(map(lambda sub: {
+            'targetType': sub['target_type'], 'target': sub['target'], 'targetName': sub['target_name'], 'cats': sub['cats'], 'tags': sub['tags']
+            }, config.list_subscribe(group_id, 'group')))
+        res[group_id] = {
+            'name': group['name'],
+            'subscribes': subs
+        }
+    return res
+
+async def get_target_name(platform_name: str, target: str, jwt_obj: dict):
+    return {'targetName': await check_sub_target(platform_name, target)}
+
+async def add_group_sub(group_number: str, platform_name: str, target: str, 
+        target_name: str, cats: list[str], tags: list[str]):
+    config = Config()
+    config.add_subscribe(group_number, 'group', target, target_name, platform_name, cats, tags)
+    return { 'status': 200, 'msg': '' }
+

src/plugins/nonebot_bison/admin_page/jwt.py

@@ -0,0 +1,20 @@
+import random
+import string
+from typing import Optional
+import jwt
+import datetime
+
+_key = ''.join(random.SystemRandom().choice(string.ascii_letters) for _ in range(16))
+
+def pack_jwt(obj: dict) -> str:
+    return jwt.encode(
+            {'exp': datetime.datetime.utcnow() + datetime.timedelta(hours=1), **obj},
+            _key, algorithm='HS256'
+            )
+
+def load_jwt(token: str) -> Optional[dict]:
+    try:
+        return jwt.decode(token, _key, algorithms=['HS256'])
+    except:
+        return None
+

src/plugins/nonebot_bison/admin_page/token_manager.py

@@ -0,0 +1,24 @@
+from typing import Optional
+from expiringdict import ExpiringDict
+import random
+import string 
+
+class TokenManager:
+
+    def __init__(self):
+        self.token_manager = ExpiringDict(max_len=100, max_age_seconds=60*10)
+
+    def get_user(self, token: str) -> Optional[tuple]:
+        res = self.token_manager.get(token)
+        assert(res is None or isinstance(res, tuple))
+        return res
+
+    def save_user(self, token: str, qq: tuple) -> None:
+        self.token_manager[token] = qq
+
+    def get_user_token(self, qq: tuple) -> str:
+        token = ''.join(random.choices(string.ascii_letters + string.digits, k=16))
+        self.save_user(token, qq)
+        return token
+
+token_manager = TokenManager()