ospd_openvas.vthelper.VtHelper.get_notus_driver_oids() - Code Metrics - Inspection of "Don't expose notus driver VTs to clients" - greenbone/ospd-openvas - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#367)

by Juan José

created 2021-01-04 08:14 UTC

VtHelper.get_notus_driver_oids() A

↳ Parent: ospd_openvas.vthelper

Complexity

Conditions

Size

Total Lines	8
Code Lines	4

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	4
nop	1
dl	0
loc	8
rs	10
c	0
b	0
f	0

# -*- coding: utf-8 -*-
# Copyright (C) 2014-2020 Greenbone Networks GmbH
#
# SPDX-License-Identifier: AGPL-3.0-or-later
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.


""" Provide functions to handle VT Info. """

from hashlib import sha256
from typing import Optional, Dict, List, Tuple, Iterator

from ospd_openvas.nvticache import NVTICache
from ospd_openvas.notus.metadata import NotusMetadataHandler


class VtHelper:
    def __init__(self, nvticache: NVTICache):
        self.nvti = nvticache

    def get_single_vt(self, vt_id: str, oids=None) -> Optional[Dict[str, any]]:
        custom = self.nvti.get_nvt_metadata(vt_id)

        if not custom:
            return None

        vt_params = custom.pop('vt_params')
        vt_refs = custom.pop('refs')
        name = custom.pop('name')
        vt_creation_time = custom.pop('creation_date')
        vt_modification_time = custom.pop('last_modification')

        if oids:
            vt_dependencies = list()
            if 'dependencies' in custom:
                deps = custom.pop('dependencies')
                deps_list = deps.split(', ')
                for dep in deps_list:
                    vt_dependencies.append(oids.get(dep))
        else:
            vt_dependencies = None

        summary = None
        impact = None
        affected = None
        insight = None
        solution = None
        solution_t = None
        solution_m = None
        vuldetect = None
        qod_t = None
        qod_v = None

        if 'summary' in custom:
            summary = custom.pop('summary')
        if 'impact' in custom:
            impact = custom.pop('impact')
        if 'affected' in custom:
            affected = custom.pop('affected')
        if 'insight' in custom:
            insight = custom.pop('insight')
        if 'solution' in custom:
            solution = custom.pop('solution')
            if 'solution_type' in custom:
                solution_t = custom.pop('solution_type')
            if 'solution_method' in custom:
                solution_m = custom.pop('solution_method')

        if 'vuldetect' in custom:
            vuldetect = custom.pop('vuldetect')
        if 'qod_type' in custom:
            qod_t = custom.pop('qod_type')
        elif 'qod' in custom:
            qod_v = custom.pop('qod')

        severity = dict()
        if 'severity_vector' in custom:
            severity_vector = custom.pop('severity_vector')
        else:
            severity_vector = custom.pop('cvss_base_vector')
        severity['severity_base_vector'] = severity_vector

        if "CVSS:3" in severity_vector:
            severity_type = 'cvss_base_v3'
        else:
            severity_type = 'cvss_base_v2'
        severity['severity_type'] = severity_type

        if 'severity_date' in custom:
            severity['severity_date'] = custom.pop('severity_date')
        else:
            severity['severity_date'] = vt_creation_time

        if 'severity_origin' in custom:
            severity['severity_origin'] = custom.pop('severity_origin')

        if name is None:
            name = ''

        vt = {'name': name}
        if custom is not None:
            vt["custom"] = custom
        if vt_params is not None:
            vt["vt_params"] = vt_params
        if vt_refs is not None:
            vt["vt_refs"] = vt_refs
        if vt_dependencies is not None:
            vt["vt_dependencies"] = vt_dependencies
        if vt_creation_time is not None:
            vt["creation_time"] = vt_creation_time
        if vt_modification_time is not None:
            vt["modification_time"] = vt_modification_time
        if summary is not None:
            vt["summary"] = summary
        if impact is not None:
            vt["impact"] = impact
        if affected is not None:
            vt["affected"] = affected
        if insight is not None:
            vt["insight"] = insight

        if solution is not None:
            vt["solution"] = solution
            if solution_t is not None:
                vt["solution_type"] = solution_t
            if solution_m is not None:
                vt["solution_method"] = solution_m

        if vuldetect is not None:
            vt["detection"] = vuldetect

        if qod_t is not None:
            vt["qod_type"] = qod_t
        elif qod_v is not None:
            vt["qod"] = qod_v

        if severity is not None:
            vt["severities"] = severity

        return vt

    def get_notus_driver_oids(self) -> List[str]:
        """Return a list of oid corresponding to notus driver which
        are considered backend entities and must not be exposed to
        the OSP client."""
        notus = NotusMetadataHandler()
        lsc_families_and_drivers = notus.get_family_driver_linkers()

        return lsc_families_and_drivers.values()

    def get_vt_iterator(
        self, vt_selection: List[str] = None, details: bool = True
    ) -> Iterator[Tuple[str, Dict]]:
        """ Yield the vts from the Redis NVTicache. """

        oids = None
        if not vt_selection or details:
            vt_collection = dict(self.nvti.get_oids())

            if not vt_selection:
                vt_selection = vt_collection.values()

            if details:
                oids = vt_collection

        # Notus driver oid list which are not sent.
        drivers = self.get_notus_driver_oids()
        for vt_id in vt_selection:
            if vt_id in drivers:
                continue
            vt = self.get_single_vt(vt_id, oids)
            yield (vt_id, vt)

    def calculate_vts_collection_hash(self) -> str:
        """ Calculate the vts collection sha256 hash. """
        m = sha256()  # pylint: disable=invalid-name

        # Notus driver oid list which are not sent.
        drivers = self.get_notus_driver_oids()

        # for a reproducible hash calculation
        # the vts must already be sorted in the dictionary.
        for vt_id, vt in self.get_vt_iterator(details=False):
            if vt_id in drivers:
                continue
            param_chain = ""
            vt_params = vt.get('vt_params')
            if vt_params:
                for _, param in sorted(vt_params.items()):
                    param_chain += (
                        param.get('id')
                        + param.get('name')
                        + param.get('default')
                    )

            m.update(
                (vt_id + vt.get('modification_time')).encode('utf-8')
                + param_chain.encode('utf-8')
            )

        return m.hexdigest()


1			# -- coding: utf-8 --
2			# Copyright (C) 2014-2020 Greenbone Networks GmbH
3			#
4			# SPDX-License-Identifier: AGPL-3.0-or-later
5			#
6			# This program is free software: you can redistribute it and/or modify
7			# it under the terms of the GNU Affero General Public License as
8			# published by the Free Software Foundation, either version 3 of the
9			# License, or (at your option) any later version.
10			#
11			# This program is distributed in the hope that it will be useful,
12			# but WITHOUT ANY WARRANTY; without even the implied warranty of
13			# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
14			# GNU Affero General Public License for more details.
15			#
16			# You should have received a copy of the GNU Affero General Public License
17			# along with this program. If not, see <http://www.gnu.org/licenses/>.
18
19
20			""" Provide functions to handle VT Info. """
21
22			from hashlib import sha256
23			from typing import Optional, Dict, List, Tuple, Iterator
24
25			from ospd_openvas.nvticache import NVTICache
26			from ospd_openvas.notus.metadata import NotusMetadataHandler
27
28
29			class VtHelper:
30			def __init__(self, nvticache: NVTICache):
31			self.nvti = nvticache
32
33			def get_single_vt(self, vt_id: str, oids=None) -> Optional[Dict[str, any]]:
34			custom = self.nvti.get_nvt_metadata(vt_id)
35
36			if not custom:
37			return None
38
39			vt_params = custom.pop('vt_params')
40			vt_refs = custom.pop('refs')
41			name = custom.pop('name')
42			vt_creation_time = custom.pop('creation_date')
43			vt_modification_time = custom.pop('last_modification')
44
45			if oids:
46			vt_dependencies = list()
47			if 'dependencies' in custom:
48			deps = custom.pop('dependencies')
49			deps_list = deps.split(', ')
50			for dep in deps_list:
51			vt_dependencies.append(oids.get(dep))
52			else:
53			vt_dependencies = None
54
55			summary = None
56			impact = None
57			affected = None
58			insight = None
59			solution = None
60			solution_t = None
61			solution_m = None
62			vuldetect = None
63			qod_t = None
64			qod_v = None
65
66			if 'summary' in custom:
67			summary = custom.pop('summary')
68			if 'impact' in custom:
69			impact = custom.pop('impact')
70			if 'affected' in custom:
71			affected = custom.pop('affected')
72			if 'insight' in custom:
73			insight = custom.pop('insight')
74			if 'solution' in custom:
75			solution = custom.pop('solution')
76			if 'solution_type' in custom:
77			solution_t = custom.pop('solution_type')
78			if 'solution_method' in custom:
79			solution_m = custom.pop('solution_method')
80
81			if 'vuldetect' in custom:
82			vuldetect = custom.pop('vuldetect')
83			if 'qod_type' in custom:
84			qod_t = custom.pop('qod_type')
85			elif 'qod' in custom:
86			qod_v = custom.pop('qod')
87
88			severity = dict()
89			if 'severity_vector' in custom:
90			severity_vector = custom.pop('severity_vector')
91			else:
92			severity_vector = custom.pop('cvss_base_vector')
93			severity['severity_base_vector'] = severity_vector
94
95			if "CVSS:3" in severity_vector:
96			severity_type = 'cvss_base_v3'
97			else:
98			severity_type = 'cvss_base_v2'
99			severity['severity_type'] = severity_type
100
101			if 'severity_date' in custom:
102			severity['severity_date'] = custom.pop('severity_date')
103			else:
104			severity['severity_date'] = vt_creation_time
105
106			if 'severity_origin' in custom:
107			severity['severity_origin'] = custom.pop('severity_origin')
108
109			if name is None:
110			name = ''
111
112			vt = {'name': name}
113			if custom is not None:
114			vt["custom"] = custom
115			if vt_params is not None:
116			vt["vt_params"] = vt_params
117			if vt_refs is not None:
118			vt["vt_refs"] = vt_refs
119			if vt_dependencies is not None:
120			vt["vt_dependencies"] = vt_dependencies
121			if vt_creation_time is not None:
122			vt["creation_time"] = vt_creation_time
123			if vt_modification_time is not None:
124			vt["modification_time"] = vt_modification_time
125			if summary is not None:
126			vt["summary"] = summary
127			if impact is not None:
128			vt["impact"] = impact
129			if affected is not None:
130			vt["affected"] = affected
131			if insight is not None:
132			vt["insight"] = insight
133
134			if solution is not None:
135			vt["solution"] = solution
136			if solution_t is not None:
137			vt["solution_type"] = solution_t
138			if solution_m is not None:
139			vt["solution_method"] = solution_m
140
141			if vuldetect is not None:
142			vt["detection"] = vuldetect
143
144			if qod_t is not None:
145			vt["qod_type"] = qod_t
146			elif qod_v is not None:
147			vt["qod"] = qod_v
148
149			if severity is not None:
150			vt["severities"] = severity
151
152			return vt
153
154			def get_notus_driver_oids(self) -> List[str]:
155			"""Return a list of oid corresponding to notus driver which
156			are considered backend entities and must not be exposed to
157			the OSP client."""
158			notus = NotusMetadataHandler()
159			lsc_families_and_drivers = notus.get_family_driver_linkers()
160
161			return lsc_families_and_drivers.values()
162
163			def get_vt_iterator(
164			self, vt_selection: List[str] = None, details: bool = True
165			) -> Iterator[Tuple[str, Dict]]:
166			""" Yield the vts from the Redis NVTicache. """
167
168			oids = None
169			if not vt_selection or details:
170			vt_collection = dict(self.nvti.get_oids())
171
172			if not vt_selection:
173			vt_selection = vt_collection.values()
174
175			if details:
176			oids = vt_collection
177
178			# Notus driver oid list which are not sent.
179			drivers = self.get_notus_driver_oids()
180			for vt_id in vt_selection:
181			if vt_id in drivers:
182			continue
183			vt = self.get_single_vt(vt_id, oids)
184			yield (vt_id, vt)
185
186			def calculate_vts_collection_hash(self) -> str:
187			""" Calculate the vts collection sha256 hash. """
188			m = sha256() # pylint: disable=invalid-name
189
190			# Notus driver oid list which are not sent.
191			drivers = self.get_notus_driver_oids()
192
193			# for a reproducible hash calculation
194			# the vts must already be sorted in the dictionary.
195			for vt_id, vt in self.get_vt_iterator(details=False):
196			if vt_id in drivers:
197			continue
198			param_chain = ""
199			vt_params = vt.get('vt_params')
200			if vt_params:
201			for _, param in sorted(vt_params.items()):
202			param_chain += (
203			param.get('id')
204			+ param.get('name')
205			+ param.get('default')
206			)
207
208			m.update(
209			(vt_id + vt.get('modification_time')).encode('utf-8')
210			+ param_chain.encode('utf-8')
211			)
212
213			return m.hexdigest()
214

greenbone / ospd-openvas

Pull Request — master (#367)

VtHelper.get_notus_driver_oids() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like