logs: collect-logs ignore instance-data-sensitive.json on non-root user

Since /run/cloud-init/instance-data-sensitive.json is root read-only,
ignore this file if non-root user runs collect-logs.

If --include-userdata is provided on the command line, exit in error
if non-root user attempts this operation.

Lastly, update the __main__ to exit based on return value of main.

LP: #1805201

2 files changed, 60 insertions, 14 deletions

diff --git a/cloudinit/cmd/devel/logs.py b/cloudinit/cmd/devel/logs.py
index df725204..4c086b51 100644
--- a/cloudinit/cmd/devel/logs.py
+++ b/cloudinit/cmd/devel/logs.py
@@ -5,14 +5,16 @@
 """Define 'collect-logs' utility and handler to include in cloud-init cmd."""
 
 import argparse
-from cloudinit.util import (
-    ProcessExecutionError, chdir, copy, ensure_dir, subp, write_file)
-from cloudinit.temp_utils import tempdir
 from datetime import datetime
 import os
 import shutil
 import sys
 
+from cloudinit.sources import INSTANCE_JSON_SENSITIVE_FILE
+from cloudinit.temp_utils import tempdir
+from cloudinit.util import (
+    ProcessExecutionError, chdir, copy, ensure_dir, subp, write_file)
+
 
 CLOUDINIT_LOGS = ['/var/log/cloud-init.log', '/var/log/cloud-init-output.log']
 CLOUDINIT_RUN_DIR = '/run/cloud-init'
@@ -46,6 +48,13 @@ def get_parser(parser=None):
     return parser
 
 
+def _copytree_ignore_sensitive_files(curdir, files):
+    """Return a list of files to ignore if we are non-root"""
+    if os.getuid() == 0:
+        return ()
+    return (INSTANCE_JSON_SENSITIVE_FILE,)  # Ignore root-permissioned files
+
+
 def _write_command_output_to_file(cmd, filename, msg, verbosity):
     """Helper which runs a command and writes output or error to filename."""
     try:
@@ -78,6 +87,11 @@ def collect_logs(tarfile, include_userdata, verbosity=0):
     @param tarfile: The path of the tar-gzipped file to create.
     @param include_userdata: Boolean, true means include user-data.
     """
+    if include_userdata and os.getuid() != 0:
+        sys.stderr.write(
+            "To include userdata, root user is required."
+            " Try sudo cloud-init collect-logs\n")
+        return 1
     tarfile = os.path.abspath(tarfile)
     date = datetime.utcnow().date().strftime('%Y-%m-%d')
     log_dir = 'cloud-init-logs-{0}'.format(date)
@@ -110,7 +124,8 @@ def collect_logs(tarfile, include_userdata, verbosity=0):
         ensure_dir(run_dir)
         if os.path.exists(CLOUDINIT_RUN_DIR):
             shutil.copytree(CLOUDINIT_RUN_DIR,
-                            os.path.join(run_dir, 'cloud-init'))
+                            os.path.join(run_dir, 'cloud-init'),
+                            ignore=_copytree_ignore_sensitive_files)
             _debug("collected dir %s\n" % CLOUDINIT_RUN_DIR, 1, verbosity)
         else:
             _debug("directory '%s' did not exist\n" % CLOUDINIT_RUN_DIR, 1,
@@ -118,21 +133,21 @@ def collect_logs(tarfile, include_userdata, verbosity=0):
         with chdir(tmp_dir):
             subp(['tar', 'czvf', tarfile, log_dir.replace(tmp_dir + '/', '')])
     sys.stderr.write("Wrote %s\n" % tarfile)
+    return 0
 
 
 def handle_collect_logs_args(name, args):
     """Handle calls to 'cloud-init collect-logs' as a subcommand."""
-    collect_logs(args.tarfile, args.userdata, args.verbosity)
+    return collect_logs(args.tarfile, args.userdata, args.verbosity)
 
 
 def main():
     """Tool to collect and tar all cloud-init related logs."""
     parser = get_parser()
-    handle_collect_logs_args('collect-logs', parser.parse_args())
-    return 0
+    return handle_collect_logs_args('collect-logs', parser.parse_args())
 
 
 if __name__ == '__main__':
-    main()
+    sys.exit(main())
 
 # vi: ts=4 expandtab
diff --git a/cloudinit/cmd/devel/tests/test_logs.py b/cloudinit/cmd/devel/tests/test_logs.py
index 98b47560..4951797b 100644
--- a/cloudinit/cmd/devel/tests/test_logs.py
+++ b/cloudinit/cmd/devel/tests/test_logs.py
@@ -1,13 +1,17 @@
 # This file is part of cloud-init. See LICENSE file for license information.
 
-from cloudinit.cmd.devel import logs
-from cloudinit.util import ensure_dir, load_file, subp, write_file
-from cloudinit.tests.helpers import FilesystemMockingTestCase, wrap_and_call
 from datetime import datetime
-import mock
 import os
+from six import StringIO
+
+from cloudinit.cmd.devel import logs
+from cloudinit.sources import INSTANCE_JSON_SENSITIVE_FILE
+from cloudinit.tests.helpers import (
+    FilesystemMockingTestCase, mock, wrap_and_call)
+from cloudinit.util import ensure_dir, load_file, subp, write_file
 
 
+@mock.patch('cloudinit.cmd.devel.logs.os.getuid')
 class TestCollectLogs(FilesystemMockingTestCase):
 
     def setUp(self):
@@ -15,14 +19,29 @@ class TestCollectLogs(FilesystemMockingTestCase):
         self.new_root = self.tmp_dir()
         self.run_dir = self.tmp_path('run', self.new_root)
 
-    def test_collect_logs_creates_tarfile(self):
+    def test_collect_logs_with_userdata_requires_root_user(self, m_getuid):
+        """collect-logs errors when non-root user collects userdata ."""
+        m_getuid.return_value = 100  # non-root
+        output_tarfile = self.tmp_path('logs.tgz')
+        with mock.patch('sys.stderr', new_callable=StringIO) as m_stderr:
+            self.assertEqual(
+                1, logs.collect_logs(output_tarfile, include_userdata=True))
+        self.assertEqual(
+            'To include userdata, root user is required.'
+            ' Try sudo cloud-init collect-logs\n',
+            m_stderr.getvalue())
+
+    def test_collect_logs_creates_tarfile(self, m_getuid):
         """collect-logs creates a tarfile with all related cloud-init info."""
+        m_getuid.return_value = 100
         log1 = self.tmp_path('cloud-init.log', self.new_root)
         write_file(log1, 'cloud-init-log')
         log2 = self.tmp_path('cloud-init-output.log', self.new_root)
         write_file(log2, 'cloud-init-output-log')
         ensure_dir(self.run_dir)
         write_file(self.tmp_path('results.json', self.run_dir), 'results')
+        write_file(self.tmp_path(INSTANCE_JSON_SENSITIVE_FILE, self.run_dir),
+                   'sensitive')
         output_tarfile = self.tmp_path('logs.tgz')
 
         date = datetime.utcnow().date().strftime('%Y-%m-%d')
@@ -59,6 +78,11 @@ class TestCollectLogs(FilesystemMockingTestCase):
         # unpack the tarfile and check file contents
         subp(['tar', 'zxvf', output_tarfile, '-C', self.new_root])
         out_logdir = self.tmp_path(date_logdir, self.new_root)
+        self.assertFalse(
+            os.path.exists(
+                os.path.join(out_logdir, 'run', 'cloud-init',
+                             INSTANCE_JSON_SENSITIVE_FILE)),
+            'Unexpected file found: %s' % INSTANCE_JSON_SENSITIVE_FILE)
         self.assertEqual(
             '0.7fake\n',
             load_file(os.path.join(out_logdir, 'dpkg-version')))
@@ -82,8 +106,9 @@ class TestCollectLogs(FilesystemMockingTestCase):
                 os.path.join(out_logdir, 'run', 'cloud-init', 'results.json')))
         fake_stderr.write.assert_any_call('Wrote %s\n' % output_tarfile)
 
-    def test_collect_logs_includes_optional_userdata(self):
+    def test_collect_logs_includes_optional_userdata(self, m_getuid):
         """collect-logs include userdata when --include-userdata is set."""
+        m_getuid.return_value = 0
         log1 = self.tmp_path('cloud-init.log', self.new_root)
         write_file(log1, 'cloud-init-log')
         log2 = self.tmp_path('cloud-init-output.log', self.new_root)
@@ -92,6 +117,8 @@ class TestCollectLogs(FilesystemMockingTestCase):
         write_file(userdata, 'user-data')
         ensure_dir(self.run_dir)
         write_file(self.tmp_path('results.json', self.run_dir), 'results')
+        write_file(self.tmp_path(INSTANCE_JSON_SENSITIVE_FILE, self.run_dir),
+                   'sensitive')
         output_tarfile = self.tmp_path('logs.tgz')
 
         date = datetime.utcnow().date().strftime('%Y-%m-%d')
@@ -132,4 +159,8 @@ class TestCollectLogs(FilesystemMockingTestCase):
         self.assertEqual(
             'user-data',
             load_file(os.path.join(out_logdir, 'user-data.txt')))
+        self.assertEqual(
+            'sensitive',
+            load_file(os.path.join(out_logdir, 'run', 'cloud-init',
+                                   INSTANCE_JSON_SENSITIVE_FILE)))
         fake_stderr.write.assert_any_call('Wrote %s\n' % output_tarfile)