CVE-2022-4314: Improper Privilege Management (CWE-269)
value: Improper Privilege Management in GitHub repository ikus060/rdiffweb prior to 2.5.2.
CWE description: The product does not properly assign, modify, track, or check privileges for an actor, creating an unintended sphere of control for that actor.
CVSS v3 base score: 9.8
Published: 2022-12-12T18:15Z
Repository: https://github.com/ikus060/rdiffweb
Language: Python
Files changed: README.md, rdiffweb/controller/tests/test_page_admin_users.py, rdiffweb/core/librdiff.py, rdiffweb/core/model/_repo.py, rdiffweb/core/model/tests/test_user.py, rdiffweb/core/rdw_templating.py, rdiffweb/core/tests/test_librdiff.py, tox.ini
Commit date: 2022-11-24 14:05:17 -0500
Commit message: Block repository access when user_root directory is empty or a relative path

Diff: diff --git a/README.md b/README.md
index 0bfeef702..dfff8bdc2 100644
--- a/README.md
+++ b/README.md
@@ -106,11 +106,15 @@ Bug reports should be reported on the Rdiffweb Gitlab at https://gitlab.com/ikus
 
 Professional support for Rdiffweb is available by contacting [IKUS Soft](https://www.ikus-soft.com/en/support/#form).
 
 # Changelog
 
-## Next Rlease - 2.5.1
+## Next Rlease 2.5.2
+
+* Block repository access when user_root directory is empty or a relative path
+
+## 2.5.1 (2022-11-11)
 
 * Add support for Ubuntu Kinetic #240
 * Disable filesize for deleted files to improve page loading #241
 
 ## 2.5.0 (2022-11-09)
diff --git a/rdiffweb/controller/tests/test_page_admin_users.py b/rdiffweb/controller/tests/test_page_admin_users.py
index b437d67cc..55e8a1c32 100644
--- a/rdiffweb/controller/tests/test_page_admin_users.py
+++ b/rdiffweb/controller/tests/test_page_admin_users.py
@@ -394,16 +394,12 @@ class AbstractAdminTest(rdiffweb.test.WebCase):
         # Then the user list is displayed with an error message
         self.assertStatus(200)
         self.assertInBody("Cannot edit user `invalid`: user doesn&#39;t exists")
 
     def test_user_invalid_root(self):
-        # Delete all user's
-        for user in UserObject.query.all():
-            if user.username != self.USERNAME:
-                user.delete().commit()
         # Change the user's root
-        user = UserObject.get_user('admin')
+        user = UserObject.get_user(self.USERNAME)
         user.user_root = "/invalid"
         user.commit()
         self.getPage("/admin/users")
         self.assertInBody("Root directory not accessible!")
 
diff --git a/rdiffweb/core/librdiff.py b/rdiffweb/core/librdiff.py
index 65e90b9d4..3669d8da5 100644
--- a/rdiffweb/core/librdiff.py
+++ b/rdiffweb/core/librdiff.py
@@ -843,25 +843,20 @@ class MetadataDict(object):
 
 class RdiffRepo(object):
 
     """Represent one rdiff-backup repository."""
 
-    def __init__(self, user_root, path, encoding):
-        if isinstance(user_root, str):
-            user_root = os.fsencode(user_root)
-        if isinstance(path, str):
-            path = os.fsencode(path)
-        assert isinstance(user_root, bytes)
-        assert isinstance(path, bytes)
-        assert encoding
+    def __init__(self, full_path, encoding):
+        assert encoding, 'encoding is required'
         self._encoding = encodings.search_function(encoding)
-        assert self._encoding
-        self.path = path.strip(b"/")
-        if self.path:
-            self.full_path = os.path.normpath(os.path.join(user_root, self.path))
-        else:
-            self.full_path = os.path.normpath(user_root)
+        assert self._encoding, 'encoding must be a valid charset'
+
+        # Validate and sanitize the full_path
+        assert full_path, 'full path is required'
+        self.full_path = os.fsencode(full_path) if isinstance(full_path, str) else full_path
+        assert os.path.isabs(self.full_path), 'full_path must be absolute path'
+        self.full_path = os.path.normpath(self.full_path)
 
         # The location of rdiff-backup-data directory.
         self._data_path = os.path.join(self.full_path, RDIFF_BACKUP_DATA)
         assert isinstance(self._data_path, bytes)
         self._increment_path = os.path.join(self._data_path, INCREMENTS)
@@ -1085,14 +1080,12 @@ class RdiffRepo(object):
         Return proper display name of the given path according to repository encoding and quoted characters.
         """
         assert isinstance(path, bytes)
         path = path.strip(b'/')
         if path in [b'.', b'']:
-            # For repository we use either path if defined or the directory base name
-            if not self.path:
-                return self._decode(unquote(os.path.basename(self.full_path)))
-            return self._decode(unquote(self.path))
+            # For repository the directory base name
+            return self._decode(unquote(os.path.basename(self.full_path)))
         else:
             # For path, we use the dir name
             return self._decode(unquote(os.path.basename(path)))
 
     def remove_older(self, remove_older_than):
diff --git a/rdiffweb/core/model/_repo.py b/rdiffweb/core/model/_repo.py
index a09c0fe18..2431490ed 100644
--- a/rdiffweb/core/model/_repo.py
+++ b/rdiffweb/core/model/_repo.py
@@ -131,18 +131,18 @@ class RepoObject(Base, RdiffRepo):
         except ValueError:
             raise DoesNotExistError(path)
 
     @orm.reconstructor
     def __init_on_load__(self):
-        RdiffRepo.__init__(
-            self, self.user.user_root, self.repopath, encoding=self.encoding or RepoObject.DEFAULT_REPO_ENCODING
-        )
-
-    @property
-    def displayname(self):
-        # Repository displayName is the "repopath" too.
-        return self.repopath.strip('/')
+        # RdiffRepo required an absolute full path, When the user_root is invalid, let generate an invalid full path.
+        if not self.user.user_root:
+            full_path = os.path.join('/user_has_an_empty_user_root/', self.repopath.strip('/'))
+        elif not os.path.isabs(self.user.user_root):
+            full_path = os.path.join('/user_has_a_relative_user_root/', self.repopath.strip('/'))
+        else:
+            full_path = os.path.join(self.user.user_root, self.repopath.strip('/'))
+        RdiffRepo.__init__(self, full_path, encoding=self.encoding or RepoObject.DEFAULT_REPO_ENCODING)
 
     @property
     def name(self):
         # Repository name is the "repopath"
         return self.repopath
diff --git a/rdiffweb/core/model/tests/test_user.py b/rdiffweb/core/model/tests/test_user.py
index 497e0738c..4db6c6ddf 100644
--- a/rdiffweb/core/model/tests/test_user.py
+++ b/rdiffweb/core/model/tests/test_user.py
@@ -431,10 +431,30 @@ class UserObjectTest(rdiffweb.test.WebCase):
         userobj.commit()
         # Then the list invlaid the invalid repo and new repos
         userobj.expire()
         self.assertEqual([''], sorted([r.name for r in userobj.repo_objs]))
 
+    def test_refresh_repos_with_empty_userroot(self):
+        # Given a user with valid repositories relative to root
+        userobj = UserObject.get_user(self.USERNAME)
+        for repo in userobj.repo_objs:
+            repo.repopath = self.testcases[1:] + '/' + repo.repopath
+            repo.add().commit()
+        userobj.user_root = '/'
+        userobj.add().commit()
+        self.assertEqual(['interrupted', 'ok'], sorted([r.status[0] for r in userobj.repo_objs]))
+        # When updating it's userroot directory to an empty value
+        userobj.user_root = ''
+        userobj.add().commit()
+        UserObject.session.expire_all()
+        # Then close session
+        cherrypy.tools.db.on_end_resource()
+        # Then repo status is "broken"
+        userobj = UserObject.get_user(self.USERNAME)
+        self.assertFalse(userobj.valid_user_root())
+        self.assertEqual(['failed', 'failed'], [r.status[0] for r in userobj.repo_objs])
+
 
 class UserObjectWithAdminPassword(rdiffweb.test.WebCase):
 
     # password: test
     default_config = {'admin-password': '{SSHA}wbSK4hlEX7mtGJplFi2oN6ABm6Y3Bo1e'}
diff --git a/rdiffweb/core/rdw_templating.py b/rdiffweb/core/rdw_templating.py
index 74100666a..07f8dd87b 100644
--- a/rdiffweb/core/rdw_templating.py
+++ b/rdiffweb/core/rdw_templating.py
@@ -165,16 +165,16 @@ def url_for(*args, **kwargs):
     """
     path = ""
     for chunk in args:
         if not chunk:
             continue
-        if hasattr(chunk, 'owner') and hasattr(chunk, 'path'):
+        if hasattr(chunk, 'owner') and hasattr(chunk, 'repopath'):
             # This is a RepoObject
             path += "/"
             path += chunk.owner
             path += "/"
-            path += rdw_helpers.quote_url(chunk.path.strip(b"/"))
+            path += rdw_helpers.quote_url(chunk.repopath.strip("/"))
         elif hasattr(chunk, 'path'):
             # This is a DirEntry
             if chunk.path:
                 path += "/"
                 path += rdw_helpers.quote_url(chunk.path.strip(b"/"))
diff --git a/rdiffweb/core/tests/test_librdiff.py b/rdiffweb/core/tests/test_librdiff.py
index 6b53cef23..6b2bc7dfb 100644
--- a/rdiffweb/core/tests/test_librdiff.py
+++ b/rdiffweb/core/tests/test_librdiff.py
@@ -50,11 +50,11 @@ from rdiffweb.core.librdiff import (
 
 
 class MockRdiffRepo(RdiffRepo):
     def __init__(self):
         p = bytes(pkg_resources.resource_filename('rdiffweb.core', 'tests'), encoding='utf-8')  # @UndefinedVariable
-        RdiffRepo.__init__(self, os.path.dirname(p), os.path.basename(p), encoding='utf-8')
+        RdiffRepo.__init__(self, p, encoding='utf-8')
         self.root_path = MockDirEntry(self)
 
 
 class MockDirEntry(RdiffDirEntry):
     def __init__(self, repo):
@@ -219,27 +219,26 @@ class RdiffRepoTest(unittest.TestCase):
         self.temp_dir = tempfile.mkdtemp(prefix='rdiffweb_tests_')
         tarfile.open(testcases).extractall(self.temp_dir)
         # Define location of testcases
         self.testcases_dir = os.path.normpath(os.path.join(self.temp_dir, 'testcases'))
         self.testcases_dir = self.testcases_dir.encode('utf8')
-        self.repo = RdiffRepo(self.temp_dir, b'testcases', encoding='utf-8')
+        self.repo = RdiffRepo(os.path.join(self.temp_dir, 'testcases'), encoding='utf-8')
 
     def tearDown(self):
         shutil.rmtree(self.temp_dir.encode('utf8'), True)
 
     def test_init(self):
         self.assertEqual('testcases', self.repo.display_name)
 
     def test_init_with_absolute(self):
-        self.repo = RdiffRepo(self.temp_dir, '/testcases', encoding='utf-8')
+        self.repo = RdiffRepo(os.path.join(self.temp_dir, '/testcases'), encoding='utf-8')
         self.assertEqual('testcases', self.repo.display_name)
 
     def test_init_with_invalid(self):
-        self.repo = RdiffRepo(self.temp_dir, 'invalid', encoding='utf-8')
+        self.repo = RdiffRepo(os.path.join(self.temp_dir, 'invalid'), encoding='utf-8')
         self.assertEqual('failed', self.repo.status[0])
         self.assertEqual(None, self.repo.last_backup_date)
-        self.assertEqual(b'invalid', self.repo.path)
         self.assertEqual('invalid', self.repo.display_name)
 
     @parameterized.expand(
         [
             (
@@ -532,22 +531,22 @@ class RdiffRepoTest(unittest.TestCase):
         os.chmod(
             os.path.join(self.testcases_dir, b'rdiff-backup-data', b'current_mirror.2016-02-02T16:30:40-05:00.data'),
             0000,
         )
         # Create repo again to query status
-        self.repo = RdiffRepo(self.temp_dir, b'testcases', encoding='utf-8')
+        self.repo = RdiffRepo(os.path.join(self.temp_dir, 'testcases'), encoding='utf-8')
         status = self.repo.status
         self.assertEqual('failed', status[0])
 
     def test_status_access_denied_rdiff_backup_data(self):
         # Skip test if running as root. Because root as access to everything.
         if os.geteuid() == 0:
             return
         # Change the permissions of the files.
         os.chmod(os.path.join(self.testcases_dir, b'rdiff-backup-data'), 0000)
         # Query status.
-        self.repo = RdiffRepo(self.temp_dir, b'testcases', encoding='utf-8')
+        self.repo = RdiffRepo(os.path.join(self.temp_dir, 'testcases'), encoding='utf-8')
         status = self.repo.status
         self.assertEqual('failed', status[0])
         # Make sure history entry doesn't raise error
         list(self.repo.mirror_metadata)
 
diff --git a/tox.ini b/tox.ini
index 24924f863..2e5ae6df0 100644
--- a/tox.ini
+++ b/tox.ini
@@ -96,14 +96,18 @@ deps = isort>=5.0.1
 commands = isort --check --diff setup.py rdiffweb
 skip_install = true
 
 [flake8]
 ignore =
-  E203 # whitespace before ':'
-  E501 # line too long (86 > 79 characters)
-  W503 # line break before binary operator
-  E741 # ambiguous variable name 'I'
+  # whitespace before ':'
+  E203
+  # line too long (86 > 79 characters)
+  E501
+  # line break before binary operator
+  W503
+  # ambiguous variable name 'I'
+  E741
 filename =
   *.py
   setup.py
 max-complexity = 20
 

Vulnerable code: ## Next Rlease - 2.5.1
        # Delete all user's
        for user in UserObject.query.all():
            if user.username != self.USERNAME:
                user.delete().commit()
        user = UserObject.get_user('admin')
    def __init__(self, user_root, path, encoding):
        if isinstance(user_root, str):
            user_root = os.fsencode(user_root)
        if isinstance(path, str):
            path = os.fsencode(path)
        assert isinstance(user_root, bytes)
        assert isinstance(path, bytes)
        assert encoding
        assert self._encoding
        self.path = path.strip(b"/")
        if self.path:
            self.full_path = os.path.normpath(os.path.join(user_root, self.path))
        else:
            self.full_path = os.path.normpath(user_root)
            # For repository we use either path if defined or the directory base name
            if not self.path:
                return self._decode(unquote(os.path.basename(self.full_path)))
            return self._decode(unquote(self.path))
        RdiffRepo.__init__(
            self, self.user.user_root, self.repopath, encoding=self.encoding or RepoObject.DEFAULT_REPO_ENCODING
        )

    @property
    def displayname(self):
        # Repository displayName is the "repopath" too.
        return self.repopath.strip('/')
        if hasattr(chunk, 'owner') and hasattr(chunk, 'path'):
            path += rdw_helpers.quote_url(chunk.path.strip(b"/"))
        RdiffRepo.__init__(self, os.path.dirname(p), os.path.basename(p), encoding='utf-8')
        self.repo = RdiffRepo(self.temp_dir, b'testcases', encoding='utf-8')
        self.repo = RdiffRepo(self.temp_dir, '/testcases', encoding='utf-8')
        self.repo = RdiffRepo(self.temp_dir, 'invalid', encoding='utf-8')
        self.assertEqual(b'invalid', self.repo.path)
        self.repo = RdiffRepo(self.temp_dir, b'testcases', encoding='utf-8')
        self.repo = RdiffRepo(self.temp_dir, b'testcases', encoding='utf-8')
  E203 # whitespace before ':'
  E501 # line too long (86 > 79 characters)
  W503 # line break before binary operator
  E741 # ambiguous variable name 'I'
Fixed code: ## Next Rlease 2.5.2

* Block repository access when user_root directory is empty or a relative path

## 2.5.1 (2022-11-11)
        user = UserObject.get_user(self.USERNAME)
    def __init__(self, full_path, encoding):
        assert encoding, 'encoding is required'
        assert self._encoding, 'encoding must be a valid charset'

        # Validate and sanitize the full_path
        assert full_path, 'full path is required'
        self.full_path = os.fsencode(full_path) if isinstance(full_path, str) else full_path
        assert os.path.isabs(self.full_path), 'full_path must be absolute path'
        self.full_path = os.path.normpath(self.full_path)
            # For repository the directory base name
            return self._decode(unquote(os.path.basename(self.full_path)))
        # RdiffRepo required an absolute full path, When the user_root is invalid, let generate an invalid full path.
        if not self.user.user_root:
            full_path = os.path.join('/user_has_an_empty_user_root/', self.repopath.strip('/'))
        elif not os.path.isabs(self.user.user_root):
            full_path = os.path.join('/user_has_a_relative_user_root/', self.repopath.strip('/'))
        else:
            full_path = os.path.join(self.user.user_root, self.repopath.strip('/'))
        RdiffRepo.__init__(self, full_path, encoding=self.encoding or RepoObject.DEFAULT_REPO_ENCODING)
    def test_refresh_repos_with_empty_userroot(self):
        # Given a user with valid repositories relative to root
        userobj = UserObject.get_user(self.USERNAME)
        for repo in userobj.repo_objs:
            repo.repopath = self.testcases[1:] + '/' + repo.repopath
            repo.add().commit()
        userobj.user_root = '/'
        userobj.add().commit()
        self.assertEqual(['interrupted', 'ok'], sorted([r.status[0] for r in userobj.repo_objs]))
        # When updating it's userroot directory to an empty value
        userobj.user_root = ''
        userobj.add().commit()
        UserObject.session.expire_all()
        # Then close session
        cherrypy.tools.db.on_end_resource()
        # Then repo status is "broken"
        userobj = UserObject.get_user(self.USERNAME)
        self.assertFalse(userobj.valid_user_root())
        self.assertEqual(['failed', 'failed'], [r.status[0] for r in userobj.repo_objs])

        if hasattr(chunk, 'owner') and hasattr(chunk, 'repopath'):
            path += rdw_helpers.quote_url(chunk.repopath.strip("/"))
        RdiffRepo.__init__(self, p, encoding='utf-8')
        self.repo = RdiffRepo(os.path.join(self.temp_dir, 'testcases'), encoding='utf-8')
        self.repo = RdiffRepo(os.path.join(self.temp_dir, '/testcases'), encoding='utf-8')
        self.repo = RdiffRepo(os.path.join(self.temp_dir, 'invalid'), encoding='utf-8')
        self.repo = RdiffRepo(os.path.join(self.temp_dir, 'testcases'), encoding='utf-8')
        self.repo = RdiffRepo(os.path.join(self.temp_dir, 'testcases'), encoding='utf-8')
  # whitespace before ':'
  E203
  # line too long (86 > 79 characters)
  E501
  # line break before binary operator
  W503
  # ambiguous variable name 'I'
  E741