Merge pull request #213 from reconurge/fix/flow-object-level-authz

fix(api): enforce object-level authorization on Flow update/delete/read

flowsint-api/alembic/versions/c7d1f0a92b3e_add_owner_id_to_flows.py

@@ -0,0 +1,45 @@
+"""add_owner_id_to_flows
+
+Revision ID: c7d1f0a92b3e
+Revises: f4d42260273d
+Create Date: 2026-08-23 00:00:00.000000
+
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = "c7d1f0a92b3e"
+down_revision: Union[str, None] = "f4d42260273d"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Upgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.add_column("flows", sa.Column("owner_id", sa.Uuid(), nullable=True))
+    op.create_foreign_key(
+        "flows_owner_id_fkey",
+        "flows",
+        "profiles",
+        ["owner_id"],
+        ["id"],
+        onupdate="CASCADE",
+        ondelete="SET NULL",
+    )
+    # Pre-existing flows are left with owner_id=NULL: treated as legacy/shared
+    # (readable by everyone, editable/deletable by nobody via the API).
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    """Downgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_constraint("flows_owner_id_fkey", "flows", type_="foreignkey")
+    op.drop_column("flows", "owner_id")
+    # ### end Alembic commands ###

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

@@ -148,6 +148,7 @@ def create_flow(
         description=payload.description,
         category=payload.category,
         flow_schema=payload.flow_schema,
+        owner_id=current_user.id,
     )
 
 
@@ -159,9 +160,11 @@ def get_flow_by_id(
 ):
     service = create_flow_service(db)
     try:
-        return service.get_by_id(flow_id)
+        return service.get_by_id(flow_id, current_user.id)
     except NotFoundError:
         raise HTTPException(status_code=404, detail="Flow not found")
+    except PermissionDeniedError:
+        raise HTTPException(status_code=403, detail="Forbidden")
 
 
 @router.put("/{flow_id}", response_model=FlowRead)
@@ -173,9 +176,13 @@ def update_flow(
 ):
     service = create_flow_service(db)
     try:
-        return service.update(flow_id, payload.model_dump(exclude_unset=True))
+        return service.update(
+            flow_id, current_user.id, payload.model_dump(exclude_unset=True)
+        )
     except NotFoundError:
         raise HTTPException(status_code=404, detail="Flow not found")
+    except PermissionDeniedError:
+        raise HTTPException(status_code=403, detail="Forbidden")
 
 
 @router.delete("/{flow_id}", status_code=status.HTTP_204_NO_CONTENT)
@@ -186,10 +193,12 @@ def delete_flow(
 ):
     service = create_flow_service(db)
     try:
-        service.delete(flow_id)
+        service.delete(flow_id, current_user.id)
         return None
     except NotFoundError:
         raise HTTPException(status_code=404, detail="Flow not found")
+    except PermissionDeniedError:
+        raise HTTPException(status_code=403, detail="Forbidden")
 
 
 @router.post("/{flow_id}/launch")
@@ -201,7 +210,7 @@ async def launch_flow(
 ):
     service = create_flow_service(db)
     try:
-        flow = service.get_by_id(UUID(flow_id))
+        flow = service.get_by_id(UUID(flow_id), current_user.id)
         service.get_sketch_for_launch(payload.sketch_id, current_user.id)
 
         # Retrieve entities from Neo4J by their element IDs

flowsint-api/app/api/schemas/flow.py

@@ -19,6 +19,7 @@ class FlowRead(ORMBase):
     description: Optional[str]
     category: Optional[List[str]]
     flow_schema: Optional[Dict[str, Any]]
+    owner_id: Optional[UUID4]
     created_at: datetime
     last_updated_at: datetime
 

flowsint-core/src/flowsint_core/core/models.py

@@ -211,6 +211,11 @@ class Flow(Base):
     flow_schema = mapped_column(JSON, nullable=True)
     created_at = mapped_column(DateTime(timezone=True), server_default=func.now())
     last_updated_at = mapped_column(DateTime(timezone=True), server_default=func.now())
+    owner_id = mapped_column(
+        Uuid,
+        ForeignKey("profiles.id", onupdate="CASCADE", ondelete="SET NULL"),
+        nullable=True,
+    )
 
 
 class Analysis(Base):

flowsint-core/src/flowsint_core/core/repositories/flow_repository.py

@@ -2,6 +2,8 @@
 
 from typing import List, Optional
 
+from sqlalchemy import or_
+
 from ..models import Flow
 from .base import BaseRepository
 
@@ -10,9 +12,13 @@ class FlowRepository(BaseRepository[Flow]):
     model = Flow
 
     def get_all_with_optional_category(
-        self, category: Optional[str] = None
+        self, category: Optional[str] = None, user_id: Optional[UUID] = None
     ) -> List[Flow]:
-        query = self._db.query(Flow).order_by(Flow.last_updated_at.desc())
+        query = (
+            self._db.query(Flow)
+            .filter(or_(Flow.owner_id.is_(None), Flow.owner_id == user_id))
+            .order_by(Flow.last_updated_at.desc())
+        )
 
         if not category or category.lower() == "undefined":
             return query.all()

flowsint-core/src/flowsint_core/core/services/flow_service.py

@@ -43,7 +43,7 @@ class FlowService(BaseService):
         self, category: Optional[str], user_id: UUID
     ) -> List[Dict[str, Any]]:
         if not category or category.lower() == "undefined":
-            return self._flow_repo.get_all_with_optional_category(None)
+            return self._flow_repo.get_all_with_optional_category(None, user_id)
 
         # Check if category is a custom type
         custom_type = self._custom_type_repo.get_published_by_name_and_owner(
@@ -51,7 +51,7 @@ class FlowService(BaseService):
         )
 
         if custom_type:
-            flows = self._flow_repo.get_all_with_optional_category(None)
+            flows = self._flow_repo.get_all_with_optional_category(None, user_id)
             return []
             return [
                 {
@@ -61,12 +61,14 @@ class FlowService(BaseService):
                 for flow in flows
             ]
 
-        return self._flow_repo.get_all_with_optional_category(category)
+        return self._flow_repo.get_all_with_optional_category(category, user_id)
 
-    def get_by_id(self, flow_id: UUID) -> Flow:
+    def get_by_id(self, flow_id: UUID, user_id: UUID) -> Flow:
         flow = self._flow_repo.get_by_id(flow_id)
         if not flow:
             raise NotFoundError("Flow not found")
+        if flow.owner_id is not None and flow.owner_id != user_id:
+            raise PermissionDeniedError("Forbidden")
         return flow
 
     def create(
@@ -75,6 +77,7 @@ class FlowService(BaseService):
         description: Optional[str],
         category: List[str],
         flow_schema: Dict[str, Any],
+        owner_id: UUID,
     ) -> Flow:
         new_flow = Flow(
             id=uuid4(),
@@ -82,6 +85,7 @@ class FlowService(BaseService):
             description=description,
             category=category,
             flow_schema=flow_schema,
+            owner_id=owner_id,
             created_at=datetime.now(timezone.utc),
             last_updated_at=datetime.now(timezone.utc),
         )
@@ -90,10 +94,12 @@ class FlowService(BaseService):
         self._refresh(new_flow)
         return new_flow
 
-    def update(self, flow_id: UUID, updates: Dict[str, Any]) -> Flow:
+    def update(self, flow_id: UUID, user_id: UUID, updates: Dict[str, Any]) -> Flow:
         flow = self._flow_repo.get_by_id(flow_id)
         if not flow:
             raise NotFoundError("Flow not found")
+        if flow.owner_id != user_id:
+            raise PermissionDeniedError("Forbidden")
 
         for key, value in updates.items():
             if key == "category":
@@ -106,10 +112,12 @@ class FlowService(BaseService):
         self._refresh(flow)
         return flow
 
-    def delete(self, flow_id: UUID) -> None:
+    def delete(self, flow_id: UUID, user_id: UUID) -> None:
         flow = self._flow_repo.get_by_id(flow_id)
         if not flow:
             raise NotFoundError("Flow not found")
+        if flow.owner_id != user_id:
+            raise PermissionDeniedError("Forbidden")
 
         self._flow_repo.delete(flow)
         self._commit()

flowsint-core/tests/services/test_timezone_timestamps.py

@@ -67,6 +67,7 @@ def test_flow_service_create_uses_timezone_aware_timestamps():
         description=None,
         category=["Domain"],
         flow_schema={"nodes": [], "edges": []},
+        owner_id=uuid4(),
     )
 
     _assert_utc(flow.created_at)