tests.test_ssh_daemon

Complexity

Total Complexity

19

Size/Duplication

Total Lines	196
Duplicated Lines	22.45 %

Importance

Changes

0

17 Methods

Rating	Name	Duplication	Size	Complexity
A	FakeFile.readlines()	0	2	1
A	FakeSSHClient.connect()	0	2	1
A	fakeparamiko.SSHClient()	0	3	1
A	FakeSSHClient.__init__()	0	3	1
A	FakeSSHClient.exec_command()	0	3	1
A	FakeSSHClient.close()	0	2	1
A	FakeSSHClient.set_missing_host_key_policy()	0	2	1
A	FakeFile.__init__()	0	2	1
A	fakeparamiko.AutoAddPolicy()	0	3	1
A	DummyWrapper.check()	0	2	1
A	DummyWrapper.__init__()	0	5	1
A	SSHDaemonTestCase.test_run_command()	22	22	1
A	SSHDaemonTestCase.test_run_command_no_credential()	0	21	2
A	DummyWrapper.exec_scan()	0	2	1
A	SSHDaemonTestCase.test_run_command_legacy_credential()	22	22	1
A	SSHDaemonTestCase.test_no_paramiko()	0	5	2
A	SSHDaemonTestCase.test_run_command_new_credential()	0	33	1

# Copyright (C) 2014-2021 Greenbone Networks GmbH
#
# SPDX-License-Identifier: AGPL-3.0-or-later
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.

""" Test module for ospd ssh support.
"""

import unittest

from ospd import ospd_ssh
from ospd.ospd_ssh import OSPDaemonSimpleSSH
from .helper import FakeDataManager


class FakeFile(object):
    def __init__(self, content):
        self.content = content

    def readlines(self):
        return self.content.split('\n')


commands = None  # pylint: disable=invalid-name


class FakeSSHClient(object):
    def __init__(self):
        global commands  # pylint: disable=global-statement,invalid-name
        commands = []

    def set_missing_host_key_policy(self, policy):
        pass

    def connect(self, **kwargs):
        pass

    def exec_command(self, cmd):
        commands.append(cmd)
        return None, FakeFile(''), None

    def close(self):
        pass


class FakeExceptions(object):
    AuthenticationException = None  # pylint: disable=invalid-name


class fakeparamiko(object):  # pylint: disable=invalid-name
    @staticmethod
    def SSHClient(*args):  # pylint: disable=invalid-name
        return FakeSSHClient(*args)

    @staticmethod
    def AutoAddPolicy():  # pylint: disable=invalid-name
        pass

    ssh_exception = FakeExceptions


class DummyWrapper(OSPDaemonSimpleSSH):
    def __init__(self, niceness=10):
        super().__init__(niceness=niceness)
        self.scan_collection.data_manager = FakeDataManager()
        self.scan_collection.file_storage_dir = '/tmp'
        self.initialized = True

    def check(self):
        return True

    def exec_scan(self, scan_id: str):
        return


class SSHDaemonTestCase(unittest.TestCase):
    def test_no_paramiko(self):
        ospd_ssh.paramiko = None

        with self.assertRaises(ImportError):
            OSPDaemonSimpleSSH()

    def test_run_command(self):

This code seems to be duplicated in your project.

        ospd_ssh.paramiko = fakeparamiko

        daemon = DummyWrapper(niceness=10)
        scanid = daemon.create_scan(
            None,
            {
                'target': 'host.example.com',
                'ports': '80, 443',
                'credentials': {},
                'exclude_hosts': '',
                'finished_hosts': '',
                'options': {},
            },
            dict(port=5, ssh_timeout=15, username_password='dummy:pw'),
            '',
        )
        daemon.start_queued_scans()
        res = daemon.run_command(scanid, 'host.example.com', 'cat /etc/passwd')

        self.assertIsInstance(res, list)
        self.assertEqual(commands, ['nice -n 10 cat /etc/passwd'])

    def test_run_command_legacy_credential(self):

This code seems to be duplicated in your project.

        ospd_ssh.paramiko = fakeparamiko

        daemon = DummyWrapper(niceness=10)
        scanid = daemon.create_scan(
            None,
            {
                'target': 'host.example.com',
                'ports': '80, 443',
                'credentials': {},
                'exclude_hosts': '',
                'finished_hosts': '',
                'options': {},
            },
            dict(port=5, ssh_timeout=15, username='dummy', password='pw'),
            '',
        )
        daemon.start_queued_scans()
        res = daemon.run_command(scanid, 'host.example.com', 'cat /etc/passwd')

        self.assertIsInstance(res, list)
        self.assertEqual(commands, ['nice -n 10 cat /etc/passwd'])

    def test_run_command_new_credential(self):
        ospd_ssh.paramiko = fakeparamiko

        daemon = DummyWrapper(niceness=10)

        cred_dict = {
            'ssh': {
                'type': 'up',
                'password': 'mypass',
                'port': '22',
                'username': 'scanuser',
            },
            'smb': {'type': 'up', 'password': 'mypass', 'username': 'smbuser'},
        }

        scanid = daemon.create_scan(
            None,
            {
                'target': 'host.example.com',
                'ports': '80, 443',
                'credentials': cred_dict,
                'exclude_hosts': '',
                'finished_hosts': '',
                'options': {},
            },
            dict(port=5, ssh_timeout=15),
            '',
        )
        daemon.start_queued_scans()
        res = daemon.run_command(scanid, 'host.example.com', 'cat /etc/passwd')

        self.assertIsInstance(res, list)
        self.assertEqual(commands, ['nice -n 10 cat /etc/passwd'])

    def test_run_command_no_credential(self):
        ospd_ssh.paramiko = fakeparamiko

        daemon = DummyWrapper(niceness=10)
        scanid = daemon.create_scan(
            None,
            {
                'target': 'host.example.com',
                'ports': '80, 443',
                'credentials': {},
                'exclude_hosts': '',
                'finished_hosts': '',
                'options': {},
            },
            dict(port=5, ssh_timeout=15),
            '',
        )
        daemon.start_queued_scans()

        with self.assertRaises(ValueError):
            daemon.run_command(scanid, 'host.example.com', 'cat /etc/passwd')