From 3f19efb7e3393415e64102f446003689e952a050 Mon Sep 17 00:00:00 2001 From: igor04091968 Date: Thu, 4 Jun 2026 09:19:08 +0300 Subject: [PATCH] feat(agent): add rust telemetry agent v0.3 --- adk-rust/Cargo.lock | 13 + adk-rust/Cargo.toml | 1 + adk-rust/crates/awatch-agent-rs/Cargo.toml | 22 ++ .../awatch-agent-rs/awatch-agent.example.toml | 13 + .../awatch-agent-rs/src/collectors/common.rs | 126 ++++++++ .../awatch-agent-rs/src/collectors/freebsd.rs | 49 ++++ .../awatch-agent-rs/src/collectors/linux.rs | 270 ++++++++++++++++++ .../awatch-agent-rs/src/collectors/mod.rs | 22 ++ .../awatch-agent-rs/src/collectors/windows.rs | 49 ++++ adk-rust/crates/awatch-agent-rs/src/config.rs | 143 ++++++++++ adk-rust/crates/awatch-agent-rs/src/main.rs | 105 +++++++ .../crates/awatch-agent-rs/src/telemetry.rs | 214 ++++++++++++++ .../crates/awatch-agent-rs/src/transport.rs | 205 +++++++++++++ adk-rust/crates/detmir-portal/src/main.rs | 217 ++++++++++++++ .../crates/detmir-portal/src/static/app.css | 6 +- .../crates/detmir-portal/src/static/app.js | 76 ++++- docs/AGENT_ARCHITECTURE_RU.md | 97 +++++++ docs/AGENT_BSD_SUPPORT_RU.md | 44 +++ docs/AGENT_DEPLOYMENT_RU.md | 91 ++++++ docs/PFSENSE_INTEGRATION_RU.md | 35 +++ docs/PORTAL_RU.md | 69 +++++ docs/WORKFORCE_UEBA_MODEL_RU.md | 79 +++++ scripts/detmir-portal-tabs-smoke.mjs | 7 +- 23 files changed, 1949 insertions(+), 4 deletions(-) create mode 100644 adk-rust/crates/awatch-agent-rs/Cargo.toml create mode 100644 adk-rust/crates/awatch-agent-rs/awatch-agent.example.toml create mode 100644 adk-rust/crates/awatch-agent-rs/src/collectors/common.rs create mode 100644 adk-rust/crates/awatch-agent-rs/src/collectors/freebsd.rs create mode 100644 adk-rust/crates/awatch-agent-rs/src/collectors/linux.rs create mode 100644 adk-rust/crates/awatch-agent-rs/src/collectors/mod.rs create mode 100644 adk-rust/crates/awatch-agent-rs/src/collectors/windows.rs create mode 100644 adk-rust/crates/awatch-agent-rs/src/config.rs create mode 100644 adk-rust/crates/awatch-agent-rs/src/main.rs create mode 100644 adk-rust/crates/awatch-agent-rs/src/telemetry.rs create mode 100644 adk-rust/crates/awatch-agent-rs/src/transport.rs create mode 100644 docs/AGENT_ARCHITECTURE_RU.md create mode 100644 docs/AGENT_BSD_SUPPORT_RU.md create mode 100644 docs/AGENT_DEPLOYMENT_RU.md create mode 100644 docs/PFSENSE_INTEGRATION_RU.md create mode 100644 docs/PORTAL_RU.md create mode 100644 docs/WORKFORCE_UEBA_MODEL_RU.md diff --git a/adk-rust/Cargo.lock b/adk-rust/Cargo.lock index 54afa94..c4ee712 100644 --- a/adk-rust/Cargo.lock +++ b/adk-rust/Cargo.lock @@ -283,6 +283,19 @@ dependencies = [ "urlencoding", ] +[[package]] +name = "awatch-agent-rs" +version = "0.3.0" +dependencies = [ + "anyhow", + "chrono", + "clap", + "reqwest", + "serde", + "serde_json", + "tempfile", +] + [[package]] name = "base64" version = "0.22.1" diff --git a/adk-rust/Cargo.toml b/adk-rust/Cargo.toml index 85b93ea..d429e23 100644 --- a/adk-rust/Cargo.toml +++ b/adk-rust/Cargo.toml @@ -1,6 +1,7 @@ [workspace] resolver = "3" members = [ + "crates/awatch-agent-rs", "crates/detmir-auto", "crates/detmir-aw-client", "crates/aw-db-health", diff --git a/adk-rust/crates/awatch-agent-rs/Cargo.toml b/adk-rust/crates/awatch-agent-rs/Cargo.toml new file mode 100644 index 0000000..b66662b --- /dev/null +++ b/adk-rust/crates/awatch-agent-rs/Cargo.toml @@ -0,0 +1,22 @@ +[package] +name = "awatch-agent-rs" +version = "0.3.0" +edition.workspace = true +rust-version.workspace = true +license.workspace = true +publish.workspace = true + +[dependencies] +anyhow.workspace = true +chrono.workspace = true +clap.workspace = true +reqwest.workspace = true +serde.workspace = true +serde_json.workspace = true + +[dev-dependencies] +tempfile.workspace = true + +[features] +default = [] +legacy-powershell = [] diff --git a/adk-rust/crates/awatch-agent-rs/awatch-agent.example.toml b/adk-rust/crates/awatch-agent-rs/awatch-agent.example.toml new file mode 100644 index 0000000..a8e8767 --- /dev/null +++ b/adk-rust/crates/awatch-agent-rs/awatch-agent.example.toml @@ -0,0 +1,13 @@ +server_url = "https://awatch.local/api/telemetry" +api_key = "change-me" +collect_interval_seconds = 60 +role = "workstation" + +enable_processes = true +enable_network = true +enable_security_events = true +enable_workforce_activity = true + +spool_dir = "/var/lib/awatch-agent/spool" +timeout_seconds = 10 +retry_attempts = 3 diff --git a/adk-rust/crates/awatch-agent-rs/src/collectors/common.rs b/adk-rust/crates/awatch-agent-rs/src/collectors/common.rs new file mode 100644 index 0000000..8687440 --- /dev/null +++ b/adk-rust/crates/awatch-agent-rs/src/collectors/common.rs @@ -0,0 +1,126 @@ +use std::env; +use std::fs; +use std::path::Path; +use std::process::Command; + +use chrono::Utc; + +use crate::config::AgentRole; +use crate::telemetry::{SecurityEventInfo, SessionInfo}; + +pub fn command_output(program: &str, args: &[&str]) -> Option { + let output = Command::new(program).args(args).output().ok()?; + if !output.status.success() { + return None; + } + Some(String::from_utf8_lossy(&output.stdout).trim().to_string()) +} + +pub fn hostname() -> String { + env::var("HOSTNAME") + .ok() + .filter(|value| !value.trim().is_empty()) + .or_else(|| fs::read_to_string("/etc/hostname").ok()) + .or_else(|| command_output("hostname", &[])) + .map(|value| value.trim().to_string()) + .filter(|value| !value.is_empty()) + .unwrap_or_else(|| "HOST-EXAMPLE".to_string()) +} + +pub fn username() -> String { + env::var("USER") + .or_else(|_| env::var("USERNAME")) + .unwrap_or_else(|_| "unknown".to_string()) +} + +pub fn domain() -> String { + env::var("USERDOMAIN") + .or_else(|_| env::var("DOMAIN")) + .unwrap_or_default() +} + +pub fn agent_id(hostname: &str) -> String { + format!("awatch-{hostname}") +} + +pub fn role_security_events(role: AgentRole) -> Vec { + if role == AgentRole::Firewall { + vec![SecurityEventInfo { + event_id: "pfsense-mode-prototype".to_string(), + source: "awatch-agent-rs".to_string(), + severity: "INFO".to_string(), + summary: "pfSense/firewall mode enabled; counters are collected from platform-specific probes when available".to_string(), + timestamp: Utc::now(), + evidence: vec!["read-only mode".to_string()], + }] + } else { + Vec::new() + } +} + +pub fn current_session(session_type: &str) -> SessionInfo { + SessionInfo { + session_id: format!("{}-{}", session_type, username()), + username: username(), + session_type: session_type.to_string(), + remote_addr: std::env::var("SSH_CLIENT") + .ok() + .and_then(|value| value.split_whitespace().next().map(str::to_string)), + started_at: None, + active: true, + } +} + +pub fn read_trimmed(path: impl AsRef) -> Option { + fs::read_to_string(path) + .ok() + .map(|value| value.trim().to_string()) + .filter(|value| !value.is_empty()) +} + +pub fn parse_os_release(path: &Path) -> (String, String) { + let text = fs::read_to_string(path).unwrap_or_default(); + let mut name = String::new(); + let mut version = String::new(); + for line in text.lines() { + if let Some(value) = line.strip_prefix("NAME=") { + name = value.trim_matches('"').to_string(); + } + if let Some(value) = line.strip_prefix("VERSION_ID=") { + version = value.trim_matches('"').to_string(); + } + } + if name.is_empty() { + name = "Linux".to_string(); + } + (name, version) +} + +pub fn parse_hex_ipv4(value: &str) -> Option { + if value.len() != 8 { + return None; + } + let raw = u32::from_str_radix(value, 16).ok()?; + let bytes = raw.to_le_bytes(); + Some(format!( + "{}.{}.{}.{}", + bytes[0], bytes[1], bytes[2], bytes[3] + )) +} + +pub fn tcp_state(value: &str) -> &'static str { + match value { + "01" => "ESTABLISHED", + "02" => "SYN_SENT", + "03" => "SYN_RECV", + "04" => "FIN_WAIT1", + "05" => "FIN_WAIT2", + "06" => "TIME_WAIT", + "07" => "CLOSE", + "08" => "CLOSE_WAIT", + "09" => "LAST_ACK", + "0A" => "LISTEN", + "0B" => "CLOSING", + _ => "UNKNOWN", + } +} diff --git a/adk-rust/crates/awatch-agent-rs/src/collectors/freebsd.rs b/adk-rust/crates/awatch-agent-rs/src/collectors/freebsd.rs new file mode 100644 index 0000000..cba3c81 --- /dev/null +++ b/adk-rust/crates/awatch-agent-rs/src/collectors/freebsd.rs @@ -0,0 +1,49 @@ +use anyhow::{Result, bail}; + +use crate::config::AgentRole; +use crate::telemetry::{ + IdentityInfo, NetworkSnapshot, ProcessInfo, ResourceInfo, SecurityEventInfo, SessionSnapshot, + TelemetryCollector, WorkforceActivityInfo, +}; + +#[derive(Debug, Clone)] +pub struct FreeBsdCollector { + role: AgentRole, +} + +impl FreeBsdCollector { + pub fn new(role: AgentRole) -> Self { + Self { role } + } +} + +impl TelemetryCollector for FreeBsdCollector { + fn collect_identity(&self) -> Result { + let _ = self.role; + unsupported() + } + fn collect_sessions(&self) -> Result { + unsupported() + } + fn collect_processes(&self) -> Result> { + unsupported() + } + fn collect_resources(&self) -> Result { + unsupported() + } + fn collect_network(&self) -> Result { + unsupported() + } + fn collect_security_events(&self) -> Result> { + unsupported() + } + fn collect_workforce_activity(&self) -> Result { + unsupported() + } +} + +fn unsupported() -> Result { + bail!( + "FreeBSD/pfSense collector skeleton is present; sysctl/procstat/kvm probes are planned behind the same TelemetryRecord API" + ) +} diff --git a/adk-rust/crates/awatch-agent-rs/src/collectors/linux.rs b/adk-rust/crates/awatch-agent-rs/src/collectors/linux.rs new file mode 100644 index 0000000..6ed0482 --- /dev/null +++ b/adk-rust/crates/awatch-agent-rs/src/collectors/linux.rs @@ -0,0 +1,270 @@ +use std::fs; +use std::path::Path; + +use anyhow::Result; +use chrono::Utc; + +use crate::collectors::common::{ + agent_id, current_session, domain, hostname, parse_hex_ipv4, parse_os_release, read_trimmed, + role_security_events, tcp_state, username, +}; +use crate::config::AgentRole; +use crate::telemetry::{ + IdentityInfo, NetworkConnectionInfo, NetworkInterfaceInfo, NetworkSnapshot, ProcessInfo, + ResourceInfo, SecurityEventInfo, SessionSnapshot, TelemetryCollector, WorkforceActivityInfo, + empty_workforce_activity, +}; + +#[derive(Debug, Clone)] +pub struct LinuxCollector { + role: AgentRole, +} + +impl LinuxCollector { + pub fn new(role: AgentRole) -> Self { + Self { role } + } +} + +impl TelemetryCollector for LinuxCollector { + fn collect_identity(&self) -> Result { + let host = hostname(); + let (os_name, os_version) = parse_os_release(Path::new("/etc/os-release")); + Ok(IdentityInfo { + agent_id: agent_id(&host), + hostname: host, + os_name, + os_version, + platform: "linux".to_string(), + username: username(), + domain: domain(), + }) + } + + fn collect_sessions(&self) -> Result { + let mut active = vec![current_session("local")]; + let mut ssh = Vec::new(); + if std::env::var("SSH_CLIENT").is_ok() || std::env::var("SSH_TTY").is_ok() { + let session = current_session("ssh"); + ssh.push(session.clone()); + active.push(session); + } + Ok(SessionSnapshot { + active_sessions: active, + rdp_sessions: Vec::new(), + ssh_sessions: ssh, + }) + } + + fn collect_processes(&self) -> Result> { + Ok(read_processes(128)) + } + + fn collect_resources(&self) -> Result { + let uptime_seconds = fs::read_to_string("/proc/uptime") + .ok() + .and_then(|text| text.split_whitespace().next()?.parse::().ok()) + .map(|value| value as u64) + .unwrap_or(0); + let (memory_total, memory_available) = read_meminfo(); + Ok(ResourceInfo { + uptime_seconds, + cpu_usage_percent: read_loadavg_percent(), + memory_total, + memory_used: memory_total.saturating_sub(memory_available), + }) + } + + fn collect_network(&self) -> Result { + Ok(NetworkSnapshot { + interfaces: read_interfaces(), + connections: read_connections(), + }) + } + + fn collect_security_events(&self) -> Result> { + let mut events = role_security_events(self.role); + if let Some(summary) = recent_syslog_summary() { + events.push(SecurityEventInfo { + event_id: "linux-syslog-summary".to_string(), + source: "syslog".to_string(), + severity: "INFO".to_string(), + summary, + timestamp: Utc::now(), + evidence: vec!["/var/log/syslog or /var/log/messages".to_string()], + }); + } + Ok(events) + } + + fn collect_workforce_activity(&self) -> Result { + let mut activity = empty_workforce_activity(); + activity.active_today = true; + activity.explanation = vec![ + "Linux collector reports presence and process/network context; application weighting is calculated server-side".to_string(), + ]; + Ok(activity) + } +} + +fn read_meminfo() -> (u64, u64) { + let mut total = 0; + let mut available = 0; + let text = fs::read_to_string("/proc/meminfo").unwrap_or_default(); + for line in text.lines() { + if let Some(value) = line.strip_prefix("MemTotal:") { + total = parse_kib(value); + } + if let Some(value) = line.strip_prefix("MemAvailable:") { + available = parse_kib(value); + } + } + (total, available) +} + +fn parse_kib(value: &str) -> u64 { + value + .split_whitespace() + .next() + .and_then(|item| item.parse::().ok()) + .unwrap_or(0) + * 1024 +} + +fn read_loadavg_percent() -> f64 { + fs::read_to_string("/proc/loadavg") + .ok() + .and_then(|text| text.split_whitespace().next()?.parse::().ok()) + .map(|load| (load * 100.0).clamp(0.0, 100.0)) + .unwrap_or(0.0) +} + +fn read_processes(limit: usize) -> Vec { + let mut items = fs::read_dir("/proc") + .ok() + .into_iter() + .flat_map(|entries| entries.filter_map(|entry| entry.ok())) + .filter_map(|entry| { + let pid = entry.file_name().to_string_lossy().parse::().ok()?; + let stat = fs::read_to_string(entry.path().join("stat")).ok()?; + let name = stat.split_once('(')?.1.split_once(')')?.0.to_string(); + let exe = fs::read_link(entry.path().join("exe")) + .ok() + .map(|path| path.display().to_string()); + let status = fs::read_to_string(entry.path().join("status")).unwrap_or_default(); + let ppid = status + .lines() + .find_map(|line| line.strip_prefix("PPid:")) + .and_then(|value| value.trim().parse::().ok()); + let memory_bytes = status + .lines() + .find_map(|line| line.strip_prefix("VmRSS:")) + .map(parse_kib); + Some(ProcessInfo { + pid, + ppid, + name, + exe, + username: None, + cpu_percent: None, + memory_bytes, + started_at: None, + }) + }) + .collect::>(); + items.sort_by_key(|item| item.pid); + items.truncate(limit); + items +} + +fn read_interfaces() -> Vec { + fs::read_dir("/sys/class/net") + .ok() + .into_iter() + .flat_map(|entries| entries.filter_map(|entry| entry.ok())) + .map(|entry| { + let path = entry.path(); + let name = entry.file_name().to_string_lossy().to_string(); + let up = read_trimmed(path.join("operstate")).is_some_and(|state| state == "up"); + let rx_bytes = + read_trimmed(path.join("statistics/rx_bytes")).and_then(|v| v.parse().ok()); + let tx_bytes = + read_trimmed(path.join("statistics/tx_bytes")).and_then(|v| v.parse().ok()); + NetworkInterfaceInfo { + name, + mac: read_trimmed(path.join("address")), + addresses: Vec::new(), + up, + rx_bytes, + tx_bytes, + } + }) + .collect() +} + +fn read_connections() -> Vec { + let mut items = Vec::new(); + read_proc_net("/proc/net/tcp", "tcp", &mut items); + read_proc_net("/proc/net/udp", "udp", &mut items); + items.truncate(256); + items +} + +fn read_proc_net(path: &str, protocol: &str, items: &mut Vec) { + let text = fs::read_to_string(path).unwrap_or_default(); + for line in text.lines().skip(1) { + let cols = line.split_whitespace().collect::>(); + if cols.len() < 4 { + continue; + } + let Some((local_addr, local_port)) = parse_addr(cols[1]) else { + continue; + }; + let (remote_addr, remote_port) = parse_addr(cols[2]).unwrap_or_default(); + items.push(NetworkConnectionInfo { + protocol: protocol.to_string(), + local_addr, + local_port, + remote_addr: if remote_addr == "0.0.0.0" { + None + } else { + Some(remote_addr) + }, + remote_port: if remote_port == 0 { + None + } else { + Some(remote_port) + }, + state: if protocol == "tcp" { + tcp_state(cols[3]).to_string() + } else { + "UDP".to_string() + }, + pid: None, + }); + } +} + +fn parse_addr(value: &str) -> Option<(String, u16)> { + let (ip, port) = value.split_once(':')?; + Some((parse_hex_ipv4(ip)?, u16::from_str_radix(port, 16).ok()?)) +} + +fn recent_syslog_summary() -> Option { + for path in ["/var/log/syslog", "/var/log/messages"] { + let Ok(text) = fs::read_to_string(path) else { + continue; + }; + let count = text + .lines() + .rev() + .take(200) + .filter(|line| { + let lower = line.to_lowercase(); + lower.contains("error") || lower.contains("fail") || lower.contains("denied") + }) + .count(); + return Some(format!("recent syslog warning/error lines: {count}")); + } + None +} diff --git a/adk-rust/crates/awatch-agent-rs/src/collectors/mod.rs b/adk-rust/crates/awatch-agent-rs/src/collectors/mod.rs new file mode 100644 index 0000000..4dfdf3a --- /dev/null +++ b/adk-rust/crates/awatch-agent-rs/src/collectors/mod.rs @@ -0,0 +1,22 @@ +pub mod common; +pub mod freebsd; +pub mod linux; +pub mod windows; + +use anyhow::{Result, bail}; + +use crate::config::AgentRole; +use crate::telemetry::TelemetryCollector; + +pub fn platform_collector(role: AgentRole) -> Result> { + if cfg!(target_os = "linux") { + return Ok(Box::new(linux::LinuxCollector::new(role))); + } + if cfg!(target_os = "windows") { + return Ok(Box::new(windows::WindowsCollector::new(role))); + } + if cfg!(target_os = "freebsd") { + return Ok(Box::new(freebsd::FreeBsdCollector::new(role))); + } + bail!("unsupported platform for awatch-agent-rs") +} diff --git a/adk-rust/crates/awatch-agent-rs/src/collectors/windows.rs b/adk-rust/crates/awatch-agent-rs/src/collectors/windows.rs new file mode 100644 index 0000000..41d9491 --- /dev/null +++ b/adk-rust/crates/awatch-agent-rs/src/collectors/windows.rs @@ -0,0 +1,49 @@ +use anyhow::{Result, bail}; + +use crate::config::AgentRole; +use crate::telemetry::{ + IdentityInfo, NetworkSnapshot, ProcessInfo, ResourceInfo, SecurityEventInfo, SessionSnapshot, + TelemetryCollector, WorkforceActivityInfo, +}; + +#[derive(Debug, Clone)] +pub struct WindowsCollector { + role: AgentRole, +} + +impl WindowsCollector { + pub fn new(role: AgentRole) -> Self { + Self { role } + } +} + +impl TelemetryCollector for WindowsCollector { + fn collect_identity(&self) -> Result { + let _ = self.role; + unsupported() + } + fn collect_sessions(&self) -> Result { + unsupported() + } + fn collect_processes(&self) -> Result> { + unsupported() + } + fn collect_resources(&self) -> Result { + unsupported() + } + fn collect_network(&self) -> Result { + unsupported() + } + fn collect_security_events(&self) -> Result> { + unsupported() + } + fn collect_workforce_activity(&self) -> Result { + unsupported() + } +} + +fn unsupported() -> Result { + bail!( + "Windows collector requires target_os=windows WinAPI/WMI implementation; PowerShell is not a primary collector" + ) +} diff --git a/adk-rust/crates/awatch-agent-rs/src/config.rs b/adk-rust/crates/awatch-agent-rs/src/config.rs new file mode 100644 index 0000000..36a6666 --- /dev/null +++ b/adk-rust/crates/awatch-agent-rs/src/config.rs @@ -0,0 +1,143 @@ +use std::fs; +use std::path::{Path, PathBuf}; + +use anyhow::{Context, Result}; + +#[derive(Debug, Clone, PartialEq)] +pub struct AgentConfig { + pub server_url: String, + pub api_key: String, + pub collect_interval_seconds: u64, + pub role: AgentRole, + pub enable_processes: bool, + pub enable_network: bool, + pub enable_security_events: bool, + pub enable_workforce_activity: bool, + pub spool_dir: PathBuf, + pub timeout_seconds: u64, + pub retry_attempts: u32, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum AgentRole { + Workstation, + Server, + Firewall, +} + +impl AgentRole { + pub fn parse(value: &str) -> Self { + match value.trim().to_lowercase().as_str() { + "firewall" | "pfsense" => Self::Firewall, + "server" => Self::Server, + _ => Self::Workstation, + } + } +} + +impl Default for AgentConfig { + fn default() -> Self { + Self { + server_url: "https://awatch.local/api/telemetry".to_string(), + api_key: "change-me".to_string(), + collect_interval_seconds: 60, + role: AgentRole::Workstation, + enable_processes: true, + enable_network: true, + enable_security_events: true, + enable_workforce_activity: true, + spool_dir: default_spool_dir(), + timeout_seconds: 10, + retry_attempts: 3, + } + } +} + +pub fn default_config_path() -> PathBuf { + if cfg!(windows) { + PathBuf::from(r"C:\ProgramData\AWatch\agent\awatch-agent.toml") + } else { + PathBuf::from("/etc/awatch-agent/awatch-agent.toml") + } +} + +fn default_spool_dir() -> PathBuf { + if cfg!(windows) { + PathBuf::from(r"C:\ProgramData\AWatch\agent\spool") + } else { + PathBuf::from("/var/lib/awatch-agent/spool") + } +} + +impl AgentConfig { + pub fn load(path: &Path) -> Result { + let text = fs::read_to_string(path).with_context(|| format!("read {}", path.display()))?; + Self::parse_toml_like(&text) + } + + pub fn parse_toml_like(text: &str) -> Result { + let mut config = Self::default(); + for raw in text.lines() { + let line = raw.split('#').next().unwrap_or("").trim(); + if line.is_empty() { + continue; + } + let Some((key, value)) = line.split_once('=') else { + continue; + }; + let key = key.trim(); + let value = value.trim().trim_matches('"'); + match key { + "server_url" => config.server_url = value.to_string(), + "api_key" => config.api_key = value.to_string(), + "collect_interval_seconds" => { + config.collect_interval_seconds = value.parse().unwrap_or(60) + } + "role" => config.role = AgentRole::parse(value), + "enable_processes" => config.enable_processes = parse_bool(value, true), + "enable_network" => config.enable_network = parse_bool(value, true), + "enable_security_events" => config.enable_security_events = parse_bool(value, true), + "enable_workforce_activity" => { + config.enable_workforce_activity = parse_bool(value, true) + } + "spool_dir" => config.spool_dir = PathBuf::from(value), + "timeout_seconds" => config.timeout_seconds = value.parse().unwrap_or(10), + "retry_attempts" => config.retry_attempts = value.parse().unwrap_or(3), + _ => {} + } + } + Ok(config) + } +} + +fn parse_bool(value: &str, fallback: bool) -> bool { + match value.trim().to_lowercase().as_str() { + "1" | "true" | "yes" | "on" => true, + "0" | "false" | "no" | "off" => false, + _ => fallback, + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn parses_agent_config() { + let config = AgentConfig::parse_toml_like( + r#" +server_url = "https://awatch.local/api/telemetry" +api_key = "change-me" +collect_interval_seconds = 30 +role = "firewall" +enable_processes = false +spool_dir = "/tmp/awatch-spool" +"#, + ) + .unwrap(); + assert_eq!(config.role, AgentRole::Firewall); + assert_eq!(config.collect_interval_seconds, 30); + assert!(!config.enable_processes); + assert_eq!(config.spool_dir, PathBuf::from("/tmp/awatch-spool")); + } +} diff --git a/adk-rust/crates/awatch-agent-rs/src/main.rs b/adk-rust/crates/awatch-agent-rs/src/main.rs new file mode 100644 index 0000000..1a1febd --- /dev/null +++ b/adk-rust/crates/awatch-agent-rs/src/main.rs @@ -0,0 +1,105 @@ +mod collectors; +mod config; +mod telemetry; +mod transport; + +use std::path::PathBuf; +use std::thread; +use std::time::Duration; + +use anyhow::{Context, Result}; +use clap::Parser; +use config::{AgentConfig, AgentRole, default_config_path}; +use transport::{TelemetryTransport, spool_health}; + +#[derive(Debug, Parser)] +#[command(about = "AWatch-rus Rust telemetry agent")] +struct Cli { + #[arg(long, env = "AWATCH_AGENT_CONFIG")] + config: Option, + + #[arg(long, env = "AWATCH_AGENT_SERVER_URL")] + server_url: Option, + + #[arg(long, env = "AWATCH_AGENT_API_KEY")] + api_key: Option, + + #[arg(long, env = "AWATCH_AGENT_ROLE")] + role: Option, + + #[arg(long)] + once: bool, + + #[arg(long)] + print_json: bool, + + #[arg(long)] + flush_spool: bool, + + #[arg(long)] + spool_health: bool, +} + +fn main() { + let code = match run() { + Ok(code) => code, + Err(err) => { + eprintln!("{err:#}"); + 1 + } + }; + std::process::exit(code); +} + +fn run() -> Result { + let cli = Cli::parse(); + let mut config = load_config(cli.config.as_ref())?; + if let Some(server_url) = cli.server_url { + config.server_url = server_url; + } + if let Some(api_key) = cli.api_key { + config.api_key = api_key; + } + if let Some(role) = cli.role { + config.role = AgentRole::parse(&role); + } + if cli.spool_health { + println!( + "{}", + serde_json::to_string_pretty(&spool_health(&config.spool_dir))? + ); + return Ok(0); + } + + let transport = TelemetryTransport::new(&config); + if cli.flush_spool { + let flushed = transport.flush_spool()?; + println!("{}", serde_json::json!({"ok": true, "flushed": flushed})); + return Ok(0); + } + + loop { + let collector = collectors::platform_collector(config.role)?; + let record = collector.collect_all()?; + if cli.print_json { + println!("{}", serde_json::to_string_pretty(&record)?); + } else if let Err(err) = transport.send_or_spool(&record) { + eprintln!("{err:#}"); + } + if cli.once { + break; + } + thread::sleep(Duration::from_secs(config.collect_interval_seconds)); + } + Ok(0) +} + +fn load_config(path: Option<&PathBuf>) -> Result { + let path = path.cloned().unwrap_or_else(default_config_path); + if path.exists() { + AgentConfig::load(&path) + } else { + AgentConfig::parse_toml_like("") + .with_context(|| format!("load default config because {} is absent", path.display())) + } +} diff --git a/adk-rust/crates/awatch-agent-rs/src/telemetry.rs b/adk-rust/crates/awatch-agent-rs/src/telemetry.rs new file mode 100644 index 0000000..5b3fb87 --- /dev/null +++ b/adk-rust/crates/awatch-agent-rs/src/telemetry.rs @@ -0,0 +1,214 @@ +use anyhow::Result; +use chrono::{DateTime, Utc}; +use serde::{Deserialize, Serialize}; + +pub const COLLECTOR_VERSION: &str = env!("CARGO_PKG_VERSION"); + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +pub struct TelemetryRecord { + pub agent_id: String, + pub hostname: String, + pub os_name: String, + pub os_version: String, + pub platform: String, + pub username: String, + pub domain: String, + pub timestamp: DateTime, + pub uptime_seconds: u64, + pub cpu_usage_percent: f64, + pub memory_total: u64, + pub memory_used: u64, + pub active_sessions: Vec, + pub rdp_sessions: Vec, + pub ssh_sessions: Vec, + pub processes: Vec, + pub network_interfaces: Vec, + pub network_connections: Vec, + pub workforce_activity: WorkforceActivityInfo, + pub security_events: Vec, + pub collector_version: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +pub struct IdentityInfo { + pub agent_id: String, + pub hostname: String, + pub os_name: String, + pub os_version: String, + pub platform: String, + pub username: String, + pub domain: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +pub struct ResourceInfo { + pub uptime_seconds: u64, + pub cpu_usage_percent: f64, + pub memory_total: u64, + pub memory_used: u64, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +pub struct SessionInfo { + pub session_id: String, + pub username: String, + pub session_type: String, + pub remote_addr: Option, + pub started_at: Option>, + pub active: bool, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +pub struct ProcessInfo { + pub pid: u32, + pub ppid: Option, + pub name: String, + pub exe: Option, + pub username: Option, + pub cpu_percent: Option, + pub memory_bytes: Option, + pub started_at: Option>, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +pub struct NetworkInterfaceInfo { + pub name: String, + pub mac: Option, + pub addresses: Vec, + pub up: bool, + pub rx_bytes: Option, + pub tx_bytes: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +pub struct NetworkConnectionInfo { + pub protocol: String, + pub local_addr: String, + pub local_port: u16, + pub remote_addr: Option, + pub remote_port: Option, + pub state: String, + pub pid: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +pub struct WorkforceActivityInfo { + pub active_today: bool, + pub activity_index: Option, + pub department: Option, + pub owner: Option, + pub work_applications: Vec, + pub idle_seconds: Option, + pub explanation: Vec, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +pub struct SecurityEventInfo { + pub event_id: String, + pub source: String, + pub severity: String, + pub summary: String, + pub timestamp: DateTime, + pub evidence: Vec, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +pub struct SessionSnapshot { + pub active_sessions: Vec, + pub rdp_sessions: Vec, + pub ssh_sessions: Vec, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +pub struct NetworkSnapshot { + pub interfaces: Vec, + pub connections: Vec, +} + +pub trait TelemetryCollector { + fn collect_identity(&self) -> Result; + fn collect_sessions(&self) -> Result; + fn collect_processes(&self) -> Result>; + fn collect_resources(&self) -> Result; + fn collect_network(&self) -> Result; + fn collect_security_events(&self) -> Result>; + fn collect_workforce_activity(&self) -> Result; + + fn collect_all(&self) -> Result { + let identity = self.collect_identity()?; + let sessions = self.collect_sessions()?; + let resources = self.collect_resources()?; + let network = self.collect_network()?; + Ok(TelemetryRecord { + agent_id: identity.agent_id, + hostname: identity.hostname, + os_name: identity.os_name, + os_version: identity.os_version, + platform: identity.platform, + username: identity.username, + domain: identity.domain, + timestamp: Utc::now(), + uptime_seconds: resources.uptime_seconds, + cpu_usage_percent: resources.cpu_usage_percent, + memory_total: resources.memory_total, + memory_used: resources.memory_used, + active_sessions: sessions.active_sessions, + rdp_sessions: sessions.rdp_sessions, + ssh_sessions: sessions.ssh_sessions, + processes: self.collect_processes()?, + network_interfaces: network.interfaces, + network_connections: network.connections, + workforce_activity: self.collect_workforce_activity()?, + security_events: self.collect_security_events()?, + collector_version: COLLECTOR_VERSION.to_string(), + }) + } +} + +pub fn empty_workforce_activity() -> WorkforceActivityInfo { + WorkforceActivityInfo { + active_today: false, + activity_index: None, + department: None, + owner: None, + work_applications: Vec::new(), + idle_seconds: None, + explanation: vec!["activity scoring requires workstation activity events".to_string()], + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn telemetry_record_serializes_required_fields() { + let record = TelemetryRecord { + agent_id: "agent-1".to_string(), + hostname: "HOST-EXAMPLE".to_string(), + os_name: "Linux".to_string(), + os_version: "test".to_string(), + platform: "linux".to_string(), + username: "user".to_string(), + domain: "".to_string(), + timestamp: Utc::now(), + uptime_seconds: 1, + cpu_usage_percent: 0.0, + memory_total: 10, + memory_used: 5, + active_sessions: Vec::new(), + rdp_sessions: Vec::new(), + ssh_sessions: Vec::new(), + processes: Vec::new(), + network_interfaces: Vec::new(), + network_connections: Vec::new(), + workforce_activity: empty_workforce_activity(), + security_events: Vec::new(), + collector_version: COLLECTOR_VERSION.to_string(), + }; + let value = serde_json::to_value(record).unwrap(); + assert_eq!(value["agent_id"], "agent-1"); + assert!(value.get("network_connections").unwrap().is_array()); + assert!(value.get("workforce_activity").is_some()); + } +} diff --git a/adk-rust/crates/awatch-agent-rs/src/transport.rs b/adk-rust/crates/awatch-agent-rs/src/transport.rs new file mode 100644 index 0000000..34b3724 --- /dev/null +++ b/adk-rust/crates/awatch-agent-rs/src/transport.rs @@ -0,0 +1,205 @@ +use std::fs; +use std::path::{Path, PathBuf}; +use std::thread; +use std::time::Duration; + +use anyhow::{Context, Result, anyhow}; +use chrono::Utc; +use reqwest::blocking::Client; +use reqwest::header::{HeaderMap, HeaderValue}; + +use crate::config::AgentConfig; +use crate::telemetry::TelemetryRecord; + +#[derive(Debug, Clone)] +pub struct TelemetryTransport { + server_url: String, + api_key: String, + spool_dir: PathBuf, + timeout: Duration, + retry_attempts: u32, +} + +impl TelemetryTransport { + pub fn new(config: &AgentConfig) -> Self { + Self { + server_url: config.server_url.clone(), + api_key: config.api_key.clone(), + spool_dir: config.spool_dir.clone(), + timeout: Duration::from_secs(config.timeout_seconds), + retry_attempts: config.retry_attempts, + } + } + + pub fn send_or_spool(&self, record: &TelemetryRecord) -> Result<()> { + match self.send(record) { + Ok(()) => Ok(()), + Err(err) => { + self.spool(record)?; + Err(err) + } + } + } + + pub fn send(&self, record: &TelemetryRecord) -> Result<()> { + let client = Client::builder() + .timeout(self.timeout) + .build() + .context("build telemetry HTTP client")?; + let mut headers = HeaderMap::new(); + headers.insert( + "x-api-key", + HeaderValue::from_str(&self.api_key).context("invalid api key header")?, + ); + let mut last_error = None; + for attempt in 0..self.retry_attempts.max(1) { + let result = client + .post(&self.server_url) + .headers(headers.clone()) + .json(record) + .send() + .and_then(|response| response.error_for_status()) + .map(|_| ()); + match result { + Ok(()) => return Ok(()), + Err(err) => { + last_error = Some(err); + let backoff = Duration::from_millis(250 * u64::from(attempt + 1)); + thread::sleep(backoff); + } + } + } + Err(anyhow!( + "telemetry POST failed: {}", + last_error + .map(|err| err.to_string()) + .unwrap_or_else(|| "unknown error".to_string()) + )) + } + + pub fn spool(&self, record: &TelemetryRecord) -> Result { + fs::create_dir_all(&self.spool_dir) + .with_context(|| format!("create spool {}", self.spool_dir.display()))?; + let file_name = format!( + "{}-{}.json", + record.timestamp.format("%Y%m%dT%H%M%S%.3fZ"), + sanitize_file_part(&record.agent_id) + ); + let path = self.spool_dir.join(file_name); + fs::write(&path, serde_json::to_vec(record)?) + .with_context(|| format!("write spool {}", path.display()))?; + Ok(path) + } + + pub fn flush_spool(&self) -> Result { + flush_spool_dir(&self.spool_dir, |record| self.send(record)) + } +} + +pub fn flush_spool_dir(spool_dir: &Path, mut sender: F) -> Result +where + F: FnMut(&TelemetryRecord) -> Result<()>, +{ + if !spool_dir.exists() { + return Ok(0); + } + let mut sent = 0; + let mut entries = fs::read_dir(spool_dir) + .with_context(|| format!("read spool {}", spool_dir.display()))? + .filter_map(|entry| entry.ok()) + .map(|entry| entry.path()) + .filter(|path| path.extension().is_some_and(|ext| ext == "json")) + .collect::>(); + entries.sort(); + for path in entries { + let data = fs::read(&path).with_context(|| format!("read {}", path.display()))?; + let record: TelemetryRecord = + serde_json::from_slice(&data).with_context(|| format!("parse {}", path.display()))?; + sender(&record)?; + fs::remove_file(&path).with_context(|| format!("remove {}", path.display()))?; + sent += 1; + } + Ok(sent) +} + +fn sanitize_file_part(value: &str) -> String { + value + .chars() + .map(|ch| { + if ch.is_ascii_alphanumeric() || ch == '-' || ch == '_' { + ch + } else { + '_' + } + }) + .collect() +} + +pub fn spool_health(spool_dir: &Path) -> serde_json::Value { + let queued = fs::read_dir(spool_dir) + .ok() + .into_iter() + .flat_map(|entries| entries.filter_map(|entry| entry.ok())) + .filter(|entry| entry.path().extension().is_some_and(|ext| ext == "json")) + .count(); + serde_json::json!({ + "generated_at_utc": Utc::now(), + "spool_dir": spool_dir.display().to_string(), + "queued": queued, + }) +} + +#[cfg(test)] +mod tests { + use tempfile::tempdir; + + use super::*; + use crate::telemetry::{TelemetryRecord, empty_workforce_activity}; + + fn record() -> TelemetryRecord { + TelemetryRecord { + agent_id: "agent/1".to_string(), + hostname: "HOST-EXAMPLE".to_string(), + os_name: "Linux".to_string(), + os_version: "test".to_string(), + platform: "linux".to_string(), + username: "user".to_string(), + domain: "".to_string(), + timestamp: Utc::now(), + uptime_seconds: 1, + cpu_usage_percent: 0.0, + memory_total: 1, + memory_used: 1, + active_sessions: Vec::new(), + rdp_sessions: Vec::new(), + ssh_sessions: Vec::new(), + processes: Vec::new(), + network_interfaces: Vec::new(), + network_connections: Vec::new(), + workforce_activity: empty_workforce_activity(), + security_events: Vec::new(), + collector_version: "test".to_string(), + } + } + + #[test] + fn spools_and_flushes_records() { + let dir = tempdir().unwrap(); + let config = AgentConfig { + spool_dir: dir.path().to_path_buf(), + ..AgentConfig::default() + }; + let transport = TelemetryTransport::new(&config); + let path = transport.spool(&record()).unwrap(); + assert!(path.is_file()); + let mut seen = 0; + let flushed = flush_spool_dir(dir.path(), |_| { + seen += 1; + Ok(()) + }) + .unwrap(); + assert_eq!(flushed, 1); + assert_eq!(seen, 1); + assert!(!path.exists()); + } +} diff --git a/adk-rust/crates/detmir-portal/src/main.rs b/adk-rust/crates/detmir-portal/src/main.rs index 1b5ae8c..98152a4 100644 --- a/adk-rust/crates/detmir-portal/src/main.rs +++ b/adk-rust/crates/detmir-portal/src/main.rs @@ -138,6 +138,20 @@ struct Cli { #[arg(long, env = "DETMIR_PORTAL_EVIDENCE_UPLOAD_TOKEN")] evidence_upload_token: Option, + + #[arg( + long, + default_value = "change-me", + env = "DETMIR_PORTAL_TELEMETRY_API_KEY" + )] + telemetry_api_key: String, + + #[arg( + long, + default_value = "/var/lib/detmir-portal/telemetry.jsonl", + env = "DETMIR_PORTAL_TELEMETRY_STORE_PATH" + )] + telemetry_store_path: PathBuf, } #[derive(Clone, Debug, Serialize)] @@ -607,6 +621,9 @@ fn handle_request(request: Request, args: &Cli, snapshot_cache: &SnapshotCache) if method == Method::Post && path == "/api/incidents/action" { return handle_incident_action(request, args); } + if method == Method::Post && path == "/api/telemetry" { + return handle_telemetry_ingest(request, args); + } if method != Method::Get { return respond_text(request, StatusCode(405), "Method Not Allowed", "text/plain"); } @@ -3282,6 +3299,119 @@ fn handle_incident_action(mut request: Request, args: &Cli) -> Result<()> { } } +fn handle_telemetry_ingest(mut request: Request, args: &Cli) -> Result<()> { + if !telemetry_authorized(&request, args) { + return respond_json_status( + request, + StatusCode(401), + &json!({ + "ok": false, + "error": "telemetry api key is missing or invalid" + }), + ); + } + let mut body = String::new(); + request + .as_reader() + .take(1024 * 1024) + .read_to_string(&mut body)?; + let response = apply_telemetry_ingest(args, &body); + match response { + Ok(response) => respond_json(request, &response), + Err(err) => respond_json_status( + request, + StatusCode(400), + &json!({ + "ok": false, + "error": err.to_string() + }), + ), + } +} + +fn apply_telemetry_ingest(args: &Cli, body: &str) -> Result { + let payload: Value = + serde_json::from_str(body).map_err(|err| anyhow!("invalid telemetry JSON: {err}"))?; + validate_telemetry_payload(&payload)?; + let received_at_utc = now(); + let envelope = json!({ + "received_at_utc": received_at_utc, + "prototype": true, + "record": payload, + }); + if let Some(parent) = args.telemetry_store_path.parent() { + fs::create_dir_all(parent).with_context(|| format!("create {}", parent.display()))?; + } + let mut file = OpenOptions::new() + .create(true) + .append(true) + .open(&args.telemetry_store_path) + .with_context(|| format!("open {}", args.telemetry_store_path.display()))?; + writeln!(file, "{}", serde_json::to_string(&envelope)?) + .with_context(|| format!("append {}", args.telemetry_store_path.display()))?; + Ok(json!({ + "ok": true, + "prototype": true, + "stored": "file-backed-jsonl", + "received_at_utc": received_at_utc, + })) +} + +fn validate_telemetry_payload(payload: &Value) -> Result<()> { + let Some(object) = payload.as_object() else { + return Err(anyhow!("telemetry payload must be a JSON object")); + }; + for field in [ + "agent_id", + "hostname", + "os_name", + "os_version", + "platform", + "username", + "timestamp", + "uptime_seconds", + "cpu_usage_percent", + "memory_total", + "memory_used", + "active_sessions", + "rdp_sessions", + "ssh_sessions", + "processes", + "network_interfaces", + "network_connections", + "workforce_activity", + "security_events", + "collector_version", + ] { + if !object.contains_key(field) { + return Err(anyhow!("telemetry field is missing: {field}")); + } + } + for field in [ + "active_sessions", + "rdp_sessions", + "ssh_sessions", + "processes", + "network_interfaces", + "network_connections", + "security_events", + ] { + if payload.get(field).and_then(Value::as_array).is_none() { + return Err(anyhow!("telemetry field must be an array: {field}")); + } + } + if payload + .get("workforce_activity") + .and_then(Value::as_object) + .is_none() + { + return Err(anyhow!( + "telemetry field must be an object: workforce_activity" + )); + } + Ok(()) +} + fn apply_incident_action(args: &Cli, actor: &str, body: &str) -> Result { let action: IncidentActionRequest = serde_json::from_str(body).map_err(|err| anyhow!("invalid incident action JSON: {err}"))?; @@ -4059,6 +4189,24 @@ fn upload_authorized(request: &Request, args: &Cli) -> bool { constant_time_eq(actual.as_bytes(), expected.as_bytes()) } +fn telemetry_authorized(request: &Request, args: &Cli) -> bool { + let expected = args.telemetry_api_key.trim(); + if expected.is_empty() || expected == "change-me" { + return false; + } + let actual = request + .headers() + .iter() + .find(|header| header.field.equiv("x-api-key")) + .map(|header| header.value.as_str().trim().to_string()) + .or_else(|| bearer_token(request)); + actual + .as_deref() + .filter(|value| !value.is_empty()) + .map(|value| constant_time_eq(value.as_bytes(), expected.as_bytes())) + .unwrap_or(false) +} + fn bearer_token(request: &Request) -> Option { request .headers() @@ -4444,6 +4592,15 @@ fn respond_json(request: Request, value: &T) -> Result<()> { ) } +fn respond_json_status( + request: Request, + status: StatusCode, + value: &T, +) -> Result<()> { + let body = serde_json::to_string_pretty(value)?; + respond_text(request, status, &body, "application/json; charset=utf-8") +} + fn respond_text( request: Request, status: StatusCode, @@ -4610,6 +4767,8 @@ mod tests { json_smoke: false, evidence_only: false, evidence_upload_token: None, + telemetry_api_key: "test-key".to_string(), + telemetry_store_path: dir.path().join("telemetry.jsonl"), }; let found = resolve_screenshot_file(&args, &None, &Some(digest.clone())) .unwrap() @@ -4651,6 +4810,64 @@ mod tests { assert!(!constant_time_eq(b"secret", b"secret2")); } + #[test] + fn telemetry_ingest_validates_and_appends_jsonl() { + let dir = tempfile::tempdir().unwrap(); + let args = Cli { + bind: "127.0.0.1:0".to_string(), + status_cmd: "true".to_string(), + check_cmd: "true".to_string(), + failed_units_cmd: "true".to_string(), + worktime_url: "http://127.0.0.1".to_string(), + one_c_url: "http://127.0.0.1".to_string(), + workforce_policy_path: dir.path().join("workforce-policy.json"), + ueba_policy_path: dir.path().join("ueba-policy.yaml"), + timeout_seconds: 1, + state_dir: dir.path().join("state"), + dlp_db_path: dir.path().join("dlp.sqlite"), + evidence_root: dir.path().to_path_buf(), + readiness_bundle_dir: dir.path().join("readiness-bundle"), + evidence_limit: 10, + evidence_max_bytes: 1024, + json_smoke: false, + evidence_only: false, + evidence_upload_token: None, + telemetry_api_key: "test-key".to_string(), + telemetry_store_path: dir.path().join("telemetry/telemetry.jsonl"), + }; + let payload = json!({ + "agent_id": "agent-1", + "hostname": "HOST-EXAMPLE", + "os_name": "Linux", + "os_version": "test", + "platform": "linux", + "username": "user", + "domain": "", + "timestamp": "2026-06-04T00:00:00Z", + "uptime_seconds": 1, + "cpu_usage_percent": 0.0, + "memory_total": 1, + "memory_used": 1, + "active_sessions": [], + "rdp_sessions": [], + "ssh_sessions": [], + "processes": [], + "network_interfaces": [], + "network_connections": [], + "workforce_activity": {"active_today": true, "explanation": []}, + "security_events": [], + "collector_version": "0.3.0" + }); + let response = apply_telemetry_ingest(&args, &serde_json::to_string(&payload).unwrap()) + .expect("valid telemetry should be accepted"); + assert_eq!(response["ok"], true); + assert_eq!(response["stored"], "file-backed-jsonl"); + let stored = fs::read_to_string(&args.telemetry_store_path).unwrap(); + assert!(stored.contains("\"prototype\":true")); + assert!(stored.contains("HOST-EXAMPLE")); + assert!(apply_telemetry_ingest(&args, r#"{"agent_id":"only"}"#).is_err()); + } + #[test] fn human_duration_formats_hhmm() { assert_eq!(human_duration(0), "00:00"); diff --git a/adk-rust/crates/detmir-portal/src/static/app.css b/adk-rust/crates/detmir-portal/src/static/app.css index b2f48aa..2120926 100644 --- a/adk-rust/crates/detmir-portal/src/static/app.css +++ b/adk-rust/crates/detmir-portal/src/static/app.css @@ -721,14 +721,16 @@ h1 { margin-top: 14px; } -.leader-grid { +.leader-grid, +.investigation-grid { display: grid; grid-template-columns: repeat(4, minmax(0, 1fr)); gap: 8px; margin: 12px 0; } -.leader-grid > div { +.leader-grid > div, +.investigation-grid > div { display: grid; gap: 4px; min-height: 64px; diff --git a/adk-rust/crates/detmir-portal/src/static/app.js b/adk-rust/crates/detmir-portal/src/static/app.js index 7e32583..99d1f87 100644 --- a/adk-rust/crates/detmir-portal/src/static/app.js +++ b/adk-rust/crates/detmir-portal/src/static/app.js @@ -1179,10 +1179,80 @@ function renderDlpEvidence(evidence) { `).join("")}`; } +function buildAutoInvestigation(report) { + report = periodReport(report || state.reports || {}); + const rows = departmentRows(report); + const risky = [...rows].sort((a, b) => statusWeight(b.status) - statusWeight(a.status) || (a.activity ?? 101) - (b.activity ?? 101))[0]; + const reasons = Array.isArray(report?.ueba_risk?.reasons) ? report.ueba_risk.reasons : []; + const risk = reasons[0] || {}; + const status = risky?.status || risk.status || report?.ueba_risk?.status || "INFO"; + const department = risky?.label || "Портфель"; + const owner = risky?.responsible || "ответственный не назначен"; + const generated = report?.generated_at_utc || new Date().toISOString(); + const summary = risk.label || risk.code || risky?.reason || "Система сформировала риск-сигнал для ручной проверки"; + return { + incident_id: `auto-${String(department).toLowerCase().replace(/[^a-zа-я0-9]+/gi, "-").replace(/^-|-$/g, "") || "risk"}`, + risk_id: risk.code || risk.label || "workforce-ueba-risk", + department, + owner, + activity_index: risky?.activityText || "нет данных", + deviation: risky?.deviation || "нет данных", + status, + summary, + why_it_is_risk: risky?.reason || risk.value || "риск может указывать на просадку активности, отклонение от нормы или событие безопасности", + what_to_check: risky?.check || risk.recommendation || "проверить первичные события ActivityWatch, RDP/1C активность, процессы и сетевые сигналы", + recommended_actions: [ + "назначить ответственного за ручную проверку", + "сопоставить риск с журналами активности и бизнес-задачей", + "зафиксировать вывод в отчете по инциденту" + ], + evidence: [ + "RDP activity: проверяется по событиям рабочего времени", + "process activity: проверяется по процессам и приложениям", + "network activity: проверяется по сетевым сигналам", + "proxy activity: подключается как внешний источник", + "pfSense events: учитываются при наличии интеграционного слоя" + ], + generated_at: generated + }; +} + +function renderAutoInvestigationCard(report) { + const card = buildAutoInvestigation(report); + return ` +
+
+
+

Автоматическая карточка расследования

+

Расследование сформировано автоматически. Решение принимает ответственный сотрудник.

+
+ ${ui(card.status)} +
+
+
incident_id${ui(card.incident_id)}
+
risk_id${ui(card.risk_id)}
+
department${ui(card.department)}
+
owner${ui(card.owner)}
+
activity_index${ui(card.activity_index)}
+
deviation${ui(card.deviation)}
+
generated_at${ui(card.generated_at)}
+
+
+
summary${ui(card.summary)}
+
why_it_is_risk${ui(card.why_it_is_risk)}
+
what_to_check${ui(card.what_to_check)}
+
recommended_actions${ui(card.recommended_actions.join("; "))}
+
evidence${ui(card.evidence.join("; "))}
+
+
+ `; +} + function renderIncidents(data) { const links = state.links || {}; const incidents = Array.isArray(data) ? data : data.incidents; const evidence = Array.isArray(data) ? null : data.evidence; + const reports = Array.isArray(data) ? state.reports : data.reports; return `
@@ -1201,6 +1271,7 @@ function renderIncidents(data) { ${renderDlpLinks(links)}
+ ${renderAutoInvestigationCard(reports)}

Материалы: скриншоты, хеши, файлы

${renderDlpEvidence(evidence)} @@ -1391,7 +1462,9 @@ async function refresh() { if (state.tab === "incidents") { const data = await loadJson("/incidents"); const evidence = await loadJson("/dlp/evidence").catch(error => ({ ok: false, error: error.message, items: [] })); - content.innerHTML = renderIncidents({ incidents: data, evidence }); + const reports = await loadJson("/reports").catch(() => state.reports || {}); + state.reports = reports; + content.innerHTML = renderIncidents({ incidents: data, evidence, reports }); } if (state.tab === "perimeter") { const data = await loadJson("/owner"); @@ -1450,6 +1523,7 @@ document.addEventListener("click", event => { document.addEventListener("click", event => { const button = event.target.closest("[data-open-investigation]"); if (!button) return; + state.investigationRequested = true; setTab("incidents"); }); diff --git a/docs/AGENT_ARCHITECTURE_RU.md b/docs/AGENT_ARCHITECTURE_RU.md new file mode 100644 index 0000000..36967c1 --- /dev/null +++ b/docs/AGENT_ARCHITECTURE_RU.md @@ -0,0 +1,97 @@ +# AWatch-rus Rust Agent Architecture + +Документ описывает агентный слой AWatch-rus v0.3. + +## Назначение + +`awatch-agent-rs` является собственным Rust-агентом телеметрии для контура: + +```text +Agent -> Telemetry -> Analytics -> Risk -> Investigation -> Report +``` + +Агент собирает техническую и операционную телеметрию рабочего места или сервера и отправляет ее в серверный endpoint `POST /api/telemetry`. + +## Границы реализации v0.3 + +- Linux collector собирает реальные данные через `/proc`, `/sys`, окружение сессии и системные журналы. +- Windows collector имеет стабильный публичный интерфейс и подготовлен под WinAPI, ETW, Event Log API и WMI-библиотеки Rust. +- FreeBSD collector имеет стабильный публичный интерфейс и подготовлен под `sysctl`, `procstat`, `kvm` и стандартные интерфейсы FreeBSD. +- PowerShell не является основным механизмом сбора. Он допускается только как будущий `legacy` fallback под feature flag. +- Агент не содержит скрытых функций, драйверов ядра, кейлоггера, записи экрана, перехвата документов и контентного DLP-анализа. + +## Crate + +```text +adk-rust/crates/awatch-agent-rs +``` + +Структура: + +```text +src/main.rs +src/config.rs +src/telemetry.rs +src/transport.rs +src/collectors/mod.rs +src/collectors/common.rs +src/collectors/linux.rs +src/collectors/windows.rs +src/collectors/freebsd.rs +``` + +## Единая модель TelemetryRecord + +Все платформы должны отдавать один JSON-контракт: + +- `agent_id` +- `hostname` +- `os_name` +- `os_version` +- `platform` +- `username` +- `domain` +- `timestamp` +- `uptime_seconds` +- `cpu_usage_percent` +- `memory_total` +- `memory_used` +- `active_sessions` +- `rdp_sessions` +- `ssh_sessions` +- `processes` +- `network_interfaces` +- `network_connections` +- `workforce_activity` +- `security_events` +- `collector_version` + +Дополнительные структуры: `SessionInfo`, `ProcessInfo`, `NetworkInterfaceInfo`, `NetworkConnectionInfo`, `WorkforceActivityInfo`, `SecurityEventInfo`. + +## Collector trait + +`TelemetryCollector` задает единый контракт: + +```text +collect_identity() +collect_sessions() +collect_processes() +collect_resources() +collect_network() +collect_security_events() +collect_workforce_activity() +collect_all() +``` + +Такой интерфейс позволяет расширять Windows, Linux, FreeBSD и pfSense mode без изменения серверного API. + +## Транспорт + +Транспорт использует JSON over HTTPS: + +```text +POST /api/telemetry +x-api-key: +``` + +При недоступности сервера запись не теряется: агент сохраняет JSON в spool и повторно отправляет после восстановления связи. diff --git a/docs/AGENT_BSD_SUPPORT_RU.md b/docs/AGENT_BSD_SUPPORT_RU.md new file mode 100644 index 0000000..ac10f24 --- /dev/null +++ b/docs/AGENT_BSD_SUPPORT_RU.md @@ -0,0 +1,44 @@ +# AWatch-rus BSD Support + +## Цель + +BSD-слой нужен для будущей поддержки FreeBSD и pfSense mode без изменения публичного `TelemetryRecord`. + +## FreeBSD collector + +В v0.3 FreeBSD collector имеет стабильный интерфейс и честно возвращает ограничение реализации, если бинарник собран под FreeBSD до включения нативных источников. + +Планируемые источники: + +- `sysctl`; +- `procstat`; +- `kvm`; +- сетевые интерфейсы ОС; +- системный syslog; +- сведения об SSH-сессиях. + +## pfSense mode + +Роль: + +```bash +awatch-agent-rs --role firewall +``` + +Назначение: read-only сбор телеметрии сетевого периметра. + +Будущие сигналы: + +- interfaces; +- gateway status; +- VPN sessions; +- firewall counters; +- pf statistics; +- NAT counters; +- DNS statistics; +- Suricata summary, если установлен; +- Unbound summary, если установлен. + +## Ограничение безопасности + +pfSense mode не должен менять правила firewall, NAT, DNS, VPN или маршрутизацию. В рамках v0.3 это только наблюдение. diff --git a/docs/AGENT_DEPLOYMENT_RU.md b/docs/AGENT_DEPLOYMENT_RU.md new file mode 100644 index 0000000..a6a64c0 --- /dev/null +++ b/docs/AGENT_DEPLOYMENT_RU.md @@ -0,0 +1,91 @@ +# AWatch-rus Agent Deployment + +## Конфигурация + +Пример: + +```toml +server_url = "https://awatch.local/api/telemetry" +api_key = "change-me" +collect_interval_seconds = 60 +role = "workstation" + +enable_processes = true +enable_network = true +enable_security_events = true +enable_workforce_activity = true + +spool_dir = "/var/lib/awatch-agent/spool" +timeout_seconds = 10 +retry_attempts = 3 +``` + +Linux/BSD путь по умолчанию: + +```text +/etc/awatch-agent/awatch-agent.toml +``` + +Windows путь по умолчанию: + +```text +C:\ProgramData\AWatch\agent\awatch-agent.toml +``` + +## Запуск + +Однократный сбор и печать JSON: + +```bash +awatch-agent-rs --once --print-json +``` + +Однократная отправка: + +```bash +awatch-agent-rs --once --config /etc/awatch-agent/awatch-agent.toml +``` + +Проверка очереди: + +```bash +awatch-agent-rs --spool-health +``` + +Отправка накопленной очереди: + +```bash +awatch-agent-rs --flush-spool +``` + +## Server API + +Портал принимает телеметрию через: + +```text +POST /api/telemetry +``` + +Требования: + +- JSON object; +- заголовок `x-api-key` или `Authorization: Bearer`; +- обязательные поля `TelemetryRecord`; +- в v0.3 хранение является prototype file-backed JSONL. + +Серверные переменные: + +```text +DETMIR_PORTAL_TELEMETRY_API_KEY +DETMIR_PORTAL_TELEMETRY_STORE_PATH +``` + +По умолчанию `change-me` не авторизует прием телеметрии. Для пилота ключ нужно задать явно. + +## Проверка после установки + +1. Запустить агент с `--once --print-json`. +2. Проверить наличие `TelemetryRecord` и `collector_version`. +3. Запустить агент с реальным `server_url`. +4. Проверить HTTP-ответ `{ "ok": true, "stored": "file-backed-jsonl" }`. +5. Проверить отсутствие файлов в spool после успешной отправки. diff --git a/docs/PFSENSE_INTEGRATION_RU.md b/docs/PFSENSE_INTEGRATION_RU.md new file mode 100644 index 0000000..a48dc7f --- /dev/null +++ b/docs/PFSENSE_INTEGRATION_RU.md @@ -0,0 +1,35 @@ +# pfSense Integration + +pfSense рассматривается как интеграционный слой сетевого периметра, а не обязательная часть продукта. + +## Режим v0.3 + +- read-only; +- без изменения правил; +- без автоматического карантина; +- без управления маршрутизацией; +- без зависимости портала от pfSense. + +## Место в архитектуре + +```text +pfSense / firewall telemetry + | +awatch-agent-rs --role firewall + | +POST /api/telemetry + | +Workforce/UEBA risk context +``` + +## Коммерческая ценность + +Интеграция позволяет объяснять риски не только по активности рабочего места, но и по сетевому контексту: unusual destinations, VPN sessions, gateway status, proxy/DNS signals. + +## Не реализуется в v0.3 + +- NAC; +- SOAR-автоматизация; +- блокировка VLAN; +- изменение firewall rules; +- управление VPN-доступом. diff --git a/docs/PORTAL_RU.md b/docs/PORTAL_RU.md new file mode 100644 index 0000000..5fdcc56 --- /dev/null +++ b/docs/PORTAL_RU.md @@ -0,0 +1,69 @@ +# AWatch-rus Portal + +## Назначение + +Портал является рабочим кабинетом для трех ролей: + +- руководитель: пульс организации, активность, подразделения, риски; +- ИБ: риск-сигналы, события безопасности, evidence; +- расследователь: карточка инцидента, причины, проверяемые источники, отчет. + +## Главные экраны + +- `Обзор` - Executive Dashboard. +- `Сотрудники` - карточки сотрудников и объяснение индекса. +- `Подразделения` - сравнение групп, тренды и ответственные. +- `Риски` - read-only контроль безопасности. +- `Расследования` - инциденты, evidence и автоматическая read-only карточка расследования. +- `Сетевой периметр` - read-only контекст внешних сетевых сигналов. +- `Отчеты` - Markdown/PDF/JSON и управленческий текст. +- `Настройки` - read-only параметры расчета. + +## Executive Dashboard + +Показывает: + +- сотрудников в работе; +- средний индекс активности; +- WARN/FAIL подразделения; +- критические риски; +- недельный тренд; +- топ-5 лучших и проблемных подразделений; +- Heat Map; +- блок `Требует внимания`. + +## Risk -> Investigation + +Кнопка `Открыть расследование` переводит пользователя в read-only карточку. + +Карточка содержит: + +- `incident_id`; +- `risk_id`; +- `department`; +- `owner`; +- `activity_index`; +- `deviation`; +- `status`; +- `summary`; +- `why_it_is_risk`; +- `what_to_check`; +- `recommended_actions`; +- `evidence`; +- `generated_at`. + +Предупреждение обязательно: + +```text +Расследование сформировано автоматически. Решение принимает ответственный сотрудник. +``` + +## Проверка + +Smoke-тест: + +```bash +node scripts/detmir-portal-tabs-smoke.mjs +``` + +Тест проверяет все вкладки, read-only настройки и переход Risk -> Investigation. diff --git a/docs/WORKFORCE_UEBA_MODEL_RU.md b/docs/WORKFORCE_UEBA_MODEL_RU.md new file mode 100644 index 0000000..0c6e160 --- /dev/null +++ b/docs/WORKFORCE_UEBA_MODEL_RU.md @@ -0,0 +1,79 @@ +# Workforce / UEBA Model + +## Позиционирование + +AWatch-rus v0.3 использует Workforce-first подход: + +```text +Телеметрия -> Активность -> Риск -> Расследование -> Отчет +``` + +Это не сертифицированная СЗИ, не промышленная SIEM и не полнофункциональная DLP. Корректная формулировка для v0.3: + +```text +UEBA-compatible rule-based risk scoring v1. +``` + +## Workforce scoring + +Расчетные поля: + +- `activity_index`; +- `active_today`; +- `department_activity_index`; +- `owner_activity_index`; +- `trend_status`; +- `anomaly_status`; +- `risk_level`. + +Базовая proxy-формула: + +```text +Индекс активности = active_seconds / planned_seconds * 100 +``` + +Взвешенный индекс использует роли и веса приложений, если задан workforce policy. + +## Статусы + +- `OK` - состояние в норме; +- `WARN` - требуется внимание; +- `FAIL` - требуется действие. + +Каждый риск обязан объяснять: + +- что произошло; +- почему это риск; +- что проверить; +- рекомендуемое действие. + +## UEBA risk v1 + +Риск рассчитывается по правилам, а не по скрытой ML-модели. + +Поля объяснимости: + +- `confidence`; +- `risk_sources`; +- `baseline_status`; +- `policy_version`; +- `calculated_from`; +- `baseline_window_days`; +- `user_baseline_available`; +- `department_baseline_available`; +- `deviation_score`; +- `baseline_samples`. + +## Evidence + +Evidence повышает достоверность вывода, но само по себе не является фактором риска. Риск должен исходить из события, отклонения, тренда или правила. + +## Не реализуется + +- перехват содержимого документов; +- запись экрана; +- кейлоггер; +- скрытый агент; +- контентный DLP-анализ; +- автоматическое редактирование инцидентов; +- автоматическое наказание или блокировка сотрудника. diff --git a/scripts/detmir-portal-tabs-smoke.mjs b/scripts/detmir-portal-tabs-smoke.mjs index 74d81fa..1eea062 100644 --- a/scripts/detmir-portal-tabs-smoke.mjs +++ b/scripts/detmir-portal-tabs-smoke.mjs @@ -163,9 +163,14 @@ async function main() { { timeout }, ); const activeTab = await page.locator(".tab.is-active").innerText({ timeout }); + const incidentText = await page.locator("#content").innerText({ timeout }); checks.push({ name: "risk_open_investigation_readonly_navigation", - ok: activeTab.trim() === "Расследования", + ok: activeTab.trim() === "Расследования" + && containsText(incidentText, "Расследование сформировано автоматически") + && containsText(incidentText, "incident_id") + && containsText(incidentText, "risk_id") + && containsText(incidentText, "evidence"), buttons: investigationButtons, }); } else {