<s> import logging <EOL> from urllib import urlencode <EOL> from urlparse import parse_qs <EOL> from urlparse import urlsplit <EOL> import time <EOL> import ldap <EOL> from saml2 import SAMLError <EOL> from saml2 . aes import AESCipher <EOL> from saml2 . httputil import Response <EOL> from saml2 . httputil import make_cookie <EOL> from saml2 . httputil import Redirect <EOL> from saml2 . httputil import Unauthorized <EOL> from saml2 . httputil import parse_cookie <EOL> __author__ = '<STR_LIT>' <EOL> logger = logging . getLogger ( __name__ ) <EOL> class AuthnFailure ( SAMLError ) : <EOL> pass <EOL> class EncodeError ( SAMLError ) : <EOL> pass <EOL> class UserAuthnMethod ( object ) : <EOL> def __init__ ( self , srv ) : <EOL> self . srv = srv <EOL> def __call__ ( self , * args , ** kwargs ) : <EOL> raise NotImplemented <EOL> def authenticated_as ( self , ** kwargs ) : <EOL> raise NotImplemented <EOL> def verify ( self , ** kwargs ) : <EOL> raise NotImplemented <EOL> def is_equal ( a , b ) : <EOL> if len ( a ) != len ( b ) : <EOL> return False <EOL> result = <NUM_LIT:0> <EOL> for x , y in zip ( a , b ) : <EOL> result |= x ^ y <EOL> return result == <NUM_LIT:0> <EOL> def url_encode_params ( params = None ) : <EOL> if not isinstance ( params , dict ) : <EOL> raise EncodeError ( "<STR_LIT>" ) <EOL> params_list = [ ] <EOL> for k , v in params . items ( ) : <EOL> if isinstance ( v , list ) : <EOL> params_list . extend ( [ ( k , x ) for x in v ] ) <EOL> else : <EOL> params_list . append ( ( k , v ) ) <EOL> return urlencode ( params_list ) <EOL> def create_return_url ( base , query , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> part = urlsplit ( base ) <EOL> if part . fragment : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> for key , values in parse_qs ( query ) . items ( ) : <EOL> if key in kwargs : <EOL> if isinstance ( kwargs [ key ] , basestring ) : <EOL> kwargs [ key ] = [ kwargs [ key ] ] <EOL> kwargs [ key ] . extend ( values ) <EOL> else : <EOL> kwargs [ key ] = values <EOL> if part . query : <EOL> for key , values in parse_qs ( part . query ) . items ( ) : <EOL> if key in kwargs : <EOL> if isinstance ( kwargs [ key ] , basestring ) : <EOL> kwargs [ key ] = [ kwargs [ key ] ] <EOL> kwargs [ key ] . extend ( values ) <EOL> else : <EOL> kwargs [ key ] = values <EOL> _pre = base . split ( "<STR_LIT:?>" ) [ <NUM_LIT:0> ] <EOL> else : <EOL> _pre = base <EOL> logger . debug ( "<STR_LIT>" % kwargs ) <EOL> return "<STR_LIT>" % ( _pre , url_encode_params ( kwargs ) ) <EOL> class UsernamePasswordMako ( UserAuthnMethod ) : <EOL> """<STR_LIT>""" <EOL> cookie_name = "<STR_LIT>" <EOL> def __init__ ( self , srv , mako_template , template_lookup , pwd , return_to ) : <EOL> """<STR_LIT>""" <EOL> UserAuthnMethod . __init__ ( self , srv ) <EOL> self . mako_template = mako_template <EOL> self . template_lookup = template_lookup <EOL> self . passwd = pwd <EOL> self . return_to = return_to <EOL> self . active = { } <EOL> self . query_param = "<STR_LIT>" <EOL> self . aes = AESCipher ( self . srv . symkey , srv . iv ) <EOL> def __call__ ( self , cookie = None , policy_url = None , logo_url = None , <EOL> query = "<STR_LIT>" , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> if cookie : <EOL> headers = [ cookie ] <EOL> else : <EOL> headers = [ ] <EOL> resp = Response ( headers = headers ) <EOL> argv = { "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT:password>" : "<STR_LIT>" , <EOL> "<STR_LIT:action>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : policy_url , <EOL> "<STR_LIT>" : logo_url , <EOL> "<STR_LIT>" : query } <EOL> logger . info ( "<STR_LIT>" % argv ) <EOL> mte = self . template_lookup . get_template ( self . mako_template ) <EOL> resp . message = mte . render ( ** argv ) <EOL> return resp <EOL> def _verify ( self , pwd , user ) : <EOL> assert is_equal ( pwd , self . passwd [ user ] ) <EOL> def verify ( self , request , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( request , basestring ) : <EOL> _dict = parse_qs ( request ) <EOL> elif isinstance ( request , dict ) : <EOL> _dict = request <EOL> else : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> try : <EOL> self . _verify ( _dict [ "<STR_LIT:password>" ] [ <NUM_LIT:0> ] , _dict [ "<STR_LIT>" ] [ <NUM_LIT:0> ] ) <EOL> timestamp = str ( int ( time . mktime ( time . gmtime ( ) ) ) ) <EOL> info = self . aes . encrypt ( "<STR_LIT>" . join ( [ _dict [ "<STR_LIT>" ] [ <NUM_LIT:0> ] , timestamp ] ) ) <EOL> self . active [ info ] = timestamp <EOL> cookie = make_cookie ( self . cookie_name , info , self . srv . seed ) <EOL> return_to = create_return_url ( self . return_to , _dict [ "<STR_LIT>" ] [ <NUM_LIT:0> ] , <EOL> ** { self . query_param : "<STR_LIT:true>" } ) <EOL> resp = Redirect ( return_to , headers = [ cookie ] ) <EOL> except ( AssertionError , KeyError ) : <EOL> resp = Unauthorized ( "<STR_LIT>" ) <EOL> return resp <EOL> def authenticated_as ( self , cookie = None , ** kwargs ) : <EOL> if cookie is None : <EOL> return None <EOL> else : <EOL> logger . debug ( "<STR_LIT>" % kwargs ) <EOL> try : <EOL> info , timestamp = parse_cookie ( self . cookie_name , <EOL> self . srv . seed , cookie ) <EOL> if self . active [ info ] == timestamp : <EOL> uid , _ts = self . aes . decrypt ( info ) . split ( "<STR_LIT>" ) <EOL> if timestamp == _ts : <EOL> return { "<STR_LIT>" : uid } <EOL> except Exception : <EOL> pass <EOL> return None <EOL> def done ( self , areq ) : <EOL> try : <EOL> _ = areq [ self . query_param ] <EOL> return False <EOL> except KeyError : <EOL> return True <EOL> class SocialService ( UserAuthnMethod ) : <EOL> def __init__ ( self , social ) : <EOL> UserAuthnMethod . __init__ ( self , None ) <EOL> self . social = social <EOL> def __call__ ( self , server_env , cookie = None , sid = "<STR_LIT>" , query = "<STR_LIT>" , ** kwargs ) : <EOL> return self . social . begin ( server_env , cookie , sid , query ) <EOL> def callback ( self , server_env , cookie = None , sid = "<STR_LIT>" , query = "<STR_LIT>" , ** kwargs ) : <EOL> return self . social . callback ( server_env , cookie , sid , query , ** kwargs ) <EOL> class AuthnMethodChooser ( object ) : <EOL> def __init__ ( self , methods = None ) : <EOL> self . methods = methods <EOL> def __call__ ( self , ** kwargs ) : <EOL> if not self . methods : <EOL> raise SAMLError ( "<STR_LIT>" ) <EOL> elif len ( self . methods ) == <NUM_LIT:1> : <EOL> return self . methods [ <NUM_LIT:0> ] <EOL> else : <EOL> pass <EOL> class LDAPAuthn ( UsernamePasswordMako ) : <EOL> def __init__ ( self , srv , ldapsrv , return_to , <EOL> dn_pattern , mako_template , template_lookup ) : <EOL> """<STR_LIT>""" <EOL> UsernamePasswordMako . __init__ ( self , srv , mako_template , template_lookup , <EOL> None , return_to ) <EOL> self . ldap = ldap . initialize ( ldapsrv ) <EOL> self . ldap . protocol_version = <NUM_LIT:3> <EOL> self . ldap . set_option ( ldap . OPT_REFERRALS , <NUM_LIT:0> ) <EOL> self . dn_pattern = dn_pattern <EOL> def _verify ( self , pwd , user ) : <EOL> """<STR_LIT>""" <EOL> _dn = self . dn_pattern % user <EOL> try : <EOL> self . ldap . simple_bind_s ( _dn , pwd ) <EOL> except Exception : <EOL> raise AssertionError ( ) </s>
