Chris CoutinhoandClaude Opus 4.8
ed32519563
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>
872 lines
33 KiB
Python
872 lines
33 KiB
Python
"""
|
|
Unit tests for UnifiedTokenVerifier (ADR-005).
|
|
|
|
Tests multi-audience token validation without requiring real network calls or
|
|
IdP connections.
|
|
"""
|
|
|
|
import time
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import httpx
|
|
import jwt
|
|
import pytest
|
|
|
|
from nextcloud_mcp_server.auth.unified_verifier import UnifiedTokenVerifier
|
|
from nextcloud_mcp_server.config import Settings
|
|
|
|
pytestmark = pytest.mark.unit
|
|
|
|
|
|
@pytest.fixture
|
|
def base_settings():
|
|
"""Create base settings for testing."""
|
|
return Settings(
|
|
oidc_client_id="test-client-id",
|
|
oidc_client_secret="test-client-secret",
|
|
oidc_issuer="https://idp.example.com",
|
|
nextcloud_host="https://nextcloud.example.com",
|
|
nextcloud_mcp_server_url="http://localhost:8000",
|
|
nextcloud_resource_uri="http://localhost:8080",
|
|
jwks_uri="https://idp.example.com/jwks",
|
|
introspection_uri="https://idp.example.com/introspect",
|
|
)
|
|
|
|
|
|
class TestUnifiedTokenVerifierInit:
|
|
"""Test UnifiedTokenVerifier initialization."""
|
|
|
|
def test_init(self, base_settings):
|
|
"""Test verifier initialization (multi-audience only; no token exchange)."""
|
|
verifier = UnifiedTokenVerifier(base_settings)
|
|
assert verifier.settings == base_settings
|
|
|
|
|
|
class TestAudienceValidation:
|
|
"""Test audience validation logic."""
|
|
|
|
def test_validate_multi_audience_both_present(self, base_settings):
|
|
"""Test MCP audience validation with both audiences present."""
|
|
verifier = UnifiedTokenVerifier(base_settings)
|
|
payload = {
|
|
"aud": ["test-client-id", "http://localhost:8080"],
|
|
"sub": "testuser",
|
|
"exp": int(time.time() + 3600),
|
|
}
|
|
|
|
assert verifier._has_mcp_audience(payload) is True
|
|
|
|
def test_validate_multi_audience_server_url_and_resource(self, base_settings):
|
|
"""Test MCP audience validation with server URL instead of client ID."""
|
|
verifier = UnifiedTokenVerifier(base_settings)
|
|
payload = {
|
|
"aud": ["http://localhost:8000", "http://localhost:8080"],
|
|
"sub": "testuser",
|
|
"exp": int(time.time() + 3600),
|
|
}
|
|
|
|
assert verifier._has_mcp_audience(payload) is True
|
|
|
|
def test_validate_multi_audience_missing_mcp(self, base_settings):
|
|
"""Test MCP audience validation fails without MCP audience."""
|
|
verifier = UnifiedTokenVerifier(base_settings)
|
|
payload = {
|
|
"aud": ["http://localhost:8080"], # Only Nextcloud
|
|
"sub": "testuser",
|
|
"exp": int(time.time() + 3600),
|
|
}
|
|
|
|
assert verifier._has_mcp_audience(payload) is False
|
|
|
|
def test_validate_multi_audience_missing_nextcloud(self, base_settings):
|
|
"""Test MCP audience validation succeeds with only MCP audience (RFC 7519 compliant)."""
|
|
verifier = UnifiedTokenVerifier(base_settings)
|
|
payload = {
|
|
"aud": ["test-client-id"], # Only MCP
|
|
"sub": "testuser",
|
|
"exp": int(time.time() + 3600),
|
|
}
|
|
|
|
# Per RFC 7519, we only validate MCP audience. Nextcloud validates its own.
|
|
assert verifier._has_mcp_audience(payload) is True
|
|
|
|
def test_validate_multi_audience_string_audience(self, base_settings):
|
|
"""Test MCP audience validation with string audience works (RFC 7519 compliant)."""
|
|
verifier = UnifiedTokenVerifier(base_settings)
|
|
payload = {
|
|
"aud": "test-client-id", # Single audience as string
|
|
"sub": "testuser",
|
|
"exp": int(time.time() + 3600),
|
|
}
|
|
|
|
# Should pass - we only validate MCP audience per RFC 7519
|
|
assert verifier._has_mcp_audience(payload) is True
|
|
|
|
def test_has_mcp_audience_with_client_id(self, base_settings):
|
|
"""Test MCP audience validation with client ID."""
|
|
verifier = UnifiedTokenVerifier(base_settings)
|
|
payload = {
|
|
"aud": ["test-client-id"],
|
|
"sub": "testuser",
|
|
"exp": int(time.time() + 3600),
|
|
}
|
|
|
|
assert verifier._has_mcp_audience(payload) is True
|
|
|
|
def test_has_mcp_audience_with_server_url(self, base_settings):
|
|
"""Test MCP audience validation with server URL."""
|
|
verifier = UnifiedTokenVerifier(base_settings)
|
|
payload = {
|
|
"aud": ["http://localhost:8000"],
|
|
"sub": "testuser",
|
|
"exp": int(time.time() + 3600),
|
|
}
|
|
|
|
assert verifier._has_mcp_audience(payload) is True
|
|
|
|
def test_has_mcp_audience_missing(self, base_settings):
|
|
"""Test MCP audience validation fails without MCP audience."""
|
|
verifier = UnifiedTokenVerifier(base_settings)
|
|
payload = {
|
|
"aud": ["http://localhost:8080"], # Wrong audience
|
|
"sub": "testuser",
|
|
"exp": int(time.time() + 3600),
|
|
}
|
|
|
|
assert verifier._has_mcp_audience(payload) is False
|
|
|
|
|
|
class TestTokenFormatDetection:
|
|
"""Test JWT format detection."""
|
|
|
|
def test_is_jwt_format_valid(self, base_settings):
|
|
"""Test JWT format detection with valid JWT."""
|
|
verifier = UnifiedTokenVerifier(base_settings)
|
|
jwt_token = "eyJhbGc.eyJzdWI.signature"
|
|
assert verifier._is_jwt_format(jwt_token) is True
|
|
|
|
def test_is_jwt_format_opaque(self, base_settings):
|
|
"""Test JWT format detection with opaque token."""
|
|
verifier = UnifiedTokenVerifier(base_settings)
|
|
opaque_token = "opaque-token-12345"
|
|
assert verifier._is_jwt_format(opaque_token) is False
|
|
|
|
|
|
class TestTokenCaching:
|
|
"""Test token caching functionality."""
|
|
|
|
async def test_cache_stores_and_retrieves(self, base_settings):
|
|
"""Test token caching stores and retrieves tokens."""
|
|
verifier = UnifiedTokenVerifier(base_settings)
|
|
|
|
# Create a valid access token
|
|
payload = {
|
|
"aud": ["test-client-id", "http://localhost:8080"],
|
|
"sub": "testuser",
|
|
"scope": "openid profile",
|
|
"exp": int(time.time() + 3600),
|
|
"client_id": "test-client-id",
|
|
}
|
|
test_token = jwt.encode(payload, "secret", algorithm="HS256")
|
|
|
|
# Create AccessToken and cache it
|
|
access_token = verifier._create_access_token(test_token, payload)
|
|
assert access_token is not None
|
|
|
|
# Should retrieve from cache
|
|
cached = verifier._get_cached_token(test_token)
|
|
assert cached is not None
|
|
assert cached.resource == "testuser"
|
|
assert cached.scopes == ["openid", "profile"]
|
|
|
|
async def test_cache_respects_expiry(self, base_settings):
|
|
"""Test that expired tokens are not returned from cache."""
|
|
verifier = UnifiedTokenVerifier(base_settings)
|
|
|
|
# Create expired token payload
|
|
payload = {
|
|
"aud": ["test-client-id", "http://localhost:8080"],
|
|
"sub": "testuser",
|
|
"scope": "openid profile",
|
|
"exp": int(time.time() - 100), # Expired 100 seconds ago
|
|
"client_id": "test-client-id",
|
|
}
|
|
test_token = jwt.encode(payload, "secret", algorithm="HS256")
|
|
|
|
# Create and cache
|
|
access_token = verifier._create_access_token(test_token, payload)
|
|
assert access_token is not None
|
|
|
|
# Should not retrieve expired token
|
|
cached = verifier._get_cached_token(test_token)
|
|
assert cached is None
|
|
|
|
async def test_cache_clear(self, base_settings):
|
|
"""Test cache clearing."""
|
|
verifier = UnifiedTokenVerifier(base_settings)
|
|
|
|
# Create and cache token
|
|
payload = {
|
|
"aud": ["test-client-id", "http://localhost:8080"],
|
|
"sub": "testuser",
|
|
"exp": int(time.time() + 3600),
|
|
}
|
|
test_token = jwt.encode(payload, "secret", algorithm="HS256")
|
|
verifier._create_access_token(test_token, payload)
|
|
|
|
# Clear cache
|
|
verifier.clear_cache()
|
|
|
|
# Should not retrieve after clear
|
|
cached = verifier._get_cached_token(test_token)
|
|
assert cached is None
|
|
|
|
|
|
class TestMultiAudienceVerification:
|
|
"""Test multi-audience token verification."""
|
|
|
|
async def test_verify_multi_audience_with_introspection(self, base_settings):
|
|
"""Test multi-audience verification using introspection."""
|
|
verifier = UnifiedTokenVerifier(base_settings)
|
|
|
|
# Mock introspection response
|
|
introspection_response = {
|
|
"active": True,
|
|
"sub": "testuser",
|
|
"aud": ["test-client-id", "http://localhost:8080"],
|
|
"scope": "openid profile",
|
|
"exp": int(time.time() + 3600),
|
|
"client_id": "test-client-id",
|
|
}
|
|
|
|
with patch.object(
|
|
verifier, "_introspect_token", return_value=introspection_response
|
|
):
|
|
opaque_token = "opaque-token-12345"
|
|
result = await verifier._verify_mcp_audience(opaque_token)
|
|
|
|
assert result is not None
|
|
assert result.resource == "testuser"
|
|
assert result.scopes == ["openid", "profile"]
|
|
|
|
async def test_verify_multi_audience_fails_without_both_audiences(
|
|
self, base_settings
|
|
):
|
|
"""Test MCP audience verification succeeds with only MCP audience (RFC 7519 compliant)."""
|
|
verifier = UnifiedTokenVerifier(base_settings)
|
|
|
|
# Mock introspection response with only MCP audience
|
|
introspection_response = {
|
|
"active": True,
|
|
"sub": "testuser",
|
|
"aud": [
|
|
"test-client-id"
|
|
], # Only MCP audience (Nextcloud validates its own)
|
|
"scope": "openid profile",
|
|
"exp": int(time.time() + 3600),
|
|
}
|
|
|
|
with patch.object(
|
|
verifier, "_introspect_token", return_value=introspection_response
|
|
):
|
|
opaque_token = "opaque-token-12345"
|
|
result = await verifier._verify_mcp_audience(opaque_token)
|
|
|
|
# Should succeed with only MCP audience per RFC 7519
|
|
assert result is not None
|
|
assert result.resource == "testuser"
|
|
|
|
|
|
class TestMcpAudienceVerification:
|
|
"""Test MCP audience verification."""
|
|
|
|
async def test_verify_mcp_audience_only_success(self, base_settings):
|
|
"""Test MCP-only audience verification succeeds with MCP audience."""
|
|
verifier = UnifiedTokenVerifier(base_settings)
|
|
|
|
# Mock introspection response with MCP audience only
|
|
introspection_response = {
|
|
"active": True,
|
|
"sub": "testuser",
|
|
"aud": ["test-client-id"],
|
|
"scope": "openid profile",
|
|
"exp": int(time.time() + 3600),
|
|
"client_id": "test-client-id",
|
|
}
|
|
|
|
with patch.object(
|
|
verifier, "_introspect_token", return_value=introspection_response
|
|
):
|
|
opaque_token = "opaque-token-12345"
|
|
result = await verifier._verify_mcp_audience(opaque_token)
|
|
|
|
assert result is not None
|
|
assert result.resource == "testuser"
|
|
|
|
async def test_verify_mcp_audience_only_fails_without_mcp(self, base_settings):
|
|
"""Test MCP audience verification fails without MCP audience."""
|
|
verifier = UnifiedTokenVerifier(base_settings)
|
|
|
|
# Mock introspection response without MCP audience
|
|
introspection_response = {
|
|
"active": True,
|
|
"sub": "testuser",
|
|
"aud": ["http://localhost:8080"], # Wrong audience
|
|
"scope": "openid profile",
|
|
"exp": int(time.time() + 3600),
|
|
}
|
|
|
|
with patch.object(
|
|
verifier, "_introspect_token", return_value=introspection_response
|
|
):
|
|
opaque_token = "opaque-token-12345"
|
|
result = await verifier._verify_mcp_audience(opaque_token)
|
|
|
|
assert result is None
|
|
|
|
|
|
class TestIntrospection:
|
|
"""Test token introspection."""
|
|
|
|
async def test_introspect_active_token(self, base_settings):
|
|
"""Test introspection of active token."""
|
|
verifier = UnifiedTokenVerifier(base_settings)
|
|
|
|
# Mock HTTP response
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
mock_response.json.return_value = {
|
|
"active": True,
|
|
"sub": "testuser",
|
|
"aud": ["test-client-id", "http://localhost:8080"],
|
|
"scope": "openid profile",
|
|
"exp": int(time.time() + 3600),
|
|
"client_id": "test-client-id",
|
|
}
|
|
|
|
verifier.http_client.post = AsyncMock(return_value=mock_response)
|
|
|
|
result = await verifier._introspect_token("test-token")
|
|
assert result is not None
|
|
assert result["active"] is True
|
|
assert result["sub"] == "testuser"
|
|
|
|
async def test_introspect_inactive_token(self, base_settings):
|
|
"""Test introspection of inactive token."""
|
|
verifier = UnifiedTokenVerifier(base_settings)
|
|
|
|
# Mock HTTP response
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
mock_response.json.return_value = {"active": False}
|
|
|
|
verifier.http_client.post = AsyncMock(return_value=mock_response)
|
|
|
|
result = await verifier._introspect_token("test-token")
|
|
assert result is None
|
|
|
|
async def test_introspect_without_endpoint(self, base_settings):
|
|
"""Test introspection when endpoint not configured."""
|
|
base_settings.introspection_uri = None
|
|
verifier = UnifiedTokenVerifier(base_settings)
|
|
|
|
result = await verifier._introspect_token("test-token")
|
|
assert result is None
|
|
|
|
|
|
class TestAccessTokenCreation:
|
|
"""Test AccessToken object creation."""
|
|
|
|
def test_create_access_token_success(self, base_settings):
|
|
"""Test successful AccessToken creation."""
|
|
verifier = UnifiedTokenVerifier(base_settings)
|
|
|
|
payload = {
|
|
"sub": "testuser",
|
|
"scope": "openid profile email",
|
|
"exp": int(time.time() + 3600),
|
|
"client_id": "test-client-id",
|
|
}
|
|
token = "test-token-123"
|
|
|
|
result = verifier._create_access_token(token, payload)
|
|
assert result is not None
|
|
assert result.token == token
|
|
assert result.resource == "testuser"
|
|
assert result.scopes == ["openid", "profile", "email"]
|
|
assert result.client_id == "test-client-id"
|
|
|
|
def test_create_access_token_with_preferred_username(self, base_settings):
|
|
"""Test AccessToken creation with preferred_username fallback."""
|
|
verifier = UnifiedTokenVerifier(base_settings)
|
|
|
|
payload = {
|
|
"preferred_username": "testuser", # No 'sub' claim
|
|
"scope": "openid profile",
|
|
"exp": int(time.time() + 3600),
|
|
}
|
|
token = "test-token-123"
|
|
|
|
result = verifier._create_access_token(token, payload)
|
|
assert result is not None
|
|
assert result.resource == "testuser"
|
|
|
|
def test_create_access_token_no_username(self, base_settings):
|
|
"""Test AccessToken creation fails without username."""
|
|
verifier = UnifiedTokenVerifier(base_settings)
|
|
|
|
payload = {
|
|
# No sub or preferred_username
|
|
"scope": "openid profile",
|
|
"exp": int(time.time() + 3600),
|
|
}
|
|
token = "test-token-123"
|
|
|
|
result = verifier._create_access_token(token, payload)
|
|
assert result is None
|
|
|
|
def test_create_access_token_no_expiry(self, base_settings):
|
|
"""Test AccessToken creation uses default TTL without expiry."""
|
|
verifier = UnifiedTokenVerifier(base_settings)
|
|
|
|
payload = {
|
|
"sub": "testuser",
|
|
"scope": "openid profile",
|
|
# No exp claim
|
|
}
|
|
token = "test-token-123"
|
|
|
|
result = verifier._create_access_token(token, payload)
|
|
assert result is not None
|
|
# Should have set a default expiry
|
|
assert result.expires_at > int(time.time())
|
|
|
|
|
|
class TestVerifyTokenFlow:
|
|
"""Test complete verify_token flow."""
|
|
|
|
async def test_verify_token_from_cache(self, base_settings):
|
|
"""Test verify_token returns cached token."""
|
|
verifier = UnifiedTokenVerifier(base_settings)
|
|
|
|
payload = {
|
|
"aud": ["test-client-id", "http://localhost:8080"],
|
|
"sub": "testuser",
|
|
"scope": "openid profile",
|
|
"exp": int(time.time() + 3600),
|
|
}
|
|
token = jwt.encode(payload, "secret", algorithm="HS256")
|
|
|
|
# First call - should cache
|
|
result1 = verifier._create_access_token(token, payload)
|
|
assert result1 is not None
|
|
|
|
# Mock _verify_mcp_audience to ensure it's not called
|
|
with patch.object(verifier, "_verify_mcp_audience") as mock_verify:
|
|
result2 = await verifier.verify_token(token)
|
|
assert result2 is not None
|
|
assert result2.resource == "testuser"
|
|
# Should not call verification since it's cached
|
|
mock_verify.assert_not_called()
|
|
|
|
async def test_verify_token_multi_audience_mode(self, base_settings):
|
|
"""Test verify_token in multi-audience mode."""
|
|
verifier = UnifiedTokenVerifier(base_settings)
|
|
|
|
introspection_response = {
|
|
"active": True,
|
|
"sub": "testuser",
|
|
"aud": ["test-client-id", "http://localhost:8080"],
|
|
"scope": "openid profile",
|
|
"exp": int(time.time() + 3600),
|
|
}
|
|
|
|
with patch.object(
|
|
verifier, "_introspect_token", return_value=introspection_response
|
|
):
|
|
result = await verifier.verify_token("opaque-token")
|
|
assert result is not None
|
|
assert result.resource == "testuser"
|
|
|
|
async def test_verify_token_mcp_audience_only(self, base_settings):
|
|
"""Test verify_token with MCP audience only."""
|
|
verifier = UnifiedTokenVerifier(base_settings)
|
|
|
|
introspection_response = {
|
|
"active": True,
|
|
"sub": "testuser",
|
|
"aud": ["test-client-id"], # MCP audience only
|
|
"scope": "openid profile",
|
|
"exp": int(time.time() + 3600),
|
|
}
|
|
|
|
with patch.object(
|
|
verifier, "_introspect_token", return_value=introspection_response
|
|
):
|
|
result = await verifier.verify_token("opaque-token")
|
|
assert result is not None
|
|
assert result.resource == "testuser"
|
|
|
|
|
|
class TestManagementApiAllowlist:
|
|
"""Test ALLOWED_MGMT_CLIENT enforcement in verify_token_for_management_api."""
|
|
|
|
@staticmethod
|
|
def _underlying_token(client_id: str = "astrolabe"):
|
|
from mcp.server.auth.provider import AccessToken
|
|
|
|
return AccessToken(
|
|
token="t",
|
|
client_id=client_id,
|
|
scopes=["openid"],
|
|
expires_at=int(time.time() + 3600),
|
|
resource="testuser",
|
|
)
|
|
|
|
async def test_unset_allowlist_rejects_all(self, monkeypatch, base_settings):
|
|
monkeypatch.delenv("ALLOWED_MGMT_CLIENT", raising=False)
|
|
verifier = UnifiedTokenVerifier(base_settings)
|
|
assert verifier._allowed_mgmt_clients == frozenset()
|
|
|
|
with patch.object(
|
|
verifier,
|
|
"_verify_without_audience_check",
|
|
return_value=self._underlying_token("astrolabe"),
|
|
):
|
|
result = await verifier.verify_token_for_management_api("any-token")
|
|
assert result is None
|
|
|
|
async def test_empty_allowlist_rejects_all(self, monkeypatch, base_settings):
|
|
monkeypatch.setenv("ALLOWED_MGMT_CLIENT", " , ,")
|
|
verifier = UnifiedTokenVerifier(base_settings)
|
|
assert verifier._allowed_mgmt_clients == frozenset()
|
|
|
|
with patch.object(
|
|
verifier,
|
|
"_verify_without_audience_check",
|
|
return_value=self._underlying_token("astrolabe"),
|
|
):
|
|
result = await verifier.verify_token_for_management_api("any-token")
|
|
assert result is None
|
|
|
|
async def test_allowlisted_client_accepted(self, monkeypatch, base_settings):
|
|
monkeypatch.setenv("ALLOWED_MGMT_CLIENT", "astrolabe, admin-tool")
|
|
verifier = UnifiedTokenVerifier(base_settings)
|
|
assert verifier._allowed_mgmt_clients == {"astrolabe", "admin-tool"}
|
|
|
|
underlying = self._underlying_token("astrolabe")
|
|
with patch.object(
|
|
verifier, "_verify_without_audience_check", return_value=underlying
|
|
):
|
|
result = await verifier.verify_token_for_management_api("any-token")
|
|
assert result is underlying
|
|
|
|
async def test_non_allowlisted_client_rejected(self, monkeypatch, base_settings):
|
|
monkeypatch.setenv("ALLOWED_MGMT_CLIENT", "astrolabe")
|
|
verifier = UnifiedTokenVerifier(base_settings)
|
|
|
|
with patch.object(
|
|
verifier,
|
|
"_verify_without_audience_check",
|
|
return_value=self._underlying_token("some-other-client"),
|
|
):
|
|
result = await verifier.verify_token_for_management_api("any-token")
|
|
assert result is None
|
|
|
|
async def test_token_missing_client_id_rejected(self, monkeypatch, base_settings):
|
|
monkeypatch.setenv("ALLOWED_MGMT_CLIENT", "astrolabe")
|
|
verifier = UnifiedTokenVerifier(base_settings)
|
|
|
|
with patch.object(
|
|
verifier,
|
|
"_verify_without_audience_check",
|
|
return_value=self._underlying_token(""),
|
|
):
|
|
result = await verifier.verify_token_for_management_api("any-token")
|
|
assert result is None
|
|
|
|
async def test_underlying_verification_failure_propagates(
|
|
self, monkeypatch, base_settings
|
|
):
|
|
monkeypatch.setenv("ALLOWED_MGMT_CLIENT", "astrolabe")
|
|
verifier = UnifiedTokenVerifier(base_settings)
|
|
|
|
with patch.object(
|
|
verifier, "_verify_without_audience_check", return_value=None
|
|
):
|
|
result = await verifier.verify_token_for_management_api("any-token")
|
|
assert result is None
|
|
|
|
async def test_cache_hit_also_enforces_allowlist(self, monkeypatch, base_settings):
|
|
"""A previously-cached token must still be re-checked against the allowlist."""
|
|
import hashlib
|
|
|
|
monkeypatch.setenv("ALLOWED_MGMT_CLIENT", "astrolabe")
|
|
verifier = UnifiedTokenVerifier(base_settings)
|
|
|
|
token = "cached-token"
|
|
cache_key = f"mgmt:{hashlib.sha256(token.encode()).hexdigest()}"
|
|
verifier._token_cache[cache_key] = (
|
|
{
|
|
"sub": "testuser",
|
|
"scope": "openid",
|
|
"client_id": "not-allowlisted",
|
|
},
|
|
time.time() + 3600,
|
|
)
|
|
|
|
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_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
|
|
|
|
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_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()
|
|
|
|
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.
|
|
|
|
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)
|
|
|
|
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(
|
|
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
|