//! Cross-server namespace checks.
//!
//! Why this is a separate module: the surface checks in [`crate::security::surface`] are
//! per-definition predicates run by the [`crate::security::SecurityEngine`]. Namespace
//! risks are relational, they only exist when two or more servers are seen
//! together (a second server shadowing a tool name, a typosquatted server
//! name). They do not fit the per-[`crate::security::Definition`] [`crate::security::SecurityCheck`]
//! trait, so they live here as
//! a standalone function that takes the full multi-server snapshot at once.
//!
//! Each check is deterministic. Matchers are conservative to keep false
//! positives low, since a near-duplicate heuristic that fires on every shared
//! prefix would be noise rather than signal.

use serde_json::Value;

use super::{CheckCategory, DefinitionKind, SecurityFinding, SecuritySeverity};

/// Run every namespace check over a multi-server snapshot.
///
/// Each element of `servers` is one server's name paired with its
/// `tools/list`-style JSON value (the same shape the per-definition engine
/// consumes). Only the `tools` array of each value is read, because the
/// namespace risks (shadowing, squatting, ambiguous resolution) are about tool
/// names a client routes on. Findings come back in check-ID order so a report
/// reads top to bottom by rule.
///
/// ```
/// use mcptest_core::security::namespace_findings;
/// use serde_json::json;
///
/// let servers = vec![
///     ("alpha".to_string(), json!({ "tools": [{ "name": "search" }] })),
///     ("beta".to_string(), json!({ "tools": [{ "name": "search" }] })),
/// ];
/// let findings = namespace_findings(&servers);
/// assert!(findings.iter().any(|f| f.rule_id == "SEC-010"));
/// ```
pub fn namespace_findings(servers: &[(String, Value)]) -> Vec<SecurityFinding> {
    let tools = flatten_tools(servers);
    let mut findings = Vec::new();
    findings.extend(duplicate_tool_name(&tools));
    findings.extend(tool_name_squat(&tools));
    findings.extend(server_name_squat(servers));
    findings.extend(ambiguous_resolution(&tools));
    findings
}

/// One tool as the namespace checks see it: which server served it, its name,
/// and the input schema (used to tell a shadow from an ambiguous resolution).
struct ToolRef<'a> {
    server: &'a str,
    name: String,
    schema: &'a Value,
}

/// Flatten every server's `tools` array into a flat list tagged by server.
///
/// A tool with no string `name` is skipped: it cannot collide or squat on a
/// name it does not have, and a missing name is the surface lane's concern.
fn flatten_tools(servers: &[(String, Value)]) -> Vec<ToolRef<'_>> {
    let mut out = Vec::new();
    for (server, snapshot) in servers {
        let Some(arr) = snapshot.get("tools").and_then(|v| v.as_array()) else {
            continue;
        };
        for tool in arr {
            let Some(name) = tool.get("name").and_then(|v| v.as_str()) else {
                continue;
            };
            out.push(ToolRef {
                server,
                name: name.to_string(),
                schema: tool.get("inputSchema").unwrap_or(&Value::Null),
            });
        }
    }
    out
}

/// Build a finding for a namespace check. Namespace findings are about a tool
/// or server name rather than a parsed [`Definition`], so `definition` carries
/// that name and `kind` is always [`DefinitionKind::Tool`] (the namespace lane
/// only routes on tools).
fn finding(
    rule_id: &str,
    name: &str,
    severity: SecuritySeverity,
    definition: String,
    message: String,
    evidence: String,
) -> SecurityFinding {
    SecurityFinding {
        rule_id: rule_id.to_string(),
        name: name.to_string(),
        severity,
        category: CheckCategory::Namespace,
        definition,
        kind: DefinitionKind::Tool,
        message,
        evidence,
    }
}

// ---------------------------------------------------------------------------
// SEC-010 duplicate-tool-name
// ---------------------------------------------------------------------------

/// The same tool name served by more than one server. A client that routes on
/// the bare name cannot tell which server it reaches, so a second server can
/// shadow a trusted tool. One finding per duplicated name.
fn duplicate_tool_name(tools: &[ToolRef<'_>]) -> Vec<SecurityFinding> {
    let mut out = Vec::new();
    for (i, tool) in tools.iter().enumerate() {
        // Report each duplicated name once, at its first occurrence.
        if tools[..i].iter().any(|t| t.name == tool.name) {
            continue;
        }
        let servers: Vec<&str> = tools
            .iter()
            .filter(|t| t.name == tool.name)
            .map(|t| t.server)
            .collect();
        if servers.len() > 1 {
            out.push(finding(
                "SEC-010",
                "duplicate-tool-name",
                SecuritySeverity::High,
                tool.name.clone(),
                "the same tool name is served by more than one server, so a client cannot tell which it reaches".into(),
                servers.join(", "),
            ));
        }
    }
    out
}

// ---------------------------------------------------------------------------
// SEC-011 tool-name-squat
// ---------------------------------------------------------------------------

/// A tool name on one server that is a near-duplicate of a tool name on a
/// different server: edit distance 1, or the same name once separators and case
/// are normalized. This catches a typosquat that a human skims past. Exact
/// duplicates are SEC-010's job and are excluded here.
fn tool_name_squat(tools: &[ToolRef<'_>]) -> Vec<SecurityFinding> {
    let mut out = Vec::new();
    for (i, a) in tools.iter().enumerate() {
        for b in &tools[i + 1..] {
            if a.server == b.server || a.name == b.name {
                continue;
            }
            if is_near_duplicate(&a.name, &b.name) {
                out.push(finding(
                    "SEC-011",
                    "tool-name-squat",
                    SecuritySeverity::Medium,
                    a.name.clone(),
                    "a tool name closely resembles another server's tool name, a likely typosquat"
                        .into(),
                    format!("{} ({}) vs {} ({})", a.name, a.server, b.name, b.server),
                ));
            }
        }
    }
    out
}

// ---------------------------------------------------------------------------
// SEC-012 server-name-squat
// ---------------------------------------------------------------------------

/// A server name that is a near-duplicate of another server name, by the same
/// edit-distance-1 or separator/case rule. An impersonating server name is how
/// a shadow server gets installed next to the one a user meant to add.
fn server_name_squat(servers: &[(String, Value)]) -> Vec<SecurityFinding> {
    let mut out = Vec::new();
    for (i, (a, _)) in servers.iter().enumerate() {
        for (b, _) in &servers[i + 1..] {
            if a == b {
                continue;
            }
            if is_near_duplicate(a, b) {
                out.push(finding(
                    "SEC-012",
                    "server-name-squat",
                    SecuritySeverity::Medium,
                    a.clone(),
                    "a server name closely resembles another server name, a likely impersonation"
                        .into(),
                    format!("{a} vs {b}"),
                ));
            }
        }
    }
    out
}

// ---------------------------------------------------------------------------
// SEC-013 ambiguous-resolution
// ---------------------------------------------------------------------------

/// Two servers expose the same tool name with different input schemas. Unlike a
/// plain duplicate, the diverging schema means a client cannot even be sure the
/// two tools take the same arguments, so which one resolves changes behavior.
/// One finding per name that collides with a schema mismatch.
fn ambiguous_resolution(tools: &[ToolRef<'_>]) -> Vec<SecurityFinding> {
    let mut out = Vec::new();
    for (i, tool) in tools.iter().enumerate() {
        if tools[..i].iter().any(|t| t.name == tool.name) {
            continue;
        }
        let same_name: Vec<&ToolRef<'_>> = tools.iter().filter(|t| t.name == tool.name).collect();
        if same_name.len() < 2 {
            continue;
        }
        // Ambiguous only when at least two of the same-named tools disagree on
        // schema. Identical schemas are an ordinary duplicate (SEC-010).
        let first = same_name[0].schema;
        if same_name.iter().any(|t| t.schema != first) {
            let servers: Vec<&str> = same_name.iter().map(|t| t.server).collect();
            out.push(finding(
                "SEC-013",
                "ambiguous-resolution",
                SecuritySeverity::Medium,
                tool.name.clone(),
                "the same tool name has different input schemas across servers, so resolution is ambiguous".into(),
                servers.join(", "),
            ));
        }
    }
    out
}

// ---------------------------------------------------------------------------
// Near-duplicate heuristic
// ---------------------------------------------------------------------------

/// True when two names are confusable: identical after normalizing separators
/// and case, or within Levenshtein distance 1. Returns false for equal inputs,
/// since an exact match is a duplicate (SEC-010), not a squat.
fn is_near_duplicate(a: &str, b: &str) -> bool {
    if a == b {
        return false;
    }
    if normalize(a) == normalize(b) {
        return true;
    }
    levenshtein_within_one(a, b)
}

/// Lowercase the name and strip the `-`, `_`, `.`, and ` ` separators, so
/// `get_user`, `get-user`, and `GetUser` collapse to one form.
fn normalize(name: &str) -> String {
    name.chars()
        .filter(|c| !matches!(c, '-' | '_' | '.' | ' '))
        .flat_map(|c| c.to_lowercase())
        .collect()
}

/// True when `a` and `b` are within Levenshtein distance 1 (one insertion,
/// deletion, or substitution). A length gap above 1 cannot be one edit, so it
/// short-circuits before the full computation.
fn levenshtein_within_one(a: &str, b: &str) -> bool {
    let ac: Vec<char> = a.chars().collect();
    let bc: Vec<char> = b.chars().collect();
    if ac.len().abs_diff(bc.len()) > 1 {
        return false;
    }
    levenshtein(&ac, &bc) <= 1
}

/// Standard dynamic-programming Levenshtein distance over char slices, with a
/// single rolling row so the cost is O(len_a * len_b) time and O(len_b) space.
fn levenshtein(a: &[char], b: &[char]) -> usize {
    let mut prev: Vec<usize> = (0..=b.len()).collect();
    let mut curr = vec![0usize; b.len() + 1];
    for (i, ca) in a.iter().enumerate() {
        curr[0] = i + 1;
        for (j, cb) in b.iter().enumerate() {
            let cost = if ca == cb { 0 } else { 1 };
            curr[j + 1] = (prev[j + 1] + 1).min(curr[j] + 1).min(prev[j] + cost);
        }
        std::mem::swap(&mut prev, &mut curr);
    }
    prev[b.len()]
}

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

    fn server(name: &str, tools: Value) -> (String, Value) {
        (name.to_string(), json!({ "tools": tools }))
    }

    #[test]
    fn duplicate_tool_name_fires_across_servers_and_is_clean_when_unique() {
        let dup = vec![
            server("alpha", json!([{ "name": "search" }])),
            server("beta", json!([{ "name": "search" }])),
        ];
        let found = duplicate_tool_name(&flatten_tools(&dup));
        assert_eq!(found.len(), 1);
        assert_eq!(found[0].rule_id, "SEC-010");
        assert_eq!(found[0].definition, "search");

        let unique = vec![
            server("alpha", json!([{ "name": "search" }])),
            server("beta", json!([{ "name": "lookup" }])),
        ];
        assert!(duplicate_tool_name(&flatten_tools(&unique)).is_empty());
    }

    #[test]
    fn tool_name_squat_fires_on_typo_and_separator_variants_only() {
        // Edit distance 1 (missing letter) across two servers.
        let typo = vec![
            server("alpha", json!([{ "name": "delete_record" }])),
            server("beta", json!([{ "name": "delete_recrd" }])),
        ];
        let found = tool_name_squat(&flatten_tools(&typo));
        assert_eq!(found.len(), 1);
        assert_eq!(found[0].rule_id, "SEC-011");

        // Separator and case difference only.
        let sep = vec![
            server("alpha", json!([{ "name": "get_user" }])),
            server("beta", json!([{ "name": "GetUser" }])),
        ];
        assert_eq!(tool_name_squat(&flatten_tools(&sep)).len(), 1);

        // Unrelated names do not fire.
        let clean = vec![
            server("alpha", json!([{ "name": "search" }])),
            server("beta", json!([{ "name": "weather" }])),
        ];
        assert!(tool_name_squat(&flatten_tools(&clean)).is_empty());
    }

    #[test]
    fn tool_name_squat_ignores_exact_duplicates_and_same_server() {
        // Exact duplicate is SEC-010, not a squat.
        let exact = vec![
            server("alpha", json!([{ "name": "search" }])),
            server("beta", json!([{ "name": "search" }])),
        ];
        assert!(tool_name_squat(&flatten_tools(&exact)).is_empty());

        // Two similar names on the same server are not a cross-server squat.
        let same = vec![server(
            "alpha",
            json!([{ "name": "get_user" }, { "name": "get_users" }]),
        )];
        assert!(tool_name_squat(&flatten_tools(&same)).is_empty());
    }

    #[test]
    fn server_name_squat_fires_on_near_duplicate_server_name() {
        let squat = vec![server("github", json!([])), server("gith_b", json!([]))];
        let found = server_name_squat(&squat);
        assert_eq!(found.len(), 1);
        assert_eq!(found[0].rule_id, "SEC-012");

        let clean = vec![server("github", json!([])), server("gitlab", json!([]))];
        assert!(server_name_squat(&clean).is_empty());
    }

    #[test]
    fn ambiguous_resolution_fires_only_on_schema_mismatch() {
        let mismatch = vec![
            server(
                "alpha",
                json!([{ "name": "search", "inputSchema": { "properties": { "q": {} } } }]),
            ),
            server(
                "beta",
                json!([{ "name": "search", "inputSchema": { "properties": { "query": {} } } }]),
            ),
        ];
        let found = ambiguous_resolution(&flatten_tools(&mismatch));
        assert_eq!(found.len(), 1);
        assert_eq!(found[0].rule_id, "SEC-013");

        // Same name, identical schema is a plain duplicate, not ambiguous.
        let identical = vec![
            server(
                "alpha",
                json!([{ "name": "search", "inputSchema": { "properties": { "q": {} } } }]),
            ),
            server(
                "beta",
                json!([{ "name": "search", "inputSchema": { "properties": { "q": {} } } }]),
            ),
        ];
        assert!(ambiguous_resolution(&flatten_tools(&identical)).is_empty());
    }

    #[test]
    fn namespace_findings_reports_in_rule_id_order() {
        let servers = vec![
            server(
                "alpha",
                json!([{ "name": "search", "inputSchema": { "properties": { "q": {} } } }]),
            ),
            server(
                "beta",
                json!([{ "name": "search", "inputSchema": { "properties": { "query": {} } } }]),
            ),
        ];
        let found = namespace_findings(&servers);
        // A duplicate name with a diverging schema fires both SEC-010 and SEC-013.
        assert!(found.iter().any(|f| f.rule_id == "SEC-010"));
        assert!(found.iter().any(|f| f.rule_id == "SEC-013"));
    }

    #[test]
    fn near_duplicate_rejects_distance_two() {
        assert!(!is_near_duplicate("search", "serch_x"));
        assert!(is_near_duplicate("search", "serch"));
    }

    #[test]
    fn levenshtein_counts_single_edits() {
        let s: Vec<char> = "kitten".chars().collect();
        let t: Vec<char> = "sitting".chars().collect();
        assert_eq!(levenshtein(&s, &t), 3);
    }
}
