Nexus Agent
72d82d737b
外网无 token WebSocket 的 HTTP 403 为应用层拒绝而非反代故障；边端 agent 在中心 401 时停止心跳重试；install 锁定路由级 404 且归档失败 fail-closed。 同步安全规范与 BL-06 一致，门控 7/7 PASS。 Co-authored-by: Cursor <cursoragent@cursor.com>
141 lines
3.6 KiB
Python
141 lines
3.6 KiB
Python
"""BL-05: locked install API returns 404 (no fingerprint) on all endpoints."""
|
|
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
|
|
|
|
@pytest.fixture
|
|
def locked():
|
|
with patch("server.api.install._is_locked", return_value=True):
|
|
yield
|
|
|
|
|
|
@pytest.fixture
|
|
def unlocked():
|
|
with patch("server.api.install._is_locked", return_value=False):
|
|
yield
|
|
|
|
|
|
def _assert_not_found(exc: HTTPException) -> None:
|
|
assert exc.status_code == 404
|
|
assert exc.detail == "Not found"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_status_locked_404(locked):
|
|
from server.api.install import install_status
|
|
|
|
with pytest.raises(HTTPException) as exc:
|
|
await install_status()
|
|
_assert_not_found(exc.value)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_status_unlocked_200(unlocked):
|
|
from server.api.install import install_status
|
|
|
|
with patch("server.api.install._has_env", return_value=False):
|
|
result = await install_status()
|
|
assert result["locked"] is False
|
|
assert "configured" in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_env_check_locked_404(locked):
|
|
from server.api.install import env_check
|
|
|
|
with pytest.raises(HTTPException) as exc:
|
|
await env_check()
|
|
_assert_not_found(exc.value)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_connection_check_locked_404(locked):
|
|
from server.api.install import connection_check
|
|
|
|
with pytest.raises(HTTPException) as exc:
|
|
await connection_check()
|
|
_assert_not_found(exc.value)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_state_locked_404(locked):
|
|
from server.api.install import get_install_state
|
|
|
|
with pytest.raises(HTTPException) as exc:
|
|
await get_install_state()
|
|
_assert_not_found(exc.value)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_lock_endpoint_locked_404(locked):
|
|
from server.api.install import lock_install
|
|
|
|
with pytest.raises(HTTPException) as exc:
|
|
await lock_install()
|
|
_assert_not_found(exc.value)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_init_db_locked_404(locked):
|
|
from server.api.install import init_db, InitDbRequest
|
|
|
|
req = InitDbRequest(
|
|
db_host="localhost",
|
|
db_port="3306",
|
|
db_name="nexus",
|
|
db_user="root",
|
|
db_pass="x",
|
|
)
|
|
with pytest.raises(HTTPException) as exc:
|
|
await init_db(req)
|
|
_assert_not_found(exc.value)
|
|
|
|
|
|
def test_reject_if_locked_raises_404():
|
|
from server.api.install import _reject_if_locked
|
|
|
|
with patch("server.api.install._is_locked", return_value=True):
|
|
with pytest.raises(HTTPException) as exc:
|
|
_reject_if_locked()
|
|
_assert_not_found(exc.value)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_test_credentials_locked_404(locked):
|
|
from server.api.install import test_credentials, InitDbRequest
|
|
|
|
req = InitDbRequest(
|
|
db_host="localhost",
|
|
db_port="3306",
|
|
db_name="nexus",
|
|
db_user="root",
|
|
db_pass="x",
|
|
)
|
|
with pytest.raises(HTTPException) as exc:
|
|
await test_credentials(req)
|
|
_assert_not_found(exc.value)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_admin_locked_404(locked):
|
|
from server.api.install import create_admin, CreateAdminRequest
|
|
|
|
req = CreateAdminRequest(
|
|
install_token="tok",
|
|
db_pass="x",
|
|
admin_password="secret12",
|
|
)
|
|
with pytest.raises(HTTPException) as exc:
|
|
await create_admin(req)
|
|
_assert_not_found(exc.value)
|
|
|
|
|
|
def test_install_router_has_lock_dependency():
|
|
"""Router-level lock guard runs before per-route handlers (BL-05)."""
|
|
from server.api.install import router
|
|
|
|
assert router.dependencies, "install router must enforce lock before body validation"
|