Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 7 additions & 3 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -48,9 +48,13 @@ coretrace-python-analyzer --emit-ir example.py
PYTHONPATH=src python -m coretrace_python --emit-ir example.py
```

Currently supported inside functions: arguments, assignments, names, constants, binary
operations, unary operations, comparisons, calls, attribute access, indexing, returns,
`if`/`elif`/`else`, `while`, `for`, `break`, `continue` and `raise`. Each function is
Currently supported inside functions: parameters with defaults and keyword-only or star
forms, decorators, assignments to names, attributes, items and unpacked tuples, augmented
assignment, list, tuple and dict literals, `and`/`or`, chained comparisons, keyword
arguments, `with`, `assert`, `if`/`elif`/`else`, `while`, `for`, `break`, `continue` and
`raise`. Methods of module-level classes are analysed like functions; other module-level
code is skipped. A function using syntax outside this subset is reported by `--check` as
an `unsupported-syntax` note and the other functions are still analysed. Each function is
emitted as its control-flow graph: one block per basic block, ending in an explicit
terminator (`branch`, `jump`, `return`, `raise`, `for_next`). Add `--ssa` to print the
static single assignment form: locals become numbered values, merges get `phi`
Expand Down
14 changes: 14 additions & 0 deletions src/coretrace_python/cfg/builder.py
Original file line number Diff line number Diff line change
Expand Up @@ -102,9 +102,23 @@ def statement(
return self.conditional(node, block, continuation)
if isinstance(node, nodes.While | nodes.For):
return self.loop_statement(node, block, continuation)
if isinstance(node, nodes.With):
return self.with_statement(node, block)
block.statements.append(node)
return block

def with_statement(self, node: nodes.With, block: _Open) -> _Open | None:
"""Lay the body out inline; an early exit skips the ``ExitWith`` statements."""

for item in node.items:
block.statements.append(nodes.EnterWith(item, node.span))
current = self.sequence(node.body, block, None, node.span)
if current is None:
return None
for item in reversed(node.items):
current.statements.append(nodes.ExitWith(item, node.span))
return current

def conditional(
self, node: nodes.If, block: _Open, continuation: BlockId | None
) -> _Open | None:
Expand Down
33 changes: 29 additions & 4 deletions src/coretrace_python/engine.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,12 +12,17 @@
from coretrace_python import __version__
from coretrace_python.abstract import ConstantPropagation
from coretrace_python.analysis import AnalysisManager, AnyAnalysis
from coretrace_python.cfg import CFGAnalysis, DominanceAnalysis, PostDominanceAnalysis
from coretrace_python.findings import Finding
from coretrace_python.cfg import CFGAnalysis, CFGError, DominanceAnalysis, PostDominanceAnalysis
from coretrace_python.findings import Confidence, Finding, Severity
from coretrace_python.frontend import build_hir
from coretrace_python.hir import nodes
from coretrace_python.ir.defuse import DefUseAnalysis
from coretrace_python.ir.lowering import ModuleIRAnalysis, PyIRAnalysis
from coretrace_python.ir.lowering import (
LoweringError,
ModuleIRAnalysis,
PyIRAnalysis,
analyzable_functions,
)
from coretrace_python.ir.ssa import SSAAnalysis
from coretrace_python.plugins import PluginRegistry, discover_plugins, run_plugins
from coretrace_python.reporters import Report
Expand Down Expand Up @@ -68,7 +73,27 @@ def check(source: SourceFile, plugin_roots: Sequence[Path]) -> tuple[Finding, ..
for loaded in registry:
models.register(*loaded.plugin.models)
manager.provide(SecurityModelAnalysis, models.freeze())
return run_plugins(manager, [loaded.plugin for loaded in registry])

supported: list[nodes.Function] = []
notes: list[Finding] = []
for function in analyzable_functions(manager.module):
try:
manager.get(SSAAnalysis, function)
except (LoweringError, CFGError) as error:
notes.append(
Finding(
rule_id="unsupported-syntax",
message=str(error),
severity=Severity.INFO,
confidence=Confidence.HIGH,
span=function.span,
function=function.name,
)
)
else:
supported.append(function)
findings = run_plugins(manager, [loaded.plugin for loaded in registry], tuple(supported))
return (*findings, *notes)


def _register_all(module: nodes.Module) -> AnalysisManager:
Expand Down
132 changes: 102 additions & 30 deletions src/coretrace_python/frontend/ast_adapter.py
Original file line number Diff line number Diff line change
Expand Up @@ -83,22 +83,40 @@ def expression(self, node: ast.expr) -> nodes.Expression:
self.fail(node.op)
return nodes.UnaryOp(operator, self.expression(node.operand), span)
if isinstance(node, ast.Compare):
if len(node.ops) != 1 or len(node.comparators) != 1:
self.fail(node, "chained comparisons are not supported yet")
operator = _COMPARE_OPERATORS.get(type(node.ops[0]))
if operator is None:
self.fail(node.ops[0])
return nodes.Compare(
operator,
self.expression(node.left),
self.expression(node.comparators[0]),
span,
)
comparisons: list[nodes.Expression] = []
left = self.expression(node.left)
for op, comparator in zip(node.ops, node.comparators, strict=True):
operator = _COMPARE_OPERATORS.get(type(op))
if operator is None:
self.fail(op)
right = self.expression(comparator)
comparisons.append(nodes.Compare(operator, left, right, span))
left = right
if len(comparisons) == 1:
return comparisons[0]
return nodes.BoolOp("and", tuple(comparisons), span)
if isinstance(node, ast.BoolOp):
operator = "and" if isinstance(node.op, ast.And) else "or"
return nodes.BoolOp(operator, tuple(self.expression(v) for v in node.values), span)
if isinstance(node, ast.Tuple):
return nodes.Tuple(self.elements(node.elts), span)
if isinstance(node, ast.List):
return nodes.List(self.elements(node.elts), span)
if isinstance(node, ast.Dict):
items = []
for key, value in zip(node.keys, node.values, strict=True):
if key is None:
self.fail(value, "dict unpacking is not supported yet")
items.append((self.expression(key), self.expression(value)))
return nodes.Dict(tuple(items), span)
if isinstance(node, ast.Attribute):
return nodes.Attribute(self.expression(node.value), node.attr, span)
if isinstance(node, ast.Subscript):
return nodes.Subscript(self.expression(node.value), self.expression(node.slice), span)
if isinstance(node, ast.Call):
for argument in node.args:
if isinstance(argument, ast.Starred):
self.fail(argument, "star arguments are not supported yet")
arguments = tuple(self.expression(argument) for argument in node.args)
keywords = tuple(
nodes.Keyword(keyword.arg, self.expression(keyword.value), self.span(keyword))
Expand All @@ -112,6 +130,26 @@ def expression(self, node: ast.expr) -> nodes.Expression:
)
self.fail(node)

def elements(self, elements: list[ast.expr]) -> tuple[nodes.Expression, ...]:
for element in elements:
if isinstance(element, ast.Starred):
self.fail(element, "star expressions are not supported yet")
return tuple(self.expression(element) for element in elements)

def target(self, node: ast.expr) -> nodes.Target:
if isinstance(node, ast.Name):
return nodes.Name(node.id, self.span(node))
if isinstance(node, (ast.Attribute, ast.Subscript)):
target = self.expression(node)
assert isinstance(target, (nodes.Attribute, nodes.Subscript))
return target
if isinstance(node, (ast.Tuple, ast.List)):
for element in node.elts:
if isinstance(element, ast.Starred):
self.fail(element, "starred assignment targets are not supported yet")
return nodes.Tuple(tuple(self.target(element) for element in node.elts), self.span(node))
self.fail(node, f"unsupported assignment target: {type(node).__name__}")

def generator(self, node: ast.comprehension) -> nodes.ComprehensionGenerator:
if node.is_async:
self.fail(node.target, "async comprehensions are not supported yet")
Expand All @@ -133,10 +171,36 @@ def generator(self, node: ast.comprehension) -> nodes.ComprehensionGenerator:
def statement(self, node: ast.stmt) -> nodes.Statement:
span = self.span(node)
if isinstance(node, ast.Assign):
if len(node.targets) != 1 or not isinstance(node.targets[0], ast.Name):
self.fail(node, "only assignment to one local name is supported")
target = nodes.Name(node.targets[0].id, self.span(node.targets[0]))
return nodes.Assign(target, self.expression(node.value), span)
if len(node.targets) != 1:
self.fail(node, "chained assignment is not supported yet")
return nodes.Assign(self.target(node.targets[0]), self.expression(node.value), span)
if isinstance(node, ast.AugAssign):
operator = _BINARY_OPERATORS.get(type(node.op))
if operator is None:
self.fail(node.op)
target = self.target(node.target)
if isinstance(target, nodes.Tuple):
self.fail(node.target, "unsupported augmented assignment target")
return nodes.AugAssign(target, operator, self.expression(node.value), span)
if isinstance(node, ast.Assert):
message = self.expression(node.msg) if node.msg is not None else None
return nodes.Assert(self.expression(node.test), message, span)
if isinstance(node, (ast.With, ast.AsyncWith)):
items = []
for item in node.items:
bound: nodes.Name | None = None
if item.optional_vars is not None:
if not isinstance(item.optional_vars, ast.Name):
self.fail(
item.optional_vars, "only a single name is supported as a with target"
)
bound = nodes.Name(item.optional_vars.id, self.span(item.optional_vars))
items.append(
nodes.WithItem(
self.expression(item.context_expr), bound, self.span(item.context_expr)
)
)
return nodes.With(tuple(items), self.block(node.body), isinstance(node, ast.AsyncWith), span)
if isinstance(node, ast.Return):
value = self.expression(node.value) if node.value is not None else None
return nodes.Return(value, span)
Expand Down Expand Up @@ -198,35 +262,43 @@ def block(self, statements: list[ast.stmt]) -> tuple[nodes.Statement, ...]:
return tuple(self.statement(statement) for statement in statements)

def class_definition(self, node: ast.ClassDef) -> nodes.Class:
if node.decorator_list:
self.fail(node, "decorated classes are not supported yet")
if node.keywords:
self.fail(node, "class keyword arguments are not supported yet")
bases = tuple(self.expression(base) for base in node.bases)
body = tuple(self.statement(statement) for statement in node.body)
return nodes.Class(node.name, bases, body, self.span(node))
decorators = tuple(self.expression(d) for d in node.decorator_list)
return nodes.Class(node.name, bases, body, self.span(node), decorators)

def function(self, node: ast.FunctionDef | ast.AsyncFunctionDef) -> nodes.Function:
if node.decorator_list:
self.fail(node, "decorated functions are not supported yet")
arguments = node.args
has_advanced_arguments = (
arguments.posonlyargs or arguments.kwonlyargs or arguments.vararg or arguments.kwarg
)
if has_advanced_arguments or arguments.defaults or arguments.kw_defaults:
self.fail(arguments, "only required positional arguments are supported")
parameters = tuple(
nodes.Parameter(argument.arg, self.span(argument)) for argument in arguments.args
)
body = tuple(self.statement(statement) for statement in node.body)
return nodes.Function(
name=node.name,
parameters=parameters,
parameters=self.parameters(node.args),
body=body,
is_async=isinstance(node, ast.AsyncFunctionDef),
span=self.span(node),
decorators=tuple(self.expression(d) for d in node.decorator_list),
)

def parameters(self, arguments: ast.arguments) -> tuple[nodes.Parameter, ...]:
positional = [*arguments.posonlyargs, *arguments.args]
defaults: list[ast.expr | None] = [None] * (len(positional) - len(arguments.defaults))
defaults.extend(arguments.defaults)
parameters: list[nodes.Parameter] = []
for argument, default in zip(positional, defaults, strict=True):
parameters.append(self.parameter(argument, default, "positional"))
if arguments.vararg is not None:
parameters.append(self.parameter(arguments.vararg, None, "var_positional"))
for argument, default in zip(arguments.kwonlyargs, arguments.kw_defaults, strict=True):
parameters.append(self.parameter(argument, default, "keyword"))
if arguments.kwarg is not None:
parameters.append(self.parameter(arguments.kwarg, None, "var_keyword"))
return tuple(parameters)

def parameter(self, argument: ast.arg, default: ast.expr | None, kind: str) -> nodes.Parameter:
value = self.expression(default) if default is not None else None
return nodes.Parameter(argument.arg, self.span(argument), value, kind)

def module(self, tree: ast.Module) -> nodes.Module:
body = tuple(self.statement(statement) for statement in tree.body)
if tree.body:
Expand Down
4 changes: 2 additions & 2 deletions src/coretrace_python/hir/__init__.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
"""Parser-independent high-level representation of Python source."""

from coretrace_python.hir.nodes import Module
from coretrace_python.hir.nodes import HIR_SCHEMA_VERSION, Module

__all__ = ["Module"]
__all__ = ["HIR_SCHEMA_VERSION", "Module"]
Loading
Loading