<s> import argparse <EOL> import base64 <EOL> import re <EOL> import logging <EOL> import time <EOL> from hashlib import sha1 <EOL> from urlparse import parse_qs <EOL> from Cookie import SimpleCookie <EOL> import os <EOL> from saml2 import server <EOL> from saml2 import BINDING_HTTP_ARTIFACT <EOL> from saml2 import BINDING_URI <EOL> from saml2 import BINDING_PAOS <EOL> from saml2 import BINDING_SOAP <EOL> from saml2 import BINDING_HTTP_REDIRECT <EOL> from saml2 import BINDING_HTTP_POST <EOL> from saml2 import time_util <EOL> from saml2 . authn import is_equal <EOL> from saml2 . authn_context import AuthnBroker <EOL> from saml2 . authn_context import PASSWORD <EOL> from saml2 . authn_context import UNSPECIFIED <EOL> from saml2 . authn_context import authn_context_class_ref <EOL> from saml2 . httputil import Response <EOL> from saml2 . httputil import NotFound <EOL> from saml2 . httputil import geturl <EOL> from saml2 . httputil import get_post <EOL> from saml2 . httputil import Redirect <EOL> from saml2 . httputil import Unauthorized <EOL> from saml2 . httputil import BadRequest <EOL> from saml2 . httputil import ServiceError <EOL> from saml2 . ident import Unknown <EOL> from saml2 . metadata import create_metadata_string <EOL> from saml2 . s_utils import rndstr , exception_trace <EOL> from saml2 . s_utils import UnknownPrincipal <EOL> from saml2 . s_utils import UnsupportedBinding <EOL> from saml2 . s_utils import PolicyError <EOL> from saml2 . sigver import verify_redirect_signature <EOL> logger = logging . getLogger ( "<STR_LIT>" ) <EOL> class Cache ( object ) : <EOL> def __init__ ( self ) : <EOL> self . user2uid = { } <EOL> self . uid2user = { } <EOL> def _expiration ( timeout , tformat = "<STR_LIT>" ) : <EOL> """<STR_LIT>""" <EOL> if timeout == "<STR_LIT>" : <EOL> return time_util . instant ( tformat ) <EOL> elif timeout == "<STR_LIT>" : <EOL> return time . strftime ( tformat , time . gmtime ( <NUM_LIT:0> ) ) <EOL> else : <EOL> return time_util . in_a_while ( minutes = timeout , format = tformat ) <EOL> def get_eptid ( idp , req_info , session ) : <EOL> return idp . eptid . get ( idp . config . entityid , <EOL> req_info . sender ( ) , session [ "<STR_LIT>" ] , <EOL> session [ "<STR_LIT>" ] ) <EOL> def dict2list_of_tuples ( d ) : <EOL> return [ ( k , v ) for k , v in d . items ( ) ] <EOL> class Service ( object ) : <EOL> def __init__ ( self , environ , start_response , user = None ) : <EOL> self . environ = environ <EOL> logger . debug ( "<STR_LIT>" % environ ) <EOL> self . start_response = start_response <EOL> self . user = user <EOL> def unpack_redirect ( self ) : <EOL> if "<STR_LIT>" in self . environ : <EOL> _qs = self . environ [ "<STR_LIT>" ] <EOL> return dict ( [ ( k , v [ <NUM_LIT:0> ] ) for k , v in parse_qs ( _qs ) . items ( ) ] ) <EOL> else : <EOL> return None <EOL> def unpack_post ( self ) : <EOL> _dict = parse_qs ( get_post ( self . environ ) ) <EOL> logger . debug ( "<STR_LIT>" % _dict ) <EOL> try : <EOL> return dict ( [ ( k , v [ <NUM_LIT:0> ] ) for k , v in _dict . items ( ) ] ) <EOL> except Exception : <EOL> return None <EOL> def unpack_soap ( self ) : <EOL> try : <EOL> query = get_post ( self . environ ) <EOL> return { "<STR_LIT>" : query , "<STR_LIT>" : "<STR_LIT>" } <EOL> except Exception : <EOL> return None <EOL> def unpack_either ( self ) : <EOL> if self . environ [ "<STR_LIT>" ] == "<STR_LIT:GET>" : <EOL> _dict = self . unpack_redirect ( ) <EOL> elif self . environ [ "<STR_LIT>" ] == "<STR_LIT:POST>" : <EOL> _dict = self . unpack_post ( ) <EOL> else : <EOL> _dict = None <EOL> logger . debug ( "<STR_LIT>" % _dict ) <EOL> return _dict <EOL> def operation ( self , _dict , binding ) : <EOL> logger . debug ( "<STR_LIT>" % _dict ) <EOL> if not _dict or not '<STR_LIT>' in _dict : <EOL> resp = BadRequest ( '<STR_LIT>' ) <EOL> return resp ( self . environ , self . start_response ) <EOL> else : <EOL> try : <EOL> return self . do ( _dict [ "<STR_LIT>" ] , binding , <EOL> _dict [ "<STR_LIT>" ] ) <EOL> except KeyError : <EOL> return self . do ( _dict [ "<STR_LIT>" ] , binding ) <EOL> def artifact_operation ( self , _dict ) : <EOL> if not _dict : <EOL> resp = BadRequest ( "<STR_LIT>" ) <EOL> return resp ( self . environ , self . start_response ) <EOL> else : <EOL> request = IDP . artifact2message ( _dict [ "<STR_LIT>" ] , "<STR_LIT>" ) <EOL> try : <EOL> return self . do ( request , BINDING_HTTP_ARTIFACT , <EOL> _dict [ "<STR_LIT>" ] ) <EOL> except KeyError : <EOL> return self . do ( request , BINDING_HTTP_ARTIFACT ) <EOL> def response ( self , binding , http_args ) : <EOL> if binding == BINDING_HTTP_ARTIFACT : <EOL> resp = Redirect ( ) <EOL> else : <EOL> resp = Response ( http_args [ "<STR_LIT:data>" ] , headers = http_args [ "<STR_LIT>" ] ) <EOL> return resp ( self . environ , self . start_response ) <EOL> def do ( self , query , binding , relay_state = "<STR_LIT>" ) : <EOL> pass <EOL> def redirect ( self ) : <EOL> """<STR_LIT>""" <EOL> _dict = self . unpack_redirect ( ) <EOL> return self . operation ( _dict , BINDING_HTTP_REDIRECT ) <EOL> def post ( self ) : <EOL> """<STR_LIT>""" <EOL> _dict = self . unpack_post ( ) <EOL> return self . operation ( _dict , BINDING_HTTP_POST ) <EOL> def artifact ( self ) : <EOL> _dict = self . unpack_either ( ) <EOL> return self . artifact_operation ( _dict ) <EOL> def soap ( self ) : <EOL> """<STR_LIT>""" <EOL> logger . debug ( "<STR_LIT>" ) <EOL> _dict = self . unpack_soap ( ) <EOL> logger . debug ( "<STR_LIT>" % _dict ) <EOL> return self . operation ( _dict , BINDING_SOAP ) <EOL> def uri ( self ) : <EOL> _dict = self . unpack_either ( ) <EOL> return self . operation ( _dict , BINDING_SOAP ) <EOL> def not_authn ( self , key , requested_authn_context ) : <EOL> ruri = geturl ( self . environ , query = False ) <EOL> return do_authentication ( self . environ , self . start_response , <EOL> authn_context = requested_authn_context , <EOL> key = key , redirect_uri = ruri ) <EOL> REPOZE_ID_EQUIVALENT = "<STR_LIT>" <EOL> FORM_SPEC = """<STR_LIT>""" <EOL> class AuthenticationNeeded ( Exception ) : <EOL> def __init__ ( self , authn_context = None , * args , ** kwargs ) : <EOL> Exception . __init__ ( * args , ** kwargs ) <EOL> self . authn_context = authn_context <EOL> class SSO ( Service ) : <EOL> def __init__ ( self , environ , start_response , user = None ) : <EOL> Service . __init__ ( self , environ , start_response , user ) <EOL> self . binding = "<STR_LIT>" <EOL> self . response_bindings = None <EOL> self . resp_args = { } <EOL> self . binding_out = None <EOL> self . destination = None <EOL> self . req_info = None <EOL> def verify_request ( self , query , binding ) : <EOL> """<STR_LIT>""" <EOL> resp_args = { } <EOL> if not query : <EOL> logger . info ( "<STR_LIT>" ) <EOL> resp = Unauthorized ( '<STR_LIT>' ) <EOL> return resp_args , resp ( self . environ , self . start_response ) <EOL> if not self . req_info : <EOL> self . req_info = IDP . parse_authn_request ( query , binding ) <EOL> logger . info ( "<STR_LIT>" ) <EOL> _authn_req = self . req_info . message <EOL> logger . debug ( "<STR_LIT:%s>" % _authn_req ) <EOL> self . binding_out , self . destination = IDP . pick_binding ( <EOL> "<STR_LIT>" , <EOL> bindings = self . response_bindings , <EOL> entity_id = _authn_req . issuer . text ) <EOL> logger . debug ( "<STR_LIT>" % ( self . binding_out , <EOL> self . destination ) ) <EOL> resp_args = { } <EOL> try : <EOL> resp_args = IDP . response_args ( _authn_req ) <EOL> _resp = None <EOL> except UnknownPrincipal , excp : <EOL> _resp = IDP . create_error_response ( _authn_req . id , <EOL> self . destination , excp ) <EOL> except UnsupportedBinding , excp : <EOL> _resp = IDP . create_error_response ( _authn_req . id , <EOL> self . destination , excp ) <EOL> return resp_args , _resp <EOL> def do ( self , query , binding_in , relay_state = "<STR_LIT>" ) : <EOL> try : <EOL> resp_args , _resp = self . verify_request ( query , binding_in ) <EOL> except UnknownPrincipal , excp : <EOL> logger . error ( "<STR_LIT>" % ( excp , ) ) <EOL> resp = ServiceError ( "<STR_LIT>" % ( excp , ) ) <EOL> return resp ( self . environ , self . start_response ) <EOL> except UnsupportedBinding , excp : <EOL> logger . error ( "<STR_LIT>" % ( excp , ) ) <EOL> resp = ServiceError ( "<STR_LIT>" % ( excp , ) ) <EOL> return resp ( self . environ , self . start_response ) <EOL> if not _resp : <EOL> identity = USERS [ self . user ] . copy ( ) <EOL> logger . info ( "<STR_LIT>" % ( identity , ) ) <EOL> if REPOZE_ID_EQUIVALENT : <EOL> identity [ REPOZE_ID_EQUIVALENT ] = self . user <EOL> try : <EOL> sign_assertion = IDP . config . getattr ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> if sign_assertion is None : <EOL> sign_assertion = False <EOL> _resp = IDP . create_authn_response ( <EOL> identity , userid = self . user , <EOL> authn = AUTHN_BROKER [ self . environ [ "<STR_LIT>" ] ] , sign_assertion = sign_assertion , <EOL> sign_response = False , ** resp_args ) <EOL> except Exception , excp : <EOL> logging . error ( exception_trace ( excp ) ) <EOL> resp = ServiceError ( "<STR_LIT>" % ( excp , ) ) <EOL> return resp ( self . environ , self . start_response ) <EOL> logger . info ( "<STR_LIT>" % _resp ) <EOL> http_args = IDP . apply_binding ( self . binding_out , <EOL> "<STR_LIT:%s>" % _resp , self . destination , <EOL> relay_state , response = True ) <EOL> logger . debug ( "<STR_LIT>" % http_args ) <EOL> return self . response ( self . binding_out , http_args ) <EOL> def _store_request ( self , _dict ) : <EOL> logger . debug ( "<STR_LIT>" % _dict ) <EOL> key = sha1 ( _dict [ "<STR_LIT>" ] ) . hexdigest ( ) <EOL> IDP . ticket [ key ] = _dict <EOL> return key <EOL> def redirect ( self ) : <EOL> """<STR_LIT>""" <EOL> logger . info ( "<STR_LIT>" ) <EOL> _info = self . unpack_redirect ( ) <EOL> try : <EOL> _key = _info [ "<STR_LIT:key>" ] <EOL> _info = IDP . ticket [ _key ] <EOL> self . req_info = _info [ "<STR_LIT>" ] <EOL> del IDP . ticket [ _key ] <EOL> except KeyError : <EOL> try : <EOL> self . req_info = IDP . parse_authn_request ( _info [ "<STR_LIT>" ] , <EOL> BINDING_HTTP_REDIRECT ) <EOL> except KeyError : <EOL> resp = BadRequest ( "<STR_LIT>" ) <EOL> return resp ( self . environ , self . start_response ) <EOL> _req = self . req_info . message <EOL> if "<STR_LIT>" in _info and "<STR_LIT>" in _info : <EOL> issuer = _req . issuer . text <EOL> _certs = IDP . metadata . certs ( issuer , "<STR_LIT>" , "<STR_LIT>" ) <EOL> verified_ok = False <EOL> for cert in _certs : <EOL> if verify_redirect_signature ( _info , cert ) : <EOL> verified_ok = True <EOL> break <EOL> if not verified_ok : <EOL> resp = BadRequest ( "<STR_LIT>" ) <EOL> return resp ( self . environ , self . start_response ) <EOL> if self . user : <EOL> if _req . force_authn : <EOL> _info [ "<STR_LIT>" ] = self . req_info <EOL> key = self . _store_request ( _info ) <EOL> return self . not_authn ( key , _req . requested_authn_context ) <EOL> else : <EOL> return self . operation ( _info , BINDING_HTTP_REDIRECT ) <EOL> else : <EOL> _info [ "<STR_LIT>" ] = self . req_info <EOL> key = self . _store_request ( _info ) <EOL> return self . not_authn ( key , _req . requested_authn_context ) <EOL> else : <EOL> return self . operation ( _info , BINDING_HTTP_REDIRECT ) <EOL> def post ( self ) : <EOL> """<STR_LIT>""" <EOL> logger . info ( "<STR_LIT>" ) <EOL> _info = self . unpack_either ( ) <EOL> self . req_info = IDP . parse_authn_request ( <EOL> _info [ "<STR_LIT>" ] , BINDING_HTTP_POST ) <EOL> _req = self . req_info . message <EOL> if self . user : <EOL> if _req . force_authn : <EOL> _info [ "<STR_LIT>" ] = self . req_info <EOL> key = self . _store_request ( _info ) <EOL> return self . not_authn ( key , _req . requested_authn_context ) <EOL> else : <EOL> return self . operation ( _info , BINDING_HTTP_POST ) <EOL> else : <EOL> _info [ "<STR_LIT>" ] = self . req_info <EOL> key = self . _store_request ( _info ) <EOL> return self . not_authn ( key , _req . requested_authn_context ) <EOL> def ecp ( self ) : <EOL> logger . info ( "<STR_LIT>" ) <EOL> resp = None <EOL> try : <EOL> authz_info = self . environ [ "<STR_LIT>" ] <EOL> if authz_info . startswith ( "<STR_LIT>" ) : <EOL> try : <EOL> _info = base64 . b64decode ( authz_info [ <NUM_LIT:6> : ] ) <EOL> except TypeError : <EOL> resp = Unauthorized ( ) <EOL> else : <EOL> logger . debug ( "<STR_LIT>" % _info ) <EOL> try : <EOL> ( user , passwd ) = _info . split ( "<STR_LIT::>" ) <EOL> if is_equal ( PASSWD [ user ] , passwd ) : <EOL> resp = Unauthorized ( ) <EOL> self . user = user <EOL> except ( ValueError , TypeError ) : <EOL> resp = Unauthorized ( ) <EOL> else : <EOL> resp = Unauthorized ( ) <EOL> except KeyError : <EOL> resp = Unauthorized ( ) <EOL> if resp : <EOL> return resp ( self . environ , self . start_response ) <EOL> _dict = self . unpack_soap ( ) <EOL> self . response_bindings = [ BINDING_PAOS ] <EOL> return self . operation ( _dict , BINDING_SOAP ) <EOL> def do_authentication ( environ , start_response , authn_context , key , <EOL> redirect_uri ) : <EOL> """<STR_LIT>""" <EOL> logger . debug ( "<STR_LIT>" ) <EOL> auth_info = AUTHN_BROKER . pick ( authn_context ) <EOL> if len ( auth_info ) : <EOL> method , reference = auth_info [ <NUM_LIT:0> ] <EOL> logger . debug ( "<STR_LIT>" % ( method , reference ) ) <EOL> return method ( environ , start_response , reference , key , redirect_uri ) <EOL> else : <EOL> resp = Unauthorized ( "<STR_LIT>" ) <EOL> return resp ( environ , start_response ) <EOL> PASSWD = { "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" } <EOL> def username_password_authn ( environ , start_response , reference , key , <EOL> redirect_uri ) : <EOL> """<STR_LIT>""" <EOL> logger . info ( "<STR_LIT>" ) <EOL> headers = [ ] <EOL> resp = Response ( mako_template = "<STR_LIT>" , template_lookup = LOOKUP , <EOL> headers = headers ) <EOL> argv = { <EOL> "<STR_LIT:action>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT:password>" : "<STR_LIT>" , <EOL> "<STR_LIT:key>" : key , <EOL> "<STR_LIT>" : reference , <EOL> "<STR_LIT>" : redirect_uri <EOL> } <EOL> logger . info ( "<STR_LIT>" % argv ) <EOL> return resp ( environ , start_response , ** argv ) <EOL> def verify_username_and_password ( dic ) : <EOL> global PASSWD <EOL> if PASSWD [ dic [ "<STR_LIT>" ] [ <NUM_LIT:0> ] ] == dic [ "<STR_LIT:password>" ] [ <NUM_LIT:0> ] : <EOL> return True , dic [ "<STR_LIT>" ] [ <NUM_LIT:0> ] <EOL> else : <EOL> return False , "<STR_LIT>" <EOL> def do_verify ( environ , start_response , _ ) : <EOL> query = parse_qs ( get_post ( environ ) ) <EOL> logger . debug ( "<STR_LIT>" % query ) <EOL> try : <EOL> _ok , user = verify_username_and_password ( query ) <EOL> except KeyError : <EOL> _ok = False <EOL> user = None <EOL> if not _ok : <EOL> resp = Unauthorized ( "<STR_LIT>" ) <EOL> else : <EOL> uid = rndstr ( <NUM_LIT> ) <EOL> IDP . cache . uid2user [ uid ] = user <EOL> IDP . cache . user2uid [ user ] = uid <EOL> logger . debug ( "<STR_LIT>" % ( user , uid ) ) <EOL> kaka = set_cookie ( "<STR_LIT>" , "<STR_LIT:/>" , uid , query [ "<STR_LIT>" ] [ <NUM_LIT:0> ] ) <EOL> lox = "<STR_LIT>" % ( query [ "<STR_LIT>" ] [ <NUM_LIT:0> ] , uid , <EOL> query [ "<STR_LIT:key>" ] [ <NUM_LIT:0> ] ) <EOL> logger . debug ( "<STR_LIT>" % lox ) <EOL> resp = Redirect ( lox , headers = [ kaka ] , content = "<STR_LIT>" ) <EOL> return resp ( environ , start_response ) <EOL> def not_found ( environ , start_response ) : <EOL> """<STR_LIT>""" <EOL> resp = NotFound ( ) <EOL> return resp ( environ , start_response ) <EOL> class SLO ( Service ) : <EOL> def do ( self , request , binding , relay_state = "<STR_LIT>" ) : <EOL> logger . info ( "<STR_LIT>" ) <EOL> try : <EOL> _ , body = request . split ( "<STR_LIT:\n>" ) <EOL> logger . debug ( "<STR_LIT>" % body ) <EOL> req_info = IDP . parse_logout_request ( body , binding ) <EOL> except Exception , exc : <EOL> logger . error ( "<STR_LIT>" % exc ) <EOL> resp = BadRequest ( "<STR_LIT:%s>" % exc ) <EOL> return resp ( self . environ , self . start_response ) <EOL> msg = req_info . message <EOL> if msg . name_id : <EOL> lid = IDP . ident . find_local_id ( msg . name_id ) <EOL> logger . info ( "<STR_LIT>" % lid ) <EOL> if lid in IDP . cache . user2uid : <EOL> uid = IDP . cache . user2uid [ lid ] <EOL> if uid in IDP . cache . uid2user : <EOL> del IDP . cache . uid2user [ uid ] <EOL> del IDP . cache . user2uid [ lid ] <EOL> try : <EOL> IDP . session_db . remove_authn_statements ( msg . name_id ) <EOL> except KeyError , exc : <EOL> logger . error ( "<STR_LIT>" % exc ) <EOL> resp = ServiceError ( "<STR_LIT:%s>" % exc ) <EOL> return resp ( self . environ , self . start_response ) <EOL> resp = IDP . create_logout_response ( msg , [ binding ] ) <EOL> try : <EOL> hinfo = IDP . apply_binding ( binding , "<STR_LIT:%s>" % resp , "<STR_LIT>" , relay_state ) <EOL> except Exception , exc : <EOL> logger . error ( "<STR_LIT>" % exc ) <EOL> resp = ServiceError ( "<STR_LIT:%s>" % exc ) <EOL> return resp ( self . environ , self . start_response ) <EOL> delco = delete_cookie ( self . environ , "<STR_LIT>" ) <EOL> if delco : <EOL> hinfo [ "<STR_LIT>" ] . append ( delco ) <EOL> logger . info ( "<STR_LIT>" % ( hinfo [ "<STR_LIT>" ] , ) ) <EOL> resp = Response ( hinfo [ "<STR_LIT:data>" ] , headers = hinfo [ "<STR_LIT>" ] ) <EOL> return resp ( self . environ , self . start_response ) <EOL> class NMI ( Service ) : <EOL> def do ( self , query , binding , relay_state = "<STR_LIT>" ) : <EOL> logger . info ( "<STR_LIT>" ) <EOL> req = IDP . parse_manage_name_id_request ( query , binding ) <EOL> request = req . message <EOL> name_id = IDP . ident . handle_manage_name_id_request ( <EOL> request . name_id , request . new_id , request . new_encrypted_id , <EOL> request . terminate ) <EOL> logger . debug ( "<STR_LIT>" % name_id ) <EOL> _resp = IDP . create_manage_name_id_response ( request ) <EOL> hinfo = IDP . apply_binding ( BINDING_SOAP , "<STR_LIT:%s>" % _resp , "<STR_LIT>" , <EOL> relay_state , response = True ) <EOL> resp = Response ( hinfo [ "<STR_LIT:data>" ] , headers = hinfo [ "<STR_LIT>" ] ) <EOL> return resp ( self . environ , self . start_response ) <EOL> class AIDR ( Service ) : <EOL> def do ( self , aid , binding , relay_state = "<STR_LIT>" ) : <EOL> logger . info ( "<STR_LIT>" ) <EOL> try : <EOL> assertion = IDP . create_assertion_id_request_response ( aid ) <EOL> except Unknown : <EOL> resp = NotFound ( aid ) <EOL> return resp ( self . environ , self . start_response ) <EOL> hinfo = IDP . apply_binding ( BINDING_URI , "<STR_LIT:%s>" % assertion , response = True ) <EOL> logger . debug ( "<STR_LIT>" % hinfo ) <EOL> resp = Response ( hinfo [ "<STR_LIT:data>" ] , headers = hinfo [ "<STR_LIT>" ] ) <EOL> return resp ( self . environ , self . start_response ) <EOL> def operation ( self , _dict , binding , ** kwargs ) : <EOL> logger . debug ( "<STR_LIT>" % _dict ) <EOL> if not _dict or "<STR_LIT>" not in _dict : <EOL> resp = BadRequest ( '<STR_LIT>' ) <EOL> return resp ( self . environ , self . start_response ) <EOL> return self . do ( _dict [ "<STR_LIT>" ] , binding , ** kwargs ) <EOL> class ARS ( Service ) : <EOL> def do ( self , request , binding , relay_state = "<STR_LIT>" ) : <EOL> _req = IDP . parse_artifact_resolve ( request , binding ) <EOL> msg = IDP . create_artifact_response ( _req , _req . artifact . text ) <EOL> hinfo = IDP . apply_binding ( BINDING_SOAP , "<STR_LIT:%s>" % msg , "<STR_LIT>" , "<STR_LIT>" , <EOL> response = True ) <EOL> resp = Response ( hinfo [ "<STR_LIT:data>" ] , headers = hinfo [ "<STR_LIT>" ] ) <EOL> return resp ( self . environ , self . start_response ) <EOL> class AQS ( Service ) : <EOL> def do ( self , request , binding , relay_state = "<STR_LIT>" ) : <EOL> logger . info ( "<STR_LIT>" ) <EOL> _req = IDP . parse_authn_query ( request , binding ) <EOL> _query = _req . message <EOL> msg = IDP . create_authn_query_response ( _query . subject , <EOL> _query . requested_authn_context , <EOL> _query . session_index ) <EOL> logger . debug ( "<STR_LIT>" % msg ) <EOL> hinfo = IDP . apply_binding ( BINDING_SOAP , "<STR_LIT:%s>" % msg , "<STR_LIT>" , "<STR_LIT>" , <EOL> response = True ) <EOL> resp = Response ( hinfo [ "<STR_LIT:data>" ] , headers = hinfo [ "<STR_LIT>" ] ) <EOL> return resp ( self . environ , self . start_response ) <EOL> class ATTR ( Service ) : <EOL> def do ( self , request , binding , relay_state = "<STR_LIT>" ) : <EOL> logger . info ( "<STR_LIT>" ) <EOL> _req = IDP . parse_attribute_query ( request , binding ) <EOL> _query = _req . message <EOL> name_id = _query . subject . name_id <EOL> uid = name_id . text <EOL> logger . debug ( "<STR_LIT>" % uid ) <EOL> identity = EXTRA [ uid ] <EOL> args = IDP . response_args ( _query , [ BINDING_SOAP ] ) <EOL> msg = IDP . create_attribute_response ( identity , <EOL> name_id = name_id , ** args ) <EOL> logger . debug ( "<STR_LIT>" % msg ) <EOL> hinfo = IDP . apply_binding ( BINDING_SOAP , "<STR_LIT:%s>" % msg , "<STR_LIT>" , "<STR_LIT>" , <EOL> response = True ) <EOL> resp = Response ( hinfo [ "<STR_LIT:data>" ] , headers = hinfo [ "<STR_LIT>" ] ) <EOL> return resp ( self . environ , self . start_response ) <EOL> class NIM ( Service ) : <EOL> def do ( self , query , binding , relay_state = "<STR_LIT>" ) : <EOL> req = IDP . parse_name_id_mapping_request ( query , binding ) <EOL> request = req . message <EOL> try : <EOL> name_id = IDP . ident . handle_name_id_mapping_request ( <EOL> request . name_id , request . name_id_policy ) <EOL> except Unknown : <EOL> resp = BadRequest ( "<STR_LIT>" ) <EOL> return resp ( self . environ , self . start_response ) <EOL> except PolicyError : <EOL> resp = BadRequest ( "<STR_LIT>" ) <EOL> return resp ( self . environ , self . start_response ) <EOL> info = IDP . response_args ( request ) <EOL> _resp = IDP . create_name_id_mapping_response ( name_id , ** info ) <EOL> hinfo = IDP . apply_binding ( BINDING_SOAP , "<STR_LIT:%s>" % _resp , "<STR_LIT>" , "<STR_LIT>" , <EOL> response = True ) <EOL> resp = Response ( hinfo [ "<STR_LIT:data>" ] , headers = hinfo [ "<STR_LIT>" ] ) <EOL> return resp ( self . environ , self . start_response ) <EOL> def info_from_cookie ( kaka ) : <EOL> logger . debug ( "<STR_LIT>" % kaka ) <EOL> if kaka : <EOL> cookie_obj = SimpleCookie ( kaka ) <EOL> morsel = cookie_obj . get ( "<STR_LIT>" , None ) <EOL> if morsel : <EOL> try : <EOL> key , ref = base64 . b64decode ( morsel . value ) . split ( "<STR_LIT::>" ) <EOL> return IDP . cache . uid2user [ key ] , ref <EOL> except ( KeyError , TypeError ) : <EOL> return None , None <EOL> else : <EOL> logger . debug ( "<STR_LIT>" ) <EOL> return None , None <EOL> def delete_cookie ( environ , name ) : <EOL> kaka = environ . get ( "<STR_LIT>" , '<STR_LIT>' ) <EOL> logger . debug ( "<STR_LIT>" % kaka ) <EOL> if kaka : <EOL> cookie_obj = SimpleCookie ( kaka ) <EOL> morsel = cookie_obj . get ( name , None ) <EOL> cookie = SimpleCookie ( ) <EOL> cookie [ name ] = "<STR_LIT>" <EOL> cookie [ name ] [ '<STR_LIT:path>' ] = "<STR_LIT:/>" <EOL> logger . debug ( "<STR_LIT>" % morsel ) <EOL> cookie [ name ] [ "<STR_LIT>" ] = _expiration ( "<STR_LIT>" ) <EOL> return tuple ( cookie . output ( ) . split ( "<STR_LIT>" , <NUM_LIT:1> ) ) <EOL> return None <EOL> def set_cookie ( name , _ , * args ) : <EOL> cookie = SimpleCookie ( ) <EOL> cookie [ name ] = base64 . b64encode ( "<STR_LIT::>" . join ( args ) ) <EOL> cookie [ name ] [ '<STR_LIT:path>' ] = "<STR_LIT:/>" <EOL> cookie [ name ] [ "<STR_LIT>" ] = _expiration ( <NUM_LIT:5> ) <EOL> logger . debug ( "<STR_LIT>" % cookie [ name ] [ "<STR_LIT>" ] ) <EOL> return tuple ( cookie . output ( ) . split ( "<STR_LIT>" , <NUM_LIT:1> ) ) <EOL> AUTHN_URLS = [ <EOL> ( r'<STR_LIT>' , ( SSO , "<STR_LIT>" ) ) , <EOL> ( r'<STR_LIT>' , ( SSO , "<STR_LIT>" ) ) , <EOL> ( r'<STR_LIT>' , ( SSO , "<STR_LIT>" ) ) , <EOL> ( r'<STR_LIT>' , ( SSO , "<STR_LIT>" ) ) , <EOL> ( r'<STR_LIT>' , ( SSO , "<STR_LIT>" ) ) , <EOL> ( r'<STR_LIT>' , ( SSO , "<STR_LIT>" ) ) , <EOL> ( r'<STR_LIT>' , ( SLO , "<STR_LIT>" ) ) , <EOL> ( r'<STR_LIT>' , ( SLO , "<STR_LIT>" ) ) , <EOL> ( r'<STR_LIT>' , ( SLO , "<STR_LIT>" ) ) , <EOL> ( r'<STR_LIT>' , ( SLO , "<STR_LIT>" ) ) , <EOL> ( r'<STR_LIT>' , ( SLO , "<STR_LIT>" ) ) , <EOL> ( r'<STR_LIT>' , ( SLO , "<STR_LIT>" ) ) , <EOL> ( r'<STR_LIT>' , ( AIDR , "<STR_LIT>" ) ) , <EOL> ( r'<STR_LIT>' , ( ARS , "<STR_LIT>" ) ) , <EOL> ( r'<STR_LIT>' , ( NMI , "<STR_LIT>" ) ) , <EOL> ( r'<STR_LIT>' , ( NMI , "<STR_LIT>" ) ) , <EOL> ( r'<STR_LIT>' , ( NMI , "<STR_LIT>" ) ) , <EOL> ( r'<STR_LIT>' , ( NMI , "<STR_LIT>" ) ) , <EOL> ( r'<STR_LIT>' , ( NMI , "<STR_LIT>" ) ) , <EOL> ( r'<STR_LIT>' , ( NMI , "<STR_LIT>" ) ) , <EOL> ( r'<STR_LIT>' , ( NMI , "<STR_LIT>" ) ) , <EOL> ( r'<STR_LIT>' , ( NMI , "<STR_LIT>" ) ) , <EOL> ( r'<STR_LIT>' , ( NIM , "<STR_LIT>" ) ) , <EOL> ( r'<STR_LIT>' , ( NIM , "<STR_LIT>" ) ) , <EOL> ( r'<STR_LIT>' , ( AQS , "<STR_LIT>" ) ) , <EOL> ( r'<STR_LIT>' , ( ATTR , "<STR_LIT>" ) ) <EOL> ] <EOL> NON_AUTHN_URLS = [ <EOL> ( r'<STR_LIT>' , do_verify ) , <EOL> ( r'<STR_LIT>' , ( SSO , "<STR_LIT>" ) ) , <EOL> ] <EOL> def metadata ( environ , start_response ) : <EOL> try : <EOL> path = args . path <EOL> if path is None or len ( path ) == <NUM_LIT:0> : <EOL> path = os . path . dirname ( os . path . abspath ( __file__ ) ) <EOL> if path [ - <NUM_LIT:1> ] != "<STR_LIT:/>" : <EOL> path += "<STR_LIT:/>" <EOL> metadata = create_metadata_string ( path + args . config , IDP . config , <EOL> args . valid , args . cert , args . keyfile , <EOL> args . id , args . name , args . sign ) <EOL> start_response ( '<STR_LIT>' , [ ( '<STR_LIT:Content-Type>' , "<STR_LIT>" ) ] ) <EOL> return metadata <EOL> except Exception as ex : <EOL> logger . error ( "<STR_LIT>" + ex . message ) <EOL> return not_found ( environ , start_response ) <EOL> def staticfile ( environ , start_response ) : <EOL> try : <EOL> path = args . path <EOL> if path is None or len ( path ) == <NUM_LIT:0> : <EOL> path = os . path . dirname ( os . path . abspath ( __file__ ) ) <EOL> if path [ - <NUM_LIT:1> ] != "<STR_LIT:/>" : <EOL> path += "<STR_LIT:/>" <EOL> path += environ . get ( '<STR_LIT>' , '<STR_LIT>' ) . lstrip ( '<STR_LIT:/>' ) <EOL> start_response ( '<STR_LIT>' , [ ( '<STR_LIT:Content-Type>' , "<STR_LIT>" ) ] ) <EOL> return open ( path , '<STR_LIT:r>' ) . read ( ) <EOL> except Exception as ex : <EOL> logger . error ( "<STR_LIT>" + ex . message ) <EOL> return not_found ( environ , start_response ) <EOL> def application ( environ , start_response ) : <EOL> """<STR_LIT>""" <EOL> path = environ . get ( '<STR_LIT>' , '<STR_LIT>' ) . lstrip ( '<STR_LIT:/>' ) <EOL> if path == "<STR_LIT>" : <EOL> return metadata ( environ , start_response ) <EOL> kaka = environ . get ( "<STR_LIT>" , None ) <EOL> logger . info ( "<STR_LIT>" % path ) <EOL> if kaka : <EOL> logger . info ( "<STR_LIT>" ) <EOL> user , authn_ref = info_from_cookie ( kaka ) <EOL> environ [ "<STR_LIT>" ] = authn_ref <EOL> else : <EOL> try : <EOL> query = parse_qs ( environ [ "<STR_LIT>" ] ) <EOL> logger . debug ( "<STR_LIT>" % query ) <EOL> user = IDP . cache . uid2user [ query [ "<STR_LIT:id>" ] [ <NUM_LIT:0> ] ] <EOL> except KeyError : <EOL> user = None <EOL> url_patterns = AUTHN_URLS <EOL> if not user : <EOL> logger . info ( "<STR_LIT>" ) <EOL> url_patterns = NON_AUTHN_URLS + url_patterns <EOL> for regex , callback in url_patterns : <EOL> match = re . search ( regex , path ) <EOL> if match is not None : <EOL> try : <EOL> environ [ '<STR_LIT>' ] = match . groups ( ) [ <NUM_LIT:0> ] <EOL> except IndexError : <EOL> environ [ '<STR_LIT>' ] = path <EOL> logger . debug ( "<STR_LIT>" % ( callback , ) ) <EOL> if isinstance ( callback , tuple ) : <EOL> cls = callback [ <NUM_LIT:0> ] ( environ , start_response , user ) <EOL> func = getattr ( cls , callback [ <NUM_LIT:1> ] ) <EOL> return func ( ) <EOL> return callback ( environ , start_response , user ) <EOL> if re . search ( r'<STR_LIT>' , path ) is not None : <EOL> return staticfile ( environ , start_response ) <EOL> return not_found ( environ , start_response ) <EOL> args = type ( '<STR_LIT>' , ( object , ) , { } ) <EOL> args . config = '<STR_LIT>' <EOL> args . mako_root = '<STR_LIT>' <EOL> args . path = None <EOL> import socket <EOL> from idp_user import USERS <EOL> from idp_user import EXTRA <EOL> from mako . lookup import TemplateLookup <EOL> AUTHN_BROKER = AuthnBroker ( ) <EOL> AUTHN_BROKER . add ( authn_context_class_ref ( PASSWORD ) , <EOL> username_password_authn , <NUM_LIT:10> , <EOL> "<STR_LIT>" % socket . gethostname ( ) ) <EOL> AUTHN_BROKER . add ( authn_context_class_ref ( UNSPECIFIED ) , <EOL> "<STR_LIT>" , <NUM_LIT:0> , "<STR_LIT>" % socket . gethostname ( ) ) <EOL> IDP = server . Server ( args . config , cache = Cache ( ) ) <EOL> IDP . ticket = { } <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> from wsgiref . simple_server import make_server <EOL> parser = argparse . ArgumentParser ( ) <EOL> parser . add_argument ( '<STR_LIT>' , dest = '<STR_LIT:path>' , help = '<STR_LIT>' ) <EOL> parser . add_argument ( '<STR_LIT>' , dest = '<STR_LIT>' , <EOL> help = "<STR_LIT>" ) <EOL> parser . add_argument ( '<STR_LIT:-c>' , dest = '<STR_LIT>' , help = '<STR_LIT>' ) <EOL> parser . add_argument ( '<STR_LIT>' , dest = '<STR_LIT:id>' , <EOL> help = "<STR_LIT>" ) <EOL> parser . add_argument ( '<STR_LIT>' , dest = '<STR_LIT>' , <EOL> help = "<STR_LIT>" ) <EOL> parser . add_argument ( '<STR_LIT>' , dest = '<STR_LIT:name>' ) <EOL> parser . add_argument ( '<STR_LIT>' , dest = '<STR_LIT>' , action = '<STR_LIT:store_true>' , <EOL> help = "<STR_LIT>" ) <EOL> parser . add_argument ( '<STR_LIT>' , dest = '<STR_LIT>' , default = "<STR_LIT>" ) <EOL> parser . add_argument ( dest = "<STR_LIT>" ) <EOL> args = parser . parse_args ( ) <EOL> _rot = args . mako_root <EOL> LOOKUP = TemplateLookup ( directories = [ _rot + '<STR_LIT>' , _rot + '<STR_LIT>' ] , <EOL> module_directory = _rot + '<STR_LIT>' , <EOL> input_encoding = '<STR_LIT:utf-8>' , output_encoding = '<STR_LIT:utf-8>' ) <EOL> HOST = '<STR_LIT:127.0.0.1>' <EOL> PORT = <NUM_LIT> <EOL> SRV = make_server ( HOST , PORT , application ) <EOL> print "<STR_LIT>" % ( HOST , PORT ) <EOL> SRV . serve_forever ( ) <EOL> else : <EOL> _rot = args . mako_root <EOL> LOOKUP = TemplateLookup ( directories = [ _rot + '<STR_LIT>' , _rot + '<STR_LIT>' ] , <EOL> module_directory = _rot + '<STR_LIT>' , <EOL> input_encoding = '<STR_LIT:utf-8>' , output_encoding = '<STR_LIT:utf-8>' ) </s>
