From 47a39ee241fc7dccd5f03103045774ca992a4e32 Mon Sep 17 00:00:00 2001 From: Alessandro Date: Wed, 1 Jul 2026 17:12:21 +0200 Subject: [PATCH 1/9] feat: add dauth job secret endpoints --- extensions/business/dauth/dauth_manager.py | 51 +++++ extensions/business/dauth/dauth_mixin.py | 146 +++++++++++++- .../dauth/test_dauth_registry_gating.py | 184 +++++++++++++++++- 3 files changed, 376 insertions(+), 5 deletions(-) diff --git a/extensions/business/dauth/dauth_manager.py b/extensions/business/dauth/dauth_manager.py index f60f9010..9d98f90c 100644 --- a/extensions/business/dauth/dauth_manager.py +++ b/extensions/business/dauth/dauth_manager.py @@ -343,3 +343,54 @@ def get_auth_data(self, body: dict): **data }) return response + + @BasePlugin.endpoint(method="post") + # /add_secrets + def add_secrets(self, body: dict): + """ + Store the full dAuth secret bundle for a job. Only protocol oracles can write. + """ + if not self._is_dauth_server_enabled(): + response = self.__get_response({ + 'error': 'dAuth server is not registered as a dAuth oracle' + }) + 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({ + **data + }) + return response + + @BasePlugin.endpoint(method="post") + # /get_secrets + def get_secrets(self, body: dict): + """ + Return the full dAuth secret bundle for a job. Only nodes currently running the + job in the R1FS-stored pipeline can read. + """ + if not self._is_dauth_server_enabled(): + response = self.__get_response({ + 'error': 'dAuth server is not registered as a dAuth oracle' + }) + 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({ + **data + }) + return response diff --git a/extensions/business/dauth/dauth_mixin.py b/extensions/business/dauth/dauth_mixin.py index b98bcb5d..237c971f 100644 --- a/extensions/business/dauth/dauth_mixin.py +++ b/extensions/business/dauth/dauth_mixin.py @@ -15,6 +15,10 @@ """ +DAUTH_JOB_SECRETS_CSTORE_HKEY = "DAUTH_JOB_SECRETS" +DEEPLOY_JOBS_CSTORE_HKEY = "DEEPLOY_DEPLOYED_JOBS" + + def version_to_int(version): """ Convert a version string to an integer. @@ -159,8 +163,146 @@ def check_if_node_allowed( node_address_eth, self.evm_network, e ) return result, msg - - + + def _verify_signed_dauth_body(self, body): + if not isinstance(body, dict): + raise ValueError("Invalid request body.") + + bcct = self.const.BASE_CT.BCctbase + requester = body.get(bcct.SENDER) + requester_send_eth = body.get(bcct.ETH_SENDER) + if not requester: + raise ValueError("No sender address in request.") + if not requester_send_eth: + raise ValueError("No sender ETH address in request.") + + requester_eth = self.bc.node_address_to_eth_address(requester) + if requester_eth.lower() != requester_send_eth.lower(): + raise ValueError("Sender ETH address and recovered ETH address do not match.") + + verify_data = self.bc.verify(body, return_full_info=True) + if not verify_data.valid: + raise ValueError("Invalid request signature: {}".format(verify_data.message)) + + return requester, requester_eth + + def _is_protocol_oracle_eth(self, node_address_eth): + eth_oracles = self.bc.get_eth_oracles() + if len(eth_oracles) == 0: + raise ValueError("No oracles found - this is a critical issue!") + return node_address_eth.lower() in [addr.lower() for addr in eth_oracles] + + def _normalize_dauth_job_id(self, job_id): + if job_id in [None, ""]: + raise ValueError("Job ID is required.") + return str(job_id) + + def _build_secret_bundle_from_request(self, body, job_id): + secret_bundle = body.get("secret_bundle") + if secret_bundle is None: + plugin_secrets = body.get("plugin_secrets") + if plugin_secrets is None: + raise ValueError("Secret bundle is required.") + secret_bundle = { + "job_id": job_id, + "plugin_secrets": plugin_secrets, + } + if not isinstance(secret_bundle, dict): + raise ValueError("Secret bundle must be a dictionary.") + + bundle_job_id = secret_bundle.get("job_id", job_id) + if str(bundle_job_id) != job_id: + raise ValueError("Secret bundle job_id does not match request job_id.") + + secret_bundle = self.deepcopy(secret_bundle) + secret_bundle["job_id"] = job_id + return secret_bundle + + 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, + ) + if not result: + raise ValueError(f"Failed to store dAuth secrets for job {job_id}.") + return result + + def _load_dauth_job_secret_bundle(self, job_id): + return self.chainstore_hget( + hkey=DAUTH_JOB_SECRETS_CSTORE_HKEY, + key=job_id, + ) + + def _load_dauth_job_pipeline(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) + + def _pipeline_runner_nodes(self, pipeline): + if not isinstance(pipeline, dict): + return [] + specs = pipeline.get("DEEPLOY_SPECS") or pipeline.get("deeploy_specs") or {} + if not isinstance(specs, dict): + return [] + nodes = specs.get("current_target_nodes") or specs.get("CURRENT_TARGET_NODES") or [] + if isinstance(nodes, str): + nodes = [nodes] + if not isinstance(nodes, list): + return [] + return [node for node in nodes if isinstance(node, str) and len(node) > 0] + + def _normalize_node_address_for_compare(self, node_address): + try: + return self.bc.maybe_add_prefix(node_address) + except Exception: + return node_address + + def _is_node_running_dauth_job(self, job_id, node_address): + pipeline = self._load_dauth_job_pipeline(job_id) + runner_nodes = self._pipeline_runner_nodes(pipeline) + requester = self._normalize_node_address_for_compare(node_address) + runner_nodes = [ + self._normalize_node_address_for_compare(node) + for node in runner_nodes + ] + return requester in runner_nodes + + def process_dauth_add_secrets_request(self, body): + requester, requester_eth = self._verify_signed_dauth_body(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) + 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 { + "status": "success", + "job_id": job_id, + } + + def process_dauth_get_secret_request(self, body): + requester, _ = self._verify_signed_dauth_body(body) + job_id = self._normalize_dauth_job_id(body.get("job_id")) + + if not self._is_node_running_dauth_job(job_id, requester): + 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}.") + + return { + "status": "success", + "job_id": job_id, + "secret_bundle": secret_bundle, + } + def chainstore_store_dauth_request( self, node_address : str, diff --git a/extensions/business/dauth/test_dauth_registry_gating.py b/extensions/business/dauth/test_dauth_registry_gating.py index 23f5740a..fb98b783 100644 --- a/extensions/business/dauth/test_dauth_registry_gating.py +++ b/extensions/business/dauth/test_dauth_registry_gating.py @@ -2,9 +2,14 @@ import queue import threading import unittest +from copy import deepcopy from pathlib import Path -from extensions.business.dauth.dauth_mixin import _DauthMixin +from extensions.business.dauth.dauth_mixin import ( + DAUTH_JOB_SECRETS_CSTORE_HKEY, + DEEPLOY_JOBS_CSTORE_HKEY, + _DauthMixin, +) ROOT = Path(__file__).resolve().parents[3] @@ -138,7 +143,13 @@ class _FakeDauthConst: DAUTH_WHITELIST = "DAUTH_WHITELIST" +class _FakeBCBaseConst: + SENDER = "EE_SENDER" + ETH_SENDER = "EE_ETH_SENDER" + + class _FakeBaseConst: + BCctbase = _FakeBCBaseConst dAuth = _FakeDauthConst @@ -155,9 +166,15 @@ class _FakeConst: class _FakeBC: - def __init__(self, *, dauth_oracle=True, protocol_oracles=None): + def __init__(self, *, dauth_oracle=True, protocol_oracles=None, valid_signature=True): self.dauth_oracle = dauth_oracle self.protocol_oracles = protocol_oracles or ["node-oracle"] + self.valid_signature = valid_signature + self.node_eth = { + "node-oracle": "0xORACLE", + "node-runner": "0xRUNNER", + "node-other": "0xOTHER", + } def get_oracles(self, include_eth_addrs=False): names = ["Oracle"] * len(self.protocol_oracles) @@ -174,18 +191,52 @@ def is_dauth_oracle(self, node_address_eth=None): # pylint: disable=unused-argu raise self.dauth_oracle return self.dauth_oracle + def get_eth_oracles(self): + return [self.node_eth.get(node, "0xORACLE") for node in self.protocol_oracles] + + def node_address_to_eth_address(self, node_address): + return self.node_eth[node_address] + + def verify(self, body, return_full_info=False): # pylint: disable=unused-argument + class _VerifyData: + pass + + data = _VerifyData() + data.valid = self.valid_signature + data.message = "ok" if self.valid_signature else "bad signature" + return data + + def maybe_add_prefix(self, node_address): + if node_address.startswith("0xai_"): + return node_address + return "0xai_" + node_address + + +class _FakeR1FS: + + def __init__(self, data): + self.data = data + + def get_json(self, cid, show_logs=False): # pylint: disable=unused-argument + return self.data[cid] + class _DauthHarness(_DauthMixin): pass -def _make_dauth_harness(*, dauth_oracle=True, protocol_oracles=None): +def _make_dauth_harness(*, dauth_oracle=True, protocol_oracles=None, valid_signature=True): plugin = _DauthHarness() plugin.const = _FakeConst plugin.bc = _FakeBC( dauth_oracle=dauth_oracle, protocol_oracles=protocol_oracles, + valid_signature=valid_signature, ) + plugin.deepcopy = deepcopy + plugin._chainstore = {} + plugin._r1fs_data = {} + plugin.r1fs = _FakeR1FS(plugin._r1fs_data) plugin.evm_network = "devnet" plugin.cfg_auth_env_keys = [] plugin.cfg_auth_node_env_keys = [] @@ -211,6 +262,11 @@ def _make_dauth_harness(*, dauth_oracle=True, protocol_oracles=None): 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__( + (hkey, str(key)), + deepcopy(value), + ) or True + plugin.chainstore_hget = lambda hkey, key: plugin._chainstore.get((hkey, str(key))) return plugin @@ -276,6 +332,128 @@ def test_dauth_token_fails_closed_when_dauth_registry_check_fails(self): self.assertEqual(data["EE_CLOUDFLARE_TOKEN_DEEPLOY_MANAGER"], "deeploy-secret") +class DauthJobSecretEndpointTests(unittest.TestCase): + + def test_add_secrets_allows_protocol_oracle_and_overwrites_bundle(self): + plugin = _make_dauth_harness(protocol_oracles=["node-oracle"]) + plugin._chainstore[(DAUTH_JOB_SECRETS_CSTORE_HKEY, "7")] = { + "job_id": "7", + "old": True, + } + body = { + "EE_SENDER": "node-oracle", + "EE_ETH_SENDER": "0xORACLE", + "job_id": 7, + "plugin_secrets": { + "plugins": { + "CONTAINER_APP_RUNNER": [{ + "instance_conf": { + "ENV": { + "API_KEY": "secret", + }, + }, + }], + }, + }, + } + + response = plugin.process_dauth_add_secrets_request(body) + + self.assertEqual(response["status"], "success") + self.assertEqual(response["job_id"], "7") + self.assertEqual( + plugin._chainstore[(DAUTH_JOB_SECRETS_CSTORE_HKEY, "7")], + { + "job_id": "7", + "plugin_secrets": body["plugin_secrets"], + }, + ) + + def test_add_secrets_rejects_non_oracle_writer(self): + plugin = _make_dauth_harness(protocol_oracles=["node-oracle"]) + body = { + "EE_SENDER": "node-runner", + "EE_ETH_SENDER": "0xRUNNER", + "job_id": "7", + "plugin_secrets": {"plugins": {}}, + } + + with self.assertRaisesRegex(ValueError, "not an oracle"): + plugin.process_dauth_add_secrets_request(body) + + self.assertNotIn((DAUTH_JOB_SECRETS_CSTORE_HKEY, "7"), plugin._chainstore) + + def test_add_secrets_rejects_invalid_signature(self): + plugin = _make_dauth_harness(valid_signature=False) + body = { + "EE_SENDER": "node-oracle", + "EE_ETH_SENDER": "0xORACLE", + "job_id": "7", + "plugin_secrets": {"plugins": {}}, + } + + with self.assertRaisesRegex(ValueError, "Invalid request signature"): + plugin.process_dauth_add_secrets_request(body) + + self.assertNotIn((DAUTH_JOB_SECRETS_CSTORE_HKEY, "7"), plugin._chainstore) + + def test_get_secrets_returns_bundle_for_node_running_job_from_r1fs_pipeline(self): + plugin = _make_dauth_harness() + bundle = { + "job_id": "7", + "plugin_secrets": { + "plugins": { + "CONTAINER_APP_RUNNER": [{ + "instance_conf": { + "ENV": { + "API_KEY": "secret", + }, + }, + }], + }, + }, + } + plugin._chainstore[(DAUTH_JOB_SECRETS_CSTORE_HKEY, "7")] = bundle + plugin._chainstore[(DEEPLOY_JOBS_CSTORE_HKEY, "7")] = "cid-7" + plugin._r1fs_data["cid-7"] = { + "deeploy_specs": { + "current_target_nodes": ["node-runner"], + }, + } + body = { + "EE_SENDER": "node-runner", + "EE_ETH_SENDER": "0xRUNNER", + "job_id": "7", + } + + response = plugin.process_dauth_get_secret_request(body) + + self.assertEqual(response["status"], "success") + self.assertEqual(response["job_id"], "7") + self.assertEqual(response["secret_bundle"], bundle) + + def test_get_secrets_rejects_node_not_running_job(self): + plugin = _make_dauth_harness() + plugin._chainstore[(DAUTH_JOB_SECRETS_CSTORE_HKEY, "7")] = { + "job_id": "7", + "plugin_secrets": {"plugins": {}}, + } + plugin._chainstore[(DEEPLOY_JOBS_CSTORE_HKEY, "7")] = "cid-7" + plugin._r1fs_data["cid-7"] = { + "DEEPLOY_SPECS": { + "current_target_nodes": ["node-runner"], + }, + } + body = { + "EE_SENDER": "node-other", + "EE_ETH_SENDER": "0xOTHER", + "job_id": "7", + } + + with self.assertRaisesRegex(ValueError, "not running job"): + plugin.process_dauth_get_secret_request(body) + + class DauthServerRegistryGateTests(unittest.TestCase): def _make_manager(self, *, dauth_oracle): From 4be3dcac0e84a45ec19b72f59aa1ca0692cebac4 Mon Sep 17 00:00:00 2001 From: Alessandro Date: Mon, 6 Jul 2026 12:22:06 +0200 Subject: [PATCH 2/9] fix: require job_secrets bundle shape --- extensions/business/dauth/dauth_mixin.py | 26 +++++-------------- .../dauth/test_dauth_registry_gating.py | 26 ++++++++++++++----- 2 files changed, 27 insertions(+), 25 deletions(-) diff --git a/extensions/business/dauth/dauth_mixin.py b/extensions/business/dauth/dauth_mixin.py index 237c971f..479b93aa 100644 --- a/extensions/business/dauth/dauth_mixin.py +++ b/extensions/business/dauth/dauth_mixin.py @@ -198,25 +198,13 @@ def _normalize_dauth_job_id(self, job_id): return str(job_id) def _build_secret_bundle_from_request(self, body, job_id): - secret_bundle = body.get("secret_bundle") - if secret_bundle is None: - plugin_secrets = body.get("plugin_secrets") - if plugin_secrets is None: - raise ValueError("Secret bundle is required.") - secret_bundle = { - "job_id": job_id, - "plugin_secrets": plugin_secrets, - } - if not isinstance(secret_bundle, dict): - raise ValueError("Secret bundle must be a dictionary.") - - bundle_job_id = secret_bundle.get("job_id", job_id) - if str(bundle_job_id) != job_id: - raise ValueError("Secret bundle job_id does not match request job_id.") - - secret_bundle = self.deepcopy(secret_bundle) - secret_bundle["job_id"] = job_id - return secret_bundle + 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), + } def _save_dauth_job_secret_bundle(self, job_id, secret_bundle): result = self.chainstore_hset( diff --git a/extensions/business/dauth/test_dauth_registry_gating.py b/extensions/business/dauth/test_dauth_registry_gating.py index fb98b783..84054586 100644 --- a/extensions/business/dauth/test_dauth_registry_gating.py +++ b/extensions/business/dauth/test_dauth_registry_gating.py @@ -344,7 +344,7 @@ def test_add_secrets_allows_protocol_oracle_and_overwrites_bundle(self): "EE_SENDER": "node-oracle", "EE_ETH_SENDER": "0xORACLE", "job_id": 7, - "plugin_secrets": { + "job_secrets": { "plugins": { "CONTAINER_APP_RUNNER": [{ "instance_conf": { @@ -365,7 +365,7 @@ def test_add_secrets_allows_protocol_oracle_and_overwrites_bundle(self): plugin._chainstore[(DAUTH_JOB_SECRETS_CSTORE_HKEY, "7")], { "job_id": "7", - "plugin_secrets": body["plugin_secrets"], + "job_secrets": body["job_secrets"], }, ) @@ -375,7 +375,7 @@ def test_add_secrets_rejects_non_oracle_writer(self): "EE_SENDER": "node-runner", "EE_ETH_SENDER": "0xRUNNER", "job_id": "7", - "plugin_secrets": {"plugins": {}}, + "job_secrets": {"plugins": {}}, } with self.assertRaisesRegex(ValueError, "not an oracle"): @@ -389,7 +389,7 @@ def test_add_secrets_rejects_invalid_signature(self): "EE_SENDER": "node-oracle", "EE_ETH_SENDER": "0xORACLE", "job_id": "7", - "plugin_secrets": {"plugins": {}}, + "job_secrets": {"plugins": {}}, } with self.assertRaisesRegex(ValueError, "Invalid request signature"): @@ -397,11 +397,25 @@ def test_add_secrets_rejects_invalid_signature(self): self.assertNotIn((DAUTH_JOB_SECRETS_CSTORE_HKEY, "7"), plugin._chainstore) + def test_add_secrets_rejects_legacy_plugin_secrets_shape(self): + plugin = _make_dauth_harness() + body = { + "EE_SENDER": "node-oracle", + "EE_ETH_SENDER": "0xORACLE", + "job_id": "7", + "plugin_secrets": {"plugins": {}}, + } + + with self.assertRaisesRegex(ValueError, "job_secrets must be a dictionary"): + plugin.process_dauth_add_secrets_request(body) + + self.assertNotIn((DAUTH_JOB_SECRETS_CSTORE_HKEY, "7"), plugin._chainstore) + def test_get_secrets_returns_bundle_for_node_running_job_from_r1fs_pipeline(self): plugin = _make_dauth_harness() bundle = { "job_id": "7", - "plugin_secrets": { + "job_secrets": { "plugins": { "CONTAINER_APP_RUNNER": [{ "instance_conf": { @@ -436,7 +450,7 @@ def test_get_secrets_rejects_node_not_running_job(self): plugin = _make_dauth_harness() plugin._chainstore[(DAUTH_JOB_SECRETS_CSTORE_HKEY, "7")] = { "job_id": "7", - "plugin_secrets": {"plugins": {}}, + "job_secrets": {"plugins": {}}, } plugin._chainstore[(DEEPLOY_JOBS_CSTORE_HKEY, "7")] = "cid-7" plugin._r1fs_data["cid-7"] = { From 49db53d46cf8a2370bc5b1ade0a25da4eee0f78a Mon Sep 17 00:00:00 2001 From: Alessandro Date: Mon, 6 Jul 2026 12:25:20 +0200 Subject: [PATCH 3/9] fix: simplification --- extensions/business/dauth/dauth_manager.py | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) diff --git a/extensions/business/dauth/dauth_manager.py b/extensions/business/dauth/dauth_manager.py index 9d98f90c..98f133a4 100644 --- a/extensions/business/dauth/dauth_manager.py +++ b/extensions/business/dauth/dauth_manager.py @@ -227,7 +227,6 @@ def on_resume(self): self._start_request_monitor_thread() return - def on_request(self, request): self._track_request(request) return @@ -325,7 +324,7 @@ def get_auth_data(self, body: dict): } } """ - if not self._is_dauth_server_enabled(): + if not self._dauth_server_enabled: response = self.__get_response({ 'error': 'dAuth server is not registered as a dAuth oracle' }) @@ -350,7 +349,7 @@ def add_secrets(self, body: dict): """ Store the full dAuth secret bundle for a job. Only protocol oracles can write. """ - if not self._is_dauth_server_enabled(): + if not self._dauth_server_enabled: response = self.__get_response({ 'error': 'dAuth server is not registered as a dAuth oracle' }) @@ -376,7 +375,7 @@ def get_secrets(self, body: dict): Return the full dAuth secret bundle for a job. Only nodes currently running the job in the R1FS-stored pipeline can read. """ - if not self._is_dauth_server_enabled(): + if not self._dauth_server_enabled: response = self.__get_response({ 'error': 'dAuth server is not registered as a dAuth oracle' }) From 238e26c1eb2c45e104e9650fd2c6ef45eb91a6b1 Mon Sep 17 00:00:00 2001 From: Alessandro Date: Thu, 23 Jul 2026 15:45:42 +0200 Subject: [PATCH 4/9] fix: protect dAuth secret requests from replay --- AGENTS.md | 9 ++ extensions/business/dauth/dauth_manager.py | 21 +++-- extensions/business/dauth/dauth_mixin.py | 30 ++++++- .../dauth/test_dauth_registry_gating.py | 88 ++++++++++++++++++- 4 files changed, 141 insertions(+), 7 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index dccc2061..1289f3a4 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -695,3 +695,12 @@ 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` diff --git a/extensions/business/dauth/dauth_manager.py b/extensions/business/dauth/dauth_manager.py index 98f133a4..7f9313fe 100644 --- a/extensions/business/dauth/dauth_manager.py +++ b/extensions/business/dauth/dauth_manager.py @@ -347,11 +347,16 @@ def get_auth_data(self, body: dict): # /add_secrets def add_secrets(self, body: dict): """ - Store the full dAuth secret bundle for a job. Only protocol oracles can write. + 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._dauth_server_enabled: response = self.__get_response({ - 'error': 'dAuth server is not registered as a dAuth oracle' + 'error': 'dAuth server is not registered as a dAuth oracle', + 'nonce': request_nonce, }) return response @@ -364,6 +369,7 @@ def add_secrets(self, body: dict): } response = self.__get_response({ + 'nonce': request_nonce, **data }) return response @@ -372,12 +378,16 @@ def add_secrets(self, body: dict): # /get_secrets def get_secrets(self, body: dict): """ - Return the full dAuth secret bundle for a job. Only nodes currently running the - job in the R1FS-stored pipeline can read. + 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._dauth_server_enabled: response = self.__get_response({ - 'error': 'dAuth server is not registered as a dAuth oracle' + 'error': 'dAuth server is not registered as a dAuth oracle', + 'nonce': request_nonce, }) return response @@ -390,6 +400,7 @@ def get_secrets(self, body: dict): } response = self.__get_response({ + 'nonce': request_nonce, **data }) return response diff --git a/extensions/business/dauth/dauth_mixin.py b/extensions/business/dauth/dauth_mixin.py index 479b93aa..c7c7da0e 100644 --- a/extensions/business/dauth/dauth_mixin.py +++ b/extensions/business/dauth/dauth_mixin.py @@ -17,6 +17,7 @@ DAUTH_JOB_SECRETS_CSTORE_HKEY = "DAUTH_JOB_SECRETS" DEEPLOY_JOBS_CSTORE_HKEY = "DEEPLOY_DEPLOYED_JOBS" +DAUTH_SECRET_REQUEST_MAX_AGE_SECONDS = 120 def version_to_int(version): @@ -192,6 +193,23 @@ def _is_protocol_oracle_eth(self, node_address_eth): raise ValueError("No oracles found - this is a critical issue!") return node_address_eth.lower() in [addr.lower() for addr in eth_oracles] + def _validate_dauth_secret_request_nonce(self, body): + """Validate the signed hex-millisecond timestamp nonce.""" + nonce = body.get(self.const.BASE_CT.dAuth.DAUTH_NONCE) + if not isinstance(nonce, str) or not nonce: + raise ValueError("dAuth request nonce is required.") + try: + request_time = int(nonce, 16) / 1000 + except (TypeError, ValueError) as exc: + raise ValueError("dAuth request nonce is invalid.") from exc + + request_age = self.time() - request_time + if request_age < 0: + raise ValueError("dAuth request nonce is from the future.") + if request_age > DAUTH_SECRET_REQUEST_MAX_AGE_SECONDS: + raise ValueError("dAuth request nonce is expired.") + return nonce + def _normalize_dauth_job_id(self, job_id): if job_id in [None, ""]: raise ValueError("Job ID is required.") @@ -262,6 +280,7 @@ def _is_node_running_dauth_job(self, job_id, node_address): 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.") @@ -272,10 +291,12 @@ def process_dauth_add_secrets_request(self, body): return { "status": "success", "job_id": job_id, + self.const.BASE_CT.dAuth.DAUTH_NONCE: request_nonce, } def process_dauth_get_secret_request(self, body): requester, _ = self._verify_signed_dauth_body(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): @@ -284,11 +305,18 @@ def process_dauth_get_secret_request(self, body): 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}.") + encrypted_secret_bundle = self.bc.encrypt_str( + str_data=self.json_dumps(secret_bundle), + str_recipient=requester, + ) + if not isinstance(encrypted_secret_bundle, str) or not encrypted_secret_bundle: + raise ValueError(f"Failed to encrypt dAuth secrets for job {job_id}.") return { "status": "success", "job_id": job_id, - "secret_bundle": secret_bundle, + self.const.BASE_CT.dAuth.DAUTH_NONCE: request_nonce, + "encrypted_secret_bundle": encrypted_secret_bundle, } def chainstore_store_dauth_request( diff --git a/extensions/business/dauth/test_dauth_registry_gating.py b/extensions/business/dauth/test_dauth_registry_gating.py index 84054586..92e543ec 100644 --- a/extensions/business/dauth/test_dauth_registry_gating.py +++ b/extensions/business/dauth/test_dauth_registry_gating.py @@ -1,4 +1,5 @@ from collections import deque +import json import queue import threading import unittest @@ -13,6 +14,8 @@ ROOT = Path(__file__).resolve().parents[3] +REQUEST_TIME = 1_700_000_000 +REQUEST_NONCE = hex(REQUEST_TIME * 1000) class _FakeProcess: @@ -139,6 +142,7 @@ def _load_dauth_manager_class(): class _FakeDauthConst: + DAUTH_NONCE = "nonce" DAUTH_ENV_KEYS_PREFIX = "EE_" DAUTH_WHITELIST = "DAUTH_WHITELIST" @@ -170,6 +174,7 @@ def __init__(self, *, dauth_oracle=True, protocol_oracles=None, valid_signature= self.dauth_oracle = dauth_oracle self.protocol_oracles = protocol_oracles or ["node-oracle"] self.valid_signature = valid_signature + self.encrypt_calls = [] self.node_eth = { "node-oracle": "0xORACLE", "node-runner": "0xRUNNER", @@ -211,6 +216,10 @@ def maybe_add_prefix(self, node_address): return node_address return "0xai_" + node_address + def encrypt_str(self, str_data, str_recipient): + self.encrypt_calls.append((str_data, str_recipient)) + return "encrypted-secret-bundle" + class _FakeR1FS: @@ -234,6 +243,8 @@ def _make_dauth_harness(*, dauth_oracle=True, protocol_oracles=None, valid_signa valid_signature=valid_signature, ) plugin.deepcopy = deepcopy + plugin.json_dumps = json.dumps + plugin.time = lambda: REQUEST_TIME plugin._chainstore = {} plugin._r1fs_data = {} plugin.r1fs = _FakeR1FS(plugin._r1fs_data) @@ -334,6 +345,30 @@ def test_dauth_token_fails_closed_when_dauth_registry_check_fails(self): class DauthJobSecretEndpointTests(unittest.TestCase): + def test_secret_request_nonce_accepts_only_last_120_seconds(self): + plugin = _make_dauth_harness() + + self.assertEqual( + plugin._validate_dauth_secret_request_nonce({"nonce": REQUEST_NONCE}), + REQUEST_NONCE, + ) + boundary_nonce = hex(int((REQUEST_TIME - 120) * 1000)) + self.assertEqual( + plugin._validate_dauth_secret_request_nonce({"nonce": boundary_nonce}), + boundary_nonce, + ) + + invalid_nonces = ( + ({}, "required"), + ({"nonce": "not-hex"}, "invalid"), + ({"nonce": hex(int((REQUEST_TIME + 1) * 1000))}, "future"), + ({"nonce": hex(int((REQUEST_TIME - 121) * 1000))}, "expired"), + ) + for body, message in invalid_nonces: + with self.subTest(body=body): + with self.assertRaisesRegex(ValueError, message): + plugin._validate_dauth_secret_request_nonce(body) + def test_add_secrets_allows_protocol_oracle_and_overwrites_bundle(self): plugin = _make_dauth_harness(protocol_oracles=["node-oracle"]) plugin._chainstore[(DAUTH_JOB_SECRETS_CSTORE_HKEY, "7")] = { @@ -343,6 +378,7 @@ def test_add_secrets_allows_protocol_oracle_and_overwrites_bundle(self): body = { "EE_SENDER": "node-oracle", "EE_ETH_SENDER": "0xORACLE", + "nonce": REQUEST_NONCE, "job_id": 7, "job_secrets": { "plugins": { @@ -361,6 +397,7 @@ def test_add_secrets_allows_protocol_oracle_and_overwrites_bundle(self): self.assertEqual(response["status"], "success") self.assertEqual(response["job_id"], "7") + self.assertEqual(response["nonce"], REQUEST_NONCE) self.assertEqual( plugin._chainstore[(DAUTH_JOB_SECRETS_CSTORE_HKEY, "7")], { @@ -374,6 +411,7 @@ def test_add_secrets_rejects_non_oracle_writer(self): body = { "EE_SENDER": "node-runner", "EE_ETH_SENDER": "0xRUNNER", + "nonce": REQUEST_NONCE, "job_id": "7", "job_secrets": {"plugins": {}}, } @@ -383,11 +421,27 @@ def test_add_secrets_rejects_non_oracle_writer(self): self.assertNotIn((DAUTH_JOB_SECRETS_CSTORE_HKEY, "7"), plugin._chainstore) + def test_add_secrets_rejects_expired_nonce_before_write(self): + plugin = _make_dauth_harness() + body = { + "EE_SENDER": "node-oracle", + "EE_ETH_SENDER": "0xORACLE", + "nonce": hex(int((REQUEST_TIME - 121) * 1000)), + "job_id": "7", + "job_secrets": {"plugins": {}}, + } + + with self.assertRaisesRegex(ValueError, "nonce is expired"): + plugin.process_dauth_add_secrets_request(body) + + self.assertNotIn((DAUTH_JOB_SECRETS_CSTORE_HKEY, "7"), plugin._chainstore) + def test_add_secrets_rejects_invalid_signature(self): plugin = _make_dauth_harness(valid_signature=False) body = { "EE_SENDER": "node-oracle", "EE_ETH_SENDER": "0xORACLE", + "nonce": REQUEST_NONCE, "job_id": "7", "job_secrets": {"plugins": {}}, } @@ -402,6 +456,7 @@ def test_add_secrets_rejects_legacy_plugin_secrets_shape(self): body = { "EE_SENDER": "node-oracle", "EE_ETH_SENDER": "0xORACLE", + "nonce": REQUEST_NONCE, "job_id": "7", "plugin_secrets": {"plugins": {}}, } @@ -437,6 +492,7 @@ def test_get_secrets_returns_bundle_for_node_running_job_from_r1fs_pipeline(self body = { "EE_SENDER": "node-runner", "EE_ETH_SENDER": "0xRUNNER", + "nonce": REQUEST_NONCE, "job_id": "7", } @@ -444,7 +500,16 @@ def test_get_secrets_returns_bundle_for_node_running_job_from_r1fs_pipeline(self self.assertEqual(response["status"], "success") self.assertEqual(response["job_id"], "7") - self.assertEqual(response["secret_bundle"], bundle) + self.assertEqual(response["nonce"], REQUEST_NONCE) + self.assertEqual( + response["encrypted_secret_bundle"], + "encrypted-secret-bundle", + ) + self.assertNotIn("secret_bundle", response) + self.assertEqual( + plugin.bc.encrypt_calls, + [(json.dumps(bundle), "node-runner")], + ) def test_get_secrets_rejects_node_not_running_job(self): plugin = _make_dauth_harness() @@ -461,6 +526,7 @@ def test_get_secrets_rejects_node_not_running_job(self): body = { "EE_SENDER": "node-other", "EE_ETH_SENDER": "0xOTHER", + "nonce": REQUEST_NONCE, "job_id": "7", } @@ -500,8 +566,28 @@ def is_dauth_oracle(self): plugin._init_request_tracking = lambda: None plugin.bc.address = "node-address" plugin.bc.eth_address = "0xNODE" + plugin._DauthManagerPlugin__get_response = lambda data: data return plugin + def test_secret_endpoint_errors_echo_request_nonce(self): + plugin = self._make_manager(dauth_oracle=True) + plugin._dauth_server_enabled = True + plugin.process_dauth_add_secrets_request = lambda body: (_ for _ in ()).throw( + ValueError("add failed") + ) + plugin.process_dauth_get_secret_request = lambda body: (_ for _ in ()).throw( + ValueError("get failed") + ) + body = {"nonce": REQUEST_NONCE} + + add_response = plugin.add_secrets(body) + get_response = plugin.get_secrets(body) + + self.assertEqual(add_response["nonce"], REQUEST_NONCE) + self.assertEqual(add_response["error"], "add failed") + self.assertEqual(get_response["nonce"], REQUEST_NONCE) + self.assertEqual(get_response["error"], "get failed") + def test_startup_lookup_is_cached_across_repeated_lifecycle_predicates(self): plugin = self._make_manager(dauth_oracle=True) From 13a739a704d1f786da543f33c48fd84ccd3d7fee Mon Sep 17 00:00:00 2001 From: Alessandro Date: Fri, 31 Jul 2026 18:04:07 +0200 Subject: [PATCH 5/9] fix --- extensions/business/dauth/dauth_manager.py | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/extensions/business/dauth/dauth_manager.py b/extensions/business/dauth/dauth_manager.py index 7f9313fe..330f5fdd 100644 --- a/extensions/business/dauth/dauth_manager.py +++ b/extensions/business/dauth/dauth_manager.py @@ -227,6 +227,7 @@ def on_resume(self): self._start_request_monitor_thread() return + def on_request(self, request): self._track_request(request) return @@ -324,7 +325,7 @@ def get_auth_data(self, body: dict): } } """ - if not self._dauth_server_enabled: + if not self._is_dauth_server_enabled(): response = self.__get_response({ 'error': 'dAuth server is not registered as a dAuth oracle' }) @@ -353,7 +354,7 @@ def add_secrets(self, body: dict): than 120 seconds. """ request_nonce = body.get("nonce") if isinstance(body, dict) else None - if not self._dauth_server_enabled: + if not self._is_dauth_server_enabled(): response = self.__get_response({ 'error': 'dAuth server is not registered as a dAuth oracle', 'nonce': request_nonce, @@ -384,7 +385,7 @@ def get_secrets(self, body: dict): than 120 seconds. The signed response echoes that nonce. """ request_nonce = body.get("nonce") if isinstance(body, dict) else None - if not self._dauth_server_enabled: + if not self._is_dauth_server_enabled(): response = self.__get_response({ 'error': 'dAuth server is not registered as a dAuth oracle', 'nonce': request_nonce, From 0e2aa6449f16096298e3c093bfeccbf6e2afbcec Mon Sep 17 00:00:00 2001 From: Alessandro Date: Fri, 31 Jul 2026 18:52:36 +0200 Subject: [PATCH 6/9] feat: sync dAuth secrets through registry peers --- AGENTS.md | 9 ++ extensions/business/dauth/dauth_manager.py | 47 +++++++- extensions/business/dauth/dauth_mixin.py | 4 + extensions/business/dauth/dauth_registry.py | 77 +++++++++++++ .../dauth/test_dauth_registry_gating.py | 83 +++++++++++++- .../dauth/test_dauth_secret_routing.py | 105 ++++++++++++++++++ 6 files changed, 319 insertions(+), 6 deletions(-) create mode 100644 extensions/business/dauth/dauth_registry.py create mode 100644 extensions/business/dauth/test_dauth_secret_routing.py diff --git a/AGENTS.md b/AGENTS.md index 1289f3a4..cba7d4b0 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -704,3 +704,12 @@ Entry format: - 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` diff --git a/extensions/business/dauth/dauth_manager.py b/extensions/business/dauth/dauth_manager.py index 330f5fdd..c581321c 100644 --- a/extensions/business/dauth/dauth_manager.py +++ b/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,8 @@ 'REQUESTS_MAX_RECORDS': 2, 'REQUESTS_LOG_INTERVAL': 5 * 60, + 'DAUTH_JOB_SECRETS_HSYNC_INTERVAL': 60, + 'SUPRESS_LOGS_AFTER_INTERVAL' : 300, # required ENV keys are defined in plugin template and should be added here @@ -105,6 +114,9 @@ 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_job_secrets_hsync = None self._dauth_web_app_initialized = False self._dauth_pause_teardown_succeeded = True return @@ -117,6 +129,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 @@ -134,7 +148,13 @@ def _check_dauth_server_enabled_on_start(self): 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 + ] + if enabled: + self._dauth_registry_eth_oracles = eth_oracles + self._dauth_registry_internal_peers = peers except Exception as e: enabled = False error = str(e) @@ -237,11 +257,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. diff --git a/extensions/business/dauth/dauth_mixin.py b/extensions/business/dauth/dauth_mixin.py index c7c7da0e..7ece280c 100644 --- a/extensions/business/dauth/dauth_mixin.py +++ b/extensions/business/dauth/dauth_mixin.py @@ -15,6 +15,9 @@ """ +from extensions.business.dauth.dauth_registry import dauth_registry_write_kwargs + + DAUTH_JOB_SECRETS_CSTORE_HKEY = "DAUTH_JOB_SECRETS" DEEPLOY_JOBS_CSTORE_HKEY = "DEEPLOY_DEPLOYED_JOBS" DAUTH_SECRET_REQUEST_MAX_AGE_SECONDS = 120 @@ -229,6 +232,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}.") diff --git a/extensions/business/dauth/dauth_registry.py b/extensions/business/dauth/dauth_registry.py new file mode 100644 index 00000000..17bde0c4 --- /dev/null +++ b/extensions/business/dauth/dauth_registry.py @@ -0,0 +1,77 @@ +"""dAuth registry lookup and ChainStore routing helpers.""" + + +def resolve_dauth_registry_internal_peers(plugin, eth_oracles): + """Resolve cached registry ETH addresses through current NetMon state.""" + current_eth = plugin.bc.eth_address.lower() + peers = [] + for eth_address in eth_oracles: + internal_address = plugin.bc.eth_addr_to_internal_addr(eth_address) + if internal_address is None and eth_address.lower() == current_eth: + internal_address = plugin.bc.address + if isinstance(internal_address, str) and internal_address: + peers.append(internal_address) + return list(dict.fromkeys(peers)) + + +def load_dauth_registry_snapshot(plugin): + """Load the dAuth registry once and resolve its currently known peers.""" + eth_oracles = plugin.bc.get_eth_dauth_oracles() + eth_oracles = list(dict.fromkeys( + address + for address in eth_oracles or [] + if isinstance(address, str) and address + )) + if not eth_oracles: + raise ValueError("No dAuth oracles are registered.") + + peers = resolve_dauth_registry_internal_peers(plugin, eth_oracles) + if not peers: + raise ValueError("No dAuth registry internal peers are available.") + return peers, eth_oracles + + +def get_cached_dauth_registry_internal_peers(plugin): + """Return the startup-cached dAuth oracle internal addresses.""" + eth_oracles = getattr(plugin, "_dauth_registry_eth_oracles", None) + if eth_oracles: + peers = resolve_dauth_registry_internal_peers(plugin, eth_oracles) + if peers: + plugin._dauth_registry_internal_peers = peers + peers = getattr(plugin, "_dauth_registry_internal_peers", None) + if not peers: + raise ValueError("dAuth registry peers were not cached at startup.") + return list(peers) + + +def get_dauth_registry_internal_peers(plugin): + """Return cached peers when available, otherwise load a registry snapshot.""" + if ( + getattr(plugin, "_dauth_registry_eth_oracles", None) + or hasattr(plugin, "_dauth_registry_internal_peers") + ): + return get_cached_dauth_registry_internal_peers(plugin) + peers, _ = load_dauth_registry_snapshot(plugin) + return peers + + +def dauth_registry_write_kwargs(plugin, peers=None): + """Route a ChainStore 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..8ae3c617 100644 --- a/extensions/business/dauth/test_dauth_registry_gating.py +++ b/extensions/business/dauth/test_dauth_registry_gating.py @@ -11,6 +11,7 @@ DEEPLOY_JOBS_CSTORE_HKEY, _DauthMixin, ) +from extensions.business.dauth.dauth_registry import load_dauth_registry_snapshot ROOT = Path(__file__).resolve().parents[3] @@ -124,14 +125,33 @@ def _load_dauth_manager_class(): "", ) source = source.replace( - "from extensions.business.dauth.dauth_mixin import _DauthMixin\n", + "from extensions.business.dauth.dauth_mixin import (\n" + " DAUTH_JOB_SECRETS_CSTORE_HKEY,\n" + " _DauthMixin,\n" + ")\n", + "", + ) + source = source.replace( + "from extensions.business.dauth.dauth_registry import (\n" + " dauth_registry_write_kwargs,\n" + " load_dauth_registry_snapshot,\n" + ")\n", "", ) namespace = { "BasePlugin": _FakeBasePlugin, + "DAUTH_JOB_SECRETS_CSTORE_HKEY": DAUTH_JOB_SECRETS_CSTORE_HKEY, "_DauthMixin": _FakeDauthMixin, "_NodeTagsMixin": _FakeNodeTagsMixin, "_RequestTrackingMixin": _FakeRequestTrackingMixin, + "dauth_registry_write_kwargs": ( + lambda plugin: { + "extra_peers": list(plugin._dauth_registry_internal_peers), + "include_default_peers": False, + "include_configured_peers": False, + } + ), + "load_dauth_registry_snapshot": load_dauth_registry_snapshot, "__name__": "loaded_dauth_manager", } exec(compile(source, str(source_path), "exec"), namespace) # noqa: S102 @@ -273,7 +293,8 @@ 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._dauth_registry_internal_peers = ["node-oracle"] + plugin.chainstore_hset = lambda hkey, key, value, **kwargs: plugin._chainstore.__setitem__( (hkey, str(key)), deepcopy(value), ) or True @@ -542,11 +563,17 @@ def __init__(self, result): self.result = result self.calls = 0 - def is_dauth_oracle(self): + def get_eth_dauth_oracles(self): self.calls += 1 if isinstance(self.result, Exception): raise self.result - return self.result + return ["0xNODE", "0xPEER"] if self.result else ["0xPEER"] + + def eth_addr_to_internal_addr(self, eth_address): + return { + "0xnode": "node-address", + "0xpeer": "peer-address", + }.get(eth_address.lower()) plugin = DauthManagerPlugin.__new__(DauthManagerPlugin) plugin.bc = _ManagerBC(dauth_oracle) @@ -566,6 +593,14 @@ def is_dauth_oracle(self): plugin._init_request_tracking = lambda: None plugin.bc.address = "node-address" plugin.bc.eth_address = "0xNODE" + plugin._dauth_registry_eth_oracles = None + plugin._dauth_registry_internal_peers = None + plugin._last_dauth_job_secrets_hsync = None + plugin.cfg_dauth_job_secrets_hsync_interval = 60 + plugin._hsync_calls = [] + plugin.chainstore_hsync = lambda **kwargs: plugin._hsync_calls.append(kwargs) or { + "hkey": kwargs["hkey"], + } plugin._DauthManagerPlugin__get_response = lambda data: data return plugin @@ -601,6 +636,45 @@ def test_startup_lookup_is_cached_across_repeated_lifecycle_predicates(self): self.assertEqual(plugin.bc.calls, 1) self.assertEqual(plugin._base_init_calls, 1) self.assertEqual(plugin._lifecycle_events, ["base_init"]) + self.assertEqual( + plugin._dauth_registry_internal_peers, + ["node-address", "peer-address"], + ) + + def test_secret_hsync_runs_at_startup_and_once_per_minute_on_cached_peers(self): + plugin = self._make_manager(dauth_oracle=True) + + plugin.on_init() + plugin.process() + plugin._now += 59 + plugin.process() + plugin._now += 1 + plugin.process() + + self.assertEqual(plugin.bc.calls, 1) + self.assertEqual(len(plugin._hsync_calls), 2) + for call in plugin._hsync_calls: + self.assertEqual(call["hkey"], DAUTH_JOB_SECRETS_CSTORE_HKEY) + self.assertEqual(call["extra_peers"], ["node-address", "peer-address"]) + self.assertFalse(call["include_default_peers"]) + self.assertFalse(call["include_configured_peers"]) + + def test_secret_hsync_failure_waits_until_next_interval(self): + plugin = self._make_manager(dauth_oracle=True) + attempts = [] + + def fail_hsync(**kwargs): + attempts.append(kwargs) + raise ValueError("sync unavailable") + + plugin.chainstore_hsync = fail_hsync + plugin.on_init() + plugin.process() + plugin._now += 60 + plugin.process() + + self.assertEqual(len(attempts), 2) + self.assertTrue(any("sync unavailable" in message for message in plugin._messages)) def test_false_startup_lookup_fails_closed_and_tears_down_fastapi(self): plugin = self._make_manager(dauth_oracle=False) @@ -610,6 +684,7 @@ def test_false_startup_lookup_fails_closed_and_tears_down_fastapi(self): self.assertTrue(plugin.should_pause()) self.assertFalse(plugin.should_resume()) self.assertEqual(plugin.bc.calls, 1) + self.assertEqual(plugin._hsync_calls, []) self.assertTrue(plugin._stop_request_monitor.is_set()) self.assertFalse(plugin._request_monitor_thread.is_alive()) self.assertEqual(plugin.start_commands_started, [False, False]) 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..6b4fe843 --- /dev/null +++ b/extensions/business/dauth/test_dauth_secret_routing.py @@ -0,0 +1,105 @@ +import unittest + +from extensions.business.dauth.dauth_mixin import _DauthMixin +from extensions.business.dauth.dauth_registry import ( + dauth_registry_write_kwargs, + load_dauth_registry_snapshot, + pipeline_registry_write_kwargs, +) + + +class _RegistryBCStub: + address = "node-local" + eth_address = "0xLOCAL" + + def __init__(self, remote_available=True): + self.calls = 0 + self.remote_available = remote_available + + def get_eth_dauth_oracles(self): + self.calls += 1 + return ["0xLOCAL", "0xREMOTE", "0xUNKNOWN"] + + def eth_addr_to_internal_addr(self, eth_address): + if self.remote_available and eth_address == "0xREMOTE": + return "node-remote" + return None + + +class _DauthStub(_DauthMixin): + def __init__(self): + self._dauth_registry_internal_peers = ["dauth-a", "dauth-b"] + self.writes = [] + + def chainstore_hset(self, **kwargs): + self.writes.append(kwargs) + return True + + +class DauthSecretRoutingTests(unittest.TestCase): + def test_registry_snapshot_uses_one_rpc_and_maps_every_known_peer(self): + class _Plugin: + bc = _RegistryBCStub() + + peers, eth_oracles = load_dauth_registry_snapshot(_Plugin()) + + self.assertEqual(_Plugin.bc.calls, 1) + self.assertEqual(peers, ["node-local", "node-remote"]) + self.assertEqual(eth_oracles, ["0xLOCAL", "0xREMOTE", "0xUNKNOWN"]) + + def test_routing_refreshes_internal_mappings_without_another_rpc(self): + class _Plugin: + bc = _RegistryBCStub(remote_available=False) + + plugin = _Plugin() + peers, eth_oracles = load_dauth_registry_snapshot(plugin) + plugin._dauth_registry_eth_oracles = eth_oracles + plugin._dauth_registry_internal_peers = peers + plugin.bc.remote_available = True + + routing = dauth_registry_write_kwargs(plugin) + + self.assertEqual(plugin.bc.calls, 1) + self.assertEqual(routing["extra_peers"], ["node-local", "node-remote"]) + + def test_secret_storage_targets_only_cached_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_explicit_peer_builders_support_non_manager_callers(self): + plugin = object() + + secret_routing = dauth_registry_write_kwargs(plugin, peers=["dauth-a"]) + pipeline_routing = pipeline_registry_write_kwargs(plugin, peers=["dauth-a"]) + + self.assertEqual(secret_routing["extra_peers"], ["dauth-a"]) + self.assertFalse(secret_routing["include_default_peers"]) + self.assertFalse(secret_routing["include_configured_peers"]) + self.assertEqual(pipeline_routing["extra_peers"], ["dauth-a"]) + self.assertTrue(pipeline_routing["include_default_peers"]) + self.assertTrue(pipeline_routing["include_configured_peers"]) + + def test_secret_storage_fails_without_startup_cached_peers(self): + plugin = _DauthStub() + plugin._dauth_registry_internal_peers = [] + + with self.assertRaisesRegex(ValueError, "not cached"): + plugin._save_dauth_job_secret_bundle( + "7", + {"job_id": "7", "job_secrets": {}}, + ) + + self.assertEqual(plugin.writes, []) + + +if __name__ == "__main__": + unittest.main() From c72119f9eb01e79ac39035d69120579b1abd19ee Mon Sep 17 00:00:00 2001 From: Alessandro Date: Mon, 3 Aug 2026 18:23:59 +0200 Subject: [PATCH 7/9] fix: refresh dAuth registry authorization --- AGENTS.md | 9 + extensions/business/dauth/dauth_manager.py | 98 +++++++++- extensions/business/dauth/dauth_registry.py | 6 +- .../dauth/test_dauth_registry_gating.py | 184 +++++++++++++++++- 4 files changed, 280 insertions(+), 17 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index cba7d4b0..53c2534e 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -713,3 +713,12 @@ Entry format: - 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 or 30-second timed-out reads clear cached peers, fail closed, and retry after one minute. Timed-out lookup results are isolated from authorization state, and at most two lookup workers may remain pending so one abandoned call cannot block recovery or cause unbounded thread growth. 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` diff --git a/extensions/business/dauth/dauth_manager.py b/extensions/business/dauth/dauth_manager.py index c581321c..50ffc6bc 100644 --- a/extensions/business/dauth/dauth_manager.py +++ b/extensions/business/dauth/dauth_manager.py @@ -20,6 +20,8 @@ """ +import threading + 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 @@ -49,7 +51,11 @@ 'REQUESTS_MAX_RECORDS': 2, 'REQUESTS_LOG_INTERVAL': 5 * 60, - 'DAUTH_JOB_SECRETS_HSYNC_INTERVAL': 60, + 'DAUTH_JOB_SECRETS_HSYNC_INTERVAL': 10 * 60, + 'DAUTH_REGISTRY_REFRESH_INTERVAL': 60 * 60, + 'DAUTH_REGISTRY_REFRESH_RETRY_INTERVAL': 60, + 'DAUTH_REGISTRY_REFRESH_TIMEOUT': 30, + 'DAUTH_REGISTRY_MAX_PENDING_LOOKUPS': 2, 'SUPRESS_LOGS_AFTER_INTERVAL' : 300, @@ -116,6 +122,9 @@ def __init__(self, **kwargs): 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._dauth_registry_lookup_threads = [] self._last_dauth_job_secrets_hsync = None self._dauth_web_app_initialized = False self._dauth_pause_teardown_succeeded = True @@ -146,26 +155,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: - peers, eth_oracles = load_dauth_registry_snapshot(self) + peers, eth_oracles = self._load_dauth_registry_snapshot_with_timeout() enabled = self.bc.eth_address.lower() in [ address.lower() for address in eth_oracles ] - if enabled: - self._dauth_registry_eth_oracles = eth_oracles - self._dauth_registry_internal_peers = peers 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})", @@ -175,13 +210,52 @@ def _check_dauth_server_enabled_on_start(self): # endif return enabled + def _load_dauth_registry_snapshot_with_timeout(self): + lookup_threads = [ + thread for thread in getattr(self, "_dauth_registry_lookup_threads", []) + if thread.is_alive() + ] + self._dauth_registry_lookup_threads = lookup_threads + if len(lookup_threads) >= self.cfg_dauth_registry_max_pending_lookups: + raise TimeoutError("too many dAuth registry lookups are still running") + # endif + + result = {} + + def load_registry(): + try: + result["snapshot"] = load_dauth_registry_snapshot(self) + except Exception as exc: + result["error"] = exc + # end try + return + + lookup_thread = threading.Thread(target=load_registry, daemon=True) + self._dauth_registry_lookup_threads.append(lookup_thread) + lookup_thread.start() + lookup_thread.join(timeout=self.cfg_dauth_registry_refresh_timeout) + if lookup_thread.is_alive(): + raise TimeoutError( + f"dAuth registry lookup timed out after " + f"{self.cfg_dauth_registry_refresh_timeout} seconds" + ) + # endif + + self._dauth_registry_lookup_threads.remove(lookup_thread) + error = result.get("error") + if error is not None: + raise error + return result["snapshot"] + 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): @@ -189,6 +263,7 @@ def on_pause(self): 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: @@ -246,6 +321,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): diff --git a/extensions/business/dauth/dauth_registry.py b/extensions/business/dauth/dauth_registry.py index 17bde0c4..a9951972 100644 --- a/extensions/business/dauth/dauth_registry.py +++ b/extensions/business/dauth/dauth_registry.py @@ -15,7 +15,7 @@ def resolve_dauth_registry_internal_peers(plugin, eth_oracles): def load_dauth_registry_snapshot(plugin): - """Load the dAuth registry once and resolve its currently known peers.""" + """Load the current dAuth registry and resolve its currently known peers.""" eth_oracles = plugin.bc.get_eth_dauth_oracles() eth_oracles = list(dict.fromkeys( address @@ -32,7 +32,7 @@ def load_dauth_registry_snapshot(plugin): def get_cached_dauth_registry_internal_peers(plugin): - """Return the startup-cached dAuth oracle internal addresses.""" + """Return the latest cached dAuth oracle internal addresses.""" eth_oracles = getattr(plugin, "_dauth_registry_eth_oracles", None) if eth_oracles: peers = resolve_dauth_registry_internal_peers(plugin, eth_oracles) @@ -40,7 +40,7 @@ def get_cached_dauth_registry_internal_peers(plugin): plugin._dauth_registry_internal_peers = peers peers = getattr(plugin, "_dauth_registry_internal_peers", None) if not peers: - raise ValueError("dAuth registry peers were not cached at startup.") + raise ValueError("dAuth registry peers are not cached.") return list(peers) diff --git a/extensions/business/dauth/test_dauth_registry_gating.py b/extensions/business/dauth/test_dauth_registry_gating.py index 8ae3c617..31835093 100644 --- a/extensions/business/dauth/test_dauth_registry_gating.py +++ b/extensions/business/dauth/test_dauth_registry_gating.py @@ -96,6 +96,14 @@ def _start_request_monitor_thread(self): self._request_monitor_thread = _FakeThread() return + def set_plugin_ready(self, ready=True): + self._is_plugin_ready = ready + return + + def on_log_handler(self, text, key=None): # pylint: disable=unused-argument + self._lifecycle_events.append("log") + return + class _FakeDauthMixin: pass @@ -567,12 +575,17 @@ def get_eth_dauth_oracles(self): self.calls += 1 if isinstance(self.result, Exception): raise self.result + if callable(self.result): + return self.result() + if isinstance(self.result, list): + return self.result return ["0xNODE", "0xPEER"] if self.result else ["0xPEER"] def eth_addr_to_internal_addr(self, eth_address): return { "0xnode": "node-address", "0xpeer": "peer-address", + "0xnew": "new-peer-address", }.get(eth_address.lower()) plugin = DauthManagerPlugin.__new__(DauthManagerPlugin) @@ -595,8 +608,16 @@ def eth_addr_to_internal_addr(self, eth_address): plugin.bc.eth_address = "0xNODE" plugin._dauth_registry_eth_oracles = None plugin._dauth_registry_internal_peers = None + plugin._last_dauth_registry_refresh = None + plugin._dauth_registry_refresh_failed = False + plugin._dauth_registry_lookup_threads = [] plugin._last_dauth_job_secrets_hsync = None - plugin.cfg_dauth_job_secrets_hsync_interval = 60 + plugin.cfg_dauth_job_secrets_hsync_interval = 10 * 60 + plugin.cfg_dauth_registry_refresh_interval = 60 * 60 + plugin.cfg_dauth_registry_refresh_retry_interval = 60 + plugin.cfg_dauth_registry_refresh_timeout = 30 + plugin.cfg_dauth_registry_max_pending_lookups = 2 + plugin._is_plugin_ready = None plugin._hsync_calls = [] plugin.chainstore_hsync = lambda **kwargs: plugin._hsync_calls.append(kwargs) or { "hkey": kwargs["hkey"], @@ -623,7 +644,7 @@ def test_secret_endpoint_errors_echo_request_nonce(self): self.assertEqual(get_response["nonce"], REQUEST_NONCE) self.assertEqual(get_response["error"], "get failed") - def test_startup_lookup_is_cached_across_repeated_lifecycle_predicates(self): + def test_registry_lookup_is_cached_between_hourly_lifecycle_refreshes(self): plugin = self._make_manager(dauth_oracle=True) plugin.on_init() @@ -641,12 +662,20 @@ def test_startup_lookup_is_cached_across_repeated_lifecycle_predicates(self): ["node-address", "peer-address"], ) - def test_secret_hsync_runs_at_startup_and_once_per_minute_on_cached_peers(self): + plugin._now += (60 * 60) - 1 + self.assertFalse(plugin.should_pause()) + self.assertEqual(plugin.bc.calls, 1) + + plugin._now += 1 + self.assertFalse(plugin.should_pause()) + self.assertEqual(plugin.bc.calls, 2) + + def test_secret_hsync_runs_at_startup_and_every_ten_minutes_on_cached_peers(self): plugin = self._make_manager(dauth_oracle=True) plugin.on_init() plugin.process() - plugin._now += 59 + plugin._now += (10 * 60) - 1 plugin.process() plugin._now += 1 plugin.process() @@ -670,7 +699,7 @@ def fail_hsync(**kwargs): plugin.chainstore_hsync = fail_hsync plugin.on_init() plugin.process() - plugin._now += 60 + plugin._now += 10 * 60 plugin.process() self.assertEqual(len(attempts), 2) @@ -705,7 +734,7 @@ def test_false_startup_lookup_fails_closed_and_tears_down_fastapi(self): ], ) - def test_startup_lookup_error_fails_closed_and_is_not_retried(self): + def test_startup_lookup_error_fails_closed_and_retries_after_one_minute(self): plugin = self._make_manager(dauth_oracle=RuntimeError("registry unavailable")) plugin.on_init() @@ -717,6 +746,144 @@ def test_startup_lookup_error_fails_closed_and_is_not_retried(self): self.assertEqual(plugin._dauth_server_enabled_message, "registry unavailable") self.assertTrue(plugin._dauth_pause_teardown_succeeded) + plugin._now += 59 + self.assertTrue(plugin.should_pause()) + self.assertEqual(plugin.bc.calls, 1) + + plugin._now += 1 + self.assertTrue(plugin.should_pause()) + self.assertEqual(plugin.bc.calls, 2) + + def test_registry_lookup_timeout_fails_closed_without_late_state_update(self): + lookup_release = threading.Event() + + def delayed_registry_lookup(): + lookup_release.wait() + return ["0xNODE", "0xPEER"] + + plugin = self._make_manager(dauth_oracle=delayed_registry_lookup) + plugin.cfg_dauth_registry_refresh_timeout = 0.001 + + plugin.on_init() + + self.assertFalse(plugin._is_dauth_server_enabled()) # pylint: disable=protected-access + self.assertIn("timed out", plugin._dauth_server_enabled_message) + self.assertEqual(len(plugin._dauth_registry_lookup_threads), 1) + self.assertTrue(plugin._dauth_registry_lookup_threads[0].is_alive()) + + plugin.bc.result = True + plugin._now += 60 + self.assertTrue(plugin.should_resume()) + self.assertEqual(plugin.bc.calls, 2) + + lookup_release.set() + plugin._dauth_registry_lookup_threads[0].join(timeout=1) + self.assertTrue(plugin._is_dauth_server_enabled()) # pylint: disable=protected-access + + def test_registry_lookup_timeouts_cap_abandoned_workers(self): + lookup_release = threading.Event() + + def blocked_registry_lookup(): + lookup_release.wait() + return ["0xNODE", "0xPEER"] + + plugin = self._make_manager(dauth_oracle=blocked_registry_lookup) + plugin.cfg_dauth_registry_refresh_timeout = 0.001 + plugin.on_init() + + plugin._now += 60 + self.assertFalse(plugin.should_resume()) + plugin._now += 60 + self.assertFalse(plugin.should_resume()) + + self.assertEqual(plugin.bc.calls, 2) + self.assertEqual(len(plugin._dauth_registry_lookup_threads), 2) + self.assertIn("too many", plugin._dauth_server_enabled_message) + + lookup_release.set() + for lookup_thread in plugin._dauth_registry_lookup_threads: + lookup_thread.join(timeout=1) + # endfor + + def test_hourly_refresh_revokes_server_and_secret_replication(self): + plugin = self._make_manager(dauth_oracle=True) + plugin.on_init() + plugin.bc.result = False + + plugin._now += (60 * 60) - 1 + self.assertFalse(plugin.should_pause()) + self.assertEqual(plugin.bc.calls, 1) + + plugin._now += 1 + self.assertTrue(plugin.should_pause()) + plugin.on_pause() + self.assertEqual(plugin.bc.calls, 2) + self.assertIsNone(plugin._dauth_registry_eth_oracles) + self.assertIsNone(plugin._dauth_registry_internal_peers) + self.assertTrue(plugin._stop_request_monitor.is_set()) + self.assertEqual(plugin.start_commands_processes, [None, None]) + + hsync_calls = len(plugin._hsync_calls) + plugin._now += 10 * 60 + plugin.process() + self.assertEqual(len(plugin._hsync_calls), hsync_calls) + + def test_hourly_refresh_replaces_removed_replication_peers(self): + plugin = self._make_manager(dauth_oracle=True) + plugin.on_init() + plugin.bc.result = ["0xNODE", "0xNEW"] + + plugin._now += 60 * 60 + self.assertFalse(plugin.should_pause()) + + self.assertEqual(plugin.bc.calls, 2) + self.assertEqual(plugin._dauth_registry_eth_oracles, ["0xNODE", "0xNEW"]) + self.assertEqual( + plugin._dauth_registry_internal_peers, + ["node-address", "new-peer-address"], + ) + plugin.process() + self.assertEqual( + plugin._hsync_calls[-1]["extra_peers"], + ["node-address", "new-peer-address"], + ) + + def test_hourly_refresh_allows_newly_registered_server_to_resume(self): + plugin = self._make_manager(dauth_oracle=False) + plugin.on_init() + plugin.bc.result = True + + plugin._now += 60 * 60 + self.assertTrue(plugin.should_resume()) + + self.assertEqual(plugin.bc.calls, 2) + self.assertEqual( + plugin._dauth_registry_internal_peers, + ["node-address", "peer-address"], + ) + + def test_hourly_refresh_error_revokes_server_and_clears_peers(self): + plugin = self._make_manager(dauth_oracle=True) + plugin.on_init() + plugin.bc.result = RuntimeError("registry unavailable") + + plugin._now += 60 * 60 + self.assertTrue(plugin.should_pause()) + + self.assertEqual(plugin.bc.calls, 2) + self.assertEqual(plugin._dauth_server_enabled_message, "registry unavailable") + self.assertIsNone(plugin._dauth_registry_eth_oracles) + self.assertIsNone(plugin._dauth_registry_internal_peers) + + plugin.bc.result = True + plugin._now += 59 + self.assertFalse(plugin.should_resume()) + self.assertEqual(plugin.bc.calls, 2) + + plugin._now += 1 + self.assertTrue(plugin.should_resume()) + self.assertEqual(plugin.bc.calls, 3) + def test_pause_tears_down_and_resume_restarts_only_request_monitor(self): plugin = self._make_manager(dauth_oracle=True) plugin.on_init() @@ -727,6 +894,7 @@ def test_pause_tears_down_and_resume_restarts_only_request_monitor(self): plugin.on_pause() self.assertTrue(plugin._dauth_pause_teardown_succeeded) + self.assertFalse(plugin._is_plugin_ready) self.assertEqual(plugin.start_commands_processes, [None, None]) self.assertEqual(list(plugin._incoming_requests), []) self.assertEqual(list(plugin.postponed_requests), []) @@ -738,9 +906,13 @@ def test_pause_tears_down_and_resume_restarts_only_request_monitor(self): self.assertFalse(plugin.failed) self.assertFalse(plugin._stop_request_monitor.is_set()) self.assertTrue(plugin._request_monitor_thread.is_alive()) + self.assertFalse(plugin._is_plugin_ready) self.assertEqual(plugin._lifecycle_events[-1], "start_monitor") self.assertEqual(plugin.bc.calls, 1) + plugin.on_log_handler("Uvicorn running on http://0.0.0.0:1234 (Press CTRL+C to quit)") + self.assertTrue(plugin._is_plugin_ready) + def test_ineligible_server_cannot_resume(self): plugin = self._make_manager(dauth_oracle=False) plugin.on_init() From 47580169bbc06d6c1094b28abd8ff16ea127c3b4 Mon Sep 17 00:00:00 2001 From: Alessandro Date: Mon, 3 Aug 2026 18:31:09 +0200 Subject: [PATCH 8/9] refactor: simplify dAuth registry refresh --- AGENTS.md | 2 +- extensions/business/dauth/dauth_manager.py | 44 +-------------- .../dauth/test_dauth_registry_gating.py | 56 ------------------- 3 files changed, 2 insertions(+), 100 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 53c2534e..35335263 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -719,6 +719,6 @@ Entry format: - 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 or 30-second timed-out reads clear cached peers, fail closed, and retry after one minute. Timed-out lookup results are isolated from authorization state, and at most two lookup workers may remain pending so one abandoned call cannot block recovery or cause unbounded thread growth. 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. +- 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` diff --git a/extensions/business/dauth/dauth_manager.py b/extensions/business/dauth/dauth_manager.py index 50ffc6bc..61c9e931 100644 --- a/extensions/business/dauth/dauth_manager.py +++ b/extensions/business/dauth/dauth_manager.py @@ -20,8 +20,6 @@ """ -import threading - 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 @@ -54,8 +52,6 @@ 'DAUTH_JOB_SECRETS_HSYNC_INTERVAL': 10 * 60, 'DAUTH_REGISTRY_REFRESH_INTERVAL': 60 * 60, 'DAUTH_REGISTRY_REFRESH_RETRY_INTERVAL': 60, - 'DAUTH_REGISTRY_REFRESH_TIMEOUT': 30, - 'DAUTH_REGISTRY_MAX_PENDING_LOOKUPS': 2, 'SUPRESS_LOGS_AFTER_INTERVAL' : 300, @@ -124,7 +120,6 @@ def __init__(self, **kwargs): self._dauth_registry_internal_peers = None self._last_dauth_registry_refresh = None self._dauth_registry_refresh_failed = False - self._dauth_registry_lookup_threads = [] self._last_dauth_job_secrets_hsync = None self._dauth_web_app_initialized = False self._dauth_pause_teardown_succeeded = True @@ -179,7 +174,7 @@ def _refresh_dauth_registry(self, force=False): error = None try: - peers, eth_oracles = self._load_dauth_registry_snapshot_with_timeout() + peers, eth_oracles = load_dauth_registry_snapshot(self) enabled = self.bc.eth_address.lower() in [ address.lower() for address in eth_oracles ] @@ -210,43 +205,6 @@ def _refresh_dauth_registry(self, force=False): # endif return enabled - def _load_dauth_registry_snapshot_with_timeout(self): - lookup_threads = [ - thread for thread in getattr(self, "_dauth_registry_lookup_threads", []) - if thread.is_alive() - ] - self._dauth_registry_lookup_threads = lookup_threads - if len(lookup_threads) >= self.cfg_dauth_registry_max_pending_lookups: - raise TimeoutError("too many dAuth registry lookups are still running") - # endif - - result = {} - - def load_registry(): - try: - result["snapshot"] = load_dauth_registry_snapshot(self) - except Exception as exc: - result["error"] = exc - # end try - return - - lookup_thread = threading.Thread(target=load_registry, daemon=True) - self._dauth_registry_lookup_threads.append(lookup_thread) - lookup_thread.start() - lookup_thread.join(timeout=self.cfg_dauth_registry_refresh_timeout) - if lookup_thread.is_alive(): - raise TimeoutError( - f"dAuth registry lookup timed out after " - f"{self.cfg_dauth_registry_refresh_timeout} seconds" - ) - # endif - - self._dauth_registry_lookup_threads.remove(lookup_thread) - error = result.get("error") - if error is not None: - raise error - return result["snapshot"] - def _is_dauth_server_enabled(self): return getattr(self, "_dauth_server_enabled", None) is True diff --git a/extensions/business/dauth/test_dauth_registry_gating.py b/extensions/business/dauth/test_dauth_registry_gating.py index 31835093..eac59129 100644 --- a/extensions/business/dauth/test_dauth_registry_gating.py +++ b/extensions/business/dauth/test_dauth_registry_gating.py @@ -575,8 +575,6 @@ def get_eth_dauth_oracles(self): self.calls += 1 if isinstance(self.result, Exception): raise self.result - if callable(self.result): - return self.result() if isinstance(self.result, list): return self.result return ["0xNODE", "0xPEER"] if self.result else ["0xPEER"] @@ -610,13 +608,10 @@ def eth_addr_to_internal_addr(self, eth_address): plugin._dauth_registry_internal_peers = None plugin._last_dauth_registry_refresh = None plugin._dauth_registry_refresh_failed = False - plugin._dauth_registry_lookup_threads = [] plugin._last_dauth_job_secrets_hsync = None plugin.cfg_dauth_job_secrets_hsync_interval = 10 * 60 plugin.cfg_dauth_registry_refresh_interval = 60 * 60 plugin.cfg_dauth_registry_refresh_retry_interval = 60 - plugin.cfg_dauth_registry_refresh_timeout = 30 - plugin.cfg_dauth_registry_max_pending_lookups = 2 plugin._is_plugin_ready = None plugin._hsync_calls = [] plugin.chainstore_hsync = lambda **kwargs: plugin._hsync_calls.append(kwargs) or { @@ -754,57 +749,6 @@ def test_startup_lookup_error_fails_closed_and_retries_after_one_minute(self): self.assertTrue(plugin.should_pause()) self.assertEqual(plugin.bc.calls, 2) - def test_registry_lookup_timeout_fails_closed_without_late_state_update(self): - lookup_release = threading.Event() - - def delayed_registry_lookup(): - lookup_release.wait() - return ["0xNODE", "0xPEER"] - - plugin = self._make_manager(dauth_oracle=delayed_registry_lookup) - plugin.cfg_dauth_registry_refresh_timeout = 0.001 - - plugin.on_init() - - self.assertFalse(plugin._is_dauth_server_enabled()) # pylint: disable=protected-access - self.assertIn("timed out", plugin._dauth_server_enabled_message) - self.assertEqual(len(plugin._dauth_registry_lookup_threads), 1) - self.assertTrue(plugin._dauth_registry_lookup_threads[0].is_alive()) - - plugin.bc.result = True - plugin._now += 60 - self.assertTrue(plugin.should_resume()) - self.assertEqual(plugin.bc.calls, 2) - - lookup_release.set() - plugin._dauth_registry_lookup_threads[0].join(timeout=1) - self.assertTrue(plugin._is_dauth_server_enabled()) # pylint: disable=protected-access - - def test_registry_lookup_timeouts_cap_abandoned_workers(self): - lookup_release = threading.Event() - - def blocked_registry_lookup(): - lookup_release.wait() - return ["0xNODE", "0xPEER"] - - plugin = self._make_manager(dauth_oracle=blocked_registry_lookup) - plugin.cfg_dauth_registry_refresh_timeout = 0.001 - plugin.on_init() - - plugin._now += 60 - self.assertFalse(plugin.should_resume()) - plugin._now += 60 - self.assertFalse(plugin.should_resume()) - - self.assertEqual(plugin.bc.calls, 2) - self.assertEqual(len(plugin._dauth_registry_lookup_threads), 2) - self.assertIn("too many", plugin._dauth_server_enabled_message) - - lookup_release.set() - for lookup_thread in plugin._dauth_registry_lookup_threads: - lookup_thread.join(timeout=1) - # endfor - def test_hourly_refresh_revokes_server_and_secret_replication(self): plugin = self._make_manager(dauth_oracle=True) plugin.on_init() From f25a5351579da761f708697ab97820fe2ede9cc2 Mon Sep 17 00:00:00 2001 From: Alessandro Date: Mon, 3 Aug 2026 19:19:53 +0200 Subject: [PATCH 9/9] chore: inc ver --- ver.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/ver.py b/ver.py index bacaa6bb..cdbe10e7 100644 --- a/ver.py +++ b/ver.py @@ -1 +1 @@ -__VER__ = '2.10.401' +__VER__ = '2.10.402'