cmueller/ansible
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

iosxr_user refactor for cliconf and netconf (#34892)

Browse source 
* * refactor iosxr_user for cliconf and netconf

(cherry picked from commit 5d0994ef598f1601fca00a0c1eff4ebb05ebbf1b)

* * Purge and units test changes
This commit is contained in:
Kedar Kekan 2018-01-17 19:58:58 +05:30 committed by Chris Alfonso
parent db61f8f967
commit 0f692f1fe7
9 changed files with 669 additions and 274 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

18
lib/ansible/module_utils/network/iosxr/iosxr.py
View file

@ -67,6 +67,8 @@ NS_DICT = {
'INTERFACE-PROPERTIES_NSMAP': {None: "http://cisco.com/ns/yang/Cisco-IOS-XR-ifmgr-oper"},
'IP-DOMAIN_NSMAP': {None: "http://cisco.com/ns/yang/Cisco-IOS-XR-ip-domain-cfg"},
'SYSLOG_NSMAP': {None: "http://cisco.com/ns/yang/Cisco-IOS-XR-infra-syslog-cfg"},
'AAA_NSMAP': {None: "http://cisco.com/ns/yang/Cisco-IOS-XR-aaa-lib-cfg"},
'AAA_LOCALD_NSMAP': {None: "http://cisco.com/ns/yang/Cisco-IOS-XR-aaa-locald-cfg"},
}
iosxr_provider_spec = {
@ -148,21 +150,21 @@ def build_xml_subtree(container_ele, xmap, param=None, opcode=None):
if ((opcode in ('delete', 'merge') and meta.get('operation', 'unknown') == 'edit') or
meta.get('operation', None) is None):
if meta.get('tag', False):
if meta.get('tag', False) is True:
if parent.tag == container_ele.tag:
if meta.get('ns', None) is True:
if meta.get('ns', False) is True:
child = etree.Element(candidates[-1], nsmap=NS_DICT[key.upper() + "_NSMAP"])
else:
child = etree.Element(candidates[-1])
meta_subtree.append(child)
sub_root = child
else:
if meta.get('ns', None) is True:
if meta.get('ns', False) is True:
child = etree.SubElement(parent, candidates[-1], nsmap=NS_DICT[key.upper() + "_NSMAP"])
else:
child = etree.SubElement(parent, candidates[-1])
if meta.get('attrib', None) and opcode in ('delete', 'merge'):
if meta.get('attrib', None) is not None and opcode in ('delete', 'merge'):
child.set(BASE_1_0 + meta.get('attrib'), opcode)
continue
@ -170,20 +172,20 @@ def build_xml_subtree(container_ele, xmap, param=None, opcode=None):
text = None
param_key = key.split(":")
if param_key[0] == 'a':
if param.get(param_key[1], None):
if param is not None and param.get(param_key[1], None) is not None:
text = param.get(param_key[1])
elif param_key[0] == 'm':
if meta.get('value', None):
if meta.get('value', None) is not None:
text = meta.get('value')
if text:
if meta.get('ns', None) is True:
if meta.get('ns', False) is True:
child = etree.SubElement(parent, candidates[-1], nsmap=NS_DICT[key.upper() + "_NSMAP"])
else:
child = etree.SubElement(parent, candidates[-1])
child.text = text
if meta.get('attrib', None) and opcode in ('delete', 'merge'):
if meta.get('attrib', None) is not None and opcode in ('delete', 'merge'):
child.set(BASE_1_0 + meta.get('attrib'), opcode)
if len(meta_subtree) > 1:

698
lib/ansible/modules/network/iosxr/iosxr_user.py
View file

@ -19,6 +19,7 @@ version_added: "2.4"
author:
- "Trishna Guha (@trishnaguha)"
- "Sebastiaan van Doesselaar (@sebasdoes)"
- "Kedar Kekan (@kedarX)"
short_description: Manage the aggregate of local users on Cisco IOS XR device
description:
- This module provides declarative management of the local usernames
@ -28,7 +29,7 @@ description:
configuration that are not explicitly defined.
extends_documentation_fragment: iosxr
notes:
- Tested against IOS XR 6.1.2
- Tested against IOS XRv 6.1.2
options:
aggregate:
description:
@ -45,9 +46,10 @@ options:
Please note that this option is not same as C(provider username).
configured_password:
description:
- The password to be configured on the Cisco IOS XR device. The
password needs to be provided in clear and it will be encrypted
on the device.
- The password to be configured on the Cisco IOS XR device. The password
needs to be provided in clear text. Password is encrypted on the device
when used with I(cli) and by Ansible when used with I(netconf)
using the same MD5 hash technique with salt size of 3.
Please note that this option is not same as C(provider password).
update_password:
description:
@ -78,7 +80,7 @@ options:
- Instructs the module to consider the
resource definition absolute. It will remove any previously
configured usernames on the device with the exception of the
`admin` user (the current defined set of users).
`admin` user and the current defined set of users.
type: bool
default: false
state:
@ -119,11 +121,11 @@ EXAMPLES = """
- name: create a new user
iosxr_user:
name: ansible
configured_password: test
configured_password: mypassword
state: present
- name: remove all users except admin
iosxr_user:
purge: yes
purge: True
- name: set multiple users to group sys-admin
iosxr_user:
aggregate:
@ -161,15 +163,38 @@ commands:
sample:
- username ansible secret password group sysadmin
- username admin secret admin
xml:
description: NetConf rpc xml sent to device with transport C(netconf)
returned: always (empty list when no xml rpc to send)
type: list
version_added: 2.5
sample:
- '<config xmlns:xc=\"urn:ietf:params:xml:ns:netconf:base:1.0\">
<aaa xmlns=\"http://cisco.com/ns/yang/Cisco-IOS-XR-aaa-lib-cfg\">
<usernames xmlns=\"http://cisco.com/ns/yang/Cisco-IOS-XR-aaa-locald-cfg\">
<username xc:operation=\"merge\">
<name>test7</name>
<usergroup-under-usernames>
<usergroup-under-username>
<name>sysadmin</name>
</usergroup-under-username>
</usergroup-under-usernames>
<secret>$1$ZsXC$zZ50wqhDC543ZWQkkAHLW0</secret>
</username>
</usernames>
</aaa>
</config>'
"""
from functools import partial
import os
from functools import partial
from copy import deepcopy
import collections
from ansible.module_utils.basic import AnsibleModule
from ansible.module_utils.network.common.utils import remove_default_spec
from ansible.module_utils.network.iosxr.iosxr import get_config, load_config
from ansible.module_utils.network.iosxr.iosxr import iosxr_argument_spec
from ansible.module_utils.network.iosxr.iosxr import get_config, load_config, is_netconf, is_cliconf
from ansible.module_utils.network.iosxr.iosxr import iosxr_argument_spec, build_xml, etree_findall
try:
from base64 import b64decode
@ -184,6 +209,138 @@ except ImportError:
HAS_PARAMIKO = False
class PublicKeyManager(object):
def __init__(self, module, result):
self._module = module
self._result = result
def convert_key_to_base64(self):
""" IOS-XR only accepts base64 decoded files, this converts the public key to a temp file.
"""
if self._module.params['aggregate']:
name = 'aggregate'
else:
name = self._module.params['name']
if self._module.params['public_key_contents']:
key = self._module.params['public_key_contents']
elif self._module.params['public_key']:
readfile = open(self._module.params['public_key'], 'r')
key = readfile.read()
splitfile = key.split()[1]
base64key = b64decode(splitfile)
base64file = open('/tmp/publickey_%s.b64' % (name), 'wb')
base64file.write(base64key)
base64file.close()
return '/tmp/publickey_%s.b64' % (name)
def copy_key_to_node(self, base64keyfile):
""" Copy key to IOS-XR node. We use SFTP because older IOS-XR versions don't handle SCP very well.
"""
if (self._module.params['host'] is None or self._module.params['provider']['host'] is None):
return False
if (self._module.params['username'] is None or self._module.params['provider']['username'] is None):
return False
if self._module.params['aggregate']:
name = 'aggregate'
else:
name = self._module.params['name']
src = base64keyfile
dst = '/harddisk:/publickey_%s.b64' % (name)
user = self._module.params['username'] or self._module.params['provider']['username']
node = self._module.params['host'] or self._module.params['provider']['host']
password = self._module.params['password'] or self._module.params['provider']['password']
ssh_keyfile = self._module.params['ssh_keyfile'] or self._module.params['provider']['ssh_keyfile']
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
if not ssh_keyfile:
ssh.connect(node, username=user, password=password)
else:
ssh.connect(node, username=user, allow_agent=True)
sftp = ssh.open_sftp()
sftp.put(src, dst)
sftp.close()
ssh.close()
def addremovekey(self, command):
""" Add or remove key based on command
"""
if (self._module.params['host'] is None or self._module.params['provider']['host'] is None):
return False
if (self._module.params['username'] is None or self._module.params['provider']['username'] is None):
return False
user = self._module.params['username'] or self._module.params['provider']['username']
node = self._module.params['host'] or self._module.params['provider']['host']
password = self._module.params['password'] or self._module.params['provider']['password']
ssh_keyfile = self._module.params['ssh_keyfile'] or self._module.params['provider']['ssh_keyfile']
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
if not ssh_keyfile:
ssh.connect(node, username=user, password=password)
else:
ssh.connect(node, username=user, allow_agent=True)
ssh_stdin, ssh_stdout, ssh_stderr = ssh.exec_command('%s \r' % (command))
readmsg = ssh_stdout.read(100) # We need to read a bit to actually apply for some reason
if ('already' in readmsg) or ('removed' in readmsg) or ('really' in readmsg):
ssh_stdin.write('yes\r')
ssh_stdout.read(1) # We need to read a bit to actually apply for some reason
ssh.close()
return readmsg
def run(self):
if self._module.params['state'] == 'present':
if not self._module.check_mode:
key = self.convert_key_to_base64()
copykeys = self.copy_key_to_node(key)
if copykeys is False:
self._result['warnings'].append('Please set up your provider before running this playbook')
if self._module.params['aggregate']:
for user in self._module.params['aggregate']:
cmdtodo = "admin crypto key import authentication rsa username %s harddisk:/publickey_aggregate.b64" % (user)
addremove = self.addremovekey(cmdtodo)
if addremove is False:
self._result['warnings'].append('Please set up your provider before running this playbook')
else:
cmdtodo = "admin crypto key import authentication rsa username %s harddisk:/publickey_%s.b64" % \
(self._module.params['name'], self._module.params['name'])
addremove = self.addremovekey(cmdtodo)
if addremove is False:
self._result['warnings'].append('Please set up your provider before running this playbook')
elif self._module.params['state'] == 'absent':
if not self._module.check_mode:
if self._module.params['aggregate']:
for user in self._module.params['aggregate']:
cmdtodo = "admin crypto key zeroize authentication rsa username %s" % (user)
addremove = self.addremovekey(cmdtodo)
if addremove is False:
self._result['warnings'].append('Please set up your provider before running this playbook')
else:
cmdtodo = "admin crypto key zeroize authentication rsa username %s" % (self._module.params['name'])
addremove = self.addremovekey(cmdtodo)
if addremove is False:
self._result['warnings'].append('Please set up your provider before running this playbook')
elif self._module.params['purge'] is True:
if not self._module.check_mode:
cmdtodo = "admin crypto key zeroize authentication rsa all"
addremove = self.addremovekey(cmdtodo)
if addremove is False:
self._result['warnings'].append('Please set up your provider before running this playbook')
return self._result
def search_obj_in_list(name, lst):
for o in lst:
if o['name'] == name:
@ -192,213 +349,284 @@ def search_obj_in_list(name, lst):
return None
def map_obj_to_commands(updates, module):
commands = list()
want, have = updates
class ConfigBase(object):
def __init__(self, module, result, flag=None):
self._module = module
self._result = result
self._want = list()
self._have = list()
for w in want:
name = w['name']
state = w['state']
def get_param_value(self, key, item):
# if key doesn't exist in the item, get it from module.params
if not item.get(key):
value = self._module.params[key]
obj_in_have = search_obj_in_list(name, have)
if state == 'absent' and obj_in_have:
commands.append('no username ' + name)
elif state == 'present' and not obj_in_have:
user_cmd = 'username ' + name
commands.append(user_cmd)
if w['configured_password']:
commands.append(user_cmd + ' secret ' + w['configured_password'])
if w['group']:
commands.append(user_cmd + ' group ' + w['group'])
elif w['groups']:
for group in w['groups']:
commands.append(user_cmd + ' group ' + group)
elif state == 'present' and obj_in_have:
user_cmd = 'username ' + name
if module.params['update_password'] == 'always' and w['configured_password']:
commands.append(user_cmd + ' secret ' + w['configured_password'])
if w['group'] and w['group'] != obj_in_have['group']:
commands.append(user_cmd + ' group ' + w['group'])
elif w['groups']:
for group in w['groups']:
commands.append(user_cmd + ' group ' + group)
return commands
def map_config_to_obj(module):
data = get_config(module, config_filter='username')
users = data.strip().rstrip('!').split('!')
if not users:
return list()
instances = list()
for user in users:
user_config = user.strip().splitlines()
name = user_config[0].strip().split()[1]
group = None
if len(user_config) > 1:
group_or_secret = user_config[1].strip().split()
if group_or_secret[0] == 'group':
group = group_or_secret[1]
obj = {
'name': name,
'state': 'present',
'configured_password': None,
'group': group
}
instances.append(obj)
return instances
def get_param_value(key, item, module):
# if key doesn't exist in the item, get it from module.params
if not item.get(key):
value = module.params[key]
# if key does exist, do a type check on it to validate it
else:
value_type = module.argument_spec[key].get('type', 'str')
type_checker = module._CHECK_ARGUMENT_TYPES_DISPATCHER[value_type]
type_checker(item[key])
value = item[key]
# validate the param value (if validator func exists)
validator = globals().get('validate_%s' % key)
if all((value, validator)):
validator(value, module)
return value
def map_params_to_obj(module):
users = module.params['aggregate']
if not users:
if not module.params['name'] and module.params['purge']:
return list()
elif not module.params['name']:
module.fail_json(msg='username is required')
# if key does exist, do a type check on it to validate it
else:
aggregate = [{'name': module.params['name']}]
else:
value_type = self._module.argument_spec[key].get('type', 'str')
type_checker = self._module._CHECK_ARGUMENT_TYPES_DISPATCHER[value_type]
type_checker(item[key])
value = item[key]
# validate the param value (if validator func exists)
validator = globals().get('validate_%s' % key)
if all((value, validator)):
validator(value, self._module)
return value
def map_params_to_obj(self):
users = self._module.params['aggregate']
aggregate = list()
for item in users:
if not isinstance(item, dict):
aggregate.append({'name': item})
elif 'name' not in item:
module.fail_json(msg='name is required')
if not users:
if not self._module.params['name'] and self._module.params['purge']:
pass
elif not self._module.params['name']:
self._module.fail_json(msg='username is required')
else:
aggregate.append(item)
aggregate = [{'name': self._module.params['name']}]
else:
for item in users:
if not isinstance(item, dict):
aggregate.append({'name': item})
elif 'name' not in item:
self._module.fail_json(msg='name is required')
else:
aggregate.append(item)
objects = list()
for item in aggregate:
get_value = partial(get_param_value, item=item, module=module)
item['configured_password'] = get_value('configured_password')
item['group'] = get_value('group')
item['groups'] = get_value('groups')
item['state'] = get_value('state')
objects.append(item)
return objects
for item in aggregate:
get_value = partial(self.get_param_value, item=item)
item['configured_password'] = get_value('configured_password')
item['group'] = get_value('group')
item['groups'] = get_value('groups')
item['state'] = get_value('state')
self._want.append(item)
def convert_key_to_base64(module):
""" IOS-XR only accepts base64 decoded files, this converts the public key to a temp file.
"""
if module.params['aggregate']:
name = 'aggregate'
else:
name = module.params['name']
class CliConfiguration(ConfigBase):
def __init__(self, module, result):
super(CliConfiguration, self).__init__(module, result)
if module.params['public_key_contents']:
key = module.params['public_key_contents']
elif module.params['public_key']:
readfile = open(module.params['public_key'], 'r')
key = readfile.read()
splitfile = key.split()[1]
def map_config_to_obj(self):
data = get_config(self._module, config_filter='username')
users = data.strip().rstrip('!').split('!')
base64key = b64decode(splitfile)
base64file = open('/tmp/publickey_%s.b64' % (name), 'wb')
base64file.write(base64key)
base64file.close()
for user in users:
user_config = user.strip().splitlines()
return '/tmp/publickey_%s.b64' % (name)
name = user_config[0].strip().split()[1]
group = None
if len(user_config) > 1:
group_or_secret = user_config[1].strip().split()
if group_or_secret[0] == 'group':
group = group_or_secret[1]
obj = {
'name': name,
'state': 'present',
'configured_password': None,
'group': group
}
self._have.append(obj)
def map_obj_to_commands(self):
commands = list()
for w in self._want:
name = w['name']
state = w['state']
obj_in_have = search_obj_in_list(name, self._have)
if state == 'absent' and obj_in_have:
commands.append('no username ' + name)
elif state == 'present' and not obj_in_have:
user_cmd = 'username ' + name
commands.append(user_cmd)
if w['configured_password']:
commands.append(user_cmd + ' secret ' + w['configured_password'])
if w['group']:
commands.append(user_cmd + ' group ' + w['group'])
elif w['groups']:
for group in w['groups']:
commands.append(user_cmd + ' group ' + group)
elif state == 'present' and obj_in_have:
user_cmd = 'username ' + name
if self._module.params['update_password'] == 'always' and w['configured_password']:
commands.append(user_cmd + ' secret ' + w['configured_password'])
if w['group'] and w['group'] != obj_in_have['group']:
commands.append(user_cmd + ' group ' + w['group'])
elif w['groups']:
for group in w['groups']:
commands.append(user_cmd + ' group ' + group)
if self._module.params['purge']:
want_users = [x['name'] for x in self._want]
have_users = [x['name'] for x in self._have]
for item in set(have_users).difference(set(want_users)):
if item != 'admin':
commands.append('no username %s' % item)
if 'no username admin' in commands:
self._module.fail_json(msg='cannot delete the `admin` account')
self._result['commands'] = []
if commands:
commit = not self._module.check_mode
diff = load_config(self._module, commands, commit=commit)
if diff:
self._result['diff'] = dict(prepared=diff)
self._result['commands'] = commands
self._result['changed'] = True
def run(self):
self.map_params_to_obj()
self.map_config_to_obj()
self.map_obj_to_commands()
return self._result
def copy_key_to_node(module, base64keyfile):
""" Copy key to IOS-XR node. We use SFTP because older IOS-XR versions don't handle SCP very well.
"""
if (module.params['host'] is None or module.params['provider']['host'] is None):
return False
class NCConfiguration(ConfigBase):
def __init__(self, module, result):
super(NCConfiguration, self).__init__(module, result)
self._locald_meta = collections.OrderedDict()
self._locald_group_meta = collections.OrderedDict()
if (module.params['username'] is None or module.params['provider']['username'] is None):
return False
def generate_md5_hash(self, arg):
'''
Generate MD5 hash with randomly generated salt size of 3.
:param arg:
:return passwd:
'''
cmd = "openssl passwd -salt `openssl rand -base64 3` -1 "
return os.popen(cmd + arg).readlines()[0].strip()
if module.params['aggregate']:
name = 'aggregate'
else:
name = module.params['name']
def map_obj_to_xml_rpc(self):
self._locald_meta.update([
('aaa_locald', {'xpath': 'aaa/usernames', 'tag': True, 'ns': True}),
('username', {'xpath': 'aaa/usernames/username', 'tag': True, 'attrib': "operation"}),
('a:name', {'xpath': 'aaa/usernames/username/name'}),
('a:configured_password', {'xpath': 'aaa/usernames/username/secret', 'operation': 'edit'}),
])
src = base64keyfile
dst = '/harddisk:/publickey_%s.b64' % (name)
self._locald_group_meta.update([
('aaa_locald', {'xpath': 'aaa/usernames', 'tag': True, 'ns': True}),
('username', {'xpath': 'aaa/usernames/username', 'tag': True, 'attrib': "operation"}),
('a:name', {'xpath': 'aaa/usernames/username/name'}),
('usergroups', {'xpath': 'aaa/usernames/username/usergroup-under-usernames', 'tag': True, 'operation': 'edit'}),
('usergroup', {'xpath': 'aaa/usernames/username/usergroup-under-usernames/usergroup-under-username', 'tag': True, 'operation': 'edit'}),
('a:group', {'xpath': 'aaa/usernames/username/usergroup-under-usernames/usergroup-under-username/name', 'operation': 'edit'}),
])
user = module.params['username'] or module.params['provider']['username']
node = module.params['host'] or module.params['provider']['host']
password = module.params['password'] or module.params['provider']['password']
ssh_keyfile = module.params['ssh_keyfile'] or module.params['provider']['ssh_keyfile']
state = self._module.params['state']
_get_filter = build_xml('aaa', opcode="filter")
running = get_config(self._module, source='running', config_filter=_get_filter)
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
if not ssh_keyfile:
ssh.connect(node, username=user, password=password)
else:
ssh.connect(node, username=user, allow_agent=True)
sftp = ssh.open_sftp()
sftp.put(src, dst)
sftp.close()
ssh.close()
elements = etree_findall(running, 'username')
users = list()
for element in elements:
name_list = etree_findall(element, 'name')
users.append(name_list[0].text)
list_size = len(name_list)
if list_size == 1:
self._have.append({'name': name_list[0].text, 'group': None, 'groups': None})
elif list_size == 2:
self._have.append({'name': name_list[0].text, 'group': name_list[1].text, 'groups': None})
elif list_size > 2:
name_iter = iter(name_list)
next(name_iter)
tmp_list = list()
for name in name_iter:
tmp_list.append(name.text)
self._have.append({'name': name_list[0].text, 'group': None, 'groups': tmp_list})
def addremovekey(module, command):
""" Add or remove key based on command
"""
if (module.params['host'] is None or module.params['provider']['host'] is None):
return False
locald_params = list()
locald_group_params = list()
opcode = None
if (module.params['username'] is None or module.params['provider']['username'] is None):
return False
if state == 'absent':
opcode = "delete"
for want_item in self._want:
if want_item['name'] in users:
want_item['configured_password'] = None
locald_params.append(want_item)
elif state == 'present':
opcode = "merge"
for want_item in self._want:
if want_item['name'] not in users:
want_item['configured_password'] = self.generate_md5_hash(want_item['configured_password'])
locald_params.append(want_item)
user = module.params['username'] or module.params['provider']['username']
node = module.params['host'] or module.params['provider']['host']
password = module.params['password'] or module.params['provider']['password']
ssh_keyfile = module.params['ssh_keyfile'] or module.params['provider']['ssh_keyfile']
if want_item['group'] is not None:
locald_group_params.append(want_item)
if want_item['groups'] is not None:
for group in want_item['groups']:
want_item['group'] = group
locald_group_params.append(want_item.copy())
else:
if self._module.params['update_password'] == 'always' and want_item['configured_password'] is not None:
want_item['configured_password'] = self.generate_md5_hash(want_item['configured_password'])
locald_params.append(want_item)
else:
want_item['configured_password'] = None
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
if not ssh_keyfile:
ssh.connect(node, username=user, password=password)
else:
ssh.connect(node, username=user, allow_agent=True)
ssh_stdin, ssh_stdout, ssh_stderr = ssh.exec_command('%s \r' % (command))
readmsg = ssh_stdout.read(100) # We need to read a bit to actually apply for some reason
if ('already' in readmsg) or ('removed' in readmsg) or ('really' in readmsg):
ssh_stdin.write('yes\r')
ssh_stdout.read(1) # We need to read a bit to actually apply for some reason
ssh.close()
obj_in_have = search_obj_in_list(want_item['name'], self._have)
if want_item['group'] is not None and want_item['group'] != obj_in_have['group']:
locald_group_params.append(want_item)
elif want_item['groups'] is not None:
for group in want_item['groups']:
want_item['group'] = group
locald_group_params.append(want_item.copy())
return readmsg
purge_params = list()
if self._module.params['purge']:
want_users = [x['name'] for x in self._want]
have_users = [x['name'] for x in self._have]
for item in set(have_users).difference(set(want_users)):
if item != 'admin':
purge_params.append({'name': item})
self._result['xml'] = []
_edit_filter_list = list()
if opcode is not None:
if locald_params:
_edit_filter_list.append(build_xml('aaa', xmap=self._locald_meta,
params=locald_params, opcode=opcode))
if locald_group_params:
_edit_filter_list.append(build_xml('aaa', xmap=self._locald_group_meta,
params=locald_group_params, opcode=opcode))
if purge_params:
_edit_filter_list.append(build_xml('aaa', xmap=self._locald_meta,
params=purge_params, opcode="delete"))
diff = None
if _edit_filter_list:
commit = not self._module.check_mode
diff = load_config(self._module, _edit_filter_list, commit=commit, running=running,
nc_get_filter=_get_filter)
if diff:
if self._module._diff:
self._result['diff'] = dict(prepared=diff)
self._result['xml'] = _edit_filter_list
self._result['changed'] = True
def run(self):
self.map_params_to_obj()
self.map_obj_to_xml_rpc()
return self._result
def main():
@ -424,14 +652,16 @@ def main():
# remove default in aggregate spec, to handle common arguments
remove_default_spec(aggregate_spec)
mutually_exclusive = [('name', 'aggregate'), ('public_key', 'public_key_contents'), ('group', 'groups')]
argument_spec = dict(
aggregate=dict(type='list', elements='dict', options=aggregate_spec, aliases=['users', 'collection']),
aggregate=dict(type='list', elements='dict', options=aggregate_spec, aliases=['users', 'collection'],
mutually_exclusive=mutually_exclusive),
purge=dict(type='bool', default=False)
)
argument_spec.update(element_spec)
argument_spec.update(iosxr_argument_spec)
mutually_exclusive = [('name', 'aggregate'), ('public_key', 'public_key_contents'), ('group', 'groups')]
module = AnsibleModule(argument_spec=argument_spec,
mutually_exclusive=mutually_exclusive,
@ -449,77 +679,27 @@ def main():
'installed. It can be installed using `pip install paramiko`'
)
warnings = list()
result = {'changed': False, 'warnings': []}
if module.params['password'] and not module.params['configured_password']:
warnings.append(
result['warnings'].append(
'The "password" argument is used to authenticate the current connection. ' +
'To set a user password use "configured_password" instead.'
)
result = {'changed': False}
config_object = None
if is_cliconf(module):
module.deprecate(msg="cli support for 'iosxr_user' is deprecated. Use transport netconf instead",
version="4 releases from v2.5")
config_object = CliConfiguration(module, result)
elif is_netconf(module):
config_object = NCConfiguration(module, result)
want = map_params_to_obj(module)
have = map_config_to_obj(module)
if config_object:
result = config_object.run()
commands = map_obj_to_commands((want, have), module)
if module.params['purge']:
want_users = [x['name'] for x in want]
have_users = [x['name'] for x in have]
for item in set(have_users).difference(want_users):
if item != 'admin':
commands.append('no username %s' % item)
result['commands'] = commands
result['warnings'] = warnings
if 'no username admin' in commands:
module.fail_json(msg='cannot delete the `admin` account')
if commands:
commit = not module.check_mode
diff = load_config(module, commands, commit=commit)
if diff:
result['diff'] = dict(prepared=diff)
result['changed'] = True
if module.params['state'] == 'present' and (module.params['public_key_contents'] or module.params['public_key']):
if not module.check_mode:
key = convert_key_to_base64(module)
copykeys = copy_key_to_node(module, key)
if copykeys is False:
warnings.append('Please set up your provider before running this playbook')
if module.params['aggregate']:
for user in module.params['aggregate']:
cmdtodo = "admin crypto key import authentication rsa username %s harddisk:/publickey_aggregate.b64" % (user)
addremove = addremovekey(module, cmdtodo)
if addremove is False:
warnings.append('Please set up your provider before running this playbook')
else:
cmdtodo = "admin crypto key import authentication rsa username %s harddisk:/publickey_%s.b64" % (module.params['name'], module.params['name'])
addremove = addremovekey(module, cmdtodo)
if addremove is False:
warnings.append('Please set up your provider before running this playbook')
elif module.params['state'] == 'absent':
if not module.check_mode:
if module.params['aggregate']:
for user in module.params['aggregate']:
cmdtodo = "admin crypto key zeroize authentication rsa username %s" % (user)
addremove = addremovekey(module, cmdtodo)
if addremove is False:
warnings.append('Please set up your provider before running this playbook')
else:
cmdtodo = "admin crypto key zeroize authentication rsa username %s" % (module.params['name'])
addremove = addremovekey(module, cmdtodo)
if addremove is False:
warnings.append('Please set up your provider before running this playbook')
elif module.params['purge'] is True:
if not module.check_mode:
cmdtodo = "admin crypto key zeroize authentication rsa all"
addremove = addremovekey(module, cmdtodo)
if addremove is False:
warnings.append('Please set up your provider before running this playbook')
if module.params['public_key_contents'] or module.params['public_key']:
pubkey_object = PublicKeyManager(module, result)
result = pubkey_object.run()
module.exit_json(**result)

7
test/integration/targets/iosxr_interface/tests/netconf/net_interface.yaml
View file

@ -4,13 +4,6 @@
# Add minimal testcase to check args are passed correctly to
# implementation module and module run is successful.
- name: Enable Netconf service
iosxr_netconf:
netconf_port: 830
netconf_vrf: 'default'
state: present
register: result
- name: Setup interface
net_interface:
name: GigabitEthernet0/0/0/1

11
test/integration/targets/iosxr_user/tasks/cli.yaml
View file

@ -1,4 +1,11 @@
---
- name: collect all common test cases
find:
paths: "{{ role_path }}/tests/common"
patterns: "{{ testcase }}.yaml"
register: common_test_cases
delegate_to: localhost
- name: collect all cli test cases
find:
paths: "{{ role_path }}/tests/cli"
@ -6,6 +13,10 @@
register: test_cases
delegate_to: localhost
- set_fact:
test_cases:
files: "{{ common_test_cases.files }} + {{ test_cases.files }}"
- name: set test_items
set_fact: test_items="{{ test_cases.files | map(attribute='path') | list }}"

1
test/integration/targets/iosxr_user/tasks/main.yaml
View file

@ -1,2 +1,3 @@
---
- { include: cli.yaml, tags: ['cli'] }
- { include: netconf.yaml, tags: ['netconf'] }

33
test/integration/targets/iosxr_user/tasks/netconf.yaml Normal file
View file

@ -0,0 +1,33 @@
---
- name: collect all common test cases
find:
paths: "{{ role_path }}/tests/common"
patterns: "{{ testcase }}.yaml"
register: common_test_cases
delegate_to: localhost
- name: collect all netconf test cases
find:
paths: "{{ role_path }}/tests/netconf"
patterns: "{{ testcase }}.yaml"
register: test_cases
delegate_to: localhost
- set_fact:
test_cases:
files: "{{ common_test_cases.files }} + {{ test_cases.files }}"
- name: set test_items
set_fact: test_items="{{ test_cases.files | map(attribute='path') | list }}"
- name: run test case (connection=local)
include: "{{ test_case_to_run }} ansible_connection=local"
with_items: "{{ test_items }}"
loop_control:
loop_var: test_case_to_run
#- name: run test case (connection=netconf)
#include: "{{ test_case_to_run }} ansible_connection=network_cli"
#with_items: "{{ test_items }}"
#loop_control:
# loop_var: test_case_to_run

0
test/integration/targets/iosxr_user/tests/cli/auth.yaml → test/integration/targets/iosxr_user/tests/common/auth.yaml
View file

170
test/integration/targets/iosxr_user/tests/netconf/basic.yaml Normal file
View file

@ -0,0 +1,170 @@
---
- name: Remove users prior to tests
iosxr_config:
lines:
- no username ansible1
- no username ansible2
- no username ansible3
provider: "{{ cli }}"
- name: Create user (SetUp)
iosxr_user:
name: ansible1
configured_password: password
state: present
provider: "{{ netconf }}"
register: result
- assert:
that:
- 'result.changed == true'
- '"ansible1" in result.xml[0]'
- '"secret" in result.xml[0]'
- name: Create user with update_password always (not idempotent)
iosxr_user:
name: ansible1
configured_password: password
update_password: always
state: present
provider: "{{ netconf }}"
register: result
- assert:
that:
- 'result.changed == true'
- '"ansible1" in result.xml[0]'
- '"secret" in result.xml[0]'
- name: Create user again with update_password on_create (idempotent)
iosxr_user:
name: ansible1
configured_password: password
update_password: on_create
state: present
provider: "{{ netconf }}"
register: result
- assert:
that:
- 'result.changed == false'
- 'result.xml | length == 0'
- name: Modify user group
iosxr_user:
name: ansible1
configured_password: password
update_password: on_create
group: sysadmin
state: present
provider: "{{ netconf }}"
register: result
- assert:
that:
- 'result.changed == true'
- '"ansible1" in result.xml[0]'
- '"sysadmin" in result.xml[0]'
- name: Modify user group again (idempotent)
iosxr_user:
name: ansible1
configured_password: password
update_password: on_create
group: sysadmin
state: present
provider: "{{ netconf }}"
register: result
- assert:
that:
- 'result.changed == false'
- 'result.xml | length == 0'
- name: Collection of users (SetUp)
iosxr_user:
aggregate:
- name: ansible2
- name: ansible3
configured_password: password
state: present
group: sysadmin
provider: "{{ netconf }}"
register: result
- assert:
that:
- 'result.changed == true'
- '"ansible2" in result.xml[0]'
- '"secret" in result.xml[0]'
- '"sysadmin" in result.xml[1]'
- '"ansible2" in result.xml[0]'
- '"secret" in result.xml[0]'
- '"sysadmin" in result.xml[1]'
- name: Add collection of users again with update_password always (not idempotent)
iosxr_user:
aggregate:
- name: ansible2
- name: ansible3
configured_password: password
state: present
group: sysadmin
provider: "{{ netconf }}"
register: result
- assert:
that:
- 'result.changed == true'
- '"ansible2" in result.xml[0]'
- '"ansible3" in result.xml[0]'
- '"secret" in result.xml[0]'
- name: Add collection of users again with update_password on_create (idempotent)
iosxr_user:
aggregate:
- name: ansible2
- name: ansible3
configured_password: password
update_password: on_create
state: present
group: sysadmin
provider: "{{ netconf }}"
register: result
- assert:
that:
- 'result.changed == false'
- 'result.xml | length == 0'
- name: Delete collection of users
iosxr_user:
aggregate:
- name: ansible1
- name: ansible2
- name: ansible3
state: absent
provider: "{{ netconf }}"
register: result
- assert:
that:
- 'result.changed == true'
- '"ansible1" in result.xml[0]'
- '"ansible2" in result.xml[0]'
- '"ansible3" in result.xml[0]'
- name: Delete collection of users again (idempotent)
iosxr_user:
aggregate:
- name: ansible1
- name: ansible2
- name: ansible3
state: absent
provider: "{{ netconf }}"
register: result
- assert:
that:
- 'result.changed == false'
- 'result.xml | length == 0'

5
test/units/modules/network/iosxr/test_iosxr_user.py
View file

@ -38,15 +38,20 @@ class TestIosxrUserModule(TestIosxrModule):
self.mock_load_config = patch('ansible.modules.network.iosxr.iosxr_user.load_config')
self.load_config = self.mock_load_config.start()
self.mock_is_cliconf = patch('ansible.modules.network.iosxr.iosxr_user.is_cliconf')
self.is_cliconf = self.mock_is_cliconf.start()
def tearDown(self):
super(TestIosxrUserModule, self).tearDown()
self.mock_get_config.stop()
self.mock_load_config.stop()
self.mock_is_cliconf.stop()
def load_fixtures(self, commands=None, transport='cli'):
self.get_config.return_value = load_fixture('iosxr_user_config.cfg')
self.load_config.return_value = dict(diff=None, session='session')
self.is_cliconf.return_value = True
def test_iosxr_user_delete(self):
set_module_args(dict(name='ansible', state='absent'))