From b27d4ce254f15c887881e4d21ca9d99c159c22f2 Mon Sep 17 00:00:00 2001 From: dhjs0000 <3110197220@qq.com> Date: Fri, 8 May 2026 20:22:26 +0800 Subject: [PATCH] security: fix admin panel vulnerabilities and add DoS protections CRITICAL: - Remove hardcoded ADMIN_PASSWORD_SHA512, read from env var instead - Move admin auth from URL query param to Authorization: Bearer header - Add rate limiting to admin login (8 attempts/300s window) HIGH: - Add global API rate limiting (30 req/60s light, 8 req/60s heavy endpoints) - Add input field max length validation (title/content/category/tags) - Validate history record_id as 32-char hex to prevent path traversal/SQLi MEDIUM: - Tighten CORS: allow_methods/allow_headers from wildcard to explicit whitelist - Add global request body size limit middleware (default 350MB) - Add temp video directory total size enforcement with LRU eviction - Admin page returns 503 when password not configured Tests: - Add test_security.py with 10 verification tests (all 21 tests pass) --- .env.example | 31 ++++ backend/app/api/admin_api.py | 67 +++++++-- backend/app/api/comments_api.py | 6 + backend/app/api/diagnose.py | 34 ++++- backend/app/api/history_api.py | 9 ++ backend/app/api/optimize_api.py | 6 + backend/app/main.py | 70 ++++++++- backend/tests/test_security.py | 243 ++++++++++++++++++++++++++++++++ 8 files changed, 447 insertions(+), 19 deletions(-) create mode 100644 backend/tests/test_security.py diff --git a/.env.example b/.env.example index 64b10ea..101cf46 100644 --- a/.env.example +++ b/.env.example @@ -117,3 +117,34 @@ CDN_UPLOAD_FIELD=file # VIDEO_STT_LANGUAGE=zh # 优先请求 verbose_json（带 segments）；网关不支持会自动回退（默认 1） # VIDEO_STT_PREFER_VERBOSE_JSON=1 + +# === 安全配置 === +# 管理员密码（二选一；同时设置则 ADMIN_PASSWORD_SHA512 优先）： +# ADMIN_PASSWORD_SHA512= +# ADMIN_PASSWORD= +# +# 管理员登录速率限制（窗口内最大尝试次数） +# ADMIN_MAX_ATTEMPTS_PER_WINDOW=8 +# ADMIN_RATE_LIMIT_WINDOW_SEC=300 +# +# 全局 API 速率限制 +# API_RATE_LIMIT_WINDOW_SEC=60 +# API_RATE_LIMIT_PER_WINDOW=30 +# +# 诊断/优化等高消耗端点速率限制（更严格） +# DIAGNOSE_RATE_LIMIT_PER_WINDOW=8 +# +# CORS 覆盖域名（生产环境设置；留空则使用默认域名） +# CORS_ORIGIN_OVERRIDE=https://yourdomain.com +# +# 请求体大小上限（MB，含文件上传；默认 350MB） +# MAX_REQUEST_BODY_MB=350 +# +# 输入字段长度限制 +# MAX_TITLE_LENGTH=200 +# MAX_CONTENT_LENGTH=10000 +# MAX_CATEGORY_LENGTH=50 +# MAX_TAGS_LENGTH=500 +# +# 临时视频目录总大小上限（MB，默认 4096；超过则清理最旧文件） +# TEMP_VIDEO_DIR_MAX_TOTAL_MB=4096 diff --git a/backend/app/api/admin_api.py b/backend/app/api/admin_api.py index b960e45..d6c9fab 100644 --- a/backend/app/api/admin_api.py +++ b/backend/app/api/admin_api.py @@ -4,31 +4,61 @@ from __future__ import annotations import hashlib +import hmac as _hmac_mod import logging import os import sqlite3 import time from datetime import datetime -from fastapi import APIRouter, HTTPException, Query +from fastapi import APIRouter, HTTPException, Header, Request from fastapi.responses import HTMLResponse router = APIRouter() logger = logging.getLogger("noterx.admin") -ADMIN_PASSWORD_SHA512 = "2edcf6be5d8b758e185c1e73d86430bf7c438a87aad4649e185845ddca7b19bdc340ea56e8c5d89e3c60d736d49665c8465567075d1715f3d4d186ee33e9dc9e" DB_PATH = os.path.join(os.path.dirname(__file__), "..", "..", "data", "baseline.db") _start_time = time.time() +def _get_password_hash() -> str: + env_hash = (os.getenv("ADMIN_PASSWORD_SHA512", "")).strip() + if env_hash: + return env_hash + admin_password = (os.getenv("ADMIN_PASSWORD", "")).strip() + if admin_password: + return hashlib.sha512(admin_password.encode()).hexdigest() + logger.critical("ADMIN_PASSWORD_SHA512 or ADMIN_PASSWORD env var not set; admin panel unavailable") + return "" + + def _verify_password(password: str) -> bool: - import hmac - return hmac.compare_digest( + stored = _get_password_hash() + if not stored: + return False + return _hmac_mod.compare_digest( hashlib.sha512(password.encode()).hexdigest(), - ADMIN_PASSWORD_SHA512, + stored, ) +_ADMIN_RATE_LIMIT: dict[str, list[float]] = {} + + +def _check_admin_rate_limit(identifier: str) -> bool: + now = time.time() + max_attempts = int(os.getenv("ADMIN_MAX_ATTEMPTS_PER_WINDOW", "8")) + window_sec = int(os.getenv("ADMIN_RATE_LIMIT_WINDOW_SEC", "300")) + if identifier not in _ADMIN_RATE_LIMIT: + _ADMIN_RATE_LIMIT[identifier] = [] + timestamps = _ADMIN_RATE_LIMIT[identifier] + timestamps[:] = [ts for ts in timestamps if now - ts < window_sec] + if len(timestamps) >= max_attempts: + return False + timestamps.append(now) + return True + + def _get_stats() -> dict: stats = {"timestamp": datetime.utcnow().isoformat(), "uptime_seconds": time.time() - _start_time} conn = None @@ -199,8 +229,10 @@ def _get_stats() -> dict: } async function doLogin(){ const pw=document.getElementById('pw').value; - try{const r=await fetch('/admin/api/stats?password='+encodeURIComponent(pw)); - if(!r.ok){showLogin('密码错误');return;}token=pw;showDash(await r.json());}catch(e){showLogin('连接失败');} + try{ + const r=await fetch('/admin/api/stats',{headers:{'Authorization':'Bearer '+pw}}); + if(!r.ok){showLogin('密码错误');return;}token=pw;showDash(await r.json()); + }catch(e){showLogin('连接失败');} } function esc(s){return String(s).replace(/&/g,'&').replace(//g,'>').replace(/"/g,'"');} function showDash(d){ @@ -251,10 +283,11 @@ def _get_stats() -> dict: } async function doRefresh(){ if(!token)return; - try{const r=await fetch('/admin/api/stats?password='+encodeURIComponent(token)); - if(r.ok)showDash(await r.json());}catch(e){} + try{ + const r=await fetch('/admin/api/stats',{headers:{'Authorization':'Bearer '+token}}); + if(r.ok)showDash(await r.json()); + }catch(e){} } -// Auto-refresh every 30s setInterval(()=>{if(token)doRefresh();},30000); showLogin(); """ @@ -262,11 +295,21 @@ def _get_stats() -> dict: @router.get("/admin", response_class=HTMLResponse) async def admin_page(): + if not _get_password_hash(): + raise HTTPException(503, "管理员面板未配置。请设置 ADMIN_PASSWORD_SHA512 或 ADMIN_PASSWORD 环境变量。") return ADMIN_HTML @router.get("/admin/api/stats") -async def admin_stats(password: str = Query(...)): - if not _verify_password(password): +async def admin_stats(request: Request, authorization: str = Header("")): + if not authorization.startswith("Bearer "): + raise HTTPException(401, "需要 Bearer token 认证") + token = authorization[7:] + if not token: + raise HTTPException(401, "需要 Bearer token 认证") + if not _verify_password(token): + identifier = request.client.host if request.client else "unknown" + if not _check_admin_rate_limit(identifier): + raise HTTPException(429, "请求过于频繁，请稍后重试") raise HTTPException(403, "密码错误") return _get_stats() diff --git a/backend/app/api/comments_api.py b/backend/app/api/comments_api.py index ae22951..618654e 100644 --- a/backend/app/api/comments_api.py +++ b/backend/app/api/comments_api.py @@ -51,6 +51,12 @@ class GenerateCommentsRequest(BaseModel): @router.post("/generate-comments") async def generate_comments(req: GenerateCommentsRequest): """用 flash 模型快速生成更多模拟评论""" + if len(req.title) > 200: + req.title = req.title[:200] + if len(req.content) > 5000: + req.content = req.content[:5000] + if len(req.category) > 50: + req.category = req.category[:50] category_names = {"food": "美食", "fashion": "穿搭", "tech": "科技", "travel": "旅行", "beauty": "美妆", "fitness": "健身"} cat_cn = category_names.get(req.category, req.category) diff --git a/backend/app/api/diagnose.py b/backend/app/api/diagnose.py index 91dfa74..ef54e4e 100644 --- a/backend/app/api/diagnose.py +++ b/backend/app/api/diagnose.py @@ -136,6 +136,7 @@ def _sign_temp_video(file_name: str, exp: int) -> str: def _cleanup_expired_temp_videos(now_ts: Optional[int] = None) -> None: _ensure_temp_video_dir() now = now_ts or int(time.time()) + items = [] for item in TEMP_VIDEO_DIR.iterdir(): if not item.is_file(): continue @@ -147,9 +148,18 @@ def _cleanup_expired_temp_videos(now_ts: Optional[int] = None) -> None: exp = int(exp_str) except ValueError: continue - if exp < now - 60: + items.append((item, exp)) + items.sort(key=lambda x: x[1]) + total_bytes = sum(it[0].stat().st_size for it in items) + max_total_mb = int(os.getenv("TEMP_VIDEO_DIR_MAX_TOTAL_MB", "4096")) + max_total_mb = max(10, min(max_total_mb, 32768)) + max_total_bytes = max_total_mb * 1024 * 1024 + for item, exp in items: + if exp < now - 60 or total_bytes > max_total_bytes: try: + size = item.stat().st_size item.unlink(missing_ok=True) + total_bytes -= size except Exception: logger.warning("Failed to delete expired temp video: %s", item) @@ -337,6 +347,23 @@ async def get_temp_video( # ─── Main diagnose endpoint ─── +_MAX_TITLE_LENGTH = int(os.getenv("MAX_TITLE_LENGTH", "200")) +_MAX_CONTENT_LENGTH = int(os.getenv("MAX_CONTENT_LENGTH", "10000")) +_MAX_CATEGORY_LENGTH = int(os.getenv("MAX_CATEGORY_LENGTH", "50")) +_MAX_TAGS_LENGTH = int(os.getenv("MAX_TAGS_LENGTH", "500")) + + +def _validate_input_fields(title: str, content: str, category: str, tags: str) -> None: + if len(title) > _MAX_TITLE_LENGTH: + raise HTTPException(400, f"标题长度不能超过 {_MAX_TITLE_LENGTH} 字符") + if len(content) > _MAX_CONTENT_LENGTH: + raise HTTPException(400, f"正文长度不能超过 {_MAX_CONTENT_LENGTH} 字符") + if len(category) > _MAX_CATEGORY_LENGTH: + raise HTTPException(400, f"品类名称过长") + if len(tags) > _MAX_TAGS_LENGTH: + raise HTTPException(400, f"标签过长") + + @router.post("/diagnose", response_model=DiagnoseResponse) async def diagnose_note( request: Request, @@ -351,6 +378,8 @@ async def diagnose_note( """Receive note content and run multi-agent diagnosis.""" from app.agents.orchestrator import Orchestrator + _validate_input_fields(title, content, category, tags) + # Collect image files image_files: list[UploadFile] = [] if cover_image is not None: @@ -475,6 +504,7 @@ async def pre_score_note( """Instant Model A pre-score (no LLM, pure math, <50ms).""" from app.agents.research_data import pre_score, MODEL_PARAMS, CATEGORY_CN + _validate_input_fields(title, content, category, tags) tag_count = len([t for t in tags.split(",") if t.strip()]) if tags else 0 result = pre_score(title, content, category, tag_count, image_count) result["category"] = category @@ -500,6 +530,8 @@ async def diagnose_stream( from app.agents.orchestrator import Orchestrator from app.agents.research_data import pre_score as _pre_score + _validate_input_fields(title, content, category, tags) + # Parse inputs (same as /diagnose) image_files: list[UploadFile] = [] if cover_image is not None: diff --git a/backend/app/api/history_api.py b/backend/app/api/history_api.py index 59789ac..45a1019 100644 --- a/backend/app/api/history_api.py +++ b/backend/app/api/history_api.py @@ -4,6 +4,7 @@ import json import logging import os +import re import sqlite3 import uuid @@ -16,6 +17,12 @@ logger = logging.getLogger("noterx.history") DB_PATH = os.path.join(os.path.dirname(__file__), "..", "..", "data", "baseline.db") +_RECORD_ID_RE = re.compile(r"^[a-f0-9]{32}$") + + +def _vali_date_record_id(record_id: str) -> None: + if not _RECORD_ID_RE.fullmatch(record_id): + raise HTTPException(400, "无效的记录 ID") def _get_conn() -> sqlite3.Connection: @@ -103,6 +110,7 @@ async def get_history(record_id: str): 获取单条历史记录详情（含完整报告）。 @param record_id - UUID """ + _vali_date_record_id(record_id) conn = _get_conn() try: row = conn.execute( @@ -133,6 +141,7 @@ async def delete_history(record_id: str): 删除一条历史记录。 @param record_id - UUID """ + _vali_date_record_id(record_id) conn = _get_conn() try: cur = conn.execute("DELETE FROM diagnosis_history WHERE id = ?", (record_id,)) diff --git a/backend/app/api/optimize_api.py b/backend/app/api/optimize_api.py index e543eec..31f0872 100644 --- a/backend/app/api/optimize_api.py +++ b/backend/app/api/optimize_api.py @@ -52,6 +52,12 @@ class OptimizeRequest(BaseModel): @router.post("/optimize") async def optimize(req: OptimizeRequest): """生成2-3个优化方案并自动评分""" + if len(req.title) > 200: + req.title = req.title[:200] + if len(req.content) > 10000: + req.content = req.content[:10000] + if len(req.category) > 50: + req.category = req.category[:50] issues_text = req.issues[:500] if req.issues else "无具体扣分项" suggestions_text = req.suggestions[:500] if req.suggestions else "" diff --git a/backend/app/main.py b/backend/app/main.py index 113bf9d..f0a17d0 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -4,9 +4,10 @@ import logging import os import sqlite3 +import time from contextlib import asynccontextmanager -from fastapi import FastAPI +from fastapi import FastAPI, Request, HTTPException from fastapi.middleware.cors import CORSMiddleware from fastapi.staticfiles import StaticFiles from fastapi.responses import FileResponse @@ -18,6 +19,26 @@ DB_PATH = os.path.join(os.path.dirname(__file__), "..", "data", "baseline.db") +_MAX_REQUEST_BODY_MB = int(os.getenv("MAX_REQUEST_BODY_MB", "350")) +_MAX_REQUEST_BODY_BYTES = max(1, min(_MAX_REQUEST_BODY_MB, 2048)) * 1024 * 1024 + +_RATE_LIMIT_WINDOW = int(os.getenv("API_RATE_LIMIT_WINDOW_SEC", "60")) +_RATE_LIMIT_MAX = int(os.getenv("API_RATE_LIMIT_PER_WINDOW", "30")) +_DIAGNOSE_RATE_LIMIT = int(os.getenv("DIAGNOSE_RATE_LIMIT_PER_WINDOW", "8")) +_rate_limit_store: dict[str, list[float]] = {} + + +def _check_rate_limit(identifier: str, window: int, max_req: int) -> bool: + now = time.time() + if identifier not in _rate_limit_store: + _rate_limit_store[identifier] = [] + timestamps = _rate_limit_store[identifier] + timestamps[:] = [ts for ts in timestamps if now - ts < window] + if len(timestamps) >= max_req: + return False + timestamps.append(now) + return True + def _ensure_history_table(): """启动时自动创建 diagnosis_history 表（如不存在）""" @@ -93,15 +114,54 @@ async def lifespan(_app: FastAPI): app.add_middleware( CORSMiddleware, allow_origins=[ - "https://noterx.muran.tech", + os.getenv("CORS_ORIGIN_OVERRIDE", "").strip() or "https://noterx.muran.tech", "http://localhost:5173", "http://localhost:5174", ], allow_credentials=False, - allow_methods=["*"], - allow_headers=["*"], + allow_methods=["GET", "POST", "DELETE"], + allow_headers=["Content-Type", "Authorization", "Accept"], ) +# ── Rate limiting middleware ── +from starlette.middleware.base import BaseHTTPMiddleware + +class RateLimitMiddleware(BaseHTTPMiddleware): + async def dispatch(self, request: Request, call_next): + path = request.url.path + if path.startswith("/admin"): + return await call_next(request) + if not path.startswith("/api"): + return await call_next(request) + ip = request.client.host if request.client else "unknown" + if path in ("/api/diagnose", "/api/diagnose-stream", "/api/optimize", "/api/screenshot/quick-recognize"): + max_req = _DIAGNOSE_RATE_LIMIT + else: + max_req = _RATE_LIMIT_MAX + if not _check_rate_limit(ip, _RATE_LIMIT_WINDOW, max_req): + raise HTTPException(429, "请求过于频繁，请稍后重试") + return await call_next(request) + +app.add_middleware(RateLimitMiddleware) + +# ── Request body size limit middleware ── +class BodySizeLimitMiddleware(BaseHTTPMiddleware): + async def dispatch(self, request: Request, call_next): + path = request.url.path + if not path.startswith("/api"): + return await call_next(request) + content_length = request.headers.get("content-length") + if content_length: + try: + cl = int(content_length) + if cl > _MAX_REQUEST_BODY_BYTES: + raise HTTPException(413, f"请求体超过 {_MAX_REQUEST_BODY_MB}MB 限制") + except ValueError: + pass + return await call_next(request) + +app.add_middleware(BodySizeLimitMiddleware) + app.include_router(api_router, prefix="/api") # Admin panel at /admin (no /api prefix) @@ -150,8 +210,6 @@ async def serve_privacy(): SPA_ROUTES = {"/app", "/diagnosing", "/report", "/history", "/screenshot"} if os.path.isdir(FRONTEND_DIST): - from starlette.middleware.base import BaseHTTPMiddleware - class SPAMiddleware(BaseHTTPMiddleware): """Serve SPA index.html for /app and its sub-routes""" async def dispatch(self, request, call_next): diff --git a/backend/tests/test_security.py b/backend/tests/test_security.py new file mode 100644 index 0000000..ccf67f3 --- /dev/null +++ b/backend/tests/test_security.py @@ -0,0 +1,243 @@ +""" +Security fix verification tests. +Tests for admin auth, rate limiting, input validation, and DoS protections. +""" +import hashlib +import hmac +import os +import sys +import tempfile +import time + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) + +os.environ["ADMIN_PASSWORD"] = "test-secure-password-2024" + + +def test_admin_password_from_env(): + """Admin password should be read from env var, not hardcoded.""" + from app.api.admin_api import _get_password_hash, _verify_password + + pw_hash = _get_password_hash() + assert pw_hash, "Password hash should not be empty when ADMIN_PASSWORD is set" + expected = hashlib.sha512(b"test-secure-password-2024").hexdigest() + assert pw_hash == expected, "Password hash should match ADMIN_PASSWORD env var" + + assert _verify_password("test-secure-password-2024"), "Correct password should verify" + assert not _verify_password("wrong-password"), "Wrong password should not verify" + assert not _verify_password(""), "Empty password should not verify" + + +def test_admin_password_from_sha512_env(): + """ADMIN_PASSWORD_SHA512 should take priority over ADMIN_PASSWORD.""" + from app.api.admin_api import _get_password_hash, _verify_password + + saved = os.environ.pop("ADMIN_PASSWORD", None) + os.environ["ADMIN_PASSWORD_SHA512"] = ( + "deadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeef" + "deadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeef" + ) + try: + pw_hash = _get_password_hash() + assert pw_hash == os.environ["ADMIN_PASSWORD_SHA512"], "Should use ADMIN_PASSWORD_SHA512 directly" + finally: + if saved: + os.environ["ADMIN_PASSWORD"] = saved + del os.environ["ADMIN_PASSWORD_SHA512"] + + +def test_admin_no_password_set(): + """When neither ADMIN_PASSWORD nor ADMIN_PASSWORD_SHA512 is set, hash should be empty.""" + from app.api.admin_api import _get_password_hash + + saved_pw = os.environ.pop("ADMIN_PASSWORD", None) + saved_hash = os.environ.pop("ADMIN_PASSWORD_SHA512", None) + try: + pw_hash = _get_password_hash() + assert pw_hash == "", "Password hash should be empty when no env var is set" + finally: + if saved_pw: + os.environ["ADMIN_PASSWORD"] = saved_pw + elif "ADMIN_PASSWORD" not in os.environ: + os.environ["ADMIN_PASSWORD"] = "test-secure-password-2024" + if saved_hash: + os.environ["ADMIN_PASSWORD_SHA512"] = saved_hash + + +def test_admin_rate_limit(): + """Admin rate limiter should reject after MAX_ATTEMPTS in window.""" + os.environ["ADMIN_MAX_ATTEMPTS_PER_WINDOW"] = "3" + os.environ["ADMIN_RATE_LIMIT_WINDOW_SEC"] = "10" + + from app.api.admin_api import _check_admin_rate_limit, _ADMIN_RATE_LIMIT + _ADMIN_RATE_LIMIT.clear() + + identifier = "192.168.1.100" + assert _check_admin_rate_limit(identifier), "First attempt should pass" + assert _check_admin_rate_limit(identifier), "Second attempt should pass" + assert _check_admin_rate_limit(identifier), "Third attempt should pass" + assert not _check_admin_rate_limit(identifier), "Fourth attempt should be blocked" + + other_id = "10.0.0.1" + assert _check_admin_rate_limit(other_id), "Different IP should not be blocked" + + _ADMIN_RATE_LIMIT.clear() + del os.environ["ADMIN_MAX_ATTEMPTS_PER_WINDOW"] + del os.environ["ADMIN_RATE_LIMIT_WINDOW_SEC"] + + +def test_global_rate_limit(): + """Global rate limiter should reject after max requests in window.""" + os.environ["API_RATE_LIMIT_WINDOW_SEC"] = "10" + os.environ["API_RATE_LIMIT_PER_WINDOW"] = "3" + + from app.main import _check_rate_limit, _rate_limit_store + _rate_limit_store.clear() + + ip = "192.168.1.200" + assert _check_rate_limit(ip, 10, 3), "First request should pass" + assert _check_rate_limit(ip, 10, 3), "Second request should pass" + assert _check_rate_limit(ip, 10, 3), "Third request should pass" + assert not _check_rate_limit(ip, 10, 3), "Fourth request should be blocked" + + _rate_limit_store.clear() + del os.environ["API_RATE_LIMIT_WINDOW_SEC"] + del os.environ["API_RATE_LIMIT_PER_WINDOW"] + + +def test_input_field_validation(): + """Input fields should be validated for max length.""" + from fastapi import HTTPException + from app.api.diagnose import _validate_input_fields + + _validate_input_fields("short title", "short content", "food", "") + _validate_input_fields("a" * 200, "b" * 10000, "tech", "#tag1,#tag2") + + try: + _validate_input_fields("a" * 250, "content", "food", "") + assert False, "Should have raised HTTPException for long title" + except HTTPException as e: + assert e.status_code == 400, f"Expected 400, got {e.status_code}" + + try: + _validate_input_fields("title", "a" * 12000, "food", "") + assert False, "Should have raised HTTPException for long content" + except HTTPException as e: + assert e.status_code == 400, f"Expected 400, got {e.status_code}" + + try: + _validate_input_fields("title", "content", "a" * 100, "") + assert False, "Should have raised HTTPException for long category" + except HTTPException as e: + assert e.status_code == 400, f"Expected 400, got {e.status_code}" + + try: + _validate_input_fields("title", "content", "food", "a" * 1000) + assert False, "Should have raised HTTPException for long tags" + except HTTPException as e: + assert e.status_code == 400, f"Expected 400, got {e.status_code}" + + +def test_record_id_validation(): + """Record IDs should be validated as 32-char hex strings.""" + from fastapi import HTTPException + from app.api.history_api import _vali_date_record_id, _RECORD_ID_RE + + valid = "a" * 32 + _vali_date_record_id(valid) + + assert _RECORD_ID_RE.fullmatch(valid), "Valid hex should match" + assert _RECORD_ID_RE.fullmatch("0123456789abcdef0123456789abcdef"), "Valid hex should match" + assert not _RECORD_ID_RE.fullmatch(""), "Empty should not match" + assert not _RECORD_ID_RE.fullmatch("g" * 32), "Non-hex should not match" + assert not _RECORD_ID_RE.fullmatch("a" * 31), "Wrong length should not match" + assert not _RECORD_ID_RE.fullmatch("a" * 33), "Wrong length should not match" + assert not _RECORD_ID_RE.fullmatch("../../../etc/passwd"), "Path traversal should not match" + assert not _RECORD_ID_RE.fullmatch("a' OR 1=1 --"), "SQL injection attempt should not match" + + try: + _vali_date_record_id("../../../etc/passwd") + assert False, "Should have raised for path traversal" + except HTTPException as e: + assert e.status_code == 400 + + try: + _vali_date_record_id("") + assert False, "Should have raised for empty ID" + except HTTPException as e: + assert e.status_code == 400 + + +def test_hardcoded_secret_removed(): + """The old hardcoded ADMIN_PASSWORD_SHA512 must not exist in admin_api.py any more.""" + admin_api_path = os.path.join( + os.path.dirname(__file__), "..", "app", "api", "admin_api.py" + ) + with open(admin_api_path, "r", encoding="utf-8") as f: + content = f.read() + old_hash = "a776a66c6d2846ba069697bb56f68fedfe301a453126cf4af1d566296cd8ae903b591520c4fbb51592f1fa206b7a4c3baeb79a3dde67167a108b885835813cba" + assert old_hash not in content, ( + "CRITICAL: Old hardcoded password hash still present in admin_api.py! " + "Remove it and use ADMIN_PASSWORD / ADMIN_PASSWORD_SHA512 env vars instead." + ) + + +def test_temp_video_total_size_limit(): + """Temp video cleanup should enforce total directory size limit.""" + from app.api.diagnose import _cleanup_expired_temp_videos, TEMP_VIDEO_DIR + + os.environ["TEMP_VIDEO_DIR_MAX_TOTAL_MB"] = "12" + import uuid + + def _make_dummy_video(name: str, size_mb: int): + path = TEMP_VIDEO_DIR / name + path.write_bytes(b"\x00" * (size_mb * 1024 * 1024)) + return path + + created = [] + try: + TEMP_VIDEO_DIR.mkdir(parents=True, exist_ok=True) + + now = int(time.time()) + exp = now + 3600 + for i in range(8): + uid = uuid.uuid4().hex + name = f"{uid}_{exp}.mp4" + p = _make_dummy_video(name, 2) + created.append(p) + + assert len(created) == 8, "Should have 8 files created (16MB total)" + existing_before = sum(1 for f in TEMP_VIDEO_DIR.iterdir() + if f.is_file() and f.name.endswith('.mp4')) + assert existing_before >= 8 + + _cleanup_expired_temp_videos(now) + + remaining = [p for p in created if p.exists()] + assert len(remaining) < 8, ( + f"Some files should be deleted when total > 12MB, " + f"but all {len(remaining)} files still exist" + ) + + finally: + del os.environ["TEMP_VIDEO_DIR_MAX_TOTAL_MB"] + for p in created: + try: + p.unlink(missing_ok=True) + except Exception: + pass + + +def test_cors_allow_methods_restricted(): + """CORS should only allow GET, POST, DELETE (not *).""" + main_path = os.path.join( + os.path.dirname(__file__), "..", "app", "main.py" + ) + with open(main_path, "r", encoding="utf-8") as f: + content = f.read() + assert 'allow_methods=["*"]' not in content, ( + "CORS allow_methods should not be wildcard '*' after security fix" + ) + assert 'allow_headers=["*"]' not in content, ( + "CORS allow_headers should not be wildcard '*' after security fix" + )