From 0294a99cd4c195b8f321eb9ef2ad33ede4709d0d Mon Sep 17 00:00:00 2001
From: Chris Coutinho <chris@coutinho.io>
Date: Wed, 17 Jun 2026 18:53:47 +0200
Subject: [PATCH 01/11] fix(auth): validate opaque access tokens via userinfo
 fallback
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The management API (used by the Astrolabe PHP app for /api/v1/apps and
/api/v1/webhooks) only accepted JWT access tokens. Opaque tokens were
sent to Nextcloud's oidc introspection endpoint, which returns
`active: false` for tokens minted for a *different* OIDC client (e.g.
Astrolabe) even when they are live — so every call 401'd. This surfaced
on the nx101294 tenant: webhook setup failed and the webhook-preset UI
(including the Files preset) showed empty, because getWebhookPresets
errors out before its `files`-always-available filter runs.

Add a userinfo-endpoint fallback in UnifiedTokenVerifier: when
introspection reports an opaque token inactive, validate it against the
discovered userinfo_endpoint (a 200 with a `sub` proves a live bearer
regardless of issuing client). userinfo returns no client_id/scope, so
such tokens are stamped `_auth_via_userinfo` and the ALLOWED_MGMT_CLIENT
allowlist is relaxed for that path only — authorization is still
enforced per-user (token sub == requested resource owner) by every
management endpoint. JWT and introspection paths are unchanged and still
enforce the allowlist.

Also bumps the astrolabe submodule to 0.29.0 (the deployed version that
exhibits the issue).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 nextcloud_mcp_server/auth/unified_verifier.py |  96 +++++++++++++++-
 tests/unit/test_unified_verifier.py           | 103 ++++++++++++++++++
 third_party/astrolabe                         |   2 +-
 3 files changed, 198 insertions(+), 3 deletions(-)

diff --git a/nextcloud_mcp_server/auth/unified_verifier.py b/nextcloud_mcp_server/auth/unified_verifier.py
index 8dc83b2a..1afd5106 100644
--- a/nextcloud_mcp_server/auth/unified_verifier.py
+++ b/nextcloud_mcp_server/auth/unified_verifier.py
@@ -75,6 +75,16 @@ 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 = getattr(settings, "userinfo_uri", None)
+        if self.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)
@@ -210,6 +220,21 @@ 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).
+        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,
+            )
+            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:
@@ -345,8 +370,17 @@ class UnifiedTokenVerifier(TokenVerifier):
                 if payload:
                     record_oauth_token_validation("introspect", "valid")
                 else:
-                    record_oauth_token_validation("introspect", "invalid")
-                    return None
+                    # 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.
+                    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")
+                        return None
 
             # Check payload is valid
             if not payload:
@@ -545,6 +579,64 @@ 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``, 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).
+
+        Args:
+            token: Bearer token to validate.
+
+        Returns:
+            Userinfo claims (with ``_auth_via_userinfo`` set) if valid, else None.
+        """
+        if not self.userinfo_uri:
+            logger.debug("No userinfo endpoint configured")
+            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"))
+        # 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(
         self, token: str, payload: dict[str, Any]
     ) -> AccessToken | None:
diff --git a/tests/unit/test_unified_verifier.py b/tests/unit/test_unified_verifier.py
index 70ca88c5..6f35e2c2 100644
--- a/tests/unit/test_unified_verifier.py
+++ b/tests/unit/test_unified_verifier.py
@@ -616,3 +616,106 @@ 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"
+        assert result["_auth_via_userinfo"] is True
+
+    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_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", "_auth_via_userinfo": True}),
+            ),
+        ):
+            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
+
+    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()
diff --git a/third_party/astrolabe b/third_party/astrolabe
index 5fc20a78..a8bb3c4e 160000
--- a/third_party/astrolabe
+++ b/third_party/astrolabe
@@ -1 +1 @@
-Subproject commit 5fc20a78d049670b7e298cf5487b27c7d63b028a
+Subproject commit a8bb3c4e03f230fabe35f513f2bc6ca2231cddbf

From b128780aac3936bccc591ccb435a5ff90956d258 Mon Sep 17 00:00:00 2001
From: Chris Coutinho <chris@coutinho.io>
Date: Wed, 17 Jun 2026 19:00:30 +0200
Subject: [PATCH 02/11] 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) <noreply@anthropic.com>
---
 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

From 8acfe9655b3022b00612509865ff65a84501a7bb Mon Sep 17 00:00:00 2001
From: Chris Coutinho <chris@coutinho.io>
Date: Wed, 17 Jun 2026 20:04:54 +0200
Subject: [PATCH 03/11] fix(auth): harden userinfo fallback (anti-forgery, SSRF
 guard, unconfigured-introspection)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

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>
---
 nextcloud_mcp_server/auth/unified_verifier.py | 88 +++++++++++++------
 tests/unit/test_unified_verifier.py           | 85 ++++++++++++++----
 2 files changed, 131 insertions(+), 42 deletions(-)

diff --git a/nextcloud_mcp_server/auth/unified_verifier.py b/nextcloud_mcp_server/auth/unified_verifier.py
index 2023d5bd..4aa314dd 100644
--- a/nextcloud_mcp_server/auth/unified_verifier.py
+++ b/nextcloud_mcp_server/auth/unified_verifier.py
@@ -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(
diff --git a/tests/unit/test_unified_verifier.py b/tests/unit/test_unified_verifier.py
index ca44fa34..f24e7084 100644
--- a/tests/unit/test_unified_verifier.py
+++ b/tests/unit/test_unified_verifier.py
@@ -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.

From bafe82c897039aa8602b7bcf4ed4e643ae812e38 Mon Sep 17 00:00:00 2001
From: Chris Coutinho <chris@coutinho.io>
Date: Wed, 17 Jun 2026 20:11:52 +0200
Subject: [PATCH 04/11] fix(auth): quiet cache-hit userinfo log, test real-exp
 userinfo path
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Address claude-review round 3 on #919:
- Log spam: the userinfo allowlist-relaxation notice fired at WARNING on every
  request (incl. cache hits — frequent Astrolabe polling). Warn once on fresh
  validation; cache-hit re-validations now log at DEBUG.
- Test: add coverage for a userinfo response that DOES carry `exp` — the real
  token expiry must win over the short userinfo TTL.

Not changed:
- USERINFO_URI auto-discovery: already auto-populated from the OIDC discovery
  document in app.py (settings.userinfo_uri = discovery["userinfo_endpoint"],
  mirroring jwks_uri/introspection_uri), so OIDC_DISCOVERY_URL deployments need
  no extra env var. The reviewer's note only inspected config.py.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 nextcloud_mcp_server/auth/unified_verifier.py | 21 ++++++++++++++-----
 tests/unit/test_unified_verifier.py           | 16 ++++++++++++++
 2 files changed, 32 insertions(+), 5 deletions(-)

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(

From ed32519563dc0cdc71262de9a76d2dac5c11514d Mon Sep 17 00:00:00 2001
From: Chris Coutinho <chris@coutinho.io>
Date: Wed, 17 Jun 2026 20:18:21 +0200
Subject: [PATCH 05/11] fix(auth): document introspection-error fall-through,
 drop misleading userinfo metric
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Address claude-review round 4 on #919:
- Functional concern: document that _introspect_token returns None for both an
  active=false response (the cross-client case we must handle) AND a network
  error, so both fall through to userinfo. This is safe — userinfo is itself an
  authoritative live check, so a flapping introspection endpoint can't cause an
  invalid token to be accepted.
- Observability nit: only record a ("userinfo", ...) metric when userinfo was
  actually attempted (userinfo_uri configured); a no-validators-configured
  opaque token now returns None without a misleading userinfo-failure metric.
  Added test_opaque_rejected_when_no_validators_configured.
- Added a comment on the post-validation cache re-read explaining why the entry
  is always present (write-then-read with no await between).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 nextcloud_mcp_server/auth/unified_verifier.py | 23 +++++++++++++++----
 tests/unit/test_unified_verifier.py           | 14 +++++++++++
 2 files changed, 33 insertions(+), 4 deletions(-)

diff --git a/nextcloud_mcp_server/auth/unified_verifier.py b/nextcloud_mcp_server/auth/unified_verifier.py
index 3aafe2d0..915e4ff3 100644
--- a/nextcloud_mcp_server/auth/unified_verifier.py
+++ b/nextcloud_mcp_server/auth/unified_verifier.py
@@ -231,6 +231,10 @@ class UnifiedTokenVerifier(TokenVerifier):
         # 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:
@@ -393,10 +397,16 @@ class UnifiedTokenVerifier(TokenVerifier):
                     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.
+                # 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:
@@ -405,6 +415,11 @@ class UnifiedTokenVerifier(TokenVerifier):
                         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
             if not payload:
                 return None
diff --git a/tests/unit/test_unified_verifier.py b/tests/unit/test_unified_verifier.py
index 7d00292a..c124d1bc 100644
--- a/tests/unit/test_unified_verifier.py
+++ b/tests/unit/test_unified_verifier.py
@@ -741,6 +741,20 @@ class TestUserinfoFallback:
         assert result.resource == "testuser"
         introspect_mock.assert_not_called()  # skipped when unconfigured
 
+    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
     ):

From e1e9c9b918fd766b800038161f332a73f64ff23f Mon Sep 17 00:00:00 2001
From: Chris Coutinho <chris@coutinho.io>
Date: Wed, 17 Jun 2026 20:23:16 +0200
Subject: [PATCH 06/11] fix(auth): quiet per-validation userinfo TTL log; test
 introspection-timeout fall-through
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Address claude-review round 5 on #919:
- The "userinfo has no exp; caching for Ns only" log fired on every fresh
  userinfo validation (userinfo never returns exp) — downgrade WARNING → DEBUG;
  the bounded-staleness window is already documented on _validate_via_userinfo.
- Add test_introspection_timeout_falls_through_to_userinfo: drives a real
  introspection timeout (httpx.TimeoutException on the POST, caught inside
  _introspect_token → None) through to a successful userinfo validation,
  pinning the documented error fall-through end to end.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 nextcloud_mcp_server/auth/unified_verifier.py |  5 +++-
 tests/unit/test_unified_verifier.py           | 27 +++++++++++++++++++
 2 files changed, 31 insertions(+), 1 deletion(-)

diff --git a/nextcloud_mcp_server/auth/unified_verifier.py b/nextcloud_mcp_server/auth/unified_verifier.py
index 915e4ff3..520ad375 100644
--- a/nextcloud_mcp_server/auth/unified_verifier.py
+++ b/nextcloud_mcp_server/auth/unified_verifier.py
@@ -758,7 +758,10 @@ class UnifiedTokenVerifier(TokenVerifier):
             # opaque token can't be honored for the full hour-long default TTL.
             if via_userinfo:
                 ttl = self.userinfo_cache_ttl
-                logger.warning(
+                # 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,
                 )
diff --git a/tests/unit/test_unified_verifier.py b/tests/unit/test_unified_verifier.py
index c124d1bc..04ac13cd 100644
--- a/tests/unit/test_unified_verifier.py
+++ b/tests/unit/test_unified_verifier.py
@@ -741,6 +741,33 @@ class TestUserinfoFallback:
         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_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."""

From bc6595b139dcf4b1c02552b3b3c46d201f5c502a Mon Sep 17 00:00:00 2001
From: Chris Coutinho <chris@coutinho.io>
Date: Wed, 17 Jun 2026 20:29:20 +0200
Subject: [PATCH 07/11] test(auth): make sync userinfo tests def; note
 defensive userinfo guard
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Address claude-review round 6 (LGTM) nits on #919:
- test_userinfo_token_cached_with_short_ttl and
  test_userinfo_token_with_exp_uses_real_expiry call only the sync
  _create_access_token_with_cache_key — declare them as plain def (no await).
- Comment the userinfo_uri guard in _validate_via_userinfo as defensive /
  direct-call support (the management caller already gates on userinfo_uri).

Left as-is: the hasattr(settings, "userinfo_uri") guard — kept to mirror the
adjacent introspection_uri block (consistency requested in round 2). The
_verify_mcp_audience metric-when-unconfigured note is a pre-existing, out-of-
scope item for a follow-up.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 nextcloud_mcp_server/auth/unified_verifier.py | 3 +++
 tests/unit/test_unified_verifier.py           | 4 ++--
 2 files changed, 5 insertions(+), 2 deletions(-)

diff --git a/nextcloud_mcp_server/auth/unified_verifier.py b/nextcloud_mcp_server/auth/unified_verifier.py
index 520ad375..b9a6b873 100644
--- a/nextcloud_mcp_server/auth/unified_verifier.py
+++ b/nextcloud_mcp_server/auth/unified_verifier.py
@@ -651,6 +651,9 @@ class UnifiedTokenVerifier(TokenVerifier):
         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
diff --git a/tests/unit/test_unified_verifier.py b/tests/unit/test_unified_verifier.py
index 04ac13cd..e34c3b7b 100644
--- a/tests/unit/test_unified_verifier.py
+++ b/tests/unit/test_unified_verifier.py
@@ -833,7 +833,7 @@ class TestUserinfoFallback:
         # 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):
+    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
@@ -850,7 +850,7 @@ 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):
+    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)

From a53e6e77219701f2a753be0c1657f6859dbab052 Mon Sep 17 00:00:00 2001
From: Chris Coutinho <chris@coutinho.io>
Date: Wed, 17 Jun 2026 20:34:31 +0200
Subject: [PATCH 08/11] test(auth): cover userinfo SSRF scheme guard; note
 empty-scope caveat
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Address claude-review round 7 nits on #919:
- Add test_validate_via_userinfo_rejects_non_http_scheme — a non-http(s)
  userinfo_uri is refused before any request (covers the SSRF scheme guard).
- Docstring caution on _validate_via_userinfo: userinfo-validated tokens carry
  empty scopes, so management endpoints must not gate on scopes for this path.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 nextcloud_mcp_server/auth/unified_verifier.py |  5 +++++
 tests/unit/test_unified_verifier.py           | 12 ++++++++++++
 2 files changed, 17 insertions(+)

diff --git a/nextcloud_mcp_server/auth/unified_verifier.py b/nextcloud_mcp_server/auth/unified_verifier.py
index b9a6b873..50341040 100644
--- a/nextcloud_mcp_server/auth/unified_verifier.py
+++ b/nextcloud_mcp_server/auth/unified_verifier.py
@@ -640,6 +640,11 @@ class UnifiedTokenVerifier(TokenVerifier):
         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
diff --git a/tests/unit/test_unified_verifier.py b/tests/unit/test_unified_verifier.py
index e34c3b7b..95d4dd52 100644
--- a/tests/unit/test_unified_verifier.py
+++ b/tests/unit/test_unified_verifier.py
@@ -665,6 +665,18 @@ class TestUserinfoFallback:
             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)

From 7ef0e9d83bf864176c96924fa2488bad1417c91e Mon Sep 17 00:00:00 2001
From: Chris Coutinho <chris@coutinho.io>
Date: Wed, 17 Jun 2026 20:40:05 +0200
Subject: [PATCH 09/11] docs(auth): document userinfo path in security model;
 drop dead guard; pin MCP asymmetry
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Address claude-review round 8 on #919:
- Security-model docstring: note that opaque cross-client tokens authenticate
  via the userinfo liveness check (not JWKS/expiry) and bypass the client
  allowlist, with per-user authz as the gate.
- Remove the redundant `if not payload: return None` after the JWT/opaque
  branches (both already return None on failure) — replace with a comment.
- Add test_mcp_path_does_not_use_userinfo_for_opaque_token to pin that the
  userinfo fallback is management-path-only (MCP path still 401s).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 nextcloud_mcp_server/auth/unified_verifier.py | 11 ++++++++---
 tests/unit/test_unified_verifier.py           | 15 +++++++++++++++
 2 files changed, 23 insertions(+), 3 deletions(-)

diff --git a/nextcloud_mcp_server/auth/unified_verifier.py b/nextcloud_mcp_server/auth/unified_verifier.py
index 50341040..d49a7642 100644
--- a/nextcloud_mcp_server/auth/unified_verifier.py
+++ b/nextcloud_mcp_server/auth/unified_verifier.py
@@ -175,6 +175,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
@@ -420,9 +426,8 @@ class UnifiedTokenVerifier(TokenVerifier):
                     # a userinfo failure metric when userinfo was never attempted.
                     return None
 
-            # Check payload is valid
-            if not payload:
-                return None
+            # Both branches above either set a populated payload or have already
+            # returned None, so payload is guaranteed truthy here.
 
             # Skip audience validation - any valid Nextcloud token is accepted
             logger.debug(
diff --git a/tests/unit/test_unified_verifier.py b/tests/unit/test_unified_verifier.py
index 95d4dd52..a290f7b0 100644
--- a/tests/unit/test_unified_verifier.py
+++ b/tests/unit/test_unified_verifier.py
@@ -780,6 +780,21 @@ class TestUserinfoFallback:
         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."""

From a926210a510a8888d227d866312619f327ea2b9b Mon Sep 17 00:00:00 2001
From: Chris Coutinho <chris@coutinho.io>
Date: Wed, 17 Jun 2026 20:45:43 +0200
Subject: [PATCH 10/11] fix(auth): clarify empty-allowlist startup warning when
 userinfo is configured

Address claude-review round 9 on #919: an empty ALLOWED_MGMT_CLIENT is no longer
a kill switch when userinfo_uri is configured (opaque tokens validated via the
userinfo fallback bypass the allowlist). Distinguish the two cases in the
startup warning so operators aren't surprised that Astrolabe tokens are still
accepted.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 nextcloud_mcp_server/auth/unified_verifier.py | 18 ++++++++++++++----
 1 file changed, 14 insertions(+), 4 deletions(-)

diff --git a/nextcloud_mcp_server/auth/unified_verifier.py b/nextcloud_mcp_server/auth/unified_verifier.py
index d49a7642..be55f2b3 100644
--- a/nextcloud_mcp_server/auth/unified_verifier.py
+++ b/nextcloud_mcp_server/auth/unified_verifier.py
@@ -115,10 +115,20 @@ class UnifiedTokenVerifier(TokenVerifier):
             if entry.strip()
         )
         if not self._allowed_mgmt_clients:
-            logger.warning(
-                "ALLOWED_MGMT_CLIENT is unset or empty: management API will reject "
-                "all requests until configured."
-            )
+            if self.userinfo_uri:
+                # An empty allowlist is NOT a kill switch when userinfo is
+                # 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)

From 63671b439707b0366d225e8dd6df938ff3122b13 Mon Sep 17 00:00:00 2001
From: Chris Coutinho <chris@coutinho.io>
Date: Wed, 17 Jun 2026 21:36:03 +0200
Subject: [PATCH 11/11] test(auth): assert userinfo tokens have empty scopes
 (contract guard)

Address claude-review round 10 (Option B): pin the empty-scope contract for
userinfo-validated tokens in test_mgmt_opaque_userinfo_fallback_accepted_despite_allowlist,
so a future @require_scopes on a management endpoint that would silently reject
cross-client callers is caught by a test rather than only the docstring.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 tests/unit/test_unified_verifier.py | 3 +++
 1 file changed, 3 insertions(+)

diff --git a/tests/unit/test_unified_verifier.py b/tests/unit/test_unified_verifier.py
index a290f7b0..3a94c133 100644
--- a/tests/unit/test_unified_verifier.py
+++ b/tests/unit/test_unified_verifier.py
@@ -704,6 +704,9 @@ class TestUserinfoFallback:
         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