BaseCLIApp._get_config_file_path() - Code Metrics - Inspection of "Merge pull request #3612 from StackStorm/more_v2.3..." - Plexxi/st2 - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — plexxi-v2.3.2 ( 5d46fe )

unknown

created 2017-08-22 17:03 UTC

BaseCLIApp._get_config_file_path() A

↳ Parent: BaseCLIApp

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	4
c	0
b	0
f	0
dl	0
loc	16
rs	9.2

# Licensed to the StackStorm, Inc ('StackStorm') under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import os
import json
import logging
import time
import calendar
import traceback

import six
import requests

from st2client import models
from st2client.config_parser import CLIConfigParser
from st2client.config_parser import ST2_CONFIG_DIRECTORY
from st2client.config_parser import ST2_CONFIG_PATH
from st2client.client import Client
from st2client.config import get_config
from st2client.utils.date import parse as parse_isotime
from st2client.utils.misc import merge_dicts

__all__ = [
    'BaseCLIApp'
]

# How many seconds before the token actual expiration date we should consider the token as
# expired. This is used to prevent the operation from failing durig the API request because the
# token was just about to expire.
TOKEN_EXPIRATION_GRACE_PERIOD_SECONDS = 15

CONFIG_OPTION_TO_CLIENT_KWARGS_MAP = {
    'base_url': ['general', 'base_url'],
    'auth_url': ['auth', 'url'],
    'stream_url': ['stream', 'url'],
    'api_url': ['api', 'url'],
    'api_version': ['general', 'api_version'],
    'api_key': ['credentials', 'api_key'],
    'cacert': ['general', 'cacert'],
    'debug': ['cli', 'debug']
}


class BaseCLIApp(object):
    """
    Base class for StackStorm CLI apps.
    """

    LOG = logging.getLogger(__name__)  # logger instance to use
    client = None  # st2client instance

    # A list of command classes for which automatic authentication should be skipped.
    SKIP_AUTH_CLASSES = []

    def get_client(self, args, debug=False):
        ST2_CLI_SKIP_CONFIG = os.environ.get('ST2_CLI_SKIP_CONFIG', 0)
        ST2_CLI_SKIP_CONFIG = int(ST2_CLI_SKIP_CONFIG)

        skip_config = args.skip_config
        skip_config = skip_config or ST2_CLI_SKIP_CONFIG

        # Note: Options provided as the CLI argument have the highest precedence
        # Precedence order: cli arguments > environment variables > rc file variables
        cli_options = ['base_url', 'auth_url', 'api_url', 'stream_url', 'api_version', 'cacert']
        cli_options = {opt: getattr(args, opt, None) for opt in cli_options}
        config_file_options = self._get_config_file_options(args=args)

        kwargs = {}

        if not skip_config:
            # Config parsing is not skipped
            kwargs = merge_dicts(kwargs, config_file_options)

        kwargs = merge_dicts(kwargs, cli_options)
        kwargs['debug'] = debug

        client = Client(**kwargs)

        if skip_config:
            # Config parsing is skipped
            self.LOG.info('Skipping parsing CLI config')
            return client

        # Ok to use config at this point
        rc_config = get_config()

        # Silence SSL warnings
        silence_ssl_warnings = rc_config.get('general', {}).get('silence_ssl_warnings', False)
        if silence_ssl_warnings:
            requests.packages.urllib3.disable_warnings()

        # We skip automatic authentication for some commands such as auth
        try:
            command_class_name = args.func.im_class.__name__
        except Exception:
            command_class_name = None

        if command_class_name in self.SKIP_AUTH_CLASSES:
            return client

        # We also skip automatic authentication if token is provided via the environment variable
        # or as a command line argument
        env_var_token = os.environ.get('ST2_AUTH_TOKEN', None)
        cli_argument_token = getattr(args, 'token', None)
        env_var_api_key = os.environ.get('ST2_API_KEY', None)
        cli_argument_api_key = getattr(args, 'api_key', None)
        if env_var_token or cli_argument_token or env_var_api_key or cli_argument_api_key:
            return client

        # If credentials are provided in the CLI config use them and try to authenticate
        credentials = rc_config.get('credentials', {})
        username = credentials.get('username', None)
        password = credentials.get('password', None)
        cache_token = rc_config.get('cli', {}).get('cache_token', False)

        if username:
            # Credentials are provided, try to authenticate agaist the API
            try:
                token = self._get_auth_token(client=client, username=username, password=password,
                                             cache_token=cache_token)
            except requests.exceptions.ConnectionError as e:
                self.LOG.warn('Auth API server is not available, skipping authentication.')
                self.LOG.exception(e)
                return client
            except Exception as e:
                print('Failed to authenticate with credentials provided in the config.')
                raise e
            client.token = token
            # TODO: Hack, refactor when splitting out the client
            os.environ['ST2_AUTH_TOKEN'] = token

        return client

    def _get_config_file_options(self, args):
        """
        Parse the config and return kwargs which can be passed to the Client
        constructor.

        :rtype: ``dict``
        """
        rc_options = self._parse_config_file(args=args)
        result = {}
        for kwarg_name, (section, option) in six.iteritems(CONFIG_OPTION_TO_CLIENT_KWARGS_MAP):
            result[kwarg_name] = rc_options.get(section, {}).get(option, None)

        return result

    def _parse_config_file(self, args):
        config_file_path = self._get_config_file_path(args=args)

        parser = CLIConfigParser(config_file_path=config_file_path, validate_config_exists=False)
        result = parser.parse()
        return result

    def _get_config_file_path(self, args):
        """
        Retrieve path to the CLI configuration file.

        :rtype: ``str``
        """
        path = os.environ.get('ST2_CONFIG_FILE', ST2_CONFIG_PATH)

        if args.config_file:
            path = args.config_file

        path = os.path.abspath(os.path.expanduser(path))
        if path != ST2_CONFIG_PATH and not os.path.isfile(path):
            raise ValueError('Config "%s" not found' % (path))

        return path

    def _get_auth_token(self, client, username, password, cache_token):
        """
        Retrieve a valid auth token.

        If caching is enabled, we will first try to retrieve cached token from a
        file system. If cached token is expired or not available, we will try to
        authenticate using the provided credentials and retrieve a new auth
        token.

        :rtype: ``str``
        """
        if cache_token:
            token = self._get_cached_auth_token(client=client, username=username,
                                                password=password)
        else:
            token = None
        if not token:
            # Token is either expired or not available
            token_obj = self._authenticate_and_retrieve_auth_token(client=client,
                                                                   username=username,
                                                                   password=password)

            self._cache_auth_token(token_obj=token_obj)
            token = token_obj.token

        return token

    def _get_cached_auth_token(self, client, username, password):
        """
        Retrieve cached auth token from the file in the config directory.

        :rtype: ``str``
        """
        if not os.path.isdir(ST2_CONFIG_DIRECTORY):
            os.makedirs(ST2_CONFIG_DIRECTORY)

        cached_token_path = self._get_cached_token_path_for_user(username=username)

        if not os.access(ST2_CONFIG_DIRECTORY, os.R_OK):
            # We don't have read access to the file with a cached token
            message = ('Unable to retrieve cached token from "%s" (user %s doesn\'t have read '
                       'access to the parent directory). Subsequent requests won\'t use a '
                       'cached token meaning they may be slower.' % (cached_token_path,
                                                                     os.getlogin()))
            self.LOG.warn(message)
            return None

        if not os.path.isfile(cached_token_path):
            return None

        if not os.access(cached_token_path, os.R_OK):
            # We don't have read access to the file with a cached token
            message = ('Unable to retrieve cached token from "%s" (user %s doesn\'t have read '
                       'access to this file). Subsequent requests won\'t use a cached token '
                       'meaning they may be slower.' % (cached_token_path, os.getlogin()))
            self.LOG.warn(message)
            return None

        # Safety check for too permissive permissions
        file_st_mode = oct(os.stat(cached_token_path).st_mode & 0777)
        others_st_mode = int(file_st_mode[-1])

        if others_st_mode >= 4:
            # Every user has access to this file which is dangerous
            message = ('Permissions (%s) for cached token file "%s" are to permissive. Please '
                       'restrict the permissions and make sure only your own user can read '
                       'from the file' % (file_st_mode, cached_token_path))
            self.LOG.warn(message)

        with open(cached_token_path) as fp:
            data = fp.read()

        try:
            data = json.loads(data)

            token = data['token']
            expire_timestamp = data['expire_timestamp']
        except Exception as e:
            msg = ('File "%s" with cached token is corrupted or invalid (%s). Please delete '
                   ' this file' % (cached_token_path, str(e)))
            raise ValueError(msg)

        now = int(time.time())
        if (expire_timestamp - TOKEN_EXPIRATION_GRACE_PERIOD_SECONDS) < now:
            self.LOG.debug('Cached token from file "%s" has expired' % (cached_token_path))
            # Token has expired
            return None

        self.LOG.debug('Using cached token from file "%s"' % (cached_token_path))
        return token

    def _cache_auth_token(self, token_obj):
        """
        Cache auth token in the config directory.

        :param token_obj: Token object.
        :type token_obj: ``object``
        """
        if not os.path.isdir(ST2_CONFIG_DIRECTORY):
            os.makedirs(ST2_CONFIG_DIRECTORY)

        username = token_obj.user
        cached_token_path = self._get_cached_token_path_for_user(username=username)

        if not os.access(ST2_CONFIG_DIRECTORY, os.W_OK):
            # We don't have write access to the file with a cached token
            message = ('Unable to write token to "%s" (user %s doesn\'t have write '
                       'access to the parent directory). Subsequent requests won\'t use a '
                       'cached token meaning they may be slower.' % (cached_token_path,
                                                                     os.getlogin()))
            self.LOG.warn(message)
            return None

        if os.path.isfile(cached_token_path) and not os.access(cached_token_path, os.W_OK):
            # We don't have write access to the file with a cached token
            message = ('Unable to write token to "%s" (user %s doesn\'t have write '
                       'access to this file). Subsequent requests won\'t use a '
                       'cached token meaning they may be slower.' % (cached_token_path,
                                                                     os.getlogin()))
            self.LOG.warn(message)
            return None

        token = token_obj.token
        expire_timestamp = parse_isotime(token_obj.expiry)
        expire_timestamp = calendar.timegm(expire_timestamp.timetuple())

        data = {}
        data['token'] = token
        data['expire_timestamp'] = expire_timestamp
        data = json.dumps(data)

        # Note: We explictly use fdopen instead of open + chmod to avoid a security issue.
        # open + chmod are two operations which means that during a short time frame (between
        # open and chmod) when file can potentially be read by other users if the default
        # permissions used during create allow that.
        fd = os.open(cached_token_path, os.O_WRONLY | os.O_CREAT, 0600)
        with os.fdopen(fd, 'w') as fp:
            fp.write(data)

        self.LOG.debug('Token has been cached in "%s"' % (cached_token_path))
        return True

    def _authenticate_and_retrieve_auth_token(self, client, username, password):
        manager = models.ResourceManager(models.Token, client.endpoints['auth'],
                                         cacert=client.cacert, debug=client.debug)
        instance = models.Token()
        instance = manager.create(instance, auth=(username, password))
        return instance

    def _get_cached_token_path_for_user(self, username):
        """
        Retrieve cached token path for the provided username.
        """
        file_name = 'token-%s' % (username)
        result = os.path.abspath(os.path.join(ST2_CONFIG_DIRECTORY, file_name))
        return result

    def _print_config(self, args):
        config = self._parse_config_file(args=args)

        for section, options in six.iteritems(config):
            print('[%s]' % (section))

            for name, value in six.iteritems(options):
                print('%s = %s' % (name, value))

    def _print_debug_info(self, args):
        # Print client settings
        self._print_client_settings(args=args)

        # Print exception traceback
        traceback.print_exc()

    def _print_client_settings(self, args):
        client = self.client

        if not client:
            return

        config_file_path = self._get_config_file_path(args=args)

        print('CLI settings:')
        print('----------------')
        print('Config file path: %s' % (config_file_path))
        print('Client settings:')
        print('----------------')
        print('ST2_BASE_URL: %s' % (client.endpoints['base']))
        print('ST2_AUTH_URL: %s' % (client.endpoints['auth']))
        print('ST2_API_URL: %s' % (client.endpoints['api']))
        print('ST2_STREAM_URL: %s' % (client.endpoints['stream']))
        print('ST2_AUTH_TOKEN: %s' % (os.environ.get('ST2_AUTH_TOKEN')))
        print('')
        print('Proxy settings:')
        print('---------------')
        print('HTTP_PROXY: %s' % (os.environ.get('HTTP_PROXY', '')))
        print('HTTPS_PROXY: %s' % (os.environ.get('HTTPS_PROXY', '')))
        print('')


1			# Licensed to the StackStorm, Inc ('StackStorm') under one or more
2			# contributor license agreements. See the NOTICE file distributed with
3			# this work for additional information regarding copyright ownership.
4			# The ASF licenses this file to You under the Apache License, Version 2.0
5			# (the "License"); you may not use this file except in compliance with
6			# the License. You may obtain a copy of the License at
7			#
8			# http://www.apache.org/licenses/LICENSE-2.0
9			#
10			# Unless required by applicable law or agreed to in writing, software
11			# distributed under the License is distributed on an "AS IS" BASIS,
12			# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13			# See the License for the specific language governing permissions and
14			# limitations under the License.
15
16			import os
17			import json
18			import logging
19			import time
20			import calendar
21			import traceback
22
23			import six
24			import requests
25
26			from st2client import models
27			from st2client.config_parser import CLIConfigParser
28			from st2client.config_parser import ST2_CONFIG_DIRECTORY
29			from st2client.config_parser import ST2_CONFIG_PATH
30			from st2client.client import Client
31			from st2client.config import get_config
32			from st2client.utils.date import parse as parse_isotime
33			from st2client.utils.misc import merge_dicts
34
35			__all__ = [
36			'BaseCLIApp'
37			]
38
39			# How many seconds before the token actual expiration date we should consider the token as
40			# expired. This is used to prevent the operation from failing durig the API request because the
41			# token was just about to expire.
42			TOKEN_EXPIRATION_GRACE_PERIOD_SECONDS = 15
43
44			CONFIG_OPTION_TO_CLIENT_KWARGS_MAP = {
45			'base_url': ['general', 'base_url'],
46			'auth_url': ['auth', 'url'],
47			'stream_url': ['stream', 'url'],
48			'api_url': ['api', 'url'],
49			'api_version': ['general', 'api_version'],
50			'api_key': ['credentials', 'api_key'],
51			'cacert': ['general', 'cacert'],
52			'debug': ['cli', 'debug']
53			}
54
55
56			class BaseCLIApp(object):
57			"""
58			Base class for StackStorm CLI apps.
59			"""
60
61			LOG = logging.getLogger(__name__) # logger instance to use
62			client = None # st2client instance
63
64			# A list of command classes for which automatic authentication should be skipped.
65			SKIP_AUTH_CLASSES = []
66
67			def get_client(self, args, debug=False):
68			ST2_CLI_SKIP_CONFIG = os.environ.get('ST2_CLI_SKIP_CONFIG', 0)
69			ST2_CLI_SKIP_CONFIG = int(ST2_CLI_SKIP_CONFIG)
70
71			skip_config = args.skip_config
72			skip_config = skip_config or ST2_CLI_SKIP_CONFIG
73
74			# Note: Options provided as the CLI argument have the highest precedence
75			# Precedence order: cli arguments > environment variables > rc file variables
76			cli_options = ['base_url', 'auth_url', 'api_url', 'stream_url', 'api_version', 'cacert']
77			cli_options = {opt: getattr(args, opt, None) for opt in cli_options}
78			config_file_options = self._get_config_file_options(args=args)
79
80			kwargs = {}
81
82			if not skip_config:
83			# Config parsing is not skipped
84			kwargs = merge_dicts(kwargs, config_file_options)
85
86			kwargs = merge_dicts(kwargs, cli_options)
87			kwargs['debug'] = debug
88
89			client = Client(**kwargs)
90
91			if skip_config:
92			# Config parsing is skipped
93			self.LOG.info('Skipping parsing CLI config')
94			return client
95
96			# Ok to use config at this point
97			rc_config = get_config()
98
99			# Silence SSL warnings
100			silence_ssl_warnings = rc_config.get('general', {}).get('silence_ssl_warnings', False)
101			if silence_ssl_warnings:
102			requests.packages.urllib3.disable_warnings()
103
104			# We skip automatic authentication for some commands such as auth
105			try:
106			command_class_name = args.func.im_class.__name__
107			except Exception:
108			command_class_name = None
109
110			if command_class_name in self.SKIP_AUTH_CLASSES:
111			return client
112
113			# We also skip automatic authentication if token is provided via the environment variable
114			# or as a command line argument
115			env_var_token = os.environ.get('ST2_AUTH_TOKEN', None)
116			cli_argument_token = getattr(args, 'token', None)
117			env_var_api_key = os.environ.get('ST2_API_KEY', None)
118			cli_argument_api_key = getattr(args, 'api_key', None)
119			if env_var_token or cli_argument_token or env_var_api_key or cli_argument_api_key:
120			return client
121
122			# If credentials are provided in the CLI config use them and try to authenticate
123			credentials = rc_config.get('credentials', {})
124			username = credentials.get('username', None)
125			password = credentials.get('password', None)
126			cache_token = rc_config.get('cli', {}).get('cache_token', False)
127
128			if username:
129			# Credentials are provided, try to authenticate agaist the API
130			try:
131			token = self._get_auth_token(client=client, username=username, password=password,
132			cache_token=cache_token)
133			except requests.exceptions.ConnectionError as e:
134			self.LOG.warn('Auth API server is not available, skipping authentication.')
135			self.LOG.exception(e)
136			return client
137			except Exception as e:
138			print('Failed to authenticate with credentials provided in the config.')
139			raise e
140			client.token = token
141			# TODO: Hack, refactor when splitting out the client
142			os.environ['ST2_AUTH_TOKEN'] = token
143
144			return client
145
146			def _get_config_file_options(self, args):
147			"""
148			Parse the config and return kwargs which can be passed to the Client
149			constructor.
150
151			:rtype: ``dict``
152			"""
153			rc_options = self._parse_config_file(args=args)
154			result = {}
155			for kwarg_name, (section, option) in six.iteritems(CONFIG_OPTION_TO_CLIENT_KWARGS_MAP):
156			result[kwarg_name] = rc_options.get(section, {}).get(option, None)
157
158			return result
159
160			def _parse_config_file(self, args):
161			config_file_path = self._get_config_file_path(args=args)
162
163			parser = CLIConfigParser(config_file_path=config_file_path, validate_config_exists=False)
164			result = parser.parse()
165			return result
166
167			def _get_config_file_path(self, args):
168			"""
169			Retrieve path to the CLI configuration file.
170
171			:rtype: ``str``
172			"""
173			path = os.environ.get('ST2_CONFIG_FILE', ST2_CONFIG_PATH)
174
175			if args.config_file:
176			path = args.config_file
177
178			path = os.path.abspath(os.path.expanduser(path))
179			if path != ST2_CONFIG_PATH and not os.path.isfile(path):
180			raise ValueError('Config "%s" not found' % (path))
181
182			return path
183
184			def _get_auth_token(self, client, username, password, cache_token):
185			"""
186			Retrieve a valid auth token.
187
188			If caching is enabled, we will first try to retrieve cached token from a
189			file system. If cached token is expired or not available, we will try to
190			authenticate using the provided credentials and retrieve a new auth
191			token.
192
193			:rtype: ``str``
194			"""
195			if cache_token:
196			token = self._get_cached_auth_token(client=client, username=username,
197			password=password)
198			else:
199			token = None
200			if not token:
201			# Token is either expired or not available
202			token_obj = self._authenticate_and_retrieve_auth_token(client=client,
203			username=username,
204			password=password)
205
206			self._cache_auth_token(token_obj=token_obj)
207			token = token_obj.token
208
209			return token
210
211			def _get_cached_auth_token(self, client, username, password):
212			"""
213			Retrieve cached auth token from the file in the config directory.
214
215			:rtype: ``str``
216			"""
217			if not os.path.isdir(ST2_CONFIG_DIRECTORY):
218			os.makedirs(ST2_CONFIG_DIRECTORY)
219
220			cached_token_path = self._get_cached_token_path_for_user(username=username)
221
222			if not os.access(ST2_CONFIG_DIRECTORY, os.R_OK):
223			# We don't have read access to the file with a cached token
224			message = ('Unable to retrieve cached token from "%s" (user %s doesn\'t have read '
225			'access to the parent directory). Subsequent requests won\'t use a '
226			'cached token meaning they may be slower.' % (cached_token_path,
227			os.getlogin()))
228			self.LOG.warn(message)
229			return None
230
231			if not os.path.isfile(cached_token_path):
232			return None
233
234			if not os.access(cached_token_path, os.R_OK):
235			# We don't have read access to the file with a cached token
236			message = ('Unable to retrieve cached token from "%s" (user %s doesn\'t have read '
237			'access to this file). Subsequent requests won\'t use a cached token '
238			'meaning they may be slower.' % (cached_token_path, os.getlogin()))
239			self.LOG.warn(message)
240			return None
241
242			# Safety check for too permissive permissions
243			file_st_mode = oct(os.stat(cached_token_path).st_mode & 0777)
244			others_st_mode = int(file_st_mode[-1])
245
246			if others_st_mode >= 4:
247			# Every user has access to this file which is dangerous
248			message = ('Permissions (%s) for cached token file "%s" are to permissive. Please '
249			'restrict the permissions and make sure only your own user can read '
250			'from the file' % (file_st_mode, cached_token_path))
251			self.LOG.warn(message)
252
253			with open(cached_token_path) as fp:
254			data = fp.read()
255
256			try:
257			data = json.loads(data)
258
259			token = data['token']
260			expire_timestamp = data['expire_timestamp']
261			except Exception as e:
262			msg = ('File "%s" with cached token is corrupted or invalid (%s). Please delete '
263			' this file' % (cached_token_path, str(e)))
264			raise ValueError(msg)
265
266			now = int(time.time())
267			if (expire_timestamp - TOKEN_EXPIRATION_GRACE_PERIOD_SECONDS) < now:
268			self.LOG.debug('Cached token from file "%s" has expired' % (cached_token_path))
269			# Token has expired
270			return None
271
272			self.LOG.debug('Using cached token from file "%s"' % (cached_token_path))
273			return token
274
275			def _cache_auth_token(self, token_obj):
276			"""
277			Cache auth token in the config directory.
278
279			:param token_obj: Token object.
280			:type token_obj: ``object``
281			"""
282			if not os.path.isdir(ST2_CONFIG_DIRECTORY):
283			os.makedirs(ST2_CONFIG_DIRECTORY)
284
285			username = token_obj.user
286			cached_token_path = self._get_cached_token_path_for_user(username=username)
287
288			if not os.access(ST2_CONFIG_DIRECTORY, os.W_OK):
289			# We don't have write access to the file with a cached token
290			message = ('Unable to write token to "%s" (user %s doesn\'t have write '
291			'access to the parent directory). Subsequent requests won\'t use a '
292			'cached token meaning they may be slower.' % (cached_token_path,
293			os.getlogin()))
294			self.LOG.warn(message)
295			return None
296
297			if os.path.isfile(cached_token_path) and not os.access(cached_token_path, os.W_OK):
298			# We don't have write access to the file with a cached token
299			message = ('Unable to write token to "%s" (user %s doesn\'t have write '
300			'access to this file). Subsequent requests won\'t use a '
301			'cached token meaning they may be slower.' % (cached_token_path,
302			os.getlogin()))
303			self.LOG.warn(message)
304			return None
305
306			token = token_obj.token
307			expire_timestamp = parse_isotime(token_obj.expiry)
308			expire_timestamp = calendar.timegm(expire_timestamp.timetuple())
309
310			data = {}
311			data['token'] = token
312			data['expire_timestamp'] = expire_timestamp
313			data = json.dumps(data)
314
315			# Note: We explictly use fdopen instead of open + chmod to avoid a security issue.
316			# open + chmod are two operations which means that during a short time frame (between
317			# open and chmod) when file can potentially be read by other users if the default
318			# permissions used during create allow that.
319			fd = os.open(cached_token_path, os.O_WRONLY \| os.O_CREAT, 0600)
320			with os.fdopen(fd, 'w') as fp:
321			fp.write(data)
322
323			self.LOG.debug('Token has been cached in "%s"' % (cached_token_path))
324			return True
325
326			def _authenticate_and_retrieve_auth_token(self, client, username, password):
327			manager = models.ResourceManager(models.Token, client.endpoints['auth'],
328			cacert=client.cacert, debug=client.debug)
329			instance = models.Token()
330			instance = manager.create(instance, auth=(username, password))
331			return instance
332
333			def _get_cached_token_path_for_user(self, username):
334			"""
335			Retrieve cached token path for the provided username.
336			"""
337			file_name = 'token-%s' % (username)
338			result = os.path.abspath(os.path.join(ST2_CONFIG_DIRECTORY, file_name))
339			return result
340
341			def _print_config(self, args):
342			config = self._parse_config_file(args=args)
343
344			for section, options in six.iteritems(config):
345			print('[%s]' % (section))
346
347			for name, value in six.iteritems(options):
348			print('%s = %s' % (name, value))
349
350			def _print_debug_info(self, args):
351			# Print client settings
352			self._print_client_settings(args=args)
353
354			# Print exception traceback
355			traceback.print_exc()
356
357			def _print_client_settings(self, args):
358			client = self.client
359
360			if not client:
361			return
362
363			config_file_path = self._get_config_file_path(args=args)
364
365			print('CLI settings:')
366			print('----------------')
367			print('Config file path: %s' % (config_file_path))
368			print('Client settings:')
369			print('----------------')
370			print('ST2_BASE_URL: %s' % (client.endpoints['base']))
371			print('ST2_AUTH_URL: %s' % (client.endpoints['auth']))
372			print('ST2_API_URL: %s' % (client.endpoints['api']))
373			print('ST2_STREAM_URL: %s' % (client.endpoints['stream']))
374			print('ST2_AUTH_TOKEN: %s' % (os.environ.get('ST2_AUTH_TOKEN')))
375			print('')
376			print('Proxy settings:')
377			print('---------------')
378			print('HTTP_PROXY: %s' % (os.environ.get('HTTP_PROXY', '')))
379			print('HTTPS_PROXY: %s' % (os.environ.get('HTTPS_PROXY', '')))
380			print('')
381

Plexxi / st2

GitHub Access Token became invalid

Push — plexxi-v2.3.2 ( 5d46fe )

BaseCLIApp._get_config_file_path() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like