diff --git a/sim/net_core/parser/parser_sanity.py b/sim/net_core/parser/parser_sanity.py index bff250e..c2526dd 100644 --- a/sim/net_core/parser/parser_sanity.py +++ b/sim/net_core/parser/parser_sanity.py @@ -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) diff --git a/src/net_core/parser.sv b/src/net_core/parser.sv index 15ba2b7..a975e19 100644 --- a/src/net_core/parser.sv +++ b/src/net_core/parser.sv @@ -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