Subdomain Enumeration Tool

Subdomain Enumeration Tool: A Production-Ready Tutorial

Problem Statement

During security assessments and reconnaissance, one of the first tasks is mapping an organization's attack surface. DNS records often expose staging servers, internal admin panels, and forgotten development environments that aren't linked anywhere on the main website — but they're still publicly resolvable. Manually running dig against hundreds of candidate names is tedious and error-prone.

This script automates subdomain discovery by combining two complementary techniques: DNS brute-forcing (testing a wordlist of common subdomain prefixes against the target domain) and passive enumeration via Certificate Transparency logs (querying crt.sh to find subdomains that appear in public TLS certificates). By running both in parallel with asyncio, it can enumerate thousands of candidates in seconds rather than minutes.

How It Works

The tool is built around an async producer/consumer pattern. A wordlist is loaded into memory, then split into a bounded asyncio.Queue that worker coroutines pull from. Each worker sends a DNS query for <prefix>.<domain> using dnspython's async resolver, which is wrapped in asyncio.to_thread() because dnspython's async API depends on having a running event loop with specific backend support. Concurrently, a separate task queries the crt.sh JSON API via aiohttp and parses out unique subdomain names from certificate entries.

Results are deduplicated in a shared set protected by an asyncio.Lock, then printed as they're discovered (sorted output at the end). A --concurrency flag controls worker count to avoid hammering the resolver, and a --timeout parameter bounds each DNS lookup. The design separates concerns: dns_worker handles brute-force, ct_lookup handles passive discovery, and main orchestrates them.

Full Script

#!/usr/bin/env python3
"""
Subdomain Enumeration Tool

Discovers subdomains for a target domain using two techniques:
  1. DNS brute-force against a supplied wordlist
  2. Passive enumeration via Certificate Transparency logs (crt.sh)

Usage:
    python3 subdomain_enum.py --domain example.com --wordlist wordlist.txt
"""

import argparse
import asyncio
import json
import os
import sys
from typing import Iterable

import aiohttp
import dns.asyncresolver
import dns.exception
import dns.resolver

# ---------------------------------------------------------------------------
# Configuration (override via environment variables)
# ---------------------------------------------------------------------------
DEFAULT_TIMEOUT = float(os.environ.get("SUBENUM_TIMEOUT", "3.0"))
DEFAULT_CONCURRENCY = int(os.environ.get("SUBENUM_CONCURRENCY", "50"))
CT_API_URL = os.environ.get("SUBENUM_CT_URL", "https://crt.sh/?q=%25.{domain}&output=json")
USER_AGENT = os.environ.get("SUBENUM_UA", "subdomain-enum/1.0 (+security-recon)")


class SubdomainEnumerator:
    """Coordinates DNS brute-force and CT log lookups for a single domain."""

    def __init__(self, domain: str, timeout: float, concurrency: int) -> None:
        self.domain = domain.rstrip(".")
        self.timeout = timeout
        self.concurrency = concurrency
        self.found: set[str] = set()
        self._lock = asyncio.Lock()
        # A single shared resolver with a tuned timeout and retry policy.
        self.resolver = dns.asyncresolver.Resolver()
        self.resolver.timeout = timeout
        self.resolver.lifetime = timeout
        self.resolver.nameservers = ["1.1.1.1", "8.8.8.8"]

    async def _record(self, name: str) -> None:
        """Add a discovered subdomain to the shared set and print it live."""
        name = name.lower().strip().rstrip(".")
        if not name or name == self.domain or not name.endswith(self.domain):
            return
        async with self._lock:
            if name not in self.found:
                self.found.add(name)
                print(f"[+] {name}")

    async def _resolve(self, fqdn: str) -> bool:
        """Return True if the FQDN resolves to at least one A/AAAA record."""
        try:
            # dnspython's async resolver needs an active loop; wrap in to_thread
            # for maximum compatibility across dnspython versions/backends.
            await asyncio.to_thread(
                self.resolver.resolve, fqdn, "A", raise_on_no_answer=False
            )
            return True
        except (dns.resolver.NXDOMAIN, dns.resolver.NoAnswer):
            return False
        except dns.resolver.NoNameservers:
            # Transient resolver failure — treat as non-existent rather than crash.
            return False
        except dns.exception.Timeout:
            return False
        except Exception as exc:  # noqa: BLE001 - defensive; log and continue
            print(f"[!] Unexpected DNS error for {fqdn}: {exc}", file=sys.stderr)
            return False

    async def dns_worker(self, queue: "asyncio.Queue[str]") -> None:
        """Consume prefixes from the queue and test each as a subdomain."""
        while True:
            try:
                prefix = queue.get_nowait()
            except asyncio.QueueEmpty:
                return
            try:
                fqdn = f"{prefix}.{self.domain}"
                if await self._resolve(fqdn):
                    await self._record(fqdn)
            finally:
                queue.task_done()

    async def brute_force(self, wordlist: Iterable[str]) -> None:
        """Fan out wordlist entries across a pool of DNS worker coroutines."""
        queue: "asyncio.Queue[str]" = asyncio.Queue()
        for raw in wordlist:
            prefix = raw.strip()
            if prefix and not prefix.startswith("#"):
                queue.put_nowait(prefix)
        if queue.empty():
            return
        workers = [
            asyncio.create_task(self.dns_worker(queue))
            for _ in range(min(self.concurrency, queue.qsize()))
        ]
        await asyncio.gather(*workers)

    async def ct_lookup(self, session: aiohttp.ClientSession) -> None:
        """Query crt.sh for certificate entries mentioning the target domain."""
        url = CT_API_URL.format(domain=self.domain)
        try:
            async with session.get(url, timeout=aiohttp.ClientTimeout(total=30)) as resp:
                if resp.status != 200:
                    print(f"[!] crt.sh returned HTTP {resp.status}", file=sys.stderr)
                    return
                payload = await resp.json(content_type=None)
        except (aiohttp.ClientError, asyncio.TimeoutError, json.JSONDecodeError) as exc:
            print(f"[!] Certificate Transparency lookup failed: {exc}", file=sys.stderr)
            return

        for entry in payload if isinstance(payload, list) else []:
            # The "name_value" field can contain newline-separated names and
            # wildcards like *.example.com — handle both.
            for name in str(entry.get("name_value", "")).splitlines():
                name = name.replace("*.", "").strip().lower()
                if name.endswith(self.domain):
                    await self._record(name)

    async def run(self, wordlist: Iterable[str]) -> list[str]:
        """Run both enumeration strategies concurrently and return sorted hits."""
        headers = {"User-Agent": USER_AGENT}
        async with aiohttp.ClientSession(headers=headers) as session:
            await asyncio.gather(
                self.brute_force(wordlist),
                self.ct_lookup(session),
            )
        return sorted(self.found)


def load_wordlist(path: str) -> list[str]:
    """Read a newline-delimited wordlist; raise if the file is missing."""
    if not os.path.isfile(path):
        raise FileNotFoundError(f"Wordlist not found: {path}")
    with open(path, "r", encoding="utf-8", errors="ignore") as fh:
        return [line.strip() for line in fh if line.strip()]


def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Async subdomain enumerator")
    parser.add_argument("--domain", required=True, help="Root domain, e.g. example.com")
    parser.add_argument("--wordlist", required=True, help="Path to subdomain wordlist")
    parser.add_argument(
        "--concurrency", type=int, default=DEFAULT_CONCURRENCY,
        help=f"Parallel DNS workers (default: {DEFAULT_CONCURRENCY})",
    )
    parser.add_argument(
        "--timeout", type=float, default=DEFAULT_TIMEOUT,
        help=f"Per-query DNS timeout in seconds (default: {DEFAULT_TIMEOUT})",
    )
    parser.add_argument(
        "--no-ct", action="store_true",
        help="Skip Certificate Transparency lookup",
    )
    return parser.parse_args(argv)


async def async_main(args: argparse.Namespace) -> int:
    try:
        words = load_wordlist(args.wordlist)
    except FileNotFoundError as exc:
        print(f"error: {exc}", file=sys.stderr)
        return 2

    enumerator = SubdomainEnumerator(
        domain=args.domain,
        timeout=args.timeout,
        concurrency=args.concurrency,
    )

    if args.no_ct:
        await enumerator.brute_force(words)
        results = sorted(enumerator.found)
    else:
        results = await enumerator.run(words)

    print(f"\n[*] Enumeration complete: {len(results)} unique subdomain(s).")
    return 0


def main() -> int:
    args = parse_args()
    try:
        return asyncio.run(async_main(args))
    except KeyboardInterrupt:
        print("\n[!] Interrupted by user.", file=sys.stderr)
        return 130


if __name__ == "__main__":
    sys.exit(main())

Configuration

Usage Examples

1. Standard scan with the default wordlist:

$ python3 subdomain_enum.py --domain example.com --wordlist wordlist.txt
[+] api.example.com
[+] mail.example.com
[+] staging.example.com
[+] www.example.com

[*] Enumeration complete: 4 unique subdomain(s).

2. Aggressive scan with higher concurrency and shorter timeout:

$ python3 subdomain_enum.py --domain example.com --wordlist big.txt --concurrency 200 --timeout 1.5

3. Pure brute-force (no external CT API calls):

$ python3 subdomain_enum.py --domain example.com --wordlist wordlist.txt --no-ct

Error Handling

• Missing wordlist — load_wordlist() raises FileNotFoundError, which is caught in async_main and exits with code 2 plus a clear message.
• NXDOMAIN / NoAnswer — Treated as "subdomain does not exist" and silently skipped.
• DNS timeouts — Caught per-query so a single slow name doesn't stall the worker pool.
• NoNameservers — Indicates a resolver-level failure (e.g., rate limiting); logged implicitly by returning False so enumeration continues.
• CT API failures — HTTP errors, timeouts, and JSON parse errors are all caught and printed to stderr; the DNS brute-force continues unaffected.
• Ctrl-C — Handled at the top level, prints a clean message, and exits with code 130 (standard for SIGINT).
• Unexpected exceptions — A defensive except Exception in _resolve logs the error and keeps the worker alive rather than killing the whole run.

Extending Ideas

• Recursive enumeration — Feed newly discovered subdomains back into the queue as new root domains to catch deeper trees like dev.api.example.com.
• Multiple output formats — Add --format json|csv|txt and write results to a file for ingestion by tools like nmap, httpx, or a SIEM.
• Wildcard detection — Before brute-forcing, resolve a random string like xyzabc123.example.com; if it resolves, the domain uses a wildcard DNS record, and you should filter out false positives.
• Additional passive sources — Integrate SecurityTrails, VirusTotal, or PassiveTotal APIs alongside crt.sh for broader coverage; abstract them behind a common PassiveSource protocol.