feat(enterprise): add auth TLS and recovery hardening

rust/crates/rustdb-auth/Cargo.toml

@@ -0,0 +1,20 @@
+[package]
+name = "rustdb-auth"
+version.workspace = true
+edition.workspace = true
+license.workspace = true
+authors.workspace = true
+description = "Authentication primitives for RustDb"
+
+[dependencies]
+base64 = { workspace = true }
+bson = { workspace = true }
+hmac = { workspace = true }
+pbkdf2 = { workspace = true }
+rand = { workspace = true }
+rustdb-config = { workspace = true }
+serde = { workspace = true }
+serde_json = { workspace = true }
+sha2 = { workspace = true }
+subtle = { workspace = true }
+thiserror = { workspace = true }

rust/crates/rustdb-auth/src/lib.rs

@@ -0,0 +1,565 @@
+use std::collections::HashMap;
+use std::io::Write;
+use std::path::{Path, PathBuf};
+use std::sync::RwLock;
+
+use base64::{engine::general_purpose::STANDARD as BASE64_STANDARD, Engine as _};
+use hmac::{Hmac, Mac};
+use pbkdf2::pbkdf2_hmac;
+use rand::{rngs::OsRng, RngCore};
+use serde::{Deserialize, Serialize};
+use sha2::{Digest, Sha256};
+use subtle::ConstantTimeEq;
+
+use rustdb_config::{AuthOptions, AuthUserOptions};
+
+type HmacSha256 = Hmac<Sha256>;
+
+const SCRAM_SHA_256: &str = "SCRAM-SHA-256";
+
+#[derive(Debug, thiserror::Error)]
+pub enum AuthError {
+    #[error("authentication is disabled")]
+    Disabled,
+    #[error("unsupported authentication mechanism: {0}")]
+    UnsupportedMechanism(String),
+    #[error("invalid SCRAM payload: {0}")]
+    InvalidPayload(String),
+    #[error("authentication failed")]
+    AuthenticationFailed,
+    #[error("unknown SASL conversation")]
+    UnknownConversation,
+    #[error("user already exists: {0}")]
+    UserAlreadyExists(String),
+    #[error("user not found: {0}")]
+    UserNotFound(String),
+    #[error("auth metadata persistence failed: {0}")]
+    Persistence(String),
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum AuthAction {
+    Read,
+    Write,
+    DbAdmin,
+    UserAdmin,
+    ClusterMonitor,
+}
+
+#[derive(Debug, Clone)]
+pub struct AuthenticatedUser {
+    pub username: String,
+    pub database: String,
+    pub roles: Vec<String>,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+struct ScramCredential {
+    salt: Vec<u8>,
+    iterations: u32,
+    stored_key: Vec<u8>,
+    server_key: Vec<u8>,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+struct AuthUser {
+    username: String,
+    database: String,
+    roles: Vec<String>,
+    scram_sha256: ScramCredential,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, Default)]
+struct PersistedAuthState {
+    users: Vec<AuthUser>,
+}
+
+#[derive(Debug, Clone)]
+pub struct ScramConversation {
+    user: AuthenticatedUser,
+    client_first_bare: String,
+    server_first: String,
+    nonce: String,
+    stored_key: Vec<u8>,
+    server_key: Vec<u8>,
+}
+
+#[derive(Debug, Clone)]
+pub struct ScramStartResult {
+    pub payload: Vec<u8>,
+    pub conversation: ScramConversation,
+}
+
+#[derive(Debug, Clone)]
+pub struct ScramContinueResult {
+    pub payload: Vec<u8>,
+    pub user: AuthenticatedUser,
+}
+
+#[derive(Debug)]
+pub struct AuthEngine {
+    enabled: bool,
+    users: RwLock<HashMap<String, AuthUser>>,
+    users_path: Option<PathBuf>,
+    scram_iterations: u32,
+}
+
+impl AuthEngine {
+    pub fn from_options(options: &AuthOptions) -> Result<Self, AuthError> {
+        let users_path = options.users_path.as_ref().map(PathBuf::from);
+        let mut users = if let Some(ref path) = users_path {
+            load_users(path)?
+        } else {
+            HashMap::new()
+        };
+
+        let mut changed = false;
+        for user_options in &options.users {
+            let key = user_key(&user_options.database, &user_options.username);
+            if !users.contains_key(&key) {
+                let user = AuthUser::from_options(user_options, options.scram_iterations);
+                users.insert(key, user);
+                changed = true;
+            }
+        }
+
+        if changed {
+            if let Some(ref path) = users_path {
+                persist_users(path, &users)?;
+            }
+        }
+
+        Ok(Self {
+            enabled: options.enabled,
+            users: RwLock::new(users),
+            users_path,
+            scram_iterations: options.scram_iterations,
+        })
+    }
+
+    pub fn disabled() -> Self {
+        Self {
+            enabled: false,
+            users: RwLock::new(HashMap::new()),
+            users_path: None,
+            scram_iterations: 15000,
+        }
+    }
+
+    pub fn enabled(&self) -> bool {
+        self.enabled
+    }
+
+    pub fn supported_mechanisms(&self, namespace_user: &str) -> Vec<String> {
+        let Some((database, username)) = namespace_user.split_once('.') else {
+            return Vec::new();
+        };
+        let users = self.users.read().unwrap_or_else(|poisoned| poisoned.into_inner());
+        if users.contains_key(&user_key(database, username)) {
+            vec![SCRAM_SHA_256.to_string()]
+        } else {
+            Vec::new()
+        }
+    }
+
+    pub fn is_authorized(
+        &self,
+        authenticated_users: &[AuthenticatedUser],
+        target_db: &str,
+        action: AuthAction,
+    ) -> bool {
+        authenticated_users
+            .iter()
+            .any(|user| user.roles.iter().any(|role| role_allows(role, user, target_db, action)))
+    }
+
+    pub fn create_user(
+        &self,
+        database: &str,
+        username: &str,
+        password: &str,
+        roles: Vec<String>,
+    ) -> Result<(), AuthError> {
+        let key = user_key(database, username);
+        let mut users = self.users.write().unwrap_or_else(|poisoned| poisoned.into_inner());
+        if users.contains_key(&key) {
+            return Err(AuthError::UserAlreadyExists(format!("{database}.{username}")));
+        }
+        let options = AuthUserOptions {
+            username: username.to_string(),
+            password: password.to_string(),
+            database: database.to_string(),
+            roles,
+        };
+        users.insert(key, AuthUser::from_options(&options, self.scram_iterations));
+        self.persist_locked(&users)
+    }
+
+    pub fn drop_user(&self, database: &str, username: &str) -> Result<(), AuthError> {
+        let key = user_key(database, username);
+        let mut users = self.users.write().unwrap_or_else(|poisoned| poisoned.into_inner());
+        if users.remove(&key).is_none() {
+            return Err(AuthError::UserNotFound(format!("{database}.{username}")));
+        }
+        self.persist_locked(&users)
+    }
+
+    pub fn update_user(
+        &self,
+        database: &str,
+        username: &str,
+        password: Option<&str>,
+        roles: Option<Vec<String>>,
+    ) -> Result<(), AuthError> {
+        let key = user_key(database, username);
+        let mut users = self.users.write().unwrap_or_else(|poisoned| poisoned.into_inner());
+        let user = users
+            .get_mut(&key)
+            .ok_or_else(|| AuthError::UserNotFound(format!("{database}.{username}")))?;
+        if let Some(new_roles) = roles {
+            user.roles = new_roles;
+        }
+        if let Some(new_password) = password {
+            let options = AuthUserOptions {
+                username: username.to_string(),
+                password: new_password.to_string(),
+                database: database.to_string(),
+                roles: user.roles.clone(),
+            };
+            user.scram_sha256 = AuthUser::from_options(&options, self.scram_iterations).scram_sha256;
+        }
+        self.persist_locked(&users)
+    }
+
+    pub fn grant_roles(
+        &self,
+        database: &str,
+        username: &str,
+        roles: Vec<String>,
+    ) -> Result<(), AuthError> {
+        let key = user_key(database, username);
+        let mut users = self.users.write().unwrap_or_else(|poisoned| poisoned.into_inner());
+        let user = users
+            .get_mut(&key)
+            .ok_or_else(|| AuthError::UserNotFound(format!("{database}.{username}")))?;
+        for role in roles {
+            if !user.roles.contains(&role) {
+                user.roles.push(role);
+            }
+        }
+        self.persist_locked(&users)
+    }
+
+    pub fn revoke_roles(
+        &self,
+        database: &str,
+        username: &str,
+        roles: Vec<String>,
+    ) -> Result<(), AuthError> {
+        let key = user_key(database, username);
+        let mut users = self.users.write().unwrap_or_else(|poisoned| poisoned.into_inner());
+        let user = users
+            .get_mut(&key)
+            .ok_or_else(|| AuthError::UserNotFound(format!("{database}.{username}")))?;
+        user.roles.retain(|role| !roles.contains(role));
+        self.persist_locked(&users)
+    }
+
+    pub fn users_info(&self, database: &str, username: Option<&str>) -> Vec<AuthenticatedUser> {
+        let users = self.users.read().unwrap_or_else(|poisoned| poisoned.into_inner());
+        users
+            .values()
+            .filter(|user| user.database == database)
+            .filter(|user| username.map(|name| user.username == name).unwrap_or(true))
+            .map(AuthUser::to_authenticated_user)
+            .collect()
+    }
+
+    pub fn start_scram_sha256(
+        &self,
+        database: &str,
+        payload: &[u8],
+    ) -> Result<ScramStartResult, AuthError> {
+        if !self.enabled {
+            return Err(AuthError::Disabled);
+        }
+
+        let message = std::str::from_utf8(payload)
+            .map_err(|_| AuthError::InvalidPayload("payload is not valid UTF-8".to_string()))?;
+        let client_first_bare = message
+            .strip_prefix("n,,")
+            .ok_or_else(|| AuthError::InvalidPayload("expected SCRAM gs2 header 'n,,'".to_string()))?;
+        let attrs = parse_scram_attrs(client_first_bare);
+        let raw_username = attrs
+            .get("n")
+            .ok_or_else(|| AuthError::InvalidPayload("missing username".to_string()))?;
+        let username = decode_scram_name(raw_username);
+        let client_nonce = attrs
+            .get("r")
+            .ok_or_else(|| AuthError::InvalidPayload("missing client nonce".to_string()))?;
+
+        let users = self.users.read().unwrap_or_else(|poisoned| poisoned.into_inner());
+        let user = users
+            .get(&user_key(database, &username))
+            .ok_or(AuthError::AuthenticationFailed)?;
+
+        let nonce = format!("{}{}", client_nonce, secure_base64(18));
+        let server_first = format!(
+            "r={},s={},i={}",
+            nonce,
+            BASE64_STANDARD.encode(&user.scram_sha256.salt),
+            user.scram_sha256.iterations,
+        );
+
+        Ok(ScramStartResult {
+            payload: server_first.as_bytes().to_vec(),
+            conversation: ScramConversation {
+                user: user.to_authenticated_user(),
+                client_first_bare: client_first_bare.to_string(),
+                server_first: server_first.clone(),
+                nonce,
+                stored_key: user.scram_sha256.stored_key.clone(),
+                server_key: user.scram_sha256.server_key.clone(),
+            },
+        })
+    }
+
+    pub fn continue_scram_sha256(
+        &self,
+        conversation: ScramConversation,
+        payload: &[u8],
+    ) -> Result<ScramContinueResult, AuthError> {
+        let message = std::str::from_utf8(payload)
+            .map_err(|_| AuthError::InvalidPayload("payload is not valid UTF-8".to_string()))?;
+        let proof_marker = ",p=";
+        let proof_pos = message
+            .rfind(proof_marker)
+            .ok_or_else(|| AuthError::InvalidPayload("missing client proof".to_string()))?;
+        let client_final_without_proof = &message[..proof_pos];
+        let proof_b64 = &message[proof_pos + proof_marker.len()..];
+        let attrs = parse_scram_attrs(client_final_without_proof);
+        let nonce = attrs
+            .get("r")
+            .ok_or_else(|| AuthError::InvalidPayload("missing nonce".to_string()))?;
+        if nonce != &conversation.nonce {
+            return Err(AuthError::AuthenticationFailed);
+        }
+
+        let client_proof = BASE64_STANDARD
+            .decode(proof_b64.as_bytes())
+            .map_err(|_| AuthError::InvalidPayload("invalid client proof encoding".to_string()))?;
+        if client_proof.len() != 32 || conversation.stored_key.len() != 32 {
+            return Err(AuthError::AuthenticationFailed);
+        }
+
+        let auth_message = format!(
+            "{},{},{}",
+            conversation.client_first_bare,
+            conversation.server_first,
+            client_final_without_proof,
+        );
+        let client_signature = hmac_sha256(&conversation.stored_key, auth_message.as_bytes());
+        let client_key: Vec<u8> = client_proof
+            .iter()
+            .zip(client_signature.iter())
+            .map(|(proof_byte, signature_byte)| proof_byte ^ signature_byte)
+            .collect();
+        let computed_stored_key = Sha256::digest(&client_key).to_vec();
+
+        if computed_stored_key.ct_eq(&conversation.stored_key).unwrap_u8() != 1 {
+            return Err(AuthError::AuthenticationFailed);
+        }
+
+        let server_signature = hmac_sha256(&conversation.server_key, auth_message.as_bytes());
+        let server_final = format!("v={}", BASE64_STANDARD.encode(server_signature));
+
+        Ok(ScramContinueResult {
+            payload: server_final.as_bytes().to_vec(),
+            user: conversation.user,
+        })
+    }
+
+    fn persist_locked(&self, users: &HashMap<String, AuthUser>) -> Result<(), AuthError> {
+        if let Some(ref path) = self.users_path {
+            persist_users(path, users)?;
+        }
+        Ok(())
+    }
+}
+
+impl Default for AuthEngine {
+    fn default() -> Self {
+        Self::disabled()
+    }
+}
+
+impl AuthUser {
+    fn from_options(options: &AuthUserOptions, iterations: u32) -> Self {
+        let salt = secure_random(24);
+        let salted_password = salted_password(options.password.as_bytes(), &salt, iterations);
+        let client_key = hmac_sha256(&salted_password, b"Client Key");
+        let stored_key = Sha256::digest(&client_key).to_vec();
+        let server_key = hmac_sha256(&salted_password, b"Server Key");
+
+        Self {
+            username: options.username.clone(),
+            database: options.database.clone(),
+            roles: options.roles.clone(),
+            scram_sha256: ScramCredential {
+                salt,
+                iterations,
+                stored_key,
+                server_key,
+            },
+        }
+    }
+
+    fn to_authenticated_user(&self) -> AuthenticatedUser {
+        AuthenticatedUser {
+            username: self.username.clone(),
+            database: self.database.clone(),
+            roles: self.roles.clone(),
+        }
+    }
+}
+
+fn role_allows(role: &str, user: &AuthenticatedUser, target_db: &str, action: AuthAction) -> bool {
+    let (role_db, role_name) = role.split_once('.').unwrap_or(("", role));
+    if role_name == "root" {
+        return true;
+    }
+
+    let any_database = role_name.ends_with("AnyDatabase");
+    let scoped_db = if role_db.is_empty() { &user.database } else { role_db };
+    if !any_database && scoped_db != target_db {
+        return false;
+    }
+
+    match role_name {
+        "read" | "readAnyDatabase" => action == AuthAction::Read,
+        "readWrite" | "readWriteAnyDatabase" => {
+            matches!(action, AuthAction::Read | AuthAction::Write)
+        }
+        "dbAdmin" | "dbAdminAnyDatabase" => action == AuthAction::DbAdmin,
+        "userAdmin" | "userAdminAnyDatabase" => action == AuthAction::UserAdmin,
+        "clusterMonitor" => action == AuthAction::ClusterMonitor,
+        _ => false,
+    }
+}
+
+fn load_users(path: &Path) -> Result<HashMap<String, AuthUser>, AuthError> {
+    if !path.exists() {
+        return Ok(HashMap::new());
+    }
+    let data = std::fs::read_to_string(path).map_err(|e| AuthError::Persistence(e.to_string()))?;
+    let persisted: PersistedAuthState = serde_json::from_str(&data)
+        .map_err(|e| AuthError::Persistence(format!("failed to parse users file: {e}")))?;
+    Ok(persisted
+        .users
+        .into_iter()
+        .map(|user| (user_key(&user.database, &user.username), user))
+        .collect())
+}
+
+fn persist_users(path: &Path, users: &HashMap<String, AuthUser>) -> Result<(), AuthError> {
+    if let Some(parent) = path.parent() {
+        std::fs::create_dir_all(parent).map_err(|e| AuthError::Persistence(e.to_string()))?;
+    }
+
+    let mut user_list: Vec<AuthUser> = users.values().cloned().collect();
+    user_list.sort_by(|a, b| a.database.cmp(&b.database).then(a.username.cmp(&b.username)));
+    let payload = serde_json::to_vec_pretty(&PersistedAuthState { users: user_list })
+        .map_err(|e| AuthError::Persistence(e.to_string()))?;
+
+    let tmp_path = path.with_extension("tmp");
+    {
+        let mut file = std::fs::File::create(&tmp_path)
+            .map_err(|e| AuthError::Persistence(e.to_string()))?;
+        file.write_all(&payload)
+            .map_err(|e| AuthError::Persistence(e.to_string()))?;
+        file.sync_all()
+            .map_err(|e| AuthError::Persistence(e.to_string()))?;
+    }
+    std::fs::rename(&tmp_path, path).map_err(|e| AuthError::Persistence(e.to_string()))?;
+    if let Some(parent) = path.parent() {
+        if let Ok(dir) = std::fs::File::open(parent) {
+            let _ = dir.sync_all();
+        }
+    }
+    Ok(())
+}
+
+fn user_key(database: &str, username: &str) -> String {
+    format!("{}\0{}", database, username)
+}
+
+fn salted_password(password: &[u8], salt: &[u8], iterations: u32) -> Vec<u8> {
+    let mut output = [0u8; 32];
+    pbkdf2_hmac::<Sha256>(password, salt, iterations, &mut output);
+    output.to_vec()
+}
+
+fn hmac_sha256(key: &[u8], message: &[u8]) -> Vec<u8> {
+    let mut mac = HmacSha256::new_from_slice(key).expect("HMAC-SHA256 accepts keys of any size");
+    mac.update(message);
+    mac.finalize().into_bytes().to_vec()
+}
+
+fn secure_random(len: usize) -> Vec<u8> {
+    let mut bytes = vec![0u8; len];
+    OsRng.fill_bytes(&mut bytes);
+    bytes
+}
+
+fn secure_base64(len: usize) -> String {
+    BASE64_STANDARD.encode(secure_random(len))
+}
+
+fn parse_scram_attrs(input: &str) -> HashMap<String, String> {
+    let mut result = HashMap::new();
+    for part in input.split(',') {
+        if let Some((key, value)) = part.split_once('=') {
+            result.insert(key.to_string(), value.to_string());
+        }
+    }
+    result
+}
+
+fn decode_scram_name(input: &str) -> String {
+    input.replace("=2C", ",").replace("=3D", "=")
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn mechanism_lookup_returns_scram_sha256() {
+        let options = AuthOptions {
+            enabled: true,
+            users: vec![AuthUserOptions {
+                username: "root".to_string(),
+                password: "secret".to_string(),
+                database: "admin".to_string(),
+                roles: vec!["root".to_string()],
+            }],
+            users_path: None,
+            scram_iterations: 4096,
+        };
+        let engine = AuthEngine::from_options(&options).unwrap();
+        assert_eq!(engine.supported_mechanisms("admin.root"), vec![SCRAM_SHA_256.to_string()]);
+    }
+
+    #[test]
+    fn read_write_role_allows_read_and_write_only_on_own_db() {
+        let user = AuthenticatedUser {
+            username: "app".to_string(),
+            database: "appdb".to_string(),
+            roles: vec!["readWrite".to_string()],
+        };
+        assert!(role_allows("readWrite", &user, "appdb", AuthAction::Read));
+        assert!(role_allows("readWrite", &user, "appdb", AuthAction::Write));
+        assert!(!role_allows("readWrite", &user, "other", AuthAction::Read));
+        assert!(!role_allows("readWrite", &user, "appdb", AuthAction::DbAdmin));
+    }
+}

rust/crates/rustdb-commands/Cargo.toml

@@ -22,3 +22,4 @@ rustdb-query = { workspace = true }
 rustdb-storage = { workspace = true }
 rustdb-index = { workspace = true }
 rustdb-txn = { workspace = true }
+rustdb-auth = { workspace = true }

rust/crates/rustdb-commands/src/context.rs

@@ -2,6 +2,7 @@ use std::sync::Arc;
 
 use bson::{Bson, Document};
 use dashmap::DashMap;
+use rustdb_auth::{AuthEngine, AuthenticatedUser, ScramConversation};
 use rustdb_index::{IndexEngine, IndexOptions};
 use rustdb_storage::{OpLog, StorageAdapter};
 use rustdb_txn::{SessionEngine, TransactionEngine};
@@ -22,6 +23,8 @@ pub struct CommandContext {
     pub start_time: std::time::Instant,
     /// Operation log for point-in-time replay.
     pub oplog: Arc<OpLog>,
+    /// Authentication engine and user store.
+    pub auth: Arc<AuthEngine>,
 }
 
 impl CommandContext {
@@ -85,6 +88,43 @@ impl CommandContext {
     }
 }
 
+/// Per-client connection state. Authentication is socket-scoped in MongoDB.
+pub struct ConnectionState {
+    pub authenticated_users: Vec<AuthenticatedUser>,
+    pub sasl_conversations: std::collections::HashMap<i32, ScramConversation>,
+    next_conversation_id: i32,
+}
+
+impl ConnectionState {
+    pub fn new() -> Self {
+        Self {
+            authenticated_users: Vec::new(),
+            sasl_conversations: std::collections::HashMap::new(),
+            next_conversation_id: 1,
+        }
+    }
+
+    pub fn is_authenticated(&self) -> bool {
+        !self.authenticated_users.is_empty()
+    }
+
+    pub fn next_conversation_id(&mut self) -> i32 {
+        let id = self.next_conversation_id;
+        self.next_conversation_id += 1;
+        id
+    }
+
+    pub fn authenticate(&mut self, user: AuthenticatedUser) {
+        self.authenticated_users.push(user);
+    }
+}
+
+impl Default for ConnectionState {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
 /// State of an open cursor from a find or aggregate command.
 pub struct CursorState {
     /// Documents remaining to be returned.

rust/crates/rustdb-commands/src/error.rs

@@ -30,6 +30,15 @@ pub enum CommandError {
     #[error("immutable field: {0}")]
     ImmutableField(String),
 
+    #[error("unauthorized: {0}")]
+    Unauthorized(String),
+
+    #[error("authentication failed")]
+    AuthenticationFailed,
+
+    #[error("illegal operation: {0}")]
+    IllegalOperation(String),
+
     #[error("internal error: {0}")]
     InternalError(String),
 }
@@ -47,6 +56,9 @@ impl CommandError {
             CommandError::NamespaceExists(_) => (48, "NamespaceExists"),
             CommandError::DuplicateKey(_) => (11000, "DuplicateKey"),
             CommandError::ImmutableField(_) => (66, "ImmutableField"),
+            CommandError::Unauthorized(_) => (13, "Unauthorized"),
+            CommandError::AuthenticationFailed => (18, "AuthenticationFailed"),
+            CommandError::IllegalOperation(_) => (20, "IllegalOperation"),
             CommandError::InternalError(_) => (1, "InternalError"),
         };
 

rust/crates/rustdb-commands/src/handlers/admin_handler.rs

@@ -98,6 +98,18 @@ pub async fn handle(
             "ok": 1.0,
         }),
 
+        "createUser" => handle_create_user(cmd, db, ctx).await,
+
+        "updateUser" => handle_update_user(cmd, db, ctx).await,
+
+        "dropUser" => handle_drop_user(cmd, db, ctx).await,
+
+        "usersInfo" => handle_users_info(cmd, db, ctx).await,
+
+        "grantRolesToUser" => handle_grant_roles_to_user(cmd, db, ctx).await,
+
+        "revokeRolesFromUser" => handle_revoke_roles_from_user(cmd, db, ctx).await,
+
         "listDatabases" => handle_list_databases(cmd, ctx).await,
 
         "listCollections" => handle_list_collections(cmd, db, ctx).await,
@@ -144,15 +156,9 @@ pub async fn handle(
             Ok(doc! { "ok": 1.0 })
         }
 
-        "commitTransaction" => {
-            // Stub: acknowledge.
-            Ok(doc! { "ok": 1.0 })
-        }
-
-        "abortTransaction" => {
-            // Stub: acknowledge.
-            Ok(doc! { "ok": 1.0 })
-        }
+        "commitTransaction" | "abortTransaction" => Err(CommandError::IllegalOperation(
+            "Transaction numbers are only allowed on a replica set member or mongos".into(),
+        )),
 
         // Auth stubs - accept silently.
         "saslStart" => Ok(doc! {
@@ -189,6 +195,166 @@ pub async fn handle(
     }
 }
 
+async fn handle_create_user(
+    cmd: &Document,
+    db: &str,
+    ctx: &CommandContext,
+) -> CommandResult<Document> {
+    let username = cmd
+        .get_str("createUser")
+        .map_err(|_| CommandError::InvalidArgument("missing 'createUser' field".into()))?;
+    let password = cmd
+        .get_str("pwd")
+        .map_err(|_| CommandError::InvalidArgument("missing 'pwd' field".into()))?;
+    let roles = parse_roles(cmd, db, "roles")?;
+    ctx.auth
+        .create_user(db, username, password, roles)
+        .map_err(auth_error_to_command_error)?;
+    Ok(doc! { "ok": 1.0 })
+}
+
+async fn handle_update_user(
+    cmd: &Document,
+    db: &str,
+    ctx: &CommandContext,
+) -> CommandResult<Document> {
+    let username = cmd
+        .get_str("updateUser")
+        .map_err(|_| CommandError::InvalidArgument("missing 'updateUser' field".into()))?;
+    let password = cmd.get_str("pwd").ok();
+    let roles = if cmd.contains_key("roles") {
+        Some(parse_roles(cmd, db, "roles")?)
+    } else {
+        None
+    };
+    ctx.auth
+        .update_user(db, username, password, roles)
+        .map_err(auth_error_to_command_error)?;
+    Ok(doc! { "ok": 1.0 })
+}
+
+async fn handle_drop_user(
+    cmd: &Document,
+    db: &str,
+    ctx: &CommandContext,
+) -> CommandResult<Document> {
+    let username = cmd
+        .get_str("dropUser")
+        .map_err(|_| CommandError::InvalidArgument("missing 'dropUser' field".into()))?;
+    ctx.auth
+        .drop_user(db, username)
+        .map_err(auth_error_to_command_error)?;
+    Ok(doc! { "ok": 1.0 })
+}
+
+async fn handle_users_info(
+    cmd: &Document,
+    db: &str,
+    ctx: &CommandContext,
+) -> CommandResult<Document> {
+    let username = match cmd.get("usersInfo") {
+        Some(Bson::String(name)) => Some(name.as_str()),
+        Some(Bson::Document(user_doc)) => user_doc.get_str("user").ok(),
+        _ => None,
+    };
+    let users = ctx.auth.users_info(db, username);
+    let user_docs: Vec<Bson> = users
+        .into_iter()
+        .map(|user| {
+            let roles: Vec<Bson> = user
+                .roles
+                .iter()
+                .map(|role| Bson::Document(role_to_document(&user.database, role)))
+                .collect();
+            Bson::Document(doc! {
+                "user": user.username,
+                "db": user.database,
+                "roles": roles,
+                "mechanisms": ["SCRAM-SHA-256"],
+            })
+        })
+        .collect();
+    Ok(doc! { "users": user_docs, "ok": 1.0 })
+}
+
+async fn handle_grant_roles_to_user(
+    cmd: &Document,
+    db: &str,
+    ctx: &CommandContext,
+) -> CommandResult<Document> {
+    let username = cmd
+        .get_str("grantRolesToUser")
+        .map_err(|_| CommandError::InvalidArgument("missing 'grantRolesToUser' field".into()))?;
+    let roles = parse_roles(cmd, db, "roles")?;
+    ctx.auth
+        .grant_roles(db, username, roles)
+        .map_err(auth_error_to_command_error)?;
+    Ok(doc! { "ok": 1.0 })
+}
+
+async fn handle_revoke_roles_from_user(
+    cmd: &Document,
+    db: &str,
+    ctx: &CommandContext,
+) -> CommandResult<Document> {
+    let username = cmd
+        .get_str("revokeRolesFromUser")
+        .map_err(|_| CommandError::InvalidArgument("missing 'revokeRolesFromUser' field".into()))?;
+    let roles = parse_roles(cmd, db, "roles")?;
+    ctx.auth
+        .revoke_roles(db, username, roles)
+        .map_err(auth_error_to_command_error)?;
+    Ok(doc! { "ok": 1.0 })
+}
+
+fn parse_roles(cmd: &Document, db: &str, key: &str) -> CommandResult<Vec<String>> {
+    let role_values = cmd
+        .get_array(key)
+        .map_err(|_| CommandError::InvalidArgument(format!("missing '{key}' array")))?;
+    let mut roles = Vec::with_capacity(role_values.len());
+    for role_value in role_values {
+        match role_value {
+            Bson::String(role) => roles.push(role.clone()),
+            Bson::Document(role_doc) => {
+                let role = role_doc
+                    .get_str("role")
+                    .map_err(|_| CommandError::InvalidArgument("role document missing 'role'".into()))?;
+                let role_db = role_doc.get_str("db").unwrap_or(db);
+                if role_db == db {
+                    roles.push(role.to_string());
+                } else {
+                    roles.push(format!("{role_db}.{role}"));
+                }
+            }
+            _ => return Err(CommandError::InvalidArgument("roles must be strings or documents".into())),
+        }
+    }
+    Ok(roles)
+}
+
+fn role_to_document(default_db: &str, role: &str) -> Document {
+    if let Some((role_db, role_name)) = role.split_once('.') {
+        doc! { "role": role_name, "db": role_db }
+    } else {
+        doc! { "role": role, "db": default_db }
+    }
+}
+
+fn auth_error_to_command_error(error: rustdb_auth::AuthError) -> CommandError {
+    match error {
+        rustdb_auth::AuthError::UserAlreadyExists(message) => CommandError::DuplicateKey(message),
+        rustdb_auth::AuthError::UserNotFound(message) => CommandError::NamespaceNotFound(message),
+        rustdb_auth::AuthError::Persistence(message) => CommandError::InternalError(message),
+        rustdb_auth::AuthError::AuthenticationFailed => CommandError::AuthenticationFailed,
+        rustdb_auth::AuthError::InvalidPayload(message) => CommandError::InvalidArgument(message),
+        rustdb_auth::AuthError::UnsupportedMechanism(message) => CommandError::InvalidArgument(message),
+        rustdb_auth::AuthError::Disabled => CommandError::Unauthorized("authentication is disabled".into()),
+        rustdb_auth::AuthError::UnknownConversation => {
+            CommandError::InvalidArgument("unknown SASL conversation".into())
+        }
+    }
+}
+
 /// Handle `listDatabases` command.
 async fn handle_list_databases(
     cmd: &Document,

rust/crates/rustdb-commands/src/handlers/auth_handler.rs

@@ -0,0 +1,87 @@
+use bson::{doc, Binary, Bson, Document};
+
+use crate::context::{CommandContext, ConnectionState};
+use crate::error::{CommandError, CommandResult};
+
+pub async fn handle_sasl_start(
+    cmd: &Document,
+    db: &str,
+    ctx: &CommandContext,
+    connection: &mut ConnectionState,
+) -> CommandResult<Document> {
+    let mechanism = cmd
+        .get_str("mechanism")
+        .map_err(|_| CommandError::InvalidArgument("missing SASL mechanism".into()))?;
+    if mechanism != "SCRAM-SHA-256" {
+        return Err(CommandError::InvalidArgument(format!(
+            "unsupported SASL mechanism: {mechanism}"
+        )));
+    }
+
+    let payload = payload_bytes(cmd)?;
+    let result = ctx
+        .auth
+        .start_scram_sha256(db, &payload)
+        .map_err(map_auth_error)?;
+    let conversation_id = connection.next_conversation_id();
+    connection
+        .sasl_conversations
+        .insert(conversation_id, result.conversation);
+
+    Ok(doc! {
+        "conversationId": conversation_id,
+        "done": false,
+        "payload": Binary { subtype: bson::spec::BinarySubtype::Generic, bytes: result.payload },
+        "ok": 1.0,
+    })
+}
+
+pub async fn handle_sasl_continue(
+    cmd: &Document,
+    ctx: &CommandContext,
+    connection: &mut ConnectionState,
+) -> CommandResult<Document> {
+    let conversation_id = cmd
+        .get_i32("conversationId")
+        .map_err(|_| CommandError::InvalidArgument("missing SASL conversationId".into()))?;
+    let payload = payload_bytes(cmd)?;
+    let conversation = connection
+        .sasl_conversations
+        .remove(&conversation_id)
+        .ok_or_else(|| CommandError::InvalidArgument("unknown SASL conversation".into()))?;
+    let result = ctx
+        .auth
+        .continue_scram_sha256(conversation, &payload)
+        .map_err(map_auth_error)?;
+    connection.authenticate(result.user);
+
+    Ok(doc! {
+        "conversationId": conversation_id,
+        "done": true,
+        "payload": Binary { subtype: bson::spec::BinarySubtype::Generic, bytes: result.payload },
+        "ok": 1.0,
+    })
+}
+
+fn payload_bytes(cmd: &Document) -> CommandResult<Vec<u8>> {
+    match cmd.get("payload") {
+        Some(Bson::Binary(binary)) => Ok(binary.bytes.clone()),
+        Some(Bson::String(value)) => Ok(value.as_bytes().to_vec()),
+        _ => Err(CommandError::InvalidArgument("missing SASL payload".into())),
+    }
+}
+
+fn map_auth_error(error: rustdb_auth::AuthError) -> CommandError {
+    match error {
+        rustdb_auth::AuthError::InvalidPayload(message) => CommandError::InvalidArgument(message),
+        rustdb_auth::AuthError::UnsupportedMechanism(message) => CommandError::InvalidArgument(message),
+        rustdb_auth::AuthError::Disabled => CommandError::Unauthorized("authentication is disabled".into()),
+        rustdb_auth::AuthError::UnknownConversation => {
+            CommandError::InvalidArgument("unknown SASL conversation".into())
+        }
+        rustdb_auth::AuthError::AuthenticationFailed => CommandError::AuthenticationFailed,
+        rustdb_auth::AuthError::UserAlreadyExists(message) => CommandError::DuplicateKey(message),
+        rustdb_auth::AuthError::UserNotFound(message) => CommandError::NamespaceNotFound(message),
+        rustdb_auth::AuthError::Persistence(message) => CommandError::InternalError(message),
+    }
+}

rust/crates/rustdb-commands/src/handlers/hello_handler.rs

@@ -1,4 +1,4 @@
-use bson::{doc, Document};
+use bson::{doc, Bson, Document};
 
 use crate::context::CommandContext;
 use crate::error::CommandResult;
@@ -7,12 +7,13 @@ use crate::error::CommandResult;
 ///
 /// Returns server capabilities matching wire protocol expectations.
 pub async fn handle(
-    _cmd: &Document,
+    cmd: &Document,
     _db: &str,
-    _ctx: &CommandContext,
+    ctx: &CommandContext,
 ) -> CommandResult<Document> {
-    Ok(doc! {
+    let mut response = doc! {
         "ismaster": true,
+        "helloOk": true,
         "isWritablePrimary": true,
         "maxBsonObjectSize": 16_777_216_i32,
         "maxMessageSizeBytes": 48_000_000_i32,
@@ -24,5 +25,19 @@ pub async fn handle(
         "maxWireVersion": 21_i32,
         "readOnly": false,
         "ok": 1.0,
-    })
+    };
+
+    if ctx.auth.enabled() {
+        if let Ok(namespace_user) = cmd.get_str("saslSupportedMechs") {
+            let mechanisms: Vec<Bson> = ctx
+                .auth
+                .supported_mechanisms(namespace_user)
+                .into_iter()
+                .map(Bson::String)
+                .collect();
+            response.insert("saslSupportedMechs", Bson::Array(mechanisms));
+        }
+    }
+
+    Ok(response)
 }

rust/crates/rustdb-commands/src/handlers/mod.rs

@@ -1,5 +1,6 @@
 pub mod admin_handler;
 pub mod aggregate_handler;
+pub mod auth_handler;
 pub mod delete_handler;
 pub mod find_handler;
 pub mod hello_handler;

rust/crates/rustdb-commands/src/lib.rs

@@ -3,6 +3,6 @@ pub mod error;
 pub mod handlers;
 mod router;
 
-pub use context::{CommandContext, CursorState};
+pub use context::{CommandContext, ConnectionState, CursorState};
 pub use error::{CommandError, CommandResult};
 pub use router::CommandRouter;

rust/crates/rustdb-commands/src/router.rs

@@ -1,11 +1,12 @@
 use std::sync::Arc;
 
-use bson::Document;
+use bson::{Bson, Document};
 use tracing::{debug, warn};
 
 use rustdb_wire::ParsedCommand;
+use rustdb_auth::AuthAction;
 
-use crate::context::CommandContext;
+use crate::context::{CommandContext, ConnectionState};
 use crate::error::CommandError;
 use crate::handlers;
 
@@ -21,12 +22,46 @@ impl CommandRouter {
     }
 
     /// Route a parsed command to the appropriate handler, returning a BSON response document.
-    pub async fn route(&self, cmd: &ParsedCommand) -> Document {
+    pub async fn route(&self, cmd: &ParsedCommand, connection: &mut ConnectionState) -> Document {
         let db = &cmd.database;
         let command_name = cmd.command_name.as_str();
 
         debug!(command = %command_name, database = %db, "routing command");
 
+        if self.ctx.auth.enabled()
+            && !connection.is_authenticated()
+            && !allows_unauthenticated(command_name)
+        {
+            return CommandError::Unauthorized(format!(
+                "command '{}' requires authentication",
+                command_name,
+            ))
+            .to_error_doc();
+        }
+
+        if self.ctx.auth.enabled() && connection.is_authenticated() {
+            if let Some(action) = required_action(command_name, &cmd.command) {
+                if !self
+                    .ctx
+                    .auth
+                    .is_authorized(&connection.authenticated_users, db, action)
+                {
+                    return CommandError::Unauthorized(format!(
+                        "command '{}' is not authorized for database '{}'",
+                        command_name, db,
+                    ))
+                    .to_error_doc();
+                }
+            }
+        }
+
+        if transaction_command_unsupported(command_name, &cmd.command) {
+            return CommandError::IllegalOperation(
+                "Transaction numbers are only allowed on a replica set member or mongos".into(),
+            )
+            .to_error_doc();
+        }
+
         // Extract session id if present, and touch the session.
         if let Some(lsid) = cmd.command.get("lsid") {
             if let Some(session_id) = rustdb_txn::SessionEngine::extract_session_id(lsid) {
@@ -40,6 +75,14 @@ impl CommandRouter {
                 handlers::hello_handler::handle(&cmd.command, db, &self.ctx).await
             }
 
+            // -- authentication --
+            "saslStart" => {
+                handlers::auth_handler::handle_sasl_start(&cmd.command, db, &self.ctx, connection).await
+            }
+            "saslContinue" => {
+                handlers::auth_handler::handle_sasl_continue(&cmd.command, &self.ctx, connection).await
+            }
+
             // -- query commands --
             "find" => {
                 handlers::find_handler::handle(&cmd.command, db, &self.ctx).await
@@ -88,7 +131,9 @@ impl CommandRouter {
             | "dbStats" | "collStats" | "validate" | "explain"
             | "startSession" | "endSessions" | "killSessions"
             | "commitTransaction" | "abortTransaction"
-            | "saslStart" | "saslContinue" | "authenticate" | "logout"
+            | "authenticate" | "logout"
+            | "createUser" | "updateUser" | "dropUser" | "usersInfo"
+            | "grantRolesToUser" | "revokeRolesFromUser"
             | "currentOp" | "killOp" | "top" | "profile"
             | "compact" | "reIndex" | "fsync" | "connPoolSync" => {
                 handlers::admin_handler::handle(&cmd.command, db, &self.ctx, command_name).await
@@ -107,3 +152,64 @@ impl CommandRouter {
         }
     }
 }
+
+fn allows_unauthenticated(command_name: &str) -> bool {
+    matches!(
+        command_name,
+        "hello" | "ismaster" | "isMaster" | "saslStart" | "saslContinue" | "getnonce"
+    )
+}
+
+fn required_action(command_name: &str, command: &Document) -> Option<AuthAction> {
+    match command_name {
+        "hello" | "ismaster" | "isMaster" | "saslStart" | "saslContinue" | "getnonce" => None,
+        "ping" | "buildInfo" | "buildinfo" | "hostInfo" | "whatsmyuri" | "getLog"
+        | "getCmdLineOpts" | "getParameter" | "getFreeMonitoringStatus" | "setFreeMonitoring"
+        | "getShardMap" | "shardingState" | "atlasVersion" | "connectionStatus"
+        | "startSession" | "endSessions" | "killSessions" | "authenticate" | "logout" => None,
+
+        "find" | "getMore" | "killCursors" | "count" | "distinct" | "listIndexes"
+        | "listCollections" | "collStats" | "dbStats" | "validate" | "explain" => {
+            Some(AuthAction::Read)
+        }
+
+        "aggregate" => Some(if aggregate_writes(command) {
+            AuthAction::Write
+        } else {
+            AuthAction::Read
+        }),
+
+        "insert" | "update" | "findAndModify" | "delete" | "commitTransaction"
+        | "abortTransaction" => Some(AuthAction::Write),
+
+        "createIndexes" | "dropIndexes" | "create" | "drop" | "dropDatabase"
+        | "renameCollection" | "compact" | "reIndex" | "fsync" | "profile" => {
+            Some(AuthAction::DbAdmin)
+        }
+
+        "createUser" | "updateUser" | "dropUser" | "usersInfo" | "grantRolesToUser"
+        | "revokeRolesFromUser" => Some(AuthAction::UserAdmin),
+
+        "serverStatus" | "listDatabases" | "currentOp" | "killOp" | "top" => {
+            Some(AuthAction::ClusterMonitor)
+        }
+
+        _ => None,
+    }
+}
+
+fn aggregate_writes(command: &Document) -> bool {
+    let Ok(pipeline) = command.get_array("pipeline") else {
+        return false;
+    };
+    pipeline.last().and_then(|stage| match stage {
+        Bson::Document(doc) => Some(doc.contains_key("$out") || doc.contains_key("$merge")),
+        _ => None,
+    }).unwrap_or(false)
+}
+
+fn transaction_command_unsupported(command_name: &str, command: &Document) -> bool {
+    matches!(command_name, "commitTransaction" | "abortTransaction")
+        || matches!(command.get("startTransaction"), Some(Bson::Boolean(true)))
+        || matches!(command.get("autocommit"), Some(Bson::Boolean(false)))
+}

rust/crates/rustdb-config/src/lib.rs

@@ -46,6 +46,99 @@ pub struct RustDbOptions {
     /// Interval in ms for periodic persistence (default: 60000)
     #[serde(default = "default_persist_interval")]
     pub persist_interval_ms: u64,
+
+    /// Authentication configuration.
+    #[serde(default)]
+    pub auth: AuthOptions,
+
+    /// TLS transport configuration for TCP listeners.
+    #[serde(default)]
+    pub tls: TlsOptions,
+}
+
+/// Authentication configuration for the embedded server.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct AuthOptions {
+    /// Whether clients must authenticate before issuing protected commands.
+    #[serde(default)]
+    pub enabled: bool,
+
+    /// Bootstrap users loaded at startup. Passwords are converted into SCRAM credentials in memory.
+    #[serde(default)]
+    pub users: Vec<AuthUserOptions>,
+
+    /// Optional path for persisted SCRAM user metadata. Stores derived credentials, never plaintext passwords.
+    #[serde(default)]
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub users_path: Option<String>,
+
+    /// SCRAM iteration count used for bootstrap credentials.
+    #[serde(default = "default_scram_iterations")]
+    pub scram_iterations: u32,
+}
+
+impl Default for AuthOptions {
+    fn default() -> Self {
+        Self {
+            enabled: false,
+            users: Vec::new(),
+            users_path: None,
+            scram_iterations: default_scram_iterations(),
+        }
+    }
+}
+
+/// TLS transport configuration for the embedded server.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct TlsOptions {
+    /// Whether TCP client connections must use TLS.
+    #[serde(default)]
+    pub enabled: bool,
+
+    /// PEM-encoded server certificate chain.
+    #[serde(default)]
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub cert_path: Option<String>,
+
+    /// PEM-encoded server private key.
+    #[serde(default)]
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub key_path: Option<String>,
+
+    /// PEM-encoded client CA roots for mTLS verification.
+    #[serde(default)]
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub ca_path: Option<String>,
+
+    /// Require clients to present a certificate signed by caPath.
+    #[serde(default)]
+    pub require_client_cert: bool,
+}
+
+impl Default for TlsOptions {
+    fn default() -> Self {
+        Self {
+            enabled: false,
+            cert_path: None,
+            key_path: None,
+            ca_path: None,
+            require_client_cert: false,
+        }
+    }
+}
+
+/// A bootstrap user for SCRAM authentication.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct AuthUserOptions {
+    pub username: String,
+    pub password: String,
+    #[serde(default = "default_auth_database")]
+    pub database: String,
+    #[serde(default)]
+    pub roles: Vec<String>,
 }
 
 fn default_port() -> u16 {
@@ -60,6 +153,14 @@ fn default_persist_interval() -> u64 {
     60000
 }
 
+fn default_scram_iterations() -> u32 {
+    15000
+}
+
+fn default_auth_database() -> String {
+    "admin".to_string()
+}
+
 impl Default for RustDbOptions {
     fn default() -> Self {
         Self {
@@ -70,6 +171,8 @@ impl Default for RustDbOptions {
             storage_path: None,
             persist_path: None,
             persist_interval_ms: default_persist_interval(),
+            auth: AuthOptions::default(),
+            tls: TlsOptions::default(),
         }
     }
 }
@@ -92,6 +195,59 @@ impl RustDbOptions {
                 "storagePath is required when storage is 'file'".to_string(),
             ));
         }
+        if self.auth.enabled {
+            if self.auth.users.is_empty() && self.auth.users_path.is_none() {
+                return Err(ConfigError::ValidationError(
+                    "auth.users or auth.usersPath must be set when auth.enabled is true".to_string(),
+                ));
+            }
+            if self.auth.scram_iterations < 4096 {
+                return Err(ConfigError::ValidationError(
+                    "auth.scramIterations must be at least 4096".to_string(),
+                ));
+            }
+            for user in &self.auth.users {
+                if user.username.is_empty() {
+                    return Err(ConfigError::ValidationError(
+                        "auth.users[].username must not be empty".to_string(),
+                    ));
+                }
+                if user.password.is_empty() {
+                    return Err(ConfigError::ValidationError(
+                        format!("auth user '{}' must have a non-empty password", user.username),
+                    ));
+                }
+                if user.database.is_empty() {
+                    return Err(ConfigError::ValidationError(
+                        format!("auth user '{}' must have a non-empty database", user.username),
+                    ));
+                }
+            }
+        }
+        if self.tls.enabled {
+            if self.socket_path.is_some() {
+                return Err(ConfigError::ValidationError(
+                    "tls.enabled is only supported for TCP listeners".to_string(),
+                ));
+            }
+            if self.tls.cert_path.as_deref().unwrap_or_default().is_empty() {
+                return Err(ConfigError::ValidationError(
+                    "tls.certPath is required when tls.enabled is true".to_string(),
+                ));
+            }
+            if self.tls.key_path.as_deref().unwrap_or_default().is_empty() {
+                return Err(ConfigError::ValidationError(
+                    "tls.keyPath is required when tls.enabled is true".to_string(),
+                ));
+            }
+            if self.tls.require_client_cert
+                && self.tls.ca_path.as_deref().unwrap_or_default().is_empty()
+            {
+                return Err(ConfigError::ValidationError(
+                    "tls.caPath is required when tls.requireClientCert is true".to_string(),
+                ));
+            }
+        }
         Ok(())
     }
 
@@ -101,7 +257,12 @@ impl RustDbOptions {
             let encoded = urlencoding(socket_path);
             format!("mongodb://{}", encoded)
         } else {
-            format!("mongodb://{}:{}", self.host, self.port)
+            let base = format!("mongodb://{}:{}", self.host, self.port);
+            if self.tls.enabled {
+                format!("{}/?tls=true", base)
+            } else {
+                base
+            }
         }
     }
 }

rust/crates/rustdb-storage/src/file.rs

@@ -187,6 +187,27 @@ impl CollectionState {
     }
 }
 
+fn truncate_invalid_tail(
+    data_path: &PathBuf,
+    stats: &crate::keydir::BuildStats,
+) -> StorageResult<()> {
+    if stats.invalid_tail_bytes == 0 {
+        return Ok(());
+    }
+
+    tracing::warn!(
+        path = %data_path.display(),
+        valid_data_end = stats.valid_data_end,
+        invalid_tail_bytes = stats.invalid_tail_bytes,
+        "truncating invalid data file tail"
+    );
+
+    let file = std::fs::OpenOptions::new().write(true).open(data_path)?;
+    file.set_len(stats.valid_data_end)?;
+    file.sync_all()?;
+    Ok(())
+}
+
 // ---------------------------------------------------------------------------
 // Collection cache key: "db\0coll"
 // ---------------------------------------------------------------------------
@@ -279,7 +300,8 @@ impl FileStorageAdapter {
                                 hint_path, stored_size, actual_size
                             );
                         }
-                        let (kd, dead, _stats) = KeyDir::build_from_data_file(&data_path)?;
+                        let (kd, dead, stats) = KeyDir::build_from_data_file(&data_path)?;
+                        truncate_invalid_tail(&data_path, &stats)?;
                         (kd, dead, false)
                     } else {
                         // Size matches — validate entry integrity with spot-checks
@@ -296,19 +318,22 @@ impl FileStorageAdapter {
                             (kd, dead, true)
                         } else {
                             tracing::warn!("hint file {:?} failed validation, rebuilding from data file", hint_path);
-                            let (kd, dead, _stats) = KeyDir::build_from_data_file(&data_path)?;
+                            let (kd, dead, stats) = KeyDir::build_from_data_file(&data_path)?;
+                            truncate_invalid_tail(&data_path, &stats)?;
                             (kd, dead, false)
                         }
                     }
                 }
                 _ => {
                     debug!("hint file invalid, rebuilding KeyDir from data file");
-                    let (kd, dead, _stats) = KeyDir::build_from_data_file(&data_path)?;
+                    let (kd, dead, stats) = KeyDir::build_from_data_file(&data_path)?;
+                    truncate_invalid_tail(&data_path, &stats)?;
                     (kd, dead, false)
                 }
             }
         } else if data_path.exists() {
-            let (kd, dead, _stats) = KeyDir::build_from_data_file(&data_path)?;
+            let (kd, dead, stats) = KeyDir::build_from_data_file(&data_path)?;
+            truncate_invalid_tail(&data_path, &stats)?;
             (kd, dead, false)
         } else {
             (KeyDir::new(), 0, false)

rust/crates/rustdb-storage/src/keydir.rs

@@ -14,7 +14,7 @@ use dashmap::DashMap;
 
 use crate::error::{StorageError, StorageResult};
 use crate::record::{
-    DataRecord, FileHeader, FileType, RecordScanner, FILE_HEADER_SIZE, FORMAT_VERSION,
+    DataRecord, FileHeader, FileType, FILE_HEADER_SIZE, FORMAT_VERSION,
 };
 
 // ---------------------------------------------------------------------------
@@ -49,6 +49,10 @@ pub struct BuildStats {
     pub tombstones: u64,
     /// Number of records superseded by a later write for the same key.
     pub superseded_records: u64,
+    /// Byte offset immediately after the last valid record.
+    pub valid_data_end: u64,
+    /// Number of invalid tail bytes after the last valid record.
+    pub invalid_tail_bytes: u64,
 }
 
 // ---------------------------------------------------------------------------
@@ -137,6 +141,7 @@ impl KeyDir {
     /// stale records (superseded by later writes or tombstoned).
     pub fn build_from_data_file(path: &Path) -> StorageResult<(Self, u64, BuildStats)> {
         let file = std::fs::File::open(path)?;
+        let file_len = file.metadata()?.len();
         let mut reader = BufReader::new(file);
 
         // Read and validate file header
@@ -152,13 +157,49 @@ impl KeyDir {
 
         let keydir = KeyDir::new();
         let mut dead_bytes: u64 = 0;
-        let mut stats = BuildStats::default();
+        let mut stats = BuildStats {
+            valid_data_end: FILE_HEADER_SIZE as u64,
+            ..BuildStats::default()
+        };
 
-        let scanner = RecordScanner::new(reader, FILE_HEADER_SIZE as u64);
-        for result in scanner {
-            let (offset, record) = result?;
+        loop {
+            let record_offset = stats.valid_data_end;
+            let (record, disk_size) = match DataRecord::decode_from(&mut reader) {
+                Ok(Some((record, disk_size))) => (record, disk_size),
+                Ok(None) => {
+                    if file_len > record_offset {
+                        stats.invalid_tail_bytes = file_len - record_offset;
+                    }
+                    break;
+                }
+                Err(StorageError::IoError(e)) if e.kind() == io::ErrorKind::UnexpectedEof => {
+                    stats.invalid_tail_bytes = file_len.saturating_sub(record_offset);
+                    break;
+                }
+                Err(StorageError::ChecksumMismatch { expected, actual }) => {
+                    tracing::warn!(
+                        path = %path.display(),
+                        offset = record_offset,
+                        "stopping data file scan at checksum mismatch: expected 0x{expected:08X}, got 0x{actual:08X}"
+                    );
+                    stats.invalid_tail_bytes = file_len.saturating_sub(record_offset);
+                    break;
+                }
+                Err(StorageError::CorruptRecord(message)) => {
+                    tracing::warn!(
+                        path = %path.display(),
+                        offset = record_offset,
+                        "stopping data file scan at corrupt record: {message}"
+                    );
+                    stats.invalid_tail_bytes = file_len.saturating_sub(record_offset);
+                    break;
+                }
+                Err(e) => return Err(e),
+            };
+
+            stats.valid_data_end += disk_size as u64;
             let is_tombstone = record.is_tombstone();
-            let disk_size = record.disk_size() as u32;
+            let disk_size = disk_size as u32;
             let value_len = record.value.len() as u32;
             let timestamp = record.timestamp;
             let key = String::from_utf8(record.key)
@@ -175,7 +216,7 @@ impl KeyDir {
                 dead_bytes += disk_size as u64;
             } else {
                 let entry = KeyDirEntry {
-                    offset,
+                    offset: record_offset,
                     record_len: disk_size,
                     value_len,
                     timestamp,

rust/crates/rustdb/Cargo.toml

@@ -21,9 +21,12 @@ rustdb-query = { workspace = true }
 rustdb-storage = { workspace = true }
 rustdb-index = { workspace = true }
 rustdb-txn = { workspace = true }
+rustdb-auth = { workspace = true }
 rustdb-commands = { workspace = true }
 tokio = { workspace = true }
 tokio-util = { workspace = true }
+tokio-rustls = { workspace = true }
+rustls-pemfile = { workspace = true }
 tracing = { workspace = true }
 tracing-subscriber = { workspace = true }
 clap = { workspace = true }

rust/crates/rustdb/src/lib.rs

@@ -1,10 +1,12 @@
 pub mod management;
 
+use std::fs::File;
+use std::io::BufReader;
 use std::path::PathBuf;
 use std::sync::Arc;
 use std::time::Duration;
 
-use anyhow::Result;
+use anyhow::{Context, Result};
 use dashmap::DashMap;
 use tokio::net::TcpListener;
 #[cfg(unix)]
@@ -12,13 +14,17 @@ use tokio::net::UnixListener;
 use tokio_util::codec::Framed;
 use tokio_util::sync::CancellationToken;
 
-use rustdb_config::{RustDbOptions, StorageType};
+use rustdb_config::{RustDbOptions, StorageType, TlsOptions};
 use rustdb_wire::{WireCodec, OP_QUERY};
 use rustdb_wire::{encode_op_msg_response, encode_op_reply_response};
 use rustdb_storage::{StorageAdapter, MemoryStorageAdapter, FileStorageAdapter, OpLog};
 use rustdb_index::{IndexEngine, IndexOptions};
 use rustdb_txn::{TransactionEngine, SessionEngine};
-use rustdb_commands::{CommandRouter, CommandContext};
+use rustdb_auth::AuthEngine;
+use rustdb_commands::{CommandRouter, CommandContext, ConnectionState};
+use tokio_rustls::rustls::{RootCertStore, ServerConfig};
+use tokio_rustls::rustls::server::WebPkiClientVerifier;
+use tokio_rustls::TlsAcceptor;
 
 /// The main RustDb server.
 pub struct RustDb {
@@ -150,6 +156,8 @@ impl RustDb {
             }
         }
 
+        let auth = Arc::new(AuthEngine::from_options(&options.auth)?);
+
         let ctx = Arc::new(CommandContext {
             storage,
             indexes,
@@ -158,6 +166,7 @@ impl RustDb {
             cursors: Arc::new(DashMap::new()),
             start_time: std::time::Instant::now(),
             oplog: Arc::new(OpLog::new()),
+            auth,
         });
 
         let router = Arc::new(CommandRouter::new(ctx.clone()));
@@ -215,7 +224,12 @@ impl RustDb {
         } else {
             let addr = format!("{}:{}", self.options.host, self.options.port);
             let listener = TcpListener::bind(&addr).await?;
-            tracing::info!("RustDb listening on {}", addr);
+            let tls_acceptor = if self.options.tls.enabled {
+                Some(build_tls_acceptor(&self.options.tls)?)
+            } else {
+                None
+            };
+            tracing::info!(tls = self.options.tls.enabled, "RustDb listening on {}", addr);
 
             let handle = tokio::spawn(async move {
                 loop {
@@ -226,9 +240,21 @@ impl RustDb {
                                 Ok((stream, _addr)) => {
                                     let _ = stream.set_nodelay(true);
                                     let router = router.clone();
-                                    tokio::spawn(async move {
-                                        handle_connection(stream, router).await;
-                                    });
+                                    match tls_acceptor.clone() {
+                                        Some(acceptor) => {
+                                            tokio::spawn(async move {
+                                                match acceptor.accept(stream).await {
+                                                    Ok(tls_stream) => handle_connection(tls_stream, router).await,
+                                                    Err(e) => tracing::debug!("TLS handshake failed: {}", e),
+                                                }
+                                            });
+                                        }
+                                        None => {
+                                            tokio::spawn(async move {
+                                                handle_connection(stream, router).await;
+                                            });
+                                        }
+                                    }
                                 }
                                 Err(e) => {
                                     tracing::error!("Accept error: {}", e);
@@ -275,14 +301,88 @@ impl RustDb {
     }
 }
 
+fn build_tls_acceptor(options: &TlsOptions) -> Result<TlsAcceptor> {
+    let cert_path = options
+        .cert_path
+        .as_deref()
+        .context("tls.certPath is required when tls.enabled is true")?;
+    let key_path = options
+        .key_path
+        .as_deref()
+        .context("tls.keyPath is required when tls.enabled is true")?;
+
+    let certs = load_certs(cert_path)?;
+    let key = load_private_key(key_path)?;
+
+    let config = if options.require_client_cert {
+        let ca_path = options
+            .ca_path
+            .as_deref()
+            .context("tls.caPath is required when tls.requireClientCert is true")?;
+        let roots = load_root_store(ca_path)?;
+        let verifier = WebPkiClientVerifier::builder(Arc::new(roots))
+            .build()
+            .context("failed to build TLS client certificate verifier")?;
+        ServerConfig::builder()
+            .with_client_cert_verifier(verifier)
+            .with_single_cert(certs, key)
+            .context("failed to build TLS server configuration")?
+    } else {
+        ServerConfig::builder()
+            .with_no_client_auth()
+            .with_single_cert(certs, key)
+            .context("failed to build TLS server configuration")?
+    };
+
+    Ok(TlsAcceptor::from(Arc::new(config)))
+}
+
+fn load_certs(path: &str) -> Result<Vec<tokio_rustls::rustls::pki_types::CertificateDer<'static>>> {
+    let file = File::open(path).with_context(|| format!("failed to open TLS certificate file '{}'", path))?;
+    let mut reader = BufReader::new(file);
+    let certs = rustls_pemfile::certs(&mut reader)
+        .collect::<std::result::Result<Vec<_>, _>>()
+        .with_context(|| format!("failed to parse TLS certificate file '{}'", path))?;
+
+    if certs.is_empty() {
+        anyhow::bail!("TLS certificate file '{}' did not contain any certificates", path);
+    }
+
+    Ok(certs)
+}
+
+fn load_private_key(path: &str) -> Result<tokio_rustls::rustls::pki_types::PrivateKeyDer<'static>> {
+    let file = File::open(path).with_context(|| format!("failed to open TLS private key file '{}'", path))?;
+    let mut reader = BufReader::new(file);
+    rustls_pemfile::private_key(&mut reader)
+        .with_context(|| format!("failed to parse TLS private key file '{}'", path))?
+        .with_context(|| format!("TLS private key file '{}' did not contain a private key", path))
+}
+
+fn load_root_store(path: &str) -> Result<RootCertStore> {
+    let mut roots = RootCertStore::empty();
+    for cert in load_certs(path)? {
+        roots
+            .add(cert)
+            .with_context(|| format!("failed to add TLS client CA certificate from '{}'", path))?;
+    }
+
+    if roots.is_empty() {
+        anyhow::bail!("TLS client CA file '{}' did not contain usable certificates", path);
+    }
+
+    Ok(roots)
+}
+
 /// Handle a single client connection using the wire protocol codec.
 async fn handle_connection<S>(stream: S, router: Arc<CommandRouter>)
 where
-    S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin,
+    S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin + Send + 'static,
 {
     use futures_util::{SinkExt, StreamExt};
 
     let mut framed = Framed::new(stream, WireCodec);
+    let mut connection = ConnectionState::new();
 
     while let Some(result) = framed.next().await {
         match result {
@@ -290,7 +390,7 @@ where
                 let request_id = parsed_cmd.request_id;
                 let op_code = parsed_cmd.op_code;
 
-                let response_doc = router.route(&parsed_cmd).await;
+                let response_doc = router.route(&parsed_cmd, &mut connection).await;
 
                 let response_id = next_request_id();
 

rust/crates/rustdb/src/management.rs

@@ -167,6 +167,9 @@ async fn handle_start(
         Ok(o) => o,
         Err(e) => return ManagementResponse::err(id.to_string(), format!("Invalid config: {}", e)),
     };
+    if let Err(e) = options.validate() {
+        return ManagementResponse::err(id.to_string(), format!("Invalid config: {}", e));
+    }
 
     let connection_uri = options.connection_uri();