fix(security): 路径穿越防护+导出行数上限+并发同步锁(3 critical/important)

- media.py: GET /api/media/{hash} 现强制 file 必须位于 Settings.media_dir
  沙箱内(resolve + relative_to),杜绝 local_path 越权读取任意文件。
- export.py: /api/export/chat-html 加 EXPORT_MAX_ROWS=5000 硬上限(.limit()),
  超出时仅渲染前 5000 条并插入 .export-warning 横幅,防止 OOM。
- scheduler.py: _safe_run 增加 SyncRun advisory lock(60min TTL),发现同
  entity 存在 status='running' 且在 TTL 内的记录即返回 skipped SyncResult;
  同时为本轮 stake 一行 wrapping SyncRun,让并发 CLI/scheduler/手动触发
  能相互避让。崩溃场景由 TTL 自动恢复。
- tests: 新增 path-traversal 拒绝/export 截断警告/advisory lock skip+TTL
  恢复 5 个测试;test_api_media 的 fixture 改为 MEDIA_DIR=tmp_path 以适配
  新沙箱语义。

Co-Authored-By: Claude <noreply@anthropic.com>

yunke-server/app/api/routers/export.py

@@ -21,6 +21,13 @@ router = APIRouter(prefix="/api/export", tags=["export"])
 
 SessionDep = Annotated[AsyncSession, Depends(get_session)]
 
+# Hard cap on rows an HTML export will load + render. An account-wide export
+# with no date range can match millions of messages; rendering all of them in
+# memory risks OOM. The cap is a safety valve, not a tight budget — at 5k rows
+# the HTML is already unwieldy in any browser. When hit, the response carries
+# a visible warning banner so the operator knows to narrow the date range.
+EXPORT_MAX_ROWS = 5000
+
 
 def _ts_text(ts: int | None) -> str:
     if not ts:
@@ -73,7 +80,14 @@ def _render_media(m) -> str:
     return ""
 
 
-def _render_chat_html(rows: list[Message], channel: str, owner: str, peer: str) -> str:
+def _render_chat_html(
+    rows: list[Message],
+    channel: str,
+    owner: str,
+    peer: str,
+    *,
+    warning: str = "",
+) -> str:
     bubbles = []
     for m in rows:
         side = "right" if m.is_send else "left"
@@ -90,6 +104,7 @@ def _render_chat_html(rows: list[Message], channel: str, owner: str, peer: str)
 <title>{title}</title><style>
 body{{font-family:-apple-system,Segoe UI,sans-serif;background:#f5f5f5;margin:0;padding:16px}}
 h1{{font-size:16px;margin:0 0 12px}}
+.export-warning{{background:#fff3cd;border:1px solid #ffeeba;color:#856404;padding:8px 12px;border-radius:4px;margin:0 0 12px;font-size:13px}}
 .bubble{{max-width:70%;margin:6px 0;padding:8px 12px;border-radius:8px;clear:both}}
 .bubble.right{{float:right;background:#95ec69;text-align:right}}
 .bubble.left{{float:left;background:#fff}}
@@ -98,7 +113,7 @@ h1{{font-size:16px;margin:0 0 12px}}
 .bubble .media{{margin-top:4px;font-size:13px}}
 .bubble .media audio,.bubble .media video,.bubble .media img{{display:block;max-width:100%}}
 .bubble .media .dur{{font-size:12px;color:#666;margin-left:4px}}
-</style></head><body><h1>{title}</h1>
+</style></head><body>{warning}<h1>{title}</h1>
 {''.join(bubbles) or '<p>暂无消息</p>'}
 </body></html>"""
 
@@ -132,9 +147,21 @@ async def export_chat_html(
         q = q.where(Message.timestamp >= start)
     if end is not None:
         q = q.where(Message.timestamp <= end)
-    q = q.order_by(sa.nullslast(Message.timestamp.asc()), Message.msg_id.asc())
+    q = q.order_by(sa.nullslast(Message.timestamp.asc()), Message.msg_id.asc()).limit(
+        EXPORT_MAX_ROWS
+    )
     rows = (await session.execute(q)).scalars().all()
-    html = _render_chat_html(list(rows), channel, owner_account_id, peer)
+    # If we filled the cap there may be more rows beyond it — surface a visible
+    # warning so the operator knows the export was truncated and can re-run
+    # with a tighter date range.
+    capped = len(rows) >= EXPORT_MAX_ROWS
+    warning = (
+        f'<div class="export-warning">⚠️ 结果超过 {EXPORT_MAX_ROWS} 条，'
+        f'仅导出前 {EXPORT_MAX_ROWS} 条。请缩小日期范围。</div>'
+        if capped
+        else ""
+    )
+    html = _render_chat_html(list(rows), channel, owner_account_id, peer, warning=warning)
     safe = peer or "all"
     filename = f"chat-{channel}-{owner_account_id}-{safe}.html".replace("@", "_").replace("/", "_")
     return Response(

yunke-server/app/api/routers/media.py

@@ -22,6 +22,7 @@ from fastapi.responses import FileResponse
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.api.deps import get_session
+from app.config import get_settings
 from app.db.models import Media
 from app.scraper.media import guess_mime_type
 
@@ -41,7 +42,19 @@ async def get_media(media_hash: str, session: SessionDep) -> FileResponse:
     if row is None or not row.local_path:
         raise HTTPException(status_code=404, detail="media not found")
 
-    path = Path(row.local_path)
+    # Path-traversal defense: resolve absolute and confirm the file lives under
+    # the configured ``media_dir`` sandbox. A row whose ``local_path`` escapes
+    # media_dir (an absolute path elsewhere, or a ``..``-laced value) is treated
+    # as not found — never served. Defense-in-depth: even if a future bug or a
+    # hostile scrape payload ever lets ``local_path`` point outside the sandbox,
+    # this check keeps the file server constrained to media_dir.
+    media_dir = get_settings().media_dir.resolve()
+    path = Path(row.local_path).resolve()
+    try:
+        path.relative_to(media_dir)
+    except ValueError:
+        raise HTTPException(status_code=404, detail="media not found")
+
     if not path.is_file():
         # Row exists but the blob was removed out-of-band — treat as not-found.
         raise HTTPException(status_code=404, detail="media file missing")

yunke-server/app/sync/scheduler.py

@@ -50,6 +50,7 @@ from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_asyn
 from app.config import get_settings
 from app.db import models  # noqa: F401  (register tables on Base.metadata)
 from app.db.base import Base
+from app.db.models import SyncRun
 from app.session.yunke_client import YunkeClient
 from app.storage.object_store import ObjectStore
 from app.sync.orchestrator import (
@@ -83,6 +84,14 @@ __all__ = ["SyncScheduler", "ENTITY_FUNCS"]
 # request, so 200 conversations × N pages each can still take a while).
 DEFAULT_MAX_CONVERSATIONS_PER_RUN = 200
 
+# Advisory-lock TTL: a ``running`` SyncRun row older than this is considered
+# stale (the process probably crashed mid-run) and a new invocation may
+# proceed despite it. 60 min is generous — any entity fits inside — but
+# short enough that a crashed run auto-recovers within an hour. Tuned to
+# protect against concurrent CLI + scheduler invocations hammering 云控
+# with the same scope twice.
+ADVISORY_LOCK_TTL_MINUTES = 60
+
 
 # ---------------------------------------------------------------------------
 # Conversation discovery (drives the incremental message sync)
@@ -561,6 +570,15 @@ class SyncScheduler:
         This is the failure-policy wrapper that protects the scheduler loop:
         a raised exception becomes a logged error and an empty result list,
         and the scheduler keeps its next appointment.
+
+        Concurrency: an advisory lock refuses to run if another ``_safe_run``
+        for the same entity is still active. The lock is implemented by
+        staking a ``running`` SyncRun row keyed on the scheduler entity name
+        (e.g. ``"all"`` / ``"messages"`` / ``"friends"``) at the start and
+        finalizing it on completion. A concurrent invocation sees the row and
+        skips. The 60-min TTL (:data:`ADVISORY_LOCK_TTL_MINUTES`) auto-recovers
+        from a crashed process: a stale ``running`` row eventually falls
+        outside the window and a new run may proceed.
         """
         started_at = dt.datetime.now(dt.UTC)
         client: YunkeClient | None = None
@@ -568,8 +586,74 @@ class SyncScheduler:
         try:
             client = await _build_client_from_env()
             async with self._session_maker() as session:
+                # --- Advisory lock check ------------------------------
+                # ``SyncRun.started_at`` is a non-tz-aware DateTime; reading
+                # a row back yields a naive datetime, so the cutoff is built
+                # naive-UTC to keep the comparison type-consistent.
+                cutoff = (
+                    dt.datetime.now(dt.UTC).replace(tzinfo=None)
+                    - dt.timedelta(minutes=ADVISORY_LOCK_TTL_MINUTES)
+                )
+                existing = (
+                    await session.execute(
+                        sa.select(SyncRun)
+                        .where(SyncRun.entity_type == entity)
+                        .where(SyncRun.status == "running")
+                        .where(SyncRun.started_at > cutoff)
+                        .limit(1)
+                    )
+                ).scalar_one_or_none()
+                if existing is not None:
+                    logger.warning(
+                        "sync %s: skipped — another run is active"
+                        " (started %s)",
+                        entity,
+                        existing.started_at.isoformat(),
+                    )
+                    skipped = SyncResult(
+                        entity=entity,
+                        status="skipped",
+                        fetched=0,
+                        upserted=0,
+                        error=f"another sync is running for {entity}",
+                    )
+                    if log:
+                        _log_run(entity, [skipped], started_at=started_at)
+                    return [skipped]
+
+                # --- Stake our own 'running' row ----------------------
+                # A wrapping SyncRun at the scheduler-entity level (distinct
+                # from any sub-entity rows the orchestrator creates internally).
+                # On a caught exception the session rolls back and this row
+                # disappears with everything else — no phantom to recover
+                # from. On the success path the row is finalized + committed
+                # below. If the process hard-crashes between flush and commit,
+                # the row stays 'running' and blocks retries for up to TTL.
+                run = SyncRun(
+                    entity_type=entity,
+                    status="running",
+                    started_at=dt.datetime.now(dt.UTC),
+                )
+                session.add(run)
+                await session.flush()
+
                 runner = ENTITY_FUNCS[entity]
                 results = await runner(session, client, ObjectStore())
+
+                # --- Finalize the wrapping SyncRun --------------------
+                run.status = (
+                    "success"
+                    if all(r.status == "success" for r in results)
+                    else "failed"
+                )
+                run.records_fetched = sum(r.fetched for r in results)
+                run.records_upserted = sum(r.upserted for r in results)
+                run.finished_at = dt.datetime.now(dt.UTC)
+                if run.status == "failed":
+                    joined = "; ".join(
+                        filter(None, (r.error for r in results))
+                    )
+                    run.error = joined or None
                 await session.commit()
         except Exception as exc:  # noqa: BLE001 — scheduler MUST NOT crash
             logger.exception("sync %s failed", entity)

yunke-server/tests/test_api_export.py

@@ -6,8 +6,10 @@
 
 from __future__ import annotations
 
+import functools
 from sqlalchemy.ext.asyncio import AsyncSession
 
+from app.api.routers.export import EXPORT_MAX_ROWS
 from app.db.models import Message
 
 
@@ -39,6 +41,26 @@ async def _seed(session: AsyncSession) -> None:
     await session.commit()
 
 
+async def _seed_many(session: AsyncSession, n: int) -> None:
+    """Seed ``n`` DM messages so the row cap can be exercised."""
+    session.add_all([
+        Message(
+            msg_id=f"cap{i}",
+            channel="wechat",
+            owner_account_id="wx_owner",
+            talker_id="wx_friend",
+            room_id=None,
+            text=f"msg{i}",
+            timestamp=1000 + i,
+            is_send=i % 2 == 0,
+            type="text",
+            type_raw=1,
+        )
+        for i in range(n)
+    ])
+    await session.commit()
+
+
 async def test_export_dm_returns_html(make_client):
     client = await make_client(_seed)
     r = await client.get("/api/export/chat-html", params={
@@ -83,3 +105,35 @@ async def test_export_embeds_media(make_client):
     # 本地路径 → 占位提示,不嵌 <audio>
     assert "[本地媒体]" in body
     assert "/api/media/abc123" not in body
+
+
+async def test_export_caps_rows_and_emits_warning(make_client):
+    """When the cap is hit, only EXPORT_MAX_ROWS rows render + a warning shows."""
+    # Seed twice the cap so the truncation is unambiguous.
+    seed = functools.partial(_seed_many, n=EXPORT_MAX_ROWS * 2)
+    client = await make_client(seed)
+    r = await client.get("/api/export/chat-html", params={
+        "channel": "wechat", "owner_account_id": "wx_owner", "talker_id": "wx_friend"})
+    assert r.status_code == 200
+    body = r.text
+    # The warning DIV is rendered (not just the CSS class in <style>).
+    assert '<div class="export-warning">' in body
+    assert str(EXPORT_MAX_ROWS) in body
+    # The first EXPORT_MAX_ROWS seeded rows (timestamps 1000..1000+N-1) render;
+    # anything past the cap must NOT appear. msg_id pattern "cap{i}" lets us
+    # count rendered bubbles precisely.
+    last_kept = EXPORT_MAX_ROWS - 1
+    last_dropped = EXPORT_MAX_ROWS  # first row beyond the cap
+    assert f"msg{last_kept}" in body
+    assert f"msg{last_dropped}" not in body
+
+
+async def test_export_no_warning_below_cap(make_client):
+    """Sub-cap exports render no warning DIV."""
+    seed = functools.partial(_seed_many, n=10)
+    client = await make_client(seed)
+    r = await client.get("/api/export/chat-html", params={
+        "channel": "wechat", "owner_account_id": "wx_owner", "talker_id": "wx_friend"})
+    assert r.status_code == 200
+    # No warning DIV (the CSS class still lives in <style> — that's fine).
+    assert '<div class="export-warning">' not in r.text

yunke-server/tests/test_api_media.py

@@ -27,6 +27,7 @@ from sqlalchemy.pool import StaticPool
 
 from app.api import deps
 from app.api.main import app
+from app.config import get_settings
 from app.db.base import Base
 from app.db.models import Media
 
@@ -34,7 +35,7 @@ PNG_BYTES = b"\x89PNG\r\n\x1a\n fake image body"
 
 
 @pytest.fixture
-async def client(tmp_path: Path):
+async def client(tmp_path: Path, monkeypatch):
     """Yield an ``AsyncClient`` against ``app`` with a seeded in-memory DB.
 
     Seeds one ``Media`` row (``abc123``) pointing at a real PNG-bytes file
@@ -42,7 +43,15 @@ async def client(tmp_path: Path):
     deliberately missing on disk (to exercise the file-missing 404 path), plus
     a third (``nomime``) whose ``mime_type`` is NULL (to exercise the
     ext-derived content-type fallback).
+
+    ``MEDIA_DIR`` is pointed at ``tmp_path`` so the endpoint's path-traversal
+    sandbox check (file must live under ``Settings.media_dir``) admits the
+    seeded rows. The settings cache is cleared around the override so the new
+    value is picked up.
     """
+    monkeypatch.setenv("MEDIA_DIR", str(tmp_path))
+    get_settings.cache_clear()
+
     engine = create_async_engine(
         "sqlite+aiosqlite:///:memory:",
         poolclass=StaticPool,
@@ -104,6 +113,65 @@ async def client(tmp_path: Path):
     finally:
         app.dependency_overrides.clear()
         await engine.dispose()
+        get_settings.cache_clear()
+
+
+@pytest.fixture
+async def escape_client(tmp_path: Path, monkeypatch):
+    """Variant of :func:`client` that pins ``MEDIA_DIR`` to a sandbox subdir
+    while seeding a row whose ``local_path`` points OUTSIDE that sandbox.
+
+    Used to verify the path-traversal defense: an outside-sandbox file must be
+    refused with 404 even when it exists on disk and the row is otherwise valid.
+    """
+    sandbox = tmp_path / "media_dir"
+    sandbox.mkdir()
+    outside = tmp_path / "outside"
+    outside.mkdir()
+
+    monkeypatch.setenv("MEDIA_DIR", str(sandbox))
+    get_settings.cache_clear()
+
+    engine = create_async_engine(
+        "sqlite+aiosqlite:///:memory:",
+        poolclass=StaticPool,
+        connect_args={"check_same_thread": False},
+    )
+    async with engine.begin() as conn:
+        await conn.run_sync(Base.metadata.create_all)
+
+    escape_file = outside / "secret.png"
+    escape_file.write_bytes(b"TOPSECRET")
+
+    SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
+    async with SessionLocal() as session:
+        session.add(
+            Media(
+                media_hash="escape",
+                channel="wechat",
+                media_type="image",
+                original_url="https://wx.oss.yunkecn.com/x/secret.png",
+                local_path=str(escape_file),
+                size_bytes=len(b"TOPSECRET"),
+                mime_type="image/png",
+            )
+        )
+        await session.commit()
+
+    async def _override_session():
+        async with SessionLocal() as s:
+            yield s
+
+    app.dependency_overrides[deps.get_session] = _override_session
+    try:
+        async with AsyncClient(
+            transport=ASGITransport(app=app), base_url="http://test"
+        ) as ac:
+            yield ac
+    finally:
+        app.dependency_overrides.clear()
+        await engine.dispose()
+        get_settings.cache_clear()
 
 
 async def test_get_media_returns_bytes_with_content_type(client: AsyncClient):
@@ -145,3 +213,17 @@ async def test_get_media_falls_back_to_ext_when_mime_null(client: AsyncClient):
     assert r.status_code == 200
     assert r.headers["content-type"].startswith("image/webp")
     assert r.content == b"WEBPBYTES"
+
+
+async def test_get_media_refuses_path_outside_media_dir(escape_client: AsyncClient):
+    """A row whose local_path escapes media_dir is refused with 404 — even
+    though the file exists on disk and the row is otherwise valid.
+
+    Pins the path-traversal defense: a hostile or buggy ``local_path`` value
+    must never let the endpoint serve files outside the configured sandbox.
+    """
+    r = await escape_client.get("/api/media/escape")
+    assert r.status_code == 404
+    # The file must NOT be served — only the 404 JSON body is returned.
+    assert b"TOPSECRET" not in r.content
+    assert "media not found" in r.json()["detail"]

yunke-server/tests/test_sync_scheduler.py

@@ -274,6 +274,128 @@ async def test_run_once_swallows_client_build_failure(engine, session_maker, mon
     assert results == []
 
 
+# ---------------------------------------------------------------------------
+# Advisory lock — concurrent-invocation protection (Fix 3)
+# ---------------------------------------------------------------------------
+
+
+def _utcnow_naive() -> dt.datetime:
+    """Naive UTC 'now', matching how SQLAlchemy reads back ``SyncRun.started_at``.
+
+    The column is a non-tz-aware ``DateTime``; an aware ``datetime`` would
+    raise TypeError when compared against DB-read values inside ``_safe_run``.
+    """
+    return dt.datetime.now(dt.UTC).replace(tzinfo=None)
+
+
+async def _make_static_engine():
+    """In-memory engine with ``StaticPool`` so cross-session rows are visible.
+
+    The default ``:memory:`` pool gives each checkout a fresh DB; the advisory
+    lock test seeds a row in one session and needs ``_safe_run``'s own session
+    to see it, so the engine must pin a single connection.
+    """
+    from sqlalchemy.pool import StaticPool
+
+    engine = create_async_engine(
+        "sqlite+aiosqlite:///:memory:",
+        poolclass=StaticPool,
+        connect_args={"check_same_thread": False},
+    )
+    async with engine.begin() as conn:
+        await conn.run_sync(Base.metadata.create_all)
+    return engine
+
+
+async def test_run_once_skips_when_another_run_is_active(monkeypatch):
+    """A 'running' SyncRun for the same entity started < TTL ago → skip."""
+    from app.db.models import SyncRun
+
+    engine = await _make_static_engine()
+    session_maker = async_sessionmaker(engine, expire_on_commit=False)
+
+    # Seed a 'running' SyncRun for "friends" started just now.
+    async with session_maker() as s:
+        s.add(
+            SyncRun(
+                entity_type="friends",
+                status="running",
+                started_at=_utcnow_naive(),
+            )
+        )
+        await s.commit()
+
+    fake_client = MagicMock()
+    monkeypatch.setattr(
+        scheduler_module,
+        "_build_client_from_env",
+        AsyncMock(return_value=fake_client),
+    )
+
+    called = {"count": 0}
+
+    async def fake_runner(session, client, store):
+        called["count"] += 1
+        return [_mk_result("wechat_friend", "success", 1, 1)]
+
+    monkeypatch.setitem(ENTITY_FUNCS, "friends", fake_runner)
+
+    try:
+        sched = SyncScheduler(engine=engine, session_maker=session_maker)
+        results = await sched.run_once("friends")
+    finally:
+        await engine.dispose()
+
+    # Skipped: the runner must NOT have run.
+    assert called["count"] == 0
+    # The skip surfaces as a single 'skipped' SyncResult with a reason.
+    assert len(results) == 1
+    assert results[0].status == "skipped"
+    assert "another sync is running" in (results[0].error or "")
+
+
+async def test_run_once_proceeds_after_stale_run_expires(monkeypatch):
+    """A 'running' SyncRun older than the TTL does NOT block — auto-recover."""
+    from app.db.models import SyncRun
+
+    engine = await _make_static_engine()
+    session_maker = async_sessionmaker(engine, expire_on_commit=False)
+
+    # Seed a 'running' SyncRun for "friends" 2 hours old — past the 60-min TTL.
+    async with session_maker() as s:
+        s.add(
+            SyncRun(
+                entity_type="friends",
+                status="running",
+                started_at=_utcnow_naive() - dt.timedelta(hours=2),
+            )
+        )
+        await s.commit()
+
+    fake_client = MagicMock()
+    monkeypatch.setattr(
+        scheduler_module,
+        "_build_client_from_env",
+        AsyncMock(return_value=fake_client),
+    )
+
+    async def fake_runner(session, client, store):
+        return [_mk_result("wechat_friend", "success", 3, 3)]
+
+    monkeypatch.setitem(ENTITY_FUNCS, "friends", fake_runner)
+
+    try:
+        sched = SyncScheduler(engine=engine, session_maker=session_maker)
+        results = await sched.run_once("friends")
+    finally:
+        await engine.dispose()
+
+    # Not skipped: the runner ran and returned its success result.
+    assert len(results) == 1
+    assert results[0].status == "success"
+    assert results[0].fetched == 3
+
+
 # ---------------------------------------------------------------------------
 # SyncScheduler.start — job registration
 # ---------------------------------------------------------------------------