Merge pull request #919 from cbcoutinho/fix/nx101294-opaque-token-support

fix(auth): validate opaque access tokens via userinfo fallback

nextcloud_mcp_server/auth/unified_verifier.py

@@ -75,6 +75,17 @@ class UnifiedTokenVerifier(TokenVerifier):
             self.introspection_uri = settings.introspection_uri
             logger.info("Token introspection enabled: %s", self.introspection_uri)
 
+        # Userinfo fallback (for opaque tokens minted for a *different* OIDC
+        # 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 = 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
+            )
+
         # Build list of valid issuers (internal + public may differ in Docker)
         # AS proxy obtains tokens via internal URL (e.g. http://app:80), while
         # NEXTCLOUD_PUBLIC_ISSUER_URL is the browser-facing URL (e.g. http://localhost:8080)
@@ -89,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
@@ -100,10 +115,20 @@ class UnifiedTokenVerifier(TokenVerifier):
             if entry.strip()
         )
         if not self._allowed_mgmt_clients:
-            logger.warning(
+            if self.userinfo_uri:
-                "ALLOWED_MGMT_CLIENT is unset or empty: management API will reject "
+                # An empty allowlist is NOT a kill switch when userinfo is
-                "all requests until configured."
+                # configured: opaque tokens validated via the userinfo fallback
-            )
+                # bypass ALLOWED_MGMT_CLIENT (per-user authz still applies).
+                logger.warning(
+                    "ALLOWED_MGMT_CLIENT is unset: JWT/introspection management "
+                    "tokens will be rejected, but opaque tokens may still be "
+                    "accepted via the userinfo fallback."
+                )
+            else:
+                logger.warning(
+                    "ALLOWED_MGMT_CLIENT is unset or empty: management API will "
+                    "reject all requests until configured."
+                )
         else:
             logger.info(
                 "Management API allowlist: %s", sorted(self._allowed_mgmt_clients)
@@ -160,6 +185,12 @@ class UnifiedTokenVerifier(TokenVerifier):
            - Verifies token signature against Nextcloud's JWKS (cryptographic proof)
            - Verifies token is not expired
            - Extracts user identity from validated token claims
+           - NOTE: for opaque cross-client tokens (e.g. Astrolabe) that
+             introspection reports inactive, authentication falls back to the
+             userinfo endpoint — a live IdP liveness check (200 + ``sub``)
+             rather than local JWKS/expiry verification. Such tokens are stamped
+             ``_auth_via_userinfo`` and bypass the client allowlist (step 4);
+             per-user authorization (step 2) remains the security gate.
 
         2. **Authorization layer** (management API endpoints):
            - EVERY endpoint verifies: token.sub == requested_resource_owner
@@ -203,6 +234,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)
@@ -210,6 +242,35 @@ class UnifiedTokenVerifier(TokenVerifier):
         if access_token is None:
             return None
 
+        # Opaque tokens validated via the userinfo fallback carry no verifiable
+        # client_id, so the ALLOWED_MGMT_CLIENT allowlist cannot apply. Such
+        # tokens are stamped with ``_auth_via_userinfo`` in the cache; for them
+        # we rely on the per-user authorization every management endpoint
+        # enforces (token sub == requested resource owner).
+        # Recover the via-userinfo flag from the cache entry. On a cache miss
+        # this is the entry _verify_without_audience_check just wrote (no await
+        # between that write and this read, so it is always present); on a cache
+        # hit it was written by an earlier call.
+        cached_entry = self._token_cache.get(cache_key)
+        via_userinfo = bool(cached_entry and cached_entry[0].get("_auth_via_userinfo"))
+        if via_userinfo:
+            # 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)
         token_client_id = access_token.client_id
         if not token_client_id or token_client_id not in self._allowed_mgmt_clients:
@@ -339,18 +400,44 @@ class UnifiedTokenVerifier(TokenVerifier):
                     record_oauth_token_validation("jwt", "invalid")
                     return None
             else:
-                # Fall back to introspection for opaque tokens
+                # Opaque token: try introspection first (only when configured),
-                validation_method = "introspect"
+                # then fall back to userinfo. userinfo validates opaque tokens
-                payload = await self._introspect_token(token)
+                # minted for a *different* OIDC client (e.g. Astrolabe) that
-                if payload:
+                # introspection reports inactive cross-client.
-                    record_oauth_token_validation("introspect", "valid")
+                payload = None
-                else:
+                if self.introspection_uri:
-                    record_oauth_token_validation("introspect", "invalid")
+                    validation_method = "introspect"
+                    payload = await self._introspect_token(token)
+                    if payload:
+                        record_oauth_token_validation("introspect", "valid")
+                    else:
+                        record_oauth_token_validation("introspect", "invalid")
+
+                # Fall through to userinfo when introspection is unconfigured or
+                # returned None. NOTE: _introspect_token returns None for BOTH an
+                # active=false response (the nx101294 cross-client case we must
+                # handle) AND a network/timeout error — both reach userinfo here.
+                # That is safe: userinfo is itself an authoritative live check (a
+                # revoked/invalid token gets a 401), so a flapping introspection
+                # endpoint cannot cause an invalid token to be accepted.
+                if payload is None and self.userinfo_uri:
+                    # Set 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:
+                        record_oauth_token_validation("userinfo", "valid")
+                    else:
+                        record_oauth_token_validation("userinfo", "invalid")
+                        return None
+
+                if payload is None:
+                    # No validator was configured, or none succeeded. Don't record
+                    # a userinfo failure metric when userinfo was never attempted.
                     return None
 
-            # Check payload is valid
+            # Both branches above either set a populated payload or have already
-            if not payload:
+            # returned None, so payload is guaranteed truthy here.
-                return None
 
             # Skip audience validation - any valid Nextcloud token is accepted
             logger.debug(
@@ -358,8 +445,15 @@ class UnifiedTokenVerifier(TokenVerifier):
                 payload.get("sub"),
             )
 
-            # Cache and return the token
+            # Cache and return the token. via_userinfo is derived from how we
-            return self._create_access_token_with_cache_key(token, payload, cache_key)
+            # 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)
@@ -545,6 +639,83 @@ class UnifiedTokenVerifier(TokenVerifier):
             logger.error("Unexpected error during token introspection: %s", e)
             return None
 
+    async def _validate_via_userinfo(self, token: str) -> dict[str, Any] | None:
+        """Validate an opaque token by calling the OIDC userinfo endpoint.
+
+        Fallback for opaque access tokens that the Nextcloud ``oidc`` app's
+        introspection endpoint reports ``active=false`` cross-client (i.e.
+        tokens minted for a *different* client such as Astrolabe). A 200 from
+        userinfo with a ``sub`` claim proves the bearer is a valid, unexpired
+        token for that user.
+
+        Unlike introspection, userinfo returns neither ``client_id`` nor
+        ``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).
+
+        Caution: userinfo-validated tokens carry **empty scopes**. Callers must
+        not gate management endpoints on scopes for this path (e.g. a future
+        ``@require_scopes``) or they would silently reject valid cross-client
+        tokens; the per-user ``sub`` check is the authorization gate.
+
+        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.
+
+        Returns:
+            Userinfo claims if valid, else None.
+        """
+        # Defensive: the management-API caller already gates on
+        # self.userinfo_uri before invoking this, but the guard keeps the method
+        # safe to call directly (e.g. in unit tests).
+        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,
+                headers={"Authorization": f"Bearer {token}"},
+            )
+        except httpx.TimeoutException:
+            logger.error("Timeout while validating token via userinfo")
+            return None
+        except httpx.RequestError as e:
+            logger.error("Network error while validating token via userinfo: %s", e)
+            return None
+
+        if response.status_code != 200:
+            logger.warning(
+                "Userinfo token validation failed: HTTP %s", response.status_code
+            )
+            return None
+
+        try:
+            data = response.json()
+        except Exception as e:
+            logger.error("Failed to parse userinfo response: %s", e)
+            return None
+
+        if not data.get("sub"):
+            logger.warning("Userinfo response missing 'sub' claim")
+            return None
+
+        logger.debug("Token validated via userinfo for user: %s", data.get("sub"))
+        return data
+
     def _create_access_token(
         self, token: str, payload: dict[str, Any]
     ) -> AccessToken | None:
@@ -563,7 +734,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.
@@ -572,6 +748,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
@@ -596,15 +776,39 @@ class UnifiedTokenVerifier(TokenVerifier):
         # Extract expiration
         exp = payload.get("exp")
         if not exp:
-            logger.warning("No 'exp' claim in token, using default TTL")
+            # userinfo-validated tokens never carry exp (userinfo describes the
-            exp = int(time.time() + self.cache_ttl)
+            # 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 via_userinfo:
+                ttl = self.userinfo_cache_ttl
+                # userinfo never returns exp, so this fires on every fresh
+                # userinfo validation — keep it at DEBUG (the bounded-staleness
+                # window is documented on _validate_via_userinfo).
+                logger.debug(
+                    "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
+        # 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

@@ -8,6 +8,7 @@ IdP connections.
 import time
 from unittest.mock import AsyncMock, MagicMock, patch
 
+import httpx
 import jwt
 import pytest
 
@@ -616,3 +617,312 @@ class TestManagementApiAllowlist:
 
         result = await verifier.verify_token_for_management_api(token)
         assert result is None
+
+
+class TestUserinfoFallback:
+    """Opaque tokens that introspection reports inactive fall back to userinfo.
+
+    Covers the nx101294 case: the Astrolabe OIDC client issues opaque access
+    tokens that Nextcloud's oidc app introspection reports active=false
+    cross-client. The userinfo endpoint validates them regardless of client.
+    """
+
+    @pytest.fixture
+    def userinfo_settings(self, base_settings):
+        base_settings.userinfo_uri = "https://idp.example.com/userinfo"
+        return base_settings
+
+    async def test_validate_via_userinfo_success(self, userinfo_settings):
+        verifier = UnifiedTokenVerifier(userinfo_settings)
+        mock_resp = MagicMock()
+        mock_resp.status_code = 200
+        mock_resp.json.return_value = {"sub": "testuser"}
+        with patch.object(
+            verifier.http_client, "get", AsyncMock(return_value=mock_resp)
+        ):
+            result = await verifier._validate_via_userinfo("opaque-token")
+        assert result is not None
+        assert result["sub"] == "testuser"
+
+    async def test_validate_via_userinfo_non_200(self, userinfo_settings):
+        verifier = UnifiedTokenVerifier(userinfo_settings)
+        mock_resp = MagicMock()
+        mock_resp.status_code = 401
+        with patch.object(
+            verifier.http_client, "get", AsyncMock(return_value=mock_resp)
+        ):
+            result = await verifier._validate_via_userinfo("opaque-token")
+        assert result is None
+
+    async def test_validate_via_userinfo_missing_sub(self, userinfo_settings):
+        verifier = UnifiedTokenVerifier(userinfo_settings)
+        mock_resp = MagicMock()
+        mock_resp.status_code = 200
+        mock_resp.json.return_value = {"name": "no sub claim"}
+        with patch.object(
+            verifier.http_client, "get", AsyncMock(return_value=mock_resp)
+        ):
+            result = await verifier._validate_via_userinfo("opaque-token")
+        assert result is None
+
+    async def test_validate_via_userinfo_rejects_non_http_scheme(
+        self, userinfo_settings
+    ):
+        """A non-http(s) userinfo_uri is refused before any request (SSRF guard)."""
+        verifier = UnifiedTokenVerifier(userinfo_settings)
+        verifier.userinfo_uri = "ftp://evil/userinfo"
+        get_mock = AsyncMock()
+        with patch.object(verifier.http_client, "get", get_mock):
+            result = await verifier._validate_via_userinfo("opaque-token")
+        assert result is None
+        get_mock.assert_not_called()
+
+    async def test_validate_via_userinfo_not_configured(self, base_settings):
+        base_settings.userinfo_uri = None
+        verifier = UnifiedTokenVerifier(base_settings)
+        result = await verifier._validate_via_userinfo("opaque-token")
+        assert result is None
+
+    async def test_mgmt_opaque_userinfo_fallback_accepted_despite_allowlist(
+        self, monkeypatch, userinfo_settings
+    ):
+        """Introspection inactive -> userinfo validates -> accepted even though
+        no client_id matches the allowlist (per-user authorization applies)."""
+        monkeypatch.setenv("ALLOWED_MGMT_CLIENT", "astrolabe")
+        verifier = UnifiedTokenVerifier(userinfo_settings)
+
+        with (
+            patch.object(verifier, "_introspect_token", AsyncMock(return_value=None)),
+            patch.object(
+                verifier,
+                "_validate_via_userinfo",
+                AsyncMock(return_value={"sub": "testuser"}),
+            ),
+        ):
+            result = await verifier.verify_token_for_management_api("opaque-token-123")
+
+        assert result is not None
+        assert result.resource == "testuser"
+        assert result.client_id == ""  # userinfo provides no client_id
+        # Contract: userinfo tokens carry empty scopes — management endpoints
+        # must not gate on scopes for this path (per-user authz is the gate).
+        assert result.scopes == []
+
+    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_introspection_timeout_falls_through_to_userinfo(
+        self, monkeypatch, userinfo_settings
+    ):
+        """A real introspection timeout (caught inside _introspect_token, which
+        returns None) falls through to userinfo — the authoritative live check —
+        exercising the whole chain, not just a mocked _introspect_token."""
+        monkeypatch.setenv("ALLOWED_MGMT_CLIENT", "astrolabe")
+        verifier = UnifiedTokenVerifier(userinfo_settings)
+
+        userinfo_resp = MagicMock()
+        userinfo_resp.status_code = 200
+        userinfo_resp.json.return_value = {"sub": "testuser"}
+        with (
+            patch.object(
+                verifier.http_client,
+                "post",
+                AsyncMock(side_effect=httpx.TimeoutException("introspect down")),
+            ),
+            patch.object(
+                verifier.http_client, "get", AsyncMock(return_value=userinfo_resp)
+            ),
+        ):
+            result = await verifier.verify_token_for_management_api("opaque-timeout")
+
+        assert result is not None
+        assert result.resource == "testuser"
+
+    async def test_mcp_path_does_not_use_userinfo_for_opaque_token(
+        self, userinfo_settings
+    ):
+        """The userinfo fallback applies only to the management API path, never
+        the MCP-audience path — an opaque token there is still rejected."""
+        verifier = UnifiedTokenVerifier(userinfo_settings)
+        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),
+        ):
+            result = await verifier.verify_token("opaque-astrolabe-token")
+        assert result is None
+        userinfo_mock.assert_not_called()
+
+    async def test_opaque_rejected_when_no_validators_configured(self, base_settings):
+        """With neither introspection nor userinfo configured, an opaque token is
+        rejected without recording a misleading userinfo-failure metric."""
+        base_settings.introspection_uri = None
+        base_settings.userinfo_uri = None
+        verifier = UnifiedTokenVerifier(base_settings)
+        assert verifier.introspection_uri is None
+        assert verifier.userinfo_uri is None
+
+        result = await verifier._verify_without_audience_check(
+            "opaque-no-validator", "mgmt:none"
+        )
+        assert result is None
+
+    async def test_mgmt_userinfo_not_called_when_introspection_succeeds(
+        self, monkeypatch, userinfo_settings
+    ):
+        monkeypatch.setenv("ALLOWED_MGMT_CLIENT", "astrolabe")
+        verifier = UnifiedTokenVerifier(userinfo_settings)
+
+        introspection_payload = {
+            "sub": "testuser",
+            "client_id": "astrolabe",
+            "scope": "openid",
+            "exp": int(time.time() + 3600),
+        }
+        userinfo_mock = AsyncMock(return_value={"sub": "x", "_auth_via_userinfo": True})
+        with (
+            patch.object(
+                verifier,
+                "_introspect_token",
+                AsyncMock(return_value=introspection_payload),
+            ),
+            patch.object(verifier, "_validate_via_userinfo", userinfo_mock),
+        ):
+            result = await verifier.verify_token_for_management_api("opaque-token-123")
+
+        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 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)
+
+        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")
+
+        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()
+
+    def test_userinfo_token_cached_with_short_ttl(self, userinfo_settings):
+        """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_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.
+        assert access_token.expires_at <= int(before + 300) + 2
+        assert access_token.expires_at < int(before + verifier.cache_ttl)
+
+    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(
+            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