pre-create subdir if needed (POSIX, par2) (#1802)

* pre-create subdir it needed

* pre-create subdir it needed: check if already exists

* use os.makedirs() to handle subdir1/subdir2/blabla

* protect against malicous "..", and better naming

* check for Windows \ and POSIX /

* check again within path, typo and formatting

* regex: square brackets

* cleanup: only "/" can occur in par2

* cleanup: better logging

* unit test: testing of filesystem.renamer()

* if subdir specified in par2: let filesystem.renamer() do all the work

* if create_local_directories=True, then renamer() must stay within specified directory. Plus unittest for that.

* if create_local_directories=True, then renamer() must stay within specified directory. Plus unittest for that.

* more comments in code

* use filesystem.create_all_dirs(), less logging, clearer "..", and other feedback from Safihre

* make remote black happy too

* Small changes in wording of comments and error

Co-authored-by: Safihre <safihre@sabnzbd.org>

sabnzbd/filesystem.py

@@ -806,8 +806,9 @@ def get_filepath(path: str, nzo, filename: str):
 
 
 @synchronized(DIR_LOCK)
-def renamer(old: str, new: str):
-    """ Rename file/folder with retries for Win32 """
+def renamer(old: str, new: str, create_local_directories: bool = False):
+    """Rename file/folder with retries for Win32
+    Optionally alows the creation of local directories if they don't exist yet"""
     # Sanitize last part of new name
     path, name = os.path.split(new)
     new = os.path.join(path, sanitize_filename(name))
@@ -816,6 +817,19 @@ def renamer(old: str, new: str):
     if old == new:
         return
 
+    # In case we want nonexistent directories to be created, check for directory escape (forbidden)
+    if create_local_directories:
+        oldpath, _ = os.path.split(old)
+        # Check not outside directory
+        # In case of "same_file() == 1": same directory, so nothing to do
+        if same_file(oldpath, path) == 0:
+            # Outside current directory, this is most likely malicious
+            logging.error(T("Blocked attempt to create directory %s"), path)
+            raise OSError("Refusing to go outside directory")
+        elif same_file(oldpath, path) == 2:
+            # Sub-directory, so create if does not yet exist:
+            create_all_dirs(path)
+
     logging.debug('Renaming "%s" to "%s"', old, new)
     if sabnzbd.WIN32:
         retries = 10

sabnzbd/newsunpack.py

@@ -55,6 +55,7 @@ from sabnzbd.filesystem import (
     setname_from_path,
     get_ext,
     get_filename,
+    same_file,
 )
 from sabnzbd.nzbstuff import NzbObject, NzbFile
 from sabnzbd.sorting import SeriesSorter
@@ -2077,7 +2078,14 @@ def quick_check_set(set, nzo):
             if nzf.md5sum == md5pack[file]:
                 try:
                     logging.debug("Quick-check will rename %s to %s", nzf.filename, file)
-                    renamer(os.path.join(nzo.download_path, nzf.filename), os.path.join(nzo.download_path, file))
+
+                    # Note: file can and is allowed to be in a subdirectory.
+                    # Subdirectories in par2 always contain "/", not "\"
+                    renamer(
+                        os.path.join(nzo.download_path, nzf.filename),
+                        os.path.join(nzo.download_path, file),
+                        create_local_directories=True,
+                    )
                     renames[file] = nzf.filename
                     nzf.filename = file
                     result &= True

tests/test_filesystem.py

@@ -21,6 +21,9 @@ tests.test_filesystem - Testing functions in filesystem.py
 import stat
 import sys
 import os
+import random
+import shutil
+from pathlib import Path
 
 import pyfakefs.fake_filesystem_unittest as ffs
 
@@ -983,3 +986,73 @@ class TestSetPermissions(ffs.TestCase, PermissionCheckerHelper):
     def test_dir1755_umask4755_setting(self):
         # Sticky bit on directory, umask with setuid
         self._runner("1755", "4755")
+
+
+class TestRenamer:
+    # test filesystem.renamer() for different scenario's
+    def test_renamer(self):
+        # First of all, create a working directory (with a random name)
+        dirname = os.path.join(SAB_DATA_DIR, "testdir" + str(random.randint(10000, 99999)))
+        os.mkdir(dirname)
+
+        # base case: rename file within directory
+        filename = os.path.join(dirname, "myfile.txt")
+        Path(filename).touch()  # create file
+        newfilename = os.path.join(dirname, "newfile.txt")
+        filesystem.renamer(filename, newfilename)  # rename() does not return a value ...
+        assert not os.path.isfile(filename)
+        assert os.path.isfile(newfilename)
+
+        # standard behaviour: renaming (moving) into an exiting other directory *is* allowed
+        filename = os.path.join(dirname, "myfile.txt")
+        Path(filename).touch()  # create file
+        sameleveldirname = os.path.join(SAB_DATA_DIR, "othertestdir" + str(random.randint(10000, 99999)))
+        os.mkdir(sameleveldirname)
+        newfilename = os.path.join(sameleveldirname, "newfile.txt")
+        filesystem.renamer(filename, newfilename)
+        assert not os.path.isfile(filename)
+        assert os.path.isfile(newfilename)
+        shutil.rmtree(sameleveldirname)
+
+        # Default: renaming into a non-existing subdirectory not allowed
+        Path(filename).touch()  # create file
+        newfilename = os.path.join(dirname, "nonexistingsubdir", "newfile.txt")
+        try:
+            filesystem.renamer(filename, newfilename)  # rename() does not return a value ...
+        except:
+            pass
+        assert os.path.isfile(filename)
+        assert not os.path.isfile(newfilename)
+
+        # Creation of subdirectory is allowed if create_local_directories=True
+        Path(filename).touch()
+        newfilename = os.path.join(dirname, "newsubdir", "newfile.txt")
+        try:
+            filesystem.renamer(filename, newfilename, create_local_directories=True)
+        except:
+            pass
+        assert not os.path.isfile(filename)
+        assert os.path.isfile(newfilename)
+
+        # Creation of subdirectory plus deeper sudbdir is allowed if create_local_directories=True
+        Path(filename).touch()
+        newfilename = os.path.join(dirname, "newsubdir", "deepersubdir", "newfile.txt")
+        try:
+            filesystem.renamer(filename, newfilename, create_local_directories=True)
+        except:
+            pass
+        assert not os.path.isfile(filename)
+        assert os.path.isfile(newfilename)
+
+        # ... escaping the directory plus subdir creation is not allowed
+        Path(filename).touch()
+        newfilename = os.path.join(dirname, "..", "newsubdir", "newfile.txt")
+        try:
+            filesystem.renamer(filename, newfilename, create_local_directories=True)
+        except:
+            pass
+        assert os.path.isfile(filename)
+        assert not os.path.isfile(newfilename)
+
+        # Cleanup working directory
+        shutil.rmtree(dirname)