During security assessments and threat modeling, security engineers and penetration testers must map out an organization's external attack surface. Direct
During security assessments and threat modeling, security engineers and penetration testers must map out an organization's external attack surface. Direct active scanning (such as aggressive port scanning or brute-forcing subdomains) often triggers Intrusion Detection Systems (IDS), triggers Web Application Firewalls (WAFs), or violates rules of engagement prior to formal authorization.
This script solves that problem by performing passive reconnaissance. It aggregates publicly exposed subdomain and endpoint data from third-party OSINT sources (Certificate Transparency logs, web archives, and DNS aggregator APIs) without sending a single packet directly to the target infrastructure. It consolidates these disparate feeds into a single deduplicated dataset, optionally resolving IP addresses via external recursive DNS resolvers.
The tool uses a modular, multi-source architecture. When executed, it initializes an aggregator orchestrator that queries requested OSINT sources in parallel using thread pools to ensure low latency.
Data flows through three core phases:
dnspython queries configured upstream DNS resolvers to resolve active A records for discovered hostnames. Results are formatted as structured JSON or human-readable tables.#!/usr/bin/env python3
"""
Passive Reconnaissance Aggregator
---------------------------------
Aggregates subdomains and historical target data from passive OSINT sources
including CRT.sh, Wayback Machine, and HackerTarget without directly contacting
the target servers.
Dependencies: requests, dnspython
Usage: python3 passive_recon.py --target example.com --sources certsh,wayback,hackertarget
"""
import argparse
import concurrent.futures
import json
import logging
import os
import re
import sys
from typing import Any
from urllib.parse import urlparse
import dns.resolver
import requests
# Config section with environment variables
DEFAULT_TIMEOUT = int(os.environ.get("RECON_TIMEOUT", "15"))
DEFAULT_USER_AGENT = os.environ.get(
"RECON_USER_AGENT",
"PassiveReconAggregator/1.0 (+https://github.com/security-tools)"
)
DEFAULT_MAX_WORKERS = int(os.environ.get("RECON_MAX_THREADS", "10"))
# Configure logging
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(message)s",
handlers=[logging.StreamHandler(sys.stderr)]
)
logger = logging.getLogger("PassiveRecon")
class ReconAggregator:
"""Orchestrates passive target collection across multiple OSINT services."""
def __init__(self, target: str, timeout: int = DEFAULT_TIMEOUT):
self.target = target.lower().strip()
self.timeout = timeout
self.headers = {"User-Agent": DEFAULT_USER_AGENT}
self.session = requests.Session()
self.session.headers.update(self.headers)
def _sanitize_domain(self, domain: str) -> str | None:
"""Sanitizes raw subdomain strings extracted from third-party logs."""
domain = domain.strip().lower()
if domain.startswith("*."):
domain = domain[2:]
if domain.startswith("."):
domain = domain[1:]
# Basic domain format validation
domain_regex = r"^([a-z0-9]+(-[a-z0-9]+)*\.)+[a-z]{2,}$"
if re.match(domain_regex, domain) and (domain == self.target or domain.endswith(f".{self.target}")):
return domain
return None
def fetch_certsh(self) -> set[str]:
"""Fetch hostnames from Certificate Transparency logs via CRT.sh."""
logger.info("Querying CRT.sh Certificate Transparency logs...")
subdomains: set[str] = set()
url = f"https://crt.sh/?q=%.{self.target}&output=json"
try:
response = self.session.get(url, timeout=self.timeout)
if response.status_code == 200:
data = response.json()
for entry in data:
name_value = entry.get("name_value", "")
# Certificate fields can contain multiline SAN entries
for name in name_value.split("\n"):
clean_name = self._sanitize_domain(name)
if clean_name:
subdomains.add(clean_name)
else:
logger.warning(f"CRT.sh returned unexpected HTTP status: {response.status_code}")
except requests.RequestException as err:
logger.error(f"Failed fetching CRT.sh data: {err}")
except json.JSONDecodeError:
logger.error("Failed to parse JSON response from CRT.sh")
return subdomains
def fetch_wayback(self) -> set[str]:
"""Fetch historical hostnames from the Wayback Machine CDX API."""
logger.info("Querying Wayback Machine CDX Index...")
subdomains: set[str] = set()
url = (
f"http://web.archive.org/cdx/search/cdx?"
f"url=*.{self.target}/*&output=json&fl=original&collapse=urlkey&limit=5000"
)
try:
response = self.session.get(url, timeout=self.timeout)
if response.status_code == 200:
data = response.json()
# Skip header row if present
rows = data[1:] if len(data) > 0 and data[0][0] == "original" else data
for row in rows:
if not row:
continue
raw_url = row[0]
parsed = urlparse(raw_url)
hostname = parsed.netloc.split(":")[0] # Remove port if present
clean_name = self._sanitize_domain(hostname)
if clean_name:
subdomains.add(clean_name)
else:
logger.warning(f"Wayback API returned unexpected HTTP status: {response.status_code}")
except requests.RequestException as err:
logger.error(f"Failed fetching Wayback Machine data: {err}")
except json.JSONDecodeError:
logger.error("Failed to parse JSON response from Wayback API")
return subdomains
def fetch_hackertarget(self) -> set[str]:
"""Fetch hostnames from HackerTarget host search API."""
logger.info("Querying HackerTarget Host Search API...")
subdomains: set[str] = set()
url = f"https://api.hackertarget.com/hostsearch/?q={self.target}"
try:
response = self.session.get(url, timeout=self.timeout)
if response.status_code == 200:
text = response.text
if "API count exceed" in text:
logger.warning("HackerTarget API limit reached for current IP.")
return subdomains
for line in text.splitlines():
parts = line.split(",")
if parts:
clean_name = self._sanitize_domain(parts[0])
if clean_name:
subdomains.add(clean_name)
else:
logger.warning(f"HackerTarget returned unexpected HTTP status: {response.status_code}")
except requests.RequestException as err:
logger.error(f"Failed fetching HackerTarget data: {err}")
return subdomains
def resolve_subdomain(domain: str) -> dict[str, Any]:
"""Resolves DNS IPv4 addresses (A records) for a given domain using dnspython."""
resolver = dns.resolver.Resolver()
resolver.timeout = 3.0
resolver.lifetime = 3.0
ips: list[str] = []
try:
answers = resolver.resolve(domain, "A")
ips = [str(rdata) for rdata in answers]
except (dns.resolver.NXDOMAIN, dns.resolver.NoAnswer, dns.resolver.NoNameservers):
pass
except dns.exception.DNSException as err:
logger.debug(f"DNS resolution error for {domain}: {err}")
return {"domain": domain, "ips": ips, "active": len(ips) > 0}
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(
description="Passive Reconnaissance Aggregator: Aggregate targets without direct interaction."
)
parser.add_argument(
"--target",
required=True,
help="Base domain to target (e.g., example.com)"
)
parser.add_argument(
"--sources",
default="certsh,wayback,hackertarget",
help="Comma-separated list of sources to query (certsh, wayback, hackertarget)"
)
parser.add_argument(
"--resolve",
action="store_true",
help="Attempt to perform DNS IPv4 resolution on collected hosts"
)
parser.add_argument(
"--json",
dest="json_output",
action="store_true",
help="Output final structured results as JSON"
)
return parser.parse_args()
def main() -> None:
args = parse_args()
target_domain = args.target.strip().lower()
selected_sources = [s.strip().lower() for s in args.sources.split(",") if s.strip()]
aggregator = ReconAggregator(target=target_domain)
discovered_hosts: set[str] = set()
# Map available methods
source_map = {
"certsh": aggregator.fetch_certsh,
"wayback": aggregator.fetch_wayback,
"hackertarget": aggregator.fetch_hackertarget,
}
# Execute source collection in parallel
with concurrent.futures.ThreadPoolExecutor(max_workers=DEFAULT_MAX_WORKERS) as executor:
future_to_source = {
executor.submit(source_map[src]): src
for src in selected_sources
if src in source_map
}
for future in concurrent.futures.as_completed(future_to_source):
source_name = future_to_source[future]
try:
hosts = future.result()
logger.info(f"Source '{source_name}' returned {len(hosts)} host(s).")
discovered_hosts.update(hosts)
except Exception as err:
logger.error(f"Error running source '{source_name}': {err}")
logger.info(f"Total unique hosts discovered across all sources: {len(discovered_hosts)}")
results: list[dict[str, Any]] = []
if args.resolve and discovered_hosts:
logger.info("Resolving active IP addresses for collected hosts...")
with concurrent.futures.ThreadPoolExecutor(max_workers=DEFAULT_MAX_WORKERS) as executor:
resolved_results = list(executor.map(resolve_subdomain, sorted(discovered_hosts)))
results = resolved_results
else:
results = [{"domain": host, "ips": [], "active": None} for host in sorted(discovered_hosts)]
if args.json_output:
output_data = {
"target": target_domain,
"total_count": len(results),
"results": results,
}
print(json.dumps(output_data, indent=2))
else:
print("\n" + "=" * 60)
print(f" RECON RESULTS FOR: {target_domain}")
print("=" * 60)
for entry in results:
ip_str = f" [{', '.join(entry['ips'])}]" if entry["ips"] else ""
status = " [ACTIVE]" if entry["active"] is True else ""
print(f" -> {entry['domain']}{status}{ip_str}")
print("=" * 60)
if __name__ == "__main__":
main()
The script leverages environment variables for non-sensitive runtime adjustments. You can set these directly in your shell session or Dockerfile:
RECON_TIMEOUT: Custom HTTP request timeout in seconds (Default: 15).RECON_USER_AGENT: Custom HTTP User-Agent header passed to third-party APIs (Default: PassiveReconAggregator/1.0 (+https://github.com/security-tools)).RECON_MAX_THREADS: Number of concurrent workers for executing HTTP requests and DNS resolutions (Default: 10).Example export before execution:
export RECON_TIMEOUT=30
export RECON_MAX_THREADS=20
Runs all passive sources against example.com and outputs formatted standard text:
python3 passive_recon.py --target example.com --sources certsh,wayback,hackertarget
Output:
2023-10-27 10:15:00 [INFO] Querying CRT.sh Certificate Transparency logs...
2023-10-27 10:15:00 [INFO] Querying Wayback Machine CDX Index...
2023-10-27 10:15:00 [INFO] Querying HackerTarget Host Search API...
2023-10-27 10:15:02 [INFO] Source 'hackertarget' returned 2 host(s).
2023-10-27 10:15:03 [INFO] Source 'certsh' returned 5 host(s).
2023-10-27 10:15:04 [INFO] Source 'wayback' returned 12 host(s).
2023-10-27 10:15:04 [INFO] Total unique hosts discovered across all sources: 14
============================================================
RECON RESULTS FOR: example.com
============================================================
-> api.example.com
-> dev.example.com
-> example.com
-> mail.example.com
-> m.example.com
-> staging.example.com
-> www.example.com
============================================================
Pass --resolve and --json to verify host status via public DNS resolution and pipe into tools like jq:
python3 passive_recon.py --target example.com --sources certsh,hackertarget --resolve --json
Output:
{
"target": "example.com",
"total_count": 3,
"results": [
{
"domain": "api.example.com",
"ips": [
"93.184.216.34"
],
"active": true
},
{
"domain": "dev-old.example.com",
"ips": [],
"active": false
},
{
"domain": "www.example.com",
"ips": [
"93.184.216.34"
],
"active": true
}
]
}
requests.RequestException), logged gracefully as warnings or errors, and do not crash the pipeline.json.JSONDecodeError blocks to prevent parser crashes on HTML error pages.dns.resolver.NXDOMAIN or dns.resolver.NoAnswer exceptions. These are handled silently during resolution, marking active: false without interrupting execution for remaining hosts.ReconAggregator with methods querying paid or key-based APIs like Shodan, VirusTotal, SecurityTrails, or Censys by inspecting os.environ.get("SHODAN_API_KEY").