From b128780aac3936bccc591ccb435a5ff90956d258 Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Wed, 17 Jun 2026 19:00:30 +0200 Subject: [PATCH] fix(auth): tighten userinfo-token cache TTL and metric labelling MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address claude-review round 1 on #919: - Security: userinfo responses carry no `exp`, so userinfo-validated opaque tokens were cached for the 1h default TTL — a revoked/expired token could be honored for up to an hour. Cache them for `userinfo_cache_ttl` (5 min) instead, and document the bounded-staleness window in the docstring. - Metrics: when introspection AND userinfo both fail, record ("introspect","invalid") + ("userinfo","invalid") separately and set validation_method="userinfo" before the userinfo call so a userinfo exception caught by the outer handler is attributed correctly. - Style: use the hasattr(...) + truthy pattern for userinfo_uri, matching the introspection block above it. - Tests: cache-hit allowlist bypass for via-userinfo tokens; short-TTL assertion; userinfo timeout / connect-error / malformed-JSON fail-closed. Co-Authored-By: Claude Opus 4.8 (1M context) --- nextcloud_mcp_server/auth/unified_verifier.py | 36 ++++++++-- tests/unit/test_unified_verifier.py | 67 +++++++++++++++++++ 2 files changed, 97 insertions(+), 6 deletions(-) diff --git a/nextcloud_mcp_server/auth/unified_verifier.py b/nextcloud_mcp_server/auth/unified_verifier.py index 1afd5106..2023d5bd 100644 --- a/nextcloud_mcp_server/auth/unified_verifier.py +++ b/nextcloud_mcp_server/auth/unified_verifier.py @@ -79,8 +79,9 @@ class UnifiedTokenVerifier(TokenVerifier): # client, e.g. Astrolabe, which the Nextcloud oidc app's introspection # endpoint reports inactive cross-client). A 200 from userinfo proves # the bearer is a live token for its user regardless of issuing client. - self.userinfo_uri: str | None = getattr(settings, "userinfo_uri", None) - if self.userinfo_uri: + self.userinfo_uri: str | None = None + if hasattr(settings, "userinfo_uri") and settings.userinfo_uri: + self.userinfo_uri = settings.userinfo_uri logger.info( "Userinfo token validation fallback enabled: %s", self.userinfo_uri ) @@ -99,6 +100,10 @@ class UnifiedTokenVerifier(TokenVerifier): # Token cache: token_hash -> (userinfo, expiry_timestamp) self._token_cache: dict[str, tuple[dict[str, Any], float]] = {} self.cache_ttl = 3600 # 1 hour default + # Userinfo responses carry no token `exp`, so userinfo-validated opaque + # tokens fall back to a TTL here. Keep it short: a revoked/expired token + # is still honored from cache until this window elapses (no exp to gate). + self.userinfo_cache_ttl = 300 # 5 minutes # NOTE: ALLOWED_MCP_CLIENTS and ALLOWED_MGMT_CLIENT are currently separate # env vars to keep the MCP-route and management-API auth surfaces @@ -370,16 +375,19 @@ class UnifiedTokenVerifier(TokenVerifier): 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. + validation_method = "userinfo" payload = await self._validate_via_userinfo(token) if payload: - validation_method = "userinfo" record_oauth_token_validation("userinfo", "valid") else: - record_oauth_token_validation("introspect", "invalid") + record_oauth_token_validation("userinfo", "invalid") return None # Check payload is valid @@ -593,6 +601,11 @@ class UnifiedTokenVerifier(TokenVerifier): and the management-API allowlist is relaxed for this path (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 + than the 1-hour default. A revoked/expired opaque token may therefore be + honored from cache for up to that window before re-validation. + Args: token: Bearer token to validate. @@ -688,8 +701,19 @@ class UnifiedTokenVerifier(TokenVerifier): # Extract expiration exp = payload.get("exp") if not exp: - logger.warning("No 'exp' claim in token, using default TTL") - exp = int(time.time() + self.cache_ttl) + # 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"): + ttl = self.userinfo_cache_ttl + logger.warning( + "Token validated via userinfo has no 'exp'; caching for %ss only", + ttl, + ) + else: + ttl = self.cache_ttl + logger.warning("No 'exp' claim in token, using default TTL") + exp = int(time.time() + ttl) # Cache the result with the provided key userinfo = { diff --git a/tests/unit/test_unified_verifier.py b/tests/unit/test_unified_verifier.py index 6f35e2c2..ca44fa34 100644 --- a/tests/unit/test_unified_verifier.py +++ b/tests/unit/test_unified_verifier.py @@ -8,6 +8,7 @@ IdP connections. import time from unittest.mock import AsyncMock, MagicMock, patch +import httpx import jwt import pytest @@ -719,3 +720,69 @@ class TestUserinfoFallback: assert result is not None assert result.client_id == "astrolabe" userinfo_mock.assert_not_called() + + async def test_mgmt_cache_hit_also_bypasses_allowlist_for_userinfo_tokens( + 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 + + 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, + ) + + result = await verifier.verify_token_for_management_api(token) + assert result is not None + assert result.resource == "testuser" + + 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.""" + 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} + ) + assert access_token is not None + # Expiry should sit within the short userinfo window, well under 1h. + assert access_token.expires_at <= int(before + 300) + 2 + assert access_token.expires_at < int(before + verifier.cache_ttl) + + async def test_validate_via_userinfo_timeout(self, userinfo_settings): + verifier = UnifiedTokenVerifier(userinfo_settings) + with patch.object( + verifier.http_client, + "get", + AsyncMock(side_effect=httpx.TimeoutException("timeout")), + ): + result = await verifier._validate_via_userinfo("opaque-token") + assert result is None + + async def test_validate_via_userinfo_request_error(self, userinfo_settings): + verifier = UnifiedTokenVerifier(userinfo_settings) + with patch.object( + verifier.http_client, + "get", + AsyncMock(side_effect=httpx.ConnectError("boom")), + ): + result = await verifier._validate_via_userinfo("opaque-token") + assert result is None + + async def test_validate_via_userinfo_malformed_json(self, userinfo_settings): + verifier = UnifiedTokenVerifier(userinfo_settings) + mock_resp = MagicMock() + mock_resp.status_code = 200 + mock_resp.json.side_effect = ValueError("not json") + with patch.object( + verifier.http_client, "get", AsyncMock(return_value=mock_resp) + ): + result = await verifier._validate_via_userinfo("opaque-token") + assert result is None