Skip to content

Commit

Permalink
Merge pull request #136 from vrdel/devel
Browse files Browse the repository at this point in the history 
ARGO-1674 Use requests library in connectors
  • Loading branch information
@themiszamani
themiszamani authored Feb 15, 2019
2 parents 2000850 + acd8092 commit d1afec3
Show file tree
Hide file tree
Showing 2 changed files with 23 additions and 96 deletions.
3 changes: 2 additions & 1 deletion argo-egi-connectors.spec
View file
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,8 @@ Obsoletes: ar-sync
Prefix: %{_prefix}
Requires: argo-ams-library
Requires: avro
Requires: pyOpenSSL
Requires: python-requests
Requires: python2-ndg_httpsclient
Source0: %{name}-%{version}.tar.gz

BuildArch: noarch
Expand Down
116 changes: 21 additions & 95 deletions modules/input.py
View file
Original file line number Diff line number Diff line change
@@ -1,17 +1,11 @@
import base64
import httplib
import json
import os
import requests
import socket
import xml.dom.minidom

from argo_egi_connectors.helpers import retry

from OpenSSL.SSL import TLSv1_METHOD, Context, Connection
from OpenSSL.SSL import VERIFY_PEER
from OpenSSL.SSL import WantReadError as SSLWantReadError
from ssl import SSLError
from time import sleep
from xml.parsers.expat import ExpatError
from urlparse import urlparse

Expand All @@ -25,49 +19,45 @@ def connection(logger, msgprefix, globopts, scheme, host, url, custauth=None):
try:
buf = None

if scheme.startswith('https'):
if eval(globopts['AuthenticationVerifyServerCert'.lower()]):
verify_cert(host, int(globopts['ConnectionTimeout'.lower()]),
globopts['AuthenticationCAPath'.lower()],
globopts['AuthenticationCAFile'.lower()])
conn = httplib.HTTPSConnection(host, 443,
globopts['AuthenticationHostKey'.lower()],
globopts['AuthenticationHostCert'.lower()],
timeout=int(globopts['ConnectionTimeout'.lower()]))
else:
conn = httplib.HTTPConnection(host, 80,
timeout=int(globopts['ConnectionTimeout'.lower()]))

headers = {}
if custauth and eval(custauth['AuthenticationUsePlainHttpAuth'.lower()]):
userpass = base64.b64encode(custauth['AuthenticationHttpUser'.lower()] + ':' \
+ custauth['AuthenticationHttpPass'.lower()])
headers={'Authorization': 'Basic ' + userpass}

conn.request('GET', url, headers=headers)
resp = conn.getresponse()
if scheme.startswith('https'):
response = requests.get('https://'+ host + url, headers=headers,
cert=(globopts['AuthenticationHostCert'.lower()],
globopts['AuthenticationHostKey'.lower()]),
verify=eval(globopts['AuthenticationVerifyServerCert'.lower()]),
timeout=int(globopts['ConnectionTimeout'.lower()]))
response.raise_for_status()
else:
response = requests.get('http://'+ host + url, headers=headers,
timeout=int(globopts['ConnectionTimeout'.lower()]))

if resp.status >= 300 and resp.status < 400:
headers = resp.getheaders()

if response.status_code >= 300 and response.status_code < 400:
headers = response.headers
location = filter(lambda h: 'location' in h[0], headers)
if location:
redir = urlparse(location[0][1])
else:
raise httplib.HTTPException('No Location header set for redirect')
raise requests.exceptions.RequestException('No Location header set for redirect')

return connection(logger, msgprefix, globopts, scheme, redir.netloc, redir.path + '?' + redir.query, custauth=custauth)

elif resp.status == 200:
buf = resp.read()
elif response.status_code == 200:
buf = response.content
if not buf:
raise httplib.HTTPException('Empty response')
raise requests.exceptions.RequestException('Empty response')

else:
raise httplib.HTTPException('Response: %s %s' % (resp.status, resp.reason))
raise requests.exceptions.RequestException('response: %s %s' % (response.status_code, response.reason))

return buf

except SSLError as e:
except requests.exceptions.SSLError as e:
if (getattr(e, 'args', False) and type(e.args) == tuple
and type(e.args[0]) == str
and 'timed out' in e.args[0]):
Expand All @@ -86,7 +76,7 @@ def connection(logger, msgprefix, globopts, scheme, host, url, custauth=None):
repr(e)))
raise e

except httplib.HTTPException as e:
except requests.exceptions.RequestException as e:
logger.warn('%sCustomer:%s Job:%s HTTP error %s - %s' % (msgprefix + ' ' if msgprefix else '',
logger.customer, logger.job,
scheme + '://' + host + url,
Expand All @@ -100,9 +90,6 @@ def connection(logger, msgprefix, globopts, scheme, host, url, custauth=None):
repr(e)))
return False

finally:
conn.close()


def parse_xml(logger, objname, globopts, buf, method):
try:
Expand Down Expand Up @@ -134,64 +121,3 @@ def parse_json(logger, objname, globopts, buf, method):

else:
return doc


def verify_cert(host, timeout, capath, cafile):
def verify_cert(host, ca, timeout):
server_ctx = Context(TLSv1_METHOD)
server_cert_chain = []

if os.path.isdir(ca):
server_ctx.load_verify_locations(None, ca)
else:
server_ctx.load_verify_locations(ca, None)

def verify_cb(conn, cert, errnum, depth, ok):
server_cert_chain.append(cert)
return ok
server_ctx.set_verify(VERIFY_PEER, verify_cb)

sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setblocking(1)
sock.settimeout(timeout)
sock.connect((host, 443))

server_conn = Connection(server_ctx, sock)
server_conn.set_connect_state()

def iosock_try():
ok = True
try:
server_conn.do_handshake()
sleep(0.5)
except SSLWantReadError as e:
ok = False
pass
except Exception as e:
raise e
return ok

try:
while True:
if iosock_try():
break

server_subject = server_cert_chain[-1].get_subject()
if host != server_subject.CN:
raise SSLError('Server certificate CN does not match %s' % host)

except SSLError as e:
raise e
finally:
server_conn.shutdown()
server_conn.close()

return True

try:
verify_cert(host, capath, timeout)
except SSLError:
verify_cert(host, cafile, timeout)


return True

0 comments on commit d1afec3

Please sign in to comment.