Add: Draft new implementation for GMP protocol classes base on the co…

…re module

Use the core module to re-implement the current GMP protocol classes.

gvm/protocols/gmp/__init__.py

@@ -0,0 +1,12 @@
+# SPDX-FileCopyrightText: 2024 Greenbone AG
+#
+# SPDX-License-Identifier: GPL-3.0-or-later
+
+from ._gmp import GMP
+
+Gmp = GMP  # for backwards compatibility
+
+__all__ = (
+    "GMP",
+    "Gmp",
+)

gvm/protocols/gmp.py → gvm/protocols/gmp/_gmp.py

@@ -1,29 +1,23 @@
 # SPDX-FileCopyrightText: 2019-2024 Greenbone AG
 #
 # SPDX-License-Identifier: GPL-3.0-or-later
-#
 
-"""
-Module for communication with gvmd
-"""
 from types import TracebackType
-from typing import Any, Callable, Optional, Type, Union
+from typing import Callable, Optional, Type, Union
 
+from gvm.connections import GvmConnection
 from gvm.errors import GvmError
-from gvm.protocols.base import GvmConnection, GvmProtocol
-from gvm.protocols.gmpv208 import Gmp as Gmpv208
-from gvm.protocols.gmpv214 import Gmp as Gmpv214
-from gvm.protocols.gmpv224 import Gmp as Gmpv224
-from gvm.protocols.gmpv225 import Gmp as Gmpv225
-from gvm.transforms import EtreeCheckCommandTransform
-from gvm.xml import XmlCommand
 
-SUPPORTED_GMP_VERSIONS = Union[  # pylint: disable=invalid-name
-    Gmpv208, Gmpv214, Gmpv224, Gmpv225
-]
+from .._protocol import GvmProtocol, T, str_transform
+from ._gmp224 import GMPv224
+from ._gmp225 import GMPv225
+from .core import Response
+from .core.requests import Version
+
+SUPPORTED_GMP_VERSIONS = Union[GMPv224, GMPv225]
 
 
-class Gmp(GvmProtocol):
+class GMP(GvmProtocol[T]):
     """Dynamically select supported GMP protocol of the remote manager daemon.
 
     Must be used as a `Context Manager
@@ -59,19 +53,18 @@ def __init__(
         self,
         connection: GvmConnection,
         *,
-        transform: Optional[Callable[[str], Any]] = None,
+        transform: Callable[[Response], T] = str_transform,  # type: ignore[assignment] # this should work with mypy 1.9.0 without an ignore
     ):
-        super().__init__(connection, transform=EtreeCheckCommandTransform())
-        self._gmp_transform = transform
+        super().__init__(connection, transform=transform)
 
     def determine_remote_gmp_version(self) -> str:
         """Determine the supported GMP version of the remote daemon"""
         self.connect()
-        resp = self._send_xml_command(XmlCommand("get_version"))
+        resp = self._send_command(Version.get_version())
         self.disconnect()
 
-        version_el = resp.find("version")
-        if version_el is None:
+        version_el = resp.xml().find("version")
+        if version_el is None or not version_el.text:
             raise GvmError(
                 "Invalid response from manager daemon while requesting the "
                 "version information."
@@ -86,21 +79,19 @@ def determine_supported_gmp(self) -> SUPPORTED_GMP_VERSIONS:
         version_str = self.determine_remote_gmp_version().split(".", 1)
         major_version = int(version_str[0])
         minor_version = int(version_str[1])
-        if major_version == 20:
-            gmp_class = Gmpv208
-        elif major_version == 21 and minor_version == 4:
-            gmp_class = Gmpv214
-        elif major_version == 22 and minor_version == 4:
-            gmp_class = Gmpv224
+        # if major_version == 22 and minor_version == 4:
+        # gmp_class = Gmpv224
+        if major_version == 22 and minor_version == 4:
+            gmp_class = GMPv224
         elif major_version == 22 and minor_version == 5:
-            gmp_class = Gmpv225
+            gmp_class = GMPv225
         else:
             raise GvmError(
                 "Remote manager daemon uses an unsupported version of GMP. "
                 f"The GMP version was {major_version}.{minor_version}"
             )
 
-        return gmp_class(self._connection, transform=self._gmp_transform)
+        return gmp_class(self._connection, transform=self._transform_callable)
 
     def __enter__(self):
         self._gmp = self.determine_supported_gmp()
@@ -114,6 +105,6 @@ def __exit__(
         exc_type: Optional[Type[BaseException]],
         exc_value: Optional[BaseException],
         traceback: Optional[TracebackType],
-    ) -> Any:
+    ) -> None:
         self._gmp.disconnect()
         self._gmp = None

gvm/protocols/gmp/_gmp224.py

@@ -0,0 +1,149 @@
+# SPDX-FileCopyrightText: 2024 Greenbone AG
+#
+# SPDX-License-Identifier: GPL-3.0-or-later
+
+from typing import Optional, Union
+
+from .._protocol import GvmProtocol, T
+from .core.requests import (
+    Authentication,
+    PortList,
+    PortRangeType,
+    Version,
+)
+
+
+class GMPv224(GvmProtocol[T]):
+    _authenticated = False
+
+    def is_authenticated(self) -> bool:
+        """Checks if the user is authenticated
+
+        If the user is authenticated privileged GMP commands like get_tasks
+        may be send to gvmd.
+
+        Returns:
+            bool: True if an authenticated connection to gvmd has been
+            established.
+        """
+        return self._authenticated
+
+    def authenticate(self, username: str, password: str) -> T:
+        """Authenticate to gvmd.
+
+        The generated authenticate command will be send to server.
+        Afterwards the response is read, transformed and returned.
+
+        Arguments:
+            username: Username
+            password: Password
+        """
+        response = self._send_command(
+            Authentication.authenticate(username=username, password=password)
+        )
+
+        if response.is_success:
+            self._authenticated = True
+
+        return self._transform(response)
+
+    def describe_auth(self) -> T:
+        """Describe authentication methods
+
+        Returns a list of all used authentication methods if such a list is
+        available.
+
+        Returns:
+            The response. See :py:meth:`send_command` for details.
+        """
+        return self._send_and_transform_command(Authentication.describe_auth())
+
+    def modify_auth(
+        self, group_name: str, auth_conf_settings: dict[str, str]
+    ) -> T:
+        """Modifies an existing auth.
+
+        Arguments:
+            group_name: Name of the group to be modified.
+            auth_conf_settings: The new auth config.
+        """
+        return self._send_and_transform_command(
+            Authentication.modify_auth(group_name, auth_conf_settings)
+        )
+
+    def get_version(self) -> T:
+        return self._send_and_transform_command(Version.get_version())
+
+    def clone_port_list(self, port_list_id: str) -> T:
+        return self._send_and_transform_command(
+            PortList.clone_port_list(port_list_id)
+        )
+
+    def create_port_list(
+        self, name: str, port_range: str, *, comment: Optional[str] = None
+    ) -> T:
+        return self._send_and_transform_command(
+            PortList.create_port_list(name, port_range, comment=comment)
+        )
+
+    def create_port_range(
+        self,
+        port_list_id: str,
+        start: int,
+        end: int,
+        port_range_type: Union[str, PortRangeType],
+        *,
+        comment: Optional[str] = None,
+    ) -> T:
+        return self._send_and_transform_command(
+            PortList.create_port_range(
+                port_list_id, start, end, port_range_type, comment=comment
+            )
+        )
+
+    def delete_port_list(
+        self, port_list_id: str, *, ultimate: bool = False
+    ) -> T:
+        return self._send_and_transform_command(
+            PortList.delete_port_list(port_list_id, ultimate=ultimate)
+        )
+
+    def delete_port_range(self, port_range_id: str) -> T:
+        return self._send_and_transform_command(
+            PortList.delete_port_range(port_range_id)
+        )
+
+    def get_port_lists(
+        self,
+        *,
+        filter_string: Optional[str] = None,
+        filter_id: Optional[str] = None,
+        details: Optional[bool] = None,
+        targets: Optional[bool] = None,
+        trash: Optional[bool] = None,
+    ) -> T:
+        return self._send_and_transform_command(
+            PortList.get_port_lists(
+                filter_string=filter_string,
+                filter_id=filter_id,
+                details=details,
+                targets=targets,
+                trash=trash,
+            )
+        )
+
+    def get_port_list(self, port_list_id: str) -> T:
+        return self._send_and_transform_command(
+            PortList.get_port_list(port_list_id)
+        )
+
+    def modify_port_list(
+        self,
+        port_list_id: str,
+        *,
+        comment: Optional[str] = None,
+        name: Optional[str] = None,
+    ) -> T:
+        return self._send_and_transform_command(
+            PortList.modify_port_list(port_list_id, comment=comment, name=name)
+        )

gvm/protocols/gmp/_gmp225.py

@@ -0,0 +1,55 @@
+# SPDX-FileCopyrightText: 2024 Greenbone AG
+#
+# SPDX-License-Identifier: GPL-3.0-or-later
+
+from typing import Optional
+
+from .._protocol import T
+from ._gmp224 import GMPv224
+from .core.requests import (
+    ResourceNames,
+    ResourceType,
+)
+
+
+class GMPv225(GMPv224[T]):
+    def get_resource_names(
+        self,
+        resource_type: ResourceType,
+        *,
+        filter_string: Optional[str] = None,
+    ) -> T:
+        """Request a list of resource names and IDs
+
+        Arguments:
+            resource_type: Type must be either ALERT, CERT_BUND_ADV,
+                CONFIG, CPE, CREDENTIAL, CVE, DFN_CERT_ADV, FILTER,
+                GROUP, HOST, NOTE, NVT, OS, OVERRIDE, PERMISSION,
+                PORT_LIST, REPORT_FORMAT, REPORT, RESULT, ROLE,
+                SCANNER, SCHEDULE, TARGET, TASK, TLS_CERTIFICATE
+                or USER
+            filter_string: Filter term to use for the query
+        """
+        return self._send_and_transform_command(
+            ResourceNames.get_resource_names(
+                resource_type, filter_string=filter_string
+            )
+        )
+
+    def get_resource_name(
+        self, resource_id: str, resource_type: ResourceType
+    ) -> T:
+        """Request a single resource name
+
+        Arguments:
+            resource_id: ID of an existing resource
+            resource_type: Type must be either ALERT, CERT_BUND_ADV,
+                CONFIG, CPE, CREDENTIAL, CVE, DFN_CERT_ADV, FILTER,
+                GROUP, HOST, NOTE, NVT, OS, OVERRIDE, PERMISSION,
+                PORT_LIST, REPORT_FORMAT, REPORT, RESULT, ROLE,
+                SCANNER, SCHEDULE, TARGET, TASK, TLS_CERTIFICATE
+                or USER
+        """
+        return self._send_and_transform_command(
+            ResourceNames.get_resource_name(resource_id, resource_type)
+        )