<s> import urllib . parse <EOL> import time <EOL> import warnings <EOL> import pprint <EOL> from openid . message import Message , OPENID_NS , OPENID2_NS , IDENTIFIER_SELECT , OPENID1_NS , BARE_NS <EOL> from openid import cryptutil , oidutil , kvform <EOL> from openid . store . nonce import mkNonce , split as splitNonce <EOL> from openid . consumer . discover import OpenIDServiceEndpoint , OPENID_2_0_TYPE , OPENID_1_1_TYPE <EOL> from openid . consumer . consumer import AuthRequest , GenericConsumer , SUCCESS , FAILURE , CANCEL , SETUP_NEEDED , SuccessResponse , FailureResponse , SetupNeededResponse , CancelResponse , DiffieHellmanSHA1ConsumerSession , Consumer , PlainTextConsumerSession , SetupNeededError , DiffieHellmanSHA256ConsumerSession , ServerError , ProtocolError , _httpResponseToMessage <EOL> from openid import association <EOL> from openid . server . server import PlainTextServerSession , DiffieHellmanSHA1ServerSession <EOL> from openid . yadis . manager import Discovery <EOL> from openid . yadis . discover import DiscoveryFailure <EOL> from openid . dh import DiffieHellman <EOL> from openid . fetchers import HTTPResponse , HTTPFetchingError <EOL> from openid import fetchers <EOL> from openid . store import memstore <EOL> from . support import CatchLogs <EOL> assocs = [ <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT:\x00>' * <NUM_LIT:20> , '<STR_LIT>' ) , <EOL> ] <EOL> def mkSuccess ( endpoint , q ) : <EOL> """<STR_LIT>""" <EOL> signed_list = [ '<STR_LIT>' + k for k in list ( q . keys ( ) ) ] <EOL> return SuccessResponse ( endpoint , Message . fromOpenIDArgs ( q ) , signed_list ) <EOL> def parseQuery ( qs ) : <EOL> q = { } <EOL> for ( k , v ) in urllib . parse . parse_qsl ( qs ) : <EOL> assert k not in q <EOL> q [ k ] = v <EOL> return q <EOL> def associate ( qs , assoc_secret , assoc_handle ) : <EOL> """<STR_LIT>""" <EOL> q = parseQuery ( qs ) <EOL> assert q [ '<STR_LIT>' ] == '<STR_LIT>' <EOL> assert q [ '<STR_LIT>' ] == '<STR_LIT>' <EOL> reply_dict = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : assoc_handle , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> if q . get ( '<STR_LIT>' ) == '<STR_LIT>' : <EOL> assert len ( q ) == <NUM_LIT:6> or len ( q ) == <NUM_LIT:4> <EOL> message = Message . fromPostArgs ( q ) <EOL> session = DiffieHellmanSHA1ServerSession . fromMessage ( message ) <EOL> reply_dict [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> else : <EOL> assert len ( q ) == <NUM_LIT:2> <EOL> session = PlainTextServerSession . fromQuery ( q ) <EOL> reply_dict . update ( session . answer ( assoc_secret ) ) <EOL> return kvform . dictToKV ( reply_dict ) <EOL> GOODSIG = "<STR_LIT>" <EOL> class GoodAssociation : <EOL> expiresIn = <NUM_LIT> <EOL> handle = "<STR_LIT>" <EOL> def checkMessageSignature ( self , message ) : <EOL> return message . getArg ( OPENID_NS , '<STR_LIT>' ) == GOODSIG <EOL> class GoodAssocStore ( memstore . MemoryStore ) : <EOL> def getAssociation ( self , server_url , handle = None ) : <EOL> return GoodAssociation ( ) <EOL> class TestFetcher ( object ) : <EOL> def __init__ ( self , user_url , user_page , xxx_todo_changeme ) : <EOL> ( assoc_secret , assoc_handle ) = xxx_todo_changeme <EOL> self . get_responses = { <EOL> user_url : self . response ( user_url , <NUM_LIT:200> , user_page ) <EOL> } <EOL> self . assoc_secret = assoc_secret <EOL> self . assoc_handle = assoc_handle <EOL> self . num_assocs = <NUM_LIT:0> <EOL> def response ( self , url , status , body ) : <EOL> return HTTPResponse ( <EOL> final_url = url , status = status , headers = { } , body = body ) <EOL> def fetch ( self , url , body = None , headers = None ) : <EOL> if body is None : <EOL> if url in self . get_responses : <EOL> return self . get_responses [ url ] <EOL> else : <EOL> try : <EOL> body . index ( '<STR_LIT>' ) <EOL> except ValueError : <EOL> pass <EOL> else : <EOL> assert body . find ( '<STR_LIT>' ) != - <NUM_LIT:1> <EOL> response = associate ( <EOL> body , self . assoc_secret , self . assoc_handle ) <EOL> self . num_assocs += <NUM_LIT:1> <EOL> return self . response ( url , <NUM_LIT:200> , response ) <EOL> return self . response ( url , <NUM_LIT> , '<STR_LIT>' ) <EOL> def makeFastConsumerSession ( ) : <EOL> """<STR_LIT>""" <EOL> dh = DiffieHellman ( <NUM_LIT> , <NUM_LIT:2> ) <EOL> return DiffieHellmanSHA1ConsumerSession ( dh ) <EOL> def setConsumerSession ( con ) : <EOL> con . session_types = { '<STR_LIT>' : makeFastConsumerSession } <EOL> def _test_success ( server_url , user_url , delegate_url , links , immediate = False ) : <EOL> if isinstance ( server_url , bytes ) : <EOL> server_url = str ( server_url , encoding = "<STR_LIT:utf-8>" ) <EOL> if isinstance ( user_url , bytes ) : <EOL> user_url = str ( user_url , encoding = "<STR_LIT:utf-8>" ) <EOL> if isinstance ( delegate_url , bytes ) : <EOL> delegate_url = str ( delegate_url , encoding = "<STR_LIT:utf-8>" ) <EOL> if isinstance ( links , bytes ) : <EOL> links = str ( links , encoding = "<STR_LIT:utf-8>" ) <EOL> store = memstore . MemoryStore ( ) <EOL> if immediate : <EOL> mode = '<STR_LIT>' <EOL> else : <EOL> mode = '<STR_LIT>' <EOL> endpoint = OpenIDServiceEndpoint ( ) <EOL> endpoint . claimed_id = user_url <EOL> endpoint . server_url = server_url <EOL> endpoint . local_id = delegate_url <EOL> endpoint . type_uris = [ OPENID_1_1_TYPE ] <EOL> fetcher = TestFetcher ( None , None , assocs [ <NUM_LIT:0> ] ) <EOL> fetchers . setDefaultFetcher ( fetcher , wrap_exceptions = False ) <EOL> def run ( ) : <EOL> trust_root = str ( consumer_url , encoding = "<STR_LIT:utf-8>" ) <EOL> consumer = GenericConsumer ( store ) <EOL> setConsumerSession ( consumer ) <EOL> request = consumer . begin ( endpoint ) <EOL> return_to = str ( consumer_url , encoding = "<STR_LIT:utf-8>" ) <EOL> m = request . getMessage ( trust_root , return_to , immediate ) <EOL> redirect_url = request . redirectURL ( trust_root , return_to , immediate ) <EOL> if isinstance ( redirect_url , bytes ) : <EOL> redirect_url = str ( redirect_url , encoding = "<STR_LIT:utf-8>" ) <EOL> parsed = urllib . parse . urlparse ( redirect_url ) <EOL> qs = parsed [ <NUM_LIT:4> ] <EOL> q = parseQuery ( qs ) <EOL> new_return_to = q [ '<STR_LIT>' ] <EOL> del q [ '<STR_LIT>' ] <EOL> expected = { <EOL> '<STR_LIT>' : mode , <EOL> '<STR_LIT>' : delegate_url , <EOL> '<STR_LIT>' : trust_root , <EOL> '<STR_LIT>' : fetcher . assoc_handle , <EOL> } <EOL> assert q == expected , pprint . pformat ( ( q , expected ) ) <EOL> assert new_return_to . startswith ( return_to ) <EOL> assert redirect_url . startswith ( server_url ) <EOL> parsed = urllib . parse . urlparse ( new_return_to ) <EOL> query = parseQuery ( parsed [ <NUM_LIT:4> ] ) <EOL> query . update ( { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : new_return_to , <EOL> '<STR_LIT>' : delegate_url , <EOL> '<STR_LIT>' : fetcher . assoc_handle , <EOL> } ) <EOL> assoc = store . getAssociation ( server_url , fetcher . assoc_handle ) <EOL> message = Message . fromPostArgs ( query ) <EOL> message = assoc . signMessage ( message ) <EOL> info = consumer . complete ( message , request . endpoint , new_return_to ) <EOL> assert info . status == SUCCESS , info . message <EOL> assert info . identity_url == user_url <EOL> assert fetcher . num_assocs == <NUM_LIT:0> <EOL> run ( ) <EOL> assert fetcher . num_assocs == <NUM_LIT:1> <EOL> run ( ) <EOL> assert fetcher . num_assocs == <NUM_LIT:1> <EOL> store . removeAssociation ( server_url , fetcher . assoc_handle ) <EOL> run ( ) <EOL> assert fetcher . num_assocs == <NUM_LIT:2> <EOL> run ( ) <EOL> assert fetcher . num_assocs == <NUM_LIT:2> <EOL> import unittest <EOL> http_server_url = b'<STR_LIT>' <EOL> consumer_url = b'<STR_LIT>' <EOL> https_server_url = b'<STR_LIT>' <EOL> class TestSuccess ( unittest . TestCase , CatchLogs ) : <EOL> server_url = http_server_url <EOL> user_url = b'<STR_LIT>' <EOL> delegate_url = b'<STR_LIT>' <EOL> def setUp ( self ) : <EOL> CatchLogs . setUp ( self ) <EOL> self . links = '<STR_LIT>' % ( <EOL> self . server_url , ) <EOL> self . delegate_links = ( '<STR_LIT>' <EOL> '<STR_LIT>' ) % ( <EOL> self . server_url , self . delegate_url ) <EOL> def tearDown ( self ) : <EOL> CatchLogs . tearDown ( self ) <EOL> def test_nodelegate ( self ) : <EOL> _test_success ( self . server_url , self . user_url , <EOL> self . user_url , self . links ) <EOL> def test_nodelegateImmediate ( self ) : <EOL> _test_success ( self . server_url , self . user_url , <EOL> self . user_url , self . links , True ) <EOL> def test_delegate ( self ) : <EOL> _test_success ( self . server_url , self . user_url , <EOL> self . delegate_url , self . delegate_links ) <EOL> def test_delegateImmediate ( self ) : <EOL> _test_success ( self . server_url , self . user_url , <EOL> self . delegate_url , self . delegate_links , True ) <EOL> class TestSuccessHTTPS ( TestSuccess ) : <EOL> server_url = https_server_url <EOL> class TestConstruct ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . store_sentinel = object ( ) <EOL> def test_construct ( self ) : <EOL> oidc = GenericConsumer ( self . store_sentinel ) <EOL> self . assertTrue ( oidc . store is self . store_sentinel ) <EOL> def test_nostore ( self ) : <EOL> self . assertRaises ( TypeError , GenericConsumer ) <EOL> class TestIdRes ( unittest . TestCase , CatchLogs ) : <EOL> consumer_class = GenericConsumer <EOL> def setUp ( self ) : <EOL> CatchLogs . setUp ( self ) <EOL> self . store = memstore . MemoryStore ( ) <EOL> self . consumer = self . consumer_class ( self . store ) <EOL> self . return_to = "<STR_LIT>" <EOL> self . endpoint = OpenIDServiceEndpoint ( ) <EOL> self . endpoint . claimed_id = self . consumer_id = "<STR_LIT>" <EOL> self . endpoint . server_url = self . server_url = "<STR_LIT>" <EOL> self . endpoint . local_id = self . server_id = "<STR_LIT>" <EOL> self . endpoint . type_uris = [ OPENID_1_1_TYPE ] <EOL> def disableDiscoveryVerification ( self ) : <EOL> """<STR_LIT>""" <EOL> def dummyVerifyDiscover ( _ , endpoint ) : <EOL> return endpoint <EOL> self . consumer . _verifyDiscoveryResults = dummyVerifyDiscover <EOL> def disableReturnToChecking ( self ) : <EOL> def checkReturnTo ( unused1 , unused2 ) : <EOL> return True <EOL> self . consumer . _checkReturnTo = checkReturnTo <EOL> complete = self . consumer . complete <EOL> def callCompleteWithoutReturnTo ( message , endpoint ) : <EOL> return complete ( message , endpoint , None ) <EOL> self . consumer . complete = callCompleteWithoutReturnTo <EOL> class TestIdResCheckSignature ( TestIdRes ) : <EOL> def setUp ( self ) : <EOL> TestIdRes . setUp ( self ) <EOL> self . assoc = GoodAssociation ( ) <EOL> self . assoc . handle = "<STR_LIT>" <EOL> self . store . storeAssociation ( self . endpoint . server_url , self . assoc ) <EOL> self . message = Message . fromPostArgs ( { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : GOODSIG , <EOL> '<STR_LIT>' : self . assoc . handle , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } ) <EOL> def test_sign ( self ) : <EOL> self . consumer . _idResCheckSignature ( self . message , <EOL> self . endpoint . server_url ) <EOL> def test_signFailsWithBadSig ( self ) : <EOL> self . message . setArg ( OPENID_NS , '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertRaises ( <EOL> ProtocolError , self . consumer . _idResCheckSignature , <EOL> self . message , self . endpoint . server_url ) <EOL> def test_stateless ( self ) : <EOL> self . message . setArg ( OPENID_NS , "<STR_LIT>" , "<STR_LIT>" ) <EOL> self . consumer . _processCheckAuthResponse = ( <EOL> lambda response , server_url : True ) <EOL> self . consumer . _makeKVPost = lambda args , server_url : { } <EOL> self . consumer . _idResCheckSignature ( self . message , <EOL> self . endpoint . server_url ) <EOL> def test_statelessRaisesError ( self ) : <EOL> self . message . setArg ( OPENID_NS , "<STR_LIT>" , "<STR_LIT>" ) <EOL> self . consumer . _checkAuth = lambda unused1 , unused2 : False <EOL> self . assertRaises ( <EOL> ProtocolError , self . consumer . _idResCheckSignature , <EOL> self . message , self . endpoint . server_url ) <EOL> def test_stateless_noStore ( self ) : <EOL> self . message . setArg ( OPENID_NS , "<STR_LIT>" , "<STR_LIT>" ) <EOL> self . consumer . store = None <EOL> self . consumer . _processCheckAuthResponse = ( <EOL> lambda response , server_url : True ) <EOL> self . consumer . _makeKVPost = lambda args , server_url : { } <EOL> self . consumer . _idResCheckSignature ( self . message , <EOL> self . endpoint . server_url ) <EOL> def test_statelessRaisesError_noStore ( self ) : <EOL> self . message . setArg ( OPENID_NS , "<STR_LIT>" , "<STR_LIT>" ) <EOL> self . consumer . _checkAuth = lambda unused1 , unused2 : False <EOL> self . consumer . store = None <EOL> self . assertRaises ( <EOL> ProtocolError , self . consumer . _idResCheckSignature , <EOL> self . message , self . endpoint . server_url ) <EOL> class TestQueryFormat ( TestIdRes ) : <EOL> def test_notAList ( self ) : <EOL> query = { '<STR_LIT>' : [ '<STR_LIT>' ] } <EOL> try : <EOL> r = Message . fromPostArgs ( query ) <EOL> except TypeError as err : <EOL> self . assertTrue ( str ( err ) . find ( '<STR_LIT>' ) != - <NUM_LIT:1> , err ) <EOL> else : <EOL> self . fail ( "<STR_LIT>" % ( r , ) ) <EOL> class TestComplete ( TestIdRes ) : <EOL> """<STR_LIT>""" <EOL> def test_setupNeededIdRes ( self ) : <EOL> message = Message . fromOpenIDArgs ( { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> setup_url_sentinel = object ( ) <EOL> def raiseSetupNeeded ( msg ) : <EOL> self . assertTrue ( msg is message ) <EOL> raise SetupNeededError ( setup_url_sentinel ) <EOL> self . consumer . _checkSetupNeeded = raiseSetupNeeded <EOL> response = self . consumer . complete ( message , None , None ) <EOL> self . assertEqual ( SETUP_NEEDED , response . status ) <EOL> self . assertTrue ( setup_url_sentinel is response . setup_url ) <EOL> def test_cancel ( self ) : <EOL> message = Message . fromPostArgs ( { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> self . disableReturnToChecking ( ) <EOL> r = self . consumer . complete ( message , self . endpoint ) <EOL> self . assertEqual ( r . status , CANCEL ) <EOL> self . assertTrue ( r . identity_url == self . endpoint . claimed_id ) <EOL> def test_cancel_with_return_to ( self ) : <EOL> message = Message . fromPostArgs ( { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> r = self . consumer . complete ( message , self . endpoint , self . return_to ) <EOL> self . assertEqual ( r . status , CANCEL ) <EOL> self . assertTrue ( r . identity_url == self . endpoint . claimed_id ) <EOL> def test_error ( self ) : <EOL> msg = '<STR_LIT>' <EOL> message = Message . fromPostArgs ( { '<STR_LIT>' : '<STR_LIT:error>' , <EOL> '<STR_LIT>' : msg , <EOL> } ) <EOL> self . disableReturnToChecking ( ) <EOL> r = self . consumer . complete ( message , self . endpoint ) <EOL> self . assertEqual ( r . status , FAILURE ) <EOL> self . assertTrue ( r . identity_url == self . endpoint . claimed_id ) <EOL> self . assertEqual ( r . message , msg ) <EOL> def test_errorWithNoOptionalKeys ( self ) : <EOL> msg = '<STR_LIT>' <EOL> contact = '<STR_LIT>' <EOL> message = Message . fromPostArgs ( { '<STR_LIT>' : '<STR_LIT:error>' , <EOL> '<STR_LIT>' : msg , <EOL> '<STR_LIT>' : contact , <EOL> } ) <EOL> self . disableReturnToChecking ( ) <EOL> r = self . consumer . complete ( message , self . endpoint ) <EOL> self . assertEqual ( r . status , FAILURE ) <EOL> self . assertTrue ( r . identity_url == self . endpoint . claimed_id ) <EOL> self . assertTrue ( r . contact == contact ) <EOL> self . assertTrue ( r . reference is None ) <EOL> self . assertEqual ( r . message , msg ) <EOL> def test_errorWithOptionalKeys ( self ) : <EOL> msg = '<STR_LIT>' <EOL> contact = '<STR_LIT>' <EOL> reference = '<STR_LIT>' <EOL> message = Message . fromPostArgs ( { '<STR_LIT>' : '<STR_LIT:error>' , <EOL> '<STR_LIT>' : msg , '<STR_LIT>' : reference , <EOL> '<STR_LIT>' : contact , '<STR_LIT>' : OPENID2_NS , <EOL> } ) <EOL> r = self . consumer . complete ( message , self . endpoint , None ) <EOL> self . assertEqual ( r . status , FAILURE ) <EOL> self . assertTrue ( r . identity_url == self . endpoint . claimed_id ) <EOL> self . assertTrue ( r . contact == contact ) <EOL> self . assertTrue ( r . reference == reference ) <EOL> self . assertEqual ( r . message , msg ) <EOL> def test_noMode ( self ) : <EOL> message = Message . fromPostArgs ( { } ) <EOL> r = self . consumer . complete ( message , self . endpoint , None ) <EOL> self . assertEqual ( r . status , FAILURE ) <EOL> self . assertTrue ( r . identity_url == self . endpoint . claimed_id ) <EOL> def test_idResMissingField ( self ) : <EOL> message = Message . fromPostArgs ( { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> self . assertRaises ( ProtocolError , self . consumer . _doIdRes , <EOL> message , self . endpoint , None ) <EOL> def test_idResURLMismatch ( self ) : <EOL> class VerifiedError ( Exception ) : <EOL> pass <EOL> def discoverAndVerify ( claimed_id , _to_match_endpoints ) : <EOL> raise VerifiedError <EOL> self . consumer . _discoverAndVerify = discoverAndVerify <EOL> self . disableReturnToChecking ( ) <EOL> message = Message . fromPostArgs ( <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : GOODSIG , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } ) <EOL> self . consumer . store = GoodAssocStore ( ) <EOL> self . assertRaises ( VerifiedError , <EOL> self . consumer . complete , <EOL> message , self . endpoint ) <EOL> self . failUnlessLogMatches ( '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> class TestCompleteMissingSig ( unittest . TestCase , CatchLogs ) : <EOL> def setUp ( self ) : <EOL> self . store = GoodAssocStore ( ) <EOL> self . consumer = GenericConsumer ( self . store ) <EOL> self . server_url = "<STR_LIT>" <EOL> CatchLogs . setUp ( self ) <EOL> claimed_id = '<STR_LIT>' <EOL> self . message = Message . fromOpenIDArgs ( <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : claimed_id , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : GOODSIG , <EOL> '<STR_LIT>' : mkNonce ( ) , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : claimed_id , <EOL> '<STR_LIT>' : self . server_url , <EOL> '<STR_LIT>' : OPENID2_NS , <EOL> } ) <EOL> self . endpoint = OpenIDServiceEndpoint ( ) <EOL> self . endpoint . server_url = self . server_url <EOL> self . endpoint . claimed_id = claimed_id <EOL> self . consumer . _checkReturnTo = lambda unused1 , unused2 : True <EOL> def tearDown ( self ) : <EOL> CatchLogs . tearDown ( self ) <EOL> def test_idResMissingNoSigs ( self ) : <EOL> def _vrfy ( resp_msg , endpoint = None ) : <EOL> return endpoint <EOL> self . consumer . _verifyDiscoveryResults = _vrfy <EOL> r = self . consumer . complete ( self . message , self . endpoint , None ) <EOL> self . failUnlessSuccess ( r ) <EOL> def test_idResNoIdentity ( self ) : <EOL> self . message . delArg ( OPENID_NS , '<STR_LIT>' ) <EOL> self . message . delArg ( OPENID_NS , '<STR_LIT>' ) <EOL> self . endpoint . claimed_id = None <EOL> self . message . setArg ( <EOL> OPENID_NS , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> r = self . consumer . complete ( self . message , self . endpoint , None ) <EOL> self . failUnlessSuccess ( r ) <EOL> def test_idResMissingIdentitySig ( self ) : <EOL> self . message . setArg ( <EOL> OPENID_NS , '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> r = self . consumer . complete ( self . message , self . endpoint , None ) <EOL> self . assertEqual ( r . status , FAILURE ) <EOL> def test_idResMissingReturnToSig ( self ) : <EOL> self . message . setArg ( <EOL> OPENID_NS , '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> r = self . consumer . complete ( self . message , self . endpoint , None ) <EOL> self . assertEqual ( r . status , FAILURE ) <EOL> def test_idResMissingAssocHandleSig ( self ) : <EOL> self . message . setArg ( <EOL> OPENID_NS , '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> r = self . consumer . complete ( self . message , self . endpoint , None ) <EOL> self . assertEqual ( r . status , FAILURE ) <EOL> def test_idResMissingClaimedIDSig ( self ) : <EOL> self . message . setArg ( <EOL> OPENID_NS , '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> r = self . consumer . complete ( self . message , self . endpoint , None ) <EOL> self . assertEqual ( r . status , FAILURE ) <EOL> def failUnlessSuccess ( self , response ) : <EOL> if response . status != SUCCESS : <EOL> self . fail ( "<STR_LIT>" % ( response , ) ) <EOL> class TestCheckAuthResponse ( TestIdRes , CatchLogs ) : <EOL> def setUp ( self ) : <EOL> CatchLogs . setUp ( self ) <EOL> TestIdRes . setUp ( self ) <EOL> def tearDown ( self ) : <EOL> CatchLogs . tearDown ( self ) <EOL> def _createAssoc ( self ) : <EOL> issued = time . time ( ) <EOL> lifetime = <NUM_LIT:1000> <EOL> assoc = association . Association ( <EOL> '<STR_LIT>' , '<STR_LIT>' , issued , lifetime , '<STR_LIT>' ) <EOL> store = self . consumer . store <EOL> store . storeAssociation ( self . server_url , assoc ) <EOL> assoc2 = store . getAssociation ( self . server_url ) <EOL> self . assertEqual ( assoc , assoc2 ) <EOL> def test_goodResponse ( self ) : <EOL> """<STR_LIT>""" <EOL> response = Message . fromOpenIDArgs ( { '<STR_LIT>' : '<STR_LIT:true>' } ) <EOL> r = self . consumer . _processCheckAuthResponse ( response , self . server_url ) <EOL> self . assertTrue ( r ) <EOL> def test_missingAnswer ( self ) : <EOL> """<STR_LIT>""" <EOL> response = Message . fromOpenIDArgs ( { } ) <EOL> r = self . consumer . _processCheckAuthResponse ( response , self . server_url ) <EOL> self . assertFalse ( r ) <EOL> def test_badResponse ( self ) : <EOL> """<STR_LIT>""" <EOL> response = Message . fromOpenIDArgs ( { '<STR_LIT>' : '<STR_LIT:false>' } ) <EOL> r = self . consumer . _processCheckAuthResponse ( response , self . server_url ) <EOL> self . assertFalse ( r ) <EOL> def test_badResponseInvalidate ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _createAssoc ( ) <EOL> response = Message . fromOpenIDArgs ( { <EOL> '<STR_LIT>' : '<STR_LIT:false>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } ) <EOL> r = self . consumer . _processCheckAuthResponse ( response , self . server_url ) <EOL> self . assertFalse ( r ) <EOL> self . assertTrue ( <EOL> self . consumer . store . getAssociation ( self . server_url ) is None ) <EOL> def test_invalidateMissing ( self ) : <EOL> """<STR_LIT>""" <EOL> response = Message . fromOpenIDArgs ( { <EOL> '<STR_LIT>' : '<STR_LIT:true>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } ) <EOL> r = self . consumer . _processCheckAuthResponse ( response , self . server_url ) <EOL> self . assertTrue ( r ) <EOL> self . failUnlessLogMatches ( <EOL> '<STR_LIT>' <EOL> ) <EOL> def test_invalidateMissing_noStore ( self ) : <EOL> """<STR_LIT>""" <EOL> response = Message . fromOpenIDArgs ( { <EOL> '<STR_LIT>' : '<STR_LIT:true>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } ) <EOL> self . consumer . store = None <EOL> r = self . consumer . _processCheckAuthResponse ( response , self . server_url ) <EOL> self . assertTrue ( r ) <EOL> self . failUnlessLogMatches ( <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> def test_invalidatePresent ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _createAssoc ( ) <EOL> response = Message . fromOpenIDArgs ( { <EOL> '<STR_LIT>' : '<STR_LIT:true>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } ) <EOL> r = self . consumer . _processCheckAuthResponse ( response , self . server_url ) <EOL> self . assertTrue ( r ) <EOL> self . assertTrue ( <EOL> self . consumer . store . getAssociation ( self . server_url ) is None ) <EOL> class TestSetupNeeded ( TestIdRes ) : <EOL> def failUnlessSetupNeeded ( self , expected_setup_url , message ) : <EOL> try : <EOL> self . consumer . _checkSetupNeeded ( message ) <EOL> except SetupNeededError as why : <EOL> self . assertEqual ( expected_setup_url , why . user_setup_url ) <EOL> else : <EOL> self . fail ( "<STR_LIT>" ) <EOL> def test_setupNeededOpenID1 ( self ) : <EOL> """<STR_LIT>""" <EOL> setup_url = '<STR_LIT>' <EOL> message = Message . fromPostArgs ( { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : setup_url , <EOL> } ) <EOL> self . assertTrue ( message . isOpenID1 ( ) ) <EOL> self . failUnlessSetupNeeded ( setup_url , message ) <EOL> def test_setupNeededOpenID1_extra ( self ) : <EOL> """<STR_LIT>""" <EOL> setup_url = '<STR_LIT>' <EOL> message = Message . fromPostArgs ( { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : setup_url , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } ) <EOL> self . assertTrue ( message . isOpenID1 ( ) ) <EOL> self . failUnlessSetupNeeded ( setup_url , message ) <EOL> def test_noSetupNeededOpenID1 ( self ) : <EOL> """<STR_LIT>""" <EOL> message = Message . fromOpenIDArgs ( { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> self . assertTrue ( message . isOpenID1 ( ) ) <EOL> self . consumer . _checkSetupNeeded ( message ) <EOL> def test_setupNeededOpenID2 ( self ) : <EOL> message = Message . fromOpenIDArgs ( { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : OPENID2_NS , <EOL> } ) <EOL> self . assertTrue ( message . isOpenID2 ( ) ) <EOL> response = self . consumer . complete ( message , None , None ) <EOL> self . assertEqual ( '<STR_LIT>' , response . status ) <EOL> self . assertEqual ( None , response . setup_url ) <EOL> def test_setupNeededDoesntWorkForOpenID1 ( self ) : <EOL> message = Message . fromOpenIDArgs ( { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } ) <EOL> self . consumer . _checkSetupNeeded ( message ) <EOL> response = self . consumer . complete ( message , None , None ) <EOL> self . assertEqual ( '<STR_LIT>' , response . status ) <EOL> self . assertTrue ( response . message . startswith ( '<STR_LIT>' ) ) <EOL> def test_noSetupNeededOpenID2 ( self ) : <EOL> message = Message . fromOpenIDArgs ( { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : OPENID2_NS , <EOL> } ) <EOL> self . assertTrue ( message . isOpenID2 ( ) ) <EOL> self . consumer . _checkSetupNeeded ( message ) <EOL> class IdResCheckForFieldsTest ( TestIdRes ) : <EOL> def setUp ( self ) : <EOL> self . consumer = GenericConsumer ( None ) <EOL> def mkSuccessTest ( openid_args , signed_list ) : <EOL> def test ( self ) : <EOL> message = Message . fromOpenIDArgs ( openid_args ) <EOL> message . setArg ( OPENID_NS , '<STR_LIT>' , '<STR_LIT:U+002C>' . join ( signed_list ) ) <EOL> self . consumer . _idResCheckForFields ( message ) <EOL> return test <EOL> test_openid1Success = mkSuccessTest ( <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } , <EOL> [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> test_openid2Success = mkSuccessTest ( <EOL> { '<STR_LIT>' : OPENID2_NS , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } , <EOL> [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> test_openid2Success_identifiers = mkSuccessTest ( <EOL> { '<STR_LIT>' : OPENID2_NS , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } , <EOL> [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> def mkMissingFieldTest ( openid_args ) : <EOL> def test ( self ) : <EOL> message = Message . fromOpenIDArgs ( openid_args ) <EOL> try : <EOL> self . consumer . _idResCheckForFields ( message ) <EOL> except ProtocolError as why : <EOL> self . assertTrue ( str ( why ) . startswith ( '<STR_LIT>' ) ) <EOL> else : <EOL> self . fail ( '<STR_LIT>' ) <EOL> return test <EOL> def mkMissingSignedTest ( openid_args ) : <EOL> def test ( self ) : <EOL> message = Message . fromOpenIDArgs ( openid_args ) <EOL> try : <EOL> self . consumer . _idResCheckForFields ( message ) <EOL> except ProtocolError as why : <EOL> self . assertTrue ( str ( why ) . endswith ( '<STR_LIT>' ) ) <EOL> else : <EOL> self . fail ( '<STR_LIT>' ) <EOL> return test <EOL> test_openid1Missing_returnToSig = mkMissingSignedTest ( <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } ) <EOL> test_openid1Missing_identitySig = mkMissingSignedTest ( <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } ) <EOL> test_openid2Missing_opEndpointSig = mkMissingSignedTest ( <EOL> { '<STR_LIT>' : OPENID2_NS , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } ) <EOL> test_openid1MissingReturnTo = mkMissingFieldTest ( <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } ) <EOL> test_openid1MissingAssocHandle = mkMissingFieldTest ( <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } ) <EOL> class CheckAuthHappened ( Exception ) : <EOL> pass <EOL> class CheckNonceVerifyTest ( TestIdRes , CatchLogs ) : <EOL> def setUp ( self ) : <EOL> CatchLogs . setUp ( self ) <EOL> TestIdRes . setUp ( self ) <EOL> self . consumer . openid1_nonce_query_arg_name = '<STR_LIT>' <EOL> def tearDown ( self ) : <EOL> CatchLogs . tearDown ( self ) <EOL> def test_openid1Success ( self ) : <EOL> """<STR_LIT>""" <EOL> nonce_value = mkNonce ( ) <EOL> self . return_to = '<STR_LIT>' % ( nonce_value , ) <EOL> self . response = Message . fromOpenIDArgs ( { '<STR_LIT>' : self . return_to } ) <EOL> self . response . setArg ( BARE_NS , '<STR_LIT>' , nonce_value ) <EOL> self . consumer . _idResCheckNonce ( self . response , self . endpoint ) <EOL> self . failUnlessLogEmpty ( ) <EOL> def test_openid1Missing ( self ) : <EOL> """<STR_LIT>""" <EOL> self . response = Message . fromOpenIDArgs ( { } ) <EOL> n = self . consumer . _idResGetNonceOpenID1 ( self . response , self . endpoint ) <EOL> self . assertTrue ( n is None , n ) <EOL> self . failUnlessLogEmpty ( ) <EOL> def test_consumerNonceOpenID2 ( self ) : <EOL> """<STR_LIT>""" <EOL> self . return_to = '<STR_LIT>' % ( mkNonce ( ) , ) <EOL> self . response = Message . fromOpenIDArgs ( <EOL> { '<STR_LIT>' : self . return_to , '<STR_LIT>' : OPENID2_NS } ) <EOL> self . assertRaises ( ProtocolError , self . consumer . _idResCheckNonce , <EOL> self . response , self . endpoint ) <EOL> self . failUnlessLogEmpty ( ) <EOL> def test_serverNonce ( self ) : <EOL> """<STR_LIT>""" <EOL> self . response = Message . fromOpenIDArgs ( <EOL> { '<STR_LIT>' : OPENID2_NS , '<STR_LIT>' : mkNonce ( ) } ) <EOL> self . consumer . _idResCheckNonce ( self . response , self . endpoint ) <EOL> self . failUnlessLogEmpty ( ) <EOL> def test_serverNonceOpenID1 ( self ) : <EOL> """<STR_LIT>""" <EOL> self . response = Message . fromOpenIDArgs ( <EOL> { '<STR_LIT>' : OPENID1_NS , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : mkNonce ( ) } ) <EOL> self . assertRaises ( ProtocolError , self . consumer . _idResCheckNonce , <EOL> self . response , self . endpoint ) <EOL> self . failUnlessLogEmpty ( ) <EOL> def test_badNonce ( self ) : <EOL> """<STR_LIT>""" <EOL> nonce = mkNonce ( ) <EOL> stamp , salt = splitNonce ( nonce ) <EOL> self . store . useNonce ( self . server_url , stamp , salt ) <EOL> self . response = Message . fromOpenIDArgs ( <EOL> { '<STR_LIT>' : nonce , <EOL> '<STR_LIT>' : OPENID2_NS , <EOL> } ) <EOL> self . assertRaises ( ProtocolError , self . consumer . _idResCheckNonce , <EOL> self . response , self . endpoint ) <EOL> def test_successWithNoStore ( self ) : <EOL> """<STR_LIT>""" <EOL> self . consumer . store = None <EOL> self . response = Message . fromOpenIDArgs ( <EOL> { '<STR_LIT>' : mkNonce ( ) , <EOL> '<STR_LIT>' : OPENID2_NS , <EOL> } ) <EOL> self . consumer . _idResCheckNonce ( self . response , self . endpoint ) <EOL> self . failUnlessLogEmpty ( ) <EOL> def test_tamperedNonce ( self ) : <EOL> """<STR_LIT>""" <EOL> self . response = Message . fromOpenIDArgs ( <EOL> { '<STR_LIT>' : OPENID2_NS , <EOL> '<STR_LIT>' : '<STR_LIT>' } ) <EOL> self . assertRaises ( ProtocolError , self . consumer . _idResCheckNonce , <EOL> self . response , self . endpoint ) <EOL> def test_missingNonce ( self ) : <EOL> """<STR_LIT>""" <EOL> self . response = Message . fromOpenIDArgs ( <EOL> { '<STR_LIT>' : self . return_to } ) <EOL> self . assertRaises ( ProtocolError , self . consumer . _idResCheckNonce , <EOL> self . response , self . endpoint ) <EOL> class CheckAuthDetectingConsumer ( GenericConsumer ) : <EOL> def _checkAuth ( self , * args ) : <EOL> raise CheckAuthHappened ( args ) <EOL> def _idResCheckNonce ( self , * args ) : <EOL> """<STR_LIT>""" <EOL> return True <EOL> class TestCheckAuthTriggered ( TestIdRes , CatchLogs ) : <EOL> consumer_class = CheckAuthDetectingConsumer <EOL> def setUp ( self ) : <EOL> TestIdRes . setUp ( self ) <EOL> CatchLogs . setUp ( self ) <EOL> self . disableDiscoveryVerification ( ) <EOL> def test_checkAuthTriggered ( self ) : <EOL> message = Message . fromPostArgs ( { <EOL> '<STR_LIT>' : self . return_to , <EOL> '<STR_LIT>' : self . server_id , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : GOODSIG , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } ) <EOL> self . disableReturnToChecking ( ) <EOL> try : <EOL> result = self . consumer . _doIdRes ( message , self . endpoint , None ) <EOL> except CheckAuthHappened : <EOL> pass <EOL> else : <EOL> self . fail ( '<STR_LIT>' % <EOL> ( result , self . messages ) ) <EOL> def test_checkAuthTriggeredWithAssoc ( self ) : <EOL> issued = time . time ( ) <EOL> lifetime = <NUM_LIT:1000> <EOL> assoc = association . Association ( <EOL> '<STR_LIT>' , '<STR_LIT>' , issued , lifetime , '<STR_LIT>' ) <EOL> self . store . storeAssociation ( self . server_url , assoc ) <EOL> self . disableReturnToChecking ( ) <EOL> message = Message . fromPostArgs ( { <EOL> '<STR_LIT>' : self . return_to , <EOL> '<STR_LIT>' : self . server_id , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : GOODSIG , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } ) <EOL> try : <EOL> result = self . consumer . _doIdRes ( message , self . endpoint , None ) <EOL> except CheckAuthHappened : <EOL> pass <EOL> else : <EOL> self . fail ( '<STR_LIT>' % ( result , ) ) <EOL> def test_expiredAssoc ( self ) : <EOL> issued = time . time ( ) - <NUM_LIT:10> <EOL> lifetime = <NUM_LIT:0> <EOL> handle = '<STR_LIT>' <EOL> assoc = association . Association ( <EOL> handle , '<STR_LIT>' , issued , lifetime , '<STR_LIT>' ) <EOL> self . assertTrue ( assoc . expiresIn <= <NUM_LIT:0> ) <EOL> self . store . storeAssociation ( self . server_url , assoc ) <EOL> message = Message . fromPostArgs ( { <EOL> '<STR_LIT>' : self . return_to , <EOL> '<STR_LIT>' : self . server_id , <EOL> '<STR_LIT>' : handle , <EOL> '<STR_LIT>' : GOODSIG , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } ) <EOL> self . disableReturnToChecking ( ) <EOL> self . assertRaises ( ProtocolError , self . consumer . _doIdRes , <EOL> message , self . endpoint , None ) <EOL> def test_newerAssoc ( self ) : <EOL> lifetime = <NUM_LIT:1000> <EOL> good_issued = time . time ( ) - <NUM_LIT:10> <EOL> good_handle = '<STR_LIT>' <EOL> good_assoc = association . Association ( <EOL> good_handle , '<STR_LIT>' , good_issued , lifetime , '<STR_LIT>' ) <EOL> self . store . storeAssociation ( self . server_url , good_assoc ) <EOL> bad_issued = time . time ( ) - <NUM_LIT:5> <EOL> bad_handle = '<STR_LIT>' <EOL> bad_assoc = association . Association ( <EOL> bad_handle , '<STR_LIT>' , bad_issued , lifetime , '<STR_LIT>' ) <EOL> self . store . storeAssociation ( self . server_url , bad_assoc ) <EOL> query = { <EOL> '<STR_LIT>' : self . return_to , <EOL> '<STR_LIT>' : self . server_id , <EOL> '<STR_LIT>' : good_handle , <EOL> } <EOL> message = Message . fromOpenIDArgs ( query ) <EOL> message = good_assoc . signMessage ( message ) <EOL> self . disableReturnToChecking ( ) <EOL> info = self . consumer . _doIdRes ( message , self . endpoint , None ) <EOL> self . assertEqual ( info . status , SUCCESS , info . message ) <EOL> self . assertEqual ( self . consumer_id , info . identity_url ) <EOL> class TestReturnToArgs ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> store = object ( ) <EOL> self . consumer = GenericConsumer ( store ) <EOL> def test_returnToArgsOkay ( self ) : <EOL> query = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:foo>' : '<STR_LIT:bar>' , <EOL> } <EOL> self . consumer . _verifyReturnToArgs ( query ) <EOL> def test_returnToArgsUnexpectedArg ( self ) : <EOL> query = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:foo>' : '<STR_LIT:bar>' , <EOL> } <EOL> self . assertRaises ( ProtocolError , <EOL> self . consumer . _verifyReturnToArgs , query ) <EOL> def test_returnToMismatch ( self ) : <EOL> query = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> self . assertRaises ( ValueError , <EOL> self . consumer . _verifyReturnToArgs , query ) <EOL> query [ '<STR_LIT:foo>' ] = '<STR_LIT>' <EOL> self . assertRaises ( ValueError , <EOL> self . consumer . _verifyReturnToArgs , query ) <EOL> def test_noReturnTo ( self ) : <EOL> query = { '<STR_LIT>' : '<STR_LIT>' } <EOL> self . assertRaises ( ValueError , <EOL> self . consumer . _verifyReturnToArgs , query ) <EOL> def test_completeBadReturnTo ( self ) : <EOL> """<STR_LIT>""" <EOL> return_to = "<STR_LIT>" <EOL> bad_return_tos = [ <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> ] <EOL> m = Message ( OPENID1_NS ) <EOL> m . setArg ( OPENID_NS , '<STR_LIT>' , '<STR_LIT>' ) <EOL> m . setArg ( BARE_NS , '<STR_LIT:foo>' , '<STR_LIT:bar>' ) <EOL> endpoint = None <EOL> for bad in bad_return_tos : <EOL> m . setArg ( OPENID_NS , '<STR_LIT>' , bad ) <EOL> self . assertFalse ( self . consumer . _checkReturnTo ( m , return_to ) ) <EOL> def test_completeGoodReturnTo ( self ) : <EOL> """<STR_LIT>""" <EOL> return_to = "<STR_LIT>" <EOL> good_return_tos = [ <EOL> ( return_to , { } ) , <EOL> ( return_to + "<STR_LIT>" , { ( BARE_NS , '<STR_LIT>' ) : '<STR_LIT>' } ) , <EOL> ( return_to + "<STR_LIT>" , <EOL> { ( BARE_NS , '<STR_LIT>' ) : '<STR_LIT>' } ) , <EOL> ( "<STR_LIT>" + return_to [ <NUM_LIT:4> : ] , { } ) , <EOL> ( return_to . replace ( '<STR_LIT:url>' , '<STR_LIT>' ) , { } ) , <EOL> ( "<STR_LIT>" , { } ) , <EOL> ( "<STR_LIT>" , { } ) , <EOL> ( "<STR_LIT>" , { } ) , <EOL> ] <EOL> endpoint = None <EOL> for good , extra in good_return_tos : <EOL> m = Message ( OPENID1_NS ) <EOL> m . setArg ( OPENID_NS , '<STR_LIT>' , '<STR_LIT>' ) <EOL> for ns , key in extra : <EOL> m . setArg ( ns , key , extra [ ( ns , key ) ] ) <EOL> m . setArg ( OPENID_NS , '<STR_LIT>' , good ) <EOL> result = self . consumer . complete ( m , endpoint , return_to ) <EOL> self . assertTrue ( <EOL> isinstance ( result , CancelResponse ) , <EOL> "<STR_LIT>" % ( result , good ) ) <EOL> class MockFetcher ( object ) : <EOL> def __init__ ( self , response = None ) : <EOL> self . response = response or HTTPResponse ( ) <EOL> self . fetches = [ ] <EOL> def fetch ( self , url , body = None , headers = None ) : <EOL> self . fetches . append ( ( url , body , headers ) ) <EOL> return self . response <EOL> class ExceptionRaisingMockFetcher ( object ) : <EOL> class MyException ( Exception ) : <EOL> pass <EOL> def fetch ( self , url , body = None , headers = None ) : <EOL> raise self . MyException ( '<STR_LIT>' ) <EOL> class BadArgCheckingConsumer ( GenericConsumer ) : <EOL> def _makeKVPost ( self , args , _ ) : <EOL> assert args == { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:foo>' , <EOL> '<STR_LIT>' : OPENID1_NS <EOL> } , args <EOL> return None <EOL> class TestCheckAuth ( unittest . TestCase , CatchLogs ) : <EOL> consumer_class = GenericConsumer <EOL> def setUp ( self ) : <EOL> CatchLogs . setUp ( self ) <EOL> self . store = memstore . MemoryStore ( ) <EOL> self . consumer = self . consumer_class ( self . store ) <EOL> self . _orig_fetcher = fetchers . getDefaultFetcher ( ) <EOL> self . fetcher = MockFetcher ( ) <EOL> fetchers . setDefaultFetcher ( self . fetcher ) <EOL> def tearDown ( self ) : <EOL> CatchLogs . tearDown ( self ) <EOL> fetchers . setDefaultFetcher ( self . _orig_fetcher , wrap_exceptions = False ) <EOL> def test_error ( self ) : <EOL> self . fetcher . response = HTTPResponse ( <EOL> "<STR_LIT>" , <NUM_LIT> , { '<STR_LIT>' : '<STR_LIT>' } , '<STR_LIT>' ) <EOL> query = { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> r = self . consumer . _checkAuth ( Message . fromPostArgs ( query ) , <EOL> http_server_url ) <EOL> self . assertFalse ( r ) <EOL> self . assertTrue ( self . messages ) <EOL> def test_bad_args ( self ) : <EOL> query = { <EOL> '<STR_LIT>' : '<STR_LIT:foo>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> consumer = BadArgCheckingConsumer ( self . store ) <EOL> consumer . _checkAuth ( Message . fromPostArgs ( query ) , '<STR_LIT>' ) <EOL> def test_signedList ( self ) : <EOL> query = Message . fromOpenIDArgs ( { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:foo>' : '<STR_LIT:bar>' , <EOL> } ) <EOL> args = self . consumer . _createCheckAuthRequest ( query ) <EOL> self . assertTrue ( args . isOpenID1 ( ) ) <EOL> for signed_arg in query . getArg ( OPENID_NS , '<STR_LIT>' ) . split ( '<STR_LIT:U+002C>' ) : <EOL> self . assertTrue ( args . getAliasedArg ( signed_arg ) , signed_arg ) <EOL> def test_112 ( self ) : <EOL> args = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:none>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:0>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } <EOL> self . assertEqual ( OPENID2_NS , args [ '<STR_LIT>' ] ) <EOL> incoming = Message . fromPostArgs ( args ) <EOL> self . assertTrue ( incoming . isOpenID2 ( ) ) <EOL> car = self . consumer . _createCheckAuthRequest ( incoming ) <EOL> expected_args = args . copy ( ) <EOL> expected_args [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> expected = Message . fromPostArgs ( expected_args ) <EOL> self . assertTrue ( expected . isOpenID2 ( ) ) <EOL> self . assertEqual ( expected , car ) <EOL> car_args = car . toPostArgs ( ) <EOL> self . assertEqual ( set ( expected_args . keys ( ) ) , set ( car_args . keys ( ) ) ) <EOL> self . assertEqual ( set ( expected_args . values ( ) ) , set ( car_args . values ( ) ) ) <EOL> self . assertEqual ( expected_args , car . toPostArgs ( ) ) <EOL> class TestFetchAssoc ( unittest . TestCase , CatchLogs ) : <EOL> consumer_class = GenericConsumer <EOL> def setUp ( self ) : <EOL> CatchLogs . setUp ( self ) <EOL> self . store = memstore . MemoryStore ( ) <EOL> self . fetcher = MockFetcher ( ) <EOL> fetchers . setDefaultFetcher ( self . fetcher ) <EOL> self . consumer = self . consumer_class ( self . store ) <EOL> def test_error_404 ( self ) : <EOL> """<STR_LIT>""" <EOL> self . fetcher . response = HTTPResponse ( <EOL> "<STR_LIT>" , <NUM_LIT> , { '<STR_LIT>' : '<STR_LIT>' } , '<STR_LIT>' ) <EOL> self . assertRaises ( <EOL> fetchers . HTTPFetchingError , <EOL> self . consumer . _makeKVPost , <EOL> Message . fromPostArgs ( { '<STR_LIT>' : '<STR_LIT>' } ) , <EOL> "<STR_LIT>" ) <EOL> def test_error_exception_unwrapped ( self ) : <EOL> """<STR_LIT>""" <EOL> self . fetcher = ExceptionRaisingMockFetcher ( ) <EOL> fetchers . setDefaultFetcher ( self . fetcher , wrap_exceptions = False ) <EOL> self . assertRaises ( self . fetcher . MyException , <EOL> self . consumer . _makeKVPost , <EOL> Message . fromPostArgs ( { '<STR_LIT>' : '<STR_LIT>' } ) , <EOL> "<STR_LIT>" ) <EOL> e = OpenIDServiceEndpoint ( ) <EOL> e . server_url = '<STR_LIT>' <EOL> self . assertRaises ( self . fetcher . MyException , <EOL> self . consumer . _getAssociation , e ) <EOL> self . assertRaises ( self . fetcher . MyException , <EOL> self . consumer . _checkAuth , <EOL> Message . fromPostArgs ( { '<STR_LIT>' : '<STR_LIT>' } ) , <EOL> '<STR_LIT>' ) <EOL> def test_error_exception_wrapped ( self ) : <EOL> """<STR_LIT>""" <EOL> self . fetcher = ExceptionRaisingMockFetcher ( ) <EOL> fetchers . setDefaultFetcher ( self . fetcher ) <EOL> self . assertRaises ( fetchers . HTTPFetchingError , <EOL> self . consumer . _makeKVPost , <EOL> Message . fromOpenIDArgs ( { '<STR_LIT>' : '<STR_LIT>' } ) , <EOL> "<STR_LIT>" ) <EOL> e = OpenIDServiceEndpoint ( ) <EOL> e . server_url = '<STR_LIT>' <EOL> self . assertTrue ( self . consumer . _getAssociation ( e ) is None ) <EOL> msg = Message . fromPostArgs ( { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> self . assertFalse ( self . consumer . _checkAuth ( msg , '<STR_LIT>' ) ) <EOL> class TestSuccessResponse ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . endpoint = OpenIDServiceEndpoint ( ) <EOL> self . endpoint . claimed_id = '<STR_LIT>' <EOL> def test_extensionResponse ( self ) : <EOL> resp = mkSuccess ( self . endpoint , { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:1>' , <EOL> '<STR_LIT>' : '<STR_LIT:2>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } ) <EOL> utargs = resp . extensionResponse ( '<STR_LIT>' , False ) <EOL> self . assertEqual ( utargs , { '<STR_LIT>' : '<STR_LIT:1>' , '<STR_LIT>' : '<STR_LIT:2>' } ) <EOL> sregargs = resp . extensionResponse ( '<STR_LIT>' , False ) <EOL> self . assertEqual ( sregargs , { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> def test_extensionResponseSigned ( self ) : <EOL> args = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:1>' , <EOL> '<STR_LIT>' : '<STR_LIT:2>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> signed_list = [ '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ] <EOL> msg = Message . fromOpenIDArgs ( args ) <EOL> resp = SuccessResponse ( self . endpoint , msg , signed_list ) <EOL> sregargs = resp . extensionResponse ( '<STR_LIT>' , True ) <EOL> self . assertEqual ( sregargs , { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } ) <EOL> utargs = resp . extensionResponse ( '<STR_LIT>' , True ) <EOL> self . assertEqual ( utargs , None ) <EOL> def test_noReturnTo ( self ) : <EOL> resp = mkSuccess ( self . endpoint , { } ) <EOL> self . assertTrue ( resp . getReturnTo ( ) is None ) <EOL> def test_returnTo ( self ) : <EOL> resp = mkSuccess ( self . endpoint , { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> self . assertEqual ( resp . getReturnTo ( ) , '<STR_LIT>' ) <EOL> def test_displayIdentifierClaimedId ( self ) : <EOL> resp = mkSuccess ( self . endpoint , { } ) <EOL> self . assertEqual ( resp . getDisplayIdentifier ( ) , <EOL> resp . endpoint . claimed_id ) <EOL> def test_displayIdentifierOverride ( self ) : <EOL> self . endpoint . display_identifier = "<STR_LIT>" <EOL> resp = mkSuccess ( self . endpoint , { } ) <EOL> self . assertEqual ( resp . getDisplayIdentifier ( ) , <EOL> "<STR_LIT>" ) <EOL> class StubConsumer ( object ) : <EOL> def __init__ ( self ) : <EOL> self . assoc = object ( ) <EOL> self . response = None <EOL> self . endpoint = None <EOL> def begin ( self , service ) : <EOL> auth_req = AuthRequest ( service , self . assoc ) <EOL> self . endpoint = service <EOL> return auth_req <EOL> def complete ( self , message , endpoint , return_to ) : <EOL> assert endpoint is self . endpoint <EOL> return self . response <EOL> class ConsumerTest ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> self . endpoint = OpenIDServiceEndpoint ( ) <EOL> self . endpoint . claimed_id = self . identity_url = '<STR_LIT>' <EOL> self . store = None <EOL> self . session = { } <EOL> self . consumer = Consumer ( self . session , self . store ) <EOL> self . consumer . consumer = StubConsumer ( ) <EOL> self . discovery = Discovery ( self . session , <EOL> self . identity_url , <EOL> self . consumer . session_key_prefix ) <EOL> def test_setAssociationPreference ( self ) : <EOL> self . consumer . setAssociationPreference ( [ ] ) <EOL> self . assertTrue ( isinstance ( self . consumer . consumer . negotiator , <EOL> association . SessionNegotiator ) ) <EOL> self . assertEqual ( [ ] , <EOL> self . consumer . consumer . negotiator . allowed_types ) <EOL> self . consumer . setAssociationPreference ( [ ( '<STR_LIT>' , '<STR_LIT>' ) ] ) <EOL> self . assertEqual ( [ ( '<STR_LIT>' , '<STR_LIT>' ) ] , <EOL> self . consumer . consumer . negotiator . allowed_types ) <EOL> def withDummyDiscovery ( self , callable , dummy_getNextService ) : <EOL> class DummyDisco ( object ) : <EOL> def __init__ ( self , * ignored ) : <EOL> pass <EOL> getNextService = dummy_getNextService <EOL> import openid . consumer . consumer <EOL> old_discovery = openid . consumer . consumer . Discovery <EOL> try : <EOL> openid . consumer . consumer . Discovery = DummyDisco <EOL> callable ( ) <EOL> finally : <EOL> openid . consumer . consumer . Discovery = old_discovery <EOL> def test_beginHTTPError ( self ) : <EOL> """<STR_LIT>""" <EOL> def getNextService ( self , ignored ) : <EOL> raise HTTPFetchingError ( "<STR_LIT>" ) <EOL> def test ( ) : <EOL> try : <EOL> self . consumer . begin ( '<STR_LIT>' ) <EOL> except DiscoveryFailure as why : <EOL> self . assertTrue ( str ( why ) . startswith ( '<STR_LIT>' ) ) <EOL> self . assertFalse ( str ( why ) . find ( '<STR_LIT>' ) == - <NUM_LIT:1> ) <EOL> else : <EOL> self . fail ( '<STR_LIT>' ) <EOL> self . withDummyDiscovery ( test , getNextService ) <EOL> def test_beginNoServices ( self ) : <EOL> def getNextService ( self , ignored ) : <EOL> return None <EOL> url = '<STR_LIT>' <EOL> def test ( ) : <EOL> try : <EOL> self . consumer . begin ( url ) <EOL> except DiscoveryFailure as why : <EOL> self . assertTrue ( str ( why ) . startswith ( '<STR_LIT>' ) ) <EOL> self . assertFalse ( str ( why ) . find ( url ) == - <NUM_LIT:1> ) <EOL> else : <EOL> self . fail ( '<STR_LIT>' ) <EOL> self . withDummyDiscovery ( test , getNextService ) <EOL> def test_beginWithoutDiscovery ( self ) : <EOL> result = self . consumer . beginWithoutDiscovery ( self . endpoint ) <EOL> self . assertTrue ( isinstance ( result , AuthRequest ) ) <EOL> self . assertTrue ( <EOL> self . session [ self . consumer . _token_key ] is result . endpoint ) <EOL> self . assertTrue ( result . endpoint is self . endpoint ) <EOL> def test_completeEmptySession ( self ) : <EOL> text = "<STR_LIT>" <EOL> def checkEndpoint ( message , endpoint , return_to ) : <EOL> self . assertTrue ( endpoint is None ) <EOL> return FailureResponse ( endpoint , text ) <EOL> self . consumer . consumer . complete = checkEndpoint <EOL> response = self . consumer . complete ( { } , None ) <EOL> self . assertEqual ( response . status , FAILURE ) <EOL> self . assertEqual ( response . message , text ) <EOL> self . assertTrue ( response . identity_url is None ) <EOL> def _doResp ( self , auth_req , exp_resp ) : <EOL> """<STR_LIT>""" <EOL> self . consumer . consumer . response = exp_resp <EOL> self . assertTrue ( self . session ) <EOL> resp = self . consumer . complete ( { } , None ) <EOL> if self . endpoint . claimed_id != IDENTIFIER_SELECT : <EOL> self . assertTrue ( resp . identity_url is self . identity_url ) <EOL> self . assertFalse ( self . consumer . _token_key in self . session ) <EOL> self . assertEqual ( resp . status , exp_resp . status ) <EOL> return resp <EOL> def _doRespNoDisco ( self , exp_resp ) : <EOL> """<STR_LIT>""" <EOL> auth_req = self . consumer . beginWithoutDiscovery ( self . endpoint ) <EOL> resp = self . _doResp ( auth_req , exp_resp ) <EOL> self . assertFalse ( self . session ) <EOL> return resp <EOL> def test_noDiscoCompleteSuccessWithToken ( self ) : <EOL> self . _doRespNoDisco ( mkSuccess ( self . endpoint , { } ) ) <EOL> def test_noDiscoCompleteCancelWithToken ( self ) : <EOL> self . _doRespNoDisco ( CancelResponse ( self . endpoint ) ) <EOL> def test_noDiscoCompleteFailure ( self ) : <EOL> msg = '<STR_LIT>' <EOL> resp = self . _doRespNoDisco ( FailureResponse ( self . endpoint , msg ) ) <EOL> self . assertTrue ( resp . message is msg ) <EOL> def test_noDiscoCompleteSetupNeeded ( self ) : <EOL> setup_url = '<STR_LIT>' <EOL> resp = self . _doRespNoDisco ( <EOL> SetupNeededResponse ( self . endpoint , setup_url ) ) <EOL> self . assertTrue ( resp . setup_url is setup_url ) <EOL> def _doRespDisco ( self , is_clean , exp_resp ) : <EOL> """<STR_LIT>""" <EOL> self . discovery . createManager ( [ self . endpoint ] , self . identity_url ) <EOL> auth_req = self . consumer . begin ( self . identity_url ) <EOL> resp = self . _doResp ( auth_req , exp_resp ) <EOL> manager = self . discovery . getManager ( ) <EOL> if is_clean : <EOL> self . assertTrue ( self . discovery . getManager ( ) is None , manager ) <EOL> else : <EOL> self . assertFalse ( self . discovery . getManager ( ) is None , manager ) <EOL> return resp <EOL> def test_completeSuccess ( self ) : <EOL> self . _doRespDisco ( True , mkSuccess ( self . endpoint , { } ) ) <EOL> def test_completeCancel ( self ) : <EOL> self . _doRespDisco ( True , CancelResponse ( self . endpoint ) ) <EOL> def test_completeFailure ( self ) : <EOL> msg = '<STR_LIT>' <EOL> resp = self . _doRespDisco ( False , FailureResponse ( self . endpoint , msg ) ) <EOL> self . assertTrue ( resp . message is msg ) <EOL> def test_completeSetupNeeded ( self ) : <EOL> setup_url = '<STR_LIT>' <EOL> resp = self . _doRespDisco ( <EOL> False , <EOL> SetupNeededResponse ( self . endpoint , setup_url ) ) <EOL> self . assertTrue ( resp . setup_url is setup_url ) <EOL> def test_successDifferentURL ( self ) : <EOL> """<STR_LIT>""" <EOL> self . identity_url = '<STR_LIT>' <EOL> self . endpoint . claimed_id = self . endpoint . local_id = IDENTIFIER_SELECT <EOL> resp_endpoint = OpenIDServiceEndpoint ( ) <EOL> resp_endpoint . claimed_id = "<STR_LIT>" <EOL> resp = self . _doRespDisco ( True , mkSuccess ( resp_endpoint , { } ) ) <EOL> self . assertTrue ( self . discovery . getManager ( force = True ) is None ) <EOL> def test_begin ( self ) : <EOL> self . discovery . createManager ( [ self . endpoint ] , self . identity_url ) <EOL> auth_req = self . consumer . begin ( self . identity_url ) <EOL> self . assertTrue ( isinstance ( auth_req , AuthRequest ) ) <EOL> self . assertTrue ( auth_req . endpoint is self . endpoint ) <EOL> self . assertTrue ( auth_req . endpoint is self . consumer . consumer . endpoint ) <EOL> self . assertTrue ( auth_req . assoc is self . consumer . consumer . assoc ) <EOL> class IDPDrivenTest ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . store = GoodAssocStore ( ) <EOL> self . consumer = GenericConsumer ( self . store ) <EOL> self . endpoint = OpenIDServiceEndpoint ( ) <EOL> self . endpoint . server_url = "<STR_LIT>" <EOL> def test_idpDrivenBegin ( self ) : <EOL> self . consumer . begin ( self . endpoint ) <EOL> def test_idpDrivenComplete ( self ) : <EOL> identifier = '<STR_LIT>' <EOL> message = Message . fromPostArgs ( { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:x>' , <EOL> '<STR_LIT>' : '<STR_LIT:z>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : GOODSIG , <EOL> } ) <EOL> discovered_endpoint = OpenIDServiceEndpoint ( ) <EOL> discovered_endpoint . claimed_id = identifier <EOL> discovered_endpoint . server_url = self . endpoint . server_url <EOL> discovered_endpoint . local_id = identifier <EOL> iverified = [ ] <EOL> def verifyDiscoveryResults ( identifier , endpoint ) : <EOL> self . assertTrue ( endpoint is self . endpoint ) <EOL> iverified . append ( discovered_endpoint ) <EOL> return discovered_endpoint <EOL> self . consumer . _verifyDiscoveryResults = verifyDiscoveryResults <EOL> self . consumer . _idResCheckNonce = lambda * args : True <EOL> self . consumer . _checkReturnTo = lambda unused1 , unused2 : True <EOL> response = self . consumer . _doIdRes ( message , self . endpoint , None ) <EOL> self . failUnlessSuccess ( response ) <EOL> self . assertEqual ( response . identity_url , "<STR_LIT>" ) <EOL> self . assertEqual ( iverified , [ discovered_endpoint ] ) <EOL> def test_idpDrivenCompleteFraud ( self ) : <EOL> message = Message . fromPostArgs ( { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:x>' , <EOL> '<STR_LIT>' : '<STR_LIT:z>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : GOODSIG , <EOL> } ) <EOL> def verifyDiscoveryResults ( identifier , endpoint ) : <EOL> raise DiscoveryFailure ( "<STR_LIT>" , None ) <EOL> self . consumer . _verifyDiscoveryResults = verifyDiscoveryResults <EOL> self . consumer . _checkReturnTo = lambda unused1 , unused2 : True <EOL> self . assertRaises ( DiscoveryFailure , self . consumer . _doIdRes , <EOL> message , self . endpoint , None ) <EOL> def failUnlessSuccess ( self , response ) : <EOL> if response . status != SUCCESS : <EOL> self . fail ( "<STR_LIT>" % ( response , ) ) <EOL> class TestDiscoveryVerification ( unittest . TestCase ) : <EOL> services = [ ] <EOL> def setUp ( self ) : <EOL> self . store = GoodAssocStore ( ) <EOL> self . consumer = GenericConsumer ( self . store ) <EOL> self . consumer . _discover = self . discoveryFunc <EOL> self . identifier = "<STR_LIT>" <EOL> self . server_url = "<STR_LIT>" <EOL> self . message = Message . fromPostArgs ( { <EOL> '<STR_LIT>' : OPENID2_NS , <EOL> '<STR_LIT>' : self . identifier , <EOL> '<STR_LIT>' : self . identifier , <EOL> '<STR_LIT>' : self . server_url , <EOL> } ) <EOL> self . endpoint = OpenIDServiceEndpoint ( ) <EOL> self . endpoint . server_url = self . server_url <EOL> def test_theGoodStuff ( self ) : <EOL> endpoint = OpenIDServiceEndpoint ( ) <EOL> endpoint . type_uris = [ OPENID_2_0_TYPE ] <EOL> endpoint . claimed_id = self . identifier <EOL> endpoint . server_url = self . server_url <EOL> endpoint . local_id = self . identifier <EOL> self . services = [ endpoint ] <EOL> r = self . consumer . _verifyDiscoveryResults ( self . message , endpoint ) <EOL> self . assertEqual ( r , endpoint ) <EOL> def test_otherServer ( self ) : <EOL> text = "<STR_LIT>" <EOL> def discoverAndVerify ( claimed_id , to_match_endpoints ) : <EOL> self . assertEqual ( claimed_id , self . identifier ) <EOL> for to_match in to_match_endpoints : <EOL> self . assertEqual ( claimed_id , to_match . claimed_id ) <EOL> raise ProtocolError ( text ) <EOL> self . consumer . _discoverAndVerify = discoverAndVerify <EOL> endpoint = OpenIDServiceEndpoint ( ) <EOL> endpoint . type_uris = [ OPENID_2_0_TYPE ] <EOL> endpoint . claimed_id = self . identifier <EOL> endpoint . server_url = "<STR_LIT>" <EOL> endpoint . local_id = self . identifier <EOL> self . services = [ endpoint ] <EOL> try : <EOL> r = self . consumer . _verifyDiscoveryResults ( self . message , endpoint ) <EOL> except ProtocolError as e : <EOL> self . assertTrue ( str ( e ) , text ) <EOL> else : <EOL> self . fail ( "<STR_LIT>" % ( r , ) ) <EOL> def test_foreignDelegate ( self ) : <EOL> text = "<STR_LIT>" <EOL> def discoverAndVerify ( claimed_id , to_match_endpoints ) : <EOL> self . assertEqual ( claimed_id , self . identifier ) <EOL> for to_match in to_match_endpoints : <EOL> self . assertEqual ( claimed_id , to_match . claimed_id ) <EOL> raise ProtocolError ( text ) <EOL> self . consumer . _discoverAndVerify = discoverAndVerify <EOL> endpoint = OpenIDServiceEndpoint ( ) <EOL> endpoint . type_uris = [ OPENID_2_0_TYPE ] <EOL> endpoint . claimed_id = self . identifier <EOL> endpoint . server_url = self . server_url <EOL> endpoint . local_id = "<STR_LIT>" <EOL> try : <EOL> r = self . consumer . _verifyDiscoveryResults ( self . message , endpoint ) <EOL> except ProtocolError as e : <EOL> self . assertEqual ( str ( e ) , text ) <EOL> else : <EOL> self . fail ( "<STR_LIT>" % ( r , ) ) <EOL> def test_nothingDiscovered ( self ) : <EOL> self . services = [ ] <EOL> self . assertRaises ( DiscoveryFailure , <EOL> self . consumer . _verifyDiscoveryResults , <EOL> self . message , self . endpoint ) <EOL> def discoveryFunc ( self , identifier ) : <EOL> return identifier , self . services <EOL> class TestCreateAssociationRequest ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> class DummyEndpoint ( object ) : <EOL> use_compatibility = False <EOL> def compatibilityMode ( self ) : <EOL> return self . use_compatibility <EOL> self . endpoint = DummyEndpoint ( ) <EOL> self . consumer = GenericConsumer ( store = None ) <EOL> self . assoc_type = '<STR_LIT>' <EOL> def test_noEncryptionSendsType ( self ) : <EOL> session_type = '<STR_LIT>' <EOL> session , args = self . consumer . _createAssociateRequest ( <EOL> self . endpoint , self . assoc_type , session_type ) <EOL> self . assertTrue ( isinstance ( session , PlainTextConsumerSession ) ) <EOL> expected = Message . fromOpenIDArgs ( <EOL> { '<STR_LIT>' : OPENID2_NS , <EOL> '<STR_LIT>' : session_type , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : self . assoc_type , <EOL> } ) <EOL> self . assertEqual ( expected , args ) <EOL> def test_noEncryptionCompatibility ( self ) : <EOL> self . endpoint . use_compatibility = True <EOL> session_type = '<STR_LIT>' <EOL> session , args = self . consumer . _createAssociateRequest ( <EOL> self . endpoint , self . assoc_type , session_type ) <EOL> self . assertTrue ( isinstance ( session , PlainTextConsumerSession ) ) <EOL> self . assertEqual ( Message . fromOpenIDArgs ( { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : self . assoc_type , <EOL> } ) , args ) <EOL> def test_dhSHA1Compatibility ( self ) : <EOL> setConsumerSession ( self . consumer ) <EOL> self . endpoint . use_compatibility = True <EOL> session_type = '<STR_LIT>' <EOL> session , args = self . consumer . _createAssociateRequest ( <EOL> self . endpoint , self . assoc_type , session_type ) <EOL> self . assertTrue ( isinstance ( session , DiffieHellmanSHA1ConsumerSession ) ) <EOL> self . assertTrue ( args . getArg ( OPENID1_NS , '<STR_LIT>' ) ) <EOL> args . delArg ( OPENID1_NS , '<STR_LIT>' ) <EOL> expected = Message . fromOpenIDArgs ( { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : self . assoc_type , <EOL> '<STR_LIT>' : b'<STR_LIT>' , <EOL> '<STR_LIT>' : b'<STR_LIT>' , <EOL> } ) <EOL> self . assertEqual ( expected , args ) <EOL> class _TestingDiffieHellmanResponseParameters ( object ) : <EOL> session_cls = None <EOL> message_namespace = None <EOL> def setUp ( self ) : <EOL> self . server_dh = DiffieHellman ( <NUM_LIT> , <NUM_LIT:2> ) <EOL> self . consumer_dh = DiffieHellman ( <NUM_LIT> , <NUM_LIT:2> ) <EOL> self . dh_server_public = cryptutil . longToBase64 ( self . server_dh . public ) <EOL> self . secret = cryptutil . randomString ( self . session_cls . secret_size ) <EOL> self . enc_mac_key = oidutil . toBase64 ( <EOL> self . server_dh . xorSecret ( self . consumer_dh . public , <EOL> self . secret , <EOL> self . session_cls . hash_func ) ) <EOL> self . consumer_session = self . session_cls ( self . consumer_dh ) <EOL> self . msg = Message ( self . message_namespace ) <EOL> def testExtractSecret ( self ) : <EOL> self . msg . setArg ( OPENID_NS , '<STR_LIT>' , self . dh_server_public ) <EOL> self . msg . setArg ( OPENID_NS , '<STR_LIT>' , self . enc_mac_key ) <EOL> extracted = self . consumer_session . extractSecret ( self . msg ) <EOL> self . assertEqual ( extracted , self . secret ) <EOL> def testAbsentServerPublic ( self ) : <EOL> self . msg . setArg ( OPENID_NS , '<STR_LIT>' , self . enc_mac_key ) <EOL> self . assertRaises ( KeyError , self . consumer_session . extractSecret , <EOL> self . msg ) <EOL> def testAbsentMacKey ( self ) : <EOL> self . msg . setArg ( OPENID_NS , '<STR_LIT>' , self . dh_server_public ) <EOL> self . assertRaises ( KeyError , self . consumer_session . extractSecret , <EOL> self . msg ) <EOL> def testInvalidBase64Public ( self ) : <EOL> self . msg . setArg ( OPENID_NS , '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . msg . setArg ( OPENID_NS , '<STR_LIT>' , self . enc_mac_key ) <EOL> self . assertRaises ( ValueError , <EOL> self . consumer_session . extractSecret , <EOL> self . msg ) <EOL> def testInvalidBase64MacKey ( self ) : <EOL> self . msg . setArg ( OPENID_NS , '<STR_LIT>' , self . dh_server_public ) <EOL> self . msg . setArg ( OPENID_NS , '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertRaises ( ValueError , <EOL> self . consumer_session . extractSecret , <EOL> self . msg ) <EOL> class TestOpenID1SHA1 ( _TestingDiffieHellmanResponseParameters , <EOL> unittest . TestCase ) : <EOL> session_cls = DiffieHellmanSHA1ConsumerSession <EOL> message_namespace = OPENID1_NS <EOL> class TestOpenID2SHA1 ( _TestingDiffieHellmanResponseParameters , <EOL> unittest . TestCase ) : <EOL> session_cls = DiffieHellmanSHA1ConsumerSession <EOL> message_namespace = OPENID2_NS <EOL> if cryptutil . SHA256_AVAILABLE : <EOL> class TestOpenID2SHA256 ( _TestingDiffieHellmanResponseParameters , <EOL> unittest . TestCase ) : <EOL> session_cls = DiffieHellmanSHA256ConsumerSession <EOL> message_namespace = OPENID2_NS <EOL> else : <EOL> warnings . warn ( "<STR_LIT>" ) <EOL> class TestNoStore ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . consumer = GenericConsumer ( None ) <EOL> def test_completeNoGetAssoc ( self ) : <EOL> """<STR_LIT>""" <EOL> def notCalled ( unused ) : <EOL> self . fail ( '<STR_LIT>' ) <EOL> endpoint = OpenIDServiceEndpoint ( ) <EOL> endpoint . claimed_id = '<STR_LIT>' <EOL> self . consumer . _getAssociation = notCalled <EOL> auth_request = self . consumer . begin ( endpoint ) <EOL> class NonAnonymousAuthRequest ( object ) : <EOL> endpoint = '<STR_LIT>' <EOL> def setAnonymous ( self , unused ) : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> class TestConsumerAnonymous ( unittest . TestCase ) : <EOL> def test_beginWithoutDiscoveryAnonymousFail ( self ) : <EOL> """<STR_LIT>""" <EOL> sess = { } <EOL> consumer = Consumer ( sess , None ) <EOL> def bogusBegin ( unused ) : <EOL> return NonAnonymousAuthRequest ( ) <EOL> consumer . consumer . begin = bogusBegin <EOL> self . assertRaises ( <EOL> ProtocolError , <EOL> consumer . beginWithoutDiscovery , None ) <EOL> class TestDiscoverAndVerify ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . consumer = GenericConsumer ( None ) <EOL> self . discovery_result = None <EOL> def dummyDiscover ( unused_identifier ) : <EOL> return self . discovery_result <EOL> self . consumer . _discover = dummyDiscover <EOL> self . to_match = OpenIDServiceEndpoint ( ) <EOL> def failUnlessDiscoveryFailure ( self ) : <EOL> self . assertRaises ( <EOL> DiscoveryFailure , <EOL> self . consumer . _discoverAndVerify , <EOL> '<STR_LIT>' , <EOL> [ self . to_match ] ) <EOL> def test_noServices ( self ) : <EOL> """<STR_LIT>""" <EOL> self . discovery_result = ( None , [ ] ) <EOL> self . failUnlessDiscoveryFailure ( ) <EOL> def test_noMatches ( self ) : <EOL> """<STR_LIT>""" <EOL> self . discovery_result = ( None , [ '<STR_LIT>' ] ) <EOL> def raiseProtocolError ( unused1 , unused2 ) : <EOL> raise ProtocolError ( '<STR_LIT>' ) <EOL> self . consumer . _verifyDiscoverySingle = raiseProtocolError <EOL> self . failUnlessDiscoveryFailure ( ) <EOL> def test_matches ( self ) : <EOL> """<STR_LIT>""" <EOL> matching_endpoint = '<STR_LIT>' <EOL> self . discovery_result = ( None , [ matching_endpoint ] ) <EOL> def returnTrue ( unused1 , unused2 ) : <EOL> return True <EOL> self . consumer . _verifyDiscoverySingle = returnTrue <EOL> result = self . consumer . _discoverAndVerify ( <EOL> '<STR_LIT>' , [ self . to_match ] ) <EOL> self . assertEqual ( matching_endpoint , result ) <EOL> from openid . extension import Extension <EOL> class SillyExtension ( Extension ) : <EOL> ns_uri = '<STR_LIT>' <EOL> ns_alias = '<STR_LIT>' <EOL> def getExtensionArgs ( self ) : <EOL> return { '<STR_LIT>' : '<STR_LIT>' } <EOL> class TestAddExtension ( unittest . TestCase ) : <EOL> def test_SillyExtension ( self ) : <EOL> ext = SillyExtension ( ) <EOL> ar = AuthRequest ( OpenIDServiceEndpoint ( ) , None ) <EOL> ar . addExtension ( ext ) <EOL> ext_args = ar . message . getArgs ( ext . ns_uri ) <EOL> self . assertEqual ( ext . getExtensionArgs ( ) , ext_args ) <EOL> class TestKVPost ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . server_url = '<STR_LIT>' % ( self . id ( ) , ) <EOL> def test_200 ( self ) : <EOL> from openid . fetchers import HTTPResponse <EOL> response = HTTPResponse ( ) <EOL> response . status = <NUM_LIT:200> <EOL> response . body = "<STR_LIT>" <EOL> r = _httpResponseToMessage ( response , self . server_url ) <EOL> expected_msg = Message . fromOpenIDArgs ( { '<STR_LIT:foo>' : '<STR_LIT:bar>' , '<STR_LIT>' : '<STR_LIT>' } ) <EOL> self . assertEqual ( expected_msg , r ) <EOL> def test_400 ( self ) : <EOL> response = HTTPResponse ( ) <EOL> response . status = <NUM_LIT> <EOL> response . body = "<STR_LIT>" <EOL> try : <EOL> r = _httpResponseToMessage ( response , self . server_url ) <EOL> except ServerError as e : <EOL> self . assertEqual ( e . error_text , '<STR_LIT>' ) <EOL> self . assertEqual ( e . error_code , '<STR_LIT>' ) <EOL> else : <EOL> self . fail ( "<STR_LIT>" % ( r , ) ) <EOL> def test_500 ( self ) : <EOL> response = HTTPResponse ( ) <EOL> response . status = <NUM_LIT> <EOL> response . body = "<STR_LIT>" <EOL> self . assertRaises ( fetchers . HTTPFetchingError , <EOL> _httpResponseToMessage , response , <EOL> self . server_url ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> unittest . main ( ) </s>
