diff --git a/AGENTS.md b/AGENTS.md index 1289f3a4..948863c1 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 c7c7da0e..6f3aa08f 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 + DAUTH_JOB_SECRETS_CSTORE_HKEY = "DAUTH_JOB_SECRETS" DEEPLOY_JOBS_CSTORE_HKEY = "DEEPLOY_DEPLOYED_JOBS" @@ -215,13 +221,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): @@ -229,6 +236,7 @@ def _save_dauth_job_secret_bundle(self, job_id, secret_bundle): hkey=DAUTH_JOB_SECRETS_CSTORE_HKEY, key=job_id, value=secret_bundle, + **dauth_registry_write_kwargs(self), ) if not result: raise ValueError(f"Failed to store dAuth secrets for job {job_id}.") @@ -240,14 +248,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): @@ -278,6 +290,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) @@ -285,7 +375,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 { @@ -299,12 +401,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 new file mode 100644 index 00000000..72972631 --- /dev/null +++ b/extensions/business/dauth/dauth_registry.py @@ -0,0 +1,38 @@ +"""Shared routing helpers for dAuth registry CStore writes.""" + + +DAUTH_SECRET_PIPELINE_CID_KEY = "pipeline_cid" + + +def get_dauth_registry_internal_peers(plugin): + """Return the currently registered dAuth oracle internal addresses.""" + 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 + + +def dauth_registry_write_kwargs(plugin, peers=None): + """Route a CStore write exclusively to dAuth registry peers.""" + if peers is None: + peers = get_dauth_registry_internal_peers(plugin) + return { + "extra_peers": list(peers), + "include_default_peers": False, + "include_configured_peers": False, + } + + +def pipeline_registry_write_kwargs(plugin, peers=None): + """Add dAuth peers without disabling normal pipeline metadata peers.""" + if peers is None: + peers = get_dauth_registry_internal_peers(plugin) + return { + "extra_peers": list(peers), + "include_default_peers": True, + "include_configured_peers": True, + } diff --git a/extensions/business/dauth/test_dauth_registry_gating.py b/extensions/business/dauth/test_dauth_registry_gating.py index 92e543ec..89546d96 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, @@ -199,6 +201,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] @@ -273,7 +278,7 @@ def _make_dauth_harness(*, dauth_oracle=True, protocol_oracles=None, valid_signa plugin.fetch_node_tags = lambda node_address_eth=None: {} plugin.P = lambda *args, **kwargs: None plugin.Pd = lambda *args, **kwargs: None - plugin.chainstore_hset = lambda hkey, key, value: plugin._chainstore.__setitem__( + plugin.chainstore_hset = lambda hkey, key, value, **kwargs: plugin._chainstore.__setitem__( (hkey, str(key)), deepcopy(value), ) or True @@ -392,6 +397,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) @@ -403,6 +409,7 @@ def test_add_secrets_allows_protocol_oracle_and_overwrites_bundle(self): { "job_id": "7", "job_secrets": body["job_secrets"], + "pipeline_cid": "cid-7", }, ) @@ -470,6 +477,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": [{ @@ -511,6 +519,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 new file mode 100644 index 00000000..87e609a7 --- /dev/null +++ b/extensions/business/dauth/test_dauth_secret_routing.py @@ -0,0 +1,46 @@ +import copy +import unittest + +from extensions.business.dauth.dauth_mixin import _DauthMixin + + +class _BCStub: + def get_dauth_oracles(self): + return ["dauth-a", "dauth-b"], ["a", "b"] + + def get_eth_oracles(self): + return ["0xOracleA", "0xOracleB"] + + +class _DauthStub(_DauthMixin): + def __init__(self): + self.bc = _BCStub() + self.writes = [] + self.deepcopy = copy.deepcopy + + def chainstore_hset(self, **kwargs): + self.writes.append(kwargs) + return True + + +class DauthSecretRoutingTests(unittest.TestCase): + def test_add_secrets_storage_targets_only_dauth_registry_peers(self): + plugin = _DauthStub() + + plugin._save_dauth_job_secret_bundle("7", {"job_id": "7", "job_secrets": {}}) + + write = plugin.writes[0] + self.assertEqual(write["extra_peers"], ["dauth-a", "dauth-b"]) + self.assertFalse(write["include_default_peers"]) + self.assertFalse(write["include_configured_peers"]) + + def test_all_protocol_oracles_remain_authorized_writers(self): + plugin = _DauthStub() + + 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..efb0709b 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,12 +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) - 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') - 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 " @@ -960,12 +955,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 +963,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 +998,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 +1047,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 +1073,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 +1116,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 +1136,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 +1153,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 +1210,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 +1273,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 +1317,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 +1671,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 +1706,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 +1721,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 +1756,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..a668674e 100644 --- a/extensions/business/deeploy/tests/test_update_requests.py +++ b/extensions/business/deeploy/tests/test_update_requests.py @@ -91,11 +91,49 @@ 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), ) + + 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 + 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) def check_and_deploy_pipelines(**kwargs): @@ -1682,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] @@ -3244,6 +3283,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