auto-save 2026-05-15 15:15 (~4)

api/main.py

@@ -2,9 +2,12 @@ from __future__ import annotations
 
 import asyncio
 import base64
+import hashlib
+import hmac
 import json
 import os
 import random
+import secrets
 import shutil
 import subprocess
 import threading
@@ -16,7 +19,7 @@ from typing import Literal
 
 import httpx
 from dotenv import load_dotenv
-from fastapi import BackgroundTasks, FastAPI, File, HTTPException, UploadFile
+from fastapi import BackgroundTasks, FastAPI, File, HTTPException, Request, Response, UploadFile
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.responses import FileResponse
 from pydantic import BaseModel, Field
@@ -91,6 +94,12 @@ VIDEO_MODEL_ALIASES = {
 }
 VIDEO_API_BASE_URL = os.getenv("VIDEO_API_BASE_URL", "").strip()
 VIDEO_API_KEY = os.getenv("VIDEO_API_KEY", "").strip()
+WEB_AUTH_USERNAME = os.getenv("WEB_AUTH_USERNAME", "").strip()
+WEB_AUTH_PASSWORD = os.getenv("WEB_AUTH_PASSWORD", "").strip()
+WEB_AUTH_SESSION_SECRET = os.getenv("WEB_AUTH_SESSION_SECRET", "").strip()
+WEB_AUTH_COOKIE_NAME = os.getenv("WEB_AUTH_COOKIE_NAME", "skg_marketing_session").strip() or "skg_marketing_session"
+WEB_AUTH_COOKIE_SECURE = os.getenv("WEB_AUTH_COOKIE_SECURE", "true").strip().lower() not in {"0", "false", "no"}
+WEB_AUTH_CONFIGURED = bool(WEB_AUTH_USERNAME and WEB_AUTH_PASSWORD and WEB_AUTH_SESSION_SECRET)
 
 
 def default_video_gateway_paths(base_url: str) -> tuple[str, str, str]:
@@ -455,6 +464,12 @@ class Job(BaseModel):
     error: str = ""
 
 
+class AuthLoginPayload(BaseModel):
+    username: str
+    password: str
+    remember: bool = False
+
+
 JOBS: dict[str, Job] = {}
 ANALYZE_QUEUE: list[AnalyzeTask] = []
 ANALYZE_WORKER_RUNNING = False
@@ -462,6 +477,58 @@ AUDIO_WORKERS_RUNNING: set[str] = set()
 AUDIO_WORKERS_LOCK = threading.Lock()
 
 
+def ensure_auth_configured() -> None:
+    if not WEB_AUTH_CONFIGURED:
+        raise HTTPException(503, "WEB_AUTH_USERNAME、WEB_AUTH_PASSWORD 或 WEB_AUTH_SESSION_SECRET 未配置")
+
+
+def _auth_signature(body: str) -> str:
+    return hmac.new(WEB_AUTH_SESSION_SECRET.encode("utf-8"), body.encode("utf-8"), hashlib.sha256).hexdigest()
+
+
+def _encode_auth_body(payload: dict) -> str:
+    raw = json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
+    return base64.urlsafe_b64encode(raw).decode("ascii").rstrip("=")
+
+
+def _decode_auth_body(body: str) -> dict:
+    padded = body + "=" * (-len(body) % 4)
+    raw = base64.urlsafe_b64decode(padded.encode("ascii"))
+    data = json.loads(raw.decode("utf-8"))
+    return data if isinstance(data, dict) else {}
+
+
+def make_auth_token(username: str, ttl_seconds: int) -> str:
+    body = _encode_auth_body({
+        "u": username,
+        "exp": int(time.time()) + ttl_seconds,
+        "n": secrets.token_hex(8),
+    })
+    return f"{body}.{_auth_signature(body)}"
+
+
+def verify_auth_token(token: str) -> str | None:
+    if not WEB_AUTH_CONFIGURED or "." not in token:
+        return None
+    body, supplied_sig = token.rsplit(".", 1)
+    if not hmac.compare_digest(_auth_signature(body), supplied_sig):
+        return None
+    try:
+        payload = _decode_auth_body(body)
+        username = str(payload.get("u") or "")
+        expires_at = int(payload.get("exp") or 0)
+    except Exception:
+        return None
+    if username != WEB_AUTH_USERNAME or expires_at < int(time.time()):
+        return None
+    return username
+
+
+def auth_username_from_request(request: Request) -> str | None:
+    token = request.cookies.get(WEB_AUTH_COOKIE_NAME, "")
+    return verify_auth_token(token)
+
+
 def job_dir(job_id: str) -> Path:
     d = JOBS_DIR / job_id
     d.mkdir(parents=True, exist_ok=True)
@@ -724,6 +791,48 @@ app.add_middleware(
 )
 
 
+@app.get("/auth/check")
+def auth_check(request: Request) -> Response:
+    ensure_auth_configured()
+    if not auth_username_from_request(request):
+        raise HTTPException(401, "unauthorized")
+    return Response(status_code=204)
+
+
+@app.post("/auth/login")
+def auth_login(payload: AuthLoginPayload, response: Response) -> dict:
+    ensure_auth_configured()
+    username = payload.username.strip()
+    password = payload.password
+    valid_user = hmac.compare_digest(username, WEB_AUTH_USERNAME)
+    valid_password = hmac.compare_digest(password, WEB_AUTH_PASSWORD)
+    if not (valid_user and valid_password):
+        raise HTTPException(401, "用户名或密码不正确")
+
+    ttl_seconds = 60 * 60 * 24 * 30 if payload.remember else 60 * 60 * 12
+    response.set_cookie(
+        key=WEB_AUTH_COOKIE_NAME,
+        value=make_auth_token(WEB_AUTH_USERNAME, ttl_seconds),
+        max_age=ttl_seconds,
+        httponly=True,
+        secure=WEB_AUTH_COOKIE_SECURE,
+        samesite="lax",
+        path="/",
+    )
+    return {"ok": True, "username": WEB_AUTH_USERNAME}
+
+
+@app.post("/auth/logout")
+def auth_logout(response: Response) -> dict:
+    response.delete_cookie(
+        key=WEB_AUTH_COOKIE_NAME,
+        path="/",
+        secure=WEB_AUTH_COOKIE_SECURE,
+        samesite="lax",
+    )
+    return {"ok": True}
+
+
 # ---------- Pipeline 实现 ----------
 
 def run(cmd: list[str], cwd: Path | None = None) -> str:
@@ -2157,6 +2266,7 @@ def health() -> dict:
     return {
         "ok": True,
         "llm_configured": bool(LLM_API_KEY),
+        "auth_configured": WEB_AUTH_CONFIGURED,
         "base_url": LLM_BASE_URL or "openai-default",
         "models": {
             "asr": ASR_MODEL,