diff --git a/apps/admin/services.py b/apps/admin/services.py index 2c647e3..015558c 100644 --- a/apps/admin/services.py +++ b/apps/admin/services.py @@ -8,6 +8,7 @@ from apps.base.models import FileCodes, KeyValue from apps.base.utils import get_expire_info, get_file_path_name from fastapi import HTTPException from core.settings import data_root +from core.utils import hash_password, is_password_hashed class FileService: @@ -76,6 +77,9 @@ class ConfigService: if admin_token is None or admin_token == "": raise HTTPException(status_code=400, detail="管理员密码不能为空") + if not is_password_hashed(admin_token): + data["admin_token"] = hash_password(admin_token) + for key, value in data.items(): if key not in settings.default_config: continue diff --git a/apps/admin/views.py b/apps/admin/views.py index bbcc0c2..b531453 100644 --- a/apps/admin/views.py +++ b/apps/admin/views.py @@ -17,18 +17,16 @@ from core.response import APIResponse from apps.base.models import FileCodes, KeyValue from apps.admin.dependencies import create_token from core.settings import settings -from core.utils import get_now +from core.utils import get_now, verify_password admin_api = APIRouter(prefix="/admin", tags=["管理"]) @admin_api.post("/login") async def login(data: LoginData): - # 验证管理员密码 - if data.password != settings.admin_token: + if not verify_password(data.password, settings.admin_token): raise HTTPException(status_code=401, detail="密码错误") - # 生成包含管理员身份的token token = create_token({"is_admin": True}) return APIResponse(detail={"token": token, "token_type": "Bearer"}) diff --git a/core/utils.py b/core/utils.py index 8abe2d9..0aca619 100644 --- a/core/utils.py +++ b/core/utils.py @@ -47,7 +47,9 @@ async def get_select_token(code: str): :return: """ token = settings.admin_token - return hashlib.sha256(f"{code}{int(time.time() / 1000)}000{token}".encode()).hexdigest() + return hashlib.sha256( + f"{code}{int(time.time() / 1000)}000{token}".encode() + ).hexdigest() async def get_file_url(code: str): @@ -95,6 +97,44 @@ async def max_save_times_desc(max_save_seconds: int): return desc_zh, desc_en +def hash_password(password: str) -> str: + """ + 使用 SHA256 + salt 哈希密码 + 返回格式: sha256$$ + """ + salt = os.urandom(16).hex() + password_hash = hashlib.sha256(f"{salt}{password}".encode()).hexdigest() + return f"sha256${salt}${password_hash}" + + +def verify_password(password: str, hashed: str) -> bool: + """ + 验证密码是否匹配 + 支持新格式 (sha256$salt$hash) 和旧格式 (明文) + """ + if not hashed: + return False + + # 新格式: sha256$salt$hash + if hashed.startswith("sha256$"): + parts = hashed.split("$") + if len(parts) != 3: + return False + _, salt, stored_hash = parts + password_hash = hashlib.sha256(f"{salt}{password}".encode()).hexdigest() + return password_hash == stored_hash + + # 旧格式: 明文比较 (兼容迁移前的数据) + return password == hashed + + +def is_password_hashed(password: str) -> bool: + """ + 检查密码是否已经是哈希格式 + """ + return password.startswith("sha256$") and len(password.split("$")) == 3 + + async def sanitize_filename(filename: str) -> str: """ 安全处理文件名: @@ -105,15 +145,15 @@ async def sanitize_filename(filename: str) -> str: filename = os.path.basename(filename) illegal_chars = r'[\\/*?:"<>|\x00-\x1F]' # 包含控制字符 # 替换非法字符为下划线 - cleaned = re.sub(illegal_chars, '_', filename) + cleaned = re.sub(illegal_chars, "_", filename) # 处理空格(可选替换为_) - cleaned = cleaned.replace(' ', '_') + cleaned = cleaned.replace(" ", "_") # 处理连续下划线 - cleaned = re.sub(r'_+', '_', cleaned) + cleaned = re.sub(r"_+", "_", cleaned) # 处理首尾特殊字符 - cleaned = cleaned.strip('._') + cleaned = cleaned.strip("._") # 处理空文件名情况 if not cleaned: - cleaned = 'unnamed_file' + cleaned = "unnamed_file" # 长度限制(按需调整) return cleaned[:255] diff --git a/main.py b/main.py index a4ae4f4..051bb4a 100644 --- a/main.py +++ b/main.py @@ -22,6 +22,7 @@ from core.logger import logger from core.response import APIResponse from core.settings import data_root, settings, BASE_DIR, DEFAULT_CONFIG from core.tasks import delete_expire_files, clean_incomplete_uploads +from core.utils import hash_password, is_password_hashed @asynccontextmanager @@ -63,13 +64,26 @@ async def load_config(): key="sys_start", defaults={"value": int(time.time() * 1000)} ) settings.user_config = user_config.value - # 更新 ip_limit 配置 + + await migrate_password_to_hash() + ip_limit["error"].minutes = settings.errorMinute ip_limit["error"].count = settings.errorCount ip_limit["upload"].minutes = settings.uploadMinute ip_limit["upload"].count = settings.uploadCount +async def migrate_password_to_hash(): + if not is_password_hashed(settings.admin_token): + hashed = hash_password(settings.admin_token) + settings.admin_token = hashed + config_record = await KeyValue.filter(key="settings").first() + if config_record and config_record.value: + config_record.value["admin_token"] = hashed + await config_record.save() + logger.info("已将管理员密码迁移为哈希存储") + + app = FastAPI(lifespan=lifespan) app.add_middleware( @@ -149,5 +163,9 @@ if __name__ == "__main__": import uvicorn uvicorn.run( - app="main:app", host=settings.serverHost, port=settings.serverPort, reload=False, workers=settings.serverWorkers + app="main:app", + host=settings.serverHost, + port=settings.serverPort, + reload=False, + workers=settings.serverWorkers, )