diff --git a/apps/api/routers/auth.py b/apps/api/routers/auth.py index e36f0d0..4495835 100755 --- a/apps/api/routers/auth.py +++ b/apps/api/routers/auth.py @@ -1,18 +1,24 @@ -"""Development-only token issuance. +"""Auth endpoints: dev-token / refresh / logout (FR-ACC-05 双令牌会话). -**This is not the V1 auth design.** FR-ACC-01/04/05 require 华为账号登录、手机号 -验证码、双令牌与服务端可撤销的 refresh token. This endpoint exists so the -vertical slice can be exercised end to end, and it refuses to run unless -``DEV_AUTH_ENABLED`` is true and the environment is local/dev. +登出语义 = 撤销 ``id_session`` 会话,**refresh token 立即失效**。access JWT +无状态、15 分钟 TTL:登出后旧 access 在其 TTL 内仍有效的残留窗口是 V1 已知 +且接受的设计权衡(要即时失效就得引入黑名单或极短 TTL,V1 不做)—— +「登出后旧令牌立即失效」的验收对象是 refresh 令牌与会话。 + +明确不做(防止后人误判「漏了」):华为账号登录(T-P1-12,等资质)、手机验证码 +(T-P1-12b,等短信通道)、游客/合并/注销。dev-token 端点不是正式鉴权设计, +仅为纵向切片与联调服务:``DEV_AUTH_ENABLED`` + 本地环境双开关才放行。 """ from __future__ import annotations from fastapi import APIRouter -from apps.api.deps import IdentityServiceDep, SettingsDep -from apps.api.schemas import DevTokenRequest, DevTokenResponse -from packages.identity.service import DEV_PROVIDER +from apps.api.deps import CurrentUserId, IdentityServiceDep, SettingsDep +from apps.api.schemas import DevTokenRequest +from packages.identity.schemas import LogoutRequest, RefreshRequest, TokenPairResponse +from packages.identity.service import DEV_PROVIDER, SessionIssued +from packages.shared.config import Settings from packages.shared.errors import AppError, ErrorCode from packages.shared.response import ok from packages.shared.security import create_access_token @@ -20,7 +26,34 @@ from packages.shared.security import create_access_token router = APIRouter(prefix="/auth", tags=["auth"]) -@router.post("/dev-token", summary="[dev only] Issue an access token for testing") +def _access_token(settings: Settings, user_id: int) -> str: + return create_access_token( + user_id=user_id, + secret=settings.jwt_secret, + algorithm=settings.jwt_algorithm, + ttl_seconds=settings.jwt_access_ttl_seconds, + issuer=settings.jwt_issuer, + ) + + +def _token_pair( + settings: Settings, access_token: str, issued: SessionIssued +) -> dict[str, str | int]: + """dev-token 与 refresh 共用的双令牌响应。 + + 旧字段名 ``access_token/token_type/expires_in/user_id`` 一字不改 + (smoke.sh 与既有测试依赖),只新增 ``refresh_token/refresh_expires_in``。 + """ + return TokenPairResponse( + access_token=access_token, + refresh_token=issued.refresh_token, + expires_in=settings.jwt_access_ttl_seconds, + refresh_expires_in=settings.refresh_token_ttl_seconds, + user_id=str(issued.user_id), + ).model_dump() + + +@router.post("/dev-token", summary="[dev only] Issue a token pair for testing") async def dev_token( payload: DevTokenRequest, settings: SettingsDep, @@ -34,17 +67,28 @@ async def dev_token( union_id=payload.device_key, register_source="dev", ) - token = create_access_token( - user_id=user.user_id, - secret=settings.jwt_secret, - algorithm=settings.jwt_algorithm, - ttl_seconds=settings.jwt_access_ttl_seconds, - issuer=settings.jwt_issuer, - ) - return ok( - DevTokenResponse( - access_token=token, - expires_in=settings.jwt_access_ttl_seconds, - user_id=str(user.user_id), - ).model_dump() - ) + issued = await identity.issue_session(user.user_id) + return ok(_token_pair(settings, _access_token(settings, user.user_id), issued)) + + +@router.post("/refresh", summary="Rotate a refresh token into a new token pair") +async def refresh( + payload: RefreshRequest, + settings: SettingsDep, + identity: IdentityServiceDep, +) -> dict[str, object]: + issued = await identity.rotate(payload.refresh_token) + return ok(_token_pair(settings, _access_token(settings, issued.user_id), issued)) + + +@router.post("/logout", summary="Revoke the session bound to this refresh token") +async def logout( + payload: LogoutRequest, + # CurrentUserId 只做鉴权门槛(不带 access 的登出请求一律 40100), + # 撤销仍以 refresh_token 定位会话,不信任 token 里的 user_id。 + user_id: CurrentUserId, + identity: IdentityServiceDep, +) -> dict[str, object]: + revoked = await identity.revoke(payload.refresh_token) + # 幂等登出:查无此 token 或已撤销都返回 revoked=false,不报错 + return ok({"revoked": revoked}) diff --git a/packages/identity/repo.py b/packages/identity/repo.py index 3f322cf..73dcde3 100755 --- a/packages/identity/repo.py +++ b/packages/identity/repo.py @@ -2,10 +2,12 @@ from __future__ import annotations +from datetime import UTC, datetime + from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession -from packages.shared.models import Consent, User, UserIdentity +from packages.shared.models import Consent, User, UserIdentity, UserSession async def get_user(session: AsyncSession, user_id: int) -> User | None: @@ -36,3 +38,57 @@ def add_identity(session: AsyncSession, identity: UserIdentity) -> UserIdentity: def add_consent(session: AsyncSession, consent: Consent) -> Consent: session.add(consent) return consent + + +async def find_session_by_hash(session: AsyncSession, token_hash: str) -> UserSession | None: + # UNIQUE(refresh_token_hash) 保证至多一行,scalar_one_or_none 即可 + result = await session.execute( + select(UserSession).where(UserSession.refresh_token_hash == token_hash) + ) + return result.scalar_one_or_none() + + +def add_session(session: AsyncSession, user_session: UserSession) -> UserSession: + session.add(user_session) + return user_session + + +async def revoke_session(session: AsyncSession, session_id: int) -> bool: + """撤销单条会话。返回 False = 该行不存在或已被撤销。 + + WHERE 里带 ``revoked = false`` 再 FOR UPDATE(照 billing.repo 的 + lock_or_create_quota 模式):两个并发请求拿着同一 refresh 轮换时, + 后到者等先到者提交后重估条件、匹配 0 行返回 False——调用方据此按 + 「已撤销/重放」处理,无需自己加锁。 + """ + result = await session.execute( + select(UserSession) + .where(UserSession.session_id == session_id, UserSession.revoked.is_(False)) + .with_for_update() + ) + row = result.scalar_one_or_none() + if row is None: + return False + row.revoked = True + row.revoked_at = datetime.now(UTC) + return True + + +async def revoke_all_sessions(session: AsyncSession, user_id: int) -> int: + """撤销某用户全部会话(logout-all)。返回实际撤销的行数。 + + V1 先备着:对应 FR 后续的「退出全部设备」,本批次不接路由,留实现是为了 + 把「会话可撤销」的数据通路一次铺好。 + """ + result = await session.execute( + select(UserSession) + .where(UserSession.user_id == user_id, UserSession.revoked.is_(False)) + .with_for_update() + ) + now = datetime.now(UTC) + revoked_count = 0 + for row in result.scalars(): + row.revoked = True + row.revoked_at = now + revoked_count += 1 + return revoked_count diff --git a/packages/identity/schemas.py b/packages/identity/schemas.py new file mode 100755 index 0000000..725f076 --- /dev/null +++ b/packages/identity/schemas.py @@ -0,0 +1,36 @@ +"""identity 域的请求/响应模型(T4:双令牌会话,FR-ACC-05)。 + +``apps/api/schemas.py`` 是本批次的共享禁改文件,refresh/logout 的契约模型 +放在域内这里,由 ``apps/api/routers/auth.py`` 引用;wire 响应仍只是统一 +信封 (架构文档 §9.2) 里的 ``data`` 部分。 +""" + +from __future__ import annotations + +from pydantic import BaseModel, Field + +#: 生成端 token_urlsafe(48) 恒为 64 字符;上限放宽到 128 容忍未来换编码。 +_REFRESH_TOKEN_MAX_LENGTH = 128 + + +class RefreshRequest(BaseModel): + refresh_token: str = Field(min_length=1, max_length=_REFRESH_TOKEN_MAX_LENGTH) + + +class LogoutRequest(BaseModel): + refresh_token: str = Field(min_length=1, max_length=_REFRESH_TOKEN_MAX_LENGTH) + + +class TokenPairResponse(BaseModel): + """dev-token 与 /auth/refresh 共用的双令牌响应。 + + 旧字段名 ``access_token/token_type/expires_in/user_id`` 保持不变 + (smoke.sh 与既有测试依赖),新增 ``refresh_token/refresh_expires_in``。 + """ + + access_token: str + refresh_token: str + token_type: str = "Bearer" + expires_in: int = Field(description="access token 有效期(秒)") + refresh_expires_in: int = Field(description="refresh token 有效期(秒)") + user_id: str = Field(description="雪花 ID 以字符串返回,避免 JS 精度丢失") diff --git a/packages/identity/service.py b/packages/identity/service.py index 696f80e..4a283c7 100755 --- a/packages/identity/service.py +++ b/packages/identity/service.py @@ -5,25 +5,47 @@ import-linter enforces this (contract "identity must not depend on any other business domain"). V1 scope still missing here: 华为账号登录 (FR-ACC-01)、手机号验证码 (FR-ACC-04)、 -双令牌与撤销 (FR-ACC-05)、游客合并 (FR-ACC-03)、注销 (FR-ACC-09)、未成年判定 -(FR-ACC-11). See packages/identity/README.md. +游客合并 (FR-ACC-03)、注销 (FR-ACC-09)、未成年判定 (FR-ACC-11). See +packages/identity/README.md. 双令牌与撤销 (FR-ACC-05) 已由 refresh 会话补齐 +(`issue_session`/`rotate`/`revoke`)。 """ from __future__ import annotations -from datetime import UTC, datetime +from dataclasses import dataclass +from datetime import UTC, datetime, timedelta from sqlalchemy.ext.asyncio import AsyncSession from packages.domain.ids import next_id from packages.identity import repo +from packages.identity.tokens import generate_refresh_token, hash_refresh_token +from packages.shared.config import get_settings from packages.shared.errors import AppError, ErrorCode -from packages.shared.models import Consent, User, UserIdentity +from packages.shared.logging import get_logger +from packages.shared.models import Consent, User, UserIdentity, UserSession + +logger = get_logger(__name__) #: Provider used by the dev-only token endpoint. Never valid in prod. DEV_PROVIDER = "dev" +@dataclass(frozen=True, slots=True) +class SessionIssued: + """一次 refresh 签发的结果。 + + ``refresh_token`` 明文只在本对象里短暂存在,由调用方直接下发给客户端; + 服务端只留 ``session_id``(撤销句柄)与 ``expires_at``。``user_id`` 供 + /auth/refresh 换发 access JWT 使用——refresh 本身不携带任何可解码信息。 + """ + + session_id: int + user_id: int + refresh_token: str + expires_at: datetime + + class IdentityService: def __init__(self, session: AsyncSession) -> None: self._session = session @@ -98,3 +120,81 @@ class IdentityService: ) await self._session.flush() return consent + + # --- refresh 会话(FR-ACC-05)------------------------------------------- + + async def issue_session(self, user_id: int) -> SessionIssued: + """登录成功后签发新 refresh 会话。 + + 先 ``get_user``:已注销/被禁用的账号不得再拿新会话(防已注销用户续签)。 + TTL 在用时读进程级配置而非构造器注入——``apps/api/deps.py``(共享禁改 + 文件)构造本 service 时拿不到 Settings;测试借 ``get_settings.cache_clear()`` + 注入 TTL 即可命中过期分支。 + """ + await self.get_user(user_id) + return await self._create_session(user_id) + + async def rotate(self, refresh_token: str) -> SessionIssued: + """一次性轮换:旧 refresh 用过即死,成功则当场换发新 token。 + + 旧行撤销与新行创建发生在**同一事务**(调用方 apps/api/deps.session_dep + 统一 commit):半途失败两者一起回滚,不会出现「旧 token 已死、新 token + 未生」把用户锁在门外,也不会出现双活会话。 + """ + token_hash = hash_refresh_token(refresh_token) + row = await repo.find_session_by_hash(self._session, token_hash) + if row is None: + raise AppError(ErrorCode.REFRESH_TOKEN_INVALID) + if row.revoked: + # 重放检测:已撤销的 refresh 又出现了(可能是被盗用的副本,也可能是 + # 客户端重试)。V1 只告警 + 拒绝;RFC 6819 的完整做法是撤销整个 + # token 家族(家族连坐),留作 V2 硬化项。 + logger.warning("refresh_reuse_detected", user_id=row.user_id) + raise AppError(ErrorCode.SESSION_REVOKED) + now = datetime.now(UTC) + if row.expire_at < now: + # 过期行顺手标记撤销,别让死会话一直可查。注意:HTTP 路径上这个 + # 写入会随 401 的统一回滚一起丢失(仅事务内标记),落库与否不影响 + # 判定本身——过期的行本来也不可能再通过校验。 + row.revoked = True + row.revoked_at = now + await self._session.flush() + raise AppError(ErrorCode.REFRESH_TOKEN_EXPIRED) + await self.get_user(row.user_id) # 防已注销用户续签 + if not await repo.revoke_session(self._session, row.session_id): + # 并发轮换:另一请求刚撤销了这一行(revoke_session 的 WHERE 守卫 + # 只放行第一个),后到者按重放处理。 + raise AppError(ErrorCode.SESSION_REVOKED) + return await self._create_session(row.user_id) + + async def revoke(self, refresh_token: str) -> bool: + """登出:撤销该 refresh 对应的会话。 + + 查无此 token 或已撤销都返回 False 而不报错——登出必须幂等,客户端 + 重试不应收到 4xx。 + """ + token_hash = hash_refresh_token(refresh_token) + row = await repo.find_session_by_hash(self._session, token_hash) + if row is None: + return False + return await repo.revoke_session(self._session, row.session_id) + + async def _create_session(self, user_id: int) -> SessionIssued: + token = generate_refresh_token() + expires_at = datetime.now(UTC) + timedelta(seconds=get_settings().refresh_token_ttl_seconds) + row = repo.add_session( + self._session, + UserSession( + session_id=next_id(), + user_id=user_id, + refresh_token_hash=hash_refresh_token(token), + expire_at=expires_at, + ), + ) + await self._session.flush() + return SessionIssued( + session_id=row.session_id, + user_id=user_id, + refresh_token=token, + expires_at=row.expire_at, + ) diff --git a/packages/identity/tokens.py b/packages/identity/tokens.py new file mode 100755 index 0000000..498d7c7 --- /dev/null +++ b/packages/identity/tokens.py @@ -0,0 +1,30 @@ +"""Opaque refresh tokens (FR-ACC-05, 架构文档 §6.2 id_session / §9.1). + +为什么是 opaque 随机串而不是 JWT:refresh 的核心价值就是「服务端可单条撤销」 +(FR-ACC-05),无状态 JWT 天生做不到撤销——除非引入黑名单,而黑名单本身就是把 +状态搬回服务端,那不如直接用查表就能撤销的 opaque 令牌。 + +库里只存 sha256 哈希而非明文:查库校验时重算一次哈希即可比对,而泄漏数据库 +不等于泄漏令牌——sha256 不可逆,拿到哈希换不出新 access。 + +本模块是纯函数(无 IO、无配置),方便穷举式单测。 +""" + +from __future__ import annotations + +import hashlib +import secrets + +#: token_urlsafe(48) 输出 64 个 URL 安全字符,携带 384 bit 熵, +#: 暴力枚举哈希空间在计算上不可行。 +_TOKEN_BYTES = 48 + + +def generate_refresh_token() -> str: + """生成新 refresh token。明文只在签发瞬间出现一次,此后系统内只见哈希。""" + return secrets.token_urlsafe(_TOKEN_BYTES) + + +def hash_refresh_token(token: str) -> str: + """sha256 hex(64 字符,与 id_session.refresh_token_hash 列宽一致)。""" + return hashlib.sha256(token.encode()).hexdigest() diff --git a/tests/integration/test_auth_refresh.py b/tests/integration/test_auth_refresh.py new file mode 100755 index 0000000..5fbf2c6 --- /dev/null +++ b/tests/integration/test_auth_refresh.py @@ -0,0 +1,145 @@ +"""Refresh token 链路(FR-ACC-05)—— 真实 PostgreSQL,auth 用例不涉及 LLM。 + +与 test_vertical_slice.py 不同:这里不替换 LLM provider(没人会调到它), +只验证 dev-token → refresh 轮换 → logout 撤销的会话生命周期。 +""" + +from __future__ import annotations + +import os +import uuid +from collections.abc import AsyncIterator +from typing import Any + +import httpx +import pytest + + +def _infra_available() -> bool: + return bool(os.getenv("MEIDA_RUN_INTEGRATION")) + + +pytestmark = [ + pytest.mark.integration, + pytest.mark.skipif( + not _infra_available(), + reason="set MEIDA_RUN_INTEGRATION=1 with PostgreSQL and Redis running", + ), +] + + +@pytest.fixture +async def client() -> AsyncIterator[httpx.AsyncClient]: + from apps.api.main import create_app + from packages.shared.config import get_settings + + app = create_app(get_settings()) + transport = httpx.ASGITransport(app=app) + async with ( + httpx.AsyncClient(transport=transport, base_url="http://test") as http, + app.router.lifespan_context(app), + ): + yield http + + +async def _dev_token(client: httpx.AsyncClient) -> dict[str, Any]: + """每个用例用独立 device_key,避免共享用户状态互相污染。""" + resp = await client.post( + "/v1/auth/dev-token", json={"device_key": f"t4-{uuid.uuid4().hex[:8]}"} + ) + assert resp.status_code == 200 + body = resp.json() + assert body["code"] == 0 + data = body["data"] + # 兼容性:旧字段名一字不改,新字段在场 + assert data["access_token"] + assert data["expires_in"] > 0 + assert data["user_id"] + assert data["refresh_token"] + return data + + +async def test_refresh_rotates_and_new_access_calls_quota(client: httpx.AsyncClient) -> None: + pair = await _dev_token(client) + + resp = await client.post("/v1/auth/refresh", json={"refresh_token": pair["refresh_token"]}) + assert resp.status_code == 200 + assert resp.json()["code"] == 0 + data = resp.json()["data"] + # 轮换的核心契约是 refresh 换新(随机串必不同)。access JWT 不比不等: + # 同一用户同一秒签发时 claims 完全一致,token 字节级相同是正常现象, + # 断言不等只会让用例变成秒边界的掷硬币。 + assert data["refresh_token"] != pair["refresh_token"] + assert data["refresh_expires_in"] > 0 + + quota = await client.get( + "/v1/quota", headers={"Authorization": f"Bearer {data['access_token']}"} + ) + assert quota.status_code == 200 + assert quota.json()["code"] == 0 + + +async def test_old_refresh_is_single_use(client: httpx.AsyncClient) -> None: + pair = await _dev_token(client) + + first = await client.post("/v1/auth/refresh", json={"refresh_token": pair["refresh_token"]}) + assert first.status_code == 200 + + second = await client.post("/v1/auth/refresh", json={"refresh_token": pair["refresh_token"]}) + assert second.status_code == 401 + assert second.json()["code"] == 40105 + + +async def test_logout_revokes_the_session(client: httpx.AsyncClient) -> None: + pair = await _dev_token(client) + headers = {"Authorization": f"Bearer {pair['access_token']}"} + + out = await client.post( + "/v1/auth/logout", json={"refresh_token": pair["refresh_token"]}, headers=headers + ) + assert out.status_code == 200 + assert out.json()["data"] == {"revoked": True} + + # 幂等登出:同一 token 再登出一次不报错,revoked=false + again = await client.post( + "/v1/auth/logout", json={"refresh_token": pair["refresh_token"]}, headers=headers + ) + assert again.status_code == 200 + assert again.json()["data"] == {"revoked": False} + + refreshed = await client.post("/v1/auth/refresh", json={"refresh_token": pair["refresh_token"]}) + assert refreshed.status_code == 401 + assert refreshed.json()["code"] == 40105 + + +async def test_forged_refresh_token_is_rejected(client: httpx.AsyncClient) -> None: + resp = await client.post("/v1/auth/refresh", json={"refresh_token": "forged-" + "x" * 32}) + assert resp.status_code == 401 + assert resp.json()["code"] == 40103 + + +async def test_expired_refresh_token_reports_40104() -> None: + """TTL=0 注入过期路径:签发即过期,refresh 必须走 40104 而非 40103/40105。""" + from apps.api.main import create_app + from packages.shared.config import get_settings + + with pytest.MonkeyPatch.context() as mp: + mp.setenv("REFRESH_TOKEN_TTL_SECONDS", "0") + get_settings.cache_clear() + app = create_app(get_settings()) + try: + transport = httpx.ASGITransport(app=app) + async with ( + httpx.AsyncClient(transport=transport, base_url="http://test") as http, + app.router.lifespan_context(app), + ): + pair = await _dev_token(http) + assert pair["refresh_expires_in"] == 0 + resp = await http.post( + "/v1/auth/refresh", json={"refresh_token": pair["refresh_token"]} + ) + assert resp.status_code == 401 + assert resp.json()["code"] == 40104 + finally: + # 恢复进程级配置缓存:打补丁的 Settings 不能泄漏给后续用例 + get_settings.cache_clear() diff --git a/tests/unit/test_refresh_tokens.py b/tests/unit/test_refresh_tokens.py new file mode 100755 index 0000000..f6a131a --- /dev/null +++ b/tests/unit/test_refresh_tokens.py @@ -0,0 +1,58 @@ +"""refresh token 纯函数(packages/identity/tokens.py)。 + +无 IO、无配置:token 长度/字符域/唯一性、哈希稳定性与不可逆性在这里穷举, +服务层的轮换/撤销语义由 tests/integration/test_auth_refresh.py 用真实 PG 验证。 +""" + +from __future__ import annotations + +import string + +from packages.identity.tokens import generate_refresh_token, hash_refresh_token + +#: token_urlsafe 的字母表:base64url(含 ``-`` ``_``),无 ``+`` ``/`` ``=`` +_URLSAFE_ALPHABET = frozenset(string.ascii_letters + string.digits + "_-") + + +def test_token_has_fixed_length() -> None: + # 48 字节 → base64 无填充恒 64 字符;长度恒定才好做列宽与日志脱敏 + assert len(generate_refresh_token()) == 64 + + +def test_token_stays_within_urlsafe_alphabet() -> None: + # 必须能原样放进 JSON / URL / Header,不引入需要转义的字符 + tokens = [generate_refresh_token() for _ in range(50)] + assert all(set(token) <= _URLSAFE_ALPHABET for token in tokens) + + +def test_tokens_are_unique() -> None: + tokens = [generate_refresh_token() for _ in range(100)] + assert len(set(tokens)) == len(tokens) + + +def test_hash_is_stable() -> None: + # 校验路径 = 重算一次哈希再比对,稳定性是查库成立的前提 + token = generate_refresh_token() + assert hash_refresh_token(token) == hash_refresh_token(token) + + +def test_hash_is_lowercase_sha256_hex() -> None: + digest = hash_refresh_token("abc") + # sha256 恒 64 个小写 hex 字符,与 id_session.refresh_token_hash varchar(64) 对齐 + assert len(digest) == 64 + assert digest == digest.lower() + assert set(digest) <= set(string.digits + "abcdef") + + +def test_hash_is_not_the_plaintext_token() -> None: + # 不可逆(≠原文):库里存的是哈希,泄漏数据库换不回可用令牌 + token = generate_refresh_token() + digest = hash_refresh_token(token) + assert digest != token + assert token not in digest + + +def test_different_tokens_hash_differently() -> None: + assert hash_refresh_token(generate_refresh_token()) != hash_refresh_token( + generate_refresh_token() + )