Improper Privilege Management in ikus060/rdiffweb

@@ -396,12 +396,8 @@ def test_edit_user_with_not_existing_username(self):
         self.assertInBody("Cannot edit user `invalid`: user doesn'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")

@@ -845,21 +845,16 @@ 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)
@@ -1087,10 +1082,8 @@ def get_display_name(self, path):
         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)))

@@ -133,14 +133,14 @@ def get_repo_path(cls, path, as_user=None, refresh=False):
 
     @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):

@@ -433,6 +433,26 @@ def test_refresh_repos_with_single_repo(self):
         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):
 

@@ -167,12 +167,12 @@ def url_for(*args, **kwargs):
     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:

@@ -52,7 +52,7 @@
 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)
 
 
@@ -221,7 +221,7 @@ def setUp(self):
         # 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)
@@ -230,14 +230,13 @@ 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(
@@ -534,7 +533,7 @@ def test_status_access_denied_current_mirror(self):
             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])
 
@@ -545,7 +544,7 @@ def test_status_access_denied_rdiff_backup_data(self):
         # 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

@@ -108,7 +108,11 @@ Professional support for Rdiffweb is available by contacting [IKUS Soft](https:/
 
 # 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

@@ -98,10 +98,14 @@ 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