diff --git a/CHANGELOG.md b/CHANGELOG.md index 911ca3b2e5..e659d4973d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2792,6 +2792,13 @@ This release closes that milestone with **765 merged pull requests**. ### Security +- **auth:** `POST /api/v1/auth/initialize` no longer lets two concurrent + first-boot requests both create an admin. The handler counted admins in one + session and created the account in another, so two requests with different + emails both saw an empty system; the loser now gets the documented + `409 system_already_initialized`. The count and the insert share one + transaction with writers serialized first (SQLite `BEGIN IMMEDIATE`, + PostgreSQL advisory lock). ([#5776]) - **uploads:** Document conversion no longer re-opens the upload by name. The Gateway converted the committed file and the embedded client converted the copy it had just placed in the thread's uploads directory, so a sandbox that @@ -4371,3 +4378,4 @@ with **180 merged pull requests** since the first 2.0 milestone tag. [#5611]: https://github.com/bytedance/deer-flow/pull/5611 [#5673]: https://github.com/bytedance/deer-flow/pull/5673 [#5734]: https://github.com/bytedance/deer-flow/pull/5734 +[#5776]: https://github.com/bytedance/deer-flow/pull/5776 diff --git a/CHANGELOG_zh.md b/CHANGELOG_zh.md index 15d69aaf72..04a11f86d5 100644 --- a/CHANGELOG_zh.md +++ b/CHANGELOG_zh.md @@ -2095,6 +2095,11 @@ ### 安全 +- **认证:** `POST /api/v1/auth/initialize` 不再让两个并发的首次初始化请求都创建 admin。 + 此前处理器在一个会话中统计 admin 数量、在另一个会话中创建账号,因此两个使用不同邮箱的请求 + 会同时看到空系统;现在失败方会返回文档所述的 `409 system_already_initialized`。统计与插入现在 + 在同一事务内完成,并先对写入串行化(SQLite 用 `BEGIN IMMEDIATE`,PostgreSQL 用 + advisory lock)。([#5776]) - **上传:** 文档转换不再按文件名重新打开上传文件。此前 Gateway 转换的是已提交的文件,嵌入式 客户端转换的是刚放入线程 uploads 目录的副本,因此沙箱若在此期间把该文件名替换为符号链接, 宿主文件的内容就会被转换成该线程的 `.md` 配套文件。现在 Gateway 通过自己写入时持有的文件 @@ -3549,3 +3554,4 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写,核心包含子 [#5611]: https://github.com/bytedance/deer-flow/pull/5611 [#5673]: https://github.com/bytedance/deer-flow/pull/5673 [#5734]: https://github.com/bytedance/deer-flow/pull/5734 +[#5776]: https://github.com/bytedance/deer-flow/pull/5776 diff --git a/backend/app/gateway/auth/local_provider.py b/backend/app/gateway/auth/local_provider.py index 11a7d8e3f8..be1f6c6897 100644 --- a/backend/app/gateway/auth/local_provider.py +++ b/backend/app/gateway/auth/local_provider.py @@ -83,6 +83,23 @@ class LocalAuthProvider(AuthProvider): ) return await self._repo.create_user(user) + async def create_first_admin(self, email: str, password: str) -> User | None: + """Create the first admin account, or return None if one already exists. + + The check and the insert are one atomic claim in the repository, so + concurrent first-boot requests cannot both succeed. + + Args: + email: Admin email address + password: Plain text password (will be hashed) + + Returns: + Created admin User, or None if the system already has an admin + """ + password_hash = await hash_password_async(password) + user = User(email=email, password_hash=password_hash, system_role="admin", needs_setup=False) + return await self._repo.create_first_admin(user) + async def get_user_by_oauth(self, provider: str, oauth_id: str) -> User | None: """Get user by OAuth provider and ID.""" return await self._repo.get_user_by_oauth(provider, oauth_id) diff --git a/backend/app/gateway/auth/repositories/base.py b/backend/app/gateway/auth/repositories/base.py index c34751a32b..cb66ced733 100644 --- a/backend/app/gateway/auth/repositories/base.py +++ b/backend/app/gateway/auth/repositories/base.py @@ -40,6 +40,22 @@ class UserRepository(ABC): """ raise NotImplementedError + @abstractmethod + async def create_first_admin(self, user: User) -> User | None: + """Create *user* as the first admin, atomically. + + Implementations must read the admin count and insert in one + serialized transaction: a count-then-create pair lets two concurrent + first-boot requests both find an empty system. + + Returns: + The created User, or None if an admin already exists. + + Raises: + ValueError: If email already exists + """ + raise NotImplementedError + @abstractmethod async def get_user_by_id(self, user_id: str) -> User | None: """Get user by ID. diff --git a/backend/app/gateway/auth/repositories/sqlite.py b/backend/app/gateway/auth/repositories/sqlite.py index 1de9f2195f..c8f50941e7 100644 --- a/backend/app/gateway/auth/repositories/sqlite.py +++ b/backend/app/gateway/auth/repositories/sqlite.py @@ -12,10 +12,11 @@ construct this after ``init_engine_from_config()`` has run. from __future__ import annotations +import hashlib from datetime import UTC from uuid import UUID -from sqlalchemy import func, select +from sqlalchemy import func, select, text from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker @@ -132,6 +133,11 @@ def _normalize_email(email: str) -> str: return email.lower() +# Fixed 63-bit key for pg_advisory_xact_lock(bigint); scopes the first-admin +# claim without colliding with other advisory-lock users in this database. +_FIRST_ADMIN_LOCK_KEY = int.from_bytes(hashlib.sha256(b"deerflow:auth:first-admin-claim").digest()[:8], "big") & 0x7FFFFFFFFFFFFFFF + + class SQLiteUserRepository(UserRepository): """Async user repository backed by the shared SQLAlchemy engine.""" @@ -183,43 +189,95 @@ class SQLiteUserRepository(UserRepository): unique constraint enforces case-insensitive uniqueness for new rows and the returned ``User`` reflects the stored form. """ + async with self._sf() as session: + await self._insert_user(session, user) + await session.commit() + return user + + async def _insert_user(self, session: AsyncSession, user: User) -> User: + """Pre-check the email and flush *user*; the caller owns the transaction. + + Shared by :meth:`create_user` and :meth:`create_first_admin` so both + report the same uniqueness conflicts as ``ValueError``. + """ user.email = _normalize_email(user.email) row = self._user_to_row(user) + # The unique constraint is case-sensitive, so it cannot catch a + # canonical address colliding with a mixed-case legacy row. + existing = select(UserRow.id).where(func.lower(UserRow.email) == user.email).limit(1) + if await session.scalar(existing) is not None: + raise ValueError(f"Email already registered: {user.email}") + session.add(row) + try: + await session.flush() + except IntegrityError as exc: + await session.rollback() + # The email pre-check above already ruled out an email + # collision under normal (non-racing) conditions, so + # IntegrityErrors reaching here are usually + # idx_users_oauth_identity -- but not always (a duplicate + # primary key, or an email collision that raced past the + # pre-check). Attribute the failure to the constraint that + # actually fired instead of assuming any one of them. + if _is_oauth_identity_violation(exc): + raise ValueError(f"OAuth account already linked: {user.oauth_provider}/{user.oauth_id}") from exc + if _is_email_violation(exc): + # A duplicate address that got past the pre-check: a + # concurrent insert of the same email. + raise ValueError(f"Email already registered: {user.email}") from exc + if _is_uniqueness_violation(exc): + # Some other unique index / primary key (in practice a + # duplicate id). "Already exists" fits, but don't dress it + # up as an email conflict for an address that isn't + # registered. + constraint = _violated_constraint(exc) + raise ValueError(f"User already exists (constraint: {constraint})" if constraint else "User already exists") from exc + # A NOT NULL / CHECK / foreign-key IntegrityError is not a + # "user already exists" condition and not part of this + # method's ValueError contract -- let it propagate. + raise + return user + + @staticmethod + async def _serialize_first_admin_claim(session: AsyncSession) -> None: + """Serialize concurrent first-admin claims before the count is read. + + SQLite takes the database write lock up front (``BEGIN IMMEDIATE``), + the idiom the project repositories use for read-then-write + transactions. Postgres has no row to lock — the table is empty on a + first boot — so it takes a transaction-scoped advisory lock on a + fixed key, the way the channel OAuth scope cap does. + + A dialect with neither strategy raises: this is the point that makes + :meth:`create_first_admin` atomic, and falling through would leave a + plain check-then-act that lets two first-boot requests both create an + admin. The engine builds only these two dialects today, so the raise + is a guard for a future backend, not a reachable path. + """ + dialect = session.get_bind().dialect.name + if dialect == "sqlite": + await session.execute(text("BEGIN IMMEDIATE")) + elif dialect == "postgresql": + await session.execute(text("SELECT pg_advisory_xact_lock(:lock_key)"), {"lock_key": _FIRST_ADMIN_LOCK_KEY}) + else: + raise RuntimeError(f"Cannot serialize the first-admin claim: no locking strategy for SQL dialect {dialect!r}") + + async def create_first_admin(self, user: User) -> User | None: + """Insert *user* as the first admin, or return None if one already exists. + + The admin count and the insert share one transaction and writers are + serialized first, so two concurrent first-boot requests cannot both + read an empty system and both create an admin. ``None`` means the + claim was lost — the caller reports "already initialized" — and the + uniqueness conflicts of :meth:`create_user` still raise ``ValueError``. + """ async with self._sf() as session: - # The unique constraint is case-sensitive, so it cannot catch a - # canonical address colliding with a mixed-case legacy row. - existing = select(UserRow.id).where(func.lower(UserRow.email) == user.email).limit(1) - if await session.scalar(existing) is not None: - raise ValueError(f"Email already registered: {user.email}") - session.add(row) - try: - await session.commit() - except IntegrityError as exc: - await session.rollback() - # The email pre-check above already ruled out an email - # collision under normal (non-racing) conditions, so - # IntegrityErrors reaching here are usually - # idx_users_oauth_identity -- but not always (a duplicate - # primary key, or an email collision that raced past the - # pre-check). Attribute the failure to the constraint that - # actually fired instead of assuming any one of them. - if _is_oauth_identity_violation(exc): - raise ValueError(f"OAuth account already linked: {user.oauth_provider}/{user.oauth_id}") from exc - if _is_email_violation(exc): - # A duplicate address that got past the pre-check: a - # concurrent insert of the same email. - raise ValueError(f"Email already registered: {user.email}") from exc - if _is_uniqueness_violation(exc): - # Some other unique index / primary key (in practice a - # duplicate id). "Already exists" fits, but don't dress it - # up as an email conflict for an address that isn't - # registered. - constraint = _violated_constraint(exc) - raise ValueError(f"User already exists (constraint: {constraint})" if constraint else "User already exists") from exc - # A NOT NULL / CHECK / foreign-key IntegrityError is not a - # "user already exists" condition and not part of this - # method's ValueError contract -- let it propagate. - raise + await self._serialize_first_admin_claim(session) + admin_count = await session.scalar(select(func.count()).select_from(UserRow).where(UserRow.system_role == "admin")) + if admin_count: + return None + await self._insert_user(session, user) + await session.commit() return user async def get_user_by_id(self, user_id: str) -> User | None: diff --git a/backend/app/gateway/routers/auth.py b/backend/app/gateway/routers/auth.py index 06c9e41fe0..23be4a0168 100644 --- a/backend/app/gateway/routers/auth.py +++ b/backend/app/gateway/routers/auth.py @@ -790,32 +790,35 @@ async def initialize_admin(request: Request, response: Response, body: Initializ """Create the first admin account on initial system setup. Only callable when no admin exists. Returns 409 Conflict if an admin - already exists. + already exists, including when a concurrent first-boot request won the + claim: the account is created through an atomic check-and-insert rather + than a count followed by a create. On success, the admin account is created with ``needs_setup=False`` and the session cookie is set. """ - admin_count = await get_local_provider().count_admin_users() - if admin_count > 0: - raise HTTPException( - status_code=status.HTTP_409_CONFLICT, - detail=AuthErrorResponse(code=AuthErrorCode.SYSTEM_ALREADY_INITIALIZED, message="System already initialized").model_dump(), - ) + already_initialized = HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail=AuthErrorResponse(code=AuthErrorCode.SYSTEM_ALREADY_INITIALIZED, message="System already initialized").model_dump(), + ) + + # Fast path only: an initialized system answers without hashing a + # password. The claim below is what actually decides, because a count + # read here cannot exclude a request already in flight. + if await get_local_provider().count_admin_users() > 0: + raise already_initialized try: - user = await get_local_provider().create_user(email=body.email, password=body.password, system_role="admin", needs_setup=False) + user = await get_local_provider().create_first_admin(email=body.email, password=body.password) except ValueError: - admin_count = await get_local_provider().count_admin_users() - if admin_count == 0: - raise HTTPException( - status_code=status.HTTP_400_BAD_REQUEST, - detail=AuthErrorResponse(code=AuthErrorCode.EMAIL_ALREADY_EXISTS, message="Email already registered").model_dump(), - ) raise HTTPException( - status_code=status.HTTP_409_CONFLICT, - detail=AuthErrorResponse(code=AuthErrorCode.SYSTEM_ALREADY_INITIALIZED, message="System already initialized").model_dump(), + status_code=status.HTTP_400_BAD_REQUEST, + detail=AuthErrorResponse(code=AuthErrorCode.EMAIL_ALREADY_EXISTS, message="Email already registered").model_dump(), ) + if user is None: + raise already_initialized + token = create_access_token(str(user.id), token_version=user.token_version) _set_session_cookie(response, token, request, remember_me=body.remember_me) diff --git a/backend/docs/AUTH_DESIGN.md b/backend/docs/AUTH_DESIGN.md index fb85c9f0ad..1520740a39 100644 --- a/backend/docs/AUTH_DESIGN.md +++ b/backend/docs/AUTH_DESIGN.md @@ -99,7 +99,7 @@ enum UserScope: 5. 服务端确认当前没有 admin,创建 `system_role="admin"`、`needs_setup=false` 的用户。 6. 服务端设置 `access_token` HttpOnly cookie,用户进入 workspace。 -`/api/v1/auth/initialize` 只在没有 admin 时可用。并发初始化由数据库唯一约束兜底,失败方返回 409。 +`/api/v1/auth/initialize` 只在没有 admin 时可用。并发初始化由存储层的原子认领兜底:admin 计数与插入在同一个事务内完成,且写入先被串行化(SQLite 使用 `BEGIN IMMEDIATE`,PostgreSQL 使用事务级 advisory lock),因此同时到达的两个首次初始化请求不会都看到空系统,失败方返回 409。邮箱唯一约束只覆盖相同邮箱的重复提交,无法阻止两个不同邮箱同时成为 admin。 ### 普通登录 diff --git a/backend/tests/test_initialize_admin.py b/backend/tests/test_initialize_admin.py index 841dfc88c5..152bb03d96 100644 --- a/backend/tests/test_initialize_admin.py +++ b/backend/tests/test_initialize_admin.py @@ -258,3 +258,91 @@ async def test_setup_status_single_flight_per_ip(monkeypatch): assert all(result["needs_setup"] is True for result in results) assert provider.calls == 1 + + +@pytest.mark.asyncio +async def test_concurrent_initialize_creates_only_one_admin(_setup_auth): + """Two first-boot requests arriving together must not both become admins.""" + import httpx + + from app.gateway.app import create_app + from app.gateway.deps import get_local_provider + + app = create_app() + transport = httpx.ASGITransport(app=app) + async with httpx.AsyncClient(transport=transport, base_url="http://setup.local") as http: + first, second = await asyncio.gather( + http.post("/api/v1/auth/initialize", json=_init_payload(email="operator@example.com")), + http.post("/api/v1/auth/initialize", json=_init_payload(email="attacker@example.com")), + ) + + assert sorted([first.status_code, second.status_code]) == [201, 409] + assert await get_local_provider().count_admin_users() == 1 + + +@pytest.mark.asyncio +async def test_create_first_admin_claim_is_atomic(_setup_auth): + """The storage claim itself is the guard: only one concurrent caller wins.""" + from app.gateway.auth.models import User + from app.gateway.auth.repositories.sqlite import SQLiteUserRepository + from deerflow.persistence.engine import get_session_factory + + repo = SQLiteUserRepository(get_session_factory()) + + def _admin(email: str) -> User: + return User(email=email, password_hash="hash", system_role="admin", needs_setup=False) + + winner, loser = await asyncio.gather( + repo.create_first_admin(_admin("operator@example.com")), + repo.create_first_admin(_admin("attacker@example.com")), + ) + + created = [result for result in (winner, loser) if result is not None] + assert len(created) == 1, "exactly one caller may claim the first admin" + assert await repo.count_admin_users() == 1 + + +@pytest.mark.asyncio +async def test_create_first_admin_declines_once_an_admin_exists(_setup_auth): + from app.gateway.auth.models import User + from app.gateway.auth.repositories.sqlite import SQLiteUserRepository + from deerflow.persistence.engine import get_session_factory + + repo = SQLiteUserRepository(get_session_factory()) + await repo.create_user(User(email="first@example.com", password_hash="hash", system_role="admin", needs_setup=False)) + + result = await repo.create_first_admin(User(email="second@example.com", password_hash="hash", system_role="admin", needs_setup=False)) + + assert result is None + assert await repo.count_admin_users() == 1 + + +@pytest.mark.asyncio +async def test_create_first_admin_reports_a_taken_email(_setup_auth): + """A regular account already holding the address is still an email conflict.""" + from app.gateway.auth.models import User + from app.gateway.auth.repositories.sqlite import SQLiteUserRepository + from deerflow.persistence.engine import get_session_factory + + repo = SQLiteUserRepository(get_session_factory()) + await repo.create_user(User(email="taken@example.com", password_hash="hash", system_role="user", needs_setup=False)) + + with pytest.raises(ValueError, match="Email already registered"): + await repo.create_first_admin(User(email="taken@example.com", password_hash="hash", system_role="admin", needs_setup=False)) + + assert await repo.count_admin_users() == 0 + + +@pytest.mark.asyncio +async def test_create_first_admin_refuses_a_dialect_it_cannot_serialize(): + """A new backend must fail loudly rather than fall back to check-then-act.""" + from types import SimpleNamespace + + from app.gateway.auth.repositories.sqlite import SQLiteUserRepository + + class _UnknownDialectSession: + def get_bind(self): + return SimpleNamespace(dialect=SimpleNamespace(name="mysql")) + + with pytest.raises(RuntimeError, match="serializ"): + await SQLiteUserRepository._serialize_first_admin_claim(_UnknownDialectSession())