From 5c8b065d0b851056dfde76af694475d6418d73c0 Mon Sep 17 00:00:00 2001 From: Alessandro Date: Wed, 8 Jul 2026 16:13:37 +0200 Subject: [PATCH 1/5] Hide Deeploy mandatory secrets in dAuth --- .../business/deeploy/deeploy_manager_api.py | 24 +++- extensions/business/deeploy/deeploy_mixin.py | 127 ++++++++++++++++++ .../deeploy/tests/test_create_requests.py | 63 +++++++++ .../deeploy/tests/test_process_request.py | 38 +++++- .../deeploy/tests/test_update_requests.py | 11 ++ 5 files changed, 257 insertions(+), 6 deletions(-) diff --git a/extensions/business/deeploy/deeploy_manager_api.py b/extensions/business/deeploy/deeploy_manager_api.py index 5dc1eef2..f6675af3 100644 --- a/extensions/business/deeploy/deeploy_manager_api.py +++ b/extensions/business/deeploy/deeploy_manager_api.py @@ -781,6 +781,7 @@ def _process_pipeline_request( skip_create_response_key_reset = False previous_pipeline_cid = None update_context_from_persisted_pipeline = False + dauth_secrets_stored = False if is_create: is_valid = self.deeploy_check_payment_and_job_owner(inputs, auth_result[DEEPLOY_KEYS.ESCROW_OWNER], is_create=is_create, debug=self.cfg_deeploy_verbose > 1) if not is_valid: @@ -959,7 +960,12 @@ def _process_pipeline_request( ) skip_create_response_key_reset = True - # All validations and response-key resets passed; remove the running job and redeploy. + job_secrets = self._extract_dauth_job_secrets_from_prepared_deploy_plan( + prepared_create_deploy_plan + ) + dauth_secrets_stored = self._store_deeploy_dauth_job_secrets(job_id, job_secrets) + + # All validations, response-key resets, and dAuth writes passed; remove the running job and redeploy. if update_context_from_persisted_pipeline: # TODO: stop stale offline old-node pipelines through ChainDist reconciliation when they return. self.Pd( @@ -1002,6 +1008,20 @@ def _process_pipeline_request( pipeline_params=pipeline_params, ) + if prepared_create_deploy_plan is None: + prepared_create_deploy_plan = self._prepare_create_pipeline_deploy_plan( + nodes=deployment_nodes, + inputs=inputs, + app_id=app_id, + job_app_type=job_app_type, + dct_deeploy_specs=deeploy_specs_payload, + ) + if not dauth_secrets_stored: + job_secrets = self._extract_dauth_job_secrets_from_prepared_deploy_plan( + prepared_create_deploy_plan + ) + self._store_deeploy_dauth_job_secrets(job_id, job_secrets) + dct_status, str_status, response_keys, pipeline_to_persist = self.check_and_deploy_pipelines( owner=auth_result[DEEPLOY_KEYS.ESCROW_OWNER], inputs=inputs, @@ -1027,7 +1047,7 @@ def _process_pipeline_request( return_request = request.get(DEEPLOY_KEYS.RETURN_REQUEST, False) if return_request: - dct_request = self.deepcopy(request) + dct_request = self._redact_deeploy_dauth_secrets_for_response(request) dct_request.pop(DEEPLOY_KEYS.APP_PARAMS, None) else: # Build simplified request summary (no app_params - data is in plugins array now) diff --git a/extensions/business/deeploy/deeploy_mixin.py b/extensions/business/deeploy/deeploy_mixin.py index b66728a0..8d86337a 100644 --- a/extensions/business/deeploy/deeploy_mixin.py +++ b/extensions/business/deeploy/deeploy_mixin.py @@ -57,6 +57,25 @@ PREFERRED_NODES_MAX_PAYLOAD_BYTES = 32 * 1024 PREFERRED_NODE_ALIAS_MAX_LENGTH = 128 PREFERRED_NODE_DESCRIPTION_MAX_LENGTH = 512 +DEEPLOY_DAUTH_SECRET_PLACEHOLDER = "__R1_DAUTH_SECRET__" +DEEPLOY_DAUTH_JOB_SECRETS_HKEY = "DAUTH_JOB_SECRETS" +DEEPLOY_DAUTH_SECRET_PATH_SUFFIXES = ( + ("CLOUDFLARE_TOKEN",), + ("NGROK_AUTH_TOKEN",), + ("EXPOSED_PORTS", "*", "token"), + ("EXPOSED_PORTS", "*", "tunnel", "token"), + ("VCS_DATA", "TOKEN"), + ("CR_DATA", "PASSWORD"), + ("ENV", "R1EN_CSTORE_AUTH_SECRET"), + ("ENV", "R1EN_CSTORE_AUTH_BOOTSTRAP_ADMIN_PWD"), + ("ENV", "CF_TUNNEL_TOKEN"), + ("ENV", "CRDB_PASSWORD"), + ("ENV", "CRDB_CA_CRT"), + ("ENV", "CRDB_NODE_CRT"), + ("ENV", "CRDB_NODE_KEY"), + ("ENV", "CRDB_CLIENT_ROOT_CRT"), + ("ENV", "CRDB_CLIENT_ROOT_KEY"), +) SENSITIVE_LOG_KEY_PARTS = ( "BEGINPRIVATEKEY", "BEGINRSAPRIVATEKEY", @@ -3716,6 +3735,114 @@ def redact(value): redact(redacted) return redacted + def _matches_deeploy_dauth_secret_path(self, path): + path = [str(part) for part in path] + for suffix in DEEPLOY_DAUTH_SECRET_PATH_SUFFIXES: + if len(path) < len(suffix): + continue + tail = path[-len(suffix):] + if all(expected == "*" or expected == actual for expected, actual in zip(suffix, tail)): + return True + return False + + def _has_deeploy_dauth_secret_value(self, value): + if isinstance(value, (dict, list)): + return False + if value is None or value == "": + return False + return value != DEEPLOY_DAUTH_SECRET_PLACEHOLDER + + def _merge_deeploy_dauth_secret_fragments(self, target, source): + if source is None: + return target + if target is None: + return self.deepcopy(source) + if isinstance(target, dict) and isinstance(source, dict): + for key, value in source.items(): + target[key] = self._merge_deeploy_dauth_secret_fragments(target.get(key), value) + return target + if isinstance(target, list) and isinstance(source, list): + while len(target) < len(source): + target.append(None) + for idx, value in enumerate(source): + target[idx] = self._merge_deeploy_dauth_secret_fragments(target[idx], value) + return target + return self.deepcopy(source) + + def _extract_and_redact_deeploy_dauth_secrets(self, payload): + redacted = self.deepcopy(payload) + + def walk(value, path): + if isinstance(value, dict): + secrets = {} + for key, item in list(value.items()): + item_path = path + [key] + if ( + self._matches_deeploy_dauth_secret_path(item_path) + and self._has_deeploy_dauth_secret_value(item) + ): + secrets[key] = self.deepcopy(item) + value[key] = DEEPLOY_DAUTH_SECRET_PLACEHOLDER + continue + child_secrets = walk(item, item_path) + if child_secrets is not None: + secrets[key] = child_secrets + return secrets or None + if isinstance(value, list): + secrets = [None] * len(value) + found = False + for idx, item in enumerate(value): + child_secrets = walk(item, path + [idx]) + if child_secrets is not None: + secrets[idx] = child_secrets + found = True + return secrets if found else None + return None + + return redacted, walk(redacted, []) + + def _redact_deeploy_dauth_secrets_for_response(self, payload): + redacted, _ = self._extract_and_redact_deeploy_dauth_secrets(payload) + return redacted + + def _extract_dauth_job_secrets_from_prepared_deploy_plan(self, prepared_deploy_plan): + if not isinstance(prepared_deploy_plan, dict): + return None + node_plugins_by_addr = prepared_deploy_plan.get("node_plugins_by_addr") + if not isinstance(node_plugins_by_addr, dict): + return None + + merged_plugins_secrets = None + for node, plugins in list(node_plugins_by_addr.items()): + redacted_plugins, plugins_secrets = self._extract_and_redact_deeploy_dauth_secrets(plugins) + node_plugins_by_addr[node] = redacted_plugins + merged_plugins_secrets = self._merge_deeploy_dauth_secret_fragments( + merged_plugins_secrets, + plugins_secrets, + ) + if merged_plugins_secrets is None: + return None + return {"PLUGINS": merged_plugins_secrets} + + def _store_deeploy_dauth_job_secrets(self, job_id, job_secrets): + if not job_secrets: + return False + if job_id in [None, ""]: + raise ValueError("Cannot store dAuth secrets without job_id.") + job_id = str(job_id) + bundle = { + "job_id": job_id, + "job_secrets": self.deepcopy(job_secrets), + } + ok = self.chainstore_hset( + hkey=DEEPLOY_DAUTH_JOB_SECRETS_HKEY, + key=job_id, + value=bundle, + ) + if not ok: + raise ValueError(f"Failed to store dAuth secrets for job {job_id}.") + return True + def _iter_per_node_configs(self, plugins): for plugin in plugins or []: instances = plugin.get(self.ct.CONFIG_PLUGIN.K_INSTANCES) or [] diff --git a/extensions/business/deeploy/tests/test_create_requests.py b/extensions/business/deeploy/tests/test_create_requests.py index 83efc477..df808c09 100644 --- a/extensions/business/deeploy/tests/test_create_requests.py +++ b/extensions/business/deeploy/tests/test_create_requests.py @@ -16,6 +16,7 @@ DEEPLOY_PLUGIN_DATA, JOB_APP_TYPES, ) +from extensions.business.deeploy.deeploy_mixin import DEEPLOY_DAUTH_SECRET_PLACEHOLDER from extensions.business.deeploy.tests.support import make_deeploy_plugin, make_inputs, make_plugin_entry @@ -257,6 +258,68 @@ def test_log_redaction_masks_per_node_config_and_token_keys(self): self.assertIn("'R1EN_CSTORE_AUTH_BOOTSTRAP_ADMIN_PWD': '***'", serialized) self.assertIn("'PER_NODE_CONFIG': '***'", serialized) + def test_dauth_secret_extraction_redacts_only_mandatory_paths(self): + plugin = make_deeploy_plugin() + payload = { + "PLUGINS": [{ + "INSTANCES": [{ + "CLOUDFLARE_TOKEN": "cf-token", + "NGROK_AUTH_TOKEN": "ngrok-token", + "EXPOSED_PORTS": { + "26257": { + "token": "port-token", + "tunnel": {"token": "tunnel-token"}, + }, + }, + "VCS_DATA": {"TOKEN": "github-token", "BRANCH": "main"}, + "CR_DATA": {"USERNAME": "user", "PASSWORD": "registry-password"}, + "ENV": { + "R1EN_CSTORE_AUTH_SECRET": "cstore-secret", + "R1EN_CSTORE_AUTH_BOOTSTRAP_ADMIN_PWD": "admin-password", + "CF_TUNNEL_TOKEN": "node-tunnel-token", + "CRDB_PASSWORD": "crdb-password", + "CRDB_CA_CRT": "ca-crt", + "CRDB_NODE_CRT": "node-crt", + "CRDB_NODE_KEY": "node-key", + "CRDB_CLIENT_ROOT_CRT": "root-crt", + "CRDB_CLIENT_ROOT_KEY": "root-key", + "POSTGRES_PASSWORD": "user-env-password", + }, + "CHAINSTORE_RESPONSE_KEY": "response-key", + }], + }], + } + + redacted, secrets = plugin._extract_and_redact_deeploy_dauth_secrets(payload) + serialized_redacted = str(redacted) + serialized_secrets = str(secrets) + + for value in ( + "cf-token", + "ngrok-token", + "port-token", + "tunnel-token", + "github-token", + "registry-password", + "cstore-secret", + "admin-password", + "node-tunnel-token", + "crdb-password", + "ca-crt", + "node-crt", + "node-key", + "root-crt", + "root-key", + ): + self.assertNotIn(value, serialized_redacted) + self.assertIn(value, serialized_secrets) + + self.assertIn(DEEPLOY_DAUTH_SECRET_PLACEHOLDER, serialized_redacted) + self.assertIn("user-env-password", serialized_redacted) + self.assertIn("response-key", serialized_redacted) + self.assertNotIn("user-env-password", serialized_secrets) + self.assertNotIn("response-key", serialized_secrets) + def test_cockroachdb_secure_config_generates_node_certs_and_redacts_them(self): plugin = make_deeploy_plugin() inputs = make_inputs( diff --git a/extensions/business/deeploy/tests/test_process_request.py b/extensions/business/deeploy/tests/test_process_request.py index 2bfdd9e2..5b1246da 100644 --- a/extensions/business/deeploy/tests/test_process_request.py +++ b/extensions/business/deeploy/tests/test_process_request.py @@ -2,10 +2,15 @@ import sys import types import unittest +from collections import defaultdict from naeural_core import constants as ct from extensions.business.deeploy.deeploy_const import DEEPLOY_KEYS, DEEPLOY_STATUS +from extensions.business.deeploy.deeploy_mixin import ( + DEEPLOY_DAUTH_JOB_SECRETS_HKEY, + DEEPLOY_DAUTH_SECRET_PLACEHOLDER, +) class _BasePluginStub: @@ -95,6 +100,16 @@ def _queue_pipeline_persistence(self, persistence_state): self.queued_persistence = persistence_state return True + def chainstore_hset(self, hkey, key, value): + if not hasattr(self, "chainstore_writes"): + self.chainstore_writes = [] + self.chainstore_writes.append({ + "hkey": hkey, + "key": key, + "value": copy.deepcopy(value), + }) + return True + class DeeployProcessRequestTests(unittest.TestCase): @@ -103,15 +118,16 @@ def test_create_pipeline_accepts_ui_cockroach_single_plugin_top_level_per_node_c plugin.ct = ct plugin.bc = _BCStub() plugin.deepcopy = copy.deepcopy + plugin.defaultdict = defaultdict plugin.sanitize_name = lambda value: str(value).replace("/", "_").replace(" ", "_") plugin.uuid = lambda size=7: "abc1234"[:size] plugin.cfg_deeploy_verbose = 0 plugin.queued_persistence = None + plugin.chainstore_writes = [] captured = {} def check_and_deploy_pipelines(**kwargs): captured.update(kwargs) - captured["prepared_plugins"] = plugin.deeploy_prepare_plugins(kwargs["inputs"]) return {}, DEEPLOY_STATUS.COMMAND_DELIVERED, {}, { "CONFIG_STREAMS": [{"NAME": kwargs["app_id"]}], } @@ -152,7 +168,10 @@ def check_and_deploy_pipelines(**kwargs): deployed_inputs = captured["inputs"] self.assertNotIn("PER_NODE_CONFIG", deployed_inputs) deployed_plugin = deployed_inputs[DEEPLOY_KEYS.PLUGINS][0] - prepared_plugin = captured["prepared_plugins"][0][plugin.ct.CONFIG_PLUGIN.K_INSTANCES][0] + prepared_plugin = ( + captured["prepared_create_deploy_plan"]["node_plugins_by_addr"]["0xai_node_a"][0] + [plugin.ct.CONFIG_PLUGIN.K_INSTANCES][0] + ) self.assertEqual( deployed_plugin["PER_NODE_CONFIG"]["byNode"]["0xai_node_b"]["ENV"]["CRDB_NODE_ID"], "2", @@ -162,9 +181,20 @@ def check_and_deploy_pipelines(**kwargs): "2", ) self.assertEqual(plugin.bc.submitted, [(97, ["eth_0xai_node_a", "eth_0xai_node_b"])]) - self.assertIn("token-a", str(res[DEEPLOY_KEYS.REQUEST])) - self.assertIn("token-b", str(res[DEEPLOY_KEYS.REQUEST])) + self.assertNotIn("token-a", str(res[DEEPLOY_KEYS.REQUEST])) + self.assertNotIn("token-b", str(res[DEEPLOY_KEYS.REQUEST])) + self.assertIn(DEEPLOY_DAUTH_SECRET_PLACEHOLDER, str(res[DEEPLOY_KEYS.REQUEST])) self.assertIsInstance(res[DEEPLOY_KEYS.REQUEST]["PER_NODE_CONFIG"], dict) + self.assertEqual(plugin.chainstore_writes[0]["hkey"], DEEPLOY_DAUTH_JOB_SECRETS_HKEY) + self.assertEqual(plugin.chainstore_writes[0]["key"], "97") + stored = str(plugin.chainstore_writes[0]["value"]) + self.assertIn("token-a", stored) + self.assertIn("token-b", stored) + self.assertNotIn(DEEPLOY_DAUTH_SECRET_PLACEHOLDER, stored) + self.assertEqual( + prepared_plugin["PER_NODE_CONFIG"]["byNode"]["0xai_node_a"]["ENV"]["CF_TUNNEL_TOKEN"], + DEEPLOY_DAUTH_SECRET_PLACEHOLDER, + ) def test_error_handler_redacts_secret_request_values(self): plugin = _ProcessRequestStub.__new__(_ProcessRequestStub) diff --git a/extensions/business/deeploy/tests/test_update_requests.py b/extensions/business/deeploy/tests/test_update_requests.py index 14d0ac29..d08a3637 100644 --- a/extensions/business/deeploy/tests/test_update_requests.py +++ b/extensions/business/deeploy/tests/test_update_requests.py @@ -79,6 +79,17 @@ def _make_process_update_plugin(self, discovered_instances, nodes=None, deeploy_ } plugin._get_pipeline_from_cstore = lambda job_id: None plugin._check_nodes_availability = lambda inputs: nodes or ["node-1"] + plugin.chainstore_writes = [] + + def chainstore_hset(hkey, key, value): + plugin.chainstore_writes.append({ + "hkey": hkey, + "key": key, + "value": copy.deepcopy(value), + }) + return True + + plugin.chainstore_hset = chainstore_hset called = {"delete": 0, "deploy": 0, "deploy_kwargs": None, "queued": 0, "bc_update": 0} plugin.bc = types.SimpleNamespace( From 43024deec31e3b7511bce3e6e698bfbe340d8590 Mon Sep 17 00:00:00 2001 From: Alessandro Date: Tue, 28 Jul 2026 17:24:22 +0200 Subject: [PATCH 2/5] test: update Cockroach dAuth fixture --- extensions/business/deeploy/tests/test_process_request.py | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/extensions/business/deeploy/tests/test_process_request.py b/extensions/business/deeploy/tests/test_process_request.py index 5b1246da..1670a57e 100644 --- a/extensions/business/deeploy/tests/test_process_request.py +++ b/extensions/business/deeploy/tests/test_process_request.py @@ -152,7 +152,12 @@ def check_and_deploy_pipelines(**kwargs): DEEPLOY_KEYS.PLUGIN_SIGNATURE: "CONTAINER_APP_RUNNER", "IMAGE": "ghcr.io/ratio1/deeploy-cockroachdb-service:main", "CONTAINER_RESOURCES": {"cpu": 1, "memory": "2g", "storage": "8g"}, - "ENV": {"CRDB_MAX_OFFSET": "500ms"}, + "ENV": { + "CRDB_DATABASE": "appdb", + "CRDB_USER": "appuser", + "CRDB_PASSWORD": "secret-password", + "CRDB_MAX_OFFSET": "500ms", + }, }], "PER_NODE_CONFIG": { "byNode": { From 93b7fb14c5f88e71b7c19827d6641cea2349c530 Mon Sep 17 00:00:00 2001 From: Alessandro Date: Thu, 16 Jul 2026 17:44:45 +0200 Subject: [PATCH 3/5] feat: stage worker dAuth secrets before dispatch --- AGENTS.md | 9 + Dockerfile_devnet | 2 +- Dockerfile_mainnet | 2 +- Dockerfile_testnet | 2 +- .../container_apps/container_app_runner.py | 8 +- .../container_apps/container_utils.py | 6 +- .../tests/test_sensitive_platform_logs.py | 20 + extensions/business/dauth/dauth_mixin.py | 135 ++++++- extensions/business/dauth/dauth_registry.py | 13 +- .../dauth/test_dauth_registry_gating.py | 120 ++++++ .../dauth/test_dauth_secret_routing.py | 13 + .../deeploy/deeploy_cmdapi_integration.py | 20 + .../business/deeploy/deeploy_job_mixin.py | 161 +++++++- .../business/deeploy/deeploy_manager_api.py | 115 ++++-- extensions/business/deeploy/deeploy_mixin.py | 344 +++++++++++++++--- extensions/business/deeploy/test_deeploy.py | 12 +- extensions/business/deeploy/tests/support.py | 36 ++ .../tests/test_cmdapi_staging_integration.py | 104 ++++++ .../deeploy/tests/test_process_request.py | 34 ++ .../deeploy/tests/test_secret_staging.py | 221 +++++++++++ .../deeploy/tests/test_update_requests.py | 75 ++++ requirements.txt | 2 +- 22 files changed, 1351 insertions(+), 103 deletions(-) create mode 100644 extensions/business/container_apps/tests/test_sensitive_platform_logs.py create mode 100644 extensions/business/deeploy/deeploy_cmdapi_integration.py create mode 100644 extensions/business/deeploy/tests/test_cmdapi_staging_integration.py create mode 100644 extensions/business/deeploy/tests/test_secret_staging.py diff --git a/AGENTS.md b/AGENTS.md index 35335263..ca8b8450 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -696,6 +696,15 @@ Entry format: - Verification: `python3 -m unittest extensions.serving.test_th_hf_model_base extensions.serving.test_th_text_classifier extensions.serving.test_th_privacy_filter extensions.business.edge_inference_api.test_text_classifier_inference_api extensions.business.edge_inference_api.test_privacy_filter_inference_api`; `python3 -m py_compile extensions/serving/default_inference/nlp/th_hf_model_base.py extensions/business/edge_inference_api/text_classifier_inference_api.py`; required serving gate `python3 -m unittest extensions.serving.model_testing.test_llm_servings` currently fails at import with `ImportError: cannot import name 'Logger' from 'naeural_core'`. - Links: `extensions/serving/default_inference/nlp/th_hf_model_base.py`, `extensions/business/edge_inference_api/text_classifier_inference_api.py`, `extensions/serving/test_th_hf_model_base.py` +- ID: `ML-20260716-001` +- Timestamp: `2026-07-16T15:06:19Z` +- Type: `change` +- Summary: Deeploy now stages redacted R1FS authorization metadata and complete dAuth secret bundles before worker dispatch. +- Criticality: Cross-cutting security and deployment transaction change affecting Deeploy create, replacement update, scale-up, dAuth replication, rollback, and worker secret authorization. +- Details: Exact pipeline command metadata is built before dispatch, mandatory secrets are replaced with the shared `__R1_DAUTH_SECRET__` marker, unchanged placeholders are reconstructed from the prior same-path bundle, and unresolved paths fail closed. Pipeline metadata is replicated to normal peers plus registry-selected dAuth peers; secret bundles target only dAuth peers and are bound to the exact R1FS CID so interleaved writes cannot return mixed generations. Structurally complete legacy bundles are CID-bound lazily after a pointer recheck. R1FS is the scale-up source of truth and persisted offline target nodes survive scale-up. Successful deployments remove the superseded CID, while failures and timeouts restore the prior pointer and bundle only if both staged values are still current. Container platform logs no longer render full environment dictionaries, tunnel-token commands, configured start commands, dynamic secret values, or exec shell text. +- Verification: `python -m unittest discover -s extensions/business/deeploy/tests -p 'test_*.py'` (215 passed); focused dAuth/staging/log tests (18 passed); focused cross-surface tests (75 passed). Full container-app discovery has three documented local-environment failures unrelated to this change. +- Links: `extensions/business/deeploy/deeploy_manager_api.py`, `extensions/business/deeploy/deeploy_job_mixin.py`, `extensions/business/deeploy/deeploy_mixin.py`, `extensions/business/dauth/dauth_registry.py`, `extensions/business/container_apps/container_app_runner.py` + - ID: `ML-20260723-001` - Timestamp: `2026-07-23T13:45:20Z` - Type: `change` diff --git a/Dockerfile_devnet b/Dockerfile_devnet index 0c7d9775..ce3e809d 100644 --- a/Dockerfile_devnet +++ b/Dockerfile_devnet @@ -75,7 +75,7 @@ ENV EE_DEBUG_R1FS=true # ENV EE_DEVICE=cuda:0 RUN uv pip install --system --no-cache -r requirements.txt -RUN uv pip install --system --no-cache --no-deps naeural-core +RUN uv pip install --system --no-cache --no-deps 'naeural-core>=7.7.318' #RUN pip install --no-cache-dir -r requirements.txt #RUN pip install --no-cache-dir --no-deps naeural-core diff --git a/Dockerfile_mainnet b/Dockerfile_mainnet index 7c001b5e..8d076f59 100644 --- a/Dockerfile_mainnet +++ b/Dockerfile_mainnet @@ -76,7 +76,7 @@ ENV EE_DEBUG_R1FS=false # ENV EE_DEVICE cuda:0 RUN uv pip install --system --no-cache -r requirements.txt -RUN uv pip install --system --no-cache --no-deps naeural-core +RUN uv pip install --system --no-cache --no-deps 'naeural-core>=7.7.318' #RUN pip install --no-cache-dir -r requirements.txt #RUN pip install --no-cache-dir --no-deps naeural-core diff --git a/Dockerfile_testnet b/Dockerfile_testnet index 2f9e077d..6390d6fc 100644 --- a/Dockerfile_testnet +++ b/Dockerfile_testnet @@ -75,7 +75,7 @@ ENV EE_DEBUG_R1FS=true # ENV EE_DEVICE=cuda:0 RUN uv pip install --system --no-cache -r requirements.txt -RUN uv pip install --system --no-cache --no-deps naeural-core +RUN uv pip install --system --no-cache --no-deps 'naeural-core>=7.7.318' #RUN pip install --no-cache-dir -r requirements.txt #RUN pip install --no-cache-dir --no-deps naeural-core diff --git a/extensions/business/container_apps/container_app_runner.py b/extensions/business/container_apps/container_app_runner.py index e778edfc..ddd64d61 100644 --- a/extensions/business/container_apps/container_app_runner.py +++ b/extensions/business/container_apps/container_app_runner.py @@ -2036,7 +2036,6 @@ def _start_extra_tunnel(self, container_port, tunnel_config): try: host_port = self._get_host_port_for_container_port(container_port) self.P(f"Starting Cloudflare tunnel for container port {container_port} (host port {host_port})...") - self.Pd(f" Command: {' '.join(command)}") # Use list-based subprocess to prevent shell injection popen_kwargs = dict( @@ -2365,13 +2364,14 @@ def start_container(self): log_str += f"Container data:\n" log_str += f" Image: {self.cfg_image}\n" log_str += f" Ports: {self.json_dumps(self.inverted_ports_mapping) if self.inverted_ports_mapping else 'None'}\n" - log_str += f" Env: {self.json_dumps(self.env) if self.env else 'None'}\n" + env_names = sorted(self.env) if self.env else [] + log_str += f" Env vars: {len(env_names)} configured\n" log_str += f" Volumes: {self.json_dumps(self.volumes) if self.volumes else 'None'}\n" log_str += f" Resources: {self.json_dumps(self.cfg_container_resources) if self.cfg_container_resources else 'None'}\n" log_str += f" Restart policy: {self.cfg_restart_policy}\n" log_str += f" Pull policy: {self.cfg_image_pull_policy}\n" log_str += f" Entrypoint: {self._entrypoint if self._entrypoint else 'Image default'}\n" - log_str += f" Start command: {self._start_command if self._start_command else 'Image default'}\n" + log_str += f" Start command: {'configured' if self._start_command else 'Image default'}\n" log_str += f" User: {self.cfg_container_user if self.cfg_container_user else 'Image default'}\n" self.P(log_str) @@ -2649,7 +2649,7 @@ def _run_container_exec(self, shell_cmd): self._record_restart_failure() return - self.P(f"Running container exec command: {shell_cmd}") + self.P("Running configured container exec command") exec_kwargs = dict( stream=True, detach=False, diff --git a/extensions/business/container_apps/container_utils.py b/extensions/business/container_apps/container_utils.py index 666c8568..39133df1 100644 --- a/extensions/business/container_apps/container_utils.py +++ b/extensions/business/container_apps/container_utils.py @@ -792,7 +792,7 @@ def _configure_dynamic_env(self): variable_value += candidate_value # endfor each part self.dynamic_env[variable_name] = variable_value - self.P(f"Dynamic env var {variable_name} = {variable_value}") + self.P(f"Resolved dynamic env var {variable_name}") #endfor each variable ## END CONTAINER MIXIN ### @@ -1148,12 +1148,12 @@ def _setup_env_and_ports(self): "=" * 60, "SEMAPHORE ENV INJECTION", "=" * 60, - f" Adding {len(semaphore_env)} env vars from semaphored plugins:", + f" Adding {len(semaphore_env)} env vars from semaphored plugins.", ] for key, value in semaphore_env.items(): sanitized_key = self._sanitize_semaphore_env_var_name(key) sanitized_semaphore_env[sanitized_key] = value - log_lines.append(f" {sanitized_key} = {value}") + log_lines.append(f" {sanitized_key}") log_lines.append("=" * 60) self.Pd("\n".join(log_lines)) self.env.update(sanitized_semaphore_env) diff --git a/extensions/business/container_apps/tests/test_sensitive_platform_logs.py b/extensions/business/container_apps/tests/test_sensitive_platform_logs.py new file mode 100644 index 00000000..44edec05 --- /dev/null +++ b/extensions/business/container_apps/tests/test_sensitive_platform_logs.py @@ -0,0 +1,20 @@ +import unittest +from pathlib import Path + + +class SensitivePlatformLogTests(unittest.TestCase): + def test_platform_logs_do_not_render_secret_bearing_values_or_commands(self): + root = Path(__file__).resolve().parents[1] + runner_source = (root / "container_app_runner.py").read_text(encoding="utf-8") + utils_source = (root / "container_utils.py").read_text(encoding="utf-8") + + self.assertNotIn("self.json_dumps(self.env)", runner_source) + self.assertNotIn("' '.join(command)", runner_source) + self.assertNotIn("Running container exec command: {shell_cmd}", runner_source) + self.assertNotIn("Start command: {self._start_command", runner_source) + self.assertNotIn("{sanitized_key} = {value}", utils_source) + self.assertNotIn("{variable_name} = {variable_value}", utils_source) + + +if __name__ == "__main__": + unittest.main() diff --git a/extensions/business/dauth/dauth_mixin.py b/extensions/business/dauth/dauth_mixin.py index 7ece280c..9a465c29 100644 --- a/extensions/business/dauth/dauth_mixin.py +++ b/extensions/business/dauth/dauth_mixin.py @@ -14,6 +14,12 @@ """ +from .dauth_registry import ( + DAUTH_SECRET_PIPELINE_CID_KEY, + dauth_registry_write_kwargs, +) +from ratio1.const.base import dAuth + from extensions.business.dauth.dauth_registry import dauth_registry_write_kwargs @@ -218,13 +224,14 @@ def _normalize_dauth_job_id(self, job_id): raise ValueError("Job ID is required.") return str(job_id) - def _build_secret_bundle_from_request(self, body, job_id): + def _build_secret_bundle_from_request(self, body, job_id, pipeline_cid): job_secrets = body.get("job_secrets") if not isinstance(job_secrets, dict): raise ValueError("job_secrets must be a dictionary.") return { "job_id": job_id, "job_secrets": self.deepcopy(job_secrets), + DAUTH_SECRET_PIPELINE_CID_KEY: pipeline_cid, } def _save_dauth_job_secret_bundle(self, job_id, secret_bundle): @@ -244,14 +251,18 @@ def _load_dauth_job_secret_bundle(self, job_id): key=job_id, ) - def _load_dauth_job_pipeline(self, job_id): + def _load_dauth_job_pipeline_record(self, job_id): cid = self.chainstore_hget( hkey=DEEPLOY_JOBS_CSTORE_HKEY, key=job_id, ) if not cid: - return None - return self.r1fs.get_json(cid, show_logs=False) + return None, None + return cid, self.r1fs.get_json(cid, show_logs=False) + + def _load_dauth_job_pipeline(self, job_id): + _, pipeline = self._load_dauth_job_pipeline_record(job_id) + return pipeline def _pipeline_runner_nodes(self, pipeline): if not isinstance(pipeline, dict): @@ -282,6 +293,84 @@ def _is_node_running_dauth_job(self, job_id, node_address): ] return requester in runner_nodes + def _extract_current_legacy_secret_paths(self, pipeline, secret_bundle): + """Rebuild legacy secrets using only current exact placeholder paths.""" + if not isinstance(pipeline, dict) or not isinstance(secret_bundle, dict): + raise ValueError("Legacy dAuth pipeline or bundle is invalid.") + job_secrets = secret_bundle.get("job_secrets") + if not isinstance(job_secrets, dict): + raise ValueError("Legacy dAuth job_secrets is invalid.") + plugins_key = "PLUGINS" if "PLUGINS" in pipeline else "plugins" + redacted_plugins = pipeline.get(plugins_key) + secret_plugins = job_secrets.get( + plugins_key, + job_secrets.get("plugins" if plugins_key == "PLUGINS" else "PLUGINS"), + ) + + def contains_placeholder(value): + if value == dAuth.DAUTH_SECRET_PLACEHOLDER: + return True + if isinstance(value, dict): + return any(contains_placeholder(item) for item in value.values()) + if isinstance(value, list): + return any(contains_placeholder(item) for item in value) + return False + + def extract(redacted, secret): + if redacted == dAuth.DAUTH_SECRET_PLACEHOLDER: + if secret is None or isinstance(secret, (dict, list)): + raise ValueError("Legacy dAuth secret path is incomplete.") + return self.deepcopy(secret) + if not contains_placeholder(redacted): + return None + if isinstance(redacted, dict): + if not isinstance(secret, dict): + raise ValueError("Legacy dAuth secret structure is incomplete.") + return { + key: extract(value, secret.get(key)) + for key, value in redacted.items() + if contains_placeholder(value) + } + if isinstance(redacted, list): + if not isinstance(secret, list) or len(secret) < len(redacted): + raise ValueError("Legacy dAuth secret structure is incomplete.") + return [ + extract(value, secret[idx]) if contains_placeholder(value) else None + for idx, value in enumerate(redacted) + ] + raise ValueError("Legacy dAuth secret structure is incomplete.") + + if not contains_placeholder(redacted_plugins): + raise ValueError("Current pipeline has no dAuth secret placeholders.") + return {plugins_key: extract(redacted_plugins, secret_plugins)} + + def _bind_legacy_secret_bundle(self, job_id, pipeline_cid, pipeline, secret_bundle): + """Bind a structurally complete pre-CID bundle to the current pipeline.""" + if str(secret_bundle.get("job_id")) != str(job_id): + raise ValueError(f"Legacy dAuth secret bundle job mismatch for job {job_id}.") + try: + current_job_secrets = self._extract_current_legacy_secret_paths( + pipeline, + secret_bundle, + ) + except ValueError as exc: + raise ValueError( + f"Legacy dAuth secret bundle is incomplete for job {job_id}." + ) from exc + current_pipeline_cid = self.chainstore_hget( + hkey=DEEPLOY_JOBS_CSTORE_HKEY, + key=job_id, + ) + if current_pipeline_cid != pipeline_cid: + raise ValueError(f"dAuth pipeline generation changed for job {job_id}.") + bound_bundle = { + "job_id": str(job_id), + "job_secrets": current_job_secrets, + DAUTH_SECRET_PIPELINE_CID_KEY: pipeline_cid, + } + self._save_dauth_job_secret_bundle(job_id, bound_bundle) + return bound_bundle + def process_dauth_add_secrets_request(self, body): requester, requester_eth = self._verify_signed_dauth_body(body) request_nonce = self._validate_dauth_secret_request_nonce(body) @@ -289,7 +378,19 @@ def process_dauth_add_secrets_request(self, body): raise ValueError(f"Sender {requester_eth} is not an oracle.") job_id = self._normalize_dauth_job_id(body.get("job_id")) - secret_bundle = self._build_secret_bundle_from_request(body, job_id) + if not isinstance(body.get("job_secrets"), dict): + raise ValueError("job_secrets must be a dictionary.") + pipeline_cid = self.chainstore_hget( + hkey=DEEPLOY_JOBS_CSTORE_HKEY, + key=job_id, + ) + if not pipeline_cid: + raise ValueError(f"No persisted pipeline found for job {job_id}.") + secret_bundle = self._build_secret_bundle_from_request( + body, + job_id, + pipeline_cid, + ) self._save_dauth_job_secret_bundle(job_id, secret_bundle) self.Pd(f"dAuth stored secret bundle for job {job_id} from oracle {requester}.") return { @@ -303,12 +404,34 @@ def process_dauth_get_secret_request(self, body): request_nonce = self._validate_dauth_secret_request_nonce(body) job_id = self._normalize_dauth_job_id(body.get("job_id")) - if not self._is_node_running_dauth_job(job_id, requester): + pipeline_cid, pipeline = self._load_dauth_job_pipeline_record(job_id) + runner_nodes = self._pipeline_runner_nodes(pipeline) + normalized_requester = self._normalize_node_address_for_compare(requester) + normalized_runner_nodes = [ + self._normalize_node_address_for_compare(node) + for node in runner_nodes + ] + if normalized_requester not in normalized_runner_nodes: raise ValueError(f"Sender {requester} is not running job {job_id}.") secret_bundle = self._load_dauth_job_secret_bundle(job_id) if not isinstance(secret_bundle, dict): raise ValueError(f"No dAuth secret bundle found for job {job_id}.") + if DAUTH_SECRET_PIPELINE_CID_KEY not in secret_bundle: + secret_bundle = self._bind_legacy_secret_bundle( + job_id=job_id, + pipeline_cid=pipeline_cid, + pipeline=pipeline, + secret_bundle=secret_bundle, + ) + if secret_bundle.get(DAUTH_SECRET_PIPELINE_CID_KEY) != pipeline_cid: + raise ValueError(f"dAuth secret bundle generation mismatch for job {job_id}.") + current_pipeline_cid = self.chainstore_hget( + hkey=DEEPLOY_JOBS_CSTORE_HKEY, + key=job_id, + ) + if current_pipeline_cid != pipeline_cid: + raise ValueError(f"dAuth pipeline generation changed for job {job_id}.") encrypted_secret_bundle = self.bc.encrypt_str( str_data=self.json_dumps(secret_bundle), str_recipient=requester, diff --git a/extensions/business/dauth/dauth_registry.py b/extensions/business/dauth/dauth_registry.py index a9951972..9a72fa1a 100644 --- a/extensions/business/dauth/dauth_registry.py +++ b/extensions/business/dauth/dauth_registry.py @@ -1,6 +1,9 @@ """dAuth registry lookup and ChainStore routing helpers.""" +DAUTH_SECRET_PIPELINE_CID_KEY = "pipeline_cid" + + def resolve_dauth_registry_internal_peers(plugin, eth_oracles): """Resolve cached registry ETH addresses through current NetMon state.""" current_eth = plugin.bc.eth_address.lower() @@ -45,13 +48,19 @@ def get_cached_dauth_registry_internal_peers(plugin): def get_dauth_registry_internal_peers(plugin): - """Return cached peers when available, otherwise load a registry snapshot.""" + """Return cached manager peers or resolve peers for another plugin.""" if ( getattr(plugin, "_dauth_registry_eth_oracles", None) or hasattr(plugin, "_dauth_registry_internal_peers") ): return get_cached_dauth_registry_internal_peers(plugin) - peers, _ = load_dauth_registry_snapshot(plugin) + peers, _ = plugin.bc.get_dauth_oracles() + peers = list(dict.fromkeys( + peer for peer in peers or [] + if isinstance(peer, str) and peer + )) + if not peers: + raise ValueError("No dAuth registry internal peers are available.") return peers diff --git a/extensions/business/dauth/test_dauth_registry_gating.py b/extensions/business/dauth/test_dauth_registry_gating.py index eac59129..d8b16a25 100644 --- a/extensions/business/dauth/test_dauth_registry_gating.py +++ b/extensions/business/dauth/test_dauth_registry_gating.py @@ -6,6 +6,8 @@ from copy import deepcopy from pathlib import Path +from ratio1.const.base import dAuth + from extensions.business.dauth.dauth_mixin import ( DAUTH_JOB_SECRETS_CSTORE_HKEY, DEEPLOY_JOBS_CSTORE_HKEY, @@ -227,6 +229,9 @@ def is_dauth_oracle(self, node_address_eth=None): # pylint: disable=unused-argu def get_eth_oracles(self): return [self.node_eth.get(node, "0xORACLE") for node in self.protocol_oracles] + def get_dauth_oracles(self): + return list(self.protocol_oracles), ["Oracle"] * len(self.protocol_oracles) + def node_address_to_eth_address(self, node_address): return self.node_eth[node_address] @@ -421,6 +426,7 @@ def test_add_secrets_allows_protocol_oracle_and_overwrites_bundle(self): }, }, } + plugin._chainstore[(DEEPLOY_JOBS_CSTORE_HKEY, "7")] = "cid-7" response = plugin.process_dauth_add_secrets_request(body) @@ -432,6 +438,7 @@ def test_add_secrets_allows_protocol_oracle_and_overwrites_bundle(self): { "job_id": "7", "job_secrets": body["job_secrets"], + "pipeline_cid": "cid-7", }, ) @@ -499,6 +506,7 @@ def test_get_secrets_returns_bundle_for_node_running_job_from_r1fs_pipeline(self plugin = _make_dauth_harness() bundle = { "job_id": "7", + "pipeline_cid": "cid-7", "job_secrets": { "plugins": { "CONTAINER_APP_RUNNER": [{ @@ -540,6 +548,118 @@ def test_get_secrets_returns_bundle_for_node_running_job_from_r1fs_pipeline(self [(json.dumps(bundle), "node-runner")], ) + def test_get_secrets_rejects_bundle_bound_to_another_pipeline_generation(self): + plugin = _make_dauth_harness() + plugin._chainstore[(DAUTH_JOB_SECRETS_CSTORE_HKEY, "7")] = { + "job_id": "7", + "pipeline_cid": "cid-old", + "job_secrets": {"plugins": {}}, + } + plugin._chainstore[(DEEPLOY_JOBS_CSTORE_HKEY, "7")] = "cid-new" + plugin._r1fs_data["cid-new"] = { + "DEEPLOY_SPECS": {"current_target_nodes": ["node-runner"]}, + } + + with self.assertRaisesRegex(ValueError, "generation mismatch"): + plugin.process_dauth_get_secret_request({ + "EE_SENDER": "node-runner", + "EE_ETH_SENDER": "0xRUNNER", + "nonce": REQUEST_NONCE, + "job_id": "7", + }) + + def test_get_secrets_safely_binds_complete_legacy_bundle(self): + plugin = _make_dauth_harness() + legacy_bundle = { + "job_id": "7", + "job_secrets": { + "PLUGINS": [{"INSTANCES": [{"ENV": { + "TOKEN": "secret", + "REMOVED_TOKEN": "must-not-return", + }}]}], + "REMOVED_SECTION": {"PASSWORD": "must-not-return"}, + }, + } + plugin._chainstore[(DAUTH_JOB_SECRETS_CSTORE_HKEY, "7")] = legacy_bundle + plugin._chainstore[(DEEPLOY_JOBS_CSTORE_HKEY, "7")] = "cid-7" + plugin._r1fs_data["cid-7"] = { + "PLUGINS": [{ + "INSTANCES": [{"ENV": {"TOKEN": dAuth.DAUTH_SECRET_PLACEHOLDER}}], + }], + "DEEPLOY_SPECS": {"current_target_nodes": ["node-runner"]}, + } + + response = plugin.process_dauth_get_secret_request({ + "EE_SENDER": "node-runner", + "EE_ETH_SENDER": "0xRUNNER", + "nonce": REQUEST_NONCE, + "job_id": "7", + }) + + self.assertEqual(response["nonce"], REQUEST_NONCE) + self.assertEqual(response["encrypted_secret_bundle"], "encrypted-secret-bundle") + self.assertNotIn("secret_bundle", response) + encrypted_bundle = json.loads(plugin.bc.encrypt_calls[-1][0]) + self.assertEqual(encrypted_bundle["pipeline_cid"], "cid-7") + migrated_job_secrets = encrypted_bundle["job_secrets"] + self.assertEqual( + migrated_job_secrets, + {"PLUGINS": [{"INSTANCES": [{"ENV": {"TOKEN": "secret"}}]}]}, + ) + self.assertEqual( + plugin._chainstore[(DAUTH_JOB_SECRETS_CSTORE_HKEY, "7")]["pipeline_cid"], + "cid-7", + ) + + def test_get_secrets_rejects_incomplete_legacy_bundle(self): + plugin = _make_dauth_harness() + plugin._chainstore[(DAUTH_JOB_SECRETS_CSTORE_HKEY, "7")] = { + "job_id": "7", + "job_secrets": {"PLUGINS": [{"INSTANCES": [{"ENV": {}}]}]}, + } + plugin._chainstore[(DEEPLOY_JOBS_CSTORE_HKEY, "7")] = "cid-7" + plugin._r1fs_data["cid-7"] = { + "PLUGINS": [{ + "INSTANCES": [{"ENV": {"TOKEN": dAuth.DAUTH_SECRET_PLACEHOLDER}}], + }], + "DEEPLOY_SPECS": {"current_target_nodes": ["node-runner"]}, + } + + with self.assertRaisesRegex(ValueError, "incomplete"): + plugin.process_dauth_get_secret_request({ + "EE_SENDER": "node-runner", + "EE_ETH_SENDER": "0xRUNNER", + "nonce": REQUEST_NONCE, + "job_id": "7", + }) + + def test_get_secrets_does_not_bind_legacy_bundle_without_current_placeholders(self): + plugin = _make_dauth_harness() + plugin._chainstore[(DAUTH_JOB_SECRETS_CSTORE_HKEY, "7")] = { + "job_id": "7", + "job_secrets": {"PLUGINS": [{"INSTANCES": [{"ENV": { + "REMOVED_TOKEN": "must-not-return", + }}]}]}, + } + plugin._chainstore[(DEEPLOY_JOBS_CSTORE_HKEY, "7")] = "cid-7" + plugin._r1fs_data["cid-7"] = { + "PLUGINS": [{"INSTANCES": [{"ENV": {"PUBLIC": "value"}}]}], + "DEEPLOY_SPECS": {"current_target_nodes": ["node-runner"]}, + } + + with self.assertRaisesRegex(ValueError, "incomplete"): + plugin.process_dauth_get_secret_request({ + "EE_SENDER": "node-runner", + "EE_ETH_SENDER": "0xRUNNER", + "nonce": REQUEST_NONCE, + "job_id": "7", + }) + + self.assertNotIn( + "pipeline_cid", + plugin._chainstore[(DAUTH_JOB_SECRETS_CSTORE_HKEY, "7")], + ) + def test_get_secrets_rejects_node_not_running_job(self): plugin = _make_dauth_harness() plugin._chainstore[(DAUTH_JOB_SECRETS_CSTORE_HKEY, "7")] = { diff --git a/extensions/business/dauth/test_dauth_secret_routing.py b/extensions/business/dauth/test_dauth_secret_routing.py index 6b4fe843..0a43a722 100644 --- a/extensions/business/dauth/test_dauth_secret_routing.py +++ b/extensions/business/dauth/test_dauth_secret_routing.py @@ -26,6 +26,11 @@ def eth_addr_to_internal_addr(self, eth_address): return None +class _ProtocolOracleBCStub: + def get_eth_oracles(self): + return ["0xOracleA", "0xOracleB"] + + class _DauthStub(_DauthMixin): def __init__(self): self._dauth_registry_internal_peers = ["dauth-a", "dauth-b"] @@ -100,6 +105,14 @@ def test_secret_storage_fails_without_startup_cached_peers(self): self.assertEqual(plugin.writes, []) + def test_all_protocol_oracles_remain_authorized_writers(self): + plugin = _DauthStub() + plugin.bc = _ProtocolOracleBCStub() + + self.assertTrue(plugin._is_protocol_oracle_eth("0xoraclea")) + self.assertTrue(plugin._is_protocol_oracle_eth("0xOracleB")) + self.assertFalse(plugin._is_protocol_oracle_eth("0xNotOracle")) + if __name__ == "__main__": unittest.main() diff --git a/extensions/business/deeploy/deeploy_cmdapi_integration.py b/extensions/business/deeploy/deeploy_cmdapi_integration.py new file mode 100644 index 00000000..c0767a90 --- /dev/null +++ b/extensions/business/deeploy/deeploy_cmdapi_integration.py @@ -0,0 +1,20 @@ +"""Isolated integration with Core's pure CmdAPI pipeline builder.""" + + +def build_pipeline_config(plugin, **kwargs): + """Build pipeline config without registering or dispatching a command.""" + builder = getattr(plugin, "cmdapi_build_pipeline_config", None) + if not callable(builder): + raise RuntimeError("Core CmdAPI pure pipeline builder is unavailable.") + pipeline = builder(**kwargs) + if not isinstance(pipeline, dict): + raise RuntimeError("Core CmdAPI pipeline builder returned invalid metadata.") + return pipeline + + +def dispatch_pipeline_config(plugin, node_address, pipeline): + """Dispatch an already built and redacted pipeline config.""" + dispatcher = getattr(plugin, "cmdapi_start_pipeline", None) + if not callable(dispatcher): + raise RuntimeError("Core CmdAPI pipeline dispatcher is unavailable.") + dispatcher(config=pipeline, node_address=node_address) diff --git a/extensions/business/deeploy/deeploy_job_mixin.py b/extensions/business/deeploy/deeploy_job_mixin.py index cbde7b92..023d2df2 100644 --- a/extensions/business/deeploy/deeploy_job_mixin.py +++ b/extensions/business/deeploy/deeploy_job_mixin.py @@ -1,6 +1,15 @@ +from extensions.business.dauth.dauth_registry import ( + DAUTH_SECRET_PIPELINE_CID_KEY, + dauth_registry_write_kwargs, + get_dauth_registry_internal_peers, + pipeline_registry_write_kwargs, +) + NONCE = 42 R1FS_FILENAME = 'data.json' DEEPLOY_JOBS_CSTORE_HKEY = "DEEPLOY_DEPLOYED_JOBS" +DAUTH_JOB_SECRETS_CSTORE_HKEY = "DAUTH_JOB_SECRETS" + class _DeeployJobMixin: """ @@ -114,13 +123,163 @@ def save_job_pipeline_in_cstore(self, pipeline: dict, job_id: int): pipeline_key = str(job_id) - result = self.chainstore_hset(hkey=DEEPLOY_JOBS_CSTORE_HKEY, key=pipeline_key, value=cid) + result = self.chainstore_hset( + hkey=DEEPLOY_JOBS_CSTORE_HKEY, + key=pipeline_key, + value=cid, + **pipeline_registry_write_kwargs(self), + ) except Exception as e: self.P(f"Error saving pipeline for job {job_id} to CSTORE: {e}", color='r') return False return result + def _load_dauth_job_secret_bundle(self, job_id): + return self.chainstore_hget( + hkey=DAUTH_JOB_SECRETS_CSTORE_HKEY, + key=str(job_id), + ) + + def _write_dauth_job_secret_bundle(self, job_id, bundle, write_kwargs=None): + if write_kwargs is None: + write_kwargs = dauth_registry_write_kwargs(self) + return self.chainstore_hset( + hkey=DAUTH_JOB_SECRETS_CSTORE_HKEY, + key=str(job_id), + value=bundle, + **write_kwargs, + ) + + def _write_job_pipeline_cid(self, job_id, cid, write_kwargs=None): + if write_kwargs is None: + write_kwargs = pipeline_registry_write_kwargs(self) + return self.chainstore_hset( + hkey=DEEPLOY_JOBS_CSTORE_HKEY, + key=str(job_id), + value=cid, + **write_kwargs, + ) + + def stage_job_pipeline_and_secrets(self, pipeline, job_id, secret_bundle): + """Stage redacted pipeline metadata and its complete dAuth bundle.""" + if job_id in [None, ""]: + raise ValueError("Cannot stage Deeploy metadata without job_id.") + if not isinstance(pipeline, dict): + raise ValueError("Cannot stage invalid Deeploy pipeline metadata.") + if not isinstance(secret_bundle, dict): + raise ValueError("Cannot stage invalid dAuth secret bundle.") + + job_id = str(job_id) + # Resolve both routes before creating an R1FS object, so a missing registry + # cannot leave an unreferenced staged CID behind. + registry_peers = get_dauth_registry_internal_peers(self) + pipeline_write_kwargs = pipeline_registry_write_kwargs(self, peers=registry_peers) + secret_write_kwargs = dauth_registry_write_kwargs(self, peers=registry_peers) + prior_cid = self._get_pipeline_from_cstore(job_id) + prior_bundle = self._load_dauth_job_secret_bundle(job_id) + sanitized_pipeline = self.extract_invariable_data_from_pipeline(pipeline) + sorted_pipeline = self._recursively_sort_pipeline_data(sanitized_pipeline) + staged_cid = self._save_pipeline_to_r1fs(sorted_pipeline) + if not staged_cid: + raise ValueError(f"Failed to stage pipeline metadata for job {job_id} in R1FS.") + + bound_secret_bundle = self.deepcopy(secret_bundle) + bound_secret_bundle[DAUTH_SECRET_PIPELINE_CID_KEY] = staged_cid + state = { + "job_id": job_id, + "prior_cid": prior_cid, + "prior_bundle": self.deepcopy(prior_bundle), + "staged_cid": staged_cid, + "staged_bundle": self.deepcopy(bound_secret_bundle), + "pipeline_staged": False, + "bundle_staged": False, + "pipeline_write_kwargs": pipeline_write_kwargs, + "secret_write_kwargs": secret_write_kwargs, + } + try: + if not self._write_dauth_job_secret_bundle( + job_id, + self.deepcopy(bound_secret_bundle), + write_kwargs=secret_write_kwargs, + ): + raise ValueError(f"Failed to stage dAuth secrets for job {job_id}.") + state["bundle_staged"] = True + if not self._write_job_pipeline_cid( + job_id, + staged_cid, + write_kwargs=pipeline_write_kwargs, + ): + raise ValueError(f"Failed to stage pipeline CID for job {job_id}.") + state["pipeline_staged"] = True + except Exception: + self.rollback_staged_job_pipeline_and_secrets(state) + raise + return state + + def commit_staged_job_pipeline_and_secrets(self, state): + """Commit a staged transaction by removing its superseded R1FS object.""" + if not isinstance(state, dict): + return False + staged_cid = state.get("staged_cid") + try: + current_cid = self._get_pipeline_from_cstore(state.get("job_id")) + current_bundle = self._load_dauth_job_secret_bundle(state.get("job_id")) + except Exception as exc: + self.Pd(f"Unable to verify staged pipeline commit: {exc}", color='y') + return False + if ( + current_cid != staged_cid + or current_bundle != state.get("staged_bundle") + ): + return False + prior_cid = state.get("prior_cid") + if prior_cid and prior_cid != staged_cid: + self._delete_pipeline_cid_from_r1fs(prior_cid) + return True + + def rollback_staged_job_pipeline_and_secrets(self, state): + """Restore matching staged state without disturbing a newer deployment.""" + if not isinstance(state, dict): + return False + job_id = state.get("job_id") + staged_cid = state.get("staged_cid") + restored = False + try: + current_cid = self._get_pipeline_from_cstore(job_id) + current_bundle = self._load_dauth_job_secret_bundle(job_id) + expected_cid = ( + staged_cid if state.get("pipeline_staged") else state.get("prior_cid") + ) + expected_bundle = ( + state.get("staged_bundle") if state.get("bundle_staged") + else state.get("prior_bundle") + ) + if current_cid == expected_cid and current_bundle == expected_bundle: + try: + pipeline_ok = True + if state.get("pipeline_staged"): + pipeline_ok = self._write_job_pipeline_cid( + job_id, + state.get("prior_cid"), + write_kwargs=state.get("pipeline_write_kwargs"), + ) + bundle_ok = True + if state.get("bundle_staged"): + bundle_ok = self._write_dauth_job_secret_bundle( + job_id, + self.deepcopy(state.get("prior_bundle")), + write_kwargs=state.get("secret_write_kwargs"), + ) + restored = bool(pipeline_ok and bundle_ok) + if restored: + self._delete_pipeline_cid_from_r1fs(staged_cid) + except Exception as exc: + self.Pd(f"Unable to restore staged Deeploy metadata for job {job_id}: {exc}", color='y') + except Exception as exc: + self.Pd(f"Unable to verify staged Deeploy rollback for job {job_id}: {exc}", color='y') + return restored + def list_all_deployed_jobs_from_cstore(self): """ Get all the job pipelines from CSTORE. diff --git a/extensions/business/deeploy/deeploy_manager_api.py b/extensions/business/deeploy/deeploy_manager_api.py index f6675af3..82bc28ee 100644 --- a/extensions/business/deeploy/deeploy_manager_api.py +++ b/extensions/business/deeploy/deeploy_manager_api.py @@ -706,6 +706,7 @@ def _process_pipeline_request( dict The response dictionary """ + staging_state = None try: self.__ensure_eth_balance() request_type = "create pipeline" if is_create else "update pipeline" @@ -778,10 +779,10 @@ def _process_pipeline_request( deeploy_specs_for_update = None deeploy_specs_payload = None prepared_create_deploy_plan = None + prepared_pipeline_configs = None skip_create_response_key_reset = False - previous_pipeline_cid = None update_context_from_persisted_pipeline = False - dauth_secrets_stored = False + delete_existing_after_stage = False if is_create: is_valid = self.deeploy_check_payment_and_job_owner(inputs, auth_result[DEEPLOY_KEYS.ESCROW_OWNER], is_create=is_create, debug=self.cfg_deeploy_verbose > 1) if not is_valid: @@ -870,11 +871,14 @@ def _process_pipeline_request( inputs.target_nodes = deployment_targets inputs[DEEPLOY_KEYS.TARGET_NODES_COUNT] = len(deployment_targets) inputs.target_nodes_count = len(deployment_targets) - if job_id is not None: - try: - previous_pipeline_cid = self._get_pipeline_from_cstore(job_id) - except Exception as exc: - self.Pd(f"Unable to read previous pipeline CID for job {job_id}: {exc}", color='y') + # Ensure plugin IDs are preserved for existing instances before any destructive action. + self._ensure_plugin_instance_ids( + inputs, + discovered_plugin_instances=discovered_plugin_instances, + owner=auth_result[DEEPLOY_KEYS.ESCROW_OWNER], + app_id=app_id, + job_id=job_id, + ) if deeploy_specs_for_update is not None and not isinstance(deeploy_specs_for_update, dict): msg = ( @@ -960,12 +964,6 @@ def _process_pipeline_request( ) skip_create_response_key_reset = True - job_secrets = self._extract_dauth_job_secrets_from_prepared_deploy_plan( - prepared_create_deploy_plan - ) - dauth_secrets_stored = self._store_deeploy_dauth_job_secrets(job_id, job_secrets) - - # All validations, response-key resets, and dAuth writes passed; remove the running job and redeploy. if update_context_from_persisted_pipeline: # TODO: stop stale offline old-node pipelines through ChainDist reconciliation when they return. self.Pd( @@ -974,17 +972,10 @@ def _process_pipeline_request( color='y', ) else: - self.delete_pipeline_from_nodes( - app_id=app_id, - job_id=job_id, - owner=auth_result[DEEPLOY_KEYS.ESCROW_OWNER], - discovered_instances=discovered_plugin_instances, - ) - + delete_existing_after_stage = True deployment_nodes = list(validated_nodes) confirmation_nodes = list(validated_nodes) nodes_changed = set(current_nodes) != set(deployment_nodes) - discovered_plugin_instances = [] inputs[DEEPLOY_KEYS.TARGET_NODES] = deployment_nodes inputs.target_nodes = deployment_nodes @@ -1016,11 +1007,43 @@ def _process_pipeline_request( job_app_type=job_app_type, dct_deeploy_specs=deeploy_specs_payload, ) - if not dauth_secrets_stored: - job_secrets = self._extract_dauth_job_secrets_from_prepared_deploy_plan( - prepared_create_deploy_plan + prepared_pipeline_configs = self._build_create_pipeline_configs( + inputs=inputs, + app_id=app_id, + app_alias=app_alias, + app_type=app_type, + owner=auth_result[DEEPLOY_KEYS.ESCROW_OWNER], + prepared_deploy_plan=prepared_create_deploy_plan, + ) + prior_bundle = self._load_dauth_job_secret_bundle(job_id) + prepared_pipeline_configs, complete_secret_bundle = ( + self._redact_pipeline_configs_and_build_secret_bundle( + job_id=job_id, + pipeline_configs=prepared_pipeline_configs, + prior_bundle=prior_bundle, + ) + ) + node_plugins_by_addr = prepared_create_deploy_plan.get("node_plugins_by_addr", {}) + for node, pipeline_config in prepared_pipeline_configs.items(): + if node in node_plugins_by_addr: + node_plugins_by_addr[node] = self.deepcopy( + pipeline_config.get(self.ct.CONFIG_STREAM.K_PLUGINS, []) + ) + pipeline_to_stage = next(iter(prepared_pipeline_configs.values())) + staging_state = self.stage_job_pipeline_and_secrets( + pipeline=pipeline_to_stage, + job_id=job_id, + secret_bundle=complete_secret_bundle, + ) + + if delete_existing_after_stage: + self.delete_pipeline_from_nodes( + app_id=app_id, + job_id=job_id, + owner=auth_result[DEEPLOY_KEYS.ESCROW_OWNER], + discovered_instances=discovered_plugin_instances, ) - self._store_deeploy_dauth_job_secrets(job_id, job_secrets) + discovered_plugin_instances = [] dct_status, str_status, response_keys, pipeline_to_persist = self.check_and_deploy_pipelines( owner=auth_result[DEEPLOY_KEYS.ESCROW_OWNER], @@ -1033,18 +1056,11 @@ def _process_pipeline_request( discovered_plugin_instances=discovered_plugin_instances, dct_deeploy_specs_create=deeploy_specs_payload, prepared_create_deploy_plan=prepared_create_deploy_plan, + prepared_pipeline_configs=prepared_pipeline_configs, skip_create_response_key_reset=skip_create_response_key_reset, job_app_type=job_app_type, wait_for_responses=not async_mode, ) - persistence_state = self._build_pipeline_persistence_state( - job_id=job_id, - pipeline=pipeline_to_persist, - app_id=app_id, - previous_cid=previous_pipeline_cid, - delete_previous=not is_create, - ) - return_request = request.get(DEEPLOY_KEYS.RETURN_REQUEST, False) if return_request: dct_request = self._redact_deeploy_dauth_secrets_for_response(request) @@ -1066,7 +1082,8 @@ def _process_pipeline_request( # dct_request[DEEPLOY_KEYS.PIPELINE_PARAMS] = pipeline_params if async_mode: if len(response_keys) == 0: - self._queue_pipeline_persistence(persistence_state) + self.commit_staged_job_pipeline_and_secrets(staging_state) + staging_state = None if nodes_changed and not is_confirmable_job: eth_nodes = [self.bc.node_addr_to_eth_addr(node) for node in confirmation_nodes] eth_nodes = sorted(eth_nodes) @@ -1108,13 +1125,10 @@ def _process_pipeline_request( 'is_confirmable_job': is_confirmable_job, 'job_id': job_id, }, - 'persistence': persistence_state, + 'staging': staging_state, } return {'__pending__': pending_state} - if str_status in [DEEPLOY_STATUS.SUCCESS, DEEPLOY_STATUS.COMMAND_DELIVERED]: - self._queue_pipeline_persistence(persistence_state) - if nodes_changed and str_status in [DEEPLOY_STATUS.SUCCESS, DEEPLOY_STATUS.COMMAND_DELIVERED]: if (dct_status is not None and is_confirmable_job and len(confirmation_nodes) == len(dct_status)) or not is_confirmable_job: eth_nodes = [self.bc.node_addr_to_eth_addr(node) for node in confirmation_nodes] @@ -1131,6 +1145,12 @@ def _process_pipeline_request( #endif #endif + if str_status in [DEEPLOY_STATUS.SUCCESS, DEEPLOY_STATUS.COMMAND_DELIVERED]: + self.commit_staged_job_pipeline_and_secrets(staging_state) + else: + self.rollback_staged_job_pipeline_and_secrets(staging_state) + staging_state = None + result = { DEEPLOY_KEYS.STATUS: str_status, DEEPLOY_KEYS.STATUS_DETAILS: dct_status, @@ -1142,6 +1162,8 @@ def _process_pipeline_request( if self.cfg_deeploy_verbose > 1: self.P(f"Request Result: {result}") except Exception as e: + if staging_state is not None: + self.rollback_staged_job_pipeline_and_secrets(staging_state) result = self.__handle_error(e, request) #endtry @@ -1197,6 +1219,7 @@ def maybe_mark_timed_out_request(self, pending_id: str, pending, now: float = No if now is None: now = self.time() if (now - pending['start_time']) > pending['timeout']: + self.rollback_staged_job_pipeline_and_secrets(pending.get('staging')) if pending.get('kind') == 'scale_up': result = { DEEPLOY_KEYS.STATUS: DEEPLOY_STATUS.TIMEOUT, @@ -1259,7 +1282,9 @@ def finalize_pending_request_pipeline( # endif nodes changed and success or delivered if str_status in [DEEPLOY_STATUS.SUCCESS, DEEPLOY_STATUS.COMMAND_DELIVERED]: - self._queue_pipeline_persistence(pending.get('persistence')) + self.commit_staged_job_pipeline_and_secrets(pending.get('staging')) + else: + self.rollback_staged_job_pipeline_and_secrets(pending.get('staging')) return { DEEPLOY_KEYS.STATUS: str_status, @@ -1301,6 +1326,10 @@ def finalize_pending_request_scale_up( job_id=job_id, is_confirmable_job=is_confirmable_job, ) + if str_status in [DEEPLOY_STATUS.SUCCESS, DEEPLOY_STATUS.COMMAND_DELIVERED]: + self.commit_staged_job_pipeline_and_secrets(pending.get('staging')) + else: + self.rollback_staged_job_pipeline_and_secrets(pending.get('staging')) return { DEEPLOY_KEYS.STATUS: str_status, DEEPLOY_KEYS.STATUS_DETAILS: dct_status, @@ -1651,6 +1680,7 @@ def scale_up_job_workers(self, dict A dictionary with the result of the operation """ + staging_state = None try: self.__ensure_eth_balance() sender, inputs = self.deeploy_verify_and_get_inputs(request, request_type="scale up workers") @@ -1685,7 +1715,7 @@ def scale_up_job_workers(self, update_nodes = list(running_apps_for_job.keys()) new_nodes = self._check_nodes_availability(inputs) - dct_status, str_status, response_keys = self.scale_up_job( + dct_status, str_status, response_keys, staging_state = self.scale_up_job( new_nodes=new_nodes, update_nodes=update_nodes, owner=auth_result[DEEPLOY_KEYS.ESCROW_OWNER], @@ -1700,6 +1730,8 @@ def scale_up_job_workers(self, else: dct_request = None if len(response_keys) == 0: + self.commit_staged_job_pipeline_and_secrets(staging_state) + staging_state = None if not is_confirmable_job: nodes = list(set(update_nodes + new_nodes)) self.Pd(f"Nodes to confirm (non-confirmable job): {self.json_dumps(nodes, indent=2)}") @@ -1733,10 +1765,13 @@ def scale_up_job_workers(self, 'is_confirmable_job': is_confirmable_job, 'request': dct_request, 'auth': auth_result, + 'staging': staging_state, } return self._register_pending_deploy_request(pending_state) except Exception as e: + if staging_state is not None: + self.rollback_staged_job_pipeline_and_secrets(staging_state) result = self.__handle_error(e, request) #endtry diff --git a/extensions/business/deeploy/deeploy_mixin.py b/extensions/business/deeploy/deeploy_mixin.py index 8d86337a..e9917f55 100644 --- a/extensions/business/deeploy/deeploy_mixin.py +++ b/extensions/business/deeploy/deeploy_mixin.py @@ -9,14 +9,21 @@ from naeural_core.constants import BASE_CT from naeural_core.main.net_mon import NetMonCt + +from extensions.business.dauth.dauth_registry import dauth_registry_write_kwargs from naeural_core import constants as ct from ratio1.bc.base import compact_canonical_sha256 +from ratio1.const.base import dAuth from extensions.business.deeploy.deeploy_const import DEEPLOY_ERRORS, DEEPLOY_KEYS, \ DEEPLOY_STATUS, DEEPLOY_PLUGIN_DATA, DEEPLOY_FORBIDDEN_SIGNATURES, CONTAINER_APP_RUNNER_SIGNATURE, \ DEEPLOY_RESOURCES, JOB_TYPE_RESOURCE_SPECS, WORKER_APP_RUNNER_SIGNATURE, JOB_APP_TYPES, JOB_APP_TYPES_ALL, \ CONTAINERIZED_APPS_SIGNATURES, DEEPLOY_PLUS_PREFERRED_NODES_HKEY, DEEPLOY_RUNTIME_KEYS, \ DEEPLOY_DYNAMIC_ENV_KEYS, DEEPLOY_DYNAMIC_ENV_TYPES +from extensions.business.deeploy.deeploy_cmdapi_integration import ( + build_pipeline_config, + dispatch_pipeline_config, +) from extensions.utils.memory_formatter import parse_memory_to_mb from extensions.utils.per_node_config import ( @@ -57,7 +64,7 @@ PREFERRED_NODES_MAX_PAYLOAD_BYTES = 32 * 1024 PREFERRED_NODE_ALIAS_MAX_LENGTH = 128 PREFERRED_NODE_DESCRIPTION_MAX_LENGTH = 512 -DEEPLOY_DAUTH_SECRET_PLACEHOLDER = "__R1_DAUTH_SECRET__" +DEEPLOY_DAUTH_SECRET_PLACEHOLDER = dAuth.DAUTH_SECRET_PLACEHOLDER DEEPLOY_DAUTH_JOB_SECRETS_HKEY = "DAUTH_JOB_SECRETS" DEEPLOY_DAUTH_SECRET_PATH_SUFFIXES = ( ("CLOUDFLARE_TOKEN",), @@ -932,6 +939,7 @@ def __create_pipeline_on_nodes( job_app_type=None, dct_deeploy_specs=None, prepared_deploy_plan=None, + prepared_pipeline_configs=None, skip_response_key_reset=False, ): """ @@ -960,28 +968,27 @@ def __create_pipeline_on_nodes( context=f"create pipeline '{app_alias}'", ) + if prepared_pipeline_configs is None: + prepared_pipeline_configs = self._build_create_pipeline_configs( + inputs=inputs, + app_id=app_id, + app_alias=app_alias, + app_type=app_type, + owner=owner, + prepared_deploy_plan=prepared_deploy_plan, + ) + saved_pipeline = None - for addr, node_plugins in node_plugins_by_addr.items(): - node_plugins = self.deepcopy(node_plugins) + for addr, pipeline_config in prepared_pipeline_configs.items(): + pipeline_config = self.deepcopy(pipeline_config) msg = '' if self.cfg_deeploy_verbose > 1: - msg = f":\n {self.json_dumps(self._redact_per_node_config_for_log(node_plugins), indent=2)}" + msg = f":\n {self.json_dumps(self._redact_per_node_config_for_log(pipeline_config), indent=2)}" self.P(f"Creating pipeline '{app_alias}' on {addr}{msg}") if addr is not None: - - saved_pipeline = self.cmdapi_start_pipeline_by_params( - name=app_id, - app_alias=app_alias, - pipeline_type=app_type, - node_address=addr, - owner=owner, - url=inputs.pipeline_input_uri, - plugins=node_plugins, - is_deeployed=True, - deeploy_specs=self.deepcopy(dct_deeploy_specs), - **pipeline_kwargs, - ) + saved_pipeline = pipeline_config + dispatch_pipeline_config(self, addr, pipeline_config) # endif addr is valid # endfor each target node @@ -989,6 +996,62 @@ def __create_pipeline_on_nodes( cleaned_response_keys = prepared_response_keys if enable_chainstore_response else {} return cleaned_response_keys, saved_pipeline + def _build_create_pipeline_configs( + self, + inputs, + app_id, + app_alias, + app_type, + owner, + prepared_deploy_plan, + ): + """Use Core's pure builder to create exact per-node pipeline metadata.""" + dct_deeploy_specs = self.deepcopy(prepared_deploy_plan.get("deeploy_specs", {})) + pipeline_kwargs = self.deepcopy(prepared_deploy_plan.get("pipeline_kwargs", {})) + node_plugins_by_addr = prepared_deploy_plan.get("node_plugins_by_addr", {}) + configs = {} + for addr, node_plugins in node_plugins_by_addr.items(): + configs[addr] = build_pipeline_config( + self, + name=app_id, + stream_type=app_type, + app_alias=app_alias, + owner=owner, + url=inputs.get(DEEPLOY_KEYS.PIPELINE_INPUT_URI), + plugins=self.deepcopy(node_plugins), + is_deeployed=True, + deeploy_specs=self.deepcopy(dct_deeploy_specs), + **pipeline_kwargs, + ) + return configs + + def _redact_pipeline_configs_and_build_secret_bundle( + self, + job_id, + pipeline_configs, + prior_bundle, + ): + """Redact exact command metadata and reconstruct its complete bundle.""" + redacted_configs = {} + merged_new_secrets = None + for node, pipeline in pipeline_configs.items(): + redacted, new_secrets = self._extract_and_redact_deeploy_dauth_secrets(pipeline) + redacted_configs[node] = redacted + merged_new_secrets = self._merge_deeploy_dauth_secret_fragments( + merged_new_secrets, + new_secrets, + ) + if not redacted_configs: + raise ValueError("Cannot stage Deeploy metadata without target pipeline configs.") + representative = next(iter(redacted_configs.values())) + bundle = self._build_complete_dauth_job_secret_bundle( + job_id=job_id, + redacted_pipeline=representative, + new_job_secrets=merged_new_secrets, + prior_bundle=prior_bundle, + ) + return redacted_configs, bundle + def __update_pipeline_on_nodes( self, nodes, @@ -3824,6 +3887,97 @@ def _extract_dauth_job_secrets_from_prepared_deploy_plan(self, prepared_deploy_p return None return {"PLUGINS": merged_plugins_secrets} + def _get_dauth_secret_fragment_value(self, fragment, path): + current = fragment + for part in path: + if isinstance(current, dict) and part in current: + current = current[part] + elif isinstance(current, list) and isinstance(part, int) and part < len(current): + current = current[part] + else: + return False, None + return True, current + + def _set_dauth_secret_fragment_value(self, fragment, path, value): + current = fragment + for idx, part in enumerate(path): + is_last = idx == len(path) - 1 + next_part = None if is_last else path[idx + 1] + if isinstance(part, int): + while len(current) <= part: + current.append(None) + if is_last: + current[part] = self.deepcopy(value) + continue + if current[part] is None: + current[part] = [] if isinstance(next_part, int) else {} + current = current[part] + else: + if is_last: + current[part] = self.deepcopy(value) + continue + if part not in current or current[part] is None: + current[part] = [] if isinstance(next_part, int) else {} + current = current[part] + return fragment + + def _iter_deeploy_dauth_placeholder_paths(self, payload): + def walk(value, path): + if isinstance(value, dict): + for key, item in value.items(): + item_path = path + [key] + if item == DEEPLOY_DAUTH_SECRET_PLACEHOLDER: + yield item_path + else: + yield from walk(item, item_path) + elif isinstance(value, list): + for idx, item in enumerate(value): + yield from walk(item, path + [idx]) + yield from walk(payload, []) + + def _build_complete_dauth_job_secret_bundle( + self, + job_id, + redacted_pipeline, + new_job_secrets, + prior_bundle=None, + ): + """Reconstruct the complete current bundle from exact placeholder paths.""" + prior_job_secrets = {} + if isinstance(prior_bundle, dict): + candidate = prior_bundle.get("job_secrets") + if isinstance(candidate, (dict, list)): + prior_job_secrets = candidate + if not isinstance(new_job_secrets, (dict, list)): + new_job_secrets = {} + + complete = {} + unresolved = [] + for path in self._iter_deeploy_dauth_placeholder_paths(redacted_pipeline): + found, value = self._get_dauth_secret_fragment_value(new_job_secrets, path) + if not found: + found, value = self._get_dauth_secret_fragment_value(prior_job_secrets, path) + if ( + not found + or value is None + or value == DEEPLOY_DAUTH_SECRET_PLACEHOLDER + or isinstance(value, (dict, list)) + ): + unresolved.append("/".join(str(part) for part in path)) + continue + self._set_dauth_secret_fragment_value(complete, path, value) + + if unresolved: + raise ValueError( + "Unresolved dAuth secret placeholder path(s): {}.".format( + ", ".join(unresolved) + ) + ) + return { + "job_id": str(job_id), + "job_secrets": complete, + } + def _store_deeploy_dauth_job_secrets(self, job_id, job_secrets): if not job_secrets: return False @@ -3838,6 +3992,7 @@ def _store_deeploy_dauth_job_secrets(self, job_id, job_secrets): hkey=DEEPLOY_DAUTH_JOB_SECRETS_HKEY, key=job_id, value=bundle, + **dauth_registry_write_kwargs(self), ) if not ok: raise ValueError(f"Failed to store dAuth secrets for job {job_id}.") @@ -4665,6 +4820,7 @@ def check_and_deploy_pipelines( dct_deeploy_specs=None, job_app_type=None, dct_deeploy_specs_create=None, prepared_create_deploy_plan=None, + prepared_pipeline_configs=None, skip_create_response_key_reset=False, wait_for_responses=True ): @@ -4732,6 +4888,7 @@ def check_and_deploy_pipelines( job_app_type=job_app_type, dct_deeploy_specs=dct_deeploy_specs_create, prepared_deploy_plan=prepared_create_deploy_plan, + prepared_pipeline_configs=prepared_pipeline_configs, skip_response_key_reset=skip_create_response_key_reset, ) response_keys.update(new_response_keys) @@ -4756,37 +4913,69 @@ def scale_up_job(self, new_nodes, update_nodes, job_id, owner, running_apps_for_ Scale up the job workers. """ - # todo: get pipeline from R1FS. - # Prepare updated app pipeline - base_pipeline = self.get_job_base_pipeline_from_apps(running_apps_for_job) + base_pipeline = self.get_job_base_pipeline_from_r1fs( + job_id, + owner=owner, + expected_nodes=update_nodes, + running_apps_for_job=running_apps_for_job, + ) create_pipelines, update_pipelines, chainstore_response_keys = ( self.prepare_create_update_pipelines(base_pipeline, new_nodes, update_nodes, running_apps_for_job)) + prepared_create_configs = self._build_scale_up_create_pipeline_configs( + create_pipelines=create_pipelines, + owner=owner, + ) + prior_bundle = self._load_dauth_job_secret_bundle(job_id) + prepared_create_configs, complete_secret_bundle = ( + self._redact_pipeline_configs_and_build_secret_bundle( + job_id=job_id, + pipeline_configs=prepared_create_configs, + prior_bundle=prior_bundle, + ) + ) + for node, pipeline in update_pipelines.items(): + redacted_plugins, _ = self._extract_and_redact_deeploy_dauth_secrets( + pipeline.get("plugins", []) + ) + pipeline["plugins"] = redacted_plugins + pipeline_to_stage = next(iter(prepared_create_configs.values())) + staging_state = self.stage_job_pipeline_and_secrets( + pipeline=pipeline_to_stage, + job_id=job_id, + secret_bundle=complete_secret_bundle, + ) + self.P(f"Prepared create pipelines: {self.json_dumps(self._redact_per_node_config_for_log(create_pipelines))}") self.P(f"Prepared update pipelines: {self.json_dumps(self._redact_per_node_config_for_log(update_pipelines))}") self.P(f"Prepared chainstore response keys: {self.json_dumps(chainstore_response_keys)}") - # RESET chainstore_response_keys here - self.P(f"Resetting chainstore keys: {self.json_dumps(chainstore_response_keys)}") - self._reset_chainstore_response_keys( - chainstore_response_keys, - context=f"scale up job {job_id}", - ) + try: + self.P(f"Resetting chainstore keys: {self.json_dumps(chainstore_response_keys)}") + self._reset_chainstore_response_keys( + chainstore_response_keys, + context=f"scale up job {job_id}", + ) - # Start pipelines on nodes. - self._start_create_update_pipelines(create_pipelines=create_pipelines, - update_pipelines=update_pipelines, - owner=owner) + self._start_create_update_pipelines( + create_pipelines=create_pipelines, + update_pipelines=update_pipelines, + owner=owner, + prepared_create_configs=prepared_create_configs, + ) + except Exception: + self.rollback_staged_job_pipeline_and_secrets(staging_state) + raise if wait_for_responses: dct_status, str_status = self._get_pipeline_responses(chainstore_response_keys, 300) else: dct_status, str_status = {}, DEEPLOY_STATUS.PENDING - return dct_status, str_status, chainstore_response_keys + return dct_status, str_status, chainstore_response_keys, staging_state def _discover_plugin_instances( self, @@ -5129,6 +5318,57 @@ def get_job_base_pipeline_from_apps(self, apps): "pipeline_params": pipeline_params, } + def get_job_base_pipeline_from_r1fs( + self, + job_id, + owner=None, + expected_nodes=None, + running_apps_for_job=None, + ): + """Load scale-up source metadata from the persisted R1FS pipeline.""" + pipeline = self.get_job_pipeline_from_cstore(job_id) + if not isinstance(pipeline, dict): + raise ValueError(f"No persisted R1FS pipeline found for job {job_id}.") + + name_key = self.ct.CONFIG_STREAM.K_NAME + type_key = self.ct.CONFIG_STREAM.K_TYPE + plugins_key = self.ct.CONFIG_STREAM.K_PLUGINS + specs_key = self.ct.CONFIG_STREAM.DEEPLOY_SPECS + plugins = pipeline.get(plugins_key) + specs = pipeline.get(specs_key) + if not isinstance(plugins, list) or not isinstance(specs, dict): + raise ValueError(f"Persisted R1FS pipeline for job {job_id} is incomplete.") + persisted_job_id = specs.get(DEEPLOY_KEYS.JOB_ID) + if persisted_job_id in [None, ""] or str(persisted_job_id) != str(job_id): + raise ValueError(f"Persisted R1FS pipeline job ID does not match job {job_id}.") + + persisted_owner = pipeline.get(NetMonCt.OWNER.upper()) + if owner is not None and persisted_owner != owner: + raise ValueError(f"Persisted R1FS pipeline owner does not match job {job_id}.") + + app_id = pipeline.get(name_key) + current_target_nodes = specs.get(DEEPLOY_KEYS.CURRENT_TARGET_NODES, []) + for node in expected_nodes or []: + if node not in current_target_nodes: + raise ValueError( + f"Persisted R1FS pipeline does not authorize existing node {node}." + ) + node_apps = (running_apps_for_job or {}).get(node) + if not isinstance(node_apps, dict) or app_id not in node_apps: + raise ValueError( + f"Persisted R1FS pipeline app ID does not match running node {node}." + ) + pipeline_params = self._get_pipeline_params_from_deeploy_specs(specs) or {} + return { + "base_pipeline": self.deepcopy(pipeline), + "app_id": pipeline.get(name_key), + "deeploy_specs": self.deepcopy(specs), + "plugins": self.deepcopy(plugins), + "pipeline_type": pipeline.get(type_key, "void"), + "url": pipeline.get(self.ct.CONFIG_STREAM.K_URL), + "pipeline_params": self.deepcopy(pipeline_params), + } + def prepare_create_update_pipelines(self, base_pipeline, new_nodes, update_nodes, running_apps_for_job): """ Prepare the create and update pipelines. @@ -5154,8 +5394,11 @@ def prepare_create_update_pipelines(self, base_pipeline, new_nodes, update_nodes raw_deeploy_specs = base_pipeline.get(NetMonCt.DEEPLOY_SPECS, {}) deeploy_specs = self.deepcopy(raw_deeploy_specs) if isinstance(raw_deeploy_specs, dict) else {} + persisted_nodes = deeploy_specs.get(DEEPLOY_KEYS.CURRENT_TARGET_NODES, []) + if not isinstance(persisted_nodes, list): + persisted_nodes = [] requested_nodes = [] - for node in list(update_nodes or []) + list(new_nodes or []): + for node in persisted_nodes + list(update_nodes or []) + list(new_nodes or []): if node not in requested_nodes: requested_nodes.append(node) chainstore_peers = self._ordered_nodes_for_per_node_config(requested_nodes, deeploy_specs) @@ -5286,10 +5529,8 @@ def prepare_create_update_pipelines(self, base_pipeline, new_nodes, update_nodes return create_pipelines, update_pipelines, prepared_response_keys - def _start_create_update_pipelines(self, create_pipelines, update_pipelines, owner): - """ - Start the create and update pipelines. - """ + def _build_scale_up_create_pipeline_configs(self, create_pipelines, owner): + configs = {} for node, pipeline in create_pipelines.items(): pipeline_params = pipeline.get('pipeline_params', {}) if not isinstance(pipeline_params, dict): @@ -5298,17 +5539,38 @@ def _start_create_update_pipelines(self, create_pipelines, update_pipelines, own pipeline_params, reserved_keys={"app_alias", "owner", "is_deeployed", "deeploy_specs"}, ) - self.cmdapi_start_pipeline_by_params( + configs[node] = build_pipeline_config( + self, name=pipeline['app_id'], - pipeline_type=pipeline['pipeline_type'], - node_address=node, - owner=owner, + stream_type=pipeline['pipeline_type'], + owner=owner, url=pipeline.get('url'), plugins=pipeline['plugins'], is_deeployed=True, deeploy_specs=pipeline['deeploy_specs'], **pipeline_kwargs, - ) + ) + if not configs: + raise ValueError("Scale-up did not produce a new-node pipeline config.") + return configs + + def _start_create_update_pipelines( + self, + create_pipelines, + update_pipelines, + owner, + prepared_create_configs=None, + ): + """ + Start the create and update pipelines. + """ + if prepared_create_configs is None: + prepared_create_configs = self._build_scale_up_create_pipeline_configs( + create_pipelines, + owner, + ) + for node, pipeline_config in prepared_create_configs.items(): + dispatch_pipeline_config(self, node, self.deepcopy(pipeline_config)) for node, pipeline in update_pipelines.items(): # For update pipelines, we need to iterate through the plugins and instances for plugin in pipeline['plugins']: diff --git a/extensions/business/deeploy/test_deeploy.py b/extensions/business/deeploy/test_deeploy.py index 695180cd..feca5ec8 100644 --- a/extensions/business/deeploy/test_deeploy.py +++ b/extensions/business/deeploy/test_deeploy.py @@ -338,7 +338,7 @@ def scale_up_job(self, new_nodes, update_nodes, job_id, owner, running_apps_for_ (dct_status, str_status, response_keys) """ response_keys = {"node1": ["k1"]} - return {}, DEEPLOY_STATUS.PENDING, response_keys + return {}, DEEPLOY_STATUS.PENDING, response_keys, None class DeeployPostponedTests(unittest.TestCase): @@ -398,9 +398,13 @@ def test_solve_postponed_pipeline_timeout(self): 'confirm': {}, } self.plugin._now = 5 + rolled_back = [] + self.plugin.rollback_staged_job_pipeline_and_secrets = lambda state: rolled_back.append(state) + self.plugin._DeeployManagerApiPlugin__pending_deploy_requests[pending_id]['staging'] = {"cid": "staged"} res = self.plugin.solve_postponed_deploy_request(pending_id) self.assertEqual(res[DEEPLOY_KEYS.STATUS], DEEPLOY_STATUS.TIMEOUT) self.assertEqual(res[DEEPLOY_KEYS.APP_ID], "app1") + self.assertEqual(rolled_back, [{"cid": "staged"}]) def test_solve_postponed_scale_up_timeout(self): """ @@ -470,9 +474,13 @@ def test_solve_postponed_pipeline_success_confirms(self): } self.plugin._chainstore["k1"] = {"ok": True} self.plugin._chainstore["k2"] = {"ok": True} + committed = [] + self.plugin.commit_staged_job_pipeline_and_secrets = lambda state: committed.append(state) + self.plugin._DeeployManagerApiPlugin__pending_deploy_requests[pending_id]['staging'] = {"cid": "staged"} res = self.plugin.solve_postponed_deploy_request(pending_id) self.assertEqual(res[DEEPLOY_KEYS.STATUS], DEEPLOY_STATUS.SUCCESS) self.assertEqual(self.plugin.bc.submitted, [(77, ["eth_nodeA"])]) + self.assertEqual(committed, [{"cid": "staged"}]) def test_solve_postponed_missing_pending(self): """ @@ -548,7 +556,7 @@ def test_scale_up_job_workers_command_delivered(self): Ensure scale_up_job_workers returns command-delivered when no keys exist. """ def scale_up_job(new_nodes, update_nodes, job_id, owner, running_apps_for_job, wait_for_responses=True): - return {}, DEEPLOY_STATUS.PENDING, {} + return {}, DEEPLOY_STATUS.PENDING, {}, None self.plugin.scale_up_job = scale_up_job req = { DEEPLOY_KEYS.JOB_ID: 10, diff --git a/extensions/business/deeploy/tests/support.py b/extensions/business/deeploy/tests/support.py index 796cbc4e..90555b6e 100644 --- a/extensions/business/deeploy/tests/support.py +++ b/extensions/business/deeploy/tests/support.py @@ -170,6 +170,42 @@ def make_deeploy_plugin(): plugin.Pd = lambda *args, **kwargs: None plugin.json_dumps = lambda obj, **kwargs: str(obj) + def build_pipeline_config(**kwargs): + config = { + "NAME": kwargs["name"], + "TYPE": kwargs["stream_type"], + } + if kwargs.get("url") is not None: + config["URL"] = kwargs["url"] + if kwargs.get("plugins") is not None: + config["PLUGINS"] = copy.deepcopy(kwargs["plugins"]) + ignored = {"name", "stream_type", "url", "plugins", "node_address", "send_immediately"} + config.update({ + key.upper(): copy.deepcopy(value) + for key, value in kwargs.items() + if key not in ignored + }) + return config + + plugin.cmdapi_build_pipeline_config = build_pipeline_config + + def dispatch_pipeline_config(config, node_address=None): + config = copy.deepcopy(config) + name = config.pop("NAME") + pipeline_type = config.pop("TYPE") + url = config.pop("URL", None) + plugins = config.pop("PLUGINS", None) + return plugin.cmdapi_start_pipeline_by_params( + name=name, + pipeline_type=pipeline_type, + node_address=node_address, + url=url, + plugins=plugins, + **{key.lower(): value for key, value in config.items()}, + ) + + plugin.cmdapi_start_pipeline = dispatch_pipeline_config + uuid_counter = {'value': 0} def uuid(size=6): diff --git a/extensions/business/deeploy/tests/test_cmdapi_staging_integration.py b/extensions/business/deeploy/tests/test_cmdapi_staging_integration.py new file mode 100644 index 00000000..6c81a29e --- /dev/null +++ b/extensions/business/deeploy/tests/test_cmdapi_staging_integration.py @@ -0,0 +1,104 @@ +import importlib.util +import unittest +from pathlib import Path + +from naeural_core import constants as core_constants + +from extensions.business.deeploy.deeploy_cmdapi_integration import ( + build_pipeline_config, + dispatch_pipeline_config, +) +from extensions.business.deeploy.tests.support import make_deeploy_plugin + + +class _CmdApiPluginStub: + def __init__(self): + self.built = [] + self.dispatched = [] + + def cmdapi_build_pipeline_config(self, **kwargs): + self.built.append(kwargs) + return { + "NAME": kwargs["name"], + "TYPE": kwargs["stream_type"], + "PLUGINS": kwargs["plugins"], + } + + def cmdapi_start_pipeline(self, config, node_address=None): + self.dispatched.append((node_address, config)) + + +class CmdApiStagingIntegrationTests(unittest.TestCase): + def test_installed_core_exposes_required_pure_builder(self): + cmdapi_path = Path(core_constants.__file__).resolve().parent / "business" / "mixins_base" / "cmdapi.py" + spec = importlib.util.spec_from_file_location("edge_test_core_cmdapi", cmdapi_path) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + + self.assertTrue(callable(getattr(module._CmdAPIMixin, "cmdapi_build_pipeline_config", None))) + + def test_pure_builder_does_not_dispatch_before_explicit_send(self): + plugin = _CmdApiPluginStub() + + pipeline = build_pipeline_config( + plugin, + name="app", + stream_type="void", + plugins=[{"SIGNATURE": "A", "INSTANCES": []}], + ) + + self.assertEqual(len(plugin.built), 1) + self.assertEqual(plugin.dispatched, []) + dispatch_pipeline_config(plugin, "node-a", pipeline) + self.assertEqual(plugin.dispatched, [("node-a", pipeline)]) + + def test_scale_up_base_pipeline_comes_from_r1fs_metadata(self): + plugin = make_deeploy_plugin() + plugin.get_job_pipeline_from_cstore = lambda job_id: { + "NAME": "app", + "TYPE": "void", + "URL": "https://example.invalid/input", + "OWNER": "0xowner", + "PLUGINS": [{"SIGNATURE": "A", "INSTANCES": []}], + "DEEPLOY_SPECS": { + "job_id": job_id, + "current_target_nodes": ["node-a"], + }, + } + plugin._get_pipeline_params_from_deeploy_specs = lambda specs: {"CUSTOM": 1} + + base = plugin.get_job_base_pipeline_from_r1fs(7) + + self.assertEqual(base["app_id"], "app") + self.assertEqual(base["plugins"][0]["SIGNATURE"], "A") + self.assertEqual(base["pipeline_params"], {"CUSTOM": 1}) + + def test_scale_up_base_rejects_mismatched_job_owner_and_running_app(self): + plugin = make_deeploy_plugin() + plugin.get_job_pipeline_from_cstore = lambda job_id: { + "NAME": "persisted-app", + "TYPE": "void", + "OWNER": "0xowner", + "PLUGINS": [], + "DEEPLOY_SPECS": { + "job_id": "7", + "current_target_nodes": ["node-a"], + }, + } + plugin._get_pipeline_params_from_deeploy_specs = lambda specs: {} + + with self.assertRaisesRegex(ValueError, "job ID"): + plugin.get_job_base_pipeline_from_r1fs(8) + with self.assertRaisesRegex(ValueError, "owner"): + plugin.get_job_base_pipeline_from_r1fs(7, owner="0xother") + with self.assertRaisesRegex(ValueError, "app ID"): + plugin.get_job_base_pipeline_from_r1fs( + 7, + owner="0xowner", + expected_nodes=["node-a"], + running_apps_for_job={"node-a": {"different-app": {}}}, + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/extensions/business/deeploy/tests/test_process_request.py b/extensions/business/deeploy/tests/test_process_request.py index 1670a57e..137d1d08 100644 --- a/extensions/business/deeploy/tests/test_process_request.py +++ b/extensions/business/deeploy/tests/test_process_request.py @@ -110,6 +110,40 @@ def chainstore_hset(self, hkey, key, value): }) return True + def cmdapi_build_pipeline_config(self, **kwargs): + config = { + "NAME": kwargs["name"], + "TYPE": kwargs["stream_type"], + } + if kwargs.get("url") is not None: + config["URL"] = kwargs["url"] + if kwargs.get("plugins") is not None: + config["PLUGINS"] = copy.deepcopy(kwargs["plugins"]) + ignored = {"name", "stream_type", "url", "plugins"} + config.update({ + key.upper(): copy.deepcopy(value) + for key, value in kwargs.items() + if key not in ignored + }) + return config + + def _load_dauth_job_secret_bundle(self, job_id): + return None + + def stage_job_pipeline_and_secrets(self, pipeline, job_id, secret_bundle): + self.chainstore_hset( + hkey=DEEPLOY_DAUTH_JOB_SECRETS_HKEY, + key=str(job_id), + value=secret_bundle, + ) + return {"job_id": str(job_id), "pipeline": copy.deepcopy(pipeline)} + + def commit_staged_job_pipeline_and_secrets(self, state): + return True + + def rollback_staged_job_pipeline_and_secrets(self, state): + return True + class DeeployProcessRequestTests(unittest.TestCase): diff --git a/extensions/business/deeploy/tests/test_secret_staging.py b/extensions/business/deeploy/tests/test_secret_staging.py new file mode 100644 index 00000000..648df16f --- /dev/null +++ b/extensions/business/deeploy/tests/test_secret_staging.py @@ -0,0 +1,221 @@ +import copy +import unittest + +from extensions.business.deeploy.deeploy_job_mixin import ( + DAUTH_JOB_SECRETS_CSTORE_HKEY, + DEEPLOY_JOBS_CSTORE_HKEY, + _DeeployJobMixin, +) +from extensions.business.deeploy.deeploy_mixin import ( + DEEPLOY_DAUTH_SECRET_PLACEHOLDER, + _DeeployMixin, +) + + +class _BCStub: + def get_dauth_oracles(self): + return ["dauth-a", "dauth-b"], ["a", "b"] + + +class _R1FSStub: + def __init__(self, events): + self.events = events + self.next_cid = "staged-cid" + + def add_json(self, payload, **kwargs): + self.events.append(("r1fs_add", copy.deepcopy(payload), kwargs)) + return self.next_cid + + def calculate_json_cid(self, payload, **kwargs): + return self.next_cid + + def delete_file(self, cid, **kwargs): + self.events.append(("r1fs_delete", cid, kwargs)) + + +class _SecretStagingPlugin(_DeeployMixin, _DeeployJobMixin): + def __init__(self): + self.events = [] + self.bc = _BCStub() + self.r1fs = _R1FSStub(self.events) + self.store = { + (DEEPLOY_JOBS_CSTORE_HKEY, "7"): "prior-cid", + (DAUTH_JOB_SECRETS_CSTORE_HKEY, "7"): { + "job_id": "7", + "job_secrets": { + "PLUGINS": [{ + "INSTANCES": [{ + "ENV": { + "CF_TUNNEL_TOKEN": "prior-token", + "CRDB_PASSWORD": "removed-password", + }, + }], + }], + }, + }, + } + self.deepcopy = copy.deepcopy + self.cfg_deeploy_verbose = 0 + + def chainstore_hget(self, hkey, key): + return copy.deepcopy(self.store.get((hkey, str(key)))) + + def chainstore_hset(self, hkey, key, value, **kwargs): + self.events.append(("hset", hkey, str(key), copy.deepcopy(value), kwargs)) + self.store[(hkey, str(key))] = copy.deepcopy(value) + return True + + def _get_pipeline_from_cstore(self, job_id): + return self.chainstore_hget(DEEPLOY_JOBS_CSTORE_HKEY, str(job_id)) + + def _redact_per_node_config_for_log(self, payload): + return "redacted" + + def json_dumps(self, payload, **kwargs): + return str(payload) + + def Pd(self, *args, **kwargs): + return + + +def _pipeline(token=DEEPLOY_DAUTH_SECRET_PLACEHOLDER): + return { + "NAME": "app", + "TYPE": "void", + "PLUGINS": [{ + "SIGNATURE": "CONTAINER_APP_RUNNER", + "INSTANCES": [{ + "INSTANCE_ID": "car", + "ENV": {"CF_TUNNEL_TOKEN": token}, + }], + }], + } + + +class DeeploySecretBundleTests(unittest.TestCase): + def setUp(self): + self.plugin = _SecretStagingPlugin() + + def test_reconstruction_uses_new_value_before_prior_same_path(self): + pipeline = _pipeline() + new_secrets = { + "PLUGINS": [{"INSTANCES": [{"ENV": {"CF_TUNNEL_TOKEN": "new-token"}}]}], + } + + bundle = self.plugin._build_complete_dauth_job_secret_bundle( + 7, + pipeline, + new_secrets, + self.plugin._load_dauth_job_secret_bundle(7), + ) + + env = bundle["job_secrets"]["PLUGINS"][0]["INSTANCES"][0]["ENV"] + self.assertEqual(env, {"CF_TUNNEL_TOKEN": "new-token"}) + + def test_reconstruction_reuses_prior_same_path_and_omits_removed_paths(self): + bundle = self.plugin._build_complete_dauth_job_secret_bundle( + 7, + _pipeline(), + {}, + self.plugin._load_dauth_job_secret_bundle(7), + ) + + env = bundle["job_secrets"]["PLUGINS"][0]["INSTANCES"][0]["ENV"] + self.assertEqual(env, {"CF_TUNNEL_TOKEN": "prior-token"}) + self.assertNotIn("CRDB_PASSWORD", env) + + def test_reconstruction_rejects_unresolved_placeholder(self): + with self.assertRaisesRegex(ValueError, "PLUGINS/0/INSTANCES/0/ENV/CF_TUNNEL_TOKEN"): + self.plugin._build_complete_dauth_job_secret_bundle(8, _pipeline(), {}, None) + + def test_reconstruction_rejects_unknown_placeholder_without_same_path_value(self): + pipeline = _pipeline() + pipeline["PLUGINS"][0]["INSTANCES"][0]["FUTURE_SECRET"] = ( + DEEPLOY_DAUTH_SECRET_PLACEHOLDER + ) + + with self.assertRaisesRegex(ValueError, "FUTURE_SECRET"): + self.plugin._build_complete_dauth_job_secret_bundle( + 7, + pipeline, + {}, + self.plugin._load_dauth_job_secret_bundle(7), + ) + + +class DeeploySecretStagingTests(unittest.TestCase): + def setUp(self): + self.plugin = _SecretStagingPlugin() + self.bundle = {"job_id": "7", "job_secrets": {"PLUGINS": []}} + + def test_stage_writes_complete_bundle_before_publishing_pipeline_pointer(self): + state = self.plugin.stage_job_pipeline_and_secrets(_pipeline(), 7, self.bundle) + + self.assertEqual(state["prior_cid"], "prior-cid") + self.assertEqual([event[0] for event in self.plugin.events[:3]], ["r1fs_add", "hset", "hset"]) + secret_write = self.plugin.events[1] + pipeline_write = self.plugin.events[2] + self.assertEqual(pipeline_write[1], DEEPLOY_JOBS_CSTORE_HKEY) + self.assertEqual(pipeline_write[4]["extra_peers"], ["dauth-a", "dauth-b"]) + self.assertTrue(pipeline_write[4]["include_default_peers"]) + self.assertTrue(pipeline_write[4]["include_configured_peers"]) + self.assertEqual(secret_write[1], DAUTH_JOB_SECRETS_CSTORE_HKEY) + self.assertEqual(secret_write[3]["pipeline_cid"], "staged-cid") + self.assertFalse(secret_write[4]["include_default_peers"]) + self.assertFalse(secret_write[4]["include_configured_peers"]) + + def test_commit_deletes_prior_cid_only_while_stage_is_current(self): + state = self.plugin.stage_job_pipeline_and_secrets(_pipeline(), 7, self.bundle) + self.assertTrue(self.plugin.commit_staged_job_pipeline_and_secrets(state)) + self.assertIn(("r1fs_delete", "prior-cid", { + "show_logs": False, + "raise_on_error": False, + }), self.plugin.events) + + def test_rollback_restores_prior_pointer_and_bundle_then_deletes_stage(self): + prior_bundle = copy.deepcopy(self.plugin.store[(DAUTH_JOB_SECRETS_CSTORE_HKEY, "7")]) + state = self.plugin.stage_job_pipeline_and_secrets(_pipeline(), 7, self.bundle) + + self.assertTrue(self.plugin.rollback_staged_job_pipeline_and_secrets(state)) + self.assertEqual(self.plugin.store[(DEEPLOY_JOBS_CSTORE_HKEY, "7")], "prior-cid") + self.assertEqual(self.plugin.store[(DAUTH_JOB_SECRETS_CSTORE_HKEY, "7")], prior_bundle) + self.assertEqual(self.plugin.events[-1][0:2], ("r1fs_delete", "staged-cid")) + + def test_rollback_does_not_overwrite_newer_concurrent_stage(self): + state = self.plugin.stage_job_pipeline_and_secrets(_pipeline(), 7, self.bundle) + newer_bundle = {"job_id": "7", "job_secrets": {"newer": True}} + self.plugin.store[(DEEPLOY_JOBS_CSTORE_HKEY, "7")] = "newer-cid" + self.plugin.store[(DAUTH_JOB_SECRETS_CSTORE_HKEY, "7")] = newer_bundle + event_count = len(self.plugin.events) + + self.assertFalse(self.plugin.rollback_staged_job_pipeline_and_secrets(state)) + self.assertEqual(self.plugin.store[(DEEPLOY_JOBS_CSTORE_HKEY, "7")], "newer-cid") + self.assertEqual(self.plugin.store[(DAUTH_JOB_SECRETS_CSTORE_HKEY, "7")], newer_bundle) + self.assertEqual(len(self.plugin.events), event_count) + + def test_rollback_does_not_overwrite_newer_bundle_with_same_cid(self): + state = self.plugin.stage_job_pipeline_and_secrets(_pipeline(), 7, self.bundle) + newer_bundle = {"job_id": "7", "job_secrets": {"newer": True}} + self.plugin.store[(DAUTH_JOB_SECRETS_CSTORE_HKEY, "7")] = newer_bundle + + self.assertFalse(self.plugin.rollback_staged_job_pipeline_and_secrets(state)) + self.assertEqual(self.plugin.store[(DEEPLOY_JOBS_CSTORE_HKEY, "7")], "staged-cid") + self.assertEqual(self.plugin.store[(DAUTH_JOB_SECRETS_CSTORE_HKEY, "7")], newer_bundle) + self.assertNotEqual(self.plugin.events[-1][0:2], ("r1fs_delete", "staged-cid")) + + def test_commit_requires_staged_bundle_to_still_be_current(self): + state = self.plugin.stage_job_pipeline_and_secrets(_pipeline(), 7, self.bundle) + self.plugin.store[(DAUTH_JOB_SECRETS_CSTORE_HKEY, "7")] = { + "job_id": "7", + "job_secrets": {"newer": True}, + } + + self.assertFalse(self.plugin.commit_staged_job_pipeline_and_secrets(state)) + self.assertNotIn(("r1fs_delete", "prior-cid", { + "show_logs": False, + "raise_on_error": False, + }), self.plugin.events) + + +if __name__ == "__main__": + unittest.main() diff --git a/extensions/business/deeploy/tests/test_update_requests.py b/extensions/business/deeploy/tests/test_update_requests.py index d08a3637..cd868074 100644 --- a/extensions/business/deeploy/tests/test_update_requests.py +++ b/extensions/business/deeploy/tests/test_update_requests.py @@ -96,6 +96,33 @@ def chainstore_hset(hkey, key, value): node_addr_to_eth_addr=lambda node: node, submit_node_update=lambda **kwargs: called.__setitem__("bc_update", called["bc_update"] + 1), ) + + def build_pipeline_config(**kwargs): + config = { + "NAME": kwargs["name"], + "TYPE": kwargs["stream_type"], + } + if kwargs.get("url") is not None: + config["URL"] = kwargs["url"] + if kwargs.get("plugins") is not None: + config["PLUGINS"] = copy.deepcopy(kwargs["plugins"]) + ignored = {"name", "stream_type", "url", "plugins"} + config.update({ + key.upper(): copy.deepcopy(value) + for key, value in kwargs.items() + if key not in ignored + }) + return config + + plugin.cmdapi_build_pipeline_config = build_pipeline_config + plugin._load_dauth_job_secret_bundle = lambda job_id: None + plugin.stage_job_pipeline_and_secrets = lambda pipeline, job_id, secret_bundle: { + "job_id": str(job_id), + "pipeline": copy.deepcopy(pipeline), + "secret_bundle": copy.deepcopy(secret_bundle), + } + plugin.commit_staged_job_pipeline_and_secrets = lambda state: True + plugin.rollback_staged_job_pipeline_and_secrets = lambda state: True plugin.delete_pipeline_from_nodes = lambda **kwargs: called.__setitem__("delete", called["delete"] + 1) def check_and_deploy_pipelines(**kwargs): @@ -3244,6 +3271,54 @@ def test_scale_up_prepare_refreshes_full_per_node_target_order(self): self.assertEqual(updated["CHAINSTORE_PEERS"], expected_nodes) self.assertEqual(updated["PER_NODE_TARGET_NODES"], expected_nodes) + def test_scale_up_prepare_preserves_offline_persisted_targets(self): + plugin = make_deeploy_plugin() + plugin.defaultdict = defaultdict + plugin.time = lambda: 1000 + base_pipeline = { + "app_id": "app-1", + "pipeline_type": "Void", + "url": "", + "pipeline_params": {}, + "deeploy_specs": { + DEEPLOY_KEYS.CURRENT_TARGET_NODES: ["online-node", "offline-node"], + DEEPLOY_KEYS.JOB_APP_TYPE: JOB_APP_TYPES.SERVICE, + }, + "plugins": [{ + "SIGNATURE": "CONTAINER_APP_RUNNER", + "INSTANCES": [{"INSTANCE_ID": "stale", "ENV": {}}], + }], + } + running_apps = { + "online-node": { + "app-1": { + "plugins": { + "CONTAINER_APP_RUNNER": [{ + "instance": "existing", + "instance_conf": {"CHAINSTORE_RESPONSE_KEY": "response"}, + }], + }, + }, + }, + } + + create_pipelines, update_pipelines, _ = plugin.prepare_create_update_pipelines( + base_pipeline=base_pipeline, + new_nodes=["new-node"], + update_nodes=["online-node"], + running_apps_for_job=running_apps, + ) + + expected = ["online-node", "offline-node", "new-node"] + self.assertEqual( + create_pipelines["new-node"]["deeploy_specs"][DEEPLOY_KEYS.CURRENT_TARGET_NODES], + expected, + ) + self.assertEqual( + update_pipelines["online-node"]["deeploy_specs"][DEEPLOY_KEYS.CURRENT_TARGET_NODES], + expected, + ) + if __name__ == "__main__": unittest.main() diff --git a/requirements.txt b/requirements.txt index 18686e4e..916946d4 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,5 +1,5 @@ kmonitor -ratio1>=3.5.41 +ratio1>=3.5.50 decentra-vision python-telegram-bot[rate-limiter] protobuf From c7ae865c36d339594a79ecaa114e8c18a453189f Mon Sep 17 00:00:00 2001 From: Alessandro Date: Tue, 28 Jul 2026 17:28:30 +0200 Subject: [PATCH 4/5] fix: preserve update behavior in dAuth staging --- .../business/deeploy/deeploy_manager_api.py | 9 ------- .../deeploy/tests/test_update_requests.py | 26 ++++++++++++++----- 2 files changed, 19 insertions(+), 16 deletions(-) diff --git a/extensions/business/deeploy/deeploy_manager_api.py b/extensions/business/deeploy/deeploy_manager_api.py index 82bc28ee..efb0709b 100644 --- a/extensions/business/deeploy/deeploy_manager_api.py +++ b/extensions/business/deeploy/deeploy_manager_api.py @@ -871,15 +871,6 @@ def _process_pipeline_request( inputs.target_nodes = deployment_targets inputs[DEEPLOY_KEYS.TARGET_NODES_COUNT] = len(deployment_targets) inputs.target_nodes_count = len(deployment_targets) - # Ensure plugin IDs are preserved for existing instances before any destructive action. - self._ensure_plugin_instance_ids( - inputs, - discovered_plugin_instances=discovered_plugin_instances, - owner=auth_result[DEEPLOY_KEYS.ESCROW_OWNER], - app_id=app_id, - job_id=job_id, - ) - if deeploy_specs_for_update is not None and not isinstance(deeploy_specs_for_update, dict): msg = ( f"{DEEPLOY_ERRORS.REQUEST3}. Unexpected 'deeploy_specs' payload type " diff --git a/extensions/business/deeploy/tests/test_update_requests.py b/extensions/business/deeploy/tests/test_update_requests.py index cd868074..a668674e 100644 --- a/extensions/business/deeploy/tests/test_update_requests.py +++ b/extensions/business/deeploy/tests/test_update_requests.py @@ -91,7 +91,14 @@ def chainstore_hset(hkey, key, value): plugin.chainstore_hset = chainstore_hset - called = {"delete": 0, "deploy": 0, "deploy_kwargs": None, "queued": 0, "bc_update": 0} + called = { + "delete": 0, + "deploy": 0, + "deploy_kwargs": None, + "queued": 0, + "bc_update": 0, + "stage": 0, + } plugin.bc = types.SimpleNamespace( node_addr_to_eth_addr=lambda node: node, submit_node_update=lambda **kwargs: called.__setitem__("bc_update", called["bc_update"] + 1), @@ -116,11 +123,15 @@ def build_pipeline_config(**kwargs): plugin.cmdapi_build_pipeline_config = build_pipeline_config plugin._load_dauth_job_secret_bundle = lambda job_id: None - plugin.stage_job_pipeline_and_secrets = lambda pipeline, job_id, secret_bundle: { - "job_id": str(job_id), - "pipeline": copy.deepcopy(pipeline), - "secret_bundle": copy.deepcopy(secret_bundle), - } + def stage_job_pipeline_and_secrets(pipeline, job_id, secret_bundle): + called["stage"] += 1 + return { + "job_id": str(job_id), + "pipeline": copy.deepcopy(pipeline), + "secret_bundle": copy.deepcopy(secret_bundle), + } + + plugin.stage_job_pipeline_and_secrets = stage_job_pipeline_and_secrets plugin.commit_staged_job_pipeline_and_secrets = lambda state: True plugin.rollback_staged_job_pipeline_and_secrets = lambda state: True plugin.delete_pipeline_from_nodes = lambda **kwargs: called.__setitem__("delete", called["delete"] + 1) @@ -1709,7 +1720,8 @@ def test_process_update_uses_persisted_pipeline_when_all_old_nodes_are_offline(s self.assertEqual(response[DEEPLOY_KEYS.STATUS], DEEPLOY_STATUS.COMMAND_DELIVERED) self.assertEqual(called["delete"], 0) self.assertEqual(called["deploy"], 1) - self.assertEqual(called["queued"], 1) + self.assertEqual(called["stage"], 1) + self.assertEqual(called["queued"], 0) self.assertEqual(called["bc_update"], 1) self.assertEqual(called["deploy_kwargs"]["new_nodes"], ["new-node-1"]) redeploy_plugins = called["deploy_kwargs"]["inputs"][DEEPLOY_KEYS.PLUGINS] From e70de646a2736a76ce242f2a94d65105d07a53a6 Mon Sep 17 00:00:00 2001 From: Alessandro Date: Tue, 4 Aug 2026 18:07:35 +0200 Subject: [PATCH 5/5] fix --- Dockerfile_devnet | 2 +- Dockerfile_mainnet | 2 +- Dockerfile_testnet | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/Dockerfile_devnet b/Dockerfile_devnet index ce3e809d..0c7d9775 100644 --- a/Dockerfile_devnet +++ b/Dockerfile_devnet @@ -75,7 +75,7 @@ ENV EE_DEBUG_R1FS=true # ENV EE_DEVICE=cuda:0 RUN uv pip install --system --no-cache -r requirements.txt -RUN uv pip install --system --no-cache --no-deps 'naeural-core>=7.7.318' +RUN uv pip install --system --no-cache --no-deps naeural-core #RUN pip install --no-cache-dir -r requirements.txt #RUN pip install --no-cache-dir --no-deps naeural-core diff --git a/Dockerfile_mainnet b/Dockerfile_mainnet index 8d076f59..7c001b5e 100644 --- a/Dockerfile_mainnet +++ b/Dockerfile_mainnet @@ -76,7 +76,7 @@ ENV EE_DEBUG_R1FS=false # ENV EE_DEVICE cuda:0 RUN uv pip install --system --no-cache -r requirements.txt -RUN uv pip install --system --no-cache --no-deps 'naeural-core>=7.7.318' +RUN uv pip install --system --no-cache --no-deps naeural-core #RUN pip install --no-cache-dir -r requirements.txt #RUN pip install --no-cache-dir --no-deps naeural-core diff --git a/Dockerfile_testnet b/Dockerfile_testnet index 6390d6fc..2f9e077d 100644 --- a/Dockerfile_testnet +++ b/Dockerfile_testnet @@ -75,7 +75,7 @@ ENV EE_DEBUG_R1FS=true # ENV EE_DEVICE=cuda:0 RUN uv pip install --system --no-cache -r requirements.txt -RUN uv pip install --system --no-cache --no-deps 'naeural-core>=7.7.318' +RUN uv pip install --system --no-cache --no-deps naeural-core #RUN pip install --no-cache-dir -r requirements.txt #RUN pip install --no-cache-dir --no-deps naeural-core