| hg-reviewers |
| Automatic diff as part of commit; lint not applicable. |
| Automatic diff as part of commit; unit tests not applicable. |
| Path | Packages | |||
|---|---|---|---|---|
| M | mercurial/pathutil.py (21 lines) |
| Status | Author | Revision | |
|---|---|---|---|
| Closed | mharbison72 | ||
| Closed | mharbison72 |
| from . import ( | from . import ( | ||||
| encoding, | encoding, | ||||
| error, | error, | ||||
| policy, | policy, | ||||
| pycompat, | pycompat, | ||||
| util, | util, | ||||
| ) | ) | ||||
| if pycompat.TYPE_CHECKING: | |||||
| from typing import ( | |||||
| Any, | |||||
| Callable, | |||||
| Iterator, | |||||
| Optional, | |||||
| ) | |||||
| rustdirs = policy.importrust('dirstate', 'Dirs') | rustdirs = policy.importrust('dirstate', 'Dirs') | ||||
| parsers = policy.importmod('parsers') | parsers = policy.importmod('parsers') | ||||
| def _lowerclean(s): | def _lowerclean(s): | ||||
| # type: (bytes) -> bytes | |||||
| return encoding.hfsignoreclean(s.lower()) | return encoding.hfsignoreclean(s.lower()) | ||||
| class pathauditor(object): | class pathauditor(object): | ||||
| """ensure that a filesystem path contains no banned components. | """ensure that a filesystem path contains no banned components. | ||||
| the following properties of a path are checked: | the following properties of a path are checked: | ||||
| - ends with a directory separator | - ends with a directory separator | ||||
| self._cached = cached | self._cached = cached | ||||
| self.callback = callback | self.callback = callback | ||||
| if os.path.lexists(root) and not util.fscasesensitive(root): | if os.path.lexists(root) and not util.fscasesensitive(root): | ||||
| self.normcase = util.normcase | self.normcase = util.normcase | ||||
| else: | else: | ||||
| self.normcase = lambda x: x | self.normcase = lambda x: x | ||||
| def __call__(self, path, mode=None): | def __call__(self, path, mode=None): | ||||
| # type: (bytes, Optional[Any]) -> None | |||||
| """Check the relative path. | """Check the relative path. | ||||
| path may contain a pattern (e.g. foodir/**.txt)""" | path may contain a pattern (e.g. foodir/**.txt)""" | ||||
| path = util.localpath(path) | path = util.localpath(path) | ||||
| normpath = self.normcase(path) | normpath = self.normcase(path) | ||||
| if normpath in self.audited: | if normpath in self.audited: | ||||
| return | return | ||||
| # AIX ignores "/" at end of path, others raise EISDIR. | # AIX ignores "/" at end of path, others raise EISDIR. | ||||
| self._checkfs(prefix, path) | self._checkfs(prefix, path) | ||||
| if self._cached: | if self._cached: | ||||
| self.auditeddir.add(normprefix) | self.auditeddir.add(normprefix) | ||||
| if self._cached: | if self._cached: | ||||
| self.audited.add(normpath) | self.audited.add(normpath) | ||||
| def _checkfs(self, prefix, path): | def _checkfs(self, prefix, path): | ||||
| # type: (bytes, bytes) -> None | |||||
| """raise exception if a file system backed check fails""" | """raise exception if a file system backed check fails""" | ||||
| curpath = os.path.join(self.root, prefix) | curpath = os.path.join(self.root, prefix) | ||||
| try: | try: | ||||
| st = os.lstat(curpath) | st = os.lstat(curpath) | ||||
| except OSError as err: | except OSError as err: | ||||
| # EINVAL can be raised as invalid path syntax under win32. | # EINVAL can be raised as invalid path syntax under win32. | ||||
| # They must be ignored for patterns can be checked too. | # They must be ignored for patterns can be checked too. | ||||
| if err.errno not in (errno.ENOENT, errno.ENOTDIR, errno.EINVAL): | if err.errno not in (errno.ENOENT, errno.ENOTDIR, errno.EINVAL): | ||||
| raise | raise | ||||
| else: | else: | ||||
| if stat.S_ISLNK(st.st_mode): | if stat.S_ISLNK(st.st_mode): | ||||
| msg = _(b'path %r traverses symbolic link %r') % ( | msg = _(b'path %r traverses symbolic link %r') % ( | ||||
| pycompat.bytestr(path), | pycompat.bytestr(path), | ||||
| pycompat.bytestr(prefix), | pycompat.bytestr(prefix), | ||||
| ) | ) | ||||
| raise error.Abort(msg) | raise error.Abort(msg) | ||||
| elif stat.S_ISDIR(st.st_mode) and os.path.isdir( | elif stat.S_ISDIR(st.st_mode) and os.path.isdir( | ||||
| os.path.join(curpath, b'.hg') | os.path.join(curpath, b'.hg') | ||||
| ): | ): | ||||
| if not self.callback or not self.callback(curpath): | if not self.callback or not self.callback(curpath): | ||||
| msg = _(b"path '%s' is inside nested repo %r") | msg = _(b"path '%s' is inside nested repo %r") | ||||
| raise error.Abort(msg % (path, pycompat.bytestr(prefix))) | raise error.Abort(msg % (path, pycompat.bytestr(prefix))) | ||||
| def check(self, path): | def check(self, path): | ||||
| # type: (bytes) -> bool | |||||
| try: | try: | ||||
| self(path) | self(path) | ||||
| return True | return True | ||||
| except (OSError, error.Abort): | except (OSError, error.Abort): | ||||
| return False | return False | ||||
| @contextlib.contextmanager | @contextlib.contextmanager | ||||
| def cached(self): | def cached(self): | ||||
| if self._cached: | if self._cached: | ||||
| yield | yield | ||||
| else: | else: | ||||
| try: | try: | ||||
| self._cached = True | self._cached = True | ||||
| yield | yield | ||||
| finally: | finally: | ||||
| self.audited.clear() | self.audited.clear() | ||||
| self.auditeddir.clear() | self.auditeddir.clear() | ||||
| self._cached = False | self._cached = False | ||||
| def canonpath(root, cwd, myname, auditor=None): | def canonpath(root, cwd, myname, auditor=None): | ||||
| # type: (bytes, bytes, bytes, Optional[pathauditor]) -> bytes | |||||
| """return the canonical path of myname, given cwd and root | """return the canonical path of myname, given cwd and root | ||||
| >>> def check(root, cwd, myname): | >>> def check(root, cwd, myname): | ||||
| ... a = pathauditor(root, realfs=False) | ... a = pathauditor(root, realfs=False) | ||||
| ... try: | ... try: | ||||
| ... return canonpath(root, cwd, myname, a) | ... return canonpath(root, cwd, myname, a) | ||||
| ... except error.Abort: | ... except error.Abort: | ||||
| ... return 'aborted' | ... return 'aborted' | ||||
| pass | pass | ||||
| raise error.Abort( | raise error.Abort( | ||||
| _(b"%s not under root '%s'") % (myname, root), hint=hint | _(b"%s not under root '%s'") % (myname, root), hint=hint | ||||
| ) | ) | ||||
| def normasprefix(path): | def normasprefix(path): | ||||
| # type: (bytes) -> bytes | |||||
| """normalize the specified path as path prefix | """normalize the specified path as path prefix | ||||
| Returned value can be used safely for "p.startswith(prefix)", | Returned value can be used safely for "p.startswith(prefix)", | ||||
| "p[len(prefix):]", and so on. | "p[len(prefix):]", and so on. | ||||
| For efficiency, this expects "path" argument to be already | For efficiency, this expects "path" argument to be already | ||||
| normalized by "os.path.normpath", "os.path.realpath", and so on. | normalized by "os.path.normpath", "os.path.realpath", and so on. | ||||
| See also issue3033 for detail about need of this function. | See also issue3033 for detail about need of this function. | ||||
| >>> normasprefix(b'/foo/bar').replace(pycompat.ossep, b'/') | >>> normasprefix(b'/foo/bar').replace(pycompat.ossep, b'/') | ||||
| '/foo/bar/' | '/foo/bar/' | ||||
| >>> normasprefix(b'/').replace(pycompat.ossep, b'/') | >>> normasprefix(b'/').replace(pycompat.ossep, b'/') | ||||
| '/' | '/' | ||||
| """ | """ | ||||
| d, p = os.path.splitdrive(path) | d, p = os.path.splitdrive(path) | ||||
| if len(p) != len(pycompat.ossep): | if len(p) != len(pycompat.ossep): | ||||
| return path + pycompat.ossep | return path + pycompat.ossep | ||||
| else: | else: | ||||
| return path | return path | ||||
| def finddirs(path): | def finddirs(path): | ||||
| # type: (bytes) -> Iterator[bytes] | |||||
| pos = path.rfind(b'/') | pos = path.rfind(b'/') | ||||
| while pos != -1: | while pos != -1: | ||||
| yield path[:pos] | yield path[:pos] | ||||
| pos = path.rfind(b'/', 0, pos) | pos = path.rfind(b'/', 0, pos) | ||||
| yield b'' | yield b'' | ||||
| class dirs(object): | class dirs(object): | ||||
| raise error.ProgrammingError( | raise error.ProgrammingError( | ||||
| b"skip character is only supported with a dict source" | b"skip character is only supported with a dict source" | ||||
| ) | ) | ||||
| else: | else: | ||||
| for f in map: | for f in map: | ||||
| addpath(f) | addpath(f) | ||||
| def addpath(self, path): | def addpath(self, path): | ||||
| # type: (bytes) -> None | |||||
| dirs = self._dirs | dirs = self._dirs | ||||
| for base in finddirs(path): | for base in finddirs(path): | ||||
| if base.endswith(b'/'): | if base.endswith(b'/'): | ||||
| raise ValueError( | raise ValueError( | ||||
| "found invalid consecutive slashes in path: %r" % base | "found invalid consecutive slashes in path: %r" % base | ||||
| ) | ) | ||||
| if base in dirs: | if base in dirs: | ||||
| dirs[base] += 1 | dirs[base] += 1 | ||||
| return | return | ||||
| dirs[base] = 1 | dirs[base] = 1 | ||||
| def delpath(self, path): | def delpath(self, path): | ||||
| # type: (bytes) -> None | |||||
| dirs = self._dirs | dirs = self._dirs | ||||
| for base in finddirs(path): | for base in finddirs(path): | ||||
| if dirs[base] > 1: | if dirs[base] > 1: | ||||
| dirs[base] -= 1 | dirs[base] -= 1 | ||||
| return | return | ||||
| del dirs[base] | del dirs[base] | ||||
| def __iter__(self): | def __iter__(self): | ||||
| return iter(self._dirs) | return iter(self._dirs) | ||||
| def __contains__(self, d): | def __contains__(self, d): | ||||
| # type: (bytes) -> bool | |||||
| return d in self._dirs | return d in self._dirs | ||||
| if util.safehasattr(parsers, 'dirs'): | if util.safehasattr(parsers, 'dirs'): | ||||
| dirs = parsers.dirs | dirs = parsers.dirs | ||||
| if rustdirs is not None: | if rustdirs is not None: | ||||
| dirs = rustdirs | dirs = rustdirs | ||||
| # forward two methods from posixpath that do what we need, but we'd | # forward two methods from posixpath that do what we need, but we'd | ||||
| # rather not let our internals know that we're thinking in posix terms | # rather not let our internals know that we're thinking in posix terms | ||||
| # - instead we'll let them be oblivious. | # - instead we'll let them be oblivious. | ||||
| join = posixpath.join | join = posixpath.join | ||||
| dirname = posixpath.dirname | dirname = posixpath.dirname # type: Callable[[bytes], bytes] | ||||