From 8550799bf81c61d9d02b45416d39285bb3572561 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Mon, 9 Feb 2026 11:51:58 +0000 Subject: [PATCH] Address code review feedback: improve XSS prevention, structured errors, API key logging, CSP warnings Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com> --- backend/app/api/api_v1/endpoints/reports.py | 6 +-- backend/app/core/security.py | 23 ++++++++- backend/app/main.py | 14 ++++-- backend/app/middleware/security.py | 56 +++------------------ backend/app/tests/test_security.py | 10 ++-- backend/app/utils/domain_validator.py | 46 +++++++++++------ 6 files changed, 77 insertions(+), 78 deletions(-) diff --git a/backend/app/api/api_v1/endpoints/reports.py b/backend/app/api/api_v1/endpoints/reports.py index 43456c1..c2c1c4d 100644 --- a/backend/app/api/api_v1/endpoints/reports.py +++ b/backend/app/api/api_v1/endpoints/reports.py @@ -5,7 +5,7 @@ import logging from app.services.dmarc_parser import DMARCParser from app.services.report_store import ReportStore -from app.utils.domain_validator import validate_domain +from app.utils.domain_validator import validate_domain, DomainValidationError logger = logging.getLogger(__name__) @@ -133,8 +133,8 @@ async def upload_report(file: UploadFile = File(...)): ) # Validate domain format (not DNS resolution to avoid external calls) - is_valid, error_msg = validate_domain(domain) - if not is_valid and "could not be resolved" not in error_msg.lower(): + is_valid, error_msg, error_code = validate_domain(domain, check_dns=False) + if not is_valid and error_code != DomainValidationError.DNS_RESOLUTION_FAILED: # Allow domains that fail DNS resolution but have valid format raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, diff --git a/backend/app/core/security.py b/backend/app/core/security.py index 84f3536..8e45f4b 100644 --- a/backend/app/core/security.py +++ b/backend/app/core/security.py @@ -19,9 +19,30 @@ pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") security_bearer = HTTPBearer(auto_error=False) api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False) -# In-memory API keys storage (for MVP - should be moved to database in production) +# In-memory API keys storage +# ⚠️ WARNING: This is a simple in-memory implementation suitable for: +# - Development and testing environments +# - Single-instance deployments +# - MVP/prototype applications +# +# ⚠️ NOT SUITABLE FOR PRODUCTION when: +# - Running multiple application instances (keys not shared) +# - Requiring key persistence across restarts +# - Needing key rotation and management +# +# For production, implement: +# - Database-backed key storage (with encryption at rest) +# - Redis or similar distributed cache for shared key storage +# - Integration with external secret management (AWS Secrets Manager, HashiCorp Vault, etc.) +# - Proper key rotation policies _api_keys = set() +logger.warning( + "Using in-memory API key storage. " + "Keys will be lost on restart. " + "Not suitable for production multi-instance deployments." +) + def generate_api_key() -> str: """ diff --git a/backend/app/main.py b/backend/app/main.py index 3f8548e..a1614ea 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -112,18 +112,26 @@ def create_app() -> FastAPI: """Initialize background tasks and security on application startup""" global background_task - # Generate and log initial API key for admin access + # Generate and provide admin API key api_key = generate_api_key() add_api_key(api_key) + + # Security: Log only last 8 characters for reference logger.warning( "=" * 80 + "\n" "IMPORTANT: Admin API Key Generated\n" - f"API Key: {api_key}\n" - "Save this key securely - it will not be shown again.\n" + f"API Key (last 8 chars): ...{api_key[-8:]}\n" + "Full key stored securely in memory.\n" + "For production, retrieve the key through secure configuration management.\n" "Use this key in the X-API-Key header for admin endpoints.\n" "=" * 80 ) + # In development, also log the full key for convenience + # This should be removed in production or controlled by environment variable + if os.getenv("ENVIRONMENT", "development") == "development": + logger.info(f"Development Mode - Full API Key: {api_key}") + # Check if IMAP credentials are configured if all([settings.IMAP_SERVER, settings.IMAP_USERNAME, settings.IMAP_PASSWORD]): logger.info("Starting IMAP polling background task") diff --git a/backend/app/middleware/security.py b/backend/app/middleware/security.py index e02c16d..c5bdcc7 100644 --- a/backend/app/middleware/security.py +++ b/backend/app/middleware/security.py @@ -51,10 +51,14 @@ class SecurityHeadersMiddleware(BaseHTTPMiddleware): # Content Security Policy (CSP) # Restricts sources of content that can be loaded + # TODO: Remove 'unsafe-inline' and 'unsafe-eval' and use nonces/hashes instead csp_directives = [ "default-src 'self'", - "script-src 'self' 'unsafe-inline' 'unsafe-eval'", # Allow inline scripts for now - "style-src 'self' 'unsafe-inline' https://fonts.googleapis.com", + # Note: 'unsafe-inline' and 'unsafe-eval' weaken XSS protection + # These should be removed and replaced with nonces or CSP hashes + # See: https://developer.mozilla.org/en-US/docs/Web/HTTP/CSP + "script-src 'self' 'unsafe-inline' 'unsafe-eval'", # TODO: Use nonces + "style-src 'self' 'unsafe-inline' https://fonts.googleapis.com", # TODO: Use nonces "font-src 'self' https://fonts.gstatic.com", "img-src 'self' data: https:", "connect-src 'self'", @@ -112,51 +116,3 @@ class SecurityHeadersMiddleware(BaseHTTPMiddleware): response.headers["Expires"] = "0" return response - - -class RateLimitMiddleware(BaseHTTPMiddleware): - """ - Simple rate limiting middleware to prevent abuse. - For production, consider using a more robust solution like slowapi or Redis-based rate limiting. - """ - - def __init__(self, app, requests_per_minute: int = 60): - """ - Initialize rate limiting middleware. - - Args: - app: FastAPI application instance - requests_per_minute: Maximum requests allowed per minute per IP - """ - super().__init__(app) - self.requests_per_minute = requests_per_minute - self.request_counts = {} # Simple in-memory store (not production-ready) - logger.warning( - "Using in-memory rate limiting. " - "For production, use Redis or similar distributed storage." - ) - - async def dispatch(self, request: Request, call_next: Callable) -> Response: - """ - Check rate limit and process request. - - Args: - request: Incoming HTTP request - call_next: Next middleware/handler in the chain - - Returns: - HTTP response or 429 Too Many Requests if rate limit exceeded - """ - # Get client IP - client_ip = request.client.host if request.client else "unknown" - - # For now, just log and pass through - # TODO: Implement actual rate limiting logic with time windows - # This is a placeholder for the actual implementation - - response = await call_next(request) - - # Add rate limit headers for transparency - response.headers["X-RateLimit-Limit"] = str(self.requests_per_minute) - - return response diff --git a/backend/app/tests/test_security.py b/backend/app/tests/test_security.py index 8062287..48b9c6f 100644 --- a/backend/app/tests/test_security.py +++ b/backend/app/tests/test_security.py @@ -70,7 +70,7 @@ class TestDomainValidation: ] for domain in valid_domains: - is_valid, error = validate_domain(domain, check_dns=False) + is_valid, error, error_code = validate_domain(domain, check_dns=False) assert is_valid, f"Domain {domain} should be valid: {error}" def test_invalid_domain_format(self): @@ -89,7 +89,7 @@ class TestDomainValidation: ] for domain in invalid_domains: - is_valid, error = validate_domain(domain, check_dns=False) + is_valid, error, error_code = validate_domain(domain, check_dns=False) assert not is_valid, f"Domain '{domain}' should be invalid" assert error is not None @@ -107,21 +107,21 @@ class TestDomainValidation: ] for domain in malicious_domains: - is_valid, error = validate_domain(domain, check_dns=False) + is_valid, error, error_code = validate_domain(domain, check_dns=False) assert not is_valid, f"Malicious domain '{domain}' should be rejected" def test_domain_length_limits(self): """Test domain length validation.""" # Max label is 63 characters - this should be caught by label length check long_label = "a" * 64 + ".example.com" - is_valid, error = validate_domain(long_label, check_dns=False) + is_valid, error, error_code = validate_domain(long_label, check_dns=False) assert not is_valid # Could be caught by format check or label length check assert error is not None # Max domain is 253 characters long_domain = "a" * 254 # 254 chars, no dot - is_valid, error = validate_domain(long_domain, check_dns=False) + is_valid, error, error_code = validate_domain(long_domain, check_dns=False) assert not is_valid assert "too long" in error.lower() or "invalid" in error.lower() diff --git a/backend/app/utils/domain_validator.py b/backend/app/utils/domain_validator.py index 22ec123..e8e3fe4 100644 --- a/backend/app/utils/domain_validator.py +++ b/backend/app/utils/domain_validator.py @@ -2,8 +2,19 @@ import re import socket from typing import Dict, Tuple, Union, Optional +# Error codes for structured error handling +class DomainValidationError: + """Domain validation error codes""" + EMPTY = "empty" + TOO_LONG = "too_long" + INVALID_FORMAT = "invalid_format" + INVALID_CHARACTERS = "invalid_characters" + LABEL_TOO_LONG = "label_too_long" + INVALID_LABEL = "invalid_label" + DNS_RESOLUTION_FAILED = "dns_resolution_failed" -def validate_domain(domain_name: str, check_dns: bool = True) -> Tuple[bool, Optional[str]]: + +def validate_domain(domain_name: str, check_dns: bool = True) -> Tuple[bool, Optional[str], Optional[str]]: """ Validates a domain name for format and optionally resolvability. @@ -12,25 +23,26 @@ def validate_domain(domain_name: str, check_dns: bool = True) -> Tuple[bool, Opt check_dns: Whether to perform DNS resolution check (default: True) Returns: - Tuple containing (is_valid, error_message) + Tuple containing (is_valid, error_message, error_code) - is_valid: Boolean indicating if domain is valid - error_message: String with error message if not valid, None if valid + - error_code: Error code constant for programmatic handling, None if valid """ # Security: Check for empty or None domain if not domain_name: - return False, "Domain name cannot be empty" + return False, "Domain name cannot be empty", DomainValidationError.EMPTY # Security: Check maximum length (DNS standard is 253 characters) if len(domain_name) > 253: - return False, "Domain name too long (max 253 characters)" + return False, "Domain name too long (max 253 characters)", DomainValidationError.TOO_LONG # Security: Check for whitespace if ' ' in domain_name or '\t' in domain_name or '\n' in domain_name: - return False, "Domain name cannot contain whitespace" + return False, "Domain name cannot contain whitespace", DomainValidationError.INVALID_CHARACTERS # Security: Check for suspicious characters if any(char in domain_name for char in ['<', '>', '"', "'", '\\', '|', ';', '&', '$', '`']): - return False, "Domain name contains invalid characters" + return False, "Domain name contains invalid characters", DomainValidationError.INVALID_CHARACTERS # Check domain format with regex # This regex allows domain names with alphanumeric characters, hyphens, @@ -38,27 +50,27 @@ def validate_domain(domain_name: str, check_dns: bool = True) -> Tuple[bool, Opt # Updated to be more strict and prevent potential attacks domain_pattern = r'^(?:[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?\.)+[a-z0-9][a-z0-9-]{0,61}[a-z0-9]$' if not re.match(domain_pattern, domain_name.lower()): - return False, "Invalid domain format" + return False, "Invalid domain format", DomainValidationError.INVALID_FORMAT # Security: Check each label length (max 63 characters per label) labels = domain_name.split('.') for label in labels: if len(label) > 63: - return False, f"Domain label too long: '{label}' (max 63 characters per label)" + return False, f"Domain label too long: '{label}' (max 63 characters per label)", DomainValidationError.LABEL_TOO_LONG if label.startswith('-') or label.endswith('-'): - return False, f"Domain label cannot start or end with hyphen: '{label}'" + return False, f"Domain label cannot start or end with hyphen: '{label}'", DomainValidationError.INVALID_LABEL # Check if domain exists by attempting to resolve DNS (optional) if check_dns: try: socket.gethostbyname(domain_name) - return True, None + return True, None, None except socket.gaierror: # We could consider this valid if we don't require DNS resolution, # but since DMARC requires valid DNS, we'll mark it as warning - return False, "Domain could not be resolved (DNS lookup failed)" + return False, "Domain could not be resolved (DNS lookup failed)", DomainValidationError.DNS_RESOLUTION_FAILED - return True, None + return True, None, None def validate_domain_config(domain_data: Dict) -> Dict[str, Union[bool, str]]: @@ -78,7 +90,7 @@ def validate_domain_config(domain_data: Dict) -> Dict[str, Union[bool, str]]: # Validate domain name if "name" in domain_data: # Don't check DNS for domain config validation - is_valid, error_msg = validate_domain(domain_data["name"], check_dns=False) + is_valid, error_msg, error_code = validate_domain(domain_data["name"], check_dns=False) if not is_valid: errors["name"] = error_msg else: @@ -88,9 +100,11 @@ def validate_domain_config(domain_data: Dict) -> Dict[str, Union[bool, str]]: if "description" in domain_data and domain_data["description"]: if len(domain_data["description"]) > 500: errors["description"] = "Description is too long (max 500 characters)" - # Security: Check for suspicious content in description - if any(char in domain_data["description"] for char in ['