diff --git a/desloppify/languages/python/detectors/dict_keys/shared.py b/desloppify/languages/python/detectors/dict_keys/shared.py index 023585942..b886ed9fe 100644 --- a/desloppify/languages/python/detectors/dict_keys/shared.py +++ b/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 diff --git a/desloppify/languages/python/detectors/dict_keys/visitor.py b/desloppify/languages/python/detectors/dict_keys/visitor.py index f4f169af9..91aac58db 100644 --- a/desloppify/languages/python/detectors/dict_keys/visitor.py +++ b/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 -- diff --git a/desloppify/languages/python/detectors/dict_keys/visitor_helpers.py b/desloppify/languages/python/detectors/dict_keys/visitor_helpers.py index e3aaef318..a2b037bf9 100644 --- a/desloppify/languages/python/detectors/dict_keys/visitor_helpers.py +++ b/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( diff --git a/desloppify/languages/python/tests/test_py_dict_keys.py b/desloppify/languages/python/tests/test_py_dict_keys.py index c4e8ef131..acf6794bc 100644 --- a/desloppify/languages/python/tests/test_py_dict_keys.py +++ b/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) ──────────────────