summaryrefslogtreecommitdiff
path: root/tests/test_63_ecp.py
blob: 0db36003ec630c48a1399f1e6c70670ee348e889 (plain)
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
from contextlib import closing
from saml2.authn_context import INTERNETPROTOCOLPASSWORD
from saml2.httpbase import set_list2dict
from saml2.profile.ecp import RelayState
from saml2.profile.paos import Request
from saml2.server import Server
from saml2.samlp import Response
from saml2.samlp import STATUS_SUCCESS
from saml2.samlp import AuthnRequest
from saml2 import ecp_client
from saml2 import BINDING_SOAP
from saml2 import BINDING_PAOS
from saml2 import create_class_from_xml_string

from saml2.profile import ecp as ecp_prof
from saml2.client import Saml2Client

from pathutils import dotname, full_path

__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(object):
    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