# posix.py - Posix utility function implementations for Mercurial

#

#  Copyright 2005-2009 Olivia Mackall <olivia@selenic.com> and others

#

# This software may be used and distributed according to the terms of the

# GNU General Public License version 2 or any later version.

from __future__ import annotations

import errno

import fcntl

import getpass

import grp

import os

import pwd

import re

import select

import stat

import sys

import tempfile

import typing

import unicodedata

from typing import (

    Any,

    AnyStr,

    Callable,

    Iterable,

    Iterator,

    List,

    Match,

    NoReturn,

    Optional,

    Sequence,

    Tuple,

    Union,

)

from .i18n import _

from . import (

    encoding,

    error,

    policy,

    pycompat,

)

osutil = policy.importmod('osutil')

normpath = os.path.normpath

samestat = os.path.samestat

abspath = os.path.abspath  # re-exports

try:

    oslink = os.link

except AttributeError:

    # Some platforms build Python without os.link on systems that are

    # vaguely unix-like but don't have hardlink support. For those

    # poor souls, just say we tried and that it failed so we fall back

    # to copies.

    def oslink(src: bytes, dst: bytes) -> NoReturn:

        raise OSError(

            errno.EINVAL, b'hardlinks not supported: %s to %s' % (src, dst)

        )

readlink = os.readlink

unlink = os.unlink

rename = os.rename

removedirs = os.removedirs

if typing.TYPE_CHECKING:

    def normpath(path: bytes) -> bytes:

        raise NotImplementedError

    def abspath(path: AnyStr) -> AnyStr:

        raise NotImplementedError

    def oslink(src: bytes, dst: bytes) -> None:

        raise NotImplementedError

    def readlink(path: bytes) -> bytes:

        raise NotImplementedError

    def unlink(path: bytes) -> None:

        raise NotImplementedError

    def rename(src: bytes, dst: bytes) -> None:

        raise NotImplementedError

    def removedirs(name: bytes) -> None:

        raise NotImplementedError

expandglobs: bool = False

umask: int = os.umask(0)

os.umask(umask)

posixfile = pycompat.open

def split(p: bytes) -> Tuple[bytes, bytes]:

    """Same as posixpath.split, but faster

    >>> import posixpath

    >>> for f in [b'/absolute/path/to/file',

    ...           b'relative/path/to/file',

    ...           b'file_alone',

    ...           b'path/to/directory/',

    ...           b'/multiple/path//separators',

    ...           b'/file_at_root',

    ...           b'///multiple_leading_separators_at_root',

    ...           b'']:

    ...     assert split(f) == posixpath.split(f), f

    """

    ht = p.rsplit(b'/', 1)

    if len(ht) == 1:

        return b'', p

    nh = ht[0].rstrip(b'/')

    if nh:

        return nh, ht[1]

    return ht[0] + b'/', ht[1]

def openhardlinks() -> bool:

    '''return true if it is safe to hold open file handles to hardlinks'''

    return True

def nlinks(name: bytes) -> int:

    '''return number of hardlinks for the given file'''

    return os.lstat(name).st_nlink

def parsepatchoutput(output_line: bytes) -> bytes:

    """parses the output produced by patch and returns the filename"""

    pf = output_line[14:]

    if pycompat.sysplatform == b'OpenVMS':

        if pf[0] == b'`':

            pf = pf[1:-1]  # Remove the quotes

    else:

        if pf.startswith(b"'") and pf.endswith(b"'") and b" " in pf:

            pf = pf[1:-1]  # Remove the quotes

    return pf

def sshargs(

    sshcmd: bytes, host: bytes, user: Optional[bytes], port: Optional[bytes]

) -> bytes:

    '''Build argument list for ssh'''

    args = user and (b"%s@%s" % (user, host)) or host

    if b'-' in args[:1]:

        raise error.Abort(

            _(b'illegal ssh hostname or username starting with -: %s') % args

        )

    args = shellquote(args)

    if port:

        args = b'-p %s %s' % (shellquote(port), args)

    return args

def isexec(f: bytes) -> bool:

    """check whether a file is executable"""

    return os.lstat(f).st_mode & 0o100 != 0

def setflags(f: bytes, l: bool, x: bool) -> None:

    st = os.lstat(f)

    s = st.st_mode

    if l:

        if not stat.S_ISLNK(s):

            # switch file to link

            with open(f, 'rb') as fp:

                data = fp.read()

            unlink(f)

            try:

                os.symlink(data, f)

            except OSError:

                # failed to make a link, rewrite file

                with open(f, "wb") as fp:

                    fp.write(data)

        # no chmod needed at this point

        return

    if stat.S_ISLNK(s):

        # switch link to file

        data = os.readlink(f)

        unlink(f)

        with open(f, "wb") as fp:

            fp.write(data)

        s = 0o666 & ~umask  # avoid restatting for chmod

    sx = s & 0o100

    if st.st_nlink > 1 and bool(x) != bool(sx):

        # the file is a hardlink, break it

        with open(f, "rb") as fp:

            data = fp.read()

        unlink(f)

        with open(f, "wb") as fp:

            fp.write(data)

    if x and not sx:

        # Turn on +x for every +r bit when making a file executable

        # and obey umask.

        os.chmod(f, s | (s & 0o444) >> 2 & ~umask)

    elif not x and sx:

        # Turn off all +x bits

        os.chmod(f, s & 0o666)

def copymode(

    src: bytes,

    dst: bytes,

    mode: Optional[int] = None,

    enforcewritable: bool = False,

) -> None:

    """Copy the file mode from the file at path src to dst.

    If src doesn't exist, we're using mode instead. If mode is None, we're

    using umask."""

    try:

        st_mode = os.lstat(src).st_mode & 0o777

    except FileNotFoundError:

        st_mode = mode

        if st_mode is None:

            st_mode = ~umask

        st_mode &= 0o666

    new_mode = st_mode

    if enforcewritable:

        new_mode |= stat.S_IWUSR

    os.chmod(dst, new_mode)

def checkexec(path: bytes) -> bool:

    """

    Check whether the given path is on a filesystem with UNIX-like exec flags

    Requires a directory (like /foo/.hg)

    """

    # VFAT on some Linux versions can flip mode but it doesn't persist

    # a FS remount. Frequently we can detect it if files are created

    # with exec bit on.

    try:

        EXECFLAGS = stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH

        basedir = os.path.join(path, b'.hg')

        cachedir = os.path.join(basedir, b'wcache')

        storedir = os.path.join(basedir, b'store')

        if not os.path.exists(cachedir):

            try:

                # we want to create the 'cache' directory, not the '.hg' one.

                # Automatically creating '.hg' directory could silently spawn

                # invalid Mercurial repositories. That seems like a bad idea.

                os.mkdir(cachedir)

                if os.path.exists(storedir):

                    copymode(storedir, cachedir)

                else:

                    copymode(basedir, cachedir)

                # we other fallback logic triggers

                pass

        if os.path.isdir(cachedir):

            checkisexec = os.path.join(cachedir, b'checkisexec')

            checknoexec = os.path.join(cachedir, b'checknoexec')

            try:

                m = os.stat(checkisexec).st_mode

            except FileNotFoundError:

                # checkisexec does not exist - fall through ...

                pass

            else:

                # checkisexec exists, check if it actually is exec

                if m & EXECFLAGS != 0:

                    # ensure checknoexec exists, check it isn't exec

                    try:

                        m = os.stat(checknoexec).st_mode

                    except FileNotFoundError:

                        open(checknoexec, 'w').close()  # might fail

                        m = os.stat(checknoexec).st_mode

                    if m & EXECFLAGS == 0:

                        # check-exec is exec and check-no-exec is not exec

                        return True

                    # checknoexec exists but is exec - delete it

                    unlink(checknoexec)

                # checkisexec exists but is not exec - delete it

                unlink(checkisexec)

            # check using one file, leave it as checkisexec

            checkdir = cachedir

        else:

            # check directly in path and don't leave checkisexec behind

            checkdir = path

            checkisexec = None

        fh, fn = pycompat.mkstemp(dir=checkdir, prefix=b'hg-checkexec-')

        try:

            os.close(fh)

            m = os.stat(fn).st_mode

            if m & EXECFLAGS == 0:

                m = 0o666 & ~umask

                os.chmod(fn, m & 0o777 | EXECFLAGS)

                if os.stat(fn).st_mode & EXECFLAGS != 0:

                    if checkisexec is not None:

                        os.rename(fn, checkisexec)

                        fn = None

                    return True

        finally:

            if fn is not None:

                unlink(fn)

        # we don't care, the user probably won't be able to commit anyway

        return False

def checklink(path: bytes) -> bool:

    """check whether the given path is on a symlink-capable filesystem"""

    # mktemp is not racy because symlink creation will fail if the

    # file already exists

    while True:

        cachedir = os.path.join(path, b'.hg', b'wcache')

        checklink = os.path.join(cachedir, b'checklink')

        # try fast path, read only

        if os.path.islink(checklink):

            return True

        if os.path.isdir(cachedir):

            checkdir = cachedir

        else:

            checkdir = path

            cachedir = None

        name = tempfile.mktemp(

            dir=pycompat.fsdecode(checkdir), prefix=r'checklink-'

        )

        name = pycompat.fsencode(name)

        try:

            fd = None

            if cachedir is None:

                fd = pycompat.namedtempfile(

                    dir=checkdir, prefix=b'hg-checklink-'

                )

                target = os.path.basename(fd.name)

            else:

                # create a fixed file to link to; doesn't matter if it

                # already exists.

                target = b'checklink-target'

                try:

                    fullpath = os.path.join(cachedir, target)

                    open(fullpath, 'w').close()

                except PermissionError:

                    # If we can't write to cachedir, just pretend

                    # that the fs is readonly and by association

                    # that the fs won't support symlinks. This

                    # seems like the least dangerous way to avoid

                    # data loss.

                    return False

            try:

                os.symlink(target, name)

                if cachedir is None:

                    unlink(name)

                else:

                    try:

                        os.rename(name, checklink)

                    except OSError:

                        unlink(name)

                return True

            except FileExistsError:

                # link creation might race, try again

                continue

            finally:

                if fd is not None:

                    fd.close()

        except AttributeError:

            return False

        except OSError as inst:

            # sshfs might report failure while successfully creating the link

            if inst.errno == errno.EIO and os.path.exists(name):

                unlink(name)

            return False

def checkosfilename(path: bytes) -> Optional[bytes]:

    """Check that the base-relative path is a valid filename on this platform.

    Returns None if the path is ok, or a UI string describing the problem."""

    return None  # on posix platforms, every path is ok

def getfsmountpoint(path: bytes) -> Optional[bytes]:

    """Get the filesystem mount point from a directory (best-effort)

    Returns None if we are unsure. Raises OSError on ENOENT, EPERM, etc.

    """

    return getattr(osutil, 'getfsmountpoint', lambda x: None)(path)

def getfstype(path: bytes) -> Optional[bytes]:

    """Get the filesystem type name from a directory (best-effort)

    Returns None if we are unsure. Raises OSError on ENOENT, EPERM, etc.

    """

    return getattr(osutil, 'getfstype', lambda x: None)(path)

def get_password() -> bytes:

    return encoding.strtolocal(getpass.getpass(''))

def setbinary(fd) -> None:

    pass

def pconvert(path: bytes) -> bytes:

    return path

def localpath(path: bytes) -> bytes:

    return path

def samefile(fpath1: bytes, fpath2: bytes) -> bool:

    """Returns whether path1 and path2 refer to the same file. This is only

    guaranteed to work for files, not directories."""

    return os.path.samefile(fpath1, fpath2)

def samedevice(fpath1: bytes, fpath2: bytes) -> bool:

    """Returns whether fpath1 and fpath2 are on the same device. This is only

    guaranteed to work for files, not directories."""

    st1 = os.lstat(fpath1)

    st2 = os.lstat(fpath2)

    return st1.st_dev == st2.st_dev

# os.path.normcase is a no-op, which doesn't help us on non-native filesystems

def normcase(path: bytes) -> bytes:

    return path.lower()

# what normcase does to ASCII strings

normcasespec: int = encoding.normcasespecs.lower

# fallback normcase function for non-ASCII strings

normcasefallback = normcase

if pycompat.isdarwin:

    def normcase(path: bytes) -> bytes:

        """

        Normalize a filename for OS X-compatible comparison:

        - escape-encode invalid characters

        - decompose to NFD

        - lowercase

        - omit ignored characters [200c-200f, 202a-202e, 206a-206f,feff]

        >>> normcase(b'UPPER')

        'upper'

        >>> normcase(b'Caf\\xc3\\xa9')

        'cafe\\xcc\\x81'

        >>> normcase(b'\\xc3\\x89')

        'e\\xcc\\x81'

        >>> normcase(b'\\xb8\\xca\\xc3\\xca\\xbe\\xc8.JPG') # issue3918

        '%b8%ca%c3\\xca\\xbe%c8.jpg'

        """

        try:

            return encoding.asciilower(path)  # exception for non-ASCII

        except UnicodeDecodeError:

            return normcasefallback(path)

    normcasespec = encoding.normcasespecs.lower

    def normcasefallback(path: bytes) -> bytes:

        try:

            u = path.decode('utf-8')

        except UnicodeDecodeError:

            # OS X percent-encodes any bytes that aren't valid utf-8

            s = b''

            pos = 0

            l = len(path)

            while pos < l:

                try: