acts_tests/tests/google/net/UsbTetheringTest.py - platform/tools/test/connectivity - Git at Google

 from acts import asserts
 from acts import base_test
 from acts.test_decorators import test_tracker_info
 from acts_contrib.test_utils.net import net_test_utils as nutils
 from acts_contrib.test_utils.wifi import wifi_test_utils as wutils
 from scapy.all import get_if_raw_hwaddr
 from scapy.layers.dns import DNS, DNSQR
 from scapy.layers.inet import IP, ICMP, UDP, TCP, RandShort, sr1
 from scapy.layers.inet6 import IPv6, ICMPv6EchoRequest
 from scapy.config import conf
 from scapy.sendrecv import send, sendp, srp1
 from scapy.layers.l2 import Ether
 import time

 DUMSYS_CMD = "dumpsys connectivity tethering"
 UPSTREAM_WANTED_STRING = "Upstream wanted"
 CURRENT_UPSTREAM_STRING = "Current upstream interface"
 SSID = wutils.WifiEnums.SSID_KEY
 GOOGLE_DNS_IP_ADDRESS = "8.8.8.8"
 DEFAULT_DOMAIN_NAME = "www.google.com"
 DEFAULT_DOMAIN_NAME_IPV4 = "ipv4.google.com"
 DEFAULT_DOMAIN_NAME_IPV6 = "ipv6.google.com"


 class UsbTetheringTest(base_test.BaseTestClass):
   """Tests for USB tethering.

   Prerequisite:
   1. Android phone should connect to the desktop with USB cable
   2. DUT should be able to connect to cellular network and Wi-Fi network
   3. Set the CAP_NET_RAW capability before run the test.
      e.g., `sudo setcap cap_net_raw=eip /usr/local/bin/act.py`
   """

   def setup_class(self):
     self.dut = self.android_devices[0]
     self.USB_TETHERED = False

     nutils.verify_lte_data_and_tethering_supported(self.dut)
     nutils.set_cap_net_raw_capability()
     req_params = ("wifi_network",)
     self.unpack_userparams(req_params)
     # Enable USB tethering and get the USB network interface
     iflist_before = nutils.get_if_list()
     serial = self.dut.device_info['serial']
     nutils.start_usb_tethering(self.dut)
     self.dut.recreate_services(serial)
     self.iface = nutils.wait_for_new_iface(iflist_before)
     conf.route6.resync()
     self.log.info("USB tethering interface: %s." % self.iface)
     if not self.check_upstream_ready():
       raise asserts.fail("Upstream interface is not active.")

   def teardown_class(self):
     nutils.stop_usb_tethering(self.dut)
     self.USB_TETHERED = False
     wutils.reset_wifi(self.dut)

   def on_fail(self, test_name, begin_time):
     self.dut.take_bug_report(test_name, begin_time)

   @test_tracker_info(uuid="d4da7695-4342-4564-b7b0-0a30895f23eb")
   def test_icmp_connectivity(self):
     """Tests connectivity under ICMP.

     Steps:
     1. Enable USB tethering on Android devices
     2. Generate ICMP packet and send to target IP address
     3. Verify that the response contains an ICMP layer
     """
     icmp = IP(dst=GOOGLE_DNS_IP_ADDRESS)/ICMP()
     resp = sr1(icmp, timeout=2, iface=self.iface)
     resp_msg = resp.show(dump=True) if resp else "null"
     asserts.assert_true(
         resp and resp.haslayer(ICMP), "Failed to send ICMP: " + resp_msg)

   @test_tracker_info(uuid="0dc7d049-11bf-42f9-918a-263f4470a7e8")
   def test_icmpv6_connectivity(self):
     """Tests connectivity under ICMPv6.

     Steps:
     1. Enable USB tethering on Android devices
     2. Generate ICMPv6 echo request packet and send to target URL
     3. Verify that the response contains an IPv6 layer
     """
     icmpv6 = IPv6(dst=DEFAULT_DOMAIN_NAME_IPV6)/ICMPv6EchoRequest()
     resp = sr1(icmpv6, timeout=2, iface=self.iface)
     resp_msg = resp.show(dump=True) if resp else "null"
     asserts.assert_true(
         resp and resp.haslayer(IPv6), "Failed to send ICMPv6: " + resp_msg)

   @test_tracker_info(uuid="34aaffb8-8dd4-4a1f-a158-2732b8df5e59")
   def test_dns_query_connectivity(self):
     """Tests connectivity of DNS query.

     Steps:
     1. Enable USB tethering on Android devices
     2. Generate DNS query and send to target DNS server
     3. Verify that the response contains a DNS layer
     """
     dnsqr = IP(dst=GOOGLE_DNS_IP_ADDRESS) \
             /UDP(sport=RandShort(), dport=53) \
             /DNS(rd=1, qd=DNSQR(qname=DEFAULT_DOMAIN_NAME))
     resp = sr1(dnsqr, timeout=2, iface=self.iface)
     resp_msg = resp.show(dump=True) if resp else "null"
     asserts.assert_true(
         resp and resp.haslayer(DNS), "Failed to send DNS query: " + resp_msg)

   @test_tracker_info(uuid="b9bed0fa-3178-4456-92e0-736b3a8cc181")
   def test_tcp_connectivity(self):
     """Tests connectivity under TCP.

     Steps:
     1. Enable USB tethering on Android devices
     2. Generate TCP packet and send to target URL
     3. Verify that the response contains a TCP layer
     """
     tcp = IP(dst=DEFAULT_DOMAIN_NAME)/TCP(dport=[80, 443])
     resp = sr1(tcp, timeout=2, iface=self.iface)
     resp_msg = resp.show(dump=True) if resp else "null"
     asserts.assert_true(
         resp and resp.haslayer(TCP), "Failed to send TCP packet:" + resp_msg)

   @test_tracker_info(uuid="5e2f31f4-0b18-44be-a1ba-d82bf9050996")
   def test_tcp_ipv6_connectivity(self):
     """Tests connectivity under IPv6.

     Steps:
     1. Enable USB tethering on Android devices
     2. Generate IPv6 packet and send to target URL (e.g., ipv6.google.com)
     3. Verify that the response contains an IPv6 layer
     """
     tcp_ipv6 = IPv6(dst=DEFAULT_DOMAIN_NAME_IPV6)/TCP(dport=[80, 443])
     resp = sr1(tcp_ipv6, timeout=2, iface=self.iface)
     resp_msg = resp.show(dump=True) if resp else "null"
     asserts.assert_true(
         resp and resp.haslayer(IPv6), "Failed to send TCP packet over IPv6, resp: " + resp_msg)

   def _send_http_get(self, destination, interface):
     try:
         syn_ack = srp1(Ether() / IPv6(dst=destination) / TCP(dport=80, flags="S"), timeout=2, iface=interface)
         sendp(Ether() / IPv6(dst=destination) / TCP(dport=80, sport=syn_ack[TCP].dport,seq=syn_ack[TCP].ack, ack=syn_ack[TCP].seq + 1, flags='A'), iface=interface)
         http_get_str = "GET / HTTP/1.1\r\nHost:" + destination + "\r\nAccept-Encoding: gzip, deflate\r\n\r\n"
         req = Ether() / IPv6(dst=destination)/TCP(dport=80, sport=syn_ack[TCP].dport, seq=syn_ack[TCP].ack, ack=syn_ack[TCP].seq + 1, flags='A')/http_get_str
         return srp1(req, timeout=2, iface=interface, retry = 3, filter = "tcp port 80")
     except TypeError:
         self.log.warn("Packet is sent but no answer from the server %s." % destination)
         return None

   @test_tracker_info(uuid="96115afb-e0d3-40a8-8f04-b64cedc6588f")
   def test_http_connectivity(self):
     """Tests connectivity under HTTP.

     Steps:
     1. Enable USB tethering on Android devices
     2. Implement TCP 3-way handshake to simulate HTTP GET
     3. Verify that the 3-way handshake works and response contains a TCP layer
     """
     resp = None
     for _ in range(3):
       resp = self._send_http_get(DEFAULT_DOMAIN_NAME, self.iface)
       if resp:
         break
       time.sleep(3)
     resp_msg = resp.show(dump=True) if resp else "null"
     asserts.assert_true(
         resp and resp.haslayer(TCP), "Failed to send HTTP request, resp: " + resp_msg)

   @test_tracker_info(uuid="140a064b-1ab0-4a92-8bdb-e52dde03d5b8")
   def test_usb_tethering_over_wifi(self):
     """Tests connectivity over Wi-Fi.

     Steps:
     1. Connects to a Wi-Fi network
     2. Enable USB tethering
     3. Verifies Wi-Fi is preferred upstream over data connection
     """

     wutils.start_wifi_connection_scan_and_ensure_network_found(
         self.dut, self.wifi_network[SSID])
     wutils.wifi_connect(self.dut, self.wifi_network)
     wifi_network = self.dut.droid.connectivityGetActiveNetwork()
     self.log.info("wifi network %s" % wifi_network)

     self.USB_TETHERED = True
     self.real_hwaddr = get_if_raw_hwaddr(self.iface)

     output = self.dut.adb.shell(DUMSYS_CMD)
     for line in output.split("\n"):
       if UPSTREAM_WANTED_STRING in line:
         asserts.assert_true("true" in line, "Upstream interface is not active")
         self.log.info("Upstream interface is active")
       if CURRENT_UPSTREAM_STRING in line:
         asserts.assert_true("wlan" in line, "WiFi is not the upstream "
                             "interface")
         self.log.info("WiFi is the upstream interface")

   def check_upstream_ready(self, retry=3):
     """Check the upstream is activated

     Check the upstream is activated with retry
     """
     for i in range(0, retry):
       output = self.dut.adb.shell(DUMSYS_CMD)
       for line in output.split("\n"):
         if UPSTREAM_WANTED_STRING in line:
           if "true" in line:
             self.log.info("Upstream interface is active")
           elif i == retry:
             return False
     return True
	from acts import asserts
	from acts import base_test
	from acts.test_decorators import test_tracker_info
	from acts_contrib.test_utils.net import net_test_utils as nutils
	from acts_contrib.test_utils.wifi import wifi_test_utils as wutils
	from scapy.all import get_if_raw_hwaddr
	from scapy.layers.dns import DNS, DNSQR
	from scapy.layers.inet import IP, ICMP, UDP, TCP, RandShort, sr1
	from scapy.layers.inet6 import IPv6, ICMPv6EchoRequest
	from scapy.config import conf
	from scapy.sendrecv import send, sendp, srp1
	from scapy.layers.l2 import Ether
	import time

	DUMSYS_CMD = "dumpsys connectivity tethering"
	UPSTREAM_WANTED_STRING = "Upstream wanted"
	CURRENT_UPSTREAM_STRING = "Current upstream interface"
	SSID = wutils.WifiEnums.SSID_KEY
	GOOGLE_DNS_IP_ADDRESS = "8.8.8.8"
	DEFAULT_DOMAIN_NAME = "www.google.com"
	DEFAULT_DOMAIN_NAME_IPV4 = "ipv4.google.com"
	DEFAULT_DOMAIN_NAME_IPV6 = "ipv6.google.com"


	class UsbTetheringTest(base_test.BaseTestClass):
	"""Tests for USB tethering.

	Prerequisite:
	1. Android phone should connect to the desktop with USB cable
	2. DUT should be able to connect to cellular network and Wi-Fi network
	3. Set the CAP_NET_RAW capability before run the test.
	e.g., `sudo setcap cap_net_raw=eip /usr/local/bin/act.py`
	"""

	def setup_class(self):
	self.dut = self.android_devices[0]
	self.USB_TETHERED = False

	nutils.verify_lte_data_and_tethering_supported(self.dut)
	nutils.set_cap_net_raw_capability()
	req_params = ("wifi_network",)
	self.unpack_userparams(req_params)
	# Enable USB tethering and get the USB network interface
	iflist_before = nutils.get_if_list()
	serial = self.dut.device_info['serial']
	nutils.start_usb_tethering(self.dut)
	self.dut.recreate_services(serial)
	self.iface = nutils.wait_for_new_iface(iflist_before)
	conf.route6.resync()
	self.log.info("USB tethering interface: %s." % self.iface)
	if not self.check_upstream_ready():
	raise asserts.fail("Upstream interface is not active.")

	def teardown_class(self):
	nutils.stop_usb_tethering(self.dut)
	self.USB_TETHERED = False
	wutils.reset_wifi(self.dut)

	def on_fail(self, test_name, begin_time):
	self.dut.take_bug_report(test_name, begin_time)

	@test_tracker_info(uuid="d4da7695-4342-4564-b7b0-0a30895f23eb")
	def test_icmp_connectivity(self):
	"""Tests connectivity under ICMP.

	Steps:
	1. Enable USB tethering on Android devices
	2. Generate ICMP packet and send to target IP address
	3. Verify that the response contains an ICMP layer
	"""
	icmp = IP(dst=GOOGLE_DNS_IP_ADDRESS)/ICMP()
	resp = sr1(icmp, timeout=2, iface=self.iface)
	resp_msg = resp.show(dump=True) if resp else "null"
	asserts.assert_true(
	resp and resp.haslayer(ICMP), "Failed to send ICMP: " + resp_msg)

	@test_tracker_info(uuid="0dc7d049-11bf-42f9-918a-263f4470a7e8")
	def test_icmpv6_connectivity(self):
	"""Tests connectivity under ICMPv6.

	Steps:
	1. Enable USB tethering on Android devices
	2. Generate ICMPv6 echo request packet and send to target URL
	3. Verify that the response contains an IPv6 layer
	"""
	icmpv6 = IPv6(dst=DEFAULT_DOMAIN_NAME_IPV6)/ICMPv6EchoRequest()
	resp = sr1(icmpv6, timeout=2, iface=self.iface)
	resp_msg = resp.show(dump=True) if resp else "null"
	asserts.assert_true(
	resp and resp.haslayer(IPv6), "Failed to send ICMPv6: " + resp_msg)

	@test_tracker_info(uuid="34aaffb8-8dd4-4a1f-a158-2732b8df5e59")
	def test_dns_query_connectivity(self):
	"""Tests connectivity of DNS query.

	Steps:
	1. Enable USB tethering on Android devices
	2. Generate DNS query and send to target DNS server
	3. Verify that the response contains a DNS layer
	"""
	dnsqr = IP(dst=GOOGLE_DNS_IP_ADDRESS) \
	/UDP(sport=RandShort(), dport=53) \
	/DNS(rd=1, qd=DNSQR(qname=DEFAULT_DOMAIN_NAME))
	resp = sr1(dnsqr, timeout=2, iface=self.iface)
	resp_msg = resp.show(dump=True) if resp else "null"
	asserts.assert_true(
	resp and resp.haslayer(DNS), "Failed to send DNS query: " + resp_msg)

	@test_tracker_info(uuid="b9bed0fa-3178-4456-92e0-736b3a8cc181")
	def test_tcp_connectivity(self):
	"""Tests connectivity under TCP.

	Steps:
	1. Enable USB tethering on Android devices
	2. Generate TCP packet and send to target URL
	3. Verify that the response contains a TCP layer
	"""
	tcp = IP(dst=DEFAULT_DOMAIN_NAME)/TCP(dport=[80, 443])
	resp = sr1(tcp, timeout=2, iface=self.iface)
	resp_msg = resp.show(dump=True) if resp else "null"
	asserts.assert_true(
	resp and resp.haslayer(TCP), "Failed to send TCP packet:" + resp_msg)

	@test_tracker_info(uuid="5e2f31f4-0b18-44be-a1ba-d82bf9050996")
	def test_tcp_ipv6_connectivity(self):
	"""Tests connectivity under IPv6.

	Steps:
	1. Enable USB tethering on Android devices
	2. Generate IPv6 packet and send to target URL (e.g., ipv6.google.com)
	3. Verify that the response contains an IPv6 layer
	"""
	tcp_ipv6 = IPv6(dst=DEFAULT_DOMAIN_NAME_IPV6)/TCP(dport=[80, 443])
	resp = sr1(tcp_ipv6, timeout=2, iface=self.iface)
	resp_msg = resp.show(dump=True) if resp else "null"
	asserts.assert_true(
	resp and resp.haslayer(IPv6), "Failed to send TCP packet over IPv6, resp: " + resp_msg)

	def _send_http_get(self, destination, interface):
	try:
	syn_ack = srp1(Ether() / IPv6(dst=destination) / TCP(dport=80, flags="S"), timeout=2, iface=interface)
	sendp(Ether() / IPv6(dst=destination) / TCP(dport=80, sport=syn_ack[TCP].dport,seq=syn_ack[TCP].ack, ack=syn_ack[TCP].seq + 1, flags='A'), iface=interface)
	http_get_str = "GET / HTTP/1.1\r\nHost:" + destination + "\r\nAccept-Encoding: gzip, deflate\r\n\r\n"
	req = Ether() / IPv6(dst=destination)/TCP(dport=80, sport=syn_ack[TCP].dport, seq=syn_ack[TCP].ack, ack=syn_ack[TCP].seq + 1, flags='A')/http_get_str
	return srp1(req, timeout=2, iface=interface, retry = 3, filter = "tcp port 80")
	except TypeError:
	self.log.warn("Packet is sent but no answer from the server %s." % destination)
	return None

	@test_tracker_info(uuid="96115afb-e0d3-40a8-8f04-b64cedc6588f")
	def test_http_connectivity(self):
	"""Tests connectivity under HTTP.

	Steps:
	1. Enable USB tethering on Android devices
	2. Implement TCP 3-way handshake to simulate HTTP GET
	3. Verify that the 3-way handshake works and response contains a TCP layer
	"""
	resp = None
	for _ in range(3):
	resp = self._send_http_get(DEFAULT_DOMAIN_NAME, self.iface)
	if resp:
	break
	time.sleep(3)
	resp_msg = resp.show(dump=True) if resp else "null"
	asserts.assert_true(
	resp and resp.haslayer(TCP), "Failed to send HTTP request, resp: " + resp_msg)

	@test_tracker_info(uuid="140a064b-1ab0-4a92-8bdb-e52dde03d5b8")
	def test_usb_tethering_over_wifi(self):
	"""Tests connectivity over Wi-Fi.

	Steps:
	1. Connects to a Wi-Fi network
	2. Enable USB tethering
	3. Verifies Wi-Fi is preferred upstream over data connection
	"""

	wutils.start_wifi_connection_scan_and_ensure_network_found(
	self.dut, self.wifi_network[SSID])
	wutils.wifi_connect(self.dut, self.wifi_network)
	wifi_network = self.dut.droid.connectivityGetActiveNetwork()
	self.log.info("wifi network %s" % wifi_network)

	self.USB_TETHERED = True
	self.real_hwaddr = get_if_raw_hwaddr(self.iface)

	output = self.dut.adb.shell(DUMSYS_CMD)
	for line in output.split("\n"):
	if UPSTREAM_WANTED_STRING in line:
	asserts.assert_true("true" in line, "Upstream interface is not active")
	self.log.info("Upstream interface is active")
	if CURRENT_UPSTREAM_STRING in line:
	asserts.assert_true("wlan" in line, "WiFi is not the upstream "
	"interface")
	self.log.info("WiFi is the upstream interface")

	def check_upstream_ready(self, retry=3):
	"""Check the upstream is activated

	Check the upstream is activated with retry
	"""
	for i in range(0, retry):
	output = self.dut.adb.shell(DUMSYS_CMD)
	for line in output.split("\n"):
	if UPSTREAM_WANTED_STRING in line:
	if "true" in line:
	self.log.info("Upstream interface is active")
	elif i == retry:
	return False
	return True