openstack/deb-python-pysaml2
Code Issues Proposed changes

More informative variable names.

Browse Source
This commit is contained in:
Roland Hedberg
2014-05-13 10:42:12 +02:00
parent a6ef5141a9
commit 6ff7f3755c
2 changed files with 55 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files

90
example/idp2/idp.py
View File

@@ -121,30 +121,32 @@ class Service(object):
logger.debug("_dict: %s" % _dict) logger.debug("_dict: %s" % _dict)
return _dict return _dict
def operation(self, _dict, binding): def operation(self, saml_msg, binding):
logger.debug("_operation: %s" % _dict) logger.debug("_operation: %s" % saml_msg)
if not _dict or not 'SAMLRequest' in _dict: if not saml_msg or not 'SAMLRequest' in saml_msg:
resp = BadRequest('Error parsing request or no request') resp = BadRequest('Error parsing request or no request')
return resp(self.environ, self.start_response) return resp(self.environ, self.start_response)
else: else:
try: try:
_encrypt_cert = encrypt_cert_from_item(_dict["req_info"].message) _encrypt_cert = encrypt_cert_from_item(
return self.do(_dict["SAMLRequest"], binding, saml_msg["req_info"].message)
_dict["RelayState"], encrypt_cert=_encrypt_cert) return self.do(saml_msg["SAMLRequest"], binding,
saml_msg["RelayState"],
encrypt_cert=_encrypt_cert)
except KeyError: except KeyError:
# Can live with no relay state # Can live with no relay state
return self.do(_dict["SAMLRequest"], binding) return self.do(saml_msg["SAMLRequest"], binding)
def artifact_operation(self, _dict): def artifact_operation(self, saml_msg):
if not _dict: if not saml_msg:
resp = BadRequest("Missing query") resp = BadRequest("Missing query")
return resp(self.environ, self.start_response) return resp(self.environ, self.start_response)
else: else:
# exchange artifact for request # exchange artifact for request
request = IDP.artifact2message(_dict["SAMLart"], "spsso") request = IDP.artifact2message(saml_msg["SAMLart"], "spsso")
try: try:
return self.do(request, BINDING_HTTP_ARTIFACT, return self.do(request, BINDING_HTTP_ARTIFACT,
_dict["RelayState"]) saml_msg["RelayState"])
except KeyError: except KeyError:
return self.do(request, BINDING_HTTP_ARTIFACT) return self.do(request, BINDING_HTTP_ARTIFACT)
@@ -188,24 +190,6 @@ class Service(object):
_dict = self.unpack_either() _dict = self.unpack_either()
return self.operation(_dict, BINDING_SOAP) return self.operation(_dict, BINDING_SOAP)
# def not_authn(self, key):
# """
#
#
# :return:
# """
# loc = "http://%s/login" % (self.environ["HTTP_HOST"])
# loc += "?%s" % urllib.urlencode({"came_from": self.environ[
# "PATH_INFO"], "key": key})
# headers = [('Content-Type', 'text/plain')]
#
# logger.debug("location: %s" % loc)
# logger.debug("headers: %s" % headers)
#
# resp = Redirect(loc, headers=headers)
#
# return resp(self.environ, self.start_response)
def not_authn(self, key, requested_authn_context): def not_authn(self, key, requested_authn_context):
ruri = geturl(self.environ, query=False) ruri = geturl(self.environ, query=False)
return do_authentication(self.environ, self.start_response, return do_authentication(self.environ, self.start_response,
@@ -345,27 +329,27 @@ class SSO(Service):
logger.debug("HTTPargs: %s" % http_args) logger.debug("HTTPargs: %s" % http_args)
return self.response(self.binding_out, http_args) return self.response(self.binding_out, http_args)
def _store_request(self, _dict): def _store_request(self, saml_msg):
logger.debug("_store_request: %s" % _dict) logger.debug("_store_request: %s" % saml_msg)
key = sha1(_dict["SAMLRequest"]).hexdigest() key = sha1(saml_msg["SAMLRequest"]).hexdigest()
# store the AuthnRequest # store the AuthnRequest
IDP.ticket[key] = _dict IDP.ticket[key] = saml_msg
return key return key
def redirect(self): def redirect(self):
""" This is the HTTP-redirect endpoint """ """ This is the HTTP-redirect endpoint """
logger.info("--- In SSO Redirect ---") logger.info("--- In SSO Redirect ---")
_info = self.unpack_redirect() saml_msg = self.unpack_redirect()
try: try:
_key = _info["key"] _key = saml_msg["key"]
_info = IDP.ticket[_key] saml_msg = IDP.ticket[_key]
self.req_info = _info["req_info"] self.req_info = saml_msg["req_info"]
del IDP.ticket[_key] del IDP.ticket[_key]
except KeyError: except KeyError:
try: try:
self.req_info = IDP.parse_authn_request(_info["SAMLRequest"], self.req_info = IDP.parse_authn_request(saml_msg["SAMLRequest"],
BINDING_HTTP_REDIRECT) BINDING_HTTP_REDIRECT)
except KeyError: except KeyError:
resp = BadRequest("Message signature verification failure") resp = BadRequest("Message signature verification failure")
@@ -373,12 +357,12 @@ class SSO(Service):
_req = self.req_info.message _req = self.req_info.message
if "SigAlg" in _info and "Signature" in _info: # Signed request if "SigAlg" in saml_msg and "Signature" in saml_msg: # Signed request
issuer = _req.issuer.text issuer = _req.issuer.text
_certs = IDP.metadata.certs(issuer, "any", "signing") _certs = IDP.metadata.certs(issuer, "any", "signing")
verified_ok = False verified_ok = False
for cert in _certs: for cert in _certs:
if verify_redirect_signature(_info, cert): if verify_redirect_signature(saml_msg, cert):
verified_ok = True verified_ok = True
break break
if not verified_ok: if not verified_ok:
@@ -387,37 +371,37 @@ class SSO(Service):
if self.user: if self.user:
if _req.force_authn: if _req.force_authn:
_info["req_info"] = self.req_info saml_msg["req_info"] = self.req_info
key = self._store_request(_info) key = self._store_request(saml_msg)
return self.not_authn(key, _req.requested_authn_context) return self.not_authn(key, _req.requested_authn_context)
else: else:
return self.operation(_info, BINDING_HTTP_REDIRECT) return self.operation(saml_msg, BINDING_HTTP_REDIRECT)
else: else:
_info["req_info"] = self.req_info saml_msg["req_info"] = self.req_info
key = self._store_request(_info) key = self._store_request(saml_msg)
return self.not_authn(key, _req.requested_authn_context) return self.not_authn(key, _req.requested_authn_context)
else: else:
return self.operation(_info, BINDING_HTTP_REDIRECT) return self.operation(saml_msg, BINDING_HTTP_REDIRECT)
def post(self): def post(self):
""" """
The HTTP-Post endpoint The HTTP-Post endpoint
""" """
logger.info("--- In SSO POST ---") logger.info("--- In SSO POST ---")
_info = self.unpack_either() saml_msg = self.unpack_either()
self.req_info = IDP.parse_authn_request( self.req_info = IDP.parse_authn_request(
_info["SAMLRequest"], BINDING_HTTP_POST) saml_msg["SAMLRequest"], BINDING_HTTP_POST)
_req = self.req_info.message _req = self.req_info.message
if self.user: if self.user:
if _req.force_authn: if _req.force_authn:
_info["req_info"] = self.req_info saml_msg["req_info"] = self.req_info
key = self._store_request(_info) key = self._store_request(saml_msg)
return self.not_authn(key, _req.requested_authn_context) return self.not_authn(key, _req.requested_authn_context)
else: else:
return self.operation(_info, BINDING_HTTP_POST) return self.operation(saml_msg, BINDING_HTTP_POST)
else: else:
_info["req_info"] = self.req_info saml_msg["req_info"] = self.req_info
key = self._store_request(_info) key = self._store_request(saml_msg)
return self.not_authn(key, _req.requested_authn_context) return self.not_authn(key, _req.requested_authn_context)
# def artifact(self): # def artifact(self):

30
src/saml2/sigver.py
View File

@@ -610,35 +610,35 @@ REQ_ORDER = ["SAMLRequest", "RelayState", "SigAlg"]
RESP_ORDER = ["SAMLResponse", "RelayState", "SigAlg"] RESP_ORDER = ["SAMLResponse", "RelayState", "SigAlg"]
def verify_redirect_signature(info, cert): def verify_redirect_signature(saml_msg, cert):
""" """
:param info: A dictionary as produced by parse_qs, means all values are :param saml_msg: A dictionary as produced by parse_qs, means all values are
lists. lists.
:param cert: A certificate to use when verifying the signature :param cert: A certificate to use when verifying the signature
:return: True, if signature verified :return: True, if signature verified
""" """
try: try:
signer = SIGNER_ALGS[info["SigAlg"][0]] signer = SIGNER_ALGS[saml_msg["SigAlg"][0]]
except KeyError: except KeyError:
raise Unsupported("Signature algorithm: %s" % info["SigAlg"]) raise Unsupported("Signature algorithm: %s" % saml_msg["SigAlg"])
else: else:
if info["SigAlg"][0] == RSA_SHA1: if saml_msg["SigAlg"][0] == RSA_SHA1:
if "SAMLRequest" in info: if "SAMLRequest" in saml_msg:
_order = REQ_ORDER _order = REQ_ORDER
elif "SAMLResponse" in info: elif "SAMLResponse" in saml_msg:
_order = RESP_ORDER _order = RESP_ORDER
else: else:
raise Unsupported( raise Unsupported(
"Verifying signature on something that should not be " "Verifying signature on something that should not be "
"signed") "signed")
args = info.copy() args = saml_msg.copy()
del args["Signature"] # everything but the signature del args["Signature"] # everything but the signature
string = "&".join( string = "&".join(
[urllib.urlencode({k: args[k][0]}) for k in _order]) [urllib.urlencode({k: args[k][0]}) for k in _order])
_key = extract_rsa_key_from_x509_cert(pem_format(cert)) _key = extract_rsa_key_from_x509_cert(pem_format(cert))
_sign = base64.b64decode(info["Signature"][0]) _sign = base64.b64decode(saml_msg["Signature"][0])
try: try:
signer.verify(string, _sign, _key) signer.verify(string, _sign, _key)
return True return True
@@ -1082,13 +1082,19 @@ class CertHandler(object):
Initiates the class for handling certificates. Enables the certificates Initiates the class for handling certificates. Enables the certificates
to either be a single certificate as base functionality or makes it to either be a single certificate as base functionality or makes it
possible to generate a new certificate for each call to the function. possible to generate a new certificate for each call to the function.
:param key_file:
:param key_type: :param security_context:
:param cert_file: :param cert_file:
:param cert_type: :param cert_type:
:param generate_cert: :param key_file:
:param key_type:
:param generate_cert_info:
:param cert_handler_extra_class: :param cert_handler_extra_class:
:param tmp_cert_file:
:param tmp_key_file:
:param verify_cert:
""" """
self._verify_cert = False self._verify_cert = False
self._generate_cert = False self._generate_cert = False
#This cert do not have to be valid, it is just the last cert to be #This cert do not have to be valid, it is just the last cert to be