qinglong
2da52e80d0
面板密码安全: - 支持哈希存储 (password_hash/password_salt) - 登录使用 secrets.compare_digest 时序安全比较 - 回退明文兼容旧配置 - 默认密码 admin 时打印 CRITICAL 警告 文件上传安全: - 拒绝危险扩展名: .exe/.dll/.so/.sh/.bat/.ps1 等 - 拒绝敏感文件名: .htaccess/Makefile/Dockerfile 等 - 上传时检查并返回 403 CSRF 防护: - panel_auth(csrf_protect=True) 参数 - 写操作需 X-CSRF-Token header 匹配 session token - 保护范围: 文件删除/写入/创建/上传/重命名 + 项目启停 + 代理管理 Co-Authored-By: Claude <noreply@anthropic.com>
113 lines
4.7 KiB
Python
113 lines
4.7 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
|
|
import os
|
|
import logging
|
|
import hashlib
|
|
import secrets
|
|
from pathlib import Path
|
|
from aiohttp import web
|
|
from .routes import auth, status, plugins, commands, logs, projects, proxy, plugin_web, files
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def _hash_pw(password: str, salt: str) -> str:
|
|
return hashlib.sha256((password + salt).encode()).hexdigest()
|
|
|
|
|
|
class WebPanelManager:
|
|
def __init__(self, config: dict, service_manager):
|
|
panel_cfg = config.get('panel', {}).get('entrance', {})
|
|
self.base_path = panel_cfg.get('path', '/panel')
|
|
self.panel_user = os.environ.get('SENSU_PANEL_USER', panel_cfg.get('username', 'admin'))
|
|
|
|
raw = os.environ.get('SENSU_PANEL_PASS') or panel_cfg.get('password', 'admin')
|
|
salt = panel_cfg.get('password_salt', '') or secrets.token_hex(16)
|
|
pw_hash = panel_cfg.get('password_hash', '') or _hash_pw(raw, salt)
|
|
self.panel_pass = raw # 保留向后兼容
|
|
self._pw_hash = pw_hash
|
|
self._pw_salt = salt
|
|
|
|
if raw == 'admin' and not panel_cfg.get('password_hash'):
|
|
logger.critical("⚠️ 面板使用默认密码 admin!")
|
|
|
|
self.base_path = f"/{self.base_path.strip('/')}"
|
|
self.sm = service_manager
|
|
self.project_root = Path(__file__).resolve().parent.parent.parent
|
|
|
|
async def start(self):
|
|
internet = self.sm.get_service("internet")
|
|
if not internet or not internet.http_app:
|
|
logger.error("❌ 网络服务未就绪，无法注册面板路由")
|
|
return False
|
|
|
|
app = internet.http_app
|
|
logger.info(f"🌐 向网络服务注册面板路由 (前缀: {self.base_path})...")
|
|
|
|
# 依赖注入
|
|
app['service_manager'] = self.sm
|
|
app['auth_service'] = self.sm.get_service("auth")
|
|
app['log_service'] = self.sm.get_service("log")
|
|
app['sensu_db'] = self.sm.get_service("sensu_db")
|
|
app['panel_config'] = {
|
|
'username': self.panel_user,
|
|
'password': self.panel_pass,
|
|
'password_hash': self._pw_hash,
|
|
'password_salt': self._pw_salt,
|
|
'index_path': self.project_root / "static" / "web_panel" / "index.html",
|
|
'home_path': self.project_root / "static" / "web_panel" / "home.html",
|
|
}
|
|
|
|
# 注册静态文件
|
|
# URL 前缀: /SenSu/static/ -> 物理路径: .../static/web_panel/
|
|
static_dir = self.project_root / "static" / "web_panel"
|
|
if static_dir.exists():
|
|
app.router.add_static(f'{self.base_path}/static/', path=str(static_dir))
|
|
logger.info(f"📂 静态资源已挂载: {self.base_path}/static/")
|
|
else:
|
|
logger.warning(f"⚠️ 静态目录缺失: {static_dir}")
|
|
|
|
# 注册首页 (登录页)
|
|
app.router.add_get(self.base_path, self._redirect_slash)
|
|
app.router.add_get(f'{self.base_path}/', self._serve_index)
|
|
app.router.add_get(f'{self.base_path}/index.html', self._serve_index)
|
|
|
|
# 🟢 新增: 注册面板主页 (/SenSu/home.html -> home.html)
|
|
app.router.add_get(f'{self.base_path}/home.html', self._serve_home)
|
|
|
|
# 注册 API 路由
|
|
auth.setup_routes(app, self.base_path)
|
|
status.setup_routes(app, self.base_path)
|
|
plugins.setup_routes(app, self.base_path)
|
|
commands.setup_routes(app, self.base_path)
|
|
logs.setup_routes(app, self.base_path)
|
|
projects.setup_project_routes(app, self.sm, self.base_path)
|
|
proxy.setup_proxy_routes(app, self.sm, self.base_path)
|
|
plugin_web.setup_plugin_web_routes(app, self.sm)
|
|
files.setup_file_routes(app, self.sm, self.base_path)
|
|
|
|
# 注册日志广播
|
|
ls = self.sm.get_service("log")
|
|
if ls and hasattr(ls, 'add_log_consumer'):
|
|
ls.add_log_consumer(logs.broadcast_log)
|
|
logger.info("📡 日志广播已连接")
|
|
|
|
logger.info(f"✅ 面板路由注册完成 (复用原有网络服务路由器)")
|
|
return True
|
|
|
|
async def _redirect_slash(self, req):
|
|
return web.HTTPFound(f'{self.base_path}/')
|
|
|
|
async def _serve_index(self, req):
|
|
"""提供登录页"""
|
|
path = req.app['panel_config']['index_path']
|
|
if path.exists(): return web.FileResponse(path)
|
|
return web.Response(text=f"❌ 找不到 index.html\n路径: {path}", status=404)
|
|
|
|
async def _serve_home(self, req):
|
|
"""提供面板主页"""
|
|
path = req.app['panel_config']['home_path']
|
|
if path.exists(): return web.FileResponse(path)
|
|
return web.Response(text=f"❌ 找不到 home.html\n路径: {path}", status=404)
|