sentry - dummy event_aggregate route

sentry - `Session` and fix `Channel.creator` usage

Author: elpiel

sentry/src/access.rs

@@ -5,6 +5,7 @@ use redis::aio::MultiplexedConnection;
 use primitives::event_submission::{RateLimit, Rule};
 use primitives::sentry::Event;
 use primitives::Channel;
+use std::cmp::PartialEq;
 
 use crate::Session;
 
@@ -54,7 +55,7 @@ pub async fn check_access(
 
     let default_rules = [
         Rule {
-            uids: Some(vec![channel.creator.clone()]),
+            uids: Some(vec![channel.creator.to_string()]),
             rate_limit: None,
         },
         Rule {
@@ -74,7 +75,7 @@ pub async fn check_access(
     let rules = allow_rules
         .iter()
         .filter(|r| match &r.uids {
-            Some(uids) => uids.iter().any(|uid| &session.uid == uid),
+            Some(uids) => uids.iter().any(|uid| uid.eq(&session.uid.to_string())),
             None => true,
         })
         .collect::<Vec<_>>();
@@ -107,16 +108,11 @@ async fn apply_rule(
     match &rule.rate_limit {
         Some(rate_limit) => {
             let key = if &rate_limit.limit_type == "sid" {
-                // @TODO: Is this really necessary?
-                if session.uid.is_empty() {
-                    Err("rateLimit: unauthenticated request".to_string())
-                } else {
-                    Ok(format!(
-                        "adexRateLimit:{}:{}",
-                        hex::encode(channel.id),
-                        session.uid
-                    ))
-                }
+                Ok(format!(
+                    "adexRateLimit:{}:{}",
+                    hex::encode(channel.id),
+                    session.uid
+                ))
             } else if &rate_limit.limit_type == "ip" {
                 if events.len() != 1 {
                     Err("rateLimit: only allows 1 event".to_string())
@@ -163,7 +159,7 @@ mod test {
     use primitives::config::configuration;
     use primitives::event_submission::{RateLimit, Rule};
     use primitives::sentry::Event;
-    use primitives::util::tests::prep_db::DUMMY_CHANNEL;
+    use primitives::util::tests::prep_db::{DUMMY_CHANNEL, IDS};
     use primitives::{Channel, Config, EventSubmission};
 
     use crate::db::redis_connection;
@@ -207,7 +203,7 @@ mod test {
 
         let session = Session {
             era: 0,
-            uid: "response".to_string(),
+            uid: IDS["follower"].clone(),
             ip: Default::default(),
         };
 
@@ -241,7 +237,7 @@ mod test {
 
         let session = Session {
             era: 0,
-            uid: "response".to_string(),
+            uid: IDS["follower"].clone(),
             ip: Default::default(),
         };
 

sentry/src/event_reducer.rs

@@ -13,7 +13,7 @@ pub(crate) fn reduce(channel: &Channel, initial_aggr: &mut EventAggregate, ev: &
             initial_aggr.events.insert("IMPRESSION".to_owned(), merge);
         }
         Event::Close => {
-            let creator = channel.creator.clone();
+            let creator = channel.creator.to_string();
             let close_event = AggregateEvents {
                 event_counts: Some(vec![(creator.clone(), 1.into())].into_iter().collect()),
                 event_payouts: vec![(creator, channel.deposit_amount.clone())]
@@ -36,14 +36,14 @@ fn merge_impression_ev(
     let event_count = impression
         .event_counts
         .get_or_insert_with(Default::default)
-        .entry(earner.into())
+        .entry(earner.to_string())
         .or_insert_with(|| 0.into());
 
     *event_count += &1.into();
 
     let event_payouts = impression
         .event_payouts
-        .entry(earner.into())
+        .entry(earner.to_string())
         .or_insert_with(|| 0.into());
     *event_payouts += &channel.spec.min_per_impression;
 

sentry/src/lib.rs

@@ -7,11 +7,12 @@ use crate::middleware::auth;
 use crate::middleware::channel::channel_load;
 use crate::middleware::cors::{cors, Cors};
 use crate::routes::channel::channel_status;
+use crate::routes::event_aggregate::list_channel_event_aggregates;
 use crate::routes::validator_message::{extract_params, list_validator_messages};
 use hyper::{Body, Method, Request, Response, StatusCode};
 use lazy_static::lazy_static;
 use primitives::adapter::Adapter;
-use primitives::Config;
+use primitives::{Config, ValidatorId};
 use redis::aio::MultiplexedConnection;
 use regex::Regex;
 use routes::cfg::config;
@@ -23,11 +24,13 @@ pub mod middleware {
     pub mod auth;
     pub mod channel;
     pub mod cors;
+    pub mod event_aggregate;
 }
 
 pub mod routes {
     pub mod cfg;
     pub mod channel;
+    pub mod event_aggregate;
     pub mod validator_message;
 }
 
@@ -43,6 +46,7 @@ lazy_static! {
     static ref CHANNEL_STATUS_BY_CHANNEL_ID: Regex = Regex::new(r"^/channel/0x([a-zA-Z0-9]{64})/status/?$").expect("The regex should be valid");
     // Only the initial Regex to be matched.
     static ref CHANNEL_VALIDATOR_MESSAGES: Regex = Regex::new(r"^/channel/0x([a-zA-Z0-9]{64})/validator-messages(/.*)?$").expect("The regex should be valid");
+    static ref CHANNEL_EVENT_AGGREGATES: Regex = Regex::new(r"^/channel/0x([a-zA-Z0-9]{64})/event-aggregates(/|/0x[a-zA-Z0-9]{40}/?)?$").expect("The regex should be valid");
     // @TODO define other regex routes
 }
 
@@ -180,6 +184,29 @@ impl<A: Adapter + 'static> Application<A> {
                         };
 
                     list_validator_messages(req, &self, &extract_params.0, &extract_params.1).await
+                } else if let (Some(caps), &Method::GET) =
+                    (CHANNEL_EVENT_AGGREGATES.captures(path), method)
+                {
+                    if req.extensions().get::<Session>().is_none() {
+                        return map_response_error(ResponseError::Unauthorized);
+                    }
+
+                    let param = RouteParams(vec![
+                        caps.get(1)
+                            .map_or("".to_string(), |m| m.as_str().to_string()),
+                        caps.get(2)
+                            .map_or("".to_string(), |m| m.as_str().trim_matches('/').to_string()),
+                    ]);
+                    req.extensions_mut().insert(param);
+
+                    let req = match chain(req, &self, vec![channel_load]).await {
+                        Ok(req) => req,
+                        Err(error) => {
+                            return map_response_error(error);
+                        }
+                    };
+
+                    list_channel_event_aggregates(req, &self).await
                 } else {
                     Err(ResponseError::NotFound)
                 }
@@ -198,6 +225,7 @@ impl<A: Adapter + 'static> Application<A> {
 pub enum ResponseError {
     NotFound,
     BadRequest(String),
+    Unauthorized,
 }
 
 impl<T> From<T> for ResponseError
@@ -215,6 +243,7 @@ pub fn map_response_error(error: ResponseError) -> Response<Body> {
     match error {
         ResponseError::NotFound => not_found(),
         ResponseError::BadRequest(e) => bad_response(e),
+        ResponseError::Unauthorized => unauthorized(),
     }
 }
 
@@ -242,6 +271,13 @@ pub fn bad_response(response_body: String) -> Response<Body> {
     response
 }
 
+pub fn unauthorized() -> Response<Body> {
+    let mut response = Response::new(Body::from("Unauthorized"));
+    let status = response.status_mut();
+    *status = StatusCode::UNAUTHORIZED;
+    response
+}
+
 pub fn success_response(response_body: String) -> Response<Body> {
     let body = Body::from(response_body);
 
@@ -260,6 +296,6 @@ pub fn success_response(response_body: String) -> Response<Body> {
 #[derive(Debug, Clone)]
 pub struct Session {
     pub era: i64,
-    pub uid: String,
+    pub uid: ValidatorId,
     pub ip: Option<String>,
 }

sentry/src/middleware/auth.rs

@@ -60,7 +60,7 @@ pub(crate) async fn for_request(
 
         let session = Session {
             era: adapter_session.era,
-            uid: adapter_session.uid.to_hex_non_prefix_string(),
+            uid: adapter_session.uid,
             ip: get_request_ip(&req),
         };
 
@@ -165,8 +165,8 @@ mod test {
             .extensions()
             .get::<Session>()
             .expect("There should be a Session set inside the request");
-        let leader_id = IDS["leader"].to_hex_non_prefix_string();
-        assert_eq!(leader_id, session.uid);
+
+        assert_eq!(IDS["leader"], session.uid);
         assert!(session.ip.is_none());
     }
 }

sentry/src/middleware/event_aggregate.rs

@@ -0,0 +1,30 @@
+use crate::{Application, ResponseError, RouteParams};
+use hyper::{Body, Request};
+use primitives::adapter::Adapter;
+use primitives::sentry::EarnerAddress;
+
+/// channel_load & channel_if_exist
+pub async fn earner_load<A: Adapter>(
+    mut req: Request<Body>,
+    _app: &Application<A>,
+) -> Result<Request<Body>, ResponseError> {
+    let earner = req
+        .extensions()
+        .get::<RouteParams>()
+        .ok_or_else(|| ResponseError::BadRequest("Route params not found".to_string()))?
+        .get(1)
+        .ok_or_else(|| ResponseError::BadRequest("No earner param".to_string()))?;
+
+    let earner_option: Option<EarnerAddress> = if earner.is_empty() {
+        None
+    } else {
+        Some(
+            serde_json::from_str(&earner)
+                .map_err(|_| ResponseError::BadRequest("Invalid earner param".to_string()))?,
+        )
+    };
+
+    req.extensions_mut().insert(earner_option);
+
+    Ok(req)
+}

sentry/src/routes/event_aggregate.rs

@@ -0,0 +1,24 @@
+use crate::{Application, ResponseError, Session};
+use hyper::{Body, Request, Response};
+use primitives::adapter::Adapter;
+use primitives::Channel;
+
+pub async fn list_channel_event_aggregates<A: Adapter>(
+    req: Request<Body>,
+    _app: &Application<A>,
+) -> Result<Response<Body>, ResponseError> {
+    let channel = req
+        .extensions()
+        .get::<Channel>()
+        .expect("Request should have Channel");
+
+    // TODO: Auth required middleware
+    let session = req
+        .extensions()
+        .get::<Session>()
+        .ok_or_else(|| ResponseError::Unauthorized)?;
+
+    let _is_superuser = channel.spec.validators.find(&session.uid).is_some();
+
+    unimplemented!("Still need to finish it")
+}