mirror of
https://github.com/q3alique/codeflow
synced 2026-06-21 14:06:02 +00:00
q3alique
acc1b4f3e7
Static taint-analysis and visualization tool for source-code security review. Supports deep analysis for Python, JavaScript/TypeScript, Java, Go, and C#, with structural support for all other languages via the generic extractor. Outputs: interactive HTML report, LLM-ready Markdown review document, and optional Burp Suite JSON export. Self-bootstrapping launcher (run.py) requires no virtual environment.
111 lines
3.6 KiB
Python
111 lines
3.6 KiB
Python
"""Task 025: CLI Python application end-to-end integration test."""
|
|
import pytest
|
|
import json
|
|
from pathlib import Path
|
|
from codeflow.orchestrator import run_analysis
|
|
from codeflow.models.node import NodeType
|
|
from codeflow.models.edge import EdgeType
|
|
from tests.fixtures import CLI_FIXTURE_DIR
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def cli_result(tmp_path_factory):
|
|
output = tmp_path_factory.mktemp("cli_out") / "report.html"
|
|
return run_analysis(
|
|
repo_path=CLI_FIXTURE_DIR,
|
|
output_path=output,
|
|
generate_burp=True,
|
|
)
|
|
|
|
|
|
class TestCLIIntegration:
|
|
def test_output_files_exist(self, cli_result):
|
|
assert Path(cli_result["html_path"]).exists()
|
|
burp_path = Path(cli_result["burp_path"])
|
|
assert burp_path.exists()
|
|
|
|
def test_file_count(self, cli_result):
|
|
assert cli_result["file_count"] >= 2
|
|
|
|
def test_node_count(self, cli_result):
|
|
assert cli_result["node_count"] >= 5
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def cli_graph():
|
|
from codeflow.discovery.file_scanner import scan
|
|
from codeflow.analysis.graph_builder import build
|
|
from codeflow.analysis.taint_tracker import propagate_taint
|
|
from codeflow.analysis.boundary_marker import mark_dangerous_boundaries
|
|
file_infos = scan(CLI_FIXTURE_DIR)
|
|
g = build(file_infos)
|
|
g = propagate_taint(g)
|
|
g = mark_dangerous_boundaries(g)
|
|
return g
|
|
|
|
|
|
class TestCLIGraphStructure:
|
|
def test_has_module_nodes(self, cli_graph):
|
|
mods = cli_graph.nodes_by_type(NodeType.MODULE)
|
|
assert len(mods) >= 2
|
|
|
|
def test_has_source_nodes(self, cli_graph):
|
|
sources = cli_graph.nodes_by_type(NodeType.SOURCE)
|
|
assert len(sources) >= 1
|
|
|
|
def test_argv_source_detected(self, cli_graph):
|
|
found = False
|
|
for n in cli_graph.nodes():
|
|
node = cli_graph.get_node(n)
|
|
if node and node.node_type == NodeType.SOURCE and "argv" in node.label:
|
|
found = True
|
|
break
|
|
assert found
|
|
|
|
def test_has_sink_nodes(self, cli_graph):
|
|
sinks = cli_graph.nodes_by_type(NodeType.SINK)
|
|
assert len(sinks) >= 2
|
|
|
|
def test_file_read_sink(self, cli_graph):
|
|
found = False
|
|
for n in cli_graph.nodes():
|
|
node = cli_graph.get_node(n)
|
|
if node and node.node_type == NodeType.SINK and node.metadata.get("sink_type") == "file_read":
|
|
found = True
|
|
break
|
|
assert found
|
|
|
|
def test_file_write_sink(self, cli_graph):
|
|
found = False
|
|
for n in cli_graph.nodes():
|
|
node = cli_graph.get_node(n)
|
|
if node and node.node_type == NodeType.SINK and node.metadata.get("sink_type") == "file_write":
|
|
found = True
|
|
break
|
|
assert found
|
|
|
|
def test_command_execution_sink(self, cli_graph):
|
|
found = False
|
|
for n in cli_graph.nodes():
|
|
node = cli_graph.get_node(n)
|
|
if node and node.node_type == NodeType.SINK and node.metadata.get("sink_type") == "command_execution":
|
|
found = True
|
|
break
|
|
assert found
|
|
|
|
def test_env_var_detected(self, cli_graph):
|
|
env_vars = cli_graph.nodes_by_type(NodeType.ENV_VAR)
|
|
assert len(env_vars) >= 1
|
|
|
|
def test_calls_edges(self, cli_graph):
|
|
has_calls = False
|
|
for _, _, edge in cli_graph.edges():
|
|
if edge.edge_type == EdgeType.CALLS:
|
|
has_calls = True
|
|
break
|
|
assert has_calls
|
|
|
|
def test_burp_json_valid(self, cli_result):
|
|
data = json.loads(Path(cli_result["burp_path"]).read_text(encoding="utf-8"))
|
|
assert "schema_version" in data
|