gvm.protocols.gmpv208.gmpv208.GmpV208Mixin.clone_agent() - Code Metrics - Inspection of "Override FeedType and get_feed for 20.08" - greenbone/python-gvm - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#434)

by Jaspar

created 2021-04-26 13:56 UTC

GmpV208Mixin.clone_agent() A

↳ Parent: gvm.protocols.gmpv208.gmpv208

Complexity

Conditions

Size

Total Lines	7
Code Lines	6

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	6
nop	2
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/>.

# pylint: disable=arguments-differ, redefined-builtin, too-many-lines

"""
Module for communication with gvmd in
`Greenbone Management Protocol version 20.08`_

.. _Greenbone Management Protocol version 20.08:
    https://docs.greenbone.net/API/GMP/gmp-20.08.html
"""
import warnings

from typing import Any, List, Optional, Callable

from gvm.utils import deprecation
from gvm.xml import XmlCommand

from gvm.protocols.gmpv7.gmpv7 import _to_bool, _to_comma_list, _add_filter
from gvm.connections import GvmConnection
from gvm.errors import InvalidArgumentType, RequiredArgument

from gvm.protocols.base import GvmProtocol

from . import types
from .types import *  # pylint: disable=unused-wildcard-import, wildcard-import

_EMPTY_POLICY_ID = '085569ce-73ed-11df-83c3-002264764cea'

PROTOCOL_VERSION = (20, 8)


class GmpV208Mixin(GvmProtocol):
    types = types


    def __init__(
        self,
        connection: GvmConnection,
        *,
        transform: Optional[Callable[[str], Any]] = None,
    ):
        super().__init__(connection, transform=transform)

        # Is authenticated on gvmd
        self._authenticated = False

    def create_agent(
        self,
        installer: str,
        signature: str,
        name: str,
        *,
        comment: Optional[str] = None,
        howto_install: Optional[str] = None,
        howto_use: Optional[str] = None,
    ) -> None:
        # pylint: disable=unused-argument
        deprecation(
            "{} has been removed in GMP version {}.{}".format(
                self.create_agent.__name__,
                self.get_protocol_version()[0],
                self.get_protocol_version()[1],
            )
        )

    def clone_agent(self, agent_id: str) -> None:
        # pylint: disable=unused-argument
        deprecation(
            "{} has been removed in GMP version {}.{}".format(
                self.clone_agent.__name__,
                self.get_protocol_version()[0],
                self.get_protocol_version()[1],
            )
        )

    def modify_agent(
        self,
        agent_id: str,
        *,
        name: Optional[str] = None,
        comment: Optional[str] = None,
    ) -> None:
        # pylint: disable=unused-argument
        deprecation(
            "{} has been removed in GMP version {}.{}".format(
                self.clone_agent.__name__,
                self.get_protocol_version()[0],
                self.get_protocol_version()[1],
            )
        )

    def delete_agent(
        self,
        agent_id: str,
        *,
        ultimate: Optional[bool] = False
        # pylint: disable=unused-argument
    ) -> None:
        deprecation(
            "{} has been removed in GMP version {}.{}".format(
                self.delete_agent.__name__,
                self.get_protocol_version()[0],
                self.get_protocol_version()[1],
            )
        )

    def verify_agent(self, agent_id: str) -> None:
        # pylint: disable=unused-argument
        deprecation(
            "{} has been removed in GMP version {}.{}".format(
                self.verify_agent.__name__,
                self.get_protocol_version()[0],
                self.get_protocol_version()[1],
            )
        )

    def get_agent(self, agent_id: str) -> None:
        # pylint: disable=unused-argument
        deprecation(
            "{} has been removed in GMP version {}.{}".format(
                self.get_agent.__name__,
                self.get_protocol_version()[0],
                self.get_protocol_version()[1],
            )
        )

    def get_info_list(

        self,
        info_type: InfoType,
        *,
        filter: Optional[str] = None,
        filter_id: Optional[str] = None,
        name: Optional[str] = None,
        details: Optional[bool] = None,
    ) -> Any:
        """Request a list of security information

        Arguments:
            info_type: Type must be either CERT_BUND_ADV, CPE, CVE,
                DFN_CERT_ADV, OVALDEF, NVT or ALLINFO
            filter: Filter term to use for the query
            filter_id: UUID of an existing filter to use for the query
            name: Name or identifier of the requested information
            details: Whether to include information about references to this
                information

        Returns:
            The response. See :py:meth:`send_command` for details.
        """
        if not info_type:
            raise RequiredArgument(
                function=self.get_info_list.__name__, argument='info_type'
            )

        if not isinstance(info_type, InfoType):
            raise InvalidArgumentType(
                function=self.get_info_list.__name__,
                argument='info_type',
                arg_type=InfoType.__name__,
            )

        cmd = XmlCommand("get_info")

        cmd.set_attribute("type", info_type.value)

        _add_filter(cmd, filter, filter_id)

        if name:
            cmd.set_attribute("name", name)

        if details is not None:
            cmd.set_attribute("details", _to_bool(details))

        return self._send_xml_command(cmd)

    def get_info(self, info_id: str, info_type: InfoType) -> Any:

        """Request a single secinfo

        Arguments:
            info_id: UUID of an existing secinfo
            info_type: Type must be either CERT_BUND_ADV, CPE, CVE,
                DFN_CERT_ADV, OVALDEF, NVT

        Returns:
            The response. See :py:meth:`send_command` for details.
        """
        if not info_type:
            raise RequiredArgument(
                function=self.get_info.__name__, argument='info_type'
            )

        if not isinstance(info_type, InfoType):
            raise InvalidArgumentType(
                function=self.get_info.__name__,
                argument='info_type',
                arg_type=InfoType.__name__,
            )

        if not info_id:
            raise RequiredArgument(
                function=self.get_info.__name__, argument='info_id'
            )

        cmd = XmlCommand("get_info")
        cmd.set_attribute("info_id", info_id)

        cmd.set_attribute("type", info_type.value)

        # for single entity always request all details
        cmd.set_attribute("details", "1")
        return self._send_xml_command(cmd)

    def create_target(
        self,
        name: str,
        *,
        make_unique: Optional[bool] = None,
        asset_hosts_filter: Optional[str] = None,
        hosts: Optional[List[str]] = None,
        comment: Optional[str] = None,
        exclude_hosts: Optional[List[str]] = None,
        ssh_credential_id: Optional[str] = None,
        ssh_credential_port: Optional[int] = None,
        smb_credential_id: Optional[str] = None,
        esxi_credential_id: Optional[str] = None,
        snmp_credential_id: Optional[str] = None,
        alive_test: Optional[AliveTest] = None,
        reverse_lookup_only: Optional[bool] = None,
        reverse_lookup_unify: Optional[bool] = None,
        port_range: Optional[str] = None,
        port_list_id: Optional[str] = None,
    ) -> Any:
        """Create a new target

        Arguments:
            name: Name of the target
            make_unique: Deprecated. Will be ignored.
            asset_hosts_filter: Filter to select target host from assets hosts
            hosts: List of hosts addresses to scan
            exclude_hosts: List of hosts addresses to exclude from scan
            comment: Comment for the target
            ssh_credential_id: UUID of a ssh credential to use on target
            ssh_credential_port: The port to use for ssh credential
            smb_credential_id: UUID of a smb credential to use on target
            snmp_credential_id: UUID of a snmp credential to use on target
            esxi_credential_id: UUID of a esxi credential to use on target
            alive_test: Which alive test to use
            reverse_lookup_only: Whether to scan only hosts that have names
            reverse_lookup_unify: Whether to scan only one IP when multiple IPs
                have the same name.
            port_range: Port range for the target
            port_list_id: UUID of the port list to use on target

        Returns:
            The response. See :py:meth:`send_command` for details.
        """

        cmd = XmlCommand("create_target")
        _xmlname = cmd.add_element("name", name)

        if make_unique is not None:
            warnings.warn(
                'create_target make_unique argument is deprecated '
                'and will be ignored.',
                DeprecationWarning,
            )

        if not name:
            raise RequiredArgument(
                function=self.create_target.__name__, argument='name'
            )

        if asset_hosts_filter:
            cmd.add_element(
                "asset_hosts", attrs={"filter": str(asset_hosts_filter)}
            )
        elif hosts:
            cmd.add_element("hosts", _to_comma_list(hosts))
        else:
            raise RequiredArgument(
                function=self.create_target.__name__,
                argument='hosts or asset_hosts_filter',
            )

        if comment:
            cmd.add_element("comment", comment)

        if exclude_hosts:
            cmd.add_element("exclude_hosts", _to_comma_list(exclude_hosts))

        if ssh_credential_id:
            _xmlssh = cmd.add_element(
                "ssh_credential", attrs={"id": ssh_credential_id}
            )
            if ssh_credential_port:
                _xmlssh.add_element("port", str(ssh_credential_port))

        if smb_credential_id:
            cmd.add_element("smb_credential", attrs={"id": smb_credential_id})

        if esxi_credential_id:
            cmd.add_element("esxi_credential", attrs={"id": esxi_credential_id})

        if snmp_credential_id:
            cmd.add_element("snmp_credential", attrs={"id": snmp_credential_id})

        if alive_test:
            if not isinstance(alive_test, AliveTest):
                raise InvalidArgumentType(
                    function=self.create_target.__name__,
                    argument='alive_test',
                    arg_type=AliveTest.__name__,
                )

            cmd.add_element("alive_tests", alive_test.value)

        if reverse_lookup_only is not None:
            cmd.add_element(
                "reverse_lookup_only", _to_bool(reverse_lookup_only)
            )

        if reverse_lookup_unify is not None:
            cmd.add_element(
                "reverse_lookup_unify", _to_bool(reverse_lookup_unify)
            )

        # since 20.08 one of port_range or port_list_id is required!
        if not port_range and not port_list_id:
            raise RequiredArgument(
                function=self.create_target.__name__,
                argument='port_range or port_list_id',
            )

        if port_range:
            cmd.add_element("port_range", port_range)

        if port_list_id:
            cmd.add_element("port_list", attrs={"id": port_list_id})

        return self._send_xml_command(cmd)

    def get_feed(self, feed_type: Optional[FeedType]) -> Any:

        """Request a single feed

        Arguments:
            feed_type: Type of single feed to get: NVT, CERT or SCAP

        Returns:
            The response. See :py:meth:`send_command` for details.
        """
        if not feed_type:
            raise RequiredArgument(
                function=self.get_feed.__name__, argument='feed_type'
            )

        if not isinstance(feed_type, FeedType):
            raise InvalidArgumentType(
                function=self.get_feed.__name__,
                argument='feed_type',
                arg_type=FeedType.__name__,
            )

        cmd = XmlCommand("get_feeds")
        cmd.set_attribute("type", feed_type.value)

        return self._send_xml_command(cmd)


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		# pylint: disable=arguments-differ, redefined-builtin, too-many-lines
20
21		"""
22		Module for communication with gvmd in
23		`Greenbone Management Protocol version 20.08`_
24
25		.. _Greenbone Management Protocol version 20.08:
26		https://docs.greenbone.net/API/GMP/gmp-20.08.html
27		"""
28		import warnings
29
30		from typing import Any, List, Optional, Callable
31
32		from gvm.utils import deprecation
33		from gvm.xml import XmlCommand
34
35		from gvm.protocols.gmpv7.gmpv7 import _to_bool, _to_comma_list, _add_filter
36		from gvm.connections import GvmConnection
37		from gvm.errors import InvalidArgumentType, RequiredArgument
38
39		from gvm.protocols.base import GvmProtocol
40
41		from . import types
42		from .types import * # pylint: disable=unused-wildcard-import, wildcard-import
43
44		_EMPTY_POLICY_ID = '085569ce-73ed-11df-83c3-002264764cea'
45
46		PROTOCOL_VERSION = (20, 8)
47
48
49		class GmpV208Mixin(GvmProtocol):
50		types = types
		0 ignored issues – show Comprehensibility Best Practice introduced 2020-07-06 11:37 UTC by Report Bug Copy Issue Report The variable `types` does not seem to be defined. Loading history...
51
52		def __init__(
53		self,
54		connection: GvmConnection,
55		*,
56		transform: Optional[Callable[[str], Any]] = None,
57		):
58		super().__init__(connection, transform=transform)
59
60		# Is authenticated on gvmd
61		self._authenticated = False
62
63		def create_agent(
64		self,
65		installer: str,
66		signature: str,
67		name: str,
68		*,
69		comment: Optional[str] = None,
70		howto_install: Optional[str] = None,
71		howto_use: Optional[str] = None,
72		) -> None:
73		# pylint: disable=unused-argument
74		deprecation(
75		"{} has been removed in GMP version {}.{}".format(
76		self.create_agent.__name__,
77		self.get_protocol_version()[0],
78		self.get_protocol_version()[1],
79		)
80		)
81
82		def clone_agent(self, agent_id: str) -> None:
83		# pylint: disable=unused-argument
84		deprecation(
85		"{} has been removed in GMP version {}.{}".format(
86		self.clone_agent.__name__,
87		self.get_protocol_version()[0],
88		self.get_protocol_version()[1],
89		)
90		)
91
92		def modify_agent(
93		self,
94		agent_id: str,
95		*,
96		name: Optional[str] = None,
97		comment: Optional[str] = None,
98		) -> None:
99		# pylint: disable=unused-argument
100		deprecation(
101		"{} has been removed in GMP version {}.{}".format(
102		self.clone_agent.__name__,
103		self.get_protocol_version()[0],
104		self.get_protocol_version()[1],
105		)
106		)
107
108		def delete_agent(
109		self,
110		agent_id: str,
111		*,
112		ultimate: Optional[bool] = False
113		# pylint: disable=unused-argument
114		) -> None:
115		deprecation(
116		"{} has been removed in GMP version {}.{}".format(
117		self.delete_agent.__name__,
118		self.get_protocol_version()[0],
119		self.get_protocol_version()[1],
120		)
121		)
122
123		def verify_agent(self, agent_id: str) -> None:
124		# pylint: disable=unused-argument
125		deprecation(
126		"{} has been removed in GMP version {}.{}".format(
127		self.verify_agent.__name__,
128		self.get_protocol_version()[0],
129		self.get_protocol_version()[1],
130		)
131		)
132
133		def get_agent(self, agent_id: str) -> None:
134		# pylint: disable=unused-argument
135		deprecation(
136		"{} has been removed in GMP version {}.{}".format(
137		self.get_agent.__name__,
138		self.get_protocol_version()[0],
139		self.get_protocol_version()[1],
140		)
141		)
142
143	View Code Duplication	def get_info_list(
		0 ignored issues – show Duplication introduced 2020-12-14 07:38 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
144		self,
145		info_type: InfoType,
146		*,
147		filter: Optional[str] = None,
148		filter_id: Optional[str] = None,
149		name: Optional[str] = None,
150		details: Optional[bool] = None,
151		) -> Any:
152		"""Request a list of security information
153
154		Arguments:
155		info_type: Type must be either CERT_BUND_ADV, CPE, CVE,
156		DFN_CERT_ADV, OVALDEF, NVT or ALLINFO
157		filter: Filter term to use for the query
158		filter_id: UUID of an existing filter to use for the query
159		name: Name or identifier of the requested information
160		details: Whether to include information about references to this
161		information
162
163		Returns:
164		The response. See :py:meth:`send_command` for details.
165		"""
166		if not info_type:
167		raise RequiredArgument(
168		function=self.get_info_list.__name__, argument='info_type'
169		)
170
171		if not isinstance(info_type, InfoType):
172		raise InvalidArgumentType(
173		function=self.get_info_list.__name__,
174		argument='info_type',
175		arg_type=InfoType.__name__,
176		)
177
178		cmd = XmlCommand("get_info")
179
180		cmd.set_attribute("type", info_type.value)
181
182		_add_filter(cmd, filter, filter_id)
183
184		if name:
185		cmd.set_attribute("name", name)
186
187		if details is not None:
188		cmd.set_attribute("details", _to_bool(details))
189
190		return self._send_xml_command(cmd)
191
192	View Code Duplication	def get_info(self, info_id: str, info_type: InfoType) -> Any:
		0 ignored issues – show Duplication introduced 2020-07-06 13:09 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
193		"""Request a single secinfo
194
195		Arguments:
196		info_id: UUID of an existing secinfo
197		info_type: Type must be either CERT_BUND_ADV, CPE, CVE,
198		DFN_CERT_ADV, OVALDEF, NVT
199
200		Returns:
201		The response. See :py:meth:`send_command` for details.
202		"""
203		if not info_type:
204		raise RequiredArgument(
205		function=self.get_info.__name__, argument='info_type'
206		)
207
208		if not isinstance(info_type, InfoType):
209		raise InvalidArgumentType(
210		function=self.get_info.__name__,
211		argument='info_type',
212		arg_type=InfoType.__name__,
213		)
214
215		if not info_id:
216		raise RequiredArgument(
217		function=self.get_info.__name__, argument='info_id'
218		)
219
220		cmd = XmlCommand("get_info")
221		cmd.set_attribute("info_id", info_id)
222
223		cmd.set_attribute("type", info_type.value)
224
225		# for single entity always request all details
226		cmd.set_attribute("details", "1")
227		return self._send_xml_command(cmd)
228
229		def create_target(
230		self,
231		name: str,
232		*,
233		make_unique: Optional[bool] = None,
234		asset_hosts_filter: Optional[str] = None,
235		hosts: Optional[List[str]] = None,
236		comment: Optional[str] = None,
237		exclude_hosts: Optional[List[str]] = None,
238		ssh_credential_id: Optional[str] = None,
239		ssh_credential_port: Optional[int] = None,
240		smb_credential_id: Optional[str] = None,
241		esxi_credential_id: Optional[str] = None,
242		snmp_credential_id: Optional[str] = None,
243		alive_test: Optional[AliveTest] = None,
244		reverse_lookup_only: Optional[bool] = None,
245		reverse_lookup_unify: Optional[bool] = None,
246		port_range: Optional[str] = None,
247		port_list_id: Optional[str] = None,
248		) -> Any:
249		"""Create a new target
250
251		Arguments:
252		name: Name of the target
253		make_unique: Deprecated. Will be ignored.
254		asset_hosts_filter: Filter to select target host from assets hosts
255		hosts: List of hosts addresses to scan
256		exclude_hosts: List of hosts addresses to exclude from scan
257		comment: Comment for the target
258		ssh_credential_id: UUID of a ssh credential to use on target
259		ssh_credential_port: The port to use for ssh credential
260		smb_credential_id: UUID of a smb credential to use on target
261		snmp_credential_id: UUID of a snmp credential to use on target
262		esxi_credential_id: UUID of a esxi credential to use on target
263		alive_test: Which alive test to use
264		reverse_lookup_only: Whether to scan only hosts that have names
265		reverse_lookup_unify: Whether to scan only one IP when multiple IPs
266		have the same name.
267		port_range: Port range for the target
268		port_list_id: UUID of the port list to use on target
269
270		Returns:
271		The response. See :py:meth:`send_command` for details.
272		"""
273
274		cmd = XmlCommand("create_target")
275		_xmlname = cmd.add_element("name", name)
276
277		if make_unique is not None:
278		warnings.warn(
279		'create_target make_unique argument is deprecated '
280		'and will be ignored.',
281		DeprecationWarning,
282		)
283
284		if not name:
285		raise RequiredArgument(
286		function=self.create_target.__name__, argument='name'
287		)
288
289		if asset_hosts_filter:
290		cmd.add_element(
291		"asset_hosts", attrs={"filter": str(asset_hosts_filter)}
292		)
293		elif hosts:
294		cmd.add_element("hosts", _to_comma_list(hosts))
295		else:
296		raise RequiredArgument(
297		function=self.create_target.__name__,
298		argument='hosts or asset_hosts_filter',
299		)
300
301		if comment:
302		cmd.add_element("comment", comment)
303
304		if exclude_hosts:
305		cmd.add_element("exclude_hosts", _to_comma_list(exclude_hosts))
306
307		if ssh_credential_id:
308		_xmlssh = cmd.add_element(
309		"ssh_credential", attrs={"id": ssh_credential_id}
310		)
311		if ssh_credential_port:
312		_xmlssh.add_element("port", str(ssh_credential_port))
313
314		if smb_credential_id:
315		cmd.add_element("smb_credential", attrs={"id": smb_credential_id})
316
317		if esxi_credential_id:
318		cmd.add_element("esxi_credential", attrs={"id": esxi_credential_id})
319
320		if snmp_credential_id:
321		cmd.add_element("snmp_credential", attrs={"id": snmp_credential_id})
322
323		if alive_test:
324		if not isinstance(alive_test, AliveTest):
325		raise InvalidArgumentType(
326		function=self.create_target.__name__,
327		argument='alive_test',
328		arg_type=AliveTest.__name__,
329		)
330
331		cmd.add_element("alive_tests", alive_test.value)
332
333		if reverse_lookup_only is not None:
334		cmd.add_element(
335		"reverse_lookup_only", _to_bool(reverse_lookup_only)
336		)
337
338		if reverse_lookup_unify is not None:
339		cmd.add_element(
340		"reverse_lookup_unify", _to_bool(reverse_lookup_unify)
341		)
342
343		# since 20.08 one of port_range or port_list_id is required!
344		if not port_range and not port_list_id:
345		raise RequiredArgument(
346		function=self.create_target.__name__,
347		argument='port_range or port_list_id',
348		)
349
350		if port_range:
351		cmd.add_element("port_range", port_range)
352
353		if port_list_id:
354		cmd.add_element("port_list", attrs={"id": port_list_id})
355
356		return self._send_xml_command(cmd)
357
358	View Code Duplication	def get_feed(self, feed_type: Optional[FeedType]) -> Any:
		0 ignored issues – show Duplication introduced 2021-04-26 07:59 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
359		"""Request a single feed
360
361		Arguments:
362		feed_type: Type of single feed to get: NVT, CERT or SCAP
363
364		Returns:
365		The response. See :py:meth:`send_command` for details.
366		"""
367		if not feed_type:
368		raise RequiredArgument(
369		function=self.get_feed.__name__, argument='feed_type'
370		)
371
372		if not isinstance(feed_type, FeedType):
373		raise InvalidArgumentType(
374		function=self.get_feed.__name__,
375		argument='feed_type',
376		arg_type=FeedType.__name__,
377		)
378
379		cmd = XmlCommand("get_feeds")
380		cmd.set_attribute("type", feed_type.value)
381
382		return self._send_xml_command(cmd)
383

greenbone / python-gvm

Pull Request — master (#434)

GmpV208Mixin.clone_agent() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like