import { Inject, Injectable, UnauthorizedException } from '@nestjs/common'; import type { AuthSessionResponse, AuthUser, ChangePasswordRequest, LoginRequest, RegisterRequest, UpdateCurrentUserRequest, Workspace, } from '@atlas/contracts'; import type { Database, DatabaseTransaction } from '@atlas/database'; import { ApiError } from '../common/api-error.js'; import { IdempotencyService } from '../common/idempotency.service.js'; import { APP_CONFIG, type AppConfig } from '../config/app-config.module.js'; import { DATABASE } from '../database/database.module.js'; import { AuthRateLimiter } from '../security/auth-rate-limiter.service.js'; import { createOpaqueToken, hashPassword, signAccessToken, tokenHash, verifyAccessToken, verifyPassword, } from './auth.crypto.js'; type UserRow = { id: string; email: string; display_name: string | null; avatar_asset_id: string | null; email_verified_at: Date | null; created_at: Date; revision: number; status: 'active' | 'disabled' | 'pending_review'; password_hash: string; }; type WorkspaceRow = { id: string; name: string; role: Workspace['role'] }; @Injectable() export class AuthService { // access token 短期存在于浏览器内存;refresh session 的有效期更长且可服务端撤销。 private readonly accessTtlSeconds = 900; private readonly refreshTtlDays = 30; public constructor( @Inject(DATABASE) private readonly database: Database, @Inject(APP_CONFIG) private readonly config: AppConfig, @Inject(AuthRateLimiter) private readonly rateLimiter: AuthRateLimiter, @Inject(IdempotencyService) private readonly idempotency: IdempotencyService, ) {} public async register(input: RegisterRequest, metadata: RequestMetadata) { if (input.password !== input.password_confirmation) throw validationError( 'password_confirmation', 'PASSWORD_CONFIRMATION_MISMATCH', '两次输入的密码不一致', ); if (input.password.length < this.config.AUTH_PASSWORD_MIN_LENGTH) throw validationError( 'password', 'PASSWORD_TOO_SHORT', `密码至少需要 ${this.config.AUTH_PASSWORD_MIN_LENGTH} 个字符`, ); const email = input.email.trim().toLowerCase(); // 限流必须早于密码哈希和用户查询,避免攻击者消耗 CPU 或探测账户状态。 await this.rateLimiter.consume('register', { ip: metadata.ip, email }); const passwordHash = await hashPassword(input.password); try { return await this.database.transaction(async (transaction) => { // 用户、个人工作空间和 owner membership 必须原子创建,不能留下半成品租户。 const user = await insertUser( transaction, email, passwordHash, input.display_name?.trim() || null, ); const workspace = await createPersonalWorkspace( transaction, user.id, user.display_name ?? email, ); return this.createSessionResponse( transaction, user, workspace, metadata, ); }); } catch (error: unknown) { if (isUniqueViolation(error)) throw new ApiError( 409, 'EMAIL_ALREADY_REGISTERED', '该邮箱已注册', [ { field: 'email', code: 'EMAIL_ALREADY_REGISTERED', message: '该邮箱已注册', }, ], ); throw error; } } public async login(input: LoginRequest, metadata: RequestMetadata) { const email = input.email.trim().toLowerCase(); // 限流与 citext 查询使用同一规范化邮箱,避免大小写绕过限流桶。 await this.rateLimiter.consume('login', { ip: metadata.ip, email }); const result = await this.database.query( 'SELECT id, email, display_name, avatar_asset_id, email_verified_at, created_at, revision, status, password_hash FROM atlas.users WHERE email = $1', [email], ); const user = result.rows[0]; if ( !user || !(await verifyPassword(input.password, user.password_hash)) ) throw new ApiError(401, 'INVALID_CREDENTIALS', '邮箱或密码错误'); if (user.status === 'disabled') throw new ApiError(403, 'ACCOUNT_DISABLED', '账户已停用'); const workspace = await this.findWorkspace(user.id); return this.database.transaction(async (transaction) => { await transaction.query( 'UPDATE atlas.users SET last_login_at = now() WHERE id = $1', [user.id], ); return this.createSessionResponse( transaction, user, workspace, metadata, ); }); } public async refresh( refreshToken: string | undefined, metadata: RequestMetadata, ) { // refresh token 验证前没有可信邮箱,只能按 IP 维度限流。 await this.rateLimiter.consume('refresh', { ip: metadata.ip }); if (!refreshToken) throw new ApiError(401, 'SESSION_EXPIRED', '登录会话已过期'); return this.database.transaction(async (transaction) => { const sessions = await transaction.query< UserRow & { session_id: string } >( `SELECT u.id, u.email, u.display_name, u.avatar_asset_id, u.email_verified_at, u.created_at, u.revision, u.status, u.password_hash, s.id AS session_id FROM atlas.user_sessions s JOIN atlas.users u ON u.id = s.user_id WHERE s.refresh_token_hash = $1 AND s.revoked_at IS NULL AND s.expires_at > now() FOR UPDATE`, [tokenHash(refreshToken)], ); const user = sessions.rows[0]; if (!user || user.status !== 'active') throw new ApiError(401, 'SESSION_EXPIRED', '登录会话已过期'); // 行锁保证并发 refresh 时旧 token 只能成功使用一次。 await transaction.query( 'UPDATE atlas.user_sessions SET revoked_at = now() WHERE id = $1', [user.session_id], ); const workspace = await findWorkspaceInTransaction( transaction, user.id, ); return this.createSessionResponse( transaction, user, workspace, metadata, ); }); } public async logout(sessionId: string) { await this.database.query( 'UPDATE atlas.user_sessions SET revoked_at = now() WHERE id = $1 AND revoked_at IS NULL', [sessionId], ); } public async authenticate(accessToken: string | undefined) { const token = accessToken?.replace(/^Bearer\s+/i, ''); const claims = token ? verifyAccessToken(token, this.config.AUTH_JWT_SECRET) : null; if (!claims) throw new UnauthorizedException(); // JWT 签名有效仍不足够;必须确认对应 session 未撤销且未过期。 const active = await this.database.query( 'SELECT 1 FROM atlas.user_sessions WHERE id = $1 AND user_id = $2 AND revoked_at IS NULL AND expires_at > now()', [claims.sid, claims.sub], ); if (!active.rowCount) throw new UnauthorizedException(); return { userId: claims.sub, sessionId: claims.sid }; } public async me(userId: string, workspace: Workspace) { // workspace 由全局守卫解析,避免控制器或服务各自选择租户造成边界不一致。 const users = await this.database.query( 'SELECT id, email, display_name, avatar_asset_id, email_verified_at, created_at, revision, status, password_hash FROM atlas.users WHERE id = $1', [userId], ); const user = users.rows[0]; if (!user) throw new UnauthorizedException(); const platforms = await this.database.workspaceTransaction( workspace.id, async (transaction) => { const rows = await transaction.query<{ id: string; code: string; name: string; }>( "SELECT id, code, name FROM atlas.platforms WHERE status = 'active' ORDER BY CASE WHEN code = 'OZON' THEN 0 ELSE 1 END, name", ); return rows.rows; }, ); return { user: publicUser(user), workspace, platforms, suggested_context: { platform_id: platforms[0]?.id ?? null, default_store: null, }, }; } public async updateProfile( workspace: Workspace, userId: string, idempotencyKey: string, revision: number, input: UpdateCurrentUserRequest, ) { return this.idempotency.run( { workspaceId: workspace.id, userId, method: 'PATCH', path: '/api/auth/me', key: idempotencyKey, request: { if_match: revision, ...input }, }, async (transaction) => { const user = await findUserInTransaction( transaction, userId, true, ); if (user.revision !== revision) { throw versionConflict(user.revision); } if (input.avatar_asset_id) { // 头像只能引用当前工作空间内已完成入库的素材,不能接受任意对象路径或 URL。 const asset = await transaction.query<{ id: string }>( `SELECT id FROM atlas.media_assets WHERE id = $1`, [input.avatar_asset_id], ); if (!asset.rows[0]) { throw new ApiError( 422, 'AVATAR_ASSET_UNAVAILABLE', '头像素材不存在或不属于当前工作空间', ); } } const updated = await transaction.query( `UPDATE atlas.users SET display_name = CASE WHEN $1 THEN $2 ELSE display_name END, avatar_asset_id = CASE WHEN $3 THEN $4::uuid ELSE avatar_asset_id END, revision = revision + 1, updated_at = now() WHERE id = $5 RETURNING id, email, display_name, avatar_asset_id, email_verified_at, created_at, revision, status, password_hash`, [ Object.hasOwn(input, 'display_name'), input.display_name ?? null, Object.hasOwn(input, 'avatar_asset_id'), input.avatar_asset_id ?? null, userId, ], ); return { status: 200, body: { data: publicUser(updated.rows[0]) }, }; }, ); } public async changePassword( workspace: Workspace, userId: string, sessionId: string, input: ChangePasswordRequest, metadata: RequestMetadata, ) { if (input.password !== input.password_confirmation) { throw validationError( 'password_confirmation', 'PASSWORD_CONFIRMATION_MISMATCH', '两次输入的密码不一致', ); } if (input.password.length < this.config.AUTH_PASSWORD_MIN_LENGTH) { throw validationError( 'password', 'PASSWORD_TOO_SHORT', `密码至少需要 ${this.config.AUTH_PASSWORD_MIN_LENGTH} 个字符`, ); } // 修改密码按已认证用户和来源 IP 双维度限流,避免已泄露会话被用于高频猜测旧密码。 await this.rateLimiter.consume('change_password', { ip: metadata.ip, userId, }); return this.database.transaction(async (transaction) => { const session = await transaction.query<{ id: string }>( `SELECT id FROM atlas.user_sessions WHERE id = $1 AND user_id = $2 AND revoked_at IS NULL AND expires_at > now() FOR UPDATE`, [sessionId, userId], ); if (!session.rows[0]) throw new UnauthorizedException(); const user = await findUserInTransaction(transaction, userId, true); if (user.status !== 'active') { throw new ApiError(403, 'ACCOUNT_DISABLED', '账户已停用'); } if ( !(await verifyPassword(input.old_password, user.password_hash)) ) { throw validationError( 'old_password', 'CURRENT_PASSWORD_INVALID', '当前密码错误', ); } const passwordHash = await hashPassword(input.password); const updated = await transaction.query( `UPDATE atlas.users SET password_hash = $1, updated_at = now() WHERE id = $2 RETURNING id, email, display_name, avatar_asset_id, email_verified_at, created_at, revision, status, password_hash`, [passwordHash, userId], ); // 先撤销所有旧 session,再创建新 session;旧 access token 会立即失效。 await transaction.query( `UPDATE atlas.user_sessions SET revoked_at = now() WHERE user_id = $1 AND revoked_at IS NULL`, [userId], ); return this.createSessionResponse( transaction, updated.rows[0], workspace, metadata, ); }); } private async findWorkspace(userId: string) { const result = await this.database.query( `SELECT w.id, w.name, m.role FROM atlas.workspace_memberships m JOIN atlas.workspaces w ON w.id = m.workspace_id WHERE m.user_id = $1 AND w.status = 'active' ORDER BY m.joined_at LIMIT 1`, [userId], ); return requireWorkspace(result.rows[0]); } private async createSessionResponse( transaction: DatabaseTransaction, user: UserRow, workspace: WorkspaceRow, metadata: RequestMetadata, ) { // refresh token 仅通过 HttpOnly cookie 交给浏览器,响应体只暴露短期 access token。 const refreshToken = createOpaqueToken(); const session = await transaction.query<{ id: string }>( `INSERT INTO atlas.user_sessions (user_id, refresh_token_hash, expires_at, ip_hash, user_agent) VALUES ($1, $2, now() + ($3 || ' days')::interval, $4, $5) RETURNING id`, [ user.id, tokenHash(refreshToken), this.refreshTtlDays, metadata.ipHash, metadata.userAgent, ], ); const now = Math.floor(Date.now() / 1000); const accessToken = signAccessToken( { sub: user.id, sid: session.rows[0].id, iat: now, exp: now + this.accessTtlSeconds, }, this.config.AUTH_JWT_SECRET, ); const payload: AuthSessionResponse = { access_token: accessToken, access_token_expires_in: this.accessTtlSeconds, user: publicUser(user), workspace, }; return { payload, refreshToken }; } } export type RequestMetadata = { ip: string | null; ipHash: string | null; userAgent: string | null; }; function validationError(field: string, code: string, message: string) { return new ApiError(422, 'VALIDATION_FAILED', '请求参数不符合要求', [ { field, code, message }, ]); } function isUniqueViolation(error: unknown): error is { code: string } { return ( typeof error === 'object' && error !== null && 'code' in error && (error as { code: string }).code === '23505' ); } async function findUserInTransaction( transaction: DatabaseTransaction, userId: string, lock: boolean, ): Promise { const result = await transaction.query( `SELECT id, email, display_name, avatar_asset_id, email_verified_at, created_at, revision, status, password_hash FROM atlas.users WHERE id = $1${lock ? ' FOR UPDATE' : ''}`, [userId], ); if (!result.rows[0]) throw new UnauthorizedException(); return result.rows[0]; } function versionConflict(currentRevision: number) { return new ApiError(409, 'VERSION_CONFLICT', '账户资料已被其他操作更新', [ { field: 'If-Match', code: 'VERSION_CONFLICT', message: `当前版本为 ${currentRevision}`, current_revision: currentRevision, }, ]); } function publicUser(user: UserRow): AuthUser { // 白名单映射防止 password_hash 等内部字段随查询结果泄露给客户端。 return { id: user.id, email: user.email, display_name: user.display_name, avatar_asset_id: user.avatar_asset_id, email_verified_at: user.email_verified_at?.toISOString() ?? null, created_at: user.created_at.toISOString(), revision: user.revision, }; } async function insertUser( transaction: DatabaseTransaction, email: string, passwordHash: string, displayName: string | null, ) { // slug 使用用户 UUID,显示名称变化不会影响个人工作空间的稳定标识。 const result = await transaction.query( 'INSERT INTO atlas.users (email, password_hash, display_name) VALUES ($1, $2, $3) RETURNING id, email, display_name, avatar_asset_id, email_verified_at, created_at, revision, status, password_hash', [email, passwordHash, displayName], ); return result.rows[0]; } async function createPersonalWorkspace( transaction: DatabaseTransaction, userId: string, label: string, ) { const workspace = await transaction.query<{ id: string; name: string }>( 'INSERT INTO atlas.workspaces (name, slug) VALUES ($1, $2) RETURNING id, name', [`${label} 的工作空间`, `personal-${userId}`], ); await transaction.query( "INSERT INTO atlas.workspace_memberships (workspace_id, user_id, role) VALUES ($1, $2, 'owner')", [workspace.rows[0].id, userId], ); return { ...workspace.rows[0], role: 'owner' as const }; } async function findWorkspaceInTransaction( transaction: DatabaseTransaction, userId: string, ): Promise { const result = await transaction.query( `SELECT w.id, w.name, m.role FROM atlas.workspace_memberships m JOIN atlas.workspaces w ON w.id = m.workspace_id WHERE m.user_id = $1 AND w.status = 'active' ORDER BY m.joined_at LIMIT 1`, [userId], ); return requireWorkspace(result.rows[0]); } function requireWorkspace(workspace: WorkspaceRow | undefined): WorkspaceRow { if (!workspace) throw new ApiError( 403, 'WORKSPACE_MEMBERSHIP_REQUIRED', '当前账户没有可用工作空间', ); return workspace; }