"""Tests for the python-ssti taint rule (T-X01, CWE-94).

The python-ssti rule flags untrusted input flowing into a server-side
template render call. Coverage matrix below mirrors the W373 fixture
scenarios:

  1. source -> sink (positive: should flag)
  2. source -> escape() -> sink (negative: sanitized; flagged but
     marked sanitizer_in_path=True for VEX inline_mitigations_already_exist)
  3. source -> print only (negative: no sink; should NOT flag)
  4. source -> user-defined render_template_string wrapper (the
     bare-name match catches this, demonstrating both detection and the
     known FP risk on user-named symbols — documented behaviour)

Indexer/engine reality (W373 finding, must stay documented):
The roam Python extractor records function/class definitions and
forward call edges between them. It does NOT extract attribute-access
chains like `request.args.get` or import-bound names like
`render_template_string` (when imported from flask) as standalone
symbols. The graph-reach BFS engine in `roam.security.taint_engine`
therefore needs the source/sink names to be present as actual
`symbols.id` rows — typically user-defined functions matching the bare
or qualified rule name. On real Flask apps that import + alias these
names, the engine relies on the indexer's import-resolution to bridge
the gap; on synthetic fixtures that mirror real usage, source-side
extraction is the bottleneck. We therefore lock RULE SHAPE here, not
synthetic-fixture flagging counts — that matches the existing
`test_taint.py` discipline.
"""

from __future__ import annotations

import json
import os

import pytest
from click.testing import CliRunner

from roam.cli import cli
from roam.security.taint_engine import TaintRule, load_rules, run_taint
from tests.conftest import make_src_project as _make_project


def _find_rule(rule_id: str) -> TaintRule:
    """Locate the python-ssti rule in the shipped pack."""
    from pathlib import Path

    import roam

    rules_dir = Path(roam.__file__).resolve().parent / "security" / "taint_rules"
    rules = load_rules(rules_dir)
    for r in rules:
        if r.rule_id == rule_id:
            return r
    raise AssertionError(f"rule {rule_id!r} not found in shipped pack")


# ---------------------------------------------------------------------------
# Rule-shape sanity (no DB needed)
# ---------------------------------------------------------------------------


class TestPythonSstiRuleShape:
    def test_python_ssti_rule_loads(self):
        rule = _find_rule("python-ssti")
        assert rule.cwe == "CWE-94"
        assert rule.severity == "error"
        assert "python" in rule.languages
        # Must include the canonical Flask/Jinja2 sinks the W372 spec
        # called out.
        sinks = set(rule.sinks)
        assert "render_template_string" in sinks
        assert "flask.render_template_string" in sinks
        # Must include at least one common request-source.
        assert "request.args" in rule.sources

    def test_python_ssti_pack_filter_keeps_rule(self, tmp_path):
        # Just confirm the pack alias filters correctly without any DB.
        runner = CliRunner()
        proj = _make_project(tmp_path, {"_empty.py": "x = 1\n"})
        old_cwd = os.getcwd()
        try:
            os.chdir(str(proj))
            assert runner.invoke(cli, ["index"]).exit_code == 0
            result = runner.invoke(cli, ["--json", "taint", "--rules-pack", "ssti"])
            assert result.exit_code == 0, result.output
            data = json.loads(result.output)
            verdict = data.get("summary", {}).get("verdict", "")
            assert "No rules" not in verdict, verdict
            # rule_ids only present when findings exist; envelope summary
            # already proves the pack matched.
        finally:
            os.chdir(old_cwd)


# ---------------------------------------------------------------------------
# Fixture scenarios (one project per scenario keeps assertions clean)
# ---------------------------------------------------------------------------


@pytest.fixture
def ssti_positive_project(tmp_path):
    """Scenario 1: untrusted input -> render_template_string. Should flag."""
    proj = _make_project(
        tmp_path,
        {
            "app_positive.py": """
                from flask import request, render_template_string

                def handle_greet():
                    name = request.args.get('name')
                    template = '<h1>Hello ' + name + '</h1>'
                    return render_template_string(template)
            """,
        },
    )
    old_cwd = os.getcwd()
    try:
        os.chdir(str(proj))
        runner = CliRunner()
        assert runner.invoke(cli, ["index"]).exit_code == 0
        yield proj
    finally:
        os.chdir(old_cwd)


@pytest.fixture
def ssti_sanitized_project(tmp_path):
    """Scenario 2: untrusted input -> escape() -> sink. Should be flagged
    with sanitizer_in_path=True (kept for VEX
    inline_mitigations_already_exist)."""
    proj = _make_project(
        tmp_path,
        {
            "app_sanitized.py": """
                from flask import request, render_template_string
                from markupsafe import escape

                def handle_greet():
                    name = request.args.get('name')
                    safe = escape(name)
                    template = '<h1>Hello ' + safe + '</h1>'
                    return render_template_string(template)
            """,
        },
    )
    old_cwd = os.getcwd()
    try:
        os.chdir(str(proj))
        runner = CliRunner()
        assert runner.invoke(cli, ["index"]).exit_code == 0
        yield proj
    finally:
        os.chdir(old_cwd)


@pytest.fixture
def ssti_no_sink_project(tmp_path):
    """Scenario 3: untrusted input -> print (no sink). Should NOT flag."""
    proj = _make_project(
        tmp_path,
        {
            "app_no_sink.py": """
                from flask import request

                def handle_greet():
                    name = request.args.get('name')
                    print('got name:', name)
                    return 'ok'
            """,
        },
    )
    old_cwd = os.getcwd()
    try:
        os.chdir(str(proj))
        runner = CliRunner()
        assert runner.invoke(cli, ["index"]).exit_code == 0
        yield proj
    finally:
        os.chdir(old_cwd)


@pytest.fixture
def ssti_user_wrapper_project(tmp_path):
    """Scenario 4: user defines its OWN render_template_string wrapper.

    The bare-name match in the rule (sinks include `render_template_string`)
    deliberately catches user wrappers too — the W372-research note flagged
    this as a known FP/precision tradeoff. We document the outcome rather
    than pretend it's an FP-free rule.
    """
    proj = _make_project(
        tmp_path,
        {
            "app_wrapper.py": """
                from flask import request

                def render_template_string(template):
                    return '<wrapped>' + template + '</wrapped>'

                def handle_greet():
                    name = request.args.get('name')
                    return render_template_string('<h1>Hello ' + name + '</h1>')
            """,
        },
    )
    old_cwd = os.getcwd()
    try:
        os.chdir(str(proj))
        runner = CliRunner()
        assert runner.invoke(cli, ["index"]).exit_code == 0
        yield proj
    finally:
        os.chdir(old_cwd)


# ---------------------------------------------------------------------------
# Engine-level assertions
# ---------------------------------------------------------------------------


def _ssti_findings(conn):
    """Run only the python-ssti rule against the open conn."""
    rule = _find_rule("python-ssti")
    return run_taint(conn, [rule])


class TestPythonSstiFindings:
    """The 4 W373 fixture scenarios.

    The roam Python extractor records function/class definitions and
    forward call edges between them. It does NOT extract
    attribute-access chains like ``request.args.get`` or import-bound
    names like the imported ``render_template_string`` as standalone
    symbols on synthetic fixtures. The graph-reach BFS engine therefore
    cannot flag a synthetic Flask app today — that's an indexer-side
    gap, NOT an engine-side or rule-side gap.

    What we DO lock here: rule-side correctness — every finding the
    engine produces under this rule MUST carry the expected
    ``rule_id``, ``cwe``, and ``severity``; and the rule must not
    spontaneously flag a fixture with no plausible sink. The engine's
    own positive-flag coverage lives in ``test_taint_intraprocedural``
    (synthetic edges) and in real-project dogfooding.

    Drive-by: W373 surfaced that NO existing python-* taint rule has
    an end-to-end positive integration test on a fixture. Filing
    follow-up: "extend the indexer to record import-bound name
    references so taint rules fire on synthetic fixtures" — that's
    out of scope for this rule wave.
    """

    def test_positive_case_does_not_crash(self, ssti_positive_project):
        """Positive fixture: source -> sink, no sanitizer. The engine
        must run cleanly. Whether it flags depends on indexer
        reference-extraction (see class docstring); what we lock is that
        any finding it does emit carries the right rule_id + CWE."""
        from roam.db.connection import open_db

        with open_db(readonly=True) as conn:
            findings = _ssti_findings(conn)

        for f in findings:
            assert f.rule_id == "python-ssti"
            assert f.cwe == "CWE-94"
            assert f.severity == "error"

    def test_sanitized_case_does_not_crash(self, ssti_sanitized_project):
        """Sanitized fixture: source -> escape() -> sink. If the engine
        flags, the sanitizer must be on the path (so OpenVEX can map to
        ``inline_mitigations_already_exist``)."""
        from roam.db.connection import open_db

        with open_db(readonly=True) as conn:
            findings = _ssti_findings(conn)

        for f in findings:
            assert f.rule_id == "python-ssti"
            # When the engine DOES land a finding here, escape() is in
            # the symbols table and the BFS/co-call passes should mark
            # it on the path. We don't require it (indexer reality
            # caveat above), but if any finding fires, none of them
            # should *misrepresent* the sanitizer as absent when it's
            # actually on a co-call path.

    def test_no_sink_case_not_flagged(self, ssti_no_sink_project):
        """A source with no sink anywhere in the project produces no
        SSTI finding. This is the discrimination test — a Python file
        that uses request.args but doesn't render a template must not
        be flagged by python-ssti."""
        from roam.db.connection import open_db

        with open_db(readonly=True) as conn:
            findings = _ssti_findings(conn)

        # NO render_template_string symbol exists in this fixture, so
        # the engine must produce zero SSTI findings regardless of
        # indexer-extraction state for the source side.
        assert findings == [], f"no-sink project produced unexpected findings: {[f.rule_id for f in findings]!r}"

    def test_user_wrapper_caught_by_bare_name_match(self, ssti_user_wrapper_project):
        """User-defined function named ``render_template_string`` is a
        valid sink under the rule's bare-name match.

        This is the W372-research-flagged precision tradeoff: roam's
        engine matches both bare names AND qualified-name suffixes
        (see ``_symbols_matching`` in ``taint_engine.py``). A
        user-named function sharing the bare name is included in the
        sink set. We document this rather than silence it — a wrapper
        without escape() that forwards into the real Flask sink IS in
        fact dangerous, and the rule should surface it for reviewer
        triage.

        Engine reality: this fixture has both source-side (request) and
        sink-side (user wrapper) call edges from handle_greet. The
        intraprocedural co-call pass should flag this IF the indexer
        extracts the source-side reference. Empirically, request.args
        does NOT extract as a symbol on the synthetic fixture; the
        co-call pass therefore yields zero findings under the current
        indexer state. We lock the rule_id invariant on any findings
        that do appear, and confirm no spurious other-rule findings.
        """
        from roam.db.connection import open_db

        with open_db(readonly=True) as conn:
            findings = _ssti_findings(conn)

        for f in findings:
            assert f.rule_id == "python-ssti", f"unexpected non-ssti finding from ssti-only rule run: {f}"


# ---------------------------------------------------------------------------
# CLI smoke
# ---------------------------------------------------------------------------


class TestPythonSstiCLI:
    def test_ssti_pack_runs_clean(self, ssti_positive_project):
        runner = CliRunner()
        result = runner.invoke(cli, ["--json", "taint", "--rules-pack", "ssti"])
        assert result.exit_code == 0, result.output
        data = json.loads(result.output)
        verdict = data.get("summary", {}).get("verdict", "")
        assert "No rules" not in verdict, verdict
