pathutil.py
378 lines
| 12.1 KiB
| text/x-python
|
PythonLexer
/ mercurial / pathutil.py
|
r45392 | import contextlib | ||
|
r25964 | import errno | ||
import os | ||||
import posixpath | ||||
import stat | ||||
from .i18n import _ | ||||
from . import ( | ||||
encoding, | ||||
|
r26587 | error, | ||
r43923 | policy, | |||
|
r30614 | pycompat, | ||
|
r25964 | util, | ||
) | ||||
|
r20033 | |||
|
r47392 | if pycompat.TYPE_CHECKING: | ||
from typing import ( | ||||
Any, | ||||
Callable, | ||||
Iterator, | ||||
Optional, | ||||
) | ||||
r43923 | rustdirs = policy.importrust('dirstate', 'Dirs') | |||
parsers = policy.importmod('parsers') | ||||
|
r43346 | |||
|
r23598 | def _lowerclean(s): | ||
|
r47392 | # type: (bytes) -> bytes | ||
|
r23598 | return encoding.hfsignoreclean(s.lower()) | ||
|
r43346 | |||
|
r49801 | class pathauditor: | ||
|
r46554 | """ensure that a filesystem path contains no banned components. | ||
|
r20033 | the following properties of a path are checked: | ||
- ends with a directory separator | ||||
- under top-level .hg | ||||
- starts at the root of a windows drive | ||||
- contains ".." | ||||
|
r27232 | |||
More check are also done about the file system states: | ||||
|
r20033 | - traverses a symlink (e.g. a/symlink_here/b) | ||
- inside a nested repository (a callback can be used to approve | ||||
some nested repositories, e.g., subrepositories) | ||||
|
r27232 | |||
The file system checks are only done when 'realfs' is set to True (the | ||||
default). They should be disable then we are auditing path for operation on | ||||
stored history. | ||||
|
r33722 | |||
If 'cached' is set to True, audited paths and sub-directories are cached. | ||||
Be careful to not keep the cache of unmanaged directories for long because | ||||
audited paths may be replaced with symlinks. | ||||
|
r46554 | """ | ||
|
r20033 | |||
|
r33722 | def __init__(self, root, callback=None, realfs=True, cached=False): | ||
|
r20033 | self.audited = set() | ||
self.auditeddir = set() | ||||
self.root = root | ||||
|
r27232 | self._realfs = realfs | ||
|
r33722 | self._cached = cached | ||
|
r20033 | self.callback = callback | ||
|
r29889 | if os.path.lexists(root) and not util.fscasesensitive(root): | ||
|
r20033 | self.normcase = util.normcase | ||
else: | ||||
self.normcase = lambda x: x | ||||
|
r33435 | def __call__(self, path, mode=None): | ||
|
r47392 | # type: (bytes, Optional[Any]) -> None | ||
|
r46554 | """Check the relative path. | ||
path may contain a pattern (e.g. foodir/**.txt)""" | ||||
|
r20033 | |||
path = util.localpath(path) | ||||
normpath = self.normcase(path) | ||||
if normpath in self.audited: | ||||
return | ||||
# AIX ignores "/" at end of path, others raise EISDIR. | ||||
if util.endswithsep(path): | ||||
|
r49192 | raise error.InputError( | ||
_(b"path ends in directory separator: %s") % path | ||||
) | ||||
|
r20033 | parts = util.splitpath(path) | ||
|
r43346 | if ( | ||
os.path.splitdrive(path)[0] | ||||
|
r43347 | or _lowerclean(parts[0]) in (b'.hg', b'.hg.', b'') | ||
|
r43346 | or pycompat.ospardir in parts | ||
): | ||||
|
r49192 | raise error.InputError( | ||
_(b"path contains illegal component: %s") % path | ||||
) | ||||
|
r23599 | # Windows shortname aliases | ||
for p in parts: | ||||
|
r43347 | if b"~" in p: | ||
first, last = p.split(b"~", 1) | ||||
if last.isdigit() and first.upper() in [b"HG", b"HG8B6C"]: | ||||
|
r49192 | raise error.InputError( | ||
|
r43347 | _(b"path contains illegal component: %s") % path | ||
|
r43346 | ) | ||
|
r43347 | if b'.hg' in _lowerclean(path): | ||
|
r44641 | lparts = [_lowerclean(p) for p in parts] | ||
|
r43347 | for p in b'.hg', b'.hg.': | ||
|
r20033 | if p in lparts[1:]: | ||
pos = lparts.index(p) | ||||
base = os.path.join(*parts[:pos]) | ||||
|
r49192 | raise error.InputError( | ||
|
r43347 | _(b"path '%s' is inside nested repo %r") | ||
|
r43346 | % (path, pycompat.bytestr(base)) | ||
) | ||||
|
r20033 | |||
normparts = util.splitpath(normpath) | ||||
assert len(parts) == len(normparts) | ||||
parts.pop() | ||||
normparts.pop() | ||||
|
r28087 | # It's important that we check the path parts starting from the root. | ||
|
r44834 | # We don't want to add "foo/bar/baz" to auditeddir before checking if | ||
# there's a "foo/.hg" directory. This also means we won't accidentally | ||||
# traverse a symlink into some other filesystem (which is potentially | ||||
# expensive to access). | ||||
|
r28087 | for i in range(len(parts)): | ||
|
r43346 | prefix = pycompat.ossep.join(parts[: i + 1]) | ||
normprefix = pycompat.ossep.join(normparts[: i + 1]) | ||||
|
r20033 | if normprefix in self.auditeddir: | ||
|
r28087 | continue | ||
|
r27232 | if self._realfs: | ||
self._checkfs(prefix, path) | ||||
|
r44656 | if self._cached: | ||
self.auditeddir.add(normprefix) | ||||
|
r20033 | |||
|
r33722 | if self._cached: | ||
self.audited.add(normpath) | ||||
|
r20033 | |||
|
r27231 | def _checkfs(self, prefix, path): | ||
|
r47392 | # type: (bytes, bytes) -> None | ||
|
r27231 | """raise exception if a file system backed check fails""" | ||
curpath = os.path.join(self.root, prefix) | ||||
try: | ||||
st = os.lstat(curpath) | ||||
except OSError as err: | ||||
# EINVAL can be raised as invalid path syntax under win32. | ||||
# They must be ignored for patterns can be checked too. | ||||
if err.errno not in (errno.ENOENT, errno.ENOTDIR, errno.EINVAL): | ||||
raise | ||||
else: | ||||
if stat.S_ISLNK(st.st_mode): | ||||
|
r43347 | msg = _(b'path %r traverses symbolic link %r') % ( | ||
|
r43346 | pycompat.bytestr(path), | ||
pycompat.bytestr(prefix), | ||||
) | ||||
|
r27235 | raise error.Abort(msg) | ||
|
r43346 | elif stat.S_ISDIR(st.st_mode) and os.path.isdir( | ||
|
r43347 | os.path.join(curpath, b'.hg') | ||
|
r43346 | ): | ||
|
r27231 | if not self.callback or not self.callback(curpath): | ||
|
r43347 | msg = _(b"path '%s' is inside nested repo %r") | ||
|
r36667 | raise error.Abort(msg % (path, pycompat.bytestr(prefix))) | ||
|
r27231 | |||
|
r20033 | def check(self, path): | ||
|
r47392 | # type: (bytes) -> bool | ||
|
r20033 | try: | ||
self(path) | ||||
return True | ||||
|
r26587 | except (OSError, error.Abort): | ||
|
r20033 | return False | ||
|
r45392 | @contextlib.contextmanager | ||
def cached(self): | ||||
if self._cached: | ||||
yield | ||||
else: | ||||
try: | ||||
self._cached = True | ||||
yield | ||||
finally: | ||||
self.audited.clear() | ||||
self.auditeddir.clear() | ||||
self._cached = False | ||||
|
r43346 | |||
|
r20033 | def canonpath(root, cwd, myname, auditor=None): | ||
|
r47392 | # type: (bytes, bytes, bytes, Optional[pathauditor]) -> bytes | ||
|
r46554 | """return the canonical path of myname, given cwd and root | ||
|
r34981 | |||
>>> def check(root, cwd, myname): | ||||
... a = pathauditor(root, realfs=False) | ||||
... try: | ||||
... return canonpath(root, cwd, myname, a) | ||||
... except error.Abort: | ||||
... return 'aborted' | ||||
>>> def unixonly(root, cwd, myname, expected='aborted'): | ||||
... if pycompat.iswindows: | ||||
... return expected | ||||
... return check(root, cwd, myname) | ||||
>>> def winonly(root, cwd, myname, expected='aborted'): | ||||
... if not pycompat.iswindows: | ||||
... return expected | ||||
... return check(root, cwd, myname) | ||||
>>> winonly(b'd:\\\\repo', b'c:\\\\dir', b'filename') | ||||
'aborted' | ||||
>>> winonly(b'c:\\\\repo', b'c:\\\\dir', b'filename') | ||||
'aborted' | ||||
>>> winonly(b'c:\\\\repo', b'c:\\\\', b'filename') | ||||
'aborted' | ||||
>>> winonly(b'c:\\\\repo', b'c:\\\\', b'repo\\\\filename', | ||||
... b'filename') | ||||
'filename' | ||||
>>> winonly(b'c:\\\\repo', b'c:\\\\repo', b'filename', b'filename') | ||||
'filename' | ||||
>>> winonly(b'c:\\\\repo', b'c:\\\\repo\\\\subdir', b'filename', | ||||
... b'subdir/filename') | ||||
'subdir/filename' | ||||
>>> unixonly(b'/repo', b'/dir', b'filename') | ||||
'aborted' | ||||
>>> unixonly(b'/repo', b'/', b'filename') | ||||
'aborted' | ||||
>>> unixonly(b'/repo', b'/', b'repo/filename', b'filename') | ||||
'filename' | ||||
>>> unixonly(b'/repo', b'/repo', b'filename', b'filename') | ||||
'filename' | ||||
>>> unixonly(b'/repo', b'/repo/subdir', b'filename', b'subdir/filename') | ||||
'subdir/filename' | ||||
|
r46554 | """ | ||
|
r20033 | if util.endswithsep(root): | ||
rootsep = root | ||||
else: | ||||
|
r30614 | rootsep = root + pycompat.ossep | ||
|
r20033 | name = myname | ||
if not os.path.isabs(name): | ||||
name = os.path.join(root, cwd, name) | ||||
name = os.path.normpath(name) | ||||
if auditor is None: | ||||
auditor = pathauditor(root) | ||||
if name != rootsep and name.startswith(rootsep): | ||||
|
r43346 | name = name[len(rootsep) :] | ||
|
r20033 | auditor(name) | ||
return util.pconvert(name) | ||||
elif name == root: | ||||
|
r43347 | return b'' | ||
|
r20033 | else: | ||
# Determine whether `name' is in the hierarchy at or beneath `root', | ||||
# by iterating name=dirname(name) until that causes no change (can't | ||||
# check name == '/', because that doesn't work on windows). The list | ||||
# `rel' holds the reversed list of components making up the relative | ||||
# file name we want. | ||||
rel = [] | ||||
while True: | ||||
try: | ||||
s = util.samefile(name, root) | ||||
except OSError: | ||||
s = False | ||||
if s: | ||||
if not rel: | ||||
# name was actually the same as root (maybe a symlink) | ||||
|
r43347 | return b'' | ||
|
r20033 | rel.reverse() | ||
name = os.path.join(*rel) | ||||
auditor(name) | ||||
return util.pconvert(name) | ||||
dirname, basename = util.split(name) | ||||
rel.append(basename) | ||||
if dirname == name: | ||||
break | ||||
name = dirname | ||||
|
r25011 | # A common mistake is to use -R, but specify a file relative to the repo | ||
# instead of cwd. Detect that case, and provide a hint to the user. | ||||
hint = None | ||||
try: | ||||
|
r25022 | if cwd != root: | ||
canonpath(root, root, myname, auditor) | ||||
|
r43347 | relpath = util.pathto(root, cwd, b'') | ||
|
r38611 | if relpath.endswith(pycompat.ossep): | ||
|
r34966 | relpath = relpath[:-1] | ||
|
r43347 | hint = _(b"consider using '--cwd %s'") % relpath | ||
|
r26587 | except error.Abort: | ||
|
r25011 | pass | ||
|
r43346 | raise error.Abort( | ||
|
r43347 | _(b"%s not under root '%s'") % (myname, root), hint=hint | ||
|
r43346 | ) | ||
|
r21568 | |||
def normasprefix(path): | ||||
|
r47392 | # type: (bytes) -> bytes | ||
|
r46554 | """normalize the specified path as path prefix | ||
|
r21568 | |||
|
r23139 | Returned value can be used safely for "p.startswith(prefix)", | ||
|
r21568 | "p[len(prefix):]", and so on. | ||
For efficiency, this expects "path" argument to be already | ||||
normalized by "os.path.normpath", "os.path.realpath", and so on. | ||||
See also issue3033 for detail about need of this function. | ||||
|
r34255 | >>> normasprefix(b'/foo/bar').replace(pycompat.ossep, b'/') | ||
|
r21568 | '/foo/bar/' | ||
|
r34255 | >>> normasprefix(b'/').replace(pycompat.ossep, b'/') | ||
|
r21568 | '/' | ||
|
r46554 | """ | ||
|
r21568 | d, p = os.path.splitdrive(path) | ||
|
r30614 | if len(p) != len(pycompat.ossep): | ||
return path + pycompat.ossep | ||||
|
r21568 | else: | ||
return path | ||||
|
r25281 | |||
|
r43346 | |||
|
r44032 | def finddirs(path): | ||
|
r47392 | # type: (bytes) -> Iterator[bytes] | ||
|
r44032 | pos = path.rfind(b'/') | ||
while pos != -1: | ||||
yield path[:pos] | ||||
pos = path.rfind(b'/', 0, pos) | ||||
yield b'' | ||||
|
r49801 | class dirs: | ||
r43923 | '''a multiset of directory names from a set of file paths''' | |||
r48756 | def __init__(self, map, only_tracked=False): | |||
|
r46554 | """ | ||
|
r45116 | a dict map indicates a dirstate while a list indicates a manifest | ||
|
r46554 | """ | ||
r43923 | self._dirs = {} | |||
addpath = self.addpath | ||||
r48756 | if isinstance(map, dict) and only_tracked: | |||
|
r49768 | for f, s in map.items(): | ||
r48756 | if s.state != b'r': | |||
r43923 | addpath(f) | |||
r48756 | elif only_tracked: | |||
msg = b"`only_tracked` is only supported with a dict source" | ||||
raise error.ProgrammingError(msg) | ||||
r43923 | else: | |||
for f in map: | ||||
addpath(f) | ||||
def addpath(self, path): | ||||
|
r47392 | # type: (bytes) -> None | ||
r43923 | dirs = self._dirs | |||
|
r44032 | for base in finddirs(path): | ||
r43923 | if base.endswith(b'/'): | |||
raise ValueError( | ||||
"found invalid consecutive slashes in path: %r" % base | ||||
) | ||||
if base in dirs: | ||||
dirs[base] += 1 | ||||
return | ||||
dirs[base] = 1 | ||||
def delpath(self, path): | ||||
|
r47392 | # type: (bytes) -> None | ||
r43923 | dirs = self._dirs | |||
|
r44032 | for base in finddirs(path): | ||
r43923 | if dirs[base] > 1: | |||
dirs[base] -= 1 | ||||
return | ||||
del dirs[base] | ||||
def __iter__(self): | ||||
return iter(self._dirs) | ||||
def __contains__(self, d): | ||||
|
r47392 | # type: (bytes) -> bool | ||
r43923 | return d in self._dirs | |||
if util.safehasattr(parsers, 'dirs'): | ||||
dirs = parsers.dirs | ||||
if rustdirs is not None: | ||||
dirs = rustdirs | ||||
|
r25286 | # 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 | ||||
# - instead we'll let them be oblivious. | ||||
join = posixpath.join | ||||
|
r47392 | dirname = posixpath.dirname # type: Callable[[bytes], bytes] | ||