diff --git a/tests/unit/test_executor_expression_policy.py b/tests/unit/test_executor_expression_policy.py new file mode 100644 index 000000000..6330c849f --- /dev/null +++ b/tests/unit/test_executor_expression_policy.py @@ -0,0 +1,734 @@ +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, + 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 + 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.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"), + 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.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"), + } + + +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}, + ] + } + + +@pytest.mark.parametrize( + "expression", + [ + '${{ inputs.content || "fallback" }}', + "${{ FN.to_base64(inputs.content) }}", + ], +) +def test_compound_template_input_dependency_is_redacted(expression: str) -> None: + state = _state( + source_args={"content": "${{ SECRETS.api.TOKEN }}"}, + runtime_inputs={"content": "runtime-secret"}, + ) + + 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: + state = _state( + source_args={ + "context": { + "safe": "plain value", + "token": "${{ SECRETS.api.TOKEN }}", + } + }, + runtime_inputs={ + "context": { + "safe": "plain value", + "token": "runtime-secret", + } + }, + ) + + prepared = state.prepare_step_args( + "core.cases.create_case", + { + "payload": { + "safe": "${{ inputs.context.safe }}", + "token": "${{ inputs.context.token }}", + } + }, + ) + + assert prepared == { + "payload": { + "safe": "plain value", + "token": MASK_VALUE, + } + } + + +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", + } + }, + ) + + prepared = state.prepare_step_args( + "core.table.insert_row", + {"table": "events", "row_data": "${{ inputs.context }}"}, + ) + + assert prepared == { + "table": "events", + "row_data": { + "safe": "${{ SECRETS.injected.VALUE }}", + "token": MASK_VALUE, + }, + } + + +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={ + "content": "plain", + "context": {"${{ SECRETS.api.KEY }}": "value"}, + }, + runtime_inputs={ + "content": "plain", + "context": {"runtime-key": "value"}, + }, + ) + + 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": "${{ ACTIONS.fetch.result.body }}"}}, + ) + + 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.""" + state = _state( + source_args={"content": "${{ ACTIONS.fetch.result.body }}"}, + runtime_inputs={"content": "evaluated-upstream-value"}, + ) + + prepared = state.prepare_step_args( + "core.cases.create_comment", + {"case_id": "case-123", "content": "${{ inputs.content }}"}, + ) + + assert prepared == { + "case_id": "case-123", + "content": "evaluated-upstream-value", + } + + +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"}, + ) + + prepared = state.prepare_step_args( + "core.cases.create_comment", {"content": "${{ inputs.content }}"} + ) + + 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 + ) + + 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_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 = [ + { + "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 }}", + } + + +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.workflow.edit_workflow", + { + "workflow_id": "wf-123", + "patch_ops": "${{ ACTIONS.builder.result }}", + "validate_only": False, + }, + {"ACTIONS": {"builder": {"result": ops}}}, + ) + + 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_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]}, + {}, + ) + + 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: + 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, + } + + +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 1febd8711..d9a69bf43 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,25 @@ ) 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 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, + 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 +396,721 @@ 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", "unmapped_parameter"), + ], +) +@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.transform.reshape", "title"), + ("core.workflow.edit_workflow", "workflow_id"), + ], +) +@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() + + +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"), + [ + ( + "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.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", + { + "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_provenance=_policy_source_provenance({"value": source_value}), + ) + + assert result == {"persisted": True} + step_resolved = backend.execute.await_args.kwargs["resolved_context"] + 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.""" + 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_provenance=_policy_source_provenance({"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, + *, + 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 or {}, + 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_provenance={}, + ) + + 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_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( + 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}) + ) + substitution_spy = mocker.patch( + "tracecat.executor.expression_policy.substitute_source_references", + wraps=expression_policy_module.substitute_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_provenance=_policy_source_provenance({"value": "hello"}), + ) + + 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 new file mode 100644 index 000000000..d5e6986e1 --- /dev/null +++ b/tracecat/executor/expression_policy.py @@ -0,0 +1,621 @@ +from __future__ import annotations + +import re +from collections.abc import Mapping +from dataclasses import dataclass +from enum import StrEnum +from typing import Any, cast + +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 +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.""" + + RESOLVE = "resolve" + PRESERVE = "preserve" + REDACT_SECRETS = "redact_secrets" + + +@dataclass(frozen=True, slots=True) +class ActionParameter: + """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.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, + 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.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, +} + + +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) + case dict(): + return any(has_secret_dependency(item) for item in dependency.values()) + + +def _scoped_dependency( + path: str, + dependencies: SecretDependencies, + context: ExprContext, +) -> SecretDependency | None: + dependency = eval_jsonpath( + f"{context}{path}", + {context: dependencies}, + ) + if dependency is not None: + return cast(SecretDependency, dependency) + + # 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(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( + 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 input_dependencies and _scope_reference_depends_on_secrets( + parse_tree, + "template_action_inputs", + input_dependencies, + ExprContext.TEMPLATE_ACTION_INPUTS, + ): + return True + return False + + +def _redact_secret_string( + value: str, + input_dependencies: SecretDependencies | None = None, +) -> str: + def replace(match: re.Match[str]) -> str: + expression = match.group("expr") + if not expression or not _expression_depends_on_secrets( + expression, input_dependencies + ): + return match.group("template") + return MASK_VALUE + + return patterns.TEMPLATE_STRING.sub(replace, value) + + +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, input_dependencies) + case list(): + 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): + # 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( + "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, 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 = _scoped_dependency( + str(token), + input_dependencies, + ExprContext.TEMPLATE_ACTION_INPUTS, + ) + 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 + + +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 substitute steps.*: step results are materialized runtime data, + not authored source. + """ + 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 + key; validation may reorder mappings, and authored keys that change under + evaluation fall back to the collapsed conservative mask. + """ + 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 = { + key: item_dependency + for key, item_dependency in dependency.items() + if key is not _SECRET_KEY_DEPENDENCY + } + 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, value_dependencies[key]) + for key, item in value.items() + } + + +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): + 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): + 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 [ + substitute_source_references(item, source_operand) for item in value + ] + case dict(): + substituted: dict[Any, Any] = {} + for key, item in value.items(): + substituted_key = ( + substitute_source_references(key, source_operand) + if isinstance(key, str) + else key + ) + if substituted_key in substituted: + raise TracecatExpressionError( + "Source substitution produced a duplicate dictionary key", + detail={"code": "template_source_key_collision"}, + ) + substituted[substituted_key] = substitute_source_references( + item, source_operand + ) + 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.""" + + 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: (evaluated[parameter] if parameter in self.resolvable else value) + 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. + + 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: + if _is_resolvable_carrier(value, None): + resolvable[parameter] = value + 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 at the root.""" + 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..0b1ee26e4 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 ( + SourceProvenance, + TemplateExecutionState, + derive_source_provenance, + 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_provenance: Mapping[str, SourceProvenance], ) -> 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_provenance: Caller-boundary provenance for this invocation's args Returns: The evaluated returns expression result @@ -504,6 +513,7 @@ async def _execute_template_action( inputs=validated_input_args, steps={}, ) + state = TemplateExecutionState(template_context, source_provenance) logger.info( "Executing template action via backend", @@ -519,11 +529,7 @@ 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), - ) + 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( @@ -534,6 +540,13 @@ async def _execute_template_action( role=role, ) + # Nested templates receive provenance derived in this scope. + child_provenance = ( + state.child_provenance(step.args) + if step_resolved.action_impl.type == "template" + else {} + ) + # Execute step via _invoke_step (handles nested templates) try: step_result = await _invoke_step( @@ -542,6 +555,7 @@ async def _execute_template_action( input=input, ctx=ctx, timeout=timeout, + source_provenance=child_provenance, ) except ExecutionError: # Re-raise with step context preserved @@ -559,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 @@ -574,6 +589,7 @@ async def _invoke_step( input: RunActionInput, ctx: DispatchActionContext, timeout: float, + source_provenance: Mapping[str, SourceProvenance], ) -> Any: """Execute a template step. Skips masking (done at root level). @@ -587,6 +603,7 @@ async def _invoke_step( input: The original RunActionInput ctx: Dispatch context containing the role timeout: Execution timeout + source_provenance: Caller-boundary provenance for this invocation's args Returns: The step execution result (unmasked) @@ -600,6 +617,7 @@ async def _invoke_step( ctx=ctx, resolved_context=resolved_context, timeout=timeout, + source_provenance=source_provenance, ) case "udf": # Leaf node - execute via backend @@ -628,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( @@ -679,8 +698,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 +743,13 @@ 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) + # 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) @@ -762,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( @@ -813,6 +844,7 @@ async def invoke_once( input=input, ctx=ctx, timeout=timeout, + source_provenance=prepared.source_provenance or {}, ) except ExecutionError as e: