Support class keyword arguments and any assignable with target

docs/usage.md

@@ -365,7 +365,8 @@ boolean operators, chained comparisons, keyword and starred arguments, `with`, `
 `else` clauses, `break`, `continue`, `raise` with `from`, conditional expressions,
 comprehensions, lambdas, nested functions and classes, `global`, `nonlocal`, `del`,
 assignment expressions (`(y := f(x))`, laid out as an assignment before the statement),
-`yield from`, starred assignment targets (`first, *rest = items`) and `match` with every
+`yield from`, starred assignment targets (`first, *rest = items`), class keyword arguments
+such as `metaclass=`, `with` targets of any assignable form and `match` with every
 pattern kind. `finally` is modelled on the normal path only. A
 `nonlocal` write made by a nested function flows back to the enclosing variable when the
 enclosing function calls that nested function directly by name; a nested function passed

src/coretrace_python/frontend/ast_adapter.py

@@ -249,13 +249,9 @@ def statement(self, node: ast.stmt) -> nodes.Statement:
         if isinstance(node, (ast.With, ast.AsyncWith)):
             items = []
             for item in node.items:
-                bound: nodes.Name | None = None
+                bound: nodes.Target | 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))
+                    bound = self.target(item.optional_vars)
                 items.append(
                     nodes.WithItem(
                         self.expression(item.context_expr), bound, self.span(item.context_expr)
@@ -389,12 +385,14 @@ def block(self, statements: list[ast.stmt]) -> tuple[nodes.Statement, ...]:
         return tuple(found for statement in statements for found in self.statements(statement))
 
     def class_definition(self, node: ast.ClassDef) -> nodes.Class:
-        if node.keywords:
-            self.fail(node, "class keyword arguments are not supported yet")
         bases = tuple(self.expression(base) for base in node.bases)
+        keywords = tuple(
+            nodes.Keyword(keyword.arg, self.expression(keyword.value), self.span(keyword))
+            for keyword in node.keywords
+        )
         body = self.block(node.body)
         decorators = tuple(self.expression(d) for d in node.decorator_list)
-        return nodes.Class(node.name, bases, body, self.span(node), decorators)
+        return nodes.Class(node.name, bases, body, self.span(node), decorators, keywords)
 
     def function(self, node: ast.FunctionDef | ast.AsyncFunctionDef) -> nodes.Function:
         body = self.block(node.body)

src/coretrace_python/hir/nodes.py

@@ -453,7 +453,7 @@ class Raise:
 @dataclass(frozen=True, slots=True)
 class WithItem:
     context: Expression
-    target: Name | None
+    target: Target | None
     span: SourceSpan
 
 
@@ -558,6 +558,8 @@ class Class:
     body: tuple[Statement, ...]
     span: SourceSpan
     decorators: tuple[Expression, ...] = ()
+    # ``metaclass=`` and the other keyword arguments of the class statement.
+    keywords: tuple[Keyword, ...] = ()
 
 
 Statement: TypeAlias = (

src/coretrace_python/ir/lowering.py

@@ -297,6 +297,8 @@ def statement(self, node: nodes.Statement) -> None:
                 self.expression(decorator)
             for base in node.bases:
                 self.expression(base)
+            for keyword in node.keywords:
+                self.expression(keyword.value)
             made_class = self.emit(MakeClass(self.new_value(), node.span, node.name))
             self.store(nodes.Name(node.name, node.span), made_class)
             return

src/coretrace_python/semantic/scopes.py

@@ -237,7 +237,7 @@ def statement(self, node: nodes.Statement, scope: _ScopeBuilder) -> None:
             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.target(item.target, scope)
             self.body(node.body, scope)
         elif isinstance(node, nodes.Try):
             self.body(node.body, scope)
@@ -291,6 +291,8 @@ def statement(self, node: nodes.Statement, scope: _ScopeBuilder) -> None:
                 self.expression(decorator, scope)
             for base in node.bases:
                 self.expression(base, scope)
+            for keyword in node.keywords:
+                self.expression(keyword.value, scope)
             scope.bind(node.name, BindingKind.CLASS, node.span)
             self.body(node.body, self.open(scope, ScopeKind.CLASS, node.name, node.span))
         elif isinstance(node, nodes.If | nodes.While):

tests/regression/expected/full-stack-fastapi-template.json

@@ -1,9 +1,9 @@
 {
   "coverage": {
     "files": 43,
-    "files_analysed": 41,
-    "functions": 139,
-    "functions_analysed": 139
+    "files_analysed": 42,
+    "functions": 140,
+    "functions_analysed": 140
   },
   "findings": [
     {
@@ -50,8 +50,8 @@
     },
     {
       "file": "backend/app/models.py",
-      "line": 1,
-      "rule": "syntax-error",
+      "line": 123,
+      "rule": "hardcoded-credential",
       "function": null
     },
     {

tests/regression/expected/httpie.json

@@ -1,9 +1,9 @@
 {
   "coverage": {
     "files": 132,
-    "files_analysed": 129,
-    "functions": 1016,
-    "functions_analysed": 1016
+    "files_analysed": 132,
+    "functions": 1070,
+    "functions_analysed": 1070
   },
   "findings": [
     {
@@ -12,12 +12,6 @@
       "rule": "missing-timeout",
       "function": "fetch"
     },
-    {
-      "file": "extras/profiling/run.py",
-      "line": 1,
-      "rule": "syntax-error",
-      "function": null
-    },
     {
       "file": "httpie/client.py",
       "line": 114,
@@ -42,12 +36,6 @@
       "rule": "missing-timeout",
       "function": "_fetch_updates"
     },
-    {
-      "file": "httpie/output/streams.py",
-      "line": 1,
-      "rule": "syntax-error",
-      "function": null
-    },
     {
       "file": "tests/test_binary.py",
       "line": 49,
@@ -68,9 +56,9 @@
     },
     {
       "file": "tests/test_uploads.py",
-      "line": 1,
-      "rule": "syntax-error",
-      "function": null
+      "line": 103,
+      "rule": "command-injection",
+      "function": "stdin_processes"
     }
   ]
 }

tests/regression/expected/luigi.json

@@ -1,9 +1,9 @@
 {
   "coverage": {
     "files": 259,
-    "files_analysed": 245,
-    "functions": 4462,
-    "functions_analysed": 4462
+    "files_analysed": 259,
+    "functions": 5169,
+    "functions_analysed": 5169
   },
   "findings": [
     {
@@ -18,30 +18,12 @@
       "rule": "hardcoded-credential",
       "function": null
     },
-    {
-      "file": "luigi/contrib/beam_dataflow.py",
-      "line": 1,
-      "rule": "syntax-error",
-      "function": null
-    },
     {
       "file": "luigi/contrib/esindex.py",
       "line": 159,
       "rule": "weak-crypto",
       "function": "marker_index_document_id"
     },
-    {
-      "file": "luigi/contrib/hdfs/abstract_client.py",
-      "line": 1,
-      "rule": "syntax-error",
-      "function": null
-    },
-    {
-      "file": "luigi/contrib/hive.py",
-      "line": 1,
-      "rule": "syntax-error",
-      "function": null
-    },
     {
       "file": "luigi/contrib/lsf.py",
       "line": 263,
@@ -84,18 +66,6 @@
       "rule": "command-injection",
       "function": "track_and_progress"
     },
-    {
-      "file": "luigi/contrib/presto.py",
-      "line": 1,
-      "rule": "syntax-error",
-      "function": null
-    },
-    {
-      "file": "luigi/contrib/pyspark_runner.py",
-      "line": 1,
-      "rule": "syntax-error",
-      "function": null
-    },
     {
       "file": "luigi/contrib/salesforce.py",
       "line": 258,
@@ -186,42 +156,6 @@
       "rule": "command-injection",
       "function": "_run_job"
     },
-    {
-      "file": "luigi/metrics.py",
-      "line": 1,
-      "rule": "syntax-error",
-      "function": null
-    },
-    {
-      "file": "luigi/parameter.py",
-      "line": 1,
-      "rule": "syntax-error",
-      "function": null
-    },
-    {
-      "file": "luigi/rpc.py",
-      "line": 1,
-      "rule": "syntax-error",
-      "function": null
-    },
-    {
-      "file": "luigi/target.py",
-      "line": 1,
-      "rule": "syntax-error",
-      "function": null
-    },
-    {
-      "file": "luigi/task.py",
-      "line": 1,
-      "rule": "syntax-error",
-      "function": null
-    },
-    {
-      "file": "luigi/task_history.py",
-      "line": 1,
-      "rule": "syntax-error",
-      "function": null
-    },
     {
       "file": "test/_test_ftp.py",
       "line": 43,
@@ -402,24 +336,6 @@
       "rule": "high-entropy-string",
       "function": "test_task_list_filter_by_search_long_family_name"
     },
-    {
-      "file": "test/task_test.py",
-      "line": 1,
-      "rule": "syntax-error",
-      "function": null
-    },
-    {
-      "file": "test/worker_scheduler_com_test.py",
-      "line": 1,
-      "rule": "syntax-error",
-      "function": null
-    },
-    {
-      "file": "test/worker_test.py",
-      "line": 1,
-      "rule": "syntax-error",
-      "function": null
-    },
     {
       "file": "tox.ini",
       "line": 55,

tests/test_class_keywords_with_targets.py

@@ -0,0 +1,91 @@
+"""Acceptance tests for class keyword arguments and general ``with`` targets (issue #71).
+
+``class Task(Base, metaclass=Register)`` and ``with ctx() as self.conn`` or
+``as (a, b)`` rejected the whole file as a ``syntax-error`` on the broadened corpus (12
+files in luigi, 2 in httpie, 1 in the FastAPI template). Class keyword arguments are kept
+on the class and evaluated where the class is defined; a ``with`` target is any assignable
+target and receives what the context manager enters.
+
+Expected to remain red until ``nodes.Class`` has ``keywords`` and ``WithItem.target`` is a target.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+from coretrace_python import engine
+from coretrace_python.findings import Finding
+from coretrace_python.frontend import build_hir
+from coretrace_python.hir import nodes
+from coretrace_python.source import SourceManager
+
+MISSING = None if "keywords" in nodes.Class.__dataclass_fields__ else "Class has no keywords"
+
+
+@pytest.fixture(autouse=True)
+def require_forms() -> None:
+    if MISSING is not None:
+        pytest.fail(f"the forms are not supported yet: {MISSING}")
+
+
+REPO = Path(__file__).resolve().parent.parent
+PLUGINS = REPO / "src" / "coretrace_python" / "bundled"
+
+
+def hir(text: str) -> nodes.Module:
+    return build_hir(SourceManager().add_source("m.py", text))
+
+
+def check(text: str) -> tuple[Finding, ...]:
+    return engine.check(SourceManager().add_source("m.py", text), [PLUGINS])
+
+
+def rules(findings: tuple[Finding, ...]) -> list[str]:
+    return sorted(f.rule_id for f in findings)
+
+
+def test_class_keyword_arguments_are_kept_on_the_class() -> None:
+    module = hir("class Task(Base, metaclass=Register, flag=True):\n    pass\n")
+    cls = module.body[0]
+
+    assert isinstance(cls, nodes.Class)
+    assert [k.name for k in cls.keywords] == ["metaclass", "flag"]
+    assert isinstance(cls.keywords[0].value, nodes.Name) and cls.keywords[0].value.identifier == "Register"
+
+
+def test_with_targets_may_be_attributes_subscripts_and_tuples() -> None:
+    module = hir("def f(self, d):\n    with a() as self.conn, b() as d['k'], c() as (x, y):\n        pass\n")
+    function = module.body[0]
+    assert isinstance(function, nodes.Function)
+    statement = function.body[0]
+    assert isinstance(statement, nodes.With)
+    targets = [item.target for item in statement.items]
+
+    assert isinstance(targets[0], nodes.Attribute)
+    assert isinstance(targets[1], nodes.Subscript)
+    assert isinstance(targets[2], nodes.Tuple)
+
+
+def test_methods_of_a_class_with_keyword_arguments_are_analysed() -> None:
+    text = "import os\n\nclass Task(object, metaclass=type):\n    def run(self):\n        os.system(input())\n"
+    analysis = engine.analyze_file(SourceManager().add_source("m.py", text), [PLUGINS])
+
+    assert rules(analysis.findings) == ["command-injection"]
+    assert analysis.coverage.summary() == "coverage: 1/1 files, 1/1 functions"
+
+
+def test_a_local_class_with_keyword_arguments_evaluates_them() -> None:
+    text = "import os\n\ndef make(meta):\n    class Task(metaclass=meta):\n        def run(self):\n            os.system(input())\n    return Task\n"
+    analysis = engine.analyze_file(SourceManager().add_source("m.py", text), [PLUGINS])
+
+    assert rules(analysis.findings) == ["command-injection"]
+    assert analysis.coverage.summary() == "coverage: 1/1 files, 2/2 functions"
+
+
+def test_with_targets_receive_the_entered_value() -> None:
+    text = "import os\n\ndef run(ctx):\n    with ctx(input()) as (cmd, other):\n        os.system(cmd)\n"
+    assert rules(check(text)) == ["command-injection"]
+    text = "import os\n\nclass R:\n    def run(self, ctx):\n        with ctx(input()) as self.cmd:\n            os.system(self.cmd)\n"
+    assert rules(check(text)) == ["command-injection"]