org_fedora_oscap.common.get_content_name() - Code Metrics - Inspection of "Merge pull request #136 from poncovka/master-dbus_..." - OpenSCAP/oscap-anaconda-addon - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( f4924a...c72d79 )

by Matěj

created 2021-04-22 14:05 UTC

org_fedora_oscap.common.get_content_name() A

↳ Parent: org_fedora_oscap.common

Complexity

Conditions

Size

Total Lines	15
Code Lines	12

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	5
eloc	12
nop	1
dl	0
loc	15
rs	9.3333
c	0
b	0
f	0

#
# 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


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

SSG_DIR = "/usr/share/xml/scap/ssg/content/"
SSG_CONTENT = "ssg-rhel7-ds.xml"
if constants.shortProductName != 'anaconda':
    if constants.shortProductName == 'fedora':
        SSG_CONTENT  = "ssg-fedora-ds.xml"
    else:
        SSG_CONTENT = "ssg-%s%s-ds.xml" % (constants.shortProductName,
                                            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", )

SUPPORTED_CONTENT_TYPES = (
    "datastream", "rpm", "archive", "scap-security-guide",
)

SUPPORTED_URL_PREFIXES = (
    "http://", "https://", "ftp://",  # LABEL:?, hdaX:?,
)

# 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):
        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.")

    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]

    """

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

    if archive.endswith(".zip"):
        # ZIP file
        try:
            zfile = zipfile.ZipFile(archive, "r")
        except zipfile.BadZipfile as err:
            raise ExtractionError(str(err))

        # 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()
        return result
    elif archive.endswith(".tar"):
        # plain tarball
        return _extract_tarball(archive, out_dir, ensure_has_files, None)
    elif archive.endswith(".tar.gz"):
        # gzipped tarball
        return _extract_tarball(archive, out_dir, ensure_has_files, "gz")
    elif archive.endswith(".tar.bz2"):
        # bzipped tarball
        return _extract_tarball(archive, out_dir, ensure_has_files, "bz2")
    elif archive.endswith(".rpm"):
        # RPM
        return _extract_rpm(archive, out_dir, ensure_has_files)
    # elif other types of archives
    else:
        raise ExtractionError("Unsuported archive type")


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 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(utils.join_paths(root, SSG_DIR + SSG_CONTENT))


def get_content_name(data):
    if data.content_type == "scap-security-guide":
        raise ValueError("Using scap-security-guide, no single content file")

    rest = "/anonymous_content"
    for prefix in SUPPORTED_URL_PREFIXES:
        if data.content_url.startswith(prefix):
            rest = data.content_url[len(prefix):]
            break

    parts = rest.rsplit("/", 1)
    if len(parts) != 2:
        raise ValueError("Unsupported url '%s'" % data.content_url)

    return parts[1]


def get_raw_preinst_content_path(data):
    """Path to the raw (unextracted, ...) pre-installation content file"""
    if data.content_type == "scap-security-guide":
        log.debug("Using scap-security-guide, no single content file")
        return None

    content_name = get_content_name(data)
    return utils.join_paths(INSTALLATION_CONTENT_DIR, content_name)


def get_preinst_content_path(data):
    """Path to the pre-installation content file"""
    if data.content_type == "scap-security-guide":
        # SSG is not copied to the standard place
        return data.content_path

    if data.content_type == "datastream":
        return utils.join_paths(
            INSTALLATION_CONTENT_DIR,
            get_content_name(data)
        )

    return utils.join_paths(
        INSTALLATION_CONTENT_DIR,
        data.content_path
    )


def get_postinst_content_path(data):
    """Path to the post-installation content file"""
    if data.content_type == "datastream":
        return utils.join_paths(
            TARGET_CONTENT_DIR,
            get_content_name(data)
        )

    if data.content_type in ("rpm", "scap-security-guide"):
        # no path magic in case of RPM (SSG is installed as an RPM)
        return data.content_path

    return utils.join_paths(
        TARGET_CONTENT_DIR,
        data.content_path
    )


def get_preinst_tailoring_path(data):
    """Path to the pre-installation tailoring file (if any)"""
    if not data.tailoring_path:
        return ""

    return utils.join_paths(
        INSTALLATION_CONTENT_DIR,
        data.tailoring_path
    )


def get_postinst_tailoring_path(data):
    """Path to the post-installation tailoring file (if any)"""
    if not data.tailoring_path:
        return ""

    if data.content_type == "rpm":
        # no path magic in case of RPM
        return data.tailoring_path

    return utils.join_paths(
        TARGET_CONTENT_DIR,
        data.tailoring_path
    )


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			INSTALLATION_CONTENT_DIR = "/tmp/openscap_data/"
61			TARGET_CONTENT_DIR = "/root/openscap_data/"
62
63			SSG_DIR = "/usr/share/xml/scap/ssg/content/"
64			SSG_CONTENT = "ssg-rhel7-ds.xml"
65			if constants.shortProductName != 'anaconda':
66			if constants.shortProductName == 'fedora':
67			SSG_CONTENT = "ssg-fedora-ds.xml"
68			else:
69			SSG_CONTENT = "ssg-%s%s-ds.xml" % (constants.shortProductName,
70			constants.productVersion.strip(".")[0])
71
72			RESULTS_PATH = utils.join_paths(TARGET_CONTENT_DIR,
73			"eval_remediate_results.xml")
74			REPORT_PATH = utils.join_paths(TARGET_CONTENT_DIR,
75			"eval_remediate_report.html")
76
77			PRE_INSTALL_FIX_SYSTEM_ATTR = "urn:redhat:anaconda:pre"
78
79			THREAD_FETCH_DATA = "AnaOSCAPdataFetchThread"
80
81			SUPPORTED_ARCHIVES = (".zip", ".tar", ".tar.gz", ".tar.bz2", )
82
83			SUPPORTED_CONTENT_TYPES = (
84			"datastream", "rpm", "archive", "scap-security-guide",
85			)
86
87			SUPPORTED_URL_PREFIXES = (
88			"http://", "https://", "ftp://", # LABEL:?, hdaX:?,
89			)
90
91			# buffer size for reading and writing out data (in bytes)
92			IO_BUF_SIZE = 2 * 1024 * 1024
93
94
95			class OSCAPaddonError(Exception):
96			"""Exception class for OSCAP addon related errors."""
97
98			pass
99
100
101			class OSCAPaddonNetworkError(OSCAPaddonError):
102			"""Exception class for OSCAP addon related network errors."""
103
104			pass
105
106
107			class ExtractionError(OSCAPaddonError):
108			"""Exception class for the extraction errors."""
109
110			pass
111
112
113			MESSAGE_TYPE_FATAL = 0
114			MESSAGE_TYPE_WARNING = 1
115			MESSAGE_TYPE_INFO = 2
116
117			# namedtuple for messages returned from the rules evaluation
118			# origin -- class (inherited from RuleHandler) that generated the message
119			# type -- one of the MESSAGE_TYPE_* constants defined above
120			# text -- the actual message that should be displayed, logged, ...
121			RuleMessage = namedtuple("RuleMessage", ["origin", "type", "text"])
122
123
124			class SubprocessLauncher(object):
125			def __init__(self, args):
126			self.args = args
127			self.stdout = ""
128			self.stderr = ""
129			self.messages = []
130			self.returncode = None
131
132			def execute(self, ** kwargs):
133			try:
134			proc = subprocess.Popen(self.args, stdout=subprocess.PIPE,
135			stderr=subprocess.PIPE, ** kwargs)
136			except OSError as oserr:
137			msg = "Failed to run the oscap tool: %s" % oserr
138			raise OSCAPaddonError(msg)
139
140			(stdout, stderr) = proc.communicate()
141			self.stdout = stdout.decode()
142			self.stderr = stderr.decode(errors="replace")
143			self.messages = re.findall(r'OpenSCAP Error:.*', self.stderr)
144			self.messages = self.messages + re.findall(r'E: oscap:.*', self.stderr)
145
146			self.returncode = proc.returncode
147
148			def log_messages(self):
149			for message in self.messages:
150			log.warning("OSCAP addon: " + message)
151
152
153			def get_fix_rules_pre(profile, fpath, ds_id="", xccdf_id="", tailoring=""):
154			"""
155			Get fix rules for the pre-installation environment for a given profile in a
156			given datastream and checklist in a given file.
157
158			:see: run_oscap_remediate
159			:see: _run_oscap_gen_fix
160			:return: fix rules for a given profile
161			:rtype: str
162
163			"""
164
165			return _run_oscap_gen_fix(profile, fpath, PRE_INSTALL_FIX_SYSTEM_ATTR,
166			ds_id=ds_id, xccdf_id=xccdf_id,
167			tailoring=tailoring)
168
169
170			def _run_oscap_gen_fix(profile, fpath, template, ds_id="", xccdf_id="",
171			tailoring=""):
172			"""
173			Run oscap tool on a given file to get the contents of fix elements with the
174			'system' attribute equal to a given template for a given datastream,
175			checklist and profile.
176
177			:see: run_oscap_remediate
178			:param template: the value of the 'system' attribute of the fix elements
179			:type template: str
180			:return: oscap tool's stdout
181			:rtype: str
182
183			"""
184
185			if not profile:
186			return ""
187
188			args = ["oscap", "xccdf", "generate", "fix"]
189			args.append("--template=%s" % template)
190
191			# oscap uses the default profile by default
192			if profile.lower() != "default":
193			args.append("--profile=%s" % profile)
194			if ds_id:
195			args.append("--datastream-id=%s" % ds_id)
196			if xccdf_id:
197			args.append("--xccdf-id=%s" % xccdf_id)
198			if tailoring:
199			args.append("--tailoring-file=%s" % tailoring)
200
201			args.append(fpath)
202
203			proc = SubprocessLauncher(args)
204			proc.execute()
205			proc.log_messages()
206			if proc.returncode != 0:
207			msg = "Failed to generate fix rules with the oscap tool: %s" % proc.stderr
208			raise OSCAPaddonError(msg)
209
210			return proc.stdout
211
212
213			def run_oscap_remediate(profile, fpath, ds_id="", xccdf_id="", tailoring="",
214			chroot=""):
215			"""
216			Run the evaluation and remediation with the oscap tool on a given file,
217			doing the remediation as defined in a given profile defined in a given
218			checklist that is a part of a given datastream. If requested, run in
219			chroot.
220
221			:param profile: id of the profile that will drive the remediation
222			:type profile: str
223			:param fpath: path to a file with SCAP content
224			:type fpath: str
225			:param ds_id: ID of the datastream that contains the checklist defining
226			the profile
227			:type ds_id: str
228			:param xccdf_id: ID of the checklist that defines the profile
229			:type xccdf_id: str
230			:param tailoring: path to a tailoring file
231			:type tailoring: str
232			:param chroot: path to the root the oscap tool should be run in
233			:type chroot: str
234			:return: oscap tool's stdout (summary of the rules, checks and fixes)
235			:rtype: str
236
237			"""
238
239			if not profile:
240			return ""
241
242			def do_chroot():
243			"""Helper function doing the chroot if requested."""
244			if chroot and chroot != "/":
245			os.chroot(chroot)
246			os.chdir("/")
247
248			# make sure the directory for the results exists
249			results_dir = os.path.dirname(RESULTS_PATH)
250			if chroot:
251			results_dir = os.path.normpath(chroot + "/" + results_dir)
252			utils.ensure_dir_exists(results_dir)
253
254			args = ["oscap", "xccdf", "eval"]
255			args.append("--remediate")
256			args.append("--results=%s" % RESULTS_PATH)
257			args.append("--report=%s" % REPORT_PATH)
258
259			# oscap uses the default profile by default
260			if profile.lower() != "default":
261			args.append("--profile=%s" % profile)
262			if ds_id:
263			args.append("--datastream-id=%s" % ds_id)
264			if xccdf_id:
265			args.append("--xccdf-id=%s" % xccdf_id)
266			if tailoring:
267			args.append("--tailoring-file=%s" % tailoring)
268
269			args.append(fpath)
270
271			proc = SubprocessLauncher(args)
272			proc.execute(preexec_fn=do_chroot)
273			proc.log_messages()
274
275			if proc.returncode not in (0, 2):
276			# 0 -- success; 2 -- no error, but checks/remediation failed
277			msg = "Content evaluation and remediation with the oscap tool "\
278			"failed: %s" % proc.stderr
279			raise OSCAPaddonError(msg)
280
281			return proc.stdout
282
283
284			def wait_and_fetch_net_data(url, out_file, ca_certs=None):
285			"""
286			Function that waits for network connection and starts a thread that fetches
287			data over network.
288
289			:see: org_fedora_oscap.data_fetch.fetch_data
290			:return: the name of the thread running fetch_data
291			:rtype: str
292
293			"""
294
295			# get thread that tries to establish a network connection
296			nm_conn_thread = threadMgr.get(constants.THREAD_WAIT_FOR_CONNECTING_NM)
297			if nm_conn_thread:
298			# NM still connecting, wait for it to finish
299			nm_conn_thread.join()
300
301			network_proxy = NETWORK.get_proxy()
302			if not network_proxy.Connected:
303			raise OSCAPaddonNetworkError("Network connection needed to fetch data.")
304
305			fetch_data_thread = AnacondaThread(name=THREAD_FETCH_DATA,
306			target=fetch_data,
307			args=(url, out_file, ca_certs),
308			fatal=False)
309
310			# register and run the thread
311			threadMgr.add(fetch_data_thread)
312
313			return THREAD_FETCH_DATA
314
315
316			def extract_data(archive, out_dir, ensure_has_files=None):
317			"""
318			Fuction that extracts the given archive to the given output directory. It
319			tries to find out the archive type by the file name.
320
321			:param archive: path to the archive file that should be extracted
322			:type archive: str
323			:param out_dir: output directory the archive should be extracted to
324			:type out_dir: str
325			:param ensure_has_files: relative paths to the files that must exist in the
326			archive
327			:type ensure_has_files: iterable of strings or None
328			:return: a list of files and directories extracted from the archive
329			:rtype: [str]
330
331			"""
332
333			# get rid of empty file paths
334			ensure_has_files = [fpath for fpath in ensure_has_files if fpath]
335
336			if archive.endswith(".zip"):
337			# ZIP file
338			try:
339			zfile = zipfile.ZipFile(archive, "r")
340			except zipfile.BadZipfile as err:
341			raise ExtractionError(str(err))
342
343			# generator for the paths of the files found in the archive (dirs end
344			# with "/")
345			files = set(info.filename for info in zfile.filelist
346			if not info.filename.endswith("/"))
347			for fpath in ensure_has_files or ():
348			if fpath not in files:
349			msg = "File '%s' not found in the archive '%s'" % (fpath,
350			archive)
351			raise ExtractionError(msg)
352
353			utils.ensure_dir_exists(out_dir)
354			zfile.extractall(path=out_dir)
355			result = [utils.join_paths(out_dir, info.filename) for info in zfile.filelist]
356			zfile.close()
357			return result
358			elif archive.endswith(".tar"):
359			# plain tarball
360			return _extract_tarball(archive, out_dir, ensure_has_files, None)
361			elif archive.endswith(".tar.gz"):
362			# gzipped tarball
363			return _extract_tarball(archive, out_dir, ensure_has_files, "gz")
364			elif archive.endswith(".tar.bz2"):
365			# bzipped tarball
366			return _extract_tarball(archive, out_dir, ensure_has_files, "bz2")
367			elif archive.endswith(".rpm"):
368			# RPM
369			return _extract_rpm(archive, out_dir, ensure_has_files)
370			# elif other types of archives
371			else:
372			raise ExtractionError("Unsuported archive type")
373
374
375			def _extract_tarball(archive, out_dir, ensure_has_files, alg):
376			"""
377			Extract the given TAR archive to the given output directory and make sure
378			the given file exists in the archive.
379
380			:see: extract_data
381			:param alg: compression algorithm used for the tarball
382			:type alg: str (one of "gz", "bz2") or None
383			:return: a list of files and directories extracted from the archive
384			:rtype: [str]
385
386			"""
387
388			if alg and alg not in ("gz", "bz2",):
389			raise ExtractionError("Unsupported compression algorithm")
390
391			mode = "r"
392			if alg:
393			mode += ":%s" % alg
394
395			try:
396			tfile = tarfile.TarFile.open(archive, mode)
397			except tarfile.TarError as err:
398			raise ExtractionError(str(err))
399
400			# generator for the paths of the files found in the archive
401			files = set(member.path for member in tfile.getmembers()
402			if member.isfile())
403
404			for fpath in ensure_has_files or ():
405			if fpath not in files:
406			msg = "File '%s' not found in the archive '%s'" % (fpath, archive)
407			raise ExtractionError(msg)
408
409			utils.ensure_dir_exists(out_dir)
410			tfile.extractall(path=out_dir)
411			result = [utils.join_paths(out_dir, member.path) for member in tfile.getmembers()]
412			tfile.close()
413
414			return result
415
416
417			def _extract_rpm(rpm_path, root="/", ensure_has_files=None):
418			"""
419			Extract the given RPM into the directory tree given by the root argument
420			and make sure the given file exists in the archive.
421
422			:param rpm_path: path to the RPM file that should be extracted
423			:type rpm_path: str
424			:param root: root of the directory tree the RPM should be extracted into
425			:type root: str
426			:param ensure_has_files: relative paths to the files that must exist in the
427			RPM
428			:type ensure_has_files: iterable of strings or None
429			:return: a list of files and directories extracted from the archive
430			:rtype: [str]
431
432			"""
433
434			# run rpm2cpio and process the output with the cpioarchive module
435			temp_fd, temp_path = tempfile.mkstemp(prefix="oscap_rpm")
436			proc = subprocess.Popen(["rpm2cpio", rpm_path], stdout=temp_fd)
437			proc.wait()
438			if proc.returncode != 0:
439			msg = "Failed to convert RPM '%s' to cpio archive" % rpm_path
440			raise ExtractionError(msg)
441
442			os.close(temp_fd)
443
444			try:
445			archive = cpioarchive.CpioArchive(temp_path)
446			except cpioarchive.CpioError as err:
447			raise ExtractionError(str(err))
448
449			# get entries from the archive (supports only iteration over entries)
450			entries = set(entry for entry in archive)
451
452			# cpio entry names (paths) start with the dot
453			entry_names = [entry.name.lstrip(".") for entry in entries]
454
455			for fpath in ensure_has_files or ():
456			# RPM->cpio entries have absolute paths
457			if fpath not in entry_names and \
458			os.path.join("/", fpath) not in entry_names:
459			msg = "File '%s' not found in the archive '%s'" % (fpath, rpm_path)
460			raise ExtractionError(msg)
461
462			try:
463			for entry in entries:
464			if entry.size == 0:
465			continue
466			dirname = os.path.dirname(entry.name.lstrip("."))
467			out_dir = os.path.normpath(root + dirname)
468			utils.ensure_dir_exists(out_dir)
469
470			out_fpath = os.path.normpath(root + entry.name.lstrip("."))
471			if os.path.exists(out_fpath):
472			continue
473			with open(out_fpath, "wb") as out_file:
474			buf = entry.read(IO_BUF_SIZE)
475			while buf:
476			out_file.write(buf)
477			buf = entry.read(IO_BUF_SIZE)
478			except (IOError, cpioarchive.CpioError) as e:
479			raise ExtractionError(e)
480
481			# cleanup
482			archive.close()
483			os.unlink(temp_path)
484
485			return [os.path.normpath(root + name) for name in entry_names]
486
487
488			def strip_content_dir(fpaths, phase="preinst"):
489			"""
490			Strip content directory prefix from the file paths for either
491			pre-installation or post-installation phase.
492
493			:param fpaths: iterable of file paths to strip content directory prefix
494			from
495			:type fpaths: iterable of strings
496			:param phase: specifies pre-installation or post-installation phase
497			:type phase: "preinst" or "postinst"
498			:return: the same iterable of file paths as given with the content
499			directory prefix stripped
500			:rtype: same type as fpaths
501
502			"""
503
504			if phase == "preinst":
505			remove_prefix = lambda x: x[len(INSTALLATION_CONTENT_DIR):]
506			else:
507			remove_prefix = lambda x: x[len(TARGET_CONTENT_DIR):]
508
509			return utils.keep_type_map(remove_prefix, fpaths)
510
511
512			def ssg_available(root="/"):
513			"""
514			Tries to find the SCAP Security Guide under the given root.
515
516			:return: True if SSG was found under the given root, False otherwise
517
518			"""
519
520			return os.path.exists(utils.join_paths(root, SSG_DIR + SSG_CONTENT))
521
522
523			def get_content_name(data):
524			if data.content_type == "scap-security-guide":
525			raise ValueError("Using scap-security-guide, no single content file")
526
527			rest = "/anonymous_content"
528			for prefix in SUPPORTED_URL_PREFIXES:
529			if data.content_url.startswith(prefix):
530			rest = data.content_url[len(prefix):]
531			break
532
533			parts = rest.rsplit("/", 1)
534			if len(parts) != 2:
535			raise ValueError("Unsupported url '%s'" % data.content_url)
536
537			return parts[1]
538
539
540			def get_raw_preinst_content_path(data):
541			"""Path to the raw (unextracted, ...) pre-installation content file"""
542			if data.content_type == "scap-security-guide":
543			log.debug("Using scap-security-guide, no single content file")
544			return None
545
546			content_name = get_content_name(data)
547			return utils.join_paths(INSTALLATION_CONTENT_DIR, content_name)
548
549
550			def get_preinst_content_path(data):
551			"""Path to the pre-installation content file"""
552			if data.content_type == "scap-security-guide":
553			# SSG is not copied to the standard place
554			return data.content_path
555
556			if data.content_type == "datastream":
557			return utils.join_paths(
558			INSTALLATION_CONTENT_DIR,
559			get_content_name(data)
560			)
561
562			return utils.join_paths(
563			INSTALLATION_CONTENT_DIR,
564			data.content_path
565			)
566
567
568			def get_postinst_content_path(data):
569			"""Path to the post-installation content file"""
570			if data.content_type == "datastream":
571			return utils.join_paths(
572			TARGET_CONTENT_DIR,
573			get_content_name(data)
574			)
575
576			if data.content_type in ("rpm", "scap-security-guide"):
577			# no path magic in case of RPM (SSG is installed as an RPM)
578			return data.content_path
579
580			return utils.join_paths(
581			TARGET_CONTENT_DIR,
582			data.content_path
583			)
584
585
586			def get_preinst_tailoring_path(data):
587			"""Path to the pre-installation tailoring file (if any)"""
588			if not data.tailoring_path:
589			return ""
590
591			return utils.join_paths(
592			INSTALLATION_CONTENT_DIR,
593			data.tailoring_path
594			)
595
596
597			def get_postinst_tailoring_path(data):
598			"""Path to the post-installation tailoring file (if any)"""
599			if not data.tailoring_path:
600			return ""
601
602			if data.content_type == "rpm":
603			# no path magic in case of RPM
604			return data.tailoring_path
605
606			return utils.join_paths(
607			TARGET_CONTENT_DIR,
608			data.tailoring_path
609			)
610

OpenSCAP / oscap-anaconda-addon

Push — master ( f4924a...c72d79 )

org_fedora_oscap.common.get_content_name() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like