repotoire 0.8.0

//! AST-driven extraction of [`super::predict::Evidence`] for Python
//! path-traversal call sites.
//!
//! # Why a separate module
//!
//! The scorer in [`super::predict`] takes plain data
//! ([`super::predict::Evidence`]) so it can be unit-tested without an
//! AST. This module's job is to populate that struct from a
//! `tree_sitter::Node` for a Python `call` expression.
//!
//! Splitting the two halves matches Phase 2a's
//! `insecure_crypto::evidence` and the typical Phase 1 ↔ Phase 2 split:
//! the predictor knows about weights and signals; the extractor knows
//! about Python AST shapes.
//!
//! # What this module knows about
//!
//! - Walking up from the call node to the enclosing
//!   `function_definition` (for parameter names + name) and
//!   `class_definition` (informational).
//! - Classifying the first positional argument's origin into one of
//!   `Literal` / `ConfigSource` / `RequestSource` / `Parameter` / `Unknown`.
//! - Detecting `os.path.basename(...)` wrapping any positional argument
//!   (defensive idiom).
//! - Reading the source line for `# repotoire: internal-path[<reason>]`
//!   or `# repotoire: user-controlled[<source>]` annotations.
//!
//! # What this module deliberately does NOT do
//!
//! - Does not look for evidence in non-Python languages. Phase 2b
//!   scope is Python-only per decisions doc D2.
//! - Does not follow data flow across statements. The first-arg-origin
//!   classification looks at the syntactic form of the argument: bare
//!   identifier matched against the enclosing function's parameters,
//!   attribute chain matched against config/request lexicons, string
//!   literal, or unknown. Anything fancier is a future Phase.
//! - Does not consult the graph (Phase 1c) for enclosing scope. AST
//!   walking is sufficient and avoids a detector → graph dependency
//!   we don't need yet (matches Phase 2a's posture).

use super::predict::{
    extract_internal_path_reason, extract_user_controlled_source, matches_config_object,
    matches_request_object, Evidence, FirstArgOrigin,
};
use crate::detectors::security::ast_helpers::{
    collect_named_args, enclosing_python_function, node_text, python_function_param_names,
};
use tree_sitter::Node;

/// Extract typed evidence from a Python path-traversal call node.
///
/// `call_node` must be a `call` AST node whose function names a
/// path-traversal-relevant API (e.g. `os.path.join`, `open`,
/// `flask.send_file`). `source` is the file's raw bytes. `lines` is
/// the pre-split source-line slice the scanner already builds; we use
/// it only to read the source line for annotation lookup.
///
/// Never panics; missing fields produce `None`/`false`/`Unknown` in
/// the corresponding Evidence field.
pub(super) fn extract_python_evidence<'a>(
    call_node: Node<'a>,
    source: &'a [u8],
    lines: &[&str],
) -> Evidence {
    let mut ev = Evidence::default();

    // ── Enclosing function (for name + parameter list) and class. ──
    let enclosing_fn = enclosing_python_function(call_node);
    if let Some(fn_node) = enclosing_fn {
        if let Some(name_node) = fn_node.child_by_field_name("name") {
            if let Some(name) = node_text(name_node, source) {
                ev.enclosing_function = Some(name.to_string());
            }
        }
    }
    ev.enclosing_class = enclosing_python_class_name(call_node, source);

    let param_names: Vec<String> = enclosing_fn
        .map(|fn_node| python_function_param_names(fn_node, source))
        .unwrap_or_default();

    // ── First positional argument origin classification. ──
    let mut positional_args: Vec<Node<'_>> = Vec::new();
    if let Some(args_node) = call_node.child_by_field_name("arguments") {
        let all = collect_named_args(args_node);
        positional_args = all
            .into_iter()
            .filter(|a| a.kind() != "keyword_argument" && a.kind() != "comment")
            .collect();
    }
    if let Some(first) = positional_args.first() {
        ev.first_arg_origin = Some(classify_first_arg_origin(*first, source, &param_names));
    }

    // ── basename wrapping (any positional arg). ──
    ev.basename_applied = positional_args
        .iter()
        .any(|arg| expression_contains_os_path_basename(*arg, source));

    // ── Source-line annotations. ──
    let line_idx = call_node.start_position().row;
    if let Some(line) = lines.get(line_idx) {
        ev.internal_path_annotation = extract_internal_path_reason(line);
        ev.user_controlled_annotation = extract_user_controlled_source(line);
    }

    ev
}

// ─────────────────────────────────────────────────────────────────────────────
// First-arg classification
// ─────────────────────────────────────────────────────────────────────────────

/// Classify a Python expression node into a [`FirstArgOrigin`].
///
/// Decisions, in priority order:
///
/// 1. String literal (`"foo"`, `b"foo"`, f-string with no interpolation)
///    → `Literal`.
/// 2. Bare identifier that matches a parameter of the enclosing
///    function → `Parameter { name }`.
/// 3. Attribute access whose textual chain matches the request lexicon
///    → `RequestSource`.
/// 4. Attribute access or call whose textual chain matches the config
///    lexicon → `ConfigSource`.
/// 5. Subscript on a request-object identifier (`request.GET["x"]`) →
///    `RequestSource`.
/// 6. Anything else → `Unknown`.
///
/// String literals with f-string interpolation are treated as `Unknown`
/// (not `Literal`) because the interpolated value may itself be
/// untrusted; conservative posture matches the existing
/// `classify_path_arg_python` logic.
fn classify_first_arg_origin(
    node: Node<'_>,
    source: &[u8],
    param_names: &[String],
) -> FirstArgOrigin {
    match node.kind() {
        "string" => {
            // Pure string literal unless it contains an interpolation.
            let mut cursor = node.walk();
            for child in node.children(&mut cursor) {
                if child.kind() == "interpolation" {
                    return FirstArgOrigin::Unknown;
                }
            }
            FirstArgOrigin::Literal
        }
        "concatenated_string" => {
            // `"foo" "bar"` adjacent literal concat: Literal if all are
            // literals, Unknown otherwise.
            let mut cursor = node.walk();
            for child in node.children(&mut cursor) {
                if classify_first_arg_origin(child, source, param_names) != FirstArgOrigin::Literal
                {
                    return FirstArgOrigin::Unknown;
                }
            }
            FirstArgOrigin::Literal
        }
        "identifier" => {
            let Some(name) = node_text(node, source) else {
                return FirstArgOrigin::Unknown;
            };
            if param_names.iter().any(|p| p == name) {
                FirstArgOrigin::Parameter {
                    name: name.to_string(),
                }
            } else {
                FirstArgOrigin::Unknown
            }
        }
        "attribute" | "call" | "subscript" => {
            // Use the full source text of the node for lexicon match.
            // This catches `flask.request.args.get("x")`,
            // `request.GET["file"]`, `settings.BASE_DIR`,
            // `os.environ.get("HOME")` uniformly.
            let Some(text) = node_text(node, source) else {
                return FirstArgOrigin::Unknown;
            };
            // Request lexicon takes priority over config: a hypothetical
            // identifier like `settings_request` shouldn't matter (the
            // lexicons don't overlap on real names) but the priority is
            // documented here for determinism.
            if matches_request_object(text) {
                FirstArgOrigin::RequestSource
            } else if matches_config_object(text) {
                FirstArgOrigin::ConfigSource
            } else {
                FirstArgOrigin::Unknown
            }
        }
        "parenthesized_expression" => {
            // Unwrap one level of parens and recurse.
            for i in 0..node.named_child_count() {
                if let Some(c) = node.named_child(i) {
                    return classify_first_arg_origin(c, source, param_names);
                }
            }
            FirstArgOrigin::Unknown
        }
        _ => FirstArgOrigin::Unknown,
    }
}

// ─────────────────────────────────────────────────────────────────────────────
// Helpers
// ─────────────────────────────────────────────────────────────────────────────

/// Walk up from `node` to the nearest enclosing `class_definition` and
/// return its name. Returns `None` at module level or when only enclosed
/// by functions.
///
/// (Same shape as Phase 2a's `enclosing_python_class_name`. Will be
/// hoisted to `ast_helpers.rs` during the Phase 2c rule-of-three
/// consolidation.)
fn enclosing_python_class_name<'a>(node: Node<'a>, source: &'a [u8]) -> Option<String> {
    let mut cur = node.parent()?;
    loop {
        if cur.kind() == "class_definition" {
            let name = cur.child_by_field_name("name")?;
            return node_text(name, source).map(str::to_string);
        }
        if cur.kind() == "module" {
            return None;
        }
        cur = cur.parent()?;
    }
}

/// True if `expr` (or any of its descendants) is a call to
/// `os.path.basename(...)`.
///
/// Walks the entire subtree; matching is by attribute name + receiver
/// chain. Catches `os.path.basename(name)`, raw imports
/// `from os.path import basename; basename(name)`, and indirect uses
/// inside binary expressions.
fn expression_contains_os_path_basename<'a>(expr: Node<'a>, source: &'a [u8]) -> bool {
    if is_os_path_basename_call(expr, source) {
        return true;
    }
    let mut cursor = expr.walk();
    for child in expr.children(&mut cursor) {
        if expression_contains_os_path_basename(child, source) {
            return true;
        }
    }
    false
}

fn is_os_path_basename_call<'a>(node: Node<'a>, source: &'a [u8]) -> bool {
    if node.kind() != "call" {
        return false;
    }
    let Some(func) = node.child_by_field_name("function") else {
        return false;
    };
    match func.kind() {
        "attribute" => {
            // os.path.basename(...) — receiver is os.path, attr is basename.
            let Some(attr) = func.child_by_field_name("attribute") else {
                return false;
            };
            if node_text(attr, source) != Some("basename") {
                return false;
            }
            // Receiver must end in `.path` (typically `os.path`).
            let Some(obj) = func.child_by_field_name("object") else {
                return false;
            };
            // Either bare `path` (from `from os import path`) or
            // attribute ending in `.path`.
            match obj.kind() {
                "identifier" => node_text(obj, source) == Some("path"),
                "attribute" => obj
                    .child_by_field_name("attribute")
                    .and_then(|a| node_text(a, source))
                    .map(|n| n == "path")
                    .unwrap_or(false),
                _ => false,
            }
        }
        "identifier" => {
            // Bare `basename(name)` — likely from `from os.path import basename`.
            node_text(func, source) == Some("basename")
        }
        _ => false,
    }
}

// ─────────────────────────────────────────────────────────────────────────────
// Tests
// ─────────────────────────────────────────────────────────────────────────────

#[cfg(test)]
mod tests {
    use super::*;
    use crate::detectors::ast_fingerprint::parse_root_ext;
    use crate::parsers::lightweight::Language;

    /// Parse `source` as Python and find the first `call` node whose
    /// function chain ends in the named attribute. Anchors evidence
    /// extraction tests on a specific call site without index counting.
    fn first_call_with_attr<'tree>(
        tree: &'tree tree_sitter::Tree,
        source: &[u8],
        attr_name: &str,
    ) -> tree_sitter::Node<'tree> {
        fn walk<'a>(
            node: tree_sitter::Node<'a>,
            source: &[u8],
            attr_name: &str,
        ) -> Option<tree_sitter::Node<'a>> {
            if node.kind() == "call" {
                if let Some(func) = node.child_by_field_name("function") {
                    if func.kind() == "attribute" {
                        if let Some(attr) = func.child_by_field_name("attribute") {
                            if node_text(attr, source) == Some(attr_name) {
                                return Some(node);
                            }
                        }
                    } else if func.kind() == "identifier"
                        && node_text(func, source) == Some(attr_name)
                    {
                        return Some(node);
                    }
                }
            }
            let mut cursor = node.walk();
            for child in node.children(&mut cursor) {
                if let Some(found) = walk(child, source, attr_name) {
                    return Some(found);
                }
            }
            None
        }
        walk(tree.root_node(), source, attr_name)
            .unwrap_or_else(|| panic!("no call ending in `.{attr_name}` in source"))
    }

    fn extract(source: &str, attr: &str) -> Evidence {
        let bytes = source.as_bytes();
        let tree = parse_root_ext(source, Language::Python, "py").expect("parse python");
        let lines: Vec<&str> = source.lines().collect();
        let call = first_call_with_attr(&tree, bytes, attr);
        extract_python_evidence(call, bytes, &lines)
    }

    // ── Enclosing scope ──

    #[test]
    fn extracts_enclosing_class_and_function() {
        let src = "
import os
class FileServer:
    def serve(self, name):
        return open(os.path.join('/var/www', name))
";
        let ev = extract(src, "open");
        assert_eq!(ev.enclosing_class.as_deref(), Some("FileServer"));
        assert_eq!(ev.enclosing_function.as_deref(), Some("serve"));
    }

    #[test]
    fn module_level_call_has_no_enclosing_function() {
        let src = "import os\nopen(os.path.join('/tmp', 'x'))\n";
        let ev = extract(src, "open");
        assert!(ev.enclosing_function.is_none());
        assert!(ev.enclosing_class.is_none());
    }

    // ── First-arg origin: literal ──

    #[test]
    fn literal_first_arg_classified_as_literal() {
        let src = "import os\np = os.path.join('/var/www', x)\n";
        let ev = extract(src, "join");
        assert_eq!(ev.first_arg_origin, Some(FirstArgOrigin::Literal));
    }

    #[test]
    fn bytes_literal_first_arg_classified_as_literal() {
        let src = "import os\np = os.path.join(b'/var/www', x)\n";
        let ev = extract(src, "join");
        assert_eq!(ev.first_arg_origin, Some(FirstArgOrigin::Literal));
    }

    #[test]
    fn fstring_with_interpolation_first_arg_is_unknown() {
        // Conservative: f-string with interpolation could carry untrusted
        // data, so don't claim Literal.
        let src = "import os\np = os.path.join(f'/var/{base}', x)\n";
        let ev = extract(src, "join");
        assert_eq!(ev.first_arg_origin, Some(FirstArgOrigin::Unknown));
    }

    // ── First-arg origin: parameter ──

    #[test]
    fn first_arg_matching_parameter_classified_as_parameter() {
        let src = "
import os
def serve(name):
    return open(os.path.join('/var/www', name))
";
        // Note: in this case the FIRST arg is the literal `'/var/www'`,
        // not `name`. Pin that. The Parameter signal would hit if `name`
        // were first.
        let ev = extract(src, "join");
        assert_eq!(ev.first_arg_origin, Some(FirstArgOrigin::Literal));
    }

    #[test]
    fn parameter_in_first_position_classified_as_parameter() {
        let src = "
import os
def get_path(folder, name):
    return os.path.join(folder, name)
";
        let ev = extract(src, "join");
        match ev.first_arg_origin {
            Some(FirstArgOrigin::Parameter { ref name }) if name == "folder" => {}
            other => panic!("expected Parameter {{ folder }}, got {other:?}"),
        }
    }

    #[test]
    fn non_param_identifier_in_first_position_is_unknown() {
        let src = "
import os
def make():
    folder = compute_folder()
    return os.path.join(folder, 'x')
";
        let ev = extract(src, "join");
        assert_eq!(ev.first_arg_origin, Some(FirstArgOrigin::Unknown));
    }

    // ── First-arg origin: request source ──

    #[test]
    fn request_args_first_arg_classified_as_request() {
        let src = "
def view(request):
    return open(request.GET['file'])
";
        let ev = extract(src, "open");
        assert_eq!(ev.first_arg_origin, Some(FirstArgOrigin::RequestSource));
    }

    #[test]
    fn flask_request_first_arg_classified_as_request() {
        let src = "
import os
from flask import request
def view():
    return os.path.join(request.args['name'])
";
        let ev = extract(src, "join");
        assert_eq!(ev.first_arg_origin, Some(FirstArgOrigin::RequestSource));
    }

    // ── First-arg origin: config source ──

    #[test]
    fn settings_attribute_first_arg_classified_as_config() {
        let src = "
import os
from django.conf import settings
def make():
    return os.path.join(settings.BASE_DIR, 'static')
";
        let ev = extract(src, "join");
        assert_eq!(ev.first_arg_origin, Some(FirstArgOrigin::ConfigSource));
    }

    #[test]
    fn os_environ_get_first_arg_classified_as_config() {
        let src = "
import os
def make():
    return os.path.join(os.environ.get('HOME'), 'data')
";
        let ev = extract(src, "join");
        assert_eq!(ev.first_arg_origin, Some(FirstArgOrigin::ConfigSource));
    }

    #[test]
    fn os_path_expanduser_first_arg_classified_as_config() {
        let src = "
import os
def make():
    return os.path.join(os.path.expanduser('~'), 'data')
";
        let ev = extract(src, "join");
        assert_eq!(ev.first_arg_origin, Some(FirstArgOrigin::ConfigSource));
    }

    // ── basename wrapping ���─

    #[test]
    fn os_path_basename_wrapping_detected() {
        let src = "
import os
def serve(name):
    return open(os.path.join('/var/www', os.path.basename(name)))
";
        let ev = extract(src, "join");
        assert!(ev.basename_applied);
    }

    #[test]
    fn no_basename_wrapping_not_detected() {
        let src = "
import os
def serve(name):
    return open(os.path.join('/var/www', name))
";
        let ev = extract(src, "join");
        assert!(!ev.basename_applied);
    }

    #[test]
    fn bare_basename_call_detected() {
        // `from os.path import basename` style.
        let src = "
import os
from os.path import basename
def serve(name):
    return open(os.path.join('/var/www', basename(name)))
";
        let ev = extract(src, "join");
        assert!(ev.basename_applied);
    }

    // ── Annotations ──

    #[test]
    fn internal_path_annotation_extracted() {
        let src = "import os\np = os.path.join(folder, x)  # repotoire: internal-path[validated]\n";
        let ev = extract(src, "join");
        assert_eq!(ev.internal_path_annotation.as_deref(), Some("validated"));
        assert!(ev.user_controlled_annotation.is_none());
    }

    #[test]
    fn user_controlled_annotation_extracted() {
        let src = "import os\np = os.path.join(folder, x)  # repotoire: user-controlled[GET]\n";
        let ev = extract(src, "join");
        assert_eq!(ev.user_controlled_annotation.as_deref(), Some("GET"));
        assert!(ev.internal_path_annotation.is_none());
    }

    #[test]
    fn no_annotation_yields_none() {
        let src = "import os\np = os.path.join(folder, x)\n";
        let ev = extract(src, "join");
        assert!(ev.internal_path_annotation.is_none());
        assert!(ev.user_controlled_annotation.is_none());
    }

    #[test]
    fn unrelated_annotation_yields_none() {
        let src = "import os\np = os.path.join(folder, x)  # repotoire: protocol-required[RFC]\n";
        let ev = extract(src, "join");
        assert!(ev.internal_path_annotation.is_none());
        assert!(ev.user_controlled_annotation.is_none());
    }

    // ── Integration: Click `utils.py:489` shape ──

    #[test]
    fn click_utils_489_shape_extracts_canonical_signals() {
        // The decisions doc's worked example: Click's `get_app_dir`
        // returns `os.path.join(folder, app_name)` where:
        //   - `folder` is bound from `os.environ.get(...)` earlier
        //     (config source — but we don't follow data flow, so the
        //     bare `folder` identifier classifies as Parameter only IF
        //     it's in the function's param list)
        //   - `app_name` is the function's parameter
        //
        // Pin what the v0 evidence extractor actually produces.
        let src = "
import os
def get_app_dir(app_name, folder=None):
    if folder is None:
        folder = os.environ.get('XDG_CONFIG_HOME', '~/.config')
    return os.path.join(folder, app_name)
";
        let ev = extract(src, "join");
        assert_eq!(ev.enclosing_function.as_deref(), Some("get_app_dir"));
        // First arg is `folder`, which IS a parameter.
        match ev.first_arg_origin {
            Some(FirstArgOrigin::Parameter { ref name }) if name == "folder" => {}
            other => panic!("expected Parameter {{ folder }}, got {other:?}"),
        }
        assert!(!ev.basename_applied);
        assert!(ev.internal_path_annotation.is_none());
    }
}