oscap.py

import re
import enum
import contextlib
import collections
import types
import xml.etree.ElementTree as ET
from pathlib import Path

from lib import util, results

FixType = enum.Flag(
    'FixType',
    ['bash', 'ansible', 'anaconda', 'kickstart', 'blueprint', 'bootc'],
)


def parse_xml(path):
    """
    Parse an XML file, yielding tuples of
        (frames, elements)
    where each is an ordered list of namespace-free tag names ('frames') and the
    actual ElementTree objects ('elements') as it appears during a top-down
    recursive traversal.
    The yielded tuples are returned as child-first (as the parser *exits* the
    elements) in order to return complete Element objects.

    Ie. for a <Tag1> containing <Tag2>, this would yield:

        (['Tag1', 'Tag2'], [Element <Tag1> at 0x...>, <Element 'Tag2' at 0x...>])
        (['Tag1'], [Element <Tag1> at 0x...>])

    The intention is for the caller to match a specific part of the XML file
    by comparing the last N members of the frames list, and/or the element list,
    extracting further details from the last element.
    """
    # parse input XML tream in 10KB binary chunks (arbitrary reasonable value),
    # pass them to ElementTree parser, which returns element start/end events
    parser = ET.XMLPullParser(events=['start', 'end'])
    frames = []
    elements = []
    with open(path, 'rb') as f:
        while True:
            chunk = f.read(10000)
            if not chunk:
                break
            parser.feed(chunk)
            for event, elem in parser.read_events():
                if event == 'start':
                    frames.append(elem.tag.partition('}')[2] or elem.tag)
                    elements.append(elem)
                else:
                    yield (frames, elements)
                    frames.pop()
                    elements.pop()


class Datastream:
    def __init__(self, xml_file):
        # extracted datastream metadata
        #   self.profiles = {
        #     'ospp': namespace(
        #       .title = 'Some text',
        #       .rules = set( 'audit_delete_success' , 'service_firewalld_enabled' , ...),
        #       .values = set( ('var_rekey_limit_size','1G') , ...),
        #     ),
        #   }
        #   self.rules = {
        #     'configure_crypto_policy': namespace(
        #       .fixes = FixType.bash | FixType.ansible | ...
        #     ),
        #     'account_password_selinux_faillock_dir': namespace(
        #       .fixes = FixType.bash,
        #     ),
        #   }
        #   self.path = Path(file_the_datastream_was_parsed_from)
        self.profiles = collections.defaultdict()
        self.rules = collections.defaultdict()
        self._parse_datastream_xml(xml_file)
        self.path = Path(xml_file)

    def _parse_datastream_xml(self, xml_file):
        def make_profile():
            return types.SimpleNamespace(title=None, rules=set(), values=set())

        def make_rule():
            return types.SimpleNamespace(fixes=FixType(0), has_sce=False, has_oval=False)

        self.profiles.default_factory = make_profile
        self.rules.default_factory = make_rule

        for frames, elements in parse_xml(xml_file):
            # optimize a bit - filter out elements too shallow for anything below
            if len(frames) < 4:
                continue

            # the logic below tries to match the last one/two elements
            # in the stack, to hopefully limit false positive matches
            # elsewhere in the XML tree (if only element name was used)

            # profiles
            if frames[-1] == 'Profile':
                profile = elements[-1].get('id')
                profile = profile.removeprefix('xccdf_org.ssgproject.content_profile_')
                self.profiles[profile]  # let defaultdict fill in the values

            # profile contents
            elif frames[-2] == 'Profile':
                profile = elements[-2].get('id')
                profile = profile.removeprefix('xccdf_org.ssgproject.content_profile_')
                # title
                if frames[-1] == 'title':
                    text = elements[-1].text
                    self.profiles[profile].title = text
                # rule selection
                elif frames[-1] == 'select':
                    if elements[-1].get('selected') == 'true':
                        rule = elements[-1].get('idref')
                        rule = rule.removeprefix('xccdf_org.ssgproject.content_rule_')
                        self.profiles[profile].rules.add(rule)
                # variable refinement
                elif frames[-1] == 'refine-value':
                    name = elements[-1].get('idref')
                    name = name.removeprefix('xccdf_org.ssgproject.content_value_')
                    contents = elements[-1].get('selector')
                    self.profiles[profile].values.add((name, contents))

            # rules
            elif frames[-1] == 'Rule':
                rule_id = elements[-1].get('id')
                rule_id = rule_id.removeprefix('xccdf_org.ssgproject.content_rule_')
                self.rules[rule_id]  # let defaultdict fill in the values

            # fixes / remediations
            elif frames[-2:] == ['Rule', 'fix']:
                system = elements[-1].get('system')
                for_rule = elements[-1].get('id')
                if system == 'urn:xccdf:fix:script:sh':
                    self.rules[for_rule].fixes |= FixType.bash
                elif system == 'urn:xccdf:fix:script:ansible':
                    self.rules[for_rule].fixes |= FixType.ansible
                elif system == 'urn:redhat:anaconda:pre':
                    self.rules[for_rule].fixes |= FixType.anaconda
                elif system == 'urn:xccdf:fix:script:kickstart':
                    self.rules[for_rule].fixes |= FixType.kickstart
                elif system == 'urn:redhat:osbuild:blueprint':
                    self.rules[for_rule].fixes |= FixType.blueprint
                elif system == 'urn:xccdf:fix:script:bootc':
                    self.rules[for_rule].fixes |= FixType.bootc

            # checks (OVAL OR sce)
            elif frames[-2:] == ['Rule', 'check']:
                system = elements[-1].get('system')
                for_rule = elements[-2].get('id')
                for_rule = for_rule.removeprefix('xccdf_org.ssgproject.content_rule_')
                if system == 'http://open-scap.org/page/SCE':
                    self.rules[for_rule].has_sce = True
                if system == 'http://oval.mitre.org/XMLSchema/oval-definitions-5':
                    self.rules[for_rule].has_oval = True

        # "convert" to regular dict, make external logic get KeyError
        # on bad profile or rule name
        self.profiles.default_factory = None
        self.rules.default_factory = None

    def has_remediation(self, rule, remediation_type):
        """
        'rule' is a rule name, as returned by 'oscap xccdf eval --progress'
        and without the 'xccdf_org.ssgproject.content_rule_' prefix.

        'remediation_type' (FixType enum value or an expression of FixType enum values)
        contains the remediation (fix) types which we want to check for in the 'rule'
        remediations found in the datastream.

        Return True if 'rule' has remediations of 'remediation_type', False otherwise.
        """
        if not remediation_type:
            raise ValueError("remediation_type must be specified")

        if rule not in self.rules:
            raise ValueError(f"rule {rule} not found in datastream")

        return bool(self.rules[rule].fixes & remediation_type)

    def get_all_profiles_rules(self):
        """
        Return a deduplicated unified set of all rules from all profiles.
        """
        return {rule for profile in self.profiles.values() for rule in profile.rules}


# "global" datastream singleton, based on a xml file location decided by
# util/content.py, useful for the vast majority of tests that work with only
# one datastream, as provided by an installed RPM, or via a user-specified
# environment variable
# - any tests that work with .xml files directly (and need access to profiles
#   or rules inside them) should instantiate class Datastream() themselves
_cached_global_ds = None


def global_ds():
    global _cached_global_ds
    if _cached_global_ds is None:
        _cached_global_ds = Datastream(util.get_datastream())
    return _cached_global_ds


def rule_from_verbose(line):
    """
    Get (rulename, status) from an oscap info verbose output line.

    Return None if the input line is not a valid oscap verbose result line.
    """
    match = re.match(r'^xccdf_org.ssgproject.content_rule_(.+):([a-z]+)$', line)
    if match:
        return (match.group(1), match.group(2))
    else:
        return None


def report_from_verbose(lines, to_file='oscap.log'):
    """
    Report results from oscap output.

    All oscap output lines are written to 'to_file' (and streamed to ATEX
    when available) instead of being printed to stdout. The log file is
    pre-registered via register_log() so it is preserved even if the test
    errors out (similar to ATEX crash-safety). Callers should not pass
    'to_file' to results.add_log() or results.report_and_exit(logs=...)
    as it is already registered with the main test result.

    Note that this expects 'oscap xccdf eval' to be run:
      - with --progress
      - with stdout parsed into lines, fed to this function
      - with stderr discarded or left on the console
    """
    total = 0
    total_nonresults = 0
    log_path = results.register_log(to_file)

    with open(log_path, 'w') as out_file:
        for line in lines:
            results.atex_upload_log_data(to_file, f'{line}\n')
            out_file.write(f'{line}\n')
            out_file.flush()

            match = rule_from_verbose(line)
            if not match:
                continue
            rule, status = match
            total += 1
            note = None

            if status in ['pass', 'error', 'fail']:
                pass
            elif status in ['notapplicable', 'notchecked', 'notselected', 'informational']:
                total_nonresults += 1
                note = status
                status = 'skip'
            else:
                note = status
                status = 'error'

            results.report(status, rule, note)

    if total == 0:
        raise RuntimeError("oscap returned no results")
    if total == total_nonresults:
        raise RuntimeError("oscap didn't return any pass/fail/error results")

    util.log(f"all done: {total} total results")


@contextlib.contextmanager
def unselect_rules(orig_ds, new_ds, rules):
    """
    Given
    - a source XML file path as 'orig_ds',
    - a destination XML file path as 'new_ds',
    - an iterable of rules (partial or full rule names),
    copy the source datastream to the destination one, disabling the
    specified rules.
    """
    prefix = 'xccdf_org.ssgproject.content_rule_'
    # prefix rules once and store in a set for O(1) membership checks
    prefixed_rules = {
        (x if x.startswith(prefix) else prefix + x)
        for x in rules
    }

    rule_def_re = re.compile(r'<[^>]*Rule[^>]*\bid="([^\"]+)"')
    select_re = re.compile(r'<[^>]*select[^>]*\bidref="([^\"]+)"')

    new_ds = Path(new_ds)
    # remove a possible existing/old file
    if new_ds.exists():
        new_ds.unlink()

    util.log(f"reading {orig_ds}, writing to {new_ds}")
    with open(orig_ds) as orig_ds_f, open(new_ds, 'w') as new_ds_f:
        for line in orig_ds_f:
            matched = False

            rule_def_match = rule_def_re.search(line)
            if rule_def_match and rule_def_match.group(1) in prefixed_rules:
                matched = True
            else:
                select_match = select_re.search(line)
                if select_match and select_match.group(1) in prefixed_rules:
                    matched = True

            if matched:
                line = line.replace('selected="true"', 'selected="false"')
                util.log(f"unselected {line.strip()}")
            new_ds_f.write(line)