diff --git a/yunke-server/.env.example b/yunke-server/.env.example
index 0a28082..bc5e213 100644
--- a/yunke-server/.env.example
+++ b/yunke-server/.env.example
@@ -7,3 +7,7 @@ MEDIA_DIR=./media
 # Fernet key (e.g. from cryptography.fernet.Fernet.generate_key()) used to encrypt
 # stored credentials. Required for real runs; left blank for tests.
 ENCRYPTION_KEY=
+# Shared bearer token for the viewing API. Empty (default) -> auth DISABLED
+# (trust the network boundary). When set, API requests must present it via
+# `Authorization: Bearer <token>` or `?token=<token>`.
+API_TOKEN=
diff --git a/yunke-server/app/api/auth.py b/yunke-server/app/api/auth.py
new file mode 100644
index 0000000..9589c22
--- /dev/null
+++ b/yunke-server/app/api/auth.py
@@ -0,0 +1,74 @@
+"""Simple shared-token auth for the yunke data-mirror API (Task 2.9).
+
+Trust model (per design doc §8 + plan Task 2.9):
+
+* **No token configured** (``Settings.api_token`` empty, the default) -> auth
+  is DISABLED. This is a local/intranet tool; when no token is set we trust the
+  network boundary and every request passes through. This keeps local dev and
+  single-box deployments zero-config.
+* **Token configured** -> every request must present the same token via either
+  the ``Authorization: Bearer <token>`` header or the ``?token=<token>`` query
+  parameter. Missing / mismatched tokens get ``401 Unauthorized``.
+
+The dependency is applied globally on the FastAPI app (see
+:mod:`app.api.main`). ``GET /api/health`` is exempt so liveness probes can
+reach it without knowing the token — that exemption is encoded here in
+:data:`_PUBLIC_PATHS` so the rule lives with the auth logic itself.
+"""
+
+from __future__ import annotations
+
+import secrets
+
+from fastapi import HTTPException, Request, status
+
+from app.config import get_settings
+
+# Request paths that bypass auth even when a token is configured. Keep this
+# minimal so the protected surface is easy to reason about — currently just the
+# liveness probe.
+_PUBLIC_PATHS = frozenset({"/api/health"})
+
+
+async def require_auth(request: Request) -> None:
+    """No-op when ``api_token`` is unset; otherwise enforce a shared bearer token.
+
+    Settings are read on every call (``get_settings`` is ``lru_cache``-ed, so
+    this is cheap) so rotating the token — env change plus
+    ``get_settings.cache_clear()`` — takes effect without a process restart.
+    The comparison uses :func:`secrets.compare_digest` on UTF-8 bytes to avoid
+    a timing side channel and to tolerate non-ASCII token values.
+    """
+    settings = get_settings()
+    expected = settings.api_token
+    if not expected:
+        # No token configured -> trust the network boundary.
+        return
+
+    if request.url.path in _PUBLIC_PATHS:
+        return
+
+    presented = _extract_token(request)
+    if presented is None or not secrets.compare_digest(
+        presented.encode("utf-8"), expected.encode("utf-8")
+    ):
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="invalid or missing API token",
+            headers={"WWW-Authenticate": "Bearer"},
+        )
+
+
+def _extract_token(request: Request) -> str | None:
+    """Pull the presented token from the ``Authorization`` header or ``?token=``.
+
+    Header wins when present and well-formed (``Bearer <value>``); otherwise we
+    fall back to the ``token`` query parameter, which is convenient for
+    browser-visible endpoints like media streaming.
+    """
+    auth_header = request.headers.get("Authorization")
+    if auth_header:
+        scheme, _, value = auth_header.partition(" ")
+        if scheme.lower() == "bearer" and value:
+            return value
+    return request.query_params.get("token")
diff --git a/yunke-server/app/api/main.py b/yunke-server/app/api/main.py
index 26ad462..9a8f99c 100644
--- a/yunke-server/app/api/main.py
+++ b/yunke-server/app/api/main.py
@@ -10,17 +10,25 @@ Includes:
 * ``GET /api/media/{media_hash}`` (see :mod:`app.api.routers.media`).
 
 CORS is configured permissively (wildcard origin, no credentials) so the local
-frontend can call the API during development. Auth / finer CORS are LATER tasks.
+frontend can call the API during development.
+
+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
+empty (the default) auth is disabled — we trust the network boundary (this is a
+local/intranet tool). When a token is configured, every request must present it
+via ``Authorization: Bearer <token>`` or ``?token=``. ``/api/health`` is exempt
+so liveness probes work without the token. See :mod:`app.api.auth`.
 """
 
 from __future__ import annotations
 
-from fastapi import FastAPI
+from fastapi import Depends, FastAPI
 from fastapi.middleware.cors import CORSMiddleware
 
+from app.api.auth import require_auth
 from app.api.routers import customer, media
 
-app = FastAPI(title="yunke data mirror")
+app = FastAPI(title="yunke data mirror", dependencies=[Depends(require_auth)])
 
 app.add_middleware(
     CORSMiddleware,
diff --git a/yunke-server/app/config.py b/yunke-server/app/config.py
index 4935fb4..1d85efe 100644
--- a/yunke-server/app/config.py
+++ b/yunke-server/app/config.py
@@ -22,6 +22,11 @@ class Settings(BaseSettings):
     # Fernet key for encrypting stored credentials. Required for real runs;
     # empty default lets tests construct Settings without a key.
     encryption_key: str = ""
+    # Shared bearer token for the viewing API. Empty (default) -> auth DISABLED
+    # (we trust the network boundary; this is a local/intranet tool). When set,
+    # every API request must present this value via ``Authorization: Bearer
+    # <token>`` or ``?token=<token>``. See app/api/auth.py.
+    api_token: str = ""
 
 
 @lru_cache
diff --git a/yunke-server/tests/test_api_auth.py b/yunke-server/tests/test_api_auth.py
new file mode 100644
index 0000000..89b6c2a
--- /dev/null
+++ b/yunke-server/tests/test_api_auth.py
@@ -0,0 +1,164 @@
+"""Tests for the shared-token API auth middleware (Task 2.9).
+
+Trust model being pinned down here (per design doc §8 + plan Task 2.9):
+
+* **No token configured** (``API_TOKEN`` unset / empty) -> auth is DISABLED.
+  This is a local/intranet tool; when no token is set we trust the network
+  boundary and every request passes through.
+* **Token configured** -> every request must present the same token, either as
+  ``Authorization: Bearer <token>`` or ``?token=<token>``. Missing / wrong
+  tokens get ``401 Unauthorized``.
+* ``GET /api/health`` is always reachable (liveness probes must not need to
+  know the token).
+
+The app is driven over an in-process httpx transport (``ASGITransport``) with
+the ``get_session`` dependency overridden to an isolated in-memory SQLite
+session. ``Settings`` is manipulated by mutating the environment plus
+``get_settings.cache_clear()`` so the next request re-reads the config.
+"""
+
+from __future__ import annotations
+
+import pytest
+from httpx import ASGITransport, AsyncClient
+from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
+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 Customer
+
+
+@pytest.fixture
+async def client_factory(monkeypatch):
+    """Return a factory that builds an ``AsyncClient`` against the in-memory app.
+
+    The factory takes an optional ``token``: when given, ``API_TOKEN`` is set in
+    the environment and the ``get_settings`` cache is cleared so the running app
+    picks up the new value on the next request. When ``None``, any leftover
+    ``API_TOKEN`` env var is removed (also cache-cleared) so auth is disabled.
+
+    Each call seeds a fresh in-memory SQLite DB with one ``Customer`` row so
+    ``GET /api/customers`` has something to return beyond an empty page.
+    Resources (client + engine) are torn down at fixture finalization.
+    """
+    created: list[tuple[AsyncClient, object]] = []
+
+    async def _factory(token: str | None = None) -> AsyncClient:
+        # Reconfigure settings for this call: clear cache first so the next
+        # get_settings() inside require_auth observes the env change.
+        get_settings.cache_clear()
+        monkeypatch.delenv("API_TOKEN", raising=False)
+        if token is not None:
+            monkeypatch.setenv("API_TOKEN", token)
+        # Second cache_clear guarantees the *next* get_settings() call rebuilds.
+        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)
+
+        SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
+        async with SessionLocal() as session:
+            session.add(
+                Customer(
+                    friend_id="wx_a",
+                    channel="wechat",
+                    owner_account_id="wx_owner_1",
+                    user_id="u1",
+                    friend_nickname="Alice",
+                    last_chat_time=1000,
+                    version="v1",
+                )
+            )
+            await session.commit()
+
+        async def _override_session():
+            async with SessionLocal() as s:
+                yield s
+
+        app.dependency_overrides[deps.get_session] = _override_session
+        ac = AsyncClient(transport=ASGITransport(app=app), base_url="http://test")
+        await ac.__aenter__()
+        created.append((ac, engine))
+        return ac
+
+    try:
+        yield _factory
+    finally:
+        for ac, engine in created:
+            await ac.aclose()
+            await engine.dispose()
+        app.dependency_overrides.clear()
+
+
+async def test_no_token_configured_allows_access(client_factory):
+    """Default (empty api_token) -> auth disabled, GET /api/customers is 200."""
+    client = await client_factory(token=None)
+    r = await client.get("/api/customers")
+    assert r.status_code == 200
+    assert r.json()["total"] == 1
+
+
+async def test_correct_bearer_token_allows_access(client_factory):
+    """Token configured + matching ``Authorization: Bearer`` -> 200."""
+    client = await client_factory(token="test-secret")
+    r = await client.get(
+        "/api/customers", headers={"Authorization": "Bearer test-secret"}
+    )
+    assert r.status_code == 200
+    assert r.json()["total"] == 1
+
+
+async def test_correct_token_via_query_param_allows_access(client_factory):
+    """Token configured + matching ``?token=`` query param -> 200."""
+    client = await client_factory(token="test-secret")
+    r = await client.get("/api/customers", params={"token": "test-secret"})
+    assert r.status_code == 200
+
+
+async def test_missing_token_returns_401(client_factory):
+    """Token configured + no header/query -> 401."""
+    client = await client_factory(token="test-secret")
+    r = await client.get("/api/customers")
+    assert r.status_code == 401
+
+
+async def test_wrong_bearer_token_returns_401(client_factory):
+    """Token configured + wrong bearer value -> 401."""
+    client = await client_factory(token="test-secret")
+    r = await client.get(
+        "/api/customers", headers={"Authorization": "Bearer wrong-value"}
+    )
+    assert r.status_code == 401
+
+
+async def test_malformed_authorization_header_returns_401(client_factory):
+    """Token configured + header without ``Bearer`` scheme -> 401."""
+    client = await client_factory(token="test-secret")
+    r = await client.get(
+        "/api/customers", headers={"Authorization": "test-secret"}
+    )
+    assert r.status_code == 401
+
+
+async def test_health_accessible_without_token_when_auth_enabled(client_factory):
+    """``/api/health`` is exempt: 200 even when a token is configured."""
+    client = await client_factory(token="test-secret")
+    r = await client.get("/api/health")
+    assert r.status_code == 200
+    assert r.json() == {"status": "ok"}
+
+
+async def test_health_accessible_when_auth_disabled(client_factory):
+    """``/api/health`` is 200 when auth is disabled (sanity check)."""
+    client = await client_factory(token=None)
+    r = await client.get("/api/health")
+    assert r.status_code == 200
+    assert r.json() == {"status": "ok"}