<s> """<STR_LIT>""" <EOL> __all__ = [ <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> ] <EOL> import time <EOL> from openid import cryptutil <EOL> from openid import kvform <EOL> from openid import oidutil <EOL> from openid . message import OPENID_NS <EOL> all_association_types = [ <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> ] <EOL> if hasattr ( cryptutil , '<STR_LIT>' ) : <EOL> supported_association_types = list ( all_association_types ) <EOL> default_association_order = [ <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ] <EOL> only_encrypted_association_order = [ <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ] <EOL> else : <EOL> supported_association_types = [ '<STR_LIT>' ] <EOL> default_association_order = [ <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ] <EOL> only_encrypted_association_order = [ <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ] <EOL> def getSessionTypes ( assoc_type ) : <EOL> """<STR_LIT>""" <EOL> assoc_to_session = { <EOL> '<STR_LIT>' : [ '<STR_LIT>' , '<STR_LIT>' ] , <EOL> '<STR_LIT>' : [ '<STR_LIT>' , '<STR_LIT>' ] , <EOL> } <EOL> return assoc_to_session . get ( assoc_type , [ ] ) <EOL> def checkSessionType ( assoc_type , session_type ) : <EOL> """<STR_LIT>""" <EOL> if session_type not in getSessionTypes ( assoc_type ) : <EOL> raise ValueError ( <EOL> '<STR_LIT>' <EOL> % ( session_type , assoc_type ) ) <EOL> class SessionNegotiator ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , allowed_types ) : <EOL> self . setAllowedTypes ( allowed_types ) <EOL> def copy ( self ) : <EOL> return self . __class__ ( list ( self . allowed_types ) ) <EOL> def setAllowedTypes ( self , allowed_types ) : <EOL> """<STR_LIT>""" <EOL> for ( assoc_type , session_type ) in allowed_types : <EOL> checkSessionType ( assoc_type , session_type ) <EOL> self . allowed_types = allowed_types <EOL> def addAllowedType ( self , assoc_type , session_type = None ) : <EOL> """<STR_LIT>""" <EOL> if self . allowed_types is None : <EOL> self . allowed_types = [ ] <EOL> if session_type is None : <EOL> available = getSessionTypes ( assoc_type ) <EOL> if not available : <EOL> raise ValueError ( '<STR_LIT>' <EOL> % ( assoc_type , ) ) <EOL> for session_type in getSessionTypes ( assoc_type ) : <EOL> self . addAllowedType ( assoc_type , session_type ) <EOL> else : <EOL> checkSessionType ( assoc_type , session_type ) <EOL> self . allowed_types . append ( ( assoc_type , session_type ) ) <EOL> def isAllowed ( self , assoc_type , session_type ) : <EOL> """<STR_LIT>""" <EOL> assoc_good = ( assoc_type , session_type ) in self . allowed_types <EOL> matches = session_type in getSessionTypes ( assoc_type ) <EOL> return assoc_good and matches <EOL> def getAllowedType ( self ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> return self . allowed_types [ <NUM_LIT:0> ] <EOL> except IndexError : <EOL> return ( None , None ) <EOL> default_negotiator = SessionNegotiator ( default_association_order ) <EOL> encrypted_negotiator = SessionNegotiator ( only_encrypted_association_order ) <EOL> def getSecretSize ( assoc_type ) : <EOL> if assoc_type == '<STR_LIT>' : <EOL> return <NUM_LIT:20> <EOL> elif assoc_type == '<STR_LIT>' : <EOL> return <NUM_LIT:32> <EOL> else : <EOL> raise ValueError ( '<STR_LIT>' % ( assoc_type , ) ) <EOL> class Association ( object ) : <EOL> """<STR_LIT>""" <EOL> assoc_keys = [ <EOL> '<STR_LIT:version>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> ] <EOL> _macs = { <EOL> '<STR_LIT>' : cryptutil . hmacSha1 , <EOL> '<STR_LIT>' : cryptutil . hmacSha256 , <EOL> } <EOL> @ classmethod <EOL> def fromExpiresIn ( cls , expires_in , handle , secret , assoc_type ) : <EOL> """<STR_LIT>""" <EOL> issued = int ( time . time ( ) ) <EOL> lifetime = expires_in <EOL> return cls ( handle , secret , issued , lifetime , assoc_type ) <EOL> def __init__ ( self , handle , secret , issued , lifetime , assoc_type ) : <EOL> """<STR_LIT>""" <EOL> if assoc_type not in all_association_types : <EOL> fmt = '<STR_LIT>' <EOL> raise ValueError ( fmt % ( assoc_type , ) ) <EOL> self . handle = handle <EOL> if isinstance ( secret , str ) : <EOL> secret = secret . encode ( "<STR_LIT:utf-8>" ) <EOL> self . secret = secret <EOL> self . issued = issued <EOL> self . lifetime = lifetime <EOL> self . assoc_type = assoc_type <EOL> @ property <EOL> def expiresIn ( self , now = None ) : <EOL> """<STR_LIT>""" <EOL> if now is None : <EOL> now = int ( time . time ( ) ) <EOL> return max ( <NUM_LIT:0> , self . issued + self . lifetime - now ) <EOL> def __eq__ ( self , other ) : <EOL> """<STR_LIT>""" <EOL> return type ( self ) is type ( other ) and self . __dict__ == other . __dict__ <EOL> def __ne__ ( self , other ) : <EOL> """<STR_LIT>""" <EOL> return not ( self == other ) <EOL> def serialize ( self ) : <EOL> """<STR_LIT>""" <EOL> data = { <EOL> '<STR_LIT:version>' : '<STR_LIT:2>' , <EOL> '<STR_LIT>' : self . handle , <EOL> '<STR_LIT>' : oidutil . toBase64 ( self . secret ) , <EOL> '<STR_LIT>' : str ( int ( self . issued ) ) , <EOL> '<STR_LIT>' : str ( int ( self . lifetime ) ) , <EOL> '<STR_LIT>' : self . assoc_type <EOL> } <EOL> assert len ( data ) == len ( self . assoc_keys ) <EOL> pairs = [ ] <EOL> for field_name in self . assoc_keys : <EOL> pairs . append ( ( field_name , data [ field_name ] ) ) <EOL> return kvform . seqToKV ( pairs , strict = True ) <EOL> @ classmethod <EOL> def deserialize ( cls , assoc_s ) : <EOL> """<STR_LIT>""" <EOL> pairs = kvform . kvToSeq ( assoc_s , strict = True ) <EOL> keys = [ ] <EOL> values = [ ] <EOL> for k , v in pairs : <EOL> keys . append ( k ) <EOL> values . append ( v ) <EOL> if keys != cls . assoc_keys : <EOL> raise ValueError ( '<STR_LIT>' , keys ) <EOL> version , handle , secret , issued , lifetime , assoc_type = values <EOL> if version != '<STR_LIT:2>' : <EOL> raise ValueError ( '<STR_LIT>' % version ) <EOL> issued = int ( issued ) <EOL> lifetime = int ( lifetime ) <EOL> secret = oidutil . fromBase64 ( secret ) <EOL> return cls ( handle , secret , issued , lifetime , assoc_type ) <EOL> def sign ( self , pairs ) : <EOL> """<STR_LIT>""" <EOL> kv = kvform . seqToKV ( pairs ) <EOL> try : <EOL> mac = self . _macs [ self . assoc_type ] <EOL> except KeyError : <EOL> raise ValueError ( <EOL> '<STR_LIT>' % ( self . assoc_type , ) ) <EOL> return mac ( self . secret , kv ) <EOL> def getMessageSignature ( self , message ) : <EOL> """<STR_LIT>""" <EOL> pairs = self . _makePairs ( message ) <EOL> return oidutil . toBase64 ( self . sign ( pairs ) ) <EOL> def signMessage ( self , message ) : <EOL> """<STR_LIT>""" <EOL> if ( message . hasKey ( OPENID_NS , '<STR_LIT>' ) or <EOL> message . hasKey ( OPENID_NS , '<STR_LIT>' ) ) : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> extant_handle = message . getArg ( OPENID_NS , '<STR_LIT>' ) <EOL> if extant_handle and extant_handle != self . handle : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> signed_message = message . copy ( ) <EOL> signed_message . setArg ( OPENID_NS , '<STR_LIT>' , self . handle ) <EOL> message_keys = list ( signed_message . toPostArgs ( ) . keys ( ) ) <EOL> signed_list = [ k [ <NUM_LIT:7> : ] for k in message_keys <EOL> if k . startswith ( '<STR_LIT>' ) ] <EOL> signed_list . append ( '<STR_LIT>' ) <EOL> signed_list . sort ( ) <EOL> signed_message . setArg ( OPENID_NS , '<STR_LIT>' , '<STR_LIT:U+002C>' . join ( signed_list ) ) <EOL> sig = self . getMessageSignature ( signed_message ) <EOL> signed_message . setArg ( OPENID_NS , '<STR_LIT>' , sig ) <EOL> return signed_message <EOL> def checkMessageSignature ( self , message ) : <EOL> """<STR_LIT>""" <EOL> message_sig = message . getArg ( OPENID_NS , '<STR_LIT>' ) <EOL> if not message_sig : <EOL> raise ValueError ( "<STR_LIT>" % ( message , ) ) <EOL> calculated_sig = self . getMessageSignature ( message ) <EOL> calculated_sig = calculated_sig . decode ( '<STR_LIT:utf-8>' ) <EOL> return cryptutil . const_eq ( calculated_sig , message_sig ) <EOL> def _makePairs ( self , message ) : <EOL> signed = message . getArg ( OPENID_NS , '<STR_LIT>' ) <EOL> if not signed : <EOL> raise ValueError ( '<STR_LIT>' % ( message , ) ) <EOL> signed_list = signed . split ( '<STR_LIT:U+002C>' ) <EOL> pairs = [ ] <EOL> data = message . toPostArgs ( ) <EOL> for field in signed_list : <EOL> pairs . append ( ( field , data . get ( '<STR_LIT>' + field , '<STR_LIT>' ) ) ) <EOL> return pairs <EOL> def __repr__ ( self ) : <EOL> return "<STR_LIT>" % ( <EOL> self . __class__ . __module__ , <EOL> self . __class__ . __name__ , <EOL> self . assoc_type , <EOL> self . handle ) </s>
