fix: validate summary-backed NetMon control paths (#437)

* fix: validate summary-backed NetMon control paths

What changed:
- add local NetMon API probe and comms testbed validation coverage
- use control-liveness APIs at edge call sites that depend on summary-backed visibility
- fail closed on unknown supervisor state during Deeploy target selection

Why:
- normal nodes need safe command/config visibility from summaries without treating summaries as heartbeat evidence

* Bump version from 2.10.312 to 2.10.313

Author: cristibleotiu

docker-compose_comms.yaml

@@ -29,6 +29,9 @@ x-edge-comms-node: &edge-comms-node
     EE_MQTT_COMMAND_QOS: "2"
     EE_NETMON_ORACLE_ONLY_HEARTBEAT_MODE: "1"
     EE_ENABLE_NETMON_API_PROBE: "1"
+    # The comms testbed does not run child workloads, so DinD TLS only adds a
+    # local certificate-validity dependency that can stop worker startup.
+    DOCKER_TLS_CERTDIR: ""
     # Only the isolated non-EVM comms testbed may trust supervisor-marked
     # NET_MON_01 summaries without a blockchain oracle registry.
     EE_NETMON_ACCEPT_LOCAL_SUPERVISOR_SUMMARY: "1"

extensions/business/deeploy/deeploy_mixin.py

@@ -87,6 +87,19 @@ def _normalize_node_specs_address(self, node_addr):
     return raw_addr if raw_addr.startswith("0xai_") else f"0xai_{raw_addr}"
 
 
+  def _netmon_node_is_online_for_control(self, node_addr):
+    """
+    Use NetMon's explicit command/control liveness predicate when available.
+
+    Summary-backed liveness is only a preflight for control/deploy requests; the
+    actual mutation still depends on command dispatch and response handling.
+    """
+    checker = getattr(self.netmon, "network_node_is_online_for_control", None)
+    if callable(checker):
+      return checker(node_addr)
+    return self.netmon.network_node_is_online(node_addr)
+
+
   def _get_node_specs(self, target_nodes):
     """
     Return total and live-available CPU, memory, and disk specs for nodes.
@@ -107,7 +120,7 @@ def _get_node_specs(self, target_nodes):
       try:
         result[node_addr] = {
           "node_alias": self.netmon.network_node_eeid(node_addr),
-          "node_is_online": self.netmon.network_node_is_online(node_addr),
+          "node_is_online": self._netmon_node_is_online_for_control(node_addr),
           "cpu": {
             "total": self._node_specs_number(self.netmon.network_node_total_cpu_cores(node_addr)),
             "available": self._node_specs_number(self.netmon.network_node_avail_cpu_cores(node_addr)),

extensions/business/deeploy/deeploy_target_nodes_mixin.py

@@ -1,3 +1,5 @@
+import math
+
 from naeural_core.main.net_mon import NetMonCt
 
 from extensions.business.deeploy.deeploy_const import (
@@ -73,6 +75,89 @@ def _parse_memory(self, mem):
     else:
       return int(float(mem))  # assume bytes
 
+  def _parse_node_telemetry_cpu(self, value):
+    """
+    Parse node CPU telemetry. Missing or malformed node telemetry must fail
+    closed as zero capacity instead of crashing deployment preflight.
+    """
+    try:
+      cpu = self._parse_cpu_value(value, default=0.0)
+    except ValueError:
+      return 0.0
+    if cpu is None or not math.isfinite(cpu) or cpu < 0:
+      return 0.0
+    return cpu
+
+  def _parse_node_telemetry_memory_bytes(self, value):
+    """
+    Parse node memory telemetry. NetMon reports memory in GB, but mixed-version
+    or malformed summaries must fail closed as zero capacity, not abort deploy.
+    """
+    try:
+      memory_bytes = self._parse_memory(value)
+    except (AttributeError, OverflowError, TypeError, ValueError):
+      return 0
+    if not math.isfinite(memory_bytes) or memory_bytes < 0:
+      return 0
+    return int(memory_bytes)
+
+  def _parse_node_telemetry_bytes(self, value):
+    """
+    Parse byte-valued node telemetry with the same fail-closed policy.
+    """
+    try:
+      parsed = float(value)
+    except (TypeError, ValueError):
+      return 0
+    if not math.isfinite(parsed) or parsed < 0:
+      return 0
+    return int(parsed)
+
+  def _parse_node_telemetry_bool(self, value):
+    """
+    Parse boolean node telemetry with a fail-closed policy. This protects
+    Deeploy against mixed-version NetMon summaries returning "false"/"0" as
+    strings, which Python would otherwise treat as truthy.
+    """
+    if isinstance(value, bool):
+      return value
+    if isinstance(value, str):
+      value = value.strip().lower()
+      if value in {"1", "true", "yes", "y"}:
+        return True
+      if value in {"0", "false", "no", "n", ""}:
+        return False
+      return False
+    if isinstance(value, (int, float)) and not isinstance(value, bool):
+      return math.isfinite(value) and value == 1
+    return False
+
+
+  def _parse_node_supervisor_state(self, value):
+    """
+    Parse supervisor telemetry for target selection.
+
+    Unlike generic capability booleans, unknown supervisor state must not be
+    coerced to False: deploying to a supervisor is unsafe, so missing/malformed
+    summary data is treated as "unknown" and filtered out by callers.
+    """
+    if isinstance(value, bool):
+      return value
+    if isinstance(value, str):
+      value = value.strip().lower()
+      if value == "":
+        return None
+      if value in {"1", "true", "yes", "y"}:
+        return True
+      if value in {"0", "false", "no", "n"}:
+        return False
+      return None
+    if isinstance(value, (int, float)) and not isinstance(value, bool):
+      if math.isfinite(value) and value in (0, 1):
+        return bool(value)
+      return None
+    return None
+
   def _get_request_plugin_signatures_from_pipeline(self, inputs):
     """
     Extract plugin signatures from normalized request payload.
@@ -268,14 +353,16 @@ def __find_suitable_nodes_for_container_app(self, nodes_with_resources, containe
       self.Pd(f"Node {addr} projected usage (with new app): CPU={used_cpu} cores, Memory={used_memory} bytes")
 
       # Check if the node has enough resources
-      self.Pd(f"Node {addr} total resources: CPU={node_resources['cpu']} cores, Memory={node_resources['memory']} bytes")
+      node_cpu = self._parse_node_telemetry_cpu(node_resources.get('cpu'))
+      node_memory = self._parse_node_telemetry_bytes(node_resources.get('memory'))
+      self.Pd(f"Node {addr} total resources: CPU={node_cpu} cores, Memory={node_memory} bytes")
       has_failed = False
-      if used_cpu > node_resources['cpu']:
-        self.Pd(f"Node {addr} has not enough CPU cores. used_cpu ({used_cpu}) > node_cpu ({node_resources['cpu']})")
+      if used_cpu > node_cpu:
+        self.Pd(f"Node {addr} has not enough CPU cores. used_cpu ({used_cpu}) > node_cpu ({node_cpu})")
         has_failed = True
 
-      if used_memory > node_resources['memory']:
-        self.Pd(f"Node {addr} has not enough RAM. used_memory ({used_memory}) > node_memory ({node_resources['memory']})")
+      if used_memory > node_memory:
+        self.Pd(f"Node {addr} has not enough RAM. used_memory ({used_memory}) > node_memory ({node_memory})")
         has_failed = True
 
       if has_failed:
@@ -371,7 +458,7 @@ def __check_nodes_capabilities_and_extract_resources(self, nodes: list['str'], i
     for addr in nodes:
       # Check if the node supports the requested plugin
       if requires_container_capabilities:
-        is_did_supported = self.netmon.network_node_has_did(addr=addr)
+        is_did_supported = self._parse_node_telemetry_bool(self.netmon.network_node_has_did(addr=addr))
         if not is_did_supported:
           self.Pd(f"Node {addr} does not support container deployments. Skipping...")
           continue
@@ -383,23 +470,24 @@ def __check_nodes_capabilities_and_extract_resources(self, nodes: list['str'], i
           continue
 
       self.Pd(f"Node {addr} cont in function.")
-      total_cpu = self.netmon.network_node_total_cpu_cores(addr)
+      total_cpu = self._parse_node_telemetry_cpu(self.netmon.network_node_total_cpu_cores(addr))
 
       total_memory = self.netmon.network_node_total_mem(addr)
-      total_memory_bytes = self._parse_memory(total_memory)
+      total_memory_bytes = self._parse_node_telemetry_memory_bytes(total_memory)
 
       current_node_total_resources = {
         'cpu': total_cpu,
         'memory': total_memory_bytes,
       }
 
       if node_res_req:
+        node_req_cpu = self._parse_cpu_value(node_req_cpu, default=None)
 
-        if total_cpu < node_req_cpu:
+        if node_req_cpu is not None and total_cpu < node_req_cpu:
           self.Pd(f"Node {addr} has not enough CPU cores in total. Skipping...")
           continue
 
-        if total_memory_bytes < node_req_memory_bytes:
+        if node_req_memory_bytes > 0 and total_memory_bytes < node_req_memory_bytes:
           self.Pd(f"Node {addr} has not enought RAM in total. Skipping...")
           continue
 
@@ -426,8 +514,9 @@ def _find_nodes_for_deeployment(self, inputs):
     for addr, value in network_nodes.items():
       ai_addr = self.bc.maybe_add_prefix(addr)
 
-      is_online = self.netmon.network_node_is_online(ai_addr)
-      if value.get('is_supervisor') is True or not is_online:
+      is_online = self._netmon_node_is_online_for_control(ai_addr)
+      is_supervisor = self._parse_node_supervisor_state(self.netmon.network_node_is_supervisor(ai_addr))
+      if is_supervisor is not False or not is_online:
       # FIXME: Disabled for now, as the most of the nodes are are marked as non-trusted.
       # if value.get('is_supervisor') is True or not value.get('trusted', False) or not is_online:
         continue
@@ -537,11 +626,11 @@ def _check_nodes_availability(self, inputs, skip_resource_check=False):
 
     for node in inputs.target_nodes:
       addr = self._check_and_maybe_convert_address(node)
-      is_supervisor = self.netmon.network_node_is_supervisor(addr=addr)
-      if is_supervisor:
-        msg = f"{DEEPLOY_ERRORS.NODES6}: Node {addr} is a supervisor node and cannot be used for deeployment"
+      is_supervisor = self._parse_node_supervisor_state(self.netmon.network_node_is_supervisor(addr=addr))
+      if is_supervisor is not False:
+        msg = f"{DEEPLOY_ERRORS.NODES6}: Node {addr} is a supervisor node or has unknown supervisor state and cannot be used for deeployment"
         raise ValueError(msg)
-      is_online = self.netmon.network_node_is_online(addr)
+      is_online = self._netmon_node_is_online_for_control(addr)
       if is_online:
         if skip_resources:
           self.Pd(f"Skipping resource validation for node {addr}")
@@ -588,15 +677,18 @@ def check_node_available_resources(self, addr, inputs):
       return result
 
     # Get available resources
-    avail_cpu = self.netmon.network_node_get_cpu_avail_cores(addr)
+    avail_cpu = self._parse_node_telemetry_cpu(self.netmon.network_node_get_cpu_avail_cores(addr))
     avail_mem = self.netmon.network_node_available_memory(addr)  # in GB
-    avail_mem_bytes = self._parse_memory(f"{avail_mem}g")
-    avail_disk = self.netmon.network_node_available_disk(addr)  # in bytes
+    avail_mem_bytes = self._parse_node_telemetry_memory_bytes(avail_mem)
+    avail_disk = self._parse_node_telemetry_bytes(self.netmon.network_node_available_disk(addr))  # in bytes
 
     # Get required resources from the request
     required_resources = self._aggregate_container_resources(inputs) or {}
     required_mem = required_resources.get(DEEPLOY_RESOURCES.MEMORY, DEFAULT_CONTAINER_RESOURCES.MEMORY)
-    required_cpu = required_resources.get(DEEPLOY_RESOURCES.CPU, DEFAULT_CONTAINER_RESOURCES.CPU)
+    required_cpu = self._parse_cpu_value(
+      required_resources.get(DEEPLOY_RESOURCES.CPU, DEFAULT_CONTAINER_RESOURCES.CPU),
+      default=DEFAULT_CONTAINER_RESOURCES.CPU,
+    )
 
     required_mem_bytes = self._parse_memory(required_mem)
 

extensions/business/deeploy/tests/test_node_specs.py

@@ -21,14 +21,27 @@ def __init__(self):
       "0xai_node_beta": {
         "alias": "beta",
         "online": False,
+        "control_online": False,
         "cpu_total": 4,
         "cpu_avail": 1,
         "mem_total": 16,
         "mem_avail": 8,
         "disk_total": 256,
         "disk_avail": 120,
       },
+      "0xai_node_gamma": {
+        "alias": "gamma",
+        "online": False,
+        "control_online": True,
+        "cpu_total": 12,
+        "cpu_avail": 9,
+        "mem_total": 64,
+        "mem_avail": 40,
+        "disk_total": 1024,
+        "disk_avail": 900,
+      },
     }
+    self.nodes["0xai_node_alpha"]["control_online"] = self.nodes["0xai_node_alpha"]["online"]
 
   def _get(self, addr, key):
     if addr not in self.nodes:
@@ -41,6 +54,9 @@ def network_node_eeid(self, addr):
   def network_node_is_online(self, addr):
     return self._get(addr, "online")
 
+  def network_node_is_online_for_control(self, addr):
+    return self._get(addr, "control_online")
+
   def network_node_total_cpu_cores(self, addr):
     return self._get(addr, "cpu_total")
 
@@ -84,6 +100,18 @@ def test_builds_node_specs_for_unique_requested_nodes(self):
     )
     self.assertEqual(specs["0xai_node_beta"]["node_is_online"], False)
 
+  def test_node_specs_use_control_liveness_for_summary_backed_nodes(self):
+    plugin = make_deeploy_plugin()
+    plugin.netmon = _NetmonStub()
+    plugin.bc = SimpleNamespace(
+      maybe_add_prefix=lambda addr: addr if str(addr).startswith("0xai_") else f"0xai_{addr}"
+    )
+
+    specs = plugin._get_node_specs(["node_gamma"])
+
+    self.assertFalse(plugin.netmon.network_node_is_online("0xai_node_gamma"))
+    self.assertTrue(specs["0xai_node_gamma"]["node_is_online"])
+
   def test_returns_per_node_error_without_failing_entire_specs_request(self):
     plugin = make_deeploy_plugin()
     plugin.netmon = _NetmonStub()