Merge pull request #745 from cbcoutinho/feat/strict-auth-allowlists

feat(auth): drop test-client defaults, add ALLOWED_MGMT_CLIENT allowlist

tests/unit/test_client_registry.py

@@ -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"

tests/unit/test_unified_verifier.py

@@ -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