583a140485
SHA-256 hashed keys, fmcp_ prefix, soft-delete revocation. Tests cover pure logic (hash, prefix, format, uniqueness, scope validation) and auth middleware bearer token path (pending auth.py update). Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
167 lines
5.1 KiB
Python
167 lines
5.1 KiB
Python
"""Unit tests for api_keys service — pure logic, DB calls mocked."""
|
|
import hashlib
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from fabledassistant.services.api_keys import (
|
|
_hash_key,
|
|
_key_prefix,
|
|
generate_key,
|
|
create_api_key,
|
|
list_api_keys,
|
|
revoke_api_key,
|
|
lookup_key,
|
|
)
|
|
|
|
|
|
def test_generate_key_format():
|
|
key = generate_key()
|
|
assert key.startswith("fmcp_")
|
|
assert len(key) > 12
|
|
|
|
|
|
def test_generate_key_uniqueness():
|
|
keys = {generate_key() for _ in range(100)}
|
|
assert len(keys) == 100
|
|
|
|
|
|
def test_hash_key_is_sha256():
|
|
key = "fmcp_testkey"
|
|
h = _hash_key(key)
|
|
expected = hashlib.sha256(key.encode()).hexdigest()
|
|
assert h == expected
|
|
|
|
|
|
def test_key_prefix_returns_first_12_chars():
|
|
key = "fmcp_abcdefghijklmnop"
|
|
assert _key_prefix(key) == "fmcp_abcdefg"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_api_key_returns_full_key():
|
|
mock_key_obj = MagicMock()
|
|
mock_key_obj.id = 1
|
|
mock_key_obj.to_dict.return_value = {"id": 1, "name": "test", "scope": "read", "key_prefix": "fmcp_xxx"}
|
|
|
|
with patch("fabledassistant.services.api_keys.async_session") as mock_session_ctx:
|
|
mock_session = AsyncMock()
|
|
mock_session.__aenter__ = AsyncMock(return_value=mock_session)
|
|
mock_session.__aexit__ = AsyncMock(return_value=False)
|
|
mock_session.add = MagicMock()
|
|
mock_session.commit = AsyncMock()
|
|
mock_session.refresh = AsyncMock(side_effect=lambda obj: None)
|
|
# Return our fake object from refresh
|
|
mock_session.refresh = AsyncMock()
|
|
mock_key_obj.to_dict.return_value = {"id": 1, "name": "test", "scope": "read", "key_prefix": "fmcp_xxx"}
|
|
mock_session_ctx.return_value = mock_session
|
|
|
|
# Patch the select result
|
|
with patch("fabledassistant.services.api_keys.ApiKey") as mock_model:
|
|
mock_model.return_value = mock_key_obj
|
|
full_key, key_dict = await create_api_key(user_id=1, name="test", scope="read")
|
|
|
|
assert full_key.startswith("fmcp_")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_lookup_key_returns_none_for_unknown():
|
|
with patch("fabledassistant.services.api_keys.async_session") as mock_session_ctx:
|
|
mock_session = AsyncMock()
|
|
mock_session.__aenter__ = AsyncMock(return_value=mock_session)
|
|
mock_session.__aexit__ = AsyncMock(return_value=False)
|
|
mock_result = MagicMock()
|
|
mock_result.scalars.return_value.first.return_value = None
|
|
mock_session.execute = AsyncMock(return_value=mock_result)
|
|
mock_session_ctx.return_value = mock_session
|
|
|
|
result = await lookup_key("fmcp_doesnotexist")
|
|
|
|
assert result is None
|
|
|
|
|
|
def test_hash_key_deterministic():
|
|
key = "fmcp_some_test_key_value"
|
|
assert _hash_key(key) == _hash_key(key)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scope_validation():
|
|
with pytest.raises(ValueError, match="scope must be"):
|
|
await create_api_key(1, "bad", "superadmin")
|
|
|
|
|
|
# --- Auth middleware tests ---
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_read_only_key_blocked_on_post():
|
|
"""Read-only API key returns 403 on non-GET requests."""
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
from quart import Quart
|
|
|
|
fake_key = MagicMock()
|
|
fake_key.user_id = 7
|
|
fake_key.scope = "read"
|
|
|
|
fake_user = MagicMock()
|
|
fake_user.role = "user"
|
|
|
|
app = Quart(__name__)
|
|
|
|
async def dummy():
|
|
from quart import jsonify
|
|
return jsonify({"ok": True})
|
|
|
|
from fabledassistant.auth import _check_auth
|
|
wrapped = _check_auth(dummy)
|
|
|
|
async with app.test_request_context(
|
|
"/test", method="POST",
|
|
headers={"Authorization": "Bearer fmcp_readonly"},
|
|
):
|
|
with patch("fabledassistant.auth.lookup_key", AsyncMock(return_value=fake_key)), \
|
|
patch("fabledassistant.auth.get_user_by_id", AsyncMock(return_value=fake_user)):
|
|
resp = await wrapped()
|
|
assert resp[1] == 403
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_bearer_token_valid_write_key_passes():
|
|
"""Valid write-scoped bearer token passes auth on POST."""
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
from quart import Quart, g
|
|
|
|
fake_key = MagicMock()
|
|
fake_key.user_id = 7
|
|
fake_key.scope = "write"
|
|
|
|
fake_user = MagicMock()
|
|
fake_user.role = "user"
|
|
fake_user.id = 7
|
|
|
|
app = Quart(__name__)
|
|
|
|
result_holder = {}
|
|
|
|
async def dummy():
|
|
result_holder["user"] = g.user
|
|
result_holder["api_key"] = g.api_key
|
|
from quart import jsonify
|
|
return jsonify({"ok": True})
|
|
|
|
from fabledassistant.auth import _check_auth
|
|
wrapped = _check_auth(dummy)
|
|
|
|
async with app.test_request_context(
|
|
"/test", method="POST",
|
|
headers={"Authorization": "Bearer fmcp_writekey"},
|
|
):
|
|
with patch("fabledassistant.auth.lookup_key", AsyncMock(return_value=fake_key)), \
|
|
patch("fabledassistant.auth.get_user_by_id", AsyncMock(return_value=fake_user)):
|
|
resp = await wrapped()
|
|
|
|
# Should not be a tuple (no error response)
|
|
assert not isinstance(resp, tuple) or resp[1] == 200
|
|
assert result_holder.get("user") is fake_user
|
|
assert result_holder.get("api_key") is fake_key
|