ospd.ospd_ssh.OSPDaemonSimpleSSH.run_command() - Code Metrics - Inspection of "Add support for credentials in multi-target mode." - greenbone/ospd - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#41)

unknown

created 2018-09-03 06:58 UTC

ospd.ospd_ssh.OSPDaemonSimpleSSH.run_command() B

↳ Parent: ospd.ospd_ssh

Complexity

Conditions

Size

Total Lines	54
Code Lines	31

Duplication

Lines	54
Ratio	100 %

Importance

Changes

Metric	Value
cc	7
eloc	31
nop	4
dl	54
loc	54
rs	7.736
c	0
b	0
f	0

How to fix Long Method

# -*- coding: utf-8 -*-
# $Id$
# Description:
# OSP Daemon class for simple remote SSH-based command execution.
#
# Authors:
# Jan-Oliver Wagner <[email protected]>
#
# 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

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):


    """
    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):

        """
        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


1		# -- coding: utf-8 --
2		# $Id$
3		# Description:
4		# OSP Daemon class for simple remote SSH-based command execution.
5		#
6		# Authors:
7		# Jan-Oliver Wagner <[email protected]>
8		#
9		# Copyright:
10		# Copyright (C) 2015 Greenbone Networks GmbH
11		#
12		# This program is free software; you can redistribute it and/or
13		# modify it under the terms of the GNU General Public License
14		# as published by the Free Software Foundation; either version 2
15		# of the License, or (at your option) any later version.
16		#
17		# This program is distributed in the hope that it will be useful,
18		# but WITHOUT ANY WARRANTY; without even the implied warranty of
19		# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
20		# GNU General Public License for more details.
21		#
22		# You should have received a copy of the GNU General Public License
23		# along with this program; if not, write to the Free Software
24		# Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
25
26		""" OSP Daemon class for simple remote SSH-based command execution. """
27
28		# This is needed for older pythons as our current module is called the same
29		# as the ospd package
30		# Another solution would be to rename that file.
31		from __future__ import absolute_import
32
33		from ospd.ospd import OSPDaemon
34
35		import socket
		0 ignored issues – show introduced 2018-06-12 10:47 UTC by Report Bug Copy Issue Report standard import "import socket" should be placed before "from ospd.ospd import OSPDaemon" Loading history...
36		try:
37		import paramiko
38		except ImportError:
39		paramiko = None
40
41		SSH_SCANNER_PARAMS = {
42		'username_password': {
43		'type': 'credential_up',
44		'name': 'SSH credentials',
45		'default': '',
46		'mandatory': 0,
47		'description': 'The SSH credentials in username:password format. Used'
48		' to log into the target and to run the commands on'
49		' that target. This should not be a privileged user'
50		' like "root", a regular privileged user account'
51		' should be sufficient in most cases.',
52		},
53		'port': {
54		'type': 'integer',
55		'name': 'SSH Port',
56		'default': 22,
57		'mandatory': 0,
58		'description': 'The SSH port which to use for logging in with the'
59		' given username_password.',
60		},
61		'ssh_timeout': {
62		'type': 'integer',
63		'name': 'SSH timeout',
64		'default': 30,
65		'mandatory': 0,
66		'description': 'Timeout when communicating with the target via SSH.',
67		},
68		}
69
70
71	View Code Duplication	class OSPDaemonSimpleSSH(OSPDaemon):
		0 ignored issues – show Bug introduced 2018-06-12 10:47 UTC by Report Bug Copy Issue Report The method `check` which was declared abstract in the super-class `OSPDaemon` was not overridden. Methods which raise `NotImplementedError` should be overridden in concrete child classes. Loading history... Bug introduced 2018-06-12 10:47 UTC by Report Bug Copy Issue Report The method `exec_scan` which was declared abstract in the super-class `OSPDaemon` was not overridden. Methods which raise `NotImplementedError` should be overridden in concrete child classes. Loading history... Unused Code introduced 2018-06-12 10:47 UTC by Report Bug Copy Issue Report The variable `__class__` seems to be unused. Loading history... Duplication introduced 2018-06-18 10:10 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
72
73		"""
74		OSP Daemon class for simple remote SSH-based command execution.
75
76		This class automatically adds scanner parameters to handle remote
77		ssh login into the target systems: username, password, port and
78		ssh_timout
79
80		The method run_command can be used to execute a single command
81		on the given remote system. The stdout result is returned as
82		an array.
83		"""
84
85		def __init__(self, certfile, keyfile, cafile):
86		""" Initializes the daemon and add parameters needed to remote SSH execution. """
87
88		super(OSPDaemonSimpleSSH, self).__init__(certfile=certfile, keyfile=keyfile,
89		cafile=cafile)
90
91		if paramiko is None:
92		raise ImportError('paramiko needs to be installed in order to use'
93		' the %s class.' % self.__class__.__name__)
94
95		for name, param in SSH_SCANNER_PARAMS.items():
96		self.add_scanner_param(name, param)
97
98
99		def run_command(self, scan_id, host, cmd):
		0 ignored issues – show Comprehensibility introduced 2018-08-31 10:10 UTC by Report Bug Copy Issue Report This function exceeds the maximum number of variables (17/15). Loading history...
100		"""
101		Run a single command via SSH and return the content of stdout or
102		None in case of an Error. A scan error is issued in the latter
103		case.
104
105		For logging into 'host', the scan options 'port', 'username',
106		'password' and 'ssh_timeout' are used.
107		"""
108
109		ssh = paramiko.SSHClient()
110		ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
111
112		options = self.get_scan_options(scan_id)
113
114		port = int(options['port'])
115		timeout = int(options['ssh_timeout'])
116
117		# For backward compatibility, consider the legacy mode to get
118		# credentials as scan_option.
119		# First and second modes should be removed in future releases.
120		# On the third case it receives the credentials as a subelement of
121		# the <target>.
122		credentials = self.get_scan_credentials(scan_id, host)
123		if ('username_password' in options and
124		':' in options['username_password']):
125		username, password = options['username_password'].split(':', 1)
126		elif 'username' in options and options['username']:
127		username = options['username']
128		password = options['password']
129		elif credentials:
130		cred_params = credentials.get('ssh')
131		username = cred_params.get('username', '')
132		password = cred_params.get('password', '')
133		else:
134		self.add_scan_error(scan_id, host=host,
135		value='Erroneous username_password value')
136		raise ValueError('Erroneous username_password value')
137
138		try:
139		ssh.connect(hostname=host, username=username, password=password,
140		timeout=timeout, port=port)
141		except (paramiko.ssh_exception.AuthenticationException,
142		socket.error) as err:
143		# Errors: No route to host, connection timeout, authentication
144		# failure etc,.
145		self.add_scan_error(scan_id, host=host, value=str(err))
146		return None
147
148		_, stdout, _ = ssh.exec_command(cmd)
149		result = stdout.readlines()
150		ssh.close()
151
152		return result
153

greenbone / ospd

Pull Request — master (#41)

ospd.ospd_ssh.OSPDaemonSimpleSSH.run_command() B

Complexity

Size

Duplication

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like