feat(server): modularize backend and admin console

app/core/__init__.py

@@ -0,0 +1 @@
+"""Core configuration, security and middleware helpers."""

app/core/audit.py

@@ -0,0 +1,43 @@
+from fastapi import Request
+
+import db
+from app.core.security import token_digest
+
+
+# 管理后台写操作统一审计；不保存令牌明文和请求正文。
+async def admin_audit_middleware(request: Request, call_next):
+    should_audit = (
+        request.url.path.startswith("/admin/")
+        and request.method not in ("GET", "HEAD", "OPTIONS")
+        and not request.url.path.startswith("/admin/audit-log/")
+    )
+    status_code = 500
+    try:
+        response = await call_next(request)
+        status_code = response.status_code
+        return response
+    finally:
+        if should_audit:
+            try:
+                token = request.headers.get("X-Admin-Token", "")
+                conn = db.get_conn()
+                conn.execute(
+                    """INSERT INTO admin_audit_logs
+                    (actor_hash,action,method,path,target,result,status_code,ip,user_agent)
+                    VALUES(?,?,?,?,?,?,?,?,?)""",
+                    (
+                        token_digest(token)[:16] if token else "anonymous",
+                        request.url.path.removeprefix("/admin/"),
+                        request.method,
+                        request.url.path,
+                        request.url.query[:500],
+                        "success" if 200 <= status_code < 400 else "fail",
+                        status_code,
+                        request.client.host if request.client else "",
+                        request.headers.get("user-agent", "")[:300],
+                    ),
+                )
+                conn.commit()
+                conn.close()
+            except Exception as audit_error:
+                print("管理员审计日志写入失败:", audit_error)

app/core/config.py

@@ -0,0 +1,34 @@
+from pathlib import Path
+import os
+
+from dotenv import load_dotenv
+
+
+BASE_DIR = Path(__file__).resolve().parents[2]
+ENV_FILE_PATH = Path(os.getenv("ENV_FILE_PATH", str(BASE_DIR / ".env"))).expanduser()
+if not ENV_FILE_PATH.is_absolute():
+    ENV_FILE_PATH = BASE_DIR / ENV_FILE_PATH
+
+load_dotenv(ENV_FILE_PATH)
+
+
+def env_bool(name: str, default: bool = False) -> bool:
+    value = os.getenv(name)
+    return default if value is None else value.strip().lower() in {"1", "true", "yes", "on"}
+
+
+def configured_path(name: str, default: str) -> Path:
+    path = Path(os.getenv(name, default)).expanduser()
+    return path if path.is_absolute() else BASE_DIR / path
+
+
+class Settings:
+    """Centralized settings facade used during the backend template migration."""
+
+    server_host = os.getenv("SERVER_HOST", "0.0.0.0")
+    server_port = int(os.getenv("SERVER_PORT", "8000"))
+    service_title = os.getenv("SERVICE_TITLE", "软件自动升级服务")
+    admin_token = os.getenv("ADMIN_TOKEN", "").strip()
+
+
+settings = Settings()

app/core/security.py

@@ -0,0 +1,61 @@
+from pathlib import Path
+import hashlib
+import secrets
+
+from fastapi import Header, HTTPException
+
+from app.core.config import ENV_FILE_PATH, settings
+
+
+if not settings.admin_token:
+    raise RuntimeError("请在环境变量或 .env 中配置 ADMIN_TOKEN")
+
+
+def token_digest(token: str) -> str:
+    return hashlib.sha256(token.encode("utf-8")).hexdigest()
+
+
+def quote_env_value(value: str) -> str:
+    unsafe_chars = set(" \t\n\r#\"'")
+    if value and not any(ch in unsafe_chars for ch in value):
+        return value
+    return '"' + value.replace("\\", "\\\\").replace('"', '\\"') + '"'
+
+
+def persist_env_value(path: Path, key: str, value: str) -> bool:
+    line = f"{key}={quote_env_value(value)}\n"
+    if path.exists():
+        lines = path.read_text(encoding="utf-8").splitlines(keepends=True)
+        replaced = False
+        for index, item in enumerate(lines):
+            stripped = item.lstrip()
+            if stripped.startswith("#") or "=" not in stripped:
+                continue
+            name = stripped.split("=", 1)[0].strip()
+            if name == key:
+                lines[index] = line
+                replaced = True
+                break
+        if not replaced:
+            if lines and not lines[-1].endswith(("\n", "\r")):
+                lines[-1] += "\n"
+            lines.append(line)
+        path.write_text("".join(lines), encoding="utf-8")
+        return True
+    path.write_text(line, encoding="utf-8")
+    return True
+
+
+def admin_auth(X_Admin_Token: str = Header("")):
+    if not secrets.compare_digest(X_Admin_Token, settings.admin_token):
+        raise HTTPException(status_code=403, detail="后台密钥错误，禁止访问")
+    return True
+
+
+ADMIN_TOKEN = settings.admin_token
+
+
+def set_admin_token(new_token: str):
+    global ADMIN_TOKEN
+    settings.admin_token = new_token
+    ADMIN_TOKEN = new_token