Merge pull request #221 from seleniumbase/proxy-with-auth

Ability to use Proxy servers that require auth

Author: mdmintz

.gitignore

@@ -62,6 +62,7 @@ report.html
 
 # Other
 selenium-server-standalone.jar
+proxy.zip
 verbose_hub_server.dat
 verbose_node_server.dat
 ip_of_grid_hub.dat

README.md

@@ -336,6 +336,12 @@ If you wish to use a proxy server for your browser tests (Chrome and Firefox onl
 pytest proxy_test.py --proxy=IP_ADDRESS:PORT
 ```
 
+If the proxy server that you wish to use requires authentication, you can do the following (Chrome only):
+
+```
+pytest proxy_test.py --proxy=USERNAME:PASSWORD@IP_ADDRESS:PORT
+```
+
 To make things easier, you can add your frequently-used proxies to PROXY_LIST in [proxy_list.py](https://github.com/seleniumbase/SeleniumBase/blob/master/seleniumbase/config/proxy_list.py), and then use ``--proxy=KEY_FROM_PROXY_LIST`` to use the IP_ADDRESS:PORT of that key.
 
 ```

examples/my_first_test.py

@@ -4,7 +4,7 @@
 class MyTestClass(BaseCase):
 
     def test_basic(self):
-        self.open('https://xkcd.com/353/')            # Navigate to the web page
+        self.open('https://xkcd.com/353/')           # Navigate to the web page
         self.assert_element('img[alt="Python"]')       # Assert element on page
         self.click('a[rel="license"]')                  # Click element on page
         self.assert_text('free to copy', 'div center')    # Assert text on page

examples/proxy_test.py

@@ -10,6 +10,6 @@ def test_proxy(self):
         self.open('https://ipinfo.io/%s' % ip_address)
         print("\n\nIP Address = %s\n" % ip_address)
         print("Displaying Host Info:")
-        print(self.get_text('table.table'))
+        print(self.get_text('ul.address-list'))
         print("\nThe browser will close automatically in 7 seconds...")
         time.sleep(7)

examples/tour_examples/xkcd_tour.py

@@ -17,6 +17,5 @@ def test_basic(self):
         self.add_tour_step("Click for the license here.", 'a[rel="license"]')
         self.add_tour_step("This selects a random comic.", 'a[href*="random"]')
         self.add_tour_step("Thanks for taking this tour!")
-        # self.export_tour()  # Use this to export the tour as a .js file
+        # self.export_tour()  # Use this to export the tour as [my_tour.js]
         self.play_tour()
-

help_docs/customizing_test_runs.md

@@ -36,6 +36,8 @@ pytest my_test_suite.py --server=IP_ADDRESS --port=4444
 
 pytest my_test_suite.py --proxy=IP_ADDRESS:PORT
 
+pytest my_test_suite.py --proxy=USERNAME:PASSWORD@IP_ADDRESS:PORT
+
 pytest test_fail.py -s --pdb --pdb-failures
 ```
 

help_docs/features_list.md

@@ -11,6 +11,7 @@
 * Uses a [global config file](https://github.com/seleniumbase/SeleniumBase/blob/master/seleniumbase/config/settings.py) for configuring SeleniumBase to your specific needs.
 * Backwards-compatible with [WebDriver](http://www.seleniumhq.org/projects/webdriver/). (Use ``self.driver`` anywhere.)
 * Can run tests through a proxy server. (Use ``--proxy=IP_ADDRESS:PORT``)
+* Can use an authenticated proxy server. (``--proxy=USERNAME:PASSWORD@IP_ADDRESS:PORT``)
 * Can handle Google Authenticator logins by using the [Python one-time password library](https://pyotp.readthedocs.io/en/latest/).
 * Includes a hybrid-automation solution called **[MasterQA](https://github.com/seleniumbase/SeleniumBase/blob/master/seleniumbase/masterqa/ReadMe.md)** to speed up manual testing.
 * Includes integrations with [MySQL](https://github.com/seleniumbase/SeleniumBase/blob/master/seleniumbase/core/testcase_manager.py), [Selenium Grid](https://github.com/seleniumbase/SeleniumBase/tree/master/seleniumbase/utilities/selenium_grid), [Google Cloud](https://github.com/seleniumbase/SeleniumBase/tree/master/integrations/google_cloud/ReadMe.md), [Amazon S3](https://github.com/seleniumbase/SeleniumBase/blob/master/seleniumbase/plugins/s3_logging_plugin.py), and [NodeJS](https://github.com/seleniumbase/SeleniumBase/tree/master/integrations/node_js).

requirements.txt

@@ -2,7 +2,7 @@ pip
 ipython
 setuptools
 selenium==3.14.1
-pytest>=3.8.2
+pytest>=3.9.1
 pytest-cov>=2.6.0
 pytest-html>=1.19.0
 pytest-rerunfailures>=4.2
@@ -14,8 +14,8 @@ pyotp>=2.2.6
 requests>=2.19.1
 unittest2>=1.1.0
 chardet>=3.0.4
-urllib3>=1.23
 boto>=2.49.0
+urllib3==1.23
 nose==1.3.7
 ipdb==0.11
 flake8==3.5.0

seleniumbase/config/proxy_list.py

@@ -15,13 +15,14 @@
 Example proxies in PROXY_LIST below are not guaranteed to be active or secure.
 If you don't already have a proxy server to connect to,
 you can try finding one from one of following sites:
+* https://www.proxynova.com/proxy-server-list/port-8080/
 * https://www.us-proxy.org/
 * https://hidemy.name/en/proxy-list/?country=US&type=h#list
 * http://proxyservers.pro/proxy/list/protocol/http/country/US/
 """
 
 PROXY_LIST = {
-    # "example1": "64.33.247.157:3128",  # (Example) - set your own proxy here
+    "example1": "104.248.122.30:8080",  # (Example) - set your own proxy here
     "proxy1": None,
     "proxy2": None,
     "proxy3": None,

seleniumbase/core/browser_launcher.py

@@ -1,16 +1,20 @@
 import os
 import re
 import sys
+import threading
+import time
 import warnings
 from selenium import webdriver
 from selenium.common.exceptions import WebDriverException
 from selenium.webdriver.common.desired_capabilities import DesiredCapabilities
 from seleniumbase.config import proxy_list
 from seleniumbase.core import download_helper
+from seleniumbase.core import proxy_helper
 from seleniumbase.fixtures import constants
 from seleniumbase.fixtures import page_utils
 from seleniumbase import drivers  # webdriver storage folder for SeleniumBase
 DRIVER_DIR = os.path.dirname(os.path.realpath(drivers.__file__))
+PROXY_ZIP_PATH = proxy_helper.PROXY_ZIP_PATH
 PLATFORM = sys.platform
 IS_WINDOWS = False
 LOCAL_CHROMEDRIVER = None
@@ -49,7 +53,31 @@ def make_driver_executable_if_not(driver_path):
         make_executable(driver_path)
 
 
-def _set_chrome_options(downloads_path, proxy_string):
+def _add_chrome_proxy_extension(
+        chrome_options, proxy_string, proxy_user, proxy_pass):
+    """ Implementation of https://stackoverflow.com/a/35293284 for
+        https://stackoverflow.com/questions/12848327/
+        (Run Selenium on a proxy server that requires authentication.)
+        The retry_on_exception is only needed for multithreaded runs
+        because proxy.zip is a common file shared between all tests
+        in a single run. """
+    if not "".join(sys.argv) == "-c":
+        # Single-threaded
+        proxy_helper.create_proxy_zip(proxy_string, proxy_user, proxy_pass)
+    else:
+        # Pytest multi-threaded test
+        lock = threading.Lock()
+        with lock:
+            if not os.path.exists(PROXY_ZIP_PATH):
+                proxy_helper.create_proxy_zip(
+                    proxy_string, proxy_user, proxy_pass)
+            time.sleep(0.3)
+    chrome_options.add_extension(PROXY_ZIP_PATH)
+    return chrome_options
+
+
+def _set_chrome_options(
+        downloads_path, proxy_string, proxy_auth, proxy_user, proxy_pass):
     chrome_options = webdriver.ChromeOptions()
     prefs = {
         "download.default_directory": downloads_path,
@@ -71,6 +99,10 @@ def _set_chrome_options(downloads_path, proxy_string):
     chrome_options.add_argument("--disable-translate")
     chrome_options.add_argument("--disable-web-security")
     if proxy_string:
+        if proxy_auth:
+            chrome_options = _add_chrome_proxy_extension(
+                chrome_options, proxy_string, proxy_user, proxy_pass)
+            chrome_options.add_extension(DRIVER_DIR + "/proxy.zip")
         chrome_options.add_argument('--proxy-server=%s' % proxy_string)
     if "win32" in sys.platform or "win64" in sys.platform:
         chrome_options.add_argument("--log-level=3")
@@ -155,22 +187,55 @@ def validate_proxy_string(proxy_string):
 
 def get_driver(browser_name, headless=False, use_grid=False,
                servername='localhost', port=4444, proxy_string=None):
+    proxy_auth = False
+    proxy_user = None
+    proxy_pass = None
     if proxy_string:
+        username_and_password = None
+        if "@" in proxy_string:
+            # Format => username:password@hostname:port
+            try:
+                username_and_password = proxy_string.split('@')[0]
+                proxy_string = proxy_string.split('@')[1]
+                proxy_user = username_and_password.split(':')[0]
+                proxy_pass = username_and_password.split(':')[1]
+            except Exception:
+                raise Exception(
+                    'The format for using a proxy server with authentication '
+                    'is: "username:password@hostname:port". If using a proxy '
+                    'server without auth, the format is: "hostname:port".')
+            if browser_name != constants.Browser.GOOGLE_CHROME:
+                raise Exception(
+                    "Chrome is required when using a proxy server that has "
+                    "authentication! (If using a proxy server without auth, "
+                    "either Chrome or Firefox may be used.)")
         proxy_string = validate_proxy_string(proxy_string)
+        if proxy_string and proxy_user and proxy_pass:
+            if not os.path.exists(PROXY_ZIP_PATH):
+                proxy_helper.create_proxy_zip(
+                    proxy_string, proxy_user, proxy_pass)
+            proxy_auth = True
     if use_grid:
         return get_remote_driver(
-            browser_name, headless, servername, port, proxy_string)
+            browser_name, headless, servername, port, proxy_string, proxy_auth,
+            proxy_user, proxy_pass)
     else:
-        return get_local_driver(browser_name, headless, proxy_string)
+        return get_local_driver(
+            browser_name, headless, proxy_string, proxy_auth,
+            proxy_user, proxy_pass)
 
 
-def get_remote_driver(browser_name, headless, servername, port, proxy_string):
+def get_remote_driver(
+        browser_name, headless, servername, port, proxy_string, proxy_auth,
+        proxy_user, proxy_pass):
     downloads_path = download_helper.get_downloads_folder()
     download_helper.reset_downloads_folder()
     address = "http://%s:%s/wd/hub" % (servername, port)
 
     if browser_name == constants.Browser.GOOGLE_CHROME:
-        chrome_options = _set_chrome_options(downloads_path, proxy_string)
+        chrome_options = _set_chrome_options(
+            downloads_path, proxy_string, proxy_auth,
+            proxy_user, proxy_pass)
         if headless:
             chrome_options.add_argument("--headless")
             chrome_options.add_argument("--disable-gpu")
@@ -237,7 +302,9 @@ def get_remote_driver(browser_name, headless, servername, port, proxy_string):
                     webdriver.DesiredCapabilities.PHANTOMJS))
 
 
-def get_local_driver(browser_name, headless, proxy_string):
+def get_local_driver(
+        browser_name, headless, proxy_string, proxy_auth,
+        proxy_user, proxy_pass):
     '''
     Spins up a new web browser and returns the driver.
     Can also be used to spin up additional browsers for the same test.
@@ -326,7 +393,9 @@ def get_local_driver(browser_name, headless, proxy_string):
             return webdriver.PhantomJS()
     elif browser_name == constants.Browser.GOOGLE_CHROME:
         try:
-            chrome_options = _set_chrome_options(downloads_path, proxy_string)
+            chrome_options = _set_chrome_options(
+                downloads_path, proxy_string, proxy_auth,
+                proxy_user, proxy_pass)
             if headless:
                 chrome_options.add_argument("--headless")
                 chrome_options.add_argument("--disable-gpu")

seleniumbase/core/log_helper.py

@@ -33,7 +33,7 @@ def log_test_failure_data(test, test_logpath, driver, browser):
     if sys.version.startswith('3') and hasattr(test, '_outcome'):
         if test._outcome.errors:
             try:
-                exc_message = test._outcome.errors[0][1][1].msg
+                exc_message = test._outcome.errors[0][1][1]
                 traceback_address = test._outcome.errors[0][1][2]
                 traceback_list = traceback.format_list(
                     traceback.extract_tb(traceback_address)[1:])
@@ -42,7 +42,7 @@ def log_test_failure_data(test, test_logpath, driver, browser):
                 exc_message = "(Unknown Exception)"
                 traceback_message = "(Unknown Traceback)"
             data_to_save.append("Traceback: " + traceback_message)
-            data_to_save.append("Exception: " + exc_message)
+            data_to_save.append("Exception: " + str(exc_message))
     else:
         data_to_save.append("Traceback: " + ''.join(
             traceback.format_exception(sys.exc_info()[0],

seleniumbase/core/proxy_helper.py

@@ -0,0 +1,82 @@
+import os
+import threading
+import zipfile
+from seleniumbase import drivers
+DRIVER_DIR = os.path.dirname(os.path.realpath(drivers.__file__))
+PROXY_ZIP_PATH = "%s/%s" % (DRIVER_DIR, "proxy.zip")
+
+
+def create_proxy_zip(proxy_string, proxy_user, proxy_pass):
+    """ Implementation of https://stackoverflow.com/a/35293284 for
+        https://stackoverflow.com/questions/12848327/
+        (Run Selenium on a proxy server that requires authentication.)
+        Solution involves creating & adding a Chrome extension on the fly.
+        * CHROME-ONLY for now! *
+    """
+    proxy_host = proxy_string.split(':')[0]
+    proxy_port = proxy_string.split(':')[1]
+    proxy_zip = DRIVER_DIR + '/proxy.zip'
+    background_js = (
+        """var config = {\n"""
+        """    mode: "fixed_servers",\n"""
+        """    rules: {\n"""
+        """      singleProxy: {\n"""
+        """        scheme: "http",\n"""
+        """        host: "%s",\n"""
+        """        port: parseInt("%s")\n"""
+        """      },\n"""
+        """    }\n"""
+        """  };\n"""
+        """chrome.proxy.settings.set("""
+        """{value: config, scope: "regular"}, function() {"""
+        """});\n"""
+        """function callbackFn(details) {\n"""
+        """    return {\n"""
+        """        authCredentials: {\n"""
+        """            username: "%s",\n"""
+        """            password: "%s"\n"""
+        """        }\n"""
+        """    };\n"""
+        """}\n"""
+        """chrome.webRequest.onAuthRequired.addListener(\n"""
+        """        callbackFn,\n"""
+        """        {urls: ["<all_urls>"]},\n"""
+        """        ['blocking']\n"""
+        """);""" % (proxy_host, proxy_port, proxy_user, proxy_pass))
+    manifest_json = (
+        '''{\n'''
+        '''"version": "1.0.0",\n'''
+        '''"manifest_version": 2,\n'''
+        '''"name": "Chrome Proxy",\n'''
+        '''"permissions": [\n'''
+        '''    "proxy",\n'''
+        '''    "tabs",\n'''
+        '''    "unlimitedStorage",\n'''
+        '''    "storage",\n'''
+        '''    "<all_urls>",\n'''
+        '''    "webRequest",\n'''
+        '''    "webRequestBlocking"\n'''
+        '''],\n'''
+        '''"background": {\n'''
+        '''    "scripts": ["background.js"]\n'''
+        '''},\n'''
+        '''"minimum_chrome_version":"22.0.0"\n'''
+        '''}''')
+    lock = threading.RLock()  # Support multi-threaded test runs with Pytest
+    with lock:
+        zf = zipfile.ZipFile(proxy_zip, mode='w')
+        zf.writestr("background.js", background_js)
+        zf.writestr("manifest.json", manifest_json)
+        zf.close()
+
+
+def remove_proxy_zip_if_present():
+    """ Remove Chrome extension zip file used for proxy server authentication.
+        Used in the implementation of https://stackoverflow.com/a/35293284
+        for https://stackoverflow.com/questions/12848327/
+    """
+    try:
+        if os.path.exists(PROXY_ZIP_PATH):
+            os.remove(PROXY_ZIP_PATH)
+    except Exception:
+        pass

seleniumbase/plugins/pytest_plugin.py

@@ -3,6 +3,7 @@
 import optparse
 import pytest
 from seleniumbase.core import log_helper
+from seleniumbase.core import proxy_helper
 from seleniumbase.fixtures import constants
 
 
@@ -84,6 +85,7 @@ def pytest_addoption(parser):
                      default=None,
                      help="""Designates the proxy server:port to use.
                           Format: servername:port.  OR
+                                  username:password@servername:port  OR
                                   A dict key from proxy_list.PROXY_LIST
                           Default: None.""")
     parser.addoption('--headless', action="store_true",
@@ -138,11 +140,12 @@ def pytest_configure(config):
     if with_testing_base:
         log_path = config.getoption('log_path')
         log_helper.log_folder_setup(log_path)
+    proxy_helper.remove_proxy_zip_if_present()
 
 
 def pytest_unconfigure():
-    """ This runs after all tests have completed with pytest """
-    pass
+    """ This runs after all tests have completed with pytest. """
+    proxy_helper.remove_proxy_zip_if_present()
 
 
 def pytest_runtest_setup():