tests.notus.test_notus.NotusTestCase.test_get_family_driver_linkers() - Code Metrics - Inspection of "Launch NVTs or notus driver, depending on availabi..." - greenbone/ospd-openvas - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#366)

by Juan José

created 2021-01-04 07:50 UTC

NotusTestCase.test_get_family_driver_linkers() A

↳ Parent: tests.notus.test_notus

Complexity

Conditions

Size

Total Lines	16
Code Lines	12

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	12
nop	2
dl	0
loc	16
rs	9.8
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/>.

import logging
import unittest

from csv import DictReader
from pathlib import Path, PurePath
from collections import OrderedDict
from unittest.mock import patch, MagicMock

from ospd_openvas.notus.metadata import (
    NotusMetadataHandler,
    EXPECTED_FIELD_NAMES_LIST,
    METADATA_DIRECTORY_NAME,
)
from ospd_openvas.errors import OspdOpenvasError


class NotusTestCase(unittest.TestCase):
    @patch('ospd_openvas.nvticache.NVTICache')
    def setUp(self, MockNvti):
        self.nvti = MockNvti()

    @patch('ospd_openvas.notus.metadata.Openvas')
    def test_set_openvas_settings(self, MockOpenvas):
        openvas = MockOpenvas()
        openvas.get_settings.return_value = {'nasl_no_signature_check': 0}

        notus = NotusMetadataHandler()
        no_signature_check = notus.openvas_setting.get(
            "nasl_no_signature_check"
        )

        self.assertEqual(no_signature_check, 0)

    @patch('ospd_openvas.notus.metadata.Openvas')
    def test_metadata_path(self, MockOpenvas):
        openvas = MockOpenvas()
        openvas.get_settings.return_value = {'plugins_folder': './tests/notus'}
        notus = NotusMetadataHandler()

        self.assertIsNone(notus._metadata_path)
        self.assertEqual(
            notus.metadata_path, f'./tests/notus/{METADATA_DIRECTORY_NAME}/'
        )
        self.assertEqual(
            notus._metadata_path, f'./tests/notus/{METADATA_DIRECTORY_NAME}/'
        )

    def test_is_checksum_correct_check_disable(self):
        notus = NotusMetadataHandler()
        notus._openvas_settings_dict = {'nasl_no_signature_check': 1}

        self.assertTrue(notus.is_checksum_correct(Path("foo")))

    def test_is_checksum_correct_enabled_false(self):
        notus = NotusMetadataHandler(nvti=self.nvti)
        notus.nvti.get_file_checksum.return_value = "abc123"
        notus._openvas_settings_dict = {'nasl_no_signature_check': 0}

        self.assertFalse(
            notus.is_checksum_correct(Path("./tests/notus/example.csv"))
        )

    def test_is_checksum_correct_enabled_true(self):
        notus = NotusMetadataHandler(nvti=self.nvti)
        notus.nvti.get_file_checksum.return_value = (
            "76b112033d3badbe1331454fb08652bda08e6d6c1f3bca5b4b533ab00d55388e"
        )
        notus._openvas_settings_dict = {'nasl_no_signature_check': 0}

        self.assertTrue(
            notus.is_checksum_correct(Path("./tests/notus/example.csv"))
        )

    def test_check_advisory_dict(self):
        advisory_dict = OrderedDict(
            [
                ('OID', '1.3.6.1.4.1.25623.1.1.2.2020.1234'),
                (
                    'TITLE',
                    'VendorOS: Security Advisory for git (VendorOS-2020-1234)',
                ),
                ('CREATION_DATE', '1600269468'),
                ('LAST_MODIFICATION', '1601380531'),
                ('SOURCE_PKGS', "['git']"),
                ('ADVISORY_ID', 'VendorOS-2020-1234'),
                ('CVSS_BASE_VECTOR', 'AV:N/AC:L/Au:N/C:C/I:C/A:C'),
                ('CVSS_BASE', '10.0'),
                ('ADVISORY_XREF', 'https://example.com'),
                ('DESCRIPTION', 'The remote host is missing an update.'),
                ('INSIGHT', 'buffer overflow'),
                ('AFFECTED', "'p1' package(s) on VendorOS V2.0SP1"),
                ('CVE_LIST', "['CVE-2020-1234']"),
                (
                    'BINARY_PACKAGES_FOR_RELEASES',
                    "{'VendorOS V2.0SP1': ['p1-1.1']}",
                ),
                ('XREFS', '[]'),
            ]
        )

        notus = NotusMetadataHandler()
        self.assertTrue(notus._check_advisory_dict(advisory_dict))

    def test_check_advisory_dict_no_value(self):
        advisory_dict = OrderedDict(
            [
                ('OID', '1.3.6.1.4.1.25623.1.1.2.2020.1234'),
                (
                    'TITLE',
                    'VendorOS: Security Advisory for git (VendorOS-2020-1234)',
                ),
                ('CREATION_DATE', None),
                ('LAST_MODIFICATION', '1601380531'),
                ('SOURCE_PKGS', "['git']"),
                ('ADVISORY_ID', 'VendorOS-2020-1234'),
                ('CVSS_BASE_VECTOR', 'AV:N/AC:L/Au:N/C:C/I:C/A:C'),
                ('CVSS_BASE', '10.0'),
                ('ADVISORY_XREF', 'https://example.com'),
                ('DESCRIPTION', 'The remote host is missing an update.'),
                ('INSIGHT', 'buffer overflow'),
                ('AFFECTED', "'p1' package(s) on VendorOS V2.0SP1"),
                ('CVE_LIST', "['CVE-2020-1234']"),
                (
                    'BINARY_PACKAGES_FOR_RELEASES',
                    "{'VendorOS V2.0SP1': ['p1-1.1']}",
                ),
                ('XREFS', '[]'),
            ]
        )

        notus = NotusMetadataHandler()
        self.assertFalse(notus._check_advisory_dict(advisory_dict))

    def test_check_advisory_dict_no_package(self):
        advisory_dict = OrderedDict(
            [
                ('OID', '1.3.6.1.4.1.25623.1.1.2.2020.1234'),
                (
                    'TITLE',
                    'VendorOS: Security Advisory for git (VendorOS-2020-1234)',
                ),
                ('CREATION_DATE', '1600269468'),
                ('LAST_MODIFICATION', '1601380531'),
                ('SOURCE_PKGS', "[]"),
                ('ADVISORY_ID', 'VendorOS-2020-1234'),
                ('CVSS_BASE_VECTOR', 'AV:N/AC:L/Au:N/C:C/I:C/A:C'),
                ('CVSS_BASE', '10.0'),
                ('ADVISORY_XREF', 'https://example.com'),
                ('DESCRIPTION', 'The remote host is missing an update.'),
                ('INSIGHT', 'buffer overflow'),
                ('AFFECTED', "'p1' package(s) on VendorOS V2.0SP1"),
                ('CVE_LIST', "['CVE-2020-1234']"),
                (
                    'BINARY_PACKAGES_FOR_RELEASES',
                    "{'VendorOS V2.0SP1': ['p1-1.1']}",
                ),
                ('XREFS', '[]'),
            ]
        )

        notus = NotusMetadataHandler()
        self.assertFalse(notus._check_advisory_dict(advisory_dict))

    def test_check_advisory_dict_valerr(self):
        advisory_dict = OrderedDict(
            [
                ('OID', '1.3.6.1.4.1.25623.1.1.2.2020.1234'),
                (
                    'TITLE',
                    'VendorOS: Security Advisory for git (VendorOS-2020-1234)',
                ),
                ('CREATION_DATE', '1600269468'),
                ('LAST_MODIFICATION', '1601380531'),
                ('SOURCE_PKGS', "a"),
                ('ADVISORY_ID', 'VendorOS-2020-1234'),
                ('CVSS_BASE_VECTOR', 'AV:N/AC:L/Au:N/C:C/I:C/A:C'),
                ('CVSS_BASE', '10.0'),
                ('ADVISORY_XREF', 'https://example.com'),
                ('DESCRIPTION', 'The remote host is missing an update.'),
                ('INSIGHT', 'buffer overflow'),
                ('AFFECTED', "'p1' package(s) on VendorOS V2.0SP1"),
                ('CVE_LIST', "['CVE-2020-1234']"),
                (
                    'BINARY_PACKAGES_FOR_RELEASES',
                    "{'VendorOS V2.0SP1': ['p1-1.1']}",
                ),
                ('XREFS', '[]'),
            ]
        )

        notus = NotusMetadataHandler()
        self.assertFalse(notus._check_advisory_dict(advisory_dict))

    def test_format_xrefs(self):
        notus = NotusMetadataHandler()
        ret = notus._format_xrefs(
            "https://example.com", ["www.foo.net", "www.bar.net"]
        )

        self.assertEqual(
            ret, "URL:https://example.com, URL:www.foo.net, URL:www.bar.net"
        )

    def test_check_field_names_lsc(self):
        notus = NotusMetadataHandler()
        field_names_list = [
            "OID",
            "TITLE",
            "CREATION_DATE",
            "LAST_MODIFICATION",
            "SOURCE_PKGS",
            "ADVISORY_ID",
            "SEVERITY_ORIGIN",
            "SEVERITY_DATE",
            "SEVERITY_VECTOR",
            "ADVISORY_XREF",
            "DESCRIPTION",
            "INSIGHT",
            "AFFECTED",
            "CVE_LIST",
            "BINARY_PACKAGES_FOR_RELEASES",
            "XREFS",
            "FILENAME",
        ]
        self.assertTrue(notus._check_field_names_lsc(field_names_list))

    def test_check_field_names_lsc_unordered(self):
        notus = NotusMetadataHandler()
        field_names_list = [
            "TITLE",
            "OID",
            "CREATION_DATE",
            "LAST_MODIFICATION",
            "SOURCE_PKGS",
            "ADVISORY_ID",
            "SEVERITY_ORIGIN",
            "SEVERITY_DATE",
            "SEVERITY_VECTOR",
            "ADVISORY_XREF",
            "DESCRIPTION",
            "INSIGHT",
            "AFFECTED",
            "CVE_LIST",
            "BINARY_PACKAGES_FOR_RELEASES",
            "XREFS",
        ]

        self.assertFalse(notus._check_field_names_lsc(field_names_list))

    def test_check_field_names_lsc_missing(self):
        notus = NotusMetadataHandler()
        field_names_list = [
            "OID",
            "CREATION_DATE",
            "LAST_MODIFICATION",
            "SOURCE_PKGS",
            "ADVISORY_ID",
            "CVSS_BASE_VECTOR",
            "CVSS_BASE",
            "ADVISORY_XREF",
            "DESCRIPTION",
            "INSIGHT",
            "AFFECTED",
            "CVE_LIST",
            "BINARY_PACKAGES_FOR_RELEASES",
            "XREFS",
        ]

        self.assertFalse(notus._check_field_names_lsc(field_names_list))

    def test_get_csv_filepath(self):
        path = Path("./tests/notus/example.csv").resolve()

        notus = NotusMetadataHandler(metadata_path="./tests/notus/")
        ret = notus._get_csv_filepaths()

        self.assertEqual(ret, [path])

    @patch('ospd_openvas.notus.metadata.Openvas')
    def test_update_metadata_warning(self, MockOpenvas):
        notus = NotusMetadataHandler()
        logging.Logger.warning = MagicMock()
        path = Path("./tests/notus/example.csv").resolve()
        openvas = MockOpenvas()
        openvas.get_settings.return_value = {'table_driven_lsc': 1}

        notus._get_csv_filepaths = MagicMock(return_value=[path])
        notus.is_checksum_correct = MagicMock(return_value=False)

        notus.update_metadata()
        logging.Logger.warning.assert_called_with(
            f'Checksum for %s failed', path
        )

    @patch('ospd_openvas.notus.metadata.Openvas')

    def test_update_metadata_field_name_failed(self, MockOpenvas):
        notus = NotusMetadataHandler(metadata_path="./tests/notus")
        logging.Logger.warning = MagicMock()
        path = Path("./tests/notus/example.csv").resolve()
        openvas = MockOpenvas()
        openvas.get_settings.return_value = {'table_driven_lsc': 1}

        notus._get_csv_filepaths = MagicMock(return_value=[path])
        notus.is_checksum_correct = MagicMock(return_value=True)
        notus._check_field_names_lsc = MagicMock(return_value=False)

        notus.update_metadata()

        logging.Logger.warning.assert_called_with(
            f'Field names check for %s failed', path
        )

    @patch('ospd_openvas.notus.metadata.Openvas')

    def test_update_metadata_failed(self, MockOpenvas):
        notus = NotusMetadataHandler(metadata_path="./tests/notus")
        logging.Logger.warning = MagicMock()
        path = Path("./tests/notus/example.csv").resolve()
        openvas = MockOpenvas()
        openvas.get_settings.return_value = {'table_driven_lsc': 1}

        notus._get_csv_filepaths = MagicMock(return_value=[path])
        notus.is_checksum_correct = MagicMock(return_value=True)
        notus._check_field_names_lsc = MagicMock(return_value=True)
        notus.upload_lsc_from_csv_reader = MagicMock(return_value=False)

        notus.update_metadata()

        logging.Logger.warning.assert_called_with(
            "Some advaisory was not loaded from %s", path.name
        )

    def test_update_metadata_disabled(self):
        notus = NotusMetadataHandler(metadata_path="./tests/notus")
        ret = notus.update_metadata()
        self.assertIsNone(ret)

    @patch('ospd_openvas.notus.metadata.Openvas')
    def test_update_metadata_success(self, MockOpenvas):
        notus = NotusMetadataHandler(metadata_path="./tests/notus")
        logging.Logger.warning = MagicMock()
        path = Path("./tests/notus/example.csv").resolve()
        purepath = PurePath(path).name
        openvas = MockOpenvas()
        openvas.get_settings.return_value = {'table_driven_lsc': 1}

        notus._get_csv_filepaths = MagicMock(return_value=[path])
        notus.is_checksum_correct = MagicMock(return_value=True)
        notus._check_field_names_lsc = MagicMock(return_value=True)
        notus.upload_lsc_from_csv_reader = MagicMock(return_value=True)

        notus.update_metadata()

        logging.Logger.warning.assert_not_called()

    def test_upload_lsc_from_csv_reader_failed(self):

        general_metadata_dict = {
            'VULDETECT': 'Checks if a vulnerable package version is present on the target host.',
            'SOLUTION': 'Please install the updated package(s).',
            'SOLUTION_TYPE': 'VendorFix',
            'QOD_TYPE': 'package',
        }

        notus = NotusMetadataHandler(
            nvti=self.nvti, metadata_path="./tests/notus"
        )
        notus.nvti.add_vt_to_cache.side_effect = OspdOpenvasError
        logging.Logger.debug = MagicMock()
        path = Path("./tests/notus/example.csv").resolve()
        purepath = PurePath(path).name
        family = "Some family"
        with path.open("r") as openfile:
            for line_string in openfile:
                if line_string.startswith("{"):
                    break
            reader = DictReader(openfile)

            ret = notus.upload_lsc_from_csv_reader(
                purepath, family, general_metadata_dict, reader
            )

        self.assertFalse(ret)
        logging.Logger.debug.assert_called_with(
            "Loaded %d/%d advisories from %s", 0, 1, purepath
        )

    def test_upload_lsc_from_csv_reader_sucess(self):

        general_metadata_dict = {
            'VULDETECT': 'Checks if a vulnerable package version is present on the target host.',
            'SOLUTION': 'Please install the updated package(s).',
            'SOLUTION_TYPE': 'VendorFix',
            'QOD_TYPE': 'package',
        }

        notus = NotusMetadataHandler(
            nvti=self.nvti, metadata_path="./tests/notus"
        )
        notus.nvti.add_vt_to_cache.return_value = None
        logging.Logger.debug = MagicMock()
        path = Path("./tests/notus/example.csv").resolve()
        purepath = PurePath(path).name
        family = "Some family"
        with path.open("r") as openfile:
            for line_string in openfile:
                if line_string.startswith("{"):
                    break
            reader = DictReader(openfile)

            ret = notus.upload_lsc_from_csv_reader(
                purepath, family, general_metadata_dict, reader
            )

        self.assertTrue(ret)
        logging.Logger.debug.assert_called_with(
            "Loaded %d/%d advisories from %s", 1, 1, purepath
        )

    @patch('ospd_openvas.notus.metadata.Openvas')
    def test_get_family_driver_linkers(self, MockOpenvas):
        notus = NotusMetadataHandler(metadata_path="./tests/notus")

        path = Path("./tests/notus/example.csv").resolve()
        notus._get_csv_filepaths = MagicMock(return_value=[path])
        notus.is_checksum_correct = MagicMock(return_value=True)

        openvas = MockOpenvas()
        openvas.get_settings.return_value = {'table_driven_lsc': 1}

        family_and_driver = {
            'Local Security Checks': '1.3.6.1.4.1.25623.1.2.3.4'
        }
        ret = notus.get_family_driver_linkers()
        self.assertEqual(ret, family_and_driver)


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		import logging
20		import unittest
21
22		from csv import DictReader
23		from pathlib import Path, PurePath
24		from collections import OrderedDict
25		from unittest.mock import patch, MagicMock
26
27		from ospd_openvas.notus.metadata import (
28		NotusMetadataHandler,
29		EXPECTED_FIELD_NAMES_LIST,
30		METADATA_DIRECTORY_NAME,
31		)
32		from ospd_openvas.errors import OspdOpenvasError
33
34
35		class NotusTestCase(unittest.TestCase):
36		@patch('ospd_openvas.nvticache.NVTICache')
37		def setUp(self, MockNvti):
38		self.nvti = MockNvti()
39
40		@patch('ospd_openvas.notus.metadata.Openvas')
41		def test_set_openvas_settings(self, MockOpenvas):
42		openvas = MockOpenvas()
43		openvas.get_settings.return_value = {'nasl_no_signature_check': 0}
44
45		notus = NotusMetadataHandler()
46		no_signature_check = notus.openvas_setting.get(
47		"nasl_no_signature_check"
48		)
49
50		self.assertEqual(no_signature_check, 0)
51
52		@patch('ospd_openvas.notus.metadata.Openvas')
53		def test_metadata_path(self, MockOpenvas):
54		openvas = MockOpenvas()
55		openvas.get_settings.return_value = {'plugins_folder': './tests/notus'}
56		notus = NotusMetadataHandler()
57
58		self.assertIsNone(notus._metadata_path)
59		self.assertEqual(
60		notus.metadata_path, f'./tests/notus/{METADATA_DIRECTORY_NAME}/'
61		)
62		self.assertEqual(
63		notus._metadata_path, f'./tests/notus/{METADATA_DIRECTORY_NAME}/'
64		)
65
66		def test_is_checksum_correct_check_disable(self):
67		notus = NotusMetadataHandler()
68		notus._openvas_settings_dict = {'nasl_no_signature_check': 1}
69
70		self.assertTrue(notus.is_checksum_correct(Path("foo")))
71
72		def test_is_checksum_correct_enabled_false(self):
73		notus = NotusMetadataHandler(nvti=self.nvti)
74		notus.nvti.get_file_checksum.return_value = "abc123"
75		notus._openvas_settings_dict = {'nasl_no_signature_check': 0}
76
77		self.assertFalse(
78		notus.is_checksum_correct(Path("./tests/notus/example.csv"))
79		)
80
81		def test_is_checksum_correct_enabled_true(self):
82		notus = NotusMetadataHandler(nvti=self.nvti)
83		notus.nvti.get_file_checksum.return_value = (
84		"76b112033d3badbe1331454fb08652bda08e6d6c1f3bca5b4b533ab00d55388e"
85		)
86		notus._openvas_settings_dict = {'nasl_no_signature_check': 0}
87
88		self.assertTrue(
89		notus.is_checksum_correct(Path("./tests/notus/example.csv"))
90		)
91
92		def test_check_advisory_dict(self):
93		advisory_dict = OrderedDict(
94		[
95		('OID', '1.3.6.1.4.1.25623.1.1.2.2020.1234'),
96		(
97		'TITLE',
98		'VendorOS: Security Advisory for git (VendorOS-2020-1234)',
99		),
100		('CREATION_DATE', '1600269468'),
101		('LAST_MODIFICATION', '1601380531'),
102		('SOURCE_PKGS', "['git']"),
103		('ADVISORY_ID', 'VendorOS-2020-1234'),
104		('CVSS_BASE_VECTOR', 'AV:N/AC:L/Au:N/C:C/I:C/A:C'),
105		('CVSS_BASE', '10.0'),
106		('ADVISORY_XREF', 'https://example.com'),
107		('DESCRIPTION', 'The remote host is missing an update.'),
108		('INSIGHT', 'buffer overflow'),
109		('AFFECTED', "'p1' package(s) on VendorOS V2.0SP1"),
110		('CVE_LIST', "['CVE-2020-1234']"),
111		(
112		'BINARY_PACKAGES_FOR_RELEASES',
113		"{'VendorOS V2.0SP1': ['p1-1.1']}",
114		),
115		('XREFS', '[]'),
116		]
117		)
118
119		notus = NotusMetadataHandler()
120		self.assertTrue(notus._check_advisory_dict(advisory_dict))
121
122		def test_check_advisory_dict_no_value(self):
123		advisory_dict = OrderedDict(
124		[
125		('OID', '1.3.6.1.4.1.25623.1.1.2.2020.1234'),
126		(
127		'TITLE',
128		'VendorOS: Security Advisory for git (VendorOS-2020-1234)',
129		),
130		('CREATION_DATE', None),
131		('LAST_MODIFICATION', '1601380531'),
132		('SOURCE_PKGS', "['git']"),
133		('ADVISORY_ID', 'VendorOS-2020-1234'),
134		('CVSS_BASE_VECTOR', 'AV:N/AC:L/Au:N/C:C/I:C/A:C'),
135		('CVSS_BASE', '10.0'),
136		('ADVISORY_XREF', 'https://example.com'),
137		('DESCRIPTION', 'The remote host is missing an update.'),
138		('INSIGHT', 'buffer overflow'),
139		('AFFECTED', "'p1' package(s) on VendorOS V2.0SP1"),
140		('CVE_LIST', "['CVE-2020-1234']"),
141		(
142		'BINARY_PACKAGES_FOR_RELEASES',
143		"{'VendorOS V2.0SP1': ['p1-1.1']}",
144		),
145		('XREFS', '[]'),
146		]
147		)
148
149		notus = NotusMetadataHandler()
150		self.assertFalse(notus._check_advisory_dict(advisory_dict))
151
152		def test_check_advisory_dict_no_package(self):
153		advisory_dict = OrderedDict(
154		[
155		('OID', '1.3.6.1.4.1.25623.1.1.2.2020.1234'),
156		(
157		'TITLE',
158		'VendorOS: Security Advisory for git (VendorOS-2020-1234)',
159		),
160		('CREATION_DATE', '1600269468'),
161		('LAST_MODIFICATION', '1601380531'),
162		('SOURCE_PKGS', "[]"),
163		('ADVISORY_ID', 'VendorOS-2020-1234'),
164		('CVSS_BASE_VECTOR', 'AV:N/AC:L/Au:N/C:C/I:C/A:C'),
165		('CVSS_BASE', '10.0'),
166		('ADVISORY_XREF', 'https://example.com'),
167		('DESCRIPTION', 'The remote host is missing an update.'),
168		('INSIGHT', 'buffer overflow'),
169		('AFFECTED', "'p1' package(s) on VendorOS V2.0SP1"),
170		('CVE_LIST', "['CVE-2020-1234']"),
171		(
172		'BINARY_PACKAGES_FOR_RELEASES',
173		"{'VendorOS V2.0SP1': ['p1-1.1']}",
174		),
175		('XREFS', '[]'),
176		]
177		)
178
179		notus = NotusMetadataHandler()
180		self.assertFalse(notus._check_advisory_dict(advisory_dict))
181
182		def test_check_advisory_dict_valerr(self):
183		advisory_dict = OrderedDict(
184		[
185		('OID', '1.3.6.1.4.1.25623.1.1.2.2020.1234'),
186		(
187		'TITLE',
188		'VendorOS: Security Advisory for git (VendorOS-2020-1234)',
189		),
190		('CREATION_DATE', '1600269468'),
191		('LAST_MODIFICATION', '1601380531'),
192		('SOURCE_PKGS', "a"),
193		('ADVISORY_ID', 'VendorOS-2020-1234'),
194		('CVSS_BASE_VECTOR', 'AV:N/AC:L/Au:N/C:C/I:C/A:C'),
195		('CVSS_BASE', '10.0'),
196		('ADVISORY_XREF', 'https://example.com'),
197		('DESCRIPTION', 'The remote host is missing an update.'),
198		('INSIGHT', 'buffer overflow'),
199		('AFFECTED', "'p1' package(s) on VendorOS V2.0SP1"),
200		('CVE_LIST', "['CVE-2020-1234']"),
201		(
202		'BINARY_PACKAGES_FOR_RELEASES',
203		"{'VendorOS V2.0SP1': ['p1-1.1']}",
204		),
205		('XREFS', '[]'),
206		]
207		)
208
209		notus = NotusMetadataHandler()
210		self.assertFalse(notus._check_advisory_dict(advisory_dict))
211
212		def test_format_xrefs(self):
213		notus = NotusMetadataHandler()
214		ret = notus._format_xrefs(
215		"https://example.com", ["www.foo.net", "www.bar.net"]
216		)
217
218		self.assertEqual(
219		ret, "URL:https://example.com, URL:www.foo.net, URL:www.bar.net"
220		)
221
222		def test_check_field_names_lsc(self):
223		notus = NotusMetadataHandler()
224		field_names_list = [
225		"OID",
226		"TITLE",
227		"CREATION_DATE",
228		"LAST_MODIFICATION",
229		"SOURCE_PKGS",
230		"ADVISORY_ID",
231		"SEVERITY_ORIGIN",
232		"SEVERITY_DATE",
233		"SEVERITY_VECTOR",
234		"ADVISORY_XREF",
235		"DESCRIPTION",
236		"INSIGHT",
237		"AFFECTED",
238		"CVE_LIST",
239		"BINARY_PACKAGES_FOR_RELEASES",
240		"XREFS",
241		"FILENAME",
242		]
243		self.assertTrue(notus._check_field_names_lsc(field_names_list))
244
245		def test_check_field_names_lsc_unordered(self):
246		notus = NotusMetadataHandler()
247		field_names_list = [
248		"TITLE",
249		"OID",
250		"CREATION_DATE",
251		"LAST_MODIFICATION",
252		"SOURCE_PKGS",
253		"ADVISORY_ID",
254		"SEVERITY_ORIGIN",
255		"SEVERITY_DATE",
256		"SEVERITY_VECTOR",
257		"ADVISORY_XREF",
258		"DESCRIPTION",
259		"INSIGHT",
260		"AFFECTED",
261		"CVE_LIST",
262		"BINARY_PACKAGES_FOR_RELEASES",
263		"XREFS",
264		]
265
266		self.assertFalse(notus._check_field_names_lsc(field_names_list))
267
268		def test_check_field_names_lsc_missing(self):
269		notus = NotusMetadataHandler()
270		field_names_list = [
271		"OID",
272		"CREATION_DATE",
273		"LAST_MODIFICATION",
274		"SOURCE_PKGS",
275		"ADVISORY_ID",
276		"CVSS_BASE_VECTOR",
277		"CVSS_BASE",
278		"ADVISORY_XREF",
279		"DESCRIPTION",
280		"INSIGHT",
281		"AFFECTED",
282		"CVE_LIST",
283		"BINARY_PACKAGES_FOR_RELEASES",
284		"XREFS",
285		]
286
287		self.assertFalse(notus._check_field_names_lsc(field_names_list))
288
289		def test_get_csv_filepath(self):
290		path = Path("./tests/notus/example.csv").resolve()
291
292		notus = NotusMetadataHandler(metadata_path="./tests/notus/")
293		ret = notus._get_csv_filepaths()
294
295		self.assertEqual(ret, [path])
296
297		@patch('ospd_openvas.notus.metadata.Openvas')
298		def test_update_metadata_warning(self, MockOpenvas):
299		notus = NotusMetadataHandler()
300		logging.Logger.warning = MagicMock()
301		path = Path("./tests/notus/example.csv").resolve()
302		openvas = MockOpenvas()
303		openvas.get_settings.return_value = {'table_driven_lsc': 1}
304
305		notus._get_csv_filepaths = MagicMock(return_value=[path])
306		notus.is_checksum_correct = MagicMock(return_value=False)
307
308		notus.update_metadata()
309		logging.Logger.warning.assert_called_with(
310		f'Checksum for %s failed', path
311		)
312
313	View Code Duplication	@patch('ospd_openvas.notus.metadata.Openvas')
		0 ignored issues – show Duplication introduced 2020-12-17 11:57 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
314		def test_update_metadata_field_name_failed(self, MockOpenvas):
315		notus = NotusMetadataHandler(metadata_path="./tests/notus")
316		logging.Logger.warning = MagicMock()
317		path = Path("./tests/notus/example.csv").resolve()
318		openvas = MockOpenvas()
319		openvas.get_settings.return_value = {'table_driven_lsc': 1}
320
321		notus._get_csv_filepaths = MagicMock(return_value=[path])
322		notus.is_checksum_correct = MagicMock(return_value=True)
323		notus._check_field_names_lsc = MagicMock(return_value=False)
324
325		notus.update_metadata()
326
327		logging.Logger.warning.assert_called_with(
328		f'Field names check for %s failed', path
329		)
330
331	View Code Duplication	@patch('ospd_openvas.notus.metadata.Openvas')
		0 ignored issues – show Duplication introduced 2020-12-17 11:57 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
332		def test_update_metadata_failed(self, MockOpenvas):
333		notus = NotusMetadataHandler(metadata_path="./tests/notus")
334		logging.Logger.warning = MagicMock()
335		path = Path("./tests/notus/example.csv").resolve()
336		openvas = MockOpenvas()
337		openvas.get_settings.return_value = {'table_driven_lsc': 1}
338
339		notus._get_csv_filepaths = MagicMock(return_value=[path])
340		notus.is_checksum_correct = MagicMock(return_value=True)
341		notus._check_field_names_lsc = MagicMock(return_value=True)
342		notus.upload_lsc_from_csv_reader = MagicMock(return_value=False)
343
344		notus.update_metadata()
345
346		logging.Logger.warning.assert_called_with(
347		"Some advaisory was not loaded from %s", path.name
348		)
349
350		def test_update_metadata_disabled(self):
351		notus = NotusMetadataHandler(metadata_path="./tests/notus")
352		ret = notus.update_metadata()
353		self.assertIsNone(ret)
354
355		@patch('ospd_openvas.notus.metadata.Openvas')
356		def test_update_metadata_success(self, MockOpenvas):
357		notus = NotusMetadataHandler(metadata_path="./tests/notus")
358		logging.Logger.warning = MagicMock()
359		path = Path("./tests/notus/example.csv").resolve()
360		purepath = PurePath(path).name
361		openvas = MockOpenvas()
362		openvas.get_settings.return_value = {'table_driven_lsc': 1}
363
364		notus._get_csv_filepaths = MagicMock(return_value=[path])
365		notus.is_checksum_correct = MagicMock(return_value=True)
366		notus._check_field_names_lsc = MagicMock(return_value=True)
367		notus.upload_lsc_from_csv_reader = MagicMock(return_value=True)
368
369		notus.update_metadata()
370
371		logging.Logger.warning.assert_not_called()
372
373	View Code Duplication	def test_upload_lsc_from_csv_reader_failed(self):
		0 ignored issues – show Duplication introduced 2020-11-03 15:17 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
374		general_metadata_dict = {
375		'VULDETECT': 'Checks if a vulnerable package version is present on the target host.',
376		'SOLUTION': 'Please install the updated package(s).',
377		'SOLUTION_TYPE': 'VendorFix',
378		'QOD_TYPE': 'package',
379		}
380
381		notus = NotusMetadataHandler(
382		nvti=self.nvti, metadata_path="./tests/notus"
383		)
384		notus.nvti.add_vt_to_cache.side_effect = OspdOpenvasError
385		logging.Logger.debug = MagicMock()
386		path = Path("./tests/notus/example.csv").resolve()
387		purepath = PurePath(path).name
388		family = "Some family"
389		with path.open("r") as openfile:
390		for line_string in openfile:
391		if line_string.startswith("{"):
392		break
393		reader = DictReader(openfile)
394
395		ret = notus.upload_lsc_from_csv_reader(
396		purepath, family, general_metadata_dict, reader
397		)
398
399		self.assertFalse(ret)
400		logging.Logger.debug.assert_called_with(
401		"Loaded %d/%d advisories from %s", 0, 1, purepath
402		)
403
404	View Code Duplication	def test_upload_lsc_from_csv_reader_sucess(self):
		0 ignored issues – show Duplication introduced 2020-11-03 15:17 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
405		general_metadata_dict = {
406		'VULDETECT': 'Checks if a vulnerable package version is present on the target host.',
407		'SOLUTION': 'Please install the updated package(s).',
408		'SOLUTION_TYPE': 'VendorFix',
409		'QOD_TYPE': 'package',
410		}
411
412		notus = NotusMetadataHandler(
413		nvti=self.nvti, metadata_path="./tests/notus"
414		)
415		notus.nvti.add_vt_to_cache.return_value = None
416		logging.Logger.debug = MagicMock()
417		path = Path("./tests/notus/example.csv").resolve()
418		purepath = PurePath(path).name
419		family = "Some family"
420		with path.open("r") as openfile:
421		for line_string in openfile:
422		if line_string.startswith("{"):
423		break
424		reader = DictReader(openfile)
425
426		ret = notus.upload_lsc_from_csv_reader(
427		purepath, family, general_metadata_dict, reader
428		)
429
430		self.assertTrue(ret)
431		logging.Logger.debug.assert_called_with(
432		"Loaded %d/%d advisories from %s", 1, 1, purepath
433		)
434
435		@patch('ospd_openvas.notus.metadata.Openvas')
436		def test_get_family_driver_linkers(self, MockOpenvas):
437		notus = NotusMetadataHandler(metadata_path="./tests/notus")
438
439		path = Path("./tests/notus/example.csv").resolve()
440		notus._get_csv_filepaths = MagicMock(return_value=[path])
441		notus.is_checksum_correct = MagicMock(return_value=True)
442
443		openvas = MockOpenvas()
444		openvas.get_settings.return_value = {'table_driven_lsc': 1}
445
446		family_and_driver = {
447		'Local Security Checks': '1.3.6.1.4.1.25623.1.2.3.4'
448		}
449		ret = notus.get_family_driver_linkers()
450		self.assertEqual(ret, family_and_driver)
451

greenbone / ospd-openvas

Pull Request — master (#366)

NotusTestCase.test_get_family_driver_linkers() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like