flowscope 0.25.2

//! Threshold Random Walk port-scan detector (Jung et al.,
//! "Fast Portscan Detection Using Sequential Hypothesis
//! Testing", IEEE S&P 2004).
//!
//! Per-source sequential hypothesis test on connection
//! outcomes. The detector maintains a log-likelihood ratio λ:
//!
//! ```text
//!   on success: λ += log(θ1 / θ0)        ≈ −1.386
//!   on failure: λ += log((1−θ1)/(1−θ0))  ≈ +1.386
//! ```
//!
//! with default priors θ0 = P(success | benign) = 0.8 and
//! θ1 = P(success | scanner) = 0.2.
//!
//! Decision thresholds (α = β = 0.01):
//!
//! ```text
//!   η1 = log((1−β) / α)     ≈ +4.595  → declare scanner
//!   η0 = log(β / (1−α))     ≈ −4.595  → declare benign
//! ```
//!
//! With the defaults, ~4 consecutive failures or ~4 consecutive
//! successes from a fresh source pushes the verdict to
//! `Scanner` or `Benign` respectively. The detector never
//! reuses state across verdicts — once a verdict crosses a
//! threshold, the next `observe` for that key starts at λ = 0.
//! Consumers wanting persistent classification keep their own
//! map.

use std::{collections::HashMap, hash::Hash};

/// Sources tracked before the least-recently-touched is evicted.
/// Matches every sibling detector's default.
const DEFAULT_MAX_SOURCES: usize = 10_000;

/// Per-source scanner-likelihood detector.
///
/// State is bounded: a source is normally dropped the moment its
/// λ crosses either threshold, but a source that stays
/// `Inconclusive` — one SYN from each of a million spoofed
/// addresses, say — would otherwise persist forever. At
/// `max_sources` the least-recently-touched entry is evicted.
/// Evicting an `Inconclusive` source is harmless by the
/// detector's own semantics: it restarts at λ = 0, exactly as it
/// would after a verdict.
pub struct PortScanDetector<K>
where
    K: Hash + Eq + Clone,
{
    sources: HashMap<K, ScannerState>,
    success_step: f64,
    failure_step: f64,
    threshold_scanner: f64,
    threshold_benign: f64,
    max_sources: usize,
    /// Monotonic counter stamped into each touched entry. Cheaper
    /// than a `Timestamp` and, unlike one, it does not force a
    /// clock into `observe`'s signature.
    clock: u64,
    evicted: u64,
}

#[derive(Debug, Clone, Copy)]
struct ScannerState {
    log_likelihood: f64,
    n_observed: u32,
    last_touch: u64,
}

/// Verdict from a single observation.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub enum ScanVerdict {
    /// Threshold crossed in the scanner direction.
    Scanner,
    /// Threshold crossed in the benign direction.
    Benign,
    /// Neither threshold crossed yet.
    Inconclusive,
}

#[derive(Debug, Clone)]
#[non_exhaustive]
pub struct ScanScore<K> {
    pub key: K,
    pub verdict: ScanVerdict,
    pub log_likelihood: f64,
    pub n_observed: u32,
}

impl<K> PortScanDetector<K>
where
    K: Hash + Eq + Clone,
{
    /// Default-tuned: θ0=0.8, θ1=0.2, α=β=0.01.
    pub fn new() -> Self {
        Self::with_params(0.8, 0.2, 0.01, 0.01)
    }

    /// Custom-tuned: caller supplies success priors + error
    /// rates. Asserts the parameters are in `(0, 1)` open
    /// intervals.
    pub fn with_params(theta0: f64, theta1: f64, alpha: f64, beta: f64) -> Self {
        assert!(
            (0.0..1.0).contains(&theta0) && theta0 > 0.0,
            "theta0 must be in (0, 1)"
        );
        assert!(
            (0.0..1.0).contains(&theta1) && theta1 > 0.0,
            "theta1 must be in (0, 1)"
        );
        assert!(
            (0.0..1.0).contains(&alpha) && alpha > 0.0,
            "alpha must be in (0, 1)"
        );
        assert!(
            (0.0..1.0).contains(&beta) && beta > 0.0,
            "beta must be in (0, 1)"
        );
        Self {
            sources: HashMap::new(),
            success_step: (theta1 / theta0).ln(),
            failure_step: ((1.0 - theta1) / (1.0 - theta0)).ln(),
            threshold_scanner: ((1.0 - beta) / alpha).ln(),
            threshold_benign: (beta / (1.0 - alpha)).ln(),
            max_sources: DEFAULT_MAX_SOURCES,
            clock: 0,
            evicted: 0,
        }
    }

    /// Cap on concurrently tracked sources (default 10 000).
    ///
    /// Only sources still awaiting a verdict occupy a slot — one
    /// that crosses either threshold is released immediately. Size
    /// this to the number of *distinct hosts mid-evaluation* you
    /// expect, not to your host count.
    #[must_use]
    pub fn with_capacity(mut self, max_sources: usize) -> Self {
        self.max_sources = max_sources;
        self
    }

    /// Sources dropped by the capacity bound rather than by a
    /// verdict. A climbing count means scan-like traffic from more
    /// distinct sources than the detector can hold — itself worth
    /// alerting on.
    pub fn evicted(&self) -> u64 {
        self.evicted
    }

    /// Make room for one more source. O(n), and only on overflow —
    /// the same trade the DNS correlator makes.
    fn evict_lru(&mut self) {
        if self.sources.len() < self.max_sources {
            return;
        }
        if let Some(victim) = self
            .sources
            .iter()
            .min_by_key(|(_, s)| s.last_touch)
            .map(|(k, _)| k.clone())
        {
            self.sources.remove(&victim);
            self.evicted += 1;
        }
    }

    /// Record a connection outcome.
    ///
    /// `success = true` for completed 3-way handshakes (or any
    /// other "connection succeeded" signal the consumer
    /// supplies); `false` for SYN-only / RST / timeout.
    pub fn observe(&mut self, key: K, success: bool) -> ScanScore<K> {
        if !self.sources.contains_key(&key) {
            self.evict_lru();
        }
        self.clock += 1;
        let clock = self.clock;
        let entry = self.sources.entry(key.clone()).or_insert(ScannerState {
            log_likelihood: 0.0,
            n_observed: 0,
            last_touch: clock,
        });
        entry.last_touch = clock;
        entry.log_likelihood += if success {
            self.success_step
        } else {
            self.failure_step
        };
        entry.n_observed += 1;
        let verdict = if entry.log_likelihood >= self.threshold_scanner {
            ScanVerdict::Scanner
        } else if entry.log_likelihood <= self.threshold_benign {
            ScanVerdict::Benign
        } else {
            ScanVerdict::Inconclusive
        };
        let score = ScanScore {
            key: key.clone(),
            verdict,
            log_likelihood: entry.log_likelihood,
            n_observed: entry.n_observed,
        };
        if matches!(verdict, ScanVerdict::Scanner | ScanVerdict::Benign) {
            self.sources.remove(&key);
        }
        score
    }

    /// Drop per-key state without emitting a verdict.
    pub fn forget(&mut self, key: &K) {
        self.sources.remove(key);
    }

    /// Number of sources currently being tracked.
    pub fn tracked(&self) -> usize {
        self.sources.len()
    }
}

#[cfg(feature = "tracker")]
impl<K> ScanScore<K>
where
    K: crate::KeyFields + Clone,
{
    /// Convert into the canonical [`OwnedAnomaly`](crate::OwnedAnomaly) shape with
    /// the given timestamp. Severity follows `verdict`:
    /// `Scanner` → `Warning`, `Benign`/`Inconclusive` → `Info`.
    ///
    /// Metrics emitted: `log_likelihood`, `n_observed`.
    /// Observations emitted: `verdict` (slug).
    ///
    /// Gated on the `tracker` feature — `OwnedAnomaly` lives
    /// there.
    pub fn into_anomaly(self, ts: crate::Timestamp) -> crate::OwnedAnomaly {
        let severity = match self.verdict {
            ScanVerdict::Scanner => crate::event::Severity::Warning,
            ScanVerdict::Benign | ScanVerdict::Inconclusive => crate::event::Severity::Info,
        };
        let verdict_slug = match self.verdict {
            ScanVerdict::Scanner => "scanner",
            ScanVerdict::Benign => "benign",
            ScanVerdict::Inconclusive => "inconclusive",
        };
        crate::OwnedAnomaly::new(crate::DetectorKind::PortScanTrw, severity, ts)
            .with_key(&self.key)
            .with_observation("verdict", verdict_slug)
            .with_metric("log_likelihood", self.log_likelihood)
            .with_metric("n_observed", self.n_observed as f64)
    }
}

#[cfg(feature = "tracker")]
impl<K> crate::DetectorScore for ScanScore<K>
where
    K: crate::KeyFields + Clone,
{
    fn kind(&self) -> crate::DetectorKind {
        crate::DetectorKind::PortScanTrw
    }

    fn into_anomaly(self, ts: crate::Timestamp) -> crate::OwnedAnomaly {
        self.into_anomaly(ts)
    }
}

impl<K> Default for PortScanDetector<K>
where
    K: Hash + Eq + Clone,
{
    fn default() -> Self {
        Self::new()
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn fresh_source_is_inconclusive_after_one_observation() {
        let mut d: PortScanDetector<u32> = PortScanDetector::new();
        let s = d.observe(1, true);
        assert_eq!(s.verdict, ScanVerdict::Inconclusive);
    }

    #[test]
    fn consecutive_failures_eventually_classify_scanner() {
        let mut d: PortScanDetector<u32> = PortScanDetector::new();
        let mut classifications = Vec::new();
        for _ in 0..8 {
            classifications.push(d.observe(1, false).verdict);
        }
        // With default thresholds (~4.595) and per-failure step
        // (~+1.386), 4 failures push λ to ~5.544 > 4.595.
        assert!(
            classifications.contains(&ScanVerdict::Scanner),
            "expected Scanner verdict after enough failures, got: {classifications:?}"
        );
    }

    #[test]
    fn consecutive_successes_eventually_classify_benign() {
        let mut d: PortScanDetector<u32> = PortScanDetector::new();
        let mut classifications = Vec::new();
        for _ in 0..8 {
            classifications.push(d.observe(1, true).verdict);
        }
        assert!(
            classifications.contains(&ScanVerdict::Benign),
            "expected Benign verdict after enough successes, got: {classifications:?}"
        );
    }

    #[test]
    fn mixed_outcomes_stay_inconclusive() {
        let mut d: PortScanDetector<u32> = PortScanDetector::new();
        for i in 0..20 {
            let s = d.observe(1, i % 2 == 0);
            assert_eq!(
                s.verdict,
                ScanVerdict::Inconclusive,
                "step {i}: λ={}, expected Inconclusive",
                s.log_likelihood
            );
        }
    }

    #[test]
    fn custom_priors_change_step_magnitude() {
        let mut strict: PortScanDetector<u32> = PortScanDetector::with_params(0.9, 0.1, 0.01, 0.01);
        // With θ0=0.9, θ1=0.1, failure step is log(0.9/0.1)
        // ≈ +2.197 — single failure shouldn't trigger.
        let s = strict.observe(1, false);
        assert!(s.log_likelihood < 4.595);
        assert_eq!(s.verdict, ScanVerdict::Inconclusive);
    }

    #[test]
    fn per_key_isolation() {
        let mut d: PortScanDetector<u32> = PortScanDetector::new();
        // Source 1: scanner pattern.
        for _ in 0..4 {
            d.observe(1, false);
        }
        // Source 2: still fresh; should be unaffected.
        let s2 = d.observe(2, true);
        assert_eq!(s2.verdict, ScanVerdict::Inconclusive);
    }

    #[test]
    fn verdict_resets_state() {
        let mut d: PortScanDetector<u32> = PortScanDetector::new();
        // Drive to a Scanner verdict by observing failures until
        // the threshold is crossed (3 inconclusive + 1 scanner
        // under default tuning).
        let mut crossed = false;
        for _ in 0..6 {
            let s = d.observe(1, false);
            if matches!(s.verdict, ScanVerdict::Scanner) {
                crossed = true;
                break;
            }
        }
        assert!(crossed, "should cross Scanner threshold");
        // Right after the verdict, state for this key is gone.
        assert_eq!(d.tracked(), 0);
        // Next observation starts from a clean slate.
        let s = d.observe(1, true);
        assert_eq!(s.n_observed, 1);
        assert_eq!(s.verdict, ScanVerdict::Inconclusive);
    }

    #[test]
    fn forget_drops_state() {
        let mut d: PortScanDetector<u32> = PortScanDetector::new();
        d.observe(1, false);
        d.observe(1, false);
        assert_eq!(d.tracked(), 1);
        d.forget(&1);
        assert_eq!(d.tracked(), 0);
    }

    /// Issue #187: a spoofed-source flood must not grow the map.
    ///
    /// Sources are normally released the moment a verdict crosses a
    /// threshold, but one SYN each from a million addresses leaves
    /// every entry `Inconclusive` — and therefore resident forever.
    #[test]
    fn inconclusive_sources_are_bounded() {
        let mut d: PortScanDetector<u32> = PortScanDetector::new().with_capacity(64);
        // One observation each: never enough to reach a verdict.
        for src in 0..100_000u32 {
            assert_eq!(d.observe(src, false).verdict, ScanVerdict::Inconclusive);
            assert!(d.tracked() <= 64, "cap held at every step");
        }
        assert_eq!(d.tracked(), 64);
        assert_eq!(d.evicted(), 100_000 - 64);
    }

    /// Eviction takes the source that has gone longest without being
    /// observed, so a scan in progress is not displaced by
    /// background noise.
    #[test]
    fn eviction_takes_the_least_recently_touched_source() {
        let mut d: PortScanDetector<u32> = PortScanDetector::new().with_capacity(4);
        for src in 0..4u32 {
            d.observe(src, false);
        }
        // Keep source 0 warm, let 1..4 go cold.
        d.observe(0, false);
        // A fifth source evicts source 1, the coldest — not 0.
        d.observe(99, false);
        assert_eq!(d.tracked(), 4);

        // Source 0 kept its accumulated λ across the eviction — it
        // is at two failures, so two more reach the threshold, where
        // a source that had been evicted and restarted would need
        // four.
        assert_eq!(d.observe(0, false).verdict, ScanVerdict::Inconclusive);
        assert_eq!(d.observe(0, false).verdict, ScanVerdict::Scanner);
    }

    /// An evicted source is not mis-scored — it simply restarts,
    /// which is what a verdict does too.
    #[test]
    fn an_evicted_source_restarts_cleanly() {
        let mut d: PortScanDetector<u32> = PortScanDetector::new().with_capacity(1);
        d.observe(1, false);
        d.observe(2, false); // evicts source 1
        let s = d.observe(1, false);
        assert_eq!(s.n_observed, 1, "source 1 starts over");
        assert_eq!(s.verdict, ScanVerdict::Inconclusive);
    }
}