test_cli_code_vuln.py

from __future__ import annotations

import json

from security_scanner.cli import main
from security_scanner.core.vulnerability.model import (
    VulnerabilityFinding,
    VulnerabilityLocation,
)
from security_scanner.storage.vulnerability_jsonl_store import VulnerabilityJsonlStore


def _sarif_payload() -> dict:
    return {
        "version": "2.1.0",
        "runs": [
            {
                "tool": {
                    "driver": {
                        "name": "Semgrep OSS",
                        "semanticVersion": "1.2.3",
                        "rules": [
                            {
                                "id": "python.lang.security.audit.sql-injection",
                                "name": "SQL injection",
                                "properties": {
                                    "precision": "high",
                                    "security-severity": "8.2",
                                    "tags": ["external/cwe/cwe-089"],
                                },
                            }
                        ],
                    }
                },
                "results": [
                    {
                        "ruleId": "python.lang.security.audit.sql-injection",
                        "message": {"text": "Potential SQL injection."},
                        "locations": [
                            {
                                "physicalLocation": {
                                    "artifactLocation": {"uri": "src/app.py"},
                                    "region": {"startLine": 12},
                                }
                            }
                        ],
                        "partialFingerprints": {
                            "primaryLocationLineHash": "abc123"
                        },
                    }
                ],
            }
        ],
    }


def _finding(**overrides) -> VulnerabilityFinding:
    defaults = dict(
        finding_id="vuln_1234567890abcdef",
        source_tool="semgrep",
        rule_id="python.lang.security.audit.sql-injection",
        rule_name="SQL injection",
        message="Potential SQL injection.",
        severity="HIGH",
        precision="HIGH",
        cwe_ids=("CWE-89",),
        primary_location=VulnerabilityLocation(
            file_path="src/app.py",
            line_start=12,
        ),
    )
    defaults.update(overrides)
    return VulnerabilityFinding(**defaults)


def test_import_sarif_cli_writes_vuln_finding_artifact(tmp_path, capsys):
    sarif = tmp_path / "result.sarif"
    output = tmp_path / "vuln-findings.jsonl"
    sarif.write_text(json.dumps(_sarif_payload()), encoding="utf-8")

    assert main(["import-sarif", "--sarif", str(sarif), "-o", str(output)]) == 0

    assert len(VulnerabilityJsonlStore(output).read_all()) == 1
    captured = capsys.readouterr().out
    assert "Imported 1 code-vuln finding(s)" in captured


def test_import_sarif_cli_can_redact_relative_paths(tmp_path):
    sarif = tmp_path / "result.sarif"
    output = tmp_path / "vuln-findings.jsonl"
    sarif.write_text(json.dumps(_sarif_payload()), encoding="utf-8")

    assert (
        main(
            [
                "import-sarif",
                "--sarif",
                str(sarif),
                "-o",
                str(output),
                "--path-policy",
                "redacted",
            ]
        )
        == 0
    )

    finding = VulnerabilityJsonlStore(output).read_all()[0]
    assert finding.primary_location.path_kind == "relative-redacted"
    assert "src/app.py" not in finding.primary_location.file_path


def test_code_vuln_report_gate_and_evaluate(tmp_path, capsys):
    findings = tmp_path / "vuln-findings.jsonl"
    expected = tmp_path / "expected.json"
    VulnerabilityJsonlStore(findings).write_all([_finding()])
    expected.write_text(
        json.dumps(
            {
                "schemaVersion": 1,
                "name": "synthetic-code-vuln",
                "expectedFindings": [
                    {
                        "filePath": "src/app.py",
                        "lineStart": 12,
                        "ruleId": "python.lang.security.audit.sql-injection",
                    }
                ],
            }
        ),
        encoding="utf-8",
    )

    assert main(["report", "--category", "code-vuln", "--findings", str(findings)]) == 0
    assert "Code Vulnerability Report" in capsys.readouterr().out

    assert main(["gate", "--category", "code-vuln", "--findings", str(findings)]) == 1
    assert "FAIL: 1 blocking code-vuln finding(s)" in capsys.readouterr().out

    assert (
        main(
            [
                "gate",
                "--category",
                "code-vuln",
                "--findings",
                str(findings),
                "--max",
                "1",
            ]
        )
        == 0
    )

    assert (
        main(
            [
                "evaluate",
                "--category",
                "code-vuln",
                "--findings",
                str(findings),
                "--expected",
                str(expected),
            ]
        )
        == 0
    )
    captured = capsys.readouterr().out
    assert "Code Vulnerability Evaluation Report" in captured
    assert "True positives: 1" in captured
    assert "Precision: 1.0000" in captured


def test_scan_vuln_cli_delegates_to_runtime(monkeypatch, tmp_path, capsys):
    calls = []
    output = tmp_path / "vuln-findings.jsonl"

    class Result:
        output_path = output
        sarif_path = tmp_path / "scan.sarif"
        finding_count = 2

    def fake_run(request):
        calls.append(request)
        return Result()

    monkeypatch.setattr(
        "security_scanner.cli.commands.vulnerability.run_vulnerability_scan",
        fake_run,
    )

    assert (
        main(
            [
                "scan-vuln",
                "--root",
                "synthetic-repo",
                "--semgrep-config",
                "auto",
                "-o",
                str(output),
            ]
        )
        == 0
    )

    assert calls[0].root == "synthetic-repo"
    assert calls[0].output_path == str(output)
    assert calls[0].path_policy == "redacted"
    assert "imported 2 code-vuln finding(s)" in capsys.readouterr().out


def test_code_vuln_category_rejects_dynamodb_storage(capsys):
    assert (
        main(
            [
                "report",
                "--category",
                "code-vuln",
                "--storage-backend",
                "dynamodb",
                "--dynamodb-table",
                "SecurityScannerLocal",
            ]
        )
        == 2
    )
    assert "JSONL artifacts only" in capsys.readouterr().err