openstack/deb-python-pysaml2
Code Issues Proposed changes

Made things work after the last merge.

Browse Source
This commit is contained in:
Roland Hedberg
2013-05-09 11:38:12 +02:00
parent ce1c2c95fa
commit a432390da2
6 changed files with 161 additions and 117 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
src/saml2/assertion.py
View File

@@ -97,8 +97,11 @@ def filter_on_attributes(ava, required=None, optional=None):
found = False found = False
nform = "" nform = ""
for nform in ["friendly_name", "name"]: for nform in ["friendly_name", "name"]:
if nform in attr: try:
_fn = _match(attr[nform], ava) _fn = _match(attr[nform], ava)
except KeyError:
pass
else:
if _fn: if _fn:
try: try:
values = [av["text"] for av in attr["attribute_value"]] values = [av["text"] for av in attr["attribute_value"]]
@@ -239,6 +242,8 @@ def filter_attribute_value_assertions(ava, attribute_restrictions=None):
else: else:
if _rests is None: if _rests is None:
continue continue
if isinstance(vals, basestring):
vals = [vals]
rvals = [] rvals = []
for restr in _rests: for restr in _rests:
for val in vals: for val in vals:
@@ -289,6 +294,8 @@ class Policy(object):
self._restrictions = restrictions.copy() self._restrictions = restrictions.copy()
for who, spec in self._restrictions.items(): for who, spec in self._restrictions.items():
if spec is None:
continue
try: try:
items = spec["entity_categories"] items = spec["entity_categories"]
except KeyError: except KeyError:
@@ -311,14 +318,14 @@ class Policy(object):
if restr is None: if restr is None:
continue continue
_are = {}
for key, values in restr.items(): for key, values in restr.items():
if not values: if not values:
spec["attribute_restrictions"][key.lower()] = None _are[key.lower()] = None
continue continue
spec["attribute_restrictions"][key.lower()] = \ _are[key.lower()] = [re.compile(value) for value in values]
[re.compile(value) for value in values] spec["attribute_restrictions"] = _are
logger.debug("policy restrictions: %s" % self._restrictions) logger.debug("policy restrictions: %s" % self._restrictions)
return self._restrictions return self._restrictions
@@ -430,6 +437,7 @@ class Policy(object):
for attr in attrs: for attr in attrs:
restrictions[attr] = None restrictions[attr] = None
if mds:
try: try:
ecs = mds.entity_categories(sp_entity_id) ecs = mds.entity_categories(sp_entity_id)
except KeyError: except KeyError:

18
src/saml2/sigver.py
View File

@@ -72,7 +72,7 @@ def signed(item):
return False return False
def _get_xmlsec_binary(paths=None): def get_xmlsec_binary(paths=None):
""" """
Tries to find the xmlsec1 binary. Tries to find the xmlsec1 binary.
@@ -107,6 +107,7 @@ def _get_xmlsec_binary(paths=None):
raise Exception("Can't find %s" % bin_name) raise Exception("Can't find %s" % bin_name)
def _get_xmlsec_cryptobackend(path=None, search_paths=None, debug=False): def _get_xmlsec_cryptobackend(path=None, search_paths=None, debug=False):
""" """
Initialize a CryptoBackendXmlSec1 crypto backend. Initialize a CryptoBackendXmlSec1 crypto backend.
@@ -114,7 +115,7 @@ def _get_xmlsec_cryptobackend(path=None, search_paths=None, debug=False):
This function is now internal to this module. This function is now internal to this module.
""" """
if path is None: if path is None:
path=_get_xmlsec_binary(paths=search_paths) path = get_xmlsec_binary(paths=search_paths)
return CryptoBackendXmlSec1(path, debug=debug) return CryptoBackendXmlSec1(path, debug=debug)
@@ -144,7 +145,6 @@ class DecryptError(Exception):
# -------------------------------------------------------------------------- # --------------------------------------------------------------------------
def _make_vals(val, klass, seccont, klass_inst=None, prop=None, part=False, def _make_vals(val, klass, seccont, klass_inst=None, prop=None, part=False,
base64encode=False, elements_to_sign=None): base64encode=False, elements_to_sign=None):
""" """
@@ -173,7 +173,8 @@ def _make_vals(val, klass, seccont, klass_inst=None, prop=None, part=False,
except ValueError: except ValueError:
if not part: if not part:
cis = [_make_vals(sval, klass, seccont, klass_inst, prop, cis = [_make_vals(sval, klass, seccont, klass_inst, prop,
True, base64encode, elements_to_sign) for sval in val] True, base64encode, elements_to_sign) for sval
in val]
setattr(klass_inst, prop, cis) setattr(klass_inst, prop, cis)
else: else:
raise raise
@@ -485,6 +486,7 @@ def sha1_digest(msg):
class Signer(object): class Signer(object):
"""Abstract base class for signing algorithms.""" """Abstract base class for signing algorithms."""
def sign(self, msg, key): def sign(self, msg, key):
"""Sign ``msg`` with ``key`` and return the signature.""" """Sign ``msg`` with ``key`` and return the signature."""
raise NotImplementedError raise NotImplementedError
@@ -544,6 +546,7 @@ def verify_redirect_signature(info, cert):
else: else:
raise Unsupported("Signature algorithm: %s" % info["SigAlg"]) raise Unsupported("Signature algorithm: %s" % info["SigAlg"])
LOG_LINE = 60 * "=" + "\n%s\n" + 60 * "-" + "\n%s" + 60 * "=" LOG_LINE = 60 * "=" + "\n%s\n" + 60 * "-" + "\n%s" + 60 * "="
LOG_LINE_2 = 60 * "=" + "\n%s\n%s\n" + 60 * "-" + "\n%s" + 60 * "=" LOG_LINE_2 = 60 * "=" + "\n%s\n%s\n" + 60 * "-" + "\n%s" + 60 * "="
@@ -588,7 +591,6 @@ def read_cert_from_file(cert_file, cert_type):
class CryptoBackend(): class CryptoBackend():
def __init__(self, debug=False): def __init__(self, debug=False):
self.debug = debug self.debug = debug
@@ -767,6 +769,7 @@ class CryptoBackendXmlSec1(CryptoBackend):
ntf.seek(0) ntf.seek(0)
return p_out, p_err, ntf.read() return p_out, p_err, ntf.read()
class CryptoBackendXMLSecurity(CryptoBackend): class CryptoBackendXMLSecurity(CryptoBackend):
""" """
CryptoBackend implementation using pyXMLSecurity to sign and verify CryptoBackend implementation using pyXMLSecurity to sign and verify
@@ -804,6 +807,7 @@ class CryptoBackendXMLSecurity(CryptoBackend):
""" """
import xmlsec import xmlsec
import lxml.etree import lxml.etree
xml = xmlsec.parse_xml(statement) xml = xmlsec.parse_xml(statement)
signed = xmlsec.sign(xml, key_file) signed = xmlsec.sign(xml, key_file)
return lxml.etree.tostring(signed, xml_declaration=True) return lxml.etree.tostring(signed, xml_declaration=True)
@@ -825,12 +829,14 @@ class CryptoBackendXMLSecurity(CryptoBackend):
if cert_type != "pem": if cert_type != "pem":
raise Unsupported("Only PEM certs supported here") raise Unsupported("Only PEM certs supported here")
import xmlsec import xmlsec
xml = xmlsec.parse_xml(signedtext) xml = xmlsec.parse_xml(signedtext)
try: try:
return xmlsec.verify(xml, cert_file) return xmlsec.verify(xml, cert_file)
except xmlsec.XMLSigException: except xmlsec.XMLSigException:
return False return False
def security_context(conf, debug=None): def security_context(conf, debug=None):
""" Creates a security context based on the configuration """ Creates a security context based on the configuration
@@ -852,7 +858,7 @@ def security_context(conf, debug=None):
if conf.crypto_backend == 'xmlsec1': if conf.crypto_backend == 'xmlsec1':
xmlsec_binary = conf.xmlsec_binary xmlsec_binary = conf.xmlsec_binary
if not xmlsec_binary: if not xmlsec_binary:
xmlsec_binary = _get_xmlsec_binary() xmlsec_binary = get_xmlsec_binary()
# verify that xmlsec is where it's supposed to be # verify that xmlsec is where it's supposed to be
if not os.path.exists(xmlsec_binary): if not os.path.exists(xmlsec_binary):
#if not os.access(, os.F_OK): #if not os.access(, os.F_OK):

10
tests/idp_conf_ec.py
View File

@@ -1,8 +1,12 @@
from saml2 import BINDING_SOAP, BINDING_HTTP_REDIRECT, BINDING_HTTP_POST from saml2.sigver import get_xmlsec_binary
from saml2.saml import NAMEID_FORMAT_PERSISTENT from saml2 import BINDING_SOAP
from saml2 import BINDING_HTTP_REDIRECT
from saml2 import BINDING_HTTP_POST
from saml2.saml import NAME_FORMAT_URI from saml2.saml import NAME_FORMAT_URI
from pathutils import full_path, xmlsec_path from pathutils import full_path
xmlsec_path = get_xmlsec_binary(["/opt/local/bin"])
BASE = "http://localhost:8088" BASE = "http://localhost:8088"

21
tests/test_20_assertion.py
View File

@@ -172,15 +172,15 @@ def test_ava_filter_2():
"surName": "Jeter", "surName": "Jeter",
"mail": "derek@example.com"} "mail": "derek@example.com"}
raises(Exception, policy.filter, ava, 'urn:mace:umu.se:saml:roland:sp', raises(MissingValue, policy.filter, ava, 'urn:mace:umu.se:saml:roland:sp',
[mail], [gn, sn]) None, [mail], [gn, sn])
ava = {"givenName": "Derek", ava = {"givenName": "Derek",
"surName": "Jeter"} "surName": "Jeter"}
# it wasn't there to begin with # it wasn't there to begin with
raises(Exception, policy.filter, ava, 'urn:mace:umu.se:saml:roland:sp', raises(Exception, policy.filter, ava, 'urn:mace:umu.se:saml:roland:sp',
[gn, sn, mail]) None, [gn, sn, mail])
def test_filter_attribute_value_assertions_0(AVA): def test_filter_attribute_value_assertions_0(AVA):
@@ -643,7 +643,7 @@ def test_req_opt():
'uid': 'rohe0002', 'edupersonaffiliation': 'staff'} 'uid': 'rohe0002', 'edupersonaffiliation': 'staff'}
sp_entity_id = "urn:mace:example.com:saml:curt:sp" sp_entity_id = "urn:mace:example.com:saml:curt:sp"
fava = policy.filter(ava, sp_entity_id, req, opt) fava = policy.filter(ava, sp_entity_id, None, req, opt)
assert fava assert fava
@@ -736,19 +736,20 @@ def test_filter_ava_5():
"default": { "default": {
"lifetime": {"minutes": 15}, "lifetime": {"minutes": 15},
#"attribute_restrictions": None # means all I have #"attribute_restrictions": None # means all I have
"entity_categories": ["swami", "edugain"] "entity_categories": ["swamid", "edugain"]
} }
}) })
ava = {"givenName": ["Derek"], "surName": ["Jeter"], ava = {"givenName": ["Derek"], "surName": ["Jeter"],
"mail": ["derek@nyy.mlb.com", "dj@example.com"]} "mail": ["derek@nyy.mlb.com", "dj@example.com"]}
# No restrictions apply ava = policy.filter(ava, "urn:mace:example.com:saml:curt:sp", None, [], [])
ava = policy.filter(ava, "urn:mace:example.com:saml:curt:sp", [], [])
assert _eq(ava.keys(), ['mail', 'givenName', 'surName']) # using entity_categories means there *always* are restrictions
assert _eq(ava["mail"], ["derek@nyy.mlb.com", "dj@example.com"]) # in this case the only allowed attribute is eduPersonTargetedID
# which isn't available in the ava hence zip is returned.
assert ava == {}
if __name__ == "__main__": if __name__ == "__main__":
test_assertion_with_noop_attribute_conv() test_filter_ava_5()

49
tests/test_31_config.py
View File

@@ -18,11 +18,13 @@ sp1 = {
"service": { "service": {
"sp": { "sp": {
"endpoints": { "endpoints": {
"assertion_consumer_service" : ["http://lingon.catalogix.se:8087/"], "assertion_consumer_service": [
"http://lingon.catalogix.se:8087/"],
}, },
"name": "test", "name": "test",
"idp": { "idp": {
"urn:mace:example.com:saml:roland:idp": {'single_sign_on_service': "urn:mace:example.com:saml:roland:idp": {
'single_sign_on_service':
{'urn:oasis:names:tc:SAML:2.0:bindings:HTTP-Redirect': {'urn:oasis:names:tc:SAML:2.0:bindings:HTTP-Redirect':
'http://localhost:8088/sso/'}}, 'http://localhost:8088/sso/'}},
} }
@@ -53,7 +55,8 @@ sp2 = {
"service": { "service": {
"sp": { "sp": {
"endpoints": { "endpoints": {
"assertion_consumer_service" : ["http://lingon.catalogix.se:8087/"], "assertion_consumer_service": [
"http://lingon.catalogix.se:8087/"],
}, },
"required_attributes": ["surName", "givenName", "mail"], "required_attributes": ["surName", "givenName", "mail"],
"optional_attributes": ["title"], "optional_attributes": ["title"],
@@ -96,7 +99,8 @@ IDP2 = {
"idp": { "idp": {
"endpoints": { "endpoints": {
"single_sign_on_service": ["http://localhost:8088/"], "single_sign_on_service": ["http://localhost:8088/"],
"single_logout_service" : [("http://localhost:8088/", BINDING_HTTP_REDIRECT)], "single_logout_service": [
("http://localhost:8088/", BINDING_HTTP_REDIRECT)],
}, },
"policy": { "policy": {
"default": { "default": {
@@ -147,7 +151,8 @@ ECP_SP = {
"service": { "service": {
"sp": { "sp": {
"endpoints": { "endpoints": {
"assertion_consumer_service" : ["http://lingon.catalogix.se:8087/"], "assertion_consumer_service": [
"http://lingon.catalogix.se:8087/"],
}, },
"ecp": { "ecp": {
"130.239.": "http://example.com/idp", "130.239.": "http://example.com/idp",
@@ -157,9 +162,11 @@ ECP_SP = {
#"xmlsec_binary" : "/opt/local/bin/xmlsec1", #"xmlsec_binary" : "/opt/local/bin/xmlsec1",
} }
def _eq(l1, l2): def _eq(l1, l2):
return set(l1) == set(l2) return set(l1) == set(l2)
def test_1(): def test_1():
c = SPConfig().load(sp1) c = SPConfig().load(sp1)
c.context = "sp" c.context = "sp"
@@ -173,11 +180,13 @@ def test_1():
assert len(c._sp_idp) == 1 assert len(c._sp_idp) == 1
assert c._sp_idp.keys() == ["urn:mace:example.com:saml:roland:idp"] assert c._sp_idp.keys() == ["urn:mace:example.com:saml:roland:idp"]
assert c._sp_idp.values() == [{'single_sign_on_service': assert c._sp_idp.values() == [{'single_sign_on_service':
{'urn:oasis:names:tc:SAML:2.0:bindings:HTTP-Redirect': {
'urn:oasis:names:tc:SAML:2.0:bindings:HTTP-Redirect':
'http://localhost:8088/sso/'}}] 'http://localhost:8088/sso/'}}]
assert c.only_use_keys_in_metadata assert c.only_use_keys_in_metadata
def test_2(): def test_2():
c = SPConfig().load(sp2) c = SPConfig().load(sp2)
c.context = "sp" c.context = "sp"
@@ -192,9 +201,11 @@ def test_2():
assert len(c._sp_idp) == 1 assert len(c._sp_idp) == 1
assert c._sp_idp.keys() == [""] assert c._sp_idp.keys() == [""]
assert c._sp_idp.values() == ["https://example.com/saml2/idp/SSOService.php"] assert c._sp_idp.values() == [
"https://example.com/saml2/idp/SSOService.php"]
assert c.only_use_keys_in_metadata is True assert c.only_use_keys_in_metadata is True
def test_minimum(): def test_minimum():
minimum = { minimum = {
"entityid": "urn:mace:example.com:saml:roland:sp", "entityid": "urn:mace:example.com:saml:roland:sp",
@@ -217,6 +228,7 @@ def test_minimum():
assert c is not None assert c is not None
def test_idp_1(): def test_idp_1():
c = IdPConfig().load(IDP1) c = IdPConfig().load(IDP1)
c.context = "idp" c.context = "idp"
@@ -224,8 +236,10 @@ def test_idp_1():
print c print c
assert c.endpoint("single_sign_on_service")[0] == 'http://localhost:8088/' assert c.endpoint("single_sign_on_service")[0] == 'http://localhost:8088/'
attribute_restrictions = c.getattr("policy","idp").get_attribute_restriction("") attribute_restrictions = c.getattr("policy",
assert attribute_restrictions["eduPersonAffiliation"][0].match("staff") "idp").get_attribute_restriction("")
assert attribute_restrictions["edupersonaffiliation"][0].match("staff")
def test_idp_2(): def test_idp_2():
c = IdPConfig().load(IDP2) c = IdPConfig().load(IDP2)
@@ -237,8 +251,10 @@ def test_idp_2():
assert c.endpoint("single_logout_service", assert c.endpoint("single_logout_service",
BINDING_HTTP_REDIRECT) == ["http://localhost:8088/"] BINDING_HTTP_REDIRECT) == ["http://localhost:8088/"]
attribute_restrictions = c.getattr("policy","idp").get_attribute_restriction("") attribute_restrictions = c.getattr("policy",
assert attribute_restrictions["eduPersonAffiliation"][0].match("staff") "idp").get_attribute_restriction("")
assert attribute_restrictions["edupersonaffiliation"][0].match("staff")
def test_wayf(): def test_wayf():
c = SPConfig().load_file("server_conf") c = SPConfig().load_file("server_conf")
@@ -266,6 +282,7 @@ def test_wayf():
assert root_logger.name == "saml2" assert root_logger.name == "saml2"
assert root_logger.level == 20 assert root_logger.level == 20
def test_conf_syslog(): def test_conf_syslog():
c = SPConfig().load_file("server_conf_syslog") c = SPConfig().load_file("server_conf_syslog")
c.context = "sp" c.context = "sp"
@@ -307,12 +324,14 @@ def test_3():
assert cnf.metadata is not None assert cnf.metadata is not None
assert cnf.attribute_converters is not None assert cnf.attribute_converters is not None
def test_sp(): def test_sp():
cnf = SPConfig() cnf = SPConfig()
cnf.load_file(dotname("sp_1_conf")) cnf.load_file(dotname("sp_1_conf"))
assert cnf.endpoint("assertion_consumer_service") == \ assert cnf.endpoint("assertion_consumer_service") == \
["http://lingon.catalogix.se:8087/"] ["http://lingon.catalogix.se:8087/"]
def test_dual(): def test_dual():
cnf = Config().load_file(dotname("idp_sp_conf")) cnf = Config().load_file(dotname("idp_sp_conf"))
@@ -322,6 +341,7 @@ def test_dual():
assert idpe assert idpe
assert spe != idpe assert spe != idpe
def test_ecp(): def test_ecp():
cnf = SPConfig() cnf = SPConfig()
cnf.load(ECP_SP) cnf.load(ECP_SP)
@@ -332,6 +352,7 @@ def test_ecp():
eid = cnf.ecp_endpoint("130.238.20.20") eid = cnf.ecp_endpoint("130.238.20.20")
assert eid is None assert eid is None
def test_assertion_consumer_service(): def test_assertion_consumer_service():
c = IdPConfig() c = IdPConfig()
c.load_file(dotname("idp_conf")) c.load_file(dotname("idp_conf"))
@@ -342,4 +363,8 @@ def test_assertion_consumer_service():
entity_id = "https://www.zimride.com/shibboleth" entity_id = "https://www.zimride.com/shibboleth"
acs = c.metadata.assertion_consumer_service(entity_id) acs = c.metadata.assertion_consumer_service(entity_id)
assert len(acs) == 1 assert len(acs) == 1
assert acs[0]["location"] == 'https://www.zimride.com/Shibboleth.sso/SAML2/POST' assert acs[0][
"location"] == 'https://www.zimride.com/Shibboleth.sso/SAML2/POST'
if __name__ == "__main__":
test_idp_1()

10
tests/test_37_entity_categories.py
View File

@@ -50,12 +50,12 @@ def test_filter_ava():
} }
}) })
ava = {"givenName": ["Derek"], "surname": ["Jeter"], ava = {"givenName": ["Derek"], "sn": ["Jeter"],
"email": ["derek@nyy.mlb.com", "dj@example.com"], "c": ["USA"]} "email": ["derek@nyy.mlb.com", "dj@example.com"], "c": ["USA"]}
ava = policy.filter(ava, "https://connect.sunet.se/shibboleth", MDS) ava = policy.filter(ava, "https://connect.sunet.se/shibboleth", MDS)
assert _eq(ava.keys(), ['email', 'givenName', 'surname', 'c']) assert _eq(ava.keys(), ['email', 'givenName', 'sn', 'c'])
assert _eq(ava["email"], ["derek@nyy.mlb.com", "dj@example.com"]) assert _eq(ava["email"], ["derek@nyy.mlb.com", "dj@example.com"])
@@ -68,7 +68,7 @@ def test_filter_ava2():
} }
}) })
ava = {"givenName": ["Derek"], "surname": ["Jeter"], ava = {"givenName": ["Derek"], "sn": ["Jeter"],
"email": ["derek@nyy.mlb.com"], "c": ["USA"], "email": ["derek@nyy.mlb.com"], "c": ["USA"],
"eduPersonTargetedID": "foo!bar!xyz"} "eduPersonTargetedID": "foo!bar!xyz"}
@@ -92,7 +92,7 @@ def test_filter_ava3():
disable_ssl_certificate_validation=True) disable_ssl_certificate_validation=True)
mds.imp({"local": [full_path("entity_cat_sfs_hei.xml")]}) mds.imp({"local": [full_path("entity_cat_sfs_hei.xml")]})
ava = {"givenName": ["Derek"], "surname": ["Jeter"], ava = {"givenName": ["Derek"], "sn": ["Jeter"],
"email": ["derek@nyy.mlb.com"], "c": ["USA"], "email": ["derek@nyy.mlb.com"], "c": ["USA"],
"eduPersonTargetedID": "foo!bar!xyz", "eduPersonTargetedID": "foo!bar!xyz",
"norEduPersonNIN": "19800101134"} "norEduPersonNIN": "19800101134"}
@@ -105,7 +105,7 @@ def test_filter_ava3():
def test_idp_policy_filter(): def test_idp_policy_filter():
idp = Server("idp_conf_ec") idp = Server("idp_conf_ec")
ava = {"givenName": ["Derek"], "surname": ["Jeter"], ava = {"givenName": ["Derek"], "sn": ["Jeter"],
"email": ["derek@nyy.mlb.com"], "c": ["USA"], "email": ["derek@nyy.mlb.com"], "c": ["USA"],
"eduPersonTargetedID": "foo!bar!xyz", "eduPersonTargetedID": "foo!bar!xyz",
"norEduPersonNIN": "19800101134"} "norEduPersonNIN": "19800101134"}