Merge "LegacyVpnTest.py: Script for VPN testing"

2 files changed, 247 insertions, 1 deletions

diff --git a/acts/framework/acts/test_utils/net/connectivity_const.py b/acts/framework/acts/test_utils/net/connectivity_const.py
index 99ef56f76e..3888e914eb 100644
--- a/acts/framework/acts/test_utils/net/connectivity_const.py
+++ b/acts/framework/acts/test_utils/net/connectivity_const.py
@@ -14,6 +14,8 @@
 #   See the License for the specific language governing permissions and
 #   limitations under the License.
 
+import enum
+
 ######################################################
 # ConnectivityManager.NetworkCallback events
 ######################################################
@@ -36,4 +38,63 @@ NETWORK_CB_KEY_ID = "id"
 NETWORK_CB_KEY_EVENT = "networkCallbackEvent"
 NETWORK_CB_KEY_MAX_MS_TO_LIVE = "maxMsToLive"
 NETWORK_CB_KEY_RSSI = "rssi"
-NETWORK_CB_KEY_INTERFACE_NAME = "interfaceName"
\ No newline at end of file
+NETWORK_CB_KEY_INTERFACE_NAME = "interfaceName"
+
+# Constants for VPN connection status
+VPN_STATE_DISCONNECTED = 0
+VPN_STATE_INITIALIZING = 1
+VPN_STATE_CONNECTING = 2
+VPN_STATE_CONNECTED = 3
+VPN_STATE_TIMEOUT = 4
+VPN_STATE_FAILED = 5
+# TODO gmoturu: determine the exact timeout value
+# This is a random value as of now
+VPN_TIMEOUT = 15
+
+# Constants for VpnProfile
+class VpnProfile(object):
+    """ This class contains all the possible
+        parameters required for VPN connection
+    """
+    NAME = "name"
+    TYPE = "type"
+    SERVER = "server"
+    USER = "username"
+    PWD = "password"
+    DNS = "dnsServers"
+    SEARCH_DOMAINS = "searchDomains"
+    ROUTES = "routes"
+    MPPE = "mppe"
+    L2TP_SECRET = "l2tpSecret"
+    IPSEC_ID = "ipsecIdentifier"
+    IPSEC_SECRET = "ipsecSecret"
+    IPSEC_USER_CERT = "ipsecUserCert"
+    IPSEC_CA_CERT = "ipsecCaCert"
+    IPSEC_SERVER_CERT = "ipsecServerCert"
+
+# Enums for VPN profile types
+class VpnProfileType(enum.Enum):
+    """ Integer constant for each type of VPN
+    """
+    PPTP = 0
+    L2TP_IPSEC_PSK = 1
+    L2TP_IPSEC_RSA = 2
+    IPSEC_XAUTH_PSK = 3
+    IPSEC_XAUTH_RSA = 4
+    IPSEC_HYBRID_RSA = 5
+
+# Constants for config file
+class VpnReqParams(object):
+    """ Config file parameters required for
+        VPN connection
+    """
+    wifi_network = "wifi_network"
+    vpn_server_addresses = "vpn_server_addresses"
+    vpn_verify_address = "vpn_verify_address"
+    vpn_username = "vpn_username"
+    vpn_password = "vpn_password"
+    psk_secret = "psk_secret"
+    client_pkcs_file_name = "client_pkcs_file_name"
+    cert_path_vpnserver = "cert_path_vpnserver"
+    cert_password = "cert_password"
+    pptp_mppe = "pptp_mppe"
diff --git a/acts/tests/google/net/LegacyVpnTest.py b/acts/tests/google/net/LegacyVpnTest.py
new file mode 100644
index 0000000000..a0d844f7f4
--- /dev/null
+++ b/acts/tests/google/net/LegacyVpnTest.py
@@ -0,0 +1,185 @@
+#
+#   Copyright 2016 - 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.
+
+import logging
+import os
+import pprint
+import time
+import urllib.request
+
+from acts import asserts
+from acts import base_test
+from acts import test_runner
+from acts.controllers import adb
+from acts.test_utils.wifi import wifi_test_utils
+from acts.test_utils.net import connectivity_const
+
+VPN_CONST = connectivity_const.VpnProfile
+VPN_TYPE = connectivity_const.VpnProfileType
+VPN_PARAMS = connectivity_const.VpnReqParams
+
+
+class LegacyVpnTest(base_test.BaseTestClass):
+    """ Tests for Legacy VPN in Android
+
+        Testbed requirement:
+            1. One Android device
+            2. A Wi-Fi network that can reach the VPN servers
+    """
+
+    def setup_class(self):
+        """ Setup wi-fi connection and unpack params
+        """
+        self.dut = self.android_devices[0]
+        required_params = dir(VPN_PARAMS)
+        required_params = [x for x in required_params if not x.startswith('__')]
+        self.unpack_userparams(required_params)
+        wifi_test_utils.wifi_test_device_init(self.dut)
+        wifi_test_utils.wifi_connect(self.dut, self.wifi_network)
+
+    def teardown_class(self):
+        """ Reset wifi to make sure VPN tears down cleanly
+        """
+        wifi_test_utils.reset_wifi(self.dut)
+
+    def download_load_certs(self,vpn_type_name):
+        """ Download the certificates from VPN server and push to sdcard of DUT
+
+            Args:
+                VPN type name
+
+            Returns:
+                Client cert file name on DUT's sdcard
+        """
+        url = "http://%s%s%s" % (self.vpn_server_addresses[vpn_type_name],
+                                 self.cert_path_vpnserver,
+                                 self.client_pkcs_file_name)
+        local_cert_name = "%s_%s" % (vpn_type_name,self.client_pkcs_file_name)
+        local_file_path = os.path.join(self.log_path, local_cert_name)
+        try:
+            ret = urllib.request.urlopen(url)
+            with open(file_path, "wb") as f:
+                f.write(ret.read())
+        except:
+            asserts.fail("Unable to download certificate from the server")
+        f.close()
+        self.dut.adb.push("%s sdcard/" % local_file_path)
+        return local_cert_name
+
+    def generate_legacy_vpn_profile(self,vpn_type):
+        """ Generate legacy VPN profile for a VPN
+
+            Args:
+                VpnProfileType
+        """
+        vpn_profile = {VPN_CONST.USER: self.vpn_username,
+                       VPN_CONST.PWD: self.vpn_password,
+                       VPN_CONST.TYPE: vpn_type.value,
+                       VPN_CONST.SERVER: self.vpn_server_addresses[vpn_type.name],}
+        vpn_profile[VPN_CONST.NAME] = "test_%s" % vpn_type.name
+        psk_set = set(["L2TP_IPSEC_PSK", "IPSEC_XAUTH_PSK"])
+        rsa_set = set(["L2TP_IPSEC_RSA", "IPSEC_XAUTH_RSA", "IPSEC_HYBRID_RSA"])
+        if vpn_type.name in psk_set:
+            vpn_profile[VPN_CONST.IPSEC_SECRET] = self.psk_secret
+        elif vpn_type.name in rsa_set:
+            cert_name = self.download_load_certs(vpn_type.name)
+            vpn_profile[VPN_CONST.IPSEC_USER_CERT] = "%s_%s" % (vpn_type.name,
+                                                      cert_name.split('.')[0])
+            vpn_profile[VPN_CONST.IPSEC_CA_CERT] = "%s_%s" % (vpn_type.name,
+                                                   cert_name.split('.')[0])
+            self.dut.droid.installCertificate(vpn_profile,cert_name,self.cert_password)
+        else:
+            vpn_profile[VPN_CONST.MPPE] = self.pptp_mppe
+        return vpn_profile
+
+    def verify_ping_to_vpn_ip(self, connected_vpn_info):
+        """ Verify if IP behind VPN server is pingable
+            Ping should pass, if VPN is connected
+            Ping should fail, if VPN is disconnected
+
+            Args:
+                connected_vpn_info which specifies the VPN connection status
+        """
+        try:
+            ping_result = self.dut.adb.shell("ping -c 3 -W 2 %s"
+                                             % self.vpn_verify_address)
+            if not connected_vpn_info and "100% packet loss" \
+                not in "%s" % ping_result:
+                  asserts.fail("VPN is disconnected.\
+                               Ping to the internal IP expected to fail")
+        except adb.AdbError as ping_error:
+            ping_error = "%s" % ping_error
+            if connected_vpn_info and "100% packet loss" in ping_error:
+                asserts.fail("Ping to the internal IP failed.\
+                             Expected to pass as VPN is connected")
+
+    def legacy_vpn_connection_test_logic(self, vpn_type):
+        """ Test logic for each legacy VPN connection
+
+            Steps:
+                1. Generate profile for the VPN type
+                2. Establish connection to the server
+                3. Verify that connection is established using LegacyVpnInfo
+                4. Verify the connection by pinging the IP behind VPN
+                5. Stop the VPN connection
+                6. Check the connection status
+                7. Verify that ping to IP behind VPN fails
+
+            Args:
+                VpnProfileType (1 of the 6 types supported by Android)
+        """
+        vpn_profile = self.generate_legacy_vpn_profile(vpn_type)
+        logging.info("Connecting to: %s", vpn_profile)
+        self.dut.droid.vpnStartLegacyVpn(vpn_profile)
+        time.sleep(connectivity_const.VPN_TIMEOUT)
+        connected_vpn_info = self.dut.droid.vpnGetLegacyVpnInfo()
+        asserts.assert_equal(connected_vpn_info["state"],
+                             connectivity_const.VPN_STATE_CONNECTED,
+                             "Unable to establish VPN connection for %s"
+                             % vpn_profile)
+        self.verify_ping_to_vpn_ip(connected_vpn_info)
+        self.dut.droid.vpnStopLegacyVpn()
+        connected_vpn_info = self.dut.droid.vpnGetLegacyVpnInfo()
+        asserts.assert_true(not connected_vpn_info,
+                            "Unable to terminate VPN connection for %s"
+                            % vpn_profile)
+        self.verify_ping_to_vpn_ip(connected_vpn_info)
+
+    """ Test Cases """
+
+    def test_connection_to_legacy_vpn(self):
+        """ Verify VPN connection for all configurations.
+            Supported VPN configurations are
+            1.) PPTP            2.) L2TP IPSEC PSK
+            3.) IPSEC XAUTH PSK 4.) L2TP IPSEC RSA
+            5.) IPSEC XAUTH RSA 6.) IPSec Hybrid RSA
+
+            Steps:
+                1. Call legacy_vpn_connection_test_logic() for each VPN which
+                tests the connection to the corresponding server
+
+            Return:
+                Pass: if all VPNs pass
+                Fail: if any one VPN fails
+        """
+        def gen_name(vpn_type):
+            return "test_legacy_vpn_%s" % vpn_type.name
+
+        result = self.run_generated_testcases(self.legacy_vpn_connection_test_logic,
+                                              VPN_TYPE,
+                                              name_func=gen_name,)
+        msg = ("The following configs failed vpn connection %s"
+               % pprint.pformat(result))
+        asserts.assert_equal(len(result), 0, msg)