Move DISA Alignment Ansible test to parent directory as ansible.py a

move common functions to shared library for other DISA Alignment tests.

scanning/disa-alignment/ansible/main.fmf → scanning/disa-alignment/ansible.fmf

@@ -1,21 +1,12 @@
 summary: Compare SSG and DISA STIG benchmark scan results after Ansible remediation
-test: python3 -m lib.runtest ./test.py
+test: python3 -m lib.runtest ./ansible.py
 result: custom
 environment+:
-    PYTHONPATH: ../../..
+    PYTHONPATH: ../..
 require+:
-  # virt library dependencies
-  - libvirt-daemon
-  - libvirt-daemon-driver-qemu
-  - libvirt-daemon-driver-storage-core
-  - libvirt-daemon-driver-network
-  - firewalld
-  - qemu-kvm
-  - libvirt-client
-  - virt-install
-  - rpm-build
-  - createrepo
+  # ansible-core replaced ansible on RHEL-8+
   - ansible-core
+  # needed for the ini_file ansible plugin, and more
   - rhc-worker-playbook
 duration: 1h
 extra-hardware: |

scanning/disa-alignment/ansible.py

@@ -0,0 +1,61 @@
+#!/usr/bin/python3
+import os
+import subprocess
+
+import shared
+from lib import util, results, virt, versions, ansible
+from conf import partitions, remediation
+
+
+ansible.install_deps()
+virt.Host.setup()
+
+g = virt.Guest('minimal_with_oscap')
+
+if not g.can_be_snapshotted():
+    ks = virt.Kickstart(partitions=partitions.partitions)
+    g.install(kickstart=ks)
+    g.prepare_for_snapshot()
+
+# the VM guest ssh code doesn't use $HOME/.known_hosts, so Ansible blocks
+# on trying to accept its ssh key - tell it to ignore this
+os.environ['ANSIBLE_HOST_KEY_CHECKING'] = 'False'
+
+with g.snapshotted():
+    playbook = util.get_playbook(shared.profile)
+    skip_tags = ','.join(remediation.excludes())
+    skip_tags_arg = ['--skip-tags', skip_tags] if skip_tags else []
+    ansible_cmd = [
+        'ansible-playbook', '-v', '-i', f'{g.ipaddr},',
+        '--private-key', g.ssh_keyfile_path,
+        *skip_tags_arg,
+        playbook,
+    ]
+    util.subprocess_run(ansible_cmd, check=True)
+    g.soft_reboot()
+
+    with util.get_content() as content_dir:
+        g.copy_to(util.get_datastream(), 'ssg-ds.xml')
+        shared.content_scan(g, 'ssg-ds.xml', html='ssg-report.html', arf='ssg-arf.xml')
+        g.copy_from('ssg-report.html')
+        g.copy_from('ssg-arf.xml')
+
+        # There is always one (the latest) DISA benchmark in content src
+        references = content_dir / 'shared' / 'references'
+        disa_ds = next(
+            references.glob(f'disa-stig-rhel{versions.rhel.major}-*-xccdf-scap.xml')
+        )
+        g.copy_to(disa_ds, 'disa-ds.xml')
+        shared.disa_scan(g, 'disa-ds.xml', html='disa-report.html', arf='disa-arf.xml')
+        g.copy_from('disa-report.html')
+        g.copy_from('disa-arf.xml')
+
+        # Compare ARFs via CaC/content script and report results from output
+        compare_script = content_dir / 'utils' / 'compare_results.py'
+        env = os.environ.copy()
+        env['PYTHONPATH'] = str(content_dir)
+        cmd = [compare_script, 'ssg-arf.xml', 'disa-arf.xml']
+        proc = util.subprocess_run(cmd, env=env, universal_newlines=True, stdout=subprocess.PIPE)
+        shared.comparison_report(proc.stdout.rstrip('\n'))
+
+results.report_and_exit(logs=['ssg-report.html', 'disa-report.html'])

scanning/disa-alignment/main.fmf

@@ -0,0 +1,12 @@
+require+:
+  # virt library dependencies
+  - libvirt-daemon
+  - libvirt-daemon-driver-qemu
+  - libvirt-daemon-driver-storage-core
+  - libvirt-daemon-driver-network
+  - firewalld
+  - qemu-kvm
+  - libvirt-client
+  - virt-install
+  - rpm-build
+  - createrepo

scanning/disa-alignment/shared.py

@@ -0,0 +1,74 @@
+import re
+
+from lib import results, versions
+
+
+profile = 'stig'
+profile_full = f'xccdf_org.ssgproject.content_profile_{profile}'
+
+# old RHEL-7 oscap mixes errors into --progress rule names without a newline
+redir = '2>&1' if versions.oscap >= 1.3 else ''
+# RHEL-7 HTML report doesn't contain OVAL findings by default
+oval_results = '' if versions.oscap >= 1.3 else '--results results.xml --oval-results'
+
+shared_cmd = ['oscap', 'xccdf', 'eval', '--progress', oval_results]
+
+
+def content_scan(host, ds, html, arf):
+    """
+    Scan machine and prepare ARF results for STIG Viewer.
+    Return HTML report and ARF file.
+    """
+    cmd = [
+        *shared_cmd,
+        '--profile', profile_full,
+        '--report', html,
+        '--stig-viewer', arf,
+        ds, redir,
+    ]
+    proc = host.ssh(' '. join(cmd))
+    if proc.returncode not in [0,2]:
+        raise RuntimeError(f"remediation oscap failed with {proc.returncode}")
+
+
+def disa_scan(host, ds, html, arf):
+    """
+    Scan machine with datastream without profiles via '--profile (all)'.
+    Return HTML report and ARF file.
+    """
+    cmd = [
+        *shared_cmd,
+        '--profile', '\'(all)\'',
+        '--report', html,
+        '--results-arf', arf,
+        ds, redir
+    ]
+    proc = host.ssh(' '. join(cmd))
+    if proc.returncode not in [0,2]:
+        raise RuntimeError(f"remediation oscap failed with {proc.returncode}")
+
+
+def comparison_report(comparison_output):
+    """
+    Parse CaC/content utils/compare_results.py output and report different results.
+      Same result format:      CCE CCI - DISA_RULE_ID SSG_RULE_ID     RESULT
+      Different result format: CCE CCI - DISA_RULE_ID SSG_RULE_ID     SSG_RESULT - DISA_RESULT
+    """
+    result_regex = re.compile(r'[\w-]+ [\w-]+ - [\w-]+ (\w*)\s+(\w+)(?: - *(\w+))*')
+    for match in result_regex.finditer(comparison_output):
+        rule_id, ssg_result, disa_result = match.groups()
+        if not rule_id:
+            rule_id = 'rule_id_not_found'
+        # Only 1 result matched - same results
+        if not disa_result:
+            results.report('pass', rule_id)
+        # SSG CPE checks can make rule notapplicable by different reason (package not
+        # installed, architecture, RHEL version). DISA bechmark doesn't have this
+        # capability, it just 'pass'. Ignore such result misalignments
+        elif ssg_result == 'notapplicable' and disa_result == 'pass':
+            result_note = 'SSG result: notapplicable, DISA result: pass'
+            results.report('pass', rule_id, result_note)
+        # Different results
+        else:
+            result_note = f'SSG result: {ssg_result}, DISA result: {disa_result}'
+            results.report('fail', rule_id, result_note)