diff --git a/.env.example b/.env.example index e43958c..0e4ec45 100644 --- a/.env.example +++ b/.env.example @@ -5,18 +5,32 @@ # Application Settings PROJECT_NAME="DMARQ" + +# SECURITY: Generate a secure random secret key +# Use: openssl rand -hex 32 +# NEVER use the default value in production! SECRET_KEY="CHANGE_THIS_TO_A_RANDOM_SECRET_IN_PRODUCTION" +# Environment (development/production) +# Affects HSTS and other security settings +ENVIRONMENT="development" + # Database DATABASE_URL="sqlite:///./dmarq.db" +# For production, use PostgreSQL: +# DATABASE_URL="postgresql://user:password@localhost/dmarq" # IMAP Settings for DMARC Report Retrieval IMAP_SERVER="mail.example.com" # Required for IMAP polling IMAP_PORT=993 # Default for SSL IMAP_USERNAME="dmarc@example.com" IMAP_PASSWORD="your_imap_password" # Consider using a secrets manager in production + # CORS Origins (comma separated) +# SECURITY: Be specific - avoid wildcards in production BACKEND_CORS_ORIGINS="http://localhost:3000,http://localhost:5173" +# For production: +# BACKEND_CORS_ORIGINS="https://yourdomain.com" # Admin User (first-time setup) FIRST_SUPERUSER="admin@example.com" diff --git a/backend/app/api/api_v1/endpoints/imap.py b/backend/app/api/api_v1/endpoints/imap.py index 16527db..8d23afe 100644 --- a/backend/app/api/api_v1/endpoints/imap.py +++ b/backend/app/api/api_v1/endpoints/imap.py @@ -1,13 +1,17 @@ from fastapi import APIRouter, Depends, HTTPException, BackgroundTasks from typing import Dict, Any from datetime import datetime +import logging from app.services.imap_client import IMAPClient +from app.core.security import require_admin_auth router = APIRouter() +logger = logging.getLogger(__name__) @router.post("/test-connection") async def test_imap_connection( + auth: dict = Depends(require_admin_auth), server: str = None, port: int = 993, username: str = None, @@ -16,7 +20,18 @@ async def test_imap_connection( ) -> Dict[str, Any]: """ Test connection to an IMAP server and gather mailbox statistics + + Security: Requires authentication (X-API-Key or Bearer token) + Note: Credentials should be passed in request body, not query params """ + # Security: Don't accept credentials in query parameters (they get logged) + if any([server, username, password]): + logger.warning("IMAP credentials passed as query parameters - this is insecure") + raise HTTPException( + status_code=400, + detail="Credentials should be passed in request body, not query parameters" + ) + imap_client = IMAPClient( server=server, port=port, @@ -40,12 +55,22 @@ async def test_imap_connection( @router.post("/fetch-reports") async def fetch_imap_reports( background_tasks: BackgroundTasks, + auth: dict = Depends(require_admin_auth), days: int = 7, delete_emails: bool = False ) -> Dict[str, Any]: """ Fetch DMARC reports from the configured IMAP mailbox + + Security: Requires authentication (X-API-Key or Bearer token) """ + # Security: Validate parameters + if days < 1 or days > 365: + raise HTTPException( + status_code=400, + detail="Days parameter must be between 1 and 365" + ) + imap_client = IMAPClient(delete_emails=delete_emails) # Run in background if it might take a while @@ -58,64 +83,41 @@ async def fetch_imap_reports( } # Otherwise run immediately - results = imap_client.fetch_reports(days=days) - - return { - "success": results["success"], - "processed_emails": results["processed"], - "reports_found": results["reports_found"], - "new_domains": results["new_domains"], - "errors": results["errors"] if "errors" in results and results["errors"] else None, - "timestamp": datetime.now().isoformat() - } + try: + results = imap_client.fetch_reports(days=days) + + return { + "success": results["success"], + "processed_emails": results["processed"], + "reports_found": results["reports_found"], + "new_domains": results["new_domains"], + "errors": results["errors"] if "errors" in results and results["errors"] else None, + "timestamp": datetime.now().isoformat() + } + except Exception as e: + logger.error(f"Error fetching IMAP reports: {str(e)}") + raise HTTPException( + status_code=500, + detail="Failed to fetch reports. Check server logs for details." + ) @router.get("/status") -async def get_imap_status() -> Dict[str, Any]: +async def get_imap_status(auth: dict = Depends(require_admin_auth)) -> Dict[str, Any]: """ Get the current status of IMAP polling background processes + + Security: Requires authentication (X-API-Key or Bearer token) """ # In a real implementation this would check a persistent store # or a global variable tracking the status of background tasks - # For now, returning mock data as this is MVP + # For now, returning simplified status - # Get the last check time if available - last_check_time = None - try: - # In a production app, this would be stored in database - # For MVP, using a simple file-based approach - import os - status_file = os.path.join(os.path.dirname(__file__), "../../../../../tmp/imap_last_check.txt") - if os.path.exists(status_file): - with open(status_file, "r") as f: - last_check_time = f.read().strip() - except: - pass - - # If status file doesn't exist, create the directory - try: - os.makedirs(os.path.dirname(os.path.join(os.path.dirname(__file__), "../../../../../tmp")), exist_ok=True) - except: - pass - - # For demonstration purposes, update the last check time to now - # In a real app, this would be updated by the background process - try: - with open(os.path.join(os.path.dirname(__file__), "../../../../../tmp/imap_last_check.txt"), "w") as f: - now = datetime.now().isoformat() - f.write(now) - # If there was no previous check time, set it to now - if not last_check_time: - last_check_time = now - except: - pass - - # Return the status return { "is_running": True, # In a real app, check if the background task is running - "last_check": last_check_time, - "next_check": None, # In production, this would be calculated based on polling interval - "messages_processed": 0, # In production, this would track actual messages processed - "reports_found": 0, # In production, this would track reports found + "last_check": None, # In production, track actual last check time + "next_check": None, # In production, calculate based on polling interval + "messages_processed": 0, # In production, track actual messages processed + "reports_found": 0, # In production, track reports found "timestamp": datetime.now().isoformat() } \ No newline at end of file diff --git a/backend/app/api/api_v1/endpoints/reports.py b/backend/app/api/api_v1/endpoints/reports.py index b254333..43456c1 100644 --- a/backend/app/api/api_v1/endpoints/reports.py +++ b/backend/app/api/api_v1/endpoints/reports.py @@ -1,12 +1,38 @@ from typing import Dict, List, Any from fastapi import APIRouter, File, HTTPException, UploadFile, status from pydantic import BaseModel +import logging from app.services.dmarc_parser import DMARCParser from app.services.report_store import ReportStore +from app.utils.domain_validator import validate_domain + +logger = logging.getLogger(__name__) + +# Try to import python-magic for MIME type detection +try: + import magic + HAS_MAGIC = True +except ImportError: + HAS_MAGIC = False + logger.warning("python-magic not installed. MIME type validation will be skipped.") router = APIRouter() +# Security: Allowed MIME types for DMARC report uploads +ALLOWED_MIME_TYPES = { + 'text/xml', + 'application/xml', + 'application/zip', + 'application/x-zip-compressed', + 'application/gzip', + 'application/x-gzip', + 'application/octet-stream' # Sometimes zip/gzip are detected as this +} + +# Security: Allowed file extensions +ALLOWED_EXTENSIONS = {'.xml', '.zip', '.gz', '.gzip'} + class UploadResponse(BaseModel): """Response model for report upload""" success: bool @@ -45,21 +71,80 @@ class PaginatedReportResponse(BaseModel): async def upload_report(file: UploadFile = File(...)): """ Upload and process a DMARC aggregate report file (XML, ZIP, or GZIP) + + Security: + - File type validation (extension and MIME type) + - File size limits enforced in parser + - Zip bomb protection + - Sanitized error messages """ try: + # Security: Validate filename is provided + if not file.filename: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="Filename is required" + ) + + # Security: Validate file extension + file_ext = '.' + file.filename.rsplit('.', 1)[-1].lower() if '.' in file.filename else '' + if file_ext not in ALLOWED_EXTENSIONS: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=f"Invalid file type. Allowed types: {', '.join(ALLOWED_EXTENSIONS)}" + ) + # Read the file content file_content = await file.read() - filename = file.filename + + # Security: Validate file is not empty + if len(file_content) == 0: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="File is empty" + ) + + # Security: Validate MIME type using python-magic (if available) + if HAS_MAGIC: + try: + mime_type = magic.from_buffer(file_content, mime=True) + if mime_type not in ALLOWED_MIME_TYPES: + logger.warning(f"Rejected file with MIME type: {mime_type}") + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=f"Invalid file type. File must be XML, ZIP, or GZIP format." + ) + except Exception as e: + # If magic fails, log but continue (fallback to extension check) + logger.warning(f"MIME type detection failed: {str(e)}") + else: + logger.debug("MIME type validation skipped (python-magic not available)") # Parse the report parser = DMARCParser() - report = parser.parse_file(file_content, filename) + report = parser.parse_file(file_content, file.filename) + + # Security: Validate domain from report + domain = report.get("domain", "") + if not domain: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="Report does not contain a valid domain" + ) + + # 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(): + # Allow domains that fail DNS resolution but have valid format + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=f"Invalid domain in report: {error_msg}" + ) # Store the report store = ReportStore.get_instance() store.add_report(report) - domain = report.get("domain", "unknown") processed_records = report.get("summary", {}).get("total_count", 0) return UploadResponse( @@ -69,10 +154,36 @@ async def upload_report(file: UploadFile = File(...)): processed_records=processed_records ) + except HTTPException: + # Re-raise HTTP exceptions as-is + raise + except ValueError as e: + # Security: Sanitize error messages from parser + error_message = str(e) + # Log full error for debugging + logger.error(f"ValueError processing report {file.filename}: {error_message}") + # Return sanitized message + if "too large" in error_message.lower(): + raise HTTPException( + status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE, + detail="File too large" + ) + elif "zip bomb" in error_message.lower(): + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="Invalid archive file" + ) + else: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="Invalid report format" + ) except Exception as e: + # Security: Don't expose internal errors to client + logger.error(f"Unexpected error processing report {file.filename}: {str(e)}") raise HTTPException( - status_code=status.HTTP_400_BAD_REQUEST, - detail=f"Error processing report: {str(e)}" + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail="Error processing report. Please contact support if this persists." ) @router.get("/domains", response_model=List[str]) diff --git a/backend/app/core/config.py b/backend/app/core/config.py index 4e0f3fd..2f848de 100644 --- a/backend/app/core/config.py +++ b/backend/app/core/config.py @@ -1,5 +1,7 @@ from functools import lru_cache from typing import Optional, List, Union +import secrets +import logging # Try to import from pydantic_settings first (newer versions) try: @@ -9,6 +11,8 @@ except ImportError: # Fall back to older pydantic version from pydantic import BaseSettings, EmailStr, validator +logger = logging.getLogger(__name__) + class Settings(BaseSettings): """Application settings""" @@ -21,7 +25,7 @@ class Settings(BaseSettings): DATABASE_URL: str = "sqlite:///./dmarq.db" # JWT Authentication - SECRET_KEY: str = "CHANGE_THIS_TO_A_RANDOM_SECRET_IN_PRODUCTION" + SECRET_KEY: Optional[str] = None ALGORITHM: str = "HS256" ACCESS_TOKEN_EXPIRE_MINUTES: int = 60 # 1 hour @@ -42,6 +46,32 @@ class Settings(BaseSettings): CLOUDFLARE_API_TOKEN: Optional[str] = None CLOUDFLARE_ZONE_ID: Optional[str] = None + @validator("SECRET_KEY", pre=True, always=True) + def validate_secret_key(cls, v: Optional[str]) -> str: + """Validate and generate SECRET_KEY if not provided.""" + # Default insecure key that should never be used + DEFAULT_INSECURE_KEY = "CHANGE_THIS_TO_A_RANDOM_SECRET_IN_PRODUCTION" + + if v is None or v == "" or v == DEFAULT_INSECURE_KEY: + # Generate a secure random key + generated_key = secrets.token_hex(32) + logger.warning( + "SECRET_KEY not configured or using default value! " + "Generated a random key for this session. " + "For production, set SECRET_KEY in your .env file using: " + f"openssl rand -hex 32" + ) + return generated_key + + # Check if key is too short + if len(v) < 32: + logger.warning( + f"SECRET_KEY is too short ({len(v)} characters). " + "Recommended minimum is 32 characters for security." + ) + + return v + @validator("BACKEND_CORS_ORIGINS", pre=True) def assemble_cors_origins(cls, v: Union[str, List[str]]) -> List[str]: if isinstance(v, str) and not v.startswith("["): diff --git a/backend/app/core/security.py b/backend/app/core/security.py index 74cc7ae..84f3536 100644 --- a/backend/app/core/security.py +++ b/backend/app/core/security.py @@ -1,15 +1,179 @@ from datetime import datetime, timedelta -from typing import Any, Union +from typing import Any, Union, Optional +import secrets +import logging -from jose import jwt +from fastapi import HTTPException, Security, status +from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials, APIKeyHeader +from jose import jwt, JWTError from passlib.context import CryptContext from app.core.config import get_settings settings = get_settings() +logger = logging.getLogger(__name__) pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") +# Security schemes for authentication +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) +_api_keys = set() + + +def generate_api_key() -> str: + """ + Generate a secure random API key. + + Returns: + A 32-character hexadecimal API key + """ + return secrets.token_hex(32) + + +def add_api_key(api_key: str) -> bool: + """ + Add an API key to the valid keys set. + + Args: + api_key: The API key to add + + Returns: + True if key was added, False if it already existed + """ + if api_key in _api_keys: + return False + _api_keys.add(api_key) + logger.info(f"API key added (ends with: ...{api_key[-8:]})") + return True + + +def verify_api_key(api_key: str) -> bool: + """ + Verify an API key is valid. + + Args: + api_key: The API key to verify + + Returns: + True if key is valid, False otherwise + """ + return api_key in _api_keys + + +async def get_api_key( + api_key_header: Optional[str] = Security(api_key_header) +) -> str: + """ + Dependency to verify API key authentication. + + Args: + api_key_header: API key from X-API-Key header + + Returns: + The validated API key + + Raises: + HTTPException: If API key is missing or invalid + """ + if not api_key_header: + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="Missing API key", + headers={"WWW-Authenticate": "ApiKey"}, + ) + + if not verify_api_key(api_key_header): + logger.warning(f"Invalid API key attempt: ...{api_key_header[-8:] if len(api_key_header) >= 8 else 'invalid'}") + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="Invalid API key", + headers={"WWW-Authenticate": "ApiKey"}, + ) + + return api_key_header + + +async def verify_token( + credentials: Optional[HTTPAuthorizationCredentials] = Security(security_bearer) +) -> dict: + """ + Dependency to verify JWT token authentication. + + Args: + credentials: Bearer token from Authorization header + + Returns: + Decoded token payload + + Raises: + HTTPException: If token is missing or invalid + """ + if not credentials: + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="Missing authentication token", + headers={"WWW-Authenticate": "Bearer"}, + ) + + token = credentials.credentials + + try: + payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM]) + return payload + except JWTError as e: + logger.warning(f"Invalid JWT token: {str(e)}") + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="Invalid authentication token", + headers={"WWW-Authenticate": "Bearer"}, + ) + + +async def require_admin_auth( + api_key: Optional[str] = Security(api_key_header), + bearer: Optional[HTTPAuthorizationCredentials] = Security(security_bearer) +) -> dict: + """ + Dependency to require either API key or JWT token authentication for admin endpoints. + + Checks API key first, then falls back to JWT token. + + Args: + api_key: Optional API key from X-API-Key header + bearer: Optional JWT token from Authorization header + + Returns: + Authentication context (api_key or token payload) + + Raises: + HTTPException: If no valid authentication is provided + """ + # Try API key first + if api_key and verify_api_key(api_key): + return {"auth_type": "api_key", "api_key": api_key} + + # Try JWT token + if bearer: + try: + payload = jwt.decode( + bearer.credentials, + settings.SECRET_KEY, + algorithms=[settings.ALGORITHM] + ) + return {"auth_type": "jwt", "payload": payload} + except JWTError as e: + logger.warning(f"Invalid JWT token: {str(e)}") + + # No valid authentication provided + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="Authentication required. Provide either X-API-Key header or Bearer token.", + headers={"WWW-Authenticate": "ApiKey, Bearer"}, + ) + def create_access_token( subject: Union[str, Any], expires_delta: timedelta = None diff --git a/backend/app/main.py b/backend/app/main.py index f66d248..3f8548e 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -1,4 +1,4 @@ -from fastapi import FastAPI, Request, BackgroundTasks +from fastapi import FastAPI, Request, BackgroundTasks, Depends from fastapi.middleware.cors import CORSMiddleware from fastapi.staticfiles import StaticFiles from fastapi.templating import Jinja2Templates @@ -10,6 +10,8 @@ from datetime import datetime from app.api.api_v1.api import api_router from app.core.config import get_settings +from app.core.security import require_admin_auth, generate_api_key, add_api_key +from app.middleware.security import SecurityHeadersMiddleware from app.services.imap_client import IMAPClient from app.services.report_store import ReportStore @@ -70,15 +72,32 @@ def create_app() -> FastAPI: openapi_url=f"{settings.API_V1_STR}/openapi.json", version="0.1.0", ) + + # Add security headers middleware + # Determine environment from settings or environment variable + environment = os.getenv("ENVIRONMENT", "development") + app.add_middleware(SecurityHeadersMiddleware, environment=environment) - # Set all CORS enabled origins + # Improved CORS configuration - restrict to specific methods and headers if settings.BACKEND_CORS_ORIGINS: app.add_middleware( CORSMiddleware, allow_origins=[str(origin) for origin in settings.BACKEND_CORS_ORIGINS], allow_credentials=True, - allow_methods=["*"], - allow_headers=["*"], + # Security: Restrict to only necessary HTTP methods + allow_methods=["GET", "POST", "PUT", "DELETE", "OPTIONS"], + # Security: Specify allowed headers instead of wildcard + allow_headers=[ + "Content-Type", + "Authorization", + "X-API-Key", + "Accept", + "Origin", + "X-Requested-With" + ], + # Security: Limit exposed headers + expose_headers=["Content-Length", "X-RateLimit-Limit"], + max_age=600, # Cache preflight requests for 10 minutes ) # Include API router @@ -90,9 +109,21 @@ def create_app() -> FastAPI: # Set up event handlers for startup and shutdown @app.on_event("startup") async def startup_event(): - """Initialize background tasks on application startup""" + """Initialize background tasks and security on application startup""" global background_task + # Generate and log initial API key for admin access + api_key = generate_api_key() + add_api_key(api_key) + 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" + "Use this key in the X-API-Key header for admin endpoints.\n" + "=" * 80 + ) + # Check if IMAP credentials are configured if all([settings.IMAP_SERVER, settings.IMAP_USERNAME, settings.IMAP_PASSWORD]): logger.info("Starting IMAP polling background task") @@ -193,8 +224,15 @@ async def upload_page(request: Request): # API endpoint to manually trigger IMAP polling @app.post("/api/v1/admin/trigger-poll") -async def trigger_imap_poll(background_tasks: BackgroundTasks): - """Manually trigger IMAP polling (admin only)""" +async def trigger_imap_poll( + background_tasks: BackgroundTasks, + auth: dict = Depends(require_admin_auth) +): + """ + Manually trigger IMAP polling (admin only - requires authentication) + + Security: Requires either X-API-Key header or Bearer token + """ global last_check_time try: @@ -210,23 +248,29 @@ async def trigger_imap_poll(background_tasks: BackgroundTasks): "timestamp": last_check_time.isoformat(), "processed": results["processed"], "reports_found": results["reports_found"], - "new_domains": results["new_domains"] + "new_domains": results["new_domains"], + "authenticated_by": auth.get("auth_type") } except Exception as e: logger.error(f"Error triggering IMAP poll: {str(e)}") return { "success": False, - "error": str(e) + "error": "Failed to trigger IMAP poll. Check server logs for details." } # API endpoint to check status of IMAP polling @app.get("/api/v1/admin/poll-status") -async def get_poll_status(): - """Get the status of IMAP polling""" +async def get_poll_status(auth: dict = Depends(require_admin_auth)): + """ + Get the status of IMAP polling (admin only - requires authentication) + + Security: Requires either X-API-Key header or Bearer token + """ global last_check_time return { "is_running": background_task is not None and not background_task.done(), - "last_check": last_check_time.isoformat() if last_check_time else None + "last_check": last_check_time.isoformat() if last_check_time else None, + "authenticated_by": auth.get("auth_type") } \ No newline at end of file diff --git a/backend/app/middleware/__init__.py b/backend/app/middleware/__init__.py new file mode 100644 index 0000000..fe0ef2f --- /dev/null +++ b/backend/app/middleware/__init__.py @@ -0,0 +1 @@ +"""Middleware package for DMARQ application.""" diff --git a/backend/app/middleware/security.py b/backend/app/middleware/security.py new file mode 100644 index 0000000..e02c16d --- /dev/null +++ b/backend/app/middleware/security.py @@ -0,0 +1,162 @@ +""" +Security headers middleware for DMARQ application. + +Implements various security headers to protect against common web vulnerabilities: +- Content Security Policy (CSP) +- X-Frame-Options +- X-Content-Type-Options +- Strict-Transport-Security (HSTS) +- X-XSS-Protection +- Referrer-Policy +- Permissions-Policy +""" + +from fastapi import Request +from starlette.middleware.base import BaseHTTPMiddleware +from starlette.responses import Response +from typing import Callable +import logging + +logger = logging.getLogger(__name__) + + +class SecurityHeadersMiddleware(BaseHTTPMiddleware): + """ + Middleware to add security headers to all HTTP responses. + """ + + def __init__(self, app, environment: str = "development"): + """ + Initialize security headers middleware. + + Args: + app: FastAPI application instance + environment: Application environment (development/production) + """ + super().__init__(app) + self.environment = environment + + async def dispatch(self, request: Request, call_next: Callable) -> Response: + """ + Process the request and add security headers to the response. + + Args: + request: Incoming HTTP request + call_next: Next middleware/handler in the chain + + Returns: + HTTP response with security headers added + """ + response = await call_next(request) + + # Content Security Policy (CSP) + # Restricts sources of content that can be loaded + 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", + "font-src 'self' https://fonts.gstatic.com", + "img-src 'self' data: https:", + "connect-src 'self'", + "frame-ancestors 'none'", # Prevent framing + "base-uri 'self'", + "form-action 'self'" + ] + response.headers["Content-Security-Policy"] = "; ".join(csp_directives) + + # X-Frame-Options: Prevent clickjacking attacks + # 'DENY' prevents the page from being displayed in a frame + response.headers["X-Frame-Options"] = "DENY" + + # X-Content-Type-Options: Prevent MIME type sniffing + # Forces browsers to respect the declared Content-Type + response.headers["X-Content-Type-Options"] = "nosniff" + + # X-XSS-Protection: Enable browser XSS protection + # Note: Modern browsers rely more on CSP, but this provides defense-in-depth + response.headers["X-XSS-Protection"] = "1; mode=block" + + # Referrer-Policy: Control referrer information + # 'strict-origin-when-cross-origin' provides good balance of privacy and functionality + response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin" + + # Permissions-Policy: Control browser features + # Disable features that aren't needed + permissions_policies = [ + "accelerometer=()", + "camera=()", + "geolocation=()", + "gyroscope=()", + "magnetometer=()", + "microphone=()", + "payment=()", + "usb=()" + ] + response.headers["Permissions-Policy"] = ", ".join(permissions_policies) + + # Strict-Transport-Security (HSTS): Force HTTPS + # Only enable in production with HTTPS + if self.environment == "production": + # max-age=31536000 = 1 year + # includeSubDomains applies to all subdomains + # preload allows inclusion in browser HSTS preload lists + response.headers["Strict-Transport-Security"] = ( + "max-age=31536000; includeSubDomains; preload" + ) + + # Cache-Control for sensitive pages + # Prevent caching of potentially sensitive data + if request.url.path.startswith("/api/"): + response.headers["Cache-Control"] = "no-store, no-cache, must-revalidate, private" + response.headers["Pragma"] = "no-cache" + 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/services/dmarc_parser.py b/backend/app/services/dmarc_parser.py index 4322d6e..11aa68e 100644 --- a/backend/app/services/dmarc_parser.py +++ b/backend/app/services/dmarc_parser.py @@ -4,13 +4,18 @@ import gzip import io from datetime import datetime from typing import Any, Dict, List, Optional, Union -import xml.etree.ElementTree as ET +import defusedxml.ElementTree as ET import logging # Set up logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) +# Security constants for file upload protection +MAX_FILE_SIZE = 10 * 1024 * 1024 # 10 MB +MAX_UNCOMPRESSED_SIZE = 100 * 1024 * 1024 # 100 MB for zip bomb protection +MAX_FILES_IN_ARCHIVE = 10 # Maximum number of files in a zip archive + class DMARCParser: """ Parser for DMARC Aggregate Reports (XML format) @@ -27,11 +32,26 @@ class DMARCParser: Returns: Dict containing the parsed report data + + Raises: + ValueError: If file is invalid, too large, or potentially malicious """ + # Security: Check file size + if len(file_content) > MAX_FILE_SIZE: + raise ValueError(f"File too large. Maximum size is {MAX_FILE_SIZE / (1024*1024):.1f} MB") + # Determine file type and extract XML content xml_content = DMARCParser._extract_xml_content(file_content, filename) if not xml_content: raise ValueError("Could not extract XML content from file") + + # Security: Check uncompressed XML size + if len(xml_content) > MAX_UNCOMPRESSED_SIZE: + raise ValueError( + f"Uncompressed content too large ({len(xml_content) / (1024*1024):.1f} MB). " + f"Maximum is {MAX_UNCOMPRESSED_SIZE / (1024*1024):.1f} MB. " + "Possible zip bomb attack detected." + ) # Parse the XML content return DMARCParser._parse_xml(xml_content) @@ -40,14 +60,39 @@ class DMARCParser: def _extract_xml_content(file_content: bytes, filename: str) -> Optional[bytes]: """ Extract XML content from various file formats (ZIP, GZIP, or plain XML) + + Raises: + ValueError: If archive contains too many files or is potentially malicious """ # Try to handle as ZIP file if filename.lower().endswith('.zip'): try: with zipfile.ZipFile(io.BytesIO(file_content)) as z: + # Security: Check number of files in archive + file_list = z.infolist() + if len(file_list) > MAX_FILES_IN_ARCHIVE: + raise ValueError( + f"ZIP archive contains too many files ({len(file_list)}). " + f"Maximum is {MAX_FILES_IN_ARCHIVE}." + ) + + # Security: Check for zip bomb by examining compression ratios + total_uncompressed = sum(f.file_size for f in file_list) + if total_uncompressed > MAX_UNCOMPRESSED_SIZE: + raise ValueError( + f"ZIP archive uncompressed size too large ({total_uncompressed / (1024*1024):.1f} MB). " + f"Maximum is {MAX_UNCOMPRESSED_SIZE / (1024*1024):.1f} MB. " + "Possible zip bomb attack detected." + ) + # Find the first XML file in the archive - for file_info in z.infolist(): + for file_info in file_list: if file_info.filename.lower().endswith('.xml'): + # Security: Double-check individual file size + if file_info.file_size > MAX_UNCOMPRESSED_SIZE: + raise ValueError( + f"XML file in archive too large ({file_info.file_size / (1024*1024):.1f} MB)" + ) return z.read(file_info.filename) except zipfile.BadZipFile: pass diff --git a/backend/app/utils/domain_validator.py b/backend/app/utils/domain_validator.py index 37167dc..22ec123 100644 --- a/backend/app/utils/domain_validator.py +++ b/backend/app/utils/domain_validator.py @@ -3,37 +3,62 @@ import socket from typing import Dict, Tuple, Union, Optional -def validate_domain(domain_name: str) -> Tuple[bool, Optional[str]]: +def validate_domain(domain_name: str, check_dns: bool = True) -> Tuple[bool, Optional[str]]: """ - Validates a domain name for format and resolvability. + Validates a domain name for format and optionally resolvability. Args: domain_name: The domain name to validate + check_dns: Whether to perform DNS resolution check (default: True) Returns: Tuple containing (is_valid, error_message) - is_valid: Boolean indicating if domain is valid - error_message: String with error message if not valid, None if valid """ - # Check for empty domain + # Security: Check for empty or None domain if not domain_name: return False, "Domain name cannot be empty" + # Security: Check maximum length (DNS standard is 253 characters) + if len(domain_name) > 253: + return False, "Domain name too long (max 253 characters)" + + # 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" + + # Security: Check for suspicious characters + if any(char in domain_name for char in ['<', '>', '"', "'", '\\', '|', ';', '&', '$', '`']): + return False, "Domain name contains invalid characters" + # Check domain format with regex # This regex allows domain names with alphanumeric characters, hyphens, # and periods as separators. It enforces proper domain structure. - domain_pattern = r'^(?:[a-zA-Z0-9](?:[a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?\.)+[a-zA-Z0-9][a-zA-Z0-9-]{0,61}[a-zA-Z0-9]$' - if not re.match(domain_pattern, domain_name): + # 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" - # Check if domain exists by attempting to resolve DNS - try: - socket.gethostbyname(domain_name) - return True, 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)" + # 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)" + if label.startswith('-') or label.endswith('-'): + return False, f"Domain label cannot start or end with hyphen: '{label}'" + + # Check if domain exists by attempting to resolve DNS (optional) + if check_dns: + try: + socket.gethostbyname(domain_name) + return True, 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 True, None def validate_domain_config(domain_data: Dict) -> Dict[str, Union[bool, str]]: @@ -52,7 +77,8 @@ def validate_domain_config(domain_data: Dict) -> Dict[str, Union[bool, str]]: # Validate domain name if "name" in domain_data: - is_valid, error_msg = validate_domain(domain_data["name"]) + # Don't check DNS for domain config validation + is_valid, error_msg = validate_domain(domain_data["name"], check_dns=False) if not is_valid: errors["name"] = error_msg else: @@ -60,8 +86,11 @@ def validate_domain_config(domain_data: Dict) -> Dict[str, Union[bool, str]]: # Validate description (optional but with max length) if "description" in domain_data and domain_data["description"]: - if len(domain_data["description"]) > 255: - errors["description"] = "Description is too long (max 255 characters)" + 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 ['