diff --git a/nextcloud_mcp_server/auth/unified_verifier.py b/nextcloud_mcp_server/auth/unified_verifier.py index 4aa314dd..3aafe2d0 100644 --- a/nextcloud_mcp_server/auth/unified_verifier.py +++ b/nextcloud_mcp_server/auth/unified_verifier.py @@ -218,6 +218,7 @@ class UnifiedTokenVerifier(TokenVerifier): else: del self._token_cache[cache_key] + from_cache = access_token is not None if access_token is None: oauth_token_cache_hits_total.labels(hit="false").inc() access_token = await self._verify_without_audience_check(token, cache_key) @@ -233,11 +234,21 @@ class UnifiedTokenVerifier(TokenVerifier): cached_entry = self._token_cache.get(cache_key) via_userinfo = bool(cached_entry and cached_entry[0].get("_auth_via_userinfo")) if via_userinfo: - logger.warning( - "Opaque token validated via userinfo endpoint; ALLOWED_MGMT_CLIENT " - "allowlist not enforced for user %s (per-user authorization applies)", - access_token.resource, - ) + # Warn once on fresh validation; subsequent cache-hit re-validations + # (frequent Astrolabe polling) log at DEBUG to avoid flooding. + if from_cache: + logger.debug( + "Opaque token (userinfo-validated) served from cache for " + "user %s; allowlist not enforced", + access_token.resource, + ) + else: + logger.warning( + "Opaque token validated via userinfo endpoint; " + "ALLOWED_MGMT_CLIENT allowlist not enforced for user %s " + "(per-user authorization applies)", + access_token.resource, + ) return access_token # Enforce ALLOWED_MGMT_CLIENT allowlist (fail-closed when unset) diff --git a/tests/unit/test_unified_verifier.py b/tests/unit/test_unified_verifier.py index f24e7084..7d00292a 100644 --- a/tests/unit/test_unified_verifier.py +++ b/tests/unit/test_unified_verifier.py @@ -809,6 +809,22 @@ class TestUserinfoFallback: assert access_token.expires_at <= int(before + 300) + 2 assert access_token.expires_at < int(before + verifier.cache_ttl) + async def test_userinfo_token_with_exp_uses_real_expiry(self, userinfo_settings): + """When userinfo (unusually) returns an exp, the real token expiry wins + over the short userinfo TTL.""" + verifier = UnifiedTokenVerifier(userinfo_settings) + verifier.userinfo_cache_ttl = 300 + real_exp = int(time.time() + 4000) # far beyond the 300s short TTL + + access_token = verifier._create_access_token_with_cache_key( + "opaque-token", + {"sub": "testuser", "exp": real_exp}, + "mgmt:test-exp", + via_userinfo=True, + ) + assert access_token is not None + assert access_token.expires_at == real_exp + async def test_validate_via_userinfo_timeout(self, userinfo_settings): verifier = UnifiedTokenVerifier(userinfo_settings) with patch.object(