PowerDNS-Admin/app/lib/utils.py

import re
import sys
import json
import requests
import urlparse
import hashlib

from app import app
from distutils.version import StrictVersion
from datetime import datetime,timedelta
from threading import Thread
from onelogin.saml2.auth import OneLogin_Saml2_Auth
from onelogin.saml2.utils import OneLogin_Saml2_Utils
from onelogin.saml2.settings import OneLogin_Saml2_Settings
from onelogin.saml2.idp_metadata_parser import OneLogin_Saml2_IdPMetadataParser

idp_timestamp = datetime(1970,1,1)
idp_data = None
if app.config['SAML_ENABLED']:
    idp_data = OneLogin_Saml2_IdPMetadataParser.parse_remote(app.config['SAML_METADATA_URL'])
    if idp_data == None:
        print('SAML: IDP Metadata initial load failed')
        exit(-1)
    idp_timestamp = datetime.now()

def get_idp_data():
    global idp_data, idp_timestamp
    lifetime = timedelta(minutes=app.config['SAML_METADATA_CACHE_LIFETIME'])
    if idp_timestamp+lifetime < datetime.now():
        background_thread = Thread(target=retreive_idp_data)
        background_thread.start()
    return idp_data

def retreive_idp_data():
    global idp_data, idp_timestamp
    new_idp_data = OneLogin_Saml2_IdPMetadataParser.parse_remote(app.config['SAML_METADATA_URL'])
    if new_idp_data != None:
        idp_data = new_idp_data
        idp_timestamp = datetime.now()
        print("SAML: IDP Metadata successfully retreived from: " + app.config['SAML_METADATA_URL'])
    else:
        print("SAML: IDP Metadata could not be retreived")

if 'TIMEOUT' in app.config.keys():
    TIMEOUT = app.config['TIMEOUT']
else:
    TIMEOUT = 10

def auth_from_url(url):
    auth = None
    parsed_url = urlparse.urlparse(url).netloc
    if '@' in parsed_url:
        auth = parsed_url.split('@')[0].split(':')
        auth = requests.auth.HTTPBasicAuth(auth[0], auth[1])
    return auth


def fetch_remote(remote_url, method='GET', data=None, accept=None, params=None, timeout=None, headers=None):
    if data is not None and type(data) != str:
        data = json.dumps(data)

    if timeout is None:
        timeout = TIMEOUT

    verify = False

    our_headers = {
        'user-agent': 'powerdnsadmin/0',
        'pragma': 'no-cache',
        'cache-control': 'no-cache'
    }
    if accept is not None:
        our_headers['accept'] = accept
    if headers is not None:
        our_headers.update(headers)

    r = requests.request(
        method,
        remote_url,
        headers=headers,
        verify=verify,
        auth=auth_from_url(remote_url),
        timeout=timeout,
        data=data,
        params=params
        )
    try:
        if r.status_code not in (200, 400, 422):
            r.raise_for_status()
    except Exception as e:
        raise RuntimeError("While fetching " + remote_url + ": " + str(e)), None, sys.exc_info()[2]

    return r


def fetch_json(remote_url, method='GET', data=None, params=None, headers=None):
    r = fetch_remote(remote_url, method=method, data=data, params=params, headers=headers, accept='application/json; q=1')

    if method == "DELETE":
        return True

    if r.status_code == 204:
        return {}

    try:
        assert('json' in r.headers['content-type'])
    except Exception as e:
        raise Exception("While fetching " + remote_url + ": " + str(e)), None, sys.exc_info()[2]

    # don't use r.json here, as it will read from r.text, which will trigger
    # content encoding auto-detection in almost all cases, WHICH IS EXTREMELY
    # SLOOOOOOOOOOOOOOOOOOOOOOW. just don't.
    data = None
    try:
        data = json.loads(r.content)
    except UnicodeDecodeError:
        data = json.loads(r.content, 'iso-8859-1')
    return data


def display_record_name(data):
    record_name, domain_name = data
    if record_name == domain_name:
        return '@'
    else:
        return record_name.replace('.'+domain_name, '')

def display_master_name(data):
    """
    input data: "[u'127.0.0.1', u'8.8.8.8']"
    """
    matches = re.findall(r'\'(.+?)\'', data)
    return ", ".join(matches)

def display_time(amount, units='s', remove_seconds=True):
    """
    Convert timestamp to normal time format
    """
    amount = int(amount)
    INTERVALS = [(lambda mlsec:divmod(mlsec, 1000), 'ms'),
                 (lambda seconds:divmod(seconds, 60), 's'),
                 (lambda minutes:divmod(minutes, 60), 'm'),
                 (lambda hours:divmod(hours, 24), 'h'),
                 (lambda days:divmod(days, 7), 'D'),
                 (lambda weeks:divmod(weeks, 4), 'W'),
                 (lambda years:divmod(years, 12), 'M'),
                 (lambda decades:divmod(decades, 10), 'Y')]

    for index_start, (interval, unit) in enumerate(INTERVALS):
        if unit == units:
            break

    amount_abrev = []
    last_index = 0
    amount_temp = amount
    for index, (formula, abrev) in enumerate(INTERVALS[index_start: len(INTERVALS)]):
        divmod_result = formula(amount_temp)
        amount_temp = divmod_result[0]
        amount_abrev.append((divmod_result[1], abrev))
        if divmod_result[1] > 0:
            last_index = index
    amount_abrev_partial = amount_abrev[0: last_index + 1]
    amount_abrev_partial.reverse()

    final_string = ''
    for amount, abrev in amount_abrev_partial:
        final_string += str(amount) + abrev + ' '

    if remove_seconds and 'm' in final_string:
        final_string = final_string[:final_string.rfind(' ')]
        return final_string[:final_string.rfind(' ')]

    return final_string

def pdns_api_extended_uri(version):
    """
    Check the pdns version
    """
    if StrictVersion(version) >= StrictVersion('4.0.0'):
        return "/api/v1"
    else:
        return ""

def email_to_gravatar_url(email, size=100):
    """
    AD doesn't necessarily have email
    """

    if not email:
        email=""


    hash_string = hashlib.md5(email).hexdigest()
    return "https://s.gravatar.com/avatar/%s?s=%s" % (hash_string, size)

def prepare_flask_request(request):
    # If server is behind proxys or balancers use the HTTP_X_FORWARDED fields
    url_data = urlparse.urlparse(request.url)
    return {
        'https': 'on' if request.scheme == 'https' else 'off',
        'http_host': request.host,
        'server_port': url_data.port,
        'script_name': request.path,
        'get_data': request.args.copy(),
        'post_data': request.form.copy(),
        # Uncomment if using ADFS as IdP, https://github.com/onelogin/python-saml/pull/144
        # 'lowercase_urlencoding': True,
        'query_string': request.query_string
    }

def init_saml_auth(req):
    own_url = ''
    if req['https'] == 'on':
        own_url = 'https://'
    else:
        own_url = 'http://'
    own_url += req['http_host']
    metadata = get_idp_data()
    settings = {}
    settings['sp'] = {}
    settings['sp']['NameIDFormat'] = idp_data['sp']['NameIDFormat']
    settings['sp']['entityId'] = app.config['SAML_SP_ENTITY_ID']
    settings['sp']['privateKey'] = ''
    settings['sp']['x509cert'] = ''
    settings['sp']['assertionConsumerService'] = {}
    settings['sp']['assertionConsumerService']['binding'] = 'urn:oasis:names:tc:SAML:2.0:bindings:HTTP-POST'
    settings['sp']['assertionConsumerService']['url'] = own_url+'/saml/authorized'
    settings['sp']['attributeConsumingService'] = {}
    settings['sp']['singleLogoutService'] = {}
    settings['sp']['singleLogoutService']['binding'] = 'urn:oasis:names:tc:SAML:2.0:bindings:HTTP-Redirect'
    settings['sp']['singleLogoutService']['url'] = own_url+'/saml/sls'
    settings['idp'] = metadata['idp']
    settings['strict'] = True
    settings['debug'] = app.config['SAML_DEBUG']
    settings['security'] = {}
    settings['security']['digestAlgorithm'] = 'http://www.w3.org/2000/09/xmldsig#sha1'
    settings['security']['metadataCacheDuration'] = None
    settings['security']['metadataValidUntil'] = None
    settings['security']['requestedAuthnContext'] = True
    settings['security']['signatureAlgorithm'] = 'http://www.w3.org/2000/09/xmldsig#rsa-sha1'
    settings['security']['wantAssertionsEncrypted'] = False
    settings['security']['wantAttributeStatement'] = True
    settings['security']['wantNameId'] = True
    settings['security']['authnRequestsSigned'] = False
    settings['security']['logoutRequestSigned'] = False
    settings['security']['logoutResponseSigned'] = False
    settings['security']['nameIdEncrypted'] = False
    settings['security']['signMetadata'] = False
    settings['security']['wantAssertionsSigned'] = True
    settings['security']['wantMessagesSigned'] = True
    settings['security']['wantNameIdEncrypted'] = False
    settings['contactPerson'] = {}
    settings['contactPerson']['support'] = {}
    settings['contactPerson']['support']['emailAddress'] = app.config['SAML_SP_CONTACT_NAME']
    settings['contactPerson']['support']['givenName'] = app.config['SAML_SP_CONTACT_MAIL']
    settings['contactPerson']['technical'] = {}
    settings['contactPerson']['technical']['emailAddress'] = app.config['SAML_SP_CONTACT_NAME']
    settings['contactPerson']['technical']['givenName'] = app.config['SAML_SP_CONTACT_MAIL']
    settings['organization'] = {}
    settings['organization']['en-US'] = {}
    settings['organization']['en-US']['displayname'] = 'PowerDNS-Admin'
    settings['organization']['en-US']['name'] = 'PowerDNS-Admin'
    settings['organization']['en-US']['url'] = own_url
    auth = OneLogin_Saml2_Auth(req, settings)
    return auth