From e93f22285e8a3d61282a958055f2c990015255f5 Mon Sep 17 00:00:00 2001 From: Jordan Umusu Date: Fri, 31 Jul 2026 17:52:03 -0400 Subject: [PATCH 1/5] fix(chat): expression resolution hardening --- tests/unit/test_executor_expression_policy.py | 182 ++++++ tests/unit/test_executor_service.py | 543 ++++++++++++++++++ tracecat/executor/expression_policy.py | 215 +++++++ tracecat/executor/service.py | 74 ++- 4 files changed, 1006 insertions(+), 8 deletions(-) create mode 100644 tests/unit/test_executor_expression_policy.py create mode 100644 tracecat/executor/expression_policy.py diff --git a/tests/unit/test_executor_expression_policy.py b/tests/unit/test_executor_expression_policy.py new file mode 100644 index 000000000..b6da4893a --- /dev/null +++ b/tests/unit/test_executor_expression_policy.py @@ -0,0 +1,182 @@ +import pytest + +from tracecat.exceptions import TracecatExpressionError +from tracecat.executor.expression_policy import ( + POLICY_MAP, + ActionParameter, + ExpressionPolicy, + expression_policy, + partition_action_args, + redact_secret_expressions, +) +from tracecat.secrets.constants import MASK_VALUE + + +def test_action_parameter_policy_scope_is_explicit() -> None: + preserved = { + parameter + for parameter, policy in POLICY_MAP.items() + if policy is ExpressionPolicy.PRESERVE + } + redacted = { + parameter + for parameter, policy in POLICY_MAP.items() + if policy is ExpressionPolicy.REDACT_SECRETS + } + + assert preserved == { + ActionParameter( + action="core.workflow.edit_workflow", + parameter="patch_ops", + ), + ActionParameter( + action="core.workflow.create_workflow", + parameter="definition_yaml", + ), + } + assert redacted == { + ActionParameter(action="core.cases.create_case", parameter="summary"), + ActionParameter(action="core.cases.create_case", parameter="description"), + ActionParameter(action="core.cases.create_case", parameter="fields"), + ActionParameter(action="core.cases.create_case", parameter="payload"), + ActionParameter(action="core.cases.update_case", parameter="summary"), + ActionParameter(action="core.cases.update_case", parameter="description"), + ActionParameter(action="core.cases.update_case", parameter="fields"), + ActionParameter(action="core.cases.update_case", parameter="payload"), + ActionParameter(action="core.cases.create_comment", parameter="content"), + ActionParameter(action="core.cases.reply_to_comment", parameter="content"), + ActionParameter(action="core.cases.update_comment", parameter="content"), + ActionParameter(action="core.table.insert_row", parameter="row_data"), + ActionParameter(action="core.table.insert_rows", parameter="rows_data"), + ActionParameter(action="core.table.update_row", parameter="row_data"), + ActionParameter(action="core.cases.insert_row", parameter="row"), + ActionParameter(action="ai.agent.create_preset", parameter="instructions"), + ActionParameter(action="ai.agent.update_preset", parameter="instructions"), + } + + +def test_expression_policy_requires_exact_action_parameter_pair() -> None: + assert ( + expression_policy("core.workflow.edit_workflow", "patch_ops") + is ExpressionPolicy.PRESERVE + ) + assert ( + expression_policy("core.workflow.edit_workflow", "workflow_id") + is ExpressionPolicy.RESOLVE + ) + assert ( + expression_policy("core.transform.reshape", "patch_ops") + is ExpressionPolicy.RESOLVE + ) + + +def test_redact_secret_expressions_preserves_non_secret_templates() -> None: + value = ( + "Host: ${{ VARS.api.host }}, " + "token: ${{ SECRETS.api.TOKEN }}, " + "time: ${{ FN.now() }}, " + "result: ${{ ACTIONS.lookup.result }}" + ) + + assert redact_secret_expressions(value) == ( + f"Host: ${{{{ VARS.api.host }}}}, token: {MASK_VALUE}, " + "time: ${{ FN.now() }}, result: ${{ ACTIONS.lookup.result }}" + ) + + +def test_redact_secret_expressions_replaces_complex_occurrence() -> None: + value = "${{ FN.to_base64(SECRETS.api.TOKEN + VARS.api.suffix) }}" + + assert redact_secret_expressions(value) == MASK_VALUE + + +def test_redact_secret_expressions_uses_ast_dependencies() -> None: + value = ( + "Plain SECRETS.api.TOKEN; " + "literal ${{ 'SECRETS.api.TOKEN' }}; " + "reference ${{ SECRETS.api.TOKEN }}" + ) + + assert redact_secret_expressions(value) == ( + "Plain SECRETS.api.TOKEN; " + "literal ${{ 'SECRETS.api.TOKEN' }}; " + f"reference {MASK_VALUE}" + ) + + +def test_redact_secret_expressions_rejects_secret_dependent_keys() -> None: + value = { + "${{ SECRETS.api.KEY }}": 1, + "${{ SECRETS.other.KEY }}": 2, + } + + with pytest.raises(TracecatExpressionError) as exc_info: + redact_secret_expressions(value) + + assert exc_info.value.detail == {"code": "secret_expression_in_key"} + + +def test_redact_secret_expressions_recurses_through_values() -> None: + value = { + "keep ${{ VARS.api.key }}": [ + "keep ${{ VARS.api.value }}", + {"nested": "${{ SECRETS.api.TOKEN }}"}, + ] + } + + assert redact_secret_expressions(value) == { + "keep ${{ VARS.api.key }}": [ + "keep ${{ VARS.api.value }}", + {"nested": MASK_VALUE}, + ] + } + + +def test_partition_redacts_before_evaluation_and_restores_order() -> None: + args = { + "case_id": "${{ VARS.case.id }}", + "content": "Host ${{ VARS.api.host }}, token ${{ SECRETS.api.TOKEN }}", + "workflow_id": None, + } + + partitioned = partition_action_args("core.cases.create_comment", args) + + assert partitioned.resolvable == { + "case_id": "${{ VARS.case.id }}", + "content": f"Host ${{{{ VARS.api.host }}}}, token {MASK_VALUE}", + "workflow_id": None, + } + assert partitioned.merge( + { + "case_id": "case-123", + "content": f"Host example.com, token {MASK_VALUE}", + "workflow_id": None, + } + ) == { + "case_id": "case-123", + "content": f"Host example.com, token {MASK_VALUE}", + "workflow_id": None, + } + + +def test_partition_restores_preserved_subtree_without_reordering() -> None: + source = { + "${{ VARS.source.key }}": ["${{ SECRETS.source.TOKEN }}"], + } + args = { + "workflow_id": "${{ VARS.runtime.workflow_id }}", + "patch_ops": source, + "validate_only": False, + } + + partitioned = partition_action_args("core.workflow.edit_workflow", args) + + assert partitioned.resolvable == { + "workflow_id": "${{ VARS.runtime.workflow_id }}", + "validate_only": False, + } + assert partitioned.merge({"workflow_id": "wf-123", "validate_only": False}) == { + "workflow_id": "wf-123", + "patch_ops": source, + "validate_only": False, + } diff --git a/tests/unit/test_executor_service.py b/tests/unit/test_executor_service.py index 1febd8711..36c5024de 100644 --- a/tests/unit/test_executor_service.py +++ b/tests/unit/test_executor_service.py @@ -1,4 +1,7 @@ +import uuid +from collections.abc import Mapping from contextlib import asynccontextmanager +from datetime import UTC, datetime from uuid import UUID import pytest @@ -10,10 +13,23 @@ ) from tracecat.auth.types import Role +from tracecat.dsl.common import create_default_execution_context +from tracecat.dsl.schemas import ActionStatement, RunActionInput, RunContext from tracecat.exceptions import TracecatCredentialsError from tracecat.executor import service as executor_service +from tracecat.executor.schemas import ( + ActionImplementation, + ExecutorResultSuccess, + ResolvedContext, +) +from tracecat.executor.secret_preprocessors import SecretEnvProjection +from tracecat.executor.service import prepare_resolved_context +from tracecat.identifiers import InternalServiceID +from tracecat.identifiers.workflow import WorkflowUUID, generate_exec_id from tracecat.integrations.enums import OAuthGrantType +from tracecat.registry.lock.types import RegistryLock from tracecat.secrets import secrets_manager +from tracecat.secrets.constants import MASK_VALUE def test_flatten_secrets_supports_runtime_scalar_entries() -> None: @@ -378,3 +394,530 @@ async def test_invoke_once_offloads_root_secret_masking(mocker): action_result, masks={"secret"}, ) + + +def _expression_policy_role(service_id: InternalServiceID) -> Role: + return Role( + type="service", + organization_id=UUID(int=1), + workspace_id=UUID(int=2), + service_id=service_id, + ) + + +def _expression_policy_input( + action_name: str, args: Mapping[str, object] +) -> RunActionInput: + wf_id = WorkflowUUID.new_uuid4() + return RunActionInput( + task=ActionStatement(ref="a", action=action_name, args=args), + exec_context=create_default_execution_context(), + run_context=RunContext( + wf_id=wf_id, + wf_exec_id=generate_exec_id(wf_id), + wf_run_id=uuid.uuid4(), + environment="default", + logical_time=datetime.now(UTC), + ), + registry_lock=RegistryLock( + origins={"tracecat_registry": "v1"}, + actions={action_name: "tracecat_registry"}, + ), + ) + + +def _patch_expression_policy_resolution( + mocker, + *, + action_name: str, + action_secrets: set[RegistrySecretType], + fetched_secrets: dict[str, dict[str, str]], + workspace_variables: dict[str, dict[str, str]], +): + """Stub registry and credential IO around argument expression handling.""" + mocker.patch.object( + executor_service.registry_resolver, + "resolve_action", + new=mocker.AsyncMock( + return_value=ActionImplementation( + type="udf", + action_name=action_name, + module="tracecat_registry.integrations.core.transform", + name="reshape", + ) + ), + ) + mocker.patch.object( + executor_service.registry_resolver, + "collect_action_secrets_from_manifest", + new=mocker.AsyncMock(return_value=action_secrets), + ) + get_action_secrets = mocker.patch.object( + executor_service.secrets_manager, + "get_action_secrets", + new=mocker.AsyncMock(return_value=fetched_secrets), + ) + get_workspace_variables = mocker.patch.object( + executor_service, + "get_workspace_variables", + new=mocker.AsyncMock(return_value=workspace_variables), + ) + mocker.patch.object( + executor_service, + "_mint_action_executor_token", + return_value="token", + ) + project_secret_env = mocker.patch.object( + executor_service, + "project_secret_env", + new=mocker.AsyncMock( + return_value=SecretEnvProjection( + env={"TOKEN": "runtime-secret"}, + mask_values={"runtime-secret"}, + ) + ), + ) + return get_action_secrets, get_workspace_variables, project_secret_env + + +@pytest.mark.parametrize("service_id", ["tracecat-executor", "tracecat-mcp"]) +@pytest.mark.parametrize( + ("action_name", "preserved_parameter", "runtime_parameter"), + [ + ("core.workflow.edit_workflow", "patch_ops", "workflow_id"), + ("core.workflow.create_workflow", "definition_yaml", "title"), + ], +) +@pytest.mark.anyio +async def test_prepare_resolved_context_preserves_only_mapped_parameter( + mocker, + service_id: InternalServiceID, + action_name: str, + preserved_parameter: str, + runtime_parameter: str, +): + """Mapped workflow source stays literal for workflow and agent callers.""" + preserved_source: object + if preserved_parameter == "patch_ops": + preserved_source = [ + { + "op": "add", + "path": "/definition/actions/-", + "value": { + "${{ VARS.source.key }}": [ + "${{ SECRETS.source.TOKEN }}", + "${{ FN.now() }}", + ] + }, + } + ] + else: + preserved_source = ( + "definition:\n" + " actions:\n" + " - args:\n" + " token: ${{ SECRETS.source.TOKEN }}\n" + " generated_at: ${{ FN.now() }}\n" + ) + args = { + runtime_parameter: ("${{ SECRETS.runtime.TOKEN }}:${{ VARS.runtime.value }}"), + preserved_parameter: preserved_source, + } + action_secrets: set[RegistrySecretType] = { + RegistrySecret(name="declared", keys=["KEY"], optional=False) + } + get_action_secrets, get_workspace_variables, _ = ( + _patch_expression_policy_resolution( + mocker, + action_name=action_name, + action_secrets=action_secrets, + fetched_secrets={ + "runtime": {"TOKEN": "runtime-secret"}, + "declared": {"KEY": "declared-secret"}, + }, + workspace_variables={"runtime": {"value": "runtime-variable"}}, + ) + ) + mocker.patch.object( + executor_service.config, + "TRACECAT__UNSAFE_DISABLE_SM_MASKING", + False, + ) + + prepared = await prepare_resolved_context( + input=_expression_policy_input(action_name, args), + role=_expression_policy_role(service_id), + ) + + assert prepared.resolved_context.evaluated_args == { + runtime_parameter: "runtime-secret:runtime-variable", + preserved_parameter: preserved_source, + } + assert get_action_secrets.await_args.kwargs == { + "secret_exprs": {"runtime.TOKEN"}, + "action_secrets": action_secrets, + } + assert get_workspace_variables.await_args.kwargs["variable_exprs"] == {"runtime"} + assert prepared.mask_values == {"runtime-secret"} + + +@pytest.mark.parametrize("service_id", ["tracecat-executor", "tracecat-mcp"]) +@pytest.mark.anyio +async def test_prepare_resolved_context_redacts_secrets_before_collection( + mocker, + service_id: InternalServiceID, +): + """Durable content resolves safe expressions without fetching direct secrets.""" + action_name = "core.cases.create_comment" + action_secrets: set[RegistrySecretType] = { + RegistrySecret(name="declared", keys=["KEY"], optional=False) + } + get_action_secrets, get_workspace_variables, _ = ( + _patch_expression_policy_resolution( + mocker, + action_name=action_name, + action_secrets=action_secrets, + fetched_secrets={ + "runtime": {"TOKEN": "runtime-secret"}, + "declared": {"KEY": "declared-secret"}, + }, + workspace_variables={"runtime": {"value": "api.example.com"}}, + ) + ) + args = { + "case_id": "${{ SECRETS.runtime.TOKEN }}", + "content": ( + "Host: ${{ VARS.runtime.value }}, " + "token: ${{ SECRETS.source.TOKEN }}, " + "encoded: " + "${{ FN.to_base64(SECRETS.source.TOKEN + VARS.source.suffix) }}" + ), + } + + prepared = await prepare_resolved_context( + input=_expression_policy_input(action_name, args), + role=_expression_policy_role(service_id), + ) + + assert prepared.resolved_context.evaluated_args == { + "case_id": "runtime-secret", + "content": ( + f"Host: api.example.com, token: {MASK_VALUE}, encoded: {MASK_VALUE}" + ), + } + assert get_action_secrets.await_args.kwargs == { + "secret_exprs": {"runtime.TOKEN"}, + "action_secrets": action_secrets, + } + assert get_workspace_variables.await_args.kwargs["variable_exprs"] == {"runtime"} + + +@pytest.mark.parametrize( + ("action_name", "parameter"), + [ + ("core.transform.reshape", "patch_ops"), + ("core.transform.reshape", "content"), + ("core.workflow.edit_workflow", "workflow_id"), + ("core.workflow.create_workflow", "title"), + ], +) +@pytest.mark.anyio +async def test_prepare_resolved_context_resolves_unmapped_parameters( + mocker, + action_name: str, + parameter: str, +): + """Policy matching requires the exact action and parameter pair.""" + get_action_secrets, get_workspace_variables, _ = ( + _patch_expression_policy_resolution( + mocker, + action_name=action_name, + action_secrets=set(), + fetched_secrets={"runtime": {"TOKEN": "runtime-secret"}}, + workspace_variables={}, + ) + ) + args = {parameter: "${{ SECRETS.runtime.TOKEN }}"} + + prepared = await prepare_resolved_context( + input=_expression_policy_input(action_name, args), + role=_expression_policy_role("tracecat-executor"), + ) + + assert prepared.resolved_context.evaluated_args == {parameter: "runtime-secret"} + assert get_action_secrets.await_args.kwargs["secret_exprs"] == {"runtime.TOKEN"} + assert get_workspace_variables.await_args.kwargs["variable_exprs"] == set() + + +@pytest.mark.parametrize( + ("step_action", "step_args", "source_value", "evaluated_value", "expected_args"), + [ + ( + "core.cases.create_comment", + { + "case_id": "case-123", + "content": ("Host ${{ VARS.runtime.host }}, token ${{ inputs.value }}"), + }, + "${{ SECRETS.runtime.TOKEN }}", + "runtime-secret", + { + "case_id": "case-123", + "content": f"Host api.example.com, token {MASK_VALUE}", + }, + ), + ( + "core.workflow.edit_workflow", + { + "workflow_id": "wf-123", + "patch_ops": "${{ inputs.value }}", + }, + [ + { + "op": "add", + "path": "/definition/actions/-", + "value": "${{ SECRETS.runtime.TOKEN }}", + } + ], + [ + { + "op": "add", + "path": "/definition/actions/-", + "value": "runtime-secret", + } + ], + { + "workflow_id": "wf-123", + "patch_ops": [ + { + "op": "add", + "path": "/definition/actions/-", + "value": "${{ SECRETS.runtime.TOKEN }}", + } + ], + }, + ), + ], +) +@pytest.mark.anyio +async def test_template_step_applies_target_action_expression_policy( + mocker, + step_action: str, + step_args: dict[str, object], + source_value: object, + evaluated_value: object, + expected_args: dict[str, object], +): + """Template input source reaches the target action's policy boundary.""" + template_action = "testing.policy_wrapper" + action_input = _expression_policy_input( + template_action, + {"value": source_value}, + ) + role = _expression_policy_role("tracecat-executor") + parent_resolved = ResolvedContext( + secrets={"runtime": {"TOKEN": "runtime-secret"}}, + variables={"runtime": {"host": "api.example.com"}}, + action_impl=ActionImplementation( + type="template", + action_name=template_action, + template_definition={ + "name": "policy_wrapper", + "namespace": "testing", + "title": "Policy wrapper", + "description": "Exercises a protected sink", + "display_group": "Testing", + "expects": {}, + "steps": [ + { + "ref": "persist", + "action": step_action, + "args": step_args, + } + ], + "returns": "${{ steps.persist.result }}", + }, + ), + evaluated_args={"value": evaluated_value}, + workspace_id=str(role.workspace_id), + workflow_id=str(action_input.run_context.wf_id), + run_id=str(action_input.run_context.wf_run_id), + executor_token="parent-token", + ) + mocker.patch.object( + executor_service.registry_resolver, + "resolve_action", + new=mocker.AsyncMock( + return_value=ActionImplementation(type="udf", action_name=step_action) + ), + ) + mocker.patch.object( + executor_service, + "_mint_action_executor_token", + return_value="step-token", + ) + backend = mocker.Mock() + backend.execute = mocker.AsyncMock( + return_value=ExecutorResultSuccess(result={"persisted": True}) + ) + + result = await executor_service._execute_template_action( + backend=backend, + input=action_input, + ctx=executor_service.DispatchActionContext(role=role), + resolved_context=parent_resolved, + timeout=30, + source_args={"value": source_value}, + ) + + assert result == {"persisted": True} + step_resolved = backend.execute.await_args.kwargs["resolved_context"] + assert step_resolved.evaluated_args == expected_args + + +def _policy_wrapper_resolved( + action_input: RunActionInput, + role: Role, + *, + steps: list[dict[str, object]], + evaluated_args: dict[str, object], + variables: dict[str, dict[str, str]], +) -> ResolvedContext: + return ResolvedContext( + secrets={}, + variables=variables, + action_impl=ActionImplementation( + type="template", + action_name="testing.policy_wrapper", + template_definition={ + "name": "policy_wrapper", + "namespace": "testing", + "title": "Policy wrapper", + "description": "Exercises a protected sink", + "display_group": "Testing", + "expects": {}, + "steps": steps, + "returns": "done", + }, + ), + evaluated_args=evaluated_args, + workspace_id=str(role.workspace_id), + workflow_id=str(action_input.run_context.wf_id), + run_id=str(action_input.run_context.wf_run_id), + executor_token="parent-token", + ) + + +@pytest.mark.anyio +async def test_template_step_result_stays_inert_in_redact_parameter(mocker): + """Materialized step results are grafted as data, never re-expanded as source.""" + action_input = _expression_policy_input("testing.policy_wrapper", {}) + role = _expression_policy_role("tracecat-executor") + parent_resolved = _policy_wrapper_resolved( + action_input, + role, + steps=[ + { + "ref": "fetch", + "action": "core.transform.reshape", + "args": {"value": "external"}, + }, + { + "ref": "persist", + "action": "core.cases.create_comment", + "args": { + "case_id": "case-123", + "content": "Summary: ${{ steps.fetch.result.note }}", + }, + }, + ], + evaluated_args={}, + variables={"runtime": {"host": "api.example.com"}}, + ) + mocker.patch.object( + executor_service.registry_resolver, + "resolve_action", + new=mocker.AsyncMock( + return_value=ActionImplementation( + type="udf", action_name="core.transform.reshape" + ) + ), + ) + mocker.patch.object( + executor_service, "_mint_action_executor_token", return_value="step-token" + ) + backend = mocker.Mock() + backend.execute = mocker.AsyncMock( + side_effect=[ + ExecutorResultSuccess(result={"note": "${{ VARS.runtime.host }}"}), + ExecutorResultSuccess(result={"persisted": True}), + ] + ) + + await executor_service._execute_template_action( + backend=backend, + input=action_input, + ctx=executor_service.DispatchActionContext(role=role), + resolved_context=parent_resolved, + timeout=30, + source_args={}, + ) + + persist_resolved = backend.execute.await_args_list[1].kwargs["resolved_context"] + assert persist_resolved.evaluated_args == { + "case_id": "case-123", + "content": "Summary: ${{ VARS.runtime.host }}", + } + + +@pytest.mark.anyio +async def test_template_step_skips_source_expansion_without_policy_parameters(mocker): + """Pure-RESOLVE UDF steps never pay for the source expansion walk.""" + action_input = _expression_policy_input("testing.policy_wrapper", {}) + role = _expression_policy_role("tracecat-executor") + parent_resolved = _policy_wrapper_resolved( + action_input, + role, + steps=[ + { + "ref": "transform", + "action": "core.transform.reshape", + "args": {"value": "${{ inputs.value }}"}, + } + ], + evaluated_args={"value": "hello"}, + variables={}, + ) + mocker.patch.object( + executor_service.registry_resolver, + "resolve_action", + new=mocker.AsyncMock( + return_value=ActionImplementation( + type="udf", action_name="core.transform.reshape" + ) + ), + ) + mocker.patch.object( + executor_service, "_mint_action_executor_token", return_value="step-token" + ) + backend = mocker.Mock() + backend.execute = mocker.AsyncMock( + return_value=ExecutorResultSuccess(result={"ok": True}) + ) + expansion_spy = mocker.patch.object( + executor_service, + "expand_template_source_references", + wraps=executor_service.expand_template_source_references, + ) + + await executor_service._execute_template_action( + backend=backend, + input=action_input, + ctx=executor_service.DispatchActionContext(role=role), + resolved_context=parent_resolved, + timeout=30, + source_args={"value": "hello"}, + ) + + expansion_spy.assert_not_called() + step_resolved = backend.execute.await_args.kwargs["resolved_context"] + assert step_resolved.evaluated_args == {"value": "hello"} diff --git a/tracecat/executor/expression_policy.py b/tracecat/executor/expression_policy.py new file mode 100644 index 000000000..a1b386852 --- /dev/null +++ b/tracecat/executor/expression_policy.py @@ -0,0 +1,215 @@ +from __future__ import annotations + +import re +from collections.abc import Mapping +from dataclasses import dataclass +from enum import StrEnum +from typing import Any, NamedTuple, cast + +from tracecat.exceptions import TracecatExpressionError +from tracecat.expressions import patterns +from tracecat.expressions.common import ExprContext +from tracecat.expressions.core import ( + Expression, + SecretPathExtractor, + TemplateExpression, +) +from tracecat.expressions.eval import eval_templated_object +from tracecat.expressions.parser.core import parser +from tracecat.secrets.constants import MASK_VALUE + + +class ExpressionPolicy(StrEnum): + """Controls expression evaluation for one top-level action parameter.""" + + RESOLVE = "resolve" + PRESERVE = "preserve" + REDACT_SECRETS = "redact_secrets" + + +class ActionParameter(NamedTuple): + """Identifies one parameter on a registry action.""" + + action: str + parameter: str + + +_PRESERVE = ExpressionPolicy.PRESERVE +_REDACT_SECRETS = ExpressionPolicy.REDACT_SECRETS + +# PRESERVE: workflow-authoring source executed later; evaluating it here is wrong. +# REDACT_SECRETS: free-form content persisted into Tracecat-owned durable state. +# Unlisted parameters default to RESOLVE. +POLICY_MAP: Mapping[ActionParameter, ExpressionPolicy] = { + ActionParameter("core.workflow.edit_workflow", "patch_ops"): _PRESERVE, + ActionParameter("core.workflow.create_workflow", "definition_yaml"): _PRESERVE, + ActionParameter("core.cases.create_case", "summary"): _REDACT_SECRETS, + ActionParameter("core.cases.create_case", "description"): _REDACT_SECRETS, + ActionParameter("core.cases.create_case", "fields"): _REDACT_SECRETS, + ActionParameter("core.cases.create_case", "payload"): _REDACT_SECRETS, + ActionParameter("core.cases.update_case", "summary"): _REDACT_SECRETS, + ActionParameter("core.cases.update_case", "description"): _REDACT_SECRETS, + ActionParameter("core.cases.update_case", "fields"): _REDACT_SECRETS, + ActionParameter("core.cases.update_case", "payload"): _REDACT_SECRETS, + ActionParameter("core.cases.create_comment", "content"): _REDACT_SECRETS, + ActionParameter("core.cases.reply_to_comment", "content"): _REDACT_SECRETS, + ActionParameter("core.cases.update_comment", "content"): _REDACT_SECRETS, + ActionParameter("core.table.insert_row", "row_data"): _REDACT_SECRETS, + ActionParameter("core.table.insert_rows", "rows_data"): _REDACT_SECRETS, + ActionParameter("core.table.update_row", "row_data"): _REDACT_SECRETS, + ActionParameter("core.cases.insert_row", "row"): _REDACT_SECRETS, + ActionParameter("ai.agent.create_preset", "instructions"): _REDACT_SECRETS, + ActionParameter("ai.agent.update_preset", "instructions"): _REDACT_SECRETS, +} + + +def _expression_references_secrets(expression: str) -> bool: + extractor = SecretPathExtractor() + results = Expression(expression, visitor=extractor).visit() + return bool(results.get(ExprContext.SECRETS)) + + +def _redact_secret_string(value: str) -> str: + def replace(match: re.Match[str]) -> str: + expression = match.group("expr") + if expression and _expression_references_secrets(expression): + return MASK_VALUE + return match.group("template") + + return patterns.TEMPLATE_STRING.sub(replace, value) + + +def redact_secret_expressions(value: Any) -> Any: + """Replace complete secret-dependent expression occurrences recursively.""" + match value: + case str(): + return _redact_secret_string(value) + case list(): + return [redact_secret_expressions(item) for item in value] + case dict(): + redacted: dict[Any, Any] = {} + for key, item in value.items(): + if isinstance(key, str): + redacted_key = _redact_secret_string(key) + if redacted_key != key: + raise TracecatExpressionError( + "Secret expressions are not allowed in dictionary keys", + detail={"code": "secret_expression_in_key"}, + ) + else: + redacted_key = key + redacted[redacted_key] = redact_secret_expressions(item) + return redacted + case _: + return value + + +def _is_direct_template_reference(template: str) -> bool: + """Return whether a template is exactly an inputs.* reference. + + Never expand steps.*: step results are materialized runtime data, and + splicing them in as evaluable source would let fetched content execute + expressions. + """ + match = patterns.TEMPLATE_STRING.fullmatch(template) + if match is None or not (expression := match.group("expr")): + return False + parse_tree = parser.parse(expression) + return parse_tree is not None and parse_tree.data == "template_action_inputs" + + +def expand_template_source_references(value: Any, context: Mapping[str, Any]) -> Any: + """Expand direct template input references while retaining nested source.""" + match value: + case str() if _is_direct_template_reference(value): + return TemplateExpression(value, operand=context).result() + case str(): + + def replace(match: re.Match[str]) -> str: + template = match.group("template") + if _is_direct_template_reference(template): + return str(TemplateExpression(template, operand=context).result()) + return template + + return patterns.TEMPLATE_STRING.sub(replace, value) + case list(): + return [expand_template_source_references(item, context) for item in value] + case dict(): + expanded: dict[Any, Any] = {} + for key, item in value.items(): + expanded_key = ( + expand_template_source_references(key, context) + if isinstance(key, str) + else key + ) + if expanded_key in expanded: + raise TracecatExpressionError( + "Template source expansion produced a duplicate dictionary key", + detail={"code": "template_source_key_collision"}, + ) + expanded[expanded_key] = expand_template_source_references( + item, context + ) + return expanded + case _: + return value + + +@dataclass(frozen=True, slots=True) +class PartitionedActionArgs: + """Action arguments split into resolvable and preserved parameters.""" + + action: str + original: Mapping[str, Any] + resolvable: Mapping[str, Any] + + def merge(self, evaluated: Mapping[str, Any]) -> dict[str, Any]: + """Restore preserved values without changing parameter order.""" + return { + parameter: ( + value + if expression_policy(self.action, parameter) + is ExpressionPolicy.PRESERVE + else evaluated[parameter] + ) + for parameter, value in self.original.items() + } + + +def expression_policy(action: str, parameter: str) -> ExpressionPolicy: + """Return the expression policy for an action parameter.""" + return POLICY_MAP.get(ActionParameter(action, parameter), ExpressionPolicy.RESOLVE) + + +def partition_action_args( + action: str, args: Mapping[str, Any] +) -> PartitionedActionArgs: + """Apply pre-evaluation policy and exclude preserved source subtrees.""" + resolvable: dict[str, Any] = {} + for parameter, value in args.items(): + match expression_policy(action, parameter): + case ExpressionPolicy.PRESERVE: + continue + case ExpressionPolicy.REDACT_SECRETS: + resolvable[parameter] = redact_secret_expressions(value) + case ExpressionPolicy.RESOLVE: + resolvable[parameter] = value + return PartitionedActionArgs( + action=action, + original=args, + resolvable=resolvable, + ) + + +def prepare_action_args( + action: str, + args: Mapping[str, Any], + context: Mapping[str, Any], +) -> dict[str, Any]: + """Apply field policy and evaluate one action's arguments.""" + partitioned = partition_action_args(action, args) + evaluated = cast( + Mapping[str, Any], + eval_templated_object(partitioned.resolvable, operand=context), + ) + return partitioned.merge(evaluated) diff --git a/tracecat/executor/service.py b/tracecat/executor/service.py index cb4055962..5bd075215 100644 --- a/tracecat/executor/service.py +++ b/tracecat/executor/service.py @@ -45,6 +45,13 @@ ) from tracecat.executor import registry_resolver from tracecat.executor.backends.base import ExecutorBackend +from tracecat.executor.expression_policy import ( + ExpressionPolicy, + expand_template_source_references, + expression_policy, + partition_action_args, + prepare_action_args, +) from tracecat.executor.schemas import ( ExecutorActionErrorInfo, ExecutorResultSuccess, @@ -446,6 +453,7 @@ async def _execute_template_action( ctx: DispatchActionContext, resolved_context: ResolvedContext, timeout: float, + source_args: Mapping[str, Any], ) -> Any: """Execute a template action by orchestrating its steps. @@ -461,6 +469,7 @@ async def _execute_template_action( ctx: Dispatch context containing the role resolved_context: Pre-resolved context with secrets and template definition timeout: Execution timeout + source_args: Unevaluated arguments supplied to this template invocation Returns: The evaluated returns expression result @@ -504,6 +513,15 @@ async def _execute_template_action( inputs=validated_input_args, steps={}, ) + source_input_args = dict(validated_input_args) + source_input_args.update(source_args) + source_context = TemplateExecutionContext( + SECRETS=secrets_context, + ENV=env_context, + VARS=vars_context, + inputs=source_input_args, + steps=template_context["steps"], + ) logger.info( "Executing template action via backend", @@ -519,10 +537,32 @@ async def _execute_template_action( step_action=step.action, ) - # Evaluate step args with template context - evaled_args = cast( - dict[str, Any], - eval_templated_object(step.args, operand=template_context), + # Expand direct input references to their source form only when a + # parameter's policy diverges from RESOLVE; the parse walk is wasted + # work otherwise. + source_step_args: dict[str, Any] | None = None + policy_args: Mapping[str, Any] = step.args + if any( + expression_policy(step.action, parameter) is not ExpressionPolicy.RESOLVE + for parameter in step.args + ): + source_step_args = cast( + dict[str, Any], + expand_template_source_references(step.args, source_context), + ) + policy_args = { + parameter: ( + source_step_args[parameter] + if expression_policy(step.action, parameter) + is not ExpressionPolicy.RESOLVE + else value + ) + for parameter, value in step.args.items() + } + evaled_args = prepare_action_args( + step.action, + policy_args, + template_context, ) # Prepare step context (reuses parent secrets, no re-fetch) @@ -534,6 +574,17 @@ async def _execute_template_action( role=role, ) + # Nested templates still route source to their own policy steps. + if source_step_args is None: + source_step_args = ( + cast( + dict[str, Any], + expand_template_source_references(step.args, source_context), + ) + if step_resolved.action_impl.type == "template" + else {} + ) + # Execute step via _invoke_step (handles nested templates) try: step_result = await _invoke_step( @@ -542,6 +593,7 @@ async def _execute_template_action( input=input, ctx=ctx, timeout=timeout, + source_args=source_step_args, ) except ExecutionError: # Re-raise with step context preserved @@ -574,6 +626,7 @@ async def _invoke_step( input: RunActionInput, ctx: DispatchActionContext, timeout: float, + source_args: Mapping[str, Any], ) -> Any: """Execute a template step. Skips masking (done at root level). @@ -587,6 +640,7 @@ async def _invoke_step( input: The original RunActionInput ctx: Dispatch context containing the role timeout: Execution timeout + source_args: Unevaluated arguments supplied to this action invocation Returns: The step execution result (unmasked) @@ -600,6 +654,7 @@ async def _invoke_step( ctx=ctx, resolved_context=resolved_context, timeout=timeout, + source_args=source_args, ) case "udf": # Leaf node - execute via backend @@ -679,8 +734,11 @@ async def prepare_resolved_context( action_name, input.registry_lock, role.organization_id ) - # Collect expressions to know what secrets/variables are needed - collected = collect_expressions(task.args) + # Apply field policy before expression collection: preserved source is + # excluded, while secret-dependent occurrences in durable content are + # replaced before any argument-driven secret lookup. + partitioned_args = partition_action_args(action_name, task.args) + collected = collect_expressions(partitioned_args.resolvable) # Fetch secrets and variables secrets = await secrets_manager.get_action_secrets( @@ -721,8 +779,7 @@ async def prepare_resolved_context( has_interaction=input.interaction_context is not None, ) - # Evaluate templated args (now with logical_time and interaction context set) - evaluated_args = evaluate_templated_args(task, context) + evaluated_args = prepare_action_args(action_name, task.args, context) finally: ctx_logical_time.reset(logical_time_token) ctx_interaction.reset(interaction_token) @@ -813,6 +870,7 @@ async def invoke_once( input=input, ctx=ctx, timeout=timeout, + source_args=input.task.args, ) except ExecutionError as e: From a3d4cf7d308e289dc7b4a6f56a58312dc1d2f341 Mon Sep 17 00:00:00 2001 From: Jordan Umusu Date: Mon, 3 Aug 2026 16:40:52 -0700 Subject: [PATCH 2/5] fix(executor): propagate secret dependencies through template inputs --- tests/unit/test_executor_expression_policy.py | 171 ++++++++++++++ tests/unit/test_executor_service.py | 116 ++++++++- tracecat/executor/expression_policy.py | 221 ++++++++++++++++-- tracecat/executor/service.py | 42 +++- 4 files changed, 518 insertions(+), 32 deletions(-) diff --git a/tests/unit/test_executor_expression_policy.py b/tests/unit/test_executor_expression_policy.py index b6da4893a..20fbeda48 100644 --- a/tests/unit/test_executor_expression_policy.py +++ b/tests/unit/test_executor_expression_policy.py @@ -5,8 +5,11 @@ POLICY_MAP, ActionParameter, ExpressionPolicy, + derive_secret_dependencies, + expand_template_source_references, expression_policy, partition_action_args, + prepare_action_args, redact_secret_expressions, ) from tracecat.secrets.constants import MASK_VALUE @@ -132,6 +135,174 @@ def test_redact_secret_expressions_recurses_through_values() -> None: } +@pytest.mark.parametrize( + "expression", + [ + '${{ inputs.content || "fallback" }}', + "${{ FN.to_base64(inputs.content) }}", + ], +) +def test_compound_template_input_dependency_is_redacted(expression: str) -> None: + dependencies = derive_secret_dependencies({"content": "${{ SECRETS.api.TOKEN }}"}) + assert isinstance(dependencies, dict) + + prepared = prepare_action_args( + "core.cases.create_comment", + {"content": expression}, + {"inputs": {"content": "runtime-secret"}}, + dependencies, + ) + + assert prepared == {"content": MASK_VALUE} + + +def test_template_input_dependencies_preserve_structured_paths() -> None: + dependencies = derive_secret_dependencies( + { + "context": { + "safe": "plain source", + "token": "${{ SECRETS.api.TOKEN }}", + } + } + ) + assert isinstance(dependencies, dict) + + prepared = prepare_action_args( + "core.cases.create_case", + { + "payload": { + "safe": "${{ inputs.context.safe }}", + "token": "${{ inputs.context.token }}", + } + }, + { + "inputs": { + "context": { + "safe": "plain value", + "token": "runtime-secret", + } + } + }, + dependencies, + ) + + assert prepared == { + "payload": { + "safe": "plain value", + "token": MASK_VALUE, + } + } + + +def test_compound_dependency_propagates_across_template_boundaries() -> None: + outer_dependencies = derive_secret_dependencies( + {"value": "${{ SECRETS.api.TOKEN }}"} + ) + assert isinstance(outer_dependencies, dict) + inner_dependencies = derive_secret_dependencies( + {"value": '${{ inputs.value || "fallback" }}'}, + outer_dependencies, + ) + + assert inner_dependencies == {"value": True} + + +@pytest.mark.parametrize( + "source", + [ + "${{ ACTIONS.fetch.result.body }}", + "${{ steps.prev.result }}", + "${{ TRIGGER.payload }}", + '${{ inputs.other || "x" }}', + "${{ var.item }}", + ], +) +def test_expand_keeps_reference_for_caller_scoped_source(source: str) -> None: + """Non-portable caller source must not be spliced into template scope.""" + expanded = expand_template_source_references( + { + "content": "${{ inputs.content }}", + "note": "Token: ${{ inputs.content }}", + }, + {"inputs": {"content": source}}, + ) + + assert expanded == { + "content": "${{ inputs.content }}", + "note": "Token: ${{ inputs.content }}", + } + + +def test_caller_scoped_source_resolves_to_runtime_value_at_sink() -> None: + """Regression: upstream action data through a sink must not become None.""" + source_context = {"inputs": {"content": "${{ ACTIONS.fetch.result.body }}"}} + step_args = {"case_id": "case-123", "content": "${{ inputs.content }}"} + + expanded = expand_template_source_references(step_args, source_context) + dependencies = derive_secret_dependencies(source_context["inputs"]) + assert isinstance(dependencies, dict) + + prepared = prepare_action_args( + "core.cases.create_comment", + expanded, + {"inputs": {"content": "evaluated-upstream-value"}}, + dependencies, + ) + + assert prepared == { + "case_id": "case-123", + "content": "evaluated-upstream-value", + } + + +def test_portable_secret_source_still_expands_for_granular_redaction() -> None: + source = "Host ${{ VARS.api.host }}, token ${{ SECRETS.api.TOKEN }}" + + expanded = expand_template_source_references( + {"content": "${{ inputs.content }}"}, + {"inputs": {"content": source}}, + ) + + assert expanded == {"content": source} + + +def test_non_portable_secret_source_masks_whole_expression() -> None: + """Secret mixed with caller-scoped context falls back to the dependency tree.""" + source_context = { + "inputs": { + "content": "${{ ACTIONS.fetch.result.body }} ${{ SECRETS.api.TOKEN }}" + } + } + step_args = {"content": "${{ inputs.content }}"} + + expanded = expand_template_source_references(step_args, source_context) + assert expanded == step_args + + dependencies = derive_secret_dependencies(source_context["inputs"]) + assert isinstance(dependencies, dict) + + prepared = prepare_action_args( + "core.cases.create_comment", + expanded, + {"inputs": {"content": "runtime-value"}}, + dependencies, + ) + + assert prepared == {"content": MASK_VALUE} + + +def test_adjacent_templates_do_not_crash_analysis() -> None: + """Regression: lazy fullmatch must not span "${{ a }} ${{ b }}".""" + source_context = {"inputs": {"a": "left", "b": "${{ SECRETS.api.TOKEN }}"}} + args = {"content": "${{ inputs.a }} ${{ inputs.b }}"} + + dependencies = derive_secret_dependencies(source_context["inputs"]) + assert dependencies == {"a": False, "b": True} + + expanded = expand_template_source_references(args, source_context) + assert expanded == {"content": "left ${{ SECRETS.api.TOKEN }}"} + + def test_partition_redacts_before_evaluation_and_restores_order() -> None: args = { "case_id": "${{ VARS.case.id }}", diff --git a/tests/unit/test_executor_service.py b/tests/unit/test_executor_service.py index 36c5024de..eb7a85227 100644 --- a/tests/unit/test_executor_service.py +++ b/tests/unit/test_executor_service.py @@ -665,6 +665,32 @@ async def test_prepare_resolved_context_resolves_unmapped_parameters( "content": f"Host api.example.com, token {MASK_VALUE}", }, ), + ( + "core.cases.create_comment", + { + "case_id": "case-123", + "content": '${{ inputs.value || "fallback" }}', + }, + "${{ SECRETS.runtime.TOKEN }}", + "runtime-secret", + { + "case_id": "case-123", + "content": MASK_VALUE, + }, + ), + ( + "core.cases.create_comment", + { + "case_id": "case-123", + "content": "${{ inputs.value }}", + }, + "${{ ACTIONS.fetch.result.body }}", + "upstream-value", + { + "case_id": "case-123", + "content": "upstream-value", + }, + ), ( "core.workflow.edit_workflow", { @@ -774,6 +800,93 @@ async def test_template_step_applies_target_action_expression_policy( assert step_resolved.evaluated_args == expected_args +@pytest.mark.anyio +async def test_compound_secret_dependency_reaches_nested_template_sink(mocker): + """Secret dependency survives composition and a nested template boundary.""" + source_value = "${{ SECRETS.runtime.TOKEN }}" + action_input = _expression_policy_input( + "testing.policy_wrapper", + {"value": source_value}, + ) + role = _expression_policy_role("tracecat-executor") + inner_template = ActionImplementation( + type="template", + action_name="testing.inner_wrapper", + template_definition={ + "name": "inner_wrapper", + "namespace": "testing", + "title": "Inner wrapper", + "description": "Calls a protected sink", + "display_group": "Testing", + "expects": {}, + "steps": [ + { + "ref": "persist", + "action": "core.cases.create_comment", + "args": { + "case_id": "case-123", + "content": "${{ inputs.value }}", + }, + } + ], + "returns": "done", + }, + ) + parent_resolved = _policy_wrapper_resolved( + action_input, + role, + steps=[ + { + "ref": "inner", + "action": "testing.inner_wrapper", + "args": { + "value": '${{ inputs.value || "fallback" }}', + }, + } + ], + evaluated_args={"value": "runtime-secret"}, + variables={}, + secrets={"runtime": {"TOKEN": "runtime-secret"}}, + ) + mocker.patch.object( + executor_service.registry_resolver, + "resolve_action", + new=mocker.AsyncMock( + side_effect=[ + inner_template, + ActionImplementation( + type="udf", + action_name="core.cases.create_comment", + ), + ] + ), + ) + mocker.patch.object( + executor_service, + "_mint_action_executor_token", + return_value="step-token", + ) + backend = mocker.Mock() + backend.execute = mocker.AsyncMock( + return_value=ExecutorResultSuccess(result={"persisted": True}) + ) + + await executor_service._execute_template_action( + backend=backend, + input=action_input, + ctx=executor_service.DispatchActionContext(role=role), + resolved_context=parent_resolved, + timeout=30, + source_args={"value": source_value}, + ) + + sink_resolved = backend.execute.await_args.kwargs["resolved_context"] + assert sink_resolved.evaluated_args == { + "case_id": "case-123", + "content": MASK_VALUE, + } + + def _policy_wrapper_resolved( action_input: RunActionInput, role: Role, @@ -781,9 +894,10 @@ def _policy_wrapper_resolved( steps: list[dict[str, object]], evaluated_args: dict[str, object], variables: dict[str, dict[str, str]], + secrets: dict[str, dict[str, str]] | None = None, ) -> ResolvedContext: return ResolvedContext( - secrets={}, + secrets=secrets or {}, variables=variables, action_impl=ActionImplementation( type="template", diff --git a/tracecat/executor/expression_policy.py b/tracecat/executor/expression_policy.py index a1b386852..f909b7412 100644 --- a/tracecat/executor/expression_policy.py +++ b/tracecat/executor/expression_policy.py @@ -6,18 +6,21 @@ from enum import StrEnum from typing import Any, NamedTuple, cast +from lark import Token + from tracecat.exceptions import TracecatExpressionError from tracecat.expressions import patterns -from tracecat.expressions.common import ExprContext -from tracecat.expressions.core import ( - Expression, - SecretPathExtractor, - TemplateExpression, -) +from tracecat.expressions.common import ExprContext, eval_jsonpath +from tracecat.expressions.core import TemplateExpression from tracecat.expressions.eval import eval_templated_object from tracecat.expressions.parser.core import parser from tracecat.secrets.constants import MASK_VALUE +type SecretDependency = bool | list[SecretDependency] | dict[Any, SecretDependency] +type SecretDependencies = Mapping[str, SecretDependency] + +_SECRET_KEY_DEPENDENCY = object() + class ExpressionPolicy(StrEnum): """Controls expression evaluation for one top-level action parameter.""" @@ -63,34 +66,94 @@ class ActionParameter(NamedTuple): } -def _expression_references_secrets(expression: str) -> bool: - extractor = SecretPathExtractor() - results = Expression(expression, visitor=extractor).visit() - return bool(results.get(ExprContext.SECRETS)) +def _has_secret_dependency(dependency: SecretDependency) -> bool: + match dependency: + case bool(): + return dependency + case list(): + return any(_has_secret_dependency(item) for item in dependency) + case dict(): + return any(_has_secret_dependency(item) for item in dependency.values()) + + +def _input_dependency( + path: str, + input_dependencies: SecretDependencies, +) -> SecretDependency | None: + dependency = eval_jsonpath( + f"{ExprContext.TEMPLATE_ACTION_INPUTS}{path}", + {ExprContext.TEMPLATE_ACTION_INPUTS: input_dependencies}, + ) + if dependency is not None: + return cast(SecretDependency, dependency) + + # A compound expression can collapse a structured input's dependency to + # True. Any later access beneath that input must remain conservatively + # secret-dependent even though the boolean tree has no child to traverse. + match = re.match(r"^\.([A-Za-z_][A-Za-z0-9_]*)", path) + if match is None: + return True if _has_secret_dependency(dict(input_dependencies)) else None + return input_dependencies.get(match.group(1)) + + +def _expression_depends_on_secrets( + expression: str, + input_dependencies: SecretDependencies | None = None, +) -> bool: + parse_tree = parser.parse(expression) + if parse_tree is None: + raise TracecatExpressionError( + f"Parser returned None for expression {expression!r}" + ) + if next(parse_tree.find_data("secrets"), None) is not None: + return True + if not input_dependencies: + return False + + for node in parse_tree.find_data("template_action_inputs"): + token = node.children[0] + if not isinstance(token, Token): + raise TracecatExpressionError( + f"Expected template input path token, got {type(token).__name__}" + ) + dependency = _input_dependency(str(token), input_dependencies) + if dependency is not None and _has_secret_dependency(dependency): + return True + return False -def _redact_secret_string(value: str) -> str: +def _redact_secret_string( + value: str, + input_dependencies: SecretDependencies | None = None, +) -> str: def replace(match: re.Match[str]) -> str: expression = match.group("expr") - if expression and _expression_references_secrets(expression): + if expression and _expression_depends_on_secrets( + expression, input_dependencies + ): return MASK_VALUE return match.group("template") return patterns.TEMPLATE_STRING.sub(replace, value) -def redact_secret_expressions(value: Any) -> Any: +def redact_secret_expressions( + value: Any, + input_dependencies: SecretDependencies | None = None, +) -> Any: """Replace complete secret-dependent expression occurrences recursively.""" match value: case str(): - return _redact_secret_string(value) + return _redact_secret_string(value, input_dependencies) case list(): - return [redact_secret_expressions(item) for item in value] + return [ + redact_secret_expressions(item, input_dependencies) for item in value + ] case dict(): redacted: dict[Any, Any] = {} for key, item in value.items(): if isinstance(key, str): - redacted_key = _redact_secret_string(key) + redacted_key = _redact_secret_string(key, input_dependencies) if redacted_key != key: raise TracecatExpressionError( "Secret expressions are not allowed in dictionary keys", @@ -98,12 +161,116 @@ def redact_secret_expressions(value: Any) -> Any: ) else: redacted_key = key - redacted[redacted_key] = redact_secret_expressions(item) + redacted[redacted_key] = redact_secret_expressions( + item, input_dependencies + ) return redacted case _: return value +def derive_secret_dependencies( + value: Any, + input_dependencies: SecretDependencies | None = None, +) -> SecretDependency: + """Derive expression-level secret dependencies without evaluating values.""" + match value: + case str(): + # Lazy fullmatch spans adjacent templates ("${{ a }} ${{ b }}"), + # so gate on the standalone pattern before extracting. + direct_match = ( + patterns.TEMPLATE_STRING.fullmatch(value) + if patterns.STANDALONE_TEMPLATE.match(value) + else None + ) + if direct_match is not None and (expression := direct_match.group("expr")): + parse_tree = parser.parse(expression) + if ( + parse_tree is not None + and parse_tree.data == "template_action_inputs" + and input_dependencies + ): + token = parse_tree.children[0] + if not isinstance(token, Token): + raise TracecatExpressionError( + "Expected template input path token" + ) + dependency = _input_dependency(str(token), input_dependencies) + return dependency if dependency is not None else False + + return any( + _expression_depends_on_secrets(expression, input_dependencies) + for match in patterns.TEMPLATE_STRING.finditer(value) + if (expression := match.group("expr")) is not None + ) + case list(): + return [ + derive_secret_dependencies(item, input_dependencies) for item in value + ] + case dict(): + dependencies = { + key: derive_secret_dependencies(item, input_dependencies) + for key, item in value.items() + } + if any( + isinstance(key, str) + and _has_secret_dependency( + derive_secret_dependencies(key, input_dependencies) + ) + for key in value + ): + dependencies[_SECRET_KEY_DEPENDENCY] = True + return dependencies + case _: + return False + + +# Context references a template's evaluation scope cannot resolve. Caller +# source containing them must stay behind its inputs.* reference; the +# dependency tree still governs redaction for that reference. +_NON_PORTABLE_NODES = frozenset( + { + "actions", + "trigger", + "local_vars", + "local_vars_assignment", + "template_action_inputs", + "template_action_steps", + } +) + + +def _is_portable_source(value: Any) -> bool: + """Whether expanded caller source can evaluate inside a template scope.""" + match value: + case str(): + for match_ in patterns.TEMPLATE_STRING.finditer(value): + expression = match_.group("expr") + if not expression: + continue + try: + parse_tree = parser.parse(expression) + except TracecatExpressionError: + return False + if parse_tree is None: + return False + if any( + subtree.data in _NON_PORTABLE_NODES + for subtree in parse_tree.iter_subtrees() + ): + return False + return True + case list(): + return all(_is_portable_source(item) for item in value) + case dict(): + return all( + _is_portable_source(key) and _is_portable_source(item) + for key, item in value.items() + ) + case _: + return True + + def _is_direct_template_reference(template: str) -> bool: """Return whether a template is exactly an inputs.* reference. @@ -111,6 +278,8 @@ def _is_direct_template_reference(template: str) -> bool: splicing them in as evaluable source would let fetched content execute expressions. """ + if patterns.STANDALONE_TEMPLATE.match(template) is None: + return False match = patterns.TEMPLATE_STRING.fullmatch(template) if match is None or not (expression := match.group("expr")): return False @@ -122,13 +291,16 @@ def expand_template_source_references(value: Any, context: Mapping[str, Any]) -> """Expand direct template input references while retaining nested source.""" match value: case str() if _is_direct_template_reference(value): - return TemplateExpression(value, operand=context).result() + expanded = TemplateExpression(value, operand=context).result() + return expanded if _is_portable_source(expanded) else value case str(): def replace(match: re.Match[str]) -> str: template = match.group("template") if _is_direct_template_reference(template): - return str(TemplateExpression(template, operand=context).result()) + expanded = TemplateExpression(template, operand=context).result() + if _is_portable_source(expanded): + return str(expanded) return template return patterns.TEMPLATE_STRING.sub(replace, value) @@ -182,7 +354,9 @@ def expression_policy(action: str, parameter: str) -> ExpressionPolicy: def partition_action_args( - action: str, args: Mapping[str, Any] + action: str, + args: Mapping[str, Any], + input_dependencies: SecretDependencies | None = None, ) -> PartitionedActionArgs: """Apply pre-evaluation policy and exclude preserved source subtrees.""" resolvable: dict[str, Any] = {} @@ -191,7 +365,9 @@ def partition_action_args( case ExpressionPolicy.PRESERVE: continue case ExpressionPolicy.REDACT_SECRETS: - resolvable[parameter] = redact_secret_expressions(value) + resolvable[parameter] = redact_secret_expressions( + value, input_dependencies + ) case ExpressionPolicy.RESOLVE: resolvable[parameter] = value return PartitionedActionArgs( @@ -205,9 +381,10 @@ def prepare_action_args( action: str, args: Mapping[str, Any], context: Mapping[str, Any], + input_dependencies: SecretDependencies | None = None, ) -> dict[str, Any]: """Apply field policy and evaluate one action's arguments.""" - partitioned = partition_action_args(action, args) + partitioned = partition_action_args(action, args, input_dependencies) evaluated = cast( Mapping[str, Any], eval_templated_object(partitioned.resolvable, operand=context), diff --git a/tracecat/executor/service.py b/tracecat/executor/service.py index 5bd075215..ab7fdf740 100644 --- a/tracecat/executor/service.py +++ b/tracecat/executor/service.py @@ -47,6 +47,9 @@ from tracecat.executor.backends.base import ExecutorBackend from tracecat.executor.expression_policy import ( ExpressionPolicy, + SecretDependencies, + SecretDependency, + derive_secret_dependencies, expand_template_source_references, expression_policy, partition_action_args, @@ -454,6 +457,7 @@ async def _execute_template_action( resolved_context: ResolvedContext, timeout: float, source_args: Mapping[str, Any], + source_dependencies: SecretDependencies | None = None, ) -> Any: """Execute a template action by orchestrating its steps. @@ -470,6 +474,7 @@ async def _execute_template_action( resolved_context: Pre-resolved context with secrets and template definition timeout: Execution timeout source_args: Unevaluated arguments supplied to this template invocation + source_dependencies: Secret dependencies derived by the caller Returns: The evaluated returns expression result @@ -515,13 +520,15 @@ async def _execute_template_action( ) source_input_args = dict(validated_input_args) source_input_args.update(source_args) - source_context = TemplateExecutionContext( - SECRETS=secrets_context, - ENV=env_context, - VARS=vars_context, - inputs=source_input_args, - steps=template_context["steps"], + input_dependencies = cast( + dict[str, SecretDependency], + derive_secret_dependencies(source_input_args), ) + if source_dependencies: + input_dependencies.update(source_dependencies) + # Source expansion is a raw inputs.* lookup. Do not expose runtime secrets, + # variables, environment, or step results to this analysis context. + source_context: Mapping[str, Any] = {"inputs": source_input_args} logger.info( "Executing template action via backend", @@ -537,9 +544,9 @@ async def _execute_template_action( step_action=step.action, ) - # Expand direct input references to their source form only when a - # parameter's policy diverges from RESOLVE; the parse walk is wasted - # work otherwise. + # Non-resolving policies need the caller's source form for direct input + # references. The parallel dependency tree covers compound expressions + # that cannot be source-expanded without rewriting their AST. source_step_args: dict[str, Any] | None = None policy_args: Mapping[str, Any] = step.args if any( @@ -563,6 +570,7 @@ async def _execute_template_action( step.action, policy_args, template_context, + input_dependencies, ) # Prepare step context (reuses parent secrets, no re-fetch) @@ -584,6 +592,14 @@ async def _execute_template_action( if step_resolved.action_impl.type == "template" else {} ) + step_dependencies = ( + cast( + dict[str, SecretDependency], + derive_secret_dependencies(step.args, input_dependencies), + ) + if step_resolved.action_impl.type == "template" + else None + ) # Execute step via _invoke_step (handles nested templates) try: @@ -594,6 +610,7 @@ async def _execute_template_action( ctx=ctx, timeout=timeout, source_args=source_step_args, + source_dependencies=step_dependencies, ) except ExecutionError: # Re-raise with step context preserved @@ -627,6 +644,7 @@ async def _invoke_step( ctx: DispatchActionContext, timeout: float, source_args: Mapping[str, Any], + source_dependencies: SecretDependencies | None = None, ) -> Any: """Execute a template step. Skips masking (done at root level). @@ -641,6 +659,7 @@ async def _invoke_step( ctx: Dispatch context containing the role timeout: Execution timeout source_args: Unevaluated arguments supplied to this action invocation + source_dependencies: Secret dependencies derived by the caller Returns: The step execution result (unmasked) @@ -655,6 +674,7 @@ async def _invoke_step( resolved_context=resolved_context, timeout=timeout, source_args=source_args, + source_dependencies=source_dependencies, ) case "udf": # Leaf node - execute via backend @@ -871,6 +891,10 @@ async def invoke_once( ctx=ctx, timeout=timeout, source_args=input.task.args, + source_dependencies=cast( + dict[str, SecretDependency], + derive_secret_dependencies(input.task.args), + ), ) except ExecutionError as e: From d85c596071197aeb407abc178f8406afb42e0185 Mon Sep 17 00:00:00 2001 From: Jordan Umusu Date: Thu, 6 Aug 2026 17:12:40 -0400 Subject: [PATCH 3/5] fix(executor): mask secrets at sinks instead of tainting template state --- tests/unit/test_executor_expression_policy.py | 438 +++++++++++++---- tests/unit/test_executor_service.py | 105 +++- tracecat/executor/expression_policy.py | 450 +++++++++++++----- tracecat/executor/service.py | 114 ++--- 4 files changed, 811 insertions(+), 296 deletions(-) diff --git a/tests/unit/test_executor_expression_policy.py b/tests/unit/test_executor_expression_policy.py index 20fbeda48..62c800d1e 100644 --- a/tests/unit/test_executor_expression_policy.py +++ b/tests/unit/test_executor_expression_policy.py @@ -1,20 +1,40 @@ +from typing import Any + import pytest +from tracecat.dsl.schemas import TemplateExecutionContext from tracecat.exceptions import TracecatExpressionError from tracecat.executor.expression_policy import ( POLICY_MAP, ActionParameter, ExpressionPolicy, + TemplateExecutionState, derive_secret_dependencies, - expand_template_source_references, + derive_source_provenance, expression_policy, partition_action_args, prepare_action_args, redact_secret_expressions, + substitute_source_references, ) from tracecat.secrets.constants import MASK_VALUE +def _state( + source_args: dict[str, Any], + runtime_inputs: dict[str, Any], + variables: dict[str, Any] | None = None, +) -> TemplateExecutionState: + provenance = derive_source_provenance(source_args) + context = TemplateExecutionContext( + SECRETS={}, + VARS=variables or {}, + inputs=runtime_inputs, + steps={}, + ) + return TemplateExecutionState(context, provenance) + + def test_action_parameter_policy_scope_is_explicit() -> None: preserved = { parameter @@ -38,6 +58,10 @@ def test_action_parameter_policy_scope_is_explicit() -> None: ), } assert redacted == { + ActionParameter(action="core.workflow.create_workflow", parameter="title"), + ActionParameter( + action="core.workflow.create_workflow", parameter="description" + ), ActionParameter(action="core.cases.create_case", parameter="summary"), ActionParameter(action="core.cases.create_case", parameter="description"), ActionParameter(action="core.cases.create_case", parameter="fields"), @@ -143,31 +167,35 @@ def test_redact_secret_expressions_recurses_through_values() -> None: ], ) def test_compound_template_input_dependency_is_redacted(expression: str) -> None: - dependencies = derive_secret_dependencies({"content": "${{ SECRETS.api.TOKEN }}"}) - assert isinstance(dependencies, dict) + state = _state( + source_args={"content": "${{ SECRETS.api.TOKEN }}"}, + runtime_inputs={"content": "runtime-secret"}, + ) - prepared = prepare_action_args( - "core.cases.create_comment", - {"content": expression}, - {"inputs": {"content": "runtime-secret"}}, - dependencies, + prepared = state.prepare_step_args( + "core.cases.create_comment", {"content": expression} ) assert prepared == {"content": MASK_VALUE} def test_template_input_dependencies_preserve_structured_paths() -> None: - dependencies = derive_secret_dependencies( - { + state = _state( + source_args={ "context": { - "safe": "plain source", + "safe": "plain value", "token": "${{ SECRETS.api.TOKEN }}", } - } + }, + runtime_inputs={ + "context": { + "safe": "plain value", + "token": "runtime-secret", + } + }, ) - assert isinstance(dependencies, dict) - prepared = prepare_action_args( + prepared = state.prepare_step_args( "core.cases.create_case", { "payload": { @@ -175,15 +203,6 @@ def test_template_input_dependencies_preserve_structured_paths() -> None: "token": "${{ inputs.context.token }}", } }, - { - "inputs": { - "context": { - "safe": "plain value", - "token": "runtime-secret", - } - } - }, - dependencies, ) assert prepared == { @@ -194,59 +213,111 @@ def test_template_input_dependencies_preserve_structured_paths() -> None: } -def test_compound_dependency_propagates_across_template_boundaries() -> None: - outer_dependencies = derive_secret_dependencies( - {"value": "${{ SECRETS.api.TOKEN }}"} +def test_whole_structured_input_is_tree_masked_without_re_evaluation() -> None: + state = _state( + source_args={ + "context": { + "safe": "${{ ACTIONS.fetch.result.note }}", + "token": "${{ SECRETS.api.TOKEN }}", + } + }, + runtime_inputs={ + "context": { + "safe": "${{ SECRETS.injected.VALUE }}", + "token": "runtime-secret", + } + }, ) - assert isinstance(outer_dependencies, dict) - inner_dependencies = derive_secret_dependencies( - {"value": '${{ inputs.value || "fallback" }}'}, - outer_dependencies, + + prepared = state.prepare_step_args( + "core.table.insert_row", + {"table": "events", "row_data": "${{ inputs.context }}"}, ) - assert inner_dependencies == {"value": True} + assert prepared == { + "table": "events", + "row_data": { + "safe": "${{ SECRETS.injected.VALUE }}", + "token": MASK_VALUE, + }, + } + +def test_secret_dependent_input_key_is_rejected_only_at_sink() -> None: + state = _state( + source_args={ + "content": "plain", + "context": {"${{ SECRETS.api.KEY }}": "value"}, + }, + runtime_inputs={ + "content": "plain", + "context": {"runtime-key": "value"}, + }, + ) -@pytest.mark.parametrize( - "source", - [ - "${{ ACTIONS.fetch.result.body }}", - "${{ steps.prev.result }}", - "${{ TRIGGER.payload }}", - '${{ inputs.other || "x" }}', - "${{ var.item }}", - ], -) -def test_expand_keeps_reference_for_caller_scoped_source(source: str) -> None: - """Non-portable caller source must not be spliced into template scope.""" - expanded = expand_template_source_references( + assert state.prepare_step_args( + "core.cases.create_comment", + {"content": "${{ inputs.content }}"}, + ) == {"content": "plain"} + + with pytest.raises(TracecatExpressionError) as exc_info: + state.prepare_step_args( + "core.cases.create_case", + {"payload": "${{ inputs.context }}"}, + ) + + assert exc_info.value.detail == {"code": "secret_expression_in_key"} + + +def test_compound_dependency_propagates_across_template_boundaries() -> None: + outer = _state( + source_args={"value": "${{ SECRETS.api.TOKEN }}"}, + runtime_inputs={"value": "runtime-secret"}, + ) + + child = outer.child_provenance({"value": '${{ inputs.value || "fallback" }}'}) + + assert child["value"].dependency is True + assert child["value"].source == '${{ inputs.value || "fallback" }}' + + +def test_substitution_splices_caller_source_without_evaluating() -> None: + """Direct input references become the caller's raw source, verbatim.""" + substituted = substitute_source_references( { "content": "${{ inputs.content }}", "note": "Token: ${{ inputs.content }}", + "step": "${{ steps.prev.result }}", }, - {"inputs": {"content": source}}, + {"inputs": {"content": "${{ ACTIONS.fetch.result.body }}"}}, ) - assert expanded == { - "content": "${{ inputs.content }}", - "note": "Token: ${{ inputs.content }}", + assert substituted == { + "content": "${{ ACTIONS.fetch.result.body }}", + "note": "Token: ${{ ACTIONS.fetch.result.body }}", + "step": "${{ steps.prev.result }}", } +def test_substitution_keeps_unresolvable_references() -> None: + substituted = substitute_source_references( + {"content": "${{ inputs.missing }}"}, + {"inputs": {"other": "value"}}, + ) + + assert substituted == {"content": "${{ inputs.missing }}"} + + def test_caller_scoped_source_resolves_to_runtime_value_at_sink() -> None: """Regression: upstream action data through a sink must not become None.""" - source_context = {"inputs": {"content": "${{ ACTIONS.fetch.result.body }}"}} - step_args = {"case_id": "case-123", "content": "${{ inputs.content }}"} - - expanded = expand_template_source_references(step_args, source_context) - dependencies = derive_secret_dependencies(source_context["inputs"]) - assert isinstance(dependencies, dict) + state = _state( + source_args={"content": "${{ ACTIONS.fetch.result.body }}"}, + runtime_inputs={"content": "evaluated-upstream-value"}, + ) - prepared = prepare_action_args( + prepared = state.prepare_step_args( "core.cases.create_comment", - expanded, - {"inputs": {"content": "evaluated-upstream-value"}}, - dependencies, + {"case_id": "case-123", "content": "${{ inputs.content }}"}, ) assert prepared == { @@ -255,52 +326,237 @@ def test_caller_scoped_source_resolves_to_runtime_value_at_sink() -> None: } -def test_portable_secret_source_still_expands_for_granular_redaction() -> None: - source = "Host ${{ VARS.api.host }}, token ${{ SECRETS.api.TOKEN }}" +def test_whole_mixed_string_input_is_conservatively_masked() -> None: + state = _state( + source_args={ + "content": "${{ ACTIONS.fetch.result.body }} ${{ SECRETS.api.TOKEN }}" + }, + runtime_inputs={"content": "upstream s3cret"}, + ) - expanded = expand_template_source_references( - {"content": "${{ inputs.content }}"}, - {"inputs": {"content": source}}, + prepared = state.prepare_step_args( + "core.cases.create_comment", {"content": "${{ inputs.content }}"} ) - assert expanded == {"content": source} + assert prepared == {"content": MASK_VALUE} + +def test_adjacent_templates_mask_independently() -> None: + state = _state( + source_args={"a": "left", "b": "${{ SECRETS.api.TOKEN }}"}, + runtime_inputs={"a": "left", "b": "runtime-secret"}, + ) + assert ( + derive_source_provenance({"a": "left", "b": "${{ SECRETS.api.TOKEN }}"})[ + "b" + ].dependency + is True + ) -def test_non_portable_secret_source_masks_whole_expression() -> None: - """Secret mixed with caller-scoped context falls back to the dependency tree.""" - source_context = { - "inputs": { - "content": "${{ ACTIONS.fetch.result.body }} ${{ SECRETS.api.TOKEN }}" + prepared = state.prepare_step_args( + "core.cases.create_comment", + {"content": "${{ inputs.a }} ${{ inputs.b }}"}, + ) + + assert prepared == {"content": f"left {MASK_VALUE}"} + + +def test_step_results_are_runtime_data_without_taint() -> None: + state = _state( + source_args={"token": "${{ SECRETS.api.TOKEN }}"}, + runtime_inputs={"token": "runtime-secret"}, + ) + state.record_step("normalize", {"result": "runtime-secret"}) + state.record_step("plain", {"result": "ok"}) + + prepared = state.prepare_step_args( + "core.cases.create_comment", + {"content": "${{ steps.normalize.result }} / ${{ steps.plain.result }}"}, + ) + + assert prepared == {"content": "runtime-secret / ok"} + + +def test_step_results_do_not_taint_child_provenance() -> None: + state = _state( + source_args={"token": "${{ SECRETS.api.TOKEN }}"}, + runtime_inputs={"token": "runtime-secret"}, + ) + state.record_step("normalize", {"result": "runtime-secret"}) + + child = state.child_provenance({"value": "${{ steps.normalize.result }}"}) + + assert child["value"].dependency is False + + +def test_workflow_metadata_redacts_secret_expressions() -> None: + prepared = prepare_action_args( + "core.workflow.create_workflow", + { + "title": "Sync ${{ SECRETS.api.KEY }}", + "description": "Uses ${{ VARS.api.host }}", + }, + {"VARS": {"api": {"host": "api.example.com"}}}, + ) + + assert prepared == { + "title": f"Sync {MASK_VALUE}", + "description": "Uses api.example.com", + } + + +def test_preserve_substitutes_caller_source_for_direct_reference() -> None: + ops = [ + { + "op": "add", + "path": "/definition/actions/-", + "value": {"token": "${{ SECRETS.source.TOKEN }}"}, } + ] + state = _state( + source_args={"ops": ops}, + runtime_inputs={"ops": [{"op": "add", "value": {"token": "runtime-secret"}}]}, + ) + + prepared = state.prepare_step_args( + "core.workflow.edit_workflow", + {"workflow_id": "wf-123", "patch_ops": "${{ inputs.ops }}"}, + ) + + assert prepared == {"workflow_id": "wf-123", "patch_ops": ops} + + +def test_preserve_carrier_source_materializes_runtime_value() -> None: + """A caller source that is itself a carrier resolves to the runtime value.""" + runtime_ops = [{"op": "add", "path": "/x", "value": 1}] + state = _state( + source_args={"ops": "${{ ACTIONS.builder.result }}"}, + runtime_inputs={"ops": runtime_ops}, + ) + + prepared = state.prepare_step_args( + "core.workflow.edit_workflow", + {"workflow_id": "wf-123", "patch_ops": "${{ inputs.ops }}"}, + ) + + assert prepared == {"workflow_id": "wf-123", "patch_ops": runtime_ops} + + +def test_secret_dependent_carrier_stays_preserved_in_template() -> None: + state = _state( + source_args={"ops": "${{ SECRETS.api.OPS }}"}, + runtime_inputs={"ops": "runtime-secret"}, + ) + + prepared = state.prepare_step_args( + "core.workflow.edit_workflow", + {"workflow_id": "wf-123", "patch_ops": "${{ inputs.ops }}"}, + ) + + assert prepared == { + "workflow_id": "wf-123", + "patch_ops": "${{ SECRETS.api.OPS }}", } - step_args = {"content": "${{ inputs.content }}"} - expanded = expand_template_source_references(step_args, source_context) - assert expanded == step_args - dependencies = derive_secret_dependencies(source_context["inputs"]) - assert isinstance(dependencies, dict) +def test_preserve_untainted_step_carrier_materializes() -> None: + runtime_ops = [{"op": "add", "path": "/x", "value": 1}] + state = _state( + source_args={}, + runtime_inputs={}, + ) + state.record_step("build", {"result": runtime_ops}) + + prepared = state.prepare_step_args( + "core.workflow.edit_workflow", + {"workflow_id": "wf-123", "patch_ops": "${{ steps.build.result }}"}, + ) + + assert prepared == {"workflow_id": "wf-123", "patch_ops": runtime_ops} + +def test_preserve_step_carrier_materializes_without_step_taint() -> None: + state = _state( + source_args={"token": "${{ SECRETS.api.TOKEN }}"}, + runtime_inputs={"token": "runtime-secret"}, + ) + state.record_step("build", {"result": "runtime-secret"}) + + prepared = state.prepare_step_args( + "core.workflow.edit_workflow", + {"workflow_id": "wf-123", "patch_ops": "${{ steps.build.result }}"}, + ) + + assert prepared == { + "workflow_id": "wf-123", + "patch_ops": "runtime-secret", + } + + +def test_preserve_carrier_expression_materializes_at_root() -> None: + """A bare expression can never be valid preserved source; evaluate it.""" + ops = [{"op": "add", "path": "/definition/actions/-", "value": "${{ VARS.x }}"}] prepared = prepare_action_args( - "core.cases.create_comment", - expanded, - {"inputs": {"content": "runtime-value"}}, - dependencies, + "core.workflow.edit_workflow", + { + "workflow_id": "wf-123", + "patch_ops": "${{ ACTIONS.builder.result }}", + "validate_only": False, + }, + {"ACTIONS": {"builder": {"result": ops}}}, ) - assert prepared == {"content": MASK_VALUE} + assert prepared == { + "workflow_id": "wf-123", + "patch_ops": ops, + "validate_only": False, + } + + +def test_preserve_carrier_expression_materializes_definition_yaml() -> None: + yaml_source = "definition:\n title: Generated\n" + prepared = prepare_action_args( + "core.workflow.create_workflow", + {"definition_yaml": "${{ ACTIONS.gen.result }}"}, + {"ACTIONS": {"gen": {"result": yaml_source}}}, + ) + assert prepared == {"definition_yaml": yaml_source} -def test_adjacent_templates_do_not_crash_analysis() -> None: - """Regression: lazy fullmatch must not span "${{ a }} ${{ b }}".""" - source_context = {"inputs": {"a": "left", "b": "${{ SECRETS.api.TOKEN }}"}} - args = {"content": "${{ inputs.a }} ${{ inputs.b }}"} - dependencies = derive_secret_dependencies(source_context["inputs"]) - assert dependencies == {"a": False, "b": True} +def test_secret_dependent_carrier_stays_preserved_at_root() -> None: + prepared = prepare_action_args( + "core.workflow.edit_workflow", + {"workflow_id": "wf-123", "patch_ops": "${{ SECRETS.api.OPS }}"}, + {"SECRETS": {"api": {"OPS": "runtime-secret"}}}, + ) + + assert prepared == { + "workflow_id": "wf-123", + "patch_ops": "${{ SECRETS.api.OPS }}", + } + + +def test_preserve_literal_source_is_untouched_by_carrier_carveout() -> None: + source = {"op": "add", "path": "/x", "value": "${{ ACTIONS.a.result }}"} + prepared = prepare_action_args( + "core.workflow.edit_workflow", + {"workflow_id": "wf-123", "patch_ops": [source]}, + {}, + ) - expanded = expand_template_source_references(args, source_context) - assert expanded == {"content": "left ${{ SECRETS.api.TOKEN }}"} + assert prepared == {"workflow_id": "wf-123", "patch_ops": [source]} + + +def test_source_provenance_never_evaluates_authored_source() -> None: + provenance = derive_source_provenance( + {"content": "${{ FN.uuid4() }} ${{ SECRETS.api.TOKEN }}"}, + ) + + assert provenance["content"].dependency is True + assert provenance["content"].source == ( + "${{ FN.uuid4() }} ${{ SECRETS.api.TOKEN }}" + ) def test_partition_redacts_before_evaluation_and_restores_order() -> None: @@ -351,3 +607,11 @@ def test_partition_restores_preserved_subtree_without_reordering() -> None: "patch_ops": source, "validate_only": False, } + + +def test_derive_secret_dependencies_does_not_taint_step_results() -> None: + dependencies = derive_secret_dependencies( + {"value": "${{ steps.normalize.result }}"} + ) + + assert dependencies == {"value": False} diff --git a/tests/unit/test_executor_service.py b/tests/unit/test_executor_service.py index eb7a85227..d9a69bf43 100644 --- a/tests/unit/test_executor_service.py +++ b/tests/unit/test_executor_service.py @@ -16,7 +16,9 @@ from tracecat.dsl.common import create_default_execution_context from tracecat.dsl.schemas import ActionStatement, RunActionInput, RunContext from tracecat.exceptions import TracecatCredentialsError +from tracecat.executor import expression_policy as expression_policy_module from tracecat.executor import service as executor_service +from tracecat.executor.expression_policy import derive_source_provenance from tracecat.executor.schemas import ( ActionImplementation, ExecutorResultSuccess, @@ -485,7 +487,7 @@ def _patch_expression_policy_resolution( ("action_name", "preserved_parameter", "runtime_parameter"), [ ("core.workflow.edit_workflow", "patch_ops", "workflow_id"), - ("core.workflow.create_workflow", "definition_yaml", "title"), + ("core.workflow.create_workflow", "definition_yaml", "unmapped_parameter"), ], ) @pytest.mark.anyio @@ -617,8 +619,8 @@ async def test_prepare_resolved_context_redacts_secrets_before_collection( [ ("core.transform.reshape", "patch_ops"), ("core.transform.reshape", "content"), + ("core.transform.reshape", "title"), ("core.workflow.edit_workflow", "workflow_id"), - ("core.workflow.create_workflow", "title"), ], ) @pytest.mark.anyio @@ -649,6 +651,10 @@ async def test_prepare_resolved_context_resolves_unmapped_parameters( assert get_workspace_variables.await_args.kwargs["variable_exprs"] == set() +def _policy_source_provenance(args: dict[str, object]): + return derive_source_provenance(args) + + @pytest.mark.parametrize( ("step_action", "step_args", "source_value", "evaluated_value", "expected_args"), [ @@ -792,7 +798,7 @@ async def test_template_step_applies_target_action_expression_policy( ctx=executor_service.DispatchActionContext(role=role), resolved_context=parent_resolved, timeout=30, - source_args={"value": source_value}, + source_provenance=_policy_source_provenance({"value": source_value}), ) assert result == {"persisted": True} @@ -800,6 +806,78 @@ async def test_template_step_applies_target_action_expression_policy( assert step_resolved.evaluated_args == expected_args +@pytest.mark.anyio +async def test_template_step_result_is_not_tainted_by_its_arguments(mocker): + """Step results stay runtime data across the accepted implementation boundary.""" + source_value = "${{ SECRETS.runtime.TOKEN }}" + action_input = _expression_policy_input( + "testing.policy_wrapper", + {"value": source_value}, + ) + role = _expression_policy_role("tracecat-executor") + parent_resolved = _policy_wrapper_resolved( + action_input, + role, + steps=[ + { + "ref": "normalize", + "action": "core.transform.reshape", + "args": {"value": "${{ inputs.value }}"}, + }, + { + "ref": "persist", + "action": "core.cases.create_comment", + "args": { + "case_id": "case-123", + "content": "${{ steps.normalize.result }}", + }, + }, + ], + evaluated_args={"value": "runtime-secret"}, + variables={}, + secrets={"runtime": {"TOKEN": "runtime-secret"}}, + ) + mocker.patch.object( + executor_service.registry_resolver, + "resolve_action", + new=mocker.AsyncMock( + side_effect=[ + ActionImplementation(type="udf", action_name="core.transform.reshape"), + ActionImplementation( + type="udf", action_name="core.cases.create_comment" + ), + ] + ), + ) + mocker.patch.object( + executor_service, + "_mint_action_executor_token", + return_value="step-token", + ) + backend = mocker.Mock() + backend.execute = mocker.AsyncMock( + side_effect=[ + ExecutorResultSuccess(result="runtime-secret"), + ExecutorResultSuccess(result={"persisted": True}), + ] + ) + + await executor_service._execute_template_action( + backend=backend, + input=action_input, + ctx=executor_service.DispatchActionContext(role=role), + resolved_context=parent_resolved, + timeout=30, + source_provenance=_policy_source_provenance({"value": source_value}), + ) + + sink_resolved = backend.execute.await_args_list[1].kwargs["resolved_context"] + assert sink_resolved.evaluated_args == { + "case_id": "case-123", + "content": "runtime-secret", + } + + @pytest.mark.anyio async def test_compound_secret_dependency_reaches_nested_template_sink(mocker): """Secret dependency survives composition and a nested template boundary.""" @@ -877,7 +955,7 @@ async def test_compound_secret_dependency_reaches_nested_template_sink(mocker): ctx=executor_service.DispatchActionContext(role=role), resolved_context=parent_resolved, timeout=30, - source_args={"value": source_value}, + source_provenance=_policy_source_provenance({"value": source_value}), ) sink_resolved = backend.execute.await_args.kwargs["resolved_context"] @@ -973,7 +1051,7 @@ async def test_template_step_result_stays_inert_in_redact_parameter(mocker): ctx=executor_service.DispatchActionContext(role=role), resolved_context=parent_resolved, timeout=30, - source_args={}, + source_provenance={}, ) persist_resolved = backend.execute.await_args_list[1].kwargs["resolved_context"] @@ -984,8 +1062,10 @@ async def test_template_step_result_stays_inert_in_redact_parameter(mocker): @pytest.mark.anyio -async def test_template_step_skips_source_expansion_without_policy_parameters(mocker): - """Pure-RESOLVE UDF steps never pay for the source expansion walk.""" +async def test_template_step_skips_source_substitution_without_policy_parameters( + mocker, +): + """Pure-RESOLVE UDF steps never pay for the source substitution walk.""" action_input = _expression_policy_input("testing.policy_wrapper", {}) role = _expression_policy_role("tracecat-executor") parent_resolved = _policy_wrapper_resolved( @@ -1017,10 +1097,9 @@ async def test_template_step_skips_source_expansion_without_policy_parameters(mo backend.execute = mocker.AsyncMock( return_value=ExecutorResultSuccess(result={"ok": True}) ) - expansion_spy = mocker.patch.object( - executor_service, - "expand_template_source_references", - wraps=executor_service.expand_template_source_references, + substitution_spy = mocker.patch( + "tracecat.executor.expression_policy.substitute_source_references", + wraps=expression_policy_module.substitute_source_references, ) await executor_service._execute_template_action( @@ -1029,9 +1108,9 @@ async def test_template_step_skips_source_expansion_without_policy_parameters(mo ctx=executor_service.DispatchActionContext(role=role), resolved_context=parent_resolved, timeout=30, - source_args={"value": "hello"}, + source_provenance=_policy_source_provenance({"value": "hello"}), ) - expansion_spy.assert_not_called() + substitution_spy.assert_not_called() step_resolved = backend.execute.await_args.kwargs["resolved_context"] assert step_resolved.evaluated_args == {"value": "hello"} diff --git a/tracecat/executor/expression_policy.py b/tracecat/executor/expression_policy.py index f909b7412..de3579f17 100644 --- a/tracecat/executor/expression_policy.py +++ b/tracecat/executor/expression_policy.py @@ -6,8 +6,9 @@ from enum import StrEnum from typing import Any, NamedTuple, cast -from lark import Token +from lark import Token, Tree +from tracecat.dsl.schemas import TemplateExecutionContext from tracecat.exceptions import TracecatExpressionError from tracecat.expressions import patterns from tracecat.expressions.common import ExprContext, eval_jsonpath @@ -46,6 +47,8 @@ class ActionParameter(NamedTuple): POLICY_MAP: Mapping[ActionParameter, ExpressionPolicy] = { ActionParameter("core.workflow.edit_workflow", "patch_ops"): _PRESERVE, ActionParameter("core.workflow.create_workflow", "definition_yaml"): _PRESERVE, + ActionParameter("core.workflow.create_workflow", "title"): _REDACT_SECRETS, + ActionParameter("core.workflow.create_workflow", "description"): _REDACT_SECRETS, ActionParameter("core.cases.create_case", "summary"): _REDACT_SECRETS, ActionParameter("core.cases.create_case", "description"): _REDACT_SECRETS, ActionParameter("core.cases.create_case", "fields"): _REDACT_SECRETS, @@ -66,34 +69,54 @@ class ActionParameter(NamedTuple): } -def _has_secret_dependency(dependency: SecretDependency) -> bool: +def has_secret_dependency(dependency: SecretDependency) -> bool: + """Collapse a dependency tree to whether any part depends on secrets.""" match dependency: case bool(): return dependency case list(): - return any(_has_secret_dependency(item) for item in dependency) + return any(has_secret_dependency(item) for item in dependency) case dict(): - return any(_has_secret_dependency(item) for item in dependency.values()) + return any(has_secret_dependency(item) for item in dependency.values()) -def _input_dependency( +def _scoped_dependency( path: str, - input_dependencies: SecretDependencies, + dependencies: SecretDependencies, + context: ExprContext, ) -> SecretDependency | None: dependency = eval_jsonpath( - f"{ExprContext.TEMPLATE_ACTION_INPUTS}{path}", - {ExprContext.TEMPLATE_ACTION_INPUTS: input_dependencies}, + f"{context}{path}", + {context: dependencies}, ) if dependency is not None: return cast(SecretDependency, dependency) - # A compound expression can collapse a structured input's dependency to - # True. Any later access beneath that input must remain conservatively + # A compound expression can collapse a structured entry's dependency to + # True. Any later access beneath that entry must remain conservatively # secret-dependent even though the boolean tree has no child to traverse. match = re.match(r"^\.([A-Za-z_][A-Za-z0-9_]*)", path) if match is None: - return True if _has_secret_dependency(dict(input_dependencies)) else None - return input_dependencies.get(match.group(1)) + return True if has_secret_dependency(dict(dependencies)) else None + return dependencies.get(match.group(1)) + + +def _scope_reference_depends_on_secrets( + parse_tree: Tree[Token], + node_name: str, + dependencies: SecretDependencies, + context: ExprContext, +) -> bool: + for node in parse_tree.find_data(node_name): + token = node.children[0] + if not isinstance(token, Token): + raise TracecatExpressionError( + f"Expected {node_name} path token, got {type(token).__name__}" + ) + dependency = _scoped_dependency(str(token), dependencies, context) + if dependency is not None and has_secret_dependency(dependency): + return True + return False def _expression_depends_on_secrets( @@ -107,18 +130,13 @@ def _expression_depends_on_secrets( ) if next(parse_tree.find_data("secrets"), None) is not None: return True - if not input_dependencies: - return False - - for node in parse_tree.find_data("template_action_inputs"): - token = node.children[0] - if not isinstance(token, Token): - raise TracecatExpressionError( - f"Expected template input path token, got {type(token).__name__}" - ) - dependency = _input_dependency(str(token), input_dependencies) - if dependency is not None and _has_secret_dependency(dependency): - return True + if input_dependencies and _scope_reference_depends_on_secrets( + parse_tree, + "template_action_inputs", + input_dependencies, + ExprContext.TEMPLATE_ACTION_INPUTS, + ): + return True return False @@ -128,11 +146,11 @@ def _redact_secret_string( ) -> str: def replace(match: re.Match[str]) -> str: expression = match.group("expr") - if expression and _expression_depends_on_secrets( + if not expression or not _expression_depends_on_secrets( expression, input_dependencies ): - return MASK_VALUE - return match.group("template") + return match.group("template") + return MASK_VALUE return patterns.TEMPLATE_STRING.sub(replace, value) @@ -153,6 +171,9 @@ def redact_secret_expressions( redacted: dict[Any, Any] = {} for key, item in value.items(): if isinstance(key, str): + # Keys never take the projection path: a masked or + # projected key is ambiguous, so secret-dependent keys + # are rejected outright. redacted_key = _redact_secret_string(key, input_dependencies) if redacted_key != key: raise TracecatExpressionError( @@ -195,7 +216,11 @@ def derive_secret_dependencies( raise TracecatExpressionError( "Expected template input path token" ) - dependency = _input_dependency(str(token), input_dependencies) + dependency = _scoped_dependency( + str(token), + input_dependencies, + ExprContext.TEMPLATE_ACTION_INPUTS, + ) return dependency if dependency is not None else False return any( @@ -214,7 +239,7 @@ def derive_secret_dependencies( } if any( isinstance(key, str) - and _has_secret_dependency( + and has_secret_dependency( derive_secret_dependencies(key, input_dependencies) ) for key in value @@ -225,108 +250,308 @@ def derive_secret_dependencies( return False -# Context references a template's evaluation scope cannot resolve. Caller -# source containing them must stay behind its inputs.* reference; the -# dependency tree still governs redaction for that reference. -_NON_PORTABLE_NODES = frozenset( - { - "actions", - "trigger", - "local_vars", - "local_vars_assignment", - "template_action_inputs", - "template_action_steps", - } -) - - -def _is_portable_source(value: Any) -> bool: - """Whether expanded caller source can evaluate inside a template scope.""" - match value: - case str(): - for match_ in patterns.TEMPLATE_STRING.finditer(value): - expression = match_.group("expr") - if not expression: - continue - try: - parse_tree = parser.parse(expression) - except TracecatExpressionError: - return False - if parse_tree is None: - return False - if any( - subtree.data in _NON_PORTABLE_NODES - for subtree in parse_tree.iter_subtrees() - ): - return False - return True - case list(): - return all(_is_portable_source(item) for item in value) - case dict(): - return all( - _is_portable_source(key) and _is_portable_source(item) - for key, item in value.items() - ) - case _: - return True +def _direct_template_input_path(template: str) -> str | None: + """Return the path for an exact inputs.* reference.""" + if patterns.STANDALONE_TEMPLATE.match(template) is None: + return None + match = patterns.TEMPLATE_STRING.fullmatch(template) + if match is None or not (expression := match.group("expr")): + return None + parse_tree = parser.parse(expression) + if parse_tree is None or parse_tree.data != "template_action_inputs": + return None + token = parse_tree.children[0] + if not isinstance(token, Token): + raise TracecatExpressionError("Expected template input path token") + return str(token) def _is_direct_template_reference(template: str) -> bool: """Return whether a template is exactly an inputs.* reference. - Never expand steps.*: step results are materialized runtime data, and - splicing them in as evaluable source would let fetched content execute - expressions. + Never substitute steps.*: step results are materialized runtime data, + not authored source. """ - if patterns.STANDALONE_TEMPLATE.match(template) is None: - return False - match = patterns.TEMPLATE_STRING.fullmatch(template) - if match is None or not (expression := match.group("expr")): - return False - parse_tree = parser.parse(expression) - return parse_tree is not None and parse_tree.data == "template_action_inputs" + return _direct_template_input_path(template) is not None + + +def _redact_runtime_value(value: Any, dependency: SecretDependency) -> Any: + """Mask materialized input data according to its authored dependency tree. + + Runtime containers retain their shape. Mapping dependencies are paired by + insertion order because non-secret expressions may change authored keys + during the caller's normal evaluation. + """ + match dependency: + case bool(): + if not dependency: + return value + if isinstance(value, list): + return [_redact_runtime_value(item, True) for item in value] + if isinstance(value, Mapping): + return { + key: _redact_runtime_value(item, True) + for key, item in value.items() + } + return MASK_VALUE + case list(): + if not isinstance(value, list) or len(value) != len(dependency): + return _redact_runtime_value(value, has_secret_dependency(dependency)) + return [ + _redact_runtime_value(item, item_dependency) + for item, item_dependency in zip(value, dependency, strict=True) + ] + case dict(): + key_dependency = dependency.get(_SECRET_KEY_DEPENDENCY, False) + if has_secret_dependency(key_dependency): + raise TracecatExpressionError( + "Secret expressions are not allowed in dictionary keys", + detail={"code": "secret_expression_in_key"}, + ) + value_dependencies = [ + item_dependency + for key, item_dependency in dependency.items() + if key is not _SECRET_KEY_DEPENDENCY + ] + if not isinstance(value, Mapping) or len(value) != len(value_dependencies): + return _redact_runtime_value(value, has_secret_dependency(dependency)) + return { + key: _redact_runtime_value(item, item_dependency) + for (key, item), item_dependency in zip( + value.items(), value_dependencies, strict=True + ) + } -def expand_template_source_references(value: Any, context: Mapping[str, Any]) -> Any: - """Expand direct template input references while retaining nested source.""" +def substitute_source_references(value: Any, source_operand: Mapping[str, Any]) -> Any: + """Replace direct template input references with the caller's raw source. + + Substitution splices authored source as inert data; it never evaluates + it, so the result is safe regardless of which contexts the source + references. Unresolvable references are left as written. + """ match value: case str() if _is_direct_template_reference(value): - expanded = TemplateExpression(value, operand=context).result() - return expanded if _is_portable_source(expanded) else value + substituted = TemplateExpression(value, operand=source_operand).result() + return value if substituted is None else substituted case str(): def replace(match: re.Match[str]) -> str: template = match.group("template") if _is_direct_template_reference(template): - expanded = TemplateExpression(template, operand=context).result() - if _is_portable_source(expanded): - return str(expanded) + substituted = TemplateExpression( + template, operand=source_operand + ).result() + if substituted is not None: + return str(substituted) return template return patterns.TEMPLATE_STRING.sub(replace, value) case list(): - return [expand_template_source_references(item, context) for item in value] + return [ + substitute_source_references(item, source_operand) for item in value + ] case dict(): - expanded: dict[Any, Any] = {} + substituted: dict[Any, Any] = {} for key, item in value.items(): - expanded_key = ( - expand_template_source_references(key, context) + substituted_key = ( + substitute_source_references(key, source_operand) if isinstance(key, str) else key ) - if expanded_key in expanded: + if substituted_key in substituted: raise TracecatExpressionError( - "Template source expansion produced a duplicate dictionary key", + "Source substitution produced a duplicate dictionary key", detail={"code": "template_source_key_collision"}, ) - expanded[expanded_key] = expand_template_source_references( - item, context + substituted[substituted_key] = substitute_source_references( + item, source_operand ) - return expanded + return substituted case _: return value +def _is_resolvable_carrier( + value: Any, + input_dependencies: SecretDependencies | None, +) -> bool: + """Whether a preserved value is a dynamic carrier safe to evaluate. + + A bare standalone expression can never itself be valid preserved source + (patch ops must be a list, a definition must be YAML text), so the caller + is constructing the source dynamically. Evaluate it unless it depends on + secrets; secret-dependent carriers stay preserved and fail loudly at the + action's own validation instead of leaking. + """ + if not isinstance(value, str): + return False + if patterns.STANDALONE_TEMPLATE.match(value) is None: + return False + match = patterns.TEMPLATE_STRING.fullmatch(value) + if match is None or not (expression := match.group("expr")): + return False + try: + return not _expression_depends_on_secrets(expression, input_dependencies) + except TracecatExpressionError: + return False + + +@dataclass(frozen=True, slots=True) +class SourceProvenance: + """Authored source and secret dependency for one template argument.""" + + source: Any + dependency: SecretDependency + + +def derive_source_provenance( + args: Mapping[str, Any], +) -> dict[str, SourceProvenance]: + """Build unevaluated caller-boundary provenance for a template invocation.""" + return { + parameter: SourceProvenance( + source=value, + dependency=derive_secret_dependencies(value), + ) + for parameter, value in args.items() + } + + +class TemplateExecutionState: + """Pairs a template's materialized context with argument provenance.""" + + def __init__( + self, + context: TemplateExecutionContext, + provenance: Mapping[str, SourceProvenance], + ) -> None: + self._context = context + self._provenance = dict(provenance) + + @property + def context(self) -> TemplateExecutionContext: + return self._context + + def _input_dependencies(self) -> dict[str, SecretDependency]: + return { + parameter: provenance.dependency + for parameter, provenance in self._provenance.items() + } + + def _source_operand(self) -> dict[str, Any]: + # Defaulted parameters have no provenance; their validated values are + # the template author's literals and stand in as source. + sources: dict[str, Any] = dict(self._context.get("inputs") or {}) + for parameter, provenance in self._provenance.items(): + sources[parameter] = provenance.source + return {str(ExprContext.TEMPLATE_ACTION_INPUTS): sources} + + def record_step(self, ref: str, result: Any) -> None: + """Store materialized step data without interpreting it as source.""" + self._context["steps"][ref] = result + + def prepare_step_args(self, action: str, args: Mapping[str, Any]) -> dict[str, Any]: + """Apply each parameter's policy and evaluate one step's arguments.""" + prepared: dict[str, Any] = {} + for parameter, value in args.items(): + match expression_policy(action, parameter): + case ExpressionPolicy.RESOLVE: + prepared[parameter] = eval_templated_object( + value, operand=cast(Mapping[str, Any], self._context) + ) + case ExpressionPolicy.REDACT_SECRETS: + prepared[parameter] = self.project_redacted(value) + case ExpressionPolicy.PRESERVE: + prepared[parameter] = self._preserve(value) + return prepared + + def project_redacted(self, value: Any) -> Any: + """Evaluate a protected field while masking authored input dependencies. + + Exact tainted input references are returned directly from materialized + runtime data after tree-shaped masking. They are never recursively + evaluated as expression source. Compound tainted expressions are + replaced before ordinary evaluation. + """ + input_dependencies = self._input_dependencies() + match value: + case str(): + if (path := _direct_template_input_path(value)) is not None: + dependency = _scoped_dependency( + path, + input_dependencies, + ExprContext.TEMPLATE_ACTION_INPUTS, + ) + if dependency is not None and has_secret_dependency(dependency): + runtime_value = eval_jsonpath( + f"{ExprContext.TEMPLATE_ACTION_INPUTS}{path}", + cast(Mapping[str, Any], self._context), + ) + return _redact_runtime_value(runtime_value, dependency) + redacted = redact_secret_expressions(value, input_dependencies) + return eval_templated_object( + redacted, operand=cast(Mapping[str, Any], self._context) + ) + case list(): + return [self.project_redacted(item) for item in value] + case dict(): + projected: dict[Any, Any] = {} + for key, item in value.items(): + if isinstance(key, str): + redacted_key = redact_secret_expressions( + key, input_dependencies + ) + if redacted_key != key: + raise TracecatExpressionError( + "Secret expressions are not allowed in dictionary keys", + detail={"code": "secret_expression_in_key"}, + ) + projected_key = eval_templated_object( + key, operand=cast(Mapping[str, Any], self._context) + ) + else: + projected_key = key + if projected_key in projected: + raise TracecatExpressionError( + "Redaction produced a duplicate dictionary key", + detail={"code": "redacted_key_collision"}, + ) + projected[projected_key] = self.project_redacted(item) + return projected + case _: + return value + + def _preserve(self, value: Any) -> Any: + substituted = substitute_source_references(value, self._source_operand()) + if _is_resolvable_carrier(substituted, self._input_dependencies()): + # A bare expression is never valid preserved source. When the + # caller's own source is the carrier, its runtime value is + # already materialized in inputs — evaluate the original + # reference in this scope instead of the caller's text. + target = ( + value + if isinstance(value, str) and substituted != value + else substituted + ) + return eval_templated_object( + target, operand=cast(Mapping[str, Any], self._context) + ) + return substituted + + def child_provenance(self, args: Mapping[str, Any]) -> dict[str, SourceProvenance]: + """Build provenance for a nested template invocation's arguments.""" + provenance: dict[str, SourceProvenance] = {} + for parameter, value in args.items(): + provenance[parameter] = SourceProvenance( + source=substitute_source_references(value, self._source_operand()), + dependency=derive_secret_dependencies( + value, self._input_dependencies() + ), + ) + return provenance + + @dataclass(frozen=True, slots=True) class PartitionedActionArgs: """Action arguments split into resolvable and preserved parameters.""" @@ -338,12 +563,7 @@ class PartitionedActionArgs: def merge(self, evaluated: Mapping[str, Any]) -> dict[str, Any]: """Restore preserved values without changing parameter order.""" return { - parameter: ( - value - if expression_policy(self.action, parameter) - is ExpressionPolicy.PRESERVE - else evaluated[parameter] - ) + parameter: (evaluated[parameter] if parameter in self.resolvable else value) for parameter, value in self.original.items() } @@ -356,18 +576,21 @@ def expression_policy(action: str, parameter: str) -> ExpressionPolicy: def partition_action_args( action: str, args: Mapping[str, Any], - input_dependencies: SecretDependencies | None = None, ) -> PartitionedActionArgs: - """Apply pre-evaluation policy and exclude preserved source subtrees.""" + """Apply pre-evaluation policy and exclude preserved source subtrees. + + Root-level boundary: template steps route through + ``TemplateExecutionState.prepare_step_args`` instead, which carries + input provenance. + """ resolvable: dict[str, Any] = {} for parameter, value in args.items(): match expression_policy(action, parameter): case ExpressionPolicy.PRESERVE: - continue + if _is_resolvable_carrier(value, None): + resolvable[parameter] = value case ExpressionPolicy.REDACT_SECRETS: - resolvable[parameter] = redact_secret_expressions( - value, input_dependencies - ) + resolvable[parameter] = redact_secret_expressions(value) case ExpressionPolicy.RESOLVE: resolvable[parameter] = value return PartitionedActionArgs( @@ -381,10 +604,9 @@ def prepare_action_args( action: str, args: Mapping[str, Any], context: Mapping[str, Any], - input_dependencies: SecretDependencies | None = None, ) -> dict[str, Any]: - """Apply field policy and evaluate one action's arguments.""" - partitioned = partition_action_args(action, args, input_dependencies) + """Apply field policy and evaluate one action's arguments at the root.""" + partitioned = partition_action_args(action, args) evaluated = cast( Mapping[str, Any], eval_templated_object(partitioned.resolvable, operand=context), diff --git a/tracecat/executor/service.py b/tracecat/executor/service.py index ab7fdf740..0b1ee26e4 100644 --- a/tracecat/executor/service.py +++ b/tracecat/executor/service.py @@ -46,12 +46,9 @@ from tracecat.executor import registry_resolver from tracecat.executor.backends.base import ExecutorBackend from tracecat.executor.expression_policy import ( - ExpressionPolicy, - SecretDependencies, - SecretDependency, - derive_secret_dependencies, - expand_template_source_references, - expression_policy, + SourceProvenance, + TemplateExecutionState, + derive_source_provenance, partition_action_args, prepare_action_args, ) @@ -456,8 +453,7 @@ async def _execute_template_action( ctx: DispatchActionContext, resolved_context: ResolvedContext, timeout: float, - source_args: Mapping[str, Any], - source_dependencies: SecretDependencies | None = None, + source_provenance: Mapping[str, SourceProvenance], ) -> Any: """Execute a template action by orchestrating its steps. @@ -473,8 +469,7 @@ async def _execute_template_action( ctx: Dispatch context containing the role resolved_context: Pre-resolved context with secrets and template definition timeout: Execution timeout - source_args: Unevaluated arguments supplied to this template invocation - source_dependencies: Secret dependencies derived by the caller + source_provenance: Caller-boundary provenance for this invocation's args Returns: The evaluated returns expression result @@ -518,17 +513,7 @@ async def _execute_template_action( inputs=validated_input_args, steps={}, ) - source_input_args = dict(validated_input_args) - source_input_args.update(source_args) - input_dependencies = cast( - dict[str, SecretDependency], - derive_secret_dependencies(source_input_args), - ) - if source_dependencies: - input_dependencies.update(source_dependencies) - # Source expansion is a raw inputs.* lookup. Do not expose runtime secrets, - # variables, environment, or step results to this analysis context. - source_context: Mapping[str, Any] = {"inputs": source_input_args} + state = TemplateExecutionState(template_context, source_provenance) logger.info( "Executing template action via backend", @@ -544,34 +529,7 @@ async def _execute_template_action( step_action=step.action, ) - # Non-resolving policies need the caller's source form for direct input - # references. The parallel dependency tree covers compound expressions - # that cannot be source-expanded without rewriting their AST. - source_step_args: dict[str, Any] | None = None - policy_args: Mapping[str, Any] = step.args - if any( - expression_policy(step.action, parameter) is not ExpressionPolicy.RESOLVE - for parameter in step.args - ): - source_step_args = cast( - dict[str, Any], - expand_template_source_references(step.args, source_context), - ) - policy_args = { - parameter: ( - source_step_args[parameter] - if expression_policy(step.action, parameter) - is not ExpressionPolicy.RESOLVE - else value - ) - for parameter, value in step.args.items() - } - evaled_args = prepare_action_args( - step.action, - policy_args, - template_context, - input_dependencies, - ) + evaled_args = state.prepare_step_args(step.action, step.args) # Prepare step context (reuses parent secrets, no re-fetch) step_resolved = await _prepare_step_context( @@ -582,23 +540,11 @@ async def _execute_template_action( role=role, ) - # Nested templates still route source to their own policy steps. - if source_step_args is None: - source_step_args = ( - cast( - dict[str, Any], - expand_template_source_references(step.args, source_context), - ) - if step_resolved.action_impl.type == "template" - else {} - ) - step_dependencies = ( - cast( - dict[str, SecretDependency], - derive_secret_dependencies(step.args, input_dependencies), - ) + # Nested templates receive provenance derived in this scope. + child_provenance = ( + state.child_provenance(step.args) if step_resolved.action_impl.type == "template" - else None + else {} ) # Execute step via _invoke_step (handles nested templates) @@ -609,8 +555,7 @@ async def _execute_template_action( input=input, ctx=ctx, timeout=timeout, - source_args=source_step_args, - source_dependencies=step_dependencies, + source_provenance=child_provenance, ) except ExecutionError: # Re-raise with step context preserved @@ -628,9 +573,10 @@ async def _execute_template_action( ) from e # Store step result for subsequent steps (materialized for expression access) - template_context["steps"][step.ref] = TaskResult.from_result( - step_result - ).to_materialized_dict() + state.record_step( + step.ref, + TaskResult.from_result(step_result).to_materialized_dict(), + ) logger.trace("Template step completed", step_ref=step.ref) # Evaluate returns expression with final template context @@ -643,8 +589,7 @@ async def _invoke_step( input: RunActionInput, ctx: DispatchActionContext, timeout: float, - source_args: Mapping[str, Any], - source_dependencies: SecretDependencies | None = None, + source_provenance: Mapping[str, SourceProvenance], ) -> Any: """Execute a template step. Skips masking (done at root level). @@ -658,8 +603,7 @@ async def _invoke_step( input: The original RunActionInput ctx: Dispatch context containing the role timeout: Execution timeout - source_args: Unevaluated arguments supplied to this action invocation - source_dependencies: Secret dependencies derived by the caller + source_provenance: Caller-boundary provenance for this invocation's args Returns: The step execution result (unmasked) @@ -673,8 +617,7 @@ async def _invoke_step( ctx=ctx, resolved_context=resolved_context, timeout=timeout, - source_args=source_args, - source_dependencies=source_dependencies, + source_provenance=source_provenance, ) case "udf": # Leaf node - execute via backend @@ -703,6 +646,7 @@ class PreparedContext: resolved_context: ResolvedContext mask_values: set[str] | None + source_provenance: Mapping[str, SourceProvenance] | None = None async def _get_template_secret_projection( @@ -800,6 +744,12 @@ async def prepare_resolved_context( ) evaluated_args = prepare_action_args(action_name, task.args, context) + # Preserve authored dependencies for policy-aware template steps. + source_provenance = ( + derive_source_provenance(task.args) + if action_impl.type == "template" + else None + ) finally: ctx_logical_time.reset(logical_time_token) ctx_interaction.reset(interaction_token) @@ -839,7 +789,11 @@ async def prepare_resolved_context( secret_projection=secret_projection, ) - return PreparedContext(resolved_context=resolved_context, mask_values=mask_values) + return PreparedContext( + resolved_context=resolved_context, + mask_values=mask_values, + source_provenance=source_provenance, + ) async def invoke_once( @@ -890,11 +844,7 @@ async def invoke_once( input=input, ctx=ctx, timeout=timeout, - source_args=input.task.args, - source_dependencies=cast( - dict[str, SecretDependency], - derive_secret_dependencies(input.task.args), - ), + source_provenance=prepared.source_provenance or {}, ) except ExecutionError as e: From 27d0dc1508a8fc24574227c2cb25f2aba43dd985 Mon Sep 17 00:00:00 2001 From: Jordan Umusu Date: Thu, 6 Aug 2026 18:07:15 -0400 Subject: [PATCH 4/5] fix(executor): pair mapping redaction by key, not insertion order --- tests/unit/test_executor_expression_policy.py | 61 +++++++++++++++++++ tracecat/executor/expression_policy.py | 18 +++--- 2 files changed, 69 insertions(+), 10 deletions(-) diff --git a/tests/unit/test_executor_expression_policy.py b/tests/unit/test_executor_expression_policy.py index 62c800d1e..95a8d3725 100644 --- a/tests/unit/test_executor_expression_policy.py +++ b/tests/unit/test_executor_expression_policy.py @@ -243,6 +243,67 @@ def test_whole_structured_input_is_tree_masked_without_re_evaluation() -> None: } +def test_reordered_runtime_mapping_masks_by_key() -> None: + # expects-model validation may reorder mapping keys; pairing must not + # rely on insertion order. + state = _state( + source_args={ + "context": { + "safe": "plain", + "token": "${{ SECRETS.api.TOKEN }}", + } + }, + runtime_inputs={ + "context": { + "token": "runtime-secret", + "safe": "plain", + } + }, + ) + + prepared = state.prepare_step_args( + "core.table.insert_row", + {"row_data": "${{ inputs.context }}"}, + ) + + assert prepared == { + "row_data": { + "token": MASK_VALUE, + "safe": "plain", + } + } + + +def test_dynamic_authored_key_falls_back_to_conservative_mask() -> None: + state = _state( + source_args={ + "context": { + "${{ VARS.col }}": "plain", + "token": "${{ SECRETS.api.TOKEN }}", + } + }, + runtime_inputs={ + "context": { + "events": "plain", + "token": "runtime-secret", + } + }, + variables={"col": "events"}, + ) + + prepared = state.prepare_step_args( + "core.table.insert_row", + {"row_data": "${{ inputs.context }}"}, + ) + + assert prepared == { + "row_data": { + "events": MASK_VALUE, + "token": MASK_VALUE, + } + } + + def test_secret_dependent_input_key_is_rejected_only_at_sink() -> None: state = _state( source_args={ diff --git a/tracecat/executor/expression_policy.py b/tracecat/executor/expression_policy.py index de3579f17..1e5edb14c 100644 --- a/tracecat/executor/expression_policy.py +++ b/tracecat/executor/expression_policy.py @@ -279,8 +279,8 @@ def _redact_runtime_value(value: Any, dependency: SecretDependency) -> Any: """Mask materialized input data according to its authored dependency tree. Runtime containers retain their shape. Mapping dependencies are paired by - insertion order because non-secret expressions may change authored keys - during the caller's normal evaluation. + key; validation may reorder mappings, and authored keys that change under + evaluation fall back to the collapsed conservative mask. """ match dependency: case bool(): @@ -308,18 +308,16 @@ def _redact_runtime_value(value: Any, dependency: SecretDependency) -> Any: "Secret expressions are not allowed in dictionary keys", detail={"code": "secret_expression_in_key"}, ) - value_dependencies = [ - item_dependency + value_dependencies = { + key: item_dependency for key, item_dependency in dependency.items() if key is not _SECRET_KEY_DEPENDENCY - ] - if not isinstance(value, Mapping) or len(value) != len(value_dependencies): + } + if not isinstance(value, Mapping) or set(value) != set(value_dependencies): return _redact_runtime_value(value, has_secret_dependency(dependency)) return { - key: _redact_runtime_value(item, item_dependency) - for (key, item), item_dependency in zip( - value.items(), value_dependencies, strict=True - ) + key: _redact_runtime_value(item, value_dependencies[key]) + for key, item in value.items() } From a210b9db65a650c9feb1b56b474857cadac90a04 Mon Sep 17 00:00:00 2001 From: Jordan Umusu Date: Thu, 6 Aug 2026 18:24:36 -0400 Subject: [PATCH 5/5] fix(executor): redact secrets in preset and table definition metadata --- tests/unit/test_executor_expression_policy.py | 56 +++++++++++++++++++ tracecat/executor/expression_policy.py | 13 ++++- 2 files changed, 67 insertions(+), 2 deletions(-) diff --git a/tests/unit/test_executor_expression_policy.py b/tests/unit/test_executor_expression_policy.py index 95a8d3725..6330c849f 100644 --- a/tests/unit/test_executor_expression_policy.py +++ b/tests/unit/test_executor_expression_policy.py @@ -73,12 +73,20 @@ def test_action_parameter_policy_scope_is_explicit() -> None: ActionParameter(action="core.cases.create_comment", parameter="content"), ActionParameter(action="core.cases.reply_to_comment", parameter="content"), ActionParameter(action="core.cases.update_comment", parameter="content"), + ActionParameter(action="core.table.create_table", parameter="columns"), + ActionParameter(action="core.table.create_column", parameter="column"), + ActionParameter(action="core.table.update_column", parameter="update"), ActionParameter(action="core.table.insert_row", parameter="row_data"), ActionParameter(action="core.table.insert_rows", parameter="rows_data"), ActionParameter(action="core.table.update_row", parameter="row_data"), ActionParameter(action="core.cases.insert_row", parameter="row"), ActionParameter(action="ai.agent.create_preset", parameter="instructions"), ActionParameter(action="ai.agent.update_preset", parameter="instructions"), + ActionParameter(action="ai.agent.create_preset", parameter="name"), + ActionParameter(action="ai.agent.create_preset", parameter="description"), + ActionParameter(action="ai.agent.update_preset", parameter="name"), + ActionParameter(action="ai.agent.update_preset", parameter="description"), + ActionParameter(action="ai.agent.update_preset", parameter="new_slug"), } @@ -466,6 +474,54 @@ def test_workflow_metadata_redacts_secret_expressions() -> None: } +def test_preset_metadata_redacts_secret_expressions() -> None: + prepared = prepare_action_args( + "ai.agent.update_preset", + { + "slug": "analyst", + "description": "Uses ${{ SECRETS.api.KEY }}", + "name": "Analyst ${{ VARS.api.env }}", + }, + {"VARS": {"api": {"env": "prod"}}}, + ) + + assert prepared == { + "slug": "analyst", + "description": f"Uses {MASK_VALUE}", + "name": "Analyst prod", + } + + +def test_column_definitions_redact_secret_expressions() -> None: + prepared = prepare_action_args( + "core.table.create_table", + { + "name": "alerts", + "columns": [ + { + "name": "token", + "type": "TEXT", + "default": "${{ SECRETS.api.KEY }}", + }, + { + "name": "region", + "type": "TEXT", + "default": "${{ VARS.api.region }}", + }, + ], + }, + {"VARS": {"api": {"region": "us-east-1"}}}, + ) + + assert prepared == { + "name": "alerts", + "columns": [ + {"name": "token", "type": "TEXT", "default": MASK_VALUE}, + {"name": "region", "type": "TEXT", "default": "us-east-1"}, + ], + } + + def test_preserve_substitutes_caller_source_for_direct_reference() -> None: ops = [ { diff --git a/tracecat/executor/expression_policy.py b/tracecat/executor/expression_policy.py index 1e5edb14c..d5e6986e1 100644 --- a/tracecat/executor/expression_policy.py +++ b/tracecat/executor/expression_policy.py @@ -4,7 +4,7 @@ from collections.abc import Mapping from dataclasses import dataclass from enum import StrEnum -from typing import Any, NamedTuple, cast +from typing import Any, cast from lark import Token, Tree @@ -31,7 +31,8 @@ class ExpressionPolicy(StrEnum): REDACT_SECRETS = "redact_secrets" -class ActionParameter(NamedTuple): +@dataclass(frozen=True, slots=True) +class ActionParameter: """Identifies one parameter on a registry action.""" action: str @@ -60,12 +61,20 @@ class ActionParameter(NamedTuple): ActionParameter("core.cases.create_comment", "content"): _REDACT_SECRETS, ActionParameter("core.cases.reply_to_comment", "content"): _REDACT_SECRETS, ActionParameter("core.cases.update_comment", "content"): _REDACT_SECRETS, + ActionParameter("core.table.create_table", "columns"): _REDACT_SECRETS, + ActionParameter("core.table.create_column", "column"): _REDACT_SECRETS, + ActionParameter("core.table.update_column", "update"): _REDACT_SECRETS, ActionParameter("core.table.insert_row", "row_data"): _REDACT_SECRETS, ActionParameter("core.table.insert_rows", "rows_data"): _REDACT_SECRETS, ActionParameter("core.table.update_row", "row_data"): _REDACT_SECRETS, ActionParameter("core.cases.insert_row", "row"): _REDACT_SECRETS, ActionParameter("ai.agent.create_preset", "instructions"): _REDACT_SECRETS, ActionParameter("ai.agent.update_preset", "instructions"): _REDACT_SECRETS, + ActionParameter("ai.agent.create_preset", "name"): _REDACT_SECRETS, + ActionParameter("ai.agent.create_preset", "description"): _REDACT_SECRETS, + ActionParameter("ai.agent.update_preset", "name"): _REDACT_SECRETS, + ActionParameter("ai.agent.update_preset", "description"): _REDACT_SECRETS, + ActionParameter("ai.agent.update_preset", "new_slug"): _REDACT_SECRETS, }