2023年11月5日 星期日

Python 開發筆記 - L2 Multicast / L3 Multicast 與 IGMP Snooping 測試

幫公司測試了一下 Professional Audiovisual (ProAV) 產品在不同 switch 上的情況,剛好有分別走 L2 Multicast 跟 L3 Multicast 溝通模式,而這些設備會使用 IGMP 通報 switch ,盡可能讓 Multicast 封包不要影響不必要的網路設備,避免俗稱的流量洪水事件。

由於過去經驗都是比較上層應用,這次就多多請教 chatGPT 惡補一下知識,其中 IGMP 的知識倒是被 Cloudflare 的教學文給救贖了一下:


IGMP 是用於 IPv4 的多點傳送通訊協定,是網際網路通訊協定的第四版。IPv6 依賴於多點傳送接聽程式探索 (MLD) 進行多點傳送。IPv6 網路使用 MLD 窺探而不是 IGMP 窺探。

就這句話,我反而在想 L2 Multicast 缺少 L3 (IP層) 的封包資訊,是不是讓 switch 做不到 IGMP snooping 功能?甚至舊款 switch 直接判定是未知 multicast 而進行永久性 broadcast 容錯設計?實測的結果是不同的 switch 行為是不一樣的。

整體上只需要簡單的 python 小程式,透過 scapy 套件,可非常輕鬆拼湊出想製作的 raw socket 來進行測試:

RX app: 

from scapy.all import *
from scapy.contrib.igmp import IGMP

multicastMac = "01:00:5e:00:02:01"
multicastGroupIP = '224.0.2.1'
nicName = '乙太網路'

# IGMPv2 Membership Report
igmpReportPacket = Ether(dst=multicastMac)/IP(dst=multicastGroupIP)/IGMP(type=0x16, mrcode=0, gaddr=multicastGroupIP)
sendp(igmpReportPacket, iface=nicName)

while True:
    multicastReceivePacket = sniff(iface=nicName, filter=f"ether dst {multicastMac}", count=1)
    if multicastReceivePacket:
        print("Received: ", multicastReceivePacket[0][Raw].load.decode("utf-8"))
        #print("Received: ", multicastReceivePacket[0][Raw].load)

TX app:

from scapy.all import *
from scapy.contrib.igmp import IGMP

multicastMac = "01:00:5e:00:02:01"
multicastGroupIP = '224.0.2.1'
nicName = '乙太網路2'

# IGMP Membership Report
igmpReportPacket = Ether(dst=multicastMac)/IP(dst=multicastGroupIP)/IGMP(type=0x16, mrcode=0, gaddr=multicastGroupIP)
sendp(igmpReportPacket, iface=nicName)

i = 0
while True:
    # Multicast
    if i % 3 == 0: 
        multicastTestMessagePacket = Ether(dst=multicastMac, type=0xffff)/"hello"
    elif i % 3 == 1:
        multicastTestMessagePacket = Ether(dst=multicastMac)/IP(dst=multicastGroupIP)/UDP(sport=12345, dport=12345)/"world"
    else:
        multicastTestMessagePacket = Ether(dst=multicastMac)/IP(src="192.168.1.2", dst=multicastGroupIP)/UDP(sport=12345, dport=12345)/"hello world"

    sendp(multicastTestMessagePacket, iface=nicName)

    # IGMPv2 Report
    if i % 600:
        sendp(igmpReportPacket, iface=nicName)
    i += 1

    time.sleep(1/10)

如此,弄一台筆電,讓他上頭有兩張 RJ45 網卡(可靠USB擴充),分別插上待側的 switch ,一邊當 RX 收訊息,另一邊當 TX 送訊息,接著調整 switch 設定後,觀察封包傳遞的狀況。可以的話,其實筆電可以設法弄個 3個網孔?接著另一網孔可以插在 switch 上,看看有沒有流量洪水(traffic floods)

收工