Phil Skentelbery
052d3ad3e1
Implements both IndieAuth flows per W3C specification: - Authentication flow (response_type=id): Code redeemed at authorization endpoint, returns only user identity - Authorization flow (response_type=code): Code redeemed at token endpoint, returns access token Changes: - Authorization endpoint GET: Accept response_type=id (default) and code - Authorization endpoint POST: Handle code verification for authentication flow - Token endpoint: Validate response_type=code for authorization flow - Store response_type in authorization code metadata - Update metadata endpoint: response_types_supported=[code, id], code_challenge_methods_supported=[S256] The default behavior now correctly defaults to response_type=id when omitted, per IndieAuth spec section 5.2. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
267 lines
8.8 KiB
Python
267 lines
8.8 KiB
Python
"""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 for domain={domain}")
|
|
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,
|
|
response_type: str = "id"
|
|
) -> 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
|
|
response_type: "id" for authentication, "code" for authorization
|
|
|
|
Returns:
|
|
Authorization code
|
|
"""
|
|
# Generate authorization code
|
|
authorization_code = self._generate_authorization_code()
|
|
|
|
# Create metadata including response_type for flow determination during redemption
|
|
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,
|
|
"response_type": response_type,
|
|
"created_at": int(time.time()),
|
|
"expires_at": int(time.time()) + 600,
|
|
"used": False
|
|
}
|
|
|
|
# Store with prefix (CodeStore handles dict values natively)
|
|
storage_key = f"authz:{authorization_code}"
|
|
self.code_storage.store(storage_key, 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)
|