saltstack/salt
1
0
Fork 0
mirror of https://github.com/saltstack/salt.git synced 2025-04-17 10:10:20 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 2

Merge pull request #26759 from terminalmage/bp-26726

Browse source 
Backport PR #26726 to 2015.5 branch
This commit is contained in:
Mike Place 2015-08-31 08:39:20 -06:00
commit 645998dbd3
6 changed files with 620 additions and 164 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

85
salt/modules/cmdmod.py
View file

@ -53,6 +53,19 @@ def __virtual__():
return __virtualname__
def _check_cb(cb_):
'''
If the callback is None or is not callable, return a lambda that returns
the value passed.
'''
if cb_ is not None:
if hasattr(cb_, '__call__'):
return cb_
else:
log.error('log_callback is not callable, ignoring')
return lambda x: x
def _python_shell_default(python_shell, __pub_jid):
'''
Set python_shell default based on remote execution and __opts__['cmd_safe']
@ -200,6 +213,7 @@ def _run(cmd,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
output_loglevel='debug',
log_callback=None,
runas=None,
shell=DEFAULT_SHELL,
python_shell=False,
@ -225,6 +239,8 @@ def _run(cmd,
'Check to ensure that the shell <{0}> is valid for this user.'
.format(shell))
log_callback = _check_cb(log_callback)
# Set the default working directory to the home directory of the user
# salt-minion is running as. Defaults to home directory of user under which
# the minion is running.
@ -331,11 +347,12 @@ def _run(cmd,
# Always log the shell commands at INFO unless quiet logging is
# requested. The command output is what will be controlled by the
# 'loglevel' parameter.
log.info(
msg = (
'Executing command {0!r} {1}in directory {2!r}'.format(
cmd, 'as user {0!r} '.format(runas) if runas else '', cwd
)
)
log.info(log_callback(msg))
if reset_system_locale is True:
if not salt.utils.is_windows():
@ -442,7 +459,8 @@ def _run(cmd,
if timeout:
to = ' (timeout: {0}s)'.format(timeout)
if _check_loglevel(output_loglevel) is not None:
log.debug('Running {0} in VT{1}'.format(cmd, to))
msg = 'Running {0} in VT{1}'.format(cmd, to)
log.debug(log_callback(msg))
stdout, stderr = '', ''
now = time.time()
if timeout:
@ -540,6 +558,7 @@ def _run_quiet(cmd,
stdin=stdin,
stderr=subprocess.STDOUT,
output_loglevel='quiet',
log_callback=None,
shell=shell,
python_shell=python_shell,
env=env,
@ -578,6 +597,7 @@ def _run_all_quiet(cmd,
python_shell=python_shell,
env=env,
output_loglevel='quiet',
log_callback=None,
template=template,
umask=umask,
timeout=timeout,
@ -599,6 +619,7 @@ def run(cmd,
rstrip=True,
umask=None,
output_loglevel='debug',
log_callback=None,
timeout=None,
reset_system_locale=True,
ignore_retcode=False,
@ -676,6 +697,7 @@ def run(cmd,
rstrip=rstrip,
umask=umask,
output_loglevel=output_loglevel,
log_callback=log_callback,
timeout=timeout,
reset_system_locale=reset_system_locale,
ignore_retcode=ignore_retcode,
@ -684,6 +706,8 @@ def run(cmd,
pillar_override=kwargs.get('pillar'),
use_vt=use_vt)
log_callback = _check_cb(log_callback)
if 'pid' in ret and '__pub_jid' in kwargs:
# Stuff the child pid in the JID file
try:
@ -710,11 +734,12 @@ def run(cmd,
if not ignore_retcode and ret['retcode'] != 0:
if lvl < LOG_LEVELS['error']:
lvl = LOG_LEVELS['error']
log.error(
msg = (
'Command {0!r} failed with return code: {1}'
.format(cmd, ret['retcode'])
)
log.log(lvl, 'output: {0}'.format(ret['stdout']))
log.error(log_callback(msg))
log.log(lvl, 'output: {0}'.format(log_callback(ret['stdout'])))
return ret['stdout']
@ -729,6 +754,7 @@ def shell(cmd,
rstrip=True,
umask=None,
output_loglevel='debug',
log_callback=None,
quiet=False,
timeout=None,
reset_system_locale=True,
@ -803,6 +829,7 @@ def shell(cmd,
rstrip=rstrip,
umask=umask,
output_loglevel=output_loglevel,
log_callback=log_callback,
quiet=quiet,
timeout=timeout,
reset_system_locale=reset_system_locale,
@ -825,6 +852,7 @@ def run_stdout(cmd,
rstrip=True,
umask=None,
output_loglevel='debug',
log_callback=None,
timeout=None,
reset_system_locale=True,
ignore_retcode=False,
@ -873,6 +901,7 @@ def run_stdout(cmd,
rstrip=rstrip,
umask=umask,
output_loglevel=output_loglevel,
log_callback=log_callback,
timeout=timeout,
reset_system_locale=reset_system_locale,
ignore_retcode=ignore_retcode,
@ -881,19 +910,22 @@ def run_stdout(cmd,
pillar_override=kwargs.get('pillar'),
use_vt=use_vt)
log_callback = _check_cb(log_callback)
lvl = _check_loglevel(output_loglevel)
if lvl is not None:
if not ignore_retcode and ret['retcode'] != 0:
if lvl < LOG_LEVELS['error']:
lvl = LOG_LEVELS['error']
log.error(
msg = (
'Command {0!r} failed with return code: {1}'
.format(cmd, ret['retcode'])
)
log.error(log_callback(msg))
if ret['stdout']:
log.log(lvl, 'stdout: {0}'.format(ret['stdout']))
log.log(lvl, 'stdout: {0}'.format(log_callback(ret['stdout'])))
if ret['stderr']:
log.log(lvl, 'stderr: {0}'.format(ret['stderr']))
log.log(lvl, 'stderr: {0}'.format(log_callback(ret['stderr'])))
if ret['retcode']:
log.log(lvl, 'retcode: {0}'.format(ret['retcode']))
return ret['stdout']
@ -911,6 +943,7 @@ def run_stderr(cmd,
rstrip=True,
umask=None,
output_loglevel='debug',
log_callback=None,
timeout=None,
reset_system_locale=True,
ignore_retcode=False,
@ -959,6 +992,7 @@ def run_stderr(cmd,
rstrip=rstrip,
umask=umask,
output_loglevel=output_loglevel,
log_callback=log_callback,
timeout=timeout,
reset_system_locale=reset_system_locale,
ignore_retcode=ignore_retcode,
@ -967,19 +1001,22 @@ def run_stderr(cmd,
pillarenv=kwargs.get('pillarenv'),
pillar_override=kwargs.get('pillar'))
log_callback = _check_cb(log_callback)
lvl = _check_loglevel(output_loglevel)
if lvl is not None:
if not ignore_retcode and ret['retcode'] != 0:
if lvl < LOG_LEVELS['error']:
lvl = LOG_LEVELS['error']
log.error(
msg = (
'Command {0!r} failed with return code: {1}'
.format(cmd, ret['retcode'])
)
log.error(log_callback(msg))
if ret['stdout']:
log.log(lvl, 'stdout: {0}'.format(ret['stdout']))
log.log(lvl, 'stdout: {0}'.format(log_callback(ret['stdout'])))
if ret['stderr']:
log.log(lvl, 'stderr: {0}'.format(ret['stderr']))
log.log(lvl, 'stderr: {0}'.format(log_callback(ret['stderr'])))
if ret['retcode']:
log.log(lvl, 'retcode: {0}'.format(ret['retcode']))
return ret['stderr']
@ -997,6 +1034,7 @@ def run_all(cmd,
rstrip=True,
umask=None,
output_loglevel='debug',
log_callback=None,
timeout=None,
reset_system_locale=True,
ignore_retcode=False,
@ -1045,6 +1083,7 @@ def run_all(cmd,
rstrip=rstrip,
umask=umask,
output_loglevel=output_loglevel,
log_callback=log_callback,
timeout=timeout,
reset_system_locale=reset_system_locale,
ignore_retcode=ignore_retcode,
@ -1053,19 +1092,22 @@ def run_all(cmd,
pillar_override=kwargs.get('pillar'),
use_vt=use_vt)
log_callback = _check_cb(log_callback)
lvl = _check_loglevel(output_loglevel)
if lvl is not None:
if not ignore_retcode and ret['retcode'] != 0:
if lvl < LOG_LEVELS['error']:
lvl = LOG_LEVELS['error']
log.error(
msg = (
'Command {0!r} failed with return code: {1}'
.format(cmd, ret['retcode'])
)
log.error(log_callback(msg))
if ret['stdout']:
log.log(lvl, 'stdout: {0}'.format(ret['stdout']))
log.log(lvl, 'stdout: {0}'.format(log_callback(ret['stdout'])))
if ret['stderr']:
log.log(lvl, 'stderr: {0}'.format(ret['stderr']))
log.log(lvl, 'stderr: {0}'.format(log_callback(ret['stderr'])))
if ret['retcode']:
log.log(lvl, 'retcode: {0}'.format(ret['retcode']))
return ret
@ -1082,6 +1124,7 @@ def retcode(cmd,
template=None,
umask=None,
output_loglevel='debug',
log_callback=None,
timeout=None,
reset_system_locale=True,
ignore_retcode=False,
@ -1132,6 +1175,7 @@ def retcode(cmd,
template=template,
umask=umask,
output_loglevel=output_loglevel,
log_callback=log_callback,
timeout=timeout,
reset_system_locale=reset_system_locale,
ignore_retcode=ignore_retcode,
@ -1140,16 +1184,19 @@ def retcode(cmd,
pillar_override=kwargs.get('pillar'),
use_vt=use_vt)
log_callback = _check_cb(log_callback)
lvl = _check_loglevel(output_loglevel)
if lvl is not None:
if not ignore_retcode and ret['retcode'] != 0:
if lvl < LOG_LEVELS['error']:
lvl = LOG_LEVELS['error']
log.error(
msg = (
'Command {0!r} failed with return code: {1}'
.format(cmd, ret['retcode'])
)
log.log(lvl, 'output: {0}'.format(ret['stdout']))
log.error(log_callback(msg))
log.log(lvl, 'output: {0}'.format(log_callback(ret['stdout'])))
return ret['retcode']
@ -1164,6 +1211,7 @@ def _retcode_quiet(cmd,
template=None,
umask=None,
output_loglevel='quiet',
log_callback=None,
timeout=None,
reset_system_locale=True,
ignore_retcode=False,
@ -1185,6 +1233,7 @@ def _retcode_quiet(cmd,
template=template,
umask=umask,
output_loglevel=output_loglevel,
log_callback=log_callback,
timeout=timeout,
reset_system_locale=reset_system_locale,
ignore_retcode=ignore_retcode,
@ -1204,6 +1253,7 @@ def script(source,
template=None,
umask=None,
output_loglevel='debug',
log_callback=None,
quiet=False,
timeout=None,
reset_system_locale=True,
@ -1292,6 +1342,7 @@ def script(source,
cwd=cwd,
stdin=stdin,
output_loglevel=output_loglevel,
log_callback=log_callback,
runas=runas,
shell=shell,
python_shell=python_shell,
@ -1322,6 +1373,7 @@ def script_retcode(source,
__env__=None,
saltenv='base',
output_loglevel='debug',
log_callback=None,
use_vt=False,
**kwargs):
'''
@ -1367,6 +1419,7 @@ def script_retcode(source,
__env__=__env__,
saltenv=saltenv,
output_loglevel=output_loglevel,
log_callback=log_callback,
use_vt=use_vt,
**kwargs)['retcode']
@ -1489,6 +1542,7 @@ def run_chroot(root,
rstrip=True,
umask=None,
output_loglevel='quiet',
log_callback=None,
quiet=False,
timeout=None,
reset_system_locale=True,
@ -1551,6 +1605,7 @@ def run_chroot(root,
rstrip=rstrip,
umask=umask,
output_loglevel=output_loglevel,
log_callback=log_callback,
quiet=quiet,
timeout=timeout,
reset_system_locale=reset_system_locale,

166
salt/modules/git.py
View file

@ -7,15 +7,13 @@ from __future__ import absolute_import
# Import python libs
import logging
import os
import re
import subprocess
import sys
# Import salt libs
from salt import utils
import salt.utils
import salt.utils.files
import salt.utils.url
from salt.exceptions import SaltInvocationError, CommandExecutionError
from salt.ext.six.moves.urllib.parse import urlparse as _urlparse # pylint: disable=no-name-in-module,import-error
from salt.ext.six.moves.urllib.parse import urlunparse as _urlunparse # pylint: disable=no-name-in-module,import-error
log = logging.getLogger(__name__)
@ -24,7 +22,7 @@ def __virtual__():
'''
Only load if git exists on the system
'''
return True if utils.which('git') else False
return True if salt.utils.which('git') else False
def _git_run(cmd, cwd=None, runas=None, identity=None, **kwargs):
@ -37,15 +35,6 @@ def _git_run(cmd, cwd=None, runas=None, identity=None, **kwargs):
'''
env = {}
if '<redacted>' in _remove_sensitive_data(cmd):
loglevel = 'quiet'
log.debug(
'HTTPS user/password in git command, the command and '
'output will redacted'
)
else:
loglevel = 'debug'
if identity:
stderrs = []
@ -63,23 +52,26 @@ def _git_run(cmd, cwd=None, runas=None, identity=None, **kwargs):
# copy wrapper to area accessible by ``runas`` user
# currently no suppport in windows for wrapping git ssh
if not utils.is_windows():
ssh_id_wrapper = os.path.join(utils.templates.TEMPLATE_DIRNAME,
'git/ssh-id-wrapper')
tmp_file = utils.mkstemp()
utils.files.copyfile(ssh_id_wrapper, tmp_file)
if not salt.utils.is_windows():
ssh_id_wrapper = os.path.join(
salt.utils.templates.TEMPLATE_DIRNAME,
'git/ssh-id-wrapper'
)
tmp_file = salt.utils.mkstemp()
salt.utils.files.copyfile(ssh_id_wrapper, tmp_file)
os.chmod(tmp_file, 0o500)
os.chown(tmp_file, __salt__['file.user_to_uid'](runas), -1)
env['GIT_SSH'] = tmp_file
try:
result = __salt__['cmd.run_all'](cmd,
cwd=cwd,
runas=runas,
env=env,
python_shell=False,
output_loglevel=loglevel,
**kwargs)
result = __salt__['cmd.run_all'](
cmd,
cwd=cwd,
runas=runas,
env=env,
python_shell=False,
log_callback=salt.utils.url.redact_http_basic_auth,
**kwargs)
finally:
if 'GIT_SSH' in env:
os.remove(env['GIT_SSH'])
@ -88,56 +80,36 @@ def _git_run(cmd, cwd=None, runas=None, identity=None, **kwargs):
if result['retcode'] == 0:
return result['stdout']
else:
stderr = _remove_sensitive_data(result['stderr'])
stderr = \
salt.utils.url.redact_http_basic_auth(result['stderr'])
stderrs.append(stderr)
# we've tried all IDs and still haven't passed, so error out
raise CommandExecutionError("\n\n".join(stderrs))
else:
result = __salt__['cmd.run_all'](cmd,
cwd=cwd,
runas=runas,
env=env,
python_shell=False,
output_loglevel=loglevel,
**kwargs)
result = __salt__['cmd.run_all'](
cmd,
cwd=cwd,
runas=runas,
env=env,
python_shell=False,
log_callback=salt.utils.url.redact_http_basic_auth,
**kwargs)
retcode = result['retcode']
if retcode == 0:
return result['stdout']
else:
stderr = _remove_sensitive_data(result['stderr'])
stderr = salt.utils.url.redact_http_basic_auth(result['stderr'])
raise CommandExecutionError(
'Command {0!r} failed. Stderr: {1!r}'.format(
_remove_sensitive_data(cmd),
salt.utils.url.redact_http_basic_auth(cmd),
stderr
)
)
def _remove_sensitive_data(output):
'''
Remove HTTP user and password
'''
# We can't use re.compile because re.compile(someregex).sub() doesn't
# support flags even in Python 2.7.
url_re = '(https?)://.*@'
redacted = r'\1://<redacted>@'
if sys.version_info >= (2, 7):
# re.sub() supports flags as of 2.7, use this to do a case-insensitive
# match.
return re.sub(url_re, redacted, output, flags=re.IGNORECASE)
else:
# We're on python 2.6, test if a lowercased version of the output
# string matches the regex...
if re.search(url_re, output.lower()):
# ... and if it does, perform the regex substitution.
return re.sub(url_re, redacted, output.lower())
# No match, just return the original string
return output
def _git_getdir(cwd, user=None):
'''
Returns the absolute path to the top-level of a given repo because some Git
@ -157,24 +129,7 @@ def _check_git():
'''
Check if git is available
'''
utils.check_or_die('git')
def _add_http_basic_auth(repository, https_user=None, https_pass=None):
if https_user is None and https_pass is None:
return repository
else:
urltuple = _urlparse(repository)
if urltuple.scheme == 'https':
if https_pass:
auth_string = "{0}:{1}".format(https_user, https_pass)
else:
auth_string = https_user
netloc = "{0}@{1}".format(auth_string, urltuple.netloc)
urltuple = urltuple._replace(netloc=netloc)
return _urlunparse(urltuple)
else:
raise ValueError('Basic Auth only supported for HTTPS scheme')
salt.utils.check_or_die('git')
def current_branch(cwd, user=None):
@ -262,11 +217,17 @@ def clone(cwd, repository, opts=None, user=None, identity=None,
'''
_check_git()
repository = _add_http_basic_auth(repository, https_user, https_pass)
try:
repository = salt.utils.url.add_http_basic_auth(repository,
https_user,
https_pass,
https_only=True)
except ValueError as exc:
raise SaltInvocationError(exc.__str__())
if not opts:
opts = ''
if utils.is_windows():
if salt.utils.is_windows():
cmd = 'git clone {0} {1} {2}'.format(repository, cwd, opts)
else:
cmd = 'git clone {0} {1!r} {2}'.format(repository, cwd, opts)
@ -735,7 +696,7 @@ def push(cwd, remote_name, branch='master', user=None, opts=None,
return _git_run(cmd, cwd=cwd, runas=user, identity=identity)
def remotes(cwd, user=None):
def remotes(cwd, user=None, redact_auth=True):
'''
Get remotes like git remote -v
@ -756,11 +717,14 @@ def remotes(cwd, user=None):
res = dict()
for remote_name in ret.splitlines():
remote = remote_name.strip()
res[remote] = remote_get(cwd, remote, user=user)
res[remote] = remote_get(cwd,
remote,
user=user,
redact_auth=redact_auth)
return res
def remote_get(cwd, remote='origin', user=None):
def remote_get(cwd, remote='origin', user=None, redact_auth=True):
'''
get the fetch and push URL for a specified remote name
@ -770,6 +734,19 @@ def remote_get(cwd, remote='origin', user=None):
user : None
Run git as a user other than what the minion runs as
redact_auth : True
Set to ``False`` to include the username/password if the remote uses
HTTPS Basic Auth. Otherwise, this information will be redacted.
.. warning::
Setting this to ``False`` will not only reveal any HTTPS Basic Auth
that is configured, but the return data will also be written to the
job cache. When possible, it is recommended to use SSH for
authentication.
.. versionadded:: 2015.5.6
CLI Example:
.. code-block:: bash
@ -784,6 +761,11 @@ def remote_get(cwd, remote='origin', user=None):
remote_fetch_url = lines[1].replace('Fetch URL: ', '').strip()
remote_push_url = lines[2].replace('Push URL: ', '').strip()
if remote_fetch_url != remote and remote_push_url != remote:
if redact_auth:
remote_fetch_url = \
salt.utils.url.redact_http_basic_auth(remote_fetch_url)
remote_push_url = \
salt.utils.url.redact_http_basic_auth(remote_push_url)
res = (remote_fetch_url, remote_push_url)
return res
else:
@ -826,8 +808,14 @@ def remote_set(cwd, name='origin', url=None, user=None, https_user=None,
if remote_get(cwd, name):
cmd = 'git remote rm {0}'.format(name)
_git_run(cmd, cwd=cwd, runas=user)
url = _add_http_basic_auth(url, https_user, https_pass)
cmd = 'git remote add {0} {1}'.format(name, url)
try:
url = salt.utils.url.add_http_basic_auth(url,
https_user,
https_pass,
https_only=True)
except ValueError as exc:
raise SaltInvocationError(exc.__str__())
cmd = 'git remote add {0} \'{1}\''.format(name, url)
_git_run(cmd, cwd=cwd, runas=user)
return remote_get(cwd=cwd, remote=name, user=None)
@ -1025,6 +1013,14 @@ def ls_remote(cwd, repository="origin", branch="master", user=None,
'''
_check_git()
repository = _add_http_basic_auth(repository, https_user, https_pass)
try:
repository = salt.utils.url.add_http_basic_auth(repository,
https_user,
https_pass,
https_only=True)
except ValueError as exc:
raise SaltInvocationError(exc.__str__())
cmd = ' '.join(["git", "ls-remote", "-h", "'" + str(repository) + "'", str(branch), "| cut -f 1"])
return _git_run(cmd, cwd=cwd, runas=user, identity=identity)

26
salt/states/git.py
View file

@ -23,6 +23,8 @@ import os.path
import shutil
# Import salt libs
import salt.utils
import salt.utils.url
from salt.exceptions import CommandExecutionError
log = logging.getLogger(__name__)
@ -176,8 +178,22 @@ def latest(name,
if not target:
return _fail(ret, '"target" option is required')
target = os.path.expanduser(target)
try:
desired_fetch_url = salt.utils.url.add_http_basic_auth(
name,
https_user,
https_pass,
https_only=True
)
except ValueError as exc:
return _fail(ret, exc.__str__())
redacted_fetch_url = \
salt.utils.url.redact_http_basic_auth(desired_fetch_url)
run_check_cmd_kwargs = {'runas': user}
if 'shell' in __grains__:
run_check_cmd_kwargs['shell'] = __grains__['shell']
@ -243,8 +259,9 @@ def latest(name,
# check remote if fetch_url not == name set it
remote = __salt__['git.remote_get'](target,
remote=remote_name,
user=user)
if remote is None or remote[0] != name:
user=user,
redact_auth=False)
if remote is None or remote[0] != desired_fetch_url:
__salt__['git.remote_set'](target,
name=remote_name,
url=name,
@ -252,7 +269,10 @@ def latest(name,
https_user=https_user,
https_pass=https_pass)
ret['changes']['remote/{0}'.format(remote_name)] = (
"{0} => {1}".format(str(remote), name)
"{0} => {1}".format(
salt.utils.url.redact_http_basic_auth(str(remote)),
redacted_fetch_url
)
)
# Set to fetch later since we just added the remote and
# need to get the refs

53
salt/utils/url.py
View file

@ -6,6 +6,7 @@ URL utils
# Import python libs
from __future__ import absolute_import
import re
import sys
# Import salt libs
from salt.ext.six.moves.urllib.parse import urlparse, urlunparse # pylint: disable=import-error,no-name-in-module
@ -146,3 +147,55 @@ def strip_proto(url):
was present.
'''
return re.sub('^[^:/]+://', '', url)
def add_http_basic_auth(url,
user=None,
password=None,
https_only=False):
'''
Return a string with http basic auth incorporated into it
'''
if user is None and password is None:
return url
else:
urltuple = urlparse(url)
if https_only and urltuple.scheme != 'https':
raise ValueError('Basic Auth only supported for HTTPS')
if password is None:
netloc = '{0}@{1}'.format(
user,
urltuple.netloc
)
urltuple = urltuple._replace(netloc=netloc)
return urlunparse(urltuple)
else:
netloc = '{0}:{1}@{2}'.format(
user,
password,
urltuple.netloc
)
urltuple = urltuple._replace(netloc=netloc)
return urlunparse(urltuple)
def redact_http_basic_auth(output):
'''
Remove HTTP user and password
'''
# We can't use re.compile because re.compile(someregex).sub() doesn't
# support flags even in Python 2.7.
url_re = '(https?)://.*@'
redacted = r'\1://<redacted>@'
if sys.version_info >= (2, 7):
# re.sub() supports flags as of 2.7, use this to do a case-insensitive
# match.
return re.sub(url_re, redacted, output, flags=re.IGNORECASE)
else:
# We're on python 2.6, test if a lowercased version of the output
# string matches the regex...
if re.search(url_re, output.lower()):
# ... and if it does, perform the regex substitution.
return re.sub(url_re, redacted, output.lower())
# No match, just return the original string
return output

61
tests/unit/modules/git_test.py
View file

@ -1,61 +0,0 @@
# -*- coding: utf-8 -*-
'''
:codeauthor: :email:`Tarjei Husøy <git@thusoy.com>`
'''
# Import Salt Testing Libs
from salttesting import TestCase
from salttesting.helpers import ensure_in_syspath
ensure_in_syspath('../../')
# Import Salt Libs
from salt.modules import git
class GitTestCase(TestCase):
'''
TestCase for salt.modules.git module
'''
def test_http_basic_authentication(self):
'''
Test that HTTP Basic auth works as intended.
'''
# ((user, pass), expected) tuples
test_inputs = [
((None, None), 'https://example.com'),
(('user', None), 'https://user@example.com'),
(('user', 'pass'), 'https://user:pass@example.com'),
]
for (user, password), expected in test_inputs:
kwargs = {
'https_user': user,
'https_pass': password,
'repository': 'https://example.com',
}
result = git._add_http_basic_auth(**kwargs)
self.assertEqual(result, expected)
def test_https_user_and_pw_is_confidential(self):
sensitive_outputs = (
'https://deadbeaf@example.com',
'https://user:pw@example.com',
)
sanitized = 'https://<redacted>@example.com'
for sensitive_output in sensitive_outputs:
result = git._remove_sensitive_data(sensitive_output)
self.assertEqual(result, sanitized)
def test_git_ssh_user_is_not_treated_as_sensitive(self):
not_sensitive_outputs = (
'ssh://user@example.com',
)
for not_sensitive_output in not_sensitive_outputs:
result = git._remove_sensitive_data(not_sensitive_output)
self.assertEqual(result, not_sensitive_output)
if __name__ == '__main__':
from integration import run_tests
run_tests(GitTestCase, needs_daemon=False)

393
tests/unit/utils/url_test.py Normal file
View file

@ -0,0 +1,393 @@
# -*- coding: utf-8 -*-
# Import python libs
from __future__ import absolute_import
# Import Salt Libs
import salt.utils.url
# Import Salt Testing Libs
from salttesting import TestCase, skipIf
from salttesting.helpers import ensure_in_syspath
from salttesting.mock import (
MagicMock,
patch,
NO_MOCK,
NO_MOCK_REASON
)
ensure_in_syspath('../../')
@patch('salt.utils.is_windows', MagicMock(return_value=False))
@skipIf(NO_MOCK, NO_MOCK_REASON)
class UrlTestCase(TestCase):
'''
TestCase for salt.utils.url module
'''
# parse tests
def test_parse_path(self):
'''
Test parsing an ordinary path
'''
path = 'interesting?/path&.conf:and other things'
self.assertEqual(salt.utils.url.parse(path), (path, None))
def test_parse_salt_url(self):
'''
Test parsing a 'salt://' URL
'''
path = '?funny/path with {interesting|chars}'
url = 'salt://' + path
self.assertEqual(salt.utils.url.parse(url), (path, None))
def test_parse_salt_env(self):
'''
Test parsing a 'salt://' URL with an '?env=' query
'''
env = 'milieu'
path = '?funny/path&with {interesting|chars}'
url = 'salt://' + path + '?saltenv=' + env
self.assertEqual(salt.utils.url.parse(url), (path, env))
def test_parse_salt_saltenv(self):
'''
Test parsing a 'salt://' URL with a '?saltenv=' query
'''
saltenv = 'ambience'
path = '?funny/path&with {interesting|chars}'
url = 'salt://' + path + '?saltenv=' + saltenv
self.assertEqual(salt.utils.url.parse(url), (path, saltenv))
# create tests
def test_create_url(self):
'''
Test creating a 'salt://' URL
'''
path = '? interesting/&path.filetype'
url = 'salt://' + path
self.assertEqual(salt.utils.url.create(path), url)
def test_create_url_saltenv(self):
'''
Test creating a 'salt://' URL with a saltenv
'''
saltenv = 'raumklang'
path = '? interesting/&path.filetype'
url = 'salt://' + path + '?saltenv=' + saltenv
self.assertEqual(salt.utils.url.create(path, saltenv), url)
# is_escaped tests
def test_is_escaped_windows(self):
'''
Test not testing a 'salt://' URL on windows
'''
url = 'salt://dir/file.ini'
with patch('salt.utils.is_windows', MagicMock(return_value=True)):
self.assertFalse(salt.utils.url.is_escaped(url))
def test_is_escaped_escaped_path(self):
'''
Test testing an escaped path
'''
path = '|dir/file.conf?saltenv=basic'
self.assertTrue(salt.utils.url.is_escaped(path))
def test_is_escaped_unescaped_path(self):
'''
Test testing an unescaped path
'''
path = 'dir/file.conf'
self.assertFalse(salt.utils.url.is_escaped(path))
def test_is_escaped_escaped_url(self):
'''
Test testing an escaped 'salt://' URL
'''
url = 'salt://|dir/file.conf?saltenv=basic'
self.assertTrue(salt.utils.url.is_escaped(url))
def test_is_escaped_unescaped_url(self):
'''
Test testing an unescaped 'salt://' URL
'''
url = 'salt://dir/file.conf'
self.assertFalse(salt.utils.url.is_escaped(url))
def test_is_escaped_generic_url(self):
'''
Test testing an unescaped 'salt://' URL
'''
url = 'https://gentoo.org/'
self.assertFalse(salt.utils.url.is_escaped(url))
# escape tests
def test_escape_windows(self):
'''
Test not escaping a 'salt://' URL on windows
'''
url = 'salt://dir/file.ini'
with patch('salt.utils.is_windows', MagicMock(return_value=True)):
self.assertEqual(salt.utils.url.escape(url), url)
def test_escape_escaped_path(self):
'''
Test escaping an escaped path
'''
resource = '|dir/file.conf?saltenv=basic'
self.assertEqual(salt.utils.url.escape(resource), resource)
def test_escape_unescaped_path(self):
'''
Test escaping an unescaped path
'''
path = 'dir/file.conf'
escaped_path = '|' + path
self.assertEqual(salt.utils.url.escape(path), escaped_path)
def test_escape_escaped_url(self):
'''
Test testing an escaped 'salt://' URL
'''
url = 'salt://|dir/file.conf?saltenv=basic'
self.assertEqual(salt.utils.url.escape(url), url)
def test_escape_unescaped_url(self):
'''
Test testing an unescaped 'salt://' URL
'''
path = 'dir/file.conf'
url = 'salt://' + path
escaped_url = 'salt://|' + path
self.assertEqual(salt.utils.url.escape(url), escaped_url)
def test_escape_generic_url(self):
'''
Test testing an unescaped 'salt://' URL
'''
url = 'https://gentoo.org/'
self.assertEqual(salt.utils.url.escape(url), url)
# unescape tests
def test_unescape_windows(self):
'''
Test not escaping a 'salt://' URL on windows
'''
url = 'salt://dir/file.ini'
with patch('salt.utils.is_windows', MagicMock(return_value=True)):
self.assertEqual(salt.utils.url.unescape(url), url)
def test_unescape_escaped_path(self):
'''
Test escaping an escaped path
'''
resource = 'dir/file.conf?saltenv=basic'
escaped_path = '|' + resource
self.assertEqual(salt.utils.url.unescape(escaped_path), resource)
def test_unescape_unescaped_path(self):
'''
Test escaping an unescaped path
'''
path = 'dir/file.conf'
self.assertEqual(salt.utils.url.unescape(path), path)
def test_unescape_escaped_url(self):
'''
Test testing an escaped 'salt://' URL
'''
resource = 'dir/file.conf?saltenv=basic'
url = 'salt://' + resource
escaped_url = 'salt://|' + resource
self.assertEqual(salt.utils.url.unescape(escaped_url), url)
def test_unescape_unescaped_url(self):
'''
Test testing an unescaped 'salt://' URL
'''
url = 'salt://dir/file.conf'
self.assertEqual(salt.utils.url.unescape(url), url)
def test_unescape_generic_url(self):
'''
Test testing an unescaped 'salt://' URL
'''
url = 'https://gentoo.org/'
self.assertEqual(salt.utils.url.unescape(url), url)
# add_env tests
def test_add_env_not_salt(self):
'''
Test not adding a saltenv to a non 'salt://' URL
'''
saltenv = 'higgs'
url = 'https://pdg.lbl.gov/'
self.assertEqual(salt.utils.url.add_env(url, saltenv), url)
def test_add_env(self):
'''
Test adding a saltenv to a 'salt://' URL
'''
saltenv = 'erstwhile'
url = 'salt://salted/file.conf'
url_env = url + '?saltenv=' + saltenv
self.assertEqual(salt.utils.url.add_env(url, saltenv), url_env)
# split_env tests
def test_split_env_non_salt(self):
'''
Test not splitting a saltenv from a non 'salt://' URL
'''
saltenv = 'gravitodynamics'
url = 'https://arxiv.org/find/all/?' + saltenv
self.assertEqual(salt.utils.url.split_env(url), (url, None))
def test_split_env(self):
'''
Test splitting a 'salt://' URL
'''
saltenv = 'elsewhere'
url = 'salt://salted/file.conf'
url_env = url + '?saltenv=' + saltenv
self.assertEqual(salt.utils.url.split_env(url_env), (url, saltenv))
# validate tests
def test_validate_valid(self):
'''
Test URL valid validation
'''
url = 'salt://config/file.name?saltenv=vapid'
protos = ['salt', 'pepper', 'cinnamon', 'melange']
self.assertTrue(salt.utils.url.validate(url, protos))
def test_validate_invalid(self):
'''
Test URL invalid validation
'''
url = 'cumin://config/file.name?saltenv=vapid'
protos = ['salt', 'pepper', 'cinnamon', 'melange']
self.assertFalse(salt.utils.url.validate(url, protos))
# strip tests
def test_strip_url_with_scheme(self):
'''
Test stripping of URL scheme
'''
scheme = 'git+salt+rsync+AYB://'
resource = 'all/the/things.stuff;parameter?query=I guess'
url = scheme + resource
self.assertEqual(salt.utils.url.strip_proto(url), resource)
def test_strip_url_without_scheme(self):
'''
Test stripping of a URL without a scheme
'''
resource = 'all/the/things.stuff;parameter?query=I guess'
self.assertEqual(salt.utils.url.strip_proto(resource), resource)
def test_http_basic_auth(self):
'''
Tests that adding basic auth to a URL works as expected
'''
# ((user, password), expected) tuples
test_inputs = (
((None, None), 'http://example.com'),
(('user', None), 'http://user@example.com'),
(('user', 'pass'), 'http://user:pass@example.com'),
)
for (user, password), expected in test_inputs:
kwargs = {
'url': 'http://example.com',
'user': user,
'password': password,
}
# Test http
result = salt.utils.url.add_http_basic_auth(**kwargs)
self.assertEqual(result, expected)
# Test https
kwargs['url'] = kwargs['url'].replace('http://', 'https://', 1)
expected = expected.replace('http://', 'https://', 1)
result = salt.utils.url.add_http_basic_auth(**kwargs)
self.assertEqual(result, expected)
def test_http_basic_auth_https_only(self):
'''
Tests that passing a non-https URL with https_only=True will raise a
ValueError.
'''
kwargs = {
'url': 'http://example.com',
'user': 'foo',
'password': 'bar',
'https_only': True,
}
self.assertRaises(
ValueError,
salt.utils.url.add_http_basic_auth,
**kwargs
)
def test_redact_http_basic_auth(self):
sensitive_outputs = (
'https://deadbeaf@example.com',
'https://user:pw@example.com',
)
sanitized = 'https://<redacted>@example.com'
for sensitive_output in sensitive_outputs:
result = salt.utils.url.redact_http_basic_auth(sensitive_output)
self.assertEqual(result, sanitized)
def test_redact_non_auth_output(self):
non_auth_output = 'This is just normal output'
self.assertEqual(
non_auth_output,
salt.utils.url.redact_http_basic_auth(non_auth_output)
)
if __name__ == '__main__':
from integration import run_tests
run_tests(UrlTestCase, needs_daemon=False)