431 lines
18 KiB
Plaintext
431 lines
18 KiB
Plaintext
import libxml2
|
|
import urllib
|
|
import urlparse
|
|
import httplib
|
|
import re
|
|
import os
|
|
|
|
from quixote import get_field, get_request, get_response, get_session, get_session_manager, redirect
|
|
from quixote.directory import Directory
|
|
from quixote.http_request import parse_header
|
|
|
|
import lasso
|
|
|
|
from qommon import get_logger
|
|
from qommon.form import *
|
|
from qommon.template import *
|
|
|
|
import misc
|
|
import storage
|
|
from users import User
|
|
from hosts import Host
|
|
from federations import Federation
|
|
import site_authentication
|
|
|
|
class Liberty(Directory):
|
|
_q_exports = ['', 'login', 'assertionConsumer', 'soapEndpoint',
|
|
'singleLogout', 'singleLogoutReturn',
|
|
'federationTermination', 'federationTerminationReturn',
|
|
('metadata.xml', 'metadata'), 'public_key', 'local_auth']
|
|
|
|
def perform_login(self, idp = None):
|
|
server = misc.get_lasso_server()
|
|
login = lasso.Login(server)
|
|
login.initAuthnRequest(idp, lasso.HTTP_METHOD_REDIRECT)
|
|
login.request.nameIdPolicy = 'federated'
|
|
login.request.forceAuthn = False
|
|
login.request.isPassive = False
|
|
login.request.consent = 'urn:liberty:consent:obtained'
|
|
login.buildAuthnRequestMsg()
|
|
return redirect(login.msgUrl)
|
|
|
|
def assertionConsumer(self):
|
|
server = misc.get_lasso_server()
|
|
if not server:
|
|
return error_page(_('Liberty support is not yet configured'))
|
|
login = lasso.Login(server)
|
|
request = get_request()
|
|
if request.get_method() == 'GET' or get_field('LAREQ'):
|
|
if request.get_method() == 'GET':
|
|
login.initRequest(request.get_query(), lasso.HTTP_METHOD_REDIRECT)
|
|
else:
|
|
login.initRequest(get_field('LAREQ'), lasso.HTTP_METHOD_POST)
|
|
|
|
login.buildRequestMsg()
|
|
try:
|
|
soap_answer = soap_call(login.msgUrl, login.msgBody)
|
|
except SOAPException:
|
|
return error_page(_('Failure to communicate with identity provider'))
|
|
try:
|
|
login.processResponseMsg(soap_answer)
|
|
except lasso.Error, error:
|
|
if error[0] == lasso.LOGIN_ERROR_STATUS_NOT_SUCCESS:
|
|
return error_page(_('Unknown authentication failure'))
|
|
if hasattr(lasso, 'LOGIN_ERROR_UNKNOWN_PRINCIPAL'):
|
|
if error[0] == lasso.LOGIN_ERROR_UNKNOWN_PRINCIPAL:
|
|
return error_page(_('Authentication failure; unknown principal'))
|
|
return error_page(_("Identity Provider didn't accept artifact transaction."))
|
|
else:
|
|
login.processAuthnResponseMsg(get_field('LARES'))
|
|
login.acceptSso()
|
|
session = get_session()
|
|
if login.isSessionDirty:
|
|
if login.session:
|
|
session.lasso_session_dumps[server.providerId] = login.session.dump()
|
|
else:
|
|
session.lasso_session_dumps[server.providerId] = None
|
|
|
|
# Look for an existing user
|
|
user = self.lookup_user(session, login)
|
|
|
|
# Check if it is for Larpe administration or token
|
|
host = Host.get_host_from_url()
|
|
if host is None:
|
|
return redirect('%s/' % get_request().environ['SCRIPT_NAME'])
|
|
if host.name == 'larpe':
|
|
session.name_identifier = login.nameIdentifier.content
|
|
session.lasso_dump = login.identity.dump()
|
|
session.provider_id = server.providerId
|
|
|
|
if user:
|
|
session.set_user(user.id, server.providerId)
|
|
if hasattr(session, 'after_url') and session.after_url is not None and \
|
|
session.after_url.find('admin') != -1:
|
|
return redirect(session.after_url)
|
|
else:
|
|
if not user or not user.is_admin:
|
|
if hasattr(session, 'after_url') and session.after_url is not None \
|
|
and session.after_url.find('token') != -1:
|
|
return redirect(session.after_url)
|
|
return redirect('%s/token' % get_request().environ['SCRIPT_NAME'])
|
|
return redirect('%s/admin/' % get_request().environ['SCRIPT_NAME'])
|
|
|
|
|
|
# Set session user
|
|
if not user:
|
|
user = User()
|
|
user.name_identifiers = [ login.nameIdentifier.content ]
|
|
user.lasso_dumps = [ login.identity.dump() ]
|
|
user.store()
|
|
session.set_user(user.id, server.providerId)
|
|
|
|
# elif request.user is not None:
|
|
# user = request.user
|
|
# session.set_user(user.id)
|
|
# else:
|
|
# session.set_user('anonymous-%s' % login.nameIdentifier.content)
|
|
# user = session.get_user(login.nameIdentifier.content)
|
|
|
|
# if not user.name_identifiers.has_key(server.providerId):
|
|
# user.name_identifiers[server.providerId] = [ login.nameIdentifier.content ]
|
|
# elif not login.nameIdentifier.content in user.name_identifiers[server.providerId]:
|
|
# user.name_identifiers[server.providerId].append(login.nameIdentifier.content)
|
|
# user.lasso_dumps[server.providerId] = login.identity.dump()
|
|
# user.store()
|
|
# request.user = user
|
|
|
|
federations = Federation.select(lambda x: host.id == x.host_id \
|
|
and user.name_identifiers[0] in x.name_identifiers)
|
|
|
|
if federations:
|
|
return site_authentication.get_site_authentication(host).sso_local_login(federations[0])
|
|
else:
|
|
if hasattr(session, 'token'):
|
|
# FIXME : this method doesn't exist anymore
|
|
self.federate_token(session.token)
|
|
else:
|
|
response = get_response()
|
|
if session.after_url:
|
|
after_url = session.after_url
|
|
session.after_url = None
|
|
return redirect(after_url)
|
|
response.set_status(303)
|
|
response.headers['location'] = urlparse.urljoin(request.get_url(), str('local_auth'))
|
|
response.content_type = 'text/plain'
|
|
return 'Your browser should redirect you'
|
|
|
|
def lookup_user(self, session, login):
|
|
found_users = list(User.select(lambda x: login.nameIdentifier.content in x.name_identifiers))
|
|
if found_users:
|
|
return found_users[0]
|
|
return None
|
|
|
|
def singleLogout(self):
|
|
request = get_request()
|
|
logout = lasso.Logout(misc.get_lasso_server())
|
|
if lasso.isLibertyQuery(request.get_query()):
|
|
try:
|
|
logout.processRequestMsg(request.get_query())
|
|
except lasso.Error, error:
|
|
if error[0] == lasso.DS_ERROR_INVALID_SIGNATURE:
|
|
return error_page(_('Failed to check single logout request signature.'))
|
|
raise
|
|
session = get_session()
|
|
if not session.id:
|
|
# session has not been found, this may be because the user has
|
|
# its browser configured so that cookies are not sent for
|
|
# remote queries and IdP is using image-based SLO.
|
|
# so we look up a session with the appropriate name identifier
|
|
for session in get_session_manager().values():
|
|
# This block differs from qommon
|
|
user = session.get_user(logout.server.providerId)
|
|
if user and logout.nameIdentifier.content in user.name_identifiers:
|
|
break
|
|
else:
|
|
session = get_session()
|
|
return self.slo_idp(logout, session)
|
|
else:
|
|
return self.slo_sp(logout, get_session())
|
|
|
|
def singleLogoutReturn(self):
|
|
logout = lasso.Logout(misc.get_lasso_server())
|
|
host = Host.get_host_from_url()
|
|
if host is None:
|
|
return redirect('%s/' % get_request().environ['SCRIPT_NAME'])
|
|
|
|
try:
|
|
logout.processResponseMsg(get_request().get_query())
|
|
except lasso.Error, error:
|
|
if error[0] == lasso.PROFILE_ERROR_INVALID_QUERY:
|
|
raise AccessError()
|
|
if error[0] == lasso.DS_ERROR_INVALID_SIGNATURE:
|
|
return error_page(_('Failed to check single logout request signature.'))
|
|
if hasattr(lasso, 'LOGOUT_ERROR_REQUEST_DENIED') and \
|
|
error[0] == lasso.LOGOUT_ERROR_REQUEST_DENIED:
|
|
return redirect(host.get_root_url()) # ignore silently
|
|
elif error[0] == lasso.ERROR_UNDEFINED:
|
|
# XXX: unknown status; ignoring for now.
|
|
return redirect(host.get_root_url()) # ignore silently
|
|
raise
|
|
return redirect(host.get_root_url())
|
|
|
|
def slo_idp(self, logout, session):
|
|
'''Single Logout initiated by IdP'''
|
|
# This block differs from qommon
|
|
if session.lasso_session_dumps.has_key(logout.server.providerId):
|
|
logout.setSessionFromDump(session.lasso_session_dumps[logout.server.providerId])
|
|
user = session.get_user(logout.server.providerId)
|
|
if user and user.lasso_dumps:
|
|
logout.setIdentityFromDump(user.lasso_dumps[0])
|
|
if user and logout.nameIdentifier.content not in user.name_identifiers:
|
|
raise 'No appropriate name identifier in user (%s and %s)' % (
|
|
logout.nameIdentifier.content, user.name_identifiers)
|
|
|
|
host = Host.get_host_with_provider_id(logout.server.providerId)
|
|
if host is not None:
|
|
site_authentication.get_site_authentication(host).local_logout(user=user)
|
|
|
|
try:
|
|
logout.validateRequest()
|
|
except lasso.Error, error:
|
|
if error[0] != lasso.PROFILE_ERROR_SESSION_NOT_FOUND:
|
|
raise
|
|
else:
|
|
get_session_manager().expire_session(logout.server.providerId)
|
|
|
|
logout.buildResponseMsg()
|
|
if logout.msgBody: # soap answer
|
|
return logout.msgBody
|
|
else:
|
|
return redirect(logout.msgUrl)
|
|
|
|
def slo_sp(self, logout, session):
|
|
host = Host.get_host_from_url()
|
|
if host is None:
|
|
return redirect('%s/' % get_request().environ['SCRIPT_NAME'])
|
|
|
|
if not session.id or not session.users.has_key(logout.server.providerId) \
|
|
or not session.lasso_session_dumps.has_key(logout.server.providerId):
|
|
get_session_manager().expire_session(logout.server.providerId)
|
|
return redirect(host.get_root_url())
|
|
|
|
logout.setSessionFromDump(session.lasso_session_dumps[logout.server.providerId])
|
|
user = session.get_user(logout.server.providerId)
|
|
site_authentication.get_site_authentication(host).local_logout(user=user)
|
|
if user and user.lasso_dumps:
|
|
logout.setIdentityFromDump(user.lasso_dumps[0])
|
|
|
|
return self.slo_sp_redirect(logout, host)
|
|
|
|
def slo_sp_redirect(self, logout, host):
|
|
try:
|
|
logout.initRequest(None, lasso.HTTP_METHOD_REDIRECT)
|
|
except lasso.Error, error:
|
|
if error[0] == lasso.PROFILE_ERROR_NAME_IDENTIFIER_NOT_FOUND:
|
|
get_session_manager().expire_session()
|
|
return redirect(host.get_root_url())
|
|
raise
|
|
logout.buildRequestMsg()
|
|
get_session_manager().expire_session(logout.server.providerId)
|
|
return redirect(logout.msgUrl)
|
|
|
|
def soapEndpoint(self):
|
|
request = get_request()
|
|
ctype = request.environ.get('CONTENT_TYPE')
|
|
if not ctype:
|
|
return
|
|
|
|
ctype, ctype_params = parse_header(ctype)
|
|
if ctype != 'text/xml':
|
|
return
|
|
|
|
response = get_response()
|
|
response.set_content_type('text/xml')
|
|
|
|
length = int(request.environ.get('CONTENT_LENGTH'))
|
|
soap_message = request.stdin.read(length)
|
|
|
|
request_type = lasso.getRequestTypeFromSoapMsg(soap_message)
|
|
|
|
if request_type == lasso.REQUEST_TYPE_LOGOUT:
|
|
logout = lasso.Logout(misc.get_lasso_server())
|
|
logout.processRequestMsg(soap_message)
|
|
name_identifier = logout.nameIdentifier.content
|
|
for session in get_session_manager().values():
|
|
user = session.get_user(logout.server.providerId)
|
|
if user and logout.nameIdentifier.content in user.name_identifiers:
|
|
break
|
|
else:
|
|
session = None
|
|
return self.slo_idp(logout, session)
|
|
|
|
if request_type == lasso.REQUEST_TYPE_DEFEDERATION:
|
|
defederation = lasso.Defederation(misc.get_lasso_server())
|
|
defederation.processNotificationMsg(soap_message)
|
|
for session in get_session_manager().values():
|
|
user = session.get_user(defederation.server.providerId)
|
|
if user and defederation.nameIdentifier.content in user.name_identifiers:
|
|
break
|
|
else:
|
|
session = None
|
|
return self.fedterm(defederation, session)
|
|
|
|
def federationTermination(self):
|
|
request = get_request()
|
|
if not lasso.isLibertyQuery(request.get_query()):
|
|
return redirect('%s/' % get_request().environ['SCRIPT_NAME'])
|
|
|
|
defederation = lasso.Defederation(misc.get_lasso_server())
|
|
defederation.processNotificationMsg(request.get_query())
|
|
return self.fedterm(defederation, get_session())
|
|
|
|
def fedterm(self, defederation, session):
|
|
if session is not None:
|
|
host = Host.get_host_with_provider_id(defederation.server.providerId)
|
|
if host is not None:
|
|
site_authentication.get_site_authentication(host).local_defederate(session, defederation.server.providerId)
|
|
if session.lasso_session_dumps.has_key(defederation.server.providerId):
|
|
defederation.setSessionFromDump(session.lasso_session_dumps[defederation.server.providerId])
|
|
user = session.get_user(defederation.server.providerId)
|
|
if user and user.lasso_dumps:
|
|
defederation.setIdentityFromDump(user.lasso_dumps[0])
|
|
else:
|
|
user = None
|
|
|
|
try:
|
|
defederation.validateNotification()
|
|
except lasso.Error, error:
|
|
pass # ignore failure (?)
|
|
else:
|
|
if user:
|
|
if not defederation.identity:
|
|
# if it was the last federation the whole identity dump collapsed
|
|
del user.lasso_dumps[0]
|
|
else:
|
|
user.lasso_dumps[0] = defederation.identity.dump()
|
|
user.store()
|
|
|
|
if user and defederation.nameIdentifier.content:
|
|
user.remove_name_identifier(defederation.server.providerId, defederation.nameIdentifier.content)
|
|
user.store()
|
|
|
|
if defederation.isSessionDirty and session is not None:
|
|
if not defederation.session:
|
|
del session.lasso_session_dumps[defederation.server.providerId]
|
|
else:
|
|
session.lasso_session_dumps[defederation.server.providerId] = defederation.session.dump()
|
|
session.store()
|
|
|
|
get_session_manager().expire_session(defederation.server.providerId)
|
|
|
|
if defederation.msgUrl:
|
|
return redirect(defederation.msgUrl)
|
|
else:
|
|
response = get_response()
|
|
response.set_status(204)
|
|
return ''
|
|
|
|
def federationTerminationReturn(self):
|
|
host = Host.get_host_from_url()
|
|
if host is None:
|
|
return redirect('%s/' % get_request().environ['SCRIPT_NAME'])
|
|
return redirect(host.get_return_url())
|
|
|
|
# def defederate(self, idp = None):
|
|
# session = get_session()
|
|
# user = get_request().user
|
|
|
|
# server = misc.get_lasso_server()
|
|
|
|
# self.local_defederate(session, server.providerId)
|
|
#
|
|
# defederation = lasso.Defederation(server)
|
|
# defederation.setSessionFromDump(session.lasso_session_dumps[defederation.server.providerId])
|
|
# if user and user.lasso_dumps.has_key(defederation.server.providerId):
|
|
# defederation.setIdentityFromDump(user.lasso_dumps[defederation.server.providerId])
|
|
|
|
# defederation.initNotification(idp, lasso.HTTP_METHOD_SOAP);
|
|
# defederation.buildNotificationMsg();
|
|
|
|
# try:
|
|
# soap_call(defederation.msgUrl, defederation.msgBody);
|
|
# except SOAPException:
|
|
# return error_page(_('Failure to communicate with identity provider'))
|
|
|
|
# rootUrl = '/' + misc.get_proxied_site_name() + '/'
|
|
# return redirect(rootUrl)
|
|
|
|
def local_auth(self):
|
|
host = Host.get_host_from_url()
|
|
if host is None:
|
|
return redirect('%s/' % get_request().environ['SCRIPT_NAME'])
|
|
return site_authentication.get_site_authentication(host).local_auth
|
|
local_auth = property(local_auth)
|
|
|
|
def metadata(self):
|
|
host = Host.get_host_from_url()
|
|
if host is None:
|
|
return redirect('%s/' % get_request().environ['SCRIPT_NAME'])
|
|
get_response().set_content_type('text/xml', 'utf-8')
|
|
metadata = unicode(open(host.metadata).read(), 'utf-8')
|
|
return metadata
|
|
|
|
def public_key(self):
|
|
host = Host.get_host_from_url()
|
|
if host is None:
|
|
return redirect('%s/' % get_request().environ['SCRIPT_NAME'])
|
|
get_response().set_content_type('text/plain')
|
|
public_key = open(host.public_key).read()
|
|
return public_key
|
|
|
|
|
|
class SOAPException(Exception):
|
|
pass
|
|
|
|
|
|
def soap_call(url, msg):
|
|
if url.startswith('http://'):
|
|
host, query = urllib.splithost(url[5:])
|
|
conn = httplib.HTTPConnection(host)
|
|
else:
|
|
host, query = urllib.splithost(url[6:])
|
|
conn = httplib.HTTPSConnection(host)
|
|
conn.request('POST', query, msg, {'Content-Type': 'text/xml'})
|
|
response = conn.getresponse()
|
|
data = response.read()
|
|
conn.close()
|
|
if response.status not in (200, 204): # 204 ok for federation termination
|
|
get_logger().warn('SOAP error (%s) (on %s)' % (response.status, url))
|
|
raise SOAPException()
|
|
return data
|