layer2 fun

receive.py

import argparse

from pypacker import psocket
from pypacker.pypacker import mac_bytes_to_str
from pypacker.layer12 import ethernet

parser = argparse.ArgumentParser()
parser.add_argument("-v", "--verbose", help="print whole packets", action="store_true")
parser.add_argument("-s", "--source", help="filter packet with source mac (all caps)")
parser.add_argument("-i", "--interface", help="interface to use", default="ens5")

args = parser.parse_args()

# default mode is MODE_LAYER_2
psock = psocket.SocketHndl(timeout=10, iface_name=args.interface)

for raw_bytes in psock:
    eth = ethernet.Ethernet(raw_bytes)
    src_mac = mac_bytes_to_str(eth[ethernet.Ethernet].src)
    dst_mac = mac_bytes_to_str(eth[ethernet.Ethernet].dst)
    if args.source and args.source != src_mac:
        continue

    print("from:", mac_bytes_to_str(eth[ethernet.Ethernet].src), "\tto: ", mac_bytes_to_str(eth[ethernet.Ethernet].dst))
    if args.verbose:
        print(eth)

psock.close()

requirements.txt

getmac
pypacker

send.py

import argparse
import time

from getmac import get_mac_address
from pypacker import psocket
from pypacker.pypacker import mac_bytes_to_str, mac_str_to_bytes
from pypacker.layer12 import ethernet

parser = argparse.ArgumentParser()
parser.add_argument("-v", "--verbose", help="print whole packets", action="store_true")
parser.add_argument("-d", "--dest", help="broadcast mac (default: src=dst)")
parser.add_argument("-i", "--interface", help="interface to use", default="ens5")
args = parser.parse_args()

# default mode is MODE_LAYER_2
psock = psocket.SocketHndl(timeout=10, iface_name=args.interface)

while True:
    src_mac = mac_str_to_bytes(get_mac_address(interface=args.interface).upper())
    dest_mac = src_mac
    if args.dest:
        dest_mac = mac_str_to_bytes(args.dest.upper())
    # TODO: Are we expecting MPLS_MCAST and related header???
    protocol = ethernet.ETH_TYPE_MPLS_MCAST # 0x8848
    protocol = b'\x8848'
    message = b'Hello World!'
    eth = ethernet.Ethernet(dest_mac + src_mac + protocol + message)
    if args.verbose:
        print(eth)
    time.sleep(1)

psock.close()