<s> __copyright__ = '<STR_LIT>' <EOL> from BaseHTTPServer import HTTPServer , BaseHTTPRequestHandler <EOL> from urlparse import urlparse <EOL> import time <EOL> import Cookie <EOL> import cgi <EOL> import cgitb <EOL> import sys <EOL> def quoteattr ( s ) : <EOL> qs = cgi . escape ( s , <NUM_LIT:1> ) <EOL> return '<STR_LIT>' % ( qs , ) <EOL> try : <EOL> import openid <EOL> except ImportError : <EOL> sys . stderr . write ( """<STR_LIT>""" ) <EOL> sys . exit ( <NUM_LIT:1> ) <EOL> from openid . extensions import sreg <EOL> from openid . server import server <EOL> from openid . store . filestore import FileOpenIDStore <EOL> from openid . consumer import discover <EOL> class OpenIDHTTPServer ( HTTPServer ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , * args , ** kwargs ) : <EOL> HTTPServer . __init__ ( self , * args , ** kwargs ) <EOL> if self . server_port != <NUM_LIT> : <EOL> self . base_url = ( '<STR_LIT>' % <EOL> ( self . server_name , self . server_port ) ) <EOL> else : <EOL> self . base_url = '<STR_LIT>' % ( self . server_name , ) <EOL> self . openid = None <EOL> self . approved = { } <EOL> self . lastCheckIDRequest = { } <EOL> def setOpenIDServer ( self , oidserver ) : <EOL> self . openid = oidserver <EOL> class ServerHandler ( BaseHTTPRequestHandler ) : <EOL> def __init__ ( self , * args , ** kwargs ) : <EOL> self . user = None <EOL> BaseHTTPRequestHandler . __init__ ( self , * args , ** kwargs ) <EOL> def do_GET ( self ) : <EOL> try : <EOL> self . parsed_uri = urlparse ( self . path ) <EOL> self . query = { } <EOL> for k , v in cgi . parse_qsl ( self . parsed_uri [ <NUM_LIT:4> ] ) : <EOL> self . query [ k ] = v <EOL> self . setUser ( ) <EOL> path = self . parsed_uri [ <NUM_LIT:2> ] . lower ( ) <EOL> if path == '<STR_LIT:/>' : <EOL> self . showMainPage ( ) <EOL> elif path == '<STR_LIT>' : <EOL> self . serverEndPoint ( self . query ) <EOL> elif path == '<STR_LIT>' : <EOL> self . showLoginPage ( '<STR_LIT:/>' , '<STR_LIT:/>' ) <EOL> elif path == '<STR_LIT>' : <EOL> self . doLogin ( ) <EOL> elif path . startswith ( '<STR_LIT>' ) : <EOL> self . showIdPage ( path ) <EOL> elif path . startswith ( '<STR_LIT>' ) : <EOL> self . showYadis ( path [ <NUM_LIT:7> : ] ) <EOL> elif path == '<STR_LIT>' : <EOL> self . showServerYadis ( ) <EOL> else : <EOL> self . send_response ( <NUM_LIT> ) <EOL> self . end_headers ( ) <EOL> except ( KeyboardInterrupt , SystemExit ) : <EOL> raise <EOL> except : <EOL> self . send_response ( <NUM_LIT> ) <EOL> self . send_header ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . end_headers ( ) <EOL> self . wfile . write ( cgitb . html ( sys . exc_info ( ) , context = <NUM_LIT:10> ) ) <EOL> def do_POST ( self ) : <EOL> try : <EOL> self . parsed_uri = urlparse ( self . path ) <EOL> self . setUser ( ) <EOL> content_length = int ( self . headers [ '<STR_LIT>' ] ) <EOL> post_data = self . rfile . read ( content_length ) <EOL> self . query = { } <EOL> for k , v in cgi . parse_qsl ( post_data ) : <EOL> self . query [ k ] = v <EOL> path = self . parsed_uri [ <NUM_LIT:2> ] <EOL> if path == '<STR_LIT>' : <EOL> self . serverEndPoint ( self . query ) <EOL> elif path == '<STR_LIT>' : <EOL> self . handleAllow ( self . query ) <EOL> else : <EOL> self . send_response ( <NUM_LIT> ) <EOL> self . end_headers ( ) <EOL> except ( KeyboardInterrupt , SystemExit ) : <EOL> raise <EOL> except : <EOL> self . send_response ( <NUM_LIT> ) <EOL> self . send_header ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . end_headers ( ) <EOL> self . wfile . write ( cgitb . html ( sys . exc_info ( ) , context = <NUM_LIT:10> ) ) <EOL> def handleAllow ( self , query ) : <EOL> request = self . server . lastCheckIDRequest . get ( self . user ) <EOL> if '<STR_LIT:yes>' in query : <EOL> if '<STR_LIT>' in query : <EOL> self . user = self . query [ '<STR_LIT>' ] <EOL> if request . idSelect ( ) : <EOL> identity = self . server . base_url + '<STR_LIT>' + query [ '<STR_LIT>' ] <EOL> else : <EOL> identity = request . identity <EOL> trust_root = request . trust_root <EOL> if self . query . get ( '<STR_LIT>' , '<STR_LIT>' ) == '<STR_LIT:yes>' : <EOL> self . server . approved [ ( identity , trust_root ) ] = '<STR_LIT>' <EOL> response = self . approved ( request , identity ) <EOL> elif '<STR_LIT>' in query : <EOL> response = request . answer ( False ) <EOL> else : <EOL> assert False , '<STR_LIT>' % ( query , ) <EOL> self . displayResponse ( response ) <EOL> def setUser ( self ) : <EOL> cookies = self . headers . get ( '<STR_LIT>' ) <EOL> if cookies : <EOL> morsel = Cookie . BaseCookie ( cookies ) . get ( '<STR_LIT:user>' ) <EOL> if morsel : <EOL> self . user = morsel . value <EOL> def isAuthorized ( self , identity_url , trust_root ) : <EOL> if self . user is None : <EOL> return False <EOL> if identity_url != self . server . base_url + '<STR_LIT>' + self . user : <EOL> return False <EOL> key = ( identity_url , trust_root ) <EOL> return self . server . approved . get ( key ) is not None <EOL> def serverEndPoint ( self , query ) : <EOL> try : <EOL> request = self . server . openid . decodeRequest ( query ) <EOL> except server . ProtocolError , why : <EOL> self . displayResponse ( why ) <EOL> return <EOL> if request is None : <EOL> self . showAboutPage ( ) <EOL> return <EOL> if request . mode in [ "<STR_LIT>" , "<STR_LIT>" ] : <EOL> self . handleCheckIDRequest ( request ) <EOL> else : <EOL> response = self . server . openid . handleRequest ( request ) <EOL> self . displayResponse ( response ) <EOL> def addSRegResponse ( self , request , response ) : <EOL> sreg_req = sreg . SRegRequest . fromOpenIDRequest ( request ) <EOL> sreg_data = { <EOL> '<STR_LIT>' : self . user <EOL> } <EOL> sreg_resp = sreg . SRegResponse . extractResponse ( sreg_req , sreg_data ) <EOL> response . addExtension ( sreg_resp ) <EOL> def approved ( self , request , identifier = None ) : <EOL> response = request . answer ( True , identity = identifier ) <EOL> self . addSRegResponse ( request , response ) <EOL> return response <EOL> def handleCheckIDRequest ( self , request ) : <EOL> is_authorized = self . isAuthorized ( request . identity , request . trust_root ) <EOL> if is_authorized : <EOL> response = self . approved ( request ) <EOL> self . displayResponse ( response ) <EOL> elif request . immediate : <EOL> response = request . answer ( False ) <EOL> self . displayResponse ( response ) <EOL> else : <EOL> self . server . lastCheckIDRequest [ self . user ] = request <EOL> self . showDecidePage ( request ) <EOL> def displayResponse ( self , response ) : <EOL> try : <EOL> webresponse = self . server . openid . encodeResponse ( response ) <EOL> except server . EncodingError , why : <EOL> text = why . response . encodeToKVForm ( ) <EOL> self . showErrorPage ( '<STR_LIT>' % cgi . escape ( text ) ) <EOL> return <EOL> self . send_response ( webresponse . code ) <EOL> for header , value in webresponse . headers . iteritems ( ) : <EOL> self . send_header ( header , value ) <EOL> self . writeUserHeader ( ) <EOL> self . end_headers ( ) <EOL> if webresponse . body : <EOL> self . wfile . write ( webresponse . body ) <EOL> def doLogin ( self ) : <EOL> if '<STR_LIT>' in self . query : <EOL> if '<STR_LIT:user>' in self . query : <EOL> self . user = self . query [ '<STR_LIT:user>' ] <EOL> else : <EOL> self . user = None <EOL> self . redirect ( self . query [ '<STR_LIT>' ] ) <EOL> elif '<STR_LIT>' in self . query : <EOL> self . redirect ( self . query [ '<STR_LIT>' ] ) <EOL> else : <EOL> assert <NUM_LIT:0> , '<STR_LIT>' % ( self . query , ) <EOL> def redirect ( self , url ) : <EOL> self . send_response ( <NUM_LIT> ) <EOL> self . send_header ( '<STR_LIT>' , url ) <EOL> self . writeUserHeader ( ) <EOL> self . end_headers ( ) <EOL> def writeUserHeader ( self ) : <EOL> if self . user is None : <EOL> t1970 = time . gmtime ( <NUM_LIT:0> ) <EOL> expires = time . strftime ( <EOL> '<STR_LIT>' , t1970 ) <EOL> self . send_header ( '<STR_LIT>' , '<STR_LIT>' % expires ) <EOL> else : <EOL> self . send_header ( '<STR_LIT>' , '<STR_LIT>' % self . user ) <EOL> def showAboutPage ( self ) : <EOL> endpoint_url = self . server . base_url + '<STR_LIT>' <EOL> def link ( url ) : <EOL> url_attr = quoteattr ( url ) <EOL> url_text = cgi . escape ( url ) <EOL> return '<STR_LIT>' % ( url_attr , url_text ) <EOL> def term ( url , text ) : <EOL> return '<STR_LIT>' % ( link ( url ) , text ) <EOL> resources = [ <EOL> ( self . server . base_url , "<STR_LIT>" ) , <EOL> ( '<STR_LIT>' , <EOL> '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ] <EOL> resource_markup = '<STR_LIT>' . join ( [ term ( url , text ) for url , text in resources ] ) <EOL> self . showPage ( <NUM_LIT:200> , '<STR_LIT>' , msg = """<STR_LIT>""" % ( link ( endpoint_url ) , resource_markup , ) ) <EOL> def showErrorPage ( self , error_message ) : <EOL> self . showPage ( <NUM_LIT> , '<STR_LIT>' , err = '''<STR_LIT>''' % error_message ) <EOL> def showDecidePage ( self , request ) : <EOL> id_url_base = self . server . base_url + '<STR_LIT>' <EOL> assert ( request . identity . startswith ( id_url_base ) or <EOL> request . idSelect ( ) ) , repr ( ( request . identity , id_url_base ) ) <EOL> expected_user = request . identity [ len ( id_url_base ) : ] <EOL> if request . idSelect ( ) : <EOL> msg = '''<STR_LIT>''' <EOL> fdata = { <EOL> '<STR_LIT>' : id_url_base , <EOL> '<STR_LIT>' : request . trust_root , <EOL> } <EOL> form = '''<STR_LIT>''' % fdata <EOL> elif expected_user == self . user : <EOL> msg = '''<STR_LIT>''' <EOL> fdata = { <EOL> '<STR_LIT>' : request . identity , <EOL> '<STR_LIT>' : request . trust_root , <EOL> } <EOL> form = '''<STR_LIT>''' % fdata <EOL> else : <EOL> mdata = { <EOL> '<STR_LIT>' : expected_user , <EOL> '<STR_LIT:user>' : self . user , <EOL> } <EOL> msg = '''<STR_LIT>''' % mdata <EOL> fdata = { <EOL> '<STR_LIT>' : request . identity , <EOL> '<STR_LIT>' : request . trust_root , <EOL> '<STR_LIT>' : expected_user , <EOL> } <EOL> form = '''<STR_LIT>''' % fdata <EOL> self . showPage ( <NUM_LIT:200> , '<STR_LIT>' , msg = msg , form = form ) <EOL> def showIdPage ( self , path ) : <EOL> link_tag = '<STR_LIT>' % self . server . base_url <EOL> yadis_loc_tag = '<STR_LIT>' % ( self . server . base_url + '<STR_LIT>' + path [ <NUM_LIT:4> : ] ) <EOL> disco_tags = link_tag + yadis_loc_tag <EOL> ident = self . server . base_url + path [ <NUM_LIT:1> : ] <EOL> approved_trust_roots = [ ] <EOL> for ( aident , trust_root ) in self . server . approved . keys ( ) : <EOL> if aident == ident : <EOL> trs = '<STR_LIT>' % cgi . escape ( trust_root ) <EOL> approved_trust_roots . append ( trs ) <EOL> if approved_trust_roots : <EOL> prepend = '<STR_LIT>' <EOL> approved_trust_roots . insert ( <NUM_LIT:0> , prepend ) <EOL> approved_trust_roots . append ( '<STR_LIT>' ) <EOL> msg = '<STR_LIT>' . join ( approved_trust_roots ) <EOL> else : <EOL> msg = '<STR_LIT>' <EOL> self . showPage ( <NUM_LIT:200> , '<STR_LIT>' , head_extras = disco_tags , msg = '''<STR_LIT>''' % ( ident , msg ) ) <EOL> def showYadis ( self , user ) : <EOL> self . send_response ( <NUM_LIT:200> ) <EOL> self . send_header ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . end_headers ( ) <EOL> endpoint_url = self . server . base_url + '<STR_LIT>' <EOL> user_url = self . server . base_url + '<STR_LIT>' + user <EOL> self . wfile . write ( """<STR_LIT>""" % ( discover . OPENID_2_0_TYPE , discover . OPENID_1_0_TYPE , <EOL> endpoint_url , user_url ) ) <EOL> def showServerYadis ( self ) : <EOL> self . send_response ( <NUM_LIT:200> ) <EOL> self . send_header ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . end_headers ( ) <EOL> endpoint_url = self . server . base_url + '<STR_LIT>' <EOL> self . wfile . write ( """<STR_LIT>""" % ( discover . OPENID_IDP_2_0_TYPE , endpoint_url , ) ) <EOL> def showMainPage ( self ) : <EOL> yadis_tag = '<STR_LIT>' % ( self . server . base_url + '<STR_LIT>' ) <EOL> if self . user : <EOL> openid_url = self . server . base_url + '<STR_LIT>' + self . user <EOL> user_message = """<STR_LIT>""" % ( self . user , quoteattr ( openid_url ) , openid_url ) <EOL> else : <EOL> user_message = """<STR_LIT>""" <EOL> self . showPage ( <NUM_LIT:200> , '<STR_LIT>' , head_extras = yadis_tag , msg = '''<STR_LIT>''' % ( user_message , quoteattr ( self . server . base_url ) , self . server . base_url ) ) <EOL> def showLoginPage ( self , success_to , fail_to ) : <EOL> self . showPage ( <NUM_LIT:200> , '<STR_LIT>' , form = '''<STR_LIT>''' % ( success_to , fail_to ) ) <EOL> def showPage ( self , response_code , title , <EOL> head_extras = '<STR_LIT>' , msg = None , err = None , form = None ) : <EOL> if self . user is None : <EOL> user_link = '<STR_LIT>' <EOL> else : <EOL> user_link = '<STR_LIT>' % ( self . user , self . user ) <EOL> body = '<STR_LIT>' <EOL> if err is not None : <EOL> body += '''<STR_LIT>''' % err <EOL> if msg is not None : <EOL> body += '''<STR_LIT>''' % msg <EOL> if form is not None : <EOL> body += '''<STR_LIT>''' % form <EOL> contents = { <EOL> '<STR_LIT:title>' : '<STR_LIT>' + title , <EOL> '<STR_LIT>' : head_extras , <EOL> '<STR_LIT:body>' : body , <EOL> '<STR_LIT>' : user_link , <EOL> } <EOL> self . send_response ( response_code ) <EOL> self . writeUserHeader ( ) <EOL> self . send_header ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . end_headers ( ) <EOL> self . wfile . write ( '''<STR_LIT>''' % contents ) <EOL> def main ( host , port , data_path ) : <EOL> addr = ( host , port ) <EOL> httpserver = OpenIDHTTPServer ( addr , ServerHandler ) <EOL> store = FileOpenIDStore ( data_path ) <EOL> oidserver = server . Server ( store , httpserver . base_url + '<STR_LIT>' ) <EOL> httpserver . setOpenIDServer ( oidserver ) <EOL> print '<STR_LIT>' <EOL> print httpserver . base_url <EOL> httpserver . serve_forever ( ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> host = '<STR_LIT:localhost>' <EOL> data_path = '<STR_LIT>' <EOL> port = <NUM_LIT> <EOL> try : <EOL> import optparse <EOL> except ImportError : <EOL> pass <EOL> else : <EOL> parser = optparse . OptionParser ( '<STR_LIT>' ) <EOL> parser . add_option ( <EOL> '<STR_LIT>' , '<STR_LIT>' , dest = '<STR_LIT>' , default = data_path , <EOL> help = '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> parser . add_option ( <EOL> '<STR_LIT>' , '<STR_LIT>' , dest = '<STR_LIT:port>' , type = '<STR_LIT:int>' , default = port , <EOL> help = '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> parser . add_option ( <EOL> '<STR_LIT>' , '<STR_LIT>' , dest = '<STR_LIT:host>' , default = host , <EOL> help = '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> options , args = parser . parse_args ( ) <EOL> if args : <EOL> parser . error ( '<STR_LIT>' % args ) <EOL> host = options . host <EOL> port = options . port <EOL> data_path = options . data_path <EOL> main ( host , port , data_path ) </s>
