admin/mcp-nextcloud
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(auth): harden userinfo fallback (anti-forgery, SSRF guard, unconfigured-introspection)

Browse Source 
Address claude-review round 2 on #919:

- Anti-forgery: the `_auth_via_userinfo` allowlist-bypass flag is now sourced
  ONLY from an explicit in-process `via_userinfo` argument (derived from how
  the token was validated), never from the IdP payload. The payload claim is
  stripped from the cached entry, so a malicious introspection/userinfo
  response can't forge the bypass. Added a regression test.
- SSRF (CWE-918): guard the userinfo_uri scheme (http/https) before the request
  — documents the trusted-source assumption and fails fast on misconfig.
- Introspection-unconfigured: only attempt introspection (and record its
  metric) when an introspection endpoint is configured; otherwise go straight
  to userinfo. Avoids mislabelled introspect-invalid metrics. Added a test.
- Tests: cache-hit test now seeds via a real first call (behavior, not cache
  internals) and asserts the network is probed once; short-TTL test uses the
  explicit via_userinfo arg; moved hashlib usage out.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Chris Coutinho
2026-06-17 20:04:54 +02:00
co-authored by Claude Opus 4.8
parent b128780aac
commit 8acfe9655b
2 changed files with 131 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files
nextcloud_mcp_server/auth/unified_verifier.py
+62 -26
View File
@@ -369,19 +369,23 @@ class UnifiedTokenVerifier(TokenVerifier):
record_oauth_token_validation("jwt", "invalid")
return None
else:
# Fall back to introspection for opaque tokens
validation_method = "introspect"
payload = await self._introspect_token(token)
if payload:
record_oauth_token_validation("introspect", "valid")
else:
record_oauth_token_validation("introspect", "invalid")
# Introspection can report opaque tokens minted for a
# *different* OIDC client (e.g. Astrolabe) as inactive even
# when they are live. Fall back to the userinfo endpoint,
# which validates any live bearer regardless of client.
# Update validation_method first so a userinfo exception
# caught by the outer handler is attributed correctly.
# Opaque token: try introspection first (only when configured),
# then fall back to userinfo. userinfo validates opaque tokens
# minted for a *different* OIDC client (e.g. Astrolabe) that
# introspection reports inactive cross-client.
payload = None
if self.introspection_uri:
validation_method = "introspect"
payload = await self._introspect_token(token)
if payload:
record_oauth_token_validation("introspect", "valid")
else:
record_oauth_token_validation("introspect", "invalid")
if payload is None:
# Introspection is unconfigured, or it reported the token
# inactive. Set validation_method first so a userinfo
# exception caught by the outer handler is attributed right.
validation_method = "userinfo"
payload = await self._validate_via_userinfo(token)
if payload:
@@ -400,8 +404,15 @@ class UnifiedTokenVerifier(TokenVerifier):
payload.get("sub"),
)
# Cache and return the token
return self._create_access_token_with_cache_key(token, payload, cache_key)
# Cache and return the token. via_userinfo is derived from how we
# actually validated — never from a payload claim (see
# _create_access_token_with_cache_key).
return self._create_access_token_with_cache_key(
token,
payload,
cache_key,
via_userinfo=(validation_method == "userinfo"),
)
except Exception as e:
logger.error("Management API token verification failed: %s", e)
@@ -597,9 +608,11 @@ class UnifiedTokenVerifier(TokenVerifier):
token for that user.
Unlike introspection, userinfo returns neither ``client_id`` nor
``scope``, so the returned payload is stamped with ``_auth_via_userinfo``
and the management-API allowlist is relaxed for this path (authorization
is still enforced per-user by every management endpoint).
``scope``. The caller signals this path via the ``via_userinfo`` argument
to :meth:`_create_access_token_with_cache_key` (never inferred from a
payload claim, so a malicious IdP response cannot forge it), and the
management-API allowlist is relaxed for it (authorization is still
enforced per-user by every management endpoint).
Security note — bounded staleness: userinfo carries no token ``exp``, so
a validated token is cached for ``userinfo_cache_ttl`` (5 min) rather
@@ -610,12 +623,19 @@ class UnifiedTokenVerifier(TokenVerifier):
token: Bearer token to validate.
Returns:
Userinfo claims (with ``_auth_via_userinfo`` set) if valid, else None.
Userinfo claims if valid, else None.
"""
if not self.userinfo_uri:
logger.debug("No userinfo endpoint configured")
return None
# userinfo_uri comes from the OIDC discovery document (admin-configured),
# not from user input — but guard the scheme anyway to satisfy SSRF
# scanners and to fail fast on a misconfigured endpoint.
if not self.userinfo_uri.startswith(("https://", "http://")):
logger.error("Refusing non-HTTP userinfo_uri: %s", self.userinfo_uri)
return None
try:
response = await self.http_client.get(
self.userinfo_uri,
@@ -645,9 +665,6 @@ class UnifiedTokenVerifier(TokenVerifier):
return None
logger.debug("Token validated via userinfo for user: %s", data.get("sub"))
# Stamp so verify_token_for_management_api can relax the client allowlist
# for tokens that legitimately lack a verifiable client_id.
data["_auth_via_userinfo"] = True
return data
def _create_access_token(
@@ -668,7 +685,12 @@ class UnifiedTokenVerifier(TokenVerifier):
return self._create_access_token_with_cache_key(token, payload, cache_key)
def _create_access_token_with_cache_key(
self, token: str, payload: dict[str, Any], cache_key: str
self,
token: str,
payload: dict[str, Any],
cache_key: str,
*,
via_userinfo: bool = False,
) -> AccessToken | None:
"""
Create AccessToken object from validated token payload with custom cache key.
@@ -677,6 +699,10 @@ class UnifiedTokenVerifier(TokenVerifier):
token: The bearer token
payload: Validated token payload
cache_key: Key to use for caching (allows separate caches for MCP vs management API)
via_userinfo: True when the token was validated via the userinfo
fallback. Sourced from the caller (how validation happened), never
from a payload claim — it gates the allowlist relaxation and the
short cache TTL, so it must not be forgeable by the IdP response.
Returns:
AccessToken object or None if required fields missing
@@ -704,7 +730,7 @@ class UnifiedTokenVerifier(TokenVerifier):
# userinfo-validated tokens never carry exp (userinfo describes the
# user, not the token). Cache them only briefly so a revoked/expired
# opaque token can't be honored for the full hour-long default TTL.
if payload.get("_auth_via_userinfo"):
if via_userinfo:
ttl = self.userinfo_cache_ttl
logger.warning(
"Token validated via userinfo has no 'exp'; caching for %ss only",
@@ -715,12 +741,22 @@ class UnifiedTokenVerifier(TokenVerifier):
logger.warning("No 'exp' claim in token, using default TTL")
exp = int(time.time() + ttl)
# Cache the result with the provided key
# Cache the result with the provided key. Drop any `_auth_via_userinfo`
# carried in the IdP payload — that flag is the allowlist-bypass signal
# and must originate ONLY from the trusted in-process `via_userinfo`
# argument, never from a (potentially malicious) introspection/userinfo
# claim.
userinfo = {
"sub": username,
"scope": scope_string,
**{k: v for k, v in payload.items() if k not in ["sub", "scope"]},
**{
k: v
for k, v in payload.items()
if k not in ("sub", "scope", "_auth_via_userinfo")
},
}
if via_userinfo:
userinfo["_auth_via_userinfo"] = True
self._token_cache[cache_key] = (userinfo, exp)
return AccessToken(
tests/unit/test_unified_verifier.py
+69 -16
View File
@@ -643,7 +643,6 @@ class TestUserinfoFallback:
result = await verifier._validate_via_userinfo("opaque-token")
assert result is not None
assert result["sub"] == "testuser"
assert result["_auth_via_userinfo"] is True
async def test_validate_via_userinfo_non_200(self, userinfo_settings):
verifier = UnifiedTokenVerifier(userinfo_settings)
@@ -685,7 +684,7 @@ class TestUserinfoFallback:
patch.object(
verifier,
"_validate_via_userinfo",
AsyncMock(return_value={"sub": "testuser", "_auth_via_userinfo": True}),
AsyncMock(return_value={"sub": "testuser"}),
),
):
result = await verifier.verify_token_for_management_api("opaque-token-123")
@@ -694,6 +693,54 @@ class TestUserinfoFallback:
assert result.resource == "testuser"
assert result.client_id == "" # userinfo provides no client_id
async def test_introspection_cannot_forge_userinfo_bypass(
self, monkeypatch, userinfo_settings
):
"""A malicious `_auth_via_userinfo` claim in an introspection response
must NOT bypass the allowlist — the bypass flag is derived from how we
validated (validation_method), never from the IdP payload."""
monkeypatch.setenv("ALLOWED_MGMT_CLIENT", "astrolabe")
verifier = UnifiedTokenVerifier(userinfo_settings)
malicious = {
"sub": "testuser",
"client_id": "not-allowlisted",
"exp": int(time.time() + 3600),
"_auth_via_userinfo": True,
}
with patch.object(
verifier, "_introspect_token", AsyncMock(return_value=malicious)
):
result = await verifier.verify_token_for_management_api("opaque-evil")
assert result is None # allowlist still enforced; forged flag ignored
async def test_userinfo_used_when_introspection_unconfigured(
self, monkeypatch, base_settings
):
"""With no introspection endpoint but a userinfo endpoint, opaque tokens
go straight to userinfo (introspection is not even attempted)."""
monkeypatch.setenv("ALLOWED_MGMT_CLIENT", "astrolabe")
base_settings.introspection_uri = None
base_settings.userinfo_uri = "https://idp.example.com/userinfo"
verifier = UnifiedTokenVerifier(base_settings)
assert verifier.introspection_uri is None
introspect_mock = AsyncMock(return_value=None)
with (
patch.object(verifier, "_introspect_token", introspect_mock),
patch.object(
verifier,
"_validate_via_userinfo",
AsyncMock(return_value={"sub": "testuser"}),
),
):
result = await verifier.verify_token_for_management_api("opaque-x")
assert result is not None
assert result.resource == "testuser"
introspect_mock.assert_not_called() # skipped when unconfigured
async def test_mgmt_userinfo_not_called_when_introspection_succeeds(
self, monkeypatch, userinfo_settings
):
@@ -725,31 +772,37 @@ class TestUserinfoFallback:
self, monkeypatch, userinfo_settings
):
"""A second call (cache hit) with a via-userinfo token still bypasses
the allowlist — the stamp is preserved in the cached entry."""
import hashlib
the allowlist and is served from cache (no second network probe).
Seeds the cache via a real first call rather than constructing the cache
key by hand, so the test exercises behavior, not cache internals."""
monkeypatch.setenv("ALLOWED_MGMT_CLIENT", "astrolabe")
verifier = UnifiedTokenVerifier(userinfo_settings)
token = "opaque-token-cached"
cache_key = f"mgmt:{hashlib.sha256(token.encode()).hexdigest()}"
verifier._token_cache[cache_key] = (
{"sub": "testuser", "scope": "", "_auth_via_userinfo": True},
time.time() + 3600,
)
userinfo_mock = AsyncMock(return_value={"sub": "testuser"})
with (
patch.object(verifier, "_introspect_token", AsyncMock(return_value=None)),
patch.object(verifier, "_validate_via_userinfo", userinfo_mock),
):
first = await verifier.verify_token_for_management_api("opaque-cached")
second = await verifier.verify_token_for_management_api("opaque-cached")
result = await verifier.verify_token_for_management_api(token)
assert result is not None
assert result.resource == "testuser"
assert first is not None and second is not None
assert second.resource == "testuser"
# Second call served from cache — userinfo probed only once.
userinfo_mock.assert_awaited_once()
async def test_userinfo_token_cached_with_short_ttl(self, userinfo_settings):
"""userinfo tokens (no exp) get the short userinfo TTL, not the 1h default."""
"""userinfo tokens (no exp) get the short userinfo TTL, not the 1h default.
The short TTL is keyed off the explicit via_userinfo argument, not a
payload claim."""
verifier = UnifiedTokenVerifier(userinfo_settings)
verifier.userinfo_cache_ttl = 300
before = time.time()
access_token = verifier._create_access_token(
"opaque-token", {"sub": "testuser", "_auth_via_userinfo": True}
access_token = verifier._create_access_token_with_cache_key(
"opaque-token", {"sub": "testuser"}, "mgmt:test", via_userinfo=True
)
assert access_token is not None
# Expiry should sit within the short userinfo window, well under 1h.