org_fedora_oscap.service.kickstart.OSCAPKickstartData.__init__() - Code Metrics - Inspection of "Create the OSCAP DBus module" - OpenSCAP/oscap-anaconda-addon - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#136)

unknown

created 2021-02-05 14:35 UTC

OSCAPKickstartData.init() A

↳ Parent: org_fedora_oscap.service.kickstart

Complexity

Conditions

Size

Total Lines	17
Code Lines	12

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	12
nop	1
dl	0
loc	17
rs	9.8
c	0
b	0
f	0

#
# Copyright (C) 2020 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.
#
import logging
import re

from pyanaconda.core.kickstart import KickstartSpecification
from pyanaconda.core.kickstart.addon import AddonData
from pykickstart.errors import KickstartValueError, KickstartParseError

from org_fedora_oscap import common, utils

log = logging.getLogger(__name__)

__all__ = ["OSCAPKickstartSpecification"]


FINGERPRINT_REGEX = re.compile(r'^[a-z0-9]+$')


def key_value_pair(key, value, indent=4):
    return "%s%s = %s" % (indent * " ", key, value)


class OSCAPKickstartData(AddonData):
    """The kickstart data for the org_fedora_oscap add-on."""

    def __init__(self):
        super().__init__()
        # values specifying the content
        self.content_type = ""
        self.content_url = ""
        self.datastream_id = ""
        self.xccdf_id = ""
        self.profile_id = ""
        self.content_path = ""
        self.cpe_path = ""
        self.tailoring_path = ""

        # additional values
        self.fingerprint = ""

        # certificate to verify HTTPS connection or signed data
        self.certificates = ""

    @property
    def name(self):
        """The name of the %addon section."""
        return "org_fedora_oscap"

    def handle_header(self, args, line_number=None):
        """Handle the arguments of the %addon line.

        :param args: a list of additional arguments
        :param line_number: a line number
        :raise: KickstartParseError for invalid arguments
        """
        pass

    def handle_line(self, line, line_number=None):
        """Handle one line of the section.

        :param line: a line to parse
        :param line_number: a line number
        :raise: KickstartParseError for invalid lines
        """
        actions = {
            "content-type": self._parse_content_type,
            "content-url": self._parse_content_url,
            "content-path": self._parse_content_path,
            "datastream-id": self._parse_datastream_id,
            "profile": self._parse_profile_id,
            "xccdf-id": self._parse_xccdf_id,
            "xccdf-path": self._parse_content_path,
            "cpe-path": self._parse_cpe_path,
            "tailoring-path": self._parse_tailoring_path,
            "fingerprint": self._parse_fingerprint,
            "certificates": self._parse_certificates,
        }

        line = line.strip()
        (pre, sep, post) = line.partition("=")
        pre = pre.strip()
        post = post.strip()
        post = post.strip('"')

        try:
            actions[pre](post)
        except KeyError:
            msg = "Unknown item '%s' for %s addon" % (line, self.name)
            raise KickstartParseError(msg)

    def _parse_content_type(self, value):
        value_low = value.lower()
        if value_low in common.SUPPORTED_CONTENT_TYPES:
            self.content_type = value_low
        else:
            msg = "Unsupported content type '%s' in the %s addon" % (value,
                                                                     self.name)
            raise KickstartValueError(msg)

    def _parse_content_url(self, value):
        if any(value.startswith(prefix)
               for prefix in common.SUPPORTED_URL_PREFIXES):
            self.content_url = value
        else:
            msg = "Unsupported url '%s' in the %s addon" % (value, self.name)
            raise KickstartValueError(msg)

    def _parse_datastream_id(self, value):
        # need to be checked?
        self.datastream_id = value

    def _parse_xccdf_id(self, value):
        # need to be checked?
        self.xccdf_id = value

    def _parse_profile_id(self, value):
        # need to be checked?
        self.profile_id = value

    def _parse_content_path(self, value):
        # need to be checked?
        self.content_path = value

    def _parse_cpe_path(self, value):
        # need to be checked?
        self.cpe_path = value

    def _parse_tailoring_path(self, value):
        # need to be checked?
        self.tailoring_path = value

    def _parse_fingerprint(self, value):
        if FINGERPRINT_REGEX.match(value) is None:
            msg = "Unsupported or invalid fingerprint"
            raise KickstartValueError(msg)

        if utils.get_hashing_algorithm(value) is None:
            msg = "Unsupported fingerprint"
            raise KickstartValueError(msg)

        self.fingerprint = value

    def _parse_certificates(self, value):
        self.certificates = value

    def handle_end(self):
        """Handle the end of the section."""
        tmpl = "%s missing for the %s addon"

        # check provided data
        if not self.content_type:
            raise KickstartValueError(tmpl % ("content-type", self.name))

        if self.content_type != "scap-security-guide" and not self.content_url:
            raise KickstartValueError(tmpl % ("content-url", self.name))

        if not self.profile_id:
            self.profile_id = "default"

        if self.content_type in ("rpm", "archive") and not self.content_path:
            msg = "Path to the XCCDF file has to be given if content in RPM "\
                  "or archive is used"
            raise KickstartValueError(msg)

        if self.content_type == "rpm" and not self.content_url.endswith(".rpm"):
            msg = "Content type set to RPM, but the content URL doesn't end "\
                  "with '.rpm'"
            raise KickstartValueError(msg)

        if self.content_type == "archive":
            supported_archive = any(
                self.content_url.endswith(arch_type)
                for arch_type in common.SUPPORTED_ARCHIVES
            )
            if not supported_archive:
                msg = "Unsupported archive type of the content "\
                      "file '%s'" % self.content_url
                raise KickstartValueError(msg)

        # do some initialization magic in case of SSG
        if self.content_type == "scap-security-guide":
            if not common.ssg_available():
                msg = "SCAP Security Guide not found on the system"
                raise KickstartValueError(msg)

            self.content_path = common.SSG_DIR + common.SSG_CONTENT

    def __str__(self):
        """Generate the kickstart representation.

        What should end up in the resulting kickstart file,
        i.e. string representation of the stored data.

        :return: a string
        """
        if not self.profile_id:
            return ""

        ret = "%%addon %s" % self.name
        ret += "\n%s" % key_value_pair("content-type", self.content_type)

        if self.content_url:
            ret += "\n%s" % key_value_pair("content-url", self.content_url)

        if self.datastream_id:
            ret += "\n%s" % key_value_pair("datastream-id", self.datastream_id)

        if self.xccdf_id:
            ret += "\n%s" % key_value_pair("xccdf-id", self.xccdf_id)

        if self.content_path and self.content_type != "scap-security-guide":
            ret += "\n%s" % key_value_pair("content-path", self.content_path)

        if self.cpe_path:
            ret += "\n%s" % key_value_pair("cpe-path", self.cpe_path)

        if self.tailoring_path:
            ret += "\n%s" % key_value_pair("tailoring-path", self.tailoring_path)

        ret += "\n%s" % key_value_pair("profile", self.profile_id)

        if self.fingerprint:
            ret += "\n%s" % key_value_pair("fingerprint", self.fingerprint)

        if self.certificates:
            ret += "\n%s" % key_value_pair("certificates", self.certificates)

        ret += "\n%end\n\n"
        return ret


class OSCAPKickstartSpecification(KickstartSpecification):
    """The kickstart specification of the OSCAP service."""

    addons = {
        "org_fedora_oscap": OSCAPKickstartData
    }


1			#
2			# Copyright (C) 2020 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			import logging
19			import re
20
21			from pyanaconda.core.kickstart import KickstartSpecification
22			from pyanaconda.core.kickstart.addon import AddonData
23			from pykickstart.errors import KickstartValueError, KickstartParseError
24
25			from org_fedora_oscap import common, utils
26
27			log = logging.getLogger(__name__)
28
29			__all__ = ["OSCAPKickstartSpecification"]
30
31
32			FINGERPRINT_REGEX = re.compile(r'^[a-z0-9]+$')
33
34
35			def key_value_pair(key, value, indent=4):
36			return "%s%s = %s" % (indent * " ", key, value)
37
38
39			class OSCAPKickstartData(AddonData):
40			"""The kickstart data for the org_fedora_oscap add-on."""
41
42			def __init__(self):
43			super().__init__()
44			# values specifying the content
45			self.content_type = ""
46			self.content_url = ""
47			self.datastream_id = ""
48			self.xccdf_id = ""
49			self.profile_id = ""
50			self.content_path = ""
51			self.cpe_path = ""
52			self.tailoring_path = ""
53
54			# additional values
55			self.fingerprint = ""
56
57			# certificate to verify HTTPS connection or signed data
58			self.certificates = ""
59
60			@property
61			def name(self):
62			"""The name of the %addon section."""
63			return "org_fedora_oscap"
64
65			def handle_header(self, args, line_number=None):
66			"""Handle the arguments of the %addon line.
67
68			:param args: a list of additional arguments
69			:param line_number: a line number
70			:raise: KickstartParseError for invalid arguments
71			"""
72			pass
73
74			def handle_line(self, line, line_number=None):
75			"""Handle one line of the section.
76
77			:param line: a line to parse
78			:param line_number: a line number
79			:raise: KickstartParseError for invalid lines
80			"""
81			actions = {
82			"content-type": self._parse_content_type,
83			"content-url": self._parse_content_url,
84			"content-path": self._parse_content_path,
85			"datastream-id": self._parse_datastream_id,
86			"profile": self._parse_profile_id,
87			"xccdf-id": self._parse_xccdf_id,
88			"xccdf-path": self._parse_content_path,
89			"cpe-path": self._parse_cpe_path,
90			"tailoring-path": self._parse_tailoring_path,
91			"fingerprint": self._parse_fingerprint,
92			"certificates": self._parse_certificates,
93			}
94
95			line = line.strip()
96			(pre, sep, post) = line.partition("=")
97			pre = pre.strip()
98			post = post.strip()
99			post = post.strip('"')
100
101			try:
102			actions[pre](post)
103			except KeyError:
104			msg = "Unknown item '%s' for %s addon" % (line, self.name)
105			raise KickstartParseError(msg)
106
107			def _parse_content_type(self, value):
108			value_low = value.lower()
109			if value_low in common.SUPPORTED_CONTENT_TYPES:
110			self.content_type = value_low
111			else:
112			msg = "Unsupported content type '%s' in the %s addon" % (value,
113			self.name)
114			raise KickstartValueError(msg)
115
116			def _parse_content_url(self, value):
117			if any(value.startswith(prefix)
118			for prefix in common.SUPPORTED_URL_PREFIXES):
119			self.content_url = value
120			else:
121			msg = "Unsupported url '%s' in the %s addon" % (value, self.name)
122			raise KickstartValueError(msg)
123
124			def _parse_datastream_id(self, value):
125			# need to be checked?
126			self.datastream_id = value
127
128			def _parse_xccdf_id(self, value):
129			# need to be checked?
130			self.xccdf_id = value
131
132			def _parse_profile_id(self, value):
133			# need to be checked?
134			self.profile_id = value
135
136			def _parse_content_path(self, value):
137			# need to be checked?
138			self.content_path = value
139
140			def _parse_cpe_path(self, value):
141			# need to be checked?
142			self.cpe_path = value
143
144			def _parse_tailoring_path(self, value):
145			# need to be checked?
146			self.tailoring_path = value
147
148			def _parse_fingerprint(self, value):
149			if FINGERPRINT_REGEX.match(value) is None:
150			msg = "Unsupported or invalid fingerprint"
151			raise KickstartValueError(msg)
152
153			if utils.get_hashing_algorithm(value) is None:
154			msg = "Unsupported fingerprint"
155			raise KickstartValueError(msg)
156
157			self.fingerprint = value
158
159			def _parse_certificates(self, value):
160			self.certificates = value
161
162			def handle_end(self):
163			"""Handle the end of the section."""
164			tmpl = "%s missing for the %s addon"
165
166			# check provided data
167			if not self.content_type:
168			raise KickstartValueError(tmpl % ("content-type", self.name))
169
170			if self.content_type != "scap-security-guide" and not self.content_url:
171			raise KickstartValueError(tmpl % ("content-url", self.name))
172
173			if not self.profile_id:
174			self.profile_id = "default"
175
176			if self.content_type in ("rpm", "archive") and not self.content_path:
177			msg = "Path to the XCCDF file has to be given if content in RPM "\
178			"or archive is used"
179			raise KickstartValueError(msg)
180
181			if self.content_type == "rpm" and not self.content_url.endswith(".rpm"):
182			msg = "Content type set to RPM, but the content URL doesn't end "\
183			"with '.rpm'"
184			raise KickstartValueError(msg)
185
186			if self.content_type == "archive":
187			supported_archive = any(
188			self.content_url.endswith(arch_type)
189			for arch_type in common.SUPPORTED_ARCHIVES
190			)
191			if not supported_archive:
192			msg = "Unsupported archive type of the content "\
193			"file '%s'" % self.content_url
194			raise KickstartValueError(msg)
195
196			# do some initialization magic in case of SSG
197			if self.content_type == "scap-security-guide":
198			if not common.ssg_available():
199			msg = "SCAP Security Guide not found on the system"
200			raise KickstartValueError(msg)
201
202			self.content_path = common.SSG_DIR + common.SSG_CONTENT
203
204			def __str__(self):
205			"""Generate the kickstart representation.
206
207			What should end up in the resulting kickstart file,
208			i.e. string representation of the stored data.
209
210			:return: a string
211			"""
212			if not self.profile_id:
213			return ""
214
215			ret = "%%addon %s" % self.name
216			ret += "\n%s" % key_value_pair("content-type", self.content_type)
217
218			if self.content_url:
219			ret += "\n%s" % key_value_pair("content-url", self.content_url)
220
221			if self.datastream_id:
222			ret += "\n%s" % key_value_pair("datastream-id", self.datastream_id)
223
224			if self.xccdf_id:
225			ret += "\n%s" % key_value_pair("xccdf-id", self.xccdf_id)
226
227			if self.content_path and self.content_type != "scap-security-guide":
228			ret += "\n%s" % key_value_pair("content-path", self.content_path)
229
230			if self.cpe_path:
231			ret += "\n%s" % key_value_pair("cpe-path", self.cpe_path)
232
233			if self.tailoring_path:
234			ret += "\n%s" % key_value_pair("tailoring-path", self.tailoring_path)
235
236			ret += "\n%s" % key_value_pair("profile", self.profile_id)
237
238			if self.fingerprint:
239			ret += "\n%s" % key_value_pair("fingerprint", self.fingerprint)
240
241			if self.certificates:
242			ret += "\n%s" % key_value_pair("certificates", self.certificates)
243
244			ret += "\n%end\n\n"
245			return ret
246
247
248			class OSCAPKickstartSpecification(KickstartSpecification):
249			"""The kickstart specification of the OSCAP service."""
250
251			addons = {
252			"org_fedora_oscap": OSCAPKickstartData
253			}
254

OpenSCAP / oscap-anaconda-addon

Pull Request — master (#136)

OSCAPKickstartData.__init__() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

OSCAPKickstartData.init() A