tests.test_daemon.TestOspdOpenvas.test_process_vts_not_found() - Code Metrics - Inspection of "Merge pull request #110 from jjnicola/finished-tar..." - greenbone/ospd-openvas - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 524bc6...41b97c )

unknown

created 2019-09-18 07:28 UTC

TestOspdOpenvas.test_process_vts_not_found() A

↳ Parent: tests.test_daemon

Complexity

Conditions

Size

Total Lines	10
Code Lines	9

Duplication

Lines	0
Ratio	0 %

Importance

Changes

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

# -*- coding: utf-8 -*-
# Copyright (C) 2018-2019 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

# pylint: disable=invalid-name,line-too-long

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

import unittest
import io

from unittest.mock import patch, mock_open

from tests.dummydaemon import DummyDaemon

from ospd_openvas.daemon import OSPD_PARAMS, OpenVasVtsFilter, Path
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, openvas 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.daemon.subprocess')
    def test_redis_nvticache_init(self, mock_subproc, mock_nvti, mock_db):
        mock_subproc.check_call.return_value = True
        w = DummyDaemon(mock_nvti, mock_db)
        mock_subproc.reset_mock()
        w.redis_nvticache_init()
        self.assertEqual(mock_subproc.check_call.call_count, 1)

    @patch('ospd_openvas.daemon.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 = DummyDaemon(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')

    @patch('ospd_openvas.daemon.subprocess')
    def test_sudo_available(self, mock_subproc, mock_nvti, mock_db):
        mock_subproc.check_call.return_value = 0
        w = DummyDaemon(mock_nvti, mock_db)
        w._sudo_available = None  # pylint: disable=protected-access
        w.sudo_available  # pylint: disable=pointless-statement
        self.assertTrue(w.sudo_available)

    def test_load_vts(self, mock_nvti, mock_db):
        w = DummyDaemon(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 = DummyDaemon(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 = DummyDaemon(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 = DummyDaemon(mock_nvti, mock_db)
        out = (
            '<params><param type="checkbox" id="2"><name>Do '
            'not randomize the  order  in  which ports are scanned</name'
            '><default>no</default></param><param type="ent'
            'ry" id="1"><name>Data length :</name><'
            '/param></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 = DummyDaemon(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 = DummyDaemon(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 = DummyDaemon(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 = DummyDaemon(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 = DummyDaemon(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 = DummyDaemon(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 = DummyDaemon(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 = DummyDaemon(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 = DummyDaemon(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 = DummyDaemon(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 = DummyDaemon(mock_nvti, mock_db)

        cred_out = [
            '1.3.6.1.4.1.25623.1.0.105058:1:entry:ESXi login name:|||username',
            '1.3.6.1.4.1.25623.1.0.105058:2:password:ESXi login password:|||pass',
            'auth_port_ssh|||22',
            '1.3.6.1.4.1.25623.1.0.103591:1:entry:SSH login name:|||username',
            '1.3.6.1.4.1.25623.1.0.103591:2:password:SSH key passphrase:|||pass',
            '1.3.6.1.4.1.25623.1.0.103591:4:file:SSH private key:|||',
            '1.3.6.1.4.1.25623.1.0.90023:1:entry:SMB login:|||username',
            '1.3.6.1.4.1.25623.1.0.90023:2:password]:SMB password :|||pass',
            '1.3.6.1.4.1.25623.1.0.105076:1:password:SNMP Community:some comunity',
            '1.3.6.1.4.1.25623.1.0.105076:2:entry:SNMPv3 Username:username',
            '1.3.6.1.4.1.25623.1.0.105076:3:password:SNMPv3 Password:pass',
            '1.3.6.1.4.1.25623.1.0.105076:4:radio:SNMPv3 Authentication Algorithm:some auth algo',
            '1.3.6.1.4.1.25623.1.0.105076:5:password:SNMPv3 Privacy Password:privacy pass',
            '1.3.6.1.4.1.25623.1.0.105076:6:radio:SNMPv3 Privacy Algorithm:privacy algo',
        ]
        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)
        self.assertIn(
            '1.3.6.1.4.1.25623.1.0.90023:1:entry:SMB login:|||username',
            cred_out,
        )

    def test_build_credentials_ssh_up(self, mock_nvti, mock_db):
        w = DummyDaemon(mock_nvti, mock_db)
        cred_out = [
            'auth_port_ssh|||22',
            '1.3.6.1.4.1.25623.1.0.103591:1:entry:SSH login name:|||username',
            '1.3.6.1.4.1.25623.1.0.103591:3: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': {'1': 'new value'},
            'vt_groups': ['family=debian', 'family=general'],
        }
        vt_out = (
            ['1.3.6.1.4.1.25623.1.0.100061'],
            [
                [
                    '1.3.6.1.4.1.25623.1.0.100061:1:entry:Data length :',
                    'new value',
                ]
            ],
        )
        w = DummyDaemon(mock_nvti, mock_db)
        w.load_vts()
        ret = w.process_vts(vts)
        self.assertEqual(ret, vt_out)

    def test_process_vts_bad_param_id(self, mock_nvti, mock_db):
        vts = {
            '1.3.6.1.4.1.25623.1.0.100061': {'3': 'new value'},
            'vt_groups': ['family=debian', 'family=general'],
        }
        w = DummyDaemon(mock_nvti, mock_db)
        w.load_vts()
        ret = w.process_vts(vts)
        self.assertFalse(ret[1])

    @patch('logging.Logger.warning')
    def test_process_vts_not_found(self, mock_logger, mock_nvti, mock_db):
        vts = {
            '1.3.6.1.4.1.25623.1.0.100065': {'3': 'new value'},
            'vt_groups': ['family=debian', 'family=general'],
        }
        w = DummyDaemon(mock_nvti, mock_db)
        w.load_vts()
        ret = w.process_vts(vts)
        self.assertTrue(mock_logger.called)

    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 = DummyDaemon(mock_nvti, mock_db)
        targets = [['192.168.0.1', 'port', 'cred', 'exclude_host']]
        w.create_scan('123-456', targets, 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 = DummyDaemon(mock_nvti, mock_db)
        targets = [['192.168.0.1', 'port', 'cred', 'exclude_host']]
        w.create_scan('123-456', targets, 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_host_is_finished(self, mock_nvti, mock_db):
        mock_db.get_single_item.return_value = 'finished'
        w = DummyDaemon(mock_nvti, mock_db)
        ret = w.host_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 = DummyDaemon(mock_nvti, mock_db)
        ret = w.scan_is_stopped('123-456')
        self.assertEqual(ret, True)

    @patch('ospd_openvas.daemon.open')
    def test_feed_is_outdated_none(self, mock_open, mock_nvti, mock_db):
        w = DummyDaemon(mock_nvti, mock_db)
        # Mock parse_param, because feed_is_oudated() will call it.
        with patch.object(w, 'parse_param', return_value=None):
            # Return None
            w.scan_only_params['plugins_folder'] = '/foo/bar'
            ret = w.feed_is_outdated('1234')
            self.assertIsNone(ret)

    def test_feed_is_outdated_true(self, mock_nvti, mock_db):
        w = DummyDaemon(mock_nvti, mock_db)
        # Mock parse_param, because feed_is_oudated() will call it.
        with patch.object(w, 'parse_param', return_value=None):
            with patch.object(Path, 'exists', return_value=True):
                read_data = 'PLUGIN_SET = "1235";'
                with patch("builtins.open",
                    return_value=io.StringIO(read_data)):
                    # Return True
                    w.scan_only_params['plugins_folder'] = '/foo/bar'
                    ret = w.feed_is_outdated('1234')
                    self.assertTrue(ret)

    def test_feed_is_outdated_false(self, mock_nvti, mock_db):
        w = DummyDaemon(mock_nvti, mock_db)
        # Mock parse_param, because feed_is_oudated() will call it.
        with patch.object(w, 'parse_param', return_value=None):
            read_data = 'PLUGIN_SET = "1234";'
            with patch.object(Path, 'exists', return_value=True):
                read_data = 'PLUGIN_SET = "1234"';
                with patch("builtins.open",
                    return_value=io.StringIO(read_data)):
                    # Return True
                    w.scan_only_params['plugins_folder'] = '/foo/bar'
                    ret = w.feed_is_outdated('1234')
                    self.assertFalse(ret)

    @patch('ospd_openvas.daemon.OSPDaemon.add_scan_log')
    def test_get_openvas_result(self, mock_ospd, mock_nvti, mock_db):
        results = ["LOG||| |||general/Host_Details||| |||Host dead", None]
        mock_db.get_result.side_effect = results
        w = DummyDaemon(mock_nvti, mock_db)
        w.load_vts()
        mock_ospd.return_value = None
        w.get_openvas_result('123-456', 'localhost')
        mock_ospd.assert_called_with(
            '123-456',
            host='localhost',
            hostname=' ',
            name='',
            port='general/Host_Details',
            qod='',
            test_id=' ',
            value='Host dead',
        )

    @patch('ospd_openvas.daemon.OSPDaemon.set_scan_target_progress')
    def test_update_progress(self, mock_ospd, mock_nvti, mock_db):
        msg = '0/-1'
        targets = [['localhost', 'port', 'cred', 'exclude_host']]
        w = DummyDaemon(mock_nvti, mock_db)
        w.create_scan('123-456', targets, None, [])

        mock_ospd.return_value = None
        w.update_progress('123-456', 'localhost', 'localhost', msg)
        mock_ospd.assert_called_with('123-456', 'localhost', 'localhost', 100)


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-2019 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			# pylint: disable=invalid-name,line-too-long
21
22			""" Unit Test for ospd-openvas """
23
24			import unittest
25			import io
26
27			from unittest.mock import patch, mock_open
28
29			from tests.dummydaemon import DummyDaemon
30
31			from ospd_openvas.daemon import OSPD_PARAMS, OpenVasVtsFilter, Path
32			from ospd_openvas.errors import OspdOpenvasError
33
34			OSPD_PARAMS_OUT = {
35			'auto_enable_dependencies': {
36			'type': 'boolean',
37			'name': 'auto_enable_dependencies',
38			'default': 1,
39			'mandatory': 1,
40			'description': 'Automatically enable the plugins that are depended on',
41			},
42			'cgi_path': {
43			'type': 'string',
44			'name': 'cgi_path',
45			'default': '/cgi-bin:/scripts',
46			'mandatory': 1,
47			'description': 'Look for default CGIs in /cgi-bin and /scripts',
48			},
49			'checks_read_timeout': {
50			'type': 'integer',
51			'name': 'checks_read_timeout',
52			'default': 5,
53			'mandatory': 1,
54			'description': 'Number of seconds that the security checks will '
55			'wait for when doing a recv()',
56			},
57			'drop_privileges': {
58			'type': 'boolean',
59			'name': 'drop_privileges',
60			'default': 0,
61			'mandatory': 1,
62			'description': '',
63			},
64			'network_scan': {
65			'type': 'boolean',
66			'name': 'network_scan',
67			'default': 0,
68			'mandatory': 1,
69			'description': '',
70			},
71			'non_simult_ports': {
72			'type': 'string',
73			'name': 'non_simult_ports',
74			'default': '22',
75			'mandatory': 1,
76			'description': 'Prevent to make two connections on the same given '
77			'ports at the same time.',
78			},
79			'open_sock_max_attempts': {
80			'type': 'integer',
81			'name': 'open_sock_max_attempts',
82			'default': 5,
83			'mandatory': 0,
84			'description': 'Number of unsuccessful retries to open the socket '
85			'before to set the port as closed.',
86			},
87			'timeout_retry': {
88			'type': 'integer',
89			'name': 'timeout_retry',
90			'default': 5,
91			'mandatory': 0,
92			'description': 'Number of retries when a socket connection attempt '
93			'timesout.',
94			},
95			'optimize_test': {
96			'type': 'integer',
97			'name': 'optimize_test',
98			'default': 5,
99			'mandatory': 0,
100			'description': 'By default, openvas does not trust the remote '
101			'host banners.',
102			},
103			'plugins_timeout': {
104			'type': 'integer',
105			'name': 'plugins_timeout',
106			'default': 5,
107			'mandatory': 0,
108			'description': 'This is the maximum lifetime, in seconds of a plugin.',
109			},
110			'report_host_details': {
111			'type': 'boolean',
112			'name': 'report_host_details',
113			'default': 1,
114			'mandatory': 1,
115			'description': '',
116			},
117			'safe_checks': {
118			'type': 'boolean',
119			'name': 'safe_checks',
120			'default': 1,
121			'mandatory': 1,
122			'description': 'Disable the plugins with potential to crash '
123			'the remote services',
124			},
125			'scanner_plugins_timeout': {
126			'type': 'integer',
127			'name': 'scanner_plugins_timeout',
128			'default': 36000,
129			'mandatory': 1,
130			'description': 'Like plugins_timeout, but for ACT_SCANNER plugins.',
131			},
132			'time_between_request': {
133			'type': 'integer',
134			'name': 'time_between_request',
135			'default': 0,
136			'mandatory': 0,
137			'description': 'Allow to set a wait time between two actions '
138			'(open, send, close).',
139			},
140			'unscanned_closed': {
141			'type': 'boolean',
142			'name': 'unscanned_closed',
143			'default': 1,
144			'mandatory': 1,
145			'description': '',
146			},
147			'unscanned_closed_udp': {
148			'type': 'boolean',
149			'name': 'unscanned_closed_udp',
150			'default': 1,
151			'mandatory': 1,
152			'description': '',
153			},
154			'use_mac_addr': {
155			'type': 'boolean',
156			'name': 'use_mac_addr',
157			'default': 0,
158			'mandatory': 0,
159			'description': 'To test the local network. '
160			'Hosts will be referred to by their MAC address.',
161			},
162			'vhosts': {
163			'type': 'string',
164			'name': 'vhosts',
165			'default': '',
166			'mandatory': 0,
167			'description': '',
168			},
169			'vhosts_ip': {
170			'type': 'string',
171			'name': 'vhosts_ip',
172			'default': '',
173			'mandatory': 0,
174			'description': '',
175			},
176			}
177
178
179			@patch('ospd_openvas.db.OpenvasDB')
180			@patch('ospd_openvas.nvticache.NVTICache')
181			class TestOspdOpenvas(unittest.TestCase):
182			@patch('ospd_openvas.daemon.subprocess')
183			def test_redis_nvticache_init(self, mock_subproc, mock_nvti, mock_db):
184			mock_subproc.check_call.return_value = True
185			w = DummyDaemon(mock_nvti, mock_db)
186			mock_subproc.reset_mock()
187			w.redis_nvticache_init()
188			self.assertEqual(mock_subproc.check_call.call_count, 1)
189
190			@patch('ospd_openvas.daemon.subprocess')
191			def test_parse_param(self, mock_subproc, mock_nvti, mock_db):
192
193			mock_subproc.check_output.return_value = (
194			'non_simult_ports = 22\nplugins_folder = /foo/bar'.encode()
195			)
196			w = DummyDaemon(mock_nvti, mock_db)
197			w.parse_param()
198			self.assertEqual(mock_subproc.check_output.call_count, 1)
199			self.assertEqual(OSPD_PARAMS, OSPD_PARAMS_OUT)
200			self.assertEqual(w.scan_only_params.get('plugins_folder'), '/foo/bar')
201
202			@patch('ospd_openvas.daemon.subprocess')
203			def test_sudo_available(self, mock_subproc, mock_nvti, mock_db):
204			mock_subproc.check_call.return_value = 0
205			w = DummyDaemon(mock_nvti, mock_db)
206			w._sudo_available = None # pylint: disable=protected-access
207			w.sudo_available # pylint: disable=pointless-statement
208			self.assertTrue(w.sudo_available)
209
210			def test_load_vts(self, mock_nvti, mock_db):
211			w = DummyDaemon(mock_nvti, mock_db)
212			w.load_vts()
213			self.maxDiff = None
214			self.assertEqual(w.vts, w.VT)
215
216			def test_get_custom_xml(self, mock_nvti, mock_db):
217			out = (
218			'<custom><required_ports>Services/www, 80</re'
219			'quired_ports><category>3</category><'
220			'excluded_keys>Settings/disable_cgi_s'
221			'canning</excluded_keys><family>Produ'
222			'ct detection</family><filename>manti'
223			's_detect.nasl</filename><timeout>0</'
224			'timeout></custom>'
225			)
226			w = DummyDaemon(mock_nvti, mock_db)
227			vt = w.VT['1.3.6.1.4.1.25623.1.0.100061']
228			res = w.get_custom_vt_as_xml_str(
229			'1.3.6.1.4.1.25623.1.0.100061', vt.get('custom')
230			)
231			self.assertEqual(len(res), len(out))
232
233			def test_get_severities_xml(self, mock_nvti, mock_db):
234			w = DummyDaemon(mock_nvti, mock_db)
235			out = (
236			'<severities><severity type="cvss_base_v2">'
237			'AV:N/AC:L/Au:N/C:N/I:N/A:N</severity></severities>'
238			)
239			vt = w.VT['1.3.6.1.4.1.25623.1.0.100061']
240			severities = vt.get('severities')
241			res = w.get_severities_vt_as_xml_str(
242			'1.3.6.1.4.1.25623.1.0.100061', severities
243			)
244
245			self.assertEqual(res, out)
246
247			def test_get_params_xml(self, mock_nvti, mock_db):
248			w = DummyDaemon(mock_nvti, mock_db)
249			out = (
250			'<params><param type="checkbox" id="2"><name>Do '
251			'not randomize the order in which ports are scanned</name'
252			'><default>no</default></param><param type="ent'
253			'ry" id="1"><name>Data length :</name><'
254			'/param></params>'
255			)
256
257			vt = w.VT['1.3.6.1.4.1.25623.1.0.100061']
258			params = vt.get('vt_params')
259			res = w.get_params_vt_as_xml_str('1.3.6.1.4.1.25623.1.0.100061', params)
260			self.assertEqual(len(res), len(out))
261
262			def test_get_refs_xml(self, mock_nvti, mock_db):
263			w = DummyDaemon(mock_nvti, mock_db)
264			out = '<refs><ref type="url" id="http://www.mantisbt.org/"/>' '</refs>'
265			vt = w.VT['1.3.6.1.4.1.25623.1.0.100061']
266			refs = vt.get('vt_refs')
267			res = w.get_refs_vt_as_xml_str('1.3.6.1.4.1.25623.1.0.100061', refs)
268
269			self.assertEqual(res, out)
270
271			def test_get_dependencies_xml(self, mock_nvti, mock_db):
272			w = DummyDaemon(mock_nvti, mock_db)
273			out = (
274			'<dependencies><dependency vt_id="1.2.3.4"/><dependency vt'
275			'_id="4.3.2.1"/></dependencies>'
276			)
277			dep = ['1.2.3.4', '4.3.2.1']
278			res = w.get_dependencies_vt_as_xml_str(
279			'1.3.6.1.4.1.25623.1.0.100061', dep
280			)
281
282			self.assertEqual(res, out)
283
284			def test_get_ctime_xml(self, mock_nvti, mock_db):
285			w = DummyDaemon(mock_nvti, mock_db)
286			out = (
287			'<creation_time>2009-03-19 11:22:36 +0100 '
288			'(Thu, 19 Mar 2009)</creation_time>'
289			)
290			vt = w.VT['1.3.6.1.4.1.25623.1.0.100061']
291			ctime = vt.get('creation_time')
292			res = w.get_creation_time_vt_as_xml_str(
293			'1.3.6.1.4.1.25623.1.0.100061', ctime
294			)
295
296			self.assertEqual(res, out)
297
298			def test_get_mtime_xml(self, mock_nvti, mock_db):
299			w = DummyDaemon(mock_nvti, mock_db)
300			out = (
301			'<modification_time>$Date: 2018-08-10 15:09:25 +0200 '
302			'(Fri, 10 Aug 2018) $</modification_time>'
303			)
304			vt = w.VT['1.3.6.1.4.1.25623.1.0.100061']
305			mtime = vt.get('modification_time')
306			res = w.get_modification_time_vt_as_xml_str(
307			'1.3.6.1.4.1.25623.1.0.100061', mtime
308			)
309
310			self.assertEqual(res, out)
311
312			def test_get_summary_xml(self, mock_nvti, mock_db):
313			w = DummyDaemon(mock_nvti, mock_db)
314			out = '<summary>some summary</summary>'
315			vt = w.VT['1.3.6.1.4.1.25623.1.0.100061']
316			summary = vt.get('summary')
317			res = w.get_summary_vt_as_xml_str(
318			'1.3.6.1.4.1.25623.1.0.100061', summary
319			)
320
321			self.assertEqual(res, out)
322
323			def test_get_impact_xml(self, mock_nvti, mock_db):
324			w = DummyDaemon(mock_nvti, mock_db)
325			out = '<impact>some impact</impact>'
326			vt = w.VT['1.3.6.1.4.1.25623.1.0.100061']
327			impact = vt.get('impact')
328			res = w.get_impact_vt_as_xml_str('1.3.6.1.4.1.25623.1.0.100061', impact)
329
330			self.assertEqual(res, out)
331
332			def test_get_insight_xml(self, mock_nvti, mock_db):
333			w = DummyDaemon(mock_nvti, mock_db)
334			out = '<insight>some insight</insight>'
335			vt = w.VT['1.3.6.1.4.1.25623.1.0.100061']
336			insight = vt.get('insight')
337			res = w.get_insight_vt_as_xml_str(
338			'1.3.6.1.4.1.25623.1.0.100061', insight
339			)
340
341			self.assertEqual(res, out)
342
343			def test_get_solution_xml(self, mock_nvti, mock_db):
344			w = DummyDaemon(mock_nvti, mock_db)
345			out = '<solution type="WillNotFix">some solution</solution>'
346			vt = w.VT['1.3.6.1.4.1.25623.1.0.100061']
347			solution = vt.get('solution')
348			solution_type = vt.get('solution_type')
349
350			res = w.get_solution_vt_as_xml_str(
351			'1.3.6.1.4.1.25623.1.0.100061', solution, solution_type
352			)
353
354			self.assertEqual(res, out)
355
356			def test_get_detection_xml(self, mock_nvti, mock_db):
357			w = DummyDaemon(mock_nvti, mock_db)
358			out = '<detection qod_type="remote_banner"/>'
359			vt = w.VT['1.3.6.1.4.1.25623.1.0.100061']
360			detection_type = vt.get('qod_type')
361
362			res = w.get_detection_vt_as_xml_str(
363			'1.3.6.1.4.1.25623.1.0.100061', qod_type=detection_type
364			)
365
366			self.assertEqual(res, out)
367
368			def test_get_affected_xml(self, mock_nvti, mock_db):
369			w = DummyDaemon(mock_nvti, mock_db)
370			out = '<affected>some affection</affected>'
371			vt = w.VT['1.3.6.1.4.1.25623.1.0.100061']
372			affected = vt.get('affected')
373
374			res = w.get_affected_vt_as_xml_str(
375			'1.3.6.1.4.1.25623.1.0.100061', affected=affected
376			)
377
378			self.assertEqual(res, out)
379
380			def test_build_credentials(self, mock_nvti, mock_db):
381			w = DummyDaemon(mock_nvti, mock_db)
382
383			cred_out = [
384			'1.3.6.1.4.1.25623.1.0.105058:1:entry:ESXi login name:\|\|\|username',
385			'1.3.6.1.4.1.25623.1.0.105058:2:password:ESXi login password:\|\|\|pass',
386			'auth_port_ssh\|\|\|22',
387			'1.3.6.1.4.1.25623.1.0.103591:1:entry:SSH login name:\|\|\|username',
388			'1.3.6.1.4.1.25623.1.0.103591:2:password:SSH key passphrase:\|\|\|pass',
389			'1.3.6.1.4.1.25623.1.0.103591:4:file:SSH private key:\|\|\|',
390			'1.3.6.1.4.1.25623.1.0.90023:1:entry:SMB login:\|\|\|username',
391			'1.3.6.1.4.1.25623.1.0.90023:2:password]:SMB password :\|\|\|pass',
392			'1.3.6.1.4.1.25623.1.0.105076:1:password:SNMP Community:some comunity',
393			'1.3.6.1.4.1.25623.1.0.105076:2:entry:SNMPv3 Username:username',
394			'1.3.6.1.4.1.25623.1.0.105076:3:password:SNMPv3 Password:pass',
395			'1.3.6.1.4.1.25623.1.0.105076:4:radio:SNMPv3 Authentication Algorithm:some auth algo',
396			'1.3.6.1.4.1.25623.1.0.105076:5:password:SNMPv3 Privacy Password:privacy pass',
397			'1.3.6.1.4.1.25623.1.0.105076:6:radio:SNMPv3 Privacy Algorithm:privacy algo',
398			]
399			cred_dict = {
400			'ssh': {
401			'type': 'ssh',
402			'port': '22',
403			'username': 'username',
404			'password': 'pass',
405			},
406			'smb': {'type': 'smb', 'username': 'username', 'password': 'pass'},
407			'esxi': {
408			'type': 'esxi',
409			'username': 'username',
410			'password': 'pass',
411			},
412			'snmp': {
413			'type': 'snmp',
414			'username': 'username',
415			'password': 'pass',
416			'community': 'some comunity',
417			'auth_algorithm': 'some auth algo',
418			'privacy_password': 'privacy pass',
419			'privacy_algorithm': 'privacy algo',
420			},
421			}
422			self.maxDiff = None
423			ret = w.build_credentials_as_prefs(cred_dict)
424			self.assertEqual(len(ret), len(cred_out))
425			self.assertIn('auth_port_ssh\|\|\|22', cred_out)
426			self.assertIn(
427			'1.3.6.1.4.1.25623.1.0.90023:1:entry:SMB login:\|\|\|username',
428			cred_out,
429			)
430
431			def test_build_credentials_ssh_up(self, mock_nvti, mock_db):
432			w = DummyDaemon(mock_nvti, mock_db)
433			cred_out = [
434			'auth_port_ssh\|\|\|22',
435			'1.3.6.1.4.1.25623.1.0.103591:1:entry:SSH login name:\|\|\|username',
436			'1.3.6.1.4.1.25623.1.0.103591:3:password:SSH password (unsafe!):\|\|\|pass',
437			]
438			cred_dict = {
439			'ssh': {
440			'type': 'up',
441			'port': '22',
442			'username': 'username',
443			'password': 'pass',
444			}
445			}
446			self.maxDiff = None
447			ret = w.build_credentials_as_prefs(cred_dict)
448			self.assertEqual(ret, cred_out)
449
450			def test_process_vts(self, mock_nvti, mock_db):
451			vts = {
452			'1.3.6.1.4.1.25623.1.0.100061': {'1': 'new value'},
453			'vt_groups': ['family=debian', 'family=general'],
454			}
455			vt_out = (
456			['1.3.6.1.4.1.25623.1.0.100061'],
457			[
458			[
459			'1.3.6.1.4.1.25623.1.0.100061:1:entry:Data length :',
460			'new value',
461			]
462			],
463			)
464			w = DummyDaemon(mock_nvti, mock_db)
465			w.load_vts()
466			ret = w.process_vts(vts)
467			self.assertEqual(ret, vt_out)
468
469			def test_process_vts_bad_param_id(self, mock_nvti, mock_db):
470			vts = {
471			'1.3.6.1.4.1.25623.1.0.100061': {'3': 'new value'},
472			'vt_groups': ['family=debian', 'family=general'],
473			}
474			w = DummyDaemon(mock_nvti, mock_db)
475			w.load_vts()
476			ret = w.process_vts(vts)
477			self.assertFalse(ret[1])
478
479			@patch('logging.Logger.warning')
480			def test_process_vts_not_found(self, mock_logger, mock_nvti, mock_db):
481			vts = {
482			'1.3.6.1.4.1.25623.1.0.100065': {'3': 'new value'},
483			'vt_groups': ['family=debian', 'family=general'],
484			}
485			w = DummyDaemon(mock_nvti, mock_db)
486			w.load_vts()
487			ret = w.process_vts(vts)
488			self.assertTrue(mock_logger.called)
489
490			def test_get_openvas_timestamp_scan_host_end(self, mock_nvti, mock_db):
491			mock_db.get_host_scan_scan_end_time.return_value = '12345'
492			w = DummyDaemon(mock_nvti, mock_db)
493			targets = [['192.168.0.1', 'port', 'cred', 'exclude_host']]
494			w.create_scan('123-456', targets, None, [])
495			w.get_openvas_timestamp_scan_host('123-456', '192.168.0.1')
496			for result in w.scan_collection.results_iterator('123-456', False):
497			self.assertEqual(result.get('value'), '12345')
498
499			def test_get_openvas_timestamp_scan_host_start(self, mock_nvti, mock_db):
500			mock_db.get_host_scan_scan_end_time.return_value = None
501			mock_db.get_host_scan_scan_end_time.return_value = '54321'
502			w = DummyDaemon(mock_nvti, mock_db)
503			targets = [['192.168.0.1', 'port', 'cred', 'exclude_host']]
504			w.create_scan('123-456', targets, None, [])
505			w.get_openvas_timestamp_scan_host('123-456', '192.168.0.1')
506			for result in w.scan_collection.results_iterator('123-456', False):
507			self.assertEqual(result.get('value'), '54321')
508
509			def test_host_is_finished(self, mock_nvti, mock_db):
510			mock_db.get_single_item.return_value = 'finished'
511			w = DummyDaemon(mock_nvti, mock_db)
512			ret = w.host_is_finished('123-456')
513			self.assertEqual(ret, True)
514
515			def test_scan_is_stopped(self, mock_nvti, mock_db):
516			mock_db.get_single_item.return_value = 'stop_all'
517			mock_db.kb_connect_item.return_value = mock_db
518			mock_db.set_redisctx.return_value = None
519			w = DummyDaemon(mock_nvti, mock_db)
520			ret = w.scan_is_stopped('123-456')
521			self.assertEqual(ret, True)
522
523			@patch('ospd_openvas.daemon.open')
524			def test_feed_is_outdated_none(self, mock_open, mock_nvti, mock_db):
525			w = DummyDaemon(mock_nvti, mock_db)
526			# Mock parse_param, because feed_is_oudated() will call it.
527			with patch.object(w, 'parse_param', return_value=None):
528			# Return None
529			w.scan_only_params['plugins_folder'] = '/foo/bar'
530			ret = w.feed_is_outdated('1234')
531			self.assertIsNone(ret)
532
533			def test_feed_is_outdated_true(self, mock_nvti, mock_db):
534			w = DummyDaemon(mock_nvti, mock_db)
535			# Mock parse_param, because feed_is_oudated() will call it.
536			with patch.object(w, 'parse_param', return_value=None):
537			with patch.object(Path, 'exists', return_value=True):
538			read_data = 'PLUGIN_SET = "1235";'
539			with patch("builtins.open",
540			return_value=io.StringIO(read_data)):
541			# Return True
542			w.scan_only_params['plugins_folder'] = '/foo/bar'
543			ret = w.feed_is_outdated('1234')
544			self.assertTrue(ret)
545
546			def test_feed_is_outdated_false(self, mock_nvti, mock_db):
547			w = DummyDaemon(mock_nvti, mock_db)
548			# Mock parse_param, because feed_is_oudated() will call it.
549			with patch.object(w, 'parse_param', return_value=None):
550			read_data = 'PLUGIN_SET = "1234";'
551			with patch.object(Path, 'exists', return_value=True):
552			read_data = 'PLUGIN_SET = "1234"';
553			with patch("builtins.open",
554			return_value=io.StringIO(read_data)):
555			# Return True
556			w.scan_only_params['plugins_folder'] = '/foo/bar'
557			ret = w.feed_is_outdated('1234')
558			self.assertFalse(ret)
559
560			@patch('ospd_openvas.daemon.OSPDaemon.add_scan_log')
561			def test_get_openvas_result(self, mock_ospd, mock_nvti, mock_db):
562			results = ["LOG\|\|\| \|\|\|general/Host_Details\|\|\| \|\|\|Host dead", None]
563			mock_db.get_result.side_effect = results
564			w = DummyDaemon(mock_nvti, mock_db)
565			w.load_vts()
566			mock_ospd.return_value = None
567			w.get_openvas_result('123-456', 'localhost')
568			mock_ospd.assert_called_with(
569			'123-456',
570			host='localhost',
571			hostname=' ',
572			name='',
573			port='general/Host_Details',
574			qod='',
575			test_id=' ',
576			value='Host dead',
577			)
578
579			@patch('ospd_openvas.daemon.OSPDaemon.set_scan_target_progress')
580			def test_update_progress(self, mock_ospd, mock_nvti, mock_db):
581			msg = '0/-1'
582			targets = [['localhost', 'port', 'cred', 'exclude_host']]
583			w = DummyDaemon(mock_nvti, mock_db)
584			w.create_scan('123-456', targets, None, [])
585
586			mock_ospd.return_value = None
587			w.update_progress('123-456', 'localhost', 'localhost', msg)
588			mock_ospd.assert_called_with('123-456', 'localhost', 'localhost', 100)
589
590
591			class TestFilters(unittest.TestCase):
592			def test_format_vt_modification_time(self):
593			ovformat = OpenVasVtsFilter()
594			td = '$Date: 2018-02-01 02:09:01 +0200 (Thu, 18 Oct 2018) $'
595			formatted = ovformat.format_vt_modification_time(td)
596			self.assertEqual(formatted, "20180201020901")
597

greenbone / ospd-openvas

Push — master ( 524bc6...41b97c )

TestOspdOpenvas.test_process_vts_not_found() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like