Add: Support for the openvasd HTTP API

A new subpackage for sending requests to HTTP APIs has been added which
also includes the first version of the openvasd API.

Author: timopollmeier

gvm/http/__init__.py

@@ -0,0 +1,3 @@
+# SPDX-FileCopyrightText: 2025 Greenbone AG
+#
+# SPDX-License-Identifier: GPL-3.0-or-later

gvm/http/core/__init__.py

@@ -0,0 +1,3 @@
+# SPDX-FileCopyrightText: 2025 Greenbone AG
+#
+# SPDX-License-Identifier: GPL-3.0-or-later

gvm/http/core/api.py

@@ -0,0 +1,28 @@
+# SPDX-FileCopyrightText: 2025 Greenbone AG
+#
+# SPDX-License-Identifier: GPL-3.0-or-later
+
+from typing import Optional
+
+from gvm.http.core.connector import HttpApiConnector
+
+
+class GvmHttpApi:
+    """
+    Base class for HTTP-based GVM APIs.
+    """
+
+    def __init__(self, connector: HttpApiConnector, *, api_key: Optional[str] = None):
+        """
+        Create a new generic GVM HTTP API instance.
+
+        Args:
+            connector: The connector handling the HTTP(S) connection
+            api_key: Optional API key for authentication
+        """
+
+        "The connector handling the HTTP(S) connection"
+        self._connector: HttpApiConnector = connector
+
+        "Optional API key for authentication"
+        self._api_key: Optional[str] = api_key

gvm/http/core/connector.py

@@ -0,0 +1,154 @@
+# SPDX-FileCopyrightText: 2025 Greenbone AG
+#
+# SPDX-License-Identifier: GPL-3.0-or-later
+
+import urllib.parse
+from typing import Optional, Tuple, Dict, Any
+
+from requests import Session
+
+from gvm.http.core.response import HttpResponse
+
+
+def url_join(base: str, rel_path: str) -> str:
+    """
+    Combines a base URL and a relative path into one URL.
+
+    Unlike `urrlib.parse.urljoin` the base path will always be the parent of the relative path as if it
+    ends with "/".
+    """
+    if base.endswith("/"):
+        return urllib.parse.urljoin(base, rel_path)
+    else:
+        return urllib.parse.urljoin(base + "/", rel_path)
+
+
+class HttpApiConnector:
+    """
+    Class for connecting to HTTP based API servers, sending requests and receiving the responses.
+    """
+
+    @classmethod
+    def _new_session(cls):
+        """
+        Creates a new session
+        """
+        return Session()
+
+    def __init__(
+            self,
+            base_url: str,
+            *,
+            server_ca_path: Optional[str] = None,
+            client_cert_paths: Optional[str | Tuple[str]] = None,
+    ):
+        """
+        Create a new HTTP API Connector.
+
+        Args:
+            base_url: The base server URL to which request-specific paths will be appended for the requests
+            server_ca_path: Optional path to a CA certificate for verifying the server.
+                If none is given, server verification is disabled.
+            client_cert_paths: Optional path to a client private key and certificate for authentication.
+                Can be a combined key and certificate file or a tuple containing separate files.
+                The key must not be encrypted.
+        """
+
+        self.base_url = base_url
+        "The base server URL to which request-specific paths will be appended for the requests"
+
+        self._session = self._new_session()
+        "Internal session handling the HTTP requests"
+        if server_ca_path:
+            self._session.verify = server_ca_path
+        if client_cert_paths:
+            self._session.cert = client_cert_paths
+
+    def update_headers(self, new_headers: Dict[str, str]) -> None:
+        """
+        Updates the headers sent with each request, e.g. for passing an API key
+
+        Args:
+            new_headers: Dict containing the new headers
+        """
+        self._session.headers.update(new_headers)
+
+    def delete(
+        self,
+        rel_path: str,
+        *,
+        raise_for_status: bool = True,
+        params: Optional[Dict[str,str]] = None,
+        headers: Optional[Dict[str,str]] = None,
+    ) -> HttpResponse:
+        """
+        Sends a ``DELETE`` request and returns the response.
+
+        Args:
+            rel_path: The relative path for the request
+            raise_for_status: Whether to raise an error if response has a non-success HTTP status code
+            params: Optional dict of URL-encoded parameters
+            headers: Optional additional headers added to the request
+
+        Return:
+            The HTTP response.
+        """
+        url = url_join(self.base_url, rel_path)
+        r = self._session.delete(url, params=params, headers=headers)
+        if raise_for_status:
+            r.raise_for_status()
+        return HttpResponse.from_requests_lib(r)
+
+    def get(
+        self,
+        rel_path: str,
+        *,
+        raise_for_status: bool = True,
+        params: Optional[Dict[str,str]] = None,
+        headers: Optional[Dict[str,str]] = None,
+    ) -> HttpResponse:
+        """
+        Sends a ``GET`` request and returns the response.
+
+        Args:
+            rel_path: The relative path for the request
+            raise_for_status: Whether to raise an error if response has a non-success HTTP status code
+            params: Optional dict of URL-encoded parameters
+            headers: Optional additional headers added to the request
+
+        Return:
+            The HTTP response.
+        """
+        url = url_join(self.base_url, rel_path)
+        r = self._session.get(url, params=params, headers=headers)
+        if raise_for_status:
+            r.raise_for_status()
+        return HttpResponse.from_requests_lib(r)
+
+    def post_json(
+        self,
+        rel_path: str,
+        json: Any,
+        *,
+        raise_for_status: bool = True,
+        params: Optional[Dict[str, str]] = None,
+        headers: Optional[Dict[str, str]] = None,
+    ) -> HttpResponse:
+        """
+        Sends a ``POST`` request, using the given JSON-compatible object as the request body, and returns the response.
+
+        Args:
+            rel_path: The relative path for the request
+            json: The object to use as the request body.
+            raise_for_status: Whether to raise an error if response has a non-success HTTP status code
+            params: Optional dict of URL-encoded parameters
+            headers: Optional additional headers added to the request
+
+        Return:
+            The HTTP response.
+        """
+        url = url_join(self.base_url, rel_path)
+        r = self._session.post(url, json=json, params=params, headers=headers)
+        if raise_for_status:
+            r.raise_for_status()
+        return HttpResponse.from_requests_lib(r)

gvm/http/core/headers.py

@@ -0,0 +1,53 @@
+# SPDX-FileCopyrightText: 2025 Greenbone AG
+#
+# SPDX-License-Identifier: GPL-3.0-or-later
+from dataclasses import dataclass
+from typing import Self, Dict, Optional
+
+
+@dataclass
+class ContentType:
+    """
+    Class representing the content type of a HTTP response.
+    """
+
+    media_type: str
+    "The MIME media type, e.g. \"application/json\""
+
+    params: Dict[str,str]
+    "Dictionary of parameters in the content type header"
+
+    charset: Optional[str]
+    "The charset parameter in the content type header if it is set"
+
+    @classmethod
+    def from_string(
+        cls,
+        header_string: str,
+        fallback_media_type: Optional[str] = "application/octet-stream"
+    ) -> Self:
+        """
+        Parse the content of content type header into a ContentType object.
+
+        Args:
+            header_string: The string to parse
+            fallback_media_type: The media type to use if the `header_string` is `None` or empty.
+        """
+        media_type = fallback_media_type
+        params = {}
+        charset = None
+
+        if header_string:
+            parts = header_string.split(";")
+            media_type = parts[0].strip()
+            for param in parts[1:]:
+                param = param.strip()
+                if "=" in param:
+                    key, value = map(lambda x: x.strip(), param.split("=", 1))
+                    params[key] = value
+                    if key == 'charset':
+                        charset = value
+                else:
+                    params[param] = True
+
+        return ContentType(media_type=media_type, params=params, charset=charset)

gvm/http/core/response.py

@@ -0,0 +1,45 @@
+# SPDX-FileCopyrightText: 2025 Greenbone AG
+#
+# SPDX-License-Identifier: GPL-3.0-or-later
+from dataclasses import dataclass
+from typing import Any, Dict, Self, Optional
+from requests import Request, Response
+
+from gvm.http.core.headers import ContentType
+
+
+@dataclass
+class HttpResponse:
+    """
+    Class representing an HTTP response.
+    """
+    body: Any
+    status: int
+    headers: Dict[str, str]
+    content_type: Optional[ContentType]
+
+    @classmethod
+    def from_requests_lib(cls, r: Response) -> Self:
+        """
+        Creates a new HTTP response object from a Request object created by the "Requests" library.
+
+        Args:
+            r: The request object to convert.
+
+        Return:
+            A HttpResponse object representing the response.
+
+            An empty body is represented by None.
+            If the content-type header in the response is set to 'application/json'.
+            A non-empty body will be parsed accordingly.
+        """
+        ct = ContentType.from_string(r.headers.get('content-type'))
+        body = r.content
+
+        if r.content == b'':
+            body = None
+        elif ct is not None:
+            if ct.media_type.lower() == 'application/json':
+                body = r.json()
+
+        return HttpResponse(body, r.status_code, r.headers, ct)

gvm/http/openvasd/__init__.py

@@ -0,0 +1,3 @@
+# SPDX-FileCopyrightText: 2025 Greenbone AG
+#
+# SPDX-License-Identifier: GPL-3.0-or-later