Hide Deeploy mandatory secrets in dAuth

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)

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 []

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(

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"]}],
       }
@@ -136,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": {
@@ -152,7 +173,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 +186,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)

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(