bslathi19/wireguard_fpga
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix bug where parser data was not resetting at the start of packet

Browse Source
This commit is contained in:
Byron Lathi
2026-08-13 19:50:27 -07:00
parent 2f2727777b
commit 9f8654f03c
2 changed files with 69 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

98
sim/net_core/parser/parser_sanity.py
View File

@@ -11,16 +11,19 @@ from cocotbext.axi import AxiStreamBus, AxiStreamSource, AxiStreamSink
import socket
from scapy.volatile import RandMAC, RandIP, RandShort, RandInt, RandLong
from scapy.volatile import RandMAC, RandIP, RandShort, RandInt, RandLong, RandIP6
from scapy.layers.l2 import Ether
from scapy.layers.l2 import Ether, ARP
from scapy.layers.inet import IP, UDP, TCP
from scapy.layers.inet6 import IPv6
from scapy.contrib.wireguard import Wireguard, WireguardTransport
from scapy.utils import str2mac
from scapy.packet import Packet
import random
CLK_PERIOD = 20
@@ -70,39 +73,61 @@ class TB:
parsed_ethertype = self.dut.ether_type.value.to_unsigned()
self.assertEqual(actual_ethertype, parsed_ethertype, "ether_type")
for f, v in zip(["src", "dst"], ["ip_source", "ip_dest"]):
actual = real_packet[IP].getfieldval(f)
parsed = socket.inet_ntoa(self.dut[v].value.to_unsigned().to_bytes(4))
self.assertEqual(actual, parsed, v)
actual_ip_len = real_packet[IP].getfieldval("len")
parsed_ip_len = self.dut.ip_length.value.to_unsigned()
self.assertEqual(actual_ip_len, parsed_ip_len, "ip_length")
if actual_ethertype == 0x0800:
for f, v in zip(["src", "dst"], ["ip_source", "ip_dest"]):
actual = real_packet[IP].getfieldval(f)
parsed = socket.inet_ntoa(self.dut[v].value.to_unsigned().to_bytes(4))
self.assertEqual(actual, parsed, v)
actual_ip_proto = real_packet[IP].getfieldval("proto")
parsed_ip_proto = self.dut.ip_proto.value.to_unsigned()
self.assertEqual(actual_ip_proto, parsed_ip_proto, "ip_proto")
actual_ip_len = real_packet[IP].getfieldval("len")
parsed_ip_len = self.dut.ip_length.value.to_unsigned()
self.assertEqual(actual_ip_len, parsed_ip_len, "ip_length")
for f, v in zip(["sport", "dport"], ["udp_source_port", "udp_dest_port"]):
actual = real_packet[IP].getfieldval(f)
parsed = self.dut[v].value.to_unsigned()
self.assertEqual(actual, parsed, v)
actual_ip_proto = real_packet[IP].getfieldval("proto")
parsed_ip_proto = self.dut.ip_proto.value.to_unsigned()
self.assertEqual(actual_ip_proto, parsed_ip_proto, "ip_proto")
else:
actual_ip_proto = 0
self.assertEqual(self.dut["ip_source"].value.to_unsigned(), 0, "ip_source")
self.assertEqual(self.dut["ip_dest"].value.to_unsigned(), 0, "ip_dest")
self.assertEqual(self.dut["ip_proto"].value.to_unsigned(), 0, "ip_proto")
actual_udp_len = real_packet[UDP].getfieldval("len")
parsed_udp_len = self.dut.udp_length.value.to_unsigned()
self.assertEqual(actual_udp_len, parsed_udp_len, "udp_length")
actual_wg_type = real_packet[Wireguard].getfieldval("message_type")
parsed_wg_type = self.dut.wg_type.value.to_unsigned()
self.assertEqual(actual_wg_type, parsed_wg_type, "wg_type")
if actual_ip_proto == 0x11:
for f, v in zip(["sport", "dport"], ["udp_source_port", "udp_dest_port"]):
actual = real_packet[UDP].getfieldval(f)
parsed = self.dut[v].value.to_unsigned()
self.assertEqual(actual, parsed, v)
actual_wg_receiver_index = real_packet[Wireguard].getfieldval("receiver_index")
parsed_wg_receiver_index = self.dut.wg_receiver_index.value.to_unsigned()
self.assertEqual(actual_wg_receiver_index, parsed_wg_receiver_index, "wg_receiver_index")
actual_udp_len = real_packet[UDP].getfieldval("len")
parsed_udp_len = self.dut.udp_length.value.to_unsigned()
self.assertEqual(actual_udp_len, parsed_udp_len, "udp_length")
actual_wg_counter = real_packet[Wireguard].getfieldval("counter")
parsed_wg_counter = self.dut.wg_counter.value.to_unsigned()
self.assertEqual(actual_wg_counter, parsed_wg_counter, "wg_counter")
# needed for the conditionals
actual_dest_port = real_packet[UDP].getfieldval("dport")
else:
actual_dest_port = 0
self.assertEqual(self.dut["udp_source_port"].value.to_unsigned(), 0, "udp_source_port")
self.assertEqual(self.dut["udp_dest_port"].value.to_unsigned(), 0,"udp_dest_port")
self.assertEqual(self.dut["udp_length"].value.to_unsigned(), 0,"udp_length")
if actual_dest_port == 51820:
actual_wg_type = real_packet[Wireguard].getfieldval("message_type")
parsed_wg_type = self.dut.wg_type.value.to_unsigned()
self.assertEqual(actual_wg_type, parsed_wg_type, "wg_type")
actual_wg_receiver_index = real_packet[Wireguard].getfieldval("receiver_index")
parsed_wg_receiver_index = self.dut.wg_receiver_index.value.to_unsigned()
self.assertEqual(actual_wg_receiver_index, parsed_wg_receiver_index, "wg_receiver_index")
actual_wg_counter = real_packet[Wireguard].getfieldval("counter")
parsed_wg_counter = self.dut.wg_counter.value.to_unsigned()
self.assertEqual(actual_wg_counter, parsed_wg_counter, "wg_counter")
else:
self.assertEqual(self.dut["wg_type"].value.to_unsigned(), 0, "wg_type")
self.assertEqual(self.dut["wg_receiver_index"].value.to_unsigned(), 0, "wg_receiver_index")
self.assertEqual(self.dut["wg_counter"].value.to_unsigned(), 0, "wg_counter")
@cocotb.test
async def test_sanity(dut):
@@ -111,8 +136,21 @@ async def test_sanity(dut):
await tb.cycle_reset()
for _ in range(16):
packet: Packet = Ether(src=RandMAC(), dst=RandMAC()) / IP(src=RandIP(), dst=RandIP()) / UDP(sport=RandShort(), dport=51820) / Wireguard() / WireguardTransport(receiver_index=RandInt(), counter=RandLong()) / b"Random encrypted data here, who knows what"
for _ in range(256):
packet_type = random.choice(["wireguard", "udp", "ip", "ip6", "arp"])
tb.log.info(packet_type)
if packet_type == "wireguard":
packet: Packet = Ether(src=RandMAC(), dst=RandMAC()) / IP(src=RandIP(), dst=RandIP()) / UDP(sport=RandShort(), dport=51820) / Wireguard() / WireguardTransport(receiver_index=RandInt(), counter=RandLong()) / b"Random encrypted data here, who knows what"
elif packet_type == "udp":
packet: Packet = Ether(src=RandMAC(), dst=RandMAC()) / IP(src=RandIP(), dst=RandIP()) / UDP(sport=RandShort(), dport=34197) / b"This is a factorio packet"
elif packet_type == "ip":
packet: Packet = Ether(src=RandMAC(), dst=RandMAC()) / IP(src=RandIP(), dst=RandIP()) / b"A Random IP packet. the protocol is not UDP though we know that."
elif packet_type == "ip6":
packet: Packet = Ether(src=RandMAC(), dst=RandMAC()) / IPv6(src=RandIP6(), dst=RandIP6()) / b"A Random IP packet. the protocol is not UDP though we know that."
elif packet_type == "arp":
packet: Packet = Ether(src=RandMAC(), dst=RandMAC()) / ARP()
packet_bytes = packet.build()
real_packet = Ether(packet_bytes)

1
src/net_core/parser.sv
View File

@@ -64,6 +64,7 @@ always_comb begin
if (s_axis.tvalid && s_axis.tready) begin
case (state)
ETH_1: begin
parser_data_next = '0;
parser_data_next.ether_dest[47:16] = s_axis.tdata;
state_next = ETH_2;
end