fix: Add missing API endpoints and validation improvements

- Add cancel_sprint and delete_sprint endpoints to sprints.py
- Add unassign_issue endpoint to issues.py
- Add remove_issue_from_sprint endpoint to sprints.py
- Add CRUD methods: remove_sprint_from_issues, unassign, remove_from_sprint
- Add validation to prevent closed issues in active/planned sprints
- Add authorization tests for SSE events endpoint
- Fix IDOR vulnerabilities in agents.py and projects.py
- Add Syndarix models migration (0004)

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
2025-12-30 15:39:51 +01:00
parent b43fa8ace2
commit cea97afe25
10 changed files with 1156 additions and 74 deletions

View File

@@ -17,6 +17,7 @@ Features:
import asyncio
import json
import logging
from typing import TYPE_CHECKING
from uuid import UUID
from fastapi import APIRouter, Depends, Header, Request
@@ -26,12 +27,16 @@ from sse_starlette.sse import EventSourceResponse
from app.api.dependencies.auth import get_current_user
from app.api.dependencies.event_bus import get_event_bus
from app.core.database import get_db
from app.core.exceptions import AuthorizationError
from app.models.user import User
from app.schemas.errors import ErrorCode
from app.schemas.events import EventType
from app.services.event_bus import EventBus
if TYPE_CHECKING:
from sqlalchemy.ext.asyncio import AsyncSession
logger = logging.getLogger(__name__)
router = APIRouter()
@@ -44,33 +49,44 @@ KEEPALIVE_INTERVAL = 30
async def check_project_access(
project_id: UUID,
user: User,
db: "AsyncSession",
) -> bool:
"""
Check if a user has access to a project's events.
This is a placeholder implementation that will be replaced
with actual project authorization logic once the Project model
is implemented. Currently allows access for all authenticated users.
Authorization rules:
- Superusers can access all projects
- Project owners can access their own projects
Args:
project_id: The project to check access for
user: The authenticated user
db: Database session for project lookup
Returns:
bool: True if user has access, False otherwise
TODO: Implement actual project authorization
- Check if user owns the project
- Check if user is a member of the project
- Check project visibility settings
"""
# Placeholder: Allow all authenticated users for now
# This will be replaced with actual project ownership/membership check
# Superusers can access all projects
if user.is_superuser:
logger.debug(
f"Project access granted for superuser {user.id} on project {project_id}"
)
return True
# Check if user owns the project
from app.crud.syndarix import project as project_crud
project = await project_crud.get(db, id=project_id)
if not project:
logger.debug(f"Project {project_id} not found for access check")
return False
has_access = bool(project.owner_id == user.id)
logger.debug(
f"Project access check for user {user.id} on project {project_id} "
"(placeholder: allowing all authenticated users)"
f"Project access {'granted' if has_access else 'denied'} "
f"for user {user.id} on project {project_id} (owner: {project.owner_id})"
)
return True
return has_access
async def event_generator(
@@ -176,6 +192,7 @@ async def stream_project_events(
project_id: UUID,
current_user: User = Depends(get_current_user),
event_bus: EventBus = Depends(get_event_bus),
db: "AsyncSession" = Depends(get_db),
last_event_id: str | None = Header(None, alias="Last-Event-ID"),
):
"""
@@ -197,7 +214,7 @@ async def stream_project_events(
)
# Check project access
has_access = await check_project_access(project_id, current_user)
has_access = await check_project_access(project_id, current_user, db)
if not has_access:
raise AuthorizationError(
message=f"You don't have access to project {project_id}",
@@ -244,6 +261,7 @@ async def send_test_event(
project_id: UUID,
current_user: User = Depends(get_current_user),
event_bus: EventBus = Depends(get_event_bus),
db: "AsyncSession" = Depends(get_db),
):
"""
Send a test event to the project's event stream.
@@ -251,7 +269,7 @@ async def send_test_event(
This is useful for testing SSE connections during development.
"""
# Check project access
has_access = await check_project_access(project_id, current_user)
has_access = await check_project_access(project_id, current_user, db)
if not has_access:
raise AuthorizationError(
message=f"You don't have access to project {project_id}",