awatch-rus/AWatch-rus
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(portal): split hardening and kpi modules

Browse Source
This commit is contained in:
igor04091968
2026-06-07 14:36:42 +03:00
parent 2fb3271558
commit 0b0bab61bb
10 changed files with 1553 additions and 1300 deletions
Download Patch File Download Diff File Expand all files Collapse all files
adk-rust/crates/detmir-portal/src/main.rs
+12 -1300
View File
File diff suppressed because it is too large Load Diff
adk-rust/crates/detmir-portal/src/production/health.rs
+10
View File
@@ -0,0 +1,10 @@
use serde_json::{Value, json};
use crate::now;
pub(crate) fn build_healthz() -> Value {
json!({
"status": "ok",
"generated_at_utc": now(),
})
}
adk-rust/crates/detmir-portal/src/production/limits.rs
+297
View File
@@ -0,0 +1,297 @@
use std::collections::BTreeSet;
use anyhow::{Result, anyhow};
use chrono::NaiveDate;
use serde_json::{Value, json};
use tiny_http::StatusCode;
use crate::{
Cli, MAX_ALLOWED_PAGE_SIZE, MAX_ALLOWED_REPORT_DATE_RANGE_DAYS, MAX_ALLOWED_REQUEST_BODY_BYTES,
MAX_ALLOWED_REQUEST_TIMEOUT_SECONDS, query_param,
};
#[derive(Debug)]
pub(crate) struct ApiLimitError {
pub(crate) status: StatusCode,
pub(crate) payload: Value,
}
pub(crate) fn validate_portal_config(args: &Cli) -> Result<()> {
let (host, port) = args
.bind
.rsplit_once(':')
.ok_or_else(|| anyhow!("invalid config bind: expected host:port"))?;
if host.trim().is_empty() {
return Err(anyhow!("invalid config host: value is empty"));
}
let port = port
.parse::<u16>()
.map_err(|_| anyhow!("invalid config port: expected 1..65535"))?;
if port == 0 {
return Err(anyhow!("invalid config port: expected 1..65535"));
}
if args.max_page_size == 0 || args.max_page_size > MAX_ALLOWED_PAGE_SIZE {
return Err(anyhow!(
"invalid config max_page_size: expected 1..={MAX_ALLOWED_PAGE_SIZE}"
));
}
if args.default_page_size == 0 || args.default_page_size > args.max_page_size {
return Err(anyhow!(
"invalid config default_page_size: expected 1..=max_page_size"
));
}
if args.max_report_date_range_days <= 0
|| args.max_report_date_range_days > MAX_ALLOWED_REPORT_DATE_RANGE_DAYS
{
return Err(anyhow!(
"invalid config max_report_date_range_days: expected 1..={MAX_ALLOWED_REPORT_DATE_RANGE_DAYS}"
));
}
if args.request_timeout_seconds == 0
|| args.request_timeout_seconds > MAX_ALLOWED_REQUEST_TIMEOUT_SECONDS
{
return Err(anyhow!(
"invalid config request_timeout_seconds: expected 1..={MAX_ALLOWED_REQUEST_TIMEOUT_SECONDS}"
));
}
if args.timeout_seconds == 0 || args.timeout_seconds > MAX_ALLOWED_REQUEST_TIMEOUT_SECONDS {
return Err(anyhow!(
"invalid config timeout_seconds: expected 1..={MAX_ALLOWED_REQUEST_TIMEOUT_SECONDS}"
));
}
if args.max_request_body_bytes < 1024
|| args.max_request_body_bytes > MAX_ALLOWED_REQUEST_BODY_BYTES
{
return Err(anyhow!(
"invalid config max_request_body_bytes: expected 1024..={MAX_ALLOWED_REQUEST_BODY_BYTES}"
));
}
if !is_safe_environment_name(&args.environment) {
return Err(anyhow!(
"invalid config environment: use 1..32 chars from A-Z, a-z, 0-9, _, -"
));
}
let modules = enabled_modules(&args.enabled_modules);
if modules.is_empty() {
return Err(anyhow!(
"invalid config enabled_modules: no modules enabled"
));
}
let allowed = [
"executive",
"workforce",
"security",
"forensics",
"admin",
"ueba",
"pfsense",
"reports",
];
for module in modules {
if !allowed.contains(&module.as_str()) {
return Err(anyhow!(
"invalid config enabled_modules: unsupported module {module}"
));
}
}
Ok(())
}
fn is_safe_environment_name(value: &str) -> bool {
let value = value.trim();
!value.is_empty()
&& value.len() <= 32
&& value
.chars()
.all(|ch| ch.is_ascii_alphanumeric() || matches!(ch, '_' | '-'))
}
fn enabled_modules(value: &str) -> BTreeSet<String> {
value
.split(',')
.map(|item| item.trim().to_ascii_lowercase())
.filter(|item| !item.is_empty())
.collect()
}
pub(crate) fn is_limited_api_route(path: &str) -> bool {
matches!(
path,
"/api/reports"
| "/api/executive"
| "/api/workforce"
| "/api/security"
| "/api/forensics"
| "/api/ueba"
| "/api/pfsense"
| "/api/workforce/kpi/explain"
)
}
pub(crate) fn validate_api_query_limits(
url: &str,
args: &Cli,
) -> std::result::Result<(), ApiLimitError> {
for key in ["page_size", "limit"] {
if let Some(value) = query_param(url, key) {
let parsed = value.parse::<u32>().ok();
if parsed.is_none_or(|page_size| page_size == 0 || page_size > args.max_page_size) {
return Err(api_limit_error(
StatusCode(400),
"invalid_page_size",
&format!("{key} must be between 1 and {}", args.max_page_size),
));
}
}
}
for (from_key, to_key) in [("date_from", "date_to"), ("from", "to"), ("start", "end")] {
let Some(from) = query_param(url, from_key).and_then(|value| parse_query_date(&value))
else {
continue;
};
let Some(to) = query_param(url, to_key).and_then(|value| parse_query_date(&value)) else {
continue;
};
let days = (to - from).num_days().abs() + 1;
if days > args.max_report_date_range_days {
return Err(api_limit_error(
StatusCode(400),
"report_range_too_large",
&format!(
"report date range must be <= {} days",
args.max_report_date_range_days
),
));
}
}
Ok(())
}
pub(crate) fn parse_query_date(value: &str) -> Option<NaiveDate> {
let date = value.split('T').next().unwrap_or(value);
NaiveDate::parse_from_str(date, "%Y-%m-%d").ok()
}
fn api_limit_error(status: StatusCode, code: &str, message: &str) -> ApiLimitError {
ApiLimitError {
status,
payload: json!({
"ok": false,
"error_code": code,
"message": message,
}),
}
}
#[cfg(test)]
mod tests {
use std::path::Path;
use super::*;
use crate::{
DEFAULT_MAX_PAGE_SIZE, DEFAULT_MAX_REPORT_DATE_RANGE_DAYS, DEFAULT_MAX_REQUEST_BODY_BYTES,
DEFAULT_PAGE_SIZE, DEFAULT_REQUEST_TIMEOUT_SECONDS, DEFAULT_SLOW_REQUEST_LOG_MS,
};
fn test_cli(dir: &Path) -> Cli {
Cli {
bind: "127.0.0.1:8720".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.join("workforce-policy.json"),
ueba_policy_path: dir.join("ueba-policy.yaml"),
timeout_seconds: 1,
max_page_size: DEFAULT_MAX_PAGE_SIZE,
default_page_size: DEFAULT_PAGE_SIZE,
max_report_date_range_days: DEFAULT_MAX_REPORT_DATE_RANGE_DAYS,
request_timeout_seconds: DEFAULT_REQUEST_TIMEOUT_SECONDS,
max_request_body_bytes: DEFAULT_MAX_REQUEST_BODY_BYTES,
slow_request_log_ms: DEFAULT_SLOW_REQUEST_LOG_MS,
environment: "test".to_string(),
enabled_modules: "executive,workforce,security,forensics,admin".to_string(),
state_dir: dir.join("state"),
dlp_db_path: dir.join("dlp.sqlite"),
evidence_root: dir.to_path_buf(),
readiness_bundle_dir: dir.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.join("telemetry.jsonl"),
expected_nodes_path: dir.join("expected_nodes.json"),
security_events_backend: "disabled".to_string(),
clickhouse_url: "http://127.0.0.1:8123".to_string(),
clickhouse_database: "analytics_1c".to_string(),
clickhouse_user: "default".to_string(),
clickhouse_password: String::new(),
}
}
#[test]
fn config_validation_rejects_bad_values() {
let dir = tempfile::tempdir().unwrap();
let args = test_cli(dir.path());
assert!(validate_portal_config(&args).is_ok());
let mut invalid = args.clone();
invalid.bind = "127.0.0.1:bad".to_string();
assert!(
validate_portal_config(&invalid)
.unwrap_err()
.to_string()
.contains("port")
);
let mut invalid = args.clone();
invalid.max_page_size = 0;
assert!(
validate_portal_config(&invalid)
.unwrap_err()
.to_string()
.contains("max_page_size")
);
let mut invalid = args.clone();
invalid.max_report_date_range_days = 0;
assert!(
validate_portal_config(&invalid)
.unwrap_err()
.to_string()
.contains("max_report_date_range_days")
);
let mut invalid = args.clone();
invalid.request_timeout_seconds = 0;
assert!(
validate_portal_config(&invalid)
.unwrap_err()
.to_string()
.contains("request_timeout_seconds")
);
}
#[test]
fn query_limits_reject_page_size_and_report_range() {
let dir = tempfile::tempdir().unwrap();
let args = test_cli(dir.path());
assert!(validate_api_query_limits("/api/reports?page_size=100", &args).is_ok());
let page_error =
validate_api_query_limits("/api/reports?page_size=999999", &args).unwrap_err();
assert_eq!(page_error.status.0, 400);
assert_eq!(page_error.payload["error_code"], "invalid_page_size");
let range_error = validate_api_query_limits(
"/api/reports?date_from=2026-01-01&date_to=2026-12-31",
&args,
)
.unwrap_err();
assert_eq!(range_error.status.0, 400);
assert_eq!(range_error.payload["error_code"], "report_range_too_large");
}
}
adk-rust/crates/detmir-portal/src/production/logging.rs
+42
View File
@@ -0,0 +1,42 @@
use serde_json::{Value, json};
use tiny_http::StatusCode;
use super::request_context::HttpRequestMetadata;
use crate::now;
pub(crate) fn log_http_request(
metadata: &HttpRequestMetadata,
status: StatusCode,
response_bytes: usize,
) {
let level = if status.0 >= 500 {
"ERROR"
} else if status.0 >= 400 {
"WARN"
} else {
"INFO"
};
let error_code = if status.0 >= 400 {
Value::String(format!("http_{}", status.0))
} else {
Value::Null
};
eprintln!(
"{}",
json!({
"timestamp": now(),
"level": level,
"request_id": &metadata.request_id,
"correlation_id": &metadata.correlation_id,
"method": &metadata.method,
"path": &metadata.path,
"route": &metadata.route,
"status": status.0,
"latency_ms": metadata.latency_ms,
"user_role": &metadata.role,
"module": &metadata.module,
"error_code": error_code,
"response_bytes": response_bytes,
})
);
}
adk-rust/crates/detmir-portal/src/production/metrics.rs
+183
View File
@@ -0,0 +1,183 @@
use std::collections::BTreeMap;
use std::fmt::Write as FmtWrite;
use std::sync::{Mutex, OnceLock};
use tiny_http::StatusCode;
use super::readiness::build_readyz;
use super::request_context::HttpRequestMetadata;
use crate::Cli;
static PORTAL_METRICS: OnceLock<Mutex<PortalMetrics>> = OnceLock::new();
#[derive(Clone, Debug, Eq, PartialEq, Ord, PartialOrd)]
struct HttpMetricKey {
method: String,
route: String,
status: u16,
module: String,
}
#[derive(Clone, Debug, Default)]
struct HttpMetricValue {
requests_total: u64,
duration_seconds_sum: f64,
duration_seconds_count: u64,
}
#[derive(Clone, Debug, Default)]
struct PortalMetrics {
http: BTreeMap<HttpMetricKey, HttpMetricValue>,
reports_generated_total: u64,
ingestion_records_total: u64,
ingestion_rejected_total: u64,
role_denied_total: u64,
}
fn portal_metrics() -> &'static Mutex<PortalMetrics> {
PORTAL_METRICS.get_or_init(|| Mutex::new(PortalMetrics::default()))
}
pub(crate) fn record_http_metric(metadata: &HttpRequestMetadata, status: StatusCode) {
if let Ok(mut metrics) = portal_metrics().lock() {
let entry = metrics
.http
.entry(HttpMetricKey {
method: metadata.method.clone(),
route: metadata.route.clone(),
status: status.0,
module: metadata.module.clone(),
})
.or_default();
entry.requests_total = entry.requests_total.saturating_add(1);
entry.duration_seconds_sum += metadata.latency_ms as f64 / 1_000.0;
entry.duration_seconds_count = entry.duration_seconds_count.saturating_add(1);
if status.0 == 403 {
metrics.role_denied_total = metrics.role_denied_total.saturating_add(1);
}
}
}
pub(crate) fn record_report_generated() {
if let Ok(mut metrics) = portal_metrics().lock() {
metrics.reports_generated_total = metrics.reports_generated_total.saturating_add(1);
}
}
pub(crate) fn record_ingestion_accepted() {
if let Ok(mut metrics) = portal_metrics().lock() {
metrics.ingestion_records_total = metrics.ingestion_records_total.saturating_add(1);
}
}
pub(crate) fn record_ingestion_rejected() {
if let Ok(mut metrics) = portal_metrics().lock() {
metrics.ingestion_rejected_total = metrics.ingestion_rejected_total.saturating_add(1);
}
}
pub(crate) fn render_prometheus_metrics(args: &Cli) -> String {
let mut text = String::new();
let ready_value = if build_readyz(args)
.get("status")
.and_then(serde_json::Value::as_str)
.unwrap_or("not_ready")
== "ready"
{
1
} else {
0
};
writeln!(
&mut text,
"# HELP awatch_http_requests_total HTTP requests handled by AWatch-rus portal"
)
.ok();
writeln!(&mut text, "# TYPE awatch_http_requests_total counter").ok();
let metrics = portal_metrics()
.lock()
.map(|guard| guard.clone())
.unwrap_or_default();
for (key, value) in &metrics.http {
writeln!(
&mut text,
"awatch_http_requests_total{{method=\"{}\",route=\"{}\",status=\"{}\",module=\"{}\"}} {}",
prom_escape(&key.method),
prom_escape(&key.route),
key.status,
prom_escape(&key.module),
value.requests_total
)
.ok();
}
writeln!(
&mut text,
"# HELP awatch_http_request_duration_seconds HTTP request duration in seconds"
)
.ok();
writeln!(
&mut text,
"# TYPE awatch_http_request_duration_seconds summary"
)
.ok();
for (key, value) in &metrics.http {
writeln!(
&mut text,
"awatch_http_request_duration_seconds_sum{{method=\"{}\",route=\"{}\",status=\"{}\",module=\"{}\"}} {:.6}",
prom_escape(&key.method),
prom_escape(&key.route),
key.status,
prom_escape(&key.module),
value.duration_seconds_sum
)
.ok();
writeln!(
&mut text,
"awatch_http_request_duration_seconds_count{{method=\"{}\",route=\"{}\",status=\"{}\",module=\"{}\"}} {}",
prom_escape(&key.method),
prom_escape(&key.route),
key.status,
prom_escape(&key.module),
value.duration_seconds_count
)
.ok();
}
for (name, help, value) in [
(
"awatch_reports_generated_total",
"Reports generated by the portal",
metrics.reports_generated_total,
),
(
"awatch_ingestion_records_total",
"Telemetry ingestion records accepted",
metrics.ingestion_records_total,
),
(
"awatch_ingestion_rejected_total",
"Telemetry ingestion records rejected",
metrics.ingestion_rejected_total,
),
(
"awatch_role_denied_total",
"Requests denied by role gates",
metrics.role_denied_total,
),
] {
writeln!(&mut text, "# HELP {name} {help}").ok();
writeln!(&mut text, "# TYPE {name} counter").ok();
writeln!(&mut text, "{name} {value}").ok();
}
writeln!(
&mut text,
"# HELP awatch_readyz_status Portal readiness status, 1=ready, 0=not_ready"
)
.ok();
writeln!(&mut text, "# TYPE awatch_readyz_status gauge").ok();
writeln!(&mut text, "awatch_readyz_status {ready_value}").ok();
text
}
fn prom_escape(value: &str) -> String {
value.replace('\\', "\\\\").replace('"', "\\\"")
}
adk-rust/crates/detmir-portal/src/production/mod.rs
+18
View File
@@ -0,0 +1,18 @@
pub(crate) mod health;
pub(crate) mod limits;
pub(crate) mod logging;
pub(crate) mod metrics;
pub(crate) mod readiness;
pub(crate) mod request_context;
pub(crate) mod version;
pub(crate) use health::build_healthz;
pub(crate) use limits::{is_limited_api_route, validate_api_query_limits, validate_portal_config};
pub(crate) use logging::log_http_request;
pub(crate) use metrics::{
record_http_metric, record_ingestion_accepted, record_ingestion_rejected,
record_report_generated, render_prometheus_metrics,
};
pub(crate) use readiness::build_readyz;
pub(crate) use request_context::{http_request_metadata, mark_request_started};
pub(crate) use version::build_version;
adk-rust/crates/detmir-portal/src/production/readiness.rs
+58
View File
@@ -0,0 +1,58 @@
use std::path::Path;
use serde_json::{Value, json};
use super::limits::validate_portal_config;
use crate::{Cli, now};
pub(crate) fn build_readyz(args: &Cli) -> Value {
let config_ok = validate_portal_config(args).is_ok();
let storage_status = storage_readiness_status(&args.state_dir);
let telemetry_status = storage_parent_readiness_status(&args.telemetry_store_path);
let evidence_status = storage_readiness_status(&args.evidence_root);
let ready = config_ok && !matches!(storage_status.as_str(), "error");
json!({
"status": if ready { "ready" } else { "not_ready" },
"generated_at_utc": now(),
"checks": {
"config": if config_ok { "ok" } else { "error" },
"storage": storage_status,
"telemetry_store": telemetry_status,
"evidence_storage": evidence_status,
"pfsense": "contract_only",
"security_events": security_events_readiness_status(args),
"clickhouse": if args.security_events_backend == "clickhouse" { "configured" } else { "not_required" }
}
})
}
fn storage_readiness_status(path: &Path) -> String {
if path.exists() {
if path.is_dir() {
"ok".to_string()
} else {
"error".to_string()
}
} else {
"not_configured".to_string()
}
}
fn storage_parent_readiness_status(path: &Path) -> String {
path.parent()
.map(storage_readiness_status)
.unwrap_or_else(|| "not_configured".to_string())
}
fn security_events_readiness_status(args: &Cli) -> &'static str {
match args
.security_events_backend
.trim()
.to_ascii_lowercase()
.as_str()
{
"disabled" | "" => "disabled",
"clickhouse" => "configured",
_ => "configured",
}
}
adk-rust/crates/detmir-portal/src/production/request_context.rs
+191
View File
@@ -0,0 +1,191 @@
use std::cell::RefCell;
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::{Instant, SystemTime, UNIX_EPOCH};
use tiny_http::Request;
use crate::{
normalize_path, parse_case_path, parse_case_status_path, parse_evidence_screenshot_path,
parse_investigation_pack_path, portal_role_from_request,
};
static REQUEST_SEQUENCE: AtomicU64 = AtomicU64::new(1);
thread_local! {
static REQUEST_STARTED_AT: RefCell<Option<Instant>> = const { RefCell::new(None) };
}
#[derive(Clone, Debug)]
pub(crate) struct HttpRequestMetadata {
pub(crate) method: String,
pub(crate) path: String,
pub(crate) route: String,
pub(crate) module: String,
pub(crate) role: String,
pub(crate) request_id: String,
pub(crate) correlation_id: String,
pub(crate) latency_ms: u64,
}
pub(crate) fn mark_request_started() {
REQUEST_STARTED_AT.with(|started| {
*started.borrow_mut() = Some(Instant::now());
});
}
fn request_latency_ms() -> u64 {
REQUEST_STARTED_AT.with(|started| {
started
.borrow()
.as_ref()
.map(|instant| instant.elapsed().as_millis() as u64)
.unwrap_or(0)
})
}
pub(crate) fn http_request_metadata(request: &Request) -> HttpRequestMetadata {
let raw_url = request.url().to_string();
let path = normalize_path(&raw_url);
let method = request.method().as_str().to_string();
let route = metrics_route(&path);
let module = metrics_module(&route).to_string();
let role = portal_role_from_request(request, &raw_url)
.as_str()
.to_string();
let request_id =
request_header(request, "X-Request-Id").or_else(|| request_header(request, "X-Request-ID"));
let correlation_id = request_header(request, "X-Correlation-Id")
.or_else(|| request_header(request, "X-Correlation-ID"));
let (request_id, correlation_id) = resolve_request_ids(request_id, correlation_id);
HttpRequestMetadata {
method,
path,
route,
module,
role,
request_id,
correlation_id,
latency_ms: request_latency_ms(),
}
}
fn request_header(request: &Request, name: &str) -> Option<String> {
request
.headers()
.iter()
.find(|header| header.field.to_string().eq_ignore_ascii_case(name))
.map(|header| header.value.as_str().to_string())
}
fn resolve_request_ids(
request_id: Option<String>,
correlation_id: Option<String>,
) -> (String, String) {
let request_id = request_id
.map(sanitize_request_token)
.filter(|value| !value.is_empty())
.unwrap_or_else(generate_request_id);
let correlation_id = correlation_id
.map(sanitize_request_token)
.filter(|value| !value.is_empty())
.unwrap_or_else(|| request_id.clone());
(request_id, correlation_id)
}
fn sanitize_request_token(value: String) -> String {
value
.chars()
.filter(|ch| ch.is_ascii_alphanumeric() || matches!(ch, '-' | '_' | '.' | ':'))
.take(96)
.collect()
}
fn generate_request_id() -> String {
let seq = REQUEST_SEQUENCE.fetch_add(1, Ordering::Relaxed);
let millis = SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|duration| duration.as_millis())
.unwrap_or(0);
format!("awatch-{millis}-{seq}")
}
fn metrics_route(path: &str) -> String {
if parse_investigation_pack_path(path).is_some() {
return "/api/investigation-pack/{candidate_id}".to_string();
}
if parse_case_path(path).is_some() {
return "/api/cases/{case_id}".to_string();
}
if parse_case_status_path(path).is_some() {
return "/api/cases/{case_id}/status".to_string();
}
if parse_evidence_screenshot_path(path).is_some() {
return "/api/dlp/evidence/{evidence_id}/asset".to_string();
}
match path {
"/" | "/operator" | "/manager" | "/owner" | "/incidents" | "/reports" | "/architecture" => {
path.to_string()
}
"/healthz" | "/api/healthz" => "/healthz".to_string(),
"/readyz" | "/api/readyz" => "/readyz".to_string(),
"/version" | "/api/version" => "/version".to_string(),
"/metrics" | "/api/metrics" => "/metrics".to_string(),
_ if path.starts_with("/api/") => path.to_string(),
_ => "other".to_string(),
}
}
fn metrics_module(route: &str) -> &'static str {
if route.starts_with("/healthz")
|| route.starts_with("/readyz")
|| route.starts_with("/version")
|| route.starts_with("/metrics")
{
"runtime"
} else if route.contains("/workforce") || route == "/manager" {
"workforce"
} else if route.contains("/security")
|| route.contains("/incidents")
|| route.contains("/incident-review")
{
"security"
} else if route.contains("/forensics")
|| route.contains("/investigation-pack")
|| route.contains("/cases")
|| route.contains("/dlp/evidence")
{
"forensics"
} else if route.contains("/ueba") {
"ueba"
} else if route.contains("/pfsense") {
"pfsense"
} else if route.contains("/reports") || route.contains("/executive") || route == "/operator" {
"reports"
} else {
"portal"
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn request_id_and_correlation_id_are_sanitized_and_linked() {
let (request_id, correlation_id) = resolve_request_ids(
Some(" req-123\nbad ".to_string()),
Some("corr-456/ignored".to_string()),
);
assert_eq!(request_id, "req-123bad");
assert_eq!(correlation_id, "corr-456ignored");
let (request_id, correlation_id) =
resolve_request_ids(Some("rid_1".to_string()), Some(" \n ".to_string()));
assert_eq!(request_id, "rid_1");
assert_eq!(correlation_id, "rid_1");
let (generated, generated_correlation) = resolve_request_ids(None, None);
assert!(generated.starts_with("awatch-"));
assert_eq!(generated, generated_correlation);
}
}
adk-rust/crates/detmir-portal/src/production/version.rs
+13
View File
@@ -0,0 +1,13 @@
use serde_json::{Value, json};
use crate::{Cli, PORTAL_SCHEMA_VERSION};
pub(crate) fn build_version(args: &Cli) -> Value {
json!({
"app_version": env!("CARGO_PKG_VERSION"),
"git_commit": option_env!("GIT_COMMIT").unwrap_or("unknown"),
"build_time": option_env!("BUILD_TIME").unwrap_or("unknown"),
"schema_version": PORTAL_SCHEMA_VERSION,
"environment": args.environment,
})
}
adk-rust/crates/detmir-portal/src/workforce_kpi_explain.rs
+729
View File
@@ -0,0 +1,729 @@
use serde_json::{Value, json};
use crate::production::limits::parse_query_date;
use crate::{
PortalRole, Snapshot, display_text_opt, query_param, role_envelope, trend_status,
workforce_index, workforce_index_status, workforce_trend_json, worktime_totals,
};
struct KpiFactorInputs<'a> {
users_count: usize,
active_seconds: i64,
apps_count: usize,
kpi_score: u8,
agent_coverage_percent: u8,
missing_sources: &'a [String],
}
#[derive(Clone, Debug, Default)]
pub(crate) struct KpiExplainQuery {
date: Option<String>,
department: Option<String>,
owner: Option<String>,
}
impl KpiExplainQuery {
pub(crate) fn from_url(url: &str) -> Self {
Self {
date: query_param(url, "date").filter(|value| parse_query_date(value).is_some()),
department: query_param(url, "department"),
owner: query_param(url, "owner"),
}
}
}
pub(crate) fn build_workforce_kpi_explain(
snapshot: &Snapshot,
policy_explain: &Value,
role: PortalRole,
query: &KpiExplainQuery,
anonymize: bool,
) -> Value {
let (users_count, active_seconds, apps_count) = worktime_totals(snapshot);
let base_index = workforce_index(users_count, active_seconds);
let policy_index = policy_explain
.get("index")
.and_then(Value::as_u64)
.map(|value| value.min(100) as u8);
let kpi_score = policy_index.or(base_index).unwrap_or(0);
let agent_coverage_percent = kpi_agent_coverage(snapshot);
let data_freshness = kpi_data_freshness(snapshot);
let missing_sources = kpi_missing_sources(snapshot, users_count, apps_count);
let confidence = kpi_confidence(
users_count,
agent_coverage_percent,
&data_freshness,
&missing_sources,
);
let factors = kpi_explain_factors(
snapshot,
policy_explain,
KpiFactorInputs {
users_count,
active_seconds,
apps_count,
kpi_score,
agent_coverage_percent,
missing_sources: &missing_sources,
},
);
let top_applications = kpi_top_applications(snapshot, anonymize);
let warnings = kpi_warnings(kpi_score, confidence, &data_freshness, &missing_sources);
let recommendations = kpi_recommendations(confidence, &missing_sources, kpi_score);
let mut payload = json!({
"ok": true,
"scope": "aggregate",
"role_context": role_envelope(role, "workforce_kpi_explain"),
"query": {
"date": query.date.clone(),
"department": query.department.clone(),
"owner": query.owner.clone(),
"employee_id_supported": false
},
"kpi_score": kpi_score,
"kpi_status": workforce_index_status(Some(kpi_score)),
"confidence": confidence,
"coverage": {
"agent_coverage_percent": agent_coverage_percent,
"data_freshness": data_freshness,
"missing_sources": missing_sources,
},
"factors": factors,
"top_applications": top_applications,
"warnings": warnings,
"recommendations": recommendations,
"formula": "rule_based: activity + business app usage - idle/afterhours/missing data with coverage confidence",
"model": {
"type": "rule_based",
"ml": false,
"llm": false,
"version": "workforce-kpi-explain-v1"
},
"generated_at_utc": snapshot.generated_at_utc,
});
filter_kpi_explain_for_role(&mut payload, role);
payload
}
fn kpi_agent_coverage(snapshot: &Snapshot) -> u8 {
if snapshot.agent_coverage_sla.expected_nodes > 0 {
snapshot.agent_coverage_sla.coverage_pct
} else if snapshot.agent_quality.sessions_collected_total > 0 {
75
} else if snapshot.worktime.ok {
60
} else {
0
}
}
fn kpi_data_freshness(snapshot: &Snapshot) -> String {
if snapshot.worktime.status.eq_ignore_ascii_case("OK")
&& snapshot
.worktime_management
.status
.eq_ignore_ascii_case("OK")
{
"fresh".to_string()
} else if snapshot.worktime.status.eq_ignore_ascii_case("DEGRADED")
|| snapshot
.worktime_management
.status
.eq_ignore_ascii_case("DEGRADED")
|| snapshot
.worktime
.summary
.to_ascii_lowercase()
.contains("stale")
|| snapshot
.worktime_management
.summary
.to_ascii_lowercase()
.contains("stale")
{
"stale".to_string()
} else {
"missing".to_string()
}
}
fn kpi_missing_sources(snapshot: &Snapshot, users_count: usize, apps_count: usize) -> Vec<String> {
let mut missing = Vec::new();
if !snapshot.worktime.ok || users_count == 0 {
missing.push("worktime".to_string());
}
if !snapshot.worktime_management.ok {
missing.push("worktime_management".to_string());
}
if apps_count == 0 {
missing.push("applications".to_string());
}
if snapshot.agent_coverage_sla.expected_nodes > 0
&& snapshot.agent_coverage_sla.coverage_pct < 50
{
missing.push("agent_coverage".to_string());
}
missing.sort();
missing.dedup();
missing
}
fn kpi_confidence(
users_count: usize,
agent_coverage_percent: u8,
data_freshness: &str,
missing_sources: &[String],
) -> &'static str {
if users_count == 0
|| agent_coverage_percent < 50
|| data_freshness == "missing"
|| missing_sources.iter().any(|item| item == "worktime")
{
"low"
} else if agent_coverage_percent < 80
|| data_freshness != "fresh"
|| !missing_sources.is_empty()
{
"medium"
} else {
"high"
}
}
fn kpi_explain_factors(
snapshot: &Snapshot,
policy_explain: &Value,
inputs: KpiFactorInputs<'_>,
) -> Vec<Value> {
let planned_seconds = (inputs.users_count as i64).saturating_mul(8 * 3600);
let idle_ratio = if planned_seconds > 0 {
((planned_seconds - inputs.active_seconds).max(0) as f64 / planned_seconds as f64)
.clamp(0.0, 1.0)
} else {
1.0
};
let weighted_apps = policy_explain
.get("matched_applications")
.and_then(Value::as_u64)
.unwrap_or(0);
let afterhours_seconds = kpi_afterhours_seconds(snapshot);
let remote_sessions = snapshot.agent_quality.rdp_sessions_total as u64;
let trend = trend_status(&workforce_trend_json(snapshot));
vec![
kpi_factor(
"productive_activity",
"Полезная активность",
positive_impact((inputs.kpi_score as i64 * 40) / 100),
if inputs.kpi_score >= 80 {
"Высокая доля активности относительно планового рабочего времени"
} else if inputs.kpi_score >= 60 {
"Активность близка к рабочему уровню, но есть просадка"
} else {
"Активность ниже ожидаемого рабочего уровня"
},
),
kpi_factor(
"business_app_usage",
"Рабочие приложения",
positive_impact(
((weighted_apps.max(inputs.apps_count as u64).min(12) as i64) * 2).min(24),
),
if weighted_apps > 0 {
"В данных есть приложения, попавшие под рабочие правила"
} else if inputs.apps_count > 0 {
"Есть активность по приложениям, но правила рабочих приложений требуют настройки"
} else {
"Данных о рабочих приложениях нет"
},
),
kpi_factor(
"idle_time",
"Простой",
negative_impact((idle_ratio * 30.0).round() as i64),
if idle_ratio > 0.35 {
"Есть значимые периоды неактивности в рабочее время"
} else {
"Простой не является основным фактором снижения индекса"
},
),
kpi_factor(
"afterhours_activity",
"Активность вне рабочего времени",
negative_impact((afterhours_seconds / 3600).min(12)),
if afterhours_seconds > 0 {
"Есть признаки активности за пределами рабочего окна"
} else {
"Существенная активность вне рабочего времени не выявлена"
},
),
kpi_factor(
"remote_session_activity",
"Удаленные сессии",
positive_impact((remote_sessions.min(5) * 2) as i64),
if remote_sessions > 0 {
"RDP/удаленные сессии подтверждают источник активности"
} else {
"Удаленные сессии не подтверждены текущим срезом"
},
),
kpi_factor(
"data_coverage",
"Полнота данных",
if inputs.agent_coverage_percent >= 80 {
positive_impact(12)
} else {
negative_impact(
((80_u8.saturating_sub(inputs.agent_coverage_percent) as i64) / 4).max(1),
)
},
if inputs.agent_coverage_percent >= 80 {
"Покрытие данных достаточно для уверенного управленческого вывода"
} else {
"Покрытие данных снижает доверие к индексу"
},
),
kpi_factor(
"missing_data",
"Отсутствующие данные",
negative_impact((inputs.missing_sources.len() as i64 * 8).min(32)),
if inputs.missing_sources.is_empty() {
"Критичных пропусков источников не выявлено"
} else {
"Есть пропуски источников, влияющие на надежность KPI"
},
),
kpi_factor(
"trend_change",
"Изменение тренда",
match trend.as_str() {
"monthly" | "weekly" => positive_impact(6),
"daily_only" => negative_impact(2),
_ => "0".to_string(),
},
match trend.as_str() {
"monthly" => "Есть месячная история для оценки тренда",
"weekly" => "Есть недельная история для оценки тренда",
"daily_only" => "Доступен только дневной срез, исторический тренд ограничен",
_ => "История тренда пока не накоплена",
},
),
]
}
fn kpi_factor(name: &str, label: &str, impact: String, explanation: &str) -> Value {
json!({
"name": name,
"label": label,
"impact": impact,
"explanation": explanation,
})
}
fn positive_impact(value: i64) -> String {
format!("+{}", value.max(0))
}
fn negative_impact(value: i64) -> String {
if value <= 0 {
"0".to_string()
} else {
format!("-{value}")
}
}
fn kpi_afterhours_seconds(snapshot: &Snapshot) -> i64 {
snapshot
.worktime_management
.payload
.as_ref()
.and_then(|payload| payload.get("department_rollups"))
.and_then(Value::as_array)
.map(|items| {
items
.iter()
.filter_map(|item| {
let total = item
.get("calendar_total_active_seconds")
.or_else(|| item.get("total_active_seconds"))
.and_then(Value::as_i64)?;
let workday = item
.get("workday_total_active_seconds")
.or_else(|| item.get("active_seconds"))
.and_then(Value::as_i64)
.unwrap_or(total);
Some((total - workday).max(0))
})
.sum()
})
.unwrap_or(0)
}
fn kpi_top_applications(snapshot: &Snapshot, anonymize: bool) -> Vec<Value> {
let Some(apps) = snapshot
.worktime
.payload
.as_ref()
.and_then(|payload| payload.get("true_active_apps"))
.and_then(Value::as_array)
else {
return Vec::new();
};
let mut items = apps
.iter()
.enumerate()
.filter_map(|(idx, app)| {
let raw_name = app.get("application").and_then(Value::as_str)?;
let seconds = app
.get("proved_work_seconds")
.or_else(|| app.get("active_seconds"))
.and_then(Value::as_i64)
.unwrap_or(0)
.max(0);
if seconds == 0 {
return None;
}
let name = if anonymize {
format!("Приложение {}", idx + 1)
} else {
display_text_opt(Some(raw_name), &format!("Приложение {}", idx + 1))
};
let category = kpi_application_category(raw_name);
let contribution = if category == "business" {
"positive"
} else {
"neutral"
};
Some(json!({
"name": name,
"category": category,
"active_minutes": seconds / 60,
"contribution": contribution,
}))
})
.collect::<Vec<_>>();
items.sort_by_key(|item| {
-item
.get("active_minutes")
.and_then(Value::as_i64)
.unwrap_or(0)
});
items.truncate(8);
items
}
fn kpi_application_category(name: &str) -> &'static str {
let lower = name.to_ascii_lowercase();
let lower_ru = name.to_lowercase();
if lower.contains("1c")
|| lower_ru.contains("1с")
|| lower.contains("erp")
|| lower.contains("sap")
|| lower.contains("excel")
|| lower.contains("office")
|| lower.contains("word")
{
"business"
} else if lower.contains("browser")
|| lower.contains("chrome")
|| lower.contains("edge")
|| lower.contains("firefox")
{
"mixed"
} else {
"other"
}
}
fn kpi_warnings(
kpi_score: u8,
confidence: &str,
data_freshness: &str,
missing_sources: &[String],
) -> Vec<String> {
let mut warnings = Vec::new();
if confidence == "low" {
warnings
.push("Низкое доверие к KPI: данных недостаточно для уверенного вывода.".to_string());
}
if data_freshness != "fresh" {
warnings.push(format!("Свежесть данных: {data_freshness}."));
}
for source in missing_sources {
warnings.push(format!("Не хватает источника данных: {source}."));
}
if kpi_score < 60 {
warnings.push("Индекс активности ниже рабочего ориентира.".to_string());
}
warnings
}
fn kpi_recommendations(confidence: &str, missing_sources: &[String], kpi_score: u8) -> Vec<String> {
let mut recommendations = Vec::new();
if !missing_sources.is_empty() {
recommendations
.push("Проверить свежесть источников и восстановить пропущенные данные.".to_string());
}
if confidence != "high" {
recommendations.push(
"Перед управленческим выводом проверить покрытие данных рабочих мест.".to_string(),
);
}
if kpi_score < 60 {
recommendations
.push("Проверить подразделения или ответственных с низкой активностью.".to_string());
}
if recommendations.is_empty() {
recommendations.push("Использовать KPI как агрегированный управленческий индикатор, не как персональную HR-оценку.".to_string());
}
recommendations
}
fn filter_kpi_explain_for_role(payload: &mut Value, role: PortalRole) {
match role {
PortalRole::Executive | PortalRole::Manager => {
payload["scope_note"] = json!("Агрегированный Workforce KPI без персональных деталей.");
}
PortalRole::Security => {
payload["scope_note"] =
json!("ИБ видит только факторы, релевантные риску и надежности данных.");
filter_kpi_factors(
payload,
&[
"afterhours_activity",
"remote_session_activity",
"data_coverage",
"missing_data",
"trend_change",
],
);
payload["top_applications"] = Value::Array(Vec::new());
}
PortalRole::Forensics => {
payload["scope_note"] = json!(
"Расследования видят только контекст надежности данных и временных отклонений."
);
filter_kpi_factors(
payload,
&["afterhours_activity", "data_coverage", "missing_data"],
);
payload["top_applications"] = Value::Array(Vec::new());
}
PortalRole::Admin => {
payload["scope_note"] =
json!("Администратор видит техническое покрытие источников и rule-based факторы.");
}
}
}
fn filter_kpi_factors(payload: &mut Value, allowed: &[&str]) {
if let Some(factors) = payload.get_mut("factors").and_then(Value::as_array_mut) {
factors.retain(|item| {
item.get("name")
.and_then(Value::as_str)
.is_some_and(|name| allowed.contains(&name))
});
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::{
AgentCoverageSla, AgentQuality, AgentQualityHistorySummary, AgentQualityNodesSummary,
SecurityEventsSummary, SourceStatus,
};
fn kpi_snapshot(
worktime_ok: bool,
coverage_pct: u8,
active_seconds: i64,
apps: Vec<Value>,
) -> Snapshot {
let rows = if active_seconds > 0 {
json!([
{"user": "USER-1", "user_id": "EMP-1", "active_seconds": active_seconds}
])
} else {
json!([])
};
Snapshot {
generated_at_utc: "2026-06-07T10:00:00Z".to_string(),
detmir_status: SourceStatus {
ok: true,
status: "OK".to_string(),
summary: String::new(),
error: None,
payload: None,
},
detmir_check: SourceStatus {
ok: true,
status: "OK".to_string(),
summary: String::new(),
error: None,
payload: None,
},
failed_units: SourceStatus {
ok: true,
status: "OK".to_string(),
summary: String::new(),
error: None,
payload: None,
},
worktime: SourceStatus {
ok: worktime_ok,
status: if worktime_ok { "OK" } else { "FAIL" }.to_string(),
summary: String::new(),
error: None,
payload: worktime_ok.then(|| {
json!({
"rows": rows,
"true_active_apps": apps
})
}),
},
worktime_management: SourceStatus {
ok: worktime_ok,
status: if worktime_ok { "OK" } else { "FAIL" }.to_string(),
summary: String::new(),
error: None,
payload: worktime_ok.then(|| {
json!({
"department_rollups": [
{
"name": "DEPT-1",
"users_count": 1,
"active_users": 1,
"portfolio_coverage_pct": coverage_pct,
"workday_total_active_seconds": active_seconds,
"calendar_total_active_seconds": active_seconds
}
],
"trend": [
{"report_date": "2026-06-06", "portfolio_coverage_pct": coverage_pct},
{"report_date": "2026-06-07", "portfolio_coverage_pct": coverage_pct}
]
})
}),
},
one_c: SourceStatus {
ok: true,
status: "OK".to_string(),
summary: String::new(),
error: None,
payload: None,
},
one_c_overview: SourceStatus {
ok: true,
status: "OK".to_string(),
summary: String::new(),
error: None,
payload: None,
},
agent_quality: AgentQuality {
collector_source: "awatch-agent-rs".to_string(),
collector_error: None,
sessions_collected_total: if worktime_ok { 1 } else { 0 },
active_sessions_total: if worktime_ok { 1 } else { 0 },
rdp_sessions_total: if worktime_ok { 1 } else { 0 },
quality_status: if worktime_ok { "OK" } else { "UNKNOWN" }.to_string(),
},
agent_quality_history: Vec::new(),
agent_quality_history_summary: AgentQualityHistorySummary::default(),
agent_quality_nodes: Vec::new(),
agent_quality_nodes_summary: AgentQualityNodesSummary::default(),
agent_coverage_sla: AgentCoverageSla {
expected_nodes: 1,
reporting_nodes_24h: if worktime_ok { 1 } else { 0 },
stale_nodes: 0,
missing_nodes: if worktime_ok { 0 } else { 1 },
coverage_pct,
freshness_pct: coverage_pct,
sla_status: if coverage_pct >= 80 { "OK" } else { "CRITICAL" }.to_string(),
problem_nodes: Vec::new(),
},
security_events_summary: SecurityEventsSummary::disabled(),
}
}
#[test]
fn confidence_is_low_when_required_data_is_missing() {
let snapshot = kpi_snapshot(false, 0, 0, Vec::new());
let explain = build_workforce_kpi_explain(
&snapshot,
&json!({"configured": false}),
PortalRole::Executive,
&KpiExplainQuery::default(),
false,
);
assert_eq!(explain["confidence"], "low");
assert_eq!(explain["kpi_score"], 0);
assert!(
explain["warnings"]
.as_array()
.unwrap()
.iter()
.any(|item| item.as_str().unwrap().contains("Низкое доверие"))
);
}
#[test]
fn factors_are_deterministic_and_role_filtered() {
let snapshot = kpi_snapshot(
true,
95,
8 * 3600,
vec![json!({"application": "1C", "proved_work_seconds": 6 * 3600})],
);
let policy = json!({
"configured": true,
"index": 82,
"matched_applications": 1
});
let explain = build_workforce_kpi_explain(
&snapshot,
&policy,
PortalRole::Executive,
&KpiExplainQuery::default(),
false,
);
assert_eq!(explain["kpi_score"], 82);
assert_eq!(explain["confidence"], "high");
let factors = explain["factors"].as_array().unwrap();
assert_eq!(factors.len(), 8);
assert_eq!(factors[0]["name"], "productive_activity");
assert!(
factors
.iter()
.any(|item| item["name"] == "business_app_usage")
);
assert!(
explain["top_applications"][0]["name"]
.as_str()
.unwrap()
.contains("1C")
);
assert!(
serde_json::to_string(&explain)
.unwrap()
.find("EMP-1")
.is_none()
);
let security = build_workforce_kpi_explain(
&snapshot,
&policy,
PortalRole::Security,
&KpiExplainQuery::default(),
false,
);
assert_eq!(security["top_applications"].as_array().unwrap().len(), 0);
assert!(
security["factors"]
.as_array()
.unwrap()
.iter()
.all(|item| item["name"] != "productive_activity")
);
}
}