diff --git a/gvm/protocols/http/openvasd/_client.py b/gvm/protocols/http/openvasd/_client.py index 167e5edb8..e4168f797 100644 --- a/gvm/protocols/http/openvasd/_client.py +++ b/gvm/protocols/http/openvasd/_client.py @@ -3,7 +3,10 @@ # SPDX-License-Identifier: GPL-3.0-or-later """ -http client for initializing a connection to the openvasd HTTP API using optional mTLS authentication. +http client for initializing a connection to the openvasd HTTP API. + +HTTPS with mTLS is used by default. Plain HTTP can be enabled explicitly +with `insecure_http=True`. """ import ssl @@ -22,11 +25,14 @@ def create_openvasd_http_client( client_cert_paths: StrOrPathLike | tuple[StrOrPathLike, StrOrPathLike] | None = None, + insecure_http: bool = False, port: int = 3000, ) -> Client: """ - Create a `httpx.Client` configured for mTLS-secured or API KEY access - to an openvasd HTTP API instance. + Create a `httpx.Client` configured for the OpenVASD HTTP API. + + mTLS is used by default. Set `insecure_http=True` to use plain HTTP. + In this case, an API key can be used for authorization. Args: host_name: Hostname or IP of the OpenVASD server (e.g., "localhost"). @@ -34,46 +40,54 @@ def create_openvasd_http_client( server_ca_path: Path to the server's CA certificate (for verifying the server). client_cert_paths: Path to the client certificate (str) or a tuple of (cert_path, key_path) for mTLS authentication. + insecure_http: If True, disables SSL verification and uses HTTP instead of HTTPS. port: The port to connect to (default: 3000). Behavior: - - If both `server_ca_path` and `client_cert_paths` are set, an mTLS connection - is established using an SSLContext. - - If not, `verify` is set to False (insecure), and HTTP is used instead of HTTPS. - HTTP connection needs api_key for authorization. + - If `insecure_http=True`, HTTP is used instead of HTTPS. + An API key can be used for authorization. + - If `insecure_http=False` (default), HTTPS with mTLS is used. + Both `server_ca_path` and `client_cert_paths` are required. + + Raises: + ValueError: If `insecure_http=False` and either + `server_ca_path` or `client_cert_paths` is missing. """ headers = {} + if api_key: + headers["X-API-KEY"] = api_key - context: ssl.SSLContext | None = None - - # Prepare mTLS SSL context if needed - if client_cert_paths and server_ca_path: - context = ssl.create_default_context( - ssl.Purpose.SERVER_AUTH, cafile=server_ca_path + if insecure_http: + return Client( + base_url=f"http://{host_name}:{port}", + headers=headers, + http2=True, + timeout=10.0, ) - if isinstance(client_cert_paths, tuple): - context.load_cert_chain( - certfile=client_cert_paths[0], keyfile=client_cert_paths[1] - ) - else: - context.load_cert_chain(certfile=client_cert_paths) - - context.check_hostname = False - context.verify_mode = ssl.CERT_REQUIRED - # Set verify based on context presence - verify: bool | ssl.SSLContext = context if context else False - - if api_key: - headers["X-API-KEY"] = api_key + if not server_ca_path or not client_cert_paths: + raise ValueError( + "Both server_ca_path and client_cert_paths must be provided " + "when insecure_http is False." + ) + # Prepare mTLS SSL context + context = ssl.create_default_context( + ssl.Purpose.SERVER_AUTH, cafile=server_ca_path + ) + if isinstance(client_cert_paths, tuple): + context.load_cert_chain( + certfile=client_cert_paths[0], keyfile=client_cert_paths[1] + ) + else: + context.load_cert_chain(certfile=client_cert_paths) - protocol = "https" if context else "http" - base_url = f"{protocol}://{host_name}:{port}" + context.check_hostname = False + context.verify_mode = ssl.CERT_REQUIRED return Client( - base_url=base_url, + base_url=f"https://{host_name}:{port}", headers=headers, - verify=verify, + verify=context, http2=True, timeout=10.0, ) diff --git a/gvm/protocols/http/openvasd/_openvasd1.py b/gvm/protocols/http/openvasd/_openvasd1.py index 8b28c8a36..49f9d04ec 100644 --- a/gvm/protocols/http/openvasd/_openvasd1.py +++ b/gvm/protocols/http/openvasd/_openvasd1.py @@ -37,6 +37,7 @@ def __init__( client_cert_paths: StrOrPathLike | tuple[StrOrPathLike, StrOrPathLike] | None = None, + insecure_http: bool = False, suppress_exceptions: bool = False, ): """ @@ -48,6 +49,7 @@ 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. + insecure_http: If True, use HTTP instead of HTTPS and disable certificate verification. suppress_exceptions: If True, suppress exceptions and return structured error responses. Default is False, which means exceptions will be raised. """ @@ -57,6 +59,7 @@ def __init__( api_key=api_key, server_ca_path=server_ca_path, client_cert_paths=client_cert_paths, + insecure_http=insecure_http, ) # Sub-API modules diff --git a/tests/protocols/http/openvasd/test_client.py b/tests/protocols/http/openvasd/test_client.py index 2224a5f82..2ee3dab5b 100644 --- a/tests/protocols/http/openvasd/test_client.py +++ b/tests/protocols/http/openvasd/test_client.py @@ -11,21 +11,29 @@ class TestOpenvasdClient(unittest.TestCase): @patch("gvm.protocols.http.openvasd._client.Client") - def test_init_without_tls_or_api_key(self, mock_httpx_client): - create_openvasd_http_client("localhost") + @patch("gvm.protocols.http.openvasd._client.ssl.create_default_context") + def test_init_without_tls_or_api_key( + self, mock_ssl_ctx_factory, mock_httpx_client + ): + create_openvasd_http_client("localhost", insecure_http=True) mock_httpx_client.assert_called_once() _, kwargs = mock_httpx_client.call_args self.assertEqual(kwargs["base_url"], "http://localhost:3000") - self.assertFalse(kwargs["verify"]) self.assertNotIn("X-API-KEY", kwargs["headers"]) + mock_ssl_ctx_factory.assert_not_called() @patch("gvm.protocols.http.openvasd._client.Client") - def test_init_with_api_key_only(self, mock_httpx_client): - create_openvasd_http_client("localhost", api_key="secret") + @patch("gvm.protocols.http.openvasd._client.ssl.create_default_context") + def test_init_with_api_key_and_insecure_http( + self, mock_ssl_ctx_factory, mock_httpx_client + ): + create_openvasd_http_client( + "localhost", api_key="secret", insecure_http=True + ) _, kwargs = mock_httpx_client.call_args self.assertEqual(kwargs["headers"]["X-API-KEY"], "secret") self.assertEqual(kwargs["base_url"], "http://localhost:3000") - self.assertFalse(kwargs["verify"]) + mock_ssl_ctx_factory.assert_not_called() @patch("gvm.protocols.http.openvasd._client.ssl.create_default_context") @patch("gvm.protocols.http.openvasd._client.Client") @@ -72,3 +80,31 @@ def test_init_with_mtls_single_cert( _, kwargs = mock_httpx_client.call_args self.assertEqual(kwargs["base_url"], "https://localhost:3000") self.assertEqual(kwargs["verify"], mock_context) + + @patch("gvm.protocols.http.openvasd._client.ssl.create_default_context") + @patch("gvm.protocols.http.openvasd._client.Client") + def test_init_with_mtls_missing_cert( + self, mock_httpx_client, mock_ssl_ctx_factory + ): + with self.assertRaises(ValueError): + create_openvasd_http_client( + "localhost", + server_ca_path="/path/ca.pem", + client_cert_paths=None, + ) + mock_httpx_client.assert_not_called() + mock_ssl_ctx_factory.assert_not_called() + + @patch("gvm.protocols.http.openvasd._client.ssl.create_default_context") + @patch("gvm.protocols.http.openvasd._client.Client") + def test_init_with_mtls_missing_ca( + self, mock_httpx_client, mock_ssl_ctx_factory + ): + with self.assertRaises(ValueError): + create_openvasd_http_client( + "localhost", + server_ca_path=None, + client_cert_paths="/path/client.pem", + ) + mock_httpx_client.assert_not_called() + mock_ssl_ctx_factory.assert_not_called() diff --git a/tests/protocols/http/openvasd/test_openvasd1.py b/tests/protocols/http/openvasd/test_openvasd1.py index 5f46de0b5..722ef27fd 100644 --- a/tests/protocols/http/openvasd/test_openvasd1.py +++ b/tests/protocols/http/openvasd/test_openvasd1.py @@ -28,6 +28,39 @@ def test_initializes_all_sub_apis(self, mock_crate_openvasd_client): api_key="test-key", server_ca_path="/path/to/ca.pem", client_cert_paths=("/path/to/client.pem", "/path/to/key.pem"), + insecure_http=False, + ) + + self.assertEqual(api._client, mock_httpx_client) + self.assertEqual(api.health._client, mock_httpx_client) + self.assertEqual(api.metadata._client, mock_httpx_client) + self.assertEqual(api.notus._client, mock_httpx_client) + self.assertEqual(api.scans._client, mock_httpx_client) + self.assertEqual(api.vts._client, mock_httpx_client) + + @patch("gvm.protocols.http.openvasd._openvasd1.create_openvasd_http_client") + def test_initializes_all_sub_apis_insecure_http( + self, mock_crate_openvasd_client + ): + mock_httpx_client = MagicMock() + mock_crate_openvasd_client.return_value = mock_httpx_client + + api = OpenvasdHttpAPIv1( + host_name="localhost", + port=3000, + api_key="test-key", + server_ca_path="/path/to/ca.pem", + client_cert_paths=("/path/to/client.pem", "/path/to/key.pem"), + insecure_http=True, + ) + + mock_crate_openvasd_client.assert_called_once_with( + host_name="localhost", + port=3000, + api_key="test-key", + server_ca_path="/path/to/ca.pem", + client_cert_paths=("/path/to/client.pem", "/path/to/key.pem"), + insecure_http=True, ) self.assertEqual(api._client, mock_httpx_client)