openstack/deb-python-pysaml2
Code Issues Proposed changes

Fixes for EncryptedAssertion and signing.

Browse Source
This commit is contained in:
Hans Hörberg
2014-03-19 15:16:07 +01:00
parent 495d5f683b
commit e43f63cae3
9 changed files with 47 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
example/idp2/idp.py
View File

@@ -301,7 +301,7 @@ class SSO(Service):
try:
_resp = IDP.create_authn_response(
identity, userid=self.user,
authn=AUTHN_BROKER[self.environ["idp.authn_ref"]], sign_response=False, encrypt_cert=encrypt_cert,
authn=AUTHN_BROKER[self.environ["idp.authn_ref"]], encrypt_cert=encrypt_cert,
**resp_args)
except Exception, excp:
logging.error(exception_trace(excp))

1
example/sp-repoze/sp_conf.py
View File

@@ -78,6 +78,7 @@ CONFIG = {
#Information needed for generated cert (NO CERT) solution.
"authn_requests_signed": "true", #Will sign the request!
"want_assertions_signed": "false", #Demands that the assertion is signed.
"want_response_signed": "true",
"allow_unsolicited": "true", #Allows the message not to be ment for this sp.
#############################################################
"name": "LocalTestSPHans",

2
src/saml2/__init__.py
View File

@@ -558,6 +558,8 @@ class SamlBase(ExtensionContainer):
except AttributeError:
# Backwards compatibility with ET < 1.3
ElementTree._namespace_map[uri] = prefix
except ValueError:
pass
return ElementTree.tostring(self._to_element_tree(), encoding="UTF-8")

4
src/saml2/client_base.py
View File

@@ -122,8 +122,9 @@ class Base(Entity):
self.allow_unsolicited = False
self.authn_requests_signed = False
self.want_assertions_signed = False
self.want_response_signed = False
for foo in ["allow_unsolicited", "authn_requests_signed",
"logout_requests_signed", "want_assertions_signed"]:
"logout_requests_signed", "want_assertions_signed", "want_response_signed"]:
v = self.config.getattr(foo, "sp")
if v is True or v == 'true':
setattr(self, foo, True)
@@ -530,6 +531,7 @@ class Base(Entity):
"outstanding_certs": outstanding_certs,
"allow_unsolicited": self.allow_unsolicited,
"want_assertions_signed": self.want_assertions_signed,
"want_response_signed": self.want_response_signed,
"return_addrs": self.service_urls(),
"entity_id": self.config.entityid,
"attribute_converters": self.config.attribute_converters,

1
src/saml2/config.py
View File

@@ -80,6 +80,7 @@ SP_ARGS = [
"idp",
"aa",
"subject_data",
"want_response_signed",
"want_assertions_signed",
"authn_requests_signed",
"name_form",

27
src/saml2/entity.py
View File

@@ -460,10 +460,16 @@ class Entity(HTTPBase):
return signed_instance_factory(response, self.sec, to_sign)
if encrypt_assertion:
sign_class = [(class_name(response), response.id)]
if sign:
response.signature = pre_signature_part(response.id, self.sec.my_cert, 1)
cbxs = CryptoBackendXmlSec1(self.config.xmlsec_binary)
_, cert_file = make_temp("%s" % encrypt_cert, decode=False)
return cbxs.encrypt_assertion(response, cert_file, pre_encryption_part())#template(response.assertion.id))
#response = response_from_string(response_str)
response = cbxs.encrypt_assertion(response, cert_file, pre_encryption_part())#template(response.assertion.id))
if sign:
return signed_instance_factory(response, self.sec, sign_class)
else:
return response
if sign:
return self.sign(response, to_sign=to_sign)
@@ -811,16 +817,18 @@ class Entity(HTTPBase):
raise
xmlstr = self.unravel(xmlstr, binding, response_cls.msgtype)
origxml = xmlstr
if outstanding_certs is not None:
_response = samlp.any_response_from_string(xmlstr)
_, cert_file = make_temp("%s" % outstanding_certs[_response.in_response_to]["key"], decode=False)
cbxs = CryptoBackendXmlSec1(self.config.xmlsec_binary)
xmlstr = cbxs.decrypt(xmlstr, cert_file)
if len(_response.encrypted_assertion) > 0:
_, cert_file = make_temp("%s" % outstanding_certs[_response.in_response_to]["key"], decode=False)
cbxs = CryptoBackendXmlSec1(self.config.xmlsec_binary)
xmlstr = cbxs.decrypt(xmlstr, cert_file)
if not xmlstr: # Not a valid reponse
return None
try:
response = response.loads(xmlstr, False)
response = response.loads(xmlstr, False, origxml=origxml)
except SigverError, err:
logger.error("Signature Error: %s" % err)
return None
@@ -831,6 +839,13 @@ class Entity(HTTPBase):
logger.debug("XMLSTR: %s" % xmlstr)
for encrypted_assertion in response.response.encrypted_assertion:
if encrypted_assertion.extension_elements is not None:
assertion_list = extension_elements_to_elements(encrypted_assertion.extension_elements, [saml])
for assertion in assertion_list:
_assertion = saml.assertion_from_string(str(assertion))
response.response.assertion.append(_assertion)
if response:
response = response.verify()

8
src/saml2/response.py
View File

@@ -268,6 +268,7 @@ class StatusResponse(object):
self.in_response_to = None
self.signature_check = self.sec.correctly_signed_response
self.require_signature = False
self.require_response_signature = False
self.not_signed = False
self.asynchop = asynchop
@@ -318,7 +319,9 @@ class StatusResponse(object):
logger.debug("xmlstr: %s" % (self.xmlstr,))
try:
self.response = self.signature_check(xmldata, origdoc=origxml, must=self.require_signature)
self.response = self.signature_check(xmldata, origdoc=origxml, must=self.require_signature,
require_response_signature=self.require_response_signature)
except TypeError:
raise
except SignatureError:
@@ -452,7 +455,7 @@ class AuthnResponse(StatusResponse):
return_addrs=None, outstanding_queries=None,
timeslack=0, asynchop=True, allow_unsolicited=False,
test=False, allow_unknown_attributes=False,
want_assertions_signed=False, **kwargs):
want_assertions_signed=False, want_response_signed=False, **kwargs):
StatusResponse.__init__(self, sec_context, return_addrs, timeslack,
asynchop=asynchop)
@@ -469,6 +472,7 @@ class AuthnResponse(StatusResponse):
self.session_not_on_or_after = 0
self.allow_unsolicited = allow_unsolicited
self.require_signature = want_assertions_signed
self.require_response_signature = want_response_signed
self.test = test
self.allow_unknown_attributes = allow_unknown_attributes
#

9
src/saml2/server.py
View File

@@ -427,7 +427,8 @@ class Server(Entity):
def create_authn_response(self, identity, in_response_to, destination,
sp_entity_id, name_id_policy=None, userid=None,
name_id=None, authn=None, issuer=None,
sign_response=False, sign_assertion=None, encrypt_cert=None, **kwargs):
sign_response=None, sign_assertion=None, encrypt_cert=None, encrypt_assertion=None,
**kwargs):
""" Constructs an AuthenticationResponse
:param identity: Information about an user
@@ -465,7 +466,11 @@ class Server(Entity):
if sign_response is None:
sign_response = False
encrypt_assertion = self.config.getattr("encrypt_assertion", "idp")
if encrypt_assertion is None:
encrypt_assertion = self.config.getattr("encrypt_assertion", "idp")
if encrypt_assertion is None:
encrypt_assertion = False
if encrypt_assertion:
if encrypt_cert is not None:
verify_encrypt_cert = self.config.getattr("verify_encrypt_cert", "idp")

18
src/saml2/sigver.py
View File

@@ -1474,7 +1474,7 @@ class SecurityContext(object):
return self.correctly_signed_message(decoded_xml, "assertion", must,
origdoc, only_valid_cert)
def correctly_signed_response(self, decoded_xml, must=False, origdoc=None):
def correctly_signed_response(self, decoded_xml, must=False, origdoc=None, require_response_signature=False):
""" Check if a instance is correctly signed, if we have metadata for
the IdP that sent the info use that, if not use the key that are in
the message if any.
@@ -1492,26 +1492,18 @@ class SecurityContext(object):
if response.signature:
self._check_signature(decoded_xml, response, class_name(response),
origdoc)
elif require_response_signature:
raise SignatureError("Signature missing for response")
if isinstance(response, Response) and (response.assertion or
response.encrypted_assertion):
# Try to find the signing cert in the assertion
for assertion in (
response.assertion or response.encrypted_assertion):
if response.encrypted_assertion:
assertion_list = extension_elements_to_elements(assertion.extension_elements, [saml])
if len(assertion_list) > 0:
assertion = saml.assertion_from_string(str(assertion_list[0]))
response.assertion.append(assertion)
else:
decoded_xml = self.decrypt(assertion.encrypted_data.to_string())
assertion = saml.assertion_from_string(decoded_xml)
response.assertion.append(assertion)
if not assertion.signature:
if not hasattr(assertion, 'signature') or not assertion.signature:
logger.debug("unsigned")
if must:
raise SignatureError("Signature missing")
raise SignatureError("Signature missing for assertion")
continue
else:
logger.debug("signed")