bathy_interpret/

rules.rs

//! The confidence ladder, the rule registry, and every protocol rule.
//!
//! # The ladder (AC-4.11)
//!
//! [`Specificity`] is the *only* place a confidence number is written down.
//! Every rule below declares a rung, never a literal `f64` -- so a `0.95`
//! anywhere means the same thing ("product and version both extracted from
//! a self-identifying banner") regardless of which protocol produced it,
//! and the whole table is auditable in one place rather than sprinkled as
//! magic numbers through match arms.
//!
//! # Byte safety (verification beyond the brief)
//!
//! Every matcher below is a plain `fn(&[u8]) -> Option<Hit>` operating on
//! attacker-controlled bytes. None of them may panic, and every
//! [`Hit::span`] they produce must be a valid range into the slice they
//! were handed. Two disciplines make that provable rather than merely
//! asserted:
//!
//! - Text-shaped rules (HTTP, SSH, SMTP) never call
//!   `String::from_utf8_lossy` over the *whole* response and then reuse
//!   `regex`'s match offsets as indices into the original bytes. That
//!   combination is unsound: a lossy conversion can change the byte length
//!   of anything after the first invalid byte (a single invalid byte
//!   becomes a 3-byte U+FFFD), so an offset computed against the lossy
//!   `String` does not necessarily land on the same byte -- or even inside
//!   bounds -- of the original slice. **This is a real defect in this
//!   task's own brief**: its worked `HTTP_NGINX` example does exactly this
//!   (`String::from_utf8_lossy(bytes)` over the whole response, then reuses
//!   `caps.get(0)`'s offsets as `matched_span` into the original `bytes`).
//!   It is also, independently, a second defect against this task's own
//!   dispatch instruction ("compile every regex once via `LazyLock`, not
//!   per call"): the brief's example constructs a fresh `regex::Regex` on
//!   every single invocation of the matcher closure. Both are fixed here:
//!   [`utf8_lines`] validates each line's bytes with `std::str::from_utf8`
//!   (strict, not lossy) *before* that line's bytes are ever used to
//!   compute an offset, skipping a line that isn't valid UTF-8 rather than
//!   guessing at where it ends; and every regex below is a module-level
//!   `LazyLock<Regex>`, compiled once for the life of the process. A
//!   response with a binary body after clean text headers (an ordinary
//!   HTTP reply with an image body, for instance) still gets its headers
//!   matched correctly under this scheme, because invalidity in one line
//!   never poisons another line's offsets.
//! - Binary-shaped rules (Postgres, MySQL, DNS, TLS) use only checked
//!   arithmetic (`checked_add`, slice `.get`) and never index past a bound
//!   they have not just verified, so a truncated, malformed, or hostile
//!   packet yields `None` rather than a panic or an out-of-range span.
//!
//! `crate::interpret::tests` property-tests both disciplines' end result
//! (span validity) over arbitrary bytes for the whole rule set at once, not
//! just per protocol.
//!
//! **Both disciplines were prose until the M7 panic-lint round.** The
//! sentence above about binary-shaped rules using "only checked arithmetic
//! ... and never index past a bound" was *false when written*: `utf8_lines`
//! sliced `bytes[start..i]`, `u16_at` indexed `s[0]`/`s[1]`,
//! `mysql_handshake_v10` sliced `bytes[VERSION_STRING_START..version_end]`
//! after an unchecked `+`, `dns_bind_version` sliced
//! `bytes[txt_start..txt_end]`, `tls_server_hello` indexed `header[0]` and
//! `header[5]`, and every text-shaped rule computed its span with a bare
//! `line_start + m.start()`. Each was in fact in bounds, and none was
//! *checked* by anything except a reader's attention -- which is precisely
//! how the `from_utf8_lossy` offset defect this module's own comment
//! describes got written in the first place, and how seven span-corrupting
//! mutants survived into three review rounds. The Global Constraint that
//! claimed `unwrap`/`expect`/indexing panics were "denied by lint" in this
//! crate was aspirational from M1 until then; `src/lib.rs` now carries the
//! lint that makes it true, and the arithmetic in this file is checked or
//! it does not compile. See [`absolute_span`], which is the one place the
//! offset arithmetic those seven mutants attacked now lives.
//!
//! # Provenance
//!
//! Every rule's `source` names an RFC section, a vendor's own protocol
//! documentation, or a capture this project ran itself in Task 2 of this
//! milestone (image, digest, and observed bytes -- see that task's report,
//! `.superpowers/sdd/2026-07-31-bathy-m4-probes-interpret/task-2-report.md`).
//! `nmap` and `nmap-service-probes`, both present on this development
//! machine, were never opened or consulted while writing any rule below --
//! confirmed structurally, not just by this comment, by
//! `crate::tests::every_rule_documents_its_non_nmap_source`.

use std::ops::Range;
use std::sync::LazyLock;

use bathy_types::confidence::Confidence;
use regex::Regex;

/// The confidence ladder. Every rule declares which rung it sits on, so
/// scores across protocols mean the same thing and are auditable in one
/// table rather than sprinkled as magic numbers through match arms
/// (AC-4.11).
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Specificity {
    /// Product and version both extracted from a self-identifying banner.
    ProductAndVersion,
    /// Product identified, version absent or unparseable.
    ProductOnly,
    /// Protocol confirmed by structure, product unknown.
    ProtocolOnly,
    /// Consistent with the service but not conclusive.
    Weak,
}

impl Specificity {
    /// Every rung, in ladder order.
    ///
    /// Exists so the exhaustiveness of the ladder is testable rather than
    /// eyeballed: `tests::every_rung_of_the_ladder_is_a_valid_confidence`
    /// walks this array, and `tests::the_ladder_array_lists_every_variant`
    /// matches on each variant exhaustively, so adding a fifth rung without
    /// adding it here fails to compile.
    pub const ALL: [Self; 4] = [
        Self::ProductAndVersion,
        Self::ProductOnly,
        Self::ProtocolOnly,
        Self::Weak,
    ];

    /// The `f64` this rung means. Split out from [`Self::confidence`] so the
    /// four numbers can be range-checked by a test without going through the
    /// fallible constructor.
    const fn value(self) -> f64 {
        match self {
            Self::ProductAndVersion => 0.95,
            Self::ProductOnly => 0.85,
            Self::ProtocolOnly => 0.70,
            Self::Weak => 0.50,
        }
    }

    pub fn confidence(self) -> Confidence {
        // NARROW ALLOW (M7 panic-lint round). `expect` is denied crate-wide
        // in non-test builds because this crate parses attacker-controlled
        // bytes; this call site takes no bytes at all. Its argument is
        // `self.value()`, a `match` over a four-variant enum returning one
        // of four literals, every one of them inside `Confidence`'s 0.0..=1.0
        // domain -- so the `Err` arm is unreachable for every value the type
        // system permits. `Confidence`'s field is private, so there is no
        // infallible constructor to reach for, and the alternatives are both
        // worse: `unwrap_or(..)` would silently substitute a confidence
        // nobody wrote, and returning `Result` would push an error nobody can
        // trigger through every rule in the file.
        //
        // The reasoning is enforced, not asserted:
        // `tests::every_rung_of_the_ladder_is_a_valid_confidence` walks
        // `Specificity::ALL` and fails if any rung's value leaves the domain,
        // and `tests::the_ladder_array_lists_every_variant` fails to compile
        // if a rung is added without joining `ALL`.
        #[allow(
            clippy::expect_used,
            reason = "four in-range literals; enforced by every_rung_of_the_ladder_is_a_valid_confidence"
        )]
        Confidence::new(self.value()).expect("ladder values are in range")
    }
}

/// Documentation for one rule, surfaced verbatim by [`explain`] (the
/// `fingerprint.explain` tool's data source in M5).
pub struct RuleDoc {
    pub id: &'static str,
    pub service: &'static str,
    pub specificity: Specificity,
    /// Human-readable explanation of what pattern justified the claim.
    pub rationale: &'static str,
    /// Provenance of this rule. Must cite an RFC, vendor documentation, or
    /// a capture from software run in this project's own lab. Never Nmap.
    pub source: &'static str,
}

/// What a rule's matcher found, before it is wrapped into a public
/// [`crate::interpret::Interpretation`].
pub(crate) struct Hit {
    pub product: Option<String>,
    pub version: Option<String>,
    /// Overrides the rule's own `doc.specificity` when a rule's confidence
    /// genuinely depends on what was found (e.g. product-with-version vs.
    /// product-without-version from the same regex). Rules whose rung never
    /// varies just echo `doc.specificity` here.
    pub specificity: Specificity,
    /// Byte range within the response that justified the claim.
    pub span: Range<usize>,
}

/// One interpretation rule: which probe it applies to, its documentation,
/// and the pure function that decides whether a response matches it.
pub(crate) struct Rule {
    pub probe_id: &'static str,
    pub doc: RuleDoc,
    pub matcher: fn(&[u8]) -> Option<Hit>,
}

/// Every rule applicable to a given probe, in registration order (the order
/// `interpret` iterates them in before its own sort makes order
/// irrelevant).
pub(crate) fn rules_for(probe_id: &str) -> impl Iterator<Item = &'static Rule> {
    ALL_RULES.iter().filter(move |r| r.probe_id == probe_id)
}

/// Every rule's documentation, for exhaustive checks like "no rule cites
/// Nmap" (AC-4.16) and for tools that want to list what this crate can
/// recognize at all.
pub fn all_rules() -> impl Iterator<Item = &'static RuleDoc> {
    ALL_RULES.iter().map(|r| &r.doc)
}

/// Documentation for one rule by id, surfaced by the `fingerprint.explain`
/// tool in M5 (AC-4.12: every rule that can fire must be explainable).
pub fn explain(rule_id: &str) -> Option<&'static RuleDoc> {
    ALL_RULES.iter().map(|r| &r.doc).find(|d| d.id == rule_id)
}

/// Every distinct probe id this crate has at least one rule for -- the
/// "registry" M4 Task 4's replay corpus (`crates/bathy-interpret/tests/replay.rs`)
/// checks each fixture's `probe_id` against, closing that task's own "the
/// corpus is data, so test the data" requirement.
///
/// Deliberately *not* `bathy_probe::framework::ProbeRegistry`'s own id list:
/// depending on `bathy-probe` from this crate, even as a dev-dependency,
/// would contradict this crate's own `src/lib.rs` doc comment (this crate
/// sits *below* `bathy-probe` in the workspace layer order specifically so
/// its tests need no upward dependency at all) and would fail
/// `xtask check-deps`, which inspects a package's dev-dependencies too, not
/// only its normal ones (`find_violations` in `xtask/src/main.rs` does not
/// filter `cargo metadata`'s dependency list by kind). This crate's own rule
/// registry is the authoritative "what probe ids do I know how to interpret"
/// answer from *inside* this crate, which is the only registry `interpret`
/// itself actually consults (see [`rules_for`]) -- a fixture naming a probe
/// id this function doesn't return could never produce a real rule match
/// regardless of what `bathy-probe` itself knows about, so it is exactly the
/// right check for a corpus that exists to regression-test `interpret`.
pub fn known_probe_ids() -> impl Iterator<Item = &'static str> {
    let mut ids: Vec<&'static str> = ALL_RULES.iter().map(|r| r.probe_id).collect();
    ids.sort_unstable();
    ids.dedup();
    ids.into_iter()
}

/// Splits `bytes` on `\n` and returns each line's starting byte offset
/// (relative to `bytes`) together with its content as `&str` -- but only
/// for lines that are themselves valid UTF-8. See this module's doc
/// comment ("Byte safety") for why per-line validation, not a single
/// whole-response `String::from_utf8_lossy`, is what keeps a match's byte
/// offsets valid indices into `bytes` itself.
fn utf8_lines(bytes: &[u8]) -> Vec<(usize, &str)> {
    let mut out = Vec::new();
    let mut start = 0usize;
    for (i, &b) in bytes.iter().enumerate() {
        if b == b'\n' {
            if let Some(Ok(s)) = bytes.get(start..i).map(std::str::from_utf8) {
                out.push((start, s));
            }
            // `i < bytes.len()`, so this cannot overflow -- but it is
            // written checked anyway, because that is exactly the kind of
            // "obviously fine" offset arithmetic this module's own history
            // is about. A `None` here would mean `start` stops advancing,
            // so bail rather than loop on a stale offset.
            let Some(next) = i.checked_add(1) else {
                return out;
            };
            start = next;
        }
    }
    if start < bytes.len()
        && let Some(Ok(s)) = bytes.get(start..).map(std::str::from_utf8)
    {
        out.push((start, s));
    }
    out
}

/// An absolute byte range into the response, from a regex match against one
/// of [`utf8_lines`]'s lines plus that line's own start offset.
///
/// The single home for the `line_start + m.start()` arithmetic that every
/// text-shaped rule below needs. It is one function rather than six copies
/// for the reason this module's "Byte safety" note gives: this exact
/// expression is what the `from_utf8_lossy` defect corrupted and what seven
/// span mutants attacked across three review rounds, and a checked add
/// written six times is six chances to write the seventh unchecked.
///
/// Returns `None` on overflow rather than wrapping or saturating: a
/// saturated span would be a *wrong* claim about which bytes justified an
/// interpretation, and this crate's whole contract is that a span points at
/// the evidence. No match, no claim.
fn absolute_span(line_start: usize, m: &regex::Match<'_>) -> Option<Range<usize>> {
    Some(line_start.checked_add(m.start())?..line_start.checked_add(m.end())?)
}

/// Compiles one of this module's own literal, compile-time-constant regex
/// patterns.
///
/// # Why this panics, and why that is the right behaviour
///
/// `expect` is denied crate-wide in non-test builds (see `src/lib.rs`)
/// because this crate parses attacker-controlled bytes. This function takes
/// no bytes: its only callers pass a `&'static str` literal written in this
/// file, so whether it compiles is decided when the source is written, not
/// by anything a scanned peer sends. There is no input that reaches this
/// `Err` arm.
///
/// The alternative -- returning `Option<Regex>` and having the rule quietly
/// not fire -- is strictly worse than a panic: it would turn a typo in a
/// pattern into a rule that silently recognizes nothing, which is the
/// "reaches nothing while reading as coverage" failure this project has
/// already measured once in a property-test strategy. A `LazyLock` that
/// dies loudly on first use is a bug you find; a rule that never matches is
/// a bug you ship.
///
/// The reasoning is enforced rather than asserted:
/// `tests::every_static_regex_in_this_module_compiles` forces every
/// `LazyLock` below, so a bad pattern fails `cargo test` rather than a scan.
#[allow(
    clippy::expect_used,
    reason = "compile-time-constant patterns only; enforced by every_static_regex_in_this_module_compiles"
)]
fn static_regex(pattern: &'static str) -> Regex {
    Regex::new(pattern).expect("a pattern literal in this module does not compile")
}

/// Reads a big-endian `u16` at `bytes[at..at+2]`, or `None` if that range
/// runs off the end of `bytes`. The one primitive every binary-shaped
/// matcher below builds its bounds-checked parsing on.
fn u16_at(bytes: &[u8], at: usize) -> Option<u16> {
    let s: [u8; 2] = bytes.get(at..at.checked_add(2)?)?.try_into().ok()?;
    Some(u16::from_be_bytes(s))
}

// =====================================================================
// HTTP -- source: RFC 9112 §4 ("Status Line": `status-line = HTTP-version
// SP status-code SP [ reason-phrase ]`), RFC 9110 §10.2.4 (`Server`).
// (Root-cause fix, M4 Task 3 review round 1: this previously cited §3,
// which is "Request Line" -- the ABNF for what a *client* sends, not a
// server's response. §4 is the section that actually defines the
// status-line shape these rules match against.) Corroborated against a
// real server: `docker.io/library/nginx:1.27-alpine`, digest
// `sha256:65645c7bb6a0661892a8b03b89d0743208a18dd2f3f17a54ef4b76fb8e2f2a10`
// (M4 Task 2 report), which replied `HTTP/1.1 200 OK\r\nServer:
// nginx/1.27.5\r\n...`.
// =====================================================================

/// The response's first line that is valid UTF-8 -- not unconditionally its
/// literal first line. A well-formed HTTP status line is always plain
/// ASCII and therefore always valid UTF-8, so for any real HTTP response
/// this distinction is moot: the first valid-UTF-8 line *is* byte 0.
/// [`utf8_lines`] is still what supplies the "valid UTF-8" half of that
/// guarantee, which is why this function is phrased in terms of it rather
/// than indexing `bytes` directly.
fn http_status_line(bytes: &[u8]) -> Option<(usize, &str)> {
    let (start, first) = *utf8_lines(bytes).first()?;
    if first.starts_with("HTTP/") {
        Some((start, first))
    } else {
        None
    }
}

static NGINX_SERVER_RE: LazyLock<Regex> =
    LazyLock::new(|| static_regex(r"(?i)^Server:[ \t]*nginx(?:/([0-9][0-9A-Za-z.\-]*))?"));

fn http_nginx(bytes: &[u8]) -> Option<Hit> {
    http_status_line(bytes)?;
    for (line_start, line) in utf8_lines(bytes) {
        let Some(caps) = NGINX_SERVER_RE.captures(line) else {
            continue;
        };
        let m = caps.get(0)?;
        let version = caps.get(1).map(|v| v.as_str().to_owned());
        let specificity = if version.is_some() {
            Specificity::ProductAndVersion
        } else {
            Specificity::ProductOnly
        };
        return Some(Hit {
            product: Some("nginx".to_owned()),
            version,
            specificity,
            span: absolute_span(line_start, &m)?,
        });
    }
    None
}

fn http_bare_protocol(bytes: &[u8]) -> Option<Hit> {
    let (start, first) = http_status_line(bytes)?;
    Some(Hit {
        product: None,
        version: None,
        specificity: Specificity::ProtocolOnly,
        span: start..start.checked_add(first.len())?,
    })
}

// =====================================================================
// SSH -- source: RFC 4253 §4.2 ("Protocol Version Exchange"). Corroborated
// against `docker.io/linuxserver/openssh-server:latest`, digest
// `sha256:96b9a4d3b5106746d08d43a6911650d4d21f7d5c7f2ac9660e792bdb5e63157c`
// (M4 Task 2 report), which sent `SSH-2.0-OpenSSH_10.3\r\n` unprompted.
//
// Both matchers below scan *every* line, not just the first, and stop at
// the first one that matches. This is not defensive-for-its-own-sake:
// §4.2 itself says "The server MAY send other lines of data before
// sending the version string... Such lines MUST NOT begin with 'SSH-'...
// Clients MUST be able to process such lines." A matcher that only ever
// looked at line 0 would false-negative on exactly this RFC-sanctioned
// case -- a real, spec-compliant server whose banner isn't byte 0. (Root-
// cause fix, M4 Task 3 review round 1: an earlier version of both
// functions here called `utf8_lines(bytes).first()`, which is *always*
// offset 0 by construction -- so it both missed this case and made the
// `line_start + …` term in `Hit::span` provably dead code, indistinguishable
// by any test from a version that dropped the offset entirely. See
// `tests::ssh_openssh_finds_the_identification_line_after_a_preamble_line`.)
// =====================================================================

static SSH_OPENSSH_RE: LazyLock<Regex> =
    LazyLock::new(|| static_regex(r"^SSH-\d\.\d+-OpenSSH_(\S+)"));

static SSH_BANNER_RE: LazyLock<Regex> = LazyLock::new(|| static_regex(r"^SSH-\d\.\d+-"));

fn ssh_openssh(bytes: &[u8]) -> Option<Hit> {
    for (line_start, line) in utf8_lines(bytes) {
        let Some(caps) = SSH_OPENSSH_RE.captures(line) else {
            continue;
        };
        let m = caps.get(0)?;
        let version = caps.get(1)?.as_str().to_owned();
        return Some(Hit {
            product: Some("OpenSSH".to_owned()),
            version: Some(version),
            specificity: Specificity::ProductAndVersion,
            span: absolute_span(line_start, &m)?,
        });
    }
    None
}

fn ssh_bare_protocol(bytes: &[u8]) -> Option<Hit> {
    for (line_start, line) in utf8_lines(bytes) {
        if let Some(m) = SSH_BANNER_RE.find(line) {
            return Some(Hit {
                product: None,
                version: None,
                specificity: Specificity::ProtocolOnly,
                span: absolute_span(line_start, &m)?,
            });
        }
    }
    None
}

// =====================================================================
// PostgreSQL -- source: PostgreSQL's own Frontend/Backend Protocol
// documentation, split across two pages of the same doc set, not one:
//
// - The *request* bytes (an 8-byte message: length 8, then the fixed
//   SSLRequest code 80877103) are "Message Formats" §SSLRequest
//   (<https://www.postgresql.org/docs/current/protocol-message-formats.html>).
//   That page defines what the client sends; it does not document the
//   server's reply at all.
// - The *reply*'s meaning is documented separately, in "Message Flow"
//   §54.2.10 ("SSL Session Encryption" -- "Message Flow" is the page's own
//   title; `protocol-flow` is only its URL slug, corrected in the M4
//   whole-branch fix wave's citation sweep,
//   <https://www.postgresql.org/docs/current/protocol-flow.html>): "The
//   server then responds with a single byte containing S or N, indicating
//   that it is willing or unwilling to perform SSL, respectively." (Root-
//   cause fix, M4 Task 3 review round 1: both rules below previously cited
//   only the request-format page for this fact too -- verified against the
//   live page, which covers the request shape only.)
//
// Corroborated against `docker.io/library/postgres:16-alpine`, digest
// `sha256:57c72fd2a128e416c7fcc499958864df5301e940bca0a56f58fddf30ffc07777`
// (M4 Task 2 report), which replied `N` (run without SSL configured) to
// exactly the 8 bytes `postgres-startup-v1` sends.
// =====================================================================

fn postgres_ssl_accepted(bytes: &[u8]) -> Option<Hit> {
    if bytes == b"S" {
        Some(Hit {
            product: None,
            version: None,
            specificity: Specificity::ProtocolOnly,
            span: 0..1,
        })
    } else {
        None
    }
}

fn postgres_ssl_declined(bytes: &[u8]) -> Option<Hit> {
    if bytes == b"N" {
        Some(Hit {
            product: None,
            version: None,
            specificity: Specificity::ProtocolOnly,
            span: 0..1,
        })
    } else {
        None
    }
}

// =====================================================================
// Redis -- source: Redis's own RESP protocol specification
// (<https://redis.io/docs/latest/develop/reference/protocol-spec/>): a
// simple string reply is `+<text>\r\n`. Corroborated against
// `docker.io/library/redis:7-alpine`, digest
// `sha256:e7723ff73d963f5cc6d9c4643ea3d989527a402a319239054e9472a7fb9219a2`
// (M4 Task 2 report), which replied `+PONG\r\n` to exactly the RESP `PING`
// `redis-ping-v1` sends.
// =====================================================================

fn redis_pong(bytes: &[u8]) -> Option<Hit> {
    let prefix = b"+PONG";
    if bytes.starts_with(prefix) {
        Some(Hit {
            product: None,
            version: None,
            specificity: Specificity::ProtocolOnly,
            span: 0..prefix.len(),
        })
    } else {
        None
    }
}

/// Weak-tier fallback: *some* RESP-shaped reply came back (one of the five
/// type sigils RESP defines -- simple string, error, integer, bulk string,
/// array), but not literally `+PONG`. Genuinely weaker evidence than
/// [`redis_pong`]: several Redis-protocol-compatible servers (e.g. KeyDB,
/// Dragonfly) reply to `PING` with a valid but non-identical RESP value, so
/// this recognizes the *wire format*, not the product -- exactly
/// [`Specificity::Weak`]'s definition ("consistent with the service but not
/// conclusive"), not a guess that it is Redis itself.
///
/// Requires an actual `\r\n` terminator (RESP's own line terminator, per
/// the RESP protocol specification's "Simple strings" section: "terminated
/// by CRLF") after the sigil, not just a matching first byte. (Root-cause
/// fix, M4 Task 3 review round 1: a single stray byte from an arbitrary
/// binary protocol -- `0x2b` alone, say -- happens to equal `+` and
/// previously matched on its own; requiring the terminator this rule's own
/// rationale claims to have found is what makes "RESP-shaped" an honest
/// description rather than a one-byte coincidence.)
fn redis_resp_shaped_reply(bytes: &[u8]) -> Option<Hit> {
    let sigil = *bytes.first()?;
    if !matches!(sigil, b'+' | b'-' | b':' | b'$' | b'*') {
        return None;
    }
    let crlf_at = bytes.windows(2).position(|w| w == b"\r\n")?;
    Some(Hit {
        product: None,
        version: None,
        specificity: Specificity::Weak,
        span: 0..crlf_at.checked_add(2)?,
    })
}

// =====================================================================
// MySQL -- source: MySQL's own "Protocol::HandshakeV10" *packet-layout*
// page
// (<https://dev.mysql.com/doc/dev/mysql-server/latest/page_protocol_connection_phase_packets_protocol_handshake_v10.html>)
// -- not the "Connection Phase" overview page this previously cited, which
// only links to the packet layout without itself listing the fields
// (verified against the live page; root-cause fix, M4 Task 3 review round
// 1). The layout page's own field table lists `protocol_version` as
// `int<1>`, "Always 10", as the first field, with `server_version` --
// `string<NUL>` -- immediately after it. Byte 4 of the packet is therefore
// `protocol_version` (0x0a for HandshakeV10), followed immediately by the
// NUL-terminated `server_version` string. Corroborated against
// `docker.io/library/mysql:8.4`, digest
// `sha256:b3b90af2a6552ae30c266fdb7d5dd55f3afb72404bb78d37fe8a23eb857fd3fb`
// (M4 Task 2 report), whose captured `HandshakeV10` packet's version string
// reads `8.4.11` -- the exact bytes reused as this rule's own test fixture
// below.
// =====================================================================

fn mysql_handshake_v10(bytes: &[u8]) -> Option<Hit> {
    const PROTOCOL_VERSION_OFFSET: usize = 4;
    const VERSION_STRING_START: usize = 5;
    if *bytes.get(PROTOCOL_VERSION_OFFSET)? != 0x0a {
        return None;
    }
    let rest = bytes.get(VERSION_STRING_START..)?;
    let nul = rest.iter().position(|&b| b == 0)?;
    if nul == 0 {
        return None; // empty version string: nothing to report
    }
    let version_end = VERSION_STRING_START.checked_add(nul)?;
    let version = std::str::from_utf8(bytes.get(VERSION_STRING_START..version_end)?).ok()?;
    Some(Hit {
        product: Some("MySQL".to_owned()),
        version: Some(version.to_owned()),
        specificity: Specificity::ProductAndVersion,
        span: VERSION_STRING_START..version_end,
    })
}

// =====================================================================
// DNS (version.bind/TXT/CHAOS) -- source: RFC 1035 §4.1.1 (header),
// §4.1.2 (question section), §3.2.2 (TXT, type 16), §3.2.4 (CH/Chaos,
// class 3), §4.2.2 (TCP's 2-byte length prefix), §3.3.14 (TXT RDATA is a
// sequence of length-prefixed character-strings); the `version.bind`
// convention itself is documented by BIND's own manual
// (<https://bind9.readthedocs.io/en/latest/reference.html>, "Built-in
// Server Information Zones"). Corroborated against
// `docker.io/internetsystemsconsortium/bind9:9.18`, digest
// `sha256:1ffb29c718ee2540c5643c1e8166629a07bbd505f99107baae535e9f86eb7eef`
// (M4 Task 2 report), whose captured reply carries a TXT record reading
// `9.18.50` -- the exact bytes reused as this rule's own test fixture
// below.
// =====================================================================

/// Skips one DNS name starting at `at` (a run of length-prefixed labels
/// terminated by a zero-length label, or a two-byte compression pointer --
/// RFC 1035 §4.1.4), returning the offset just past it, or `None` if the
/// name runs off the end of `bytes`.
fn dns_skip_name(bytes: &[u8], mut at: usize) -> Option<usize> {
    loop {
        let len = *bytes.get(at)?;
        if len == 0 {
            return at.checked_add(1);
        }
        if len & 0xC0 == 0xC0 {
            // Compression pointer: exactly 2 bytes, does not recurse into
            // the name it points at -- not needed for this rule's purpose.
            bytes.get(at.checked_add(1)?)?;
            return at.checked_add(2);
        }
        at = at.checked_add(1)?.checked_add(len as usize)?;
    }
}

fn dns_bind_version(bytes: &[u8]) -> Option<Hit> {
    let msg_len = u16_at(bytes, 0)? as usize;
    let msg_start = 2usize;
    let msg_end = msg_start.checked_add(msg_len)?;
    if msg_end > bytes.len() {
        return None;
    }

    let flags = u16_at(bytes, msg_start.checked_add(2)?)?;
    if flags & 0x8000 == 0 {
        return None; // QR bit: must be a response, not a query
    }
    let qdcount = u16_at(bytes, msg_start.checked_add(4)?)?;
    let ancount = u16_at(bytes, msg_start.checked_add(6)?)?;
    if ancount == 0 {
        return None;
    }

    let mut at = msg_start.checked_add(12)?; // past the fixed 12-byte header
    for _ in 0..qdcount {
        at = dns_skip_name(bytes, at)?;
        at = at.checked_add(4)?; // QTYPE + QCLASS
        if at > msg_end {
            return None;
        }
    }

    for _ in 0..ancount {
        at = dns_skip_name(bytes, at)?;
        let rtype = u16_at(bytes, at)?;
        let rclass = u16_at(bytes, at.checked_add(2)?)?;
        let rdlength = u16_at(bytes, at.checked_add(8)?)? as usize; // TYPE+CLASS+TTL = 8
        let rdata_start = at.checked_add(10)?;
        let rdata_end = rdata_start.checked_add(rdlength)?;
        if rdata_end > msg_end || rdata_end > bytes.len() {
            return None;
        }

        if rtype == 16 && rclass == 3 {
            // TXT/CH: RDATA is a length-prefixed character-string.
            let txt_len = *bytes.get(rdata_start)? as usize;
            let txt_start = rdata_start.checked_add(1)?;
            let txt_end = txt_start.checked_add(txt_len)?;
            if txt_end > rdata_end {
                return None;
            }
            let version = std::str::from_utf8(bytes.get(txt_start..txt_end)?).ok()?;
            if version.is_empty() {
                return None;
            }
            return Some(Hit {
                product: Some("BIND".to_owned()),
                version: Some(version.to_owned()),
                specificity: Specificity::ProductAndVersion,
                span: txt_start..txt_end,
            });
        }
        at = rdata_end;
    }
    None
}

// =====================================================================
// SMTP -- source: RFC 5321 §3.1 ("Session Initiation": "An SMTP session is
// initiated when a client opens a connection to a server and the server
// responds with an opening message" -- §3.1 itself permits a 554 reply
// here instead of 220, so this is a description of the usual case, not a
// promise about wording); §4.3.1 ("Sequencing Overview": "Normally, a
// receiver will send a 220 'Service ready' reply" -- likewise descriptive);
// §4.2 ("SMTP Replies": the `nnn-`/`nnn ` multiline reply ABNF these
// rules' regexes depend on). (Root-cause fix, M4 Task 3 review round 1:
// this previously attributed the quotation "the SMTP server MUST send a
// 220 'Service ready' reply" to §3.1 -- that sentence does not appear
// anywhere in §3.1, RFC 5321 makes no MUST-level promise about the
// greeting at all, and the real "Normally... will send" sentence is in
// §4.3.1, not §3.1. The multiline-reply ABNF was also miscited to §4.3.1
// -- it is in §4.2, "SMTP Replies". Verified against the live RFC text for
// this fix, not re-derived from the earlier, uncorroborated citation.)
// Corroborated against `docker.io/boky/postfix:latest`,
// digest `sha256:aafc772384232497bed875e1eb66b4d3e54ba1ebc86e2e185a6dc1dbc48182ef`
// (M4 Task 2 report), which replied `220 <host> ESMTP Postfix (Debian)\r\n`.
// =====================================================================

static SMTP_GREETING_RE: LazyLock<Regex> = LazyLock::new(|| static_regex(r"^220[ -]"));

static SMTP_POSTFIX_RE: LazyLock<Regex> = LazyLock::new(|| static_regex(r"^220[ -].*\bPostfix\b"));

/// Scans every line, not just the first: RFC 5321 §4.2's own ABNF for the
/// `Greeting` allows a *multiline* 220 reply -- verbatim: `( "220-"
/// (Domain / address-literal) [ SP textstring ] CRLF *( "220-" [
/// textstring ] CRLF ) "220" [ SP textstring ] CRLF )` -- so the text
/// naming a product may legitimately be on a continuation line rather than
/// the very first one. (An earlier version of this comment transcribed the
/// final line as `"220" SP [text] CRLF`, moving the `SP` outside the
/// optional group and so making it mandatory. It is not: a conformant
/// final line may be a bare `220\r\n`, which `SMTP_GREETING_RE` -- `^220[
/// -]` -- does not match. That is a real, if narrow, false negative, left
/// as-is here deliberately: widening the regex is a behaviour change to a
/// matcher, which belongs in a rule change with its own corpus fixture,
/// not in a citation correction. M4 whole-branch fix wave citation sweep.) (Root-cause fix, M4 Task 3 review round 1: an earlier
/// version only checked `utf8_lines(bytes).first()`, which is always
/// offset 0 -- making `Hit::span`'s offset term dead code for every
/// realistic single-line-greeting test, the same issue fixed in
/// `ssh_openssh` above. See
/// `tests::smtp_postfix_finds_the_product_on_a_continuation_line`.)
fn smtp_postfix(bytes: &[u8]) -> Option<Hit> {
    for (line_start, line) in utf8_lines(bytes) {
        if let Some(m) = SMTP_POSTFIX_RE.find(line) {
            return Some(Hit {
                product: Some("Postfix".to_owned()),
                version: None,
                specificity: Specificity::ProductOnly,
                span: absolute_span(line_start, &m)?,
            });
        }
    }
    None
}

fn smtp_bare_protocol(bytes: &[u8]) -> Option<Hit> {
    let (start, first) = *utf8_lines(bytes).first()?;
    let m = SMTP_GREETING_RE.find(first)?;
    Some(Hit {
        product: None,
        version: None,
        specificity: Specificity::ProtocolOnly,
        span: absolute_span(start, &m)?,
    })
}

// =====================================================================
// TLS -- source: RFC 8446 §5.1 ("Record Layer": `ContentType ...
// handshake(22)`, i.e. content type `0x16`) and §4 ("Handshake Protocol",
// which defines `enum { ... server_hello(2), ... } HandshakeType`, i.e.
// `0x02`). §4 is NOT the handshake type registry -- that is §11, as §4's
// own text says: "New handshake message types are assigned by IANA as
// described in Section 11." An earlier version of this comment and of the
// rule's own `source` string below called §4 the registry (M4 whole-branch
// review, MINOR-3). Corroborated
// against `docker.io/library/nginx:1.27-alpine` (same digest as the HTTP
// rule above) terminating TLS 1.3 with a locally generated self-signed
// certificate (M4 Task 2 report): sending `tls-v1`'s `ClientHello` elicited
// a real `ServerHello` with exactly this record/handshake-type header.
// Structural only, deliberately: RFC 8446 §4.4 moves `Certificate` into the
// encrypted handshake flight for TLS 1.3, so no product or version can be
// read from these bytes without first decrypting them, which this probe
// (and this rule) never does -- see `bathy_probe::probes::tls`'s own doc
// comment for the same point made about the probe side.
// =====================================================================

fn tls_server_hello(bytes: &[u8]) -> Option<Hit> {
    const CONTENT_TYPE_HANDSHAKE: u8 = 0x16;
    const HANDSHAKE_TYPE_SERVER_HELLO: u8 = 0x02;
    const HEADER_LEN: usize = 6; // 5-byte record header + 1-byte handshake type
    // Destructured, not indexed: the array pattern is what makes the
    // six-byte length requirement and the two field positions one check the
    // compiler sees, rather than a `get(0..6)` whose result is then indexed
    // on the reader's word that six is bigger than five.
    let &[
        CONTENT_TYPE_HANDSHAKE,
        _,
        _,
        _,
        _,
        HANDSHAKE_TYPE_SERVER_HELLO,
    ] = bytes.get(0..HEADER_LEN)?
    else {
        return None;
    };
    Some(Hit {
        product: None,
        version: None,
        specificity: Specificity::ProtocolOnly,
        span: 0..HEADER_LEN,
    })
}

// =====================================================================
// The registry.
// =====================================================================

static ALL_RULES: &[Rule] = &[
    Rule {
        probe_id: "http-get-v1",
        doc: RuleDoc {
            id: "http.server.nginx.v1",
            service: "http",
            specificity: Specificity::ProductAndVersion,
            rationale: "The `Server` response header declared `nginx`, optionally followed by a version.",
            source: "RFC 9112 §4 (\"Status Line\"), RFC 9110 §10.2.4 (`Server`); capture from \
                      nginx:1.27-alpine (digest sha256:65645c7bb6a0661892a8b03b89d0743208a18dd2f3f17a54ef4b76fb8e2f2a10), \
                      M4 Task 2 report",
        },
        matcher: http_nginx,
    },
    Rule {
        probe_id: "http-get-v1",
        doc: RuleDoc {
            id: "http.protocol.bare.v1",
            service: "http",
            specificity: Specificity::ProtocolOnly,
            rationale: "The response's first line is a well-formed HTTP status line, but no \
                        `Server` header matched any known product.",
            source: "RFC 9112 §4 (\"Status Line\": `status-line = HTTP-version SP status-code SP \
                      [ reason-phrase ]`)",
        },
        matcher: http_bare_protocol,
    },
    Rule {
        probe_id: "ssh-banner-v1",
        doc: RuleDoc {
            id: "ssh.banner.openssh.v1",
            service: "ssh",
            specificity: Specificity::ProductAndVersion,
            rationale: "The SSH identification string named the OpenSSH software version, per \
                        the `SSH-protoversion-softwareversion` format.",
            source: "RFC 4253 §4.2 (\"Protocol Version Exchange\"); capture from \
                      linuxserver/openssh-server:latest \
                      (digest sha256:96b9a4d3b5106746d08d43a6911650d4d21f7d5c7f2ac9660e792bdb5e63157c), \
                      M4 Task 2 report",
        },
        matcher: ssh_openssh,
    },
    Rule {
        probe_id: "ssh-banner-v1",
        doc: RuleDoc {
            id: "ssh.protocol.bare.v1",
            service: "ssh",
            specificity: Specificity::ProtocolOnly,
            rationale: "The response is a well-formed SSH identification string, but the \
                        software field did not match any known product.",
            source: "RFC 4253 §4.2 (\"Protocol Version Exchange\": SSH-protoversion-softwareversion)",
        },
        matcher: ssh_bare_protocol,
    },
    Rule {
        probe_id: "postgres-startup-v1",
        doc: RuleDoc {
            id: "postgres.sslrequest.accepted.v1",
            service: "postgresql",
            specificity: Specificity::ProtocolOnly,
            rationale: "The server replied with the single byte `S`, PostgreSQL's documented \
                        SSLRequest reply meaning it will negotiate SSL.",
            source: "PostgreSQL \"Message Flow\" §54.2.10 (\"SSL Session Encryption\": \"The \
                      server then responds with a single byte containing S or N, indicating \
                      that it is willing or unwilling to perform SSL, respectively.\" -- \
                      postgresql.org/docs/current/protocol-flow.html); capture from \
                      postgres:16-alpine (digest sha256:57c72fd2a128e416c7fcc499958864df5301e940bca0a56f58fddf30ffc07777), \
                      M4 Task 2 report",
        },
        matcher: postgres_ssl_accepted,
    },
    Rule {
        probe_id: "postgres-startup-v1",
        doc: RuleDoc {
            id: "postgres.sslrequest.declined.v1",
            service: "postgresql",
            specificity: Specificity::ProtocolOnly,
            rationale: "The server replied with the single byte `N`, PostgreSQL's documented \
                        SSLRequest reply meaning it will not negotiate SSL.",
            source: "PostgreSQL \"Message Flow\" §54.2.10 (\"SSL Session Encryption\": \"The \
                      server then responds with a single byte containing S or N, indicating \
                      that it is willing or unwilling to perform SSL, respectively.\" -- \
                      postgresql.org/docs/current/protocol-flow.html); capture from \
                      postgres:16-alpine (digest sha256:57c72fd2a128e416c7fcc499958864df5301e940bca0a56f58fddf30ffc07777), \
                      M4 Task 2 report -- the container itself replied `N`",
        },
        matcher: postgres_ssl_declined,
    },
    Rule {
        probe_id: "redis-ping-v1",
        doc: RuleDoc {
            id: "redis.ping.pong.v1",
            service: "redis",
            specificity: Specificity::ProtocolOnly,
            rationale: "The server replied `+PONG`, RESP's documented reply to the `PING` command.",
            source: "Redis RESP protocol specification, \"Simple strings\" (a `+`-prefixed \
                      reply \"terminated by CRLF\") plus its \"Inline commands\" example, which \
                      shows `C: PING` answered by `S: +PONG` \
                      (redis.io/docs/latest/develop/reference/protocol-spec/); capture from \
                      redis:7-alpine (digest sha256:e7723ff73d963f5cc6d9c4643ea3d989527a402a319239054e9472a7fb9219a2), \
                      M4 Task 2 report",
        },
        matcher: redis_pong,
    },
    Rule {
        probe_id: "redis-ping-v1",
        doc: RuleDoc {
            id: "redis.protocol.resp_shaped.v1",
            service: "redis",
            specificity: Specificity::Weak,
            rationale: "The reply began with a valid RESP type sigil and carried a proper CRLF \
                        line terminator, but was not the literal `+PONG` a real Redis server \
                        sends -- consistent with a RESP-compatible service, not a confirmed \
                        product.",
            source: "Redis RESP protocol specification, \"Simple strings\" (a reply is \
                      \"terminated by CRLF\") \
                      (redis.io/docs/latest/develop/reference/protocol-spec/), structural only",
        },
        matcher: redis_resp_shaped_reply,
    },
    Rule {
        probe_id: "mysql-greeting-v1",
        doc: RuleDoc {
            id: "mysql.handshake.v10.v1",
            service: "mysql",
            specificity: Specificity::ProductAndVersion,
            rationale: "The greeting's protocol-version byte was 0x0a (HandshakeV10), followed \
                        by a NUL-terminated server-version string.",
            source: "MySQL \"Protocol::HandshakeV10\" field-layout table (protocol_version: \
                      int<1>, \"Always 10\"; immediately followed by server_version: \
                      string<NUL>) -- dev.mysql.com/doc/dev/mysql-server/latest/\
                      page_protocol_connection_phase_packets_protocol_handshake_v10.html; \
                      capture from mysql:8.4 \
                      (digest sha256:b3b90af2a6552ae30c266fdb7d5dd55f3afb72404bb78d37fe8a23eb857fd3fb), \
                      M4 Task 2 report",
        },
        matcher: mysql_handshake_v10,
    },
    Rule {
        probe_id: "dns-version-bind-v1",
        doc: RuleDoc {
            id: "dns.version_bind.txt_chaos.v1",
            service: "dns",
            specificity: Specificity::ProductAndVersion,
            rationale: "The reply's answer section carried a TXT/CH record -- the documented \
                        response to a `version.bind` query -- containing a version string.",
            source: "RFC 1035 §4.1.1 (header), §4.1.2 (question), §3.2.2 (TXT), §3.2.4 (CH), \
                      §4.2.2 (TCP length prefix), §3.3.14 (TXT RDATA); BIND manual, \"Built-in \
                      Server Information Zones\" (bind9.readthedocs.io/en/latest/reference.html); \
                      capture from internetsystemsconsortium/bind9:9.18 \
                      (digest sha256:1ffb29c718ee2540c5643c1e8166629a07bbd505f99107baae535e9f86eb7eef), \
                      M4 Task 2 report",
        },
        matcher: dns_bind_version,
    },
    Rule {
        probe_id: "smtp-banner-v1",
        doc: RuleDoc {
            id: "smtp.banner.postfix.v1",
            service: "smtp",
            specificity: Specificity::ProductOnly,
            rationale: "The 220 greeting named Postfix. Postfix's greeting does not carry a \
                        version number, so no version can be extracted.",
            source: "RFC 5321 §4.2 (\"SMTP Replies\": the `nnn-`/`nnn ` multiline reply ABNF \
                      this rule's regex scans every line for a match against); capture from \
                      boky/postfix:latest \
                      (digest sha256:aafc772384232497bed875e1eb66b4d3e54ba1ebc86e2e185a6dc1dbc48182ef), \
                      M4 Task 2 report",
        },
        matcher: smtp_postfix,
    },
    Rule {
        probe_id: "smtp-banner-v1",
        doc: RuleDoc {
            id: "smtp.protocol.bare.v1",
            service: "smtp",
            specificity: Specificity::ProtocolOnly,
            rationale: "The response is a well-formed 220 SMTP greeting, but no product name in \
                        it matched any known rule.",
            source: "RFC 5321 §4.3.1 (\"Sequencing Overview\": \"Normally, a receiver will send \
                      a 220 'Service ready' reply\" -- descriptive, not a MUST; §3.1 explicitly \
                      permits a 554 reply instead), §4.2 (\"SMTP Replies\": `nnn-`/`nnn ` \
                      multiline reply ABNF)",
        },
        matcher: smtp_bare_protocol,
    },
    Rule {
        probe_id: "tls-v1",
        doc: RuleDoc {
            id: "tls.serverhello.structural.v1",
            service: "tls",
            specificity: Specificity::ProtocolOnly,
            rationale: "The reply's record layer carried content type 0x16 (handshake) with an \
                        inner handshake type of 0x02 (ServerHello) -- confirms a TLS server \
                        answered, but (for TLS 1.3) the certificate is encrypted, so no product \
                        or version can be read from these bytes.",
            source: "RFC 8446 §5.1 (\"Record Layer\": `ContentType ... handshake(22)`, i.e. \
                      0x16), §4 (\"Handshake Protocol\", which defines `enum { ... \
                      server_hello(2), ... } HandshakeType`, i.e. 0x02) -- the IANA \"TLS \
                      HandshakeType\" registry itself is §11, not §4, as §4's own text says \
                      (\"New handshake message types are assigned by IANA as described in \
                      Section 11\"); capture from \
                      nginx:1.27-alpine (same digest as http.server.nginx.v1) terminating TLS \
                      1.3 with a locally generated self-signed certificate, M4 Task 2 report",
        },
        matcher: tls_server_hello,
    },
];

#[cfg(test)]
mod tests {
    use super::*;
    // The behavioural half of the dispatch tests below drives the crate's
    // own public entry point rather than `rules_for` directly.
    use crate::interpret;
    use bathy_types::{ProbeCapture, Transport};

    // --- the two narrow `#[allow(clippy::expect_used)]`s in this file ---
    //
    // Both allows in this module claim their `Err` arm is unreachable. These
    // three tests are what make that a checked claim rather than a comment;
    // deleting one re-opens exactly the hole the M7 panic-lint round closed.

    #[test]
    fn every_rung_of_the_ladder_is_a_valid_confidence() {
        // `Specificity::confidence`'s `expect` is allowed because every rung
        // is one of four literals inside `Confidence`'s domain. This is the
        // check on that: a rung edited to 1.5 or -0.1 fails here rather than
        // panicking in the middle of a scan.
        for rung in Specificity::ALL {
            assert!(
                Confidence::new(rung.value()).is_ok(),
                "rung {rung:?} has value {} , which Confidence rejects",
                rung.value()
            );
        }
    }

    #[test]
    fn the_ladder_array_lists_every_variant() {
        // Exhaustiveness, checked by the compiler rather than by counting:
        // a fifth `Specificity` variant makes this `match` fail to build, and
        // the `assert_eq!` catches a variant dropped from `ALL` instead.
        for rung in Specificity::ALL {
            match rung {
                Specificity::ProductAndVersion
                | Specificity::ProductOnly
                | Specificity::ProtocolOnly
                | Specificity::Weak => {}
            }
        }
        let mut seen: Vec<f64> = Specificity::ALL.iter().map(|s| s.value()).collect();
        seen.sort_by(f64::total_cmp);
        seen.dedup();
        assert_eq!(
            seen.len(),
            Specificity::ALL.len(),
            "two rungs of the ladder carry the same confidence, so one of them is not a rung"
        );
    }

    #[test]
    fn every_static_regex_in_this_module_compiles() {
        // `static_regex`'s `expect` is allowed because its arguments are
        // literals in this file. This forces every one of those `LazyLock`s,
        // so a bad pattern is a red test rather than a panic on the first
        // response that reaches the rule. A regex added below without a line
        // here is caught by `every_rule_has_its_static_regex_forced`.
        let _ = NGINX_SERVER_RE.as_str();
        let _ = SSH_OPENSSH_RE.as_str();
        let _ = SSH_BANNER_RE.as_str();
        let _ = SMTP_GREETING_RE.as_str();
        let _ = SMTP_POSTFIX_RE.as_str();
    }

    #[test]
    fn every_rule_has_its_static_regex_forced() {
        // The test above is a hand-written list, which is the shape this
        // project has repeatedly watched go stale. This is the check on the
        // list: it counts `LazyLock<Regex>` declarations in this file's own
        // source and fails if one was added without joining the test.
        let source = include_str!("rules.rs");
        let declared = source
            .lines()
            .filter(|l| l.contains("LazyLock<Regex>") && l.trim_start().starts_with("static "))
            .count();
        assert_eq!(
            declared, 5,
            "this file declares {declared} `LazyLock<Regex>` statics, not 5; add the new one to \
             every_static_regex_in_this_module_compiles and update this count"
        );
    }

    // --- known_probe_ids ---

    #[test]
    fn known_probe_ids_lists_every_probe_this_crate_has_rules_for_deduped_and_sorted() {
        // Pinned against M4 Task 2's eight real probe ids by name -- a
        // change here (an id added, removed, or renamed) is exactly the
        // kind of thing M4 Task 4's replay corpus depends on staying in
        // sync with the fixtures under `testdata/captures/`.
        let ids: Vec<&str> = known_probe_ids().collect();
        assert_eq!(
            ids,
            vec![
                "dns-version-bind-v1",
                "http-get-v1",
                "mysql-greeting-v1",
                "postgres-startup-v1",
                "redis-ping-v1",
                "smtp-banner-v1",
                "ssh-banner-v1",
                "tls-v1",
            ]
        );
    }

    // --- rules_for: the dispatch itself (M4 whole-branch review,
    // IMPORTANT-5).
    //
    // `rules_for` is this crate's entire routing mechanism -- the single
    // `r.probe_id == probe_id` comparison that decides which rules a
    // capture is even offered to -- and nothing tested it. Making it ignore
    // its argument outright, so every rule ran against every capture
    // regardless of which probe produced it, survived all 69 tests: the
    // corpus's inputs are mutually exclusive enough that no rule
    // false-positives on another protocol's bytes today.
    //
    // "Today" is the problem. That property is a coincidence of the current
    // thirteen rules, not an invariant: a TLS `ServerHello` is a
    // 6-byte structural header, a MySQL greeting is `int<1> = 10`, and a
    // redis `+...CRLF` is one leading byte -- all short, all structural,
    // and any new rule that widened one of them would start silently
    // claiming another protocol's captures with no test objecting.
    // Dispatch is what makes that impossible, so dispatch is what gets
    // asserted, in both a structural and a behavioural form. Together with
    // `bathy-engine`'s `tests/probe_rule_seam.rs` (CRITICAL-1) this closes
    // the id-equality seam at both of its ends: that the ids agree across
    // the two crates, and that this crate actually routes by them. ---

    #[test]
    fn rules_for_returns_only_rules_belonging_to_the_probe_it_was_asked_about() {
        for id in known_probe_ids() {
            let selected: Vec<&str> = rules_for(id).map(|r| r.doc.id).collect();
            assert!(
                !selected.is_empty(),
                "{id} is a known probe id, so it must select at least one rule"
            );
            for r in rules_for(id) {
                assert_eq!(
                    r.probe_id, id,
                    "rules_for({id:?}) offered rule {:?}, which belongs to probe {:?} -- \
                     a capture from one probe must never be matched against another \
                     probe's rules",
                    r.doc.id, r.probe_id
                );
            }
            assert!(
                selected.len() < ALL_RULES.len(),
                "rules_for({id:?}) returned every rule in the registry ({selected:?}); \
                 dispatch is not filtering by probe id at all"
            );
        }
    }

    #[test]
    fn rules_for_partitions_the_registry_leaving_no_rule_unreachable_and_none_duplicated() {
        // Every rule is selected by exactly one probe id: the union over
        // all known ids is the whole registry (no rule is unreachable) and
        // the counts sum without overlap (no rule is offered twice).
        let mut total = 0usize;
        let mut seen: Vec<&str> = Vec::new();
        for id in known_probe_ids() {
            for r in rules_for(id) {
                total += 1;
                seen.push(r.doc.id);
            }
        }
        seen.sort_unstable();
        let mut deduped = seen.clone();
        deduped.dedup();
        assert_eq!(
            seen, deduped,
            "a rule was offered by two different probe ids"
        );
        assert_eq!(
            total,
            ALL_RULES.len(),
            "the union of rules_for() over every known probe id must be exactly the registry"
        );
    }

    #[test]
    fn a_rule_is_never_offered_a_capture_from_a_different_probe() {
        // The behavioural half, through the crate's own public entry point
        // rather than through `rules_for` directly: each of these byte
        // strings is one this crate genuinely recognizes under its OWN
        // probe id (every one of them is asserted to match in a test
        // elsewhere in this module). Delivered under any OTHER probe id,
        // `interpret` must return nothing at all -- not a lower-confidence
        // guess, nothing. This is what "the probe that produced these bytes
        // is part of the evidence" means in practice.
        let recognized: &[(&'static str, &[u8])] = &[
            (
                "http-get-v1",
                b"HTTP/1.1 200 OK\r\nServer: nginx/1.26.0\r\n\r\n",
            ),
            ("ssh-banner-v1", b"SSH-2.0-OpenSSH_10.3\r\n"),
            ("smtp-banner-v1", b"220 mail.example.com ESMTP Postfix\r\n"),
            ("redis-ping-v1", b"+PONG\r\n"),
            ("tls-v1", &[0x16, 0x03, 0x03, 0x00, 0x02, 0x02, 0x00]),
            ("postgres-startup-v1", b"S"),
        ];
        for &(owner, bytes) in recognized {
            let own = interpret(&ProbeCapture {
                probe_id: owner,
                transport: Transport::Tcp,
                port: 0,
                request: None,
                response: bytes.to_vec(),
                elapsed_micros: 0,
                truncated: false,
            });
            assert!(
                !own.is_empty(),
                "test fixture sanity: {owner} must recognize its own bytes, or the \
                 cross-feeding below proves nothing"
            );
            for other in known_probe_ids().filter(|&id| id != owner) {
                let cross = interpret(&ProbeCapture {
                    probe_id: other,
                    transport: Transport::Tcp,
                    port: 0,
                    request: None,
                    response: bytes.to_vec(),
                    elapsed_micros: 0,
                    truncated: false,
                });
                assert!(
                    cross.is_empty(),
                    "bytes only {owner} can produce were interpreted as {:?} when \
                     delivered under probe id {other}",
                    cross.iter().map(|i| i.rule_id).collect::<Vec<_>>()
                );
            }
        }
    }

    // --- The ladder ---

    #[test]
    fn ladder_orders_product_and_version_above_product_only_above_protocol_only_above_weak() {
        assert!(
            Specificity::ProductAndVersion.confidence().get()
                > Specificity::ProductOnly.confidence().get()
        );
        assert!(
            Specificity::ProductOnly.confidence().get()
                > Specificity::ProtocolOnly.confidence().get()
        );
        assert!(
            Specificity::ProtocolOnly.confidence().get() > Specificity::Weak.confidence().get()
        );
    }

    // --- utf8_lines ---

    #[test]
    fn utf8_lines_splits_on_newline_and_reports_correct_offsets() {
        let bytes = b"HTTP/1.1 200 OK\r\nServer: nginx\r\n\r\n";
        let lines = utf8_lines(bytes);
        assert_eq!(lines[0], (0, "HTTP/1.1 200 OK\r"));
        assert_eq!(lines[1].0, 17);
        assert!(lines[1].1.starts_with("Server: nginx"));
    }

    #[test]
    fn utf8_lines_skips_a_line_that_is_not_valid_utf8_but_keeps_earlier_and_later_lines() {
        let mut bytes = b"clean line one\n".to_vec();
        bytes.extend_from_slice(&[0xff, 0xfe, b'\n']); // not valid UTF-8
        bytes.extend_from_slice(b"clean line three\n");
        let lines = utf8_lines(&bytes);
        let texts: Vec<&str> = lines.iter().map(|(_, s)| *s).collect();
        assert_eq!(texts, vec!["clean line one", "clean line three"]);
    }

    #[test]
    fn utf8_lines_never_panics_on_empty_input() {
        assert!(utf8_lines(&[]).is_empty());
    }

    // --- HTTP ---

    #[test]
    fn http_nginx_extracts_product_and_version() {
        let bytes = b"HTTP/1.1 200 OK\r\nServer: nginx/1.27.5\r\n\r\n";
        let hit = http_nginx(bytes).unwrap();
        assert_eq!(hit.product.as_deref(), Some("nginx"));
        assert_eq!(hit.version.as_deref(), Some("1.27.5"));
        assert_eq!(hit.specificity, Specificity::ProductAndVersion);
        assert_eq!(&bytes[hit.span.clone()], b"Server: nginx/1.27.5");
    }

    #[test]
    fn http_nginx_without_a_version_is_product_only() {
        let hit = http_nginx(b"HTTP/1.1 200 OK\r\nServer: nginx\r\n\r\n").unwrap();
        assert!(hit.version.is_none());
        assert_eq!(hit.specificity, Specificity::ProductOnly);
    }

    /// The exact response `10.30.0.17:443` -- the lab's TLS-only nginx --
    /// returns to a plaintext request, byte for byte as `lab/run.sh verify`
    /// read it. This is the named test `lab/ground-truth.json`'s
    /// `identification_gap` at that endpoint points at.
    ///
    /// It exists to locate the gap precisely, because "bathy does not identify
    /// nginx behind TLS" has two possible causes and only one of them is true.
    /// It is NOT that the rules cannot read these bytes: they can, at full
    /// `ProductAndVersion` specificity, as this asserts. It is that
    /// `Scheduler::detect_service` stops at the first probe whose capture
    /// interprets to anything, and on 443 that is `tls-v1`, which is
    /// protocol-only by construction -- RFC 8446 §4.4 moves the certificate
    /// into the encrypted flight -- so `http-get-v1` is never reached and
    /// these bytes are never captured.
    ///
    /// If that policy changes, AC-7.5's conformance test is what goes red and
    /// demands the ground truth's `identification_gap` key be deleted. This
    /// test is what says the rule side was never the problem, and it runs with
    /// no lab, no Docker and no network.
    #[test]
    fn the_bytes_a_tls_terminator_returns_to_a_plaintext_request_name_its_product() {
        let observed: &[u8] = b"HTTP/1.1 400 Bad Request\r\nServer: nginx/1.29.8\r\n\
                                Date: Tue, 04 Aug 2026 17:29:44 GMT\r\n\
                                Content-Type: text/html\r\nContent-Length: 255\r\n\
                                Connection: close\r\n\r\n<html>\r\n\
                                <head><title>400 The plain HTTP request was sent to \
                                HTTPS port</title></head>\r\n";
        let hit = http_nginx(observed).expect(
            "the `Server` header is in cleartext ahead of any handshake; if this stops \
             matching, lab/ground-truth.json's claim at 10.30.0.17:443 has lost its basis",
        );
        assert_eq!(hit.product.as_deref(), Some("nginx"));
        assert_eq!(hit.version.as_deref(), Some("1.29.8"));
        assert_eq!(
            hit.specificity,
            Specificity::ProductAndVersion,
            "a 4xx status is still an HTTP response and still names its server"
        );
    }

    #[test]
    fn http_nginx_does_not_match_a_non_http_response() {
        assert!(http_nginx(b"Server: nginx/1.27.5\r\n").is_none());
    }

    #[test]
    fn http_nginx_does_not_match_a_different_server_header() {
        assert!(http_nginx(b"HTTP/1.1 200 OK\r\nServer: Apache/2.4.62\r\n\r\n").is_none());
    }

    // --- Soundness regression: the brief's own worked example reused
    // `String::from_utf8_lossy` offsets as indices into the original
    // bytes, which silently cites the wrong bytes (or panics) once
    // anything before the match isn't valid UTF-8. These two tests are
    // built to fail under that unsound version specifically -- reverting
    // `http_nginx` to a whole-response-lossy-conversion implementation and
    // running the suite is this task's own review-round-1 finding; see the
    // fix report for the reproduced failure. ---

    #[test]
    fn http_nginx_cites_the_correct_bytes_when_invalid_utf8_precedes_the_match() {
        // A stray 0x80 (a lone UTF-8 continuation byte, invalid on its own)
        // sits on an earlier header line, strictly before the `Server:`
        // line that actually matches. `utf8_lines` skips only that one
        // invalid line; `http_nginx` must still report a span pointing at
        // the real `Server:` bytes, at their real offset in the original
        // buffer -- not an offset computed against a lossy re-encoding of
        // the whole response (which would grow by 2 bytes at the one
        // invalid byte, a `String::from_utf8_lossy` implementation would
        // silently misalign every offset after it).
        let mut bytes = Vec::new();
        bytes.extend_from_slice(b"HTTP/1.1 200 OK\r\n");
        bytes.extend_from_slice(b"X-Bad: \x80\r\n");
        bytes.extend_from_slice(b"Server: nginx/1.26.0\r\n");
        bytes.extend_from_slice(b"\r\n");
        let hit = http_nginx(&bytes).unwrap();
        assert_eq!(hit.version.as_deref(), Some("1.26.0"));
        assert_eq!(
            &bytes[hit.span.clone()],
            b"Server: nginx/1.26.0",
            "span must index the real bytes even with invalid UTF-8 earlier in the response"
        );
    }

    #[test]
    fn http_nginx_span_stays_in_bounds_when_the_match_ends_at_the_last_byte() {
        // No trailing CRLF at all: the `Server:` line is both the match
        // and the literal last byte of the buffer. A lossy-offset
        // implementation whose earlier invalid byte inflated every
        // downstream offset by 2 would push `span.end` past
        // `bytes.len()`, which panics on the slice index below rather than
        // merely being wrong.
        let mut bytes = Vec::new();
        bytes.extend_from_slice(b"HTTP/1.1 200 OK\r\n");
        bytes.extend_from_slice(b"X-Bad: \x80\r\n");
        bytes.extend_from_slice(b"Server: nginx/1.26.0"); // ends the buffer, no CRLF
        let hit = http_nginx(&bytes).unwrap();
        assert_eq!(
            hit.span.end,
            bytes.len(),
            "the match ends exactly at the buffer's end"
        );
        assert_eq!(&bytes[hit.span.clone()], b"Server: nginx/1.26.0");
    }

    #[test]
    fn http_bare_protocol_matches_any_status_line() {
        let bytes = b"HTTP/1.0 404 Not Found\r\n\r\n";
        let hit = http_bare_protocol(bytes).unwrap();
        assert_eq!(
            &bytes[hit.span.clone()],
            b"HTTP/1.0 404 Not Found\r",
            "span must be exactly the status line, not the whole response"
        );
    }

    // --- The `*_bare_protocol` line-offset gap (M4 whole-branch review,
    // IMPORTANT-4). `http_bare_protocol` and `smtp_bare_protocol` are the
    // only two matchers that take `utf8_lines(bytes).first()` rather than
    // iterating, so in every test above their line offset happens to be 0
    // and the `start +` term in their span is indistinguishable from a
    // literal `0` -- dropping it survived the whole suite for both. The
    // offset is NOT always 0: `utf8_lines` skips lines that are not valid
    // UTF-8, so `first()` is the first VALID line, which is at a non-zero
    // offset whenever anything invalid precedes it. That is reachable from
    // a hostile or merely broken peer, which is exactly the case
    // `matched_span` -- the "which bytes justified this claim" contract --
    // must not get wrong.
    //
    // The audit that produced these two tests mutated the offset term out
    // of every one of the ten span constructions in this file, not just the
    // one reported: `http_nginx`, `ssh_openssh`, `ssh_bare_protocol`,
    // `smtp_postfix`, `mysql_handshake`, `dns_bind_version`, both redis
    // matchers and `tls_server_hello` all die already. These two were the
    // only survivors, and they share one shape. See the fix-wave report. ---

    #[test]
    fn http_bare_protocol_cites_the_correct_bytes_when_invalid_utf8_precedes_the_status_line() {
        // A lone 0x80 (an unaccompanied UTF-8 continuation byte, invalid on
        // its own) makes line 0 undecodable, so `utf8_lines` skips it and
        // the status line -- the real match -- starts at byte 9, not 0.
        let bytes = b"\x80garbage\nHTTP/1.1 200 OK\r\n\r\n";
        let hit = http_bare_protocol(bytes).unwrap();
        assert_eq!(
            hit.span,
            9..25,
            "the status line begins after the skipped invalid line, not at byte 0"
        );
        assert_eq!(
            &bytes[hit.span.clone()],
            b"HTTP/1.1 200 OK\r",
            "span must cite the status line's real bytes; dropping the line-start offset \
             cites b\"\\x80garbage\\nHTTP/1.\" instead -- the wrong bytes entirely, and \
             still a perfectly valid range, so no bounds check would ever notice"
        );
    }

    #[test]
    fn smtp_bare_protocol_cites_the_correct_bytes_when_invalid_utf8_precedes_the_greeting() {
        // Same shape as the HTTP case above, and the same defect: this is
        // the second of the two matchers that read only the first VALID
        // line. A real SMTP peer that emits a non-UTF-8 byte before its
        // greeting is unusual; a hostile one that does it deliberately, to
        // make bathy cite bytes it did not send, is the threat this
        // matters for.
        let bytes = b"\x80junk\n220 mail.example.com ESMTP Sendmail\r\n";
        let hit = smtp_bare_protocol(bytes).unwrap();
        assert_eq!(
            hit.span,
            6..10,
            "the greeting begins after the skipped invalid line, not at byte 0"
        );
        assert_eq!(
            &bytes[hit.span.clone()],
            b"220 ",
            "span must cite the greeting's real bytes, not b\"\\x80jun\""
        );
    }

    // --- SSH ---

    #[test]
    fn ssh_openssh_extracts_version_and_ignores_the_trailing_comment() {
        let bytes = b"SSH-2.0-OpenSSH_9.6p1 Ubuntu-3ubuntu13\r\n";
        let hit = ssh_openssh(bytes).unwrap();
        assert_eq!(hit.product.as_deref(), Some("OpenSSH"));
        assert_eq!(hit.version.as_deref(), Some("9.6p1"));
        assert_eq!(&bytes[hit.span.clone()], b"SSH-2.0-OpenSSH_9.6p1");
    }

    #[test]
    fn ssh_openssh_matches_the_real_captured_banner_with_no_comment() {
        // The exact banner captured from linuxserver/openssh-server:latest
        // in M4 Task 2 (see this module's source note).
        let hit = ssh_openssh(b"SSH-2.0-OpenSSH_10.3\r\n").unwrap();
        assert_eq!(hit.version.as_deref(), Some("10.3"));
    }

    #[test]
    fn ssh_openssh_does_not_match_a_non_openssh_banner() {
        assert!(ssh_openssh(b"SSH-2.0-libssh_0.9.6\r\n").is_none());
    }

    // RFC 4253 §4.2: "The server MAY send other lines of data before
    // sending the version string... Clients MUST be able to process such
    // lines." A matcher that only ever looked at line 0 would false-
    // negative here; it would also make `Hit::span`'s line-offset term
    // untestable (offset 0 either way). This is both the false-negative
    // fix and its own regression test, together.
    #[test]
    fn ssh_openssh_finds_the_identification_line_after_a_preamble_line() {
        let bytes = b"Some preamble the server sent first\r\nSSH-2.0-OpenSSH_9.6p1\r\n";
        let hit = ssh_openssh(bytes).unwrap();
        assert_eq!(hit.version.as_deref(), Some("9.6p1"));
        assert_eq!(&bytes[hit.span.clone()], b"SSH-2.0-OpenSSH_9.6p1");
        assert!(
            hit.span.start > 0,
            "the identification line is not at offset 0 here, so this also proves the \
             line-offset term in Hit::span is real, not dead code"
        );
    }

    #[test]
    fn ssh_bare_protocol_matches_any_ssh_banner() {
        let bytes = b"SSH-2.0-libssh_0.9.6\r\n";
        let hit = ssh_bare_protocol(bytes).unwrap();
        assert_eq!(&bytes[hit.span.clone()], b"SSH-2.0-");
    }

    #[test]
    fn ssh_bare_protocol_finds_the_identification_line_after_a_preamble_line() {
        let bytes = b"Some preamble the server sent first\r\nSSH-2.0-libssh_0.9.6\r\n";
        let hit = ssh_bare_protocol(bytes).unwrap();
        assert_eq!(&bytes[hit.span.clone()], b"SSH-2.0-");
        assert!(hit.span.start > 0);
    }

    // --- Postgres ---

    #[test]
    fn postgres_ssl_accepted_matches_exactly_s() {
        let hit = postgres_ssl_accepted(b"S").unwrap();
        assert_eq!(&b"S"[hit.span.clone()], b"S");
        assert!(postgres_ssl_accepted(b"N").is_none());
        assert!(postgres_ssl_accepted(b"SS").is_none());
    }

    #[test]
    fn postgres_ssl_declined_matches_the_real_captured_reply() {
        // postgres:16-alpine (M4 Task 2 report) replied exactly `N`.
        let hit = postgres_ssl_declined(b"N").unwrap();
        assert_eq!(&b"N"[hit.span.clone()], b"N");
        assert!(postgres_ssl_declined(b"S").is_none());
    }

    // --- Redis ---

    #[test]
    fn redis_pong_matches_the_real_captured_reply() {
        let bytes = b"+PONG\r\n";
        let hit = redis_pong(bytes).unwrap();
        assert_eq!(&bytes[hit.span.clone()], b"+PONG");
    }

    #[test]
    fn redis_resp_shaped_reply_is_weak_for_a_non_pong_resp_value() {
        let bytes = b"-ERR unknown command\r\n";
        let hit = redis_resp_shaped_reply(bytes).unwrap();
        assert_eq!(hit.specificity, Specificity::Weak);
        assert_eq!(&bytes[hit.span.clone()], b"-ERR unknown command\r\n");
    }

    #[test]
    fn redis_resp_shaped_reply_does_not_match_non_resp_bytes() {
        assert!(redis_resp_shaped_reply(b"HTTP/1.1 200 OK\r\n").is_none());
        assert!(redis_resp_shaped_reply(b"").is_none());
    }

    // --- Root-cause fix, M4 Task 3 review round 1: a single stray sigil
    // byte with no CRLF terminator used to match on its own -- one byte
    // from an arbitrary binary protocol is not meaningful RESP evidence.
    #[test]
    fn redis_resp_shaped_reply_rejects_a_lone_sigil_byte_with_no_crlf() {
        assert!(redis_resp_shaped_reply(b"+").is_none());
        assert!(redis_resp_shaped_reply(b"+X").is_none());
        assert!(redis_resp_shaped_reply(b"+no terminator here").is_none());
    }

    // --- MySQL ---

    // The real handshake packet captured from `mysql:8.4` (M4 Task 2
    // report), reused verbatim here as this rule's own fixture.
    const MYSQL_GREETING_HEX: &str = "4a0000000a382e342e3131000800000062215a7649740441\
                                       00ffffff0200ffdf1500000000000000000000441e1b514b\
                                       4e6e53084a5c270063616368696e675f736861325f706173\
                                       73776f726400";

    fn hex_to_bytes(hex: &str) -> Vec<u8> {
        (0..hex.len())
            .step_by(2)
            .map(|i| u8::from_str_radix(&hex[i..i + 2], 16).unwrap())
            .collect()
    }

    #[test]
    fn mysql_handshake_v10_extracts_the_real_captured_version() {
        let bytes = hex_to_bytes(MYSQL_GREETING_HEX);
        let hit = mysql_handshake_v10(&bytes).unwrap();
        assert_eq!(hit.product.as_deref(), Some("MySQL"));
        assert_eq!(hit.version.as_deref(), Some("8.4.11"));
        // Asserted independently of `hit.version` above: `version` and
        // `span` are computed from the same offsets in the real code, but
        // a mutant that shifts only `Hit::span` (leaving the string
        // extraction that produces `.version` untouched) would pass the
        // assertion above while citing the wrong bytes. Re-deriving the
        // expected text from `hit.span` itself is what catches that.
        assert_eq!(&bytes[hit.span.clone()], b"8.4.11");
    }

    #[test]
    fn mysql_handshake_v10_rejects_a_short_packet() {
        assert!(mysql_handshake_v10(b"\x0a\x00").is_none());
    }

    #[test]
    fn mysql_handshake_v10_rejects_a_non_handshake_v10_protocol_byte() {
        let mut bytes = hex_to_bytes(MYSQL_GREETING_HEX);
        bytes[4] = 0x09; // not HandshakeV10
        assert!(mysql_handshake_v10(&bytes).is_none());
    }

    #[test]
    fn mysql_handshake_v10_rejects_a_missing_nul_terminator() {
        let bytes = vec![0u8, 0, 0, 0, 0x0a, b'8', b'.', b'4']; // no trailing NUL
        assert!(mysql_handshake_v10(&bytes).is_none());
    }

    // --- DNS ---

    // The real reply captured from `internetsystemsconsortium/bind9:9.18`
    // (M4 Task 2 report), reused verbatim as this rule's own fixture.
    const BIND_REPLY_HEX: &str = "00405344840000010001000100000776657273696f6e0462696e6400001000\
                                   03c00c0010000300000000000807392e31382e3530c00c00020003000000000\
                                   002c00c";

    #[test]
    fn dns_bind_version_extracts_the_real_captured_version() {
        let bytes = hex_to_bytes(BIND_REPLY_HEX);
        let hit = dns_bind_version(&bytes).unwrap();
        assert_eq!(hit.product.as_deref(), Some("BIND"));
        assert_eq!(hit.version.as_deref(), Some("9.18.50"));
        // See the identical comment on the MySQL test above: re-derive the
        // expected text from `hit.span` itself, independent of whatever
        // internal offsets produced `.version`, so a span-only shift is
        // caught even when `.version` still happens to be correct.
        assert_eq!(&bytes[hit.span.clone()], b"9.18.50");
    }

    #[test]
    fn dns_bind_version_rejects_a_query_not_a_response() {
        // Same shape as the real query bathy_probe::probes::dns::build_query
        // sends (QR bit clear).
        let query: Vec<u8> = [
            0x00, 0x1e, 0x53, 0x44, 0x00, 0x00, 0x00, 0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 7,
            b'v', b'e', b'r', b's', b'i', b'o', b'n', 4, b'b', b'i', b'n', b'd', 0, 0x00, 0x10,
            0x00, 0x03,
        ]
        .to_vec();
        assert!(dns_bind_version(&query).is_none());
    }

    #[test]
    fn dns_bind_version_rejects_truncated_bytes_without_panicking() {
        let full = hex_to_bytes(BIND_REPLY_HEX);
        for cut in 0..full.len() {
            assert!(dns_bind_version(&full[..cut]).is_none());
        }
    }

    #[test]
    fn dns_skip_name_rejects_a_label_length_that_runs_past_the_end() {
        // A label byte of 63 (top two bits clear, so a genuine label
        // length -- not a compression pointer, which needs the top two
        // bits set per RFC 1035 §4.1.4) claiming 63 more bytes in a
        // 3-byte buffer must not panic or wrap; it must simply fail to
        // resolve.
        assert!(dns_skip_name(&[63, 1, 2], 0).is_none());
    }

    #[test]
    fn dns_skip_name_treats_a_top_bits_set_byte_as_a_two_byte_compression_pointer() {
        // 200 = 0b1100_1000: top two bits set, so RFC 1035 §4.1.4 defines
        // this as a compression pointer, not a 200-byte label -- it
        // consumes exactly 2 bytes regardless of what follows.
        assert_eq!(dns_skip_name(&[200, 1, 2], 0), Some(2));
    }

    // --- SMTP ---

    #[test]
    fn smtp_postfix_matches_the_real_captured_greeting() {
        let bytes = b"220 mail.example.com ESMTP Postfix\r\n";
        let hit = smtp_postfix(bytes).unwrap();
        assert_eq!(hit.product.as_deref(), Some("Postfix"));
        assert!(hit.version.is_none());
        assert_eq!(
            &bytes[hit.span.clone()],
            b"220 mail.example.com ESMTP Postfix"
        );
    }

    #[test]
    fn smtp_postfix_does_not_match_a_non_postfix_greeting() {
        assert!(smtp_postfix(b"220 mail.example.com ESMTP Sendmail\r\n").is_none());
    }

    // RFC 5321 §4.2's own `Greeting` ABNF allows a multiline 220 reply
    // (`"220-" Domain [SP text] CRLF *("220-" [text] CRLF) "220" SP [text]
    // CRLF`); the product name may legitimately be on a continuation line,
    // not the first one. This also forces `line_start > 0`, which is what
    // makes the offset term in `Hit::span` observable at all -- see the
    // identical point made on the SSH tests above.
    #[test]
    fn smtp_postfix_finds_the_product_on_a_continuation_line() {
        let bytes = b"220-mail.example.com ESMTP\r\n220 Postfix ready\r\n";
        let hit = smtp_postfix(bytes).unwrap();
        assert_eq!(hit.product.as_deref(), Some("Postfix"));
        assert!(
            hit.span.start > 0,
            "the matching line is not at offset 0 here"
        );
        assert_eq!(&bytes[hit.span.clone()], b"220 Postfix");
    }

    #[test]
    fn smtp_bare_protocol_matches_any_220_greeting() {
        let bytes = b"220 mail.example.com ESMTP Sendmail\r\n";
        let hit = smtp_bare_protocol(bytes).unwrap();
        assert_eq!(&bytes[hit.span.clone()], b"220 ");
    }

    // --- TLS ---

    #[test]
    fn tls_server_hello_matches_the_structural_header() {
        let reply: &[u8] = &[0x16, 0x03, 0x03, 0x00, 0x02, 0x02, 0x00];
        let hit = tls_server_hello(reply).unwrap();
        assert_eq!(
            &reply[hit.span.clone()],
            &[0x16, 0x03, 0x03, 0x00, 0x02, 0x02]
        );
    }

    #[test]
    fn tls_server_hello_does_not_match_a_non_handshake_record() {
        let alert: &[u8] = &[0x15, 0x03, 0x03, 0x00, 0x02, 0x02, 0x00];
        assert!(tls_server_hello(alert).is_none());
    }

    #[test]
    fn tls_server_hello_rejects_a_too_short_buffer_without_panicking() {
        for len in 0..6 {
            assert!(tls_server_hello(&vec![0x16; len]).is_none());
        }
    }
}