diff --git a/pyproject.toml b/pyproject.toml index bfdae9a0..ce11700e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -7,7 +7,7 @@ build-backend = "setuptools.build_meta" [project] name = "chevah-compat" -version = "1.6.1" +version = "1.7.1" maintainers = [ { name = "Adi Roiban", email = "adi.roiban@proatria.com" }, ] @@ -40,8 +40,9 @@ Homepage = "https://github.com/chevah/compat" # These are the deps required to develop. # Try to pin them as much as possible. dev = [ - "ruff ~= 0.7", + "ruff ~= 0.16", "bunch", + "psutil", "Twisted==25.5.0", "incremental", "service-identity==24.2.0", diff --git a/pythia.conf b/pythia.conf index 9d4ff03b..a26ba5ed 100644 --- a/pythia.conf +++ b/pythia.conf @@ -1,6 +1,6 @@ -PYTHON_CONFIGURATION="default@3.13.2.6899196" +PYTHON_CONFIGURATION="default@3.12.13.20260408" # This is defined as a Bash array of options to be passed to commands. -BASE_REQUIREMENTS=("chevah-brink==1.0.15" "paver==1.3.4" "wheel") +BASE_REQUIREMENTS=("chevah-brink==1.0.15" "paver==1.3.4" "wheel" "setuptools==78.1.1") # Use our private PyPi server instead of the default one set in pythia.sh. PIP_INDEX_URL="https://bin.chevah.com:20443/pypi/simple" # Use our production server instead of the GitHub releases set by default. diff --git a/release-notes.rst b/release-notes.rst index 7a45c2fb..d4e012f8 100644 --- a/release-notes.rst +++ b/release-notes.rst @@ -1,6 +1,18 @@ Release notes for chevah.compat =============================== +1.7.1 - 2026-07-29 +------------------ + +* For Unix, getRealPathFromSegments now returns the root folder instead + of an empty string. + +1.7.0 - 2026-07-22 +------------------ + +* Prevent the filesystem from modifying the root folder. + All operations that would modify the root folder will raise a CompatError. + 1.6.1 - 2025-08-07 ------------------ diff --git a/src/chevah_compat/administration.py b/src/chevah_compat/administration.py index 2623b68f..3f5f530a 100644 --- a/src/chevah_compat/administration.py +++ b/src/chevah_compat/administration.py @@ -563,7 +563,7 @@ def _setUserPassword_freebsd(self, user): ) def _setUserPassword_openbsd(self, user): - code, out = execute( + _, out = execute( command=['encrypt'], input_text=user.password.encode('utf-8'), ) @@ -959,7 +959,7 @@ def deleteUser(self, user): win32net.NetUserDel(user.pdc, user.name) except win32net.error as error: # pragma: no cover # Ignore user not found error. - (number, context, message) = error + (number, _context, _message) = error # Ignore user not found error. if number != ERROR_NONE_MAPPED: raise diff --git a/src/chevah_compat/interfaces.py b/src/chevah_compat/interfaces.py index e0134c1a..1e0a4022 100644 --- a/src/chevah_compat/interfaces.py +++ b/src/chevah_compat/interfaces.py @@ -455,6 +455,11 @@ def isLink(segments): Return True if segments points to a link. """ + def isRoot(segments): + """ + Return True if segments points to the root folder. + """ + def exists(segments): """ Return True if segments points to an existing path. diff --git a/src/chevah_compat/nt_filesystem.py b/src/chevah_compat/nt_filesystem.py index 58af2ede..68075f4b 100644 --- a/src/chevah_compat/nt_filesystem.py +++ b/src/chevah_compat/nt_filesystem.py @@ -226,6 +226,22 @@ def get_path(segments): self._validateDrivePath(result) return six.text_type(result) + def isRoot(self, segments): + """ + See `ILocalFilesystem`. + """ + normalized_segments = self.getSegments(self.getPath(segments)) + if self._lock_in_home: + return super().isRoot(normalized_segments) + + if normalized_segments in [[], ['.'], ['..']]: + return True + + return ( + len(normalized_segments) == 1 + and normalized_segments[0].lower() in self._allowed_drive_letters + ) + # Windows allows only 26 drive letters and is case insensitive. _allowed_drive_letters = [ 'a', @@ -498,6 +514,10 @@ def makeLink(self, target_segments, link_segments): if not self.process_capabilities.symbolic_link: raise NotImplementedError('makeLink not implemented on this OS.') + self._rejectRoot( + link_segments, 'Creating a link as the root folder is not allowed.' + ) + target_path = self.getRealPathFromSegments( target_segments, include_virtual=False, @@ -562,14 +582,14 @@ def _getStatus(self, path, segments): try: file_info = win32file.GetFileInformationByHandle(file_handle) ( - attributes, - created_at, - accessed_at, - written_at, + _attributes, + _created_at, + _accessed_at, + _written_at, volume_id, - file_high, - file_low, - n_links, + _file_high, + _file_low, + _n_links, index_high, index_low, ) = file_info @@ -621,6 +641,10 @@ def setAttributes(self, segments, attributes): """ See `ILocalFilesystem`. """ + self._rejectRoot( + segments, + 'Setting attributes for the nt root folder is not allowed.', + ) with self._windowsToOSError(segments): if 'uid' in attributes or 'gid' in attributes: raise OSError(errno.EPERM, 'Operation not supported') @@ -808,6 +832,11 @@ def deleteFolder(self, segments, recursive=True): For symbolic links we always force non-recursive behaviour. """ + self._rejectRoot( + segments, + 'Deleting the nt root folder is not allowed.', + ) + path = self.getRealPathFromSegments(segments, include_virtual=False) path_encoded = self.getEncodedPath(path) try: @@ -854,6 +883,10 @@ def setOwner(self, segments, owner): """ See `ILocalFilesystem`. """ + self._rejectRoot( + segments, + 'Setting owner for the nt root folder is not allowed.', + ) path = self.getRealPathFromSegments(segments, include_virtual=False) encoded_path = self.getEncodedPath(path) try: @@ -877,7 +910,7 @@ def _setOwner(self, path, owner): ) d_acl = security_descriptor.GetSecurityDescriptorDacl() - user_sid, user_domain, user_type = ( + user_sid, _user_domain, _user_type = ( win32security.LookupAccountName(None, owner) ) flags = ( @@ -950,10 +983,14 @@ def addGroup(self, segments, group, permissions=None): """ See `ILocalFilesystem`. """ + self._rejectRoot( + segments, + 'Adding group for the nt root folder is not allowed.', + ) path = self.getRealPathFromSegments(segments, include_virtual=False) encoded_path = self.getEncodedPath(path) try: - group_sid, group_domain, group_type = ( + group_sid, _group_domain, _group_type = ( win32security.LookupAccountName(None, group) ) except win32net.error: @@ -988,10 +1025,14 @@ def removeGroup(self, segments, group): """ See `ILocalFilesystem`. """ + self._rejectRoot( + segments, + 'Removing group for the nt root folder is not allowed.', + ) path = self.getRealPathFromSegments(segments, include_virtual=False) encoded_path = self.getEncodedPath(path) try: - group_sid, group_domain, group_type = ( + group_sid, _group_domain, _group_type = ( win32security.LookupAccountName(None, group) ) except win32net.error: @@ -1018,7 +1059,7 @@ def removeGroup(self, segments, group): return None index_ace_to_remove = -1 for index in range(ace_count): - ((ace_type, ace_flag), mask, sid) = dacl.GetAce(index) + ((_ace_type, _ace_flag), _mask, sid) = dacl.GetAce(index) if group_sid == sid: index_ace_to_remove = index break @@ -1045,7 +1086,7 @@ def hasGroup(self, segments, group): encoded_path = self.getEncodedPath(path) try: - group_sid, group_domain, group_type = ( + group_sid, _group_domain, _group_type = ( win32security.LookupAccountName(None, group) ) except win32net.error: @@ -1066,7 +1107,7 @@ def hasGroup(self, segments, group): # Nothing in the list. return False for index in range(ace_count): - ((ace_type, ace_flag), mask, sid) = dacl.GetAce(index) + ((_ace_type, _ace_flag), _mask, sid) = dacl.GetAce(index) if group_sid == sid: return True return False diff --git a/src/chevah_compat/nt_users.py b/src/chevah_compat/nt_users.py index 5e54440e..7ae4d72e 100644 --- a/src/chevah_compat/nt_users.py +++ b/src/chevah_compat/nt_users.py @@ -206,11 +206,11 @@ def getGroupForUser(self, username, groups, token): if not groups: raise ValueError("Groups for validation can't be empty.") - primary_domain_controller, name = self._parseUPN(username) + primary_domain_controller, _name = self._parseUPN(username) for group in groups: try: - group_sid, group_domain, group_type = ( + group_sid, _group_domain, _group_type = ( win32security.LookupAccountName( primary_domain_controller, group, diff --git a/src/chevah_compat/posix_filesystem.py b/src/chevah_compat/posix_filesystem.py index 0e6a5bde..95b10593 100644 --- a/src/chevah_compat/posix_filesystem.py +++ b/src/chevah_compat/posix_filesystem.py @@ -387,6 +387,25 @@ def isLink(self, segments): """ raise NotImplementedError('isLink') + def isRoot(self, segments): + """ + See `ILocalFilesystem`. + """ + normalized_segments = self.getSegments(self.getPath(segments)) + path = self.getRealPathFromSegments( + normalized_segments, + include_virtual=False, + ) + root = self.getRealPathFromSegments([], include_virtual=False) + return root.lower() == path.lower() + + def _rejectRoot(self, segments, message): + """ + Helper to raise an error on operations that modify the root folder. + """ + if self.isRoot(segments): + raise CompatError(1009, message) + def exists(self, segments): """See `ILocalFilesystem`.""" @@ -406,7 +425,10 @@ def exists(self, segments): return os.path.lexists(path_encoded) def createFolder(self, segments, recursive=False): - """See `ILocalFilesystem`.""" + """ + See `ILocalFilesystem`. + """ + self._rejectRoot(segments, 'Creating the root folder is not allowed.') path = self.getRealPathFromSegments(segments, include_virtual=False) path_encoded = self.getEncodedPath(path) with self._impersonateUser(): @@ -488,7 +510,15 @@ def deleteFile(self, segments, ignore_errors=False): raise def rename(self, from_segments, to_segments): - """See `ILocalFilesystem`.""" + """ + See `ILocalFilesystem`. + """ + self._rejectRoot( + from_segments, 'Renaming from the root folder is not allowed.' + ) + self._rejectRoot( + to_segments, 'Renaming to the root folder is not allowed.' + ) from_path = self.getRealPathFromSegments( from_segments, include_virtual=False, @@ -857,7 +887,13 @@ def _getPlaceholderStatus(self): return os.stat_result([0o40555, 0, 0, 0, 1, 1, 0, 1, modified, 0]) def setAttributes(self, segments, attributes): - """See `ILocalFilesystem`.""" + """ + See `ILocalFilesystem`. + """ + self._rejectRoot( + segments, + 'Setting attributes on the posix root folder is not allowed.', + ) path = self.getRealPathFromSegments(segments, include_virtual=False) path_encoded = self.getEncodedPath(path) with self._impersonateUser(): diff --git a/src/chevah_compat/testing/mockup.py b/src/chevah_compat/testing/mockup.py index e579b003..5765934f 100644 --- a/src/chevah_compat/testing/mockup.py +++ b/src/chevah_compat/testing/mockup.py @@ -53,7 +53,7 @@ def sanitizeName(cls, name): os_version = ChevahTestCase.os_version if os_name in ['aix', 'hpux', 'freebsd', 'openbsd']: return _sanitize_name_legacy_unix(name) - if os_version in ['osx-10.16']: + if os_version == 'osx-10.16': # It looks like macOS 11 can't handle full Unix group names. macosname = _sanitize_name_legacy_unix(name) return macosname.replace('_', 'Z') diff --git a/src/chevah_compat/tests/elevated/test_system_users.py b/src/chevah_compat/tests/elevated/test_system_users.py index 1509828d..cf400b7c 100644 --- a/src/chevah_compat/tests/elevated/test_system_users.py +++ b/src/chevah_compat/tests/elevated/test_system_users.py @@ -254,7 +254,7 @@ def test_checkPasswdFile_valid(self): if self.os_name in ['aix', 'hpux']: # On AIX and HPUX password is in the passwd file. self.assertTrue(result) - elif self.os_version in ['rhel-5']: + elif self.os_version == 'rhel-5': # Old RHEL/Centos contain has the password in passwd file. self.assertIsTrue(result) else: @@ -277,7 +277,7 @@ def test_checkPasswdFile_invalid(self): if self.os_name in ['aix', 'hpux']: # On AIX and HPUX invalid passwords are not accepted. self.assertFalse(result) - elif self.os_version in ['rhel-5']: + elif self.os_version == 'rhel-5': # Old RHEL/Centos contain has the password in passwd file, # and the provided password doesn't match what is in the file. self.assertIsFalse(result) diff --git a/src/chevah_compat/tests/normal/test_capabilities.py b/src/chevah_compat/tests/normal/test_capabilities.py index 3e4a1c59..b910e2e1 100644 --- a/src/chevah_compat/tests/normal/test_capabilities.py +++ b/src/chevah_compat/tests/normal/test_capabilities.py @@ -298,7 +298,7 @@ def test_getCurrentPrivilegesDescription(self): self.assertNotContains('SeBackupPrivilege:0', text) self.assertNotContains('SeRestorePrivilege', text) - if self.os_version in ['nt-5.1']: + if self.os_version == 'nt-5.1': # Windows XP has SE_CREATE_GLOBAL enabled even when # running without administrator privileges. self.assertContains('SeCreateGlobalPrivilege:3', text) diff --git a/src/chevah_compat/tests/normal/test_filesystem.py b/src/chevah_compat/tests/normal/test_filesystem.py index 8e8079c8..029eb107 100644 --- a/src/chevah_compat/tests/normal/test_filesystem.py +++ b/src/chevah_compat/tests/normal/test_filesystem.py @@ -31,6 +31,16 @@ class FilesystemTestingHelpers: Common code for running filesystem tests. """ + def assertRootOperationRejected(self, operation): + """ + Check that all segment forms resolving to the filesystem root fail. + """ + root_segments = ([], ['..'], ['child', '..']) + for segments in root_segments: + error = self.assertRaises(CompatError, operation, segments) + self.assertEqual(1009, error.event_id) + self.assertEndsWith('is not allowed.', error.message) + def makeLink(self, segments, cleanup=True): """ Create a symbolic link to `segments` and return the segments for it. @@ -90,6 +100,19 @@ class FilesystemTestMixin(FilesystemTestingHelpers): Common tests for filesystem for all OSes. """ + def test_isRoot(self): + """ + Root detection normalizes segments before comparing their real paths. + """ + self.assertIsTrue(self.filesystem.isRoot([])) + self.assertIsTrue(self.filesystem.isRoot(['..'])) + self.assertIsTrue(self.filesystem.isRoot(['child', '..'])) + if self.os_family == 'nt' and not self.filesystem._lock_in_home: + self.assertIsTrue(self.filesystem.isRoot(['c'])) + self.assertIsFalse(self.filesystem.isRoot(['c', 'child'])) + else: + self.assertIsFalse(self.filesystem.isRoot(['c'])) + def test_getSegments_upper_paths(self): """ It will properly remove parent folder (..) and root folder (.) in @@ -475,7 +498,7 @@ def test_deleteFolder_recursive_non_empty(self): """ It can delete folder even if it is not empty. """ - segments, child_name = self.createFolderWithChild() + segments, _ = self.createFolderWithChild() self.assertTrue(self.filesystem.exists(segments)) self.filesystem.deleteFolder(segments, recursive=True) @@ -620,7 +643,7 @@ def test_makeLink_windows_share(self): # We assume all slaves have the c:\temp folder. share_name = 'share-name ' + mk.string() self.makeWindowsShare(path='c:\\temp', name=share_name) - path, segments = mk.fs.makePathInTemp() + _, segments = mk.fs.makePathInTemp() self.addCleanup(self.filesystem.deleteFolder, segments) filename = mk.makeFilename() file_segments = ['c', 'temp', filename] @@ -2265,6 +2288,19 @@ def test_setAttributes_mode(self): self.assertEqual(initial.mode, after.mode) + def test_deleteFolder_drive_root(self): + """ + Deleting a drive root is rejected. + """ + error = self.assertRaises( + CompatError, + self.filesystem.deleteFolder, + ['c'], + recursive=True, + ) + self.assertEqual(1009, error.event_id) + self.assertEndsWith('is not allowed.', error.message) + def test_isAbsolutePath(self): """ Unit test for detecting which Windows path is absolute. @@ -2485,6 +2521,17 @@ def setUpClass(cls): cls.unlocked_filesystem = LocalFilesystem(avatar=DefaultAvatar()) cls.filesystem = cls.unlocked_filesystem + def test_deleteFolder_root(self): + """ + Recursive deletion rejects the unlocked filesystem root. + """ + self.assertRootOperationRejected( + lambda segments: self.unlocked_filesystem.deleteFolder( + segments, + recursive=True, + ), + ) + def test_getSegments(self): """ Check getSegments. @@ -2812,7 +2859,7 @@ def test_exists_share_link(self): """ Will return True when we have a UNC / network link. """ - path, segments = mk.fs.makePathInTemp() + _, segments = mk.fs.makePathInTemp() # Make sure path does not exists. result = self.unlocked_filesystem.exists(segments) self.assertFalse(result) @@ -2906,6 +2953,101 @@ def setUpClass(cls): cls.locked_filesystem = LocalFilesystem(avatar=cls.locked_avatar) cls.filesystem = cls.locked_filesystem + def test_createFolder_root(self): + """ + Creating the avatar root using any equivalent segments is rejected. + """ + self.assertRootOperationRejected(self.locked_filesystem.createFolder) + + def test_deleteFolder_root(self): + """ + Recursive deletion is rejected for the avatar root. + """ + self.assertRootOperationRejected( + lambda segments: self.locked_filesystem.deleteFolder( + segments, + recursive=True, + ), + ) + + def test_rename_from_root(self): + """ + Renaming the avatar root using any equivalent segments is rejected. + """ + self.assertRootOperationRejected( + lambda segments: self.locked_filesystem.rename( + segments, + ['destination'], + ), + ) + + def test_rename_to_root(self): + """ + Renaming another path over the avatar root is rejected. + """ + self.assertRootOperationRejected( + lambda segments: self.locked_filesystem.rename( + ['source'], + segments, + ), + ) + + def test_setAttributes_root(self): + """ + Changing attributes on the avatar root is rejected. + """ + self.assertRootOperationRejected( + lambda segments: self.locked_filesystem.setAttributes( + segments, + {'mode': 0o700}, + ), + ) + + def test_setOwner_root(self): + """ + Changing the owner of the avatar root is rejected. + """ + self.assertRootOperationRejected( + lambda segments: self.locked_filesystem.setOwner( + segments, + 'ignored-owner', + ), + ) + + def test_addGroup_root(self): + """ + Adding a group to the avatar root is rejected. + """ + self.assertRootOperationRejected( + lambda segments: self.locked_filesystem.addGroup( + segments, + 'ignored-group', + ), + ) + + def test_removeGroup_root(self): + """ + Removing a group from the avatar root is rejected. + """ + self.assertRootOperationRejected( + lambda segments: self.locked_filesystem.removeGroup( + segments, + 'ignored-group', + ), + ) + + @conditionals.onCapability('symbolic_link', True) + def test_makeLink_root(self): + """ + Creating a link over the avatar root is rejected. + """ + self.assertRootOperationRejected( + lambda segments: self.locked_filesystem.makeLink( + ['ignored-target'], + segments, + ), + ) + def test_getSegments_locked(self): """ Check getSegments for a locked filesystem. @@ -3108,7 +3250,7 @@ def test_readLink_inside_home(self): """ It return the virtual link of the target. """ - path, target_segments = self.tempFile() + _, target_segments = self.tempFile() link_segments = [f'{target_segments[-1]}-link'] mk.fs.makeLink( target_segments=target_segments, @@ -3265,7 +3407,7 @@ def test_init_virtual_overlap_folder_case_sensitive(self): Virtual path on Windows/OSX are case insensitive, while on other systems are case sensitive. """ - path, segments = self.tempFolder(suffix='low') + _, segments = self.tempFolder(suffix='low') virtual_shadow = segments[-1][:-3] + segments[-1][-3:].upper() if self.os_name in ['windows', 'osx']: diff --git a/src/chevah_compat/tests/normal/testing/test_testcase.py b/src/chevah_compat/tests/normal/testing/test_testcase.py index 86e63214..ec079424 100644 --- a/src/chevah_compat/tests/normal/testing/test_testcase.py +++ b/src/chevah_compat/tests/normal/testing/test_testcase.py @@ -358,9 +358,11 @@ def much_later(): # pragma: no cover """ self.EXCEPTED_DELAYED_CALLS = [ - 'TestTwistedTestCase.' - 'test_assertReactorIsClean_excepted_delayed_calls.' - '.much_later' + ( + 'TestTwistedTestCase.' + 'test_assertReactorIsClean_excepted_delayed_calls.' + '.much_later' + ) ] delayed_call = reactor.callLater(10, much_later) diff --git a/src/chevah_compat/unix_filesystem.py b/src/chevah_compat/unix_filesystem.py index 5cf6eb89..83174ed0 100644 --- a/src/chevah_compat/unix_filesystem.py +++ b/src/chevah_compat/unix_filesystem.py @@ -14,7 +14,6 @@ from zope.interface import implementer -from chevah_compat.exceptions import CompatError from chevah_compat.interfaces import ILocalFilesystem from chevah_compat.posix_filesystem import PosixFilesystemBase from chevah_compat.unix_users import UnixUsers @@ -57,7 +56,8 @@ def getRealPathFromSegments(self, segments, include_virtual=True): relative_path = '/' + '/'.join(segments) relative_path = self.getAbsoluteRealPath(relative_path).rstrip('/') - return str(self._root_path.rstrip('/') + relative_path) + path = self._root_path.rstrip('/') + relative_path + return str(path or '/') def getSegmentsFromRealPath(self, path): """ @@ -102,6 +102,10 @@ def makeLink(self, target_segments, link_segments): """ See `ILocalFilesystem`. """ + self._rejectRoot( + link_segments, + 'Creating a link in the root folder is not allowed.', + ) target_path = self.getRealPathFromSegments( target_segments, include_virtual=False, @@ -116,6 +120,10 @@ def makeLink(self, target_segments, link_segments): def setOwner(self, segments, owner): """See `ILocalFilesystem`.""" + self._rejectRoot( + segments, + 'Setting owner for the unix root folder is not allowed.', + ) path = self.getRealPathFromSegments(segments, include_virtual=False) try: uid = pwd.getpwnam(owner).pw_uid @@ -136,6 +144,10 @@ def getOwner(self, segments): def addGroup(self, segments, group, permissions=None): """See `ILocalFilesystem`.""" + self._rejectRoot( + segments, + 'Adding group for the unix root folder is not allowed.', + ) path = self.getRealPathFromSegments(segments, include_virtual=False) try: gid = grp.getgrnam(group).gr_gid @@ -157,6 +169,10 @@ def removeGroup(self, segments, group): This has no effect on Unix/Linux but raises an error if we are touching a virtual root. """ + self._rejectRoot( + segments, + 'Removing group for the unix root folder is not allowed.', + ) self.getRealPathFromSegments(segments, include_virtual=False) return @@ -203,10 +219,11 @@ def deleteFolder(self, segments, recursive=True): """ See `ILocalFilesystem`. """ + self._rejectRoot( + segments, + 'Deleting the unix root folder is not allowed.', + ) path = self.getRealPathFromSegments(segments, include_virtual=False) - if path == '/': - raise CompatError(1009, 'Deleting Unix root folder is not allowed.') - path_encoded = self.getEncodedPath(path) if self.isLink(segments): diff --git a/src/chevah_compat/unix_users.py b/src/chevah_compat/unix_users.py index 8e90d389..ae9f8da9 100644 --- a/src/chevah_compat/unix_users.py +++ b/src/chevah_compat/unix_users.py @@ -348,7 +348,7 @@ def _checkShadowFile(self, username, password): crypted_password = _get_etc_shadow(username) # Locked account - if crypted_password in ('LK',): + if crypted_password == 'LK': return False # Allow other methods to take over if password is not