config: disable ssh access to a configured user account

Cloud config can now disable ssh access to non-root users.

When defining the 'users' list in cloud-configuration a boolean
'ssh_redirect_user: true' can be provided to disable ssh logins for
that user. Any ssh 'public-keys' defined in cloud meta-data will be added
and disabled in .ssh/authorized_keys. Any attempts to ssh as this user
using acceptable ssh keys will be presented with a message like the
following:

Please login as the user "ubuntu" rather than the user "youruser".

9 files changed, 337 insertions, 23 deletions

diff --git a/cloudinit/config/cc_ssh.py b/cloudinit/config/cc_ssh.py
index 45204a07..f8f7cb35 100755
--- a/cloudinit/config/cc_ssh.py
+++ b/cloudinit/config/cc_ssh.py
@@ -101,10 +101,6 @@ from cloudinit.distros import ug_util
 from cloudinit import ssh_util
 from cloudinit import util
 
-DISABLE_ROOT_OPTS = (
-    "no-port-forwarding,no-agent-forwarding,"
-    "no-X11-forwarding,command=\"echo \'Please login as the user \\\"$USER\\\""
-    " rather than the user \\\"root\\\".\';echo;sleep 10\"")
 
 GENERATE_KEY_NAMES = ['rsa', 'dsa', 'ecdsa', 'ed25519']
 KEY_FILE_TPL = '/etc/ssh/ssh_host_%s_key'
@@ -185,7 +181,7 @@ def handle(_name, cfg, cloud, log, _args):
         (user, _user_config) = ug_util.extract_default(users)
         disable_root = util.get_cfg_option_bool(cfg, "disable_root", True)
         disable_root_opts = util.get_cfg_option_str(cfg, "disable_root_opts",
-                                                    DISABLE_ROOT_OPTS)
+                                                    ssh_util.DISABLE_USER_OPTS)
 
         keys = cloud.get_public_ssh_keys() or []
         if "ssh_authorized_keys" in cfg:
@@ -207,6 +203,7 @@ def apply_credentials(keys, user, disable_root, disable_root_opts):
         if not user:
             user = "NONE"
         key_prefix = disable_root_opts.replace('$USER', user)
+        key_prefix = key_prefix.replace('$DISABLE_USER', 'root')
     else:
         key_prefix = ''
 
diff --git a/cloudinit/config/cc_users_groups.py b/cloudinit/config/cc_users_groups.py
index c95bdaad..c32a743a 100644
--- a/cloudinit/config/cc_users_groups.py
+++ b/cloudinit/config/cc_users_groups.py
@@ -52,8 +52,17 @@ config keys for an entry in ``users`` are as follows:
       associated with the address, username and SSH keys will be requested from
       there. Default: none
     - ``ssh_authorized_keys``: Optional. List of ssh keys to add to user's
-      authkeys file. Default: none
-    - ``ssh_import_id``: Optional. SSH id to import for user. Default: none
+      authkeys file. Default: none. This key can not be combined with
+      ``ssh_redirect_user``.
+    - ``ssh_import_id``: Optional. SSH id to import for user. Default: none.
+      This key can not be combined with ``ssh_redirect_user``.
+    - ``ssh_redirect_user``: Optional. Boolean set to true to disable SSH
+      logins for this user. When specified, all cloud meta-data public ssh
+      keys will be set up in a disabled state for this username. Any ssh login
+      as this username will timeout and prompt with a message to login instead
+      as the configured <default_username> for this instance. Default: false.
+      This key can not be combined with ``ssh_import_id`` or
+      ``ssh_authorized_keys``.
     - ``sudo``: Optional. Sudo rule to use, list of sudo rules to use or False.
       Default: none. An absence of sudo key, or a value of none or false
       will result in no sudo rules being written for the user.
@@ -101,6 +110,7 @@ config keys for an entry in ``users`` are as follows:
           selinux_user: <selinux username>
           shell: <shell path>
           snapuser: <email>
+          ssh_redirect_user: <true/false>
           ssh_authorized_keys:
               - <key>
               - <key>
@@ -114,17 +124,44 @@ config keys for an entry in ``users`` are as follows:
 # since the module attribute 'distros'
 # is a list of distros that are supported, not a sub-module
 from cloudinit.distros import ug_util
+from cloudinit import log as logging
 
 from cloudinit.settings import PER_INSTANCE
 
+LOG = logging.getLogger(__name__)
+
 frequency = PER_INSTANCE
 
 
 def handle(name, cfg, cloud, _log, _args):
     (users, groups) = ug_util.normalize_users_groups(cfg, cloud.distro)
+    (default_user, _user_config) = ug_util.extract_default(users)
+    cloud_keys = cloud.get_public_ssh_keys() or []
     for (name, members) in groups.items():
         cloud.distro.create_group(name, members)
     for (user, config) in users.items():
+        ssh_redirect_user = config.pop("ssh_redirect_user", False)
+        if ssh_redirect_user:
+            if 'ssh_authorized_keys' in config or 'ssh_import_id' in config:
+                raise ValueError(
+                    'Not creating user %s. ssh_redirect_user cannot be'
+                    ' provided with ssh_import_id or ssh_authorized_keys' %
+                    user)
+            if ssh_redirect_user not in (True, 'default'):
+                raise ValueError(
+                    'Not creating user %s. Invalid value of'
+                    ' ssh_redirect_user: %s. Expected values: true, default'
+                    ' or false.' % (user, ssh_redirect_user))
+            if default_user is None:
+                LOG.warning(
+                    'Ignoring ssh_redirect_user: %s for %s.'
+                    ' No default_user defined.'
+                    ' Perhaps missing cloud configuration users: '
+                    ' [default, ..].',
+                    ssh_redirect_user, user)
+            else:
+                config['ssh_redirect_user'] = default_user
+                config['cloud_public_ssh_keys'] = cloud_keys
         cloud.distro.create_user(user, **config)
 
 # vi: ts=4 expandtab
diff --git a/cloudinit/config/tests/test_ssh.py b/cloudinit/config/tests/test_ssh.py
index 7441d9e9..c8a4271f 100644
--- a/cloudinit/config/tests/test_ssh.py
+++ b/cloudinit/config/tests/test_ssh.py
@@ -2,6 +2,7 @@
 
 
 from cloudinit.config import cc_ssh
+from cloudinit import ssh_util
 from cloudinit.tests.helpers import CiTestCase, mock
 
 MODPATH = "cloudinit.config.cc_ssh."
@@ -15,8 +16,7 @@ class TestHandleSsh(CiTestCase):
         """Apply keys for the given user and root."""
         keys = ["key1"]
         user = "clouduser"
-        options = cc_ssh.DISABLE_ROOT_OPTS
-        cc_ssh.apply_credentials(keys, user, False, options)
+        cc_ssh.apply_credentials(keys, user, False, ssh_util.DISABLE_USER_OPTS)
         self.assertEqual([mock.call(set(keys), user),
                           mock.call(set(keys), "root", options="")],
                          m_setup_keys.call_args_list)
@@ -25,8 +25,7 @@ class TestHandleSsh(CiTestCase):
         """Apply keys for root only."""
         keys = ["key1"]
         user = None
-        options = cc_ssh.DISABLE_ROOT_OPTS
-        cc_ssh.apply_credentials(keys, user, False, options)
+        cc_ssh.apply_credentials(keys, user, False, ssh_util.DISABLE_USER_OPTS)
         self.assertEqual([mock.call(set(keys), "root", options="")],
                          m_setup_keys.call_args_list)
 
@@ -34,9 +33,10 @@ class TestHandleSsh(CiTestCase):
         """Apply keys for the given user and disable root ssh."""
         keys = ["key1"]
         user = "clouduser"
-        options = cc_ssh.DISABLE_ROOT_OPTS
+        options = ssh_util.DISABLE_USER_OPTS
         cc_ssh.apply_credentials(keys, user, True, options)
         options = options.replace("$USER", user)
+        options = options.replace("$DISABLE_USER", "root")
         self.assertEqual([mock.call(set(keys), user),
                           mock.call(set(keys), "root", options=options)],
                          m_setup_keys.call_args_list)
@@ -45,9 +45,10 @@ class TestHandleSsh(CiTestCase):
         """Apply keys no user and disable root ssh."""
         keys = ["key1"]
         user = None
-        options = cc_ssh.DISABLE_ROOT_OPTS
+        options = ssh_util.DISABLE_USER_OPTS
         cc_ssh.apply_credentials(keys, user, True, options)
         options = options.replace("$USER", "NONE")
+        options = options.replace("$DISABLE_USER", "root")
         self.assertEqual([mock.call(set(keys), "root", options=options)],
                          m_setup_keys.call_args_list)
 
@@ -66,7 +67,8 @@ class TestHandleSsh(CiTestCase):
         cloud = self.tmp_cloud(
             distro='ubuntu', metadata={'public-keys': keys})
         cc_ssh.handle("name", cfg, cloud, None, None)
-        options = cc_ssh.DISABLE_ROOT_OPTS.replace("$USER", "NONE")
+        options = ssh_util.DISABLE_USER_OPTS.replace("$USER", "NONE")
+        options = options.replace("$DISABLE_USER", "root")
         m_glob.assert_called_once_with('/etc/ssh/ssh_host_*key*')
         self.assertIn(
             [mock.call('/etc/ssh/ssh_host_rsa_key'),
@@ -94,7 +96,8 @@ class TestHandleSsh(CiTestCase):
             distro='ubuntu', metadata={'public-keys': keys})
         cc_ssh.handle("name", cfg, cloud, None, None)
 
-        options = cc_ssh.DISABLE_ROOT_OPTS.replace("$USER", user)
+        options = ssh_util.DISABLE_USER_OPTS.replace("$USER", user)
+        options = options.replace("$DISABLE_USER", "root")
         self.assertEqual([mock.call(set(keys), user),
                           mock.call(set(keys), "root", options=options)],
                          m_setup_keys.call_args_list)
@@ -118,7 +121,8 @@ class TestHandleSsh(CiTestCase):
             distro='ubuntu', metadata={'public-keys': keys})
         cc_ssh.handle("name", cfg, cloud, None, None)
 
-        options = cc_ssh.DISABLE_ROOT_OPTS.replace("$USER", user)
+        options = ssh_util.DISABLE_USER_OPTS.replace("$USER", user)
+        options = options.replace("$DISABLE_USER", "root")
         self.assertEqual([mock.call(set(keys), user),
                           mock.call(set(keys), "root", options=options)],
                          m_setup_keys.call_args_list)
diff --git a/cloudinit/config/tests/test_users_groups.py b/cloudinit/config/tests/test_users_groups.py
new file mode 100644
index 00000000..ba0afae3
--- /dev/null
+++ b/cloudinit/config/tests/test_users_groups.py
@@ -0,0 +1,144 @@
+# This file is part of cloud-init. See LICENSE file for license information.
+
+
+from cloudinit.config import cc_users_groups
+from cloudinit.tests.helpers import CiTestCase, mock
+
+MODPATH = "cloudinit.config.cc_users_groups"
+
+
+@mock.patch('cloudinit.distros.ubuntu.Distro.create_group')
+@mock.patch('cloudinit.distros.ubuntu.Distro.create_user')
+class TestHandleUsersGroups(CiTestCase):
+    """Test cc_users_groups handling of config."""
+
+    with_logs = True
+
+    def test_handle_no_cfg_creates_no_users_or_groups(self, m_user, m_group):
+        """Test handle with no config will not create users or groups."""
+        cfg = {}  # merged cloud-config
+        # System config defines a default user for the distro.
+        sys_cfg = {'default_user': {'name': 'ubuntu', 'lock_passwd': True,
+                                    'groups': ['lxd', 'sudo'],
+                                    'shell': '/bin/bash'}}
+        metadata = {}
+        cloud = self.tmp_cloud(
+            distro='ubuntu', sys_cfg=sys_cfg, metadata=metadata)
+        cc_users_groups.handle('modulename', cfg, cloud, None, None)
+        m_user.assert_not_called()
+        m_group.assert_not_called()
+
+    def test_handle_users_in_cfg_calls_create_users(self, m_user, m_group):
+        """When users in config, create users with distro.create_user."""
+        cfg = {'users': ['default', {'name': 'me2'}]}  # merged cloud-config
+        # System config defines a default user for the distro.
+        sys_cfg = {'default_user': {'name': 'ubuntu', 'lock_passwd': True,
+                                    'groups': ['lxd', 'sudo'],
+                                    'shell': '/bin/bash'}}
+        metadata = {}
+        cloud = self.tmp_cloud(
+            distro='ubuntu', sys_cfg=sys_cfg, metadata=metadata)
+        cc_users_groups.handle('modulename', cfg, cloud, None, None)
+        self.assertItemsEqual(
+            m_user.call_args_list,
+            [mock.call('ubuntu', groups='lxd,sudo', lock_passwd=True,
+                       shell='/bin/bash'),
+             mock.call('me2', default=False)])
+        m_group.assert_not_called()
+
+    def test_users_with_ssh_redirect_user_passes_keys(self, m_user, m_group):
+        """When ssh_redirect_user is True pass default user and cloud keys."""
+        cfg = {
+            'users': ['default', {'name': 'me2', 'ssh_redirect_user': True}]}
+        # System config defines a default user for the distro.
+        sys_cfg = {'default_user': {'name': 'ubuntu', 'lock_passwd': True,
+                                    'groups': ['lxd', 'sudo'],
+                                    'shell': '/bin/bash'}}
+        metadata = {'public-keys': ['key1']}
+        cloud = self.tmp_cloud(
+            distro='ubuntu', sys_cfg=sys_cfg, metadata=metadata)
+        cc_users_groups.handle('modulename', cfg, cloud, None, None)
+        self.assertItemsEqual(
+            m_user.call_args_list,
+            [mock.call('ubuntu', groups='lxd,sudo', lock_passwd=True,
+                       shell='/bin/bash'),
+             mock.call('me2', cloud_public_ssh_keys=['key1'], default=False,
+                       ssh_redirect_user='ubuntu')])
+        m_group.assert_not_called()
+
+    def test_users_with_ssh_redirect_user_default_str(self, m_user, m_group):
+        """When ssh_redirect_user is 'default' pass default username."""
+        cfg = {
+            'users': ['default', {'name': 'me2',
+                                  'ssh_redirect_user': 'default'}]}
+        # System config defines a default user for the distro.
+        sys_cfg = {'default_user': {'name': 'ubuntu', 'lock_passwd': True,
+                                    'groups': ['lxd', 'sudo'],
+                                    'shell': '/bin/bash'}}
+        metadata = {'public-keys': ['key1']}
+        cloud = self.tmp_cloud(
+            distro='ubuntu', sys_cfg=sys_cfg, metadata=metadata)
+        cc_users_groups.handle('modulename', cfg, cloud, None, None)
+        self.assertItemsEqual(
+            m_user.call_args_list,
+            [mock.call('ubuntu', groups='lxd,sudo', lock_passwd=True,
+                       shell='/bin/bash'),
+             mock.call('me2', cloud_public_ssh_keys=['key1'], default=False,
+                       ssh_redirect_user='ubuntu')])
+        m_group.assert_not_called()
+
+    def test_users_with_ssh_redirect_user_non_default(self, m_user, m_group):
+        """Warn when ssh_redirect_user is not 'default'."""
+        cfg = {
+            'users': ['default', {'name': 'me2',
+                                  'ssh_redirect_user': 'snowflake'}]}
+        # System config defines a default user for the distro.
+        sys_cfg = {'default_user': {'name': 'ubuntu', 'lock_passwd': True,
+                                    'groups': ['lxd', 'sudo'],
+                                    'shell': '/bin/bash'}}
+        metadata = {'public-keys': ['key1']}
+        cloud = self.tmp_cloud(
+            distro='ubuntu', sys_cfg=sys_cfg, metadata=metadata)
+        with self.assertRaises(ValueError) as context_manager:
+            cc_users_groups.handle('modulename', cfg, cloud, None, None)
+        m_group.assert_not_called()
+        self.assertEqual(
+            'Not creating user me2. Invalid value of ssh_redirect_user:'
+            ' snowflake. Expected values: true, default or false.',
+            str(context_manager.exception))
+
+    def test_users_with_ssh_redirect_user_default_false(self, m_user, m_group):
+        """When unspecified ssh_redirect_user is false and not set up."""
+        cfg = {'users': ['default', {'name': 'me2'}]}
+        # System config defines a default user for the distro.
+        sys_cfg = {'default_user': {'name': 'ubuntu', 'lock_passwd': True,
+                                    'groups': ['lxd', 'sudo'],
+                                    'shell': '/bin/bash'}}
+        metadata = {'public-keys': ['key1']}
+        cloud = self.tmp_cloud(
+            distro='ubuntu', sys_cfg=sys_cfg, metadata=metadata)
+        cc_users_groups.handle('modulename', cfg, cloud, None, None)
+        self.assertItemsEqual(
+            m_user.call_args_list,
+            [mock.call('ubuntu', groups='lxd,sudo', lock_passwd=True,
+                       shell='/bin/bash'),
+             mock.call('me2', default=False)])
+        m_group.assert_not_called()
+
+    def test_users_ssh_redirect_user_and_no_default(self, m_user, m_group):
+        """Warn when ssh_redirect_user is True and no default user present."""
+        cfg = {
+            'users': ['default', {'name': 'me2', 'ssh_redirect_user': True}]}
+        # System config defines *no* default user for the distro.
+        sys_cfg = {}
+        metadata = {}  # no public-keys defined
+        cloud = self.tmp_cloud(
+            distro='ubuntu', sys_cfg=sys_cfg, metadata=metadata)
+        cc_users_groups.handle('modulename', cfg, cloud, None, None)
+        m_user.assert_called_once_with('me2', default=False)
+        m_group.assert_not_called()
+        self.assertEqual(
+            'WARNING: Ignoring ssh_redirect_user: True for me2. No'
+            ' default_user defined. Perhaps missing'
+            ' cloud configuration users:  [default, ..].\n',
+            self.logs.getvalue())
diff --git a/cloudinit/distros/__init__.py b/cloudinit/distros/__init__.py
index d9101ce6..b8a48e85 100755..100644
--- a/cloudinit/distros/__init__.py
+++ b/cloudinit/distros/__init__.py
@@ -381,6 +381,9 @@ class Distro(object):
         """
         Add a user to the system using standard GNU tools
         """
+        # XXX need to make add_user idempotent somehow as we
+        # still want to add groups or modify ssh keys on pre-existing
+        # users in the image.
         if util.is_user(name):
             LOG.info("User %s already exists, skipping.", name)
             return
@@ -547,10 +550,24 @@ class Distro(object):
                     LOG.warning("Invalid type '%s' detected for"
                                 " 'ssh_authorized_keys', expected list,"
                                 " string, dict, or set.", type(keys))
+                    keys = []
                 else:
                     keys = set(keys) or []
-                    ssh_util.setup_user_keys(keys, name, options=None)
-
+            ssh_util.setup_user_keys(set(keys), name)
+        if 'ssh_redirect_user' in kwargs:
+            cloud_keys = kwargs.get('cloud_public_ssh_keys', [])
+            if not cloud_keys:
+                LOG.warning(
+                    'Unable to disable ssh logins for %s given'
+                    ' ssh_redirect_user: %s. No cloud public-keys present.',
+                    name, kwargs['ssh_redirect_user'])
+            else:
+                redirect_user = kwargs['ssh_redirect_user']
+                disable_option = ssh_util.DISABLE_USER_OPTS
+                disable_option = disable_option.replace('$USER', redirect_user)
+                disable_option = disable_option.replace('$DISABLE_USER', name)
+                ssh_util.setup_user_keys(
+                    set(cloud_keys), name, options=disable_option)
         return True
 
     def lock_passwd(self, name):
diff --git a/cloudinit/ssh_util.py b/cloudinit/ssh_util.py
index 73c31772..3f99b58c 100644
--- a/cloudinit/ssh_util.py
+++ b/cloudinit/ssh_util.py
@@ -41,6 +41,12 @@ VALID_KEY_TYPES = (
 )
 
 
+DISABLE_USER_OPTS = (
+    "no-port-forwarding,no-agent-forwarding,"
+    "no-X11-forwarding,command=\"echo \'Please login as the user \\\"$USER\\\""
+    " rather than the user \\\"$DISABLE_USER\\\".\';echo;sleep 10\"")
+
+
 class AuthKeyLine(object):
     def __init__(self, source, keytype=None, base64=None,
                  comment=None, options=None):
diff --git a/doc/examples/cloud-config-user-groups.txt b/doc/examples/cloud-config-user-groups.txt
index 01ecad7b..6a363b77 100644
--- a/doc/examples/cloud-config-user-groups.txt
+++ b/doc/examples/cloud-config-user-groups.txt
@@ -36,6 +36,8 @@ users:
       - <ssh pub key 1>
       - <ssh pub key 2>
   - snapuser: joe@joeuser.io
+  - name: nosshlogins
+    ssh_redirect_user: true
 
 # Valid Values:
 #   name: The user's login name
@@ -76,6 +78,13 @@ users:
 #   no_log_init: When set to true, do not initialize lastlog and faillog database.
 #   ssh_import_id: Optional. Import SSH ids
 #   ssh_authorized_keys: Optional. [list] Add keys to user's authorized keys file
+#   ssh_redirect_user: Optional. [bool] Set true to block ssh logins for cloud
+#       ssh public keys and emit a message redirecting logins to
+#       use <default_username> instead. This option only disables cloud
+#       provided public-keys. An error will be raised if ssh_authorized_keys
+#       or ssh_import_id is provided for the same user.
+# 
+#       ssh_authorized_keys.
 #   sudo: Defaults to none. Accepts a sudo rule string, a list of sudo rule
 #         strings or False to explicitly deny sudo usage. Examples:
 #
diff --git a/doc/examples/cloud-config.txt b/doc/examples/cloud-config.txt
index 774f66b9..eb84dcf5 100644
--- a/doc/examples/cloud-config.txt
+++ b/doc/examples/cloud-config.txt
@@ -232,9 +232,22 @@ disable_root: false
 # respective key in /root/.ssh/authorized_keys if disable_root is true
 # see 'man authorized_keys' for more information on what you can do here
 #
-# The string '$USER' will be replaced with the username of the default user
-#
-# disable_root_opts: no-port-forwarding,no-agent-forwarding,no-X11-forwarding,command="echo 'Please login as the user \"$USER\" rather than the user \"root\".';echo;sleep 10"
+# The string '$USER' will be replaced with the username of the default user.
+# The string '$DISABLE_USER' will be replaced with the username to disable.
+#
+# disable_root_opts: no-port-forwarding,no-agent-forwarding,no-X11-forwarding,command="echo 'Please login as the user \"$USER\" rather than the user \"$DISABLE_USER\".';echo;sleep 10"
+
+# disable ssh access for non-root-users
+# To disable ssh access for non-root users, ssh_redirect_user: true can be
+# provided for any use in the 'users' list. This will prompt any ssh login
+# attempts as that user with a message like that in disable_root_opts which
+# redirects the person to login as <default_username>
+# This option can not be combined with either ssh_authorized_keys or
+# ssh_import_id.
+users:
+ - default
+ - name: blockeduser
+   ssh_redirect_user: true
 
 
 # set the locale to a given locale
diff --git a/tests/unittests/test_distros/test_create_users.py b/tests/unittests/test_distros/test_create_users.py
index 07176caa..c3f258d5 100644
--- a/tests/unittests/test_distros/test_create_users.py
+++ b/tests/unittests/test_distros/test_create_users.py
@@ -1,7 +1,10 @@
 # This file is part of cloud-init. See LICENSE file for license information.
 
+import re
+
 from cloudinit import distros
-from cloudinit.tests.helpers import (TestCase, mock)
+from cloudinit import ssh_util
+from cloudinit.tests.helpers import (CiTestCase, mock)
 
 
 class MyBaseDistro(distros.Distro):
@@ -44,8 +47,12 @@ class MyBaseDistro(distros.Distro):
 
 @mock.patch("cloudinit.distros.util.system_is_snappy", return_value=False)
 @mock.patch("cloudinit.distros.util.subp")
-class TestCreateUser(TestCase):
+class TestCreateUser(CiTestCase):
+
+    with_logs = True
+
     def setUp(self):
+        super(TestCreateUser, self).setUp()
         self.dist = MyBaseDistro()
 
     def _useradd2call(self, args):
@@ -153,4 +160,84 @@ class TestCreateUser(TestCase):
             [self._useradd2call([user, '-m']),
              mock.call(['passwd', '-l', user])])
 
+    @mock.patch('cloudinit.ssh_util.setup_user_keys')
+    def test_setup_ssh_authorized_keys_with_string(
+            self, m_setup_user_keys, m_subp, m_is_snappy):
+        """ssh_authorized_keys allows string and calls setup_user_keys."""
+        user = 'foouser'
+        self.dist.create_user(user, ssh_authorized_keys='mykey')
+        self.assertEqual(
+            m_subp.call_args_list,
+            [self._useradd2call([user, '-m']),
+             mock.call(['passwd', '-l', user])])
+        m_setup_user_keys.assert_called_once_with(set(['mykey']), user)
+
+    @mock.patch('cloudinit.ssh_util.setup_user_keys')
+    def test_setup_ssh_authorized_keys_with_list(
+            self, m_setup_user_keys, m_subp, m_is_snappy):
+        """ssh_authorized_keys allows lists and calls setup_user_keys."""
+        user = 'foouser'
+        self.dist.create_user(user, ssh_authorized_keys=['key1', 'key2'])
+        self.assertEqual(
+            m_subp.call_args_list,
+            [self._useradd2call([user, '-m']),
+             mock.call(['passwd', '-l', user])])
+        m_setup_user_keys.assert_called_once_with(set(['key1', 'key2']), user)
+
+    @mock.patch('cloudinit.ssh_util.setup_user_keys')
+    def test_setup_ssh_authorized_keys_with_integer(
+            self, m_setup_user_keys, m_subp, m_is_snappy):
+        """ssh_authorized_keys warns on non-iterable/string type."""
+        user = 'foouser'
+        self.dist.create_user(user, ssh_authorized_keys=-1)
+        m_setup_user_keys.assert_called_once_with(set([]), user)
+        match = re.match(
+            r'.*WARNING: Invalid type \'<(type|class) \'int\'>\' detected for'
+            ' \'ssh_authorized_keys\'.*',
+            self.logs.getvalue(),
+            re.DOTALL)
+        self.assertIsNotNone(
+            match, 'Missing ssh_authorized_keys invalid type warning')
+
+    @mock.patch('cloudinit.ssh_util.setup_user_keys')
+    def test_create_user_with_ssh_redirect_user_no_cloud_keys(
+            self, m_setup_user_keys, m_subp, m_is_snappy):
+        """Log a warning when trying to redirect a user no cloud ssh keys."""
+        user = 'foouser'
+        self.dist.create_user(user, ssh_redirect_user='someuser')
+        self.assertIn(
+            'WARNING: Unable to disable ssh logins for foouser given '
+            'ssh_redirect_user: someuser. No cloud public-keys present.\n',
+            self.logs.getvalue())
+        m_setup_user_keys.assert_not_called()
+
+    @mock.patch('cloudinit.ssh_util.setup_user_keys')
+    def test_create_user_with_ssh_redirect_user_with_cloud_keys(
+            self, m_setup_user_keys, m_subp, m_is_snappy):
+        """Disable ssh when ssh_redirect_user and cloud ssh keys are set."""
+        user = 'foouser'
+        self.dist.create_user(
+            user, ssh_redirect_user='someuser', cloud_public_ssh_keys=['key1'])
+        disable_prefix = ssh_util.DISABLE_USER_OPTS
+        disable_prefix = disable_prefix.replace('$USER', 'someuser')
+        disable_prefix = disable_prefix.replace('$DISABLE_USER', user)
+        m_setup_user_keys.assert_called_once_with(
+            set(['key1']), 'foouser', options=disable_prefix)
+
+    @mock.patch('cloudinit.ssh_util.setup_user_keys')
+    def test_create_user_with_ssh_redirect_user_does_not_disable_auth_keys(
+            self, m_setup_user_keys, m_subp, m_is_snappy):
+        """Do not disable ssh_authorized_keys when ssh_redirect_user is set."""
+        user = 'foouser'
+        self.dist.create_user(
+            user, ssh_authorized_keys='auth1', ssh_redirect_user='someuser',
+            cloud_public_ssh_keys=['key1'])
+        disable_prefix = ssh_util.DISABLE_USER_OPTS
+        disable_prefix = disable_prefix.replace('$USER', 'someuser')
+        disable_prefix = disable_prefix.replace('$DISABLE_USER', user)
+        self.assertEqual(
+            m_setup_user_keys.call_args_list,
+            [mock.call(set(['auth1']), user),  # not disabled
+             mock.call(set(['key1']), 'foouser', options=disable_prefix)])
+
 # vi: ts=4 expandtab