Merge pull request #56 from xenserver/feature/snmp-status-report

Merge feature branch `feature/snmp-status-report` into `master`

.pylintrc

@@ -190,7 +190,8 @@ disable=anomalous-backslash-in-string,
         # Py2 compat: Python2 requires calls to super() to have it's arguments:
         super-with-arguments,
         # Py2 compat: As long as we try to use conditional imports for Py2+Py3:
-        ungrouped-imports
+        ungrouped-imports,
+        use-maxsplit-arg
 
 # Enable the message, report, category or checker with the given id(s). You can
 # either give multiple identifier separated by comma (,) or put this option

tests/integration/test_system_load.py

@@ -1,7 +1,7 @@
 """tests/integration/test_system_load.py: Test xen-bugtool --entries=system-load"""
 import os
 
-from .utils import check_file, run_bugtool_entry, assert_content_from_dom0_template
+from .utils import assert_file, run_bugtool_entry, assert_content_from_dom0_template
 
 
 # In this test case we need to sleep for 1 sec, and it is sufficient
@@ -27,5 +27,5 @@ def test_system_load(output_archive_type="zip"):
     run_bugtool_entry(output_archive_type, entry)
 
     assert_content_from_dom0_template("sar-A.out", "etc/xensource-inventory")
-    assert check_file("var/log/sa/sa01") == "sa01 test data"
-    assert check_file("var/log/sa/sar31") == "sar31 test data"
+    assert_file("var/log/sa/sa01", "sa01 test data")
+    assert_file("var/log/sa/sar31", "sar31 test data")

tests/integration/test_xenserver_config.py

@@ -1,7 +1,7 @@
 """tests/integration/test_xenserver_config.py: Test xen-bugtool --entries=xenserver-config"""
 import os
 
-from .utils import assert_cmd, check_file, run_bugtool_entry, assert_content_from_dom0_template
+from .utils import assert_cmd, assert_file, run_bugtool_entry, assert_content_from_dom0_template
 
 
 def test_xenserver_config(output_archive_type):
@@ -11,9 +11,9 @@ def test_xenserver_config(output_archive_type):
     run_bugtool_entry(output_archive_type, entry)
 
     # Assert that the bugtool output archive of --entries=xenserver-config matches our expectations for it:
-    assert check_file("ls-lR-%opt%xensource.out").splitlines()[0] == "/opt/xensource:"
-    assert check_file("ls-lR-%etc%xensource%static-vdis.out") == ""
-    assert check_file("static-vdis-list.out") == "list"
+    assert_file("ls-lR-%opt%xensource.out", "/opt/xensource:", only_first_line=True)
+    assert_file("ls-lR-%etc%xensource%static-vdis.out", "")
+    assert_file("static-vdis-list.out", "list")
 
     # Assert the contents of the extracted etc/systemd.tar
     os.chdir("..")
@@ -23,3 +23,23 @@ def test_xenserver_config(output_archive_type):
     os.chdir(entry)
     assert_content_from_dom0_template("etc/systemd")
     assert_content_from_dom0_template("etc/xensource-inventory")
+
+    # Sample SNMP config files are currently not in the dom0_template!
+    # Reading them records the error message in the file content, do we want this?
+    # I think the "Failed to filter" is redundant in it.
+    # Maybe decide on a standardized error for missing files in bugtool?
+
+    # TODO: To be clarified or fixed as part of CP-46759 or a follow-up!
+
+    for input_file in [
+        "/etc/snmp/snmp.xs.conf",
+        "/etc/snmp/snmpd.xs.conf",
+        "/var/lib/net-snmp/snmpd.conf",
+    ]:
+        assert_file(
+            input_file.split("/")[-1].replace(".", "_") + ".out",
+            # That's a very long error message an the 1st two parts are redundant:
+            "Failed to filter %s "
+            "[Errno 2] "
+            "No such file or directory: '%s'" % (input_file, input_file),
+        )

tests/integration/utils.py

@@ -35,19 +35,33 @@ def run(command):
 
 def assert_cmd(cmd, remove_file):
     """Run the given command, print its stdout and stderr and return them. Remove the given remove_file"""
+    # If something causes an Exception, the remove_file will not be removed and
+    # this will cause a final test failure check to trip and fail the test too:
     stdout, stderr = run(cmd)
-    # After successfuly verficiation of the files, remove the checked output file(missed files remain):
+    # After successful verification of the file, remove the checked output file:
     os.unlink(remove_file)
     return stdout, stderr
 
 
-def check_file(path):
+def assert_file(path, expected, only_first_line=False):
     """Return the contents of the passed bugtool output file for verification"""
+    assert os.path.exists(path)
     with open(path) as handle:
-        contents = handle.read()
-    # After successfuly verficiation of the files, remove the checked output file (missed files remain):
+        read_data = handle.read()
+        if only_first_line:
+            read_data = read_data.splitlines()[0]
+        if read_data != expected:
+            error_msg = "Data in %s does not match!" % path
+            print(error_msg)
+            print("- " + expected)
+            print("+ " + read_data)
+            # Exit with an exception, and the file with unexpected contents
+            # is also kept. Any files that are left in the extracted directory
+            # will cause a final test failure check to trip and fail the test too:
+            assert RuntimeError(error_msg)
+
+    # After successful verification of the file, remove the checked output file:
     os.unlink(path)
-    return contents
 
 
 def assert_content_from_dom0_template(path, control_path=None):

tests/unit/conftest.py

@@ -5,6 +5,20 @@
 import pytest
 
 
+@pytest.fixture(scope="function")
+def builtins():
+    """
+    Pytest fixture to return the name of the built-in module.
+    The built-in module provides the built-in functions, exceptions, etc.
+    It is needed to replace built-in functions like open() with a new mock.
+
+    :returns (str): The name of the built-in module.
+    """
+    if sys.version_info < (3,):
+        return "__builtin__"  # pragma: no cover
+    return "builtins"
+
+
 @pytest.fixture(scope="session")
 def testdir():
     """Test fixture to get the directory of the unit test for finding other files relative to it"""

tests/unit/test_snmp.py

@@ -0,0 +1,50 @@
+"""Unit tests for the bugtool SNMP filter functions: filter_snmp.*_conf()"""
+
+
+def test_filter_snmp_xs_conf(bugtool, builtins, mocker):
+    """Assert that filter_snmp_xs_conf() replaces sensitive strings"""
+
+    snmp_xs_conf_input = """
+        "community": "SECRET",
+        "authentication_key": "SECRET",
+        "privacy_key": "SECRET",
+    """
+    snmp_xs_conf_output = """
+        "community": "REMOVED",
+        "authentication_key": "REMOVED",
+        "privacy_key": "REMOVED",
+    """
+    mocker.patch(builtins + ".open", mocker.mock_open(read_data=snmp_xs_conf_input))
+    assert bugtool.filter_snmp_xs_conf("_") == snmp_xs_conf_output
+
+
+def test_filter_snmpd_xs_conf(bugtool, builtins, mocker):
+    """Assert that filter_snmpd_xs_conf() replaces sensitive strings"""
+
+    snmpd_xs_conf_input = "com2sec notConfigUser default SECRET"
+    snmpd_xs_conf_output = "com2sec notConfigUser default REMOVED"
+    mocker.patch(builtins + ".open", mocker.mock_open(read_data=snmpd_xs_conf_input))
+    assert bugtool.filter_snmpd_xs_conf("_") == snmpd_xs_conf_output
+
+
+def test_filter_snmpd_conf(bugtool, builtins, mocker):
+    """Assert that filter_snmpd_conf() replaces sensitive strings"""
+
+    snmpd_conf_input = (
+        "usmUser 1 3 0x80001f8880f369b576d8b2a46500000000 0x7872746d69612d30372d3035 "
+        + "0x7872746d69612d30372d3035 NULL .1.3.6.1.6.3.10.1.1.3 "
+        + "SECRET "
+        + ".1.3.6.1.6.3.10.1.2.2 "
+        + "SECRET "
+        + "0x"
+    )
+    snmpd_conf_output = (
+        "usmUser 1 3 0x80001f8880f369b576d8b2a46500000000 0x7872746d69612d30372d3035 "
+        + "0x7872746d69612d30372d3035 NULL .1.3.6.1.6.3.10.1.1.3 "
+        + "REMOVED "
+        + ".1.3.6.1.6.3.10.1.2.2 "
+        + "REMOVED "
+        + "0x"
+    )
+    mocker.patch(builtins + ".open", mocker.mock_open(read_data=snmpd_conf_input))
+    assert bugtool.filter_snmpd_conf("_") == snmpd_conf_output

xen-bugtool

@@ -235,6 +235,10 @@ NRPE_DIR = '/etc/nrpe.d'
 XEN_BUGTOOL_LOG = 'xen-bugtool.log'
 CRON_DIRS = '/etc/cron*'
 CRON_SPOOL = '/var/spool/cron'
+SNMP_XS_CONF = '/etc/snmp/snmp.xs.conf'
+SNMPD_XS_CONF = '/etc/snmp/snmpd.xs.conf'
+SNMPD_CONF = '/var/lib/net-snmp/snmpd.conf'
+SYSCONFIG_SNMPD = '/etc/sysconfig/snmpd'
 
 #
 # External programs
@@ -1176,6 +1180,10 @@ exclude those logs from the archive.
                     'vgscan-log']])
     tree_output(CAP_XENSERVER_INSTALL, INSTALLED_REPOS_DIR)
     tree_output(CAP_XENSERVER_INSTALL, UPDATE_APPLIED_DIR)
+    func_output(CAP_XENSERVER_CONFIG, 'snmp_xs_conf', filter_snmp_xs_conf)
+    func_output(CAP_XENSERVER_CONFIG, 'snmpd_xs_conf', filter_snmpd_xs_conf)
+    func_output(CAP_XENSERVER_CONFIG, 'snmpd_conf', filter_snmpd_conf)
+    file_output(CAP_XENSERVER_CONFIG, [SYSCONFIG_SNMPD])
 
     try:
         load_plugins()
@@ -1626,6 +1634,26 @@ def multipathd_topology(cap):
 
     return stdout
 
+def filter_snmp_xs_conf(_):
+    """Filter /etc/snmp/snmp.xs.conf with keys and community removed"""
+    return snmp_regex_filter(SNMP_XS_CONF, r'(\"(community|\w*_key)\"\s*:\s*\")\S+(\",*)', r'\1REMOVED\3')
+
+def filter_snmpd_xs_conf(_):
+    """Filter /etc/snmp/snmpd.xs.conf with the com2sec community removed"""
+    return snmp_regex_filter(SNMPD_XS_CONF, r"(com2sec(\s+\S+){2}\s+)\S+", r"\1REMOVED")
+
+def filter_snmpd_conf(_):
+    """Filter /var/lib/net-snmp/snmpd.conf with the usmUser fields authKey and privKey removed"""
+    return snmp_regex_filter(SNMPD_CONF, r"(usmUser(\s+\S+){7}\s+)\S+(\s+\S+\s+)\S+(\s+\S+)", r"\1REMOVED\3REMOVED\4")
+
+def snmp_regex_filter(replace_file, regex_str, replace_str):
+    try:
+        with open(replace_file, "r") as file:
+            return re.sub(regex_str, replace_str, file.read())
+    except Exception as e:
+        return "Failed to filter %s %s" % (replace_file, str(e))
+
+
 def dp_list():
     output = io.BytesIO()
     procs = [ProcOutput([OVS_DPCTL, 'dump-dps'], caps[CAP_NETWORK_STATUS][MAX_TIME], output)]