1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
|
from contextlib import closing
from pathutils import dotname
from pathutils import full_path
from saml2 import BINDING_PAOS
from saml2 import BINDING_SOAP
from saml2 import create_class_from_xml_string
from saml2 import ecp_client
from saml2.authn_context import INTERNETPROTOCOLPASSWORD
from saml2.client import Saml2Client
from saml2.httpbase import set_list2dict
from saml2.profile import ecp as ecp_prof
from saml2.profile.ecp import RelayState
from saml2.profile.paos import Request
from saml2.samlp import STATUS_SUCCESS
from saml2.samlp import AuthnRequest
from saml2.samlp import Response
from saml2.server import Server
__author__ = "rolandh"
AUTHN = {"class_ref": INTERNETPROTOCOLPASSWORD, "authn_auth": "http://www.example.com/login"}
def _eq(l1, l2):
if len(l1) == len(l2):
return set(l1) == set(l2)
else:
return len(l1) == len(l2)
class DummyResponse:
def __init__(self, headers):
self.headers = headers
def test_complete_flow():
client = ecp_client.Client("user", "password", metadata_file=full_path("idp_all.xml"))
sp = Saml2Client(config_file=dotname("servera_conf"))
with closing(Server(config_file=dotname("idp_all_conf"))) as idp:
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)
sid, 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,
)
# ------------ @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
# Explicitly allow unsigned responses for this test
sp.want_response_signed = False
resp = sp.parse_authn_request_response(respdict["body"], None, {sid: "/"})
print(resp.response)
assert resp.response.destination == "http://lingon.catalogix.se:8087/paos"
assert resp.response.status.status_code.value == STATUS_SUCCESS
|
