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.
507 lines
22 KiB
Python
507 lines
22 KiB
Python
from __future__ import annotations
|
|
import re
|
|
from pathlib import Path
|
|
from codeflow.discovery.file_scanner import FileInfo
|
|
from codeflow.analysis.base_extractor import ExtractionResult
|
|
from codeflow.analysis.ast_parser import parse_file, query_tree, get_node_text
|
|
from codeflow.models.node import Node, NodeType, make_node_id, make_module_node_id
|
|
from codeflow.models.edge import Edge, EdgeType
|
|
|
|
|
|
# ── Sink catalogue ────────────────────────────────────────────────────────────
|
|
CSHARP_SINK_PATTERNS: dict[str, dict[str, str]] = {
|
|
"command_execution": {
|
|
"Process.Start": r"Process\.Start\s*\(",
|
|
"process.Start": r"\.Start\s*\(\s*\)", # instance process.Start()
|
|
},
|
|
"database": {
|
|
"ExecuteNonQuery": r"\.ExecuteNonQuery(?:Async)?\s*\(",
|
|
"ExecuteReader": r"\.ExecuteReader(?:Async)?\s*\(",
|
|
"ExecuteScalar": r"\.ExecuteScalar(?:Async)?\s*\(",
|
|
"FromSqlRaw": r"\.FromSqlRaw\s*\(",
|
|
"FromSqlInterpolated": r"\.FromSqlInterpolated\s*\(",
|
|
"SqliteCommand.Execute": r"new\s+SqliteCommand\s*\(",
|
|
},
|
|
"deserialization": {
|
|
"JsonConvert.DeserializeObject": r"JsonConvert\.DeserializeObject\s*[<(]",
|
|
"JsonSerializer.Deserialize": r"JsonSerializer\.Deserialize\s*[<(]",
|
|
"XmlSerializer.Deserialize": r"\.Deserialize\s*\(",
|
|
"BinaryFormatter.Deserialize": r"\.Deserialize\s*\(",
|
|
},
|
|
"file_write": {
|
|
"File.WriteAllText": r"File\.WriteAllText(?:Async)?\s*\(",
|
|
"File.WriteAllBytes": r"File\.WriteAllBytes(?:Async)?\s*\(",
|
|
"File.AppendAllText": r"File\.AppendAllText(?:Async)?\s*\(",
|
|
"StreamWriter": r"new\s+StreamWriter\s*\(",
|
|
},
|
|
"file_read": {
|
|
"File.ReadAllText": r"File\.ReadAllText(?:Async)?\s*\(",
|
|
"File.ReadAllBytes": r"File\.ReadAllBytes(?:Async)?\s*\(",
|
|
"File.OpenRead": r"File\.OpenRead\s*\(",
|
|
"StreamReader": r"new\s+StreamReader\s*\(",
|
|
},
|
|
"code_execution": {
|
|
"Activator.CreateInstance": r"Activator\.CreateInstance\s*[<(]",
|
|
"Assembly.Load": r"Assembly\.Load(?:From|File)?\s*\(",
|
|
"Type.GetType": r"Type\.GetType\s*\(",
|
|
"CSharpScript.Evaluate": r"CSharpScript\.Evaluate(?:Async)?\s*\(",
|
|
},
|
|
"http_request": {
|
|
"HttpClient.GetAsync": r"\.GetAsync\s*\(",
|
|
"HttpClient.PostAsync": r"\.PostAsync\s*\(",
|
|
"HttpClient.SendAsync": r"\.SendAsync\s*\(",
|
|
"WebClient.DownloadString": r"\.DownloadString(?:TaskAsync)?\s*\(",
|
|
},
|
|
"template_injection": {
|
|
"RazorEngine.Run": r"Engine\.Razor\.Run\s*\(",
|
|
},
|
|
}
|
|
|
|
_PRIORITY_ORDER = [
|
|
"code_execution", "command_execution", "database",
|
|
"template_injection", "deserialization",
|
|
"file_write", "file_read", "http_request",
|
|
]
|
|
|
|
# ASP.NET Core / MVC HTTP-verb attributes
|
|
_HTTP_ATTRIBUTES: dict[str, str] = {
|
|
"HttpGet": "GET",
|
|
"HttpPost": "POST",
|
|
"HttpPut": "PUT",
|
|
"HttpDelete": "DELETE",
|
|
"HttpPatch": "PATCH",
|
|
"HttpHead": "HEAD",
|
|
"HttpOptions": "OPTIONS",
|
|
"Route": "ROUTE",
|
|
}
|
|
|
|
# Parameter-binding attributes → input_type
|
|
_PARAM_ATTRS: dict[str, str] = {
|
|
"FromQuery": "query_string",
|
|
"FromBody": "json_body",
|
|
"FromRoute": "url_param",
|
|
"FromHeader": "header",
|
|
"FromForm": "form_field",
|
|
}
|
|
|
|
|
|
def _detect_csharp_framework(source_text: str) -> str:
|
|
if "Microsoft.AspNetCore" in source_text or "[ApiController]" in source_text:
|
|
return "aspnet_core"
|
|
if "System.Web.Mvc" in source_text:
|
|
return "aspnet_mvc"
|
|
if "System.Web.Http" in source_text:
|
|
return "webapi"
|
|
return "unknown"
|
|
|
|
|
|
class CSharpExtractor:
|
|
def __init__(self):
|
|
self.file_stem: str = ""
|
|
self.file_path: str = ""
|
|
self.framework: str = "unknown"
|
|
self.source_bytes: bytes = b""
|
|
self.source_lines: list[str] = []
|
|
|
|
# ── Public entry ──────────────────────────────────────────────────────────
|
|
def extract(self, file_info: FileInfo) -> ExtractionResult:
|
|
result = ExtractionResult(file_path=file_info.path)
|
|
self.file_path = file_info.path
|
|
self.file_stem = Path(file_info.path).stem
|
|
|
|
try:
|
|
self.source_bytes = Path(file_info.path).read_bytes()
|
|
self.source_lines = (
|
|
Path(file_info.path)
|
|
.read_text(encoding="utf-8", errors="replace")
|
|
.splitlines()
|
|
)
|
|
except (OSError, IOError):
|
|
result.errors.append(f"Cannot read file: {file_info.path}")
|
|
return result
|
|
|
|
tree = parse_file(file_info.path, "c_sharp")
|
|
if tree is None:
|
|
result.errors.append(f"Could not parse: {file_info.path}")
|
|
return result
|
|
|
|
source_text = self.source_bytes.decode("utf-8", errors="replace")
|
|
self.framework = _detect_csharp_framework(source_text)
|
|
|
|
self._extract_module_node(result)
|
|
self._extract_classes(result, tree)
|
|
self._extract_methods(result, tree)
|
|
self._extract_entry_points(result, source_text)
|
|
self._extract_sources(result, source_text)
|
|
self._extract_sinks(result, source_text)
|
|
self._extract_env_vars(result, source_text)
|
|
self._extract_imports(result, source_text)
|
|
self._extract_calls(result, tree)
|
|
|
|
self._ensure_unique_ids(result)
|
|
return result
|
|
|
|
# ── Helpers ───────────────────────────────────────────────────────────────
|
|
def _ensure_unique_ids(self, result: ExtractionResult):
|
|
seen: set[str] = set()
|
|
deduped: list[Node] = []
|
|
for node in result.nodes:
|
|
if node.id not in seen:
|
|
seen.add(node.id)
|
|
deduped.append(node)
|
|
result.nodes = deduped
|
|
|
|
def _find_containing_function(self, line_num: int, result: ExtractionResult) -> str | None:
|
|
for n in result.nodes:
|
|
if n.node_type in (NodeType.FUNCTION, NodeType.ENTRY_POINT):
|
|
if n.line_start and n.line_end and n.line_start <= line_num <= n.line_end:
|
|
return n.id
|
|
return None
|
|
|
|
def _get_containing_class(self, ts_node) -> str:
|
|
parent = ts_node.parent
|
|
while parent:
|
|
if parent.type == "class_declaration":
|
|
name_child = parent.child_by_field_name("name")
|
|
if name_child:
|
|
return get_node_text(name_child, self.source_bytes)
|
|
parent = parent.parent
|
|
return ""
|
|
|
|
# ── Module node ───────────────────────────────────────────────────────────
|
|
def _extract_module_node(self, result: ExtractionResult):
|
|
node_id = make_module_node_id(self.file_stem)
|
|
result.nodes.append(Node(
|
|
id=node_id,
|
|
node_type=NodeType.MODULE,
|
|
label=Path(self.file_path).name,
|
|
file_path=self.file_path,
|
|
line_start=1,
|
|
line_end=len(self.source_lines),
|
|
metadata={"language": "c_sharp", "framework": self.framework, "imports": []},
|
|
))
|
|
|
|
# ── Classes ───────────────────────────────────────────────────────────────
|
|
def _extract_classes(self, result: ExtractionResult, tree):
|
|
query = """
|
|
(class_declaration
|
|
name: (identifier) @class_name) @class_decl
|
|
"""
|
|
for cls_node, _ in query_tree(tree, "c_sharp", query, capture_filter="class_decl"):
|
|
name_child = cls_node.child_by_field_name("name")
|
|
if not name_child:
|
|
continue
|
|
cls_name = get_node_text(name_child, self.source_bytes)
|
|
if not cls_name:
|
|
continue
|
|
node_id = make_node_id(self.file_stem, cls_name)
|
|
result.nodes.append(Node(
|
|
id=node_id,
|
|
node_type=NodeType.CLASS,
|
|
label=cls_name,
|
|
file_path=self.file_path,
|
|
line_start=cls_node.start_point[0] + 1,
|
|
line_end=cls_node.end_point[0] + 1,
|
|
metadata={"language": "c_sharp"},
|
|
))
|
|
result.edges.append(Edge(
|
|
source_id=make_module_node_id(self.file_stem),
|
|
target_id=node_id,
|
|
edge_type=EdgeType.CONTAINS,
|
|
))
|
|
|
|
# ── Methods ───────────────────────────────────────────────────────────────
|
|
def _extract_methods(self, result: ExtractionResult, tree):
|
|
query = """
|
|
(method_declaration
|
|
name: (identifier) @method_name) @method_decl
|
|
"""
|
|
for method_node, _ in query_tree(tree, "c_sharp", query, capture_filter="method_decl"):
|
|
name_child = method_node.child_by_field_name("name")
|
|
if not name_child:
|
|
continue
|
|
method_name = get_node_text(name_child, self.source_bytes)
|
|
if not method_name:
|
|
continue
|
|
|
|
class_name = self._get_containing_class(method_node)
|
|
qualified = f"{class_name}.{method_name}" if class_name else method_name
|
|
node_id = make_node_id(self.file_stem, qualified)
|
|
|
|
# Collect parameter names
|
|
params: list[str] = []
|
|
params_node = method_node.child_by_field_name("parameters")
|
|
if params_node:
|
|
for child in params_node.children:
|
|
if child.type == "parameter":
|
|
pname_node = child.child_by_field_name("name")
|
|
if pname_node:
|
|
pname = get_node_text(pname_node, self.source_bytes)
|
|
if pname:
|
|
params.append(pname)
|
|
|
|
result.nodes.append(Node(
|
|
id=node_id,
|
|
node_type=NodeType.FUNCTION,
|
|
label=method_name[:40],
|
|
file_path=self.file_path,
|
|
line_start=method_node.start_point[0] + 1,
|
|
line_end=method_node.end_point[0] + 1,
|
|
metadata={
|
|
"qualified_name": qualified,
|
|
"params": params,
|
|
"class_name": class_name,
|
|
"language": "c_sharp",
|
|
},
|
|
))
|
|
result.edges.append(Edge(
|
|
source_id=make_module_node_id(self.file_stem),
|
|
target_id=node_id,
|
|
edge_type=EdgeType.CONTAINS,
|
|
))
|
|
|
|
# ── Entry points (ASP.NET) ────────────────────────────────────────────────
|
|
def _extract_entry_points(self, result: ExtractionResult, source_text: str):
|
|
if self.framework not in ("aspnet_core", "aspnet_mvc", "webapi"):
|
|
return
|
|
|
|
# Match [HttpGet], [HttpGet("/route")], [HttpGet("route")]
|
|
# Also [Route("path")] without a specific verb
|
|
attr_re = re.compile(
|
|
r"\[(?P<attr>" + "|".join(_HTTP_ATTRIBUTES.keys()) + r")"
|
|
r'(?:\s*\(\s*"?(?P<path>[^"\)\s]*)"?\s*\))?\]',
|
|
re.MULTILINE,
|
|
)
|
|
|
|
for m in attr_re.finditer(source_text):
|
|
attr_name = m.group("attr")
|
|
http_method = _HTTP_ATTRIBUTES[attr_name]
|
|
route_path = m.group("path") or "/"
|
|
attr_line = source_text[: m.start()].count("\n") + 1
|
|
|
|
# Find method node whose line_start is within the next few lines
|
|
# (ASP.NET attributes sit directly above the method signature)
|
|
best: Node | None = None
|
|
best_dist = 9999
|
|
for n in result.nodes:
|
|
if n.node_type != NodeType.FUNCTION:
|
|
continue
|
|
dist = n.line_start - attr_line
|
|
if 0 <= dist <= 8 and dist < best_dist:
|
|
best_dist = dist
|
|
best = n
|
|
|
|
if best is None:
|
|
continue
|
|
|
|
best.node_type = NodeType.ENTRY_POINT
|
|
best.is_entry_point = True
|
|
best.label = f"{http_method} {route_path}"
|
|
best.metadata["http_method"] = http_method
|
|
best.metadata["route_path"] = route_path
|
|
best.metadata["framework"] = self.framework
|
|
|
|
# ── Sources ───────────────────────────────────────────────────────────────
|
|
def _extract_sources(self, result: ExtractionResult, source_text: str):
|
|
seen: set[str] = set()
|
|
|
|
# [FromQuery] / [FromBody] / [FromRoute] / [FromHeader] / [FromForm] on parameters
|
|
# Handles: [FromQuery] string name AND [FromQuery(Name="foo")] string bar
|
|
param_re = re.compile(
|
|
r"\[(?P<attr>" + "|".join(_PARAM_ATTRS.keys()) + r")"
|
|
r"(?:\([^)]*\))?\]\s+"
|
|
r"(?:(?:[\w.<>\[\]]+)\s+)+" # type
|
|
r"(?P<pname>\w+)",
|
|
re.MULTILINE,
|
|
)
|
|
for m in param_re.finditer(source_text):
|
|
attr = m.group("attr")
|
|
input_type = _PARAM_ATTRS[attr]
|
|
param_name = m.group("pname")
|
|
line_num = source_text[: m.start()].count("\n") + 1
|
|
self._add_source(result, seen, param_name, input_type, line_num)
|
|
|
|
# Request.Query["name"], Request.Form["name"], etc.
|
|
request_patterns: list[tuple[str, str]] = [
|
|
(r'Request\.Query\s*\[\s*"(?P<p>[^"]+)"\s*\]', "query_string"),
|
|
(r'Request\.Form\s*\[\s*"(?P<p>[^"]+)"\s*\]', "form_field"),
|
|
(r'Request\.Headers\s*\[\s*"(?P<p>[^"]+)"\s*\]', "header"),
|
|
(r'Request\.Cookies\s*\[\s*"(?P<p>[^"]+)"\s*\]', "cookie"),
|
|
(r'Request\.RouteValues\s*\[\s*"(?P<p>[^"]+)"\s*\]', "url_param"),
|
|
(r'\bRequest\.Body\b', "raw_body"),
|
|
# HttpContext.Request convenience
|
|
(r'HttpContext\.Request\.Query\s*\[\s*"(?P<p>[^"]+)"\s*\]', "query_string"),
|
|
]
|
|
for pattern, input_type in request_patterns:
|
|
for m in re.finditer(pattern, source_text):
|
|
gd = m.groupdict()
|
|
param_name = gd.get("p") or f"*{input_type}*"
|
|
line_num = source_text[: m.start()].count("\n") + 1
|
|
self._add_source(result, seen, param_name, input_type, line_num)
|
|
|
|
def _add_source(
|
|
self,
|
|
result: ExtractionResult,
|
|
seen: set[str],
|
|
param_name: str,
|
|
input_type: str,
|
|
line_num: int,
|
|
):
|
|
containing_func = self._find_containing_function(line_num, result)
|
|
if not containing_func:
|
|
return
|
|
func_part = containing_func.split("::")[-1]
|
|
source_id = make_node_id(self.file_stem, f"{func_part}::{param_name}")
|
|
if source_id in seen:
|
|
return
|
|
seen.add(source_id)
|
|
result.nodes.append(Node(
|
|
id=source_id,
|
|
node_type=NodeType.SOURCE,
|
|
label=f"param: {param_name} [{input_type}]"[:40],
|
|
file_path=self.file_path,
|
|
line_start=line_num,
|
|
line_end=line_num,
|
|
metadata={
|
|
"param_name": param_name,
|
|
"input_type": input_type,
|
|
"framework": self.framework,
|
|
},
|
|
))
|
|
result.edges.append(Edge(
|
|
source_id=source_id,
|
|
target_id=containing_func,
|
|
edge_type=EdgeType.DATA_FLOW,
|
|
))
|
|
|
|
# ── Sinks ─────────────────────────────────────────────────────────────────
|
|
def _extract_sinks(self, result: ExtractionResult, source_text: str):
|
|
seen: set[str] = set()
|
|
flat: list[tuple[str, str, str]] = []
|
|
for sink_type, funcs in CSHARP_SINK_PATTERNS.items():
|
|
for func_name, pattern in funcs.items():
|
|
flat.append((pattern, sink_type, func_name))
|
|
flat.sort(key=lambda x: _PRIORITY_ORDER.index(x[1]) if x[1] in _PRIORITY_ORDER else 99)
|
|
|
|
for pattern, sink_type, sink_func in flat:
|
|
for m in re.finditer(pattern, source_text):
|
|
line_num = source_text[: m.start()].count("\n") + 1
|
|
containing_func = self._find_containing_function(line_num, result)
|
|
if not containing_func:
|
|
continue
|
|
sink_id = make_node_id(
|
|
self.file_stem, f"SINK::{sink_type}::{sink_func}::{line_num}"
|
|
)
|
|
if sink_id in seen:
|
|
continue
|
|
seen.add(sink_id)
|
|
result.nodes.append(Node(
|
|
id=sink_id,
|
|
node_type=NodeType.SINK,
|
|
label=f"sink: {sink_func} [{sink_type}]",
|
|
file_path=self.file_path,
|
|
line_start=line_num,
|
|
line_end=line_num,
|
|
metadata={"sink_type": sink_type, "function_name": sink_func},
|
|
))
|
|
result.edges.append(Edge(
|
|
source_id=containing_func,
|
|
target_id=sink_id,
|
|
edge_type=EdgeType.CALLS,
|
|
))
|
|
|
|
# ── Env vars ──────────────────────────────────────────────────────────────
|
|
def _extract_env_vars(self, result: ExtractionResult, source_text: str):
|
|
env_re = re.compile(
|
|
r'Environment\.GetEnvironmentVariable\s*\(\s*"(?P<var>[^"]+)"\s*\)'
|
|
)
|
|
for m in env_re.finditer(source_text):
|
|
var_name = m.group("var")
|
|
line_num = source_text[: m.start()].count("\n") + 1
|
|
env_id = make_node_id(self.file_stem, f"ENV::{var_name}")
|
|
if any(n.id == env_id for n in result.nodes):
|
|
continue
|
|
result.nodes.append(Node(
|
|
id=env_id,
|
|
node_type=NodeType.ENV_VAR,
|
|
label=f"env: {var_name}",
|
|
file_path=self.file_path,
|
|
line_start=line_num,
|
|
line_end=line_num,
|
|
metadata={"var_name": var_name},
|
|
))
|
|
containing_func = self._find_containing_function(line_num, result)
|
|
if containing_func:
|
|
result.edges.append(Edge(
|
|
source_id=env_id,
|
|
target_id=containing_func,
|
|
edge_type=EdgeType.DATA_FLOW,
|
|
))
|
|
|
|
# ── Imports (using directives) ─────────────────────────────────────────────
|
|
def _extract_imports(self, result: ExtractionResult, source_text: str):
|
|
module_id = make_module_node_id(self.file_stem)
|
|
using_re = re.compile(
|
|
r"^using\s+(?:static\s+)?(?P<ns>[\w.]+)\s*;", re.MULTILINE
|
|
)
|
|
imports_list = [m.group("ns") for m in using_re.finditer(source_text)]
|
|
for n in result.nodes:
|
|
if n.id == module_id:
|
|
n.metadata["imports"] = imports_list
|
|
break
|
|
|
|
# ── Call graph ────────────────────────────────────────────────────────────
|
|
def _extract_calls(self, result: ExtractionResult, tree):
|
|
# C# invocation_expression: obj.Method(...)
|
|
query = """
|
|
(invocation_expression
|
|
function: (member_access_expression
|
|
name: (identifier) @method_name)) @call
|
|
"""
|
|
func_map: dict[str, str] = {}
|
|
for n in result.nodes:
|
|
if n.node_type in (NodeType.FUNCTION, NodeType.ENTRY_POINT):
|
|
func_map[n.label] = n.id
|
|
qn = n.metadata.get("qualified_name", "")
|
|
if qn:
|
|
func_map[qn] = n.id
|
|
|
|
unresolved: list[dict] = []
|
|
for call_node, _ in query_tree(tree, "c_sharp", query, capture_filter="call"):
|
|
func_child = call_node.child_by_field_name("function")
|
|
if not func_child:
|
|
continue
|
|
# The last identifier child of member_access_expression is the method name
|
|
callee = ""
|
|
for child in reversed(func_child.children):
|
|
if child.type == "identifier":
|
|
callee = get_node_text(child, self.source_bytes)
|
|
break
|
|
if not callee:
|
|
continue
|
|
|
|
line_num = call_node.start_point[0] + 1
|
|
caller_id = self._find_containing_function(line_num, result)
|
|
if not caller_id:
|
|
continue
|
|
|
|
resolved = False
|
|
for fname, fid in func_map.items():
|
|
if fname == callee or fid.endswith(f"::{callee}"):
|
|
result.edges.append(Edge(
|
|
source_id=caller_id,
|
|
target_id=fid,
|
|
edge_type=EdgeType.CALLS,
|
|
))
|
|
resolved = True
|
|
break
|
|
|
|
if not resolved:
|
|
unresolved.append({"caller_id": caller_id, "callee_name": callee, "line": line_num})
|
|
|
|
result.metadata["unresolved_calls"] = unresolved
|
|
|
|
|
|
def extract(file_info: FileInfo) -> ExtractionResult:
|
|
return CSharpExtractor().extract(file_info)
|