fix(core): correct shared transport ssl, header, and body handling

- _create_connector: an explicit ssl=False (verification disabled) was
  silently replaced with a verifying context by 'ssl_context or default';
  only an unset context now falls back to the certifi default
- fetch/post_fetch/fetch_all: caller-supplied headers were silently
  dropped when a caller-owned session was passed; they are now forwarded
  per request, matching _open_get_response and aiohttp merge semantics
  (sessions that bake the same headers are unaffected)
- fetch: a borrowed session with default verification no longer rebuilds
  a certifi SSL context on every request; it defers to the connector
  context the session already owns
- _normalize_data: the post_fetch data='' default raised JSONDecodeError
  internally and silently returned without sending any request; empty
  bodies now send, and non-JSON strings pass through as raw bodies

Traced every owned-versus-borrowed branch; all post_fetch callers pass
json_body or valid JSON strings, and the only verify=False caller
(api_endpoints) supplies its own connector.

tests/lib/test_core.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 import asyncio
 import logging
+import ssl
 import stat
 from concurrent.futures import ThreadPoolExecutor
 from pathlib import Path
@@ -452,6 +453,103 @@ async def test_fetch_reused_session_uses_a_stable_explicit_ssl_policy(monkeypatc
     assert ssl_policies[0] is ssl_policies[1]
 
 
+@pytest.mark.asyncio
+async def test_fetch_reused_session_defers_default_ssl_policy_to_the_connector(monkeypatch) -> None:
+    session = DummySession()
+
+    def fail_if_context_built(_verify=True):
+        raise AssertionError('a borrowed session must not rebuild an SSL context per request')
+
+    monkeypatch.setattr(AsyncFetcher, '_ssl_context', staticmethod(fail_if_context_built))
+
+    await AsyncFetcher.fetch(session=session, url='https://example.com/one', verify=None)
+
+    assert session.requests == [('GET', 'https://example.com/one', {})]
+
+
+@pytest.mark.asyncio
+async def test_fetch_reused_session_forwards_caller_headers(monkeypatch) -> None:
+    session = DummySession()
+
+    await AsyncFetcher.fetch(session=session, url='https://example.com/api', headers={'X-Api-Key': 'test-key'})
+
+    _method, _url, request_options = session.requests[0]
+    assert request_options['headers']['X-Api-Key'] == 'test-key'
+
+
+@pytest.mark.asyncio
+async def test_fetch_reused_session_without_headers_keeps_session_headers(monkeypatch) -> None:
+    session = DummySession()
+
+    await AsyncFetcher.fetch(session=session, url='https://example.com/api')
+
+    assert 'headers' not in session.requests[0][2]
+
+
+@pytest.mark.asyncio
+async def test_post_fetch_reused_session_forwards_caller_headers(monkeypatch) -> None:
+    session = DummySession()
+
+    await AsyncFetcher.post_fetch(
+        'https://example.com/api',
+        headers={'X-Api-Key': 'test-key'},
+        session=session,
+        json=True,
+        json_body={'query': 'example.com'},
+    )
+
+    _method, _url, request_options = session.requests[0]
+    assert request_options['headers']['X-Api-Key'] == 'test-key'
+    assert request_options['json'] == {'query': 'example.com'}
+
+
+@pytest.mark.asyncio
+async def test_post_fetch_sends_the_default_empty_body(monkeypatch) -> None:
+    session = DummySession()
+
+    await AsyncFetcher.post_fetch('https://example.com/api', session=session, json=True)
+
+    _method, _url, request_options = session.requests[0]
+    assert request_options['data'] == ''
+
+
+@pytest.mark.asyncio
+async def test_post_fetch_passes_non_json_string_data_through(monkeypatch) -> None:
+    session = DummySession()
+
+    await AsyncFetcher.post_fetch('https://example.com/api', data='field=value', session=session)
+
+    _method, _url, request_options = session.requests[0]
+    assert request_options['data'] == 'field=value'
+
+
+@pytest.mark.asyncio
+async def test_tcp_connector_honors_disabled_verification() -> None:
+    connector = await AsyncFetcher._create_connector(None, None, False)
+    try:
+        assert connector._ssl is False
+    finally:
+        await connector.close()
+
+
+@pytest.mark.asyncio
+async def test_tcp_connector_defaults_to_a_verifying_context() -> None:
+    connector = await AsyncFetcher._create_connector(None, None, None)
+    try:
+        assert isinstance(connector._ssl, ssl.SSLContext)
+    finally:
+        await connector.close()
+
+
+@pytest.mark.asyncio
+async def test_create_session_with_verification_disabled_builds_an_unverified_connector() -> None:
+    session = await AsyncFetcher.create_session(verify=False)
+    try:
+        assert session.connector._ssl is False
+    finally:
+        await session.close()
+
+
 @pytest.mark.asyncio
 async def test_open_session_owns_one_proxy_and_cookie_policy(monkeypatch: pytest.MonkeyPatch) -> None:
     reset_dummy_sessions()

theHarvester/lib/core.py

@@ -526,7 +526,12 @@ class AsyncFetcher:
 
     @staticmethod
     def _normalize_data(data: str | dict[str, Any]) -> str | dict[str, Any]:
-        return json_loader.loads(data) if isinstance(data, str) else data
+        if isinstance(data, str) and data:
+            try:
+                return json_loader.loads(data)
+            except ValueError:
+                return data
+        return data
 
     @classmethod
     def _resolve_proxy(cls, proxy: str | bool | None) -> tuple[str | None, str | None]:
@@ -779,7 +784,9 @@ class AsyncFetcher:
             return connector
         else:
             # Use default TCP connector for HTTP proxies or no proxy
-            return aiohttp.TCPConnector(ssl=ssl_context or ssl.create_default_context(cafile=certifi.where()))
+            return aiohttp.TCPConnector(
+                ssl=ssl_context if ssl_context is not None else ssl.create_default_context(cafile=certifi.where())
+            )
 
     @classmethod
     async def post_fetch(
@@ -796,6 +803,7 @@ class AsyncFetcher:
         session: aiohttp.ClientSession | None = None,
         response_byte_limit: int | None = None,
     ) -> Any:
+        caller_headers = headers
         headers = cls._default_headers(headers)
         # By default, timeout is 5 minutes, changed to 12-minutes
         # results are well worth the wait
@@ -817,6 +825,8 @@ class AsyncFetcher:
             }
             if params != '':
                 request_kwargs['params'] = params
+            if caller_headers is not None:
+                request_kwargs['headers'] = headers
             return await cls._request(
                 session,
                 'POST',
@@ -855,7 +865,14 @@ class AsyncFetcher:
         """
         try:
             owns_session = session is None
-            ssl_arg = cls._ssl_context(verify) if owns_session or not isinstance(verify, bool) else verify
+            if owns_session:
+                ssl_arg = cls._ssl_context(verify)
+            elif isinstance(verify, bool):
+                ssl_arg = verify
+            else:
+                # A borrowed session already owns its TLS policy; defer to the
+                # session connector instead of rebuilding a context per request.
+                ssl_arg = None
             proxy_url, proxy_type = cls._resolve_proxy(proxy)
             client_timeout = cls._request_timeout(request_timeout)
             req_headers = cls._default_headers(headers)
@@ -871,9 +888,9 @@ class AsyncFetcher:
             assert session is not None
 
             try:
-                request_kwargs: dict[str, Any] = {
-                    'ssl': ssl_arg,
-                }
+                request_kwargs: dict[str, Any] = {}
+                if ssl_arg is not None:
+                    request_kwargs['ssl'] = ssl_arg
                 # For HTTP proxies, pass the proxy parameter; for SOCKS5, the connector handles it
                 if proxy_url and proxy_type == 'http':
                     request_kwargs['proxy'] = proxy_url
@@ -881,6 +898,8 @@ class AsyncFetcher:
                     request_kwargs['allow_redirects'] = follow_redirects
                 if params != '':
                     request_kwargs['params'] = params
+                if not owns_session and headers is not None:
+                    request_kwargs['headers'] = req_headers
                 return await cls._request(
                     session,
                     method,
@@ -1102,6 +1121,7 @@ class AsyncFetcher:
                             url=url,
                             params=params,
                             json=json,
+                            headers=headers,
                             include_metadata=include_metadata,
                         )
                         for url in urls