feat(api): authenticate SSE via header, drop query-param auth and dead route

flowsint-api/app/api/deps.py

@@ -1,11 +1,10 @@
-from fastapi import Depends, HTTPException, status, Request
+from fastapi import Depends, HTTPException, status
 from fastapi.security import OAuth2PasswordBearer
 from jose import JWTError, jwt
 from sqlalchemy.orm import Session
 from flowsint_core.core.auth import ALGORITHM, AUTH_SECRET
 from flowsint_core.core.postgre_db import get_db
 from flowsint_core.core.models import Profile
-from typing import Optional
 
 oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
 
@@ -29,42 +28,3 @@ def get_current_user(
     if user is None:
         raise credentials_exception
     return user
-
-
-def get_current_user_sse(
-    request: Request, db: Session = Depends(get_db)
-) -> Profile:
-    """
-    Alternative authentication for SSE endpoints that accepts token via query parameter.
-    EventSource API doesn't support custom headers, so we need to pass the token in the URL.
-    """
-    credentials_exception = HTTPException(
-        status_code=status.HTTP_401_UNAUTHORIZED,
-        detail="Could not validate credentials",
-    )
-
-    # Try to get token from query parameter
-    token: Optional[str] = request.query_params.get("token")
-
-    # Fallback to Authorization header if query param not present
-    if not token:
-        auth_header = request.headers.get("Authorization")
-        if auth_header and auth_header.startswith("Bearer "):
-            token = auth_header.replace("Bearer ", "")
-
-    if not token:
-        raise credentials_exception
-
-    try:
-        payload = jwt.decode(token, AUTH_SECRET, algorithms=[ALGORITHM])
-        email: str = payload.get("sub")
-        if email is None:
-            raise credentials_exception
-    except JWTError:
-        raise credentials_exception
-
-    user = db.query(Profile).filter(Profile.email == email).first()
-    if user is None:
-        raise credentials_exception
-
-    return user

flowsint-api/app/api/routes/events.py

@@ -14,7 +14,7 @@ from flowsint_core.core.services import (
     PermissionDeniedError,
     DatabaseError,
 )
-from app.api.deps import get_current_user, get_current_user_sse
+from app.api.deps import get_current_user
 
 router = APIRouter()
 
@@ -42,7 +42,7 @@ async def stream_events(
     request: Request,
     sketch_id: str,
     db: Session = Depends(get_db),
-    current_user: Profile = Depends(get_current_user_sse),
+    current_user: Profile = Depends(get_current_user),
 ):
     """Stream events for a specific sketch in real-time."""
     service = create_log_service(db)
@@ -58,7 +58,7 @@ async def stream_events(
         channel = sketch_id
         await event_emitter.subscribe(channel)
         try:
-            yield 'data: {"event": "connected", "data": "Connected to log stream"}\n\n'
+            yield json.dumps({"event": "connected", "data": "Connected to log stream"})
             while True:
                 if await request.is_disconnected():
                     break
@@ -115,7 +115,7 @@ async def stream_sketch_status(
     request: Request,
     sketch_id: str,
     db: Session = Depends(get_db),
-    current_user: Profile = Depends(get_current_user_sse),
+    current_user: Profile = Depends(get_current_user),
 ):
     """Stream COMPLETED events for a specific sketch (for graph refresh)."""
     service = create_log_service(db)
@@ -160,49 +160,3 @@ async def stream_sketch_status(
             "X-Accel-Buffering": "no",
         },
     )
-
-
-@router.get("/status/scan/{scan_id}/stream")
-async def stream_status(
-    request: Request,
-    scan_id: str,
-    db: Session = Depends(get_db),
-    current_user: Profile = Depends(get_current_user_sse),
-):
-    """Stream status updates for a specific scan in real-time."""
-    service = create_log_service(db)
-    try:
-        service.get_scan_with_permission(scan_id, current_user.id)
-    except NotFoundError as e:
-        raise HTTPException(status_code=404, detail=str(e))
-    except PermissionDeniedError:
-        raise HTTPException(status_code=403, detail="Forbidden")
-
-    async def status_generator():
-        print("[EventEmitter] Start status generator")
-        await event_emitter.subscribe(f"scan_{scan_id}_status")
-        try:
-            yield 'data: {"event": "connected", "data": "Connected to status stream"}\n\n'
-
-            while True:
-                data = await event_emitter.get_message(f"scan_{scan_id}_status")
-                if data is None:
-                    await asyncio.sleep(0.1)
-                    continue
-                print(f"[EventEmitter] Received status data: {data}")
-                yield f"data: {data}\n\n"
-
-        except asyncio.CancelledError:
-            print(f"[EventEmitter] Client disconnected from status stream for scan_id: {scan_id}")
-        finally:
-            await event_emitter.unsubscribe(f"scan_{scan_id}_status")
-
-    return EventSourceResponse(
-        status_generator(),
-        media_type="text/event-stream",
-        headers={
-            "Cache-Control": "no-cache",
-            "Connection": "keep-alive",
-            "X-Accel-Buffering": "no",
-        },
-    )