Fixed attribute filtering logic, take 2.

This commit is contained in:
Roland Hedberg 2014-06-03 09:28:11 +02:00
parent 884e75b1f5
commit 84a1453a35
7 changed files with 76 additions and 52 deletions

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

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
}
}

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()

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

View File

@ -98,7 +98,7 @@ class TestResponse:
outstanding_queries={
"bahigehogffohiphlfmplepdpcohkhhmheppcdie":
"http://localhost:8088/sso"},
timeslack=10000, decode=False)
timeslack=1000000, decode=False)
assert isinstance(resp, StatusResponse)
assert isinstance(resp, AuthnResponse)

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'

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