"""identity 域 service. 架构文档 §4.2: ``identity`` 不得依赖其他业务域 —— 登录链路不能被业务拖累。 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-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 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.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 async def get_user(self, user_id: int) -> User: user = await repo.get_user(self._session, user_id) if user is None or user.deleted_at is not None or user.status != "active": raise AppError(ErrorCode.UNAUTHENTICATED, message="账号不存在或已注销") return user async def get_or_create_by_external_identity( self, *, provider: str, union_id: str, open_id: str | None = None, register_source: str = "unknown", nickname: str | None = None, ) -> User: """Map an external identity onto an internal ``user_id`` (FR-ACC-01). The internal id is deliberately decoupled from the provider id so a second login channel can be attached later without a data migration. """ identity = await repo.find_identity(self._session, provider=provider, union_id=union_id) if identity is not None: return await self.get_user(identity.user_id) user = repo.add_user( self._session, User( user_id=next_id(), status="active", nickname=nickname, register_source=register_source, last_active_at=datetime.now(UTC), ), ) repo.add_identity( self._session, UserIdentity( identity_id=next_id(), user_id=user.user_id, provider=provider, union_id=union_id, open_id=open_id, ), ) await self._session.flush() return user async def record_consent( self, *, user_id: int, consent_type: str, doc_version: str, client_ip: str | None = None, user_agent: str | None = None, ) -> Consent: """CR-01 留证。人脸的单独同意 (``consent_type='face'``) 必须独立成行。""" consent = repo.add_consent( self._session, Consent( consent_id=next_id(), user_id=user_id, consent_type=consent_type, doc_version=doc_version, client_ip=client_ip, ua=user_agent, ), ) 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, )