ios_user module - implement sshkey option (#38782)

* ios_user module - add sshkey support

* ios_user - Add version_added to sshkey option

* ios_user - pep8 indentation fixes in unit tests

* ios_user - use b64decode method that works on python 2 and 3

lib/ansible/modules/network/ios/ios_user.py

@@ -78,6 +78,11 @@ options:
         defining the view name. This argument does not check if the view
         has been configured on the device.
     aliases: ['role']
+  sshkey:
+    description:
+      - Specifies the SSH public key to configure
+        for the given username.  This argument accepts a valid SSH key value.
+    version_added: "2.6"
   nopassword:
     description:
       - Defines the username without assigning
@@ -109,6 +114,7 @@ EXAMPLES = """
   ios_user:
     name: ansible
     nopassword: True
+    sshkey: "{{ lookup('file', '~/.ssh/id_rsa.pub') }}"
     state: present
 
 - name: remove all users except admin
@@ -165,6 +171,8 @@ from copy import deepcopy
 
 import re
 import json
+import base64
+import hashlib
 
 from functools import partial
 
@@ -189,6 +197,22 @@ def user_del_cmd(username):
     }
 
 
+def sshkey_fingerprint(sshkey):
+    # IOS will accept a MD5 fingerprint of the public key
+    # and is easier to configure in a single line
+    # we calculate this fingerprint here
+    if not sshkey:
+        return None
+    if ' ' in sshkey:
+        # ssh-rsa AAA...== comment
+        keyparts = sshkey.split(' ')
+        keyparts[1] = hashlib.md5(base64.b64decode(keyparts[1])).hexdigest().upper()
+        return ' '.join(keyparts)
+    else:
+        # just the key, assume rsa type
+        return 'ssh-rsa %s' % hashlib.md5(base64.b64decode(sshkey)).hexdigest().upper()
+
+
 def map_obj_to_commands(updates, module):
     commands = list()
     state = module.params['state']
@@ -200,11 +224,21 @@ def map_obj_to_commands(updates, module):
     def add(command, want, x):
         command.append('username %s %s' % (want['name'], x))
 
+    def add_ssh(command, want, x=None):
+        command.append('ip ssh pubkey-chain')
+        command.append(' no username %s' % want['name'])
+        if x:
+            command.append(' username %s' % want['name'])
+            command.append('  key-hash %s' % x)
+            command.append('  exit')
+        command.append(' exit')
+
     for update in updates:
         want, have = update
 
         if want['state'] == 'absent':
             commands.append(user_del_cmd(want['name']))
+            add_ssh(commands, want)
             continue
 
         if needs_update(want, have, 'view'):
@@ -213,6 +247,9 @@ def map_obj_to_commands(updates, module):
         if needs_update(want, have, 'privilege'):
             add(commands, want, 'privilege %s' % want['privilege'])
 
+        if needs_update(want, have, 'sshkey'):
+            add_ssh(commands, want, want['sshkey'])
+
         if needs_update(want, have, 'configured_password'):
             if update_password == 'always' or not have:
                 add(commands, want, 'secret %s' % want['configured_password'])
@@ -232,6 +269,12 @@ def parse_view(data):
         return match.group(1)
 
 
+def parse_sshkey(data):
+    match = re.search(r'key-hash (\S+ \S+(?: .+)?)$', data, re.M)
+    if match:
+        return match.group(1)
+
+
 def parse_privilege(data):
     match = re.search(r'privilege (\S+)', data, re.M)
     if match:
@@ -251,11 +294,15 @@ def map_config_to_obj(module):
         regex = r'username %s .+$' % user
         cfg = re.findall(regex, data, re.M)
         cfg = '\n'.join(cfg)
+        sshregex = r'username %s\n\s+key-hash .+$' % user
+        sshcfg = re.findall(sshregex, data, re.M)
+        sshcfg = '\n'.join(sshcfg)
         obj = {
             'name': user,
             'state': 'present',
             'nopassword': 'nopassword' in cfg,
             'configured_password': None,
+            'sshkey': parse_sshkey(sshcfg),
             'privilege': parse_privilege(cfg),
             'view': parse_view(cfg)
         }
@@ -311,6 +358,7 @@ def map_params_to_obj(module):
         item['nopassword'] = get_value('nopassword')
         item['privilege'] = get_value('privilege')
         item['view'] = get_value('view')
+        item['sshkey'] = sshkey_fingerprint(get_value('sshkey'))
         item['state'] = get_value('state')
         objects.append(item)
 
@@ -343,6 +391,8 @@ def main():
         privilege=dict(type='int'),
         view=dict(aliases=['role']),
 
+        sshkey=dict(),
+
         state=dict(default='present', choices=['present', 'absent'])
     )
     aggregate_spec = deepcopy(element_spec)

test/integration/targets/ios_user/files/test_rsa

@@ -0,0 +1,27 @@
+-----BEGIN RSA PRIVATE KEY-----
+MIIEpAIBAAKCAQEAseTsv6oTMJbsRgOSaNEVBlsqE4tKI2MBEOHzTlUnE3GBRtZi
+xlUlF9rZcjtk+hTIz3N3UrLtxMnWqv0YGM8tXZTcz50M9ANDE/oNXrXkkMKX5WWr
+OcgDCAAZYSVJ/vIt/xZkYteax6zMZQqEBIc/rKo8KuxKZTUMqj0GHkCQbA/o7lBQ
+hpdUZxtU/Y2uc9tXVhav3jVzTFUOdwZrCAJghdCCweJfwAj1DdHODip+/hNu+3Sk
+DNl0bdhWoNNnUFCYPOUK2B79OMg2+r5bwH7xgURIoTuc54HlGectJpIzI7GB6VfL
+UCsJn0uQ7HwX7XAtZjhJWP8dKPd1FuUgg5M8MwIDAQABAoIBACfBVz9GDN/Q+qBy
+7+dIwoAXI4IWoMzjtTtGo48f7Iml1hQ0mQJlyNJZ8DpdF6XjuzTRQxtSLVzLFpRD
+13zD4AmH2Qj0ug5WJEl0mkRONfQ76KI1ZFyXXEYPb5yMLssw6CKXqHuGX2q8LTlv
+bi1s5Ef8C1I0WDPh9SCeXp2oJB5h2G+HtCXDyxASK2nAKqhkpQqPhg5Rd50mBOpD
+WE/lor358hU0Aj/qhzjeWKNNK8pgeahXz5anEQZ69TUH102B6bNh8Ao4ZL2j3tr0
+6FbE3ooQT7+zOLm5xOFJ9OnJ2yDVW6Dj1Czllx2vJUcxKsKxaGF76xNCIPiSUUfS
+mnOnEfECgYEA2bBFc9Pb8AI0bZZ8Q6XE7Jqa6BOaLbzyjK6IzdyAV/LKdk3yRfEZ
+Rb2iNy8poBUYBqBUMfRsRVq5dabjYkz9X5e+75S8Lm/qiktlhTpQYWk5q+eBZdPm
+I+dG64Tdyv+Y/NwN4enIsw8LGllY472iUf37ms2+uOA8/BysQ2n7ss0CgYEA0TPD
+IhmLqNhQGkS2GU6tM8G7LyGOaIH3mmyCviYgEauWWw3bn/Hhiq/6tLtQc6pv2nIa
+ifbACnI+GiIoBFwz8ofuFA8dm76uro7o6eWP5iUizoGISYSewCFpcCpp0xn7/FNR
+3RT4YRBMt3yL8J1cVBpPRRbIwp/bZ+pRb0Ggqv8CgYEAoNAFHqHdkhou3N4UgmzN
+YvR7hwIkHbG9hIvS6DECZvYm9upyFZUcVFbYpOekWmv6ybpbOGQWL83rv6w/wfia
+HKofFSHNOojWvL8iCh+gDbYMMp/dCXpWQyOxUn9e0X2saO+vGbr41r5AN4DVl7gZ
+V3THD/75691Lb/tGjq6Wj+kCgYB6ZhadNOUJfMYhGGKSm/2qcobaJH/1lVUQ/Lvi
+FNxeek4WKB1/jz2urxe39oAzrFyVKn1sivoBIqZDFAjlxCyAkhcxlUZ1gTMi3mpX
+rwBqXv/mYtMicH2RW/scrTQNVv6fuwACoepQoADCuhQGS4thiaMngRUlCfKM8gOD
+XJpscQKBgQDIMURtVIV/2ZcGqHv/3G5jsPJPsTycv6YR4gTs5GUBy5If2Rs7DMWE
+pJLIcU+SJhMeVKTZPrePibzCp2+rMSI5pc6T+9LC79RKsfie3UybWfLZrSmtnxJx
+MgC49TR4NFP6yoYJPYiTdRJ/1Bu68WfVafFK86i9MKAI5OU2ba3/Bg==
+-----END RSA PRIVATE KEY-----

test/integration/targets/ios_user/files/test_rsa.pub

@@ -0,0 +1 @@
+ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABAQCx5Oy/qhMwluxGA5Jo0RUGWyoTi0ojYwEQ4fNOVScTcYFG1mLGVSUX2tlyO2T6FMjPc3dSsu3Eydaq/RgYzy1dlNzPnQz0A0MT+g1eteSQwpflZas5yAMIABlhJUn+8i3/FmRi15rHrMxlCoQEhz+sqjwq7EplNQyqPQYeQJBsD+juUFCGl1RnG1T9ja5z21dWFq/eNXNMVQ53BmsIAmCF0ILB4l/ACPUN0c4OKn7+E277dKQM2XRt2Fag02dQUJg85QrYHv04yDb6vlvAfvGBREihO5zngeUZ5y0mkjMjsYHpV8tQKwmfS5DsfBftcC1mOElY/x0o93UW5SCDkzwz ansible_ios_user_test

test/integration/targets/ios_user/tests/cli/auth.yaml

@@ -38,3 +38,41 @@
 
   - name: reset connection
     meta: reset_connection
+
+
+- block:
+  - name: Create user with sshkey
+    ios_user:
+      name: ssh_user
+      privilege: 15
+      role: network-operator
+      state: present
+      provider: "{{ cli }}"
+      sshkey: "{{ lookup('file', 'files/test_rsa.pub') }}"
+
+  - name: test sshkey login
+    shell: "ssh ssh_user@{{ ansible_ssh_host }} -p {{ ansible_ssh_port|default(22) }} -o IdentityFile={{ role_path }}/files/test_rsa -o UserKnownHostsFile=/dev/null -o StrictHostKeyChecking=no -o PubkeyAuthentication=yes show version"
+
+  - name: test login without sshkey (should fail)
+    expect:
+      command: "ssh ssh_user@{{ ansible_ssh_host }} -p {{ ansible_ssh_port|default(22) }} -o UserKnownHostsFile=/dev/null -o StrictHostKeyChecking=no -o PubkeyAuthentication=no show version"
+      responses:
+        (?i)password: badpass
+    ignore_errors: yes
+    register: results
+
+  - name: check that attempt failed
+    assert:
+      that:
+        - results.failed
+
+  always:
+  - name: delete user
+    ios_user:
+      name: ssh_user
+      state: absent
+      provider: "{{ cli }}"
+    register: result
+
+  - name: reset connection
+    meta: reset_connection

test/integration/targets/ios_user/tests/cli/basic.yaml

@@ -80,5 +80,5 @@
     that:
       - 'result.changed == true'
       - '"no username ansibletest1" in result.commands[0]["command"]'
-      - '"no username ansibletest2" in result.commands[1]["command"]'
-      - '"no username ansibletest3" in result.commands[2]["command"]'
+      - '"no username ansibletest2" in result.commands[4]["command"]'
+      - '"no username ansibletest3" in result.commands[8]["command"]'

test/units/modules/network/ios/test_ios_user.py

@@ -57,16 +57,21 @@ class TestIosUserModule(TestIosModule):
     def test_ios_user_delete(self):
         set_module_args(dict(name='ansible', state='absent'))
         result = self.execute_module(changed=True)
-        cmd = {
+        cmds = [
+            {
                 "command": "no username ansible", "answer": "y", "newline": False,
                 "prompt": "This operation will remove all username related configurations with same name",
-        }
+            },
+            'ip ssh pubkey-chain',
+            ' no username ansible',
+            ' exit'
+        ]
 
         result_cmd = []
         for i in result['commands']:
             result_cmd.append(i)
 
-        self.assertEqual(result_cmd, [cmd])
+        self.assertEqual(result_cmd, cmds)
 
     def test_ios_user_password(self):
         set_module_args(dict(name='ansible', configured_password='test'))
@@ -114,3 +119,16 @@ class TestIosUserModule(TestIosModule):
         set_module_args(dict(name='ansible', configured_password='test', update_password='always'))
         result = self.execute_module(changed=True)
         self.assertEqual(result['commands'], ['username ansible secret test'])
+
+    def test_ios_user_set_sshkey(self):
+        set_module_args(dict(name='ansible', sshkey='dGVzdA=='))
+        commands = [
+            'ip ssh pubkey-chain',
+            ' no username ansible',
+            ' username ansible',
+            '  key-hash ssh-rsa 098F6BCD4621D373CADE4E832627B4F6',
+            '  exit',
+            ' exit'
+        ]
+        result = self.execute_module(changed=True, commands=commands)
+        self.assertEqual(result['commands'], commands)