org_fedora_oscap.common.extract_data() - Code Metrics - Inspection of "Introduce various small fixes and refactoring" - OpenSCAP/oscap-anaconda-addon - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — rhel8-branch (#146)

by Matěj

created 2021-05-26 15:08 UTC

org_fedora_oscap.common.extract_data() C

↳ Parent: org_fedora_oscap.common

Complexity

Conditions

Size

Total Lines	69
Code Lines	38

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	38
dl	0
loc	69
rs	5.4
c	0
b	0
f	0
cc	11
nop	3

How to fix Long Method Complexity

#
# Copyright (C) 2013  Red Hat, Inc.
#
# This copyrighted material is made available to anyone wishing to use,
# modify, copy, or redistribute it subject to the terms and conditions of
# the GNU General Public License v.2, or (at your option) any later version.
# This program is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY expressed or implied, including the implied warranties of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General
# Public License for more details.  You should have received a copy of the
# GNU General Public License along with this program; if not, write to the
# Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
# 02110-1301, USA.  Any Red Hat trademarks that are incorporated in the
# source code or documentation are not subject to the GNU General Public
# License and may only be used or replicated with the express permission of
# Red Hat, Inc.
#
# Red Hat Author(s): Vratislav Podzimek <[email protected]>
#

"""
Module with various classes and functions needed by the OSCAP addon that are
not specific to any installation mode (tui, gui, ks).

"""

import os
import tempfile
import subprocess
import zipfile
import tarfile

import cpioarchive
import re
import logging

from collections import namedtuple
import gettext
from functools import wraps
from pyanaconda.core import constants
from pyanaconda.modules.common.constants.services import NETWORK
from pyanaconda.threading import threadMgr, AnacondaThread
from org_fedora_oscap import utils
from org_fedora_oscap.data_fetch import fetch_data

log = logging.getLogger("anaconda")


# mimick pyanaconda/core/i18n.py
def _(string):
    if string:
        return gettext.translation("oscap-anaconda-addon", fallback=True).gettext(string)
    else:
        return ""


def N_(string): return string


# everything else should be private
__all__ = ["run_oscap_remediate", "get_fix_rules_pre",
           "wait_and_fetch_net_data", "extract_data", "strip_content_dir",
           "OSCAPaddonError"]

INSTALLATION_CONTENT_DIR = "/tmp/openscap_data/"
TARGET_CONTENT_DIR = "/root/openscap_data/"

SSG_DIR = "/usr/share/xml/scap/ssg/content/"

# Enable patches that set the content name at package-time
DEFAULT_SSG_CONTENT_NAME = ""
SSG_CONTENT = DEFAULT_SSG_CONTENT_NAME
if not SSG_CONTENT:
    if constants.shortProductName != 'anaconda':
        if constants.shortProductName == 'fedora':
            SSG_CONTENT = "ssg-fedora-ds.xml"
        else:
            SSG_CONTENT = (
                "ssg-{name}{version}-ds.xml"
                .format(
                    name=constants.shortProductName,
                    version=constants.productVersion.strip(".")[0]))

RESULTS_PATH = utils.join_paths(TARGET_CONTENT_DIR,
                                "eval_remediate_results.xml")
REPORT_PATH = utils.join_paths(TARGET_CONTENT_DIR,
                               "eval_remediate_report.html")

PRE_INSTALL_FIX_SYSTEM_ATTR = "urn:redhat:anaconda:pre"

THREAD_FETCH_DATA = "AnaOSCAPdataFetchThread"

SUPPORTED_ARCHIVES = (".zip", ".tar", ".tar.gz", ".tar.bz2", )

# buffer size for reading and writing out data (in bytes)
IO_BUF_SIZE = 2 * 1024 * 1024


class OSCAPaddonError(Exception):
    """Exception class for OSCAP addon related errors."""

    pass


class OSCAPaddonNetworkError(OSCAPaddonError):
    """Exception class for OSCAP addon related network errors."""

    pass


class ExtractionError(OSCAPaddonError):
    """Exception class for the extraction errors."""

    pass


MESSAGE_TYPE_FATAL = 0
MESSAGE_TYPE_WARNING = 1
MESSAGE_TYPE_INFO = 2

# namedtuple for messages returned from the rules evaluation
#   origin -- class (inherited from RuleHandler) that generated the message
#   type -- one of the MESSAGE_TYPE_* constants defined above
#   text -- the actual message that should be displayed, logged, ...
RuleMessage = namedtuple("RuleMessage", ["origin", "type", "text"])


class SubprocessLauncher(object):
    def __init__(self, args):
        self.args = args
        self.stdout = ""
        self.stderr = ""
        self.messages = []
        self.returncode = None

    def execute(self, ** kwargs):
        command_string = " ".join(self.args)
        log.info(
            "OSCAP addon: Executing subprocess: '{command_string}'"
            .format(command_string=command_string))
        try:
            proc = subprocess.Popen(self.args, stdout=subprocess.PIPE,
                                    stderr=subprocess.PIPE, ** kwargs)
        except OSError as oserr:
            msg = "Failed to run the oscap tool: %s" % oserr
            raise OSCAPaddonError(msg)

        (stdout, stderr) = proc.communicate()
        self.stdout = stdout.decode()
        self.stderr = stderr.decode(errors="replace")
        self.messages = re.findall(r'OpenSCAP Error:.*', self.stderr)
        self.messages = self.messages + re.findall(r'E: oscap:.*', self.stderr)

        self.returncode = proc.returncode

    def log_messages(self):
        for message in self.messages:
            log.warning("OSCAP addon: " + message)


def get_fix_rules_pre(profile, fpath, ds_id="", xccdf_id="", tailoring=""):
    """
    Get fix rules for the pre-installation environment for a given profile in a
    given datastream and checklist in a given file.

    :see: run_oscap_remediate
    :see: _run_oscap_gen_fix
    :return: fix rules for a given profile
    :rtype: str

    """

    return _run_oscap_gen_fix(profile, fpath, PRE_INSTALL_FIX_SYSTEM_ATTR,
                              ds_id=ds_id, xccdf_id=xccdf_id,
                              tailoring=tailoring)


def _run_oscap_gen_fix(profile, fpath, template, ds_id="", xccdf_id="",
                       tailoring=""):
    """
    Run oscap tool on a given file to get the contents of fix elements with the
    'system' attribute equal to a given template for a given datastream,
    checklist and profile.

    :see: run_oscap_remediate
    :param template: the value of the 'system' attribute of the fix elements
    :type template: str
    :return: oscap tool's stdout
    :rtype: str

    """

    if not profile:
        return ""

    args = ["oscap", "xccdf", "generate", "fix"]
    args.append("--template=%s" % template)

    # oscap uses the default profile by default
    if profile.lower() != "default":
        args.append("--profile=%s" % profile)
    if ds_id:
        args.append("--datastream-id=%s" % ds_id)
    if xccdf_id:
        args.append("--xccdf-id=%s" % xccdf_id)
    if tailoring:
        args.append("--tailoring-file=%s" % tailoring)

    args.append(fpath)

    proc = SubprocessLauncher(args)
    proc.execute()
    proc.log_messages()
    if proc.returncode != 0:
        msg = "Failed to generate fix rules with the oscap tool: %s" % proc.stderr
        raise OSCAPaddonError(msg)

    return proc.stdout


def run_oscap_remediate(profile, fpath, ds_id="", xccdf_id="", tailoring="",
                        chroot=""):
    """
    Run the evaluation and remediation with the oscap tool on a given file,
    doing the remediation as defined in a given profile defined in a given
    checklist that is a part of a given datastream. If requested, run in
    chroot.

    :param profile: id of the profile that will drive the remediation
    :type profile: str
    :param fpath: path to a file with SCAP content
    :type fpath: str
    :param ds_id: ID of the datastream that contains the checklist defining
                  the profile
    :type ds_id: str
    :param xccdf_id: ID of the checklist that defines the profile
    :type xccdf_id: str
    :param tailoring: path to a tailoring file
    :type tailoring: str
    :param chroot: path to the root the oscap tool should be run in
    :type chroot: str
    :return: oscap tool's stdout (summary of the rules, checks and fixes)
    :rtype: str

    """

    if not profile:
        return ""

    def do_chroot():
        """Helper function doing the chroot if requested."""
        if chroot and chroot != "/":
            os.chroot(chroot)
            os.chdir("/")

    # make sure the directory for the results exists
    results_dir = os.path.dirname(RESULTS_PATH)
    if chroot:
        results_dir = os.path.normpath(chroot + "/" + results_dir)
    utils.ensure_dir_exists(results_dir)

    args = ["oscap", "xccdf", "eval"]
    args.append("--remediate")
    args.append("--results=%s" % RESULTS_PATH)
    args.append("--report=%s" % REPORT_PATH)

    # oscap uses the default profile by default
    if profile.lower() != "default":
        args.append("--profile=%s" % profile)
    if ds_id:
        args.append("--datastream-id=%s" % ds_id)
    if xccdf_id:
        args.append("--xccdf-id=%s" % xccdf_id)
    if tailoring:
        args.append("--tailoring-file=%s" % tailoring)

    args.append(fpath)

    proc = SubprocessLauncher(args)
    proc.execute(preexec_fn=do_chroot)
    proc.log_messages()

    if proc.returncode not in (0, 2):
        # 0 -- success; 2 -- no error, but checks/remediation failed
        msg = "Content evaluation and remediation with the oscap tool "\
            "failed: %s" % proc.stderr
        raise OSCAPaddonError(msg)

    return proc.stdout


def wait_and_fetch_net_data(url, out_file, ca_certs=None):
    """
    Function that waits for network connection and starts a thread that fetches
    data over network.

    :see: org_fedora_oscap.data_fetch.fetch_data
    :return: the name of the thread running fetch_data
    :rtype: str

    """

    # get thread that tries to establish a network connection
    nm_conn_thread = threadMgr.get(constants.THREAD_WAIT_FOR_CONNECTING_NM)
    if nm_conn_thread:
        # NM still connecting, wait for it to finish
        nm_conn_thread.join()

    network_proxy = NETWORK.get_proxy()
    if not network_proxy.Connected:
        raise OSCAPaddonNetworkError("Network connection needed to fetch data.")

    log.info("Fetching data from {url}".format(url=url))
    fetch_data_thread = AnacondaThread(name=THREAD_FETCH_DATA,
                                       target=fetch_data,
                                       args=(url, out_file, ca_certs),
                                       fatal=False)

    # register and run the thread
    threadMgr.add(fetch_data_thread)

    return THREAD_FETCH_DATA


def extract_data(archive, out_dir, ensure_has_files=None):
    """
    Fuction that extracts the given archive to the given output directory. It
    tries to find out the archive type by the file name.

    :param archive: path to the archive file that should be extracted
    :type archive: str
    :param out_dir: output directory the archive should be extracted to
    :type out_dir: str
    :param ensure_has_files: relative paths to the files that must exist in the
                             archive
    :type ensure_has_files: iterable of strings or None
    :return: a list of files and directories extracted from the archive
    :rtype: [str]

    """

    if not ensure_has_files:
        ensure_has_files = []

    # get rid of empty file paths
    ensure_has_files = [fpath for fpath in ensure_has_files if fpath]

    msg = "OSCAP addon: Extracting {archive}".format(archive=archive)
    if ensure_has_files:
        msg += ", expecting to find {files} there.".format(files=tuple(ensure_has_files))
    log.info(msg)

    result = []
    if archive.endswith(".zip"):
        # ZIP file
        try:
            zfile = zipfile.ZipFile(archive, "r")
        except Exception as exc:
            msg = "Error exctracting archive as a zipfile: {exc}".format(exc=str(exc))
            raise ExtractionError(msg)

        # generator for the paths of the files found in the archive (dirs end
        # with "/")
        files = set(info.filename for info in zfile.filelist
                    if not info.filename.endswith("/"))
        for fpath in ensure_has_files or ():
            if fpath not in files:
                msg = "File '%s' not found in the archive '%s'" % (fpath,
                                                                   archive)
                raise ExtractionError(msg)

        utils.ensure_dir_exists(out_dir)
        zfile.extractall(path=out_dir)
        result = [utils.join_paths(out_dir, info.filename) for info in zfile.filelist]
        zfile.close()
    elif archive.endswith(".tar"):
        # plain tarball
        result = _extract_tarball(archive, out_dir, ensure_has_files, None)
    elif archive.endswith(".tar.gz"):
        # gzipped tarball
        result = _extract_tarball(archive, out_dir, ensure_has_files, "gz")
    elif archive.endswith(".tar.bz2"):
        # bzipped tarball
        result = _extract_tarball(archive, out_dir, ensure_has_files, "bz2")
    elif archive.endswith(".rpm"):
        # RPM
        result = _extract_rpm(archive, out_dir, ensure_has_files)
    # elif other types of archives
    else:
        raise ExtractionError("Unsuported archive type")
    log.info("OSCAP addon: Extracted {files} from the supplied content"
             .format(files=result))
    return result


def _extract_tarball(archive, out_dir, ensure_has_files, alg):
    """
    Extract the given TAR archive to the given output directory and make sure
    the given file exists in the archive.

    :see: extract_data
    :param alg: compression algorithm used for the tarball
    :type alg: str (one of "gz", "bz2") or None
    :return: a list of files and directories extracted from the archive
    :rtype: [str]

    """

    if alg and alg not in ("gz", "bz2",):
        raise ExtractionError("Unsupported compression algorithm")

    mode = "r"
    if alg:
        mode += ":%s" % alg

    try:
        tfile = tarfile.TarFile.open(archive, mode)
    except tarfile.TarError as err:
        raise ExtractionError(str(err))

    # generator for the paths of the files found in the archive
    files = set(member.path for member in tfile.getmembers()
                if member.isfile())

    for fpath in ensure_has_files or ():
        if fpath not in files:
            msg = "File '%s' not found in the archive '%s'" % (fpath, archive)
            raise ExtractionError(msg)

    utils.ensure_dir_exists(out_dir)
    tfile.extractall(path=out_dir)
    result = [utils.join_paths(out_dir, member.path) for member in tfile.getmembers()]
    tfile.close()

    return result


def _extract_rpm(rpm_path, root="/", ensure_has_files=None):
    """
    Extract the given RPM into the directory tree given by the root argument
    and make sure the given file exists in the archive.

    :param rpm_path: path to the RPM file that should be extracted
    :type rpm_path: str
    :param root: root of the directory tree the RPM should be extracted into
    :type root: str
    :param ensure_has_files: relative paths to the files that must exist in the
                             RPM
    :type ensure_has_files: iterable of strings or None
    :return: a list of files and directories extracted from the archive
    :rtype: [str]

    """

    # run rpm2cpio and process the output with the cpioarchive module
    temp_fd, temp_path = tempfile.mkstemp(prefix="oscap_rpm")
    proc = subprocess.Popen(["rpm2cpio", rpm_path], stdout=temp_fd)
    proc.wait()
    if proc.returncode != 0:
        msg = "Failed to convert RPM '%s' to cpio archive" % rpm_path
        raise ExtractionError(msg)

    os.close(temp_fd)

    try:
        archive = cpioarchive.CpioArchive(temp_path)
    except cpioarchive.CpioError as err:
        raise ExtractionError(str(err))

    # get entries from the archive (supports only iteration over entries)
    entries = set(entry for entry in archive)

    # cpio entry names (paths) start with the dot
    entry_names = [entry.name.lstrip(".") for entry in entries]

    for fpath in ensure_has_files or ():
        # RPM->cpio entries have absolute paths
        if fpath not in entry_names and \
           os.path.join("/", fpath) not in entry_names:
            msg = "File '%s' not found in the archive '%s'" % (fpath, rpm_path)
            raise ExtractionError(msg)

    try:
        for entry in entries:
            if entry.size == 0:
                continue
            dirname = os.path.dirname(entry.name.lstrip("."))
            out_dir = os.path.normpath(root + dirname)
            utils.ensure_dir_exists(out_dir)

            out_fpath = os.path.normpath(root + entry.name.lstrip("."))
            if os.path.exists(out_fpath):
                continue
            with open(out_fpath, "wb") as out_file:
                buf = entry.read(IO_BUF_SIZE)
                while buf:
                    out_file.write(buf)
                    buf = entry.read(IO_BUF_SIZE)
    except (IOError, cpioarchive.CpioError) as e:
        raise ExtractionError(e)

    # cleanup
    archive.close()
    os.unlink(temp_path)

    return [os.path.normpath(root + name) for name in entry_names]


def strip_content_dir(fpaths, phase="preinst"):
    """
    Strip content directory prefix from the file paths for either
    pre-installation or post-installation phase.

    :param fpaths: iterable of file paths to strip content directory prefix
                   from
    :type fpaths: iterable of strings
    :param phase: specifies pre-installation or post-installation phase
    :type phase: "preinst" or "postinst"
    :return: the same iterable of file paths as given with the content
             directory prefix stripped
    :rtype: same type as fpaths

    """

    if phase == "preinst":
        remove_prefix = lambda x: x[len(INSTALLATION_CONTENT_DIR):]
    else:
        remove_prefix = lambda x: x[len(TARGET_CONTENT_DIR):]

    return utils.keep_type_map(remove_prefix, fpaths)


def get_ssg_path(root="/"):
    return utils.join_paths(root, SSG_DIR + SSG_CONTENT)


def ssg_available(root="/"):
    """
    Tries to find the SCAP Security Guide under the given root.

    :return: True if SSG was found under the given root, False otherwise

    """

    return os.path.exists(get_ssg_path(root))


def dry_run_skip(func):
    """
    Decorator that makes sure the decorated function is noop in the dry-run
    mode.

    :param func: a decorated function that needs to have the first parameter an
                 object with the _addon_data attribute referencing the OSCAP
                 addon's ksdata
    """

    @wraps(func)
    def decorated(self, *args, **kwargs):
        if self._addon_data.dry_run:
            return
        else:
            return func(self, *args, **kwargs)

    return decorated


1			#
2			# Copyright (C) 2013 Red Hat, Inc.
3			#
4			# This copyrighted material is made available to anyone wishing to use,
5			# modify, copy, or redistribute it subject to the terms and conditions of
6			# the GNU General Public License v.2, or (at your option) any later version.
7			# This program is distributed in the hope that it will be useful, but WITHOUT
8			# ANY WARRANTY expressed or implied, including the implied warranties of
9			# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General
10			# Public License for more details. You should have received a copy of the
11			# GNU General Public License along with this program; if not, write to the
12			# Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
13			# 02110-1301, USA. Any Red Hat trademarks that are incorporated in the
14			# source code or documentation are not subject to the GNU General Public
15			# License and may only be used or replicated with the express permission of
16			# Red Hat, Inc.
17			#
18			# Red Hat Author(s): Vratislav Podzimek <[email protected]>
19			#
20
21			"""
22			Module with various classes and functions needed by the OSCAP addon that are
23			not specific to any installation mode (tui, gui, ks).
24
25			"""
26
27			import os
28			import tempfile
29			import subprocess
30			import zipfile
31			import tarfile
32
33			import cpioarchive
34			import re
35			import logging
36
37			from collections import namedtuple
38			import gettext
39			from functools import wraps
40			from pyanaconda.core import constants
41			from pyanaconda.modules.common.constants.services import NETWORK
42			from pyanaconda.threading import threadMgr, AnacondaThread
43			from org_fedora_oscap import utils
44			from org_fedora_oscap.data_fetch import fetch_data
45
46			log = logging.getLogger("anaconda")
47
48
49			# mimick pyanaconda/core/i18n.py
50			def _(string):
51			if string:
52			return gettext.translation("oscap-anaconda-addon", fallback=True).gettext(string)
53			else:
54			return ""
55
56
57			def N_(string): return string
58
59
60			# everything else should be private
61			__all__ = ["run_oscap_remediate", "get_fix_rules_pre",
62			"wait_and_fetch_net_data", "extract_data", "strip_content_dir",
63			"OSCAPaddonError"]
64
65			INSTALLATION_CONTENT_DIR = "/tmp/openscap_data/"
66			TARGET_CONTENT_DIR = "/root/openscap_data/"
67
68			SSG_DIR = "/usr/share/xml/scap/ssg/content/"
69
70			# Enable patches that set the content name at package-time
71			DEFAULT_SSG_CONTENT_NAME = ""
72			SSG_CONTENT = DEFAULT_SSG_CONTENT_NAME
73			if not SSG_CONTENT:
74			if constants.shortProductName != 'anaconda':
75			if constants.shortProductName == 'fedora':
76			SSG_CONTENT = "ssg-fedora-ds.xml"
77			else:
78			SSG_CONTENT = (
79			"ssg-{name}{version}-ds.xml"
80			.format(
81			name=constants.shortProductName,
82			version=constants.productVersion.strip(".")[0]))
83
84			RESULTS_PATH = utils.join_paths(TARGET_CONTENT_DIR,
85			"eval_remediate_results.xml")
86			REPORT_PATH = utils.join_paths(TARGET_CONTENT_DIR,
87			"eval_remediate_report.html")
88
89			PRE_INSTALL_FIX_SYSTEM_ATTR = "urn:redhat:anaconda:pre"
90
91			THREAD_FETCH_DATA = "AnaOSCAPdataFetchThread"
92
93			SUPPORTED_ARCHIVES = (".zip", ".tar", ".tar.gz", ".tar.bz2", )
94
95			# buffer size for reading and writing out data (in bytes)
96			IO_BUF_SIZE = 2 * 1024 * 1024
97
98
99			class OSCAPaddonError(Exception):
100			"""Exception class for OSCAP addon related errors."""
101
102			pass
103
104
105			class OSCAPaddonNetworkError(OSCAPaddonError):
106			"""Exception class for OSCAP addon related network errors."""
107
108			pass
109
110
111			class ExtractionError(OSCAPaddonError):
112			"""Exception class for the extraction errors."""
113
114			pass
115
116
117			MESSAGE_TYPE_FATAL = 0
118			MESSAGE_TYPE_WARNING = 1
119			MESSAGE_TYPE_INFO = 2
120
121			# namedtuple for messages returned from the rules evaluation
122			# origin -- class (inherited from RuleHandler) that generated the message
123			# type -- one of the MESSAGE_TYPE_* constants defined above
124			# text -- the actual message that should be displayed, logged, ...
125			RuleMessage = namedtuple("RuleMessage", ["origin", "type", "text"])
126
127
128			class SubprocessLauncher(object):
129			def __init__(self, args):
130			self.args = args
131			self.stdout = ""
132			self.stderr = ""
133			self.messages = []
134			self.returncode = None
135
136			def execute(self, ** kwargs):
137			command_string = " ".join(self.args)
138			log.info(
139			"OSCAP addon: Executing subprocess: '{command_string}'"
140			.format(command_string=command_string))
141			try:
142			proc = subprocess.Popen(self.args, stdout=subprocess.PIPE,
143			stderr=subprocess.PIPE, ** kwargs)
144			except OSError as oserr:
145			msg = "Failed to run the oscap tool: %s" % oserr
146			raise OSCAPaddonError(msg)
147
148			(stdout, stderr) = proc.communicate()
149			self.stdout = stdout.decode()
150			self.stderr = stderr.decode(errors="replace")
151			self.messages = re.findall(r'OpenSCAP Error:.*', self.stderr)
152			self.messages = self.messages + re.findall(r'E: oscap:.*', self.stderr)
153
154			self.returncode = proc.returncode
155
156			def log_messages(self):
157			for message in self.messages:
158			log.warning("OSCAP addon: " + message)
159
160
161			def get_fix_rules_pre(profile, fpath, ds_id="", xccdf_id="", tailoring=""):
162			"""
163			Get fix rules for the pre-installation environment for a given profile in a
164			given datastream and checklist in a given file.
165
166			:see: run_oscap_remediate
167			:see: _run_oscap_gen_fix
168			:return: fix rules for a given profile
169			:rtype: str
170
171			"""
172
173			return _run_oscap_gen_fix(profile, fpath, PRE_INSTALL_FIX_SYSTEM_ATTR,
174			ds_id=ds_id, xccdf_id=xccdf_id,
175			tailoring=tailoring)
176
177
178			def _run_oscap_gen_fix(profile, fpath, template, ds_id="", xccdf_id="",
179			tailoring=""):
180			"""
181			Run oscap tool on a given file to get the contents of fix elements with the
182			'system' attribute equal to a given template for a given datastream,
183			checklist and profile.
184
185			:see: run_oscap_remediate
186			:param template: the value of the 'system' attribute of the fix elements
187			:type template: str
188			:return: oscap tool's stdout
189			:rtype: str
190
191			"""
192
193			if not profile:
194			return ""
195
196			args = ["oscap", "xccdf", "generate", "fix"]
197			args.append("--template=%s" % template)
198
199			# oscap uses the default profile by default
200			if profile.lower() != "default":
201			args.append("--profile=%s" % profile)
202			if ds_id:
203			args.append("--datastream-id=%s" % ds_id)
204			if xccdf_id:
205			args.append("--xccdf-id=%s" % xccdf_id)
206			if tailoring:
207			args.append("--tailoring-file=%s" % tailoring)
208
209			args.append(fpath)
210
211			proc = SubprocessLauncher(args)
212			proc.execute()
213			proc.log_messages()
214			if proc.returncode != 0:
215			msg = "Failed to generate fix rules with the oscap tool: %s" % proc.stderr
216			raise OSCAPaddonError(msg)
217
218			return proc.stdout
219
220
221			def run_oscap_remediate(profile, fpath, ds_id="", xccdf_id="", tailoring="",
222			chroot=""):
223			"""
224			Run the evaluation and remediation with the oscap tool on a given file,
225			doing the remediation as defined in a given profile defined in a given
226			checklist that is a part of a given datastream. If requested, run in
227			chroot.
228
229			:param profile: id of the profile that will drive the remediation
230			:type profile: str
231			:param fpath: path to a file with SCAP content
232			:type fpath: str
233			:param ds_id: ID of the datastream that contains the checklist defining
234			the profile
235			:type ds_id: str
236			:param xccdf_id: ID of the checklist that defines the profile
237			:type xccdf_id: str
238			:param tailoring: path to a tailoring file
239			:type tailoring: str
240			:param chroot: path to the root the oscap tool should be run in
241			:type chroot: str
242			:return: oscap tool's stdout (summary of the rules, checks and fixes)
243			:rtype: str
244
245			"""
246
247			if not profile:
248			return ""
249
250			def do_chroot():
251			"""Helper function doing the chroot if requested."""
252			if chroot and chroot != "/":
253			os.chroot(chroot)
254			os.chdir("/")
255
256			# make sure the directory for the results exists
257			results_dir = os.path.dirname(RESULTS_PATH)
258			if chroot:
259			results_dir = os.path.normpath(chroot + "/" + results_dir)
260			utils.ensure_dir_exists(results_dir)
261
262			args = ["oscap", "xccdf", "eval"]
263			args.append("--remediate")
264			args.append("--results=%s" % RESULTS_PATH)
265			args.append("--report=%s" % REPORT_PATH)
266
267			# oscap uses the default profile by default
268			if profile.lower() != "default":
269			args.append("--profile=%s" % profile)
270			if ds_id:
271			args.append("--datastream-id=%s" % ds_id)
272			if xccdf_id:
273			args.append("--xccdf-id=%s" % xccdf_id)
274			if tailoring:
275			args.append("--tailoring-file=%s" % tailoring)
276
277			args.append(fpath)
278
279			proc = SubprocessLauncher(args)
280			proc.execute(preexec_fn=do_chroot)
281			proc.log_messages()
282
283			if proc.returncode not in (0, 2):
284			# 0 -- success; 2 -- no error, but checks/remediation failed
285			msg = "Content evaluation and remediation with the oscap tool "\
286			"failed: %s" % proc.stderr
287			raise OSCAPaddonError(msg)
288
289			return proc.stdout
290
291
292			def wait_and_fetch_net_data(url, out_file, ca_certs=None):
293			"""
294			Function that waits for network connection and starts a thread that fetches
295			data over network.
296
297			:see: org_fedora_oscap.data_fetch.fetch_data
298			:return: the name of the thread running fetch_data
299			:rtype: str
300
301			"""
302
303			# get thread that tries to establish a network connection
304			nm_conn_thread = threadMgr.get(constants.THREAD_WAIT_FOR_CONNECTING_NM)
305			if nm_conn_thread:
306			# NM still connecting, wait for it to finish
307			nm_conn_thread.join()
308
309			network_proxy = NETWORK.get_proxy()
310			if not network_proxy.Connected:
311			raise OSCAPaddonNetworkError("Network connection needed to fetch data.")
312
313			log.info("Fetching data from {url}".format(url=url))
314			fetch_data_thread = AnacondaThread(name=THREAD_FETCH_DATA,
315			target=fetch_data,
316			args=(url, out_file, ca_certs),
317			fatal=False)
318
319			# register and run the thread
320			threadMgr.add(fetch_data_thread)
321
322			return THREAD_FETCH_DATA
323
324
325			def extract_data(archive, out_dir, ensure_has_files=None):
326			"""
327			Fuction that extracts the given archive to the given output directory. It
328			tries to find out the archive type by the file name.
329
330			:param archive: path to the archive file that should be extracted
331			:type archive: str
332			:param out_dir: output directory the archive should be extracted to
333			:type out_dir: str
334			:param ensure_has_files: relative paths to the files that must exist in the
335			archive
336			:type ensure_has_files: iterable of strings or None
337			:return: a list of files and directories extracted from the archive
338			:rtype: [str]
339
340			"""
341
342			if not ensure_has_files:
343			ensure_has_files = []
344
345			# get rid of empty file paths
346			ensure_has_files = [fpath for fpath in ensure_has_files if fpath]
347
348			msg = "OSCAP addon: Extracting {archive}".format(archive=archive)
349			if ensure_has_files:
350			msg += ", expecting to find {files} there.".format(files=tuple(ensure_has_files))
351			log.info(msg)
352
353			result = []
354			if archive.endswith(".zip"):
355			# ZIP file
356			try:
357			zfile = zipfile.ZipFile(archive, "r")
358			except Exception as exc:
359			msg = "Error exctracting archive as a zipfile: {exc}".format(exc=str(exc))
360			raise ExtractionError(msg)
361
362			# generator for the paths of the files found in the archive (dirs end
363			# with "/")
364			files = set(info.filename for info in zfile.filelist
365			if not info.filename.endswith("/"))
366			for fpath in ensure_has_files or ():
367			if fpath not in files:
368			msg = "File '%s' not found in the archive '%s'" % (fpath,
369			archive)
370			raise ExtractionError(msg)
371
372			utils.ensure_dir_exists(out_dir)
373			zfile.extractall(path=out_dir)
374			result = [utils.join_paths(out_dir, info.filename) for info in zfile.filelist]
375			zfile.close()
376			elif archive.endswith(".tar"):
377			# plain tarball
378			result = _extract_tarball(archive, out_dir, ensure_has_files, None)
379			elif archive.endswith(".tar.gz"):
380			# gzipped tarball
381			result = _extract_tarball(archive, out_dir, ensure_has_files, "gz")
382			elif archive.endswith(".tar.bz2"):
383			# bzipped tarball
384			result = _extract_tarball(archive, out_dir, ensure_has_files, "bz2")
385			elif archive.endswith(".rpm"):
386			# RPM
387			result = _extract_rpm(archive, out_dir, ensure_has_files)
388			# elif other types of archives
389			else:
390			raise ExtractionError("Unsuported archive type")
391			log.info("OSCAP addon: Extracted {files} from the supplied content"
392			.format(files=result))
393			return result
394
395
396			def _extract_tarball(archive, out_dir, ensure_has_files, alg):
397			"""
398			Extract the given TAR archive to the given output directory and make sure
399			the given file exists in the archive.
400
401			:see: extract_data
402			:param alg: compression algorithm used for the tarball
403			:type alg: str (one of "gz", "bz2") or None
404			:return: a list of files and directories extracted from the archive
405			:rtype: [str]
406
407			"""
408
409			if alg and alg not in ("gz", "bz2",):
410			raise ExtractionError("Unsupported compression algorithm")
411
412			mode = "r"
413			if alg:
414			mode += ":%s" % alg
415
416			try:
417			tfile = tarfile.TarFile.open(archive, mode)
418			except tarfile.TarError as err:
419			raise ExtractionError(str(err))
420
421			# generator for the paths of the files found in the archive
422			files = set(member.path for member in tfile.getmembers()
423			if member.isfile())
424
425			for fpath in ensure_has_files or ():
426			if fpath not in files:
427			msg = "File '%s' not found in the archive '%s'" % (fpath, archive)
428			raise ExtractionError(msg)
429
430			utils.ensure_dir_exists(out_dir)
431			tfile.extractall(path=out_dir)
432			result = [utils.join_paths(out_dir, member.path) for member in tfile.getmembers()]
433			tfile.close()
434
435			return result
436
437
438			def _extract_rpm(rpm_path, root="/", ensure_has_files=None):
439			"""
440			Extract the given RPM into the directory tree given by the root argument
441			and make sure the given file exists in the archive.
442
443			:param rpm_path: path to the RPM file that should be extracted
444			:type rpm_path: str
445			:param root: root of the directory tree the RPM should be extracted into
446			:type root: str
447			:param ensure_has_files: relative paths to the files that must exist in the
448			RPM
449			:type ensure_has_files: iterable of strings or None
450			:return: a list of files and directories extracted from the archive
451			:rtype: [str]
452
453			"""
454
455			# run rpm2cpio and process the output with the cpioarchive module
456			temp_fd, temp_path = tempfile.mkstemp(prefix="oscap_rpm")
457			proc = subprocess.Popen(["rpm2cpio", rpm_path], stdout=temp_fd)
458			proc.wait()
459			if proc.returncode != 0:
460			msg = "Failed to convert RPM '%s' to cpio archive" % rpm_path
461			raise ExtractionError(msg)
462
463			os.close(temp_fd)
464
465			try:
466			archive = cpioarchive.CpioArchive(temp_path)
467			except cpioarchive.CpioError as err:
468			raise ExtractionError(str(err))
469
470			# get entries from the archive (supports only iteration over entries)
471			entries = set(entry for entry in archive)
472
473			# cpio entry names (paths) start with the dot
474			entry_names = [entry.name.lstrip(".") for entry in entries]
475
476			for fpath in ensure_has_files or ():
477			# RPM->cpio entries have absolute paths
478			if fpath not in entry_names and \
479			os.path.join("/", fpath) not in entry_names:
480			msg = "File '%s' not found in the archive '%s'" % (fpath, rpm_path)
481			raise ExtractionError(msg)
482
483			try:
484			for entry in entries:
485			if entry.size == 0:
486			continue
487			dirname = os.path.dirname(entry.name.lstrip("."))
488			out_dir = os.path.normpath(root + dirname)
489			utils.ensure_dir_exists(out_dir)
490
491			out_fpath = os.path.normpath(root + entry.name.lstrip("."))
492			if os.path.exists(out_fpath):
493			continue
494			with open(out_fpath, "wb") as out_file:
495			buf = entry.read(IO_BUF_SIZE)
496			while buf:
497			out_file.write(buf)
498			buf = entry.read(IO_BUF_SIZE)
499			except (IOError, cpioarchive.CpioError) as e:
500			raise ExtractionError(e)
501
502			# cleanup
503			archive.close()
504			os.unlink(temp_path)
505
506			return [os.path.normpath(root + name) for name in entry_names]
507
508
509			def strip_content_dir(fpaths, phase="preinst"):
510			"""
511			Strip content directory prefix from the file paths for either
512			pre-installation or post-installation phase.
513
514			:param fpaths: iterable of file paths to strip content directory prefix
515			from
516			:type fpaths: iterable of strings
517			:param phase: specifies pre-installation or post-installation phase
518			:type phase: "preinst" or "postinst"
519			:return: the same iterable of file paths as given with the content
520			directory prefix stripped
521			:rtype: same type as fpaths
522
523			"""
524
525			if phase == "preinst":
526			remove_prefix = lambda x: x[len(INSTALLATION_CONTENT_DIR):]
527			else:
528			remove_prefix = lambda x: x[len(TARGET_CONTENT_DIR):]
529
530			return utils.keep_type_map(remove_prefix, fpaths)
531
532
533			def get_ssg_path(root="/"):
534			return utils.join_paths(root, SSG_DIR + SSG_CONTENT)
535
536
537			def ssg_available(root="/"):
538			"""
539			Tries to find the SCAP Security Guide under the given root.
540
541			:return: True if SSG was found under the given root, False otherwise
542
543			"""
544
545			return os.path.exists(get_ssg_path(root))
546
547
548			def dry_run_skip(func):
549			"""
550			Decorator that makes sure the decorated function is noop in the dry-run
551			mode.
552
553			:param func: a decorated function that needs to have the first parameter an
554			object with the _addon_data attribute referencing the OSCAP
555			addon's ksdata
556			"""
557
558			@wraps(func)
559			def decorated(self, args, *kwargs):
560			if self._addon_data.dry_run:
561			return
562			else:
563			return func(self, args, *kwargs)
564
565			return decorated
566

OpenSCAP / oscap-anaconda-addon

Pull Request — rhel8-branch (#146)

org_fedora_oscap.common.extract_data() C

Complexity

Size

Duplication

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like