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 logging |
29
|
|
|
from numbers import Integral |
30
|
|
|
|
31
|
|
|
from typing import Any, Optional |
32
|
|
|
|
33
|
|
|
from gvm.errors import InvalidArgument, RequiredArgument |
34
|
|
|
from gvm.protocols.base import GvmProtocol |
35
|
|
|
|
36
|
|
|
|
37
|
|
|
from gvm.utils import ( |
38
|
|
|
check_command_status, |
39
|
|
|
to_bool, |
40
|
|
|
) |
41
|
|
|
from gvm.xml import XmlCommand |
42
|
|
|
|
43
|
|
|
PROTOCOL_VERSION = (20, 8) |
44
|
|
|
|
45
|
|
|
|
46
|
|
|
logger = logging.getLogger(__name__) |
47
|
|
|
|
48
|
|
|
|
49
|
|
|
class GmpV208Mixin(GvmProtocol): |
50
|
|
|
"""Python interface for Greenbone Management Protocol |
51
|
|
|
|
52
|
|
|
This class implements the `Greenbone Management Protocol version 20.08`_ |
53
|
|
|
|
54
|
|
|
Arguments: |
55
|
|
|
connection: Connection to use to talk with the gvmd daemon. See |
56
|
|
|
:mod:`gvm.connections` for possible connection types. |
57
|
|
|
transform: Optional transform `callable`_ to convert response data. |
58
|
|
|
After each request the callable gets passed the plain response data |
59
|
|
|
which can be used to check the data and/or conversion into different |
60
|
|
|
representations like a xml dom. |
61
|
|
|
|
62
|
|
|
See :mod:`gvm.transforms` for existing transforms. |
63
|
|
|
|
64
|
|
|
.. _Greenbone Management Protocol version 20.08: |
65
|
|
|
https://docs.greenbone.net/API/GMP/gmp-20.08.html |
66
|
|
|
.. _callable: |
67
|
|
|
https://docs.python.org/3/library/functions.html#callable |
68
|
|
|
""" |
69
|
|
|
|
70
|
|
|
def is_authenticated(self) -> bool: |
71
|
|
|
"""Checks if the user is authenticated |
72
|
|
|
|
73
|
|
|
If the user is authenticated privileged GMP commands like get_tasks |
74
|
|
|
may be send to gvmd. |
75
|
|
|
|
76
|
|
|
Returns: |
77
|
|
|
bool: True if an authenticated connection to gvmd has been |
78
|
|
|
established. |
79
|
|
|
""" |
80
|
|
|
return self._authenticated |
81
|
|
|
|
82
|
|
|
def authenticate(self, username: str, password: str) -> Any: |
83
|
|
|
"""Authenticate to gvmd. |
84
|
|
|
|
85
|
|
|
The generated authenticate command will be send to server. |
86
|
|
|
Afterwards the response is read, transformed and returned. |
87
|
|
|
|
88
|
|
|
Arguments: |
89
|
|
|
username: Username |
90
|
|
|
password: Password |
91
|
|
|
|
92
|
|
|
Returns: |
93
|
|
|
Transformed response from server. |
94
|
|
|
""" |
95
|
|
|
cmd = XmlCommand("authenticate") |
96
|
|
|
|
97
|
|
|
if not username: |
98
|
|
|
raise RequiredArgument( |
99
|
|
|
function=self.authenticate.__name__, argument='username' |
100
|
|
|
) |
101
|
|
|
|
102
|
|
|
if not password: |
103
|
|
|
raise RequiredArgument( |
104
|
|
|
function=self.authenticate.__name__, argument='password' |
105
|
|
|
) |
106
|
|
|
|
107
|
|
|
credentials = cmd.add_element("credentials") |
108
|
|
|
credentials.add_element("username", username) |
109
|
|
|
credentials.add_element("password", password) |
110
|
|
|
|
111
|
|
|
self._send(cmd.to_string()) |
112
|
|
|
response = self._read() |
113
|
|
|
|
114
|
|
|
if check_command_status(response): |
115
|
|
|
self._authenticated = True |
116
|
|
|
|
117
|
|
|
return self._transform(response) |
118
|
|
|
|
119
|
|
|
def describe_auth(self) -> Any: |
120
|
|
|
"""Describe authentication methods |
121
|
|
|
|
122
|
|
|
Returns a list of all used authentication methods if such a list is |
123
|
|
|
available. |
124
|
|
|
|
125
|
|
|
Returns: |
126
|
|
|
The response. See :py:meth:`send_command` for details. |
127
|
|
|
""" |
128
|
|
|
return self._send_xml_command(XmlCommand("describe_auth")) |
129
|
|
|
|
130
|
|
|
def empty_trashcan(self) -> Any: |
131
|
|
|
"""Empty the trashcan |
132
|
|
|
|
133
|
|
|
Remove all entities from the trashcan. **Attention:** this command can |
134
|
|
|
not be reverted |
135
|
|
|
|
136
|
|
|
Returns: |
137
|
|
|
The response. See :py:meth:`send_command` for details. |
138
|
|
|
""" |
139
|
|
|
return self._send_xml_command(XmlCommand("empty_trashcan")) |
140
|
|
|
|
141
|
|
|
def get_system_reports( |
142
|
|
|
self, |
143
|
|
|
*, |
144
|
|
|
name: Optional[str] = None, |
145
|
|
|
duration: Optional[int] = None, |
146
|
|
|
start_time: Optional[str] = None, |
147
|
|
|
end_time: Optional[str] = None, |
148
|
|
|
brief: Optional[bool] = None, |
149
|
|
|
slave_id: Optional[str] = None, |
150
|
|
|
) -> Any: |
151
|
|
|
"""Request a list of system reports |
152
|
|
|
|
153
|
|
|
Arguments: |
154
|
|
|
name: A string describing the required system report |
155
|
|
|
duration: The number of seconds into the past that the system report |
156
|
|
|
should include |
157
|
|
|
start_time: The start of the time interval the system report should |
158
|
|
|
include in ISO time format |
159
|
|
|
end_time: The end of the time interval the system report should |
160
|
|
|
include in ISO time format |
161
|
|
|
brief: Whether to include the actual system reports |
162
|
|
|
slave_id: UUID of GMP scanner from which to get the system reports |
163
|
|
|
|
164
|
|
|
Returns: |
165
|
|
|
The response. See :py:meth:`send_command` for details. |
166
|
|
|
""" |
167
|
|
|
cmd = XmlCommand("get_system_reports") |
168
|
|
|
|
169
|
|
|
if name: |
170
|
|
|
cmd.set_attribute("name", name) |
171
|
|
|
|
172
|
|
|
if duration is not None: |
173
|
|
|
if not isinstance(duration, Integral): |
174
|
|
|
raise InvalidArgument("duration needs to be an integer number") |
175
|
|
|
|
176
|
|
|
cmd.set_attribute("duration", str(duration)) |
177
|
|
|
|
178
|
|
|
if start_time: |
179
|
|
|
cmd.set_attribute("start_time", str(start_time)) |
180
|
|
|
|
181
|
|
|
if end_time: |
182
|
|
|
cmd.set_attribute("end_time", str(end_time)) |
183
|
|
|
|
184
|
|
|
if brief is not None: |
185
|
|
|
cmd.set_attribute("brief", to_bool(brief)) |
186
|
|
|
|
187
|
|
|
if slave_id: |
188
|
|
|
cmd.set_attribute("slave_id", slave_id) |
189
|
|
|
|
190
|
|
|
return self._send_xml_command(cmd) |
191
|
|
|
|
192
|
|
|
def get_version(self) -> Any: |
193
|
|
|
"""Get the Greenbone Manager Protocol version used by the remote gvmd |
194
|
|
|
Returns: |
195
|
|
|
The response. See :py:meth:`send_command` for details. |
196
|
|
|
""" |
197
|
|
|
return self._send_xml_command(XmlCommand("get_version")) |
198
|
|
|
|
199
|
|
|
def help( |
200
|
|
|
self, *, format: Optional[str] = None, help_type: Optional[str] = None |
201
|
|
|
) -> Any: |
202
|
|
|
"""Get the help text |
203
|
|
|
|
204
|
|
|
Arguments: |
205
|
|
|
format: One of "html", "rnc", "text" or "xml |
206
|
|
|
help_type: One of "brief" or "". Default "" |
207
|
|
|
|
208
|
|
|
Returns: |
209
|
|
|
The response. See :py:meth:`send_command` for details. |
210
|
|
|
""" |
211
|
|
|
cmd = XmlCommand("help") |
212
|
|
|
|
213
|
|
|
if not help_type: |
214
|
|
|
help_type = "" |
215
|
|
|
|
216
|
|
|
if help_type not in ("", "brief"): |
217
|
|
|
raise InvalidArgument( |
218
|
|
|
'help_type argument must be an empty string or "brief"' |
219
|
|
|
) |
220
|
|
|
|
221
|
|
|
cmd.set_attribute("type", help_type) |
222
|
|
|
|
223
|
|
|
if format: |
224
|
|
|
if not format.lower() in ("html", "rnc", "text", "xml"): |
225
|
|
|
raise InvalidArgument( |
226
|
|
|
"help format Argument must be one of html, rnc, text or " |
227
|
|
|
"xml" |
228
|
|
|
) |
229
|
|
|
|
230
|
|
|
cmd.set_attribute("format", format) |
231
|
|
|
|
232
|
|
|
return self._send_xml_command(cmd) |
233
|
|
|
|
234
|
|
|
def modify_auth(self, group_name: str, auth_conf_settings: dict) -> Any: |
235
|
|
|
"""Modifies an existing auth. |
236
|
|
|
|
237
|
|
|
Arguments: |
238
|
|
|
group_name: Name of the group to be modified. |
239
|
|
|
auth_conf_settings: The new auth config. |
240
|
|
|
|
241
|
|
|
Returns: |
242
|
|
|
The response. See :py:meth:`send_command` for details. |
243
|
|
|
""" |
244
|
|
|
if not group_name: |
245
|
|
|
raise RequiredArgument( |
246
|
|
|
function=self.modify_auth.__name__, argument='group_name' |
247
|
|
|
) |
248
|
|
|
if not auth_conf_settings: |
249
|
|
|
raise RequiredArgument( |
250
|
|
|
function=self.modify_auth.__name__, |
251
|
|
|
argument='auth_conf_settings', |
252
|
|
|
) |
253
|
|
|
cmd = XmlCommand("modify_auth") |
254
|
|
|
_xmlgroup = cmd.add_element("group", attrs={"name": str(group_name)}) |
255
|
|
|
|
256
|
|
|
for key, value in auth_conf_settings.items(): |
257
|
|
|
_xmlauthconf = _xmlgroup.add_element("auth_conf_setting") |
258
|
|
|
_xmlauthconf.add_element("key", key) |
259
|
|
|
_xmlauthconf.add_element("value", value) |
260
|
|
|
|
261
|
|
|
return self._send_xml_command(cmd) |
262
|
|
|
|
263
|
|
|
def restore(self, entity_id: str) -> Any: |
264
|
|
|
"""Restore an entity from the trashcan |
265
|
|
|
|
266
|
|
|
Arguments: |
267
|
|
|
entity_id: ID of the entity to be restored from the trashcan |
268
|
|
|
|
269
|
|
|
Returns: |
270
|
|
|
The response. See :py:meth:`send_command` for details. |
271
|
|
|
""" |
272
|
|
|
if not entity_id: |
273
|
|
|
raise RequiredArgument( |
274
|
|
|
function=self.restore.__name__, argument='entity_id' |
275
|
|
|
) |
276
|
|
|
|
277
|
|
|
cmd = XmlCommand("restore") |
278
|
|
|
cmd.set_attribute("id", entity_id) |
279
|
|
|
|
280
|
|
|
return self._send_xml_command(cmd) |
281
|
|
|
|
282
|
|
|
def sync_cert(self) -> Any: |
283
|
|
|
"""Request a synchronization with the CERT feed service |
284
|
|
|
|
285
|
|
|
Returns: |
286
|
|
|
The response. See :py:meth:`send_command` for details. |
287
|
|
|
""" |
288
|
|
|
return self._send_xml_command(XmlCommand("sync_cert")) |
289
|
|
|
|
290
|
|
|
def sync_scap(self) -> Any: |
291
|
|
|
"""Request a synchronization with the SCAP feed service |
292
|
|
|
|
293
|
|
|
Returns: |
294
|
|
|
The response. See :py:meth:`send_command` for details. |
295
|
|
|
""" |
296
|
|
|
return self._send_xml_command(XmlCommand("sync_scap")) |
297
|
|
|
|