From 4c0acfb3c883ef28d4084442379c0e4c9d620737 Mon Sep 17 00:00:00 2001 From: Arseniy Dugin Date: Tue, 14 Jul 2026 12:17:59 +0000 Subject: [PATCH] Add --arg-secret to pipeline-runs submit for secret-backed inputs Support binding a pipeline input to a Tangle secret via `--arg-secret INPUT=SECRET_NAME` (repeatable) and a matching `arg_secrets` config mapping. Each reference encodes the OSS dynamic-data payload {"dynamicData":{"secret":{"name":...}}} under the input in root_task arguments. Validation runs before any file read or network call: inputs and secret names must be non-empty after trimming, duplicates are rejected, and an input supplied both as a plain value and a secret reference is rejected rather than silently overwritten. CLI `--arg-secret` overrides config `arg_secrets`. Existing --arg, --args-json, config args, hydration, dry-run, and submit-recovery behavior are unchanged. --- .../src/tangle_cli/pipeline_run_manager.py | 77 ++++ .../src/tangle_cli/pipeline_runs_cli.py | 21 +- tests/test_pipeline_runs_cli.py | 336 ++++++++++++++++++ 3 files changed, 433 insertions(+), 1 deletion(-) diff --git a/packages/tangle-cli/src/tangle_cli/pipeline_run_manager.py b/packages/tangle-cli/src/tangle_cli/pipeline_run_manager.py index 24742fa..aa7d6ad 100644 --- a/packages/tangle-cli/src/tangle_cli/pipeline_run_manager.py +++ b/packages/tangle-cli/src/tangle_cli/pipeline_run_manager.py @@ -2100,3 +2100,80 @@ def parse_json_or_key_values( result.update(loaded) result.update(parse_key_value_entries(entries)) return result + + +def secret_argument_value(secret_name: str) -> dict[str, Any]: + """Return the OSS dynamic-data payload for a Tangle secret reference.""" + + return {"dynamicData": {"secret": {"name": secret_name}}} + + +def parse_arg_secret_entries(entries: list[str] | None) -> dict[str, str]: + """Parse ``INPUT=SECRET_NAME`` secret references into a mapping. + + Trimmed input and secret names must be non-empty, and an input may not be + repeated. Raising here keeps validation before any file read or network + call in the submit path. + """ + + parsed: dict[str, str] = {} + for entry in entries or []: + if "=" not in entry: + raise PipelineRunError(f"Expected INPUT=SECRET_NAME for --arg-secret, got {entry!r}") + raw_input, raw_secret = entry.split("=", 1) + input_name = raw_input.strip() + secret_name = raw_secret.strip() + if not input_name or not secret_name: + raise PipelineRunError( + f"--arg-secret requires a non-empty input and secret name, got {entry!r}" + ) + if input_name in parsed: + raise PipelineRunError(f"Duplicate --arg-secret for input {input_name!r}") + parsed[input_name] = secret_name + return parsed + + +def normalize_arg_secret_config(value: Any) -> dict[str, str]: + """Normalize a config ``arg_secrets`` mapping of input -> secret name.""" + + if value is None: + return {} + if not isinstance(value, Mapping): + raise PipelineRunError("arg_secrets config must be a mapping of INPUT to SECRET_NAME") + normalized: dict[str, str] = {} + for raw_input, raw_secret in value.items(): + input_name = str(raw_input).strip() + if not isinstance(raw_secret, str): + raise PipelineRunError( + f"arg_secrets[{raw_input!r}] must be a secret name string, " + f"got {type(raw_secret).__name__}" + ) + secret_name = raw_secret.strip() + if not input_name or not secret_name: + raise PipelineRunError("arg_secrets entries require a non-empty input and secret name") + normalized[input_name] = secret_name + return normalized + + +def merge_secret_run_args( + run_args: dict[str, Any], + secret_names: Mapping[str, str], +) -> dict[str, Any]: + """Merge secret references into run args, rejecting input conflicts. + + An input may be supplied as a plain value (``--arg`` / ``--args-json`` / + config ``args``) or as a secret reference (``--arg-secret`` / config + ``arg_secrets``), never both. Overlap is rejected rather than silently + overwriting one form with the other. + """ + + conflicts = sorted(set(run_args) & set(secret_names)) + if conflicts: + raise PipelineRunError( + "Input(s) given as both a value and a secret reference: " + f"{', '.join(conflicts)}. Use only one of --arg/--args-json or --arg-secret per input." + ) + merged = dict(run_args) + for input_name, secret_name in secret_names.items(): + merged[input_name] = secret_argument_value(secret_name) + return merged diff --git a/packages/tangle-cli/src/tangle_cli/pipeline_runs_cli.py b/packages/tangle-cli/src/tangle_cli/pipeline_runs_cli.py index 2da0eb5..2f072ae 100644 --- a/packages/tangle-cli/src/tangle_cli/pipeline_runs_cli.py +++ b/packages/tangle-cli/src/tangle_cli/pipeline_runs_cli.py @@ -32,6 +32,9 @@ PipelineRunError, PipelineRunHooks, PipelineRunManager, + merge_secret_run_args, + normalize_arg_secret_config, + parse_arg_secret_entries, parse_json_or_key_values, parse_key_value_entries, ) @@ -134,6 +137,17 @@ def pipeline_runs_submit( Parameter(help="Pipeline argument as KEY=VALUE. Repeat for multiple.", negative_iterable=()), ] = None, args_json: Annotated[str | None, Parameter(help="Pipeline arguments as a JSON object.")] = None, + arg_secret: Annotated[ + list[str] | None, + Parameter( + name="--arg-secret", + help=( + "Pipeline argument bound to a Tangle secret as INPUT=SECRET_NAME. " + "Repeat for multiple." + ), + negative_iterable=(), + ), + ] = None, annotation: Annotated[ list[str] | None, Parameter(help="Run annotation as KEY=VALUE. Repeat for multiple.", negative_iterable=()), @@ -185,6 +199,8 @@ def pipeline_runs_submit( "arg": (arg, None), "args_json": (args_json, None), "args_config": ("args", None, None, True), + "arg_secret": (arg_secret, None), + "arg_secrets_config": ("arg_secrets", None, None, True), "annotation": (annotation, None), "hydrate": (hydrate, True), "dry_run": (dry_run, None), @@ -197,8 +213,11 @@ def pipeline_runs_submit( } def action(manager: PipelineRunManager, args: ArgsContainer) -> dict[str, Any]: + run_args = parse_json_or_key_values(args.args_json or args.args_config, args.arg) + secret_names = normalize_arg_secret_config(args.arg_secrets_config) + secret_names.update(parse_arg_secret_entries(args.arg_secret)) kwargs = { - "run_args": parse_json_or_key_values(args.args_json or args.args_config, args.arg), + "run_args": merge_secret_run_args(run_args, secret_names), "annotations": parse_key_value_entries(args.annotation), "hydrate": bool(args.hydrate), "run_as": args.run_as, diff --git a/tests/test_pipeline_runs_cli.py b/tests/test_pipeline_runs_cli.py index b74b0ad..5a56a67 100644 --- a/tests/test_pipeline_runs_cli.py +++ b/tests/test_pipeline_runs_cli.py @@ -19,6 +19,10 @@ PipelineRunManager, PipelineWaitOutcome, PipelineWaitPoll, + merge_secret_run_args, + normalize_arg_secret_config, + parse_arg_secret_entries, + secret_argument_value, ) from tangle_cli.pipeline_runner import PipelineRunner, PipelineRunnerHooks @@ -2790,3 +2794,335 @@ def cleanup_prepared_pipeline(self, preparation, *, error=None): # type: ignore assert cleaned == [(temp_effective_path, "Pipeline validation failed:\n - boom")] assert not temp_effective_path.exists() + + +def _write_secret_pipeline(path: Path) -> Path: + path.write_text( + yaml.safe_dump( + { + "name": "Secret Pipeline", + "inputs": [ + {"name": "query", "type": "String", "default": "default"}, + {"name": "api_key", "type": "String"}, + {"name": "db_password", "type": "String", "optional": True}, + {"name": "required", "type": "String"}, + ], + "implementation": {"graph": {"tasks": {}}}, + }, + sort_keys=False, + ), + encoding="utf-8", + ) + return path + + +def test_secret_argument_value_matches_oss_dynamic_data_contract() -> None: + assert secret_argument_value("OPENAI_KEY") == { + "dynamicData": {"secret": {"name": "OPENAI_KEY"}} + } + + +def test_parse_arg_secret_entries_trims_and_maps_inputs() -> None: + assert parse_arg_secret_entries([" api_key = OPENAI_KEY ", "db_password=PG"]) == { + "api_key": "OPENAI_KEY", + "db_password": "PG", + } + + +def test_parse_arg_secret_entries_defaults_to_empty() -> None: + assert parse_arg_secret_entries(None) == {} + + +@pytest.mark.parametrize("entry", ["api_key", "", " ", "=SECRET", "api_key=", "api_key= "]) +def test_parse_arg_secret_entries_rejects_malformed(entry: str) -> None: + with pytest.raises(PipelineRunError): + parse_arg_secret_entries([entry]) + + +def test_parse_arg_secret_entries_rejects_duplicate_input() -> None: + with pytest.raises(PipelineRunError, match="Duplicate --arg-secret for input 'api_key'"): + parse_arg_secret_entries(["api_key=A", "api_key=B"]) + + +def test_normalize_arg_secret_config_accepts_mapping() -> None: + assert normalize_arg_secret_config({"api_key": "OPENAI_KEY", " db ": " PG "}) == { + "api_key": "OPENAI_KEY", + "db": "PG", + } + + +def test_normalize_arg_secret_config_defaults_to_empty() -> None: + assert normalize_arg_secret_config(None) == {} + + +def test_normalize_arg_secret_config_rejects_non_mapping() -> None: + with pytest.raises(PipelineRunError, match="arg_secrets config must be a mapping"): + normalize_arg_secret_config(["api_key=OPENAI_KEY"]) + + +def test_normalize_arg_secret_config_rejects_non_string_secret() -> None: + with pytest.raises(PipelineRunError, match="must be a secret name string"): + normalize_arg_secret_config({"api_key": 123}) + + +def test_merge_secret_run_args_injects_dynamic_data() -> None: + merged = merge_secret_run_args({"required": "value"}, {"api_key": "OPENAI_KEY"}) + assert merged == { + "required": "value", + "api_key": {"dynamicData": {"secret": {"name": "OPENAI_KEY"}}}, + } + + +def test_merge_secret_run_args_rejects_value_and_secret_conflict() -> None: + with pytest.raises(PipelineRunError, match="both a value and a secret reference: api_key"): + merge_secret_run_args({"api_key": "plain"}, {"api_key": "OPENAI_KEY"}) + + +def test_pipeline_runs_submit_arg_secret_encodes_dynamic_data(monkeypatch, tmp_path: Path, capsys): + pipeline_path = _write_secret_pipeline(tmp_path / "pipeline.yaml") + fake_client = FakeClient() + monkeypatch.setattr(pipeline_runs_cli, "LazyTangleApiClient", lambda **kwargs: fake_client) + app = cli.build_app() + + run_app( + app, + [ + "sdk", + "pipeline-runs", + "submit", + str(pipeline_path), + "--no-hydrate", + "--arg", + "required=value", + "--arg-secret", + "api_key=OPENAI_KEY", + ], + ) + + capsys.readouterr() + arguments = fake_client.created[0]["root_task"]["arguments"] + assert arguments["required"] == "value" + assert arguments["api_key"] == {"dynamicData": {"secret": {"name": "OPENAI_KEY"}}} + + +def test_pipeline_runs_submit_multiple_arg_secrets(monkeypatch, tmp_path: Path, capsys): + pipeline_path = _write_secret_pipeline(tmp_path / "pipeline.yaml") + fake_client = FakeClient() + monkeypatch.setattr(pipeline_runs_cli, "LazyTangleApiClient", lambda **kwargs: fake_client) + app = cli.build_app() + + run_app( + app, + [ + "sdk", + "pipeline-runs", + "submit", + str(pipeline_path), + "--no-hydrate", + "--arg", + "required=value", + "--arg-secret", + "api_key=OPENAI_KEY", + "--arg-secret", + "db_password=PG_PASSWORD", + ], + ) + + capsys.readouterr() + arguments = fake_client.created[0]["root_task"]["arguments"] + assert arguments["api_key"] == {"dynamicData": {"secret": {"name": "OPENAI_KEY"}}} + assert arguments["db_password"] == {"dynamicData": {"secret": {"name": "PG_PASSWORD"}}} + + +def test_pipeline_runs_submit_arg_secret_dry_run_no_network(monkeypatch, tmp_path: Path, capsys): + pipeline_path = _write_secret_pipeline(tmp_path / "pipeline.yaml") + fake_client = FakeClient() + monkeypatch.setattr(pipeline_runs_cli, "LazyTangleApiClient", lambda **kwargs: fake_client) + app = cli.build_app() + + run_app( + app, + [ + "sdk", + "pipeline-runs", + "submit", + str(pipeline_path), + "--no-hydrate", + "--dry-run", + "--arg", + "required=value", + "--arg-secret", + "api_key=OPENAI_KEY", + ], + ) + + payload = json.loads(capsys.readouterr().out) + assert fake_client.created == [] + arguments = payload["root_task"]["arguments"] + assert arguments["api_key"] == {"dynamicData": {"secret": {"name": "OPENAI_KEY"}}} + + +def test_pipeline_runs_submit_arg_secret_conflict_rejected_without_submit( + monkeypatch, tmp_path: Path +): + pipeline_path = _write_secret_pipeline(tmp_path / "pipeline.yaml") + fake_client = FakeClient() + monkeypatch.setattr(pipeline_runs_cli, "LazyTangleApiClient", lambda **kwargs: fake_client) + app = cli.build_app() + + with pytest.raises(SystemExit) as excinfo: + app( + [ + "sdk", + "pipeline-runs", + "submit", + str(pipeline_path), + "--no-hydrate", + "--arg", + "api_key=plain", + "--arg-secret", + "api_key=OPENAI_KEY", + ] + ) + + assert "both a value and a secret reference" in str(excinfo.value) + assert fake_client.created == [] + + +def test_pipeline_runs_submit_arg_secret_conflicts_with_args_json(monkeypatch, tmp_path: Path): + pipeline_path = _write_secret_pipeline(tmp_path / "pipeline.yaml") + fake_client = FakeClient() + monkeypatch.setattr(pipeline_runs_cli, "LazyTangleApiClient", lambda **kwargs: fake_client) + app = cli.build_app() + + with pytest.raises(SystemExit) as excinfo: + app( + [ + "sdk", + "pipeline-runs", + "submit", + str(pipeline_path), + "--no-hydrate", + "--args-json", + json.dumps({"api_key": "plain"}), + "--arg-secret", + "api_key=OPENAI_KEY", + ] + ) + + assert "both a value and a secret reference" in str(excinfo.value) + assert fake_client.created == [] + + +def test_pipeline_runs_submit_arg_secret_malformed_skips_file_and_network( + monkeypatch, tmp_path: Path +): + fake_client = FakeClient() + monkeypatch.setattr(pipeline_runs_cli, "LazyTangleApiClient", lambda **kwargs: fake_client) + app = cli.build_app() + missing_path = tmp_path / "does-not-exist.yaml" + + with pytest.raises(SystemExit) as excinfo: + app( + [ + "sdk", + "pipeline-runs", + "submit", + str(missing_path), + "--no-hydrate", + "--arg-secret", + "api_key=", + ] + ) + + message = str(excinfo.value) + assert "non-empty input and secret name" in message + assert not missing_path.exists() + assert fake_client.created == [] + + +def test_pipeline_runs_submit_arg_secret_duplicate_rejected(monkeypatch, tmp_path: Path): + pipeline_path = _write_secret_pipeline(tmp_path / "pipeline.yaml") + fake_client = FakeClient() + monkeypatch.setattr(pipeline_runs_cli, "LazyTangleApiClient", lambda **kwargs: fake_client) + app = cli.build_app() + + with pytest.raises(SystemExit) as excinfo: + app( + [ + "sdk", + "pipeline-runs", + "submit", + str(pipeline_path), + "--no-hydrate", + "--arg-secret", + "api_key=A", + "--arg-secret", + "api_key=B", + ] + ) + + assert "Duplicate --arg-secret for input 'api_key'" in str(excinfo.value) + assert fake_client.created == [] + + +def test_pipeline_runs_submit_arg_secret_from_config(monkeypatch, tmp_path: Path): + pipeline_path = _write_secret_pipeline(tmp_path / "pipeline.yaml") + config = tmp_path / "pipeline.config.yaml" + config.write_text( + yaml.safe_dump( + { + "pipeline_path": str(pipeline_path), + "hydrate": False, + "args": {"required": "value"}, + "arg_secrets": {"api_key": "OPENAI_KEY"}, + }, + sort_keys=False, + ), + encoding="utf-8", + ) + fake_client = FakeClient() + monkeypatch.setattr(pipeline_runs_cli, "LazyTangleApiClient", lambda **kwargs: fake_client) + app = cli.build_app() + + run_app(app, ["sdk", "pipeline-runs", "submit", "--config", str(config)]) + + arguments = fake_client.created[0]["root_task"]["arguments"] + assert arguments["required"] == "value" + assert arguments["api_key"] == {"dynamicData": {"secret": {"name": "OPENAI_KEY"}}} + + +def test_pipeline_runs_submit_cli_arg_secret_overrides_config(monkeypatch, tmp_path: Path): + pipeline_path = _write_secret_pipeline(tmp_path / "pipeline.yaml") + config = tmp_path / "pipeline.config.yaml" + config.write_text( + yaml.safe_dump( + { + "pipeline_path": str(pipeline_path), + "hydrate": False, + "args": {"required": "value"}, + "arg_secrets": {"api_key": "FROM_CONFIG"}, + }, + sort_keys=False, + ), + encoding="utf-8", + ) + fake_client = FakeClient() + monkeypatch.setattr(pipeline_runs_cli, "LazyTangleApiClient", lambda **kwargs: fake_client) + app = cli.build_app() + + run_app( + app, + [ + "sdk", + "pipeline-runs", + "submit", + "--config", + str(config), + "--arg-secret", + "api_key=FROM_CLI", + ], + ) + + arguments = fake_client.created[0]["root_task"]["arguments"] + assert arguments["api_key"] == {"dynamicData": {"secret": {"name": "FROM_CLI"}}}