controller.py

"""Central controller for orchestrating MUDOS attacks."""

from __future__ import annotations

import math
import signal
import threading
import time
from typing import Any

from attacks import ATTACK_REGISTRY, BaseAttack
from config import APP_NAME, APP_VERSION
from utils.amplification_scope import (
    pin_amplification_endpoints,
    require_allowlist_for_amplification,
    resolve_broadcast_default,
    validate_amplification_endpoints,
    validate_broadcast_endpoint,
)
from utils.audit import log_event
from utils.logger import setup_logger
from utils.network import (
    TargetInfo,
    merge_target_port,
    parse_port,
    pick_scoped_address,
    resolve_phase_port,
    resolve_target,
    target_info_to_dict,
    target_info_with_address,
)
from utils.params import validate_run_params
from utils.platform import is_root_user
from utils.probe import compare_probe, probe_ports_mixed, probe_target
from utils.probe_protocol import probe_protocol_for_attack, probe_protocols_for_phases
from utils.reporter import EngagementReport
from utils.safety import (
    check_attack_safety,
    validate_attack_compatibility,
    validate_auxiliary_target_scope,
    validate_source_address_scope,
)
from utils.scope import ScopeError, combine_allowlist, is_target_in_scope, validate_engagement
from utils.stats import get_system_stats

logger = setup_logger()


def _scope_metadata(scope: dict[str, Any]) -> dict[str, Any]:
    meta = dict(scope)
    meta["target"] = meta.get("target") or meta.get("target_ip")
    return meta


class AttackController:
    """Manages attack lifecycle, campaigns, stats, and reporting."""

    _signal_handlers_registered = False
    _signal_handler_depth = 0
    _previous_signal_handlers: dict[int, Any] = {}
    _active_controller: AttackController | None = None

    def __init__(self) -> None:
        self._attack: BaseAttack | None = None
        self._stats_thread: threading.Thread | None = None
        self._shutdown = threading.Event()
        self._stats_shutdown = threading.Event()
        self._report: EngagementReport | None = None
        self._campaign_active = False
        self._skip_report = False
        self._current_attack_name: str | None = None
        self._interrupted = False
        self._orphan_abort = False
        self._report_exported = False
        self._report_export_lock = threading.Lock()
        self._campaign_scope: dict[str, Any] | None = None
        self._signal_acquisitions = 0

    def list_attacks(self) -> list[dict[str, str]]:
        return [
            {
                "name": name,
                "category": cls.category,
                "description": cls.description,
                "requires_root": cls.requires_root,
            }
            for name, cls in sorted(ATTACK_REGISTRY.items())
        ]

    def _require_root_if_needed(self, attack_type: str) -> None:
        attack_cls = ATTACK_REGISTRY[attack_type]
        if attack_cls.requires_root and not is_root_user():
            raise PermissionError(
                f"{attack_type} requires root privileges. Run with sudo."
            )

    def _warn_root_if_needed(self, attack_type: str, phase_index: int) -> None:
        attack_cls = ATTACK_REGISTRY[attack_type]
        if attack_cls.requires_root and not is_root_user():
            logger.warning(
                "Campaign dry-run phase %d (%s): live execution requires root (sudo).",
                phase_index,
                attack_type,
            )

    def create_attack(
        self,
        attack_type: str,
        target: str,
        port: int = 80,
        duration: int = 60,
        threads: int = 10,
        *,
        target_info: TargetInfo | None = None,
        check_root: bool = True,
        **options: Any,
    ) -> BaseAttack:
        if attack_type not in ATTACK_REGISTRY:
            from utils.cli_helpers import format_unknown_attack_error

            raise ValueError(format_unknown_attack_error(attack_type, list(ATTACK_REGISTRY.keys())))

        if target_info is not None:
            resolved_target = target_info.address
            options.setdefault("is_ipv6", target_info.is_ipv6)
            options.setdefault("host_header", target_info.host_header)
            options.setdefault("target_info", target_info_to_dict(target_info))
        else:
            ipv6 = options.get("ipv6", False)
            target_info = resolve_target(target, ipv6=ipv6)
            resolved_target = target_info.address
            options["is_ipv6"] = target_info.is_ipv6
            options["host_header"] = target_info.host_header
            options["target_info"] = target_info_to_dict(target_info)

        port = parse_port(port)
        attack_cls = ATTACK_REGISTRY[attack_type]

        for key in ("allowlist_configured", "require_engagement_id", "require_root"):
            options.pop(key, None)

        if check_root:
            self._require_root_if_needed(attack_type)

        self._attack = attack_cls(
            target=resolved_target,
            port=port,
            duration=duration,
            threads=threads,
            **options,
        )
        return self._attack

    def _combined_allowlist(
        self,
        allowlist: list[str] | None,
        allowlist_file: str | None,
    ) -> list[str] | None:
        return combine_allowlist(allowlist, allowlist_file)

    def _validate_attack_scope(
        self,
        attack_type: str,
        resolved_target: str,
        options: dict[str, Any],
        scope: dict[str, Any],
        *,
        dry_run: bool = False,
        pin_endpoints: bool = True,
        skip_compat_checks: bool = False,
        skip_root_check: bool = False,
    ) -> None:
        if not skip_compat_checks:
            validate_attack_compatibility(attack_type, options)
        if not skip_root_check:
            self._require_root_if_needed(attack_type)
        check_attack_safety(
            attack_type,
            {**options, "dry_run": True} if dry_run else options,
            logger,
        )
        require_allowlist_for_amplification(attack_type, scope.get("allowlist"))
        needs_source_scope = any(
            options.get(key)
            for key in ("bind_source", "spoof_source_pool", "arp_target")
        )
        from attacks import ATTACK_REGISTRY
        from utils.amplification_scope import BROADCAST_ATTACK_TYPES

        attack_cls = ATTACK_REGISTRY.get(attack_type)
        scope_sensitive = (
            attack_cls is not None
            and (attack_cls.category == "amplification" or attack_type in BROADCAST_ATTACK_TYPES)
        )
        if (
            not skip_compat_checks
            or scope.get("allowlist")
            or needs_source_scope
            or scope_sensitive
        ):
            validate_auxiliary_target_scope(attack_type, options, scope.get("allowlist"))
            validate_source_address_scope(attack_type, options, scope.get("allowlist"))
        if scope.get("allowlist"):
            validate_amplification_endpoints(
                attack_type, resolved_target, options, scope["allowlist"]
            )
            validate_broadcast_endpoint(attack_type, options, scope["allowlist"])
        if scope.get("allowlist") and pin_endpoints:
            pin_amplification_endpoints(attack_type, resolved_target, options)

    def _build_phase_options(
        self,
        phase: dict[str, Any],
        *,
        profile_ipv6: bool,
        campaign_allowlist: list[str] | None,
        target_info: TargetInfo,
        effective_overrides: dict[str, Any],
    ) -> dict[str, Any]:
        phase_options = {
            k: v
            for k, v in phase.items()
            if k not in ("attack", "port", "description", "duration", "threads")
        }
        phase_options.update(
            {
                k: v
                for k, v in effective_overrides.items()
                if k not in ("duration", "threads")
            }
        )
        phase_options["ipv6"] = profile_ipv6
        phase_options["allowlist_configured"] = bool(campaign_allowlist)
        phase_options["is_ipv6"] = target_info.is_ipv6
        from utils.params import normalize_attack_options

        return normalize_attack_options(phase_options, phase["attack"])

    def _record_skipped_phases(
        self,
        profile: dict[str, Any],
        start_index: int,
        default_port: int,
        target_info: TargetInfo,
        target: str,
        effective_overrides: dict[str, Any] | None = None,
    ) -> None:
        overrides = effective_overrides or {}
        for phase in profile["phases"][start_index:]:
            phase_duration, phase_threads = self._phase_timing(phase, overrides)
            phase_port = resolve_phase_port(phase, default_port, target_info)
            self._report.add_phase(
                attack=phase["attack"],
                config={
                    "target": target_info.address,
                    "port": phase_port,
                    "duration": phase_duration,
                    "threads": phase_threads,
                    "skipped": True,
                    "reason": "campaign_interrupted",
                },
                result={"status": "skipped", "packets_sent": 0, "errors": 0},
                description=phase.get("description"),
            )

    def export_report(self, status: str = "completed") -> dict[str, str] | None:
        """Export engagement report artifacts (JSON, HTML, PDF when available)."""
        return self._export_report(status=status)

    def _export_report(self, status: str = "completed") -> dict[str, str] | None:
        with self._report_export_lock:
            if self._skip_report or not self._report or self._report_exported:
                return None
            self._report.status = status
            last_exc: Exception | None = None
            for attempt in range(3):
                try:
                    paths = self._report.export_all()
                    self._report_exported = True
                    logger.info(f"JSON report: {paths['json']}")
                    logger.info(f"HTML report: {paths['html']}")
                    if paths.get("pdf"):
                        logger.info(f"PDF report: {paths['pdf']}")
                    result = {"json": str(paths["json"]), "html": str(paths["html"])}
                    if paths.get("pdf"):
                        result["pdf"] = str(paths["pdf"])
                    return result
                except OSError as exc:
                    last_exc = exc
                    if attempt < 2:
                        time.sleep(0.5 * (attempt + 1))
                except Exception as exc:
                    logger.exception(f"Report export failed: {exc}")
                    return None
            logger.error(f"Report export failed after retries: {last_exc}")
            return None

    def _join_stats_thread(self, report_interval: float) -> None:
        if self._stats_thread and self._stats_thread.is_alive():
            self._stats_thread.join(timeout=report_interval + 1)
            if self._stats_thread.is_alive():
                logger.warning("Stats sampling thread did not stop within timeout")
        if self._stats_thread and not self._stats_thread.is_alive():
            self._stats_thread = None

    def _cleanup_run(self, attack: BaseAttack | None, report_interval: float) -> None:
        self._stats_shutdown.set()
        if attack is not None and (
            attack.is_running() or self._attack_has_active_workers(attack)
        ):
            attack.stop()
        if attack is not None:
            worker_threads = getattr(attack, "_worker_threads", ())
            orphans = [thread for thread in worker_threads if thread.is_alive()]
            async_thread = getattr(attack, "_async_thread", None)
            if async_thread is not None and async_thread.is_alive():
                orphans.append(async_thread)
            if orphans:
                logger.error(
                    "%s left %d orphan worker thread(s) after stop",
                    attack.name,
                    len(orphans),
                )
                self._interrupted = True
                if self._campaign_active:
                    logger.error("Stopping remaining campaign phases")
                    self._orphan_abort = True
                    self._shutdown.set()
        if self._attack is attack:
            self._attack = None
        self._join_stats_thread(report_interval)
        if not self._campaign_active:
            AttackController._active_controller = None
            if self._signal_acquisitions:
                AttackController._release_signal_handlers()
                self._signal_acquisitions -= 1

    @classmethod
    def _release_signal_handlers(cls) -> None:
        if cls._signal_handler_depth <= 0:
            return
        cls._signal_handler_depth -= 1
        if cls._signal_handler_depth == 0:
            cls._restore_signal_handlers()

    @classmethod
    def _reset_signal_handlers(cls) -> None:
        cls._signal_handler_depth = 0
        cls._restore_signal_handlers()

    @classmethod
    def _restore_signal_handlers(cls) -> None:
        if not cls._signal_handlers_registered:
            return
        for sig, handler in cls._previous_signal_handlers.items():
            try:
                signal.signal(sig, handler)
            except (ValueError, OSError):
                pass
        cls._previous_signal_handlers.clear()
        cls._signal_handlers_registered = False

    def _phase_timing(
        self,
        phase: dict[str, Any],
        effective_overrides: dict[str, Any],
        *,
        strict: bool = False,
    ) -> tuple[int, int]:
        from config import MAX_DURATION_SECONDS
        from utils.params import MAX_THREADS

        def _bounded_int(value: Any, default: int, *, minimum: int, maximum: int) -> int:
            try:
                parsed = int(value)
            except (TypeError, ValueError) as exc:
                if strict:
                    raise ValueError(f"Invalid integer value: {value!r}") from exc
                return default
            return max(minimum, min(parsed, maximum))

        duration = effective_overrides.get("duration", phase.get("duration", 60))
        threads = effective_overrides.get("threads", phase.get("threads", 10))
        return (
            _bounded_int(duration, 60, minimum=1, maximum=MAX_DURATION_SECONDS),
            _bounded_int(threads, 10, minimum=1, maximum=MAX_THREADS),
        )

    def run(
        self,
        attack_type: str,
        target: str,
        port: int = 80,
        duration: int = 60,
        threads: int = 10,
        report_interval: float = 2.0,
        engagement_id: str | None = None,
        operator: str | None = None,
        allowlist: list[str] | None = None,
        allowlist_file: str | None = None,
        dry_run: bool = False,
        probe: bool = False,
        skip_report: bool = False,
        require_allowlist: bool | None = None,
        skip_scope_check: bool = False,
        skip_param_validation: bool = False,
        skip_compat_validation: bool = False,
        cached_target_info: TargetInfo | None = None,
        prevalidated_scope: dict[str, Any] | None = None,
        dashboard: bool = False,
        clear_allowlist_cache: bool = False,
        **options: Any,
    ) -> dict:
        report_interval = float(report_interval)
        if not math.isfinite(report_interval) or report_interval <= 0:
            raise ValueError("report_interval must be a positive finite number.")
        if require_allowlist is None:
            require_allowlist = not dry_run
        if not dry_run and not require_allowlist:
            logger.warning(
                "Live run with require_allowlist=False; authorized scope enforcement is disabled."
            )
            log_event(
                "scope_enforcement_disabled",
                engagement_id=engagement_id,
                operator=operator,
                details={"attack": attack_type, "target": target},
            )
        if threading.current_thread() is not threading.main_thread():
            logger.warning(
                "Attack invoked off the main thread; SIGINT/SIGTERM handlers are not registered."
            )
        options = dict(options)
        if clear_allowlist_cache:
            from utils.scope import clear_allowlist_cache

            clear_allowlist_cache()
            from utils.network import clear_dns_cache

            clear_dns_cache()
        ipv6 = options.get("ipv6", False)
        if cached_target_info is not None:
            target_info = cached_target_info
        else:
            target_info = resolve_target(target, ipv6=ipv6)
        resolved_target = target_info.address
        port = merge_target_port(port, target_info)

        if not skip_param_validation:
            options = validate_run_params(
                duration=duration,
                threads=threads,
                port=port,
                options=options,
                attack_type=attack_type,
            )
        else:
            from utils.params import prepare_attack_options

            options = prepare_attack_options(
                options, attack_type, threads=threads, duration=duration
            )

        combined_allowlist = self._combined_allowlist(allowlist, allowlist_file)
        scope_prevalidated = prevalidated_scope is not None
        effective_allowlist = combined_allowlist
        if scope_prevalidated and not effective_allowlist and prevalidated_scope:
            effective_allowlist = prevalidated_scope.get("allowlist")
        if effective_allowlist and target_info.resolved_addresses:
            scoped_address = pick_scoped_address(target_info, effective_allowlist)
            if scoped_address != resolved_target:
                target_info = target_info_with_address(target_info, scoped_address)
                resolved_target = scoped_address

        if scope_prevalidated:
            scope = dict(prevalidated_scope)
            options.pop("require_engagement_id", None)
            pre_allowlist = scope.get("allowlist")
            if pre_allowlist is not None and not is_target_in_scope(
                resolved_target, pre_allowlist
            ):
                raise ScopeError(
                    f"Target {resolved_target} is outside prevalidated authorized scope."
                )
        elif not skip_scope_check:
            scope = validate_engagement(
                resolved_target,
                engagement_id=engagement_id,
                precombined_allowlist=combined_allowlist,
                require_engagement_id=options.pop("require_engagement_id", False),
                require_allowlist=require_allowlist,
                dry_run=dry_run,
            )
        else:
            require_eid = options.pop("require_engagement_id", False)
            scope = validate_engagement(
                resolved_target,
                engagement_id=engagement_id,
                allowlist=combined_allowlist,
                allowlist_file=None,
                require_engagement_id=require_eid,
                require_allowlist=require_allowlist,
                dry_run=dry_run,
            )

        if scope.get("allowlist"):
            resolve_broadcast_default(attack_type, options, scope["allowlist"])

        options["is_ipv6"] = target_info.is_ipv6
        options["host_header"] = target_info.host_header
        options["target_info"] = target_info_to_dict(target_info)
        options["allowlist_configured"] = bool(scope.get("allowlist_enforced"))

        if self._campaign_active and self._interrupted:
            return {
                "status": "interrupted",
                "stats": {},
                "probe_delta": None,
                "skipped_before_start": True,
            }

        self._skip_report = skip_report

        if dry_run:
            self._validate_attack_scope(
                attack_type,
                resolved_target,
                options,
                scope,
                dry_run=True,
            )
            report = EngagementReport(
                engagement_id=engagement_id,
                operator=operator,
                target=resolved_target,
                report_type="dry_run",
            )
            report.set_metadata(scope=_scope_metadata(scope))
            report.add_dry_run_phase(
                attack=attack_type,
                config={
                    "target": resolved_target,
                    "port": port,
                    "duration": duration,
                    "threads": threads,
                    **{k: v for k, v in options.items() if k not in ("require_root",)},
                },
            )
            report.status = "dry_run"
            self._report = report
            logger.info(f"[DRY-RUN] Would run {attack_type} against {resolved_target}:{port}")
            from utils.reporter import sanitize_scope_metadata

            log_event(
                "dry_run",
                engagement_id=engagement_id,
                operator=operator,
                details={"scope": sanitize_scope_metadata(_scope_metadata(scope))},
            )
            report_paths = self._export_report(status="dry_run")
            return {"dry_run": True, "scope": scope, "reports": report_paths}

        if not self._campaign_active:
            self._report = EngagementReport(
                engagement_id=engagement_id,
                operator=operator,
                target=resolved_target,
                report_type="single_attack",
            )
            self._report.set_metadata(scope=_scope_metadata(scope))

        self._validate_attack_scope(
            attack_type, resolved_target, options, scope,
            skip_compat_checks=skip_compat_validation,
        )

        probe_before = None
        attack_cls = ATTACK_REGISTRY[attack_type]
        probe_protocol = probe_protocol_for_attack(attack_type, attack_cls.category)
        if probe and self._report.probe_before is None:
            probe_before = probe_target(resolved_target, port, probe_protocol)
            self._report.record_probe_before(probe_before)
            status = "reachable" if probe_before["reachable"] else "unreachable"
            logger.info(f"Pre-attack probe: {status}")
        elif probe and self._report.probe_before:
            probe_before = self._report.probe_before

        attack: BaseAttack | None = None
        try:
            attack = self.create_attack(
                attack_type=attack_type,
                target=target,
                port=port,
                duration=duration,
                threads=threads,
                target_info=target_info,
                check_root=False,
                **{
                    k: v
                    for k, v in options.items()
                    if k not in ("target_info", "is_ipv6", "host_header", "allowlist_configured")
                },
            )
            self._current_attack_name = attack_type

            log_event(
                "attack_start",
                engagement_id=engagement_id,
                operator=operator,
                details={"attack": attack_type, "target": resolved_target, "port": port},
            )

            AttackController._active_controller = self
            self._register_signal_handlers()
            logger.info(f"[{APP_NAME} v{APP_VERSION}] Starting {attack.name} -> {attack.target}:{attack.port}")
            logger.info(
                f"Duration: {duration}s | Threads: {threads} | "
                f"Rate: {options.get('rate', 'unlimited')} pps"
            )

            self._shutdown.clear()
            self._stats_shutdown.clear()
            if not self._campaign_active:
                self._interrupted = False
                self._report_exported = False
            self._stats_thread = threading.Thread(
                target=self._report_stats,
                args=(attack, report_interval, attack_type, dashboard),
                daemon=True,
            )
            self._stats_thread.start()

            attack.start()
            attack.wait()

            final_stats = attack.get_stats()
            if self._interrupted:
                final_stats = {**final_stats, "status": "interrupted"}
            system = get_system_stats()
            self._log_stats(final_stats, system)
            if self._report:
                self._report.add_timeline_sample(
                    final_stats, system, attack=attack_type
                )

            probe_delta = None
            if probe:
                probe_after = probe_target(resolved_target, port, probe_protocol)
                if not self._campaign_active:
                    self._report.record_probe_after(probe_after)
                probe_delta = compare_probe(probe_before or self._report.probe_before or {}, probe_after)
                from utils.reporter import sanitize_scope_metadata

                log_event(
                    "probe",
                    engagement_id=engagement_id,
                    operator=operator,
                    details={
                        "target": resolved_target,
                        "port": port,
                        "protocol": probe_protocol,
                        "reachable": probe_after.get("reachable"),
                        "status": probe_delta.get("status"),
                        "scope": sanitize_scope_metadata(_scope_metadata(scope)),
                    },
                )
                if not self._campaign_active:
                    logger.info(f"Post-attack probe status: {probe_delta['status']}")

            self._report.add_phase(
                attack=attack_type,
                config={
                    "target": resolved_target,
                    "port": port,
                    "duration": duration,
                    "threads": threads,
                    **{
                        k: v
                        for k, v in options.items()
                        if k
                        not in (
                            "require_root",
                            "require_engagement_id",
                            "target_info",
                            "is_ipv6",
                            "host_header",
                            "allowlist_configured",
                        )
                    },
                },
                result=final_stats,
                probe_delta=probe_delta,
            )
            if self._campaign_active and self._report:
                self._report.mark_phase_timeline_boundary(final_stats)

            log_event(
                "attack_complete",
                engagement_id=engagement_id,
                operator=operator,
                details={"attack": attack_type, "stats": final_stats},
            )

            status = "interrupted" if self._interrupted else "completed"
            logger.info(f"Attack {status}.")

            report_paths = None
            if not self._campaign_active:
                report_paths = self._export_report(status=status)

            result: dict[str, Any] = {"stats": final_stats, "probe_delta": probe_delta, "status": status}
            if report_paths:
                result["reports"] = report_paths

            return result
        except Exception:
            if not self._campaign_active:
                self._export_report(status="failed")
            raise
        finally:
            self._cleanup_run(attack, report_interval)

    def run_campaign(
        self,
        profile: dict[str, Any],
        *,
        dry_run: bool = False,
        operator: str | None = None,
        skip_report: bool | None = None,
        engagement_id: str | None = None,
        allowlist: list[str] | None = None,
        allowlist_file: str | None = None,
        require_allowlist: bool | None = None,
        require_engagement_id: bool = False,
        ipv6: bool | None = None,
        probe: bool | None = None,
        disable_probe: bool = False,
        disable_ipv6: bool = False,
        phase_overrides: dict[str, Any] | None = None,
        dashboard: bool = False,
        port_override: int | None = None,
        profile_validated: bool = False,
        phase_overrides_validated: bool = False,
        phase_compat_validated: bool = False,
    ) -> dict:
        from profiles.loader import (
            assert_profile_campaign_ready,
            build_campaign_allowlist,
            validate_campaign_phase_overrides,
            validate_profile,
        )
        from utils.network import clear_dns_cache
        from utils.scope import clear_allowlist_cache

        clear_allowlist_cache()
        clear_dns_cache()
        effective_overrides = {
            k: v for k, v in (phase_overrides or {}).items() if v is not None
        }
        if require_allowlist is None:
            require_allowlist = not dry_run
        cli_allowlist = list(allowlist) if allowlist else None
        if disable_ipv6:
            profile_ipv6 = False
        else:
            profile_ipv6 = ipv6 if ipv6 is not None else profile.get("ipv6", False)
        campaign_allowlist = build_campaign_allowlist(
            profile,
            cli_allowlist=cli_allowlist,
            cli_allowlist_file=allowlist_file,
        )
        if profile_validated:
            assert_profile_campaign_ready(
                profile,
                cli_allowlist=cli_allowlist,
                cli_allowlist_file=allowlist_file,
                ipv6_override=profile_ipv6,
                phase_overrides=effective_overrides if not phase_overrides_validated else None,
            )
        else:
            validate_profile(
                profile,
                cli_allowlist=cli_allowlist,
                cli_allowlist_file=allowlist_file,
                ipv6_override=profile_ipv6,
            )

        target = profile["target"]
        default_port = profile.get("port", 80)
        engagement_id = engagement_id or profile.get("engagement_id")
        if disable_probe:
            probe_enabled = False
        else:
            probe_enabled = probe if probe is not None else profile.get("probe", False)
        require_allowlist = require_allowlist or profile.get("require_allowlist", False)
        require_engagement_id = require_engagement_id or profile.get("require_engagement_id", False)
        profile_report = profile.get("report", True)
        effective_skip_report = skip_report if skip_report is not None else not profile_report

        target_info = resolve_target(target, ipv6=profile_ipv6)
        resolved_target = target_info.address
        default_port = merge_target_port(default_port, target_info)
        if port_override is not None:
            default_port = parse_port(port_override)

        if campaign_allowlist and target_info.resolved_addresses:
            scoped_address = pick_scoped_address(target_info, campaign_allowlist)
            if scoped_address != resolved_target:
                target_info = target_info_with_address(target_info, scoped_address)
                resolved_target = scoped_address

        scope = _scope_metadata(
            validate_engagement(
                resolved_target,
                engagement_id=engagement_id,
                allowlist=campaign_allowlist,
                allowlist_file=None,
                require_allowlist=require_allowlist,
                require_engagement_id=require_engagement_id,
                dry_run=dry_run,
            )
        )
        self._campaign_scope = scope

        self._skip_report = effective_skip_report
        self._interrupted = False
        self._orphan_abort = False
        self._shutdown.clear()
        self._stats_shutdown.clear()
        self._report_exported = False
        self._report = EngagementReport(
            engagement_id=engagement_id,
            operator=operator or profile.get("operator"),
            target=resolved_target,
            report_type="campaign",
            profile_name=profile.get("name"),
        )
        self._report.set_metadata(
            profile=profile.get("name"),
            description=profile.get("description"),
            cooldown=profile.get("cooldown", 5),
            scope=scope,
        )

        if profile.get("require_root") and not is_root_user() and not dry_run:
            raise PermissionError("Campaign profile requires root privileges. Run with sudo.")

        self._campaign_active = True

        log_event(
            "campaign_start",
            engagement_id=engagement_id,
            operator=operator,
            details={"phases": len(profile["phases"]), "target": target},
        )

        logger.info(f"Starting campaign: {profile.get('name', 'unnamed')} ({len(profile['phases'])} phases)")

        if not phase_overrides_validated:
            validate_campaign_phase_overrides(
                profile,
                effective_overrides,
                ipv6_override=profile_ipv6,
                default_port_override=port_override,
                cli_allowlist=cli_allowlist,
                cli_allowlist_file=allowlist_file,
            )
        if threading.current_thread() is not threading.main_thread():
            logger.warning(
                "Campaign invoked off the main thread; SIGINT/SIGTERM handlers are not registered."
            )

        try:
            if dry_run:
                for i, phase in enumerate(profile["phases"], 1):
                    phase_port = resolve_phase_port(phase, default_port, target_info)
                    phase_options = self._build_phase_options(
                        phase,
                        profile_ipv6=profile_ipv6,
                        campaign_allowlist=campaign_allowlist,
                        target_info=target_info,
                        effective_overrides=effective_overrides,
                    )
                    phase_duration, phase_threads = self._phase_timing(
                        phase, effective_overrides, strict=True
                    )
                    phase_options = validate_run_params(
                        duration=phase_duration,
                        threads=phase_threads,
                        port=phase_port,
                        options=phase_options,
                        attack_type=phase["attack"],
                    )
                    phase_broadcast_options = dict(phase_options)
                    if campaign_allowlist:
                        resolve_broadcast_default(
                            phase["attack"], phase_broadcast_options, campaign_allowlist
                        )
                        if "broadcast" in phase_broadcast_options:
                            phase_options["broadcast"] = phase_broadcast_options["broadcast"]
                    self._validate_attack_scope(
                        phase["attack"],
                        resolved_target,
                        phase_broadcast_options if campaign_allowlist else phase_options,
                        scope,
                        dry_run=True,
                        pin_endpoints=bool(campaign_allowlist),
                        skip_root_check=True,
                    )
                    self._warn_root_if_needed(phase["attack"], i)
                    self._report.add_dry_run_phase(
                        attack=phase["attack"],
                        config={
                            "target": resolved_target,
                            "port": phase_port,
                            "duration": phase_duration,
                            "threads": phase_threads,
                            **{
                                k: v
                                for k, v in phase_options.items()
                                if k not in ("require_root", "require_engagement_id")
                            },
                        },
                        description=phase.get("description"),
                    )
                    logger.info(f"[DRY-RUN] Phase {i}: {phase['attack']} ({phase.get('duration', 60)}s)")
                from utils.reporter import sanitize_scope_metadata

                log_event(
                    "dry_run",
                    engagement_id=engagement_id,
                    operator=operator,
                    details={
                        "scope": sanitize_scope_metadata(_scope_metadata(scope)),
                        "phases": len(profile["phases"]),
                        "profile": profile.get("name"),
                    },
                )
                report_paths = self._export_report(status="dry_run")
                return {"dry_run": True, "phases": len(profile["phases"]), "reports": report_paths}

            if probe_enabled:
                port_protocols = probe_protocols_for_phases(
                    profile["phases"],
                    lambda phase: resolve_phase_port(phase, default_port, target_info),
                )
                probe_before = probe_ports_mixed(resolved_target, port_protocols)
                self._report.record_probe_before(probe_before)
                logger.info(
                    f"Campaign pre-probe: {probe_before['reachable_count']}/{probe_before['total_ports']} ports reachable"
                )

            campaign_results = []
            for i, phase in enumerate(profile["phases"], 1):
                if self._interrupted:
                    logger.warning("Campaign interrupted. Skipping remaining phases.")
                    self._record_skipped_phases(
                        profile, i - 1, default_port, target_info, target, effective_overrides
                    )
                    break

                attack_type = phase["attack"]
                phase_port = resolve_phase_port(phase, default_port, target_info)
                description = phase.get("description", "")

                logger.info(f"--- Phase {i}/{len(profile['phases'])}: {attack_type} ---")
                if description:
                    logger.info(f"Description: {description}")

                phase_options = self._build_phase_options(
                    phase,
                    profile_ipv6=profile_ipv6,
                    campaign_allowlist=campaign_allowlist,
                    target_info=target_info,
                    effective_overrides=effective_overrides,
                )

                self._current_attack_name = attack_type
                phases_before = len(self._report.phases)
                phase_duration, phase_threads = self._phase_timing(
                    phase, effective_overrides, strict=True
                )
                phase_options = validate_run_params(
                    duration=phase_duration,
                    threads=phase_threads,
                    port=phase_port,
                    options=phase_options,
                    attack_type=attack_type,
                )
                result = self.run(
                    attack_type=attack_type,
                    target=target,
                    port=phase_port,
                    duration=phase_duration,
                    threads=phase_threads,
                    engagement_id=engagement_id,
                    operator=operator or profile.get("operator"),
                    allowlist=campaign_allowlist,
                    allowlist_file=None,
                    require_allowlist=require_allowlist,
                    dry_run=False,
                    probe=False,
                    skip_report=True,
                    skip_scope_check=True,
                    skip_param_validation=True,
                    skip_compat_validation=phase_compat_validated,
                    cached_target_info=target_info,
                    prevalidated_scope=self._campaign_scope,
                    clear_allowlist_cache=False,
                    dashboard=dashboard,
                    **phase_options,
                )
                if len(self._report.phases) > phases_before:
                    self._report.set_last_phase_description(description)
                campaign_results.append(result)

                if self._interrupted or self._orphan_abort:
                    skip_start = i - 1 if result.get("skipped_before_start") else i
                    self._record_skipped_phases(
                        profile,
                        skip_start,
                        default_port,
                        target_info,
                        target,
                        effective_overrides,
                    )
                    break

                if not self._shutdown.is_set() and not self._orphan_abort:
                    self._interrupted = False

                if i < len(profile["phases"]):
                    cooldown = profile.get("cooldown", 5)
                    logger.info(f"Cooldown: {cooldown}s before next phase")
                    if self._shutdown.wait(cooldown):
                        logger.warning("Campaign interrupted during cooldown.")
                        self._record_skipped_phases(
                            profile, i, default_port, target_info, target, effective_overrides
                        )
                        break

            if probe_enabled:
                executed_phases = profile["phases"][: len(campaign_results)]
                port_protocols = probe_protocols_for_phases(
                    executed_phases,
                    lambda phase: resolve_phase_port(phase, default_port, target_info),
                )
                probe_after = probe_ports_mixed(resolved_target, port_protocols)
                self._report.record_probe_after(probe_after)
                if self._report.probe_before:
                    delta = compare_probe(
                        self._report.probe_before,
                        probe_after,
                        ports=list(port_protocols.keys()),
                    )
                    self._report.set_metadata(campaign_probe_delta=delta)
                    logger.info(f"Campaign post-probe status: {delta['status']}")

            final_status = "interrupted" if (self._interrupted or self._orphan_abort) else "completed"
            report_paths = self._export_report(status=final_status)

            if self._interrupted:
                log_event(
                    "campaign_interrupted",
                    engagement_id=engagement_id,
                    operator=operator,
                    details={"phases_completed": len(campaign_results)},
                )
                logger.info(f"Campaign interrupted after {len(campaign_results)} phase(s).")
                return {
                    "phases": campaign_results,
                    "reports": report_paths,
                    "status": "interrupted",
                    "summary": self._report.to_dict()["summary"],
                }

            log_event(
                "campaign_complete",
                engagement_id=engagement_id,
                operator=operator,
                details={"phases": len(campaign_results)},
            )

            logger.info(f"Campaign completed. Phases: {len(campaign_results)}")

            return {
                "phases": campaign_results,
                "reports": report_paths,
                "status": "completed",
                "summary": self._report.to_dict()["summary"],
            }
        except Exception:
            self._export_report(status="failed")
            raise
        finally:
            self._campaign_active = False
            self._campaign_scope = None
            AttackController._active_controller = None
            AttackController._reset_signal_handlers()
            self._signal_acquisitions = 0

    def stop(self) -> None:
        self._shutdown.set()
        self._stats_shutdown.set()
        self._interrupted = True
        if self._attack:
            self._attack.stop()

    @staticmethod
    def _attack_has_active_workers(attack: BaseAttack) -> bool:
        """Support BaseAttack and minimal third-party/test doubles safely."""
        checker = getattr(attack, "has_active_workers", None)
        if callable(checker):
            return bool(checker())

        worker_threads = getattr(attack, "_worker_threads", ())
        if any(thread.is_alive() for thread in worker_threads):
            return True
        async_thread = getattr(attack, "_async_thread", None)
        return async_thread is not None and async_thread.is_alive()

    def _report_stats(
        self,
        attack: BaseAttack,
        interval: float,
        attack_name: str,
        dashboard: bool = False,
    ) -> None:
        from utils.dashboard import LiveStatsDashboard

        started = False
        start_deadline = time.monotonic() + min(1.0, max(0.1, interval))
        system_sample_interval = max(interval * 3.0, 6.0)
        last_system_sample = 0.0
        cached_system: dict | None = None
        with LiveStatsDashboard(attack_name) if dashboard else _NullDashboard() as live:
            while not self._shutdown.is_set() and not self._stats_shutdown.is_set():
                running = attack.is_running() or self._attack_has_active_workers(attack)
                if not running:
                    if started or time.monotonic() >= start_deadline:
                        break
                    self._stats_shutdown.wait(0.05)
                    continue

                started = True
                stats = attack.get_stats()
                now = time.monotonic()
                if live.enabled or cached_system is None or (now - last_system_sample) >= system_sample_interval:
                    cached_system = get_system_stats()
                    last_system_sample = now
                system = cached_system
                if live.enabled:
                    live.update(stats, system)
                else:
                    self._log_stats(stats, system)
                if self._report:
                    self._report.add_timeline_sample(
                        stats, system, attack=attack_name
                    )
                self._stats_shutdown.wait(interval)

    def _log_stats(self, stats: dict, system: dict | None = None) -> None:
        if system is None:
            system = get_system_stats()
        logger.info(
            f"Packets: {stats.get('packets_sent', 0)} | "
            f"Rate: {stats.get('packets_per_second', 0)} pps | "
            f"Peak: {stats.get('peak_packets_per_second', 0)} pps | "
            f"Bandwidth: {stats.get('megabits_per_second', 0)} Mbps | "
            f"Errors: {stats.get('errors', 0)} | "
            f"CPU: {system.get('cpu_percent', 0)}% | "
            f"Mem: {system.get('memory_mb', 0)} MB"
        )

    def _register_signal_handlers(self) -> None:
        if threading.current_thread() is not threading.main_thread():
            logger.debug("Signal handlers are only registered from the main thread")
            return
        if self._campaign_active and AttackController._signal_handlers_registered:
            return
        if AttackController._signal_handler_depth == 0:

            def handler(signum, frame):
                logger.warning("Interrupt received. Stopping attack...")
                inst = AttackController._active_controller
                if inst is not None:
                    inst._interrupted = True
                    inst.stop()
                else:
                    previous = AttackController._previous_signal_handlers.get(signum)
                    if callable(previous):
                        previous(signum, frame)
                    elif signum == signal.SIGINT:
                        raise KeyboardInterrupt

            for sig in (signal.SIGINT, signal.SIGTERM):
                AttackController._previous_signal_handlers[sig] = signal.getsignal(sig)
                signal.signal(sig, handler)
            AttackController._signal_handlers_registered = True
            AttackController._signal_handler_depth = 1
        else:
            AttackController._signal_handler_depth += 1
        self._signal_acquisitions += 1


class _NullDashboard:
    enabled = False

    def __enter__(self) -> "_NullDashboard":
        return self

    def __exit__(self, *args: object) -> None:
        return None