feat(server): support control plane

Author: ICKelin

README.md

@@ -204,7 +204,7 @@ listen_addr = "0.0.0.0:8080"
 chacha20poly1305 = "your-secret-key-here"
 
 # AES-256-GCM (Hardware accelerated on modern CPUs)
-# aes256gcm = "your-secret-key-here"
+# aes256 = "your-secret-key-here"
 
 # XOR (Lightweight, for testing only)
 # xor = "test-key"

etc/server.toml.example

@@ -1,24 +1,38 @@
 # Example server configuration file for Rustun VPN
 
-[server_config]
-# Address and port to listen on
-# 0.0.0.0 listens on all interfaces
-# Use 127.0.0.1 for local only
-listen_addr = "0.0.0.0:8080"
-
 [crypto_config]
 # Encryption method (choose one):
 # - chacha20poly1305: High security, good performance (recommended)
-# - aes256gcm: High security, hardware acceleration on supported CPUs
+# - aes256: High security, hardware acceleration on supported CPUs
 # - xor: Fast but low security (for testing only)
 # - plain: No encryption (for debugging only)
 
 chacha20poly1305 = "your-secret-key-here"
 
 # Alternative encryption options (uncomment to use):
-# aes256gcm = "your-secret-key-here"
+# aes256 = "your-secret-key-here"
 # xor = "your-secret-key-here"
-# plain = true
+# crypto_config="plain"
+
+[server_config]
+# Address and port to listen on
+# 0.0.0.0 listens on all interfaces
+# Use 127.0.0.1 for local only
+listen_addr = "0.0.0.0:8080"
+
+# Optional: Conf-agent configuration for syncing with control plane
+# If configured, conf-agent will periodically fetch routes and report connection status
+# [conf_agent]
+# # Control plane API URL
+# control_plane_url = "http://localhost:8080"
+# # API token for authentication (optional)
+# api_token = "your-api-token-here"
+# # Routes file path to update (should match route_config.routes_file)
+# routes_file = "/etc/rustun/routes.json"
+# # Poll interval in seconds for fetching routes (default: 60)
+# poll_interval = 60
+# # Connection reporting interval in seconds (default: 30)
+# report_interval = 30
 
 [route_config]
 # Path to the routes configuration file

src/network/connection_manager.rs

@@ -164,6 +164,17 @@ impl ConnectionManager {
 
         None
     }
+
+    pub fn dump_connection_info(&self) -> Vec<ConnectionMeta> {
+        let mut result = Vec::new();
+        let guard = self.cluster_connections.read().unwrap_or_else(|e| e.into_inner());
+        for (_, connections) in guard.iter() {
+            for conn in connections {
+                result.push(conn.clone());
+            }
+        }
+        result
+    }
 }
 
 impl Default for ConnectionManager {

src/network/mod.rs

@@ -134,6 +134,9 @@ impl PartialEq<ConnectionMeta> for &ConnectionMeta {
 }
 
 impl ConnectionMeta {
+    pub fn dump(&self) -> String {
+        format!("{},{},{},{}", self.cluster, self.identity, self.private_ip, self.last_active)
+    }
     /// Check if a destination IP matches this connection's routing rules
     ///
     /// Returns true if the destination matches the private IP or falls

src/server/client_manager.rs

@@ -1,8 +1,8 @@
-use serde::Deserialize;
+use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
 use std::sync::RwLock;
 
-#[derive(Debug, Clone, Deserialize)]
+#[derive(Debug, Clone, Deserialize, Serialize)]
 pub struct ClientConfig {
     pub cluster: String,
     pub identity: String,

src/server/conf_agent.rs

@@ -0,0 +1,234 @@
+use crate::server::client_manager::{ClientConfig, ClientManager};
+use crate::server::config::ConfAgentConfig;
+use crate::network::connection_manager::ConnectionManager;
+use serde::{Deserialize, Serialize};
+use std::sync::Arc;
+use tokio::fs;
+use tokio::time::{interval, Duration};
+
+/// Connection update request for backend API
+#[derive(Serialize, Debug)]
+struct ConnectionUpdateRequest {
+    cluster_id: u64,
+    identity: String,
+    last_active: Option<u64>,
+}
+
+/// Client config response from control plane API
+#[derive(Deserialize, Debug)]
+struct ClientConfigResponse {
+    cluster: String, // Cluster ID as string
+    identity: String,
+    private_ip: String,
+    mask: String,
+    gateway: String,
+    ciders: Vec<String>,
+}
+
+pub struct ConfAgent {
+    config: ConfAgentConfig,
+    client_manager: Arc<ClientManager>,
+    connection_manager: Arc<ConnectionManager>,
+    routes_file: String,
+}
+
+
+impl ConfAgent {
+    pub fn new(
+        config: ConfAgentConfig,
+        client_manager: Arc<ClientManager>,
+        connection_manager: Arc<ConnectionManager>,
+        routes_file: String,
+    ) -> Self {
+        Self {
+            config,
+            client_manager,
+            connection_manager,
+            routes_file,
+        }
+    }
+
+    /// Start the conf-agent service
+    pub async fn start(self: Arc<Self>) -> crate::Result<()> {
+        tracing::info!("Starting conf-agent");
+        tracing::info!("Control plane URL: {}", self.config.control_plane_url);
+        tracing::info!("Routes file: {}", self.config.routes_file);
+        tracing::info!("Poll interval: {}s", self.config.poll_interval);
+
+        // Initial fetch and report
+        if let Err(e) = self.fetch_and_update_routes().await {
+            tracing::error!("Failed to fetch routes: {:?}", e);
+        }
+        if let Err(e) = self.report_connections().await {
+            tracing::error!("Failed to report connections: {:?}", e);
+        }
+
+        // Periodic tasks: route fetching and connection reporting
+        let mut route_ticker = interval(Duration::from_secs(self.config.poll_interval));
+        let mut report_ticker = interval(Duration::from_secs(self.config.report_interval));
+
+        loop {
+            tokio::select! {
+                _ = route_ticker.tick() => {
+                    if let Err(e) = self.fetch_and_update_routes().await {
+                        tracing::error!("Failed to fetch routes: {:?}", e);
+                    }
+                }
+                _ = report_ticker.tick() => {
+                    if let Err(e) = self.report_connections().await {
+                        tracing::error!("Failed to report connections: {:?}", e);
+                    }
+                }
+            }
+        }
+    }
+
+    /// Report connections from connection manager
+    async fn report_connections(&self) -> crate::Result<()> {
+        // Get connections from connection manager
+        let connections = self.connection_manager.dump_connection_info();
+
+        if connections.is_empty() {
+            return Ok(());
+        }
+
+        // Convert ConnectionMeta to ConnectionUpdateRequest
+        let mut updates = Vec::new();
+        for meta in &connections {
+            // Parse cluster ID from string to u64
+            let cluster_id: u64 = match meta.cluster.parse() {
+                Ok(id) => id,
+                Err(_) => {
+                    tracing::warn!(
+                        "Invalid cluster ID '{}' for identity {}, skipping",
+                        meta.cluster,
+                        meta.identity
+                    );
+                    continue;
+                }
+            };
+
+            updates.push(ConnectionUpdateRequest {
+                cluster_id,
+                identity: meta.identity.clone(),
+                last_active: Some(meta.last_active),
+            });
+        }
+
+        if updates.is_empty() {
+            return Ok(());
+        }
+
+        // Send batch update to backend
+        let url = format!("{}/api/sync/connections", self.config.control_plane_url);
+        Self::send_connection_updates(&url, self.config.api_token.as_deref(), &updates).await?;
+
+        tracing::debug!("Reported {} connection updates", updates.len());
+        Ok(())
+    }
+
+    /// Fetch routes from control plane and update local routes file
+    async fn fetch_and_update_routes(&self) -> crate::Result<()> {
+        tracing::debug!("Fetching routes from control plane...");
+
+        let url = format!("{}/api/sync/clients", self.config.control_plane_url);
+        let routes = Self::fetch_routes(&url, self.config.api_token.as_deref()).await?;
+
+        tracing::info!("Fetched {} routes", routes.len());
+
+        // Update client manager
+        // self.client_manager.add_clients_config(routes.clone());
+        self.client_manager.rewrite_clients_config(routes.clone());
+
+        // Write to routes file
+        Self::write_routes(&self.routes_file, &routes).await?;
+
+        tracing::info!("Routes file updated successfully");
+        Ok(())
+    }
+
+    /// Fetch routes from control plane API
+    async fn fetch_routes(
+        url: &str,
+        token: Option<&str>,
+    ) -> crate::Result<Vec<ClientConfig>> {
+        let mut request = ureq::get(url).timeout(Duration::from_secs(30));
+
+        if let Some(token) = token {
+            request = request.set("Authorization", &format!("Bearer {}", token));
+        }
+
+        let response = request.call()?;
+
+        let status = response.status();
+        let body = response.into_string()?;
+        
+        if status != 200 {
+            return Err(format!("Control plane returned error: {} - {}", status, body).into());
+        }
+
+        let routes: Vec<ClientConfigResponse> = serde_json::from_str(&body)?;
+
+        // Convert to ClientConfig format
+        let client_configs: Vec<ClientConfig> = routes
+            .into_iter()
+            .map(|r| ClientConfig {
+                cluster: r.cluster,
+                identity: r.identity,
+                private_ip: r.private_ip,
+                mask: r.mask,
+                gateway: r.gateway,
+                ciders: r.ciders,
+            })
+            .collect();
+
+        Ok(client_configs)
+    }
+
+    /// Write routes to file atomically
+    async fn write_routes(file_path: &str, routes: &[ClientConfig]) -> crate::Result<()> {
+        // Ensure parent directory exists
+        if let Some(parent) = std::path::Path::new(file_path).parent() {
+            fs::create_dir_all(parent).await?;
+        }
+
+        // Serialize to JSON with pretty formatting
+        let json = serde_json::to_string_pretty(routes)?;
+
+        // Write to temp file first, then rename (atomic write)
+        let temp_path = format!("{}.tmp", file_path);
+        fs::write(&temp_path, json).await?;
+        fs::rename(&temp_path, file_path).await?;
+
+        Ok(())
+    }
+
+
+    /// Send connection updates to control plane API
+    async fn send_connection_updates(
+        url: &str,
+        token: Option<&str>,
+        updates: &[ConnectionUpdateRequest],
+    ) -> crate::Result<()> {
+        let json_data = serde_json::to_string(updates)?;
+
+        let mut request = ureq::post(url)
+            .set("Content-Type", "application/json")
+            .timeout(Duration::from_secs(30));
+
+        if let Some(token) = token {
+            request = request.set("Authorization", &format!("Bearer {}", token));
+        }
+
+        let response = request.send_string(&json_data)?;
+        let status = response.status();
+        let body = response.into_string()?;
+
+        if status != 200 {
+            return Err(format!("Backend returned error: {} - {}", status, body).into());
+        }
+
+        Ok(())
+    }
+}
+

src/server/config.rs

@@ -8,13 +8,40 @@ pub struct Config {
     pub server_config: ServerConfig,
     pub crypto_config: CryptoConfig,
     pub route_config: RouteConfig,
+    #[serde(default)]
+    pub conf_agent: Option<ConfAgentConfig>,
 }
 
 #[derive(Debug, Deserialize, Clone)]
 pub struct ServerConfig {
     pub listen_addr: String,
 }
 
+#[derive(Debug, Deserialize, Clone)]
+pub struct ConfAgentConfig {
+    /// Control plane API URL
+    pub control_plane_url: String,
+    /// API token for authentication
+    #[serde(default)]
+    pub api_token: Option<String>,
+    /// Routes file path to update
+    pub routes_file: String,
+    /// Poll interval in seconds for fetching routes
+    #[serde(default = "default_poll_interval")]
+    pub poll_interval: u64,
+    /// Connection reporting interval in seconds (default: 30)
+    #[serde(default = "default_report_interval")]
+    pub report_interval: u64,
+}
+
+fn default_poll_interval() -> u64 {
+    60
+}
+
+fn default_report_interval() -> u64 {
+    30
+}
+
 #[derive(Debug, Deserialize)]
 pub struct RouteConfig {
     pub routes_file: String,