primitives/checks_runner.py: failing check command does not raise in saga step → fail-open execution (CWE-755)

Component: primitives/checks_runner.py — apply_checks_plan() (station v0.97).

The defect.
In primitives/checks_runner.py:137-158, pre-execution test checks wrap command execution inside saga.step("run_checks_%s" % alias, lambda: client.run(alias)).

When a check command fails (exit_code != 0), client.run() returns a result dictionary without raising an exception. Consequently the saga records status: "ok" on disk, allowing downstream sequential actions (e.g. production deployment) to proceed despite failed tests — a classic fail-open defect.

Reproduction (standalone harness).

import os, tempfile, shutil, sqlite3, json

try:
    from primitives.sagalog import Saga, LocalSagaStore
    from primitives import checks_runner as cr
except ImportError:  # Workbench layout
    from workbench.primitives.sagalog import Saga, LocalSagaStore
    from workbench.primitives import checks_runner as cr

ws = tempfile.mkdtemp(prefix="rc_checks_")
db_path = os.path.join(ws, "sagalog.db")
store = LocalSagaStore(db_path)
saga = Saga(name="production_deploy_pipeline", db=store)

class FailingChecksClient:
    def run(self, alias: str):
        # Simulates failing pre-deploy test suite
        return {"exit_code": 1, "output": "FAIL: test_security_auth assertion failed"}

client = FailingChecksClient()
plan = {"command_alias": "run_pytest"}

# 1. Execute check command under saga
res = cr.apply_checks_plan(plan, client, saga)

# 2. Query SQLite sagalog store on disk to inspect what was persisted
conn = sqlite3.connect(db_path)
cur = conn.cursor()
row = cur.execute("SELECT state, steps FROM sagalogs WHERE tracking_id = ?", (saga.tracking_id,)).fetchone()
conn.close()

persisted_state = row[0]
persisted_steps = json.loads(row[1])

print("Check Command exit_code:", res.get("exit_code"))
print("Check Result passed flag:", res.get("passed"))
print("Saga State on Disk:", persisted_state)
print("Saga Steps Persisted to Disk:", [{"label": s["label"], "status": s.get("status")} for s in persisted_steps])
print("Fail-Open Occurred (Saga step marked ok on disk):", 
      res.get("passed") is False and persisted_steps[0].get("status") == "ok")

shutil.rmtree(ws)

Expected raw output on vulnerable code:

Check Command exit_code: 1
Check Result passed flag: False
Saga State on Disk: STARTED
Saga Steps Persisted to Disk: [{'label': 'run_checks_run_pytest', 'status': 'ok'}]
Fail-Open Occurred (Saga step marked ok on disk): True

Scope (stated honestly).

• Causes pre-flight / pre-deployment checks to fail open instead of halting the saga workflow.
• Only relevant when checks are executed as an intermediate step inside a multi-step saga.

Fix.
Raise a ChecksError when exit_code != 0 so the saga records a failed step and halts downstream execution:

--- a/primitives/checks_runner.py
+++ b/primitives/checks_runner.py
@@ -155,5 +155,7 @@
                          % (len(out) - MAX_INLINE_OUTPUT) + out[-half:])
         res["output_truncated"] = True
     res["passed"] = (res["exit_code"] == 0)
+    if not res["passed"]:
+        raise ChecksError(f"Check command {alias!r} failed with exit code {res['exit_code']}")
     return res

Classification: CWE-755 (Improper Handling of Exceptional Conditions)