Merge pull request #274 from AdExNetwork/dev

Fast forward master to dev: validator worker fixes

Author: Ivo Georgiev

adapter/src/ethereum.rs

@@ -115,7 +115,9 @@ impl Adapter for EthereumAdapter {
 
     fn sign(&self, state_root: &str) -> AdapterResult<String> {
         if let Some(wallet) = &self.wallet {
-            let message = Message::from_slice(&hash_message(state_root));
+            let state_root = hex::decode(state_root)
+                .map_err(|_| AdapterError::Signature("invalid state_root".to_string()))?;
+            let message = Message::from_slice(&hash_message(&state_root));
             let wallet_sign = wallet
                 .sign(&self.keystore_pwd, &message)
                 .map_err(|_| map_error("failed to sign messages"))?;
@@ -130,11 +132,16 @@ impl Adapter for EthereumAdapter {
     }
 
     fn verify(&self, signer: &ValidatorId, state_root: &str, sig: &str) -> AdapterResult<bool> {
-        let decoded_signature = hex::decode(sig)
+        if !sig.starts_with("0x") {
+            return Err(AdapterError::Signature("not 0x prefixed hex".to_string()));
+        }
+        let decoded_signature = hex::decode(&sig[2..])
             .map_err(|_| AdapterError::Signature("invalid signature".to_string()))?;
         let address = Address::from_slice(signer.inner());
         let signature = Signature::from_electrum(&decoded_signature);
-        let message = Message::from_slice(&hash_message(state_root));
+        let state_root = hex::decode(state_root)
+            .map_err(|_| AdapterError::Signature("invalid state_root".to_string()))?;
+        let message = Message::from_slice(&hash_message(&state_root));
 
         verify_address(&address, &signature, &message).or_else(|_| Ok(false))
     }
@@ -314,14 +321,13 @@ impl RelayerClient {
     }
 }
 
-fn hash_message(message: &str) -> [u8; 32] {
+fn hash_message(message: &[u8]) -> [u8; 32] {
     let eth = "\x19Ethereum Signed Message:\n";
     let message_length = message.len();
 
-    let encoded = format!("{}{}{}", eth, message_length, message);
-
     let mut result = Keccak::new_keccak256();
-    result.update(&encoded.as_bytes());
+    result.update(&format!("{}{}", eth, message_length).as_bytes());
+    result.update(&message);
 
     let mut res: [u8; 32] = [0; 32];
     result.finalize(&mut res);
@@ -371,10 +377,9 @@ pub fn ewt_sign(
 
     let payload_encoded =
         base64::encode_config(&serde_json::to_string(payload)?, base64::URL_SAFE_NO_PAD);
-    let message = Message::from_slice(&hash_message(&format!(
-        "{}.{}",
-        header_encoded, payload_encoded
-    )));
+    let message = Message::from_slice(&hash_message(
+        &format!("{}.{}", header_encoded, payload_encoded).as_bytes(),
+    ));
     let signature: Signature = signer
         .sign(password, &message)
         .map_err(|_| map_error("sign message"))?
@@ -394,10 +399,9 @@ pub fn ewt_verify(
     payload_encoded: &str,
     token: &str,
 ) -> Result<VerifyPayload, Box<dyn Error>> {
-    let message = Message::from_slice(&hash_message(&format!(
-        "{}.{}",
-        header_encoded, payload_encoded
-    )));
+    let message = Message::from_slice(&hash_message(
+        &format!("{}.{}", header_encoded, payload_encoded).as_bytes(),
+    ));
 
     let decoded_signature = base64::decode_config(&token, base64::URL_SAFE_NO_PAD)?;
     let signature = Signature::from_electrum(&decoded_signature);
@@ -466,19 +470,19 @@ mod test {
 
         // Sign
         let expected_response =
-            "0xce654de0b3d14d63e1cb3181eee7a7a37ef4a06c9fabc204faf96f26357441b625b1be460fbe8f5278cc02aa88a5d0ac2f238e9e3b8e4893760d33bccf77e47f1b";
+            "0x625fd46f82c4cfd135ea6a8534e85dbf50beb157046dce59d2e97aacdf4e38381d1513c0e6f002b2f05c05458038b187754ff38cc0658dfc9ba854cccfb6e13e1b";
         let message = "2bdeafae53940669daa6f519373f686c";
         let response = eth_adapter.sign(message).expect("failed to sign message");
         assert_eq!(expected_response, response, "invalid signature");
 
         // Verify
         let signature =
-            "ce654de0b3d14d63e1cb3181eee7a7a37ef4a06c9fabc204faf96f26357441b625b1be460fbe8f5278cc02aa88a5d0ac2f238e9e3b8e4893760d33bccf77e47f1b";
+            "0x9e07f12958ce7c5eb1362eb9461e4745dd9d74a42b921391393caea700bfbd6e1ad876a7d8f9202ef1fe6110dbfe87840c5676ca5c4fda9f3330694a1ac2a1fc1b";
         let verify = eth_adapter
             .verify(
-                &ValidatorId::try_from("2bDeAFAE53940669DaA6F519373f686c1f3d3393")
+                &ValidatorId::try_from("2892f6C41E0718eeeDd49D98D648C789668cA67d")
                     .expect("Failed to parse id"),
-                "2bdeafae53940669daa6f519373f686c",
+                "8bc45d8eb27f4c98cab35d17b0baecc2a263d6831ef0800f4c190cbfac6d20a3",
                 &signature,
             )
             .expect("Failed to verify signatures");

primitives/src/util/tests/prep_db.rs

@@ -1,4 +1,5 @@
 use crate::{
+    channel::{Pricing, PricingBounds},
     BigNum, Channel, ChannelId, ChannelSpec, EventSubmission, SpecValidators, ValidatorDesc,
     ValidatorId,
 };
@@ -69,7 +70,7 @@ lazy_static! {
                 title: None,
                 validators: SpecValidators::new(DUMMY_VALIDATOR_LEADER.clone(), DUMMY_VALIDATOR_FOLLOWER.clone()),
                 max_per_impression: 10.into(),
-                min_per_impression: 10.into(),
+                min_per_impression: 1.into(),
                 targeting: vec![],
                 min_targeting_score: None,
                 event_submission: Some(EventSubmission { allow: vec![] }),
@@ -79,7 +80,7 @@ lazy_static! {
                 nonce: Some(nonce),
                 withdraw_period_start: Utc.timestamp_millis(4_073_414_400_000),
                 ad_units: vec![],
-                pricing_bounds: None,
+                pricing_bounds: Some(PricingBounds {impression: None, click: Some(Pricing { max: 0.into(), min: 0.into()})}),
             },
         }
     };

sentry/src/main.rs

@@ -28,6 +28,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         )
         .arg(
             Arg::with_name("adapter")
+                .long("adapter")
                 .short("a")
                 .help("the adapter for authentication and signing")
                 .required(true)
@@ -37,18 +38,21 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         )
         .arg(
             Arg::with_name("keystoreFile")
+                .long("keystoreFile")
                 .short("k")
                 .help("path to the JSON Ethereum Keystore file")
                 .takes_value(true),
         )
         .arg(
             Arg::with_name("dummyIdentity")
+                .long("dummyIdentity")
                 .short("i")
                 .help("the identity to use with the dummy adapter")
                 .takes_value(true),
         )
         .arg(
             Arg::with_name("clustered")
+                .long("clustered")
                 .short("c")
                 .help("Run app in cluster mode with multiple workers"),
         )

validator_worker/src/error.rs

@@ -1,3 +1,4 @@
+use primitives::ChannelId;
 use serde::{Deserialize, Serialize};
 use std::error::Error;
 use std::fmt::{Display, Formatter, Result};
@@ -6,15 +7,19 @@ use std::fmt::{Display, Formatter, Result};
 pub enum ValidatorWorker {
     Configuration(String),
     Failed(String),
+    Channel(ChannelId, String),
 }
 
 impl Error for ValidatorWorker {}
 
 impl Display for ValidatorWorker {
     fn fmt(&self, f: &mut Formatter<'_>) -> Result {
         match self {
-            ValidatorWorker::Configuration(error) => write!(f, "Configuration error: {}", error),
-            ValidatorWorker::Failed(error) => write!(f, "error: {}", error),
+            ValidatorWorker::Configuration(err) => write!(f, "Configuration error: {}", err),
+            ValidatorWorker::Failed(err) => write!(f, "error: {}", err),
+            ValidatorWorker::Channel(channel_id, err) => {
+                write!(f, "Channel {}: {}", channel_id, err)
+            }
         }
     }
 }

validator_worker/src/main.rs

@@ -6,7 +6,7 @@ use std::error::Error;
 use std::time::Duration;
 
 use clap::{App, Arg};
-use futures::future::{join, try_join_all};
+use futures::future::{join, join_all};
 use tokio::runtime::Runtime;
 use tokio::time::{delay_for, timeout};
 
@@ -36,6 +36,7 @@ fn main() -> Result<(), Box<dyn Error>> {
         )
         .arg(
             Arg::with_name("adapter")
+                .long("adapter")
                 .short("a")
                 .help("the adapter for authentication and signing")
                 .required(true)
@@ -45,18 +46,21 @@ fn main() -> Result<(), Box<dyn Error>> {
         )
         .arg(
             Arg::with_name("keystoreFile")
+                .long("keystoreFile")
                 .short("k")
                 .help("path to the JSON Ethereum Keystore file")
                 .takes_value(true),
         )
         .arg(
             Arg::with_name("dummyIdentity")
+                .long("dummyIdentity")
                 .short("i")
                 .help("the identity to use with the dummy adapter")
                 .takes_value(true),
         )
         .arg(
             Arg::with_name("sentryUrl")
+                .long("sentryUrl")
                 .short("u")
                 .help("the URL to the sentry used for listing channels")
                 .default_value("http://127.0.0.1:8005")
@@ -65,6 +69,7 @@ fn main() -> Result<(), Box<dyn Error>> {
         )
         .arg(
             Arg::with_name("singleTick")
+                .long("singleTick")
                 .short("t")
                 .takes_value(false)
                 .help("runs the validator in single-tick mode and exit"),
@@ -160,26 +165,26 @@ async fn iterate_channels<A: Adapter + 'static>(args: Args<A>, logger: &Logger)
     let channels = match result {
         Ok(channels) => channels,
         Err(e) => {
-            error!(logger, "Failed to get channels {}", &e; "main" => "iterate_channels");
+            error!(logger, "Failed to get channels - {}", &e; "main" => "iterate_channels");
             return;
         }
     };
 
     let channels_size = channels.len();
 
-    let tick = try_join_all(
+    let tick_results = join_all(
         channels
             .into_iter()
             .map(|channel| validator_tick(args.adapter.clone(), channel, &args.config, logger)),
     )
     .await;
 
-    info!(logger, "processed {} channels", channels_size);
-
-    if let Err(e) = tick {
-        error!(logger, "An occurred while processing channels {}", &e; "main" => "iterate_channels");
+    for channel_err in tick_results.into_iter().filter_map(Result::err) {
+        error!(logger, "Error processing channels - {}", channel_err; "main" => "iterate_channels");
     }
 
+    info!(logger, "processed {} channels", channels_size);
+
     if channels_size >= args.config.max_channels as usize {
         error!(logger, "WARNING: channel limit cfg.MAX_CHANNELS={} reached", &args.config.max_channels; "main" => "iterate_channels");
     }
@@ -197,23 +202,33 @@ async fn validator_tick<A: Adapter + 'static>(
     let duration = Duration::from_millis(config.validator_tick_timeout as u64);
 
     match channel.spec.validators.find(&whoami) {
-        SpecValidator::Leader(_) => {
-            if let Err(e) = timeout(duration, leader::tick(&sentry)).await {
-                return Err(ValidatorWorkerError::Failed(e.to_string()));
-            }
-        }
-        SpecValidator::Follower(_) => {
-            if let Err(e) = timeout(duration, follower::tick(&sentry)).await {
-                return Err(ValidatorWorkerError::Failed(e.to_string()));
-            }
-        }
-        SpecValidator::None => {
-            return Err(ValidatorWorkerError::Failed(
-                "validatorTick: processing a channel where we are not validating".to_string(),
-            ))
-        }
-    };
-    Ok(())
+        SpecValidator::Leader(_) => match timeout(duration, leader::tick(&sentry)).await {
+            Err(timeout_e) => Err(ValidatorWorkerError::Channel(
+                channel.id,
+                timeout_e.to_string(),
+            )),
+            Ok(Err(tick_e)) => Err(ValidatorWorkerError::Channel(
+                channel.id,
+                tick_e.to_string(),
+            )),
+            _ => Ok(()),
+        },
+        SpecValidator::Follower(_) => match timeout(duration, follower::tick(&sentry)).await {
+            Err(timeout_e) => Err(ValidatorWorkerError::Channel(
+                channel.id,
+                timeout_e.to_string(),
+            )),
+            Ok(Err(tick_e)) => Err(ValidatorWorkerError::Channel(
+                channel.id,
+                tick_e.to_string(),
+            )),
+            _ => Ok(()),
+        },
+        SpecValidator::None => Err(ValidatorWorkerError::Channel(
+            channel.id,
+            "validatorTick: processing a channel which we are not validating".to_string(),
+        )),
+    }
 }
 
 fn logger() -> Logger {