<s> """<STR_LIT>""" <EOL> from BaseHTTPServer import BaseHTTPRequestHandler , HTTPServer <EOL> import urllib <EOL> import oauth . oauth as oauth <EOL> REQUEST_TOKEN_URL = '<STR_LIT>' <EOL> ACCESS_TOKEN_URL = '<STR_LIT>' <EOL> AUTHORIZATION_URL = '<STR_LIT>' <EOL> CALLBACK_URL = '<STR_LIT>' <EOL> RESOURCE_URL = '<STR_LIT>' <EOL> REALM = '<STR_LIT>' <EOL> VERIFIER = '<STR_LIT>' <EOL> class MockOAuthDataStore ( oauth . OAuthDataStore ) : <EOL> def __init__ ( self ) : <EOL> self . consumer = oauth . OAuthConsumer ( '<STR_LIT:key>' , '<STR_LIT>' ) <EOL> self . request_token = oauth . OAuthToken ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . access_token = oauth . OAuthToken ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . nonce = '<STR_LIT>' <EOL> self . verifier = VERIFIER <EOL> def lookup_consumer ( self , key ) : <EOL> if key == self . consumer . key : <EOL> return self . consumer <EOL> return None <EOL> def lookup_token ( self , token_type , token ) : <EOL> token_attrib = getattr ( self , '<STR_LIT>' % token_type ) <EOL> if token == token_attrib . key : <EOL> token_attrib . set_callback ( CALLBACK_URL ) <EOL> return token_attrib <EOL> return None <EOL> def lookup_nonce ( self , oauth_consumer , oauth_token , nonce ) : <EOL> if oauth_token and oauth_consumer . key == self . consumer . key and ( oauth_token . key == self . request_token . key or oauth_token . key == self . access_token . key ) and nonce == self . nonce : <EOL> return self . nonce <EOL> return None <EOL> def fetch_request_token ( self , oauth_consumer , oauth_callback ) : <EOL> if oauth_consumer . key == self . consumer . key : <EOL> if oauth_callback : <EOL> self . request_token . set_callback ( oauth_callback ) <EOL> return self . request_token <EOL> return None <EOL> def fetch_access_token ( self , oauth_consumer , oauth_token , oauth_verifier ) : <EOL> if oauth_consumer . key == self . consumer . key and oauth_token . key == self . request_token . key and oauth_verifier == self . verifier : <EOL> return self . access_token <EOL> return None <EOL> def authorize_request_token ( self , oauth_token , user ) : <EOL> if oauth_token . key == self . request_token . key : <EOL> return self . request_token <EOL> return None <EOL> class RequestHandler ( BaseHTTPRequestHandler ) : <EOL> def __init__ ( self , * args , ** kwargs ) : <EOL> self . oauth_server = oauth . OAuthServer ( MockOAuthDataStore ( ) ) <EOL> self . oauth_server . add_signature_method ( oauth . OAuthSignatureMethod_PLAINTEXT ( ) ) <EOL> self . oauth_server . add_signature_method ( oauth . OAuthSignatureMethod_HMAC_SHA1 ( ) ) <EOL> BaseHTTPRequestHandler . __init__ ( self , * args , ** kwargs ) <EOL> def send_oauth_error ( self , err = None ) : <EOL> self . send_error ( <NUM_LIT> , str ( err . message ) ) <EOL> header = oauth . build_authenticate_header ( realm = REALM ) <EOL> for k , v in header . iteritems ( ) : <EOL> self . send_header ( k , v ) <EOL> def do_GET ( self ) : <EOL> postdata = None <EOL> if self . command == '<STR_LIT:POST>' : <EOL> try : <EOL> length = int ( self . headers . getheader ( '<STR_LIT>' ) ) <EOL> postdata = self . rfile . read ( length ) <EOL> except : <EOL> pass <EOL> oauth_request = oauth . OAuthRequest . from_request ( self . command , self . path , headers = self . headers , query_string = postdata ) <EOL> if self . path . startswith ( REQUEST_TOKEN_URL ) : <EOL> try : <EOL> token = self . oauth_server . fetch_request_token ( oauth_request ) <EOL> self . send_response ( <NUM_LIT:200> , '<STR_LIT:OK>' ) <EOL> self . end_headers ( ) <EOL> self . wfile . write ( token . to_string ( ) ) <EOL> except oauth . OAuthError , err : <EOL> self . send_oauth_error ( err ) <EOL> return <EOL> if self . path . startswith ( AUTHORIZATION_URL ) : <EOL> try : <EOL> token = self . oauth_server . fetch_request_token ( oauth_request ) <EOL> token = self . oauth_server . authorize_token ( token , None ) <EOL> token . set_verifier ( VERIFIER ) <EOL> self . send_response ( <NUM_LIT:200> , '<STR_LIT:OK>' ) <EOL> self . end_headers ( ) <EOL> self . wfile . write ( token . get_callback_url ( ) ) <EOL> except oauth . OAuthError , err : <EOL> self . send_oauth_error ( err ) <EOL> return <EOL> if self . path . startswith ( ACCESS_TOKEN_URL ) : <EOL> try : <EOL> token = self . oauth_server . fetch_access_token ( oauth_request ) <EOL> self . send_response ( <NUM_LIT:200> , '<STR_LIT:OK>' ) <EOL> self . end_headers ( ) <EOL> self . wfile . write ( token . to_string ( ) ) <EOL> except oauth . OAuthError , err : <EOL> self . send_oauth_error ( err ) <EOL> return <EOL> if self . path . startswith ( RESOURCE_URL ) : <EOL> try : <EOL> consumer , token , params = self . oauth_server . verify_request ( oauth_request ) <EOL> self . send_response ( <NUM_LIT:200> , '<STR_LIT:OK>' ) <EOL> self . end_headers ( ) <EOL> self . wfile . write ( str ( params ) ) <EOL> except oauth . OAuthError , err : <EOL> self . send_oauth_error ( err ) <EOL> return <EOL> def do_POST ( self ) : <EOL> return self . do_GET ( ) <EOL> def main ( ) : <EOL> try : <EOL> server = HTTPServer ( ( '<STR_LIT>' , <NUM_LIT> ) , RequestHandler ) <EOL> print '<STR_LIT>' <EOL> server . serve_forever ( ) <EOL> except KeyboardInterrupt : <EOL> server . socket . close ( ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> main ( ) </s>
