openstack/deb-python-pysaml2
Code Issues Proposed changes

Show AuthnStatement together with all attributes on result page.

Browse Source
This commit is contained in:
Rebecka Gulliksson
2016-01-15 12:20:15 +01:00
parent 62d12bd346
commit ecaee0a93b
1 changed files with 23 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
example/sp-wsgi/sp.py
View File

@@ -2,11 +2,13 @@
from __future__ import print_function from __future__ import print_function
import argparse import argparse
import cgi
import importlib import importlib
import logging import logging
import os import os
import re import re
import sys import sys
import xml.dom.minidom
import six import six
from six.moves.http_cookies import SimpleCookie from six.moves.http_cookies import SimpleCookie
@@ -46,7 +48,7 @@ from saml2.samlp import Extensions
logger = logging.getLogger("") logger = logging.getLogger("")
hdlr = logging.FileHandler('spx.log') hdlr = logging.FileHandler('spx.log')
base_formatter = logging.Formatter( base_formatter = logging.Formatter(
"%(asctime)s %(name)s:%(levelname)s %(message)s") "%(asctime)s %(name)s:%(levelname)s %(message)s")
hdlr.setFormatter(base_formatter) hdlr.setFormatter(base_formatter)
logger.addHandler(hdlr) logger.addHandler(hdlr)
@@ -329,9 +331,15 @@ class Service(object):
class User(object): class User(object):
def __init__(self, name_id, data): def __init__(self, name_id, data, saml_response):
self.name_id = name_id self.name_id = name_id
self.data = data self.data = data
self.response = saml_response
@property
def authn_statement(self):
xml_doc = xml.dom.minidom.parseString(str(self.response.assertion.authn_statement[0]))
return xml_doc.toprettyxml()
class ACS(Service): class ACS(Service):
@@ -356,7 +364,7 @@ class ACS(Service):
try: try:
self.response = self.sp.parse_authn_request_response( self.response = self.sp.parse_authn_request_response(
response, binding, self.outstanding_queries, self.cache.outstanding_certs) response, binding, self.outstanding_queries, self.cache.outstanding_certs)
except UnknownPrincipal as excp: except UnknownPrincipal as excp:
logger.error("UnknownPrincipal: %s", excp) logger.error("UnknownPrincipal: %s", excp)
resp = ServiceError("UnknownPrincipal: %s" % (excp,)) resp = ServiceError("UnknownPrincipal: %s" % (excp,))
@@ -374,7 +382,7 @@ class ACS(Service):
logger.info("AVA: %s", self.response.ava) logger.info("AVA: %s", self.response.ava)
user = User(self.response.name_id, self.response.ava) user = User(self.response.name_id, self.response.ava, self.response)
cookie = self.cache.set_cookie(user) cookie = self.cache.set_cookie(user)
resp = Redirect("/", headers=[ resp = Redirect("/", headers=[
@@ -385,7 +393,7 @@ class ACS(Service):
def verify_attributes(self, ava): def verify_attributes(self, ava):
logger.info("SP: %s", self.sp.config.entityid) logger.info("SP: %s", self.sp.config.entityid)
rest = POLICY.get_entity_categories( rest = POLICY.get_entity_categories(
self.sp.config.entityid, self.sp.metadata) self.sp.config.entityid, self.sp.metadata)
akeys = [k.lower() for k in ava.keys()] akeys = [k.lower() for k in ava.keys()]
@@ -470,7 +478,7 @@ class SSO(object):
_rstate = rndstr() _rstate = rndstr()
self.cache.relay_state[_rstate] = geturl(self.environ) self.cache.relay_state[_rstate] = geturl(self.environ)
_entityid = _cli.config.ecp_endpoint( _entityid = _cli.config.ecp_endpoint(
self.environ["REMOTE_ADDR"]) self.environ["REMOTE_ADDR"])
if not _entityid: if not _entityid:
return -1, ServiceError("No IdP to talk to") return -1, ServiceError("No IdP to talk to")
@@ -522,7 +530,7 @@ class SSO(object):
elif self.discosrv: elif self.discosrv:
if query: if query:
idp_entity_id = _cli.parse_discovery_service_response( idp_entity_id = _cli.parse_discovery_service_response(
query=self.environ.get("QUERY_STRING")) query=self.environ.get("QUERY_STRING"))
if not idp_entity_id: if not idp_entity_id:
sid_ = sid() sid_ = sid()
self.cache.outstanding_queries[sid_] = came_from self.cache.outstanding_queries[sid_] = came_from
@@ -532,7 +540,7 @@ class SSO(object):
"sp")["discovery_response"][0][0] "sp")["discovery_response"][0][0]
ret += "?sid=%s" % sid_ ret += "?sid=%s" % sid_
loc = _cli.create_discovery_service_request( loc = _cli.create_discovery_service_request(
self.discosrv, eid, **{"return": ret}) self.discosrv, eid, **{"return": ret})
return -1, SeeOther(loc) return -1, SeeOther(loc)
elif len(idps) == 1: elif len(idps) == 1:
# idps is a dictionary # idps is a dictionary
@@ -549,8 +557,8 @@ class SSO(object):
try: try:
# Picks a binding to use for sending the Request to the IDP # Picks a binding to use for sending the Request to the IDP
_binding, destination = _cli.pick_binding( _binding, destination = _cli.pick_binding(
"single_sign_on_service", self.bindings, "idpsso", "single_sign_on_service", self.bindings, "idpsso",
entity_id=entity_id) entity_id=entity_id)
logger.debug("binding: %s, destination: %s", _binding, logger.debug("binding: %s, destination: %s", _binding,
destination) destination)
# Binding here is the response binding that is which binding the # Binding here is the response binding that is which binding the
@@ -569,7 +577,7 @@ class SSO(object):
"key": req_key_str "key": req_key_str
} }
spcertenc = SPCertEnc(x509_data=ds.X509Data( spcertenc = SPCertEnc(x509_data=ds.X509Data(
x509_certificate=ds.X509Certificate(text=cert_str))) x509_certificate=ds.X509Certificate(text=cert_str)))
extensions = Extensions(extension_elements=[ extensions = Extensions(extension_elements=[
element_to_extension_element(spcertenc)]) element_to_extension_element(spcertenc)])
@@ -590,7 +598,7 @@ class SSO(object):
except Exception as exc: except Exception as exc:
logger.exception(exc) logger.exception(exc)
resp = ServiceError( resp = ServiceError(
"Failed to construct the AuthnRequest: %s" % exc) "Failed to construct the AuthnRequest: %s" % exc)
return resp return resp
# remember the request # remember the request
@@ -668,7 +676,9 @@ def main(environ, start_response, sp):
return sso.do() return sso.do()
body = dict_to_table(user.data) body = dict_to_table(user.data)
body += '<br><a href="/logout">logout</a>' authn_stmt = cgi.escape(user.authn_statement)
body.append('<br><pre>' + authn_stmt + "</pre>")
body.append('<br><a href="/logout">logout</a>')
resp = Response(body) resp = Response(body)
return resp(environ, start_response) return resp(environ, start_response)