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.
336 lines
12 KiB
Python
336 lines
12 KiB
Python
from __future__ import annotations
|
|
import math
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from codeflow.models.graph import CodeGraph
|
|
from codeflow.models.node import NodeType
|
|
from codeflow.models.flow import Flow
|
|
from collections import defaultdict, deque
|
|
|
|
|
|
@dataclass
|
|
class Position:
|
|
x: float = 0
|
|
y: float = 0
|
|
width: float = 180
|
|
height: float = 56
|
|
|
|
@property
|
|
def left(self) -> float:
|
|
return self.x
|
|
|
|
@property
|
|
def right(self) -> float:
|
|
return self.x + self.width
|
|
|
|
@property
|
|
def top(self) -> float:
|
|
return self.y
|
|
|
|
@property
|
|
def bottom(self) -> float:
|
|
return self.y + self.height
|
|
|
|
|
|
# Base (width, min-height) per node type.
|
|
# Heights grow dynamically with label length — see _node_dims().
|
|
# Widths are slightly larger for shaped nodes (diamond/hexagon) so the
|
|
# inscribed text has room at the widest cross-section.
|
|
NODE_DIMENSIONS: dict[NodeType, tuple[float, float]] = {
|
|
NodeType.MODULE: (210, 70),
|
|
NodeType.FUNCTION: (190, 56),
|
|
NodeType.CLASS: (190, 58),
|
|
NodeType.ENTRY_POINT: (210, 62),
|
|
NodeType.SOURCE: (170, 68), # diamond — needs more vertical room
|
|
NodeType.SINK: (190, 64), # hexagon — needs wider bbox
|
|
NodeType.BOUNDARY: (170, 50),
|
|
NodeType.CLUSTER: (230, 66),
|
|
NodeType.ENV_VAR: (170, 52), # parallelogram
|
|
NodeType.PARAMETER: (170, 52), # parallelogram
|
|
}
|
|
|
|
# Pixels per monospace character at 12 px font, used for line-wrap estimates
|
|
_MONO_PX_PER_CHAR = 7.5
|
|
_LINE_HEIGHT_PX = 16 # px between wrapped lines
|
|
_WRAP_V_PADDING = 36 # reserved for badge + subtitle within the node
|
|
|
|
# Layout spacing constants
|
|
NODE_H_SPACING = 64 # horizontal gap between nodes in the same layer
|
|
LAYER_Y_GAP = 110 # vertical gap between layers
|
|
SVG_PADDING_X = 60 # left/right canvas margin
|
|
SVG_PADDING_Y = 60 # top/bottom canvas margin
|
|
|
|
|
|
def compute_svg_dimensions(positions: dict[str, Position]) -> tuple[float, float]:
|
|
if not positions:
|
|
return 800, 500
|
|
max_x = max(p.x + p.width for p in positions.values())
|
|
max_y = max(p.y + p.height for p in positions.values())
|
|
return max_x + SVG_PADDING_X, max_y + SVG_PADDING_Y
|
|
|
|
|
|
def _node_dims(nid: str, graph: CodeGraph) -> tuple[float, float]:
|
|
"""
|
|
Return (width, height) for a node.
|
|
Height grows with label length so wrapped text always fits.
|
|
"""
|
|
node = graph.get_node(nid)
|
|
nt = node.node_type if node else NodeType.FUNCTION
|
|
base_w, base_h = NODE_DIMENSIONS.get(nt, (190, 56))
|
|
|
|
if node and node.label:
|
|
max_chars = max(8, int(base_w / _MONO_PX_PER_CHAR))
|
|
n_lines = max(1, math.ceil(len(node.label) / max_chars))
|
|
n_lines = min(n_lines, 4) # hard cap at 4 wrapped lines
|
|
dynamic_h = base_h + max(0, n_lines - 1) * _LINE_HEIGHT_PX
|
|
return base_w, dynamic_h
|
|
|
|
return base_w, base_h
|
|
|
|
|
|
def _layer_total_width(nids: list[str], graph: CodeGraph) -> float:
|
|
"""Total rendered width of a layer (nodes + inter-node gaps)."""
|
|
if not nids:
|
|
return 0
|
|
total = sum(_node_dims(n, graph)[0] for n in nids)
|
|
total += NODE_H_SPACING * max(0, len(nids) - 1)
|
|
return total
|
|
|
|
|
|
def layout_flow(graph: CodeGraph, flow: Flow) -> dict[str, Position]:
|
|
"""
|
|
Hierarchical Sugiyama-style layout with three improvements:
|
|
1. SINK nodes are forced to the last layer so the diagram always reads
|
|
Entry → Processing → Sink top-to-bottom.
|
|
2. Each layer is horizontally centred within the canvas.
|
|
3. Vertical spacing is generous (LAYER_Y_GAP=110) for readability.
|
|
"""
|
|
subgraph_nodes = [nid for nid in flow.node_ids if graph.has_node(nid)]
|
|
if not subgraph_nodes:
|
|
return {}
|
|
|
|
# Build local adjacency from CALLS / DATA_FLOW / TAINTED_FLOW / CONTAINS
|
|
TRAVERSE = {"CALLS", "DATA_FLOW", "TAINTED_FLOW", "CONTAINS"}
|
|
adjacency: dict[str, list[str]] = {nid: [] for nid in subgraph_nodes}
|
|
for u, v, edge in graph.edges():
|
|
if u in adjacency and v in adjacency:
|
|
if edge.edge_type.value in TRAVERSE:
|
|
adjacency[u].append(v)
|
|
|
|
# In-degree for Kahn's topological sort (cycle-safe)
|
|
in_degree: dict[str, int] = {nid: 0 for nid in subgraph_nodes}
|
|
for nid in subgraph_nodes:
|
|
for succ in adjacency[nid]:
|
|
if succ in in_degree:
|
|
in_degree[succ] += 1
|
|
|
|
# Kahn's BFS for topological order
|
|
topo_order: list[str] = []
|
|
temp_degree = dict(in_degree)
|
|
temp_queue: deque[str] = deque(
|
|
nid for nid in subgraph_nodes if temp_degree.get(nid, 0) == 0
|
|
)
|
|
# If all nodes have incoming edges (cycle), start from entry point
|
|
if not temp_queue:
|
|
seed = flow.entry_node_id if flow.entry_node_id in subgraph_nodes else subgraph_nodes[0]
|
|
temp_queue = deque([seed])
|
|
|
|
while temp_queue:
|
|
node = temp_queue.popleft()
|
|
topo_order.append(node)
|
|
for succ in adjacency[node]:
|
|
if succ in temp_degree:
|
|
temp_degree[succ] -= 1
|
|
if temp_degree[succ] == 0:
|
|
temp_queue.append(succ)
|
|
|
|
# Remaining nodes (in cycles): append at the end
|
|
seen = set(topo_order)
|
|
topo_order.extend(nid for nid in subgraph_nodes if nid not in seen)
|
|
|
|
# --- Layer assignment (longest-path / critical-path algorithm) ---
|
|
# Entry node is always layer 0.
|
|
layer_of: dict[str, int] = {}
|
|
for nid in topo_order:
|
|
if nid == flow.entry_node_id:
|
|
layer_of[nid] = 0
|
|
continue
|
|
max_pred = -1
|
|
for pred, succs in adjacency.items():
|
|
if nid in succs and pred in layer_of:
|
|
max_pred = max(max_pred, layer_of[pred])
|
|
layer_of[nid] = max_pred + 1
|
|
|
|
# --- Semantic overrides ---
|
|
# Force SINK nodes to the very last layer so the visual hierarchy is always
|
|
# Entry(top) → Processing(middle) → Sink(bottom).
|
|
if layer_of:
|
|
natural_max = max(layer_of.values())
|
|
sink_layer = natural_max + 1
|
|
for nid in subgraph_nodes:
|
|
node = graph.get_node(nid)
|
|
if node and node.node_type == NodeType.SINK:
|
|
layer_of[nid] = sink_layer
|
|
|
|
# Group nodes by layer
|
|
layers: dict[int, list[str]] = defaultdict(list)
|
|
for nid, lid in layer_of.items():
|
|
layers[lid].append(nid)
|
|
|
|
# --- Barycenter heuristic: order nodes within each layer ---
|
|
sorted_layer_ids = sorted(layers.keys())
|
|
for layer_id in sorted_layer_ids:
|
|
barycenters: dict[str, float] = {}
|
|
for nid in layers[layer_id]:
|
|
pred_positions: list[float] = []
|
|
for pred, succs in adjacency.items():
|
|
if nid in succs and pred in layer_of and layer_of[pred] < layer_id:
|
|
pred_layer = layers[layer_of[pred]]
|
|
if pred in pred_layer:
|
|
pred_positions.append(pred_layer.index(pred))
|
|
barycenters[nid] = (
|
|
sum(pred_positions) / len(pred_positions) if pred_positions else 0.0
|
|
)
|
|
layers[layer_id] = sorted(layers[layer_id], key=lambda x: barycenters.get(x, 0))
|
|
|
|
# --- Centred horizontal placement ---
|
|
# First pass: left-aligned to find the content width (= widest layer)
|
|
positions: dict[str, Position] = {}
|
|
current_y = SVG_PADDING_Y
|
|
layer_y: dict[int, float] = {}
|
|
|
|
for layer_id in sorted_layer_ids:
|
|
layer_y[layer_id] = current_y
|
|
max_h = max((_node_dims(n, graph)[1] for n in layers[layer_id]), default=56)
|
|
current_y += max_h + LAYER_Y_GAP
|
|
|
|
x = SVG_PADDING_X
|
|
for nid in layers[layer_id]:
|
|
w, h = _node_dims(nid, graph)
|
|
positions[nid] = Position(x=x, y=layer_y[layer_id], width=w, height=h)
|
|
x += w + NODE_H_SPACING
|
|
|
|
# Content width = right edge of the widest layer
|
|
if not positions:
|
|
return {}
|
|
content_w = max(p.x + p.width for p in positions.values())
|
|
|
|
# Second pass: shift each layer so it is horizontally centred
|
|
for layer_id in sorted_layer_ids:
|
|
layer_w = _layer_total_width(layers[layer_id], graph)
|
|
shift = (content_w - SVG_PADDING_X - layer_w) / 2
|
|
if shift > 0.5:
|
|
for nid in layers[layer_id]:
|
|
if nid in positions:
|
|
p = positions[nid]
|
|
positions[nid] = Position(
|
|
x=p.x + shift, y=p.y, width=p.width, height=p.height
|
|
)
|
|
|
|
return positions
|
|
|
|
|
|
def layout_overview(graph: CodeGraph, flows: list[Flow]) -> dict[str, Position]:
|
|
"""
|
|
Module-card layout: each MODULE is rendered as a card; its ENTRY_POINT
|
|
children are listed inside the card as rows. Modules are arranged in a
|
|
responsive grid.
|
|
"""
|
|
module_nodes = [
|
|
nid for nid in graph.nodes()
|
|
if (n := graph.get_node(nid)) and n.node_type == NodeType.MODULE
|
|
]
|
|
|
|
entry_nodes = [
|
|
nid for nid in graph.nodes()
|
|
if (n := graph.get_node(nid)) and n.node_type == NodeType.ENTRY_POINT
|
|
]
|
|
|
|
if not module_nodes and not entry_nodes:
|
|
return {}
|
|
|
|
positions: dict[str, Position] = {}
|
|
|
|
# ---- Module-card layout ----
|
|
if module_nodes:
|
|
# Map file_path → module id
|
|
fp_to_module: dict[str, str] = {}
|
|
for mid in module_nodes:
|
|
n = graph.get_node(mid)
|
|
if n and n.file_path:
|
|
fp_to_module[n.file_path] = mid
|
|
|
|
# Group entry points by their parent module
|
|
module_entries: dict[str, list[str]] = defaultdict(list)
|
|
unattached: list[str] = []
|
|
for eid in entry_nodes:
|
|
n = graph.get_node(eid)
|
|
if n:
|
|
mid = fp_to_module.get(n.file_path)
|
|
if mid:
|
|
module_entries[mid].append(eid)
|
|
else:
|
|
unattached.append(eid)
|
|
|
|
CARD_W = 230
|
|
CARD_HEADER_H = 50
|
|
ENTRY_ROW_H = 34
|
|
ENTRY_PAD = 4
|
|
CARD_FOOT_PAD = 10
|
|
CARD_GAP_H = 28 # horizontal gap between columns
|
|
CARD_GAP_V = 20 # vertical gap between rows
|
|
COLS = 3
|
|
|
|
col_x = [SVG_PADDING_X + c * (CARD_W + CARD_GAP_H) for c in range(COLS)]
|
|
col_y = [SVG_PADDING_Y] * COLS
|
|
|
|
for i, mid in enumerate(module_nodes):
|
|
col = i % COLS
|
|
entries = module_entries.get(mid, [])
|
|
rows = max(len(entries), 1)
|
|
card_h = (
|
|
CARD_HEADER_H
|
|
+ rows * (ENTRY_ROW_H + ENTRY_PAD)
|
|
+ CARD_FOOT_PAD
|
|
)
|
|
|
|
x = col_x[col]
|
|
y = col_y[col]
|
|
positions[mid] = Position(x=x, y=y, width=CARD_W, height=card_h)
|
|
|
|
for j, eid in enumerate(entries):
|
|
ey = y + CARD_HEADER_H + j * (ENTRY_ROW_H + ENTRY_PAD)
|
|
positions[eid] = Position(
|
|
x=x + 8, y=ey, width=CARD_W - 16, height=ENTRY_ROW_H
|
|
)
|
|
|
|
col_y[col] += card_h + CARD_GAP_V
|
|
|
|
# Place any unattached entry points in a final row
|
|
if unattached:
|
|
start_y = max(col_y) + 20
|
|
for i, eid in enumerate(unattached):
|
|
col = i % COLS
|
|
positions[eid] = Position(
|
|
x=col_x[col],
|
|
y=start_y,
|
|
width=CARD_W,
|
|
height=ENTRY_ROW_H + 8,
|
|
)
|
|
|
|
else:
|
|
# No modules — just lay out entry points in a grid
|
|
COLS = 3
|
|
W, H = NODE_DIMENSIONS[NodeType.ENTRY_POINT]
|
|
for i, eid in enumerate(entry_nodes):
|
|
col = i % COLS
|
|
row = i // COLS
|
|
positions[eid] = Position(
|
|
x=SVG_PADDING_X + col * (W + 40),
|
|
y=SVG_PADDING_Y + row * (H + 20),
|
|
width=W,
|
|
height=H,
|
|
)
|
|
|
|
return positions
|