moving utilities from rest.py into recently added utils.py

parent 7540b69c
......@@ -26,17 +26,13 @@ if os.name == 'posix':
import json
from six.moves.urllib.parse import urlparse, urlencode
from base64 import b64encode
from netrc import netrc, NetrcParseError
from subprocess import Popen, PIPE
import shlex
import stat
from datetime import datetime
import six
from six.moves import map
from .exceptions import HTTPError
from .version import __version__
from .utils import event_or_superevent, handle_str_or_list_arg
from .utils import event_or_superevent, handle_str_or_list_arg, safe_netrc, \
cleanListInput, get_dt_from_openssl_output, is_expired
DEFAULT_SERVICE_URL = "https://gracedb.ligo.org/api/"
DEFAULT_BASIC_SERVICE_URL = "https://gracedb.ligo.org/apibasic/"
......@@ -61,95 +57,6 @@ if sys.hexversion < 0x20709f0:
suppress_ragged_eofs)
ssl.wrap_socket = wrap_socket_patched
#-----------------------------------------------------------------
# Utilities
class safe_netrc(netrc):
"""The netrc.netrc class from the Python standard library applies access
safety checks (requiring that the netrc file is readable only by the
current user, and not by group members or other users) only if using the
netrc file in the default location (~/.netrc). This subclass applies the
same access safety checks regardless of the path to the netrc file."""
def _parse(self, file, fp, *args, **kwargs):
# Copied and adapted from netrc.py from Python 2.7
if os.name == 'posix':
prop = os.fstat(fp.fileno())
if prop.st_uid != os.getuid():
try:
fowner = pwd.getpwuid(prop.st_uid)[0]
except KeyError:
fowner = 'uid %s' % prop.st_uid
try:
user = pwd.getpwuid(os.getuid())[0]
except KeyError:
user = 'uid %s' % os.getuid()
raise NetrcParseError(
("~/.netrc file owner (%s) does not match"
" current user (%s)") % (fowner, user),
file)
if (prop.st_mode & (stat.S_IRWXG | stat.S_IRWXO)):
raise NetrcParseError(
"~/.netrc access too permissive: access"
" permissions must restrict access to only"
" the owner", file)
return netrc._parse(self, file, fp, *args, **kwargs)
# XXX It would be nice to get rid of this if we can.
# It seems that you can pass python lists via the requests package.
# That would make putting our lists into comma-separated strings
# unnecessary.
def cleanListInput(list_arg):
stringified_list = list_arg
if isinstance(list_arg, float) or isinstance(list_arg, int):
stringified_value = str(list_arg)
return stringified_value
if not isinstance(list_arg, six.string_types):
stringified_list = ','.join(map(str,list_arg))
return stringified_list
# The following are used to check whether a user has tried to use
# an expired certificate.
# Parse a datetime object out of the openssl output.
# Note that this returns a naive datetime object.
def get_dt_from_openssl_output(s):
dt = None
err = ''
# Openssl spits out a string like "notAfter=Feb 6 15:17:54 2016 GMT"
# So we first have to split off the bit after the equal sign.
try:
date_string = s.split('=')[1].strip()
except:
err = 'Openssl output not understood.'
return dt, err
# Next, attempt to parse the date with strptime.
try:
dt = datetime.strptime(date_string, "%b %d %H:%M:%S %Y %Z")
except:
err = 'Could not parse time string from openssl.'
return dt, err
return dt, err
# Given a path to a cert file, check whether it is expired.
def is_expired(cert_file):
cmd = 'openssl x509 -enddate -noout -in %s' % cert_file
p = Popen(shlex.split(cmd), stdout=PIPE, stderr=PIPE)
out, err = p.communicate()
expired = None
if p.returncode == 0:
dt, err = get_dt_from_openssl_output(out)
if dt:
# Note that our naive datetime must be compared with a UTC
# datetime that has been rendered naive.
expired = dt <= datetime.utcnow().replace(tzinfo=None)
return expired, err
#-----------------------------------------------------------------
# HTTP/S Proxy classes
# Taken from: http://code.activestate.com/recipes/456195/
......
from functools import wraps
import os
import re
import shlex
import six
import stat
from datetime import datetime
from functools import wraps
from netrc import netrc, NetrcParseError
from subprocess import Popen, PIPE
if os.name == 'posix':
import pwd
SUPEREVENT_PREFIXES = ['S', 'GW']
superevent_prefix_regex = re.compile(r'^({prefixes})\d+'.format(
......@@ -17,6 +25,7 @@ def event_or_superevent(func):
**kwargs)
return inner
# Function for checking arguments which can be strings or lists
# If a string, converts to list for ease of processing
def handle_str_or_list_arg(arg, arg_name):
......@@ -29,4 +38,93 @@ def handle_str_or_list_arg(arg, arg_name):
raise TypeError("{0} arg is {1}, should be str or list" \
.format(arg_name, type(arg)))
return arg
# XXX It would be nice to get rid of this if we can.
# It seems that you can pass python lists via the requests package.
# That would make putting our lists into comma-separated strings
# unnecessary.
def cleanListInput(list_arg):
stringified_list = list_arg
if isinstance(list_arg, float) or isinstance(list_arg, int):
stringified_value = str(list_arg)
return stringified_value
if not isinstance(list_arg, six.string_types):
stringified_list = ','.join(map(str,list_arg))
return stringified_list
# The following are used to check whether a user has tried to use
# an expired certificate.
# Parse a datetime object out of the openssl output.
# Note that this returns a naive datetime object.
def get_dt_from_openssl_output(s):
dt = None
err = ''
# Openssl spits out a string like "notAfter=Feb 6 15:17:54 2016 GMT"
# So we first have to split off the bit after the equal sign.
try:
date_string = s.split('=')[1].strip()
except:
err = 'Openssl output not understood.'
return dt, err
# Next, attempt to parse the date with strptime.
try:
dt = datetime.strptime(date_string, "%b %d %H:%M:%S %Y %Z")
except:
err = 'Could not parse time string from openssl.'
return dt, err
return dt, err
# Given a path to a cert file, check whether it is expired.
def is_expired(cert_file):
cmd = 'openssl x509 -enddate -noout -in %s' % cert_file
p = Popen(shlex.split(cmd), stdout=PIPE, stderr=PIPE)
out, err = p.communicate()
expired = None
if p.returncode == 0:
dt, err = get_dt_from_openssl_output(out)
if dt:
# Note that our naive datetime must be compared with a UTC
# datetime that has been rendered naive.
expired = dt <= datetime.utcnow().replace(tzinfo=None)
return expired, err
class safe_netrc(netrc):
"""The netrc.netrc class from the Python standard library applies access
safety checks (requiring that the netrc file is readable only by the
current user, and not by group members or other users) only if using the
netrc file in the default location (~/.netrc). This subclass applies the
same access safety checks regardless of the path to the netrc file."""
def _parse(self, file, fp, *args, **kwargs):
# Copied and adapted from netrc.py from Python 2.7
if os.name == 'posix':
prop = os.fstat(fp.fileno())
if prop.st_uid != os.getuid():
try:
fowner = pwd.getpwuid(prop.st_uid)[0]
except KeyError:
fowner = 'uid %s' % prop.st_uid
try:
user = pwd.getpwuid(os.getuid())[0]
except KeyError:
user = 'uid %s' % os.getuid()
raise NetrcParseError(
("~/.netrc file owner (%s) does not match"
" current user (%s)") % (fowner, user),
file)
if (prop.st_mode & (stat.S_IRWXG | stat.S_IRWXO)):
raise NetrcParseError(
"~/.netrc access too permissive: access"
" permissions must restrict access to only"
" the owner", file)
return netrc._parse(self, file, fp, *args, **kwargs)
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment