Add crt.name composite hostname discovery (#2531)

CHANGELOG.md

@@ -8,6 +8,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ## [Unreleased]
 
 ### Added
+- Added bounded, keyless `crt.name` composite-index discovery as a separate source alongside `crtsh`, retaining only descendant-hostname candidates from its streamed response.
 - Added bounded, keyless APIs.guru discovery through exact target-domain directory lookups, retaining only target-scoped hostnames, contact emails, and URLs from preferred OpenAPI specifications.
 - Added bounded virtual host discovery over harvested or operator-supplied literal-IP endpoints, with aligned HTTP `Host` and TLS SNI, synthetic unknown-host controls, hard request and runtime limits, and structured observations on canonical hostname results in JSONL, SQLite, the API, and HarvestView.
 - Added HarvestView, an authenticated local browser workspace backed by a durable single-worker `/api/v1` run lifecycle with cancellation, deadlines, JSONL-only file interchange, retained partial evidence, and real-browser regression coverage.

README.md

@@ -172,6 +172,7 @@ Read the **API key** column as follows:
 | `chaos` | ✓ | No | No | No | No | No | No | No | ✓ |
 | `commoncrawl` | ✓ | No | No | No | No | No | No | No | No |
 | `criminalip` | ✓ | No | ✓ | ✓ | No | No | No | No | ✓ |
+| `crt-name` | ✓ | No | No | No | No | No | No | No | No |
 | `crtsh` | ✓ | No | No | No | No | No | No | No | No |
 | `dehashed` | No | ✓ | ✓ | No | No | No | No | No | ✓ |
 | `dnsdb` | ✓ | No | No | No | No | No | No | No | ✓ |
@@ -223,6 +224,8 @@ Read the **API key** column as follows:
 
 `apis-guru` performs P0 provider-side collection through APIs.guru's public v2 API. It requests the exact target-domain directory entry and follows every matching preferred OpenAPI specification within hard 1,000-entry and 10-minute safety ceilings. `--limit` bounds retained results per output type without truncating catalog traversal. The source retains only target-scoped hostnames, contact emails, and HTTP(S) URLs; external OAuth, CDN, and third-party server references are excluded. API specifications, operations, security declarations, version provenance, and external relationships remain deferred until the normalized evidence model can represent them without flattening their meaning.
 
+`crt-name` requests the provider's single unpaginated composite response for the exact operator-requested scope and retains only names inside that scope. It does not broaden a descendant target to its registrable domain, use `-l` / `--limit`, contact the target, or replace `crtsh`. Its results combine certificate-transparency and other public datasets, so overlap with `crtsh` is expected and a returned hostname is not proof of ownership, scope, or current liveness. The response remains subject to the shared 64 MiB stream and 90-second runtime ceilings.
+
 `sourcegraph` makes one anonymous, provider-only search capped at 5,000 code matches. It does not use `-l` / `--limit`; returned names are candidates mentioned in indexed code, not proof of ownership or liveness.
 
 Provider pricing is intentionally omitted because plans and quotas change frequently. See [Configuration and API Keys](docs/wiki/Configuration-and-API-Keys.md) and each provider's current documentation.

tests/discovery/test_crtname.py

@@ -0,0 +1,321 @@
+import asyncio
+import contextlib
+import json
+import sys
+from collections.abc import AsyncIterator
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+from theHarvester import __main__ as theharvester_main
+from theHarvester.discovery import crtname
+from theHarvester.lib.completed_result import CompletedResult
+from theHarvester.lib.core import ResponseStreamError
+
+
+class FakeResponse:
+    def __init__(
+        self,
+        records: tuple[str, ...],
+        status: int = 200,
+        error: BaseException | None = None,
+    ) -> None:
+        self.records = records
+        self.status = status
+        self.headers: dict[str, str] = {}
+        self.error = error
+
+    def __aiter__(self) -> AsyncIterator[str]:
+        async def records() -> AsyncIterator[str]:
+            for record in self.records:
+                yield record
+            if self.error is not None:
+                raise self.error
+
+        return records()
+
+
+@pytest.mark.asyncio
+async def test_crt_name_streams_one_provider_response_and_retains_scoped_descendants(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    calls: list[dict[str, Any]] = []
+
+    @contextlib.asynccontextmanager
+    async def stream_records(url: str, **kwargs: Any) -> AsyncIterator[FakeResponse]:
+        calls.append({'url': url, **kwargs})
+        yield FakeResponse(
+            (
+                'example.com',
+                'API.Example.COM.',
+                '*.wild.example.com',
+                'outside.example.net',
+                '',
+            )
+        )
+
+    monkeypatch.setattr(crtname.AsyncFetcher, 'stream_records', stream_records)
+    search = crtname.SearchCrtName(' Example.COM. ')
+
+    await search.process(proxy=True)
+
+    assert calls == [
+        {
+            'url': 'https://crt.name/v1/search',
+            'framing': 'ndjson',
+            'params': {'apex': 'example.com'},
+            'headers': {'Accept': 'text/plain'},
+            'proxy': True,
+            'follow_redirects': False,
+            'request_timeout': 90,
+        }
+    ]
+    assert await search.get_hostnames() == {'api.example.com', 'wild.example.com'}
+    assert search.execution_status == 'partial'
+    assert search.stop_reason == 'invalid-response'
+
+
+@pytest.mark.asyncio
+async def test_crt_name_queries_and_retains_only_the_requested_scope(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    calls: list[dict[str, Any]] = []
+
+    @contextlib.asynccontextmanager
+    async def stream_records(url: str, **kwargs: Any) -> AsyncIterator[FakeResponse]:
+        calls.append({'url': url, **kwargs})
+        yield FakeResponse(('www.example.com', 'deep.www.example.com'))
+
+    monkeypatch.setattr(crtname.AsyncFetcher, 'stream_records', stream_records)
+    search = crtname.SearchCrtName('www.example.com')
+
+    await search.process()
+
+    assert calls[0]['params'] == {'apex': 'www.example.com'}
+    assert await search.get_hostnames() == {'deep.www.example.com'}
+    assert search.execution_status == 'completed'
+    assert search.stop_reason is None
+
+
+@pytest.mark.asyncio
+async def test_crt_name_queries_a_syntactically_valid_unknown_suffix_without_public_suffix_lookup(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    calls: list[dict[str, Any]] = []
+
+    @contextlib.asynccontextmanager
+    async def stream_records(url: str, **kwargs: Any) -> AsyncIterator[FakeResponse]:
+        calls.append({'url': url, **kwargs})
+        yield FakeResponse(())
+
+    monkeypatch.setattr(crtname.AsyncFetcher, 'stream_records', stream_records)
+
+    await crtname.SearchCrtName('scope.example').process()
+
+    assert calls[0]['params'] == {'apex': 'scope.example'}
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ('status', 'execution_status', 'stop_reason'),
+    [
+        (400, 'failed', 'http-400'),
+        (401, 'failed', 'access-denied'),
+        (403, 'failed', 'access-denied'),
+        (429, 'rate-limited', 'http-429'),
+        (503, 'failed', 'http-503'),
+    ],
+)
+async def test_crt_name_attributes_provider_status_without_parsing_error_bodies(
+    monkeypatch: pytest.MonkeyPatch,
+    status: int,
+    execution_status: str,
+    stop_reason: str,
+) -> None:
+    @contextlib.asynccontextmanager
+    async def stream_records(*_args: Any, **_kwargs: Any) -> AsyncIterator[FakeResponse]:
+        yield FakeResponse(('must-not-be-retained.example.com',), status=status)
+
+    monkeypatch.setattr(crtname.AsyncFetcher, 'stream_records', stream_records)
+    search = crtname.SearchCrtName('example.com')
+
+    await search.process()
+
+    assert await search.get_hostnames() == set()
+    assert search.execution_status == execution_status
+    assert search.stop_reason == stop_reason
+
+
+@pytest.mark.asyncio
+async def test_crt_name_preserves_valid_prefix_when_stream_fails(monkeypatch: pytest.MonkeyPatch) -> None:
+    @contextlib.asynccontextmanager
+    async def stream_records(*_args: Any, **_kwargs: Any) -> AsyncIterator[FakeResponse]:
+        yield FakeResponse(('api.example.com',), error=ResponseStreamError('response-limit'))
+
+    monkeypatch.setattr(crtname.AsyncFetcher, 'stream_records', stream_records)
+    search = crtname.SearchCrtName('example.com')
+
+    await search.process()
+
+    assert await search.get_hostnames() == {'api.example.com'}
+    assert search.execution_status == 'partial'
+    assert search.stop_reason == 'response-limit'
+
+
+@pytest.mark.asyncio
+async def test_crt_name_preserves_valid_prefix_at_runtime_limit(monkeypatch: pytest.MonkeyPatch) -> None:
+    class SlowResponse(FakeResponse):
+        def __aiter__(self) -> AsyncIterator[str]:
+            async def records() -> AsyncIterator[str]:
+                yield 'api.example.com'
+                await asyncio.Event().wait()
+
+            return records()
+
+    @contextlib.asynccontextmanager
+    async def stream_records(*_args: Any, **_kwargs: Any) -> AsyncIterator[FakeResponse]:
+        yield SlowResponse(())
+
+    monkeypatch.setattr(crtname.SearchCrtName, 'RUNTIME_SECONDS', 0.01)
+    monkeypatch.setattr(crtname.AsyncFetcher, 'stream_records', stream_records)
+    search = crtname.SearchCrtName('example.com')
+
+    await search.process()
+
+    assert await search.get_hostnames() == {'api.example.com'}
+    assert search.execution_status == 'partial'
+    assert search.stop_reason == 'runtime-limit'
+
+
+@pytest.mark.asyncio
+async def test_crt_name_empty_response_completes_without_results(monkeypatch: pytest.MonkeyPatch) -> None:
+    @contextlib.asynccontextmanager
+    async def stream_records(*_args: Any, **_kwargs: Any) -> AsyncIterator[FakeResponse]:
+        yield FakeResponse(())
+
+    monkeypatch.setattr(crtname.AsyncFetcher, 'stream_records', stream_records)
+    search = crtname.SearchCrtName('example.com')
+
+    await search.process()
+
+    assert await search.get_hostnames() == set()
+    assert search.execution_status == 'completed'
+    assert search.stop_reason == 'no-results'
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    'target',
+    [
+        '',
+        'localhost',
+        'https://example.com',
+        '192.0.2.1',
+        '192.168.001.001',
+        'example.com..',
+        '\u212a.example.com',
+        'straße.de',
+    ],
+)
+async def test_crt_name_rejects_invalid_targets_without_requesting(
+    monkeypatch: pytest.MonkeyPatch,
+    target: str,
+) -> None:
+    def unexpected_request(*_args: Any, **_kwargs: Any) -> None:
+        raise AssertionError('invalid targets must not reach crt.name')
+
+    monkeypatch.setattr(crtname.AsyncFetcher, 'stream_records', unexpected_request)
+    search = crtname.SearchCrtName(target)
+
+    await search.process()
+
+    assert search.execution_status == 'failed'
+    assert search.stop_reason == 'invalid-target'
+
+
+@pytest.mark.asyncio
+async def test_crt_name_rejects_non_ascii_provider_records(monkeypatch: pytest.MonkeyPatch) -> None:
+    @contextlib.asynccontextmanager
+    async def stream_records(*_args: Any, **_kwargs: Any) -> AsyncIterator[FakeResponse]:
+        yield FakeResponse(('\u212a.example.com', 'api.example.com'))
+
+    monkeypatch.setattr(crtname.AsyncFetcher, 'stream_records', stream_records)
+    search = crtname.SearchCrtName('example.com')
+
+    await search.process()
+
+    assert await search.get_hostnames() == {'api.example.com'}
+    assert search.execution_status == 'partial'
+    assert search.stop_reason == 'invalid-response'
+
+
+@pytest.mark.asyncio
+async def test_crt_name_propagates_cancellation(monkeypatch: pytest.MonkeyPatch) -> None:
+    cancelled = asyncio.CancelledError()
+
+    @contextlib.asynccontextmanager
+    async def stream_records(*_args: Any, **_kwargs: Any) -> AsyncIterator[FakeResponse]:
+        yield FakeResponse((), error=cancelled)
+
+    monkeypatch.setattr(crtname.AsyncFetcher, 'stream_records', stream_records)
+
+    with pytest.raises(asyncio.CancelledError) as raised:
+        await crtname.SearchCrtName('example.com').process()
+
+    assert raised.value is cancelled
+
+
+@pytest.mark.asyncio
+async def test_crt_name_and_crtsh_share_one_result_with_both_sources(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+) -> None:
+    completed_results: list[CompletedResult] = []
+
+    class FakeResultStore:
+        async def initialize(self) -> None:
+            return None
+
+        async def record_observations(self, *_args: object) -> None:
+            return None
+
+        async def save_run(self, result: CompletedResult) -> None:
+            completed_results.append(result)
+
+    class FakeCrtsh:
+        execution_status = 'completed'
+        stop_reason = None
+
+        def __init__(self, _word: str) -> None:
+            pass
+
+        async def process(self, _proxy: bool = False) -> None:
+            return None
+
+        async def get_hostnames(self) -> set[str]:
+            return {'shared.example.com'}
+
+    @contextlib.asynccontextmanager
+    async def stream_records(*_args: Any, **_kwargs: Any) -> AsyncIterator[FakeResponse]:
+        yield FakeResponse(('shared.example.com', 'only-crt-name.example.com'))
+
+    report = tmp_path / 'crt-name-overlap'
+    monkeypatch.setattr(crtname.AsyncFetcher, 'stream_records', stream_records)
+    monkeypatch.setattr(theharvester_main.crtsh, 'SearchCrtsh', FakeCrtsh)
+    monkeypatch.setattr(theharvester_main, 'ResultStore', FakeResultStore)
+    monkeypatch.setattr(
+        sys,
+        'argv',
+        ['theHarvester', '-d', 'example.com', '-b', 'crt-name,crtsh', '-f', str(report)],
+    )
+
+    with pytest.raises(SystemExit) as exit_info:
+        await theharvester_main.start()
+
+    assert exit_info.value.code == 0
+    assert {execution.source for execution in completed_results[-1].source_executions} == {'crt-name', 'crtsh'}
+    records = [json.loads(line) for line in report.with_suffix('.jsonl').read_text().splitlines()]
+    findings = {(record['type'], record['value']): record for record in records[1:]}
+    assert findings[('hostname', 'shared.example.com')]['sources'] == ['crt-name', 'crtsh']
+    assert findings[('hostname', 'only-crt-name.example.com')]['sources'] == ['crt-name']

tests/lib/test_core.py

@@ -579,6 +579,17 @@ async def test_stream_records_preserves_complete_prefix_before_response_limit(mo
     assert DummySession.instances[0].closed is True
 
 
+@pytest.mark.asyncio
+async def test_stream_records_accepts_final_record_ending_exactly_at_response_limit(monkeypatch) -> None:
+    install_stream_response(monkeypatch, chunks=(b'one\ntwo',))
+    monkeypatch.setattr(core_module, 'MAX_PROVIDER_STREAM_BYTES', 7)
+    lines: list[str] = []
+
+    await collect_default_stream(lines)
+
+    assert lines == ['one', 'two']
+
+
 @pytest.mark.asyncio
 async def test_stream_records_rejects_oversized_record_after_complete_prefix(monkeypatch) -> None:
     install_stream_response(monkeypatch, chunks=(b'ok\n123', b'456\n'))

tests/lib/test_source_catalog.py

@@ -3,7 +3,7 @@ from pathlib import Path
 
 from theHarvester.discovery import apisguru, bevigil, builtwith, gitlabsearch, intelxsearch, rocketreach, urlscan, zoomeyesearch
 from theHarvester.lib.core import Core
-from theHarvester.lib.source_catalog import SOURCE_SPECS, ResultRoute, SourceSpec, get_source_spec
+from theHarvester.lib.source_catalog import SOURCE_SPECS, ActivityClass, ResultRoute, SourceSpec, get_source_spec
 
 
 def _scheduled_source_names() -> list[str]:
@@ -125,3 +125,11 @@ def test_unavailable_venacus_source_is_not_selectable() -> None:
 
 def test_source_lookup_preserves_case_insensitive_legacy_labels() -> None:
     assert get_source_spec('CRTsh') is SOURCE_SPECS['crtsh']
+
+
+def test_crt_name_is_a_separate_passive_hostname_source() -> None:
+    spec = get_source_spec('CRT-NAME')
+
+    assert spec is SOURCE_SPECS['crt-name']
+    assert spec.routes == frozenset({ResultRoute.SUBDOMAINS})
+    assert spec.activity is ActivityClass.PASSIVE

tests/test_readme.py

@@ -81,8 +81,8 @@ def test_readme_matches_declared_source_contracts() -> None:
     declared = _declared_source_contracts()
 
     assert '| Source | Subdomains | Emails | IPs | ASNs | URLs | People | Breaches |' in readme
-    assert len(declared) == 58
-    assert len(documented) == 58
+    assert len(declared) == 59
+    assert len(documented) == 59
     assert documented == declared
     assert {'securitytrails', 'shodaninternetdb'}.isdisjoint(documented)
 

theHarvester/__main__.py

@@ -38,6 +38,7 @@ from theHarvester.discovery import (
     chaos,
     commoncrawl,
     criminalip,
+    crtname,
     crtsh,
     dnsdb,
     dnssearch,
@@ -1082,6 +1083,13 @@ async def start(
                         else:
                             show_default_error_message(engineitem, word, e)
 
+                elif engineitem == 'crt-name':
+                    try:
+                        crt_name_search = crtname.SearchCrtName(word)
+                        stor_lst.append(store(crt_name_search, engineitem))
+                    except Exception as e:
+                        show_default_error_message(engineitem, word, e)
+
                 elif engineitem == 'crtsh':
                     try:
                         crtsh_search = crtsh.SearchCrtsh(word)

theHarvester/discovery/crtname.py

@@ -0,0 +1,109 @@
+import asyncio
+
+from theHarvester.lib.core import AsyncFetcher, ResponseStreamError
+from theHarvester.lib.hostnames import normalize_scoped_hostname
+
+
+class SearchCrtName:
+    """Collect hostname candidates for the exact operator-requested scope."""
+
+    ENDPOINT = 'https://crt.name/v1/search'
+    RUNTIME_SECONDS = 90
+
+    def __init__(self, word: str) -> None:
+        self.word = self._normalize_scope(word)
+        self.hostnames: set[str] = set()
+        self.execution_status: str | None = None
+        self.stop_reason: str | None = None
+
+    @staticmethod
+    def _valid_hostname(value: str) -> bool:
+        if not value or len(value) > 253 or not value.isascii():
+            return False
+        labels = value.split('.')
+        return all(
+            label
+            and len(label) <= 63
+            and label[0].isalnum()
+            and label[-1].isalnum()
+            and all(character.isalnum() or character == '-' for character in label)
+            for label in labels
+        )
+
+    @classmethod
+    def _normalize_scope(cls, value: str) -> str:
+        stripped = value.strip()
+        if not stripped.isascii():
+            return ''
+        normalized = stripped.lower().removesuffix('.')
+        if normalized.endswith('.') or not cls._valid_hostname(normalized) or len(normalized.split('.')) < 2:
+            return ''
+        labels = normalized.split('.')
+        if len(labels) == 4 and all(label.isdigit() and len(label) <= 3 and int(label) <= 255 for label in labels):
+            return ''
+        return normalized
+
+    def _stop(self, status: str, reason: str) -> None:
+        self.execution_status = 'partial' if self.hostnames else status
+        self.stop_reason = reason
+
+    async def _collect(self, proxy: bool) -> None:
+        async with AsyncFetcher.stream_records(
+            self.ENDPOINT,
+            framing='ndjson',
+            params={'apex': self.word},
+            headers={'Accept': 'text/plain'},
+            proxy=proxy,
+            follow_redirects=False,
+            request_timeout=self.RUNTIME_SECONDS,
+        ) as response:
+            if response.status == 429:
+                self._stop('rate-limited', 'http-429')
+                return
+            if response.status in {401, 403}:
+                self._stop('failed', 'access-denied')
+                return
+            if not 200 <= response.status < 300:
+                self._stop('failed', f'http-{response.status}')
+                return
+
+            malformed = False
+            async for record in response:
+                candidate = record.strip()
+                if not candidate:
+                    continue
+                if not candidate.isascii():
+                    malformed = True
+                    continue
+                candidate = candidate.lower().removeprefix('*.').removesuffix('.')
+                if not self._valid_hostname(candidate):
+                    malformed = True
+                    continue
+                normalized = normalize_scoped_hostname(candidate, self.word)
+                if normalized is None:
+                    malformed = True
+                elif normalized != self.word:
+                    self.hostnames.add(normalized)
+
+            if malformed:
+                self._stop('failed', 'invalid-response')
+            else:
+                self.execution_status = 'completed'
+                self.stop_reason = None if self.hostnames else 'no-results'
+
+    async def process(self, proxy: bool = False) -> None:
+        self.execution_status = None
+        self.stop_reason = None
+        if not self.word:
+            self._stop('failed', 'invalid-target')
+            return
+        try:
+            async with asyncio.timeout(self.RUNTIME_SECONDS):
+                await self._collect(proxy)
+        except ResponseStreamError as error:
+            self._stop('failed', error.reason)
+        except TimeoutError:
+            self._stop('failed', 'runtime-limit')
+
+    async def get_hostnames(self) -> set[str]:
+        return self.hostnames

theHarvester/lib/core.py

@@ -412,6 +412,7 @@ class Core:
             'chaos',
             'commoncrawl',
             'criminalip',
+            'crt-name',
             'crtsh',
             'dehashed',
             'dnsdb',

theHarvester/lib/source_catalog.py

@@ -108,6 +108,7 @@ _SPECS = (
         ResultRoute.ASNS,
         activity=ActivityClass.DIRECT,
     ),
+    _spec('crt-name', ResultRoute.SUBDOMAINS),
     _spec('crtsh', ResultRoute.SUBDOMAINS),
     _spec('dehashed', ResultRoute.EMAILS, ResultRoute.IPS),
     _spec('dnsdb', ResultRoute.SUBDOMAINS),