1272 lines
48 KiB
Python
1272 lines
48 KiB
Python
#!/usr/bin/env python
|
|
# -*- coding: utf-8 -*-
|
|
import base64
|
|
import copy
|
|
import os
|
|
from contextlib import closing
|
|
from six.moves.urllib.parse import parse_qs
|
|
import uuid
|
|
|
|
from saml2.cert import OpenSSLWrapper
|
|
from saml2.sigver import make_temp, EncryptError, CertificateError
|
|
from saml2.assertion import Policy
|
|
from saml2.authn_context import INTERNETPROTOCOLPASSWORD
|
|
from saml2.saml import NameID, NAMEID_FORMAT_TRANSIENT
|
|
from saml2.samlp import response_from_string
|
|
|
|
from saml2.server import Server
|
|
from saml2 import samlp
|
|
from saml2 import saml
|
|
from saml2 import client
|
|
from saml2 import config
|
|
from saml2 import extension_elements_to_elements
|
|
from saml2 import s_utils
|
|
from saml2 import sigver
|
|
from saml2 import time_util
|
|
from saml2.s_utils import OtherError
|
|
from saml2.s_utils import do_attribute_statement
|
|
from saml2.s_utils import factory
|
|
from saml2.soap import make_soap_enveloped_saml_thingy
|
|
from saml2 import BINDING_HTTP_POST
|
|
from saml2 import BINDING_HTTP_REDIRECT
|
|
|
|
from py.test import raises
|
|
from pathutils import full_path
|
|
import saml2.xmldsig as ds
|
|
|
|
nid = NameID(name_qualifier="foo", format=NAMEID_FORMAT_TRANSIENT,
|
|
text="123456")
|
|
|
|
AUTHN = {
|
|
"class_ref": INTERNETPROTOCOLPASSWORD,
|
|
"authn_auth": "http://www.example.com/login"
|
|
}
|
|
|
|
|
|
def _eq(l1, l2):
|
|
return set(l1) == set(l2)
|
|
|
|
|
|
BASEDIR = os.path.abspath(os.path.dirname(__file__))
|
|
|
|
|
|
def get_ava(assertion):
|
|
ava = {}
|
|
for statement in assertion.attribute_statement:
|
|
for attr in statement.attribute:
|
|
value = []
|
|
for tmp_val in attr.attribute_value:
|
|
value.append(tmp_val.text)
|
|
key = attr.friendly_name
|
|
if key is None or len(key) == 0:
|
|
key = attr.text
|
|
ava[key] = value
|
|
return ava
|
|
|
|
|
|
def generate_cert():
|
|
sn = uuid.uuid4().urn
|
|
cert_info = {
|
|
"cn": "localhost",
|
|
"country_code": "se",
|
|
"state": "ac",
|
|
"city": "Umea",
|
|
"organization": "ITS",
|
|
"organization_unit": "DIRG"
|
|
}
|
|
osw = OpenSSLWrapper()
|
|
ca_cert_str = osw.read_str_from_file(
|
|
full_path("root_cert/localhost.ca.crt"))
|
|
ca_key_str = osw.read_str_from_file(
|
|
full_path("root_cert/localhost.ca.key"))
|
|
req_cert_str, req_key_str = osw.create_certificate(cert_info, request=True,
|
|
sn=sn, key_length=2048)
|
|
cert_str = osw.create_cert_signed_certificate(ca_cert_str, ca_key_str,
|
|
req_cert_str)
|
|
return cert_str, req_key_str
|
|
|
|
|
|
class TestServer1():
|
|
|
|
def setup_class(self):
|
|
self.server = Server("idp_conf")
|
|
|
|
conf = config.SPConfig()
|
|
conf.load_file("server_conf")
|
|
self.client = client.Saml2Client(conf)
|
|
self.name_id = self.server.ident.transient_nameid(
|
|
"urn:mace:example.com:saml:roland:sp", "id12")
|
|
self.ava = {"givenName": ["Derek"], "sn": ["Jeter"],
|
|
"mail": ["derek@nyy.mlb.com"], "title": "The man"}
|
|
|
|
def teardown_class(self):
|
|
self.server.close()
|
|
|
|
def verify_assertion(self, assertion):
|
|
assert assertion
|
|
assert assertion[0].attribute_statement
|
|
|
|
ava = ava = get_ava(assertion[0])
|
|
|
|
assert ava ==\
|
|
{'mail': ['derek@nyy.mlb.com'], 'givenName': ['Derek'],
|
|
'sn': ['Jeter'], 'title': ['The man']}
|
|
|
|
|
|
def verify_encrypted_assertion(self, assertion, decr_text):
|
|
self.verify_assertion(assertion)
|
|
assert assertion[0].signature is None
|
|
|
|
assert 'EncryptedAssertion><encas1:Assertion xmlns:encas0="http://www.w3.org/2001/XMLSchema-instance" ' \
|
|
'xmlns:encas1="urn:oasis:names:tc:SAML:2.0:assertion"' in decr_text
|
|
|
|
def verify_advice_assertion(self, resp, decr_text):
|
|
assert resp.assertion[0].signature is None
|
|
|
|
assert resp.assertion[0].advice.encrypted_assertion[0].extension_elements
|
|
|
|
assertion = extension_elements_to_elements(resp.assertion[0].advice.encrypted_assertion[0].extension_elements,
|
|
[saml, samlp])
|
|
self.verify_encrypted_assertion(assertion, decr_text)
|
|
|
|
|
|
def test_issuer(self):
|
|
issuer = self.server._issuer()
|
|
assert isinstance(issuer, saml.Issuer)
|
|
assert _eq(issuer.keyswv(), ["text", "format"])
|
|
assert issuer.format == saml.NAMEID_FORMAT_ENTITY
|
|
assert issuer.text == self.server.config.entityid
|
|
|
|
def test_assertion(self):
|
|
assertion = s_utils.assertion_factory(
|
|
subject=factory(
|
|
saml.Subject, text="_aaa",
|
|
name_id=factory(saml.NameID,
|
|
format=saml.NAMEID_FORMAT_TRANSIENT)),
|
|
attribute_statement=do_attribute_statement(
|
|
{
|
|
("", "", "sn"): ("Jeter", ""),
|
|
("", "", "givenName"): ("Derek", ""),
|
|
}
|
|
),
|
|
issuer=self.server._issuer(),
|
|
)
|
|
|
|
assert _eq(assertion.keyswv(), ['attribute_statement', 'issuer', 'id',
|
|
'subject', 'issue_instant', 'version'])
|
|
assert assertion.version == "2.0"
|
|
assert assertion.issuer.text == "urn:mace:example.com:saml:roland:idp"
|
|
#
|
|
assert assertion.attribute_statement
|
|
attribute_statement = assertion.attribute_statement
|
|
assert len(attribute_statement.attribute) == 2
|
|
attr0 = attribute_statement.attribute[0]
|
|
attr1 = attribute_statement.attribute[1]
|
|
if attr0.attribute_value[0].text == "Derek":
|
|
assert attr0.friendly_name == "givenName"
|
|
assert attr1.friendly_name == "sn"
|
|
assert attr1.attribute_value[0].text == "Jeter"
|
|
else:
|
|
assert attr1.friendly_name == "givenName"
|
|
assert attr1.attribute_value[0].text == "Derek"
|
|
assert attr0.friendly_name == "sn"
|
|
assert attr0.attribute_value[0].text == "Jeter"
|
|
#
|
|
subject = assertion.subject
|
|
assert _eq(subject.keyswv(), ["text", "name_id"])
|
|
assert subject.text == "_aaa"
|
|
assert subject.name_id.format == saml.NAMEID_FORMAT_TRANSIENT
|
|
|
|
def test_response(self):
|
|
response = sigver.response_factory(
|
|
in_response_to="_012345",
|
|
destination="https:#www.example.com",
|
|
status=s_utils.success_status_factory(),
|
|
assertion=s_utils.assertion_factory(
|
|
subject=factory(saml.Subject, text="_aaa",
|
|
name_id=saml.NAMEID_FORMAT_TRANSIENT),
|
|
attribute_statement=do_attribute_statement(
|
|
{
|
|
("", "", "sn"): ("Jeter", ""),
|
|
("", "", "givenName"): ("Derek", ""),
|
|
}
|
|
),
|
|
issuer=self.server._issuer(),
|
|
),
|
|
issuer=self.server._issuer(),
|
|
)
|
|
|
|
print(response.keyswv())
|
|
assert _eq(response.keyswv(), ['destination', 'assertion', 'status',
|
|
'in_response_to', 'issue_instant',
|
|
'version', 'issuer', 'id'])
|
|
assert response.version == "2.0"
|
|
assert response.issuer.text == "urn:mace:example.com:saml:roland:idp"
|
|
assert response.destination == "https:#www.example.com"
|
|
assert response.in_response_to == "_012345"
|
|
#
|
|
status = response.status
|
|
print(status)
|
|
assert status.status_code.value == samlp.STATUS_SUCCESS
|
|
|
|
def test_parse_faulty_request(self):
|
|
req_id, authn_request = self.client.create_authn_request(
|
|
destination="http://www.example.com", id="id1")
|
|
|
|
# should raise an error because faulty spentityid
|
|
binding = BINDING_HTTP_REDIRECT
|
|
htargs = self.client.apply_binding(
|
|
binding, "%s" % authn_request, "http://www.example.com", "abcd")
|
|
_dict = parse_qs(htargs["headers"][0][1].split('?')[1])
|
|
print(_dict)
|
|
raises(OtherError, self.server.parse_authn_request,
|
|
_dict["SAMLRequest"][0], binding)
|
|
|
|
def test_parse_faulty_request_to_err_status(self):
|
|
req_id, authn_request = self.client.create_authn_request(
|
|
destination="http://www.example.com")
|
|
|
|
binding = BINDING_HTTP_REDIRECT
|
|
htargs = self.client.apply_binding(binding, "%s" % authn_request,
|
|
"http://www.example.com", "abcd")
|
|
_dict = parse_qs(htargs["headers"][0][1].split('?')[1])
|
|
print(_dict)
|
|
|
|
try:
|
|
self.server.parse_authn_request(_dict["SAMLRequest"][0], binding)
|
|
status = None
|
|
except OtherError as oe:
|
|
print(oe.args)
|
|
status = s_utils.error_status_factory(oe)
|
|
|
|
assert status
|
|
print(status)
|
|
assert _eq(status.keyswv(), ["status_code", "status_message"])
|
|
assert status.status_message.text == 'Not destined for me!'
|
|
status_code = status.status_code
|
|
assert _eq(status_code.keyswv(), ["status_code", "value"])
|
|
assert status_code.value == samlp.STATUS_RESPONDER
|
|
assert status_code.status_code.value == samlp.STATUS_UNKNOWN_PRINCIPAL
|
|
|
|
def test_parse_ok_request(self):
|
|
req_id, authn_request = self.client.create_authn_request(
|
|
message_id="id1", destination="http://localhost:8088/sso")
|
|
|
|
print(authn_request)
|
|
binding = BINDING_HTTP_REDIRECT
|
|
htargs = self.client.apply_binding(binding, "%s" % authn_request,
|
|
"http://www.example.com", "abcd")
|
|
_dict = parse_qs(htargs["headers"][0][1].split('?')[1])
|
|
print(_dict)
|
|
|
|
req = self.server.parse_authn_request(_dict["SAMLRequest"][0], binding)
|
|
# returns a dictionary
|
|
print(req)
|
|
resp_args = self.server.response_args(req.message, [BINDING_HTTP_POST])
|
|
assert resp_args["destination"] == "http://lingon.catalogix.se:8087/"
|
|
assert resp_args["in_response_to"] == "id1"
|
|
name_id_policy = resp_args["name_id_policy"]
|
|
assert _eq(name_id_policy.keyswv(), ["format", "allow_create"])
|
|
assert name_id_policy.format == saml.NAMEID_FORMAT_TRANSIENT
|
|
assert resp_args[
|
|
"sp_entity_id"] == "urn:mace:example.com:saml:roland:sp"
|
|
|
|
def test_sso_response_with_identity(self):
|
|
name_id = self.server.ident.transient_nameid(
|
|
"https://example.com/sp", "id12")
|
|
resp = self.server.create_authn_response(
|
|
{
|
|
"eduPersonEntitlement": "Short stop",
|
|
"sn": "Jeter",
|
|
"givenName": "Derek",
|
|
"mail": "derek.jeter@nyy.mlb.com",
|
|
"title": "The man"
|
|
},
|
|
"id12", # in_response_to
|
|
"http://localhost:8087/", # destination
|
|
"https://example.com/sp", # sp_entity_id
|
|
name_id=name_id,
|
|
authn=AUTHN
|
|
)
|
|
|
|
print(resp.keyswv())
|
|
assert _eq(resp.keyswv(), ['status', 'destination', 'assertion',
|
|
'in_response_to', 'issue_instant',
|
|
'version', 'id', 'issuer'])
|
|
assert resp.destination == "http://localhost:8087/"
|
|
assert resp.in_response_to == "id12"
|
|
assert resp.status
|
|
assert resp.status.status_code.value == samlp.STATUS_SUCCESS
|
|
assert resp.assertion
|
|
assertion = resp.assertion
|
|
print(assertion)
|
|
assert assertion.authn_statement
|
|
assert assertion.conditions
|
|
assert assertion.attribute_statement
|
|
attribute_statement = assertion.attribute_statement
|
|
print(attribute_statement)
|
|
assert len(attribute_statement[0].attribute) == 4
|
|
# Pick out one attribute
|
|
attr = None
|
|
for attr in attribute_statement[0].attribute:
|
|
if attr.friendly_name == "givenName":
|
|
break
|
|
assert len(attr.attribute_value) == 1
|
|
assert attr.name == "urn:mace:dir:attribute-def:givenName"
|
|
assert attr.name_format == "urn:oasis:names:tc:SAML:2.0:attrname-format:basic"
|
|
value = attr.attribute_value[0]
|
|
assert value.text.strip() == "Derek"
|
|
assert value.get_type() == "xs:string"
|
|
assert assertion.subject
|
|
assert assertion.subject.name_id
|
|
assert assertion.subject.subject_confirmation
|
|
confirmation = assertion.subject.subject_confirmation[0]
|
|
print(confirmation.keyswv())
|
|
print(confirmation.subject_confirmation_data)
|
|
assert confirmation.subject_confirmation_data.in_response_to == "id12"
|
|
|
|
def test_sso_response_without_identity(self):
|
|
resp = self.server.create_authn_response(
|
|
{},
|
|
"id12", # in_response_to
|
|
"http://localhost:8087/", # consumer_url
|
|
"urn:mace:example.com:saml:roland:sp", # sp_entity_id
|
|
userid="USER1",
|
|
authn=AUTHN,
|
|
release_policy=Policy(),
|
|
best_effort=True
|
|
)
|
|
|
|
print(resp.keyswv())
|
|
assert _eq(resp.keyswv(), ['status', 'destination', 'in_response_to',
|
|
'issue_instant', 'version', 'id', 'issuer',
|
|
'assertion'])
|
|
assert resp.destination == "http://localhost:8087/"
|
|
assert resp.in_response_to == "id12"
|
|
assert resp.status
|
|
assert resp.status.status_code.value == samlp.STATUS_SUCCESS
|
|
assert resp.issuer.text == "urn:mace:example.com:saml:roland:idp"
|
|
assert not resp.assertion.attribute_statement
|
|
|
|
def test_sso_response_specific_instant(self):
|
|
_authn = AUTHN.copy()
|
|
_authn["authn_instant"] = 1234567890
|
|
|
|
resp = self.server.create_authn_response(
|
|
{},
|
|
"id12", # in_response_to
|
|
"http://localhost:8087/", # consumer_url
|
|
"urn:mace:example.com:saml:roland:sp", # sp_entity_id
|
|
userid="USER1",
|
|
authn=_authn,
|
|
best_effort=True
|
|
)
|
|
|
|
print(resp.keyswv())
|
|
assert _eq(resp.keyswv(), ['status', 'destination', 'in_response_to',
|
|
'issue_instant', 'version', 'id', 'issuer',
|
|
'assertion'])
|
|
authn_statement = resp.assertion.authn_statement[0]
|
|
assert authn_statement.authn_instant == '2009-02-13T23:31:30Z'
|
|
|
|
def test_sso_failure_response(self):
|
|
exc = s_utils.MissingValue("eduPersonAffiliation missing")
|
|
resp = self.server.create_error_response(
|
|
"id12", "http://localhost:8087/", exc)
|
|
|
|
print(resp.keyswv())
|
|
assert _eq(resp.keyswv(), ['status', 'destination', 'in_response_to',
|
|
'issue_instant', 'version', 'id', 'issuer'])
|
|
assert resp.destination == "http://localhost:8087/"
|
|
assert resp.in_response_to == "id12"
|
|
assert resp.status
|
|
print(resp.status)
|
|
assert resp.status.status_code.value == samlp.STATUS_RESPONDER
|
|
assert resp.status.status_code.status_code.value == \
|
|
samlp.STATUS_REQUEST_UNSUPPORTED
|
|
assert resp.status.status_message.text == \
|
|
"eduPersonAffiliation missing"
|
|
assert resp.issuer.text == "urn:mace:example.com:saml:roland:idp"
|
|
assert not resp.assertion
|
|
|
|
def test_authn_response_0(self):
|
|
conf = config.SPConfig()
|
|
conf.load_file("server_conf")
|
|
self.client = client.Saml2Client(conf)
|
|
|
|
ava = {"givenName": ["Derek"], "sn": ["Jeter"],
|
|
"mail": ["derek@nyy.mlb.com"], "title": "The man"}
|
|
|
|
npolicy = samlp.NameIDPolicy(format=saml.NAMEID_FORMAT_TRANSIENT,
|
|
allow_create="true")
|
|
resp_str = "%s" % self.server.create_authn_response(
|
|
ava, "id1", "http://local:8087/",
|
|
"urn:mace:example.com:saml:roland:sp", npolicy,
|
|
"foba0001@example.com", authn=AUTHN)
|
|
|
|
response = samlp.response_from_string(resp_str)
|
|
print(response.keyswv())
|
|
assert _eq(response.keyswv(), ['status', 'destination', 'assertion',
|
|
'in_response_to', 'issue_instant',
|
|
'version', 'issuer', 'id'])
|
|
print(response.assertion[0].keyswv())
|
|
assert len(response.assertion) == 1
|
|
assert _eq(response.assertion[0].keyswv(), ['attribute_statement',
|
|
'issue_instant', 'version',
|
|
'subject', 'conditions',
|
|
'id', 'issuer',
|
|
'authn_statement'])
|
|
assertion = response.assertion[0]
|
|
assert len(assertion.attribute_statement) == 1
|
|
astate = assertion.attribute_statement[0]
|
|
print(astate)
|
|
assert len(astate.attribute) == 4
|
|
|
|
def test_signed_response(self):
|
|
name_id = self.server.ident.transient_nameid(
|
|
"urn:mace:example.com:saml:roland:sp", "id12")
|
|
ava = {"givenName": ["Derek"], "sn": ["Jeter"],
|
|
"mail": ["derek@nyy.mlb.com"], "title": "The man"}
|
|
|
|
signed_resp = self.server.create_authn_response(
|
|
ava,
|
|
"id12", # in_response_to
|
|
"http://lingon.catalogix.se:8087/", # consumer_url
|
|
"urn:mace:example.com:saml:roland:sp", # sp_entity_id
|
|
name_id=name_id,
|
|
sign_assertion=True
|
|
)
|
|
|
|
print(signed_resp)
|
|
assert signed_resp
|
|
|
|
sresponse = response_from_string(signed_resp)
|
|
# It's the assertions that are signed not the response per se
|
|
assert len(sresponse.assertion) == 1
|
|
assertion = sresponse.assertion[0]
|
|
|
|
# Since the reponse is created dynamically I don't know the signature
|
|
# value. Just that there should be one
|
|
assert assertion.signature.signature_value.text != ""
|
|
|
|
def test_signed_response_1(self):
|
|
|
|
|
|
signed_resp = self.server.create_authn_response(
|
|
self.ava,
|
|
"id12", # in_response_to
|
|
"http://lingon.catalogix.se:8087/", # consumer_url
|
|
"urn:mace:example.com:saml:roland:sp", # sp_entity_id
|
|
name_id=self.name_id,
|
|
sign_response=True,
|
|
sign_assertion=True,
|
|
)
|
|
|
|
sresponse = response_from_string(signed_resp)
|
|
|
|
valid = self.server.sec.verify_signature(signed_resp,
|
|
self.server.config.cert_file,
|
|
node_name='urn:oasis:names:tc:SAML:2.0:protocol:Response',
|
|
node_id=sresponse.id,
|
|
id_attr="")
|
|
assert valid
|
|
|
|
valid = self.server.sec.verify_signature(signed_resp,
|
|
self.server.config.cert_file,
|
|
node_name='urn:oasis:names:tc:SAML:2.0:assertion:Assertion',
|
|
node_id=sresponse.assertion[0].id,
|
|
id_attr="")
|
|
assert valid
|
|
|
|
self.verify_assertion(sresponse.assertion)
|
|
|
|
def test_signed_response_2(self):
|
|
signed_resp = self.server.create_authn_response(
|
|
self.ava,
|
|
"id12", # in_response_to
|
|
"http://lingon.catalogix.se:8087/", # consumer_url
|
|
"urn:mace:example.com:saml:roland:sp", # sp_entity_id
|
|
name_id=self.name_id,
|
|
sign_response=True,
|
|
sign_assertion=False,
|
|
)
|
|
|
|
sresponse = response_from_string(signed_resp)
|
|
|
|
valid = self.server.sec.verify_signature(signed_resp,
|
|
self.server.config.cert_file,
|
|
node_name='urn:oasis:names:tc:SAML:2.0:protocol:Response',
|
|
node_id=sresponse.id,
|
|
id_attr="")
|
|
assert valid
|
|
|
|
assert sresponse.assertion[0].signature == None
|
|
|
|
def test_signed_response_3(self):
|
|
|
|
|
|
signed_resp = self.server.create_authn_response(
|
|
self.ava,
|
|
"id12", # in_response_to
|
|
"http://lingon.catalogix.se:8087/", # consumer_url
|
|
"urn:mace:example.com:saml:roland:sp", # sp_entity_id
|
|
name_id=self.name_id,
|
|
sign_response=False,
|
|
sign_assertion=True,
|
|
)
|
|
|
|
sresponse = response_from_string(signed_resp)
|
|
|
|
assert sresponse.signature == None
|
|
|
|
valid = self.server.sec.verify_signature(signed_resp,
|
|
self.server.config.cert_file,
|
|
node_name='urn:oasis:names:tc:SAML:2.0:assertion:Assertion',
|
|
node_id=sresponse.assertion[0].id,
|
|
id_attr="")
|
|
assert valid
|
|
|
|
self.verify_assertion(sresponse.assertion)
|
|
|
|
def test_encrypted_signed_response_1(self):
|
|
|
|
cert_str, cert_key_str = generate_cert()
|
|
|
|
signed_resp = self.server.create_authn_response(
|
|
self.ava,
|
|
"id12", # in_response_to
|
|
"http://lingon.catalogix.se:8087/", # consumer_url
|
|
"urn:mace:example.com:saml:roland:sp", # sp_entity_id
|
|
name_id=self.name_id,
|
|
sign_response=True,
|
|
sign_assertion=True,
|
|
encrypt_assertion=False,
|
|
encrypt_assertion_self_contained=True,
|
|
pefim=True,
|
|
encrypt_cert_advice=cert_str,
|
|
)
|
|
|
|
sresponse = response_from_string(signed_resp)
|
|
|
|
valid = self.server.sec.verify_signature(
|
|
signed_resp, self.server.config.cert_file,
|
|
node_name='urn:oasis:names:tc:SAML:2.0:protocol:Response',
|
|
node_id=sresponse.id, id_attr="")
|
|
|
|
assert valid
|
|
|
|
valid = self.server.sec.verify_signature(
|
|
signed_resp, self.server.config.cert_file,
|
|
node_name='urn:oasis:names:tc:SAML:2.0:assertion:Assertion',
|
|
node_id=sresponse.assertion[0].id, id_attr="")
|
|
|
|
assert valid
|
|
|
|
_, key_file = make_temp(cert_key_str, decode=False)
|
|
|
|
decr_text = self.server.sec.decrypt(signed_resp, key_file)
|
|
|
|
resp = samlp.response_from_string(decr_text)
|
|
|
|
assert resp.assertion[0].advice.encrypted_assertion[0].extension_elements
|
|
|
|
assertion = extension_elements_to_elements(
|
|
resp.assertion[0].advice.encrypted_assertion[0].extension_elements,
|
|
[saml, samlp])
|
|
|
|
self.verify_assertion(assertion)
|
|
|
|
|
|
|
|
#PEFIM never signs assertions.
|
|
assert assertion[0].signature is None
|
|
#valid = self.server.sec.verify_signature(decr_text,
|
|
# self.server.config.cert_file,
|
|
# node_name='urn:oasis:names:tc:SAML:2.0:assertion:Assertion',
|
|
# node_id=assertion[0].id,
|
|
# id_attr="")
|
|
assert valid
|
|
|
|
def test_encrypted_signed_response_2(self):
|
|
cert_str, cert_key_str = generate_cert()
|
|
|
|
signed_resp = self.server.create_authn_response(
|
|
self.ava,
|
|
"id12", # in_response_to
|
|
"http://lingon.catalogix.se:8087/", # consumer_url
|
|
"urn:mace:example.com:saml:roland:sp", # sp_entity_id
|
|
name_id=self.name_id,
|
|
sign_response=True,
|
|
sign_assertion=False,
|
|
encrypt_assertion=True,
|
|
encrypt_assertion_self_contained=True,
|
|
)
|
|
|
|
sresponse = response_from_string(signed_resp)
|
|
|
|
valid = self.server.sec.verify_signature(signed_resp,
|
|
self.server.config.cert_file,
|
|
node_name='urn:oasis:names:tc:SAML:2.0:protocol:Response',
|
|
node_id=sresponse.id,
|
|
id_attr="")
|
|
assert valid
|
|
|
|
decr_text_old = copy.deepcopy("%s" % signed_resp)
|
|
|
|
decr_text = self.server.sec.decrypt(signed_resp, self.client.config.encryption_keypairs[0]["key_file"])
|
|
|
|
assert decr_text == decr_text_old
|
|
|
|
decr_text = self.server.sec.decrypt(signed_resp, self.client.config.encryption_keypairs[1]["key_file"])
|
|
|
|
assert decr_text != decr_text_old
|
|
|
|
resp = samlp.response_from_string(decr_text)
|
|
|
|
resp.assertion = extension_elements_to_elements(resp.encrypted_assertion[0].extension_elements, [saml, samlp])
|
|
|
|
assert resp.assertion[0].signature == None
|
|
|
|
self.verify_assertion(resp.assertion)
|
|
|
|
|
|
def test_encrypted_signed_response_3(self):
|
|
cert_str, cert_key_str = generate_cert()
|
|
|
|
signed_resp = self.server.create_authn_response(
|
|
self.ava,
|
|
"id12", # in_response_to
|
|
"http://lingon.catalogix.se:8087/", # consumer_url
|
|
"urn:mace:example.com:saml:roland:sp", # sp_entity_id
|
|
name_id=self.name_id,
|
|
sign_response=True,
|
|
sign_assertion=True,
|
|
encrypt_assertion=True,
|
|
encrypt_assertion_self_contained=False,
|
|
encrypt_cert_assertion=cert_str,
|
|
)
|
|
|
|
sresponse = response_from_string(signed_resp)
|
|
|
|
valid = self.server.sec.verify_signature(signed_resp,
|
|
self.server.config.cert_file,
|
|
node_name='urn:oasis:names:tc:SAML:2.0:protocol:Response',
|
|
node_id=sresponse.id,
|
|
id_attr="")
|
|
assert valid
|
|
|
|
_, key_file = make_temp(cert_key_str, decode=False)
|
|
|
|
decr_text = self.server.sec.decrypt(signed_resp, key_file)
|
|
|
|
resp = samlp.response_from_string(decr_text)
|
|
|
|
resp.assertion = extension_elements_to_elements(resp.encrypted_assertion[0].extension_elements, [saml, samlp])
|
|
|
|
valid = self.server.sec.verify_signature(decr_text,
|
|
self.server.config.cert_file,
|
|
node_name='urn:oasis:names:tc:SAML:2.0:assertion:Assertion',
|
|
node_id=resp.assertion[0].id,
|
|
id_attr="")
|
|
|
|
assert valid
|
|
|
|
self.verify_assertion(resp.assertion)
|
|
|
|
assert 'xmlns:encas' not in decr_text
|
|
|
|
|
|
def test_encrypted_signed_response_4(self):
|
|
|
|
cert_str, cert_key_str = generate_cert()
|
|
|
|
signed_resp = self.server.create_authn_response(
|
|
self.ava,
|
|
"id12", # in_response_to
|
|
"http://lingon.catalogix.se:8087/", # consumer_url
|
|
"urn:mace:example.com:saml:roland:sp", # sp_entity_id
|
|
name_id=self.name_id,
|
|
sign_response=True,
|
|
sign_assertion=True,
|
|
encrypt_assertion=True,
|
|
encrypt_assertion_self_contained=True,
|
|
pefim=True,
|
|
encrypt_cert_advice=cert_str,
|
|
)
|
|
|
|
sresponse = response_from_string(signed_resp)
|
|
|
|
valid = self.server.sec.verify_signature(signed_resp,
|
|
self.server.config.cert_file,
|
|
node_name='urn:oasis:names:tc:SAML:2.0:protocol:Response',
|
|
node_id=sresponse.id,
|
|
id_attr="")
|
|
assert valid
|
|
|
|
decr_text = self.server.sec.decrypt(signed_resp, self.client.config.encryption_keypairs[1]["key_file"])
|
|
|
|
resp = samlp.response_from_string(decr_text)
|
|
|
|
resp.assertion = extension_elements_to_elements(resp.encrypted_assertion[0].extension_elements, [saml, samlp])
|
|
|
|
valid = self.server.sec.verify_signature(decr_text,
|
|
self.server.config.cert_file,
|
|
node_name='urn:oasis:names:tc:SAML:2.0:assertion:Assertion',
|
|
node_id=resp.assertion[0].id,
|
|
id_attr="")
|
|
|
|
assert valid
|
|
|
|
_, key_file = make_temp(cert_key_str, decode=False)
|
|
|
|
decr_text = self.server.sec.decrypt(decr_text, key_file)
|
|
|
|
resp = samlp.response_from_string(decr_text)
|
|
|
|
assertion = extension_elements_to_elements(resp.encrypted_assertion[0].extension_elements, [saml, samlp])
|
|
assertion = \
|
|
extension_elements_to_elements(assertion[0].advice.encrypted_assertion[0].extension_elements,[saml, samlp])
|
|
self.verify_assertion(assertion)
|
|
|
|
#PEFIM never signs assertion in advice
|
|
assert assertion[0].signature is None
|
|
#valid = self.server.sec.verify_signature(decr_text,
|
|
# self.server.config.cert_file,
|
|
# node_name='urn:oasis:names:tc:SAML:2.0:assertion:Assertion',
|
|
# node_id=assertion[0].id,
|
|
# id_attr="")
|
|
assert valid
|
|
|
|
def test_encrypted_response_1(self):
|
|
cert_str_advice, cert_key_str_advice = generate_cert()
|
|
|
|
_resp = self.server.create_authn_response(
|
|
self.ava,
|
|
"id12", # in_response_to
|
|
"http://lingon.catalogix.se:8087/", # consumer_url
|
|
"urn:mace:example.com:saml:roland:sp", # sp_entity_id
|
|
name_id=self.name_id,
|
|
sign_response=False,
|
|
sign_assertion=False,
|
|
encrypt_assertion=False,
|
|
encrypt_assertion_self_contained=True,
|
|
pefim=True,
|
|
encrypt_cert_advice=cert_str_advice,
|
|
)
|
|
|
|
_resp = "%s" % _resp
|
|
|
|
sresponse = response_from_string(_resp)
|
|
|
|
assert sresponse.signature is None
|
|
|
|
_, key_file = make_temp(cert_key_str_advice, decode=False)
|
|
|
|
decr_text = self.server.sec.decrypt(_resp, key_file)
|
|
|
|
resp = samlp.response_from_string(decr_text)
|
|
|
|
self.verify_advice_assertion(resp, decr_text)
|
|
|
|
def test_encrypted_response_2(self):
|
|
|
|
cert_str_advice, cert_key_str_advice = generate_cert()
|
|
|
|
_resp = self.server.create_authn_response(
|
|
self.ava,
|
|
"id12", # in_response_to
|
|
"http://lingon.catalogix.se:8087/", # consumer_url
|
|
"urn:mace:example.com:saml:roland:sp", # sp_entity_id
|
|
name_id=self.name_id,
|
|
sign_response=False,
|
|
sign_assertion=False,
|
|
encrypt_assertion=True,
|
|
encrypt_assertion_self_contained=True,
|
|
pefim=True,
|
|
encrypt_cert_advice=cert_str_advice,
|
|
)
|
|
|
|
sresponse = response_from_string(_resp)
|
|
|
|
assert sresponse.signature is None
|
|
|
|
decr_text_1 = self.server.sec.decrypt(_resp, self.client.config.encryption_keypairs[1]["key_file"])
|
|
|
|
_, key_file = make_temp(cert_key_str_advice, decode=False)
|
|
|
|
decr_text_2 = self.server.sec.decrypt(decr_text_1, key_file)
|
|
|
|
resp = samlp.response_from_string(decr_text_2)
|
|
|
|
resp.assertion = extension_elements_to_elements(resp.encrypted_assertion[0].extension_elements, [saml, samlp])
|
|
|
|
self.verify_advice_assertion(resp, decr_text_2)
|
|
|
|
def test_encrypted_response_3(self):
|
|
cert_str_assertion, cert_key_str_assertion = generate_cert()
|
|
|
|
_resp = self.server.create_authn_response(
|
|
self.ava,
|
|
"id12", # in_response_to
|
|
"http://lingon.catalogix.se:8087/", # consumer_url
|
|
"urn:mace:example.com:saml:roland:sp", # sp_entity_id
|
|
name_id=self.name_id,
|
|
sign_response=False,
|
|
sign_assertion=False,
|
|
encrypt_assertion=True,
|
|
encrypt_assertion_self_contained=True,
|
|
encrypted_advice_attributes=False,
|
|
encrypt_cert_assertion=cert_str_assertion
|
|
)
|
|
|
|
sresponse = response_from_string(_resp)
|
|
|
|
assert sresponse.signature is None
|
|
|
|
_, key_file = make_temp(cert_key_str_assertion, decode=False)
|
|
|
|
decr_text = self.server.sec.decrypt(_resp, key_file)
|
|
|
|
resp = samlp.response_from_string(decr_text)
|
|
|
|
assert resp.encrypted_assertion[0].extension_elements
|
|
|
|
assertion = extension_elements_to_elements(resp.encrypted_assertion[0].extension_elements, [saml, samlp])
|
|
|
|
self.verify_encrypted_assertion(assertion, decr_text)
|
|
|
|
def test_encrypted_response_4(self):
|
|
_resp = self.server.create_authn_response(
|
|
self.ava,
|
|
"id12", # in_response_to
|
|
"http://lingon.catalogix.se:8087/", # consumer_url
|
|
"urn:mace:example.com:saml:roland:sp", # sp_entity_id
|
|
name_id=self.name_id,
|
|
sign_response=False,
|
|
sign_assertion=False,
|
|
encrypt_assertion=True,
|
|
encrypt_assertion_self_contained=True,
|
|
encrypted_advice_attributes=False,
|
|
)
|
|
|
|
sresponse = response_from_string(_resp)
|
|
|
|
assert sresponse.signature is None
|
|
|
|
decr_text = self.server.sec.decrypt(_resp, self.client.config.encryption_keypairs[1]["key_file"])
|
|
|
|
resp = samlp.response_from_string(decr_text)
|
|
|
|
assert resp.encrypted_assertion[0].extension_elements
|
|
|
|
assertion = extension_elements_to_elements(resp.encrypted_assertion[0].extension_elements, [saml, samlp])
|
|
|
|
self.verify_encrypted_assertion(assertion, decr_text)
|
|
|
|
def test_encrypted_response_5(self):
|
|
_resp = self.server.create_authn_response(
|
|
self.ava,
|
|
"id12", # in_response_to
|
|
"http://lingon.catalogix.se:8087/", # consumer_url
|
|
"urn:mace:example.com:saml:roland:sp", # sp_entity_id
|
|
name_id=self.name_id,
|
|
sign_response=False,
|
|
sign_assertion=False,
|
|
encrypt_assertion=False,
|
|
encrypt_assertion_self_contained=True,
|
|
pefim=True
|
|
)
|
|
|
|
_resp = "%s" % _resp
|
|
|
|
sresponse = response_from_string(_resp)
|
|
|
|
assert sresponse.signature is None
|
|
|
|
decr_text = self.server.sec.decrypt(_resp, self.client.config.encryption_keypairs[1]["key_file"])
|
|
|
|
resp = samlp.response_from_string(decr_text)
|
|
|
|
self.verify_advice_assertion(resp, decr_text)
|
|
|
|
def test_encrypted_response_6(self):
|
|
_server = Server("idp_conf_verify_cert")
|
|
|
|
cert_str_advice, cert_key_str_advice = generate_cert()
|
|
|
|
cert_str_assertion, cert_key_str_assertion = generate_cert()
|
|
|
|
_resp = _server.create_authn_response(
|
|
self.ava,
|
|
"id12", # in_response_to
|
|
"http://lingon.catalogix.se:8087/", # consumer_url
|
|
"urn:mace:example.com:saml:roland:sp", # sp_entity_id
|
|
name_id=self.name_id,
|
|
sign_response=False,
|
|
sign_assertion=False,
|
|
encrypt_assertion=True,
|
|
encrypt_assertion_self_contained=True,
|
|
pefim=True,
|
|
encrypt_cert_advice=cert_str_advice,
|
|
encrypt_cert_assertion=cert_str_assertion
|
|
)
|
|
|
|
sresponse = response_from_string(_resp)
|
|
|
|
assert sresponse.signature is None
|
|
|
|
_, key_file = make_temp(cert_key_str_assertion, decode=False)
|
|
|
|
decr_text_1 = _server.sec.decrypt(_resp, key_file)
|
|
|
|
_, key_file = make_temp(cert_key_str_advice, decode=False)
|
|
|
|
decr_text_2 = _server.sec.decrypt(decr_text_1, key_file)
|
|
|
|
resp = samlp.response_from_string(decr_text_2)
|
|
|
|
resp.assertion = extension_elements_to_elements(resp.encrypted_assertion[0].extension_elements, [saml, samlp])
|
|
|
|
self.verify_advice_assertion(resp, decr_text_2)
|
|
|
|
def test_encrypted_response_7(self):
|
|
_resp = self.server.create_authn_response(
|
|
self.ava,
|
|
"id12", # in_response_to
|
|
"http://lingon.catalogix.se:8087/", # consumer_url
|
|
"urn:mace:example.com:saml:roland:sp", # sp_entity_id
|
|
name_id=self.name_id,
|
|
sign_response=False,
|
|
sign_assertion=False,
|
|
encrypt_assertion=True,
|
|
encrypt_assertion_self_contained=True,
|
|
pefim=True
|
|
)
|
|
|
|
sresponse = response_from_string(_resp)
|
|
|
|
assert sresponse.signature is None
|
|
|
|
decr_text_1 = self.server.sec.decrypt(_resp, self.client.config.encryption_keypairs[1]["key_file"])
|
|
|
|
decr_text_2 = self.server.sec.decrypt(decr_text_1, self.client.config.encryption_keypairs[1]["key_file"])
|
|
|
|
resp = samlp.response_from_string(decr_text_2)
|
|
|
|
resp.assertion = extension_elements_to_elements(resp.encrypted_assertion[0].extension_elements, [saml, samlp])
|
|
|
|
self.verify_advice_assertion(resp, decr_text_2)
|
|
|
|
def test_encrypted_response_8(self):
|
|
try:
|
|
_resp = self.server.create_authn_response(
|
|
self.ava,
|
|
"id12", # in_response_to
|
|
"http://lingon.catalogix.se:8087/", # consumer_url
|
|
"urn:mace:example.com:saml:roland:sp", # sp_entity_id
|
|
name_id=self.name_id,
|
|
sign_response=False,
|
|
sign_assertion=False,
|
|
encrypt_assertion=True,
|
|
encrypt_assertion_self_contained=True,
|
|
pefim=True,
|
|
encrypt_cert_advice="whatever",
|
|
encrypt_cert_assertion="whatever"
|
|
)
|
|
assert False, "Must throw an exception"
|
|
except EncryptError as ex:
|
|
pass
|
|
except Exception as ex:
|
|
assert False, "Wrong exception!"
|
|
|
|
try:
|
|
_resp = self.server.create_authn_response(
|
|
self.ava,
|
|
"id12", # in_response_to
|
|
"http://lingon.catalogix.se:8087/", # consumer_url
|
|
"urn:mace:example.com:saml:roland:sp", # sp_entity_id
|
|
name_id=self.name_id,
|
|
sign_response=False,
|
|
sign_assertion=False,
|
|
encrypt_assertion=False,
|
|
encrypt_assertion_self_contained=True,
|
|
pefim=True,
|
|
encrypt_cert_advice="whatever",
|
|
)
|
|
assert False, "Must throw an exception"
|
|
except EncryptError as ex:
|
|
pass
|
|
except Exception as ex:
|
|
assert False, "Wrong exception!"
|
|
|
|
try:
|
|
_resp = self.server.create_authn_response(
|
|
self.ava,
|
|
"id12", # in_response_to
|
|
"http://lingon.catalogix.se:8087/", # consumer_url
|
|
"urn:mace:example.com:saml:roland:sp", # sp_entity_id
|
|
name_id=self.name_id,
|
|
sign_response=False,
|
|
sign_assertion=False,
|
|
encrypt_assertion=True,
|
|
encrypt_assertion_self_contained=True,
|
|
encrypted_advice_attributes=False,
|
|
encrypt_cert_assertion="whatever"
|
|
)
|
|
assert False, "Must throw an exception"
|
|
except EncryptError as ex:
|
|
pass
|
|
except Exception as ex:
|
|
assert False, "Wrong exception!"
|
|
|
|
_server = Server("idp_conf_verify_cert")
|
|
|
|
try:
|
|
_resp = _server.create_authn_response(
|
|
self.ava,
|
|
"id12", # in_response_to
|
|
"http://lingon.catalogix.se:8087/", # consumer_url
|
|
"urn:mace:example.com:saml:roland:sp", # sp_entity_id
|
|
name_id=self.name_id,
|
|
sign_response=False,
|
|
sign_assertion=False,
|
|
encrypt_assertion=True,
|
|
encrypt_assertion_self_contained=True,
|
|
pefim=True,
|
|
encrypt_cert_advice="whatever",
|
|
encrypt_cert_assertion="whatever"
|
|
)
|
|
assert False, "Must throw an exception"
|
|
except CertificateError as ex:
|
|
pass
|
|
except Exception as ex:
|
|
assert False, "Wrong exception!"
|
|
|
|
try:
|
|
_resp = _server.create_authn_response(
|
|
self.ava,
|
|
"id12", # in_response_to
|
|
"http://lingon.catalogix.se:8087/", # consumer_url
|
|
"urn:mace:example.com:saml:roland:sp", # sp_entity_id
|
|
name_id=self.name_id,
|
|
sign_response=False,
|
|
sign_assertion=False,
|
|
encrypt_assertion=False,
|
|
encrypt_assertion_self_contained=True,
|
|
pefim=True,
|
|
encrypt_cert_advice="whatever",
|
|
)
|
|
assert False, "Must throw an exception"
|
|
except CertificateError as ex:
|
|
pass
|
|
except Exception as ex:
|
|
assert False, "Wrong exception!"
|
|
|
|
try:
|
|
_resp = _server.create_authn_response(
|
|
self.ava,
|
|
"id12", # in_response_to
|
|
"http://lingon.catalogix.se:8087/", # consumer_url
|
|
"urn:mace:example.com:saml:roland:sp", # sp_entity_id
|
|
name_id=self.name_id,
|
|
sign_response=False,
|
|
sign_assertion=False,
|
|
encrypt_assertion=True,
|
|
encrypt_assertion_self_contained=True,
|
|
encrypted_advice_attributes=False,
|
|
encrypt_cert_assertion="whatever"
|
|
)
|
|
assert False, "Must throw an exception"
|
|
except CertificateError as ex:
|
|
pass
|
|
except Exception as ex:
|
|
assert False, "Wrong exception!"
|
|
|
|
def test_encrypted_response_9(self):
|
|
_server = Server("idp_conf_sp_no_encrypt")
|
|
|
|
_resp = _server.create_authn_response(
|
|
self.ava,
|
|
"id12", # in_response_to
|
|
"http://lingon.catalogix.se:8087/", # consumer_url
|
|
"urn:mace:example.com:saml:roland:sp", # sp_entity_id
|
|
name_id=self.name_id,
|
|
sign_response=False,
|
|
sign_assertion=False,
|
|
encrypt_assertion=True,
|
|
encrypt_assertion_self_contained=True,
|
|
pefim=True,
|
|
)
|
|
|
|
self.verify_assertion(_resp.assertion.advice.assertion)
|
|
|
|
_resp = _server.create_authn_response(
|
|
self.ava,
|
|
"id12", # in_response_to
|
|
"http://lingon.catalogix.se:8087/", # consumer_url
|
|
"urn:mace:example.com:saml:roland:sp", # sp_entity_id
|
|
name_id=self.name_id,
|
|
sign_response=False,
|
|
sign_assertion=False,
|
|
encrypt_assertion=False,
|
|
encrypt_assertion_self_contained=True,
|
|
pefim=True
|
|
)
|
|
|
|
self.verify_assertion(_resp.assertion.advice.assertion)
|
|
|
|
_resp = _server.create_authn_response(
|
|
self.ava,
|
|
"id12", # in_response_to
|
|
"http://lingon.catalogix.se:8087/", # consumer_url
|
|
"urn:mace:example.com:saml:roland:sp", # sp_entity_id
|
|
name_id=self.name_id,
|
|
sign_response=False,
|
|
sign_assertion=False,
|
|
encrypt_assertion=True,
|
|
encrypt_assertion_self_contained=True,
|
|
encrypted_advice_attributes=False,
|
|
)
|
|
|
|
self.verify_assertion([_resp.assertion])
|
|
|
|
|
|
def test_slo_http_post(self):
|
|
soon = time_util.in_a_while(days=1)
|
|
sinfo = {
|
|
"name_id": nid,
|
|
"issuer": "urn:mace:example.com:saml:roland:idp",
|
|
"not_on_or_after": soon,
|
|
"user": {
|
|
"givenName": "Leo",
|
|
"sn": "Laport",
|
|
}
|
|
}
|
|
self.client.users.add_information_about_person(sinfo)
|
|
|
|
req_id, logout_request = self.client.create_logout_request(
|
|
destination="http://localhost:8088/slop", name_id=nid,
|
|
issuer_entity_id="urn:mace:example.com:saml:roland:idp",
|
|
reason="I'm tired of this")
|
|
|
|
intermed = base64.b64encode(str(logout_request).encode('utf-8'))
|
|
|
|
#saml_soap = make_soap_enveloped_saml_thingy(logout_request)
|
|
request = self.server.parse_logout_request(intermed, BINDING_HTTP_POST)
|
|
assert request
|
|
|
|
def test_slo_soap(self):
|
|
soon = time_util.in_a_while(days=1)
|
|
sinfo = {
|
|
"name_id": nid,
|
|
"issuer": "urn:mace:example.com:saml:roland:idp",
|
|
"not_on_or_after": soon,
|
|
"user": {
|
|
"givenName": "Leo",
|
|
"sn": "Laport",
|
|
}
|
|
}
|
|
|
|
sp = client.Saml2Client(config_file="server_conf")
|
|
sp.users.add_information_about_person(sinfo)
|
|
|
|
req_id, logout_request = sp.create_logout_request(
|
|
name_id=nid, destination="http://localhost:8088/slo",
|
|
issuer_entity_id="urn:mace:example.com:saml:roland:idp",
|
|
reason="I'm tired of this")
|
|
|
|
#_ = s_utils.deflate_and_base64_encode("%s" % (logout_request,))
|
|
|
|
saml_soap = make_soap_enveloped_saml_thingy(logout_request)
|
|
self.server.ident.close()
|
|
|
|
with closing(Server("idp_soap_conf")) as idp:
|
|
request = idp.parse_logout_request(saml_soap)
|
|
idp.ident.close()
|
|
assert request
|
|
|
|
#------------------------------------------------------------------------
|
|
|
|
IDENTITY = {"eduPersonAffiliation": ["staff", "member"],
|
|
"sn": ["Jeter"], "givenName": ["Derek"],
|
|
"mail": ["foo@gmail.com"], "title": "The man"}
|
|
|
|
|
|
class TestServer2():
|
|
def setup_class(self):
|
|
self.server = Server("restrictive_idp_conf")
|
|
|
|
def teardown_class(self):
|
|
self.server.close()
|
|
|
|
def test_do_attribute_reponse(self):
|
|
aa_policy = self.server.config.getattr("policy", "idp")
|
|
print(aa_policy.__dict__)
|
|
response = self.server.create_attribute_response(
|
|
IDENTITY.copy(), "aaa", "http://example.com/sp/",
|
|
"http://www.example.com/roland/sp")
|
|
|
|
assert response is not None
|
|
assert response.destination == "http://example.com/sp/"
|
|
assert response.in_response_to == "aaa"
|
|
assert response.version == "2.0"
|
|
assert response.issuer.text == "urn:mace:example.com:saml:roland:idpr"
|
|
assert response.status.status_code.value == samlp.STATUS_SUCCESS
|
|
assert response.assertion
|
|
assertion = response.assertion
|
|
assert assertion.version == "2.0"
|
|
subject = assertion.subject
|
|
#assert subject.name_id.format == saml.NAMEID_FORMAT_TRANSIENT
|
|
assert subject.subject_confirmation
|
|
subject_conf = subject.subject_confirmation[0]
|
|
assert subject_conf.subject_confirmation_data.in_response_to == "aaa"
|
|
|
|
|
|
def _logout_request(conf_file):
|
|
conf = config.SPConfig()
|
|
conf.load_file(conf_file)
|
|
sp = client.Saml2Client(conf)
|
|
|
|
soon = time_util.in_a_while(days=1)
|
|
sinfo = {
|
|
"name_id": nid,
|
|
"issuer": "urn:mace:example.com:saml:roland:idp",
|
|
"not_on_or_after": soon,
|
|
"user": {
|
|
"givenName": "Leo",
|
|
"sn": "Laport",
|
|
}
|
|
}
|
|
sp.users.add_information_about_person(sinfo)
|
|
|
|
return sp.create_logout_request(
|
|
name_id=nid,
|
|
destination="http://localhost:8088/slo",
|
|
issuer_entity_id="urn:mace:example.com:saml:roland:idp",
|
|
reason="I'm tired of this")
|
|
|
|
|
|
class TestServerLogout():
|
|
def test_1(self):
|
|
with closing(Server("idp_slo_redirect_conf")) as server:
|
|
req_id, request = _logout_request("sp_slo_redirect_conf")
|
|
print(request)
|
|
bindings = [BINDING_HTTP_REDIRECT]
|
|
response = server.create_logout_response(request, bindings)
|
|
binding, destination = server.pick_binding("single_logout_service",
|
|
bindings, "spsso",
|
|
request)
|
|
|
|
http_args = server.apply_binding(binding, "%s" % response, destination,
|
|
"relay_state", response=True)
|
|
|
|
assert len(http_args) == 4
|
|
assert http_args["headers"][0][0] == "Location"
|
|
assert http_args["data"] == []
|
|
|
|
|
|
if __name__ == "__main__":
|
|
ts = TestServer1()
|
|
ts.setup_class()
|
|
ts.test_encrypted_signed_response_1()
|