feat: add feishu multi-user auth

api/.env.example

@@ -8,6 +8,16 @@ WEB_AUTH_PASSWORD=
 WEB_AUTH_SESSION_SECRET=
 WEB_AUTH_COOKIE_NAME=skg_marketing_session
 WEB_AUTH_COOKIE_SECURE=false
+AUTH_DATA_ISOLATION_ENABLED=true
+
+# 飞书免登录（OAuth）。生产回调地址需同步配置到飞书开放平台应用安全设置。
+FEISHU_APP_ID=
+FEISHU_APP_SECRET=
+FEISHU_REDIRECT_URI=https://marketing.skg.com/api/auth/feishu/callback
+FEISHU_OAUTH_SCOPE=
+FEISHU_ALLOWED_EMAIL_DOMAINS=
+FEISHU_ALLOWED_EMAILS=
+FEISHU_ALLOWED_TENANT_KEYS=
 
 # 模型分工
 ASR_MODEL=whisper-1

api/main.py

@@ -4,6 +4,7 @@ import asyncio
 import base64
 import hashlib
 import hmac
+import io
 import json
 import os
 import random
@@ -17,12 +18,13 @@ import uuid
 from contextlib import asynccontextmanager
 from pathlib import Path
 from typing import Literal
+from urllib.parse import urlencode
 
 import httpx
 from dotenv import load_dotenv
 from fastapi import BackgroundTasks, FastAPI, File, Form, HTTPException, Request, Response, UploadFile
 from fastapi.middleware.cors import CORSMiddleware
-from fastapi.responses import FileResponse
+from fastapi.responses import FileResponse, JSONResponse, RedirectResponse
 from pydantic import BaseModel, ConfigDict, Field
 
 load_dotenv()
@@ -194,7 +196,30 @@ 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)
+FEISHU_APP_ID = (os.getenv("FEISHU_APP_ID") or os.getenv("FEISHU_CLIENT_ID") or "").strip()
+FEISHU_APP_SECRET = (os.getenv("FEISHU_APP_SECRET") or os.getenv("FEISHU_CLIENT_SECRET") or "").strip()
+FEISHU_REDIRECT_URI = os.getenv("FEISHU_REDIRECT_URI", "").strip()
+FEISHU_OAUTH_SCOPE = os.getenv("FEISHU_OAUTH_SCOPE", "").strip()
+FEISHU_AUTHORIZE_URL = os.getenv(
+    "FEISHU_AUTHORIZE_URL",
+    "https://accounts.feishu.cn/open-apis/authen/v1/authorize",
+).strip()
+FEISHU_TOKEN_URL = os.getenv(
+    "FEISHU_TOKEN_URL",
+    "https://open.feishu.cn/open-apis/authen/v2/oauth/token",
+).strip()
+FEISHU_USER_INFO_URL = os.getenv(
+    "FEISHU_USER_INFO_URL",
+    "https://open.feishu.cn/open-apis/authen/v1/user_info",
+).strip()
+FEISHU_STATE_COOKIE_NAME = os.getenv("FEISHU_STATE_COOKIE_NAME", "skg_feishu_oauth_state").strip() or "skg_feishu_oauth_state"
+FEISHU_ALLOWED_EMAIL_DOMAINS = os.getenv("FEISHU_ALLOWED_EMAIL_DOMAINS", "").strip()
+FEISHU_ALLOWED_EMAILS = os.getenv("FEISHU_ALLOWED_EMAILS", "").strip()
+FEISHU_ALLOWED_TENANT_KEYS = os.getenv("FEISHU_ALLOWED_TENANT_KEYS", "").strip()
+AUTH_DATA_ISOLATION_ENABLED = os.getenv("AUTH_DATA_ISOLATION_ENABLED", "true").strip().lower() not in {"0", "false", "no", "off"}
+PASSWORD_AUTH_CONFIGURED = bool(WEB_AUTH_USERNAME and WEB_AUTH_PASSWORD and WEB_AUTH_SESSION_SECRET)
+FEISHU_AUTH_CONFIGURED = bool(FEISHU_APP_ID and FEISHU_APP_SECRET and WEB_AUTH_SESSION_SECRET)
+WEB_AUTH_CONFIGURED = bool(PASSWORD_AUTH_CONFIGURED or FEISHU_AUTH_CONFIGURED)
 
 
 def default_video_gateway_paths(base_url: str) -> tuple[str, str, str]:
@@ -794,6 +819,11 @@ class SubjectAgentState(BaseModel):
 class Job(BaseModel):
     id: str
     url: str
+    owner_id: str = ""
+    owner_name: str = ""
+    owner_email: str = ""
+    owner_provider: str = ""
+    tenant_key: str = ""
     status: JobStatus = "created"
     progress: int = 0
     message: str = ""
@@ -827,7 +857,17 @@ 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 未配置")
+        raise HTTPException(503, "WEB_AUTH_SESSION_SECRET 以及账号密码或飞书 OAuth 未配置")
+
+
+def ensure_password_auth_configured() -> None:
+    if not PASSWORD_AUTH_CONFIGURED:
+        raise HTTPException(503, "账号密码登录未配置")
+
+
+def ensure_feishu_auth_configured() -> None:
+    if not FEISHU_AUTH_CONFIGURED:
+        raise HTTPException(503, "飞书免登录未配置")
 
 
 def _auth_signature(body: str) -> str:
@@ -846,16 +886,80 @@ def _decode_auth_body(body: str) -> dict:
     return data if isinstance(data, dict) else {}
 
 
-def make_auth_token(username: str, ttl_seconds: int) -> str:
-    body = _encode_auth_body({
-        "u": username,
+def _csv_values(raw: str) -> set[str]:
+    return {item.strip().lower() for item in raw.split(",") if item.strip()}
+
+
+def _normalize_next_url(value: str | None) -> str:
+    value = (value or "/").strip() or "/"
+    if not value.startswith("/") or value.startswith("//"):
+        return "/"
+    return value
+
+
+def _public_base_url(request: Request) -> str:
+    proto = request.headers.get("x-forwarded-proto") or request.url.scheme
+    host = request.headers.get("host") or request.url.netloc
+    return f"{proto}://{host}".rstrip("/")
+
+
+def _feishu_redirect_uri(request: Request) -> str:
+    if FEISHU_REDIRECT_URI:
+        return FEISHU_REDIRECT_URI
+    return f"{_public_base_url(request)}/api/auth/feishu/callback"
+
+
+def _session_user_id(payload: dict | None) -> str:
+    payload = payload or {}
+    explicit = str(payload.get("uid") or "").strip()
+    if explicit:
+        return explicit
+    provider = str(payload.get("provider") or "").strip().lower()
+    if provider == "feishu":
+        for key in ("open_id", "union_id", "email", "u"):
+            value = str(payload.get(key) or "").strip()
+            if value:
+                return f"feishu:{value.lower() if key == 'email' else value}"
+    username = str(payload.get("u") or "").strip() or "anonymous"
+    return f"password:{username}"
+
+
+def _public_session(payload: dict) -> dict:
+    return {
+        "uid": _session_user_id(payload),
+        "provider": str(payload.get("provider") or "password"),
+        "username": str(payload.get("u") or payload.get("name") or ""),
+        "name": str(payload.get("name") or payload.get("u") or ""),
+        "email": str(payload.get("email") or ""),
+        "open_id": str(payload.get("open_id") or ""),
+        "union_id": str(payload.get("union_id") or ""),
+        "tenant_key": str(payload.get("tenant_key") or ""),
+        "avatar_url": str(payload.get("avatar_url") or ""),
+    }
+
+
+def make_auth_token(user: str | dict, ttl_seconds: int) -> str:
+    if isinstance(user, str):
+        payload = {
+            "u": user,
+            "name": user,
+            "provider": "password",
+            "uid": f"password:{user}",
+        }
+    else:
+        payload = dict(user)
+        payload["uid"] = _session_user_id(payload)
+        payload.setdefault("u", payload.get("name") or payload.get("email") or payload["uid"])
+        payload.setdefault("name", payload.get("u") or payload["uid"])
+    payload.update({
         "exp": int(time.time()) + ttl_seconds,
         "n": secrets.token_hex(8),
     })
+    body = _encode_auth_body(payload)
     return f"{body}.{_auth_signature(body)}"
 
 
-def verify_auth_token(token: str) -> str | None:
+def verify_auth_token(token: str) -> dict | None:
     if not WEB_AUTH_CONFIGURED or "." not in token:
         return None
     body, supplied_sig = token.rsplit(".", 1)
@@ -867,14 +971,228 @@ def verify_auth_token(token: str) -> str | None:
         expires_at = int(payload.get("exp") or 0)
     except Exception:
         return None
-    if username != WEB_AUTH_USERNAME or expires_at < int(time.time()):
+    if expires_at < int(time.time()):
         return None
-    return username
+
+    provider = str(payload.get("provider") or "").strip().lower()
+    if not provider:
+        provider = "password" if username else ""
+
+    if provider == "password":
+        if not PASSWORD_AUTH_CONFIGURED or username != WEB_AUTH_USERNAME:
+            return None
+        payload["provider"] = "password"
+        payload["uid"] = f"password:{username}"
+        payload.setdefault("name", username)
+        return _public_session(payload)
+
+    if provider == "feishu":
+        if not FEISHU_AUTH_CONFIGURED:
+            return None
+        payload["provider"] = "feishu"
+        payload["uid"] = _session_user_id(payload)
+        return _public_session(payload)
+
+    return None
+
+
+def auth_session_from_request(request: Request) -> dict | None:
+    token = request.cookies.get(WEB_AUTH_COOKIE_NAME, "")
+    return verify_auth_token(token)
 
 
 def auth_username_from_request(request: Request) -> str | None:
-    token = request.cookies.get(WEB_AUTH_COOKIE_NAME, "")
-    return verify_auth_token(token)
+    session = auth_session_from_request(request)
+    return str(session.get("username") or session.get("name") or session.get("uid")) if session else None
+
+
+def data_user_from_request(request: Request) -> dict:
+    session = auth_session_from_request(request)
+    if session:
+        return session
+    if not WEB_AUTH_CONFIGURED:
+        return {"uid": "local:dev", "provider": "local", "username": "local-dev", "name": "local-dev", "email": "", "tenant_key": ""}
+    raise HTTPException(401, "unauthorized")
+
+
+def _is_password_session(user: dict | None) -> bool:
+    return bool(user and str(user.get("provider") or "") == "password")
+
+
+def assign_owner(model: Job | "AgentRun", user: dict) -> None:
+    model.owner_id = _session_user_id(user)
+    model.owner_name = str(user.get("name") or user.get("username") or model.owner_id)
+    model.owner_email = str(user.get("email") or "")
+    model.owner_provider = str(user.get("provider") or "")
+    model.tenant_key = str(user.get("tenant_key") or "")
+
+
+def user_can_access_job(job: Job | None, user: dict | None) -> bool:
+    if not job:
+        return False
+    if not AUTH_DATA_ISOLATION_ENABLED or not WEB_AUTH_CONFIGURED:
+        return True
+    owner_id = str(getattr(job, "owner_id", "") or "").strip()
+    if owner_id:
+        return bool(user and owner_id == _session_user_id(user))
+    return _is_password_session(user)
+
+
+def _load_agent_run_for_access(run_id: str):
+    run = AGENT_RUNS.get(run_id)
+    if not run and agent_run_path(run_id).exists():
+        try:
+            run = AgentRun.model_validate_json(agent_run_path(run_id).read_text(encoding="utf-8"))
+            AGENT_RUNS[run_id] = run
+        except Exception:
+            return None
+    return run
+
+
+def user_can_access_agent_run(run_id: str, user: dict | None) -> bool:
+    if not AUTH_DATA_ISOLATION_ENABLED or not WEB_AUTH_CONFIGURED:
+        return True
+    run = _load_agent_run_for_access(run_id)
+    if not run:
+        return False
+    owner_id = str(getattr(run, "owner_id", "") or "").strip()
+    if owner_id:
+        return bool(user and owner_id == _session_user_id(user))
+    return user_can_access_job(JOBS.get(run.job_id), user) or _is_password_session(user)
+
+
+JOB_PATH_RE = re.compile(r"^/jobs/([0-9a-f]{8,32})(?:/|$)")
+COPY_TO_JOB_PATH_RE = re.compile(r"^/asset-library/[^/]+/[^/]+/copy-to-job/([0-9a-f]{8,32})(?:/|$)")
+AGENT_RUN_PATH_RE = re.compile(r"^/agent-runs/([0-9a-f]{8,32})(?:/|$)")
+
+
+def _extract_protected_job_id(path: str) -> str:
+    for pattern in (JOB_PATH_RE, COPY_TO_JOB_PATH_RE):
+        match = pattern.match(path)
+        if match:
+            return match.group(1)
+    return ""
+
+
+def _feishu_oauth_state(next_url: str) -> str:
+    body = _encode_auth_body({
+        "kind": "feishu_oauth_state",
+        "next": _normalize_next_url(next_url),
+        "exp": int(time.time()) + 600,
+        "n": secrets.token_hex(12),
+    })
+    return f"{body}.{_auth_signature(body)}"
+
+
+def _verify_feishu_oauth_state(token: str) -> dict | None:
+    if not token 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)
+    except Exception:
+        return None
+    if payload.get("kind") != "feishu_oauth_state" or int(payload.get("exp") or 0) < int(time.time()):
+        return None
+    payload["next"] = _normalize_next_url(str(payload.get("next") or "/"))
+    return payload
+
+
+def _feishu_authorize_url(request: Request, state: str) -> str:
+    params = {
+        "client_id": FEISHU_APP_ID,
+        "redirect_uri": _feishu_redirect_uri(request),
+        "response_type": "code",
+        "state": state,
+    }
+    if FEISHU_OAUTH_SCOPE:
+        params["scope"] = FEISHU_OAUTH_SCOPE
+    return f"{FEISHU_AUTHORIZE_URL}?{urlencode(params)}"
+
+
+def _exchange_feishu_code(code: str, redirect_uri: str) -> str:
+    payload = {
+        "grant_type": "authorization_code",
+        "client_id": FEISHU_APP_ID,
+        "client_secret": FEISHU_APP_SECRET,
+        "code": code,
+        "redirect_uri": redirect_uri,
+    }
+    with httpx.Client(timeout=20) as client:
+        response = client.post(FEISHU_TOKEN_URL, json=payload)
+        response.raise_for_status()
+    data = response.json()
+    if data.get("code") not in (None, 0, "0"):
+        raise HTTPException(401, f"飞书授权失败：{data.get('msg') or data.get('message') or data.get('code')}")
+    token_data = data.get("data") if isinstance(data.get("data"), dict) else data
+    token = str(
+        token_data.get("access_token")
+        or token_data.get("user_access_token")
+        or token_data.get("accessToken")
+        or ""
+    ).strip()
+    if not token:
+        raise HTTPException(401, "飞书授权未返回 user_access_token")
+    return token
+
+
+def _fetch_feishu_user(access_token: str) -> dict:
+    with httpx.Client(timeout=20) as client:
+        response = client.get(FEISHU_USER_INFO_URL, headers={"Authorization": f"Bearer {access_token}"})
+        response.raise_for_status()
+    data = response.json()
+    if data.get("code") not in (None, 0, "0"):
+        raise HTTPException(401, f"飞书用户信息获取失败：{data.get('msg') or data.get('message') or data.get('code')}")
+    user = data.get("data") if isinstance(data.get("data"), dict) else data
+    if not isinstance(user, dict):
+        raise HTTPException(401, "飞书用户信息格式异常")
+    return user
+
+
+def _build_feishu_session(user: dict) -> dict:
+    email = str(user.get("email") or user.get("enterprise_email") or "").strip().lower()
+    open_id = str(user.get("open_id") or "").strip()
+    union_id = str(user.get("union_id") or "").strip()
+    tenant_key = str(user.get("tenant_key") or "").strip()
+    name = str(user.get("name") or user.get("en_name") or user.get("nickname") or email or open_id or union_id or "Feishu User").strip()
+    avatar_url = str(
+        user.get("avatar_url")
+        or user.get("avatar_thumb")
+        or user.get("avatar_middle")
+        or user.get("avatar_big")
+        or ""
+    ).strip()
+    session = {
+        "provider": "feishu",
+        "u": name,
+        "name": name,
+        "email": email,
+        "open_id": open_id,
+        "union_id": union_id,
+        "tenant_key": tenant_key,
+        "avatar_url": avatar_url,
+    }
+    session["uid"] = _session_user_id(session)
+    return session
+
+
+def _validate_feishu_session(session: dict) -> None:
+    allowed_emails = _csv_values(FEISHU_ALLOWED_EMAILS)
+    allowed_domains = {item.lstrip("@") for item in _csv_values(FEISHU_ALLOWED_EMAIL_DOMAINS)}
+    allowed_tenants = _csv_values(FEISHU_ALLOWED_TENANT_KEYS)
+
+    email = str(session.get("email") or "").lower()
+    domain = email.rsplit("@", 1)[1] if "@" in email else ""
+    tenant_key = str(session.get("tenant_key") or "").lower()
+
+    if allowed_emails and email not in allowed_emails:
+        raise HTTPException(403, "当前飞书账号不在允许登录名单")
+    if allowed_domains and domain not in allowed_domains:
+        raise HTTPException(403, "当前飞书账号邮箱域不允许登录")
+    if allowed_tenants and tenant_key not in allowed_tenants:
+        raise HTTPException(403, "当前飞书租户不允许登录")
 
 
 def job_dir(job_id: str) -> Path:
@@ -1474,7 +1792,7 @@ async def lifespan(_: FastAPI):
     yield
 
 
-app = FastAPI(title="SKG TK 二创 API", lifespan=lifespan)
+app = FastAPI(title="SKG 营销内容工作台 API", lifespan=lifespan)
 app.add_middleware(
     CORSMiddleware,
     allow_origins=CORS_ORIGINS,
@@ -1484,17 +1802,56 @@ app.add_middleware(
 )
 
 
+@app.middleware("http")
+async def enforce_data_isolation(request: Request, call_next):
+    path = request.url.path
+    if AUTH_DATA_ISOLATION_ENABLED and WEB_AUTH_CONFIGURED:
+        try:
+            user = data_user_from_request(request)
+        except HTTPException:
+            user = None
+
+        job_id = _extract_protected_job_id(path)
+        if job_id and not user_can_access_job(JOBS.get(job_id), user):
+            return JSONResponse({"detail": "job not found"}, status_code=404)
+
+        run_match = AGENT_RUN_PATH_RE.match(path)
+        if run_match and not user_can_access_agent_run(run_match.group(1), user):
+            return JSONResponse({"detail": "agent run not found"}, status_code=404)
+
+    return await call_next(request)
+
+
 @app.get("/auth/check")
 def auth_check(request: Request) -> Response:
     ensure_auth_configured()
-    if not auth_username_from_request(request):
+    if not auth_session_from_request(request):
         raise HTTPException(401, "unauthorized")
     return Response(status_code=204)
 
 
+@app.get("/auth/config")
+def auth_config() -> dict:
+    return {
+        "ok": True,
+        "auth_configured": WEB_AUTH_CONFIGURED,
+        "password_enabled": PASSWORD_AUTH_CONFIGURED,
+        "feishu_enabled": FEISHU_AUTH_CONFIGURED,
+        "data_isolation_enabled": AUTH_DATA_ISOLATION_ENABLED,
+    }
+
+
+@app.get("/auth/me")
+def auth_me(request: Request) -> dict:
+    session = auth_session_from_request(request)
+    if not session:
+        raise HTTPException(401, "unauthorized")
+    return {"ok": True, "user": session}
+
+
 @app.post("/auth/login")
 def auth_login(payload: AuthLoginPayload, response: Response) -> dict:
-    ensure_auth_configured()
+    ensure_password_auth_configured()
     username = payload.username.strip()
     password = payload.password
     valid_user = hmac.compare_digest(username, WEB_AUTH_USERNAME)
@@ -1515,6 +1872,66 @@ def auth_login(payload: AuthLoginPayload, response: Response) -> dict:
     return {"ok": True, "username": WEB_AUTH_USERNAME}
 
 
+@app.get("/auth/feishu/start")
+def auth_feishu_start(request: Request) -> RedirectResponse:
+    ensure_feishu_auth_configured()
+    next_url = _normalize_next_url(request.query_params.get("next"))
+    state = _feishu_oauth_state(next_url)
+    response = RedirectResponse(_feishu_authorize_url(request, state), status_code=302)
+    response.set_cookie(
+        key=FEISHU_STATE_COOKIE_NAME,
+        value=state,
+        max_age=600,
+        httponly=True,
+        secure=WEB_AUTH_COOKIE_SECURE,
+        samesite="lax",
+        path="/",
+    )
+    return response
+
+
+@app.get("/auth/feishu/callback")
+def auth_feishu_callback(request: Request) -> RedirectResponse:
+    ensure_feishu_auth_configured()
+    if request.query_params.get("error"):
+        raise HTTPException(401, f"飞书授权取消或失败：{request.query_params.get('error')}")
+
+    code = str(request.query_params.get("code") or "").strip()
+    supplied_state = str(request.query_params.get("state") or "").strip()
+    cookie_state = request.cookies.get(FEISHU_STATE_COOKIE_NAME, "")
+    if not code:
+        raise HTTPException(400, "missing feishu code")
+    if not supplied_state or not cookie_state or not hmac.compare_digest(supplied_state, cookie_state):
+        raise HTTPException(401, "invalid feishu state")
+
+    state_payload = _verify_feishu_oauth_state(supplied_state)
+    if not state_payload:
+        raise HTTPException(401, "expired feishu state")
+
+    access_token = _exchange_feishu_code(code, _feishu_redirect_uri(request))
+    session = _build_feishu_session(_fetch_feishu_user(access_token))
+    _validate_feishu_session(session)
+
+    ttl_seconds = 60 * 60 * 24 * 30
+    response = RedirectResponse(_normalize_next_url(str(state_payload.get("next") or "/")), status_code=302)
+    response.set_cookie(
+        key=WEB_AUTH_COOKIE_NAME,
+        value=make_auth_token(session, ttl_seconds),
+        max_age=ttl_seconds,
+        httponly=True,
+        secure=WEB_AUTH_COOKIE_SECURE,
+        samesite="lax",
+        path="/",
+    )
+    response.delete_cookie(
+        key=FEISHU_STATE_COOKIE_NAME,
+        path="/",
+        secure=WEB_AUTH_COOKIE_SECURE,
+        samesite="lax",
+    )
+    return response
+
+
 @app.post("/auth/logout")
 def auth_logout(response: Response) -> dict:
     response.delete_cookie(
@@ -4642,6 +5059,11 @@ def health() -> dict:
         "ok": True,
         "llm_configured": bool(LLM_API_KEY),
         "auth_configured": WEB_AUTH_CONFIGURED,
+        "auth_modes": {
+            "password": PASSWORD_AUTH_CONFIGURED,
+            "feishu": FEISHU_AUTH_CONFIGURED,
+            "data_isolation": AUTH_DATA_ISOLATION_ENABLED,
+        },
         "base_url": LLM_BASE_URL or "openai-default",
         "asr_base_url": ASR_BASE_URL or LLM_BASE_URL or "openai-default",
         "image_base_url": IMAGE_BASE_URL or LLM_BASE_URL or "openai-default",
@@ -4689,6 +5111,9 @@ def health() -> dict:
 class JobSummary(BaseModel):
     id: str
     url: str
+    owner_name: str = ""
+    owner_email: str = ""
+    owner_provider: str = ""
     status: JobStatus
     progress: int = 0
     message: str = ""
@@ -4704,16 +5129,22 @@ class JobSummary(BaseModel):
 
 
 @app.get("/jobs", response_model=list[JobSummary])
-def list_jobs(limit: int | None = None) -> list[JobSummary]:
-    """所有 job 的精简列表，按磁盘 state.json mtime 倒序（最新优先）。前端无 ?job= 时用它回填历史。"""
+def list_jobs(request: Request, limit: int | None = None) -> list[JobSummary]:
+    """当前用户可见 job 的精简列表，按磁盘 state.json mtime 倒序（最新优先）。"""
+    user = data_user_from_request(request)
     items: list[JobSummary] = []
     for job_id, job in JOBS.items():
+        if not user_can_access_job(job, user):
+            continue
         state_path = JOBS_DIR / job_id / "state.json"
         mtime = state_path.stat().st_mtime if state_path.exists() else 0.0
         thumb = f"/jobs/{job_id}/frames/{job.frames[0].index}.jpg" if job.frames else ""
         items.append(JobSummary(
             id=job.id,
             url=job.url,
+            owner_name=job.owner_name,
+            owner_email=job.owner_email,
+            owner_provider=job.owner_provider,
             status=job.status,
             progress=job.progress,
             message=job.message,
@@ -4734,11 +5165,13 @@ def list_jobs(limit: int | None = None) -> list[JobSummary]:
 
 
 @app.post("/jobs", response_model=Job)
-async def create_job(req: CreateJobReq, bg: BackgroundTasks) -> Job:
+async def create_job(req: CreateJobReq, bg: BackgroundTasks, request: Request) -> Job:
     if not req.url.strip():
         raise HTTPException(400, "url required")
+    user = data_user_from_request(request)
     job_id = uuid.uuid4().hex[:12]
     job = Job(id=job_id, url=req.url.strip())
+    assign_owner(job, user)
     JOBS[job_id] = job
     save_state(job)
     bg.add_task(pipeline_download, job_id)
@@ -4772,13 +5205,14 @@ async def retry_job_download(job_id: str, bg: BackgroundTasks) -> Job:
 
 
 @app.post("/jobs/upload", response_model=Job)
-async def create_job_from_upload(bg: BackgroundTasks, file: UploadFile = File(...)) -> Job:
+async def create_job_from_upload(bg: BackgroundTasks, request: Request, file: UploadFile = File(...)) -> Job:
     if not file.filename:
         raise HTTPException(400, "file required")
     ext = Path(file.filename).suffix.lower()
     if ext not in {".mp4", ".mov", ".webm", ".mkv", ".m4v"}:
         raise HTTPException(400, f"unsupported video format: {ext}")
 
+    user = data_user_from_request(request)
     job_id = uuid.uuid4().hex[:12]
     d = job_dir(job_id)
     mp4 = d / "source.mp4"
@@ -4789,6 +5223,7 @@ async def create_job_from_upload(bg: BackgroundTasks, file: UploadFile = File(..
         raise HTTPException(500, "upload failed")
 
     job = Job(id=job_id, url=f"upload://{file.filename}")
+    assign_owner(job, user)
     JOBS[job_id] = job
     save_state(job)
     bg.add_task(pipeline_download, job_id)
@@ -4815,7 +5250,8 @@ def _write_creative_reference_frame(job_id: str, file_bytes: bytes | None = None
 
 
 @app.post("/creative/jobs/image", response_model=Job)
-async def create_creative_image_job(file: UploadFile | None = File(default=None)) -> Job:
+async def create_creative_image_job(request: Request, file: UploadFile | None = File(default=None)) -> Job:
+    user = data_user_from_request(request)
     job_id = uuid.uuid4().hex[:12]
     file_bytes: bytes | None = None
     source_label = "blank"
@@ -4838,6 +5274,7 @@ async def create_creative_image_job(file: UploadFile | None = File(default=None)
         duration=0,
         frames=[frame],
     )
+    assign_owner(job, user)
     JOBS[job_id] = job
     save_state(job)
     return job
@@ -8224,6 +8661,11 @@ class AgentRunLog(BaseModel):
 class AgentRun(BaseModel):
     id: str
     job_id: str
+    owner_id: str = ""
+    owner_name: str = ""
+    owner_email: str = ""
+    owner_provider: str = ""
+    tenant_key: str = ""
     status: Literal["draft", "queued", "executing", "reviewing", "completed", "failed"] = "queued"
     stage: str = "queued"
     progress: int = 0
@@ -8564,14 +9006,17 @@ def agent_run_worker(run_id: str, product_refs: list[dict]) -> None:
 
 @app.post("/agent-runs", response_model=AgentRun)
 async def create_agent_run(
+    request: Request,
     tk_url: str = Form(...),
     product_files: list[UploadFile] | None = File(None),
 ) -> AgentRun:
     if not tk_url.strip():
         raise HTTPException(400, "tk_url required")
+    user = data_user_from_request(request)
     job_id = uuid.uuid4().hex[:12]
     run_id = uuid.uuid4().hex[:12]
     job = Job(id=job_id, url=tk_url.strip())
+    assign_owner(job, user)
     JOBS[job_id] = job
     save_state(job)
 
@@ -8580,6 +9025,7 @@ async def create_agent_run(
         refs.append(await save_agent_product_upload(job_id, upload, index))
 
     run = AgentRun(id=run_id, job_id=job_id, status="queued", stage="queued", progress=1)
+    assign_owner(run, user)
     save_agent_run(run)
     agent_log(run, f"任务已入队 · job={job_id} · 产品图 {len(refs)} 张", status="queued", stage="queued", progress=1)
     threading.Thread(target=agent_run_worker, args=(run_id, refs), daemon=True).start()
@@ -8587,14 +9033,15 @@ async def create_agent_run(
 
 
 @app.get("/agent-runs", response_model=list[AgentRun])
-def list_agent_runs(limit: int = 20) -> list[AgentRun]:
+def list_agent_runs(request: Request, limit: int = 20) -> list[AgentRun]:
+    user = data_user_from_request(request)
     for p in AGENT_RUNS_DIR.iterdir():
         if p.is_dir() and (p / "state.json").exists() and p.name not in AGENT_RUNS:
             try:
                 AGENT_RUNS[p.name] = AgentRun.model_validate_json((p / "state.json").read_text(encoding="utf-8"))
             except Exception:
                 pass
-    items = list(AGENT_RUNS.values())
+    items = [item for item in AGENT_RUNS.values() if user_can_access_agent_run(item.id, user)]
     items.sort(key=lambda item: item.updated_at, reverse=True)
     return items[:max(1, min(100, limit))]