ansible/community.general
1
0
Fork 0
mirror of https://github.com/ansible-collections/community.general.git synced 2024-09-14 20:13:21 +02:00
Code Releases Activity

new module eos_user (#20759)

Browse source 
* adds new module for managing local user accounts on eos devices
* adds unit test cases for module

This module only supports network_cli
This commit is contained in:
Peter Sprygada 2017-01-27 14:12:36 -05:00 committed by GitHub
parent 085865cee8
commit 271c0c4574
3 changed files with 516 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

368
lib/ansible/modules/network/eos/eos_user.py Normal file
View file

@ -0,0 +1,368 @@
#!/usr/bin/python
#
# This file is part of Ansible
#
# Ansible is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Ansible is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with Ansible. If not, see <http://www.gnu.org/licenses/>.
#
DOCUMENTATION = """
---
module: eos_user
version_added: "2.3"
author: "Peter Sprygada (@privateip)"
short_description: Manage the collection of local users on EOS devices
description:
- This module provides declarative management of the local usernames
configured on Arista EOS devices. It allows playbooks to manage
either individual usernames or the collection of usernames in the
current running config. It also supports purging usernames from the
configuration that are not explicitly defined.
notes:
- This module requires connection to be network_cli
options:
users:
description:
- The set of username objects to be configured on the remote
Arista EOS device. The list entries can either be the username
or a hash of username and properties. This argument is mutually
exclusive with the C(username) argument.
required: false
default: null
username:
description:
- The username to be configured on the remote Arista EOS
device. This argument accepts a stringv value and is mutually
exclusive with the C(users) argument.
required: false
default: null
update_password:
description:
- Since passwords are encrypted in the device running config, this
argument will instruct the module when to change the password. When
set to C(always), the password will always be updated in the device
and when set to C(on_create) the password will be updated only if
the username is created.
required: false
default: always
choices: ['on_create', 'always']
privilege:
description:
- The C(privilege) argument configures the privilege level of the
user when logged into the system. This argument accepts integer
values in the range of 1 to 15.
required: false
default: null
role:
description:
- The C(role) argument configures the role for the username in the
device running configuration. The argument accepts a string value
defining the role name. This argument does not check if the role
has been configured on the device.
required: false
default: null
sshkey:
description:
- The C(sshkey) argument defines the SSH public key to configure
for the username. This argument accepts a valid SSH key value.
required: false
default: null
nopassword:
description:
- The C(nopassword) argument defines the username without assigning
a password. This will allow the user to login to the system
without being authenticated by a password. This argument accepts
boolean values.
required: false
default: null
choices: ['true', 'false']
purge:
description:
- The C(purge) argument 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 which cannot be deleted per EOS constraints.
required: false
default: false
state:
description:
- The C(state) argument configures the state of the uername definition
as it relates to the device operational configuration. When set
to I(present), the username(s) should be configured in the device active
configuration and when set to I(absent) the username(s) should not be
in the device active configuration
required: false
default: present
choices: ['present', 'absent']
"""
EXAMPLES = """
- name: create a new user
eos_user:
username: ansible
sshkey: "{{ lookup('file', '~/.ssh/id_rsa.pub') }}"
state: present
- name: remove all users except admin
eos_user:
purge: yes
- name: set multiple users to privilege level
users:
- username: netop
- username: netend
privilege: 15
state: present
"""
RETURN = """
commands:
description: The list of configuration mode commands to send to the device
returned: always
type: list
sample:
- username ansible secret password
- username admin secret admin
session_name:
description: The EOS config session name used to load the configuration
returned: when changed is True
type: str
sample: ansible_1479315771
start:
description: The time the job started
returned: always
type: str
sample: "2016-11-16 10:38:15.126146"
end:
description: The time the job ended
returned: always
type: str
sample: "2016-11-16 10:38:25.595612"
delta:
description: The time elapsed to perform all operations
returned: always
type: str
sample: "0:00:10.469466"
"""
import re
from functools import partial
from ansible.module_utils.local import LocalAnsibleModule
from ansible.module_utils.eos import get_config, load_config
from ansible.module_utils.six import iteritems
def validate_privilege(value, module):
if not 1 <= value <= 15:
module.fail_json(msg='privilege must be between 1 and 15, got %s' % value)
def map_obj_to_commands(updates, module):
commands = list()
state = module.params['state']
update_password = module.params['update_password']
for update in updates:
want, have = update
needs_update = lambda x: want.get(x) and (want.get(x) != have.get(x))
add = lambda x: commands.append('username %s %s' % (want['username'], x))
if want['state'] == 'absent':
commands.append('no username %s' % want['username'])
continue
if needs_update('role'):
add('role %s' % want['role'])
if needs_update('privilege'):
add('privilege %s' % want['privilege'])
if needs_update('password'):
if update_password == 'always' or not have:
add('secret %s' % want['password'])
if needs_update('sshkey'):
add('sshkey %s' % want['sshkey'])
if needs_update('nopassword'):
if want['nopassword']:
add('nopassword')
else:
add('no username %s nopassword' % want['username'])
return commands
def parse_role(data):
match = re.search(r'role (\S+)', data, re.M)
if match:
return match.group(1)
def parse_sshkey(data):
match = re.search(r'sshkey (.+)$', data, re.M)
if match:
return match.group(1)
def parse_privilege(data):
match = re.search(r'privilege (\S+)', data, re.M)
if match:
return int(match.group(1))
def map_config_to_obj(module):
data = get_config(module, flags=['section username'])
match = re.findall(r'^username (\S+)', data, re.M)
if not match:
return list()
instances = list()
for user in set(match):
regex = r'username %s .+$' % user
cfg = re.findall(r'username %s .+$' % user, data, re.M)
cfg = '\n'.join(cfg)
obj = {
'username': user,
'state': 'present',
'nopassword': 'nopassword' in cfg,
'password': None,
'sshkey': parse_sshkey(cfg),
'privilege': parse_privilege(cfg),
'role': parse_role(cfg)
}
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['users']
if not users:
if not module.params['username'] and module.params['purge']:
return list()
elif not module.params['username']:
module.fail_json(msg='username is required')
else:
collection = [{'username': module.params['username']}]
else:
collection = list()
for item in users:
if not isinstance(item, dict):
collection.append({'username': item})
elif 'username' not in item:
module.fail_json(msg='username is required')
else:
collection.append(item)
objects = list()
for item in collection:
get_value = partial(get_param_value, item=item, module=module)
item['password'] = get_value('password')
item['nopassword'] = get_value('nopassword')
item['privilege'] = get_value('privilege')
item['role'] = get_value('role')
item['sshkey'] = get_value('sshkey')
item['state'] = get_value('state')
objects.append(item)
return objects
def update_objects(want, have):
updates = list()
for entry in want:
item = next((i for i in have if i['username'] == entry['username']), None)
if all((item is None, entry['state'] == 'present')):
updates.append((entry, {}))
elif item:
for key, value in iteritems(entry):
if value and value != item[key]:
updates.append((entry, item))
return updates
def main():
""" main entry point for module execution
"""
argument_spec = dict(
users=dict(type='list', no_log=True),
username=dict(),
password=dict(no_log=True),
nopassword=dict(type='bool'),
update_password=dict(default='always', choices=['on_create', 'always']),
privilege=dict(type='int'),
role=dict(),
sshkey=dict(),
purge=dict(type='bool', default=False),
state=dict(default='present', choices=['present', 'absent'])
)
mutually_exclusive = [('username', 'users')]
module = LocalAnsibleModule(argument_spec=argument_spec,
mutually_exclusive=mutually_exclusive,
supports_check_mode=True)
result = {'changed': False}
want = map_params_to_obj(module)
have = map_config_to_obj(module)
commands = map_obj_to_commands(update_objects(want, have), module)
if module.params['purge']:
want_users = [x['username'] for x in want]
have_users = [x['username'] 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
# the eos cli prevents this by rule so capture it and display
# a nice failure message
if 'no username admin' in commands:
module.fail_json(msg='cannot delete the `admin` account')
if commands:
commit = not module.check_mode
response = load_config(module, commands, commit=commit)
if response.get('diff') and module._diff:
result['diff'] = {'prepared': response.get('diff')}
result['session_name'] = response.get('session')
result['changed'] = True
module.exit_json(**result)
if __name__ == '__main__':
main()

2
test/units/modules/network/eos/fixtures/eos_user_config.cfg Normal file
View file

@ -0,0 +1,2 @@
username admin role network-admin secret 5 $1$mdQIUxjg$3t3lzBpfKfITKvFm1uEIY.
username ansible role network-admin secret 5 $1$3yWSXiIi$VdzV59ChiurrNdGxlDeAW/

146
test/units/modules/network/eos/test_eos_user.py Normal file
View file

@ -0,0 +1,146 @@
# This file is part of Ansible
#
# Ansible is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Ansible is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with Ansible. If not, see <http://www.gnu.org/licenses/>.
# Make coding more python3-ish
from __future__ import (absolute_import, division, print_function)
__metaclass__ = type
import os
import json
from ansible.compat.tests import unittest
from ansible.compat.tests.mock import patch, MagicMock
from ansible.errors import AnsibleModuleExit
from ansible.modules.network.eos import eos_user
from ansible.module_utils import basic
from ansible.module_utils._text import to_bytes
def set_module_args(args):
args = json.dumps({'ANSIBLE_MODULE_ARGS': args})
basic._ANSIBLE_ARGS = to_bytes(args)
fixture_path = os.path.join(os.path.dirname(__file__), 'fixtures')
fixture_data = {}
def load_fixture(name):
path = os.path.join(fixture_path, name)
if path in fixture_data:
return fixture_data[path]
with open(path) as f:
data = f.read()
try:
data = json.loads(data)
except:
pass
fixture_data[path] = data
return data
class TestEosUserModule(unittest.TestCase):
def setUp(self):
self.mock_get_config = patch('ansible.modules.network.eos.eos_user.get_config')
self.get_config = self.mock_get_config.start()
self.mock_load_config = patch('ansible.modules.network.eos.eos_user.load_config')
self.load_config = self.mock_load_config.start()
def tearDown(self):
self.mock_get_config.stop()
self.mock_load_config.stop()
def execute_module(self, failed=False, changed=False, commands=None, sort=True):
self.get_config.return_value = load_fixture('eos_user_config.cfg')
self.load_config.return_value = dict(diff=None, session='session')
with self.assertRaises(AnsibleModuleExit) as exc:
eos_user.main()
result = exc.exception.result
if failed:
self.assertTrue(result['failed'], result)
else:
self.assertEqual(result['changed'], changed, result)
if commands:
if sort:
self.assertEqual(sorted(commands), sorted(result['commands']), result['commands'])
else:
self.assertEqual(commands, result['commands'])
return result
def test_eos_user_create(self):
set_module_args(dict(username='test', nopassword=True))
commands = ['username test nopassword']
self.execute_module(changed=True, commands=commands)
def test_eos_user_delete(self):
set_module_args(dict(username='ansible', state='absent'))
commands = ['no username ansible']
self.execute_module(changed=True, commands=commands)
def test_eos_user_password(self):
set_module_args(dict(username='ansible', password='test'))
commands = ['username ansible secret ********']
self.execute_module(changed=True, commands=commands)
def test_eos_user_privilege(self):
set_module_args(dict(username='ansible', privilege=15))
commands = ['username ansible privilege 15']
self.execute_module(changed=True, commands=commands)
def test_eos_user_privilege_invalid(self):
set_module_args(dict(username='ansible', privilege=25))
self.execute_module(failed=True)
def test_eos_user_purge(self):
set_module_args(dict(purge=True))
commands = ['no username ansible']
self.execute_module(changed=True, commands=commands)
def test_eos_user_role(self):
set_module_args(dict(username='ansible', role='test'))
commands = ['username ansible role test']
self.execute_module(changed=True, commands=commands)
def test_eos_user_sshkey(self):
set_module_args(dict(username='ansible', sshkey='test'))
commands = ['username ansible sshkey test']
self.execute_module(changed=True, commands=commands)
def test_eos_user_update_password_changed(self):
set_module_args(dict(username='test', password='test', update_password='on_create'))
commands = ['username ******** secret ********']
self.execute_module(changed=True, commands=commands)
def test_eos_user_update_password_on_create_ok(self):
set_module_args(dict(username='ansible', password='test', update_password='on_create'))
commands = []
self.execute_module(commands=commands)
def test_eos_user_update_password_always(self):
set_module_args(dict(username='ansible', password='test', update_password='always'))
commands = ['username ansible secret ********']
self.execute_module(changed=True, commands=commands)