leonxie/v1-backend
Archived
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

为admin修改文章组件添加remarkbreaks，修改admin路由鉴权机制

Browse Source
This commit is contained in:
LeonspaceX
2025-12-02 22:33:14 +08:00
parent c4fc6c0084
commit 0e8b762b22
1 changed files with 36 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

47
api_server.py
View File

@@ -90,11 +90,11 @@ def set_config(key, value):
# === 变量 ===
DEFAULT_BANNED_KEYWORDS = [
"傻逼"
"default"
]
# === 延迟初始化配置 ===
DEFAULT_ADMIN_TOKEN = "Sycamore_whisper"
DEFAULT_ADMIN_TOKEN_HASH = hashlib.sha256("Sycamore_whisper".encode('utf-8')).hexdigest()
DEFAULT_UPLOAD_FOLDER = "img"
DEFAULT_ALLOWED_EXTENSIONS = {"png", "jpg", "jpeg", "gif", "webp"}
DEFAULT_MAX_FILE_SIZE = 10 * 1024 * 1024 # 10 MB
@@ -105,7 +105,7 @@ INIT = False
NEED_AUDIT = False
# 运行时使用的变量，初始为默认值
ADMIN_TOKEN = DEFAULT_ADMIN_TOKEN
ADMIN_TOKEN_HASH = DEFAULT_ADMIN_TOKEN_HASH
UPLOAD_FOLDER = DEFAULT_UPLOAD_FOLDER
ALLOWED_EXTENSIONS = set(DEFAULT_ALLOWED_EXTENSIONS)
MAX_FILE_SIZE = DEFAULT_MAX_FILE_SIZE
@@ -123,8 +123,8 @@ def allowed_backup_file(filename):
return '.' in filename and filename.rsplit('.', 1)[1].lower() in ALLOWED_BACKUP_EXTENSIONS
def apply_config_to_globals():
global ADMIN_TOKEN, UPLOAD_FOLDER, ALLOWED_EXTENSIONS, MAX_FILE_SIZE, IMG_FOLDER, BANNED_KEYWORDS, RATE_LIMIT
ADMIN_TOKEN = CONFIG.get('ADMIN_TOKEN', DEFAULT_ADMIN_TOKEN)
global ADMIN_TOKEN_HASH, UPLOAD_FOLDER, ALLOWED_EXTENSIONS, MAX_FILE_SIZE, IMG_FOLDER, BANNED_KEYWORDS, RATE_LIMIT
ADMIN_TOKEN_HASH = CONFIG.get('ADMIN_TOKEN_HASH', DEFAULT_ADMIN_TOKEN_HASH)
UPLOAD_FOLDER = CONFIG.get('UPLOAD_FOLDER', DEFAULT_UPLOAD_FOLDER)
ALLOWED_EXTENSIONS = set(CONFIG.get('ALLOWED_EXTENSIONS', DEFAULT_ALLOWED_EXTENSIONS))
MAX_FILE_SIZE = int(CONFIG.get('MAX_FILE_SIZE', DEFAULT_MAX_FILE_SIZE))
@@ -142,8 +142,29 @@ def load_config():
cfg = importlib.reload(sys.modules['config'])
else:
cfg = importlib.import_module('config')
# 兼容性处理：如果只有 ADMIN_TOKEN 没有 ADMIN_TOKEN_HASH，则迁移
admin_token_hash = getattr(cfg, 'ADMIN_TOKEN_HASH', None)
if admin_token_hash is None:
# 尝试从旧的 ADMIN_TOKEN 迁移
old_token = getattr(cfg, 'ADMIN_TOKEN', None)
if old_token is not None:
# 对旧token做哈希
admin_token_hash = hashlib.sha256(old_token.encode('utf-8')).hexdigest()
# 重写配置文件，使用哈希值
upload_folder = getattr(cfg, 'UPLOAD_FOLDER', DEFAULT_UPLOAD_FOLDER)
allowed_extensions = set(getattr(cfg, 'ALLOWED_EXTENSIONS', DEFAULT_ALLOWED_EXTENSIONS))
max_file_size = int(getattr(cfg, 'MAX_FILE_SIZE', DEFAULT_MAX_FILE_SIZE))
banned_keywords = list(getattr(cfg, 'BANNED_KEYWORDS', DEFAULT_BANNED_KEYWORDS))
rate_limit = int(getattr(cfg, 'RATE_LIMIT', DEFAULT_RATE_LIMIT))
write_config_py(admin_token_hash, upload_folder, allowed_extensions, max_file_size, banned_keywords, rate_limit)
# 重新加载配置
importlib.invalidate_caches()
cfg = importlib.reload(sys.modules['config'])
admin_token_hash = getattr(cfg, 'ADMIN_TOKEN_HASH')
CONFIG = {
'ADMIN_TOKEN': getattr(cfg, 'ADMIN_TOKEN'),
'ADMIN_TOKEN_HASH': admin_token_hash,
'UPLOAD_FOLDER': getattr(cfg, 'UPLOAD_FOLDER'),
'ALLOWED_EXTENSIONS': set(getattr(cfg, 'ALLOWED_EXTENSIONS')),
'MAX_FILE_SIZE': int(getattr(cfg, 'MAX_FILE_SIZE')),
@@ -174,7 +195,7 @@ def gate_uninitialized():
if not INIT:
return jsonify({"status": "Fail", "reason": "Uninitialized"}), 503
def write_config_py(token, upload_folder, allowed_exts, max_file_size, banned_keywords=None, rate_limit=DEFAULT_RATE_LIMIT):
def write_config_py(token_hash, upload_folder, allowed_exts, max_file_size, banned_keywords=None, rate_limit=DEFAULT_RATE_LIMIT):
# 归一化扩展名为小写且唯一
exts = sorted(set(str(e).strip().lower() for e in allowed_exts if str(e).strip()))
# 归一化敏感词为去空格的字符串列表
@@ -182,7 +203,7 @@ def write_config_py(token, upload_folder, allowed_exts, max_file_size, banned_ke
banned = [str(w).strip() for w in banned if str(w).strip()]
content = (
"# Auto-generated by /init\n"
f"ADMIN_TOKEN = {repr(token)}\n"
f"ADMIN_TOKEN_HASH = {repr(token_hash)}\n"
f"UPLOAD_FOLDER = {repr(upload_folder)}\n"
f"ALLOWED_EXTENSIONS = {repr(exts)}\n"
f"MAX_FILE_SIZE = {int(max_file_size)}\n"
@@ -204,7 +225,9 @@ def init_service():
if missing:
return jsonify({"status": "Fail", "reason": f"Missing fields: {', '.join(missing)}"}), 400
# 接收明文 token，在后端做哈希处理
token = str(data["ADMIN_TOKEN"])
token_hash = hashlib.sha256(token.encode('utf-8')).hexdigest()
upload_folder = str(data["UPLOAD_FOLDER"]).strip()
# 接受 list 或 逗号字符串
exts = data["ALLOWED_EXTENSIONS"]
@@ -238,7 +261,7 @@ def init_service():
return jsonify({"status": "Fail", "reason": "BANNED_KEYWORDS must be list or comma string"}), 400
try:
write_config_py(token, upload_folder, allowed_exts, max_file_size, banned_keywords, rate_limit)
write_config_py(token_hash, upload_folder, allowed_exts, max_file_size, banned_keywords, rate_limit)
load_config()
initialize_database()
try:
@@ -338,7 +361,9 @@ def require_admin(func):
if not auth_header.startswith("Bearer "):
return jsonify({"status": "Fail", "reason": "Token invalid"}), 401
token = auth_header.split(" ", 1)[1]
if token != ADMIN_TOKEN:
# 对传入的token做哈希后与配置文件中的哈希值比对
token_hash = hashlib.sha256(token.encode('utf-8')).hexdigest()
if token_hash != ADMIN_TOKEN_HASH:
return jsonify({"status": "Fail", "reason": "Token invalid"}), 403
return func(*args, **kwargs)
return wrapper
@@ -729,7 +754,7 @@ def set_banned_keywords():
BANNED_KEYWORDS = new_keywords
try:
# 重写配置文件，确保重启后仍生效
write_config_py(ADMIN_TOKEN, UPLOAD_FOLDER, list(ALLOWED_EXTENSIONS), MAX_FILE_SIZE, BANNED_KEYWORDS)
write_config_py(ADMIN_TOKEN_HASH, UPLOAD_FOLDER, list(ALLOWED_EXTENSIONS), MAX_FILE_SIZE, BANNED_KEYWORDS)
load_config()
return jsonify({"status": "OK"}), 200
except Exception as e: