<s> import base64 <EOL> import sys <EOL> import os <EOL> import socket <EOL> import struct <EOL> from crtauth import exceptions <EOL> from crtauth import rsa <EOL> SSH2_AGENTC_REQUEST_IDENTITIES = <NUM_LIT:11> <EOL> SSH2_AGENT_IDENTITIES_ANSWER = <NUM_LIT:12> <EOL> SSH2_AGENTC_SIGN_REQUEST = <NUM_LIT> <EOL> SSH2_AGENT_SIGN_RESPONSE = <NUM_LIT> <EOL> def base64url_decode ( s ) : <EOL> """<STR_LIT>""" <EOL> if type ( s ) is unicode : <EOL> s = s . encode ( "<STR_LIT>" ) <EOL> if len ( s ) % <NUM_LIT:4> == <NUM_LIT:3> : <EOL> s += "<STR_LIT:=>" <EOL> elif len ( s ) % <NUM_LIT:4> == <NUM_LIT:2> : <EOL> s += "<STR_LIT>" <EOL> try : <EOL> return base64 . b64decode ( s , "<STR_LIT>" ) <EOL> except : <EOL> _ , e , tb = sys . exc_info ( ) <EOL> raise exceptions . InvalidInputException ( <EOL> "<STR_LIT>" % e ) , None , tb <EOL> def base64url_encode ( data ) : <EOL> """<STR_LIT>""" <EOL> s = base64 . b64encode ( data , "<STR_LIT>" ) <EOL> return s . rstrip ( "<STR_LIT:=>" ) <EOL> def i2s ( i ) : <EOL> return struct . pack ( "<STR_LIT>" , i ) <EOL> def write_field ( socket , data ) : <EOL> if type ( data ) is int : <EOL> data = chr ( data ) <EOL> elif type ( data ) is not str : <EOL> data = repr ( data ) <EOL> socket . send ( i2s ( len ( data ) ) + data ) <EOL> class SigningPlug ( object ) : <EOL> """<STR_LIT>""" <EOL> def sign ( self , data , fingerprint ) : <EOL> raise NotImplementedError ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> def __enter__ ( self ) : <EOL> pass <EOL> def __exit__ ( self , exc_class , exc_value , bt ) : <EOL> self . close ( ) <EOL> def close ( self ) : <EOL> pass <EOL> class SingleKeySigner ( SigningPlug ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , priv_key ) : <EOL> self . key = rsa . RSAPrivateKey ( priv_key ) <EOL> def sign ( self , data , fingerprint ) : <EOL> return self . key . sign ( data ) <EOL> class AgentSigner ( SigningPlug ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self ) : <EOL> self . sock = socket . socket ( socket . AF_UNIX ) <EOL> sock_path = os . getenv ( "<STR_LIT>" ) <EOL> if not sock_path : <EOL> raise exceptions . SshAgentError ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> self . sock . settimeout ( <NUM_LIT> ) <EOL> self . sock . connect ( sock_path ) <EOL> def __find_key ( self , key_fingerprint ) : <EOL> write_field ( self . sock , SSH2_AGENTC_REQUEST_IDENTITIES ) <EOL> length , response_code , count = struct . unpack ( "<STR_LIT>" , <EOL> self . sock . recv ( <NUM_LIT:9> ) ) <EOL> assert response_code == SSH2_AGENT_IDENTITIES_ANSWER <EOL> resp = self . sock . recv ( length - <NUM_LIT:5> ) <EOL> fields = rsa . read_fields ( resp ) <EOL> for i in xrange ( count ) : <EOL> try : <EOL> key = rsa . RSAPublicKey ( fields . next ( ) ) <EOL> except exceptions . KeyError : <EOL> fields . next ( ) <EOL> continue <EOL> fields . next ( ) <EOL> if key . fingerprint ( ) == key_fingerprint : <EOL> return key <EOL> def sign ( self , data , fingerprint ) : <EOL> try : <EOL> pub_key = self . __find_key ( fingerprint ) <EOL> if not pub_key : <EOL> raise exceptions . SshAgentError ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> self . sock . send ( i2s ( len ( pub_key ) + len ( data ) + <NUM_LIT> ) + <EOL> chr ( SSH2_AGENTC_SIGN_REQUEST ) ) <EOL> write_field ( self . sock , pub_key ) <EOL> write_field ( self . sock , data ) <EOL> self . sock . send ( "<STR_LIT>" ) <EOL> length , response_code , resp_len = struct . unpack ( "<STR_LIT>" , <EOL> self . sock . recv ( <NUM_LIT:9> ) ) <EOL> assert response_code == SSH2_AGENT_SIGN_RESPONSE <EOL> buf = self . sock . recv ( length - <NUM_LIT:5> ) <EOL> fields = rsa . read_fields ( buf ) <EOL> response_type = fields . next ( ) <EOL> assert response_type == "<STR_LIT>" <EOL> return fields . next ( ) <EOL> except socket . timeout as why : <EOL> raise exceptions . SshAgentError ( why ) <EOL> def close ( self ) : <EOL> self . sock . close ( ) </s>
