fix(auth): claim the first admin atomically instead of counting first (#5776)

* fix(auth): claim the first admin atomically instead of counting first

POST /api/v1/auth/initialize read the admin count in one session and
created the admin in another, with no lock between them. Two requests
arriving together with different emails both saw an empty system and
both became admins: reproduced over HTTP, two 201s and two admin rows,
after which setup-status reports an initialized system.

The endpoint is public and unthrottled, and during the pre-setup window
anyone who can reach the gateway can claim admin with a single request
anyway — that part is the first-boot design. What the race removes is
the only signal: without it exactly one request wins and the loser gets
409, so an operator who loses sees "already initialized" and looks into
it. With it, both succeed and the operator's setup looks ordinary.

The check and the insert now share one transaction in the repository,
with writers serialized first — SQLite takes the write lock up front
(BEGIN IMMEDIATE) and Postgres a transaction-scoped advisory lock, the
idiom the project and channel repositories already use. The same
repository serves both backends, and multi-replica deployments make the
race genuinely parallel rather than an event-loop interleaving.

create_first_admin returns None when the claim is lost, which the route
reports as 409; uniqueness conflicts still raise ValueError and stay a
400. The count in the handler is now only a fast path that avoids
hashing a password for an already-initialized system.

AUTH_DESIGN.md said concurrent initialization was backstopped by a
database unique constraint. No such constraint exists — email
uniqueness only catches the same address submitted twice — so it now
describes the claim that does the work.

* docs(changelog): note the atomic first-admin claim (#5776)

* fix(auth): refuse a dialect the first-admin claim cannot serialize

Review follow-up. The claim's serialization handled sqlite and
postgresql and fell through silently for anything else, which would
leave create_first_admin as the plain check-then-act it exists to
prevent — the one place the abstract method's atomicity contract is
enforced.

An unknown dialect now raises instead. The engine builds only these two
today, so this guards a future backend: it fails loudly at first boot
rather than quietly reopening the race.

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

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

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)

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.

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:

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)
 

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。
 
 ### 普通登录
 

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())