62 lines
1.8 KiB
Python
62 lines
1.8 KiB
Python
|
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
|