ospd.ospd_ssh

Complexity

Total Complexity

10

Size/Duplication

Total Lines	153
Duplicated Lines	53.59 %

Importance

Changes

0

2 Methods

Rating	Name	Duplication	Size	Complexity
A	OSPDaemonSimpleSSH.__init__()	12	12	3
B	OSPDaemonSimpleSSH.run_command()	54	54	7

# -*- coding: utf-8 -*-
# $Id$
# Description:
# OSP Daemon class for simple remote SSH-based command execution.
#
# Authors:
# Jan-Oliver Wagner <jan-oliver.wagner@greenbone.net>
#
# Copyright:
# Copyright (C) 2015 Greenbone Networks GmbH
#
# 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.

""" OSP Daemon class for simple remote SSH-based command execution. """

# This is needed for older pythons as our current module is called the same
# as the ospd package
# Another solution would be to rename that file.
from __future__ import absolute_import

from ospd.ospd import OSPDaemon

import socket

standard import "import socket" should be placed before "from ospd.ospd import OSPDaemon"

try:
    import paramiko
except ImportError:
    paramiko = None

SSH_SCANNER_PARAMS = {
    'username_password': {
        'type': 'credential_up',
        'name': 'SSH credentials',
        'default': '',
        'mandatory': 0,
        'description': 'The SSH credentials in username:password format. Used'
                       ' to log into the target and to run the commands on'
                       ' that target. This should not be a privileged user'
                       ' like "root", a regular privileged user account'
                       ' should be sufficient in most cases.',
    },
    'port': {
        'type': 'integer',
        'name': 'SSH Port',
        'default': 22,
        'mandatory': 0,
        'description': 'The SSH port which to use for logging in with the'
                       ' given username_password.',
    },
    'ssh_timeout': {
        'type': 'integer',
        'name': 'SSH timeout',
        'default': 30,
        'mandatory': 0,
        'description': 'Timeout when communicating with the target via SSH.',
    },
}


class OSPDaemonSimpleSSH(OSPDaemon):

The method check which was declared abstract in the super-class OSPDaemon
was not overridden.

The method exec_scan which was declared abstract in the super-class OSPDaemon
was not overridden.

The variable __class__ seems to be unused.

This code seems to be duplicated in your project.

    """
    OSP Daemon class for simple remote SSH-based command execution.

    This class automatically adds scanner parameters to handle remote
    ssh login into the target systems: username, password, port and
    ssh_timout

    The method run_command can be used to execute a single command
    on the given remote system. The stdout result is returned as
    an array.
    """

    def __init__(self, certfile, keyfile, cafile):
        """ Initializes the daemon and add parameters needed to remote SSH execution. """

        super(OSPDaemonSimpleSSH, self).__init__(certfile=certfile, keyfile=keyfile,
                                                 cafile=cafile)

        if paramiko is None:
            raise ImportError('paramiko needs to be installed in order to use'
                              ' the %s class.' % self.__class__.__name__)

        for name, param in SSH_SCANNER_PARAMS.items():
            self.add_scanner_param(name, param)


    def run_command(self, scan_id, host, cmd):

This function exceeds the maximum number of variables (17/15).

        """
        Run a single command via SSH and return the content of stdout or
        None in case of an Error. A scan error is issued in the latter
        case.

        For logging into 'host', the scan options 'port', 'username',
        'password' and 'ssh_timeout' are used.
        """

        ssh = paramiko.SSHClient()
        ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())

        options = self.get_scan_options(scan_id)

        port = int(options['port'])
        timeout = int(options['ssh_timeout'])

        # For backward compatibility, consider the legacy mode to get
        # credentials as scan_option.
        # First and second modes should be removed in future releases.
        # On the third case it receives the credentials as a subelement of
        # the <target>.
        credentials = self.get_scan_credentials(scan_id, host)
        if ('username_password' in options and
                ':' in options['username_password']):
            username, password = options['username_password'].split(':', 1)
        elif 'username' in options and options['username']:
            username = options['username']
            password = options['password']
        elif credentials:
            cred_params = credentials.get('ssh')
            username = cred_params.get('username', '')
            password = cred_params.get('password', '')
        else:
            self.add_scan_error(scan_id, host=host,
                                value='Erroneous username_password value')
            raise ValueError('Erroneous username_password value')

        try:
            ssh.connect(hostname=host, username=username, password=password,
                        timeout=timeout, port=port)
        except (paramiko.ssh_exception.AuthenticationException,
                socket.error) as err:
            # Errors: No route to host, connection timeout, authentication
            # failure etc,.
            self.add_scan_error(scan_id, host=host, value=str(err))
            return None

        _, stdout, _ = ssh.exec_command(cmd)
        result = stdout.readlines()
        ssh.close()

        return result