tests.testWrapper.TestOspdOpenvas.test_feed_is_outdated() - Code Metrics - Inspection of "Merge pull request #60 from jjnicola/check-feed-ve..." - greenbone/ospd-openvas - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 617948...65d286 )

by Juan José

created 2019-03-25 15:24 UTC

TestOspdOpenvas.test_feed_is_outdated() A

↳ Parent: tests.testWrapper

Complexity

Conditions

Size

Total Lines	15
Code Lines	12

Duplication

Lines	0
Ratio	0 %

Importance

Changes

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

# -*- coding: utf-8 -*-
# Copyright (C) 2018 Greenbone Networks GmbH
#
# SPDX-License-Identifier: GPL-2.0-or-later
#
# This program is free software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 2
# 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 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 St, Fifth Floor, Boston, MA 02110-1301 USA

""" Unit Test for ospd-openvas """

import unittest
from unittest.mock import patch
from ospd_openvas.wrapper import OSPD_PARAMS
from tests.dummywrapper import DummyWrapper
from  ospd_openvas.wrapper import OpenVasVtsFilter
from ospd_openvas.errors import OSPDOpenvasError

OSPD_PARAMS_OUT = {
    'auto_enable_dependencies': {
        'type': 'boolean',
        'name': 'auto_enable_dependencies',
        'default': 1,
        'mandatory': 1,
        'description': 'Automatically enable the plugins that are depended on',
    },
    'cgi_path': {
        'type': 'string',
        'name': 'cgi_path',
        'default': '/cgi-bin:/scripts',
        'mandatory': 1,
        'description': 'Look for default CGIs in /cgi-bin and /scripts',
    },
    'checks_read_timeout': {
        'type': 'integer',
        'name': 'checks_read_timeout',
        'default': 5,
        'mandatory': 1,
        'description': 'Number  of seconds that the security checks will ' \
                       'wait for when doing a recv()',
    },
    'drop_privileges': {
        'type': 'boolean',
        'name': 'drop_privileges',
        'default': 0,
        'mandatory': 1,
        'description': '',
    },
    'network_scan': {
        'type': 'boolean',
        'name': 'network_scan',
        'default': 0,
        'mandatory': 1,
        'description': '',
    },
    'non_simult_ports': {
        'type': 'string',
        'name': 'non_simult_ports',
        'default': '22',
        'mandatory': 1,
        'description': 'Prevent to make two connections on the same given ' \
                       'ports at the same time.',
    },
    'open_sock_max_attempts': {
        'type': 'integer',
        'name': 'open_sock_max_attempts',
        'default': 5,
        'mandatory': 0,
        'description': 'Number of unsuccessful retries to open the socket ' \
                       'before to set the port as closed.',
    },
    'timeout_retry': {
        'type': 'integer',
        'name': 'timeout_retry',
        'default': 5,
        'mandatory': 0,
        'description': 'Number of retries when a socket connection attempt ' \
                       'timesout.',
    },
    'optimize_test': {
        'type': 'integer',
        'name': 'optimize_test',
        'default': 5,
        'mandatory': 0,
        'description': 'By default, openvassd does not trust the remote ' \
                       'host banners.',
    },
    'plugins_timeout': {
        'type': 'integer',
        'name': 'plugins_timeout',
        'default': 5,
        'mandatory': 0,
        'description': 'This is the maximum lifetime, in seconds of a plugin.',
    },
    'report_host_details': {
        'type': 'boolean',
        'name': 'report_host_details',
        'default': 1,
        'mandatory': 1,
        'description': '',
    },
    'safe_checks': {
        'type': 'boolean',
        'name': 'safe_checks',
        'default': 1,
        'mandatory': 1,
        'description': 'Disable the plugins with potential to crash ' \
                       'the remote services',
    },
    'scanner_plugins_timeout': {
        'type': 'integer',
        'name': 'scanner_plugins_timeout',
        'default': 36000,
        'mandatory': 1,
        'description': 'Like plugins_timeout, but for ACT_SCANNER plugins.',
    },
    'time_between_request': {
        'type': 'integer',
        'name': 'time_between_request',
        'default': 0,
        'mandatory': 0,
        'description': 'Allow to set a wait time between two actions ' \
                       '(open, send, close).',
    },
    'unscanned_closed': {
        'type': 'boolean',
        'name': 'unscanned_closed',
        'default': 1,
        'mandatory': 1,
        'description': '',
    },
    'unscanned_closed_udp': {
        'type': 'boolean',
        'name': 'unscanned_closed_udp',
        'default': 1,
        'mandatory': 1,
        'description': '',
    },
    'use_mac_addr': {
        'type': 'boolean',
        'name': 'use_mac_addr',
        'default': 0,
        'mandatory': 0,
        'description': 'To test the local network. ' \
                       'Hosts will be referred to by their MAC address.',
    },
    'vhosts': {
        'type': 'string',
        'name': 'vhosts',
        'default': '',
        'mandatory': 0,
        'description': '',
    },
    'vhosts_ip': {
        'type': 'string',
        'name': 'vhosts_ip',
        'default': '',
        'mandatory': 0,
        'description': '',
    },
}


@patch('ospd_openvas.db.OpenvasDB')
@patch('ospd_openvas.nvticache.NVTICache')
class TestOspdOpenvas(unittest.TestCase):

    @patch('ospd_openvas.wrapper.subprocess')
    def test_redis_nvticache_init(self, mock_subproc, mock_nvti, mock_db):
        mock_subproc.check_call.return_value = True
        w = DummyWrapper(mock_nvti, mock_db)
        w.redis_nvticache_init()
        self.assertEqual(mock_subproc.check_call.call_count, 1)

    @patch('ospd_openvas.wrapper.subprocess')
    def test_parse_param(self, mock_subproc, mock_nvti, mock_db):

        mock_subproc.check_output.return_value = \
            'non_simult_ports = 22\nplugins_folder = /foo/bar'.encode()
        w =  DummyWrapper(mock_nvti, mock_db)
        w.parse_param()
        self.assertEqual(mock_subproc.check_output.call_count, 1)
        self.assertEqual(OSPD_PARAMS, OSPD_PARAMS_OUT)
        self.assertEqual(w.scan_only_params.get('plugins_folder'), '/foo/bar')

    def test_load_vts(self, mock_nvti, mock_db):
        w =  DummyWrapper(mock_nvti, mock_db)
        w.load_vts()
        self.maxDiff = None
        self.assertEqual(w.vts, w.VT)

    def test_get_custom_xml(self, mock_nvti, mock_db):
        out = '<custom><required_ports>Services/www, 80</re' \
              'quired_ports><category>3</category><' \
              'excluded_keys>Settings/disable_cgi_s' \
              'canning</excluded_keys><family>Produ' \
              'ct detection</family><filename>manti' \
              's_detect.nasl</filename><timeout>0</' \
              'timeout></custom>'
        w =  DummyWrapper(mock_nvti, mock_db)
        vt = w.VT['1.3.6.1.4.1.25623.1.0.100061']
        res = w.get_custom_vt_as_xml_str(
            '1.3.6.1.4.1.25623.1.0.100061',
            vt.get('custom'))
        self.assertEqual(len(res), len(out))

    def test_get_severities_xml(self, mock_nvti, mock_db):
        w =  DummyWrapper(mock_nvti, mock_db)
        out = '<severities><severity type="cvss_base_v2">' \
              'AV:N/AC:L/Au:N/C:N/I:N/A:N</severity></severities>'
        vt =w.VT['1.3.6.1.4.1.25623.1.0.100061']
        severities = vt.get('severities')
        res = w.get_severities_vt_as_xml_str(
            '1.3.6.1.4.1.25623.1.0.100061', severities)

        self.assertEqual(res, out)

    def test_get_params_xml(self, mock_nvti, mock_db):
        w =  DummyWrapper(mock_nvti, mock_db)
        out = '<vt_params><vt_param type="checkbox" id="2"><name>Do ' \
              'not randomize the  order  in  which ports are scanned</name' \
              '><default>no</default></vt_param><vt_param type="ent' \
              'ry" id="1"><name>Data length :</name><' \
              '/vt_param></vt_params>'

        vt = w.VT['1.3.6.1.4.1.25623.1.0.100061']
        params = vt.get('vt_params')
        res = w.get_params_vt_as_xml_str(
            '1.3.6.1.4.1.25623.1.0.100061', params)
        self.assertEqual(len(res), len(out))

    def test_get_refs_xml(self, mock_nvti, mock_db):
        w =  DummyWrapper(mock_nvti, mock_db)
        out = '<refs><ref type="url" id="http://www.mantisbt.org/"/>' \
              '</refs>'
        vt = w.VT['1.3.6.1.4.1.25623.1.0.100061']
        refs = vt.get('vt_refs')
        res = w.get_refs_vt_as_xml_str(
            '1.3.6.1.4.1.25623.1.0.100061', refs)

        self.assertEqual(res, out)

    def test_get_dependencies_xml(self, mock_nvti, mock_db):
        w =  DummyWrapper(mock_nvti, mock_db)
        out = '<dependencies><dependency vt_id="1.2.3.4"/><dependency vt' \
              '_id="4.3.2.1"/></dependencies>'
        dep = ['1.2.3.4', '4.3.2.1']
        res = w.get_dependencies_vt_as_xml_str(
            '1.3.6.1.4.1.25623.1.0.100061', dep)

        self.assertEqual(res, out)

    def test_get_ctime_xml(self, mock_nvti, mock_db):
        w =  DummyWrapper(mock_nvti, mock_db)
        out = '<creation_time>2009-03-19 11:22:36 +0100 ' \
              '(Thu, 19 Mar 2009)</creation_time>'
        vt = w.VT['1.3.6.1.4.1.25623.1.0.100061']
        ctime = vt.get('creation_time')
        res = w.get_creation_time_vt_as_xml_str(
            '1.3.6.1.4.1.25623.1.0.100061', ctime)

        self.assertEqual(res, out)

    def test_get_mtime_xml(self, mock_nvti, mock_db):
        w =  DummyWrapper(mock_nvti, mock_db)
        out = '<modification_time>$Date: 2018-08-10 15:09:25 +0200 ' \
              '(Fri, 10 Aug 2018) $</modification_time>'
        vt = w.VT['1.3.6.1.4.1.25623.1.0.100061']
        mtime = vt.get('modification_time')
        res = w.get_modification_time_vt_as_xml_str(
            '1.3.6.1.4.1.25623.1.0.100061', mtime)

        self.assertEqual(res, out)

    def test_get_summary_xml(self, mock_nvti, mock_db):
        w =  DummyWrapper(mock_nvti, mock_db)
        out = '<summary>some summary</summary>'
        vt = w.VT['1.3.6.1.4.1.25623.1.0.100061']
        summary = vt.get('summary')
        res = w.get_summary_vt_as_xml_str(
            '1.3.6.1.4.1.25623.1.0.100061', summary)

        self.assertEqual(res, out)

    def test_get_impact_xml(self, mock_nvti, mock_db):
        w =  DummyWrapper(mock_nvti, mock_db)
        out = '<impact>some impact</impact>'
        vt = w.VT['1.3.6.1.4.1.25623.1.0.100061']
        impact = vt.get('impact')
        res = w.get_impact_vt_as_xml_str(
            '1.3.6.1.4.1.25623.1.0.100061', impact)

        self.assertEqual(res, out)

    def test_get_insight_xml(self, mock_nvti, mock_db):
        w =  DummyWrapper(mock_nvti, mock_db)
        out = '<insight>some insight</insight>'
        vt = w.VT['1.3.6.1.4.1.25623.1.0.100061']
        insight = vt.get('insight')
        res = w.get_insight_vt_as_xml_str(
            '1.3.6.1.4.1.25623.1.0.100061', insight)

        self.assertEqual(res, out)

    def test_get_solution_xml(self, mock_nvti, mock_db):
        w =  DummyWrapper(mock_nvti, mock_db)
        out = '<solution type="WillNotFix">some solution</solution>'
        vt =w.VT['1.3.6.1.4.1.25623.1.0.100061']
        solution = vt.get('solution')
        solution_type = vt.get('solution_type')

        res = w.get_solution_vt_as_xml_str(
            '1.3.6.1.4.1.25623.1.0.100061', solution, solution_type)

        self.assertEqual(res, out)

    def test_get_detection_xml(self, mock_nvti, mock_db):
        w =  DummyWrapper(mock_nvti, mock_db)
        out = '<detection qod_type="remote_banner"/>'
        vt = w.VT['1.3.6.1.4.1.25623.1.0.100061']
        detection_type = vt.get('qod_type')

        res = w.get_detection_vt_as_xml_str(
            '1.3.6.1.4.1.25623.1.0.100061', qod_type=detection_type)

        self.assertEqual(res, out)

    def test_get_affected_xml(self, mock_nvti, mock_db):
        w =  DummyWrapper(mock_nvti, mock_db)
        out = '<affected>some affection</affected>'
        vt = w.VT['1.3.6.1.4.1.25623.1.0.100061']
        affected =  vt.get('affected')

        res = w.get_affected_vt_as_xml_str(
            '1.3.6.1.4.1.25623.1.0.100061', affected=affected)

        self.assertEqual(res, out)

    def test_build_credentials(self, mock_nvti, mock_db):
        w =  DummyWrapper(mock_nvti, mock_db)
        cred_out = [
            'auth_port_ssh|||22',
            'SSH Authorization[entry]:SSH login name:|||username',
            'SSH Authorization[password]:SSH key passphrase:|||pass',
            'SSH Authorization[file]:SSH private key:|||',
            'SMB Authorization[entry]:SMB login:|||username',
            'SMB Authorization[password]:SMB password :|||pass',
            'SNMP Authorization[password]:SNMP Community:some comunity',
            'SNMP Authorization[entry]:SNMPv3 Username:username',
            'SNMP Authorization[password]:SNMPv3 Password:pass',
            'SNMP Authorization[radio]:SNMPv3 Authentication Algorithm:some auth algo',
            'SNMP Authorization[password]:SNMPv3 Privacy Password:privacy pass',
            'SNMP Authorization[radio]:SNMPv3 Privacy Algorithm:privacy algo',
            'ESXi Authorization[entry]:ESXi login name:|||username',
            'ESXi Authorization[password]:ESXi login password:|||pass']
        cred_dict = {
            'ssh': {'type': 'ssh',
                    'port': '22',
                    'username': 'username',
                    'password': 'pass',
            },
            'smb': {'type': 'smb',
                    'username': 'username',
                    'password': 'pass',
            },
            'esxi': {'type': 'esxi',
                    'username': 'username',
                    'password': 'pass',
            },
            'snmp': {'type': 'snmp',
                     'username': 'username',
                     'password': 'pass',
                     'community': 'some comunity',
                     'auth_algorithm': 'some auth algo',
                     'privacy_password': 'privacy pass',
                     'privacy_algorithm': 'privacy algo',
            },
        }
        self.maxDiff=None
        ret = w.build_credentials_as_prefs(cred_dict)
        self.assertEqual(len(ret), len(cred_out))
        self.assertIn('auth_port_ssh|||22', cred_out)

    def test_build_credentials_ssh_up(self, mock_nvti, mock_db):
        w =  DummyWrapper(mock_nvti, mock_db)
        cred_out = [
            'auth_port_ssh|||22',
            'SSH Authorization[entry]:SSH login name:|||username',
            'SSH Authorization[password]:SSH password (unsafe!):|||pass']
        cred_dict = {
            'ssh': {'type': 'up',
                    'port': '22',
                    'username': 'username',
                    'password': 'pass',
            }
        }
        self.maxDiff=None
        ret = w.build_credentials_as_prefs(cred_dict)
        self.assertEqual(ret, cred_out)

    def test_process_vts(self, mock_nvti, mock_db):
        vts = {
            '1.3.6.1.4.1.25623.1.0.100061': {
                'Data length :': 'new value',
                'Do not randomize the  order  in  which ports are ' \
                'scanned': 'new value'},
            'vt_groups': ['family=debian', 'family=general']
        }
        vt_out = (
            ['1.3.6.1.4.1.25623.1.0.100061'],
            [['Mantis Detection[checkbox]:Do not randomize the  order  i' \
              'n  which ports are scanned', 'new value'],
             ['Mantis Detection[entry]:Data length : ', 'new value']]
            )
        w =  DummyWrapper(mock_nvti, mock_db)
        w.load_vts()
        ret = w.process_vts(vts)
        self.assertEqual(len(ret), len(vt_out))

    def test_get_openvas_timestamp_scan_host_end(self, mock_nvti, mock_db):
        mock_db.get_host_scan_scan_end_time.return_value = '12345'
        w =  DummyWrapper(mock_nvti, mock_db)
        w.create_scan('123-456', '192.168.0.1', '192.168.0.1', None, [])
        w.get_openvas_timestamp_scan_host('123-456', '192.168.0.1')
        for result in  w.scan_collection.results_iterator('123-456', False):
            self.assertEqual(result.get('value'), '12345')

    def test_get_openvas_timestamp_scan_host_start(self, mock_nvti, mock_db):
        mock_db.get_host_scan_scan_end_time.return_value = None
        mock_db.get_host_scan_scan_end_time.return_value = '54321'
        w =  DummyWrapper(mock_nvti, mock_db)
        w.create_scan('123-456', '192.168.0.1', '192.168.0.1', None, [])
        w.get_openvas_timestamp_scan_host('123-456', '192.168.0.1')
        for result in  w.scan_collection.results_iterator('123-456', False):
            self.assertEqual(result.get('value'), '54321')

    def test_scan_is_finished(self, mock_nvti, mock_db):
        mock_db.get_single_item.return_value = 'finished'
        w =  DummyWrapper(mock_nvti, mock_db)
        ret = w.scan_is_finished('123-456')
        self.assertEqual(ret, True)

    def test_scan_is_stopped(self, mock_nvti, mock_db):
        mock_db.get_single_item.return_value = 'stop_all'
        mock_db.kb_connect_item.return_value = mock_db
        mock_db.set_redisctx.return_value = None
        w =  DummyWrapper(mock_nvti, mock_db)
        ret = w.scan_is_stopped('123-456')
        self.assertEqual(ret, True)

    @patch('ospd_openvas.wrapper.open')
    def test_feed_is_outdated(self, mock_open, mock_nvti, mock_db):
        mock_nvti.get_feed_version.return_value = '1234'
        mock_open.return_value = ['PLUGIN_SET = "1234";']
        w =  DummyWrapper(mock_nvti, mock_db)
        self.assertRaises(OSPDOpenvasError, w.feed_is_outdated)
        # Return False
        w.scan_only_params['plugins_folder'] = '/foo/bar'
        ret = w.feed_is_outdated()
        self.assertFalse(ret)

        # Return true
        mock_open.return_value = ['PLUGIN_SET = "1235";']
        ret = w.feed_is_outdated()
        self.assertTrue(ret)

class TestFilters(unittest.TestCase):

    def test_format_vt_modification_time(self):
        ovformat = OpenVasVtsFilter()
        td = '$Date: 2018-02-01 02:09:01 +0200 (Thu, 18 Oct 2018) $'
        formatted = ovformat.format_vt_modification_time(td)
        self.assertEqual(formatted, "20180201020901")


1			# -- coding: utf-8 --
2			# Copyright (C) 2018 Greenbone Networks GmbH
3			#
4			# SPDX-License-Identifier: GPL-2.0-or-later
5			#
6			# This program is free software; you can redistribute it and/or
7			# modify it under the terms of the GNU General Public License
8			# as published by the Free Software Foundation; either version 2
9			# of the 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 General Public License for more details.
15			#
16			# You should have received a copy of the GNU General Public License
17			# along with this program; if not, write to the Free Software
18			# Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
19
20			""" Unit Test for ospd-openvas """
21
22			import unittest
23			from unittest.mock import patch
24			from ospd_openvas.wrapper import OSPD_PARAMS
25			from tests.dummywrapper import DummyWrapper
26			from ospd_openvas.wrapper import OpenVasVtsFilter
27			from ospd_openvas.errors import OSPDOpenvasError
28
29			OSPD_PARAMS_OUT = {
30			'auto_enable_dependencies': {
31			'type': 'boolean',
32			'name': 'auto_enable_dependencies',
33			'default': 1,
34			'mandatory': 1,
35			'description': 'Automatically enable the plugins that are depended on',
36			},
37			'cgi_path': {
38			'type': 'string',
39			'name': 'cgi_path',
40			'default': '/cgi-bin:/scripts',
41			'mandatory': 1,
42			'description': 'Look for default CGIs in /cgi-bin and /scripts',
43			},
44			'checks_read_timeout': {
45			'type': 'integer',
46			'name': 'checks_read_timeout',
47			'default': 5,
48			'mandatory': 1,
49			'description': 'Number of seconds that the security checks will ' \
50			'wait for when doing a recv()',
51			},
52			'drop_privileges': {
53			'type': 'boolean',
54			'name': 'drop_privileges',
55			'default': 0,
56			'mandatory': 1,
57			'description': '',
58			},
59			'network_scan': {
60			'type': 'boolean',
61			'name': 'network_scan',
62			'default': 0,
63			'mandatory': 1,
64			'description': '',
65			},
66			'non_simult_ports': {
67			'type': 'string',
68			'name': 'non_simult_ports',
69			'default': '22',
70			'mandatory': 1,
71			'description': 'Prevent to make two connections on the same given ' \
72			'ports at the same time.',
73			},
74			'open_sock_max_attempts': {
75			'type': 'integer',
76			'name': 'open_sock_max_attempts',
77			'default': 5,
78			'mandatory': 0,
79			'description': 'Number of unsuccessful retries to open the socket ' \
80			'before to set the port as closed.',
81			},
82			'timeout_retry': {
83			'type': 'integer',
84			'name': 'timeout_retry',
85			'default': 5,
86			'mandatory': 0,
87			'description': 'Number of retries when a socket connection attempt ' \
88			'timesout.',
89			},
90			'optimize_test': {
91			'type': 'integer',
92			'name': 'optimize_test',
93			'default': 5,
94			'mandatory': 0,
95			'description': 'By default, openvassd does not trust the remote ' \
96			'host banners.',
97			},
98			'plugins_timeout': {
99			'type': 'integer',
100			'name': 'plugins_timeout',
101			'default': 5,
102			'mandatory': 0,
103			'description': 'This is the maximum lifetime, in seconds of a plugin.',
104			},
105			'report_host_details': {
106			'type': 'boolean',
107			'name': 'report_host_details',
108			'default': 1,
109			'mandatory': 1,
110			'description': '',
111			},
112			'safe_checks': {
113			'type': 'boolean',
114			'name': 'safe_checks',
115			'default': 1,
116			'mandatory': 1,
117			'description': 'Disable the plugins with potential to crash ' \
118			'the remote services',
119			},
120			'scanner_plugins_timeout': {
121			'type': 'integer',
122			'name': 'scanner_plugins_timeout',
123			'default': 36000,
124			'mandatory': 1,
125			'description': 'Like plugins_timeout, but for ACT_SCANNER plugins.',
126			},
127			'time_between_request': {
128			'type': 'integer',
129			'name': 'time_between_request',
130			'default': 0,
131			'mandatory': 0,
132			'description': 'Allow to set a wait time between two actions ' \
133			'(open, send, close).',
134			},
135			'unscanned_closed': {
136			'type': 'boolean',
137			'name': 'unscanned_closed',
138			'default': 1,
139			'mandatory': 1,
140			'description': '',
141			},
142			'unscanned_closed_udp': {
143			'type': 'boolean',
144			'name': 'unscanned_closed_udp',
145			'default': 1,
146			'mandatory': 1,
147			'description': '',
148			},
149			'use_mac_addr': {
150			'type': 'boolean',
151			'name': 'use_mac_addr',
152			'default': 0,
153			'mandatory': 0,
154			'description': 'To test the local network. ' \
155			'Hosts will be referred to by their MAC address.',
156			},
157			'vhosts': {
158			'type': 'string',
159			'name': 'vhosts',
160			'default': '',
161			'mandatory': 0,
162			'description': '',
163			},
164			'vhosts_ip': {
165			'type': 'string',
166			'name': 'vhosts_ip',
167			'default': '',
168			'mandatory': 0,
169			'description': '',
170			},
171			}
172
173
174			@patch('ospd_openvas.db.OpenvasDB')
175			@patch('ospd_openvas.nvticache.NVTICache')
176			class TestOspdOpenvas(unittest.TestCase):
177
178			@patch('ospd_openvas.wrapper.subprocess')
179			def test_redis_nvticache_init(self, mock_subproc, mock_nvti, mock_db):
180			mock_subproc.check_call.return_value = True
181			w = DummyWrapper(mock_nvti, mock_db)
182			w.redis_nvticache_init()
183			self.assertEqual(mock_subproc.check_call.call_count, 1)
184
185			@patch('ospd_openvas.wrapper.subprocess')
186			def test_parse_param(self, mock_subproc, mock_nvti, mock_db):
187
188			mock_subproc.check_output.return_value = \
189			'non_simult_ports = 22\nplugins_folder = /foo/bar'.encode()
190			w = DummyWrapper(mock_nvti, mock_db)
191			w.parse_param()
192			self.assertEqual(mock_subproc.check_output.call_count, 1)
193			self.assertEqual(OSPD_PARAMS, OSPD_PARAMS_OUT)
194			self.assertEqual(w.scan_only_params.get('plugins_folder'), '/foo/bar')
195
196			def test_load_vts(self, mock_nvti, mock_db):
197			w = DummyWrapper(mock_nvti, mock_db)
198			w.load_vts()
199			self.maxDiff = None
200			self.assertEqual(w.vts, w.VT)
201
202			def test_get_custom_xml(self, mock_nvti, mock_db):
203			out = '<custom><required_ports>Services/www, 80</re' \
204			'quired_ports><category>3</category><' \
205			'excluded_keys>Settings/disable_cgi_s' \
206			'canning</excluded_keys><family>Produ' \
207			'ct detection</family><filename>manti' \
208			's_detect.nasl</filename><timeout>0</' \
209			'timeout></custom>'
210			w = DummyWrapper(mock_nvti, mock_db)
211			vt = w.VT['1.3.6.1.4.1.25623.1.0.100061']
212			res = w.get_custom_vt_as_xml_str(
213			'1.3.6.1.4.1.25623.1.0.100061',
214			vt.get('custom'))
215			self.assertEqual(len(res), len(out))
216
217			def test_get_severities_xml(self, mock_nvti, mock_db):
218			w = DummyWrapper(mock_nvti, mock_db)
219			out = '<severities><severity type="cvss_base_v2">' \
220			'AV:N/AC:L/Au:N/C:N/I:N/A:N</severity></severities>'
221			vt =w.VT['1.3.6.1.4.1.25623.1.0.100061']
222			severities = vt.get('severities')
223			res = w.get_severities_vt_as_xml_str(
224			'1.3.6.1.4.1.25623.1.0.100061', severities)
225
226			self.assertEqual(res, out)
227
228			def test_get_params_xml(self, mock_nvti, mock_db):
229			w = DummyWrapper(mock_nvti, mock_db)
230			out = '<vt_params><vt_param type="checkbox" id="2"><name>Do ' \
231			'not randomize the order in which ports are scanned</name' \
232			'><default>no</default></vt_param><vt_param type="ent' \
233			'ry" id="1"><name>Data length :</name><' \
234			'/vt_param></vt_params>'
235
236			vt = w.VT['1.3.6.1.4.1.25623.1.0.100061']
237			params = vt.get('vt_params')
238			res = w.get_params_vt_as_xml_str(
239			'1.3.6.1.4.1.25623.1.0.100061', params)
240			self.assertEqual(len(res), len(out))
241
242			def test_get_refs_xml(self, mock_nvti, mock_db):
243			w = DummyWrapper(mock_nvti, mock_db)
244			out = '<refs><ref type="url" id="http://www.mantisbt.org/"/>' \
245			'</refs>'
246			vt = w.VT['1.3.6.1.4.1.25623.1.0.100061']
247			refs = vt.get('vt_refs')
248			res = w.get_refs_vt_as_xml_str(
249			'1.3.6.1.4.1.25623.1.0.100061', refs)
250
251			self.assertEqual(res, out)
252
253			def test_get_dependencies_xml(self, mock_nvti, mock_db):
254			w = DummyWrapper(mock_nvti, mock_db)
255			out = '<dependencies><dependency vt_id="1.2.3.4"/><dependency vt' \
256			'_id="4.3.2.1"/></dependencies>'
257			dep = ['1.2.3.4', '4.3.2.1']
258			res = w.get_dependencies_vt_as_xml_str(
259			'1.3.6.1.4.1.25623.1.0.100061', dep)
260
261			self.assertEqual(res, out)
262
263			def test_get_ctime_xml(self, mock_nvti, mock_db):
264			w = DummyWrapper(mock_nvti, mock_db)
265			out = '<creation_time>2009-03-19 11:22:36 +0100 ' \
266			'(Thu, 19 Mar 2009)</creation_time>'
267			vt = w.VT['1.3.6.1.4.1.25623.1.0.100061']
268			ctime = vt.get('creation_time')
269			res = w.get_creation_time_vt_as_xml_str(
270			'1.3.6.1.4.1.25623.1.0.100061', ctime)
271
272			self.assertEqual(res, out)
273
274			def test_get_mtime_xml(self, mock_nvti, mock_db):
275			w = DummyWrapper(mock_nvti, mock_db)
276			out = '<modification_time>$Date: 2018-08-10 15:09:25 +0200 ' \
277			'(Fri, 10 Aug 2018) $</modification_time>'
278			vt = w.VT['1.3.6.1.4.1.25623.1.0.100061']
279			mtime = vt.get('modification_time')
280			res = w.get_modification_time_vt_as_xml_str(
281			'1.3.6.1.4.1.25623.1.0.100061', mtime)
282
283			self.assertEqual(res, out)
284
285			def test_get_summary_xml(self, mock_nvti, mock_db):
286			w = DummyWrapper(mock_nvti, mock_db)
287			out = '<summary>some summary</summary>'
288			vt = w.VT['1.3.6.1.4.1.25623.1.0.100061']
289			summary = vt.get('summary')
290			res = w.get_summary_vt_as_xml_str(
291			'1.3.6.1.4.1.25623.1.0.100061', summary)
292
293			self.assertEqual(res, out)
294
295			def test_get_impact_xml(self, mock_nvti, mock_db):
296			w = DummyWrapper(mock_nvti, mock_db)
297			out = '<impact>some impact</impact>'
298			vt = w.VT['1.3.6.1.4.1.25623.1.0.100061']
299			impact = vt.get('impact')
300			res = w.get_impact_vt_as_xml_str(
301			'1.3.6.1.4.1.25623.1.0.100061', impact)
302
303			self.assertEqual(res, out)
304
305			def test_get_insight_xml(self, mock_nvti, mock_db):
306			w = DummyWrapper(mock_nvti, mock_db)
307			out = '<insight>some insight</insight>'
308			vt = w.VT['1.3.6.1.4.1.25623.1.0.100061']
309			insight = vt.get('insight')
310			res = w.get_insight_vt_as_xml_str(
311			'1.3.6.1.4.1.25623.1.0.100061', insight)
312
313			self.assertEqual(res, out)
314
315			def test_get_solution_xml(self, mock_nvti, mock_db):
316			w = DummyWrapper(mock_nvti, mock_db)
317			out = '<solution type="WillNotFix">some solution</solution>'
318			vt =w.VT['1.3.6.1.4.1.25623.1.0.100061']
319			solution = vt.get('solution')
320			solution_type = vt.get('solution_type')
321
322			res = w.get_solution_vt_as_xml_str(
323			'1.3.6.1.4.1.25623.1.0.100061', solution, solution_type)
324
325			self.assertEqual(res, out)
326
327			def test_get_detection_xml(self, mock_nvti, mock_db):
328			w = DummyWrapper(mock_nvti, mock_db)
329			out = '<detection qod_type="remote_banner"/>'
330			vt = w.VT['1.3.6.1.4.1.25623.1.0.100061']
331			detection_type = vt.get('qod_type')
332
333			res = w.get_detection_vt_as_xml_str(
334			'1.3.6.1.4.1.25623.1.0.100061', qod_type=detection_type)
335
336			self.assertEqual(res, out)
337
338			def test_get_affected_xml(self, mock_nvti, mock_db):
339			w = DummyWrapper(mock_nvti, mock_db)
340			out = '<affected>some affection</affected>'
341			vt = w.VT['1.3.6.1.4.1.25623.1.0.100061']
342			affected = vt.get('affected')
343
344			res = w.get_affected_vt_as_xml_str(
345			'1.3.6.1.4.1.25623.1.0.100061', affected=affected)
346
347			self.assertEqual(res, out)
348
349			def test_build_credentials(self, mock_nvti, mock_db):
350			w = DummyWrapper(mock_nvti, mock_db)
351			cred_out = [
352			'auth_port_ssh\|\|\|22',
353			'SSH Authorization[entry]:SSH login name:\|\|\|username',
354			'SSH Authorization[password]:SSH key passphrase:\|\|\|pass',
355			'SSH Authorization[file]:SSH private key:\|\|\|',
356			'SMB Authorization[entry]:SMB login:\|\|\|username',
357			'SMB Authorization[password]:SMB password :\|\|\|pass',
358			'SNMP Authorization[password]:SNMP Community:some comunity',
359			'SNMP Authorization[entry]:SNMPv3 Username:username',
360			'SNMP Authorization[password]:SNMPv3 Password:pass',
361			'SNMP Authorization[radio]:SNMPv3 Authentication Algorithm:some auth algo',
362			'SNMP Authorization[password]:SNMPv3 Privacy Password:privacy pass',
363			'SNMP Authorization[radio]:SNMPv3 Privacy Algorithm:privacy algo',
364			'ESXi Authorization[entry]:ESXi login name:\|\|\|username',
365			'ESXi Authorization[password]:ESXi login password:\|\|\|pass']
366			cred_dict = {
367			'ssh': {'type': 'ssh',
368			'port': '22',
369			'username': 'username',
370			'password': 'pass',
371			},
372			'smb': {'type': 'smb',
373			'username': 'username',
374			'password': 'pass',
375			},
376			'esxi': {'type': 'esxi',
377			'username': 'username',
378			'password': 'pass',
379			},
380			'snmp': {'type': 'snmp',
381			'username': 'username',
382			'password': 'pass',
383			'community': 'some comunity',
384			'auth_algorithm': 'some auth algo',
385			'privacy_password': 'privacy pass',
386			'privacy_algorithm': 'privacy algo',
387			},
388			}
389			self.maxDiff=None
390			ret = w.build_credentials_as_prefs(cred_dict)
391			self.assertEqual(len(ret), len(cred_out))
392			self.assertIn('auth_port_ssh\|\|\|22', cred_out)
393
394			def test_build_credentials_ssh_up(self, mock_nvti, mock_db):
395			w = DummyWrapper(mock_nvti, mock_db)
396			cred_out = [
397			'auth_port_ssh\|\|\|22',
398			'SSH Authorization[entry]:SSH login name:\|\|\|username',
399			'SSH Authorization[password]:SSH password (unsafe!):\|\|\|pass']
400			cred_dict = {
401			'ssh': {'type': 'up',
402			'port': '22',
403			'username': 'username',
404			'password': 'pass',
405			}
406			}
407			self.maxDiff=None
408			ret = w.build_credentials_as_prefs(cred_dict)
409			self.assertEqual(ret, cred_out)
410
411			def test_process_vts(self, mock_nvti, mock_db):
412			vts = {
413			'1.3.6.1.4.1.25623.1.0.100061': {
414			'Data length :': 'new value',
415			'Do not randomize the order in which ports are ' \
416			'scanned': 'new value'},
417			'vt_groups': ['family=debian', 'family=general']
418			}
419			vt_out = (
420			['1.3.6.1.4.1.25623.1.0.100061'],
421			[['Mantis Detection[checkbox]:Do not randomize the order i' \
422			'n which ports are scanned', 'new value'],
423			['Mantis Detection[entry]:Data length : ', 'new value']]
424			)
425			w = DummyWrapper(mock_nvti, mock_db)
426			w.load_vts()
427			ret = w.process_vts(vts)
428			self.assertEqual(len(ret), len(vt_out))
429
430			def test_get_openvas_timestamp_scan_host_end(self, mock_nvti, mock_db):
431			mock_db.get_host_scan_scan_end_time.return_value = '12345'
432			w = DummyWrapper(mock_nvti, mock_db)
433			w.create_scan('123-456', '192.168.0.1', '192.168.0.1', None, [])
434			w.get_openvas_timestamp_scan_host('123-456', '192.168.0.1')
435			for result in w.scan_collection.results_iterator('123-456', False):
436			self.assertEqual(result.get('value'), '12345')
437
438			def test_get_openvas_timestamp_scan_host_start(self, mock_nvti, mock_db):
439			mock_db.get_host_scan_scan_end_time.return_value = None
440			mock_db.get_host_scan_scan_end_time.return_value = '54321'
441			w = DummyWrapper(mock_nvti, mock_db)
442			w.create_scan('123-456', '192.168.0.1', '192.168.0.1', None, [])
443			w.get_openvas_timestamp_scan_host('123-456', '192.168.0.1')
444			for result in w.scan_collection.results_iterator('123-456', False):
445			self.assertEqual(result.get('value'), '54321')
446
447			def test_scan_is_finished(self, mock_nvti, mock_db):
448			mock_db.get_single_item.return_value = 'finished'
449			w = DummyWrapper(mock_nvti, mock_db)
450			ret = w.scan_is_finished('123-456')
451			self.assertEqual(ret, True)
452
453			def test_scan_is_stopped(self, mock_nvti, mock_db):
454			mock_db.get_single_item.return_value = 'stop_all'
455			mock_db.kb_connect_item.return_value = mock_db
456			mock_db.set_redisctx.return_value = None
457			w = DummyWrapper(mock_nvti, mock_db)
458			ret = w.scan_is_stopped('123-456')
459			self.assertEqual(ret, True)
460
461			@patch('ospd_openvas.wrapper.open')
462			def test_feed_is_outdated(self, mock_open, mock_nvti, mock_db):
463			mock_nvti.get_feed_version.return_value = '1234'
464			mock_open.return_value = ['PLUGIN_SET = "1234";']
465			w = DummyWrapper(mock_nvti, mock_db)
466			self.assertRaises(OSPDOpenvasError, w.feed_is_outdated)
467			# Return False
468			w.scan_only_params['plugins_folder'] = '/foo/bar'
469			ret = w.feed_is_outdated()
470			self.assertFalse(ret)
471
472			# Return true
473			mock_open.return_value = ['PLUGIN_SET = "1235";']
474			ret = w.feed_is_outdated()
475			self.assertTrue(ret)
476
477			class TestFilters(unittest.TestCase):
478
479			def test_format_vt_modification_time(self):
480			ovformat = OpenVasVtsFilter()
481			td = '$Date: 2018-02-01 02:09:01 +0200 (Thu, 18 Oct 2018) $'
482			formatted = ovformat.format_vt_modification_time(td)
483			self.assertEqual(formatted, "20180201020901")
484

greenbone / ospd-openvas

Push — master ( 617948...65d286 )

TestOspdOpenvas.test_feed_is_outdated() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like