Use scapy to generate ARP packet in multi-device test
Test: m connectivity_multi_devices_snippet && atest CtsConnectivityMultiDevicesTestCases Test: atest NetworkStaticLibHostPythonTests Change-Id: I30adfe7ca2fef83ed4ef2ea3e5bc2291b9016530
This commit is contained in:
parent
8e8bbcabde
commit
f0343a65d1
4 changed files with 37 additions and 174 deletions
|
|
@ -1,72 +0,0 @@
|
|||
# Copyright (C) 2024 The Android Open Source Project
|
||||
#
|
||||
# Licensed under the Apache License, Version 2.0 (the "License");
|
||||
# you may not use this file except in compliance with the License.
|
||||
# You may obtain a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing, software
|
||||
# distributed under the License is distributed on an "AS IS" BASIS,
|
||||
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
# See the License for the specific language governing permissions and
|
||||
# limitations under the License.
|
||||
|
||||
from mobly import asserts
|
||||
from mobly import base_test
|
||||
from net_tests_utils.host.python import packet_utils
|
||||
|
||||
class TestPacketUtils(base_test.BaseTestClass):
|
||||
def test_unicast_arp_request(self):
|
||||
# Using scapy to generate unicast arp request packet:
|
||||
# eth = Ether(src="00:01:02:03:04:05", dst="01:02:03:04:05:06")
|
||||
# arp = ARP(op=1, pdst="192.168.1.1", hwsrc="00:01:02:03:04:05", psrc="192.168.1.2")
|
||||
# pkt = eth/arp
|
||||
expect_arp_request = """
|
||||
01020304050600010203040508060001080006040001000102030405c0a80102000000000000c0a80101
|
||||
""".upper().replace(" ", "").replace("\n", "")
|
||||
arp_request = packet_utils.construct_arp_packet(
|
||||
src_mac="00:01:02:03:04:05",
|
||||
dst_mac="01:02:03:04:05:06",
|
||||
src_ip="192.168.1.2",
|
||||
dst_ip="192.168.1.1",
|
||||
op=packet_utils.ARP_REQUEST_OP
|
||||
)
|
||||
asserts.assert_equal(expect_arp_request, arp_request)
|
||||
|
||||
def test_broadcast_arp_request(self):
|
||||
# Using scapy to generate unicast arp request packet:
|
||||
# eth = Ether(src="00:01:02:03:04:05", dst="FF:FF:FF:FF:FF:FF")
|
||||
# arp = ARP(op=1, pdst="192.168.1.1", hwsrc="00:01:02:03:04:05", psrc="192.168.1.2")
|
||||
# pkt = eth/arp
|
||||
expect_arp_request = """
|
||||
ffffffffffff00010203040508060001080006040001000102030405c0a80102000000000000c0a80101
|
||||
""".upper().replace(" ", "").replace("\n", "")
|
||||
arp_request = packet_utils.construct_arp_packet(
|
||||
src_mac="00:01:02:03:04:05",
|
||||
dst_mac=packet_utils.ETHER_BROADCAST_MAC_ADDRESS,
|
||||
src_ip="192.168.1.2",
|
||||
dst_ip="192.168.1.1",
|
||||
op=packet_utils.ARP_REQUEST_OP
|
||||
)
|
||||
asserts.assert_equal(expect_arp_request, arp_request)
|
||||
|
||||
def test_arp_reply(self):
|
||||
# Using scapy to generate unicast arp request packet:
|
||||
# eth = Ether(src="01:02:03:04:05:06", dst="00:01:02:03:04:05")
|
||||
# arp = ARP(op=2, pdst="192.168.1.2", \
|
||||
# hwsrc="01:02:03:04:05:06", \
|
||||
# psrc="192.168.1.1", \
|
||||
# hwdst="00:01:02:03:04:05")
|
||||
# pkt = eth/arp
|
||||
expect_arp_reply = """
|
||||
00010203040501020304050608060001080006040002010203040506c0a80101000102030405c0a80102
|
||||
""".upper().replace(" ", "").replace("\n", "")
|
||||
arp_reply = packet_utils.construct_arp_packet(
|
||||
src_mac="01:02:03:04:05:06",
|
||||
dst_mac="00:01:02:03:04:05",
|
||||
src_ip="192.168.1.1",
|
||||
dst_ip="192.168.1.2",
|
||||
op=packet_utils.ARP_REPLY_OP
|
||||
)
|
||||
asserts.assert_equal(expect_arp_reply, arp_reply)
|
||||
|
|
@ -18,7 +18,6 @@ import sys
|
|||
from host.python.adb_utils_test import TestAdbUtils
|
||||
from host.python.apf_utils_test import TestApfUtils
|
||||
from host.python.assert_utils_test import TestAssertUtils
|
||||
from host.python.packet_utils_test import TestPacketUtils
|
||||
from mobly import suite_runner
|
||||
|
||||
|
||||
|
|
@ -32,5 +31,5 @@ if __name__ == "__main__":
|
|||
sys.argv.pop(1)
|
||||
# TODO: make the tests can be executed without manually list classes.
|
||||
suite_runner.run_suite(
|
||||
[TestAssertUtils, TestAdbUtils, TestApfUtils, TestPacketUtils], sys.argv
|
||||
[TestAssertUtils, TestAdbUtils, TestApfUtils], sys.argv
|
||||
)
|
||||
|
|
|
|||
|
|
@ -1,70 +0,0 @@
|
|||
# Copyright (C) 2024 The Android Open Source Project
|
||||
#
|
||||
# Licensed under the Apache License, Version 2.0 (the "License");
|
||||
# you may not use this file except in compliance with the License.
|
||||
# You may obtain a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing, software
|
||||
# distributed under the License is distributed on an "AS IS" BASIS,
|
||||
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
# See the License for the specific language governing permissions and
|
||||
# limitations under the License.
|
||||
from ipaddress import IPv4Address
|
||||
from socket import inet_aton
|
||||
|
||||
ETHER_BROADCAST_MAC_ADDRESS = "FF:FF:FF:FF:FF:FF"
|
||||
ARP_REQUEST_OP = 1
|
||||
ARP_REPLY_OP = 2
|
||||
|
||||
"""
|
||||
This variable defines a template for constructing ARP packets in hexadecimal format.
|
||||
It's used to provide the common fields for ARP packet, and replaced needed fields when constructing
|
||||
"""
|
||||
ARP_TEMPLATE = (
|
||||
# Ether Header (14 bytes)
|
||||
"{dst_mac}" + # DA
|
||||
"{src_mac}" + # SA
|
||||
"0806" + # ARP
|
||||
# ARP Header (28 bytes)
|
||||
"0001" + # Hardware type (Ethernet)
|
||||
"0800" + # Protocol type (IPv4)
|
||||
"06" + # hardware address length
|
||||
"04" + # protocol address length
|
||||
"{opcode}" + # opcode
|
||||
"{sender_mac}" + # sender MAC
|
||||
"{sender_ip}" + # sender IP
|
||||
"{target_mac}" + # target MAC
|
||||
"{target_ip}" # target IP
|
||||
)
|
||||
|
||||
def construct_arp_packet(src_mac, dst_mac, src_ip, dst_ip, op) -> str:
|
||||
"""Constructs an ARP packet as a hexadecimal string.
|
||||
|
||||
This function creates an ARP packet by filling in the required fields
|
||||
in a predefined ARP packet template.
|
||||
|
||||
Args:
|
||||
src_mac: The MAC address of the sender. (e.g. "11:22:33:44:55:66")
|
||||
dst_mac: The MAC address of the recipient. (e.g. "aa:bb:cc:dd:ee:ff")
|
||||
src_ip: The IP address of the sender. (e.g. "1.1.1.1")
|
||||
dst_ip: The IP address of the target machine. (e.g. "2.2.2.2")
|
||||
op: The op code of the ARP packet, refer to ARP_*_OP
|
||||
|
||||
Returns:
|
||||
A string representing the ARP packet in hexadecimal format.
|
||||
"""
|
||||
# Replace the needed fields from packet template
|
||||
arp_pkt = ARP_TEMPLATE.format(
|
||||
dst_mac=dst_mac.replace(":",""),
|
||||
src_mac=src_mac.replace(":",""),
|
||||
opcode=str(op).rjust(4, "0"),
|
||||
sender_mac=src_mac.replace(":",""),
|
||||
sender_ip=inet_aton(src_ip).hex(),
|
||||
target_mac=("000000000000" if op == ARP_REQUEST_OP else dst_mac.replace(":", "")),
|
||||
target_ip=inet_aton(dst_ip).hex()
|
||||
)
|
||||
|
||||
# always convert to upper case hex string
|
||||
return arp_pkt.upper()
|
||||
|
|
@ -28,9 +28,9 @@ from scapy.layers.inet6 import (
|
|||
ICMPv6ND_NA,
|
||||
RouterAlert
|
||||
)
|
||||
from scapy.layers.l2 import Ether
|
||||
from scapy.layers.l2 import ARP, Ether
|
||||
from scapy.contrib.igmpv3 import IGMPv3, IGMPv3mq, IGMPv3mr, IGMPv3gr
|
||||
from net_tests_utils.host.python import apf_test_base, apf_utils, adb_utils, assert_utils, packet_utils
|
||||
from net_tests_utils.host.python import apf_test_base, apf_utils, adb_utils, assert_utils
|
||||
|
||||
APFV6_VERSION = 6000
|
||||
ARP_OFFLOAD_REPLY_LEN = 60
|
||||
|
|
@ -53,51 +53,57 @@ class ApfV6Test(apf_test_base.ApfTestBase):
|
|||
super().teardown_class()
|
||||
|
||||
def test_unicast_arp_request_offload(self):
|
||||
arp_request = packet_utils.construct_arp_packet(
|
||||
src_mac=self.server_mac_address,
|
||||
dst_mac=self.client_mac_address,
|
||||
src_ip=self.server_ipv4_addresses[0],
|
||||
dst_ip=self.client_ipv4_addresses[0],
|
||||
op=packet_utils.ARP_REQUEST_OP
|
||||
eth = Ether(src=self.server_mac_address, dst=self.client_mac_address)
|
||||
arp = ARP(
|
||||
op=1,
|
||||
psrc=self.server_ipv4_addresses[0],
|
||||
pdst=self.client_ipv4_addresses[0],
|
||||
hwsrc=self.server_mac_address
|
||||
)
|
||||
arp_request = bytes(eth/arp).hex()
|
||||
|
||||
arp_reply = packet_utils.construct_arp_packet(
|
||||
src_mac=self.client_mac_address,
|
||||
dst_mac=self.server_mac_address,
|
||||
src_ip=self.client_ipv4_addresses[0],
|
||||
dst_ip=self.server_ipv4_addresses[0],
|
||||
op=packet_utils.ARP_REPLY_OP
|
||||
eth = Ether(src=self.client_mac_address, dst=self.server_mac_address)
|
||||
arp = ARP(
|
||||
op=2,
|
||||
psrc=self.client_ipv4_addresses[0],
|
||||
pdst=self.server_ipv4_addresses[0],
|
||||
hwsrc=self.client_mac_address,
|
||||
hwdst=self.server_mac_address
|
||||
)
|
||||
expected_arp_reply = bytes(eth/arp).hex()
|
||||
|
||||
# Add zero padding up to 60 bytes, since APFv6 ARP offload always sent out 60 bytes reply
|
||||
arp_reply = arp_reply.ljust(ARP_OFFLOAD_REPLY_LEN * 2, "0")
|
||||
expected_arp_reply = expected_arp_reply.ljust(ARP_OFFLOAD_REPLY_LEN * 2, "0")
|
||||
|
||||
self.send_packet_and_expect_reply_received(
|
||||
arp_request, "DROPPED_ARP_REQUEST_REPLIED", arp_reply
|
||||
arp_request, "DROPPED_ARP_REQUEST_REPLIED", expected_arp_reply
|
||||
)
|
||||
|
||||
def test_broadcast_arp_request_offload(self):
|
||||
arp_request = packet_utils.construct_arp_packet(
|
||||
src_mac=self.server_mac_address,
|
||||
dst_mac=packet_utils.ETHER_BROADCAST_MAC_ADDRESS,
|
||||
src_ip=self.server_ipv4_addresses[0],
|
||||
dst_ip=self.client_ipv4_addresses[0],
|
||||
op=packet_utils.ARP_REQUEST_OP
|
||||
eth = Ether(src=self.server_mac_address, dst='ff:ff:ff:ff:ff:ff')
|
||||
arp = ARP(
|
||||
op=1,
|
||||
psrc=self.server_ipv4_addresses[0],
|
||||
pdst=self.client_ipv4_addresses[0],
|
||||
hwsrc=self.server_mac_address
|
||||
)
|
||||
arp_request = bytes(eth/arp).hex()
|
||||
|
||||
arp_reply = packet_utils.construct_arp_packet(
|
||||
src_mac=self.client_mac_address,
|
||||
dst_mac=self.server_mac_address,
|
||||
src_ip=self.client_ipv4_addresses[0],
|
||||
dst_ip=self.server_ipv4_addresses[0],
|
||||
op=packet_utils.ARP_REPLY_OP
|
||||
eth = Ether(src=self.client_mac_address, dst=self.server_mac_address)
|
||||
arp = ARP(
|
||||
op=2,
|
||||
psrc=self.client_ipv4_addresses[0],
|
||||
pdst=self.server_ipv4_addresses[0],
|
||||
hwsrc=self.client_mac_address,
|
||||
hwdst=self.server_mac_address
|
||||
)
|
||||
expected_arp_reply = bytes(eth/arp).hex()
|
||||
|
||||
# Add zero padding up to 60 bytes, since APFv6 ARP offload always sent out 60 bytes reply
|
||||
arp_reply = arp_reply.ljust(ARP_OFFLOAD_REPLY_LEN * 2, "0")
|
||||
expected_arp_reply = expected_arp_reply.ljust(ARP_OFFLOAD_REPLY_LEN * 2, "0")
|
||||
|
||||
self.send_packet_and_expect_reply_received(
|
||||
arp_request, "DROPPED_ARP_REQUEST_REPLIED", arp_reply
|
||||
arp_request, "DROPPED_ARP_REQUEST_REPLIED", expected_arp_reply
|
||||
)
|
||||
|
||||
def test_non_dad_ipv6_neighbor_solicitation_offload(self):
|
||||
|
|
|
|||
Loading…
Reference in a new issue