gossan-portscan 0.3.3

#![forbid(unsafe_code)]
// pedantic moved to workspace [lints.clippy] in root Cargo.toml
//
// `unwrap_used` / `todo` / `unimplemented` / `panic` stay forbidden.
#![cfg_attr(
    not(test),
    deny(
        clippy::unwrap_used,
        clippy::todo,
        clippy::unimplemented,
        clippy::panic
    )
)]
#![allow(
    clippy::module_name_repetitions,
    clippy::must_use_candidate,
    clippy::missing_errors_doc
)]

//! TCP connect scanner with banner grabbing, active service probing,
//! TLS inspection (JA3/JA3S, cert chain, cipher weakness), rate limiting,
//! IPv6 support, and checkpoint resume.
//!
//! # Configuration
//!
//! Port lists, service probes, and risky service definitions are loaded from TOML:
//! - `rules/top_ports.toml`: port list definitions
//! - `rules/risky_services.toml`: high-risk service definitions
//! - `rules/service_probes.toml`: active service probe payloads (~200+)

pub mod cdn;
pub mod cve;
pub mod jarm;
pub mod probes;
pub mod rules;
pub mod stateless;
pub mod tls;
pub mod top_ports;

#[cfg(test)]
mod integration_tests;

use std::fmt;
use std::net::{IpAddr, Ipv4Addr, SocketAddrV4};
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use std::time::Duration;

use async_trait::async_trait;
use futures::StreamExt;
use gossan_core::ratelimit::{BackoffKind, BackoffPolicy, BACKOFF_TIMEOUT_BASE_MS};
use gossan_core::{
    Config, DiscoverySource, DomainTarget, HostTarget, PortMode, Protocol, ScanInput, Scanner,
    ServiceTarget, Target,
};
use secfinding::{Evidence, Finding, FindingBuilder, Severity};
use tokio::io::AsyncReadExt;

/// Maximum number of hosts enumerated per CIDR range before the scan is
/// truncated and an informational finding is emitted. A /24 exactly fits;
/// anything larger is split by the caller.
const MAX_HOSTS_PER_CIDR: usize = 256;

use gossan_core::{EPHEMERAL_PORT_COUNT, EPHEMERAL_PORT_START};

/// Maximum retries for a TCP connect probe before giving up.
const PROBE_MAX_RETRIES: u32 = 3;

/// Well-known TLS ports (connections on these ports get cert inspection).
/// Inline test `tls_ports_well_known` pins this list against the match arm
/// in `probe_port` so the two cannot drift independently.
const TLS_PORTS: &[u16] = &[443, 8443, 465, 993, 636, 995, 587];

/// A unique key identifying a scanned target (IP address or domain) and port.
#[derive(Debug, Clone, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)]
pub struct ScanTargetKey {
    pub target: String,
    pub port: u16,
}

/// TCP port scanner with banner grabbing, TLS inspection, and CVE correlation.
pub struct PortScanner;

impl Default for PortScanner {
    fn default() -> Self {
        Self::new()
    }
}

impl PortScanner {
    /// Creates a new port scanner instance.
    pub fn new() -> Self {
        Self
    }
}

impl fmt::Display for PortScanner {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "PortScanner({})", self.name())
    }
}

/// Creates a finding builder pre-configured for portscan findings.
pub fn finding_builder(
    target: &Target,
    severity: Severity,
    title: impl Into<String>,
    detail: impl Into<String>,
) -> FindingBuilder {
    Finding::builder("portscan", target.domain().unwrap_or("?"), severity)
        .title(title)
        .detail(detail)
        .kind(secfinding::FindingKind::Exposure)
}

#[async_trait]
impl Scanner for PortScanner {
    fn name(&self) -> &'static str {
        "portscan"
    }
    fn tags(&self) -> &[&'static str] {
        &["active", "network"]
    }
    fn accepts(&self, target: &Target) -> bool {
        matches!(
            target,
            Target::Domain(_) | Target::Host(_) | Target::Network(_)
        )
    }

    async fn run(&self, input: ScanInput, config: &Config) -> anyhow::Result<()> {
        let timeout = config.timeout();
        let host_delay = Duration::from_millis(config.host_delay_ms);
        let rate_limiter = Arc::new(gossan_core::ratelimit::HostRateLimiter::new(
            config.rate_limit.max(1),
        ));

        // Drain all targets from the channel
        let mut all_input_targets = Vec::new();
        {
            let mut rx = input.target_rx.lock().await;
            // recv() until the pipeline closes the inbox — try_recv races the
            // sender and drops asynchronously delivered targets.
            while let Some(t) = rx.recv().await {
                all_input_targets.push(t);
            }
        }
        let mut expanded_targets = Vec::new();
        for t in &all_input_targets {
            if let Target::Network(net) = t {
                if let Ok(prefix) = net.cidr.parse::<ipnet::IpNet>() {
                    // Cap enumeration so IPv6 /0 (2^128 hosts) does not
                    // overflow usize during count(), ipnet emulates std
                    // overflow behavior which panics in debug builds.
                    let total_hosts = prefix.hosts().take(MAX_HOSTS_PER_CIDR + 1).count();
                    if total_hosts > MAX_HOSTS_PER_CIDR {
                        if let Some(f) = finding_builder(
                            &Target::Network(net.clone()),
                            Severity::Info,
                            format!(
                                "CIDR range {} truncated: scanning {}/{} hosts",
                                net.cidr, MAX_HOSTS_PER_CIDR, total_hosts
                            ),
                            format!(
                                "Network {} contains {} hosts but scanning is limited to {} per range. \
                                 {} hosts will NOT be scanned. Split into /24 subnets for full coverage.",
                                net.cidr, total_hosts, MAX_HOSTS_PER_CIDR,
                                total_hosts.saturating_sub(MAX_HOSTS_PER_CIDR)
                            ),
                        )
                        .tag("cidr")
                        .tag("truncation")
                        .kind(secfinding::FindingKind::InfoDisclosure)
                        .build_or_log()
                        {
                            input.emit(f).await;
                        }
                    }
                    for addr in prefix.hosts().take(MAX_HOSTS_PER_CIDR) {
                        expanded_targets.push(Target::Host(HostTarget {
                            ip: addr,
                            domain: None,
                        }));
                    }
                }
            } else {
                expanded_targets.push(t.clone());
            }
        }

        let active_ports: Vec<u16> = match &config.port_mode {
            mode => gossan_core::resolve_ports(mode),
        };

        // ── Resume support: load checkpoint if available ─────────────────────
        let completed_ports: Arc<std::sync::Mutex<std::collections::HashSet<ScanTargetKey>>> =
            Arc::new(std::sync::Mutex::new(std::collections::HashSet::new()));
        let checkpoint_path = std::env::var("GOSSAN_CHECKPOINT")
            .ok()
            .map(std::path::PathBuf::from)
            .or_else(|| Some(std::path::PathBuf::from("gossan-scan.db")));

        if let Some(ref path) = checkpoint_path {
            if path.exists() {
                // Touch the CheckpointStore as a structural soundness
                // check (corrupted DB → don't try to resume), then read
                // the per-portscan sidecar JSON. The portscan resume
                // contract is intentionally side-cared off the main
                // store: per-(IpAddr, u16) granularity would inflate
                // the SQLite write rate to one row per probed port,
                // which dominated wall time on previous benchmarks.
                if gossan_checkpoint::CheckpointStore::open(path).is_ok() {
                    if let Ok(content) =
                        std::fs::read_to_string(path.with_extension("portscan-resume.json"))
                    {
                        if let Ok(keys) = serde_json::from_str::<Vec<ScanTargetKey>>(&content) {
                            completed_ports
                                .lock()
                                .unwrap_or_else(|e| e.into_inner())
                                .extend(keys);
                            tracing::info!(
                                resumed = completed_ports
                                    .lock()
                                    .unwrap_or_else(|e| e.into_inner())
                                    .len(),
                                "resuming portscan from checkpoint"
                            );
                        } else if let Ok(old_ports) =
                            serde_json::from_str::<Vec<(IpAddr, u16)>>(&content)
                        {
                            let keys: Vec<ScanTargetKey> = old_ports
                                .into_iter()
                                .map(|(ip, port)| ScanTargetKey {
                                    target: ip.to_string(),
                                    port,
                                })
                                .collect();
                            completed_ports
                                .lock()
                                .unwrap_or_else(|e| e.into_inner())
                                .extend(keys);
                            tracing::info!(
                                resumed = completed_ports
                                    .lock()
                                    .unwrap_or_else(|e| e.into_inner())
                                    .len(),
                                "resuming portscan from legacy checkpoint"
                            );
                        }
                    }
                }
            }
        }

        // ── Stateless SYN pre-filter (Linux + CAP_NET_RAW) ───────────────────
        let want_stateless: bool = {
            #[cfg(target_os = "linux")]
            {
                stateless::transport::linux::raw_available() && config.proxy.is_none()
            }
            #[cfg(not(target_os = "linux"))]
            {
                false
            }
        };

        let domain_ips: std::collections::HashMap<String, IpAddr> = if want_stateless {
            let mut m = std::collections::HashMap::new();
            for t in &expanded_targets {
                if let Target::Domain(d) = t {
                    if m.contains_key(&d.domain) {
                        continue;
                    }
                    if let Ok(addrs) = input.resolver.lookup_ip(format!("{}.", d.domain)).await {
                        for addr in addrs {
                            m.insert(d.domain.clone(), addr);
                            break;
                        }
                    }
                }
            }
            m
        } else {
            std::collections::HashMap::new()
        };

        #[cfg(target_os = "linux")]
        let syn_open: Option<std::collections::HashSet<(Ipv4Addr, u16)>> = if want_stateless {
            let mut resolved = Vec::new();
            for t in &expanded_targets {
                match t {
                    Target::Host(h) => {
                        if let IpAddr::V4(v4) = h.ip {
                            resolved.push((h.ip.to_string(), h.domain.clone(), v4));
                        }
                    }
                    Target::Domain(d) => {
                        if let Some(&IpAddr::V4(v4)) = domain_ips.get(&d.domain) {
                            resolved.push((d.domain.clone(), Some(d.domain.clone()), v4));
                        }
                    }
                    _ => {}
                }
            }
            if resolved.is_empty() {
                None
            } else {
                let unique_ips: Vec<Ipv4Addr> = {
                    let set: std::collections::HashSet<_> =
                        resolved.iter().map(|(_, _, ip)| *ip).collect();
                    set.into_iter().collect()
                };
                let src_port =
                    EPHEMERAL_PORT_START + (std::process::id() as u16 % EPHEMERAL_PORT_COUNT);
                let src_ip = stateless::transport::local_source_ipv4(Ipv4Addr::new(8, 8, 8, 8))
                    .unwrap_or(Ipv4Addr::UNSPECIFIED);
                if src_ip.is_unspecified() {
                    tracing::warn!("could not determine local source IPv4 for stateless SYN scan");
                    None
                } else {
                    let src = SocketAddrV4::new(src_ip, src_port);
                    let cookie = stateless::cookie::SynCookie::random();
                    let seed: u64 = rand::random();
                    let unique_ips_len = unique_ips.len();
                    let mut scanner = stateless::StatelessScanner::new(
                        src,
                        unique_ips,
                        active_ports.clone(),
                        cookie,
                        seed,
                    );
                    let total = scanner.total();
                    tracing::info!(
                        targets = resolved.len(),
                        unique_ips = unique_ips_len,
                        ports = active_ports.len(),
                        total_probes = total,
                        "starting stateless SYN pre-filter"
                    );
                    let rate_limit = config.rate_limit as u64;
                    let outcomes = match tokio::task::spawn_blocking(move || {
                        let mut transport = stateless::transport::linux::RawSynTransport::new()?;
                        stateless::transport::run_blocking(
                            &mut scanner,
                            &mut transport,
                            rate_limit,
                            std::time::Duration::from_secs(2),
                        )
                    })
                    .await
                    {
                        Ok(Ok(outcomes)) => Some(outcomes),
                        Ok(Err(e)) => {
                            tracing::warn!(
                                error = %e,
                                "stateless SYN pre-filter failed; falling back to connect scan"
                            );
                            None
                        }
                        Err(e) => {
                            tracing::warn!(
                                error = %e,
                                "stateless SYN pre-filter join failed; falling back to connect scan"
                            );
                            None
                        }
                    };
                    if let Some(outcomes) = outcomes {
                        let mut open_set = std::collections::HashSet::new();
                        for o in &outcomes {
                            if let stateless::Outcome::Open(sa) = o {
                                open_set.insert((*sa.ip(), sa.port()));
                            }
                        }
                        tracing::info!(
                            open = open_set.len(),
                            total_probes = total,
                            "stateless SYN pre-filter complete"
                        );
                        Some(open_set)
                    } else {
                        None
                    }
                }
            }
        } else {
            None
        };
        #[cfg(not(target_os = "linux"))]
        let syn_open: Option<std::collections::HashSet<(Ipv4Addr, u16)>> = None;

        let mut pairs: Vec<(String, Option<String>, u16, IpAddr)> = expanded_targets
            .iter()
            .filter(|t| self.accepts(t))
            .flat_map(|t| {
                let (addr, domain, ip) = match t {
                    Target::Domain(d) => {
                        let ip = domain_ips.get(&d.domain).copied();
                        (d.domain.clone(), Some(d.domain.clone()), ip)
                    }
                    Target::Host(h) => (h.ip.to_string(), h.domain.clone(), Some(h.ip)),
                    _ => return Vec::new(),
                };
                active_ports
                    .iter()
                    .filter({
                        let completed_ports = Arc::clone(&completed_ports);
                        let target_str = addr.clone();
                        move |&&p| {
                            let key = ScanTargetKey {
                                target: target_str.clone(),
                                port: p,
                            };
                            !completed_ports
                                .lock()
                                .unwrap_or_else(|e| e.into_inner())
                                .contains(&key)
                        }
                    })
                    .map(move |&p| {
                        let ip = ip.unwrap_or_else(|| {
                            // placeholder; resolved later
                            IpAddr::from([0, 0, 0, 0])
                        });
                        (addr.clone(), domain.clone(), p, ip)
                    })
                    .collect::<Vec<_>>()
            })
            .collect();

        if let Some(ref open_set) = syn_open {
            pairs.retain(|(_, _, port, ip)| {
                if let IpAddr::V4(v4) = ip {
                    if !v4.is_unspecified() {
                        return open_set.contains(&(*v4, *port));
                    }
                }
                true
            });
        }

        let open_count = Arc::new(AtomicUsize::new(0));
        let probe_engine = Arc::new(probes::ProbeEngine::new(timeout));

        // ── Periodic checkpoint saving task ──────────────────────────────────
        let checkpoint_task = if let Some(ref path) = checkpoint_path {
            let path = path.clone();
            let completed_ports = Arc::clone(&completed_ports);
            Some(tokio::spawn(async move {
                loop {
                    tokio::time::sleep(Duration::from_secs(5)).await;
                    let resume_file = path.with_extension("portscan-resume.json");
                    let data = {
                        let locked = completed_ports.lock().unwrap_or_else(|e| e.into_inner());
                        let keys: Vec<&ScanTargetKey> = locked.iter().collect();
                        serde_json::to_string(&keys)
                    };
                    if let Ok(json) = data {
                        if let Err(e) = tokio::fs::write(&resume_file, json).await {
                            tracing::warn!(err = %e, "failed to write periodic checkpoint");
                        }
                    }
                }
            }))
        } else {
            None
        };

        let results: Vec<Option<(ServiceTarget, Vec<Finding>, Vec<Target>)>> =
            futures::stream::iter(pairs)
                .map(|(addr, domain, port, ip)| {
                    let rl = Arc::clone(&rate_limiter);
                    let proxy_opt = config.proxy.clone();
                    let engine = Arc::clone(&probe_engine);
                    let open_count = Arc::clone(&open_count);
                    let completed_ports = Arc::clone(&completed_ports);
                    async move {
                        // Per-host rate limiting
                        rl.until_ready(&addr).await;
                        tokio::time::sleep(host_delay).await;

                        let result = retry_probe(
                            &addr,
                            domain.clone(),
                            port,
                            timeout,
                            proxy_opt.as_deref(),
                            &engine,
                        )
                        .await;

                        // Mark this (ip, port) pair complete REGARDLESS
                        // of result. Resume semantics are "we already
                        // probed this once", the result (open / closed
                        // / filtered) doesn't change whether we should
                        // re-probe on resume. Without this, the loaded
                        // completed_ports set was treated read-only and
                        // every resumed run re-scanned every port from
                        // scratch, exactly the bug the warning on the
                        // unused `ip` variable was concealing.
                        completed_ports
                            .lock()
                            .unwrap_or_else(|e| e.into_inner())
                            .insert(ScanTargetKey {
                                target: addr.clone(),
                                port,
                            });

                        if let Some((ref svc, _, _)) = result {
                            tracing::debug!(host = ?svc.host.ip, port = svc.port, "open port");
                            open_count.fetch_add(1, Ordering::Relaxed);
                        }
                        result
                    }
                })
                .buffer_unordered(config.concurrency)
                .collect()
                .await;

        // ── Extract the seed's root domain for SAN filtering ─────────────────
        let seed_root = extract_root_domain(&input.seed);

        for item in results.into_iter().flatten() {
            let (svc, findings, extra_targets) = item;
            for f in findings {
                input.emit(f).await;
            }
            input.emit_target(Target::Service(svc)).await;

            for t in extra_targets {
                if let Target::Domain(ref d) = t {
                    let san_root = extract_root_domain(&d.domain);
                    if san_root == seed_root
                        || d.domain.ends_with(&format!(".{}", input.seed))
                        || input.seed.ends_with(&format!(".{}", d.domain))
                    {
                        input.emit_target(t).await;
                    } else {
                        tracing::debug!(
                            san = %d.domain,
                            seed = %input.seed,
                            "filtered out-of-scope SAN domain"
                        );
                    }
                } else {
                    input.emit_target(t).await;
                }
            }
        }

        // ── Abort periodic checkpoint task and save final state ──────────────
        if let Some(handle) = checkpoint_task {
            handle.abort();
            let _ = handle.await;
        }

        // ── Save checkpoint ──────────────────────────────────────────────────
        if let Some(path) = checkpoint_path {
            let resume_file = path.with_extension("portscan-resume.json");
            let data = {
                let guard = completed_ports.lock().unwrap_or_else(|e| e.into_inner());
                let keys: Vec<&ScanTargetKey> = guard.iter().collect();
                serde_json::to_string(&keys)?
            };
            if let Err(e) = tokio::fs::write(&resume_file, data).await {
                tracing::warn!(
                    path = %resume_file.display(),
                    error = %e,
                    "portscan resume checkpoint write failed"
                );
            }
        }

        tracing::info!(
            open = open_count.load(Ordering::Relaxed),
            "port scan complete"
        );
        Ok(())
    }
}

async fn retry_probe(
    addr: &str,
    domain: Option<String>,
    port: u16,
    timeout: Duration,
    proxy: Option<&str>,
    engine: &probes::ProbeEngine,
) -> Option<(ServiceTarget, Vec<Finding>, Vec<Target>)> {
    let backoff = probe_retry_backoff();
    for attempt in 0..PROBE_MAX_RETRIES {
        match probe_port(addr, domain.clone(), port, timeout, proxy, engine).await {
            Some(result) => return Some(result),
            None if backoff.should_retry_after(attempt) => {
                tokio::time::sleep(backoff.delay(BackoffKind::Timeout, attempt)).await;
            }
            None => return None,
        }
    }
    None
}

fn probe_retry_backoff() -> BackoffPolicy {
    BackoffPolicy::new(
        PROBE_MAX_RETRIES,
        BACKOFF_TIMEOUT_BASE_MS,
        BACKOFF_TIMEOUT_BASE_MS,
    )
}

async fn probe_port(
    addr: &str,
    domain: Option<String>,
    port: u16,
    timeout: Duration,
    proxy: Option<&str>,
    engine: &probes::ProbeEngine,
) -> Option<(ServiceTarget, Vec<Finding>, Vec<Target>)> {
    let stream = tokio::time::timeout(timeout, gossan_core::net::connect_tcp(addr, port, proxy))
        .await
        .ok()?
        .ok()?;

    let ip = stream.peer_addr().ok()?.ip();

    // Run banner grab and active probes in parallel under a shared deadline
    let deadline = timeout.max(Duration::from_secs(5));
    let probe_future = engine.probe(stream, addr, port, proxy);
    let (banner, probe_matches) = match tokio::time::timeout(deadline, probe_future).await {
        Ok((b, m)) => (b, m),
        Err(_) => {
            tracing::warn!(
                addr = %addr,
                port,
                "portscan probe timed out; continuing with empty banner/matches"
            );
            (None, Vec::new())
        }
    };

    let tls = TLS_PORTS.contains(&port);

    let svc = ServiceTarget {
        host: HostTarget { ip, domain },
        port,
        protocol: Protocol::Tcp,
        banner: banner.clone(),
        tls,
    };

    let mut findings: Vec<Finding> = Vec::new();
    let mut extra_targets: Vec<Target> = Vec::new();

    // Emit finding for high-risk port exposure (require banner confirmation)
    if let Some(r) = rules::risky_service_by_port(port) {
        let target = Target::Service(svc.clone());
        let severity = if banner.is_none() {
            Severity::Low // downgrade if we can't confirm the service
        } else {
            r.severity
        };
        // Structured tags so the masscan-grepable renderer can pick
        // out the IP / port / proto / service hint without parsing
        // the human-readable title. The hint is derived from the
        // banner via the same logic the cli uses for grepable output.
        let svc_hint = banner.as_deref().and_then(|b| {
            // Keep this in sync with cli::output::classify_service_hint.
            let bl = b.to_ascii_lowercase();
            if bl.starts_with("ssh-") || port == 22 {
                Some("ssh")
            } else if bl.contains("http/") || matches!(port, 80 | 8080 | 8000 | 8888) {
                Some("http")
            } else if matches!(port, 443 | 8443) {
                Some("https")
            } else if bl.starts_with("220 ") && (bl.contains("smtp") || port == 25) {
                Some("smtp")
            } else if bl.starts_with("220") && port == 21 {
                Some("ftp")
            } else if port == 6379 || bl.contains("noauth") {
                Some("redis")
            } else if port == 27017 || bl.contains("mongodb") {
                Some("mongodb")
            } else {
                None
            }
        });
        let mut f = finding_builder(&target, severity, r.name.clone(), r.detail.clone())
            .tag("exposure")
            .tag("network")
            .tag(format!("ip:{ip}"))
            .tag(format!("port:{port}/tcp"));
        if let Some(s) = svc_hint {
            f = f.tag(format!("service:{s}"));
        }
        if let Some(ref b) = banner {
            f = f.evidence(Evidence::Banner {
                raw: b.clone().into(),
            });
        }
        gossan_core::try_push_finding(f, &mut findings);
    }

    // Banner / probe identification
    let combined_banner = banner.as_deref().unwrap_or("");
    if let Some(id_finding) = identify_banner_or_probe(combined_banner, &probe_matches, &svc, port)
    {
        let existing_max = findings.iter().map(|f| f.severity()).max();
        if existing_max.is_none_or(|max| id_finding.severity() >= max) {
            findings.clear();
            findings.push(id_finding);
        } else {
            findings.push(id_finding);
        }
    }

    // TLS cert inspection (parallelized)
    if tls {
        let tls_deadline = timeout.max(Duration::from_secs(8));
        let tls_future = async {
            let mut all_findings = Vec::new();
            if let Some(cert) = tls::probe_tls(addr, port, timeout, proxy).await {
                let days = tls::days_until_expiry(cert.not_after_unix);
                let target = Target::Service(svc.clone());

                if days < 0 {
                    gossan_core::try_push_finding(
                        finding_builder(
                            &target,
                            Severity::Critical,
                            format!("TLS certificate expired {} days ago", -days),
                            format!(
                                "Certificate for port {} expired. Browsers will show security warnings.",
                                port
                            ),
                        )
                        .tag("tls")
                        .tag("cert")
                        .tag("expired")
                        .kind(secfinding::FindingKind::Misconfiguration),
                        &mut all_findings,
                    );
                } else if days <= 14 {
                    gossan_core::try_push_finding(
                        finding_builder(
                            &target,
                            Severity::Medium,
                            format!("TLS certificate expires in {} days", days),
                            format!(
                                "Certificate for port {} expires very soon. Immediate renewal required.",
                                port
                            ),
                        )
                        .tag("tls")
                        .tag("cert")
                        .tag("expiry")
                        .kind(secfinding::FindingKind::Misconfiguration),
                        &mut all_findings,
                    );
                } else if days <= 30 {
                    gossan_core::try_push_finding(
                        finding_builder(
                            &target,
                            Severity::Medium,
                            format!("TLS certificate expires in {} days", days),
                            format!("Certificate for port {} expiring within 30 days.", port),
                        )
                        .tag("tls")
                        .tag("cert")
                        .tag("expiry"),
                        &mut all_findings,
                    );
                }

                if cert.is_self_signed {
                    gossan_core::try_push_finding(
                        finding_builder(
                            &target,
                            Severity::Medium,
                            "Self-signed TLS certificate",
                            format!(
                                "Port {} uses a self-signed certificate, clients cannot verify authenticity.",
                                port
                            ),
                        )
                        .tag("tls")
                        .tag("cert")
                        .tag("self-signed")
                        .kind(secfinding::FindingKind::Misconfiguration),
                        &mut all_findings,
                    );
                }

                // Note: cipher_weakness and negotiated_version removed from TlsCertInfo

                for san in &cert.sans {
                    let san = san.trim_start_matches("*.").to_string();
                    if !san.is_empty() {
                        extra_targets.push(Target::Domain(DomainTarget {
                            domain: san,
                            source: DiscoverySource::CertificateTransparency,
                        }));
                    }
                }

                tracing::debug!(
                    port,
                    subject = %cert.subject,
                    issuer = %cert.issuer,
                    sans = ?cert.sans,
                    "TLS cert inspected"
                );
            }

            // Legacy TLS protocol detection
            let legacy = tls::probe_legacy(addr, port, timeout, proxy).await;
            let target = Target::Service(svc.clone());
            if legacy.supports_tls10 {
                gossan_core::try_push_finding(
                    finding_builder(
                        &target,
                        Severity::Low,
                        format!(
                            "TLS 1.0 supported on port {}: BEAST/POODLE vulnerable",
                            port
                        ),
                        format!(
                            "Port {} accepts TLS 1.0 connections. TLS 1.0 has known protocol-level \
                             vulnerabilities (BEAST, POODLE) and was deprecated by RFC 8996.",
                            port
                        ),
                    )
                    .tag("tls")
                    .tag("legacy-tls")
                    .tag("protocol")
                    .kind(secfinding::FindingKind::Misconfiguration),
                    &mut all_findings,
                );
            }
            if legacy.supports_tls11 {
                gossan_core::try_push_finding(
                    finding_builder(
                        &target,
                        Severity::Low,
                        format!("TLS 1.1 supported on port {}, deprecated (RFC 8996)", port),
                        format!(
                            "Port {} accepts TLS 1.1 connections. TLS 1.1 was deprecated alongside \
                             TLS 1.0 in RFC 8996 (March 2021). Configure the server to require \
                             TLS 1.2 or higher.",
                            port
                        ),
                    )
                    .tag("tls")
                    .tag("legacy-tls")
                    .tag("protocol")
                    .kind(secfinding::FindingKind::Misconfiguration),
                    &mut all_findings,
                );
            }

            all_findings
        };

        let tls_results = match tokio::time::timeout(tls_deadline, tls_future).await {
            Ok(v) => v,
            Err(_) => {
                tracing::warn!("TLS probe future timed out; skipping TLS findings for this service: addr={} port={}", addr, port);
                Vec::new()
            }
        };
        findings.extend(tls_results);

        // JARM (optional, default off)
        let jarm_enabled = std::env::var("GOSSAN_JARM")
            .map(|s| s == "1" || s == "true")
            .unwrap_or(false);
        if jarm_enabled {
            if let Some(fp) = jarm::fingerprint(addr, port, timeout, proxy).await {
                let target = Target::Service(svc.clone());
                let known_tag = jarm::identify(&fp);
                let (severity, title, detail) = if let Some(name) = known_tag {
                    (
                        Severity::Critical,
                        format!("JARM fingerprint matches {}", name),
                        format!(
                            "TLS fingerprint {} matches known C2/malware framework: {}.",
                            fp, name
                        ),
                    )
                } else {
                    (
                        Severity::Info,
                        "JARM TLS fingerprint".to_string(),
                        format!("JARM fingerprint: {}  (Shodan: ssl.jarm:{})", fp, fp),
                    )
                };
                gossan_core::try_push_finding(
                    finding_builder(&target, severity, title, detail)
                        .tag("jarm")
                        .tag("tls")
                        .tag("fingerprint")
                        .kind(secfinding::FindingKind::TechDetect),
                    &mut findings,
                );
            }
        }
    }

    // CVE correlation from banner + probe responses
    let banner_for_cve = if banner.is_some() {
        banner.as_deref().unwrap_or("").to_string()
    } else {
        probe_matches.join(" | ")
    };
    if !banner_for_cve.is_empty() {
        findings.extend(cve::correlate(&banner_for_cve, &svc));

        // NVD CVE database lookup (optional, needs pre-synced cache)
        #[cfg(feature = "nvd")]
        {
            let b = banner_for_cve.clone();
            let s = svc.clone();
            if let Ok(nvd_findings) =
                tokio::task::spawn_blocking(move || cve::nvd::try_search(&b, &s)).await
            {
                findings.extend(nvd_findings);
            }
        }
    }

    Some((svc, findings, extra_targets))
}

fn build_banner_finding(builder: secfinding::FindingBuilder) -> Option<Finding> {
    match builder.build() {
        Ok(f) => Some(f),
        Err(e) => {
            tracing::warn!(error = %e, "portscan banner finding build failed");
            None
        }
    }
}

fn identify_banner_or_probe(
    banner: &str,
    probe_matches: &[String],
    svc: &ServiceTarget,
    port: u16,
) -> Option<Finding> {
    let b = banner.to_lowercase();

    // SSH version disclosure
    if b.starts_with("ssh-") || banner.starts_with("SSH-") {
        let version = banner.lines().next().unwrap_or(banner).trim();
        let severity = if version.contains("OpenSSH_7")
            || version.contains("OpenSSH_6")
            || version.contains("OpenSSH_5")
            || version.contains("OpenSSH_4")
        {
            Severity::High
        } else {
            Severity::Info
        };
        return build_banner_finding(
            finding_builder(
                &Target::Service(svc.clone()),
                severity,
                format!("SSH version disclosed: {}", version),
                "SSH banner reveals server version. Old versions may have known CVEs.",
            )
            .evidence(Evidence::Banner {
                raw: banner.to_string().into(),
            })
            .tag("banner")
            .tag("ssh")
            .tag("version-disclosure")
            .kind(secfinding::FindingKind::InfoDisclosure)
        );
    }

    // FTP banner
    if (port == 21 || b.contains("ftp")) && (b.starts_with("220") || b.starts_with("230")) {
        let version = banner.lines().next().unwrap_or(banner).trim();
        return build_banner_finding(
            finding_builder(
                &Target::Service(svc.clone()),
                Severity::Info,
                format!("FTP banner: {}", version),
                "FTP banner may disclose server software and version.",
            )
            .evidence(Evidence::Banner {
                raw: banner.to_string().into(),
            })
            .tag("banner")
            .tag("ftp")
            .tag("version-disclosure")
            .kind(secfinding::FindingKind::InfoDisclosure)
        );
    }

    // SMTP banner
    if (port == 25 || port == 465 || port == 587) && b.starts_with("220") {
        let version = banner.lines().next().unwrap_or(banner).trim();
        return build_banner_finding(
            finding_builder(
                &Target::Service(svc.clone()),
                Severity::Info,
                format!("SMTP banner: {}", version),
                "SMTP banner may disclose mail server software and version.",
            )
            .evidence(Evidence::Banner {
                raw: banner.to_string().into(),
            })
            .tag("banner")
            .tag("smtp")
            .tag("version-disclosure")
            .kind(secfinding::FindingKind::InfoDisclosure)
        );
    }

    // HTTP Server header
    if b.starts_with("http/") {
        let server_line = banner
            .lines()
            .find(|l| l.to_lowercase().starts_with("server:"))
            .unwrap_or("");
        if !server_line.is_empty() {
            return build_banner_finding(
            finding_builder(
                    &Target::Service(svc.clone()),
                    Severity::Info,
                    format!("HTTP server header: {}", server_line.trim()),
                    "HTTP Server header discloses software and version.",
                )
                .evidence(Evidence::Banner {
                    raw: banner.to_string().into(),
                })
                .tag("banner")
                .tag("http")
                .tag("version-disclosure")
        );
        }
    }

    // Redis
    if port == 6379 && (b.starts_with('+') || b.starts_with('-')) {
        return build_banner_finding(
            finding_builder(
                &Target::Service(svc.clone()),
                Severity::Critical,
                "Redis responds without authentication",
                "Redis accepted connection and responded, likely unauthenticated. Full data access and potential RCE via cron/SSH key write.",
            )
            .evidence(Evidence::Banner { raw: banner.to_string().into() })
            .tag("banner")
            .tag("redis")
            .tag("no-auth")
            .kind(secfinding::FindingKind::Vulnerability)
        );
    }

    // MongoDB
    if port == 27017 && (banner.contains("MongoDB") || b.contains("ismaster")) {
        return build_banner_finding(
            finding_builder(
                &Target::Service(svc.clone()),
                Severity::Critical,
                "MongoDB responds, likely unauthenticated",
                "MongoDB accepted connection. May allow unauthenticated full database access.",
            )
            .evidence(Evidence::Banner {
                raw: banner.to_string().into(),
            })
            .tag("banner")
            .tag("mongodb")
            .tag("no-auth")
            .kind(secfinding::FindingKind::Vulnerability)
        );
    }

    // Telnet
    if port == 23 {
        return build_banner_finding(
            finding_builder(
                &Target::Service(svc.clone()),
                Severity::Critical,
                "Telnet service responds",
                "Telnet is active and responding. All traffic is plaintext, immediate credential interception risk.",
            )
            .evidence(Evidence::Banner { raw: banner.to_string().into() })
            .tag("banner")
            .tag("telnet")
            .tag("plaintext")
            .kind(secfinding::FindingKind::Vulnerability)
        );
    }

    // Elasticsearch
    if (port == 9200 || port == 9300)
        && (b.contains("lucene") || b.contains("elasticsearch") || b.contains("\"cluster_name\""))
    {
        return build_banner_finding(
            finding_builder(
                &Target::Service(svc.clone()),
                Severity::Critical,
                "Elasticsearch responds, likely unauthenticated",
                format!(
                    "Elasticsearch on port {} accepted connection and returned cluster info. \
                     Unauthenticated access allows full index enumeration, data exfiltration, \
                     and potential RCE via script queries.",
                    port
                ),
            )
            .evidence(Evidence::Banner {
                raw: banner.to_string().into(),
            })
            .tag("banner")
            .tag("elasticsearch")
            .tag("no-auth")
            .kind(secfinding::FindingKind::Vulnerability)
        );
    }

    // PostgreSQL
    if port == 5432
        && (banner.contains("PostgreSQL") || b.contains("pgsql") || b.contains("pg_hba.conf"))
    {
        let severity = if b.contains("no pg_hba.conf entry") {
            Severity::Info
        } else {
            Severity::High
        };
        return build_banner_finding(
            finding_builder(
                &Target::Service(svc.clone()),
                severity,
                "PostgreSQL service responds",
                format!("PostgreSQL on port {} is accepting connections.", port),
            )
            .evidence(Evidence::Banner {
                raw: banner.to_string().into(),
            })
            .tag("banner")
            .tag("postgresql")
            .tag("database")
            .kind(secfinding::FindingKind::Exposure)
        );
    }

    // MySQL
    if port == 3306
        && (banner.contains("mysql") || b.contains("mariadb") || b.contains("caching_sha2"))
    {
        return build_banner_finding(
            finding_builder(
                &Target::Service(svc.clone()),
                Severity::High,
                format!("MySQL/MariaDB responds on port {}", port),
                format!(
                    "MySQL on port {} is accepting connections. Direct database port exposure \
                     enables brute-force attacks and version-specific CVE exploitation.",
                    port
                ),
            )
            .evidence(Evidence::Banner {
                raw: banner.to_string().into(),
            })
            .tag("banner")
            .tag("mysql")
            .tag("database")
            .kind(secfinding::FindingKind::Exposure)
        );
    }

    // Memcached
    if port == 11211
        && (b.starts_with("stat") || b.starts_with("version") || b.starts_with("error"))
    {
        return build_banner_finding(
            finding_builder(
                &Target::Service(svc.clone()),
                Severity::Critical,
                "Memcached responds, likely unauthenticated",
                "Memcached on port 11211 accepted connection. Unauthenticated access allows full \
                 cache dump, data injection, and DDoS amplification (UDP reflection).",
            )
            .evidence(Evidence::Banner {
                raw: banner.to_string().into(),
            })
            .tag("banner")
            .tag("memcached")
            .tag("no-auth")
            .kind(secfinding::FindingKind::Vulnerability)
        );
    }

    // Kubernetes API
    if (port == 6443 || port == 443 || port == 8443)
        && b.contains("\"kind\"")
        && (b.contains("status") || b.contains("api"))
    {
        let severity = if b.contains("forbidden") || b.contains("unauthorized") {
            Severity::Medium
        } else {
            Severity::Critical
        };
        return build_banner_finding(
            finding_builder(
                &Target::Service(svc.clone()),
                severity,
                format!("Kubernetes API server detected on port {}", port),
                format!(
                    "Kubernetes API responding on port {}. {} access may expose cluster \
                     configuration, secrets, and allow container escape.",
                    port,
                    if severity == Severity::Critical {
                        "Unauthenticated"
                    } else {
                        "Authenticated"
                    }
                ),
            )
            .evidence(Evidence::Banner {
                raw: banner.to_string().into(),
            })
            .tag("banner")
            .tag("kubernetes")
            .tag("api")
            .kind(secfinding::FindingKind::Exposure)
        );
    }

    // Probe-based matches (from active probes)
    for m in probe_matches {
        return build_banner_finding(
            finding_builder(
                &Target::Service(svc.clone()),
                Severity::Info,
                format!("Service detected via active probe: {}", m),
                "An active service probe returned a positive match.",
            )
            .tag("banner")
            .tag("probe")
            .kind(secfinding::FindingKind::TechDetect)
        );
    }

    None
}

/// Attempts to grab a service banner from an open TCP connection.
pub async fn grab_banner(mut stream: tokio::net::TcpStream, timeout: Duration) -> Option<String> {
    let mut buf = vec![0u8; 512];
    let effective_timeout = timeout.max(Duration::from_millis(100));
    let n = match tokio::time::timeout(effective_timeout, stream.read(&mut buf)).await {
        Ok(Ok(n)) => n,
        Ok(Err(e)) => {
            tracing::debug!(error = %e, "failed to read banner from stream");
            return None;
        }
        Err(_) => {
            tracing::debug!("banner grab timed out");
            return None;
        }
    };

    if n == 0 {
        tracing::debug!("banner read returned 0 bytes - connection closed immediately");
        return None;
    }

    let s: String = buf[..n]
        .iter()
        .map(|&b| {
            if (0x20..0x7f).contains(&b) {
                b as char
            } else {
                '.'
            }
        })
        .collect::<String>()
        .trim()
        .to_string();

    if s.is_empty() {
        tracing::debug!("banner contained only non-printable characters");
        None
    } else {
        Some(s)
    }
}

/// Extract the registrable root domain from a full domain name.
fn extract_root_domain(domain: &str) -> String {
    let domain = domain.trim_end_matches('.').to_lowercase();
    // Fast path for simple cases
    if domain.is_empty() {
        return domain;
    }
    // Pure fallback: last two labels (or three for known two-part TLDs)
    // No external publicsuffix crate dependency needed.
    let parts: Vec<&str> = domain.split('.').collect();
    if parts.len() <= 2 {
        return domain;
    }
    let last_two = format!("{}.{}", parts[parts.len() - 2], parts[parts.len() - 1]);
    let common_two_part_suffixes = [
        "co.uk", "org.uk", "me.uk", "ltd.uk", "plc.uk", "sch.uk", "gov.uk", "ac.uk", "net.uk",
        "com.au", "net.au", "org.au", "edu.au", "gov.au", "asn.au", "id.au", "com.cn", "edu.cn",
        "gov.cn", "org.cn", "net.cn", "ac.cn", "com.br", "net.br", "org.br", "edu.br", "gov.br",
        "co.jp", "or.jp", "ne.jp", "ac.jp", "ad.jp", "ed.jp", "go.jp", "com.sg", "org.sg",
        "edu.sg", "gov.sg", "net.sg", "co.nz", "net.nz", "org.nz", "edu.nz", "gov.nz", "com.tw",
        "org.tw", "gov.tw", "edu.tw", "net.tw", "com.hk", "org.hk", "gov.hk", "edu.hk", "net.hk",
    ];
    if common_two_part_suffixes.contains(&last_two.as_str()) && parts.len() >= 3 {
        format!(
            "{}.{}.{}",
            parts[parts.len() - 3],
            parts[parts.len() - 2],
            parts[parts.len() - 1]
        )
    } else {
        last_two
    }
}

#[cfg(test)]
mod edge_tests;
#[cfg(test)]
mod tests;