Stage and resolve worker dAuth secrets

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`

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
 

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
 

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
 

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,

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)

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()

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,20 +221,22 @@ 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):
     result = self.chainstore_hset(
       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,14 +290,104 @@ 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)
     if not self._is_protocol_oracle_eth(requester_eth):
       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,

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,
+  }