Add dAuth job secret endpoints (#458)

* feat: add dauth job secret endpoints

* fix: require job_secrets bundle shape

* fix: simplification

* fix: protect dAuth secret requests from replay

* fix

* feat: sync dAuth secrets through registry peers

* fix: refresh dAuth registry authorization

* refactor: simplify dAuth registry refresh

* chore: inc ver

Author: aledefra

AGENTS.md

@@ -695,3 +695,30 @@ Entry format:
 - Details: `ThHfModelBase` keeps Transformers/PT as the default GPU and fallback path, but CPU-only `HF_RUNTIME=auto` now loads `artifact_manifest.json`, selects a declared ONNX Runtime artifact, downloads only safe allow-patterns, loads schema and contract decoder from HF artifacts, and exposes the decoded artifact contract through the existing text-classifier flow. Business API response shaping now passes through generic model/runtime metadata emitted by serving.
 - 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-20260723-001`
+- Timestamp: `2026-07-23T13:45:20Z`
+- Type: `change`
+- Summary: dAuth job-secret requests now require signed 120-second timestamp nonces, and GET responses encrypt secret bundles to the authorized runner.
+- Criticality: Security protocol change preventing indefinite signed-request/response replay and removing plaintext job secrets from HTTP responses.
+- Details: `/add_secrets` and `/get_secrets` validate signed hex-millisecond timestamp nonces and echo them in successful signed responses. `/get_secrets` encrypts the serialized bundle to the signed requester address; clients must verify the response signer and echoed nonce before decrypting.
+- Verification: `python -m unittest discover -s extensions/business/dauth -p 'test_*.py'`; cross-repo SDK dAuth client tests.
+- Links: `extensions/business/dauth/dauth_mixin.py`, `extensions/business/dauth/dauth_manager.py`
+
+- ID: `ML-20260731-001`
+- Timestamp: `2026-07-31T16:48:11Z`
+- Type: `change`
+- Summary: dAuth job-secret ChainStore writes and minute syncs now target only startup-cached dAuth registry peers.
+- Criticality: Secret-replication boundary and recovery behavior across every dAuth server.
+- Details: The dAuth manager reads registry ETH addresses once at startup, keeps local service eligibility fixed until restart, and refreshes only ETH-to-internal mappings from local NetMon state. `DAUTH_JOB_SECRETS` writes and 60-second hsync calls disable default/configured ChainStore peers. Known deferred risks: generic ChainStore does not authorize inbound operations by hash namespace, and first-response hsync has no freshness arbitration; production hardening requires an inbound ACL or dedicated authenticated replication protocol plus version-aware merges.
+- Verification: `python3 -m unittest discover -s extensions/business/dauth -p 'test_*.py'`; `python3 -m py_compile extensions/business/dauth/dauth_registry.py extensions/business/dauth/dauth_manager.py extensions/business/dauth/dauth_mixin.py extensions/business/dauth/test_dauth_registry_gating.py extensions/business/dauth/test_dauth_secret_routing.py`; `git diff --check`
+- Links: `extensions/business/dauth/dauth_registry.py`, `extensions/business/dauth/dauth_manager.py`, `extensions/business/dauth/dauth_mixin.py`
+
+- ID: `ML-20260803-001`
+- Timestamp: `2026-08-03T16:09:26Z`
+- Type: `change`
+- Summary: dAuth server eligibility and secret-replication peers now refresh from the on-chain registry every hour; secret hsync runs every 10 minutes.
+- Criticality: Authorization revocation and secret-replication routing across every dAuth server.
+- Details: Lifecycle pause/resume predicates perform the rate-limited registry refresh without adding RPC calls to endpoint request paths. Successful reads remain cached for one hour; failed reads clear cached peers, fail closed, and retry after one minute. Registry reads are synchronous and rely on the SDK Web3 provider to return or time out. A removed local node causes the web app to pause and become unready; readiness returns only after a resumed Uvicorn process reports startup. Remaining dAuth nodes replace their cached peer set on their next hourly refresh. The inbound namespace authorization and version-aware hsync limitations from `ML-20260731-001` remain open.
+- Verification: `python3 -m unittest discover -s extensions/business/dauth -p 'test_*.py'`; `python3 -m py_compile extensions/business/dauth/dauth_registry.py extensions/business/dauth/dauth_manager.py extensions/business/dauth/dauth_mixin.py extensions/business/dauth/test_dauth_registry_gating.py extensions/business/dauth/test_dauth_secret_routing.py`; `git diff --check`
+- Links: `extensions/business/dauth/dauth_manager.py`, `extensions/business/dauth/test_dauth_registry_gating.py`

extensions/business/dauth/dauth_manager.py

@@ -23,7 +23,14 @@
 from extensions.business.mixins.node_tags_mixin import _NodeTagsMixin
 from naeural_core.business.default.web_app.supervisor_fast_api_web_app import SupervisorFastApiWebApp as BasePlugin
 from extensions.business.mixins.request_tracking_mixin import _RequestTrackingMixin
-from extensions.business.dauth.dauth_mixin import _DauthMixin
+from extensions.business.dauth.dauth_mixin import (
+  DAUTH_JOB_SECRETS_CSTORE_HKEY,
+  _DauthMixin,
+)
+from extensions.business.dauth.dauth_registry import (
+  dauth_registry_write_kwargs,
+  load_dauth_registry_snapshot,
+)
 
 __VER__ = '0.3.0'
 
@@ -42,6 +49,10 @@
   'REQUESTS_MAX_RECORDS': 2,
   'REQUESTS_LOG_INTERVAL': 5 * 60,
 
+  'DAUTH_JOB_SECRETS_HSYNC_INTERVAL': 10 * 60,
+  'DAUTH_REGISTRY_REFRESH_INTERVAL': 60 * 60,
+  'DAUTH_REGISTRY_REFRESH_RETRY_INTERVAL': 60,
+
   'SUPRESS_LOGS_AFTER_INTERVAL' : 300,
 
   # required ENV keys are defined in plugin template and should be added here  
@@ -105,6 +116,11 @@ def __init__(self, **kwargs):
     super(DauthManagerPlugin, self).__init__(**kwargs)
     self._dauth_server_enabled = None
     self._dauth_server_enabled_message = None
+    self._dauth_registry_eth_oracles = None
+    self._dauth_registry_internal_peers = None
+    self._last_dauth_registry_refresh = None
+    self._dauth_registry_refresh_failed = False
+    self._last_dauth_job_secrets_hsync = None
     self._dauth_web_app_initialized = False
     self._dauth_pause_teardown_succeeded = True
     return
@@ -117,6 +133,8 @@ def on_init(self):
     self._dauth_web_app_initialized = True
     if not self._is_dauth_server_enabled():
       self.on_pause()
+    else:
+      self._maybe_hsync_dauth_job_secrets()
     # endif
     my_address = self.bc.address
     my_eth_address = self.bc.eth_address
@@ -132,20 +150,52 @@ def _check_dauth_server_enabled_on_start(self):
       return self._dauth_server_enabled
     # endif
 
+    return self._refresh_dauth_registry(force=True)
+
+  def _refresh_dauth_registry(self, force=False):
+    now = self.time()
+    last_refresh = getattr(self, "_last_dauth_registry_refresh", None)
+    refresh_interval = (
+      self.cfg_dauth_registry_refresh_retry_interval
+      if getattr(self, "_dauth_registry_refresh_failed", False)
+      else self.cfg_dauth_registry_refresh_interval
+    )
+    if (
+      not force
+      and last_refresh is not None
+      and now - last_refresh < refresh_interval
+    ):
+      return self._is_dauth_server_enabled()
+    # endif
+
+    self._last_dauth_registry_refresh = now
+    previous_enabled = getattr(self, "_dauth_server_enabled", None)
+    previous_eth_oracles = getattr(self, "_dauth_registry_eth_oracles", None)
+
     error = None
     try:
-      enabled = self.bc.is_dauth_oracle() is True
+      peers, eth_oracles = load_dauth_registry_snapshot(self)
+      enabled = self.bc.eth_address.lower() in [
+        address.lower() for address in eth_oracles
+      ]
     except Exception as e:
       enabled = False
       error = str(e)
     # end try
 
     message = None if enabled else error or "current node is not registered as a dAuth oracle"
+    self._dauth_registry_eth_oracles = eth_oracles if enabled else None
+    self._dauth_registry_internal_peers = peers if enabled else None
+    self._dauth_registry_refresh_failed = error is not None
     self._dauth_server_enabled = enabled
     self._dauth_server_enabled_message = message
-    if enabled:
-      self.P(f"{self.__class__.__name__} dAuth registry gate is enabled")
-    else:
+    registry_changed = previous_eth_oracles != self._dauth_registry_eth_oracles
+    if enabled and (previous_enabled is not True or registry_changed):
+      self.P(
+        f"{self.__class__.__name__} dAuth registry gate is enabled "
+        f"with {len(eth_oracles)} registered oracle(s)"
+      )
+    elif not enabled and (previous_enabled is not False or error is not None):
       self.P(
         f"{self.__class__.__name__} dAuth registry gate is disabled. "
         f"(cause: {message})",
@@ -159,16 +209,19 @@ def _is_dauth_server_enabled(self):
     return getattr(self, "_dauth_server_enabled", None) is True
 
   def should_pause(self):
+    self._refresh_dauth_registry()
     return not self._is_dauth_server_enabled()
 
   def should_resume(self):
+    self._refresh_dauth_registry()
     return self._is_dauth_server_enabled()
 
   def on_pause(self):
     if not getattr(self, "_dauth_web_app_initialized", False):
       return
     # endif
 
+    self.set_plugin_ready(False)
     self._dauth_pause_teardown_succeeded = False
     self._stop_request_monitor.set()
     if self._request_monitor_thread is not None:
@@ -226,6 +279,13 @@ def on_resume(self):
     self._stop_request_monitor.clear()
     self._start_request_monitor_thread()
     return
+
+  def on_log_handler(self, text, key=None):
+    super(DauthManagerPlugin, self).on_log_handler(text, key=key)
+    if self._is_dauth_server_enabled() and "Uvicorn running on " in text:
+      self.set_plugin_ready(True)
+    # endif
+    return
 
 
   def on_request(self, request):
@@ -237,11 +297,34 @@ def on_response(self, method, response):
     return
 
   def process(self):
+    self._maybe_hsync_dauth_job_secrets()
     # TODO: this will be re-enabled in the future.
     if False:
       self._maybe_log_and_save_tracked_requests()
     return
 
+  def _maybe_hsync_dauth_job_secrets(self):
+    if not self._is_dauth_server_enabled():
+      return None
+
+    now = self.time()
+    last_sync = getattr(self, "_last_dauth_job_secrets_hsync", None)
+    if (
+      last_sync is not None
+      and now - last_sync < self.cfg_dauth_job_secrets_hsync_interval
+    ):
+      return None
+
+    self._last_dauth_job_secrets_hsync = now
+    try:
+      return self.chainstore_hsync(
+        hkey=DAUTH_JOB_SECRETS_CSTORE_HKEY,
+        **dauth_registry_write_kwargs(self),
+      )
+    except Exception as exc:
+      self.P(f"Could not sync dAuth job secrets: {exc}", color="y")
+    return None
+
   def __get_current_epoch(self):
     """
     Get the current epoch of the node.
@@ -343,3 +426,65 @@ def get_auth_data(self, body: dict):
       **data
     })
     return response
+
+  @BasePlugin.endpoint(method="post")
+  # /add_secrets
+  def add_secrets(self, body: dict):
+    """
+    Store a full job secret bundle from a protocol oracle.
+
+    The signed request must include a hex-millisecond timestamp nonce no older
+    than 120 seconds.
+    """
+    request_nonce = body.get("nonce") if isinstance(body, dict) else None
+    if not self._is_dauth_server_enabled():
+      response = self.__get_response({
+        'error': 'dAuth server is not registered as a dAuth oracle',
+        'nonce': request_nonce,
+      })
+      return response
+
+    try:
+      data = self.process_dauth_add_secrets_request(body)
+    except Exception as e:
+      self.P("Error processing add_secrets request: {}".format(e), color='r')
+      data = {
+        'error' : str(e)
+      }
+
+    response = self.__get_response({
+      'nonce': request_nonce,
+      **data
+    })
+    return response
+
+  @BasePlugin.endpoint(method="post")
+  # /get_secrets
+  def get_secrets(self, body: dict):
+    """
+    Return an encrypted job secret bundle to a current R1FS job runner.
+
+    The signed request must include a hex-millisecond timestamp nonce no older
+    than 120 seconds. The signed response echoes that nonce.
+    """
+    request_nonce = body.get("nonce") if isinstance(body, dict) else None
+    if not self._is_dauth_server_enabled():
+      response = self.__get_response({
+        'error': 'dAuth server is not registered as a dAuth oracle',
+        'nonce': request_nonce,
+      })
+      return response
+
+    try:
+      data = self.process_dauth_get_secret_request(body)
+    except Exception as e:
+      self.P("Error processing get_secrets request: {}".format(e), color='r')
+      data = {
+        'error' : str(e)
+      }
+
+    response = self.__get_response({
+      'nonce': request_nonce,
+      **data
+    })
+    return response