Feat: better backend

backend/app/api/routes/admin/events.py

@@ -9,6 +9,7 @@
 from sse_starlette.sse import EventSourceResponse
 
 from app.api.dependencies import admin_user
+from app.api.routes.common import SSEResponse
 from app.domain.enums import EventType, ExportFormat
 from app.domain.events import EventFilter as DomainEventFilter
 from app.domain.replay import ReplayFilter
@@ -27,15 +28,6 @@
 from app.services.admin import AdminEventsService
 from app.services.sse import SSEService
 
-
-class _SSEResponse(EventSourceResponse):
-    """Workaround: sse-starlette sets media_type only in __init__, not as a
-    class attribute.  FastAPI reads the class attribute for OpenAPI generation,
-    so without this subclass every SSE endpoint shows application/json."""
-
-    media_type = "text/event-stream"
-
-
 router = APIRouter(
     prefix="/admin/events", tags=["admin-events"], route_class=DishkaRoute, dependencies=[Depends(admin_user)]
 )
@@ -144,7 +136,7 @@ async def replay_events(
 
 @router.get(
     "/replay/{session_id}/status",
-    response_class=_SSEResponse,
+    response_class=SSEResponse,
     responses={
         200: {"model": EventReplayStatusResponse},
         404: {"model": ErrorResponse, "description": "Replay session not found"},

backend/app/api/routes/admin/executions.py

@@ -20,6 +20,7 @@
     prefix="/admin/executions",
     tags=["admin-executions"],
     route_class=DishkaRoute,
+    dependencies=[Depends(admin_user)],
 )
 
 

backend/app/api/routes/common.py

@@ -0,0 +1,9 @@
+from sse_starlette.sse import EventSourceResponse
+
+
+class SSEResponse(EventSourceResponse):
+    """Workaround: sse-starlette sets media_type only in __init__, not as a
+    class attribute.  FastAPI reads the class attribute for OpenAPI generation,
+    so without this subclass every SSE endpoint shows application/json."""
+
+    media_type = "text/event-stream"

backend/app/api/routes/sse.py

@@ -6,26 +6,18 @@
 from sse_starlette.sse import EventSourceResponse
 
 from app.api.dependencies import current_user
+from app.api.routes.common import SSEResponse
 from app.domain.user import User
 from app.schemas_pydantic.notification import NotificationResponse
 from app.schemas_pydantic.sse import SSEExecutionEventSchema
 from app.services.sse import SSEService
 
-
-class _SSEResponse(EventSourceResponse):
-    """Workaround: sse-starlette sets media_type only in __init__, not as a
-    class attribute.  FastAPI reads the class attribute for OpenAPI generation,
-    so without this subclass every SSE endpoint shows application/json."""
-
-    media_type = "text/event-stream"
-
-
 router = APIRouter(prefix="/events", tags=["sse"], route_class=DishkaRoute)
 
 
 @router.get(
     "/notifications/stream",
-    response_class=_SSEResponse,
+    response_class=SSEResponse,
     responses={200: {"model": NotificationResponse}},
 )
 async def notification_stream(
@@ -41,7 +33,7 @@ async def notification_stream(
 
 @router.get(
     "/executions/{execution_id}",
-    response_class=_SSEResponse,
+    response_class=SSEResponse,
     responses={200: {"model": SSEExecutionEventSchema}},
 )
 async def execution_events(

backend/app/core/container.py

@@ -35,7 +35,7 @@ def create_app_container(settings: Settings) -> AsyncContainer:
     Args:
         settings: Application settings (injected via from_context).
 
-    Note: init_beanie() must be called BEFORE this container is created.
+    Note: init_beanie() is called in the async lifespan AFTER this container is created.
     KafkaBroker is created by BrokerProvider and can be retrieved
     via container.get(KafkaBroker) after container creation.
     """

backend/app/core/metrics/queue.py

@@ -30,6 +30,11 @@ def _create_instruments(self) -> None:
             description="Time spent waiting in queue before scheduling",
             unit="s",
         )
+        self._event_data_lost = self._meter.create_counter(
+            name="queue.event_data_lost",
+            description="Executions lost due to expired event data in Redis",
+            unit="1",
+        )
 
     def record_enqueue(self) -> None:
         self._enqueue_total.add(1)
@@ -45,3 +50,6 @@ def record_release(self) -> None:
 
     def record_wait_time(self, wait_seconds: float, priority: str) -> None:
         self._wait_time.record(wait_seconds, attributes={"priority": priority})
+
+    def record_event_data_lost(self) -> None:
+        self._event_data_lost.add(1)

backend/app/core/middlewares/__init__.py

@@ -3,6 +3,7 @@
 from .metrics import MetricsMiddleware, create_system_metrics, setup_metrics
 from .rate_limit import RateLimitMiddleware
 from .request_size_limit import RequestSizeLimitMiddleware
+from .security_headers import SecurityHeadersMiddleware
 
 __all__ = [
     "CacheControlMiddleware",
@@ -12,4 +13,5 @@
     "create_system_metrics",
     "RequestSizeLimitMiddleware",
     "RateLimitMiddleware",
+    "SecurityHeadersMiddleware",
 ]

backend/app/core/middlewares/csrf.py

@@ -19,7 +19,7 @@ class CSRFMiddleware:
 
     Requests are skipped if:
     - Method is safe (GET, HEAD, OPTIONS)
-    - Path is an auth endpoint (login, register, logout)
+    - Path is an auth endpoint (login, register)
     - Path is not under /api/
     - User is not authenticated (no access_token cookie)
     """

backend/app/core/middlewares/rate_limit.py

@@ -7,7 +7,6 @@
 
 from app.core.utils import get_client_ip
 from app.domain.rate_limit import RateLimitStatus
-from app.domain.user import User
 from app.services.rate_limit_service import RateLimitService
 from app.settings import Settings
 
@@ -102,10 +101,14 @@ async def send_wrapper(message: Message) -> None:
         await self.app(scope, receive, send_wrapper)
 
     # --8<-- [start:extract_user_id]
-    def _extract_user_id(self, request: Request) -> str:
-        user: User | None = request.state.__dict__.get("user")
-        if user:
-            return str(user.user_id)
+    @staticmethod
+    def _extract_user_id(request: Request) -> str:
+        """Extract rate-limit bucket key from client IP.
+
+        Middleware runs before route-level auth, so no verified identity is
+        available here. Using unverified JWT claims would let an attacker
+        craft arbitrary bucket keys to bypass IP-based limits.
+        """
         return f"ip:{get_client_ip(request)}"
     # --8<-- [end:extract_user_id]
 

backend/app/core/middlewares/request_size_limit.py

@@ -1,10 +1,15 @@
 from starlette.responses import JSONResponse
-from starlette.types import ASGIApp, Receive, Scope, Send
+from starlette.types import ASGIApp, Message, Receive, Scope, Send
 
 
 # --8<-- [start:RequestSizeLimitMiddleware]
 class RequestSizeLimitMiddleware:
-    """Middleware to limit request size, default 10MB"""
+    """Middleware to limit request size, default 10MB.
+
+    Checks Content-Length header when present for an early reject, and wraps
+    the ASGI ``receive`` callable to count bytes as they stream — this
+    catches chunked-transfer requests that omit Content-Length.
+    """
 
     def __init__(self, app: ASGIApp, max_size_mb: int = 10) -> None:
         self.app = app
@@ -29,4 +34,30 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
                 await response(scope, receive, send)
                 return
 
-        await self.app(scope, receive, send)
+        bytes_received = 0
+        max_size = self.max_size_bytes
+        exceeded = False
+
+        async def receive_wrapper() -> Message:
+            nonlocal bytes_received, exceeded
+            message = await receive()
+            if message["type"] == "http.request":
+                body = message.get("body", b"")
+                bytes_received += len(body)
+                if bytes_received > max_size:
+                    exceeded = True
+                    raise _RequestTooLarge()
+            return message
+
+        try:
+            await self.app(scope, receive_wrapper, send)
+        except _RequestTooLarge:
+            response = JSONResponse(
+                status_code=413,
+                content={"detail": f"Request too large. Maximum size is {max_size / 1024 / 1024}MB"},
+            )
+            await response(scope, receive, send)
+
+
+class _RequestTooLarge(Exception):
+    pass

backend/app/core/middlewares/security_headers.py

@@ -0,0 +1,31 @@
+from starlette.datastructures import MutableHeaders
+from starlette.types import ASGIApp, Message, Receive, Scope, Send
+
+
+class SecurityHeadersMiddleware:
+    """Add standard security response headers to every HTTP response."""
+
+    _HEADERS = {
+        "X-Content-Type-Options": "nosniff",
+        "X-Frame-Options": "DENY",
+        "Referrer-Policy": "strict-origin-when-cross-origin",
+        "Strict-Transport-Security": "max-age=63072000; includeSubDomains",
+        "Content-Security-Policy": "default-src 'self'; frame-ancestors 'none'",
+    }
+
+    def __init__(self, app: ASGIApp) -> None:
+        self.app = app
+
+    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
+        if scope["type"] != "http":
+            await self.app(scope, receive, send)
+            return
+
+        async def send_wrapper(message: Message) -> None:
+            if message["type"] == "http.response.start":
+                headers = MutableHeaders(scope=message)
+                for name, value in self._HEADERS.items():
+                    headers[name] = value
+            await send(message)
+
+        await self.app(scope, receive, send_wrapper)

backend/app/core/security.py

@@ -42,16 +42,24 @@ def create_access_token(self, data: dict[str, Any], expires_delta: timedelta) ->
         return encoded_jwt
     # --8<-- [end:create_access_token]
 
-    def decode_token(self, token: str) -> str:
-        """Decode JWT and return the username (sub claim)."""
+    def decode_token(self, token: str, *, allow_expired: bool = False) -> str:
+        """Decode JWT and return the username (sub claim).
+
+        Args:
+            token: The JWT token string.
+            allow_expired: If True, accept expired tokens (used for logout).
+        """
         try:
-            payload = jwt.decode(token, self.settings.SECRET_KEY, algorithms=[self.settings.ALGORITHM])
+            options = {"verify_exp": not allow_expired}
+            payload = jwt.decode(
+                token, self.settings.SECRET_KEY, algorithms=[self.settings.ALGORITHM], options=options,
+            )
             username: str | None = payload.get("sub")
             if username is None:
                 raise InvalidCredentialsError()
-        except jwt.PyJWTError as e:
-            raise InvalidCredentialsError() from e
-        return username
+            return username
+        except jwt.PyJWTError:
+            raise InvalidCredentialsError() from None
 
     def generate_csrf_token(self, session_id: str) -> str:
         """Generate a signed CSRF token bound to the given session (access_token).

backend/app/core/utils.py

@@ -1,4 +1,5 @@
 from enum import StrEnum
+from ipaddress import ip_address
 
 from fastapi import Request
 
@@ -32,23 +33,31 @@ def __format__(self, format_spec: str) -> str:
 
 
 def get_client_ip(request: Request) -> str:
+    """Get client IP address from request.
+
+    Uses the direct connection IP as the authoritative source.
+    X-Forwarded-For is only trusted when the direct connection comes from a
+    known local/proxy address, preventing attackers from spoofing their IP to
+    bypass rate limiting.
     """
-    Safely get client IP address from request.
-    Handles both normal connections and proxy forwarded requests.
-    """
-    # Check for proxy headers first (in order of preference)
-    forwarded_for = request.headers.get("x-forwarded-for")
-    if forwarded_for:
-        # X-Forwarded-For can contain multiple IPs, take the first one
-        return forwarded_for.split(",")[0].strip()
-
-    real_ip = request.headers.get("x-real-ip")
-    if real_ip:
-        return real_ip
-
-    # Fall back to direct client connection
-    if request.client:
-        return request.client.host
-
-    # Should rarely happen, but handle the case where client is None
-    return "127.0.0.1"
+    direct_ip = request.client.host if request.client else "127.0.0.1"
+
+    if _is_trusted_proxy(direct_ip):
+        forwarded_for = request.headers.get("x-forwarded-for")
+        if forwarded_for:
+            return forwarded_for.split(",")[0].strip()
+
+        real_ip = request.headers.get("x-real-ip")
+        if real_ip:
+            return real_ip
+
+    return direct_ip
+
+
+def _is_trusted_proxy(ip: str) -> bool:
+    """Check if an IP belongs to a trusted proxy (loopback or RFC 1918 private ranges)."""
+    try:
+        addr = ip_address(ip)
+    except ValueError:
+        return False
+    return addr.is_loopback or addr.is_private

backend/app/db/docs/resource.py

@@ -3,6 +3,8 @@
 from beanie import Document, Indexed
 from pydantic import ConfigDict, Field
 
+from app.domain.enums import AllocationStatus
+
 
 class ResourceAllocationDocument(Document):
     """Resource allocation bookkeeping document used by saga steps.
@@ -17,7 +19,7 @@ class ResourceAllocationDocument(Document):
     memory_request: str
     cpu_limit: str
     memory_limit: str
-    status: Indexed(str) = "active"  # type: ignore[valid-type]  # "active" | "released"
+    status: Indexed(str) = AllocationStatus.ACTIVE  # type: ignore[valid-type]
     allocated_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
     released_at: datetime | None = None
 

backend/app/db/repositories/event_repository.py

@@ -57,6 +57,12 @@ async def store_event(self, event: DomainEvent) -> str:
         self.logger.debug(f"Stored event {event.event_id} of type {event.event_type}")
         return event.event_id
 
+    async def mark_publish_failed(self, event_id: str) -> None:
+        """Mark an event as failed to publish to Kafka for later retry."""
+        await EventDocument.find_one(
+            EventDocument.event_id == event_id,
+        ).update({"$set": {"publish_failed": True, "publish_failed_at": datetime.now(timezone.utc)}})
+
     async def get_event(self, event_id: str) -> DomainEvent | None:
         doc = await EventDocument.find_one(EventDocument.event_id == event_id)
         if not doc:
@@ -257,8 +263,12 @@ async def delete_event_with_archival(
         deleted_at = datetime.now(timezone.utc)
         archive_fields = {"deleted_at": deleted_at, "deleted_by": deleted_by, "deletion_reason": deletion_reason}
         archived_doc = EventArchiveDocument.model_validate(doc).model_copy(update=archive_fields)
-        await archived_doc.insert()
-        await doc.delete()
+
+        async with EventDocument.get_pymongo_collection().database.client.start_session() as session:
+            async with await session.start_transaction():
+                await archived_doc.insert(session=session)
+                await doc.delete(session=session)
+
         return ArchivedEvent.model_validate(doc).model_copy(update=archive_fields)
 
     async def get_aggregate_replay_info(self, aggregate_id: str) -> EventReplayInfo | None: