ShellCommandAction.get_sanitized_full_command_string() - Code Metrics - Inspection of "Merge pull request #3867 from StackStorm/local_rem..." - StackStorm/st2 - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 6aeca2...dec5f2 )

unknown

created 2017-12-01 11:00 UTC

get_sanitized_full_command_string() A

↳ Parent: ShellCommandAction

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	2
c	0
b	0
f	0
dl	0
loc	14
rs	9.4285

# -*- coding: utf-8 -*-
# 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.
# pylint: disable=not-context-manager

import os
import pwd
import six
import sys
import copy
import traceback

from oslo_config import cfg

from st2common import log as logging
from st2common.models.base import DictSerializableClassMixin
from st2common.util.shell import quote_unix
from st2common.constants.action import LIBS_DIR as ACTION_LIBS_DIR
from st2common.util.secrets import get_secret_parameters
from st2common.util.secrets import mask_secret_parameters
from st2common.constants.secrets import MASKED_ATTRIBUTE_VALUE

__all__ = [
    'ShellCommandAction',
    'ShellScriptAction',
    'RemoteAction',
    'RemoteScriptAction',
    'ResolvedActionParameters'
]

LOG = logging.getLogger(__name__)

LOGGED_USER_USERNAME = pwd.getpwuid(os.getuid())[0]

# Flags which are passed to every sudo invocation
SUDO_COMMON_OPTIONS = [
    '-E'  # we want to preserve the environment of the user which ran sudo
]

# Flags which are only passed to sudo when not running as current user and when
# -u flag is used
SUDO_DIFFERENT_USER_OPTIONS = [
    '-H'  # we want $HOME to reflect the home directory of the requested / target user
]


class ShellCommandAction(object):
    EXPORT_CMD = 'export'

    def __init__(self, name, action_exec_id, command, user, env_vars=None, sudo=False,
                 timeout=None, cwd=None, sudo_password=None):
        self.name = name
        self.action_exec_id = action_exec_id
        self.command = command
        self.env_vars = env_vars or {}
        self.user = user
        self.sudo = sudo
        self.timeout = timeout
        self.cwd = cwd
        self.sudo_password = sudo_password

    def get_full_command_string(self):
        # Note: We pass -E to sudo because we want to preserve user provided environment variables
        if self.sudo:
            command = quote_unix(self.command)
            sudo_arguments = ' '.join(self._get_common_sudo_arguments())
            command = 'sudo %s -- bash -c %s' % (sudo_arguments, command)
        else:
            if self.user and self.user != LOGGED_USER_USERNAME:
                # Need to use sudo to run as a different (requested) user
                user = quote_unix(self.user)
                sudo_arguments = ' '.join(self._get_user_sudo_arguments(user=user))
                command = quote_unix(self.command)
                command = 'sudo %s -- bash -c %s' % (sudo_arguments, command)

            else:
                command = self.command

        return command

    def get_sanitized_full_command_string(self):
        """
        Get a command string which can be used inside the log messages (if provided, sudo password
        is masked).

        :rtype: ``password``
        """
        command_string = self.get_full_command_string()

        if self.sudo_password:
            # Mask sudo password
            command_string = 'echo -e \'%s\n\' | %s' % (MASKED_ATTRIBUTE_VALUE, command_string)

        return command_string

    def get_timeout(self):
        return self.timeout

    def get_cwd(self):
        return self.cwd

    def _get_common_sudo_arguments(self):
        """
        Retrieve a list of flags which are passed to sudo on every invocation.

        :rtype: ``list``
        """
        flags = []

        if self.sudo_password:
            # Note: We use subprocess.Popen in local runner so we provide password via subprocess
            # stdin (using echo -e won't work when using subprocess.Popen)
            flags.append('-S')

        flags = flags + SUDO_COMMON_OPTIONS

        return flags

    def _get_user_sudo_arguments(self, user):
        """
        Retrieve a list of flags which are passed to sudo when running as a different user and "-u"
        flag is used.

        :rtype: ``list``
        """
        flags = self._get_common_sudo_arguments()
        flags += SUDO_DIFFERENT_USER_OPTIONS
        flags += ['-u', user]

        return flags

    def _get_env_vars_export_string(self):
        if self.env_vars:
            # Envrionment variables could contain spaces and open us to shell
            # injection attacks. Always quote the key and the value.
            exports = ' '.join(
                '%s=%s' % (quote_unix(k), quote_unix(v))
                for k, v in self.env_vars.iteritems()
            )
            shell_env_str = '%s %s' % (ShellCommandAction.EXPORT_CMD, exports)
        else:
            shell_env_str = ''

        return shell_env_str

    def _get_command_string(self, cmd, args):
        """
        Escape the command arguments and form a command string.

        :type cmd: ``str``
        :type args: ``list``

        :rtype: ``str``
        """
        assert isinstance(args, (list, tuple))

        args = [quote_unix(arg) for arg in args]
        args = ' '.join(args)
        result = '%s %s' % (cmd, args)
        return result

    def _get_error_result(self):
        """
        Prepares a structured error result based on the exception.

        :type e: ``Exception``

        :rtype: ``dict``
        """
        _, exc_value, exc_traceback = sys.exc_info()

        exc_value = str(exc_value)
        exc_traceback = ''.join(traceback.format_tb(exc_traceback))

        result = {}
        result['failed'] = True
        result['succeeded'] = False
        result['error'] = exc_value
        result['traceback'] = exc_traceback
        return result


class ShellScriptAction(ShellCommandAction):
    def __init__(self, name, action_exec_id, script_local_path_abs, named_args=None,
                 positional_args=None, env_vars=None, user=None, sudo=False, timeout=None,
                 cwd=None, sudo_password=None):
        super(ShellScriptAction, self).__init__(name=name, action_exec_id=action_exec_id,
                                                command=None, user=user, env_vars=env_vars,
                                                sudo=sudo, timeout=timeout,
                                                cwd=cwd, sudo_password=sudo_password)
        self.script_local_path_abs = script_local_path_abs
        self.named_args = named_args
        self.positional_args = positional_args

    def get_full_command_string(self):
        return self._format_command()

    def _format_command(self):
        script_arguments = self._get_script_arguments(named_args=self.named_args,
                                                      positional_args=self.positional_args)
        if self.sudo:
            if script_arguments:
                command = quote_unix('%s %s' % (self.script_local_path_abs, script_arguments))
            else:
                command = quote_unix(self.script_local_path_abs)

            sudo_arguments = ' '.join(self._get_common_sudo_arguments())
            command = 'sudo %s -- bash -c %s' % (sudo_arguments, command)
        else:
            if self.user and self.user != LOGGED_USER_USERNAME:
                # Need to use sudo to run as a different user
                user = quote_unix(self.user)

                if script_arguments:
                    command = quote_unix('%s %s' % (self.script_local_path_abs, script_arguments))
                else:
                    command = quote_unix(self.script_local_path_abs)

                sudo_arguments = ' '.join(self._get_user_sudo_arguments(user=user))
                command = 'sudo %s -- bash -c %s' % (sudo_arguments, command)
            else:
                script_path = quote_unix(self.script_local_path_abs)

                if script_arguments:
                    command = '%s %s' % (script_path, script_arguments)
                else:
                    command = script_path
        return command

    def _get_script_arguments(self, named_args=None, positional_args=None):
        """
        Build a string of named and positional arguments which are passed to the
        script.

        :param named_args: Dictionary with named arguments.
        :type named_args: ``dict``.

        :param positional_args: List with positional arguments.
        :type positional_args: ``dict``.

        :rtype: ``str``
        """
        command_parts = []

        # add all named_args in the format <kwarg_op>name=value (e.g. --name=value)
        if named_args is not None:
            for (arg, value) in six.iteritems(named_args):
                if value is None or (isinstance(value, (str, unicode)) and len(value) < 1):

                    LOG.debug('Ignoring arg %s as its value is %s.', arg, value)
                    continue

                if isinstance(value, bool):
                    if value is True:
                        command_parts.append(arg)
                else:
                    values = (quote_unix(arg), quote_unix(six.text_type(value)))
                    command_parts.append(six.text_type('%s=%s' % values))

        # add the positional args
        if positional_args:
            quoted_pos_args = [quote_unix(pos_arg) for pos_arg in positional_args]
            pos_args_string = ' '.join(quoted_pos_args)
            command_parts.append(pos_args_string)
        return ' '.join(command_parts)


class SSHCommandAction(ShellCommandAction):
    def __init__(self, name, action_exec_id, command, env_vars, user, password=None, pkey=None,
                 hosts=None, parallel=True, sudo=False, timeout=None, cwd=None, passphrase=None,
                 sudo_password=None):
        super(SSHCommandAction, self).__init__(name=name, action_exec_id=action_exec_id,
                                               command=command, env_vars=env_vars, user=user,
                                               sudo=sudo, timeout=timeout, cwd=cwd,
                                               sudo_password=sudo_password)
        self.hosts = hosts
        self.parallel = parallel
        self.pkey = pkey
        self.passphrase = passphrase
        self.password = password

    def is_parallel(self):
        return self.parallel

    def is_sudo(self):
        return self.sudo

    def get_user(self):
        return self.user

    def get_hosts(self):
        return self.hosts

    def is_pkey_authentication(self):
        return self.pkey is not None

    def get_pkey(self):
        return self.pkey

    def get_password(self):
        return self.password

    def get_command(self):
        return self.command

    def __str__(self):
        str_rep = []
        str_rep.append('%s@%s(name: %s' % (self.__class__.__name__, id(self), self.name))
        str_rep.append('id: %s' % self.action_exec_id)
        str_rep.append('command: %s' % self.command)
        str_rep.append('user: %s' % self.user)
        str_rep.append('sudo: %s' % str(self.sudo))
        str_rep.append('parallel: %s' % str(self.parallel))
        str_rep.append('hosts: %s)' % str(self.hosts))
        return ', '.join(str_rep)


class RemoteAction(SSHCommandAction):
    def __init__(self, name, action_exec_id, command, env_vars=None, on_behalf_user=None,
                 user=None, password=None, private_key=None, hosts=None, parallel=True, sudo=False,
                 timeout=None, cwd=None, passphrase=None, sudo_password=None):
        super(RemoteAction, self).__init__(name=name, action_exec_id=action_exec_id,
                                           command=command, env_vars=env_vars, user=user,
                                           hosts=hosts, parallel=parallel, sudo=sudo,
                                           timeout=timeout, cwd=cwd, passphrase=passphrase,
                                           sudo_password=sudo_password)
        self.password = password
        self.private_key = private_key
        self.passphrase = passphrase
        self.on_behalf_user = on_behalf_user  # Used for audit purposes.
        self.timeout = timeout

    def get_on_behalf_user(self):
        return self.on_behalf_user

    def __str__(self):
        str_rep = []
        str_rep.append('%s@%s(name: %s' % (self.__class__.__name__, id(self), self.name))
        str_rep.append('id: %s' % self.action_exec_id)
        str_rep.append('command: %s' % self.command)
        str_rep.append('user: %s' % self.user)
        str_rep.append('on_behalf_user: %s' % self.on_behalf_user)
        str_rep.append('sudo: %s' % str(self.sudo))
        str_rep.append('parallel: %s' % str(self.parallel))
        str_rep.append('hosts: %s)' % str(self.hosts))
        str_rep.append('timeout: %s)' % str(self.timeout))

        return ', '.join(str_rep)


class RemoteScriptAction(ShellScriptAction):
    def __init__(self, name, action_exec_id, script_local_path_abs, script_local_libs_path_abs,
                 named_args=None, positional_args=None, env_vars=None, on_behalf_user=None,
                 user=None, password=None, private_key=None, remote_dir=None, hosts=None,
                 parallel=True, sudo=False, timeout=None, cwd=None, sudo_password=None):
        super(RemoteScriptAction, self).__init__(name=name, action_exec_id=action_exec_id,
                                                 script_local_path_abs=script_local_path_abs,
                                                 user=user,
                                                 named_args=named_args,
                                                 positional_args=positional_args, env_vars=env_vars,
                                                 sudo=sudo, timeout=timeout, cwd=cwd,
                                                 sudo_password=sudo_password)
        self.script_local_libs_path_abs = script_local_libs_path_abs
        self.script_local_dir, self.script_name = os.path.split(self.script_local_path_abs)
        self.remote_dir = remote_dir if remote_dir is not None else '/tmp'
        self.remote_libs_path_abs = os.path.join(self.remote_dir, ACTION_LIBS_DIR)
        self.on_behalf_user = on_behalf_user
        self.password = password
        self.private_key = private_key
        self.remote_script = os.path.join(self.remote_dir, quote_unix(self.script_name))
        self.hosts = hosts
        self.parallel = parallel
        self.command = self._format_command()
        LOG.debug('RemoteScriptAction: command to run on remote box: %s', self.command)

    def get_remote_script_abs_path(self):
        return self.remote_script

    def get_local_script_abs_path(self):
        return self.script_local_path_abs

    def get_remote_libs_path_abs(self):
        return self.remote_libs_path_abs

    def get_local_libs_path_abs(self):
        return self.script_local_libs_path_abs

    def get_remote_base_dir(self):
        return self.remote_dir

    def _format_command(self):
        script_arguments = self._get_script_arguments(named_args=self.named_args,
                                                      positional_args=self.positional_args)

        if script_arguments:
            command = '%s %s' % (self.remote_script, script_arguments)
        else:
            command = self.remote_script

        return command

    def __str__(self):
        str_rep = []
        str_rep.append('%s@%s(name: %s' % (self.__class__.__name__, id(self), self.name))
        str_rep.append('id: %s' % self.action_exec_id)
        str_rep.append('local_script: %s' % self.script_local_path_abs)
        str_rep.append('local_libs: %s' % self.script_local_libs_path_abs)
        str_rep.append('remote_dir: %s' % self.remote_dir)
        str_rep.append('remote_libs: %s' % self.remote_libs_path_abs)
        str_rep.append('named_args: %s' % self.named_args)
        str_rep.append('positional_args: %s' % self.positional_args)
        str_rep.append('user: %s' % self.user)
        str_rep.append('on_behalf_user: %s' % self.on_behalf_user)
        str_rep.append('sudo: %s' % self.sudo)
        str_rep.append('parallel: %s' % self.parallel)
        str_rep.append('hosts: %s)' % self.hosts)

        return ', '.join(str_rep)


class ResolvedActionParameters(DictSerializableClassMixin):
    """
    Class which contains resolved runner and action parameters for a particular action.
    """

    def __init__(self, action_db, runner_type_db, runner_parameters=None, action_parameters=None):
        self._action_db = action_db
        self._runner_type_db = runner_type_db
        self._runner_parameters = runner_parameters
        self._action_parameters = action_parameters

    def mask_secrets(self, value):
        result = copy.deepcopy(value)

        runner_parameters = result['runner_parameters']
        action_parameters = result['action_parameters']

        runner_parameters_specs = self._runner_type_db.runner_parameters
        action_parameters_sepcs = self._action_db.parameters

        secret_runner_parameters = get_secret_parameters(parameters=runner_parameters_specs)
        secret_action_parameters = get_secret_parameters(parameters=action_parameters_sepcs)

        runner_parameters = mask_secret_parameters(parameters=runner_parameters,
                                                   secret_parameters=secret_runner_parameters)
        action_parameters = mask_secret_parameters(parameters=action_parameters,
                                                   secret_parameters=secret_action_parameters)
        result['runner_parameters'] = runner_parameters
        result['action_parameters'] = action_parameters

        return result

    def to_serializable_dict(self, mask_secrets=False):
        result = {}
        result['runner_parameters'] = self._runner_parameters
        result['action_parameters'] = self._action_parameters

        if mask_secrets and cfg.CONF.log.mask_secrets:
            result = self.mask_secrets(value=result)

        return result


1			# -- coding: utf-8 --
2			# Licensed to the StackStorm, Inc ('StackStorm') under one or more
3			# contributor license agreements. See the NOTICE file distributed with
4			# this work for additional information regarding copyright ownership.
5			# The ASF licenses this file to You under the Apache License, Version 2.0
6			# (the "License"); you may not use this file except in compliance with
7			# the License. You may obtain a copy of the License at
8			#
9			# http://www.apache.org/licenses/LICENSE-2.0
10			#
11			# Unless required by applicable law or agreed to in writing, software
12			# distributed under the License is distributed on an "AS IS" BASIS,
13			# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14			# See the License for the specific language governing permissions and
15			# limitations under the License.
16			# pylint: disable=not-context-manager
17
18			import os
19			import pwd
20			import six
21			import sys
22			import copy
23			import traceback
24
25			from oslo_config import cfg
26
27			from st2common import log as logging
28			from st2common.models.base import DictSerializableClassMixin
29			from st2common.util.shell import quote_unix
30			from st2common.constants.action import LIBS_DIR as ACTION_LIBS_DIR
31			from st2common.util.secrets import get_secret_parameters
32			from st2common.util.secrets import mask_secret_parameters
33			from st2common.constants.secrets import MASKED_ATTRIBUTE_VALUE
34
35			__all__ = [
36			'ShellCommandAction',
37			'ShellScriptAction',
38			'RemoteAction',
39			'RemoteScriptAction',
40			'ResolvedActionParameters'
41			]
42
43			LOG = logging.getLogger(__name__)
44
45			LOGGED_USER_USERNAME = pwd.getpwuid(os.getuid())[0]
46
47			# Flags which are passed to every sudo invocation
48			SUDO_COMMON_OPTIONS = [
49			'-E' # we want to preserve the environment of the user which ran sudo
50			]
51
52			# Flags which are only passed to sudo when not running as current user and when
53			# -u flag is used
54			SUDO_DIFFERENT_USER_OPTIONS = [
55			'-H' # we want $HOME to reflect the home directory of the requested / target user
56			]
57
58
59			class ShellCommandAction(object):
60			EXPORT_CMD = 'export'
61
62			def __init__(self, name, action_exec_id, command, user, env_vars=None, sudo=False,
63			timeout=None, cwd=None, sudo_password=None):
64			self.name = name
65			self.action_exec_id = action_exec_id
66			self.command = command
67			self.env_vars = env_vars or {}
68			self.user = user
69			self.sudo = sudo
70			self.timeout = timeout
71			self.cwd = cwd
72			self.sudo_password = sudo_password
73
74			def get_full_command_string(self):
75			# Note: We pass -E to sudo because we want to preserve user provided environment variables
76			if self.sudo:
77			command = quote_unix(self.command)
78			sudo_arguments = ' '.join(self._get_common_sudo_arguments())
79			command = 'sudo %s -- bash -c %s' % (sudo_arguments, command)
80			else:
81			if self.user and self.user != LOGGED_USER_USERNAME:
82			# Need to use sudo to run as a different (requested) user
83			user = quote_unix(self.user)
84			sudo_arguments = ' '.join(self._get_user_sudo_arguments(user=user))
85			command = quote_unix(self.command)
86			command = 'sudo %s -- bash -c %s' % (sudo_arguments, command)
87
88			else:
89			command = self.command
90
91			return command
92
93			def get_sanitized_full_command_string(self):
94			"""
95			Get a command string which can be used inside the log messages (if provided, sudo password
96			is masked).
97
98			:rtype: ``password``
99			"""
100			command_string = self.get_full_command_string()
101
102			if self.sudo_password:
103			# Mask sudo password
104			command_string = 'echo -e \'%s\n\' \| %s' % (MASKED_ATTRIBUTE_VALUE, command_string)
105
106			return command_string
107
108			def get_timeout(self):
109			return self.timeout
110
111			def get_cwd(self):
112			return self.cwd
113
114			def _get_common_sudo_arguments(self):
115			"""
116			Retrieve a list of flags which are passed to sudo on every invocation.
117
118			:rtype: ``list``
119			"""
120			flags = []
121
122			if self.sudo_password:
123			# Note: We use subprocess.Popen in local runner so we provide password via subprocess
124			# stdin (using echo -e won't work when using subprocess.Popen)
125			flags.append('-S')
126
127			flags = flags + SUDO_COMMON_OPTIONS
128
129			return flags
130
131			def _get_user_sudo_arguments(self, user):
132			"""
133			Retrieve a list of flags which are passed to sudo when running as a different user and "-u"
134			flag is used.
135
136			:rtype: ``list``
137			"""
138			flags = self._get_common_sudo_arguments()
139			flags += SUDO_DIFFERENT_USER_OPTIONS
140			flags += ['-u', user]
141
142			return flags
143
144			def _get_env_vars_export_string(self):
145			if self.env_vars:
146			# Envrionment variables could contain spaces and open us to shell
147			# injection attacks. Always quote the key and the value.
148			exports = ' '.join(
149			'%s=%s' % (quote_unix(k), quote_unix(v))
150			for k, v in self.env_vars.iteritems()
151			)
152			shell_env_str = '%s %s' % (ShellCommandAction.EXPORT_CMD, exports)
153			else:
154			shell_env_str = ''
155
156			return shell_env_str
157
158			def _get_command_string(self, cmd, args):
159			"""
160			Escape the command arguments and form a command string.
161
162			:type cmd: ``str``
163			:type args: ``list``
164
165			:rtype: ``str``
166			"""
167			assert isinstance(args, (list, tuple))
168
169			args = [quote_unix(arg) for arg in args]
170			args = ' '.join(args)
171			result = '%s %s' % (cmd, args)
172			return result
173
174			def _get_error_result(self):
175			"""
176			Prepares a structured error result based on the exception.
177
178			:type e: ``Exception``
179
180			:rtype: ``dict``
181			"""
182			_, exc_value, exc_traceback = sys.exc_info()
183
184			exc_value = str(exc_value)
185			exc_traceback = ''.join(traceback.format_tb(exc_traceback))
186
187			result = {}
188			result['failed'] = True
189			result['succeeded'] = False
190			result['error'] = exc_value
191			result['traceback'] = exc_traceback
192			return result
193
194
195			class ShellScriptAction(ShellCommandAction):
196			def __init__(self, name, action_exec_id, script_local_path_abs, named_args=None,
197			positional_args=None, env_vars=None, user=None, sudo=False, timeout=None,
198			cwd=None, sudo_password=None):
199			super(ShellScriptAction, self).__init__(name=name, action_exec_id=action_exec_id,
200			command=None, user=user, env_vars=env_vars,
201			sudo=sudo, timeout=timeout,
202			cwd=cwd, sudo_password=sudo_password)
203			self.script_local_path_abs = script_local_path_abs
204			self.named_args = named_args
205			self.positional_args = positional_args
206
207			def get_full_command_string(self):
208			return self._format_command()
209
210			def _format_command(self):
211			script_arguments = self._get_script_arguments(named_args=self.named_args,
212			positional_args=self.positional_args)
213			if self.sudo:
214			if script_arguments:
215			command = quote_unix('%s %s' % (self.script_local_path_abs, script_arguments))
216			else:
217			command = quote_unix(self.script_local_path_abs)
218
219			sudo_arguments = ' '.join(self._get_common_sudo_arguments())
220			command = 'sudo %s -- bash -c %s' % (sudo_arguments, command)
221			else:
222			if self.user and self.user != LOGGED_USER_USERNAME:
223			# Need to use sudo to run as a different user
224			user = quote_unix(self.user)
225
226			if script_arguments:
227			command = quote_unix('%s %s' % (self.script_local_path_abs, script_arguments))
228			else:
229			command = quote_unix(self.script_local_path_abs)
230
231			sudo_arguments = ' '.join(self._get_user_sudo_arguments(user=user))
232			command = 'sudo %s -- bash -c %s' % (sudo_arguments, command)
233			else:
234			script_path = quote_unix(self.script_local_path_abs)
235
236			if script_arguments:
237			command = '%s %s' % (script_path, script_arguments)
238			else:
239			command = script_path
240			return command
241
242			def _get_script_arguments(self, named_args=None, positional_args=None):
243			"""
244			Build a string of named and positional arguments which are passed to the
245			script.
246
247			:param named_args: Dictionary with named arguments.
248			:type named_args: ``dict``.
249
250			:param positional_args: List with positional arguments.
251			:type positional_args: ``dict``.
252
253			:rtype: ``str``
254			"""
255			command_parts = []
256
257			# add all named_args in the format <kwarg_op>name=value (e.g. --name=value)
258			if named_args is not None:
259			for (arg, value) in six.iteritems(named_args):
260			if value is None or (isinstance(value, (str, unicode)) and len(value) < 1):
			0 ignored issues – show Comprehensibility Best Practice introduced 2016-05-04 06:14 UTC by Report Bug Copy Issue Report Undefined variable 'unicode' Loading history...
261			LOG.debug('Ignoring arg %s as its value is %s.', arg, value)
262			continue
263
264			if isinstance(value, bool):
265			if value is True:
266			command_parts.append(arg)
267			else:
268			values = (quote_unix(arg), quote_unix(six.text_type(value)))
269			command_parts.append(six.text_type('%s=%s' % values))
270
271			# add the positional args
272			if positional_args:
273			quoted_pos_args = [quote_unix(pos_arg) for pos_arg in positional_args]
274			pos_args_string = ' '.join(quoted_pos_args)
275			command_parts.append(pos_args_string)
276			return ' '.join(command_parts)
277
278
279			class SSHCommandAction(ShellCommandAction):
280			def __init__(self, name, action_exec_id, command, env_vars, user, password=None, pkey=None,
281			hosts=None, parallel=True, sudo=False, timeout=None, cwd=None, passphrase=None,
282			sudo_password=None):
283			super(SSHCommandAction, self).__init__(name=name, action_exec_id=action_exec_id,
284			command=command, env_vars=env_vars, user=user,
285			sudo=sudo, timeout=timeout, cwd=cwd,
286			sudo_password=sudo_password)
287			self.hosts = hosts
288			self.parallel = parallel
289			self.pkey = pkey
290			self.passphrase = passphrase
291			self.password = password
292
293			def is_parallel(self):
294			return self.parallel
295
296			def is_sudo(self):
297			return self.sudo
298
299			def get_user(self):
300			return self.user
301
302			def get_hosts(self):
303			return self.hosts
304
305			def is_pkey_authentication(self):
306			return self.pkey is not None
307
308			def get_pkey(self):
309			return self.pkey
310
311			def get_password(self):
312			return self.password
313
314			def get_command(self):
315			return self.command
316
317			def __str__(self):
318			str_rep = []
319			str_rep.append('%s@%s(name: %s' % (self.__class__.__name__, id(self), self.name))
320			str_rep.append('id: %s' % self.action_exec_id)
321			str_rep.append('command: %s' % self.command)
322			str_rep.append('user: %s' % self.user)
323			str_rep.append('sudo: %s' % str(self.sudo))
324			str_rep.append('parallel: %s' % str(self.parallel))
325			str_rep.append('hosts: %s)' % str(self.hosts))
326			return ', '.join(str_rep)
327
328
329			class RemoteAction(SSHCommandAction):
330			def __init__(self, name, action_exec_id, command, env_vars=None, on_behalf_user=None,
331			user=None, password=None, private_key=None, hosts=None, parallel=True, sudo=False,
332			timeout=None, cwd=None, passphrase=None, sudo_password=None):
333			super(RemoteAction, self).__init__(name=name, action_exec_id=action_exec_id,
334			command=command, env_vars=env_vars, user=user,
335			hosts=hosts, parallel=parallel, sudo=sudo,
336			timeout=timeout, cwd=cwd, passphrase=passphrase,
337			sudo_password=sudo_password)
338			self.password = password
339			self.private_key = private_key
340			self.passphrase = passphrase
341			self.on_behalf_user = on_behalf_user # Used for audit purposes.
342			self.timeout = timeout
343
344			def get_on_behalf_user(self):
345			return self.on_behalf_user
346
347			def __str__(self):
348			str_rep = []
349			str_rep.append('%s@%s(name: %s' % (self.__class__.__name__, id(self), self.name))
350			str_rep.append('id: %s' % self.action_exec_id)
351			str_rep.append('command: %s' % self.command)
352			str_rep.append('user: %s' % self.user)
353			str_rep.append('on_behalf_user: %s' % self.on_behalf_user)
354			str_rep.append('sudo: %s' % str(self.sudo))
355			str_rep.append('parallel: %s' % str(self.parallel))
356			str_rep.append('hosts: %s)' % str(self.hosts))
357			str_rep.append('timeout: %s)' % str(self.timeout))
358
359			return ', '.join(str_rep)
360
361
362			class RemoteScriptAction(ShellScriptAction):
363			def __init__(self, name, action_exec_id, script_local_path_abs, script_local_libs_path_abs,
364			named_args=None, positional_args=None, env_vars=None, on_behalf_user=None,
365			user=None, password=None, private_key=None, remote_dir=None, hosts=None,
366			parallel=True, sudo=False, timeout=None, cwd=None, sudo_password=None):
367			super(RemoteScriptAction, self).__init__(name=name, action_exec_id=action_exec_id,
368			script_local_path_abs=script_local_path_abs,
369			user=user,
370			named_args=named_args,
371			positional_args=positional_args, env_vars=env_vars,
372			sudo=sudo, timeout=timeout, cwd=cwd,
373			sudo_password=sudo_password)
374			self.script_local_libs_path_abs = script_local_libs_path_abs
375			self.script_local_dir, self.script_name = os.path.split(self.script_local_path_abs)
376			self.remote_dir = remote_dir if remote_dir is not None else '/tmp'
377			self.remote_libs_path_abs = os.path.join(self.remote_dir, ACTION_LIBS_DIR)
378			self.on_behalf_user = on_behalf_user
379			self.password = password
380			self.private_key = private_key
381			self.remote_script = os.path.join(self.remote_dir, quote_unix(self.script_name))
382			self.hosts = hosts
383			self.parallel = parallel
384			self.command = self._format_command()
385			LOG.debug('RemoteScriptAction: command to run on remote box: %s', self.command)
386
387			def get_remote_script_abs_path(self):
388			return self.remote_script
389
390			def get_local_script_abs_path(self):
391			return self.script_local_path_abs
392
393			def get_remote_libs_path_abs(self):
394			return self.remote_libs_path_abs
395
396			def get_local_libs_path_abs(self):
397			return self.script_local_libs_path_abs
398
399			def get_remote_base_dir(self):
400			return self.remote_dir
401
402			def _format_command(self):
403			script_arguments = self._get_script_arguments(named_args=self.named_args,
404			positional_args=self.positional_args)
405
406			if script_arguments:
407			command = '%s %s' % (self.remote_script, script_arguments)
408			else:
409			command = self.remote_script
410
411			return command
412
413			def __str__(self):
414			str_rep = []
415			str_rep.append('%s@%s(name: %s' % (self.__class__.__name__, id(self), self.name))
416			str_rep.append('id: %s' % self.action_exec_id)
417			str_rep.append('local_script: %s' % self.script_local_path_abs)
418			str_rep.append('local_libs: %s' % self.script_local_libs_path_abs)
419			str_rep.append('remote_dir: %s' % self.remote_dir)
420			str_rep.append('remote_libs: %s' % self.remote_libs_path_abs)
421			str_rep.append('named_args: %s' % self.named_args)
422			str_rep.append('positional_args: %s' % self.positional_args)
423			str_rep.append('user: %s' % self.user)
424			str_rep.append('on_behalf_user: %s' % self.on_behalf_user)
425			str_rep.append('sudo: %s' % self.sudo)
426			str_rep.append('parallel: %s' % self.parallel)
427			str_rep.append('hosts: %s)' % self.hosts)
428
429			return ', '.join(str_rep)
430
431
432			class ResolvedActionParameters(DictSerializableClassMixin):
433			"""
434			Class which contains resolved runner and action parameters for a particular action.
435			"""
436
437			def __init__(self, action_db, runner_type_db, runner_parameters=None, action_parameters=None):
438			self._action_db = action_db
439			self._runner_type_db = runner_type_db
440			self._runner_parameters = runner_parameters
441			self._action_parameters = action_parameters
442
443			def mask_secrets(self, value):
444			result = copy.deepcopy(value)
445
446			runner_parameters = result['runner_parameters']
447			action_parameters = result['action_parameters']
448
449			runner_parameters_specs = self._runner_type_db.runner_parameters
450			action_parameters_sepcs = self._action_db.parameters
451
452			secret_runner_parameters = get_secret_parameters(parameters=runner_parameters_specs)
453			secret_action_parameters = get_secret_parameters(parameters=action_parameters_sepcs)
454
455			runner_parameters = mask_secret_parameters(parameters=runner_parameters,
456			secret_parameters=secret_runner_parameters)
457			action_parameters = mask_secret_parameters(parameters=action_parameters,
458			secret_parameters=secret_action_parameters)
459			result['runner_parameters'] = runner_parameters
460			result['action_parameters'] = action_parameters
461
462			return result
463
464			def to_serializable_dict(self, mask_secrets=False):
465			result = {}
466			result['runner_parameters'] = self._runner_parameters
467			result['action_parameters'] = self._action_parameters
468
469			if mask_secrets and cfg.CONF.log.mask_secrets:
470			result = self.mask_secrets(value=result)
471
472			return result
473

StackStorm / st2

Push — master ( 6aeca2...dec5f2 )

get_sanitized_full_command_string() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like