openstack/deb-python-pysaml2
Code Issues Proposed changes

Update of pysaml2 and fix for very signature assertion.

Browse Source
This commit is contained in:
Hans Hörberg
2014-06-05 08:40:52 +02:00
parent 291bf156d6 87e51d6284
commit 5f6607a055
11 changed files with 108 additions and 83 deletions
Download Patch File Download Diff File Expand all files Collapse all files

43
src/saml2/assertion.py
View File

@@ -497,20 +497,33 @@ class Policy(object):
:return: A possibly modified AVA
"""
_rest = self.get_attribute_restrictions(sp_entity_id)
if _rest is None:
_rest = self.get_entity_categories(sp_entity_id, mdstore)
logger.debug("filter based on: %s" % _rest)
_ava = filter_attribute_value_assertions(ava.copy(), _rest)
_ava = None
if required or optional:
logger.debug("required: %s, optional: %s" % (required, optional))
ava1 = filter_on_attributes(
_ava = filter_on_attributes(
ava.copy(), required, optional, self.acs,
self.get_fail_on_missing_requested(sp_entity_id))
_ava.update(ava1)
return _ava
_rest = self.get_entity_categories(sp_entity_id, mdstore)
if _rest:
ava_ec = filter_attribute_value_assertions(ava.copy(), _rest)
if _ava is None:
_ava = ava_ec
else:
_ava.update(ava_ec)
_rest = self.get_attribute_restrictions(sp_entity_id)
if _rest:
if _ava is None:
_ava = ava.copy()
_ava = filter_attribute_value_assertions(_ava, _rest)
elif _ava is None:
_ava = ava.copy()
if _ava is None:
return {}
else:
return _ava
def restrict(self, ava, sp_entity_id, metadata=None):
""" Identity attribute names are expected to be expressed in
@@ -523,8 +536,8 @@ class Policy(object):
if metadata:
spec = metadata.attribute_requirement(sp_entity_id)
if spec:
ava = self.filter(ava, sp_entity_id, metadata,
spec["required"], spec["optional"])
return self.filter(ava, sp_entity_id, metadata,
spec["required"], spec["optional"])
return self.filter(ava, sp_entity_id, metadata, [], [])
@@ -757,5 +770,11 @@ class Assertion(dict):
policy.acs = self.acs
ava = policy.restrict(self, sp_entity_id, metadata)
self.update(ava)
for key, val in self.items():
if key in ava:
self[key] = ava[key]
else:
del self[key]
return ava

4
src/saml2/client_base.py
View File

@@ -545,8 +545,8 @@ class Base(Entity):
raise
except UnravelError:
return None
except Exception:
logger.error("XML parse error")
except Exception as err:
logger.error("XML parse error: %s" % err)
raise
#logger.debug(">> %s", resp)

5
src/saml2/entity.py
View File

@@ -127,7 +127,8 @@ class Entity(HTTPBase):
if _val.startswith("http"):
r = requests.request("GET", _val)
if r.status_code == 200:
setattr(self.config, item, r.text)
_, filename = make_temp(r.text, ".pem", False)
setattr(self.config, item, filename)
else:
raise Exception(
"Could not fetch certificate from %s" % _val)
@@ -840,7 +841,7 @@ class Entity(HTTPBase):
response = None
if self.config.accepted_time_diff:
timeslack = self.config.accepted_time_diff
kwargs["timeslack"] = self.config.accepted_time_diff
if "asynchop" not in kwargs:
if binding in [BINDING_SOAP, BINDING_PAOS]:

41
src/saml2/response.py
View File

@@ -475,7 +475,7 @@ class AuthnResponse(StatusResponse):
else:
self.outstanding_queries = {}
self.context = "AuthnReq"
self.came_from = ""
self.came_from = None
self.ava = None
self.assertion = None
self.assertions = []
@@ -507,7 +507,7 @@ class AuthnResponse(StatusResponse):
if self.asynchop:
if self.in_response_to in self.outstanding_queries:
self.came_from = self.outstanding_queries[self.in_response_to]
del self.outstanding_queries[self.in_response_to]
#del self.outstanding_queries[self.in_response_to]
try:
if not self.check_subject_confirmation_in_response_to(
self.in_response_to):
@@ -529,7 +529,7 @@ class AuthnResponse(StatusResponse):
def clear(self):
self._clear()
self.came_from = ""
self.came_from = None
self.ava = None
self.assertion = None
@@ -667,12 +667,12 @@ class AuthnResponse(StatusResponse):
if not later_than(data.not_on_or_after, data.not_before):
return False
if self.asynchop and not self.came_from:
if self.asynchop and self.came_from is None:
if data.in_response_to:
if data.in_response_to in self.outstanding_queries:
self.came_from = self.outstanding_queries[
data.in_response_to]
del self.outstanding_queries[data.in_response_to]
#del self.outstanding_queries[data.in_response_to]
elif self.allow_unsolicited:
pass
else:
@@ -744,7 +744,7 @@ class AuthnResponse(StatusResponse):
logger.info("Subject NameID: %s" % self.name_id)
return self.name_id
def _assertion(self, assertion):
def _assertion(self, assertion, verified=False):
"""
Check the assertion
:param assertion:
@@ -757,13 +757,14 @@ class AuthnResponse(StatusResponse):
raise SignatureError("Signature missing for assertion")
else:
logger.debug("signed")
try:
if self.require_signature:
self.sec.check_signature(assertion, class_name(assertion),
self.xmlstr)
except Exception as exc:
logger.error("correctly_signed_response: %s" % exc)
raise
if not verified:
try:
if self.require_signature:
self.sec.check_signature(assertion, class_name(assertion),
self.xmlstr)
except Exception as exc:
logger.error("correctly_signed_response: %s" % exc)
raise
self.assertion = assertion
logger.debug("assertion context: %s" % (self.context,))
@@ -791,14 +792,14 @@ class AuthnResponse(StatusResponse):
if self.asynchop:
if self.allow_unsolicited:
pass
elif not self.came_from:
elif self.came_from is None:
raise VerificationError("Came from")
return True
except Exception:
logger.exception("get subject")
raise
def decrypt_assertions(self, encrypted_assertions, key_file=""):
def decrypt_assertions(self, encrypted_assertions, decr_txt):
res = []
for encrypted_assertion in encrypted_assertions:
if encrypted_assertion.extension_elements:
@@ -806,8 +807,8 @@ class AuthnResponse(StatusResponse):
encrypted_assertion.extension_elements, [saml, samlp])
for assertion in assertions:
if assertion.signature:
if not self.sec.verify_signature(
"%s" % assertion, key_file,
if not self.sec.check_signature(
assertion, origdoc=decr_txt,
node_name=class_name(assertion)):
logger.error(
"Failed to verify signature on '%s'" % assertion)
@@ -826,21 +827,23 @@ class AuthnResponse(StatusResponse):
except AssertionError:
raise Exception("No assertion part")
res = []
if self.response.encrypted_assertion:
logger.debug("***Encrypted assertion/-s***")
decr_text = self.sec.decrypt(self.xmlstr, key_file)
resp = samlp.response_from_string(decr_text)
res = self.decrypt_assertions(resp.encrypted_assertion, key_file)
res = self.decrypt_assertions(resp.encrypted_assertion, decr_text)
if self.response.assertion:
self.response.assertion.extend(res)
else:
self.response.assertion = res
self.response.encrypted_assertion = []
self.xmlstr = decr_text
if self.response.assertion:
logger.debug("***Unencrypted assertion***")
for assertion in self.response.assertion:
if not self._assertion(assertion):
if not self._assertion(assertion, assertion in res):
return False
else:
self.assertions.append(assertion)

3
src/saml2/sigver.py
View File

@@ -35,7 +35,6 @@ from Crypto.PublicKey import RSA
from saml2.cert import OpenSSLWrapper
from saml2.extension import pefim
from saml2.saml import EncryptedAssertion
from saml2.samlp import Response
import xmldsig as ds
@@ -1378,7 +1377,7 @@ class SecurityContext(object):
"""
:param item: Parsed entity
:param node_name:
:param node_name: The name of the class that is signed
:param origdoc: The original XML string
:param id_attr:
:param must:

50
tests/sp_2_conf.py
View File

@@ -1,48 +1,50 @@
from pathutils import full_path
CONFIG = {
"entityid" : "urn:mace:example.com:saml:roland:sp",
"name" : "urn:mace:example.com:saml:roland:sp",
"entityid": "urn:mace:example.com:saml:roland:sp",
"name": "urn:mace:example.com:saml:roland:sp",
"description": "My own SP",
"service": {
"sp": {
"endpoints":{
"assertion_consumer_service": ["http://lingon.catalogix.se:8087/"],
},
"endpoints": {
"assertion_consumer_service": [
"http://lingon.catalogix.se:8087/"],
},
"required_attributes": ["surName", "givenName", "mail"],
"optional_attributes": ["title"],
"idp": ["urn:mace:example.com:saml:roland:idp"],
}
}
},
"debug" : 1,
"key_file" : full_path("test.key"),
"cert_file" : full_path("test.pem"),
"xmlsec_binary" : None,
"debug": 1,
"key_file": full_path("test.key"),
"cert_file": full_path("test.pem"),
"xmlsec_binary": None,
"metadata": {
"local": [full_path("idp_2.xml")],
},
"virtual_organization" : {
"urn:mace:example.com:it:tek":{
"nameid_format" : "urn:oid:1.3.6.1.4.1.1466.115.121.1.15-NameID",
},
"virtual_organization": {
"urn:mace:example.com:it:tek": {
"nameid_format": "urn:oid:1.3.6.1.4.1.1466.115.121.1.15-NameID",
"common_identifier": "umuselin",
}
}
},
"subject_data": full_path("subject_data.db"),
"accepted_time_diff": 60,
"attribute_map_dir" : full_path("attributemaps"),
"attribute_map_dir": full_path("attributemaps"),
"organization": {
"name": ("AB Exempel", "se"),
"display_name": ("AB Exempel", "se"),
"url": "http://www.example.org",
},
"contact_person": [{
"given_name": "Roland",
"sur_name": "Hedberg",
"telephone_number": "+46 70 100 0000",
"email_address": ["tech@eample.com", "tech@example.org"],
"contact_type": "technical"
},
"contact_person": [{
"given_name": "Roland",
"sur_name": "Hedberg",
"telephone_number": "+46 70 100 0000",
"email_address": ["tech@eample.com",
"tech@example.org"],
"contact_type": "technical"
},
],
"secret": "0123456789",
"only_use_keys_in_metadata": True
}
}

16
tests/test_20_assertion.py
View File

@@ -173,16 +173,20 @@ def test_ava_filter_2():
"mail": "derek@example.com"}
# mail removed because it doesn't match the regular expression
# So this should fail.
raises(MissingValue, policy.filter, ava, 'urn:mace:umu.se:saml:roland:sp',
None, [mail], [gn, sn])
_ava =policy.filter(ava, 'urn:mace:umu.se:saml:roland:sp', None, [mail],
[gn, sn])
assert _eq(_ava.keys(), ["givenName", "surName"])
ava = {"givenName": "Derek",
"surName": "Jeter"}
# it wasn't there to begin with
raises(Exception, policy.filter, ava, 'urn:mace:umu.se:saml:roland:sp',
None, [gn, sn, mail])
try:
policy.filter(ava, 'urn:mace:umu.se:saml:roland:sp', None,
[gn, sn, mail])
except MissingValue:
pass
def test_ava_filter_dont_fail():
@@ -843,4 +847,4 @@ def test_assertion_with_authn_instant():
if __name__ == "__main__":
test_ava_filter_dont_fail()
test_assertion_2()

4
tests/test_37_entity_categories.py
View File

@@ -64,7 +64,7 @@ def test_filter_ava2():
"default": {
"lifetime": {"minutes": 15},
#"attribute_restrictions": None # means all I have
"entity_categories": ["edugain"]
"entity_categories": ["refeds", "edugain"]
}
})
@@ -158,7 +158,7 @@ def test_idp_policy_filter():
"norEduPersonNIN": "19800101134"}
policy = idp.config.getattr("policy", "idp")
policy.filter(ava, "urn:mace:example.com:saml:roland:sp", idp.metadata)
ava = policy.filter(ava, "urn:mace:example.com:saml:roland:sp", idp.metadata)
print ava
assert ava.keys() == ["eduPersonTargetedID"] # because no entity category

12
tests/test_41_response.py
View File

@@ -8,13 +8,11 @@ from saml2.server import Server
from saml2.response import response_factory
from saml2.response import StatusResponse
from saml2.response import AuthnResponse
from saml2.sigver import security_context, SignatureError
from saml2.sigver import MissingKey
from pytest import raises
from saml2.sigver import SignatureError
FALSE_ASSERT_SIGNED = "saml_false_signed.xml"
TIMESLACK = 2592000 # Roughly 3 month
def _eq(l1, l2):
return set(l1) == set(l2)
@@ -73,7 +71,7 @@ class TestResponse:
"http://lingon.catalogix.se:8087/"],
outstanding_queries={
"id12": "http://localhost:8088/sso"},
timeslack=10000, decode=False)
timeslack=TIMESLACK, decode=False)
assert isinstance(resp, StatusResponse)
assert isinstance(resp, AuthnResponse)
@@ -85,7 +83,7 @@ class TestResponse:
"http://lingon.catalogix.se:8087/"],
outstanding_queries={
"id12": "http://localhost:8088/sso"},
timeslack=10000, decode=False)
timeslack=TIMESLACK, decode=False)
assert isinstance(resp, StatusResponse)
assert isinstance(resp, AuthnResponse)
@@ -98,7 +96,7 @@ class TestResponse:
outstanding_queries={
"bahigehogffohiphlfmplepdpcohkhhmheppcdie":
"http://localhost:8088/sso"},
timeslack=10000, decode=False)
timeslack=TIMESLACK, decode=False)
assert isinstance(resp, StatusResponse)
assert isinstance(resp, AuthnResponse)

2
tests/test_44_authnresp.py
View File

@@ -71,7 +71,7 @@ class TestAuthnResponse:
print self.ar.__dict__
assert self.ar.came_from == 'http://localhost:8088/sso'
assert self.ar.session_id() == "id12"
assert self.ar.ava["eduPersonAffiliation"] == IDENTITY["eduPersonAffiliation"]
assert self.ar.ava["givenName"] == IDENTITY["givenName"]
assert self.ar.name_id
assert self.ar.issuer() == 'urn:mace:example.com:saml:roland:idp'

11
tests/test_50_server.py
View File

@@ -227,18 +227,17 @@ class TestServer1():
assert assertion.attribute_statement
attribute_statement = assertion.attribute_statement
print attribute_statement
assert len(attribute_statement[0].attribute) == 5
assert len(attribute_statement[0].attribute) == 4
# Pick out one attribute
attr = None
for attr in attribute_statement[0].attribute:
if attr.friendly_name == "edupersonentitlement":
if attr.friendly_name == "givenname":
break
assert len(attr.attribute_value) == 1
assert attr.name == "urn:oid:1.3.6.1.4.1.5923.1.1.1.7"
assert attr.name_format == "urn:oasis:names:tc:SAML:2" \
".0:attrname-format:uri"
assert attr.name == "urn:oid:2.5.4.42"
assert attr.name_format == "urn:oasis:names:tc:SAML:2.0:attrname-format:uri"
value = attr.attribute_value[0]
assert value.text.strip() == "Short stop"
assert value.text.strip() == "Derek"
assert value.get_type() == "xs:string"
assert assertion.subject
assert assertion.subject.name_id