cli: add --system param to allow validating system user-data on a machine

cloudinit/config/schema.py

@@ -1,6 +1,7 @@
 # This file is part of cloud-init. See LICENSE file for license information.
 """schema.py: Set of module functions for processing cloud-config schema."""
 
+from cloudinit.cmd.devel import read_cfg_paths
 from cloudinit import importer
 from cloudinit.util import find_modules, load_file
 
@@ -173,17 +174,33 @@ def annotated_cloudconfig_file(cloudconfig, original_content, schema_errors):
 def validate_cloudconfig_file(config_path, schema, annotate=False):
     """Validate cloudconfig file adheres to a specific jsonschema.
 
-    @param config_path: Path to the yaml cloud-config file to parse.
+    @param config_path: Path to the yaml cloud-config file to parse, or None
+        to default to system userdata from Paths object.
     @param schema: Dict describing a valid jsonschema to validate against.
     @param annotate: Boolean set True to print original config file with error
         annotations on the offending lines.
 
     @raises SchemaValidationError containing any of schema_errors encountered.
     @raises RuntimeError when config_path does not exist.
     """
-    if not os.path.exists(config_path):
-        raise RuntimeError('Configfile {0} does not exist'.format(config_path))
-    content = load_file(config_path, decode=False)
+    if config_path is None:
+        # Use system's raw userdata path
+        if os.getuid() != 0:
+            raise RuntimeError(
+                "Unable to read system userdata as non-root user."
+                " Try using sudo"
+            )
+        paths = read_cfg_paths()
+        user_data_file = paths.get_ipath_cur("userdata_raw")
+        content = load_file(user_data_file, decode=False)
+    else:
+        if not os.path.exists(config_path):
+            raise RuntimeError(
+                'Configfile {0} does not exist'.format(
+                    config_path
+                )
+            )
+        content = load_file(config_path, decode=False)
     if not content.startswith(CLOUD_CONFIG_HEADER):
         errors = (
             ('format-l1.c1', 'File {0} needs to begin with "{1}"'.format(
@@ -425,6 +442,8 @@ def get_parser(parser=None):
             description='Validate cloud-config files or document schema')
     parser.add_argument('-c', '--config-file',
                         help='Path of the cloud-config yaml file to validate')
+    parser.add_argument('--system', action='store_true', default=False,
+                        help='Validate the system cloud-config userdata')
     parser.add_argument('-d', '--docs', nargs='+',
                         help=('Print schema module docs. Choices: all or'
                               ' space-delimited cc_names.'))
@@ -435,11 +454,11 @@ def get_parser(parser=None):
 
 def handle_schema_args(name, args):
     """Handle provided schema args and perform the appropriate actions."""
-    exclusive_args = [args.config_file, args.docs]
-    if not any(exclusive_args) or all(exclusive_args):
-        error('Expected either --config-file argument or --docs')
+    exclusive_args = [args.config_file, args.docs, args.system]
+    if len([arg for arg in exclusive_args if arg]) != 1:
+        error('Expected one of --config-file, --system or --docs arguments')
     full_schema = get_schema()
-    if args.config_file:
+    if args.config_file or args.system:
         try:
             validate_cloudconfig_file(
                 args.config_file, full_schema, args.annotate)
@@ -449,7 +468,11 @@ def handle_schema_args(name, args):
         except RuntimeError as e:
             error(str(e))
         else:
-            print("Valid cloud-config file {0}".format(args.config_file))
+            if args.config_file is None:
+                cfg_name = "system userdata"
+            else:
+                cfg_name = args.config_file
+            print("Valid cloud-config:", cfg_name)
     elif args.docs:
         schema_ids = [subschema['id'] for subschema in full_schema['allOf']]
         schema_ids += ['all']

doc/rtd/topics/faq.rst

@@ -141,12 +141,12 @@ that can validate your user data offline.
 
 .. _validate-yaml.py: https://github.com/canonical/cloud-init/blob/master/tools/validate-yaml.py
 
-Another option is to run the following on an instance when debugging:
+Another option is to run the following on an instance to debug userdata
+provided to the system:
 
 .. code-block:: shell-session
 
-    $ sudo cloud-init query userdata > user-data.yaml
-    $ cloud-init devel schema -c user-data.yaml --annotate
+    $ cloud-init devel schema --system --annotate
 
 As launching instances in the cloud can cost money and take a bit longer,
 sometimes it is easier to launch instances locally using Multipass or LXD:

tests/unittests/test_cli.py

@@ -214,7 +214,7 @@ def test_wb_devel_schema_subcommand_parser(self):
         self.assertEqual(1, exit_code)
         # Known whitebox output from schema subcommand
         self.assertEqual(
-            'Expected either --config-file argument or --docs\n',
+            'Expected one of --config-file, --system or --docs arguments\n',
             self.stderr.getvalue())
 
     def test_wb_devel_schema_subcommand_doc_content(self):

tests/unittests/test_handler/test_schema.py

@@ -9,9 +9,9 @@
 from cloudinit.tests.helpers import CiTestCase, mock, skipUnlessJsonSchema
 
 from copy import copy
+import itertools
 import os
 import pytest
-from io import StringIO
 from pathlib import Path
 from textwrap import dedent
 from yaml import safe_load
@@ -400,50 +400,97 @@ def test_annotated_cloudconfig_file_annotates_separate_line_items(self):
             annotated_cloudconfig_file(parsed_config, content, schema_errors))
 
 
-class MainTest(CiTestCase):
+class TestMain:
 
-    def test_main_missing_args(self):
+    exclusive_combinations = itertools.combinations(
+        ["--system", "--docs all", "--config-file something"], 2
+    )
+
+    @pytest.mark.parametrize("params", exclusive_combinations)
+    def test_main_exclusive_args(self, params, capsys):
+        """Main exits non-zero and error on required exclusive args."""
+        params = list(itertools.chain(*[a.split() for a in params]))
+        with mock.patch('sys.argv', ['mycmd'] + params):
+            with pytest.raises(SystemExit) as context_manager:
+                main()
+        assert 1 == context_manager.value.code
+
+        _out, err = capsys.readouterr()
+        expected = (
+            'Expected one of --config-file, --system or --docs arguments\n'
+        )
+        assert expected == err
+
+    def test_main_missing_args(self, capsys):
         """Main exits non-zero and reports an error on missing parameters."""
         with mock.patch('sys.argv', ['mycmd']):
-            with mock.patch('sys.stderr', new_callable=StringIO) as m_stderr:
-                with self.assertRaises(SystemExit) as context_manager:
-                    main()
-        self.assertEqual(1, context_manager.exception.code)
-        self.assertEqual(
-            'Expected either --config-file argument or --docs\n',
-            m_stderr.getvalue())
+            with pytest.raises(SystemExit) as context_manager:
+                main()
+        assert 1 == context_manager.value.code
+
+        _out, err = capsys.readouterr()
+        expected = (
+            'Expected one of --config-file, --system or --docs arguments\n'
+        )
+        assert expected == err
 
-    def test_main_absent_config_file(self):
+    def test_main_absent_config_file(self, capsys):
         """Main exits non-zero when config file is absent."""
         myargs = ['mycmd', '--annotate', '--config-file', 'NOT_A_FILE']
         with mock.patch('sys.argv', myargs):
-            with mock.patch('sys.stderr', new_callable=StringIO) as m_stderr:
-                with self.assertRaises(SystemExit) as context_manager:
-                    main()
-        self.assertEqual(1, context_manager.exception.code)
-        self.assertEqual(
-            'Configfile NOT_A_FILE does not exist\n',
-            m_stderr.getvalue())
+            with pytest.raises(SystemExit) as context_manager:
+                main()
+        assert 1 == context_manager.value.code
+        _out, err = capsys.readouterr()
+        assert 'Configfile NOT_A_FILE does not exist\n' == err
 
-    def test_main_prints_docs(self):
+    def test_main_prints_docs(self, capsys):
         """When --docs parameter is provided, main generates documentation."""
         myargs = ['mycmd', '--docs', 'all']
         with mock.patch('sys.argv', myargs):
-            with mock.patch('sys.stdout', new_callable=StringIO) as m_stdout:
-                self.assertEqual(0, main(), 'Expected 0 exit code')
-        self.assertIn('\nNTP\n---\n', m_stdout.getvalue())
-        self.assertIn('\nRuncmd\n------\n', m_stdout.getvalue())
+            assert 0 == main(), 'Expected 0 exit code'
+        out, _err = capsys.readouterr()
+        assert '\nNTP\n---\n' in out
+        assert '\nRuncmd\n------\n' in out
 
-    def test_main_validates_config_file(self):
+    def test_main_validates_config_file(self, tmpdir, capsys):
         """When --config-file parameter is provided, main validates schema."""
-        myyaml = self.tmp_path('my.yaml')
-        myargs = ['mycmd', '--config-file', myyaml]
-        write_file(myyaml, b'#cloud-config\nntp:')  # shortest ntp schema
+        myyaml = tmpdir.join('my.yaml')
+        myargs = ['mycmd', '--config-file', myyaml.strpath]
+        myyaml.write(b'#cloud-config\nntp:')  # shortest ntp schema
         with mock.patch('sys.argv', myargs):
-            with mock.patch('sys.stdout', new_callable=StringIO) as m_stdout:
-                self.assertEqual(0, main(), 'Expected 0 exit code')
-        self.assertIn(
-            'Valid cloud-config file {0}'.format(myyaml), m_stdout.getvalue())
+            assert 0 == main(), 'Expected 0 exit code'
+        out, _err = capsys.readouterr()
+        assert 'Valid cloud-config: {0}\n'.format(myyaml) == out
+
+    @mock.patch('cloudinit.config.schema.read_cfg_paths')
+    @mock.patch('cloudinit.config.schema.os.getuid', return_value=0)
+    def test_main_validates_system_userdata(
+        self, m_getuid, m_read_cfg_paths, capsys, paths
+    ):
+        """When --system is provided, main validates system userdata."""
+        m_read_cfg_paths.return_value = paths
+        ud_file = paths.get_ipath_cur("userdata_raw")
+        write_file(ud_file, b'#cloud-config\nntp:')
+        myargs = ['mycmd', '--system']
+        with mock.patch('sys.argv', myargs):
+            assert 0 == main(), 'Expected 0 exit code'
+        out, _err = capsys.readouterr()
+        assert 'Valid cloud-config: system userdata\n' == out
+
+    @mock.patch('cloudinit.config.schema.os.getuid', return_value=1000)
+    def test_main_system_userdata_requires_root(self, m_getuid, capsys, paths):
+        """Non-root user can't use --system param"""
+        myargs = ['mycmd', '--system']
+        with mock.patch('sys.argv', myargs):
+            with pytest.raises(SystemExit) as context_manager:
+                main()
+        assert 1 == context_manager.value.code
+        _out, err = capsys.readouterr()
+        expected = (
+            'Unable to read system userdata as non-root user. Try using sudo\n'
+        )
+        assert expected == err
 
 
 class CloudTestsIntegrationTest(CiTestCase):