openstack/deb-python-pysaml2
Code Issues Proposed changes

Added a PEFIM parameter for PEFIM specific configurations.

Browse Source
This commit is contained in:
Hans Hörberg
2015-05-20 08:43:29 +02:00
parent ea021d6f44
commit eb2d968476
4 changed files with 68 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
src/saml2/entity.py
View File

@@ -541,7 +541,7 @@ class Entity(HTTPBase):
def _response(self, in_response_to, consumer_url=None, status=None, def _response(self, in_response_to, consumer_url=None, status=None,
issuer=None, sign=False, to_sign=None, sp_entity_id=None, issuer=None, sign=False, to_sign=None, sp_entity_id=None,
encrypt_assertion=False, encrypt_assertion_self_contained=False, encrypted_advice_attributes=False, encrypt_assertion=False, encrypt_assertion_self_contained=False, encrypted_advice_attributes=False,
encrypt_cert_advice=None, encrypt_cert_assertion=None,sign_assertion=None, **kwargs): encrypt_cert_advice=None, encrypt_cert_assertion=None,sign_assertion=None, pefim=False, **kwargs):
""" Create a Response. """ Create a Response.
Encryption: Encryption:
encrypt_assertion must be true for encryption to be performed. If encrypted_advice_attributes also is encrypt_assertion must be true for encryption to be performed. If encrypted_advice_attributes also is
@@ -584,7 +584,8 @@ class Entity(HTTPBase):
if not has_encrypt_cert and encrypt_cert_assertion is None: if not has_encrypt_cert and encrypt_cert_assertion is None:
encrypt_assertion = False encrypt_assertion = False
if encrypt_assertion or (encrypted_advice_attributes and response.assertion.advice is not None and len(response.assertion.advice.assertion) == 1): if encrypt_assertion or (encrypted_advice_attributes and response.assertion.advice is not None and
len(response.assertion.advice.assertion) == 1):
if sign: if sign:
response.signature = pre_signature_part(response.id, response.signature = pre_signature_part(response.id,
self.sec.my_cert, 1) self.sec.my_cert, 1)
@@ -605,7 +606,7 @@ class Entity(HTTPBase):
_advice_assertions = [_advice_assertions] _advice_assertions = [_advice_assertions]
for tmp_assertion in _advice_assertions: for tmp_assertion in _advice_assertions:
to_sign_advice = [] to_sign_advice = []
if sign_assertion is not None and sign_assertion: if sign_assertion and not pefim:
tmp_assertion.signature = pre_signature_part(tmp_assertion.id, self.sec.my_cert, 1) tmp_assertion.signature = pre_signature_part(tmp_assertion.id, self.sec.my_cert, 1)
to_sign_advice.append((class_name(tmp_assertion), tmp_assertion.id)) to_sign_advice.append((class_name(tmp_assertion), tmp_assertion.id))
#tmp_assertion = response.assertion.advice.assertion[0] #tmp_assertion = response.assertion.advice.assertion[0]

8
src/saml2/response.py
View File

@@ -797,7 +797,7 @@ class AuthnResponse(StatusResponse):
logger.exception("get subject") logger.exception("get subject")
raise raise
def decrypt_assertions(self, encrypted_assertions, decr_txt): def decrypt_assertions(self, encrypted_assertions, decr_txt, issuer=None):
res = [] res = []
for encrypted_assertion in encrypted_assertions: for encrypted_assertion in encrypted_assertions:
if encrypted_assertion.extension_elements: if encrypted_assertion.extension_elements:
@@ -807,7 +807,7 @@ class AuthnResponse(StatusResponse):
if assertion.signature: if assertion.signature:
if not self.sec.check_signature( if not self.sec.check_signature(
assertion, origdoc=decr_txt, assertion, origdoc=decr_txt,
node_name=class_name(assertion)): node_name=class_name(assertion), issuer=issuer):
logger.error( logger.error(
"Failed to verify signature on '%s'" % assertion) "Failed to verify signature on '%s'" % assertion)
raise SignatureError() raise SignatureError()
@@ -842,7 +842,9 @@ class AuthnResponse(StatusResponse):
if resp.assertion: if resp.assertion:
for tmp_ass in resp.assertion: for tmp_ass in resp.assertion:
if tmp_ass.advice and tmp_ass.advice.encrypted_assertion: if tmp_ass.advice and tmp_ass.advice.encrypted_assertion:
advice_res = self.decrypt_assertions(tmp_ass.advice.encrypted_assertion, decr_text) advice_res = self.decrypt_assertions(tmp_ass.advice.encrypted_assertion,
decr_text,
tmp_ass.issuer)
if tmp_ass.advice.assertion: if tmp_ass.advice.assertion:
tmp_ass.advice.assertion.extend(advice_res) tmp_ass.advice.assertion.extend(advice_res)
else: else:

27
src/saml2/server.py
View File

@@ -324,7 +324,7 @@ class Server(Entity):
sign_assertion=False, sign_response=False, sign_assertion=False, sign_response=False,
best_effort=False, encrypt_assertion=False, best_effort=False, encrypt_assertion=False,
encrypt_cert_advice=None, encrypt_cert_assertion=None, authn_statement=None, encrypt_cert_advice=None, encrypt_cert_assertion=None, authn_statement=None,
encrypt_assertion_self_contained=False, encrypted_advice_attributes=False): encrypt_assertion_self_contained=False, encrypted_advice_attributes=False, pefim=False):
""" Create a response. A layer of indirection. """ Create a response. A layer of indirection.
:param in_response_to: The session identifier of the request :param in_response_to: The session identifier of the request
@@ -358,12 +358,14 @@ class Server(Entity):
# tmp_authn_statement = authn_statement # tmp_authn_statement = authn_statement
# authn_statement = None # authn_statement = None
if encrypted_advice_attributes: if pefim:
encrypted_advice_attributes = True
encrypt_assertion_self_contained = True
assertion_attributes = self.setup_assertion(None, sp_entity_id, None, None, None, policy, assertion_attributes = self.setup_assertion(None, sp_entity_id, None, None, None, policy,
None, None, identity, best_effort, sign_response, False) None, None, identity, best_effort, sign_response, False)
assertion = self.setup_assertion(authn, sp_entity_id, in_response_to, consumer_url, assertion = self.setup_assertion(authn, sp_entity_id, in_response_to, consumer_url,
name_id, policy, _issuer, authn_statement, [], True, name_id, policy, _issuer, authn_statement, [], True,
sign_response) sign_response)
assertion.advice = saml.Advice() assertion.advice = saml.Advice()
#assertion.advice.assertion_id_ref.append(saml.AssertionIDRef()) #assertion.advice.assertion_id_ref.append(saml.AssertionIDRef())
@@ -371,8 +373,8 @@ class Server(Entity):
assertion.advice.assertion.append(assertion_attributes) assertion.advice.assertion.append(assertion_attributes)
else: else:
assertion = self.setup_assertion(authn, sp_entity_id, in_response_to, consumer_url, assertion = self.setup_assertion(authn, sp_entity_id, in_response_to, consumer_url,
name_id, policy, _issuer, authn_statement, identity, True, name_id, policy, _issuer, authn_statement, identity, True,
sign_response) sign_response)
to_sign = [] to_sign = []
if not encrypt_assertion: if not encrypt_assertion:
@@ -405,6 +407,7 @@ class Server(Entity):
encrypt_cert_assertion=encrypt_cert_assertion, encrypt_cert_assertion=encrypt_cert_assertion,
encrypt_assertion_self_contained=encrypt_assertion_self_contained, encrypt_assertion_self_contained=encrypt_assertion_self_contained,
encrypted_advice_attributes=encrypted_advice_attributes,sign_assertion=sign_assertion, encrypted_advice_attributes=encrypted_advice_attributes,sign_assertion=sign_assertion,
pefim=pefim,
**args) **args)
# ------------------------------------------------------------------------ # ------------------------------------------------------------------------
@@ -481,7 +484,7 @@ class Server(Entity):
sign_response=None, sign_assertion=None, sign_response=None, sign_assertion=None,
encrypt_cert_advice=None, encrypt_cert_assertion=None, encrypt_assertion=None, encrypt_cert_advice=None, encrypt_cert_assertion=None, encrypt_assertion=None,
encrypt_assertion_self_contained=True, encrypt_assertion_self_contained=True,
encrypted_advice_attributes=False, encrypted_advice_attributes=False, pefim=False,
**kwargs): **kwargs):
""" Constructs an AuthenticationResponse """ Constructs an AuthenticationResponse
@@ -536,7 +539,7 @@ class Server(Entity):
if encrypted_advice_attributes is None: if encrypted_advice_attributes is None:
encrypted_advice_attributes = False encrypted_advice_attributes = False
if encrypted_advice_attributes: if encrypted_advice_attributes or pefim:
verify_encrypt_cert = self.config.getattr("verify_encrypt_cert_advice", "idp") verify_encrypt_cert = self.config.getattr("verify_encrypt_cert_advice", "idp")
if verify_encrypt_cert is not None: if verify_encrypt_cert is not None:
if encrypt_cert_advice is None: if encrypt_cert_advice is None:
@@ -613,7 +616,8 @@ class Server(Entity):
encrypt_assertion_self_contained=encrypt_assertion_self_contained, encrypt_assertion_self_contained=encrypt_assertion_self_contained,
encrypted_advice_attributes=encrypted_advice_attributes, encrypted_advice_attributes=encrypted_advice_attributes,
encrypt_cert_advice=encrypt_cert_advice, encrypt_cert_advice=encrypt_cert_advice,
encrypt_cert_assertion=encrypt_cert_assertion) encrypt_cert_assertion=encrypt_cert_assertion,
pefim=peifm)
return self._authn_response(in_response_to, # in_response_to return self._authn_response(in_response_to, # in_response_to
destination, # consumer_url destination, # consumer_url
sp_entity_id, # sp_entity_id sp_entity_id, # sp_entity_id
@@ -629,7 +633,8 @@ class Server(Entity):
encrypt_assertion_self_contained=encrypt_assertion_self_contained, encrypt_assertion_self_contained=encrypt_assertion_self_contained,
encrypted_advice_attributes=encrypted_advice_attributes, encrypted_advice_attributes=encrypted_advice_attributes,
encrypt_cert_advice=encrypt_cert_advice, encrypt_cert_advice=encrypt_cert_advice,
encrypt_cert_assertion=encrypt_cert_assertion) encrypt_cert_assertion=encrypt_cert_assertion,
pefim=pefim)
except MissingValue as exc: except MissingValue as exc:
return self.create_error_response(in_response_to, destination, return self.create_error_response(in_response_to, destination,

69
tests/test_50_server.py
View File

@@ -539,7 +539,8 @@ class TestServer1():
sign_assertion=True, sign_assertion=True,
encrypt_assertion=False, encrypt_assertion=False,
encrypt_assertion_self_contained=True, encrypt_assertion_self_contained=True,
encrypted_advice_attributes=True, pefim=True,
#encrypted_advice_attributes=True,
encrypt_cert_advice=cert_str, encrypt_cert_advice=cert_str,
) )
@@ -573,11 +574,15 @@ class TestServer1():
self.verify_assertion(assertion) self.verify_assertion(assertion)
valid = self.server.sec.verify_signature(decr_text,
self.server.config.cert_file,
node_name='urn:oasis:names:tc:SAML:2.0:assertion:Assertion', #PEFIM never signs assertions.
node_id=assertion[0].id, assert assertion[0].signature is None
id_attr="") #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 assert valid
def test_encrypted_signed_response_2(self): def test_encrypted_signed_response_2(self):
@@ -593,7 +598,6 @@ class TestServer1():
sign_assertion=False, sign_assertion=False,
encrypt_assertion=True, encrypt_assertion=True,
encrypt_assertion_self_contained=True, encrypt_assertion_self_contained=True,
encrypt_cert=cert_str,
) )
sresponse = response_from_string(signed_resp) sresponse = response_from_string(signed_resp)
@@ -649,7 +653,6 @@ class TestServer1():
resp.assertion = extension_elements_to_elements(resp.encrypted_assertion[0].extension_elements, [saml, samlp]) resp.assertion = extension_elements_to_elements(resp.encrypted_assertion[0].extension_elements, [saml, samlp])
valid = self.server.sec.verify_signature(decr_text, valid = self.server.sec.verify_signature(decr_text,
self.server.config.cert_file, self.server.config.cert_file,
node_name='urn:oasis:names:tc:SAML:2.0:assertion:Assertion', node_name='urn:oasis:names:tc:SAML:2.0:assertion:Assertion',
@@ -677,7 +680,8 @@ class TestServer1():
sign_assertion=True, sign_assertion=True,
encrypt_assertion=True, encrypt_assertion=True,
encrypt_assertion_self_contained=True, encrypt_assertion_self_contained=True,
encrypted_advice_attributes=True, #encrypted_advice_attributes=True,
pefim=True,
encrypt_cert_advice=cert_str, encrypt_cert_advice=cert_str,
) )
@@ -715,11 +719,13 @@ class TestServer1():
extension_elements_to_elements(assertion[0].advice.encrypted_assertion[0].extension_elements,[saml, samlp]) extension_elements_to_elements(assertion[0].advice.encrypted_assertion[0].extension_elements,[saml, samlp])
self.verify_assertion(assertion) self.verify_assertion(assertion)
valid = self.server.sec.verify_signature(decr_text, #PEFIM never signs assertion in advice
self.server.config.cert_file, assert assertion[0].signature is None
node_name='urn:oasis:names:tc:SAML:2.0:assertion:Assertion', #valid = self.server.sec.verify_signature(decr_text,
node_id=assertion[0].id, # self.server.config.cert_file,
id_attr="") # node_name='urn:oasis:names:tc:SAML:2.0:assertion:Assertion',
# node_id=assertion[0].id,
# id_attr="")
assert valid assert valid
def test_encrypted_response_1(self): def test_encrypted_response_1(self):
@@ -735,7 +741,8 @@ class TestServer1():
sign_assertion=False, sign_assertion=False,
encrypt_assertion=False, encrypt_assertion=False,
encrypt_assertion_self_contained=True, encrypt_assertion_self_contained=True,
encrypted_advice_attributes=True, #encrypted_advice_attributes=True,
pefim=True,
encrypt_cert_advice=cert_str_advice, encrypt_cert_advice=cert_str_advice,
) )
@@ -767,7 +774,8 @@ class TestServer1():
sign_assertion=False, sign_assertion=False,
encrypt_assertion=True, encrypt_assertion=True,
encrypt_assertion_self_contained=True, encrypt_assertion_self_contained=True,
encrypted_advice_attributes=True, #encrypted_advice_attributes=True,
pefim=True,
encrypt_cert_advice=cert_str_advice, encrypt_cert_advice=cert_str_advice,
) )
@@ -859,7 +867,8 @@ class TestServer1():
sign_assertion=False, sign_assertion=False,
encrypt_assertion=False, encrypt_assertion=False,
encrypt_assertion_self_contained=True, encrypt_assertion_self_contained=True,
encrypted_advice_attributes=True, #encrypted_advice_attributes=True,
pefim=True
) )
_resp = "%s" % _resp _resp = "%s" % _resp
@@ -891,7 +900,8 @@ class TestServer1():
sign_assertion=False, sign_assertion=False,
encrypt_assertion=True, encrypt_assertion=True,
encrypt_assertion_self_contained=True, encrypt_assertion_self_contained=True,
encrypted_advice_attributes=True, #encrypted_advice_attributes=True,
pefim=True,
encrypt_cert_advice=cert_str_advice, encrypt_cert_advice=cert_str_advice,
encrypt_cert_assertion=cert_str_assertion encrypt_cert_assertion=cert_str_assertion
) )
@@ -925,7 +935,8 @@ class TestServer1():
sign_assertion=False, sign_assertion=False,
encrypt_assertion=True, encrypt_assertion=True,
encrypt_assertion_self_contained=True, encrypt_assertion_self_contained=True,
encrypted_advice_attributes=True, #encrypted_advice_attributes=True,
pefim=True
) )
sresponse = response_from_string(_resp) sresponse = response_from_string(_resp)
@@ -954,7 +965,8 @@ class TestServer1():
sign_assertion=False, sign_assertion=False,
encrypt_assertion=True, encrypt_assertion=True,
encrypt_assertion_self_contained=True, encrypt_assertion_self_contained=True,
encrypted_advice_attributes=True, #encrypted_advice_attributes=True,
pefim=True,
encrypt_cert_advice="whatever", encrypt_cert_advice="whatever",
encrypt_cert_assertion="whatever" encrypt_cert_assertion="whatever"
) )
@@ -975,7 +987,8 @@ class TestServer1():
sign_assertion=False, sign_assertion=False,
encrypt_assertion=False, encrypt_assertion=False,
encrypt_assertion_self_contained=True, encrypt_assertion_self_contained=True,
encrypted_advice_attributes=True, #encrypted_advice_attributes=True,
pefim=True,
encrypt_cert_advice="whatever", encrypt_cert_advice="whatever",
) )
assert False, "Must throw an exception" assert False, "Must throw an exception"
@@ -1017,7 +1030,8 @@ class TestServer1():
sign_assertion=False, sign_assertion=False,
encrypt_assertion=True, encrypt_assertion=True,
encrypt_assertion_self_contained=True, encrypt_assertion_self_contained=True,
encrypted_advice_attributes=True, #encrypted_advice_attributes=True,
pefim=True,
encrypt_cert_advice="whatever", encrypt_cert_advice="whatever",
encrypt_cert_assertion="whatever" encrypt_cert_assertion="whatever"
) )
@@ -1038,7 +1052,8 @@ class TestServer1():
sign_assertion=False, sign_assertion=False,
encrypt_assertion=False, encrypt_assertion=False,
encrypt_assertion_self_contained=True, encrypt_assertion_self_contained=True,
encrypted_advice_attributes=True, #encrypted_advice_attributes=True,
pefim=True,
encrypt_cert_advice="whatever", encrypt_cert_advice="whatever",
) )
assert False, "Must throw an exception" assert False, "Must throw an exception"
@@ -1080,7 +1095,8 @@ class TestServer1():
sign_assertion=False, sign_assertion=False,
encrypt_assertion=True, encrypt_assertion=True,
encrypt_assertion_self_contained=True, encrypt_assertion_self_contained=True,
encrypted_advice_attributes=True, #encrypted_advice_attributes=True,
pefim=True,
) )
self.verify_assertion(_resp.assertion.advice.assertion) self.verify_assertion(_resp.assertion.advice.assertion)
@@ -1095,7 +1111,8 @@ class TestServer1():
sign_assertion=False, sign_assertion=False,
encrypt_assertion=False, encrypt_assertion=False,
encrypt_assertion_self_contained=True, encrypt_assertion_self_contained=True,
encrypted_advice_attributes=True, #encrypted_advice_attributes=True,
pefim=True
) )
self.verify_assertion(_resp.assertion.advice.assertion) self.verify_assertion(_resp.assertion.advice.assertion)
@@ -1253,4 +1270,4 @@ class TestServerLogout():
if __name__ == "__main__": if __name__ == "__main__":
ts = TestServer1() ts = TestServer1()
ts.setup_class() ts.setup_class()
ts.test_encrypted_response_9() ts.test_encrypted_signed_response_1()