openstack/deb-python-pysaml2
Code Issues Proposed changes

Fixed artifact handling

Browse Source
This commit is contained in:
Roland Hedberg 2013-01-18 14:59:00 +01:00
parent 21880631d9
commit 13f327c0bc
12 changed files with 151 additions and 72 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
example/idp/idp.py
View File

@ -204,7 +204,7 @@ def slo(environ, start_response, user):
bindings, "spsso", response)
http_args = IDP.apply_binding(binding, "%s" % response, destination,
query["RelayState"], "SAMLResponse")
query["RelayState"], response=True)
except Exception, exc:
resp = BadRequest('%s' % exc)

14
example/idp2/idp.py
View File

@ -180,7 +180,7 @@ def _sso(environ, start_response, user, query, binding, relay_state=""):
binding, destination = IDP.pick_binding("assertion_consumer_service",
entity_id=_authn_req.issuer.text)
http_args = IDP.apply_binding(binding, "%s" % authn_resp, destination,
relay_state, "SAMLResponse")
relay_state, response=True)
resp = Response(http_args["data"], headers=http_args["headers"])
return resp(environ, start_response)
@ -495,7 +495,7 @@ def assertion_id_request(environ, start_response, user=None):
resp_args = IDP.response_args(req_info.message, _binding, "spsso")
response = IDP.create_assertion_id_request_response(asids, **resp_args)
hinfo = IDP.apply_binding(_binding, "%s" % response, "","","SAMLResponse")
hinfo = IDP.apply_binding(_binding, "%s" % response, "","",response=True)
resp = Response(hinfo["data"], headers=hinfo["headers"])
return resp(environ, start_response)
@ -522,7 +522,7 @@ def artifact_resolve_service(environ, start_response, user=None):
msg = IDP.create_artifact_response(_req, _req.artifact.text)
hinfo = IDP.apply_binding(_binding, "%s" % msg, "","","SAMLResponse")
hinfo = IDP.apply_binding(_binding, "%s" % msg, "","",response=True)
resp = Response(hinfo["data"], headers=hinfo["headers"])
return resp(environ, start_response)
@ -553,7 +553,7 @@ def authn_query_service(environ, start_response, user=None):
_query.session_index)
logger.debug("response: %s" % msg)
hinfo = IDP.apply_binding(_binding, "%s" % msg, "","","SAMLResponse")
hinfo = IDP.apply_binding(_binding, "%s" % msg, "","",response=True)
resp = Response(hinfo["data"], headers=hinfo["headers"])
return resp(environ, start_response)
@ -585,7 +585,7 @@ def _nim(environ, start_response, user, query, binding, relay_state=""):
_resp = IDP.create_manage_name_id_response(name_id, **info)
# It's using SOAP binding
hinfo = IDP.apply_binding(binding, "%s" % _resp, "", "", "SAMLResponse")
hinfo = IDP.apply_binding(binding, "%s" % _resp, "", "", response=True)
resp = Response(hinfo["data"],
headers=dict2list_of_tuples(hinfo["headers"]))
@ -629,8 +629,8 @@ def nim_art(environ, start_response, user):
_dict = unpack_redirect(environ)
if not _dict:
_dict = unpack_post()
return _mni
_dict = unpack_post(environ)
return _artifact_oper(environ, start_response, user, _dict, _mni)
# ----------------------------------------------------------------------------
# ----------------------------------------------------------------------------

2
src/saml2/config.py
View File

@ -138,7 +138,7 @@ PREFERRED_BINDING={
"attribute_service": [BINDING_SOAP],
"authz_service": [BINDING_SOAP],
"assertion_id_request_service": [BINDING_URI],
"artifact_resolution_service": [BINDING_HTTP_REDIRECT, BINDING_HTTP_POST],
"artifact_resolution_service": [BINDING_SOAP],
"attribute_consuming_service": _RPA
}

21
src/saml2/entity.py
View File

@ -4,7 +4,7 @@ from hashlib import sha1
from saml2.metadata import ENDPOINTS
from saml2.soap import parse_soap_enveloped_saml_artifact_resolve
from saml2 import samlp, saml, response, BINDING_URI
from saml2 import samlp, saml, response, BINDING_URI, BINDING_HTTP_ARTIFACT
from saml2 import request
from saml2 import soap
from saml2 import element_to_extension_element
@ -69,7 +69,6 @@ def create_artifact(entity_id, message_handle, endpoint_index = 0):
return base64.b64encode(ter)
class Entity(HTTPBase):
def __init__(self, entity_type, config=None, config_file="",
virtual_organization=""):
@ -121,7 +120,7 @@ class Entity(HTTPBase):
format=NAMEID_FORMAT_ENTITY)
def apply_binding(self, binding, msg_str, destination="", relay_state="",
typ="SAMLRequest"):
response=False):
"""
Construct the necessary HTTP arguments dependent on Binding
@ -129,9 +128,15 @@ class Entity(HTTPBase):
:param msg_str: The return message as a string (XML)
:param destination: Where to send the message
:param relay_state: Relay_state if provided
:param typ: Which type of message this is
:param response: Which type of message this is
:return: A dictionary
"""
# unless if BINDING_HTTP_ARTIFACT
if response:
typ = "SAMLResponse"
else:
typ = "SAMLRequest"
if binding == BINDING_HTTP_POST:
logger.info("HTTP POST")
info = self.use_http_form_post(msg_str, destination,
@ -147,6 +152,14 @@ class Entity(HTTPBase):
info = self.use_soap(msg_str, destination)
elif binding == BINDING_URI:
info = self.use_http_uri(msg_str, typ, destination)
elif binding == BINDING_HTTP_ARTIFACT:
typ = "SAMLart"
if response:
info = self.use_http_artifact(msg_str, destination, relay_state)
info["method"] = "GET"
info["status"] = 302
else:
info = self.use_http_artifact(msg_str, destination, relay_state)
else:
raise Exception("Unknown binding type: %s" % binding)

22
src/saml2/httpbase.py
View File

@ -2,6 +2,7 @@ import calendar
import cookielib
import copy
import re
import urllib
import urlparse
import requests
import time
@ -206,7 +207,19 @@ class HTTPBase(object):
return http_redirect_message(message, destination, relay_state, typ)
def use_http_uri(self, message, typ, destination=""):
def use_http_artifact(self, message, destination="", relay_state=""):
if relay_state:
query = urllib.urlencode({"SAMLart": message,
"RelayState": relay_state})
else:
query = urllib.urlencode({"SAMLart": message})
info = {
"data": "",
"url": "%s?%s" % (destination, query)
}
return info
def use_http_uri(self, message, typ, destination="", relay_state=""):
if typ == "SAMLResponse":
info = {
"data": message.split("\n")[1],
@ -218,9 +231,14 @@ class HTTPBase(object):
}
elif typ == "SAMLRequest":
# msg should be an identifier
if relay_state:
query = urllib.urlencode({"ID": message,
"RelayState": relay_state})
else:
query = urllib.urlencode({"ID": message})
info = {
"data": "",
"url": "%s?ID=%s" % (destination, message)
"url": "%s?%s" % (destination, query)
}
else:
raise NotImplemented

2
src/saml2/pack.py
View File

@ -124,7 +124,7 @@ def http_redirect_message(message, location, relay_state="", typ="SAMLRequest"):
glue_char = "&" if urlparse.urlparse(location).query else "?"
login_url = glue_char.join([location, urllib.urlencode(args)])
headers = [('Location', login_url)]
body = [""]
body = []
return {"headers":headers, "data":body}

1
src/saml2/soap.py
View File

@ -125,6 +125,7 @@ def parse_soap_enveloped_saml_thingy(text, expected_tags):
"""
envelope = ElementTree.fromstring(text)
# Make sure it's a SOAP message
assert envelope.tag == '{%s}Envelope' % soapenv.NAMESPACE
assert len(envelope) >= 1

4
tests/test_50_server.py
View File

@ -458,8 +458,8 @@ class TestServerLogout():
None, "spsso", request)
http_args = server.apply_binding(binding, "%s" % response, destination,
"relay_state", "SAMLResponse")
"relay_state", response=True)
assert len(http_args) == 4
assert http_args["headers"][0][0] == "Location"
assert http_args["data"] == ['']
assert http_args["data"] == []

2
tests/test_51_client.py
View File

@ -349,7 +349,7 @@ class TestClientWithDummy():
assert isinstance(id, basestring)
assert len(http_args) == 4
assert http_args["headers"][0][0] == "Location"
assert http_args["data"] == [""]
assert http_args["data"] == []
def test_do_attribute_query(self):
response = self.client.do_attribute_query(IDP,

148
tests/test_64_artifact.py
View File

@ -1,10 +1,11 @@
import base64
from hashlib import sha1
import urlparse
from urlparse import urlparse
from urlparse import parse_qs
from saml2.saml import AUTHN_PASSWORD
from saml2 import BINDING_HTTP_REDIRECT
from saml2 import BINDING_HTTP_ARTIFACT
from saml2 import BINDING_SOAP
from saml2 import BINDING_HTTP_POST
from saml2.pack import http_redirect_message
from saml2.client import Saml2Client
from saml2.entity import create_artifact
@ -14,6 +15,34 @@ from saml2.server import Server
__author__ = 'rolandh'
TAG1 = "name=\"SAMLRequest\" value="
def get_msg(hinfo, binding, response=False):
if binding == BINDING_SOAP:
msg = hinfo["data"]
elif binding == BINDING_HTTP_POST:
_inp = hinfo["data"][3]
i = _inp.find(TAG1)
i += len(TAG1) + 1
j = _inp.find('"', i)
msg = _inp[i:j]
elif binding == BINDING_HTTP_ARTIFACT:
# either by POST or by redirect
if hinfo["data"]:
_inp = hinfo["data"][3]
i = _inp.find(TAG1)
i += len(TAG1) + 1
j = _inp.find('"', i)
msg = _inp[i:j]
else:
parts = urlparse(hinfo["url"])
msg = parse_qs(parts.query)["SAMLart"][0]
else: # BINDING_HTTP_REDIRECT
parts = urlparse(hinfo["headers"][0][1])
msg = parse_qs(parts.query)["SAMLRequest"][0]
return msg
def test_create_artifact():
b64art = create_artifact("http://sp.example.com/saml.xml",
"aabbccddeeffgghhiijj")
@ -60,64 +89,71 @@ def test_create_artifact_resolve():
assert ar.artifact.text == b64art
def test_artifact_flow():
SP = 'urn:mace:example.com:saml:roland:sp'
sp = Saml2Client(config_file="servera_conf")
idp = Server(config_file="idp_all_conf")
# ======= SP ==========
# original request
srvs = sp.metadata.single_sign_on_service(idp.config.entityid,
BINDING_HTTP_REDIRECT)
destination=srvs[0]["location"]
req = sp.create_authn_request(destination, id = "id1")
binding, destination = sp.pick_binding("single_sign_on_service",
entity_id=idp.config.entityid)
relay_state = "RS0"
req = sp.create_authn_request(destination, id="id1")
# create the artifact
artifact = sp.use_artifact(req, 1)
# HTTP args for sending the message with the artifact
args = http_redirect_message(artifact, destination, "really", "SAMLart")
# ====== IDP =========
# simulating the IDP receiver
artifact2 = None
for item, val in args["headers"]:
if item == "Location":
part = urlparse.urlparse(val)
query = urlparse.parse_qs(part.query)
artifact2 = query["SAMLart"][0]
binding, destination = sp.pick_binding("single_sign_on_service",
[BINDING_HTTP_ARTIFACT],
entity_id=idp.config.entityid)
hinfo = sp.apply_binding(binding, "%s" % artifact, destination, relay_state)
# ========== @IDP ============
artifact2 = get_msg(hinfo, binding)
assert artifact == artifact2
# The IDP now wants to replace the artifact with the real request
# Got an artifact, now want to get the original request
destination = idp.artifact2destination(artifact2, "spsso")
msg = idp.create_artifact_resolve(artifact2, destination, sid())
args = idp.use_soap(msg, destination, None, False)
hinfo = idp.use_soap(msg, destination, None, False)
# ======== SP ==========
# ======== @SP ==========
ar = sp.parse_artifact_resolve(args["data"])
msg = get_msg(hinfo, BINDING_SOAP)
print ar
ar = sp.parse_artifact_resolve(msg)
assert ar.artifact.text == artifact
# The SP picks the request out of the repository with the artifact as the key
oreq = sp.artifact[ar.artifact.text]
# Should be the same as req above
# Returns the information over the existing SOAP connection so
# no transport information needed
msg = sp.create_artifact_response(ar, ar.artifact.text)
args = sp.use_soap(msg, destination)
hinfo = sp.use_soap(msg, destination)
# ========== IDP ============
# ========== @IDP ============
spreq = idp.parse_artifact_resolve_response(args["data"])
msg = get_msg(hinfo, BINDING_SOAP)
# The IDP untangles the request from the artifact resolve response
spreq = idp.parse_artifact_resolve_response(msg)
# should be the same as req above
print spreq
assert spreq.id == req.id
# That was one way
# ------------------------------------
# Now for the other
# That was one way, the Request from the SP
# ---------------------------------------------#
# Now for the other, the response from the IDP
name_id = idp.ident.transient_nameid(sp.config.entityid, "derek")
@ -134,44 +170,54 @@ def test_artifact_flow():
print response
artifact = idp.use_artifact(response, 1)
args = http_redirect_message(artifact, resp_args["destination"], "really2",
"SAMLart")
# with the response in hand create an artifact
artifact2=None
for item, val in args["headers"]:
if item == "Location":
part = urlparse.urlparse(val)
query = urlparse.parse_qs(part.query)
artifact2 = query["SAMLart"][0]
artifact = idp.use_artifact(response, 1)
binding, destination = sp.pick_binding("single_sign_on_service",
[BINDING_HTTP_ARTIFACT],
entity_id=idp.config.entityid)
hinfo = sp.apply_binding(binding, "%s" % artifact, destination, relay_state,
response=True)
# ========== SP =========
destination = sp.artifact2destination(artifact2, "idpsso")
artifact3 = get_msg(hinfo, binding)
msg = sp.create_artifact_resolve(artifact2, destination, sid())
assert artifact == artifact3
destination = sp.artifact2destination(artifact3, "idpsso")
# Got an artifact want to replace it with the real message
msg = sp.create_artifact_resolve(artifact3, destination, sid())
print msg
args = sp.use_soap(msg, destination, None, False)
hinfo = sp.use_soap(msg, destination, None, False)
# ======== IDP ==========
ar = idp.parse_artifact_resolve(args["data"])
msg = get_msg(hinfo, BINDING_SOAP)
ar = idp.parse_artifact_resolve(msg)
print ar
assert ar.artifact.text == artifact
assert ar.artifact.text == artifact3
# The IDP retrieves the response from the database using the artifact as the key
oreq = idp.artifact[ar.artifact.text]
# Should be the same as req above
msg = idp.create_artifact_response(ar, ar.artifact.text)
args = idp.use_soap(msg, destination)
binding, destination = idp.pick_binding("artifact_resolution_service",
entity_id=sp.config.entityid)
resp = idp.create_artifact_response(ar, ar.artifact.text)
hinfo = idp.use_soap(resp, destination)
# ========== SP ============
sp_resp = sp.parse_artifact_resolve_response(args["data"])
msg = get_msg(hinfo, BINDING_SOAP)
sp_resp = sp.parse_artifact_resolve_response(msg)
assert sp_resp.id == response.id

3
tests/test_65_authn_query.py
View File

@ -119,7 +119,8 @@ def test_flow():
print p_res
hinfo = idp.apply_binding(binding, "%s" % p_res, "", "state2", "SAMLResponse")
hinfo = idp.apply_binding(binding, "%s" % p_res, "", "state2",
response=True)
# ------- @SP ----------

2
tests/test_68_assertion_id.py
View File

@ -91,7 +91,7 @@ def test_basic_flow():
resp = idp.create_assertion_id_request_response(aid)
hinfo = idp.apply_binding(binding, "%s" % resp, None, "", "SAMLResponse")
hinfo = idp.apply_binding(binding, "%s" % resp, None, "", response=True)
# ----------- @SP -------------