gvm.connections.GvmConnection.disconnect() - Code Metrics - Inspection of "Merge pull request #43 from bjoernricks/connection..." - greenbone/python-gvm - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( f97307...dda958 )

by Juan José

created 2018-11-12 08:28 UTC

gvm.connections.GvmConnection.disconnect() A

↳ Parent: gvm.connections

Complexity

Conditions

Size

Total Lines	8
Code Lines	6

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	3
eloc	6
nop	1
dl	0
loc	8
rs	10
c	0
b	0
f	0

# -*- coding: utf-8 -*-
# Copyright (C) 2018 Greenbone Networks GmbH
#
# SPDX-License-Identifier: GPL-3.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 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 General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.
"""
Module for connections to GVM server daemons like gvmd and ospd.
"""
import logging
import socket as socketlib
import ssl
import time

import paramiko

from lxml import etree

from gvm.errors import GvmError


logger = logging.getLogger(__name__)

BUF_SIZE = 1024
DEFAULT_READ_TIMEOUT = 60 # in seconds
DEFAULT_TIMEOUT = 60 # in seconds
DEFAULT_GVM_PORT = 9390
DEFAULT_UNIX_SOCKET_PATH = '/usr/local/var/run/gvmd.sock'
MAX_SSH_DATA_LENGTH = 4095

class XmlReader:
    """
    Read a XML command until its closing element
    """

    def _start_xml(self):
        self._first_element = None
        self._parser = etree.XMLPullParser(('start', 'end'))

    def _is_end_xml(self):
        for action, obj in self._parser.read_events():
            if not self._first_element and action in 'start':
                self._first_element = obj.tag

            if self._first_element and action in 'end' and \
                    str(self._first_element) == str(obj.tag):
                return True
        return False

    def _feed_xml(self, data):
        try:
            self._parser.feed(data)
        except etree.ParseError as e:
            raise GvmError("Can't parse xml response. Response data "
                           "read {0}".format(data), e)


class GvmConnection:
    """
    Base class for establishing a connection to a remote server daemon.
    """

    def __init__(self, timeout=DEFAULT_TIMEOUT):
        """
          Arguments:
            socket -- A socket
        """
        self._socket = None
        self._timeout = timeout

    def connect(self):
        """Establish a connection to gvmd
        """
        raise NotImplementedError

    def send(self, data):
        """Send data to gvmd
        """
        if isinstance(data, str):
            self._socket.send(data.encode())
        else:
            self._socket.send(data)

    def read(self):
        """Read data from gvmd
        """
        raise NotImplementedError

    def disconnect(self):
        """Close the connection to gvmd
        """
        try:
            if self._socket is not None:
                self._socket.close()
        except OSError as e:
            logger.debug('Connection closing error: %s', e)


class SSHConnection(GvmConnection, XmlReader):
    """
    SSH Class to connect, read and write from GVM via SSH
    """

    def __init__(self, timeout=DEFAULT_TIMEOUT, hostname='127.0.0.1', port=22,
                 username='gmp', password=''):
        super().__init__(timeout=timeout)

        self.hostname = hostname
        self.port = int(port)
        self.username = username
        self.password = password

    def _send_in_chunks(self, data, chunk_size):
        i_start = 0
        i_end = chunk_size
        sent_bytes = 0
        length = len(data)

        while sent_bytes < length:
            time.sleep(0.01)

            self._stdin.channel.send(data[i_start:i_end])

            i_start = i_end
            if i_end > length:
                i_end = length
            else:
                i_end = i_end + chunk_size

            sent_bytes += (i_end - i_start)

        return sent_bytes

    def connect(self):
        self._socket = paramiko.SSHClient()
        self._socket.set_missing_host_key_policy(paramiko.AutoAddPolicy())

        try:
            self._socket.connect(
                hostname=self.hostname,
                username=self.username,
                password=self.password,
                timeout=self._timeout,
                port=int(self.port),
                allow_agent=False,
                look_for_keys=False)
            self._stdin, self._stdout, self._stderr = self._socket.exec_command(
                "", get_pty=False)

        except (paramiko.BadHostKeyException,
                paramiko.AuthenticationException,
                paramiko.SSHException, OSError) as e:
            logger.debug('SSH Connection failed: %s', e)
            raise

    def read(self):
        response = ''

        self._start_xml()

        while True:
            data = self._stdout.channel.recv(BUF_SIZE)
            # Connection was closed by server
            if not data:
                break

            self._feed_xml(data)

            response += data.decode('utf-8', errors='ignore')

            if self._is_end_xml():
                break

        return response

    def send(self, data):
        logger.debug('SSH:send(): %s', data)
        if len(data) > MAX_SSH_DATA_LENGTH:
            sent_bytes = self._send_in_chunks(data, MAX_SSH_DATA_LENGTH)
            logger.debug("SSH: %s bytes sent.", sent_bytes)
        else:
            self._stdin.channel.send(data)


class TLSConnection(GvmConnection):
    """
    TLS class to connect, read and write from a remote GVM daemon via TLS
    secured socket.
    """

    def __init__(self, certfile=None, cafile=None, keyfile=None,
                 hostname='127.0.0.1', port=DEFAULT_GVM_PORT,
                 timeout=DEFAULT_TIMEOUT):
        super().__init__(timeout=timeout)

        self.hostname = hostname
        self.port = port
        self.certfile = certfile
        self.cafile = cafile
        self.keyfile = keyfile

    def _new_socket(self):
        if self.certfile and self.cafile and self.keyfile:
            context = ssl.create_default_context(ssl.Purpose.SERVER_AUTH,
                                                 cafile=self.cafile)
            context.check_hostname = False
            context.load_cert_chain(
                certfile=self.certfile, keyfile=self.keyfile)
            new_socket = socketlib.socket(socketlib.AF_INET,
                                          socketlib.SOCK_STREAM)
            sock = context.wrap_socket(new_socket, server_side=False)
        else:
            context = ssl.SSLContext(ssl.PROTOCOL_TLSv1_2)
            sock = context.wrap_socket(socketlib.socket(socketlib.AF_INET))
        return sock


    def connect(self):
        self._socket = self._new_socket()
        self._socket.settimeout(self._timeout)
        self._socket.connect((self.hostname, int(self.port)))

    def read(self):
        response = ''

        while True:
            data = self._socket.read(BUF_SIZE)

            response += data.decode('utf-8', errors='ignore')
            if len(data) < BUF_SIZE:
                break

        return response


class UnixSocketConnection(GvmConnection, XmlReader):
    """
    UNIX-Socket class to connect, read, write from a GVM server daemon via
    direct communicating UNIX-Socket
    """

    def __init__(self, path=DEFAULT_UNIX_SOCKET_PATH, timeout=DEFAULT_TIMEOUT,
                 read_timeout=DEFAULT_READ_TIMEOUT):
        super().__init__(timeout=timeout)

        self.read_timeout = read_timeout
        self.path = path

    def connect(self):
        """Connect to the UNIX socket
        """
        self._socket = socketlib.socket(
            socketlib.AF_UNIX, socketlib.SOCK_STREAM)
        self._socket.settimeout(self._timeout)
        self._socket.connect(self.path)

    def read(self):
        """Read from the UNIX socket
        """
        response = ''

        break_timeout = time.time() + self.read_timeout
        old_timeout = self._socket.gettimeout()
        self._socket.settimeout(5)  # in seconds

        self._start_xml()

        while time.time() < break_timeout:
            data = b''

            try:
                data = self._socket.recv(BUF_SIZE)
            except (socketlib.timeout) as exception:
                logger.debug('Warning: No data received '
                             'from server: %s', exception)
                continue

            self._feed_xml(data)

            response += data.decode('utf-8', errors='ignore')

            if len(data) < BUF_SIZE:
                if self._is_end_xml():
                    break

        self._socket.settimeout(old_timeout)
        return response


1			# -- coding: utf-8 --
2			# Copyright (C) 2018 Greenbone Networks GmbH
3			#
4			# SPDX-License-Identifier: GPL-3.0-or-later
5			#
6			# This program is free software: you can redistribute it and/or modify
7			# it under the terms of the GNU General Public License as published by
8			# the Free Software Foundation, either version 3 of the License, or
9			# (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, see <http://www.gnu.org/licenses/>.
18			"""
19			Module for connections to GVM server daemons like gvmd and ospd.
20			"""
21			import logging
22			import socket as socketlib
23			import ssl
24			import time
25
26			import paramiko
27
28			from lxml import etree
29
30			from gvm.errors import GvmError
31
32
33			logger = logging.getLogger(__name__)
34
35			BUF_SIZE = 1024
36			DEFAULT_READ_TIMEOUT = 60 # in seconds
37			DEFAULT_TIMEOUT = 60 # in seconds
38			DEFAULT_GVM_PORT = 9390
39			DEFAULT_UNIX_SOCKET_PATH = '/usr/local/var/run/gvmd.sock'
40			MAX_SSH_DATA_LENGTH = 4095
41
42			class XmlReader:
43			"""
44			Read a XML command until its closing element
45			"""
46
47			def _start_xml(self):
48			self._first_element = None
49			self._parser = etree.XMLPullParser(('start', 'end'))
50
51			def _is_end_xml(self):
52			for action, obj in self._parser.read_events():
53			if not self._first_element and action in 'start':
54			self._first_element = obj.tag
55
56			if self._first_element and action in 'end' and \
57			str(self._first_element) == str(obj.tag):
58			return True
59			return False
60
61			def _feed_xml(self, data):
62			try:
63			self._parser.feed(data)
64			except etree.ParseError as e:
65			raise GvmError("Can't parse xml response. Response data "
66			"read {0}".format(data), e)
67
68
69			class GvmConnection:
70			"""
71			Base class for establishing a connection to a remote server daemon.
72			"""
73
74			def __init__(self, timeout=DEFAULT_TIMEOUT):
75			"""
76			Arguments:
77			socket -- A socket
78			"""
79			self._socket = None
80			self._timeout = timeout
81
82			def connect(self):
83			"""Establish a connection to gvmd
84			"""
85			raise NotImplementedError
86
87			def send(self, data):
88			"""Send data to gvmd
89			"""
90			if isinstance(data, str):
91			self._socket.send(data.encode())
92			else:
93			self._socket.send(data)
94
95			def read(self):
96			"""Read data from gvmd
97			"""
98			raise NotImplementedError
99
100			def disconnect(self):
101			"""Close the connection to gvmd
102			"""
103			try:
104			if self._socket is not None:
105			self._socket.close()
106			except OSError as e:
107			logger.debug('Connection closing error: %s', e)
108
109
110			class SSHConnection(GvmConnection, XmlReader):
111			"""
112			SSH Class to connect, read and write from GVM via SSH
113			"""
114
115			def __init__(self, timeout=DEFAULT_TIMEOUT, hostname='127.0.0.1', port=22,
116			username='gmp', password=''):
117			super().__init__(timeout=timeout)
118
119			self.hostname = hostname
120			self.port = int(port)
121			self.username = username
122			self.password = password
123
124			def _send_in_chunks(self, data, chunk_size):
125			i_start = 0
126			i_end = chunk_size
127			sent_bytes = 0
128			length = len(data)
129
130			while sent_bytes < length:
131			time.sleep(0.01)
132
133			self._stdin.channel.send(data[i_start:i_end])
134
135			i_start = i_end
136			if i_end > length:
137			i_end = length
138			else:
139			i_end = i_end + chunk_size
140
141			sent_bytes += (i_end - i_start)
142
143			return sent_bytes
144
145			def connect(self):
146			self._socket = paramiko.SSHClient()
147			self._socket.set_missing_host_key_policy(paramiko.AutoAddPolicy())
148
149			try:
150			self._socket.connect(
151			hostname=self.hostname,
152			username=self.username,
153			password=self.password,
154			timeout=self._timeout,
155			port=int(self.port),
156			allow_agent=False,
157			look_for_keys=False)
158			self._stdin, self._stdout, self._stderr = self._socket.exec_command(
159			"", get_pty=False)
160
161			except (paramiko.BadHostKeyException,
162			paramiko.AuthenticationException,
163			paramiko.SSHException, OSError) as e:
164			logger.debug('SSH Connection failed: %s', e)
165			raise
166
167			def read(self):
168			response = ''
169
170			self._start_xml()
171
172			while True:
173			data = self._stdout.channel.recv(BUF_SIZE)
174			# Connection was closed by server
175			if not data:
176			break
177
178			self._feed_xml(data)
179
180			response += data.decode('utf-8', errors='ignore')
181
182			if self._is_end_xml():
183			break
184
185			return response
186
187			def send(self, data):
188			logger.debug('SSH:send(): %s', data)
189			if len(data) > MAX_SSH_DATA_LENGTH:
190			sent_bytes = self._send_in_chunks(data, MAX_SSH_DATA_LENGTH)
191			logger.debug("SSH: %s bytes sent.", sent_bytes)
192			else:
193			self._stdin.channel.send(data)
194
195
196			class TLSConnection(GvmConnection):
197			"""
198			TLS class to connect, read and write from a remote GVM daemon via TLS
199			secured socket.
200			"""
201
202			def __init__(self, certfile=None, cafile=None, keyfile=None,
203			hostname='127.0.0.1', port=DEFAULT_GVM_PORT,
204			timeout=DEFAULT_TIMEOUT):
205			super().__init__(timeout=timeout)
206
207			self.hostname = hostname
208			self.port = port
209			self.certfile = certfile
210			self.cafile = cafile
211			self.keyfile = keyfile
212
213			def _new_socket(self):
214			if self.certfile and self.cafile and self.keyfile:
215			context = ssl.create_default_context(ssl.Purpose.SERVER_AUTH,
216			cafile=self.cafile)
217			context.check_hostname = False
218			context.load_cert_chain(
219			certfile=self.certfile, keyfile=self.keyfile)
220			new_socket = socketlib.socket(socketlib.AF_INET,
221			socketlib.SOCK_STREAM)
222			sock = context.wrap_socket(new_socket, server_side=False)
223			else:
224			context = ssl.SSLContext(ssl.PROTOCOL_TLSv1_2)
225			sock = context.wrap_socket(socketlib.socket(socketlib.AF_INET))
226			return sock
227
228
229			def connect(self):
230			self._socket = self._new_socket()
231			self._socket.settimeout(self._timeout)
232			self._socket.connect((self.hostname, int(self.port)))
233
234			def read(self):
235			response = ''
236
237			while True:
238			data = self._socket.read(BUF_SIZE)
239
240			response += data.decode('utf-8', errors='ignore')
241			if len(data) < BUF_SIZE:
242			break
243
244			return response
245
246
247			class UnixSocketConnection(GvmConnection, XmlReader):
248			"""
249			UNIX-Socket class to connect, read, write from a GVM server daemon via
250			direct communicating UNIX-Socket
251			"""
252
253			def __init__(self, path=DEFAULT_UNIX_SOCKET_PATH, timeout=DEFAULT_TIMEOUT,
254			read_timeout=DEFAULT_READ_TIMEOUT):
255			super().__init__(timeout=timeout)
256
257			self.read_timeout = read_timeout
258			self.path = path
259
260			def connect(self):
261			"""Connect to the UNIX socket
262			"""
263			self._socket = socketlib.socket(
264			socketlib.AF_UNIX, socketlib.SOCK_STREAM)
265			self._socket.settimeout(self._timeout)
266			self._socket.connect(self.path)
267
268			def read(self):
269			"""Read from the UNIX socket
270			"""
271			response = ''
272
273			break_timeout = time.time() + self.read_timeout
274			old_timeout = self._socket.gettimeout()
275			self._socket.settimeout(5) # in seconds
276
277			self._start_xml()
278
279			while time.time() < break_timeout:
280			data = b''
281
282			try:
283			data = self._socket.recv(BUF_SIZE)
284			except (socketlib.timeout) as exception:
285			logger.debug('Warning: No data received '
286			'from server: %s', exception)
287			continue
288
289			self._feed_xml(data)
290
291			response += data.decode('utf-8', errors='ignore')
292
293			if len(data) < BUF_SIZE:
294			if self._is_end_xml():
295			break
296
297			self._socket.settimeout(old_timeout)
298			return response
299

greenbone / python-gvm

Push — master ( f97307...dda958 )

gvm.connections.GvmConnection.disconnect() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like