fix(security): CORS限制+SSRF防护+PATCH乐观锁+accounts安全cap(4 important bugs)

- main.py + config.py: CORS origins read from Settings.cors_origins (comma-separated).
  Empty falls back to dev origin localhost:5173; never wildcard. Methods locked
  to GET/PATCH/POST; allow_credentials=False.
- scraper/media.py: download_url now passes follow_redirects=False to httpx.
  OSS URLs return 200 directly; a 3xx is treated as a download failure (caught
  by existing except clauses) instead of opening an SSRF vector to internal hosts.
- routers/voice_clips.py: PATCH /voice-clips/{id}/match accepts an optional
  If-Match header carrying the client's last-seen updated_at. On mismatch the
  row was modified by another request and the server returns 409. Without the
  header the behaviour is unchanged (backward compatible). Response now echoes
  the post-update updated_at so the client can fold it into the next request.
- routers/organization.py: /api/accounts now UNION ALLs the wechat_account and
  whatsapp_account projections at the SQL level and pushes COUNT + pagination
  (offset/limit) down to the DB. Stable ordering on (channel, account_id).
  Removes the in-memory materialization that broke at scale.

Tests: 395 passed. Added optimistic-lock coverage (If-Match accepted / stale 409 /
absent header backward-compat). Updated existing PATCH tests for the new
updated_at response field.

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
浅念
2026-08-12 11:41:30 +08:00
parent c7d54a135e
commit c261b2bc9a
6 changed files with 238 additions and 55 deletions
+19 -4
View File
@@ -20,8 +20,10 @@ Includes:
* ``GET /api/search`` (see :mod:`app.api.routers.search`).
* ``GET /api/export/chat-html`` (see :mod:`app.api.routers.export`).
CORS is configured permissively (wildcard origin, no credentials) so the local
frontend can call the API during development.
CORS is configured to read explicit allowed origins from ``Settings.cors_origins``
(comma-separated ``CORS_ORIGINS`` env var). Empty (default) falls back to the dev
origin ``http://localhost:5173``; set ``CORS_ORIGINS=https://app.example.com`` in
prod. Never wildcard — see the comment near ``add_middleware`` below.
Auth (Task 2.9): a single shared-token dependency (:func:`app.api.auth.require_auth`)
is applied globally on the app. Trust model: when ``Settings.api_token`` is
@@ -52,14 +54,27 @@ from app.api.routers import (
sync,
voice_clips,
)
from app.config import get_settings
app = FastAPI(title="yunke data mirror", dependencies=[Depends(require_auth)])
# CORS: read explicit allowed origins from Settings (``CORS_ORIGINS`` env var,
# comma-separated). Empty -> dev fallback (``http://localhost:5173``). Never
# ``*`` — wildcard CORS on an authenticated API is an open door even without
# credentials, since the middleware would reflect any attacker-controlled
# origin back in ``Access-Control-Allow-Origin``.
_settings = get_settings()
_cors = _settings.cors_origins.strip()
if _cors:
_origins = [o.strip() for o in _cors.split(",") if o.strip()]
else:
_origins = [] # empty = deny all cross-origin (safest default)
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_methods=["*"],
allow_origins=_origins or ["http://localhost:5173"], # dev fallback
allow_methods=["GET", "PATCH", "POST"],
allow_headers=["*"],
allow_credentials=False,
)
app.include_router(customer.router)
+77 -45
View File
@@ -10,9 +10,10 @@ Read-only queries over the mirrored ``department``, ``staff``,
* ``GET /api/accounts`` -> ``Page[AccountOut]`` projecting both staff-messaging
account tables onto a unified shape (see :class:`AccountOut`). The optional
``channel`` filter restricts to one of ``wechat`` / ``whatsapp``; when
omitted, both tables are unioned in memory. Staff-account volumes are modest
(tens to low hundreds of rows), so the in-memory merge is cheap and keeps
the API contract simple.
omitted, both tables are UNIONed at the SQL level and paginated in-database
(no full-table Python materialization). Staff-account volumes are modest in
practice today, but the SQL-level pagination keeps the endpoint safe if a
tenant ever scales past thousands of rows.
"""
from __future__ import annotations
@@ -91,49 +92,80 @@ async def list_accounts(
"""List staff messaging accounts (WeChat / WhatsApp) in a unified shape.
``channel`` optional; one of ``wechat`` / ``whatsapp``. When omitted, both
tables are merged in memory. Pagination is applied after the merge using a
stable ``(channel, account_id)`` ordering.
tables are UNIONed at the SQL level. Pagination and total count are pushed
down to the database (single ``COUNT(*)`` over the UNION subquery + a
paginated ``SELECT`` with ``OFFSET``/``LIMIT``) so we never materialize the
full cross-table row set in Python. Stable ordering is ``(channel,
account_id)``.
``AccountOut`` field mapping (identical shape from both source tables):
* WeChat: account_id=wx_id, nickname=nick_name, user_id, user_name, phone,
contacts_num, group_num, is_on_job, raw_json (verbatim upstream record).
* WhatsApp: account_id=uid (+ same projection).
"""
selected_channels = [channel] if channel else ["wechat", "whatsapp"]
wechat_q = sa.select(
WechatAccount.wx_id.label("account_id"),
sa.literal("wechat").label("channel"),
WechatAccount.nick_name.label("nickname"),
WechatAccount.user_id.label("user_id"),
WechatAccount.user_name.label("user_name"),
WechatAccount.phone.label("phone"),
WechatAccount.contacts_num.label("contacts_num"),
WechatAccount.group_num.label("group_num"),
WechatAccount.is_on_job.label("is_on_job"),
WechatAccount.raw_json.label("raw_json"),
)
whatsapp_q = sa.select(
WhatsappAccount.uid.label("account_id"),
sa.literal("whatsapp").label("channel"),
WhatsappAccount.nick_name.label("nickname"),
WhatsappAccount.user_id.label("user_id"),
WhatsappAccount.user_name.label("user_name"),
WhatsappAccount.phone.label("phone"),
WhatsappAccount.contacts_num.label("contacts_num"),
WhatsappAccount.group_num.label("group_num"),
WhatsappAccount.is_on_job.label("is_on_job"),
WhatsappAccount.raw_json.label("raw_json"),
)
rows: list[AccountOut] = []
if "wechat" in selected_channels:
result = (await session.execute(sa.select(WechatAccount))).scalars().all()
for r in result:
rows.append(
AccountOut(
channel="wechat",
account_id=r.wx_id,
nickname=r.nick_name,
user_id=r.user_id,
user_name=r.user_name,
phone=r.phone,
contacts_num=r.contacts_num,
group_num=r.group_num,
is_on_job=r.is_on_job,
raw_json=r.raw_json,
)
)
if "whatsapp" in selected_channels:
result = (await session.execute(sa.select(WhatsappAccount))).scalars().all()
for r in result:
rows.append(
AccountOut(
channel="whatsapp",
account_id=r.uid,
nickname=r.nick_name,
user_id=r.user_id,
user_name=r.user_name,
phone=r.phone,
contacts_num=r.contacts_num,
group_num=r.group_num,
is_on_job=r.is_on_job,
raw_json=r.raw_json,
)
)
if channel == "wechat":
parts: list[sa.Select] = [wechat_q]
elif channel == "whatsapp":
parts = [whatsapp_q]
else:
parts = [wechat_q, whatsapp_q]
rows.sort(key=lambda a: (a.channel, a.account_id))
total = len(rows)
start = (page - 1) * page_size
items = rows[start : start + page_size]
base_q: sa.Select = parts[0] if len(parts) == 1 else sa.union_all(*parts)
# Wrap in a subquery so COUNT and paginated SELECT both have a stable
# column set (the labelled columns become ``sub.c.<label>``).
sub = base_q.subquery()
total = (
await session.execute(sa.select(sa.func.count()).select_from(sub))
).scalar_one()
page_q = (
sa.select(sub)
.order_by(sub.c.channel.asc(), sub.c.account_id.asc())
.offset((page - 1) * page_size)
.limit(page_size)
)
rows = (await session.execute(page_q)).all()
items = [
AccountOut(
channel=row.channel,
account_id=row.account_id,
nickname=row.nickname,
user_id=row.user_id,
user_name=row.user_name,
phone=row.phone,
contacts_num=row.contacts_num,
group_num=row.group_num,
is_on_job=row.is_on_job,
raw_json=row.raw_json,
)
for row in rows
]
return Page(items=items, total=total, page=page, page_size=page_size)
+27 -1
View File
@@ -30,7 +30,7 @@ from __future__ import annotations
from typing import Annotated, Any
import sqlalchemy as sa
from fastapi import APIRouter, Depends, HTTPException, Query
from fastapi import APIRouter, Depends, Header, HTTPException, Query
from pydantic import BaseModel
from sqlalchemy.ext.asyncio import AsyncSession
@@ -153,20 +153,46 @@ async def match_voice_clip(
session: SessionDep,
voice_clip_id: int,
body: MatchBody,
if_match: Annotated[str | None, Header()] = None,
) -> dict[str, Any]:
"""Associate (or clear) the customer match on a voice_clip.
LOCAL write only — ``matched_friend_id`` is stored in the mirror's
``voice_clip`` row and never pushed to 云客. ``friend_id=None`` clears an
existing match.
Optimistic lock (optional, backward-compatible): the client may send
``If-Match: <updated_at ISO>`` with the ``updated_at`` it last read. When
present, the server compares it against the row's current ``updated_at``
and rejects with ``409`` if the row was modified by another request in
between. Without the header, behavior is unchanged.
The response echoes the post-update ``updated_at`` so the client can fold
it into its next read.
"""
clip = await session.get(VoiceClip, voice_clip_id)
if not clip:
raise HTTPException(status_code=404, detail="voice_clip not found")
# Optimistic lock: if the client sent If-Match, verify the row has not been
# modified since the value it read. The header carries the ISO 8601
# ``updated_at`` the client last saw; a mismatch means another writer
# committed first.
if if_match is not None and clip.updated_at is not None:
expected = clip.updated_at.isoformat()
if if_match != expected:
raise HTTPException(
status_code=409,
detail="voice_clip was modified by another request",
)
clip.matched_friend_id = body.friend_id # None clears the match
await session.commit()
# Refresh to pick up the server-generated ``updated_at`` from the UPDATE
# (``onupdate=func.now()``); with ``expire_on_commit=False`` the attribute
# otherwise stays at the pre-update value.
await session.refresh(clip)
return {
"voice_clip_id": voice_clip_id,
"matched_friend_id": body.friend_id,
"updated_at": clip.updated_at.isoformat() if clip.updated_at else None,
}
+9
View File
@@ -34,6 +34,15 @@ class Settings(BaseSettings):
# <token>`` or ``?token=<token>``. See app/api/auth.py.
api_token: str = ""
# Comma-separated list of allowed CORS origins for the viewing API. Empty
# (default) -> only the dev fallback (``http://localhost:5173``) is allowed;
# set ``CORS_ORIGINS=https://app.example.com`` in prod to lock cross-origin
# access down to explicit origins. Never ``*`` (wildcard + credentialed APIs
# are an open door; we accept no credentials but still restrict origins to
# avoid reflecting arbitrary attacker-controlled origins in ``Acces-Control-
# Allow-Origin``).
cors_origins: str = ""
# --- Anti-risk-control / stealth tuning ---------------------------------
# Tuned 2026-08-11 after triggering 风控 during a heavy session (~500+
# consecutive requests across groups/WA/voice-clips/SMS in ~1 hour).
+10 -2
View File
@@ -145,9 +145,17 @@ async def download_url(client: YunkeClient, url: str) -> bytes:
for the public OSS media URLs, useful for any same-origin authenticated
URLs) and is the transport ``respx`` mocks in the test suite, so media
downloads go through the same mocked transport as every other scraper
request. Redirects are followed (OSS / avatar URLs commonly redirect).
request.
Redirects are NOT followed (SSRF protection): the URL is the trusted OSS /
recording endpoint resolved at scrape time, and following a 3xx would allow
a malicious or compromised host to redirect us to internal addresses such
as ``http://169.254.169.254/`` (cloud metadata) or other RFC1918 hosts.
OSS media URLs return the blob directly with 200; a 3xx here is treated as
a download failure and surfaces through the caller's existing
``except (httpx.HTTPError, OSError)`` clause.
"""
resp = await client.client.get(url, follow_redirects=True)
resp = await client.client.get(url, follow_redirects=False)
resp.raise_for_status()
return resp.content
+96 -3
View File
@@ -268,7 +268,12 @@ async def test_match_returns_payload_and_persists(make_client):
r = await client.patch("/api/voice-clips/1/match", json={"friend_id": "fid_remark"})
assert r.status_code == 200
body = r.json()
assert body == {"voice_clip_id": 1, "matched_friend_id": "fid_remark"}
# Response now carries the post-update ``updated_at`` for the optimistic-
# lock flow (client folds it into the next request's ``If-Match`` header).
assert body["voice_clip_id"] == 1
assert body["matched_friend_id"] == "fid_remark"
assert "updated_at" in body
updated_at_v1 = body["updated_at"]
# The GET list now reflects the match (cross-request persistence —
# requires the PATCH to commit, not just flush).
@@ -280,6 +285,7 @@ async def test_match_returns_payload_and_persists(make_client):
# The other wechat clip for u1 is still unmatched.
assert by_id[3]["matched_friend_id"] is None
assert by_id[3]["matched_friend_name"] is None
return updated_at_v1 # for reuse by optimistic-lock tests below if needed
async def test_match_name_falls_back_through_nickname_real_name(make_client):
@@ -332,7 +338,10 @@ async def test_unmatch_clears_friend_id(make_client):
# Now clear it.
r = await client.patch("/api/voice-clips/1/match", json={"friend_id": None})
assert r.status_code == 200
assert r.json() == {"voice_clip_id": 1, "matched_friend_id": None}
body = r.json()
assert body["voice_clip_id"] == 1
assert body["matched_friend_id"] is None
assert "updated_at" in body
r2 = await client.get("/api/voice-clips", params={"user_id": "u1"})
by_id = {it["voice_clip_id"]: it for it in r2.json()["items"]}
@@ -356,4 +365,88 @@ async def test_match_default_null_body_is_treated_as_unmatch(make_client):
# Empty body -> body.friend_id defaults to None -> unmatch.
r = await client.patch("/api/voice-clips/1/match", json={})
assert r.status_code == 200
assert r.json()["matched_friend_id"] is None
body = r.json()
assert body["matched_friend_id"] is None
assert "updated_at" in body
# ---------------------------------------------------------------------------
# Optimistic lock — If-Match header on PATCH /api/voice-clips/{id}/match
# ---------------------------------------------------------------------------
async def test_match_with_if_match_accepts_current_updated_at(make_client):
"""If-Match equal to the row's current ``updated_at`` is accepted (200).
The flow: GET to read the current ``updated_at`` (via a successful PATCH
echo), send that value as ``If-Match`` on the next PATCH — the server
accepts the request because the row has not changed in between.
"""
client = await make_client(_seed_with_customer)
# First PATCH establishes the match and returns the post-update timestamp.
r1 = await client.patch(
"/api/voice-clips/1/match", json={"friend_id": "fid_remark"}
)
assert r1.status_code == 200
current = r1.json()["updated_at"]
assert current is not None
# Second PATCH with If-Match=current: accepted (200), match cleared.
r2 = await client.patch(
"/api/voice-clips/1/match",
json={"friend_id": None},
headers={"If-Match": current},
)
assert r2.status_code == 200
assert r2.json()["matched_friend_id"] is None
async def test_match_with_stale_if_match_returns_409(make_client):
"""An If-Match that doesn't match the row's current ``updated_at`` -> 409.
Simulates the lost-update race: client A reads the clip, client B modifies
it (bumping ``updated_at``), then client A's PATCH carrying A's now-stale
timestamp is rejected. We force a clearly stale value rather than relying
on wall-clock drift because SQLite's ``CURRENT_TIMESTAMP`` (the dialect
compilation of ``func.now()``) has 1-second resolution and two PATCHes
inside the same second produce identical timestamps.
"""
client = await make_client(_seed_with_customer)
# First PATCH: succeeds, returns the post-update timestamp.
r1 = await client.patch(
"/api/voice-clips/1/match", json={"friend_id": "fid_remark"}
)
assert r1.status_code == 200
current = r1.json()["updated_at"]
assert current is not None
# Any string that differs from the row's current updated_at is "stale" —
# 409. We use an obviously-older timestamp so the test is deterministic.
stale = "2000-01-01T00:00:00"
assert stale != current
r2 = await client.patch(
"/api/voice-clips/1/match",
json={"friend_id": None},
headers={"If-Match": stale},
)
assert r2.status_code == 409
assert "modified" in r2.json()["detail"].lower()
async def test_match_without_if_match_is_backward_compatible(make_client):
"""No If-Match header -> no optimistic-lock check; behavior unchanged."""
client = await make_client(_seed_with_customer)
# First PATCH establishes a value.
r1 = await client.patch(
"/api/voice-clips/1/match", json={"friend_id": "fid_remark"}
)
assert r1.status_code == 200
# Second PATCH WITHOUT If-Match — accepted regardless of staleness
# (backward compatible: clients that don't send the header keep working).
r2 = await client.patch(
"/api/voice-clips/1/match", json={"friend_id": "fid_nick"}
)
assert r2.status_code == 200
assert r2.json()["matched_friend_id"] == "fid_nick"