diff --git a/README.md b/README.md index 0328981..b353594 100644 --- a/README.md +++ b/README.md @@ -137,9 +137,12 @@ PYTHONPATH=src python -m coretrace_python --emit-ir example.py 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`, `try`/`except`/`else`/`finally`, `await`, `yield`, -`if`/`elif`/`else`, `while`, `for`, `break`, `continue` and `raise`. Blocks inside a `try` +assignment, list, tuple and dict literals with `*` and `**` unpacking, f-strings, slices, +`and`/`or`, chained comparisons, keyword and starred arguments, `with`, `assert`, +`try`/`except`/`else`/`finally`, `await`, `yield`, `if`/`elif`/`else`, `while`, `for` +with name or tuple targets, `break`, `continue` and `raise`, `from` clause included. Not +yet: conditional expressions, lambdas, comprehensions, nested function definitions and +assignments to `global` names. Blocks inside a `try` body carry exception edges to the handlers; `finally` is modelled on the normal path only. Methods of module-level classes are analysed like functions; other module-level code is skipped. An import inside a function is shown where it runs, as an `import` instruction naming the module, the bound name and the canonical symbol. A function using syntax outside this subset is reported by `--check` as diff --git a/src/coretrace_python/abstract/heap.py b/src/coretrace_python/abstract/heap.py index dc56238..e6b52ab 100644 --- a/src/coretrace_python/abstract/heap.py +++ b/src/coretrace_python/abstract/heap.py @@ -195,12 +195,20 @@ def instruction(self, instruction: Instruction) -> bool: }[type(instruction)] site = AbstractObject(AllocationSite(kind, instruction.location)) changed = self.assign(result, {site}) - if isinstance(instruction, BuildList | BuildTuple): - for element in instruction.elements: + if isinstance(instruction, BuildList | BuildTuple | BuildDict): + values = ( + instruction.elements + if isinstance(instruction, BuildList | BuildTuple) + else tuple(v for _, v in instruction.items) + ) + for element in values: changed |= self.store({site}, ELEMENTS, element) - elif isinstance(instruction, BuildDict): - for _, value in instruction.items: - changed |= self.store({site}, ELEMENTS, value) + for unpacked in instruction.unpacked: + contents = self.at(self.of(unpacked), ELEMENTS) + location = self.heap.setdefault(HeapLocation(site, ELEMENTS), set()) + before = len(location) + location |= contents + changed |= len(location) != before elif isinstance(instruction, Call): receiver = mutated_by(instruction, self.defs) if receiver is not None: diff --git a/src/coretrace_python/cfg/builder.py b/src/coretrace_python/cfg/builder.py index 274c069..674bb1a 100644 --- a/src/coretrace_python/cfg/builder.py +++ b/src/coretrace_python/cfg/builder.py @@ -96,7 +96,7 @@ def statement( self.finish(block, Return(node.value, node.span)) return None if isinstance(node, nodes.Raise): - self.finish(block, Raise(node.exception, node.span)) + self.finish(block, Raise(node.exception, node.span, node.cause)) return None if isinstance(node, nodes.Break): self.finish(block, Jump(self.loop("break", node.span)[1], node.span)) diff --git a/src/coretrace_python/cfg/model.py b/src/coretrace_python/cfg/model.py index 573d237..8d61123 100644 --- a/src/coretrace_python/cfg/model.py +++ b/src/coretrace_python/cfg/model.py @@ -52,6 +52,7 @@ class Return: class Raise: exception: nodes.Expression | None span: SourceSpan + cause: nodes.Expression | None = None @dataclass(frozen=True) diff --git a/src/coretrace_python/dependency/graph.py b/src/coretrace_python/dependency/graph.py index 963ce57..88fde39 100644 --- a/src/coretrace_python/dependency/graph.py +++ b/src/coretrace_python/dependency/graph.py @@ -13,6 +13,7 @@ import tomllib from collections.abc import Mapping from dataclasses import dataclass +from pathlib import PurePath from types import MappingProxyType from typing import Any, ClassVar @@ -179,7 +180,7 @@ def merge(self, other: DependencyGraph) -> DependencyGraph: def parse_dependencies(source: SourceFile) -> DependencyGraph: """The requirements declared or pinned by one dependency file; other files are empty.""" - name = source.path.name if source.path is not None else str(source.source_id) + name = source.path.name if source.path is not None else PurePath(str(source.source_id)).name if name.startswith("requirements") and name.endswith(".txt"): return _parse_requirements_txt(source) if name == "pyproject.toml": diff --git a/src/coretrace_python/engine.py b/src/coretrace_python/engine.py index 84ae733..efd413c 100644 --- a/src/coretrace_python/engine.py +++ b/src/coretrace_python/engine.py @@ -215,11 +215,27 @@ def resolve_dependencies(root: Path, sources: SourceManager) -> DependencyGraph: graph = DependencyGraph() candidates = sorted(root.glob("requirements*.txt")) + [root / name for name in DEPENDENCY_FILES] for path in candidates: - if path.is_file(): - graph = graph.merge(parse_dependencies(sources.load_file(path))) + if not path.is_file(): + continue + try: + text = _decode_manifest(path.read_bytes()) + except (OSError, UnicodeDecodeError) as error: + graph = graph.merge(DependencyGraph(errors=(f"{path}: {error}",))) + continue + graph = graph.merge(parse_dependencies(sources.add_source(str(path), text))) return graph +def _decode_manifest(data: bytes) -> str: + """Dependency files written by ``pip freeze`` on Windows are UTF-16 with a byte + order mark; honour the mark, otherwise expect UTF-8.""" + + for mark, encoding in ((b"\xef\xbb\xbf", "utf-8-sig"), (b"\xff\xfe", "utf-16"), (b"\xfe\xff", "utf-16")): + if data.startswith(mark): + return data.decode(encoding) + return data.decode("utf-8") + + def analyze_project( root: Path, plugin_roots: Sequence[Path] = (), @@ -261,7 +277,11 @@ def analyze_project( findings.append(_note("syntax-error", str(error), sources.add_source(str(policy_path), ""), 1)) modules: dict[str, nodes.Module] = {} files: dict[str, SourceFile] = {} - for source in discover_sources(root, sources): + unreadable: list[tuple[Path, str]] = [] + discovered = discover_sources(root, sources, unreadable) + for path, reason in unreadable: + findings.append(_note("syntax-error", f"{path}: {reason}", sources.add_source(str(path), ""), 1)) + for source in discovered: try: modules[source.module_name] = build_hir(source) except (ParseError, HIRBuildError) as error: diff --git a/src/coretrace_python/frontend/ast_adapter.py b/src/coretrace_python/frontend/ast_adapter.py index 9a35223..3c81e17 100644 --- a/src/coretrace_python/frontend/ast_adapter.py +++ b/src/coretrace_python/frontend/ast_adapter.py @@ -103,20 +103,26 @@ def expression(self, node: ast.expr) -> nodes.Expression: if isinstance(node, ast.List): return nodes.List(self.elements(node.elts), span) if isinstance(node, ast.Dict): - items = [] + items: list[tuple[nodes.Expression | None, nodes.Expression]] = [] 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))) + items.append((None if key is None else self.expression(key), self.expression(value))) return nodes.Dict(tuple(items), span) + if isinstance(node, ast.JoinedStr): + return nodes.FormattedString(tuple(self.formatted_parts(node)), span) + if isinstance(node, ast.Slice): + return nodes.Slice( + self.expression(node.lower) if node.lower is not None else None, + self.expression(node.upper) if node.upper is not None else None, + self.expression(node.step) if node.step is not None else None, + span, + ) + if isinstance(node, ast.Starred): + return nodes.Starred(self.expression(node.value), 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)) @@ -138,11 +144,21 @@ 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 formatted_parts(self, node: ast.JoinedStr) -> list[nodes.Expression]: + parts: list[nodes.Expression] = [] + for value in node.values: + if isinstance(value, ast.Constant): + parts.append(nodes.Constant(value.value, self.span(value))) + elif isinstance(value, ast.FormattedValue): + parts.append(self.expression(value.value)) + if isinstance(value.format_spec, ast.JoinedStr): + parts.extend(p for p in self.formatted_parts(value.format_spec) if not isinstance(p, nodes.Constant)) + else: # pragma: no cover - the grammar allows nothing else + self.fail(value) + return parts + def target(self, node: ast.expr) -> nodes.Target: if isinstance(node, ast.Name): return nodes.Name(node.id, self.span(node)) @@ -241,13 +257,18 @@ def statement(self, node: ast.stmt) -> nodes.Statement: if isinstance(node, (ast.For, ast.AsyncFor)): if node.orelse: self.fail(node.orelse[0], "loop else clauses are not supported yet") - if not isinstance(node.target, ast.Name): - self.fail(node.target, "only a single name is supported as a loop target") - target = nodes.Name(node.target.id, self.span(node.target)) + body = self.block(node.body) + if isinstance(node.target, ast.Name): + target = nodes.Name(node.target.id, self.span(node.target)) + else: + # ``for k, v in items`` binds a hidden local and destructures it first. + target_span = self.span(node.target) + target = nodes.Name(f"_coretrace_item_{target_span.start_line}_{target_span.start_column}", target_span) + body = (nodes.Assign(self.target(node.target), target, target_span), *body) return nodes.For( target, self.expression(node.iter), - self.block(node.body), + body, isinstance(node, ast.AsyncFor), span, ) @@ -256,10 +277,9 @@ def statement(self, node: ast.stmt) -> nodes.Statement: if isinstance(node, ast.Continue): return nodes.Continue(span) if isinstance(node, ast.Raise): - if node.cause is not None: - self.fail(node.cause, "raise from is not supported yet") exception = self.expression(node.exc) if node.exc is not None else None - return nodes.Raise(exception, span) + cause = self.expression(node.cause) if node.cause is not None else None + return nodes.Raise(exception, span, cause) if isinstance(node, ast.Try): handlers = [] for handler in node.handlers: diff --git a/src/coretrace_python/hir/nodes.py b/src/coretrace_python/hir/nodes.py index f1aac2b..51a32aa 100644 --- a/src/coretrace_python/hir/nodes.py +++ b/src/coretrace_python/hir/nodes.py @@ -99,7 +99,34 @@ class List: @dataclass(frozen=True, slots=True) class Dict: - items: tuple[tuple[Expression, Expression], ...] + """Items are ``(key, value)``; a ``None`` key is a ``**mapping`` unpacking.""" + + items: tuple[tuple[Expression | None, Expression], ...] + span: SourceSpan + + +@dataclass(frozen=True, slots=True) +class FormattedString: + """An f-string: constant text and formatted expressions, in order. Conversions and + format specifications are dropped; expressions nested in a specification are parts.""" + + parts: tuple[Expression, ...] + span: SourceSpan + + +@dataclass(frozen=True, slots=True) +class Slice: + lower: Expression | None + upper: Expression | None + step: Expression | None + span: SourceSpan + + +@dataclass(frozen=True, slots=True) +class Starred: + """``*iterable`` in a call argument list or a list or tuple display.""" + + value: Expression span: SourceSpan @@ -148,6 +175,9 @@ class Comprehension: | Tuple | List | Dict + | FormattedString + | Slice + | Starred | Await | Yield | Comprehension @@ -245,6 +275,7 @@ class Continue: class Raise: exception: Expression | None span: SourceSpan + cause: Expression | None = None @dataclass(frozen=True, slots=True) diff --git a/src/coretrace_python/interprocedural/modulegraph.py b/src/coretrace_python/interprocedural/modulegraph.py index e812c54..ee357ff 100644 --- a/src/coretrace_python/interprocedural/modulegraph.py +++ b/src/coretrace_python/interprocedural/modulegraph.py @@ -24,10 +24,13 @@ IGNORED_DIRECTORIES = frozenset({"__pycache__", "node_modules", "venv", "build", "dist", "site-packages"}) -def discover_sources(root: Path, manager: SourceManager) -> tuple[SourceFile, ...]: +def discover_sources( + root: Path, manager: SourceManager, errors: list[tuple[Path, str]] | None = None +) -> tuple[SourceFile, ...]: """Load every ``.py`` file under ``root``, skipping hidden and tooling directories and virtual environments, recognised by the ``pyvenv.cfg`` at their root whatever their - name.""" + name. A file that cannot be read or decoded is skipped and reported in ``errors``: + Python could not import it either.""" found: list[SourceFile] = [] environments: dict[Path, bool] = {} @@ -37,7 +40,12 @@ def discover_sources(root: Path, manager: SourceManager) -> tuple[SourceFile, .. continue if any(_is_environment(root / Path(*relative.parts[:depth]), environments) for depth in range(1, len(relative.parts))): continue - found.append(manager.load_file(path)) + try: + found.append(manager.load_file(path)) + except (OSError, UnicodeDecodeError) as error: + if errors is None: + raise + errors.append((path, str(error))) return tuple(found) diff --git a/src/coretrace_python/interprocedural/summaries.py b/src/coretrace_python/interprocedural/summaries.py index 41f096f..d4b7a56 100644 --- a/src/coretrace_python/interprocedural/summaries.py +++ b/src/coretrace_python/interprocedural/summaries.py @@ -35,6 +35,9 @@ from coretrace_python.ir.model import ( BasicBlock, Branch, + BuildDict, + BuildList, + BuildTuple, Call, Constant, ForNext, @@ -255,13 +258,15 @@ def flow(self, cfg: CFG, block_id: BlockId, incoming: Mapping[BlockId, State]) - # ------------------------------------------------------------------ transfer - @staticmethod - def instruction(instruction: Instruction, state: Mapping[Key, Dep]) -> Dep: + def instruction(self, instruction: Instruction, state: Mapping[Key, Dep]) -> Dep: if isinstance(instruction, Constant | Global | Symbol | Undefined): return EMPTY deps = EMPTY for operand in instruction.operands(): deps |= state.get(operand, EMPTY) + if isinstance(instruction, BuildList | BuildTuple | BuildDict): + for unpacked in instruction.unpacked: + deps |= self.deep(unpacked, state) return deps def loaded(self, instruction: GetAttr | GetItem | GetIter, state: Mapping[Key, Dep]) -> Dep: @@ -281,7 +286,7 @@ def loaded(self, instruction: GetAttr | GetItem | GetIter, state: Mapping[Key, D def call(self, call: Call, state: dict[Key, Dep]) -> Dep: arguments = tuple(self.deep(a, state) for a in call.arguments) keywords = EMPTY - for _, value in call.keywords: + for value in (*call.starred, *(v for _, v in call.keywords)): keywords |= self.deep(value, state) everything = keywords for deps in arguments: diff --git a/src/coretrace_python/ir/lowering.py b/src/coretrace_python/ir/lowering.py index aa2941a..23256cc 100644 --- a/src/coretrace_python/ir/lowering.py +++ b/src/coretrace_python/ir/lowering.py @@ -25,6 +25,8 @@ Branch, BuildDict, BuildList, + BuildSlice, + BuildString, BuildTuple, Call, Catch, @@ -146,21 +148,28 @@ def expression(self, node: nodes.Expression) -> Value: return self.emit(Compare(self.new_value(), node.span, node.operator, left, right)) if isinstance(node, nodes.Call): callee = self.expression(node.callee) - arguments = tuple(self.expression(argument) for argument in node.arguments) + arguments, starred = self.spread(node.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)) + return self.emit(Call(self.new_value(), node.span, callee, arguments, keywords, starred)) 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.List | nodes.Tuple): + elements, unpacked = self.spread(node.elements) + builder = BuildList if isinstance(node, nodes.List) else BuildTuple + return self.emit(builder(self.new_value(), node.span, elements, unpacked)) 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)) + items = tuple((self.expression(k), self.expression(v)) for k, v in node.items if k is not None) + unpacked = tuple(self.expression(v) for k, v in node.items if k is None) + return self.emit(BuildDict(self.new_value(), node.span, items, unpacked)) + if isinstance(node, nodes.FormattedString): + parts = tuple(self.expression(part) for part in node.parts) + return self.emit(BuildString(self.new_value(), node.span, parts)) + if isinstance(node, nodes.Slice): + bounds = [self.expression(b) if b is not None else None for b in (node.lower, node.upper, node.step)] + return self.emit(BuildSlice(self.new_value(), node.span, bounds[0], bounds[1], bounds[2])) + if isinstance(node, nodes.Starred): + self.fail(node, "a starred expression is only supported in calls, lists and tuples") if isinstance(node, nodes.Await): return self.emit(Await(self.new_value(), node.span, self.expression(node.value))) if isinstance(node, nodes.Yield): @@ -175,6 +184,18 @@ def expression(self, node: nodes.Expression) -> Value: return self.emit(GetItem(self.new_value(), node.span, object_value, key)) self.fail(node) + def spread(self, elements: tuple[nodes.Expression, ...]) -> tuple[tuple[Value, ...], tuple[Value, ...]]: + """Plain elements and ``*iterable`` elements of a display or an argument list.""" + + plain: list[Value] = [] + unpacked: list[Value] = [] + for element in elements: + if isinstance(element, nodes.Starred): + unpacked.append(self.expression(element.value)) + else: + plain.append(self.expression(element)) + return tuple(plain), tuple(unpacked) + # ------------------------------------------------------------------ statements def store(self, target: nodes.Target, value: Value) -> None: @@ -262,7 +283,8 @@ def terminator(self, block: control_flow.BasicBlock) -> Terminator: exception = ( self.expression(terminator.exception) if terminator.exception is not None else None ) - return Raise(terminator.span, exception) + cause = self.expression(terminator.cause) if terminator.cause is not None else None + return Raise(terminator.span, exception, cause) if isinstance(terminator, control_flow.Jump): self.enter_loop(block.id, terminator.target) return Jump(terminator.span, terminator.target) diff --git a/src/coretrace_python/ir/model.py b/src/coretrace_python/ir/model.py index df29ded..f212003 100644 --- a/src/coretrace_python/ir/model.py +++ b/src/coretrace_python/ir/model.py @@ -110,14 +110,18 @@ class Compare(Operands): @dataclass(frozen=True) class Call(Operands): + """``starred`` are ``*iterable`` arguments whose positions are unknown; keywords + with a ``None`` name are ``**mapping`` arguments.""" + result: Value location: SourceSpan callee: Value arguments: tuple[Value, ...] keywords: tuple[tuple[str | None, Value], ...] = () + starred: tuple[Value, ...] = () def argument_values(self) -> tuple[Value, ...]: - return (*self.arguments, *(value for _, value in self.keywords)) + return (*self.arguments, *self.starred, *(value for _, value in self.keywords)) @dataclass(frozen=True) @@ -133,6 +137,7 @@ class BuildList(Operands): result: Value location: SourceSpan elements: tuple[Value, ...] + unpacked: tuple[Value, ...] = () @dataclass(frozen=True) @@ -140,6 +145,7 @@ class BuildTuple(Operands): result: Value location: SourceSpan elements: tuple[Value, ...] + unpacked: tuple[Value, ...] = () @dataclass(frozen=True) @@ -147,6 +153,25 @@ class BuildDict(Operands): result: Value location: SourceSpan items: tuple[tuple[Value, Value], ...] + unpacked: tuple[Value, ...] = () + + +@dataclass(frozen=True) +class BuildString(Operands): + """An f-string: its constant and formatted parts, in order.""" + + result: Value + location: SourceSpan + parts: tuple[Value, ...] + + +@dataclass(frozen=True) +class BuildSlice(Operands): + result: Value + location: SourceSpan + lower: Value | None + upper: Value | None + step: Value | None @dataclass(frozen=True) @@ -299,6 +324,8 @@ class Import(Operands): | BuildList | BuildTuple | BuildDict + | BuildString + | BuildSlice | WithEnter | Catch | Await @@ -335,6 +362,7 @@ class Jump(Operands): class Raise(Operands): location: SourceSpan exception: Value | None + cause: Value | None = None @dataclass(frozen=True) diff --git a/src/coretrace_python/ir/printer.py b/src/coretrace_python/ir/printer.py index a8c6e70..b3e0f45 100644 --- a/src/coretrace_python/ir/printer.py +++ b/src/coretrace_python/ir/printer.py @@ -8,6 +8,8 @@ Branch, BuildDict, BuildList, + BuildSlice, + BuildString, BuildTuple, Call, Catch, @@ -68,6 +70,7 @@ def _instruction(instruction: Instruction) -> str: ) if isinstance(instruction, Call): parts = [_value(argument) for argument in instruction.arguments] + parts.extend(f"*{_value(value)}" for value in instruction.starred) 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)})" @@ -76,11 +79,17 @@ def _instruction(instruction: Instruction) -> str: 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() + parts = [_value(v) for v in instruction.elements] + [f"*{_value(v)}" for v in instruction.unpacked] + return f"{_value(instruction.result)} = {kind} {', '.join(parts)}".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() + parts = [f"{_value(k)}: {_value(v)}" for k, v in instruction.items] + parts.extend(f"**{_value(v)}" for v in instruction.unpacked) + return f"{_value(instruction.result)} = build_dict {', '.join(parts)}".rstrip() + if isinstance(instruction, BuildString): + return f"{_value(instruction.result)} = build_string {', '.join(_value(v) for v in instruction.parts)}".rstrip() + if isinstance(instruction, BuildSlice): + bounds = ", ".join("none" if b is None else _value(b) for b in (instruction.lower, instruction.upper, instruction.step)) + return f"{_value(instruction.result)} = build_slice {bounds}" if isinstance(instruction, WithEnter): return f"{_value(instruction.result)} = with_enter {_value(instruction.context)}" if isinstance(instruction, WithExit): @@ -140,7 +149,10 @@ def _terminator(terminator: Terminator) -> str: if isinstance(terminator, Jump): return f"jump {terminator.target}" if isinstance(terminator, Raise): - return "raise" if terminator.exception is None else f"raise {_value(terminator.exception)}" + if terminator.exception is None: + return "raise" + text = f"raise {_value(terminator.exception)}" + return text if terminator.cause is None else f"{text} from {_value(terminator.cause)}" if isinstance(terminator, ForNext): prefix = "" if terminator.result is None else f"{_value(terminator.result)} = " return ( diff --git a/src/coretrace_python/plugins/secrets.py b/src/coretrace_python/plugins/secrets.py index 193036a..2d73cef 100644 --- a/src/coretrace_python/plugins/secrets.py +++ b/src/coretrace_python/plugins/secrets.py @@ -79,7 +79,8 @@ def _walk(node: Node, name: str | None, function: str | None) -> Iterator[Litera bound = key.value if isinstance(key.value, str) else None else: bound = None - yield from _walk(key, None, function) + if key is not None: + yield from _walk(key, None, function) yield from _walk(value, bound, function) return for child in children(node): diff --git a/src/coretrace_python/semantic/scopes.py b/src/coretrace_python/semantic/scopes.py index da3bc1f..907ed8a 100644 --- a/src/coretrace_python/semantic/scopes.py +++ b/src/coretrace_python/semantic/scopes.py @@ -307,6 +307,8 @@ def statement(self, node: nodes.Statement, scope: _ScopeBuilder) -> None: elif isinstance(node, nodes.Raise): if node.exception is not None: self.expression(node.exception, scope) + if node.cause is not None: + self.expression(node.cause, scope) elif not isinstance(node, nodes.Pass | nodes.Break | nodes.Continue): raise TypeError(f"unknown statement: {node!r}") @@ -344,8 +346,18 @@ def expression(self, node: nodes.Expression, scope: _ScopeBuilder) -> None: self.expression(element, scope) elif isinstance(node, nodes.Dict): for key, value in node.items: - self.expression(key, scope) + if key is not None: + self.expression(key, scope) self.expression(value, scope) + elif isinstance(node, nodes.FormattedString): + for part in node.parts: + self.expression(part, scope) + elif isinstance(node, nodes.Slice): + for bound in (node.lower, node.upper, node.step): + if bound is not None: + self.expression(bound, scope) + elif isinstance(node, nodes.Starred): + self.expression(node.value, scope) else: raise TypeError(f"unknown expression: {node!r}") diff --git a/src/coretrace_python/taint/engine.py b/src/coretrace_python/taint/engine.py index f366f31..26132ce 100644 --- a/src/coretrace_python/taint/engine.py +++ b/src/coretrace_python/taint/engine.py @@ -40,6 +40,9 @@ from coretrace_python.ir.model import ( BasicBlock, Branch, + BuildDict, + BuildList, + BuildTuple, Call, Compare, ForNext, @@ -251,12 +254,16 @@ def instruction(self, instruction: Instruction, state: Mapping[Key, Taint]) -> T return taint for operand in instruction.operands(): taint = taint.join(state.get(operand, Taint.none())) + if isinstance(instruction, BuildList | BuildTuple | BuildDict): + # ``[*xs]`` and ``{**d}`` copy the contents of what they unpack. + for unpacked in instruction.unpacked: + taint = taint.join(self.deep(unpacked, state)) return taint def call(self, call: Call, state: dict[Key, Taint], flows: list[TaintFlow]) -> Taint: arguments = tuple(self.deep(a, state) for a in call.arguments) keywords = Taint.none() - for _, value in call.keywords: + for value in (*call.starred, *(v for _, v in call.keywords)): keywords = keywords.join(self.deep(value, state)) everything = keywords for taint in arguments: @@ -305,12 +312,15 @@ def known( if summary.unsupported: return everything + spread = (*call.starred, *(value for _, value in call.keywords)) + def mapped(deps: frozenset[int]) -> tuple[Taint, Value | None]: taint, witness = Taint.none(), None for index in sorted(deps): - part = arguments[index] if index < len(arguments) else keywords - if part and witness is None: - witness = call.arguments[index] if index < len(arguments) else call.keywords[0][1] + positional = index < len(arguments) + part = arguments[index] if positional else keywords + if part and witness is None and (positional or spread): + witness = call.arguments[index] if positional else spread[0] taint = taint.join(part) return taint, witness diff --git a/tests/test_discovery.py b/tests/test_discovery.py index daf72dc..ca4e65e 100644 --- a/tests/test_discovery.py +++ b/tests/test_discovery.py @@ -71,3 +71,26 @@ def test_dashed_packages_are_analysed_instead_of_crashing(tmp_path: Path) -> Non ("command-injection", "__main__.py", 4) ] assert SymbolId("python.network_topologer.traceroute.probe") in analysis.index.symbols + + +def test_undecodable_files_become_notes_and_the_rest_is_analysed(tmp_path: Path) -> None: + root = project(tmp_path, {"app.py": "import os\n\ndef run():\n os.system(input())\n"}) + (root / "legacy.py").write_bytes(b"\xff\xfeprint('utf-16')\n") + + analysis = engine.analyze_project(root, [Path(__file__).resolve().parent.parent / "plugins"]) + + notes = [(f.rule_id, Path(str(f.span.source_id)).name) for f in analysis.findings] + assert ("syntax-error", "legacy.py") in notes + assert ("command-injection", "app.py") in notes + assert set(analysis.keys) == {"app"} + assert any("utf-8" in f.message for f in analysis.findings if f.rule_id == "syntax-error") + + +def test_utf16_dependency_files_are_read(tmp_path: Path) -> None: + root = project(tmp_path, {"app.py": "x = 1\n"}) + (root / "requirements.txt").write_bytes("pyyaml==5.3.1\r\nflask==2.0.0\r\n".encode("utf-16")) + + analysis = engine.analyze_project(root, [Path(__file__).resolve().parent.parent / "plugins"]) + + assert analysis.dependencies.names == ("flask", "pyyaml") + assert sorted(f.rule_id for f in analysis.findings) == ["vulnerable-dependency", "vulnerable-dependency"] diff --git a/tests/test_hir_builder.py b/tests/test_hir_builder.py index e5f901c..b963f57 100644 --- a/tests/test_hir_builder.py +++ b/tests/test_hir_builder.py @@ -210,10 +210,9 @@ def test_represents_raise_with_and_without_exception() -> None: "source_text, message", [ ("while x:\n pass\nelse:\n pass\n", "else"), - ("for a, b in items:\n pass\n", "single name"), - ("raise Error from cause\n", "from"), + ("for a, b in items:\n pass\nelse:\n pass\n", "else"), ], - ids=["loop-else", "tuple-target", "raise-from"], + ids=["while-else", "for-else"], ) def test_unsupported_control_flow_forms_are_reported(source_text: str, message: str) -> None: from coretrace_python.frontend import HIRBuildError diff --git a/tests/test_syntax_coverage.py b/tests/test_syntax_coverage.py index 31ca5ac..d9589ac 100644 --- a/tests/test_syntax_coverage.py +++ b/tests/test_syntax_coverage.py @@ -14,12 +14,10 @@ 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.frontend import build_hir from coretrace_python.hir import nodes from coretrace_python.semantic.scopes import ResolutionKind, analyze_scopes from coretrace_python.source import SourceManager @@ -121,9 +119,13 @@ def test_keyword_arguments_including_unpacking() -> None: 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_star_arguments_are_kept_as_starred_expressions() -> None: + module = build("def f(g, a):\n g(*a)\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 isinstance(call.expression.arguments[0], nodes.Starred) def test_pyhir_declares_a_schema_version() -> None: diff --git a/tests/test_syntax_expressions.py b/tests/test_syntax_expressions.py new file mode 100644 index 0000000..923c2a8 --- /dev/null +++ b/tests/test_syntax_expressions.py @@ -0,0 +1,285 @@ +"""Acceptance tests for the expression syntax found blocking real repositories +(``docs/architecture.md`` §3.2, §6; first post-roadmap syntax pass). + +Seven repositories under ``tests-project/`` were mostly rejected by the frontend for a +handful of everyday constructs. This pass covers the ones that add no control flow: +f-strings, slices, dictionary unpacking, starred arguments and elements, ``raise ... +from`` and tuple loop targets. Each lowers to PyIR that the existing analyses consume +through their generic operand rules, so taint, dependence and refutation apply unchanged. + +Expected to remain red until ``FormattedString``, ``Slice``, ``Starred``, ``Raise.cause`` +and tuple loop targets exist in the HIR and lower to ``BuildString``, ``BuildSlice``, +``Call.starred``, ``BuildList.unpacked``, ``BuildDict.unpacked`` and ``Raise.cause``. +""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from coretrace_python import engine +from coretrace_python.analysis import AnalysisManager +from coretrace_python.cli import main +from coretrace_python.findings.refutation import RefutationAnalysis, Status +from coretrace_python.frontend import build_hir +from coretrace_python.hir import nodes +from coretrace_python.ir.lowering import lower_module +from coretrace_python.ir.model import ( + BuildDict, + BuildList, + Call, + FunctionIR, + GetItem, + Instruction, + Raise, +) +from coretrace_python.semantic.symbols import SymbolId +from coretrace_python.source import SourceManager +from coretrace_python.taint import ( + SecurityModelAnalysis, + SecurityModelRegistry, + Sink, + Source, + TaintAnalysis, + TaintKind, +) + +try: + from coretrace_python.hir.nodes import FormattedString, Slice, Starred + from coretrace_python.ir.model import BuildSlice, BuildString +except ImportError as error: # pragma: no cover - red until the syntax lands + MISSING = error +else: + MISSING = None + + +@pytest.fixture(autouse=True) +def require_syntax() -> None: + if MISSING is not None: + pytest.fail(f"expression syntax pass is not implemented yet: {MISSING}") + + +REPO = Path(__file__).resolve().parent.parent +PLUGINS = REPO / "plugins" + +MODELS = ( + Source(SymbolId("python.builtins.input"), "stdin"), + Sink(SymbolId("python.os.system"), TaintKind.COMMAND), + Sink(SymbolId("python.subprocess.run"), TaintKind.COMMAND), +) + + +def hir(text: str) -> nodes.Module: + return build_hir(SourceManager().add_source("s.py", text)) + + +def lower(text: str) -> FunctionIR: + return lower_module(hir(text)).functions[0] + + +def instructions(function: FunctionIR, kind: type[Instruction]) -> list[Instruction]: + return [i for block in function.blocks for i in block.instructions if isinstance(i, kind)] + + +def manager_for(text: str) -> AnalysisManager: + manager = AnalysisManager(hir("import os\nimport subprocess\n\n" + text)) + manager.register(*engine.ALL_ANALYSES) + registry = SecurityModelRegistry() + registry.register(*MODELS) + manager.provide(SecurityModelAnalysis, registry.freeze()) + return manager + + +def first_function(manager: AnalysisManager) -> nodes.Function: + return next(s for s in manager.module.body if isinstance(s, nodes.Function)) + + +def flow_lines(text: str) -> list[int]: + manager = manager_for(text) + return sorted(f.location.start_line for f in manager.get(TaintAnalysis, first_function(manager)).flows) + + +def verdicts(text: str) -> list[Status]: + manager = manager_for(text) + return [v.status for v in manager.get(RefutationAnalysis, first_function(manager)).all()] + + +def emit(text: str, tmp_path: Path, capsys) -> str: # type: ignore[no-untyped-def] + source = tmp_path / "e.py" + source.write_text(text, encoding="utf-8") + assert main(["--emit-ir", str(source)]) == 0 + return capsys.readouterr().out + + +# --------------------------------------------------------------------------- f-strings + + +def test_fstrings_become_formatted_strings_with_their_parts() -> None: + module = hir("def f(host, n):\n return f'ping {host} -c {n:>{n}} done'\n") + function = module.body[0] + assert isinstance(function, nodes.Function) + returned = function.body[0] + assert isinstance(returned, nodes.Return) + text = returned.value + + assert isinstance(text, FormattedString) + kinds = [type(part).__name__ for part in text.parts] + assert kinds == ["Constant", "Name", "Constant", "Name", "Name", "Constant"] + assert [p.value for p in text.parts if isinstance(p, nodes.Constant)] == ["ping ", " -c ", " done"] + + +def test_fstrings_lower_to_build_string(tmp_path: Path, capsys) -> None: # type: ignore[no-untyped-def] + (built,) = instructions(lower("def f(host):\n return f'ping {host}'\n"), BuildString) + assert isinstance(built, BuildString) and len(built.parts) == 2 + output = emit("def f(host):\n return f'ping {host}'\n", tmp_path, capsys) + assert output == ( + "func @f(%0) {\n" + "entry:\n" + " %1 = const 'ping '\n" + " %2 = build_string %1, %0\n" + " return %2\n" + "}\n" + ) + + +def test_taint_and_refutation_flow_through_fstrings() -> None: + assert flow_lines("def f():\n host = input()\n os.system(f'ping {host}')\n") == [6] + assert verdicts("def f():\n host = input()\n if host.isdigit():\n os.system(f'ping -c {host}')\n") == [ + Status.REFUTED + ] + assert flow_lines("def f():\n n = 3\n os.system(f'ping -c {n}')\n") == [] + + +# --------------------------------------------------------------------------- slices + + +def test_slices_lower_to_build_slice_keys(tmp_path: Path, capsys) -> None: # type: ignore[no-untyped-def] + module = hir("def f(x, a):\n return x[1:3], x[::2], x[a:]\n") + returned = module.body[0].body[0] # type: ignore[union-attr] + assert isinstance(returned, nodes.Return) and isinstance(returned.value, nodes.Tuple) + first, second, third = returned.value.elements + assert all(isinstance(e, nodes.Subscript) and isinstance(e.key, Slice) for e in (first, second, third)) + assert isinstance(third.key, Slice) and third.key.upper is None and third.key.step is None # type: ignore[union-attr] + + (built,) = instructions(lower("def f(x):\n return x[1:]\n"), BuildSlice) + assert isinstance(built, BuildSlice) and built.upper is None and built.step is None + output = emit("def f(x):\n return x[1:]\n", tmp_path, capsys) + assert output == ( + "func @f(%0) {\n" + "entry:\n" + " %1 = const 1\n" + " %2 = build_slice %1, none, none\n" + " %3 = get_item %0, %2\n" + " return %3\n" + "}\n" + ) + + +def test_taint_flows_through_slices() -> None: + assert flow_lines("def f():\n cmd = input()\n os.system(cmd[1:])\n") == [6] + assert flow_lines("def f():\n cmd = input()\n os.system('ls'[0:2])\n") == [] + + +# --------------------------------------------------------------------------- unpacking + + +def test_dict_unpacking_and_starred_elements_lower_with_unpacked_operands() -> None: + function = lower("def f(base, extra, xs):\n d = {**base, 'k': 1}\n l = [*xs, 2]\n t = (*xs,)\n return d, l, t\n") + + (build_dict,) = instructions(function, BuildDict) + (build_list,) = instructions(function, BuildList) + assert isinstance(build_dict, BuildDict) and len(build_dict.unpacked) == 1 and len(build_dict.items) == 1 + assert isinstance(build_list, BuildList) and len(build_list.unpacked) == 1 and len(build_list.elements) == 1 + module = hir("def f(base):\n return {**base}\n") + returned = module.body[0].body[0] # type: ignore[union-attr] + assert isinstance(returned, nodes.Return) and isinstance(returned.value, nodes.Dict) + ((key, _),) = returned.value.items + assert key is None + + +def test_taint_flows_through_unpacked_collections() -> None: + assert flow_lines("def f():\n base = {'cmd': input()}\n d = {**base, 'x': 1}\n os.system(d['cmd'])\n") == [7] + assert flow_lines("def f():\n xs = [input()]\n l = [*xs, 'ls']\n os.system(l[0])\n") == [7] + assert flow_lines("def f():\n base = {'cmd': 'ls'}\n d = {**base}\n os.system(d['cmd'])\n") == [] + + +def test_starred_call_arguments_are_kept_apart_and_reach_sinks() -> None: + function = lower("def f(args, kw):\n return g(1, *args, key=2, **kw)\n") + (call,) = instructions(function, Call) + assert isinstance(call, Call) + assert len(call.arguments) == 1 and len(call.starred) == 1 and len(call.keywords) == 2 + assert len(call.argument_values()) == 4 + + module = hir("def f(args):\n return g(*args)\n") + returned = module.body[0].body[0] # type: ignore[union-attr] + assert isinstance(returned, nodes.Return) and isinstance(returned.value, nodes.Call) + assert isinstance(returned.value.arguments[0], Starred) + + +def test_taint_flows_through_starred_arguments_and_known_callees() -> None: + assert flow_lines("def f():\n parts = ['sh', '-c', input()]\n subprocess.run(*parts)\n") == [6] + assert flow_lines( + "def run(cmd):\n os.system(cmd)\n\ndef f():\n args = [input()]\n run(*args)\n" + ) == [] + manager = manager_for("def run(cmd):\n os.system(cmd)\n\ndef f():\n args = [input()]\n run(*args)\n") + f = [s for s in manager.module.body if isinstance(s, nodes.Function)][1] + assert [fl.location.start_line for fl in manager.get(TaintAnalysis, f).flows] == [9] + + +# --------------------------------------------------------------------------- raise from + + +def test_raise_from_keeps_its_cause(tmp_path: Path, capsys) -> None: # type: ignore[no-untyped-def] + module = hir("def f(err):\n raise ValueError('x') from err\n") + raised = module.body[0].body[0] # type: ignore[union-attr] + assert isinstance(raised, nodes.Raise) and isinstance(raised.cause, nodes.Name) + + function = lower("def f(err):\n raise ValueError('x') from err\n") + terminator = function.blocks[0].terminator + assert isinstance(terminator, Raise) and terminator.cause == function.parameters[0] + + output = emit("def f(err):\n raise err from None\n", tmp_path, capsys) + assert output.splitlines()[-2] == " raise %0 from %1" + + +# --------------------------------------------------------------------------- tuple loop targets + + +def test_tuple_loop_targets_destructure_each_item() -> None: + module = hir("def f(items):\n for k, (v, w) in items:\n print(k, v, w)\n") + loop = module.body[0].body[0] # type: ignore[union-attr] + assert isinstance(loop, nodes.For) + assert loop.target.identifier.isidentifier() + first = loop.body[0] + assert isinstance(first, nodes.Assign) and isinstance(first.target, nodes.Tuple) + assert isinstance(first.value, nodes.Name) and first.value.identifier == loop.target.identifier + + function = lower("def f(items):\n for k, v in items:\n print(k, v)\n") + assert len(instructions(function, GetItem)) == 2 + + +def test_taint_flows_through_tuple_loop_targets() -> None: + assert flow_lines("def f():\n pairs = [('a', input())]\n for name, cmd in pairs:\n os.system(cmd)\n") == [7] + + +# --------------------------------------------------------------------------- end to end + + +def test_a_realistic_module_is_fully_analysed() -> None: + findings = engine.check( + SourceManager().add_source( + "topology.py", + "import os\nimport subprocess\n\n" + "def probe(hosts, extra):\n" + " results = {}\n" + " for name, host in hosts.items():\n" + " try:\n" + " results[name] = subprocess.run(['ping', '-c', '1', host[:64]], *extra)\n" + " except OSError as err:\n" + " raise RuntimeError(f'cannot ping {host}') from err\n" + " return {**results, 'count': len(results)}\n", + ), + [PLUGINS], + ) + assert [f.rule_id for f in findings if f.rule_id in ("syntax-error", "unsupported-syntax")] == []