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.
483 lines
21 KiB
Python
483 lines
21 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 ────────────────────────────────────────────────────────────
|
|
GO_SINK_PATTERNS: dict[str, dict[str, str]] = {
|
|
"command_execution": {
|
|
"exec.Command": r"exec\.Command\s*\(",
|
|
"exec.CommandContext": r"exec\.CommandContext\s*\(",
|
|
"os.StartProcess": r"os\.StartProcess\s*\(",
|
|
},
|
|
"database": {
|
|
"db.Query": r"\.Query\s*\(",
|
|
"db.QueryRow": r"\.QueryRow\s*\(",
|
|
"db.Exec": r"\.Exec\s*\(",
|
|
"db.QueryContext": r"\.QueryContext\s*\(",
|
|
"db.ExecContext": r"\.ExecContext\s*\(",
|
|
"db.QueryRowContext":r"\.QueryRowContext\s*\(",
|
|
},
|
|
"file_write": {
|
|
"os.WriteFile": r"os\.WriteFile\s*\(",
|
|
"ioutil.WriteFile": r"ioutil\.WriteFile\s*\(",
|
|
"os.Create": r"os\.Create\s*\(",
|
|
"os.OpenFile": r"os\.OpenFile\s*\(",
|
|
},
|
|
"file_read": {
|
|
"os.ReadFile": r"os\.ReadFile\s*\(",
|
|
"ioutil.ReadFile": r"ioutil\.ReadFile\s*\(",
|
|
"os.Open": r"os\.Open\s*\(",
|
|
},
|
|
"http_request": {
|
|
"http.Get": r"http\.Get\s*\(",
|
|
"http.Post": r"http\.Post\s*\(",
|
|
"http.Do": r"\.Do\s*\(",
|
|
},
|
|
"deserialization": {
|
|
"json.Unmarshal": r"json\.Unmarshal\s*\(",
|
|
"json.NewDecoder": r"json\.NewDecoder\s*\(",
|
|
"xml.Unmarshal": r"xml\.Unmarshal\s*\(",
|
|
"gob.Decode": r"\.Decode\s*\(",
|
|
"yaml.Unmarshal": r"yaml\.Unmarshal\s*\(",
|
|
},
|
|
"code_execution": {
|
|
"plugin.Open": r"plugin\.Open\s*\(",
|
|
},
|
|
"template_injection": {
|
|
"template.Execute": r"\.Execute\s*\(",
|
|
"template.ExecuteTemplate": r"\.ExecuteTemplate\s*\(",
|
|
},
|
|
}
|
|
|
|
_PRIORITY_ORDER = [
|
|
"code_execution", "command_execution", "database",
|
|
"template_injection", "deserialization",
|
|
"file_write", "file_read", "http_request",
|
|
]
|
|
|
|
|
|
def _detect_go_framework(source_text: str) -> str:
|
|
if "github.com/gin-gonic/gin" in source_text:
|
|
return "gin"
|
|
if "github.com/labstack/echo" in source_text:
|
|
return "echo"
|
|
if "github.com/go-chi/chi" in source_text:
|
|
return "chi"
|
|
if "github.com/gorilla/mux" in source_text:
|
|
return "gorilla_mux"
|
|
if "net/http" in source_text and (
|
|
"http.HandleFunc" in source_text
|
|
or "http.Handle" in source_text
|
|
or "http.ListenAndServe" in source_text
|
|
):
|
|
return "net_http"
|
|
return "unknown"
|
|
|
|
|
|
class GoExtractor:
|
|
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, "go")
|
|
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_go_framework(source_text)
|
|
|
|
self._extract_module_node(result)
|
|
self._extract_functions(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
|
|
|
|
# ── 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": "go", "framework": self.framework, "imports": []},
|
|
))
|
|
|
|
# ── Functions & methods ───────────────────────────────────────────────────
|
|
def _extract_functions(self, result: ExtractionResult, tree):
|
|
module_id = make_module_node_id(self.file_stem)
|
|
|
|
# Regular top-level functions: func Foo(...)
|
|
func_query = """
|
|
(function_declaration
|
|
name: (identifier) @func_name) @func_decl
|
|
"""
|
|
for func_node, _ in query_tree(tree, "go", func_query, capture_filter="func_decl"):
|
|
name_child = func_node.child_by_field_name("name")
|
|
if not name_child:
|
|
continue
|
|
func_name = get_node_text(name_child, self.source_bytes)
|
|
if not func_name:
|
|
continue
|
|
node_id = make_node_id(self.file_stem, func_name)
|
|
result.nodes.append(Node(
|
|
id=node_id,
|
|
node_type=NodeType.FUNCTION,
|
|
label=func_name[:40],
|
|
file_path=self.file_path,
|
|
line_start=func_node.start_point[0] + 1,
|
|
line_end=func_node.end_point[0] + 1,
|
|
metadata={"language": "go"},
|
|
))
|
|
result.edges.append(Edge(
|
|
source_id=module_id, target_id=node_id, edge_type=EdgeType.CONTAINS
|
|
))
|
|
|
|
# Methods with receivers: func (h *Handler) Foo(...)
|
|
method_query = """
|
|
(method_declaration
|
|
name: (field_identifier) @method_name) @method_decl
|
|
"""
|
|
for method_node, _ in query_tree(tree, "go", method_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
|
|
|
|
# Extract receiver type for qualified name: (h *Handler) → Handler
|
|
receiver_type = ""
|
|
receiver = method_node.child_by_field_name("receiver")
|
|
if receiver:
|
|
rec_text = get_node_text(receiver, self.source_bytes)
|
|
rec_m = re.search(r"\*?(\w+)", rec_text)
|
|
if rec_m:
|
|
receiver_type = rec_m.group(1)
|
|
|
|
qualified = f"{receiver_type}.{method_name}" if receiver_type else method_name
|
|
node_id = make_node_id(self.file_stem, qualified)
|
|
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,
|
|
"receiver_type": receiver_type,
|
|
"language": "go",
|
|
},
|
|
))
|
|
result.edges.append(Edge(
|
|
source_id=module_id, target_id=node_id, edge_type=EdgeType.CONTAINS
|
|
))
|
|
|
|
# ── Entry points ──────────────────────────────────────────────────────────
|
|
def _extract_entry_points(self, result: ExtractionResult, source_text: str):
|
|
# Gin / Echo / Chi / gorilla-mux style: r.GET("/path", handlerFunc)
|
|
# Also matches: router.Get("/path", handler) for Chi (lowercase method)
|
|
route_re = re.compile(
|
|
r"\.(?P<method>GET|POST|PUT|DELETE|PATCH|OPTIONS|HEAD|"
|
|
r"Get|Post|Put|Delete|Patch|Options|Head)\s*\("
|
|
r'\s*"(?P<path>[^"]+)"\s*,\s*(?P<handler>\w+)',
|
|
re.MULTILINE,
|
|
)
|
|
for m in route_re.finditer(source_text):
|
|
http_method = m.group("method").upper()
|
|
route_path = m.group("path")
|
|
handler_name = m.group("handler")
|
|
self._promote_to_entry_point(result, handler_name, http_method, route_path)
|
|
|
|
# net/http: http.HandleFunc("/path", handlerFunc)
|
|
hf_re = re.compile(
|
|
r'http\.HandleFunc\s*\(\s*"(?P<path>[^"]+)"\s*,\s*(?P<handler>\w+)',
|
|
re.MULTILINE,
|
|
)
|
|
for m in hf_re.finditer(source_text):
|
|
route_path = m.group("path")
|
|
handler_name = m.group("handler")
|
|
self._promote_to_entry_point(result, handler_name, "HTTP", route_path)
|
|
|
|
# gorilla/mux: r.HandleFunc("/path", handlerFunc).Methods("GET")
|
|
gmux_re = re.compile(
|
|
r'\.HandleFunc\s*\(\s*"(?P<path>[^"]+)"\s*,\s*(?P<handler>\w+)',
|
|
re.MULTILINE,
|
|
)
|
|
for m in gmux_re.finditer(source_text):
|
|
route_path = m.group("path")
|
|
handler_name = m.group("handler")
|
|
self._promote_to_entry_point(result, handler_name, "HTTP", route_path)
|
|
|
|
def _promote_to_entry_point(
|
|
self, result: ExtractionResult, handler_name: str, http_method: str, route_path: str
|
|
):
|
|
func_id = make_node_id(self.file_stem, handler_name)
|
|
for n in result.nodes:
|
|
if n.id == func_id and n.node_type == NodeType.FUNCTION:
|
|
n.node_type = NodeType.ENTRY_POINT
|
|
n.is_entry_point = True
|
|
n.label = f"{http_method} {route_path}"
|
|
n.metadata.update({
|
|
"http_method": http_method,
|
|
"route_path": route_path,
|
|
"framework": self.framework,
|
|
})
|
|
return
|
|
# Also try by label (for methods with receivers)
|
|
for n in result.nodes:
|
|
if n.node_type == NodeType.FUNCTION and n.label == handler_name[:40]:
|
|
n.node_type = NodeType.ENTRY_POINT
|
|
n.is_entry_point = True
|
|
n.label = f"{http_method} {route_path}"
|
|
n.metadata.update({
|
|
"http_method": http_method,
|
|
"route_path": route_path,
|
|
"framework": self.framework,
|
|
})
|
|
return
|
|
|
|
# ── Sources ───────────────────────────────────────────────────────────────
|
|
def _extract_sources(self, result: ExtractionResult, source_text: str):
|
|
seen: set[str] = set()
|
|
|
|
all_patterns: list[tuple[str, str]] = [
|
|
# Gin
|
|
(r'[cr]\.Param\s*\(\s*"(?P<p>[^"]+)"\s*\)', "url_param"),
|
|
(r'[cr]\.Query\s*\(\s*"(?P<p>[^"]+)"\s*\)', "query_string"),
|
|
(r'[cr]\.PostForm\s*\(\s*"(?P<p>[^"]+)"\s*\)', "form_field"),
|
|
(r'[cr]\.GetHeader\s*\(\s*"(?P<p>[^"]+)"\s*\)', "header"),
|
|
(r'[cr]\.Cookie\s*\(\s*"(?P<p>[^"]+)"\s*\)', "cookie"),
|
|
(r'[cr]\.BindJSON\s*\(', "json_body"),
|
|
(r'[cr]\.ShouldBindJSON\s*\(', "json_body"),
|
|
(r'[cr]\.ShouldBind\s*\(', "json_body"),
|
|
# Echo
|
|
(r'c\.QueryParam\s*\(\s*"(?P<p>[^"]+)"\s*\)', "query_string"),
|
|
(r'c\.FormValue\s*\(\s*"(?P<p>[^"]+)"\s*\)', "form_field"),
|
|
# net/http
|
|
(r'r\.URL\.Query\(\)\.Get\s*\(\s*"(?P<p>[^"]+)"\s*\)', "query_string"),
|
|
(r'r\.FormValue\s*\(\s*"(?P<p>[^"]+)"\s*\)', "form_field"),
|
|
(r'r\.Header\.Get\s*\(\s*"(?P<p>[^"]+)"\s*\)', "header"),
|
|
(r'r\.Cookie\s*\(\s*"(?P<p>[^"]+)"\s*\)', "cookie"),
|
|
(r'r\.PathValue\s*\(\s*"(?P<p>[^"]+)"\s*\)', "url_param"), # Go 1.22+
|
|
(r'\br\.Body\b', "raw_body"),
|
|
# CLI
|
|
(r'\bos\.Args\b', "cli_arg"),
|
|
(r'flag\.\w+\s*\(\s*"(?P<p>[^"]+)"', "cli_arg"),
|
|
]
|
|
|
|
for pattern, input_type in all_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
|
|
containing_func = self._find_containing_function(line_num, result)
|
|
if not containing_func:
|
|
continue
|
|
func_part = containing_func.split("::")[-1]
|
|
source_id = make_node_id(self.file_stem, f"{func_part}::{param_name}")
|
|
if source_id in seen:
|
|
continue
|
|
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 GO_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):
|
|
for env_re in (
|
|
re.compile(r'os\.Getenv\s*\(\s*"(?P<var>\w+)"\s*\)'),
|
|
re.compile(r'os\.LookupEnv\s*\(\s*"(?P<var>\w+)"\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 ───────────────────────────────────────────────────────────────
|
|
def _extract_imports(self, result: ExtractionResult, source_text: str):
|
|
module_id = make_module_node_id(self.file_stem)
|
|
# Match both single-line `import "pkg"` and grouped imports
|
|
import_block_re = re.compile(
|
|
r'\bimport\s+(?:"[^"]+"|\((?:[^)]*)\))', re.DOTALL
|
|
)
|
|
pkg_re = re.compile(r'"(?P<pkg>[^"]+)"')
|
|
imports_list: list[str] = []
|
|
for block in import_block_re.finditer(source_text):
|
|
for pm in pkg_re.finditer(block.group(0)):
|
|
imports_list.append(pm.group("pkg"))
|
|
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):
|
|
# Simple call_expression where the function is a plain identifier
|
|
query = """
|
|
(call_expression
|
|
function: (identifier) @func_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, "go", query, capture_filter="call"):
|
|
func_child = call_node.child_by_field_name("function")
|
|
if not func_child:
|
|
continue
|
|
callee = get_node_text(func_child, self.source_bytes)
|
|
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 GoExtractor().extract(file_info)
|