Hire Dedicated Python Developers For AI Apps & Data Pipelines

# Python 3.11+ - FastAPI WebSocket with Security & Best Practices
from typing import Annotated
from datetime import datetime, timezone
from contextlib import asynccontextmanager
from urllib.parse import quote
import secrets

from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Depends, status
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.trustedhost import TrustedHostMiddleware
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import Response
from pydantic import ValidationError
import nh3

from app.core.logging import get_logger
from app.core.config import settings
from app.core.metrics import metrics
from app.core.redis import get_redis_client
from app.schemas.websocket import MessageSchema, WebSocketEventSchema
from app.services.auth import AuthService
from app.services.rate_limiter import RateLimiter
from app.managers.connection_manager import ConnectionManager

logger = get_logger(__name__)

# Initialize services with dependency injection
auth_service = AuthService()
rate_limiter = RateLimiter(redis_client=get_redis_client())
connection_manager = ConnectionManager(
    redis_client=get_redis_client(),
    logger=logger
)

# Security headers middleware
class SecurityHeadersMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next) -> Response:
        response = await call_next(request)
        response.headers["X-Content-Type-Options"] = "nosniff"
        response.headers["X-Frame-Options"] = "DENY"
        response.headers["X-XSS-Protection"] = "1; mode=block"
        response.headers["Strict-Transport-Security"] = "max-age=31536000; includeSubDomains"
        response.headers["Content-Security-Policy"] = "default-src 'self'"
        response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin"
        return response

@asynccontextmanager
async def lifespan(app: FastAPI):
    """Manage application lifecycle with proper cleanup"""
    await connection_manager.initialize()
    logger.info("Application started", version=settings.APP_VERSION)
    yield
    await connection_manager.cleanup()
    logger.info("Application shutdown complete")

# Create FastAPI app
app = FastAPI(
    title=settings.APP_NAME,
    version=settings.APP_VERSION,
    lifespan=lifespan,
    docs_url="/docs" if settings.DEBUG else None,
    redoc_url=None,  # Disable redoc in production
    openapi_url="/openapi.json" if settings.DEBUG else None
)

# Add security middleware stack
app.add_middleware(SecurityHeadersMiddleware)
app.add_middleware(TrustedHostMiddleware, allowed_hosts=settings.ALLOWED_HOSTS)
app.add_middleware(
    CORSMiddleware,
    allow_origins=settings.CORS_ORIGINS,  # Explicit origins, not "*"
    allow_credentials=True,
    allow_methods=["GET", "POST"],  # Restrict methods
    allow_headers=["Authorization", "Content-Type"],
    max_age=600,
)

def sanitize_input(text: str, max_length: int = 1000) -> str:
    """Sanitize user input to prevent XSS"""
    # Strip all HTML tags and limit length
    cleaned = nh3.clean(text, tags=set())
    return cleaned[:max_length].strip()

def validate_room_id(room_id: str) -> str:
    """Validate and sanitize room ID"""
    if not room_id or len(room_id) > 64:
        raise ValueError("Invalid room ID")
    # Only allow alphanumeric and hyphens
    if not room_id.replace("-", "").isalnum():
        raise ValueError("Room ID contains invalid characters")
    return quote(room_id, safe="")

@app.websocket("/ws/{room_id}")
async def websocket_endpoint(
    websocket: WebSocket,
    room_id: str,
    user: Annotated[str, Depends(auth_service.get_user_from_token)],
    _rate_limit: Annotated[None, Depends(rate_limiter.check)]
):
    """WebSocket endpoint with auth, rate limiting, and input validation"""

    # Validate room ID
    try:
        safe_room_id = validate_room_id(room_id)
    except ValueError as e:
        await websocket.close(code=status.WS_1008_POLICY_VIOLATION, reason=str(e))
        return

    # Check room access
    if not await auth_service.can_access_room(user, safe_room_id):
        await websocket.close(code=status.WS_1008_POLICY_VIOLATION, reason="Access denied")
        return

    # Accept the WebSocket connection
    await websocket.accept()

    # Generate secure client ID
    client_id = secrets.token_urlsafe(16)

    # Connect client
    await connection_manager.connect(
        websocket=websocket,
        room_id=safe_room_id,
        user_id=user,
        client_id=client_id
    )

    try:
        while True:
            raw_data = await websocket.receive_text()

            # Validate message size
            if len(raw_data) > 10_000:
                await connection_manager.send_error(client_id, "MESSAGE_TOO_LARGE")
                continue

            try:
                event = WebSocketEventSchema.model_validate_json(raw_data)
            except ValidationError as e:
                logger.warning("Invalid message", errors=e.errors(), client=client_id)
                await connection_manager.send_error(client_id, "INVALID_FORMAT")
                continue

            match event.type:
                case "message":
                    # Sanitize message content
                    safe_text = sanitize_input(event.payload.get("text", ""))
                    if not safe_text:
                        continue

                    message_id = await connection_manager.process_message(
                        room_id=safe_room_id,
                        user_id=user,
                        text=safe_text
                    )
                    metrics.record_message_sent(safe_room_id, user)
                    logger.debug("Message sent", message_id=message_id)

                case "typing":
                    await connection_manager.broadcast_typing(
                        room_id=safe_room_id,
                        user_id=user,
                        is_typing=bool(event.payload.get("is_typing"))
                    )

                case "heartbeat":
                    await connection_manager.heartbeat(client_id)

                case _:
                    logger.warning("Unknown event", type=event.type)

    except WebSocketDisconnect:
        logger.info("Client disconnected", client=client_id, room=safe_room_id)
    except Exception as e:
        logger.error("WebSocket error", error=str(e), client=client_id, exc_info=True)
        metrics.record_error("websocket_error")
    finally:
        await connection_manager.disconnect(client_id)

@app.get("/health")
async def health_check():
    """Health check with service status"""
    redis_ok = await connection_manager.check_redis_health()
    return {
        "status": "healthy" if redis_ok else "degraded",
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "version": settings.APP_VERSION
    }