# `flowscope::detect::patterns` — named detectors
Three canonical security FAQ detectors, packaged as parameter-
tuned named types so consumers stop reinventing them. Each
returns a typed **score** (not a verdict) so policy stays in
the consumer's hands.
Always-on. No new runtime deps. No feature gate. New in 0.12
(plan 143).
## `PortScanDetector<K>` — Threshold Random Walk
Per-source scanner-likelihood detector implementing the
sequential hypothesis test from Jung, Paxson, Berger, and
Balakrishnan, "Fast Portscan Detection Using Sequential
Hypothesis Testing" (IEEE S&P 2004).
Per source `K` (typically `IpAddr`), maintain a log-likelihood
ratio λ. On each observed connection outcome:
```text
success → λ += log(θ1 / θ0) ≈ −1.386
failure → λ += log((1−θ1)/(1−θ0)) ≈ +1.386
```
Verdict thresholds (α = β = 0.01 false-positive / false-
negative rates):
```text
η1 = log((1−β) / α) ≈ +4.595 → Scanner
η0 = log(β / (1−α)) ≈ −4.595 → Benign
```
State **resets** the moment a verdict crosses a threshold — the
next `observe(k, …)` starts fresh at λ = 0. Consumers wanting
persistent classification keep their own map keyed by
`ScanScore::key`.
```rust,ignore
use flowscope::detect::patterns::{PortScanDetector, ScanVerdict};
let mut d: PortScanDetector<std::net::IpAddr> = PortScanDetector::new();
let s = d.observe(client_ip, three_way_handshake_completed);
match s.verdict {
ScanVerdict::Scanner => alert(s.key),
ScanVerdict::Benign => whitelist(s.key),
ScanVerdict::Inconclusive => {}
}
```
### Tuning
`with_params(theta0, theta1, alpha, beta)` for stricter or
laxer priors. Smaller α tightens the Scanner threshold;
smaller β tightens the Benign threshold.
### Limitations
- Misses **distributed** scans (many sources, few connections
each). Pair with a separate cross-source aggregator if
you care about that.
- Legitimate **fan-out** sources (CDN edges, asset scanners
like Shodan / Censys) need an allowlist on the consumer
side — the detector classifies them as Scanner.
## `BeaconDetector<K>` — coefficient-of-variation periodicity
Per-key rolling window of (timestamp, bytes) tuples. Score
range 0.0 (no beacon) → 1.0 (perfect beacon). Composite
RITA-style:
```text
score = 0.5 · (1 − CV_dt)
+ 0.3 · (1 − CV_bytes)
+ 0.2 · duration_bonus
```
Where:
- `CV_dt` = stddev / mean of inter-arrival intervals.
- `CV_bytes` = stddev / mean of payload byte counts.
- `duration_bonus` = `min(1, window_span / 30 min)` —
rewards beacons that sustain over time.
**Suppresses chatty short-lived flows**: requires `n ≥ 10`
observations AND mean interval ∈ `[10 s, 24 h]` before
returning `Some(…)`.
```rust,ignore
use flowscope::detect::patterns::BeaconDetector;
use flowscope::extract::FiveTupleKey;
let mut d: BeaconDetector<FiveTupleKey> = BeaconDetector::new();
if let Some(score) = d.observe(key, packet_ts, packet_bytes) {
if score.score > 0.85 {
eprintln!("beacon: {:?} every {:?}", score.key, score.mean_interval);
}
}
```
### Tuning
- `with_window(n)` — rolling window size; default 20.
- `with_interval_range(min, max)` — gate; default 10 s – 24 h.
### Reference
ActiveCM RITA: <https://github.com/activecm/rita> — the
composite-score weights and threshold methodology this
implementation follows.
## `DgaScorer` — bigram log-likelihood
Mean per-bigram log-likelihood over a 37-class alphabet
(a-z + 0-9 + `-`). The bigram probability table is compiled
once at first use from an embedded ~80-domain English baseline
(google / facebook / amazon / etc.), Laplace-smoothed
(pseudo-count 1).
```text
Higher (less negative) → more "natural" English-like. Lower →
more DGA-like. Default verdict threshold:
`log_likelihood < -3.5`.
```rust,ignore
use flowscope::detect::patterns::DgaScorer;
let scorer = DgaScorer::new();
for qname in dns_queries {
let sld = strip_tld(&qname);
let score = scorer.score(&sld);
if score.log_likelihood < -3.5 {
flag(qname, score);
}
}
```
`DgaScore` also surfaces auxiliary features useful for
composite scoring:
| `length` | SLD length post-classification |
| `vowel_ratio` | vowels / alphabetic chars |
| `digit_ratio` | digits / total |
| `max_consonant_run` | longest run of consonants |
| `char_entropy` | Shannon entropy bits/char |
### Tuning
- `is_dga_with_threshold(sld, threshold)` overrides the −3.5
default.
### Limitations
- **Misses dictionary-based DGAs** (Gozi, Suppobox) — these
use real-word concatenation and score in the same band as
legitimate domains. Pair with the auxiliary features
(length > 12 + 0 digits + low entropy is a signal) or use
a labelled-supervised classifier downstream.
- Embedded baseline is small (~80 SLDs). Consumers wanting
tight FPR control compute their own bigram table from a
representative SLD list (Tranco, internal allowlist) and
ship a custom scorer.
- TLD stripping is the **caller's** job. The shipped
`dga_finder` example uses a naive "last-2-labels minus the
rightmost" heuristic; for strict eTLD+1 handling pair with
the `publicsuffix` crate.
## Emitting detector output via `OwnedAnomaly` (0.13)
Every shipped detector's score implements
[`flowscope::DetectorScore`] and has an inherent
`into_anomaly(ts: Timestamp) -> OwnedAnomaly` method that
converts it to the canonical [`flowscope::OwnedAnomaly`] shape:
- **`ScanScore<K>`** → `DetectorKind::PortScanTrw`
(slug `"PortScanTRW"`); verdict
observation + `log_likelihood` + `n_observed` metrics;
severity follows verdict (`Scanner` → `Warning`,
`Benign`/`Inconclusive` → `Info`).
- **`BeaconScore<K>`** → `DetectorKind::BeaconCv`
(slug `"BeaconCv"`); `score` + `cv_dt`
+ `cv_bytes` + `mean_interval_secs` + `n` metrics; severity
`Warning` (beacon only emits when it has a hit).
- **`DgaScore`** → `DetectorKind::Dga` (slug
`"DgaScorer"`); six feature metrics
(`log_likelihood`, `length`, `vowel_ratio`, `digit_ratio`,
`max_consonant_run`, `char_entropy`); severity `Info`
(consumer thresholds on `log_likelihood` to escalate).
DGA is keyless on the detector side but consumers usually have
the DNS-query flow context — the inherent method takes
`Option<&dyn KeyFields>` for the flow key. The `DetectorScore`
trait impl is the keyless convenience (`into_anomaly(ts)`
delegates to `into_anomaly(ts, None)`).
End-to-end emit example:
```rust,ignore
use flowscope::detect::patterns::PortScanDetector;
use flowscope::emit::EveJsonWriter;
use flowscope::extract::FiveTupleKey;
let mut port_scan: PortScanDetector<FiveTupleKey> =
PortScanDetector::new();
let mut eve = EveJsonWriter::new(std::io::stdout());
for (key, success) in connection_outcomes {
let score = port_scan.observe(key, success);
eve.write_owned_anomaly(&score.into_anomaly(ts))?;
}
```
For generic-over-detector routing, write the function bound
on `DetectorScore`:
```rust,ignore
use flowscope::{DetectorScore, Timestamp};
use flowscope::emit::EveJsonWriter;
use std::io::Write;
fn emit<S: DetectorScore, W: Write>(
eve: &mut EveJsonWriter<W>,
score: S,
ts: Timestamp,
) -> std::io::Result<()> {
eve.write_owned_anomaly(&score.into_anomaly(ts))
}
```
For custom detectors, implement `DetectorScore` on your score
type and you get the same uniform emit path:
```rust,ignore
use flowscope::{DetectorKind, DetectorScore, OwnedAnomaly, Timestamp};
use flowscope::event::Severity;
struct MyScore { hits: u32, max_burst: u32 }
impl DetectorScore for MyScore {
fn kind(&self) -> DetectorKind { DetectorKind::Other("MyCustomDetector") }
fn into_anomaly(self, ts: Timestamp) -> OwnedAnomaly {
OwnedAnomaly::new(DetectorKind::Other("MyCustomDetector"), Severity::Warning, ts)
.with_metric("hits", self.hits as f64)
.with_metric("max_burst", self.max_burst as f64)
}
}
```
See [`docs/eve-format.md`](eve-format.md) §"Custom anomaly
emission via `OwnedAnomaly` (0.13)" for the EVE schema mapping
and [`docs/recipes.md`](recipes.md) §"0.13 patterns" for the
end-to-end recipe.
## Unified `Detector` trait + `DetectorRegistry` (0.21, #131)
Registering detectors on a [`DetectorRegistry`] drives a
heterogeneous set from **one** call per event, instead of
hand-wiring each detector's feed/drain loop. The registry keeps
the genuinely-different observe inputs but unifies dispatch:
```rust,ignore
use flowscope::detect::{DetectorRegistry, DgaDetector, HostPair, SrcHost};
use flowscope::detect::patterns::{BeaconDetector, PortScanDetector};
use flowscope::extract::FiveTupleKey;
let mut registry: DetectorRegistry<FiveTupleKey> = DetectorRegistry::new();
registry
.register(BeaconDetector::<HostPair>::new()) // aggregates per (src, dst, port)
.register(PortScanDetector::<SrcHost>::new()) // aggregates per source IP
.register(DgaDetector::new()); // feeds off DNS query names
// per flow event (from FlowTracker or the typed Driver):
registry.observe(&flow_event, &mut anomalies); // or observe_event(&driver_event, …)
// per DNS query name (from a DNS slot drain) — the message's orientation
// says which endpoint sent the query:
registry.observe_dns(&msg.key, msg.orientation, qname, ts, &mut anomalies);
// periodically:
registry.evict_expired(now);
```
Each detector implements the [`Detector`] trait — a set of
defaulted, no-op **lifecycle hooks** (`on_flow_start`,
`on_flow_end`, `on_dns_query`, …), so a detector implements only
the feeds it consumes. Anomalies append to a caller-owned
`Vec<OwnedAnomaly>` (the `track_into` idiom — zero per-event
allocation, no unbounded internal queues).
**Who is the source?** A bidirectional flow key is address-sorted
(`FiveTupleKey` keeps `a < b`), so `key.src_ip()` is the *lower
address*, not the host that opened the flow. The hooks say which
endpoint acted: `on_flow_start(key, initiator, ..)` gets the
initiator's orientation, `on_dns_query(key, sender, ..)` the query
sender's, and flow-end / tick hooks have `stats.initiator_orientation`.
Read the key through [`Directed`](crate::Directed) —
`Directed::new(key, initiator)` / `Directed::initiator(key, stats)` —
to get the acting host as the source. Every shipped detector does
(since 0.26; before, a scanner with a higher address than its targets
was reported as each target).
**Derived aggregation keys.** Registry detectors see the *flow*
key `K`, but beaconing and scanning aggregate *across* flows
(every beacon ping is its own ephemeral 5-tuple). The shipped
[`SrcHost`] (source IP) and [`HostPair`] (`src`, `dst`,
`dst_port`) keys implement `KeyFields + Hash + Eq`, derive from
any flow key via `from_key`, and replace the `SrcIpKey` newtype
consumers used to re-declare.
**Actionable-only emission.** The shipped impls gate on
per-detector thresholds + cooldowns
(`BeaconDetector::with_anomaly_threshold` / `with_cooldown`;
port-scan emits only on a `Scanner` verdict;
`DgaDetector::with_threshold` + per-`(src, domain)` suppression),
so a registry drain is alert-shaped, not score-shaped. Call the
detectors' raw `observe` methods directly when you want every
score.
**Port-scan success signal** is derived statelessly at flow end:
for TCP, the responder answered the handshake (`'s'` in the
Zeek-style [`HistoryString`]); for UDP, any responder packet.
An unanswered SYN or a RST refusal both read as failure.
Every emitted [`OwnedAnomaly`] carries its typed
[`DetectorKind`] (0.21, #133) — so `kind.attack_technique()`
gives the MITRE ATT&CK technique ID and `EveJsonWriter` emits
`anomaly.attack_technique` automatically.
## NDR detectors (0.21, #132)
Four network-detection-and-response detectors built on the
[`correlate`](crate::correlate) streaming primitives, each a
[`Detector`] you can drop into a [`DetectorRegistry`] or drive
standalone via its `observe` method. All take **pre-extracted**
values (an `IpAddr`, a `&str` qname, a byte count), so `detect`
stays independent of the `dns` Cargo feature, and all bound their
state + gate emission behind thresholds and cooldowns (a drain is
alert-shaped, not score-shaped).
| [`DnsTunnelDetector`] | `on_dns_query` | `TimeBucketedSet<(IpAddr, domain), u64>` — distinct hashed subdomains | ≥ 50 distinct ≥ 50-byte qnames under one registered domain in 300 s | T1071.004 |
| [`NewlyObservedDomainDetector`] | `on_dns_query` | `FirstSeen<String>` (cap 100 k, 7-day TTL) | first sight of a registered domain, after a 600 s warmup | T1568 |
| [`ConnectionFloodDetector`] | `on_flow_start` | `TimeBucketedCounter<IpAddr>` | ≥ 100 new flows per source per 10 s | T1498 |
| [`DataExfilDetector`] | `on_flow_end` | `EwmaVar<IpAddr>` over `bytes_initiator` | flow > mean + 3σ of the source's own baseline, ≥ 10 samples, ≥ 1 MiB floor | T1048 |
Notes and gotchas:
- **Registered-domain rollup** uses a PSL-free last-two-labels
rule (`a.b.evil.com` → `evil.com`), the same caveat as
[`detect::risk`](crate::detect::risk) — imperfect for
`co.uk`-style public suffixes; pre-extract or allowlist when
it matters.
- **`DataExfilDetector` needs a non-degenerate baseline.** A
source that sends *exactly* the same byte count every flow has
zero variance, so no z-score exists and it never alarms
(`EwmaVar::zscore` returns 0.0 during zero-variance warmup — by
design, so a cold/flat baseline can't false-positive). Real
traffic jitters; a perfectly constant baseline is a test
artifact. The absolute `min_bytes` floor is the backstop.
- **`DataExfilDetector` observes at flow end, not per tick** —
tick stats are cumulative, so naive per-tick recording would
double-count a flow's bytes. Delta-tracking is a follow-up.
- **`ConnectionFloodDetector` is rate-shaped (DoS, T1498)**;
scan-shaped fan-out across many destinations is
[`PortScanDetector`]'s job (T1046). Both can fire on the same
aggressive source — that's fine, they answer different
questions.
- **NOD is a context signal, not a verdict** (`Info` severity) —
every legitimate first visit is "newly observed". Pair it with
beaconing / reputation / volume.
## Examples
- `examples/03-detection/c2_beacon_finder.rs` — replay a pcap,
flag top beacon candidates.
- `examples/03-detection/dga_finder.rs` — replay a pcap,
score DNS query names, rank suspicious SLDs.
- `examples/03-detection/port_scan_detector.rs` (pre-existing
in 0.10 cycle) — refactor target if you want a
`PortScanDetector` migration recipe.
- `examples/03-detection/detector_registry.rs` (0.21, #131) —
the unified `DetectorRegistry` over a real typed `Driver`:
register beacon / RITA-beacon / port-scan / DGA once, drive
them all from one event stream, emit each anomaly (with its
ATT&CK technique) as EVE.
- `examples/03-detection/dns_tunnel_detector.rs` (0.21, #132) —
the `DnsTunnelDetector` distinct-subdomain heuristic driven
through a registry over a DNS pcap, emitting T1071.004 EVE
anomalies.
## Why these three
The 0.12 strategic review identified these as the canonical
"how do I detect X" FAQ recipes consumers keep rebuilding from
the existing `correlate::TopK` / `Ewma` / `BurstDetector` and
`detect::shannon_entropy` primitives.
Packaging them as named, threshold-tuned APIs:
- saves consumers ~200 LoC per pipeline,
- pins canonical algorithm references (so a future spec drift
surfaces as a documented change, not a silent regression),
- keeps verdict policy in the consumer's hands (we emit
scores, not actions).