"""Enterprise-Grade Security Middleware for Project Memo. Implements OWASP Top 10 protections: 1. Content-Security-Policy (CSP) headers 2. X-Frame-Options: DENY (clickjacking prevention) 3. X-Content-Type-Options: nosniff 4. Referrer-Policy 5. Permissions-Policy 6. Strict-Transport-Security (HSTS) — production only 7. Input sanitization via nh3 (Rust-based HTML/JS stripping) 8. Request body size limits 9. Security event logging """ from __future__ import annotations import logging import re import time from typing import Any import nh3 from fastapi import Request, Response from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint logger = logging.getLogger("memo.security") # ── Allowed HTML tags for product descriptions (very restrictive) ── ALLOWED_TAGS: set[str] = {"b", "i", "em", "strong", "ul", "ol", "li", "p", "br"} ALLOWED_ATTRIBUTES: dict[str, set[str]] = {} # ── Regex patterns for suspicious input ────────────────────────── _XSS_PATTERNS = [ re.compile(r" Response: response = await call_next(request) # ── OWASP-recommended headers ──────────────────────────── # X-Frame-Options is intentionally REMOVED for Hugging Face Spaces. # HF embeds the app in an iframe. CSP frame-ancestors controls # framing instead (set above to allow 'self' + huggingface.co). response.headers["X-Content-Type-Options"] = "nosniff" response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin" response.headers["Permissions-Policy"] = ( "camera=(), microphone=(), geolocation=(), " "payment=(self), usb=(), magnetometer=(), gyroscope=()" ) response.headers["Content-Security-Policy"] = CSP_DIRECTIVES response.headers["X-Permitted-Cross-Domain-Policies"] = "none" # COEP/COOP are intentionally relaxed to allow: # - External product images from AliExpress CDNs (no CORP headers) # - PayPal popup checkout flow (needs cross-origin window.opener) response.headers["Cross-Origin-Resource-Policy"] = "cross-origin" # ── HSTS — only in production (HTTPS) ──────────────────── # In development (DEBUG mode), HSTS would break HTTP access # response.headers["Strict-Transport-Security"] = "max-age=63072000; includeSubDomains; preload" return response # ── Input Sanitization Middleware ──────────────────────────────── class InputSanitizationMiddleware(BaseHTTPMiddleware): """Sanitize incoming request data to prevent XSS injection. This middleware: 1. Checks query parameters for XSS patterns 2. Validates Content-Length to prevent oversized payloads 3. Logs suspicious requests for audit trail """ MAX_BODY_SIZE = 2 * 1024 * 1024 # 2 MB async def dispatch(self, request: Request, call_next: RequestResponseEndpoint) -> Response: # ── Check Content-Length ────────────────────────────────── content_length = request.headers.get("content-length") if content_length and int(content_length) > self.MAX_BODY_SIZE: logger.warning( "Rejected oversized request: %d bytes from %s", int(content_length), request.client.host if request.client else "unknown", ) from fastapi.responses import JSONResponse return JSONResponse( status_code=413, content={"detail": "Request body too large"}, ) # ── Check query parameters for XSS ─────────────────────── for key, value in request.query_params.items(): if _contains_xss(value): client_ip = request.client.host if request.client else "unknown" logger.warning( "XSS attempt blocked in query param '%s' from %s [%s]", key, client_ip, value[:100], ) from fastapi.responses import JSONResponse return JSONResponse( status_code=400, content={"detail": "Invalid input detected"}, ) # ── Check path for traversal/injection ─────────────────── path = request.url.path if ".." in path or "%2e%2e" in path.lower(): client_ip = request.client.host if request.client else "unknown" logger.warning("Path traversal attempt from %s: %s", client_ip, path) from fastapi.responses import JSONResponse return JSONResponse(status_code=400, content={"detail": "Invalid path"}) response = await call_next(request) return response # ── Request Logging Middleware ──────────────────────────────────── class RequestLoggingMiddleware(BaseHTTPMiddleware): """Log all requests with timing for security audit trail.""" SKIP_PATHS = {"/health", "/static", "/favicon.ico"} async def dispatch(self, request: Request, call_next: RequestResponseEndpoint) -> Response: if any(request.url.path.startswith(p) for p in self.SKIP_PATHS): return await call_next(request) start = time.monotonic() client_ip = request.client.host if request.client else "unknown" response = await call_next(request) elapsed_ms = (time.monotonic() - start) * 1000 logger.info( "%s %s %d (%.0fms) [%s]", request.method, request.url.path, response.status_code, elapsed_ms, client_ip, ) return response # ── Utility Functions ──────────────────────────────────────────── def _contains_xss(value: str) -> bool: """Check if a string contains potential XSS patterns.""" for pattern in _XSS_PATTERNS: if pattern.search(value): return True return False def sanitize_html(text: str) -> str: """Strip all HTML tags except a safe whitelist using nh3 (Rust-based). Use this for user-generated content like product descriptions. """ if not text: return text return nh3.clean( text, tags=ALLOWED_TAGS, attributes=ALLOWED_ATTRIBUTES, strip_comments=True, ) def sanitize_text(text: str) -> str: """Strip ALL HTML tags from plain text fields (titles, names, etc.). This is the strictest sanitizer — use for fields that should never contain any HTML. """ if not text: return text return nh3.clean(text, tags=set(), attributes={}, strip_comments=True) # ── ID / URL Validation (shared utility) ──────────────────────── _ID_RE = re.compile(r"^[0-9a-zA-Z_-]{1,100}$") def safe_id(raw: str, max_length: int = 100) -> str: """Validate and return an ID string (alphanumeric, hyphens, underscores). Use for product IDs, post IDs, and other opaque identifiers. Returns empty string if the value doesn't match the expected pattern. """ val = str(raw).strip()[:max_length] return val if _ID_RE.match(val) else "" def safe_url(url: str) -> str: """Validate a URL is safe (starts with https://, no injection). Use for external URLs fetched from scraping or APIs. Uses regex validation instead of nh3.clean which would mangle URL characters like & into &. Returns empty string if the URL is malformed or suspicious. """ if not url: return "" url = url.strip() # Reject URLs containing angle brackets entirely (legitimate URLs never have them) if "<" in url or ">" in url: logger.warning("Blocked URL with embedded tags: %s", url[:80]) return "" # Quick scheme check if not url.startswith("https://") and not url.startswith("http://"): logger.debug("URL validation dropped: %s", url[:100]) return "" cleaned = url # Block javascript: and data: URIs if re.match(r"^(javascript|data|vbscript):", cleaned, re.IGNORECASE): logger.warning("Blocked dangerous URL scheme: %s", cleaned[:80]) return "" # Optionally validate against allowed base domains if allowed_bases and not any(cleaned.startswith(base) for base in allowed_bases): logger.debug("URL not in allowed bases: %s", cleaned[:80]) return "" return cleaned[:2000] def sanitize_dict(data: dict[str, Any], text_fields: set[str] | None = None) -> dict[str, Any]: """Recursively sanitize string values in a dict. text_fields: set of field names that should use strict text sanitization. All other string fields get HTML sanitization with safe tags. """ text_fields = text_fields or set() sanitized: dict[str, Any] = {} for key, value in data.items(): if isinstance(value, str): if key in text_fields: sanitized[key] = sanitize_text(value) else: sanitized[key] = sanitize_html(value) elif isinstance(value, dict): sanitized[key] = sanitize_dict(value, text_fields) elif isinstance(value, list): sanitized[key] = [ sanitize_text(item) if isinstance(item, str) else item for item in value ] else: sanitized[key] = value return sanitized