mirror of
https://github.com/saltstack/salt.git
synced 2025-04-17 10:10:20 +00:00
Refactor publish func in master/masterapi to use check_authentication
In a previous PR, the runner and wheel functions in master.py and masterapi.py were refactored to reduce common code via the salt.auth.check_authentication method. The publish function also can utilize the check_authentication function in master.py and masterapi.py. Consolidation of this code will help us be able to differentiate between authorization and authentication errors in the future.
This commit is contained in:
rallytime
parent
b651ff0534
commit
74acaad46e
5 changed files with 319 additions and 166 deletions
|
|
@ -6,8 +6,6 @@ This system allows for authentication to be managed in a module pluggable way
|
|||
so that any external authentication system can be used inside of Salt
|
||||
'''
|
||||
|
||||
from __future__ import absolute_import
|
||||
|
||||
# 1. Create auth loader instance
|
||||
# 2. Accept arguments as a dict
|
||||
# 3. Verify with function introspection
|
||||
|
|
@ -16,7 +14,7 @@ from __future__ import absolute_import
|
|||
# 6. Interface to verify tokens
|
||||
|
||||
# Import python libs
|
||||
from __future__ import print_function
|
||||
from __future__ import absolute_import, print_function
|
||||
import collections
|
||||
import time
|
||||
import logging
|
||||
|
|
@ -31,6 +29,7 @@ import salt.transport.client
|
|||
import salt.utils.args
|
||||
import salt.utils.dictupdate
|
||||
import salt.utils.files
|
||||
import salt.utils.master
|
||||
import salt.utils.minions
|
||||
import salt.utils.user
|
||||
import salt.utils.versions
|
||||
|
|
@ -430,13 +429,26 @@ class LoadAuth(object):
|
|||
|
||||
auth_list = self.get_auth_list(load)
|
||||
elif auth_type == 'user':
|
||||
if not self.authenticate_key(load, key):
|
||||
auth_ret = self.authenticate_key(load, key)
|
||||
msg = 'Authentication failure of type "user" occurred'
|
||||
if not auth_ret: # auth_ret can be a boolean or the effective user id
|
||||
if show_username:
|
||||
msg = 'Authentication failure of type "user" occurred for user {0}.'.format(username)
|
||||
else:
|
||||
msg = 'Authentication failure of type "user" occurred'
|
||||
msg = '{0} for user {1}.'.format(msg, username)
|
||||
ret['error'] = {'name': 'UserAuthenticationError', 'message': msg}
|
||||
return ret
|
||||
|
||||
# Verify that the caller has root on master
|
||||
if auth_ret is not True:
|
||||
if AuthUser(load['user']).is_sudo():
|
||||
if not self.opts['sudo_acl'] or not self.opts['publisher_acl']:
|
||||
auth_ret = True
|
||||
|
||||
if auth_ret is not True:
|
||||
auth_list = salt.utils.master.get_values_of_matching_keys(
|
||||
self.opts['publisher_acl'], auth_ret)
|
||||
if not auth_list:
|
||||
ret['error'] = {'name': 'UserAuthenticationError', 'message': msg}
|
||||
return ret
|
||||
else:
|
||||
ret['error'] = {'name': 'SaltInvocationError',
|
||||
'message': 'Authentication type not supported.'}
|
||||
|
|
|
|||
|
|
@ -14,6 +14,7 @@ import time
|
|||
import stat
|
||||
|
||||
# Import salt libs
|
||||
import salt.acl
|
||||
import salt.crypt
|
||||
import salt.cache
|
||||
import salt.client
|
||||
|
|
@ -1176,88 +1177,50 @@ class LocalFuncs(object):
|
|||
)
|
||||
minions = _res['minions']
|
||||
|
||||
# Check for external auth calls
|
||||
if extra.get('token', False):
|
||||
# Authenticate
|
||||
token = self.loadauth.authenticate_token(extra)
|
||||
if not token:
|
||||
return ''
|
||||
|
||||
# Get acl from eauth module.
|
||||
auth_list = self.loadauth.get_auth_list(extra, token)
|
||||
|
||||
# Authorize the request
|
||||
if not self.ckminions.auth_check(
|
||||
auth_list,
|
||||
load['fun'],
|
||||
load['arg'],
|
||||
load['tgt'],
|
||||
load.get('tgt_type', 'glob'),
|
||||
minions=minions,
|
||||
# always accept find_job
|
||||
whitelist=['saltutil.find_job'],
|
||||
):
|
||||
log.warning('Authentication failure of type "token" occurred.')
|
||||
return ''
|
||||
load['user'] = token['name']
|
||||
log.debug('Minion tokenized user = "{0}"'.format(load['user']))
|
||||
elif 'eauth' in extra:
|
||||
# Authenticate.
|
||||
if not self.loadauth.authenticate_eauth(extra):
|
||||
return ''
|
||||
|
||||
# Get acl from eauth module.
|
||||
auth_list = self.loadauth.get_auth_list(extra)
|
||||
|
||||
# Authorize the request
|
||||
if not self.ckminions.auth_check(
|
||||
auth_list,
|
||||
load['fun'],
|
||||
load['arg'],
|
||||
load['tgt'],
|
||||
load.get('tgt_type', 'glob'),
|
||||
minions=minions,
|
||||
# always accept find_job
|
||||
whitelist=['saltutil.find_job'],
|
||||
):
|
||||
log.warning('Authentication failure of type "eauth" occurred.')
|
||||
return ''
|
||||
load['user'] = self.loadauth.load_name(extra) # The username we are attempting to auth with
|
||||
# Verify that the caller has root on master
|
||||
# Check for external auth calls and authenticate
|
||||
auth_type, err_name, key = self._prep_auth_info(extra)
|
||||
if auth_type == 'user':
|
||||
auth_check = self.loadauth.check_authentication(load, auth_type, key=key)
|
||||
else:
|
||||
auth_ret = self.loadauth.authenticate_key(load, self.key)
|
||||
if auth_ret is False:
|
||||
auth_check = self.loadauth.check_authentication(extra, auth_type)
|
||||
|
||||
# Setup authorization list variable and error information
|
||||
auth_list = auth_check.get('auth_list', [])
|
||||
error = auth_check.get('error')
|
||||
err_msg = 'Authentication failure of type "{0}" occurred.'.format(auth_type)
|
||||
|
||||
if error:
|
||||
# Authentication error occurred: do not continue.
|
||||
log.warning(err_msg)
|
||||
return ''
|
||||
|
||||
# All Token, Eauth, and non-root users must pass the authorization check
|
||||
if auth_type != 'user' or (auth_type == 'user' and auth_list):
|
||||
# Authorize the request
|
||||
authorized = self.ckminions.auth_check(
|
||||
auth_list,
|
||||
load['fun'],
|
||||
load['arg'],
|
||||
load['tgt'],
|
||||
load.get('tgt_type', 'glob'),
|
||||
minions=minions,
|
||||
# always accept find_job
|
||||
whitelist=['saltutil.find_job'],
|
||||
)
|
||||
|
||||
if not authorized:
|
||||
# Authorization error occurred. Log warning and do not continue.
|
||||
log.warning(err_msg)
|
||||
return ''
|
||||
|
||||
if auth_ret is not True:
|
||||
if salt.auth.AuthUser(load['user']).is_sudo():
|
||||
if not self.opts['sudo_acl'] or not self.opts['publisher_acl']:
|
||||
auth_ret = True
|
||||
|
||||
if auth_ret is not True:
|
||||
# Avoid circular import
|
||||
import salt.utils.master
|
||||
auth_list = salt.utils.master.get_values_of_matching_keys(
|
||||
self.opts['publisher_acl'],
|
||||
auth_ret)
|
||||
if not auth_list:
|
||||
log.warning(
|
||||
'Authentication failure of type "user" occurred.'
|
||||
)
|
||||
return ''
|
||||
|
||||
if not self.ckminions.auth_check(
|
||||
auth_list,
|
||||
load['fun'],
|
||||
load['arg'],
|
||||
load['tgt'],
|
||||
load.get('tgt_type', 'glob'),
|
||||
minions=minions,
|
||||
# always accept find_job
|
||||
whitelist=['saltutil.find_job'],
|
||||
):
|
||||
log.warning('Authentication failure of type "user" occurred.')
|
||||
return ''
|
||||
# Perform some specific auth_type tasks after the authorization check
|
||||
if auth_type == 'token':
|
||||
username = auth_check.get('username')
|
||||
load['user'] = username
|
||||
log.debug('Minion tokenized user = "{0}"'.format(username))
|
||||
elif auth_type == 'eauth':
|
||||
# The username we are attempting to auth with
|
||||
load['user'] = self.loadauth.load_name(extra)
|
||||
|
||||
# If we order masters (via a syndic), don't short circuit if no minions
|
||||
# are found
|
||||
|
|
|
|||
121
salt/master.py
121
salt/master.py
|
|
@ -1840,89 +1840,52 @@ class ClearFuncs(object):
|
|||
clear_load.get(u'tgt_type', u'glob'),
|
||||
delimiter
|
||||
)
|
||||
minions = _res.get('minions', list())
|
||||
missing = _res.get('missing', list())
|
||||
minions = _res.get(u'minions', list())
|
||||
missing = _res.get(u'missing', list())
|
||||
|
||||
# Check for external auth calls
|
||||
if extra.get(u'token', False):
|
||||
# Authenticate.
|
||||
token = self.loadauth.authenticate_token(extra)
|
||||
if not token:
|
||||
return u''
|
||||
|
||||
# Get acl
|
||||
auth_list = self.loadauth.get_auth_list(extra, token)
|
||||
|
||||
# Authorize the request
|
||||
if not self.ckminions.auth_check(
|
||||
auth_list,
|
||||
clear_load[u'fun'],
|
||||
clear_load[u'arg'],
|
||||
clear_load[u'tgt'],
|
||||
clear_load.get(u'tgt_type', u'glob'),
|
||||
minions=minions,
|
||||
# always accept find_job
|
||||
whitelist=[u'saltutil.find_job'],
|
||||
):
|
||||
log.warning(u'Authentication failure of type "token" occurred.')
|
||||
return u''
|
||||
clear_load[u'user'] = token[u'name']
|
||||
log.debug(u'Minion tokenized user = "%s"', clear_load[u'user'])
|
||||
elif u'eauth' in extra:
|
||||
# Authenticate.
|
||||
if not self.loadauth.authenticate_eauth(extra):
|
||||
return u''
|
||||
|
||||
# Get acl from eauth module.
|
||||
auth_list = self.loadauth.get_auth_list(extra)
|
||||
|
||||
# Authorize the request
|
||||
if not self.ckminions.auth_check(
|
||||
auth_list,
|
||||
clear_load[u'fun'],
|
||||
clear_load[u'arg'],
|
||||
clear_load[u'tgt'],
|
||||
clear_load.get(u'tgt_type', u'glob'),
|
||||
minions=minions,
|
||||
# always accept find_job
|
||||
whitelist=[u'saltutil.find_job'],
|
||||
):
|
||||
log.warning(u'Authentication failure of type "eauth" occurred.')
|
||||
return u''
|
||||
clear_load[u'user'] = self.loadauth.load_name(extra) # The username we are attempting to auth with
|
||||
# Verify that the caller has root on master
|
||||
# Check for external auth calls and authenticate
|
||||
auth_type, err_name, key, sensitive_load_keys = self._prep_auth_info(extra)
|
||||
if auth_type == 'user':
|
||||
auth_check = self.loadauth.check_authentication(clear_load, auth_type, key=key)
|
||||
else:
|
||||
auth_ret = self.loadauth.authenticate_key(clear_load, self.key)
|
||||
if auth_ret is False:
|
||||
auth_check = self.loadauth.check_authentication(extra, auth_type)
|
||||
|
||||
# Setup authorization list variable and error information
|
||||
auth_list = auth_check.get(u'auth_list', [])
|
||||
err_msg = u'Authentication failure of type "{0}" occurred.'.format(auth_type)
|
||||
|
||||
if auth_check.get(u'error'):
|
||||
# Authentication error occurred: do not continue.
|
||||
log.warning(err_msg)
|
||||
return u''
|
||||
|
||||
# All Token, Eauth, and non-root users must pass the authorization check
|
||||
if auth_type != u'user' or (auth_type == u'user' and auth_list):
|
||||
# Authorize the request
|
||||
authorized = self.ckminions.auth_check(
|
||||
auth_list,
|
||||
clear_load[u'fun'],
|
||||
clear_load[u'arg'],
|
||||
clear_load[u'tgt'],
|
||||
clear_load.get(u'tgt_type', u'glob'),
|
||||
minions=minions,
|
||||
# always accept find_job
|
||||
whitelist=[u'saltutil.find_job'],
|
||||
)
|
||||
|
||||
if not authorized:
|
||||
# Authorization error occurred. Do not continue.
|
||||
log.warning(err_msg)
|
||||
return u''
|
||||
|
||||
if auth_ret is not True:
|
||||
if salt.auth.AuthUser(clear_load[u'user']).is_sudo():
|
||||
if not self.opts[u'sudo_acl'] or not self.opts[u'publisher_acl']:
|
||||
auth_ret = True
|
||||
|
||||
if auth_ret is not True:
|
||||
auth_list = salt.utils.master.get_values_of_matching_keys(
|
||||
self.opts[u'publisher_acl'],
|
||||
auth_ret)
|
||||
if not auth_list:
|
||||
log.warning(
|
||||
u'Authentication failure of type "user" occurred.'
|
||||
)
|
||||
return u''
|
||||
|
||||
if not self.ckminions.auth_check(
|
||||
auth_list,
|
||||
clear_load[u'fun'],
|
||||
clear_load[u'arg'],
|
||||
clear_load[u'tgt'],
|
||||
clear_load.get(u'tgt_type', u'glob'),
|
||||
minions=minions,
|
||||
# always accept find_job
|
||||
whitelist=[u'saltutil.find_job'],
|
||||
):
|
||||
log.warning(u'Authentication failure of type "user" occurred.')
|
||||
return u''
|
||||
# Perform some specific auth_type tasks after the authorization check
|
||||
if auth_type == u'token':
|
||||
username = auth_check.get(u'username')
|
||||
clear_load[u'user'] = username
|
||||
log.debug(u'Minion tokenized user = "%s"', username)
|
||||
elif auth_type == u'eauth':
|
||||
# The username we are attempting to auth with
|
||||
clear_load[u'user'] = self.loadauth.load_name(extra)
|
||||
|
||||
# If we order masters (via a syndic), don't short circuit if no minions
|
||||
# are found
|
||||
|
|
|
|||
|
|
@ -8,13 +8,16 @@ import salt.config
|
|||
import salt.daemons.masterapi as masterapi
|
||||
|
||||
# Import Salt Testing Libs
|
||||
from tests.support.unit import TestCase
|
||||
from tests.support.unit import TestCase, skipIf
|
||||
from tests.support.mock import (
|
||||
patch,
|
||||
MagicMock,
|
||||
NO_MOCK,
|
||||
NO_MOCK_REASON
|
||||
)
|
||||
|
||||
|
||||
@skipIf(NO_MOCK, NO_MOCK_REASON)
|
||||
class LocalFuncsTestCase(TestCase):
|
||||
'''
|
||||
TestCase for salt.daemons.masterapi.LocalFuncs class
|
||||
|
|
@ -24,6 +27,8 @@ class LocalFuncsTestCase(TestCase):
|
|||
opts = salt.config.master_config(None)
|
||||
self.local_funcs = masterapi.LocalFuncs(opts, 'test-key')
|
||||
|
||||
# runner tests
|
||||
|
||||
def test_runner_token_not_authenticated(self):
|
||||
'''
|
||||
Asserts that a TokenAuthenticationError is returned when the token can't authenticate.
|
||||
|
|
@ -107,6 +112,8 @@ class LocalFuncsTestCase(TestCase):
|
|||
|
||||
self.assertDictEqual(mock_ret, ret)
|
||||
|
||||
# wheel tests
|
||||
|
||||
def test_wheel_token_not_authenticated(self):
|
||||
'''
|
||||
Asserts that a TokenAuthenticationError is returned when the token can't authenticate.
|
||||
|
|
@ -199,3 +206,105 @@ class LocalFuncsTestCase(TestCase):
|
|||
u'user UNKNOWN.'}}
|
||||
ret = self.local_funcs.wheel({})
|
||||
self.assertDictEqual(mock_ret, ret)
|
||||
|
||||
# publish tests
|
||||
|
||||
def test_publish_user_is_blacklisted(self):
|
||||
'''
|
||||
Asserts that an empty string is returned when the user has been blacklisted.
|
||||
'''
|
||||
with patch('salt.acl.PublisherACL.user_is_blacklisted', MagicMock(return_value=True)):
|
||||
self.assertEqual(u'', self.local_funcs.publish({u'user': u'foo', u'fun': u'test.arg'}))
|
||||
|
||||
def test_publish_cmd_blacklisted(self):
|
||||
'''
|
||||
Asserts that an empty string returned when the command has been blacklisted.
|
||||
'''
|
||||
with patch('salt.acl.PublisherACL.user_is_blacklisted', MagicMock(return_value=False)), \
|
||||
patch('salt.acl.PublisherACL.cmd_is_blacklisted', MagicMock(return_value=True)):
|
||||
self.assertEqual(u'', self.local_funcs.publish({u'user': u'foo', u'fun': u'test.arg'}))
|
||||
|
||||
def test_publish_token_not_authenticated(self):
|
||||
'''
|
||||
Asserts that an empty string is returned when the token can't authenticate.
|
||||
'''
|
||||
load = {u'user': u'foo', u'fun': u'test.arg', u'tgt': u'test_minion',
|
||||
u'kwargs': {u'token': u'asdfasdfasdfasdf'}}
|
||||
with patch('salt.acl.PublisherACL.user_is_blacklisted', MagicMock(return_value=False)), \
|
||||
patch('salt.acl.PublisherACL.cmd_is_blacklisted', MagicMock(return_value=False)):
|
||||
self.assertEqual(u'', self.local_funcs.publish(load))
|
||||
|
||||
def test_publish_token_authorization_error(self):
|
||||
'''
|
||||
Asserts that an empty string is returned when the token authenticates, but is not
|
||||
authorized.
|
||||
'''
|
||||
token = u'asdfasdfasdfasdf'
|
||||
load = {u'user': u'foo', u'fun': u'test.arg', u'tgt': u'test_minion',
|
||||
u'arg': u'bar', u'kwargs': {u'token': token}}
|
||||
mock_token = {u'token': token, u'eauth': u'foo', u'name': u'test'}
|
||||
|
||||
with patch('salt.acl.PublisherACL.user_is_blacklisted', MagicMock(return_value=False)), \
|
||||
patch('salt.acl.PublisherACL.cmd_is_blacklisted', MagicMock(return_value=False)), \
|
||||
patch('salt.auth.LoadAuth.authenticate_token', MagicMock(return_value=mock_token)), \
|
||||
patch('salt.auth.LoadAuth.get_auth_list', MagicMock(return_value=[])):
|
||||
self.assertEqual(u'', self.local_funcs.publish(load))
|
||||
|
||||
def test_publish_eauth_not_authenticated(self):
|
||||
'''
|
||||
Asserts that an empty string is returned when the user can't authenticate.
|
||||
'''
|
||||
load = {u'user': u'test', u'fun': u'test.arg', u'tgt': u'test_minion',
|
||||
u'kwargs': {u'eauth': u'foo'}}
|
||||
with patch('salt.acl.PublisherACL.user_is_blacklisted', MagicMock(return_value=False)), \
|
||||
patch('salt.acl.PublisherACL.cmd_is_blacklisted', MagicMock(return_value=False)):
|
||||
self.assertEqual(u'', self.local_funcs.publish(load))
|
||||
|
||||
def test_publish_eauth_authorization_error(self):
|
||||
'''
|
||||
Asserts that an empty string is returned when the user authenticates, but is not
|
||||
authorized.
|
||||
'''
|
||||
load = {u'user': u'test', u'fun': u'test.arg', u'tgt': u'test_minion',
|
||||
u'kwargs': {u'eauth': u'foo'}, u'arg': u'bar'}
|
||||
with patch('salt.acl.PublisherACL.user_is_blacklisted', MagicMock(return_value=False)), \
|
||||
patch('salt.acl.PublisherACL.cmd_is_blacklisted', MagicMock(return_value=False)), \
|
||||
patch('salt.auth.LoadAuth.authenticate_eauth', MagicMock(return_value=True)), \
|
||||
patch('salt.auth.LoadAuth.get_auth_list', MagicMock(return_value=[])):
|
||||
self.assertEqual(u'', self.local_funcs.publish(load))
|
||||
|
||||
def test_publish_user_not_authenticated(self):
|
||||
'''
|
||||
Asserts that an empty string is returned when the user can't authenticate.
|
||||
'''
|
||||
load = {u'user': u'test', u'fun': u'test.arg', u'tgt': u'test_minion'}
|
||||
with patch('salt.acl.PublisherACL.user_is_blacklisted', MagicMock(return_value=False)), \
|
||||
patch('salt.acl.PublisherACL.cmd_is_blacklisted', MagicMock(return_value=False)):
|
||||
self.assertEqual(u'', self.local_funcs.publish(load))
|
||||
|
||||
def test_publish_user_authenticated_missing_auth_list(self):
|
||||
'''
|
||||
Asserts that an empty string is returned when the user has an effective user id and is
|
||||
authenticated, but the auth_list is empty.
|
||||
'''
|
||||
load = {u'user': u'test', u'fun': u'test.arg', u'tgt': u'test_minion',
|
||||
u'kwargs': {u'user': u'test'}, u'arg': u'foo'}
|
||||
with patch('salt.acl.PublisherACL.user_is_blacklisted', MagicMock(return_value=False)), \
|
||||
patch('salt.acl.PublisherACL.cmd_is_blacklisted', MagicMock(return_value=False)), \
|
||||
patch('salt.auth.LoadAuth.authenticate_key', MagicMock(return_value='fake-user-key')), \
|
||||
patch('salt.utils.master.get_values_of_matching_keys', MagicMock(return_value=[])):
|
||||
self.assertEqual(u'', self.local_funcs.publish(load))
|
||||
|
||||
def test_publish_user_authorization_error(self):
|
||||
'''
|
||||
Asserts that an empty string is returned when the user authenticates, but is not
|
||||
authorized.
|
||||
'''
|
||||
load = {u'user': u'test', u'fun': u'test.arg', u'tgt': u'test_minion',
|
||||
u'kwargs': {u'user': u'test'}, u'arg': u'foo'}
|
||||
with patch('salt.acl.PublisherACL.user_is_blacklisted', MagicMock(return_value=False)), \
|
||||
patch('salt.acl.PublisherACL.cmd_is_blacklisted', MagicMock(return_value=False)), \
|
||||
patch('salt.auth.LoadAuth.authenticate_key', MagicMock(return_value='fake-user-key')), \
|
||||
patch('salt.utils.master.get_values_of_matching_keys', MagicMock(return_value=['test'])), \
|
||||
patch('salt.utils.minions.CkMinions.auth_check', MagicMock(return_value=False)):
|
||||
self.assertEqual(u'', self.local_funcs.publish(load))
|
||||
|
|
|
|||
|
|
@ -24,6 +24,8 @@ class ClearFuncsTestCase(TestCase):
|
|||
opts = salt.config.master_config(None)
|
||||
self.clear_funcs = salt.master.ClearFuncs(opts, {})
|
||||
|
||||
# runner tests
|
||||
|
||||
def test_runner_token_not_authenticated(self):
|
||||
'''
|
||||
Asserts that a TokenAuthenticationError is returned when the token can't authenticate.
|
||||
|
|
@ -116,6 +118,8 @@ class ClearFuncsTestCase(TestCase):
|
|||
ret = self.clear_funcs.runner({})
|
||||
self.assertDictEqual(mock_ret, ret)
|
||||
|
||||
# wheel tests
|
||||
|
||||
def test_wheel_token_not_authenticated(self):
|
||||
'''
|
||||
Asserts that a TokenAuthenticationError is returned when the token can't authenticate.
|
||||
|
|
@ -207,3 +211,105 @@ class ClearFuncsTestCase(TestCase):
|
|||
u'message': u'Authentication failure of type "user" occurred'}}
|
||||
ret = self.clear_funcs.wheel({})
|
||||
self.assertDictEqual(mock_ret, ret)
|
||||
|
||||
# publish tests
|
||||
|
||||
def test_publish_user_is_blacklisted(self):
|
||||
'''
|
||||
Asserts that an empty string is returned when the user has been blacklisted.
|
||||
'''
|
||||
with patch('salt.acl.PublisherACL.user_is_blacklisted', MagicMock(return_value=True)):
|
||||
self.assertEqual(u'', self.clear_funcs.publish({u'user': u'foo', u'fun': u'test.arg'}))
|
||||
|
||||
def test_publish_cmd_blacklisted(self):
|
||||
'''
|
||||
Asserts that an empty string returned when the command has been blacklisted.
|
||||
'''
|
||||
with patch('salt.acl.PublisherACL.user_is_blacklisted', MagicMock(return_value=False)), \
|
||||
patch('salt.acl.PublisherACL.cmd_is_blacklisted', MagicMock(return_value=True)):
|
||||
self.assertEqual(u'', self.clear_funcs.publish({u'user': u'foo', u'fun': u'test.arg'}))
|
||||
|
||||
def test_publish_token_not_authenticated(self):
|
||||
'''
|
||||
Asserts that an empty string is returned when the token can't authenticate.
|
||||
'''
|
||||
load = {u'user': u'foo', u'fun': u'test.arg', u'tgt': u'test_minion',
|
||||
u'kwargs': {u'token': u'asdfasdfasdfasdf'}}
|
||||
with patch('salt.acl.PublisherACL.user_is_blacklisted', MagicMock(return_value=False)), \
|
||||
patch('salt.acl.PublisherACL.cmd_is_blacklisted', MagicMock(return_value=False)):
|
||||
self.assertEqual(u'', self.clear_funcs.publish(load))
|
||||
|
||||
def test_publish_token_authorization_error(self):
|
||||
'''
|
||||
Asserts that an empty string is returned when the token authenticates, but is not
|
||||
authorized.
|
||||
'''
|
||||
token = u'asdfasdfasdfasdf'
|
||||
load = {u'user': u'foo', u'fun': u'test.arg', u'tgt': u'test_minion',
|
||||
u'arg': u'bar', u'kwargs': {u'token': token}}
|
||||
mock_token = {u'token': token, u'eauth': u'foo', u'name': u'test'}
|
||||
|
||||
with patch('salt.acl.PublisherACL.user_is_blacklisted', MagicMock(return_value=False)), \
|
||||
patch('salt.acl.PublisherACL.cmd_is_blacklisted', MagicMock(return_value=False)), \
|
||||
patch('salt.auth.LoadAuth.authenticate_token', MagicMock(return_value=mock_token)), \
|
||||
patch('salt.auth.LoadAuth.get_auth_list', MagicMock(return_value=[])):
|
||||
self.assertEqual(u'', self.clear_funcs.publish(load))
|
||||
|
||||
def test_publish_eauth_not_authenticated(self):
|
||||
'''
|
||||
Asserts that an empty string is returned when the user can't authenticate.
|
||||
'''
|
||||
load = {u'user': u'test', u'fun': u'test.arg', u'tgt': u'test_minion',
|
||||
u'kwargs': {u'eauth': u'foo'}}
|
||||
with patch('salt.acl.PublisherACL.user_is_blacklisted', MagicMock(return_value=False)), \
|
||||
patch('salt.acl.PublisherACL.cmd_is_blacklisted', MagicMock(return_value=False)):
|
||||
self.assertEqual(u'', self.clear_funcs.publish(load))
|
||||
|
||||
def test_publish_eauth_authorization_error(self):
|
||||
'''
|
||||
Asserts that an empty string is returned when the user authenticates, but is not
|
||||
authorized.
|
||||
'''
|
||||
load = {u'user': u'test', u'fun': u'test.arg', u'tgt': u'test_minion',
|
||||
u'kwargs': {u'eauth': u'foo'}, u'arg': u'bar'}
|
||||
with patch('salt.acl.PublisherACL.user_is_blacklisted', MagicMock(return_value=False)), \
|
||||
patch('salt.acl.PublisherACL.cmd_is_blacklisted', MagicMock(return_value=False)), \
|
||||
patch('salt.auth.LoadAuth.authenticate_eauth', MagicMock(return_value=True)), \
|
||||
patch('salt.auth.LoadAuth.get_auth_list', MagicMock(return_value=[])):
|
||||
self.assertEqual(u'', self.clear_funcs.publish(load))
|
||||
|
||||
def test_publish_user_not_authenticated(self):
|
||||
'''
|
||||
Asserts that an empty string is returned when the user can't authenticate.
|
||||
'''
|
||||
load = {u'user': u'test', u'fun': u'test.arg', u'tgt': u'test_minion'}
|
||||
with patch('salt.acl.PublisherACL.user_is_blacklisted', MagicMock(return_value=False)), \
|
||||
patch('salt.acl.PublisherACL.cmd_is_blacklisted', MagicMock(return_value=False)):
|
||||
self.assertEqual(u'', self.clear_funcs.publish(load))
|
||||
|
||||
def test_publish_user_authenticated_missing_auth_list(self):
|
||||
'''
|
||||
Asserts that an empty string is returned when the user has an effective user id and is
|
||||
authenticated, but the auth_list is empty.
|
||||
'''
|
||||
load = {u'user': u'test', u'fun': u'test.arg', u'tgt': u'test_minion',
|
||||
u'kwargs': {u'user': u'test'}, u'arg': u'foo'}
|
||||
with patch('salt.acl.PublisherACL.user_is_blacklisted', MagicMock(return_value=False)), \
|
||||
patch('salt.acl.PublisherACL.cmd_is_blacklisted', MagicMock(return_value=False)), \
|
||||
patch('salt.auth.LoadAuth.authenticate_key', MagicMock(return_value='fake-user-key')), \
|
||||
patch('salt.utils.master.get_values_of_matching_keys', MagicMock(return_value=[])):
|
||||
self.assertEqual(u'', self.clear_funcs.publish(load))
|
||||
|
||||
def test_publish_user_authorization_error(self):
|
||||
'''
|
||||
Asserts that an empty string is returned when the user authenticates, but is not
|
||||
authorized.
|
||||
'''
|
||||
load = {u'user': u'test', u'fun': u'test.arg', u'tgt': u'test_minion',
|
||||
u'kwargs': {u'user': u'test'}, u'arg': u'foo'}
|
||||
with patch('salt.acl.PublisherACL.user_is_blacklisted', MagicMock(return_value=False)), \
|
||||
patch('salt.acl.PublisherACL.cmd_is_blacklisted', MagicMock(return_value=False)), \
|
||||
patch('salt.auth.LoadAuth.authenticate_key', MagicMock(return_value='fake-user-key')), \
|
||||
patch('salt.utils.master.get_values_of_matching_keys', MagicMock(return_value=['test'])), \
|
||||
patch('salt.utils.minions.CkMinions.auth_check', MagicMock(return_value=False)):
|
||||
self.assertEqual(u'', self.clear_funcs.publish(load))
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue