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.
257 lines
9.3 KiB
Python
257 lines
9.3 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
|
|
|
|
|
|
FUNCTION_QUERIES: dict[str, str] = {
|
|
"java": """
|
|
(method_declaration
|
|
name: (identifier) @func_name) @method_decl
|
|
""",
|
|
"go": """
|
|
(function_declaration
|
|
name: (identifier) @func_name) @func_decl
|
|
""",
|
|
"ruby": """
|
|
(method
|
|
name: (identifier) @func_name) @method_def
|
|
""",
|
|
"rust": """
|
|
(function_item
|
|
name: (identifier) @func_name) @func_item
|
|
""",
|
|
"c": """
|
|
(function_definition
|
|
declarator: (function_declarator
|
|
declarator: (identifier) @func_name)) @func_def
|
|
""",
|
|
"cpp": """
|
|
(function_definition
|
|
declarator: (function_declarator
|
|
declarator: (identifier) @func_name)) @func_def
|
|
""",
|
|
"php": """
|
|
(function_definition
|
|
name: (name) @func_name) @func_def
|
|
""",
|
|
"c_sharp": """
|
|
(method_declaration
|
|
name: (identifier) @func_name) @method_decl
|
|
""",
|
|
"kotlin": """
|
|
(function_declaration
|
|
name: (simple_identifier) @func_name) @func_decl
|
|
""",
|
|
"swift": """
|
|
(function_declaration
|
|
name: (simple_identifier) @func_name) @func_decl
|
|
""",
|
|
"scala": """
|
|
(function_definition
|
|
name: (identifier) @func_name) @func_def
|
|
""",
|
|
"bash": """
|
|
(function_definition
|
|
name: (word) @func_name) @func_def
|
|
""",
|
|
}
|
|
|
|
CLASS_QUERIES: dict[str, str] = {
|
|
"java": """
|
|
(class_declaration
|
|
name: (identifier) @class_name) @class_decl
|
|
""",
|
|
"go": """
|
|
(type_declaration
|
|
(type_spec
|
|
name: (type_identifier) @class_name)) @type_decl
|
|
""",
|
|
"ruby": """
|
|
(class
|
|
name: (constant) @class_name) @class_def
|
|
""",
|
|
"rust": """
|
|
(struct_item
|
|
name: (type_identifier) @class_name) @struct_item
|
|
|
|
(impl_item
|
|
trait: (type_identifier) @class_name) @impl_item
|
|
""",
|
|
"c_sharp": """
|
|
(class_declaration
|
|
name: (identifier) @class_name) @class_decl
|
|
""",
|
|
"kotlin": """
|
|
(class_declaration
|
|
name: (simple_identifier) @class_name) @class_decl
|
|
""",
|
|
"swift": """
|
|
(class_declaration
|
|
name: (type_identifier) @class_name) @class_decl
|
|
""",
|
|
}
|
|
|
|
|
|
class GenericExtractor:
|
|
def __init__(self):
|
|
self.file_stem: str = ""
|
|
self.file_path: str = ""
|
|
self.source_bytes: bytes = b""
|
|
|
|
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()
|
|
except (OSError, IOError):
|
|
result.errors.append(f"Cannot read file: {file_info.path}")
|
|
result.metadata["extraction_quality"] = "failed"
|
|
return result
|
|
|
|
tree = parse_file(file_info.path, file_info.language)
|
|
if tree is None:
|
|
result.metadata["extraction_quality"] = "failed"
|
|
return result
|
|
|
|
try:
|
|
self.file_stem = Path(file_info.path).stem
|
|
result.file_path = file_info.path
|
|
except Exception:
|
|
pass
|
|
|
|
lang = file_info.language
|
|
|
|
module_id = make_module_node_id(self.file_stem)
|
|
module_node = Node(
|
|
id=module_id,
|
|
node_type=NodeType.MODULE,
|
|
label=Path(self.file_path).name,
|
|
file_path=self.file_path,
|
|
line_start=1,
|
|
line_end=0,
|
|
metadata={"language": lang, "imports": []},
|
|
)
|
|
result.nodes.append(module_node)
|
|
|
|
_FUNC_CAP = {
|
|
"java": "method_decl", "go": "func_decl", "ruby": "method_def",
|
|
"rust": "func_item", "c": "func_def", "cpp": "func_def",
|
|
"php": "func_def", "c_sharp": "method_decl", "kotlin": "func_decl",
|
|
"swift": "func_decl", "scala": "func_def", "bash": "func_def",
|
|
}
|
|
if lang in FUNCTION_QUERIES:
|
|
captures = query_tree(tree, lang, FUNCTION_QUERIES[lang], capture_filter=_FUNC_CAP.get(lang))
|
|
for cap_node, _ in captures:
|
|
func_name = get_node_text(cap_node, self.source_bytes)
|
|
if not func_name:
|
|
continue
|
|
node_id = make_node_id(self.file_stem, func_name)
|
|
try:
|
|
start = cap_node.start_point[0] + 1
|
|
end = cap_node.end_point[0] + 1
|
|
except Exception:
|
|
start, end = 0, 0
|
|
node = Node(
|
|
id=node_id,
|
|
node_type=NodeType.FUNCTION,
|
|
label=func_name[:40],
|
|
file_path=self.file_path,
|
|
line_start=start,
|
|
line_end=end,
|
|
metadata={"language": lang},
|
|
)
|
|
result.nodes.append(node)
|
|
result.edges.append(Edge(
|
|
source_id=module_id,
|
|
target_id=node_id,
|
|
edge_type=EdgeType.CONTAINS,
|
|
))
|
|
|
|
_CLASS_CAP: dict[str, str | set[str]] = {
|
|
"java": "class_decl", "go": "type_decl", "ruby": "class_def",
|
|
"rust": {"struct_item", "impl_item"}, "c_sharp": "class_decl",
|
|
"kotlin": "class_decl", "swift": "class_decl",
|
|
}
|
|
if lang in CLASS_QUERIES:
|
|
captures = query_tree(tree, lang, CLASS_QUERIES[lang], capture_filter=_CLASS_CAP.get(lang))
|
|
for cap_node, _ in captures:
|
|
cls_name = get_node_text(cap_node, self.source_bytes)
|
|
if not cls_name:
|
|
continue
|
|
node_id = make_node_id(self.file_stem, cls_name)
|
|
try:
|
|
start = cap_node.start_point[0] + 1
|
|
end = cap_node.end_point[0] + 1
|
|
except Exception:
|
|
start, end = 0, 0
|
|
node = Node(
|
|
id=node_id,
|
|
node_type=NodeType.CLASS,
|
|
label=cls_name,
|
|
file_path=self.file_path,
|
|
line_start=start,
|
|
line_end=end,
|
|
metadata={"language": lang},
|
|
)
|
|
result.nodes.append(node)
|
|
result.edges.append(Edge(
|
|
source_id=module_id,
|
|
target_id=node_id,
|
|
edge_type=EdgeType.CONTAINS,
|
|
))
|
|
|
|
source_text = self.source_bytes.decode("utf-8", errors="replace")
|
|
|
|
source_patterns = {
|
|
"java": [r"System\.getenv\s*\(\s*['\"](\w+)['\"]\s*\)", r"System\.console\(\)"],
|
|
"go": [r"os\.Getenv\s*\(\s*['\"](\w+)['\"]\s*\)", r"os\.Args"],
|
|
"ruby": [r"ENV\s*\[\s*['\"](\w+)['\"]\s*\]", r"ARGV"],
|
|
"php": [r"\$_GET\s*\[\s*['\"](\w+)['\"]\s*\]", r"\$_POST\s*\[\s*['\"](\w+)['\"]\s*\]",
|
|
r"\$_REQUEST\s*\[\s*['\"](\w+)['\"]\s*\]", r"\$_SERVER"],
|
|
}
|
|
|
|
if lang in source_patterns:
|
|
for pattern in source_patterns[lang]:
|
|
for match in re.finditer(pattern, source_text):
|
|
line_num = source_text[:match.start()].count("\n") + 1
|
|
param_name = match.group(1) if match.lastindex and match.lastindex >= 1 else match.group(0)[:20]
|
|
containing_func = None
|
|
for n in result.nodes:
|
|
if n.node_type in (NodeType.FUNCTION, NodeType.ENTRY_POINT):
|
|
if n.line_start <= line_num <= n.line_end:
|
|
containing_func = n.id
|
|
break
|
|
if containing_func:
|
|
func_part = containing_func.split("::")[-1]
|
|
source_id = make_node_id(self.file_stem, f"{func_part}::{param_name}")
|
|
if not any(n.id == source_id for n in result.nodes):
|
|
result.nodes.append(Node(
|
|
id=source_id,
|
|
node_type=NodeType.SOURCE,
|
|
label=f"param: {param_name} [generic]"[:40],
|
|
file_path=self.file_path,
|
|
line_start=line_num,
|
|
line_end=line_num,
|
|
metadata={"param_name": str(param_name), "input_type": "generic", "language": lang},
|
|
))
|
|
result.edges.append(Edge(
|
|
source_id=source_id,
|
|
target_id=containing_func,
|
|
edge_type=EdgeType.DATA_FLOW,
|
|
))
|
|
|
|
result.metadata["extraction_quality"] = "partial"
|
|
return result
|
|
|
|
|
|
def extract(file_info: FileInfo) -> ExtractionResult:
|
|
extractor = GenericExtractor()
|
|
return extractor.extract(file_info)
|