Spaces:

yusef75
/

smart-agent-store

Sleeping

smart-agent-store / middleware /security.py

Critical fix: remove X-Frame-Options DENY and allow huggingface.co in CSP frame-ancestors so HF Spaces iframe can load the app

c62aa29 4 months ago

Raw History Blame Contribute Delete

11.5 kB

"""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"<script", re.IGNORECASE),
    re.compile(r"javascript:", re.IGNORECASE),
    re.compile(r"on\w+\s*=", re.IGNORECASE),  # onclick=, onerror=, etc.
    re.compile(r"data:\s*text/html", re.IGNORECASE),
    re.compile(r"vbscript:", re.IGNORECASE),
    re.compile(r"expression\s*\(", re.IGNORECASE),
]

# ── Content-Security-Policy ──────────────────────────────────────

CSP_DIRECTIVES = (
    "default-src 'self'; "
    "script-src 'self' 'unsafe-inline' https://cdn.tailwindcss.com https://cdnjs.cloudflare.com https://www.paypal.com https://www.sandbox.paypal.com; "
    "style-src 'self' 'unsafe-inline' https://fonts.googleapis.com https://cdnjs.cloudflare.com https://cdn.tailwindcss.com; "
    "font-src 'self' https://fonts.gstatic.com https://cdnjs.cloudflare.com; "
    "img-src 'self' data: https: blob:; "
    "connect-src 'self' https://api-m.paypal.com https://api-m.sandbox.paypal.com; "
    "frame-src https://www.paypal.com https://www.sandbox.paypal.com; "
    "object-src 'none'; "
    "base-uri 'self'; "
    "form-action 'self'; "
    "frame-ancestors 'self' https://huggingface.co; "
    "upgrade-insecure-requests"
)


# ── Security Headers Middleware ───────────────────────────────────

class SecurityHeadersMiddleware(BaseHTTPMiddleware):
    """Inject comprehensive security headers into every response."""

    async def dispatch(self, request: Request, call_next: RequestResponseEndpoint) -> 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 &amp;.
    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

	"""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"<script", re.IGNORECASE),
	re.compile(r"javascript:", re.IGNORECASE),
	re.compile(r"on\w+\s*=", re.IGNORECASE), # onclick=, onerror=, etc.
	re.compile(r"data:\s*text/html", re.IGNORECASE),
	re.compile(r"vbscript:", re.IGNORECASE),
	re.compile(r"expression\s*\(", re.IGNORECASE),
	]

	# ── Content-Security-Policy ──────────────────────────────────────

	CSP_DIRECTIVES = (
	"default-src 'self'; "
	"script-src 'self' 'unsafe-inline' https://cdn.tailwindcss.com https://cdnjs.cloudflare.com https://www.paypal.com https://www.sandbox.paypal.com; "
	"style-src 'self' 'unsafe-inline' https://fonts.googleapis.com https://cdnjs.cloudflare.com https://cdn.tailwindcss.com; "
	"font-src 'self' https://fonts.gstatic.com https://cdnjs.cloudflare.com; "
	"img-src 'self' data: https: blob:; "
	"connect-src 'self' https://api-m.paypal.com https://api-m.sandbox.paypal.com; "
	"frame-src https://www.paypal.com https://www.sandbox.paypal.com; "
	"object-src 'none'; "
	"base-uri 'self'; "
	"form-action 'self'; "
	"frame-ancestors 'self' https://huggingface.co; "
	"upgrade-insecure-requests"
	)


	# ── Security Headers Middleware ───────────────────────────────────

	class SecurityHeadersMiddleware(BaseHTTPMiddleware):
	"""Inject comprehensive security headers into every response."""

	async def dispatch(self, request: Request, call_next: RequestResponseEndpoint) -> 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