main.py

"""Command-line interface for MUDOS."""

from __future__ import annotations

from datetime import datetime

import click
from click.shell_completion import CompletionItem

from config import (
    APP_EDITION,
    APP_LICENSE,
    APP_LICENSE_NAME,
    APP_NAME,
    APP_VERSION,
    DEVELOPER_EMAIL,
    DEVELOPER_GITHUB,
    DEVELOPER_LINKEDIN,
    DEVELOPER_NAME,
    ETHICS_WARNING,
)
from controller import AttackController
from utils.cli_helpers import (
    apply_tls_vector_defaults,
    build_campaign_phase_overrides,
    build_run_attack_options,
    format_unknown_attack_error,
    require_confirm_or_abort,
    resolve_spoof_for_run,
    resolve_ssl_for_attack,
    validate_campaign_numeric_overrides,
    validate_mutually_exclusive_flags,
)
from utils.cli_options import campaign_attack_overrides, run_attack_options
from utils.cli_paths import ProjectProfilePath
from utils.console import (
    banner as render_banner,
)
from utils.console import (
    key_values,
    note,
    section,
    table,
    warning_panel,
)
from utils.console import (
    report_paths as render_report_paths,
)
from utils.logger import setup_logger
from utils.platform import is_root_user
from utils.probe import probe_target
from utils.scope import ScopeError

logger = setup_logger()


def _complete_attack_names(ctx, param, incomplete: str):
    """Shell completion callback for attack names."""
    from attacks import ATTACK_REGISTRY

    prefix = incomplete.lower()
    return [
        CompletionItem(name, help=cls.description)
        for name, cls in sorted(ATTACK_REGISTRY.items())
        if name.lower().startswith(prefix)
    ]




def _complete_profile_names(ctx, param, incomplete: str):
    """Shell completion callback for campaign profile names."""
    from profiles.loader import list_profile_names

    prefix = incomplete.lower()
    return [
        CompletionItem(name, help="campaign profile")
        for name in list_profile_names()
        if name.lower().startswith(prefix)
    ]


@click.group()
@click.version_option(version=APP_VERSION, prog_name=APP_NAME)
def cli():
    """MUDOS - professional multi-vector DDoS testing tool."""


@cli.command("completion")
@click.argument("shell", type=click.Choice(["bash", "zsh", "fish"]))
@click.option(
    "--prog-name",
    default="mudos",
    show_default=True,
    help="Executable name to generate completion for",
)
def completion_cmd(shell: str, prog_name: str):
    """Generate shell completion script."""
    from click.shell_completion import get_completion_class

    completion_cls = get_completion_class(shell)
    if completion_cls is None:
        raise click.ClickException(f"Unsupported shell: {shell}")

    complete_var = f"_{prog_name.replace('-', '_').replace('.', '_').upper()}_COMPLETE"
    click.echo(completion_cls(cli, {}, prog_name, complete_var).source())


@cli.command("list")
def list_attacks():
    """List all available attack vectors."""
    controller = AttackController()
    attacks = controller.list_attacks()

    rows = [
        (
            attack["name"],
            "yes" if attack["requires_root"] else "no",
            attack["category"],
            attack["description"],
        )
        for attack in attacks
    ]
    table(
        ("Attack", "Root", "Category", "Description"),
        rows,
        title="Attack Catalog",
        footer=f"Total: {len(attacks)} attack vectors",
        max_col_widths=(24, 6, 14, 70),
    )


@cli.command("profiles")
def list_profiles_cmd():
    """List available campaign profiles."""
    from profiles.loader import list_profiles

    profiles = list_profiles()
    if not profiles:
        note("No profiles found in profiles/ directory.", fg="yellow")
        return

    table(
        ("Profile", "Phases", "Description"),
        [(p["name"], p["phases"], p["description"]) for p in profiles],
        title="Campaign Profiles",
        max_col_widths=(24, 8, 78),
    )


@cli.command("about")
def about_cmd():
    """Show edition, developer, and project contact information."""
    key_values(
        "About MUDOS",
        [
            ("Edition", APP_EDITION, "ok"),
            ("Version", APP_VERSION, None),
            ("Developer", DEVELOPER_NAME, None),
            ("Email", DEVELOPER_EMAIL, None),
            ("LinkedIn", DEVELOPER_LINKEDIN, None),
            ("GitHub", DEVELOPER_GITHUB, None),
            ("License", f"{APP_LICENSE} ({APP_LICENSE_NAME})", "warn"),
        ],
    )


@cli.command("coverage")
@click.option(
    "--show",
    "show_filter",
    default="all",
    type=click.Choice(["all", "covered", "missing", "out-of-scope"]),
    show_default=True,
    help="Filter vector families by coverage status",
)
def coverage_cmd(show_filter: str):
    """Show DDoS vector-family coverage against the implemented catalog."""
    from attacks import ATTACK_REGISTRY
    from attacks.taxonomy import coverage_rows, coverage_summary

    summary = coverage_summary(ATTACK_REGISTRY.keys())
    key_values(
        "DDoS Coverage Summary",
        [
            ("Registered Vectors", len(ATTACK_REGISTRY), "ok"),
            ("In-Scope Families", summary["in_scope_families"], "ok"),
            ("Covered Families", summary["covered_families"], "ok"),
            (
                "Missing Families",
                summary["missing_families"],
                "fail" if summary["missing_families"] else "ok",
            ),
            ("Out of Scope", summary["out_of_scope_families"], "warn"),
            ("Coverage", f"{summary['coverage_percent']}%", "ok"),
        ],
    )

    if summary["problems"]:
        table(
            ("Problem",),
            [(problem,) for problem in summary["problems"]],
            title="Catalog Integrity Problems",
            max_col_widths=(110,),
        )

    rows = coverage_rows(ATTACK_REGISTRY.keys())
    normalized_filter = show_filter.replace("-", "_")
    if normalized_filter != "all":
        rows = [row for row in rows if row["status"] == normalized_filter]

    table(
        ("Domain", "MITRE", "Status", "Family", "Vectors"),
        [
            (
                row["domain"],
                row["mitre"],
                str(row["status"]).replace("_", "-"),
                row["family"],
                ", ".join(row["covered_by"]) if row["covered_by"] else row["notes"],
            )
            for row in rows
        ],
        title="Vector Family Coverage",
        max_col_widths=(22, 24, 12, 42, 80),
    )


@cli.command("check")
@click.option(
    "--strict",
    is_flag=True,
    default=False,
    help="Exit with code 1 when libpcap, scapy, h2, or aioquic is missing.",
)
@click.option(
    "--quick",
    is_flag=True,
    default=False,
    help="Check core dependencies only (skip vector registry and coverage scan).",
)
def check_cmd(strict: bool, quick: bool):
    """Verify dependencies and environment readiness."""
    raise SystemExit(check_environment(strict=strict, quick=quick))


@cli.command("probe")
@click.option("--target", "-t", required=True, help="Target IP or hostname")
@click.option("--port", "-p", default=80, show_default=True, help="Target port")
@click.option("--protocol", default="tcp", type=click.Choice(["tcp", "udp"]), help="Probe protocol")
@click.option("--ipv6", is_flag=True, default=False, help="Resolve target over IPv6")
@click.option("--allowlist", multiple=True, help="Authorized target CIDR/IP (repeatable)")
@click.option("--allowlist-file", default=None, help="File with authorized targets")
@click.option("--confirm", is_flag=True, default=False, help="Confirm authorized use")
def probe_cmd(
    target: str,
    port: int,
    protocol: str,
    ipv6: bool,
    allowlist: tuple,
    allowlist_file: str | None,
    confirm: bool,
):
    """Probe target reachability and latency."""
    from utils.network import merge_target_port, parse_port, port_override_notice, resolve_target
    from utils.scope import ScopeError, combine_allowlist, validate_engagement

    warning_panel(ETHICS_WARNING, err=True)
    try:
        require_confirm_or_abort(dry_run=False, confirm=confirm)
    except SystemExit:
        note("Aborted. Re-run with --confirm.", fg="red", bold=True, err=True)
        raise

    info = resolve_target(target, ipv6=ipv6)
    port = parse_port(merge_target_port(port, info))
    notice_text = port_override_notice(info, port)
    if notice_text:
        note(notice_text, fg="yellow", err=True)

    combined_allowlist = combine_allowlist(list(allowlist) or None, allowlist_file)
    scope_target = info.address
    if combined_allowlist and info.resolved_addresses:
        from utils.network import pick_scoped_address

        scope_target = pick_scoped_address(info, combined_allowlist)
    try:
        validate_engagement(
            scope_target,
            precombined_allowlist=combined_allowlist,
            require_allowlist=True,
            dry_run=False,
        )
    except ScopeError as exc:
        note(str(exc), fg="red", err=True)
        raise SystemExit(1) from exc

    result = probe_target(scope_target, port, protocol)

    from utils.audit import log_event
    from utils.reporter import sanitize_scope_metadata

    log_event(
        "probe",
        details={
            "target": scope_target,
            "port": port,
            "protocol": protocol,
            "reachable": result["reachable"],
            "scope": sanitize_scope_metadata(
                {
                    "target": scope_target,
                    "allowlist": combined_allowlist,
                    "allowlist_enforced": bool(combined_allowlist),
                }
            ),
        },
    )

    rows = [
        ("Target", f"{scope_target}:{port}", None),
        ("Protocol", protocol.upper(), None),
        ("IPv6", "yes" if info.is_ipv6 else "no", "ok" if info.is_ipv6 else None),
        (
            "Reachable",
            "yes" if result["reachable"] else "no",
            "ok" if result["reachable"] else "fail",
        ),
    ]
    if result.get("reachability_confidence"):
        confidence = result["reachability_confidence"]
        rows.append(
            (
                "Confidence",
                confidence,
                "warn" if confidence == "inconclusive" else "ok",
            )
        )
    if result.get("udp_status"):
        rows.append(("UDP Status", result["udp_status"], None))
    if result.get("latency_ms") is not None:
        rows.append(("Latency", f"{result['latency_ms']} ms", None))
    if result.get("error"):
        rows.append(("Error", result["error"], "fail"))
    key_values("Probe Result", rows)
    if not result["reachable"]:
        raise SystemExit(1)


@cli.command("campaign")
@click.option(
    "--profile",
    "-f",
    required=True,
    type=ProjectProfilePath(),
    help="Campaign YAML profile",
    shell_complete=_complete_profile_names,
)
@click.option("--operator", default=None, help="Operator name for audit log")
@click.option("--engagement-id", default=None, help="Override profile engagement ID")
@click.option("--allowlist", multiple=True, help="Override/add authorized target CIDR/IP")
@click.option("--allowlist-file", default=None, help="Override profile allowlist file")
@click.option(
    "--require-allowlist",
    is_flag=True,
    default=False,
    help="Require allowlist during dry-run too (live runs always require it)",
)
@click.option("--require-engagement-id", is_flag=True, default=False, help="Require engagement ID")
@click.option("--ipv6", is_flag=True, default=False, help="Resolve target over IPv6")
@click.option("--no-ipv6", is_flag=True, default=False, help="Disable IPv6 even if set in profile")
@click.option("--probe", is_flag=True, default=False, help="Probe target before/after campaign")
@click.option(
    "--no-probe", is_flag=True, default=False, help="Disable probe even if set in profile"
)
@click.option(
    "--dashboard",
    is_flag=True,
    default=False,
    help="Show a live terminal stats dashboard during each phase",
)
@click.option("--dry-run", is_flag=True, default=False, help="Validate profile without executing")
@click.option(
    "--no-report",
    "skip_report",
    is_flag=True,
    default=False,
    help="Disable automatic reports",
)
@click.option("--confirm", is_flag=True, default=False, help="Confirm authorized use")
@campaign_attack_overrides
def campaign_cmd(
    profile: str,
    operator: str | None,
    engagement_id: str | None,
    allowlist: tuple,
    allowlist_file: str | None,
    require_allowlist: bool,
    require_engagement_id: bool,
    ipv6: bool,
    no_ipv6: bool,
    probe: bool,
    no_probe: bool,
    dry_run: bool,
    skip_report: bool,
    confirm: bool,
    port: int | None,
    rate: int | None,
    broadcast: str | None,
    max_held_connections: int | None,
    payload_size: int | None,
    packet_size: int | None,
    spoof: bool,
    connection_limit: int | None,
    slow_delay: float | None,
    ramp_up: int | None,
    duration: int | None,
    threads: int | None,
    max_sockets: int | None,
    query_domain: str | None,
    concurrent_streams: int | None,
    streams_per_conn: int | None,
    community: str | None,
    bypass_preset: str | None,
    iface: str | None,
    bind_source: str | None,
    spoof_source_file: str | None,
    grpc_method: str | None,
    ws_channel: str | None,
    ws_path: str | None,
    arp_target: str | None,
    connect_port: int | None,
    sse_path: str | None,
    doh_path: str | None,
    ssl: bool,
    no_ssl: bool,
    no_spoof: bool,
    insecure: bool,
    dashboard: bool = False,
):
    """Run a multi-phase attack campaign from a YAML profile."""
    from profiles.loader import load_profile, validate_campaign_ready
    from utils.platform import is_root_user

    render_banner()

    if probe and no_probe:
        raise click.UsageError("--probe and --no-probe are mutually exclusive.")
    validate_mutually_exclusive_flags(
        (("--ipv6", ipv6), ("--no-ipv6", no_ipv6)),
        (("--ssl", ssl), ("--no-ssl", no_ssl)),
        (("--spoof", spoof), ("--no-spoof", no_spoof)),
    )
    try:
        validate_campaign_numeric_overrides(port=port, duration=duration, threads=threads)
    except click.UsageError as exc:
        note(str(exc), fg="red", err=True)
        raise SystemExit(1) from exc

    ipv6_override = False if no_ipv6 else (True if ipv6 else None)

    try:
        profile_data = load_profile(profile)
        from profiles.loader import build_campaign_allowlist

        campaign_allowlist = build_campaign_allowlist(
            profile_data,
            cli_allowlist=list(allowlist) or None,
            cli_allowlist_file=allowlist_file,
        )
        if (require_allowlist or not dry_run) and not campaign_allowlist:
            raise ValueError(
                "Campaign allowlist is required for live execution or when "
                "--require-allowlist is set. "
                "Use --allowlist CIDR or --allowlist-file path."
            )
        from utils.network import (
            merge_target_port,
            parse_port,
            port_override_notice,
            resolve_target,
        )

        profile_target_info = resolve_target(
            profile_data["target"],
            ipv6=ipv6_override if ipv6_override is not None else profile_data.get("ipv6", False),
        )
        if port is not None:
            profile_port = parse_port(port)
        else:
            profile_port = parse_port(
                merge_target_port(profile_data.get("port", 80), profile_target_info)
            )
        port_notice = port_override_notice(profile_target_info, profile_port)
        if port_notice:
            note(port_notice, fg="yellow", err=True)
    except (ValueError, FileNotFoundError) as exc:
        note(str(exc), fg="red", err=True)
        raise SystemExit(1) from exc
    except ScopeError as exc:
        note(str(exc), fg="red", err=True)
        raise SystemExit(1) from exc

    phase_overrides = build_campaign_phase_overrides(
        rate=rate,
        broadcast=broadcast,
        max_held_connections=max_held_connections,
        payload_size=payload_size,
        packet_size=packet_size,
        connection_limit=connection_limit,
        slow_delay=slow_delay,
        ramp_up=ramp_up,
        duration=duration,
        threads=threads,
        max_sockets=max_sockets,
        query_domain=query_domain,
        concurrent_streams=concurrent_streams,
        streams_per_conn=streams_per_conn,
        community=community,
        bypass_preset=bypass_preset,
        iface=iface,
        bind_source=bind_source,
        spoof_source_file=spoof_source_file,
        grpc_method=grpc_method,
        ws_channel=ws_channel,
        ws_path=ws_path,
        arp_target=arp_target,
        connect_port=connect_port,
        sse_path=sse_path,
        doh_path=doh_path,
        spoof=spoof,
        ssl=ssl,
        insecure=insecure,
        no_ssl=no_ssl,
        no_spoof=no_spoof,
    )
    phase_overrides = apply_tls_vector_defaults(
        profile_data, phase_overrides, no_ssl=no_ssl
    )

    try:
        from profiles.loader import validate_campaign_ready

        validate_campaign_ready(
            profile_data,
            phase_overrides,
            cli_allowlist=list(allowlist) or None,
            cli_allowlist_file=allowlist_file,
            ipv6_override=ipv6_override,
            default_port_override=port,
        )
    except ValueError as exc:
        note(str(exc), fg="red", err=True)
        raise SystemExit(1) from exc

    if profile_data.get("require_root") and not is_root_user():
        note(
            "Campaign profile requires root privileges (sudo) for live execution.",
            fg="yellow",
            bold=True,
            err=True,
        )
        if not dry_run:
            raise SystemExit(1)

    warning_panel(ETHICS_WARNING, err=True)
    try:
        require_confirm_or_abort(dry_run=dry_run, confirm=confirm)
    except SystemExit:
        note("Aborted. Re-run with --confirm.", fg="red", bold=True, err=True)
        raise

    controller = AttackController()
    try:
        result = controller.run_campaign(
            profile_data,
            dry_run=dry_run,
            operator=operator,
            skip_report=skip_report if skip_report else None,
            engagement_id=engagement_id,
            allowlist=list(allowlist) or None,
            allowlist_file=allowlist_file,
            require_allowlist=require_allowlist or not dry_run,
            require_engagement_id=require_engagement_id,
            ipv6=ipv6 if ipv6 else None,
            probe=probe if probe else None,
            disable_ipv6=no_ipv6,
            disable_probe=no_probe,
            phase_overrides=phase_overrides,
            dashboard=dashboard,
            port_override=port,
            profile_validated=True,
            phase_overrides_validated=True,
            phase_compat_validated=True,
        )
        render_report_paths(result.get("reports"))
    except KeyboardInterrupt:
        controller.stop()
        paths = controller.export_report(status="interrupted")
        render_report_paths(paths)
        raise SystemExit(0) from None
    except (ValueError, FileNotFoundError, ScopeError, PermissionError) as exc:
        logger.error(str(exc))
        raise SystemExit(1) from exc
    except Exception as exc:
        logger.error(f"Campaign failed: {exc}")
        controller.stop()
        paths = controller.export_report(status="failed")
        render_report_paths(paths)
        raise SystemExit(1) from exc


@cli.command("run")
@click.option("--attack", "-a", required=True, help="Attack type", shell_complete=_complete_attack_names)
@click.option("--target", "-t", required=True, help="Target IP, hostname, or URL")
@run_attack_options
def run_attack(
    attack: str,
    target: str,
    port: int,
    duration: int,
    threads: int,
    ssl: bool,
    insecure: bool,
    ipv6: bool,
    packet_size: int,
    payload_size: int,
    slow_delay: float,
    rate: int,
    ramp_up: int,
    spoof: bool,
    no_ssl: bool,
    no_spoof: bool,
    connection_limit: int,
    broadcast: str | None,
    max_held_connections: int | None,
    engagement_id: str | None,
    operator: str | None,
    allowlist: tuple,
    allowlist_file: str | None,
    probe: bool,
    dashboard: bool,
    skip_report: bool,
    dry_run: bool,
    confirm: bool,
    require_allowlist: bool,
    require_engagement_id: bool,
    max_sockets: int | None,
    query_domain: str | None,
    concurrent_streams: int | None,
    streams_per_conn: int | None,
    community: str | None,
    bypass_preset: str | None,
    iface: str | None,
    bind_source: str | None,
    spoof_source_file: str | None,
    grpc_method: str | None,
    ws_channel: str | None,
    ws_path: str | None,
    arp_target: str | None,
    connect_port: int | None,
    sse_path: str | None,
    doh_path: str | None,
):
    """Execute a DDoS test attack against the specified target."""
    from attacks import ATTACK_REGISTRY
    from utils.network import port_override_notice
    from utils.platform import is_root_user
    from utils.preflight import preflight_single_run

    render_banner()

    validate_mutually_exclusive_flags(
        (("--ssl", ssl), ("--no-ssl", no_ssl)),
        (("--spoof", spoof), ("--no-spoof", no_spoof)),
    )

    ssl = resolve_ssl_for_attack(attack, ssl, no_ssl)
    spoof = resolve_spoof_for_run(spoof, no_spoof)

    attack_options = build_run_attack_options(
        ssl=ssl,
        insecure=insecure,
        ipv6=ipv6,
        packet_size=packet_size,
        payload_size=payload_size,
        slow_delay=slow_delay,
        rate=rate,
        ramp_up=ramp_up,
        spoof=spoof,
        connection_limit=connection_limit,
        broadcast=broadcast,
        max_held_connections=max_held_connections,
        max_sockets=max_sockets,
        query_domain=query_domain,
        concurrent_streams=concurrent_streams,
        streams_per_conn=streams_per_conn,
        community=community,
        bypass_preset=bypass_preset,
        iface=iface,
        bind_source=bind_source,
        spoof_source_file=spoof_source_file,
        grpc_method=grpc_method,
        ws_channel=ws_channel,
        ws_path=ws_path,
        arp_target=arp_target,
        connect_port=connect_port,
        sse_path=sse_path,
        doh_path=doh_path,
    )

    try:
        target_info, effective_port = preflight_single_run(
            attack,
            target,
            port,
            duration,
            threads,
            attack_options,
            ipv6=ipv6,
        )
        attack_cls = ATTACK_REGISTRY[attack]
        from utils.network import pick_scoped_address
        from utils.scope import combine_allowlist, validate_engagement

        combined_allowlist = combine_allowlist(list(allowlist) or None, allowlist_file)
        scope_target = target_info.address
        if combined_allowlist and target_info.resolved_addresses:
            scope_target = pick_scoped_address(target_info, combined_allowlist)
            if scope_target != target_info.address:
                from utils.network import target_info_with_address

                target_info = target_info_with_address(target_info, scope_target)
        prevalidated_scope = validate_engagement(
            scope_target,
            engagement_id=engagement_id,
            precombined_allowlist=combined_allowlist,
            require_engagement_id=require_engagement_id,
            require_allowlist=require_allowlist or not dry_run,
            dry_run=dry_run,
        )
        port_notice = port_override_notice(target_info, effective_port)
        if port_notice:
            note(port_notice, fg="yellow", err=True)
    except ValueError as exc:
        note(str(exc), fg="red", err=True)
        raise SystemExit(1) from exc

    if attack_cls.requires_root and not is_root_user():
        note(
            f"{attack} requires root privileges (sudo). "
            "Dry-run also validates this requirement.",
            fg="yellow",
            bold=True,
            err=True,
        )
        if not dry_run:
            raise SystemExit(1)

    warning_panel(ETHICS_WARNING, err=True)
    try:
        require_confirm_or_abort(dry_run=dry_run, confirm=confirm)
    except SystemExit:
        note("Aborted. Re-run with --confirm.", fg="red", bold=True, err=True)
        raise

    controller = AttackController()
    try:
        result = controller.run(
            attack_type=attack,
            target=target,
            port=effective_port,
            duration=duration,
            threads=threads,
            engagement_id=engagement_id,
            operator=operator,
            allowlist=list(allowlist) or None,
            allowlist_file=allowlist_file,
            require_allowlist=require_allowlist or not dry_run,
            require_engagement_id=require_engagement_id,
            probe=probe,
            dashboard=dashboard,
            skip_report=skip_report,
            dry_run=dry_run,
            cached_target_info=target_info,
            prevalidated_scope=prevalidated_scope,
            skip_scope_check=True,
            skip_param_validation=True,
            skip_compat_validation=True,
            **attack_options,
        )
        render_report_paths(result.get("reports"))
    except KeyboardInterrupt:
        controller.stop()
        paths = controller.export_report(status="interrupted")
        render_report_paths(paths)
        raise SystemExit(0) from None
    except (ValueError, FileNotFoundError, ScopeError, PermissionError) as exc:
        logger.error(str(exc))
        raise SystemExit(1) from exc
    except Exception as exc:
        logger.error(f"Attack failed: {exc}")
        controller.stop()
        paths = controller.export_report(status="failed")
        render_report_paths(paths)
        raise SystemExit(1) from exc


@cli.command("info")
@click.argument("attack_name", shell_complete=_complete_attack_names)
def attack_info(attack_name: str):
    """Show details about a specific attack type."""
    from attacks import ATTACK_REGISTRY
    from utils.cli_helpers import (
        attack_ssl_example_flag,
        attack_supports_rate,
        attack_transport_hint,
    )

    if attack_name not in ATTACK_REGISTRY:
        click.echo(format_unknown_attack_error(attack_name, list(ATTACK_REGISTRY.keys())))
        raise SystemExit(1)

    cls = ATTACK_REGISTRY[attack_name]
    supports_rate = attack_supports_rate(cls.name, cls.category)
    transport = attack_transport_hint(cls.name, cls.category)

    rows = [
        ("Name", cls.name, None),
        ("Category", cls.category, None),
        ("Description", cls.description, None),
        (
            "Root",
            "required" if cls.requires_root else "not required",
            "warn" if cls.requires_root else "ok",
        ),
        ("Rate Limit", "supported (--rate)" if supports_rate else "not applicable", None),
    ]
    if transport:
        rows.append(("Transport", transport, None))
    key_values("Attack Details", rows)
    section("Example")
    ssl_flag = attack_ssl_example_flag(cls.name, cls.category)
    note(
        f"  mudos run -a {cls.name} -t TARGET --confirm "
        f"--allowlist TARGET/32 -d 30 -n 10{ssl_flag}"
    )


@cli.command("reports")
@click.option("--limit", "-n", default=10, type=click.IntRange(min=1), help="Number of recent reports to show")
def list_reports(limit: int):
    """List recently generated HTML reports."""
    from utils.reporter import HTML_DIR

    html_dir = HTML_DIR
    if not html_dir.exists():
        note("No reports found. Run an attack to generate reports.", fg="yellow")
        return

    files = sorted(html_dir.glob("*.html"), key=lambda p: p.stat().st_mtime, reverse=True)
    if not files:
        note("No HTML reports found.", fg="yellow")
        return

    rows = []
    for path in files[:limit]:
        mtime = datetime.fromtimestamp(path.stat().st_mtime).strftime("%Y-%m-%d %H:%M:%S")
        try:
            display_path = path.relative_to(html_dir)
        except ValueError:
            display_path = path.name
        rows.append((mtime, str(display_path)))
    table(
        ("Generated", "Report File"),
        rows,
        title="Recent HTML Reports",
        footer=f"Total: {len(files)} reports in {html_dir.name}/",
        max_col_widths=(22, 90),
    )


def check_environment(*, strict: bool = False, quick: bool = False) -> int:
    """Verify dependencies and environment readiness. Returns process exit code."""
    import sys
    from importlib.metadata import version as pkg_version

    import aiohttp
    import colorama
    import psutil
    import yaml

    try:
        import click as click_lib
        click_version = pkg_version("click")
    except Exception:
        import click as click_lib
        click_version = getattr(click_lib, "__version__", "unknown")
    try:
        jsonschema_version = pkg_version("jsonschema")
    except Exception:
        jsonschema_version = "installed"
    try:
        rich_version = pkg_version("rich")
    except Exception:
        rich_version = "installed"

    try:
        import scapy
        scapy_version = scapy.__version__
    except ImportError:
        scapy_version = "not installed (FAIL for root-required attacks)"

    libpcap_status = "n/a"
    libpcap_level = None
    if not scapy_version.startswith("not installed"):
        from utils.scapy_runtime import activate_scapy_libpcap

        if activate_scapy_libpcap():
            libpcap_status = "active (Scapy pcap)"
            libpcap_level = "ok"
        else:
            libpcap_status = "not available — install libpcap for raw/L2 captures"
            libpcap_level = "warn"

    try:
        import httpx
        httpx_version = httpx.__version__
    except ImportError:
        httpx_version = "not installed"

    try:
        import h2

        h2_version = getattr(h2, "__version__", "installed")
    except ImportError:
        h2_version = "not installed"

    try:
        import aioquic

        aioquic_version = getattr(aioquic, "__version__", "installed")
    except ImportError:
        aioquic_version = "not installed"

    is_root = is_root_user()
    rows = [
        ("Edition", APP_EDITION, "ok"),
        ("Python", sys.version.split()[0], "ok" if sys.version_info >= (3, 10) else "fail"),
        ("aiohttp", aiohttp.__version__, "ok"),
        ("httpx", httpx_version, "fail" if httpx_version == "not installed" else "ok"),
        ("h2", h2_version, "fail" if h2_version == "not installed" else "ok"),
        ("aioquic", aioquic_version, "fail" if aioquic_version == "not installed" else "ok"),
        ("scapy", scapy_version, "warn" if scapy_version.startswith("not installed") else "ok"),
        ("libpcap", libpcap_status, libpcap_level),
        ("psutil", psutil.__version__, "ok"),
        ("colorama", colorama.__version__, "ok"),
        ("click", click_version, "ok"),
        ("pyyaml", yaml.__version__, "ok"),
        ("jsonschema", jsonschema_version, "ok"),
        ("rich", rich_version, "ok"),
        (
            "Root Access",
            "yes" if is_root else "no (required for raw socket attacks)",
            "ok" if is_root else "warn",
        ),
    ]

    from attacks.plugins import plugin_check_status

    plugin_text, plugin_level = plugin_check_status()
    rows.append(("Attack Plugins", plugin_text, plugin_level))

    if not quick:
        from attacks import ATTACK_REGISTRY
        from attacks.taxonomy import coverage_summary
        from profiles.loader import list_profile_names

        profile_count = len(list_profile_names())
        coverage = coverage_summary(ATTACK_REGISTRY.keys())
        rows.extend(
            [
                ("Attacks", f"{len(ATTACK_REGISTRY)} vectors registered", "ok"),
                (
                    "DDoS Coverage",
                    f"{coverage['covered_families']}/{coverage['in_scope_families']} families",
                    "fail" if coverage["missing_families"] else "ok",
                ),
                ("Profiles", f"{profile_count} campaign profiles", "ok"),
            ]
        )
    else:
        coverage = {"missing_families": []}

    rows.extend(
        [
            ("Developer", DEVELOPER_NAME, None),
            ("Contact", DEVELOPER_EMAIL, None),
        ]
    )
    key_values(f"{APP_NAME} v{APP_VERSION} Environment", rows)

    exit_code = 0
    if scapy_version.startswith("not installed"):
        exit_code = 1
    elif libpcap_level == "warn":
        exit_code = 1 if strict else 0
    if coverage["missing_families"]:
        exit_code = 1
    if strict and (h2_version == "not installed" or aioquic_version == "not installed"):
        exit_code = 1
    if sys.version_info < (3, 10):
        exit_code = 1
    if httpx_version == "not installed":
        exit_code = 1
    return exit_code


if __name__ == "__main__":
    cli()