feat(auth): drop test-client defaults, add ALLOWED_MGMT_CLIENT allowlist
Both auth surfaces now fail-closed by default: - ALLOWED_MCP_CLIENTS: removed the silent `claude-desktop` and `test-mcp-client` fallbacks. Empty/unset env var leaves the registry empty so /oauth/authorize rejects every client_id. - ALLOWED_MGMT_CLIENT (new): comma-separated list of OIDC client_ids whose tokens are accepted by /api/management/*. Enforced in verify_token_for_management_api on both the cache-hit and cache-miss paths against the token's client_id claim. Unset/empty rejects all. Compose: set ALLOWED_MGMT_CLIENT=nextcloudMcpServerUIPublicClient on mcp-multi-user-basic so the existing Astrolabe integration test (test_astrolabe_chunk_context.py) still passes. env.sample documents both vars and notes they may be consolidated later. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 4.7
parent
2bd1c18b43
commit
bd7702ad12
@@ -88,20 +88,23 @@ def test_http_non_localhost_rejected(monkeypatch, caplog):
|
||||
assert "evil.com" in caplog.text
|
||||
|
||||
|
||||
def test_empty_string_uses_well_known(monkeypatch):
|
||||
registry = _get_registry(monkeypatch, "")
|
||||
clients = registry.list_clients()
|
||||
client_ids = {c.client_id for c in clients}
|
||||
assert "claude-desktop" in client_ids
|
||||
assert "test-mcp-client" in client_ids
|
||||
def test_empty_string_yields_empty_registry(monkeypatch, caplog):
|
||||
with caplog.at_level(logging.WARNING):
|
||||
registry = _get_registry(monkeypatch, "")
|
||||
assert registry.list_clients() == []
|
||||
valid, err = registry.validate_client("claude-desktop")
|
||||
assert valid is False
|
||||
assert "Unknown client_id" in err
|
||||
assert "ALLOWED_MCP_CLIENTS is unset or empty" in caplog.text
|
||||
|
||||
|
||||
def test_unset_env_uses_well_known(monkeypatch):
|
||||
registry = _get_registry(monkeypatch, None)
|
||||
clients = registry.list_clients()
|
||||
client_ids = {c.client_id for c in clients}
|
||||
assert "claude-desktop" in client_ids
|
||||
assert "test-mcp-client" in client_ids
|
||||
def test_unset_env_yields_empty_registry(monkeypatch, caplog):
|
||||
with caplog.at_level(logging.WARNING):
|
||||
registry = _get_registry(monkeypatch, None)
|
||||
assert registry.list_clients() == []
|
||||
valid, err = registry.validate_client("test-mcp-client")
|
||||
assert valid is False
|
||||
assert "Unknown client_id" in err
|
||||
|
||||
|
||||
def test_malformed_entries_skipped_with_warning(monkeypatch, caplog):
|
||||
@@ -153,14 +156,6 @@ def test_validate_redirect_uri_localhost_wildcard(monkeypatch):
|
||||
assert err is None
|
||||
|
||||
|
||||
def test_well_known_clients_wildcard_scopes(monkeypatch):
|
||||
registry = _get_registry(monkeypatch, None)
|
||||
for client in registry.list_clients():
|
||||
assert client.allowed_scopes == ["*"], (
|
||||
f"Well-known client {client.client_id} should have wildcard scopes"
|
||||
)
|
||||
|
||||
|
||||
def test_client_name_resolution(monkeypatch):
|
||||
registry = _get_registry(monkeypatch, "claude-desktop, custom-tool")
|
||||
assert registry.get_client("claude-desktop").name == "Claude Desktop"
|
||||
|
||||
@@ -512,3 +512,114 @@ class TestVerifyTokenFlow:
|
||||
result = await verifier.verify_token("opaque-token")
|
||||
assert result is not None
|
||||
assert result.resource == "testuser"
|
||||
|
||||
|
||||
class TestManagementApiAllowlist:
|
||||
"""Test ALLOWED_MGMT_CLIENT enforcement in verify_token_for_management_api."""
|
||||
|
||||
@staticmethod
|
||||
def _underlying_token(client_id: str = "astrolabe"):
|
||||
from mcp.server.auth.provider import AccessToken
|
||||
|
||||
return AccessToken(
|
||||
token="t",
|
||||
client_id=client_id,
|
||||
scopes=["openid"],
|
||||
expires_at=int(time.time() + 3600),
|
||||
resource="testuser",
|
||||
)
|
||||
|
||||
async def test_unset_allowlist_rejects_all(self, monkeypatch, base_settings):
|
||||
monkeypatch.delenv("ALLOWED_MGMT_CLIENT", raising=False)
|
||||
verifier = UnifiedTokenVerifier(base_settings)
|
||||
assert verifier._allowed_mgmt_clients == frozenset()
|
||||
|
||||
with patch.object(
|
||||
verifier,
|
||||
"_verify_without_audience_check",
|
||||
return_value=self._underlying_token("astrolabe"),
|
||||
):
|
||||
result = await verifier.verify_token_for_management_api("any-token")
|
||||
assert result is None
|
||||
|
||||
async def test_empty_allowlist_rejects_all(self, monkeypatch, base_settings):
|
||||
monkeypatch.setenv("ALLOWED_MGMT_CLIENT", " , ,")
|
||||
verifier = UnifiedTokenVerifier(base_settings)
|
||||
assert verifier._allowed_mgmt_clients == frozenset()
|
||||
|
||||
with patch.object(
|
||||
verifier,
|
||||
"_verify_without_audience_check",
|
||||
return_value=self._underlying_token("astrolabe"),
|
||||
):
|
||||
result = await verifier.verify_token_for_management_api("any-token")
|
||||
assert result is None
|
||||
|
||||
async def test_allowlisted_client_accepted(self, monkeypatch, base_settings):
|
||||
monkeypatch.setenv("ALLOWED_MGMT_CLIENT", "astrolabe, admin-tool")
|
||||
verifier = UnifiedTokenVerifier(base_settings)
|
||||
assert verifier._allowed_mgmt_clients == {"astrolabe", "admin-tool"}
|
||||
|
||||
underlying = self._underlying_token("astrolabe")
|
||||
with patch.object(
|
||||
verifier, "_verify_without_audience_check", return_value=underlying
|
||||
):
|
||||
result = await verifier.verify_token_for_management_api("any-token")
|
||||
assert result is underlying
|
||||
|
||||
async def test_non_allowlisted_client_rejected(self, monkeypatch, base_settings):
|
||||
monkeypatch.setenv("ALLOWED_MGMT_CLIENT", "astrolabe")
|
||||
verifier = UnifiedTokenVerifier(base_settings)
|
||||
|
||||
with patch.object(
|
||||
verifier,
|
||||
"_verify_without_audience_check",
|
||||
return_value=self._underlying_token("some-other-client"),
|
||||
):
|
||||
result = await verifier.verify_token_for_management_api("any-token")
|
||||
assert result is None
|
||||
|
||||
async def test_token_missing_client_id_rejected(self, monkeypatch, base_settings):
|
||||
monkeypatch.setenv("ALLOWED_MGMT_CLIENT", "astrolabe")
|
||||
verifier = UnifiedTokenVerifier(base_settings)
|
||||
|
||||
with patch.object(
|
||||
verifier,
|
||||
"_verify_without_audience_check",
|
||||
return_value=self._underlying_token(""),
|
||||
):
|
||||
result = await verifier.verify_token_for_management_api("any-token")
|
||||
assert result is None
|
||||
|
||||
async def test_underlying_verification_failure_propagates(
|
||||
self, monkeypatch, base_settings
|
||||
):
|
||||
monkeypatch.setenv("ALLOWED_MGMT_CLIENT", "astrolabe")
|
||||
verifier = UnifiedTokenVerifier(base_settings)
|
||||
|
||||
with patch.object(
|
||||
verifier, "_verify_without_audience_check", return_value=None
|
||||
):
|
||||
result = await verifier.verify_token_for_management_api("any-token")
|
||||
assert result is None
|
||||
|
||||
async def test_cache_hit_also_enforces_allowlist(self, monkeypatch, base_settings):
|
||||
"""A previously-cached token must still be re-checked against the allowlist."""
|
||||
import hashlib
|
||||
|
||||
monkeypatch.setenv("ALLOWED_MGMT_CLIENT", "astrolabe")
|
||||
verifier = UnifiedTokenVerifier(base_settings)
|
||||
|
||||
token = "cached-token"
|
||||
cache_key = f"mgmt:{hashlib.sha256(token.encode()).hexdigest()}"
|
||||
verifier._token_cache[cache_key] = (
|
||||
{
|
||||
"sub": "testuser",
|
||||
"scope": "openid",
|
||||
"client_id": "not-allowlisted",
|
||||
},
|
||||
time.time() + 3600,
|
||||
)
|
||||
|
||||
result = await verifier.verify_token_for_management_api(token)
|
||||
assert result is None
|
||||
|
||||
Reference in New Issue
Block a user