fix: track open dict copy keysets

desloppify/languages/python/detectors/dict_keys/shared.py

@@ -17,6 +17,7 @@ class TrackedDict:
     returned_or_passed: bool = False
     has_dynamic_key: bool = False
     has_star_unpack: bool = False
+    keyset_is_open: bool = False
     writes: dict[str, list[int]] = field(default_factory=lambda: defaultdict(list))
     reads: dict[str, list[int]] = field(default_factory=lambda: defaultdict(list))
     bulk_read: bool = False

desloppify/languages/python/detectors/dict_keys/visitor.py

@@ -9,8 +9,10 @@
     _get_name,
     _get_str_key,
 )
+
 from .visitor_helpers import (
     analyze_scope_issues,
+    dict_source_provenance,
     mark_assignment_escape,
     mark_returned_or_passed,
     record_call_interactions,
@@ -38,9 +40,15 @@ def _track(
         *,
         locally_created: bool,
         initial_keys: list[str] | None = None,
+        keyset_is_open: bool = False,
     ) -> TrackedDict:
         scope = self._current_scope()
-        td = TrackedDict(name=name, created_line=line, locally_created=locally_created)
+        td = TrackedDict(
+            name=name,
+            created_line=line,
+            locally_created=locally_created,
+            keyset_is_open=keyset_is_open,
+        )
         if initial_keys:
             for k in initial_keys:
                 td.writes[k].append(line)
@@ -112,17 +120,24 @@ def visit_AugAssign(self, node: ast.AugAssign) -> None:
         self._check_subscript_write(node.target, node.lineno)
         self.generic_visit(node)
 
-    def _check_dict_creation(self, name: str, value: ast.expr, line: int):
-        """Detect d = {}, d = dict(), d = {"k": v, ...}."""
-        initial_keys: list[str] = []
-        is_creation = False
+    def _check_dict_creation(self, name: str, value: ast.expr, line: int) -> None:
+        """Track aliases and newly created dict expressions."""
+
+        source_name = _get_name(value)
+        if source_name:
+            tracked = self._get_tracked(source_name)
+            if tracked is not None:
+                self._current_scope()[name] = tracked
+                if name.startswith("self.") and self._in_init_or_setup:
+                    self._class_dicts[name] = tracked
+                return
+
+        provenance = dict_source_provenance(self, value)
+        if provenance is None:
+            return
+        initial_keys, keyset_is_open = provenance
 
         if isinstance(value, ast.Dict):
-            is_creation = True
-            for k in value.keys:
-                sk = _get_str_key(k) if k else None
-                if sk:
-                    initial_keys.append(sk)
             # Collect dict literal for schema drift
             if (
                 all(
@@ -140,23 +155,16 @@ def _check_dict_creation(self, name: str, value: ast.expr, line: int):
                         "keys": frozenset(keys),
                     }
                 )
-        elif (
-            isinstance(value, ast.Call)
-            and isinstance(value.func, ast.Name)
-            and value.func.id == "dict"
-        ):
-            is_creation = True
-            for kw in value.keywords:
-                if kw.arg:
-                    initial_keys.append(kw.arg)
-
-        if is_creation:
-            td = self._track(
-                name, line, locally_created=True, initial_keys=initial_keys
-            )
-            # Store as class dict if it's self.x
-            if name.startswith("self.") and self._in_init_or_setup:
-                self._class_dicts[name] = td
+        td = self._track(
+            name,
+            line,
+            locally_created=True,
+            initial_keys=initial_keys,
+            keyset_is_open=keyset_is_open,
+        )
+        # Store as class dict if it's self.x
+        if name.startswith("self.") and self._in_init_or_setup:
+            self._class_dicts[name] = td
 
     def _check_subscript_write(self, target: ast.expr, line: int):
         """Handle d["key"] = val or d["key"] += val."""
@@ -173,6 +181,7 @@ def _check_subscript_write(self, target: ast.expr, line: int):
             td.writes[key].append(line)
         else:
             td.has_dynamic_key = True
+            td.keyset_is_open = True
 
     # -- Dict reads --
 

desloppify/languages/python/detectors/dict_keys/visitor_helpers.py

@@ -54,6 +54,80 @@ def mark_assignment_escape(visitor, targets: list[ast.expr], value: ast.expr) ->
             return
 
 
+def dict_source_provenance(
+    visitor,
+    value: ast.expr,
+) -> tuple[list[str], bool] | None:
+    """Return known keys and whether a dict expression may contain other keys."""
+
+    source_name = _get_name(value)
+    if source_name:
+        tracked = visitor._get_tracked(source_name)
+        if tracked is None:
+            return None
+        return list(tracked.writes), tracked.keyset_is_open
+
+    if isinstance(value, ast.Dict):
+        known_keys: list[str] = []
+        keyset_is_open = False
+        for key_node, value_node in zip(value.keys, value.values, strict=True):
+            if key_node is None:
+                provenance = dict_source_provenance(visitor, value_node)
+                if provenance is None:
+                    keyset_is_open = True
+                else:
+                    source_keys, source_is_open = provenance
+                    known_keys.extend(source_keys)
+                    keyset_is_open |= source_is_open
+                continue
+            key = _get_str_key(key_node)
+            if key is None:
+                keyset_is_open = True
+            else:
+                known_keys.append(key)
+        return known_keys, keyset_is_open
+
+    if not isinstance(value, ast.Call):
+        return None
+
+    if isinstance(value.func, ast.Name) and value.func.id == "dict":
+        known_keys = []
+        keyset_is_open = False
+        for argument in value.args:
+            provenance = dict_source_provenance(visitor, argument)
+            if provenance is None:
+                keyset_is_open = True
+            else:
+                source_keys, source_is_open = provenance
+                known_keys.extend(source_keys)
+                keyset_is_open |= source_is_open
+        for keyword in value.keywords:
+            if keyword.arg is not None:
+                known_keys.append(keyword.arg)
+                continue
+            provenance = dict_source_provenance(visitor, keyword.value)
+            if provenance is None:
+                keyset_is_open = True
+            else:
+                source_keys, source_is_open = provenance
+                known_keys.extend(source_keys)
+                keyset_is_open |= source_is_open
+        return known_keys, keyset_is_open
+
+    if (
+        isinstance(value.func, ast.Attribute)
+        and value.func.attr == "copy"
+        and not value.args
+        and not value.keywords
+    ):
+        source_name = _get_name(value.func.value)
+        if source_name:
+            tracked = visitor._get_tracked(source_name)
+            if tracked is not None:
+                return list(tracked.writes), tracked.keyset_is_open
+    return None
+
+
 def record_call_interactions(visitor, node: ast.Call) -> None:
     """Update tracked dict read/write metadata from a call expression."""
     if isinstance(node.func, ast.Attribute):
@@ -77,19 +151,29 @@ def record_call_interactions(visitor, node: ast.Call) -> None:
                             tracked.writes[key].append(node.lineno)
                         else:
                             tracked.has_dynamic_key = True
+                            tracked.keyset_is_open = True
                 elif method == "update":
-                    if node.args and isinstance(node.args[0], ast.Dict):
-                        for key_node in node.args[0].keys:
-                            key = _get_str_key(key_node) if key_node else None
-                            if key:
+                    if node.args:
+                        provenance = dict_source_provenance(visitor, node.args[0])
+                        if provenance is None:
+                            tracked.keyset_is_open = True
+                        else:
+                            source_keys, source_is_open = provenance
+                            for key in source_keys:
                                 tracked.writes[key].append(node.lineno)
-                            elif key_node is None:
-                                tracked.has_dynamic_key = True
+                            tracked.keyset_is_open |= source_is_open
                     for kw in node.keywords:
                         if kw.arg:
                             tracked.writes[kw.arg].append(node.lineno)
                         else:
-                            tracked.has_dynamic_key = True
+                            provenance = dict_source_provenance(visitor, kw.value)
+                            if provenance is None:
+                                tracked.keyset_is_open = True
+                            else:
+                                source_keys, source_is_open = provenance
+                                for key in source_keys:
+                                    tracked.writes[key].append(node.lineno)
+                                tracked.keyset_is_open |= source_is_open
                 elif method in _BULK_READ_METHODS:
                     tracked.bulk_read = True
 
@@ -115,7 +199,12 @@ def analyze_scope_issues(
 ) -> list[dict]:
     """Analyze a completed scope and return dict-key issues."""
     issues: list[dict] = []
+    seen_tracked: set[int] = set()
     for tracked in scope.values():
+        tracked_identity = id(tracked)
+        if tracked_identity in seen_tracked:
+            continue
+        seen_tracked.add(tracked_identity)
         if not tracked.locally_created:
             continue
 
@@ -157,7 +246,7 @@ def analyze_scope_issues(
                     }
                 )
 
-        phantom_keys = read_keys - written_keys
+        phantom_keys = set() if tracked.keyset_is_open else read_keys - written_keys
         for key in sorted(phantom_keys):
             line = tracked.reads[key][0]
             issues.append(

desloppify/languages/python/tests/test_py_dict_keys.py

@@ -94,6 +94,128 @@ def build():
         entries, _ = detect_dict_key_flow(path)
         assert "phantom_read" not in _kinds(entries)
 
+    def test_open_positional_dict_copy_suppresses_phantom_reads(self, tmp_path):
+        path = _write_py(
+            tmp_path,
+            """\
+            def read_external(source):
+                payload = dict(source)
+                return payload.get("message"), payload.pop("error_code", None)
+        """,
+        )
+
+        entries, _ = detect_dict_key_flow(path)
+
+        assert "phantom_read" not in _kinds(entries)
+
+    def test_open_unpack_copies_suppress_phantom_reads(self, tmp_path):
+        path = _write_py(
+            tmp_path,
+            """\
+            def read_external(source):
+                call_copy = dict(**source)
+                literal_copy = {**source}
+                return call_copy.get("message"), literal_copy.get("error")
+        """,
+        )
+
+        entries, _ = detect_dict_key_flow(path)
+
+        assert "phantom_read" not in _kinds(entries)
+
+    def test_open_keyset_suppresses_derived_near_miss(self, tmp_path):
+        path = _write_py(
+            tmp_path,
+            """\
+            def read_external(source):
+                payload = dict(source, message="fallback")
+                return payload.get("messag")
+        """,
+        )
+
+        entries, _ = detect_dict_key_flow(path)
+
+        assert not {"phantom_read", "near_miss"} & _kinds(entries)
+
+    def test_closed_dict_constructors_still_detect_typos(self, tmp_path):
+        path = _write_py(
+            tmp_path,
+            """\
+            def read_closed():
+                literal = {"message": "one"}
+                keyword = dict(message="two")
+                return literal.get("messag"), keyword.get("messag")
+        """,
+        )
+
+        entries, _ = detect_dict_key_flow(path)
+
+        assert len(_find_kind(entries, "phantom_read")) == 2
+        assert len(_find_kind(entries, "near_miss")) == 2
+
+    def test_closed_keyset_survives_literal_constructor_alias_and_copy(self, tmp_path):
+        path = _write_py(
+            tmp_path,
+            """\
+            def read_closed():
+                original = {"message": "one"}
+                alias = original
+                constructor = dict(alias)
+                method_copy = alias.copy()
+                unpack_copy = dict(**{"message": "two"})
+                positional_literal = dict({"message": "three"})
+                return (
+                    alias.get("messag"),
+                    constructor.get("messag"),
+                    method_copy.get("messag"),
+                    unpack_copy.get("messag"),
+                    positional_literal.get("messag"),
+                )
+        """,
+        )
+
+        entries, _ = detect_dict_key_flow(path)
+
+        assert len(_find_kind(entries, "phantom_read")) == 5
+        assert len(_find_kind(entries, "near_miss")) == 5
+
+    def test_open_keyset_survives_alias_and_tracked_copies(self, tmp_path):
+        path = _write_py(
+            tmp_path,
+            """\
+            def read_external(source):
+                original = dict(source)
+                alias = original
+                constructor = dict(alias)
+                method_copy = alias.copy()
+                return constructor.get("message"), method_copy.get("error")
+        """,
+        )
+
+        entries, _ = detect_dict_key_flow(path)
+
+        assert "phantom_read" not in _kinds(entries)
+
+    def test_update_propagates_open_and_closed_keysets(self, tmp_path):
+        path = _write_py(
+            tmp_path,
+            """\
+            def read_external(source):
+                open_payload = {}
+                open_payload.update(source)
+                closed_payload = {}
+                closed_payload.update({"message": "fallback"})
+                return open_payload.get("message"), closed_payload.get("messag")
+        """,
+        )
+
+        entries, _ = detect_dict_key_flow(path)
+
+        phantom = _find_kind(entries, "phantom_read")
+        near_miss = _find_kind(entries, "near_miss")
+        assert [issue["variable"] for issue in phantom] == ["closed_payload"]
+        assert [issue["variable"] for issue in near_miss] == ["closed_payload"]
+
 
 # ── Dead writes (written key never read) ──────────────────