# -*- coding: utf-8 -*-
# SPDX-FileCopyrightText: 2014-2023 Greenbone AG
#
# SPDX-License-Identifier: AGPL-3.0-or-later


# pylint: disable=invalid-name,line-too-long,no-value-for-parameter

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

import io
import logging
from pathlib import Path

from unittest import TestCase
from unittest.mock import patch, Mock, MagicMock

from ospd.protocol import OspRequest

from tests.dummydaemon import DummyDaemon
from tests.helper import assert_called_once

from ospd_openvas.daemon import (
    OSPD_PARAMS,
    OpenVasVtsFilter,
)
from ospd_openvas.openvas import Openvas
from ospd_openvas.notus import Notus, hashsum_verificator

OSPD_PARAMS_OUT = {
    'auto_enable_dependencies': {
        'type': 'boolean',
        'name': 'auto_enable_dependencies',
        'default': 1,
        'mandatory': 1,
        'visible_for_client': True,
        'description': 'Automatically enable the plugins that are depended on',
    },
    'cgi_path': {
        'type': 'string',
        'name': 'cgi_path',
        'default': '/cgi-bin:/scripts',
        'mandatory': 1,
        'visible_for_client': True,
        'description': 'Look for default CGIs in /cgi-bin and /scripts',
    },
    'checks_read_timeout': {
        'type': 'integer',
        'name': 'checks_read_timeout',
        'default': 5,
        'mandatory': 1,
        'visible_for_client': True,
        'description': (
            'Number  of seconds that the security checks will '
            + 'wait for when doing a recv()'
        ),
    },
    'non_simult_ports': {
        'type': 'string',
        'name': 'non_simult_ports',
        'default': '139, 445, 3389, Services/irc',
        'mandatory': 1,
        'visible_for_client': True,
        '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,
        'visible_for_client': True,
        '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,
        'visible_for_client': True,
        'description': (
            'Number of retries when a socket connection attempt ' + 'timesout.'
        ),
    },
    'optimize_test': {
        'type': 'boolean',
        'name': 'optimize_test',
        'default': 1,
        'mandatory': 0,
        'visible_for_client': True,
        'description': (
            'By default, optimize_test is enabled which means openvas does '
            + 'trust the remote host banners and is only launching plugins '
            + 'against the services they have been designed to check. '
            + 'For example it will check a web server claiming to be IIS only '
            + 'for IIS related flaws but will skip plugins testing for Apache '
            + 'flaws, and so on. This default behavior is used to optimize '
            + 'the scanning performance and to avoid false positives. '
            + 'If you are not sure that the banners of the remote host '
            + 'have been tampered with, you can disable this option.'
        ),
    },
    'plugins_timeout': {
        'type': 'integer',
        'name': 'plugins_timeout',
        'default': 5,
        'mandatory': 0,
        'visible_for_client': True,
        'description': 'This is the maximum lifetime, in seconds of a plugin.',
    },
    'report_host_details': {
        'type': 'boolean',
        'name': 'report_host_details',
        'default': 1,
        'mandatory': 1,
        'visible_for_client': True,
        'description': '',
    },
    'safe_checks': {
        'type': 'boolean',
        'name': 'safe_checks',
        'default': 1,
        'mandatory': 1,
        'visible_for_client': True,
        'description': (
            'Disable the plugins with potential to crash '
            + 'the remote services'
        ),
    },
    'scanner_plugins_timeout': {
        'type': 'integer',
        'name': 'scanner_plugins_timeout',
        'default': 36000,
        'mandatory': 1,
        'visible_for_client': True,
        'description': 'Like plugins_timeout, but for ACT_SCANNER plugins.',
    },
    'time_between_request': {
        'type': 'integer',
        'name': 'time_between_request',
        'default': 0,
        'mandatory': 0,
        'visible_for_client': True,
        'description': (
            'Allow to set a wait time between two actions '
            + '(open, send, close).'
        ),
    },
    'unscanned_closed': {
        'type': 'boolean',
        'name': 'unscanned_closed',
        'default': 1,
        'mandatory': 1,
        'visible_for_client': True,
        'description': '',
    },
    'unscanned_closed_udp': {
        'type': 'boolean',
        'name': 'unscanned_closed_udp',
        'default': 1,
        'mandatory': 1,
        'visible_for_client': True,
        'description': '',
    },
    'expand_vhosts': {
        'type': 'boolean',
        'name': 'expand_vhosts',
        'default': 1,
        'mandatory': 0,
        'visible_for_client': True,
        'description': 'Whether to expand the target hosts '
        + 'list of vhosts with values gathered from sources '
        + 'such as reverse-lookup queries and VT checks '
        + 'for SSL/TLS certificates.',
    },
    'test_empty_vhost': {
        'type': 'boolean',
        'name': 'test_empty_vhost',
        'default': 0,
        'mandatory': 0,
        'visible_for_client': True,
        'description': 'If  set  to  yes, the scanner will '
        + 'also test the target by using empty vhost value '
        + 'in addition to the targets associated vhost values.',
    },
    'max_hosts': {
        'type': 'integer',
        'name': 'max_hosts',
        'default': 30,
        'mandatory': 0,
        'visible_for_client': False,
        'description': (
            'The maximum number of hosts to test at the same time which '
            + 'should be given to the client (which can override it). '
            + 'This value must be computed given your bandwidth, '
            + 'the number of hosts you want to test, your amount of '
            + 'memory and the performance of your processor(s).'
        ),
    },
    'max_checks': {
        'type': 'integer',
        'name': 'max_checks',
        'default': 10,
        'mandatory': 0,
        'visible_for_client': False,
        'description': (
            'The number of plugins that will run against each host being '
            + 'tested. Note that the total number of process will be max '
            + 'checks x max_hosts so you need to find a balance between '
            + 'these two options. Note that launching too many plugins at '
            + 'the same time may disable the remote host, either temporarily '
            + '(ie: inetd closes its ports) or definitely (the remote host '
            + 'crash because it is asked to do too many things at the '
            + 'same time), so be careful.'
        ),
    },
    'port_range': {
        'type': 'string',
        'name': 'port_range',
        'default': '',
        'mandatory': 0,
        'visible_for_client': False,
        'description': (
            'This is the default range of ports that the scanner plugins will '
            + 'probe. The syntax of this option is flexible, it can be a '
            + 'single range ("1-1500"), several ports ("21,23,80"), several '
            + 'ranges of ports ("1-1500,32000-33000"). Note that you can '
            + 'specify UDP and TCP ports by prefixing each range by T or U. '
            + 'For instance, the following range will make openvas scan UDP '
            + 'ports 1 to 1024 and TCP ports 1 to 65535 : '
            + '"T:1-65535,U:1-1024".'
        ),
    },
    'alive_test_ports': {
        'type': 'string',
        'name': 'alive_test_ports',
        'default': '21-23,25,53,80,110-111,135,139,143,443,445,'
        + '993,995,1723,3306,3389,5900,8080',
        'mandatory': 0,
        'visible_for_client': True,
        'description': ('Port list used for host alive detection.'),
    },
    'test_alive_hosts_only': {
        'type': 'boolean',
        'name': 'test_alive_hosts_only',
        'default': 0,
        'mandatory': 0,
        'visible_for_client': False,
        'description': (
            'If this option is set, openvas will scan the target list for '
            + 'alive hosts in a separate process while only testing those '
            + 'hosts which are identified as alive. This boosts the scan '
            + 'speed of target ranges with a high amount of dead hosts '
            + 'significantly.'
        ),
    },
    'test_alive_wait_timeout': {
        'type': 'integer',
        'name': 'test_alive_wait_timeout',
        'default': 1,
        'mandatory': 0,
        'visible_for_client': True,
        'description': (
            'This is the default timeout to wait for replies after last '
            + 'packet was sent.'
        ),
    },
    'icmp_retries': {
        'type': 'integer',
        'name': 'icmp_retries',
        'default': 1,
        'mandatory': 0,
        'visible_for_client': True,
        'description': (
            'This is the default amount of icmp packets that will be '
            + 'sent to the host target during an alive test.'
        ),
    },
    'icmp_grace_period': {
        'type': 'integer',
        'name': 'icmp_grace_period',
        'default': 0,
        'mandatory': 0,
        'visible_for_client': True,
        'description': (
            'Wait time between icmp packets during alive tests. '
            + 'Useful for sensitive targets. It can slow '
            + 'down the alive test.'
        ),
    },
    'hosts_allow': {
        'type': 'string',
        'name': 'hosts_allow',
        'default': '',
        'mandatory': 0,
        'visible_for_client': False,
        'description': (
            'Comma-separated list of the only targets that are authorized '
            + 'to be scanned. Supports the same syntax as the list targets. '
            + 'Both target hostnames and the address to which they resolve '
            + 'are checked. Hostnames in hosts_allow list are not resolved '
            + 'however.'
        ),
    },
    'hosts_deny': {
        'type': 'string',
        'name': 'hosts_deny',
        'default': '',
        'mandatory': 0,
        'visible_for_client': False,
        'description': (
            'Comma-separated list of targets that are not authorized to '
            + 'be scanned. Supports the same syntax as the list targets. '
            + 'Both target hostnames and the address to which they resolve '
            + 'are checked. Hostnames in hosts_deny list are not '
            + 'resolved however.'
        ),
    },
    'results_per_host': {
        'type': 'integer',
        'name': 'results_per_host',
        'default': 10,
        'mandatory': 0,
        'visible_for_client': True,
        'description': (
            'Amount of fake results generated per each host in the target '
            + 'list for a dry run scan.'
        ),
    },
    'table_driven_lsc': {
        'type': 'boolean',
        'name': 'table_driven_lsc',
        'default': 1,
        'mandatory': 0,
        'visible_for_client': True,
        'description': (
            'If this option is enabled a scanner for table_driven_lsc will '
            + 'scan package results.'
        ),
    },
    'max_mem_kb': {
        'type': 'integer',
        'name': 'max_mem_kb',
        'default': 0,
        'mandatory': 0,
        'visible_for_client': True,
        'description': (
            'Maximum amount of memory (in MB) allowed to use for a single '
            + 'script. If this value is set, the amount of memory put into '
            + 'redis is tracked for every Script. If the amount of memory '
            + 'exceeds this limit, the script is not able to set more kb '
            + 'items. The tracked the value written into redis is only '
            + 'estimated, as it does not check, if a value was replaced or '
            + 'appended. The size of the key is also not tracked. If this '
            + 'value is not set or <= 0, the maximum amount is unlimited '
            + '(Default).'
        ),
    },
}


class TestOspdOpenvas(TestCase):
    def test_return_disabled_verifier(self):
        verifier = hashsum_verificator(Path('/tmp'), True)
        self.assertEqual(verifier(Path('/tmp')), True)

    @patch('ospd_openvas.daemon.Openvas')
    def test_set_params_from_openvas_settings(self, mock_openvas: Openvas):
        mock_openvas.get_settings.return_value = {
            'non_simult_ports': '139, 445, 3389, Services/irc',
            'plugins_folder': '/foo/bar',
        }
        w = DummyDaemon()
        w.set_params_from_openvas_settings()

        self.assertEqual(mock_openvas.get_settings.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.Openvas')
    def test_sudo_available(self, mock_openvas):
        mock_openvas.check_sudo.return_value = True

        w = DummyDaemon()
        w._sudo_available = None  # pylint: disable=protected-access
        w._is_running_as_root = False  # pylint: disable=protected-access

        self.assertTrue(w.sudo_available)

    def test_update_vts(self):
        daemon = DummyDaemon()
        daemon.notus = MagicMock(spec=Notus)
        daemon.update_vts()
        self.assertEqual(daemon.notus.reload_cache.call_count, 1)

    @patch('ospd_openvas.daemon.Path.exists')
    @patch('ospd_openvas.daemon.Path.open')
    def test_get_feed_info(
        self,
        mock_path_open: MagicMock,
        mock_path_exists: MagicMock,
    ):
        read_data = 'PLUGIN_SET = "1235";'

        mock_path_exists.return_value = True
        mock_read = MagicMock(name='Path open context manager')
        mock_read.__enter__ = MagicMock(return_value=io.StringIO(read_data))
        mock_path_open.return_value = mock_read

        w = DummyDaemon()

        # Return True
        w.scan_only_params['plugins_folder'] = '/foo/bar'

        ret = w.get_feed_info()
        self.assertEqual(ret, {"PLUGIN_SET": "1235"})

        self.assertEqual(mock_path_exists.call_count, 1)
        self.assertEqual(mock_path_open.call_count, 1)

    @patch('ospd_openvas.daemon.Path.exists')
    @patch('ospd_openvas.daemon.OSPDopenvas.set_params_from_openvas_settings')
    def test_get_feed_info_none(
        self, mock_set_params: MagicMock, mock_path_exists: MagicMock
    ):
        w = DummyDaemon()

        w.scan_only_params['plugins_folder'] = '/foo/bar'

        # Return None
        mock_path_exists.return_value = False

        ret = w.get_feed_info()
        self.assertEqual(ret, {})

        self.assertEqual(mock_set_params.call_count, 1)
        self.assertEqual(mock_path_exists.call_count, 1)

    @patch('ospd_openvas.daemon.Path.exists')
    @patch('ospd_openvas.daemon.Path.open')
    def test_feed_is_outdated_true(
        self,
        mock_path_open: MagicMock,
        mock_path_exists: MagicMock,
    ):
        read_data = 'PLUGIN_SET = "1235";'

        mock_path_exists.return_value = True
        mock_read = MagicMock(name='Path open context manager')
        mock_read.__enter__ = MagicMock(return_value=io.StringIO(read_data))
        mock_path_open.return_value = mock_read

        w = DummyDaemon()

        # Return True
        w.scan_only_params['plugins_folder'] = '/foo/bar'

        ret = w.feed_is_outdated('1234')
        self.assertTrue(ret)

        self.assertEqual(mock_path_exists.call_count, 1)
        self.assertEqual(mock_path_open.call_count, 1)

    @patch('ospd_openvas.daemon.Path.exists')
    @patch('ospd_openvas.daemon.Path.open')
    def test_feed_is_outdated_false(
        self,
        mock_path_open: MagicMock,
        mock_path_exists: MagicMock,
    ):
        mock_path_exists.return_value = True

        read_data = 'PLUGIN_SET = "1234"'
        mock_path_exists.return_value = True
        mock_read = MagicMock(name='Path open context manager')
        mock_read.__enter__ = MagicMock(return_value=io.StringIO(read_data))
        mock_path_open.return_value = mock_read

        w = DummyDaemon()
        w.scan_only_params['plugins_folder'] = '/foo/bar'

        ret = w.feed_is_outdated('1234')
        self.assertFalse(ret)

        self.assertEqual(mock_path_exists.call_count, 1)
        self.assertEqual(mock_path_open.call_count, 1)

    def test_check_feed_cache_unavailable(self):
        w = DummyDaemon()
        w.vts.is_cache_available = False
        w.feed_is_outdated = Mock()

        w.feed_is_outdated.assert_not_called()

    @patch('ospd_openvas.daemon.BaseDB')
    @patch('ospd_openvas.daemon.ResultList.add_scan_log_to_list')
    def test_get_openvas_result(self, mock_add_scan_log_to_list, MockDBClass):
        w = DummyDaemon()

        target_element = w.create_xml_target()
        targets = OspRequest.process_target_element(target_element)
        w.create_scan('123-456', targets, None, [])

        results = [
            "LOG|||192.168.0.1|||localhost|||general/Host_Details||||||Host"
            " dead",
        ]
        MockDBClass.get_result.return_value = results
        mock_add_scan_log_to_list.return_value = None

        w.report_openvas_results(MockDBClass, '123-456')
        mock_add_scan_log_to_list.assert_called_with(
            host='192.168.0.1',
            hostname='localhost',
            name='',
            port='general/Host_Details',
            qod='',
            test_id='',
            uri='',
            value='Host dead',
        )

    @patch('ospd_openvas.daemon.BaseDB')
    @patch('ospd_openvas.daemon.ResultList.add_scan_error_to_list')
    def test_get_openvas_result_host_deny(
        self, mock_add_scan_error_to_list, MockDBClass
    ):
        w = DummyDaemon()

        target_element = w.create_xml_target()
        targets = OspRequest.process_target_element(target_element)
        w.create_scan('123-456', targets, None, [])

        results = [
            "ERRMSG|||127.0.0.1|||localhost|||||||||Host access denied.",
        ]
        MockDBClass.get_result.return_value = results
        mock_add_scan_error_to_list.return_value = None

        w.report_openvas_results(MockDBClass, '123-456')
        mock_add_scan_error_to_list.assert_called_with(
            host='127.0.0.1',
            hostname='localhost',
            name='',
            port='',
            test_id='',
            uri='',
            value='Host access denied.',
        )

    @patch('ospd_openvas.daemon.BaseDB')
    def test_get_openvas_result_dead_hosts(self, MockDBClass):
        w = DummyDaemon()
        target_element = w.create_xml_target()
        targets = OspRequest.process_target_element(target_element)
        w.create_scan('123-456', targets, None, [])

        results = [
            "DEADHOST||| ||| ||| ||| |||4",
        ]
        MockDBClass.get_result.return_value = results
        w.scan_collection.set_amount_dead_hosts = MagicMock()

        w.report_openvas_results(MockDBClass, '123-456')
        w.scan_collection.set_amount_dead_hosts.assert_called_with(
            '123-456',
            total_dead=4,
        )

    @patch('ospd_openvas.daemon.BaseDB')
    @patch('ospd_openvas.daemon.ResultList.add_scan_log_to_list')
    def test_get_openvas_result_host_start(
        self, mock_add_scan_log_to_list, MockDBClass
    ):
        w = DummyDaemon()
        target_element = w.create_xml_target()
        targets = OspRequest.process_target_element(target_element)
        w.create_scan('123-456', targets, None, [])

        results = [
            "HOST_START|||192.168.10.124||| ||| ||||||today 1",
        ]

        MockDBClass.get_result.return_value = results
        mock_add_scan_log_to_list.return_value = None

        w.report_openvas_results(MockDBClass, '123-456')

        mock_add_scan_log_to_list.assert_called_with(
            host='192.168.10.124',
            name='HOST_START',
            value='today 1',
        )

    @patch('ospd_openvas.daemon.BaseDB')
    def test_get_openvas_result_hosts_count(self, MockDBClass):
        w = DummyDaemon()
        target_element = w.create_xml_target()
        targets = OspRequest.process_target_element(target_element)
        w.create_scan('123-456', targets, None, [])

        results = [
            "HOSTS_COUNT||| ||| ||| ||| |||4",
        ]
        MockDBClass.get_result.return_value = results
        w.set_scan_total_hosts = MagicMock()

        w.report_openvas_results(MockDBClass, '123-456')
        w.set_scan_total_hosts.assert_called_with(
            '123-456',
            4,
        )

    @patch('ospd_openvas.daemon.BaseDB')
    @patch('ospd_openvas.daemon.ResultList.add_scan_alarm_to_list')
    def test_result_without_vt_oid(
        self, mock_add_scan_alarm_to_list, MockDBClass
    ):
        w = DummyDaemon()
        logging.Logger.warning = Mock()

        target_element = w.create_xml_target()
        targets = OspRequest.process_target_element(target_element)
        w.create_scan('123-456', targets, None, [])
        w.scan_collection.scans_table['123-456']['results'] = list()
        results = ["ALARM||| ||| ||| ||| |||some alarm|||path", None]
        MockDBClass.get_result.return_value = results
        mock_add_scan_alarm_to_list.return_value = None

        w.report_openvas_results(MockDBClass, '123-456')

        assert_called_once(logging.Logger.warning)

    @patch('psutil.Popen')
    def test_openvas_is_alive_already_stopped(self, mock_process):
        w = DummyDaemon()

        mock_process.is_running.return_value = True
        ret = w.is_openvas_process_alive(mock_process)
        self.assertTrue(ret)

    @patch('psutil.Popen')
    def test_openvas_is_alive_still(self, mock_process):
        w = DummyDaemon()

        mock_process.is_running.return_value = False
        ret = w.is_openvas_process_alive(mock_process)
        self.assertFalse(ret)

    @patch('ospd_openvas.daemon.OSPDaemon.set_scan_progress_batch')
    @patch('ospd_openvas.daemon.OSPDaemon.sort_host_finished')
    @patch('ospd_openvas.db.KbDB')
    def test_report_openvas_scan_status(
        self, mock_db, mock_sort_host_finished, mock_set_scan_progress_batch
    ):
        w = DummyDaemon()

        mock_set_scan_progress_batch.return_value = None
        mock_sort_host_finished.return_value = None
        mock_db.get_scan_status.return_value = [
            '192.168.0.1/15/1000',
            '192.168.0.2/15/0',
            '192.168.0.3/15/-1',
            '192.168.0.4/1500/1500',
        ]

        target_element = w.create_xml_target()
        targets = OspRequest.process_target_element(target_element)

        w.create_scan('123-456', targets, None, [])
        w.report_openvas_scan_status(mock_db, '123-456')

        mock_set_scan_progress_batch.assert_called_with(
            '123-456',
            host_progress={
                '192.168.0.1': 1,
                '192.168.0.3': -1,
                '192.168.0.4': 100,
            },
        )

        mock_sort_host_finished.assert_called_with(
            '123-456', ['192.168.0.3', '192.168.0.4']
        )


class TestFilters(TestCase):
    def test_format_vt_modification_time(self):
        ovformat = OpenVasVtsFilter(None, None)
        td = '1517443741'
        formatted = ovformat.format_vt_modification_time(td)
        self.assertEqual(formatted, "20180201000901")

    def test_get_filtered_vts_false(self):
        w = DummyDaemon()
        vts_collection = ['1234', '1.3.6.1.4.1.25623.1.0.100061']

        ovfilter = OpenVasVtsFilter(w.nvti, None)
        res = ovfilter.get_filtered_vts_list(
            vts_collection, "modification_time<10"
        )
        self.assertNotIn('1.3.6.1.4.1.25623.1.0.100061', res)

    def test_get_filtered_vts_true(self):
        w = DummyDaemon()
        vts_collection = ['1234', '1.3.6.1.4.1.25623.1.0.100061']

        ovfilter = OpenVasVtsFilter(w.nvti, None)
        res = ovfilter.get_filtered_vts_list(
            vts_collection, "modification_time>10"
        )
        self.assertIn('1.3.6.1.4.1.25623.1.0.100061', res)