Codex Release Bot
100 lines
3.5 KiB
Python
100 lines
3.5 KiB
Python
"""BL-06: Agent auth must use per-server agent_api_key only (no global API_KEY fallback)."""
|
|
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
import pytest
|
|
|
|
from server.api.agent import _verify_server_api_key
|
|
from server.application.services.script_service import ScriptService
|
|
from server.config import settings
|
|
|
|
|
|
def _fake_server(*, agent_api_key=None, name="test-srv"):
|
|
srv = MagicMock()
|
|
srv.name = name
|
|
srv.agent_api_key = agent_api_key
|
|
return srv
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_service():
|
|
return AsyncMock()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_verify_server_key_matches_per_server(mock_service):
|
|
mock_service.get_server.return_value = _fake_server(agent_api_key="nxs-abc")
|
|
result = await _verify_server_api_key(1, "nxs-abc", mock_service)
|
|
assert result is not None
|
|
assert result.agent_api_key == "nxs-abc"
|
|
mock_service.get_server.assert_called_once_with(1)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_verify_server_key_single_get_server_per_auth_check(mock_service):
|
|
"""Regression: auth path must not double-query the same server_id (BL-06 patrol)."""
|
|
mock_service.get_server.return_value = _fake_server(agent_api_key="nxs-once")
|
|
await _verify_server_api_key(7, "nxs-once", mock_service)
|
|
mock_service.get_server.assert_called_once_with(7)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_verify_server_key_rejects_wrong_key(mock_service):
|
|
mock_service.get_server.return_value = _fake_server(agent_api_key="nxs-abc")
|
|
assert await _verify_server_api_key(1, "nxs-wrong", mock_service) is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_verify_server_key_rejects_global_api_key(mock_service, monkeypatch):
|
|
monkeypatch.setattr(settings, "API_KEY", "global-secret-key")
|
|
mock_service.get_server.return_value = _fake_server(agent_api_key="nxs-abc")
|
|
assert await _verify_server_api_key(1, "global-secret-key", mock_service) is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_verify_server_key_no_per_server_key_rejects_global(mock_service, monkeypatch):
|
|
monkeypatch.setattr(settings, "API_KEY", "global-secret-key")
|
|
mock_service.get_server.return_value = _fake_server(agent_api_key=None)
|
|
assert await _verify_server_api_key(1, "global-secret-key", mock_service) is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_verify_server_key_server_not_found(mock_service):
|
|
mock_service.get_server.return_value = None
|
|
assert await _verify_server_api_key(99, "any-key", mock_service) is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_verify_server_key_db_error_fail_closed(mock_service):
|
|
mock_service.get_server.side_effect = RuntimeError("db down")
|
|
assert await _verify_server_api_key(1, "any-key", mock_service) is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_script_service_agent_key_requires_per_server():
|
|
repo = AsyncMock()
|
|
repo.get_by_id.return_value = _fake_server(agent_api_key=None)
|
|
svc = ScriptService(
|
|
script_repo=AsyncMock(),
|
|
execution_repo=AsyncMock(),
|
|
credential_repo=AsyncMock(),
|
|
server_repo=repo,
|
|
audit_repo=AsyncMock(),
|
|
)
|
|
with pytest.raises(ValueError, match="未设置 Agent API Key"):
|
|
await svc._agent_api_key_for_server(42)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_script_service_agent_key_returns_per_server():
|
|
repo = AsyncMock()
|
|
repo.get_by_id.return_value = _fake_server(agent_api_key="nxs-callback")
|
|
svc = ScriptService(
|
|
script_repo=AsyncMock(),
|
|
execution_repo=AsyncMock(),
|
|
credential_repo=AsyncMock(),
|
|
server_repo=repo,
|
|
audit_repo=AsyncMock(),
|
|
)
|
|
assert await svc._agent_api_key_for_server(1) == "nxs-callback"
|