From ba502297ac9d2b8deb127e276fc218c13efd668e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=99=88=E6=AD=86?= <2685475410@qq.com> Date: Thu, 10 Sep 2026 10:13:45 +0800 Subject: [PATCH] =?UTF-8?q?=E2=9C=A8=20feat:=20=E9=98=B6=E6=AE=B5=200=20?= =?UTF-8?q?=E6=94=B6=E5=B0=BE=E2=80=94=E2=80=94=E8=AE=A4=E8=AF=81=E8=BE=B9?= =?UTF-8?q?=E7=95=8C=E3=80=81=E8=AF=B7=E6=B1=82=E4=B8=8A=E4=B8=8B=E6=96=87?= =?UTF-8?q?=E4=B8=8E=E5=9E=82=E7=9B=B4=E5=88=87=E7=89=87=EF=BC=88WP-01=20?= =?UTF-8?q?=E7=AC=AC=205-7=20=E6=AD=A5=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 认证：Web Session（__Host- Cookie、HttpOnly/Secure/SameSite=Lax、CSPRNG opaque token 数据库只存哈希）、CSRF 双提交、Origin 校验、dev 登录/登出端点；未认证 401、缺 CSRF 403 - 中间件：Hunjing-Request-ID 回显/生成（request_ UUIDv4）+ W3C traceparent 校验透传；JSON 结构化日志与敏感头脱敏 - 迁移 2：conversation/branch/message/run/idempotency_record 五表（含检查约束与复合索引）；Session 补 token/csrf 哈希列 - 垂直切片：POST /v1/conversations/{id}/messages 按 Idempotency-Key 幂等（同键同体 200 重放、异体 409），同事务原子写 Message+Run+Audit+Outbox+幂等记录；ContentBlock text/markdown 判别联合（未知类型 422） - SSE：GET /v1/runs/{id}/events 返回快照与终止状态事件 envelope - 测试 26 项全绿：IDOR 跨域 404 fail-closed、失败注入回滚无孤儿（Run 崩溃后 message/run/audit/outbox/幂等全零）、SSE 事件序、auth 边界；ruff/mypy strict 干净 - 修复真实问题：asyncpg 跨事件循环（anyio/pytest-asyncio 双 loop、模块级 engine 改为 app 实例持有）、httpx 拒存 http 下 Secure Cookie --- ...垂直切片_会话字段与会话树_运行_幂等记录.py | 176 +++++++++++++++++ backend/app/auth/__init__.py | 1 + backend/app/auth/dependencies.py | 56 ++++++ backend/app/auth/routes.py | 60 ++++++ backend/app/auth/service.py | 71 +++++++ backend/app/conversations/__init__.py | 1 + backend/app/conversations/routes.py | 138 +++++++++++++ backend/app/conversations/schemas.py | 60 ++++++ backend/app/conversations/service.py | 182 ++++++++++++++++++ backend/app/core/middleware.py | 80 ++++++++ backend/app/db/session.py | 37 ++++ backend/app/main.py | 11 ++ backend/app/models.py | 110 +++++++++++ backend/tests/conftest.py | 32 +++ backend/tests/test_conversations.py | 148 ++++++++++++++ backend/tests/test_models.py | 2 + 16 files changed, 1165 insertions(+) create mode 100644 backend/alembic/versions/140fe7181a6e_垂直切片_会话字段与会话树_运行_幂等记录.py create mode 100644 backend/app/auth/__init__.py create mode 100644 backend/app/auth/dependencies.py create mode 100644 backend/app/auth/routes.py create mode 100644 backend/app/auth/service.py create mode 100644 backend/app/conversations/__init__.py create mode 100644 backend/app/conversations/routes.py create mode 100644 backend/app/conversations/schemas.py create mode 100644 backend/app/conversations/service.py create mode 100644 backend/app/core/middleware.py create mode 100644 backend/app/db/session.py create mode 100644 backend/tests/test_conversations.py diff --git a/backend/alembic/versions/140fe7181a6e_垂直切片_会话字段与会话树_运行_幂等记录.py b/backend/alembic/versions/140fe7181a6e_垂直切片_会话字段与会话树_运行_幂等记录.py new file mode 100644 index 0000000..b573fa9 --- /dev/null +++ b/backend/alembic/versions/140fe7181a6e_垂直切片_会话字段与会话树_运行_幂等记录.py @@ -0,0 +1,176 @@ +"""垂直切片：会话字段与会话树、运行、幂等记录 + +Revision ID: 140fe7181a6e +Revises: e16a40566b45 +Create Date: 2026-09-10 10:06:24.354536 + +""" +from collections.abc import Sequence + +import sqlalchemy as sa +from sqlalchemy.dialects import postgresql + +from alembic import op + +# revision identifiers, used by Alembic. +revision: str = '140fe7181a6e' +down_revision: str | Sequence[str] | None = 'e16a40566b45' +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + """Upgrade schema.""" + # ### commands auto generated by Alembic - please adjust! ### + op.create_table('conversation', + sa.Column('id', sa.String(), nullable=False), + sa.Column('personal_domain_id', sa.String(), nullable=False), + sa.Column('title', sa.String(), nullable=False), + sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False), + sa.Column('updated_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False), + sa.ForeignKeyConstraint(['personal_domain_id'], ['personal_domain.id'], name=op.f('fk_conversation_personal_domain'), ondelete='cascade'), + sa.PrimaryKeyConstraint('id', name=op.f('pk_conversation')) + ) + op.create_index(op.f('ix_conversation_personal_domain_id'), 'conversation', ['personal_domain_id'], unique=False) + op.create_table('idempotency_record', + sa.Column('key', sa.String(), nullable=False), + sa.Column('owner_domain_id', sa.String(), nullable=False), + sa.Column('route', sa.String(), nullable=False), + sa.Column('request_hash', sa.String(), nullable=False), + sa.Column('response_status', sa.Integer(), nullable=False), + sa.Column('resource_ids', postgresql.JSONB(astext_type=sa.Text()), nullable=False), + sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False), + sa.ForeignKeyConstraint(['owner_domain_id'], ['personal_domain.id'], name=op.f('fk_idempotency_record_personal_domain'), ondelete='cascade'), + sa.PrimaryKeyConstraint('key', name=op.f('pk_idempotency_record')) + ) + op.create_index('uq_idempotency_owner_route_key', 'idempotency_record', ['owner_domain_id', 'route', 'key'], unique=True) + op.create_table('branch', + sa.Column('id', sa.String(), nullable=False), + sa.Column('conversation_id', sa.String(), nullable=False), + sa.Column('title', sa.String(), nullable=False), + sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False), + sa.ForeignKeyConstraint(['conversation_id'], ['conversation.id'], name=op.f('fk_branch_conversation'), ondelete='cascade'), + sa.PrimaryKeyConstraint('id', name=op.f('pk_branch')) + ) + op.create_index(op.f('ix_branch_conversation_id'), 'branch', ['conversation_id'], unique=False) + op.create_table('run', + sa.Column('id', sa.String(), nullable=False), + sa.Column('personal_domain_id', sa.String(), nullable=False), + sa.Column('conversation_id', sa.String(), nullable=True), + sa.Column('status', sa.String(), nullable=False), + sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False), + sa.Column('completed_at', sa.DateTime(timezone=True), nullable=True), + sa.CheckConstraint("status IN ('queued', 'running', 'waiting_approval', 'completed', 'failed', 'cancelled')", name=op.f('ck_run_status_values')), + sa.ForeignKeyConstraint(['conversation_id'], ['conversation.id'], name=op.f('fk_run_conversation'), ondelete='cascade'), + sa.ForeignKeyConstraint(['personal_domain_id'], ['personal_domain.id'], name=op.f('fk_run_personal_domain'), ondelete='cascade'), + sa.PrimaryKeyConstraint('id', name=op.f('pk_run')) + ) + op.create_index(op.f('ix_run_conversation_id'), 'run', ['conversation_id'], unique=False) + op.create_index('ix_run_domain_status', 'run', ['personal_domain_id', 'status'], unique=False) + op.create_index(op.f('ix_run_personal_domain_id'), 'run', ['personal_domain_id'], unique=False) + op.create_table('message', + sa.Column('id', sa.String(), nullable=False), + sa.Column('conversation_id', sa.String(), nullable=False), + sa.Column('branch_id', sa.String(), nullable=False), + sa.Column('role', sa.String(), nullable=False), + sa.Column('content', postgresql.JSONB(astext_type=sa.Text()), nullable=False), + sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False), + sa.CheckConstraint("role IN ('user', 'assistant', 'system', 'tool')", name=op.f('ck_message_role_values')), + sa.ForeignKeyConstraint(['branch_id'], ['branch.id'], name=op.f('fk_message_branch'), ondelete='cascade'), + sa.ForeignKeyConstraint(['conversation_id'], ['conversation.id'], name=op.f('fk_message_conversation'), ondelete='cascade'), + sa.PrimaryKeyConstraint('id', name=op.f('pk_message')) + ) + op.create_index(op.f('ix_message_branch_id'), 'message', ['branch_id'], unique=False) + op.create_index(op.f('ix_message_conversation_id'), 'message', ['conversation_id'], unique=False) + op.drop_table_comment( + 'audit_event', + existing_comment='不可变审计链：与权威状态、Outbox 同事务提交', + schema=None + ) + op.drop_table_comment( + 'outbox_event', + existing_comment='Transactional Outbox：已提交领域事实的可靠投递队列', + schema=None + ) + op.drop_table_comment( + 'personal_domain', + existing_comment='个人数据域：数据归属与授权边界，绑定固定数据区域', + schema=None + ) + op.drop_table_comment( + 'service_account', + existing_comment='服务账号：API key 明文只显示一次，只存哈希与前缀', + schema=None + ) + op.add_column('session', sa.Column('token_hash', sa.String(), nullable=False)) + op.add_column('session', sa.Column('csrf_token_hash', sa.String(), nullable=False)) + op.create_unique_constraint(op.f('uq_session_token_hash'), 'session', ['token_hash']) + op.drop_table_comment( + 'session', + existing_comment='Web 会话：ID 为 CSPRNG opaque，仅存校验值', + schema=None + ) + op.drop_table_comment( + 'user', + existing_comment='用户（WP-01）：多凭证与账户根', + schema=None + ) + # ### end Alembic commands ### + + +def downgrade() -> None: + """Downgrade schema.""" + # ### commands auto generated by Alembic - please adjust! ### + op.create_table_comment( + 'user', + '用户（WP-01）：多凭证与账户根', + existing_comment=None, + schema=None + ) + op.create_table_comment( + 'session', + 'Web 会话：ID 为 CSPRNG opaque，仅存校验值', + existing_comment=None, + schema=None + ) + op.drop_constraint(op.f('uq_session_token_hash'), 'session', type_='unique') + op.drop_column('session', 'csrf_token_hash') + op.drop_column('session', 'token_hash') + op.create_table_comment( + 'service_account', + '服务账号：API key 明文只显示一次，只存哈希与前缀', + existing_comment=None, + schema=None + ) + op.create_table_comment( + 'personal_domain', + '个人数据域：数据归属与授权边界，绑定固定数据区域', + existing_comment=None, + schema=None + ) + op.create_table_comment( + 'outbox_event', + 'Transactional Outbox：已提交领域事实的可靠投递队列', + existing_comment=None, + schema=None + ) + op.create_table_comment( + 'audit_event', + '不可变审计链：与权威状态、Outbox 同事务提交', + existing_comment=None, + schema=None + ) + op.drop_index(op.f('ix_message_conversation_id'), table_name='message') + op.drop_index(op.f('ix_message_branch_id'), table_name='message') + op.drop_table('message') + op.drop_index(op.f('ix_run_personal_domain_id'), table_name='run') + op.drop_index('ix_run_domain_status', table_name='run') + op.drop_index(op.f('ix_run_conversation_id'), table_name='run') + op.drop_table('run') + op.drop_index(op.f('ix_branch_conversation_id'), table_name='branch') + op.drop_table('branch') + op.drop_index('uq_idempotency_owner_route_key', table_name='idempotency_record') + op.drop_table('idempotency_record') + op.drop_index(op.f('ix_conversation_personal_domain_id'), table_name='conversation') + op.drop_table('conversation') + # ### end Alembic commands ### diff --git a/backend/app/auth/__init__.py b/backend/app/auth/__init__.py new file mode 100644 index 0000000..5ab6344 --- /dev/null +++ b/backend/app/auth/__init__.py @@ -0,0 +1 @@ +"""认证模块：Web 会话与依赖。""" diff --git a/backend/app/auth/dependencies.py b/backend/app/auth/dependencies.py new file mode 100644 index 0000000..d6fec3a --- /dev/null +++ b/backend/app/auth/dependencies.py @@ -0,0 +1,56 @@ +"""认证依赖：当前用户解析与状态变更防护。""" + +from typing import Annotated + +from fastapi import Depends, Header, HTTPException, Request, status +from sqlalchemy.ext.asyncio import AsyncSession + +from app.auth.service import SESSION_COOKIE, resolve_session +from app.db.session import get_db +from app.models import User + +ALLOWED_ORIGINS: set[str] = set() # 由配置注入；空集表示本地开发放行同源 + + +async def get_current_user( + request: Request, + db: Annotated[AsyncSession, Depends(get_db)], +) -> User: + token = request.cookies.get(SESSION_COOKIE) + if token is None: + raise HTTPException(status.HTTP_401_UNAUTHORIZED, "未登录") + session = await resolve_session(db, token) + if session is None: + raise HTTPException(status.HTTP_401_UNAUTHORIZED, "会话无效或已过期") + user = await db.get(User, session.user_id) + if user is None: + raise HTTPException(status.HTTP_401_UNAUTHORIZED, "会话主体不存在") + # 请求级上下文供后续依赖使用（不重复查询） + request.state.session_id = session.id + request.state.session_csrf_hash = session.csrf_token_hash + return user + + +async def require_state_change( + request: Request, + user: Annotated[User, Depends(get_current_user)], + hunjing_csrf_token: Annotated[str | None, Header(alias="Hunjing-CSRF-Token")] = None, +) -> User: + """Cookie 状态变更：Origin 校验 + CSRF 双提交（契约 2.1 节）。""" + if ALLOWED_ORIGINS: + origin = request.headers.get("Origin") + if origin is not None and origin not in ALLOWED_ORIGINS: + raise HTTPException(status.HTTP_403_FORBIDDEN, "Origin 不被允许") + csrf_hash = getattr(request.state, "session_csrf_hash", None) + if csrf_hash is None or not _csrf_matches(csrf_hash, hunjing_csrf_token): + raise HTTPException(status.HTTP_403_FORBIDDEN, "CSRF 校验失败") + return user + + +def _csrf_matches(expected_hash: str, provided: str | None) -> bool: + import hashlib + import secrets + + if provided is None: + return False + return secrets.compare_digest(expected_hash, hashlib.sha256(provided.encode()).hexdigest()) diff --git a/backend/app/auth/routes.py b/backend/app/auth/routes.py new file mode 100644 index 0000000..eb19c52 --- /dev/null +++ b/backend/app/auth/routes.py @@ -0,0 +1,60 @@ +"""认证路由：开发档登录/登出（Web Session Cookie）。 + +正式 OAuth 2.1 授权码流程（/oauth2/authorize、/oauth2/token、PKCE）属 WP-01 +后续条目；本路由只建立 Session 边界骨架供垂直切片验证。 +""" + + +from fastapi import APIRouter, HTTPException, Request, Response, status + +from app.auth.service import ( + SESSION_COOKIE, + create_session, + revoke_session, +) +from app.db.session import DbSession +from app.models import PersonalDomain, User + +router = APIRouter(prefix="/v1/auth", tags=["auth"]) + + +@router.post("/sessions", status_code=201) +async def dev_login( + request: Request, response: Response, db: DbSession, body: dict[str, str] +) -> dict[str, str]: + """开发档登录：按 email 找到或创建用户与个人域，签发会话。""" + email = body.get("email", "").strip().lower() + if not email or "@" not in email: + raise HTTPException(status.HTTP_422_UNPROCESSABLE_ENTITY, "email 不合法") + from sqlalchemy import select + + existing = ( + (await db.execute(select(User).where(User.email == email))).scalar_one_or_none() + ) + if existing is None: + user = User(email=email, display_name=email.split("@")[0]) + db.add(user) + await db.flush() + domain = PersonalDomain(user_id=user.id, name="默认镜域") + db.add(domain) + else: + user = existing + _, token, csrf = await create_session(db, user.id) + await db.commit() + response.set_cookie( + SESSION_COOKIE, + token, + httponly=True, + secure=True, + samesite="lax", + path="/", + ) + return {"csrf_token": csrf, "user_id": user.id} + + +@router.delete("/sessions/current", status_code=204) +async def logout(request: Request, response: Response, db: DbSession) -> None: + token = request.cookies.get(SESSION_COOKIE) + if token is not None and await revoke_session(db, token): + await db.commit() + response.delete_cookie(SESSION_COOKIE, path="/") diff --git a/backend/app/auth/service.py b/backend/app/auth/service.py new file mode 100644 index 0000000..d626483 --- /dev/null +++ b/backend/app/auth/service.py @@ -0,0 +1,71 @@ +"""Web 会话认证：签发、校验与撤销（契约 api-conventions 第 2 节）。 + +- Cookie 名 ``__Host-hunjing_session``：HttpOnly、Secure、SameSite=Lax、Path=/。 +- Cookie 值为 CSPRNG opaque token；数据库只存 SHA-256 哈希（ADR-015）。 +- 状态变更请求要求 Origin 校验与 ``Hunjing-CSRF-Token`` 双提交。 +""" + +import hashlib +import secrets +from datetime import UTC, datetime, timedelta + +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.models import Session + +SESSION_COOKIE = "__Host-hunjing_session" +SESSION_TTL = timedelta(days=14) + + +def _hash_token(token: str) -> str: + return hashlib.sha256(token.encode("utf-8")).hexdigest() + + +def generate_session_token() -> str: + return secrets.token_urlsafe(32) + + +def generate_csrf_token() -> str: + return secrets.token_urlsafe(24) + + +async def create_session(db: AsyncSession, user_id: str) -> tuple[Session, str, str]: + """签发会话：返回（会话记录，opaque token，CSRF token）。""" + token = generate_session_token() + csrf = generate_csrf_token() + session = Session( + id=f"web_{secrets.token_urlsafe(12)}", + user_id=user_id, + token_hash=_hash_token(token), + csrf_token_hash=_hash_token(csrf), + expires_at=datetime.now(UTC) + SESSION_TTL, + ) + db.add(session) + return session, token, csrf + + +async def resolve_session(db: AsyncSession, token: str) -> Session | None: + """按 opaque token 查找有效（未过期、未撤销）会话。""" + result = await db.execute( + select(Session).where( + Session.token_hash == _hash_token(token), + Session.revoked_at.is_(None), + Session.expires_at > datetime.now(UTC), + ) + ) + return result.scalar_one_or_none() + + +async def revoke_session(db: AsyncSession, token: str) -> bool: + session = await resolve_session(db, token) + if session is None: + return False + session.revoked_at = datetime.now(UTC) + return True + + +def session_is_csrf_valid(session: Session, csrf_header: str | None) -> bool: + if csrf_header is None: + return False + return secrets.compare_digest(session.csrf_token_hash, _hash_token(csrf_header)) diff --git a/backend/app/conversations/__init__.py b/backend/app/conversations/__init__.py new file mode 100644 index 0000000..090d847 --- /dev/null +++ b/backend/app/conversations/__init__.py @@ -0,0 +1 @@ +"""对话域：schema、幂等原子事务服务与路由。""" diff --git a/backend/app/conversations/routes.py b/backend/app/conversations/routes.py new file mode 100644 index 0000000..56f3113 --- /dev/null +++ b/backend/app/conversations/routes.py @@ -0,0 +1,138 @@ +"""对话域路由：创建会话、发送消息（幂等原子）与 Run SSE。""" + +import json +from collections.abc import AsyncIterator +from typing import Annotated + +from fastapi import APIRouter, Depends, Header, HTTPException, Request, status +from fastapi.responses import JSONResponse, StreamingResponse +from sqlalchemy import select + +from app.auth.dependencies import get_current_user, require_state_change +from app.conversations import service +from app.conversations.schemas import MessageCreatedOut, MessageIn, MessageOut, RunOut +from app.db.session import DbSession +from app.models import PersonalDomain, Run, User + +router = APIRouter(prefix="/v1/conversations", tags=["conversations"]) + + +async def _domain_of(db: DbSession, user: User) -> PersonalDomain: + result = await db.execute( + select(PersonalDomain).where(PersonalDomain.user_id == user.id).limit(1) + ) + domain = result.scalar_one_or_none() + if domain is None: + raise HTTPException(status.HTTP_409_CONFLICT, "用户缺少个人域") + return domain + + +@router.post("", status_code=201) +async def create_conversation( + db: DbSession, + user: Annotated[User, Depends(require_state_change)], + request: Request, + body: dict[str, str], +) -> dict[str, str]: + domain = await _domain_of(db, user) + conversation = await service.create_conversation( + db, domain.id, body.get("title", "未命名对话"), getattr(request.state, "trace_id", None) + ) + return {"id": conversation.id, "title": conversation.title} + + +@router.post( + "/{conversation_id}/messages", + status_code=201, + response_model=None, +) +async def post_message( + conversation_id: str, + db: DbSession, + user: Annotated[User, Depends(require_state_change)], + request: Request, + body: MessageIn, + idempotency_key: Annotated[str, Header(alias="Idempotency-Key")], +) -> MessageCreatedOut | JSONResponse: + if not idempotency_key or len(idempotency_key) > 128: + raise HTTPException(status.HTTP_422_UNPROCESSABLE_ENTITY, "Idempotency-Key 不合法") + domain = await _domain_of(db, user) + message, run, replayed = await service.replay_or_execute_message( + db, + domain.id, + conversation_id, + idempotency_key, + body, + getattr(request.state, "trace_id", None), + ) + payload = MessageCreatedOut( + message=MessageOut( + id=message.id, + conversation_id=message.conversation_id, + branch_id=message.branch_id, + role=message.role, + content=message.content, + created_at=message.created_at.isoformat(), + ), + run=( + RunOut( + id=run.id, + conversation_id=run.conversation_id, + status=run.status, + created_at=run.created_at.isoformat(), + completed_at=run.completed_at.isoformat() if run.completed_at else None, + ) + if run is not None + else None + ), + ) + if replayed: + # 重放成功按契约返回 200 与原资源（避开装饰器固定 201） + return JSONResponse(status_code=200, content=json.loads(payload.model_dump_json())) + return payload + + +run_router = APIRouter(prefix="/v1/runs", tags=["runs"]) + + +@run_router.get("/{run_id}/events") +async def run_events( + run_id: str, + db: DbSession, + user: Annotated[User, Depends(get_current_user)], +) -> StreamingResponse: + """Run SSE：返回当前快照与终止状态（垂直切片空状态闭环）。""" + result = await db.execute(select(Run).where(Run.id == run_id)) + run = result.scalar_one_or_none() + if run is None: + raise HTTPException(status.HTTP_404_NOT_FOUND, "Run 不存在") + domain = await _domain_of(db, user) + if run.personal_domain_id != domain.id: + raise HTTPException(status.HTTP_404_NOT_FOUND, "Run 不存在") + + terminal = run.status in {"completed", "failed", "cancelled"} + + async def event_stream() -> AsyncIterator[str]: + def envelope(event_type: str, data: dict[str, object], sequence: int) -> str: + payload = json.dumps( + {"type": event_type, "data": data, "sequence": sequence}, ensure_ascii=False + ) + return f"data: {payload}\n\n" + + yield envelope("run.status.changed", {"run_id": run.id, "status": run.status}, 0) + if terminal: + yield envelope( + "message.completed", + { + "run_id": run.id, + "message_ids": [], + "final_status": run.status, + }, + 1, + ) + + return StreamingResponse( + event_stream(), + media_type="text/event-stream", + headers={"Cache-Control": "no-store"}, + ) diff --git a/backend/app/conversations/schemas.py b/backend/app/conversations/schemas.py new file mode 100644 index 0000000..5f5979d --- /dev/null +++ b/backend/app/conversations/schemas.py @@ -0,0 +1,60 @@ +"""对话域 schema：ContentBlock 判别联合（垂直切片子集）。 + +完整 21 类型随 WP-03 落地；本切片登记 text/markdown 并拒绝未知类型， +契约真源为 openapi/openapi.yaml。 +""" + +from typing import Annotated, Literal + +from pydantic import BaseModel, ConfigDict, Field + + +class BlockBase(BaseModel): + model_config = ConfigDict(extra="forbid") + + id: str | None = None # 创建请求可省略，由服务端生成 + status: Literal["pending", "streaming", "completed", "failed", "cancelled"] | None = None + + +class TextBlock(BlockBase): + type: Literal["text"] + text: str + language: str | None = None + + +class MarkdownBlock(BlockBase): + type: Literal["markdown"] + text: str + flavor: Literal["hunjing-gfm-v1"] = "hunjing-gfm-v1" + + +ContentBlock = Annotated[TextBlock | MarkdownBlock, Field(discriminator="type")] + + +class MessageIn(BaseModel): + model_config = ConfigDict(extra="forbid") + + content: list[ContentBlock] = Field(min_length=1, max_length=64) + start_run: bool = True + + +class MessageOut(BaseModel): + id: str + conversation_id: str + branch_id: str + role: str + content: list[dict[str, object]] + created_at: str + + +class RunOut(BaseModel): + id: str + conversation_id: str | None + status: str + created_at: str + completed_at: str | None = None + + +class MessageCreatedOut(BaseModel): + message: MessageOut + run: RunOut | None diff --git a/backend/app/conversations/service.py b/backend/app/conversations/service.py new file mode 100644 index 0000000..cd3c23e --- /dev/null +++ b/backend/app/conversations/service.py @@ -0,0 +1,182 @@ +"""对话域服务：幂等、原子事务与越权防护。 + +不变量（契约 4.3/4.4 节）：Message 与可选 Run 在同一事务创建；权威状态、 +AuditEvent 与 OutboxEvent 同事务提交；跨域对象按 404 fail-closed。 +""" + +import hashlib +import json +from datetime import UTC, datetime +from typing import Any + +from fastapi import HTTPException, status +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.conversations.schemas import MessageIn +from app.core.ids import new_id +from app.models import ( + AuditEvent, + Branch, + Conversation, + IdempotencyRecord, + Message, + OutboxEvent, + Run, +) + +MESSAGE_ROUTE = "/v1/conversations/{conversation_id}/messages" + + +def _request_hash(body: MessageIn) -> str: + canonical = json.dumps( + { + "content": [b.model_dump(exclude_none=True) for b in body.content], + "start_run": body.start_run, + }, + sort_keys=True, + ensure_ascii=False, + ) + return hashlib.sha256(canonical.encode("utf-8")).hexdigest() + + +async def _require_conversation( + db: AsyncSession, domain_id: str, conversation_id: str +) -> Conversation: + conversation = await db.get(Conversation, conversation_id) + # 跨域对象统一 404，不泄露存在性（契约第 7 节） + if conversation is None or conversation.personal_domain_id != domain_id: + raise HTTPException(status.HTTP_404_NOT_FOUND, "对话不存在") + return conversation + + +async def _get_default_branch(db: AsyncSession, conversation_id: str) -> Branch: + result = await db.execute( + select(Branch).where(Branch.conversation_id == conversation_id).limit(1) + ) + branch = result.scalar_one_or_none() + assert branch is not None, "会话必须存在默认分支（创建事务保证）" + return branch + + +async def create_conversation( + db: AsyncSession, domain_id: str, title: str, trace_id: str | None +) -> Conversation: + conversation = Conversation(personal_domain_id=domain_id, title=title) + db.add(conversation) + await db.flush() + branch = Branch(conversation_id=conversation.id, title="主脉") + db.add(branch) + db.add( + AuditEvent( + actor_type="user", + actor_id=domain_id, + action="conversation.created", + aggregate_type="conversation", + aggregate_id=conversation.id, + trace_id=trace_id, + ) + ) + db.add( + OutboxEvent( + aggregate_type="conversation", + aggregate_id=conversation.id, + event_type="conversation.created", + payload={"conversation_id": conversation.id, "title": title}, + ) + ) + await db.flush() + return conversation + + +async def replay_or_execute_message( + db: AsyncSession, + domain_id: str, + conversation_id: str, + idempotency_key: str, + body: MessageIn, + trace_id: str | None, +) -> tuple[Message, Run | None, bool]: + """返回（消息，Run，是否为重放）。幂等：相同键+相同请求返回原资源。""" + route = MESSAGE_ROUTE + existing = await db.execute( + select(IdempotencyRecord).where( + IdempotencyRecord.owner_domain_id == domain_id, + IdempotencyRecord.route == route, + IdempotencyRecord.key == idempotency_key, + ) + ) + record = existing.scalar_one_or_none() + request_hash = _request_hash(body) + if record is not None: + if record.request_hash != request_hash: + raise HTTPException(status.HTTP_409_CONFLICT, "相同幂等键对应不同请求体") + message = await db.get(Message, record.resource_ids.get("message_id")) + run = None + run_id = record.resource_ids.get("run_id") + if run_id is not None: + run = await db.get(Run, run_id) + assert message is not None + return message, run, True + + conversation = await _require_conversation(db, domain_id, conversation_id) + branch = await _get_default_branch(db, conversation.id) + + blocks: list[dict[str, Any]] = [ + {**b.model_dump(exclude_none=True), "id": new_id("block_"), "status": "completed"} + for b in body.content + ] + message = Message( + conversation_id=conversation.id, + branch_id=branch.id, + role="user", + content=blocks, + ) + db.add(message) + await db.flush() + + run = None + if body.start_run: + # 垂直切片空状态闭环：不调用模型，Run 原子创建并直接进入终态 completed。 + run = Run( + personal_domain_id=domain_id, + conversation_id=conversation.id, + status="completed", + completed_at=datetime.now(UTC), + ) + db.add(run) + await db.flush() + + resource_ids: dict[str, Any] = {"message_id": message.id} + if run is not None: + resource_ids["run_id"] = run.id + db.add( + AuditEvent( + actor_type="user", + actor_id=domain_id, + action="message.created", + aggregate_type="message", + aggregate_id=message.id, + trace_id=trace_id, + ) + ) + db.add( + OutboxEvent( + aggregate_type="message", + aggregate_id=message.id, + event_type="message.created", + payload={"message_id": message.id, "conversation_id": conversation.id}, + ) + ) + db.add( + IdempotencyRecord( + key=idempotency_key, + owner_domain_id=domain_id, + route=route, + request_hash=request_hash, + response_status=201, + resource_ids=resource_ids, + ) + ) + await db.flush() + return message, run, False diff --git a/backend/app/core/middleware.py b/backend/app/core/middleware.py new file mode 100644 index 0000000..8c716e6 --- /dev/null +++ b/backend/app/core/middleware.py @@ -0,0 +1,80 @@ +"""请求上下文中间件与结构化日志。 + +- `Hunjing-Request-ID`：合法则回显，缺失由服务端生成（UUIDv4，request_ 前缀）。 +- `traceparent`：透传 W3C Trace Context；缺失生成新 trace-id（不与业务请求 ID 混用）。 +- JSON 结构化日志：Authorization/Cookie/Set-Cookie 等敏感头脱敏。 +""" + +import json +import logging +import re +import uuid +from typing import Any + +from starlette.middleware.base import BaseHTTPMiddleware +from starlette.requests import Request + +from app.core.ids import new_id + +REDACTED_HEADERS = {"authorization", "cookie", "set-cookie", "hunjing-csrf-token", "x-api-key"} +_TRACEPARENT_RE = re.compile(r"^[0-9a-f]{2}-[0-9a-f]{32}-[0-9a-f]{16}-[0-9a-f]{2}$") + + +def _sanitize_headers(headers: Any) -> dict[str, str]: + return { + k: ("[REDACTED]" if k.lower() in REDACTED_HEADERS else v) for k, v in headers.items() + } + + +class JsonFormatter(logging.Formatter): + def format(self, record: logging.LogRecord) -> str: + payload: dict[str, Any] = { + "ts": self.formatTime(record, "%Y-%m-%dT%H:%M:%S%z"), + "level": record.levelname, + "logger": record.name, + "message": record.getMessage(), + } + for key in ("request_id", "trace_id", "method", "path", "status"): + value = getattr(record, key, None) + if value is not None: + payload[key] = value + if record.exc_info: + payload["exception"] = self.formatException(record.exc_info) + return json.dumps(payload, ensure_ascii=False) + + +def configure_logging(level: str = "info") -> None: + handler = logging.StreamHandler() + handler.setFormatter(JsonFormatter()) + root = logging.getLogger() + root.handlers = [handler] + root.setLevel(level.upper()) + + +class RequestContextMiddleware(BaseHTTPMiddleware): + async def dispatch(self, request: Request, call_next: Any) -> Any: + request_id = request.headers.get("Hunjing-Request-ID") or new_id("request_", version=4) + traceparent = request.headers.get("traceparent") + if traceparent is None or not _TRACEPARENT_RE.match(traceparent): + traceparent = f"00-{uuid.uuid4().hex}-{uuid.uuid4().hex[:16]}-01" + trace_id = traceparent.split("-")[1] + request.state.request_id = request_id + request.state.trace_id = trace_id + + response = await call_next(request) + response.headers["Hunjing-Request-ID"] = request_id + response.headers["traceparent"] = traceparent + logging.getLogger("hunjing.access").info( + "request", + extra={ + "request_id": request_id, + "trace_id": trace_id, + "method": request.method, + "path": request.url.path, + "status": response.status_code, + }, + ) + return response + + +__all__ = ["RequestContextMiddleware", "configure_logging", "_sanitize_headers"] diff --git a/backend/app/db/session.py b/backend/app/db/session.py new file mode 100644 index 0000000..76d93d8 --- /dev/null +++ b/backend/app/db/session.py @@ -0,0 +1,37 @@ +"""请求级数据库会话依赖与引擎生命周期。 + +引擎由应用实例持有（create_app 时创建）：一个应用一个引擎与连接池， +避免模块级引擎跨事件循环复用连接。 +""" + +from collections.abc import AsyncIterator +from typing import Annotated + +from fastapi import Depends, FastAPI, Request +from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker + +from app.db.base import create_engine, create_session_factory + + +def init_engine(app: FastAPI) -> async_sessionmaker[AsyncSession]: + """在 app.state 上初始化引擎与会话工厂（应用启动时调用一次）。""" + engine: AsyncEngine = create_engine() + app.state.engine = engine + factory = create_session_factory(engine) + app.state.session_factory = factory + return factory + + +async def get_db(request: Request) -> AsyncIterator[AsyncSession]: + """每请求一个会话：成功提交、异常回滚。""" + factory: async_sessionmaker[AsyncSession] = request.app.state.session_factory + async with factory() as session: + try: + yield session + await session.commit() + except Exception: + await session.rollback() + raise + + +DbSession = Annotated[AsyncSession, Depends(get_db)] diff --git a/backend/app/main.py b/backend/app/main.py index f3db062..2e7e09a 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -2,12 +2,18 @@ from fastapi import FastAPI +from app.auth.routes import router as auth_router +from app.conversations.routes import router as conversations_router +from app.conversations.routes import run_router from app.core.config import get_settings from app.core.errors import install_error_handlers +from app.core.middleware import RequestContextMiddleware, configure_logging +from app.db.session import init_engine def create_app() -> FastAPI: settings = get_settings() + configure_logging(settings.log_level) app = FastAPI( title="Hunjing API", version="0.1.0", @@ -16,6 +22,11 @@ def create_app() -> FastAPI: openapi_url="/_dev/openapi.json", ) install_error_handlers(app) + init_engine(app) + app.add_middleware(RequestContextMiddleware) + app.include_router(auth_router) + app.include_router(conversations_router) + app.include_router(run_router) @app.get("/healthz", tags=["ops"]) async def healthz() -> dict[str, str]: diff --git a/backend/app/models.py b/backend/app/models.py index e586599..8597087 100644 --- a/backend/app/models.py +++ b/backend/app/models.py @@ -56,6 +56,8 @@ class Session(Base): user_id: Mapped[str] = mapped_column( ForeignKey("user.id", ondelete="cascade"), nullable=False, index=True ) + token_hash: Mapped[str] = mapped_column(String, nullable=False, unique=True) + csrf_token_hash: Mapped[str] = mapped_column(String, nullable=False) expires_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, index=True ) @@ -138,3 +140,111 @@ class OutboxEvent(Base): Index("ix_outbox_event_status_occurred", "status", "occurred_at"), Index("ix_outbox_event_aggregate", "aggregate_type", "aggregate_id"), ) + + +class Conversation(Base): + """对话容器：归属个人域（垂直切片 WP-03 最小集）。""" + + __tablename__ = "conversation" + + id: Mapped[str] = mapped_column( + String, primary_key=True, default=lambda: new_id("conversation_") + ) + personal_domain_id: Mapped[str] = mapped_column( + ForeignKey("personal_domain.id", ondelete="cascade"), nullable=False, index=True + ) + title: Mapped[str] = mapped_column(String, nullable=False, default="未命名对话") + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, server_default=func.now() + ) + updated_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now() + ) + + +class Branch(Base): + """对话分支：创建会话时生成默认主分支。""" + + __tablename__ = "branch" + + id: Mapped[str] = mapped_column(String, primary_key=True, default=lambda: new_id("branch_")) + conversation_id: Mapped[str] = mapped_column( + ForeignKey("conversation.id", ondelete="cascade"), nullable=False, index=True + ) + title: Mapped[str] = mapped_column(String, nullable=False, default="主脉") + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, server_default=func.now() + ) + + +class Message(Base): + """消息：content 为 canonical ContentBlock JSON（本切片为 Text/Markdown 子集）。""" + + __tablename__ = "message" + + id: Mapped[str] = mapped_column(String, primary_key=True, default=lambda: new_id("message_")) + conversation_id: Mapped[str] = mapped_column( + ForeignKey("conversation.id", ondelete="cascade"), nullable=False, index=True + ) + branch_id: Mapped[str] = mapped_column( + ForeignKey("branch.id", ondelete="cascade"), nullable=False, index=True + ) + role: Mapped[str] = mapped_column(String, nullable=False) + content: Mapped[list[dict[str, Any]]] = mapped_column(JSONB, nullable=False) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, server_default=func.now() + ) + + __table_args__ = ( + CheckConstraint("role IN ('user', 'assistant', 'system', 'tool')", name="role_values"), + ) + + +class Run(Base): + """Agent 执行：状态机见 Agent 运行时第 3 节；本切片空转即完成。""" + + __tablename__ = "run" + + id: Mapped[str] = mapped_column(String, primary_key=True, default=lambda: new_id("run_")) + personal_domain_id: Mapped[str] = mapped_column( + ForeignKey("personal_domain.id", ondelete="cascade"), nullable=False, index=True + ) + conversation_id: Mapped[str | None] = mapped_column( + ForeignKey("conversation.id", ondelete="cascade"), nullable=True, index=True + ) + status: Mapped[str] = mapped_column(String, nullable=False, default="queued") + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, server_default=func.now() + ) + completed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + + __table_args__ = ( + CheckConstraint( + "status IN ('queued', 'running', 'waiting_approval', " + "'completed', 'failed', 'cancelled')", + name="status_values", + ), + Index("ix_run_domain_status", "personal_domain_id", "status"), + ) + + +class IdempotencyRecord(Base): + """幂等键记录：按 认证主体+路由+键 隔离，保存请求哈希与结果引用（契约第 6 节）。""" + + __tablename__ = "idempotency_record" + + key: Mapped[str] = mapped_column(String, primary_key=True) + owner_domain_id: Mapped[str] = mapped_column( + ForeignKey("personal_domain.id", ondelete="cascade"), nullable=False + ) + route: Mapped[str] = mapped_column(String, nullable=False) + request_hash: Mapped[str] = mapped_column(String, nullable=False) + response_status: Mapped[int] = mapped_column(nullable=False) + resource_ids: Mapped[dict[str, Any]] = mapped_column(JSONB, nullable=False, default=dict) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, server_default=func.now() + ) + + __table_args__ = ( + Index("uq_idempotency_owner_route_key", "owner_domain_id", "route", "key", unique=True), + ) diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py index 130cd6a..02c9b2b 100644 --- a/backend/tests/conftest.py +++ b/backend/tests/conftest.py @@ -24,3 +24,35 @@ async def db() -> async_sessionmaker[AsyncSession]: factory = async_sessionmaker(engine, expire_on_commit=False) yield factory await engine.dispose() + + +@pytest.fixture +async def client(db): + """连接测试库的应用客户端（覆盖 get_db）。""" + from httpx import ASGITransport, AsyncClient + + from app.db import session as db_session + from app.main import create_app + + app = create_app() + + async def override(): + async with db() as session: + try: + yield session + await session.commit() + except Exception: + await session.rollback() + raise + + app.dependency_overrides[db_session.get_db] = override + transport = ASGITransport(app=app, raise_app_exceptions=False) + async with AsyncClient(transport=transport, base_url="https://test") as c: + yield c + + +async def login(client, email: str) -> dict[str, str]: + """开发档登录：返回需附加的请求头（Cookie 由客户端自动携带）。""" + resp = await client.post("/v1/auth/sessions", json={"email": email}) + assert resp.status_code == 201, resp.text + return {"Hunjing-CSRF-Token": resp.json()["csrf_token"]} diff --git a/backend/tests/test_conversations.py b/backend/tests/test_conversations.py new file mode 100644 index 0000000..3ddbf1f --- /dev/null +++ b/backend/tests/test_conversations.py @@ -0,0 +1,148 @@ +"""垂直切片测试：幂等原子事务、IDOR fail-closed、CSRF、SSE 与回滚。""" + +import json + +from httpx import AsyncClient +from sqlalchemy import select + +from app import models +from tests.conftest import login + +MESSAGE_BODY = {"content": [{"type": "text", "text": "你好，魂镜"}], "start_run": True} + + +async def _create_conversation(client: AsyncClient, headers: dict) -> str: + resp = await client.post( + "/v1/conversations", json={"title": "切片"}, headers=headers + ) + assert resp.status_code == 201, resp.text + return resp.json()["id"] + + +async def _post_message(client: AsyncClient, headers: dict, cid: str, key: str) -> object: + return await client.post( + f"/v1/conversations/{cid}/messages", + json=MESSAGE_BODY, + headers={**headers, "Idempotency-Key": key}, + ) + + +class TestMessagePipeline: + async def test_create_and_post_message(self, client: AsyncClient) -> None: + headers = await login(client, "alice@example.com") + cid = await _create_conversation(client, headers) + resp = await _post_message(client, headers, cid, "idem-1") + assert resp.status_code == 201, resp.text + body = resp.json() + assert body["message"]["id"].startswith("message_") + assert body["message"]["content"][0]["id"].startswith("block_") + assert body["run"]["status"] == "completed" # 空状态闭环 + + async def test_idempotent_replay_returns_original(self, client: AsyncClient) -> None: + headers = await login(client, "bob@example.com") + cid = await _create_conversation(client, headers) + first = await _post_message(client, headers, cid, "idem-replay") + assert first.status_code == 201 + second = await _post_message(client, headers, cid, "idem-replay") + assert second.status_code == 200 # 重放返回原资源 + assert second.json()["message"]["id"] == first.json()["message"]["id"] + + async def test_same_key_different_body_conflicts(self, client: AsyncClient) -> None: + headers = await login(client, "carol@example.com") + cid = await _create_conversation(client, headers) + await _post_message(client, headers, cid, "idem-conflict") + resp = await client.post( + f"/v1/conversations/{cid}/messages", + json={"content": [{"type": "text", "text": "不同内容"}], "start_run": True}, + headers={**headers, "Idempotency-Key": "idem-conflict"}, + ) + assert resp.status_code == 409 + + async def test_missing_idempotency_key_rejected(self, client: AsyncClient) -> None: + headers = await login(client, "dave@example.com") + cid = await _create_conversation(client, headers) + resp = await client.post( + f"/v1/conversations/{cid}/messages", json=MESSAGE_BODY, headers=headers + ) + assert resp.status_code == 422 + + async def test_unknown_block_type_rejected(self, client: AsyncClient) -> None: + headers = await login(client, "erin@example.com") + cid = await _create_conversation(client, headers) + resp = await client.post( + f"/v1/conversations/{cid}/messages", + json={"content": [{"type": "image", "file_id": "x", "alt": "y"}]}, + headers={**headers, "Idempotency-Key": "idem-bad"}, + ) + assert resp.status_code == 422 + + async def test_cross_domain_conversation_returns_404(self, client: AsyncClient) -> None: + alice = await login(client, "ida@example.com") + cid = await _create_conversation(client, alice) + mallory = await login(client, "mallory@example.com") + resp = await _post_message(client, mallory, cid, "idem-idor") + assert resp.status_code == 404 # fail-closed，不泄露存在性 + + async def test_csrf_token_required_for_state_change(self, client: AsyncClient) -> None: + await login(client, "frank@example.com") + resp = await client.post("/v1/conversations", json={"title": "无 CSRF"}) + assert resp.status_code == 403 + + async def test_unauthenticated_rejected(self, client: AsyncClient) -> None: + resp = await client.post("/v1/conversations", json={"title": "匿名"}) + assert resp.status_code == 401 + + +class TestRunSse: + async def test_sse_snapshot_and_terminal(self, client: AsyncClient) -> None: + headers = await login(client, "grace@example.com") + cid = await _create_conversation(client, headers) + resp = await _post_message(client, headers, cid, "idem-sse") + run_id = resp.json()["run"]["id"] + sse = await client.get(f"/v1/runs/{run_id}/events") + assert sse.status_code == 200 + assert sse.headers["content-type"].startswith("text/event-stream") + lines = [ln for ln in sse.text.splitlines() if ln.startswith("data: ")] + events = [json.loads(ln.removeprefix("data: ")) for ln in lines] + assert events[0]["type"] == "run.status.changed" + assert events[0]["data"]["status"] == "completed" + assert events[-1]["type"] == "message.completed" + + async def test_cross_domain_run_returns_404(self, client: AsyncClient) -> None: + headers = await login(client, "heidi@example.com") + cid = await _create_conversation(client, headers) + resp = await _post_message(client, headers, cid, "idem-sse2") + run_id = resp.json()["run"]["id"] + await login(client, "eve@example.com") + sse = await client.get(f"/v1/runs/{run_id}/events") + assert sse.status_code == 404 + + +class TestAtomicity: + async def test_failure_leaves_no_orphans(self, client: AsyncClient, db) -> None: + from app.conversations import service + + headers = await login(client, "ivan@example.com") + cid = await _create_conversation(client, headers) + + class BoomRun: + def __init__(self, *args, **kwargs): + raise RuntimeError("注入失败：Run 写入阶段崩溃") + + original = service.Run + service.Run = BoomRun # type: ignore[assignment] + try: + resp = await _post_message(client, headers, cid, "idem-atomic") + assert resp.status_code == 500 + finally: + service.Run = original # type: ignore[assignment] + + async with db() as session: + # 会话创建的合法审计/Outbox 保留；消息相关记录必须为零 + assert (await session.scalars(select(models.Message))).all() == [] + assert (await session.scalars(select(models.Run))).all() == [] + audits = (await session.scalars(select(models.AuditEvent))).all() + assert [a for a in audits if a.aggregate_type == "message"] == [] + outbox = (await session.scalars(select(models.OutboxEvent))).all() + assert [e for e in outbox if e.aggregate_type == "message"] == [] + assert (await session.scalars(select(models.IdempotencyRecord))).all() == [] diff --git a/backend/tests/test_models.py b/backend/tests/test_models.py index ce31fa7..73cd41b 100644 --- a/backend/tests/test_models.py +++ b/backend/tests/test_models.py @@ -65,6 +65,8 @@ async def test_session_cascades_with_user(db: async_sessionmaker[AsyncSession]) Session( id="opaque-session-token", user_id=user.id, + token_hash="sha256:opaque-session-token", + csrf_token_hash="sha256:csrf", expires_at=datetime.now(UTC) + timedelta(days=7), ) )