acts_tests/tests/google/net/UsbTetheringTest.py

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

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)
    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)
    asserts.assert_true(
        resp and resp.haslayer(ICMP),
        "Failed to send ICMP: " + resp.show(dump=True) if resp else "null")

  @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)
    asserts.assert_true(
        resp and resp.haslayer(IPv6),
        "Failed to send ICMPv6: " + resp.show(dump=True) if resp else "null")

  @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)
    asserts.assert_true(
        resp and resp.haslayer(DNS),
        "Failed to send DNS query: " + resp.show(dump=True) if resp else "null")

  @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)
    asserts.assert_true(
        resp and resp.haslayer(TCP),
        "Failed to send TCP packet:" + resp.show(dump=True) if resp else "null")

  @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)
    asserts.assert_true(
        resp and resp.haslayer(IPv6),
        "Failed to send TCP packet over IPv6, resp: " +
        resp.show(dump=True) if resp else "null")

  @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
    """
    syn_ack = sr1(IP(dst=DEFAULT_DOMAIN_NAME)
                  / TCP(dport=80, flags="S"), timeout=2, iface=self.iface)
    get_str = "GET / HTTP/1.1\r\nHost: " + DEFAULT_DOMAIN_NAME + "\r\n\r\n"
    req = IP(dst=DEFAULT_DOMAIN_NAME)/TCP(dport=80, sport=syn_ack[TCP].dport,
             seq=syn_ack[TCP].ack, ack=syn_ack[TCP].seq + 1, flags="A")/get_str
    resp = sr1(req, timeout=2, iface=self.iface)
    asserts.assert_true(
        resp and resp.haslayer(TCP),
        "Failed to send HTTP request, resp: " +
        resp.show(dump=True) if resp else "null")

  @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