""" 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. Updated for session-based authentication flow: - GET /authorize -> verify_code.html (email verification) - POST /authorize/verify-code -> consent page - POST /authorize/consent -> redirect with auth code """ import pytest from datetime import datetime, timedelta from fastapi.testclient import TestClient from unittest.mock import AsyncMock, Mock, patch from tests.conftest import extract_code_from_redirect def create_mock_dns_service(verify_success=True): """Create a mock DNS service.""" mock_service = Mock() mock_service.verify_txt_record.return_value = verify_success return mock_service def create_mock_email_service(): """Create a mock email service.""" mock_service = Mock() mock_service.send_verification_code = Mock() return mock_service def create_mock_html_fetcher(email="test@example.com"): """Create a mock HTML fetcher that returns a page with rel=me email.""" mock_fetcher = Mock() if email: html = f''' Email ''' else: html = '' mock_fetcher.fetch.return_value = html return mock_fetcher def create_mock_auth_session_service(session_id="test_session_123", code="123456", verified=True, response_type="code", me="https://user.example.com", state="test123", scope=""): """Create a mock auth session service.""" from gondulf.services.auth_session import AuthSessionService mock_service = Mock(spec=AuthSessionService) mock_service.create_session.return_value = { "session_id": session_id, "verification_code": code, "expires_at": datetime.utcnow() + timedelta(minutes=10) } session_data = { "session_id": session_id, "me": me, "email": "test@example.com", "code_verified": verified, "client_id": "https://app.example.com", "redirect_uri": "https://app.example.com/callback", "state": state, "code_challenge": "E9Melhoa2OwvFrEMTJguCHaoeK1t8URWbuGJSstw-cM", "code_challenge_method": "S256", "scope": scope, "response_type": response_type } mock_service.get_session.return_value = session_data mock_service.verify_code.return_value = session_data mock_service.is_session_verified.return_value = verified mock_service.delete_session = Mock() return mock_service def create_mock_happ_parser(): """Create a mock h-app parser.""" from gondulf.services.happ_parser import ClientMetadata mock_parser = Mock() mock_parser.fetch_and_parse = AsyncMock(return_value=ClientMetadata( name="E2E Test App", url="https://app.example.com", logo="https://app.example.com/logo.png" )) return mock_parser @pytest.fixture def e2e_app_with_mocks(monkeypatch, tmp_path): """Create app with all dependencies mocked 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 from gondulf.dependencies import ( get_dns_service, get_email_service, get_html_fetcher, get_relme_parser, get_happ_parser, get_auth_session_service, get_database ) from gondulf.database.connection import Database from gondulf.services.relme_parser import RelMeParser from sqlalchemy import text # Initialize database db = Database(f"sqlite:///{db_path}") db.initialize() # Add verified domain now = datetime.utcnow() with db.get_engine().begin() as conn: conn.execute( text(""" INSERT OR REPLACE INTO domains (domain, email, verification_code, verified, verified_at, last_checked, two_factor) VALUES (:domain, '', '', 1, :now, :now, 0) """), {"domain": "user.example.com", "now": now} ) app.dependency_overrides[get_database] = lambda: db app.dependency_overrides[get_dns_service] = lambda: create_mock_dns_service(True) app.dependency_overrides[get_email_service] = lambda: create_mock_email_service() app.dependency_overrides[get_html_fetcher] = lambda: create_mock_html_fetcher("test@example.com") app.dependency_overrides[get_relme_parser] = lambda: RelMeParser() app.dependency_overrides[get_happ_parser] = create_mock_happ_parser yield app, db app.dependency_overrides.clear() @pytest.fixture def e2e_app(monkeypatch, tmp_path): """Create app for E2E testing (without mocks, for error tests).""" 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.mark.e2e class TestCompleteAuthorizationFlow: """E2E tests for complete authorization code flow.""" def test_full_authorization_to_token_flow(self, e2e_app_with_mocks): """Test complete flow: authorization request -> verify code -> consent -> token exchange.""" app, db = e2e_app_with_mocks from gondulf.dependencies import get_auth_session_service # Create mock session service with verified session mock_session = create_mock_auth_session_service( verified=True, response_type="code", state="e2e_test_state_12345" ) app.dependency_overrides[get_auth_session_service] = lambda: mock_session with TestClient(app) as client: # Step 1: Authorization request - should show verification page 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 = client.get("/authorize", params=auth_params) # Should show verification page assert auth_response.status_code == 200 assert "text/html" in auth_response.headers["content-type"] assert "session_id" in auth_response.text.lower() or "verify" in auth_response.text.lower() # Step 2: Submit consent form (session is already verified in mock) consent_response = client.post( "/authorize/consent", data={"session_id": "test_session_123"}, 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 = 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_app_with_mocks): """Test that state parameter is preserved throughout the flow.""" app, db = e2e_app_with_mocks from gondulf.dependencies import get_auth_session_service state = "unique_state_for_csrf_protection" # Create mock session service with the specific state mock_session = create_mock_auth_session_service( verified=True, response_type="code", state=state ) app.dependency_overrides[get_auth_session_service] = lambda: mock_session with TestClient(app) as client: # Consent submission consent_response = client.post( "/authorize/consent", data={"session_id": "test_session_123"}, 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_app_with_mocks): """Test multiple authorization flows can run concurrently.""" app, db = e2e_app_with_mocks from gondulf.dependencies import get_auth_session_service flows = [] with TestClient(app) as client: # Start 3 authorization flows for i in range(3): # Create unique mock session for each flow mock_session = create_mock_auth_session_service( session_id=f"session_{i}", verified=True, response_type="code", state=f"flow_{i}", me=f"https://user{i}.example.com" ) app.dependency_overrides[get_auth_session_service] = lambda ms=mock_session: ms consent_response = client.post( "/authorize/consent", data={"session_id": f"session_{i}"}, 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 = 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): """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", "response_type": "code", # Authorization flow - exchange at token endpoint "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_app_with_mocks): """Test authorization code single-use enforcement.""" app, db = e2e_app_with_mocks from gondulf.dependencies import get_auth_session_service mock_session = create_mock_auth_session_service(verified=True, response_type="code") app.dependency_overrides[get_auth_session_service] = lambda: mock_session with TestClient(app) as client: # Get a valid code consent_response = client.post( "/authorize/consent", data={"session_id": "test_session_123"}, follow_redirects=False ) code = extract_code_from_redirect(consent_response.headers["location"]) # First exchange should succeed response1 = 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 = 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_app_with_mocks): """Test token exchange with wrong client_id is rejected.""" app, db = e2e_app_with_mocks from gondulf.dependencies import get_auth_session_service mock_session = create_mock_auth_session_service(verified=True, response_type="code") app.dependency_overrides[get_auth_session_service] = lambda: mock_session with TestClient(app) as client: # Get a code for one client consent_response = client.post( "/authorize/consent", data={"session_id": "test_session_123"}, follow_redirects=False ) code = extract_code_from_redirect(consent_response.headers["location"]) # Try to exchange with different client_id response = 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_app_with_mocks): """Test the token obtained through E2E flow has correct format.""" app, db = e2e_app_with_mocks from gondulf.dependencies import get_auth_session_service mock_session = create_mock_auth_session_service(verified=True, response_type="code") app.dependency_overrides[get_auth_session_service] = lambda: mock_session with TestClient(app) as client: # Complete the flow consent_response = client.post( "/authorize/consent", data={"session_id": "test_session_123"}, follow_redirects=False ) code = extract_code_from_redirect(consent_response.headers["location"]) token_response = 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_app_with_mocks): """Test token response includes all required IndieAuth fields.""" app, db = e2e_app_with_mocks from gondulf.dependencies import get_auth_session_service mock_session = create_mock_auth_session_service( verified=True, response_type="code", scope="profile" ) app.dependency_overrides[get_auth_session_service] = lambda: mock_session with TestClient(app) as client: # Complete the flow consent_response = client.post( "/authorize/consent", data={"session_id": "test_session_123"}, follow_redirects=False ) code = extract_code_from_redirect(consent_response.headers["location"]) token_response = 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