diff --git a/README.md b/README.md index 2ccb077..3245d0e 100644 --- a/README.md +++ b/README.md @@ -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` diff --git a/src/coretrace_python/cfg/builder.py b/src/coretrace_python/cfg/builder.py index 9df6d68..fade2b9 100644 --- a/src/coretrace_python/cfg/builder.py +++ b/src/coretrace_python/cfg/builder.py @@ -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: diff --git a/src/coretrace_python/engine.py b/src/coretrace_python/engine.py index 9c46024..8d6fa71 100644 --- a/src/coretrace_python/engine.py +++ b/src/coretrace_python/engine.py @@ -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 @@ -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: diff --git a/src/coretrace_python/frontend/ast_adapter.py b/src/coretrace_python/frontend/ast_adapter.py index 88201de..93140e7 100644 --- a/src/coretrace_python/frontend/ast_adapter.py +++ b/src/coretrace_python/frontend/ast_adapter.py @@ -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)) @@ -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") @@ -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) @@ -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: diff --git a/src/coretrace_python/hir/__init__.py b/src/coretrace_python/hir/__init__.py index 4083bf0..013689c 100644 --- a/src/coretrace_python/hir/__init__.py +++ b/src/coretrace_python/hir/__init__.py @@ -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"] diff --git a/src/coretrace_python/hir/nodes.py b/src/coretrace_python/hir/nodes.py index 93151c5..0814694 100644 --- a/src/coretrace_python/hir/nodes.py +++ b/src/coretrace_python/hir/nodes.py @@ -7,11 +7,7 @@ from coretrace_python.source import SourceSpan - -@dataclass(frozen=True, slots=True) -class Parameter: - name: str - span: SourceSpan +HIR_SCHEMA_VERSION = 1 @dataclass(frozen=True, slots=True) @@ -41,6 +37,15 @@ class UnaryOp: span: SourceSpan +@dataclass(frozen=True, slots=True) +class BoolOp: + """``and`` / ``or`` over two or more values; chained comparisons lower to ``and``.""" + + operator: str + values: tuple[Expression, ...] + span: SourceSpan + + @dataclass(frozen=True, slots=True) class Compare: operator: str @@ -65,6 +70,8 @@ class Subscript: @dataclass(frozen=True, slots=True) class Keyword: + """A keyword argument; ``name`` is ``None`` for ``**mapping`` unpacking.""" + name: str | None value: Expression span: SourceSpan @@ -78,6 +85,24 @@ class Call: span: SourceSpan +@dataclass(frozen=True, slots=True) +class Tuple: + elements: tuple[Expression, ...] + span: SourceSpan + + +@dataclass(frozen=True, slots=True) +class List: + elements: tuple[Expression, ...] + span: SourceSpan + + +@dataclass(frozen=True, slots=True) +class Dict: + items: tuple[tuple[Expression, Expression], ...] + span: SourceSpan + + @dataclass(frozen=True, slots=True) class ComprehensionGenerator: """One ``for target in iterable if condition...`` clause of a comprehension.""" @@ -103,56 +128,66 @@ class Comprehension: | Constant | BinaryOp | UnaryOp + | BoolOp | Compare | Attribute | Subscript | Call + | Tuple + | List + | Dict | Comprehension ) +Target: TypeAlias = Name | Attribute | Subscript | Tuple + @dataclass(frozen=True, slots=True) -class Assign: - target: Name - value: Expression +class Parameter: + """``kind`` is ``positional``, ``keyword``, ``var_positional`` or ``var_keyword``.""" + + name: str span: SourceSpan + default: Expression | None = None + kind: str = "positional" @dataclass(frozen=True, slots=True) -class Return: - value: Expression | None +class Assign: + target: Target + value: Expression span: SourceSpan @dataclass(frozen=True, slots=True) -class ExpressionStatement: - expression: Expression +class AugAssign: + target: Name | Attribute | Subscript + operator: str + value: Expression span: SourceSpan @dataclass(frozen=True, slots=True) -class Pass: +class Return: + value: Expression | None span: SourceSpan @dataclass(frozen=True, slots=True) -class ImportAlias: - name: str - as_name: str | None +class ExpressionStatement: + expression: Expression span: SourceSpan @dataclass(frozen=True, slots=True) -class Import: - names: tuple[ImportAlias, ...] +class Pass: span: SourceSpan @dataclass(frozen=True, slots=True) -class ImportFrom: - module: str | None - names: tuple[ImportAlias, ...] - level: int +class Assert: + test: Expression + message: Expression | None span: SourceSpan @@ -196,6 +231,58 @@ class Raise: span: SourceSpan +@dataclass(frozen=True, slots=True) +class WithItem: + context: Expression + target: Name | None + span: SourceSpan + + +@dataclass(frozen=True, slots=True) +class With: + items: tuple[WithItem, ...] + body: tuple[Statement, ...] + is_async: bool + span: SourceSpan + + +@dataclass(frozen=True, slots=True) +class EnterWith: + """Synthetic statement the CFG builder emits when a ``with`` body starts.""" + + item: WithItem + span: SourceSpan + + +@dataclass(frozen=True, slots=True) +class ExitWith: + """Synthetic statement the CFG builder emits when a ``with`` body falls through.""" + + item: WithItem + span: SourceSpan + + +@dataclass(frozen=True, slots=True) +class ImportAlias: + name: str + as_name: str | None + span: SourceSpan + + +@dataclass(frozen=True, slots=True) +class Import: + names: tuple[ImportAlias, ...] + span: SourceSpan + + +@dataclass(frozen=True, slots=True) +class ImportFrom: + module: str | None + names: tuple[ImportAlias, ...] + level: int + span: SourceSpan + + @dataclass(frozen=True, slots=True) class Global: names: tuple[str, ...] @@ -215,6 +302,7 @@ class Function: body: tuple[Statement, ...] is_async: bool span: SourceSpan + decorators: tuple[Expression, ...] = () @dataclass(frozen=True, slots=True) @@ -223,19 +311,25 @@ class Class: bases: tuple[Expression, ...] body: tuple[Statement, ...] span: SourceSpan + decorators: tuple[Expression, ...] = () Statement: TypeAlias = ( Assign + | AugAssign | Return | ExpressionStatement | Pass + | Assert | If | While | For | Break | Continue | Raise + | With + | EnterWith + | ExitWith | Import | ImportFrom | Global @@ -250,4 +344,3 @@ class Module: name: str body: tuple[Statement, ...] span: SourceSpan - diff --git a/src/coretrace_python/hir/visitors.py b/src/coretrace_python/hir/visitors.py index a7d8b7c..608652f 100644 --- a/src/coretrace_python/hir/visitors.py +++ b/src/coretrace_python/hir/visitors.py @@ -7,21 +7,30 @@ from coretrace_python.hir import nodes -Node = nodes.Statement | nodes.Expression | nodes.ComprehensionGenerator | nodes.Keyword +Node = ( + nodes.Statement + | nodes.Expression + | nodes.ComprehensionGenerator + | nodes.Keyword + | nodes.WithItem + | nodes.Parameter +) def children(node: Node) -> Iterator[Node]: """Yield the direct child nodes of ``node`` in source order.""" for field in fields(node): - value = getattr(node, field.name) - if _is_node(value): - yield value - elif isinstance(value, tuple): - for item in value: - if _is_node(item): - yield item + yield from _nodes_in(getattr(node, field.name)) + + +def _nodes_in(value: object) -> Iterator[Node]: + if _is_node(value): + yield value # type: ignore[misc] + elif isinstance(value, tuple): + for item in value: + yield from _nodes_in(item) def _is_node(value: object) -> bool: - return hasattr(value, "span") and not isinstance(value, nodes.Parameter | nodes.ImportAlias) + return hasattr(value, "span") and not isinstance(value, nodes.ImportAlias) diff --git a/src/coretrace_python/ir/lowering.py b/src/coretrace_python/ir/lowering.py index 923149e..e9bc64c 100644 --- a/src/coretrace_python/ir/lowering.py +++ b/src/coretrace_python/ir/lowering.py @@ -17,12 +17,18 @@ from coretrace_python.hir import nodes from coretrace_python.hir.visitors import Node, children from coretrace_python.ir.model import ( + Assert, BasicBlock, BinaryOp, + BoolOp, Branch, + BuildDict, + BuildList, + BuildTuple, Call, Compare, Constant, + EffectInstruction, ForNext, FunctionIR, GetAttr, @@ -35,12 +41,16 @@ ModuleIR, Raise, Return, + SetAttr, + SetItem, StoreLocal, Symbol, Terminator, UnaryOp, Value, ValueInstruction, + WithEnter, + WithExit, ) from coretrace_python.semantic import SEMANTIC_ANALYSES from coretrace_python.semantic.scopes import ( @@ -48,9 +58,11 @@ ResolutionKind, Scope, ScopeAnalysis, + ScopeKind, ScopeTable, ) from coretrace_python.semantic.symbols import SymbolAnalysis, SymbolId, SymbolTable +from coretrace_python.source import SourceSpan class LoweringError(Exception): @@ -67,6 +79,7 @@ class _FunctionLowerer: parameters: dict[str, Value] = field(default_factory=dict) instructions: list[Instruction] = field(default_factory=list) iterators: dict[BlockId, Value] = field(default_factory=dict) + contexts: dict[SourceSpan, Value] = field(default_factory=dict) def fail( self, @@ -85,7 +98,7 @@ def emit(self, instruction: ValueInstruction) -> Value: self.instructions.append(instruction) return instruction.result - def emit_effect(self, instruction: StoreLocal) -> None: + def emit_effect(self, instruction: EffectInstruction) -> None: self.instructions.append(instruction) def resolve(self, name: str) -> Resolution: @@ -128,11 +141,22 @@ def expression(self, node: nodes.Expression) -> Value: right = self.expression(node.right) return self.emit(Compare(self.new_value(), node.span, node.operator, left, right)) if isinstance(node, nodes.Call): - if node.keywords: - self.fail(node, "keyword arguments are not supported yet") callee = self.expression(node.callee) arguments = tuple(self.expression(argument) for argument in node.arguments) - return self.emit(Call(self.new_value(), node.span, callee, arguments)) + keywords = tuple((k.name, self.expression(k.value)) for k in node.keywords) + return self.emit(Call(self.new_value(), node.span, callee, arguments, keywords)) + if isinstance(node, nodes.BoolOp): + values = tuple(self.expression(value) for value in node.values) + return self.emit(BoolOp(self.new_value(), node.span, node.operator, values)) + if isinstance(node, nodes.List): + elements = tuple(self.expression(element) for element in node.elements) + return self.emit(BuildList(self.new_value(), node.span, elements)) + if isinstance(node, nodes.Tuple): + elements = tuple(self.expression(element) for element in node.elements) + return self.emit(BuildTuple(self.new_value(), node.span, elements)) + if isinstance(node, nodes.Dict): + items = tuple((self.expression(k), self.expression(v)) for k, v in node.items) + return self.emit(BuildDict(self.new_value(), node.span, items)) if isinstance(node, nodes.Attribute): object_value = self.expression(node.value) return self.emit(GetAttr(self.new_value(), node.span, object_value, node.name)) @@ -144,15 +168,50 @@ def expression(self, node: nodes.Expression) -> Value: # ------------------------------------------------------------------ statements - def store(self, target: nodes.Name, value: Value) -> None: - if self.resolve(target.identifier).kind is not ResolutionKind.LOCAL: - self.fail(target, "assignment to a global or nonlocal name is not supported yet") - self.emit_effect(StoreLocal(None, target.span, target.identifier, value)) + def store(self, target: nodes.Target, value: Value) -> None: + if isinstance(target, nodes.Name): + if self.resolve(target.identifier).kind is not ResolutionKind.LOCAL: + self.fail(target, "assignment to a global or nonlocal name is not supported yet") + self.emit_effect(StoreLocal(None, target.span, target.identifier, value)) + elif isinstance(target, nodes.Attribute): + object_value = self.expression(target.value) + self.emit_effect(SetAttr(None, target.span, object_value, target.name, value)) + elif isinstance(target, nodes.Subscript): + object_value = self.expression(target.value) + key = self.expression(target.key) + self.emit_effect(SetItem(None, target.span, object_value, key, value)) + else: + for index, element in enumerate(target.elements): + position = self.emit(Constant(self.new_value(), element.span, index)) + item = self.emit(GetItem(self.new_value(), element.span, value, position)) + assert isinstance(element, nodes.Name | nodes.Attribute | nodes.Subscript | nodes.Tuple) + self.store(element, item) def statement(self, node: nodes.Statement) -> None: if isinstance(node, nodes.Assign): self.store(node.target, self.expression(node.value)) return + if isinstance(node, nodes.AugAssign): + current = self.expression(node.target) + operand = self.expression(node.value) + result = self.emit(BinaryOp(self.new_value(), node.span, node.operator, current, operand)) + self.store(node.target, result) + return + if isinstance(node, nodes.Assert): + test = self.expression(node.test) + message = self.expression(node.message) if node.message is not None else None + self.emit_effect(Assert(None, node.span, test, message)) + return + if isinstance(node, nodes.EnterWith): + context = self.expression(node.item.context) + self.contexts[node.item.span] = context + entered = self.emit(WithEnter(self.new_value(), node.item.span, context)) + if node.item.target is not None: + self.store(node.item.target, entered) + return + if isinstance(node, nodes.ExitWith): + self.emit_effect(WithExit(None, node.item.span, self.contexts[node.item.span])) + return if isinstance(node, nodes.ExpressionStatement): self.expression(node.expression) return @@ -218,7 +277,17 @@ def function(self, node: nodes.Function) -> FunctionIR: self.statement(statement) terminator = self.terminator(cfg_block) blocks.append(BasicBlock(cfg_block.id, tuple(self.instructions), terminator)) - return FunctionIR(node.name, parameter_values, self.cfg.entry, tuple(blocks), node.span) + return FunctionIR( + self.qualified_name(node), parameter_values, self.cfg.entry, tuple(blocks), node.span + ) + + def qualified_name(self, node: nodes.Function) -> str: + names = [node.name] + parent = self.scopes.scope(self.scope.parent) if self.scope.parent else None + while parent is not None and parent.kind is not ScopeKind.MODULE: + names.append(parent.name) + parent = self.scopes.scope(parent.parent) if parent.parent else None + return ".".join(reversed(names)) def _reassigned_parameters(function: nodes.Function) -> frozenset[str]: @@ -226,9 +295,16 @@ def _reassigned_parameters(function: nodes.Function) -> frozenset[str]: assigned: set[str] = set() + def names(target: nodes.Target | None) -> None: + if isinstance(target, nodes.Name): + assigned.add(target.identifier) + elif isinstance(target, nodes.Tuple): + for element in target.elements: + names(element) # type: ignore[arg-type] + def walk(node: Node) -> None: - if isinstance(node, nodes.Assign | nodes.For): - assigned.add(node.target.identifier) + if isinstance(node, nodes.Assign | nodes.AugAssign | nodes.For | nodes.WithItem): + names(node.target) if isinstance(node, nodes.Function | nodes.Class | nodes.Comprehension): return for child in children(node): @@ -259,27 +335,15 @@ def compute(cls, ctx: AnalysisContext, function: nodes.Function) -> FunctionIR: return lowerer.function(function) -def top_level_functions(module: nodes.Module) -> tuple[nodes.Function, ...]: - """The functions a module lowers to, rejecting module syntax outside the subset.""" +def analyzable_functions(module: nodes.Module) -> tuple[nodes.Function, ...]: + """Top-level functions and the methods of top-level classes, in source order.""" functions: list[nodes.Function] = [] for statement in module.body: if isinstance(statement, nodes.Function): functions.append(statement) - elif isinstance(statement, (nodes.Import, nodes.ImportFrom)): - continue - elif ( - isinstance(statement, nodes.ExpressionStatement) - and isinstance(statement.expression, nodes.Constant) - and isinstance(statement.expression.value, str) - ): - # Permit module docstrings. - continue - else: - raise LoweringError( - f"{statement.span.display()}: " - f"unsupported module syntax: {type(statement).__name__}" - ) + elif isinstance(statement, nodes.Class): + functions.extend(s for s in statement.body if isinstance(s, nodes.Function)) return tuple(functions) @@ -292,7 +356,7 @@ class ModuleIRAnalysis(Analysis[ModuleIR]): @classmethod def compute(cls, ctx: AnalysisContext) -> ModuleIR: return ModuleIR( - tuple(ctx.get(PyIRAnalysis, function) for function in top_level_functions(ctx.module)) + tuple(ctx.get(PyIRAnalysis, function) for function in analyzable_functions(ctx.module)) ) @@ -309,5 +373,5 @@ def lower_module(module: nodes.Module, *, ssa: bool = False) -> ModuleIR: manager.register(DominanceAnalysis, SSAAnalysis) return ModuleIR( - tuple(manager.get(SSAAnalysis, function) for function in top_level_functions(module)) + tuple(manager.get(SSAAnalysis, function) for function in analyzable_functions(module)) ) diff --git a/src/coretrace_python/ir/model.py b/src/coretrace_python/ir/model.py index f093e01..85ccbfb 100644 --- a/src/coretrace_python/ir/model.py +++ b/src/coretrace_python/ir/model.py @@ -20,20 +20,27 @@ class Operands: def operands(self) -> tuple[Value, ...]: found: list[Value] = [] for field_info in fields(self): # type: ignore[arg-type] - if field_info.name == "result": - continue - value = getattr(self, field_info.name) - if isinstance(value, Value): - found.append(value) - elif isinstance(value, tuple): - for item in value: - if isinstance(item, Value): - found.append(item) - elif isinstance(item, tuple) and item and isinstance(item[0], Value): - found.append(item[0]) + if field_info.name != "result": + found.extend(_values_in(getattr(self, field_info.name))) return tuple(found) +def _values_in(value: object) -> list[Value]: + if isinstance(value, Value): + return [value] + if isinstance(value, tuple): + return [v for item in value for v in _values_in(item)] + return [] + + +def _map_values(value: Any, mapping: Callable[[Value], Value]) -> Any: + if isinstance(value, Value): + return mapping(value) + if isinstance(value, tuple): + return tuple(_map_values(item, mapping) for item in value) + return value + + N = TypeVar("N", bound=Operands) @@ -46,17 +53,8 @@ def substitute(node: N, mapping: Callable[[Value], Value], *, include_result: bo if field_info.name == "result": if include_result and isinstance(value, Value): changes["result"] = mapping(value) - elif isinstance(value, Value): - changes[field_info.name] = mapping(value) - elif isinstance(value, tuple): - changes[field_info.name] = tuple( - mapping(item) - if isinstance(item, Value) - else (mapping(item[0]), *item[1:]) - if isinstance(item, tuple) and item and isinstance(item[0], Value) - else item - for item in value - ) + else: + changes[field_info.name] = _map_values(value, mapping) return replace(node, **changes) # type: ignore[type-var] @@ -113,6 +111,46 @@ class Call(Operands): location: SourceSpan callee: Value arguments: tuple[Value, ...] + keywords: tuple[tuple[str | None, Value], ...] = () + + def argument_values(self) -> tuple[Value, ...]: + return (*self.arguments, *(value for _, value in self.keywords)) + + +@dataclass(frozen=True) +class BoolOp(Operands): + result: Value + location: SourceSpan + operator: str + values: tuple[Value, ...] + + +@dataclass(frozen=True) +class BuildList(Operands): + result: Value + location: SourceSpan + elements: tuple[Value, ...] + + +@dataclass(frozen=True) +class BuildTuple(Operands): + result: Value + location: SourceSpan + elements: tuple[Value, ...] + + +@dataclass(frozen=True) +class BuildDict(Operands): + result: Value + location: SourceSpan + items: tuple[tuple[Value, Value], ...] + + +@dataclass(frozen=True) +class WithEnter(Operands): + result: Value + location: SourceSpan + context: Value @dataclass(frozen=True) @@ -172,6 +210,39 @@ class StoreLocal(Operands): value: Value +@dataclass(frozen=True) +class SetAttr(Operands): + result: None + location: SourceSpan + object: Value + attribute: str + value: Value + + +@dataclass(frozen=True) +class SetItem(Operands): + result: None + location: SourceSpan + object: Value + key: Value + value: Value + + +@dataclass(frozen=True) +class WithExit(Operands): + result: None + location: SourceSpan + context: Value + + +@dataclass(frozen=True) +class Assert(Operands): + result: None + location: SourceSpan + test: Value + message: Value | None + + ValueInstruction: TypeAlias = ( Constant | Global @@ -186,8 +257,13 @@ class StoreLocal(Operands): | LoadLocal | Phi | Undefined + | BoolOp + | BuildList + | BuildTuple + | BuildDict + | WithEnter ) -EffectInstruction: TypeAlias = StoreLocal +EffectInstruction: TypeAlias = StoreLocal | SetAttr | SetItem | WithExit | Assert Instruction: TypeAlias = ValueInstruction | EffectInstruction diff --git a/src/coretrace_python/ir/printer.py b/src/coretrace_python/ir/printer.py index 06d695a..a302ddc 100644 --- a/src/coretrace_python/ir/printer.py +++ b/src/coretrace_python/ir/printer.py @@ -1,8 +1,13 @@ from __future__ import annotations from coretrace_python.ir.model import ( + Assert, BinaryOp, + BoolOp, Branch, + BuildDict, + BuildList, + BuildTuple, Call, Compare, Constant, @@ -18,12 +23,16 @@ Phi, Raise, Return, + SetAttr, + SetItem, StoreLocal, Symbol, Terminator, UnaryOp, Undefined, Value, + WithEnter, + WithExit, ) @@ -54,8 +63,34 @@ def _instruction(instruction: Instruction) -> str: f"{_value(instruction.left)}, {_value(instruction.right)}" ) if isinstance(instruction, Call): - arguments = ", ".join(_value(argument) for argument in instruction.arguments) - return f"{_value(instruction.result)} = call {_value(instruction.callee)}({arguments})" + parts = [_value(argument) for argument in instruction.arguments] + for name, value in instruction.keywords: + parts.append(f"{name}={_value(value)}" if name is not None else f"**{_value(value)}") + return f"{_value(instruction.result)} = call {_value(instruction.callee)}({', '.join(parts)})" + if isinstance(instruction, BoolOp): + values = ", ".join(_value(v) for v in instruction.values) + return f"{_value(instruction.result)} = bool_op.{instruction.operator} {values}" + if isinstance(instruction, BuildList | BuildTuple): + kind = "build_list" if isinstance(instruction, BuildList) else "build_tuple" + elements = ", ".join(_value(v) for v in instruction.elements) + return f"{_value(instruction.result)} = {kind} {elements}".rstrip() + if isinstance(instruction, BuildDict): + items = ", ".join(f"{_value(k)}: {_value(v)}" for k, v in instruction.items) + return f"{_value(instruction.result)} = build_dict {items}".rstrip() + if isinstance(instruction, WithEnter): + return f"{_value(instruction.result)} = with_enter {_value(instruction.context)}" + if isinstance(instruction, WithExit): + return f"with_exit {_value(instruction.context)}" + if isinstance(instruction, SetAttr): + return f"set_attr {_value(instruction.object)}, {instruction.attribute!r}, {_value(instruction.value)}" + if isinstance(instruction, SetItem): + return ( + f"set_item {_value(instruction.object)}, {_value(instruction.key)}, " + f"{_value(instruction.value)}" + ) + if isinstance(instruction, Assert): + message = "" if instruction.message is None else f", {_value(instruction.message)}" + return f"assert {_value(instruction.test)}{message}" if isinstance(instruction, GetAttr): return ( f"{_value(instruction.result)} = get_attr {_value(instruction.object)}, " diff --git a/src/coretrace_python/plugins/api.py b/src/coretrace_python/plugins/api.py index 60b2b4c..94f9e86 100644 --- a/src/coretrace_python/plugins/api.py +++ b/src/coretrace_python/plugins/api.py @@ -22,6 +22,7 @@ from coretrace_python.analysis.provider import R from coretrace_python.findings import Finding from coretrace_python.hir import nodes +from coretrace_python.ir.lowering import analyzable_functions from coretrace_python.taint import Model PLUGIN_API_VERSION = 1 @@ -50,16 +51,24 @@ def analyze(self, ctx: PluginContext) -> Sequence[Finding]: class PluginContext: """What one plugin may see while analysing one module.""" - def __init__(self, manager: AnalysisManager, plugin: Plugin) -> None: + def __init__( + self, + manager: AnalysisManager, + plugin: Plugin, + functions: tuple[nodes.Function, ...] | None = None, + ) -> None: self._manager = manager self._plugin = plugin + self._functions = analyzable_functions(manager.module) if functions is None else functions @property def module(self) -> nodes.Module: return self._manager.module def functions(self) -> tuple[nodes.Function, ...]: - return tuple(s for s in self.module.body if isinstance(s, nodes.Function)) + """Top-level functions and methods the engine can analyse.""" + + return self._functions @overload def get(self, analysis: type[Analysis[R]], function: None = None) -> R: ... @@ -79,10 +88,14 @@ def get(self, analysis: AnyAnalysis, function: nodes.Function | None = None) -> return self._manager.get(analysis, function) -def run_plugins(manager: AnalysisManager, plugins: Iterable[Plugin]) -> tuple[Finding, ...]: +def run_plugins( + manager: AnalysisManager, + plugins: Iterable[Plugin], + functions: tuple[nodes.Function, ...] | None = None, +) -> tuple[Finding, ...]: """Run every plugin against one shared manager and concatenate their findings.""" findings: list[Finding] = [] for plugin in plugins: - findings.extend(plugin.analyze(PluginContext(manager, plugin))) + findings.extend(plugin.analyze(PluginContext(manager, plugin, functions))) return tuple(findings) diff --git a/src/coretrace_python/semantic/imports.py b/src/coretrace_python/semantic/imports.py index c99c6b5..c251fbe 100644 --- a/src/coretrace_python/semantic/imports.py +++ b/src/coretrace_python/semantic/imports.py @@ -75,7 +75,7 @@ def body(self, statements: tuple[nodes.Statement, ...], scope_id: ScopeId) -> No elif isinstance(statement, nodes.If): self.body(statement.body, scope_id) self.body(statement.orelse, scope_id) - elif isinstance(statement, (nodes.While, nodes.For)): + elif isinstance(statement, (nodes.While, nodes.For, nodes.With)): self.body(statement.body, scope_id) def base_module(self, statement: nodes.ImportFrom) -> str: diff --git a/src/coretrace_python/semantic/scopes.py b/src/coretrace_python/semantic/scopes.py index ca7ec0c..2d2cb03 100644 --- a/src/coretrace_python/semantic/scopes.py +++ b/src/coretrace_python/semantic/scopes.py @@ -226,9 +226,21 @@ def body(self, statements: tuple[nodes.Statement, ...], scope: _ScopeBuilder) -> self.statement(statement, scope) def statement(self, node: nodes.Statement, scope: _ScopeBuilder) -> None: - if isinstance(node, nodes.Assign): + if isinstance(node, nodes.Assign | nodes.AugAssign): self.expression(node.value, scope) - scope.bind(node.target.identifier, BindingKind.LOCAL, node.target.span) + self.target(node.target, scope) + elif isinstance(node, nodes.Assert): + self.expression(node.test, scope) + if node.message is not None: + self.expression(node.message, scope) + elif isinstance(node, nodes.With): + for item in node.items: + self.expression(item.context, scope) + if item.target is not None: + scope.bind(item.target.identifier, BindingKind.LOCAL, item.target.span) + self.body(node.body, scope) + elif isinstance(node, nodes.EnterWith | nodes.ExitWith): + pass elif isinstance(node, nodes.Return): if node.value is not None: self.expression(node.value, scope) @@ -254,12 +266,19 @@ def statement(self, node: nodes.Statement, scope: _ScopeBuilder) -> None: scope.bind(name, BindingKind.NONLOCAL, node.span) scope.declarations.append((node, name)) elif isinstance(node, nodes.Function): + for decorator in node.decorators: + self.expression(decorator, scope) + for parameter in node.parameters: + if parameter.default is not None: + self.expression(parameter.default, scope) scope.bind(node.name, BindingKind.FUNCTION, node.span) function = self.open(scope, ScopeKind.FUNCTION, node.name, node.span) for parameter in node.parameters: function.bind(parameter.name, BindingKind.PARAMETER, parameter.span) self.body(node.body, function) elif isinstance(node, nodes.Class): + for decorator in node.decorators: + self.expression(decorator, scope) for base in node.bases: self.expression(base, scope) scope.bind(node.name, BindingKind.CLASS, node.span) @@ -302,9 +321,29 @@ def expression(self, node: nodes.Expression, scope: _ScopeBuilder) -> None: self.expression(keyword.value, scope) elif isinstance(node, nodes.Comprehension): self.comprehension(node, scope) + elif isinstance(node, nodes.BoolOp): + for value in node.values: + self.expression(value, scope) + elif isinstance(node, nodes.Tuple | nodes.List): + for element in node.elements: + self.expression(element, scope) + elif isinstance(node, nodes.Dict): + for key, value in node.items: + self.expression(key, scope) + self.expression(value, scope) else: raise TypeError(f"unknown expression: {node!r}") + def target(self, node: nodes.Target, scope: _ScopeBuilder) -> None: + if isinstance(node, nodes.Name): + scope.bind(node.identifier, BindingKind.LOCAL, node.span) + elif isinstance(node, nodes.Tuple): + for element in node.elements: + assert isinstance(element, nodes.Name | nodes.Attribute | nodes.Subscript | nodes.Tuple) + self.target(element, scope) + else: + self.expression(node, scope) + def comprehension(self, node: nodes.Comprehension, scope: _ScopeBuilder) -> None: # The first iterable is evaluated in the enclosing scope; everything else runs # inside the comprehension's own scope. diff --git a/src/coretrace_python/taint/engine.py b/src/coretrace_python/taint/engine.py index de1d649..de4ac88 100644 --- a/src/coretrace_python/taint/engine.py +++ b/src/coretrace_python/taint/engine.py @@ -161,12 +161,12 @@ def instruction(self, instruction: Instruction, state: Mapping[Value, Taint]) -> def call(self, call: Call, state: Mapping[Value, Taint], flows: list[TaintFlow]) -> Taint: symbol = self.symbols.get(call.callee) arguments = Taint.none() - for argument in call.arguments: + for argument in call.argument_values(): arguments = arguments.join(state.get(argument, Taint.none())) if symbol is not None: sink = self.models.sink(symbol) if sink is not None: - for argument in call.arguments: + for argument in call.argument_values(): taint = state.get(argument, Taint.none()) reaching = taint.kinds & sink.kinds if reaching: diff --git a/tests/test_syntax_coverage.py b/tests/test_syntax_coverage.py new file mode 100644 index 0000000..31ca5ac --- /dev/null +++ b/tests/test_syntax_coverage.py @@ -0,0 +1,307 @@ +"""Acceptance tests widening the PyHIR node set and the PyIR instruction set. + +``docs/architecture.md`` §3.2 PyHIR, §6 PyIR. Everyday syntax must reach the IR: keyword +arguments, parameter defaults and kinds, decorators, augmented assignment, attribute and +item stores, tuple unpacking, list/tuple/dict literals, ``and``/``or``, chained +comparisons, ``with``, ``assert``, module-level statements and methods of module-level +classes. A function that still uses unsupported syntax must not abort ``--check``: it is +reported as an ``unsupported-syntax`` note and the other functions are analysed. + +Expected to remain red until the nodes, instructions and tolerant check exist. +""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from coretrace_python import engine +from coretrace_python.cli import main +from coretrace_python.findings import Severity +from coretrace_python.frontend import HIRBuildError, build_hir +from coretrace_python.hir import nodes +from coretrace_python.semantic.scopes import ResolutionKind, analyze_scopes +from coretrace_python.source import SourceManager + +REPO = Path(__file__).resolve().parent.parent +PLUGINS = REPO / "plugins" + + +def build(source_text: str) -> nodes.Module: + return build_hir(SourceManager().add_source("cover.py", source_text)) + + +def emit(source_text: str, tmp_path: Path, capsys, *flags: str) -> str: # type: ignore[no-untyped-def] + path = tmp_path / "cover.py" + path.write_text(source_text, encoding="utf-8") + assert main(["--emit-ir", *flags, str(path)]) == 0, capsys.readouterr().err + return str(capsys.readouterr().out) + + +# --------------------------------------------------------------------------- PyHIR nodes + + +def test_parameters_carry_defaults_and_kinds() -> None: + module = build("def f(a, b=1, *args, c, d=2, **kwargs):\n pass\n") + function = module.body[0] + assert isinstance(function, nodes.Function) + + assert [(p.name, p.kind) for p in function.parameters] == [ + ("a", "positional"), + ("b", "positional"), + ("args", "var_positional"), + ("c", "keyword"), + ("d", "keyword"), + ("kwargs", "var_keyword"), + ] + assert function.parameters[0].default is None + assert isinstance(function.parameters[1].default, nodes.Constant) + assert isinstance(function.parameters[4].default, nodes.Constant) + + +def test_decorators_are_kept_on_functions_and_classes() -> None: + module = build("@app.route('/')\ndef index():\n pass\n\n@dataclass\nclass C:\n pass\n") + function, klass = module.body + assert isinstance(function, nodes.Function) and isinstance(klass, nodes.Class) + assert len(function.decorators) == 1 + assert isinstance(function.decorators[0], nodes.Call) + assert isinstance(klass.decorators[0], nodes.Name) + + +def test_augmented_assignment_and_store_targets() -> None: + module = build("def f(o, d, k):\n o.n += 1\n d[k] = o\n a, (b, c) = o\n") + function = module.body[0] + assert isinstance(function, nodes.Function) + aug, item, unpack = function.body + assert isinstance(aug, nodes.AugAssign) + assert aug.operator == "add" + assert isinstance(aug.target, nodes.Attribute) + assert isinstance(item, nodes.Assign) and isinstance(item.target, nodes.Subscript) + assert isinstance(unpack, nodes.Assign) and isinstance(unpack.target, nodes.Tuple) + assert isinstance(unpack.target.elements[1], nodes.Tuple) + + +def test_literals_boolean_operators_and_chained_comparisons() -> None: + module = build("def f(a, b, c):\n return [a], (a, b), {a: b}, a and b or c, a < b <= c\n") + function = module.body[0] + assert isinstance(function, nodes.Function) + returned = function.body[0] + assert isinstance(returned, nodes.Return) and isinstance(returned.value, nodes.Tuple) + lst, tup, dct, boolean, chain = returned.value.elements + assert isinstance(lst, nodes.List) and isinstance(tup, nodes.Tuple) and isinstance(dct, nodes.Dict) + assert dct.items[0][1].identifier == "b" # type: ignore[union-attr] + assert isinstance(boolean, nodes.BoolOp) and boolean.operator == "or" + assert isinstance(boolean.values[0], nodes.BoolOp) and boolean.values[0].operator == "and" + assert isinstance(chain, nodes.BoolOp) and chain.operator == "and" + assert [c.operator for c in chain.values] == ["lt", "lt_eq"] # type: ignore[union-attr] + + +def test_with_and_assert_statements() -> None: + module = build("def f(p):\n with open(p) as fh, lock:\n assert fh, 'open'\n") + function = module.body[0] + assert isinstance(function, nodes.Function) + with_statement = function.body[0] + assert isinstance(with_statement, nodes.With) + first, second = with_statement.items + assert isinstance(first.context, nodes.Call) and first.target is not None + assert first.target.identifier == "fh" + assert second.target is None + assertion = with_statement.body[0] + assert isinstance(assertion, nodes.Assert) + assert isinstance(assertion.message, nodes.Constant) + + +def test_keyword_arguments_including_unpacking() -> None: + module = build("def f(g, k):\n g(1, key=2, **k)\n") + function = module.body[0] + assert isinstance(function, nodes.Function) + call = function.body[0] + assert isinstance(call, nodes.ExpressionStatement) and isinstance(call.expression, nodes.Call) + assert [k.name for k in call.expression.keywords] == ["key", None] + + +def test_star_arguments_are_still_rejected_explicitly() -> None: + with pytest.raises(HIRBuildError, match="star arguments"): + build("def f(g, a):\n g(*a)\n") + + +def test_pyhir_declares_a_schema_version() -> None: + from coretrace_python.hir import HIR_SCHEMA_VERSION + + assert HIR_SCHEMA_VERSION >= 1 + + +# --------------------------------------------------------------------------- scopes + + +def test_defaults_and_decorators_resolve_in_the_enclosing_scope() -> None: + module = build("LIMIT = 1\n\n@register\ndef f(a, b=LIMIT):\n return b\n") + scopes = analyze_scopes(module) + f = next(s for s in scopes.children(scopes.module_scope.id) if s.name == "f") + + assert "LIMIT" not in f.bindings + assert f.bindings["b"].kind.name == "PARAMETER" + assert scopes.resolve(scopes.module_scope.id, "LIMIT").kind is ResolutionKind.GLOBAL + + +def test_with_targets_and_unpacked_names_are_locals() -> None: + module = build("def f(x):\n with x as y:\n a, b = y\n return a\n") + scopes = analyze_scopes(module) + f = next(s for s in scopes.children(scopes.module_scope.id) if s.name == "f") + + assert {name for name in ("y", "a", "b")} <= set(f.bindings) + + +# --------------------------------------------------------------------------- PyIR + + +def test_emit_ir_keyword_arguments(tmp_path, capsys) -> None: # type: ignore[no-untyped-def] + output = emit("def f(g, a, k):\n g(a, key=1, **k)\n", tmp_path, capsys) + assert " %3 = const 1\n %4 = call %0(%1, key=%3, **%2)\n" in output + + +def test_emit_ir_augmented_and_indexed_stores(tmp_path, capsys) -> None: # type: ignore[no-untyped-def] + output = emit("def f(o, d, k):\n o.n += 1\n d[k] = o\n", tmp_path, capsys) + assert output == ( + "func @f(%0, %1, %2) {\n" + "entry:\n" + " %3 = get_attr %0, 'n'\n" + " %4 = const 1\n" + " %5 = binary.add %3, %4\n" + " set_attr %0, 'n', %5\n" + " set_item %1, %2, %0\n" + " return\n" + "}\n" + ) + + +def test_emit_ir_unpacking_and_literals(tmp_path, capsys) -> None: # type: ignore[no-untyped-def] + output = emit("def f(x):\n a, b = x\n return [a, b], (a,), {a: b}\n", tmp_path, capsys) + assert " %1 = const 0\n %2 = get_item %0, %1\n store_local \"a\", %2\n" in output + assert " %3 = const 1\n %4 = get_item %0, %3\n store_local \"b\", %4\n" in output + assert "build_list %5, %6\n" in output + assert "build_tuple %8\n" in output + assert "build_dict %10: %11\n" in output + assert output.rstrip().endswith(" return %13\n}") + + +def test_emit_ir_boolean_operators_and_chains(tmp_path, capsys) -> None: # type: ignore[no-untyped-def] + output = emit("def f(a, b, c):\n return a < b < c and a\n", tmp_path, capsys) + assert output == ( + "func @f(%0, %1, %2) {\n" + "entry:\n" + " %3 = compare.lt %0, %1\n" + " %4 = compare.lt %1, %2\n" + " %5 = bool_op.and %3, %4\n" + " %6 = bool_op.and %5, %0\n" + " return %6\n" + "}\n" + ) + + +def test_emit_ir_with_statement(tmp_path, capsys) -> None: # type: ignore[no-untyped-def] + output = emit( + "def f(p):\n with open(p) as fh:\n data = fh.read()\n return data\n", tmp_path, capsys + ) + assert output == ( + "func @f(%0) {\n" + "entry:\n" + " %1 = symbol @python.builtins.open\n" + " %2 = call %1(%0)\n" + " %3 = with_enter %2\n" + ' store_local "fh", %3\n' + ' %4 = load_local "fh"\n' + " %5 = get_attr %4, 'read'\n" + " %6 = call %5()\n" + ' store_local "data", %6\n' + " with_exit %2\n" + ' %7 = load_local "data"\n' + " return %7\n" + "}\n" + ) + + +def test_emit_ir_early_return_inside_with(tmp_path, capsys) -> None: # type: ignore[no-untyped-def] + output = emit("def f(x):\n with x:\n if x:\n return 1\n return 0\n", tmp_path, capsys) + assert "with_enter %0" in output + assert "then_1:\n %2 = const 1\n return %2\n" in output + assert "merge_1:\n with_exit %0\n" in output + + +def test_emit_ir_assert_evaluates_its_expressions(tmp_path, capsys) -> None: # type: ignore[no-untyped-def] + output = emit("def f(x):\n assert x, 'm'\n return x\n", tmp_path, capsys) + assert output == "func @f(%0) {\nentry:\n %1 = const 'm'\n assert %0, %1\n return %0\n}\n" + + +def test_emit_ir_defaults_and_decorators_do_not_change_the_function(tmp_path, capsys) -> None: # type: ignore[no-untyped-def] + output = emit("@deco\ndef f(a, b=1, *, c=2):\n return c\n", tmp_path, capsys) + assert output == "func @f(%0, %1, %2) {\nentry:\n return %2\n}\n" + + +def test_emit_ir_module_level_code_and_methods(tmp_path, capsys) -> None: # type: ignore[no-untyped-def] + output = emit( + "LIMIT = 3\n\nclass Store:\n size = 0\n\n def get(self, key):\n return self.data[key]\n\n" + "def top():\n return LIMIT\n", + tmp_path, + capsys, + ) + assert output.startswith("func @Store.get(%0, %1) {\n") + assert "func @top() {\n" in output + + +def test_emit_ssa_handles_the_new_instructions(tmp_path, capsys) -> None: # type: ignore[no-untyped-def] + output = emit("def f(o, k):\n o.n += k\n xs = [o, k]\n return xs\n", tmp_path, capsys, "--ssa") + assert "set_attr %0, 'n', %3\n" in output + assert "%4 = build_list %0, %1\n return %4\n" in output + + +# --------------------------------------------------------------------------- tolerant check + + +def test_check_reports_unsupported_functions_and_keeps_going(tmp_path, capsys) -> None: # type: ignore[no-untyped-def] + source = tmp_path / "mixed.py" + source.write_text( + "CONFIG = {}\n\n" + "def outer():\n" + " def inner():\n" + " return 1\n" + " return inner\n\n" + "class Runner:\n" + " def go(self, code):\n" + " eval(code)\n", + encoding="utf-8", + ) + + exit_code = main(["--check", str(source), "--plugins", str(PLUGINS)]) + output = capsys.readouterr().out + + assert exit_code == 1 + assert f"{source}:3:1: info unsupported-syntax: " in output + assert "Function" in output + assert f"{source}:10:9: high dangerous-eval:" in output + assert output.endswith("2 findings\n") + + +def test_unsupported_syntax_findings_are_notes() -> None: + findings = engine.check(SourceManager().add_source("n.py", "def f():\n def g():\n pass\n"), [PLUGINS]) + + assert [f.rule_id for f in findings] == ["unsupported-syntax"] + assert findings[0].severity is Severity.INFO + assert findings[0].function == "f" + + +def test_taint_flows_through_list_literals() -> None: + findings = engine.check( + SourceManager().add_source("cmd.py", "import subprocess\n\ndef run():\n subprocess.run(['sh', '-c', input()], check=True)\n"), + [PLUGINS], + ) + assert [f.rule_id for f in findings] == ["command-injection"] + + +def test_taint_flows_through_keyword_arguments() -> None: + findings = engine.check( + SourceManager().add_source("cmd.py", "import subprocess\n\ndef run():\n subprocess.run('ls', input=input())\n"), + [PLUGINS], + ) + assert [f.rule_id for f in findings] == ["command-injection"]