From d40e6f70a5edabffcbfff22912163520da3a29e2 Mon Sep 17 00:00:00 2001 From: Brian Curtin Date: Thu, 8 Jul 2010 21:39:08 +0000 Subject: Implement #1578269. Patch by Jason R. Coombs. Added Windows support for os.symlink when run on Windows 6.0 or greater, aka Vista. Previous Windows versions will raise NotImplementedError when trying to symlink. Includes numerous test updates and additions to test_os, including a symlink_support module because of the fact that privilege escalation is required in order to run the tests to ensure that the user is able to create symlinks. By default, accounts do not have the required privilege, so the escalation code will have to be exposed later (or documented on how to do so). I'll be following up with that work next. Note that the tests use ctypes, which was agreed on during the PyCon language summit. --- Doc/library/os.path.rst | 12 +- Doc/library/os.rst | 24 +- Lib/ntpath.py | 43 +++- Lib/tarfile.py | 12 +- Lib/test/support.py | 25 +- Lib/test/symlink_support.py | 203 ++++++++++++++++ Lib/test/test_glob.py | 14 +- Lib/test/test_httpservers.py | 2 +- Lib/test/test_os.py | 82 ++++++- Lib/test/test_platform.py | 34 +-- Lib/test/test_posixpath.py | 345 ++++++++++++++------------- Lib/test/test_shutil.py | 57 ++--- Lib/test/test_sys.py | 2 +- Lib/test/test_sysconfig.py | 14 +- Lib/test/test_tarfile.py | 29 ++- Misc/ACKS | 1 + Misc/NEWS | 3 + Modules/posixmodule.c | 546 +++++++++++++++++++++++++++++++++++++++++-- 18 files changed, 1167 insertions(+), 281 deletions(-) create mode 100644 Lib/test/symlink_support.py diff --git a/Doc/library/os.path.rst b/Doc/library/os.path.rst index c838983..7fcb2a9 100644 --- a/Doc/library/os.path.rst +++ b/Doc/library/os.path.rst @@ -231,11 +231,15 @@ applications should use string objects to access all files. .. function:: samefile(path1, path2) - Return ``True`` if both pathname arguments refer to the same file or directory - (as indicated by device number and i-node number). Raise an exception if a - :func:`os.stat` call on either pathname fails. + Return ``True`` if both pathname arguments refer to the same file or directory. + On Unix, this is determined by the device number and i-node number and raises an + exception if a :func:`os.stat` call on either pathname fails. - Availability: Unix. + On Windows, two files are the same if they resolve to the same final path + name using the Windows API call GetFinalPathNameByHandle and this function + raises an exception if handles cannot be obtained to either file. + + Availability: Windows, Unix. .. function:: sameopenfile(fp1, fp2) diff --git a/Doc/library/os.rst b/Doc/library/os.rst index b74350d..d8835f6 100644 --- a/Doc/library/os.rst +++ b/Doc/library/os.rst @@ -1065,7 +1065,7 @@ Files and Directories Like :func:`stat`, but do not follow symbolic links. This is an alias for :func:`stat` on platforms that do not support symbolic links, such as - Windows. + Windows prior to 6.0 (Vista). .. function:: mkfifo(path[, mode]) @@ -1181,7 +1181,7 @@ Files and Directories and the call may raise an UnicodeDecodeError. If the *path* is a bytes object, the result will be a bytes object. - Availability: Unix. + Availability: Unix, Windows. .. function:: remove(path) @@ -1341,9 +1341,25 @@ Files and Directories .. function:: symlink(source, link_name) - Create a symbolic link pointing to *source* named *link_name*. + Create a symbolic link pointing to *source* named *link_name*. On Windows, + symlink version takes an additional, optional parameter, + *target_is_directory*, which defaults to False. - Availability: Unix. + symlink(source, link_name, target_is_directory=False) + + On Windows, a symlink represents a file or a directory, and does not + morph to the target dynamically. For this reason, when creating a + symlink on Windows, if the target is not already present, the symlink + will default to being a file symlink. If *target_is_directory* is set to + True, the symlink will be created as a directory symlink. This + parameter is ignored if the target exists (and the symlink is created + with the same type as the target). + + Symbolic link support was introduced in Windows 6.0 (Vista). *symlink* + will raise a NotImplementedError on Windows versions earlier than 6.0. The + SeCreateSymbolicLinkPrivilege is required in order to create symlinks. + + Availability: Unix, Windows 6.0. .. function:: unlink(path) diff --git a/Lib/ntpath.py b/Lib/ntpath.py index 7972aa9..ab0b318 100644 --- a/Lib/ntpath.py +++ b/Lib/ntpath.py @@ -16,7 +16,8 @@ __all__ = ["normcase","isabs","join","splitdrive","split","splitext", "getatime","getctime", "islink","exists","lexists","isdir","isfile", "ismount", "expanduser","expandvars","normpath","abspath", "splitunc","curdir","pardir","sep","pathsep","defpath","altsep", - "extsep","devnull","realpath","supports_unicode_filenames","relpath"] + "extsep","devnull","realpath","supports_unicode_filenames","relpath", + "samefile",] # strings representing various path-related bits and pieces # These are primarily for export; internally, they are hardcoded. @@ -309,16 +310,28 @@ def dirname(p): return split(p)[0] # Is a path a symbolic link? -# This will always return false on systems where posix.lstat doesn't exist. +# This will always return false on systems where os.lstat doesn't exist. def islink(path): - """Test for symbolic link. - On WindowsNT/95 and OS/2 always returns false + """Test whether a path is a symbolic link. + This will always return false for Windows prior to 6.0 + and for OS/2. """ - return False - -# alias exists to lexists -lexists = exists + try: + st = os.lstat(path) + except (os.error, AttributeError): + return False + return stat.S_ISLNK(st.st_mode) + +# Being true for dangling symbolic links is also useful. + +def lexists(path): + """Test whether a path exists. Returns True for broken symbolic links""" + try: + st = os.lstat(path) + except (os.error, WindowsError): + return False + return True # Is a path a mount point? Either a root (with or without drive letter) # or an UNC path with at most a / or \ after the mount point. @@ -612,3 +625,17 @@ def relpath(path, start=curdir): if not rel_list: return _get_dot(path) return join(*rel_list) + + +# determine if two files are in fact the same file +def samefile(f1, f2): + "Test whether two pathnames reference the same actual file" + try: + from nt import _getfinalpathname + return _getfinalpathname(f1) == _getfinalpathname(f2) + except (NotImplementedError, ImportError): + # On Windows XP and earlier, two files are the same if their + # absolute pathnames are the same. + # Also, on other operating systems, fake this method with a + # Windows-XP approximation. + return abspath(f1) == abspath(f2) diff --git a/Lib/tarfile.py b/Lib/tarfile.py index 4839eb1..aca934a 100644 --- a/Lib/tarfile.py +++ b/Lib/tarfile.py @@ -2273,7 +2273,7 @@ class TarFile(object): (platform limitation), we try to make a copy of the referenced file instead of a link. """ - if hasattr(os, "symlink") and hasattr(os, "link"): + try: # For systems that support symbolic and hard links. if tarinfo.issym(): os.symlink(tarinfo.linkname, targetpath) @@ -2282,7 +2282,15 @@ class TarFile(object): if os.path.exists(tarinfo._link_target): os.link(tarinfo._link_target, targetpath) else: - self._extract_member(self._find_link_target(tarinfo), targetpath) + self._extract_mem + except (AttributeError, NotImplementedError, WindowsError): + # AttributeError if no os.symlink + # NotImplementedError if on Windows XP + # WindowsError (1314) if the required privilege is not held by the client + if tarinfo.issym(): + linkpath = os.path.join(os.path.dirname(tarinfo.name),tarinfo.linkname) + else: + linkpath = tarinfo.linkname else: try: self._extract_member(self._find_link_target(tarinfo), targetpath) diff --git a/Lib/test/support.py b/Lib/test/support.py index 08105df..83d7ba8 100644 --- a/Lib/test/support.py +++ b/Lib/test/support.py @@ -17,6 +17,7 @@ import unittest import importlib import collections import re +import subprocess import imp import time try: @@ -38,8 +39,7 @@ __all__ = [ "set_memlimit", "bigmemtest", "bigaddrspacetest", "BasicTestRunner", "run_unittest", "run_doctest", "threading_setup", "threading_cleanup", "reap_children", "cpython_only", "check_impl_detail", "get_attribute", - "swap_item", "swap_attr", - ] + "swap_item", "swap_attr", "can_symlink", "skip_unless_symlink"] class Error(Exception): @@ -1169,6 +1169,27 @@ def reap_children(): except: break +try: + from .symlink_support import enable_symlink_privilege +except: + enable_symlink_privilege = lambda: True + +def can_symlink(): + """It's no longer sufficient to test for the presence of symlink in the + os module - on Windows XP and earlier, os.symlink exists but a + NotImplementedError is thrown. + """ + has_symlink = hasattr(os, 'symlink') + is_old_windows = sys.platform == "win32" and sys.getwindowsversion().major < 6 + has_privilege = False if is_old_windows else enable_symlink_privilege() + return has_symlink and (not is_old_windows) and has_privilege + +def skip_unless_symlink(test): + """Skip decorator for tests that require functional symlink""" + selector = can_symlink() + msg = "Requires functional symlink implementation" + return [unittest.skip(msg)(test), test][selector] + @contextlib.contextmanager def swap_attr(obj, attr, new_val): """Temporary swap out an attribute with a new object. diff --git a/Lib/test/symlink_support.py b/Lib/test/symlink_support.py new file mode 100644 index 0000000..eab7667 --- /dev/null +++ b/Lib/test/symlink_support.py @@ -0,0 +1,203 @@ +""" +A module built to test if the current process has the privilege to +create symlinks on Windows. +""" + +# allow script to run natively under python 2.6+ +from __future__ import print_function + +import ctypes +from ctypes import wintypes + +GetCurrentProcess = ctypes.windll.kernel32.GetCurrentProcess +GetCurrentProcess.restype = wintypes.HANDLE +OpenProcessToken = ctypes.windll.advapi32.OpenProcessToken +OpenProcessToken.argtypes = (wintypes.HANDLE, wintypes.DWORD, ctypes.POINTER(wintypes.HANDLE)) +OpenProcessToken.restype = wintypes.BOOL + +class LUID(ctypes.Structure): + _fields_ = [ + ('low_part', wintypes.DWORD), + ('high_part', wintypes.LONG), + ] + + def __eq__(self, other): + return ( + self.high_part == other.high_part and + self.low_part == other.low_part + ) + + def __ne__(self, other): + return not (self==other) + +LookupPrivilegeValue = ctypes.windll.advapi32.LookupPrivilegeValueW +LookupPrivilegeValue.argtypes = ( + wintypes.LPWSTR, # system name + wintypes.LPWSTR, # name + ctypes.POINTER(LUID), + ) +LookupPrivilegeValue.restype = wintypes.BOOL + +class TOKEN_INFORMATION_CLASS: + TokenUser = 1 + TokenGroups = 2 + TokenPrivileges = 3 + # ... see http://msdn.microsoft.com/en-us/library/aa379626%28VS.85%29.aspx + +SE_PRIVILEGE_ENABLED_BY_DEFAULT = (0x00000001) +SE_PRIVILEGE_ENABLED = (0x00000002) +SE_PRIVILEGE_REMOVED = (0x00000004) +SE_PRIVILEGE_USED_FOR_ACCESS = (0x80000000) + +class LUID_AND_ATTRIBUTES(ctypes.Structure): + _fields_ = [ + ('LUID', LUID), + ('attributes', wintypes.DWORD), + ] + + def is_enabled(self): + return bool(self.attributes & SE_PRIVILEGE_ENABLED) + + def enable(self): + self.attributes |= SE_PRIVILEGE_ENABLED + + def get_name(self): + size = wintypes.DWORD(10240) + buf = ctypes.create_unicode_buffer(size.value) + res = LookupPrivilegeName(None, self.LUID, buf, size) + if res == 0: + raise RuntimeError + return buf[:size.value] + + def __str__(self): + name = self.name + fmt = ['{name}', '{name} (enabled)'][self.is_enabled()] + return fmt.format(**vars()) + +LookupPrivilegeName = ctypes.windll.advapi32.LookupPrivilegeNameW +LookupPrivilegeName.argtypes = ( + wintypes.LPWSTR, # lpSystemName + ctypes.POINTER(LUID), # lpLuid + wintypes.LPWSTR, # lpName + ctypes.POINTER(wintypes.DWORD), #cchName + ) +LookupPrivilegeName.restype = wintypes.BOOL + +class TOKEN_PRIVILEGES(ctypes.Structure): + _fields_ = [ + ('count', wintypes.DWORD), + ('privileges', LUID_AND_ATTRIBUTES*0), + ] + + def get_array(self): + array_type = LUID_AND_ATTRIBUTES*self.count + privileges = ctypes.cast(self.privileges, ctypes.POINTER(array_type)).contents + return privileges + + def __iter__(self): + return iter(self.get_array()) + +PTOKEN_PRIVILEGES = ctypes.POINTER(TOKEN_PRIVILEGES) + +GetTokenInformation = ctypes.windll.advapi32.GetTokenInformation +GetTokenInformation.argtypes = [ + wintypes.HANDLE, # TokenHandle + ctypes.c_uint, # TOKEN_INFORMATION_CLASS value + ctypes.c_void_p, # TokenInformation + wintypes.DWORD, # TokenInformationLength + ctypes.POINTER(wintypes.DWORD), # ReturnLength + ] +GetTokenInformation.restype = wintypes.BOOL + +# http://msdn.microsoft.com/en-us/library/aa375202%28VS.85%29.aspx +AdjustTokenPrivileges = ctypes.windll.advapi32.AdjustTokenPrivileges +AdjustTokenPrivileges.restype = wintypes.BOOL +AdjustTokenPrivileges.argtypes = [ + wintypes.HANDLE, # TokenHandle + wintypes.BOOL, # DisableAllPrivileges + PTOKEN_PRIVILEGES, # NewState (optional) + wintypes.DWORD, # BufferLength of PreviousState + PTOKEN_PRIVILEGES, # PreviousState (out, optional) + ctypes.POINTER(wintypes.DWORD), # ReturnLength + ] + +def get_process_token(): + "Get the current process token" + token = wintypes.HANDLE() + TOKEN_ALL_ACCESS = 0xf01ff + res = OpenProcessToken(GetCurrentProcess(), TOKEN_ALL_ACCESS, token) + if not res > 0: + raise RuntimeError("Couldn't get process token") + return token + +def get_symlink_luid(): + "Get the LUID for the SeCreateSymbolicLinkPrivilege" + symlink_luid = LUID() + res = LookupPrivilegeValue(None, "SeCreateSymbolicLinkPrivilege", symlink_luid) + if not res > 0: + raise RuntimeError("Couldn't lookup privilege value") + return symlink_luid + +def get_privilege_information(): + "Get all privileges associated with the current process." + # first call with zero length to determine what size buffer we need + + return_length = wintypes.DWORD() + params = [ + get_process_token(), + TOKEN_INFORMATION_CLASS.TokenPrivileges, + None, + 0, + return_length, + ] + + res = GetTokenInformation(*params) + + # assume we now have the necessary length in return_length + + buffer = ctypes.create_string_buffer(return_length.value) + params[2] = buffer + params[3] = return_length.value + + res = GetTokenInformation(*params) + assert res > 0, "Error in second GetTokenInformation (%d)" % res + + privileges = ctypes.cast(buffer, ctypes.POINTER(TOKEN_PRIVILEGES)).contents + return privileges + +def report_privilege_information(): + "Report all privilege information assigned to the current process." + privileges = get_privilege_information() + print("found {0} privileges".format(privileges.count)) + tuple(map(print, privileges)) + +def enable_symlink_privilege(): + """ + Try to assign the symlink privilege to the current process token. + Return True if the assignment is successful. + """ + # create a space in memory for a TOKEN_PRIVILEGES structure + # with one element + size = ctypes.sizeof(TOKEN_PRIVILEGES) + size += ctypes.sizeof(LUID_AND_ATTRIBUTES) + buffer = ctypes.create_string_buffer(size) + tp = ctypes.cast(buffer, ctypes.POINTER(TOKEN_PRIVILEGES)).contents + tp.count = 1 + tp.get_array()[0].enable() + tp.get_array()[0].LUID = get_symlink_luid() + token = get_process_token() + res = AdjustTokenPrivileges(token, False, tp, 0, None, None) + if res == 0: + raise RuntimeError("Error in AdjustTokenPrivileges") + + ERROR_NOT_ALL_ASSIGNED = 1300 + return ctypes.windll.kernel32.GetLastError() != ERROR_NOT_ALL_ASSIGNED + +def main(): + assigned = enable_symlink_privilege() + msg = ['failure', 'success'][assigned] + + print("Symlink privilege assignment completed with {0}".format(msg)) + +if __name__ == '__main__': + main() diff --git a/Lib/test/test_glob.py b/Lib/test/test_glob.py index 5a3be94..05e29b5 100644 --- a/Lib/test/test_glob.py +++ b/Lib/test/test_glob.py @@ -1,5 +1,5 @@ import unittest -from test.support import run_unittest, TESTFN +from test.support import run_unittest, TESTFN, skip_unless_symlink, can_symlink import glob import os import shutil @@ -25,7 +25,7 @@ class GlobTests(unittest.TestCase): self.mktemp('ZZZ') self.mktemp('a', 'bcd', 'EF') self.mktemp('a', 'bcd', 'efg', 'ha') - if hasattr(os, 'symlink'): + if can_symlink(): os.symlink(self.norm('broken'), self.norm('sym1')) os.symlink(self.norm('broken'), self.norm('sym2')) @@ -98,12 +98,12 @@ class GlobTests(unittest.TestCase): # either of these results are reasonable self.assertIn(res[0], [self.tempdir, self.tempdir + os.sep]) + @skip_unless_symlink def test_glob_broken_symlinks(self): - if hasattr(os, 'symlink'): - eq = self.assertSequencesEqual_noorder - eq(self.glob('sym*'), [self.norm('sym1'), self.norm('sym2')]) - eq(self.glob('sym1'), [self.norm('sym1')]) - eq(self.glob('sym2'), [self.norm('sym2')]) + eq = self.assertSequencesEqual_noorder + eq(self.glob('sym*'), [self.norm('sym1'), self.norm('sym2')]) + eq(self.glob('sym1'), [self.norm('sym1')]) + eq(self.glob('sym2'), [self.norm('sym2')]) def test_main(): diff --git a/Lib/test/test_httpservers.py b/Lib/test/test_httpservers.py index 6a93800..4895b65 100644 --- a/Lib/test/test_httpservers.py +++ b/Lib/test/test_httpservers.py @@ -299,7 +299,7 @@ class CGIHTTPServerTestCase(BaseTestCase): # The shebang line should be pure ASCII: use symlink if possible. # See issue #7668. - if hasattr(os, 'symlink'): + if support.can_symlink(): self.pythonexe = os.path.join(self.parent_dir, 'python') os.symlink(sys.executable, self.pythonexe) else: diff --git a/Lib/test/test_os.py b/Lib/test/test_os.py index 14a9a7a..4693164 100644 --- a/Lib/test/test_os.py +++ b/Lib/test/test_os.py @@ -520,7 +520,7 @@ class WalkTests(unittest.TestCase): f = open(path, "w") f.write("I'm " + path + " and proud of it. Blame test_os.\n") f.close() - if hasattr(os, "symlink"): + if support.can_symlink(): os.symlink(os.path.abspath(t2_path), link_path) sub2_tree = (sub2_path, ["link"], ["tmp3"]) else: @@ -564,7 +564,7 @@ class WalkTests(unittest.TestCase): self.assertEqual(all[flipped + 1], (sub1_path, ["SUB11"], ["tmp2"])) self.assertEqual(all[2 - 2 * flipped], sub2_tree) - if hasattr(os, "symlink"): + if support.can_symlink(): # Walk, following symlinks. for root, dirs, files in os.walk(walk_path, followlinks=True): if root == link_path: @@ -1033,6 +1033,83 @@ class Win32KillTests(unittest.TestCase): self._kill_with_event(signal.CTRL_BREAK_EVENT, "CTRL_BREAK_EVENT") +def skipUnlessWindows6(test): + if hasattr(sys, 'getwindowsversion') and sys.getwindowsversion().major >= 6: + return test + return unittest.skip("Requires Windows Vista or later")(test) + +@unittest.skipUnless(sys.platform == "win32", "Win32 specific tests") +@support.skip_unless_symlink +class Win32SymlinkTests(unittest.TestCase): + filelink = 'filelinktest' + filelink_target = os.path.abspath(__file__) + dirlink = 'dirlinktest' + dirlink_target = os.path.dirname(filelink_target) + missing_link = 'missing link' + + def setUp(self): + assert os.path.exists(self.dirlink_target) + assert os.path.exists(self.filelink_target) + assert not os.path.exists(self.dirlink) + assert not os.path.exists(self.filelink) + assert not os.path.exists(self.missing_link) + + def tearDown(self): + if os.path.exists(self.filelink): + os.remove(self.filelink) + if os.path.exists(self.dirlink): + os.rmdir(self.dirlink) + if os.path.lexists(self.missing_link): + os.remove(self.missing_link) + + def test_directory_link(self): + os.symlink(self.dirlink_target, self.dirlink) + self.assertTrue(os.path.exists(self.dirlink)) + self.assertTrue(os.path.isdir(self.dirlink)) + self.assertTrue(os.path.islink(self.dirlink)) + self.check_stat(self.dirlink, self.dirlink_target) + + def test_file_link(self): + os.symlink(self.filelink_target, self.filelink) + self.assertTrue(os.path.exists(self.filelink)) + self.assertTrue(os.path.isfile(self.filelink)) + self.assertTrue(os.path.islink(self.filelink)) + self.check_stat(self.filelink, self.filelink_target) + + def _create_missing_dir_link(self): + 'Create a "directory" link to a non-existent target' + linkname = self.missing_link + if os.path.lexists(linkname): + os.remove(linkname) + target = r'c:\\target does not exist.29r3c740' + assert not os.path.exists(target) + target_is_dir = True + os.symlink(target, linkname, target_is_dir) + + def test_remove_directory_link_to_missing_target(self): + self._create_missing_dir_link() + # For compatibility with Unix, os.remove will check the + # directory status and call RemoveDirectory if the symlink + # was created with target_is_dir==True. + os.remove(self.missing_link) + + @unittest.skip("currently fails; consider for improvement") + def test_isdir_on_directory_link_to_missing_target(self): + self._create_missing_dir_link() + # consider having isdir return true for directory links + self.assertTrue(os.path.isdir(self.missing_link)) + + @unittest.skip("currently fails; consider for improvement") + def test_rmdir_on_directory_link_to_missing_target(self): + self._create_missing_dir_link() + # consider allowing rmdir to remove directory links + os.rmdir(self.missing_link) + + def check_stat(self, link, target): + self.assertEqual(os.stat(link), os.stat(target)) + self.assertNotEqual(os.lstat(link), os.stat(link)) + + class MiscTests(unittest.TestCase): @unittest.skipIf(os.name == "nt", "POSIX specific test") @@ -1056,6 +1133,7 @@ def test_main(): PosixUidGidTests, Pep383Tests, Win32KillTests, + Win32SymlinkTests, MiscTests, ) diff --git a/Lib/test/test_platform.py b/Lib/test/test_platform.py index e4051f9..51f47db 100644 --- a/Lib/test/test_platform.py +++ b/Lib/test/test_platform.py @@ -10,20 +10,26 @@ class PlatformTest(unittest.TestCase): def test_architecture(self): res = platform.architecture() - if hasattr(os, "symlink"): - def test_architecture_via_symlink(self): # issue3762 - def get(python): - cmd = [python, '-c', - 'import platform; print(platform.architecture())'] - p = subprocess.Popen(cmd, stdout=subprocess.PIPE) - return p.communicate() - real = os.path.realpath(sys.executable) - link = os.path.abspath(support.TESTFN) - os.symlink(real, link) - try: - self.assertEqual(get(real), get(link)) - finally: - os.remove(link) + @support.skip_unless_symlink + def test_architecture_via_symlink(self): # issue3762 + # On Windows, the EXE needs to know where pythonXY.dll is at so we have + # to add the directory to the path. + if sys.platform == "win32": + os.environ["Path"] = "{};{}".format(os.path.dirname(sys.executable), + os.environ["Path"]) + + def get(python): + cmd = [python, '-c', + 'import platform; print(platform.architecture())'] + p = subprocess.Popen(cmd, stdout=subprocess.PIPE) + return p.communicate() + real = os.path.realpath(sys.executable) + link = os.path.abspath(support.TESTFN) + os.symlink(real, link) + try: + self.assertEqual(get(real), get(link)) + finally: + os.remove(link) def test_platform(self): for aliased in (False, True): diff --git a/Lib/test/test_posixpath.py b/Lib/test/test_posixpath.py index 7ba95f7..34b9689 100644 --- a/Lib/test/test_posixpath.py +++ b/Lib/test/test_posixpath.py @@ -1,7 +1,9 @@ import unittest from test import support, test_genericpath -import posixpath, os +import posixpath +import os +import sys from posixpath import realpath, abspath, dirname, basename # An absolute path to a temporary filename for testing. We can't rely on TESTFN @@ -9,6 +11,16 @@ from posixpath import realpath, abspath, dirname, basename ABSTFN = abspath(support.TESTFN) +def skip_if_ABSTFN_contains_backslash(test): + """ + On Windows, posixpath.abspath still returns paths with backslashes + instead of posix forward slashes. If this is the case, several tests + fail, so skip them. + """ + found_backslash = '\\' in ABSTFN + msg = "ABSTFN is not a posix path - tests fail" + return [test, unittest.skip(msg)(test)][found_backslash] + def safe_rmdir(dirname): try: os.rmdir(dirname) @@ -143,7 +155,7 @@ class PosixPathTest(unittest.TestCase): f.write(b"foo") f.close() self.assertIs(posixpath.islink(support.TESTFN + "1"), False) - if hasattr(os, "symlink"): + if support.can_symlink(): os.symlink(support.TESTFN + "1", support.TESTFN + "2") self.assertIs(posixpath.islink(support.TESTFN + "2"), True) os.remove(support.TESTFN + "1") @@ -154,85 +166,59 @@ class PosixPathTest(unittest.TestCase): if not f.close(): f.close() + @staticmethod + def _create_file(filename): + with open(filename, 'wb') as f: + f.write(b'foo') + def test_samefile(self): - f = open(support.TESTFN + "1", "wb") - try: - f.write(b"foo") - f.close() - self.assertIs( - posixpath.samefile( - support.TESTFN + "1", - support.TESTFN + "1" - ), - True - ) - # If we don't have links, assume that os.stat doesn't return resonable - # inode information and thus, that samefile() doesn't work - if hasattr(os, "symlink"): - os.symlink( - support.TESTFN + "1", - support.TESTFN + "2" - ) - self.assertIs( - posixpath.samefile( - support.TESTFN + "1", - support.TESTFN + "2" - ), - True - ) - os.remove(support.TESTFN + "2") - f = open(support.TESTFN + "2", "wb") - f.write(b"bar") - f.close() - self.assertIs( - posixpath.samefile( - support.TESTFN + "1", - support.TESTFN + "2" - ), - False - ) - finally: - if not f.close(): - f.close() + test_fn = support.TESTFN + "1" + self._create_file(test_fn) + self.assertTrue(posixpath.samefile(test_fn, test_fn)) + self.assertRaises(TypeError, posixpath.samefile) + + @unittest.skipIf( + sys.platform.startswith('win'), + "posixpath.samefile does not work on links in Windows") + @support.skip_unless_symlink + def test_samefile_on_links(self): + test_fn1 = support.TESTFN + "1" + test_fn2 = support.TESTFN + "2" + self._create_file(test_fn1) + + os.symlink(test_fn1, test_fn2) + self.assertTrue(posixpath.samefile(test_fn1, test_fn2)) + os.remove(test_fn2) + + self._create_file(test_fn2) + self.assertFalse(posixpath.samefile(test_fn1, test_fn2)) + def test_samestat(self): - f = open(support.TESTFN + "1", "wb") - try: - f.write(b"foo") - f.close() - self.assertIs( - posixpath.samestat( - os.stat(support.TESTFN + "1"), - os.stat(support.TESTFN + "1") - ), - True - ) - # If we don't have links, assume that os.stat() doesn't return resonable - # inode information and thus, that samefile() doesn't work - if hasattr(os, "symlink"): - if hasattr(os, "symlink"): - os.symlink(support.TESTFN + "1", support.TESTFN + "2") - self.assertIs( - posixpath.samestat( - os.stat(support.TESTFN + "1"), - os.stat(support.TESTFN + "2") - ), - True - ) - os.remove(support.TESTFN + "2") - f = open(support.TESTFN + "2", "wb") - f.write(b"bar") - f.close() - self.assertIs( - posixpath.samestat( - os.stat(support.TESTFN + "1"), - os.stat(support.TESTFN + "2") - ), - False - ) - finally: - if not f.close(): - f.close() + test_fn = support.TESTFN + "1" + self._create_file(test_fn) + test_fns = [test_fn]*2 + stats = map(os.stat, test_fns) + self.assertTrue(posixpath.samestat(*stats)) + + @unittest.skipIf( + sys.platform.startswith('win'), + "posixpath.samestat does not work on links in Windows") + @support.skip_unless_symlink + def test_samestat_on_links(self): + test_fn1 = support.TESTFN + "1" + test_fn2 = support.TESTFN + "2" + test_fns = (test_fn1, test_fn2) + os.symlink(*test_fns) + stats = map(os.stat, test_fns) + self.assertTrue(posixpath.samestat(*stats)) + os.remove(test_fn2) + + self._create_file(test_fn2) + stats = map(os.stat, test_fns) + self.assertFalse(posixpath.samestat(*stats)) + + self.assertRaises(TypeError, posixpath.samestat) def test_ismount(self): self.assertIs(posixpath.ismount("/"), True) @@ -286,103 +272,112 @@ class PosixPathTest(unittest.TestCase): self.assertEqual(posixpath.normpath(b"///..//./foo/.//bar"), b"/foo/bar") - if hasattr(os, "symlink"): - def test_realpath_basic(self): - # Basic operation. - try: - os.symlink(ABSTFN+"1", ABSTFN) - self.assertEqual(realpath(ABSTFN), ABSTFN+"1") - finally: - support.unlink(ABSTFN) - - def test_realpath_symlink_loops(self): - # Bug #930024, return the path unchanged if we get into an infinite - # symlink loop. - try: - old_path = abspath('.') - os.symlink(ABSTFN, ABSTFN) - self.assertEqual(realpath(ABSTFN), ABSTFN) - - os.symlink(ABSTFN+"1", ABSTFN+"2") - os.symlink(ABSTFN+"2", ABSTFN+"1") - self.assertEqual(realpath(ABSTFN+"1"), ABSTFN+"1") - self.assertEqual(realpath(ABSTFN+"2"), ABSTFN+"2") - - # Test using relative path as well. - os.chdir(dirname(ABSTFN)) - self.assertEqual(realpath(basename(ABSTFN)), ABSTFN) - finally: - os.chdir(old_path) - support.unlink(ABSTFN) - support.unlink(ABSTFN+"1") - support.unlink(ABSTFN+"2") - - def test_realpath_resolve_parents(self): - # We also need to resolve any symlinks in the parents of a relative - # path passed to realpath. E.g.: current working directory is - # /usr/doc with 'doc' being a symlink to /usr/share/doc. We call - # realpath("a"). This should return /usr/share/doc/a/. - try: - old_path = abspath('.') - os.mkdir(ABSTFN) - os.mkdir(ABSTFN + "/y") - os.symlink(ABSTFN + "/y", ABSTFN + "/k") - - os.chdir(ABSTFN + "/k") - self.assertEqual(realpath("a"), ABSTFN + "/y/a") - finally: - os.chdir(old_path) - support.unlink(ABSTFN + "/k") - safe_rmdir(ABSTFN + "/y") - safe_rmdir(ABSTFN) - - def test_realpath_resolve_before_normalizing(self): - # Bug #990669: Symbolic links should be resolved before we - # normalize the path. E.g.: if we have directories 'a', 'k' and 'y' - # in the following hierarchy: - # a/k/y - # - # and a symbolic link 'link-y' pointing to 'y' in directory 'a', - # then realpath("link-y/..") should return 'k', not 'a'. - try: - old_path = abspath('.') - os.mkdir(ABSTFN) - os.mkdir(ABSTFN + "/k") - os.mkdir(ABSTFN + "/k/y") - os.symlink(ABSTFN + "/k/y", ABSTFN + "/link-y") - - # Absolute path. - self.assertEqual(realpath(ABSTFN + "/link-y/.."), ABSTFN + "/k") - # Relative path. - os.chdir(dirname(ABSTFN)) - self.assertEqual(realpath(basename(ABSTFN) + "/link-y/.."), - ABSTFN + "/k") - finally: - os.chdir(old_path) - support.unlink(ABSTFN + "/link-y") - safe_rmdir(ABSTFN + "/k/y") - safe_rmdir(ABSTFN + "/k") - safe_rmdir(ABSTFN) - - def test_realpath_resolve_first(self): - # Bug #1213894: The first component of the path, if not absolute, - # must be resolved too. - - try: - old_path = abspath('.') - os.mkdir(ABSTFN) - os.mkdir(ABSTFN + "/k") - os.symlink(ABSTFN, ABSTFN + "link") - os.chdir(dirname(ABSTFN)) - - base = basename(ABSTFN) - self.assertEqual(realpath(base + "link"), ABSTFN) - self.assertEqual(realpath(base + "link/k"), ABSTFN + "/k") - finally: - os.chdir(old_path) - support.unlink(ABSTFN + "link") - safe_rmdir(ABSTFN + "/k") - safe_rmdir(ABSTFN) + @support.skip_unless_symlink + @skip_if_ABSTFN_contains_backslash + def test_realpath_basic(self): + # Basic operation. + try: + os.symlink(ABSTFN+"1", ABSTFN) + self.assertEqual(realpath(ABSTFN), ABSTFN+"1") + finally: + support.unlink(ABSTFN) + + @support.skip_unless_symlink + @skip_if_ABSTFN_contains_backslash + def test_realpath_symlink_loops(self): + # Bug #930024, return the path unchanged if we get into an infinite + # symlink loop. + try: + old_path = abspath('.') + os.symlink(ABSTFN, ABSTFN) + self.assertEqual(realpath(ABSTFN), ABSTFN) + + os.symlink(ABSTFN+"1", ABSTFN+"2") + os.symlink(ABSTFN+"2", ABSTFN+"1") + self.assertEqual(realpath(ABSTFN+"1"), ABSTFN+"1") + self.assertEqual(realpath(ABSTFN+"2"), ABSTFN+"2") + + # Test using relative path as well. + os.chdir(dirname(ABSTFN)) + self.assertEqual(realpath(basename(ABSTFN)), ABSTFN) + finally: + os.chdir(old_path) + support.unlink(ABSTFN) + support.unlink(ABSTFN+"1") + support.unlink(ABSTFN+"2") + + @support.skip_unless_symlink + @skip_if_ABSTFN_contains_backslash + def test_realpath_resolve_parents(self): + # We also need to resolve any symlinks in the parents of a relative + # path passed to realpath. E.g.: current working directory is + # /usr/doc with 'doc' being a symlink to /usr/share/doc. We call + # realpath("a"). This should return /usr/share/doc/a/. + try: + old_path = abspath('.') + os.mkdir(ABSTFN) + os.mkdir(ABSTFN + "/y") + os.symlink(ABSTFN + "/y", ABSTFN + "/k") + + os.chdir(ABSTFN + "/k") + self.assertEqual(realpath("a"), ABSTFN + "/y/a") + finally: + os.chdir(old_path) + support.unlink(ABSTFN + "/k") + safe_rmdir(ABSTFN + "/y") + safe_rmdir(ABSTFN) + + @support.skip_unless_symlink + @skip_if_ABSTFN_contains_backslash + def test_realpath_resolve_before_normalizing(self): + # Bug #990669: Symbolic links should be resolved before we + # normalize the path. E.g.: if we have directories 'a', 'k' and 'y' + # in the following hierarchy: + # a/k/y + # + # and a symbolic link 'link-y' pointing to 'y' in directory 'a', + # then realpath("link-y/..") should return 'k', not 'a'. + try: + old_path = abspath('.') + os.mkdir(ABSTFN) + os.mkdir(ABSTFN + "/k") + os.mkdir(ABSTFN + "/k/y") + os.symlink(ABSTFN + "/k/y", ABSTFN + "/link-y") + + # Absolute path. + self.assertEqual(realpath(ABSTFN + "/link-y/.."), ABSTFN + "/k") + # Relative path. + os.chdir(dirname(ABSTFN)) + self.assertEqual(realpath(basename(ABSTFN) + "/link-y/.."), + ABSTFN + "/k") + finally: + os.chdir(old_path) + support.unlink(ABSTFN + "/link-y") + safe_rmdir(ABSTFN + "/k/y") + safe_rmdir(ABSTFN + "/k") + safe_rmdir(ABSTFN) + + @support.skip_unless_symlink + @skip_if_ABSTFN_contains_backslash + def test_realpath_resolve_first(self): + # Bug #1213894: The first component of the path, if not absolute, + # must be resolved too. + + try: + old_path = abspath('.') + os.mkdir(ABSTFN) + os.mkdir(ABSTFN + "/k") + os.symlink(ABSTFN, ABSTFN + "link") + os.chdir(dirname(ABSTFN)) + + base = basename(ABSTFN) + self.assertEqual(realpath(base + "link"), ABSTFN) + self.assertEqual(realpath(base + "link/k"), ABSTFN + "/k") + finally: + os.chdir(old_path) + support.unlink(ABSTFN + "link") + safe_rmdir(ABSTFN + "/k") + safe_rmdir(ABSTFN) def test_relpath(self): (real_getcwd, os.getcwd) = (os.getcwd, lambda: r"/home/user/bar") diff --git a/Lib/test/test_shutil.py b/Lib/test/test_shutil.py index 9f4871b..baa1943 100644 --- a/Lib/test/test_shutil.py +++ b/Lib/test/test_shutil.py @@ -271,7 +271,7 @@ class TestShutil(unittest.TestCase): shutil.rmtree(src_dir) shutil.rmtree(os.path.dirname(dst_dir)) - @unittest.skipUnless(hasattr(os, 'symlink'), 'requires os.symlink') + @support.skip_unless_symlink def test_dont_copy_file_onto_link_to_itself(self): # bug 851123. os.mkdir(TESTFN) @@ -282,10 +282,11 @@ class TestShutil(unittest.TestCase): f.write('cheddar') f.close() - os.link(src, dst) - self.assertRaises(shutil.Error, shutil.copyfile, src, dst) - self.assertEqual(open(src,'r').read(), 'cheddar') - os.remove(dst) + if hasattr(os, "link"): + os.link(src, dst) + self.assertRaises(shutil.Error, shutil.copyfile, src, dst) + self.assertEqual(open(src,'r').read(), 'cheddar') + os.remove(dst) # Using `src` here would mean we end up with a symlink pointing # to TESTFN/TESTFN/cheese, while it should point at @@ -300,30 +301,30 @@ class TestShutil(unittest.TestCase): except OSError: pass - @unittest.skipUnless(hasattr(os, 'symlink'), 'requires os.symlink') - def test_rmtree_on_symlink(self): - # bug 1669. - os.mkdir(TESTFN) - try: - src = os.path.join(TESTFN, 'cheese') - dst = os.path.join(TESTFN, 'shop') - os.mkdir(src) - os.symlink(src, dst) - self.assertRaises(OSError, shutil.rmtree, dst) - finally: - shutil.rmtree(TESTFN, ignore_errors=True) - - @unittest.skipUnless(hasattr(os, 'mkfifo'), 'requires os.mkfifo') - # Issue #3002: copyfile and copytree block indefinitely on named pipes - def test_copyfile_named_pipe(self): - os.mkfifo(TESTFN) + @support.skip_unless_symlink + def test_rmtree_on_symlink(self): + # bug 1669. + os.mkdir(TESTFN) try: - self.assertRaises(shutil.SpecialFileError, - shutil.copyfile, TESTFN, TESTFN2) - self.assertRaises(shutil.SpecialFileError, - shutil.copyfile, __file__, TESTFN) + src = os.path.join(TESTFN, 'cheese') + dst = os.path.join(TESTFN, 'shop') + os.mkdir(src) + os.symlink(src, dst) + self.assertRaises(OSError, shutil.rmtree, dst) finally: - os.remove(TESTFN) + shutil.rmtree(TESTFN, ignore_errors=True) + + if hasattr(os, "mkfifo"): + # Issue #3002: copyfile and copytree block indefinitely on named pipes + def test_copyfile_named_pipe(self): + os.mkfifo(TESTFN) + try: + self.assertRaises(shutil.SpecialFileError, + shutil.copyfile, TESTFN, TESTFN2) + self.assertRaises(shutil.SpecialFileError, + shutil.copyfile, __file__, TESTFN) + finally: + os.remove(TESTFN) @unittest.skipUnless(hasattr(os, 'mkfifo'), 'requires os.mkfifo') def test_copytree_named_pipe(self): @@ -361,7 +362,7 @@ class TestShutil(unittest.TestCase): shutil.copytree(src_dir, dst_dir, copy_function=_copy) self.assertEquals(len(copied), 2) - @unittest.skipUnless(hasattr(os, 'symlink'), 'requires os.symlink') + @support.skip_unless_symlink def test_copytree_dangling_symlinks(self): # a dangling symlink raises an error at the end diff --git a/Lib/test/test_sys.py b/Lib/test/test_sys.py index d55da62..2f1fffd 100644 --- a/Lib/test/test_sys.py +++ b/Lib/test/test_sys.py @@ -263,7 +263,7 @@ class SysModuleTest(unittest.TestCase): # Raise SkipTest if sys doesn't have getwindowsversion attribute test.support.get_attribute(sys, "getwindowsversion") v = sys.getwindowsversion() - self.assertEqual(len(v), 5) + self.assertEqual(len(v), 9) self.assertIsInstance(v[0], int) self.assertIsInstance(v[1], int) self.assertIsInstance(v[2], int) diff --git a/Lib/test/test_sysconfig.py b/Lib/test/test_sysconfig.py index f4a3c8e..8314e66 100644 --- a/Lib/test/test_sysconfig.py +++ b/Lib/test/test_sysconfig.py @@ -12,7 +12,7 @@ import shutil from copy import copy, deepcopy from test.support import (run_unittest, TESTFN, unlink, get_attribute, - captured_stdout) + captured_stdout, skip_unless_symlink) import sysconfig from sysconfig import (get_paths, get_platform, get_config_vars, @@ -239,17 +239,23 @@ class TestSysConfig(unittest.TestCase): 'posix_home', 'posix_prefix', 'posix_user') self.assertEquals(get_scheme_names(), wanted) + @skip_unless_symlink def test_symlink(self): + # On Windows, the EXE needs to know where pythonXY.dll is at so we have + # to add the directory to the path. + if sys.platform == "win32": + os.environ["Path"] = "{};{}".format(os.path.dirname(sys.executable), + os.environ["Path"]) + # Issue 7880 - symlink = get_attribute(os, "symlink") def get(python): cmd = [python, '-c', 'import sysconfig; print(sysconfig.get_platform())'] - p = subprocess.Popen(cmd, stdout=subprocess.PIPE) + p = subprocess.Popen(cmd, stdout=subprocess.PIPE, env=os.environ) return p.communicate() real = os.path.realpath(sys.executable) link = os.path.abspath(TESTFN) - symlink(real, link) + os.symlink(real, link) try: self.assertEqual(get(real), get(link)) finally: diff --git a/Lib/test/test_tarfile.py b/Lib/test/test_tarfile.py index 858f45b..5c064fc 100644 --- a/Lib/test/test_tarfile.py +++ b/Lib/test/test_tarfile.py @@ -291,6 +291,8 @@ class MiscReadTest(CommonReadTest): self.assertTrue(self.tar.getmembers()[-1].name == "misc/eof", "could not find all members") + @unittest.skipUnless(hasattr(os, "link"), "Missing hardlink implementation") + @support.skip_unless_symlink def test_extract_hardlink(self): # Test hardlink extraction (e.g. bug #857297). tar = tarfile.open(tarname, errorlevel=1, encoding="iso8859-1") @@ -695,16 +697,16 @@ class WriteTest(WriteTestBase): os.remove(target) os.remove(link) + @support.skip_unless_symlink def test_symlink_size(self): - if hasattr(os, "symlink"): - path = os.path.join(TEMPDIR, "symlink") - os.symlink("link_target", path) - try: - tar = tarfile.open(tmpname, self.mode) - tarinfo = tar.gettarinfo(path) - self.assertEqual(tarinfo.size, 0) - finally: - os.remove(path) + path = os.path.join(TEMPDIR, "symlink") + os.symlink("link_target", path) + try: + tar = tarfile.open(tmpname, self.mode) + tarinfo = tar.gettarinfo(path) + self.assertEqual(tarinfo.size, 0) + finally: + os.remove(path) def test_add_self(self): # Test for #1257255. @@ -1408,15 +1410,24 @@ class LinkEmulationTest(ReadTest): data = open(os.path.join(TEMPDIR, name), "rb").read() self.assertEqual(md5sum(data), md5_regtype) + # When 8879 gets fixed, this will need to change. Currently on Windows + # we have os.path.islink but no os.link, so these tests fail without the + # following skip until link is completed. + @unittest.skipIf(hasattr(os.path, "islink"), + "Skip emulation - has os.path.islink but not os.link") def test_hardlink_extraction1(self): self._test_link_extraction("ustar/lnktype") + @unittest.skipIf(hasattr(os.path, "islink"), + "Skip emulation - has os.path.islink but not os.link") def test_hardlink_extraction2(self): self._test_link_extraction("./ustar/linktest2/lnktype") + @unittest.skipIf(hasattr(os, "symlink"), "Skip emulation if symlink exists") def test_symlink_extraction1(self): self._test_link_extraction("ustar/symtype") + @unittest.skipIf(hasattr(os, "symlink"), "Skip emulation if symlink exists") def test_symlink_extraction2(self): self._test_link_extraction("./ustar/linktest2/symtype") diff --git a/Misc/ACKS b/Misc/ACKS index 4608843..ec42bf9 100644 --- a/Misc/ACKS +++ b/Misc/ACKS @@ -154,6 +154,7 @@ Benjamin Collar Jeffery Collins Robert Collins Paul Colomiets +Jason R. Coombs Geremy Condra Juan José Conti Matt Conway diff --git a/Misc/NEWS b/Misc/NEWS index d1edda2..729c010 100644 --- a/Misc/NEWS +++ b/Misc/NEWS @@ -1406,6 +1406,9 @@ Library Extension Modules ----------------- +- Issue #1578269: Implement os.symlink for Windows 6.0+. Patch by + Jason R. Coombs + - In struct.pack, correctly propogate exceptions from computing the truth of an object in the '?' format. diff --git a/Modules/posixmodule.c b/Modules/posixmodule.c index e9ea6cd..6fb990a 100644 --- a/Modules/posixmodule.c +++ b/Modules/posixmodule.c @@ -571,7 +571,7 @@ posix_error_with_allocated_filename(PyObject* name) #ifdef MS_WINDOWS static PyObject * -win32_error(char* function, char* filename) +win32_error(char* function, const char* filename) { /* XXX We should pass the function name along in the future. (winreg.c also wants to pass the function name.) @@ -978,12 +978,28 @@ attributes_from_dir_w(LPCWSTR pszFile, LPWIN32_FILE_ATTRIBUTE_DATA pfad) return TRUE; } -static int -win32_stat(const char* path, struct win32_stat *result) +/* About the following functions: win32_lstat, win32_lstat_w, win32_stat, + win32_stat_w + + In Posix, stat automatically traverses symlinks and returns the stat + structure for the target. In Windows, the equivalent GetFileAttributes by + default does not traverse symlinks and instead returns attributes for + the symlink. + + Therefore, win32_lstat will get the attributes traditionally, and + win32_stat will first explicitly resolve the symlink target and then will + call win32_lstat on that result. + + The _w represent Unicode equivalents of the aformentioned ANSI functions. */ + +static int +win32_lstat(const char* path, struct win32_stat *result) { WIN32_FILE_ATTRIBUTE_DATA info; int code; char *dot; + WIN32_FIND_DATAA find_data; + HANDLE find_data_handle; if (!GetFileAttributesExA(path, GetFileExInfoStandard, &info)) { if (GetLastError() != ERROR_SHARING_VIOLATION) { /* Protocol violation: we explicitly clear errno, instead of @@ -1000,27 +1016,44 @@ win32_stat(const char* path, struct win32_stat *result) } } } + code = attribute_data_to_stat(&info, result); if (code != 0) return code; + + /* Get WIN32_FIND_DATA structure for the path to determine if + it is a symlink */ + if(info.dwFileAttributes & FILE_ATTRIBUTE_REPARSE_POINT) { + find_data_handle = FindFirstFileA(path, &find_data); + if(find_data_handle != INVALID_HANDLE_VALUE) { + if(find_data.dwReserved0 == IO_REPARSE_TAG_SYMLINK) { + /* first clear the S_IFMT bits */ + result->st_mode ^= (result->st_mode & 0170000); + /* now set the bits that make this a symlink */ + result->st_mode |= 0120000; + } + FindClose(find_data_handle); + } + } + /* Set S_IFEXEC if it is an .exe, .bat, ... */ dot = strrchr(path, '.'); if (dot) { - if (stricmp(dot, ".bat") == 0 || - stricmp(dot, ".cmd") == 0 || - stricmp(dot, ".exe") == 0 || - stricmp(dot, ".com") == 0) + if (stricmp(dot, ".bat") == 0 || stricmp(dot, ".cmd") == 0 || + stricmp(dot, ".exe") == 0 || stricmp(dot, ".com") == 0) result->st_mode |= 0111; } return code; } static int -win32_wstat(const wchar_t* path, struct win32_stat *result) +win32_lstat_w(const wchar_t* path, struct win32_stat *result) { int code; const wchar_t *dot; WIN32_FILE_ATTRIBUTE_DATA info; + WIN32_FIND_DATAW find_data; + HANDLE find_data_handle; if (!GetFileAttributesExW(path, GetFileExInfoStandard, &info)) { if (GetLastError() != ERROR_SHARING_VIOLATION) { /* Protocol violation: we explicitly clear errno, instead of @@ -1028,8 +1061,8 @@ win32_wstat(const wchar_t* path, struct win32_stat *result) errno = 0; return -1; } else { - /* Could not get attributes on open file. Fall back to - reading the directory. */ + /* Could not get attributes on open file. Fall back to reading + the directory. */ if (!attributes_from_dir_w(path, &info)) { /* Very strange. This should not fail now */ errno = 0; @@ -1040,18 +1073,198 @@ win32_wstat(const wchar_t* path, struct win32_stat *result) code = attribute_data_to_stat(&info, result); if (code < 0) return code; + + /* Get WIN32_FIND_DATA structure for the path to determine if + it is a symlink */ + if(info.dwFileAttributes & FILE_ATTRIBUTE_REPARSE_POINT) { + find_data_handle = FindFirstFileW(path, &find_data); + if(find_data_handle != INVALID_HANDLE_VALUE) { + if(find_data.dwReserved0 == IO_REPARSE_TAG_SYMLINK) { + /* first clear the S_IFMT bits */ + result->st_mode ^= (result->st_mode & 0170000); + /* now set the bits that make this a symlink */ + result->st_mode |= 0120000; + } + FindClose(find_data_handle); + } + } + /* Set IFEXEC if it is an .exe, .bat, ... */ dot = wcsrchr(path, '.'); if (dot) { - if (_wcsicmp(dot, L".bat") == 0 || - _wcsicmp(dot, L".cmd") == 0 || - _wcsicmp(dot, L".exe") == 0 || - _wcsicmp(dot, L".com") == 0) + if (_wcsicmp(dot, L".bat") == 0 || _wcsicmp(dot, L".cmd") == 0 || + _wcsicmp(dot, L".exe") == 0 || _wcsicmp(dot, L".com") == 0) result->st_mode |= 0111; } return code; } +/* Grab GetFinalPathNameByHandle dynamically from kernel32 */ +static int has_GetFinalPathNameByHandle = 0; +static DWORD (CALLBACK *Py_GetFinalPathNameByHandleA)(HANDLE, LPSTR, DWORD, + DWORD); +static DWORD (CALLBACK *Py_GetFinalPathNameByHandleW)(HANDLE, LPWSTR, DWORD, + DWORD); +static int +check_GetFinalPathNameByHandle() +{ + HINSTANCE hKernel32; + /* only recheck */ + if (!has_GetFinalPathNameByHandle) + { + hKernel32 = GetModuleHandle("KERNEL32"); + *(FARPROC*)&Py_GetFinalPathNameByHandleA = GetProcAddress(hKernel32, + "GetFinalPathNameByHandleA"); + *(FARPROC*)&Py_GetFinalPathNameByHandleW = GetProcAddress(hKernel32, + "GetFinalPathNameByHandleW"); + has_GetFinalPathNameByHandle = Py_GetFinalPathNameByHandleA && Py_GetFinalPathNameByHandleW; + } + return has_GetFinalPathNameByHandle; +} + +static int +win32_stat(const char* path, struct win32_stat *result) +{ + /* Traverse the symlink to the target using + GetFinalPathNameByHandle() + */ + int code; + HANDLE hFile; + int buf_size; + char *target_path; + int result_length; + WIN32_FILE_ATTRIBUTE_DATA info; + + if(!check_GetFinalPathNameByHandle()) { + /* if the OS doesn't have GetFinalPathNameByHandle, it doesn't + have symlinks, so just fall back to the traditional behavior + found in lstat. */ + return win32_lstat(path, result); + } + + hFile = CreateFileA( + path, + 0, /* desired access */ + 0, /* share mode */ + NULL, /* security attributes */ + OPEN_EXISTING, + /* FILE_FLAG_BACKUP_SEMANTICS is required to open a directory */ + FILE_FLAG_BACKUP_SEMANTICS, + NULL); + + if(hFile == INVALID_HANDLE_VALUE) { + /* Either the target doesn't exist, or we don't have access to + get a handle to it. If the former, we need to return an error. + If the latter, we can use attributes_from_dir. */ + if (GetLastError() != ERROR_SHARING_VIOLATION) { + /* Protocol violation: we explicitly clear errno, instead of + setting it to a POSIX error. Callers should use GetLastError. */ + errno = 0; + return -1; + } else { + /* Could not get attributes on open file. Fall back to + reading the directory. */ + if (!attributes_from_dir(path, &info)) { + /* Very strange. This should not fail now */ + errno = 0; + return -1; + } + } + code = attribute_data_to_stat(&info, result); + } + + buf_size = Py_GetFinalPathNameByHandleA(hFile, 0, 0, VOLUME_NAME_DOS); + if(!buf_size) return -1; + target_path = (char *)malloc((buf_size+1)*sizeof(char)); + result_length = Py_GetFinalPathNameByHandleA(hFile, target_path, + buf_size, VOLUME_NAME_DOS); + + if(!result_length) + return -1; + + if(!CloseHandle(hFile)) + return -1; + + target_path[result_length] = 0; + code = win32_lstat(target_path, result); + free(target_path); + + return code; +} + +static int +win32_stat_w(const wchar_t* path, struct win32_stat *result) +{ + /* Traverse the symlink to the target using GetFinalPathNameByHandle() */ + int code; + HANDLE hFile; + int buf_size; + wchar_t *target_path; + int result_length; + WIN32_FILE_ATTRIBUTE_DATA info; + + if(!check_GetFinalPathNameByHandle()) { + /* If the OS doesn't have GetFinalPathNameByHandle, it doesn't have + symlinks, so just fall back to the traditional behavior found + in lstat. */ + return win32_lstat_w(path, result); + } + + hFile = CreateFileW( + path, + 0, /* desired access */ + 0, /* share mode */ + NULL, /* security attributes */ + OPEN_EXISTING, + /* FILE_FLAG_BACKUP_SEMANTICS is required to open a directory */ + FILE_ATTRIBUTE_NORMAL|FILE_FLAG_BACKUP_SEMANTICS, + NULL); + + if(hFile == INVALID_HANDLE_VALUE) { + /* Either the target doesn't exist, or we don't have access to + get a handle to it. If the former, we need to return an error. + If the latter, we can use attributes_from_dir. */ + if (GetLastError() != ERROR_SHARING_VIOLATION) { + /* Protocol violation: we explicitly clear errno, instead of + setting it to a POSIX error. Callers should use GetLastError. */ + errno = 0; + return -1; + } else { + /* Could not get attributes on open file. Fall back to + reading the directory. */ + if (!attributes_from_dir_w(path, &info)) { + /* Very strange. This should not fail now */ + errno = 0; + return -1; + } + } + code = attribute_data_to_stat(&info, result); + } + else { + /* We have a good handle to the target, use it to determine the target + path name (then we'll call lstat on it). */ + buf_size = Py_GetFinalPathNameByHandleW(hFile, 0, 0, VOLUME_NAME_DOS); + if(!buf_size) + return -1; + + target_path = (wchar_t *)malloc((buf_size+1)*sizeof(wchar_t)); + result_length = Py_GetFinalPathNameByHandleW(hFile, target_path, + buf_size, VOLUME_NAME_DOS); + + if(!result_length) + return -1; + + if(!CloseHandle(hFile)) + return -1; + + target_path[result_length] = 0; + code = win32_lstat_w(target_path, result); + free(target_path); + } + + return code; +} + static int win32_fstat(int file_number, struct win32_stat *result) { @@ -2443,6 +2656,69 @@ posix__getfullpathname(PyObject *self, PyObject *args) } return PyBytes_FromString(outbuf); } /* end of posix__getfullpathname */ + +/* A helper function for samepath on windows */ +static PyObject * +posix__getfinalpathname(PyObject *self, PyObject *args) +{ + HANDLE hFile; + int buf_size; + wchar_t *target_path; + int result_length; + PyObject *result; + wchar_t *path; + + if (!PyArg_ParseTuple(args, "u|:_getfullpathname", &path)) { + return NULL; + } + + if(!check_GetFinalPathNameByHandle()) { + /* If the OS doesn't have GetFinalPathNameByHandle, return a + NotImplementedError. */ + return PyErr_Format(PyExc_NotImplementedError, + "GetFinalPathNameByHandle not available on this platform"); + } + + hFile = CreateFileW( + path, + 0, /* desired access */ + 0, /* share mode */ + NULL, /* security attributes */ + OPEN_EXISTING, + /* FILE_FLAG_BACKUP_SEMANTICS is required to open a directory */ + FILE_FLAG_BACKUP_SEMANTICS, + NULL); + + if(hFile == INVALID_HANDLE_VALUE) { + return win32_error_unicode("GetFinalPathNamyByHandle", path); + return PyErr_Format(PyExc_RuntimeError, "Could not get a handle to file."); + } + + /* We have a good handle to the target, use it to determine the + target path name. */ + buf_size = Py_GetFinalPathNameByHandleW(hFile, 0, 0, VOLUME_NAME_NT); + + if(!buf_size) + return win32_error_unicode("GetFinalPathNameByHandle", path); + + target_path = (wchar_t *)malloc((buf_size+1)*sizeof(wchar_t)); + if(!target_path) + return PyErr_NoMemory(); + + result_length = Py_GetFinalPathNameByHandleW(hFile, target_path, + buf_size, VOLUME_NAME_DOS); + if(!result_length) + return win32_error_unicode("GetFinalPathNamyByHandle", path); + + if(!CloseHandle(hFile)) + return win32_error_unicode("GetFinalPathNameByHandle", path); + + target_path[result_length] = 0; + result = PyUnicode_FromUnicode(target_path, result_length); + free(target_path); + return result; + +} /* end of posix__getfinalpathname */ #endif /* MS_WINDOWS */ PyDoc_STRVAR(posix_mkdir__doc__, @@ -2623,7 +2899,7 @@ static PyObject * posix_stat(PyObject *self, PyObject *args) { #ifdef MS_WINDOWS - return posix_do_stat(self, args, "O&:stat", STAT, "U:stat", win32_wstat); + return posix_do_stat(self, args, "O&:stat", STAT, "U:stat", win32_stat_w); #else return posix_do_stat(self, args, "O&:stat", STAT, NULL, NULL); #endif @@ -2681,6 +2957,41 @@ posix_umask(PyObject *self, PyObject *args) return PyLong_FromLong((long)i); } +#ifdef MS_WINDOWS + +/* override the default DeleteFileW behavior so that directory +symlinks can be removed with this function, the same as with +Unix symlinks */ +BOOL WINAPI Py_DeleteFileW(LPCWSTR lpFileName) +{ + WIN32_FILE_ATTRIBUTE_DATA info; + WIN32_FIND_DATAW find_data; + HANDLE find_data_handle; + int is_directory = 0; + int is_link = 0; + + if (GetFileAttributesExW(lpFileName, GetFileExInfoStandard, &info)) { + is_directory = info.dwFileAttributes & FILE_ATTRIBUTE_DIRECTORY; + + /* Get WIN32_FIND_DATA structure for the path to determine if + it is a symlink */ + if(is_directory && + info.dwFileAttributes & FILE_ATTRIBUTE_REPARSE_POINT) { + find_data_handle = FindFirstFileW(lpFileName, &find_data); + + if(find_data_handle != INVALID_HANDLE_VALUE) { + is_link = find_data.dwReserved0 == IO_REPARSE_TAG_SYMLINK; + FindClose(find_data_handle); + } + } + } + + if (is_directory && is_link) + return RemoveDirectoryW(lpFileName); + + return DeleteFileW(lpFileName); +} +#endif /* MS_WINDOWS */ PyDoc_STRVAR(posix_unlink__doc__, "unlink(path)\n\n\ @@ -2694,7 +3005,7 @@ static PyObject * posix_unlink(PyObject *self, PyObject *args) { #ifdef MS_WINDOWS - return win32_1str(args, "remove", "y:remove", DeleteFileA, "U:remove", DeleteFileW); + return win32_1str(args, "remove", "y:remove", DeleteFileA, "U:remove", Py_DeleteFileW); #else return posix_1str(args, "O&:remove", unlink); #endif @@ -4544,7 +4855,8 @@ posix_lstat(PyObject *self, PyObject *args) return posix_do_stat(self, args, "O&:lstat", lstat, NULL, NULL); #else /* !HAVE_LSTAT */ #ifdef MS_WINDOWS - return posix_do_stat(self, args, "O&:lstat", STAT, "U:lstat", win32_wstat); + return posix_do_stat(self, args, "O&:lstat", STAT, "U:lstat", + win32_lstat_w); #else return posix_do_stat(self, args, "O&:lstat", STAT, NULL, NULL); #endif @@ -4609,6 +4921,193 @@ posix_symlink(PyObject *self, PyObject *args) } #endif /* HAVE_SYMLINK */ +#if !defined(HAVE_READLINK) && defined(MS_WINDOWS) + +PyDoc_STRVAR(win_readlink__doc__, +"readlink(path) -> path\n\n\ +Return a string representing the path to which the symbolic link points."); + +/* The following structure was copied from + http://msdn.microsoft.com/en-us/library/ms791514.aspx as the required + include doesn't seem to be present in the Windows SDK (at least as included + with Visual Studio Express). */ +typedef struct _REPARSE_DATA_BUFFER { + ULONG ReparseTag; + USHORT ReparseDataLength; + USHORT Reserved; + union { + struct { + USHORT SubstituteNameOffset; + USHORT SubstituteNameLength; + USHORT PrintNameOffset; + USHORT PrintNameLength; + ULONG Flags; + WCHAR PathBuffer[1]; + } SymbolicLinkReparseBuffer; + + struct { + USHORT SubstituteNameOffset; + USHORT SubstituteNameLength; + USHORT PrintNameOffset; + USHORT PrintNameLength; + WCHAR PathBuffer[1]; + } MountPointReparseBuffer; + + struct { + UCHAR DataBuffer[1]; + } GenericReparseBuffer; + }; +} REPARSE_DATA_BUFFER, *PREPARSE_DATA_BUFFER; + +#define REPARSE_DATA_BUFFER_HEADER_SIZE FIELD_OFFSET(REPARSE_DATA_BUFFER, GenericReparseBuffer) + +#define MAXIMUM_REPARSE_DATA_BUFFER_SIZE ( 16 * 1024 ) + +/* Windows readlink implementation */ +static PyObject * +win_readlink(PyObject *self, PyObject *args) +{ + wchar_t *path; + DWORD n_bytes_returned; + DWORD io_result; + PyObject *result; + HANDLE reparse_point_handle; + + char target_buffer[MAXIMUM_REPARSE_DATA_BUFFER_SIZE]; + REPARSE_DATA_BUFFER *rdb = (REPARSE_DATA_BUFFER *)target_buffer; + wchar_t *print_name; + + if (!PyArg_ParseTuple(args, + "u:readlink", + &path)) + return NULL; + + /* First get a handle to the reparse point */ + Py_BEGIN_ALLOW_THREADS + reparse_point_handle = CreateFileW( + path, + 0, + 0, + 0, + OPEN_EXISTING, + FILE_FLAG_OPEN_REPARSE_POINT|FILE_FLAG_BACKUP_SEMANTICS, + 0); + Py_END_ALLOW_THREADS + + if (reparse_point_handle==INVALID_HANDLE_VALUE) + { + return win32_error_unicode("readlink", path); + } + + Py_BEGIN_ALLOW_THREADS + /* New call DeviceIoControl to read the reparse point */ + io_result = DeviceIoControl( + reparse_point_handle, + FSCTL_GET_REPARSE_POINT, + 0, 0, /* in buffer */ + target_buffer, sizeof(target_buffer), + &n_bytes_returned, + 0 /* we're not using OVERLAPPED_IO */ + ); + CloseHandle(reparse_point_handle); + Py_END_ALLOW_THREADS + + if (io_result==0) + { + return win32_error_unicode("readlink", path); + } + + if (rdb->ReparseTag != IO_REPARSE_TAG_SYMLINK) + { + PyErr_SetString(PyExc_ValueError, + "not a symbolic link"); + return NULL; + } + print_name = rdb->SymbolicLinkReparseBuffer.PathBuffer + rdb->SymbolicLinkReparseBuffer.PrintNameOffset; + result = PyUnicode_FromWideChar(print_name, rdb->SymbolicLinkReparseBuffer.PrintNameLength/2); + return result; +} + +#endif /* !defined(HAVE_READLINK) && defined(MS_WINDOWS) */ + +#if !defined(HAVE_SYMLINK) && defined(MS_WINDOWS) + +/* Grab CreateSymbolicLinkW dynamically from kernel32 */ +static int has_CreateSymbolicLinkW = 0; +static DWORD (CALLBACK *Py_CreateSymbolicLinkW)(LPWSTR, LPWSTR, DWORD); +static int +check_CreateSymbolicLinkW() +{ + HINSTANCE hKernel32; + /* only recheck */ + if (has_CreateSymbolicLinkW) + return has_CreateSymbolicLinkW; + hKernel32 = GetModuleHandle("KERNEL32"); + *(FARPROC*)&Py_CreateSymbolicLinkW = GetProcAddress(hKernel32, "CreateSymbolicLinkW"); + if (Py_CreateSymbolicLinkW) + has_CreateSymbolicLinkW = 1; + return has_CreateSymbolicLinkW; +} + +PyDoc_STRVAR(win_symlink__doc__, +"symlink(src, dst, target_is_directory=False)\n\n\ +Create a symbolic link pointing to src named dst.\n\ +target_is_directory is required if the target is to be interpreted as\n\ +a directory.\n\ +This function requires Windows 6.0 or greater, and raises a\n\ +NotImplementedError otherwise."); + +static PyObject * +win_symlink(PyObject *self, PyObject *args, PyObject *kwargs) +{ + static char *kwlist[] = {"src", "dest", "target_is_directory", NULL}; + PyObject *src, *dest; + int target_is_directory = 0; + DWORD res; + WIN32_FILE_ATTRIBUTE_DATA src_info; + + if (!check_CreateSymbolicLinkW()) + { + /* raise NotImplementedError */ + return PyErr_Format(PyExc_NotImplementedError, + "CreateSymbolicLinkW not found"); + } + if (!PyArg_ParseTupleAndKeywords(args, kwargs, "OO|i:symlink", + kwlist, &src, &dest, &target_is_directory)) + return NULL; + if (!convert_to_unicode(&src)) { return NULL; } + if (!convert_to_unicode(&dest)) { + Py_DECREF(src); + return NULL; + } + + /* if src is a directory, ensure target_is_directory==1 */ + if( + GetFileAttributesExW( + PyUnicode_AsUnicode(src), GetFileExInfoStandard, &src_info + )) + { + target_is_directory = target_is_directory || + (src_info.dwFileAttributes & FILE_ATTRIBUTE_DIRECTORY); + } + + Py_BEGIN_ALLOW_THREADS + res = Py_CreateSymbolicLinkW( + PyUnicode_AsUnicode(dest), + PyUnicode_AsUnicode(src), + target_is_directory); + Py_END_ALLOW_THREADS + Py_DECREF(src); + Py_DECREF(dest); + if (!res) + { + return win32_error_unicode("symlink", PyUnicode_AsUnicode(src)); + } + + Py_INCREF(Py_None); + return Py_None; +} +#endif /* !defined(HAVE_SYMLINK) && defined(MS_WINDOWS) */ #ifdef HAVE_TIMES #if defined(PYCC_VACPP) && defined(PYOS_OS2) @@ -7076,13 +7575,19 @@ static PyMethodDef posix_methods[] = { #ifdef HAVE_READLINK {"readlink", posix_readlink, METH_VARARGS, posix_readlink__doc__}, #endif /* HAVE_READLINK */ - {"rename", posix_rename, METH_VARARGS, posix_rename__doc__}, - {"rmdir", posix_rmdir, METH_VARARGS, posix_rmdir__doc__}, - {"stat", posix_stat, METH_VARARGS, posix_stat__doc__}, +#if !defined(HAVE_READLINK) && defined(MS_WINDOWS) + {"readlink", win_readlink, METH_VARARGS, win_readlink__doc__}, +#endif /* !defined(HAVE_READLINK) && defined(MS_WINDOWS) */ + {"rename", posix_rename, METH_VARARGS, posix_rename__doc__}, + {"rmdir", posix_rmdir, METH_VARARGS, posix_rmdir__doc__}, + {"stat", posix_stat, METH_VARARGS, posix_stat__doc__}, {"stat_float_times", stat_float_times, METH_VARARGS, stat_float_times__doc__}, #ifdef HAVE_SYMLINK {"symlink", posix_symlink, METH_VARARGS, posix_symlink__doc__}, #endif /* HAVE_SYMLINK */ +#if !defined(HAVE_SYMLINK) && defined(MS_WINDOWS) + {"symlink", (PyCFunction)win_symlink, METH_VARARGS | METH_KEYWORDS, win_symlink__doc__}, +#endif /* !defined(HAVE_SYMLINK) && defined(MS_WINDOWS) */ #ifdef HAVE_SYSTEM {"system", posix_system, METH_VARARGS, posix_system__doc__}, #endif @@ -7307,6 +7812,7 @@ static PyMethodDef posix_methods[] = { {"abort", posix_abort, METH_NOARGS, posix_abort__doc__}, #ifdef MS_WINDOWS {"_getfullpathname", posix__getfullpathname, METH_VARARGS, NULL}, + {"_getfinalpathname", posix__getfinalpathname, METH_VARARGS, NULL}, #endif #ifdef HAVE_GETLOADAVG {"getloadavg", posix_getloadavg, METH_NOARGS, posix_getloadavg__doc__}, -- cgit v0.12