115 lines
3.4 KiB
Python
115 lines
3.4 KiB
Python
"""Configuration read/write endpoints."""
|
|
|
|
from typing import Dict, Optional, Any
|
|
from fastapi import APIRouter, HTTPException, Depends
|
|
from pydantic import BaseModel
|
|
|
|
from ..auth.dependencies import get_current_user
|
|
|
|
router = APIRouter(prefix="/api/config", tags=["config"])
|
|
|
|
# Keys that should be masked in GET responses
|
|
_SENSITIVE_KEYS = {"api_key", "secret_key", "token", "password", "api_secret", "access_key"}
|
|
|
|
# Sections to expose (match actual config.ini)
|
|
_ALLOWED_SECTIONS = {"API", "Paths", "Performance", "File", "Templates", "Gitea", "WebAuth"}
|
|
|
|
|
|
class ConfigUpdate(BaseModel):
|
|
section: str
|
|
key: str
|
|
value: str
|
|
|
|
|
|
class ConfigBulkUpdate(BaseModel):
|
|
updates: list[ConfigUpdate]
|
|
|
|
|
|
def _get_config():
|
|
from app.config.settings import ConfigManager
|
|
return ConfigManager()
|
|
|
|
|
|
def _mask_value(key: str, value: str) -> str:
|
|
if any(s in key.lower() for s in _SENSITIVE_KEYS):
|
|
if len(value) > 4:
|
|
return value[:2] + "*" * (len(value) - 4) + value[-2:]
|
|
return "****"
|
|
return value
|
|
|
|
|
|
@router.get("")
|
|
async def get_config(
|
|
section: Optional[str] = None,
|
|
current_user: dict = Depends(get_current_user),
|
|
):
|
|
cfg = _get_config()
|
|
if section:
|
|
if section not in _ALLOWED_SECTIONS and section != "DEFAULT":
|
|
raise HTTPException(403, f"不允许访问配置节: {section}")
|
|
items = {}
|
|
for key, value in cfg.config.items(section):
|
|
items[key] = _mask_value(key, value)
|
|
return {"section": section, "items": items}
|
|
|
|
result = {}
|
|
for sec in _ALLOWED_SECTIONS:
|
|
try:
|
|
items = {}
|
|
for key, value in cfg.config.items(sec):
|
|
items[key] = _mask_value(key, value)
|
|
result[sec] = items
|
|
except Exception:
|
|
pass
|
|
return result
|
|
|
|
|
|
def _is_masked(key: str, value: str) -> bool:
|
|
"""Check if a value looks like a masked sensitive field (contains asterisks)."""
|
|
return any(s in key.lower() for s in _SENSITIVE_KEYS) and '*' in value
|
|
|
|
|
|
@router.put("")
|
|
async def update_config(
|
|
body: ConfigUpdate,
|
|
current_user: dict = Depends(get_current_user),
|
|
):
|
|
if body.section not in _ALLOWED_SECTIONS:
|
|
raise HTTPException(403, f"不允许修改配置节: {body.section}")
|
|
|
|
if _is_masked(body.key, body.value):
|
|
raise HTTPException(400, "敏感字段不能直接提交掩码值，请先清除输入框再输入真实值")
|
|
|
|
cfg = _get_config()
|
|
try:
|
|
cfg.update(body.section, body.key, body.value)
|
|
cfg.save_config()
|
|
return {"message": f"已更新 [{body.section}] {body.key}"}
|
|
except Exception as e:
|
|
raise HTTPException(500, f"保存失败: {e}")
|
|
|
|
|
|
@router.put("/bulk")
|
|
async def bulk_update_config(
|
|
body: ConfigBulkUpdate,
|
|
current_user: dict = Depends(get_current_user),
|
|
):
|
|
cfg = _get_config()
|
|
updated = []
|
|
skipped = []
|
|
for item in body.updates:
|
|
if item.section not in _ALLOWED_SECTIONS:
|
|
continue
|
|
# Skip masked sensitive values to prevent destroying real credentials
|
|
if _is_masked(item.key, item.value):
|
|
skipped.append(f"[{item.section}] {item.key}")
|
|
continue
|
|
cfg.update(item.section, item.key, item.value)
|
|
updated.append(f"[{item.section}] {item.key}")
|
|
|
|
cfg.save_config()
|
|
msg = f"已更新 {len(updated)} 项"
|
|
if skipped:
|
|
msg += f"，跳过 {len(skipped)} 项掩码值"
|
|
return {"message": msg, "updated": updated, "skipped": skipped}
|