Skip to main content

brokk_bifrost_python/
package_paths.rs

1//! Structured interpretation of surviving package search-path extensions.
2
3use crate::syntax::{python_plain_string_literal, python_static_attribute_path};
4use brokk_bifrost_core::analyzer::tree_walk::{WalkControl, walk_named_tree_preorder};
5use brokk_bifrost_core::hash::{HashMap, HashSet};
6use brokk_bifrost_core::path_normalization::NormalizePath;
7use std::path::{Component, PathBuf};
8use tree_sitter::Node;
9
10/// Surviving package-location-derived extensions, relative to the initializer
11/// directory. Literals may be joined onto the anchor, but never establish one.
12/// The anchor is symbolic, so these facts remain valid when a blob is remounted.
13pub fn appended_path_dirs(root: Node<'_>, source: &str) -> Vec<PathBuf> {
14    let mut bound: HashMap<String, PathValue> = HashMap::default();
15    bound.insert(
16        "__file__".to_owned(),
17        PathValue::Anchored {
18            parents: 0,
19            suffix: PathBuf::from("__init__.py"),
20        },
21    );
22    let mut appended: Vec<PathBuf> = Vec::new();
23    // Once `__path__` reaches code this reader does not follow (an alias, a
24    // call argument, a function body), that code can change the list, so no
25    // recorded directory is proven to survive.
26    let mut escaped = false;
27    let mut copying = CopyingNames::default();
28    walk_named_tree_preorder(root, true, |node| {
29        match node.kind() {
30            // Only module-level code runs the extension, so a name bound inside
31            // a function or class body says nothing about what `__path__`
32            // holds after the marker is executed. A body that names `__path__`
33            // can still change it when called, which this reader does not follow.
34            "function_definition" | "class_definition" | "lambda" => {
35                python_unbind(python_bound_names(node, source), &mut bound, &mut copying);
36                escaped |= python_subtree_names_path(node, source);
37                python_forget_names_in_body(node, source, &mut copying);
38                return WalkControl::SkipChildren;
39            }
40            "identifier" => {
41                if python_names_path(Some(node), source)
42                    && !python_path_use_is_modelled(node, source, &copying)
43                {
44                    escaped = true;
45                }
46                python_check_copying_use(node, source, &mut copying);
47            }
48            "assignment" => {
49                if node
50                    .parent()
51                    .is_some_and(|parent| parent.kind() == "assignment")
52                {
53                    return WalkControl::Continue;
54                }
55                let mut assignment = node;
56                loop {
57                    if python_targets_path(assignment.child_by_field_name("left"), source) {
58                        appended.clear();
59                    }
60                    match assignment.child_by_field_name("right") {
61                        Some(right) if right.kind() == "assignment" => assignment = right,
62                        _ => break,
63                    }
64                }
65                if node
66                    .parent()
67                    .is_some_and(|parent| parent.kind() == "expression_statement")
68                    && node
69                        .parent()
70                        .and_then(|parent| parent.parent())
71                        .is_some_and(|parent| parent.kind() == "module")
72                {
73                    for name in python_bound_names(node, source) {
74                        copying.forget(&name);
75                    }
76                    python_bind_assignment(node, source, &mut bound);
77                } else {
78                    python_unbind(python_bound_names(node, source), &mut bound, &mut copying);
79                }
80            }
81            "augmented_assignment" | "delete_statement" => {
82                let target = if node.kind() == "augmented_assignment" {
83                    node.child_by_field_name("left")
84                } else {
85                    node.named_child(0)
86                };
87                if python_targets_path(target, source) {
88                    appended.clear();
89                }
90                python_unbind(python_bound_names(node, source), &mut bound, &mut copying);
91            }
92            "for_statement" | "with_item" | "except_clause" | "named_expression" => {
93                python_unbind(python_bound_names(node, source), &mut bound, &mut copying);
94            }
95            "import_statement" | "import_from_statement" => {
96                python_unbind(python_bound_names(node, source), &mut bound, &mut copying);
97                if node
98                    .parent()
99                    .is_some_and(|parent| parent.kind() == "module")
100                {
101                    python_bind_pkgutil_import(node, source, &mut copying);
102                }
103                // The imported names are bindings, not uses of a trusted callee.
104                return WalkControl::SkipChildren;
105            }
106            "call" => {
107                let Some(function) = node.child_by_field_name("function") else {
108                    return WalkControl::Continue;
109                };
110                let Some(path) = python_static_attribute_path(function) else {
111                    return WalkControl::Continue;
112                };
113                let [object, method] = path.as_slice() else {
114                    return WalkControl::Continue;
115                };
116                if !python_names_path(Some(*object), source) {
117                    return WalkControl::Continue;
118                }
119                let method = method.utf8_text(source.as_bytes()).unwrap_or("");
120                if !matches!(method, "append" | "insert" | "extend") {
121                    appended.clear();
122                    return WalkControl::Continue;
123                }
124                let directories = python_path_appended_dirs_in_call(node, source, &bound);
125                if !python_extension_statement_runs(node, source, &bound, &directories) {
126                    appended.clear();
127                    return WalkControl::Continue;
128                }
129                for directory in directories
130                    .into_iter()
131                    .filter_map(PathValue::relative_directory)
132                {
133                    if !appended.contains(&directory) {
134                        appended.push(directory);
135                    }
136                }
137            }
138            _ => {}
139        }
140        if escaped {
141            WalkControl::Break
142        } else {
143            WalkControl::Continue
144        }
145    });
146    if escaped {
147        return Vec::new();
148    }
149    appended
150}
151
152/// Whether one `__path__` occurrence is a use this reader models: the receiver
153/// of a method call (modelled or invalidating in the call arm), an assignment or
154/// `del` target (invalidating in those arms), an element read, a read-only
155/// membership test or loop over it, or the argument of a call whose result
156/// replaces `__path__`. Any other use can hand the list to code that changes it
157/// under another name.
158fn python_path_use_is_modelled(identifier: Node<'_>, source: &str, copying: &CopyingNames) -> bool {
159    let Some(parent) = identifier.parent() else {
160        return false;
161    };
162    if parent.kind() == "argument_list" {
163        // `__path__ = extend_path(__path__, __name__)`: pkgutil.extend_path returns
164        // a new list and keeps no reference to its argument, and the assignment
165        // arm drops every recorded directory. Any other callee could keep the list
166        // and return it, so a later change under the kept name would go unseen.
167        return parent.named_child(0) == Some(identifier)
168            && parent
169                .parent()
170                .filter(|call| call.kind() == "call")
171                .is_some_and(|call| {
172                    python_calls_extend_path(call, source, copying)
173                        && call.parent().is_some_and(|holder| {
174                            holder.kind() == "assignment"
175                                && holder.child_by_field_name("right") == Some(call)
176                                && python_names_path(holder.child_by_field_name("left"), source)
177                        })
178                });
179    }
180    if parent.kind() == "attribute" {
181        return parent.child_by_field_name("object") == Some(identifier)
182            && parent.parent().is_some_and(|call| {
183                call.kind() == "call" && call.child_by_field_name("function") == Some(parent)
184            });
185    }
186    if parent.kind() == "comparison_operator" {
187        let mut cursor = parent.walk();
188        let membership = parent
189            .children_by_field_name("operators", &mut cursor)
190            .all(|operator| matches!(operator.kind(), "in" | "not in"));
191        return membership && parent.named_child(0) != Some(identifier);
192    }
193    if parent.kind() == "for_statement" {
194        return parent.child_by_field_name("right") == Some(identifier);
195    }
196    let mut top = identifier;
197    while let Some(up) = top
198        .parent()
199        .filter(|up| up.kind() == "subscript" && up.child_by_field_name("value") == Some(top))
200    {
201        top = up;
202    }
203    let Some(holder) = top.parent() else {
204        return false;
205    };
206    match holder.kind() {
207        "assignment" | "augmented_assignment" => {
208            holder.child_by_field_name("left") == Some(top) || top != identifier
209        }
210        "delete_statement" => holder.named_child(0) == Some(top),
211        "pattern_list" | "tuple_pattern" | "list_pattern" | "as_pattern_target" => false,
212        "for_statement" => holder.child_by_field_name("left") != Some(top) && top != identifier,
213        // A subscript elsewhere reads one element, which cannot change the list.
214        _ => top != identifier,
215    }
216}
217
218#[derive(Clone, Copy, Debug, PartialEq, Eq)]
219enum CopyingCallee {
220    PkgutilModule,
221    ExtendPath,
222}
223
224/// The callees this reader trusts to return a new list without keeping their
225/// argument: module-level bindings of `pkgutil` and its `extend_path`, and the
226/// builtin `__import__` while nothing in the initializer rebinds it. A trusted
227/// name stays trusted only while every reference to it is that call. Any other
228/// reference (an attribute write, an argument, a mention in a function body)
229/// can change the one shared pkgutil module or its function, so it taints every
230/// route to them for the rest of the initializer: other aliases, a later
231/// import, and `__import__('pkgutil')`. Reaching pkgutil dynamically
232/// (`importlib`, `sys.modules`, another module's code) is outside what a reader
233/// of this file can see, like `sys.modules[__name__].__path__`.
234#[derive(Default)]
235struct CopyingNames {
236    callees: HashMap<String, CopyingCallee>,
237    import_shadowed: bool,
238    pkgutil_tainted: bool,
239    /// Names mentioned in a function, class or lambda body. Python resolves them
240    /// when the body runs, which can be after a later import binds them, so such
241    /// a name never becomes a trusted callee.
242    body_names: HashSet<String>,
243}
244
245impl CopyingNames {
246    /// A rebinding replaces the name only; the module it named is unchanged.
247    fn forget(&mut self, name: &str) {
248        self.callees.remove(name);
249        if name == "__import__" {
250            self.import_shadowed = true;
251        }
252    }
253
254    fn taint_pkgutil(&mut self) {
255        self.callees.clear();
256        self.pkgutil_tainted = true;
257    }
258}
259
260fn python_unbind(
261    names: Vec<String>,
262    bound: &mut HashMap<String, PathValue>,
263    copying: &mut CopyingNames,
264) {
265    for name in names {
266        bound.remove(&name);
267        copying.forget(&name);
268    }
269}
270
271/// Forget a trusted callee when one reference to its name is not the trusted
272/// call: `p.extend_path(...)` for a `pkgutil` module, `e(...)` for an
273/// `extend_path` function, or a call of `__import__`. An attribute's member name
274/// is not a reference to a binding, except that writing a `__import__` member
275/// (`builtins.__import__ = ...`) can replace the builtin.
276fn python_check_copying_use(identifier: Node<'_>, source: &str, copying: &mut CopyingNames) {
277    let Ok(name) = identifier.utf8_text(source.as_bytes()) else {
278        return;
279    };
280    let parent = identifier.parent();
281    let called = parent.is_some_and(|call| {
282        call.kind() == "call" && call.child_by_field_name("function") == Some(identifier)
283    });
284    if name == "__import__" {
285        match parent.filter(|_| called) {
286            None => copying.forget(name),
287            // `__import__('pkgutil')` reaches the shared pkgutil module, so any use
288            // of its result other than the trusted call can change it.
289            Some(call) => {
290                if python_imports_pkgutil(call, source)
291                    && !python_is_extend_path_call_object(call, source)
292                {
293                    copying.taint_pkgutil();
294                }
295            }
296        }
297        return;
298    }
299    let Some(callee) = copying.callees.get(name).copied() else {
300        return;
301    };
302    let Some(parent) = parent else {
303        return;
304    };
305    if parent.kind() == "attribute" && parent.child_by_field_name("attribute") == Some(identifier) {
306        return;
307    }
308    let trusted = match callee {
309        CopyingCallee::ExtendPath => called,
310        CopyingCallee::PkgutilModule => python_is_extend_path_call_object(identifier, source),
311    };
312    if !trusted {
313        copying.taint_pkgutil();
314    }
315}
316
317/// Whether `node` is the receiver of a called `.extend_path`: `node.extend_path(...)`.
318fn python_is_extend_path_call_object(node: Node<'_>, source: &str) -> bool {
319    node.parent().is_some_and(|attribute| {
320        attribute.kind() == "attribute"
321            && attribute.child_by_field_name("object") == Some(node)
322            && attribute
323                .child_by_field_name("attribute")
324                .and_then(|member| member.utf8_text(source.as_bytes()).ok())
325                == Some("extend_path")
326            && attribute.parent().is_some_and(|call| {
327                call.kind() == "call" && call.child_by_field_name("function") == Some(attribute)
328            })
329    })
330}
331
332/// Whether `call` is `__import__('pkgutil')`.
333fn python_imports_pkgutil(call: Node<'_>, source: &str) -> bool {
334    let importer = call
335        .child_by_field_name("function")
336        .is_some_and(|function| {
337            function.kind() == "identifier"
338                && function.utf8_text(source.as_bytes()).ok() == Some("__import__")
339        });
340    let module = call.child_by_field_name("arguments").and_then(|arguments| {
341        let mut cursor = arguments.walk();
342        let arguments: Vec<Node<'_>> = arguments.named_children(&mut cursor).collect();
343        match arguments.as_slice() {
344            [only] => python_plain_string_literal(*only, source),
345            _ => None,
346        }
347    });
348    importer && module == Some("pkgutil")
349}
350
351/// A function, class or lambda body runs when called, which this reader does not
352/// follow. One that names pkgutil, `extend_path`, `__import__` or a trusted
353/// callee can change the shared pkgutil module whenever it runs, and its names
354/// resolve at that time, so none of them can become trusted later either.
355fn python_forget_names_in_body(node: Node<'_>, source: &str, copying: &mut CopyingNames) {
356    walk_named_tree_preorder(node, false, |child| {
357        if child.kind() == "identifier"
358            && let Ok(name) = child.utf8_text(source.as_bytes())
359        {
360            if name == "__import__" {
361                copying.forget(name);
362            }
363            if matches!(name, "pkgutil" | "extend_path" | "__import__")
364                || copying.callees.contains_key(name)
365            {
366                copying.taint_pkgutil();
367            }
368            copying.body_names.insert(name.to_owned());
369        }
370        WalkControl::Continue
371    });
372}
373
374/// Record a module-level `import pkgutil [as p]` or
375/// `from pkgutil import extend_path [as e]`.
376fn python_bind_pkgutil_import(statement: Node<'_>, source: &str, copying: &mut CopyingNames) {
377    if copying.pkgutil_tainted {
378        return;
379    }
380    let text = |node: Node<'_>| node.utf8_text(source.as_bytes()).ok();
381    let (imported, callee) = if statement.kind() == "import_from_statement" {
382        if statement.child_by_field_name("module_name").and_then(text) != Some("pkgutil") {
383            return;
384        }
385        ("extend_path", CopyingCallee::ExtendPath)
386    } else {
387        ("pkgutil", CopyingCallee::PkgutilModule)
388    };
389    let mut cursor = statement.walk();
390    for name in statement.children_by_field_name("name", &mut cursor) {
391        let (target, binding) = if name.kind() == "aliased_import" {
392            (
393                name.child_by_field_name("name"),
394                name.child_by_field_name("alias"),
395            )
396        } else {
397            (Some(name), Some(name))
398        };
399        if target.and_then(text) == Some(imported)
400            && let Some(binding) = binding.and_then(text)
401            && !copying.body_names.contains(binding)
402        {
403            copying.callees.insert(binding.to_owned(), callee);
404        }
405    }
406}
407
408/// Whether `call` calls pkgutil's `extend_path` through a module-level binding,
409/// `p.extend_path` for a bound `pkgutil` module, or `__import__('pkgutil')`.
410fn python_calls_extend_path(call: Node<'_>, source: &str, copying: &CopyingNames) -> bool {
411    let text = |node: Node<'_>| node.utf8_text(source.as_bytes()).ok();
412    let Some(function) = call.child_by_field_name("function") else {
413        return false;
414    };
415    match function.kind() {
416        "identifier" => {
417            text(function).and_then(|name| copying.callees.get(name))
418                == Some(&CopyingCallee::ExtendPath)
419        }
420        "attribute" => {
421            if function.child_by_field_name("attribute").and_then(text) != Some("extend_path") {
422                return false;
423            }
424            let Some(object) = function.child_by_field_name("object") else {
425                return false;
426            };
427            match object.kind() {
428                "identifier" => {
429                    text(object).and_then(|name| copying.callees.get(name))
430                        == Some(&CopyingCallee::PkgutilModule)
431                }
432                "call" => {
433                    python_imports_pkgutil(object, source)
434                        && !copying.import_shadowed
435                        && !copying.pkgutil_tainted
436                }
437                _ => false,
438            }
439        }
440        _ => false,
441    }
442}
443
444/// Whether any identifier under `node` is `__path__`.
445fn python_subtree_names_path(node: Node<'_>, source: &str) -> bool {
446    let mut found = false;
447    walk_named_tree_preorder(node, false, |child| {
448        if child.kind() == "identifier" && python_names_path(Some(child), source) {
449            found = true;
450            return WalkControl::Break;
451        }
452        WalkControl::Continue
453    });
454    found
455}
456
457/// The identifiers a binding construct writes in the module scope. Target walks
458/// visit destructuring patterns but never treat an attribute or subscript's
459/// object as a binding.
460fn python_bound_names(node: Node<'_>, source: &str) -> Vec<String> {
461    let mut targets = Vec::new();
462    match node.kind() {
463        "assignment" => {
464            let mut current = node;
465            loop {
466                if let Some(left) = current.child_by_field_name("left") {
467                    targets.push(left);
468                }
469                match current.child_by_field_name("right") {
470                    Some(right) if right.kind() == "assignment" => current = right,
471                    _ => break,
472                }
473            }
474        }
475        "augmented_assignment" | "for_statement" => {
476            targets.extend(node.child_by_field_name("left"));
477        }
478        "named_expression" | "function_definition" | "class_definition" => {
479            targets.extend(node.child_by_field_name("name"));
480        }
481        "except_clause" => targets.extend(node.child_by_field_name("alias")),
482        "with_item" => {
483            let mut cursor = node.walk();
484            targets.extend(
485                node.named_children(&mut cursor)
486                    .filter(|child| child.kind() == "as_pattern")
487                    .filter_map(|child| child.child_by_field_name("alias")),
488            );
489        }
490        "delete_statement" => targets.extend(node.named_child(0)),
491        "import_statement" | "import_from_statement" => {
492            let mut cursor = node.walk();
493            for child in node.children_by_field_name("name", &mut cursor) {
494                let name = if child.kind() == "aliased_import" {
495                    child.child_by_field_name("alias")
496                } else {
497                    child.named_child(0)
498                };
499                targets.extend(name);
500            }
501        }
502        _ => {}
503    }
504    let mut names = Vec::new();
505    while let Some(target) = targets.pop() {
506        match target.kind() {
507            "identifier" => {
508                if let Ok(name) = target.utf8_text(source.as_bytes()) {
509                    names.push(name.to_owned());
510                }
511            }
512            "pattern_list" | "tuple_pattern" | "list_pattern" | "expression_list"
513            | "as_pattern" | "as_pattern_target" => {
514                let mut cursor = target.walk();
515                targets.extend(target.named_children(&mut cursor));
516            }
517            _ => {}
518        }
519    }
520    names
521}
522
523fn python_targets_path(target: Option<Node<'_>>, source: &str) -> bool {
524    let Some(mut target) = target else {
525        return false;
526    };
527    while target.kind() == "subscript" {
528        let Some(value) = target.child_by_field_name("value") else {
529            return false;
530        };
531        target = value;
532    }
533    python_names_path(Some(target), source)
534}
535
536/// Whether `node` -- when it is one -- is the marker's own `__path__` name.
537fn python_names_path(node: Option<Node<'_>>, source: &str) -> bool {
538    node.is_some_and(|node| {
539        node.kind() == "identifier" && node.utf8_text(source.as_bytes()).ok() == Some("__path__")
540    })
541}
542
543/// Whether the statement holding one `__path__` extension call is one the
544/// interpreter runs before the imports the marker serves.
545///
546/// Module-level statements run. Inside a branch they run only when the branch
547/// tests exactly the fact that makes the extension necessary: the pinned
548/// shim's membership guard `if _impl not in __path__:`, whose tested name
549/// denotes the directory the branch appends. Every other branch shape is not
550/// evidence -- `if False:` never runs the append, a predicate the reader
551/// cannot answer says nothing, and a loop, `try` or `else` body is no more
552/// proven than a comparison the guard does not state -- so the marker leaves
553/// an honest resolution gap rather than a binding. The same rule applies to
554/// the guard's own position: the append runs only when the guard does, so a
555/// guard that is itself inside another branch -- `if False:` around the pinned
556/// membership guard, say -- states nothing and leaves the gap (#3506).
557fn python_extension_statement_runs(
558    call: Node<'_>,
559    source: &str,
560    bound: &HashMap<String, PathValue>,
561    directories: &[PathValue],
562) -> bool {
563    let Some(statement) = call.parent() else {
564        return false;
565    };
566    if statement.kind() != "expression_statement" {
567        return false;
568    }
569    let Some(container) = statement.parent() else {
570        return false;
571    };
572    if container.kind() == "module" {
573        return true;
574    }
575    // The call's statement is a guarded body only when its own block is that
576    // `if`'s consequence: `else`, `elif`, loop, `try` and `with` bodies all
577    // share the block shape, and none of them is this guard. A one-line
578    // `if x: y()` puts the statement directly under the `if` instead.
579    let guarded = if container.kind() == "block" {
580        container.parent().filter(|parent| {
581            parent.kind() == "if_statement"
582                && parent.child_by_field_name("consequence") == Some(container)
583        })
584    } else if container.kind() == "if_statement"
585        && container.child_by_field_name("consequence") == Some(statement)
586    {
587        Some(container)
588    } else {
589        None
590    };
591    let Some(guarded) = guarded else {
592        return false;
593    };
594    // Execution certainty has to hold for the guard's ancestors too, and the
595    // only enclosing scope this reader can vouch for is the marker's own body:
596    // a guard under another branch, loop, `with` or `try` is no more proven
597    // than the shape that holds it.
598    if guarded.parent().map(|parent| parent.kind()) != Some("module") {
599        return false;
600    }
601    let Some(condition) = guarded.child_by_field_name("condition") else {
602        return false;
603    };
604    python_condition_justifies_path_extension(condition, source, bound, directories)
605}
606
607/// Whether a branch condition tests exactly the fact that makes the extension
608/// necessary: the directory the branch appends is the one missing from
609/// `__path__`, which is what the pinned shim's `if _impl not in __path__:`
610/// states.
611fn python_condition_justifies_path_extension(
612    condition: Node<'_>,
613    source: &str,
614    bound: &HashMap<String, PathValue>,
615    directories: &[PathValue],
616) -> bool {
617    if condition.kind() != "comparison_operator" {
618        return false;
619    }
620    let mut cursor = condition.walk();
621    let operators = condition
622        .children_by_field_name("operators", &mut cursor)
623        .map(|operator| operator.kind())
624        .collect::<Vec<_>>();
625    if operators.as_slice() != ["not in"] {
626        return false;
627    }
628    // The grammar gives a comparison no operand fields: its named children are
629    // the operands in source order.
630    let mut cursor = condition.walk();
631    let operands: Vec<Node<'_>> = condition.named_children(&mut cursor).collect();
632    let [left, right] = operands.as_slice() else {
633        return false;
634    };
635    if !python_names_path(Some(*right), source) {
636        return false;
637    }
638    python_path_expression_dir(*left, source, bound)
639        .is_some_and(|directory| directories.contains(&directory))
640}
641
642/// Bind the names one `name = value` statement assigns -- and every name of a
643/// chained `a = b = value` -- to the directory the value denotes. A value no
644/// shape answers leaves the name unbound, so a later use of it contributes no
645/// evidence either.
646fn python_bind_assignment(
647    assignment: Node<'_>,
648    source: &str,
649    bound: &mut HashMap<String, PathValue>,
650) {
651    let mut targets: Vec<Node<'_>> = Vec::new();
652    let mut current = assignment;
653    loop {
654        let (Some(left), Some(right)) = (
655            current.child_by_field_name("left"),
656            current.child_by_field_name("right"),
657        ) else {
658            return;
659        };
660        targets.push(left);
661        if right.kind() != "assignment" {
662            let resolved = python_path_expression_dir(right, source, bound);
663            for target in targets {
664                if target.kind() != "identifier" {
665                    continue;
666                }
667                let Ok(name) = target.utf8_text(source.as_bytes()) else {
668                    continue;
669                };
670                match &resolved {
671                    Some(directory) => {
672                        bound.insert(name.to_owned(), directory.clone());
673                    }
674                    None => {
675                        bound.remove(name);
676                    }
677                }
678            }
679            return;
680        }
681        current = right;
682    }
683}
684
685/// The directories one `__path__.append(...)`, `__path__.insert(...)` or
686/// `__path__.extend(...)` call adds.
687///
688/// The call has to be a method of the marker's own `__path__`, which is the
689/// name the interpreter gives the search path it is building; a same-named
690/// attribute of anything else is not that list.
691fn python_path_appended_dirs_in_call(
692    call: Node<'_>,
693    source: &str,
694    bound: &HashMap<String, PathValue>,
695) -> Vec<PathValue> {
696    let Some(function) = call.child_by_field_name("function") else {
697        return Vec::new();
698    };
699    let Some(path) = python_static_attribute_path(function) else {
700        return Vec::new();
701    };
702    let [object, method] = path.as_slice() else {
703        return Vec::new();
704    };
705    if object.utf8_text(source.as_bytes()).ok() != Some("__path__") {
706        return Vec::new();
707    }
708    let Ok(method) = method.utf8_text(source.as_bytes()) else {
709        return Vec::new();
710    };
711    let Some(arguments) = call.child_by_field_name("arguments") else {
712        return Vec::new();
713    };
714    let mut cursor = arguments.walk();
715    let arguments: Vec<Node<'_>> = arguments.named_children(&mut cursor).collect();
716    let named: Vec<Node<'_>> = match method {
717        "append" => arguments.first().copied().into_iter().collect(),
718        "insert" => arguments.get(1).copied().into_iter().collect(),
719        // `extend` adds the elements of its iterable. Only a list or tuple
720        // literal states them; any other iterable, including a path string
721        // (whose elements are its characters), adds entries this reader cannot
722        // name, so none is recorded.
723        "extend" => match arguments.first() {
724            Some(literal) if matches!(literal.kind(), "list" | "tuple") => {
725                let mut cursor = literal.walk();
726                literal.named_children(&mut cursor).collect()
727            }
728            _ => Vec::new(),
729        },
730        _ => Vec::new(),
731    };
732    named
733        .into_iter()
734        .filter_map(|node| python_path_expression_dir(node, source, bound))
735        .collect()
736}
737
738// Parents produced by dirname(anchor) are distinct from a literal ".." suffix:
739// dirname(join(anchor, "..")) removes that suffix before any normalization.
740#[derive(Clone, Debug, PartialEq, Eq)]
741enum PathValue {
742    Anchored { parents: usize, suffix: PathBuf },
743    Literal(PathBuf),
744}
745impl PathValue {
746    fn dirname(self) -> Option<Self> {
747        match self {
748            Self::Anchored {
749                mut parents,
750                mut suffix,
751            } => {
752                if !suffix.pop() {
753                    parents += 1;
754                }
755                Some(Self::Anchored { parents, suffix })
756            }
757            Self::Literal(path) => path
758                .parent()
759                .map(|parent| Self::Literal(parent.to_path_buf())),
760        }
761    }
762    fn join(self, tail: Self) -> Self {
763        match (self, tail) {
764            // The symbolic anchor stands for an absolute package location.
765            (_, anchored @ Self::Anchored { .. }) => anchored,
766            (_, Self::Literal(path))
767                if path.has_root()
768                    || path
769                        .components()
770                        .any(|part| matches!(part, Component::Prefix(_))) =>
771            {
772                Self::Literal(path)
773            }
774            (
775                Self::Anchored {
776                    parents,
777                    mut suffix,
778                },
779                Self::Literal(path),
780            ) => {
781                suffix.push(path);
782                Self::Anchored { parents, suffix }
783            }
784            (Self::Literal(mut head), Self::Literal(tail)) => {
785                head.push(tail);
786                Self::Literal(head)
787            }
788        }
789    }
790    fn normalized(self) -> Self {
791        let (anchored, mut parents, suffix) = match self {
792            Self::Anchored { parents, suffix } => (true, parents, suffix),
793            Self::Literal(path)
794                if path.has_root()
795                    || path
796                        .components()
797                        .any(|part| matches!(part, Component::Prefix(_))) =>
798            {
799                return Self::Literal(path.normalize());
800            }
801            Self::Literal(path) => (false, 0, path),
802        };
803        let mut normalized = PathBuf::new();
804        for component in suffix.components() {
805            match component {
806                Component::CurDir => {}
807                Component::ParentDir => {
808                    if !normalized.pop() {
809                        parents += 1;
810                    }
811                }
812                Component::Normal(name) => normalized.push(name),
813                Component::Prefix(_) | Component::RootDir => {
814                    unreachable!("symbolic suffix is relative")
815                }
816            }
817        }
818        if anchored {
819            Self::Anchored {
820                parents,
821                suffix: normalized,
822            }
823        } else {
824            Self::Literal(Self::relative_path(parents, normalized))
825        }
826    }
827    fn relative_path(parents: usize, suffix: PathBuf) -> PathBuf {
828        let mut relative = PathBuf::new();
829        for _ in 0..parents {
830            relative.push("..");
831        }
832        relative.push(suffix);
833        if relative.as_os_str().is_empty() {
834            relative.push(".");
835        }
836        relative
837    }
838    fn relative_directory(self) -> Option<PathBuf> {
839        match self.normalized() {
840            Self::Anchored { parents, suffix } => Some(Self::relative_path(parents, suffix)),
841            Self::Literal(_) => None,
842        }
843    }
844}
845
846/// Evaluate structured path expressions without recursive Rust calls. Only
847/// anchored values can be made absolute without knowing the process CWD.
848fn python_path_expression_dir(
849    node: Node<'_>,
850    source: &str,
851    bound: &HashMap<String, PathValue>,
852) -> Option<PathValue> {
853    #[derive(Clone, Copy, Debug)]
854    enum Operation {
855        Dirname,
856        Join,
857        Normalize,
858        Absolute,
859    }
860    #[derive(Debug)]
861    enum Step<'tree> {
862        Value(Node<'tree>),
863        Apply(Operation, usize),
864    }
865    let mut pending = vec![Step::Value(node)];
866    let mut values: Vec<PathValue> = Vec::new();
867    while let Some(step) = pending.pop() {
868        match step {
869            Step::Value(node) => {
870                if let Some(literal) = python_plain_string_literal(node, source) {
871                    values.push(PathValue::Literal(PathBuf::from(literal)));
872                    continue;
873                }
874                match node.kind() {
875                    "identifier" => {
876                        values.push(bound.get(node.utf8_text(source.as_bytes()).ok()?)?.clone())
877                    }
878                    "call" => {
879                        let path =
880                            python_static_attribute_path(node.child_by_field_name("function")?)?;
881                        let names = path
882                            .iter()
883                            .map(|part| part.utf8_text(source.as_bytes()).ok())
884                            .collect::<Option<Vec<_>>>()?;
885                        let operation = match names.as_slice() {
886                            ["os", "path", "dirname"] => Operation::Dirname,
887                            ["os", "path", "join"] => Operation::Join,
888                            ["os", "path", "normpath"] => Operation::Normalize,
889                            ["os", "path", "abspath"] => Operation::Absolute,
890                            _ => return None,
891                        };
892                        let arguments = node.child_by_field_name("arguments")?;
893                        let mut cursor = arguments.walk();
894                        let mut arguments =
895                            arguments.named_children(&mut cursor).collect::<Vec<_>>();
896                        if arguments.is_empty() {
897                            return None;
898                        }
899                        if !matches!(operation, Operation::Join) {
900                            arguments.truncate(1);
901                        }
902                        pending.push(Step::Apply(operation, arguments.len()));
903                        pending.extend(arguments.into_iter().rev().map(Step::Value));
904                    }
905                    _ => return None,
906                }
907            }
908            Step::Apply(operation, count) => {
909                let arguments = values.split_off(
910                    values
911                        .len()
912                        .checked_sub(count)
913                        .expect("expression operands"),
914                );
915                let mut arguments = arguments.into_iter();
916                let first = arguments.next().expect("path operation argument");
917                values.push(match operation {
918                    Operation::Dirname => first.dirname()?,
919                    Operation::Join => arguments.fold(first, PathValue::join),
920                    Operation::Normalize => first.normalized(),
921                    Operation::Absolute => match first {
922                        anchored @ PathValue::Anchored { .. } => anchored.normalized(),
923                        PathValue::Literal(path) if path.is_absolute() => {
924                            PathValue::Literal(path.normalize())
925                        }
926                        PathValue::Literal(_) => return None,
927                    },
928                });
929            }
930        }
931    }
932    assert_eq!(values.len(), 1, "one path expression result");
933    values.pop()
934}
935
936#[cfg(test)]
937mod tests {
938    use super::appended_path_dirs;
939    use brokk_bifrost_test_macros::hardwon;
940    use std::path::PathBuf;
941
942    fn dirs(source: &str) -> Vec<PathBuf> {
943        let mut parser = tree_sitter::Parser::new();
944        parser
945            .set_language(&tree_sitter_python::LANGUAGE.into())
946            .unwrap();
947        let tree = parser.parse(source, None).unwrap();
948        assert!(!tree.root_node().has_error(), "{source}");
949        appended_path_dirs(tree.root_node(), source)
950    }
951
952    #[test]
953    #[hardwon(
954        "#9077: package path extensions trusted conditional rebinding and ignored __path__ removals"
955    )]
956    fn unproven_bindings_do_not_establish_extensions() {
957        let prefix = "import os\n_dir = os.path.dirname(__file__)\n";
958        let cases = [
959            "if False:\n    _dir = os.path.join(_dir, 'other')\n",
960            "for _dir in []:\n    pass\n",
961            "with open(__file__) as _dir:\n    pass\n",
962            "(_dir := os.path.join(_dir, 'other'))\n",
963            "_dir += 'other'\n",
964            "def _dir():\n    pass\n",
965            "class _dir:\n    pass\n",
966        ];
967        for case in cases {
968            let source = format!("{prefix}{case}__path__.append(_dir)\n");
969            assert!(dirs(&source).is_empty(), "{case}");
970        }
971    }
972
973    #[test]
974    #[hardwon(
975        "#9077: package path extensions trusted conditional rebinding and ignored __path__ removals"
976    )]
977    fn path_removals_invalidate_prior_extensions() {
978        let prefix = "import os\n_dir = os.path.join(os.path.dirname(__file__), 'impl')\n__path__.append(_dir)\n";
979        for removal in [
980            "__path__.clear()",
981            "__path__.remove(_dir)",
982            "__path__.pop()",
983            "del __path__[0]",
984            "__path__[0] = _dir",
985            "__path__ += [_dir]",
986        ] {
987            assert!(
988                dirs(&format!("{prefix}{removal}\n")).is_empty(),
989                "{removal}"
990            );
991        }
992        assert_eq!(
993            dirs(&format!("__path__.clear()\n{prefix}")),
994            vec![PathBuf::from("impl")]
995        );
996        assert_eq!(
997            dirs(
998                "import os\n_impl = os.path.join(os.path.dirname(__file__), 'impl')\nif _impl not in __path__:\n    __path__.append(_impl)\n"
999            ),
1000            vec![PathBuf::from("impl")]
1001        );
1002    }
1003
1004    #[test]
1005    #[hardwon(
1006        "#9077: an alias or callee of __path__ could remove a recorded package path extension"
1007    )]
1008    fn escaped_path_list_leaves_extensions_unproven() {
1009        let prefix = "import os\n_dir = os.path.join(os.path.dirname(__file__), 'impl')\n";
1010        for (before, after) in [
1011            ("paths = __path__\n", "paths.clear()\n"),
1012            ("", "paths = __path__\npaths.clear()\n"),
1013            ("", "forget(__path__)\n"),
1014            (
1015                "",
1016                "paths = extend_path(__path__, __name__)\npaths.clear()\n",
1017            ),
1018            ("__path__ = keep(__path__)\n", "saved.clear()\n"),
1019            (
1020                "import pkgutil\npkgutil.extend_path = keep\n__path__ = pkgutil.extend_path(__path__, __name__)\n",
1021                "saved.clear()\n",
1022            ),
1023            (
1024                "import pkgutil\nmonkey(pkgutil)\n__path__ = pkgutil.extend_path(__path__, __name__)\n",
1025                "",
1026            ),
1027            (
1028                "import pkgutil\ndef patch():\n    pkgutil.extend_path = keep\npatch()\n__path__ = pkgutil.extend_path(__path__, __name__)\n",
1029                "",
1030            ),
1031            (
1032                "__import__ = fake\n__path__ = __import__('pkgutil').extend_path(__path__, __name__)\n",
1033                "",
1034            ),
1035            (
1036                "import pkgutil as p\np.extend_path = keep\n__path__ = __import__('pkgutil').extend_path(__path__, __name__)\n",
1037                "",
1038            ),
1039            (
1040                "__import__('pkgutil').extend_path = keep\n__path__ = __import__('pkgutil').extend_path(__path__, __name__)\n",
1041                "saved.clear()\n",
1042            ),
1043            (
1044                "module = __import__('pkgutil')\n__path__ = __import__('pkgutil').extend_path(__path__, __name__)\n",
1045                "",
1046            ),
1047            (
1048                "def patch():\n    __import__('pkgutil').extend_path = keep\npatch()\nimport pkgutil\n__path__ = pkgutil.extend_path(__path__, __name__)\n",
1049                "saved.clear()\n",
1050            ),
1051            (
1052                "def patch():\n    p.extend_path = keep\nimport pkgutil as p\npatch()\n__path__ = p.extend_path(__path__, __name__)\n",
1053                "saved.clear()\n",
1054            ),
1055            (
1056                "import pkgutil\npkgutil.extend_path = keep\nimport pkgutil\n__path__ = pkgutil.extend_path(__path__, __name__)\n",
1057                "",
1058            ),
1059            (
1060                "import pkgutil\npkgutil.extend_path = keep\nfrom pkgutil import extend_path\n__path__ = extend_path(__path__, __name__)\n",
1061                "",
1062            ),
1063            (
1064                "import pkgutil as p\nimport pkgutil as q\np.extend_path = keep\n__path__ = q.extend_path(__path__, __name__)\n",
1065                "",
1066            ),
1067            (
1068                "import builtins\nbuiltins.__import__ = fake\n__path__ = __import__('pkgutil').extend_path(__path__, __name__)\n",
1069                "",
1070            ),
1071            (
1072                "from pkgutil import extend_path\nextend_path = keep\n__path__ = extend_path(__path__, __name__)\n",
1073                "saved.clear()\n",
1074            ),
1075            (
1076                "try:\n    from pkgutil import extend_path\nexcept ImportError:\n    pass\n__path__ = extend_path(__path__, __name__)\n",
1077                "",
1078            ),
1079            ("", "clear = __path__.clear\nclear()\n"),
1080            ("def forget():\n    __path__.clear()\n", "forget()\n"),
1081            ("", "(lambda: __path__.clear())()\n"),
1082            ("", "a, __path__ = 1, []\n"),
1083        ] {
1084            let source = format!("{prefix}{before}__path__.append(_dir)\n{after}");
1085            assert!(dirs(&source).is_empty(), "{source}");
1086        }
1087        // `extend` records only the elements a list or tuple literal states; a
1088        // path string extends `__path__` with its characters.
1089        assert!(dirs(&format!("{prefix}__path__.extend(_dir)\n")).is_empty());
1090        for literal in ["[_dir]", "(_dir,)"] {
1091            let source = format!("{prefix}__path__.extend({literal})\n");
1092            assert_eq!(dirs(&source), vec![PathBuf::from("impl")], "{source}");
1093        }
1094        // pkgutil's extend_path returns a new list, so replacing `__path__` with
1095        // its result before the append keeps the extension.
1096        for replacement in [
1097            "from pkgutil import extend_path\n__path__ = extend_path(__path__, __name__)\n",
1098            "from pkgutil import extend_path as grow\n__path__ = grow(__path__, __name__)\n",
1099            "import pkgutil\n__path__ = pkgutil.extend_path(__path__, __name__)\n",
1100            "__path__ = __import__('pkgutil').extend_path(__path__, __name__)\n",
1101        ] {
1102            let source = format!("{prefix}{replacement}__path__.append(_dir)\n");
1103            assert_eq!(dirs(&source), vec![PathBuf::from("impl")], "{source}");
1104        }
1105        // Reads that cannot change the list keep the extension.
1106        for read in [
1107            "first = __path__[0]\n",
1108            "for entry in __path__:\n    pass\n",
1109            "present = _dir in __path__\n",
1110        ] {
1111            let source = format!("{prefix}__path__.append(_dir)\n{read}");
1112            assert_eq!(dirs(&source), vec![PathBuf::from("impl")], "{source}");
1113        }
1114    }
1115}