Merge pull request #181 from AdExNetwork/issue-177-auth-middleware

Issue #177 auth middleware

Author: elpiel

Cargo.lock

@@ -10,7 +10,6 @@ primitives = {path = "../primitives"}
 futures-preview = {version = "=0.3.0-alpha.19"}
 # Time handling
 chrono = "0.4"
-time = "0.1.42"
 # To/From Hex
 hex = "0.3.2"
 serde =  {version = "^1.0", features = ['derive']}

adapter/Cargo.toml

@@ -1,5 +1,6 @@
 use crate::channel_validator::ChannelValidator;
 use crate::{Channel, DomainError, ValidatorId};
+use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
 use std::convert::From;
 use std::error::Error;
@@ -53,7 +54,7 @@ pub struct KeystoreOptions {
     pub keystore_pwd: String,
 }
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct Session {
     pub era: i64,
     pub uid: ValidatorId,

primitives/src/adapter.rs

@@ -117,7 +117,7 @@ async fn apply_rule(
                     Ok(format!(
                         "adexRateLimit:{}:{}",
                         hex::encode(channel.id),
-                        session.ip
+                        session.ip.as_ref().unwrap_or(&String::new())
                     ))
                 }
             } else {
@@ -157,13 +157,25 @@ mod test {
     use primitives::event_submission::{RateLimit, Rule};
     use primitives::sentry::Event;
     use primitives::util::tests::prep_db::DUMMY_CHANNEL;
-    use primitives::{Channel, EventSubmission};
+    use primitives::{Channel, Config, EventSubmission};
 
     use crate::db::redis_connection;
     use crate::Session;
 
     use super::*;
 
+    async fn setup() -> (Config, SharedConnection) {
+        let mut redis = redis_connection().await.expect("Couldn't connect to Redis");
+        let config = configuration("development", None).expect("Failed to get dev configuration");
+
+        // run `FLUSHALL` to clean any leftovers of other tests
+        let _ = redis::cmd("FLUSHALL")
+            .query_async::<_, String>(&mut redis)
+            .await;
+
+        (config, redis)
+    }
+
     fn get_channel(with_rule: Rule) -> Channel {
         let mut channel = DUMMY_CHANNEL.clone();
 
@@ -184,8 +196,7 @@ mod test {
 
     #[tokio::test]
     async fn session_uid_rate_limit() {
-        let redis = redis_connection().await.expect("Couldn't connect to Redis");
-        let config = configuration("development", None).expect("Failed to get dev configuration");
+        let (config, redis) = setup().await;
 
         let session = Session {
             era: 0,
@@ -219,8 +230,7 @@ mod test {
 
     #[tokio::test]
     async fn ip_rate_limit() {
-        let redis = redis_connection().await.expect("Couldn't connect to Redis");
-        let config = configuration("development", None).expect("Failed to get dev configuration");
+        let (config, redis) = setup().await;
 
         let session = Session {
             era: 0,

sentry/src/access.rs

@@ -1,12 +1,19 @@
 #![deny(clippy::all)]
 #![deny(rust_2018_idioms)]
 
+use crate::middleware::auth;
+use crate::middleware::cors::{cors, Cors};
 use hyper::service::{make_service_fn, service_fn};
 use hyper::{Body, Error, Method, Request, Response, Server, StatusCode};
 use primitives::adapter::Adapter;
 use primitives::Config;
+use redis::aio::SharedConnection;
 use slog::{error, info, Logger};
 
+pub mod middleware {
+    pub mod auth;
+    pub mod cors;
+}
 pub mod routes {
     pub mod channel;
     pub mod cfg {
@@ -32,33 +39,57 @@ pub mod event_reducer;
 
 pub struct Application<A: Adapter> {
     adapter: A,
-    logger: slog::Logger,
+    logger: Logger,
+    redis: SharedConnection,
     _clustered: bool,
     port: u16,
     config: Config,
 }
 
 impl<A: Adapter + 'static> Application<A> {
-    pub fn new(adapter: A, config: Config, logger: Logger, clustered: bool, port: u16) -> Self {
+    pub fn new(
+        adapter: A,
+        config: Config,
+        logger: Logger,
+        redis: SharedConnection,
+        clustered: bool,
+        port: u16,
+    ) -> Self {
         Self {
             adapter,
             config,
             logger,
+            redis,
             _clustered: clustered,
             port,
         }
     }
 
+    /// Starts the `hyper` `Server`.
     pub async fn run(&self) {
         let addr = ([127, 0, 0, 1], self.port).into();
         info!(&self.logger, "Listening on port {}!", self.port);
 
         let make_service = make_service_fn(move |_| {
             let adapter_config = (self.adapter.clone(), self.config.clone());
+            let redis = self.redis.clone();
+            let logger = self.logger.clone();
             async move {
                 Ok::<_, Error>(service_fn(move |req| {
                     let adapter_config = adapter_config.clone();
-                    async move { Ok::<_, Error>(handle_routing(req, adapter_config).await) }
+                    let redis = redis.clone();
+                    let logger = logger.clone();
+                    async move {
+                        Ok::<_, Error>(
+                            handle_routing(
+                                req,
+                                (&adapter_config.0, &adapter_config.1),
+                                redis,
+                                &logger,
+                            )
+                            .await,
+                        )
+                    }
                 }))
             }
         });
@@ -88,19 +119,45 @@ where
 
 async fn handle_routing(
     req: Request<Body>,
-    (adapter, config): (impl Adapter, Config),
+    (adapter, config): (&impl Adapter, &Config),
+    redis: SharedConnection,
+    logger: &Logger,
 ) -> Response<Body> {
-    match (req.uri().path(), req.method()) {
+    let headers = match cors(&req) {
+        Some(Cors::Simple(headers)) => headers,
+        // if we have a Preflight, just return the response directly
+        Some(Cors::Preflight(response)) => return response,
+        None => Default::default(),
+    };
+
+    let req = match auth::for_request(req, adapter, redis.clone()).await {
+        Ok(req) => req,
+        Err(error) => {
+            error!(&logger, "{}", &error; "module" => "middleware-auth");
+
+            return map_response_error(ResponseError::BadRequest(error));
+        }
+    };
+
+    let mut response = match (req.uri().path(), req.method()) {
         ("/cfg", &Method::GET) => crate::routes::cfg::return_config(&config),
         (route, _) if route.starts_with("/channel") => {
             crate::routes::channel::handle_channel_routes(req, adapter).await
         }
         _ => Err(ResponseError::NotFound),
     }
-    .unwrap_or_else(|response_err| match response_err {
+    .unwrap_or_else(map_response_error);
+
+    // extend the headers with the initial headers we have from CORS (if there are some)
+    response.headers_mut().extend(headers);
+    response
+}
+
+fn map_response_error(error: ResponseError) -> Response<Body> {
+    match error {
         ResponseError::NotFound => not_found(),
         ResponseError::BadRequest(error) => bad_request(error),
-    })
+    }
 }
 
 pub fn not_found() -> Response<Body> {
@@ -110,8 +167,8 @@ pub fn not_found() -> Response<Body> {
     response
 }
 
-pub fn bad_request(error: Box<dyn std::error::Error>) -> Response<Body> {
-    let body = Body::from(format!("Bad Request: {}", error));
+pub fn bad_request(_: Box<dyn std::error::Error>) -> Response<Body> {
+    let body = Body::from("Bad Request: try again later");
     let mut response = Response::new(body);
     let status = response.status_mut();
     *status = StatusCode::BAD_REQUEST;
@@ -123,5 +180,5 @@ pub fn bad_request(error: Box<dyn std::error::Error>) -> Response<Body> {
 pub struct Session {
     pub era: i64,
     pub uid: String,
-    pub ip: String,
+    pub ip: Option<String>,
 }

sentry/src/lib.rs

@@ -9,14 +9,15 @@ use primitives::config::configuration;
 use primitives::util::logging::{Async, PrefixedCompactFormat, TermDecorator};
 use primitives::util::tests::prep_db::{AUTH, IDS};
 use primitives::ValidatorId;
+use sentry::db::redis_connection;
 use sentry::Application;
 use slog::{o, Drain, Logger};
 use std::convert::TryFrom;
 
 const DEFAULT_PORT: u16 = 8005;
 
 #[tokio::main]
-async fn main() {
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let cli = App::new("Sentry")
         .version("0.1")
         .arg(
@@ -96,19 +97,21 @@ async fn main() {
     };
 
     let logger = logger();
+    let redis = redis_connection().await?;
 
     match adapter {
         AdapterTypes::EthereumAdapter(adapter) => {
-            Application::new(*adapter, config, logger, clustered, port)
+            Application::new(*adapter, config, logger, redis, clustered, port)
                 .run()
                 .await
         }
         AdapterTypes::DummyAdapter(adapter) => {
-            Application::new(*adapter, config, logger, clustered, port)
+            Application::new(*adapter, config, logger, redis, clustered, port)
                 .run()
                 .await
         }
     }
+    Ok(())
 }
 
 fn logger() -> Logger {