gvm.connections.GvmConnection.disconnect() - Code Metrics - greenbone/python-gvm - Measure and Improve Code Quality continuously with Scrutinizer

gvm.connections.GvmConnection.disconnect() A
last analyzed 2021-05-24 13:33 UTC

↳ Parent: gvm.connections

Complexity

Conditions

Size

Total Lines	7
Code Lines	6

Duplication

Lines	0
Ratio	0 %

Importance

Changes

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

# -*- coding: utf-8 -*-
# Copyright (C) 2018-2021 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

from typing import Optional, Union

import paramiko

from lxml import etree

from gvm.errors import GvmError


logger = logging.getLogger(__name__)

BUF_SIZE = 16 * 1024
DEFAULT_READ_TIMEOUT = 60  # in seconds
DEFAULT_TIMEOUT = 60  # in seconds
DEFAULT_GVM_PORT = 9390
DEFAULT_UNIX_SOCKET_PATH = "/var/run/gvmd.sock"
DEFAULT_SSH_PORT = 22
DEFAULT_SSH_USERNAME = "gmp"
DEFAULT_SSH_PASSWORD = ""
DEFAULT_HOSTNAME = '127.0.0.1'
MAX_SSH_DATA_LENGTH = 4095


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

    def _start_xml(self):
        self._first_element = None
        # act on start and end element events and
        # allow huge text data (for report content)
        self._parser = etree.XMLPullParser(
            events=("start", "end"), huge_tree=True
        )

    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(
                "Cannot parse XML response. Response data "
                "read {0}".format(data),
                e,
            ) from None


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

    Arguments:
        timeout: Timeout in seconds for the connection. None to
            wait indefinitely
    """

    def __init__(self, timeout: Optional[int] = DEFAULT_TIMEOUT):
        self._socket = None
        self._timeout = timeout if timeout is not None else DEFAULT_TIMEOUT

    def _read(self):
        return self._socket.recv(BUF_SIZE)

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

    def send(self, data: Union[bytes, str]):
        """Send data to the connected remote server

        Arguments:
            data: Data to be send to the server. Either utf-8 encoded string or
                bytes.
        """
        if self._socket is None:
            raise GvmError("Socket is not connected")

        if isinstance(data, str):
            self._socket.sendall(data.encode())
        else:
            self._socket.sendall(data)

    def read(self) -> str:
        """Read data from the remote server

        Returns:
            str: data as utf-8 encoded string
        """
        response = ""

        self._start_xml()

        if self._timeout is not None:
            now = time.time()

            break_timeout = now + self._timeout

        while True:
            data = self._read()

            if not data:
                # Connection was closed by server
                raise GvmError("Remote closed the connection")

            self._feed_xml(data)

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

            if self._is_end_xml():
                break

            if self._timeout is not None:
                now = time.time()

                if now > break_timeout:

                    raise GvmError("Timeout while reading the response")

        return response

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

    def finish_send(self):
        """Indicate to the remote server you are done with sending data"""
        # shutdown socket for sending. only allow reading data afterwards
        self._socket.shutdown(socketlib.SHUT_WR)


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

    Arguments:
        timeout: Timeout in seconds for the connection.
        hostname: DNS name or IP address of the remote server. Default is
            127.0.0.1.
        port: Port of the remote SSH server. Default is port 22.
        username: Username to use for SSH login. Default is "gmp".
        password: Passwort to use for SSH login. Default is "".
    """

    def __init__(
        self,
        *,
        timeout: Optional[int] = DEFAULT_TIMEOUT,
        hostname: Optional[str] = DEFAULT_HOSTNAME,
        port: Optional[int] = DEFAULT_SSH_PORT,
        username: Optional[str] = DEFAULT_SSH_USERNAME,
        password: Optional[str] = DEFAULT_SSH_PASSWORD,
    ):
        super().__init__(timeout=timeout)

        self.hostname = hostname if hostname is not None else DEFAULT_HOSTNAME
        self.port = int(port) if port is not None else DEFAULT_SSH_PORT
        self.username = (
            username if username is not None else DEFAULT_SSH_USERNAME
        )
        self.password = (
            password if password is not None else DEFAULT_SSH_PASSWORD
        )

    def _send_all(self, data):
        while data:
            sent = self._stdin.channel.send(data)

            if not sent:
                # Connection was closed by server
                raise GvmError("Remote closed the connection")

            data = data[sent:]

    def connect(self):
        """
        Connect to the SSH server and authenticate to it
        """
        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,
        ) as e:
            raise GvmError("SSH Connection failed", e) from None

    def _read(self):
        return self._stdout.channel.recv(BUF_SIZE)

    def send(self, data: Union[bytes, str]):
        self._send_all(data)

    def finish_send(self):
        # shutdown socket for sending. only allow reading data afterwards
        self._stdout.channel.shutdown(socketlib.SHUT_WR)

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

        if self._socket:
            del self._socket, self._stdin, self._stdout, self._stderr


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

    Arguments:
        timeout: Timeout in seconds for the connection.
        hostname: DNS name or IP address of the remote TLS server.
        port: Port for the TLS connection. Default is 9390.
        certfile: Path to PEM encoded certificate file. See
            `python certificates`_ for details.
        cafile: Path to PEM encoded CA file. See `python certificates`_
            for details.
        keyfile: Path to PEM encoded private key. See `python certificates`_
            for details.
        password: Password for the private key. If the password argument is not
            specified and a password is required it will be interactively prompt
            the user for a password.

    .. _python certificates:
        https://docs.python.org/3/library/ssl.html#certificates
    """

    def __init__(
        self,
        *,
        certfile: Optional[str] = None,
        cafile: Optional[str] = None,
        keyfile: Optional[str] = None,
        hostname: Optional[str] = DEFAULT_HOSTNAME,
        port: Optional[int] = DEFAULT_GVM_PORT,
        password: Optional[str] = None,
        timeout: Optional[int] = DEFAULT_TIMEOUT,
    ):
        super().__init__(timeout=timeout)

        self.hostname = hostname if hostname is not None else DEFAULT_HOSTNAME
        self.port = port if port is not None else DEFAULT_GVM_PORT
        self.certfile = certfile
        self.cafile = cafile
        self.keyfile = keyfile
        self.password = password

    def _new_socket(self):
        transport_socket = socketlib.socket(
            socketlib.AF_INET, socketlib.SOCK_STREAM
        )

        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,
                password=self.password,
            )
            sock = context.wrap_socket(transport_socket, server_side=False)
        else:
            context = ssl.SSLContext(ssl.PROTOCOL_TLSv1_2)
            sock = context.wrap_socket(transport_socket)

        sock.settimeout(self._timeout)

        return sock

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


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

    Arguments:
        path: Path to the socket. Default is "/var/run/gvmd.sock".
        timeout: Timeout in seconds for the connection. Default is 60 seconds.
    """

    def __init__(
        self,
        *,
        path: Optional[str] = DEFAULT_UNIX_SOCKET_PATH,
        timeout: Optional[int] = DEFAULT_TIMEOUT,
    ):
        super().__init__(timeout=timeout)

        self.path = path if path is not None else DEFAULT_UNIX_SOCKET_PATH

    def connect(self):
        """Connect to the UNIX socket"""
        self._socket = socketlib.socket(
            socketlib.AF_UNIX, socketlib.SOCK_STREAM
        )
        self._socket.settimeout(self._timeout)
        try:
            self._socket.connect(self.path)
        except FileNotFoundError:
            raise GvmError(
                "Socket {path} does not exist".format(path=self.path)
            ) from None
        except ConnectionError:
            raise GvmError(
                "Could not connect to socket {}".format(self.path)
            ) from None


class DebugConnection:
    """Wrapper around a connection for debugging purposes

    Allows to debug the connection flow including send and read data. Internally
    it uses the python `logging`_ framework to create debug messages. Please
    take a look at `the logging tutorial
    <https://docs.python.org/3/howto/logging.html#logging-basic-tutorial>`_
    for further details.

    Usage example:

    .. code-block:: python

        import logging

        logging.basicConfig(level=logging.DEBUG)

        socketconnection = UnixSocketConnection(path='/var/run/gvm.sock')
        connection = DebugConnection(socketconnection)
        gmp = Gmp(connection=connection)

    Arguments:
        connection: GvmConnection to observe

    .. _logging:
        https://docs.python.org/3/library/logging.html
    """

    def __init__(self, connection: GvmConnection):
        self._connection = connection

    def read(self) -> str:
        data = self._connection.read()

        logger.debug("Read %s characters. Data %s", len(data), data)

        self.last_read_data = data
        return data

    def send(self, data):
        self.last_send_data = data

        logger.debug("Sending %s characters. Data %s", len(data), data)

        return self._connection.send(data)

    def connect(self):
        logger.debug("Connecting")

        return self._connection.connect()

    def disconnect(self):
        logger.debug("Disconnecting")

        return self._connection.disconnect()

    def finish_send(self):
        logger.debug("Finish send")

        self._connection.finish_send()


1			# -- coding: utf-8 --
2			# Copyright (C) 2018-2021 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			from typing import Optional, Union
27
28			import paramiko
29
30			from lxml import etree
31
32			from gvm.errors import GvmError
33
34
35			logger = logging.getLogger(__name__)
36
37			BUF_SIZE = 16 * 1024
38			DEFAULT_READ_TIMEOUT = 60 # in seconds
39			DEFAULT_TIMEOUT = 60 # in seconds
40			DEFAULT_GVM_PORT = 9390
41			DEFAULT_UNIX_SOCKET_PATH = "/var/run/gvmd.sock"
42			DEFAULT_SSH_PORT = 22
43			DEFAULT_SSH_USERNAME = "gmp"
44			DEFAULT_SSH_PASSWORD = ""
45			DEFAULT_HOSTNAME = '127.0.0.1'
46			MAX_SSH_DATA_LENGTH = 4095
47
48
49			class XmlReader:
50			"""
51			Read a XML command until its closing element
52			"""
53
54			def _start_xml(self):
55			self._first_element = None
56			# act on start and end element events and
57			# allow huge text data (for report content)
58			self._parser = etree.XMLPullParser(
59			events=("start", "end"), huge_tree=True
60			)
61
62			def _is_end_xml(self):
63			for action, obj in self._parser.read_events():
64			if not self._first_element and action in "start":
65			self._first_element = obj.tag
66
67			if (
68			self._first_element
69			and action in "end"
70			and str(self._first_element) == str(obj.tag)
71			):
72			return True
73			return False
74
75			def _feed_xml(self, data):
76			try:
77			self._parser.feed(data)
78			except etree.ParseError as e:
79			raise GvmError(
80			"Cannot parse XML response. Response data "
81			"read {0}".format(data),
82			e,
83			) from None
84
85
86			class GvmConnection(XmlReader):
87			"""
88			Base class for establishing a connection to a remote server daemon.
89
90			Arguments:
91			timeout: Timeout in seconds for the connection. None to
92			wait indefinitely
93			"""
94
95			def __init__(self, timeout: Optional[int] = DEFAULT_TIMEOUT):
96			self._socket = None
97			self._timeout = timeout if timeout is not None else DEFAULT_TIMEOUT
98
99			def _read(self):
100			return self._socket.recv(BUF_SIZE)
101
102			def connect(self):
103			"""Establish a connection to a remote server"""
104			raise NotImplementedError
105
106			def send(self, data: Union[bytes, str]):
107			"""Send data to the connected remote server
108
109			Arguments:
110			data: Data to be send to the server. Either utf-8 encoded string or
111			bytes.
112			"""
113			if self._socket is None:
114			raise GvmError("Socket is not connected")
115
116			if isinstance(data, str):
117			self._socket.sendall(data.encode())
118			else:
119			self._socket.sendall(data)
120
121			def read(self) -> str:
122			"""Read data from the remote server
123
124			Returns:
125			str: data as utf-8 encoded string
126			"""
127			response = ""
128
129			self._start_xml()
130
131			if self._timeout is not None:
132			now = time.time()
133
134			break_timeout = now + self._timeout
135
136			while True:
137			data = self._read()
138
139			if not data:
140			# Connection was closed by server
141			raise GvmError("Remote closed the connection")
142
143			self._feed_xml(data)
144
145			response += data.decode("utf-8", errors="ignore")
146
147			if self._is_end_xml():
148			break
149
150			if self._timeout is not None:
151			now = time.time()
152
153			if now > break_timeout:
			0 ignored issues – show introduced 2018-11-30 08:56 UTC by Report Bug Copy Issue Report The variable `break_timeout` does not seem to be defined in case `self._timeout is not None` on line `131` is `False`. Are you sure this can never be the case? Loading history...
154			raise GvmError("Timeout while reading the response")
155
156			return response
157
158			def disconnect(self):
159			"""Disconnect and close the connection to the remote server"""
160			try:
161			if self._socket is not None:
162			self._socket.close()
163			except OSError as e:
164			logger.debug("Connection closing error: %s", e)
165
166			def finish_send(self):
167			"""Indicate to the remote server you are done with sending data"""
168			# shutdown socket for sending. only allow reading data afterwards
169			self._socket.shutdown(socketlib.SHUT_WR)
170
171
172			class SSHConnection(GvmConnection):
173			"""
174			SSH Class to connect, read and write from GVM via SSH
175
176			Arguments:
177			timeout: Timeout in seconds for the connection.
178			hostname: DNS name or IP address of the remote server. Default is
179			127.0.0.1.
180			port: Port of the remote SSH server. Default is port 22.
181			username: Username to use for SSH login. Default is "gmp".
182			password: Passwort to use for SSH login. Default is "".
183			"""
184
185			def __init__(
186			self,
187			*,
188			timeout: Optional[int] = DEFAULT_TIMEOUT,
189			hostname: Optional[str] = DEFAULT_HOSTNAME,
190			port: Optional[int] = DEFAULT_SSH_PORT,
191			username: Optional[str] = DEFAULT_SSH_USERNAME,
192			password: Optional[str] = DEFAULT_SSH_PASSWORD,
193			):
194			super().__init__(timeout=timeout)
195
196			self.hostname = hostname if hostname is not None else DEFAULT_HOSTNAME
197			self.port = int(port) if port is not None else DEFAULT_SSH_PORT
198			self.username = (
199			username if username is not None else DEFAULT_SSH_USERNAME
200			)
201			self.password = (
202			password if password is not None else DEFAULT_SSH_PASSWORD
203			)
204
205			def _send_all(self, data):
206			while data:
207			sent = self._stdin.channel.send(data)
208
209			if not sent:
210			# Connection was closed by server
211			raise GvmError("Remote closed the connection")
212
213			data = data[sent:]
214
215			def connect(self):
216			"""
217			Connect to the SSH server and authenticate to it
218			"""
219			self._socket = paramiko.SSHClient()
220			self._socket.set_missing_host_key_policy(paramiko.AutoAddPolicy())
221
222			try:
223			self._socket.connect(
224			hostname=self.hostname,
225			username=self.username,
226			password=self.password,
227			timeout=self._timeout,
228			port=int(self.port),
229			allow_agent=False,
230			look_for_keys=False,
231			)
232			self._stdin, self._stdout, self._stderr = self._socket.exec_command(
233			"", get_pty=False
234			)
235
236			except (
237			paramiko.BadHostKeyException,
238			paramiko.AuthenticationException,
239			paramiko.SSHException,
240			) as e:
241			raise GvmError("SSH Connection failed", e) from None
242
243			def _read(self):
244			return self._stdout.channel.recv(BUF_SIZE)
245
246			def send(self, data: Union[bytes, str]):
247			self._send_all(data)
248
249			def finish_send(self):
250			# shutdown socket for sending. only allow reading data afterwards
251			self._stdout.channel.shutdown(socketlib.SHUT_WR)
252
253			def disconnect(self):
254			"""Disconnect and close the connection to the remote server"""
255			try:
256			if self._socket is not None:
257			self._socket.close()
258			except OSError as e:
259			logger.debug("Connection closing error: %s", e)
260
261			if self._socket:
262			del self._socket, self._stdin, self._stdout, self._stderr
263
264
265			class TLSConnection(GvmConnection):
266			"""
267			TLS class to connect, read and write from a remote GVM daemon via TLS
268			secured socket.
269
270			Arguments:
271			timeout: Timeout in seconds for the connection.
272			hostname: DNS name or IP address of the remote TLS server.
273			port: Port for the TLS connection. Default is 9390.
274			certfile: Path to PEM encoded certificate file. See
275			`python certificates`_ for details.
276			cafile: Path to PEM encoded CA file. See `python certificates`_
277			for details.
278			keyfile: Path to PEM encoded private key. See `python certificates`_
279			for details.
280			password: Password for the private key. If the password argument is not
281			specified and a password is required it will be interactively prompt
282			the user for a password.
283
284			.. _python certificates:
285			https://docs.python.org/3/library/ssl.html#certificates
286			"""
287
288			def __init__(
289			self,
290			*,
291			certfile: Optional[str] = None,
292			cafile: Optional[str] = None,
293			keyfile: Optional[str] = None,
294			hostname: Optional[str] = DEFAULT_HOSTNAME,
295			port: Optional[int] = DEFAULT_GVM_PORT,
296			password: Optional[str] = None,
297			timeout: Optional[int] = DEFAULT_TIMEOUT,
298			):
299			super().__init__(timeout=timeout)
300
301			self.hostname = hostname if hostname is not None else DEFAULT_HOSTNAME
302			self.port = port if port is not None else DEFAULT_GVM_PORT
303			self.certfile = certfile
304			self.cafile = cafile
305			self.keyfile = keyfile
306			self.password = password
307
308			def _new_socket(self):
309			transport_socket = socketlib.socket(
310			socketlib.AF_INET, socketlib.SOCK_STREAM
311			)
312
313			if self.certfile and self.cafile and self.keyfile:
314			context = ssl.create_default_context(
315			ssl.Purpose.SERVER_AUTH, cafile=self.cafile
316			)
317			context.check_hostname = False
318			context.load_cert_chain(
319			certfile=self.certfile,
320			keyfile=self.keyfile,
321			password=self.password,
322			)
323			sock = context.wrap_socket(transport_socket, server_side=False)
324			else:
325			context = ssl.SSLContext(ssl.PROTOCOL_TLSv1_2)
326			sock = context.wrap_socket(transport_socket)
327
328			sock.settimeout(self._timeout)
329
330			return sock
331
332			def connect(self):
333			self._socket = self._new_socket()
334			self._socket.connect((self.hostname, int(self.port)))
335
336
337			class UnixSocketConnection(GvmConnection):
338			"""
339			UNIX-Socket class to connect, read, write from a GVM server daemon via
340			direct communicating UNIX-Socket
341
342			Arguments:
343			path: Path to the socket. Default is "/var/run/gvmd.sock".
344			timeout: Timeout in seconds for the connection. Default is 60 seconds.
345			"""
346
347			def __init__(
348			self,
349			*,
350			path: Optional[str] = DEFAULT_UNIX_SOCKET_PATH,
351			timeout: Optional[int] = DEFAULT_TIMEOUT,
352			):
353			super().__init__(timeout=timeout)
354
355			self.path = path if path is not None else DEFAULT_UNIX_SOCKET_PATH
356
357			def connect(self):
358			"""Connect to the UNIX socket"""
359			self._socket = socketlib.socket(
360			socketlib.AF_UNIX, socketlib.SOCK_STREAM
361			)
362			self._socket.settimeout(self._timeout)
363			try:
364			self._socket.connect(self.path)
365			except FileNotFoundError:
366			raise GvmError(
367			"Socket {path} does not exist".format(path=self.path)
368			) from None
369			except ConnectionError:
370			raise GvmError(
371			"Could not connect to socket {}".format(self.path)
372			) from None
373
374
375			class DebugConnection:
376			"""Wrapper around a connection for debugging purposes
377
378			Allows to debug the connection flow including send and read data. Internally
379			it uses the python `logging`_ framework to create debug messages. Please
380			take a look at `the logging tutorial
381			<https://docs.python.org/3/howto/logging.html#logging-basic-tutorial>`_
382			for further details.
383
384			Usage example:
385
386			.. code-block:: python
387
388			import logging
389
390			logging.basicConfig(level=logging.DEBUG)
391
392			socketconnection = UnixSocketConnection(path='/var/run/gvm.sock')
393			connection = DebugConnection(socketconnection)
394			gmp = Gmp(connection=connection)
395
396			Arguments:
397			connection: GvmConnection to observe
398
399			.. _logging:
400			https://docs.python.org/3/library/logging.html
401			"""
402
403			def __init__(self, connection: GvmConnection):
404			self._connection = connection
405
406			def read(self) -> str:
407			data = self._connection.read()
408
409			logger.debug("Read %s characters. Data %s", len(data), data)
410
411			self.last_read_data = data
412			return data
413
414			def send(self, data):
415			self.last_send_data = data
416
417			logger.debug("Sending %s characters. Data %s", len(data), data)
418
419			return self._connection.send(data)
420
421			def connect(self):
422			logger.debug("Connecting")
423
424			return self._connection.connect()
425
426			def disconnect(self):
427			logger.debug("Disconnecting")
428
429			return self._connection.disconnect()
430
431			def finish_send(self):
432			logger.debug("Finish send")
433
434			self._connection.finish_send()
435

greenbone / python-gvm

gvm.connections.GvmConnection.disconnect() A last analyzed 2021-05-24 13:33 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

gvm.connections.GvmConnection.disconnect() A
last analyzed 2021-05-24 13:33 UTC