Add multi user support

server/Cargo.toml

@@ -78,6 +78,7 @@ pyroscope = { version = "0.5.3", optional = true }
 pyroscope_pprofrs = { version = "0.2", optional = true }
 uptime_lib = "0.2.2"
 regex = "1.7.3"
+argon2 = "0.5.0"
 
 [build-dependencies]
 static-files = "0.2"

server/src/handlers/http.rs

@@ -20,7 +20,8 @@ use std::fs::File;
 use std::io::BufReader;
 
 use actix_cors::Cors;
-use actix_web::dev::ServiceRequest;
+use actix_web::dev::{Service, ServiceRequest};
+use actix_web::error::ErrorBadRequest;
 use actix_web::{middleware, web, App, HttpServer};
 use actix_web_httpauth::extractors::basic::BasicAuth;
 use actix_web_httpauth::middleware::HttpAuthentication;
@@ -30,11 +31,13 @@ use rustls::{Certificate, PrivateKey, ServerConfig};
 use rustls_pemfile::{certs, pkcs8_private_keys};
 
 use crate::option::CONFIG;
+use crate::rbac::get_user_map;
 
 mod health_check;
 mod ingest;
 mod logstream;
 mod query;
+mod rbac;
 
 include!(concat!(env!("OUT_DIR"), "/generated.rs"));
 
@@ -63,10 +66,13 @@ async fn validator(
     req: ServiceRequest,
     credentials: BasicAuth,
 ) -> Result<ServiceRequest, (actix_web::Error, ServiceRequest)> {
-    if credentials.user_id().trim() == CONFIG.parseable.username
-        && credentials.password().unwrap().trim() == CONFIG.parseable.password
-    {
-        return Ok(req);
+    let username = credentials.user_id().trim();
+    let password = credentials.password().unwrap().trim();
+
+    if let Some(user) = get_user_map().read().unwrap().get(username) {
+        if user.verify(password) {
+            return Ok(req);
+        }
     }
 
     Err((actix_web::error::ErrorUnauthorized("Unauthorized"), req))
@@ -157,6 +163,25 @@ pub fn configure_routes(cfg: &mut web::ServiceConfig) {
                 // GET "/logstream/{logstream}/retention" ==> Get retention for given logstream
                 .route(web::get().to(logstream::get_retention)),
         );
+    let user_api = web::scope("/user").service(
+        web::resource("/{username}")
+            // POST /user/{username} => Create a new user
+            .route(web::put().to(rbac::put_user))
+            // DELETE /user/{username} => Delete a user
+            .route(web::delete().to(rbac::delete_user))
+            .wrap_fn(|req, srv| {
+                // deny request if username is same as username from config
+                let username = req.match_info().get("username").unwrap_or("");
+                let is_root = username == CONFIG.parseable.username;
+                let call = srv.call(req);
+                async move {
+                    if is_root {
+                        return Err(ErrorBadRequest("Cannot call this API for root admin user"));
+                    }
+                    call.await
+                }
+            }),
+    );
 
     cfg.service(
         // Base path "{url}/api/v1"
@@ -184,6 +209,7 @@ pub fn configure_routes(cfg: &mut web::ServiceConfig) {
                         logstream_api,
                     ),
             )
+            .service(user_api)
             .wrap(HttpAuthentication::basic(validator)),
     )
     // GET "/" ==> Serve the static frontend directory

server/src/handlers/http/rbac.rs

@@ -0,0 +1,153 @@
+/*
+ * Parseable Server (C) 2022 - 2023 Parseable, Inc.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ */
+
+use crate::{
+    option::CONFIG,
+    rbac::{
+        get_user_map,
+        user::{PassCode, User},
+    },
+    storage::{self, ObjectStorageError, StorageMetadata},
+    validator::{self, error::UsernameValidationError},
+};
+use actix_web::{http::header::ContentType, web, Responder};
+use http::StatusCode;
+use tokio::sync::Mutex;
+
+// async aware lock for updating storage metadata and user map atomicically
+static UPDATE_LOCK: Mutex<()> = Mutex::const_new(());
+
+// Handler for PUT /api/v1/user/{username}
+// Creates a new user by username if it does not exists
+// Otherwise make a call to reset password
+// returns password generated for this user
+pub async fn put_user(username: web::Path<String>) -> Result<impl Responder, RBACError> {
+    let username = username.into_inner();
+    validator::verify_username(&username)?;
+    let _ = UPDATE_LOCK.lock().await;
+    let user_exists = get_user_map().read().unwrap().contains_key(&username);
+    if user_exists {
+        reset_password(username).await
+    } else {
+        let mut metadata = get_metadata().await?;
+        if metadata.users.iter().any(|user| user.username == username) {
+            // should be unreachable given state is always consistent
+            return Err(RBACError::UserExists);
+        }
+
+        let (user, password) = User::create_new(username);
+        metadata.users.push(user.clone());
+        put_metadata(&metadata).await?;
+        // set this user to user map
+        get_user_map().write().unwrap().insert(user);
+
+        Ok(password)
+    }
+}
+
+// Handler for DELETE /api/v1/user/delete/{username}
+pub async fn delete_user(username: web::Path<String>) -> Result<impl Responder, RBACError> {
+    let username = username.into_inner();
+    let _ = UPDATE_LOCK.lock().await;
+    // fail this request if the user does not exists
+    if !get_user_map().read().unwrap().contains_key(&username) {
+        return Err(RBACError::UserDoesNotExist);
+    };
+    // delete from parseable.json first
+    let mut metadata = get_metadata().await?;
+    metadata.users.retain(|user| user.username != username);
+    put_metadata(&metadata).await?;
+    // update in mem table
+    get_user_map().write().unwrap().remove(&username);
+    Ok(format!("deleted user: {}", username))
+}
+
+// Reset password for given username
+// returns new password generated for this user
+pub async fn reset_password(username: String) -> Result<String, RBACError> {
+    // get new password for this user
+    let PassCode { password, hash } = User::gen_new_password();
+    // update parseable.json first
+    let mut metadata = get_metadata().await?;
+    if let Some(user) = metadata
+        .users
+        .iter_mut()
+        .find(|user| user.username == username)
+    {
+        user.password_hash.clone_from(&hash);
+    } else {
+        // should be unreachable given state is always consistent
+        return Err(RBACError::UserDoesNotExist);
+    }
+    put_metadata(&metadata).await?;
+
+    // update in mem table
+    get_user_map()
+        .write()
+        .unwrap()
+        .get_mut(&username)
+        .expect("checked that user exists in map")
+        .password_hash = hash;
+
+    Ok(password)
+}
+
+async fn get_metadata() -> Result<crate::storage::StorageMetadata, ObjectStorageError> {
+    let metadata = CONFIG
+        .storage()
+        .get_object_store()
+        .get_metadata()
+        .await?
+        .expect("metadata is initialized");
+    Ok(metadata)
+}
+
+async fn put_metadata(metadata: &StorageMetadata) -> Result<(), ObjectStorageError> {
+    storage::put_remote_metadata(metadata).await?;
+    storage::put_staging_metadata(metadata)?;
+    Ok(())
+}
+
+#[derive(Debug, thiserror::Error)]
+pub enum RBACError {
+    #[error("User exists already")]
+    UserExists,
+    #[error("User does not exist")]
+    UserDoesNotExist,
+    #[error("Failed to connect to storage: {0}")]
+    ObjectStorageError(#[from] ObjectStorageError),
+    #[error("invalid Username: {0}")]
+    ValidationError(#[from] UsernameValidationError),
+}
+
+impl actix_web::ResponseError for RBACError {
+    fn status_code(&self) -> http::StatusCode {
+        match self {
+            Self::UserExists => StatusCode::BAD_REQUEST,
+            Self::UserDoesNotExist => StatusCode::NOT_FOUND,
+            Self::ValidationError(_) => StatusCode::BAD_REQUEST,
+            Self::ObjectStorageError(_) => StatusCode::INTERNAL_SERVER_ERROR,
+        }
+    }
+
+    fn error_response(&self) -> actix_web::HttpResponse<actix_web::body::BoxBody> {
+        actix_web::HttpResponse::build(self.status_code())
+            .insert_header(ContentType::plaintext())
+            .body(self.to_string())
+    }
+}

server/src/main.rs

@@ -41,6 +41,7 @@ mod metrics;
 mod migration;
 mod option;
 mod query;
+mod rbac;
 mod response;
 mod stats;
 mod storage;
@@ -60,9 +61,11 @@ async fn main() -> anyhow::Result<()> {
     CONFIG.validate();
     let storage = CONFIG.storage().get_object_store();
     CONFIG.validate_staging()?;
+    migration::run_metadata_migration(&CONFIG).await?;
     let metadata = storage::resolve_parseable_metadata().await?;
+    banner::print(&CONFIG, &metadata).await;
+    rbac::set_user_map(metadata.users.clone());
     metadata.set_global();
-    banner::print(&CONFIG, storage::StorageMetadata::global()).await;
     let prometheus = metrics::build_metrics_handler();
     CONFIG.storage().register_store_metrics(&prometheus);