refactor: distribute Acaua as a Docker image instead of a desktop app
Release / Plan (reuse the published image?) (push) Successful in 3s
CI / Docker image (build and smoke test) (push) Skipped
CI / Format and clippy (push) Successful in 4m32s
CI / Dependency audit (push) Successful in 11s
CI / Tests (Linux) (push) Successful in 9m15s
Release / ci (push) Successful in 9m18s
Release / Build image (arm64) (push) Failing after 7m7s
Release / Build image (amd64) (push) Failing after 13m53s
Release / Publish tags (push) Skipped
Release / Gitea release (push) Skipped

Acaua is a collective SOC daemon consuming only Zeek log files, not a
single-user tool needing local hardware like its siblings. Drop the
Tauri desktop shell, installers and self-update entirely; the server
and its web dashboard ship as a container, updated by pulling a new
image tag.

Link against the official prebuilt libduckdb instead of compiling it
from source via the bundled feature, cutting image build time from
15+ minutes to about 3. Fix a real bug found along the way: as PID 1
in a container, the daemon ignored SIGTERM and was always killed
after the stop timeout; it now shuts down gracefully.

Also corrects the Pillar 1 beaconing score: the period estimate had a
harmonic alias (a perfectly periodic 4h series resolved to 600s), and
the old score formula flagged pure random traffic as suspicious
(p95 = 0.839 on Poisson noise). The new formula was validated against
300 samples per scenario.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VAyr8d7HuAdUGB6aA5xmDD
This commit is contained in:
Lucas RayanandClaude Sonnet 5 committed 2026-10-31 19:54:45 -03:00
1 parent 1b974bb90d
commit 8a80504c45
72 files changed
+2524 -6384

No files matched your search

+33 -26
View File
@@ -25,6 +25,8 @@ enum Command {
},
Analyze {
directory: PathBuf,
#[arg(long, default_value = ".acaua")]
data_dir: PathBuf,
#[arg(long)]
json: bool,
},
@@ -32,10 +34,7 @@ enum Command {
#[arg(long)]
json: bool,
},
Update {
#[arg(long)]
apply: bool,
},
Update,
}
#[tokio::main]
@@ -50,8 +49,12 @@ async fn main() {
async fn run(cli: Cli) -> acaua_core::Result<()> {
match cli.command {
Command::Serve { bind, data_dir } => web::serve(&bind, data_dir).await,
Command::Analyze { directory, json } => {
let report = zeek::analyze_dir(&directory)?;
Command::Analyze {
directory,
data_dir,
json,
} => {
let report = zeek::analyze_dir(&directory, &data_dir)?;
if json {
println!("{}", serde_json::to_string_pretty(&report)?);
} else {
@@ -59,6 +62,9 @@ async fn run(cli: Cli) -> acaua_core::Result<()> {
println!("logs lidos: {}", report.logs_seen);
println!("registros: {}", report.records);
println!("alertas: {}", report.alerts.len());
for alert in &report.alerts {
println!(" [{:.3}] {}", alert.score, alert.summary);
}
println!("estado: {}", report.status);
}
Ok(())
@@ -80,26 +86,27 @@ async fn run(cli: Cli) -> acaua_core::Result<()> {
}
Ok(())
}
Command::Update { apply } => {
if apply {
let executable = std::env::current_exe()?;
let tag = update::apply(&executable)?;
println!("atualizado para {tag}");
} else {
let status = update::check(env!("CARGO_PKG_VERSION"))?;
println!(
"versão atual: {} | última: {}",
status.current_version,
status.latest_version.as_deref().unwrap_or("desconhecida")
);
println!(
"atualização disponível: {}",
if status.update_available {
"sim"
} else {
"não"
}
);
Command::Update => {
let status = update::check(env!("CARGO_PKG_VERSION"))?;
println!(
"versão atual: {} | última: {}",
status.current_version,
status.latest_version.as_deref().unwrap_or("desconhecida")
);
println!(
"atualização disponível: {}",
if status.update_available {
"sim"
} else {
"não"
}
);
if status.update_available {
if std::env::var("ACAUA_RUNTIME").as_deref() == Ok("container") {
println!("para atualizar: docker compose pull && docker compose up -d");
} else {
println!("para atualizar: baixe a nova imagem ou recompile a partir da tag");
}
}
Ok(())
}
+31 -20
View File
@@ -75,16 +75,19 @@ async fn alerts() -> Json<Vec<Alert>> {
}
async fn analyze(
State(state): State<AppState>,
Json(request): Json<AnalyzeRequest>,
) -> std::result::Result<Json<AnalysisReport>, (StatusCode, Json<ErrorBody>)> {
zeek::analyze_dir(&request.path).map(Json).map_err(|error| {
(
StatusCode::BAD_REQUEST,
Json(ErrorBody {
error: error.to_string(),
}),
)
})
zeek::analyze_dir(&request.path, state.storage.root())
.map(Json)
.map_err(|error| {
(
StatusCode::BAD_REQUEST,
Json(ErrorBody {
error: error.to_string(),
}),
)
})
}
async fn update_check() -> std::result::Result<Json<UpdateStatus>, (StatusCode, Json<ErrorBody>)> {
@@ -100,15 +103,6 @@ async fn update_check() -> std::result::Result<Json<UpdateStatus>, (StatusCode,
})
}
async fn update_apply() -> (StatusCode, Json<ErrorBody>) {
(
StatusCode::NOT_IMPLEMENTED,
Json(ErrorBody {
error: "update --apply is only available through the acaua CLI".to_string(),
}),
)
}
async fn logo() -> impl IntoResponse {
([(header::CONTENT_TYPE, "image/png")], Body::from(LOGO))
}
@@ -129,21 +123,38 @@ async fn asset(uri: Uri) -> Response {
}
async fn shutdown_signal() {
let _ = tokio::signal::ctrl_c().await;
let interrupt = async {
let _ = tokio::signal::ctrl_c().await;
};
#[cfg(unix)]
let terminate = async {
match tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate()) {
Ok(mut stream) => {
stream.recv().await;
}
Err(_) => std::future::pending::<()>().await,
}
};
#[cfg(not(unix))]
let terminate = std::future::pending::<()>();
tokio::select! {
_ = interrupt => {}
_ = terminate => {}
}
}
pub async fn serve(bind: &str, data_dir: PathBuf) -> Result<()> {
let _ = tracing_subscriber::fmt().try_init();
std::fs::create_dir_all(&data_dir)?;
let state = AppState {
storage: Arc::new(Storage::open(&data_dir)),
storage: Arc::new(Storage::open(&data_dir)?),
};
let app = Router::new()
.route("/api/health", get(health))
.route("/api/status", get(status))
.route("/api/alerts", get(alerts))
.route("/api/analyze", post(analyze))
.route("/api/update", get(update_check).post(update_apply))
.route("/api/update", get(update_check))
.route("/logo.png", get(logo))
.fallback(asset)
.with_state(state)
+5 -5
View File
@@ -8,17 +8,17 @@ repository.workspace = true
authors.workspace = true
rust-version.workspace = true
[features]
default = []
duckdb = ["dep:duckdb"]
[dependencies]
serde = { version = "1", features = ["derive"] }
serde_json = "1"
sha2 = "0.10"
thiserror = "1"
time = { version = "0.3", features = ["formatting", "macros"] }
duckdb = { version = "1", features = ["bundled"], optional = true }
chrono = { version = "0.4", default-features = false, features = ["std"] }
# Linked against the official prebuilt libduckdb (scripts/fetch-libduckdb.sh):
# never enable `bundled`, `parquet`, `json` or `extensions-full`, which compile
# DuckDB from source. Parquet, JSON and ICU are already inside libduckdb.so.
duckdb = "~1.10506.1"
[dev-dependencies]
tempfile = "3"
+60 -21
View File
@@ -14,40 +14,79 @@ pub struct DoctorReport {
pub status: &'static str,
}
fn command_available(name: &str, arg: &str) -> Check {
match Command::new(name).arg(arg).output() {
Ok(output) if output.status.success() => Check {
name: if name == "zeek" { "zeek" } else { "runtime" },
fn zeek_check() -> Check {
if std::env::var("ACAUA_RUNTIME").as_deref() == Ok("container") {
Check {
name: "zeek",
ok: true,
detail: format!("{name} is available"),
detail: "zeek runs in a dedicated container".to_string(),
}
} else {
match Command::new("zeek").arg("--version").output() {
Ok(output) if output.status.success() => Check {
name: "zeek",
ok: true,
detail: "zeek is available".to_string(),
},
Ok(_) => Check {
name: "zeek",
ok: false,
detail: "zeek returned an error".to_string(),
},
Err(error) => Check {
name: "zeek",
ok: false,
detail: format!("zeek not found: {error}"),
},
}
}
}
fn curl_check() -> Check {
match Command::new("curl").arg("--version").output() {
Ok(output) if output.status.success() => Check {
name: "curl",
ok: true,
detail: "curl is available".to_string(),
},
Ok(_) => Check {
name: if name == "zeek" { "zeek" } else { "runtime" },
name: "curl",
ok: false,
detail: format!("{name} returned an error"),
detail: "curl returned an error".to_string(),
},
Err(error) => Check {
name: if name == "zeek" { "zeek" } else { "runtime" },
name: "curl",
ok: false,
detail: format!("{name} not found: {error}"),
detail: format!("curl not found: {error}"),
},
}
}
fn duckdb_check() -> Check {
let result = duckdb::Connection::open_in_memory().and_then(|conn| {
conn.query_row("SELECT library_version FROM pragma_version()", [], |row| {
row.get::<_, String>(0)
})
});
match result {
Ok(version) => Check {
name: "duckdb",
ok: true,
detail: format!("libduckdb {version} linked"),
},
Err(error) => Check {
name: "duckdb",
ok: false,
detail: error.to_string(),
},
}
}
pub fn run() -> DoctorReport {
let checks = vec![
command_available("zeek", "--version"),
command_available("curl", "--version"),
command_available("tar", "--version"),
Check {
name: "duckdb",
ok: cfg!(feature = "duckdb"),
detail: if cfg!(feature = "duckdb") {
"bundled DuckDB feature is enabled".to_string()
} else {
"bundled DuckDB feature is disabled in this build".to_string()
},
},
zeek_check(),
curl_check(),
duckdb_check(),
Check {
name: "research",
ok: true,
+3
View File
@@ -16,6 +16,9 @@ pub enum AcauaError {
#[error("serialization error: {0}")]
Serde(#[from] serde_json::Error),
#[error("database error: {0}")]
Database(#[from] duckdb::Error),
}
pub type Result<T> = std::result::Result<T, AcauaError>;
+356 -5
View File
@@ -6,19 +6,370 @@ pub struct BeaconingInput {
pub timestamps: Vec<f64>,
}
#[derive(Debug, Clone, Serialize)]
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct GridInfo {
pub min_period_s: f64,
pub max_period_s: f64,
pub points: usize,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct WeightsInfo {
pub significance: f64,
pub regularity: f64,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct BeaconingScore {
pub score: f64,
pub observations: usize,
pub period_detected_seconds: f64,
pub p_value: f64,
pub peak_power: f64,
pub madm_normalized: f64,
pub skew_bowley: f64,
pub degenerate_quartiles: bool,
pub method: &'static str,
pub grid: GridInfo,
pub weights: WeightsInfo,
pub formula_version: &'static str,
pub status: &'static str,
}
fn quantile_cont(sorted: &[f64], q: f64) -> f64 {
if sorted.is_empty() {
return 0.0;
}
if sorted.len() == 1 {
return sorted[0];
}
let idx = (sorted.len() - 1) as f64 * q;
let lower = idx.floor() as usize;
let upper = idx.ceil() as usize;
let frac = idx - lower as f64;
if lower == upper {
sorted[lower]
} else {
sorted[lower] + frac * (sorted[upper] - sorted[lower])
}
}
pub fn score(input: &BeaconingInput) -> Result<BeaconingScore> {
let grid_info = GridInfo {
min_period_s: 600.0,
max_period_s: 43200.0,
points: 200,
};
let weights_info = WeightsInfo {
significance: 0.5,
regularity: 0.5,
};
let n = input.timestamps.len();
if n < 20 {
return Ok(BeaconingScore {
score: 0.0,
observations: n,
period_detected_seconds: 0.0,
p_value: 1.0,
peak_power: 0.0,
madm_normalized: 0.0,
skew_bowley: 0.0,
degenerate_quartiles: false,
method: "rayleigh-schuster",
grid: grid_info,
weights: weights_info,
formula_version: "1.1",
status: "insufficient_data",
});
}
let mut ts = input.timestamps.clone();
ts.sort_by(|a, b| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal));
let mut deltas = Vec::with_capacity(n - 1);
for i in 1..n {
deltas.push(ts[i] - ts[i - 1]);
}
let mut sorted_deltas = deltas.clone();
sorted_deltas.sort_by(|a, b| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal));
let q1 = quantile_cont(&sorted_deltas, 0.25);
let q2 = quantile_cont(&sorted_deltas, 0.50);
let q3 = quantile_cont(&sorted_deltas, 0.75);
const EPSILON: f64 = 1e-9;
let (skew_bowley, degenerate_quartiles) = if (q3 - q1).abs() < EPSILON {
(0.0, true)
} else {
((q3 + q1 - 2.0 * q2) / (q3 - q1), false)
};
let mut abs_devs: Vec<f64> = deltas.iter().map(|&d| (d - q2).abs()).collect();
abs_devs.sort_by(|a, b| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal));
let madm = quantile_cont(&abs_devs, 0.50);
let (madm_normalized, degenerate_deltas) = if q2 <= EPSILON {
(0.0, true)
} else {
(madm / q2, false)
};
// Rayleigh/Schuster periodogram with M = 200 log-spaced periods from 600s to 43200s
let t0 = ts[0];
let num_periods = 200;
let mut periods = Vec::with_capacity(num_periods);
let mut powers = Vec::with_capacity(num_periods);
for j in 0..num_periods {
let period = 600.0 * 72.0_f64.powf(j as f64 / 199.0);
periods.push(period);
let f = 1.0 / period;
let mut sum_cos = 0.0;
let mut sum_sin = 0.0;
for &t in &ts {
let phase = 2.0 * std::f64::consts::PI * f * (t - t0);
sum_cos += phase.cos();
sum_sin += phase.sin();
}
let power = (sum_cos * sum_cos + sum_sin * sum_sin) / (n as f64);
powers.push(power);
}
let mut p_max = 0.0_f64;
for &p in &powers {
if p > p_max {
p_max = p;
}
}
// Harmonic resistance: largest period in the grid with P >= 0.5 * P_max
let threshold_power = 0.5 * p_max;
let mut detected_period = periods[0];
for j in (0..num_periods).rev() {
if powers[j] >= threshold_power {
detected_period = periods[j];
break;
}
}
let m_points = num_periods as f64;
let p_value = (m_points * (-p_max).exp()).min(1.0);
// Formula v1.1
let ln_m = m_points.ln();
let six_ln_10 = 6.0 * 10.0_f64.ln();
let sig = ((p_max - ln_m) / six_ln_10).clamp(0.0, 1.0);
let regularity = (1.0 - 2.0 * madm_normalized).clamp(0.0, 1.0);
let status = if degenerate_deltas {
"degenerate_deltas"
} else {
"ok"
};
let score = if degenerate_deltas {
0.5 * sig
} else {
0.5 * sig + 0.5 * regularity
};
Ok(BeaconingScore {
score: 0.0,
observations: input.timestamps.len(),
method: "rayleigh-schuster-stub",
status: "skeleton",
score,
observations: n,
period_detected_seconds: detected_period,
p_value,
peak_power: p_max,
madm_normalized,
skew_bowley,
degenerate_quartiles,
method: "rayleigh-schuster",
grid: grid_info,
weights: weights_info,
formula_version: "1.1",
status,
})
}
#[cfg(test)]
mod tests {
use super::*;
struct SplitMix64 {
state: u64,
}
impl SplitMix64 {
fn new(seed: u64) -> Self {
Self { state: seed }
}
fn next_u64(&mut self) -> u64 {
self.state = self.state.wrapping_add(0x9e3779b97f4a7c15);
let mut z = self.state;
z = (z ^ (z >> 30)).wrapping_mul(0xbf58476d1ce4e5b9);
z = (z ^ (z >> 27)).wrapping_mul(0x94d049bb133111eb);
z ^ (z >> 31)
}
fn next_f64(&mut self) -> f64 {
(self.next_u64() >> 11) as f64 / (1u64 << 53) as f64
}
fn uniform(&mut self, low: f64, high: f64) -> f64 {
low + (high - low) * self.next_f64()
}
fn exponential(&mut self, mean: f64) -> f64 {
let u = self.next_f64().max(1e-15);
-mean * u.ln()
}
}
#[test]
fn test_1_perfectly_periodic_series() {
let mut timestamps = Vec::with_capacity(42);
for k in 0..42 {
timestamps.push(k as f64 * 14400.0);
}
let res = score(&BeaconingInput { timestamps }).unwrap();
assert!(res.score >= 0.99, "score: {}", res.score);
let rel_err = (res.period_detected_seconds - 14400.0).abs() / 14400.0;
assert!(
rel_err < 0.02,
"period: {}, rel_err: {}",
res.period_detected_seconds,
rel_err
);
assert_eq!(res.status, "ok");
}
#[test]
fn test_2_grid_jitter() {
let mut min_score = f64::MAX;
let mut max_score = f64::MIN;
for seed in 1..=50 {
let mut rng = SplitMix64::new(seed);
let t0 = 1000.0;
let mut timestamps = Vec::with_capacity(42);
for k in 0..42 {
let jitter = rng.uniform(-0.18, 0.18) * 14400.0;
timestamps.push(t0 + k as f64 * 14400.0 + jitter);
}
let res = score(&BeaconingInput { timestamps }).unwrap();
assert!(res.score >= 0.80, "seed {seed} score {} < 0.80", res.score);
if res.score < min_score {
min_score = res.score;
}
if res.score > max_score {
max_score = res.score;
}
}
println!("Test 2 (grid jitter) 50 seeds: min={min_score:.4}, max={max_score:.4}");
}
#[test]
fn test_3_cumulative_jitter() {
let mut min_score = f64::MAX;
let mut max_score = f64::MIN;
let mut period_within_10pct = 0;
for seed in 1..=50 {
let mut rng = SplitMix64::new(seed);
let mut timestamps = Vec::with_capacity(42);
let mut curr = 1000.0;
timestamps.push(curr);
for _ in 1..42 {
curr += 14400.0 * (1.0 + rng.uniform(-0.18, 0.18));
timestamps.push(curr);
}
let res = score(&BeaconingInput { timestamps }).unwrap();
assert!(res.score >= 0.40, "seed {seed} score {} < 0.40", res.score);
if res.score < min_score {
min_score = res.score;
}
if res.score > max_score {
max_score = res.score;
}
let rel_err = (res.period_detected_seconds - 14400.0).abs() / 14400.0;
if rel_err < 0.10 {
period_within_10pct += 1;
}
}
println!(
"Test 3 (cumulative jitter) 50 seeds: min={min_score:.4}, max={max_score:.4}, period_hits={period_within_10pct}/50"
);
assert!(
period_within_10pct >= 45,
"period hits: {period_within_10pct}/50"
);
}
#[test]
fn test_4_poisson_random() {
let mut min_score = f64::MAX;
let mut max_score = f64::MIN;
let mean = 7.0 * 86400.0 / 40.0;
for seed in 1..=50 {
let mut rng = SplitMix64::new(seed);
let mut timestamps = Vec::with_capacity(40);
let mut curr = 0.0;
for _ in 0..40 {
curr += rng.exponential(mean);
timestamps.push(curr);
}
let res = score(&BeaconingInput { timestamps }).unwrap();
assert!(res.score < 0.30, "seed {seed} score {} >= 0.30", res.score);
if res.score < min_score {
min_score = res.score;
}
if res.score > max_score {
max_score = res.score;
}
}
println!("Test 4 (Poisson) 50 seeds: min={min_score:.4}, max={max_score:.4}");
}
#[test]
fn test_5_sentinel_q3_eq_q1() {
let mut timestamps = Vec::with_capacity(25);
for k in 0..25 {
timestamps.push(k as f64 * 1000.0);
}
let res = score(&BeaconingInput { timestamps }).unwrap();
assert_eq!(res.skew_bowley, 0.0);
assert!(res.degenerate_quartiles);
assert!(!res.score.is_nan() && !res.score.is_infinite());
}
#[test]
fn test_6_median_delta_zero() {
let timestamps = vec![1000.0; 25];
let res = score(&BeaconingInput { timestamps }).unwrap();
assert_eq!(res.status, "degenerate_deltas");
assert!(!res.score.is_nan() && !res.score.is_infinite());
}
#[test]
fn test_7_insufficient_data() {
let timestamps: Vec<f64> = (0..19).map(|k| k as f64 * 1000.0).collect();
let res = score(&BeaconingInput { timestamps }).unwrap();
assert_eq!(res.status, "insufficient_data");
assert_eq!(res.score, 0.0);
}
#[test]
fn test_8_determinism() {
let timestamps: Vec<f64> = (0..42).map(|k| k as f64 * 14400.0).collect();
let s1 = score(&BeaconingInput {
timestamps: timestamps.clone(),
})
.unwrap();
let s2 = score(&BeaconingInput { timestamps }).unwrap();
let j1 = serde_json::to_vec(&s1).unwrap();
let j2 = serde_json::to_vec(&s2).unwrap();
assert_eq!(j1, j2);
}
}
+649 -8
View File
@@ -1,5 +1,9 @@
use crate::error::Result;
use crate::error::{AcauaError, Result};
use chrono::NaiveDate;
use serde::Serialize;
use sha2::{Digest, Sha256};
use std::collections::HashSet;
use std::fs;
use std::path::{Path, PathBuf};
#[derive(Debug, Clone, Serialize)]
@@ -18,32 +22,669 @@ impl DatasetSummary {
partitions: 0,
ttl_days: 7,
backend: "duckdb-parquet",
status: "skeleton",
status: "empty",
}
}
}
#[derive(Debug, Clone, Default)]
pub struct IngestionStats {
pub files_seen: usize,
pub records_ingested: u64,
pub lines_skipped: u64,
pub partitions_affected: usize,
}
#[derive(Debug, Clone)]
pub struct TupleTimestamps {
pub orig_h: String,
pub resp_h: String,
pub resp_p: u16,
pub timestamps: Vec<f64>,
}
#[derive(Debug, Clone)]
pub struct Storage {
root: PathBuf,
}
#[derive(Debug, Clone)]
pub struct ConnRecord {
pub ts: f64,
pub uid: String,
pub orig_h: String,
pub orig_p: u16,
pub resp_h: String,
pub resp_p: u16,
pub proto: String,
pub service: Option<String>,
pub duration: Option<f64>,
}
impl Storage {
pub fn open(root: impl AsRef<Path>) -> Self {
Self {
root: root.as_ref().to_path_buf(),
}
pub fn open(root: impl AsRef<Path>) -> Result<Self> {
let root = root.as_ref();
fs::create_dir_all(root.join("raw").join("conn"))?;
fs::create_dir_all(root.join("tmp"))?;
let canonical_root = fs::canonicalize(root)?;
let test_file = canonical_root.join("tmp").join(".write_check");
fs::write(&test_file, b"ok")?;
let _ = fs::remove_file(&test_file);
Ok(Self {
root: canonical_root,
})
}
pub fn root(&self) -> &Path {
&self.root
}
pub fn open_connection(&self) -> Result<duckdb::Connection> {
let conn = duckdb::Connection::open_in_memory()?;
let data_dir_str = self.root.to_string_lossy().replace('\'', "''");
let tmp_dir_str = self.root.join("tmp").to_string_lossy().replace('\'', "''");
conn.execute_batch(&format!(
"SET TimeZone = 'UTC';
SET temp_directory = '{tmp_dir_str}';
SET autoinstall_known_extensions = false;
SET autoload_known_extensions = false;
SET allowed_directories = ['{data_dir_str}/'];
SET enable_external_access = false;
SET lock_configuration = true;"
))?;
Ok(conn)
}
fn has_parquet_files(&self) -> bool {
let conn_dir = self.root.join("raw").join("conn");
if let Ok(entries) = fs::read_dir(conn_dir) {
for entry in entries.flatten() {
if let Ok(sub) = fs::read_dir(entry.path()) {
for f in sub.flatten() {
if f.path().extension().and_then(|s| s.to_str()) == Some("parquet") {
return true;
}
}
}
}
}
false
}
pub fn summary(&self) -> Result<DatasetSummary> {
Ok(DatasetSummary::empty())
if !self.has_parquet_files() {
return Ok(DatasetSummary::empty());
}
let conn = self.open_connection()?;
let parquet_glob = self
.root
.join("raw")
.join("conn")
.join("*")
.join("*.parquet")
.to_string_lossy()
.replace('\'', "''");
let query = format!(
"SELECT count(*)::BIGINT, count(DISTINCT date)::BIGINT
FROM read_parquet('{parquet_glob}', hive_partitioning = true);"
);
let row = conn.query_row(&query, [], |r| {
let records: i64 = r.get(0)?;
let partitions: i64 = r.get(1)?;
Ok((records as u64, partitions as usize))
});
match row {
Ok((records, partitions)) => Ok(DatasetSummary {
records,
partitions,
ttl_days: 7,
backend: "duckdb-parquet",
status: "ok",
}),
Err(_) => Ok(DatasetSummary::empty()),
}
}
pub fn prune_expired(&self) -> Result<usize> {
Ok(0)
let conn_dir = self.root.join("raw").join("conn");
let entries = match fs::read_dir(&conn_dir) {
Ok(e) => e,
Err(_) => return Ok(0),
};
let mut dates = Vec::new();
let mut dirs = Vec::new();
for entry in entries.flatten() {
let path = entry.path();
if !path.is_dir() {
continue;
}
if let Some(name) = path.file_name().and_then(|n| n.to_str()) {
if let Some(date_str) = name.strip_prefix("date=") {
if let Ok(d) = NaiveDate::parse_from_str(date_str, "%Y-%m-%d") {
dates.push(d);
dirs.push((d, path));
}
}
}
}
if dates.is_empty() {
return Ok(0);
}
let max_date = dates.into_iter().max().unwrap();
let cutoff = max_date - chrono::Duration::days(7);
let mut pruned = 0;
for (date, path) in dirs {
if date <= cutoff && fs::remove_dir_all(&path).is_ok() {
pruned += 1;
}
}
Ok(pruned)
}
pub fn parse_conn_log(content: &[u8]) -> (Vec<ConnRecord>, u64, HashSet<String>) {
let mut records = Vec::new();
let mut skipped = 0;
let mut dates = HashSet::new();
let mut separator = '\t';
let mut unset_field = String::from("-");
let mut fields: Vec<String> = Vec::new();
let mut idx_ts = None;
let mut idx_uid = None;
let mut idx_orig_h = None;
let mut idx_orig_p = None;
let mut idx_resp_h = None;
let mut idx_resp_p = None;
let mut idx_proto = None;
let mut idx_service = None;
let mut idx_duration = None;
let text = String::from_utf8_lossy(content);
for line in text.lines() {
let line = line.trim_end_matches(['\r', '\n']);
if line.is_empty() {
continue;
}
if line.starts_with('#') {
if let Some(rest) = line.strip_prefix("#separator ") {
let sep_str = rest.trim();
if let Some(hex_code) = sep_str.strip_prefix("\\x") {
if let Ok(b) = u8::from_str_radix(hex_code, 16) {
separator = b as char;
}
} else if !sep_str.is_empty() {
separator = sep_str.chars().next().unwrap();
}
} else if let Some(rest) = line.strip_prefix("#unset_field") {
unset_field = rest.trim().to_string();
} else if let Some(rest) = line.strip_prefix("#fields") {
let rest = rest.trim();
fields = rest
.split(separator)
.map(|s| s.trim().to_string())
.filter(|s| !s.is_empty())
.collect();
idx_ts = fields.iter().position(|s| s == "ts");
idx_uid = fields.iter().position(|s| s == "uid");
idx_orig_h = fields.iter().position(|s| s == "id.orig_h");
idx_orig_p = fields.iter().position(|s| s == "id.orig_p");
idx_resp_h = fields.iter().position(|s| s == "id.resp_h");
idx_resp_p = fields.iter().position(|s| s == "id.resp_p");
idx_proto = fields.iter().position(|s| s == "proto");
idx_service = fields.iter().position(|s| s == "service");
idx_duration = fields.iter().position(|s| s == "duration");
}
continue;
}
// Must have fields mapped
let (i_ts, i_uid, i_oh, i_op, i_rh, i_rp, i_pr) = match (
idx_ts, idx_uid, idx_orig_h, idx_orig_p, idx_resp_h, idx_resp_p, idx_proto,
) {
(Some(a), Some(b), Some(c), Some(d), Some(e), Some(f), Some(g)) => {
(a, b, c, d, e, f, g)
}
_ => {
skipped += 1;
continue;
}
};
let parts: Vec<&str> = line.split(separator).collect();
if parts.len() < fields.len() {
skipped += 1;
continue;
}
let ts: f64 = match parts[i_ts].parse() {
Ok(v) => v,
Err(_) => {
skipped += 1;
continue;
}
};
let orig_p: u16 = match parts[i_op].parse() {
Ok(v) => v,
Err(_) => {
skipped += 1;
continue;
}
};
let resp_p: u16 = match parts[i_rp].parse() {
Ok(v) => v,
Err(_) => {
skipped += 1;
continue;
}
};
let service = idx_service.and_then(|idx| {
let val = parts.get(idx)?;
if *val == unset_field.as_str() || *val == "-" {
None
} else {
Some(val.to_string())
}
});
let duration = idx_duration.and_then(|idx| {
let val = parts.get(idx)?;
if *val == unset_field.as_str() || *val == "-" {
None
} else {
val.parse::<f64>().ok()
}
});
// Track day partition (UTC)
let date_ts = chrono::DateTime::from_timestamp(ts.floor() as i64, 0);
if let Some(dt) = date_ts {
dates.insert(dt.format("%Y-%m-%d").to_string());
}
records.push(ConnRecord {
ts,
uid: parts[i_uid].to_string(),
orig_h: parts[i_oh].to_string(),
orig_p,
resp_h: parts[i_rh].to_string(),
resp_p,
proto: parts[i_pr].to_string(),
service,
duration,
});
}
(records, skipped, dates)
}
pub fn ingest_file(&self, path: impl AsRef<Path>) -> Result<(u64, u64, usize)> {
let path = path.as_ref();
let content = fs::read(path)?;
let mut hasher = Sha256::new();
hasher.update(&content);
let hash = hasher.finalize();
let mut sha16 = String::with_capacity(16);
for b in &hash[..8] {
use std::fmt::Write;
let _ = write!(sha16, "{b:02x}");
}
let (records, skipped, dates) = Self::parse_conn_log(&content);
if records.is_empty() {
return Ok((0, skipped, 0));
}
let conn = self.open_connection()?;
conn.execute_batch(
"CREATE TEMP TABLE t (
ts DOUBLE,
uid VARCHAR,
orig_h VARCHAR,
orig_p USMALLINT,
resp_h VARCHAR,
resp_p USMALLINT,
proto VARCHAR,
service VARCHAR,
duration DOUBLE
);",
)?;
{
let mut appender = conn.appender("t")?;
for r in &records {
appender.append_row(duckdb::params![
r.ts, r.uid, r.orig_h, r.orig_p, r.resp_h, r.resp_p, r.proto, r.service,
r.duration
])?;
}
}
let raw_conn_dir = self
.root
.join("raw")
.join("conn")
.to_string_lossy()
.replace('\'', "''");
let copy_sql = format!(
"COPY (
SELECT *, CAST(to_timestamp(ts) AS DATE) AS date FROM t
) TO '{raw_conn_dir}' (
FORMAT parquet,
PARTITION_BY (date),
FILENAME_PATTERN 'src_{sha16}_{{i}}',
OVERWRITE_OR_IGNORE
);"
);
conn.execute_batch(&copy_sql)?;
conn.execute_batch("DROP TABLE t;")?;
Ok((records.len() as u64, skipped, dates.len()))
}
pub fn ingest_dir(&self, directory: impl AsRef<Path>) -> Result<IngestionStats> {
let directory = directory.as_ref();
if !directory.is_dir() {
return Err(AcauaError::InvalidInput(format!(
"{} is not a directory",
directory.display()
)));
}
let canonical_dir = fs::canonicalize(directory)?;
let mut stats = IngestionStats::default();
let mut all_dates = HashSet::new();
let entries = fs::read_dir(&canonical_dir)?;
for entry in entries.flatten() {
let path = entry.path();
// Path confinement: do not follow symlink to outside directory
if let Ok(canon) = fs::canonicalize(&path) {
if !canon.starts_with(&canonical_dir) {
continue;
}
} else {
continue;
}
let file_name = match path.file_name().and_then(|n| n.to_str()) {
Some(n) => n,
None => continue,
};
// Skip .gz archives with note
if file_name.ends_with(".gz") {
continue;
}
// Only conn.log and conn.*.log directly in directory
let is_conn = file_name == "conn.log"
|| (file_name.starts_with("conn.") && file_name.ends_with(".log"));
if is_conn && path.is_file() {
stats.files_seen += 1;
match self.ingest_file(&path) {
Ok((recs, sk, affected)) => {
stats.records_ingested += recs;
stats.lines_skipped += sk;
for _ in 0..affected {
all_dates.insert(file_name.to_string());
}
}
Err(_) => {
stats.lines_skipped += 1;
}
}
}
}
stats.partitions_affected = all_dates.len();
Ok(stats)
}
pub fn query_7day_tuples(&self) -> Result<Vec<TupleTimestamps>> {
if !self.has_parquet_files() {
return Ok(Vec::new());
}
let conn = self.open_connection()?;
let parquet_glob = self
.root
.join("raw")
.join("conn")
.join("*")
.join("*.parquet")
.to_string_lossy()
.replace('\'', "''");
// Find max date present in data
let max_date_query = format!(
"SELECT max(date)::VARCHAR
FROM read_parquet('{parquet_glob}', hive_partitioning = true);"
);
let max_date: Option<String> = conn
.query_row(&max_date_query, [], |r| r.get(0))
.unwrap_or(None);
let max_date_str = match max_date {
Some(d) if !d.is_empty() => d,
_ => return Ok(Vec::new()),
};
let query = format!(
"WITH daily AS (
SELECT date, orig_h, resp_h, resp_p, list(ts ORDER BY ts) AS tss
FROM read_parquet('{parquet_glob}', hive_partitioning = true)
WHERE date > CAST('{max_date_str}' AS DATE) - INTERVAL 7 DAY
GROUP BY ALL
)
SELECT orig_h, resp_h, resp_p, list_sort(flatten(list(tss))) AS timestamps
FROM daily
GROUP BY ALL
ORDER BY orig_h, resp_h, resp_p;"
);
let mut stmt = conn.prepare(&query)?;
let rows = stmt.query_map([], |row| {
let orig_h: String = row.get(0)?;
let resp_h: String = row.get(1)?;
let resp_p: u16 = row.get(2)?;
let timestamps_val: duckdb::types::Value = row.get(3)?;
let mut timestamps = Vec::new();
if let duckdb::types::Value::List(items) = timestamps_val {
for item in items {
if let duckdb::types::Value::Double(t) = item {
timestamps.push(t);
}
}
}
Ok(TupleTimestamps {
orig_h,
resp_h,
resp_p,
timestamps,
})
})?;
let mut result = Vec::new();
for r in rows {
result.push(r?);
}
Ok(result)
}
}
#[cfg(test)]
mod tests {
use super::*;
const CONN_LOG_SAMPLE_1: &str = "#separator \\x09\n\
#set_separator\t,\n\
#empty_field\t(empty)\n\
#unset_field\t-\n\
#path\tconn\n\
#open\t2026-10-31-12-00-00\n\
#fields\tts\tuid\tid.orig_h\tid.orig_p\tid.resp_h\tid.resp_p\tproto\tservice\tduration\tip_proto\n\
#types\ttime\tstring\taddr\tport\taddr\tport\tenum\tstring\tinterval\tcount\n\
1793461200.000000\tCuid1\t192.168.1.10\t45000\t10.0.0.1\t443\ttcp\tssl\t1.234\t6\n\
1793461205.000000\tCuid2\t192.168.1.10\t45001\t10.0.0.1\t443\ttcp\t-\t-\t6\n\
#close\t2026-10-31-12-01-00\n";
const CONN_LOG_PERMUTED_COLS: &str = "#separator \\x09\n\
#fields\tid.resp_p\tid.orig_h\tservice\tuid\tduration\tproto\tts\tid.resp_h\tid.orig_p\n\
443\t192.168.1.10\thttp\tCuid3\t0.5\ttcp\t1793461210.000000\t10.0.0.1\t45002\n";
#[test]
fn test_parser_extracts_correct_fields_and_unset_as_none() {
let (records, skipped, dates) = Storage::parse_conn_log(CONN_LOG_SAMPLE_1.as_bytes());
assert_eq!(skipped, 0);
assert_eq!(records.len(), 2);
assert_eq!(dates.len(), 1);
assert_eq!(records[0].uid, "Cuid1");
assert_eq!(records[0].orig_h, "192.168.1.10");
assert_eq!(records[0].orig_p, 45000);
assert_eq!(records[0].resp_h, "10.0.0.1");
assert_eq!(records[0].resp_p, 443);
assert_eq!(records[0].proto, "tcp");
assert_eq!(records[0].service.as_deref(), Some("ssl"));
assert_eq!(records[0].duration, Some(1.234));
assert_eq!(records[1].uid, "Cuid2");
assert_eq!(records[1].service, None);
assert_eq!(records[1].duration, None);
}
#[test]
fn test_parser_maps_by_name_not_position() {
let (records, skipped, _) = Storage::parse_conn_log(CONN_LOG_PERMUTED_COLS.as_bytes());
assert_eq!(skipped, 0);
assert_eq!(records.len(), 1);
let r = &records[0];
assert_eq!(r.uid, "Cuid3");
assert_eq!(r.orig_h, "192.168.1.10");
assert_eq!(r.orig_p, 45002);
assert_eq!(r.resp_h, "10.0.0.1");
assert_eq!(r.resp_p, 443);
assert_eq!(r.proto, "tcp");
assert_eq!(r.service.as_deref(), Some("http"));
assert_eq!(r.duration, Some(0.5));
}
#[test]
fn test_storage_roundtrip_and_idempotence() {
let temp_dir = tempfile::tempdir().unwrap();
let store = Storage::open(temp_dir.path()).unwrap();
let log_file = temp_dir.path().join("conn.log");
fs::write(&log_file, CONN_LOG_SAMPLE_1).unwrap();
let (ingested1, skipped1, _) = store.ingest_file(&log_file).unwrap();
assert_eq!(ingested1, 2);
assert_eq!(skipped1, 0);
let summary1 = store.summary().unwrap();
assert_eq!(summary1.records, 2);
assert_eq!(summary1.partitions, 1);
// Idempotence: re-ingesting the exact same file does not duplicate records
let (ingested2, skipped2, _) = store.ingest_file(&log_file).unwrap();
assert_eq!(ingested2, 2);
assert_eq!(skipped2, 0);
let summary2 = store.summary().unwrap();
assert_eq!(summary2.records, 2);
// Ingesting a different file for the same tuple/day adds to it
let log_file_2 = temp_dir.path().join("conn.extra.log");
fs::write(&log_file_2, CONN_LOG_PERMUTED_COLS).unwrap();
store.ingest_file(&log_file_2).unwrap();
let summary3 = store.summary().unwrap();
assert_eq!(summary3.records, 3);
let tuples = store.query_7day_tuples().unwrap();
assert_eq!(tuples.len(), 1);
assert_eq!(tuples[0].orig_h, "192.168.1.10");
assert_eq!(tuples[0].resp_h, "10.0.0.1");
assert_eq!(tuples[0].resp_p, 443);
assert_eq!(tuples[0].timestamps.len(), 3);
assert_eq!(tuples[0].timestamps[0], 1793461200.0);
assert_eq!(tuples[0].timestamps[1], 1793461205.0);
assert_eq!(tuples[0].timestamps[2], 1793461210.0);
}
#[test]
fn test_storage_utc_timezone_partitioning() {
let temp_dir = tempfile::tempdir().unwrap();
let store = Storage::open(temp_dir.path()).unwrap();
// 2026-10-31 02:00:00 UTC = 1793412000
let log = "#separator \\x09\n#fields\tts\tuid\tid.orig_h\tid.orig_p\tid.resp_h\tid.resp_p\tproto\n1793412000.000000\tu1\t10.0.0.1\t1000\t10.0.0.2\t80\ttcp\n";
let log_file = temp_dir.path().join("conn.log");
fs::write(&log_file, log).unwrap();
store.ingest_file(&log_file).unwrap();
let partition_dir = temp_dir
.path()
.join("raw")
.join("conn")
.join("date=2026-10-31");
assert!(
partition_dir.is_dir(),
"partition date=2026-10-31 should exist regardless of local timezone"
);
}
#[test]
fn test_storage_sandbox_blocks_unauthorized_paths() {
let temp_dir = tempfile::tempdir().unwrap();
let store = Storage::open(temp_dir.path()).unwrap();
let conn = store.open_connection().unwrap();
let res = conn.execute_batch("SELECT * FROM read_csv('/etc/passwd');");
assert!(res.is_err(), "sandbox must block access outside data_dir");
}
#[test]
fn test_storage_prune_expired() {
let temp_dir = tempfile::tempdir().unwrap();
let store = Storage::open(temp_dir.path()).unwrap();
let raw_conn = temp_dir.path().join("raw").join("conn");
// Day 1 (10 days ago) and Day 10 (recent)
let old_part = raw_conn.join("date=2026-10-01");
let new_part = raw_conn.join("date=2026-10-11");
fs::create_dir_all(&old_part).unwrap();
fs::create_dir_all(&new_part).unwrap();
let pruned = store.prune_expired().unwrap();
assert_eq!(pruned, 1);
assert!(!old_part.exists());
assert!(new_part.exists());
}
}
+4 -530
View File
@@ -1,25 +1,13 @@
use crate::error::{AcauaError, Result};
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
use std::ffi::OsStr;
use std::fs;
use std::path::{Path, PathBuf};
use std::process::Command;
const LATEST_RELEASE_URL: &str =
"https://git.lucasrguerra.dev.br/api/v1/repos/ciberfauna/acaua/releases/latest";
const RELEASES_PAGE_URL: &str = "https://git.lucasrguerra.dev.br/ciberfauna/acaua/releases";
#[derive(Debug, Deserialize)]
struct Release {
tag_name: String,
assets: Vec<Asset>,
}
#[derive(Debug, Deserialize)]
struct Asset {
name: String,
browser_download_url: String,
}
#[derive(Debug, Clone, Serialize)]
@@ -46,42 +34,6 @@ fn curl_text(url: &str) -> Result<String> {
Ok(String::from_utf8_lossy(&output.stdout).into_owned())
}
fn curl_download(url: &str, destination: &Path) -> Result<()> {
let status = Command::new("curl")
.args(["-fsSL", "-H", "User-Agent: acaua-updater", "-o"])
.arg(destination)
.arg(url)
.status()
.map_err(|error| AcauaError::CommandFailed {
cmd: format!("curl {url}"),
detail: error.to_string(),
})?;
if !status.success() {
return Err(AcauaError::CommandFailed {
cmd: format!("curl {url}"),
detail: "download failed".to_string(),
});
}
Ok(())
}
fn asset_suffix_for_platform() -> String {
let os = if cfg!(target_os = "linux") {
"linux"
} else if cfg!(target_os = "macos") {
"macos"
} else {
"windows"
};
let arch = if cfg!(target_arch = "x86_64") {
"x86_64"
} else {
"aarch64"
};
let extension = if cfg!(windows) { "zip" } else { "tar.gz" };
format!("-{os}-{arch}.{extension}")
}
pub fn check(current_version: &str) -> Result<UpdateStatus> {
let body = match curl_text(LATEST_RELEASE_URL) {
Ok(body) => body,
@@ -103,492 +55,14 @@ pub fn check(current_version: &str) -> Result<UpdateStatus> {
})
}
pub fn apply(current_exe: &Path) -> Result<String> {
let is_cli = current_exe
.file_stem()
.map(|stem| stem == "acaua")
.unwrap_or(false);
if !is_cli {
return Err(AcauaError::CommandFailed {
cmd: "update".to_string(),
detail: format!(
"automatic updates only replace the acaua CLI; download the desktop installer from {RELEASES_PAGE_URL}"
),
});
}
let body = curl_text(LATEST_RELEASE_URL)?;
let release: Release = serde_json::from_str(&body)?;
let suffix = asset_suffix_for_platform();
let asset = release
.assets
.iter()
.find(|asset| asset.name.ends_with(&suffix))
.ok_or_else(|| AcauaError::CommandFailed {
cmd: "update".to_string(),
detail: format!(
"no asset ending in `{suffix}` in release {}",
release.tag_name
),
})?;
let checksum_asset = release
.assets
.iter()
.find(|candidate| candidate.name == format!("{}.sha256", asset.name));
let temp_dir = std::env::temp_dir().join("acaua-update");
fs::create_dir_all(&temp_dir)?;
let archive_path = temp_dir.join(&asset.name);
curl_download(&asset.browser_download_url, &archive_path)?;
if let Some(checksum_asset) = checksum_asset {
let expected = curl_text(&checksum_asset.browser_download_url)?;
let expected_hash = expected
.split_whitespace()
.next()
.unwrap_or("")
.to_lowercase();
let data = fs::read(&archive_path)?;
let mut hasher = Sha256::new();
hasher.update(&data);
let actual_hash = format!("{:x}", hasher.finalize());
if actual_hash != expected_hash {
return Err(AcauaError::CommandFailed {
cmd: "update".to_string(),
detail: "SHA-256 checksum mismatch; update discarded".to_string(),
});
}
}
let extract_dir = temp_dir.join("extracted");
let _ = fs::remove_dir_all(&extract_dir);
fs::create_dir_all(&extract_dir)?;
let status = Command::new("tar")
.arg("xf")
.arg(&archive_path)
.arg("-C")
.arg(&extract_dir)
.status()
.map_err(|error| AcauaError::CommandFailed {
cmd: "tar xf".to_string(),
detail: error.to_string(),
})?;
if !status.success() {
return Err(AcauaError::CommandFailed {
cmd: "tar xf".to_string(),
detail: "extraction failed".to_string(),
});
}
let new_binary = find_binary(&extract_dir)?;
let backup = current_exe.with_extension("bak");
let _ = fs::remove_file(&backup);
fs::rename(current_exe, &backup)?;
fs::copy(&new_binary, current_exe)?;
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt;
let mut permissions = fs::metadata(current_exe)?.permissions();
permissions.set_mode(0o755);
fs::set_permissions(current_exe, permissions)?;
}
Ok(release.tag_name)
}
fn find_binary(directory: &Path) -> Result<PathBuf> {
fn is_binary(name: &OsStr) -> bool {
name == "acaua" || name == "acaua.exe"
}
for entry in fs::read_dir(directory)? {
let entry = entry?;
if is_binary(&entry.file_name()) {
return Ok(entry.path());
}
if entry.file_type().map(|kind| kind.is_dir()).unwrap_or(false) {
for inner in fs::read_dir(entry.path())? {
let inner = inner?;
if is_binary(&inner.file_name()) {
return Ok(inner.path());
}
}
}
}
Err(AcauaError::CommandFailed {
cmd: "update".to_string(),
detail: "binary not found in the downloaded package".to_string(),
})
}
/// Desktop GUI installation type detected at runtime.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum DesktopInstall {
/// Linux AppImage: real filesystem path of the .AppImage file (from APPIMAGE env var).
AppImage(PathBuf),
/// Linux Debian package installed via dpkg: `current_exe` is reliable here.
Deb,
/// Windows NSIS installer: downloaded setup executable run with `/S`
/// (silent). Does not relaunch the app on its own — see the comment on
/// the `WindowsInstaller` apply path.
WindowsInstaller,
}
/// Detects the desktop installation type given an optional `APPIMAGE` environment
/// variable value and the current executable path.
pub fn detect_desktop_install_from(
appimage_env: Option<&std::ffi::OsStr>,
current_exe: &Path,
) -> Result<DesktopInstall> {
#[cfg(windows)]
{
let _ = (appimage_env, current_exe);
Ok(DesktopInstall::WindowsInstaller)
}
#[cfg(target_os = "linux")]
{
if let Some(val) = appimage_env {
let path = PathBuf::from(val);
if path.is_file() {
return Ok(DesktopInstall::AppImage(path));
}
}
if current_exe.is_file()
&& current_exe
.extension()
.and_then(|e| e.to_str())
.map(|e| e.eq_ignore_ascii_case("appimage"))
.unwrap_or(false)
{
return Ok(DesktopInstall::AppImage(current_exe.to_path_buf()));
}
Ok(DesktopInstall::Deb)
}
#[cfg(not(any(windows, target_os = "linux")))]
{
let _ = (appimage_env, current_exe);
Err(AcauaError::CommandFailed {
cmd: "update".to_string(),
detail: format!(
"automatic self-update is not supported on this platform; download manually from {RELEASES_PAGE_URL}"
),
})
}
}
/// Detects the desktop installation type using the live environment.
pub fn detect_desktop_install() -> Result<DesktopInstall> {
let exe = std::env::current_exe()?;
detect_desktop_install_from(std::env::var_os("APPIMAGE").as_deref(), &exe)
}
/// Returns the expected release asset suffix for a desktop installation and CPU architecture.
pub fn gui_asset_suffix_for(install: &DesktopInstall, arch: &str) -> Result<&'static str> {
match install {
DesktopInstall::AppImage(_) => match arch {
"x86_64" => Ok("_amd64.AppImage"),
"aarch64" => Ok("_aarch64.AppImage"),
other => Err(AcauaError::CommandFailed {
cmd: "update".to_string(),
detail: format!("unsupported architecture for AppImage: {other}"),
}),
},
DesktopInstall::Deb => match arch {
"x86_64" => Ok("_amd64.deb"),
"aarch64" => Ok("_arm64.deb"),
other => Err(AcauaError::CommandFailed {
cmd: "update".to_string(),
detail: format!("unsupported architecture for deb package: {other}"),
}),
},
DesktopInstall::WindowsInstaller => match arch {
"x86_64" => Ok("_x64-setup.exe"),
other => Err(AcauaError::CommandFailed {
cmd: "update".to_string(),
detail: format!("unsupported architecture for Windows installer: {other}"),
}),
},
}
}
/// Returns the expected release asset suffix for the current architecture.
pub fn gui_asset_suffix(install: &DesktopInstall) -> Result<&'static str> {
gui_asset_suffix_for(install, std::env::consts::ARCH)
}
/// Downloads, verifies, and installs the latest desktop GUI release in place.
pub fn apply_desktop(current_exe: &Path) -> Result<String> {
apply_desktop_with_progress(current_exe, |_| {})
}
/// Downloads, verifies, and installs the latest desktop GUI release in place,
/// notifying the progress callback of phase transitions (`"downloading"`,
/// `"verifying"`, `"installing"`, `"restarting"`).
pub fn apply_desktop_with_progress<F>(current_exe: &Path, on_progress: F) -> Result<String>
where
F: Fn(&str),
{
let install =
detect_desktop_install_from(std::env::var_os("APPIMAGE").as_deref(), current_exe)?;
let body = curl_text(LATEST_RELEASE_URL)?;
let release: Release = serde_json::from_str(&body)?;
let suffix = gui_asset_suffix(&install)?;
let asset = release
.assets
.iter()
.find(|asset| asset.name.ends_with(suffix))
.ok_or_else(|| AcauaError::CommandFailed {
cmd: "update".to_string(),
detail: format!(
"no asset ending with `{suffix}` in release {}",
release.tag_name
),
})?;
let checksum_asset = release
.assets
.iter()
.find(|asset| asset.name == format!("{}.sha256", asset.name));
let tmp_dir = std::env::temp_dir().join("acaua-desktop-update");
fs::create_dir_all(&tmp_dir)?;
let download_path = tmp_dir.join(&asset.name);
on_progress("downloading");
curl_download(&asset.browser_download_url, &download_path)?;
if let Some(checksum_asset) = checksum_asset {
on_progress("verifying");
let expected = curl_text(&checksum_asset.browser_download_url)?;
let expected_hash = expected
.split_whitespace()
.next()
.unwrap_or("")
.to_lowercase();
let data = fs::read(&download_path)?;
let mut hasher = Sha256::new();
hasher.update(&data);
let actual_hash: String = hasher
.finalize()
.iter()
.map(|byte| format!("{byte:02x}"))
.collect();
if actual_hash != expected_hash {
return Err(AcauaError::CommandFailed {
cmd: "update".to_string(),
detail: "SHA-256 checksum mismatch — update discarded".to_string(),
});
}
}
on_progress("installing");
match install {
DesktopInstall::AppImage(target_appimage) => {
#[cfg(target_os = "linux")]
{
let parent_dir = target_appimage.parent().unwrap_or_else(|| Path::new("."));
let temp_name = format!(
".{}.tmp-{}",
target_appimage
.file_name()
.and_then(|n| n.to_str())
.unwrap_or("appimage"),
std::process::id()
);
let temp_file = parent_dir.join(temp_name);
fs::copy(&download_path, &temp_file)?;
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt;
let mut perms = fs::metadata(&temp_file)?.permissions();
perms.set_mode(0o755);
fs::set_permissions(&temp_file, perms)?;
}
fs::rename(&temp_file, &target_appimage)?;
let _ = fs::remove_file(&download_path);
on_progress("restarting");
#[cfg(unix)]
{
use std::os::unix::process::CommandExt;
let err = Command::new(&target_appimage)
.args(std::env::args().skip(1))
.exec();
return Err(AcauaError::CommandFailed {
cmd: "update".to_string(),
detail: format!("failed to relaunch AppImage: {err}"),
});
}
}
#[cfg(not(target_os = "linux"))]
{
let _ = (&target_appimage, &download_path);
return Err(AcauaError::CommandFailed {
cmd: "update".to_string(),
detail: "AppImage updates are only supported on Linux".to_string(),
});
}
}
DesktopInstall::Deb => {
#[cfg(target_os = "linux")]
{
let status = Command::new("dpkg")
.args(["-i"])
.arg(&download_path)
.status()
.map_err(|e| AcauaError::CommandFailed {
cmd: "dpkg -i".to_string(),
detail: e.to_string(),
})?;
if !status.success() {
return Err(AcauaError::CommandFailed {
cmd: "dpkg -i".to_string(),
detail: format!("dpkg -i failed with exit code {:?}", status.code()),
});
}
let _ = fs::remove_file(&download_path);
on_progress("restarting");
#[cfg(unix)]
{
use std::os::unix::process::CommandExt;
let err = Command::new(current_exe)
.args(std::env::args().skip(1))
.exec();
return Err(AcauaError::CommandFailed {
cmd: "update".to_string(),
detail: format!("failed to relaunch updated app: {err}"),
});
}
}
#[cfg(not(target_os = "linux"))]
{
let _ = (&download_path, current_exe);
return Err(AcauaError::CommandFailed {
cmd: "update".to_string(),
detail: "deb updates are only supported on Linux".to_string(),
});
}
}
DesktopInstall::WindowsInstaller => {
#[cfg(windows)]
{
// `/S` is the one NSIS silent-install switch with a documented,
// stable meaning. There is no standard NSIS or Tauri flag that
// makes a silent install relaunch the app afterwards — that was
// claimed in an earlier draft of this code and could not be
// verified against NSIS's or Tauri's documentation; passing an
// invented flag to a stock NSIS installer is silently ignored,
// not harmful, but also not a relaunch mechanism. Spawning
// detached and exiting immediately (instead of waiting on the
// child) also avoids a self-deadlock: if the installer's script
// waits for this process to close before it can overwrite the
// running .exe, blocking here on the installer's exit status
// would hang forever. The practical effect: the update installs
// silently, but the app does not reopen itself — the user needs
// to start it again. This needs real-world confirmation on
// Windows before relying on it; if testing shows the bundled
// NSIS template does support an explicit relaunch flag or step,
// wire it in here instead of this comment.
on_progress("restarting");
Command::new(&download_path)
.arg("/S")
.spawn()
.map_err(|e| AcauaError::CommandFailed {
cmd: "installer".to_string(),
detail: format!("failed to launch installer: {e}"),
})?;
std::process::exit(0);
}
#[cfg(not(windows))]
{
let _ = &download_path;
return Err(AcauaError::CommandFailed {
cmd: "update".to_string(),
detail: "Windows installer updates are only supported on Windows".to_string(),
});
}
}
}
#[allow(unreachable_code)]
Ok(release.tag_name)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn apply_refuses_to_overwrite_the_desktop_app() {
let error = apply(Path::new("/opt/acaua/acaua-gui")).unwrap_err();
assert!(error.to_string().contains("CLI"), "{error}");
assert!(error.to_string().contains(RELEASES_PAGE_URL), "{error}");
}
#[test]
fn detect_desktop_install_variants() {
#[cfg(target_os = "linux")]
{
let temp_appimage =
std::env::temp_dir().join(format!("test-acaua-{}.AppImage", std::process::id()));
let _ = fs::write(&temp_appimage, b"dummy appimage");
let res = detect_desktop_install_from(
Some(temp_appimage.as_os_str()),
Path::new("/usr/bin/acaua-gui"),
)
.unwrap();
assert_eq!(res, DesktopInstall::AppImage(temp_appimage.clone()));
let res_fallback = detect_desktop_install_from(None, &temp_appimage).unwrap();
assert_eq!(
res_fallback,
DesktopInstall::AppImage(temp_appimage.clone())
);
let _ = fs::remove_file(&temp_appimage);
let res_deb =
detect_desktop_install_from(None, Path::new("/usr/bin/acaua-gui")).unwrap();
assert_eq!(res_deb, DesktopInstall::Deb);
}
#[cfg(windows)]
{
let res = detect_desktop_install_from(
None,
Path::new("C:/Program Files/Acaua/acaua-gui.exe"),
)
.unwrap();
assert_eq!(res, DesktopInstall::WindowsInstaller);
}
}
#[test]
fn gui_asset_suffix_matches_known_release_naming() {
let appimage = DesktopInstall::AppImage(PathBuf::from("/tmp/Acaua.AppImage"));
assert_eq!(
gui_asset_suffix_for(&appimage, "x86_64").unwrap(),
"_amd64.AppImage"
);
assert_eq!(
gui_asset_suffix_for(&appimage, "aarch64").unwrap(),
"_aarch64.AppImage"
);
let deb = DesktopInstall::Deb;
assert_eq!(gui_asset_suffix_for(&deb, "x86_64").unwrap(), "_amd64.deb");
assert_eq!(gui_asset_suffix_for(&deb, "aarch64").unwrap(), "_arm64.deb");
let win = DesktopInstall::WindowsInstaller;
assert_eq!(
gui_asset_suffix_for(&win, "x86_64").unwrap(),
"_x64-setup.exe"
);
assert!(gui_asset_suffix_for(&appimage, "mips").is_err());
assert!(gui_asset_suffix_for(&deb, "riscv64").is_err());
assert!(gui_asset_suffix_for(&win, "arm").is_err());
fn parses_latest_release_tag() {
let json = r#"{"tag_name": "v0.2.0"}"#;
let release: Release = serde_json::from_str(json).unwrap();
assert_eq!(release.tag_name, "v0.2.0");
}
}
+78 -5
View File
@@ -1,7 +1,11 @@
use crate::error::{AcauaError, Result};
use crate::fusion::Alert;
use crate::pillar1::{self, BeaconingInput};
use crate::storage::Storage;
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
use std::path::Path;
use time::OffsetDateTime;
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub enum ZeekLogKind {
@@ -26,7 +30,10 @@ pub struct AnalysisReport {
pub status: &'static str,
}
pub fn analyze_dir(directory: impl AsRef<Path>) -> Result<AnalysisReport> {
pub fn analyze_dir(
directory: impl AsRef<Path>,
data_dir: impl AsRef<Path>,
) -> Result<AnalysisReport> {
let directory = directory.as_ref();
if !directory.is_dir() {
return Err(AcauaError::InvalidInput(format!(
@@ -34,12 +41,78 @@ pub fn analyze_dir(directory: impl AsRef<Path>) -> Result<AnalysisReport> {
directory.display()
)));
}
let store = Storage::open(data_dir)?;
let stats = store.ingest_dir(directory)?;
let tuples = store.query_7day_tuples()?;
let mut alerts = Vec::new();
for tuple in tuples {
if tuple.timestamps.len() < 20 {
continue;
}
let beacon_score = pillar1::score(&BeaconingInput {
timestamps: tuple.timestamps.clone(),
})?;
if beacon_score.score >= 0.40 {
let latest_ts = tuple.timestamps.last().copied().unwrap_or(0.0);
let latest_date_str = match OffsetDateTime::from_unix_timestamp(latest_ts.floor() as i64)
{
Ok(dt) => {
let format = time::macros::format_description!("[year]-[month]-[day]");
dt.date()
.format(&format)
.unwrap_or_else(|_| "unknown-date".to_string())
}
Err(_) => "unknown-date".to_string(),
};
let mut hasher = Sha256::new();
let raw_id = format!(
"1|{}|{}|{}|{}",
tuple.orig_h, tuple.resp_h, tuple.resp_p, latest_date_str
);
hasher.update(raw_id.as_bytes());
let hash = hasher.finalize();
let mut alert_id = String::with_capacity(16);
for b in &hash[..8] {
use std::fmt::Write;
let _ = write!(alert_id, "{b:02x}");
}
let summary = format!(
"Beaconing detected: {}:{} -> {}:{} (period ~{:.0}s, score {:.3})",
tuple.orig_h,
"any",
tuple.resp_h,
tuple.resp_p,
beacon_score.period_detected_seconds,
beacon_score.score
);
let evidence = serde_json::to_value(&beacon_score)?;
let created_at = OffsetDateTime::now_utc().to_string();
alerts.push(Alert {
id: alert_id,
pillar: 1,
score: beacon_score.score,
summary,
evidence,
created_at,
});
}
}
Ok(AnalysisReport {
directory: directory.display().to_string(),
logs_seen: 0,
records: 0,
alerts: Vec::new(),
status: "skeleton",
logs_seen: stats.files_seen,
records: stats.records_ingested,
alerts,
status: "ok",
})
}
+76
View File
@@ -0,0 +1,76 @@
#separator \x09
#set_separator ,
#empty_field (empty)
#unset_field -
#path conn
#open 2026-10-31-00-00-00
#fields ts uid id.orig_h id.orig_p id.resp_h id.resp_p proto service duration ip_proto
#types time string addr port addr port enum string interval count
1793400000.000000 Cbeacon000 192.168.1.50 49152 198.51.100.1 443 tcp ssl 0.123 6
1793414400.000000 Cbeacon001 192.168.1.50 49152 198.51.100.1 443 tcp ssl 0.123 6
1793420401.205745 Crandom000 192.168.1.50 49153 198.51.100.2 80 tcp http 0.050 6
1793420907.782526 Crandom001 192.168.1.50 49153 198.51.100.2 80 tcp http 0.050 6
1793427340.263808 Crandom002 192.168.1.50 49153 198.51.100.2 80 tcp http 0.050 6
1793428800.000000 Cbeacon002 192.168.1.50 49152 198.51.100.1 443 tcp ssl 0.123 6
1793432391.987521 Crandom003 192.168.1.50 49153 198.51.100.2 80 tcp http 0.050 6
1793443200.000000 Cbeacon003 192.168.1.50 49152 198.51.100.1 443 tcp ssl 0.123 6
1793457600.000000 Cbeacon004 192.168.1.50 49152 198.51.100.1 443 tcp ssl 0.123 6
1793459063.840977 Crandom004 192.168.1.50 49153 198.51.100.2 80 tcp http 0.050 6
1793472000.000000 Cbeacon005 192.168.1.50 49152 198.51.100.1 443 tcp ssl 0.123 6
1793481647.301153 Crandom005 192.168.1.50 49153 198.51.100.2 80 tcp http 0.050 6
1793486400.000000 Cbeacon006 192.168.1.50 49152 198.51.100.1 443 tcp ssl 0.123 6
1793500800.000000 Cbeacon007 192.168.1.50 49152 198.51.100.1 443 tcp ssl 0.123 6
1793515200.000000 Cbeacon008 192.168.1.50 49152 198.51.100.1 443 tcp ssl 0.123 6
1793526193.063144 Crandom006 192.168.1.50 49153 198.51.100.2 80 tcp http 0.050 6
1793528012.111237 Crandom007 192.168.1.50 49153 198.51.100.2 80 tcp http 0.050 6
1793529600.000000 Cbeacon009 192.168.1.50 49152 198.51.100.1 443 tcp ssl 0.123 6
1793538973.034425 Crandom008 192.168.1.50 49153 198.51.100.2 80 tcp http 0.050 6
1793539578.037969 Crandom009 192.168.1.50 49153 198.51.100.2 80 tcp http 0.050 6
1793544000.000000 Cbeacon010 192.168.1.50 49152 198.51.100.1 443 tcp ssl 0.123 6
1793544512.371888 Crandom010 192.168.1.50 49153 198.51.100.2 80 tcp http 0.050 6
1793558400.000000 Cbeacon011 192.168.1.50 49152 198.51.100.1 443 tcp ssl 0.123 6
1793558590.682445 Crandom011 192.168.1.50 49153 198.51.100.2 80 tcp http 0.050 6
1793559128.570519 Crandom012 192.168.1.50 49153 198.51.100.2 80 tcp http 0.050 6
1793563562.403902 Crandom013 192.168.1.50 49153 198.51.100.2 80 tcp http 0.050 6
1793572800.000000 Cbeacon012 192.168.1.50 49152 198.51.100.1 443 tcp ssl 0.123 6
1793584552.243927 Crandom014 192.168.1.50 49153 198.51.100.2 80 tcp http 0.050 6
1793587200.000000 Cbeacon013 192.168.1.50 49152 198.51.100.1 443 tcp ssl 0.123 6
1793600298.829012 Crandom015 192.168.1.50 49153 198.51.100.2 80 tcp http 0.050 6
1793601600.000000 Cbeacon014 192.168.1.50 49152 198.51.100.1 443 tcp ssl 0.123 6
1793605279.357391 Crandom016 192.168.1.50 49153 198.51.100.2 80 tcp http 0.050 6
1793616000.000000 Cbeacon015 192.168.1.50 49152 198.51.100.1 443 tcp ssl 0.123 6
1793623075.531517 Crandom017 192.168.1.50 49153 198.51.100.2 80 tcp http 0.050 6
1793630400.000000 Cbeacon016 192.168.1.50 49152 198.51.100.1 443 tcp ssl 0.123 6
1793644800.000000 Cbeacon017 192.168.1.50 49152 198.51.100.1 443 tcp ssl 0.123 6
1793656230.293402 Crandom018 192.168.1.50 49153 198.51.100.2 80 tcp http 0.050 6
1793656360.692773 Crandom019 192.168.1.50 49153 198.51.100.2 80 tcp http 0.050 6
1793659200.000000 Cbeacon018 192.168.1.50 49152 198.51.100.1 443 tcp ssl 0.123 6
1793673600.000000 Cbeacon019 192.168.1.50 49152 198.51.100.1 443 tcp ssl 0.123 6
1793688000.000000 Cbeacon020 192.168.1.50 49152 198.51.100.1 443 tcp ssl 0.123 6
1793689140.010015 Crandom020 192.168.1.50 49153 198.51.100.2 80 tcp http 0.050 6
1793702400.000000 Cbeacon021 192.168.1.50 49152 198.51.100.1 443 tcp ssl 0.123 6
1793713095.808834 Crandom021 192.168.1.50 49153 198.51.100.2 80 tcp http 0.050 6
1793716800.000000 Cbeacon022 192.168.1.50 49152 198.51.100.1 443 tcp ssl 0.123 6
1793721413.710564 Crandom022 192.168.1.50 49153 198.51.100.2 80 tcp http 0.050 6
1793724793.435926 Crandom023 192.168.1.50 49153 198.51.100.2 80 tcp http 0.050 6
1793731200.000000 Cbeacon023 192.168.1.50 49152 198.51.100.1 443 tcp ssl 0.123 6
1793745600.000000 Cbeacon024 192.168.1.50 49152 198.51.100.1 443 tcp ssl 0.123 6
1793760000.000000 Cbeacon025 192.168.1.50 49152 198.51.100.1 443 tcp ssl 0.123 6
1793774400.000000 Cbeacon026 192.168.1.50 49152 198.51.100.1 443 tcp ssl 0.123 6
1793787823.888895 Crandom024 192.168.1.50 49153 198.51.100.2 80 tcp http 0.050 6
1793788800.000000 Cbeacon027 192.168.1.50 49152 198.51.100.1 443 tcp ssl 0.123 6
1793803200.000000 Cbeacon028 192.168.1.50 49152 198.51.100.1 443 tcp ssl 0.123 6
1793817600.000000 Cbeacon029 192.168.1.50 49152 198.51.100.1 443 tcp ssl 0.123 6
1793832000.000000 Cbeacon030 192.168.1.50 49152 198.51.100.1 443 tcp ssl 0.123 6
1793846400.000000 Cbeacon031 192.168.1.50 49152 198.51.100.1 443 tcp ssl 0.123 6
1793860800.000000 Cbeacon032 192.168.1.50 49152 198.51.100.1 443 tcp ssl 0.123 6
1793875200.000000 Cbeacon033 192.168.1.50 49152 198.51.100.1 443 tcp ssl 0.123 6
1793889600.000000 Cbeacon034 192.168.1.50 49152 198.51.100.1 443 tcp ssl 0.123 6
1793904000.000000 Cbeacon035 192.168.1.50 49152 198.51.100.1 443 tcp ssl 0.123 6
1793918400.000000 Cbeacon036 192.168.1.50 49152 198.51.100.1 443 tcp ssl 0.123 6
1793932800.000000 Cbeacon037 192.168.1.50 49152 198.51.100.1 443 tcp ssl 0.123 6
1793947200.000000 Cbeacon038 192.168.1.50 49152 198.51.100.1 443 tcp ssl 0.123 6
1793961600.000000 Cbeacon039 192.168.1.50 49152 198.51.100.1 443 tcp ssl 0.123 6
1793976000.000000 Cbeacon040 192.168.1.50 49152 198.51.100.1 443 tcp ssl 0.123 6
1793990400.000000 Cbeacon041 192.168.1.50 49152 198.51.100.1 443 tcp ssl 0.123 6
#close 2026-10-31-23-59-59
+4 -3
View File
@@ -7,7 +7,7 @@ fn pillar1_returns_a_skeleton_score() {
};
let score = pillar1::score(&input).unwrap();
assert_eq!(score.observations, 3);
assert_eq!(score.status, "skeleton");
assert_eq!(score.status, "insufficient_data");
}
#[test]
@@ -44,7 +44,7 @@ fn fusion_returns_no_alerts_yet() {
#[test]
fn storage_reports_the_reserved_backend() {
let store = storage::Storage::open(std::env::temp_dir());
let store = storage::Storage::open(std::env::temp_dir()).unwrap();
let summary = store.summary().unwrap();
assert_eq!(summary.ttl_days, 7);
assert_eq!(summary.records, 0);
@@ -52,7 +52,8 @@ fn storage_reports_the_reserved_backend() {
#[test]
fn zeek_rejects_a_missing_directory() {
let error = zeek::analyze_dir("/definitely/not/a/zeek/directory").unwrap_err();
let error =
zeek::analyze_dir("/definitely/not/a/zeek/directory", std::env::temp_dir()).unwrap_err();
assert!(matches!(error, AcauaError::InvalidInput(_)));
}