feat: Validate user subscription data in autoendpoint (#160)

* Update autoendpoint so only one cadence version is used

* Remove unnecessary Rc and Box in DynamoStorage and support Send + Sync

DynamoDbClient and StatsdClient both are internally Arc, so clones are
already cheap and refer to the same data. This change also makes
DynamoStorage Send + Sync.

* Autoendpoint: Connect to DynamoDB

* Add DynamoStorage::get_user_channels

* Read and validate user data in the Subscription extractor

* Remove router_data from DynamoDbUser

The associated validation is not included, so it is unused.

Closes #156

autoendpoint/Cargo.toml

@@ -12,7 +12,7 @@ actix-cors = "0.2.0"
 autopush_common = { path = "../autopush-common" }
 backtrace = "0.3"
 base64 = "0.12.1"
-cadence = "0.19.1"
+cadence = "0.20"
 config = "0.10.1"
 docopt = "1.1.0"
 fernet = "0.1.3"
@@ -33,5 +33,6 @@ slog-stdlog = "4.0"
 slog-term = "2.5"
 thiserror = "1.0"
 url = "2.1"
+uuid = { version = "0.8.1", features = ["serde", "v4"] }
 validator = "0.10.0"
 validator_derive = "0.10.0"

autoendpoint/src/error.rs

@@ -59,12 +59,21 @@ pub enum ApiErrorKind {
     #[error(transparent)]
     VapidError(#[from] VapidError),
 
+    #[error(transparent)]
+    Uuid(#[from] uuid::Error),
+
     #[error("Error while validating token")]
     TokenHashValidation(#[from] openssl::error::ErrorStack),
 
+    #[error("Database error: {0}")]
+    Database(#[source] autopush_common::errors::Error),
+
     #[error("Invalid token")]
     InvalidToken,
 
+    #[error("No such subscription")]
+    NoSubscription,
+
     /// A specific issue with the encryption headers
     #[error("{0}")]
     InvalidEncryption(String),
@@ -88,17 +97,21 @@ impl ApiErrorKind {
 
             ApiErrorKind::Validation(_)
             | ApiErrorKind::InvalidEncryption(_)
-            | ApiErrorKind::TokenHashValidation(_) => StatusCode::BAD_REQUEST,
+            | ApiErrorKind::TokenHashValidation(_)
+            | ApiErrorKind::Uuid(_) => StatusCode::BAD_REQUEST,
+
+            ApiErrorKind::NoSubscription => StatusCode::GONE,
 
             ApiErrorKind::VapidError(_) => StatusCode::UNAUTHORIZED,
 
             ApiErrorKind::InvalidToken | ApiErrorKind::InvalidApiVersion => StatusCode::NOT_FOUND,
 
             ApiErrorKind::PayloadTooLarge(_) => StatusCode::PAYLOAD_TOO_LARGE,
 
-            ApiErrorKind::Io(_) | ApiErrorKind::Metrics(_) | ApiErrorKind::Internal(_) => {
-                StatusCode::INTERNAL_SERVER_ERROR
-            }
+            ApiErrorKind::Io(_)
+            | ApiErrorKind::Metrics(_)
+            | ApiErrorKind::Database(_)
+            | ApiErrorKind::Internal(_) => StatusCode::INTERNAL_SERVER_ERROR,
         }
     }
 }

autoendpoint/src/server/extractors/mod.rs

@@ -5,3 +5,4 @@ pub mod notification;
 pub mod notification_headers;
 pub mod subscription;
 pub mod token_info;
+pub mod user;

autoendpoint/src/server/extractors/subscription.rs

@@ -1,21 +1,25 @@
 use crate::error::{ApiError, ApiErrorKind, ApiResult};
 use crate::server::extractors::token_info::TokenInfo;
+use crate::server::extractors::user::validate_user;
 use crate::server::headers::crypto_key::CryptoKeyHeader;
 use crate::server::headers::vapid::{VapidHeader, VapidVersionData};
 use crate::server::{ServerState, VapidError};
 use actix_http::{Payload, PayloadStream};
 use actix_web::web::Data;
 use actix_web::{FromRequest, HttpRequest};
+use autopush_common::db::DynamoDbUser;
 use cadence::{Counted, StatsdClient};
+use futures::compat::Future01CompatExt;
 use futures::future::LocalBoxFuture;
 use futures::FutureExt;
 use openssl::hash;
 use std::borrow::Cow;
+use uuid::Uuid;
 
 /// Extracts subscription data from `TokenInfo` and verifies auth/crypto headers
 pub struct Subscription {
-    pub uaid: String,
-    pub channel_id: String,
+    pub user: DynamoDbUser,
+    pub channel_id: Uuid,
     pub vapid: Option<VapidHeader>,
     pub public_key: Option<String>,
 }
@@ -33,10 +37,10 @@ impl FromRequest for Subscription {
             let token_info = TokenInfo::extract(&req).await?;
             let state: Data<ServerState> =
                 Data::extract(&req).await.expect("No server state found");
-            let fernet = state.fernet.as_ref();
 
             // Decrypt the token
-            let token = fernet
+            let token = state
+                .fernet
                 .decrypt(&repad_base64(&token_info.token))
                 .map_err(|_| ApiErrorKind::InvalidToken)?;
 
@@ -53,9 +57,20 @@ impl FromRequest for Subscription {
                 version_1_validation(&token)?;
             }
 
+            // Load and validate user data
+            let uaid = Uuid::from_slice(&token[..16])?;
+            let channel_id = Uuid::from_slice(&token[16..32])?;
+            let user = state
+                .ddb
+                .get_user(&uaid)
+                .compat()
+                .await
+                .map_err(ApiErrorKind::Database)?;
+            validate_user(&user, &channel_id, &state).await?;
+
             Ok(Subscription {
-                uaid: hex::encode(&token[..16]),
-                channel_id: hex::encode(&token[16..32]),
+                user,
+                channel_id,
                 vapid,
                 public_key,
             })

autoendpoint/src/server/extractors/user.rs

@@ -0,0 +1,85 @@
+//! User validations
+
+use crate::error::{ApiErrorKind, ApiResult};
+use crate::server::ServerState;
+use autopush_common::db::{DynamoDbUser, DynamoStorage};
+use cadence::{Counted, StatsdClient};
+use futures::compat::Future01CompatExt;
+use uuid::Uuid;
+
+/// Valid `DynamoDbUser::router_type` values
+const VALID_ROUTERS: [&str; 5] = ["webpush", "gcm", "fcm", "apns", "adm"];
+
+/// Perform some validations on the user, including:
+/// - Validate router type
+/// - (WebPush) Check that the subscription/channel exists
+/// - (WebPush) Drop user if inactive
+pub async fn validate_user(
+    user: &DynamoDbUser,
+    channel_id: &Uuid,
+    state: &ServerState,
+) -> ApiResult<()> {
+    if !VALID_ROUTERS.contains(&user.router_type.as_str()) {
+        debug!("Unknown router type, dropping user"; "user" => ?user);
+        drop_user(&user.uaid, &state.ddb, &state.metrics).await?;
+        return Err(ApiErrorKind::NoSubscription.into());
+    }
+
+    if user.router_type == "webpush" {
+        validate_webpush_user(user, channel_id, &state.ddb, &state.metrics).await?;
+    }
+
+    Ok(())
+}
+
+/// Make sure the user is not inactive and the subscription channel exists
+async fn validate_webpush_user(
+    user: &DynamoDbUser,
+    channel_id: &Uuid,
+    ddb: &DynamoStorage,
+    metrics: &StatsdClient,
+) -> ApiResult<()> {
+    // Make sure the user is active (has a valid message table)
+    let message_table = match user.current_month.as_ref() {
+        Some(table) => table,
+        None => {
+            debug!("Missing `current_month` value, dropping user"; "user" => ?user);
+            drop_user(&user.uaid, ddb, metrics).await?;
+            return Err(ApiErrorKind::NoSubscription.into());
+        }
+    };
+
+    if !ddb.message_table_names.contains(message_table) {
+        debug!("User is inactive, dropping user"; "user" => ?user);
+        drop_user(&user.uaid, ddb, metrics).await?;
+        return Err(ApiErrorKind::NoSubscription.into());
+    }
+
+    // Make sure the subscription channel exists
+    let channel_ids = ddb
+        .get_user_channels(&user.uaid, message_table)
+        .compat()
+        .await
+        .map_err(ApiErrorKind::Database)?;
+
+    if !channel_ids.contains(channel_id) {
+        return Err(ApiErrorKind::NoSubscription.into());
+    }
+
+    Ok(())
+}
+
+/// Drop a user and increment associated metric
+async fn drop_user(uaid: &Uuid, ddb: &DynamoStorage, metrics: &StatsdClient) -> ApiResult<()> {
+    metrics
+        .incr_with_tags("updates.drop_user")
+        .with_tag("errno", "102")
+        .send();
+
+    ddb.drop_uaid(uaid)
+        .compat()
+        .await
+        .map_err(ApiErrorKind::Database)?;
+
+    Ok(())
+}

autoendpoint/src/server/mod.rs

@@ -1,18 +1,18 @@
 //! Main application server
 
-use actix_cors::Cors;
-use actix_web::{
-    dev, http::StatusCode, middleware::errhandlers::ErrorHandlers, web, App, HttpServer,
-};
-use cadence::StatsdClient;
-
-use crate::error::{ApiError, ApiResult};
+use crate::error::{ApiError, ApiErrorKind, ApiResult};
 use crate::metrics;
 use crate::server::routes::health::{
     health_route, lb_heartbeat_route, status_route, version_route,
 };
 use crate::server::routes::webpush::webpush_route;
 use crate::settings::Settings;
+use actix_cors::Cors;
+use actix_web::{
+    dev, http::StatusCode, middleware::errhandlers::ErrorHandlers, web, App, HttpServer,
+};
+use autopush_common::db::DynamoStorage;
+use cadence::StatsdClient;
 use fernet::MultiFernet;
 use std::sync::Arc;
 
@@ -28,6 +28,7 @@ pub struct ServerState {
     pub metrics: StatsdClient,
     pub settings: Settings,
     pub fernet: Arc<MultiFernet>,
+    pub ddb: DynamoStorage,
 }
 
 pub struct Server;
@@ -37,10 +38,17 @@ impl Server {
         let metrics = metrics::metrics_from_opts(&settings)?;
         let bind_address = format!("{}:{}", settings.host, settings.port);
         let fernet = Arc::new(settings.make_fernet());
+        let ddb = DynamoStorage::from_opts(
+            &settings.message_table_name,
+            &settings.router_table_name,
+            metrics.clone(),
+        )
+        .map_err(ApiErrorKind::Database)?;
         let state = ServerState {
             metrics,
             settings,
             fernet,
+            ddb,
         };
 
         let server = HttpServer::new(move || {

autoendpoint/src/settings.rs

@@ -19,6 +19,9 @@ pub struct Settings {
     #[cfg(any(test, feature = "db_test"))]
     pub database_use_test_transactions: bool,
 
+    pub router_table_name: String,
+    pub message_table_name: String,
+
     pub max_data_bytes: usize,
     pub crypto_keys: String,
     pub human_logs: bool,
@@ -38,6 +41,8 @@ impl Default for Settings {
             database_pool_max_size: None,
             #[cfg(any(test, feature = "db_test"))]
             database_use_test_transactions: false,
+            router_table_name: "router".to_string(),
+            message_table_name: "message".to_string(),
             max_data_bytes: 4096,
             crypto_keys: format!("[{}]", Fernet::generate_key()),
             human_logs: false,