diff --git a/nextcloud_mcp_server/auth/unified_verifier.py b/nextcloud_mcp_server/auth/unified_verifier.py index 8dc83b2a..be55f2b3 100644 --- a/nextcloud_mcp_server/auth/unified_verifier.py +++ b/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( - "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) @@ -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 - validation_method = "introspect" - payload = await self._introspect_token(token) - if payload: - record_oauth_token_validation("introspect", "valid") - else: - record_oauth_token_validation("introspect", "invalid") + # 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") + + # 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 - 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( @@ -358,8 +445,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) @@ -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") - 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 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( diff --git a/tests/unit/test_unified_verifier.py b/tests/unit/test_unified_verifier.py index 70ca88c5..3a94c133 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 @@ -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 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