diff --git a/easytier-contrib/easytier-ohrs/src/config.rs b/easytier-contrib/easytier-ohrs/src/config.rs new file mode 100644 index 00000000..cf25a9bb --- /dev/null +++ b/easytier-contrib/easytier-ohrs/src/config.rs @@ -0,0 +1,4 @@ +pub(crate) mod services; +pub(crate) mod repository; +pub(crate) mod storage; +pub(crate) mod types; diff --git a/easytier-contrib/easytier-ohrs/src/config/repository/mod.rs b/easytier-contrib/easytier-ohrs/src/config/repository/mod.rs new file mode 100644 index 00000000..1b66eb24 --- /dev/null +++ b/easytier-contrib/easytier-ohrs/src/config/repository/mod.rs @@ -0,0 +1,13 @@ +#[path = "../../config_repo/field_store.rs"] +mod field_store; +#[path = "../../config_repo/import_export.rs"] +mod import_export; +#[path = "../../config_repo/legacy_migration.rs"] +mod legacy_migration; +#[path = "../../config_repo/validation.rs"] +mod validation; + +#[path = "../../config_repo.rs"] +mod repo; + +pub use repo::*; diff --git a/easytier-contrib/easytier-ohrs/src/config/services/mod.rs b/easytier-contrib/easytier-ohrs/src/config/services/mod.rs new file mode 100644 index 00000000..88b329de --- /dev/null +++ b/easytier-contrib/easytier-ohrs/src/config/services/mod.rs @@ -0,0 +1,2 @@ +pub(crate) mod schema_service; +pub(crate) mod share_link_service; diff --git a/easytier-contrib/easytier-ohrs/src/config/services/schema_service.rs b/easytier-contrib/easytier-ohrs/src/config/services/schema_service.rs new file mode 100644 index 00000000..a6c9ae83 --- /dev/null +++ b/easytier-contrib/easytier-ohrs/src/config/services/schema_service.rs @@ -0,0 +1,394 @@ +use easytier::proto::ALL_DESCRIPTOR_BYTES; +use napi_derive_ohos::napi; +use once_cell::sync::Lazy; +use prost_reflect::{Cardinality, DescriptorPool, FieldDescriptor, Kind, MessageDescriptor}; +use serde::Serialize; + +#[derive(Debug, Clone, Serialize)] +#[napi(object)] +pub struct FieldOption { + pub label: String, + pub value: String, +} + +#[derive(Debug, Clone, Serialize)] +#[napi(object)] +pub struct ValidationRule { + pub rule_type: String, + pub arg: String, + pub message: String, +} + +#[derive(Debug, Clone, Serialize)] +#[napi(object)] +pub struct NetworkConfigSchema { + pub node_kind: String, + pub name: String, + pub field_number: i32, + pub type_name: Option, + pub semantic_type: Option, + pub value_kind: String, + pub is_list: bool, + pub required: bool, + pub default_value_text: Option, + pub enum_options: Vec, + pub validations: Vec, + pub children: Vec, + pub definitions: Vec, +} + +#[derive(Debug, Clone, Serialize)] +#[napi(object)] +pub struct ConfigFieldMapping { + pub field_name: String, + pub field_number: i32, +} + +static DESCRIPTOR_POOL: Lazy = Lazy::new(|| { + DescriptorPool::decode(ALL_DESCRIPTOR_BYTES) + .expect("easytier descriptor pool should decode from embedded protobuf descriptors") +}); + +const NETWORK_CONFIG_MESSAGE_NAME: &str = "api.manage.NetworkConfig"; + +fn descriptor_pool() -> &'static DescriptorPool { + &DESCRIPTOR_POOL +} + +fn network_config_descriptor() -> MessageDescriptor { + descriptor_pool() + .get_message_by_name(NETWORK_CONFIG_MESSAGE_NAME) + .expect("api.manage.NetworkConfig descriptor should exist") +} + +fn field_default_value_text(field: &FieldDescriptor) -> Option { + if field.is_list() || field.is_map() { + return Some("[]".to_string()); + } + + match field.kind() { + Kind::Bool => Some("false".to_string()), + Kind::String => Some("\"\"".to_string()), + Kind::Bytes => Some("\"\"".to_string()), + Kind::Int32 + | Kind::Sint32 + | Kind::Sfixed32 + | Kind::Int64 + | Kind::Sint64 + | Kind::Sfixed64 + | Kind::Uint32 + | Kind::Fixed32 + | Kind::Uint64 + | Kind::Fixed64 + | Kind::Float + | Kind::Double => Some("0".to_string()), + Kind::Enum(enum_desc) => enum_desc.get_value(0).map(|value| value.number().to_string()), + Kind::Message(_) => None, + } +} + +fn field_type_name(field: &FieldDescriptor) -> Option { + match field.kind() { + Kind::Enum(enum_desc) => Some(enum_desc.full_name().to_string()), + Kind::Message(message_desc) => Some(message_desc.full_name().to_string()), + _ => None, + } +} + +fn field_semantic_type(field: &FieldDescriptor) -> Option { + match field.name() { + "virtual_ipv4" => Some("cidr_ip".to_string()), + "network_length" => Some("cidr_mask".to_string()), + "peer_urls" => Some("peer[]".to_string()), + "proxy_cidrs" => Some("cidr[]".to_string()), + "listener_urls" => Some("listener[]".to_string()), + "routes" => Some("route[]".to_string()), + "exit_nodes" => Some("ip[]".to_string()), + "relay_network_whitelist" => Some("network_name[]".to_string()), + "mapped_listeners" => Some("mapped_listener[]".to_string()), + "port_forwards" => Some("port_forward[]".to_string()), + _ => None, + } +} + +fn enum_options(kind: Kind) -> Vec { + match kind { + Kind::Enum(enum_desc) => enum_desc + .values() + .map(|value| FieldOption { + label: value.name().to_string(), + value: value.number().to_string(), + }) + .collect(), + _ => Vec::new(), + } +} + +fn should_expose_field(field: &FieldDescriptor) -> bool { + match field.containing_oneof() { + Some(_) => field.field_descriptor_proto().proto3_optional.unwrap_or(false), + None => true, + } +} + +fn build_validations(field: &FieldDescriptor) -> Vec { + if field.cardinality() == Cardinality::Required { + return vec![ValidationRule { + rule_type: "required".to_string(), + arg: String::new(), + message: format!("{} is required", field.name()), + }]; + } + + Vec::new() +} + +fn kind_to_value_kind(field: &FieldDescriptor) -> String { + if field.is_map() { + return "object".to_string(); + } + + match field.kind() { + Kind::Bool => "boolean".to_string(), + Kind::String | Kind::Bytes => "string".to_string(), + Kind::Int32 + | Kind::Sint32 + | Kind::Sfixed32 + | Kind::Int64 + | Kind::Sint64 + | Kind::Sfixed64 + | Kind::Uint32 + | Kind::Fixed32 + | Kind::Uint64 + | Kind::Fixed64 + | Kind::Float + | Kind::Double => "number".to_string(), + Kind::Enum(_) => "enum".to_string(), + Kind::Message(_) => "object".to_string(), + } +} + +fn build_node( + node_kind: &str, + name: String, + field_number: i32, + type_name: Option, + semantic_type: Option, + value_kind: String, + is_list: bool, + required: bool, + default_value_text: Option, + enum_options: Vec, + validations: Vec, + children: Vec, + definitions: Vec, +) -> NetworkConfigSchema { + NetworkConfigSchema { + node_kind: node_kind.to_string(), + name, + field_number, + type_name, + semantic_type, + value_kind, + is_list, + required, + default_value_text, + enum_options, + validations, + children, + definitions, + } +} + +fn build_map_entry_node(message_desc: &MessageDescriptor) -> NetworkConfigSchema { + let key_field = message_desc.map_entry_key_field(); + let value_field = message_desc.map_entry_value_field(); + + build_node( + "object", + message_desc.name().to_string(), + 0, + Some(message_desc.full_name().to_string()), + None, + "object".to_string(), + false, + true, + None, + Vec::new(), + Vec::new(), + vec![build_schema_field_node(&key_field), build_schema_field_node(&value_field)], + Vec::new(), + ) +} + +fn field_children(field: &FieldDescriptor) -> Vec { + if field.is_map() { + if let Kind::Message(message_desc) = field.kind() { + return vec![build_map_entry_node(&message_desc)]; + } + } + + match field.kind() { + Kind::Message(message_desc) => build_message_children(&message_desc), + _ => Vec::new(), + } +} + +fn build_message_children(message_desc: &MessageDescriptor) -> Vec { + message_desc + .fields() + .filter(should_expose_field) + .map(|field| build_schema_field_node(&field)) + .collect() +} + +fn build_schema_field_node(field: &FieldDescriptor) -> NetworkConfigSchema { + build_node( + "field", + field.name().to_string(), + field.number() as i32, + field_type_name(field), + field_semantic_type(field), + kind_to_value_kind(field), + field.is_list() || field.is_map(), + field.cardinality() == Cardinality::Required, + field_default_value_text(field), + enum_options(field.kind()), + build_validations(field), + field_children(field), + Vec::new(), + ) +} + +fn collect_definitions() -> Vec { + let mut definitions = Vec::new(); + + for message_desc in descriptor_pool().all_messages() { + let full_name = message_desc.full_name(); + if full_name == NETWORK_CONFIG_MESSAGE_NAME || message_desc.is_map_entry() { + continue; + } + + definitions.push(build_node( + "object", + full_name.to_string(), + 0, + Some(full_name.to_string()), + None, + "object".to_string(), + false, + true, + None, + Vec::new(), + Vec::new(), + build_message_children(&message_desc), + Vec::new(), + )); + } + + for enum_desc in descriptor_pool().all_enums() { + definitions.push(build_node( + "enum", + enum_desc.full_name().to_string(), + 0, + Some(enum_desc.full_name().to_string()), + None, + "enum".to_string(), + false, + false, + None, + enum_options(Kind::Enum(enum_desc.clone())), + Vec::new(), + Vec::new(), + Vec::new(), + )); + } + + definitions.sort_by(|a, b| a.name.cmp(&b.name)); + definitions +} + +fn build_network_config_schema() -> NetworkConfigSchema { + let network_config = network_config_descriptor(); + build_node( + "schema", + network_config.name().to_string(), + 0, + Some(network_config.full_name().to_string()), + None, + "object".to_string(), + false, + true, + None, + Vec::new(), + Vec::new(), + build_message_children(&network_config), + collect_definitions(), + ) +} + +fn build_network_config_field_mappings() -> Vec { + network_config_descriptor() + .fields() + .filter(should_expose_field) + .map(|field| ConfigFieldMapping { + field_name: field.name().to_string(), + field_number: field.number() as i32, + }) + .collect() +} + +pub fn get_network_config_schema() -> NetworkConfigSchema { + build_network_config_schema() +} + +pub fn get_network_config_field_mappings() -> Vec { + build_network_config_field_mappings() +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn schema_is_exposed_as_single_tree_type() { + let schema = get_network_config_schema(); + assert_eq!(schema.node_kind, "schema"); + assert_eq!(schema.name, "NetworkConfig"); + assert_eq!(schema.type_name.as_deref(), Some("api.manage.NetworkConfig")); + + let virtual_ipv4 = schema + .children + .iter() + .find(|field| field.name == "virtual_ipv4") + .expect("virtual_ipv4 field"); + assert_eq!(virtual_ipv4.semantic_type.as_deref(), Some("cidr_ip")); + + let secure_mode = schema + .children + .iter() + .find(|field| field.name == "secure_mode") + .expect("secure_mode field"); + assert!(secure_mode.children.iter().any(|field| field.name == "enabled")); + + let secure_mode_definition = schema + .definitions + .iter() + .find(|definition| definition.name == "common.SecureModeConfig") + .expect("secure mode definition"); + assert!(secure_mode_definition + .children + .iter() + .any(|field| field.name == "local_private_key")); + + let networking_method_definition = schema + .definitions + .iter() + .find(|definition| definition.name == "api.manage.NetworkingMethod") + .expect("networking method enum definition"); + assert!(networking_method_definition + .enum_options + .iter() + .any(|option| option.label == "PublicServer")); + } +} diff --git a/easytier-contrib/easytier-ohrs/src/config/services/share_link_service.rs b/easytier-contrib/easytier-ohrs/src/config/services/share_link_service.rs new file mode 100644 index 00000000..bcbbc5b7 --- /dev/null +++ b/easytier-contrib/easytier-ohrs/src/config/services/share_link_service.rs @@ -0,0 +1,187 @@ +use crate::config::repository::{get_config_record, save_config_record}; +use crate::config::services::schema_service::get_network_config_field_mappings; +use crate::config::types::stored_config::SharedConfigLinkPayload; +use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD}; +use easytier::proto::api::manage::NetworkConfig; +use flate2::{Compression, read::ZlibDecoder, write::ZlibEncoder}; +use gethostname::gethostname; +use std::collections::HashMap; +use std::io::{Read, Write}; +use url::Url; +use uuid::Uuid; + +const SHARE_LINK_HOST: &str = "easytier.cn"; +const SHARE_LINK_PATH: &str = "/comp_cfg"; + +fn field_name_to_id_map() -> HashMap { + get_network_config_field_mappings() + .into_iter() + .map(|mapping| (mapping.field_name, mapping.field_number.to_string())) + .collect() +} + +fn field_id_to_name_map() -> HashMap { + get_network_config_field_mappings() + .into_iter() + .map(|mapping| (mapping.field_number.to_string(), mapping.field_name)) + .collect() +} + +fn prune_empty(value: &serde_json::Value) -> Option { + match value { + serde_json::Value::Null => None, + serde_json::Value::Array(values) if values.is_empty() => None, + _ => Some(value.clone()), + } +} + +fn map_config_json(config: &NetworkConfig) -> Result { + let field_name_to_id = field_name_to_id_map(); + let raw = serde_json::to_value(config).map_err(|err| err.to_string())?; + let mut mapped = serde_json::Map::new(); + + for (key, value) in raw.as_object().cloned().unwrap_or_default() { + let Some(value) = prune_empty(&value) else { + continue; + }; + let mapped_key = field_name_to_id.get(&key).cloned().unwrap_or(key); + mapped.insert(mapped_key, value); + } + + serde_json::to_string(&mapped).map_err(|err| err.to_string()) +} + +fn unmap_config_json(raw: &str) -> Result { + let field_id_to_name = field_id_to_name_map(); + let value = serde_json::from_str::(raw).map_err(|err| err.to_string())?; + let mut mapped = serde_json::Map::new(); + for (key, value) in value.as_object().cloned().unwrap_or_default() { + let field_name = field_id_to_name.get(&key).cloned().unwrap_or(key); + mapped.insert(field_name, value); + } + serde_json::from_value(serde_json::Value::Object(mapped)).map_err(|err| err.to_string()) +} + +fn compress_to_base64url(raw: &str) -> Result { + let mut encoder = ZlibEncoder::new(Vec::new(), Compression::best()); + encoder + .write_all(raw.as_bytes()) + .map_err(|err| err.to_string())?; + let compressed = encoder.finish().map_err(|err| err.to_string())?; + Ok(URL_SAFE_NO_PAD.encode(compressed)) +} + +fn decompress_from_base64url(raw: &str) -> Result { + let compressed = URL_SAFE_NO_PAD.decode(raw).map_err(|err| err.to_string())?; + let mut decoder = ZlibDecoder::new(compressed.as_slice()); + let mut out = String::new(); + decoder.read_to_string(&mut out).map_err(|err| err.to_string())?; + Ok(out) +} + +pub fn build_config_share_link( + config_id: &str, + display_name: Option, + only_start: bool, +) -> Option { + let record = get_config_record(config_id)?; + let config = serde_json::from_str::(&record.config_json).ok()?; + let mapped_json = map_config_json(&config).ok()?; + let compressed = compress_to_base64url(&mapped_json).ok()?; + let final_name = display_name.or(Some(record.meta.display_name)).filter(|name| !name.is_empty()); + + let mut url = Url::parse(&format!("https://{SHARE_LINK_HOST}{SHARE_LINK_PATH}")).ok()?; + url.query_pairs_mut().append_pair("cfg", &compressed); + if let Some(name) = final_name { + url.query_pairs_mut().append_pair("name", &name); + } + if only_start { + url.query_pairs_mut().append_pair("only_start", "true"); + } + Some(url.to_string()) +} + +pub fn parse_config_share_link(share_link: &str) -> Option { + let url = Url::parse(share_link).ok()?; + if url.host_str()? != SHARE_LINK_HOST || url.path() != SHARE_LINK_PATH { + return None; + } + + let cfg = url.query_pairs().find(|(key, _)| key == "cfg")?.1.to_string(); + let mapped_json = decompress_from_base64url(&cfg).ok()?; + let mut config = unmap_config_json(&mapped_json).ok()?; + config.instance_id = Some(Uuid::new_v4().to_string()); + let hostname = gethostname().to_string_lossy().to_string(); + if !hostname.is_empty() { + config.hostname = Some(hostname); + } + + let config_json = serde_json::to_string(&config).ok()?; + let display_name = url + .query_pairs() + .find(|(key, _)| key == "name") + .map(|(_, value)| value.to_string()) + .filter(|name| !name.is_empty()); + let only_start = url + .query_pairs() + .find(|(key, _)| key == "only_start") + .map(|(_, value)| value == "true") + .unwrap_or(false); + + Some(SharedConfigLinkPayload { + config_json, + display_name, + only_start, + }) +} + +pub fn import_config_share_link( + share_link: &str, + display_name_override: Option, +) -> Option { + let payload = parse_config_share_link(share_link)?; + let config = serde_json::from_str::(&payload.config_json).ok()?; + let config_id = config.instance_id.clone()?; + let display_name = display_name_override + .filter(|name| !name.is_empty()) + .or(payload.display_name) + .unwrap_or_else(|| config_id.clone()); + + save_config_record(config_id.clone(), display_name, payload.config_json)?; + Some(config_id) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::config_repo::{create_config_record, init_config_store}; + use std::time::{SystemTime, UNIX_EPOCH}; + + fn test_root() -> String { + let unique = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_nanos(); + std::env::temp_dir() + .join(format!("easytier_ohrs_share_test_{unique}")) + .to_string_lossy() + .into_owned() + } + + #[test] + fn share_link_roundtrip_works() { + assert!(init_config_store(test_root())); + create_config_record("cfg-share".to_string(), "share-demo".to_string()).expect("create config"); + + let link = build_config_share_link("cfg-share", None, true).expect("share link"); + let payload = parse_config_share_link(&link).expect("parse link"); + let config = serde_json::from_str::(&payload.config_json).expect("config json"); + + assert!(payload.only_start); + assert_eq!(payload.display_name.as_deref(), Some("share-demo")); + assert_ne!(config.instance_id.as_deref(), Some("cfg-share")); + + let imported_id = import_config_share_link(&link, None).expect("import link"); + assert_ne!(imported_id, "cfg-share"); + } +} diff --git a/easytier-contrib/easytier-ohrs/src/config/storage/config_meta.rs b/easytier-contrib/easytier-ohrs/src/config/storage/config_meta.rs new file mode 100644 index 00000000..4855c1ee --- /dev/null +++ b/easytier-contrib/easytier-ohrs/src/config/storage/config_meta.rs @@ -0,0 +1,322 @@ +use crate::config::types::stored_config::{StoredConfigList, StoredConfigMeta}; +use ohos_hilog_binding::{hilog_debug, hilog_error}; +use rusqlite::{Connection, OptionalExtension, params}; +use std::path::PathBuf; +use std::sync::Mutex; +use std::time::{SystemTime, UNIX_EPOCH}; + +static CONFIG_DB_PATH: Mutex> = Mutex::new(None); +const CONFIG_DB_FILE_NAME: &str = "easytier-config-store.db"; + +#[derive(Debug, Clone)] +struct StoredConfigMetaRecord { + config_id: String, + display_name: String, + created_at: String, + updated_at: String, + favorite: bool, + temporary: bool, +} + +pub(crate) fn now_ts_string() -> String { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_secs().to_string()) + .unwrap_or_else(|_| "0".to_string()) +} + +fn db_file_path() -> Option { + CONFIG_DB_PATH + .lock() + .ok() + .and_then(|guard| guard.as_ref().cloned()) +} + +fn init_schema(conn: &Connection) -> rusqlite::Result<()> { + conn.execute_batch( + "PRAGMA foreign_keys = ON; + CREATE TABLE IF NOT EXISTS stored_configs ( + config_id TEXT PRIMARY KEY, + display_name TEXT NOT NULL, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + favorite INTEGER NOT NULL DEFAULT 0, + temporary INTEGER NOT NULL DEFAULT 0 + ); + CREATE TABLE IF NOT EXISTS stored_config_fields ( + config_id TEXT NOT NULL, + field_name TEXT NOT NULL, + field_json TEXT NOT NULL, + updated_at TEXT NOT NULL, + PRIMARY KEY (config_id, field_name), + FOREIGN KEY (config_id) REFERENCES stored_configs(config_id) ON DELETE CASCADE + ); + CREATE INDEX IF NOT EXISTS idx_stored_config_fields_config_id + ON stored_config_fields(config_id);", + ) +} + +pub(crate) fn open_db() -> Option { + let path = db_file_path()?; + let conn = match Connection::open(&path) { + Ok(conn) => conn, + Err(e) => { + hilog_error!("[Rust] failed to open config db {}: {}", path.display(), e); + return None; + } + }; + + if let Err(e) = init_schema(&conn) { + hilog_error!("[Rust] failed to initialize config db {}: {}", path.display(), e); + return None; + } + + Some(conn) +} + +fn row_to_meta(row: &rusqlite::Row<'_>) -> rusqlite::Result { + Ok(StoredConfigMetaRecord { + config_id: row.get(0)?, + display_name: row.get(1)?, + created_at: row.get(2)?, + updated_at: row.get(3)?, + favorite: row.get::<_, i64>(4)? != 0, + temporary: row.get::<_, i64>(5)? != 0, + }) +} + +fn load_meta_record(conn: &Connection, config_id: &str) -> Option { + conn.query_row( + "SELECT config_id, display_name, created_at, updated_at, favorite, temporary + FROM stored_configs WHERE config_id = ?1", + params![config_id], + row_to_meta, + ) + .optional() + .ok() + .flatten() +} + +fn to_meta(record: StoredConfigMetaRecord) -> StoredConfigMeta { + StoredConfigMeta { + config_id: record.config_id, + display_name: record.display_name, + created_at: record.created_at, + updated_at: record.updated_at, + favorite: record.favorite, + temporary: record.temporary, + } +} + +pub fn init_config_meta_store(root_dir: String) -> bool { + let root = PathBuf::from(root_dir); + if let Err(e) = std::fs::create_dir_all(&root) { + hilog_error!("[Rust] failed to create config db dir {}: {}", root.display(), e); + return false; + } + + let db_path = root.join(CONFIG_DB_FILE_NAME); + match CONFIG_DB_PATH.lock() { + Ok(mut guard) => { + *guard = Some(db_path.clone()); + } + Err(e) => { + hilog_error!("[Rust] failed to lock config db path: {}", e); + return false; + } + } + + if open_db().is_none() { + return false; + } + + hilog_debug!("[Rust] initialized config db at {}", db_path.display()); + true +} + +pub fn list_config_meta_entries() -> StoredConfigList { + let Some(conn) = open_db() else { + return StoredConfigList { configs: vec![] }; + }; + + let mut stmt = match conn.prepare( + "SELECT config_id, display_name, created_at, updated_at, favorite, temporary + FROM stored_configs + ORDER BY updated_at DESC, display_name ASC", + ) { + Ok(stmt) => stmt, + Err(e) => { + hilog_error!("[Rust] failed to prepare list meta query: {}", e); + return StoredConfigList { configs: vec![] }; + } + }; + + let rows = match stmt.query_map([], row_to_meta) { + Ok(rows) => rows, + Err(e) => { + hilog_error!("[Rust] failed to list config meta rows: {}", e); + return StoredConfigList { configs: vec![] }; + } + }; + + let configs = rows.filter_map(Result::ok).map(to_meta).collect(); + StoredConfigList { configs } +} + +pub fn get_config_display_name(config_id: &str) -> Option { + let conn = open_db()?; + load_meta_record(&conn, config_id).map(|record| record.display_name) +} + +pub fn get_config_meta(config_id: &str) -> Option { + let conn = open_db()?; + load_meta_record(&conn, config_id).map(to_meta) +} + +pub fn upsert_config_meta( + config_id: String, + display_name: String, + favorite: bool, + temporary: bool, +) -> StoredConfigMeta { + let now = now_ts_string(); + let Some(conn) = open_db() else { + return StoredConfigMeta { + config_id, + display_name, + created_at: now.clone(), + updated_at: now, + favorite, + temporary, + }; + }; + + let created_at = load_meta_record(&conn, &config_id) + .map(|record| record.created_at) + .unwrap_or_else(|| now.clone()); + + if let Err(e) = conn.execute( + "INSERT INTO stored_configs ( + config_id, display_name, created_at, updated_at, favorite, temporary + ) VALUES (?1, ?2, ?3, ?4, ?5, ?6) + ON CONFLICT(config_id) DO UPDATE SET + display_name = excluded.display_name, + updated_at = excluded.updated_at, + favorite = excluded.favorite, + temporary = excluded.temporary", + params![ + config_id, + display_name, + created_at, + now, + if favorite { 1 } else { 0 }, + if temporary { 1 } else { 0 } + ], + ) { + hilog_error!("[Rust] failed to upsert config meta: {}", e); + } + + get_config_meta(&config_id).unwrap_or(StoredConfigMeta { + config_id, + display_name, + created_at, + updated_at: now, + favorite, + temporary, + }) +} + +pub(crate) fn upsert_config_meta_in_tx( + tx: &rusqlite::Transaction<'_>, + config_id: String, + display_name: String, + favorite: bool, + temporary: bool, +) -> Option { + let now = now_ts_string(); + let created_at = tx + .query_row( + "SELECT config_id, display_name, created_at, updated_at, favorite, temporary + FROM stored_configs WHERE config_id = ?1", + params![config_id], + row_to_meta, + ) + .optional() + .ok() + .flatten() + .map(|record| record.created_at) + .unwrap_or_else(|| now.clone()); + + tx.execute( + "INSERT INTO stored_configs ( + config_id, display_name, created_at, updated_at, favorite, temporary + ) VALUES (?1, ?2, ?3, ?4, ?5, ?6) + ON CONFLICT(config_id) DO UPDATE SET + display_name = excluded.display_name, + updated_at = excluded.updated_at, + favorite = excluded.favorite, + temporary = excluded.temporary", + params![ + config_id, + display_name, + created_at, + now, + if favorite { 1 } else { 0 }, + if temporary { 1 } else { 0 } + ], + ) + .ok()?; + + tx.query_row( + "SELECT config_id, display_name, created_at, updated_at, favorite, temporary + FROM stored_configs WHERE config_id = ?1", + params![config_id], + row_to_meta, + ) + .optional() + .ok() + .flatten() + .map(to_meta) + .or(Some(StoredConfigMeta { + config_id, + display_name, + created_at, + updated_at: now, + favorite, + temporary, + })) +} + +pub fn set_config_display_name(config_id: String, display_name: String) -> Option { + let conn = open_db()?; + let mut record = load_meta_record(&conn, &config_id)?; + record.display_name = display_name; + record.updated_at = now_ts_string(); + + conn.execute( + "UPDATE stored_configs + SET display_name = ?2, updated_at = ?3 + WHERE config_id = ?1", + params![config_id, record.display_name, record.updated_at], + ) + .ok()?; + + Some(to_meta(record)) +} + +pub fn delete_config_meta(config_id: &str) -> bool { + let Some(conn) = open_db() else { + return false; + }; + + match conn.execute( + "DELETE FROM stored_configs WHERE config_id = ?1", + params![config_id], + ) { + Ok(rows) => rows > 0, + Err(e) => { + hilog_error!("[Rust] failed to delete config meta {}: {}", config_id, e); + false + } + } +} diff --git a/easytier-contrib/easytier-ohrs/src/config/storage/mod.rs b/easytier-contrib/easytier-ohrs/src/config/storage/mod.rs new file mode 100644 index 00000000..765a7267 --- /dev/null +++ b/easytier-contrib/easytier-ohrs/src/config/storage/mod.rs @@ -0,0 +1 @@ +pub(crate) mod config_meta; diff --git a/easytier-contrib/easytier-ohrs/src/config/types/mod.rs b/easytier-contrib/easytier-ohrs/src/config/types/mod.rs new file mode 100644 index 00000000..71a56173 --- /dev/null +++ b/easytier-contrib/easytier-ohrs/src/config/types/mod.rs @@ -0,0 +1 @@ +pub(crate) mod stored_config; diff --git a/easytier-contrib/easytier-ohrs/src/config/types/stored_config.rs b/easytier-contrib/easytier-ohrs/src/config/types/stored_config.rs new file mode 100644 index 00000000..86375416 --- /dev/null +++ b/easytier-contrib/easytier-ohrs/src/config/types/stored_config.rs @@ -0,0 +1,68 @@ +use napi_derive_ohos::napi; +use serde::Serialize; + +#[derive(Debug, Clone, Serialize)] +#[serde(rename_all = "camelCase")] +#[napi(object)] +pub struct StoredConfigMeta { + pub config_id: String, + pub display_name: String, + pub created_at: String, + pub updated_at: String, + pub favorite: bool, + pub temporary: bool, +} + +#[derive(Debug, Clone, Serialize)] +#[serde(rename_all = "camelCase")] +#[napi(object)] +pub struct StoredConfigRecord { + pub meta: StoredConfigMeta, + pub config_json: String, +} + +#[derive(Debug, Clone, Serialize)] +#[serde(rename_all = "camelCase")] +#[napi(object)] +pub struct StoredConfigList { + pub configs: Vec, +} + +#[derive(Debug, Clone, Serialize)] +#[serde(rename_all = "camelCase")] +#[napi(object)] +pub struct ExportTomlResult { + pub toml_text: String, +} + +#[derive(Debug, Clone, Serialize)] +#[serde(rename_all = "camelCase")] +#[napi(object)] +pub struct StoredConfigSummary { + pub config_id: String, + pub display_name: String, +} + +#[derive(Debug, Clone, Serialize)] +#[serde(rename_all = "camelCase")] +#[napi(object)] +pub struct SharedConfigLinkPayload { + pub config_json: String, + pub display_name: Option, + pub only_start: bool, +} + +#[derive(Debug, Clone, Serialize)] +#[serde(rename_all = "camelCase")] +#[napi(object)] +pub struct LocalSocketSyncMessage { + pub message_type: String, + pub payload_json: String, +} + +#[derive(Debug, Clone, Serialize)] +#[napi(object)] +pub struct KeyValuePair { + pub key: String, + pub value: String, +} diff --git a/easytier-contrib/easytier-ohrs/src/config_repo/field_store.rs b/easytier-contrib/easytier-ohrs/src/config_repo/field_store.rs new file mode 100644 index 00000000..1d7acf3b --- /dev/null +++ b/easytier-contrib/easytier-ohrs/src/config_repo/field_store.rs @@ -0,0 +1,59 @@ +use crate::config::storage::config_meta::{now_ts_string, open_db}; +use ohos_hilog_binding::hilog_error; +use rusqlite::{Connection, params}; +use serde_json::{Map, Value}; + +pub(super) fn load_config_map_from_db(config_id: &str) -> Option> { + let conn = open_db()?; + let mut stmt = conn + .prepare( + "SELECT field_name, field_json + FROM stored_config_fields + WHERE config_id = ?1", + ) + .ok()?; + let rows = stmt + .query_map(params![config_id], |row| { + let field_name: String = row.get(0)?; + let field_json: String = row.get(1)?; + Ok((field_name, field_json)) + }) + .ok()?; + + let mut object = Map::new(); + for row in rows { + let (field_name, field_json) = row.ok()?; + let value = serde_json::from_str::(&field_json).ok()?; + object.insert(field_name, value); + } + + if object.is_empty() { None } else { Some(object) } +} + +pub(super) fn replace_config_fields( + tx: &Connection, + config_id: &str, + fields: Map, +) -> Option<()> { + if let Err(e) = tx.execute( + "DELETE FROM stored_config_fields WHERE config_id = ?1", + params![config_id], + ) { + hilog_error!("[Rust] failed to clear existing config fields {}: {}", config_id, e); + return None; + } + + for (field_name, value) in fields { + let field_json = serde_json::to_string(&value).ok()?; + if let Err(e) = tx.execute( + "INSERT INTO stored_config_fields (config_id, field_name, field_json, updated_at) + VALUES (?1, ?2, ?3, ?4)", + params![config_id, field_name, field_json, now_ts_string()], + ) { + hilog_error!("[Rust] failed to persist config field {}: {}", config_id, e); + return None; + } + } + + Some(()) +} diff --git a/easytier-contrib/easytier-ohrs/src/config_repo/import_export.rs b/easytier-contrib/easytier-ohrs/src/config_repo/import_export.rs new file mode 100644 index 00000000..2dcd2fdc --- /dev/null +++ b/easytier-contrib/easytier-ohrs/src/config_repo/import_export.rs @@ -0,0 +1,39 @@ +use crate::config::types::stored_config::{ExportTomlResult, StoredConfigRecord}; +use easytier::common::config::{ConfigLoader, TomlConfigLoader}; +use easytier::proto::api::manage::NetworkConfig; + +pub(super) fn export_config_toml_from_record(record: &StoredConfigRecord) -> Option { + let config = serde_json::from_str::(&record.config_json).ok()?; + let toml = config.gen_config().ok()?; + Some(ExportTomlResult { toml_text: toml.dump() }) +} + +pub(super) fn import_toml_to_record( + toml_text: String, + display_name: Option, + save_config_record: impl Fn(String, String, String) -> Option, +) -> Option { + let config = NetworkConfig::new_from_config(TomlConfigLoader::new_from_str(&toml_text).ok()?).ok()?; + + let config_id = config.instance_id.clone()?; + let name_from_toml = toml_text + .lines() + .find_map(|line| { + let trimmed = line.trim(); + if !trimmed.starts_with("instance_name") { + return None; + } + trimmed + .split_once('=') + .map(|(_, value)| value.trim().trim_matches('"').trim_matches('\'').to_string()) + }) + .filter(|name| !name.is_empty()); + + let final_name = display_name + .filter(|name| !name.is_empty()) + .or(name_from_toml) + .unwrap_or_else(|| config_id.clone()); + + let config_json = serde_json::to_string(&config).ok()?; + save_config_record(config_id, final_name, config_json) +} diff --git a/easytier-contrib/easytier-ohrs/src/config_repo/legacy_migration.rs b/easytier-contrib/easytier-ohrs/src/config_repo/legacy_migration.rs new file mode 100644 index 00000000..3c289e8e --- /dev/null +++ b/easytier-contrib/easytier-ohrs/src/config_repo/legacy_migration.rs @@ -0,0 +1,30 @@ +use crate::config::storage::config_meta::get_config_meta; +use ohos_hilog_binding::hilog_error; +use std::path::PathBuf; + +pub(super) fn legacy_config_file_path(root_dir: &Option, config_dir_name: &str, config_id: &str) -> Option { + root_dir.as_ref().map(|root| root.join(config_dir_name).join(format!("{}.json", config_id))) +} + +pub(super) fn migrate_legacy_file_if_needed( + root_dir: &Option, + config_dir_name: &str, + config_id: &str, + save_config_record: impl Fn(String, String, String) -> Option, +) -> Option<()> { + let legacy_path = legacy_config_file_path(root_dir, config_dir_name, config_id)?; + if !legacy_path.exists() { + return Some(()); + } + + let raw = std::fs::read_to_string(&legacy_path).ok()?; + let display_name = get_config_meta(config_id) + .map(|meta| meta.display_name) + .unwrap_or_else(|| config_id.to_string()); + save_config_record(config_id.to_string(), display_name, raw)?; + + if let Err(e) = std::fs::remove_file(&legacy_path) { + hilog_error!("[Rust] failed to remove legacy config file {}: {}", legacy_path.display(), e); + } + Some(()) +} diff --git a/easytier-contrib/easytier-ohrs/src/config_repo/validation.rs b/easytier-contrib/easytier-ohrs/src/config_repo/validation.rs new file mode 100644 index 00000000..cc7551fb --- /dev/null +++ b/easytier-contrib/easytier-ohrs/src/config_repo/validation.rs @@ -0,0 +1,30 @@ +use easytier::proto::api::manage::NetworkConfig; +use serde_json::{Map, Value}; + +pub(super) fn normalize_config_id( + mut config: NetworkConfig, + requested_id: String, +) -> Result { + if requested_id.is_empty() { + return Err("config_id is required".to_string()); + } + config.instance_id = Some(requested_id); + Ok(config) +} + +pub(super) fn validate_config_json( + config_json: &str, + config_id: String, +) -> Result { + let config = serde_json::from_str::(config_json) + .map_err(|e| format!("parse config json failed: {}", e))?; + let config = normalize_config_id(config, config_id)?; + config + .gen_config() + .map_err(|e| format!("generate toml failed: {}", e))?; + Ok(config) +} + +pub(super) fn config_to_top_level_map(config: &NetworkConfig) -> Option> { + serde_json::to_value(config).ok()?.as_object().cloned() +} diff --git a/easytier-contrib/easytier-ohrs/src/exports.rs b/easytier-contrib/easytier-ohrs/src/exports.rs new file mode 100644 index 00000000..57588876 --- /dev/null +++ b/easytier-contrib/easytier-ohrs/src/exports.rs @@ -0,0 +1,2 @@ +pub(crate) mod config_api; +pub(crate) mod runtime_api; diff --git a/easytier-contrib/easytier-ohrs/src/exports/config_api.rs b/easytier-contrib/easytier-ohrs/src/exports/config_api.rs new file mode 100644 index 00000000..02f97a8a --- /dev/null +++ b/easytier-contrib/easytier-ohrs/src/exports/config_api.rs @@ -0,0 +1,45 @@ +use crate::config; + +pub(crate) fn init_config_store(root_dir: String) -> bool { + config::repository::init_config_store(root_dir) +} + +pub(crate) fn list_configs() -> String { + config::repository::list_config_meta_json() +} + +pub(crate) fn save_config(config_id: String, display_name: String, config_json: String) -> bool { + config::repository::save_config_record(config_id, display_name, config_json).is_some() +} + +pub(crate) fn create_config(config_id: String, display_name: String) -> bool { + config::repository::create_config_record(config_id, display_name).is_some() +} + +pub(crate) fn delete_stored_config_meta(config_id: String) -> bool { + config::repository::delete_config_record(&config_id) +} + +pub(crate) fn get_config(config_id: String) -> Option { + config::repository::load_config_json(&config_id) +} + +pub(crate) fn get_default_config() -> Option { + config::repository::get_default_config_json() +} + +pub(crate) fn get_config_field(config_id: String, field: String) -> Option { + config::repository::get_config_field_value(&config_id, &field) +} + +pub(crate) fn set_config_field(config_id: String, field: String, json_value: String) -> bool { + config::repository::set_config_field_value(&config_id, &field, &json_value) +} + +pub(crate) fn import_toml(toml_text: String, display_name: Option) -> Option { + config::repository::import_toml_config(toml_text, display_name).map(|record| record.meta.config_id) +} + +pub(crate) fn export_toml(config_id: String) -> Option { + config::repository::export_config_toml(&config_id).map(|ret| ret.toml_text) +} diff --git a/easytier-contrib/easytier-ohrs/src/exports/runtime_api.rs b/easytier-contrib/easytier-ohrs/src/exports/runtime_api.rs new file mode 100644 index 00000000..afa159ea --- /dev/null +++ b/easytier-contrib/easytier-ohrs/src/exports/runtime_api.rs @@ -0,0 +1,163 @@ +use crate::config::repository::load_config_json; +use crate::kernel_bridge::{aggregate_requested_tun_routes, start_local_socket_server as start_local_socket_server_inner, stop_local_socket_server as stop_local_socket_server_inner}; +use crate::runtime::state::runtime_state::{RuntimeAggregateState, TunAggregateState, clear_tun_attached, mark_tun_attached, runtime_instance_from_running_info}; +use crate::{ASYNC_RUNTIME, EASYTIER_VERSION, INSTANCE_MANAGER, WEB_CLIENTS}; +use crate::config::storage::config_meta::get_config_display_name; +use crate::config::types::stored_config::KeyValuePair; +use easytier::proto::api::manage::NetworkConfig; +use ohos_hilog_binding::{hilog_error, hilog_info}; +use std::sync::Arc; + +pub(crate) fn start_kernel(config_id: String, start_kernel_with_config_id: impl Fn(&str) -> bool) -> bool { + start_kernel_with_config_id(&config_id) +} + +pub(crate) fn stop_kernel( + config_id: String, + stop_web_client: impl Fn(&str) -> bool, + parse_instance_uuid: impl Fn(&str) -> Option, + maybe_stop_local_socket_server: impl Fn(), +) -> bool { + clear_tun_attached(&config_id); + if stop_web_client(&config_id) { + return true; + } + + let Some(instance_id) = parse_instance_uuid(&config_id) else { + return false; + }; + + let ret = INSTANCE_MANAGER + .delete_network_instance(vec![instance_id]) + .map(|_| true) + .unwrap_or_else(|err| { + hilog_error!("[Rust] stop_kernel failed {}: {}", config_id, err); + false + }); + maybe_stop_local_socket_server(); + ret +} + +pub(crate) fn stop_network_instance( + config_ids: Vec, + stop_kernel: impl Fn(String) -> bool, +) -> bool { + let mut ok = true; + for config_id in config_ids { + ok = stop_kernel(config_id) && ok; + } + ok +} + +pub(crate) fn collect_network_infos() -> Vec { + let infos = match INSTANCE_MANAGER.collect_network_infos_sync() { + Ok(infos) => infos, + Err(err) => { + hilog_error!("[Rust] collect network infos failed {}", err); + return vec![]; + } + }; + + infos.into_iter() + .filter_map(|(key, value)| { + serde_json::to_string(&value).ok().map(|value_json| KeyValuePair { + key: key.to_string(), + value: value_json, + }) + }) + .collect() +} + +pub(crate) fn set_tun_fd( + config_id: String, + fd: i32, + parse_instance_uuid: impl Fn(&str) -> Option, +) -> bool { + let Some(instance_id) = parse_instance_uuid(&config_id) else { + hilog_error!("[Rust] set_tun_fd invalid instance id: {}", config_id); + return false; + }; + + INSTANCE_MANAGER + .set_tun_fd(&instance_id, fd) + .map(|_| { + mark_tun_attached(&config_id); + hilog_info!("[Rust] set_tun_fd success instance={} fd={} marked_attached=true", config_id, fd); + true + }) + .unwrap_or_else(|err| { + hilog_error!("[Rust] set_tun_fd failed {}: {}", config_id, err); + false + }) +} + +pub(crate) fn get_runtime_snapshot() -> RuntimeAggregateState { + get_runtime_snapshot_inner() +} + +pub(crate) fn get_runtime_snapshot_inner() -> RuntimeAggregateState { + let infos = match INSTANCE_MANAGER.collect_network_infos_sync() { + Ok(infos) => infos, + Err(err) => { + hilog_error!("[Rust] collect network infos failed {}", err); + return RuntimeAggregateState { + instances: vec![], + tun: TunAggregateState { + active: false, + attached_instance_ids: vec![], + aggregated_routes: vec![], + dns_servers: vec![], + need_rebuild: false, + }, + running_instance_count: 0, + }; + } + }; + + let mut instances = Vec::with_capacity(infos.len()); + for (instance_uuid, info) in infos { + let config_id = instance_uuid.to_string(); + let display_name = get_config_display_name(&config_id).unwrap_or_else(|| config_id.clone()); + let config_json = load_config_json(&config_id); + let stored_config = config_json + .as_deref() + .and_then(|raw| serde_json::from_str::(raw).ok()); + let magic_dns_enabled = stored_config + .as_ref() + .and_then(|cfg| cfg.enable_magic_dns) + .unwrap_or(false); + let need_exit_node = stored_config + .as_ref() + .map(|cfg| !cfg.exit_nodes.is_empty()) + .unwrap_or(false); + instances.push(runtime_instance_from_running_info( + config_id, + display_name, + magic_dns_enabled, + need_exit_node, + info, + )); + } + + instances.sort_by(|a, b| a.display_name.cmp(&b.display_name).then_with(|| a.instance_id.cmp(&b.instance_id))); + let attached_instance_ids = instances + .iter() + .filter(|instance| instance.tun_required) + .map(|instance| instance.instance_id.clone()) + .collect::>(); + let aggregated_routes = aggregate_requested_tun_routes(&instances); + let running_instance_count = instances.iter().filter(|instance| instance.running).count() as i32; + let tun_active = !attached_instance_ids.is_empty(); + + RuntimeAggregateState { + instances, + tun: TunAggregateState { + active: tun_active, + attached_instance_ids, + aggregated_routes, + dns_servers: vec![], + need_rebuild: false, + }, + running_instance_count, + } +} diff --git a/easytier-contrib/easytier-ohrs/src/kernel_bridge.rs b/easytier-contrib/easytier-ohrs/src/kernel_bridge.rs new file mode 100644 index 00000000..f01868b0 --- /dev/null +++ b/easytier-contrib/easytier-ohrs/src/kernel_bridge.rs @@ -0,0 +1,6 @@ +mod protocol; +mod routing; +mod socket_server; + +pub(crate) use routing::aggregate_requested_tun_routes; +pub use socket_server::{start_local_socket_server, stop_local_socket_server}; diff --git a/easytier-contrib/easytier-ohrs/src/kernel_bridge/protocol.rs b/easytier-contrib/easytier-ohrs/src/kernel_bridge/protocol.rs new file mode 100644 index 00000000..13cc6aed --- /dev/null +++ b/easytier-contrib/easytier-ohrs/src/kernel_bridge/protocol.rs @@ -0,0 +1,50 @@ +use crate::config::types::stored_config::LocalSocketSyncMessage; +use serde::Serialize; +use std::io::{Error, ErrorKind, Write}; +use std::os::unix::net::UnixStream; + +#[derive(Debug, Clone, Serialize)] +#[serde(rename_all = "camelCase")] +pub(crate) struct TunRequestPayload { + pub config_id: String, + pub instance_id: String, + pub display_name: String, + pub virtual_ipv4: Option, + pub virtual_ipv4_cidr: Option, + pub aggregated_routes: Vec, + pub magic_dns_enabled: bool, + pub need_exit_node: bool, +} + +pub(crate) fn send_local_socket_message( + stream: &mut UnixStream, + message_type: &str, + payload_json: String, +) -> std::io::Result<()> { + let message = LocalSocketSyncMessage { + message_type: message_type.to_string(), + payload_json, + }; + let mut raw = serde_json::to_vec(&message) + .map_err(|err| Error::new(ErrorKind::InvalidData, err.to_string()))?; + raw.push(b'\n'); + stream.write_all(&raw)?; + Ok(()) +} + +pub(crate) fn broadcast_local_socket_message( + clients: &mut Vec, + message_type: &str, + payload_json: &str, +) -> bool { + let mut active_clients = Vec::with_capacity(clients.len()); + let mut delivered = false; + for mut client in clients.drain(..) { + if send_local_socket_message(&mut client, message_type, payload_json.to_string()).is_ok() { + delivered = true; + active_clients.push(client); + } + } + *clients = active_clients; + delivered +} diff --git a/easytier-contrib/easytier-ohrs/src/kernel_bridge/routing.rs b/easytier-contrib/easytier-ohrs/src/kernel_bridge/routing.rs new file mode 100644 index 00000000..c7638197 --- /dev/null +++ b/easytier-contrib/easytier-ohrs/src/kernel_bridge/routing.rs @@ -0,0 +1,87 @@ +use crate::config::repository::load_config_json; +use crate::runtime::state::runtime_state::RuntimeInstanceState; +use easytier::proto::api::manage::NetworkConfig; +use ipnet::IpNet; +use std::collections::HashSet; + +pub(crate) fn load_manual_routes(config_id: &str) -> Vec { + load_config_json(config_id) + .and_then(|raw| serde_json::from_str::(&raw).ok()) + .map(|config| config.routes) + .unwrap_or_default() +} + +fn normalize_route_cidr(route: &str) -> Option { + route.parse::().ok().map(|network| match network { + IpNet::V4(net) => net.trunc().to_string(), + IpNet::V6(net) => net.trunc().to_string(), + }) +} + +fn simplify_routes(routes: Vec) -> Vec { + let mut parsed = routes + .into_iter() + .filter_map(|route| normalize_route_cidr(&route)) + .filter_map(|route| route.parse::().ok()) + .collect::>(); + parsed.sort_by(|left, right| { + left.prefix_len() + .cmp(&right.prefix_len()) + .then_with(|| left.network().to_string().cmp(&right.network().to_string())) + }); + + let mut simplified = Vec::::new(); + 'outer: for route in parsed { + for existing in &simplified { + if existing.contains(&route.network()) && existing.prefix_len() <= route.prefix_len() { + continue 'outer; + } + } + simplified.retain(|existing| { + !(route.contains(&existing.network()) && route.prefix_len() <= existing.prefix_len()) + }); + simplified.push(route); + } + + let mut seen = HashSet::new(); + simplified + .into_iter() + .map(|route| route.to_string()) + .filter(|route| seen.insert(route.clone())) + .collect() +} + +pub(crate) fn aggregate_tun_routes(instance: &RuntimeInstanceState) -> Vec { + let virtual_ipv4_cidr = instance + .my_node_info + .as_ref() + .and_then(|info| info.virtual_ipv4_cidr.clone()); + let manual_routes = load_manual_routes(&instance.config_id); + let proxy_cidrs = instance + .routes + .iter() + .flat_map(|route| route.proxy_cidrs.iter().cloned()) + .collect::>(); + let mut raw_routes = Vec::new(); + + if let Some(cidr) = virtual_ipv4_cidr.clone() { + raw_routes.push(cidr); + } + + raw_routes.extend(manual_routes.iter().cloned()); + raw_routes.extend(proxy_cidrs.iter().cloned()); + simplify_routes(raw_routes) +} + +pub(crate) fn aggregate_requested_tun_routes(instances: &[RuntimeInstanceState]) -> Vec { + let mut aggregated_routes = Vec::new(); + let mut seen_routes = HashSet::new(); + for instance in instances.iter().filter(|instance| instance.tun_required) { + for route in aggregate_tun_routes(instance) { + if seen_routes.insert(route.clone()) { + aggregated_routes.push(route); + } + } + } + aggregated_routes +} diff --git a/easytier-contrib/easytier-ohrs/src/kernel_bridge/socket_server.rs b/easytier-contrib/easytier-ohrs/src/kernel_bridge/socket_server.rs new file mode 100644 index 00000000..064ff531 --- /dev/null +++ b/easytier-contrib/easytier-ohrs/src/kernel_bridge/socket_server.rs @@ -0,0 +1,187 @@ +use super::protocol::{TunRequestPayload, broadcast_local_socket_message}; +use super::routing::aggregate_tun_routes; +use crate::config::repository::kernel_socket_path; +use crate::get_runtime_snapshot_inner; +use ohos_hilog_binding::hilog_error; +use once_cell::sync::Lazy; +use std::collections::{HashMap, HashSet}; +use std::io::ErrorKind; +use std::os::unix::net::{UnixListener, UnixStream}; +use std::path::PathBuf; +use std::sync::Mutex; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::thread::{self, JoinHandle}; +use std::time::Duration; + +struct LocalSocketState { + stop_flag: std::sync::Arc, + socket_path: PathBuf, + worker: JoinHandle<()>, +} + +static LOCAL_SOCKET_STATE: Lazy>> = Lazy::new(|| Mutex::new(None)); + +pub fn start_local_socket_server() -> bool { + let socket_path = match kernel_socket_path() { + Some(path) => path, + None => { + hilog_error!("[Rust] kernel socket path unavailable"); + return false; + } + }; + + match LOCAL_SOCKET_STATE.lock() { + Ok(guard) if guard.is_some() => return true, + Ok(_) => {} + Err(err) => { + hilog_error!("[Rust] lock localsocket state failed: {}", err); + return false; + } + } + + if socket_path.exists() { + let _ = std::fs::remove_file(&socket_path); + } + + let listener = match UnixListener::bind(&socket_path) { + Ok(listener) => listener, + Err(err) => { + hilog_error!("[Rust] bind localsocket failed {}: {}", socket_path.display(), err); + return false; + } + }; + if let Err(err) = listener.set_nonblocking(true) { + hilog_error!("[Rust] set localsocket nonblocking failed: {}", err); + let _ = std::fs::remove_file(&socket_path); + return false; + } + + let stop_flag = std::sync::Arc::new(AtomicBool::new(false)); + let worker_stop_flag = stop_flag.clone(); + let worker = thread::spawn(move || { + let mut last_snapshot_json = String::new(); + let mut delivered_tun_requests = HashSet::new(); + let mut last_tun_route_signatures = HashMap::::new(); + let mut clients = Vec::::new(); + + while !worker_stop_flag.load(Ordering::Relaxed) { + let mut accepted_client = false; + loop { + match listener.accept() { + Ok((stream, _addr)) => { + accepted_client = true; + clients.push(stream); + } + Err(err) if err.kind() == ErrorKind::WouldBlock => break, + Err(err) => { + hilog_error!("[Rust] accept localsocket failed: {}", err); + break; + } + } + } + + let snapshot = get_runtime_snapshot_inner(); + let snapshot_json = match serde_json::to_string(&snapshot) { + Ok(json) => json, + Err(err) => { + hilog_error!("[Rust] serialize runtime snapshot failed: {}", err); + thread::sleep(Duration::from_millis(250)); + continue; + } + }; + + if accepted_client || snapshot_json != last_snapshot_json { + let _ = broadcast_local_socket_message(&mut clients, "runtime_snapshot", &snapshot_json); + last_snapshot_json = snapshot_json; + } + + for instance in snapshot.instances.iter() { + if instance.running && instance.tun_required { + let virtual_ipv4 = instance + .my_node_info + .as_ref() + .and_then(|info| info.virtual_ipv4.clone()); + let virtual_ipv4_cidr = instance + .my_node_info + .as_ref() + .and_then(|info| info.virtual_ipv4_cidr.clone()); + if clients.is_empty() { + continue; + } + if virtual_ipv4.is_none() || virtual_ipv4_cidr.is_none() { + continue; + } + let aggregated_routes = aggregate_tun_routes(instance); + let route_signature = serde_json::to_string(&aggregated_routes) + .unwrap_or_else(|_| "[]".to_string()); + let should_send = !delivered_tun_requests.contains(&instance.instance_id) + || last_tun_route_signatures + .get(&instance.instance_id) + .map(|value| value != &route_signature) + .unwrap_or(true); + if !should_send { + continue; + } + let payload = TunRequestPayload { + config_id: instance.config_id.clone(), + instance_id: instance.instance_id.clone(), + display_name: instance.display_name.clone(), + virtual_ipv4, + virtual_ipv4_cidr, + aggregated_routes, + magic_dns_enabled: instance.magic_dns_enabled, + need_exit_node: instance.need_exit_node, + }; + let payload_json = match serde_json::to_string(&payload) { + Ok(json) => json, + Err(err) => { + hilog_error!("[Rust] serialize tun request failed: {}", err); + continue; + } + }; + if broadcast_local_socket_message(&mut clients, "tun_request", &payload_json) { + delivered_tun_requests.insert(instance.instance_id.clone()); + last_tun_route_signatures.insert(instance.instance_id.clone(), route_signature); + } + } else { + delivered_tun_requests.remove(&instance.instance_id); + last_tun_route_signatures.remove(&instance.instance_id); + } + } + + thread::sleep(Duration::from_millis(250)); + } + }); + + match LOCAL_SOCKET_STATE.lock() { + Ok(mut guard) => { + *guard = Some(LocalSocketState { + stop_flag, + socket_path, + worker, + }); + true + } + Err(err) => { + hilog_error!("[Rust] lock localsocket state failed: {}", err); + false + } + } +} + +pub fn stop_local_socket_server() -> bool { + let state = match LOCAL_SOCKET_STATE.lock() { + Ok(mut guard) => guard.take(), + Err(err) => { + hilog_error!("[Rust] lock localsocket state failed: {}", err); + return false; + } + }; + + if let Some(state) = state { + state.stop_flag.store(true, Ordering::Relaxed); + let _ = state.worker.join(); + let _ = std::fs::remove_file(state.socket_path); + } + true +} diff --git a/easytier-contrib/easytier-ohrs/src/platform.rs b/easytier-contrib/easytier-ohrs/src/platform.rs new file mode 100644 index 00000000..6a79dd07 --- /dev/null +++ b/easytier-contrib/easytier-ohrs/src/platform.rs @@ -0,0 +1 @@ +pub(crate) mod logging; diff --git a/easytier-contrib/easytier-ohrs/src/platform/logging/mod.rs b/easytier-contrib/easytier-ohrs/src/platform/logging/mod.rs new file mode 100644 index 00000000..0b44f5b3 --- /dev/null +++ b/easytier-contrib/easytier-ohrs/src/platform/logging/mod.rs @@ -0,0 +1 @@ +pub(crate) mod native_log; diff --git a/easytier-contrib/easytier-ohrs/src/platform/logging/native_log.rs b/easytier-contrib/easytier-ohrs/src/platform/logging/native_log.rs new file mode 100644 index 00000000..abfbc611 --- /dev/null +++ b/easytier-contrib/easytier-ohrs/src/platform/logging/native_log.rs @@ -0,0 +1,96 @@ +use napi_derive_ohos::napi; +use ohos_hilog_binding::{ + LogOptions, hilog_debug, hilog_error, hilog_info, hilog_warn, set_global_options, +}; +use std::collections::HashMap; +use std::panic; +use tracing::{Event, Subscriber}; +use tracing_core::Level; +use tracing_subscriber::layer::{Context, Layer}; +use tracing_subscriber::prelude::*; + +static INITIALIZED: std::sync::Once = std::sync::Once::new(); +fn panic_hook(info: &panic::PanicHookInfo) { + hilog_error!("RUST PANIC: {}", info); +} + +#[napi] +pub fn init_panic_hook() { + INITIALIZED.call_once(|| { + panic::set_hook(Box::new(panic_hook)); + }); +} + +#[napi] +pub fn hilog_global_options(domain: u32, tag: String) { + ohos_hilog_binding::forward_stdio_to_hilog(); + set_global_options(LogOptions { + domain, + tag: Box::leak(tag.clone().into_boxed_str()), + }) +} + +#[napi] +pub fn init_tracing_subscriber() { + tracing_subscriber::registry() + .with(CallbackLayer { + callback: Box::new(tracing_callback), + }) + .init(); +} + +fn tracing_callback(event: &Event, fields: HashMap) { + let metadata = event.metadata(); + #[cfg(target_env = "ohos")] + { + let loc = metadata.target().split("::").last().unwrap(); + match *metadata.level() { + Level::TRACE => { + hilog_debug!("[{}] {:?}", loc, fields.values().collect::>()); + } + Level::DEBUG => { + hilog_debug!("[{}] {:?}", loc, fields.values().collect::>()); + } + Level::INFO => { + hilog_info!("[{}] {:?}", loc, fields.values().collect::>()); + } + Level::WARN => { + hilog_warn!("[{}] {:?}", loc, fields.values().collect::>()); + } + Level::ERROR => { + hilog_error!("[{}] {:?}", loc, fields.values().collect::>()); + } + } + } +} + +struct CallbackLayer { + callback: Box) + Send + Sync>, +} + +impl Layer for CallbackLayer { + fn on_event(&self, event: &Event, _ctx: Context) { + // 使用 fmt::format::FmtSpan 提取字段值 + let mut fields = HashMap::new(); + let mut visitor = FieldCollector(&mut fields); + event.record(&mut visitor); + (self.callback)(event, fields); + } +} + +struct FieldCollector<'a>(&'a mut HashMap); + +impl<'a> tracing::field::Visit for FieldCollector<'a> { + fn record_i64(&mut self, field: &tracing::field::Field, value: i64) { + self.0.insert(field.name().to_string(), value.to_string()); + } + + fn record_str(&mut self, field: &tracing::field::Field, value: &str) { + self.0.insert(field.name().to_string(), value.to_string()); + } + + fn record_debug(&mut self, field: &tracing::field::Field, value: &dyn std::fmt::Debug) { + self.0 + .insert(field.name().to_string(), format!("{:?}", value)); + } +} diff --git a/easytier-contrib/easytier-ohrs/src/runtime.rs b/easytier-contrib/easytier-ohrs/src/runtime.rs new file mode 100644 index 00000000..33e14d22 --- /dev/null +++ b/easytier-contrib/easytier-ohrs/src/runtime.rs @@ -0,0 +1 @@ +pub(crate) mod state; diff --git a/easytier-contrib/easytier-ohrs/src/runtime/state/mod.rs b/easytier-contrib/easytier-ohrs/src/runtime/state/mod.rs new file mode 100644 index 00000000..f84ecc4a --- /dev/null +++ b/easytier-contrib/easytier-ohrs/src/runtime/state/mod.rs @@ -0,0 +1 @@ +pub(crate) mod runtime_state; diff --git a/easytier-contrib/easytier-ohrs/src/runtime/state/runtime_state.rs b/easytier-contrib/easytier-ohrs/src/runtime/state/runtime_state.rs new file mode 100644 index 00000000..5cce26b9 --- /dev/null +++ b/easytier-contrib/easytier-ohrs/src/runtime/state/runtime_state.rs @@ -0,0 +1,283 @@ +use easytier::proto::{api, common}; +use napi_derive_ohos::napi; +use serde::Serialize; +use std::collections::HashSet; +use std::sync::Mutex; + +static ATTACHED_TUN_INSTANCE_IDS: once_cell::sync::Lazy>> = + once_cell::sync::Lazy::new(|| Mutex::new(HashSet::new())); + +pub fn mark_tun_attached(instance_id: &str) { + if let Ok(mut guard) = ATTACHED_TUN_INSTANCE_IDS.lock() { + guard.insert(instance_id.to_string()); + } +} + +pub fn clear_tun_attached(instance_id: &str) { + if let Ok(mut guard) = ATTACHED_TUN_INSTANCE_IDS.lock() { + guard.remove(instance_id); + } +} + +pub fn is_tun_attached(instance_id: &str) -> bool { + ATTACHED_TUN_INSTANCE_IDS + .lock() + .map(|guard| guard.contains(instance_id)) + .unwrap_or(false) +} + +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +#[napi(object)] +pub struct PeerConnStats { + pub rx_bytes: i64, + pub tx_bytes: i64, + pub rx_packets: i64, + pub tx_packets: i64, + pub latency_us: i64, +} + +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +#[napi(object)] +pub struct PeerConnInfo { + pub conn_id: String, + pub my_peer_id: i64, + pub peer_id: i64, + pub features: Vec, + pub tunnel_type: Option, + pub local_addr: Option, + pub remote_addr: Option, + pub resolved_remote_addr: Option, + pub stats: Option, + pub loss_rate: Option, + pub is_client: bool, + pub network_name: Option, + pub is_closed: bool, + pub secure_auth_level: Option, + pub peer_identity_type: Option, +} + +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +#[napi(object)] +pub struct PeerInfo { + pub peer_id: i64, + pub default_conn_id: Option, + pub directly_connected_conns: Vec, + pub conns: Vec, +} + +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +#[napi(object)] +pub struct RouteView { + pub peer_id: i64, + pub hostname: Option, + pub ipv4: Option, + pub ipv4_cidr: Option, + pub ipv6_cidr: Option, + pub proxy_cidrs: Vec, + pub next_hop_peer_id: Option, + pub cost: Option, + pub path_latency: Option, + pub udp_nat_type: Option, + pub tcp_nat_type: Option, + pub inst_id: Option, + pub version: Option, + pub is_public_server: Option, +} + +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +#[napi(object)] +pub struct MyNodeInfo { + pub virtual_ipv4: Option, + pub virtual_ipv4_cidr: Option, + pub hostname: Option, + pub version: Option, + pub peer_id: Option, + pub listeners: Vec, + pub vpn_portal_cfg: Option, + pub udp_nat_type: Option, + pub tcp_nat_type: Option, +} + +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +#[napi(object)] +pub struct RuntimeInstanceState { + pub config_id: String, + pub instance_id: String, + pub display_name: String, + pub running: bool, + pub tun_required: bool, + pub tun_attached: bool, + pub magic_dns_enabled: bool, + pub need_exit_node: bool, + pub error_message: Option, + pub my_node_info: Option, + pub events: Vec, + pub routes: Vec, + pub peers: Vec, +} + +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +#[napi(object)] +pub struct TunAggregateState { + pub active: bool, + pub attached_instance_ids: Vec, + pub aggregated_routes: Vec, + pub dns_servers: Vec, + pub need_rebuild: bool, +} + +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +#[napi(object)] +pub struct RuntimeAggregateState { + pub instances: Vec, + pub tun: TunAggregateState, + pub running_instance_count: i32, +} + +fn stringify_ipv4_inet(value: Option) -> Option { + value.map(|v| v.to_string()) +} + +fn stringify_ipv6_inet(value: Option) -> Option { + value.map(|v| v.to_string()) +} + +fn stringify_url(value: Option) -> Option { + value.map(|v| v.to_string()) +} + +fn stringify_uuid(value: Option) -> Option { + value.map(|v| v.to_string()) +} + +fn optional_u32_to_i64(value: Option) -> Option { + value.map(|v| v as i64) +} + +fn optional_i32_to_i64(value: Option) -> Option { + value.map(|v| v as i64) +} + +fn route_to_view(route: api::instance::Route) -> RouteView { + let stun = route.stun_info; + let feature_flag = route.feature_flag; + RouteView { + peer_id: route.peer_id as i64, + hostname: (!route.hostname.is_empty()).then_some(route.hostname), + ipv4: route + .ipv4_addr + .as_ref() + .and_then(|inet| inet.address.as_ref()) + .map(|addr| addr.to_string()), + ipv4_cidr: stringify_ipv4_inet(route.ipv4_addr), + ipv6_cidr: stringify_ipv6_inet(route.ipv6_addr), + proxy_cidrs: route.proxy_cidrs, + next_hop_peer_id: optional_u32_to_i64(route.next_hop_peer_id_latency_first) + .or_else(|| Some(route.next_hop_peer_id as i64)), + cost: Some(route.cost), + path_latency: optional_i32_to_i64(route.path_latency_latency_first) + .or_else(|| Some(route.path_latency as i64)), + udp_nat_type: stun.as_ref().map(|info| info.udp_nat_type), + tcp_nat_type: stun.as_ref().map(|info| info.tcp_nat_type), + inst_id: (!route.inst_id.is_empty()).then_some(route.inst_id), + version: (!route.version.is_empty()).then_some(route.version), + is_public_server: feature_flag.map(|flag| flag.is_public_server), + } +} + +fn peer_conn_to_view(conn: api::instance::PeerConnInfo) -> PeerConnInfo { + let stats = conn.stats.map(|stats| PeerConnStats { + rx_bytes: stats.rx_bytes as i64, + tx_bytes: stats.tx_bytes as i64, + rx_packets: stats.rx_packets as i64, + tx_packets: stats.tx_packets as i64, + latency_us: stats.latency_us as i64, + }); + + PeerConnInfo { + conn_id: conn.conn_id, + my_peer_id: conn.my_peer_id as i64, + peer_id: conn.peer_id as i64, + features: conn.features, + tunnel_type: conn.tunnel.as_ref().map(|t| t.tunnel_type.clone()), + local_addr: conn.tunnel.as_ref().and_then(|t| stringify_url(t.local_addr.clone())), + remote_addr: conn.tunnel.as_ref().and_then(|t| stringify_url(t.remote_addr.clone())), + resolved_remote_addr: conn + .tunnel + .as_ref() + .and_then(|t| stringify_url(t.resolved_remote_addr.clone())), + stats, + loss_rate: Some(conn.loss_rate as f64), + is_client: conn.is_client, + network_name: (!conn.network_name.is_empty()).then_some(conn.network_name), + is_closed: conn.is_closed, + secure_auth_level: Some(conn.secure_auth_level), + peer_identity_type: Some(conn.peer_identity_type), + } +} + +fn peer_to_view(peer: api::instance::PeerInfo) -> PeerInfo { + PeerInfo { + peer_id: peer.peer_id as i64, + default_conn_id: stringify_uuid(peer.default_conn_id), + directly_connected_conns: peer + .directly_connected_conns + .into_iter() + .map(|id| id.to_string()) + .collect(), + conns: peer.conns.into_iter().map(peer_conn_to_view).collect(), + } +} + +fn my_node_info_to_view(info: api::manage::MyNodeInfo) -> MyNodeInfo { + MyNodeInfo { + virtual_ipv4: info + .virtual_ipv4 + .as_ref() + .and_then(|inet| inet.address.as_ref()) + .map(|addr| addr.to_string()), + virtual_ipv4_cidr: stringify_ipv4_inet(info.virtual_ipv4), + hostname: (!info.hostname.is_empty()).then_some(info.hostname), + version: (!info.version.is_empty()).then_some(info.version), + peer_id: Some(info.peer_id as i64), + listeners: info.listeners.into_iter().map(|url| url.to_string()).collect(), + vpn_portal_cfg: info.vpn_portal_cfg, + udp_nat_type: info.stun_info.as_ref().map(|stun| stun.udp_nat_type), + tcp_nat_type: info.stun_info.as_ref().map(|stun| stun.tcp_nat_type), + } +} + +pub fn runtime_instance_from_running_info( + config_id: String, + display_name: String, + magic_dns_enabled: bool, + need_exit_node: bool, + info: api::manage::NetworkInstanceRunningInfo, +) -> RuntimeInstanceState { + let tun_attached = info.running && is_tun_attached(&config_id); + let tun_required = info.running && (info.dev_name != "no_tun" || tun_attached); + + RuntimeInstanceState { + config_id: config_id.clone(), + instance_id: config_id, + display_name, + running: info.running, + tun_required, + tun_attached, + magic_dns_enabled, + need_exit_node, + error_message: info.error_msg, + my_node_info: info.my_node_info.map(my_node_info_to_view), + events: info.events, + routes: info.routes.into_iter().map(route_to_view).collect(), + peers: info.peers.into_iter().map(peer_to_view).collect(), + } +}