From 3f3feadb4865276c45ab889a47eb8ef202d921a4 Mon Sep 17 00:00:00 2001 From: taly Date: Wed, 1 Jul 2026 10:37:55 +0300 Subject: [PATCH] fix: skip unknown transform ops instead of silently running base64_decode _normalize_transform_op silently rewrote an unknown/unsupported transform op to base64_decode, so a flow authored with a newer op ran the WRONG operation against live 24/7 traffic and returned plausible-but-wrong output. execute_transform_ops now SKIPS an unknown op with a structured, machine- readable warning (logged as TRANSFORM_OP_UNSUPPORTED and surfaced by the step handler) and continues; the op's output variable is left unset. _normalize_transform_op now raises on an unknown op as defense-in-depth. Adds 6 unit tests (op-level skip/no-downgrade, normalize-raises, TransformOp instance path, missing op/set, step-level no-halt, all-ops-unknown) and a severity-flagged RELEASE_NOTES entry with an audit note. Part of the cross-app FlowMap graceful-degradation strategy. Co-Authored-By: Claude Opus 4.8 --- RELEASE_NOTES.md | 15 +++++ flow_runner.py | 43 +++++++++++++- tests/unit/test_flow_runner.py | 102 +++++++++++++++++++++++++++++++++ 3 files changed, 159 insertions(+), 1 deletion(-) diff --git a/RELEASE_NOTES.md b/RELEASE_NOTES.md index 748e24f..c9d6d6a 100644 --- a/RELEASE_NOTES.md +++ b/RELEASE_NOTES.md @@ -1,3 +1,18 @@ +# Release Notes: FlowRunner CLI (Unreleased) + +## ⚠️ Behavior fix — severity: HIGH — unknown transform ops no longer silently mis-execute + +**What changed.** An unknown/unsupported transform `op` was previously **silently rewritten to `base64_decode`** and executed (`_normalize_transform_op`), so a flow authored with a newer/unrecognized op ran the WRONG operation against live traffic and produced plausible-but-wrong output. It is now **skipped with a machine-readable warning** rather than executed: +- `execute_transform_ops` logs a machine-readable `TRANSFORM_OP_UNSUPPORTED op=... set=... index=...` marker at ERROR level and records a structured entry in its returned `warnings[]` (`{"type": "unsupported_transform_op", "op": , "set": , "index": , "status": "skipped"}`), which the step handler surfaces to the run log. The op's output variable is left unset and the remaining ops run; the skip itself does not halt the flow. (If a *later* op or step hard-references the unset variable via `{{var}}`, that still resolves as an undefined reference and halts as usual, exactly as before — the skip neither invents new failures nor papers over genuinely missing data.) +- `_normalize_transform_op` now **raises** on an unknown op (defense-in-depth) instead of downgrading to `base64_decode`. +- `_execute_transform_step` surfaces each warning at WARNING level in the run log. + +**Audit note.** Run artifacts produced by earlier CLI versions may contain values from an unintended `base64_decode` of an unknown op's first argument. To audit, review flows using transform ops outside the supported set (`base64_encode/decode`, `jwt_encode/decode`, `json_set`, `math_add/sub/mul/div`, `to_number/string/boolean`, `boolean_not`); any output variable whose value looks like a base64 decode of an unexpected input may have been affected. + +This is part of the cross-app FlowMap graceful-degradation strategy (unknown step type / transform op / operator ⇒ skip-with-warning, never crash, never mis-execute). See the FlowRunner UI repo's `docs/flowmap-evolution.md` and `gotchas.md` "Cross-app FlowMap contract". + +--- + # Release Notes: FlowRunner CLI v1.2.0 ## Highlights diff --git a/flow_runner.py b/flow_runner.py index 6fc6944..7b36494 100644 --- a/flow_runner.py +++ b/flow_runner.py @@ -733,7 +733,10 @@ def _normalize_transform_op(op: Any) -> Dict[str, Any]: op_name = safe_op.get("op") if op_name not in _TRANSFORM_OP_DEFS: - op_name = "base64_decode" + # Never silently substitute a different op (historically base64_decode). Callers + # must skip unknown ops with a warning; this raise is defense-in-depth so no code + # path can execute the wrong operation against live traffic. + raise ValueError(f'Unsupported transform op "{op_name}".') defn = _TRANSFORM_OP_DEFS[op_name] raw_args = safe_op.get("args") @@ -1168,6 +1171,40 @@ def execute_transform_ops(ops: Any, context: Dict[str, Any], evaluate_path: Opti output = {"updatedVars": [], "warnings": []} ops_list = ops if isinstance(ops, list) else [] for index, op in enumerate(ops_list): + # Graceful degradation: an unknown/newer transform op is SKIPPED with a + # machine-readable warning rather than silently downgraded to base64_decode + # (which would run the wrong operation against live traffic) or raised (which + # would set flow_error and halt the whole flow). Known ops proceed normally. + if isinstance(op, TransformOp): + raw_op = op.model_dump() + elif isinstance(op, dict): + raw_op = op + else: + raw_op = {} + op_name = raw_op.get("op") + # Unknown-op detection mirrors _normalize_transform_op's check below; keep the + # two in sync if op-name normalization/aliasing is ever introduced. + if op_name not in _TRANSFORM_OP_DEFS: + set_hint = raw_op.get("set") if isinstance(raw_op.get("set"), str) else None + var_phrase = f'output variable "{set_hint}" left unset' if set_hint else "no output variable set" + output["warnings"].append({ + "type": "unsupported_transform_op", + "op": op_name, + "set": set_hint, + "index": index, + "status": "skipped", + "message": ( + f'Unsupported transform op "{op_name}" at position {index + 1}; ' + f'skipped ({var_phrase}). This runner may be older than the flow ' + f'that produced it.' + ), + }) + logger.error( + "TRANSFORM_OP_UNSUPPORTED op=%r set=%r index=%d - skipped, not executed " + "(refusing to silently substitute base64_decode).", + op_name, set_hint, index, + ) + continue normalized = _normalize_transform_op(op) set_name = normalized.get("set") if not isinstance(set_name, str) or not set_name: @@ -2809,6 +2846,10 @@ async def _execute_transform_step( ops = step.ops or [] logger.debug(f"{indent}User {user_id_log}: Transform {step_identifier}: Executing {len(ops)} ops.") output = execute_transform_ops(ops, context, evaluate_path=get_value_from_context) + for warning in output.get("warnings", []): + logger.warning( + f"{indent}User {user_id_log}: Transform {step_identifier}: {warning.get('message')}" + ) logger.debug(f"{indent}User {user_id_log}: Transform {step_identifier}: Updated vars {output.get('updatedVars', [])}.") except Exception as exc: logger.error( diff --git a/tests/unit/test_flow_runner.py b/tests/unit/test_flow_runner.py index ad11135..78f0aa1 100644 --- a/tests/unit/test_flow_runner.py +++ b/tests/unit/test_flow_runner.py @@ -54,12 +54,15 @@ def decorator(fn): LoopStep, ConditionStep, TransformStep, + TransformOp, ConditionData, Metrics, StartRequest, get_value_from_context, _MISSING, set_value_in_context, + execute_transform_ops, + _normalize_transform_op, logger as fr_logger, ) @@ -205,6 +208,105 @@ async def test_transform_step_updates_context(base_config, empty_flow): assert context["payload"]["exp"] == 110 +def test_execute_transform_ops_skips_unknown_op_without_downgrade(): + # An unknown/newer transform op must NOT be silently rewritten to base64_decode. + # It is skipped with a machine-readable warning; later known ops still run. + context: Dict[str, Any] = {} + ops = [ + # "SGVsbG8" base64url-decodes to "Hello". If the old bug downgrades this to + # base64_decode, context["decoded"] would become "Hello". It must stay unset. + {"op": "totally_unknown_future_op", "set": "decoded", "args": ["SGVsbG8"]}, + {"op": "math_add", "set": "sum", "args": [1, 2]}, + ] + output = execute_transform_ops(ops, context) + # unknown op skipped: variable never set (definitely not base64-decoded to "Hello") + assert "decoded" not in context + # subsequent known op still executed + assert context["sum"] == 3 + assert "sum" in output["updatedVars"] + assert "decoded" not in output["updatedVars"] + # a machine-readable warning was recorded for the skipped op + warnings = output["warnings"] + assert len(warnings) == 1 + w = warnings[0] + assert w["op"] == "totally_unknown_future_op" + assert w["set"] == "decoded" + assert w["status"] == "skipped" + + +def test_normalize_transform_op_raises_on_unknown_op(): + # Defense in depth: normalization must never silently substitute base64_decode. + with pytest.raises(ValueError): + _normalize_transform_op({"op": "nope_not_real", "set": "x", "args": ["SGVsbG8"]}) + + +@pytest.mark.asyncio +async def test_transform_step_skips_unknown_op_without_halting(base_config, empty_flow): + # Graceful degradation: an unknown op does not halt the flow (no flow_error is set) + # and known ops in the same step still apply. + runner = make_runner(base_config, empty_flow) + context: Dict[str, Any] = {} + step = TransformStep( + id="t2", + name="Transform", + type="transform", + ops=[ + {"op": "totally_unknown_future_op", "set": "decoded", "args": ["SGVsbG8"]}, + {"op": "math_add", "set": "sum", "args": [2, 3]}, + ], + ) + await runner._execute_transform_step(step, context, depth=0, user_id_log="test") + # no crash / no halt + assert get_value_from_context(context, "flow_error") is _MISSING + # unknown op did not run (not downgraded to base64_decode) + assert "decoded" not in context + # known op still ran + assert context["sum"] == 5 + + +def test_execute_transform_ops_skips_unknown_transformop_instance(): + # Exercises the model_dump() branch of the guard directly: an unknown op passed as a + # TransformOp model instance (the shape ops actually take after step validation). + context: Dict[str, Any] = {} + op = TransformOp.model_validate({"op": "unknown_model_op", "set": "decoded", "args": ["SGVsbG8"]}) + output = execute_transform_ops([op], context) + assert "decoded" not in context + assert output["updatedVars"] == [] + assert len(output["warnings"]) == 1 + assert output["warnings"][0]["op"] == "unknown_model_op" + assert output["warnings"][0]["status"] == "skipped" + + +def test_execute_transform_ops_handles_missing_op_name_and_set(): + # op is None (missing) and set is missing -> each skipped with a warning, no crash. + context: Dict[str, Any] = {} + output = execute_transform_ops([{"args": []}, {"op": "still_unknown"}], context) + assert output["updatedVars"] == [] + assert len(output["warnings"]) == 2 + assert output["warnings"][0]["op"] is None + assert output["warnings"][0]["set"] is None + + +@pytest.mark.asyncio +async def test_transform_step_all_ops_unknown_does_not_halt(base_config, empty_flow): + # Every op unknown: the whole step degrades to a no-op and the flow is not halted. + runner = make_runner(base_config, empty_flow) + context: Dict[str, Any] = {} + step = TransformStep( + id="t3", + name="Transform", + type="transform", + ops=[ + {"op": "unknown_a", "set": "a", "args": []}, + {"op": "unknown_b", "set": "b", "args": []}, + ], + ) + await runner._execute_transform_step(step, context, depth=0, user_id_log="test") + assert get_value_from_context(context, "flow_error") is _MISSING + assert "a" not in context + assert "b" not in context + + def test_extract_data_status_headers_and_body(base_config, empty_flow): runner = make_runner(base_config, empty_flow) ctx: Dict[str, Any] = {}