feat: add completed-result JSONL reports

CHANGELOG.md

@@ -8,6 +8,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ## [Unreleased]
 
 ### Added
+- Added deterministic JSONL report companions finalized after selected one-shot actions complete.
 - Added DNSDB passive DNS discovery with API key configuration, shared transport handling, result parsing, and offline tests ([9b41b78e](https://github.com/laramies/theHarvester/commit/9b41b78e), [aba9fec6](https://github.com/laramies/theHarvester/commit/aba9fec6)).
 - Added `--verbose` diagnostic logging while keeping normal operator output available at the default log level ([8a7b8b71](https://github.com/laramies/theHarvester/commit/8a7b8b71)).
 - Added an opt-in passive-provider smoke workflow and a network guard that keeps routine tests offline by default ([72e5820f](https://github.com/laramies/theHarvester/commit/72e5820f)).

README.md

@@ -15,7 +15,7 @@ It is built for the early reconnaissance stage of authorized security assessment
 - **Useful result types:** collect hostnames, email addresses, IP addresses, URLs, ASNs, and people.
 - **Enrichment after discovery:** optionally resolve DNS, query Shodan, check for subdomain takeovers, brute-force DNS names, scan common API paths, and capture screenshots.
 - **CLI and browser-accessible API:** use the command line interactively or run the FastAPI service for automation and interactive Swagger/ReDoc documentation.
-- **Repeatable output:** print results, write JSON and XML reports, and retain host, email, and IP findings in a local SQLite database.
+- **Repeatable output:** print results, write JSON, XML, and JSONL reports, and retain host, email, and IP findings in a local SQLite database.
 - **Operational controls:** select individual sources, set result limits, use HTTP or SOCKS proxies, choose DNS resolvers, and suppress missing-key noise.
 
 Source availability, quotas, and response formats are controlled by third parties and can change independently of theHarvester.
@@ -56,7 +56,7 @@ uv run theHarvester -d example.com -b emails,urls,certspotter
 
 Capability selectors form a union and choose which sources run. They do not discard other result types returned by those sources. Available selectors are `subdomains`, `emails`, `ips`, `asns`, `urls`, and `people`. `-b all` continues to run every registered source.
 
-Save both JSON and XML reports:
+Save JSON, XML, and JSONL reports:
 
 ```bash
 uv run theHarvester -d example.com -b crtsh,certspotter -f report
@@ -206,7 +206,7 @@ Never commit populated configuration files, API keys, account details, or provid
 ## Results and local data
 
 - Terminal output shows consolidated findings. Separately selected actions, such as `-s` / `--shodan`, may print their own enrichment.
-- `-f NAME` writes `NAME.json` and `NAME.xml`.
+- `-f NAME` writes `NAME.json`, `NAME.xml`, and `NAME.jsonl`.
 - Screenshots are written to the directory passed to `--screenshot`.
 - Host, email, IP, and related scan records are stored in `~/.local/share/theHarvester/stash.sqlite`.
 - REST queries return JSON.
@@ -229,6 +229,14 @@ The JSON report is a single object and is the more complete format for automatio
 
 The XML report contains the command, emails, hosts, and virtual hosts. Use JSON when you need the additional result types above.
 
+The JSONL report is finalized after the selected one-shot actions finish. Its first line is a summary with an independent run UUID, the target, UTC timestamps, counts, and schema version. Each remaining line is one deterministic, deduplicated string finding. The format does not claim provider success or record source attribution.
+
+List every JSONL finding as tab-separated type and value columns:
+
+```bash
+jq -r 'select(.type != "summary") | [.type, .value] | @tsv' report.jsonl
+```
+
 List discovered hosts with [`jq`](https://jqlang.org/):
 
 ```bash

tests/discovery/test_rapiddns.py

@@ -5,6 +5,7 @@ import xml.etree.ElementTree as ElementTree
 from argparse import Namespace
 from pathlib import Path
 from typing import Any
+from uuid import UUID
 
 import pytest
 
@@ -76,15 +77,84 @@ async def test_rapiddns_evidence_reaches_existing_outputs(
         async def get_ips(self) -> set[str]:
             return {'198.51.100.2'}
 
+    class FakeApiEndpoints:
+        def __init__(self, *, word: str, wordlist: str) -> None:
+            assert word == 'example.com'
+            assert wordlist.endswith('api_endpoints.txt')
+
+        async def do_search(self) -> None:
+            return None
+
+        def get_found_endpoints(self) -> dict[str, object]:
+            return {'/health': object()}
+
+        def get_interesting_endpoints(self) -> dict[str, object]:
+            return {'/health': object()}
+
+        def get_auth_required(self) -> dict[str, object]:
+            return {}
+
+        def get_api_versions(self) -> set[str]:
+            return set()
+
+        def get_rate_limits(self) -> dict[str, object]:
+            return {}
+
+        def get_methods(self) -> set[str]:
+            return {'GET'}
+
+        def get_status_codes(self) -> set[int]:
+            return {200}
+
+    class FakeSecurityScorecard:
+        created = 0
+
+        def __init__(self, _domain: str) -> None:
+            self.is_late_action = self.created > 0
+            type(self).created += 1
+
+        async def process(self, _proxy: bool) -> None:
+            return None
+
+        async def get_hostnames(self) -> set[str]:
+            return set()
+
+        async def get_ips(self) -> set[str]:
+            if self.is_late_action:
+                return {'2001:0DB8::1', '198.51.100.9', 'not-an-ip'}
+            return set()
+
+    async def fake_reverse_all_ips_in_range(
+        iprange: str,
+        callback: Any,
+        nameservers: list[str] | None = None,
+    ) -> None:
+        assert iprange in {'192.0.2.0/24', '198.51.100.0/24'}
+        assert nameservers is None
+        callback('reverse.example.com')
+
     report = tmp_path / 'rapiddns-report'
     monkeypatch.setattr(rapiddns.AsyncFetcher, 'fetch_all', fake_fetch_all)
     monkeypatch.setattr(theharvester_main.stash, 'StashManager', FakeStash)
     monkeypatch.setattr(theharvester_main.hostchecker, 'Checker', UnexpectedChecker)
     monkeypatch.setattr(theharvester_main.search_dehashed, 'SearchDehashed', FakeDehashed)
+    monkeypatch.setattr(theharvester_main.api_endpoints, 'SearchApiEndpoints', FakeApiEndpoints)
+    monkeypatch.setattr(theharvester_main.securityscorecard, 'SearchSecurityScorecard', FakeSecurityScorecard)
+    monkeypatch.setattr(theharvester_main.dnssearch, 'reverse_all_ips_in_range', fake_reverse_all_ips_in_range)
     monkeypatch.setattr(
         sys,
         'argv',
-        ['theHarvester', '-d', 'example.com', '-b', 'rapiddns', '-f', str(report)],
+        [
+            'theHarvester',
+            '-d',
+            'example.com',
+            '-b',
+            'rapiddns,securityscorecard',
+            '-a',
+            '-n',
+            '-f',
+            str(report),
+        ],
     )
     configure_logging(verbose=False)
 
@@ -92,14 +162,25 @@ async def test_rapiddns_evidence_reaches_existing_outputs(
         await theharvester_main.start()
 
     assert exit_info.value.code == 0
-    assert stored == [
-        ('host', ('alias.example.com', 'api.example.com', 'broken.example.com'), 'rapiddns'),
-        ('ip', ('192.0.2.1',), 'rapiddns'),
-    ]
+    assert ('host', ('alias.example.com', 'api.example.com', 'broken.example.com'), 'rapiddns') in stored
+    assert ('ip', ('192.0.2.1',), 'rapiddns') in stored
+    assert stored.count(('api_endpoint', ('/health',), 'api_scan')) == 2
 
     report_json = json.loads(report.with_suffix('.json').read_text())
     assert report_json['hosts'] == ['alias.example.com', 'api.example.com', 'broken.example.com']
     assert report_json['ips'] == ['192.0.2.1']
+    assert 'interesting_urls' not in report_json
+
+    jsonl_records = [json.loads(line) for line in report.with_suffix('.jsonl').read_text().splitlines()]
+    assert jsonl_records[0]['type'] == 'summary'
+    assert jsonl_records[0]['target'] == 'example.com'
+    UUID(jsonl_records[0]['run_id'])
+    assert {'type': 'interesting-url', 'value': 'https://example.com/health'} in jsonl_records
+    assert {'type': 'url', 'value': 'https://example.com/health'} in jsonl_records
+    assert {'type': 'hostname', 'value': 'reverse.example.com'} in jsonl_records
+    assert {'type': 'ip-address', 'value': '198.51.100.9'} in jsonl_records
+    assert {'type': 'ip-address', 'value': '2001:db8::1'} in jsonl_records
+    assert not any(record.get('value') == 'not-an-ip' for record in jsonl_records)
 
     xml_hosts = {
         (element.findtext('hostname') or (element.text or '').strip(), element.findtext('ip'))
@@ -109,11 +190,13 @@ async def test_rapiddns_evidence_reaches_existing_outputs(
         ('alias.example.com', None),
         ('api.example.com', '192.0.2.1'),
         ('broken.example.com', None),
+        ('reverse.example.com', None),
     }
 
     console = capsys.readouterr().out
     assert {'alias.example.com', 'api.example.com', 'broken.example.com', '192.0.2.1'} <= set(console.splitlines())
 
+    stored_before_rest = len(stored)
     rest_results = await theharvester_main.start(
         Namespace(
             source='dehashed,rapiddns',
@@ -133,7 +216,7 @@ async def test_rapiddns_evidence_reaches_existing_outputs(
     )
     assert set(rest_results[6]) == {'192.0.2.1', '198.51.100.2'}
     assert rest_results[8] == ['alias.example.com', 'api.example.com', 'broken.example.com']
-    assert stored[2:] == [
+    assert stored[stored_before_rest:] == [
         ('ip', ('198.51.100.2',), 'dehashed'),
         ('host', ('alias.example.com', 'api.example.com', 'broken.example.com'), 'rapiddns'),
         ('ip', ('192.0.2.1',), 'rapiddns'),

tests/lib/test_completed_result.py

@@ -0,0 +1,109 @@
+import json
+from datetime import UTC, datetime
+from typing import Any
+from uuid import UUID
+
+import pytest
+
+from theHarvester.lib.completed_result import CompletedResult
+
+
+def test_completed_result_is_deterministic_and_deduplicated() -> None:
+    result = CompletedResult.finish(
+        run_id=UUID('f047261c-0afb-4e18-89d5-28a7d977f51f'),
+        target='example.com',
+        started_at=datetime(2026, 8, 5, 12, 0, tzinfo=UTC),
+        completed_at=datetime(2026, 8, 5, 12, 1, tzinfo=UTC),
+        groups={
+            'hostname': ['www.example.com', 'api.example.com', 'api.example.com'],
+            'email': ['admin@example.com'],
+        },
+    )
+
+    records = [json.loads(line) for line in result.jsonl().splitlines()]
+
+    assert records == [
+        {
+            'completed_at': '2026-08-05T12:01:00Z',
+            'counts': {'email': 1, 'hostname': 2},
+            'result_count': 3,
+            'run_id': 'f047261c-0afb-4e18-89d5-28a7d977f51f',
+            'schema_version': 'theharvester-results-v1',
+            'started_at': '2026-08-05T12:00:00Z',
+            'target': 'example.com',
+            'type': 'summary',
+        },
+        {
+            'type': 'email',
+            'value': 'admin@example.com',
+        },
+        {
+            'type': 'hostname',
+            'value': 'api.example.com',
+        },
+        {
+            'type': 'hostname',
+            'value': 'www.example.com',
+        },
+    ]
+    assert result.jsonl().endswith('\n')
+
+
+@pytest.mark.parametrize('value', ['', '   ', 7])
+def test_completed_result_rejects_invalid_findings(value: Any) -> None:
+    with pytest.raises(ValueError, match='non-empty string'):
+        CompletedResult.finish(
+            target='example.com',
+            started_at=datetime(2026, 8, 5, 12, 0, tzinfo=UTC),
+            completed_at=datetime(2026, 8, 5, 12, 1, tzinfo=UTC),
+            groups={'hostname': [value]},
+        )
+
+
+def test_completed_result_rejects_invalid_completion() -> None:
+    started_at = datetime(2026, 8, 5, 12, 0, tzinfo=UTC)
+
+    with pytest.raises(ValueError, match='target'):
+        CompletedResult.finish(
+            target=' ',
+            started_at=started_at,
+            completed_at=started_at,
+            groups={},
+        )
+    with pytest.raises(ValueError, match='timezone-aware'):
+        CompletedResult.finish(
+            target='example.com',
+            started_at=datetime(2026, 8, 5, 12, 0),
+            completed_at=started_at,
+            groups={},
+        )
+    with pytest.raises(ValueError, match='earlier'):
+        CompletedResult.finish(
+            target='example.com',
+            started_at=started_at,
+            completed_at=datetime(2026, 8, 5, 11, 59, tzinfo=UTC),
+            groups={},
+        )
+
+
+def test_completed_result_rejects_unknown_kind() -> None:
+    groups: Any = {'source-status': ['complete']}
+
+    with pytest.raises(ValueError, match='unknown result kind'):
+        CompletedResult.finish(
+            target='example.com',
+            started_at=datetime(2026, 8, 5, 12, 0, tzinfo=UTC),
+            completed_at=datetime(2026, 8, 5, 12, 1, tzinfo=UTC),
+            groups=groups,
+        )
+
+
+def test_completed_result_rejects_direct_whitespace_finding() -> None:
+    with pytest.raises(ValueError, match='non-empty string'):
+        CompletedResult(
+            run_id=UUID('f047261c-0afb-4e18-89d5-28a7d977f51f'),
+            target='example.com',
+            started_at=datetime(2026, 8, 5, 12, 0, tzinfo=UTC),
+            completed_at=datetime(2026, 8, 5, 12, 1, tzinfo=UTC),
+            results=(('hostname', ' '),),
+        )

theHarvester/__main__.py

@@ -9,6 +9,8 @@ import sys
 import time
 import traceback
 from collections.abc import Iterable
+from datetime import UTC, datetime
+from ipaddress import ip_address
 from typing import TYPE_CHECKING, Any
 
 import anyio
@@ -79,6 +81,7 @@ from theHarvester.discovery import (
 )
 from theHarvester.discovery.constants import MissingKey
 from theHarvester.lib import hostchecker, stash
+from theHarvester.lib.completed_result import CompletedResult, ResultKind
 from theHarvester.lib.core import DATA_DIR, Core, show_default_error_message
 from theHarvester.lib.hostnames import normalize_scoped_hostname
 from theHarvester.lib.output import configure_logging, output_logger, print_linkedin_sections, print_section, sorted_unique
@@ -100,6 +103,18 @@ def _normalize_hosts_for_storage(discovered_hosts: Iterable[object], target: str
     }
 
 
+def _normalize_ip_addresses(values: Iterable[object]) -> set[str]:
+    addresses: set[str] = set()
+    for value in values:
+        if not isinstance(value, str):
+            continue
+        try:
+            addresses.add(str(ip_address(value.strip())))
+        except ValueError:
+            continue
+    return addresses
+
+
 def sanitize_for_xml(text: str) -> str:
     """Sanitize text for safe inclusion in XML documents."""
     text = text.replace('&', '&')
@@ -198,7 +213,7 @@ async def start(rest_args: argparse.Namespace | None = None):
     parser.add_argument(
         '-f',
         '--filename',
-        help='Save the results to an XML and JSON file.',
+        help='Save the results to XML, JSON, and JSONL files.',
         default='',
         type=str,
     )
@@ -269,6 +284,7 @@ async def start(rest_args: argparse.Namespace | None = None):
         else:
             # For relative paths, sanitize the entire filename
             filename = sanitize_filename(filename)
+    report_started_at = datetime.now(UTC) if filename else None
 
     all_emails: list = []
     all_hosts: list = []
@@ -1861,6 +1877,33 @@ async def start(rest_args: argparse.Namespace | None = None):
                 else:
                     output_logger.info(f'An exception has occurred in BuiltWith scanning: {e}')
 
+    if filename and report_started_at is not None:
+        try:
+            groups: dict[ResultKind, Iterable[str]] = {
+                'asn': map(str, total_asns),
+                'email': map(str, all_emails),
+                'hostname': _normalize_hosts_for_storage((*all_hosts, *dnsrev), word),
+                'interesting-url': map(str, interesting_urls),
+                'ip-address': _normalize_ip_addresses(all_ip),
+                'linkedin-link': map(str, linkedin_links_tracker),
+                'linkedin-person': map(str, linkedin_people_list_tracker),
+                'twitter-person': map(str, twitter_people_list_tracker),
+                'url': map(str, all_urls),
+                'vhost': map(str, vhost),
+            }
+            completed_result = CompletedResult.finish(
+                target=word,
+                started_at=report_started_at,
+                completed_at=datetime.now(UTC),
+                groups=groups,
+            )
+            jsonl_filename = filename.rsplit('.', 1)[0] + '.jsonl'
+            async with await anyio.open_file(jsonl_filename, 'w+', encoding='UTF-8') as fp:
+                await fp.write(completed_result.jsonl())
+            output_logger.info('[*] JSONL File saved.')
+        except (OSError, ValueError, TypeError, UnicodeEncodeError) as error:
+            output_logger.info(f'[!] An error occurred while saving the JSONL file: {error}')
+
     if rest_args is not None:
         all_hosts = sorted({host.replace('www.', '') for host in all_hosts})
         return (

theHarvester/lib/completed_result.py

@@ -0,0 +1,95 @@
+import json
+from collections import Counter
+from collections.abc import Iterable, Mapping
+from dataclasses import dataclass
+from datetime import UTC, datetime
+from typing import Literal, Self, get_args
+from uuid import UUID, uuid4
+
+ResultKind = Literal[
+    'asn',
+    'email',
+    'hostname',
+    'interesting-url',
+    'ip-address',
+    'linkedin-link',
+    'linkedin-person',
+    'twitter-person',
+    'url',
+    'vhost',
+]
+
+SCHEMA_VERSION = 'theharvester-results-v1'
+RESULT_KINDS: frozenset[str] = frozenset(get_args(ResultKind))
+
+
+def _isoformat_utc(value: datetime) -> str:
+    return value.astimezone(UTC).isoformat().replace('+00:00', 'Z')
+
+
+@dataclass(frozen=True, slots=True)
+class CompletedResult:
+    run_id: UUID
+    target: str
+    started_at: datetime
+    completed_at: datetime
+    results: tuple[tuple[ResultKind, str], ...]
+
+    def __post_init__(self) -> None:
+        if not self.target.strip():
+            raise ValueError('target must not be empty')
+        if self.started_at.tzinfo is None or self.started_at.utcoffset() is None:
+            raise ValueError('started_at must be timezone-aware')
+        if self.completed_at.tzinfo is None or self.completed_at.utcoffset() is None:
+            raise ValueError('completed_at must be timezone-aware')
+        if self.completed_at < self.started_at:
+            raise ValueError('completed_at must not be earlier than started_at')
+        if not isinstance(self.run_id, UUID):
+            raise ValueError('run_id must be a UUID')
+        if any(kind not in RESULT_KINDS or not isinstance(value, str) or not value.strip() for kind, value in self.results):
+            raise ValueError('results must contain known kinds and non-empty string values')
+        if self.results != tuple(sorted(set(self.results))):
+            raise ValueError('results must be deduplicated and sorted')
+
+    @classmethod
+    def finish(
+        cls,
+        *,
+        run_id: UUID | None = None,
+        target: str,
+        started_at: datetime,
+        completed_at: datetime,
+        groups: Mapping[ResultKind, Iterable[str]],
+    ) -> Self:
+        results: set[tuple[ResultKind, str]] = set()
+        for kind, values in groups.items():
+            if kind not in RESULT_KINDS:
+                raise ValueError(f'unknown result kind: {kind}')
+            for value in values:
+                if not isinstance(value, str) or not value.strip():
+                    raise ValueError('results must contain non-empty string values')
+                results.add((kind, value.strip()))
+        return cls(
+            run_id=run_id or uuid4(),
+            target=target.strip(),
+            started_at=started_at,
+            completed_at=completed_at,
+            results=tuple(sorted(results)),
+        )
+
+    def jsonl(self) -> str:
+        counts = Counter(kind for kind, _value in self.results)
+        records = [
+            {
+                'completed_at': _isoformat_utc(self.completed_at),
+                'counts': dict(sorted(counts.items())),
+                'result_count': len(self.results),
+                'run_id': str(self.run_id),
+                'schema_version': SCHEMA_VERSION,
+                'started_at': _isoformat_utc(self.started_at),
+                'target': self.target,
+                'type': 'summary',
+            },
+            *({'type': kind, 'value': value} for kind, value in self.results),
+        ]
+        return ''.join(json.dumps(record, ensure_ascii=False, separators=(',', ':'), sort_keys=True) + '\n' for record in records)