43bc58770d
Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
214 lines
6.5 KiB
Python
214 lines
6.5 KiB
Python
"""
|
|
Mock OAuth2/OIDC Server for testing authentication flows.
|
|
|
|
Uses testcontainers to spin up a mock-oauth2-server instance that provides
|
|
a complete OIDC provider with .well-known/openid-configuration, JWKS, token,
|
|
and userinfo endpoints.
|
|
|
|
This allows for realistic OAuth testing without requiring a real IdP.
|
|
"""
|
|
|
|
import logging
|
|
import time
|
|
from typing import Dict, Optional
|
|
|
|
import requests
|
|
from testcontainers.core.container import DockerContainer
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class MockOAuth2ServerContainer(DockerContainer):
|
|
"""
|
|
Testcontainer for mock-oauth2-server.
|
|
|
|
Provides a complete OIDC provider for testing OAuth2 flows.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
image: str = "ghcr.io/navikt/mock-oauth2-server:2.1.1",
|
|
port: int = 8080,
|
|
issuer_id: str = "default",
|
|
):
|
|
"""
|
|
Initialize the mock OAuth2 server container.
|
|
|
|
Args:
|
|
image: Docker image to use
|
|
port: Internal container port (default 8080)
|
|
issuer_id: Issuer identifier for the mock server
|
|
"""
|
|
super().__init__(image)
|
|
self.port = port
|
|
self.issuer_id = issuer_id
|
|
self.with_exposed_ports(port)
|
|
|
|
def get_base_url(self) -> str:
|
|
"""Get the base URL for the mock OAuth server."""
|
|
host = self.get_container_host_ip()
|
|
port = self.get_exposed_port(self.port)
|
|
return f"http://{host}:{port}"
|
|
|
|
def get_issuer_url(self) -> str:
|
|
"""Get the issuer URL for the OIDC provider."""
|
|
return f"{self.get_base_url()}/{self.issuer_id}"
|
|
|
|
def get_well_known_url(self) -> str:
|
|
"""Get the .well-known/openid-configuration URL."""
|
|
return f"{self.get_issuer_url()}/.well-known/openid-configuration"
|
|
|
|
def get_token_endpoint(self) -> str:
|
|
"""Get the token endpoint URL."""
|
|
return f"{self.get_issuer_url()}/token"
|
|
|
|
def get_authorization_endpoint(self) -> str:
|
|
"""Get the authorization endpoint URL."""
|
|
return f"{self.get_issuer_url()}/authorize"
|
|
|
|
def get_userinfo_endpoint(self) -> str:
|
|
"""Get the userinfo endpoint URL."""
|
|
return f"{self.get_issuer_url()}/userinfo"
|
|
|
|
def get_jwks_uri(self) -> str:
|
|
"""Get the JWKS URI."""
|
|
return f"{self.get_issuer_url()}/jwks"
|
|
|
|
def wait_for_ready(self, timeout: int = 30) -> None:
|
|
"""
|
|
Wait for the OAuth server to be ready by checking the well-known endpoint.
|
|
|
|
Args:
|
|
timeout: Maximum time to wait in seconds
|
|
"""
|
|
start_time = time.time()
|
|
while time.time() - start_time < timeout:
|
|
try:
|
|
response = requests.get(self.get_well_known_url(), timeout=5)
|
|
if response.status_code == 200:
|
|
logger.info(f"Mock OAuth2 server is ready at {self.get_base_url()}")
|
|
return
|
|
except requests.exceptions.RequestException:
|
|
pass
|
|
time.sleep(0.5)
|
|
|
|
raise TimeoutError(f"Mock OAuth2 server did not become ready within {timeout}s")
|
|
|
|
def get_config(self) -> Dict[str, str]:
|
|
"""
|
|
Get the OAuth configuration for the mock server.
|
|
|
|
Returns:
|
|
Dictionary with OAuth endpoints and configuration
|
|
"""
|
|
return {
|
|
"issuer": self.get_issuer_url(),
|
|
"authorization_endpoint": self.get_authorization_endpoint(),
|
|
"token_endpoint": self.get_token_endpoint(),
|
|
"userinfo_endpoint": self.get_userinfo_endpoint(),
|
|
"jwks_uri": self.get_jwks_uri(),
|
|
"well_known_url": self.get_well_known_url(),
|
|
"base_url": self.get_base_url(),
|
|
}
|
|
|
|
def create_token(
|
|
self,
|
|
subject: str = "test-user",
|
|
claims: Optional[Dict] = None,
|
|
audience: str = "test-client",
|
|
) -> str:
|
|
"""
|
|
Create a mock JWT token.
|
|
|
|
The mock-oauth2-server will generate a valid JWT that can be verified
|
|
using its JWKS endpoint.
|
|
|
|
Args:
|
|
subject: Subject (sub) claim for the token
|
|
claims: Additional claims to include in the token
|
|
audience: Audience (aud) claim
|
|
|
|
Returns:
|
|
JWT token string
|
|
"""
|
|
if claims is None:
|
|
claims = {}
|
|
|
|
# Add standard claims
|
|
token_claims = {
|
|
"sub": subject,
|
|
"aud": audience,
|
|
**claims,
|
|
}
|
|
|
|
# The debugger endpoint expects a different format
|
|
# For simpler testing, we'll use the token endpoint directly
|
|
# with a mock authorization code flow
|
|
|
|
# Note: For actual tests, we'll mock the token exchange in the tests
|
|
# This method is mainly for documentation/example purposes
|
|
logger.info(f"Creating token for subject: {subject}")
|
|
|
|
# Return a placeholder - in actual tests we'll mock the OAuth flow
|
|
return f"mock-token-{subject}"
|
|
|
|
|
|
def create_test_userinfo(
|
|
sub: str = "test-user-123",
|
|
email: str = "test@example.com",
|
|
name: str = "Test User",
|
|
preferred_username: str = "testuser",
|
|
groups: Optional[list] = None,
|
|
) -> Dict:
|
|
"""
|
|
Create a test userinfo response.
|
|
|
|
Args:
|
|
sub: Subject identifier
|
|
email: User email address
|
|
name: Full name
|
|
preferred_username: Username
|
|
groups: List of group names
|
|
|
|
Returns:
|
|
Dictionary with userinfo claims
|
|
"""
|
|
if groups is None:
|
|
groups = ["admin"]
|
|
|
|
return {
|
|
"sub": sub,
|
|
"email": email,
|
|
"email_verified": True,
|
|
"name": name,
|
|
"preferred_username": preferred_username,
|
|
"groups": groups,
|
|
"picture": f"https://www.gravatar.com/avatar/{sub}?d=identicon",
|
|
}
|
|
|
|
|
|
def configure_mock_oauth_response(
|
|
container: MockOAuth2ServerContainer,
|
|
code: str,
|
|
userinfo: Optional[Dict] = None,
|
|
access_token: Optional[str] = None,
|
|
) -> None:
|
|
"""
|
|
Configure the mock OAuth server to return specific responses for a code.
|
|
|
|
This is useful for testing the OAuth callback flow.
|
|
|
|
Args:
|
|
container: The mock OAuth server container
|
|
code: Authorization code to configure
|
|
userinfo: Userinfo response to return
|
|
access_token: Access token to return (if None, server generates one)
|
|
"""
|
|
if userinfo is None:
|
|
userinfo = create_test_userinfo()
|
|
|
|
# The mock-oauth2-server automatically handles code exchange
|
|
# and returns the configured userinfo
|
|
# This is a placeholder for any additional configuration needed
|
|
logger.info(f"Configured mock OAuth response for code: {code}")
|