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",
@@ -3697,25 +3716,136 @@ def _redact_per_node_config_for_log(self, plugins):
     deepcopy = getattr(self, "deepcopy", copy.deepcopy)
     redacted = deepcopy(plugins)
 
-    def redact(value):
+    def redact(value, path):
       if isinstance(value, dict):
         for key, item in list(value.items()):
+          item_path = path + [key]
           if key in PER_NODE_CONFIG_KEYS and item:
             value[key] = "***"
+          elif self._matches_deeploy_dauth_secret_path(item_path) and item:
+            value[key] = "***"
           elif any(
             part in re.sub(r"[^A-Z0-9]", "", str(key).upper())
             for part in SENSITIVE_LOG_KEY_PARTS
           ):
             value[key] = "***" if item else item
           else:
-            redact(item)
+            redact(item, item_path)
       elif isinstance(value, list):
-        for item in value:
-          redact(item)
+        for idx, item in enumerate(value):
+          redact(item, path + [idx])
 
-    redact(redacted)
+    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

@@ -2,6 +2,7 @@
 from collections import defaultdict
 import sys
 import types
+from unittest.mock import patch
 
 
 for _mod_name in ("torch", "torch.nn", "torch.nn.functional"):
@@ -16,6 +17,11 @@
   DEEPLOY_PLUGIN_DATA,
   JOB_APP_TYPES,
 )
+from extensions.business.deeploy import deeploy_mixin
+from extensions.business.deeploy.deeploy_mixin import (
+  DEEPLOY_DAUTH_SECRET_PATH_SUFFIXES,
+  DEEPLOY_DAUTH_SECRET_PLACEHOLDER,
+)
 from extensions.business.deeploy.tests.support import make_deeploy_plugin, make_inputs, make_plugin_entry
 
 
@@ -257,6 +263,99 @@ 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_log_redaction_covers_every_dauth_secret_path_without_key_heuristics(self):
+    plugin = make_deeploy_plugin()
+
+    with patch.object(deeploy_mixin, "SENSITIVE_LOG_KEY_PARTS", ()):
+      for suffix in DEEPLOY_DAUTH_SECRET_PATH_SUFFIXES:
+        with self.subTest(path=suffix):
+          instance = {}
+          payload = {"PLUGINS": [{"INSTANCES": [instance]}]}
+          current = instance
+          for part in suffix[:-1]:
+            key = "wildcard" if part == "*" else part
+            current[key] = {}
+            current = current[key]
+          leaf = "wildcard" if suffix[-1] == "*" else suffix[-1]
+          current[leaf] = "must-not-be-logged"
+
+          redacted = plugin._redact_per_node_config_for_log(payload)
+
+          self.assertNotIn("must-not-be-logged", str(redacted))
+          self.assertIn("***", str(redacted))
+
+      list_shaped_ports = {
+        "EXPOSED_PORTS": [{
+          "token": "list-port-token",
+          "tunnel": {"token": "list-tunnel-token"},
+        }],
+      }
+      redacted_ports = plugin._redact_per_node_config_for_log(list_shaped_ports)
+      self.assertNotIn("list-port-token", str(redacted_ports))
+      self.assertNotIn("list-tunnel-token", str(redacted_ports))
+
+  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(