Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
59 changes: 39 additions & 20 deletions sos/cleaner/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -35,8 +35,8 @@
SoSCollectorDirectory)
from sos.cleaner.archives.generic import DataDirArchive, TarballArchive
from sos.cleaner.archives.insights import InsightsArchive
from sos.utilities import (get_human_readable, import_module,
ImporterHelper, is_executable)
from sos.utilities import (get_directory_size, get_human_readable,
import_module, ImporterHelper, is_executable)


# an auxiliary method to kick off child processes over its instances
Expand Down Expand Up @@ -410,6 +410,42 @@ def review_parser_values(self):
self.opts.skip_cleaning_files = [fnmatch.translate(p) for p in
self.opts.skip_cleaning_files]

def display_cleaner_results(self, final_path, map_path):
"""Print the location of the obfuscated output.

Size is the packed archive size, or the cumulative size of files
under a directory. Directory inode metadata is not used, as that
is commonly 4KiB rather than the obfuscated data size.

:param final_path: Path to the obfuscated archive or directory
:type final_path: ``str``

:param map_path: Path to the private mapping file
:type map_path: ``str``
"""
arcstat = os.stat(final_path)
if os.path.isfile(final_path):
size = arcstat.st_size
else:
size = get_directory_size(final_path)

# while these messages won't be included in the log file in the
# archive some facilities, such as our avocado test suite, will
# sometimes not capture print() output, so leverage the ui_log
# to print to console
self.ui_log.info(
f"A mapping of obfuscated elements is available at\n\t{map_path}"
)
self.ui_log.info(
f"\nThe obfuscated archive is available at\n\t{final_path}\n"
)
self.ui_log.info(f"\tSize\t{get_human_readable(size)}")
self.ui_log.info(f"\tOwner\t{getpwuid(arcstat.st_uid).pw_name}\n")
self.ui_log.info(
"Please send the obfuscated archive to your support\n"
"representative and keep the mapping file private."
)

def execute(self):
"""SoSCleaner will begin by inspecting the TARGET option to determine
if it is a directory, archive, or archive of archives.
Expand Down Expand Up @@ -500,24 +536,7 @@ def execute(self):
self.obfuscate_string(arc_path.split('/')[-1])
)
shutil.move(arc_path, final_path)
arcstat = os.stat(final_path)

# while these messages won't be included in the log file in the archive
# some facilities, such as our avocado test suite, will sometimes not
# capture print() output, so leverage the ui_log to print to console
self.ui_log.info(
f"A mapping of obfuscated elements is available at\n\t{map_path}"
)
self.ui_log.info(
f"\nThe obfuscated archive is available at\n\t{final_path}\n"
)

self.ui_log.info(f"\tSize\t{get_human_readable(arcstat.st_size)}")
self.ui_log.info(f"\tOwner\t{getpwuid(arcstat.st_uid).pw_name}\n")
self.ui_log.info(
"Please send the obfuscated archive to your support\n"
"representative and keep the mapping file private."
)
self.display_cleaner_results(final_path, map_path)

self.cleanup()
return None
Expand Down
20 changes: 20 additions & 0 deletions sos/utilities.py
Original file line number Diff line number Diff line change
Expand Up @@ -528,6 +528,26 @@ def get_human_readable(size, precision=2):
return f"{size:.{precision}f}{suffixes[suffixindex]}"


def get_directory_size(path):
"""Return the cumulative size of files under ``path``.

Directory inode sizes are not included. Permission errors are ignored
so a size display cannot fail the calling command.
"""
total_size = 0
try:
with os.scandir(path) as flist:
for _f in flist:
if _f.is_file(follow_symlinks=False):
total_size += _f.stat(follow_symlinks=False).st_size
elif _f.is_dir(follow_symlinks=False):
total_size += get_directory_size(_f.path)
except PermissionError:
# ignore these instead of bailing out on size calculation
pass
return total_size


def _os_wrapper(path, sysroot, method, module=os.path):
if sysroot and sysroot != os.sep:
if not path.startswith(sysroot):
Expand Down
67 changes: 67 additions & 0 deletions tests/unittests/cleaner_tests.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,12 +6,16 @@
#
# See the LICENSE file in the source distribution for further information.

import os
import tempfile
import unittest
from ipaddress import ip_interface
from os.path import join
from unittest import mock

import sos.policies
from sos.cleaner import SoSCleaner
from sos.utilities import get_directory_size, get_human_readable
from sos.cleaner.parsers.ip_parser import SoSIPParser
from sos.cleaner.parsers.mac_parser import SoSMacParser
from sos.cleaner.parsers.hostname_parser import SoSHostnameParser
Expand Down Expand Up @@ -723,3 +727,66 @@ def test_packed_dirs_empty_for_premature_manifest(self):
with mock.patch.object(self.archive, 'get_file_content',
return_value='{"components": {"report": {}}}'):
self.assertEqual(self.archive._load_packed_dirs(), [])


class CleanerDisplayResultsTests(unittest.TestCase):
"""Verify sos clean reports Size for archives and directories.

Directory output must use the cumulative file size, not
os.stat().st_size of the directory inode (often 4KiB).
"""

def setUp(self):
self.cleaner = mock.Mock()
self.cleaner.ui_log = mock.Mock()

def _info_messages(self):
return [call.args[0] for call in
self.cleaner.ui_log.info.call_args_list]

def test_archive_reports_file_size(self):
with tempfile.NamedTemporaryFile(delete=False) as tfile:
tfile.write(b'x' * 2048)
tfile.flush()
path = tfile.name
try:
SoSCleaner.display_cleaner_results(
self.cleaner, path, '/tmp/private_map'
)
msgs = self._info_messages()
expected = f"\tSize\t{get_human_readable(os.stat(path).st_size)}"
self.assertIn(expected, msgs)
self.assertTrue(any('\tOwner\t' in msg for msg in msgs))
self.assertTrue(any(path in msg for msg in msgs))
finally:
os.unlink(path)

def test_directory_reports_cumulative_size(self):
with tempfile.TemporaryDirectory() as tmpdir:
with open(os.path.join(tmpdir, 'data'), 'wb') as dfile:
dfile.write(b'x' * 8192)
nested = os.path.join(tmpdir, 'nested')
os.mkdir(nested)
with open(os.path.join(nested, 'more'), 'wb') as nfile:
nfile.write(b'y' * 1024)
SoSCleaner.display_cleaner_results(
self.cleaner, tmpdir, '/tmp/private_map'
)
msgs = self._info_messages()
expected_bytes = get_directory_size(tmpdir)
expected = f"\tSize\t{get_human_readable(expected_bytes)}"
inode_bytes = os.stat(tmpdir).st_size
inode_size = f"\tSize\t{get_human_readable(inode_bytes)}"
self.assertEqual(expected_bytes, 9216)
self.assertIn(expected, msgs)
self.assertNotEqual(expected, inode_size)
self.assertTrue(any('\tOwner\t' in msg for msg in msgs))
self.assertTrue(any(tmpdir in msg for msg in msgs))
self.assertTrue(
any('private_map' in msg for msg in msgs)
)

def test_directory_size_ignores_permission_errors(self):
with mock.patch('sos.utilities.os.scandir',
side_effect=PermissionError):
self.assertEqual(get_directory_size('/restricted'), 0)