Add (script add) and restrict (script remove) to owners (#1025)

Author: folex

Cargo.lock

@@ -21,6 +21,7 @@ use aquamarine_vm::AquamarineVM;
 use particle_protocol::Particle;
 
 use futures::FutureExt;
+use std::ops::Mul;
 use std::{
     collections::VecDeque,
     fmt::Debug,
@@ -42,11 +43,12 @@ impl Deadline {
         }
     }
 
-    pub fn is_expired(&self, now: u64) -> bool {
+    pub fn is_expired(&self, now_ms: u64) -> bool {
         self.timestamp
+            .mul(1000)
             .checked_add(self.ttl as u64)
             // Whether ts is in the past
-            .map(|ts| ts < now)
+            .map(|ts| ts < now_ms)
             // If timestamp + ttl gives overflow, consider particle expired
             .unwrap_or_else(|| {
                 log::warn!("timestamp {} + ttl {} overflowed", self.timestamp, self.ttl);
@@ -73,8 +75,8 @@ impl Actor {
         }
     }
 
-    pub fn is_expired(&self, now: u64) -> bool {
-        self.deadline.is_expired(now)
+    pub fn is_expired(&self, now_ms: u64) -> bool {
+        self.deadline.is_expired(now_ms)
     }
 
     pub fn ingest(&mut self, particle: AwaitedParticle) {

aquamarine/src/actor.rs

@@ -27,13 +27,13 @@ use std::{
 
 /// Get current time from OS
 #[cfg(not(test))]
-use real_time::now;
+use real_time::now_ms;
 
 use crate::awaited_particle::{AwaitedEffects, AwaitedParticle};
 use futures::task::Waker;
 /// For tests, mocked time is used
 #[cfg(test)]
-use mock_time::now;
+use mock_time::now_ms;
 
 pub struct Plumber {
     events: VecDeque<AwaitedEffects>,
@@ -58,7 +58,7 @@ impl Plumber {
         self.wake();
 
         let deadline = Deadline::from(&particle);
-        if deadline.is_expired(now()) {
+        if deadline.is_expired(now_ms()) {
             log::info!("Particle {} is expired, ignoring", particle.id);
             self.events.push_back(AwaitedEffects::expired(particle));
             return;
@@ -80,7 +80,7 @@ impl Plumber {
         }
 
         // Remove expired actors
-        let now = now();
+        let now = now_ms();
         self.actors.retain(|_, actor| !actor.is_expired(now));
 
         // Gather effects and put VMs back
@@ -124,8 +124,8 @@ impl Plumber {
 /// Implements `now` by taking number of non-leap seconds from `Utc::now()`
 mod real_time {
     #[allow(dead_code)]
-    pub fn now() -> u64 {
-        chrono::Utc::now().timestamp() as u64
+    pub fn now_ms() -> u64 {
+        (chrono::Utc::now().timestamp() * 1000) as u64
     }
 }
 
@@ -214,7 +214,7 @@ pub mod mock_time {
         static MOCK_TIME: RefCell<u64> = RefCell::new(0);
     }
 
-    pub fn now() -> u64 {
+    pub fn now_ms() -> u64 {
         MOCK_TIME.with(|cell| cell.borrow().clone())
     }
 

aquamarine/src/plumber.rs

@@ -158,6 +158,7 @@ impl ConnectionPoolBehaviour {
             // If particle is sent to the current node, process it locally
             self.queue.push_back(particle);
             outlet.send(true).ok();
+            self.wake();
         } else {
             // Send particle to remote peer
             self.push_event(NetworkBehaviourAction::NotifyHandler {

connection-pool/src/behaviour.rs

@@ -9,7 +9,7 @@ config-utils = { path = "../config-utils" }
 trust-graph = "0.2.0"
 particle-protocol = { path = "../../particle-protocol"}
 
-air-interpreter-wasm = "0.3.4"
+air-interpreter-wasm = "0.4.2"
 libp2p = { package = "fluence-fork-libp2p", version = "0.34.0" }
 
 serde = { version = "1.0.118", features = ["derive"] }

crates/server-config/Cargo.toml

@@ -19,7 +19,7 @@ connection-pool = { path = "../../connection-pool" }
 script-storage = { path = "../../script-storage" }
 
 stepper-interface = "0.1.2"
-air-interpreter-wasm = "0.3.4"
+air-interpreter-wasm = "0.4.2"
 aquamarine-vm = "0.1.29"
 
 libp2p = { package = "fluence-fork-libp2p", version = "0.34.0" }

crates/test-utils/Cargo.toml

@@ -25,6 +25,7 @@ use crate::{
 use fluence_client::{Client, Transport};
 use particle_protocol::Particle;
 
+use anyhow::bail;
 use async_std::task;
 use core::ops::Deref;
 use libp2p::{core::Multiaddr, PeerId};
@@ -175,4 +176,19 @@ impl ConnectedClient {
         let particle = self.receive();
         read_args(particle, &self.peer_id)
     }
+
+    /// Wait for a particle with specified `particle_id`, and read "op" "return" result from it
+    pub fn wait_particle_args(&mut self, particle_id: String) -> anyhow::Result<Vec<JValue>> {
+        let mut max = 10;
+        loop {
+            max -= 1;
+            if max <= 0 {
+                bail!("timed out waiting for particle {}", particle_id);
+            }
+            let particle = self.receive();
+            if particle.id == particle_id {
+                break Ok(read_args(particle, &self.peer_id));
+            }
+        }
+    }
 }

crates/test-utils/src/connected_client.rs

@@ -13,7 +13,7 @@
       "159.89.2.70",
       "157.230.98.75"
     ],
-    "branch": "script_storage",
+    "branch": "script_storage_list",
     "bootstrap": "138.197.177.2",
     "nodes": [
       "138.197.177.2"

deploy/deployment_config.json

@@ -37,3 +37,4 @@ parking_lot = "0.11.1"
 uuid = "0.8.2"
 chrono = "0.4.19"
 thiserror = "1.0.23"
+humantime-serde = "1.0.1"

particle-closures/Cargo.toml

@@ -28,9 +28,11 @@ use script_storage::ScriptStorageApi;
 use server_config::ServicesConfig;
 
 use async_std::task;
+use humantime_serde::re::humantime::format_duration as pretty;
 use libp2p::{core::Multiaddr, PeerId};
 use multihash::{Code, MultihashDigest};
 use serde_json::{json, Value as JValue};
+use std::borrow::Borrow;
 use std::num::ParseIntError;
 use std::time::Duration;
 use std::{str::FromStr, sync::Arc};
@@ -92,7 +94,7 @@ impl<C: Clone + Send + Sync + 'static + AsRef<KademliaApi> + AsRef<ConnectionPoo
         })
     }
 
-    fn route(&self, particle: ParticleParameters, args: Vec<IValue>) -> Option<IValue> {
+    fn route(&self, params: ParticleParameters, args: Vec<IValue>) -> Option<IValue> {
         let args = match Args::parse(args) {
             Ok(args) => args,
             Err(err) => {
@@ -118,7 +120,7 @@ impl<C: Clone + Send + Sync + 'static + AsRef<KademliaApi> + AsRef<ConnectionPoo
             ("dht", "add_provider")    => (self.add_provider)(args),
             ("dht", "get_providers")   => (self.get_providers)(args),
 
-            ("srv", "create")          => (self.create_service)(particle, args),
+            ("srv", "create")          => (self.create_service)(params, args),
             ("srv", "get_interface")   => (self.get_interface)(args),
             ("srv", "get_interfaces")  => (self.get_active_interfaces)(args),
 
@@ -127,13 +129,14 @@ impl<C: Clone + Send + Sync + 'static + AsRef<KademliaApi> + AsRef<ConnectionPoo
             ("dist", "get_modules")    => (self.get_modules)(args),
             ("dist", "get_blueprints") => (self.get_blueprints)(args),
 
-            ("script", "add")          => wrap(self.add_script(args)),
-            ("script", "remove")       => wrap(self.remove_script(args)),
+            ("script", "add")          => wrap(self.add_script(args, params)),
+            ("script", "remove")       => wrap(self.remove_script(args, params)),
+            ("script", "list")         => wrap(self.list_scripts()),
 
             ("op", "identify")         => (self.identify)(args),
             ("op", "identity")         => ok(Array(args.function_args)),
 
-            _ => (self.call_service)(particle, args),
+            _ => (self.call_service)(params, args),
         }
     }
 
@@ -174,11 +177,10 @@ impl<C: Clone + Send + Sync + 'static + AsRef<KademliaApi> + AsRef<ConnectionPoo
         Ok(contact.map(|c| json!(c)))
     }
 
-    fn add_script(&self, args: Args) -> Result<JValue, JError> {
+    fn add_script(&self, args: Args, params: ParticleParameters) -> Result<JValue, JError> {
         #[derive(thiserror::Error, Debug)]
         #[error("Error while deserializing field interval_sec: not a valid u64")]
         struct Error(#[source] ParseIntError);
-
         let mut args = args.function_args.into_iter();
 
         let script: String = Args::next("script", &mut args)?;
@@ -188,18 +190,46 @@ impl<C: Clone + Send + Sync + 'static + AsRef<KademliaApi> + AsRef<ConnectionPoo
             .transpose()
             .map_err(Error)?;
         let interval = interval.map(Duration::from_secs);
-        let id = self.script_storage.add_script(script, interval)?;
+        let creator = PeerId::from_str(&params.init_user_id)?;
+        let id = self.script_storage.add_script(script, interval, creator)?;
 
         Ok(json!(id))
     }
 
-    fn remove_script(&self, args: Args) -> Result<JValue, JError> {
-        let uuid: String = Args::next("uuid", &mut args.function_args.into_iter())?;
-        let ok = task::block_on(self.script_storage.remove_script(uuid))?;
+    fn remove_script(&self, args: Args, params: ParticleParameters) -> Result<JValue, JError> {
+        let mut args = args.function_args.into_iter();
+
+        let uuid: String = Args::next("uuid", &mut args)?;
+        let force: Option<String> = Args::maybe_next("force", &mut args)?;
+        // TODO HACK: this is a hack to allow anyone to delete any script if they know this secret
+        let force = force.map_or(false, |s| s == "--force");
+        let actor = PeerId::from_str(&params.init_user_id)?;
+
+        let ok = task::block_on(self.script_storage.remove_script(uuid, actor, force))?;
 
         Ok(json!(ok))
     }
 
+    fn list_scripts(&self) -> Result<JValue, JError> {
+        let scripts = task::block_on(self.script_storage.list_scripts())?;
+
+        Ok(JValue::Array(
+            scripts
+                .into_iter()
+                .map(|(id, script)| {
+                    let id: &String = id.borrow();
+                    json!({
+                        "id": id,
+                        "src": script.src,
+                        "failures": script.failures,
+                        "interval": script.interval.map(|i| pretty(i).to_string()),
+                        "owner": script.owner.to_string(),
+                    })
+                })
+                .collect(),
+        ))
+    }
+
     fn kademlia(&self) -> &KademliaApi {
         self.connectivity.as_ref()
     }

particle-closures/src/host_closures.rs

@@ -20,7 +20,7 @@ config-utils = { path = "../crates/config-utils" }
 kademlia = { path = "../crates/kademlia" }
 
 trust-graph = "0.2.0"
-air-interpreter-wasm = "0.3.4"
+air-interpreter-wasm = "0.4.2"
 
 libp2p = { package = "fluence-fork-libp2p", version = "0.34.0" }
 multihash ="0.13"