From 8f9dbd86847ee862e11ae32aafe912fc7ddccf71 Mon Sep 17 00:00:00 2001 From: Ondrej Nosek Date: Feb 18 2020 07:50:21 +0000 Subject: (new-)sources should fail with git tracked files Implements additional protection (except .gitignore warning) that prevents overwrite git tracked files when x-pkg sources is run. And also forbids uploading tracked files that may not belong to the lookaside cache. Fixes: https://pagure.io/fedpkg/issue/241 JIRA: COMPOSE-2689 Signed-off-by: Ondrej Nosek --- diff --git a/pyrpkg/__init__.py b/pyrpkg/__init__.py index 92e4693..6ee2348 100644 --- a/pyrpkg/__init__.py +++ b/pyrpkg/__init__.py @@ -46,7 +46,7 @@ from pyrpkg.errors import (HashtypeMixingError, UnknownTargetError, rpkgAuthError, rpkgError) from pyrpkg.lookaside import CGILookasideCache from pyrpkg.sources import SourcesFile -from pyrpkg.utils import cached_property, find_me, log_result +from pyrpkg.utils import cached_property, find_me, is_file_tracked, log_result from .gitignore import GitIgnore @@ -2065,6 +2065,11 @@ class Commands(object): for entry in sourcesf.entries: outfile = os.path.join(outdir, entry.file) + if is_file_tracked(outfile, outdir): + raise rpkgError( + "Error: Attempting a download '{0}' that would override a git tracked file. " + "Either remove the corresponding line from 'sources' file to keep the git " + "tracked one or 'git rm' the file to allow the download.".format(outfile)) self.lookasidecache.download( self.ns_repo_name if self.lookaside_namespaced else self.repo_name, entry.file, entry.hash, outfile, @@ -2842,6 +2847,13 @@ class Commands(object): raise rpkgError(msg) gitignore.add('/%s' % file_basename) + if is_file_tracked(f, self.path): # need full file path + raise rpkgError( + "Error: Attempting to upload a git tracked file '{0}'. Only upload files not " + "tracked by git. You can use 'git rm --cached' to untrack a file from git.\n" + "Hint: Use git for text files like the spec file, patches or helper scripts. " + "Use the lookaside cache for binary blobs, usually upstream source " + "tarballs.".format(f)) self.lookasidecache.upload( self.ns_repo_name if self.lookaside_namespaced else self.repo_name, f, file_hash, offline=offline) diff --git a/pyrpkg/utils.py b/pyrpkg/utils.py index 8513b7c..957a7e9 100644 --- a/pyrpkg/utils.py +++ b/pyrpkg/utils.py @@ -17,6 +17,7 @@ import argparse import os import sys +import git import six if six.PY3: @@ -156,3 +157,84 @@ Running Tasks: %s""" % (progname, '\n'.join(tlist))) if 'handler_reference' not in globals(): handler_reference = koji_watch_tasks_handler return handler_reference + + +def is_file_tracked(file_path, repo_path): + """ + Finds out whether input file is currently tracked in the Git repository + + :param file_path: path to a file that should be checked (relative or absolute) + :type file_path: str + :param repo_path: path to Git repository (relative or absolute) + :type repo_path: str + :return: is file staged in git repo + :rtype: bool + """ + if not file_path: + raise ValueError("empty file path") + if not repo_path: + raise ValueError("empty repo path") + + # file can be external and file like this is not tracked + relative_file_path = is_file_in_directory(file_path, repo_path) + if not relative_file_path: + return False + + # create a repo object from our path + try: + repo = git.Repo(repo_path) + except (git.InvalidGitRepositoryError, git.NoSuchPathError): + raise RuntimeError("%s is not a valid repo" % repo_path) + + if relative_file_path in repo.untracked_files: + return False + + # search entries for the file name + for entry in repo.index.entries.keys(): + (entry_file_name, _) = entry + if entry_file_name == relative_file_path: + return True + + return False + + +def is_file_in_directory(file_path, dir_path): + """ + Compares two different paths - file and dictionary. + Method doesn't check whether files exist. + :param file_path: relative or absolute path to the file + :type file_path: str + :param file_path: relative or absolute path to the directory + :type file_path: str + :return: file path relative to the dictionary if the file is inside + of the directory otherwise None + :rtype: str or None + """ + try: + real_file_path = os.path.realpath(file_path) + real_dir_path = os.path.realpath(dir_path) + except TypeError: + print( + "Wrong value of the file name(s): ({0}, {1})".format( + file_path, dir_path + ), + file=sys.stderr + ) + raise + + # file is definitely outside of the repository + if len(real_dir_path) > len(real_file_path): + return + + # this case is not defined + if real_file_path == real_dir_path: + raise ValueError("Wrong input, paths are the same.") + + # paths have common prefix that equals to dir path -> file is inside the directory + # NOTE: there is more suitable method 'os.path.commonpath' since Python 3.5. + # It returns valid path. + if os.path.commonprefix((real_file_path, real_dir_path)) == real_dir_path: + # what is the file name relative to the directory + # (length of filename is safely longer than length of directory) + return real_file_path[len(real_dir_path):].strip("/") + return diff --git a/tests/test_cli.py b/tests/test_cli.py index aa69379..425ddca 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -1279,6 +1279,7 @@ class TestUpload(LookasideCacheMock, CliTestCase): with patch('pyrpkg.lookaside.CGILookasideCache.upload', new=self.lookasidecache_upload): cli.upload() + # git tracked files are not allowed to be uploaded to lookaside cache readme_rst = os.path.join(self.cloned_repo_path, 'README.rst') self.make_changes(filename=readme_rst, content='# dockpkg', commit=True) @@ -1286,13 +1287,14 @@ class TestUpload(LookasideCacheMock, CliTestCase): with patch('sys.argv', new=cli_cmd): cli = self.new_cli() with patch('pyrpkg.lookaside.CGILookasideCache.upload', new=self.lookasidecache_upload): - cli.upload() + six.assertRaisesRegex( + self, rpkgError, + r'/README.rst.+upload files not tracked by git', + cli.upload) expected_sources_content = [ '{0} {1}'.format(self.hash_file(self.readme_patch), os.path.basename(self.readme_patch)), - '{0} {1}'.format(self.hash_file(readme_rst), - os.path.basename(readme_rst)), ] self.assertEqual(expected_sources_content, self.read_file(self.sources_file).strip().split('\n')) diff --git a/tests/test_utils.py b/tests/test_utils.py index 2f9d2cc..1ad5925 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -1,9 +1,18 @@ +import os +import tempfile import unittest import warnings import mock -from pyrpkg.utils import cached_property, log_result, warn_deprecated +from pyrpkg.utils import ( + cached_property, + is_file_in_directory, + is_file_tracked, + log_result, + warn_deprecated, +) +from utils import CommandTestCase class CachedPropertyTestCase(unittest.TestCase): @@ -196,3 +205,86 @@ class LogResultTestCase(unittest.TestCase): ] log_result(self.log_func, obj) self.assertEqual(self.logs, expected) + + +class FileInDirectoryTestCase(unittest.TestCase): + def test_is_file_in_directory(self): + expected = ( + # file, directory, expected result + ("/repo/fedpkg/sources", "/repo/fedpkg", "sources"), + ("/repo/fedpkg/subdir/sources", "/repo/fedpkg", "subdir/sources"), + ("/repo/fedpkg/sources/", "/repo/fedpkg", "sources"), + ("/repo/fedpkg/sources", "/repo/fedpkg/", "sources"), + ("/repo/file", "/", "repo/file"), + ("/file", "/", "file"), + ("sources", ".", "sources"), + ("data/file", ".", "data/file"), + ("dir/myrepo/sources", "dir/myrepo", "sources"), + + ("/sources", "/repo/fedpkg", None), + ("/", "bad_dir_name", None), + ("./", "./a", None), + ("./b", "./a/", None), + ("sources", "./dir", None), + ("f", "./dir", None), + ("file", "dir/myrepo", None), + ("a/bbb", "/a/b", None), + ("a/b", "/a/bbb", None), + ) + for num, case in enumerate(expected): + (file_path, dir_path, expected_result) = case + self.assertEqual( + is_file_in_directory(file_path, dir_path), + expected_result, + "The case num {0} has failed".format(num), + ) + + +class FileTrackedTestCase(CommandTestCase): + def test_file_outside_of_repo(self): + # check file outside of the test repo + self.assertFalse( + is_file_tracked("external_filename", self.repo_path) + ) + + def test_actual_file_in_repo(self): + # check actual file from the test repo + self.assertTrue( + is_file_tracked( + os.path.join(self.repo_path, "sources"), + self.repo_path + ) + ) + + def test_newly_created_file_in_repo(self): + # check newly created file in the test repo + temp_file = tempfile.NamedTemporaryFile(dir=self.repo_path) + self.assertFalse( + is_file_tracked(temp_file.name, self.repo_path) + ) + temp_file.close() + + def test_newly_added_file_to_stage(self): + # check newly added file to stage in the test repo + temp_file = tempfile.NamedTemporaryFile(dir=self.repo_path) + temp_file_basename = os.path.basename(temp_file.name) + self.run_cmd(["git", "add", temp_file_basename], cwd=self.repo_path) + self.assertTrue( + is_file_tracked(temp_file.name, self.repo_path) + ) + temp_file.close() + + def test_not_existing_file(self): + # check not existing file + self.assertFalse( + is_file_tracked("/file_doesnt_exist", self.repo_path) + ) + + def test_not_existing_file_should_belong_to_repo(self): + # check not existing file that should belong to the test repo + self.assertFalse( + is_file_tracked( + os.path.join(self.repo_path, "file_doesnt_exist"), + self.repo_path + ) + )