"""Utilities for Url handling
"""
#=========================================================================================
# Licence, Reference and Credits
#=========================================================================================
__copyright__ = "Copyright (C) CCPN project (https://www.ccpn.ac.uk) 2014 - 2022"
__credits__ = ("Ed Brooksbank, Joanna Fox, Victoria A Higman, Luca Mureddu, Eliza Płoskoń",
"Timothy J Ragan, Brian O Smith, Gary S Thompson & Geerten W Vuister")
__licence__ = ("CCPN licence. See https://ccpn.ac.uk/software/licensing/")
__reference__ = ("Skinner, S.P., Fogh, R.H., Boucher, W., Ragan, T.J., Mureddu, L.G., & Vuister, G.W.",
"CcpNmr AnalysisAssign: a flexible platform for integrated NMR analysis",
"J.Biomol.Nmr (2016), 66, 111-124, http://doi.org/10.1007/s10858-016-0060-y")
#=========================================================================================
# Last code modification
#=========================================================================================
__modifiedBy__ = "$modifiedBy: Ed Brooksbank $"
__dateModified__ = "$dateModified: 2022-03-04 00:17:35 +0000 (Fri, March 04, 2022) $"
__version__ = "$Revision: 3.1.0 $"
#=========================================================================================
# Created
#=========================================================================================
__author__ = "$Author: CCPN $"
__date__ = "$Date: 2017-04-07 10:28:41 +0000 (Fri, April 07, 2017) $"
#=========================================================================================
# Start of code
#=========================================================================================
BAD_DOWNLOAD = 'Exception: '
from ccpn.util.Logging import getLogger
[docs]def fetchHttpResponse(method, url, data=None, headers=None, proxySettings=None):
"""Generate an http, and return the response
"""
import os
import ssl
import certifi
import urllib
from urllib.request import getproxies
import urllib3.contrib.pyopenssl
from urllib.parse import urlencode, quote
context = ssl.create_default_context()
context.check_hostname = False
context.verify_mode = ssl.CERT_NONE
if not headers:
headers = {'Content-Type': 'application/x-www-form-urlencoded;charset=UTF-8'}
try:
body = urlencode(data or {}, quote_via=quote).encode('utf-8')
except Exception as err:
body = urlencode({}, quote_via=quote).encode('utf-8')
urllib3.contrib.pyopenssl.inject_into_urllib3()
# create the options list for creating an http connection
options = {
'cert_reqs': 'CERT_REQUIRED',
'ca_certs' : certifi.where(),
# 'timeout' : urllib3.Timeout(connect=3.0, read=3.0),
# 'retries' : urllib3.Retry(1, redirect=False)
}
# check whether a proxy is required
from ccpn.util.UserPreferences import UserPreferences, USEPROXY, USEPROXYPASSWORD, PROXYADDRESS, \
PROXYPORT, PROXYUSERNAME, PROXYPASSWORD, USESYSTEMPROXY, VERIFYSSL
def _getProxyIn(proxyDict):
"""Get the first occurrence of a proxy type in the supplied dict
"""
# define a list of proxy identifiers
proxyCheckList = ['HTTPS_PROXY', 'https', 'HTTP_PROXY', 'http']
for pCheck in proxyCheckList:
proxyUrl = proxyDict.get(pCheck, None)
if proxyUrl:
return proxyUrl
if proxySettings:
verifySSL = proxySettings.get(VERIFYSSL)
if verifySSL:
options.update(cert_reqs=ssl.CERT_REQUIRED)
else:
options.update(cert_reqs=ssl.CERT_NONE)
urllib3.disable_warnings()
getLogger().warning('SSL certificates validity check skipped.')
if proxySettings and proxySettings.get(USEPROXY):
# Use the user settings if set
useProxyPassword = proxySettings.get(USEPROXYPASSWORD)
proxyAddress = proxySettings.get(PROXYADDRESS)
proxyPort = proxySettings.get(PROXYPORT)
proxyUsername = proxySettings.get(PROXYUSERNAME)
proxyPassword = proxySettings.get(PROXYPASSWORD)
if useProxyPassword:
# grab the decode from the userPreferences
_userPreferences = UserPreferences(readPreferences=False)
options.update({'headers': urllib3.make_headers(proxy_basic_auth='%s:%s' %
(proxyUsername,
_userPreferences.decodeValue(proxyPassword)))})
proxyUrl = "http://%s:%s/" % (str(proxyAddress), str(proxyPort)) if proxyAddress else None
else:
# read the environment/system proxies if exist
proxyUrl = _getProxyIn(os.environ) or _getProxyIn(urllib.request.getproxies())
# ED: issues - @"HTTPProxyAuthenticated" key on system?. If it exists, the value is a boolean (NSNumber) indicating whether or not the proxy is authentified,
# get the username if the proxy is authenticated: check @"HTTPProxyUsername"
# proxy may still not be defined
if proxyUrl:
http = urllib3.ProxyManager(proxyUrl, **options)
else:
http = urllib3.PoolManager(**options)
# generate an http request
response = http.request(method, url,
headers=headers,
body=body,
preload_content=True,
decode_content=False)
# return the http response
return response
[docs]def fetchUrl(url, data=None, headers=None, timeout=5, proxySettings=None, decodeResponse=True):
"""Fetch url request from the server
"""
import logging
from ccpn.core.lib.ContextManagers import Timeout as timer
from ccpn.util.Common import isWindowsOS
timeoutMessage = 'Could not connect to server. Check connection'
if isWindowsOS():
# Windows does not have signal.SIGALRM - temporary fix
urllib3_logger = logging.getLogger('urllib3')
urllib3_logger.setLevel(logging.CRITICAL)
if not proxySettings:
# read the proxySettings from the preferences
from ccpn.util.UserPreferences import UserPreferences
_userPreferences = UserPreferences(readPreferences=True)
if _userPreferences.proxyDefined:
proxyNames = ['useProxy', 'proxyAddress', 'proxyPort', 'useProxyPassword',
'proxyUsername', 'proxyPassword', 'verifySSL']
proxySettings = {}
for name in proxyNames:
proxySettings[name] = _userPreferences._getPreferencesParameter(name)
response = fetchHttpResponse('POST', url, data=data, headers=headers,
proxySettings=proxySettings)
else:
# use the timeout limited call
with timer(seconds=timeout or 5, timeoutMessage=timeoutMessage):
urllib3_logger = logging.getLogger('urllib3')
urllib3_logger.setLevel(logging.CRITICAL)
if not proxySettings:
# read the proxySettings from the preferences
from ccpn.util.UserPreferences import UserPreferences
_userPreferences = UserPreferences(readPreferences=True)
if _userPreferences.proxyDefined:
proxyNames = ['useProxy', 'proxyAddress', 'proxyPort', 'useProxyPassword',
'proxyUsername', 'proxyPassword', 'verifySSL']
proxySettings = {}
for name in proxyNames:
proxySettings[name] = _userPreferences._getPreferencesParameter(name)
response = fetchHttpResponse('POST', url, data=data, headers=headers,
proxySettings=proxySettings)
return response.data.decode('utf-8') if decodeResponse else response
[docs]def uploadFile(url, fileName, data=None):
import os
if not data:
data = {}
with open(fileName, 'rb') as fp:
fileData = fp.read()
data['fileName'] = os.path.basename(fileName)
data['fileData'] = fileData
try:
return fetchUrl(url, data)
except:
return None
[docs]def checkInternetConnection():
"""Check whether an internet conection is available by testing the CCPN weblink
"""
from ccpn.framework.PathsAndUrls import ccpnUrl
try:
fetchUrl('/'.join([ccpnUrl, 'cgi-bin/checkInternet']))
return True
except Exception as es:
getLogger().exception(es)
return False