|
3 | 3 | # This module is part of GitPython and is released under the |
4 | 4 | # 3-Clause BSD License: https://opensource.org/license/bsd-3-clause/ |
5 | 5 |
|
| 6 | +import contextlib |
6 | 7 | from itertools import chain |
7 | 8 | import os.path as osp |
8 | 9 | from pathlib import Path |
|
30 | 31 |
|
31 | 32 |
|
32 | 33 | class TestRefs(TestBase): |
| 34 | + @contextlib.contextmanager |
33 | 35 | def _repo_with_initial_commit(self, base_dir): |
34 | 36 | repo_dir = base_dir / "repo" |
35 | 37 | repo = Repo.init(repo_dir) |
36 | 38 | (repo_dir / "file.txt").write_text("initial\n", encoding="utf-8") |
37 | 39 | repo.index.add(["file.txt"]) |
38 | 40 | repo.index.commit("initial") |
39 | | - return repo |
| 41 | + try: |
| 42 | + yield repo |
| 43 | + finally: |
| 44 | + repo.git.clear_cache() |
40 | 45 |
|
41 | 46 | def test_from_path(self): |
42 | 47 | # Should be able to create any reference directly. |
@@ -660,60 +665,84 @@ def test_refs_outside_repo(self): |
660 | 665 | def test_reference_create_rejects_path_traversal(self): |
661 | 666 | with tempfile.TemporaryDirectory() as tmp_dir: |
662 | 667 | base_dir = Path(tmp_dir) |
663 | | - repo = self._repo_with_initial_commit(base_dir) |
664 | | - outside_path = base_dir / "outside_write.txt" |
| 668 | + with self._repo_with_initial_commit(base_dir) as repo: |
| 669 | + outside_path = base_dir / "outside_write.txt" |
665 | 670 |
|
666 | | - self.assertRaises(ValueError, Reference.create, repo, "../../../outside_write.txt", "HEAD") |
667 | | - assert not outside_path.exists() |
| 671 | + self.assertRaises(ValueError, Reference.create, repo, "../../../outside_write.txt", "HEAD") |
| 672 | + assert not outside_path.exists() |
668 | 673 |
|
669 | 674 | def test_symbolic_reference_create_rejects_path_traversal(self): |
670 | 675 | with tempfile.TemporaryDirectory() as tmp_dir: |
671 | 676 | base_dir = Path(tmp_dir) |
672 | | - repo = self._repo_with_initial_commit(base_dir) |
673 | | - outside_path = base_dir / "outside_write.txt" |
| 677 | + with self._repo_with_initial_commit(base_dir) as repo: |
| 678 | + outside_path = base_dir / "outside_write.txt" |
674 | 679 |
|
675 | | - self.assertRaises(ValueError, SymbolicReference.create, repo, "../../outside_write.txt", "HEAD") |
676 | | - assert not outside_path.exists() |
| 680 | + self.assertRaises(ValueError, SymbolicReference.create, repo, "../../outside_write.txt", "HEAD") |
| 681 | + assert not outside_path.exists() |
677 | 682 |
|
678 | 683 | def test_symbolic_reference_set_reference_rejects_path_traversal(self): |
679 | 684 | with tempfile.TemporaryDirectory() as tmp_dir: |
680 | 685 | base_dir = Path(tmp_dir) |
681 | | - repo = self._repo_with_initial_commit(base_dir) |
682 | | - outside_path = base_dir / "outside_write.txt" |
| 686 | + with self._repo_with_initial_commit(base_dir) as repo: |
| 687 | + outside_path = base_dir / "outside_write.txt" |
683 | 688 |
|
684 | | - self.assertRaises(ValueError, SymbolicReference(repo, "../../outside_write.txt").set_reference, "HEAD") |
685 | | - assert not outside_path.exists() |
| 689 | + self.assertRaises(ValueError, SymbolicReference(repo, "../../outside_write.txt").set_reference, "HEAD") |
| 690 | + assert not outside_path.exists() |
686 | 691 |
|
687 | 692 | def test_symbolic_reference_rename_rejects_path_traversal(self): |
688 | 693 | with tempfile.TemporaryDirectory() as tmp_dir: |
689 | 694 | base_dir = Path(tmp_dir) |
690 | | - repo = self._repo_with_initial_commit(base_dir) |
691 | | - outside_path = base_dir / "outside_move.txt" |
692 | | - ref = SymbolicReference.create(repo, "SAFE_RENAME_SOURCE", "HEAD") |
| 695 | + with self._repo_with_initial_commit(base_dir) as repo: |
| 696 | + outside_path = base_dir / "outside_move.txt" |
| 697 | + ref = SymbolicReference.create(repo, "SAFE_RENAME_SOURCE", "HEAD") |
693 | 698 |
|
694 | | - self.assertRaises(ValueError, ref.rename, "../../outside_move.txt") |
695 | | - assert not outside_path.exists() |
696 | | - assert Path(ref.abspath).is_file() |
| 699 | + self.assertRaises(ValueError, ref.rename, "../../outside_move.txt") |
| 700 | + assert not outside_path.exists() |
| 701 | + assert Path(ref.abspath).is_file() |
697 | 702 |
|
698 | 703 | def test_symbolic_reference_delete_rejects_path_traversal(self): |
699 | 704 | with tempfile.TemporaryDirectory() as tmp_dir: |
700 | 705 | base_dir = Path(tmp_dir) |
701 | | - repo = self._repo_with_initial_commit(base_dir) |
702 | | - outside_path = base_dir / "outside_delete.txt" |
703 | | - outside_path.write_text("do not delete\n", encoding="utf-8") |
| 706 | + with self._repo_with_initial_commit(base_dir) as repo: |
| 707 | + outside_path = base_dir / "outside_delete.txt" |
| 708 | + outside_path.write_text("do not delete\n", encoding="utf-8") |
704 | 709 |
|
705 | | - self.assertRaises(ValueError, SymbolicReference.delete, repo, "../../outside_delete.txt") |
706 | | - assert outside_path.read_text(encoding="utf-8") == "do not delete\n" |
| 710 | + self.assertRaises(ValueError, SymbolicReference.delete, repo, "../../outside_delete.txt") |
| 711 | + assert outside_path.read_text(encoding="utf-8") == "do not delete\n" |
707 | 712 |
|
708 | 713 | def test_symbolic_reference_log_append_rejects_path_traversal(self): |
709 | 714 | with tempfile.TemporaryDirectory() as tmp_dir: |
710 | 715 | base_dir = Path(tmp_dir) |
711 | | - repo = self._repo_with_initial_commit(base_dir) |
712 | | - outside_path = base_dir / "outside_reflog.txt" |
| 716 | + with self._repo_with_initial_commit(base_dir) as repo: |
| 717 | + outside_path = base_dir / "outside_reflog.txt" |
| 718 | + |
| 719 | + ref = SymbolicReference(repo, "../../../outside_reflog.txt") |
| 720 | + self.assertRaises( |
| 721 | + ValueError, ref.log_append, Commit.NULL_BIN_SHA, "do not write", repo.head.commit.binsha |
| 722 | + ) |
| 723 | + assert not outside_path.exists() |
713 | 724 |
|
714 | | - ref = SymbolicReference(repo, "../../../outside_reflog.txt") |
715 | | - self.assertRaises(ValueError, ref.log_append, Commit.NULL_BIN_SHA, "do not write", repo.head.commit.binsha) |
716 | | - assert not outside_path.exists() |
| 725 | + def test_symbolic_reference_set_reference_rejects_symlink_escape(self): |
| 726 | + with tempfile.TemporaryDirectory() as tmp_dir: |
| 727 | + base_dir = Path(tmp_dir) |
| 728 | + with self._repo_with_initial_commit(base_dir) as repo: |
| 729 | + outside_dir = base_dir / "outside_refs" |
| 730 | + outside_dir.mkdir() |
| 731 | + outside_path = outside_dir / "escaped" |
| 732 | + |
| 733 | + refs_heads_dir = Path(repo.common_dir) / "refs" / "heads" |
| 734 | + refs_heads_dir.mkdir(parents=True, exist_ok=True) |
| 735 | + symlink_path = refs_heads_dir / "link_out" |
| 736 | + try: |
| 737 | + symlink_path.symlink_to(outside_dir, target_is_directory=True) |
| 738 | + except (OSError, NotImplementedError) as ex: |
| 739 | + self.skipTest("symlinks unavailable on this platform: %s" % ex) |
| 740 | + if osp.realpath(symlink_path / "escaped") == osp.abspath(symlink_path / "escaped"): |
| 741 | + self.skipTest("realpath does not resolve directory symlinks on this platform") |
| 742 | + |
| 743 | + ref = SymbolicReference(repo, "refs/heads/link_out/escaped") |
| 744 | + self.assertRaises(ValueError, ref.set_reference, "HEAD") |
| 745 | + assert not outside_path.exists() |
717 | 746 |
|
718 | 747 | def test_remote_reference_delete_cleanup_rejects_path_traversal(self): |
719 | 748 | with tempfile.TemporaryDirectory() as tmp_dir: |
@@ -724,8 +753,10 @@ def test_remote_reference_delete_cleanup_rejects_path_traversal(self): |
724 | 753 | outside_path.write_text("do not delete\n", encoding="utf-8") |
725 | 754 |
|
726 | 755 | class GitStub: |
| 756 | + branch_called = False |
| 757 | + |
727 | 758 | def branch(self, *args): |
728 | | - pass |
| 759 | + self.branch_called = True |
729 | 760 |
|
730 | 761 | class RepoStub: |
731 | 762 | pass |
@@ -737,6 +768,7 @@ class RepoStub: |
737 | 768 | ref = RemoteReference(repo, "../../outside_remote_delete.txt", check_path=False) |
738 | 769 |
|
739 | 770 | self.assertRaises(ValueError, RemoteReference.delete, repo, ref) |
| 771 | + assert not repo.git.branch_called |
740 | 772 | assert outside_path.read_text(encoding="utf-8") == "do not delete\n" |
741 | 773 |
|
742 | 774 | def test_validity_ref_names(self): |
|