feat(phase-2): implement domain verification system
Implements complete domain verification flow with: - rel=me link verification service - HTML fetching with security controls - Rate limiting to prevent abuse - Email validation utilities - Authorization and verification API endpoints - User-facing templates for authorization and verification flows This completes Phase 2: Domain Verification as designed. Tests: - All Phase 2 unit tests passing - Coverage: 85% overall - Migration tests updated 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
263
src/gondulf/services/domain_verification.py
Normal file
263
src/gondulf/services/domain_verification.py
Normal file
@@ -0,0 +1,263 @@
|
||||
"""Domain verification service orchestrating two-factor verification."""
|
||||
import logging
|
||||
import secrets
|
||||
import time
|
||||
from typing import Any
|
||||
|
||||
from gondulf.dns import DNSService
|
||||
from gondulf.email import EmailService
|
||||
from gondulf.services.html_fetcher import HTMLFetcherService
|
||||
from gondulf.services.relme_parser import RelMeParser
|
||||
from gondulf.storage import CodeStore
|
||||
from gondulf.utils.validation import validate_email
|
||||
|
||||
logger = logging.getLogger("gondulf.domain_verification")
|
||||
|
||||
|
||||
class DomainVerificationService:
|
||||
"""Service for orchestrating two-factor domain verification (DNS + email)."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
dns_service: DNSService,
|
||||
email_service: EmailService,
|
||||
code_storage: CodeStore,
|
||||
html_fetcher: HTMLFetcherService,
|
||||
relme_parser: RelMeParser,
|
||||
) -> None:
|
||||
"""
|
||||
Initialize domain verification service.
|
||||
|
||||
Args:
|
||||
dns_service: DNS service for TXT record verification
|
||||
email_service: Email service for sending verification codes
|
||||
code_storage: Code storage for verification codes
|
||||
html_fetcher: HTML fetcher service for retrieving user homepage
|
||||
relme_parser: rel=me parser for extracting email from HTML
|
||||
"""
|
||||
self.dns_service = dns_service
|
||||
self.email_service = email_service
|
||||
self.code_storage = code_storage
|
||||
self.html_fetcher = html_fetcher
|
||||
self.relme_parser = relme_parser
|
||||
logger.debug("DomainVerificationService initialized")
|
||||
|
||||
def generate_verification_code(self) -> str:
|
||||
"""
|
||||
Generate a 6-digit numeric verification code.
|
||||
|
||||
Returns:
|
||||
6-digit numeric code as string
|
||||
"""
|
||||
return f"{secrets.randbelow(1000000):06d}"
|
||||
|
||||
def start_verification(self, domain: str, me_url: str) -> dict[str, Any]:
|
||||
"""
|
||||
Start two-factor verification process for domain.
|
||||
|
||||
Step 1: Verify DNS TXT record
|
||||
Step 2: Fetch homepage and extract email from rel=me
|
||||
Step 3: Send verification code to email
|
||||
Step 4: Store code for later verification
|
||||
|
||||
Args:
|
||||
domain: Domain to verify (e.g., "example.com")
|
||||
me_url: User's URL for verification (e.g., "https://example.com/")
|
||||
|
||||
Returns:
|
||||
Dict with verification result:
|
||||
- success: bool
|
||||
- email: masked email if successful
|
||||
- error: error code if failed
|
||||
"""
|
||||
logger.info(f"Starting verification for domain={domain} me_url={me_url}")
|
||||
|
||||
# Step 1: Verify DNS TXT record
|
||||
dns_verified = self._verify_dns_record(domain)
|
||||
if not dns_verified:
|
||||
logger.warning(f"DNS verification failed for domain={domain}")
|
||||
return {"success": False, "error": "dns_verification_failed"}
|
||||
|
||||
logger.info(f"DNS verification successful for domain={domain}")
|
||||
|
||||
# Step 2: Fetch homepage and extract email
|
||||
email = self._discover_email(me_url)
|
||||
if not email:
|
||||
logger.warning(f"Email discovery failed for me_url={me_url}")
|
||||
return {"success": False, "error": "email_discovery_failed"}
|
||||
|
||||
logger.info(f"Email discovered for domain={domain}")
|
||||
|
||||
# Validate email format
|
||||
if not validate_email(email):
|
||||
logger.warning(f"Invalid email format discovered: {email}")
|
||||
return {"success": False, "error": "invalid_email_format"}
|
||||
|
||||
# Step 3: Generate and send verification code
|
||||
code = self.generate_verification_code()
|
||||
|
||||
try:
|
||||
self.email_service.send_verification_code(email, code, domain)
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to send verification email: {e}")
|
||||
return {"success": False, "error": "email_send_failed"}
|
||||
|
||||
# Step 4: Store code for verification
|
||||
storage_key = f"email_verify:{domain}"
|
||||
self.code_storage.store(storage_key, code)
|
||||
|
||||
# Also store the email address for later retrieval
|
||||
email_key = f"email_addr:{domain}"
|
||||
self.code_storage.store(email_key, email)
|
||||
|
||||
logger.info(f"Verification code sent for domain={domain}")
|
||||
|
||||
# Return masked email
|
||||
from gondulf.utils.validation import mask_email
|
||||
return {
|
||||
"success": True,
|
||||
"email": mask_email(email),
|
||||
"verification_method": "email"
|
||||
}
|
||||
|
||||
def verify_email_code(self, domain: str, code: str) -> dict[str, Any]:
|
||||
"""
|
||||
Verify email code for domain.
|
||||
|
||||
Args:
|
||||
domain: Domain being verified
|
||||
code: Verification code from email
|
||||
|
||||
Returns:
|
||||
Dict with verification result:
|
||||
- success: bool
|
||||
- email: full email address if successful
|
||||
- error: error code if failed
|
||||
"""
|
||||
storage_key = f"email_verify:{domain}"
|
||||
email_key = f"email_addr:{domain}"
|
||||
|
||||
# Verify code
|
||||
if not self.code_storage.verify(storage_key, code):
|
||||
logger.warning(f"Email code verification failed for domain={domain}")
|
||||
return {"success": False, "error": "invalid_code"}
|
||||
|
||||
# Retrieve email address
|
||||
email = self.code_storage.get(email_key)
|
||||
if not email:
|
||||
logger.error(f"Email address not found for domain={domain}")
|
||||
return {"success": False, "error": "email_not_found"}
|
||||
|
||||
# Clean up email address from storage
|
||||
self.code_storage.delete(email_key)
|
||||
|
||||
logger.info(f"Email verification successful for domain={domain}")
|
||||
return {"success": True, "email": email}
|
||||
|
||||
def _verify_dns_record(self, domain: str) -> bool:
|
||||
"""
|
||||
Verify DNS TXT record for domain.
|
||||
|
||||
Checks for TXT record containing "gondulf-verify-domain"
|
||||
|
||||
Args:
|
||||
domain: Domain to verify
|
||||
|
||||
Returns:
|
||||
True if DNS verification successful, False otherwise
|
||||
"""
|
||||
try:
|
||||
return self.dns_service.verify_txt_record(
|
||||
domain,
|
||||
"gondulf-verify-domain"
|
||||
)
|
||||
except Exception as e:
|
||||
logger.error(f"DNS verification error for domain={domain}: {e}")
|
||||
return False
|
||||
|
||||
def _discover_email(self, me_url: str) -> str | None:
|
||||
"""
|
||||
Discover email address from user's homepage via rel=me links.
|
||||
|
||||
Args:
|
||||
me_url: User's URL to fetch
|
||||
|
||||
Returns:
|
||||
Email address if found, None otherwise
|
||||
"""
|
||||
try:
|
||||
# Fetch HTML
|
||||
html = self.html_fetcher.fetch(me_url)
|
||||
if not html:
|
||||
logger.warning(f"Failed to fetch HTML from {me_url}")
|
||||
return None
|
||||
|
||||
# Parse rel=me links and extract email
|
||||
email = self.relme_parser.find_email(html)
|
||||
if not email:
|
||||
logger.warning(f"No email found in rel=me links at {me_url}")
|
||||
return None
|
||||
|
||||
return email
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Email discovery error for {me_url}: {e}")
|
||||
return None
|
||||
|
||||
def create_authorization_code(
|
||||
self,
|
||||
client_id: str,
|
||||
redirect_uri: str,
|
||||
state: str,
|
||||
code_challenge: str,
|
||||
code_challenge_method: str,
|
||||
scope: str,
|
||||
me: str
|
||||
) -> str:
|
||||
"""
|
||||
Create authorization code with metadata.
|
||||
|
||||
Args:
|
||||
client_id: Client identifier
|
||||
redirect_uri: Redirect URI for callback
|
||||
state: Client state parameter
|
||||
code_challenge: PKCE code challenge
|
||||
code_challenge_method: PKCE method (S256)
|
||||
scope: Requested scope
|
||||
me: Verified user identity
|
||||
|
||||
Returns:
|
||||
Authorization code
|
||||
"""
|
||||
# Generate authorization code
|
||||
authorization_code = self._generate_authorization_code()
|
||||
|
||||
# Create metadata
|
||||
metadata = {
|
||||
"client_id": client_id,
|
||||
"redirect_uri": redirect_uri,
|
||||
"state": state,
|
||||
"code_challenge": code_challenge,
|
||||
"code_challenge_method": code_challenge_method,
|
||||
"scope": scope,
|
||||
"me": me,
|
||||
"created_at": int(time.time()),
|
||||
"expires_at": int(time.time()) + 600,
|
||||
"used": False
|
||||
}
|
||||
|
||||
# Store with prefix
|
||||
storage_key = f"authz:{authorization_code}"
|
||||
self.code_storage.store(storage_key, str(metadata))
|
||||
|
||||
logger.info(f"Authorization code created for client_id={client_id}")
|
||||
return authorization_code
|
||||
|
||||
def _generate_authorization_code(self) -> str:
|
||||
"""
|
||||
Generate secure random authorization code.
|
||||
|
||||
Returns:
|
||||
URL-safe authorization code
|
||||
"""
|
||||
return secrets.token_urlsafe(32)
|
||||
Reference in New Issue
Block a user