acaua/crates/acaua-core/tests/pillar3_e2e.rs

//! End-to-end Pilar 3: a synthetic Zeek `dns.log` is ingested into DuckDB,
//! read back through the 7-day window and scored by `pillar3`. The tunnel
//! names are synthetic (SplitMix64 base32), never real captures.

use acaua_core::{pillar3, storage::Storage};
use std::fs;

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(0x9E37_79B9_7F4A_7C15);
        let mut z = self.state;
        z = (z ^ (z >> 30)).wrapping_mul(0xBF58_476D_1CE4_E5B9);
        z = (z ^ (z >> 27)).wrapping_mul(0x94D0_49BB_1331_11EB);
        z ^ (z >> 31)
    }
    fn below(&mut self, n: u64) -> u64 {
        self.next_u64() % n
    }
}

const HEADER: &str = "#separator \\x09\n\
#set_separator\t,\n\
#empty_field\t(empty)\n\
#unset_field\t-\n\
#path\tdns\n\
#fields\tts\tuid\tid.orig_h\tid.orig_p\tid.resp_h\tid.resp_p\tproto\ttrans_id\trtt\tquery\tqclass\tqclass_name\tqtype\tqtype_name\trcode\trcode_name\tAA\tTC\tRD\tRA\tZ\tanswers\tTTLs\trejected\topcode\topcode_name\n";

fn dns_line(ts: f64, uid: &str, host: &str, port: u16, query: &str) -> String {
    format!(
        "{ts:.6}\t{uid}\t{host}\t{port}\t10.0.2.3\t53\tudp\t1\t0.001\t{query}\t1\tC_INET\t16\tTXT\t0\tNOERROR\tF\tF\tT\tT\t0\t-\t-\tF\t0\tQUERY\n"
    )
}

fn base32_name(rng: &mut SplitMix64) -> String {
    const ALPHABET: &[u8] = b"abcdefghijklmnopqrstuvwxyz234567";
    let labels = 2 + rng.below(2) as usize;
    let mut parts = Vec::new();
    for _ in 0..labels {
        let len = 50 + rng.below(14) as usize;
        parts.push(
            (0..len)
                .map(|_| ALPHABET[rng.below(32) as usize] as char)
                .collect::<String>(),
        );
    }
    parts.join(".")
}

fn write_dns_log(dir: &std::path::Path, lines: &[String]) {
    let mut log = String::from(HEADER);
    for line in lines {
        log.push_str(line);
    }
    fs::write(dir.join("dns.log"), log).unwrap();
}

#[test]
#[allow(clippy::inconsistent_digit_grouping)]
fn synthetic_tunnel_is_ingested_and_alerted() {
    let temp = tempfile::tempdir().unwrap();
    let logs = temp.path().join("logs");
    fs::create_dir_all(&logs).unwrap();
    let store = Storage::open(temp.path().join("data")).unwrap();

    let mut rng = SplitMix64::new(2026_10_31);
    let mut lines = Vec::new();
    for i in 0..100 {
        lines.push(dns_line(
            1_800_000_000.0 + i as f64 * 30.0,
            &format!("C{i}"),
            "10.66.66.66",
            50000 + i,
            &format!("{}.t.example-tunnel.net", base32_name(&mut rng)),
        ));
    }
    write_dns_log(&logs, &lines);

    let stats = store.ingest_dns_directory(&logs).unwrap();
    assert_eq!(stats.files_seen, 1);
    assert_eq!(stats.records_ingested, 100);

    let queries = store.query_window_dns().unwrap();
    assert_eq!(queries.len(), 100);

    let scores = pillar3::score(&pillar3::ProtocolInput { queries }).unwrap();
    assert_eq!(scores.len(), 1);
    let score = &scores[0];
    assert_eq!(score.orig_h, "10.66.66.66");
    assert_eq!(score.base_domain, "example-tunnel.net");
    assert_eq!(score.status, "ok");
    assert!(
        score.score >= 0.40,
        "score {} bits {}",
        score.score,
        score.information_bits
    );
    assert_eq!(score.weights.information, 1.0);
    assert_eq!(score.weights.jsd, 0.0);
    assert_eq!(score.jsd_reference_groups, 0);
    assert_eq!(score.jsd, 0.0);

    let alerts = pillar3::alerts(&scores).unwrap();
    assert_eq!(alerts.len(), 1);
    assert_eq!(alerts[0].pillar, 3);
    assert_eq!(alerts[0].id.len(), 16);
    assert_eq!(alerts[0].score, score.score);
    let evidence: serde_json::Value = alerts[0].evidence.clone();
    assert_eq!(evidence["weights"]["jsd"].as_f64(), Some(0.0));
    assert_eq!(evidence["weights"]["information"].as_f64(), Some(1.0));
}

#[test]
fn synthetic_cloudfront_stays_below_the_alert_threshold() {
    let temp = tempfile::tempdir().unwrap();
    let logs = temp.path().join("logs");
    fs::create_dir_all(&logs).unwrap();
    let store = Storage::open(temp.path().join("data")).unwrap();

    let mut rng = SplitMix64::new(7);
    let mut lines = Vec::new();
    const ALPHABET: &[u8] = b"abcdefghijklmnopqrstuvwxyz0123456789";
    for i in 0..45 {
        let name: String = (0..14)
            .map(|_| ALPHABET[rng.below(ALPHABET.len() as u64) as usize] as char)
            .collect();
        lines.push(dns_line(
            1_800_000_000.0 + i as f64 * 60.0,
            &format!("C{i}"),
            "10.0.2.15",
            41000 + i,
            &format!("{name}.cloudfront.net"),
        ));
    }
    write_dns_log(&logs, &lines);
    store.ingest_dns_directory(&logs).unwrap();

    let queries = store.query_window_dns().unwrap();
    let scores = pillar3::score(&pillar3::ProtocolInput { queries }).unwrap();
    let alerts = pillar3::alerts(&scores).unwrap();
    assert!(alerts.is_empty());
    assert_eq!(scores.len(), 1);
    assert_eq!(scores[0].score, 0.0);
}

#[test]
fn analyze_dir_emits_a_pillar3_alert_without_any_conn_log() {
    let temp = tempfile::tempdir().unwrap();
    let logs = temp.path().join("logs");
    let data = temp.path().join("data");
    fs::create_dir_all(&logs).unwrap();

    let mut rng = SplitMix64::new(99);
    let mut lines = Vec::new();
    for i in 0..100 {
        lines.push(dns_line(
            1_800_000_000.0 + i as f64 * 30.0,
            &format!("C{i}"),
            "10.66.66.68",
            52000 + i,
            &format!("{}.t.example-tunnel.net", base32_name(&mut rng)),
        ));
    }
    write_dns_log(&logs, &lines);

    let report = acaua_core::zeek::analyze_dir(&logs, &data).unwrap();
    assert_eq!(report.alerts.len(), 1);
    assert_eq!(report.alerts[0].pillar, 3);
    assert!(report.alerts[0]
        .summary
        .contains("High-information DNS subdomains"));
    assert_eq!(report.alerts[0].id.len(), 16);
}