fix(auth): require email authentication every login
CRITICAL SECURITY FIX: - Email code required EVERY login (authentication, not verification) - DNS TXT check cached separately (domain verification) - New auth_sessions table for per-login state - Codes hashed with SHA-256, constant-time comparison - Max 3 attempts, 10-minute session expiry - OAuth params stored server-side (security improvement) New files: - services/auth_session.py - migrations 004, 005 - ADR-010: domain verification vs user authentication 312 tests passing, 86.21% coverage 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
@@ -3,9 +3,12 @@ Integration tests for authorization endpoint flow.
|
||||
|
||||
Tests the complete authorization endpoint behavior including parameter validation,
|
||||
client metadata fetching, consent form rendering, and code generation.
|
||||
|
||||
Updated for the session-based authentication flow (ADR-010).
|
||||
"""
|
||||
|
||||
import tempfile
|
||||
from datetime import datetime, timedelta
|
||||
from pathlib import Path
|
||||
from unittest.mock import AsyncMock, Mock, patch
|
||||
|
||||
@@ -184,7 +187,7 @@ class TestAuthorizationEndpointRedirectErrors:
|
||||
|
||||
|
||||
class TestAuthorizationConsentPage:
|
||||
"""Tests for the consent page rendering."""
|
||||
"""Tests for the consent page rendering (after email verification)."""
|
||||
|
||||
@pytest.fixture
|
||||
def complete_params(self):
|
||||
@@ -199,131 +202,279 @@ class TestAuthorizationConsentPage:
|
||||
"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)
|
||||
def test_valid_request_shows_verification_page(self, auth_app, complete_params, mock_happ_fetch):
|
||||
"""Test valid authorization request shows verification page (not consent directly)."""
|
||||
from gondulf.dependencies import (
|
||||
get_dns_service, get_email_service, get_html_fetcher,
|
||||
get_relme_parser, get_auth_session_service, get_database
|
||||
)
|
||||
from gondulf.database.connection import Database
|
||||
from sqlalchemy import text
|
||||
import tempfile
|
||||
|
||||
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
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
db_path = Path(tmpdir) / "test.db"
|
||||
db = Database(f"sqlite:///{db_path}")
|
||||
db.initialize()
|
||||
|
||||
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)
|
||||
# Setup DNS-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}
|
||||
)
|
||||
|
||||
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
|
||||
# Create mock services
|
||||
mock_dns = Mock()
|
||||
mock_dns.verify_txt_record.return_value = True
|
||||
|
||||
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)
|
||||
mock_email = Mock()
|
||||
mock_email.send_verification_code = Mock()
|
||||
|
||||
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
|
||||
mock_html = Mock()
|
||||
mock_html.fetch.return_value = '<html><a href="mailto:test@example.com" rel="me">Email</a></html>'
|
||||
|
||||
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)
|
||||
from gondulf.services.relme_parser import RelMeParser
|
||||
mock_relme = RelMeParser()
|
||||
|
||||
assert response.status_code == 200
|
||||
assert "test123" in response.text
|
||||
mock_session = Mock()
|
||||
mock_session.create_session.return_value = {
|
||||
"session_id": "test_session_123",
|
||||
"verification_code": "123456",
|
||||
"expires_at": datetime.utcnow() + timedelta(minutes=10)
|
||||
}
|
||||
|
||||
auth_app.dependency_overrides[get_database] = lambda: db
|
||||
auth_app.dependency_overrides[get_dns_service] = lambda: mock_dns
|
||||
auth_app.dependency_overrides[get_email_service] = lambda: mock_email
|
||||
auth_app.dependency_overrides[get_html_fetcher] = lambda: mock_html
|
||||
auth_app.dependency_overrides[get_relme_parser] = lambda: mock_relme
|
||||
auth_app.dependency_overrides[get_auth_session_service] = lambda: mock_session
|
||||
|
||||
try:
|
||||
with TestClient(auth_app) as client:
|
||||
response = client.get("/authorize", params=complete_params)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert "text/html" in response.headers["content-type"]
|
||||
# Should show verification page (email auth required every login)
|
||||
assert "Verify Your Identity" in response.text
|
||||
finally:
|
||||
auth_app.dependency_overrides.clear()
|
||||
|
||||
|
||||
class TestAuthorizationConsentSubmission:
|
||||
"""Tests for consent form submission."""
|
||||
"""Tests for consent form submission (via session-based flow)."""
|
||||
|
||||
@pytest.fixture
|
||||
def consent_form_data(self):
|
||||
"""Valid consent form data."""
|
||||
return {
|
||||
def test_consent_submission_redirects_with_code(self, auth_app):
|
||||
"""Test consent submission redirects to client with authorization code."""
|
||||
from gondulf.dependencies import get_auth_session_service, get_code_storage
|
||||
|
||||
# Mock verified session
|
||||
mock_session = Mock()
|
||||
mock_session.get_session.return_value = {
|
||||
"session_id": "test_session_123",
|
||||
"me": "https://user.example.com",
|
||||
"email": "test@example.com",
|
||||
"code_verified": True, # Session is verified
|
||||
"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": "",
|
||||
"response_type": "code"
|
||||
}
|
||||
mock_session.delete_session = Mock()
|
||||
|
||||
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
|
||||
)
|
||||
auth_app.dependency_overrides[get_auth_session_service] = lambda: mock_session
|
||||
|
||||
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
|
||||
try:
|
||||
with TestClient(auth_app) as client:
|
||||
response = client.post(
|
||||
"/authorize/consent",
|
||||
data={"session_id": "test_session_123"},
|
||||
follow_redirects=False
|
||||
)
|
||||
|
||||
def test_consent_submission_generates_unique_codes(self, auth_client, consent_form_data):
|
||||
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
|
||||
finally:
|
||||
auth_app.dependency_overrides.clear()
|
||||
|
||||
def test_consent_submission_generates_unique_codes(self, auth_app):
|
||||
"""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"]
|
||||
from gondulf.dependencies import get_auth_session_service
|
||||
|
||||
# Second submission
|
||||
response2 = auth_client.post(
|
||||
"/authorize/consent",
|
||||
data=consent_form_data,
|
||||
follow_redirects=False
|
||||
)
|
||||
location2 = response2.headers["location"]
|
||||
# Mock verified session
|
||||
mock_session = Mock()
|
||||
mock_session.get_session.return_value = {
|
||||
"session_id": "test_session_123",
|
||||
"me": "https://user.example.com",
|
||||
"email": "test@example.com",
|
||||
"code_verified": True,
|
||||
"client_id": "https://app.example.com",
|
||||
"redirect_uri": "https://app.example.com/callback",
|
||||
"state": "test123",
|
||||
"code_challenge": "E9Melhoa2OwvFrEMTJguCHaoeK1t8URWbuGJSstw-cM",
|
||||
"code_challenge_method": "S256",
|
||||
"scope": "",
|
||||
"response_type": "code"
|
||||
}
|
||||
mock_session.delete_session = Mock()
|
||||
|
||||
# Extract codes
|
||||
from tests.conftest import extract_code_from_redirect
|
||||
code1 = extract_code_from_redirect(location1)
|
||||
code2 = extract_code_from_redirect(location2)
|
||||
auth_app.dependency_overrides[get_auth_session_service] = lambda: mock_session
|
||||
|
||||
assert code1 != code2
|
||||
try:
|
||||
with TestClient(auth_app) as client:
|
||||
# First submission
|
||||
response1 = client.post(
|
||||
"/authorize/consent",
|
||||
data={"session_id": "test_session_123"},
|
||||
follow_redirects=False
|
||||
)
|
||||
location1 = response1.headers["location"]
|
||||
|
||||
def test_authorization_code_stored_for_exchange(self, auth_client, consent_form_data):
|
||||
# Second submission
|
||||
response2 = client.post(
|
||||
"/authorize/consent",
|
||||
data={"session_id": "test_session_123"},
|
||||
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
|
||||
finally:
|
||||
auth_app.dependency_overrides.clear()
|
||||
|
||||
def test_authorization_code_stored_for_exchange(self, auth_app):
|
||||
"""Test authorization code is stored for later token exchange."""
|
||||
response = auth_client.post(
|
||||
"/authorize/consent",
|
||||
data=consent_form_data,
|
||||
follow_redirects=False
|
||||
)
|
||||
from gondulf.dependencies import get_auth_session_service
|
||||
|
||||
from tests.conftest import extract_code_from_redirect
|
||||
code = extract_code_from_redirect(response.headers["location"])
|
||||
# Mock verified session
|
||||
mock_session = Mock()
|
||||
mock_session.get_session.return_value = {
|
||||
"session_id": "test_session_123",
|
||||
"me": "https://user.example.com",
|
||||
"email": "test@example.com",
|
||||
"code_verified": True,
|
||||
"client_id": "https://app.example.com",
|
||||
"redirect_uri": "https://app.example.com/callback",
|
||||
"state": "test123",
|
||||
"code_challenge": "E9Melhoa2OwvFrEMTJguCHaoeK1t8URWbuGJSstw-cM",
|
||||
"code_challenge_method": "S256",
|
||||
"scope": "",
|
||||
"response_type": "code"
|
||||
}
|
||||
mock_session.delete_session = Mock()
|
||||
|
||||
# Code should be non-empty and URL-safe
|
||||
assert code is not None
|
||||
assert len(code) > 20 # Should be a substantial code
|
||||
auth_app.dependency_overrides[get_auth_session_service] = lambda: mock_session
|
||||
|
||||
try:
|
||||
with TestClient(auth_app) as client:
|
||||
response = client.post(
|
||||
"/authorize/consent",
|
||||
data={"session_id": "test_session_123"},
|
||||
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
|
||||
finally:
|
||||
auth_app.dependency_overrides.clear()
|
||||
|
||||
|
||||
class TestAuthorizationSecurityHeaders:
|
||||
"""Tests for security headers on authorization endpoints."""
|
||||
|
||||
def test_authorization_page_has_security_headers(self, auth_client, mock_happ_fetch):
|
||||
def test_authorization_page_has_security_headers(self, auth_app, 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)
|
||||
from gondulf.dependencies import (
|
||||
get_dns_service, get_email_service, get_html_fetcher,
|
||||
get_relme_parser, get_auth_session_service, get_database
|
||||
)
|
||||
from gondulf.database.connection import Database
|
||||
from sqlalchemy import text
|
||||
import tempfile
|
||||
|
||||
assert "X-Frame-Options" in response.headers
|
||||
assert "X-Content-Type-Options" in response.headers
|
||||
assert response.headers["X-Frame-Options"] == "DENY"
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
db_path = Path(tmpdir) / "test.db"
|
||||
db = Database(f"sqlite:///{db_path}")
|
||||
db.initialize()
|
||||
|
||||
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}
|
||||
)
|
||||
|
||||
mock_dns = Mock()
|
||||
mock_dns.verify_txt_record.return_value = True
|
||||
|
||||
mock_email = Mock()
|
||||
mock_email.send_verification_code = Mock()
|
||||
|
||||
mock_html = Mock()
|
||||
mock_html.fetch.return_value = '<html><a href="mailto:test@example.com" rel="me">Email</a></html>'
|
||||
|
||||
from gondulf.services.relme_parser import RelMeParser
|
||||
mock_relme = RelMeParser()
|
||||
|
||||
mock_session = Mock()
|
||||
mock_session.create_session.return_value = {
|
||||
"session_id": "test_session_123",
|
||||
"verification_code": "123456",
|
||||
"expires_at": datetime.utcnow() + timedelta(minutes=10)
|
||||
}
|
||||
|
||||
auth_app.dependency_overrides[get_database] = lambda: db
|
||||
auth_app.dependency_overrides[get_dns_service] = lambda: mock_dns
|
||||
auth_app.dependency_overrides[get_email_service] = lambda: mock_email
|
||||
auth_app.dependency_overrides[get_html_fetcher] = lambda: mock_html
|
||||
auth_app.dependency_overrides[get_relme_parser] = lambda: mock_relme
|
||||
auth_app.dependency_overrides[get_auth_session_service] = lambda: mock_session
|
||||
|
||||
try:
|
||||
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",
|
||||
}
|
||||
|
||||
with TestClient(auth_app) as client:
|
||||
response = 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"
|
||||
finally:
|
||||
auth_app.dependency_overrides.clear()
|
||||
|
||||
def test_error_pages_have_security_headers(self, auth_client):
|
||||
"""Test error pages include security headers."""
|
||||
|
||||
@@ -1,10 +1,13 @@
|
||||
"""
|
||||
Integration tests for authorization endpoint domain verification.
|
||||
|
||||
Tests the security fix that requires domain verification before showing the consent page.
|
||||
Tests the authentication flow that requires email verification on EVERY login.
|
||||
See ADR-010 for the architectural decision.
|
||||
|
||||
Key principle: Email code is AUTHENTICATION (every login), not domain verification.
|
||||
"""
|
||||
|
||||
from datetime import datetime
|
||||
from datetime import datetime, timedelta
|
||||
from unittest.mock import AsyncMock, Mock, patch
|
||||
|
||||
import pytest
|
||||
@@ -25,40 +28,43 @@ def valid_auth_params():
|
||||
}
|
||||
|
||||
|
||||
def create_mock_verification_service(start_success=True, verify_success=True, start_error="dns_verification_failed"):
|
||||
"""Create a mock verification service with configurable behavior."""
|
||||
def create_mock_dns_service(verify_success=True):
|
||||
"""Create a mock DNS service."""
|
||||
mock_service = Mock()
|
||||
|
||||
if start_success:
|
||||
mock_service.start_verification.return_value = {
|
||||
"success": True,
|
||||
"email": "t***@example.com",
|
||||
"verification_method": "email"
|
||||
}
|
||||
else:
|
||||
mock_service.start_verification.return_value = {
|
||||
"success": False,
|
||||
"error": start_error
|
||||
}
|
||||
|
||||
if verify_success:
|
||||
mock_service.verify_email_code.return_value = {
|
||||
"success": True,
|
||||
"email": "test@example.com"
|
||||
}
|
||||
else:
|
||||
mock_service.verify_email_code.return_value = {
|
||||
"success": False,
|
||||
"error": "invalid_code"
|
||||
}
|
||||
|
||||
mock_service.code_storage = Mock()
|
||||
mock_service.code_storage.get.return_value = "test@example.com"
|
||||
mock_service.create_authorization_code.return_value = "test_auth_code_12345"
|
||||
|
||||
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'''
|
||||
<html>
|
||||
<body>
|
||||
<a href="mailto:{email}" rel="me">Email</a>
|
||||
</body>
|
||||
</html>
|
||||
'''
|
||||
else:
|
||||
html = '<html><body></body></html>'
|
||||
mock_fetcher.fetch.return_value = html
|
||||
return mock_fetcher
|
||||
|
||||
|
||||
def create_mock_relme_parser():
|
||||
"""Create a real RelMeParser (it's simple enough)."""
|
||||
from gondulf.services.relme_parser import RelMeParser
|
||||
return RelMeParser()
|
||||
|
||||
|
||||
def create_mock_happ_parser():
|
||||
"""Create a mock h-app parser."""
|
||||
from gondulf.services.happ_parser import ClientMetadata
|
||||
@@ -72,6 +78,55 @@ def create_mock_happ_parser():
|
||||
return mock_parser
|
||||
|
||||
|
||||
def create_mock_auth_session_service(session_id="test_session_123", code="123456", verified=False):
|
||||
"""Create a mock auth session service."""
|
||||
from gondulf.services.auth_session import (
|
||||
AuthSessionService,
|
||||
CodeVerificationError,
|
||||
SessionNotFoundError,
|
||||
)
|
||||
|
||||
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)
|
||||
}
|
||||
|
||||
mock_service.get_session.return_value = {
|
||||
"session_id": session_id,
|
||||
"me": "https://user.example.com",
|
||||
"email": "test@example.com",
|
||||
"code_verified": verified,
|
||||
"client_id": "https://app.example.com",
|
||||
"redirect_uri": "https://app.example.com/callback",
|
||||
"state": "test123",
|
||||
"code_challenge": "E9Melhoa2OwvFrEMTJguCHaoeK1t8URWbuGJSstw-cM",
|
||||
"code_challenge_method": "S256",
|
||||
"scope": "",
|
||||
"response_type": "code"
|
||||
}
|
||||
|
||||
mock_service.verify_code.return_value = {
|
||||
"session_id": session_id,
|
||||
"me": "https://user.example.com",
|
||||
"email": "test@example.com",
|
||||
"code_verified": True,
|
||||
"client_id": "https://app.example.com",
|
||||
"redirect_uri": "https://app.example.com/callback",
|
||||
"state": "test123",
|
||||
"code_challenge": "E9Melhoa2OwvFrEMTJguCHaoeK1t8URWbuGJSstw-cM",
|
||||
"code_challenge_method": "S256",
|
||||
"scope": "",
|
||||
"response_type": "code"
|
||||
}
|
||||
|
||||
mock_service.is_session_verified.return_value = verified
|
||||
mock_service.delete_session = Mock()
|
||||
|
||||
return mock_service
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def configured_app(monkeypatch, tmp_path):
|
||||
"""Create a fully configured app with fresh database."""
|
||||
@@ -87,31 +142,50 @@ def configured_app(monkeypatch, tmp_path):
|
||||
|
||||
|
||||
class TestUnverifiedDomainTriggersVerification:
|
||||
"""Tests that unverified domains trigger the verification flow."""
|
||||
"""Tests that any login triggers authentication (email code)."""
|
||||
|
||||
def test_unverified_domain_shows_verification_form(
|
||||
self, configured_app, valid_auth_params
|
||||
):
|
||||
"""Test that an unverified domain shows the verification code form."""
|
||||
app, _ = configured_app
|
||||
from gondulf.dependencies import get_verification_service, get_happ_parser
|
||||
"""Test that DNS-verified domain STILL shows verification form (email auth required)."""
|
||||
app, db_path = configured_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 sqlalchemy import text
|
||||
|
||||
mock_service = create_mock_verification_service(start_success=True)
|
||||
mock_parser = create_mock_happ_parser()
|
||||
# Setup database with DNS-verified domain
|
||||
db = Database(f"sqlite:///{db_path}")
|
||||
db.initialize()
|
||||
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_verification_service] = lambda: mock_service
|
||||
app.dependency_overrides[get_happ_parser] = lambda: mock_parser
|
||||
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] = create_mock_relme_parser
|
||||
app.dependency_overrides[get_happ_parser] = lambda: create_mock_happ_parser()
|
||||
app.dependency_overrides[get_auth_session_service] = lambda: create_mock_auth_session_service()
|
||||
|
||||
try:
|
||||
with TestClient(app) as client:
|
||||
response = client.get("/authorize", params=valid_auth_params)
|
||||
|
||||
assert response.status_code == 200
|
||||
# Should show verification form, not consent form
|
||||
# CRITICAL: Even DNS-verified domains require email verification every login
|
||||
assert "Verify Your Identity" in response.text
|
||||
assert "verification code" in response.text.lower()
|
||||
# Should show masked email
|
||||
assert "t***@example.com" in response.text
|
||||
finally:
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
@@ -119,89 +193,76 @@ class TestUnverifiedDomainTriggersVerification:
|
||||
self, configured_app, valid_auth_params
|
||||
):
|
||||
"""Test that authorization parameters are preserved in verification form."""
|
||||
app, _ = configured_app
|
||||
from gondulf.dependencies import get_verification_service, get_happ_parser
|
||||
app, db_path = configured_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 sqlalchemy import text
|
||||
|
||||
mock_service = create_mock_verification_service(start_success=True)
|
||||
mock_parser = create_mock_happ_parser()
|
||||
db = Database(f"sqlite:///{db_path}")
|
||||
db.initialize()
|
||||
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_verification_service] = lambda: mock_service
|
||||
app.dependency_overrides[get_happ_parser] = lambda: mock_parser
|
||||
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] = create_mock_relme_parser
|
||||
app.dependency_overrides[get_happ_parser] = lambda: create_mock_happ_parser()
|
||||
app.dependency_overrides[get_auth_session_service] = lambda: create_mock_auth_session_service()
|
||||
|
||||
try:
|
||||
with TestClient(app) as client:
|
||||
response = client.get("/authorize", params=valid_auth_params)
|
||||
|
||||
assert response.status_code == 200
|
||||
# Check hidden fields contain auth params
|
||||
assert 'name="client_id"' in response.text
|
||||
assert 'name="redirect_uri"' in response.text
|
||||
assert 'name="state"' in response.text
|
||||
assert 'name="code_challenge"' in response.text
|
||||
finally:
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
def test_unverified_domain_does_not_show_consent(
|
||||
self, configured_app, valid_auth_params
|
||||
):
|
||||
"""Test that unverified domain does NOT show consent form directly."""
|
||||
app, _ = configured_app
|
||||
from gondulf.dependencies import get_verification_service, get_happ_parser
|
||||
|
||||
mock_service = create_mock_verification_service(start_success=True)
|
||||
mock_parser = create_mock_happ_parser()
|
||||
|
||||
app.dependency_overrides[get_verification_service] = lambda: mock_service
|
||||
app.dependency_overrides[get_happ_parser] = lambda: mock_parser
|
||||
|
||||
try:
|
||||
with TestClient(app) as client:
|
||||
response = client.get("/authorize", params=valid_auth_params)
|
||||
|
||||
assert response.status_code == 200
|
||||
# Should NOT show consent/authorization form
|
||||
assert "Authorization Request" not in response.text
|
||||
# New flow uses session_id instead of passing all params
|
||||
assert 'name="session_id"' in response.text
|
||||
finally:
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
|
||||
class TestVerifiedDomainShowsConsent:
|
||||
"""Tests that verified domains skip verification and show consent."""
|
||||
"""Tests that verified sessions (email code verified) show consent."""
|
||||
|
||||
def test_verified_domain_shows_consent_page(
|
||||
self, configured_app, valid_auth_params
|
||||
):
|
||||
"""Test that a verified domain shows consent page directly."""
|
||||
"""Test that after email verification, consent page is shown."""
|
||||
app, db_path = configured_app
|
||||
from gondulf.dependencies import get_happ_parser, get_database
|
||||
from gondulf.database.connection import Database
|
||||
from sqlalchemy import text
|
||||
from gondulf.dependencies import get_happ_parser, get_auth_session_service
|
||||
from gondulf.services.auth_session import CodeVerificationError
|
||||
|
||||
# Create database and insert verified domain
|
||||
db = Database(f"sqlite:///{db_path}")
|
||||
db.initialize()
|
||||
# Mock auth session that succeeds on verify
|
||||
mock_session = create_mock_auth_session_service(verified=True)
|
||||
|
||||
with db.get_engine().begin() as conn:
|
||||
conn.execute(
|
||||
text("""
|
||||
INSERT INTO domains (domain, email, verification_code, verified, verified_at, two_factor)
|
||||
VALUES (:domain, :email, '', 1, :verified_at, 1)
|
||||
"""),
|
||||
{"domain": "user.example.com", "email": "test@example.com", "verified_at": datetime.utcnow()}
|
||||
)
|
||||
|
||||
# Override database to use same instance
|
||||
app.dependency_overrides[get_database] = lambda: db
|
||||
mock_parser = create_mock_happ_parser()
|
||||
app.dependency_overrides[get_happ_parser] = lambda: mock_parser
|
||||
app.dependency_overrides[get_auth_session_service] = lambda: mock_session
|
||||
app.dependency_overrides[get_happ_parser] = lambda: create_mock_happ_parser()
|
||||
|
||||
try:
|
||||
with TestClient(app) as client:
|
||||
response = client.get("/authorize", params=valid_auth_params)
|
||||
# Simulate verifying the code
|
||||
form_data = {
|
||||
"session_id": "test_session_123",
|
||||
"code": "123456",
|
||||
}
|
||||
|
||||
response = client.post("/authorize/verify-code", data=form_data)
|
||||
|
||||
# Should show consent page
|
||||
assert response.status_code == 200
|
||||
assert "Authorization Request" in response.text
|
||||
assert "Authorization Request" in response.text or "Authorize" in response.text
|
||||
assert 'action="/authorize/consent"' in response.text
|
||||
finally:
|
||||
app.dependency_overrides.clear()
|
||||
@@ -215,27 +276,19 @@ class TestVerificationCodeValidation:
|
||||
):
|
||||
"""Test that valid verification code shows consent page."""
|
||||
app, _ = configured_app
|
||||
from gondulf.dependencies import get_verification_service, get_happ_parser
|
||||
from gondulf.dependencies import get_happ_parser, get_auth_session_service
|
||||
|
||||
mock_service = create_mock_verification_service(verify_success=True)
|
||||
mock_session = create_mock_auth_session_service()
|
||||
mock_parser = create_mock_happ_parser()
|
||||
|
||||
app.dependency_overrides[get_verification_service] = lambda: mock_service
|
||||
app.dependency_overrides[get_auth_session_service] = lambda: mock_session
|
||||
app.dependency_overrides[get_happ_parser] = lambda: mock_parser
|
||||
|
||||
try:
|
||||
with TestClient(app) as client:
|
||||
form_data = {
|
||||
"domain": "user.example.com",
|
||||
"session_id": "test_session_123",
|
||||
"code": "123456",
|
||||
"client_id": valid_auth_params["client_id"],
|
||||
"redirect_uri": valid_auth_params["redirect_uri"],
|
||||
"response_type": valid_auth_params["response_type"],
|
||||
"state": valid_auth_params["state"],
|
||||
"code_challenge": valid_auth_params["code_challenge"],
|
||||
"code_challenge_method": valid_auth_params["code_challenge_method"],
|
||||
"scope": "",
|
||||
"me": valid_auth_params["me"],
|
||||
}
|
||||
|
||||
response = client.post("/authorize/verify-code", data=form_data)
|
||||
@@ -251,27 +304,22 @@ class TestVerificationCodeValidation:
|
||||
):
|
||||
"""Test that invalid code shows error and allows retry."""
|
||||
app, _ = configured_app
|
||||
from gondulf.dependencies import get_verification_service, get_happ_parser
|
||||
from gondulf.dependencies import get_happ_parser, get_auth_session_service
|
||||
from gondulf.services.auth_session import CodeVerificationError
|
||||
|
||||
mock_session = create_mock_auth_session_service()
|
||||
mock_session.verify_code.side_effect = CodeVerificationError("Invalid code")
|
||||
|
||||
mock_service = create_mock_verification_service(verify_success=False)
|
||||
mock_parser = create_mock_happ_parser()
|
||||
|
||||
app.dependency_overrides[get_verification_service] = lambda: mock_service
|
||||
app.dependency_overrides[get_auth_session_service] = lambda: mock_session
|
||||
app.dependency_overrides[get_happ_parser] = lambda: mock_parser
|
||||
|
||||
try:
|
||||
with TestClient(app) as client:
|
||||
form_data = {
|
||||
"domain": "user.example.com",
|
||||
"session_id": "test_session_123",
|
||||
"code": "000000",
|
||||
"client_id": valid_auth_params["client_id"],
|
||||
"redirect_uri": valid_auth_params["redirect_uri"],
|
||||
"response_type": valid_auth_params["response_type"],
|
||||
"state": valid_auth_params["state"],
|
||||
"code_challenge": valid_auth_params["code_challenge"],
|
||||
"code_challenge_method": valid_auth_params["code_challenge_method"],
|
||||
"scope": "",
|
||||
"me": valid_auth_params["me"],
|
||||
}
|
||||
|
||||
response = client.post("/authorize/verify-code", data=form_data)
|
||||
@@ -285,45 +333,6 @@ class TestVerificationCodeValidation:
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
|
||||
class TestDNSFailureHandling:
|
||||
"""Tests for DNS verification failure scenarios."""
|
||||
|
||||
def test_dns_failure_shows_instructions(
|
||||
self, configured_app, valid_auth_params
|
||||
):
|
||||
"""Test that DNS verification failure shows helpful instructions."""
|
||||
app, db_path = configured_app
|
||||
from gondulf.dependencies import get_verification_service, get_happ_parser, get_database
|
||||
from gondulf.database.connection import Database
|
||||
from sqlalchemy import text
|
||||
|
||||
# Clear any pre-existing verified domain to ensure test isolation
|
||||
db = Database(f"sqlite:///{db_path}")
|
||||
db.initialize()
|
||||
with db.get_engine().begin() as conn:
|
||||
conn.execute(text("DELETE FROM domains WHERE domain = :domain"), {"domain": "user.example.com"})
|
||||
|
||||
app.dependency_overrides[get_database] = lambda: db
|
||||
|
||||
mock_service = create_mock_verification_service(start_success=False, start_error="dns_verification_failed")
|
||||
mock_parser = create_mock_happ_parser()
|
||||
|
||||
app.dependency_overrides[get_verification_service] = lambda: mock_service
|
||||
app.dependency_overrides[get_happ_parser] = lambda: mock_parser
|
||||
|
||||
try:
|
||||
with TestClient(app) as client:
|
||||
response = client.get("/authorize", params=valid_auth_params)
|
||||
|
||||
assert response.status_code == 200
|
||||
# Should show error page with DNS instructions
|
||||
assert "DNS" in response.text or "dns" in response.text.lower()
|
||||
assert "TXT" in response.text
|
||||
assert "_gondulf" in response.text
|
||||
finally:
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
|
||||
class TestEmailFailureHandling:
|
||||
"""Tests for email discovery failure scenarios."""
|
||||
|
||||
@@ -332,23 +341,34 @@ class TestEmailFailureHandling:
|
||||
):
|
||||
"""Test that email discovery failure shows helpful instructions."""
|
||||
app, db_path = configured_app
|
||||
from gondulf.dependencies import get_verification_service, get_happ_parser, get_database
|
||||
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 sqlalchemy import text
|
||||
|
||||
# Clear any pre-existing verified domain to ensure test isolation
|
||||
db = Database(f"sqlite:///{db_path}")
|
||||
db.initialize()
|
||||
now = datetime.utcnow()
|
||||
with db.get_engine().begin() as conn:
|
||||
conn.execute(text("DELETE FROM domains WHERE domain = :domain"), {"domain": "user.example.com"})
|
||||
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
|
||||
|
||||
mock_service = create_mock_verification_service(start_success=False, start_error="email_discovery_failed")
|
||||
mock_parser = create_mock_happ_parser()
|
||||
|
||||
app.dependency_overrides[get_verification_service] = lambda: mock_service
|
||||
app.dependency_overrides[get_happ_parser] = lambda: mock_parser
|
||||
app.dependency_overrides[get_dns_service] = lambda: create_mock_dns_service(True)
|
||||
app.dependency_overrides[get_email_service] = lambda: create_mock_email_service()
|
||||
# HTML fetcher returns page with no email
|
||||
app.dependency_overrides[get_html_fetcher] = lambda: create_mock_html_fetcher(email=None)
|
||||
app.dependency_overrides[get_relme_parser] = create_mock_relme_parser
|
||||
app.dependency_overrides[get_happ_parser] = lambda: create_mock_happ_parser()
|
||||
app.dependency_overrides[get_auth_session_service] = lambda: create_mock_auth_session_service()
|
||||
|
||||
try:
|
||||
with TestClient(app) as client:
|
||||
@@ -367,29 +387,42 @@ class TestFullVerificationFlow:
|
||||
def test_full_flow_new_domain(
|
||||
self, configured_app, valid_auth_params
|
||||
):
|
||||
"""Test complete flow: unverified domain -> verify code -> consent."""
|
||||
"""Test complete flow: authorize -> verify code -> consent."""
|
||||
app, db_path = configured_app
|
||||
from gondulf.dependencies import get_verification_service, get_happ_parser, get_database
|
||||
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 sqlalchemy import text
|
||||
|
||||
# Clear any pre-existing verified domain to ensure test isolation
|
||||
db = Database(f"sqlite:///{db_path}")
|
||||
db.initialize()
|
||||
now = datetime.utcnow()
|
||||
with db.get_engine().begin() as conn:
|
||||
conn.execute(text("DELETE FROM domains WHERE domain = :domain"), {"domain": "user.example.com"})
|
||||
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
|
||||
|
||||
mock_service = create_mock_verification_service(start_success=True, verify_success=True)
|
||||
mock_session = create_mock_auth_session_service()
|
||||
mock_parser = create_mock_happ_parser()
|
||||
|
||||
app.dependency_overrides[get_verification_service] = lambda: mock_service
|
||||
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] = create_mock_relme_parser
|
||||
app.dependency_overrides[get_happ_parser] = lambda: mock_parser
|
||||
app.dependency_overrides[get_auth_session_service] = lambda: mock_session
|
||||
|
||||
try:
|
||||
with TestClient(app) as client:
|
||||
# Step 1: GET /authorize -> should show verification form
|
||||
# Step 1: GET /authorize -> should show verification form (always!)
|
||||
response1 = client.get("/authorize", params=valid_auth_params)
|
||||
|
||||
assert response1.status_code == 200
|
||||
@@ -397,16 +430,8 @@ class TestFullVerificationFlow:
|
||||
|
||||
# Step 2: POST /authorize/verify-code -> should show consent
|
||||
form_data = {
|
||||
"domain": "user.example.com",
|
||||
"session_id": "test_session_123",
|
||||
"code": "123456",
|
||||
"client_id": valid_auth_params["client_id"],
|
||||
"redirect_uri": valid_auth_params["redirect_uri"],
|
||||
"response_type": valid_auth_params["response_type"],
|
||||
"state": valid_auth_params["state"],
|
||||
"code_challenge": valid_auth_params["code_challenge"],
|
||||
"code_challenge_method": valid_auth_params["code_challenge_method"],
|
||||
"scope": "",
|
||||
"me": valid_auth_params["me"],
|
||||
}
|
||||
|
||||
response2 = client.post("/authorize/verify-code", data=form_data)
|
||||
@@ -422,35 +447,38 @@ class TestFullVerificationFlow:
|
||||
):
|
||||
"""Test that user can retry with correct code after failure."""
|
||||
app, _ = configured_app
|
||||
from gondulf.dependencies import get_verification_service, get_happ_parser
|
||||
from gondulf.dependencies import get_happ_parser, get_auth_session_service
|
||||
from gondulf.services.auth_session import CodeVerificationError
|
||||
|
||||
mock_service = Mock()
|
||||
# First verify_email_code call fails, second succeeds
|
||||
mock_service.verify_email_code.side_effect = [
|
||||
{"success": False, "error": "invalid_code"},
|
||||
{"success": True, "email": "test@example.com"}
|
||||
mock_session = create_mock_auth_session_service()
|
||||
# First verify_code call fails, second succeeds
|
||||
mock_session.verify_code.side_effect = [
|
||||
CodeVerificationError("Invalid code"),
|
||||
{
|
||||
"session_id": "test_session_123",
|
||||
"me": "https://user.example.com",
|
||||
"email": "test@example.com",
|
||||
"code_verified": True,
|
||||
"client_id": "https://app.example.com",
|
||||
"redirect_uri": "https://app.example.com/callback",
|
||||
"state": "test123",
|
||||
"code_challenge": "E9Melhoa2OwvFrEMTJguCHaoeK1t8URWbuGJSstw-cM",
|
||||
"code_challenge_method": "S256",
|
||||
"scope": "",
|
||||
"response_type": "code"
|
||||
}
|
||||
]
|
||||
mock_service.code_storage = Mock()
|
||||
mock_service.code_storage.get.return_value = "test@example.com"
|
||||
|
||||
mock_parser = create_mock_happ_parser()
|
||||
|
||||
app.dependency_overrides[get_verification_service] = lambda: mock_service
|
||||
app.dependency_overrides[get_auth_session_service] = lambda: mock_session
|
||||
app.dependency_overrides[get_happ_parser] = lambda: mock_parser
|
||||
|
||||
try:
|
||||
with TestClient(app) as client:
|
||||
form_data = {
|
||||
"domain": "user.example.com",
|
||||
"session_id": "test_session_123",
|
||||
"code": "000000", # Wrong code
|
||||
"client_id": valid_auth_params["client_id"],
|
||||
"redirect_uri": valid_auth_params["redirect_uri"],
|
||||
"response_type": valid_auth_params["response_type"],
|
||||
"state": valid_auth_params["state"],
|
||||
"code_challenge": valid_auth_params["code_challenge"],
|
||||
"code_challenge_method": valid_auth_params["code_challenge_method"],
|
||||
"scope": "",
|
||||
"me": valid_auth_params["me"],
|
||||
}
|
||||
|
||||
# First attempt with wrong code
|
||||
@@ -468,36 +496,48 @@ class TestFullVerificationFlow:
|
||||
|
||||
|
||||
class TestSecurityRequirements:
|
||||
"""Tests for security requirements of the fix."""
|
||||
"""Tests for security requirements - email auth required every login."""
|
||||
|
||||
def test_unverified_domain_never_sees_consent_directly(
|
||||
self, configured_app, valid_auth_params
|
||||
):
|
||||
"""Critical: Unverified domains must NEVER see consent page directly."""
|
||||
"""Critical: Even DNS-verified domains must authenticate via email every time."""
|
||||
app, db_path = configured_app
|
||||
from gondulf.dependencies import get_verification_service, get_happ_parser, get_database
|
||||
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 sqlalchemy import text
|
||||
|
||||
# Clear any pre-existing verified domain to ensure test isolation
|
||||
db = Database(f"sqlite:///{db_path}")
|
||||
db.initialize()
|
||||
now = datetime.utcnow()
|
||||
with db.get_engine().begin() as conn:
|
||||
conn.execute(text("DELETE FROM domains WHERE domain = :domain"), {"domain": "user.example.com"})
|
||||
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}
|
||||
)
|
||||
|
||||
mock_session = create_mock_auth_session_service()
|
||||
|
||||
app.dependency_overrides[get_database] = lambda: db
|
||||
|
||||
mock_service = create_mock_verification_service(start_success=True)
|
||||
mock_parser = create_mock_happ_parser()
|
||||
|
||||
app.dependency_overrides[get_verification_service] = lambda: mock_service
|
||||
app.dependency_overrides[get_happ_parser] = lambda: mock_parser
|
||||
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] = create_mock_relme_parser
|
||||
app.dependency_overrides[get_happ_parser] = lambda: create_mock_happ_parser()
|
||||
app.dependency_overrides[get_auth_session_service] = lambda: mock_session
|
||||
|
||||
try:
|
||||
with TestClient(app) as client:
|
||||
response = client.get("/authorize", params=valid_auth_params)
|
||||
|
||||
# The consent page should NOT be shown
|
||||
# CRITICAL: The consent page should NOT be shown without email verification
|
||||
assert "Authorization Request" not in response.text
|
||||
# Verify code page should be shown instead
|
||||
assert "Verify Your Identity" in response.text
|
||||
@@ -508,14 +548,36 @@ class TestSecurityRequirements:
|
||||
self, configured_app, valid_auth_params
|
||||
):
|
||||
"""Test that state parameter is preserved through verification flow."""
|
||||
app, _ = configured_app
|
||||
from gondulf.dependencies import get_verification_service, get_happ_parser
|
||||
app, db_path = configured_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 sqlalchemy import text
|
||||
|
||||
mock_service = create_mock_verification_service(start_success=True)
|
||||
mock_parser = create_mock_happ_parser()
|
||||
db = Database(f"sqlite:///{db_path}")
|
||||
db.initialize()
|
||||
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_verification_service] = lambda: mock_service
|
||||
app.dependency_overrides[get_happ_parser] = lambda: mock_parser
|
||||
mock_session = create_mock_auth_session_service()
|
||||
|
||||
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] = create_mock_relme_parser
|
||||
app.dependency_overrides[get_happ_parser] = lambda: create_mock_happ_parser()
|
||||
app.dependency_overrides[get_auth_session_service] = lambda: mock_session
|
||||
|
||||
try:
|
||||
unique_state = "unique_state_abc123xyz"
|
||||
@@ -526,7 +588,9 @@ class TestSecurityRequirements:
|
||||
response = client.get("/authorize", params=params)
|
||||
|
||||
assert response.status_code == 200
|
||||
# State should be in hidden form field
|
||||
assert f'value="{unique_state}"' in response.text or f"value='{unique_state}'" in response.text
|
||||
# State is now stored in session, so we check session_id is present
|
||||
assert 'name="session_id"' in response.text
|
||||
# The state should be stored in the session service
|
||||
assert mock_session.create_session.called
|
||||
finally:
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
630
tests/unit/test_auth_session.py
Normal file
630
tests/unit/test_auth_session.py
Normal file
@@ -0,0 +1,630 @@
|
||||
"""
|
||||
Unit tests for AuthSessionService.
|
||||
|
||||
Tests the per-login authentication session management that ensures
|
||||
email verification is required on EVERY login, never cached.
|
||||
|
||||
See ADR-010 for the architectural decision behind this design.
|
||||
"""
|
||||
import hashlib
|
||||
import time
|
||||
from datetime import datetime, timedelta
|
||||
from unittest.mock import MagicMock, Mock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from gondulf.services.auth_session import (
|
||||
MAX_CODE_ATTEMPTS,
|
||||
SESSION_TTL_MINUTES,
|
||||
AuthSessionError,
|
||||
AuthSessionService,
|
||||
CodeVerificationError,
|
||||
MaxAttemptsExceededError,
|
||||
SessionExpiredError,
|
||||
SessionNotFoundError,
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_database():
|
||||
"""Create a mock database for testing."""
|
||||
mock_db = Mock()
|
||||
mock_engine = MagicMock()
|
||||
mock_db.get_engine.return_value = mock_engine
|
||||
return mock_db
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def auth_session_service(mock_database):
|
||||
"""Create AuthSessionService with mock database."""
|
||||
return AuthSessionService(database=mock_database)
|
||||
|
||||
|
||||
class TestAuthSessionServiceInit:
|
||||
"""Tests for AuthSessionService initialization."""
|
||||
|
||||
def test_initialization(self, mock_database):
|
||||
"""Test service initializes correctly."""
|
||||
service = AuthSessionService(database=mock_database)
|
||||
assert service.database == mock_database
|
||||
|
||||
|
||||
class TestSessionIdGeneration:
|
||||
"""Tests for session ID generation."""
|
||||
|
||||
def test_generate_session_id_is_string(self, auth_session_service):
|
||||
"""Test session ID is a string."""
|
||||
session_id = auth_session_service._generate_session_id()
|
||||
assert isinstance(session_id, str)
|
||||
|
||||
def test_generate_session_id_is_unique(self, auth_session_service):
|
||||
"""Test session IDs are unique."""
|
||||
ids = [auth_session_service._generate_session_id() for _ in range(100)]
|
||||
assert len(set(ids)) == 100
|
||||
|
||||
def test_generate_session_id_is_long_enough(self, auth_session_service):
|
||||
"""Test session ID has sufficient entropy."""
|
||||
session_id = auth_session_service._generate_session_id()
|
||||
# URL-safe base64 of 32 bytes = ~43 characters
|
||||
assert len(session_id) >= 40
|
||||
|
||||
|
||||
class TestVerificationCodeGeneration:
|
||||
"""Tests for verification code generation."""
|
||||
|
||||
def test_generate_code_is_6_digits(self, auth_session_service):
|
||||
"""Test verification code is exactly 6 digits."""
|
||||
code = auth_session_service._generate_verification_code()
|
||||
assert len(code) == 6
|
||||
assert code.isdigit()
|
||||
|
||||
def test_generate_code_is_padded(self, auth_session_service):
|
||||
"""Test verification code is zero-padded."""
|
||||
# Generate many codes to test padding
|
||||
for _ in range(100):
|
||||
code = auth_session_service._generate_verification_code()
|
||||
assert len(code) == 6
|
||||
|
||||
def test_generate_code_varies(self, auth_session_service):
|
||||
"""Test verification codes are not constant."""
|
||||
codes = [auth_session_service._generate_verification_code() for _ in range(100)]
|
||||
# With 6 digits, 100 codes should have significant variation
|
||||
assert len(set(codes)) > 50
|
||||
|
||||
|
||||
class TestCodeHashing:
|
||||
"""Tests for code hashing."""
|
||||
|
||||
def test_hash_code_produces_sha256(self, auth_session_service):
|
||||
"""Test code hashing produces SHA-256 hash."""
|
||||
code = "123456"
|
||||
hashed = auth_session_service._hash_code(code)
|
||||
expected = hashlib.sha256(code.encode()).hexdigest()
|
||||
assert hashed == expected
|
||||
|
||||
def test_hash_code_is_deterministic(self, auth_session_service):
|
||||
"""Test same code produces same hash."""
|
||||
code = "123456"
|
||||
hash1 = auth_session_service._hash_code(code)
|
||||
hash2 = auth_session_service._hash_code(code)
|
||||
assert hash1 == hash2
|
||||
|
||||
def test_different_codes_produce_different_hashes(self, auth_session_service):
|
||||
"""Test different codes produce different hashes."""
|
||||
hash1 = auth_session_service._hash_code("123456")
|
||||
hash2 = auth_session_service._hash_code("654321")
|
||||
assert hash1 != hash2
|
||||
|
||||
|
||||
class TestCreateSession:
|
||||
"""Tests for session creation."""
|
||||
|
||||
def test_create_session_returns_session_id(self, auth_session_service, mock_database):
|
||||
"""Test session creation returns session ID."""
|
||||
# Setup mock to track execute calls
|
||||
mock_conn = MagicMock()
|
||||
mock_database.get_engine.return_value.begin.return_value.__enter__ = Mock(return_value=mock_conn)
|
||||
mock_database.get_engine.return_value.begin.return_value.__exit__ = Mock(return_value=False)
|
||||
|
||||
result = auth_session_service.create_session(
|
||||
me="https://user.example.com",
|
||||
email="user@example.com",
|
||||
client_id="https://app.example.com",
|
||||
redirect_uri="https://app.example.com/callback",
|
||||
state="xyz123",
|
||||
code_challenge="challenge123",
|
||||
code_challenge_method="S256",
|
||||
scope="",
|
||||
response_type="id"
|
||||
)
|
||||
|
||||
assert "session_id" in result
|
||||
assert isinstance(result["session_id"], str)
|
||||
assert len(result["session_id"]) >= 40
|
||||
|
||||
def test_create_session_returns_verification_code(self, auth_session_service, mock_database):
|
||||
"""Test session creation returns verification code."""
|
||||
mock_conn = MagicMock()
|
||||
mock_database.get_engine.return_value.begin.return_value.__enter__ = Mock(return_value=mock_conn)
|
||||
mock_database.get_engine.return_value.begin.return_value.__exit__ = Mock(return_value=False)
|
||||
|
||||
result = auth_session_service.create_session(
|
||||
me="https://user.example.com",
|
||||
email="user@example.com",
|
||||
client_id="https://app.example.com",
|
||||
redirect_uri="https://app.example.com/callback",
|
||||
state="xyz123",
|
||||
code_challenge="challenge123",
|
||||
code_challenge_method="S256",
|
||||
scope="",
|
||||
response_type="id"
|
||||
)
|
||||
|
||||
assert "verification_code" in result
|
||||
assert len(result["verification_code"]) == 6
|
||||
assert result["verification_code"].isdigit()
|
||||
|
||||
def test_create_session_returns_expiration(self, auth_session_service, mock_database):
|
||||
"""Test session creation returns expiration time."""
|
||||
mock_conn = MagicMock()
|
||||
mock_database.get_engine.return_value.begin.return_value.__enter__ = Mock(return_value=mock_conn)
|
||||
mock_database.get_engine.return_value.begin.return_value.__exit__ = Mock(return_value=False)
|
||||
|
||||
result = auth_session_service.create_session(
|
||||
me="https://user.example.com",
|
||||
email="user@example.com",
|
||||
client_id="https://app.example.com",
|
||||
redirect_uri="https://app.example.com/callback",
|
||||
state="xyz123",
|
||||
code_challenge="challenge123",
|
||||
code_challenge_method="S256",
|
||||
scope="",
|
||||
response_type="id"
|
||||
)
|
||||
|
||||
assert "expires_at" in result
|
||||
assert isinstance(result["expires_at"], datetime)
|
||||
# Expiration should be approximately SESSION_TTL_MINUTES from now
|
||||
expected_expiry = datetime.utcnow() + timedelta(minutes=SESSION_TTL_MINUTES)
|
||||
assert abs((result["expires_at"] - expected_expiry).total_seconds()) < 5
|
||||
|
||||
def test_create_session_stores_hashed_code(self, auth_session_service, mock_database):
|
||||
"""Test that verification code is stored hashed, not plain."""
|
||||
mock_conn = MagicMock()
|
||||
mock_database.get_engine.return_value.begin.return_value.__enter__ = Mock(return_value=mock_conn)
|
||||
mock_database.get_engine.return_value.begin.return_value.__exit__ = Mock(return_value=False)
|
||||
|
||||
result = auth_session_service.create_session(
|
||||
me="https://user.example.com",
|
||||
email="user@example.com",
|
||||
client_id="https://app.example.com",
|
||||
redirect_uri="https://app.example.com/callback",
|
||||
state="xyz123",
|
||||
code_challenge="challenge123",
|
||||
code_challenge_method="S256",
|
||||
scope="",
|
||||
response_type="id"
|
||||
)
|
||||
|
||||
# Verify execute was called
|
||||
assert mock_conn.execute.called
|
||||
|
||||
# Check the parameters passed to execute
|
||||
call_args = mock_conn.execute.call_args
|
||||
params = call_args[0][1]
|
||||
|
||||
# Code hash should be SHA-256 of the verification code
|
||||
expected_hash = hashlib.sha256(result["verification_code"].encode()).hexdigest()
|
||||
assert params["code_hash"] == expected_hash
|
||||
|
||||
def test_create_session_handles_database_error(self, auth_session_service, mock_database):
|
||||
"""Test session creation handles database errors."""
|
||||
mock_database.get_engine.return_value.begin.side_effect = Exception("Database error")
|
||||
|
||||
with pytest.raises(AuthSessionError) as exc_info:
|
||||
auth_session_service.create_session(
|
||||
me="https://user.example.com",
|
||||
email="user@example.com",
|
||||
client_id="https://app.example.com",
|
||||
redirect_uri="https://app.example.com/callback",
|
||||
state="xyz123",
|
||||
code_challenge="challenge123",
|
||||
code_challenge_method="S256",
|
||||
scope="",
|
||||
response_type="id"
|
||||
)
|
||||
|
||||
assert "Failed to create session" in str(exc_info.value)
|
||||
|
||||
|
||||
class TestGetSession:
|
||||
"""Tests for session retrieval."""
|
||||
|
||||
def test_get_session_not_found(self, auth_session_service, mock_database):
|
||||
"""Test getting non-existent session raises error."""
|
||||
mock_conn = MagicMock()
|
||||
mock_result = MagicMock()
|
||||
mock_result.fetchone.return_value = None
|
||||
mock_conn.execute.return_value = mock_result
|
||||
|
||||
mock_database.get_engine.return_value.connect.return_value.__enter__ = Mock(return_value=mock_conn)
|
||||
mock_database.get_engine.return_value.connect.return_value.__exit__ = Mock(return_value=False)
|
||||
|
||||
with pytest.raises(SessionNotFoundError):
|
||||
auth_session_service.get_session("nonexistent_session_id")
|
||||
|
||||
def test_get_session_expired(self, auth_session_service, mock_database):
|
||||
"""Test getting expired session raises error."""
|
||||
mock_conn = MagicMock()
|
||||
mock_result = MagicMock()
|
||||
# Return a session that expired in the past
|
||||
expired_time = datetime.utcnow() - timedelta(hours=1)
|
||||
mock_result.fetchone.return_value = (
|
||||
"session123", "https://user.example.com", "user@example.com",
|
||||
False, 0, "https://app.example.com", "https://app.example.com/callback",
|
||||
"xyz", "challenge", "S256", "", "id",
|
||||
datetime.utcnow() - timedelta(hours=2), expired_time
|
||||
)
|
||||
mock_conn.execute.return_value = mock_result
|
||||
|
||||
mock_database.get_engine.return_value.connect.return_value.__enter__ = Mock(return_value=mock_conn)
|
||||
mock_database.get_engine.return_value.connect.return_value.__exit__ = Mock(return_value=False)
|
||||
|
||||
# Also mock the delete for cleanup
|
||||
mock_del_conn = MagicMock()
|
||||
mock_database.get_engine.return_value.begin.return_value.__enter__ = Mock(return_value=mock_del_conn)
|
||||
mock_database.get_engine.return_value.begin.return_value.__exit__ = Mock(return_value=False)
|
||||
|
||||
with pytest.raises(SessionExpiredError):
|
||||
auth_session_service.get_session("session123")
|
||||
|
||||
def test_get_session_returns_data(self, auth_session_service, mock_database):
|
||||
"""Test getting valid session returns all data."""
|
||||
mock_conn = MagicMock()
|
||||
mock_result = MagicMock()
|
||||
future_time = datetime.utcnow() + timedelta(minutes=5)
|
||||
mock_result.fetchone.return_value = (
|
||||
"session123", "https://user.example.com", "user@example.com",
|
||||
True, 1, "https://app.example.com", "https://app.example.com/callback",
|
||||
"xyz", "challenge", "S256", "profile", "code",
|
||||
datetime.utcnow(), future_time
|
||||
)
|
||||
mock_conn.execute.return_value = mock_result
|
||||
|
||||
mock_database.get_engine.return_value.connect.return_value.__enter__ = Mock(return_value=mock_conn)
|
||||
mock_database.get_engine.return_value.connect.return_value.__exit__ = Mock(return_value=False)
|
||||
|
||||
result = auth_session_service.get_session("session123")
|
||||
|
||||
assert result["session_id"] == "session123"
|
||||
assert result["me"] == "https://user.example.com"
|
||||
assert result["email"] == "user@example.com"
|
||||
assert result["code_verified"] is True
|
||||
assert result["client_id"] == "https://app.example.com"
|
||||
assert result["response_type"] == "code"
|
||||
|
||||
|
||||
class TestVerifyCode:
|
||||
"""Tests for code verification - the core authentication step."""
|
||||
|
||||
def test_verify_code_success(self, auth_session_service, mock_database):
|
||||
"""Test successful code verification."""
|
||||
code = "123456"
|
||||
code_hash = hashlib.sha256(code.encode()).hexdigest()
|
||||
future_time = datetime.utcnow() + timedelta(minutes=5)
|
||||
|
||||
# Mock for initial fetch
|
||||
mock_conn = MagicMock()
|
||||
mock_result = MagicMock()
|
||||
mock_result.fetchone.return_value = (
|
||||
"session123", "https://user.example.com", "user@example.com",
|
||||
code_hash, False, 0, "https://app.example.com",
|
||||
"https://app.example.com/callback", "xyz", "challenge", "S256",
|
||||
"", "id", future_time
|
||||
)
|
||||
mock_conn.execute.return_value = mock_result
|
||||
|
||||
mock_database.get_engine.return_value.connect.return_value.__enter__ = Mock(return_value=mock_conn)
|
||||
mock_database.get_engine.return_value.connect.return_value.__exit__ = Mock(return_value=False)
|
||||
|
||||
# Mock for update
|
||||
mock_update_conn = MagicMock()
|
||||
mock_database.get_engine.return_value.begin.return_value.__enter__ = Mock(return_value=mock_update_conn)
|
||||
mock_database.get_engine.return_value.begin.return_value.__exit__ = Mock(return_value=False)
|
||||
|
||||
result = auth_session_service.verify_code("session123", code)
|
||||
|
||||
assert result["code_verified"] is True
|
||||
assert result["me"] == "https://user.example.com"
|
||||
|
||||
def test_verify_code_wrong_code(self, auth_session_service, mock_database):
|
||||
"""Test code verification with wrong code."""
|
||||
correct_code = "123456"
|
||||
wrong_code = "654321"
|
||||
code_hash = hashlib.sha256(correct_code.encode()).hexdigest()
|
||||
future_time = datetime.utcnow() + timedelta(minutes=5)
|
||||
|
||||
mock_conn = MagicMock()
|
||||
mock_result = MagicMock()
|
||||
mock_result.fetchone.return_value = (
|
||||
"session123", "https://user.example.com", "user@example.com",
|
||||
code_hash, False, 0, "https://app.example.com",
|
||||
"https://app.example.com/callback", "xyz", "challenge", "S256",
|
||||
"", "id", future_time
|
||||
)
|
||||
mock_conn.execute.return_value = mock_result
|
||||
|
||||
mock_database.get_engine.return_value.connect.return_value.__enter__ = Mock(return_value=mock_conn)
|
||||
mock_database.get_engine.return_value.connect.return_value.__exit__ = Mock(return_value=False)
|
||||
|
||||
# Mock for attempt increment
|
||||
mock_update_conn = MagicMock()
|
||||
mock_database.get_engine.return_value.begin.return_value.__enter__ = Mock(return_value=mock_update_conn)
|
||||
mock_database.get_engine.return_value.begin.return_value.__exit__ = Mock(return_value=False)
|
||||
|
||||
with pytest.raises(CodeVerificationError):
|
||||
auth_session_service.verify_code("session123", wrong_code)
|
||||
|
||||
def test_verify_code_max_attempts_exceeded(self, auth_session_service, mock_database):
|
||||
"""Test code verification fails after max attempts."""
|
||||
code = "123456"
|
||||
code_hash = hashlib.sha256(code.encode()).hexdigest()
|
||||
future_time = datetime.utcnow() + timedelta(minutes=5)
|
||||
|
||||
mock_conn = MagicMock()
|
||||
mock_result = MagicMock()
|
||||
mock_result.fetchone.return_value = (
|
||||
"session123", "https://user.example.com", "user@example.com",
|
||||
code_hash, False, MAX_CODE_ATTEMPTS, "https://app.example.com",
|
||||
"https://app.example.com/callback", "xyz", "challenge", "S256",
|
||||
"", "id", future_time
|
||||
)
|
||||
mock_conn.execute.return_value = mock_result
|
||||
|
||||
mock_database.get_engine.return_value.connect.return_value.__enter__ = Mock(return_value=mock_conn)
|
||||
mock_database.get_engine.return_value.connect.return_value.__exit__ = Mock(return_value=False)
|
||||
|
||||
# Mock for session deletion
|
||||
mock_del_conn = MagicMock()
|
||||
mock_database.get_engine.return_value.begin.return_value.__enter__ = Mock(return_value=mock_del_conn)
|
||||
mock_database.get_engine.return_value.begin.return_value.__exit__ = Mock(return_value=False)
|
||||
|
||||
with pytest.raises(MaxAttemptsExceededError):
|
||||
auth_session_service.verify_code("session123", code)
|
||||
|
||||
def test_verify_code_session_not_found(self, auth_session_service, mock_database):
|
||||
"""Test code verification with non-existent session."""
|
||||
mock_conn = MagicMock()
|
||||
mock_result = MagicMock()
|
||||
mock_result.fetchone.return_value = None
|
||||
mock_conn.execute.return_value = mock_result
|
||||
|
||||
mock_database.get_engine.return_value.connect.return_value.__enter__ = Mock(return_value=mock_conn)
|
||||
mock_database.get_engine.return_value.connect.return_value.__exit__ = Mock(return_value=False)
|
||||
|
||||
with pytest.raises(SessionNotFoundError):
|
||||
auth_session_service.verify_code("nonexistent", "123456")
|
||||
|
||||
def test_verify_code_session_expired(self, auth_session_service, mock_database):
|
||||
"""Test code verification with expired session."""
|
||||
code = "123456"
|
||||
code_hash = hashlib.sha256(code.encode()).hexdigest()
|
||||
expired_time = datetime.utcnow() - timedelta(hours=1)
|
||||
|
||||
mock_conn = MagicMock()
|
||||
mock_result = MagicMock()
|
||||
mock_result.fetchone.return_value = (
|
||||
"session123", "https://user.example.com", "user@example.com",
|
||||
code_hash, False, 0, "https://app.example.com",
|
||||
"https://app.example.com/callback", "xyz", "challenge", "S256",
|
||||
"", "id", expired_time
|
||||
)
|
||||
mock_conn.execute.return_value = mock_result
|
||||
|
||||
mock_database.get_engine.return_value.connect.return_value.__enter__ = Mock(return_value=mock_conn)
|
||||
mock_database.get_engine.return_value.connect.return_value.__exit__ = Mock(return_value=False)
|
||||
|
||||
# Mock for session deletion
|
||||
mock_del_conn = MagicMock()
|
||||
mock_database.get_engine.return_value.begin.return_value.__enter__ = Mock(return_value=mock_del_conn)
|
||||
mock_database.get_engine.return_value.begin.return_value.__exit__ = Mock(return_value=False)
|
||||
|
||||
with pytest.raises(SessionExpiredError):
|
||||
auth_session_service.verify_code("session123", code)
|
||||
|
||||
def test_verify_code_already_verified(self, auth_session_service, mock_database):
|
||||
"""Test code verification on already verified session returns success."""
|
||||
code = "123456"
|
||||
code_hash = hashlib.sha256(code.encode()).hexdigest()
|
||||
future_time = datetime.utcnow() + timedelta(minutes=5)
|
||||
|
||||
mock_conn = MagicMock()
|
||||
mock_result = MagicMock()
|
||||
mock_result.fetchone.return_value = (
|
||||
"session123", "https://user.example.com", "user@example.com",
|
||||
code_hash, True, 1, "https://app.example.com", # Already verified
|
||||
"https://app.example.com/callback", "xyz", "challenge", "S256",
|
||||
"", "id", future_time
|
||||
)
|
||||
mock_conn.execute.return_value = mock_result
|
||||
|
||||
mock_database.get_engine.return_value.connect.return_value.__enter__ = Mock(return_value=mock_conn)
|
||||
mock_database.get_engine.return_value.connect.return_value.__exit__ = Mock(return_value=False)
|
||||
|
||||
result = auth_session_service.verify_code("session123", code)
|
||||
|
||||
assert result["code_verified"] is True
|
||||
|
||||
|
||||
class TestIsSessionVerified:
|
||||
"""Tests for checking session verification status."""
|
||||
|
||||
def test_is_session_verified_true(self, auth_session_service, mock_database):
|
||||
"""Test is_session_verified returns True for verified session."""
|
||||
future_time = datetime.utcnow() + timedelta(minutes=5)
|
||||
|
||||
mock_conn = MagicMock()
|
||||
mock_result = MagicMock()
|
||||
mock_result.fetchone.return_value = (
|
||||
"session123", "https://user.example.com", "user@example.com",
|
||||
True, 1, "https://app.example.com", "https://app.example.com/callback",
|
||||
"xyz", "challenge", "S256", "", "id",
|
||||
datetime.utcnow(), future_time
|
||||
)
|
||||
mock_conn.execute.return_value = mock_result
|
||||
|
||||
mock_database.get_engine.return_value.connect.return_value.__enter__ = Mock(return_value=mock_conn)
|
||||
mock_database.get_engine.return_value.connect.return_value.__exit__ = Mock(return_value=False)
|
||||
|
||||
assert auth_session_service.is_session_verified("session123") is True
|
||||
|
||||
def test_is_session_verified_false(self, auth_session_service, mock_database):
|
||||
"""Test is_session_verified returns False for unverified session."""
|
||||
future_time = datetime.utcnow() + timedelta(minutes=5)
|
||||
|
||||
mock_conn = MagicMock()
|
||||
mock_result = MagicMock()
|
||||
mock_result.fetchone.return_value = (
|
||||
"session123", "https://user.example.com", "user@example.com",
|
||||
False, 0, "https://app.example.com", "https://app.example.com/callback",
|
||||
"xyz", "challenge", "S256", "", "id",
|
||||
datetime.utcnow(), future_time
|
||||
)
|
||||
mock_conn.execute.return_value = mock_result
|
||||
|
||||
mock_database.get_engine.return_value.connect.return_value.__enter__ = Mock(return_value=mock_conn)
|
||||
mock_database.get_engine.return_value.connect.return_value.__exit__ = Mock(return_value=False)
|
||||
|
||||
assert auth_session_service.is_session_verified("session123") is False
|
||||
|
||||
def test_is_session_verified_not_found(self, auth_session_service, mock_database):
|
||||
"""Test is_session_verified returns False for non-existent session."""
|
||||
mock_conn = MagicMock()
|
||||
mock_result = MagicMock()
|
||||
mock_result.fetchone.return_value = None
|
||||
mock_conn.execute.return_value = mock_result
|
||||
|
||||
mock_database.get_engine.return_value.connect.return_value.__enter__ = Mock(return_value=mock_conn)
|
||||
mock_database.get_engine.return_value.connect.return_value.__exit__ = Mock(return_value=False)
|
||||
|
||||
assert auth_session_service.is_session_verified("nonexistent") is False
|
||||
|
||||
|
||||
class TestDeleteSession:
|
||||
"""Tests for session deletion."""
|
||||
|
||||
def test_delete_session(self, auth_session_service, mock_database):
|
||||
"""Test session deletion."""
|
||||
mock_conn = MagicMock()
|
||||
mock_database.get_engine.return_value.begin.return_value.__enter__ = Mock(return_value=mock_conn)
|
||||
mock_database.get_engine.return_value.begin.return_value.__exit__ = Mock(return_value=False)
|
||||
|
||||
# Should not raise
|
||||
auth_session_service.delete_session("session123")
|
||||
|
||||
# Verify execute was called
|
||||
assert mock_conn.execute.called
|
||||
|
||||
|
||||
class TestCleanupExpiredSessions:
|
||||
"""Tests for expired session cleanup."""
|
||||
|
||||
def test_cleanup_returns_count(self, auth_session_service, mock_database):
|
||||
"""Test cleanup returns number of deleted sessions."""
|
||||
mock_conn = MagicMock()
|
||||
mock_result = MagicMock()
|
||||
mock_result.rowcount = 5
|
||||
mock_conn.execute.return_value = mock_result
|
||||
|
||||
mock_database.get_engine.return_value.begin.return_value.__enter__ = Mock(return_value=mock_conn)
|
||||
mock_database.get_engine.return_value.begin.return_value.__exit__ = Mock(return_value=False)
|
||||
|
||||
count = auth_session_service.cleanup_expired_sessions()
|
||||
|
||||
assert count == 5
|
||||
|
||||
def test_cleanup_handles_error(self, auth_session_service, mock_database):
|
||||
"""Test cleanup handles database errors gracefully."""
|
||||
mock_database.get_engine.return_value.begin.side_effect = Exception("Database error")
|
||||
|
||||
count = auth_session_service.cleanup_expired_sessions()
|
||||
|
||||
assert count == 0
|
||||
|
||||
|
||||
class TestSecurityProperties:
|
||||
"""
|
||||
Tests verifying security properties of the authentication flow.
|
||||
|
||||
These tests ensure the critical security requirements from ADR-010 are met.
|
||||
"""
|
||||
|
||||
def test_code_is_never_stored_in_plain_text(self, auth_session_service, mock_database):
|
||||
"""
|
||||
CRITICAL: Verify that verification codes are never stored in plain text.
|
||||
|
||||
The verification code should be hashed before storage to prevent
|
||||
database compromise from exposing valid codes.
|
||||
"""
|
||||
mock_conn = MagicMock()
|
||||
mock_database.get_engine.return_value.begin.return_value.__enter__ = Mock(return_value=mock_conn)
|
||||
mock_database.get_engine.return_value.begin.return_value.__exit__ = Mock(return_value=False)
|
||||
|
||||
result = auth_session_service.create_session(
|
||||
me="https://user.example.com",
|
||||
email="user@example.com",
|
||||
client_id="https://app.example.com",
|
||||
redirect_uri="https://app.example.com/callback",
|
||||
state="xyz123",
|
||||
code_challenge="challenge123",
|
||||
code_challenge_method="S256",
|
||||
scope="",
|
||||
response_type="id"
|
||||
)
|
||||
|
||||
plain_code = result["verification_code"]
|
||||
call_args = mock_conn.execute.call_args
|
||||
params = call_args[0][1]
|
||||
|
||||
# The plain code should NOT appear in storage
|
||||
assert params.get("code_hash") != plain_code
|
||||
# The hash should be a SHA-256 hash (64 hex characters)
|
||||
assert len(params["code_hash"]) == 64
|
||||
|
||||
def test_session_id_has_sufficient_entropy(self, auth_session_service):
|
||||
"""
|
||||
CRITICAL: Verify session IDs have sufficient entropy to prevent guessing.
|
||||
|
||||
Session IDs must be cryptographically random with enough bits
|
||||
to prevent brute-force attacks.
|
||||
"""
|
||||
session_ids = [auth_session_service._generate_session_id() for _ in range(1000)]
|
||||
|
||||
# All should be unique
|
||||
assert len(set(session_ids)) == 1000
|
||||
|
||||
# Should be at least 32 bytes of entropy (256 bits)
|
||||
# URL-safe base64 of 32 bytes is ~43 characters
|
||||
for sid in session_ids:
|
||||
assert len(sid) >= 40
|
||||
|
||||
def test_code_verification_uses_constant_time_comparison(self, auth_session_service):
|
||||
"""
|
||||
CRITICAL: Verify code comparison uses constant-time algorithm.
|
||||
|
||||
This prevents timing attacks that could leak information about
|
||||
the correct code.
|
||||
"""
|
||||
# The implementation uses secrets.compare_digest which is constant-time
|
||||
# We verify the hash comparison pattern is correct
|
||||
code1 = "123456"
|
||||
code2 = "123456"
|
||||
hash1 = auth_session_service._hash_code(code1)
|
||||
hash2 = auth_session_service._hash_code(code2)
|
||||
|
||||
# Same codes should produce same hashes
|
||||
assert hash1 == hash2
|
||||
|
||||
# Different codes should produce different hashes
|
||||
hash3 = auth_session_service._hash_code("654321")
|
||||
assert hash1 != hash3
|
||||
@@ -175,15 +175,15 @@ class TestDatabaseMigrations:
|
||||
|
||||
engine = db.get_engine()
|
||||
with engine.connect() as conn:
|
||||
# Check migrations were recorded correctly (001, 002, and 003)
|
||||
# Check migrations were recorded correctly (001-005)
|
||||
result = conn.execute(text("SELECT COUNT(*) FROM migrations"))
|
||||
count = result.fetchone()[0]
|
||||
assert count == 3
|
||||
assert count == 5
|
||||
|
||||
# Verify all migrations are present
|
||||
result = conn.execute(text("SELECT version FROM migrations ORDER BY version"))
|
||||
versions = [row[0] for row in result]
|
||||
assert versions == [1, 2, 3]
|
||||
assert versions == [1, 2, 3, 4, 5]
|
||||
|
||||
def test_initialize_full_setup(self):
|
||||
"""Test initialize performs full database setup."""
|
||||
@@ -261,6 +261,7 @@ class TestMigrationSchemaCorrectness:
|
||||
"created_at",
|
||||
"verified_at",
|
||||
"two_factor",
|
||||
"last_checked", # Added in migration 005
|
||||
}
|
||||
|
||||
assert columns == expected_columns
|
||||
|
||||
Reference in New Issue
Block a user