feat: expression syntax pass

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

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:

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))

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)

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":

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:

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:

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)

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)
 
 

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: