Source code for sas.qtgui.Utilities.ConnectionProxy

#!/usr/bin/env python
# -*- coding: utf-8 -*-
import urllib.request, urllib.error, urllib.parse
import sys
import json
import logging
import re

'''
HTTP Proxy parser and Connection

connect() function:
    - auto detects proxy in windows, osx
    - in ux systems, the http_proxy enviroment variable must be set
    - if it fails, try to find the proxy.pac address.
      - parses the file, and looks up for all possible proxies
'''


[docs]class ConnectionProxy(object):
[docs] def __init__(self, url, timeout): self.url = url self.timeout = timeout
[docs] def _get_addresses_of_proxy_pac(self): """ Return a list of possible auto proxy .pac files being used, based on the system registry (win32) or system preferences (OSX). @return: list of urls """ pac_files = [] if sys.platform == 'win32': try: import winreg as winreg # used from python 2.0-2.6 except: import winreg # used from python 2.7 onwards net = winreg.OpenKey( winreg.HKEY_CURRENT_USER, "Software\\Microsoft\\Windows\\CurrentVersion\\Internet Settings" ) n_subs, n_vals, last_mod = winreg.QueryInfoKey(net) subkeys = {} for i in range(n_vals): this_name, this_val, this_type = winreg.EnumValue(net, i) subkeys[this_name] = this_val if 'AutoConfigURL' in list(subkeys.keys()) and len(subkeys['AutoConfigURL']) > 0: pac_files.append(subkeys['AutoConfigURL']) elif sys.platform == 'darwin': import plistlib sys_prefs = plistlib.readPlist( '/Library/Preferences/SystemConfiguration/preferences.plist') networks = sys_prefs['NetworkServices'] # loop through each possible network (e.g. Ethernet, Airport...) for network in list(networks.items()): # the first part is a long identifier net_key, network = network if 'ProxyAutoConfigURLString' in list(network['Proxies'].keys()): pac_files.append( network['Proxies']['ProxyAutoConfigURLString']) return list(set(pac_files)) # remove redundant ones
[docs] def _parse_proxy_pac(self, pac_urls_list): ''' For every pac file url in pac_urls_list, it tryes to connect. If the connection is successful parses the file in search for http proxies. @param pac_urls_list: List with urls for the pac files @return: list with all found http proxies ''' proxy_url_list = [] for this_pac_url in pac_urls_list: logging.debug('Trying pac file (%s)...' % this_pac_url) try: response = urllib.request.urlopen( this_pac_url, timeout=self.timeout) logging.debug('Succeeded (%s)...' % this_pac_url) except Exception: logging.debug('Failled (%s)...' % this_pac_url) continue pacStr = response.read() possProxies = re.findall( r"PROXY\s([^\s;,:]+:[0-9]{1,5})[^0-9]", pacStr + '\n') for thisPoss in possProxies: prox_url = 'http://' + thisPoss proxy_dic = {'http': prox_url} proxy_url_list.append(proxy_dic) return proxy_url_list
[docs] def _set_proxy(self,proxy_dic=None): ''' Sets connection proxy. if proxy_dic is None get's teh proxy from the system. To disable autodetected proxy pass an empty dictionary: {} @param proxy_dic: format: {'http': 'http://www.example.com:3128/'} ''' if proxy_dic is None: # The default is to read the list of proxies from the environment variables <protocol>_proxy. # If no proxy environment variables are set, then in a Windows environment proxy settings are # obtained from the registry's Internet Settings section, and in a Mac OS X environment proxy # information is retrieved from the OS X System Configuration # Framework. proxy = urllib.request.ProxyHandler() else: # If proxies is given, it must be a dictionary mapping protocol names to # URLs of proxies. proxy = urllib.request.ProxyHandler(proxy_dic) opener = urllib.request.build_opener(proxy) urllib.request.install_opener(opener)
[docs] def connect(self): ''' Performs the request and gets a response from self.url @return: response object from urllib2.urlopen ''' req = urllib.request.Request(self.url) response = None try: logging.debug("Trying Direct connection to %s..."%self.url) response = urllib.request.urlopen(req, timeout=self.timeout) except Exception as e: logging.debug("Failed!") logging.debug(e) try: logging.debug("Trying to use system proxy if it exists...") self._set_proxy() response = urllib.request.urlopen(req, timeout=self.timeout) except Exception as e: logging.debug("Failed!") logging.debug(e) pac_urls = self._get_addresses_of_proxy_pac() proxy_urls = self._parse_proxy_pac(pac_urls) for proxy in proxy_urls: try: logging.debug("Trying to use the proxy %s found in proxy.pac configuration"%proxy) self._set_proxy(proxy) response = urllib.request.urlopen(req, timeout=self.timeout) except Exception as e: logging.debug("Failed!") logging.debug(e) if response is not None: logging.debug("The connection to %s was successful."%self.url) else: logging.warning("Connection to %s failed..."%self.url) return response
if __name__ == "__main__": from pprint import pprint c = Connection() response = c.connect() if response is not None: print(50 * '-') content = json.loads(response.read().strip()) pprint(content)