feat: add support for escrow delegates (#330)

Author: aledefra

extensions/business/deeploy/deeploy_const.py

@@ -52,9 +52,8 @@ class DEEPLOY_KEYS:
   PLUGIN_INSTANCE_ID = "instance_id"
   # Auth result keys
   SENDER = "sender"
-  SENDER_ORACLES = "sender_oracles"
-  SENDER_NODES_COUNT = "sender_nodes_count"
-  SENDER_TOTAL_COUNT = "sender_total_count"
+  SENDER_ESCROW = "sender_escrow"
+  ESCROW_OWNER = "escrow_owner"
 
   # Config keys
   DATE_UPDATED = "date_updated"

extensions/business/deeploy/deeploy_manager_api.py

@@ -122,9 +122,7 @@ def get_apps(
       sender, inputs = self.deeploy_verify_and_get_inputs(request)
       auth_result = self.deeploy_get_auth_result(inputs)
 
-      apps = self._get_online_apps(owner=sender)
-      
-      # TODO: (Vitalii) filter apps by the sender address (OWNER)
+      apps = self._get_online_apps(owner=auth_result[DEEPLOY_KEYS.ESCROW_OWNER])
 
       result = {
         DEEPLOY_KEYS.STATUS : DEEPLOY_STATUS.SUCCESS,
@@ -206,7 +204,7 @@ def _process_pipeline_request(
           raise ValueError(msg)
 
       # check payment
-      is_valid = self.deeploy_check_payment_and_job_owner(inputs, sender, is_create=is_create, debug=self.cfg_deeploy_verbose > 1)
+      is_valid = self.deeploy_check_payment_and_job_owner(inputs, auth_result[DEEPLOY_KEYS.ESCROW_OWNER], is_create=is_create, debug=self.cfg_deeploy_verbose > 1)
       if not is_valid:
         msg = f"{DEEPLOY_ERRORS.PAYMENT1}: The request job is not paid, or the job is not sent by the job owner."
         raise ValueError(msg)
@@ -225,7 +223,7 @@ def _process_pipeline_request(
       else:
         # Discover the live deployment so we can validate node affinity and reuse existing specs.
         pipeline_context = self._gather_running_pipeline_context(
-          owner=sender,
+          owner=auth_result[DEEPLOY_KEYS.ESCROW_OWNER],
           app_id=app_id,
           job_id=job_id,
         )
@@ -277,7 +275,7 @@ def _process_pipeline_request(
         self._ensure_plugin_instance_ids(
           inputs,
           discovered_plugin_instances=discovered_plugin_instances,
-          owner=sender,
+          owner=auth_result[DEEPLOY_KEYS.ESCROW_OWNER],
           app_id=app_id,
           job_id=job_id,
         )
@@ -300,7 +298,7 @@ def _process_pipeline_request(
         self.delete_pipeline_from_nodes(
           app_id=app_id,
           job_id=job_id,
-          owner=sender,
+          owner=auth_result[DEEPLOY_KEYS.ESCROW_OWNER],
           discovered_instances=discovered_plugin_instances,
         )
 
@@ -331,7 +329,7 @@ def _process_pipeline_request(
       )
 
       dct_status, str_status = self.check_and_deploy_pipelines(
-        sender=sender,
+        owner=auth_result[DEEPLOY_KEYS.ESCROW_OWNER],
         inputs=inputs,
         app_id=app_id,
         app_alias=app_alias,
@@ -629,27 +627,27 @@ def scale_up_job_workers(self,
       is_confirmable_job = inputs.chainstore_response
 
       # check payment
-      is_valid = self.deeploy_check_payment_and_job_owner(inputs, sender, is_create=False, debug=self.cfg_deeploy_verbose > 1)
+      is_valid = self.deeploy_check_payment_and_job_owner(inputs, auth_result[DEEPLOY_KEYS.ESCROW_OWNER], is_create=False, debug=self.cfg_deeploy_verbose > 1)
       if not is_valid:
         msg = f"{DEEPLOY_ERRORS.PAYMENT1}: The request job is not paid, or the job is not sent by the job owner."
         raise ValueError(msg)
 
-      running_apps_for_job = self._get_online_apps(job_id=job_id, owner=sender)
+      running_apps_for_job = self._get_online_apps(job_id=job_id, owner=auth_result[DEEPLOY_KEYS.ESCROW_OWNER])
 
       # todo: check the count of running workers and compare with the amount of allowed workers count from blockchain.
 
       self.P(f"Discovered running apps for job: {self.json_dumps(running_apps_for_job)}")
 
       if not running_apps_for_job or not len(running_apps_for_job):
-        msg = f"{DEEPLOY_ERRORS.NODES3}: No running workers found for provided job_id and owner '{sender}'."
+        msg = f"{DEEPLOY_ERRORS.NODES3}: No running workers found for provided job_id and owner '{auth_result[DEEPLOY_KEYS.ESCROW_OWNER]}'."
         raise ValueError(msg)
 
       update_nodes = list(running_apps_for_job.keys())
       new_nodes = self._check_nodes_availability(inputs)
 
       dct_status, str_status = self.scale_up_job(new_nodes=new_nodes, 
                                                  update_nodes=update_nodes, 
-                                                 sender=sender, 
+                                                 owner=auth_result[DEEPLOY_KEYS.ESCROW_OWNER], 
                                                  job_id=job_id,
                                                  running_apps_for_job=running_apps_for_job)
 
@@ -718,7 +716,7 @@ def delete_pipeline(self,
       job_id = inputs.get(DEEPLOY_KEYS.JOB_ID, None)
       app_id = inputs.get(DEEPLOY_KEYS.APP_ID, None)
 
-      discovered_instances = self.delete_pipeline_from_nodes(app_id=app_id, job_id=job_id, owner=sender)
+      discovered_instances = self.delete_pipeline_from_nodes(app_id=app_id, job_id=job_id, owner=auth_result[DEEPLOY_KEYS.ESCROW_OWNER])
       request_payload = {
         DEEPLOY_KEYS.STATUS: DEEPLOY_STATUS.SUCCESS,
         DEEPLOY_KEYS.TARGETS: discovered_instances,
@@ -778,7 +776,7 @@ def send_instance_command(self,
       # Validate the request fields.
       self._validate_send_instance_command_request(inputs)
 
-      self.send_instance_command_to_nodes(inputs, owner=sender)
+      self.send_instance_command_to_nodes(inputs, owner=auth_result[DEEPLOY_KEYS.ESCROW_OWNER])
 
       result = {
         DEEPLOY_KEYS.REQUEST : {
@@ -831,7 +829,7 @@ def send_app_command(self,
       # Validate the request fields.
       self._validate_send_app_command_request(inputs)
 
-      discovered_pipelines = self.discover_and_send_instance_command(inputs, owner=sender)
+      discovered_pipelines = self.discover_and_send_instance_command(inputs, owner=auth_result[DEEPLOY_KEYS.ESCROW_OWNER])
       targets = []
       for discovered_pipeline in discovered_pipelines:
         targets.append([discovered_pipeline[DEEPLOY_PLUGIN_DATA.NODE],

extensions/business/deeploy/deeploy_mixin.py

@@ -80,27 +80,11 @@ def _check_plugin_signature(self, signature: str):
 
   def __check_allowed_wallet(self, inputs):
     sender = inputs.get(BASE_CT.BCctbase.ETH_SENDER)
-    eth_nodes = self.bc.get_wallet_nodes(sender)
-    if len(eth_nodes) == 0:
-      raise ValueError("No nodes found for wallet {}".format(sender))
-    eth_oracles = self.bc.get_eth_oracles()
-    if len(eth_oracles) == 0:
-      raise ValueError("No oracles found - this is a critical issue!")
-    oracle_found = False
-    wallet_oracles = []
-    wallet_nodes = []
-    for node in eth_nodes:
-      if node in eth_oracles:
-        oracle_found = True
-        wallet_oracles.append(node)
-      else:
-        wallet_nodes.append(node)
-      #endif 
-    #endfor each node
-    if not oracle_found:
-      raise ValueError("No oracles found for wallet {}".format(sender))
-    inputs.wallet_nodes = wallet_nodes
-    inputs.wallet_oracles = wallet_oracles
+    escrow_details = self.bc.get_user_escrow_details(sender)
+    if not escrow_details.get('isActive', False):
+      raise ValueError("Wallet {} has no active escrow".format(sender))
+    inputs.wallet_escrow = escrow_details['escrowAddress']
+    inputs.wallet_escrow_owner = escrow_details['escrowOwner']
     return inputs
 
   def __check_is_oracle(self, inputs):
@@ -112,7 +96,7 @@ def __check_is_oracle(self, inputs):
       raise ValueError("Sender {} is not an oracle".format(sender))
     return True
 
-  def __create_pipeline_on_nodes(self, nodes, inputs, app_id, app_alias, app_type, sender, job_app_type=None, dct_deeploy_specs=None):
+  def __create_pipeline_on_nodes(self, nodes, inputs, app_id, app_alias, app_type, owner, job_app_type=None, dct_deeploy_specs=None):
     """
     Create new pipelines on each node and set CSTORE `response_key` for the "callback" action
     """
@@ -223,7 +207,7 @@ def __create_pipeline_on_nodes(self, nodes, inputs, app_id, app_alias, app_type,
           app_alias=app_alias,
           pipeline_type=app_type,
           node_address=addr,
-          owner=sender, 
+          owner=owner, 
           url=inputs.pipeline_input_uri,
           plugins=node_plugins,
           is_deeployed=True,
@@ -243,7 +227,7 @@ def __create_pipeline_on_nodes(self, nodes, inputs, app_id, app_alias, app_type,
     cleaned_response_keys = prepared_response_keys if inputs.chainstore_response else {}
     return cleaned_response_keys
 
-  def __update_pipeline_on_nodes(self, nodes, inputs, app_id, app_alias, app_type, sender, discovered_plugin_instances, dct_deeploy_specs = None, job_app_type=None):
+  def __update_pipeline_on_nodes(self, nodes, inputs, app_id, app_alias, app_type, owner, discovered_plugin_instances, dct_deeploy_specs = None, job_app_type=None):
     """
     Create new pipelines on each node and set CSTORE `response_key` for the "callback" action
     """
@@ -447,7 +431,7 @@ def __update_pipeline_on_nodes(self, nodes, inputs, app_id, app_alias, app_type,
           app_alias=app_alias,
           pipeline_type=app_type,
           node_address=addr,
-          owner=sender,
+          owner=owner,
           url=inputs.pipeline_input_uri,
           plugins=node_plugins,
           is_deeployed=True,
@@ -694,7 +678,7 @@ def deeploy_verify_and_get_inputs(self, request: dict, require_sender_is_oracle:
     if addr.lower() != sender.lower():
       raise ValueError("Invalid signature: recovered {} != {}".format(addr, sender))
 
-    # Check if the sender is allowed to create pipelines
+    # Check if the sender is allowed to make the request
     if require_sender_is_oracle:
       self.__check_is_oracle(inputs)
     else:
@@ -1098,9 +1082,8 @@ def deeploy_get_auth_result(self, inputs):
     result = {
       DEEPLOY_KEYS.SENDER: sender,
       DEEPLOY_KEYS.NONCE: self.deeploy_get_nonce(inputs.nonce),
-      DEEPLOY_KEYS.SENDER_ORACLES: inputs.wallet_oracles,
-      DEEPLOY_KEYS.SENDER_NODES_COUNT: len(inputs.wallet_nodes),
-      DEEPLOY_KEYS.SENDER_TOTAL_COUNT: len(inputs.wallet_nodes) + len(inputs.wallet_oracles),
+      DEEPLOY_KEYS.SENDER_ESCROW: inputs.wallet_escrow,
+      DEEPLOY_KEYS.ESCROW_OWNER: inputs.wallet_escrow_owner,
     }
     return result
 
@@ -1211,12 +1194,12 @@ def _organize_requested_plugins(self, inputs):
     return plugins_by_instance_id, plugins_by_signature, new_plugin_configs
   # TODO: END FIXME
 
-  def deeploy_check_payment_and_job_owner(self, inputs, sender, is_create, debug=False):
+  def deeploy_check_payment_and_job_owner(self, inputs, owner, is_create, debug=False):
     """
     Check if the payment is valid for the given job.
     """
     self.Pd(f"=== deeploy_check_payment_and_job_owner ===")
-    self.Pd(f"  sender: {sender}")
+    self.Pd(f"  owner: {owner}")
     self.Pd(f"  is_create: {is_create}")
     self.Pd(f"  debug: {debug}")
 
@@ -1229,7 +1212,7 @@ def deeploy_check_payment_and_job_owner(self, inputs, sender, is_create, debug=F
       return True
 
     job_id = inputs.get(DEEPLOY_KEYS.JOB_ID, None)
-    self.Pd(f"Checking payment for job {job_id} by sender {sender}{' (debug mode)' if debug else ''}")
+    self.Pd(f"Checking payment for job {job_id} by owner {owner}{' (debug mode)' if debug else ''}")
 
     if not job_id:
       self.Pd("  No job_id provided - validation failed")
@@ -1251,8 +1234,8 @@ def deeploy_check_payment_and_job_owner(self, inputs, sender, is_create, debug=F
     self.Pd(f"  Job owner: {job_owner}")
     self.Pd(f"  Start timestamp: {start_timestamp}")
 
-    is_valid = (sender == job_owner) if sender and job_owner else False
-    self.Pd(f"  Owner match: {is_valid} (sender={sender}, owner={job_owner})")
+    is_valid = (owner == job_owner) if owner and job_owner else False
+    self.Pd(f"  Owner match: {is_valid} (owner={owner}, job_owner={job_owner})")
 
     if is_create and start_timestamp:
       self.Pd(f"  Job already started (timestamp={start_timestamp}) but is_create=True - invalidating")
@@ -1741,7 +1724,7 @@ def deeploy_prepare_plugins(self, inputs):
     plugins = [plugin]
     return plugins
 
-  def check_and_deploy_pipelines(self, sender, inputs, app_id, app_alias, app_type, update_nodes, new_nodes, discovered_plugin_instances=[], dct_deeploy_specs=None, job_app_type=None, dct_deeploy_specs_create=None):
+  def check_and_deploy_pipelines(self, owner, inputs, app_id, app_alias, app_type, update_nodes, new_nodes, discovered_plugin_instances=[], dct_deeploy_specs=None, job_app_type=None, dct_deeploy_specs_create=None):
     """
     Validate the inputs and deploy the pipeline on the target nodes.
     """
@@ -1754,10 +1737,10 @@ def check_and_deploy_pipelines(self, sender, inputs, app_id, app_alias, app_type
     # Phase 2: Launch the pipeline on each node and set CSTORE `response_key`` for the "callback" action
     response_keys = {}
     if len(update_nodes) > 0:
-      update_response_keys = self.__update_pipeline_on_nodes(update_nodes, inputs, app_id, app_alias, app_type, sender, discovered_plugin_instances, dct_deeploy_specs, job_app_type=job_app_type)
+      update_response_keys = self.__update_pipeline_on_nodes(update_nodes, inputs, app_id, app_alias, app_type, owner, discovered_plugin_instances, dct_deeploy_specs, job_app_type=job_app_type)
       response_keys.update(update_response_keys)
     if len(new_nodes) > 0:
-      new_response_keys = self.__create_pipeline_on_nodes(new_nodes, inputs, app_id, app_alias, app_type, sender, job_app_type=job_app_type, dct_deeploy_specs=dct_deeploy_specs_create)
+      new_response_keys = self.__create_pipeline_on_nodes(new_nodes, inputs, app_id, app_alias, app_type, owner, job_app_type=job_app_type, dct_deeploy_specs=dct_deeploy_specs_create)
       response_keys.update(new_response_keys)
 
     # Phase 3: Wait until all the responses are received via CSTORE and compose status response
@@ -1770,7 +1753,7 @@ def check_and_deploy_pipelines(self, sender, inputs, app_id, app_alias, app_type
 
     return dct_status, str_status
 
-  def scale_up_job(self, new_nodes, update_nodes, job_id, sender, running_apps_for_job):
+  def scale_up_job(self, new_nodes, update_nodes, job_id, owner, running_apps_for_job):
     """
     Scale up the job workers.
     """
@@ -1800,7 +1783,7 @@ def scale_up_job(self, new_nodes, update_nodes, job_id, sender, running_apps_for
     # Start pipelines on nodes.
     self._start_create_update_pipelines(create_pipelines=create_pipelines,
                                         update_pipelines=update_pipelines,
-                                        sender=sender)
+                                        owner=owner)
 
     dct_status, str_status = self._get_pipeline_responses(chainstore_response_keys, 300)
 
@@ -2267,7 +2250,7 @@ def prepare_create_update_pipelines(self, base_pipeline, new_nodes, update_nodes
 
     return create_pipelines, update_pipelines, prepared_response_keys
 
-  def _start_create_update_pipelines(self, create_pipelines, update_pipelines, sender):
+  def _start_create_update_pipelines(self, create_pipelines, update_pipelines, owner):
     """
     Start the create and update pipelines.
     """
@@ -2283,7 +2266,7 @@ def _start_create_update_pipelines(self, create_pipelines, update_pipelines, sen
         name=pipeline['app_id'],
         pipeline_type=pipeline['pipeline_type'],
         node_address=node,
-        owner=sender, 
+        owner=owner, 
         url=pipeline.get('url'),
         plugins=pipeline['plugins'],
         is_deeployed=True,

ver.py

@@ -1 +1 @@
-__VER__ = '2.9.940'
+__VER__ = '2.9.941'