Feat: better backend (#241)

* feat: misc issues fixed, also update saga/exec status
* feat: better msg parsing from pods: using NAME+BYTESIZE instead of JSON. Also simplified logging in event mapper
* feat: issues fixing (passing exit code instead of 0, previous issues, ..)

Author: HardMax71

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