mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-05-20 15:11:09 +00:00
refactor: Remove init_token handling from admin initialization logic and related tests
This commit is contained in:
@@ -1,7 +1,7 @@
|
||||
"""Tests for the POST /api/v1/auth/initialize endpoint.
|
||||
|
||||
Covers: first-boot admin creation, rejection when system already
|
||||
initialized, invalid/missing init_token, password strength validation,
|
||||
initialized, password strength validation,
|
||||
and public accessibility (no auth cookie required).
|
||||
"""
|
||||
|
||||
@@ -16,7 +16,6 @@ os.environ.setdefault("AUTH_JWT_SECRET", "test-secret-key-initialize-admin-min-3
|
||||
from app.gateway.auth.config import AuthConfig, set_auth_config
|
||||
|
||||
_TEST_SECRET = "test-secret-key-initialize-admin-min-32"
|
||||
_INIT_TOKEN = "test-init-token-for-initialization-tests"
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
@@ -45,9 +44,6 @@ def client(_setup_auth):
|
||||
|
||||
set_auth_config(AuthConfig(jwt_secret=_TEST_SECRET))
|
||||
app = create_app()
|
||||
# Pre-set the init token on app.state (normally done by the lifespan on
|
||||
# first boot; tests don't run the lifespan because it requires config.yaml).
|
||||
app.state.init_token = _INIT_TOKEN
|
||||
# Do NOT use TestClient as a context manager — that would trigger the
|
||||
# full lifespan which requires config.yaml. The auth endpoints work
|
||||
# without the lifespan (persistence engine is set up by _setup_auth).
|
||||
@@ -55,11 +51,10 @@ def client(_setup_auth):
|
||||
|
||||
|
||||
def _init_payload(**extra):
|
||||
"""Build a valid /initialize payload with the test init_token."""
|
||||
"""Build a valid /initialize payload."""
|
||||
return {
|
||||
"email": "admin@example.com",
|
||||
"password": "Str0ng!Pass99",
|
||||
"init_token": _INIT_TOKEN,
|
||||
**extra,
|
||||
}
|
||||
|
||||
@@ -85,53 +80,12 @@ def test_initialize_needs_setup_false(client):
|
||||
assert me.json()["needs_setup"] is False
|
||||
|
||||
|
||||
# ── Token validation ──────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_initialize_rejects_wrong_token(client):
|
||||
"""Wrong init_token → 403 invalid_init_token."""
|
||||
resp = client.post(
|
||||
"/api/v1/auth/initialize",
|
||||
json={**_init_payload(), "init_token": "wrong-token"},
|
||||
)
|
||||
assert resp.status_code == 403
|
||||
assert resp.json()["detail"]["code"] == "invalid_init_token"
|
||||
|
||||
|
||||
def test_initialize_rejects_empty_token(client):
|
||||
"""Empty init_token → 403 (fails constant-time comparison against stored token)."""
|
||||
resp = client.post(
|
||||
"/api/v1/auth/initialize",
|
||||
json={**_init_payload(), "init_token": ""},
|
||||
)
|
||||
assert resp.status_code == 403
|
||||
|
||||
|
||||
def test_initialize_token_consumed_after_success(client):
|
||||
"""After a successful /initialize the token is consumed and cannot be reused."""
|
||||
client.post("/api/v1/auth/initialize", json=_init_payload())
|
||||
# The token is now None; any subsequent call with the old token must be rejected (403)
|
||||
resp2 = client.post(
|
||||
"/api/v1/auth/initialize",
|
||||
json={**_init_payload(), "email": "other@example.com"},
|
||||
)
|
||||
assert resp2.status_code == 403
|
||||
|
||||
|
||||
# ── Rejection when already initialized ───────────────────────────────────
|
||||
|
||||
|
||||
def test_initialize_rejected_when_admin_exists(client):
|
||||
"""Second call to /initialize after admin exists → 409 system_already_initialized.
|
||||
|
||||
The first call consumes the token. Re-setting it on app.state simulates
|
||||
what would happen if the operator somehow restarted or manually refreshed
|
||||
the token (e.g., in testing).
|
||||
"""
|
||||
"""Second call to /initialize after admin exists → 409 system_already_initialized."""
|
||||
client.post("/api/v1/auth/initialize", json=_init_payload())
|
||||
# Re-set the token so the second attempt can pass token validation
|
||||
# and reach the admin-exists check.
|
||||
client.app.state.init_token = _INIT_TOKEN
|
||||
resp2 = client.post(
|
||||
"/api/v1/auth/initialize",
|
||||
json={**_init_payload(), "email": "other@example.com"},
|
||||
@@ -141,24 +95,6 @@ def test_initialize_rejected_when_admin_exists(client):
|
||||
assert body["detail"]["code"] == "system_already_initialized"
|
||||
|
||||
|
||||
def test_initialize_token_not_consumed_on_admin_exists(client):
|
||||
"""Token is NOT consumed when the admin-exists guard rejects the request.
|
||||
|
||||
This prevents a DoS where an attacker calls with the correct token when
|
||||
admin already exists and permanently burns the init token.
|
||||
"""
|
||||
client.post("/api/v1/auth/initialize", json=_init_payload())
|
||||
# Token consumed by success above; re-simulate the scenario:
|
||||
# admin exists, token is still valid (re-set), call should 409 and NOT consume token.
|
||||
client.app.state.init_token = _INIT_TOKEN
|
||||
client.post(
|
||||
"/api/v1/auth/initialize",
|
||||
json={**_init_payload(), "email": "other@example.com"},
|
||||
)
|
||||
# Token must still be set (not consumed) after the 409 rejection.
|
||||
assert client.app.state.init_token == _INIT_TOKEN
|
||||
|
||||
|
||||
def test_initialize_register_does_not_block_initialization(client):
|
||||
"""/register creating a user before /initialize doesn't block admin creation."""
|
||||
# Register a regular user first
|
||||
|
||||
Reference in New Issue
Block a user