Artifacts/unraid-mcp
2
0
Fork 0
forked from HomeLab/unraid-mcp
Code Pull Requests Actions Packages Releases 5 Activity

feat: add API key bearer token authentication

Browse Source 
- ApiKeyVerifier(TokenVerifier) — validates Authorization: Bearer <key>
  against UNRAID_MCP_API_KEY; guards against empty-key bypass
- _build_auth() replaces module-level _build_google_auth() call:
  returns MultiAuth(server=google, verifiers=[api_key]) when both set,
  GoogleProvider alone, ApiKeyVerifier alone, or None
- settings.py: add UNRAID_MCP_API_KEY + is_api_key_auth_configured()
  + api_key_auth_enabled in get_config_summary()
- run_server(): improved auth status logging for all three states
- tests/test_api_key_auth.py: 9 tests covering verifier + _build_auth
- .env.example: add UNRAID_MCP_API_KEY section
- docs/GOOGLE_OAUTH.md: add API Key section
- README.md / CLAUDE.md: rename section, document both auth methods
- Fix pre-existing: test_health.py patched cache_middleware/error_middleware
  now match renamed _cache_middleware/_error_middleware in server.py
This commit is contained in:
Jacob Magar
2026-03-16 11:11:38 -04:00
parent 6f7a58a0f9
commit cc24f1ec62
16 changed files with 406 additions and 69 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
tests/mcporter/test-tools.sh
View File

@@ -134,6 +134,11 @@ check_prerequisites() {
missing=true
fi
if ! command -v jq &>/dev/null; then
log_error "jq not found in PATH. Install it and re-run."
missing=true
fi
if [[ ! -f "${PROJECT_DIR}/pyproject.toml" ]]; then
log_error "pyproject.toml not found at ${PROJECT_DIR}. Wrong directory?"
missing=true
@@ -181,10 +186,12 @@ smoke_test_server() {
import sys, json
try:
d = json.load(sys.stdin)
if 'status' in d or 'success' in d or 'error' in d:
if 'error' in d:
print('error: tool returned error key — ' + str(d.get('error', '')))
elif 'status' in d or 'success' in d:
print('ok')
else:
print('missing: no status/success/error key in response')
print('missing: no status/success key in response')
except Exception as e:
print('parse_error: ' + str(e))
" 2>/dev/null
@@ -253,6 +260,31 @@ run_test() {
return 1
fi
# Always validate JSON is parseable and not an error payload
local json_check
json_check="$(
printf '%s' "${output}" | python3 -c "
import sys, json
try:
d = json.load(sys.stdin)
if isinstance(d, dict) and ('error' in d or d.get('kind') == 'error'):
print('error: ' + str(d.get('error', d.get('message', 'unknown error'))))
else:
print('ok')
except Exception as e:
print('invalid_json: ' + str(e))
" 2>/dev/null
)" || json_check="parse_error"
if [[ "${json_check}" != "ok" ]]; then
printf "${C_RED}[FAIL]${C_RESET} %-55s ${C_DIM}%dms${C_RESET}\n" \
"${label}" "${elapsed_ms}" | tee -a "${LOG_FILE}"
printf ' response validation failed: %s\n' "${json_check}" | tee -a "${LOG_FILE}"
FAIL_COUNT=$(( FAIL_COUNT + 1 ))
FAIL_NAMES+=("${label}")
return 1
fi
# Validate optional key presence
if [[ -n "${expected_key}" ]]; then
local key_check

3
tests/schema/test_query_validation.py
View File

@@ -36,7 +36,7 @@ def _all_domain_dicts(unraid_mod: object) -> list[tuple[str, dict[str, str]]]:
"""
import types
m = unraid_mod # type: ignore[assignment]
m = unraid_mod
if not isinstance(m, types.ModuleType):
import importlib
@@ -417,7 +417,6 @@ class TestDockerQueries:
"details",
"networks",
"network_details",
"_resolve",
}
assert set(QUERIES.keys()) == expected

155
tests/test_api_key_auth.py Normal file
View File

@@ -0,0 +1,155 @@
"""Tests for ApiKeyVerifier and _build_auth() in server.py."""
import importlib
from unittest.mock import MagicMock, patch
import pytest
from unraid_mcp.server import ApiKeyVerifier, _build_auth
# ---------------------------------------------------------------------------
# ApiKeyVerifier unit tests
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_api_key_verifier_accepts_correct_key():
"""Returns AccessToken when the presented token matches the configured key."""
verifier = ApiKeyVerifier("secret-key-abc123")
result = await verifier.verify_token("secret-key-abc123")
assert result is not None
assert result.client_id == "api-key-client"
assert result.token == "secret-key-abc123"
@pytest.mark.asyncio
async def test_api_key_verifier_rejects_wrong_key():
"""Returns None when the token does not match."""
verifier = ApiKeyVerifier("secret-key-abc123")
result = await verifier.verify_token("wrong-key")
assert result is None
@pytest.mark.asyncio
async def test_api_key_verifier_rejects_empty_token():
"""Returns None for an empty string token."""
verifier = ApiKeyVerifier("secret-key-abc123")
result = await verifier.verify_token("")
assert result is None
@pytest.mark.asyncio
async def test_api_key_verifier_empty_key_rejects_empty_token():
"""When initialised with empty key, even an empty token is rejected.
An empty UNRAID_MCP_API_KEY means auth is disabled — ApiKeyVerifier
should not be instantiated in that case. But if it is, it must not
grant access via an empty bearer token.
"""
verifier = ApiKeyVerifier("")
result = await verifier.verify_token("")
assert result is None
# ---------------------------------------------------------------------------
# _build_auth() integration tests
# ---------------------------------------------------------------------------
def test_build_auth_returns_none_when_nothing_configured(monkeypatch):
"""Returns None when neither Google OAuth nor API key is set."""
monkeypatch.setenv("GOOGLE_CLIENT_ID", "")
monkeypatch.setenv("GOOGLE_CLIENT_SECRET", "")
monkeypatch.setenv("UNRAID_MCP_BASE_URL", "")
monkeypatch.setenv("UNRAID_MCP_API_KEY", "")
import unraid_mcp.config.settings as s
importlib.reload(s)
result = _build_auth()
assert result is None
def test_build_auth_returns_api_key_verifier_when_only_api_key_set(monkeypatch):
"""Returns ApiKeyVerifier when UNRAID_MCP_API_KEY is set but Google OAuth is not."""
monkeypatch.setenv("GOOGLE_CLIENT_ID", "")
monkeypatch.setenv("GOOGLE_CLIENT_SECRET", "")
monkeypatch.setenv("UNRAID_MCP_BASE_URL", "")
monkeypatch.setenv("UNRAID_MCP_API_KEY", "my-secret-api-key")
import unraid_mcp.config.settings as s
importlib.reload(s)
result = _build_auth()
assert isinstance(result, ApiKeyVerifier)
def test_build_auth_returns_google_provider_when_only_oauth_set(monkeypatch):
"""Returns GoogleProvider when Google OAuth vars are set but no API key."""
monkeypatch.setenv("GOOGLE_CLIENT_ID", "test-id.apps.googleusercontent.com")
monkeypatch.setenv("GOOGLE_CLIENT_SECRET", "GOCSPX-test-secret")
monkeypatch.setenv("UNRAID_MCP_BASE_URL", "http://10.1.0.2:6970")
monkeypatch.setenv("UNRAID_MCP_API_KEY", "")
monkeypatch.setenv("UNRAID_MCP_JWT_SIGNING_KEY", "x" * 32)
import unraid_mcp.config.settings as s
importlib.reload(s)
mock_provider = MagicMock()
with patch("unraid_mcp.server.GoogleProvider", return_value=mock_provider):
result = _build_auth()
assert result is mock_provider
def test_build_auth_returns_multi_auth_when_both_configured(monkeypatch):
"""Returns MultiAuth when both Google OAuth and UNRAID_MCP_API_KEY are set."""
from fastmcp.server.auth import MultiAuth
monkeypatch.setenv("GOOGLE_CLIENT_ID", "test-id.apps.googleusercontent.com")
monkeypatch.setenv("GOOGLE_CLIENT_SECRET", "GOCSPX-test-secret")
monkeypatch.setenv("UNRAID_MCP_BASE_URL", "http://10.1.0.2:6970")
monkeypatch.setenv("UNRAID_MCP_API_KEY", "my-secret-api-key")
monkeypatch.setenv("UNRAID_MCP_JWT_SIGNING_KEY", "x" * 32)
import unraid_mcp.config.settings as s
importlib.reload(s)
mock_provider = MagicMock()
with patch("unraid_mcp.server.GoogleProvider", return_value=mock_provider):
result = _build_auth()
assert isinstance(result, MultiAuth)
# Server is the Google provider
assert result.server is mock_provider
# One additional verifier — the ApiKeyVerifier
assert len(result.verifiers) == 1
assert isinstance(result.verifiers[0], ApiKeyVerifier)
def test_build_auth_multi_auth_api_key_verifier_uses_correct_key(monkeypatch):
"""The ApiKeyVerifier inside MultiAuth is seeded with the configured key."""
monkeypatch.setenv("GOOGLE_CLIENT_ID", "test-id.apps.googleusercontent.com")
monkeypatch.setenv("GOOGLE_CLIENT_SECRET", "GOCSPX-test-secret")
monkeypatch.setenv("UNRAID_MCP_BASE_URL", "http://10.1.0.2:6970")
monkeypatch.setenv("UNRAID_MCP_API_KEY", "super-secret-token")
monkeypatch.setenv("UNRAID_MCP_JWT_SIGNING_KEY", "x" * 32)
import unraid_mcp.config.settings as s
importlib.reload(s)
with patch("unraid_mcp.server.GoogleProvider", return_value=MagicMock()):
result = _build_auth()
verifier = result.verifiers[0]
assert verifier._api_key == "super-secret-token"

1
tests/test_customization.py
View File

@@ -3,6 +3,7 @@
from __future__ import annotations
from typing import Any
from unittest.mock import AsyncMock, patch
import pytest

4
tests/test_health.py
View File

@@ -141,8 +141,8 @@ class TestHealthActions:
"unraid_mcp.subscriptions.utils._analyze_subscription_status",
return_value=(0, []),
),
patch("unraid_mcp.server.cache_middleware", mock_cache),
patch("unraid_mcp.server.error_middleware", mock_error),
patch("unraid_mcp.server._cache_middleware", mock_cache),
patch("unraid_mcp.server._error_middleware", mock_error),
):
result = await tool_fn(action="health", subaction="diagnose")
assert "subscriptions" in result

14
tests/test_resources.py
View File

@@ -36,6 +36,8 @@ class TestLiveResourcesUseManagerCache:
with patch("unraid_mcp.subscriptions.resources.subscription_manager") as mock_mgr:
mock_mgr.get_resource_data = AsyncMock(return_value=cached)
mcp = _make_resources()
# Accessing FastMCP internals intentionally for unit test isolation.
# This may break on FastMCP upgrades — consider a make_resource_fn() helper if it does.
resource = mcp.providers[0]._components[f"resource:unraid://live/{action}@"]
result = await resource.fn()
assert json.loads(result) == cached
@@ -49,6 +51,8 @@ class TestLiveResourcesUseManagerCache:
mock_mgr.get_resource_data = AsyncMock(return_value=None)
mock_mgr.last_error = {}
mcp = _make_resources()
# Accessing FastMCP internals intentionally for unit test isolation.
# This may break on FastMCP upgrades — consider a make_resource_fn() helper if it does.
resource = mcp.providers[0]._components[f"resource:unraid://live/{action}@"]
result = await resource.fn()
parsed = json.loads(result)
@@ -61,6 +65,8 @@ class TestLiveResourcesUseManagerCache:
mock_mgr.get_resource_data = AsyncMock(return_value=None)
mock_mgr.last_error = {action: "WebSocket auth failed"}
mcp = _make_resources()
# Accessing FastMCP internals intentionally for unit test isolation.
# This may break on FastMCP upgrades — consider a make_resource_fn() helper if it does.
resource = mcp.providers[0]._components[f"resource:unraid://live/{action}@"]
result = await resource.fn()
parsed = json.loads(result)
@@ -96,6 +102,8 @@ class TestLogsStreamResource:
mock_mgr.get_resource_data = AsyncMock(return_value=None)
mcp = _make_resources()
local_provider = mcp.providers[0]
# Accessing FastMCP internals intentionally for unit test isolation.
# This may break on FastMCP upgrades — consider a make_resource_fn() helper if it does.
resource = local_provider._components["resource:unraid://logs/stream@"]
result = await resource.fn()
parsed = json.loads(result)
@@ -108,6 +116,8 @@ class TestLogsStreamResource:
mock_mgr.get_resource_data = AsyncMock(return_value={})
mcp = _make_resources()
local_provider = mcp.providers[0]
# Accessing FastMCP internals intentionally for unit test isolation.
# This may break on FastMCP upgrades — consider a make_resource_fn() helper if it does.
resource = local_provider._components["resource:unraid://logs/stream@"]
result = await resource.fn()
assert json.loads(result) == {}
@@ -131,6 +141,8 @@ class TestAutoStartDisabledFallback:
mock_mgr.last_error = {}
mock_mgr.auto_start_enabled = False
mcp = _make_resources()
# Accessing FastMCP internals intentionally for unit test isolation.
# This may break on FastMCP upgrades — consider a make_resource_fn() helper if it does.
resource = mcp.providers[0]._components[f"resource:unraid://live/{action}@"]
result = await resource.fn()
assert json.loads(result) == fallback_data
@@ -150,6 +162,8 @@ class TestAutoStartDisabledFallback:
mock_mgr.last_error = {}
mock_mgr.auto_start_enabled = False
mcp = _make_resources()
# Accessing FastMCP internals intentionally for unit test isolation.
# This may break on FastMCP upgrades — consider a make_resource_fn() helper if it does.
resource = mcp.providers[0]._components[f"resource:unraid://live/{action}@"]
result = await resource.fn()
assert json.loads(result)["status"] == "connecting"