.local-bin/aactivator.py

#!/usr/bin/env python3
# -*- coding: UTF-8 -*-
"""\
Usage: eval "$(aactivator init)"

aactivator is a script for automatically sourcing environments in an interactive shell.
The interface for using this is two files:

    - .activate.sh: when sourced, this file activates your environment
    - .deactivate.sh: when sourced, this file deactivates your environment

A typical setup in a python project:

    $ ln -vs venv/bin/activate .activate.sh
    $ echo deactivate > .deactivate.sh

If an environment is already active it will not be re-activated.
If a different project is activated, the previous project will be deactivated beforehand.

aactivator will ask before automatically sourcing environments, and optionally
remember your answer. You can later adjust your per-project preferences in the
~/.cache/aactivator/ directory.

see also: https://github.com/Yelp/aactivator
"""
from __future__ import absolute_import, print_function, unicode_literals

import io
import os.path
import stat
import sys
from os.path import relpath
from shlex import quote

ENVIRONMENT_VARIABLE = "AACTIVATOR_ACTIVE"
ACTIVATE = ".activate.sh"
DEACTIVATE = ".deactivate.sh"

__version__ = "2.0.0"


def init(arg0):
    arg0 = os.path.realpath(arg0)
    cmd = 'if [ -x {exe} ]; then  eval "`{exe}`"; fi'.format(exe=arg0)
    return """\
export AACTIVATOR_VERSION={version}
alias aactivator={arg0}
unset {varname}
if [ "$ZSH_VERSION" ]; then
    precmd_aactivator() {{ {cmd}; }}
    if ! [ "${{precmd_functions[(r)precmd_aactivator]}}" ]; then
        precmd_functions=(precmd_aactivator $precmd_functions)
    fi
else
    if ! ( echo "$PROMPT_COMMAND" | grep -Fq '{cmd}' ); then
        PROMPT_COMMAND='{cmd}; '"$PROMPT_COMMAND"
    fi
fi""".format(
        version=__version__, arg0=arg0, cmd=cmd, varname=ENVIRONMENT_VARIABLE
    )


def get_filesystem_id(path):
    try:
        return os.stat(path).st_dev
    except OSError as error:
        if error.errno == 2:  # no such file
            return None
        else:
            raise


def insecure_inode(path):
    """This particular inode can be altered by someone other than the owner"""
    pathstat = os.stat(path).st_mode
    # Directories with a sticky bit are always acceptable.
    if os.path.isdir(path) and pathstat & stat.S_ISVTX:
        return False
    # The path is writable by someone who is not us.
    elif pathstat & (stat.S_IWGRP | stat.S_IWOTH):
        return True
    else:
        return False


def first(iterable, predicate):
    for x in iterable:
        if predicate(x):
            return x


def insecure(path):
    """Find an insecure path, at or above this one"""
    return first(search_parent_paths(path), insecure_inode)


def search_parent_paths(path):
    path = os.path.abspath(path)
    original_fs_id = fs_id = get_filesystem_id(path)
    previous_path = None
    while original_fs_id == fs_id and path != previous_path:
        yield path
        previous_path = path
        path = os.path.dirname(path)
        fs_id = get_filesystem_id(path)


def error_command(message):
    return "echo %s >&2" % quote("aactivator: " + message)


def mkdirp(path):
    try:
        os.makedirs(path)
    except OSError:
        if os.path.isdir(path):
            return
        else:
            raise


def _get_lines_if_there(path):
    if os.path.exists(path):
        return io.open(path).read().splitlines()
    else:
        return []


class ConfigFile(object):
    def __init__(self, directory, name):
        self.path = os.path.join(directory, name)
        self.lines = frozenset(_get_lines_if_there(self.path))

    def write(self, mode, value):
        mkdirp(os.path.dirname(self.path))
        with io.open(self.path, mode) as file_obj:
            file_obj.write(value)

    def append(self, value):
        self.write("a", value + "\n")


def path_is_under(path, under):
    relpath = os.path.relpath(path, under).split("/")
    return not relpath[:1] == [".."]


def user_cache_dir(env):
    # stolen from pip.utils.appdirs.user_cache_dir
    # expanduser doesn't take an env argument -.-
    from os.path import expanduser

    orig, os.environ = os.environ, env
    try:
        return expanduser(env.get("XDG_CACHE_HOME", "~/.cache"))
    finally:
        os.environ = orig


class ActivateConfig(object):
    def __init__(self, env, get_input):
        self.env = env
        self.get_input = get_input
        self.path = os.path.join(user_cache_dir(self.env), "aactivator")
        self.allowed = ConfigFile(self.path, "allowed")
        self.not_now = ConfigFile(self.path, "not-now")
        self.disallowed = ConfigFile(self.path, "disallowed")

    def refresh_not_now(self, pwd):
        result = []
        for path in self.not_now.lines:
            dirname = os.path.dirname(path)
            if path_is_under(pwd, dirname):
                result.append(path)
        self.not_now.write("w", "\n".join(result))

    def _prompt_user(self, path):
        print(
            "aactivator will source {0} and {1} at {2}.".format(
                ACTIVATE,
                DEACTIVATE,
                path,
            ),
            file=sys.stderr,
        )
        while True:
            print("Acceptable? (y)es (n)o (N)ever: ", file=sys.stderr, end="")
            sys.stderr.flush()
            try:
                response = self.get_input()
            # Allow ^D to be "no"
            except EOFError:
                response = "n"

            if response.startswith("N"):
                self.disallowed.append(path)
                print(
                    "aactivator will remember this: " "~/.cache/aactivator/disallowed",
                    file=sys.stderr,
                )
                return False

            response = response.lower()
            if response.startswith("n"):
                self.not_now.append(path)
                return False
            elif response.startswith("y"):
                self.allowed.append(path)
                print(
                    "aactivator will remember this: " "~/.cache/aactivator/allowed",
                    file=sys.stderr,
                )
                return True
            else:
                print("I didn't understand your response.", file=sys.stderr)
                print(file=sys.stderr)

    def find_allowed(self, path):
        self.refresh_not_now(path)
        return first(search_parent_paths(path), self.is_allowed)

    def is_allowed(self, path, _getuid=os.getuid):
        activate = os.path.join(path, ACTIVATE)
        if not os.path.exists(activate):
            return False
        elif os.stat(activate).st_uid != _getuid():
            # If we do not own this path, short circuit on activating
            return False
        elif path in self.disallowed.lines or path in self.not_now.lines:
            return False
        elif path in self.allowed.lines:
            return True
        else:
            return self._prompt_user(path)


def security_check(path):
    if not os.path.exists(path):
        return "aactivator: File does not exist: " + path
    insecure_path = insecure(path)
    if insecure_path is not None:
        return "aactivator: Cowardly refusing to source {0} because writeable by others: {1}".format(
            relpath(path), relpath(insecure_path)
        )


def command_for_path(cmd, path, pwd):
    if path == pwd:
        return cmd
    else:
        return " &&\n".join(
            (
                'OLDPWD_bak="$OLDPWD"',
                "cd " + quote(path),
                cmd,
                'cd "$OLDPWD_bak"',
                "cd " + quote(pwd),
                "unset OLDPWD_bak",
            )
        )


def aactivate(path, pwd):
    return command_for_path(
        " &&\n".join(
            (
                "aactivator security-check " + ACTIVATE,
                "source ./" + ACTIVATE,
                "export %s=%s" % (ENVIRONMENT_VARIABLE, quote(path)),
            )
        ),
        path,
        pwd,
    )


def deaactivate(path, pwd):
    unset = "unset " + ENVIRONMENT_VARIABLE
    deactivate_path = os.path.join(path, DEACTIVATE)

    if os.path.exists(deactivate_path):
        return command_for_path(
            " &&\n".join(
                (
                    "aactivator security-check " + DEACTIVATE,
                    "source ./" + DEACTIVATE,
                )
            )
            + "\n"
            + unset,
            path,
            pwd,
        )
    else:
        return " &&\n".join(
            (
                unset,
                error_command(
                    "Cannot deactivate. File missing: {0}".format(deactivate_path)
                ),
            )
        )


def get_output(
    environ, pwd=".", get_input=sys.stdin.readline, arg0="/path/to/aactivator"
):
    try:
        pwd = os.path.realpath(pwd)
    except OSError as error:
        if error.errno == 2:  # no such file
            return ""
        else:
            raise
    config = ActivateConfig(environ, get_input)
    activate_path = config.find_allowed(pwd)
    result = []

    if environ.get("AACTIVATOR_VERSION") == __version__:
        activated_env = environ.get(ENVIRONMENT_VARIABLE)
    else:
        result.append(init(arg0))
        activated_env = None

    if (
        activated_env != activate_path
    ):  # did we already activate the current environment?
        if activated_env:  # deactivate it
            result.append(deaactivate(activated_env, pwd))
        if activate_path:
            result.append(aactivate(activate_path, pwd))
    return " &&\n".join(result)


def aactivator(args, env):
    if len(args) == 1:
        return get_output(env, arg0=args[0])
    elif len(args) == 2 and args[1] == "init":
        return init(args[0])
    elif len(args) == 3 and args[1] == "security-check":
        exit(security_check(args[2]))
    else:
        return __doc__ + "\nVersion: " + __version__


def main():
    try:
        print(aactivator(tuple(sys.argv), os.environ.copy()))
    except KeyboardInterrupt:  # pragma: no cover
        # Silence ^C
        pass


if __name__ == "__main__":
    try:
        sys.exit(main())
    except KeyboardInterrupt:
        # Silence ^C
        pass