<commit_before>"""
Collaborate with RestApiClient to make remote anonymous and authenticated calls.
Uses user_io to request user's login and password and obtain a token for calling authenticated
methods if receives AuthenticationException from RestApiClient.


Flow:

    Directly invoke a REST method in RestApiClient, example: get_conan.
    if receives AuthenticationException (not open method) will ask user for login and password
    and will invoke RestApiClient.get_token() (with LOGIN_RETRIES retries) and retry to call
    get_conan with the new token.
"""

from conans.errors import AuthenticationException, ForbiddenException,\
    ConanException
from uuid import getnode as get_mac
import hashlib
from conans.util.log import logger
from os import getenv


def input_credentials_if_unauthorized(func):
    """Decorator. Handles AuthenticationException and request user
    to input a user and a password"""
    LOGIN_RETRIES = 3

    def wrapper(self, *args, **kwargs):
        try:
            # Set custom headers of mac_digest and username
            self.set_custom_headers(self.user)
            ret = func(self, *args, **kwargs)
            return ret
        except ForbiddenException:
            raise ForbiddenException("Permission denied for user: '%s'" % self.user)
        except AuthenticationException:
            # User valid but not enough permissions
            if self.user is None or self._rest_client.token is None:
                # token is None when you change user with user command
                # Anonymous is not enough, ask for a user
                remote = self.remote
                self._user_io.out.info('Please log in to "%s" to perform this action. '
                                       'Execute "conan user" command.' % remote.name)
                if "bintray" in remote.url:
                    self._user_io.out.info('If you don\'t have an account sign up here: '
                                           'https://bintray.com/signup/oss')
                disable_login_prompt = getenv("CONAN_DISABLE_LOGIN_PROMPT", "False")
                if disable_login_prompt == "True":
                    raise AuthenticationException('User "%s" not authenticated' % self.user)
                return retry_with_new_token(self, *args, **kwargs)
            else:
                # Token expired or not valid, so clean the token and repeat the call
                # (will be anonymous call but exporting who is calling)
                logger.info("Token expired or not valid, cleaning the saved token and retrying")
                self._store_login((self.user, None))
                self._rest_client.token = None
                # Set custom headers of mac_digest and username
                self.set_custom_headers(self.user)
                return wrapper(self, *args, **kwargs)

    def retry_with_new_token(self, *args, **kwargs):
        """Try LOGIN_RETRIES to obtain a password from user input for which
        we can get a valid token from api_client. If a token is returned,
        credentials are stored in localdb and rest method is called"""
        for _ in range(LOGIN_RETRIES):
            user, password = self._user_io.request_login(self._remote.name, self.user)
            token = None
            try:
                token = self.authenticate(user, password)
            except AuthenticationException:
                if self.user is None:<commit_msg>Fix disable login prompt environment variable. (#2674)<commit_after>"""
Collaborate with RestApiClient to make remote anonymous and authenticated calls.
Uses user_io to request user's login and password and obtain a token for calling authenticated
methods if receives AuthenticationException from RestApiClient.


Flow:

    Directly invoke a REST method in RestApiClient, example: get_conan.
    if receives AuthenticationException (not open method) will ask user for login and password
    and will invoke RestApiClient.get_token() (with LOGIN_RETRIES retries) and retry to call
    get_conan with the new token.
"""

from conans.errors import AuthenticationException, ForbiddenException,\
    ConanException
from uuid import getnode as get_mac
import hashlib
from conans.util.log import logger
from conans.util.env_reader import get_env


def input_credentials_if_unauthorized(func):
    """Decorator. Handles AuthenticationException and request user
    to input a user and a password"""
    LOGIN_RETRIES = 3

    def wrapper(self, *args, **kwargs):
        try:
            # Set custom headers of mac_digest and username
            self.set_custom_headers(self.user)
            ret = func(self, *args, **kwargs)
            return ret
        except ForbiddenException:
            raise ForbiddenException("Permission denied for user: '%s'" % self.user)
        except AuthenticationException:
            # User valid but not enough permissions
            if self.user is None or self._rest_client.token is None:
                # token is None when you change user with user command
                # Anonymous is not enough, ask for a user
                remote = self.remote
                self._user_io.out.info('Please log in to "%s" to perform this action. '
                                       'Execute "conan user" command.' % remote.name)
                if "bintray" in remote.url:
                    self._user_io.out.info('If you don\'t have an account sign up here: '
                                           'https://bintray.com/signup/oss')
                if get_env("CONAN_DISABLE_LOGIN_PROMPT", False):
                    raise AuthenticationException('User "%s" not authenticated' % self.user)
                return retry_with_new_token(self, *args, **kwargs)
            else:
                # Token expired or not valid, so clean the token and repeat the call
                # (will be anonymous call but exporting who is calling)
                logger.info("Token expired or not valid, cleaning the saved token and retrying")
                self._store_login((self.user, None))
                self._rest_client.token = None
                # Set custom headers of mac_digest and username
                self.set_custom_headers(self.user)
                return wrapper(self, *args, **kwargs)

    def retry_with_new_token(self, *args, **kwargs):
        """Try LOGIN_RETRIES to obtain a password from user input for which
        we can get a valid token from api_client. If a token is returned,
        credentials are stored in localdb and rest method is called"""
        for _ in range(LOGIN_RETRIES):
            user, password = self._user_io.request_login(self._remote.name, self.user)
            token = None
            try:
                token = self.authenticate(user, password)
            except AuthenticationException:
                if self.user is None:
{'fraction_non_alphanumeric': 0.06606387633075689, 'fraction_numerical': 0.0008750182295464489, 'mean_word_length': 2.2703862660944205, 'pattern_counts': {'":': 1, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5544383', 'n_tokens_mistral': 1762, 'n_tokens_neox': 1690, 'n_words': 652}