Add http_proxy parameter to Client (#21)

Author: gcoxmoz

demo/app.py

@@ -114,6 +114,7 @@ def parse():
         host=config[config_section]['api_hostname'],
         redirect_uri=config[config_section]['redirect_uri'],
         duo_certs=config[config_section].get('duo_certs', None),
+        http_proxy=config[config_section].get('http_proxy', None),
     )
 except DuoException as e:
     print("*** Duo config error. Verify the values in duo.conf are correct ***")

demo/duo.conf

@@ -5,3 +5,5 @@ client_secret =
 api_hostname =
 redirect_uri = http://localhost:8080/duo-callback
 failmode = closed
+; Uncomment to use an HTTP proxy server
+; http_proxy = localhost:8081

duo_universal/client.py

@@ -113,7 +113,7 @@ def _create_jwt_args(self, endpoint):
         return jwt_args
 
     def __init__(self, client_id, client_secret, host,
-                 redirect_uri, duo_certs=DEFAULT_CA_CERT_PATH, use_duo_code_attribute=True):
+                 redirect_uri, duo_certs=DEFAULT_CA_CERT_PATH, use_duo_code_attribute=True, http_proxy=None):
         """
         Initializes instance of Client class
 
@@ -125,6 +125,7 @@ def __init__(self, client_id, client_secret, host,
         redirect_uri             -- Uri to redirect to after a successful auth
         duo_certs                -- (Optional) Provide custom CA certs
         use_duo_code_attribute   -- (Optional: default true) Flag to use `duo_code` instead of `code` for returned authorization parameter
+        http_proxy               -- (Optional) HTTP proxy to tunnel requests through
         """
 
         self._validate_init_config(client_id,
@@ -148,6 +149,11 @@ def __init__(self, client_id, client_secret, host,
         else:
             self._duo_certs = DEFAULT_CA_CERT_PATH
 
+        if http_proxy is not None:
+            self._http_proxy = {'https': http_proxy}
+        else:
+            self._http_proxy = None
+
     def generate_state(self):
         """
         Return a random string of 36 characters
@@ -181,7 +187,8 @@ def health_check(self):
         try:
             response = requests.post(health_check_endpoint,
                                      data=all_args,
-                                     verify=self._duo_certs)
+                                     verify=self._duo_certs,
+                                     proxies=self._http_proxy)
             res = json.loads(response.content)
             if res['stat'] != 'OK':
                 raise DuoException(res)
@@ -282,7 +289,8 @@ def exchange_authorization_code_for_2fa_result(self, duoCode, username, nonce=No
                                      params=all_args,
                                      headers={"user-agent":
                                               user_agent},
-                                     verify=self._duo_certs)
+                                     verify=self._duo_certs,
+                                     proxies=self._http_proxy)
         except Exception as e:
             raise DuoException(e)
 

tests/test_setup_client.py

@@ -11,6 +11,7 @@
 WRONG_HOST = "api-XXXXXXX.test.duosecurity.com"
 REDIRECT_URI = "https://www.example.com"
 CA_CERT_NEW = "/path/to/cert/ca_cert_new.pem"
+PROXY_HOST = "http://proxy.example.com:8001"
 NONE = None
 
 
@@ -116,6 +117,26 @@ def test_disable_duo_cert(self):
     def test_default_duo_code_attribute(self):
         self.assertEqual(self.client._use_duo_code_attribute, True)
 
+    def test_proxy_unset(self):
+        plain_client = client.Client(CLIENT_ID, CLIENT_SECRET, HOST, REDIRECT_URI)
+        self.assertEqual(plain_client._http_proxy, NONE)
+
+    def test_proxy_set_on_args(self):
+        client_with_proxy = client.Client(CLIENT_ID, CLIENT_SECRET, HOST, REDIRECT_URI, NONE, True, PROXY_HOST)
+        self.assertEqual(client_with_proxy._http_proxy, {'https': PROXY_HOST})
+
+    def test_proxy_set_on_kwargs(self):
+        client_with_proxy = client.Client(CLIENT_ID, CLIENT_SECRET, HOST, REDIRECT_URI, http_proxy=PROXY_HOST)
+        self.assertEqual(client_with_proxy._http_proxy, {'https': PROXY_HOST})
+
+    def test_proxy_set_off_args(self):
+        client_with_no_proxy = client.Client(CLIENT_ID, CLIENT_SECRET, HOST, REDIRECT_URI, NONE, True, NONE)
+        self.assertEqual(client_with_no_proxy._http_proxy, NONE)
+
+    def test_proxy_set_off_kwargs(self):
+        client_with_no_proxy = client.Client(CLIENT_ID, CLIENT_SECRET, HOST, REDIRECT_URI, http_proxy=NONE)
+        self.assertEqual(client_with_no_proxy._http_proxy, NONE)
+
 
 if __name__ == '__main__':
     unittest.main()