openstack/deb-python-pysaml2
Code Issues Proposed changes

Merge pull request #87 from HaToHo/master

Browse Source 
Only validate certificate and set client certificate tp authn request.
This commit is contained in:
Roland Hedberg
2014-03-05 18:49:23 +01:00
parent b0c0cf854b 7edc2447a5
commit a9d0b7c474
12 changed files with 63 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
.gitignore vendored
View File

@@ -149,3 +149,9 @@ example/sp/sp_nocert2.xml
example/sp/test.py
example/sp/sp_conf.py
example/sp/nocert_sp_conf/sp.xml
example/sp/nocert_sp_conf/sp_conf.py
example/sp/nocert_sp_conf/who.ini

1
example/sp/who.ini
View File

@@ -17,6 +17,7 @@ saml_conf = sp_conf
remember_name = auth_tkt
sid_store = outstanding
idp_query_param = IdPEntityId
discovery = http://130.239.201.5/role/idp.ds
[general]
request_classifier = s2repoze.plugins.challenge_decider:my_request_classifier

1
setup.py
View File

@@ -44,6 +44,7 @@ install_requires = [
'zope.interface',
'repoze.who',
'pycrypto', #'Crypto'
'pytz'
]
tests_require = [

6
src/saml2/cert.py
View File

@@ -1,11 +1,9 @@
__author__ = 'haho0032'
import base64
import traceback
from M2Crypto.util import passphrase_callback
import datetime
import dateutil.parser
import pytz
__author__ = 'haho0032'
from OpenSSL import crypto
from os.path import exists, join
from os import remove

10
src/saml2/client_base.py
View File

@@ -231,7 +231,9 @@ class Base(Entity):
:param kwargs: Extra key word arguments
:return: <samlp:AuthnRequest> instance
"""
client_crt = None
if "client_crt" in kwargs:
client_crt = kwargs["client_crt"]
args = {}
try:
args["assertion_consumer_service_url"] = kwargs[
@@ -299,9 +301,11 @@ class Base(Entity):
except KeyError:
pass
if sign and self.sec.cert_handler.generate_cert():
if (sign and self.sec.cert_handler.generate_cert()) or client_crt is not None:
with self.lock:
self.sec.cert_handler.update_cert(True)
self.sec.cert_handler.update_cert(True, client_crt)
if client_crt is not None:
sign_prepare = True
return self._message(AuthnRequest, destination, message_id, consent,
extensions, sign, sign_prepare,
protocol_binding=binding,

1
src/saml2/config.py
View File

@@ -92,6 +92,7 @@ SP_ARGS = [
AA_IDP_ARGS = [
"sign_assertion",
"want_authn_requests_signed",
"want_authn_requests_only_with_valid_cert",
"provided_attributes",
"subject_data",
"sp",

7
src/saml2/entity.py
View File

@@ -546,7 +546,12 @@ class Entity(HTTPBase):
origdoc = xmlstr
xmlstr = self.unravel(xmlstr, binding, request_cls.msgtype)
must = self.config.getattr("want_authn_requests_signed", "idp")
_request = _request.loads(xmlstr, binding, origdoc=origdoc, must=must)
only_valid_cert = self.config.getattr("want_authn_requests_only_with_valid_cert", "idp")
if only_valid_cert is None:
only_valid_cert = False
if only_valid_cert:
must = True
_request = _request.loads(xmlstr, binding, origdoc=origdoc, must=must, only_valid_cert=only_valid_cert)
_log_debug("Loaded request")

6
src/saml2/md.py
View File

@@ -1041,6 +1041,8 @@ class IDPSSODescriptorType_(SSODescriptorType_):
c_cardinality['attribute'] = {"min": 0}
c_attributes['WantAuthnRequestsSigned'] = ('want_authn_requests_signed',
'boolean', False)
c_attributes['WantAuthnRequestsOnlyWithValidCert'] = ('want_authn_requests_only_with_valid_cert',
'boolean', False)
c_child_order.extend(['single_sign_on_service', 'name_id_mapping_service',
'assertion_id_request_service', 'attribute_profile',
'attribute'])
@@ -1069,6 +1071,7 @@ class IDPSSODescriptorType_(SSODescriptorType_):
text=None,
extension_elements=None,
extension_attributes=None,
want_authn_requests_only_with_valid_cert=None,
):
SSODescriptorType_.__init__(self,
artifact_resolution_service=artifact_resolution_service,
@@ -1095,6 +1098,7 @@ class IDPSSODescriptorType_(SSODescriptorType_):
self.attribute_profile = attribute_profile or []
self.attribute = attribute or []
self.want_authn_requests_signed = want_authn_requests_signed
self.want_authn_requests_only_with_valid_cert = want_authn_requests_only_with_valid_cert
def idpsso_descriptor_type__from_string(xml_string):
@@ -2012,3 +2016,5 @@ ELEMENT_BY_TAG = {
def factory(tag, **kwargs):
return ELEMENT_BY_TAG[tag](**kwargs)

12
src/saml2/metadata.py
View File

@@ -37,6 +37,7 @@ DEFAULTS = {
"want_assertions_signed": "true",
"authn_requests_signed": "false",
"want_authn_requests_signed": "true",
"want_authn_requests_only_with_valid_cert": "false",
}
ORG_ATTR_TRANSL = {
@@ -407,6 +408,7 @@ DEFAULT = {
"want_assertions_signed": "true",
"authn_requests_signed": "false",
"want_authn_requests_signed": "false",
"want_authn_requests_only_with_valid_cert": "false",
}
@@ -527,6 +529,16 @@ def do_idpsso_descriptor(conf, cert=None):
except KeyError:
setattr(idpsso, key, DEFAULTS[key])
for key in ["want_authn_requests_only_with_valid_cert"]:
try:
val = conf.getattr(key, "idp")
if val is None:
setattr(idpsso, key, DEFAULT["want_authn_requests_only_with_valid_cert"])
else:
setattr(idpsso, key, "%s" % val)
except KeyError:
setattr(idpsso, key, DEFAULTS[key])
return idpsso

8
src/saml2/request.py
View File

@@ -36,12 +36,12 @@ class Request(object):
self.message = None
self.not_on_or_after = 0
def _loads(self, xmldata, binding=None, origdoc=None, must=None):
def _loads(self, xmldata, binding=None, origdoc=None, must=None, only_valid_cert=False):
# own copy
self.xmlstr = xmldata[:]
logger.info("xmlstr: %s" % (self.xmlstr,))
try:
self.message = self.signature_check(xmldata, origdoc=origdoc, must=must)
self.message = self.signature_check(xmldata, origdoc=origdoc, must=must, only_valid_cert=only_valid_cert)
except TypeError:
raise
except Exception, excp:
@@ -84,8 +84,8 @@ class Request(object):
assert self.issue_instant_ok()
return self
def loads(self, xmldata, binding, origdoc=None, must=None):
return self._loads(xmldata, binding, origdoc, must)
def loads(self, xmldata, binding, origdoc=None, must=None, only_valid_cert=False):
return self._loads(xmldata, binding, origdoc, must, only_valid_cert=only_valid_cert)
def verify(self):
try:

1
src/saml2/response.py
View File

@@ -553,6 +553,7 @@ class AuthnResponse(StatusResponse):
else:
self.not_on_or_after = 0
if not self.allow_unsolicited:
if not for_me(conditions, self.entity_id):
if not lax:
raise Exception("Not for me!!!")

22
src/saml2/sigver.py
View File

@@ -1006,9 +1006,13 @@ class CertHandler(object):
def generate_cert(self):
return self._generate_cert
def update_cert(self, active=False):
if self._generate_cert and active:
if self._cert_handler_extra_class is not None and self._cert_handler_extra_class.use_generate_cert_func():
def update_cert(self, active=False, client_crt=None):
if (self._generate_cert and active) or client_crt is not None:
if client_crt is not None:
self._tmp_cert_str = client_crt
#No private key for signing
self._tmp_key_str = ""
elif self._cert_handler_extra_class is not None and self._cert_handler_extra_class.use_generate_cert_func():
(self._tmp_cert_str, self._tmp_key_str) = \
self._cert_handler_extra_class.generate_cert(self._cert_info, self._cert_str, self._key_str)
else:
@@ -1127,7 +1131,7 @@ class SecurityContext(object):
)
def _check_signature(self, decoded_xml, item, node_name=NODE_NAME,
origdoc=None, id_attr="", must=False):
origdoc=None, id_attr="", must=False, only_valid_cert=False):
#print item
try:
issuer = item.issuer.text.strip()
@@ -1196,7 +1200,7 @@ class SecurityContext(object):
logger.error("check_sig: %s" % exc)
raise
if not verified:
if (not verified) and (not only_valid_cert):
raise SignatureError("Failed to verify signature")
else:
if not self.cert_handler.verify_cert(last_pem_file):
@@ -1211,7 +1215,7 @@ class SecurityContext(object):
id_attr=id_attr, must=must)
def correctly_signed_message(self, decoded_xml, msgtype, must=False,
origdoc=None):
origdoc=None, only_valid_cert=False):
"""Check if a request is correctly signed, if we have metadata for
the entity that sent the info use that, if not use the key that are in
the message if any.
@@ -1239,12 +1243,12 @@ class SecurityContext(object):
return msg
return self._check_signature(decoded_xml, msg, class_name(msg),
origdoc, must=must)
origdoc, must=must, only_valid_cert=only_valid_cert)
def correctly_signed_authn_request(self, decoded_xml, must=False,
origdoc=None):
origdoc=None, only_valid_cert=False):
return self.correctly_signed_message(decoded_xml, "authn_request",
must, origdoc)
must, origdoc, only_valid_cert=only_valid_cert)
def correctly_signed_authn_query(self, decoded_xml, must=False,
origdoc=None):