Coverage for main.py: 17%

Hot-keys on this page

r m x p toggle line displays

j k next/prev highlighted chunk

0 (zero) top of page

1 (one) first highlighted chunk

#############################################################################

# Authentication Server

# vim:ts=3:sw=3:expandtab

# Author: Lionel Litty

# A Tornado-based web server to handle user authentication.

# Since we use multiple hosts, consistent auth cookies need to be set on all

# the hosts we use. An anchor host (e.g., safe.menlosec.com) is used as the

# primary host to help avoid reauthentication and to provide bid

# continuity. The list of auxiliary hosts is either obtained via the

# get_cookied_hosts call or provided by the client (which helps with

# inconsistent DNS resolution, more below).

# An alternative would be to use a domain cookie, set on menlosec.com. However,

# this does not work in on-prem scenarios: the service is deployed at

# safe.customer.com and it's not secure to set the cookie on customer.com. It's

# also usually not practical for customers to create a more complex hierarchy

# (e.g., res.safe.customer.com and xhr.safe.customer.com).

# Since setting cookies on XHRs that are CORS can run afoul of third party

# cookie blocking in some browsers (Safari), we only set cookies when the

# request matches the page's origin (non-CORS XHRs).

# Thus, the auxiliary host cookies are set by redirecting the client browser to

# each host in the redirect list, then making an XHR that contains a token

# allowing the cookie to be set immediately without requiring the user to

# re-enter their credentials. The token specifies the uid, the bid, and the

# host for which it can be used. It is signed and has a short validity.

# The token can be obtained in two ways: it is returned when authenticating to

# the main host or it can be obtained when the main host cookie is present

# (through authentication). Through authentication is provided to help with

# changing auxiliary hosts or missing auxiliary host cookies without forcing

# reauthentication.

# Three sources are supported for authentication/authorization: LDAP, the UDB

# and a list of IPs obtained periodically from S3 from which anonymous access

# is allowed. In the anonymous access scenario, through authentication occurs

# without user input. If a user explicitly visits the login page, they can

# login even if their IP is in the anonymous list.

# The auth server is also used to verify that users are still allowed to use

# the system via authorization requests. These result in a check with the

# authoritative source (UDB, LDAP, anonymous IPs). The result is cached for a

# configurable amount of time (validity_period) to avoid overwhelming the

# authoritative source.

# Authentication can also occur from the PNR side. The mechanism is the same

# and through authentication is attempted. In addition, a token is created to

# support setting a soft ID cookie on 3rd party domains (soft in the sense that

# this id is only used for reporting and shouldn't be used for allowing access

# to backend resources). 3rd party domain cookie setting is quite frequent and

# can happen on requests that are not for the top level page (e.g., images

# fetched from a 3rd party). In the PNR case, when token generation is possible

# because a safeview-id cookie is present, the redirects take place at the

# network level.

# Note on inconsistent DNS resolution and the need for the list of auxiliary

# domains to be passed in as a url parameter to the login page: we encountered

# scenarios where safeview is used in proxy mode and the client connects to the

# us-east region. The safeview id cookie is not present, resulting in a

# redirect to the anchor domain. That anchor domain then resolves to the

# us-west cluster, causing the auxiliary domains used to be the us-west

# ones. This causes a redirect loop.

#############################################################################

"""AuthServer"""

import abc

import base64

import httplib

import json

import logging

import os

import re

import socket

import time

import urlparse

from urllib import urlencode, quote

import netifaces

import tornado.escape

import tornado.httpclient

import tornado.ioloop

import tornado.netutil

import tornado.web

import ldap_auth

from safly.config import config, get_build_id

from safly.event_reporter import EventReporter

from safly.web import RequestHandler

from safly.saml_auth import SAMLAuth, InvalidSamlConfig, InvalidSamlResponse

from safly.tenant_config import TenantConfigIni, TenantConfigPnr

from safly.app import SafelyTornadoApp, SafelyApplication

from safly.misc import www, aes_encrypt_hmac, aes_decrypt_hmac

log = logging.getLogger('root')

log.setLevel(logging.INFO)

def query_to_dict(query):

return {k: v for k, v in urlparse.parse_qsl(query, True)}

def generate_v1_pnr_token(user):

"""Generate a v1 token that is used to authenticate users in the pnr

system"""

cookie_secret = config.get('authentication', 'pnr_cookie_secret')

return base64.urlsafe_b64encode(tornado.web.create_signed_value(

cookie_secret, 'pnr_token',

json.dumps({'bid': user['bid'], 'uid': user['uid'], 'tid': user['tid']})))

def encrypt_hmac_json(**kwargs):

sc_secret = config.get('authentication', 'sc_token_secret')

return base64.urlsafe_b64encode(aes_encrypt_hmac(sc_secret,

json.dumps(kwargs)))

def decrypt_hmac_json(token):

sc_secret = config.get('authentication', 'sc_token_secret')

return json.loads(aes_decrypt_hmac(sc_secret,

base64.urlsafe_b64decode(token)))

def generate_v2_pnr_token(user):

"""Generate a v2 token that is used to authenticate users in the pnr

system"""

return "v2:" + encrypt_hmac_json(uid=user['uid'], tid=user['tid'],

ts=int(time.time()))

def encode_relay_state(tid, url, entry_mode='web'):

if config.getboolean('authentication', 'saml_use_relay_state'):

return encrypt_hmac_json(tid=int(tid), url=url, entry_mode=entry_mode)

else:

return url

def _encode_argument_hmac(name, value):

cookie_secret = config.get('authentication', 'pnr_cookie_secret')

return tornado.web.create_signed_value(cookie_secret, name, value)

def _encode_b64argument_hmac(name, value):

return base64.urlsafe_b64encode(_encode_argument_hmac(name, value))

def _decode_argument_hmac(name, value, max_age_days):

cookie_secret = config.get('authentication', 'pnr_cookie_secret')

return tornado.web.decode_signed_value(cookie_secret, name, value,

max_age_days=max_age_days)

def _decode_b64argument_hmac(name, encoded_value, max_age_days=1):

b64_value_hmac = encoded_value + '=='

value_hmac = base64.urlsafe_b64decode(b64_value_hmac.encode('utf-8'))

return _decode_argument_hmac(name, value_hmac, max_age_days)

def decode_user_hmac(encoded_user_id):

return _decode_b64argument_hmac('user', encoded_user_id)

def sign_params(**kwargs):

"""Canonical way of constructing signed params - pnr-icap-auth must be

compatible with this method."""

tok = _encode_argument_hmac('p', json.dumps(kwargs)).split('|')

b64v = tok[4].split(':')[1]

# b64 to urlsafe-b64

v = b64v.replace('=', '').replace('/', '_').replace('+', '-')

return {'v': v, 'ts': tok[2].split(':')[1], 'p': tok[5]}

def expect_signed_params(v='', ts='', p='', **kwargs):

"""Reconstruct the signed token that pnr-icap-auth produced, validate

the signature and construct dict of params. Expecting valid params to never

be empty, so if you observe an empty dict, the decoding or signature

validation failed."""

v = v.encode('utf-8')

# urlsafe-b64 to b64 encoding translation.

# (len(v) + 3) & -4 rounds up to the next multiple of 4, so the right padding

# with '=' is extracted from (...+'====').

tok = (v.replace('_', '/').replace('-', '+') + '====')[0:(len(v) + 3) & -4]

token = '2|1:0|%d:%s|1:p|%d:%s|%s' % (len(ts), ts, len(tok), tok, p)

params = _decode_argument_hmac('p', token,

config.getfloat('authentication',

'signed_token_max_age_minutes')/(24*60))

if not params:

return dict()

return json.loads(params)

class CSPHandler(RequestHandler):

# By default, the assumption is that this needs to occur over https. Derived

# classes can override.

HTTP_OK = False

def __init__(self, *args, **kwargs):

super(CSPHandler, self).__init__(*args, **kwargs)

self._connect_sources = set()

def set_csp(self, hosts):

if hosts:

self._connect_sources.update(set(hosts))

def finish(self):

self.set_csp(self.domain_scheme.get_cookied_hosts())

# FIXME: 4-level DomainScheme does not work with this implementation of

# CSP. Desupport 4-level DomainScheme, or fix this logic.

csp = ("default-src 'none'; "

"script-src 'self' https://%(safeview_static)s; "

"img-src 'self' data: https://%(safeview_static)s; "

"style-src 'self' https://%(safeview_static)s; "

"font-src 'self' https://%(safeview_static)s; "

"connect-src 'self' %(cookied_domains)s; "

"frame-ancestors 'none'; base-uri 'none'; "

"report-uri /safeview-client-logger/csp-violation;") % {

'safeview_static': self.domain_scheme.get_cluster_host(),

'cookied_domains': ' '.join(

['https://' + d for d in self._connect_sources]

)

}

self.set_header('Content-Security-Policy', csp)

super(CSPHandler, self).finish()

def ok_proto(self):

# Login should always occur over https.

return self.HTTP_OK or self.request.protocol == 'https'

class BaseHandler(CSPHandler):

# See Tornado.web

INVALID_HEADER_CHAR_RE = re.compile(br"[\x00-\x1f]")

MAX_LOC_LEN = 2 ** 14

def __init__(self, *args, **kwargs):

super(BaseHandler, self).__init__(*args, **kwargs)

# Track username, user for better logging.

self.username = None

self.user = None

self.tenant_id = -1

self.redirect_list = None

self.expired_user = None

self.req_dict = {}

self.add_header('Vary','Accept-Language')

def get_extra_logger_fields(self):

fields = super(BaseHandler, self).get_extra_logger_fields()

if self.user:

fields += [('tenant_id', self.user['tid'], 5),

('user_id', self.user.get('uid'), 25)]

if self.expired_user:

fields += [('expired_tenant_id', self.expired_user['tid'], 5),

('expired_user_id', self.expired_user.get('uid'), 25)]

if self.username:

fields += [('username', self.username, 15)]

if self.redirect_list:

fields += [('redirect_list', self.redirect_list, 50)]

if self.req_dict:

if self.req_dict.get('token', None) is not None:

fields += [('query_token_present', True, 15)]

for p in ['pnr', 'ip', 'user', 'username', 's', 'url', 'saml',

'entry_mode', 'through', 'redirect_list', 'v', 'ts', 'p']:

v = self.req_dict.get(p, None)

if v is not None:

fields += [('query_' + p, v, 15)]

return fields

def get_logger_name(self):

return 'auth-main'

def initialize(self, args, en_locale):

# gets one argument based on url spec 3rd arg - OK to disable warning.

# pylint: disable=arguments-differ

self.args = args

self._en_locale = en_locale

def finish_with_json(self, status, resp_dict):

self.set_status(status)

resp = json.dumps(resp_dict)

self.log.debug('%s response: %s.', self.request.uri, resp)

self.write(resp)

self.finish()

def finish_with_msg(self, status, msg_id):

self.finish_with_json(status, {'msg': self.locale.translate(msg_id),

'msg_id': msg_id})

class BaseTenantHandler(BaseHandler):

__metaclass__ = abc.ABCMeta

@abc.abstractmethod

def _xhr_post(self, req_dict):

"""Must be a coroutine"""

pass

@tornado.gen.coroutine

def post(self):

"""Handle XHRs"""

if not self.ok_proto():

self.log.error({'details': 'communication with auth server over http '

'not allowed.'},

event='auth-bad-request')

self.finish_with_msg(httplib.BAD_REQUEST, 'ID_ALERT_HTTP_NOT_ALLOWED')

return

try:

self.req_dict = json.loads(self.request.body)

except ValueError as e:

self.log.error({'details': 'could not parse request', 'error': e,

'request_content_type': self.request.headers['content-type']},

event='auth-bad-request', opslog=True)

self.finish_with_msg(httplib.BAD_REQUEST, 'ID_ALERT_COULD_NOT_PARSE')

return

try:

yield self._xhr_post(self.req_dict)

except KeyError as e:

self.log.error({'details': 'request missing argument',

'error': e, 'url': self.request.uri,

'present_args': self.req_dict.keys()},

event='auth-bad-request', opslog=True)

self.finish_with_msg(httplib.BAD_REQUEST, 'ID_ALERT_COULD_NOT_PARSE')

return

def try_again_later(self, delay):

msg_id = 'ID_LOGIN_TRY_AGAIN_LATER'

self.finish_with_json(httplib.INTERNAL_SERVER_ERROR,

{'msg': self.locale.translate(msg_id),

'delay': delay,

'msg_id': msg_id})

def check_udb_response(self, response, action, msg_id,

error=httplib.INTERNAL_SERVER_ERROR):

"""Check a REST call response.

Returns an error to the client and None if the response is

invalid. Otherwise, returns the response dictionary.

"""

# Construct suitable name for the event.

event_name = '-'.join(['auth', action.lower().replace(' ', '-'),

'failed'])

if response.error or response.code != httplib.OK:

self.log.error({'response_code': response.code,

'response_error': response.error,

'body': response.body},

event=event_name)

self.finish_with_msg(error, msg_id)

return

try:

resp_dict = json.loads(response.body)

result = resp_dict['result']

reason = resp_dict.get('reason', None)

delay = resp_dict.get('delay', 0)

except (ValueError, KeyError, TypeError) as e:

self.log.warning({'error': e, 'body': response.body},

event=event_name, opslog=True)

self.finish_with_msg(error, msg_id)

return

if result != 0:

self.log.warning({'details': 'non-zero result',

'result': result, 'reason': reason},

event=event_name, opslog=True)

# Be selective about the error messages we expose to the user.

if reason == 'already exists':

msg_id = 'ID_ALERT_USER_EXISTS'

elif delay:

self.try_again_later(delay)

return

self.finish_with_msg(error, msg_id)

return

return resp_dict

def signed_params(self, ts='', **kwargs):

now = time.time()

if abs(now - int(ts)) > 15*60:

self.log.error({'now': now, 'ts': ts}, event='auth-clock-skew')

return expect_signed_params(ts=ts, **kwargs)

class BaseAuthHandler(BaseTenantHandler):

@abc.abstractmethod

def _on_xhr_post(self, req_dict):

"""Must be a coroutine"""

pass

@tornado.gen.coroutine

def _xhr_post(self, req_dict):

self.prepare_redirect_list(req_dict.get('redirect_list'))

yield self._on_xhr_post(req_dict)

def _is_reauth_required(self, entry_web=True, tid=None, timestamp=None):

tid = tid or self.user['tid']

if (not self.application.tenant_config.

is_reauth_timeout_present(tid, entry_web)):

return False

reauth_timeout = self.application.tenant_config.get_reauth_timeout(

tid, entry_web)

delta_t = (int(time.time()) - timestamp if timestamp

else self.time_since_last_auth())

return delta_t >= reauth_timeout

def remote_ip(self, ip, is_pnr):

"""Validate the presence of ip param"""

# assert: require_ip_param implies "not get_ip_proxy_protocol" (can't

# require IP param, if using get_ip_proxy_protocol)

# Remove the getip functionality once the proxy protocol is standard

if config.getboolean('authentication', 'get_ip_proxy_protocol'):

return self.request.remote_ip

if ip:

return ip

# if require_ip_param is true, don't return default ip address

if is_pnr and config.getboolean('authentication', 'require_ip_param'):

self.log.info({}, event='auth-ip-hmac-missing')

return None

return self.request.remote_ip

def decode_ip_hmac(self, encoded_ip, is_pnr):

"""Validate and decode IP hmac value"""

peer_ip = self.remote_ip(encoded_ip, is_pnr)

if encoded_ip and peer_ip is encoded_ip:

return _decode_b64argument_hmac('ip', encoded_ip, max_age_days=365)

return peer_ip

def raw_redirect(self, url, status=httplib.FOUND):

"""Redirect to url with status.

Similar to redirect provided by Tornado, but enforces a looser max header

length for the redirect location.

"""

assert isinstance(status, int) and 300 <= status <= 399

if self._headers_written:

raise Exception('Cannot redirect after headers have been written')

escaped_url = tornado.escape.utf8(url)

if (len(escaped_url) > self.MAX_LOC_LEN or

self.INVALID_HEADER_CHAR_RE.search(escaped_url)):

self.log.error({'url': escaped_url[:256], 'len': len(escaped_url)},

event='auth-bad-proxy-auth-url')

self.finish_with_msg(httplib.BAD_REQUEST, 'ID_ALERT_INVALID_URL')

return

self.set_status(status)

self._headers['Location'] = escaped_url

self.finish()

def _get_login_host(self):

"""Use the current hostname as the login host.

This has to be the case since we use a cookie to pass the tid around and

we set the cookie on the current host.

FIXME - this should check instead that the request.uri host matches the

be accurate on-prem.

"""

if not config.getboolean('authentication', 'saml_use_request_host'):

return self.domain_scheme.get_login_host()

host = self.request.host

domain = self.domain_scheme.get_trust_domain()

if host != domain and not host.endswith('.' + domain):

self.log.error({'host': host, 'allowed': domain},

event='saml-invalid-login-host')

raise Exception('Invalid domain')

return host

@tornado.gen.coroutine

def validate_user_saml(self, userid, tid, justify,

saml_auth=None, tid_conf=None):

tenant_config = self.application.tenant_config

if saml_auth and tid_conf:

results = tenant_config.validate_attribute_authorization(

tenant_config.get_attributes(saml_auth, tid_conf, justify),

justify)

raise tornado.gen.Return(results)

results = yield tenant_config.validate_saml_authorization(userid, tid,

justify)

raise tornado.gen.Return(results)

def try_login(self, req_dict):

self.req_dict = req_dict

if 'v' in req_dict:

# assert: v= is present only on return from getip, which happens

# only when is_pnr=True when redirecting to getip

client_ip = self.decode_ip_hmac(req_dict.get('ip'), True)

params = self.signed_params(**req_dict)

if not params or not client_ip:

self.log.error({'details': 'getip',

'query_v': req_dict.get('v'),

'query_ts': req_dict.get('ts'),

'query_p': req_dict.get('p')},

event='auth-invalid-params-hmac')

self.unexpected_redirect()

return

# override 'ip' so it is no longer a HMACed value henceforward

params['ip'] = client_ip

req_dict.update(params)

is_pnr = bool(req_dict.get('pnr'))

# assert: if ip=... is present, it was added by a trusted party:

# POST to /login does not contain ip=..., proxy_auth and getip validate

# ip=... comes from trusted pnr-icap-auth

client_ip = self.remote_ip(req_dict.get('ip'), is_pnr)

# assert: client_ip is None if and only if is_pnr,

# and require_ip_param, and ip=... was missing

# Disable getip functionality when using the PROXY protocol

# The client IP and user ID might not be available from PnR if the

# request is SSL and not bumped by the first proxy (there are

# use-cases where the PnR proxy is 2nd in the chain)

if not client_ip or (

not config.getboolean('authentication', 'get_ip_proxy_protocol') and

is_pnr and

req_dict.get('user', None) is None and

config.get('authentication', 'user_id_header')):

getip_hostname = config.get('authentication', 'getip_hostname')

# TODO: Desupport getip including proxy_header auth before

# desupporting /login GET

old_query = dict(req_dict)

old_query.pop('ip', None)

old_query.pop('user', None)

new_query = sign_params(**old_query)

self.log.info({}, event='auth-getip')

self.raw_redirect('http://%s/account/login?%s' % (

getip_hostname, urlencode(new_query)))

return

if req_dict.get('through'):

self._cookie_refresh()

entry_mode = req_dict.get('entry_mode') or 'web'

expired_user = self.user or self.get_current_user(max_age_days=365)

ip_tid = self.application.tid_for_ip(client_ip)

tid = expired_user and expired_user['tid'] or ip_tid

allowed_methods = self.application.tid_auth_methods(tid, entry_mode)

if ((is_pnr or

config.getboolean('authentication', 'get_ip_proxy_protocol')) and

'ip' not in self.application.tid_auth_methods(ip_tid, entry_mode)):

# can trust IP (hence, ip_tid) only in case proxy told us the IP

# (is_pnr), or PROXY protocol is in use (client_ip is always the real

# peer).

# if not is_pnr, cannot enforce IP range restriction, as it may be

# a through authentication with no ip param, so proxy's IP is seen

allowed_methods = allowed_methods - {'ip'}

if expired_user:

auth_method = self.user_auth_method(expired_user)

if auth_method in allowed_methods:

# If the user is known, then it is a reauth.

# Use the same method, if it is still allowed.

allowed_methods = {auth_method}

login_prompt = 'udb' in allowed_methods or 'ldap' in allowed_methods

saml_prompt = 'saml' in allowed_methods

other_methods = bool(allowed_methods - {'udb', 'ldap', 'saml'})

# 's' is the indicator that UDB is probably needed

req_dict['s'] = '1' if login_prompt else ''

self.username = req_dict.get('username', None)

if req_dict.get('prepend', None) and entry_mode == 'web':

# Only show direct link for web entry mode prepended URLs from clients

# not identifiable as being associated with a known tenant

req_dict['direct_link'] = tid == -1 or ''

through_success = False

if req_dict.get('through') or other_methods:

# Through auth requested or auth methods for the known tid

# allow authentication without user interaction.

# We need to install the auth cookie on multiple hosts (XHR and

# resource hosts). We use CORS XHRs to install cookies on each.

through_res = self.do_through_authentication(req_dict, tid,

allowed_methods)

req_dict['through'] = json.dumps(through_res)

through_success = through_res['success']

if not through_success and saml_prompt and not login_prompt:

return self.try_redirect_idp(req_dict)

if not through_success:

# Ensure that through_success only has through-auth-success logged

# (already logged by do_through_authentication).

# Although login.js is loaded, the prompt should not be rendered

self.log.info({'direct_link': req_dict.get('direct_link', '')},

event='auth-prompt')

self.render(www('templates/login.template'), req_dict=req_dict)

def resp_redirect(self, pnr=False):

resp = {'success': True}

resp.update(self.generate_redirect_info())

# Are we done installing the auth cookie on cookied hosts?

if pnr and not resp.get('redirect'):

if pnr == '2':

pnr_token = generate_v2_pnr_token(self.user)

else:

pnr_token = generate_v1_pnr_token(self.user)

resp['pnr_token'] = pnr_token

return resp

def generate_redirect_info(self):

"""Return a dict used to authenticate to the next host in redirect_list.

The dict contains 'token', a signed token than can only be used to

authenticate the specific host for a short period of time

(MAX_TOKEN_AGE). It also contains 'redirect', the next host, and

'redirect_list', a comma-separated list of the remaining hosts.

"""

if not self.redirect_list:

return {}

next_host = self.redirect_list.pop()

token = base64.urlsafe_b64encode(self.create_signed_value(

'redirect_token', json.dumps({'bid': self.user['bid'],

'uid': self.user['uid'],

'tid': self.user['tid'],

'f': self.user.get('f',0),

'host': next_host})))

res = {'redirect': next_host, 'token': token}

if self.redirect_list:

res['redirect_list'] = ','.join(self.redirect_list)

return res

def _tid_by_uid(self, user_id):

"""Determine user's tenant ID based on the user ID or HTTP header"""

tid_header_value = self.request.headers.get(

config.get('authentication', 'tenant_id_header'), None)

return tid_header_value or -1

def _cookie_refresh(self):

"""As part of through authentication, look deep to find a pre-existing

cookie."""

self.user = self.get_current_user()

# The user may have a cookie, but it may be more than 31 days old and in

# need of refreshing. Retain the time of issue timestamp, if any is

# present, to make sure the re-authentication can be triggered.

if not self.user:

self.user = self.get_current_user(max_age_days=365)

if self.user:

self.set_current_user(self.user['uid'], self.user['tid'],

self.user['bid'], self.user.get('f',0),

self.user.get('t'))

self.log.info({}, event='cookie-refresh')

# Does the user have an auth cookie using an older name? If so,

# migrate it to the new name. This is needed to support the seamless

# transition of users from non-domain to domain auth cookie.

if not self.user and config.get('service', 'domain_scheme') == '4-level':

self.user = self.migrate_preexisting_cookie_if_any()

if self.user:

self.log.info({}, event='cookie-migrated')

def do_through_authentication(self, req_dict, tid, allowed_methods):

"""Through authentication.

This is the method to establish user identity without user interaction -

based on _cookie_refresh or other factors available from the browser - IP

address, headers, etc

"""

self.req_dict = req_dict

entry_mode = req_dict.get('entry_mode') or 'web'

auth_method = self.user and self.user_auth_method(self.user)

reauth = False

if self.user and auth_method not in allowed_methods:

self.expired_user = self.user

self.user = None

self.log.info({}, event='auth-method-change')

# Support for user ID by header (e.g. X-Authenticated-User)

if ((not self.user or self.is_user_anonymous_by_header(self.user)) and

'proxy_header' in allowed_methods and

'user' in req_dict):

user_id = req_dict.get('user') and decode_user_hmac(req_dict.get('user')) or None

if user_id or not req_dict.get('user'):

x_auth = self.request.headers.get(config.get('authentication',

'user_id_header'))

if x_auth:

# Use-case: pnr-icap-auth sometimes cannot see

# X-Authenticated-User, but auth-server can. pnr-icap-auth

# still forwards user=..., but user_id there is "unknown", so

# need to override user_id from that param with user_id seen

# in X-Authenticated-User.

user_id = x_auth # Assuming x_auth seen here is same or better

# than user_id seen by pnr-icap-auth

user_flags = self.USER_FLAGS_ID_BY_HEADER

tid_from_header = self._tid_by_uid(user_id)

if tid_from_header != -1:

user_flags |= self.USER_FLAGS_TENANT_ID_BY_HEADER

self.user = (user_id and self.set_current_user(user_id,

tid_from_header,

flags=user_flags)

) or self.set_anon_user(tid_from_header,

flags=user_flags,

prefix='xau')

self.username = self.user['uid']

self.log.info({}, event='auth-header')

else:

self.log.warning({'user': req_dict.get('user')},

event='auth-invalid-user-hmac', opslog=True)

elif self.user:

reauth = self._is_reauth_required(entry_mode == 'web',

timestamp=self.user['t'])

# If self.user, then auth_method is in allowed_methods; so if it's

# non-interactive, then refresh the cookie timestamp.

if reauth and auth_method in {'ip', 'proxy_header', 'anonymous'}:

reauth = False

self.user = self.set_current_user(self.user['uid'],

self.user['tid'],

flags=self.user['f'])

self.username = self.user['uid']

self.log.info({}, event='auth-method-confirmed')

# The user may be coming from an IP address that allows anonymous access.

elif 'ip' in allowed_methods:

# tid and allowed_methods based on IP or pre-existing cookie.

# So there is a narrow case when auth method is switched to IP-based,

# and a user presents a cookie with the right tid, but from a wrong IP:

# in this case still sets the cookie with the old tid, but not based on

# IP.

self.user = self.set_anon_user(tid, is_ip=True)

self.username = self.user['uid']

self.log.info({}, event='auth-ip-anon')

# Support for anonymous mode

elif 'anonymous' in allowed_methods:

# allowed_methods can contain anonymous only on-prem, hence tid=-1

# assumed below

self.user = self.set_anon_user(-1)

self.username = self.user['uid']

self.log.info({}, event='auth-anon')

# It's possible there is no (valid) cookie for the host.

if not self.user or reauth:

self.log.debug('Through authentication not possible - no cookie, or '

'reauth required',

event='failed-through-login')

if not req_dict.get('pnr'):

# If we get here via SV, it is an indication that something might

# be wrong. This does not constitute an error, as this will be the

# case when the user is logging in first time.

self.log.info({}, event='failed-sv-through-login')

return {'success': False}

self.log.debug('Through authentication attempt.')

# Detect third party cookie blocking. Do not do this for PNR.

if (not req_dict.get('pnr') and

self.application.is_client_looping(self, self.user['bid'])):

self.log.warning({}, event='auth-third-party-cookie-block',

opslog=True)

return {'success': False, 'looping': True}

if not self.redirect_list:

self.redirect_list = self.domain_scheme.get_cookied_hosts()

self.set_csp(self.redirect_list)

resp = self.resp_redirect()

self.log.info({}, event='through-auth-success')

return resp

def prepare_redirect_list(self, r_list):

self.redirect_list = []

if not r_list:

return

try:

redirect_list = r_list.split(',')

domain = '.%s' % self.domain_scheme.get_trust_domain()

# Prevent this mechanism from being used to set cookies on weird

# domains.

if any([not h.endswith(domain) for h in redirect_list]):

raise Exception('Invalid domain')

self.redirect_list = redirect_list

except Exception as e:

self.log.error({'error': e,

'redirect-list': r_list},

event='auth-bad-redirect')

def try_redirect_idp(self, req_dict):

"""Using a dictionary of parameters and self.request.body, redirect to

IdP, or render login dialog"""

self.req_dict = req_dict

# assert: if ip=... is present, it originated from a trusted source:

# POST does not insert ip=..., proxy_auth and getip verify ip=...

# originated from a trusted pnr-icap-auth

client_ip = self.remote_ip(req_dict.get('ip'), bool(req_dict.get('pnr')))

# If reauth, cookie is still set - reuse tenant id

self.expired_user = self.get_current_user(max_age_days=365)

tid_for_ip = self.application.tid_for_ip(client_ip)

tid = self.expired_user['tid'] if self.expired_user else tid_for_ip

tid_conf = self.application.tenant_config.tid_config.get(tid, {})

tid_saml = tid_conf.get('saml', {})

if not tid_saml.get('enabled'):

self.log.error({'details': 'Expected SAML to be enabled',

'ip': client_ip, 'tenant_id': tid},

event='saml-disabled' if tid_conf

else 'saml-tenant-config-missing')

return self.raw_redirect('/account/login')

# is_external: True, if we can tell IP address, and IP-based tid does not

# match tid

# False, if we can tell IP address, and it matches IP-based

# tid, or we cannot tell IP address

is_external = (req_dict.get('ip', req_dict.get('prepend', False)) and

tid_for_ip != tid)

if is_external and not tid_saml.get('external'):

self.log.error({'ip': client_ip, 'tenant_id': tid},

event='saml-external-disabled')

req_dict.pop('s', None)

self.render(www('templates/login.template'), req_dict=req_dict)

return

try:

login_url = self.idp_login(tid, client_ip, req_dict, is_external)

except InvalidSamlConfig as e:

self.log.error({'tenant_id': tid, 'saml_config': tid_saml,

'detail': e.error},

event='saml-tenant-config-broken')

self.render(www('templates/login.template'), req_dict=req_dict)

return

self.log.info({'idp_url': login_url}, event='saml-redirect-idp')

return self.raw_redirect(login_url)

def idp_login(self, tid, client_ip, req_dict, external=False):

self.req_dict = req_dict

tid_saml = self.application.tenant_config.tid_config.get(tid).get('saml')

entry_mode = req_dict.get('entry_mode', 'web')

saml_auth = SAMLAuth(self.request, self.domain_scheme.get_login_host(),

tid_saml, acs_host=self._get_login_host())

# Remove through= flag; it will be set on POST

state = dict(req_dict)

state.pop('s', None)

state.pop('through', None)

state.pop('username', None)

relay_state = encode_relay_state(tid,

urlencode(state),

entry_mode)

login_url = saml_auth.get_login_url(relay_state)

self.log.info({'ip': client_ip, 'relay_state': relay_state,

'login_url': login_url, 'tenant_id': tid},

event='external-saml-auth-initiated' if external

else 'saml-auth-initiated')

if not config.getboolean('authentication', 'saml_use_relay_state'):

self.set_secure_cookie('_sc_tid', str(tid), expires_days=0.04,

secure=True, httponly=True)

return login_url

def unexpected_redirect(self):

self.set_status(httplib.FORBIDDEN)

self.render_error(header=self.locale.translate('ID_UNEXPECTED_REDIRECT'),

byline=self.locale.translate('ID_UNEXPECTED_AUTH_FLOW'))

class LoginHandler(BaseAuthHandler):

# 86400 seconds is a day, so this is 600 seconds as a fraction of a day.

MAX_TOKEN_AGE = 1.0 / 86400 * 600

def prepare(self, *args, **kwargs):

# gets one argument based on url spec 3rd arg - OK to disable warning.

# pylint: disable=arguments-differ

super(LoginHandler, self).prepare(*args, **kwargs)

# Overriding set_default_headers is not possible - it is invoked before

# constructor has finished, so domain_scheme is not available

def _set_default_headers(self):

self.set_cors_headers(self.request.headers.get(

'Origin',

'https://' + self.domain_scheme.get_login_host()),

self.domain_scheme.get_ok_origins())

def do_auth_with_token(self, req_dict):

"""Token base authentication.

The client is providing a token. This happens to authenticate the XHR

host after the main host is authenticated.

"""

self.req_dict = req_dict

self._set_default_headers()

pnr = req_dict.get('pnr')

try:

token_val = base64.urlsafe_b64decode(str(req_dict['token']))

token_val = self.get_secure_cookie('redirect_token',

value=token_val,

max_age_days=self.MAX_TOKEN_AGE)

if not token_val:

raise ValueError('Failed to parse token.')

token_json = json.loads(token_val)

bid = token_json['bid']

uid = token_json['uid']

tid = token_json['tid']

host = token_json['host']

flags = token_json.get('f',0)

except (KeyError, ValueError, TypeError) as e:

self.log.error({'details': 'bad token',

'error': e, 'token': req_dict['token']},

event='auth-token-auth-failed', opslog=True)

self.finish_with_msg(httplib.BAD_REQUEST, 'ID_ALERT_COULD_NOT_PARSE')

return

if host != self.request.host:

self.log.error({'details': 'token host invalid',

'host': host},

event='auth-token-auth-failed', opslog=True)

self.finish_with_msg(httplib.BAD_REQUEST, 'ID_ALERT_COULD_NOT_PARSE')

return

self.user = self.set_current_user(uid, tid, bid, flags)

# Authenticate to the next host on the list.

# [LL]: "Do not call @do_through_authentication since that'll trigger the

# third-party cookie blocking detection logic."

resp = self.resp_redirect(pnr=pnr)

self.log.info({}, event='token-auth-success')

self.finish_with_json(httplib.OK, resp)

@tornado.gen.coroutine

def do_password_authentication(self, req_dict):

"""Username:password authentication"""

self.req_dict = req_dict

self.username, password = req_dict['username'], req_dict['password']

self.domain = req_dict.get('domain', '')

self.log.debug('Authenticating.', event='auth-attempt')

method = config.get('authentication', 'method')

if method == 'udb':

yield self.verify_user_udb(password)

elif method == 'ldap':

yield self.verify_user_ldap(password, req_dict['domain'])

else:

# This should not happen.

self.log.error({'details': 'authentication method not supported',

'method': method},

event='auth-config-processing-failed')

self.finish_with_msg(httplib.INTERNAL_SERVER_ERROR,

'ID_ALERT_UNABLE_TO_AUTH')

@tornado.gen.coroutine

def _on_xhr_post(self, req_dict):

"""Handle login XHRs."""

if req_dict.get('token'):

self.do_auth_with_token(req_dict)

return

if req_dict.get('through'):

# A through authentication that is not part of a form submit

# (remember form handling is done before this method is called)

# that will create a token to re-cookie an alternate domain,

# using the loginhost's cookie.

# Get the current user to check current cookie validity

self.user = self.get_current_user()

if not self.user:

# Current cookie was invalid, so send 403 to the client

self.log.warning({}, event='recookie-domain-failed')

self.finish_with_json(httplib.FORBIDDEN, {'success': False})

return

# Block generation of token on everything but the login host

# by cross checking request's host.

if self.domain_scheme.get_login_host() != self.request.host:

self.log.error({'login_host': self.domain_scheme.get_login_host(),

'requested_host': self.request.host},

event='recookie-domain-incorrect')

self.finish_with_json(httplib.FORBIDDEN, {'success': False})

return

# Current cookie was valid, so create a token to use to

# re-cookie the next domain.

self._set_default_headers()

self.set_csp(self.redirect_list)

resp = self.resp_redirect()

self.log.info({}, event='recookie-domain-success')

self.finish_with_json(httplib.OK, resp)

return

yield self.do_password_authentication(req_dict)

@tornado.gen.coroutine

def verify_user_udb(self, password):

response = yield self.application.make_udb_rest_call(

'/api/user/login',

{'username': self.username, 'password': password})

resp_dict = self.check_udb_response(response, 'Authentication',

'ID_ALERT_AUTH_FAILED', httplib.FORBIDDEN)

if resp_dict is None:

self.log_auth_status('Authentication failed', {}, event='auth-failed')

return

userId = resp_dict.get('userId', None)

# This is a bit of paranoia. Shouldn't happen. We normalize to lower case

# because the udb is case insensitive.

if userId.lower() != self.username.lower():

msg = 'Authentication failed: userId mismatch: %s !?' % userId

self.log_auth_status(msg,

{'details': 'user_id mismatch',

'user_id': userId}, event='auth-failed')

self.finish_with_msg(httplib.FORBIDDEN,

'ID_ALERT_UNABLE_TO_VERIFY_CREDS')

return

# We use the case returned by the UDB for the username.

self.username = userId

self.tenant_id = int(resp_dict.get('tid', -1))

self.complete_password_authentication(self.USER_FLAGS_UDB_AUTH)

@tornado.gen.coroutine

def verify_user_ldap(self, password, domain):

cb_res = yield tornado.gen.Task(ldap_auth.verify_user,

self.username, password, domain)

# cb_res is either tornado.gen.Arguments, or the literal value, if

# callback received just one argument

user, = getattr(cb_res, 'args', (cb_res,))

if user:

# We use the LDAP DN as the username.

self.username = user

self.complete_password_authentication(self.USER_FLAGS_LDAP_AUTH)

return

kwargs = getattr(cb_res, 'kwargs', {})

error = kwargs.get('error', 'ID_ALERT_UNABLE_TO_VERIFY_CREDS')

i18n_error = self._en_locale.translate(error)

msg = 'Authentication failed: %s' % i18n_error

self.log_auth_status(msg, {'details': 'ldap',

'error': i18n_error},

event='auth-failed')

self.finish_with_msg(httplib.FORBIDDEN, error)

def complete_password_authentication(self, flags):

self.user = self.set_current_user(self.username, self.tenant_id,

flags = flags)

self.application.add_authorized_user(self.username)

# If the client is accessing the service with an IP address, the XHR

# host is not used and may not be setup, so do not redirect there.

if tornado.netutil.is_valid_ip(self.request.host.split(':')[0]):

self.log_auth_status('Successful authentication.', {},

event='auth-success')

self.finish_with_json(httplib.OK, {'success': True})

return

if not self.redirect_list:

self.redirect_list = self.domain_scheme.get_cookied_hosts()

self.set_csp(self.redirect_list)

resp = self.resp_redirect()

self.log_auth_status('Successful authentication.', {},

event='auth-success')

self.finish_with_json(httplib.OK, resp)

def log_auth_status(self, msg, details, event):

event_type = 'user_auth_failed'

if event == 'auth-success':

self.log.info(details, event=event)

event_type = 'user_auth_succeeded'

else:

self.log.info(details, event=event, opslog=True)

report_args = {}

report_args['user_id'] = self.username

report_args['client_ip'] = self.request.remote_ip

report_args['ldap_domain'] = self.domain

report_args['details'] = msg

report_args['event'] = event

self.application.report(event_type, **report_args)

def get(self):

req_query = query_to_dict(self.request.query)

# GET allowed only over HTTPS, and only when PROXY protocol is not used

# and ip= or user= param is required, and v= is present, or when no params

# have been passed

if (not self.ok_proto() or

req_query and (

config.getboolean('authentication', 'get_ip_proxy_protocol') or

not (config.getboolean('authentication', 'require_ip_param') or

config.get('authentication', 'user_id_header')) or

not req_query.get('v'))):

self.log.error({'details': 'unsupported GET request to auth server',

'query_v': req_query.get('v'),

'query_ts': req_query.get('ts'),

'query_p': req_query.get('p')},

event='auth-bad-request')

self.unexpected_redirect()

return

self.request.body = self.request.query

# assert: try_login trusts that request either came over POST, or was

# validated by proxy_auth, or contains a signed v= (that is, if v= is

# present, ts= and p= are expected to form a signed token)

self._on_form_post(req_query)

def _on_form_post(self, req_dict):

self.prepare_redirect_list(req_dict.get('redirect_list'))

if 'ip' in req_dict and 'v' not in req_dict:

# If 'v' is not present, possibly 2.49 is talking to us.

# Overwrite 'ip' with the decoded value, because try_login trusts

# the caller to have validated the HMACed value.

req_dict['ip'] = self.decode_ip_hmac(req_dict.get('ip'),

bool(req_dict.get('pnr')))

self.try_login(req_dict)

@tornado.gen.coroutine

def post(self):

"""Handle auth XHRs and FORM POST."""

if self.request.query == 'form':

if not self.ok_proto():

self.unexpected_redirect()

return

try:

self.req_dict = query_to_dict(self.request.body)

except ValueError as e:

self.log.error({'details': 'could not parse request', 'error': e,

'request_content_type': self.request.headers['content-type']},

event='auth-bad-request', opslog=True)

self.render_error(header=self.locale.translate('ID_BAD_REQUEST'),

byline=self.locale.translate('ID_ALERT_COULD_NOT_PARSE'))

return

allowed_params = {'prepend', 'through', 'redirect_list', 'entry_mode',

'url'}

try:

extra_keys = set(self.req_dict.keys()) - allowed_params

if extra_keys:

self.log.error({'details': 'Dropping untrusted params',

'url': self.request.uri,

'extra_keys': extra_keys,

'present_args': self.req_dict.keys()},

event='auth-bad-request', opslog=True)

for k in extra_keys:

self.req_dict.pop(k)

self._on_form_post(self.req_dict)

except KeyError as e:

self.log.error({'details': 'request missing argument',

'error': e, 'url': self.request.uri,

'present_args': self.req_dict.keys()},

event='auth-bad-request', opslog=True)

self.render_error(header=self.locale.translate('ID_BAD_REQUEST'),

byline=self.locale.translate('ID_ALERT_COULD_NOT_PARSE'))

return

yield super(LoginHandler, self).post()

class RegisterHandler(BaseTenantHandler):

@tornado.gen.coroutine

def _xhr_post(self, req_dict):

"""Handle registration XHRs."""

self.username = req_dict['username']

self.log.info({'firstname': req_dict['firstname'],

'lastname': req_dict['lastname']},

event='registration-attempt')

response = yield self.application.make_udb_rest_call(

'/api/user/create',

{'username': self.username, 'firstname': req_dict['firstname'],

'lastname': req_dict['lastname'], 'password': req_dict['password'],

'service_url': self.request.host})

if self.check_udb_response(

response, 'Registration', 'ID_ALERT_REGISTRATION_FAILED'

) is not None:

self.log.info({'udb_response': response.body},

event='registration-success')

self.finish_with_json(httplib.OK, {'success': True})

# else: check_udb_response has logged and sent a response

def get(self):

self.render(www('templates/register.template'), req_dict={})

class ConfirmHandler(BaseTenantHandler):

@tornado.gen.coroutine

def _xhr_post(self, req_dict):

"""Handle user confirmation."""

response = yield self.application.make_udb_rest_call(

'/api/user/confirm/%s' % req_dict['token'], {}, 'GET')

if self.check_udb_response(

response, 'Confirm', 'ID_ALERT_CONFIRM_FAILED') is not None:

self.log.info({'udb_response': response.body}, event='confirm-success')

self.finish_with_json(httplib.OK, {'success': True})

# else: check_udb_response has logged and sent a response

def get(self):

self.render(www('templates/confirm.template'))

class SetHandler(BaseTenantHandler):

@tornado.gen.coroutine

def _xhr_post(self, req_dict):

"""Handle password setting XHRs."""

response = yield self.application.make_udb_rest_call(

'/api/user/setpw',

{'password': req_dict['password'], 'token': req_dict['token']})

if self.check_udb_response(

response, 'Set password', 'ID_ALERT_SET_PW_FAILED') is not None:

self.log.info({}, event='set-password-success')

self.finish_with_json(httplib.OK, {'success': True})

# else: check_udb_response has logged and sent a response

def get(self):

self.render(www('templates/set_password.template'))

class DeniedHandler(CSPHandler):

def __init__(self, *args, **kwargs):

super(DeniedHandler, self).__init__(*args, **kwargs)

self.add_header('Vary','Accept-Language')

def get(self):

svc_title = config.get('customization', 'title')

self.render(www('templates/denied.template'), svc_title=svc_title)

class ResetHandler(BaseTenantHandler):

@tornado.gen.coroutine

def _xhr_post(self, req_dict):

"""Handle password reset XHRs."""

self.username = req_dict['username']

response = yield self.application.make_udb_rest_call(

'/api/user/resetpw', {'username': self.username})

# Don't use check_udb_response - we don't want this to be a way to find

# out if a user is valid.

if response.error or response.code != httplib.OK:

self.log.error({'code': response.code, 'error': response.error,

'udb_response': response.body},

event='auth-password-reset-failed', opslog=True)

self.finish_with_msg(httplib.INTERNAL_SERVER_ERROR,

'ID_ALERT_PW_RESET_FAILED')

return

try:

resp_dict = json.loads(response.body)

result = resp_dict['result']

reason = resp_dict.get('reason', None)

except (ValueError, KeyError) as e:

self.log.error({'details': 'Could not parse response',

'error': e, 'udb_response': response.body},

event='auth-password-reset-failed', opslog=True)

self.finish_with_msg(httplib.INTERNAL_SERVER_ERROR,

'ID_ALERT_PW_RESET_FAILED')

return

if result != 0:

self.log.error({'details': 'non-zero result',

'result': result, 'reason': reason},

event='auth-password-reset-failed', opslog=True)

# This is not a known user. We do not want to disclose this, so return

# success.

else:

self.log.info({'udb_response': response.body},

event='password-reset-success')

self.finish_with_json(httplib.OK, {'success': True})

def get(self):

self.render(www('templates/reset.template'))

class ProxyAuthHandler(BaseAuthHandler):

HTTP_OK=True

@tornado.gen.coroutine

def _on_xhr_post(self, req_dict):

self.unexpected_redirect()

@staticmethod

def set_query_parameter(url, param_name, param_value):

"""Given a URL, set or replace a query parameter and return the modified

URL."""

scheme, netloc, path, query_string, fragment = urlparse.urlsplit(url)

new_query_string = '%s=%s%s%s' % (param_name, quote(param_value,

safe='~()*!.\''),

'&' if query_string else '',

query_string)

return urlparse.urlunsplit((scheme, netloc, path, new_query_string,

fragment))

def _is_saml_reauth_possible(self, client_ip):

"""Check if SAML reauth trigger is possible.

Checks if the client IP is in the gateway list or if SAML external is set

to True.

"""

return self.application.tenant_config.is_saml_reauth_possible(

client_ip, self.user['tid'])

def get(self):

# pylint: disable=too-many-branches

if not self.ok_proto():

self.log.error({'details': 'communication with auth server over http '

'not allowed.'},

event='auth-bad-request')

self.unexpected_redirect()

return

self.request.body = self.request.query

self.req_dict = query_to_dict(self.request.query)

req_dict = self.req_dict

self.user = self.get_current_user()

if 'v' in req_dict:

params = self.signed_params(**req_dict)

client_ip = self.remote_ip(params.get('ip'), True)

else:

client_ip = self.decode_ip_hmac(req_dict.get('ip'),

bool(req_dict.get('pnr')))

b64_url = req_dict.get('url') + '=='

params = {

'saml': 1 if req_dict.get('saml', '0') in ['true', '1'] else 0,

'url': base64.urlsafe_b64decode(b64_url.encode('utf-8')),

'ip': client_ip

}

if not client_ip or not params:

self.log.error({'details': 'proxy_auth',

'query_v': req_dict.get('v'),

'query_ts': req_dict.get('ts'),

'query_p': req_dict.get('p')},

event='auth-invalid-params-hmac')

self.raw_redirect('/account/login')

return

# trust signed token is from a cooperating pnr-icap-auth

req_dict.update(params)

dest_url = req_dict['url']

use_saml = req_dict['saml'] == 1

trigger_reauth = False

encoded_user_id = req_dict.get('user', None)

pnr_ver = req_dict['pnr']

if self.user:

tid = self.user['tid']

else:

tid = self.application.tid_for_ip(client_ip)

allowed_methods = self.application.tid_auth_methods(tid, 'web')

# assert self.user and allowed_methods is based on real tid_conf

# or not self.user and allowed_methods includes UDB and SAML to permit

# auth based on PAC file - including external SAML auth, which will do the

# allowed_methods check after tid is known

if self.user:

if self._is_reauth_required():

trigger_reauth = True

self.log.info({'time_since_last_auth': self.time_since_last_auth()},

event='cookie-reauth-triggered')

# Seen cases when pnr-icap-auth does not observe x-auth headers on some

# URLs, but later observes x-auth headers on other URLs. So, to reduce

# impact of having "xau_*" user_id, force through authentication

# again, if observing a cookie with anonymous user, but pnr-icap-auth

# saw a user_id

elif (encoded_user_id and

self.is_user_anonymous_by_header(self.user) and

config.get('authentication', 'user_id_header')):

trigger_reauth = True

self.log.info({'last_auth_at': self.user.get('t', 0)},

event='proxy-unknown-reauth-triggered')

elif self.user_auth_method(self.user) not in allowed_methods:

trigger_reauth = True

self.log.info({}, event='proxy-auth-method-reauth')

# It's possible there is no (valid) cookie for the host.

if not self.user or trigger_reauth:

self.log.debug('ProxyAuth authentication requires login - no cookie')

# PAC coercion: only possible on reauth; if tid is not known, both SAML

# and UDB are allowed, and selected by using a PAC for

# the right port

# - if user has a PAC for UDB port, but SAML is the only method - do

# SAML auth anyway

# - if user has a PAC for SAML port, but SAML is not enabled - do UDB

# or through auth anyway

if (use_saml and 'saml' in allowed_methods or

allowed_methods == {'saml'}):

# Check if we have got a reauth trigger, or if user is

# using saml squid and ip is recognized in which case

# redirect the user to perform saml auth

if trigger_reauth or self.application.ip_allow_saml(client_ip):

self.log.info({}, event='proxy-saml-auth')

self.try_redirect_idp({'url': dest_url, 'ip': client_ip})

return

# if user is outside of office, redirect them to page to lookup

# tenant config via email

self.log.info({}, event='proxy-saml-prompt')

self.render(www('templates/login.template'),

req_dict={'url': dest_url})

else:

self.log.info({}, event='proxy-udb-prompt')

# if user is not using saml, redirect them to regular udb login

self.try_login(

{'pnr': pnr_ver,

'url': dest_url,

# Setting through, so IP-based authentication or tid

# determination can happen.

# getip redirect seen in try_login kicks in, if client_ip

# or encoded_user_id are not known, but required in INI file.

'through': 'true',

'ip': client_ip,

'user': encoded_user_id})

return

self.log.debug(

'ProxyAuth received valid cookie - generating sc_token for %s',

dest_url)

if pnr_ver == '2':

sc_token = generate_v2_pnr_token(self.user)

else:

sc_token = generate_v1_pnr_token(self.user)

dest_url = self.set_query_parameter(dest_url, '_sc_token',

quote(sc_token, safe='~()*!.\''))

self.log.info({}, event='proxy-success')

self.raw_redirect(dest_url, httplib.TEMPORARY_REDIRECT)

# pylint: enable=too-many-branches

def post(self):

"""Both POST and GET for proxy_auth use query part of request to determine

what to do next. Both produce 307, if the user is already authenticated,

so browser resends POST body on redirect."""

self.get()

class SamlAuthHandler(BaseAuthHandler):

"""Handles SAML SSO Authentication"""

# This is not used, because post() is overridden

def _on_xhr_post(self, req_dict):

pass

@tornado.gen.coroutine

def post(self):

"""Fetch the tenant id from the cookie or RelayState and lookup the saml

configuration.

If the tenant has saml group capabilities enabled, then push the group

data to the pnr-policy server.

"""

try:

relay_state = self.decode_relay_state()

tid = relay_state['tid']

# NOTE: When RelayState is not used to pass state around

# (authentication.saml_use_relay_state == False), entry_mode is not

# available

entry_mode = relay_state.get('entry_mode', 'web')

except Exception:

# Already logged

self.set_status(403)

self.render_error(

header=self.locale.translate('ID_SAML_AUTH_REJECTED'),

byline=self.locale.translate('ID_SAML_AUTH_REJECT_REASON') %

{'reason': 'RelayState was not found in POST '

'body, or RelayState is corrupted'},

is_alert=True, is_raw=True)

return

tid_conf = self.application.tenant_config.tid_config.get(tid)

if not tid_conf:

self.log.error({'details': 'Tenant config was expected to still exist',

'tenant_id': tid},

event='saml-tenant-config-missing')

self.raw_redirect('/account/login')

return

try:

saml_auth = SAMLAuth(self.request,

self.domain_scheme.get_login_host(),

tid_conf.get('saml', {}),

acs_host=self._get_login_host())

saml_auth.process_response()

except (InvalidSamlConfig, InvalidSamlResponse) as e:

error = getattr(e, 'error', None) or e.message

self.log.error({'error': error, 'tenant_id': tid,

'saml_response': self.request.body},

event='saml-response-error')

self.set_status(401)

self.render_error(

header=self.locale.translate('ID_SAML_AUTH_REJECTED'),

byline=self.locale.translate('ID_SAML_AUTH_REJECT_REASON') %

{'reason': tornado.escape.xhtml_escape(e.message)},

is_alert=True, is_raw=True)

return

if not config.getboolean('authentication', 'saml_use_relay_state'):

relay_state['url'] = saml_auth.get_relay_url()

# reuse bid if they have one

self.user = self.get_current_user()

bid = self.user['bid'] if self.user else None

flags = self.USER_FLAGS_SAML_AUTH

trace = []

# update/create safeview-id cookie, reset the timestamp

saml_user = saml_auth.get_userid()

if self.application.tenant_config.should_push_groups(tid_conf,

trace.append):

yield self.application.tenant_config.push_group_capabilities(

saml_auth, tid_conf, tid, trace.append)

validate_trace = []

if entry_mode == 'web' and not (

yield self.validate_user_saml(saml_auth.get_userid(), tid,

validate_trace.append,

saml_auth=saml_auth, tid_conf=tid_conf)):

self.log.info({'reason': 'Invalid group or custom attribute',

'user_id': saml_user, 'tenant_id': tid,

'trace': json.dumps(trace + validate_trace),

'saml_response': self.request.body},

event='saml-auth-failed')

self.set_status(httplib.FORBIDDEN)

self.render(www('templates/saml_auth_failed.template'))

return

self.user = self.set_current_user(saml_user, tid, bid, flags)

req_dict = {k: v for k, v in urlparse.parse_qsl(relay_state['url'])}

self.prepare_redirect_list(req_dict.get('redirect_list'))

if not self.redirect_list:

self.redirect_list = self.domain_scheme.get_cookied_hosts()

self.set_csp(self.redirect_list)

req_dict['through'] = json.dumps(self.resp_redirect())

if time.time() < float(tid_conf['saml'].get('verbose_till', '-Inf')):

req_dict['trace'] = json.dumps(trace)

else:

req_dict['trace'] = ''

self.log.info({'uid': self.user['uid'], 'tid': self.user['tid']},

event='saml-auth-success')

self.render(www('templates/login.template'), req_dict=req_dict)

def decode_relay_state(self):

"""Fetch tid from cookie or RelayState"""

try:

if config.getboolean('authentication', 'saml_use_relay_state'):

relay_state = ('RelayState' in self.request.body_arguments and

self.request.body_arguments['RelayState'][0])

if not relay_state:

loggable = {'saml_response': self.request.body}

raise Exception('Expecting RelayState in saml_response')

loggable = {'relay_state': relay_state}

rs = decrypt_hmac_json(relay_state)

loggable.update(rs)

self.log.info(loggable, event='decode-relay-state')

if 'tid' not in rs:

raise Exception('Expecting tid in RelayState')

if 'url' not in rs:

rs['url'] = ''

else:

loggable = {}

rs = {'tid': int(self.get_secure_cookie('_sc_tid', max_age_days=1))}

return rs

except Exception as e:

loggable['error'] = e

self.log.warning(loggable, event='saml-missing-tid')

if config.getboolean('authentication', 'saml_use_tid_fallback'):

return {'tid': -1}

else:

raise e

class SamlExternalHandler(BaseAuthHandler):

"""Lookup SAML endpoint by email address and redirect"""

def __init__(self, *args, **kwargs):

super(SamlExternalHandler, self).__init__(*args, **kwargs)

@tornado.gen.coroutine

def _on_xhr_post(self, req_dict):

self.username = req_dict['username']

if config.getboolean('authentication', 'use_ini_settings'):

# It is not wrong to call UDB in single-tenant deployments.

# However, in some dev scenarios UDB contains more tenants

# than tenant configs - for example, tenant config is taken

# from INI, but UDB used knows about more tenant ids.

# Assuming there is one and only one tenant id.

tid = self.application.tenant_config.tid_config.keys()[0]

else:

response = yield self.application.make_udb_rest_call(

'/api/user/lookup_tenant',

{'username': self.username})

resp_dict = self.check_udb_response(response, 'Tenant Lookup',

'ID_SAML_TENANT_FAIL',

httplib.NOT_FOUND)

if not resp_dict:

return

tid = int(resp_dict['tid'])

tid_conf = self.application.tenant_config.tid_config.get(tid)

if not tid_conf:

self.log.error({'tenant_id': tid}, event='saml-tenant-config-missing')

self.finish_with_msg(httplib.BAD_REQUEST, 'ID_SAML_NOT_CONFIG')

return

saml_conf = tid_conf.get('saml')

if not saml_conf or not saml_conf.get('enabled', False):

self.log.info({'tenant_id': tid}, event='saml-disabled')

self.finish_with_msg(httplib.BAD_REQUEST, 'ID_SAML_NOT_ENABLED')

return

if not saml_conf.get('external', False):

self.log.info({'tenant_id': tid}, event='saml-external-disabled')

self.finish_with_msg(httplib.BAD_REQUEST, 'ID_SAML_EXTERNAL_ACCESS')

return

saml_required = saml_conf.get('required', False)

try:

login_url = self.idp_login(tid, self.request.remote_ip, req_dict, True)

except InvalidSamlConfig as e:

self.log.error({'tenant_id': tid, 'saml_config': saml_conf,

'detail': e.error},

event='saml-tenant-config-broken')

self.finish_with_msg(httplib.BAD_REQUEST, 'ID_SAML_NOT_CONFIG')

return

self.finish_with_json(httplib.OK, {'success': True,

'redirect': login_url,

'saml_required': saml_required})

class AuthorizedHandler(BaseAuthHandler):

# This is internal, can happen over http.

HTTP_OK=True

@tornado.gen.coroutine

def _on_xhr_post(self, req_dict):

"""Handle Authorization check XHRs.

Use the cache if possible. Otherwise, check with the server.

"""

self.username = req_dict['username']

tid = req_dict['tid']

timestamp = req_dict['t']

entry_mode = req_dict['entry_mode']

entry_web = entry_mode == 'web'

# Backwards compatibility with 2.40:

# if flags are not set, we are dealing with old cookies where UDB vs SAML

# is distinsguished based on deficient logic

if int(req_dict.get('f', 0)) == 0 and self.tid_allow_saml(tid):

auth_mode = 'saml'

else:

auth_mode = self.user_auth_method({'uid': self.username, 'tid': tid,

'f': req_dict.get('f', 0)})

allowed_methods = self.application.tid_auth_methods(tid, entry_mode)

if (auth_mode == 'saml' and

self._is_reauth_required(entry_web, tid, timestamp) or

auth_mode not in allowed_methods and

'proxy_header' not in allowed_methods): # if 'proxy_header' is required,

# can only succeed from proxy_auth - will not reauth

self.log.info({'last_auth_at': timestamp,

'auth_method': auth_mode,

'entry_mode': entry_mode}, event='auth-trigger-reauth')

self.finish_with_json(httplib.FORBIDDEN, {'success': False})

return

self.log.debug('Skipping revocation check for %s user, recent auth, '

'auth mode: %s in %s',

entry_mode, auth_mode, allowed_methods)

if self.username in self.application.cached_authorized_users:

self.log.debug('Authorized (cached).')

self.finish_with_json(httplib.OK, {'success': True})

return

if auth_mode in {'proxy_header', 'ip', 'anonymous'}:

# The downstream proxy is responsible for validating the user

self.log.debug('Authorized (%s).' % auth_mode)

self.finish_with_json(httplib.OK, {'success': True})

return

# Handle reauth timers for SAML users

if auth_mode == 'saml':

yield self.authorize_user_saml(tid, self.username, entry_web)

elif auth_mode == 'udb':

yield self.authorize_user_udb(tid)

elif auth_mode == 'ldap':

yield self.authorize_user_ldap()

else:

# This should not happen.

self.log.error({'method': auth_mode, 'flags': req_dict.get('f', 0)},

event='auth-cookie-unknown-method')

self.finish_with_msg(httplib.INTERNAL_SERVER_ERROR,

'ID_ALERT_UNABLE_TO_AUTH')

@tornado.gen.coroutine

def authorize_user_saml(self, tid, username, entry_web):

trace = []

# Validate saml attributes

if entry_web and not (yield self.validate_user_saml(username, tid,

trace.append)):

self.log.info({'reason': 'Invalid group or custom attribute',

'trace': json.dumps(trace),

'username': username, 'tid': tid},

event='saml-auth-failed')

self.finish_with_json(httplib.FORBIDDEN, {'success': False})

return

# Cache web users to prevent querying pnr-policy every pair/re-pair

self.complete_authorization(entry_web)

@tornado.gen.coroutine

def authorize_user_udb(self, tid):

# Handle anonymous access here.

if self.tid_allow_anon(tid):

# TODO: is this even reachable now?

self.log.info({'tenant_id': tid}, event='auth-anon-enabled')

self.complete_authorization(False) # No need to cache anonymous users

return

response = yield self.application.make_udb_rest_call(

'/api/user/authorized', {'username': self.username})

if self.check_udb_response(

response, 'Authorized', 'ID_ALERT_AUTHORIZED_FAILED',

httplib.FORBIDDEN) is not None:

self.complete_authorization()

# else: check_udb_response logged and sent response

@tornado.gen.coroutine

def authorize_user_ldap(self):

cb_res = yield tornado.gen.Task(ldap_auth.authorize_user, self.username)

# cb_res is either tornado.gen.Arguments, or the literal value, if

# callback received just one argument

username, = getattr(cb_res, 'args', (cb_res,))

if username:

self.complete_authorization()

return

kwargs = getattr(cb_res, 'kwargs', {})

error = kwargs.get('error', 'ID_ALERT_NOT_AUTHORIZED')

self.finish_with_msg(httplib.FORBIDDEN, error)

def complete_authorization(self, cache=True):

self.log.debug('Authorized.', event='authorize-success')

if cache:

self.application.add_authorized_user(self.username)

self.set_status(httplib.OK)

self.finish()

def tid_allow_anon(self, tid):

conf = self.application.tenant_config.tid_config.get(tid)

if not conf:

return False

return conf.get('anonAccess', False)

def tid_allow_saml(self, tid):

conf = self.application.tenant_config.tid_config.get(tid)

if not conf or 'saml' not in conf:

return False

return conf['saml'].get('enabled', False)

class UserAttributesHandler(BaseAuthHandler):

# This is internal, can happen over http.

HTTP_OK=True

def log_attr_status(self, msg, details, event):

event_type = 'user_attrs_failed'

if event == 'user-attributes-success':

self.log.info(details, event=event)

event_type = 'user_attrs_succeeded'

else:

self.log.info(details, event=event, opslog=True)

report_args = {}

report_args['user_id'] = self.username

report_args['client_ip'] = self.request.remote_ip

report_args['ldap_domain'] = self.domain

report_args['details'] = msg

report_args['event'] = event

self.application.report(event_type, **report_args)

@tornado.gen.coroutine

def _on_xhr_post(self, req_dict):

"""Handle User Attribute queries from PnR.

Will cache for the same duration as configured for AuthorizationHandler,

but this should be a policy statement.

"""

def verify_ldap_response(attributes,

error='ID_ALERT_ATTRIBS_FAILED'):

if not attributes:

tr_msg = self._en_locale.translate(error)

msg = 'Attribute retrieval failed: %s' % tr_msg

self.log_attr_status(msg, {'error': tr_msg},

event='ldap-user-attributes-failed')

self.finish_with_msg(httplib.FORBIDDEN, error)

return

self.log_attr_status('Attributes retrieved.', {},

event='user-attributes-success')

self.application.add_user_attributes(key, attributes)

self.finish_with_json(httplib.OK,

{'success': True,

'attributes': attributes})

def verify_default_ldap(attribute_values,

error='User attributes retrieval failed.'):

# assert attributes is captured from the enclosing function

# TODO: in the spirit of the rest of the code, looking only for

# canonical spelling of the attributes

# although it should be case-insensitive.

# TODO: assuming attribute names passed to this service are spelled

# canonically

upn = (attribute_values and

attribute_values.get('userPrincipalName') or [])

fqdn = (len(upn) == 1 and

any([attribute_values.get(a, None) == None for a in attributes])

and upn[0].split('@', 1)) or []

# if upn is unusable, cannot search a non-default LDAP

# if all attributes were returned, there is no need to search

# a non-default LDAP

if len(fqdn) != 2:

verify_ldap_response(attribute_values, error)

return

self.domain = fqdn[1].split('.', 1)[0]

ldap_auth.attribute_lookup(self.username, self.domain,

attributes, verify_ldap_response)

# TODO: policy should specify the caching duration

self.username = req_dict['username']

key = self.username + req_dict.get('version', '')

# TODO: should converge to using just one cache for AuthorizationHandler

if (req_dict.get('cached', True) and

key in self.application.cached_user_attributes):

# Assuming the attributes are always the same, unless the policy

# changes.

# When the policy changes, PnR will know, and use a different version

# indicator.

# This simple versioning protocol permits to bypass the cache sooner,

# as the policy

# changes, but even without that the divergence is up to cache duration

self.log.debug('Returning cached attributes.')

self.finish_with_json(httplib.OK,

{'success': True, 'cached': True,

'attributes': self.application

.cached_user_attributes[key]

})

return

attributes = [a.encode('ascii', 'ignore')

for a in req_dict['attributes']]

if not attributes:

self.finish_with_json(httplib.OK,

{'success': True,

'attributes': {}})

return

self.domain = req_dict.get('domain', None)

if not self.domain and '\\' in self.username:

self.domain, self.username = self.username.split('\\', 1)

if self.domain:

ldap_auth.attribute_lookup(self.username, self.domain,

attributes,

verify_ldap_response)

else:

attributes.append('userPrincipalName')

ldap_auth.attribute_lookup(self.username, None,

attributes,

verify_default_ldap)

# TODO: LoginInfoHandler outages can cause strange behaviors - like, switching

# to UDB login. Need to do something more clever - retry or maybe render as part

# of login template.

class LoginInfoHandler(BaseAuthHandler):

@tornado.gen.coroutine

def _on_xhr_post(self, req_dict):

"""Return general login info to client."""

iface = config.get('service', 'inbound_iface')

res = {'version': get_build_id(),

'ip': netifaces.ifaddresses(iface)[netifaces.AF_INET][0]['addr'],

'service_domain': config.get('service', 'service_host'),

'hostname': socket.gethostname()}

# If the auth method is ldap, we get the domain info, unless anonymous

# mode is enabled.

if (config.get('authentication', 'method') == 'ldap' and

not config.getboolean('authentication', 'allow_anonymous')):

domains = {}

try:

domains = ldap_auth.get_domain_dict()

except Exception as e:

self.log.error({'error': e}, event='ldap-get-domains-failed')

if domains:

res.update({'domains': domains,

'default_domain': config.get('authentication',

'ldap_default_domain')})

self.finish_with_json(httplib.OK, res)

class ThirdPartyCookieHandler(BaseHandler):

def get(self):

# Get the user if available to improve log message.

self.user = self.get_current_user()

self.log.info({}, event='third-party-cookie-page')

self.render(www('templates/third_party_cookie.template'))

class AuthApplication(SafelyApplication):

"""Authentication application."""

# Threshold for third party cookie blocking detection, in seconds.

LOOPING_THRESHOLD = 20

# Number of consecutive failures that must happen for the same bid within

# LOOPING_THRESHOLD seconds to trigger the redirect to the third party

# cookie blocking error page.

LOOPING_ATTEMPT_THRESHOLD = 5

def __init__(self, *args, **kwargs):

super(AuthApplication, self).__init__(*args, **kwargs)

# Last through attempt - we keep track of this to detect clients that

# keep retrying because they block third party cookies (the cookie is

# missing on a host used by the TC even though the client has just gone

# through the through login procedure). Keeping track of just one attempt

# is crude, we might want to refine this if this does not perform well.

self._last_attempt_bid = None

self._last_attempt_ts = None

self._failed_count = 0

# Map of authorized users.

self.cached_authorized_users = {}

self.cached_user_attributes = {}

self._reporter = EventReporter.instance()

# Get the tenant settings from ini/pnr-policy

if config.getboolean('authentication', 'use_ini_settings'):

self.tenant_config = TenantConfigIni()

else:

self.tenant_config = TenantConfigPnr()

self.tenant_config.update_config()

if config.get('authentication', 'method') == 'udb':

self.client_key = config.get('authentication', 'client_key')

self.client_cert = config.get('authentication', 'client_cert')

def report(self, event_type, **kwargs):

self._reporter.report(event_type, kwargs)

def add_authorized_user(self, username):

"""Add a user to the cache and schedule removal."""

self._add_to_cache(self.cached_authorized_users, username, True)

def add_user_attributes(self, username, attributes):

self._add_to_cache(self.cached_user_attributes, username, attributes)

def _add_to_cache(self, cache, key, value):

"""Add a key and value to the cache and schedule removal."""

validity_period = config.getfloat('authentication', 'validity_period')

if not validity_period:

return

# This can happen if multiple calls occur almost simultaneously. In

# this case, we just ignore timeout settings. We don't ignore cache value

# update: on the one hand, in concurrent cases the value should be the

# same, and the assignment operation is cheap, on the other hand, if the

# cache is updated deliberately (e.g., cached=False), it is better to keep

# the latest value.

if key in cache:

cache[key] = value

return

cache[key] = value

# schedule removal.

# note we now use monotonic timer instead of unix time

tornado.ioloop.IOLoop.instance().call_later(

validity_period,

lambda: cache.pop(key))

def is_client_looping(self, handler, bid):

now = time.time()

res = False

if (self._last_attempt_bid == bid and

(now - self._last_attempt_ts < self.LOOPING_THRESHOLD)):

self._failed_count += 1

if self._failed_count >= self.LOOPING_ATTEMPT_THRESHOLD:

res = True

else:

# This is not necessarily indicative of a problem. Several tabs may

# be reconnecting almost simultaneously after an update.

handler.log.info('Possible failed through login attempt.',

event='failed-through-login')

else:

self._last_attempt_bid = bid

self._last_attempt_ts = now

self._failed_count = 0

return res

def make_udb_rest_call(self, api, call_dict, method='POST'):

http_client = tornado.httpclient.AsyncHTTPClient(force_instance=True)

req = tornado.httpclient.HTTPRequest(

url=config.get('authentication', 'udb_rest') + api,

method=method,

body=json.dumps(call_dict) if call_dict else None,

connect_timeout=20, request_timeout=20,

#FIXME: this is terrible. The UDB uses a self-signed cert.

validate_cert=False,

client_key=self.client_key,

client_cert=self.client_cert,

headers={'content-type': 'application/json'})

return http_client.fetch(req, raise_error=False)

def ip_allow_anon(self, ip):

conf = self.tid_conf_for_ip(ip)

if not conf:

return False

return conf.get('anonAccess', False)

def ip_allow_saml(self, ip):

conf = self.tid_conf_for_ip(ip)

if not conf or 'saml' not in conf:

return False

return conf['saml'].get('enabled', False)

def tid_allow_saml(self, tid):

conf_saml = self.tenant_config.tid_config.get(tid, {}).get('saml', {})

return conf_saml.get('enabled', False)

def tid_requires_saml(self, tid):

conf_saml = self.tenant_config.tid_config.get(tid, {}).get('saml', {})

return (conf_saml.get('required', False) and

conf_saml.get('enabled', False))

def tid_for_ip(self, ip):

return self.tenant_config.ip_to_tid.get(ip, -1)

def tid_conf_for_ip(self, ip):

tid = self.tid_for_ip(ip)

return self.tenant_config.tid_config.get(tid)

def tid_auth_methods(self, tid, entry_mode):

"""Assume that either tid is real, then tid_conf is real,

or tid is -1 based on unknown IP address, then tid_conf should

allow UDB and external SAML auth to kick in."""

tid_conf = self.tenant_config.tid_config.get(

tid, {'saml': {'enabled': True}})

saml_conf = tid_conf.get('saml', {})

return {k for k, v in [

('anonymous', config.getboolean(

'authentication',

'allow_anonymous' if entry_mode == 'web'

else 'allow_anonymous_readonly')),

('proxy_header', config.get('authentication', 'user_id_header')),

('ip', tid_conf.get('anonAccess', False)),

(config.get('authentication', 'method'),

# if SAML is not required, then allow udb or ldap

not saml_conf.get('required', False) or

not saml_conf.get('enabled', False)),

('saml', saml_conf.get('enabled', False))

] if v}

class AuthServer(SafelyTornadoApp):

"""Authentication server."""

NAME='auth-server'

URL_PATH = '/safeview-auth-server'

def __init__(self, *args, **kwargs):

super(AuthServer, self).__init__(*args, **kwargs)

tornado.locale.load_translations(os.path.join(os.path.dirname(__file__),

"locale"))

def create_tornado_app(self):

handler_args = {'args': self.args,

'en_locale': tornado.locale.get('en_US')}

_handlers = [

# Internal use only.

(r'/authorized', AuthorizedHandler, handler_args),

(r'/userattributes', UserAttributesHandler, handler_args),

# Externally visible (prefixed with /account)

(r'/account/denied', DeniedHandler),

(r'/account/login', LoginHandler, handler_args),

(r'/account/register', RegisterHandler, handler_args),

(r'/account/reset', ResetHandler, handler_args),

(r'/account/confirm', ConfirmHandler, handler_args),

(r'/account/set_password', SetHandler, handler_args),

(r'/account/third_party_cookie', ThirdPartyCookieHandler, handler_args),

(r'/account/saml', SamlExternalHandler, handler_args),

# Externally accessible (prefixed with URL_PATH)

(r'%s/confirm' % self.URL_PATH, ConfirmHandler, handler_args),

(r'%s/login' % self.URL_PATH, LoginHandler, handler_args),

(r'%s/register' % self.URL_PATH, RegisterHandler, handler_args),

(r'%s/reset_password' % self.URL_PATH, ResetHandler, handler_args),

(r'%s/set_password' % self.URL_PATH, SetHandler, handler_args),

(r'%s/login_info' % self.URL_PATH, LoginInfoHandler, handler_args),

(r'%s/proxy_auth' % self.URL_PATH, ProxyAuthHandler, handler_args),

(r'%s/saml' % self.URL_PATH, SamlAuthHandler, handler_args),

(r'%s/saml_external' % self.URL_PATH, SamlExternalHandler, handler_args),

]

return AuthApplication(_handlers, need_secret=True)

if __name__ == '__main__':

AuthServer(description=__doc__.strip()).run()

Coverage for main.py : 17%

1053 statements 182 run 871 missing 0 excluded