fix: hardening del engine y limites de tamano tras auditoria de seguridad

.githooks/commit-msg

@@ -0,0 +1,7 @@
+#!/bin/sh
+# ponytail: hard block on AI co-author trailers surviving into a commit.
+# Regla del repo: commits a nombre de Mateo, nunca "Co-Authored-By: Claude/Anthropic".
+msg_file="$1"
+tmp="$(mktemp)"
+grep -viE '^Co-Authored-By:.*(claude|anthropic)' "$msg_file" > "$tmp"
+mv "$tmp" "$msg_file"

.github/workflows/guard-coauthor.yml

@@ -0,0 +1,31 @@
+name: guard-coauthor
+
+# Corta cualquier commit con trailer "Co-Authored-By: Claude/Anthropic" antes
+# de que llegue a main, sin depender del hook local (.githooks/commit-msg).
+on:
+  pull_request:
+  push:
+    branches: [main]
+
+jobs:
+  check:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+        with:
+          fetch-depth: 0
+
+      - name: No AI co-author trailers
+        env:
+          BASE: ${{ github.event.pull_request.base.sha || github.event.before }}
+          HEAD: ${{ github.event.pull_request.head.sha || github.event.after }}
+        run: |
+          if git cat-file -e "$BASE" 2>/dev/null; then
+            hits=$(git log --format=%B "$BASE..$HEAD")
+          else
+            hits=$(git log -1 --format=%B "$HEAD")
+          fi
+          if echo "$hits" | grep -qiE 'Co-Authored-By:.*(claude|anthropic)'; then
+            echo "::error::Co-Authored-By: Claude/Anthropic trailer found. Commits deben ser 100% a nombre del owner."
+            exit 1
+          fi

apps/api/app/api/routes/ws.py

@@ -29,6 +29,7 @@
 
 from __future__ import annotations
 
+import logging
 import uuid
 
 from fastapi import APIRouter, Cookie, WebSocket, WebSocketDisconnect
@@ -43,6 +44,8 @@
 
 router = APIRouter()
 
+logger = logging.getLogger("chronoflow.ws")
+
 
 @router.websocket("/runs/{run_id}")
 async def ws_run_events(
@@ -153,8 +156,10 @@ async def ws_run_events(
     except WebSocketDisconnect:
         pass
     except Exception:
-        # Don't let WS errors crash the server
-        pass
+        # Don't let WS errors crash the server — but a real bug (serialization
+        # failure, DB error mid-stream) must not vanish silently, or production
+        # issues become undiagnosable. Log with run_id for correlation.
+        logger.error("Unhandled error streaming run %s", run_id_str, exc_info=True)
     finally:
         unsubscribe(run_id_str, queue)
         try:

apps/api/app/engine/executors.py

@@ -37,6 +37,13 @@
 
 # ─── Executor registry ──────────────────────────────────────────────────────
 
+# Upper bound on a `delay` node's sleep duration. Without this, an unbounded
+# `seconds` value ties up a scheduler task (and the in-process asyncio.Task
+# pool) indefinitely — a cheap DoS against the demo's single-process runner.
+# A few minutes is generous for demonstrating the ready-set scheduler's
+# parallelism while still bounding worst-case run time.
+MAX_DELAY_SECONDS = 300  # 5 minutes
+
 
 def _assert_public_url(url: str) -> None:
     """SSRF guard for the `http` node: block requests to non-public addresses.
@@ -197,9 +204,22 @@ async def _execute_delay(node: GraphNode, context: dict[str, Any]) -> dict[str,
     A concrete example: two parallel delay branches (3s + 1s) finish
     in ~3s total (max), not 4s (sum). This is the concrete proof of the
     ready-set scheduler working correctly.
+
+    `seconds` is untrusted (comes from a user-authored workflow config), so
+    it's validated here: a negative value is rejected outright (raising
+    surfaces as this node's "failed" state via the scheduler, rather than
+    silently no-op'ing on asyncio.sleep), and the value is capped at
+    MAX_DELAY_SECONDS to bound worst-case run time.
     """
     config = node.data.config
-    seconds: float = float(config.get("seconds", 1))
+    raw_seconds = config.get("seconds", 1)
+    try:
+        seconds: float = float(raw_seconds)
+    except (TypeError, ValueError) as exc:
+        raise ValueError(f"delay node: 'seconds' must be a number, got {raw_seconds!r}") from exc
+    if seconds < 0:
+        raise ValueError(f"delay node: 'seconds' must not be negative (got {seconds})")
+    seconds = min(seconds, MAX_DELAY_SECONDS)
     await asyncio.sleep(seconds)
     return {"waited": seconds}
 

apps/api/app/engine/runner.py

@@ -21,7 +21,7 @@
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from app.engine.scheduler import EventPublisher, run_graph
+from app.engine.scheduler import NODE_ERROR_KEY, EventPublisher, run_graph
 from app.models.run import WorkflowRun
 from app.models.workflow import Workflow
 from app.schemas.graph import Graph
@@ -97,12 +97,14 @@ async def execute_run(
         run.status = "failed"
         run.error = run_error
     else:
-        # Check if any node failed (their output will have {"error": ...})
-        # We detect failure by looking at the events — if any event is "failed",
-        # the overall run is "failed".
+        # Check if any node failed (their output will carry the NODE_ERROR_KEY
+        # sentinel set by the scheduler — see scheduler.py). Deliberately NOT a
+        # bare {"error": ...} shape check: a successful `transform` node's own
+        # user-defined mappings can legitimately produce that exact dict as
+        # real output, which would otherwise be misclassified as a failure.
         failed_nodes = [
             k for k, v in final_context.items()
-            if isinstance(v, dict) and "error" in v and len(v) == 1
+            if isinstance(v, dict) and NODE_ERROR_KEY in v
         ]
         if failed_nodes:
             run.status = "failed"

apps/api/app/engine/scheduler.py

@@ -63,6 +63,16 @@
 # Type alias for the event publisher callback
 EventPublisher = Callable[[ExecutionEvent], Coroutine[Any, Any, None]]
 
+# Sentinel key used to mark a node's context entry as "this node's executor
+# raised" — namespaced so it can't collide with a legitimate node output.
+# A `transform` node's own `mappings` config is user-controlled and could
+# easily produce a real output shaped like {"error": "..."}; that must NOT
+# be mistaken for an executor failure downstream (see runner.py). Kept here
+# (not in executors.py) because the scheduler is the only place that turns
+# a caught exception into a context entry — executors themselves never
+# produce this shape, they raise.
+NODE_ERROR_KEY = "__chronoflow_error__"
+
 
 def _utcnow() -> datetime:
     return datetime.now(UTC)
@@ -287,8 +297,13 @@ async def _run_node(node: GraphNode) -> tuple[str, dict[str, Any] | None, str |
                     error=error_msg,
                     started_at=started_at,
                 )
-                # Store error in context so downstream can see it (unlikely to be used)
-                context[node_id] = {"error": error_msg}
+                # Store error in context so downstream can see it (unlikely to be used).
+                # Uses the namespaced NODE_ERROR_KEY sentinel, not a bare
+                # {"error": ...} dict — a successful `transform` node can
+                # legitimately produce that exact shape from its own
+                # user-defined mappings, which would otherwise collide with
+                # the failure check in runner.py.
+                context[node_id] = {NODE_ERROR_KEY: error_msg}
                 # NOTE: we continue processing (other branches may still complete).
                 # The runner will mark the run as "failed" at the end.
             else:

apps/api/app/schemas/graph.py

@@ -5,12 +5,25 @@
 
 from __future__ import annotations
 
+import json
 from typing import Any, Literal
 
-from pydantic import BaseModel
+from pydantic import BaseModel, Field, field_validator
 
 NodeType = Literal["start", "transform", "http", "delay", "branch", "end"]
 
+# ─── Size limits (DoS guard) ─────────────────────────────────────────────────
+# A workflow graph is user-authored input that gets stored, echoed back on
+# every load, and walked by the scheduler/validator on every run. Without a
+# bound, nothing stops an oversized graph (or an oversized node config blob)
+# from being POSTed — large enough JSON parsing/validation/DB-write/DAG-walk
+# work per request to degrade the single-process API. Limits below are
+# generous for any real workflow (the node palette is intentionally 6 types,
+# see CLAUDE.md) while still rejecting pathological input with a clean 422.
+MAX_NODES = 200
+MAX_EDGES = 400
+MAX_NODE_CONFIG_BYTES = 20_000  # 20 KB of JSON per node's `config`
+
 
 class NodePosition(BaseModel):
     x: float
@@ -21,6 +34,16 @@ class NodeData(BaseModel):
     label: str
     config: dict[str, Any] = {}
 
+    @field_validator("config")
+    @classmethod
+    def _config_not_too_large(cls, v: dict[str, Any]) -> dict[str, Any]:
+        size = len(json.dumps(v, default=str))
+        if size > MAX_NODE_CONFIG_BYTES:
+            raise ValueError(
+                f"Node config is too large ({size} bytes, max {MAX_NODE_CONFIG_BYTES})."
+            )
+        return v
+
 
 class GraphNode(BaseModel):
     id: str
@@ -41,6 +64,6 @@ class GraphEdge(BaseModel):
 
 
 class Graph(BaseModel):
-    nodes: list[GraphNode]
-    edges: list[GraphEdge]
+    nodes: list[GraphNode] = Field(max_length=MAX_NODES)
+    edges: list[GraphEdge] = Field(max_length=MAX_EDGES)
     direction: Literal["vertical", "horizontal"] = "horizontal"

apps/api/app/schemas/workflow.py

@@ -5,16 +5,21 @@
 import uuid
 from datetime import datetime
 
-from pydantic import BaseModel, ConfigDict
+from pydantic import BaseModel, ConfigDict, Field
 
 from app.schemas.graph import Graph
 
+# Same DoS-guard rationale as app/schemas/graph.py (MAX_NODES/MAX_EDGES):
+# unbounded free-text fields on user-authored input are cheap to abuse.
+MAX_NAME_LENGTH = 200
+MAX_DESCRIPTION_LENGTH = 2000
+
 
 class WorkflowIn(BaseModel):
     """Request body for creating or updating a Workflow."""
 
-    name: str
-    description: str | None = None
+    name: str = Field(max_length=MAX_NAME_LENGTH)
+    description: str | None = Field(default=None, max_length=MAX_DESCRIPTION_LENGTH)
     graph: Graph
 
 

apps/api/app/services/task_manager.py

@@ -70,6 +70,17 @@ async def _publish(event: ExecutionEvent) -> None:
 
 
 # ─── Task launcher ────────────────────────────────────────────────────────────
+#
+# run_id → live asyncio.Task registry.
+# asyncio.create_task() does NOT keep a strong reference to the task anywhere
+# by itself — per the stdlib docs, "the event loop only keeps weak references
+# to tasks... if the task is not referenced elsewhere, the garbage collector
+# may destroy it." A background run's task is never awaited by anything, so
+# without a live reference it could be GC'd mid-flight, silently killing the
+# run. Mirrors the `_subscribers` pattern above: module-level registry, added
+# right after creation, cleaned up automatically on completion.
+
+_background_tasks: set[asyncio.Task] = set()
 
 
 def launch_run(run_id: uuid.UUID, session_factory) -> None:
@@ -98,4 +109,8 @@ async def _task() -> None:
                     "Unhandled error in run %s: %s", run_id, exc, exc_info=True
                 )
 
-    asyncio.create_task(_task(), name=f"run-{run_id}")
+    task = asyncio.create_task(_task(), name=f"run-{run_id}")
+    # Keep a strong reference until the task finishes, then let it go —
+    # see the module-level comment above `_background_tasks`.
+    _background_tasks.add(task)
+    task.add_done_callback(_background_tasks.discard)

apps/api/tests/conftest.py

@@ -86,6 +86,7 @@ async def client(engine, session_factory) -> AsyncGenerator[AsyncClient, None]:
 
     auth_routes._login_limiter._hits.clear()
     auth_routes._register_limiter._hits.clear()
+    auth_routes._guest_limiter._hits.clear()
     wf_routes_for_limiter._run_limiter._hits.clear()
 
     async def override_get_session() -> AsyncGenerator[AsyncSession, None]:

apps/api/tests/test_api_workflows.py

@@ -270,3 +270,72 @@ async def test_register_seeds_three_example_workflows(client):
     assert resp.status_code == 201
     listing = await client.get("/api/workflows/")
     assert len(listing.json()) == 3
+
+
+# ─── Size limits (DoS guard, see app/schemas/graph.py + workflow.py) ─────────
+
+
+def _oversized_node_graph(count: int) -> dict:
+    """A graph with `count` isolated (unconnected) nodes past the real `start`.
+
+    We only need enough nodes to trip the Field(max_length=...) check — the
+    graph doesn't need to be a valid DAG for the Pydantic-level 422 to fire,
+    since request-body validation runs before graph validation.
+    """
+    nodes = [
+        {"id": "start", "type": "start", "position": {"x": 0, "y": 0}, "data": {"label": "Start", "config": {}}},
+    ]
+    for i in range(count):
+        nodes.append(
+            {
+                "id": f"extra-{i}",
+                "type": "end",
+                "position": {"x": 0, "y": 0},
+                "data": {"label": "End", "config": {}},
+            }
+        )
+    return {"nodes": nodes, "edges": []}
+
+
+async def test_create_workflow_rejects_too_many_nodes(auth_client):
+    resp = await auth_client.post(
+        "/api/workflows/",
+        json={"name": "Too Big", "graph": _oversized_node_graph(201)},
+    )
+    assert resp.status_code == 422
+
+
+async def test_create_workflow_accepts_max_nodes(auth_client):
+    # Exactly at the limit (1 start + 199 extras = 200) must still be accepted.
+    resp = await auth_client.post(
+        "/api/workflows/",
+        json={"name": "Right At The Limit", "graph": _oversized_node_graph(199)},
+    )
+    assert resp.status_code == 201
+
+
+async def test_create_workflow_rejects_oversized_node_config(auth_client):
+    huge_graph = {
+        "nodes": [
+            {
+                "id": "start",
+                "type": "start",
+                "position": {"x": 0, "y": 0},
+                "data": {"label": "Start", "config": {"blob": "x" * 30_000}},
+            },
+            {"id": "end", "type": "end", "position": {"x": 0, "y": 100}, "data": {"label": "End", "config": {}}},
+        ],
+        "edges": [{"id": "e1", "source": "start", "target": "end"}],
+    }
+    resp = await auth_client.post(
+        "/api/workflows/", json={"name": "Huge Config", "graph": huge_graph}
+    )
+    assert resp.status_code == 422
+
+
+async def test_create_workflow_rejects_oversized_name(auth_client):
+    resp = await auth_client.post(
+        "/api/workflows/",
+        json={"name": "x" * 500, "graph": VALID_GRAPH},
+    )
+    assert resp.status_code == 422

apps/api/tests/test_auth.py

@@ -121,13 +121,13 @@ async def test_guest_creates_session_with_seeded_workflows(client):
 
 @pytest.mark.asyncio
 async def test_guest_rate_limited(client):
-    auth_route._guest_limiter._hits.clear()
+    # `client` fixture (conftest.py) already resets `_guest_limiter` between
+    # tests, so no manual reset needed here (before or after).
     last = None
     for _ in range(auth_route._guest_limiter.max_attempts + 2):
         client.cookies.clear()
         last = await client.post("/api/auth/guest")
     assert last is not None and last.status_code == 429
-    auth_route._guest_limiter._hits.clear()
 
 
 @pytest.mark.asyncio