feat(test): add Phase 5b integration and E2E tests
Add comprehensive integration and end-to-end test suites: - Integration tests for API flows (authorization, token, verification) - Integration tests for middleware chain and security headers - Integration tests for domain verification services - E2E tests for complete authentication flows - E2E tests for error scenarios and edge cases - Shared test fixtures and utilities in conftest.py - Rename Dockerfile to Containerfile for Podman compatibility 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
@@ -1,10 +1,23 @@
|
||||
"""
|
||||
Pytest configuration and shared fixtures.
|
||||
|
||||
This module provides comprehensive test fixtures for Phase 5b integration
|
||||
and E2E testing. Fixtures are organized by category for maintainability.
|
||||
"""
|
||||
|
||||
import os
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
from typing import Any, Generator
|
||||
from unittest.mock import MagicMock, Mock, patch
|
||||
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# ENVIRONMENT SETUP FIXTURES
|
||||
# =============================================================================
|
||||
|
||||
|
||||
@pytest.fixture(scope="session", autouse=True)
|
||||
@@ -38,3 +51,675 @@ def reset_config_before_test(monkeypatch):
|
||||
monkeypatch.setenv("GONDULF_BASE_URL", "http://localhost:8000")
|
||||
monkeypatch.setenv("GONDULF_DEBUG", "true")
|
||||
monkeypatch.setenv("GONDULF_DATABASE_URL", "sqlite:///:memory:")
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# DATABASE FIXTURES
|
||||
# =============================================================================
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def test_db_path(tmp_path) -> Path:
|
||||
"""Create a temporary database path."""
|
||||
return tmp_path / "test.db"
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def test_database(test_db_path):
|
||||
"""
|
||||
Create and initialize a test database.
|
||||
|
||||
Yields:
|
||||
Database: Initialized database instance with tables created
|
||||
"""
|
||||
from gondulf.database.connection import Database
|
||||
|
||||
db = Database(f"sqlite:///{test_db_path}")
|
||||
db.ensure_database_directory()
|
||||
db.run_migrations()
|
||||
yield db
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def configured_test_app(monkeypatch, test_db_path):
|
||||
"""
|
||||
Create a fully configured FastAPI test app with temporary database.
|
||||
|
||||
This fixture handles all environment configuration and creates
|
||||
a fresh app instance for each test.
|
||||
"""
|
||||
# Set required environment variables
|
||||
monkeypatch.setenv("GONDULF_SECRET_KEY", "a" * 32)
|
||||
monkeypatch.setenv("GONDULF_BASE_URL", "https://auth.example.com")
|
||||
monkeypatch.setenv("GONDULF_DATABASE_URL", f"sqlite:///{test_db_path}")
|
||||
monkeypatch.setenv("GONDULF_DEBUG", "true")
|
||||
|
||||
# Import after environment is configured
|
||||
from gondulf.main import app
|
||||
yield app
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def test_client(configured_test_app) -> Generator[TestClient, None, None]:
|
||||
"""
|
||||
Create a TestClient with properly configured app.
|
||||
|
||||
Yields:
|
||||
TestClient: FastAPI test client with startup events run
|
||||
"""
|
||||
with TestClient(configured_test_app) as client:
|
||||
yield client
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# CODE STORAGE FIXTURES
|
||||
# =============================================================================
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def test_code_storage():
|
||||
"""
|
||||
Create a test code storage instance.
|
||||
|
||||
Returns:
|
||||
CodeStore: Fresh code storage for testing
|
||||
"""
|
||||
from gondulf.storage import CodeStore
|
||||
return CodeStore(ttl_seconds=600)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def valid_auth_code(test_code_storage) -> tuple[str, dict]:
|
||||
"""
|
||||
Create a valid authorization code with metadata.
|
||||
|
||||
Args:
|
||||
test_code_storage: Code storage fixture
|
||||
|
||||
Returns:
|
||||
Tuple of (code, metadata)
|
||||
"""
|
||||
code = "test_auth_code_12345"
|
||||
metadata = {
|
||||
"client_id": "https://client.example.com",
|
||||
"redirect_uri": "https://client.example.com/callback",
|
||||
"state": "xyz123",
|
||||
"me": "https://user.example.com",
|
||||
"scope": "",
|
||||
"code_challenge": "abc123def456",
|
||||
"code_challenge_method": "S256",
|
||||
"created_at": 1234567890,
|
||||
"expires_at": 1234568490,
|
||||
"used": False
|
||||
}
|
||||
test_code_storage.store(f"authz:{code}", metadata)
|
||||
return code, metadata
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def expired_auth_code(test_code_storage) -> tuple[str, dict]:
|
||||
"""
|
||||
Create an expired authorization code.
|
||||
|
||||
Returns:
|
||||
Tuple of (code, metadata) where the code is expired
|
||||
"""
|
||||
import time
|
||||
code = "expired_auth_code_12345"
|
||||
metadata = {
|
||||
"client_id": "https://client.example.com",
|
||||
"redirect_uri": "https://client.example.com/callback",
|
||||
"state": "xyz123",
|
||||
"me": "https://user.example.com",
|
||||
"scope": "",
|
||||
"code_challenge": "abc123def456",
|
||||
"code_challenge_method": "S256",
|
||||
"created_at": 1000000000,
|
||||
"expires_at": 1000000001, # Expired long ago
|
||||
"used": False
|
||||
}
|
||||
# Store with 0 TTL to make it immediately expired
|
||||
test_code_storage.store(f"authz:{code}", metadata, ttl=0)
|
||||
return code, metadata
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def used_auth_code(test_code_storage) -> tuple[str, dict]:
|
||||
"""
|
||||
Create an already-used authorization code.
|
||||
|
||||
Returns:
|
||||
Tuple of (code, metadata) where the code is marked as used
|
||||
"""
|
||||
code = "used_auth_code_12345"
|
||||
metadata = {
|
||||
"client_id": "https://client.example.com",
|
||||
"redirect_uri": "https://client.example.com/callback",
|
||||
"state": "xyz123",
|
||||
"me": "https://user.example.com",
|
||||
"scope": "",
|
||||
"code_challenge": "abc123def456",
|
||||
"code_challenge_method": "S256",
|
||||
"created_at": 1234567890,
|
||||
"expires_at": 1234568490,
|
||||
"used": True # Already used
|
||||
}
|
||||
test_code_storage.store(f"authz:{code}", metadata)
|
||||
return code, metadata
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# SERVICE FIXTURES
|
||||
# =============================================================================
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def test_token_service(test_database):
|
||||
"""
|
||||
Create a test token service with database.
|
||||
|
||||
Args:
|
||||
test_database: Database fixture
|
||||
|
||||
Returns:
|
||||
TokenService: Token service configured for testing
|
||||
"""
|
||||
from gondulf.services.token_service import TokenService
|
||||
return TokenService(
|
||||
database=test_database,
|
||||
token_length=32,
|
||||
token_ttl=3600
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_dns_service():
|
||||
"""
|
||||
Create a mock DNS service.
|
||||
|
||||
Returns:
|
||||
Mock: Mocked DNSService for testing
|
||||
"""
|
||||
mock = Mock()
|
||||
mock.verify_txt_record = Mock(return_value=True)
|
||||
mock.resolve_txt = Mock(return_value=["gondulf-verify-domain"])
|
||||
return mock
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_dns_service_failure():
|
||||
"""
|
||||
Create a mock DNS service that returns failures.
|
||||
|
||||
Returns:
|
||||
Mock: Mocked DNSService that simulates DNS failures
|
||||
"""
|
||||
mock = Mock()
|
||||
mock.verify_txt_record = Mock(return_value=False)
|
||||
mock.resolve_txt = Mock(return_value=[])
|
||||
return mock
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_email_service():
|
||||
"""
|
||||
Create a mock email service.
|
||||
|
||||
Returns:
|
||||
Mock: Mocked EmailService for testing
|
||||
"""
|
||||
mock = Mock()
|
||||
mock.send_verification_code = Mock(return_value=None)
|
||||
mock.messages_sent = []
|
||||
|
||||
def track_send(email, code, domain):
|
||||
mock.messages_sent.append({
|
||||
"email": email,
|
||||
"code": code,
|
||||
"domain": domain
|
||||
})
|
||||
|
||||
mock.send_verification_code.side_effect = track_send
|
||||
return mock
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_html_fetcher():
|
||||
"""
|
||||
Create a mock HTML fetcher service.
|
||||
|
||||
Returns:
|
||||
Mock: Mocked HTMLFetcherService
|
||||
"""
|
||||
mock = Mock()
|
||||
mock.fetch = Mock(return_value="<html><body></body></html>")
|
||||
return mock
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_html_fetcher_with_email():
|
||||
"""
|
||||
Create a mock HTML fetcher that returns a page with rel=me email.
|
||||
|
||||
Returns:
|
||||
Mock: Mocked HTMLFetcherService with email in page
|
||||
"""
|
||||
mock = Mock()
|
||||
html = '''
|
||||
<html>
|
||||
<body>
|
||||
<a href="mailto:test@example.com" rel="me">Email</a>
|
||||
</body>
|
||||
</html>
|
||||
'''
|
||||
mock.fetch = Mock(return_value=html)
|
||||
return mock
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_happ_parser():
|
||||
"""
|
||||
Create a mock h-app parser.
|
||||
|
||||
Returns:
|
||||
Mock: Mocked HAppParser
|
||||
"""
|
||||
from gondulf.services.happ_parser import ClientMetadata
|
||||
|
||||
mock = Mock()
|
||||
mock.fetch_and_parse = Mock(return_value=ClientMetadata(
|
||||
name="Test Application",
|
||||
url="https://app.example.com",
|
||||
logo="https://app.example.com/logo.png"
|
||||
))
|
||||
return mock
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_rate_limiter():
|
||||
"""
|
||||
Create a mock rate limiter that always allows requests.
|
||||
|
||||
Returns:
|
||||
Mock: Mocked RateLimiter
|
||||
"""
|
||||
mock = Mock()
|
||||
mock.check_rate_limit = Mock(return_value=True)
|
||||
mock.record_attempt = Mock()
|
||||
mock.reset = Mock()
|
||||
return mock
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_rate_limiter_exceeded():
|
||||
"""
|
||||
Create a mock rate limiter that blocks all requests.
|
||||
|
||||
Returns:
|
||||
Mock: Mocked RateLimiter that simulates rate limit exceeded
|
||||
"""
|
||||
mock = Mock()
|
||||
mock.check_rate_limit = Mock(return_value=False)
|
||||
mock.record_attempt = Mock()
|
||||
return mock
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# DOMAIN VERIFICATION FIXTURES
|
||||
# =============================================================================
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def verification_service(mock_dns_service, mock_email_service, mock_html_fetcher_with_email, test_code_storage):
|
||||
"""
|
||||
Create a domain verification service with all mocked dependencies.
|
||||
|
||||
Args:
|
||||
mock_dns_service: Mock DNS service
|
||||
mock_email_service: Mock email service
|
||||
mock_html_fetcher_with_email: Mock HTML fetcher with email
|
||||
test_code_storage: Code storage fixture
|
||||
|
||||
Returns:
|
||||
DomainVerificationService: Service configured with mocks
|
||||
"""
|
||||
from gondulf.services.domain_verification import DomainVerificationService
|
||||
from gondulf.services.relme_parser import RelMeParser
|
||||
|
||||
return DomainVerificationService(
|
||||
dns_service=mock_dns_service,
|
||||
email_service=mock_email_service,
|
||||
code_storage=test_code_storage,
|
||||
html_fetcher=mock_html_fetcher_with_email,
|
||||
relme_parser=RelMeParser()
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def verification_service_dns_failure(mock_dns_service_failure, mock_email_service, mock_html_fetcher_with_email, test_code_storage):
|
||||
"""
|
||||
Create a verification service where DNS verification fails.
|
||||
|
||||
Returns:
|
||||
DomainVerificationService: Service with failing DNS
|
||||
"""
|
||||
from gondulf.services.domain_verification import DomainVerificationService
|
||||
from gondulf.services.relme_parser import RelMeParser
|
||||
|
||||
return DomainVerificationService(
|
||||
dns_service=mock_dns_service_failure,
|
||||
email_service=mock_email_service,
|
||||
code_storage=test_code_storage,
|
||||
html_fetcher=mock_html_fetcher_with_email,
|
||||
relme_parser=RelMeParser()
|
||||
)
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# CLIENT CONFIGURATION FIXTURES
|
||||
# =============================================================================
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def simple_client() -> dict[str, str]:
|
||||
"""
|
||||
Basic IndieAuth client configuration.
|
||||
|
||||
Returns:
|
||||
Dict with client_id and redirect_uri
|
||||
"""
|
||||
return {
|
||||
"client_id": "https://app.example.com",
|
||||
"redirect_uri": "https://app.example.com/callback",
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client_with_metadata() -> dict[str, str]:
|
||||
"""
|
||||
Client configuration that would have h-app metadata.
|
||||
|
||||
Returns:
|
||||
Dict with client configuration
|
||||
"""
|
||||
return {
|
||||
"client_id": "https://rich-app.example.com",
|
||||
"redirect_uri": "https://rich-app.example.com/auth/callback",
|
||||
"expected_name": "Rich Application",
|
||||
"expected_logo": "https://rich-app.example.com/logo.png"
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def malicious_client() -> dict[str, Any]:
|
||||
"""
|
||||
Client with potentially malicious configuration for security testing.
|
||||
|
||||
Returns:
|
||||
Dict with malicious inputs
|
||||
"""
|
||||
return {
|
||||
"client_id": "https://evil.example.com",
|
||||
"redirect_uri": "https://evil.example.com/steal",
|
||||
"state": "<script>alert('xss')</script>",
|
||||
"me": "javascript:alert('xss')"
|
||||
}
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# AUTHORIZATION REQUEST FIXTURES
|
||||
# =============================================================================
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def valid_auth_request() -> dict[str, str]:
|
||||
"""
|
||||
Complete valid authorization request parameters.
|
||||
|
||||
Returns:
|
||||
Dict with all required authorization parameters
|
||||
"""
|
||||
return {
|
||||
"response_type": "code",
|
||||
"client_id": "https://app.example.com",
|
||||
"redirect_uri": "https://app.example.com/callback",
|
||||
"state": "random_state_12345",
|
||||
"code_challenge": "E9Melhoa2OwvFrEMTJguCHaoeK1t8URWbuGJSstw-cM",
|
||||
"code_challenge_method": "S256",
|
||||
"me": "https://user.example.com",
|
||||
"scope": ""
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def auth_request_missing_client_id(valid_auth_request) -> dict[str, str]:
|
||||
"""Authorization request missing client_id."""
|
||||
request = valid_auth_request.copy()
|
||||
del request["client_id"]
|
||||
return request
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def auth_request_missing_redirect_uri(valid_auth_request) -> dict[str, str]:
|
||||
"""Authorization request missing redirect_uri."""
|
||||
request = valid_auth_request.copy()
|
||||
del request["redirect_uri"]
|
||||
return request
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def auth_request_invalid_response_type(valid_auth_request) -> dict[str, str]:
|
||||
"""Authorization request with invalid response_type."""
|
||||
request = valid_auth_request.copy()
|
||||
request["response_type"] = "token" # Invalid - we only support "code"
|
||||
return request
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def auth_request_missing_pkce(valid_auth_request) -> dict[str, str]:
|
||||
"""Authorization request missing PKCE code_challenge."""
|
||||
request = valid_auth_request.copy()
|
||||
del request["code_challenge"]
|
||||
return request
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# TOKEN FIXTURES
|
||||
# =============================================================================
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def valid_token(test_token_service) -> tuple[str, dict]:
|
||||
"""
|
||||
Generate a valid access token.
|
||||
|
||||
Args:
|
||||
test_token_service: Token service fixture
|
||||
|
||||
Returns:
|
||||
Tuple of (token, metadata)
|
||||
"""
|
||||
token = test_token_service.generate_token(
|
||||
me="https://user.example.com",
|
||||
client_id="https://app.example.com",
|
||||
scope=""
|
||||
)
|
||||
metadata = test_token_service.validate_token(token)
|
||||
return token, metadata
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def expired_token_metadata() -> dict[str, Any]:
|
||||
"""
|
||||
Metadata representing an expired token (for manual database insertion).
|
||||
|
||||
Returns:
|
||||
Dict with expired token metadata
|
||||
"""
|
||||
from datetime import datetime, timedelta
|
||||
import hashlib
|
||||
|
||||
token = "expired_test_token_12345"
|
||||
return {
|
||||
"token": token,
|
||||
"token_hash": hashlib.sha256(token.encode()).hexdigest(),
|
||||
"me": "https://user.example.com",
|
||||
"client_id": "https://app.example.com",
|
||||
"scope": "",
|
||||
"issued_at": datetime.utcnow() - timedelta(hours=2),
|
||||
"expires_at": datetime.utcnow() - timedelta(hours=1), # Already expired
|
||||
"revoked": False
|
||||
}
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# HTTP MOCKING FIXTURES (for urllib)
|
||||
# =============================================================================
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_urlopen():
|
||||
"""
|
||||
Mock urllib.request.urlopen for HTTP request testing.
|
||||
|
||||
Yields:
|
||||
MagicMock: Mock that can be configured per test
|
||||
"""
|
||||
with patch('gondulf.services.html_fetcher.urllib.request.urlopen') as mock:
|
||||
yield mock
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_urlopen_success(mock_urlopen):
|
||||
"""
|
||||
Configure mock_urlopen to return a successful response.
|
||||
|
||||
Args:
|
||||
mock_urlopen: Base mock fixture
|
||||
|
||||
Returns:
|
||||
MagicMock: Configured mock
|
||||
"""
|
||||
mock_response = MagicMock()
|
||||
mock_response.read.return_value = b"<html><body>Test</body></html>"
|
||||
mock_response.status = 200
|
||||
mock_response.__enter__ = Mock(return_value=mock_response)
|
||||
mock_response.__exit__ = Mock(return_value=False)
|
||||
mock_urlopen.return_value = mock_response
|
||||
return mock_urlopen
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_urlopen_with_happ(mock_urlopen):
|
||||
"""
|
||||
Configure mock_urlopen to return a page with h-app metadata.
|
||||
|
||||
Args:
|
||||
mock_urlopen: Base mock fixture
|
||||
|
||||
Returns:
|
||||
MagicMock: Configured mock
|
||||
"""
|
||||
html = b'''
|
||||
<!DOCTYPE html>
|
||||
<html>
|
||||
<head><title>Test App</title></head>
|
||||
<body>
|
||||
<div class="h-app">
|
||||
<h1 class="p-name">Example Application</h1>
|
||||
<img class="u-logo" src="https://app.example.com/logo.png" alt="Logo">
|
||||
<a class="u-url" href="https://app.example.com">Home</a>
|
||||
</div>
|
||||
</body>
|
||||
</html>
|
||||
'''
|
||||
mock_response = MagicMock()
|
||||
mock_response.read.return_value = html
|
||||
mock_response.status = 200
|
||||
mock_response.__enter__ = Mock(return_value=mock_response)
|
||||
mock_response.__exit__ = Mock(return_value=False)
|
||||
mock_urlopen.return_value = mock_response
|
||||
return mock_urlopen
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_urlopen_timeout(mock_urlopen):
|
||||
"""
|
||||
Configure mock_urlopen to simulate a timeout.
|
||||
|
||||
Args:
|
||||
mock_urlopen: Base mock fixture
|
||||
|
||||
Returns:
|
||||
MagicMock: Configured mock that raises timeout
|
||||
"""
|
||||
import urllib.error
|
||||
mock_urlopen.side_effect = urllib.error.URLError("Connection timed out")
|
||||
return mock_urlopen
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# HELPER FUNCTIONS
|
||||
# =============================================================================
|
||||
|
||||
|
||||
def create_app_with_overrides(monkeypatch, tmp_path, **overrides):
|
||||
"""
|
||||
Helper to create a test app with custom dependency overrides.
|
||||
|
||||
Args:
|
||||
monkeypatch: pytest monkeypatch fixture
|
||||
tmp_path: temporary path for database
|
||||
**overrides: Dependency override functions
|
||||
|
||||
Returns:
|
||||
tuple: (app, client, overrides_applied)
|
||||
"""
|
||||
db_path = tmp_path / "test.db"
|
||||
|
||||
monkeypatch.setenv("GONDULF_SECRET_KEY", "a" * 32)
|
||||
monkeypatch.setenv("GONDULF_BASE_URL", "https://auth.example.com")
|
||||
monkeypatch.setenv("GONDULF_DATABASE_URL", f"sqlite:///{db_path}")
|
||||
monkeypatch.setenv("GONDULF_DEBUG", "true")
|
||||
|
||||
from gondulf.main import app
|
||||
|
||||
for dependency, override in overrides.items():
|
||||
app.dependency_overrides[dependency] = override
|
||||
|
||||
return app
|
||||
|
||||
|
||||
def extract_code_from_redirect(location: str) -> str:
|
||||
"""
|
||||
Extract authorization code from redirect URL.
|
||||
|
||||
Args:
|
||||
location: Redirect URL with code parameter
|
||||
|
||||
Returns:
|
||||
str: Authorization code
|
||||
"""
|
||||
from urllib.parse import parse_qs, urlparse
|
||||
parsed = urlparse(location)
|
||||
params = parse_qs(parsed.query)
|
||||
return params.get("code", [None])[0]
|
||||
|
||||
|
||||
def extract_error_from_redirect(location: str) -> dict[str, str]:
|
||||
"""
|
||||
Extract error parameters from redirect URL.
|
||||
|
||||
Args:
|
||||
location: Redirect URL with error parameters
|
||||
|
||||
Returns:
|
||||
Dict with error and error_description
|
||||
"""
|
||||
from urllib.parse import parse_qs, urlparse
|
||||
parsed = urlparse(location)
|
||||
params = parse_qs(parsed.query)
|
||||
return {
|
||||
"error": params.get("error", [None])[0],
|
||||
"error_description": params.get("error_description", [None])[0]
|
||||
}
|
||||
|
||||
1
tests/e2e/__init__.py
Normal file
1
tests/e2e/__init__.py
Normal file
@@ -0,0 +1 @@
|
||||
"""End-to-end tests for Gondulf IndieAuth server."""
|
||||
390
tests/e2e/test_complete_auth_flow.py
Normal file
390
tests/e2e/test_complete_auth_flow.py
Normal file
@@ -0,0 +1,390 @@
|
||||
"""
|
||||
End-to-end tests for complete IndieAuth authentication flow.
|
||||
|
||||
Tests the full authorization code flow from initial request through token exchange.
|
||||
Uses TestClient-based flow simulation per Phase 5b clarifications.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
from unittest.mock import AsyncMock, Mock, patch
|
||||
|
||||
from tests.conftest import extract_code_from_redirect
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def e2e_app(monkeypatch, tmp_path):
|
||||
"""Create app for E2E testing."""
|
||||
db_path = tmp_path / "test.db"
|
||||
|
||||
monkeypatch.setenv("GONDULF_SECRET_KEY", "a" * 32)
|
||||
monkeypatch.setenv("GONDULF_BASE_URL", "https://auth.example.com")
|
||||
monkeypatch.setenv("GONDULF_DATABASE_URL", f"sqlite:///{db_path}")
|
||||
monkeypatch.setenv("GONDULF_DEBUG", "true")
|
||||
|
||||
from gondulf.main import app
|
||||
return app
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def e2e_client(e2e_app):
|
||||
"""Create test client for E2E tests."""
|
||||
with TestClient(e2e_app) as client:
|
||||
yield client
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_happ_for_e2e():
|
||||
"""Mock h-app parser for E2E tests."""
|
||||
from gondulf.services.happ_parser import ClientMetadata
|
||||
|
||||
metadata = ClientMetadata(
|
||||
name="E2E Test App",
|
||||
url="https://app.example.com",
|
||||
logo="https://app.example.com/logo.png"
|
||||
)
|
||||
|
||||
with patch('gondulf.services.happ_parser.HAppParser.fetch_and_parse', new_callable=AsyncMock) as mock:
|
||||
mock.return_value = metadata
|
||||
yield mock
|
||||
|
||||
|
||||
@pytest.mark.e2e
|
||||
class TestCompleteAuthorizationFlow:
|
||||
"""E2E tests for complete authorization code flow."""
|
||||
|
||||
def test_full_authorization_to_token_flow(self, e2e_client, mock_happ_for_e2e):
|
||||
"""Test complete flow: authorization request -> consent -> token exchange."""
|
||||
# Step 1: Authorization request
|
||||
auth_params = {
|
||||
"response_type": "code",
|
||||
"client_id": "https://app.example.com",
|
||||
"redirect_uri": "https://app.example.com/callback",
|
||||
"state": "e2e_test_state_12345",
|
||||
"code_challenge": "E9Melhoa2OwvFrEMTJguCHaoeK1t8URWbuGJSstw-cM",
|
||||
"code_challenge_method": "S256",
|
||||
"me": "https://user.example.com",
|
||||
}
|
||||
|
||||
auth_response = e2e_client.get("/authorize", params=auth_params)
|
||||
|
||||
# Should show consent page
|
||||
assert auth_response.status_code == 200
|
||||
assert "text/html" in auth_response.headers["content-type"]
|
||||
|
||||
# Step 2: Submit consent form
|
||||
consent_data = {
|
||||
"client_id": "https://app.example.com",
|
||||
"redirect_uri": "https://app.example.com/callback",
|
||||
"state": "e2e_test_state_12345",
|
||||
"code_challenge": "E9Melhoa2OwvFrEMTJguCHaoeK1t8URWbuGJSstw-cM",
|
||||
"code_challenge_method": "S256",
|
||||
"me": "https://user.example.com",
|
||||
"scope": "",
|
||||
}
|
||||
|
||||
consent_response = e2e_client.post(
|
||||
"/authorize/consent",
|
||||
data=consent_data,
|
||||
follow_redirects=False
|
||||
)
|
||||
|
||||
# Should redirect with authorization code
|
||||
assert consent_response.status_code == 302
|
||||
location = consent_response.headers["location"]
|
||||
assert location.startswith("https://app.example.com/callback")
|
||||
assert "code=" in location
|
||||
assert "state=e2e_test_state_12345" in location
|
||||
|
||||
# Step 3: Extract authorization code
|
||||
auth_code = extract_code_from_redirect(location)
|
||||
assert auth_code is not None
|
||||
|
||||
# Step 4: Exchange code for token
|
||||
token_response = e2e_client.post("/token", data={
|
||||
"grant_type": "authorization_code",
|
||||
"code": auth_code,
|
||||
"client_id": "https://app.example.com",
|
||||
"redirect_uri": "https://app.example.com/callback",
|
||||
})
|
||||
|
||||
# Should receive access token
|
||||
assert token_response.status_code == 200
|
||||
token_data = token_response.json()
|
||||
assert "access_token" in token_data
|
||||
assert token_data["token_type"] == "Bearer"
|
||||
assert token_data["me"] == "https://user.example.com"
|
||||
|
||||
def test_authorization_flow_preserves_state(self, e2e_client, mock_happ_for_e2e):
|
||||
"""Test that state parameter is preserved throughout the flow."""
|
||||
state = "unique_state_for_csrf_protection"
|
||||
|
||||
# Authorization request
|
||||
auth_response = e2e_client.get("/authorize", params={
|
||||
"response_type": "code",
|
||||
"client_id": "https://app.example.com",
|
||||
"redirect_uri": "https://app.example.com/callback",
|
||||
"state": state,
|
||||
"code_challenge": "abc123",
|
||||
"code_challenge_method": "S256",
|
||||
"me": "https://user.example.com",
|
||||
})
|
||||
|
||||
assert auth_response.status_code == 200
|
||||
assert state in auth_response.text
|
||||
|
||||
# Consent submission
|
||||
consent_response = e2e_client.post(
|
||||
"/authorize/consent",
|
||||
data={
|
||||
"client_id": "https://app.example.com",
|
||||
"redirect_uri": "https://app.example.com/callback",
|
||||
"state": state,
|
||||
"code_challenge": "abc123",
|
||||
"code_challenge_method": "S256",
|
||||
"me": "https://user.example.com",
|
||||
"scope": "",
|
||||
},
|
||||
follow_redirects=False
|
||||
)
|
||||
|
||||
# State should be in redirect
|
||||
location = consent_response.headers["location"]
|
||||
assert f"state={state}" in location
|
||||
|
||||
def test_multiple_concurrent_flows(self, e2e_client, mock_happ_for_e2e):
|
||||
"""Test multiple authorization flows can run concurrently."""
|
||||
flows = []
|
||||
|
||||
# Start 3 authorization flows
|
||||
for i in range(3):
|
||||
consent_response = e2e_client.post(
|
||||
"/authorize/consent",
|
||||
data={
|
||||
"client_id": "https://app.example.com",
|
||||
"redirect_uri": "https://app.example.com/callback",
|
||||
"state": f"flow_{i}",
|
||||
"code_challenge": "abc123",
|
||||
"code_challenge_method": "S256",
|
||||
"me": f"https://user{i}.example.com",
|
||||
"scope": "",
|
||||
},
|
||||
follow_redirects=False
|
||||
)
|
||||
|
||||
code = extract_code_from_redirect(consent_response.headers["location"])
|
||||
flows.append((code, f"https://user{i}.example.com"))
|
||||
|
||||
# Exchange all codes - each should work
|
||||
for code, expected_me in flows:
|
||||
token_response = e2e_client.post("/token", data={
|
||||
"grant_type": "authorization_code",
|
||||
"code": code,
|
||||
"client_id": "https://app.example.com",
|
||||
"redirect_uri": "https://app.example.com/callback",
|
||||
})
|
||||
|
||||
assert token_response.status_code == 200
|
||||
assert token_response.json()["me"] == expected_me
|
||||
|
||||
|
||||
@pytest.mark.e2e
|
||||
class TestErrorScenariosE2E:
|
||||
"""E2E tests for error scenarios."""
|
||||
|
||||
def test_invalid_client_id_error_page(self, e2e_client):
|
||||
"""Test invalid client_id shows error page."""
|
||||
response = e2e_client.get("/authorize", params={
|
||||
"client_id": "http://insecure.example.com", # HTTP not allowed
|
||||
"redirect_uri": "http://insecure.example.com/callback",
|
||||
"response_type": "code",
|
||||
})
|
||||
|
||||
assert response.status_code == 400
|
||||
# Should show error page, not redirect
|
||||
assert "text/html" in response.headers["content-type"]
|
||||
|
||||
def test_expired_code_rejected(self, e2e_client, e2e_app, mock_happ_for_e2e):
|
||||
"""Test expired authorization code is rejected."""
|
||||
from gondulf.dependencies import get_code_storage
|
||||
from gondulf.storage import CodeStore
|
||||
|
||||
# Create code storage with very short TTL
|
||||
short_ttl_storage = CodeStore(ttl_seconds=0) # Expire immediately
|
||||
|
||||
# Store a code that will expire immediately
|
||||
code = "expired_test_code_12345"
|
||||
metadata = {
|
||||
"client_id": "https://app.example.com",
|
||||
"redirect_uri": "https://app.example.com/callback",
|
||||
"state": "test",
|
||||
"me": "https://user.example.com",
|
||||
"scope": "",
|
||||
"code_challenge": "abc123",
|
||||
"code_challenge_method": "S256",
|
||||
"created_at": 1000000000,
|
||||
"expires_at": 1000000001,
|
||||
"used": False
|
||||
}
|
||||
short_ttl_storage.store(f"authz:{code}", metadata, ttl=0)
|
||||
|
||||
e2e_app.dependency_overrides[get_code_storage] = lambda: short_ttl_storage
|
||||
|
||||
# Wait a tiny bit for expiration
|
||||
import time
|
||||
time.sleep(0.01)
|
||||
|
||||
# Try to exchange expired code
|
||||
response = e2e_client.post("/token", data={
|
||||
"grant_type": "authorization_code",
|
||||
"code": code,
|
||||
"client_id": "https://app.example.com",
|
||||
"redirect_uri": "https://app.example.com/callback",
|
||||
})
|
||||
|
||||
assert response.status_code == 400
|
||||
assert response.json()["detail"]["error"] == "invalid_grant"
|
||||
|
||||
e2e_app.dependency_overrides.clear()
|
||||
|
||||
def test_code_cannot_be_reused(self, e2e_client, mock_happ_for_e2e):
|
||||
"""Test authorization code single-use enforcement."""
|
||||
# Get a valid code
|
||||
consent_response = e2e_client.post(
|
||||
"/authorize/consent",
|
||||
data={
|
||||
"client_id": "https://app.example.com",
|
||||
"redirect_uri": "https://app.example.com/callback",
|
||||
"state": "test",
|
||||
"code_challenge": "abc123",
|
||||
"code_challenge_method": "S256",
|
||||
"me": "https://user.example.com",
|
||||
"scope": "",
|
||||
},
|
||||
follow_redirects=False
|
||||
)
|
||||
|
||||
code = extract_code_from_redirect(consent_response.headers["location"])
|
||||
|
||||
# First exchange should succeed
|
||||
response1 = e2e_client.post("/token", data={
|
||||
"grant_type": "authorization_code",
|
||||
"code": code,
|
||||
"client_id": "https://app.example.com",
|
||||
"redirect_uri": "https://app.example.com/callback",
|
||||
})
|
||||
assert response1.status_code == 200
|
||||
|
||||
# Second exchange should fail
|
||||
response2 = e2e_client.post("/token", data={
|
||||
"grant_type": "authorization_code",
|
||||
"code": code,
|
||||
"client_id": "https://app.example.com",
|
||||
"redirect_uri": "https://app.example.com/callback",
|
||||
})
|
||||
assert response2.status_code == 400
|
||||
|
||||
def test_wrong_client_id_rejected(self, e2e_client, mock_happ_for_e2e):
|
||||
"""Test token exchange with wrong client_id is rejected."""
|
||||
# Get a code for one client
|
||||
consent_response = e2e_client.post(
|
||||
"/authorize/consent",
|
||||
data={
|
||||
"client_id": "https://app.example.com",
|
||||
"redirect_uri": "https://app.example.com/callback",
|
||||
"state": "test",
|
||||
"code_challenge": "abc123",
|
||||
"code_challenge_method": "S256",
|
||||
"me": "https://user.example.com",
|
||||
"scope": "",
|
||||
},
|
||||
follow_redirects=False
|
||||
)
|
||||
|
||||
code = extract_code_from_redirect(consent_response.headers["location"])
|
||||
|
||||
# Try to exchange with different client_id
|
||||
response = e2e_client.post("/token", data={
|
||||
"grant_type": "authorization_code",
|
||||
"code": code,
|
||||
"client_id": "https://different-app.example.com", # Wrong client
|
||||
"redirect_uri": "https://app.example.com/callback",
|
||||
})
|
||||
|
||||
assert response.status_code == 400
|
||||
assert response.json()["detail"]["error"] == "invalid_client"
|
||||
|
||||
|
||||
@pytest.mark.e2e
|
||||
class TestTokenUsageE2E:
|
||||
"""E2E tests for token usage after obtaining it."""
|
||||
|
||||
def test_obtained_token_has_correct_format(self, e2e_client, mock_happ_for_e2e):
|
||||
"""Test the token obtained through E2E flow has correct format."""
|
||||
# Complete the flow
|
||||
consent_response = e2e_client.post(
|
||||
"/authorize/consent",
|
||||
data={
|
||||
"client_id": "https://app.example.com",
|
||||
"redirect_uri": "https://app.example.com/callback",
|
||||
"state": "test",
|
||||
"code_challenge": "abc123",
|
||||
"code_challenge_method": "S256",
|
||||
"me": "https://user.example.com",
|
||||
"scope": "",
|
||||
},
|
||||
follow_redirects=False
|
||||
)
|
||||
|
||||
code = extract_code_from_redirect(consent_response.headers["location"])
|
||||
|
||||
token_response = e2e_client.post("/token", data={
|
||||
"grant_type": "authorization_code",
|
||||
"code": code,
|
||||
"client_id": "https://app.example.com",
|
||||
"redirect_uri": "https://app.example.com/callback",
|
||||
})
|
||||
|
||||
assert token_response.status_code == 200
|
||||
token_data = token_response.json()
|
||||
|
||||
# Verify token has correct format
|
||||
assert "access_token" in token_data
|
||||
assert len(token_data["access_token"]) >= 32 # Should be substantial
|
||||
assert token_data["token_type"] == "Bearer"
|
||||
assert token_data["me"] == "https://user.example.com"
|
||||
|
||||
def test_token_response_includes_all_fields(self, e2e_client, mock_happ_for_e2e):
|
||||
"""Test token response includes all required IndieAuth fields."""
|
||||
# Complete the flow
|
||||
consent_response = e2e_client.post(
|
||||
"/authorize/consent",
|
||||
data={
|
||||
"client_id": "https://app.example.com",
|
||||
"redirect_uri": "https://app.example.com/callback",
|
||||
"state": "test",
|
||||
"code_challenge": "abc123",
|
||||
"code_challenge_method": "S256",
|
||||
"me": "https://user.example.com",
|
||||
"scope": "profile",
|
||||
},
|
||||
follow_redirects=False
|
||||
)
|
||||
|
||||
code = extract_code_from_redirect(consent_response.headers["location"])
|
||||
|
||||
token_response = e2e_client.post("/token", data={
|
||||
"grant_type": "authorization_code",
|
||||
"code": code,
|
||||
"client_id": "https://app.example.com",
|
||||
"redirect_uri": "https://app.example.com/callback",
|
||||
})
|
||||
|
||||
assert token_response.status_code == 200
|
||||
token_data = token_response.json()
|
||||
|
||||
# All required IndieAuth fields
|
||||
assert "access_token" in token_data
|
||||
assert "token_type" in token_data
|
||||
assert "me" in token_data
|
||||
assert "scope" in token_data
|
||||
260
tests/e2e/test_error_scenarios.py
Normal file
260
tests/e2e/test_error_scenarios.py
Normal file
@@ -0,0 +1,260 @@
|
||||
"""
|
||||
End-to-end tests for error scenarios and edge cases.
|
||||
|
||||
Tests various error conditions and ensures proper error handling throughout the system.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def error_app(monkeypatch, tmp_path):
|
||||
"""Create app for error scenario testing."""
|
||||
db_path = tmp_path / "test.db"
|
||||
|
||||
monkeypatch.setenv("GONDULF_SECRET_KEY", "a" * 32)
|
||||
monkeypatch.setenv("GONDULF_BASE_URL", "https://auth.example.com")
|
||||
monkeypatch.setenv("GONDULF_DATABASE_URL", f"sqlite:///{db_path}")
|
||||
monkeypatch.setenv("GONDULF_DEBUG", "true")
|
||||
|
||||
from gondulf.main import app
|
||||
return app
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def error_client(error_app):
|
||||
"""Create test client for error scenario tests."""
|
||||
with TestClient(error_app) as client:
|
||||
yield client
|
||||
|
||||
|
||||
@pytest.mark.e2e
|
||||
class TestAuthorizationErrors:
|
||||
"""E2E tests for authorization endpoint errors."""
|
||||
|
||||
def test_missing_all_parameters(self, error_client):
|
||||
"""Test authorization request with no parameters."""
|
||||
response = error_client.get("/authorize")
|
||||
|
||||
assert response.status_code == 400
|
||||
|
||||
def test_http_client_id_rejected(self, error_client):
|
||||
"""Test HTTP (non-HTTPS) client_id is rejected."""
|
||||
response = error_client.get("/authorize", params={
|
||||
"client_id": "http://insecure.example.com",
|
||||
"redirect_uri": "http://insecure.example.com/callback",
|
||||
"response_type": "code",
|
||||
"state": "test",
|
||||
})
|
||||
|
||||
assert response.status_code == 400
|
||||
assert "https" in response.text.lower()
|
||||
|
||||
def test_mismatched_redirect_uri_domain(self, error_client):
|
||||
"""Test redirect_uri must match client_id domain."""
|
||||
response = error_client.get("/authorize", params={
|
||||
"client_id": "https://legitimate-app.example.com",
|
||||
"redirect_uri": "https://evil-site.example.com/steal",
|
||||
"response_type": "code",
|
||||
"state": "test",
|
||||
})
|
||||
|
||||
assert response.status_code == 400
|
||||
|
||||
def test_invalid_response_type_redirects(self, error_client):
|
||||
"""Test invalid response_type redirects with error."""
|
||||
response = error_client.get("/authorize", params={
|
||||
"client_id": "https://app.example.com",
|
||||
"redirect_uri": "https://app.example.com/callback",
|
||||
"response_type": "implicit", # Not supported
|
||||
"state": "test123",
|
||||
}, follow_redirects=False)
|
||||
|
||||
assert response.status_code == 302
|
||||
location = response.headers["location"]
|
||||
assert "error=unsupported_response_type" in location
|
||||
assert "state=test123" in location
|
||||
|
||||
|
||||
@pytest.mark.e2e
|
||||
class TestTokenEndpointErrors:
|
||||
"""E2E tests for token endpoint errors."""
|
||||
|
||||
def test_invalid_grant_type(self, error_client):
|
||||
"""Test unsupported grant_type returns error."""
|
||||
response = error_client.post("/token", data={
|
||||
"grant_type": "client_credentials",
|
||||
"code": "some_code",
|
||||
"client_id": "https://app.example.com",
|
||||
"redirect_uri": "https://app.example.com/callback",
|
||||
})
|
||||
|
||||
assert response.status_code == 400
|
||||
data = response.json()
|
||||
assert data["detail"]["error"] == "unsupported_grant_type"
|
||||
|
||||
def test_missing_grant_type(self, error_client):
|
||||
"""Test missing grant_type returns validation error."""
|
||||
response = error_client.post("/token", data={
|
||||
"code": "some_code",
|
||||
"client_id": "https://app.example.com",
|
||||
"redirect_uri": "https://app.example.com/callback",
|
||||
})
|
||||
|
||||
# FastAPI validation error
|
||||
assert response.status_code == 422
|
||||
|
||||
def test_nonexistent_code(self, error_client):
|
||||
"""Test nonexistent authorization code returns error."""
|
||||
response = error_client.post("/token", data={
|
||||
"grant_type": "authorization_code",
|
||||
"code": "completely_made_up_code_12345",
|
||||
"client_id": "https://app.example.com",
|
||||
"redirect_uri": "https://app.example.com/callback",
|
||||
})
|
||||
|
||||
assert response.status_code == 400
|
||||
data = response.json()
|
||||
assert data["detail"]["error"] == "invalid_grant"
|
||||
|
||||
def test_get_method_not_allowed(self, error_client):
|
||||
"""Test GET method not allowed on token endpoint."""
|
||||
response = error_client.get("/token")
|
||||
|
||||
assert response.status_code == 405
|
||||
|
||||
|
||||
@pytest.mark.e2e
|
||||
class TestVerificationErrors:
|
||||
"""E2E tests for verification endpoint errors."""
|
||||
|
||||
def test_invalid_me_url(self, error_client):
|
||||
"""Test invalid me URL format."""
|
||||
response = error_client.post(
|
||||
"/api/verify/start",
|
||||
data={"me": "not-a-url"}
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["success"] is False
|
||||
assert data["error"] == "invalid_me_url"
|
||||
|
||||
def test_invalid_code_verification(self, error_client):
|
||||
"""Test verification with invalid code."""
|
||||
response = error_client.post(
|
||||
"/api/verify/code",
|
||||
data={"domain": "example.com", "code": "000000"}
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["success"] is False
|
||||
|
||||
|
||||
@pytest.mark.e2e
|
||||
class TestSecurityErrorHandling:
|
||||
"""E2E tests for security-related error handling."""
|
||||
|
||||
def test_xss_in_state_escaped(self, error_client):
|
||||
"""Test XSS attempt in state parameter is escaped."""
|
||||
xss_payload = "<script>alert('xss')</script>"
|
||||
|
||||
response = error_client.get("/authorize", params={
|
||||
"client_id": "https://app.example.com",
|
||||
"redirect_uri": "https://app.example.com/callback",
|
||||
"response_type": "token", # Will error and redirect
|
||||
"state": xss_payload,
|
||||
}, follow_redirects=False)
|
||||
|
||||
# Should redirect with error
|
||||
assert response.status_code == 302
|
||||
location = response.headers["location"]
|
||||
# Script tags should be URL encoded, not raw
|
||||
assert "<script>" not in location
|
||||
|
||||
def test_errors_have_security_headers(self, error_client):
|
||||
"""Test error responses include security headers."""
|
||||
response = error_client.get("/authorize") # Missing params = error
|
||||
|
||||
assert response.status_code == 400
|
||||
assert "X-Frame-Options" in response.headers
|
||||
assert response.headers["X-Frame-Options"] == "DENY"
|
||||
|
||||
def test_error_response_is_json_for_api(self, error_client):
|
||||
"""Test API error responses are JSON formatted."""
|
||||
response = error_client.post("/token", data={
|
||||
"grant_type": "authorization_code",
|
||||
"code": "invalid",
|
||||
"client_id": "https://app.example.com",
|
||||
"redirect_uri": "https://app.example.com/callback",
|
||||
})
|
||||
|
||||
assert response.status_code == 400
|
||||
# Should be JSON
|
||||
assert "application/json" in response.headers["content-type"]
|
||||
data = response.json()
|
||||
assert "detail" in data
|
||||
|
||||
|
||||
@pytest.mark.e2e
|
||||
class TestEdgeCases:
|
||||
"""E2E tests for edge cases."""
|
||||
|
||||
def test_empty_scope_accepted(self, error_client):
|
||||
"""Test empty scope is accepted."""
|
||||
from unittest.mock import AsyncMock, patch
|
||||
from gondulf.services.happ_parser import ClientMetadata
|
||||
|
||||
metadata = ClientMetadata(
|
||||
name="Test App",
|
||||
url="https://app.example.com",
|
||||
logo=None
|
||||
)
|
||||
|
||||
with patch('gondulf.services.happ_parser.HAppParser.fetch_and_parse', new_callable=AsyncMock) as mock:
|
||||
mock.return_value = metadata
|
||||
|
||||
response = error_client.get("/authorize", params={
|
||||
"client_id": "https://app.example.com",
|
||||
"redirect_uri": "https://app.example.com/callback",
|
||||
"response_type": "code",
|
||||
"state": "test",
|
||||
"code_challenge": "abc123",
|
||||
"code_challenge_method": "S256",
|
||||
"me": "https://user.example.com",
|
||||
"scope": "", # Empty scope
|
||||
})
|
||||
|
||||
# Should show consent page
|
||||
assert response.status_code == 200
|
||||
|
||||
def test_very_long_state_handled(self, error_client):
|
||||
"""Test very long state parameter is handled."""
|
||||
from unittest.mock import AsyncMock, patch
|
||||
from gondulf.services.happ_parser import ClientMetadata
|
||||
|
||||
metadata = ClientMetadata(
|
||||
name="Test App",
|
||||
url="https://app.example.com",
|
||||
logo=None
|
||||
)
|
||||
|
||||
long_state = "x" * 1000
|
||||
|
||||
with patch('gondulf.services.happ_parser.HAppParser.fetch_and_parse', new_callable=AsyncMock) as mock:
|
||||
mock.return_value = metadata
|
||||
|
||||
response = error_client.get("/authorize", params={
|
||||
"client_id": "https://app.example.com",
|
||||
"redirect_uri": "https://app.example.com/callback",
|
||||
"response_type": "code",
|
||||
"state": long_state,
|
||||
"code_challenge": "abc123",
|
||||
"code_challenge_method": "S256",
|
||||
"me": "https://user.example.com",
|
||||
})
|
||||
|
||||
# Should handle without error
|
||||
assert response.status_code == 200
|
||||
1
tests/integration/api/__init__.py
Normal file
1
tests/integration/api/__init__.py
Normal file
@@ -0,0 +1 @@
|
||||
"""API integration tests for Gondulf IndieAuth server."""
|
||||
337
tests/integration/api/test_authorization_flow.py
Normal file
337
tests/integration/api/test_authorization_flow.py
Normal file
@@ -0,0 +1,337 @@
|
||||
"""
|
||||
Integration tests for authorization endpoint flow.
|
||||
|
||||
Tests the complete authorization endpoint behavior including parameter validation,
|
||||
client metadata fetching, consent form rendering, and code generation.
|
||||
"""
|
||||
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
from unittest.mock import AsyncMock, Mock, patch
|
||||
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def auth_app(monkeypatch, tmp_path):
|
||||
"""Create app for authorization testing."""
|
||||
db_path = tmp_path / "test.db"
|
||||
|
||||
monkeypatch.setenv("GONDULF_SECRET_KEY", "a" * 32)
|
||||
monkeypatch.setenv("GONDULF_BASE_URL", "https://auth.example.com")
|
||||
monkeypatch.setenv("GONDULF_DATABASE_URL", f"sqlite:///{db_path}")
|
||||
monkeypatch.setenv("GONDULF_DEBUG", "true")
|
||||
|
||||
from gondulf.main import app
|
||||
return app
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def auth_client(auth_app):
|
||||
"""Create test client for authorization tests."""
|
||||
with TestClient(auth_app) as client:
|
||||
yield client
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_happ_fetch():
|
||||
"""Mock h-app parser to avoid network calls."""
|
||||
from gondulf.services.happ_parser import ClientMetadata
|
||||
|
||||
metadata = ClientMetadata(
|
||||
name="Test Application",
|
||||
url="https://app.example.com",
|
||||
logo="https://app.example.com/logo.png"
|
||||
)
|
||||
|
||||
with patch('gondulf.services.happ_parser.HAppParser.fetch_and_parse', new_callable=AsyncMock) as mock:
|
||||
mock.return_value = metadata
|
||||
yield mock
|
||||
|
||||
|
||||
class TestAuthorizationEndpointValidation:
|
||||
"""Tests for authorization endpoint parameter validation."""
|
||||
|
||||
def test_missing_client_id_returns_error(self, auth_client):
|
||||
"""Test that missing client_id returns 400 error."""
|
||||
response = auth_client.get("/authorize", params={
|
||||
"redirect_uri": "https://app.example.com/callback",
|
||||
"response_type": "code",
|
||||
"state": "test123",
|
||||
})
|
||||
|
||||
assert response.status_code == 400
|
||||
assert "client_id" in response.text.lower()
|
||||
|
||||
def test_missing_redirect_uri_returns_error(self, auth_client):
|
||||
"""Test that missing redirect_uri returns 400 error."""
|
||||
response = auth_client.get("/authorize", params={
|
||||
"client_id": "https://app.example.com",
|
||||
"response_type": "code",
|
||||
"state": "test123",
|
||||
})
|
||||
|
||||
assert response.status_code == 400
|
||||
assert "redirect_uri" in response.text.lower()
|
||||
|
||||
def test_http_client_id_rejected(self, auth_client):
|
||||
"""Test that HTTP client_id (non-HTTPS) is rejected."""
|
||||
response = auth_client.get("/authorize", params={
|
||||
"client_id": "http://app.example.com", # HTTP not allowed
|
||||
"redirect_uri": "https://app.example.com/callback",
|
||||
"response_type": "code",
|
||||
"state": "test123",
|
||||
})
|
||||
|
||||
assert response.status_code == 400
|
||||
assert "https" in response.text.lower()
|
||||
|
||||
def test_mismatched_redirect_uri_rejected(self, auth_client):
|
||||
"""Test that redirect_uri not matching client_id domain is rejected."""
|
||||
response = auth_client.get("/authorize", params={
|
||||
"client_id": "https://app.example.com",
|
||||
"redirect_uri": "https://evil.example.com/callback", # Different domain
|
||||
"response_type": "code",
|
||||
"state": "test123",
|
||||
})
|
||||
|
||||
assert response.status_code == 400
|
||||
assert "redirect_uri" in response.text.lower()
|
||||
|
||||
|
||||
class TestAuthorizationEndpointRedirectErrors:
|
||||
"""Tests for errors that redirect back to the client."""
|
||||
|
||||
@pytest.fixture
|
||||
def valid_params(self):
|
||||
"""Valid base authorization parameters."""
|
||||
return {
|
||||
"client_id": "https://app.example.com",
|
||||
"redirect_uri": "https://app.example.com/callback",
|
||||
"state": "test123",
|
||||
}
|
||||
|
||||
def test_invalid_response_type_redirects_with_error(self, auth_client, valid_params, mock_happ_fetch):
|
||||
"""Test invalid response_type redirects with error parameter."""
|
||||
params = valid_params.copy()
|
||||
params["response_type"] = "token" # Invalid - only "code" is supported
|
||||
|
||||
response = auth_client.get("/authorize", params=params, follow_redirects=False)
|
||||
|
||||
assert response.status_code == 302
|
||||
location = response.headers["location"]
|
||||
assert "error=unsupported_response_type" in location
|
||||
assert "state=test123" in location
|
||||
|
||||
def test_missing_code_challenge_redirects_with_error(self, auth_client, valid_params, mock_happ_fetch):
|
||||
"""Test missing PKCE code_challenge redirects with error."""
|
||||
params = valid_params.copy()
|
||||
params["response_type"] = "code"
|
||||
params["me"] = "https://user.example.com"
|
||||
# Missing code_challenge
|
||||
|
||||
response = auth_client.get("/authorize", params=params, follow_redirects=False)
|
||||
|
||||
assert response.status_code == 302
|
||||
location = response.headers["location"]
|
||||
assert "error=invalid_request" in location
|
||||
assert "code_challenge" in location.lower()
|
||||
|
||||
def test_invalid_code_challenge_method_redirects_with_error(self, auth_client, valid_params, mock_happ_fetch):
|
||||
"""Test invalid code_challenge_method redirects with error."""
|
||||
params = valid_params.copy()
|
||||
params["response_type"] = "code"
|
||||
params["me"] = "https://user.example.com"
|
||||
params["code_challenge"] = "abc123"
|
||||
params["code_challenge_method"] = "plain" # Invalid - only S256 supported
|
||||
|
||||
response = auth_client.get("/authorize", params=params, follow_redirects=False)
|
||||
|
||||
assert response.status_code == 302
|
||||
location = response.headers["location"]
|
||||
assert "error=invalid_request" in location
|
||||
assert "S256" in location
|
||||
|
||||
def test_missing_me_parameter_redirects_with_error(self, auth_client, valid_params, mock_happ_fetch):
|
||||
"""Test missing me parameter redirects with error."""
|
||||
params = valid_params.copy()
|
||||
params["response_type"] = "code"
|
||||
params["code_challenge"] = "abc123"
|
||||
params["code_challenge_method"] = "S256"
|
||||
# Missing me parameter
|
||||
|
||||
response = auth_client.get("/authorize", params=params, follow_redirects=False)
|
||||
|
||||
assert response.status_code == 302
|
||||
location = response.headers["location"]
|
||||
assert "error=invalid_request" in location
|
||||
assert "me" in location.lower()
|
||||
|
||||
def test_invalid_me_url_redirects_with_error(self, auth_client, valid_params, mock_happ_fetch):
|
||||
"""Test invalid me URL redirects with error."""
|
||||
params = valid_params.copy()
|
||||
params["response_type"] = "code"
|
||||
params["code_challenge"] = "abc123"
|
||||
params["code_challenge_method"] = "S256"
|
||||
params["me"] = "not-a-valid-url"
|
||||
|
||||
response = auth_client.get("/authorize", params=params, follow_redirects=False)
|
||||
|
||||
assert response.status_code == 302
|
||||
location = response.headers["location"]
|
||||
assert "error=invalid_request" in location
|
||||
|
||||
|
||||
class TestAuthorizationConsentPage:
|
||||
"""Tests for the consent page rendering."""
|
||||
|
||||
@pytest.fixture
|
||||
def complete_params(self):
|
||||
"""Complete valid authorization parameters."""
|
||||
return {
|
||||
"client_id": "https://app.example.com",
|
||||
"redirect_uri": "https://app.example.com/callback",
|
||||
"response_type": "code",
|
||||
"state": "test123",
|
||||
"code_challenge": "E9Melhoa2OwvFrEMTJguCHaoeK1t8URWbuGJSstw-cM",
|
||||
"code_challenge_method": "S256",
|
||||
"me": "https://user.example.com",
|
||||
}
|
||||
|
||||
def test_valid_request_shows_consent_page(self, auth_client, complete_params, mock_happ_fetch):
|
||||
"""Test valid authorization request shows consent page."""
|
||||
response = auth_client.get("/authorize", params=complete_params)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert "text/html" in response.headers["content-type"]
|
||||
# Page should contain client information
|
||||
assert "app.example.com" in response.text or "Test Application" in response.text
|
||||
|
||||
def test_consent_page_contains_required_fields(self, auth_client, complete_params, mock_happ_fetch):
|
||||
"""Test consent page contains all required form fields."""
|
||||
response = auth_client.get("/authorize", params=complete_params)
|
||||
|
||||
assert response.status_code == 200
|
||||
# Check for hidden form fields that will be POSTed
|
||||
assert "client_id" in response.text
|
||||
assert "redirect_uri" in response.text
|
||||
assert "code_challenge" in response.text
|
||||
|
||||
def test_consent_page_displays_client_metadata(self, auth_client, complete_params, mock_happ_fetch):
|
||||
"""Test consent page displays client h-app metadata."""
|
||||
response = auth_client.get("/authorize", params=complete_params)
|
||||
|
||||
assert response.status_code == 200
|
||||
# Should show client name from h-app
|
||||
assert "Test Application" in response.text or "app.example.com" in response.text
|
||||
|
||||
def test_consent_page_preserves_state(self, auth_client, complete_params, mock_happ_fetch):
|
||||
"""Test consent page preserves state parameter."""
|
||||
response = auth_client.get("/authorize", params=complete_params)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert "test123" in response.text
|
||||
|
||||
|
||||
class TestAuthorizationConsentSubmission:
|
||||
"""Tests for consent form submission."""
|
||||
|
||||
@pytest.fixture
|
||||
def consent_form_data(self):
|
||||
"""Valid consent form data."""
|
||||
return {
|
||||
"client_id": "https://app.example.com",
|
||||
"redirect_uri": "https://app.example.com/callback",
|
||||
"state": "test123",
|
||||
"code_challenge": "E9Melhoa2OwvFrEMTJguCHaoeK1t8URWbuGJSstw-cM",
|
||||
"code_challenge_method": "S256",
|
||||
"me": "https://user.example.com",
|
||||
"scope": "",
|
||||
}
|
||||
|
||||
def test_consent_submission_redirects_with_code(self, auth_client, consent_form_data):
|
||||
"""Test consent submission redirects to client with authorization code."""
|
||||
response = auth_client.post(
|
||||
"/authorize/consent",
|
||||
data=consent_form_data,
|
||||
follow_redirects=False
|
||||
)
|
||||
|
||||
assert response.status_code == 302
|
||||
location = response.headers["location"]
|
||||
assert location.startswith("https://app.example.com/callback")
|
||||
assert "code=" in location
|
||||
assert "state=test123" in location
|
||||
|
||||
def test_consent_submission_generates_unique_codes(self, auth_client, consent_form_data):
|
||||
"""Test each consent generates a unique authorization code."""
|
||||
# First submission
|
||||
response1 = auth_client.post(
|
||||
"/authorize/consent",
|
||||
data=consent_form_data,
|
||||
follow_redirects=False
|
||||
)
|
||||
location1 = response1.headers["location"]
|
||||
|
||||
# Second submission
|
||||
response2 = auth_client.post(
|
||||
"/authorize/consent",
|
||||
data=consent_form_data,
|
||||
follow_redirects=False
|
||||
)
|
||||
location2 = response2.headers["location"]
|
||||
|
||||
# Extract codes
|
||||
from tests.conftest import extract_code_from_redirect
|
||||
code1 = extract_code_from_redirect(location1)
|
||||
code2 = extract_code_from_redirect(location2)
|
||||
|
||||
assert code1 != code2
|
||||
|
||||
def test_authorization_code_stored_for_exchange(self, auth_client, consent_form_data):
|
||||
"""Test authorization code is stored for later token exchange."""
|
||||
response = auth_client.post(
|
||||
"/authorize/consent",
|
||||
data=consent_form_data,
|
||||
follow_redirects=False
|
||||
)
|
||||
|
||||
from tests.conftest import extract_code_from_redirect
|
||||
code = extract_code_from_redirect(response.headers["location"])
|
||||
|
||||
# Code should be non-empty and URL-safe
|
||||
assert code is not None
|
||||
assert len(code) > 20 # Should be a substantial code
|
||||
|
||||
|
||||
class TestAuthorizationSecurityHeaders:
|
||||
"""Tests for security headers on authorization endpoints."""
|
||||
|
||||
def test_authorization_page_has_security_headers(self, auth_client, mock_happ_fetch):
|
||||
"""Test authorization page includes security headers."""
|
||||
params = {
|
||||
"client_id": "https://app.example.com",
|
||||
"redirect_uri": "https://app.example.com/callback",
|
||||
"response_type": "code",
|
||||
"state": "test123",
|
||||
"code_challenge": "abc123",
|
||||
"code_challenge_method": "S256",
|
||||
"me": "https://user.example.com",
|
||||
}
|
||||
response = auth_client.get("/authorize", params=params)
|
||||
|
||||
assert "X-Frame-Options" in response.headers
|
||||
assert "X-Content-Type-Options" in response.headers
|
||||
assert response.headers["X-Frame-Options"] == "DENY"
|
||||
|
||||
def test_error_pages_have_security_headers(self, auth_client):
|
||||
"""Test error pages include security headers."""
|
||||
# Request without client_id should return error page
|
||||
response = auth_client.get("/authorize", params={
|
||||
"redirect_uri": "https://app.example.com/callback"
|
||||
})
|
||||
|
||||
assert response.status_code == 400
|
||||
assert "X-Frame-Options" in response.headers
|
||||
assert "X-Content-Type-Options" in response.headers
|
||||
137
tests/integration/api/test_metadata.py
Normal file
137
tests/integration/api/test_metadata.py
Normal file
@@ -0,0 +1,137 @@
|
||||
"""
|
||||
Integration tests for OAuth 2.0 metadata endpoint.
|
||||
|
||||
Tests the /.well-known/oauth-authorization-server endpoint per RFC 8414.
|
||||
"""
|
||||
|
||||
import json
|
||||
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def metadata_app(monkeypatch, tmp_path):
|
||||
"""Create app for metadata testing."""
|
||||
db_path = tmp_path / "test.db"
|
||||
|
||||
monkeypatch.setenv("GONDULF_SECRET_KEY", "a" * 32)
|
||||
monkeypatch.setenv("GONDULF_BASE_URL", "https://auth.example.com")
|
||||
monkeypatch.setenv("GONDULF_DATABASE_URL", f"sqlite:///{db_path}")
|
||||
monkeypatch.setenv("GONDULF_DEBUG", "true")
|
||||
|
||||
from gondulf.main import app
|
||||
return app
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def metadata_client(metadata_app):
|
||||
"""Create test client for metadata tests."""
|
||||
with TestClient(metadata_app) as client:
|
||||
yield client
|
||||
|
||||
|
||||
class TestMetadataEndpoint:
|
||||
"""Tests for OAuth 2.0 Authorization Server Metadata endpoint."""
|
||||
|
||||
def test_metadata_returns_json(self, metadata_client):
|
||||
"""Test metadata endpoint returns JSON response."""
|
||||
response = metadata_client.get("/.well-known/oauth-authorization-server")
|
||||
|
||||
assert response.status_code == 200
|
||||
assert "application/json" in response.headers["content-type"]
|
||||
|
||||
def test_metadata_includes_issuer(self, metadata_client):
|
||||
"""Test metadata includes issuer field."""
|
||||
response = metadata_client.get("/.well-known/oauth-authorization-server")
|
||||
|
||||
data = response.json()
|
||||
assert "issuer" in data
|
||||
assert data["issuer"] == "https://auth.example.com"
|
||||
|
||||
def test_metadata_includes_authorization_endpoint(self, metadata_client):
|
||||
"""Test metadata includes authorization endpoint."""
|
||||
response = metadata_client.get("/.well-known/oauth-authorization-server")
|
||||
|
||||
data = response.json()
|
||||
assert "authorization_endpoint" in data
|
||||
assert data["authorization_endpoint"] == "https://auth.example.com/authorize"
|
||||
|
||||
def test_metadata_includes_token_endpoint(self, metadata_client):
|
||||
"""Test metadata includes token endpoint."""
|
||||
response = metadata_client.get("/.well-known/oauth-authorization-server")
|
||||
|
||||
data = response.json()
|
||||
assert "token_endpoint" in data
|
||||
assert data["token_endpoint"] == "https://auth.example.com/token"
|
||||
|
||||
def test_metadata_includes_response_types(self, metadata_client):
|
||||
"""Test metadata includes supported response types."""
|
||||
response = metadata_client.get("/.well-known/oauth-authorization-server")
|
||||
|
||||
data = response.json()
|
||||
assert "response_types_supported" in data
|
||||
assert "code" in data["response_types_supported"]
|
||||
|
||||
def test_metadata_includes_grant_types(self, metadata_client):
|
||||
"""Test metadata includes supported grant types."""
|
||||
response = metadata_client.get("/.well-known/oauth-authorization-server")
|
||||
|
||||
data = response.json()
|
||||
assert "grant_types_supported" in data
|
||||
assert "authorization_code" in data["grant_types_supported"]
|
||||
|
||||
def test_metadata_includes_token_auth_methods(self, metadata_client):
|
||||
"""Test metadata includes token endpoint auth methods."""
|
||||
response = metadata_client.get("/.well-known/oauth-authorization-server")
|
||||
|
||||
data = response.json()
|
||||
assert "token_endpoint_auth_methods_supported" in data
|
||||
assert "none" in data["token_endpoint_auth_methods_supported"]
|
||||
|
||||
|
||||
class TestMetadataCaching:
|
||||
"""Tests for metadata endpoint caching behavior."""
|
||||
|
||||
def test_metadata_includes_cache_header(self, metadata_client):
|
||||
"""Test metadata endpoint includes Cache-Control header."""
|
||||
response = metadata_client.get("/.well-known/oauth-authorization-server")
|
||||
|
||||
assert "Cache-Control" in response.headers
|
||||
# Should allow caching
|
||||
assert "public" in response.headers["Cache-Control"]
|
||||
assert "max-age" in response.headers["Cache-Control"]
|
||||
|
||||
def test_metadata_is_cacheable(self, metadata_client):
|
||||
"""Test metadata endpoint allows public caching."""
|
||||
response = metadata_client.get("/.well-known/oauth-authorization-server")
|
||||
|
||||
cache_control = response.headers["Cache-Control"]
|
||||
# Should be cacheable for a reasonable time
|
||||
assert "public" in cache_control
|
||||
|
||||
|
||||
class TestMetadataSecurity:
|
||||
"""Security tests for metadata endpoint."""
|
||||
|
||||
def test_metadata_includes_security_headers(self, metadata_client):
|
||||
"""Test metadata endpoint includes security headers."""
|
||||
response = metadata_client.get("/.well-known/oauth-authorization-server")
|
||||
|
||||
assert "X-Frame-Options" in response.headers
|
||||
assert "X-Content-Type-Options" in response.headers
|
||||
|
||||
def test_metadata_requires_no_authentication(self, metadata_client):
|
||||
"""Test metadata endpoint is publicly accessible."""
|
||||
response = metadata_client.get("/.well-known/oauth-authorization-server")
|
||||
|
||||
# Should work without any authentication
|
||||
assert response.status_code == 200
|
||||
|
||||
def test_metadata_returns_valid_json(self, metadata_client):
|
||||
"""Test metadata returns valid parseable JSON."""
|
||||
response = metadata_client.get("/.well-known/oauth-authorization-server")
|
||||
|
||||
# Should not raise
|
||||
data = json.loads(response.content)
|
||||
assert isinstance(data, dict)
|
||||
328
tests/integration/api/test_token_flow.py
Normal file
328
tests/integration/api/test_token_flow.py
Normal file
@@ -0,0 +1,328 @@
|
||||
"""
|
||||
Integration tests for token endpoint flow.
|
||||
|
||||
Tests the complete token exchange flow including authorization code validation,
|
||||
PKCE verification, token generation, and error handling.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def token_app(monkeypatch, tmp_path):
|
||||
"""Create app for token testing."""
|
||||
db_path = tmp_path / "test.db"
|
||||
|
||||
monkeypatch.setenv("GONDULF_SECRET_KEY", "a" * 32)
|
||||
monkeypatch.setenv("GONDULF_BASE_URL", "https://auth.example.com")
|
||||
monkeypatch.setenv("GONDULF_DATABASE_URL", f"sqlite:///{db_path}")
|
||||
monkeypatch.setenv("GONDULF_DEBUG", "true")
|
||||
|
||||
from gondulf.main import app
|
||||
return app
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def token_client(token_app):
|
||||
"""Create test client for token tests."""
|
||||
with TestClient(token_app) as client:
|
||||
yield client
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def setup_auth_code(token_app, test_code_storage):
|
||||
"""Setup a valid authorization code for testing."""
|
||||
from gondulf.dependencies import get_code_storage
|
||||
|
||||
code = "integration_test_code_12345"
|
||||
metadata = {
|
||||
"client_id": "https://app.example.com",
|
||||
"redirect_uri": "https://app.example.com/callback",
|
||||
"state": "xyz123",
|
||||
"me": "https://user.example.com",
|
||||
"scope": "",
|
||||
"code_challenge": "E9Melhoa2OwvFrEMTJguCHaoeK1t8URWbuGJSstw-cM",
|
||||
"code_challenge_method": "S256",
|
||||
"created_at": 1234567890,
|
||||
"expires_at": 1234568490,
|
||||
"used": False
|
||||
}
|
||||
|
||||
# Override the code storage dependency
|
||||
token_app.dependency_overrides[get_code_storage] = lambda: test_code_storage
|
||||
test_code_storage.store(f"authz:{code}", metadata)
|
||||
|
||||
yield code, metadata, test_code_storage
|
||||
|
||||
token_app.dependency_overrides.clear()
|
||||
|
||||
|
||||
class TestTokenExchangeIntegration:
|
||||
"""Integration tests for successful token exchange."""
|
||||
|
||||
def test_valid_code_exchange_returns_token(self, token_client, setup_auth_code):
|
||||
"""Test valid authorization code exchange returns access token."""
|
||||
code, metadata, _ = setup_auth_code
|
||||
|
||||
response = token_client.post("/token", data={
|
||||
"grant_type": "authorization_code",
|
||||
"code": code,
|
||||
"client_id": metadata["client_id"],
|
||||
"redirect_uri": metadata["redirect_uri"],
|
||||
})
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert "access_token" in data
|
||||
assert data["token_type"] == "Bearer"
|
||||
assert data["me"] == metadata["me"]
|
||||
|
||||
def test_token_response_format_matches_oauth2(self, token_client, setup_auth_code):
|
||||
"""Test token response matches OAuth 2.0 specification format."""
|
||||
code, metadata, _ = setup_auth_code
|
||||
|
||||
response = token_client.post("/token", data={
|
||||
"grant_type": "authorization_code",
|
||||
"code": code,
|
||||
"client_id": metadata["client_id"],
|
||||
"redirect_uri": metadata["redirect_uri"],
|
||||
})
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
|
||||
# Required fields per OAuth 2.0 / IndieAuth
|
||||
assert "access_token" in data
|
||||
assert "token_type" in data
|
||||
assert "me" in data
|
||||
|
||||
# Token should be substantial
|
||||
assert len(data["access_token"]) >= 32
|
||||
|
||||
def test_token_response_includes_cache_headers(self, token_client, setup_auth_code):
|
||||
"""Test token response includes required cache headers."""
|
||||
code, metadata, _ = setup_auth_code
|
||||
|
||||
response = token_client.post("/token", data={
|
||||
"grant_type": "authorization_code",
|
||||
"code": code,
|
||||
"client_id": metadata["client_id"],
|
||||
"redirect_uri": metadata["redirect_uri"],
|
||||
})
|
||||
|
||||
assert response.status_code == 200
|
||||
# OAuth 2.0 requires no-store
|
||||
assert response.headers["Cache-Control"] == "no-store"
|
||||
assert response.headers["Pragma"] == "no-cache"
|
||||
|
||||
def test_authorization_code_single_use(self, token_client, setup_auth_code):
|
||||
"""Test authorization code cannot be used twice."""
|
||||
code, metadata, _ = setup_auth_code
|
||||
|
||||
# First exchange should succeed
|
||||
response1 = token_client.post("/token", data={
|
||||
"grant_type": "authorization_code",
|
||||
"code": code,
|
||||
"client_id": metadata["client_id"],
|
||||
"redirect_uri": metadata["redirect_uri"],
|
||||
})
|
||||
assert response1.status_code == 200
|
||||
|
||||
# Second exchange should fail
|
||||
response2 = token_client.post("/token", data={
|
||||
"grant_type": "authorization_code",
|
||||
"code": code,
|
||||
"client_id": metadata["client_id"],
|
||||
"redirect_uri": metadata["redirect_uri"],
|
||||
})
|
||||
assert response2.status_code == 400
|
||||
data = response2.json()
|
||||
assert data["detail"]["error"] == "invalid_grant"
|
||||
|
||||
|
||||
class TestTokenExchangeErrors:
|
||||
"""Integration tests for token exchange error conditions."""
|
||||
|
||||
def test_invalid_grant_type_rejected(self, token_client, setup_auth_code):
|
||||
"""Test invalid grant_type returns error."""
|
||||
code, metadata, _ = setup_auth_code
|
||||
|
||||
response = token_client.post("/token", data={
|
||||
"grant_type": "password", # Invalid grant type
|
||||
"code": code,
|
||||
"client_id": metadata["client_id"],
|
||||
"redirect_uri": metadata["redirect_uri"],
|
||||
})
|
||||
|
||||
assert response.status_code == 400
|
||||
data = response.json()
|
||||
assert data["detail"]["error"] == "unsupported_grant_type"
|
||||
|
||||
def test_invalid_code_rejected(self, token_client, setup_auth_code):
|
||||
"""Test invalid authorization code returns error."""
|
||||
_, metadata, _ = setup_auth_code
|
||||
|
||||
response = token_client.post("/token", data={
|
||||
"grant_type": "authorization_code",
|
||||
"code": "nonexistent_code_12345",
|
||||
"client_id": metadata["client_id"],
|
||||
"redirect_uri": metadata["redirect_uri"],
|
||||
})
|
||||
|
||||
assert response.status_code == 400
|
||||
data = response.json()
|
||||
assert data["detail"]["error"] == "invalid_grant"
|
||||
|
||||
def test_client_id_mismatch_rejected(self, token_client, setup_auth_code):
|
||||
"""Test mismatched client_id returns error."""
|
||||
code, metadata, _ = setup_auth_code
|
||||
|
||||
response = token_client.post("/token", data={
|
||||
"grant_type": "authorization_code",
|
||||
"code": code,
|
||||
"client_id": "https://different-client.example.com", # Wrong client
|
||||
"redirect_uri": metadata["redirect_uri"],
|
||||
})
|
||||
|
||||
assert response.status_code == 400
|
||||
data = response.json()
|
||||
assert data["detail"]["error"] == "invalid_client"
|
||||
|
||||
def test_redirect_uri_mismatch_rejected(self, token_client, setup_auth_code):
|
||||
"""Test mismatched redirect_uri returns error."""
|
||||
code, metadata, _ = setup_auth_code
|
||||
|
||||
response = token_client.post("/token", data={
|
||||
"grant_type": "authorization_code",
|
||||
"code": code,
|
||||
"client_id": metadata["client_id"],
|
||||
"redirect_uri": "https://app.example.com/different-callback", # Wrong URI
|
||||
})
|
||||
|
||||
assert response.status_code == 400
|
||||
data = response.json()
|
||||
assert data["detail"]["error"] == "invalid_grant"
|
||||
|
||||
def test_used_code_rejected(self, token_client, token_app, test_code_storage):
|
||||
"""Test already-used authorization code returns error."""
|
||||
from gondulf.dependencies import get_code_storage
|
||||
|
||||
code = "used_code_test_12345"
|
||||
metadata = {
|
||||
"client_id": "https://app.example.com",
|
||||
"redirect_uri": "https://app.example.com/callback",
|
||||
"state": "xyz123",
|
||||
"me": "https://user.example.com",
|
||||
"scope": "",
|
||||
"code_challenge": "abc123",
|
||||
"code_challenge_method": "S256",
|
||||
"created_at": 1234567890,
|
||||
"expires_at": 1234568490,
|
||||
"used": True # Already used
|
||||
}
|
||||
|
||||
token_app.dependency_overrides[get_code_storage] = lambda: test_code_storage
|
||||
test_code_storage.store(f"authz:{code}", metadata)
|
||||
|
||||
response = token_client.post("/token", data={
|
||||
"grant_type": "authorization_code",
|
||||
"code": code,
|
||||
"client_id": metadata["client_id"],
|
||||
"redirect_uri": metadata["redirect_uri"],
|
||||
})
|
||||
|
||||
assert response.status_code == 400
|
||||
data = response.json()
|
||||
assert data["detail"]["error"] == "invalid_grant"
|
||||
|
||||
token_app.dependency_overrides.clear()
|
||||
|
||||
|
||||
class TestTokenEndpointSecurity:
|
||||
"""Security tests for token endpoint."""
|
||||
|
||||
def test_token_endpoint_requires_post(self, token_client):
|
||||
"""Test token endpoint only accepts POST requests."""
|
||||
response = token_client.get("/token")
|
||||
assert response.status_code == 405 # Method Not Allowed
|
||||
|
||||
def test_token_endpoint_requires_form_data(self, token_client, setup_auth_code):
|
||||
"""Test token endpoint requires form-encoded data."""
|
||||
code, metadata, _ = setup_auth_code
|
||||
|
||||
# Send JSON instead of form data
|
||||
response = token_client.post("/token", json={
|
||||
"grant_type": "authorization_code",
|
||||
"code": code,
|
||||
"client_id": metadata["client_id"],
|
||||
"redirect_uri": metadata["redirect_uri"],
|
||||
})
|
||||
|
||||
# Should fail because it expects form data
|
||||
assert response.status_code == 422 # Unprocessable Entity
|
||||
|
||||
def test_token_response_security_headers(self, token_client, setup_auth_code):
|
||||
"""Test token response includes security headers."""
|
||||
code, metadata, _ = setup_auth_code
|
||||
|
||||
response = token_client.post("/token", data={
|
||||
"grant_type": "authorization_code",
|
||||
"code": code,
|
||||
"client_id": metadata["client_id"],
|
||||
"redirect_uri": metadata["redirect_uri"],
|
||||
})
|
||||
|
||||
# Security headers should be present
|
||||
assert "X-Frame-Options" in response.headers
|
||||
assert "X-Content-Type-Options" in response.headers
|
||||
|
||||
def test_error_response_format_matches_oauth2(self, token_client):
|
||||
"""Test error responses match OAuth 2.0 format."""
|
||||
response = token_client.post("/token", data={
|
||||
"grant_type": "authorization_code",
|
||||
"code": "invalid_code",
|
||||
"client_id": "https://app.example.com",
|
||||
"redirect_uri": "https://app.example.com/callback",
|
||||
})
|
||||
|
||||
assert response.status_code == 400
|
||||
data = response.json()
|
||||
|
||||
# OAuth 2.0 error format
|
||||
assert "detail" in data
|
||||
assert "error" in data["detail"]
|
||||
|
||||
|
||||
class TestPKCEHandling:
|
||||
"""Tests for PKCE code_verifier handling."""
|
||||
|
||||
def test_code_verifier_accepted(self, token_client, setup_auth_code):
|
||||
"""Test code_verifier parameter is accepted."""
|
||||
code, metadata, _ = setup_auth_code
|
||||
|
||||
response = token_client.post("/token", data={
|
||||
"grant_type": "authorization_code",
|
||||
"code": code,
|
||||
"client_id": metadata["client_id"],
|
||||
"redirect_uri": metadata["redirect_uri"],
|
||||
"code_verifier": "some_verifier_value", # PKCE verifier
|
||||
})
|
||||
|
||||
# Should succeed (PKCE validation deferred per design)
|
||||
assert response.status_code == 200
|
||||
|
||||
def test_token_exchange_works_without_verifier(self, token_client, setup_auth_code):
|
||||
"""Test token exchange works without code_verifier in v1.0.0."""
|
||||
code, metadata, _ = setup_auth_code
|
||||
|
||||
response = token_client.post("/token", data={
|
||||
"grant_type": "authorization_code",
|
||||
"code": code,
|
||||
"client_id": metadata["client_id"],
|
||||
"redirect_uri": metadata["redirect_uri"],
|
||||
# No code_verifier
|
||||
})
|
||||
|
||||
# Should succeed (PKCE not enforced in v1.0.0)
|
||||
assert response.status_code == 200
|
||||
243
tests/integration/api/test_verification_flow.py
Normal file
243
tests/integration/api/test_verification_flow.py
Normal file
@@ -0,0 +1,243 @@
|
||||
"""
|
||||
Integration tests for domain verification flow.
|
||||
|
||||
Tests the complete domain verification flow including DNS verification,
|
||||
email discovery, and code verification.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
from unittest.mock import Mock
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def verification_app(monkeypatch, tmp_path):
|
||||
"""Create app for verification testing."""
|
||||
db_path = tmp_path / "test.db"
|
||||
|
||||
monkeypatch.setenv("GONDULF_SECRET_KEY", "a" * 32)
|
||||
monkeypatch.setenv("GONDULF_BASE_URL", "https://auth.example.com")
|
||||
monkeypatch.setenv("GONDULF_DATABASE_URL", f"sqlite:///{db_path}")
|
||||
monkeypatch.setenv("GONDULF_DEBUG", "true")
|
||||
|
||||
from gondulf.main import app
|
||||
return app
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def verification_client(verification_app):
|
||||
"""Create test client for verification tests."""
|
||||
with TestClient(verification_app) as client:
|
||||
yield client
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_verification_deps(verification_app, mock_dns_service, mock_email_service, mock_html_fetcher_with_email, mock_rate_limiter, test_code_storage):
|
||||
"""Setup mock dependencies for verification."""
|
||||
from gondulf.dependencies import get_verification_service, get_rate_limiter
|
||||
from gondulf.services.domain_verification import DomainVerificationService
|
||||
from gondulf.services.relme_parser import RelMeParser
|
||||
|
||||
service = DomainVerificationService(
|
||||
dns_service=mock_dns_service,
|
||||
email_service=mock_email_service,
|
||||
code_storage=test_code_storage,
|
||||
html_fetcher=mock_html_fetcher_with_email,
|
||||
relme_parser=RelMeParser()
|
||||
)
|
||||
|
||||
verification_app.dependency_overrides[get_verification_service] = lambda: service
|
||||
verification_app.dependency_overrides[get_rate_limiter] = lambda: mock_rate_limiter
|
||||
|
||||
yield service, test_code_storage
|
||||
|
||||
verification_app.dependency_overrides.clear()
|
||||
|
||||
|
||||
class TestStartVerification:
|
||||
"""Tests for starting domain verification."""
|
||||
|
||||
def test_start_verification_success(self, verification_client, mock_verification_deps):
|
||||
"""Test successful start of domain verification."""
|
||||
response = verification_client.post(
|
||||
"/api/verify/start",
|
||||
data={"me": "https://user.example.com"}
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["success"] is True
|
||||
assert "email" in data
|
||||
# Email should be masked
|
||||
assert "*" in data["email"]
|
||||
|
||||
def test_start_verification_invalid_me_url(self, verification_client, mock_verification_deps):
|
||||
"""Test verification fails with invalid me URL."""
|
||||
response = verification_client.post(
|
||||
"/api/verify/start",
|
||||
data={"me": "not-a-valid-url"}
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["success"] is False
|
||||
assert data["error"] == "invalid_me_url"
|
||||
|
||||
def test_start_verification_rate_limited(self, verification_app, verification_client, mock_rate_limiter_exceeded, verification_service):
|
||||
"""Test verification fails when rate limited."""
|
||||
from gondulf.dependencies import get_rate_limiter, get_verification_service
|
||||
|
||||
verification_app.dependency_overrides[get_rate_limiter] = lambda: mock_rate_limiter_exceeded
|
||||
verification_app.dependency_overrides[get_verification_service] = lambda: verification_service
|
||||
|
||||
response = verification_client.post(
|
||||
"/api/verify/start",
|
||||
data={"me": "https://user.example.com"}
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["success"] is False
|
||||
assert data["error"] == "rate_limit_exceeded"
|
||||
|
||||
verification_app.dependency_overrides.clear()
|
||||
|
||||
def test_start_verification_dns_failure(self, verification_app, verification_client, verification_service_dns_failure, mock_rate_limiter):
|
||||
"""Test verification fails when DNS check fails."""
|
||||
from gondulf.dependencies import get_rate_limiter, get_verification_service
|
||||
|
||||
verification_app.dependency_overrides[get_rate_limiter] = lambda: mock_rate_limiter
|
||||
verification_app.dependency_overrides[get_verification_service] = lambda: verification_service_dns_failure
|
||||
|
||||
response = verification_client.post(
|
||||
"/api/verify/start",
|
||||
data={"me": "https://user.example.com"}
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["success"] is False
|
||||
assert data["error"] == "dns_verification_failed"
|
||||
|
||||
verification_app.dependency_overrides.clear()
|
||||
|
||||
|
||||
class TestVerifyCode:
|
||||
"""Tests for verifying email code."""
|
||||
|
||||
def test_verify_code_success(self, verification_client, mock_verification_deps):
|
||||
"""Test successful code verification."""
|
||||
service, code_storage = mock_verification_deps
|
||||
|
||||
# First start verification to store the code
|
||||
verification_client.post(
|
||||
"/api/verify/start",
|
||||
data={"me": "https://example.com/"}
|
||||
)
|
||||
|
||||
# Get the stored code
|
||||
stored_code = code_storage.get("email_verify:example.com")
|
||||
assert stored_code is not None
|
||||
|
||||
# Verify the code
|
||||
response = verification_client.post(
|
||||
"/api/verify/code",
|
||||
data={"domain": "example.com", "code": stored_code}
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["success"] is True
|
||||
assert "email" in data
|
||||
|
||||
def test_verify_code_invalid_code(self, verification_client, mock_verification_deps):
|
||||
"""Test verification fails with invalid code."""
|
||||
response = verification_client.post(
|
||||
"/api/verify/code",
|
||||
data={"domain": "example.com", "code": "000000"}
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["success"] is False
|
||||
assert data["error"] == "invalid_code"
|
||||
|
||||
def test_verify_code_wrong_domain(self, verification_client, mock_verification_deps):
|
||||
"""Test verification fails with wrong domain."""
|
||||
service, code_storage = mock_verification_deps
|
||||
|
||||
# Start verification for one domain
|
||||
verification_client.post(
|
||||
"/api/verify/start",
|
||||
data={"me": "https://example.com/"}
|
||||
)
|
||||
|
||||
# Get the stored code
|
||||
stored_code = code_storage.get("email_verify:example.com")
|
||||
|
||||
# Try to verify with different domain
|
||||
response = verification_client.post(
|
||||
"/api/verify/code",
|
||||
data={"domain": "other.example.com", "code": stored_code}
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["success"] is False
|
||||
|
||||
|
||||
class TestVerificationSecurityHeaders:
|
||||
"""Security tests for verification endpoints."""
|
||||
|
||||
def test_start_verification_security_headers(self, verification_client, mock_verification_deps):
|
||||
"""Test verification endpoints include security headers."""
|
||||
response = verification_client.post(
|
||||
"/api/verify/start",
|
||||
data={"me": "https://user.example.com"}
|
||||
)
|
||||
|
||||
assert "X-Frame-Options" in response.headers
|
||||
assert "X-Content-Type-Options" in response.headers
|
||||
|
||||
def test_verify_code_security_headers(self, verification_client, mock_verification_deps):
|
||||
"""Test code verification endpoint includes security headers."""
|
||||
response = verification_client.post(
|
||||
"/api/verify/code",
|
||||
data={"domain": "example.com", "code": "123456"}
|
||||
)
|
||||
|
||||
assert "X-Frame-Options" in response.headers
|
||||
assert "X-Content-Type-Options" in response.headers
|
||||
|
||||
|
||||
class TestVerificationResponseFormat:
|
||||
"""Tests for verification endpoint response formats."""
|
||||
|
||||
def test_start_verification_returns_json(self, verification_client, mock_verification_deps):
|
||||
"""Test start verification returns JSON."""
|
||||
response = verification_client.post(
|
||||
"/api/verify/start",
|
||||
data={"me": "https://user.example.com"}
|
||||
)
|
||||
|
||||
assert "application/json" in response.headers["content-type"]
|
||||
|
||||
def test_verify_code_returns_json(self, verification_client, mock_verification_deps):
|
||||
"""Test code verification returns JSON."""
|
||||
response = verification_client.post(
|
||||
"/api/verify/code",
|
||||
data={"domain": "example.com", "code": "123456"}
|
||||
)
|
||||
|
||||
assert "application/json" in response.headers["content-type"]
|
||||
|
||||
def test_success_response_includes_method(self, verification_client, mock_verification_deps):
|
||||
"""Test successful verification includes verification method."""
|
||||
response = verification_client.post(
|
||||
"/api/verify/start",
|
||||
data={"me": "https://user.example.com"}
|
||||
)
|
||||
|
||||
data = response.json()
|
||||
assert data["success"] is True
|
||||
assert "verification_method" in data
|
||||
1
tests/integration/middleware/__init__.py
Normal file
1
tests/integration/middleware/__init__.py
Normal file
@@ -0,0 +1 @@
|
||||
"""Middleware integration tests for Gondulf IndieAuth server."""
|
||||
219
tests/integration/middleware/test_middleware_chain.py
Normal file
219
tests/integration/middleware/test_middleware_chain.py
Normal file
@@ -0,0 +1,219 @@
|
||||
"""
|
||||
Integration tests for middleware chain.
|
||||
|
||||
Tests that security headers and HTTPS enforcement middleware work together.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def middleware_app_debug(monkeypatch, tmp_path):
|
||||
"""Create app in debug mode for middleware testing."""
|
||||
db_path = tmp_path / "test.db"
|
||||
|
||||
monkeypatch.setenv("GONDULF_SECRET_KEY", "a" * 32)
|
||||
monkeypatch.setenv("GONDULF_BASE_URL", "https://auth.example.com")
|
||||
monkeypatch.setenv("GONDULF_DATABASE_URL", f"sqlite:///{db_path}")
|
||||
monkeypatch.setenv("GONDULF_DEBUG", "true")
|
||||
|
||||
from gondulf.main import app
|
||||
return app
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def middleware_app_production(monkeypatch, tmp_path):
|
||||
"""Create app in production mode for middleware testing."""
|
||||
db_path = tmp_path / "test.db"
|
||||
|
||||
monkeypatch.setenv("GONDULF_SECRET_KEY", "a" * 32)
|
||||
monkeypatch.setenv("GONDULF_BASE_URL", "https://auth.example.com")
|
||||
monkeypatch.setenv("GONDULF_DATABASE_URL", f"sqlite:///{db_path}")
|
||||
monkeypatch.setenv("GONDULF_DEBUG", "false")
|
||||
|
||||
from gondulf.main import app
|
||||
return app
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def debug_client(middleware_app_debug):
|
||||
"""Test client in debug mode."""
|
||||
with TestClient(middleware_app_debug) as client:
|
||||
yield client
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def production_client(middleware_app_production):
|
||||
"""Test client in production mode."""
|
||||
with TestClient(middleware_app_production) as client:
|
||||
yield client
|
||||
|
||||
|
||||
class TestSecurityHeadersChain:
|
||||
"""Tests for security headers middleware."""
|
||||
|
||||
def test_all_security_headers_present(self, debug_client):
|
||||
"""Test all required security headers are present."""
|
||||
response = debug_client.get("/")
|
||||
|
||||
# Required security headers
|
||||
assert response.headers["X-Frame-Options"] == "DENY"
|
||||
assert response.headers["X-Content-Type-Options"] == "nosniff"
|
||||
assert response.headers["X-XSS-Protection"] == "1; mode=block"
|
||||
assert "Content-Security-Policy" in response.headers
|
||||
assert "Referrer-Policy" in response.headers
|
||||
assert "Permissions-Policy" in response.headers
|
||||
|
||||
def test_csp_header_format(self, debug_client):
|
||||
"""Test CSP header has correct format."""
|
||||
response = debug_client.get("/")
|
||||
|
||||
csp = response.headers["Content-Security-Policy"]
|
||||
assert "default-src 'self'" in csp
|
||||
assert "frame-ancestors 'none'" in csp
|
||||
|
||||
def test_referrer_policy_value(self, debug_client):
|
||||
"""Test Referrer-Policy has correct value."""
|
||||
response = debug_client.get("/")
|
||||
|
||||
assert response.headers["Referrer-Policy"] == "strict-origin-when-cross-origin"
|
||||
|
||||
def test_permissions_policy_value(self, debug_client):
|
||||
"""Test Permissions-Policy disables unnecessary features."""
|
||||
response = debug_client.get("/")
|
||||
|
||||
permissions = response.headers["Permissions-Policy"]
|
||||
assert "geolocation=()" in permissions
|
||||
assert "microphone=()" in permissions
|
||||
assert "camera=()" in permissions
|
||||
|
||||
def test_hsts_not_in_debug_mode(self, debug_client):
|
||||
"""Test HSTS header is not present in debug mode."""
|
||||
response = debug_client.get("/")
|
||||
|
||||
# HSTS should not be set in debug mode
|
||||
assert "Strict-Transport-Security" not in response.headers
|
||||
|
||||
|
||||
class TestMiddlewareOnAllEndpoints:
|
||||
"""Tests that middleware applies to all endpoints."""
|
||||
|
||||
@pytest.mark.parametrize("endpoint", [
|
||||
"/",
|
||||
"/health",
|
||||
"/.well-known/oauth-authorization-server",
|
||||
])
|
||||
def test_security_headers_on_endpoint(self, debug_client, endpoint):
|
||||
"""Test security headers present on various endpoints."""
|
||||
response = debug_client.get(endpoint)
|
||||
|
||||
assert "X-Frame-Options" in response.headers
|
||||
assert "X-Content-Type-Options" in response.headers
|
||||
|
||||
def test_security_headers_on_post_endpoint(self, debug_client):
|
||||
"""Test security headers on POST endpoints."""
|
||||
response = debug_client.post(
|
||||
"/api/verify/start",
|
||||
data={"me": "https://example.com"}
|
||||
)
|
||||
|
||||
assert "X-Frame-Options" in response.headers
|
||||
assert "X-Content-Type-Options" in response.headers
|
||||
|
||||
def test_security_headers_on_error_response(self, debug_client):
|
||||
"""Test security headers on 4xx error responses."""
|
||||
response = debug_client.get("/authorize") # Missing required params
|
||||
|
||||
assert response.status_code == 400
|
||||
assert "X-Frame-Options" in response.headers
|
||||
assert "X-Content-Type-Options" in response.headers
|
||||
|
||||
|
||||
class TestHTTPSEnforcementMiddleware:
|
||||
"""Tests for HTTPS enforcement middleware."""
|
||||
|
||||
def test_http_localhost_allowed_in_debug(self, debug_client):
|
||||
"""Test HTTP to localhost is allowed in debug mode."""
|
||||
# TestClient defaults to http
|
||||
response = debug_client.get("http://localhost/")
|
||||
|
||||
# Should work in debug mode
|
||||
assert response.status_code == 200
|
||||
|
||||
def test_https_always_allowed(self, debug_client):
|
||||
"""Test HTTPS requests are always allowed."""
|
||||
response = debug_client.get("/")
|
||||
|
||||
assert response.status_code == 200
|
||||
|
||||
|
||||
class TestMiddlewareOrdering:
|
||||
"""Tests for correct middleware ordering."""
|
||||
|
||||
def test_security_headers_applied_to_redirects(self, debug_client):
|
||||
"""Test security headers are applied even on redirect responses."""
|
||||
# This request should trigger a redirect due to error
|
||||
response = debug_client.get(
|
||||
"/authorize",
|
||||
params={
|
||||
"client_id": "https://app.example.com",
|
||||
"redirect_uri": "https://app.example.com/callback",
|
||||
"response_type": "token", # Invalid - should redirect with error
|
||||
"state": "test"
|
||||
},
|
||||
follow_redirects=False
|
||||
)
|
||||
|
||||
# Even on redirect, security headers should be present
|
||||
if response.status_code in (301, 302, 307, 308):
|
||||
assert "X-Frame-Options" in response.headers
|
||||
|
||||
def test_middleware_chain_complete(self, debug_client):
|
||||
"""Test full middleware chain processes correctly."""
|
||||
response = debug_client.get("/")
|
||||
|
||||
# Response should be successful
|
||||
assert response.status_code == 200
|
||||
|
||||
# Security headers from SecurityHeadersMiddleware
|
||||
assert "X-Frame-Options" in response.headers
|
||||
assert "X-Content-Type-Options" in response.headers
|
||||
|
||||
# Application response should be JSON
|
||||
data = response.json()
|
||||
assert "service" in data
|
||||
|
||||
|
||||
class TestContentSecurityPolicy:
|
||||
"""Tests for CSP header configuration."""
|
||||
|
||||
def test_csp_allows_self(self, debug_client):
|
||||
"""Test CSP allows resources from same origin."""
|
||||
response = debug_client.get("/")
|
||||
|
||||
csp = response.headers["Content-Security-Policy"]
|
||||
assert "default-src 'self'" in csp
|
||||
|
||||
def test_csp_allows_inline_styles(self, debug_client):
|
||||
"""Test CSP allows inline styles for templates."""
|
||||
response = debug_client.get("/")
|
||||
|
||||
csp = response.headers["Content-Security-Policy"]
|
||||
assert "style-src" in csp
|
||||
assert "'unsafe-inline'" in csp
|
||||
|
||||
def test_csp_allows_https_images(self, debug_client):
|
||||
"""Test CSP allows HTTPS images for h-app logos."""
|
||||
response = debug_client.get("/")
|
||||
|
||||
csp = response.headers["Content-Security-Policy"]
|
||||
assert "img-src" in csp
|
||||
assert "https:" in csp
|
||||
|
||||
def test_csp_prevents_framing(self, debug_client):
|
||||
"""Test CSP prevents page from being framed."""
|
||||
response = debug_client.get("/")
|
||||
|
||||
csp = response.headers["Content-Security-Policy"]
|
||||
assert "frame-ancestors 'none'" in csp
|
||||
1
tests/integration/services/__init__.py
Normal file
1
tests/integration/services/__init__.py
Normal file
@@ -0,0 +1 @@
|
||||
"""Service integration tests for Gondulf IndieAuth server."""
|
||||
190
tests/integration/services/test_domain_verification.py
Normal file
190
tests/integration/services/test_domain_verification.py
Normal file
@@ -0,0 +1,190 @@
|
||||
"""
|
||||
Integration tests for domain verification service.
|
||||
|
||||
Tests the complete domain verification flow with mocked external services.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
from unittest.mock import Mock
|
||||
|
||||
|
||||
class TestDomainVerificationIntegration:
|
||||
"""Integration tests for DomainVerificationService."""
|
||||
|
||||
def test_complete_verification_flow(self, verification_service, mock_email_service):
|
||||
"""Test complete DNS + email verification flow."""
|
||||
# Start verification
|
||||
result = verification_service.start_verification(
|
||||
domain="example.com",
|
||||
me_url="https://example.com/"
|
||||
)
|
||||
|
||||
assert result["success"] is True
|
||||
assert "email" in result
|
||||
assert result["verification_method"] == "email"
|
||||
|
||||
# Email should have been sent
|
||||
assert len(mock_email_service.messages_sent) == 1
|
||||
sent = mock_email_service.messages_sent[0]
|
||||
assert sent["email"] == "test@example.com"
|
||||
assert sent["domain"] == "example.com"
|
||||
assert len(sent["code"]) == 6
|
||||
|
||||
def test_dns_failure_blocks_verification(self, verification_service_dns_failure):
|
||||
"""Test that DNS verification failure stops the process."""
|
||||
result = verification_service_dns_failure.start_verification(
|
||||
domain="example.com",
|
||||
me_url="https://example.com/"
|
||||
)
|
||||
|
||||
assert result["success"] is False
|
||||
assert result["error"] == "dns_verification_failed"
|
||||
|
||||
def test_email_discovery_failure(self, mock_dns_service, mock_email_service, mock_html_fetcher, test_code_storage):
|
||||
"""Test verification fails when no email is discovered."""
|
||||
from gondulf.services.domain_verification import DomainVerificationService
|
||||
from gondulf.services.relme_parser import RelMeParser
|
||||
|
||||
# HTML fetcher returns page without email
|
||||
mock_html_fetcher.fetch = Mock(return_value="<html><body>No email here</body></html>")
|
||||
|
||||
service = DomainVerificationService(
|
||||
dns_service=mock_dns_service,
|
||||
email_service=mock_email_service,
|
||||
code_storage=test_code_storage,
|
||||
html_fetcher=mock_html_fetcher,
|
||||
relme_parser=RelMeParser()
|
||||
)
|
||||
|
||||
result = service.start_verification(
|
||||
domain="example.com",
|
||||
me_url="https://example.com/"
|
||||
)
|
||||
|
||||
assert result["success"] is False
|
||||
assert result["error"] == "email_discovery_failed"
|
||||
|
||||
def test_code_verification_success(self, verification_service, test_code_storage):
|
||||
"""Test successful code verification."""
|
||||
# Start verification to generate code
|
||||
verification_service.start_verification(
|
||||
domain="example.com",
|
||||
me_url="https://example.com/"
|
||||
)
|
||||
|
||||
# Get the stored code
|
||||
stored_code = test_code_storage.get("email_verify:example.com")
|
||||
assert stored_code is not None
|
||||
|
||||
# Verify the code
|
||||
result = verification_service.verify_email_code(
|
||||
domain="example.com",
|
||||
code=stored_code
|
||||
)
|
||||
|
||||
assert result["success"] is True
|
||||
assert result["email"] == "test@example.com"
|
||||
|
||||
def test_code_verification_invalid_code(self, verification_service, test_code_storage):
|
||||
"""Test code verification fails with wrong code."""
|
||||
# Start verification
|
||||
verification_service.start_verification(
|
||||
domain="example.com",
|
||||
me_url="https://example.com/"
|
||||
)
|
||||
|
||||
# Try to verify with wrong code
|
||||
result = verification_service.verify_email_code(
|
||||
domain="example.com",
|
||||
code="000000"
|
||||
)
|
||||
|
||||
assert result["success"] is False
|
||||
assert result["error"] == "invalid_code"
|
||||
|
||||
def test_code_single_use(self, verification_service, test_code_storage):
|
||||
"""Test verification code can only be used once."""
|
||||
# Start verification
|
||||
verification_service.start_verification(
|
||||
domain="example.com",
|
||||
me_url="https://example.com/"
|
||||
)
|
||||
|
||||
# Get the stored code
|
||||
stored_code = test_code_storage.get("email_verify:example.com")
|
||||
|
||||
# First verification should succeed
|
||||
result1 = verification_service.verify_email_code(
|
||||
domain="example.com",
|
||||
code=stored_code
|
||||
)
|
||||
assert result1["success"] is True
|
||||
|
||||
# Second verification should fail
|
||||
result2 = verification_service.verify_email_code(
|
||||
domain="example.com",
|
||||
code=stored_code
|
||||
)
|
||||
assert result2["success"] is False
|
||||
|
||||
|
||||
class TestAuthorizationCodeGeneration:
|
||||
"""Integration tests for authorization code generation."""
|
||||
|
||||
def test_create_authorization_code(self, verification_service):
|
||||
"""Test authorization code creation stores metadata."""
|
||||
code = verification_service.create_authorization_code(
|
||||
client_id="https://app.example.com",
|
||||
redirect_uri="https://app.example.com/callback",
|
||||
state="test123",
|
||||
code_challenge="abc123",
|
||||
code_challenge_method="S256",
|
||||
scope="",
|
||||
me="https://user.example.com"
|
||||
)
|
||||
|
||||
assert code is not None
|
||||
assert len(code) > 20 # Should be a substantial code
|
||||
|
||||
def test_authorization_code_unique(self, verification_service):
|
||||
"""Test each authorization code is unique."""
|
||||
codes = set()
|
||||
for _ in range(100):
|
||||
code = verification_service.create_authorization_code(
|
||||
client_id="https://app.example.com",
|
||||
redirect_uri="https://app.example.com/callback",
|
||||
state="test123",
|
||||
code_challenge="abc123",
|
||||
code_challenge_method="S256",
|
||||
scope="",
|
||||
me="https://user.example.com"
|
||||
)
|
||||
codes.add(code)
|
||||
|
||||
# All 100 codes should be unique
|
||||
assert len(codes) == 100
|
||||
|
||||
def test_authorization_code_stored_with_metadata(self, verification_service, test_code_storage):
|
||||
"""Test authorization code metadata is stored correctly."""
|
||||
code = verification_service.create_authorization_code(
|
||||
client_id="https://app.example.com",
|
||||
redirect_uri="https://app.example.com/callback",
|
||||
state="test123",
|
||||
code_challenge="abc123",
|
||||
code_challenge_method="S256",
|
||||
scope="profile",
|
||||
me="https://user.example.com"
|
||||
)
|
||||
|
||||
# Retrieve stored metadata
|
||||
metadata = test_code_storage.get(f"authz:{code}")
|
||||
|
||||
assert metadata is not None
|
||||
assert metadata["client_id"] == "https://app.example.com"
|
||||
assert metadata["redirect_uri"] == "https://app.example.com/callback"
|
||||
assert metadata["state"] == "test123"
|
||||
assert metadata["code_challenge"] == "abc123"
|
||||
assert metadata["code_challenge_method"] == "S256"
|
||||
assert metadata["scope"] == "profile"
|
||||
assert metadata["me"] == "https://user.example.com"
|
||||
assert metadata["used"] is False
|
||||
170
tests/integration/services/test_happ_parser.py
Normal file
170
tests/integration/services/test_happ_parser.py
Normal file
@@ -0,0 +1,170 @@
|
||||
"""
|
||||
Integration tests for h-app parser service.
|
||||
|
||||
Tests client metadata fetching with mocked HTTP responses.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
from unittest.mock import MagicMock, Mock, patch
|
||||
|
||||
|
||||
class TestHAppParserIntegration:
|
||||
"""Integration tests for h-app metadata parsing."""
|
||||
|
||||
@pytest.fixture
|
||||
def happ_parser_with_mock_fetcher(self):
|
||||
"""Create h-app parser with mocked HTML fetcher."""
|
||||
from gondulf.services.happ_parser import HAppParser
|
||||
|
||||
html = '''
|
||||
<!DOCTYPE html>
|
||||
<html>
|
||||
<head><title>Test App</title></head>
|
||||
<body>
|
||||
<div class="h-app">
|
||||
<h1 class="p-name">Example Application</h1>
|
||||
<img class="u-logo" src="https://app.example.com/logo.png" alt="Logo">
|
||||
<a class="u-url" href="https://app.example.com">Home</a>
|
||||
</div>
|
||||
</body>
|
||||
</html>
|
||||
'''
|
||||
|
||||
mock_fetcher = Mock()
|
||||
mock_fetcher.fetch = Mock(return_value=html)
|
||||
|
||||
return HAppParser(html_fetcher=mock_fetcher)
|
||||
|
||||
def test_fetch_and_parse_happ_metadata(self, happ_parser_with_mock_fetcher):
|
||||
"""Test fetching and parsing h-app microformat."""
|
||||
import asyncio
|
||||
result = asyncio.get_event_loop().run_until_complete(
|
||||
happ_parser_with_mock_fetcher.fetch_and_parse("https://app.example.com")
|
||||
)
|
||||
|
||||
assert result is not None
|
||||
assert result.name == "Example Application"
|
||||
assert result.logo == "https://app.example.com/logo.png"
|
||||
|
||||
def test_parse_page_without_happ(self, mock_urlopen):
|
||||
"""Test parsing page without h-app returns fallback."""
|
||||
from gondulf.services.happ_parser import HAppParser
|
||||
from gondulf.services.html_fetcher import HTMLFetcherService
|
||||
|
||||
# Setup mock to return page without h-app
|
||||
html = b'<html><head><title>Plain Page</title></head><body>No h-app</body></html>'
|
||||
mock_response = MagicMock()
|
||||
mock_response.read.return_value = html
|
||||
mock_response.status = 200
|
||||
mock_response.__enter__ = Mock(return_value=mock_response)
|
||||
mock_response.__exit__ = Mock(return_value=False)
|
||||
mock_urlopen.return_value = mock_response
|
||||
|
||||
fetcher = HTMLFetcherService()
|
||||
parser = HAppParser(html_fetcher=fetcher)
|
||||
|
||||
import asyncio
|
||||
result = asyncio.get_event_loop().run_until_complete(
|
||||
parser.fetch_and_parse("https://app.example.com")
|
||||
)
|
||||
|
||||
# Should return fallback metadata using domain
|
||||
assert result is not None
|
||||
assert "example.com" in result.name.lower() or result.name == "Plain Page"
|
||||
|
||||
def test_fetch_timeout_returns_fallback(self, mock_urlopen_timeout):
|
||||
"""Test HTTP timeout returns fallback metadata."""
|
||||
from gondulf.services.happ_parser import HAppParser
|
||||
from gondulf.services.html_fetcher import HTMLFetcherService
|
||||
|
||||
fetcher = HTMLFetcherService()
|
||||
parser = HAppParser(html_fetcher=fetcher)
|
||||
|
||||
import asyncio
|
||||
result = asyncio.get_event_loop().run_until_complete(
|
||||
parser.fetch_and_parse("https://slow-app.example.com")
|
||||
)
|
||||
|
||||
# Should return fallback metadata
|
||||
assert result is not None
|
||||
# Should use domain as fallback name
|
||||
assert "slow-app.example.com" in result.name or result.url == "https://slow-app.example.com"
|
||||
|
||||
|
||||
class TestClientMetadataCaching:
|
||||
"""Tests for client metadata caching behavior."""
|
||||
|
||||
def test_metadata_fetched_from_url(self, mock_urlopen_with_happ):
|
||||
"""Test metadata is actually fetched from URL."""
|
||||
from gondulf.services.happ_parser import HAppParser
|
||||
from gondulf.services.html_fetcher import HTMLFetcherService
|
||||
|
||||
fetcher = HTMLFetcherService()
|
||||
parser = HAppParser(html_fetcher=fetcher)
|
||||
|
||||
import asyncio
|
||||
result = asyncio.get_event_loop().run_until_complete(
|
||||
parser.fetch_and_parse("https://app.example.com")
|
||||
)
|
||||
|
||||
# urlopen should have been called
|
||||
mock_urlopen_with_happ.assert_called()
|
||||
|
||||
|
||||
class TestHAppMicroformatVariants:
|
||||
"""Tests for various h-app microformat formats."""
|
||||
|
||||
@pytest.fixture
|
||||
def create_parser_with_html(self):
|
||||
"""Factory to create parser with specific HTML content."""
|
||||
def _create(html_content):
|
||||
from gondulf.services.happ_parser import HAppParser
|
||||
|
||||
mock_fetcher = Mock()
|
||||
mock_fetcher.fetch = Mock(return_value=html_content)
|
||||
|
||||
return HAppParser(html_fetcher=mock_fetcher)
|
||||
return _create
|
||||
|
||||
def test_parse_happ_with_minimal_data(self, create_parser_with_html):
|
||||
"""Test parsing h-app with only name."""
|
||||
html = '''
|
||||
<html>
|
||||
<body>
|
||||
<div class="h-app">
|
||||
<span class="p-name">Minimal App</span>
|
||||
</div>
|
||||
</body>
|
||||
</html>
|
||||
'''
|
||||
parser = create_parser_with_html(html)
|
||||
|
||||
import asyncio
|
||||
result = asyncio.get_event_loop().run_until_complete(
|
||||
parser.fetch_and_parse("https://minimal.example.com")
|
||||
)
|
||||
|
||||
assert result.name == "Minimal App"
|
||||
|
||||
def test_parse_happ_with_logo_relative_url(self, create_parser_with_html):
|
||||
"""Test parsing h-app with relative logo URL."""
|
||||
html = '''
|
||||
<html>
|
||||
<body>
|
||||
<div class="h-app">
|
||||
<span class="p-name">Relative Logo App</span>
|
||||
<img class="u-logo" src="/logo.png">
|
||||
</div>
|
||||
</body>
|
||||
</html>
|
||||
'''
|
||||
parser = create_parser_with_html(html)
|
||||
|
||||
import asyncio
|
||||
result = asyncio.get_event_loop().run_until_complete(
|
||||
parser.fetch_and_parse("https://relative.example.com")
|
||||
)
|
||||
|
||||
assert result.name == "Relative Logo App"
|
||||
# Logo should be resolved to absolute URL
|
||||
assert result.logo is not None
|
||||
Reference in New Issue
Block a user