from ryu.
base import app_manager
from [Link] import ofp_event
from [Link] import CONFIG_DISPATCHER, MAIN_DISPATCHER, set_ev_cls
from [Link] import ofproto_v1_3
from [Link] import packet, ethernet, arp, ipv4, udp
from dnslib import DNSRecord
import time
class SpoofDetector(app_manager.RyuApp):
OFP_VERSIONS = [ofproto_v1_3.OFP_VERSION]
def __init__(self, *args, **kwargs):
super(SpoofDetector, self).__init__(*args, **kwargs)
self.arp_table = {} # {ip: mac}
self.detected_attackers = {} # {attacker_mac: (ip, time)}
self.trusted_dns_servers = {"[Link]", "[Link]", "[Link]"} # Trusted DNS
servers
self.dns_queries = {} # {(src_ip, query_id): (query_name, time)}
self.gateway_ip = "[Link]" # ← غيرها حسب شبكتك
self.gateway_mac = None
@set_ev_cls(ofp_event.EventOFPSwitchFeatures, CONFIG_DISPATCHER)
def switch_features_handler(self, ev):
datapath = [Link]
ofproto = [Link]
parser = datapath.ofproto_parser
match_arp = [Link](eth_type=0x0806)
actions_arp = [
[Link](ofproto.OFPP_CONTROLLER,
ofproto.OFPCML_NO_BUFFER),
[Link](ofproto.OFPP_NORMAL)
]
self.add_flow(datapath, 10, match_arp, actions_arp)
match_dns = [Link](eth_type=0x0800, ip_proto=17, udp_dst=53)
actions_dns = [
[Link](ofproto.OFPP_CONTROLLER,
ofproto.OFPCML_NO_BUFFER),
[Link](ofproto.OFPP_NORMAL)
]
self.add_flow(datapath, 30, match_dns, actions_dns)
match_dns_resp = [Link](eth_type=0x0800, ip_proto=17, udp_src=53)
actions_dns_resp = [
[Link](ofproto.OFPP_CONTROLLER,
ofproto.OFPCML_NO_BUFFER),
[Link](ofproto.OFPP_NORMAL)
]
self.add_flow(datapath, 30, match_dns_resp, actions_dns_resp)
match_default = [Link]()
actions_default = [[Link](ofproto.OFPP_NORMAL)]
self.add_flow(datapath, 0, match_default, actions_default)
def add_flow(self, datapath, priority, match, actions):
ofproto = [Link]
parser = datapath.ofproto_parser
inst = [[Link](ofproto.OFPIT_APPLY_ACTIONS, actions)]
mod = [Link](datapath=datapath, priority=priority, match=match,
instructions=inst)
datapath.send_msg(mod)
@set_ev_cls(ofp_event.EventOFPPacketIn, MAIN_DISPATCHER)
def packet_in_handler(self, ev):
try:
msg = [Link]
pkt = [Link]([Link])
eth = pkt.get_protocol([Link])
arp_pkt = pkt.get_protocol([Link])
if arp_pkt:
self.handle_arp(arp_pkt, eth)
ip_pkt = pkt.get_protocol(ipv4.ipv4)
if ip_pkt and ip_pkt.proto == 17:
udp_pkt = pkt.get_protocol([Link])
if udp_pkt and (udp_pkt.dst_port == 53 or udp_pkt.src_port == 53):
self.handle_dns([Link], eth, ip_pkt, udp_pkt)
except Exception as e:
[Link](f"Error in packet_in_handler: {str(e)}")
def handle_arp(self, arp_pkt, eth):
ip = arp_pkt.src_ip
mac = [Link]
if ip == self.gateway_ip:
if self.gateway_mac is None:
self.gateway_mac = mac
[Link](f"✔️ Registered Gateway MAC: {mac}")
elif self.gateway_mac != mac:
[Link]("\033[91m⚠️ ARP Spoofing: Gateway
impersonation!\033[0m")
[Link](f"IP: {ip} | Real MAC: {self.gateway_mac} |
Fake MAC: {mac}")
self.detected_attackers[mac] = (ip, [Link]())
elif ip in self.arp_table:
if self.arp_table[ip] != mac:
[Link]("\033[91m⚠️ ARP Spoofing Detected!\033[0m")
[Link](f"IP: {ip} | Old MAC: {self.arp_table[ip]} |
New MAC: {mac}")
self.detected_attackers[mac] = (ip, [Link]())
else:
self.arp_table[ip] = mac
def handle_dns(self, raw_data, eth, ip_pkt, udp_pkt):
try:
eth_len = 14
ip_len = ip_pkt.header_length * 4
udp_len = 8
header_len = eth_len + ip_len + udp_len
if len(raw_data) < header_len + 12:
return
dns_data = raw_data[header_len:]
dns_record = [Link](dns_data)
if dns_record.[Link] == 0:
self.handle_dns_query(ip_pkt.src, [Link], dns_record)
else:
self.handle_dns_response(ip_pkt.src, ip_pkt.dst, [Link],
dns_record)
except Exception as e:
[Link](f"Error in handle_dns: {str(e)}")
def handle_dns_query(self, src_ip, src_mac, dns_record):
try:
for question in dns_record.questions:
query_key = (src_ip, dns_record.[Link])
self.dns_queries[query_key] = (str([Link]).lower(),
[Link]())
except Exception as e:
[Link](f"Error in handle_dns_query: {str(e)}")
def clean_old_queries(self):
now = [Link]()
old = [k for k, v in self.dns_queries.items() if now - v[1] > 10]
for k in old:
del self.dns_queries[k]
def handle_dns_response(self, src_ip, dst_ip, src_mac, dns_record):
try:
self.clean_old_queries()
# إذا المهاجم ينتحلIP الـGateway لكنMAC مختلف
if src_ip == self.gateway_ip and src_mac != self.gateway_mac:
self.log_dns_spoofing(dst_ip, src_ip, src_mac, f"Gateway MAC:
{self.gateway_mac}", f"Fake MAC: {src_mac}")
return
if src_ip not in self.trusted_dns_servers:
query_key = (dst_ip, dns_record.[Link])
if query_key not in self.dns_queries:
if len(dns_record.rr) > 0:
self.log_dns_spoofing(dst_ip, src_ip, src_mac,
"unsolicited", str(dns_record.rr[0].rname))
return
expected_qname, query_time = self.dns_queries[query_key]
for rr in dns_record.rr:
if str([Link]).lower() != expected_qname:
self.log_dns_spoofing(dst_ip, src_ip, src_mac,
expected_qname, str([Link]))
return
response_time = [Link]() - query_time
if response_time < 0.01:
[Link]("\033[91m⚠️ Fast DNS response detected!\
033[0m")
[Link](f"{src_ip} ({src_mac}) responded in
{response_time:.4f}s")
del self.dns_queries[query_key]
except Exception as e:
[Link](f"Error in handle_dns_response: {str(e)}")
def log_dns_spoofing(self, victim_ip, attacker_ip, attacker_mac, expected,
received):
log = f"""
\033[91m⚠️ DNS SPOOFING DETECTED! ⚠️\033[0m
Time: {[Link]()}
Victim IP: {victim_ip}
Attacker IP: {attacker_ip}
Attacker MAC: {attacker_mac}
Expected: {expected}
Received: {received}
"""
[Link](log)
self.detected_attackers[attacker_mac] = (attacker_ip, [Link]())