domain_intelligence.py

"""Domain Intelligence Tool command-line entry point."""

from __future__ import annotations

import argparse
import csv
import json
import logging
import sys
import tempfile
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

try:
    from dotenv import load_dotenv
except ImportError:  # pragma: no cover
    def load_dotenv(*_args: Any, **_kwargs: Any) -> None:
        return None

from rich.console import Console
from rich.logging import RichHandler
from rich.panel import Panel
from rich.progress import Progress, SpinnerColumn, TextColumn
from rich.table import Table
from rich.text import Text

from modules.dns_lookup import DNSCollector, RECORD_TYPES
from modules.exporters import export_selected
from modules.ip_intelligence import IPIntelligenceCollector
from modules.mail_security import analyze_dmarc, analyze_mx, analyze_spf
from modules.risk_analyzer import build_findings
from modules.tls_analyzer import TLSAnalyzer
from modules.validators import normalize_domain
from modules.whois_lookup import WhoisCollector


VERSION = "1.0.0"
PROJECT_DIR = Path(__file__).resolve().parent
GITHUB_URL = "https://github.com/X3roxDev"
DISCLAIMER = (
    "This tool is intended for authorized OSINT, research, educational, and defensive security purposes only. "
    "Users are responsible for complying with applicable laws, service terms, and API usage policies."
)
BANNER = r"""
▓█████▄  ▒█████   ███▄ ▄███▓ ▄▄▄       ██▓ ███▄    █     ██▓ ███▄    █ ▄▄▄█████▓▓█████  ██▓
▒██▀ ██▌▒██▒  ██▒▓██▒▀█▀ ██▒▒████▄    ▓██▒ ██ ▀█   █    ▓██▒ ██ ▀█   █ ▓  ██▒ ▓▒▓█   ▀ ▓██▒
░██   █▌▒██░  ██▒▓██    ▓██░▒██  ▀█▄  ▒██▒▓██  ▀█ ██▒   ▒██▒▓██  ▀█ ██▒▒ ▓██░ ▒░▒███   ▒██░
░▓█▄   ▌▒██   ██░▒██    ▒██ ░██▄▄▄▄██ ░██░▓██▒  ▐▌██▒   ░██░▓██▒  ▐▌██▒░ ▓██▓ ░ ▒▓█  ▄ ▒██░
░▒████▓ ░ ████▓▒░▒██▒   ░██▒ ▓█   ▓██▒░██░▒██░   ▓██░   ░██░▒██░   ▓██░  ▒██▒ ░ ░▒████▒░██████▒
 ▒▒▓  ▒ ░ ▒░▒░▒░ ░ ▒░   ░  ░ ▒▒   ▓▒█░░▓  ░ ▒░   ▒ ▒    ░▓  ░ ▒░   ▒ ▒   ▒ ░░   ░░ ▒░ ░░ ▒░▓  ░
 ░ ▒  ▒   ░ ▒ ▒░ ░  ░      ░  ▒   ▒▒ ░ ▒ ░░ ░░   ░ ▒░    ▒ ░░ ░░   ░ ▒░    ░     ░ ░  ░░ ░ ▒  ░
 ░ ░  ░ ░ ░ ░ ▒  ░      ░     ░   ▒    ▒ ░   ░   ░ ░     ▒ ░   ░   ░ ░   ░         ░     ░ ░
   ░        ░ ░         ░         ░  ░ ░           ░     ░           ░             ░  ░    ░  ░
 ░
"""


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description="Passive domain intelligence and defensive configuration analysis.")
    parser.add_argument("domain", nargs="?", help="Domain or URL to investigate")
    parser.add_argument("--full", action="store_true", help="Run all intelligence modules")
    parser.add_argument("--whois", action="store_true", help="Run WHOIS analysis")
    parser.add_argument("--dns", action="store_true", help="Run DNS analysis")
    parser.add_argument("--mx", action="store_true", help="Run MX analysis")
    parser.add_argument("--spf", action="store_true", help="Run SPF analysis")
    parser.add_argument("--dmarc", action="store_true", help="Run DMARC analysis")
    parser.add_argument("--asn", action="store_true", help="Run ASN analysis")
    parser.add_argument("--geoip", action="store_true", help="Run GeoIP analysis")
    parser.add_argument("--tls", action="store_true", help="Run TLS analysis")
    parser.add_argument("--srv", action="store_true", help="Include SRV records during DNS analysis")
    parser.add_argument("--tls-port", type=int, default=443, help="Custom TLS port")
    parser.add_argument("--dns-server", help="Custom DNS resolver")
    parser.add_argument("--timeout", type=float, default=5.0, help="Network timeout in seconds")
    parser.add_argument("--export-json", action="store_true", help="Export results to JSON")
    parser.add_argument("--export-txt", action="store_true", help="Export results to TXT")
    parser.add_argument("--export-csv", action="store_true", help="Export suitable records to CSV")
    parser.add_argument("--output", default="results", help="Export directory")
    parser.add_argument("--no-color", action="store_true", help="Disable terminal colors")
    parser.add_argument("--verbose", action="store_true", help="Display detailed errors")
    parser.add_argument("--version", action="version", version=f"Domain Intelligence Tool {VERSION}")
    return parser


def ensure_utf8_stdio() -> None:
    """Prefer UTF-8 output on Windows terminals and redirected shells."""

    for stream in (sys.stdout, sys.stderr):
        reconfigure = getattr(stream, "reconfigure", None)
        if callable(reconfigure):
            try:
                reconfigure(encoding="utf-8", errors="replace")
            except Exception:
                pass


def setup_logging(verbose: bool) -> None:
    handlers: list[logging.Handler] = []
    for log_dir in (PROJECT_DIR / "logs", Path(tempfile.gettempdir()) / "domain-intelligence-tool-logs"):
        try:
            log_dir.mkdir(parents=True, exist_ok=True)
            handlers.append(logging.FileHandler(log_dir / "domain_intelligence.log", encoding="utf-8"))
            break
        except OSError:
            continue
    if verbose:
        handlers.append(RichHandler(rich_tracebacks=True))
    if not handlers:
        handlers.append(logging.NullHandler())
    logging.basicConfig(level=logging.INFO if not verbose else logging.DEBUG, format="%(asctime)s %(levelname)s %(message)s", handlers=handlers)


def resolve_output_base(output: str) -> str:
    """Resolve relative export paths from the project directory, not the launch directory."""

    path = Path(output).expanduser()
    if not path.is_absolute():
        path = PROJECT_DIR / path
    return str(path)


def selected_modules(args: argparse.Namespace) -> set[str]:
    modules = {"whois", "dns", "mx", "spf", "dmarc", "asn", "geoip", "tls"}
    flags = {name for name in modules if getattr(args, name)}
    return modules if args.full or not flags else flags


def base_results(domain: str, started: str) -> dict[str, Any]:
    return {
        "target": domain,
        "scan_started": started,
        "scan_finished": "",
        "whois": {},
        "dns": {},
        "mx": [],
        "spf": {},
        "dmarc": {},
        "ips": [],
        "asn": [],
        "geoip": [],
        "tls": {},
        "findings": [],
        "errors": [],
    }


def render_banner(console: Console, domain: str, started: str) -> None:
    console.print(Text(BANNER, style="bold cyan", overflow="fold"))
    console.print(Panel.fit(f"Domain Intelligence Tool\nTarget: {domain}\nStarted: {started}\n\n{DISCLAIMER}", border_style="cyan"))


def render_key_values(console: Console, title: str, data: dict[str, Any], keys: list[str]) -> None:
    table = Table(title=title, show_header=True, header_style="bold cyan")
    table.add_column("Field")
    table.add_column("Value", overflow="fold")
    for key in keys:
        value = data.get(key)
        if value in (None, "", []):
            value = "-"
        table.add_row(key.replace("_", " ").title(), str(value))
    console.print(table)


def render_dns(console: Console, dns_data: dict[str, Any]) -> None:
    for record_type, records in sorted((dns_data.get("records") or {}).items()):
        table = Table(title=f"DNS {record_type}", header_style="bold cyan")
        table.add_column("Value", overflow="fold")
        table.add_column("TTL")
        table.add_column("Details", overflow="fold")
        for record in records:
            value = record.get("value") or record.get("ip") or record.get("exchange")
            details = []
            if "priority" in record:
                details.append(f"priority={record['priority']}")
            if "reverse_dns" in record:
                details.append(f"reverse={record.get('reverse_dns') or '-'}")
                details.append(f"public={record.get('is_public')}")
            table.add_row(str(value), str(record.get("ttl", "-")), ", ".join(details) or "-")
        if records:
            console.print(table)


def render_mx(console: Console, mx_data: list[dict[str, Any]]) -> None:
    table = Table(title="MX Intelligence", header_style="bold cyan")
    table.add_column("Priority")
    table.add_column("Hostname", overflow="fold")
    table.add_column("Status")
    table.add_column("Provider Estimate")
    table.add_column("IPs", overflow="fold")
    for mx in mx_data:
        if mx.get("null_mx"):
            status = "Null MX"
        else:
            status = "Resolves" if mx.get("resolves") else "No resolution"
        table.add_row(str(mx.get("priority")), mx.get("hostname", "-"), status, mx.get("provider_estimate") or "-", ", ".join(mx.get("ipv4", []) + mx.get("ipv6", [])) or "-")
    console.print(table)


def render_findings(console: Console, findings: list[dict[str, str]]) -> None:
    table = Table(title="Risk Summary", header_style="bold magenta")
    table.add_column("Severity")
    table.add_column("Finding", overflow="fold")
    table.add_column("Detail", overflow="fold")
    for item in findings or [{"severity": "INFO", "title": "No findings generated", "detail": "No configured checks produced warnings."}]:
        table.add_row(item["severity"], item["title"], item["detail"])
    console.print(table)


def aggregate_errors(results: dict[str, Any]) -> list[str]:
    """Collect recoverable module errors for terminal display and exports."""

    errors = list(results.get("errors", []))
    for message in (results.get("whois") or {}).get("errors", []):
        errors.append(f"WHOIS: {message}")
    for item in (results.get("dns") or {}).get("errors", []):
        if isinstance(item, dict):
            if item.get("error") != "NoAnswer":
                errors.append(f"DNS {item.get('type')}: {item.get('error')}")
        else:
            errors.append(f"DNS: {item}")
    dmarc_error = (results.get("dmarc") or {}).get("error")
    if dmarc_error and not (results.get("dmarc") or {}).get("records"):
        errors.append(f"DMARC: {dmarc_error}")
    tls_error = (results.get("tls") or {}).get("error")
    if tls_error:
        errors.append(f"TLS: {tls_error}")
    return sorted({str(error) for error in errors if error})


def render_errors(console: Console, errors: list[str], verbose: bool) -> None:
    """Render concise recoverable errors without hiding important failures."""

    if not errors:
        return
    shown = errors if verbose else errors[:8]
    suffix = "" if verbose or len(errors) <= 8 else f"\n... {len(errors) - 8} more; rerun with --verbose for all details."
    console.print(Panel("\n".join(shown) + suffix, title="Lookup Notes", border_style="yellow"))


def run_with_spinner(console: Console, label: str, func: Any) -> Any:
    with Progress(SpinnerColumn(), TextColumn("[progress.description]{task.description}"), transient=True, console=console) as progress:
        progress.add_task(label, total=None)
        return func()


def collect_scan(args: argparse.Namespace, console: Console, domain: str, modules: set[str]) -> dict[str, Any]:
    started = datetime.now(timezone.utc).isoformat()
    results = base_results(domain, started)
    logging.info("Scan start for %s", domain)

    dns_collector = DNSCollector(timeout=args.timeout, dns_server=args.dns_server)
    need_dns = bool(modules & {"dns", "mx", "spf", "asn", "geoip"})

    try:
        if "whois" in modules:
            results["whois"] = run_with_spinner(console, "[1/8] WHOIS Intelligence", lambda: WhoisCollector().collect(domain))
            logging.info("Completed WHOIS")

        if need_dns:
            record_types = RECORD_TYPES + (("SRV",) if args.srv else ())
            results["dns"] = run_with_spinner(console, "[2/8] DNS Records", lambda: dns_collector.collect(domain, record_types))
            results["ips"] = results["dns"].get("ips", [])
            logging.info("Completed DNS")

        if "mx" in modules:
            results["mx"] = run_with_spinner(console, "[3/8] MX Intelligence", lambda: analyze_mx(results["dns"]))
            logging.info("Completed MX")

        if "spf" in modules:
            results["spf"] = run_with_spinner(console, "[4/8] SPF Analysis", lambda: analyze_spf(results["dns"]))
            logging.info("Completed SPF")

        if "dmarc" in modules:
            results["dmarc"] = run_with_spinner(console, "[5/8] DMARC Analysis", lambda: analyze_dmarc(domain, args.timeout, args.dns_server))
            logging.info("Completed DMARC")

        if modules & {"asn", "geoip"}:
            ips = [item["ip"] for item in results.get("ips", []) if item.get("is_public")]
            ip_data = run_with_spinner(
                console,
                "[6/8] ASN and [7/8] GeoIP Intelligence",
                lambda: IPIntelligenceCollector(timeout=args.timeout).collect(ips, "asn" in modules, "geoip" in modules),
            )
            results["asn"] = ip_data.get("asn", [])
            results["geoip"] = ip_data.get("geoip", [])
            logging.info("Completed IP intelligence")

        if "tls" in modules:
            results["tls"] = run_with_spinner(console, "[8/8] TLS Certificate", lambda: TLSAnalyzer(timeout=args.timeout).analyze(domain, args.tls_port))
            logging.info("Completed TLS")
    except KeyboardInterrupt:
        results["errors"].append("Scan cancelled by user.")
        logging.warning("Scan cancelled by user")
        raise
    finally:
        results["scan_finished"] = datetime.now(timezone.utc).isoformat()

    results["errors"] = aggregate_errors(results)
    results["findings"] = build_findings(results)
    logging.info("Scan complete for %s", domain)
    return results


def render_results(console: Console, results: dict[str, Any], verbose: bool) -> None:
    if results.get("whois"):
        render_key_values(console, "WHOIS Intelligence", results["whois"], ["domain_name", "registrar", "registrar_iana_id", "whois_server", "creation_date", "updated_date", "expiration_date", "domain_age", "remaining_until_expiration", "dnssec", "abuse_contact_email"])
    if results.get("dns"):
        render_dns(console, results["dns"])
    if results.get("mx"):
        render_mx(console, results["mx"])
    if results.get("spf"):
        render_key_values(console, "SPF Analysis", results["spf"], ["records", "version", "included_domains", "ipv4_mechanisms", "ipv6_mechanisms", "redirect", "all_policy", "lookup_mechanism_count", "policy_strength", "warnings"])
    if results.get("dmarc"):
        render_key_values(console, "DMARC Analysis", results["dmarc"], ["records", "policy", "subdomain_policy", "percentage", "aggregate_reports", "forensic_reports", "spf_alignment", "dkim_alignment", "failure_options", "warnings"])
    if results.get("asn"):
        render_key_values(console, "ASN Intelligence", {"items": results["asn"]}, ["items"])
    if results.get("geoip"):
        console.print(Panel("IP geolocation is approximate and does not represent an exact physical address.", border_style="yellow"))
        render_key_values(console, "GeoIP Intelligence", {"items": results["geoip"]}, ["items"])
    if results.get("tls"):
        render_key_values(console, "TLS Certificate", results["tls"], ["connected", "tls_version", "cipher_suite", "common_name", "issuer", "subject_alt_names", "valid_from", "valid_until", "remaining_validity_days", "sha256_fingerprint", "hostname_matches", "currently_valid", "expired", "self_signed", "warnings", "error"])
    render_errors(console, results.get("errors", []), verbose)
    render_findings(console, results.get("findings", []))


def summarize_dashboard(last_result: dict[str, Any] | None) -> dict[str, int]:
    """Build compact counters for the dashboard header."""

    if not last_result:
        return {"total": 0, "findings": 0, "high": 0, "cert_problems": 0, "secure": 0}

    findings = last_result.get("findings", [])
    high_count = sum(1 for item in findings if item.get("severity") in {"HIGH", "CRITICAL"})
    cert_problems = len((last_result.get("tls") or {}).get("warnings", []))
    secure = int(bool(last_result.get("tls", {}).get("connected")) and cert_problems == 0 and high_count == 0)
    return {
        "total": 1,
        "findings": len(findings),
        "high": high_count,
        "cert_problems": cert_problems,
        "secure": secure,
    }


def render_dashboard(console: Console, last_result: dict[str, Any] | None = None) -> None:
    """Render the compact menu-style interface."""

    stats = summarize_dashboard(last_result)
    console.clear()
    console.print(Text(BANNER.rstrip(), style="bold white", overflow="crop", no_wrap=True))
    console.print()
    console.print("[bold white]Domain Intelligence Tool[/bold white]")
    console.print(Text(f"GitHub: {GITHUB_URL}", style="bold white"))
    console.print("[dim]" + "=" * 113 + "[/dim]")
    console.print(
        f"[bold green]Total:[/bold green] {stats['total']}   "
        f"[bold green]Findings:[/bold green] {stats['findings']}   "
        f"[bold yellow]High:[/bold yellow] {stats['high']}   "
        f"[bold red]Cert Problems:[/bold red] {stats['cert_problems']}   "
        f"[bold cyan]Secure:[/bold cyan] {stats['secure']}"
    )
    console.print()
    console.print("[bold white]Menu[/bold white]")
    console.print("  [bold white][1][/bold white] Scan domain")
    console.print("  [bold white][2][/bold white] Batch scan TXT/CSV")
    console.print("  [bold white][3][/bold white] History and search")
    console.print("  [bold white][4][/bold white] Clear loaded result")
    console.print("  [bold white][0][/bold white] Exit")
    console.print()


def pause(console: Console) -> None:
    console.print()
    console.print("[bold cyan]Press Enter to continue...[/bold cyan]", end="")
    input()


def run_scan_for_input(args: argparse.Namespace, console: Console, domain_input: str) -> dict[str, Any] | None:
    """Normalize, scan, render, and optionally export one target."""

    try:
        target = normalize_domain(domain_input)
    except ValueError as exc:
        console.print(f"[red]Error:[/red] {exc}")
        return None

    started_display = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    render_banner(console, target.ascii_domain, started_display)
    modules = selected_modules(args)
    try:
        results = collect_scan(args, console, target.ascii_domain, modules)
    except KeyboardInterrupt:
        console.print("\n[yellow]Scan cancelled.[/yellow]")
        return None

    render_results(console, results, args.verbose)
    if args.export_json or args.export_txt or args.export_csv:
        paths = export_selected(results, resolve_output_base(args.output), args.export_json, args.export_txt, args.export_csv)
        for path in paths:
            logging.info("Exported %s", path)
        console.print(Panel("\n".join(str(path) for path in paths), title="Export Results", border_style="green"))
    return results


def load_batch_targets(path: Path) -> list[str]:
    """Load domains or URLs from a TXT or CSV file."""

    if path.suffix.lower() == ".csv":
        with path.open("r", newline="", encoding="utf-8-sig") as handle:
            reader = csv.DictReader(handle)
            if reader.fieldnames:
                preferred = next((name for name in reader.fieldnames if name.lower() in {"domain", "url", "target"}), reader.fieldnames[0])
                return [str(row.get(preferred, "")).strip() for row in reader if str(row.get(preferred, "")).strip()]
    return [line.strip() for line in path.read_text(encoding="utf-8-sig").splitlines() if line.strip() and not line.strip().startswith("#")]


def batch_scan(args: argparse.Namespace, console: Console) -> dict[str, Any] | None:
    """Run a simple batch scan from TXT or CSV targets."""

    console.print("[bold cyan]TXT/CSV path>[/bold cyan] ", end="")
    file_input = input().strip().strip('"')
    if not file_input:
        console.print("[yellow]No file selected.[/yellow]")
        return None
    path = Path(file_input).expanduser()
    if not path.is_absolute():
        path = PROJECT_DIR / path
    if not path.exists():
        console.print(f"[red]File not found:[/red] {path}")
        return None

    targets = load_batch_targets(path)
    if not targets:
        console.print("[yellow]No targets found in that file.[/yellow]")
        return None

    batch_args = argparse.Namespace(**vars(args))
    batch_args.export_json = True
    batch_args.export_txt = args.export_txt
    batch_args.export_csv = args.export_csv
    last_result = None
    for index, target in enumerate(targets, start=1):
        console.print(f"\n[bold cyan][{index}/{len(targets)}][/bold cyan] {target}")
        last_result = run_scan_for_input(batch_args, console, target) or last_result
    return last_result


def history_and_search(console: Console) -> None:
    """Show exported report history and allow a light filename/target search."""

    results_dir = PROJECT_DIR / "results"
    console.print("[bold cyan]Search history (blank for latest)>[/bold cyan] ", end="")
    query = input().strip().lower()
    reports = sorted(results_dir.glob("*/report.json"), key=lambda item: item.stat().st_mtime, reverse=True) if results_dir.exists() else []
    if query:
        reports = [report for report in reports if query in str(report.parent.name).lower()]
    if not reports:
        console.print("[yellow]No matching history found.[/yellow]")
        return

    table = Table(title="History", header_style="bold cyan")
    table.add_column("#")
    table.add_column("Target")
    table.add_column("Finished")
    table.add_column("Findings")
    table.add_column("Report", overflow="fold")
    for index, report in enumerate(reports[:20], start=1):
        try:
            data = json.loads(report.read_text(encoding="utf-8"))
        except (OSError, json.JSONDecodeError):
            data = {}
        table.add_row(
            str(index),
            str(data.get("target") or report.parent.name),
            str(data.get("scan_finished") or "-"),
            str(len(data.get("findings", []))),
            str(report),
        )
    console.print(table)


def interactive_menu(args: argparse.Namespace, console: Console) -> int:
    """Run the screenshot-style interactive dashboard."""

    last_result: dict[str, Any] | None = None
    while True:
        render_dashboard(console, last_result)
        console.print("[bold cyan]Select option>[/bold cyan] ", end="")
        choice = input().strip()
        if choice == "1":
            console.print("[bold cyan]Domain or URL>[/bold cyan] ", end="")
            domain_input = input().strip()
            if domain_input:
                console.clear()
                last_result = run_scan_for_input(args, console, domain_input) or last_result
            pause(console)
        elif choice == "2":
            console.clear()
            last_result = batch_scan(args, console) or last_result
            pause(console)
        elif choice == "3":
            console.clear()
            history_and_search(console)
            pause(console)
        elif choice == "4":
            last_result = None
            console.print("[green]Loaded result cleared.[/green]")
            pause(console)
        elif choice == "0":
            return 0
        else:
            console.print("[red]Invalid option.[/red]")
            pause(console)


def main(argv: list[str] | None = None) -> int:
    ensure_utf8_stdio()
    load_dotenv(PROJECT_DIR / ".env")
    parser = build_parser()
    raw_args = sys.argv[1:] if argv is None else argv
    args = parser.parse_args(raw_args)
    setup_logging(args.verbose)
    console = Console(color_system=None if args.no_color else "auto")

    if not args.domain and not raw_args:
        return interactive_menu(args, console)

    if args.domain:
        domain_input = args.domain
    else:
        console.print("[bold cyan]Domain or URL>[/bold cyan] ", end="")
        domain_input = input().strip()
    return 0 if run_scan_for_input(args, console, domain_input) else 2


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