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.
97 lines
3.2 KiB
Python
97 lines
3.2 KiB
Python
from __future__ import annotations
|
|
from collections import deque
|
|
from codeflow.models.graph import CodeGraph
|
|
from codeflow.models.node import NodeType
|
|
from codeflow.models.edge import EdgeType
|
|
from codeflow.models.flow import Flow
|
|
|
|
|
|
def _flow_name(node) -> str:
|
|
if node.metadata.get("http_method"):
|
|
return f"{node.metadata['http_method']} {node.metadata.get('route_path', '')}"
|
|
return f"flow: {node.label}"
|
|
|
|
|
|
def identify_flows(graph: CodeGraph) -> list[Flow]:
|
|
entry_points = [
|
|
nid for nid in graph.nodes()
|
|
if (node := graph.get_node(nid)) and node.node_type == NodeType.ENTRY_POINT
|
|
]
|
|
|
|
# Only follow these edge types in the FORWARD direction.
|
|
# CONTAINS is included so clustered sub-nodes are reachable.
|
|
# Predecessors are NOT traversed — that caused cross-flow contamination
|
|
# where an unrelated entry point sharing a callee would be pulled in.
|
|
TRAVERSE_EDGE_TYPES = {
|
|
EdgeType.CALLS, EdgeType.DATA_FLOW, EdgeType.TAINTED_FLOW,
|
|
EdgeType.CONTAINS,
|
|
}
|
|
|
|
flows: list[Flow] = []
|
|
|
|
for ep_id in entry_points:
|
|
ep_node = graph.get_node(ep_id)
|
|
if ep_node is None:
|
|
continue
|
|
|
|
flow = Flow(
|
|
id=ep_id,
|
|
name=_flow_name(ep_node),
|
|
entry_node_id=ep_id,
|
|
metadata={
|
|
"http_method": ep_node.metadata.get("http_method", ""),
|
|
"route_path": ep_node.metadata.get("route_path", ""),
|
|
"framework": ep_node.metadata.get("framework", ""),
|
|
},
|
|
)
|
|
|
|
# Forward-only BFS from the entry point
|
|
visited: set[str] = set()
|
|
queue = deque([ep_id])
|
|
|
|
while queue:
|
|
current = queue.popleft()
|
|
if current in visited:
|
|
continue
|
|
visited.add(current)
|
|
|
|
for successor in graph.successors(current):
|
|
edge = graph.get_edge(current, successor)
|
|
if edge and edge.edge_type in TRAVERSE_EDGE_TYPES:
|
|
if successor not in visited:
|
|
queue.append(successor)
|
|
|
|
# Explicitly pull in SOURCE and ENV_VAR nodes that feed INTO this flow.
|
|
# They are predecessors of the entry point (DATA_FLOW: SOURCE → ENTRY_POINT),
|
|
# so forward BFS alone would never visit them.
|
|
for nid in graph.nodes():
|
|
if nid in visited:
|
|
continue
|
|
n = graph.get_node(nid)
|
|
if n and n.node_type in (NodeType.SOURCE, NodeType.ENV_VAR):
|
|
# Include if at least one outgoing edge lands inside this flow
|
|
if any(succ in visited for succ in graph.successors(nid)):
|
|
visited.add(nid)
|
|
|
|
flow.node_ids = list(visited)
|
|
|
|
flow.source_node_ids = [
|
|
nid for nid in visited
|
|
if (n := graph.get_node(nid)) and n.node_type == NodeType.SOURCE
|
|
]
|
|
|
|
flow.sink_node_ids = [
|
|
nid for nid in visited
|
|
if (n := graph.get_node(nid)) and n.node_type == NodeType.SINK
|
|
]
|
|
|
|
flow.tainted_node_ids = [
|
|
nid for nid in visited
|
|
if (n := graph.get_node(nid)) and n.is_tainted
|
|
]
|
|
|
|
flows.append(flow)
|
|
|
|
flows.sort(key=lambda f: f.name)
|
|
return flows
|