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".

6 files changed, 223 insertions, 18 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):