Cleanup and improve the openvasd API

Drop the safe argument from all methods and replace it with a instance
variable. Improve the types and docstrings of all methods.

Author: bjoernricks

gvm/protocols/http/openvasd/_api.py

@@ -0,0 +1,23 @@
+# SPDX-FileCopyrightText: 2025 Greenbone AG
+#
+# SPDX-License-Identifier: GPL-3.0-or-later
+
+
+import httpx
+
+
+class OpenvasdAPI:
+    def __init__(
+        self, client: httpx.Client, *, suppress_exceptions: bool = False
+    ):
+        """
+        Initialize the OpenvasdAPI entry point.
+
+        Args:
+            client: An initialized `httpx.Client` configured for communicating
+                    with the openvasd server.
+            suppress_exceptions: If True, suppress exceptions and return structured error responses.
+                Default is False, which means exceptions will be raised.
+        """
+        self._client = client
+        self._suppress_exceptions = suppress_exceptions

gvm/protocols/http/openvasd/_health.py

@@ -8,8 +8,10 @@
 
 import httpx
 
+from ._api import OpenvasdAPI
 
-class HealthAPI:
+
+class HealthAPI(OpenvasdAPI):
     """
     Provides access to the openvasd /health endpoints, which expose the
     operational state of the scanner.
@@ -18,28 +20,16 @@ class HealthAPI:
     if the server returns an error response (4xx or 5xx).
     """
 
-    def __init__(self, client: httpx.Client):
-        """
-        Create a new HealthAPI instance.
-
-        Args:
-            client: An initialized `httpx.Client` configured for communicating
-                    with the openvasd server.
-        """
-        self._client = client
-
-    def get_alive(self, safe: bool = False) -> int:
+    def get_alive(self) -> int:
         """
         Check if the scanner process is alive.
 
-        Args:
-            safe: If True, suppress exceptions and return structured error responses.
-
         Returns:
             HTTP status code (e.g., 200 if alive).
 
         Raises:
-            httpx.HTTPStatusError: If the server response indicates failure and safe is False.
+            httpx.HTTPStatusError: If the server response indicates failure and
+                exceptions are not suppressed.
 
         See: GET /health/alive in the openvasd API documentation.
         """
@@ -48,22 +38,20 @@ def get_alive(self, safe: bool = False) -> int:
             response.raise_for_status()
             return response.status_code
         except httpx.HTTPStatusError as e:
-            if safe:
+            if self._suppress_exceptions:
                 return e.response.status_code
             raise
 
-    def get_ready(self, safe: bool = False) -> int:
+    def get_ready(self) -> int:
         """
         Check if the scanner is ready to accept requests (e.g., feed loaded).
 
-        Args:
-            safe: If True, suppress exceptions and return structured error responses.
-
         Returns:
             HTTP status code (e.g., 200 if ready).
 
         Raises:
-            httpx.HTTPStatusError: If the server response indicates failure and safe is False.
+            httpx.HTTPStatusError: If the server response indicates failure and
+                exceptions are not suppressed.
 
         See: GET /health/ready in the openvasd API documentation.
         """
@@ -72,22 +60,20 @@ def get_ready(self, safe: bool = False) -> int:
             response.raise_for_status()
             return response.status_code
         except httpx.HTTPStatusError as e:
-            if safe:
+            if self._suppress_exceptions:
                 return e.response.status_code
             raise
 
-    def get_started(self, safe: bool = False) -> int:
+    def get_started(self) -> int:
         """
         Check if the scanner has fully started.
 
-        Args:
-            safe: If True, suppress exceptions and return structured error responses.
-
         Returns:
             HTTP status code (e.g., 200 if started).
 
         Raises:
-            httpx.HTTPStatusError: If the server response indicates failure and safe is False.
+            httpx.HTTPStatusError: If the server response indicates failure and
+                exceptions are not suppressed.
 
         See: GET /health/started in the openvasd API documentation.
         """
@@ -96,6 +82,6 @@ def get_started(self, safe: bool = False) -> int:
             response.raise_for_status()
             return response.status_code
         except httpx.HTTPStatusError as e:
-            if safe:
+            if self._suppress_exceptions:
                 return e.response.status_code
             raise

gvm/protocols/http/openvasd/_metadata.py

@@ -6,10 +6,14 @@
 API wrapper for retrieving metadata from the openvasd HTTP API using HEAD requests.
 """
 
+from typing import Union
+
 import httpx
 
+from ._api import OpenvasdAPI
+
 
-class MetadataAPI:
+class MetadataAPI(OpenvasdAPI):
     """
     Provides access to metadata endpoints exposed by the openvasd server
     using lightweight HTTP HEAD requests.
@@ -23,35 +27,23 @@ class MetadataAPI:
     is handled gracefully.
     """
 
-    def __init__(self, client: httpx.Client):
-        """
-        Initialize a MetadataAPI instance.
-
-        Args:
-            client: An `httpx.Client` configured to communicate with the openvasd server.
-        """
-        self._client = client
-
-    def get(self, safe: bool = False) -> dict:
+    def get(self) -> dict[str, Union[str, int]]:
         """
         Perform a HEAD request to `/` to retrieve top-level API metadata.
 
-        Args:
-            safe: If True, suppress exceptions and return structured error responses.
-
         Returns:
             A dictionary containing:
 
               - "api-version"
               - "feed-version"
               - "authentication"
 
-            Or if safe=True and error occurs:
+            Or if exceptions are suppressed and error occurs:
 
               - {"error": str, "status_code": int}
 
         Raises:
-            httpx.HTTPStatusError: For non-401 HTTP errors if safe=False.
+            httpx.HTTPStatusError: For non-401 HTTP errors if exceptions are not suppressed.
 
         See: HEAD / in the openvasd API documentation.
         """
@@ -64,17 +56,14 @@ def get(self, safe: bool = False) -> dict:
                 "authentication": response.headers.get("authentication"),
             }
         except httpx.HTTPStatusError as e:
-            if safe:
+            if self._suppress_exceptions:
                 return {"error": str(e), "status_code": e.response.status_code}
             raise
 
-    def get_scans(self, safe: bool = False) -> dict:
+    def get_scans(self) -> dict[str, Union[str, int]]:
         """
          Perform a HEAD request to `/scans` to retrieve scan endpoint metadata.
 
-         Args:
-             safe: If True, suppress exceptions and return structured error responses.
-
         Returns:
              A dictionary containing:
 
@@ -87,7 +76,7 @@ def get_scans(self, safe: bool = False) -> dict:
                - {"error": str, "status_code": int}
 
          Raises:
-             httpx.HTTPStatusError: For non-401 HTTP errors if safe=False.
+             httpx.HTTPStatusError: For non-401 HTTP errors if exceptions are not suppressed.
 
          See: HEAD /scans in the openvasd API documentation.
         """
@@ -100,6 +89,6 @@ def get_scans(self, safe: bool = False) -> dict:
                 "authentication": response.headers.get("authentication"),
             }
         except httpx.HTTPStatusError as e:
-            if safe:
+            if self._suppress_exceptions:
                 return {"error": str(e), "status_code": e.response.status_code}
             raise

gvm/protocols/http/openvasd/_notus.py

@@ -6,36 +6,25 @@
 API wrapper for interacting with the Notus component of the openvasd HTTP API.
 """
 
-import urllib.parse
-from typing import List
+import urllib
 
 import httpx
 
+from ._api import OpenvasdAPI
 
-class NotusAPI:
+
+class NotusAPI(OpenvasdAPI):
     """
     Provides access to the Notus-related endpoints of the openvasd HTTP API.
 
     This includes retrieving supported operating systems and triggering
     package-based vulnerability scans for a specific OS.
     """
 
-    def __init__(self, client: httpx.Client):
-        """
-        Initialize a NotusAPI instance.
-
-        Args:
-            client: An `httpx.Client` configured to communicate with the openvasd server.
-        """
-        self._client = client
-
-    def get_os_list(self, safe: bool = False) -> httpx.Response:
+    def get_os_list(self) -> httpx.Response:
         """
         Retrieve the list of supported operating systems from the Notus service.
 
-        Args:
-            safe: If True, return error info on failure instead of raising.
-
         Returns:
             The full `httpx.Response` on success.
 
@@ -46,20 +35,21 @@ def get_os_list(self, safe: bool = False) -> httpx.Response:
             response.raise_for_status()
             return response
         except httpx.HTTPStatusError as e:
-            if safe:
+            if self._suppress_exceptions:
                 return e.response
             raise
 
     def run_scan(
-        self, os: str, package_list: List[str], safe: bool = False
+        self,
+        os: str,
+        package_list: list[str],
     ) -> httpx.Response:
         """
         Trigger a Notus scan for a given OS and list of packages.
 
         Args:
             os: Operating system name (e.g., "debian", "alpine").
             package_list: List of package names to evaluate for vulnerabilities.
-            safe: If True, return error info on failure instead of raising.
 
         Returns:
             The full `httpx.Response` on success.
@@ -74,6 +64,6 @@ def run_scan(
             response.raise_for_status()
             return response
         except httpx.HTTPStatusError as e:
-            if safe:
+            if self._suppress_exceptions:
                 return e.response
             raise

gvm/protocols/http/openvasd/_openvasd1.py

@@ -39,6 +39,7 @@ def __init__(
         client_cert_paths: Optional[
             Union[StrOrPathLike, Tuple[StrOrPathLike, StrOrPathLike]]
         ] = None,
+        suppress_exceptions: bool = False,
     ):
         """
         Initialize the OpenvasdHttpApiV1 entry point.
@@ -49,10 +50,8 @@ def __init__(
             api_key: Optional API key to be used for authentication.
             server_ca_path: Path to the server CA certificate (for HTTPS/mTLS).
             client_cert_paths: Path to client certificate or (cert, key) tuple for mTLS.
-
-        Behavior:
-            - Sets up an underlying `httpx.Client` using `OpenvasdClient`.
-            - Initializes sub-API modules with the shared client instance.
+            suppress_exceptions: If True, suppress exceptions and return structured error
+                responses. Default is False, which means exceptions will be raised.
         """
         self._client = create_openvasd_http_client(
             host_name=host_name,
@@ -63,8 +62,67 @@ def __init__(
         )
 
         # Sub-API modules
-        self.health = HealthAPI(self._client)
-        self.metadata = MetadataAPI(self._client)
-        self.notus = NotusAPI(self._client)
-        self.scans = ScansAPI(self._client)
-        self.vts = VtsAPI(self._client)
+        self.__health = HealthAPI(
+            self._client, suppress_exceptions=suppress_exceptions
+        )
+        self.__metadata = MetadataAPI(
+            self._client, suppress_exceptions=suppress_exceptions
+        )
+        self.__notus = NotusAPI(
+            self._client, suppress_exceptions=suppress_exceptions
+        )
+        self.__scans = ScansAPI(
+            self._client, suppress_exceptions=suppress_exceptions
+        )
+        self.__vts = VtsAPI(
+            self._client, suppress_exceptions=suppress_exceptions
+        )
+
+    @property
+    def health(self) -> HealthAPI:
+        """
+        Access the health API module.
+
+        Provides methods to check the health status of the openvasd service.
+        """
+        return self.__health
+
+    @property
+    def metadata(self) -> MetadataAPI:
+        """
+        Access the metadata API module.
+
+        Provides methods to retrieve metadata about the openvasd service,
+        including version and feed information.
+        """
+        return self.__metadata
+
+    @property
+    def notus(self) -> NotusAPI:
+        """
+        Access the Notus API module.
+
+        Provides methods to interact with the Notus service for package-based
+        vulnerability scanning.
+        """
+        return self.__notus
+
+    @property
+    def scans(self) -> ScansAPI:
+        """
+        Access the scans API module.
+
+        Provides methods to manage and interact with vulnerability scans,
+        including starting, stopping, and retrieving scan results.
+        """
+        return self.__scans
+
+    @property
+    def vts(self) -> VtsAPI:
+        """
+        Access the VTS API module.
+
+        Provides methods to manage and interact with Vulnerability Tests
+        (VTs) for vulnerability assessment.
+        """
+        return self.__vts