1use 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
10pub 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 let mut escaped = false;
27 let mut copying = CopyingNames::default();
28 walk_named_tree_preorder(root, true, |node| {
29 match node.kind() {
30 "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, ©ing)
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 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
152fn 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 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 _ => top != identifier,
215 }
216}
217
218#[derive(Clone, Copy, Debug, PartialEq, Eq)]
219enum CopyingCallee {
220 PkgutilModule,
221 ExtendPath,
222}
223
224#[derive(Default)]
235struct CopyingNames {
236 callees: HashMap<String, CopyingCallee>,
237 import_shadowed: bool,
238 pkgutil_tainted: bool,
239 body_names: HashSet<String>,
243}
244
245impl CopyingNames {
246 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
271fn 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 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
317fn 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
332fn 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
351fn 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
374fn 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
408fn 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
444fn 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
457fn 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
536fn 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
543fn 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 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 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
607fn 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 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
642fn 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
685fn 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" => 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#[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 (_, 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
846fn 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 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 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 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}