Migrate Pentest-Tools discovery to API v2 (#2497)

* Migrate Pentest-Tools discovery to API v2

* Send Pentest scan payload as JSON

CHANGELOG.md

@@ -21,6 +21,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Added root contributor and security policies, structured issue forms, repository agent guidance, discovery terminology, and an operator-focused documentation wiki ([d090a29a](https://github.com/laramies/theHarvester/commit/d090a29a), [7c491ef5](https://github.com/laramies/theHarvester/commit/7c491ef5), [8b9d420b](https://github.com/laramies/theHarvester/commit/8b9d420b)).
 
 ### Changed
+- Migrated Pentest-Tools discovery to its API v2 Bearer-authenticated scan, status, and output endpoints.
 - Included HIBP verified-domain in `all` and matching capability selectors like every other P0 source, with REST operator authentication applied after source expansion when its provider key is configured.
 - Allowed REST `/query` requests to select discovery sources by result capability, matching the CLI's union semantics while preserving explicit source selection.
 - Changed `-b all` to select every cataloged P0 passive source once while leaving P1 DNS and P2 direct sources available through explicit selection.

tests/discovery/test_pentesttools.py

@@ -8,26 +8,41 @@ from theHarvester.discovery import pentesttools
 @pytest.mark.asyncio
 async def test_successful_scan_returns_normalized_hostnames_and_ips(monkeypatch) -> None:
     monkeypatch.setattr(pentesttools.Core, 'pentest_tools_key', lambda: 'test-key')
-
+    start_requests = []
+    get_requests = []
     responses = iter(
         [
-            '{"op_status":"success","scan_id":"scan-1"}',
-            '{"op_status":"success","scan_status":"finished"}',
-            (
-                '{"op_status":"success","scan_output":{"output_json":[{"output_data":'
-                '[["Api.Example.TEST.","203.0.113.10"],["Example.TEST.","203.0.113.11"],'
-                '["NoIp.Example.TEST.","not-an-ip"],["www.notexample.test","198.51.100.1"],["broken"]]}]}}'
-            ),
+            {'data': {'status_name': 'finished'}},
+            {
+                'data': {
+                    'output_type': 'subdomain_list',
+                    'output_data': {
+                        'subdomains': [
+                            {'hostname': 'Api.Example.TEST.', 'ip_address': '203.0.113.10'},
+                            {'hostname': 'Example.TEST.', 'ip_address': '203.0.113.11'},
+                            {'hostname': 'NoIp.Example.TEST.', 'ip_address': ''},
+                            {'hostname': 'www.notexample.test', 'ip_address': '198.51.100.1'},
+                            {'hostname': 'broken'},
+                        ]
+                    },
+                }
+            },
         ]
     )
 
-    async def fake_post_fetch(*_args, **_kwargs):
+    async def fake_post_fetch(**kwargs):
+        start_requests.append(kwargs)
+        return {'data': {'created_id': 420323, 'target_id': 5426912}}
+
+    async def fake_fetch(**kwargs):
+        get_requests.append(kwargs)
         return next(responses)
 
     async def no_sleep(*_args, **_kwargs):
         return None
 
     monkeypatch.setattr(pentesttools.AsyncFetcher, 'post_fetch', fake_post_fetch)
+    monkeypatch.setattr(pentesttools.AsyncFetcher, 'fetch', fake_fetch)
     monkeypatch.setattr(pentesttools.asyncio, 'sleep', no_sleep)
 
     search = pentesttools.SearchPentestTools(' Example.TEST. ')
@@ -35,26 +50,60 @@ async def test_successful_scan_returns_normalized_hostnames_and_ips(monkeypatch)
 
     assert await search.get_hostnames() == {'api.example.test', 'noip.example.test'}
     assert await search.get_ips() == {'203.0.113.10', '203.0.113.11'}
+    headers = {
+        'Accept': 'application/json',
+        'Authorization': 'Bearer test-key',
+        'Content-Type': 'application/json',
+    }
+    assert start_requests == [
+        {
+            'url': 'https://app.pentest-tools.com/api/v2/scans',
+            'headers': headers,
+            'json_body': {
+                'tool_id': 20,
+                'target_name': 'example.test',
+                'tool_params': {'scan_type': 'light', 'web_details': False, 'unresolved_results': True},
+            },
+            'json': True,
+            'proxy': False,
+        }
+    ]
+    assert get_requests == [
+        {
+            'url': 'https://app.pentest-tools.com/api/v2/scans/420323',
+            'headers': headers,
+            'json': True,
+            'proxy': False,
+        },
+        {
+            'url': 'https://app.pentest-tools.com/api/v2/scans/420323/output',
+            'headers': headers,
+            'json': True,
+            'proxy': False,
+        },
+    ]
 
 
 @pytest.mark.asyncio
 async def test_status_error_payload_is_not_logged(monkeypatch, caplog) -> None:
     monkeypatch.setattr(pentesttools.Core, 'pentest_tools_key', lambda: 'test-key')
 
-    responses = iter(
-        [
-            '{"op_status":"success","scan_id":"scan-1"}',
-            '{"op_status":"error","error":"provider-secret-payload","details":"private target data"}',
-        ]
-    )
+    async def fake_post_fetch(**_kwargs):
+        return {'data': {'created_id': 420323}}
 
-    async def fake_post_fetch(*args, **kwargs):
-        return next(responses)
+    async def fake_fetch(**_kwargs):
+        return {
+            'data': {
+                'status_name': 'failed to start',
+                'status_message': 'provider-secret-payload private target data',
+            }
+        }
 
-    async def no_sleep(*args, **kwargs):
+    async def no_sleep(*_args, **_kwargs):
         return None
 
     monkeypatch.setattr(pentesttools.AsyncFetcher, 'post_fetch', fake_post_fetch)
+    monkeypatch.setattr(pentesttools.AsyncFetcher, 'fetch', fake_fetch)
     monkeypatch.setattr(pentesttools.asyncio, 'sleep', no_sleep)
     caplog.set_level(logging.INFO, logger=pentesttools.__name__)
 
@@ -62,11 +111,14 @@ async def test_status_error_payload_is_not_logged(monkeypatch, caplog) -> None:
 
     assert 'provider-secret-payload' not in caplog.text
     assert 'private target data' not in caplog.text
-    assert 'get_scan_status failed' in caplog.text
+    assert 'did not finish successfully' in caplog.text
 
 
 @pytest.mark.asyncio
-@pytest.mark.parametrize('response', ['not-json', '{}'])
+@pytest.mark.parametrize(
+    'response',
+    ['not-json', {}, {'data': {}}, {'data': {'created_id': 'scan-1'}}, {'data': {'created_id': True}}],
+)
 async def test_malformed_start_response_completes_without_evidence(monkeypatch, caplog, response) -> None:
     monkeypatch.setattr(pentesttools.Core, 'pentest_tools_key', lambda: 'test-key')
 
@@ -81,50 +133,51 @@ async def test_malformed_start_response_completes_without_evidence(monkeypatch,
 
     assert not await search.get_hostnames()
     assert not await search.get_ips()
-    assert 'malformed response' in caplog.text
+    assert 'malformed' in caplog.text
 
 
 @pytest.mark.asyncio
 async def test_waiting_scan_stops_after_ten_status_checks(monkeypatch, caplog) -> None:
     monkeypatch.setattr(pentesttools.Core, 'pentest_tools_key', lambda: 'test-key')
-    responses = iter(['{"op_status":"success","scan_id":"scan-1"}'] + ['{"op_status":"success","scan_status":"waiting"}'] * 10)
-    calls = 0
+    status_calls = 0
 
-    async def fake_post_fetch(*_args, **_kwargs):
-        nonlocal calls
-        calls += 1
-        return next(responses)
+    async def fake_post_fetch(**_kwargs):
+        return {'data': {'created_id': 420323}}
+
+    async def fake_fetch(**_kwargs):
+        nonlocal status_calls
+        status_calls += 1
+        return {'data': {'status_name': 'waiting'}}
 
     async def no_sleep(*_args, **_kwargs):
         return None
 
     monkeypatch.setattr(pentesttools.AsyncFetcher, 'post_fetch', fake_post_fetch)
+    monkeypatch.setattr(pentesttools.AsyncFetcher, 'fetch', fake_fetch)
     monkeypatch.setattr(pentesttools.asyncio, 'sleep', no_sleep)
     caplog.set_level(logging.INFO, logger=pentesttools.__name__)
 
     await pentesttools.SearchPentestTools('example.test').process()
 
-    assert calls == 11
+    assert status_calls == 10
     assert 'still waiting after 10 status checks' in caplog.text
 
 
 @pytest.mark.asyncio
 async def test_malformed_status_response_completes_without_evidence(monkeypatch, caplog) -> None:
     monkeypatch.setattr(pentesttools.Core, 'pentest_tools_key', lambda: 'test-key')
-    responses = iter(
-        [
-            '{"op_status":"success","scan_id":"scan-1"}',
-            '{"op_status":"success"}',
-        ]
-    )
 
-    async def fake_post_fetch(*_args, **_kwargs):
-        return next(responses)
+    async def fake_post_fetch(**_kwargs):
+        return {'data': {'created_id': 420323}}
+
+    async def fake_fetch(**_kwargs):
+        return {'data': {}}
 
     async def no_sleep(*_args, **_kwargs):
         return None
 
     monkeypatch.setattr(pentesttools.AsyncFetcher, 'post_fetch', fake_post_fetch)
+    monkeypatch.setattr(pentesttools.AsyncFetcher, 'fetch', fake_fetch)
     monkeypatch.setattr(pentesttools.asyncio, 'sleep', no_sleep)
     caplog.set_level(logging.INFO, logger=pentesttools.__name__)
 
@@ -140,19 +193,58 @@ async def test_null_output_data_completes_without_evidence(monkeypatch) -> None:
     monkeypatch.setattr(pentesttools.Core, 'pentest_tools_key', lambda: 'test-key')
     responses = iter(
         [
-            '{"op_status":"success","scan_id":"scan-1"}',
-            '{"op_status":"success","scan_status":"finished"}',
-            '{"op_status":"success","scan_output":{"output_json":[{"output_data":null}]}}',
+            {'data': {'status_name': 'finished'}},
+            {'data': {'output_type': 'subdomain_list', 'output_data': None}},
         ]
     )
 
-    async def fake_post_fetch(*_args, **_kwargs):
+    async def fake_post_fetch(**_kwargs):
+        return {'data': {'created_id': 420323}}
+
+    async def fake_fetch(**_kwargs):
         return next(responses)
 
     async def no_sleep(*_args, **_kwargs):
         return None
 
     monkeypatch.setattr(pentesttools.AsyncFetcher, 'post_fetch', fake_post_fetch)
+    monkeypatch.setattr(pentesttools.AsyncFetcher, 'fetch', fake_fetch)
+    monkeypatch.setattr(pentesttools.asyncio, 'sleep', no_sleep)
+
+    search = pentesttools.SearchPentestTools('example.test')
+    await search.process()
+
+    assert not await search.get_hostnames()
+    assert not await search.get_ips()
+
+
+@pytest.mark.asyncio
+async def test_non_subdomain_output_is_not_collected(monkeypatch) -> None:
+    monkeypatch.setattr(pentesttools.Core, 'pentest_tools_key', lambda: 'test-key')
+
+    async def fake_post_fetch(**_kwargs):
+        return {'data': {'created_id': 420323}}
+
+    responses = iter(
+        [
+            {'data': {'status_name': 'finished'}},
+            {
+                'data': {
+                    'output_type': 'finding_list',
+                    'output_data': {'subdomains': [{'hostname': 'injected.example.test', 'ip_address': '203.0.113.10'}]},
+                }
+            },
+        ]
+    )
+
+    async def fake_fetch(**_kwargs):
+        return next(responses)
+
+    async def no_sleep(*_args, **_kwargs):
+        return None
+
+    monkeypatch.setattr(pentesttools.AsyncFetcher, 'post_fetch', fake_post_fetch)
+    monkeypatch.setattr(pentesttools.AsyncFetcher, 'fetch', fake_fetch)
     monkeypatch.setattr(pentesttools.asyncio, 'sleep', no_sleep)
 
     search = pentesttools.SearchPentestTools('example.test')

tests/lib/test_core.py

@@ -455,6 +455,27 @@ async def test_post_fetch_decodes_string_payload_and_posts_params(monkeypatch) -
     ]
 
 
+@pytest.mark.asyncio
+async def test_post_fetch_sends_json_body(monkeypatch) -> None:
+    reset_dummy_sessions()
+    monkeypatch.setattr(core_module.aiohttp, 'ClientSession', DummySession)
+    monkeypatch.setattr(core_module.asyncio, 'sleep', fake_sleep)
+    monkeypatch.setattr(core_module.ssl, 'create_default_context', lambda cafile=None: 'ssl-context')
+    monkeypatch.setattr(core_module.certifi, 'where', lambda: '/tmp/cacert.pem')
+
+    result = await AsyncFetcher.post_fetch(
+        'https://example.com/api',
+        json_body={'scan': 'example'},
+        json=True,
+    )
+
+    assert result == {'ok': True}
+    session = DummySession.instances[0]
+    assert session.requests == [
+        ('POST', 'https://example.com/api', {'json': {'scan': 'example'}})
+    ]
+
+
 @pytest.mark.asyncio
 async def test_post_fetch_can_include_response_metadata(monkeypatch) -> None:
     reset_dummy_sessions()

theHarvester/discovery/pentesttools.py

@@ -2,8 +2,6 @@ import asyncio
 import logging
 from ipaddress import ip_address
 
-import ujson
-
 from theHarvester.discovery.constants import MissingKey
 from theHarvester.lib.core import AsyncFetcher, Core
 from theHarvester.lib.hostnames import normalize_scoped_hostname
@@ -19,73 +17,79 @@ class SearchPentestTools:
             raise MissingKey('PentestTools')
         self.totalhosts: set[str] = set()
         self.totalips: set[str] = set()
-        self.api = f'https://pentest-tools.com/api?key={self.key}'
+        self.api = 'https://app.pentest-tools.com/api/v2'
+        self.headers = {
+            'Accept': 'application/json',
+            'Authorization': f'Bearer {self.key}',
+            'Content-Type': 'application/json',
+        }
         self.proxy = False
 
     @staticmethod
-    def _decode_response(response: object) -> dict | None:
-        if not isinstance(response, str):
+    def _response_data(response: object) -> dict | None:
+        if not isinstance(response, dict) or not isinstance(data := response.get('data'), dict):
             logger.info('Pentest-Tools returned a malformed response')
             return None
-        try:
-            decoded = ujson.loads(response.strip())
-        except ValueError:
-            logger.info('Pentest-Tools returned a malformed response')
-            return None
-        if not isinstance(decoded, dict) or not isinstance(decoded.get('op_status'), str):
-            logger.info('Pentest-Tools returned a malformed response')
-            return None
-        return decoded
+        return data
 
-    async def poll(self, scan_id):
+    async def poll(self, scan_id: int) -> None:
         for _attempt in range(10):
             await asyncio.sleep(3)
-            # Get the status of our scan
-            scan_status_data = {'op': 'get_scan_status', 'scan_id': scan_id}
-            responses = await AsyncFetcher.post_fetch(url=self.api, data=ujson.dumps(scan_status_data), proxy=self.proxy)
-            res_json = self._decode_response(responses)
-            if res_json is None:
+            status = self._response_data(
+                await AsyncFetcher.fetch(
+                    url=f'{self.api}/scans/{scan_id}',
+                    headers=self.headers,
+                    json=True,
+                    proxy=self.proxy,
+                )
+            )
+            if status is None:
                 return
-            if res_json['op_status'] == 'success':
-                scan_status = res_json.get('scan_status')
-                if not isinstance(scan_status, str):
-                    logger.info('Pentest-Tools returned a malformed status response')
-                    return
-                if scan_status != 'waiting' and scan_status != 'running':
-                    getoutput_data = {
-                        'op': 'get_output',
-                        'scan_id': scan_id,
-                        'output_format': 'json',
-                    }
-                    responses = await AsyncFetcher.post_fetch(url=self.api, data=ujson.dumps(getoutput_data), proxy=self.proxy)
-
-                    output = self._decode_response(responses)
-                    if output is not None:
-                        await self.parse_json(output)
-                    return
+            status_name = status.get('status_name')
+            if not isinstance(status_name, str):
+                logger.info('Pentest-Tools returned a malformed status response')
+                return
+            if status_name in {'waiting', 'running'}:
+                continue
+            if status_name == 'finished':
+                output = self._response_data(
+                    await AsyncFetcher.fetch(
+                        url=f'{self.api}/scans/{scan_id}/output',
+                        headers=self.headers,
+                        json=True,
+                        proxy=self.proxy,
+                    )
+                )
+                if output is not None:
+                    await self.parse_json(output)
             else:
-                logger.info('Pentest-Tools operation get_scan_status failed')
-                return
+                logger.info('Pentest-Tools scan did not finish successfully')
+            return
         logger.info('Pentest-Tools scan is still waiting after 10 status checks')
 
     async def parse_json(self, json_results) -> None:
+        if json_results.get('output_type') != 'subdomain_list':
+            return
         try:
-            output_data = json_results['scan_output']['output_json'][0]['output_data']
-        except (KeyError, IndexError, TypeError):
+            output_data = json_results['output_data']['subdomains']
+        except (KeyError, TypeError):
             return
         if not isinstance(output_data, list):
             return
         for result in output_data:
-            if not isinstance(result, list) or len(result) < 2:
+            if not isinstance(result, dict):
                 continue
-            hostname = normalize_scoped_hostname(result[0], self.word)
+            hostname = normalize_scoped_hostname(result.get('hostname'), self.word)
             if not hostname:
                 continue
             if hostname != self.word:
                 self.totalhosts.add(hostname)
+            address_value = result.get('ip_address')
+            if not isinstance(address_value, str):
+                continue
             try:
-                address = str(ip_address(result[1]))
-            except (TypeError, ValueError):
+                address = str(ip_address(address_value))
+            except ValueError:
                 continue
             self.totalips.add(address)
 
@@ -99,24 +103,30 @@ class SearchPentestTools:
         # Pentest-Tools documents Subdomain Finder as tool 20:
         # https://pentest-tools.com/docs/api-reference/scans/start-a-scan
         subdomain_payload = {
-            'op': 'start_scan',
             'tool_id': 20,
+            'target_name': self.word,
             'tool_params': {
-                'target': f'{self.word}',
-                'web_details': 'off',
-                'do_smart_search': 'off',
+                'scan_type': 'light',
+                'web_details': False,
+                'unresolved_results': True,
             },
         }
-        responses = await AsyncFetcher.post_fetch(url=self.api, data=ujson.dumps(subdomain_payload), proxy=self.proxy)
-        res_json = self._decode_response(responses)
-        if res_json is None:
+        response = self._response_data(
+            await AsyncFetcher.post_fetch(
+                url=f'{self.api}/scans',
+                headers=self.headers,
+                json_body=subdomain_payload,
+                json=True,
+                proxy=self.proxy,
+            )
+        )
+        if response is None:
             return
-        if res_json['op_status'] == 'success':
-            scan_id = res_json.get('scan_id')
-            if scan_id is None:
-                logger.info('Pentest-Tools returned a malformed start response')
-                return
-            await self.poll(scan_id)
+        scan_id = response.get('created_id')
+        if not isinstance(scan_id, int) or isinstance(scan_id, bool):
+            logger.info('Pentest-Tools returned a malformed start response')
+            return
+        await self.poll(scan_id)
 
     async def process(self, proxy: bool = False) -> None:
         self.proxy = proxy

theHarvester/lib/core.py

@@ -531,11 +531,15 @@ class AsyncFetcher:
         url: str,
         *,
         json: bool = False,
+        json_body: dict[str, Any] | None = None,
         delay: int = 5,
         request_timeout: int | None = None,
         include_metadata: bool = False,
         **request_kwargs: Any,
     ) -> Any:
+        if json_body is not None:
+            request_kwargs.pop('data', None)
+            request_kwargs['json'] = json_body
         if request_timeout:
             async with asyncio.timeout(request_timeout):
                 async with session.request(method.upper(), url, **request_kwargs) as response:
@@ -596,6 +600,7 @@ class AsyncFetcher:
         json: bool = False,
         proxy: bool = False,
         include_metadata: bool = False,
+        json_body: dict[str, Any] | None = None,
     ):
         headers = cls._default_headers(headers)
         timeout = cls._request_timeout(720)
@@ -615,6 +620,7 @@ class AsyncFetcher:
                             params=params,
                             proxy=proxy_url if proxy_type == 'http' else None,
                             json=json,
+                            json_body=json_body,
                             delay=5,
                             include_metadata=include_metadata,
                         )
@@ -626,6 +632,7 @@ class AsyncFetcher:
                             url,
                             proxy=proxy_url if proxy_type == 'http' else None,
                             json=json,
+                            json_body=json_body,
                             delay=5,
                             include_metadata=include_metadata,
                         )
@@ -635,8 +642,9 @@ class AsyncFetcher:
                         session,
                         'POST',
                         url,
-                        data=cls._normalize_data(data),
+                        data=cls._normalize_data(data) if json_body is None else None,
                         json=json,
+                        json_body=json_body,
                         delay=3,
                         include_metadata=include_metadata,
                     )
@@ -646,10 +654,11 @@ class AsyncFetcher:
                         session,
                         'POST',
                         url,
-                        data=cls._normalize_data(data),
+                        data=cls._normalize_data(data) if json_body is None else None,
                         ssl=cls._ssl_context(),
                         params=params,
                         json=json,
+                        json_body=json_body,
                         delay=3,
                         include_metadata=include_metadata,
                     )