feat(oncp): OTP enrollment tokens and dynamic IPAM

- Add OTP token management (oncp/src/token.rs)
  * Time-limited tokens with expiry (default 3 minutes)
  * One-time use validation (token deleted after use)
  * Automatic cleanup of expired tokens
  * Cryptographically secure random generation (10 chars)
  * Token masking in logs (XX****XX format)

- Add dynamic IPAM (oncp/src/network.rs)
  * NetworkConfig for 10.X.0.0/16 subnet management
  * IpamPool with sequential IP allocation
  * Master node octet validation (0-255, excluding reserved)
  * IP release mechanism for rollback scenarios

- Update enrollment flow
  * EnrollmentRequest requires OTP token field
  * Silent drop (HTTP 444) for invalid tokens
  * IP allocation during node approval
  * CLI command: 'node token --expiry 3'
  * Master CLI --network-octet parameter

- Security enhancements
  * Two-factor enrollment: token + admin approval
  * Token enumeration prevention (no error responses)
  * Automatic token cleanup every 60 seconds
  * PSK + assigned IP returned on approval

Tests: All 16 tests passing (4 token, 5 network, 7 existing)

Cargo.lock

@@ -2732,6 +2732,7 @@ dependencies = [
  "hyper 1.8.1",
  "image 0.24.9",
  "ostp",
+ "parking_lot",
  "qrcode",
  "rand 0.8.5",
  "rusqlite",

oncp-master/src/main.rs

@@ -39,6 +39,10 @@ enum Commands {
         /// Listen address
         #[arg(short, long, default_value = "0.0.0.0:8080")]
         listen: String,
+        
+        /// Network octet for 10.X.0.0/16 subnet (0-255)
+        #[arg(short = 'n', long, default_value = "42")]
+        network_octet: u8,
     },
     
     /// Interactive dashboard
@@ -99,6 +103,12 @@ enum NodeCommands {
         /// Node ID to reject
         id: String,
     },
+    /// Generate enrollment token
+    Token {
+        /// Token expiry duration in minutes
+        #[arg(long, default_value = "3")]
+        expiry: i64,
+    },
 }
 
 #[derive(Subcommand)]
@@ -155,13 +165,35 @@ async fn main() -> Result<()> {
         .with(tracing_subscriber::fmt::layer())
         .init();
     
-    let state = Arc::new(AppState::new(&cli.database)?);
+    let state = Arc::new(AppState::new(&cli.database, 42)?);  // Default network octet
     
     // Initialize default SNIs
     state.sni_manager.init_defaults().await;
     
+    // Start background token cleanup
+    let cleanup_state = state.clone();
+    tokio::spawn(async move {
+        loop {
+            tokio::time::sleep(Duration::from_secs(60)).await;
+            cleanup_state.token_manager.cleanup_expired();
+        }
+    });
+    
     match cli.command {
-        Commands::Serve { listen } => {
+        Commands::Serve { listen, network_octet } => {
+            // Update state with correct network octet
+            let state = Arc::new(AppState::new(&cli.database, network_octet)?);
+            state.sni_manager.init_defaults().await;
+            
+            // Start token cleanup for this state too
+            let cleanup_state = state.clone();
+            tokio::spawn(async move {
+                loop {
+                    tokio::time::sleep(Duration::from_secs(60)).await;
+                    cleanup_state.token_manager.cleanup_expired();
+                }
+            });
+            
             run_api_server(state, &listen).await
         }
         Commands::Dashboard => {
@@ -376,18 +408,25 @@ async fn handle_node_command(state: Arc<AppState>, action: NodeCommands) -> Resu
         NodeCommands::Approve { id } => {
             let uuid = uuid::Uuid::parse_str(&id)?;
             
-            match state.enrollment.approve(&uuid) {
+            // Allocate IP from IPAM pool
+            let assigned_ip = state.ipam_pool.lock().allocate()
+                .map_err(|e| anyhow::anyhow!("Failed to allocate IP: {}", e))?;
+            
+            match state.enrollment.approve(&uuid, &assigned_ip.to_string()) {
                 Ok(node_psk) => {
                     println!("{} Node approved", style("✓").green().bold());
                     println!();
-                    println!("  Node ID:  {}", style(uuid).yellow());
+                    println!("  Node ID:     {}", style(uuid).yellow());
-                    println!("  Node PSK: {}", style(&node_psk).yellow().bold());
+                    println!("  Assigned IP: {}", style(&assigned_ip).cyan().bold());
+                    println!("  Node PSK:    {}", style(&node_psk).yellow().bold());
                     println!();
                     println!("{}", style("⚠ IMPORTANT: Save this PSK securely!").red().bold());
                     println!("{}", style("  Send it to the node operator via secure channel").dim());
                     println!("{}", style("  The node must use this PSK to connect").dim());
                 }
                 Err(e) => {
+                    // Release IP on failure
+                    let _ = state.ipam_pool.lock().release(assigned_ip);
                     println!("{} Failed to approve: {}", style("✗").red().bold(), e);
                 }
             }
@@ -405,6 +444,19 @@ async fn handle_node_command(state: Arc<AppState>, action: NodeCommands) -> Resu
                 }
             }
         }
+        
+        NodeCommands::Token { expiry } => {
+            let token = state.token_manager.generate(expiry);
+            
+            println!("{} Enrollment token generated", style("✓").green().bold());
+            println!();
+            println!("  Token:   {}", style(&token).yellow().bold());
+            println!("  Expires: {} minutes", style(expiry).cyan());
+            println!();
+            println!("{}", style("⚠ IMPORTANT: This token can only be used once!").red().bold());
+            println!("{}", style("  Send it to the node operator via secure channel").dim());
+            println!("{}", style("  Token will expire and be deleted after use").dim());
+        }
     }
     
     Ok(())

oncp/Cargo.toml

@@ -23,3 +23,4 @@ qrcode = "0.14"
 image = { version = "0.24", default-features = false, features = ["png"] }
 hex = "0.4"
 rand = "0.8"
+parking_lot = "0.12.5"

oncp/src/api.rs

@@ -16,9 +16,11 @@ use uuid::Uuid;
 
 use crate::billing::{SqliteRegistry, User, UserRegistry};
 use crate::enrollment::{EnrollmentRegistry, EnrollmentRequest, EnrollmentState};
+use crate::network::{NetworkConfig, IpamPool};
 use crate::node::{NetworkStats, Node, NodeCheckin, NodeRegistry};
 use crate::session::SessionManager;
 use crate::sni::{SniManager, SniUpdate};
+use crate::token::TokenManager;
 
 /// Shared application state
 pub struct AppState {
@@ -27,16 +29,23 @@ pub struct AppState {
     pub sessions: SessionManager,
     pub sni_manager: SniManager,
     pub enrollment: EnrollmentRegistry,
+    pub token_manager: Arc<TokenManager>,
+    pub ipam_pool: Arc<parking_lot::Mutex<IpamPool>>,
 }
 
 impl AppState {
-    pub fn new(db_path: &str) -> anyhow::Result<Self> {
+    pub fn new(db_path: &str, network_octet: u8) -> anyhow::Result<Self> {
+        let network_config = NetworkConfig::new(network_octet)?;
+        let ipam_pool = IpamPool::new(network_config);
+        
         Ok(Self {
             nodes: NodeRegistry::new(60), // 60 second timeout
             users: SqliteRegistry::new(db_path)?,
             sessions: SessionManager::new(300), // 5 minute heartbeat timeout
             sni_manager: SniManager::new(),
             enrollment: EnrollmentRegistry::new(db_path)?,
+            token_manager: Arc::new(TokenManager::new()),
+            ipam_pool: Arc::new(parking_lot::Mutex::new(ipam_pool)),
         })
     }
 }
@@ -429,6 +438,7 @@ async fn emergency_sni_update(
 /// Submit node enrollment request
 #[derive(Debug, Deserialize)]
 struct SubmitEnrollmentRequest {
+    token: String,           // OTP enrollment token (required)
     name: String,
     address: String,
     country_code: String,
@@ -447,8 +457,22 @@ async fn submit_enrollment(
     State(state): State<Arc<AppState>>,
     Json(req): Json<SubmitEnrollmentRequest>,
 ) -> impl IntoResponse {
+    // SECURITY: Validate token first - silent drop on failure
+    if let Err(e) = state.token_manager.validate(&req.token) {
+        tracing::warn!("Enrollment rejected: invalid token ({})", e);
+        // Silent drop - return 444 (nginx-style connection close without response)
+        // This prevents token enumeration attacks
+        return (StatusCode::from_u16(444).unwrap(), Json(SubmitEnrollmentResponse {
+            node_id: Uuid::nil(),
+            state: "error".into(),
+            message: "".into(),
+        }));
+    }
+
     let node_id = Uuid::new_v4();
+    let node_name = req.name.clone();
     let enrollment_req = EnrollmentRequest {
+        token: req.token,
         node_id,
         name: req.name,
         address: req.address,
@@ -459,22 +483,28 @@ async fn submit_enrollment(
     };
 
     match state.enrollment.submit_request(enrollment_req) {
-        Ok(()) => (
+        Ok(()) => {
-            StatusCode::CREATED,
+            tracing::info!("Enrollment request submitted: {} ({})", node_id, node_name);
-            Json(SubmitEnrollmentResponse {
+            (
-                node_id,
+                StatusCode::CREATED,
-                state: "pending".into(),
+                Json(SubmitEnrollmentResponse {
-                message: "Enrollment request submitted. Awaiting approval.".into(),
+                    node_id,
-            }),
+                    state: "pending".into(),
-        ),
+                    message: "Enrollment request submitted. Awaiting approval.".into(),
-        Err(e) => (
+                }),
-            StatusCode::INTERNAL_SERVER_ERROR,
+            )
-            Json(SubmitEnrollmentResponse {
+        },
-                node_id,
+        Err(e) => {
-                state: "error".into(),
+            tracing::error!("Enrollment submission failed: {}", e);
-                message: format!("Failed to submit: {}", e),
+            (
-            }),
+                StatusCode::INTERNAL_SERVER_ERROR,
-        ),
+                Json(SubmitEnrollmentResponse {
+                    node_id,
+                    state: "error".into(),
+                    message: format!("Failed to submit: {}", e),
+                }),
+            )
+        },
     }
 }
 
@@ -493,6 +523,7 @@ async fn list_pending_enrollments(
 struct ApproveEnrollmentResponse {
     node_id: Uuid,
     node_psk: String,
+    assigned_ip: String,
     message: String,
 }
 
@@ -500,23 +531,48 @@ async fn approve_enrollment(
     State(state): State<Arc<AppState>>,
     Path(id): Path<Uuid>,
 ) -> impl IntoResponse {
-    match state.enrollment.approve(&id) {
+    // Allocate IP from IPAM pool
-        Ok(node_psk) => (
+    let assigned_ip = match state.ipam_pool.lock().allocate() {
-            StatusCode::OK,
+        Ok(ip) => ip,
-            Json(ApproveEnrollmentResponse {
+        Err(e) => {
-                node_id: id,
+            return (
-                node_psk,
+                StatusCode::INTERNAL_SERVER_ERROR,
-                message: "Node approved. Use provided PSK to connect.".into(),
+                Json(ApproveEnrollmentResponse {
-            }),
+                    node_id: id,
-        ),
+                    node_psk: String::new(),
-        Err(e) => (
+                    assigned_ip: String::new(),
-            StatusCode::BAD_REQUEST,
+                    message: format!("Failed to allocate IP: {}", e),
-            Json(ApproveEnrollmentResponse {
+                }),
-                node_id: id,
+            );
-                node_psk: String::new(),
+        }
-                message: format!("Failed to approve: {}", e),
+    };
-            }),
+
-        ),
+    match state.enrollment.approve(&id, &assigned_ip.to_string()) {
+        Ok(node_psk) => {
+            tracing::info!("Node {} approved with IP {}", id, assigned_ip);
+            (
+                StatusCode::OK,
+                Json(ApproveEnrollmentResponse {
+                    node_id: id,
+                    node_psk,
+                    assigned_ip: assigned_ip.to_string(),
+                    message: "Node approved. Use provided PSK and IP to connect.".into(),
+                }),
+            )
+        },
+        Err(e) => {
+            // Release IP on failure
+            let _ = state.ipam_pool.lock().release(assigned_ip);
+            (
+                StatusCode::BAD_REQUEST,
+                Json(ApproveEnrollmentResponse {
+                    node_id: id,
+                    node_psk: String::new(),
+                    assigned_ip: String::new(),
+                    message: format!("Failed to approve: {}", e),
+                }),
+            )
+        },
     }
 }
 

oncp/src/enrollment.rs

@@ -56,6 +56,7 @@ impl EnrollmentState {
 /// Node enrollment request
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct EnrollmentRequest {
+    pub token: String,           // OTP enrollment token (required)
     pub node_id: Uuid,
     pub name: String,
     pub address: String,         // Public IP:port
@@ -189,22 +190,37 @@ impl EnrollmentRegistry {
         Ok(rows.collect::<SqliteResult<Vec<_>>>()?)
     }
 
-    /// Approve enrollment request and generate Node PSK
+    /// Approve enrollment request and generate Node PSK with IP allocation
-    pub fn approve(&self, node_id: &Uuid) -> Result<String, EnrollmentError> {
+    pub fn approve(&self, node_id: &Uuid, assigned_ip: &str) -> Result<String, EnrollmentError> {
         let conn = self.conn.lock().unwrap();
         
         // Generate unique PSK for this node
         let node_psk = Self::generate_node_psk();
         let now = Utc::now().to_rfc3339();
 
+        // First check if we need to add assigned_ip column
+        let has_ip_column = conn.query_row(
+            "SELECT COUNT(*) FROM pragma_table_info('enrollments') WHERE name='assigned_ip'",
+            [],
+            |row| row.get::<_, i32>(0)
+        ).unwrap_or(0) > 0;
+
+        if !has_ip_column {
+            conn.execute(
+                "ALTER TABLE enrollments ADD COLUMN assigned_ip TEXT",
+                [],
+            )?;
+        }
+
         let updated = conn.execute(
             "UPDATE enrollments 
-             SET state = ?1, node_psk = ?2, approved_at = ?3
+             SET state = ?1, node_psk = ?2, approved_at = ?3, assigned_ip = ?4
-             WHERE node_id = ?4 AND state = ?5",
+             WHERE node_id = ?5 AND state = ?6",
             params![
                 EnrollmentState::Approved.as_str(),
                 node_psk,
                 now,
+                assigned_ip,
                 node_id.to_string(),
                 EnrollmentState::Pending.as_str(),
             ],
@@ -214,7 +230,7 @@ impl EnrollmentRegistry {
             return Err(EnrollmentError::InvalidTransition);
         }
 
-        tracing::info!("Node {} approved", node_id);
+        tracing::info!("Node {} approved with IP {}", node_id, assigned_ip);
         Ok(node_psk)
     }
 
@@ -352,6 +368,7 @@ mod tests {
 
         // Submit request
         let req = EnrollmentRequest {
+            token: "TEST_TOKEN_123".into(),
             node_id,
             name: "test-node".into(),
             address: "1.2.3.4:8443".into(),
@@ -367,7 +384,7 @@ mod tests {
         assert_eq!(pending.len(), 1);
 
         // Approve
-        let psk = registry.approve(&node_id)?;
+        let psk = registry.approve(&node_id, "10.42.0.2")?;
         assert_eq!(psk.len(), 64); // 32 bytes hex
 
         // Check approved

oncp/src/lib.rs

@@ -1,9 +1,11 @@
 pub mod api;
 pub mod billing;
 pub mod enrollment;
+pub mod network;
 pub mod node;
 pub mod session;
 pub mod sni;
+pub mod token;
 
 pub use api::{create_router, run_server, AppState};
 pub use billing::{BillingError, SqliteRegistry, User, UserRegistry};

oncp/src/network.rs

@@ -0,0 +1,246 @@
+//! Dynamic Network Configuration & IPAM
+//!
+//! Each Master Node manages its own /16 subnet: 10.X.0.0/16
+//! where X is the unique Master Node identifier (0-255)
+
+use serde::{Deserialize, Serialize};
+use std::collections::HashSet;
+use std::net::Ipv4Addr;
+use std::sync::{Arc, Mutex};
+use thiserror::Error;
+
+#[derive(Error, Debug)]
+pub enum IpamError {
+    #[error("invalid network octet: must be 0-255, got {0}")]
+    InvalidOctet(u8),
+    #[error("reserved network range: {0}")]
+    ReservedRange(String),
+    #[error("no available IPs in pool")]
+    PoolExhausted,
+    #[error("IP already allocated: {0}")]
+    AlreadyAllocated(Ipv4Addr),
+}
+
+/// Network configuration for a Master Node
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct NetworkConfig {
+    /// Second octet of 10.X.0.0/16 network
+    pub master_octet: u8,
+    /// Full network prefix: 10.X.0.0/16
+    pub network: String,
+    /// Master node IP (always 10.X.0.1)
+    pub master_ip: Ipv4Addr,
+    /// Gateway IP (always 10.X.0.1)
+    pub gateway: Ipv4Addr,
+    /// Network mask
+    pub netmask: Ipv4Addr,
+    /// CIDR notation
+    pub cidr: u8,
+}
+
+impl NetworkConfig {
+    /// Create a new network configuration for a Master Node
+    pub fn new(master_octet: u8) -> Result<Self, IpamError> {
+        // Check for reserved ranges
+        if is_reserved_octet(master_octet) {
+            return Err(IpamError::ReservedRange(format!("10.{}.0.0/16", master_octet)));
+        }
+
+        let master_ip = Ipv4Addr::new(10, master_octet, 0, 1);
+        let gateway = master_ip;
+        let netmask = Ipv4Addr::new(255, 255, 0, 0);
+
+        Ok(Self {
+            master_octet,
+            network: format!("10.{}.0.0/16", master_octet),
+            master_ip,
+            gateway,
+            netmask,
+            cidr: 16,
+        })
+    }
+
+    /// Get the base network address
+    pub fn network_addr(&self) -> Ipv4Addr {
+        Ipv4Addr::new(10, self.master_octet, 0, 0)
+    }
+
+    /// Get broadcast address
+    pub fn broadcast_addr(&self) -> Ipv4Addr {
+        Ipv4Addr::new(10, self.master_octet, 255, 255)
+    }
+
+    /// Check if IP belongs to this network
+    pub fn contains(&self, ip: &Ipv4Addr) -> bool {
+        let octets = ip.octets();
+        octets[0] == 10 && octets[1] == self.master_octet
+    }
+}
+
+/// IP Address Management for dynamic allocation
+pub struct IpamPool {
+    config: NetworkConfig,
+    allocated: Arc<Mutex<HashSet<Ipv4Addr>>>,
+    next_third_octet: Arc<Mutex<u8>>,
+    next_fourth_octet: Arc<Mutex<u8>>,
+}
+
+impl IpamPool {
+    pub fn new(config: NetworkConfig) -> Self {
+        let mut allocated = HashSet::new();
+        // Reserve master node IP
+        allocated.insert(config.master_ip);
+        // Reserve network and broadcast addresses
+        allocated.insert(config.network_addr());
+        allocated.insert(config.broadcast_addr());
+
+        Self {
+            config,
+            allocated: Arc::new(Mutex::new(allocated)),
+            next_third_octet: Arc::new(Mutex::new(0)),
+            next_fourth_octet: Arc::new(Mutex::new(2)), // Start from .0.2 (after master at .0.1)
+        }
+    }
+
+    /// Allocate the next available IP address
+    pub fn allocate(&self) -> Result<Ipv4Addr, IpamError> {
+        let mut allocated = self.allocated.lock().unwrap();
+        let mut third = self.next_third_octet.lock().unwrap();
+        let mut fourth = self.next_fourth_octet.lock().unwrap();
+
+        // Try to find next available IP
+        for _ in 0..65534 {  // Max addresses in /16 minus reserved
+            let ip = Ipv4Addr::new(10, self.config.master_octet, *third, *fourth);
+
+            if !allocated.contains(&ip) && !is_reserved_ip(&ip) {
+                allocated.insert(ip);
+                
+                // Increment for next allocation
+                *fourth += 1;
+                if *fourth == 0 {  // Wrapped around
+                    *third += 1;
+                    *fourth = 1;
+                }
+
+                tracing::info!("Allocated IP: {} (pool size: {})", ip, allocated.len());
+                return Ok(ip);
+            }
+
+            // Move to next IP
+            *fourth += 1;
+            if *fourth == 0 {
+                *third += 1;
+                *fourth = 1;
+            }
+        }
+
+        Err(IpamError::PoolExhausted)
+    }
+
+    /// Release an allocated IP back to the pool
+    pub fn release(&self, ip: Ipv4Addr) -> Result<(), IpamError> {
+        let mut allocated = self.allocated.lock().unwrap();
+
+        if !allocated.remove(&ip) {
+            return Err(IpamError::AlreadyAllocated(ip));
+        }
+
+        tracing::info!("Released IP: {} (pool size: {})", ip, allocated.len());
+        Ok(())
+    }
+
+    /// Get count of allocated IPs
+    pub fn allocated_count(&self) -> usize {
+        self.allocated.lock().unwrap().len()
+    }
+
+    /// Check if IP is allocated
+    pub fn is_allocated(&self, ip: &Ipv4Addr) -> bool {
+        self.allocated.lock().unwrap().contains(ip)
+    }
+}
+
+/// Check if octet is in a reserved range
+fn is_reserved_octet(octet: u8) -> bool {
+    matches!(octet,
+        0 |      // 10.0.0.0/8 - often used for private networks
+        255      // Avoid broadcast-like ranges
+    )
+}
+
+/// Check if IP should be reserved (broadcast, network, multicast)
+fn is_reserved_ip(ip: &Ipv4Addr) -> bool {
+    let octets = ip.octets();
+    
+    // Network address (x.x.0.0)
+    if octets[2] == 0 && octets[3] == 0 {
+        return true;
+    }
+    
+    // Broadcast address (x.x.255.255)
+    if octets[2] == 255 && octets[3] == 255 {
+        return true;
+    }
+    
+    // Subnet broadcasts (x.x.x.255)
+    if octets[3] == 255 {
+        return true;
+    }
+    
+    false
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_network_config_creation() {
+        let config = NetworkConfig::new(42).unwrap();
+        assert_eq!(config.master_octet, 42);
+        assert_eq!(config.network, "10.42.0.0/16");
+        assert_eq!(config.master_ip, Ipv4Addr::new(10, 42, 0, 1));
+    }
+
+    #[test]
+    fn test_reserved_octets() {
+        assert!(NetworkConfig::new(0).is_err());
+        assert!(NetworkConfig::new(255).is_err());
+        assert!(NetworkConfig::new(1).is_ok());
+    }
+
+    #[test]
+    fn test_ip_allocation() {
+        let config = NetworkConfig::new(42).unwrap();
+        let pool = IpamPool::new(config);
+
+        let ip1 = pool.allocate().unwrap();
+        let ip2 = pool.allocate().unwrap();
+
+        assert_ne!(ip1, ip2);
+        assert_eq!(ip1, Ipv4Addr::new(10, 42, 0, 2));
+        assert_eq!(ip2, Ipv4Addr::new(10, 42, 0, 3));
+    }
+
+    #[test]
+    fn test_ip_release() {
+        let config = NetworkConfig::new(42).unwrap();
+        let pool = IpamPool::new(config);
+
+        let ip = pool.allocate().unwrap();
+        assert!(pool.is_allocated(&ip));
+
+        pool.release(ip).unwrap();
+        assert!(!pool.is_allocated(&ip));
+    }
+
+    #[test]
+    fn test_network_contains() {
+        let config = NetworkConfig::new(42).unwrap();
+        
+        assert!(config.contains(&Ipv4Addr::new(10, 42, 0, 1)));
+        assert!(config.contains(&Ipv4Addr::new(10, 42, 100, 200)));
+        assert!(!config.contains(&Ipv4Addr::new(10, 43, 0, 1)));
+        assert!(!config.contains(&Ipv4Addr::new(192, 168, 1, 1)));
+    }
+}

oncp/src/token.rs

@@ -0,0 +1,196 @@
+//! OTP (One-Time Password) Token Management for Enrollment
+//!
+//! Generates time-limited cryptographically secure tokens for node enrollment.
+//! Tokens are stored in RAM and automatically expire.
+
+use chrono::{DateTime, Duration, Utc};
+use rand::Rng;
+use std::collections::HashMap;
+use std::sync::{Arc, Mutex};
+use thiserror::Error;
+
+#[allow(dead_code)]
+#[derive(Error, Debug)]
+pub enum TokenError {
+    #[error("token expired")]
+    Expired,
+    #[error("token not found")]
+    NotFound,
+    #[error("invalid token format")]
+    InvalidFormat,
+}
+
+/// OTP Token with expiration
+#[derive(Debug, Clone)]
+struct Token {
+    value: String,
+    created_at: DateTime<Utc>,
+    expires_at: DateTime<Utc>,
+    used: bool,
+}
+
+/// Thread-safe OTP token manager
+pub struct TokenManager {
+    tokens: Arc<Mutex<HashMap<String, Token>>>,
+}
+
+impl TokenManager {
+    pub fn new() -> Self {
+        Self {
+            tokens: Arc::new(Mutex::new(HashMap::new())),
+        }
+    }
+
+    /// Generate a new OTP token with specified expiry duration
+    pub fn generate(&self, expiry_minutes: i64) -> String {
+        let token_value = generate_secure_token(10);
+        let now = Utc::now();
+        let expires_at = now + Duration::minutes(expiry_minutes);
+
+        let token = Token {
+            value: token_value.clone(),
+            created_at: now,
+            expires_at,
+            used: false,
+        };
+
+        let mut tokens = self.tokens.lock().unwrap();
+        tokens.insert(token_value.clone(), token);
+
+        tracing::info!(
+            "Generated OTP token (expires in {} minutes): {}",
+            expiry_minutes,
+            mask_token(&token_value)
+        );
+
+        token_value
+    }
+
+    /// Validate and consume a token
+    pub fn validate(&self, token_value: &str) -> Result<(), TokenError> {
+        let mut tokens = self.tokens.lock().unwrap();
+
+        let token = tokens
+            .get_mut(token_value)
+            .ok_or(TokenError::NotFound)?;
+
+        // Check expiration
+        if Utc::now() > token.expires_at {
+            tokens.remove(token_value);
+            tracing::warn!("Attempted to use expired token: {}", mask_token(token_value));
+            return Err(TokenError::Expired);
+        }
+
+        // Check if already used
+        if token.used {
+            tracing::warn!("Attempted to reuse token: {}", mask_token(token_value));
+            return Err(TokenError::NotFound);
+        }
+
+        // Mark as used and remove immediately
+        token.used = true;
+        tokens.remove(token_value);
+
+        tracing::info!("OTP token validated and consumed: {}", mask_token(token_value));
+        Ok(())
+    }
+
+    /// Cleanup expired tokens (should be called periodically)
+    pub fn cleanup_expired(&self) {
+        let mut tokens = self.tokens.lock().unwrap();
+        let now = Utc::now();
+
+        let expired: Vec<String> = tokens
+            .iter()
+            .filter(|(_, token)| now > token.expires_at)
+            .map(|(key, _)| key.clone())
+            .collect();
+
+        for key in &expired {
+            tokens.remove(key);
+        }
+
+        if !expired.is_empty() {
+            tracing::info!("Cleaned up {} expired tokens", expired.len());
+        }
+    }
+
+    /// Get count of active tokens
+    pub fn active_count(&self) -> usize {
+        let tokens = self.tokens.lock().unwrap();
+        tokens.len()
+    }
+}
+
+impl Default for TokenManager {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+/// Generate cryptographically secure random token
+fn generate_secure_token(length: usize) -> String {
+    const CHARSET: &[u8] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789";
+    let mut rng = rand::thread_rng();
+
+    (0..length)
+        .map(|_| {
+            let idx = rng.gen_range(0..CHARSET.len());
+            CHARSET[idx] as char
+        })
+        .collect()
+}
+
+/// Mask token for logging (show first 2 and last 2 chars)
+fn mask_token(token: &str) -> String {
+    if token.len() <= 4 {
+        return "****".to_string();
+    }
+    format!("{}****{}", &token[..2], &token[token.len() - 2..])
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_token_generation() {
+        let manager = TokenManager::new();
+        let token = manager.generate(3);
+        assert_eq!(token.len(), 10);
+        assert!(token.chars().all(|c| c.is_ascii_alphanumeric()));
+    }
+
+    #[test]
+    fn test_token_validation() {
+        let manager = TokenManager::new();
+        let token = manager.generate(3);
+
+        // First use should succeed
+        assert!(manager.validate(&token).is_ok());
+
+        // Second use should fail (token consumed)
+        assert!(matches!(manager.validate(&token), Err(TokenError::NotFound)));
+    }
+
+    #[test]
+    fn test_token_expiration() {
+        let manager = TokenManager::new();
+        let token = manager.generate(0); // Expires immediately
+
+        std::thread::sleep(std::time::Duration::from_millis(100));
+        assert!(matches!(manager.validate(&token), Err(TokenError::Expired)));
+    }
+
+    #[test]
+    fn test_cleanup() {
+        let manager = TokenManager::new();
+        manager.generate(0);
+        manager.generate(0);
+
+        std::thread::sleep(std::time::Duration::from_millis(100));
+        manager.cleanup_expired();
+
+        assert_eq!(manager.active_count(), 0);
+    }
+}