797 lines
36 KiB
Python
Executable File
797 lines
36 KiB
Python
Executable File
#! /usr/bin/env python
|
|
# -*- coding: UTF-8 -*-
|
|
#
|
|
# $Id: idwsf2_tests.py 3254 2007-06-05 21:23:57Z dlaniel $
|
|
#
|
|
# Python unit tests for Lasso library
|
|
#
|
|
# Copyright (C) 2004-2007 Entr'ouvert
|
|
# http://lasso.entrouvert.org
|
|
#
|
|
# Authors: See AUTHORS file in top-level directory.
|
|
#
|
|
# This program is free software; you can redistribute it and/or modify
|
|
# it under the terms of the GNU General Public License as published by
|
|
# the Free Software Foundation; either version 2 of the License, or
|
|
# (at your option) any later version.
|
|
#
|
|
# This program is distributed in the hope that it will be useful,
|
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
# GNU General Public License for more details.
|
|
#
|
|
# You should have received a copy of the GNU General Public License
|
|
# along with this program; if not, see <http://www.gnu.org/licenses/>.
|
|
|
|
|
|
import os
|
|
import unittest
|
|
import sys
|
|
from StringIO import StringIO
|
|
import logging
|
|
|
|
logging.basicConfig()
|
|
|
|
if not '..' in sys.path:
|
|
sys.path.insert(0, '..')
|
|
if not '../.libs' in sys.path:
|
|
sys.path.insert(0, '../.libs')
|
|
|
|
import lasso
|
|
|
|
try:
|
|
import lxml.etree as ET
|
|
except ImportError:
|
|
try:
|
|
import elementtree.ElementTree as ET
|
|
except ImportError:
|
|
import xml.etree.ElementTree as ET
|
|
|
|
try:
|
|
dataDir
|
|
except NameError:
|
|
dataDir = os.path.join(os.environ['TOP_SRCDIR'], 'tests', 'data')
|
|
|
|
idpSoapEndpoint = 'http://idp1/soapEndpoint'
|
|
spSoapEndpoint = 'http://sp1/soapEndpoint'
|
|
spInteractionUrl = 'http://sp1/askMeAQuestion'
|
|
|
|
class IdWsf2TestCase(unittest.TestCase):
|
|
def getWspServer(self):
|
|
wsp_metadata = os.path.join(dataDir, 'sp5-saml2/metadata.xml')
|
|
wsp_private_key = os.path.join(dataDir, 'sp5-saml2/private-key.pem')
|
|
idp_metadata = os.path.join(dataDir, 'idp5-saml2/metadata.xml')
|
|
|
|
server = lasso.Server(wsp_metadata, wsp_private_key, None, None)
|
|
server.role = lasso.PROVIDER_ROLE_SP
|
|
server.addProvider(lasso.PROVIDER_ROLE_IDP, idp_metadata, None, None)
|
|
server.setEncryptionPrivateKey(wsp_private_key);
|
|
|
|
return server;
|
|
|
|
def getWscServer(self):
|
|
wsc_metadata = os.path.join(dataDir, 'sp6-saml2/metadata.xml')
|
|
wsc_private_key = os.path.join(dataDir, 'sp6-saml2/private-key.pem')
|
|
idp_metadata = os.path.join(dataDir, 'idp5-saml2/metadata.xml')
|
|
|
|
server = lasso.Server(wsc_metadata, wsc_private_key, None, None)
|
|
server.role = lasso.PROVIDER_ROLE_SP
|
|
server.addProvider(lasso.PROVIDER_ROLE_IDP, idp_metadata, None, None)
|
|
|
|
return server;
|
|
|
|
def getIdpServer(self):
|
|
if hasattr(self, 'idp_server_dump') and self.idp_server_dump is not None:
|
|
server = lasso.Server.newFromDump(self.idp_server_dump)
|
|
server.role = lasso.PROVIDER_ROLE_IDP
|
|
else:
|
|
idp_metadata = os.path.join(dataDir, 'idp5-saml2/metadata.xml')
|
|
idp_private_key = os.path.join(dataDir, 'idp5-saml2/private-key.pem')
|
|
wsp_metadata = os.path.join(dataDir, 'sp5-saml2/metadata.xml')
|
|
wsc_metadata = os.path.join(dataDir, 'sp6-saml2/metadata.xml')
|
|
|
|
server = lasso.Server(idp_metadata, idp_private_key, None, None)
|
|
server.role = lasso.PROVIDER_ROLE_IDP
|
|
server.addProvider(lasso.PROVIDER_ROLE_SP, wsp_metadata, None, None)
|
|
server.getProvider(server.providerIds[0]).setEncryptionMode(lasso.ENCRYPTION_MODE_NAMEID);
|
|
server.addProvider(lasso.PROVIDER_ROLE_SP, wsc_metadata, None, None)
|
|
self.idp_server_dump = server.dump()
|
|
|
|
return server
|
|
|
|
def query(self, wsc, idp, idp_identity_dump, wsc_session_dump, uid, federations, services_map, service_associations, provider_ids = None, service_types = None, options = None, actions = None):
|
|
session = lasso.Session.newFromDump(wsc_session_dump)
|
|
assertion = session.getAssertion(idp.providerId)
|
|
self.failUnless(assertion is not None)
|
|
epr = assertion.idwsf2GetDiscoveryBootstrapEpr()
|
|
self.failUnless(epr is not None)
|
|
wsc_disco = lasso.IdWsf2Discovery(wsc)
|
|
wsc_disco.setEpr(epr)
|
|
wsc_disco.initQuery()
|
|
wsc_disco.addRequestedService(service_types = service_types, provider_ids = provider_ids, options = options, actions = actions)
|
|
wsc_disco.buildRequestMsg()
|
|
idp_disco = lasso.IdWsf2Discovery(idp)
|
|
idp_disco.setIdentityFromDump(idp_identity_dump)
|
|
idp_disco.processRequestMsg(wsc_disco.msgBody)
|
|
f = self.nid2tuple(idp_disco.getNameIdentifier())
|
|
uid = federations[f]
|
|
for id in service_associations[uid]:
|
|
idp_disco.addServiceMetadata(services_map[id])
|
|
idp_disco.validateRequest()
|
|
idp_disco.buildResponseMsg()
|
|
wsc_disco.processResponseMsg(idp_disco.msgBody)
|
|
return wsc_disco.endpointReferences
|
|
|
|
|
|
|
|
def metadataRegister(self, wsp, idp, session_dump, abstract = None, address = None, provider_id = None, service_types = None, services_map = None):
|
|
session = lasso.Session.newFromDump(session_dump)
|
|
assertion = session.getAssertion(idp.providerId)
|
|
self.failUnless(assertion is not None)
|
|
epr = assertion.idwsf2GetDiscoveryBootstrapEpr()
|
|
self.failUnless(epr is not None)
|
|
wsp_disco = lasso.IdWsf2Discovery(wsp)
|
|
wsp_disco.setEpr(epr)
|
|
abstract = 'Personal Profile service'
|
|
self.failUnless(abstract is not None)
|
|
self.failUnless(address is not None)
|
|
self.failUnless(service_types is not None)
|
|
self.failUnless(isinstance(services_map, dict))
|
|
wsp_disco.initMetadataRegister()
|
|
if not provider_id:
|
|
provider_id = wsp.providerId
|
|
wsp_disco.addSimpleServiceMetadata(
|
|
service_types = service_types,
|
|
abstract = abstract, provider_id = provider_id,
|
|
address = address,
|
|
security_mechanisms = (lasso.SECURITY_MECH_BEARER,))
|
|
wsp_disco.buildRequestMsg()
|
|
|
|
idp_disco = lasso.IdWsf2Discovery(idp)
|
|
idp_disco.processRequestMsg(wsp_disco.msgBody)
|
|
idp_disco.checkSecurityMechanism()
|
|
idp_disco.validateRequest()
|
|
self.failUnlessEqual(len(idp_disco.metadatas), 1)
|
|
# add metadatas to directory
|
|
sender = idp_disco.getSoapEnvelopeRequest().sb2GetProviderId()
|
|
self.failUnless(sender is not None)
|
|
metadatas = services_map.get(sender, [])
|
|
for metadata in idp_disco.metadatas:
|
|
services_map[metadata.svcMDID] = metadata
|
|
metadatas.append(metadata.svcMDID)
|
|
services_map[sender] = metadatas
|
|
idp_disco.buildResponseMsg()
|
|
wsp_disco.processResponseMsg(idp_disco.msgBody)
|
|
self.failUnlessEqual(len(wsp_disco.metadatas), 1)
|
|
self.failUnlessEqual(wsp_disco.metadatas[0].svcMDID, wsp_disco.response.svcMDID[0])
|
|
return wsp_disco.metadatas[0].svcMDID
|
|
|
|
def nid2tuple(self, nid):
|
|
return (nid.nameQualifier, nid.format, nid.sPNameQualifier, nid.content)
|
|
|
|
def addAssociation(self, wsp, idp, session_dump, svcmdid, service_maps, federations, service_associations):
|
|
self.failUnless(isinstance(service_associations, dict))
|
|
self.failUnless(isinstance(service_maps, dict))
|
|
# Get the bootstrap
|
|
session = lasso.Session.newFromDump(session_dump)
|
|
assertion = session.getAssertion(idp.providerId)
|
|
self.failUnless(assertion is not None)
|
|
epr = assertion.idwsf2GetDiscoveryBootstrapEpr()
|
|
self.failUnless(epr is not None)
|
|
wsp_disco = lasso.IdWsf2Discovery(wsp)
|
|
wsp_disco.setEpr(epr)
|
|
wsp_disco.initMetadataAssociationAdd()
|
|
wsp_disco.svcmdids = (svcmdid,)
|
|
wsp_disco.buildRequestMsg()
|
|
# Handle request
|
|
idp_disco = lasso.IdWsf2Discovery(idp)
|
|
idp_disco.processRequestMsg(wsp_disco.msgBody)
|
|
idp_disco.checkSecurityMechanism()
|
|
self.failUnlessEqual(idp_disco.svcmdids, (svcmdid,))
|
|
sender = idp_disco.getSoapEnvelopeRequest().sb2GetProviderId()
|
|
name_identifier = idp_disco.getNameIdentifier()
|
|
f = self.nid2tuple(name_identifier)
|
|
uid = federations[f]
|
|
l = service_associations.get(uid, [])
|
|
for id in idp_disco.svcmdids:
|
|
# check it exists
|
|
self.failUnless(service_maps.get(id) is not None)
|
|
# create association
|
|
if id not in l:
|
|
l.append(id)
|
|
service_associations[uid] = l
|
|
idp_disco.validateRequest()
|
|
idp_disco.buildResponseMsg()
|
|
wsp_disco.processResponseMsg(idp_disco.msgBody)
|
|
|
|
def login(self, sp, idp, user_id, federations, sp_identity_dump=None, sp_session_dump=None,
|
|
idp_identity_dump=None, idp_session_dump=None):
|
|
sp_login = lasso.Login(sp)
|
|
idp_provider_id = 'http://idp5/metadata'
|
|
sp_login.initAuthnRequest(idp_provider_id, lasso.HTTP_METHOD_REDIRECT)
|
|
sp_login.request.nameIDPolicy.format = lasso.SAML2_NAME_IDENTIFIER_FORMAT_PERSISTENT
|
|
sp_login.request.nameIDPolicy.allowCreate = True
|
|
sp_login.buildAuthnRequestMsg()
|
|
|
|
idp_login = lasso.Login(idp)
|
|
query = sp_login.msgUrl.split('?')[1]
|
|
if idp_identity_dump is not None:
|
|
idp_login.setIdentityFromDump(idp_identity_dump)
|
|
if idp_session_dump is not None:
|
|
idp_login.setSessionFromDump(idp_session_dump)
|
|
idp_login.processAuthnRequestMsg(query)
|
|
idp_login.validateRequestMsg(True, True)
|
|
idp_login.buildAssertion(lasso.SAML_AUTHENTICATION_METHOD_PASSWORD, None, None, None, None)
|
|
if idp_login.assertion.subject.encryptedId:
|
|
f = self.nid2tuple(idp_login.assertion.subject.encryptedId.originalData)
|
|
else:
|
|
f = self.nid2tuple(idp_login.assertion.subject.nameId)
|
|
federations[f] = user_id
|
|
l = federations.get(user_id, [])
|
|
l.append(f)
|
|
federations[user_id] = l
|
|
idp_login.idwsf2AddDiscoveryBootstrapEpr(url = idpSoapEndpoint, abstract = 'Discovery Service', security_mechanisms = (lasso.SECURITY_MECH_BEARER,))
|
|
idp_login.buildArtifactMsg(lasso.HTTP_METHOD_ARTIFACT_GET)
|
|
artifact_message = idp_login.artifactMessage
|
|
|
|
if idp_login.isIdentityDirty:
|
|
idp_identity_dump = idp_login.identity.dump()
|
|
if idp_login.isSessionDirty:
|
|
idp_session_dump = idp_login.session.dump()
|
|
|
|
sp_login = lasso.Login(sp)
|
|
query = idp_login.msgUrl.split('?')[1]
|
|
query = query.replace("%3D", "=")
|
|
sp_login.initRequest(query, lasso.HTTP_METHOD_ARTIFACT_GET)
|
|
sp_login.buildRequestMsg()
|
|
|
|
idp_login = lasso.Login(idp)
|
|
idp_login.processRequestMsg(sp_login.msgBody)
|
|
idp_login.artifactMessage = artifact_message
|
|
idp_login.buildResponseMsg(None)
|
|
|
|
sp_login.processResponseMsg(idp_login.msgBody)
|
|
sp_login.acceptSso()
|
|
if sp_login.isIdentityDirty:
|
|
sp_identity_dump = sp_login.identity.dump()
|
|
if sp_login.isSessionDirty:
|
|
sp_session_dump = sp_login.session.dump()
|
|
|
|
return sp_identity_dump, sp_session_dump, idp_identity_dump, idp_session_dump, sp_login.idwsf2GetDiscoveryBootstrapEpr()
|
|
|
|
|
|
class MetadataTestCase(IdWsf2TestCase):
|
|
def test01(self):
|
|
"""Test metadata registration on the IdP"""
|
|
idp = self.getIdpServer()
|
|
wsp = self.getWspServer()
|
|
wsp_identity_dump, wsp_session_dump, idp_identity_dump, idp_session_dump, dst_epr = self.login(wsp, idp, 1, {})
|
|
|
|
wsp_disco = lasso.IdWsf2Discovery(wsp)
|
|
wsp_disco.setEpr(dst_epr)
|
|
|
|
abstract = 'Personal Profile service'
|
|
wsp_disco.initMetadataRegister()
|
|
self.failUnless(wsp_disco.request is not None)
|
|
wsp_disco.addSimpleServiceMetadata(service_types =
|
|
(lasso.PP11_HREF,), abstract = abstract,
|
|
provider_id = wsp.providerId, address = spSoapEndpoint,
|
|
security_mechanisms = (lasso.SECURITY_MECH_BEARER,))
|
|
self.failUnlessEqual(len(wsp_disco.metadatas), 1)
|
|
metadata = wsp_disco.metadatas[0]
|
|
self.failUnlessEqual(metadata.abstract, abstract)
|
|
self.failUnlessEqual(metadata.providerId, wsp.providerId)
|
|
self.failUnlessEqual(len(metadata.serviceContext), 1)
|
|
self.failUnlessEqual(len(metadata.serviceContext[0].serviceType), 1)
|
|
self.failUnlessEqual(metadata.serviceContext[0].serviceType[0],
|
|
lasso.PP11_HREF)
|
|
self.failUnlessEqual(len(metadata.serviceContext[0].endpointContext), 1)
|
|
self.failUnlessEqual(
|
|
len(metadata.serviceContext[0].endpointContext[0].address),
|
|
1)
|
|
self.failUnlessEqual(metadata.serviceContext[0].endpointContext[0].address[0],
|
|
spSoapEndpoint)
|
|
self.failUnlessEqual(
|
|
len(metadata.serviceContext[0].endpointContext[0].securityMechId),
|
|
1)
|
|
self.failUnlessEqual(
|
|
metadata.serviceContext[0].endpointContext[0].securityMechId[0],
|
|
lasso.SECURITY_MECH_BEARER)
|
|
self.failUnless(metadata.svcMDID is None)
|
|
wsp_disco.buildRequestMsg()
|
|
self.failUnlessEqual(wsp_disco.msgUrl, idpSoapEndpoint)
|
|
self.failUnless(wsp_disco.msgBody is not None)
|
|
|
|
idp_disco = lasso.IdWsf2Discovery(idp)
|
|
self.failUnless(idp_disco is not None)
|
|
idp_disco.processRequestMsg(wsp_disco.msgBody)
|
|
self.failUnless(idp_disco.request is not None)
|
|
self.failUnlessEqual(len(idp_disco.request.svcMD), 1)
|
|
self.failUnless(idp_disco.request.svcMD[0].svcMDID is None)
|
|
try:
|
|
idp_disco.checkSecurityMechanism()
|
|
except lasso.Error, e:
|
|
self.fail(e)
|
|
try:
|
|
idp_disco.validateRequest()
|
|
except lasso.Error, e:
|
|
self.fail(e)
|
|
self.failUnless(idp_disco.response is not None)
|
|
self.failUnlessEqual(len(idp_disco.metadatas), 1)
|
|
metadata = idp_disco.metadatas[0]
|
|
self.failUnlessEqual(metadata.abstract, abstract)
|
|
self.failUnlessEqual(metadata.providerId, wsp.providerId)
|
|
self.failUnlessEqual(len(metadata.serviceContext), 1)
|
|
self.failUnlessEqual(len(metadata.serviceContext[0].serviceType), 1)
|
|
self.failUnlessEqual(metadata.serviceContext[0].serviceType[0],
|
|
lasso.PP11_HREF)
|
|
self.failUnlessEqual(len(metadata.serviceContext[0].endpointContext), 1)
|
|
self.failUnlessEqual(
|
|
len(metadata.serviceContext[0].endpointContext[0].address),
|
|
1)
|
|
self.failUnlessEqual(metadata.serviceContext[0].endpointContext[0].address[0],
|
|
spSoapEndpoint)
|
|
self.failUnlessEqual(
|
|
len(metadata.serviceContext[0].endpointContext[0].securityMechId),
|
|
1)
|
|
self.failUnlessEqual(
|
|
metadata.serviceContext[0].endpointContext[0].securityMechId[0],
|
|
lasso.SECURITY_MECH_BEARER)
|
|
idp_disco.buildResponseMsg()
|
|
self.failUnless(metadata.svcMDID is not None)
|
|
self.failUnless(idp_disco.msgUrl is None)
|
|
self.failUnless(idp_disco.msgBody is not None)
|
|
|
|
wsp_disco.processResponseMsg(idp_disco.msgBody)
|
|
|
|
self.failUnless(len(wsp_disco.metadatas) == 1, 'missing svcMDID')
|
|
self.failUnless(wsp_disco.metadatas[0].svcMDID, 'missing svcMDID')
|
|
|
|
def test02(self):
|
|
"Test failure by IdP for register request"
|
|
idp = self.getIdpServer()
|
|
wsp = self.getWspServer()
|
|
wsp_identity_dump, wsp_session_dump, idp_identity_dump, idp_session_dump, dst_epr = self.login(wsp, idp, 1, {})
|
|
|
|
wsp_disco = lasso.IdWsf2Discovery(wsp)
|
|
wsp_disco.setEpr(dst_epr)
|
|
|
|
abstract = 'Personal Profile service'
|
|
wsp_disco.initMetadataRegister()
|
|
self.failUnless(wsp_disco.request is not None)
|
|
wsp_disco.addSimpleServiceMetadata(service_types =
|
|
(lasso.PP11_HREF,), abstract = abstract,
|
|
provider_id = wsp.providerId, address = spSoapEndpoint,
|
|
security_mechanisms= (lasso.SECURITY_MECH_BEARER,))
|
|
self.failUnlessEqual(len(wsp_disco.metadatas), 1)
|
|
metadata = wsp_disco.metadatas[0]
|
|
self.failUnlessEqual(metadata.abstract, abstract)
|
|
self.failUnlessEqual(metadata.providerId, wsp.providerId)
|
|
self.failUnlessEqual(len(metadata.serviceContext[0].serviceType), 1)
|
|
self.failUnlessEqual(metadata.serviceContext[0].serviceType[0],
|
|
lasso.PP11_HREF)
|
|
self.failUnlessEqual(len(metadata.serviceContext[0].endpointContext), 1)
|
|
self.failUnlessEqual(
|
|
len(metadata.serviceContext[0].endpointContext[0].address),
|
|
1)
|
|
self.failUnlessEqual(metadata.serviceContext[0].endpointContext[0].address[0],
|
|
spSoapEndpoint)
|
|
self.failUnlessEqual(
|
|
len(metadata.serviceContext[0].endpointContext[0].securityMechId),
|
|
1)
|
|
self.failUnlessEqual(
|
|
metadata.serviceContext[0].endpointContext[0].securityMechId[0],
|
|
lasso.SECURITY_MECH_BEARER)
|
|
self.failUnless(metadata.svcMDID is None)
|
|
wsp_disco.buildRequestMsg()
|
|
self.failUnlessEqual(wsp_disco.msgUrl, idpSoapEndpoint)
|
|
self.failUnless(wsp_disco.msgBody is not None)
|
|
|
|
idp_disco = lasso.IdWsf2Discovery(idp)
|
|
self.failUnless(idp_disco is not None)
|
|
try:
|
|
idp_disco.processRequestMsg(wsp_disco.msgBody)
|
|
except lasso.Error, e:
|
|
self.fail(e)
|
|
self.failUnless(idp_disco.request is not None)
|
|
try:
|
|
idp_disco.checkSecurityMechanism()
|
|
except lasso.Error, e:
|
|
self.fail(e)
|
|
try:
|
|
idp_disco.failRequest(lasso.IDWSF2_DISCOVERY_STATUS_CODE_FAILED, lasso.IDWSF2_DISCOVERY_STATUS_CODE_FORBIDDEN)
|
|
except lasso.Error, e:
|
|
self.fail(e)
|
|
self.failUnless(idp_disco.response is not None)
|
|
self.failUnless(idp_disco.response.status is not None)
|
|
self.failUnless(idp_disco.response.status.code is not lasso.IDWSF2_DISCOVERY_STATUS_CODE_FAILED)
|
|
self.failUnlessEqual(len(idp_disco.response.status.status), 1)
|
|
self.failUnless(idp_disco.response.status.status[0].code is not lasso.IDWSF2_DISCOVERY_STATUS_CODE_FORBIDDEN)
|
|
idp_disco.buildResponseMsg()
|
|
self.failUnless(idp_disco.msgUrl is None)
|
|
self.failUnless(idp_disco.msgBody is not None)
|
|
|
|
try:
|
|
wsp_disco.processResponseMsg(idp_disco.msgBody)
|
|
except lasso.Idwsf2DiscoveryForbiddenError:
|
|
pass
|
|
except lasso.Error, e:
|
|
self.fail(e)
|
|
|
|
def test03(self):
|
|
"""Test metadata register with redirection"""
|
|
idp = self.getIdpServer()
|
|
wsp = self.getWspServer()
|
|
wsp_identity_dump, wsp_session_dump, idp_identity_dump, idp_session_dump, dst_epr = self.login(wsp, idp, 1, {})
|
|
|
|
wsp_disco = lasso.IdWsf2Discovery(wsp)
|
|
wsp_disco.setEpr(dst_epr)
|
|
|
|
abstract = 'Personal Profile service'
|
|
wsp_disco.initMetadataRegister()
|
|
soap_envelope = wsp_disco.getSoapEnvelopeRequest()
|
|
soap_envelope.setSb2UserInteractionHint(lasso.IDWSF2_SB2_USER_INTERACTION_HINT_INTERACT_IF_NEEDED)
|
|
self.failUnless(isinstance(soap_envelope.header, lasso.SoapHeader))
|
|
self.failUnless(len(soap_envelope.header.other) > 0)
|
|
self.failUnlessEqual(soap_envelope.getSb2UserInteractionHint(), lasso.IDWSF2_SB2_USER_INTERACTION_HINT_INTERACT_IF_NEEDED)
|
|
self.failUnless(wsp_disco.request is not None)
|
|
wsp_disco.addSimpleServiceMetadata(service_types =
|
|
(lasso.PP11_HREF,), abstract = abstract,
|
|
provider_id = wsp.providerId, address = spSoapEndpoint,
|
|
security_mechanisms = (lasso.SECURITY_MECH_BEARER,))
|
|
self.failUnlessEqual(len(wsp_disco.metadatas), 1)
|
|
metadata = wsp_disco.metadatas[0]
|
|
self.failUnlessEqual(metadata.abstract, abstract)
|
|
self.failUnlessEqual(metadata.providerId, wsp.providerId)
|
|
self.failUnlessEqual(len(metadata.serviceContext), 1)
|
|
self.failUnlessEqual(len(metadata.serviceContext[0].serviceType), 1)
|
|
self.failUnlessEqual(metadata.serviceContext[0].serviceType[0],
|
|
lasso.PP11_HREF)
|
|
self.failUnlessEqual(len(metadata.serviceContext[0].endpointContext), 1)
|
|
self.failUnlessEqual(
|
|
len(metadata.serviceContext[0].endpointContext[0].address),
|
|
1)
|
|
self.failUnlessEqual(metadata.serviceContext[0].endpointContext[0].address[0],
|
|
spSoapEndpoint)
|
|
self.failUnlessEqual(
|
|
len(metadata.serviceContext[0].endpointContext[0].securityMechId),
|
|
1)
|
|
self.failUnlessEqual(
|
|
metadata.serviceContext[0].endpointContext[0].securityMechId[0],
|
|
lasso.SECURITY_MECH_BEARER)
|
|
self.failUnless(metadata.svcMDID is None)
|
|
wsp_disco.buildRequestMsg()
|
|
self.failUnlessEqual(wsp_disco.msgUrl, idpSoapEndpoint)
|
|
self.failUnless(wsp_disco.msgBody is not None)
|
|
|
|
idp_disco = lasso.IdWsf2Discovery(idp)
|
|
self.failUnless(idp_disco is not None)
|
|
idp_disco.processRequestMsg(wsp_disco.msgBody)
|
|
self.failUnless(idp_disco.request is not None)
|
|
self.failUnlessEqual(len(idp_disco.request.svcMD), 1)
|
|
self.failUnless(idp_disco.request.svcMD[0].svcMDID is None)
|
|
soap_envelope = idp_disco.getSoapEnvelopeRequest()
|
|
self.failUnless(soap_envelope is not None)
|
|
self.failUnless(soap_envelope.getMessageId() is not None)
|
|
try:
|
|
idp_disco.checkSecurityMechanism()
|
|
except lasso.Error, e:
|
|
self.fail(e)
|
|
# redirect
|
|
interactionUrl = spInteractionUrl
|
|
idp_disco.redirectUserForInteraction(interactionUrl, False)
|
|
response = idp_disco.response
|
|
self.failUnless(isinstance(response, lasso.SoapFault))
|
|
self.failUnless(response.detail is not None)
|
|
self.failUnlessEqual(len(response.detail.any), 1)
|
|
self.failUnless(isinstance(response.detail.any[0], lasso.IdWsf2Sb2RedirectRequest))
|
|
self.failUnless(response.detail.any[0].redirectURL.startswith(interactionUrl + '?transactionID='))
|
|
try:
|
|
idp_disco.buildResponseMsg()
|
|
except lasso.Error, e:
|
|
self.fail(e)
|
|
self.failUnless(idp_disco.msgBody is not None)
|
|
|
|
|
|
self.failUnless(idp_disco.msgUrl is None)
|
|
self.failUnless(idp_disco.msgBody is not None)
|
|
|
|
try:
|
|
wsp_disco.processResponseMsg(idp_disco.msgBody)
|
|
except lasso.WsfprofileRedirectRequestError:
|
|
pass
|
|
except lasso.Error, e:
|
|
self.fail(e)
|
|
response_envelope = wsp_disco.getSoapEnvelopeResponse()
|
|
self.failUnless(response_envelope.sb2GetRedirectRequestUrl().startswith(interactionUrl + '?transactionID='))
|
|
# Here keep information about the request associated to ID: response_envelope.getMessageId().content
|
|
wsp_disco_dump = wsp_disco.dump()
|
|
wsp_disco = lasso.Node.newFromDump(wsp_disco_dump)
|
|
wsp_disco.server = wsp
|
|
request_envelope = wsp_disco.getSoapEnvelopeRequest()
|
|
self.failUnless(request_envelope is not None)
|
|
relates_to = request_envelope.getRelatesTo(True)
|
|
self.failUnless(relates_to is not None)
|
|
response_message_id = response_envelope.getMessageId().content
|
|
relates_to.content = response_message_id
|
|
wsp_disco.buildRequestMsg()
|
|
# now redo as for test01 after request building
|
|
self.failUnlessEqual(wsp_disco.msgUrl, idpSoapEndpoint)
|
|
self.failUnless(wsp_disco.msgBody is not None)
|
|
|
|
idp_disco = lasso.IdWsf2Discovery(idp)
|
|
self.failUnless(idp_disco is not None)
|
|
idp_disco.processRequestMsg(wsp_disco.msgBody)
|
|
self.failUnless(idp_disco.request is not None)
|
|
self.failUnlessEqual(len(idp_disco.request.svcMD), 1)
|
|
self.failUnless(idp_disco.request.svcMD[0].svcMDID is None)
|
|
try:
|
|
idp_disco.checkSecurityMechanism()
|
|
except lasso.Error, e:
|
|
self.fail(e)
|
|
try:
|
|
idp_disco.validateRequest()
|
|
except lasso.Error, e:
|
|
self.fail(e)
|
|
self.failUnless(idp_disco.response is not None)
|
|
self.failUnlessEqual(len(idp_disco.metadatas), 1)
|
|
metadata = idp_disco.metadatas[0]
|
|
self.failUnlessEqual(metadata.abstract, abstract)
|
|
self.failUnlessEqual(metadata.providerId, wsp.providerId)
|
|
self.failUnlessEqual(len(metadata.serviceContext), 1)
|
|
self.failUnlessEqual(len(metadata.serviceContext[0].serviceType), 1)
|
|
self.failUnlessEqual(metadata.serviceContext[0].serviceType[0],
|
|
lasso.PP11_HREF)
|
|
self.failUnlessEqual(len(metadata.serviceContext[0].endpointContext), 1)
|
|
self.failUnlessEqual(
|
|
len(metadata.serviceContext[0].endpointContext[0].address),
|
|
1)
|
|
self.failUnlessEqual(metadata.serviceContext[0].endpointContext[0].address[0],
|
|
spSoapEndpoint)
|
|
self.failUnlessEqual(
|
|
len(metadata.serviceContext[0].endpointContext[0].securityMechId),
|
|
1)
|
|
self.failUnlessEqual(
|
|
metadata.serviceContext[0].endpointContext[0].securityMechId[0],
|
|
lasso.SECURITY_MECH_BEARER)
|
|
idp_disco.buildResponseMsg()
|
|
self.failUnless(metadata.svcMDID is not None)
|
|
self.failUnless(idp_disco.msgUrl is None)
|
|
self.failUnless(idp_disco.msgBody is not None)
|
|
|
|
wsp_disco.processResponseMsg(idp_disco.msgBody)
|
|
|
|
self.failUnless(len(wsp_disco.metadatas) == 1, 'missing svcMDID')
|
|
self.failUnless(wsp_disco.metadatas[0].svcMDID, 'missing svcMDID')
|
|
|
|
def test04(self):
|
|
"""Test metadata query"""
|
|
idp = self.getIdpServer()
|
|
wsp = self.getWspServer()
|
|
federations = {}
|
|
wsp_identity_dump, wsp_session_dump, idp_identity_dump, idp_session_dump, dst_epr = self.login(wsp, idp, 1, federations)
|
|
service_map = {}
|
|
self.metadataRegister(wsp, idp, wsp_session_dump, service_types =
|
|
(lasso.PP11_HREF,), address = spSoapEndpoint,
|
|
abstract = 'My first PP service', services_map = service_map)
|
|
self.metadataRegister(wsp, idp, wsp_session_dump, service_types =
|
|
(lasso.PP11_HREF,), address = spSoapEndpoint+'2',
|
|
abstract = 'My second PP service', services_map = service_map)
|
|
wsp_identity_dump, wsp_session_dump, idp_identity_dump, idp_session_dump, dst_epr = self.login(wsp, idp, 1, federations)
|
|
wsp_disco = lasso.IdWsf2Discovery(wsp)
|
|
wsp_disco.setEpr(dst_epr)
|
|
wsp_disco.initMetadataQuery()
|
|
wsp_disco.buildRequestMsg()
|
|
|
|
idp_disco = lasso.IdWsf2Discovery(idp)
|
|
idp_disco.processRequestMsg(wsp_disco.msgBody)
|
|
idp_disco.checkSecurityMechanism()
|
|
self.failUnlessEqual(idp_disco.svcmdids, ())
|
|
sender = idp_disco.getSoapEnvelopeRequest().sb2GetProviderId()
|
|
for svcMDID in service_map.get(sender, []):
|
|
idp_disco.addServiceMetadata(service_map.get(svcMDID))
|
|
idp_disco.validateRequest()
|
|
idp_disco.buildResponseMsg()
|
|
wsp_disco.processResponseMsg(idp_disco.msgBody)
|
|
self.failUnless(len(wsp_disco.metadatas), 2)
|
|
|
|
def test05(self):
|
|
"""Test metadata delete"""
|
|
idp = self.getIdpServer()
|
|
wsp = self.getWspServer()
|
|
wsp_identity_dump, wsp_session_dump, idp_identity_dump, idp_session_dump, dst_epr = self.login(wsp, idp, 1, {})
|
|
service_map = {}
|
|
self.metadataRegister(wsp, idp, wsp_session_dump, service_types =
|
|
(lasso.PP11_HREF,), address = spSoapEndpoint,
|
|
abstract = 'My first PP service', services_map = service_map)
|
|
self.metadataRegister(wsp, idp, wsp_session_dump, service_types =
|
|
(lasso.PP11_HREF,), address = spSoapEndpoint+'2',
|
|
abstract = 'My second PP service', services_map = service_map)
|
|
wsp_identity_dump, wsp_session_dump, idp_identity_dump, idp_session_dump, dst_epr = self.login(wsp, idp, 1, {})
|
|
|
|
wsp_disco = lasso.IdWsf2Discovery(wsp)
|
|
wsp_disco.setEpr(dst_epr)
|
|
wsp_disco.initMetadataDelete()
|
|
svcmdids = tuple(service_map[wsp.providerId])
|
|
wsp_disco.setSvcmdids(svcmdids)
|
|
wsp_disco.buildRequestMsg()
|
|
|
|
idp_disco = lasso.IdWsf2Discovery(idp)
|
|
idp_disco.processRequestMsg(wsp_disco.msgBody)
|
|
idp_disco.checkSecurityMechanism()
|
|
self.failUnlessEqual(idp_disco.svcmdids, svcmdids)
|
|
sender = idp_disco.getSoapEnvelopeRequest().sb2GetProviderId()
|
|
self.failUnlessEqual(sender, wsp.providerId)
|
|
idp_disco.validateRequest()
|
|
idp_disco.buildResponseMsg()
|
|
wsp_disco.processResponseMsg(idp_disco.msgBody)
|
|
|
|
class MetadataAssociationTestCase(IdWsf2TestCase):
|
|
def test01(self):
|
|
"""Metadata association add"""
|
|
idp = self.getIdpServer()
|
|
wsp = self.getWspServer()
|
|
wsp_identity_dump, wsp_session_dump, idp_identity_dump, idp_session_dump, dst_epr = self.login(wsp, idp, 1, {})
|
|
service_map = {}
|
|
svcMDID = self.metadataRegister(wsp, idp, wsp_session_dump, service_types =
|
|
(lasso.PP11_HREF,), address = spSoapEndpoint,
|
|
abstract = 'My first PP service', services_map = service_map)
|
|
# Make the request
|
|
wsp_disco = lasso.IdWsf2Discovery(wsp)
|
|
wsp_disco.setEpr(dst_epr)
|
|
wsp_disco.initMetadataAssociationAdd()
|
|
wsp_disco.svcmdids = (svcMDID,)
|
|
wsp_disco.buildRequestMsg()
|
|
# Receive it
|
|
idp_disco = lasso.IdWsf2Discovery(idp)
|
|
idp_disco.processRequestMsg(wsp_disco.msgBody)
|
|
idp_disco.checkSecurityMechanism()
|
|
self.failUnlessEqual(idp_disco.svcmdids, (svcMDID,))
|
|
sender = idp_disco.getSoapEnvelopeRequest().sb2GetProviderId()
|
|
name_identifier = idp_disco.getNameIdentifier()
|
|
# Store the association
|
|
self.failUnless(sender is not None)
|
|
self.failUnless(name_identifier is not None)
|
|
idp_disco.validateRequest()
|
|
idp_disco.buildResponseMsg()
|
|
wsp_disco.processResponseMsg(idp_disco.msgBody)
|
|
|
|
def test02(self):
|
|
"""Metadata association query"""
|
|
idp = self.getIdpServer()
|
|
wsp = self.getWspServer()
|
|
wsc = self.getWscServer()
|
|
# Register the service, add an association
|
|
federations = {}
|
|
wsp_identity_dump, wsp_session_dump, \
|
|
idp_identity_dump, idp_session_dump, \
|
|
wsp_dst_epr = self.login(wsp, idp, 1, federations)
|
|
service_maps = {}
|
|
svcMDID = self.metadataRegister(wsp, idp, wsp_session_dump,
|
|
service_types = (lasso.PP11_HREF,), address = spSoapEndpoint,
|
|
abstract = 'My first PP service', services_map = service_maps)
|
|
service_associations = {}
|
|
self.addAssociation(wsp, idp, wsp_session_dump, svcMDID, service_maps,
|
|
federations, service_associations)
|
|
# Start a query
|
|
wsp_disco = lasso.IdWsf2Discovery(wsp)
|
|
wsp_disco.setEpr(wsp_dst_epr)
|
|
wsp_disco.initMetadataAssociationQuery()
|
|
wsp_disco.buildRequestMsg()
|
|
#
|
|
idp_disco = lasso.IdWsf2Discovery(idp)
|
|
idp_disco.processRequestMsg(wsp_disco.msgBody)
|
|
idp_disco.checkSecurityMechanism()
|
|
self.failUnlessEqual(idp_disco.svcmdids, ())
|
|
f = self.nid2tuple(idp_disco.getNameIdentifier())
|
|
uid = federations[f]
|
|
result = []
|
|
for svcmdid in service_associations[uid]:
|
|
result.append(svcmdid)
|
|
idp_disco.svcmdids = tuple(result)
|
|
idp_disco.validateRequest()
|
|
idp_disco.buildResponseMsg()
|
|
wsp_disco.processResponseMsg(idp_disco.msgBody)
|
|
self.failUnlessEqual(wsp_disco.svcmdids, (svcMDID,))
|
|
|
|
def test03(self):
|
|
"""Metadata association delete"""
|
|
pass
|
|
|
|
class QueryTestCase(IdWsf2TestCase):
|
|
def test01(self):
|
|
"""Discovery Service Query"""
|
|
idp = self.getIdpServer()
|
|
wsp = self.getWspServer()
|
|
wsc = self.getWscServer()
|
|
federations = {}
|
|
# Register the service, add an association
|
|
wsp_identity_dump, wsp_session_dump, idp_identity_dump, idp_session_dump, wsp_dst_epr = self.login(wsp, idp, 1, federations)
|
|
service_maps = {}
|
|
svcMDID = self.metadataRegister(wsp, idp, wsp_session_dump, service_types =
|
|
(lasso.PP11_HREF,), address = spSoapEndpoint,
|
|
abstract = 'My first PP service', services_map = service_maps)
|
|
service_associations = {}
|
|
self.addAssociation(wsp, idp, wsp_session_dump, svcMDID, service_maps, federations, service_associations)
|
|
# Try to find the service
|
|
wsc_identity_dump, wsc_session_dump, idp_identity_dump, idp_session_dump, wsc_dst_epr = self.login(wsc, idp, 1, federations, idp_identity_dump = idp_identity_dump, idp_session_dump = idp_session_dump)
|
|
wsc_disco = lasso.IdWsf2Discovery(wsc)
|
|
wsc_disco.setEpr(wsc_dst_epr)
|
|
wsc_disco.initQuery()
|
|
wsc_disco.addRequestedService(service_types = (lasso.PP11_HREF,))
|
|
wsc_disco.buildRequestMsg()
|
|
idp_disco = lasso.IdWsf2Discovery(idp)
|
|
idp_disco.setIdentityFromDump(idp_identity_dump)
|
|
idp_disco.processRequestMsg(wsc_disco.msgBody)
|
|
f = self.nid2tuple(idp_disco.getNameIdentifier())
|
|
uid = federations[f]
|
|
for id in service_associations[uid]:
|
|
idp_disco.addServiceMetadata(service_maps[id])
|
|
idp_disco.validateRequest()
|
|
idp_disco.buildResponseMsg()
|
|
wsc_disco.processResponseMsg(idp_disco.msgBody)
|
|
self.failUnlessEqual(len(wsc_disco.endpointReferences), 1)
|
|
|
|
class DstTestCase(IdWsf2TestCase):
|
|
def test01(self):
|
|
"""Data Service Template Query"""
|
|
content = '<pp:PP xmlns:pp="%s">Coin</pp:PP>' % lasso.PP11_HREF
|
|
idp = self.getIdpServer()
|
|
wsp = self.getWspServer()
|
|
wsc = self.getWscServer()
|
|
federations = {}
|
|
# Register the service, add an association
|
|
wsp_identity_dump, wsp_session_dump, idp_identity_dump, \
|
|
idp_session_dump, wsp_dst_epr = self.login(wsp, idp, 1,
|
|
federations)
|
|
service_maps = {}
|
|
svcMDID = self.metadataRegister(wsp, idp, wsp_session_dump,
|
|
service_types = (lasso.PP11_HREF,), address =
|
|
spSoapEndpoint, abstract = 'My first PP service',
|
|
services_map = service_maps)
|
|
service_associations = {}
|
|
self.addAssociation(wsp, idp, wsp_session_dump, svcMDID,
|
|
service_maps, federations, service_associations)
|
|
wsc_identity_dump, wsc_session_dump, idp_identity_dump, \
|
|
idp_session_dump, wsc_dst_epr = self.login(wsc, idp, 1, federations,
|
|
idp_identity_dump = idp_identity_dump, idp_session_dump =
|
|
idp_session_dump)
|
|
eprs = self.query(wsc, idp, idp_identity_dump, wsc_session_dump, 1,
|
|
federations, service_maps, service_associations,
|
|
service_types = (lasso.PP11_HREF,))
|
|
self.failUnless(len(eprs), 1)
|
|
lasso.registerIdwsf2DstService(lasso.PP11_PREFIX, lasso.PP11_HREF)
|
|
wsc_dst = lasso.IdWsf2DataService(wsc)
|
|
wsc_dst.setEpr(eprs[0])
|
|
wsc_dst.initQuery()
|
|
wsc_dst.setServiceType(lasso.PP11_PREFIX, lasso.PP11_HREF)
|
|
wsc_dst.addQueryItem('/%s:PP' % lasso.PP11_PREFIX, 'xxx')
|
|
wsc_dst.buildRequestMsg()
|
|
wsp_dst = lasso.IdWsf2DataService(wsp)
|
|
wsp_dst.processRequestMsg(wsc_dst.msgBody)
|
|
self.failUnlessEqual(wsp_dst.requestType, lasso.IDWSF2_DATA_SERVICE_REQUEST_TYPE_QUERY)
|
|
wsp_dst.checkSecurityMechanism()
|
|
data = ET.parse(StringIO(content))
|
|
for item in wsp_dst.items:
|
|
result = data.xpath(item.select, namespaces = { lasso.PP11_PREFIX: lasso.PP11_HREF })
|
|
for found in result:
|
|
wsp_dst.setQueryItemResult(item.itemId, ET.tostring(found), True)
|
|
wsp_dst.setServiceType(lasso.PP11_PREFIX, lasso.PP11_HREF)
|
|
wsp_dst.validateRequest()
|
|
wsp_dst.buildResponseMsg()
|
|
wsc_dst.processResponseMsg(wsp_dst.msgBody)
|
|
|
|
|
|
metadataSuite = unittest.makeSuite(MetadataTestCase, 'test')
|
|
metadataAssociationSuite = unittest.makeSuite(MetadataAssociationTestCase, 'test')
|
|
querySuite = unittest.makeSuite(QueryTestCase, 'test')
|
|
dstSuite = unittest.makeSuite(DstTestCase, 'test')
|
|
|
|
|
|
allTests = unittest.TestSuite((metadataSuite,
|
|
metadataAssociationSuite,querySuite,dstSuite))
|
|
|
|
if __name__ == '__main__':
|
|
sys.exit(not unittest.TextTestRunner(verbosity = 2).run(allTests).wasSuccessful())
|
|
|