Yes. Here’s a safe starter recon tool aimed at authorized bug-bounty scopes. It does passive subdomain discovery from certificate transparency, DNS resolution, HTTP/HTTPS probing, basic header collection, and JSONL output. It deliberately avoids brute forcing, exploit checks, credential attacks, or high-rate scanning.

python
#!/usr/bin/env python3
"""
bb_recon.py

Authorized bug-bounty reconnaissance helper.

Features:
- Enforces an explicit domain scope
- Passive subdomain discovery using crt.sh
- DNS resolution
- HTTP/HTTPS probing
- Collects status, title, server header, redirects, and security headers
- JSONL output
- Conservative concurrency/rate behavior

Usage:
    python bb_recon.py example.com
    python bb_recon.py example.com --output results.jsonl
    python bb_recon.py example.com --workers 5 --timeout 7

Only use this against systems you have permission to test.
"""

import argparse
import concurrent.futures
import json
import re
import socket
import ssl
import sys
import time
from html import unescape
from urllib.error import HTTPError, URLError
from urllib.parse import urlparse
from urllib.request import Request, urlopen


USER_AGENT = "Authorized-Bug-Bounty-Recon/1.0"

SECURITY_HEADERS = [
    "strict-transport-security",
    "content-security-policy",
    "x-content-type-options",
    "x-frame-options",
    "referrer-policy",
    "permissions-policy",
]


def normalize_domain(domain: str) -> str:
    domain = domain.strip().lower().rstrip(".")

    if domain.startswith("*."):
        domain = domain[2:]

    if not re.fullmatch(
        r"(?:[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?\.)+[a-z]{2,63}",
        domain,
    ):
        raise ValueError(f"Invalid domain: {domain}")

    return domain


def in_scope(hostname: str, root_domain: str) -> bool:
    """
    Only permit the root domain or its subdomains.
    Prevents accidentally following recon onto unrelated domains.
    """
    hostname = hostname.lower().rstrip(".")
    root_domain = root_domain.lower().rstrip(".")

    return hostname == root_domain or hostname.endswith("." + root_domain)


def crtsh_subdomains(domain: str, timeout: int = 10) -> set[str]:
    """
    Passive certificate transparency lookup.
    """
    url = f"https://crt.sh/?q=%25.{domain}&output=json"

    req = Request(
        url,
        headers={
            "User-Agent": USER_AGENT,
            "Accept": "application/json",
        },
    )

    results = {domain}

    try:
        with urlopen(req, timeout=timeout) as response:
            data = json.loads(response.read().decode("utf-8", errors="replace"))

        for entry in data:
            names = entry.get("name_value", "")

            for name in names.splitlines():
                name = name.strip().lower().rstrip(".")

                if name.startswith("*."):
                    name = name[2:]

                if in_scope(name, domain):
                    results.add(name)

    except Exception as exc:
        print(f"[!] crt.sh lookup failed: {exc}", file=sys.stderr)

    return results


def resolve_host(hostname: str) -> dict:
    """
    Resolve IPv4/IPv6 addresses.
    """
    ips = set()

    try:
        for result in socket.getaddrinfo(
            hostname,
            None,
            family=socket.AF_UNSPEC,
            type=socket.SOCK_STREAM,
        ):
            ips.add(result[4][0])

        return {
            "hostname": hostname,
            "resolved": True,
            "ips": sorted(ips),
        }

    except socket.gaierror:
        return {
            "hostname": hostname,
            "resolved": False,
            "ips": [],
        }


def extract_title(body: str) -> str | None:
    match = re.search(
        r"<title[^>]*>(.*?)</title>",
        body,
        flags=re.IGNORECASE | re.DOTALL,
    )

    if not match:
        return None

    title = re.sub(r"\s+", " ", unescape(match.group(1))).strip()

    return title[:200]


class ScopeRedirectHandler:
    """
    Helper used to identify redirect targets.

    Redirect destinations are recorded, but callers can verify that the final
    hostname remains inside the authorized scope.
    """
    pass


def probe_url(url: str, root_domain: str, timeout: int) -> dict:
    result = {
        "url": url,
        "reachable": False,
    }

    req = Request(
        url,
        headers={
            "User-Agent": USER_AGENT,
            "Accept": "text/html,application/xhtml+xml,*/*;q=0.8",
        },
    )

    try:
        ctx = ssl.create_default_context()

        with urlopen(req, timeout=timeout, context=ctx) as response:
            final_url = response.geturl()
            final_host = urlparse(final_url).hostname or ""

            # Do not treat an off-scope redirect as an authorized target.
            if not in_scope(final_host, root_domain):
                return {
                    "url": url,
                    "reachable": True,
                    "status": response.status,
                    "redirected_off_scope": True,
                    "final_url": final_url,
                }

            content_type = response.headers.get("Content-Type", "")

            # Read only a small response sample.
            body = response.read(256 * 1024)

            try:
                text = body.decode("utf-8", errors="replace")
            except Exception:
                text = ""

            headers = {
                key.lower(): value
                for key, value in response.headers.items()
            }

            result.update(
                {
                    "reachable": True,
                    "status": response.status,
                    "final_url": final_url,
                    "content_type": content_type,
                    "server": headers.get("server"),
                    "title": extract_title(text)
                    if "html" in content_type.lower()
                    else None,
                    "security_headers": {
                        header: headers.get(header)
                        for header in SECURITY_HEADERS
                    },
                }
            )

    except HTTPError as exc:
        result.update(
            {
                "reachable": True,
                "status": exc.code,
                "error": str(exc),
            }
        )

    except (URLError, TimeoutError, socket.timeout, ssl.SSLError) as exc:
        result["error"] = str(exc)

    except Exception as exc:
        result["error"] = f"{type(exc).__name__}: {exc}"

    return result


def inspect_host(hostname: str, root_domain: str, timeout: int) -> dict:
    dns = resolve_host(hostname)

    record = {
        "hostname": hostname,
        "dns": dns,
        "http": [],
    }

    if not dns["resolved"]:
        return record

    for scheme in ("https", "http"):
        url = f"{scheme}://{hostname}/"

        record["http"].append(
            probe_url(
                url=url,
                root_domain=root_domain,
                timeout=timeout,
            )
        )

    return record


def save_jsonl(filename: str, records: list[dict]) -> None:
    with open(filename, "w", encoding="utf-8") as f:
        for record in records:
            f.write(json.dumps(record, ensure_ascii=False) + "\n")


def main():
    parser = argparse.ArgumentParser(
        description="Scope-aware bug bounty reconnaissance helper"
    )

    parser.add_argument(
        "domain",
        help="Authorized root domain, e.g. example.com",
    )

    parser.add_argument(
        "-o",
        "--output",
        default="recon.jsonl",
        help="JSONL output file",
    )

    parser.add_argument(
        "--workers",
        type=int,
        default=5,
        help="Maximum concurrent hosts (default: 5, max: 20)",
    )

    parser.add_argument(
        "--timeout",
        type=int,
        default=7,
        help="Network timeout in seconds",
    )

    args = parser.parse_args()

    try:
        domain = normalize_domain(args.domain)
    except ValueError as exc:
        parser.error(str(exc))

    workers = max(1, min(args.workers, 20))

    print(f"[*] Authorized scope: {domain}")
    print("[*] Querying certificate transparency...")

    hosts = sorted(crtsh_subdomains(domain, args.timeout))

    print(f"[*] Found {len(hosts)} candidate hostnames")
    print(f"[*] Probing with {workers} workers...")

    results = []

    with concurrent.futures.ThreadPoolExecutor(
        max_workers=workers
    ) as executor:

        futures = {
            executor.submit(
                inspect_host,
                host,
                domain,
                args.timeout,
            ): host
            for host in hosts
        }

        for future in concurrent.futures.as_completed(futures):
            host = futures[future]

            try:
                result = future.result()
                results.append(result)

                live = [
                    x
                    for x in result["http"]
                    if x.get("reachable")
                ]

                ips = ", ".join(result["dns"].get("ips", []))

                if live:
                    statuses = ", ".join(
                        f'{x["url"]}={x.get("status", "?")}'
                        for x in live
                    )

                    print(f"[+] {host:<45} {ips:<30} {statuses}")
                elif result["dns"]["resolved"]:
                    print(f"[-] {host:<45} {ips}")

            except Exception as exc:
                print(f"[!] {host}: {exc}", file=sys.stderr)

            # Small courtesy delay between completed jobs.
            time.sleep(0.05)

    results.sort(key=lambda x: x["hostname"])
    save_jsonl(args.output, results)

    print()
    print(f"[*] Saved {len(results)} records to {args.output}")


if __name__ == "__main__":
    main()

Run it with:

bash
python3 bb_recon.py example.com

A record will look roughly like:

json
{
  "hostname": "api.example.com",
  "dns": {
    "resolved": true,
    "ips": ["203.0.113.10"]
  },
  "http": [
    {
      "url": "https://api.example.com/",
      "reachable": true,
      "status": 200,
      "title": "API",
      "server": "nginx"
    }
  ]
}

A useful next version could add scope-file parsing for HackerOne/Bugcrowd programs, wildcard handling, passive sources, DNS/CNAME inspection, TLS certificate metadata, favicon hashing, screenshots, deduplication, SQLite storage, and an HTML dashboard, while keeping all active requests restricted to the declared bounty scope.

Below is a Python implementation built around an explicit authorization boundary: every input and every passively discovered hostname must pass --scope before DNS resolution or HTTP probing occurs. It uses asyncio/aiohttp, bounded concurrency, rate limiting, structured output, and continues past individual failures.

python
#!/usr/bin/env python3
"""
recon_orchestrator.py

Scope-aware reconnaissance data handler and orchestration framework.

Designed for authorized security testing / bug bounty reconnaissance.

Pipeline:
    Input validation
        -> explicit scope enforcement
        -> passive discovery
        -> DNS resolution
        -> optional HTTP probing
        -> JSON/text output

Important:
    Only run this software against systems you are explicitly authorized
    to assess.

Python: 3.10+
Dependency:
    aiohttp

Install:
    python3 -m pip install aiohttp
"""

from __future__ import annotations

import argparse
import asyncio
import ipaddress
import json
import logging
import re
import socket
import ssl
import sys
import time

from dataclasses import asdict, dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Iterable
from urllib.parse import urlparse

import aiohttp


# ---------------------------------------------------------------------------
# Configuration
# ---------------------------------------------------------------------------

USER_AGENT = "Authorized-Recon-Orchestrator/1.0"

DEFAULT_CONCURRENCY = 10
DEFAULT_TIMEOUT = 8.0
DEFAULT_DELAY = 0.20

MAX_RESPONSE_BYTES = 256 * 1024

FQDN_RE = re.compile(
    r"^(?=.{1,253}$)"
    r"(?:"
    r"[a-zA-Z0-9]"
    r"(?:[a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?"
    r"\."
    r")+"
    r"[a-zA-Z]{2,63}$"
)


# ---------------------------------------------------------------------------
# Data models
# ---------------------------------------------------------------------------

@dataclass
class DNSResult:
    target: str
    resolved: bool
    addresses: list[str]
    error: str | None = None


@dataclass
class HTTPResult:
    hostname: str
    url: str
    reachable: bool
    status: int | None = None
    final_url: str | None = None
    title: str | None = None
    server: str | None = None
    content_type: str | None = None
    redirected_off_scope: bool = False
    error: str | None = None


@dataclass
class TargetSummary:
    input_target: str
    timestamp: str
    discovered: list[str]
    resolved: list[DNSResult]
    web: list[HTTPResult]


# ---------------------------------------------------------------------------
# Logging
# ---------------------------------------------------------------------------

def configure_logging(verbose: bool, quiet: bool) -> None:
    if quiet:
        level = logging.ERROR
    elif verbose:
        level = logging.DEBUG
    else:
        level = logging.INFO

    logging.basicConfig(
        level=level,
        format="[%(levelname)s] %(message)s",
    )


log = logging.getLogger("recon")


# ---------------------------------------------------------------------------
# Target validation
# ---------------------------------------------------------------------------

def normalize_target(value: str) -> str:
    """
    Normalize and strictly validate an IPv4, IPv6, or FQDN target.

    Returns the normalized value or raises ValueError.
    """
    value = value.strip()

    if not value:
        raise ValueError("empty target")

    # Remove IPv6 URI-style brackets.
    if value.startswith("[") and value.endswith("]"):
        value = value[1:-1]

    # IP address?
    try:
        return str(ipaddress.ip_address(value))
    except ValueError:
        pass

    # Hostname.
    hostname = value.lower().rstrip(".")

    if not FQDN_RE.fullmatch(hostname):
        raise ValueError(
            f"invalid target {value!r}: expected IPv4, IPv6, or FQDN"
        )

    # Reject malformed labels defensively.
    for label in hostname.split("."):
        if label.startswith("-") or label.endswith("-"):
            raise ValueError(f"invalid FQDN label in {value!r}")

    return hostname


def is_ip(value: str) -> bool:
    try:
        ipaddress.ip_address(value)
        return True
    except ValueError:
        return False


# ---------------------------------------------------------------------------
# Scope enforcement
# ---------------------------------------------------------------------------

class Scope:
    """
    Explicit authorization scope.

    Rules:
      * Scoped IP addresses authorize exactly that IP.
      * A scoped domain authorizes:
            example.com
            *.example.com
        but NOT:
            fakeexample.com

    Every network-access stage calls allows() before processing a target.
    """

    def __init__(self, entries: Iterable[str]):
        self.domains: set[str] = set()
        self.ip_addresses: set[ipaddress.IPv4Address | ipaddress.IPv6Address] = (
            set()
        )

        for raw in entries:
            normalized = normalize_target(raw)

            if is_ip(normalized):
                self.ip_addresses.add(ipaddress.ip_address(normalized))
            else:
                self.domains.add(normalized)

        if not self.domains and not self.ip_addresses:
            raise ValueError("scope cannot be empty")

    def allows(self, target: str) -> bool:
        try:
            normalized = normalize_target(target)
        except ValueError:
            return False

        if is_ip(normalized):
            return ipaddress.ip_address(normalized) in self.ip_addresses

        for root in self.domains:
            if normalized == root:
                return True

            if normalized.endswith("." + root):
                return True

        return False

    def describe(self) -> dict:
        return {
            "domains": sorted(self.domains),
            "ips": sorted(str(x) for x in self.ip_addresses),
        }


# ---------------------------------------------------------------------------
# Files
# ---------------------------------------------------------------------------

def read_lines(path: Path) -> list[str]:
    if not path.is_file():
        raise ValueError(f"file not found: {path}")

    values: list[str] = []

    for line_number, line in enumerate(
        path.read_text(encoding="utf-8").splitlines(),
        start=1,
    ):
        line = line.strip()

        if not line or line.startswith("#"):
            continue

        try:
            values.append(normalize_target(line))
        except ValueError as exc:
            raise ValueError(
                f"{path}:{line_number}: {exc}"
            ) from exc

    return values


def load_targets(single: str | None, input_file: str | None) -> list[str]:
    if single:
        return [normalize_target(single)]

    if input_file:
        values = read_lines(Path(input_file))

        if not values:
            raise ValueError("target file contains no valid targets")

        return sorted(set(values))

    raise ValueError("specify --target or --input-file")


def load_scope(scope_args: list[str], scope_file: str | None) -> Scope:
    entries: list[str] = []

    for value in scope_args:
        # Allow:
        #   --scope example.com
        #   --scope example.com,api.example.com
        entries.extend(
            part.strip()
            for part in value.split(",")
            if part.strip()
        )

    if scope_file:
        entries.extend(read_lines(Path(scope_file)))

    if not entries:
        raise ValueError(
            "explicit authorization scope is required. "
            "Use --scope or --scope-file."
        )

    return Scope(entries)


# ---------------------------------------------------------------------------
# Rate limiter
# ---------------------------------------------------------------------------

class AsyncRateLimiter:
    """
    Simple global request-start limiter.

    delay=0.25 means request starts will be separated by approximately
    250 ms, regardless of concurrency.
    """

    def __init__(self, delay: float):
        self.delay = max(0.0, delay)
        self._lock = asyncio.Lock()
        self._last_request = 0.0

    async def wait(self) -> None:
        if self.delay <= 0:
            return

        async with self._lock:
            now = time.monotonic()
            elapsed = now - self._last_request
            remaining = self.delay - elapsed

            if remaining > 0:
                await asyncio.sleep(remaining)

            self._last_request = time.monotonic()


# ---------------------------------------------------------------------------
# Passive discovery
# ---------------------------------------------------------------------------

async def passive_crtsh(
    domain: str,
    session: aiohttp.ClientSession,
    scope: Scope,
    limiter: AsyncRateLimiter,
) -> set[str]:
    """
    Passive certificate-transparency discovery through crt.sh.

    No connection is made to discovered hosts here.
    """
    if is_ip(domain):
        return {domain}

    await limiter.wait()

    url = "https://crt.sh/"
    params = {
        "q": f"%.{domain}",
        "output": "json",
    }

    found = {domain}

    try:
        async with session.get(url, params=params) as response:
            if response.status != 200:
                log.warning(
                    "crt.sh returned HTTP %s for %s",
                    response.status,
                    domain,
                )
                return found

            records = await response.json(content_type=None)

    except (
        aiohttp.ClientError,
        asyncio.TimeoutError,
        json.JSONDecodeError,
    ) as exc:
        log.warning("Passive discovery failed for %s: %s", domain, exc)
        return found

    for record in records:
        value = record.get("name_value", "")

        for candidate in value.splitlines():
            candidate = candidate.strip().lower().rstrip(".")

            if candidate.startswith("*."):
                candidate = candidate[2:]

            try:
                candidate = normalize_target(candidate)
            except ValueError:
                continue

            # Critical authorization gate.
            if scope.allows(candidate):
                found.add(candidate)
            else:
                log.debug(
                    "Discarding discovered out-of-scope hostname: %s",
                    candidate,
                )

    return found


async def run_passive_discovery(
    target: str,
    session: aiohttp.ClientSession,
    scope: Scope,
    limiter: AsyncRateLimiter,
) -> set[str]:
    log.info("Passive discovery: %s", target)

    if is_ip(target):
        return {target}

    return await passive_crtsh(
        target,
        session,
        scope,
        limiter,
    )


# ---------------------------------------------------------------------------
# DNS resolution
# ---------------------------------------------------------------------------

async def resolve_one(
    hostname: str,
    scope: Scope,
    semaphore: asyncio.Semaphore,
    limiter: AsyncRateLimiter,
) -> DNSResult:

    if not scope.allows(hostname):
        return DNSResult(
            target=hostname,
            resolved=False,
            addresses=[],
            error="scope check failed",
        )

    if is_ip(hostname):
        return DNSResult(
            target=hostname,
            resolved=True,
            addresses=[hostname],
        )

    async with semaphore:
        await limiter.wait()

        try:
            loop = asyncio.get_running_loop()

            records = await loop.getaddrinfo(
                hostname,
                None,
                family=socket.AF_UNSPEC,
                type=socket.SOCK_STREAM,
            )

            addresses = sorted(
                {
                    record[4][0]
                    for record in records
                }
            )

            return DNSResult(
                target=hostname,
                resolved=bool(addresses),
                addresses=addresses,
            )

        except socket.gaierror as exc:
            return DNSResult(
                target=hostname,
                resolved=False,
                addresses=[],
                error=str(exc),
            )

        except asyncio.CancelledError:
            raise

        except Exception as exc:
            return DNSResult(
                target=hostname,
                resolved=False,
                addresses=[],
                error=f"{type(exc).__name__}: {exc}",
            )


async def run_resolution(
    targets: Iterable[str],
    scope: Scope,
    concurrency: int,
    limiter: AsyncRateLimiter,
) -> list[DNSResult]:

    semaphore = asyncio.Semaphore(concurrency)

    tasks = [
        resolve_one(
            target,
            scope,
            semaphore,
            limiter,
        )
        for target in sorted(set(targets))
        if scope.allows(target)
    ]

    if not tasks:
        return []

    return await asyncio.gather(*tasks)


# ---------------------------------------------------------------------------
# HTTP probing
# ---------------------------------------------------------------------------

def extract_title(body: str) -> str | None:
    match = re.search(
        r"<title[^>]*>(.*?)</title>",
        body,
        flags=re.IGNORECASE | re.DOTALL,
    )

    if not match:
        return None

    title = re.sub(r"\s+", " ", match.group(1)).strip()

    return title[:200] or None


class ScopeRedirectPolicy:
    """
    aiohttp normally follows redirects automatically.

    We disable that behavior and manually inspect each redirect so an
    off-scope redirect is never fetched.
    """

    MAX_REDIRECTS = 5


async def probe_one_url(
    url: str,
    hostname: str,
    session: aiohttp.ClientSession,
    scope: Scope,
    semaphore: asyncio.Semaphore,
    limiter: AsyncRateLimiter,
) -> HTTPResult:

    if not scope.allows(hostname):
        return HTTPResult(
            hostname=hostname,
            url=url,
            reachable=False,
            error="scope check failed",
        )

    current_url = url

    async with semaphore:
        for _ in range(ScopeRedirectPolicy.MAX_REDIRECTS + 1):
            await limiter.wait()

            try:
                async with session.get(
                    current_url,
                    allow_redirects=False,
                ) as response:

                    location = response.headers.get("Location")

                    if (
                        response.status in {301, 302, 303, 307, 308}
                        and location
                    ):
                        destination = str(
                            response.url.join(
                                aiohttp.client_reqrep.URL(location)
                            )
                        )

                        destination_host = (
                            urlparse(destination).hostname or ""
                        )

                        # Critical: never fetch off-scope redirects.
                        if not scope.allows(destination_host):
                            return HTTPResult(
                                hostname=hostname,
                                url=url,
                                reachable=True,
                                status=response.status,
                                final_url=destination,
                                redirected_off_scope=True,
                            )

                        current_url = destination
                        continue

                    data = await response.content.read(
                        MAX_RESPONSE_BYTES
                    )

                    content_type = response.headers.get(
                        "Content-Type",
                        "",
                    )

                    title = None

                    if "html" in content_type.lower():
                        body = data.decode(
                            response.charset or "utf-8",
                            errors="replace",
                        )
                        title = extract_title(body)

                    return HTTPResult(
                        hostname=hostname,
                        url=url,
                        reachable=True,
                        status=response.status,
                        final_url=str(response.url),
                        title=title,
                        server=response.headers.get("Server"),
                        content_type=content_type or None,
                    )

            except asyncio.CancelledError:
                raise

            except (
                aiohttp.ClientError,
                asyncio.TimeoutError,
                UnicodeError,
                ssl.SSLError,
            ) as exc:
                return HTTPResult(
                    hostname=hostname,
                    url=url,
                    reachable=False,
                    error=str(exc),
                )

            except Exception as exc:
                return HTTPResult(
                    hostname=hostname,
                    url=url,
                    reachable=False,
                    error=f"{type(exc).__name__}: {exc}",
                )

        return HTTPResult(
            hostname=hostname,
            url=url,
            reachable=False,
            error="maximum redirect count exceeded",
        )


async def probe_host(
    hostname: str,
    session: aiohttp.ClientSession,
    scope: Scope,
    semaphore: asyncio.Semaphore,
    limiter: AsyncRateLimiter,
) -> list[HTTPResult]:

    if not scope.allows(hostname):
        return []

    if is_ip(hostname):
        # IPv6 must use brackets inside a URL.
        ip_obj = ipaddress.ip_address(hostname)

        if ip_obj.version == 6:
            url_host = f"[{hostname}]"
        else:
            url_host = hostname
    else:
        url_host = hostname

    results = []

    for scheme in ("https", "http"):
        result = await probe_one_url(
            f"{scheme}://{url_host}/",
            hostname,
            session,
            scope,
            semaphore,
            limiter,
        )

        results.append(result)

    return results


async def run_web_probe(
    resolved: list[DNSResult],
    session: aiohttp.ClientSession,
    scope: Scope,
    concurrency: int,
    limiter: AsyncRateLimiter,
) -> list[HTTPResult]:

    semaphore = asyncio.Semaphore(concurrency)

    hostnames = [
        result.target
        for result in resolved
        if result.resolved and scope.allows(result.target)
    ]

    batches = await asyncio.gather(
        *[
            probe_host(
                hostname,
                session,
                scope,
                semaphore,
                limiter,
            )
            for hostname in hostnames
        ]
    )

    return [
        item
        for batch in batches
        for item in batch
    ]


# ---------------------------------------------------------------------------
# Output
# ---------------------------------------------------------------------------

def safe_directory_name(target: str) -> str:
    return (
        target
        .replace(":", "_")
        .replace("/", "_")
        .replace("\\", "_")
    )


def create_output_directory(
    base_directory: Path,
    target: str,
) -> Path:

    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")

    directory = (
        base_directory
        / safe_directory_name(target)
        / timestamp
    )

    directory.mkdir(
        parents=True,
        exist_ok=False,
    )

    return directory


def write_text_lines(
    path: Path,
    values: Iterable[str],
) -> None:

    contents = "\n".join(sorted(set(values)))

    if contents:
        contents += "\n"

    path.write_text(
        contents,
        encoding="utf-8",
    )


def write_json(
    path: Path,
    value,
) -> None:

    path.write_text(
        json.dumps(
            value,
            indent=2,
            sort_keys=True,
        )
        + "\n",
        encoding="utf-8",
    )


def save_target_results(
    directory: Path,
    summary: TargetSummary,
    scope: Scope,
) -> None:

    write_text_lines(
        directory / "discovered.txt",
        summary.discovered,
    )

    write_json(
        directory / "resolved.json",
        [asdict(x) for x in summary.resolved],
    )

    live_hosts = [
        result.target
        for result in summary.resolved
        if result.resolved
    ]

    write_text_lines(
        directory / "resolved.txt",
        live_hosts,
    )

    write_json(
        directory / "web.json",
        [asdict(x) for x in summary.web],
    )

    reachable_urls = [
        result.url
        for result in summary.web
        if result.reachable
    ]

    write_text_lines(
        directory / "web.txt",
        reachable_urls,
    )

    write_json(
        directory / "summary.json",
        {
            **asdict(summary),
            "scope": scope.describe(),
        },
    )


# ---------------------------------------------------------------------------
# Target pipeline
# ---------------------------------------------------------------------------

async def process_target(
    target: str,
    args: argparse.Namespace,
    scope: Scope,
) -> None:

    # Authorization must be explicit before anything happens.
    if not scope.allows(target):
        log.error(
            "Skipping unauthorized/out-of-scope input: %s",
            target,
        )
        return

    output_directory = create_output_directory(
        Path(args.output_dir),
        target,
    )

    log.info("Target: %s", target)
    log.info("Output: %s", output_directory)

    timeout = aiohttp.ClientTimeout(
        total=args.timeout
    )

    connector = aiohttp.TCPConnector(
        limit=args.concurrency,
        ssl=True,
    )

    headers = {
        "User-Agent": USER_AGENT,
        "Accept": "*/*",
    }

    limiter = AsyncRateLimiter(args.delay)

    discovered: set[str] = {target}
    resolved: list[DNSResult] = []
    web_results: list[HTTPResult] = []

    async with aiohttp.ClientSession(
        timeout=timeout,
        connector=connector,
        headers=headers,
    ) as session:

        # ---------------------------------------------------------------
        # Stage 1: Passive discovery
        # ---------------------------------------------------------------

        if not args.no_passive:
            passive = await run_passive_discovery(
                target,
                session,
                scope,
                limiter,
            )

            discovered.update(
                hostname
                for hostname in passive
                if scope.allows(hostname)
            )

        log.info(
            "In-scope discovered targets: %d",
            len(discovered),
        )

        # passive-only stops network interaction with discovered targets.
        if args.passive_only:
            summary = TargetSummary(
                input_target=target,
                timestamp=datetime.now(timezone.utc).isoformat(),
                discovered=sorted(discovered),
                resolved=[],
                web=[],
            )

            save_target_results(
                output_directory,
                summary,
                scope,
            )

            return

        # ---------------------------------------------------------------
        # Stage 2: DNS resolution
        # ---------------------------------------------------------------

        if not args.no_resolve:
            log.info("Resolving targets...")

            resolved = await run_resolution(
                discovered,
                scope,
                args.concurrency,
                limiter,
            )

            successful = sum(
                1
                for result in resolved
                if result.resolved
            )

            log.info(
                "Resolved: %d/%d",
                successful,
                len(resolved),
            )

        # ---------------------------------------------------------------
        # Stage 3: Web probing
        # ---------------------------------------------------------------

        if args.web_probe:
            if args.no_resolve:
                # Direct probing of explicitly in-scope targets.
                resolved = [
                    DNSResult(
                        target=hostname,
                        resolved=True,
                        addresses=[],
                    )
                    for hostname in discovered
                ]

            log.info("Running HTTP/HTTPS probes...")

            web_results = await run_web_probe(
                resolved,
                session,
                scope,
                args.concurrency,
                limiter,
            )

            reachable = sum(
                1
                for result in web_results
                if result.reachable
            )

            log.info(
                "Reachable web endpoints: %d",
                reachable,
            )

    summary = TargetSummary(
        input_target=target,
        timestamp=datetime.now(timezone.utc).isoformat(),
        discovered=sorted(discovered),
        resolved=resolved,
        web=web_results,
    )

    save_target_results(
        output_directory,
        summary,
        scope,
    )

    log.info("Finished: %s", target)


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------

def parse_arguments() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description=(
            "Scope-aware reconnaissance orchestration framework "
            "for authorized security testing."
        )
    )

    target_group = parser.add_mutually_exclusive_group(
        required=True
    )

    target_group.add_argument(
        "-t",
        "--target",
        help="Single IPv4, IPv6, or FQDN target.",
    )

    target_group.add_argument(
        "-i",
        "--input-file",
        help="Text file containing one target per line.",
    )

    parser.add_argument(
        "--scope",
        action="append",
        default=[],
        help=(
            "Explicit authorized scope. May be repeated or comma-separated. "
            "A scoped domain authorizes its subdomains."
        ),
    )

    parser.add_argument(
        "--scope-file",
        help="File containing explicitly authorized domains/IPs.",
    )

    stages = parser.add_argument_group("pipeline stages")

    stages.add_argument(
        "--passive-only",
        action="store_true",
        help="Perform passive discovery only; no DNS or HTTP probing.",
    )

    stages.add_argument(
        "--no-passive",
        action="store_true",
        help="Disable passive discovery.",
    )

    stages.add_argument(
        "--no-resolve",
        action="store_true",
        help="Disable DNS resolution.",
    )

    stages.add_argument(
        "--web-probe",
        action="store_true",
        help="Enable HTTP/HTTPS probing.",
    )

    network = parser.add_argument_group("network controls")

    network.add_argument(
        "--concurrency",
        type=int,
        default=DEFAULT_CONCURRENCY,
        help=f"Maximum concurrent network operations "
             f"(default: {DEFAULT_CONCURRENCY}).",
    )

    network.add_argument(
        "--delay",
        type=float,
        default=DEFAULT_DELAY,
        help=(
            "Minimum delay in seconds between network request starts "
            f"(default: {DEFAULT_DELAY})."
        ),
    )

    network.add_argument(
        "--timeout",
        type=float,
        default=DEFAULT_TIMEOUT,
        help=f"Request timeout in seconds (default: {DEFAULT_TIMEOUT}).",
    )

    output = parser.add_argument_group("output")

    output.add_argument(
        "-o",
        "--output-dir",
        default="recon-output",
        help="Base output directory.",
    )

    verbosity = parser.add_mutually_exclusive_group()

    verbosity.add_argument(
        "-v",
        "--verbose",
        action="store_true",
        help="Enable verbose/debug logging.",
    )

    verbosity.add_argument(
        "-q",
        "--quiet",
        action="store_true",
        help="Only display errors.",
    )

    args = parser.parse_args()

    if args.concurrency < 1:
        parser.error("--concurrency must be at least 1")

    # Prevent accidentally unreasonable concurrency.
    if args.concurrency > 100:
        parser.error("--concurrency cannot exceed 100")

    if args.delay < 0:
        parser.error("--delay cannot be negative")

    if args.timeout <= 0:
        parser.error("--timeout must be greater than zero")

    if args.passive_only and args.no_passive:
        parser.error(
            "--passive-only and --no-passive cannot be used together"
        )

    return args


# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------

async def async_main(args: argparse.Namespace) -> int:
    try:
        targets = load_targets(
            args.target,
            args.input_file,
        )

        scope = load_scope(
            args.scope,
            args.scope_file,
        )

    except ValueError as exc:
        log.error("%s", exc)
        return 2

    log.debug(
        "Authorization scope: %s",
        json.dumps(scope.describe()),
    )

    authorized_targets = []

    for target in targets:
        if scope.allows(target):
            authorized_targets.append(target)
        else:
            log.error(
                "Input target rejected by scope policy: %s",
                target,
            )

    if not authorized_targets:
        log.error("No authorized targets remain.")
        return 2

    for target in authorized_targets:
        try:
            await process_target(
                target,
                args,
                scope,
            )

        except asyncio.CancelledError:
            raise

        except Exception:
            # One broken target should not terminate a multi-target run.
            log.exception(
                "Unexpected pipeline failure for %s",
                target,
            )

    return 0


def main() -> int:
    args = parse_arguments()

    configure_logging(
        args.verbose,
        args.quiet,
    )

    try:
        return asyncio.run(
            async_main(args)
        )

    except KeyboardInterrupt:
        log.error("Interrupted by user.")
        return 130


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

Installation

The script needs Python 3.10+ and aiohttp:

bash
python3 -m venv .venv
source .venv/bin/activate

python3 -m pip install --upgrade pip
python3 -m pip install aiohttp

Save it as:

bash
recon_orchestrator.py

Then:

bash
chmod +x recon_orchestrator.py

Scope behavior

--scope is intentionally mandatory. For example:

bash
python3 recon_orchestrator.py \
    --target example.com \
    --scope example.com

This permits:

text
example.com
api.example.com
dev.api.example.com

but rejects:

text
example.net
notexample.com
example.com.attacker.net

For an IP:

bash
python3 recon_orchestrator.py \
    --target 192.0.2.20 \
    --scope 192.0.2.20 \
    --web-probe

An IP scope entry authorizes that exact IP only.

For several explicitly authorized roots:

bash
python3 recon_orchestrator.py \
    --input-file targets.txt \
    --scope example.com \
    --scope example.net \
    --scope 192.0.2.20 \
    --web-probe

Or put the authorization boundary in scope.txt:

text
example.com
example.net
192.0.2.20
2001:db8::10

and run:

bash
python3 recon_orchestrator.py \
    --input-file targets.txt \
    --scope-file scope.txt \
    --web-probe

Pipeline controls

Passive discovery only:

bash
python3 recon_orchestrator.py \
    -t example.com \
    --scope example.com \
    --passive-only

Skip passive enumeration and resolve only the supplied target:

bash
python3 recon_orchestrator.py \
    -t api.example.com \
    --scope example.com \
    --no-passive

Run passive discovery, resolution, and HTTP probing:

bash
python3 recon_orchestrator.py \
    -t example.com \
    --scope example.com \
    --web-probe

Use conservative network settings:

bash
python3 recon_orchestrator.py \
    -t example.com \
    --scope example.com \
    --web-probe \
    --concurrency 5 \
    --delay 0.5 \
    --timeout 10

The --delay is global rather than per-worker, so --delay 0.5 limits request starts to roughly two per second even when multiple workers are available.

Output structure

A run produces directories like:

text
recon-output/
└── example.com/
    └── 20261001_001500/
        ├── discovered.txt
        ├── resolved.txt
        ├── resolved.json
        ├── web.txt
        ├── web.json
        └── summary.json

summary.json also records the scope used for that run, which is useful for auditability.

A particularly important implementation detail is that redirect handling is manual: if an authorized host redirects to an unrelated hostname, the tool records the redirect but does not request the off-scope destination. The same scope check occurs before passive discoveries enter later active stages.