From cd25edc4f371a089b0cab36b46d04dafc6eb7384 Mon Sep 17 00:00:00 2001 From: igor04091968 Date: Thu, 9 Jul 2026 23:38:57 +0300 Subject: [PATCH] Fix DetMir light DLP readiness and collector guard --- .../crates/aw-windows-telemetry/src/main.rs | 369 +++++++++++++++++- adk-rust/crates/detmir-readiness/src/main.rs | 73 +++- scripts/full-validation-orchestrator.mjs | 17 +- 3 files changed, 440 insertions(+), 19 deletions(-) diff --git a/adk-rust/crates/aw-windows-telemetry/src/main.rs b/adk-rust/crates/aw-windows-telemetry/src/main.rs index 94a9e0e..2fa6889 100644 --- a/adk-rust/crates/aw-windows-telemetry/src/main.rs +++ b/adk-rust/crates/aw-windows-telemetry/src/main.rs @@ -3985,9 +3985,47 @@ fn run_collector_guard_cycle(args: &CollectorGuard, runtime: &mut GuardRuntime) let server_port = json_i64(&config, &["server", "port"]).unwrap_or(5600); let api_base = format!("{server_scheme}://{server_host}:{server_port}/api/0"); let mut process_snapshot = collect_process_snapshot(); + let session_snapshot = collect_session_snapshot(); + let live_session_ids = live_session_ids(&session_snapshot.sessions); let mut problems = Vec::new(); let mut actions = Vec::new(); + let non_live_stop_plan = + non_live_session_collectors(&process_snapshot.processes, &live_session_ids); + if !non_live_stop_plan.is_empty() { + if args.mode == "enforce" { + for process in &non_live_stop_plan { + let ok = process.pid.is_some_and(terminate_process); + actions.push(json!({ + "action": "stop-non-live-session-collector", + "kind": session_scoped_collector_kind(process).unwrap_or("unknown"), + "sessionId": process.session_id, + "pid": process.pid, + "applied": true, + "ok": ok + })); + if !ok { + problems.push(format!( + "failed to stop collector pid {:?} in non-live session {:?}", + process.pid, process.session_id + )); + } + } + process_snapshot = collect_process_snapshot(); + } else { + for process in &non_live_stop_plan { + actions.push(json!({ + "action": "stop-non-live-session-collector", + "kind": session_scoped_collector_kind(process).unwrap_or("unknown"), + "sessionId": process.session_id, + "pid": process.pid, + "applied": false, + "mode": "shadow" + })); + } + } + } + let duplicate_plan = duplicate_legacy_collectors(&process_snapshot.processes); if !duplicate_plan.is_empty() { if args.mode == "enforce" { @@ -4088,7 +4126,7 @@ fn run_collector_guard_cycle(args: &CollectorGuard, runtime: &mut GuardRuntime) } let task_defs = guard_task_definitions(&config); - let missing_fileops_sessions = + let mut missing_fileops_sessions = if file_ops_enabled && file_ops_mode.eq_ignore_ascii_case("rust_primary") { missing_rust_collector_sessions( &process_snapshot.processes, @@ -4098,10 +4136,13 @@ fn run_collector_guard_cycle(args: &CollectorGuard, runtime: &mut GuardRuntime) } else { Vec::new() }; - let launch_needed = interactive_stale || !missing_fileops_sessions.is_empty(); + missing_fileops_sessions.retain(|session_id| live_session_ids.contains(session_id)); + let has_live_sessions = !live_session_ids.is_empty(); + let effective_interactive_stale = interactive_stale && has_live_sessions; + let launch_needed = effective_interactive_stale || !missing_fileops_sessions.is_empty(); if launch_needed { let active_legacy_collectors = active_legacy_collector_count(&process_snapshot.processes); - if interactive_stale + if effective_interactive_stale && missing_fileops_sessions.is_empty() && active_legacy_collectors > 0 && process_snapshot.command_line_query_ok @@ -4130,6 +4171,16 @@ fn run_collector_guard_cycle(args: &CollectorGuard, runtime: &mut GuardRuntime) problems.push(format!("refuse non-allowlisted task {}", task.task_name)); continue; } + if !task_has_live_session(&task.user_id, &session_snapshot.sessions) { + actions.push(json!({ + "action": "run-task", + "target": task.task_name, + "applied": false, + "reason": "no-live-session-for-user", + "userId": task.user_id + })); + continue; + } let key = format!("task:{}", task.task_name); let allowed = runtime.action_allowed( &key, @@ -4168,7 +4219,8 @@ fn run_collector_guard_cycle(args: &CollectorGuard, runtime: &mut GuardRuntime) } } - let status = if problems.is_empty() && (args.mode == "enforce" || !interactive_stale) { + let status = if problems.is_empty() && (args.mode == "enforce" || !effective_interactive_stale) + { "ok" } else { "warn" @@ -4187,12 +4239,25 @@ fn run_collector_guard_cycle(args: &CollectorGuard, runtime: &mut GuardRuntime) "rustWorktimeAgentRunning": rust_agent_running, "powerShellRuntimeByKind": power_shell_by_kind }, + "sessions": { + "queryOk": session_snapshot.query_ok, + "source": session_snapshot.source, + "error": session_snapshot.error, + "liveSessionIds": live_session_ids.iter().copied().collect::>(), + "records": session_snapshot.sessions.iter().map(|session| json!({ + "sessionId": session.session_id, + "userName": session.user_name, + "state": session.state, + "isLive": session.is_live + })).collect::>() + }, "buckets": bucket_checks, "tasks": task_defs.iter().map(|task| json!({ "taskName": task.task_name, "userId": task.user_id })).collect::>(), "interactiveStale": interactive_stale, + "effectiveInteractiveStale": effective_interactive_stale, "fileOperationsPresence": { "enabled": file_ops_enabled, "mode": file_ops_mode, @@ -4285,6 +4350,51 @@ fn guard_task_definitions(config: &Value) -> Vec { out } +fn live_session_ids(sessions: &[SessionInfo]) -> HashSet { + sessions + .iter() + .filter(|session| session.is_live) + .map(|session| session.session_id) + .collect() +} + +fn task_has_live_session(user_id: &str, sessions: &[SessionInfo]) -> bool { + let candidates = user_candidates(user_id); + sessions.iter().any(|session| { + session.is_live + && session + .user_name + .as_deref() + .is_some_and(|user| user_matches_candidates(user, &candidates)) + }) +} + +fn user_candidates(user_id: &str) -> HashSet { + let normalized = user_id.trim().to_ascii_lowercase(); + let mut out = HashSet::new(); + if normalized.is_empty() { + return out; + } + out.insert(normalized.clone()); + if let Some((_, leaf)) = normalized.rsplit_once('\\') { + out.insert(leaf.to_string()); + } + out +} + +fn user_matches_candidates(user_name: &str, candidates: &HashSet) -> bool { + let normalized = user_name.trim().to_ascii_lowercase(); + if normalized.is_empty() { + return false; + } + if candidates.contains(&normalized) { + return true; + } + normalized + .rsplit_once('\\') + .is_some_and(|(_, leaf)| candidates.contains(leaf)) +} + fn run_scheduled_task(task_name: &str) -> bool { if !task_name.starts_with("ActivityWatch Launch ") { return false; @@ -4296,6 +4406,54 @@ fn run_scheduled_task(task_name: &str) -> bool { .unwrap_or(false) } +fn non_live_session_collectors<'a>( + processes: &'a [ProcessInfo], + live_session_ids: &HashSet, +) -> Vec<&'a ProcessInfo> { + processes + .iter() + .filter(|process| { + let Some(session_id) = process.session_id else { + return false; + }; + session_id > 0 + && !live_session_ids.contains(&session_id) + && session_scoped_collector_kind(process).is_some() + && process.pid.is_some() + }) + .collect() +} + +fn session_scoped_collector_kind(process: &ProcessInfo) -> Option<&'static str> { + if let Some(kind) = legacy_collector_kind(process) { + return Some(kind); + } + let name = process.name.as_deref().unwrap_or_default(); + if name.eq_ignore_ascii_case("aw-watcher-afk.exe") { + return Some("afk"); + } + if name.eq_ignore_ascii_case("aw-watcher-window.exe") { + return Some("window"); + } + if name.eq_ignore_ascii_case("aw-windows-telemetry.exe") { + let command_line = process + .command_line + .as_deref() + .unwrap_or_default() + .to_ascii_lowercase(); + if command_line.contains("browser-domains-collector") { + return Some("browser"); + } + if command_line.contains("dlp-endpoint-collector") { + return Some("dlp_endpoint"); + } + if command_line.contains("file-operations-collector") { + return Some("fileops"); + } + } + None +} + #[derive(Debug, Clone, Serialize)] struct LegacyCollectorDuplicate { kind: &'static str, @@ -5014,6 +5172,22 @@ struct ProcessInfo { command_line: Option, } +#[derive(Debug, Default)] +struct SessionSnapshot { + query_ok: bool, + source: String, + error: Option, + sessions: Vec, +} + +#[derive(Debug, Clone)] +struct SessionInfo { + session_id: u32, + user_name: Option, + state: String, + is_live: bool, +} + fn validate_files(paths: &[PathBuf]) -> Value { let mut list = Vec::new(); let mut missing = Vec::new(); @@ -5205,6 +5379,118 @@ fn collect_process_snapshot() -> ProcessSnapshot { } } +fn collect_session_snapshot() -> SessionSnapshot { + if let Some(raw) = command_output_utf16le("cmd", &["/U", "/C", "query user"]) + .or_else(|| command_output_utf16le("cmd", &["/U", "/C", "quser"])) + { + let sessions = parse_query_user_sessions(&raw); + if !sessions.is_empty() { + return SessionSnapshot { + query_ok: true, + source: "quser_utf16".to_string(), + error: None, + sessions, + }; + } + } + + if let Some(raw) = command_output_lossy_combined("cmd", &["/C", "query user"]) + .or_else(|| command_output_lossy_combined("cmd", &["/C", "quser"])) + { + let sessions = parse_query_user_sessions(&raw); + if !sessions.is_empty() { + return SessionSnapshot { + query_ok: true, + source: "quser_lossy".to_string(), + error: None, + sessions, + }; + } + } + + SessionSnapshot { + query_ok: false, + source: "unavailable".to_string(), + error: Some("query user and quser returned no sessions".to_string()), + sessions: Vec::new(), + } +} + +fn command_output_utf16le(program: &str, args: &[&str]) -> Option { + let output = Command::new(program).args(args).output().ok()?; + if !output.status.success() { + return None; + } + let mut bytes = output.stdout; + bytes.extend_from_slice(&output.stderr); + if bytes.is_empty() { + return None; + } + let words = bytes + .chunks_exact(2) + .map(|chunk| u16::from_le_bytes([chunk[0], chunk[1]])) + .collect::>(); + String::from_utf16(&words) + .ok() + .map(|value| value.trim().to_string()) + .filter(|value| !value.is_empty()) +} + +fn command_output_lossy_combined(program: &str, args: &[&str]) -> Option { + let output = Command::new(program).args(args).output().ok()?; + if !output.status.success() { + return None; + } + let mut bytes = output.stdout; + bytes.extend_from_slice(&output.stderr); + Some(String::from_utf8_lossy(&bytes).trim().to_string()).filter(|value| !value.is_empty()) +} + +fn parse_query_user_sessions(raw: &str) -> Vec { + raw.lines() + .filter_map(parse_query_user_line) + .collect::>() +} + +fn parse_query_user_line(line: &str) -> Option { + let cleaned = line.trim().trim_start_matches('>').trim(); + if cleaned.is_empty() + || cleaned.to_ascii_lowercase().starts_with("username") + || cleaned.starts_with("ПОЛЬЗОВАТЕЛЬ") + { + return None; + } + let parts = cleaned.split_whitespace().collect::>(); + if parts.len() < 3 { + return None; + } + let username = parts.first()?.trim(); + let (session_id, state, has_session_name) = + if parts.get(1)?.chars().all(|ch| ch.is_ascii_digit()) { + (*parts.get(1)?, *parts.get(2)?, false) + } else { + (*parts.get(2)?, *parts.get(3).unwrap_or(&"Unknown"), true) + }; + let session_id = session_id.parse::().ok()?; + Some(SessionInfo { + session_id, + user_name: (!username.is_empty()).then(|| username.to_string()), + state: state.to_string(), + is_live: session_state_is_live(state) + || (has_session_name && !session_state_is_disconnected(state)), + }) +} + +fn session_state_is_live(state: &str) -> bool { + let lower = state.to_lowercase(); + lower.contains("active") || lower.contains("conn") || lower.contains("актив") +} + +fn session_state_is_disconnected(state: &str) -> bool { + let lower = state.to_lowercase(); + lower.contains("disc") || lower.contains("диск") +} + #[cfg(windows)] fn collect_native_process_snapshot() -> Option { use std::mem::{MaybeUninit, size_of}; @@ -6371,6 +6657,81 @@ SERVICE_NAME: AWatchRusCollectorGuard ); } + #[test] + fn collector_guard_parses_live_and_disconnected_sessions() { + let sessions = parse_query_user_sessions( + r#" + USERNAME SESSIONNAME ID STATE IDLE TIME LOGON TIME + user1 rdp-tcp#5 5 Active none 09.07.2026 10:00 + user2 2 Disc 3:14 07.07.2026 8:34 +"#, + ); + + assert_eq!(sessions.len(), 2); + assert_eq!(live_session_ids(&sessions), HashSet::from([5])); + assert!(task_has_live_session(r"SHARKON2025\user1", &sessions)); + assert!(!task_has_live_session(r"SHARKON2025\user2", &sessions)); + } + + #[test] + fn collector_guard_treats_named_rdp_session_as_live_when_state_is_localized() { + let sessions = parse_query_user_sessions( + r#" + USERNAME SESSIONNAME ID STATE IDLE TIME LOGON TIME + user1 rdp-tcp#12 12 ????? none 09.07.2026 10:00 + user2 2 ????? 3:14 07.07.2026 8:34 +"#, + ); + + assert_eq!(sessions.len(), 2); + assert_eq!(live_session_ids(&sessions), HashSet::from([12])); + assert!(task_has_live_session(r"SHARKON2025\user1", &sessions)); + assert!(!task_has_live_session(r"SHARKON2025\user2", &sessions)); + } + + #[test] + fn collector_guard_stops_session_collectors_outside_live_sessions() { + let processes = vec![ + ProcessInfo { + name: Some("aw-watcher-afk.exe".to_string()), + pid: Some(100), + session_id: Some(2), + created_unix_seconds: Some(10), + command_line: Some("aw-watcher-afk.exe --host 10.10.10.13".to_string()), + }, + ProcessInfo { + name: Some("aw-windows-telemetry.exe".to_string()), + pid: Some(101), + session_id: Some(2), + created_unix_seconds: Some(11), + command_line: Some( + "aw-windows-telemetry.exe browser-domains-collector --mode enforce".to_string(), + ), + }, + ProcessInfo { + name: Some("aw-watcher-window.exe".to_string()), + pid: Some(200), + session_id: Some(5), + created_unix_seconds: Some(20), + command_line: Some("aw-watcher-window.exe --host 10.10.10.13".to_string()), + }, + ProcessInfo { + name: Some("awatch-agent-rs.exe".to_string()), + pid: Some(300), + session_id: Some(0), + created_unix_seconds: Some(30), + command_line: Some("awatch-agent-rs.exe --config x".to_string()), + }, + ]; + let stop_plan = non_live_session_collectors(&processes, &HashSet::from([5])); + let pids = stop_plan + .iter() + .filter_map(|process| process.pid) + .collect::>(); + + assert_eq!(pids, vec![100, 101]); + } + #[test] fn validate_deployment_parses_wmic_process_csv() { let csv = br#"Node,CommandLine,Name,ProcessId,SessionId diff --git a/adk-rust/crates/detmir-readiness/src/main.rs b/adk-rust/crates/detmir-readiness/src/main.rs index 7d618e2..158a655 100644 --- a/adk-rust/crates/detmir-readiness/src/main.rs +++ b/adk-rust/crates/detmir-readiness/src/main.rs @@ -221,15 +221,17 @@ fn run(cli: &Cli) -> Result { let worktime = influx_config(&aw_env, "AW_WORKTIME_INFLUX"); let dlp = influx_config(&aw_env, "AW_DLP_INFLUX"); let dlp_enabled = env_bool(&aw_env, "AW_DLP_ENABLED", true); + let dlp_profile = env_value(&aw_env, "AW_DLP_PROFILE", "full"); + let dlp_influx_required = dlp_influx_required_for_profile(dlp_enabled, &dlp_profile); checks.push(check_influx_env(&worktime, cli.allow_disabled_influx)); - if dlp_enabled { + if dlp_influx_required { checks.push(check_influx_env(&dlp, cli.allow_disabled_influx)); } else { - checks.push(warn( + checks.push(ok( "env:AW_DLP_INFLUX", - "DLP Influx runtime disabled by AW_DLP_ENABLED=false", - json!({"enabled": false, "mode": "disabled"}), + "DLP Influx runtime is not required by the current DLP profile", + json!({"enabled": dlp_enabled, "profile": dlp_profile.as_str()}), )); } @@ -238,7 +240,7 @@ fn run(cli: &Cli) -> Result { } else { checks.extend(check_systemd_services(&systemd_services_for_mode( &cli.systemd_services, - dlp_enabled, + dlp_influx_required, ))); } @@ -250,13 +252,13 @@ fn run(cli: &Cli) -> Result { )); } else { checks.push(check_influx_write(&client, "worktime", &worktime)); - if dlp_enabled { + if dlp_influx_required { checks.push(check_influx_write(&client, "dlp", &dlp)); } else { - checks.push(warn( + checks.push(ok( "influx:write:dlp", - "DLP write probe skipped because DLP is disabled", - json!({"enabled": false, "mode": "disabled"}), + "DLP write probe skipped because DLP Influx is not required by the current profile", + json!({"enabled": dlp_enabled, "profile": dlp_profile.as_str()}), )); } } @@ -292,7 +294,7 @@ fn run(cli: &Cli) -> Result { git_commit: cli.git_commit.clone(), counts, checks, - limitations: build_limitations(cli, dlp_enabled), + limitations: build_limitations(cli, dlp_enabled, dlp_influx_required), }) } @@ -360,9 +362,19 @@ fn split_csv(value: &str) -> Vec { .collect() } -fn systemd_services_for_mode(csv: &str, dlp_enabled: bool) -> String { +fn dlp_influx_required_for_profile(dlp_enabled: bool, profile: &str) -> bool { + if !dlp_enabled { + return false; + } + !matches!( + profile.trim().to_ascii_lowercase().as_str(), + "light" | "core_only" | "disabled" | "off" | "on_demand" + ) +} + +fn systemd_services_for_mode(csv: &str, dlp_influx_required: bool) -> String { let mut services = split_csv(csv); - if dlp_enabled { + if dlp_influx_required { for service in split_csv(DEFAULT_DLP_SYSTEMD_SERVICES) { if !services.iter().any(|item| item == &service) { services.push(service); @@ -384,7 +396,7 @@ fn hostname() -> String { .unwrap_or_else(|| "unknown".to_string()) } -fn build_limitations(cli: &Cli, dlp_enabled: bool) -> Vec { +fn build_limitations(cli: &Cli, dlp_enabled: bool, dlp_influx_required: bool) -> Vec { let mut limitations = Vec::new(); limitations.push( "Проверка подтверждает состояние runtime на момент формирования акта и не заменяет аудит конфигурации, нагрузочное тестирование или приемочные испытания заказчика.".to_string(), @@ -417,6 +429,10 @@ fn build_limitations(cli: &Cli, dlp_enabled: bool) -> Vec { limitations.push( "DLP runtime отключен штатно через AW_DLP_ENABLED=false; readiness не считает DLP services/timers и DLP Influx write обязательными.".to_string(), ); + } else if !dlp_influx_required { + limitations.push( + "DLP runtime включен в лёгком профиле; readiness не считает DLP Influx timer и DLP Influx write обязательными.".to_string(), + ); } limitations } @@ -1348,6 +1364,37 @@ mod tests { assert_eq!(counts.fail, 1); } + #[test] + fn dlp_influx_is_required_only_for_full_profile() { + for profile in ["full", "enabled", "on"] { + assert!( + dlp_influx_required_for_profile(true, profile), + "{profile} should require DLP Influx" + ); + } + + for profile in ["light", "core_only", "disabled", "off", "on_demand"] { + assert!( + !dlp_influx_required_for_profile(true, profile), + "{profile} should not require DLP Influx" + ); + } + + assert!(!dlp_influx_required_for_profile(false, "full")); + assert!(dlp_influx_required_for_profile(true, "unexpected")); + } + + #[test] + fn light_profile_excludes_dlp_influx_timer() { + let services = systemd_services_for_mode(DEFAULT_SYSTEMD_SERVICES, false); + assert!(services.contains("activitywatch-server")); + assert!(services.contains("aw-worktime-influx-exporter.timer")); + assert!(!services.contains("aw-dlp-influx-exporter.timer")); + + let services = systemd_services_for_mode(DEFAULT_SYSTEMD_SERVICES, true); + assert!(services.contains("aw-dlp-influx-exporter.timer")); + } + #[test] fn renders_readiness_act_without_secrets() { let checks = vec![ok( diff --git a/scripts/full-validation-orchestrator.mjs b/scripts/full-validation-orchestrator.mjs index ecdb902..f93a52b 100644 --- a/scripts/full-validation-orchestrator.mjs +++ b/scripts/full-validation-orchestrator.mjs @@ -183,6 +183,16 @@ function writeText(file, data) { fs.writeFileSync(file, data, "utf8"); } +function copyFileCompat(source, target) { + mkdirp(path.dirname(target)); + try { + fs.copyFileSync(source, target); + } catch (error) { + if (!["EPERM", "ENOSYS", "EXDEV"].includes(error?.code)) throw error; + fs.writeFileSync(target, fs.readFileSync(source)); + } +} + function runCommand({ id, category, command, args = [], cwd = root, timeoutSeconds, env = {}, optional = false }) { const started = Date.now(); const cmdline = [command, ...args].join(" "); @@ -335,7 +345,10 @@ async function trackedFiles() { } const result = await spawnSyncText("git", ["ls-files", "-z"], root, 60); if (result.status !== 0) return []; - return result.stdout.split("\0").filter(Boolean); + return result.stdout + .split("\0") + .filter(Boolean) + .filter((file) => fs.existsSync(path.join(root, file))); } function walk(dir) { @@ -1347,7 +1360,7 @@ async function executeValidation(args) { mkdirp(latestDir); for (const [name, file] of Object.entries(outputs)) { const latestName = path.basename(file); - fs.copyFileSync(file, path.join(latestDir, latestName)); + copyFileCompat(file, path.join(latestDir, latestName)); report.outputs[`latest_${name}`] = normalizeRelative(path.join(latestDir, latestName)); } writeJson(outputs.validation_report_json, report);