sentry - axum refactor:

- routes - campaign - create
- routes - campaign - event submissions
- routes - channel - list
- routes - channel - dummy deposit
- routers - if dummy adapter middleware
- middleware - auth - authenticate & authentication_required
- middleware - campaign - load
- response - impl IntoResponse for ResponseError
- application - Qs - query string extractor based on serde_qs

Author: elpiel

sentry/src/application.rs

@@ -1,9 +1,15 @@
 use std::{
     net::{IpAddr, Ipv4Addr, SocketAddr},
     path::Path,
+    sync::Arc,
 };
 
 use adapter::client::Locked;
+use axum::{
+    extract::{FromRequest, RequestParts},
+    http::StatusCode,
+    middleware, Extension, Router,
+};
 use hyper::{
     service::{make_service_fn, service_fn},
     Error, Server,
@@ -14,19 +20,24 @@ use redis::ConnectionInfo;
 use serde::{Deserialize, Deserializer};
 use simple_hyper_server_tls::{listener_from_pem_files, Protocols, TlsListener};
 use slog::{error, info};
+use tower::ServiceBuilder;
+use tower_http::cors::CorsLayer;
 
 use crate::{
     db::{CampaignRemaining, DbPool},
     middleware::{
-        auth::Authenticate,
+        auth::{authenticate, Authenticate},
         cors::{cors, Cors},
         Middleware,
     },
     platform::PlatformApi,
     response::{map_response_error, ResponseError},
     routes::{
         get_cfg,
-        routers::{analytics_router, campaigns_router, channels_router},
+        routers::{
+            analytics_router, campaigns_router, campaigns_router_axum, channels_router,
+            channels_router_axum,
+        },
     },
 };
 use adapter::Adapter;
@@ -158,11 +169,45 @@ where
         response.headers_mut().extend(headers);
         response
     }
+
+    pub async fn axum_routing(&self) -> Router {
+        let cors = CorsLayer::new()
+            // "GET,HEAD,PUT,PATCH,POST,DELETE"
+            .allow_methods([
+                Method::GET,
+                Method::HEAD,
+                Method::PUT,
+                Method::PATCH,
+                Method::POST,
+                Method::DELETE,
+            ])
+            // allow requests from any origin
+            // "*"
+            .allow_origin(tower_http::cors::Any);
+
+        let channels = channels_router_axum::<C>();
+
+        let campaigns = campaigns_router_axum::<C>();
+
+        let router = Router::new()
+            .nest("/channel", channels)
+            .nest("/campaign", campaigns);
+
+        Router::new()
+            .nest("/v5", router)
+            .layer(
+                // keeps the order from top to bottom!
+                ServiceBuilder::new()
+                    .layer(cors)
+                    .layer(middleware::from_fn(authenticate::<C, _>)),
+            )
+            .layer(Extension(Arc::new(self.clone())))
+    }
 }
 
 impl<C: Locked + 'static> Application<C> {
     /// Starts the `hyper` `Server`.
-    pub async fn run(self, enable_tls: EnableTls) {
+    pub async fn run2(self, enable_tls: EnableTls) {
         let logger = self.logger.clone();
         let socket_addr = match &enable_tls {
             EnableTls::NoTls(socket_addr) => socket_addr,
@@ -215,6 +260,29 @@ impl<C: Locked + 'static> Application<C> {
             }
         }
     }
+
+    pub async fn run(self, enable_tls: EnableTls) {
+        let logger = self.logger.clone();
+        let socket_addr = match &enable_tls {
+            EnableTls::NoTls(socket_addr) => socket_addr,
+            EnableTls::Tls { socket_addr, .. } => socket_addr,
+        };
+
+        info!(&logger, "Listening on socket address: {}!", socket_addr);
+
+        let app = self.axum_routing().await;
+
+        let server = axum::Server::bind(socket_addr)
+            .serve(app.into_make_service())
+            .with_graceful_shutdown(shutdown_signal(logger.clone()));
+
+        tokio::pin!(server);
+
+        while let Err(e) = (&mut server).await {
+            // This is usually caused by trying to connect on HTTP instead of HTTPS
+            error!(&logger, "server error: {}", e; "main" => "run");
+        }
+    }
 }
 
 impl<C: Locked> Clone for Application<C> {
@@ -278,6 +346,27 @@ pub struct Auth {
     pub chain: primitives::Chain,
 }
 
+/// A query string deserialized using `serde_qs` instead of axum's `serde_urlencoded`
+pub struct Qs<T>(pub T);
+
+#[axum::async_trait]
+impl<T, B> FromRequest<B> for Qs<T>
+where
+    T: serde::de::DeserializeOwned,
+    B: Send,
+{
+    type Rejection = (StatusCode, String);
+
+    async fn from_request(req: &mut RequestParts<B>) -> Result<Self, Self::Rejection> {
+        let query = req.uri().query().unwrap_or_default();
+
+        match serde_qs::from_str(query) {
+            Ok(query) => Ok(Self(query)),
+            Err(err) => Err((StatusCode::BAD_REQUEST, err.to_string())),
+        }
+    }
+}
+
 /// A Ctrl+C signal to gracefully shutdown the server
 async fn shutdown_signal(logger: Logger) {
     // Wait for the Ctrl+C signal

sentry/src/middleware/auth.rs

@@ -1,8 +1,11 @@
-use std::error;
+use std::{error, sync::Arc};
 
 use async_trait::async_trait;
-use hyper::header::{AUTHORIZATION, REFERER};
-use hyper::{Body, Request};
+use axum::middleware::Next;
+use hyper::{
+    header::{AUTHORIZATION, REFERER},
+    Body, Request,
+};
 use redis::aio::MultiplexedConnection;
 
 use adapter::{prelude::*, primitives::Session as AdapterSession, Adapter};
@@ -71,6 +74,92 @@ impl<C: Locked + 'static> Middleware<C> for IsAdmin {
     }
 }
 
+pub async fn authentication_required<C: Locked + 'static, B>(
+    request: axum::http::Request<B>,
+    next: Next<B>,
+) -> Result<axum::response::Response, ResponseError> {
+    if request.extensions().get::<Auth>().is_some() {
+        Ok(next.run(request).await)
+    } else {
+        Err(ResponseError::Unauthorized)
+    }
+}
+
+pub async fn authenticate<C: Locked + 'static, B>(
+    mut request: axum::http::Request<B>,
+    next: Next<B>,
+) -> Result<axum::response::Response, ResponseError> {
+    let (adapter, redis) = {
+        let app = request
+            .extensions()
+            .get::<Arc<Application<C>>>()
+            .expect("Application should always be present");
+
+        (app.adapter.clone(), app.redis.clone())
+    };
+
+    let referrer = request
+        .headers()
+        .get(REFERER)
+        .and_then(|hv| hv.to_str().ok().map(ToString::to_string));
+
+    let session = Session {
+        ip: get_request_ip(&request),
+        country: None,
+        referrer_header: referrer,
+        os: None,
+    };
+    request.extensions_mut().insert(session);
+
+    let authorization = request.headers().get(AUTHORIZATION);
+
+    let prefix = "Bearer ";
+
+    let token = authorization
+        .and_then(|hv| {
+            hv.to_str()
+                .map(|token_str| token_str.strip_prefix(prefix))
+                .transpose()
+        })
+        .transpose()?;
+
+    if let Some(token) = token {
+        let adapter_session = match redis::cmd("GET")
+            .arg(token)
+            .query_async::<_, Option<String>>(&mut redis.clone())
+            .await?
+            .and_then(|session_str| serde_json::from_str::<AdapterSession>(&session_str).ok())
+        {
+            Some(adapter_session) => adapter_session,
+            None => {
+                // If there was a problem with the Session or the Token, this will error
+                // and a BadRequest response will be returned
+                let adapter_session = adapter.session_from_token(token).await?;
+
+                // save the Adapter Session to Redis for the next request
+                // if serde errors on deserialization this will override the value inside
+                redis::cmd("SET")
+                    .arg(token)
+                    .arg(serde_json::to_string(&adapter_session)?)
+                    .query_async(&mut redis.clone())
+                    .await?;
+
+                adapter_session
+            }
+        };
+
+        let auth = Auth {
+            era: adapter_session.era,
+            uid: ValidatorId::from(adapter_session.uid),
+            chain: adapter_session.chain,
+        };
+
+        request.extensions_mut().insert(auth);
+    }
+
+    Ok(next.run(request).await)
+}
+
 /// Check `Authorization` header for `Bearer` scheme with `Adapter::session_from_token`.
 /// If the `Adapter` fails to create an `AdapterSession`, `ResponseError::BadRequest` will be returned.
 async fn for_request<C: Locked>(
@@ -140,7 +229,7 @@ async fn for_request<C: Locked>(
     Ok(req)
 }
 
-fn get_request_ip(req: &Request<Body>) -> Option<String> {
+fn get_request_ip<B>(req: &Request<B>) -> Option<String> {
     req.headers()
         .get("true-client-ip")
         .or_else(|| req.headers().get("x-forwarded-for"))