Merge pull request #222 from NOAA-GSL/staging

Security Patches

Author: Hackshaven

poetry.lock

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "zyra"
-version = "0.1.43"
+version = "0.1.44"
 description = "A tool to ingest data from various sources and formats, create imagery or video based on that data, and send the results to various locations for dissemination."
 authors = ["Eric Hackathorn <eric.j.hackathorn@noaa.gov>"]
 include = [
@@ -34,13 +34,13 @@ pyyaml = "^6.0.2"
 numpy = "^1.26"
 fastapi = { version = ">=0.120.2,<0.122.0", optional = true }
 uvicorn = { version = "^0.30.0", optional = true }
-python-multipart = { version = ">=0.0.18", optional = true }
+python-multipart = { version = ">=0.0.22", optional = true }
 redis = { version = "^5.0.0", optional = true }
 rq = { version = "^1.15.1", optional = true }
 python-magic = { version = "^0.4.27", optional = true }
 websockets = { version = "^11.0.3", optional = true }
 prompt_toolkit = { version = "^3.0.50", optional = true }
-guardrails-ai = { version = "^0.5.0", optional = true }
+guardrails-ai = { version = "^0.8.0", optional = true }
 google-auth = { version = "^2.29.0", optional = true }
 
 # Optional feature dependencies (installed via extras)

pyproject.toml

@@ -393,17 +393,28 @@ def flush(self):  # type: ignore[override]
         sys.stdout = old_stdout
 
 
+_DEFAULT_SUBPROCESS_TIMEOUT = 120
+
+
 def _run_cli_subprocess(
     argv: list[str], input_bytes: bytes | None
 ) -> tuple[int, bytes, str]:
     import subprocess
     import sys
 
-    proc = subprocess.run(
-        [sys.executable, "-m", "zyra.cli", *argv],
-        input=input_bytes or b"",
-        capture_output=True,
+    timeout = int(
+        os.getenv("ZYRA_CLI_SUBPROCESS_TIMEOUT", str(_DEFAULT_SUBPROCESS_TIMEOUT))
+        or _DEFAULT_SUBPROCESS_TIMEOUT
     )
+    try:
+        proc = subprocess.run(
+            [sys.executable, "-m", "zyra.cli", *argv],
+            input=input_bytes or b"",
+            capture_output=True,
+            timeout=timeout,
+        )
+    except subprocess.TimeoutExpired:
+        return 2, b"", f"subprocess timed out after {timeout}s"
     stderr = (proc.stderr or b"").decode("utf-8", errors="ignore")
     return int(proc.returncode), proc.stdout or b"", stderr
 

src/zyra/pipeline_runner.py

@@ -50,7 +50,7 @@ def _load_guard(self):
         if self._guard:
             return self._guard
         path = Path(self.schema_path)
-        self._guard = _guardrails.Guard.from_rail(str(path))  # type: ignore[attr-defined]
+        self._guard = _guardrails.Guard.for_rail(str(path))  # type: ignore[attr-defined]
         return self._guard
 
     def validate(
@@ -61,25 +61,28 @@ def validate(
         for key, value in outputs.items():
             raw = value if isinstance(value, str) else json.dumps(value)
             try:
-                # guard.parse returns the validated structure (string or dict)
-                result = guard.parse(raw)
+                # guard.validate returns a ValidationOutcome with
+                # .validation_passed (bool) and .validated_output
+                result = guard.validate(raw)
             except Exception as exc:
                 msg = f"guardrails validation failed for {agent.spec.id}:{key}: {exc}"
                 if self.strict:
                     raise RuntimeError(msg) from exc
                 LOG.warning("%s", msg)
                 validated_value = value
             else:
-                validated_value = getattr(result, "validated_output", result)
-                # If validation failed but strict mode is enabled, raise
-                if (
-                    hasattr(result, "validation_passed")
-                    and not result.validation_passed
-                    and self.strict
-                ):
-                    raise RuntimeError(
-                        f"guardrails validation failed for {agent.spec.id}:{key}"
+                passed = getattr(result, "validation_passed", True)
+                if not passed:
+                    msg = (
+                        f"guardrails validation did not pass for "
+                        f"{agent.spec.id}:{key}"
                     )
+                    if self.strict:
+                        raise RuntimeError(msg)
+                    LOG.warning("%s – falling back to original value", msg)
+                    validated_value = value
+                else:
+                    validated_value = getattr(result, "validated_output", result)
             validated[key] = validated_value
         return validated
 

src/zyra/swarm/guardrails.py

@@ -1959,8 +1959,34 @@ def _run_guardrails(schema_path: str, manifest: dict[str, Any]) -> None:
             "guardrails library not installed; pip install guardrails-ai"
         ) from exc
     text = Path(schema_path).read_text(encoding="utf-8")
-    guard = Guard.from_rail(text)  # type: ignore
-    guard.parse(json.dumps(manifest, sort_keys=True))
+    guard = Guard.for_rail_string(text)  # type: ignore
+    result = guard.validate(json.dumps(manifest, sort_keys=True))
+    if hasattr(result, "validation_passed") and not result.validation_passed:
+        details = _guardrails_failure_details(result)
+        detail_suffix = f": {details}" if details else ""
+        raise RuntimeError(
+            f"guardrails validation did not pass for {schema_path}{detail_suffix}"
+        )
+
+
+def _guardrails_failure_details(result: Any) -> str:
+    """Extract a human-readable failure summary from a ValidationOutcome."""
+    parts: list[str] = []
+    for attr in ("error", "error_message"):
+        val = getattr(result, attr, None)
+        if isinstance(val, str) and val.strip():
+            parts.append(val.strip())
+            break
+    errors = getattr(result, "validation_errors", None)
+    if isinstance(errors, list):
+        for err in errors[:5]:
+            text = str(err).strip() if err else ""
+            if text:
+                parts.append(text)
+    reask = getattr(result, "reask", None)
+    if reask is not None and not parts:
+        parts.append(f"reask={reask}")
+    return "; ".join(parts)
 
 
 def _load_llm_client():  # pragma: no cover - environment dependent

src/zyra/swarm/planner.py

@@ -4,6 +4,8 @@
 import json
 from argparse import Namespace
 
+import pytest
+
 import zyra.swarm.value_engine as value_engine
 from zyra.swarm import planner as planner_cli
 from zyra.swarm.planner import (
@@ -20,6 +22,7 @@
     _map_to_capabilities,
     _normalize_args_for_command,
     _propagate_inferred_args,
+    _run_guardrails,
     _scan_frames_plan_details,
     _strip_internal_fields,
     _validate_manifest,
@@ -836,3 +839,135 @@ class FakeCaps:
     assert specs[0].command == "ftp"
     assert specs[1].behavior == "proposal"
     assert specs[1].metadata["proposal_options"] == ["swarm", "describe"]
+
+
+# --- guardrails integration via planner ---
+
+_PASS_RAIL = """\
+<rail version="0.1">
+
+<output>
+    <list name="agents" description="Pipeline agent definitions">
+        <object>
+            <string name="id" />
+            <string name="stage" />
+        </object>
+    </list>
+</output>
+
+<prompt>
+    Validate plan agents.
+    {{#block hidden=True}}
+    {{input}}
+    {{/block}}
+</prompt>
+
+</rail>
+"""
+
+_STRICT_RAIL = """\
+<rail version="0.1">
+
+<output>
+    <object name="plan">
+        <list name="agents">
+            <object>
+                <string name="id" />
+                <string name="stage" />
+                <integer name="priority" description="required priority field" />
+            </object>
+        </list>
+    </object>
+</output>
+
+<prompt>
+    Validate plan agents have a priority field.
+    {{#block hidden=True}}
+    {{input}}
+    {{/block}}
+</prompt>
+
+</rail>
+"""
+
+
+@pytest.mark.guardrails
+def test_run_guardrails_validates_manifest(tmp_path, monkeypatch):
+    """_run_guardrails should accept a valid manifest without raising."""
+    pytest.importorskip("guardrails")
+    monkeypatch.setenv("OTEL_SDK_DISABLED", "true")
+    schema = tmp_path / "plan.rail"
+    schema.write_text(_PASS_RAIL, encoding="utf-8")
+    manifest = {
+        "agents": [
+            {"id": "fetch", "stage": "acquire"},
+            {"id": "narrate", "stage": "narrate"},
+        ]
+    }
+    # Should not raise
+    _run_guardrails(str(schema), manifest)
+
+
+@pytest.mark.guardrails
+def test_run_guardrails_rejects_invalid_manifest(tmp_path, monkeypatch):
+    """_run_guardrails should raise when the manifest fails validation."""
+    pytest.importorskip("guardrails")
+    monkeypatch.setenv("OTEL_SDK_DISABLED", "true")
+    schema = tmp_path / "strict.rail"
+    schema.write_text(_STRICT_RAIL, encoding="utf-8")
+    # Manifest lacks the required "priority" integer field and is not
+    # wrapped in a "plan" key, so validation_passed will be False.
+    manifest = {
+        "agents": [
+            {"id": "fetch", "stage": "acquire"},
+        ]
+    }
+    with pytest.raises(RuntimeError, match="validation did not pass"):
+        _run_guardrails(str(schema), manifest)
+
+
+@pytest.mark.guardrails
+def test_cmd_plan_with_guardrails_flag(tmp_path, capsys, monkeypatch):
+    """The --guardrails CLI flag should invoke validation without error."""
+    pytest.importorskip("guardrails")
+    monkeypatch.setenv("OTEL_SDK_DISABLED", "true")
+    schema = tmp_path / "plan.rail"
+    schema.write_text(_PASS_RAIL, encoding="utf-8")
+    ns = Namespace(
+        intent="mock swarm plan",
+        intent_file=None,
+        output="-",
+        guardrails=str(schema),
+        strict=False,
+        memory=None,
+        no_clarify=True,
+        verbose=False,
+    )
+    rc = planner_cli._cmd_plan(ns)
+    assert rc == 0
+    out = capsys.readouterr().out
+    payload = json.loads(out)
+    assert payload["agents"][0]["stage"] == "simulate"
+
+
+@pytest.mark.guardrails
+def test_cmd_plan_strict_guardrails_rejects(tmp_path, capsys, monkeypatch):
+    """--guardrails + --strict should return exit code 2 on failure."""
+    pytest.importorskip("guardrails")
+    monkeypatch.setenv("OTEL_SDK_DISABLED", "true")
+    schema = tmp_path / "strict.rail"
+    schema.write_text(_STRICT_RAIL, encoding="utf-8")
+    ns = Namespace(
+        intent="mock swarm plan",
+        intent_file=None,
+        output="-",
+        guardrails=str(schema),
+        strict=True,
+        memory=None,
+        no_clarify=True,
+        verbose=False,
+    )
+    rc = planner_cli._cmd_plan(ns)
+    assert rc == 2
+    err = capsys.readouterr().err
+    assert "guardrails validation failed" in err

tests/swarm/test_planner.py

@@ -86,13 +86,15 @@ def test_log_events_and_dump_memory(tmp_path, capsys) -> None:
                 "id": "narrate",
                 "stage": "narrate",
                 "command": "describe",
+                "behavior": "mock",
                 "outputs": ["summary"],
                 "args": {"topic": "demo"},
             },
             {
                 "id": "export",
                 "stage": "disseminate",
                 "command": "local",
+                "behavior": "mock",
                 "stdin_from": "summary",
                 "outputs": ["artifact"],
                 "args": {"input": "-", "path": str(tmp_path / "out.txt")},