Final touch to ECP
This commit is contained in:
Roland Hedberg
parent
b295a359b9
commit
db681c0841
@ -4,6 +4,8 @@ from hashlib import sha1
|
||||
from saml2.metadata import ENDPOINTS
|
||||
from saml2.profile import paos, ecp
|
||||
from saml2.soap import parse_soap_enveloped_saml_artifact_resolve
|
||||
from saml2.soap import class_instances_from_soap_enveloped_saml_thingies
|
||||
from saml2.soap import open_soap_envelope
|
||||
|
||||
from saml2 import samlp
|
||||
from saml2 import saml
|
||||
@ -259,7 +261,7 @@ class Entity(HTTPBase):
|
||||
elif binding == BINDING_SOAP:
|
||||
func = getattr(soap, "parse_soap_enveloped_saml_%s" % msgtype)
|
||||
xmlstr = func(txt)
|
||||
elif binding == BINDING_URI:
|
||||
elif binding == BINDING_URI or binding is None:
|
||||
xmlstr = txt
|
||||
else:
|
||||
raise ValueError("Don't know how to handle '%s'" % binding)
|
||||
@ -272,11 +274,19 @@ class Entity(HTTPBase):
|
||||
:param text: The SOAP message
|
||||
:return: A dictionary with two keys "body" and "header"
|
||||
"""
|
||||
return soap.class_instances_from_soap_enveloped_saml_thingies(text,
|
||||
return class_instances_from_soap_enveloped_saml_thingies(text,
|
||||
[paos,
|
||||
ecp,
|
||||
samlp])
|
||||
|
||||
def unpack_soap_message(self, text):
|
||||
"""
|
||||
Picks out the parts of the SOAP message, body and headers apart
|
||||
:param text: The SOAP message
|
||||
:return: A dictionary with two keys "body"/"header"
|
||||
"""
|
||||
return open_soap_envelope(text)
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
def sign(self, msg, mid=None, to_sign=None):
|
||||
|
||||
@ -150,13 +150,25 @@ import re
|
||||
|
||||
NS_AND_TAG = re.compile("\{([^}]+)\}(.*)")
|
||||
|
||||
def instanciate_class(item, modules):
|
||||
m = NS_AND_TAG.match(item.tag)
|
||||
ns,tag = m.groups()
|
||||
for module in modules:
|
||||
if module.NAMESPACE == ns:
|
||||
try:
|
||||
target = module.ELEMENT_BY_TAG[tag]
|
||||
return create_class_from_element_tree(target, item)
|
||||
except KeyError:
|
||||
continue
|
||||
raise Exception("Unknown class: ns='%s', tag='%s'" % (ns, tag))
|
||||
|
||||
def class_instances_from_soap_enveloped_saml_thingies(text, modules):
|
||||
"""Parses a SOAP enveloped header and body SAML thing and returns the
|
||||
thing as a dictionary class instance.
|
||||
|
||||
:param text: The SOAP object as XML
|
||||
:param modules: modules representing xsd schemas
|
||||
:return: SAML thingy as a class instance
|
||||
:return: The body and headers as class instances
|
||||
"""
|
||||
try:
|
||||
envelope = ElementTree.fromstring(text)
|
||||
@ -170,32 +182,39 @@ def class_instances_from_soap_enveloped_saml_thingies(text, modules):
|
||||
for part in envelope:
|
||||
if part.tag == '{%s}Body' % soapenv.NAMESPACE:
|
||||
assert len(part) == 1
|
||||
m = NS_AND_TAG.match(part[0].tag)
|
||||
ns,tag = m.groups()
|
||||
for module in modules:
|
||||
if module.NAMESPACE == ns:
|
||||
try:
|
||||
target = module.ELEMENT_BY_TAG[tag]
|
||||
env["body"] = create_class_from_element_tree(target,
|
||||
part[0])
|
||||
except KeyError:
|
||||
continue
|
||||
env["body"] = instanciate_class(part[0], modules)
|
||||
elif part.tag == "{%s}Header" % soapenv.NAMESPACE:
|
||||
for item in part:
|
||||
m = NS_AND_TAG.match(item.tag)
|
||||
ns,tag = m.groups()
|
||||
for module in modules:
|
||||
if module.NAMESPACE == ns:
|
||||
try:
|
||||
target = module.ELEMENT_BY_TAG[tag]
|
||||
env["header"].append(create_class_from_element_tree(
|
||||
target,
|
||||
item))
|
||||
except KeyError:
|
||||
continue
|
||||
env["header"].append(instanciate_class(item, modules))
|
||||
|
||||
return env
|
||||
|
||||
def open_soap_envelope(text):
|
||||
"""
|
||||
|
||||
:param text: SOAP message
|
||||
:return: dictionary with two keys "body"/"header"
|
||||
"""
|
||||
try:
|
||||
envelope = ElementTree.fromstring(text)
|
||||
except Exception, exc:
|
||||
raise XmlParseError("%s" % exc)
|
||||
|
||||
assert envelope.tag == '{%s}Envelope' % soapenv.NAMESPACE
|
||||
assert len(envelope) >= 1
|
||||
content = {"header":[], "body":None}
|
||||
|
||||
for part in envelope:
|
||||
if part.tag == '{%s}Body' % soapenv.NAMESPACE:
|
||||
assert len(part) == 1
|
||||
content["body"] = ElementTree.tostring(part[0], encoding="UTF-8")
|
||||
elif part.tag == "{%s}Header" % soapenv.NAMESPACE:
|
||||
for item in part:
|
||||
_str = ElementTree.tostring(item, encoding="UTF-8")
|
||||
content["header"].append(_str)
|
||||
|
||||
return content
|
||||
|
||||
def make_soap_enveloped_saml_thingy(thingy, headers=None):
|
||||
""" Returns a soap envelope containing a SAML request
|
||||
as a text string.
|
||||
|
||||
@ -1,6 +1,15 @@
|
||||
from saml2.saml import AUTHN_PASSWORD
|
||||
from saml2.httpbase import set_list2dict
|
||||
from saml2.profile.ecp import RelayState
|
||||
from saml2.profile.paos import Request
|
||||
from saml2.request import AuthnRequest
|
||||
from saml2.server import Server
|
||||
from saml2.samlp import Response, STATUS_SUCCESS
|
||||
|
||||
__author__ = 'rolandh'
|
||||
|
||||
from saml2 import soap
|
||||
from saml2 import soap, ecp_client, BINDING_SOAP, BINDING_PAOS, \
|
||||
create_class_from_xml_string
|
||||
from saml2 import samlp
|
||||
from saml2 import config
|
||||
from saml2 import ecp
|
||||
@ -66,3 +75,125 @@ class TestECPClient(object):
|
||||
assert _eq(headers,['{urn:liberty:paos:2003-08}Request',
|
||||
#'{urn:oasis:names:tc:SAML:2.0:profiles:SSO:ecp}Request',
|
||||
'{urn:oasis:names:tc:SAML:2.0:profiles:SSO:ecp}RelayState'])
|
||||
|
||||
try:
|
||||
from saml2.sigver import get_xmlsec_binary
|
||||
except ImportError:
|
||||
get_xmlsec_binary = None
|
||||
|
||||
if get_xmlsec_binary:
|
||||
xmlsec_path = get_xmlsec_binary(["/opt/local/bin"])
|
||||
else:
|
||||
xmlsec_path = '/usr/bin/xmlsec1'
|
||||
|
||||
class DummyResponse(object):
|
||||
def __init__(self, headers):
|
||||
self.headers = headers
|
||||
|
||||
def test_complete_flow():
|
||||
client = ecp_client.Client("user", "password", metadata_file="idp_all.xml",
|
||||
xmlsec_binary=xmlsec_path)
|
||||
|
||||
sp = Saml2Client(config_file="servera_conf")
|
||||
idp = Server(config_file="idp_all_conf")
|
||||
|
||||
IDP_ENTITY_ID = idp.config.entityid
|
||||
SP_ENTITY_ID = sp.config.entityid
|
||||
|
||||
# ------------ @Client -----------------------------
|
||||
|
||||
headers = client.add_paos_headers([])
|
||||
|
||||
assert len(headers) == 2
|
||||
|
||||
# ------------ @SP -----------------------------
|
||||
|
||||
response = DummyResponse(set_list2dict(headers))
|
||||
|
||||
assert sp.can_handle_ecp_response(response)
|
||||
|
||||
id, message = sp.create_ecp_authn_request(IDP_ENTITY_ID, relay_state="XYZ")
|
||||
|
||||
# ------------ @Client -----------------------------
|
||||
|
||||
respdict = client.parse_soap_message(message)
|
||||
|
||||
cargs = client.parse_sp_ecp_response(respdict)
|
||||
|
||||
assert isinstance(respdict["body"], AuthnRequest)
|
||||
assert len(respdict["header"]) == 2
|
||||
item0 = respdict["header"][0]
|
||||
assert isinstance(item0, Request) or isinstance(item0, RelayState)
|
||||
|
||||
destination = respdict["body"].destination
|
||||
|
||||
ht_args = client.apply_binding(BINDING_SOAP, respdict["body"], destination)
|
||||
|
||||
# Time to send to the IDP
|
||||
# ----------- @IDP -------------------------------
|
||||
|
||||
req = idp.parse_authn_request(ht_args["data"], BINDING_SOAP)
|
||||
|
||||
assert isinstance(req.message, AuthnRequest)
|
||||
|
||||
# create Response and return in the SOAP response
|
||||
sp_entity_id = req.sender()
|
||||
|
||||
name_id = idp.ident.transient_nameid( "id12", sp.config.entityid)
|
||||
binding, destination = idp.pick_binding("assertion_consumer_service",
|
||||
[BINDING_PAOS],
|
||||
entity_id=sp_entity_id)
|
||||
|
||||
resp = idp.create_ecp_authn_request_response(destination,
|
||||
{
|
||||
"eduPersonEntitlement": "Short stop",
|
||||
"surName": "Jeter",
|
||||
"givenName": "Derek",
|
||||
"mail": "derek.jeter@nyy.mlb.com",
|
||||
"title": "The man"
|
||||
},
|
||||
req.message.id, destination, sp_entity_id,
|
||||
name_id=name_id, authn=(AUTHN_PASSWORD,
|
||||
"http://www.example.com/login"))
|
||||
|
||||
# ------------ @Client -----------------------------
|
||||
# The client got the response from the IDP repackage and send it to the SP
|
||||
|
||||
respdict = client.parse_soap_message(resp)
|
||||
idp_response = respdict["body"]
|
||||
|
||||
assert isinstance(idp_response, Response)
|
||||
assert len(respdict["header"]) == 1
|
||||
|
||||
_ecp_response = None
|
||||
for item in respdict["header"]:
|
||||
if item.c_tag == "Response" and item.c_namespace == ecp_prof.NAMESPACE:
|
||||
_ecp_response = item
|
||||
|
||||
_acs_url = _ecp_response.assertion_consumer_service_url
|
||||
|
||||
# done phase2 at the client
|
||||
|
||||
ht_args = client.use_soap(idp_response, cargs["rc_url"], [cargs["relay_state"]])
|
||||
|
||||
print ht_args
|
||||
|
||||
# ------------ @SP -----------------------------
|
||||
|
||||
respdict = sp.unpack_soap_message(ht_args["data"])
|
||||
|
||||
# verify the relay_state
|
||||
|
||||
for header in respdict["header"]:
|
||||
inst = create_class_from_xml_string(RelayState, header)
|
||||
if isinstance(inst, RelayState):
|
||||
assert inst.text == "XYZ"
|
||||
|
||||
# parse the response
|
||||
|
||||
resp = sp.parse_authn_request_response(respdict["body"], None, {id: "/"})
|
||||
|
||||
print resp.response
|
||||
|
||||
assert resp.response.destination == "http://lingon.catalogix.se:8087/paos"
|
||||
assert resp.response.status.status_code.value == STATUS_SUCCESS
|
||||
|
||||
Loading…
Reference in New Issue
Block a user