feat: Basic autoendpoint extractors (#151)

* Add FromRequest impl for Metrics

* Organize routes and add the rest of the health checks

* Edit the /__lbheartbeat__ route to be a little clearer

* Add a no-op webpush route

* Add basic webpush extraction structs

* Impl FromRequest for TokenInfo

* Translate wip FromRequest impl for Subscription from Python validator

Needs the meat of the validation + refactoring. Also unsure if the
crypto_keys config is correct, as the Python version enforces brackets,
even in env variables.

* Extract extractors to new module and add some docs

Pun fully intended.

* Add some more docs and refactoring to Subscription FromRequest impl

* Add a WebPushHeaders extractor and header utilities

* Add validation to WebPushHeaders

* Impl FromRequest for Notification

Based on the Python validator/extractor. Kind of annoying that we have
to `take` the payload because GATs isn't stable... Hopefully no one else
needs it!

* Fix incorrect call to get server state

* Add extractors to webpush_route

* Fix errors after rebase

* Add error for invalid token

* Make the subscription public key optional

v1 may not have a public key

* Suppress unused variable / mutability lint warnings

* Be more explicit about what extractors need the payload

* Add a PayloadError error kind to remove a TODO


Related to #103 (needs more work before it can be closed).

autoendpoint/Cargo.toml

@@ -11,11 +11,15 @@ actix-rt = "1.0"
 actix-cors = "0.2.0"
 autopush_common = { path = "../autopush-common" }
 backtrace = "0.3"
+base64 = "0.12.1"
 cadence = "0.19.1"
 config = "0.10.1"
 docopt = "1.1.0"
+fernet = "0.1.3"
 futures = { version = "0.3", features = ["compat"] }
+hex = "0.4.2"
 lazy_static = "1.4.0"
+regex = "1.3"
 sentry = { version = "0.18", features = ["with_curl_transport"] }
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
@@ -28,3 +32,5 @@ slog-stdlog = "4.0"
 slog-term = "2.5"
 thiserror = "1.0"
 url = "2.1"
+validator = "0.10.0"
+validator_derive = "0.10.0"

autoendpoint/src/error.rs

@@ -2,7 +2,7 @@
 
 use actix_web::{
     dev::{HttpResponseBuilder, ServiceResponse},
-    error::ResponseError,
+    error::{PayloadError, ResponseError},
     http::StatusCode,
     middleware::errhandlers::ErrorHandlerResponse,
     HttpResponse, Result,
@@ -48,6 +48,15 @@ pub enum ApiErrorKind {
     #[error("{0}")]
     Metrics(#[from] cadence::MetricError),
 
+    #[error("{0}")]
+    Validation(#[from] validator::ValidationErrors),
+
+    #[error("invalid token")]
+    InvalidToken,
+
+    #[error("{0}")]
+    PayloadError(PayloadError),
+
     #[error("{0}")]
     Internal(String),
 }
@@ -56,6 +65,9 @@ impl ApiErrorKind {
     /// Get the associated HTTP status code
     pub fn status(&self) -> StatusCode {
         match self {
+            ApiErrorKind::PayloadError(e) => e.status_code(),
+            ApiErrorKind::Validation(_) => StatusCode::BAD_REQUEST,
+            ApiErrorKind::InvalidToken => StatusCode::NOT_FOUND,
             ApiErrorKind::Io(_) | ApiErrorKind::Metrics(_) | ApiErrorKind::Internal(_) => {
                 StatusCode::INTERNAL_SERVER_ERROR
             }

autoendpoint/src/metrics.rs

@@ -1,7 +1,7 @@
 use std::net::UdpSocket;
 use std::time::Instant;
 
-use actix_web::{web::Data, HttpRequest};
+use actix_web::{web::Data, FromRequest, HttpRequest};
 use cadence::{
     BufferedUdpMetricSink, Counted, Metric, MetricError, NopMetricSink, QueuingMetricSink,
     StatsdClient, Timed,
@@ -10,6 +10,8 @@ use cadence::{
 use crate::server::ServerState;
 use crate::settings::Settings;
 use crate::tags::Tags;
+use actix_web::dev::{Payload, PayloadStream};
+use futures::future;
 
 #[derive(Debug, Clone)]
 pub struct MetricTimer {
@@ -55,6 +57,16 @@ impl Drop for Metrics {
     }
 }
 
+impl FromRequest for Metrics {
+    type Error = ();
+    type Future = future::Ready<Result<Self, Self::Error>>;
+    type Config = ();
+
+    fn from_request(req: &HttpRequest, _: &mut Payload<PayloadStream>) -> Self::Future {
+        future::ok(Metrics::from(req))
+    }
+}
+
 impl From<&HttpRequest> for Metrics {
     fn from(req: &HttpRequest) -> Self {
         let exts = req.extensions();

autoendpoint/src/server/extractors/mod.rs

@@ -0,0 +1,7 @@
+//! Actix extractors (`FromRequest`). These extractors transform and validate
+//! the incoming request data.
+
+pub mod notification;
+pub mod subscription;
+pub mod token_info;
+pub mod webpush_headers;

autoendpoint/src/server/extractors/notification.rs

@@ -0,0 +1,52 @@
+use crate::error::{ApiError, ApiErrorKind};
+use crate::server::extractors::subscription::Subscription;
+use crate::server::extractors::webpush_headers::WebPushHeaders;
+use actix_web::dev::{Payload, PayloadStream};
+use actix_web::{FromRequest, HttpRequest};
+use autopush_common::util::sec_since_epoch;
+use futures::{future, FutureExt, StreamExt};
+
+/// Extracts notification data from `Subscription` and request data
+pub struct Notification {
+    pub uaid: String,
+    pub channel_id: String,
+    pub version: String,
+    pub ttl: Option<u64>,
+    pub topic: Option<String>,
+    pub timestamp: u64,
+    pub data: String,
+}
+
+impl FromRequest for Notification {
+    type Error = ApiError;
+    type Future = future::LocalBoxFuture<'static, Result<Self, Self::Error>>;
+    type Config = ();
+
+    fn from_request(req: &HttpRequest, payload: &mut Payload<PayloadStream>) -> Self::Future {
+        let req = req.clone();
+        let mut payload = payload.take();
+
+        async move {
+            let headers = WebPushHeaders::extract(&req).await?;
+            let subscription = Subscription::extract(&req).await?;
+
+            // Read data and convert to base64
+            let mut data = Vec::new();
+            while let Some(item) = payload.next().await {
+                data.extend_from_slice(&item.map_err(ApiErrorKind::PayloadError)?);
+            }
+            let data = base64::encode_config(data, base64::URL_SAFE_NO_PAD);
+
+            Ok(Notification {
+                uaid: subscription.uaid,
+                channel_id: subscription.channel_id,
+                version: subscription.api_version,
+                ttl: headers.ttl,
+                topic: headers.topic,
+                timestamp: sec_since_epoch(),
+                data,
+            })
+        }
+        .boxed_local()
+    }
+}

autoendpoint/src/server/extractors/subscription.rs

@@ -0,0 +1,66 @@
+use crate::error::{ApiError, ApiErrorKind};
+use crate::server::extractors::token_info::TokenInfo;
+use crate::server::ServerState;
+use actix_http::{Payload, PayloadStream};
+use actix_web::web::Data;
+use actix_web::{FromRequest, HttpRequest};
+use futures::future;
+
+/// Extracts subscription data from `TokenInfo` and verifies auth/crypto headers
+pub struct Subscription {
+    pub uaid: String,
+    pub channel_id: String,
+    pub api_version: String,
+    pub public_key: Option<String>,
+}
+
+impl FromRequest for Subscription {
+    type Error = ApiError;
+    type Future = future::Ready<Result<Self, Self::Error>>;
+    type Config = ();
+
+    fn from_request(req: &HttpRequest, _: &mut Payload<PayloadStream>) -> Self::Future {
+        // Collect token info and server state
+        let token_info = match TokenInfo::extract(req).into_inner() {
+            Ok(t) => t,
+            Err(e) => return future::err(e),
+        };
+        let state: Data<ServerState> = Data::extract(req)
+            .into_inner()
+            .expect("No server state found");
+        let fernet = state.fernet.as_ref();
+
+        // Decrypt the token
+        let token = match fernet.decrypt(&token_info.token) {
+            Ok(t) => t,
+            Err(_) => return future::err(ApiErrorKind::InvalidToken.into()),
+        };
+
+        if token_info.api_version == "v1" && token.len() != 32 {
+            // Corrupted token
+            return future::err(ApiErrorKind::InvalidToken.into());
+        }
+
+        // Extract public key
+        let public_key = None;
+        if let Some(_crypto_key_header) = token_info.crypto_key_header {
+            todo!("Extract public key from header")
+        }
+
+        if let Some(_auth_header) = token_info.auth_header {
+            todo!("Parse vapid auth")
+        }
+
+        // Validate key data if on v2
+        if token_info.api_version == "v2" {
+            todo!("Perform v2 checks")
+        }
+
+        future::ok(Subscription {
+            uaid: hex::encode(&token[..16]),
+            channel_id: hex::encode(&token[16..32]),
+            api_version: token_info.api_version,
+            public_key,
+        })
+    }
+}

autoendpoint/src/server/extractors/token_info.rs

@@ -0,0 +1,40 @@
+use crate::error::ApiError;
+use crate::server::header_util::get_owned_header;
+use actix_http::{Payload, PayloadStream};
+use actix_web::{FromRequest, HttpRequest};
+use futures::future;
+
+/// Extracts basic token data from the webpush request path and headers
+pub struct TokenInfo {
+    pub api_version: String,
+    pub token: String,
+    pub crypto_key_header: Option<String>,
+    pub auth_header: Option<String>,
+}
+
+impl FromRequest for TokenInfo {
+    type Error = ApiError;
+    type Future = future::Ready<Result<Self, Self::Error>>;
+    type Config = ();
+
+    fn from_request(req: &HttpRequest, _: &mut Payload<PayloadStream>) -> Self::Future {
+        // Path variables
+        let api_version = req
+            .match_info()
+            .get("api_version")
+            .unwrap_or("v1")
+            .to_string();
+        let token = req
+            .match_info()
+            .get("token")
+            .expect("{token} must be part of the webpush path")
+            .to_string();
+
+        future::ok(TokenInfo {
+            api_version,
+            token,
+            crypto_key_header: get_owned_header(req, "crypto-key"),
+            auth_header: get_owned_header(req, "authorization"),
+        })
+    }
+}