tests.notus.test_notus.LockFileTestCase.test_update_metadata_failed() - Code Metrics - Inspection of "Check if Notus is enabled." - greenbone/ospd-openvas - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#364)

by Juan José

created 2020-12-17 11:56 UTC

LockFileTestCase.test_update_metadata_failed() A

↳ Parent: tests.notus.test_notus

Complexity

Conditions

Size

Total Lines	14
Code Lines	11

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	11
nop	1
dl	0
loc	14
rs	9.85
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 = (
            "2f561b9be5d1a1194f49cd5a6a024dee15a0c0bc7d94287266d0e6358e737f4e"
        )
        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",
        ]
        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)

    def test_update_metadata_success(self):
        notus = NotusMetadataHandler(metadata_path="./tests/notus")
        logging.Logger.warning = MagicMock()
        path = Path("./tests/notus/example.csv").resolve()
        purepath = PurePath(path).name

        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
        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, 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
        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, general_metadata_dict, reader
            )

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


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		"2f561b9be5d1a1194f49cd5a6a024dee15a0c0bc7d94287266d0e6358e737f4e"
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		]
242		self.assertTrue(notus._check_field_names_lsc(field_names_list))
243
244		def test_check_field_names_lsc_unordered(self):
245		notus = NotusMetadataHandler()
246		field_names_list = [
247		"TITLE",
248		"OID",
249		"CREATION_DATE",
250		"LAST_MODIFICATION",
251		"SOURCE_PKGS",
252		"ADVISORY_ID",
253		"SEVERITY_ORIGIN",
254		"SEVERITY_DATE",
255		"SEVERITY_VECTOR",
256		"ADVISORY_XREF",
257		"DESCRIPTION",
258		"INSIGHT",
259		"AFFECTED",
260		"CVE_LIST",
261		"BINARY_PACKAGES_FOR_RELEASES",
262		"XREFS",
263		]
264
265		self.assertFalse(notus._check_field_names_lsc(field_names_list))
266
267		def test_check_field_names_lsc_missing(self):
268		notus = NotusMetadataHandler()
269		field_names_list = [
270		"OID",
271		"CREATION_DATE",
272		"LAST_MODIFICATION",
273		"SOURCE_PKGS",
274		"ADVISORY_ID",
275		"CVSS_BASE_VECTOR",
276		"CVSS_BASE",
277		"ADVISORY_XREF",
278		"DESCRIPTION",
279		"INSIGHT",
280		"AFFECTED",
281		"CVE_LIST",
282		"BINARY_PACKAGES_FOR_RELEASES",
283		"XREFS",
284		]
285
286		self.assertFalse(notus._check_field_names_lsc(field_names_list))
287
288		def test_get_csv_filepath(self):
289		path = Path("./tests/notus/example.csv").resolve()
290
291		notus = NotusMetadataHandler(metadata_path="./tests/notus/")
292		ret = notus._get_csv_filepaths()
293
294		self.assertEqual(ret, [path])
295
296		@patch('ospd_openvas.notus.metadata.Openvas')
297		def test_update_metadata_warning(self, MockOpenvas):
298		notus = NotusMetadataHandler()
299		logging.Logger.warning = MagicMock()
300		path = Path("./tests/notus/example.csv").resolve()
301		openvas = MockOpenvas()
302		openvas.get_settings.return_value = {'table_driven_lsc': 1}
303
304		notus._get_csv_filepaths = MagicMock(return_value=[path])
305		notus.is_checksum_correct = MagicMock(return_value=False)
306
307		notus.update_metadata()
308		logging.Logger.warning.assert_called_with(
309		f'Checksum for %s failed', path
310		)
311
312	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...
313		def test_update_metadata_field_name_failed(self, MockOpenvas):
314		notus = NotusMetadataHandler(metadata_path="./tests/notus")
315		logging.Logger.warning = MagicMock()
316		path = Path("./tests/notus/example.csv").resolve()
317		openvas = MockOpenvas()
318		openvas.get_settings.return_value = {'table_driven_lsc': 1}
319
320		notus._get_csv_filepaths = MagicMock(return_value=[path])
321		notus.is_checksum_correct = MagicMock(return_value=True)
322		notus._check_field_names_lsc = MagicMock(return_value=False)
323
324		notus.update_metadata()
325
326		logging.Logger.warning.assert_called_with(
327		f'Field names check for %s failed', path
328		)
329
330	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...
331		def test_update_metadata_failed(self, MockOpenvas):
332		notus = NotusMetadataHandler(metadata_path="./tests/notus")
333		logging.Logger.warning = MagicMock()
334		path = Path("./tests/notus/example.csv").resolve()
335		openvas = MockOpenvas()
336		openvas.get_settings.return_value = {'table_driven_lsc': 1}
337
338		notus._get_csv_filepaths = MagicMock(return_value=[path])
339		notus.is_checksum_correct = MagicMock(return_value=True)
340		notus._check_field_names_lsc = MagicMock(return_value=True)
341		notus.upload_lsc_from_csv_reader = MagicMock(return_value=False)
342
343		notus.update_metadata()
344
345		logging.Logger.warning.assert_called_with(
346		"Some advaisory was not loaded from %s", path.name
347		)
348
349		def test_update_metadata_disabled(self):
350		notus = NotusMetadataHandler(metadata_path="./tests/notus")
351		ret = notus.update_metadata()
352		self.assertIsNone(ret)
353
354		def test_update_metadata_success(self):
355		notus = NotusMetadataHandler(metadata_path="./tests/notus")
356		logging.Logger.warning = MagicMock()
357		path = Path("./tests/notus/example.csv").resolve()
358		purepath = PurePath(path).name
359
360		notus._get_csv_filepaths = MagicMock(return_value=[path])
361		notus.is_checksum_correct = MagicMock(return_value=True)
362		notus._check_field_names_lsc = MagicMock(return_value=True)
363		notus.upload_lsc_from_csv_reader = MagicMock(return_value=True)
364
365		notus.update_metadata()
366
367		logging.Logger.warning.assert_not_called()
368
369	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...
370		general_metadata_dict = {
371		'VULDETECT': 'Checks if a vulnerable package version is present on the target host.',
372		'SOLUTION': 'Please install the updated package(s).',
373		'SOLUTION_TYPE': 'VendorFix',
374		'QOD_TYPE': 'package',
375		}
376
377		notus = NotusMetadataHandler(
378		nvti=self.nvti, metadata_path="./tests/notus"
379		)
380		notus.nvti.add_vt_to_cache.side_effect = OspdOpenvasError
381		logging.Logger.debug = MagicMock()
382		path = Path("./tests/notus/example.csv").resolve()
383		purepath = PurePath(path).name
384		with path.open("r") as openfile:
385		for line_string in openfile:
386		if line_string.startswith("{"):
387		break
388		reader = DictReader(openfile)
389
390		ret = notus.upload_lsc_from_csv_reader(
391		purepath, general_metadata_dict, reader
392		)
393
394		self.assertFalse(ret)
395		logging.Logger.debug.assert_called_with(
396		"Loaded %d/%d advisories from %s", 0, 1, purepath
397		)
398
399	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...
400		general_metadata_dict = {
401		'VULDETECT': 'Checks if a vulnerable package version is present on the target host.',
402		'SOLUTION': 'Please install the updated package(s).',
403		'SOLUTION_TYPE': 'VendorFix',
404		'QOD_TYPE': 'package',
405		}
406
407		notus = NotusMetadataHandler(
408		nvti=self.nvti, metadata_path="./tests/notus"
409		)
410		notus.nvti.add_vt_to_cache.return_value = None
411		logging.Logger.debug = MagicMock()
412		path = Path("./tests/notus/example.csv").resolve()
413		purepath = PurePath(path).name
414		with path.open("r") as openfile:
415		for line_string in openfile:
416		if line_string.startswith("{"):
417		break
418		reader = DictReader(openfile)
419
420		ret = notus.upload_lsc_from_csv_reader(
421		purepath, general_metadata_dict, reader
422		)
423
424		self.assertTrue(ret)
425		logging.Logger.debug.assert_called_with(
426		"Loaded %d/%d advisories from %s", 1, 1, purepath
427		)
428

greenbone / ospd-openvas

Pull Request — master (#364)

LockFileTestCase.test_update_metadata_failed() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like