diff --git a/.githooks/commit-msg b/.githooks/commit-msg new file mode 100644 index 0000000..84d1537 --- /dev/null +++ b/.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" diff --git a/.github/workflows/guard-coauthor.yml b/.github/workflows/guard-coauthor.yml new file mode 100644 index 0000000..fea85de --- /dev/null +++ b/.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 diff --git a/apps/api/app/api/routes/ws.py b/apps/api/app/api/routes/ws.py index 226d459..69b305a 100644 --- a/apps/api/app/api/routes/ws.py +++ b/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: diff --git a/apps/api/app/engine/executors.py b/apps/api/app/engine/executors.py index d7fd597..b671494 100644 --- a/apps/api/app/engine/executors.py +++ b/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} diff --git a/apps/api/app/engine/runner.py b/apps/api/app/engine/runner.py index b4408f0..92b3e38 100644 --- a/apps/api/app/engine/runner.py +++ b/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" diff --git a/apps/api/app/engine/scheduler.py b/apps/api/app/engine/scheduler.py index c7a70d1..a1ed1fd 100644 --- a/apps/api/app/engine/scheduler.py +++ b/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: diff --git a/apps/api/app/schemas/graph.py b/apps/api/app/schemas/graph.py index f6bbe6b..7233254 100644 --- a/apps/api/app/schemas/graph.py +++ b/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" diff --git a/apps/api/app/schemas/workflow.py b/apps/api/app/schemas/workflow.py index dfb1c7a..49a1531 100644 --- a/apps/api/app/schemas/workflow.py +++ b/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 diff --git a/apps/api/app/services/task_manager.py b/apps/api/app/services/task_manager.py index ade1327..e6cee5f 100644 --- a/apps/api/app/services/task_manager.py +++ b/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) diff --git a/apps/api/tests/conftest.py b/apps/api/tests/conftest.py index 2113bab..fa8bcf7 100644 --- a/apps/api/tests/conftest.py +++ b/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]: diff --git a/apps/api/tests/test_api_workflows.py b/apps/api/tests/test_api_workflows.py index 15e0f2b..2da3005 100644 --- a/apps/api/tests/test_api_workflows.py +++ b/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 diff --git a/apps/api/tests/test_auth.py b/apps/api/tests/test_auth.py index 8779786..67f19e7 100644 --- a/apps/api/tests/test_auth.py +++ b/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 diff --git a/apps/api/tests/test_engine.py b/apps/api/tests/test_engine.py index 7e7b611..7081f22 100644 --- a/apps/api/tests/test_engine.py +++ b/apps/api/tests/test_engine.py @@ -550,3 +550,123 @@ async def test_completed_events_have_duration_ms(db_session, session_factory, te for e in events: assert e.duration_ms is not None assert e.duration_ms >= 0 + + +# ─── Test: transform output shaped like {"error": ...} must not be misread +# as an executor failure (see NODE_ERROR_KEY in scheduler.py) ──────────── + + +def _transform_error_shaped_output_graph() -> dict: + """start → transform (whose own mapping produces {"error": "..."}) → end""" + return { + "nodes": [ + {"id": "start", "type": "start", "position": {"x": 0, "y": 0}, "data": {"label": "S", "config": {}}}, + { + "id": "tr", + "type": "transform", + "position": {"x": 0, "y": 100}, + "data": {"label": "T", "config": {"mappings": {"error": "Not a real failure"}}}, + }, + {"id": "end", "type": "end", "position": {"x": 0, "y": 200}, "data": {"label": "End", "config": {}}}, + ], + "edges": [ + {"id": "e1", "source": "start", "target": "tr"}, + {"id": "e2", "source": "tr", "target": "end"}, + ], + } + + +async def test_transform_output_shaped_like_error_dict_is_not_misclassified_as_failed( + db_session, session_factory, test_user_id +): + """A successful `transform` node whose mapping happens to produce + {"error": "..."} must still mark the run "completed" — this exact shape + collided with the old bare {"error": ...} failure heuristic in runner.py. + """ + wf = Workflow( + owner_id=test_user_id, + name="error-shaped-output-test", + graph=_transform_error_shaped_output_graph(), + ) + db_session.add(wf) + await db_session.flush() + + run = WorkflowRun(workflow_id=wf.id, status="pending", trigger_payload={}) + db_session.add(run) + await db_session.commit() + + async with session_factory() as session: + await execute_run(run_id=run.id, session=session) + + async with session_factory() as s: + refreshed = await s.get(WorkflowRun, run.id) + assert refreshed.status == "completed" + assert refreshed.final_payload["tr"] == {"error": "Not a real failure"} + + +# ─── Test: `delay` node bounds (see MAX_DELAY_SECONDS in executors.py) ────── + + +def _delay_graph(seconds) -> dict: + """start → delay(seconds) → end""" + return { + "nodes": [ + {"id": "start", "type": "start", "position": {"x": 0, "y": 0}, "data": {"label": "S", "config": {}}}, + {"id": "delay-x", "type": "delay", "position": {"x": 0, "y": 100}, "data": {"label": "D", "config": {"seconds": seconds}}}, + {"id": "end", "type": "end", "position": {"x": 0, "y": 200}, "data": {"label": "End", "config": {}}}, + ], + "edges": [ + {"id": "e1", "source": "start", "target": "delay-x"}, + {"id": "e2", "source": "delay-x", "target": "end"}, + ], + } + + +async def test_delay_negative_seconds_fails_the_node(db_session, session_factory, test_user_id): + wf = Workflow(owner_id=test_user_id, name="negative-delay-test", graph=_delay_graph(-5)) + db_session.add(wf) + await db_session.flush() + + run = WorkflowRun(workflow_id=wf.id, status="pending", trigger_payload={}) + db_session.add(run) + await db_session.commit() + + async with session_factory() as session: + await execute_run(run_id=run.id, session=session) + + async with session_factory() as s: + refreshed = await s.get(WorkflowRun, run.id) + assert refreshed.status == "failed" + + +async def test_delay_seconds_are_capped_at_max(db_session, session_factory, test_user_id, monkeypatch): + """A `seconds` value above MAX_DELAY_SECONDS is clamped, not honored as-is. + + MAX_DELAY_SECONDS is monkeypatched down to a tiny value so the test itself + doesn't have to sleep for real minutes to prove the clamp happened — + what matters is that the executed `waited` value equals the cap, not the + requested value. + """ + import app.engine.executors as executors_module + + monkeypatch.setattr(executors_module, "MAX_DELAY_SECONDS", 0.05) + + wf = Workflow( + owner_id=test_user_id, + name="oversized-delay-test", + graph=_delay_graph(1000), + ) + db_session.add(wf) + await db_session.flush() + + run = WorkflowRun(workflow_id=wf.id, status="pending", trigger_payload={}) + db_session.add(run) + await db_session.commit() + + async with session_factory() as session: + await execute_run(run_id=run.id, session=session) + + async with session_factory() as s: + refreshed = await s.get(WorkflowRun, run.id) + assert refreshed.status == "completed" + assert refreshed.final_payload["delay-x"]["waited"] == 0.05