# -*- coding: utf-8 -*-
"""企微客服助手的内置配置后台。
仅依赖 Python 标准库，提供：
- 用户名/密码登录和强制首次改密
- admin / operator / viewer 三种角色
- AI 与 MCP 配置的网页管理
- 供桌面端认证并拉取配置的 JSON API
"""
from __future__ import annotations
import argparse
import base64
import getpass
import hashlib
import hmac
import html
import ipaddress
import json
import os
import secrets
import sqlite3
import sys
import threading
import time
import urllib.parse
from collections import defaultdict, deque
from datetime import datetime
from http import HTTPStatus
from http.cookies import SimpleCookie
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from typing import Any
SCRIPT_DIR = Path(__file__).resolve().parent
DEFAULT_DB = SCRIPT_DIR / "backend.db"
DEFAULT_RUNTIME_FILE = SCRIPT_DIR / "backend_runtime.json"
DEFAULT_HOST = "127.0.0.1"
DEFAULT_PORT = 8765
DEFAULT_PORT_ATTEMPTS = 100
DEFAULT_ADMIN_PASSWORD = "Admin@123456"
ROLES = ("admin", "operator", "viewer")
ROLE_LABELS = {"admin": "管理员", "operator": "配置员", "viewer": "只读用户"}
PBKDF2_ITERATIONS = 310_000
CONFIG_KEYS = (
"AI_ENABLED",
"AI_API_BASE",
"AI_API_KEY",
"AI_MODEL",
"AI_USE_VISION",
"AI_CONTEXT_ENABLED",
"AI_CONTEXT_MAX_ROUNDS",
"AI_COUNTER_INSULT_ENABLED",
"AI_AGENT_NAME",
"AI_HOSPITAL_NAME",
"AI_MAX_TOKENS",
"AI_TEMPERATURE",
"AI_TIMEOUT",
"AI_MCP_ENABLED",
"AI_MCP_MAX_ROUNDS",
"AI_MCP_SERVERS",
)
BOOL_KEYS = {
"AI_ENABLED",
"AI_USE_VISION",
"AI_CONTEXT_ENABLED",
"AI_COUNTER_INSULT_ENABLED",
"AI_MCP_ENABLED",
}
def now_text() -> str:
return datetime.now().astimezone().isoformat(timespec="seconds")
def password_hash(password: str, salt: bytes | None = None) -> tuple[str, str]:
salt = salt or secrets.token_bytes(16)
digest = hashlib.pbkdf2_hmac(
"sha256", password.encode("utf-8"), salt, PBKDF2_ITERATIONS
)
return base64.b64encode(salt).decode("ascii"), base64.b64encode(digest).decode("ascii")
def verify_password(password: str, salt_text: str, digest_text: str) -> bool:
try:
salt = base64.b64decode(salt_text)
expected = base64.b64decode(digest_text)
except Exception:
return False
actual = hashlib.pbkdf2_hmac(
"sha256", password.encode("utf-8"), salt, PBKDF2_ITERATIONS
)
return hmac.compare_digest(actual, expected)
def token_hash(token: str) -> str:
return hashlib.sha256(token.encode("utf-8")).hexdigest()
def load_initial_config() -> dict[str, Any]:
path = SCRIPT_DIR / "ai_settings.json"
try:
saved = json.loads(path.read_text(encoding="utf-8"))
except (OSError, ValueError, TypeError):
saved = {}
defaults: dict[str, Any] = {
"AI_ENABLED": True,
"AI_API_BASE": "",
"AI_API_KEY": "",
"AI_MODEL": "",
"AI_USE_VISION": False,
"AI_CONTEXT_ENABLED": True,
"AI_CONTEXT_MAX_ROUNDS": 5,
"AI_COUNTER_INSULT_ENABLED": False,
"AI_AGENT_NAME": "客服",
"AI_HOSPITAL_NAME": "",
"AI_MAX_TOKENS": 500,
"AI_TEMPERATURE": 0.35,
"AI_TIMEOUT": 120,
"AI_MCP_ENABLED": False,
"AI_MCP_MAX_ROUNDS": 5,
"AI_MCP_SERVERS": [],
}
if isinstance(saved, dict):
defaults.update({key: saved[key] for key in CONFIG_KEYS if key in saved})
return defaults
class ClosingConnection(sqlite3.Connection):
"""提交/回滚后立即关闭，避免 Windows 上数据库文件长期被占用。"""
def __exit__(self, exc_type, exc_value, traceback):
try:
return super().__exit__(exc_type, exc_value, traceback)
finally:
self.close()
class Database:
def __init__(self, path: Path):
self.path = path
self.path.parent.mkdir(parents=True, exist_ok=True)
def connect(self) -> sqlite3.Connection:
connection = sqlite3.connect(
self.path, timeout=10, factory=ClosingConnection
)
connection.row_factory = sqlite3.Row
connection.execute("PRAGMA foreign_keys = ON")
return connection
def initialize(self, initial_password: str) -> bool:
with self.connect() as db:
db.executescript(
"""
PRAGMA journal_mode = WAL;
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
username TEXT NOT NULL UNIQUE COLLATE NOCASE,
password_salt TEXT NOT NULL,
password_digest TEXT NOT NULL,
role TEXT NOT NULL CHECK(role IN ('admin','operator','viewer')),
active INTEGER NOT NULL DEFAULT 1,
must_change_password INTEGER NOT NULL DEFAULT 1,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS auth_tokens (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
token_digest TEXT NOT NULL UNIQUE,
kind TEXT NOT NULL,
csrf_token TEXT NOT NULL,
device_name TEXT NOT NULL DEFAULT '',
created_at TEXT NOT NULL,
expires_at INTEGER NOT NULL,
last_used_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS model_config (
id INTEGER PRIMARY KEY CHECK(id = 1),
config_json TEXT NOT NULL,
version INTEGER NOT NULL DEFAULT 1,
updated_at TEXT NOT NULL,
updated_by INTEGER REFERENCES users(id)
);
CREATE TABLE IF NOT EXISTS audit_log (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER REFERENCES users(id),
action TEXT NOT NULL,
detail TEXT NOT NULL DEFAULT '',
ip_address TEXT NOT NULL DEFAULT '',
created_at TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_tokens_digest ON auth_tokens(token_digest);
CREATE INDEX IF NOT EXISTS idx_audit_created ON audit_log(created_at DESC);
"""
)
created = db.execute("SELECT COUNT(*) FROM users").fetchone()[0] == 0
if created:
salt, digest = password_hash(initial_password)
stamp = now_text()
db.execute(
"""INSERT INTO users
(username,password_salt,password_digest,role,active,must_change_password,created_at,updated_at)
VALUES (?,?,?,?,1,1,?,?)""",
("admin", salt, digest, "admin", stamp, stamp),
)
if db.execute("SELECT COUNT(*) FROM model_config").fetchone()[0] == 0:
admin = db.execute("SELECT id FROM users ORDER BY id LIMIT 1").fetchone()
db.execute(
"""INSERT INTO model_config
(id,config_json,version,updated_at,updated_by) VALUES (1,?,1,?,?)""",
(
json.dumps(load_initial_config(), ensure_ascii=False),
now_text(),
admin["id"] if admin else None,
),
)
db.commit()
return created
def authenticate(self, username: str, password: str) -> sqlite3.Row | None:
with self.connect() as db:
user = db.execute(
"SELECT * FROM users WHERE username = ? AND active = 1", (username,)
).fetchone()
if user and verify_password(password, user["password_salt"], user["password_digest"]):
return user
return None
def create_token(
self, user_id: int, kind: str, device_name: str, lifetime: int
) -> tuple[str, str]:
token = secrets.token_urlsafe(36)
csrf = secrets.token_urlsafe(24)
stamp = now_text()
with self.connect() as db:
db.execute(
"""INSERT INTO auth_tokens
(user_id,token_digest,kind,csrf_token,device_name,created_at,expires_at,last_used_at)
VALUES (?,?,?,?,?,?,?,?)""",
(
user_id,
token_hash(token),
kind,
csrf,
str(device_name or "")[:120],
stamp,
int(time.time()) + lifetime,
stamp,
),
)
db.execute("DELETE FROM auth_tokens WHERE expires_at < ?", (int(time.time()),))
db.commit()
return token, csrf
def session(self, token: str) -> sqlite3.Row | None:
if not token:
return None
with self.connect() as db:
row = db.execute(
"""SELECT u.*, t.id AS token_id, t.kind AS token_kind,
t.csrf_token, t.expires_at
FROM auth_tokens t JOIN users u ON u.id=t.user_id
WHERE t.token_digest=? AND t.expires_at>=? AND u.active=1""",
(token_hash(token), int(time.time())),
).fetchone()
if row:
db.execute(
"UPDATE auth_tokens SET last_used_at=? WHERE id=?",
(now_text(), row["token_id"]),
)
db.commit()
return row
def revoke(self, token: str) -> None:
if not token:
return
with self.connect() as db:
db.execute("DELETE FROM auth_tokens WHERE token_digest=?", (token_hash(token),))
db.commit()
def config(self) -> sqlite3.Row:
with self.connect() as db:
return db.execute(
"""SELECT c.*, u.username AS updated_by_name
FROM model_config c LEFT JOIN users u ON u.id=c.updated_by WHERE c.id=1"""
).fetchone()
def save_config(self, config: dict[str, Any], user_id: int, ip: str) -> int:
with self.connect() as db:
db.execute("BEGIN IMMEDIATE")
current = db.execute("SELECT version FROM model_config WHERE id=1").fetchone()
version = int(current["version"]) + 1
db.execute(
"""UPDATE model_config SET config_json=?,version=?,updated_at=?,updated_by=?
WHERE id=1""",
(json.dumps(config, ensure_ascii=False), version, now_text(), user_id),
)
self._audit(db, user_id, "config.update", f"version={version}", ip)
db.commit()
return version
@staticmethod
def _audit(
db: sqlite3.Connection, user_id: int | None, action: str, detail: str, ip: str
) -> None:
db.execute(
"INSERT INTO audit_log(user_id,action,detail,ip_address,created_at) VALUES(?,?,?,?,?)",
(user_id, action, detail[:500], ip[:80], now_text()),
)
def audit(self, user_id: int | None, action: str, detail: str, ip: str) -> None:
with self.connect() as db:
self._audit(db, user_id, action, detail, ip)
db.commit()
def list_users(self) -> list[sqlite3.Row]:
with self.connect() as db:
return list(db.execute("SELECT * FROM users ORDER BY id"))
def recent_audit(self, limit: int = 12) -> list[sqlite3.Row]:
with self.connect() as db:
return list(
db.execute(
"""SELECT a.*, u.username FROM audit_log a
LEFT JOIN users u ON u.id=a.user_id ORDER BY a.id DESC LIMIT ?""",
(limit,),
)
)
def create_user(
self, username: str, password: str, role: str, actor_id: int, ip: str
) -> None:
salt, digest = password_hash(password)
stamp = now_text()
with self.connect() as db:
db.execute(
"""INSERT INTO users
(username,password_salt,password_digest,role,active,must_change_password,created_at,updated_at)
VALUES(?,?,?,?,1,1,?,?)""",
(username, salt, digest, role, stamp, stamp),
)
self._audit(db, actor_id, "user.create", f"username={username}, role={role}", ip)
db.commit()
def update_user(
self,
user_id: int,
role: str,
active: bool,
new_password: str,
actor_id: int,
ip: str,
) -> None:
with self.connect() as db:
target = db.execute("SELECT * FROM users WHERE id=?", (user_id,)).fetchone()
if not target:
raise ValueError("用户不存在")
if user_id == actor_id and not active:
raise ValueError("不能停用当前登录账号")
if target["role"] == "admin" and (role != "admin" or not active):
count = db.execute(
"SELECT COUNT(*) FROM users WHERE role='admin' AND active=1"
).fetchone()[0]
if count <= 1:
raise ValueError("系统至少需要保留一个启用的管理员")
values: list[Any] = [role, int(active), now_text()]
sql = "UPDATE users SET role=?,active=?,updated_at=?"
if new_password:
salt, digest = password_hash(new_password)
sql += ",password_salt=?,password_digest=?,must_change_password=1"
values.extend([salt, digest])
sql += " WHERE id=?"
values.append(user_id)
db.execute(sql, values)
if not active or new_password:
db.execute("DELETE FROM auth_tokens WHERE user_id=?", (user_id,))
self._audit(
db,
actor_id,
"user.update",
f"username={target['username']}, role={role}, active={int(active)}, password_reset={bool(new_password)}",
ip,
)
db.commit()
def change_password(self, user_id: int, current: str, new_password: str, ip: str) -> None:
with self.connect() as db:
user = db.execute("SELECT * FROM users WHERE id=?", (user_id,)).fetchone()
if not user or not verify_password(
current, user["password_salt"], user["password_digest"]
):
raise ValueError("当前密码不正确")
salt, digest = password_hash(new_password)
db.execute(
"""UPDATE users SET password_salt=?,password_digest=?,must_change_password=0,
updated_at=? WHERE id=?""",
(salt, digest, now_text(), user_id),
)
self._audit(db, user_id, "password.change", "", ip)
db.commit()
def reset_admin_password(self, password: str) -> None:
salt, digest = password_hash(password)
with self.connect() as db:
result = db.execute(
"""UPDATE users SET password_salt=?,password_digest=?,active=1,
must_change_password=1,updated_at=? WHERE username='admin'""",
(salt, digest, now_text()),
)
if result.rowcount == 0:
stamp = now_text()
db.execute(
"""INSERT INTO users
(username,password_salt,password_digest,role,active,must_change_password,created_at,updated_at)
VALUES('admin',?,?,'admin',1,1,?,?)""",
(salt, digest, stamp, stamp),
)
db.execute(
"DELETE FROM auth_tokens WHERE user_id=(SELECT id FROM users WHERE username='admin')"
)
db.commit()
class LoginLimiter:
def __init__(self):
self._attempts: dict[str, deque[float]] = defaultdict(deque)
self._lock = threading.Lock()
def allowed(self, key: str) -> bool:
cutoff = time.time() - 300
with self._lock:
items = self._attempts[key]
while items and items[0] < cutoff:
items.popleft()
return len(items) < 8
def failure(self, key: str) -> None:
with self._lock:
self._attempts[key].append(time.time())
def success(self, key: str) -> None:
with self._lock:
self._attempts.pop(key, None)
LOGIN_LIMITER = LoginLimiter()
BASE_CSS = """
:root{--bg:#f3f7f5;--surface:#fff;--ink:#17251f;--muted:#67786f;--line:#dce7e1;
--accent:#0d9871;--deep:#102b21;--danger:#c94352;--soft:#e1f4ec}
*{box-sizing:border-box}body{margin:0;background:var(--bg);color:var(--ink);font-family:"Microsoft YaHei UI","Segoe UI",sans-serif}
a{color:var(--accent);text-decoration:none}.shell{min-height:100vh;display:grid;grid-template-columns:250px 1fr}
aside{background:var(--deep);color:white;padding:30px 24px;position:sticky;top:0;height:100vh}.brand{font-size:22px;font-weight:800;letter-spacing:.04em}
.brand small{display:block;margin-top:8px;color:#9fc0b3;font-size:12px;font-weight:500}.userbox{margin-top:44px;padding:16px;background:#17392c;border:1px solid #2b5143;border-radius:14px}
.role{display:inline-block;margin-top:8px;padding:4px 9px;background:#275643;border-radius:999px;color:#cce9de;font-size:12px}
nav{margin-top:30px;display:grid;gap:8px}nav a{color:#c6d8d0;padding:10px 12px;border-radius:9px}nav a:hover{background:#1c4435;color:white}
main{padding:34px;max-width:1450px;width:100%;margin:0 auto}.top{display:flex;justify-content:space-between;align-items:end;margin-bottom:24px}.top h1{font-size:30px;margin:0 0 7px}.muted{color:var(--muted);font-size:13px}
.grid{display:grid;grid-template-columns:1.35fr .9fr;gap:20px;align-items:start}.card{background:var(--surface);border:1px solid var(--line);border-radius:17px;padding:22px;margin-bottom:20px;box-shadow:0 10px 30px rgba(17,51,38,.035)}
.card h2{font-size:18px;margin:0 0 5px}.cardhead{display:flex;justify-content:space-between;gap:15px;align-items:start;margin-bottom:18px}.version{font:600 12px Consolas,monospace;color:var(--accent);background:var(--soft);padding:6px 9px;border-radius:8px}
.formgrid{display:grid;grid-template-columns:1fr 1fr;gap:14px}.full{grid-column:1/-1}label{display:block;color:#53675d;font-size:13px;margin-bottom:6px}input,select,textarea{width:100%;border:1px solid #ceddd5;background:#fbfcfb;border-radius:9px;padding:10px 12px;color:var(--ink);font:inherit;outline:none}
input:focus,select:focus,textarea:focus{border-color:var(--accent);box-shadow:0 0 0 3px #dff3eb}textarea{min-height:150px;font:13px Consolas,monospace;resize:vertical}.switches{display:grid;grid-template-columns:1fr 1fr;gap:10px 18px}.check{display:flex;align-items:center;gap:9px;color:var(--ink)}.check input{width:18px;height:18px;accent-color:var(--accent)}
.actions{display:flex;justify-content:flex-end;margin-top:18px;gap:10px}button,.button{border:0;border-radius:9px;padding:10px 17px;font:600 14px inherit;cursor:pointer;background:var(--accent);color:white}button:hover{filter:brightness(.93)}button.secondary{background:#edf3f0;color:#28483b}.danger{color:var(--danger)}
.flash{padding:12px 15px;border-radius:10px;margin-bottom:18px;background:var(--soft);color:#087656;border:1px solid #c9eadc}.flash.error{background:#fbeaec;color:var(--danger);border-color:#f2cbd1}
.user{display:grid;grid-template-columns:1.1fr .8fr .6fr 1.1fr auto;gap:8px;align-items:end;padding:12px 0;border-top:1px solid #edf2ef}.user:first-of-type{border-top:0}.username{font-weight:700;padding:11px 0}.user input,.user select{padding:8px 9px}.tiny{font-size:11px;color:#829189}
.audit{display:grid;gap:12px}.event{border-left:3px solid #b8ddce;padding-left:11px}.event b{font-size:13px}.event div{font-size:11px;color:#829189;margin-top:3px}
.loginwrap{min-height:100vh;display:grid;place-items:center;padding:24px}.login{width:min(430px,100%);background:white;border:1px solid var(--line);border-radius:22px;padding:34px;box-shadow:0 25px 70px rgba(12,48,34,.12)}.login h1{margin:0 0 8px;font-size:27px}.login form{display:grid;gap:15px;margin-top:26px}.login button{margin-top:5px;padding:12px}.notice{background:#fff7e8;border:1px solid #f0d6a5;color:#8a5a0a;padding:13px;border-radius:10px;font-size:13px;margin-bottom:18px}
@media(max-width:980px){.shell{grid-template-columns:1fr}aside{height:auto;position:static}.grid{grid-template-columns:1fr}.user{grid-template-columns:1fr 1fr}.user .actions{grid-column:1/-1}.formgrid{grid-template-columns:1fr}.full{grid-column:auto}main{padding:20px}}
"""
def page(title: str, body: str) -> str:
return (
"{html.escape(title)} {body}"
)
class AdminServer(ThreadingHTTPServer):
daemon_threads = True
def __init__(self, address: tuple[str, int], database: Database):
self.local_sync_token = secrets.token_urlsafe(36)
super().__init__(address, AdminHandler)
self.database = database
def _address_in_use(exc: OSError) -> bool:
"""兼容 Windows、Linux 和 macOS 的端口占用错误码。"""
# Windows 在独占地址被另一个进程监听时，可能返回 10013 而不是 10048。
return exc.errno in (13, 48, 98, 10013, 10048) or getattr(
exc, "winerror", None
) in (10013, 10048)
def create_server(
host: str,
preferred_port: int,
database: Database,
*,
max_attempts: int = DEFAULT_PORT_ATTEMPTS,
) -> tuple[AdminServer, int]:
"""创建服务；首选端口被占用时自动向后尝试可用端口。"""
if not 0 <= preferred_port <= 65535:
raise ValueError("端口必须在 0-65535 之间")
if max_attempts < 1:
raise ValueError("端口尝试次数必须大于 0")
if preferred_port == 0:
server = AdminServer((host, 0), database)
return server, int(server.server_address[1])
last_error: OSError | None = None
for port in range(preferred_port, min(65535, preferred_port + max_attempts - 1) + 1):
try:
return AdminServer((host, port), database), port
except OSError as exc:
if not _address_in_use(exc):
raise
last_error = exc
raise OSError(
f"端口 {preferred_port}-{min(65535, preferred_port + max_attempts - 1)} 均被占用"
) from last_error
def public_server_url(host: str, port: int) -> str:
display_host = "127.0.0.1" if host in ("0.0.0.0", "::", "") else host
if ":" in display_host and not display_host.startswith("["):
display_host = f"[{display_host}]"
return f"http://{display_host}:{port}"
def write_runtime_info(
path: Path,
host: str,
port: int,
*,
local_sync_token: str = "",
) -> dict[str, Any]:
"""发布实际监听端口，供同项目桌面端自动发现。"""
info = {
"pid": os.getpid(),
"host": host,
"port": int(port),
"server_url": public_server_url(host, port),
"local_sync_token": local_sync_token,
"started_at": now_text(),
}
path = path.resolve()
path.parent.mkdir(parents=True, exist_ok=True)
temporary = path.with_suffix(path.suffix + ".tmp")
temporary.write_text(json.dumps(info, ensure_ascii=False, indent=2), encoding="utf-8")
os.replace(temporary, path)
return info
def clear_runtime_info(path: Path, port: int) -> None:
"""只清理由当前进程写入的发现文件，避免影响另一个后台实例。"""
path = path.resolve()
try:
info = json.loads(path.read_text(encoding="utf-8"))
except (OSError, ValueError, TypeError):
return
if info.get("pid") == os.getpid() and info.get("port") == int(port):
try:
path.unlink()
except FileNotFoundError:
pass
class AdminHandler(BaseHTTPRequestHandler):
server_version = "WeComConfig/1.0"
@property
def db(self) -> Database:
return self.server.database # type: ignore[attr-defined]
@property
def client_ip(self) -> str:
forwarded = self.headers.get("X-Forwarded-For", "").split(",", 1)[0].strip()
return forwarded or self.client_address[0]
def log_message(self, format_string: str, *args: Any) -> None:
sys.stdout.write(
f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] {self.client_ip} {format_string % args}\n"
)
def _security_headers(self) -> None:
self.send_header("X-Content-Type-Options", "nosniff")
self.send_header("X-Frame-Options", "DENY")
self.send_header("Referrer-Policy", "no-referrer")
self.send_header("Cache-Control", "no-store")
self.send_header(
"Content-Security-Policy",
"default-src 'none'; style-src 'unsafe-inline'; form-action 'self'; base-uri 'none'; frame-ancestors 'none'",
)
def _send(self, status: int, content: str, content_type: str) -> None:
data = content.encode("utf-8")
self.send_response(status)
self._security_headers()
self.send_header("Content-Type", content_type)
self.send_header("Content-Length", str(len(data)))
self.end_headers()
self.wfile.write(data)
def json_response(self, status: int, data: dict[str, Any]) -> None:
self._send(
status,
json.dumps(data, ensure_ascii=False),
"application/json; charset=utf-8",
)
def html_response(self, status: int, content: str) -> None:
self._send(status, content, "text/html; charset=utf-8")
def redirect(self, location: str, *, cookie: str = "") -> None:
self.send_response(HTTPStatus.SEE_OTHER)
self._security_headers()
self.send_header("Location", location)
if cookie:
self.send_header("Set-Cookie", cookie)
self.send_header("Content-Length", "0")
self.end_headers()
def body_bytes(self, limit: int = 512_000) -> bytes:
try:
length = int(self.headers.get("Content-Length", "0"))
except ValueError:
raise ValueError("Content-Length 无效")
if length < 0 or length > limit:
raise ValueError("请求内容过大")
return self.rfile.read(length)
def json_body(self) -> dict[str, Any]:
try:
data = json.loads(self.body_bytes().decode("utf-8") or "{}")
except (UnicodeDecodeError, ValueError) as exc:
raise ValueError("请求 JSON 无效") from exc
if not isinstance(data, dict):
raise ValueError("请求 JSON 必须是对象")
return data
def form_body(self) -> dict[str, str]:
try:
parsed = urllib.parse.parse_qs(
self.body_bytes().decode("utf-8"), keep_blank_values=True
)
except UnicodeDecodeError as exc:
raise ValueError("表单编码无效") from exc
return {key: values[-1] for key, values in parsed.items()}
def bearer_token(self) -> str:
authorization = self.headers.get("Authorization", "")
if authorization.startswith("Bearer "):
return authorization[7:].strip()
return ""
def cookie_token(self) -> str:
cookie = SimpleCookie()
try:
cookie.load(self.headers.get("Cookie", ""))
except Exception:
return ""
return cookie.get("session_token").value if cookie.get("session_token") else ""
def auth(self, api: bool = False) -> tuple[sqlite3.Row | None, str]:
token = self.bearer_token() if api else self.cookie_token()
return self.db.session(token), token
@staticmethod
def valid_password(value: str) -> bool:
return len(value) >= 10 and any(c.isalpha() for c in value) and any(
c.isdigit() for c in value
)
def require_web_auth(
self, *, roles: tuple[str, ...] | None = None, allow_password_change: bool = False
) -> tuple[sqlite3.Row, str] | None:
user, token = self.auth(False)
if not user:
self.redirect("/login")
return None
if user["must_change_password"] and not allow_password_change:
self.redirect("/?error=" + urllib.parse.quote("首次登录请先修改密码"))
return None
if roles and user["role"] not in roles:
self.redirect("/?error=" + urllib.parse.quote("当前角色没有执行此操作的权限"))
return None
return user, token
def require_api_auth(self) -> sqlite3.Row | None:
user, _ = self.auth(True)
if not user:
self.json_response(HTTPStatus.UNAUTHORIZED, {"error": "登录已失效，请重新登录"})
return None
if user["must_change_password"]:
self.json_response(
HTTPStatus.FORBIDDEN, {"error": "请先在后台网页修改初始密码"}
)
return None
return user
def local_sync_authorized(self) -> bool:
"""允许同机桌面端使用运行时随机凭证只读模型配置。"""
supplied = self.headers.get("X-Desktop-Sync-Token", "")
expected = getattr(self.server, "local_sync_token", "")
if not supplied or not expected:
return False
try:
is_loopback = ipaddress.ip_address(self.client_address[0]).is_loopback
except ValueError:
return False
return is_loopback and hmac.compare_digest(supplied, expected)
@staticmethod
def csrf_ok(user: sqlite3.Row, form: dict[str, str]) -> bool:
return hmac.compare_digest(str(user["csrf_token"]), str(form.get("csrf", "")))
def do_GET(self) -> None:
path = urllib.parse.urlparse(self.path).path.rstrip("/") or "/"
if path == "/health":
self.json_response(HTTPStatus.OK, {"status": "ok", "time": now_text()})
elif path == "/login":
user, _ = self.auth(False)
if user:
self.redirect("/")
else:
self.render_login()
elif path == "/":
auth = self.require_web_auth(allow_password_change=True)
if auth:
self.render_dashboard(auth[0])
elif path == "/api/v1/me":
user = self.require_api_auth()
if user:
self.json_response(HTTPStatus.OK, {"user": self.public_user(user)})
elif path == "/api/v1/config":
local_sync = self.local_sync_authorized()
user = None if local_sync else self.require_api_auth()
if local_sync or user:
row = self.db.config()
self.json_response(
HTTPStatus.OK,
{
"version": row["version"],
"updated_at": row["updated_at"],
"updated_by": row["updated_by_name"] or "system",
"config": json.loads(row["config_json"]),
},
)
else:
self.json_response(HTTPStatus.NOT_FOUND, {"error": "页面不存在"})
def do_POST(self) -> None:
path = urllib.parse.urlparse(self.path).path.rstrip("/") or "/"
try:
if path == "/login":
self.web_login()
elif path == "/logout":
self.web_logout()
elif path == "/admin/config":
self.web_save_config()
elif path == "/admin/users/create":
self.web_create_user()
elif path == "/admin/users/update":
self.web_update_user()
elif path == "/admin/password":
self.web_change_password()
elif path == "/api/v1/auth/login":
self.api_login()
elif path == "/api/v1/auth/logout":
self.api_logout()
else:
self.json_response(HTTPStatus.NOT_FOUND, {"error": "接口不存在"})
except ValueError as exc:
if path.startswith("/api/"):
self.json_response(HTTPStatus.BAD_REQUEST, {"error": str(exc)})
else:
self.redirect("/?error=" + urllib.parse.quote(str(exc)))
except sqlite3.IntegrityError:
if path.startswith("/api/"):
self.json_response(HTTPStatus.CONFLICT, {"error": "用户名已存在"})
else:
self.redirect("/?error=" + urllib.parse.quote("用户名已存在"))
@staticmethod
def public_user(user: sqlite3.Row) -> dict[str, Any]:
return {
"id": user["id"],
"username": user["username"],
"role": user["role"],
"role_name": ROLE_LABELS[user["role"]],
}
def render_login(self, error: str = "") -> None:
message = (
f"{html.escape(error)}
" if error else ""
)
body = f"""
ZHEN AI ADMIN
配置管理后台 统一管理桌面端账号、角色和 AI 模型参数。
{message}
{html.escape(message)}
"
elif error:
flash = f"{html.escape(error)}
"
csrf = html.escape(user["csrf_token"], quote=True)
password_card = self.password_card(user, csrf)
if user["must_change_password"]:
content = f"""
{self.sidebar(user, csrf)}
{flash}{password_card}
{self.sidebar(user, csrf)}
模型配置中心 保存后，已登录桌面端会在启动或定时同步时自动应用。
CURRENT · v{int(row['version'])}
{users_card}{password_card}{audit_card}
当前仍在使用初始密码，必须修改后才能让桌面端登录。
"
if user["must_change_password"]
else ""
)
return f"""
"""
@staticmethod
def config_card(
user: sqlite3.Row, csrf: str, row: sqlite3.Row, config: dict[str, Any]
) -> str:
can_edit = user["role"] in ("admin", "operator")
esc = lambda key: html.escape(str(config.get(key, "")), quote=True)
checked = lambda key: " checked" if config.get(key) else ""
disabled = " disabled" if not can_edit else ""
mcp = html.escape(
json.dumps(config.get("AI_MCP_SERVERS", []), ensure_ascii=False, indent=2)
)
submit = (
"保存并发布配置
"
if can_edit
else "当前为只读角色，可查看配置但不能修改。
"
)
return f"""
最后更新：{html.escape(row['updated_at'])} · {html.escape(row['updated_by_name'] or 'system')}
"""
def users_card(self, user: sqlite3.Row, csrf: str) -> str:
if user["role"] != "admin":
return f""""""
items = []
role_options = lambda current: "".join(
f"{ROLE_LABELS[role]} "
for role in ROLES
)
for item in self.db.list_users():
active = " checked" if item["active"] else ""
items.append(
f""""""
)
return f"""
用户与角色 管理员管理全部功能，配置员可发布配置，只读用户仅同步。
新建用户
{html.escape(item['action'])} · {html.escape(item['username'] or 'anonymous')}"
f"
{html.escape(item['created_at'])} · {html.escape(item['detail'])}
{''.join(events) or '
暂无记录
'}