cf62e65c55
Community edition generated from private repo via sync pipeline. Includes 9 plugins (WordPress, WooCommerce, WP Advanced, Gitea, n8n, Supabase, OpenPanel, Appwrite, Directus) with ~587 tools. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
340 lines
11 KiB
Python
340 lines
11 KiB
Python
"""
|
|
OAuth 2.1 Integration Tests
|
|
|
|
Tests the complete OAuth flow end-to-end:
|
|
1. Client registration
|
|
2. Authorization request
|
|
3. Authorization code generation
|
|
4. Token exchange
|
|
5. Token refresh
|
|
6. JWT validation
|
|
"""
|
|
|
|
import os
|
|
import tempfile
|
|
|
|
import pytest
|
|
|
|
from core.oauth import (
|
|
OAuthError,
|
|
generate_code_challenge,
|
|
generate_code_verifier,
|
|
get_client_registry,
|
|
get_oauth_server,
|
|
get_storage,
|
|
get_token_manager,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def temp_storage():
|
|
"""Create temporary storage for tests"""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
# Set environment for storage
|
|
os.environ["OAUTH_STORAGE_PATH"] = tmpdir
|
|
os.environ["OAUTH_JWT_SECRET_KEY"] = "test_secret_key_for_integration_tests"
|
|
|
|
# Reset singletons by clearing them
|
|
from core.oauth import client_registry, server, storage, token_manager
|
|
|
|
client_registry._client_registry = None
|
|
token_manager._token_manager = None
|
|
storage._storage = None
|
|
server._oauth_server = None
|
|
|
|
yield tmpdir
|
|
|
|
|
|
@pytest.fixture
|
|
def oauth_components(temp_storage):
|
|
"""Get fresh OAuth components"""
|
|
return {
|
|
"server": get_oauth_server(),
|
|
"client_registry": get_client_registry(),
|
|
"token_manager": get_token_manager(),
|
|
"storage": get_storage(),
|
|
}
|
|
|
|
|
|
@pytest.fixture
|
|
def test_client(oauth_components):
|
|
"""Create a test OAuth client"""
|
|
client_registry = oauth_components["client_registry"]
|
|
|
|
client_id, client_secret = client_registry.create_client(
|
|
client_name="Test Client",
|
|
redirect_uris=["http://localhost:3000/callback"],
|
|
grant_types=["authorization_code", "refresh_token", "client_credentials"],
|
|
allowed_scopes=["read", "write", "admin"],
|
|
)
|
|
|
|
return {
|
|
"client_id": client_id,
|
|
"client_secret": client_secret,
|
|
"redirect_uri": "http://localhost:3000/callback",
|
|
}
|
|
|
|
|
|
def test_full_authorization_code_flow(oauth_components, test_client):
|
|
"""
|
|
Test complete Authorization Code flow with PKCE
|
|
|
|
Steps:
|
|
1. Generate PKCE verifier/challenge
|
|
2. Validate authorization request
|
|
3. Create authorization code
|
|
4. Exchange code for tokens
|
|
5. Validate access token
|
|
6. Refresh access token
|
|
"""
|
|
server = oauth_components["server"]
|
|
token_manager = oauth_components["token_manager"]
|
|
|
|
# Step 1: Generate PKCE
|
|
code_verifier = generate_code_verifier()
|
|
code_challenge = generate_code_challenge(code_verifier)
|
|
|
|
# Step 2: Validate authorization request
|
|
validated = server.validate_authorization_request(
|
|
client_id=test_client["client_id"],
|
|
redirect_uri=test_client["redirect_uri"],
|
|
response_type="code",
|
|
code_challenge=code_challenge,
|
|
code_challenge_method="S256",
|
|
scope="read write",
|
|
state="random_state_token",
|
|
)
|
|
|
|
assert validated["client_id"] == test_client["client_id"]
|
|
assert validated["scope"] == "read write"
|
|
assert validated["state"] == "random_state_token"
|
|
|
|
# Step 3: Create authorization code
|
|
auth_code = server.create_authorization_code(
|
|
client_id=validated["client_id"],
|
|
redirect_uri=validated["redirect_uri"],
|
|
scope=validated["scope"],
|
|
code_challenge=validated["code_challenge"],
|
|
code_challenge_method=validated["code_challenge_method"],
|
|
)
|
|
|
|
assert auth_code.startswith("auth_")
|
|
|
|
# Step 4: Exchange code for tokens
|
|
token_response = server.exchange_code_for_tokens(
|
|
client_id=test_client["client_id"],
|
|
client_secret=test_client["client_secret"],
|
|
code=auth_code,
|
|
redirect_uri=test_client["redirect_uri"],
|
|
code_verifier=code_verifier,
|
|
)
|
|
|
|
assert token_response.access_token
|
|
assert token_response.refresh_token
|
|
assert token_response.token_type == "Bearer"
|
|
assert token_response.expires_in > 0
|
|
assert token_response.scope == "read write"
|
|
|
|
# Step 5: Validate access token
|
|
payload = token_manager.validate_access_token(token_response.access_token)
|
|
|
|
assert payload["client_id"] == test_client["client_id"]
|
|
assert payload["scope"] == "read write"
|
|
assert "exp" in payload
|
|
assert "iat" in payload
|
|
assert "jti" in payload
|
|
|
|
# Step 6: Refresh access token
|
|
new_tokens = server.handle_refresh_token_grant(
|
|
client_id=test_client["client_id"],
|
|
client_secret=test_client["client_secret"],
|
|
refresh_token=token_response.refresh_token,
|
|
)
|
|
|
|
assert new_tokens.access_token != token_response.access_token
|
|
assert new_tokens.refresh_token != token_response.refresh_token
|
|
assert new_tokens.scope == "read write"
|
|
|
|
|
|
def test_client_credentials_flow(oauth_components, test_client):
|
|
"""
|
|
Test Client Credentials flow (machine-to-machine)
|
|
"""
|
|
server = oauth_components["server"]
|
|
token_manager = oauth_components["token_manager"]
|
|
|
|
# Request token with client credentials
|
|
token_response = server.handle_client_credentials_grant(
|
|
client_id=test_client["client_id"], client_secret=test_client["client_secret"], scope="read"
|
|
)
|
|
|
|
assert token_response.access_token
|
|
assert token_response.refresh_token is None # No refresh token for client credentials
|
|
assert token_response.scope == "read"
|
|
|
|
# Validate token
|
|
payload = token_manager.validate_access_token(token_response.access_token)
|
|
|
|
assert payload["client_id"] == test_client["client_id"]
|
|
assert payload["scope"] == "read"
|
|
|
|
|
|
def test_pkce_validation_failure(oauth_components, test_client):
|
|
"""Test that PKCE validation fails with wrong code_verifier"""
|
|
server = oauth_components["server"]
|
|
|
|
# Generate PKCE
|
|
code_verifier = generate_code_verifier()
|
|
code_challenge = generate_code_challenge(code_verifier)
|
|
|
|
# Create authorization code
|
|
validated = server.validate_authorization_request(
|
|
client_id=test_client["client_id"],
|
|
redirect_uri=test_client["redirect_uri"],
|
|
response_type="code",
|
|
code_challenge=code_challenge,
|
|
code_challenge_method="S256",
|
|
)
|
|
|
|
auth_code = server.create_authorization_code(
|
|
client_id=validated["client_id"],
|
|
redirect_uri=validated["redirect_uri"],
|
|
scope=validated["scope"],
|
|
code_challenge=validated["code_challenge"],
|
|
code_challenge_method=validated["code_challenge_method"],
|
|
)
|
|
|
|
# Try to exchange with WRONG code_verifier
|
|
wrong_verifier = generate_code_verifier()
|
|
|
|
with pytest.raises(OAuthError, match="PKCE validation failed"):
|
|
server.exchange_code_for_tokens(
|
|
client_id=test_client["client_id"],
|
|
client_secret=test_client["client_secret"],
|
|
code=auth_code,
|
|
redirect_uri=test_client["redirect_uri"],
|
|
code_verifier=wrong_verifier, # WRONG!
|
|
)
|
|
|
|
|
|
def test_authorization_code_reuse_detection(oauth_components, test_client):
|
|
"""Test that reusing an authorization code is detected"""
|
|
server = oauth_components["server"]
|
|
|
|
# Generate PKCE and authorization code
|
|
code_verifier = generate_code_verifier()
|
|
code_challenge = generate_code_challenge(code_verifier)
|
|
|
|
validated = server.validate_authorization_request(
|
|
client_id=test_client["client_id"],
|
|
redirect_uri=test_client["redirect_uri"],
|
|
response_type="code",
|
|
code_challenge=code_challenge,
|
|
code_challenge_method="S256",
|
|
)
|
|
|
|
auth_code = server.create_authorization_code(
|
|
client_id=validated["client_id"],
|
|
redirect_uri=validated["redirect_uri"],
|
|
scope=validated["scope"],
|
|
code_challenge=validated["code_challenge"],
|
|
code_challenge_method=validated["code_challenge_method"],
|
|
)
|
|
|
|
# First exchange - should succeed
|
|
token_response = server.exchange_code_for_tokens(
|
|
client_id=test_client["client_id"],
|
|
client_secret=test_client["client_secret"],
|
|
code=auth_code,
|
|
redirect_uri=test_client["redirect_uri"],
|
|
code_verifier=code_verifier,
|
|
)
|
|
|
|
assert token_response.access_token
|
|
|
|
# Second exchange with same code - should fail
|
|
with pytest.raises(OAuthError, match="already used"):
|
|
server.exchange_code_for_tokens(
|
|
client_id=test_client["client_id"],
|
|
client_secret=test_client["client_secret"],
|
|
code=auth_code,
|
|
redirect_uri=test_client["redirect_uri"],
|
|
code_verifier=code_verifier,
|
|
)
|
|
|
|
|
|
def test_invalid_client_credentials(oauth_components, test_client):
|
|
"""Test that invalid client credentials are rejected"""
|
|
server = oauth_components["server"]
|
|
|
|
# Try with wrong client secret
|
|
with pytest.raises(OAuthError, match="Invalid client credentials"):
|
|
server.handle_client_credentials_grant(
|
|
client_id=test_client["client_id"], client_secret="wrong_secret", scope="read"
|
|
)
|
|
|
|
|
|
def test_scope_validation(oauth_components, test_client):
|
|
"""Test that invalid scopes are rejected"""
|
|
server = oauth_components["server"]
|
|
|
|
# Try to request invalid scope
|
|
code_verifier = generate_code_verifier()
|
|
code_challenge = generate_code_challenge(code_verifier)
|
|
|
|
with pytest.raises(OAuthError, match="not allowed"):
|
|
server.validate_authorization_request(
|
|
client_id=test_client["client_id"],
|
|
redirect_uri=test_client["redirect_uri"],
|
|
response_type="code",
|
|
code_challenge=code_challenge,
|
|
code_challenge_method="S256",
|
|
scope="invalid_scope", # Not in allowed_scopes
|
|
)
|
|
|
|
|
|
def test_redirect_uri_validation(oauth_components, test_client):
|
|
"""Test that invalid redirect URIs are rejected"""
|
|
server = oauth_components["server"]
|
|
|
|
code_verifier = generate_code_verifier()
|
|
code_challenge = generate_code_challenge(code_verifier)
|
|
|
|
# Try with unregistered redirect_uri
|
|
with pytest.raises(OAuthError, match="Invalid redirect_uri"):
|
|
server.validate_authorization_request(
|
|
client_id=test_client["client_id"],
|
|
redirect_uri="http://evil.com/callback", # Not registered!
|
|
response_type="code",
|
|
code_challenge=code_challenge,
|
|
code_challenge_method="S256",
|
|
)
|
|
|
|
|
|
def test_expired_token(oauth_components, test_client):
|
|
"""Test that expired JWT tokens are rejected"""
|
|
import time
|
|
|
|
import jwt as pyjwt
|
|
|
|
token_manager = oauth_components["token_manager"]
|
|
|
|
# Set very short TTL
|
|
original_ttl = token_manager.access_token_ttl
|
|
token_manager.access_token_ttl = 1 # 1 second
|
|
|
|
# Generate token
|
|
access_token = token_manager.generate_access_token(
|
|
client_id=test_client["client_id"], scope="read"
|
|
)
|
|
|
|
# Wait for expiration
|
|
time.sleep(2)
|
|
|
|
# Try to validate expired token
|
|
with pytest.raises(pyjwt.ExpiredSignatureError):
|
|
token_manager.validate_access_token(access_token)
|
|
|
|
# Restore original TTL
|
|
token_manager.access_token_ttl = original_ttl
|