Source code for enstaller.config

# Copyright by Enthought, Inc.
# Author: Ilan Schnell <>
from __future__ import absolute_import, print_function

import logging
import re
import os
import sys
import platform
import tempfile
import warnings

from os.path import isfile, join

from egginst.utils import parse_assignments
from egginst.vendor.six import string_types
from egginst.vendor.six.moves import urllib

from enstaller.vendor import keyring
from enstaller.vendor.keyring.backends.file import PlaintextKeyring

from enstaller import __version__
from enstaller.auth import (DUMMY_USER,
                            subscription_message, APITokenAuth, UserInfo,
from enstaller.config_templates import RC_DEFAULT_TEMPLATE, RC_TEMPLATE
from enstaller.errors import (EnstallerException, InvalidConfiguration,
from enstaller.proxy_info import ProxyInfo
from enstaller.repository_info import (BroodRepositoryInfo,
from enstaller.utils import real_prefix, under_venv
from enstaller.vendor import requests
from enstaller.cli.utils import humanize_ssl_error_and_die
from enstaller import plat
from .utils import PY_VER, abs_expanduser, fill_url
from ._yaml_config import load_configuration_from_yaml

logger = logging.getLogger(__name__)


ENSTALLER4RC_FILENAME = ".enstaller4rc"
HOME_ENSTALLER4RC = os.path.join(abs_expanduser("~"), ENSTALLER4RC_FILENAME)

_BROOD_PREFIX = "brood+"

def _setup_keyring():
    backend = PlaintextKeyring()

        if sys.platform == "win32":
            from enstaller.vendor.keyring.backends.Windows import \
            tentative_backend = WinVaultKeyring()
        elif sys.platform == "darwin":
            from enstaller.vendor.keyring.backends.OS_X import Keyring
            tentative_backend = Keyring()
            tentative_backend = backend
        if tentative_backend.priority >= 0:
            backend = tentative_backend
    except ImportError:



def _keyring_backend_name():
    return str(type(keyring.get_keyring()))

def legacy_configuration_read_search_order():
    Return a list of directories where to look for the configuration file.

    Legacy order, don't use outside canopy.
    paths = [

    return [os.path.normpath(p) for p in paths]

def configuration_read_search_order():
    Return a list of directories where to look for the configuration file.
    paths = [sys.prefix]
    if under_venv():

    return [os.path.normpath(p) for p in paths]

def add_url(filename, config, url):
    url = fill_url(url)
    if url in config.indexed_repositories:
        print("Already configured:", url)
    prepend_url(filename, url)

def _get_writable_local_dir(local_dir):
    if not os.access(local_dir, os.F_OK):
            return local_dir
        except (OSError, IOError):
    elif os.access(local_dir, os.W_OK):
        return local_dir

    logger.warn('Warning: the following directory is not writeable '
                'with current permissions:\n'
                '    {0!r}\n'
                'Using a temporary cache for index and eggs.\n'.
    return tempfile.mkdtemp()

[docs]def write_default_config(filename): """ Write a default configuration file at the given location. Will raise an exception if a file already exists. Parameters ---------- filename : str The location to write to. """ if os.path.isfile(filename): msg = "File '{0}' already exists, not overwriting." raise EnstallerException(msg.format(filename)) else: config = Configuration() if config.proxy: proxy_line = 'proxy = %r' % str(config.proxy) else: proxy_line = ('#proxy = <proxy string> ' '# e.g. "http://<user>:<passwd>@"') variables = {"py_ver": PY_VER, "sys_prefix": sys.prefix, "version": __version__, "proxy_line": proxy_line, "auth_section": ""} with open(filename, "w") as fo: fo.write(RC_DEFAULT_TEMPLATE % variables)
def _is_using_epd_username(filename_or_fp): """ Returns True if the given configuration file uses EPD_username. """ data = parse_assignments(filename_or_fp) return "EPD_username" in data and "EPD_auth" not in data def convert_auth_if_required(filename): """ This function will convert configuration using EPD_username + keyring to using EPD_auth. Returns True if the file has been modified, False otherwise. """ did_convert = False if _is_using_epd_username(filename): msg = "Cannot convert password: no password found in keyring" config = Configuration.from_file(filename) if config.auth is None: raise EnstallerException(msg) username = config.auth.username password = _get_keyring_password(username) if password is None: raise EnstallerException(msg) else: config.update(auth=(username, password)) config._change_auth(filename) did_convert = True return did_convert def _get_keyring_password(username): return keyring.get_password(KEYRING_SERVICE_NAME, username) def _set_keyring_password(username, password): return keyring.set_password(KEYRING_SERVICE_NAME, username, password) def _create_error_message(fp, exc): pos = fp.tell() try: lines = fp.readlines() line = lines[exc.lineno-1].rstrip() if isinstance(exc, SyntaxError): msg = "Could not parse configuration file " \ "(invalid python syntax at line {0!r}: expression {1!r})".\ format(exc.lineno, line) else: msg = "Could not parse configuration file " \ "(error at line {0!r}: expression {1!r} not " \ "supported)".format(exc.lineno, line) return msg finally:
[docs]class Configuration(object): @classmethod def _from_legacy_locations(cls): warnings.warn("Using legacy location is deprecated, use " "Configuration.from_filename with an explicit " "filename instead", DeprecationWarning) config_path = None for p in legacy_configuration_read_search_order(): candidate = os.path.join(p, ENSTALLER4RC_FILENAME) if isfile(candidate): config_path = candidate break if config_path is None: raise InvalidConfiguration("No default configuration found.") else: return cls.from_file(config_path) @classmethod def from_yaml_filename(cls, filename): return load_configuration_from_yaml(cls, filename) @classmethod
[docs] def from_file(cls, filename): """ Create a new Configuration instance from the given file. Parameters ---------- filename: str or file-like object If a string, is understood as a filename to open. Understood as a file-like object otherwise. """ def _create(fp): ret = cls() def api_token_to_auth(api_token): ret.update(auth=APITokenAuth(api_token)) def epd_auth_to_auth(epd_auth): ret.update(auth=UserPasswordAuth.from_encoded_auth(epd_auth)) def epd_username_to_auth(username): if keyring is not None: password = _get_keyring_password(username) if password is not None: ret.update(auth=(username, password)) try: parsed = parse_assignments(fp) except (InvalidFormat, SyntaxError) as e: msg = _create_error_message(fp, e) raise InvalidConfiguration(msg) # We need a custom translator to manage attributes in the # configuration file without polluting the config class translator = ret._name_to_setter.copy() translator.update({ "EPD_auth": epd_auth_to_auth, "EPD_username": epd_username_to_auth, "IndexedRepos": translator["indexed_repositories"], "api_token": api_token_to_auth, }) if "EPD_auth" in parsed and "api_token" in parsed: msg = "Both 'EPD_auth' and 'api_token' set in configuration." \ "\nYou should remove one of those for consistent " \ "behaviour." warnings.warn(msg) for name, value in parsed.items(): if name in translator: translator[name](value) else: warnings.warn("Unsupported configuration setting {0}, " "ignored".format(name)) return ret if isinstance(filename, string_types): with open(filename, "r") as fp: ret = _create(fp) ret._filename = filename return ret else: return _create(filename)
def __init__(self, **kw): """ Create a new configuration instance. Any argument passed to the constructor need to be a keyword argument, and is understood as an argument to the update method, e.g.:: config = Configuration(store_url="") # equivalent to config = Configuration() config.update(store_url="") """ self._auth = None self._autoupdate = True self._noapp = False self._proxy = None self._use_pypi = True self._use_webservice = True self._prefix = os.path.normpath(sys.prefix) self._indexed_repositories = [] self._repositories = [] self._store_url = "" self._store_kind = STORE_KIND_LEGACY self._repository_cache = None self._filename = None self._platform = plat.custom_plat self._max_retries = 0 self._verify_ssl = True self._name_to_setter = {} simple_attributes = [ ("autoupdate", "_autoupdate"), ("noapp", "_noapp"), ("verify_ssl", "_verify_ssl"), ("use_pypi", "_use_pypi"), ("use_webservice", "_use_webservice"), ] for name, private_attribute in simple_attributes: self._name_to_setter[name] = \ self._simple_attribute_set_factory(private_attribute) self._name_to_setter.update({ "auth": self._set_auth, "indexed_repositories": self._set_indexed_repositories, "max_retries": self._set_max_retries, "prefix": self._set_prefix, "proxy": self._set_proxy, "repository_cache": self._set_repository_cache, "store_url": self._set_store_url, }) self.update(**kw) # ---------- # Properties # ---------- @property def api_url(self): """ Url to hit to get user information on """ return fill_url("{0}/accounts/user/info/".format(self.store_url)) @property def auth(self): """ The auth object that may be passed to Session.authenticate :return: the auth instance, or None is configured. :rtype: IAuth or None """ return self._auth @property def autoupdate(self): """ Whether enpkg should attempt updating itself. """ return self._autoupdate @property def filename(self): """ The filename this configuration was created from. May be None if the configuration was not created from a file. """ return self._filename @property def indexed_repositories(self): """ List of (old-style) repositories. Only actually used when use_webservice is False. """ return self._indexed_repositories @property def indices(self): """ Returns a list of pair (index_url, store_location) for this given configuration. Takes into account webservice/no webservice and pypi True/False .. deprecated:: 4.9.0 Use the repositories property instead """ msg = ("the indices property is deprecated: use the repositories " "property instead") warnings.warn(msg, DeprecationWarning) return tuple((repository_info.index_url, repository_info._base_url) for repository_info in self.repositories) @property def max_retries(self): """ Max attempts to retry an http connection or re-fetching data whose checksum failed. """ return self._max_retries @property def noapp(self): """ Ignore appinst entries. """ return self._noapp @property def prefix(self): """ Prefix in which enpkg operates. """ return self._prefix @property def proxy(self): """ A ProxyInfo instance or None if no proxy is configured. """ return self._proxy @property def proxy_dict(self): """ A dictionary <scheme>:<proxy_string> that can be used as the proxies argument for requests. """ if self._proxy: return {self._proxy.scheme: str(self._proxy)} else: return {} @property def repositories(self): if self.use_webservice: return (CanopyRepositoryInfo(self.store_url, self.use_pypi, self._platform),) else: return self._repositories @property def repository_cache(self): """ Absolute path where eggs will be cached. """ if self._repository_cache is None: return join(self.prefix, "LOCAL-REPO") else: return self._repository_cache @property def store_kind(self): """ Store kind (brood, legacy canopy, old-repo style). """ return self._store_kind @property def store_url(self): """ The store url to hit for indices and eggs. """ return self._store_url @property def verify_ssl(self): """ Whether to verify SSL CA or not. """ return self._verify_ssl @property def use_pypi(self): """ Whether to load pypi repositories (in `webservice` mode). """ return self._use_pypi @property def use_webservice(self): """ Whether to use canopy legacy or not. """ return self._use_webservice @property def webservice_entry_point(self): """ Whether to fetch indices and data (in `webservice` mode). """ return fill_url("{0}/eggs/{1}/". format(self.store_url, self._platform)) # -------------- # Public methods # -------------- def reset_auth(self): self._auth = None def set_auth_from_encoded(self, value): try: auth = UserPasswordAuth.from_encoded_auth(value) except Exception: raise InvalidConfiguration("Invalid EPD_auth value") else: self.update(auth=auth)
[docs] def set_repositories_from_names(self, names): """ Set repositories from their names alone, e.g. 'enthought/free'. Parameters ---------- names : list List of repository names """ repositories = [] for name in names: p = urllib.parse.urlparse(name) if p.scheme == "": repositories.append(BroodRepositoryInfo(self.store_url, name, self._platform)) elif p.scheme == "file": repositories.append(FSRepositoryInfo(name)) else: msg = "Unsupported syntax: {0!r}".format(name) raise InvalidConfiguration(msg) self._repositories = tuple(repositories)
[docs] def update(self, **kw): """ Set configuration attributes given as keyword arguments.""" for name, value in kw.items(): setter = self._name_to_setter.get(name, None) if name is None: raise ValueError("Invalid setting name: {0!r}".format(name)) else: setter(value)
[docs] def write(self, filename): """ Write this configuration to the given filename. Parameters ---------- filename : str The path of the written file. """ if self.auth is None: auth_section = "" else: auth_section = self.auth.config_string if self.proxy: proxy_line = 'proxy = %r' % str(self.proxy) else: proxy_line = ('#proxy = <proxy string> ' '# e.g. "http://<user>:<passwd>@"') # XXX: because setting indexed repositories use string replacement # (sigh...), we need to write the IndexedRepos section in a very specific # way: # # IndexedRepos = [ # url1, # url2 # ] indexed_repos_string = "\n".join(" %r," % (repo,) for repo in self.indexed_repositories) if self.store_kind == STORE_KIND_BROOD: store_url = _BROOD_PREFIX + self.store_url else: store_url = self.store_url variables = {"py_ver": PY_VER, "sys_prefix": sys.prefix, "version": __version__, "proxy_line": proxy_line, "auth_section": auth_section, "store_url": store_url, "indexed_repositories": indexed_repos_string, "use_pypi": self.use_pypi, "use_webservice": self.use_webservice, } with open(filename, "w") as fo: fo.write(RC_TEMPLATE % variables) # --------------- # Private methods # ---------------
def _change_auth(self, filename): if self.auth is None: pat = re.compile(r'^(EPD_auth|EPD_username)\s*=.*$', re.M) with open(filename, 'r') as fi: data = if data = pat.sub("", data) with open(filename, 'w') as fo: fo.write(data) else: self.auth.change_auth(filename) def _checked_change_auth(self, auth, session, filename): user = {} session.authenticate(auth) self.update(auth=auth) self._change_auth(filename) user = UserInfo.from_session(session) print(subscription_message(self, user)) return user def _set_auth(self, auth): """ Set the internal authentication information. Parameters ---------- auth : Auth-like The authentication information. May be a (username, password) tuple, or an *Auth subclass. """ if isinstance(auth, tuple) and len(auth) == 2: username, password = auth if username is None: raise InvalidConfiguration( "invalid authentication arguments: " "{0}:{1}".format(username, password)) self._auth = UserPasswordAuth(username, password) else: self._auth = auth def _set_indexed_repositories(self, urls): self._indexed_repositories = tuple(fill_url(url) for url in urls) self._repositories = tuple(OldstyleRepositoryInfo(url) for url in self._indexed_repositories) def _set_max_retries(self, raw_max_retries): try: max_retries = int(raw_max_retries) except ValueError: msg = "Invalid type for 'max_retries': {0!r}" raise InvalidConfiguration(msg.format(raw_max_retries)) else: self._max_retries = max_retries def _set_prefix(self, prefix): self._prefix = os.path.normpath(abs_expanduser(prefix)) def _set_proxy(self, proxy_string): self._proxy = ProxyInfo.from_string(proxy_string) def _set_store_url(self, url): p = urllib.parse.urlparse(url) if p.scheme.startswith(_BROOD_PREFIX): url = url[len(_BROOD_PREFIX):] self._store_kind = STORE_KIND_BROOD self._store_url = url def _set_repository_cache(self, value): normalized = os.path.normpath(abs_expanduser(value)) self._repository_cache = _get_writable_local_dir(normalized) def _simple_attribute_set_factory(self, attribute_name): return lambda value: setattr(self, attribute_name, value)
def prepend_url(filename, url): with open(filename, 'r+') as fp: data = pat = re.compile(r'^IndexedRepos\s*=\s*\[\s*$', re.M) if not sys.exit("Error: IndexedRepos section not found") data = pat.sub(r"IndexedRepos = [\n '%s'," % url, data) fp.write(data) def print_config(config, session): print("Python version:", PY_VER) print("enstaller version:", __version__) print("sys.prefix:", os.path.normpath(sys.prefix)) print("platform:", platform.platform()) print("architecture:", platform.architecture()[0]) print("use_webservice:", config.use_webservice) if config.filename is not None: print("config file:", config.filename) print("keyring backend: %s" % (_keyring_backend_name(), )) print("settings:") print(" prefix = %s" % config.prefix) print(" %s = %s" % ("repository_cache", config.repository_cache)) print(" %s = %r" % ("noapp", config.noapp)) print(" %s = %r" % ("proxy", config.proxy)) if not config.use_webservice: print(" IndexedRepos:") for repo in config.indexed_repositories: print(' %r' % repo) user = DUMMY_USER if config.auth is None: print("No valid auth information in configuration, cannot " "authenticate.") else: try: session.authenticate(config.auth) user = UserInfo.from_session(session) except requests.exceptions.SSLError as e: humanize_ssl_error_and_die(e, config.store_url) except Exception as e: print(e) print(subscription_message(config, user))