diff --git a/flowsint-api/app/api/deps.py b/flowsint-api/app/api/deps.py index 98804d3..3dd967c 100644 --- a/flowsint-api/app/api/deps.py +++ b/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 diff --git a/flowsint-api/app/api/routes/events.py b/flowsint-api/app/api/routes/events.py index 6161a0b..cc891f3 100644 --- a/flowsint-api/app/api/routes/events.py +++ b/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", - }, - )