merge: T4 刷新令牌与服务端可撤销会话(FR-ACC-05)
This commit is contained in:
+67
-23
@@ -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})
|
||||
|
||||
Reference in New Issue
Block a user