diff --git a/clients/openframe-client/src/lib.rs b/clients/openframe-client/src/lib.rs index 310a5fd94..865febf8f 100644 --- a/clients/openframe-client/src/lib.rs +++ b/clients/openframe-client/src/lib.rs @@ -86,6 +86,7 @@ use crate::services::{ InitialKeyService, LastKnownGoodService, UpdateCleanupService, UpdateHandlerService, UpdateStateService, }; +use crate::services::{MachineIdService, MACHINE_ID_HEADER}; #[derive(Debug, Clone, Serialize, Deserialize)] pub struct ServerConfig { @@ -206,8 +207,21 @@ impl Client { let config_service = AgentConfigurationService::new(directory_manager.clone()) .context("Failed to initialize device configuration service")?; + let machine_id_service = MachineIdService::new(&directory_manager); + let machine_id = machine_id_service + .get_or_create() + .context("Failed to get or create machine ID")?; + + let mut default_headers = reqwest::header::HeaderMap::new(); + default_headers.insert( + MACHINE_ID_HEADER, + reqwest::header::HeaderValue::from_str(&machine_id) + .context("Invalid machine ID for header")?, + ); + let http_client = reqwest::Client::builder() .timeout(Duration::from_secs(HTTP_CLIENT_TIMEOUT_SECS)) + .default_headers(default_headers.clone()) // disable TLS verification for dev mode only .danger_accept_invalid_certs(initial_configuration_service.is_local_mode()?) .no_proxy() @@ -218,6 +232,7 @@ impl Client { let download_client = reqwest::Client::builder() .timeout(Duration::from_secs(DOWNLOAD_CLIENT_TIMEOUT_SECS)) + .default_headers(default_headers) .danger_accept_invalid_certs(initial_configuration_service.is_local_mode()?) .no_proxy() .pool_max_idle_per_host(0) @@ -293,6 +308,7 @@ impl Client { auth_service.clone(), tls_config_provider, deactivation_service.clone(), + machine_id_service.clone(), ); // Initialize tool agent file client diff --git a/clients/openframe-client/src/logging/nats_streaming.rs b/clients/openframe-client/src/logging/nats_streaming.rs index 326776fd1..dcbcdd89b 100644 --- a/clients/openframe-client/src/logging/nats_streaming.rs +++ b/clients/openframe-client/src/logging/nats_streaming.rs @@ -9,6 +9,7 @@ use crate::platform::DirectoryManager; use crate::services::device_data_fetcher::DeviceDataFetcher; use crate::services::{ AgentConfigurationService, InitialConfigurationService, InstalledToolsService, + MachineIdService, MACHINE_ID_HEADER, }; use super::log_parser::LogBatchMessage; @@ -21,22 +22,27 @@ const RECONNECT_DELAY_SECS: u64 = 5; const INITIAL_KEY_CHECK_INTERVAL_SECS: u64 = 10; const SOURCE_DISCOVERY_INTERVAL_SECS: u64 = 30; const NATS_SUBJECT: &str = "agents.logs"; -const NATS_HEADER_MACHINE_ID: &str = "openframe-client"; - pub struct NatsLogConnection { jetstream: Option, server_host: String, tenant_domain: String, initial_key: String, + machine_id: String, } impl NatsLogConnection { - pub fn new(server_host: String, tenant_domain: String, initial_key: String) -> Self { + pub fn new( + server_host: String, + tenant_domain: String, + initial_key: String, + machine_id: String, + ) -> Self { Self { jetstream: None, server_host, tenant_domain, initial_key, + machine_id, } } @@ -51,7 +57,7 @@ impl NatsLogConnection { let client = async_nats::ConnectOptions::new() .custom_header("x-tenant-domain", &self.tenant_domain) .custom_header("x-initial-key", &self.initial_key) - .custom_header("x-machine-id", NATS_HEADER_MACHINE_ID) + .custom_header(MACHINE_ID_HEADER, &self.machine_id) .retry_on_initial_connect() .reconnect_delay_callback(|attempt| { let delay = Duration::from_secs(RECONNECT_DELAY_SECS); @@ -119,6 +125,7 @@ pub struct LogStreamingRunManager { agent_config_service: AgentConfigurationService, installed_tools_service: InstalledToolsService, directory_manager: DirectoryManager, + machine_id: String, } impl LogStreamingRunManager { @@ -138,6 +145,7 @@ impl LogStreamingRunManager { let log_file_path = directory_manager.logs_dir().join("openframe.log"); let offset_file_path = directory_manager.secured_dir().join("log_stream_offset"); + let machine_id = MachineIdService::new(directory_manager).get_or_create()?; Ok(Self { server_host, @@ -149,6 +157,7 @@ impl LogStreamingRunManager { agent_config_service: agent_config_service.clone(), installed_tools_service: installed_tools_service.clone(), directory_manager: directory_manager.clone(), + machine_id, }) } @@ -160,6 +169,7 @@ impl LogStreamingRunManager { self.server_host.clone(), self.tenant_domain.clone(), initial_key, + self.machine_id.clone(), ); loop { diff --git a/clients/openframe-client/src/services/machine_id_service.rs b/clients/openframe-client/src/services/machine_id_service.rs new file mode 100644 index 000000000..9dc6d2d27 --- /dev/null +++ b/clients/openframe-client/src/services/machine_id_service.rs @@ -0,0 +1,73 @@ +use anyhow::{Context, Result}; +use std::fs; +use std::path::PathBuf; +use std::sync::Arc; +use tracing::{debug, info}; +use uuid::Uuid; + +use crate::platform::DirectoryManager; + +pub const MACHINE_ID_HEADER: &str = "x-machine-id"; + +// Locally generated machine identity, distinct from the server-assigned machine_id in agent config. +// Persisted in the shared app-support dir so integrated tool agents (mesh, fleet) can read it. +#[derive(Clone)] +pub struct MachineIdService { + file_path: PathBuf, + cached_id: Arc>>, +} + +impl MachineIdService { + pub fn new(directory_manager: &DirectoryManager) -> Self { + Self { + file_path: directory_manager.app_support_dir().join("machine_id"), + cached_id: Arc::new(std::sync::RwLock::new(None)), + } + } + + pub fn get_or_create(&self) -> Result { + if let Some(id) = self.cached_id.read().unwrap().clone() { + return Ok(id); + } + + if let Ok(id) = self.read() { + debug!("Using existing machine ID: {}", id); + *self.cached_id.write().unwrap() = Some(id.clone()); + return Ok(id); + } + + let id = Uuid::new_v4().to_string(); + self.write(&id)?; + info!("Generated new machine ID: {}", id); + *self.cached_id.write().unwrap() = Some(id.clone()); + Ok(id) + } + + // Cached id only; empty string until get_or_create has run. + pub fn get(&self) -> String { + self.cached_id.read().unwrap().clone().unwrap_or_default() + } + + fn read(&self) -> Result { + let content = fs::read_to_string(&self.file_path) + .with_context(|| format!("Failed to read {}", self.file_path.display()))?; + + let id = content.trim(); + if id.is_empty() { + anyhow::bail!("Machine ID file is empty"); + } + Ok(id.to_string()) + } + + fn write(&self, id: &str) -> Result<()> { + if let Some(parent) = self.file_path.parent() { + fs::create_dir_all(parent) + .with_context(|| format!("Failed to create {}", parent.display()))?; + } + + fs::write(&self.file_path, id) + .with_context(|| format!("Failed to write {}", self.file_path.display()))?; + + Ok(()) + } +} diff --git a/clients/openframe-client/src/services/mod.rs b/clients/openframe-client/src/services/mod.rs index d6f2f0b30..e6f6a8b85 100644 --- a/clients/openframe-client/src/services/mod.rs +++ b/clients/openframe-client/src/services/mod.rs @@ -15,6 +15,7 @@ pub mod last_known_good_service; pub mod local_tls_config_provider; pub mod machine_heartbeat_publisher; pub mod machine_heartbeat_run_manager; +pub mod machine_id_service; pub mod mesh_self_heal_service; pub mod nats_connection_manager; pub mod nats_message_publisher; @@ -54,6 +55,7 @@ pub use last_known_good_service::LastKnownGoodService; pub use local_tls_config_provider::LocalTlsConfigProvider; pub use machine_heartbeat_publisher::MachineHeartbeatPublisher; pub use machine_heartbeat_run_manager::MachineHeartbeatRunManager; +pub use machine_id_service::{MachineIdService, MACHINE_ID_HEADER}; pub use nats_connection_manager::NatsConnectionManager; pub use nats_message_publisher::NatsMessagePublisher; pub use openframe_client_info_service::OpenFrameClientInfoService; diff --git a/clients/openframe-client/src/services/nats_connection_manager.rs b/clients/openframe-client/src/services/nats_connection_manager.rs index dc979456c..5de3848d5 100644 --- a/clients/openframe-client/src/services/nats_connection_manager.rs +++ b/clients/openframe-client/src/services/nats_connection_manager.rs @@ -1,7 +1,9 @@ use crate::services::agent_configuration_service::AgentConfigurationService; use crate::services::deactivation_service::DeactivationService; use crate::services::local_tls_config_provider::LocalTlsConfigProvider; -use crate::services::{AgentAuthService, InitialConfigurationService}; +use crate::services::{ + AgentAuthService, InitialConfigurationService, MachineIdService, MACHINE_ID_HEADER, +}; use anyhow::{Context, Result}; use async_nats::{Client, Event}; use log::error; @@ -25,6 +27,7 @@ pub struct NatsConnectionManager { initial_configuration_service: InitialConfigurationService, auth_service: AgentAuthService, deactivation: Arc, + machine_id_service: MachineIdService, } impl NatsConnectionManager { @@ -38,6 +41,7 @@ impl NatsConnectionManager { auth_service: AgentAuthService, tls_config_provider: LocalTlsConfigProvider, deactivation: Arc, + machine_id_service: MachineIdService, ) -> Self { let (reconnect_tx, _) = broadcast::channel(16); Self { @@ -49,6 +53,7 @@ impl NatsConnectionManager { initial_configuration_service, auth_service, deactivation, + machine_id_service, } } @@ -57,7 +62,9 @@ impl NatsConnectionManager { } pub async fn connect(&self) -> Result<()> { + // Server-assigned machine_id names the NATS connection; the local one goes in the header let machine_id = self.config_service.get_machine_id()?; + let local_machine_id = self.machine_id_service.get(); info!( hostname = %self.nats_server_url, @@ -129,7 +136,7 @@ impl NatsConnectionManager { .await } }) - .custom_header("X-MACHINE-ID", &machine_id); + .custom_header(MACHINE_ID_HEADER, &local_machine_id); // Only add TLS config in development mode if self.initial_configuration_service.is_local_mode()? {