<s> """<STR_LIT>""" <EOL> __all__ = [ <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> ] <EOL> from openid import extension <EOL> from openid . server . trustroot import TrustRoot <EOL> from openid . message import NamespaceMap , OPENID_NS <EOL> UNLIMITED_VALUES = "<STR_LIT>" <EOL> MINIMUM_SUPPORTED_ALIAS_LENGTH = <NUM_LIT:32> <EOL> def checkAlias ( alias ) : <EOL> """<STR_LIT>""" <EOL> if '<STR_LIT:U+002C>' in alias : <EOL> raise AXError ( "<STR_LIT>" % ( alias , ) ) <EOL> if '<STR_LIT:.>' in alias : <EOL> raise AXError ( "<STR_LIT>" % ( alias , ) ) <EOL> class AXError ( ValueError ) : <EOL> """<STR_LIT>""" <EOL> class NotAXMessage ( AXError ) : <EOL> """<STR_LIT>""" <EOL> def __repr__ ( self ) : <EOL> return self . __class__ . __name__ <EOL> def __str__ ( self ) : <EOL> return self . __class__ . __name__ <EOL> class AXMessage ( extension . Extension ) : <EOL> """<STR_LIT>""" <EOL> ns_alias = '<STR_LIT>' <EOL> ns_uri = '<STR_LIT>' <EOL> mode = None <EOL> def _checkMode ( self , ax_args ) : <EOL> """<STR_LIT>""" <EOL> mode = ax_args . get ( '<STR_LIT>' ) <EOL> if isinstance ( mode , bytes ) : <EOL> mode = str ( mode , encoding = "<STR_LIT:utf-8>" ) <EOL> if mode != self . mode : <EOL> if not mode : <EOL> raise NotAXMessage ( ) <EOL> else : <EOL> raise AXError ( <EOL> '<STR_LIT>' % ( self . mode , mode ) ) <EOL> def _newArgs ( self ) : <EOL> """<STR_LIT>""" <EOL> return { '<STR_LIT>' : self . mode } <EOL> class AttrInfo ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , type_uri , count = <NUM_LIT:1> , required = False , alias = None ) : <EOL> self . required = required <EOL> self . count = count <EOL> self . type_uri = type_uri <EOL> self . alias = alias <EOL> if self . alias is not None : <EOL> checkAlias ( self . alias ) <EOL> def wantsUnlimitedValues ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . count == UNLIMITED_VALUES <EOL> def toTypeURIs ( namespace_map , alias_list_s ) : <EOL> """<STR_LIT>""" <EOL> uris = [ ] <EOL> if alias_list_s : <EOL> for alias in alias_list_s . split ( '<STR_LIT:U+002C>' ) : <EOL> type_uri = namespace_map . getNamespaceURI ( alias ) <EOL> if type_uri is None : <EOL> raise KeyError ( <EOL> '<STR_LIT>' % ( alias , ) ) <EOL> else : <EOL> uris . append ( type_uri ) <EOL> return uris <EOL> class FetchRequest ( AXMessage ) : <EOL> """<STR_LIT>""" <EOL> mode = '<STR_LIT>' <EOL> def __init__ ( self , update_url = None ) : <EOL> AXMessage . __init__ ( self ) <EOL> self . requested_attributes = { } <EOL> self . update_url = update_url <EOL> def add ( self , attribute ) : <EOL> """<STR_LIT>""" <EOL> if attribute . type_uri in self . requested_attributes : <EOL> raise KeyError ( '<STR_LIT>' <EOL> % ( attribute . type_uri , ) ) <EOL> self . requested_attributes [ attribute . type_uri ] = attribute <EOL> def getExtensionArgs ( self ) : <EOL> """<STR_LIT>""" <EOL> aliases = NamespaceMap ( ) <EOL> required = [ ] <EOL> if_available = [ ] <EOL> ax_args = self . _newArgs ( ) <EOL> for type_uri , attribute in self . requested_attributes . items ( ) : <EOL> if attribute . alias is None : <EOL> alias = aliases . add ( type_uri ) <EOL> else : <EOL> alias = aliases . addAlias ( type_uri , attribute . alias ) <EOL> if attribute . required : <EOL> required . append ( alias ) <EOL> else : <EOL> if_available . append ( alias ) <EOL> if attribute . count != <NUM_LIT:1> : <EOL> ax_args [ '<STR_LIT>' + alias ] = str ( attribute . count ) <EOL> ax_args [ '<STR_LIT>' + alias ] = type_uri <EOL> if required : <EOL> ax_args [ '<STR_LIT>' ] = '<STR_LIT:U+002C>' . join ( required ) <EOL> if if_available : <EOL> ax_args [ '<STR_LIT>' ] = '<STR_LIT:U+002C>' . join ( if_available ) <EOL> return ax_args <EOL> def getRequiredAttrs ( self ) : <EOL> """<STR_LIT>""" <EOL> required = [ ] <EOL> for type_uri , attribute in self . requested_attributes . items ( ) : <EOL> if attribute . required : <EOL> required . append ( type_uri ) <EOL> return required <EOL> def fromOpenIDRequest ( cls , openid_request ) : <EOL> """<STR_LIT>""" <EOL> message = openid_request . message <EOL> ax_args = message . getArgs ( cls . ns_uri ) <EOL> self = cls ( ) <EOL> try : <EOL> self . parseExtensionArgs ( ax_args ) <EOL> except NotAXMessage as err : <EOL> return None <EOL> if self . update_url : <EOL> realm = message . getArg ( OPENID_NS , '<STR_LIT>' , <EOL> message . getArg ( OPENID_NS , '<STR_LIT>' ) ) <EOL> if not realm : <EOL> raise AXError ( ( "<STR_LIT>" + <EOL> "<STR_LIT>" ) % ( self . update_url , ) ) <EOL> tr = TrustRoot . parse ( realm ) <EOL> if not tr . validateURL ( self . update_url ) : <EOL> raise AXError ( <EOL> "<STR_LIT>" % <EOL> ( self . update_url , realm , ) ) <EOL> return self <EOL> fromOpenIDRequest = classmethod ( fromOpenIDRequest ) <EOL> def parseExtensionArgs ( self , ax_args ) : <EOL> """<STR_LIT>""" <EOL> self . _checkMode ( ax_args ) <EOL> aliases = NamespaceMap ( ) <EOL> for key , value in ax_args . items ( ) : <EOL> if key . startswith ( '<STR_LIT>' ) : <EOL> alias = key [ <NUM_LIT:5> : ] <EOL> type_uri = value <EOL> aliases . addAlias ( type_uri , alias ) <EOL> count_key = '<STR_LIT>' + alias <EOL> count_s = ax_args . get ( count_key ) <EOL> if count_s : <EOL> try : <EOL> count = int ( count_s ) <EOL> if count <= <NUM_LIT:0> : <EOL> raise AXError ( <EOL> "<STR_LIT>" % <EOL> ( count_key , count_s , ) ) <EOL> except ValueError : <EOL> if count_s != UNLIMITED_VALUES : <EOL> raise AXError ( <EOL> "<STR_LIT>" % <EOL> ( count_key , count_s , ) ) <EOL> count = count_s <EOL> else : <EOL> count = <NUM_LIT:1> <EOL> self . add ( AttrInfo ( type_uri , alias = alias , count = count ) ) <EOL> required = toTypeURIs ( aliases , ax_args . get ( '<STR_LIT>' ) ) <EOL> for type_uri in required : <EOL> self . requested_attributes [ type_uri ] . required = True <EOL> if_available = toTypeURIs ( aliases , ax_args . get ( '<STR_LIT>' ) ) <EOL> all_type_uris = required + if_available <EOL> for type_uri in aliases . iterNamespaceURIs ( ) : <EOL> if type_uri not in all_type_uris : <EOL> raise AXError ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' % ( type_uri , ) ) <EOL> self . update_url = ax_args . get ( '<STR_LIT>' ) <EOL> def iterAttrs ( self ) : <EOL> """<STR_LIT>""" <EOL> return iter ( self . requested_attributes . values ( ) ) <EOL> def __iter__ ( self ) : <EOL> """<STR_LIT>""" <EOL> return iter ( self . requested_attributes ) <EOL> def has_key ( self , type_uri ) : <EOL> """<STR_LIT>""" <EOL> return type_uri in self . requested_attributes <EOL> __contains__ = has_key <EOL> class AXKeyValueMessage ( AXMessage ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self ) : <EOL> AXMessage . __init__ ( self ) <EOL> self . data = { } <EOL> def addValue ( self , type_uri , value ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> values = self . data [ type_uri ] <EOL> except KeyError : <EOL> values = self . data [ type_uri ] = [ ] <EOL> values . append ( value ) <EOL> def setValues ( self , type_uri , values ) : <EOL> """<STR_LIT>""" <EOL> self . data [ type_uri ] = values <EOL> def _getExtensionKVArgs ( self , aliases = None ) : <EOL> """<STR_LIT>""" <EOL> if aliases is None : <EOL> aliases = NamespaceMap ( ) <EOL> ax_args = { } <EOL> for type_uri , values in self . data . items ( ) : <EOL> alias = aliases . add ( type_uri ) <EOL> ax_args [ '<STR_LIT>' + alias ] = type_uri <EOL> ax_args [ '<STR_LIT>' + alias ] = str ( len ( values ) ) <EOL> for i , value in enumerate ( values ) : <EOL> key = '<STR_LIT>' % ( alias , i + <NUM_LIT:1> ) <EOL> ax_args [ key ] = value <EOL> return ax_args <EOL> def parseExtensionArgs ( self , ax_args ) : <EOL> """<STR_LIT>""" <EOL> self . _checkMode ( ax_args ) <EOL> aliases = NamespaceMap ( ) <EOL> for key , value in ax_args . items ( ) : <EOL> if key . startswith ( '<STR_LIT>' ) : <EOL> type_uri = value <EOL> alias = key [ <NUM_LIT:5> : ] <EOL> checkAlias ( alias ) <EOL> aliases . addAlias ( type_uri , alias ) <EOL> for type_uri , alias in aliases . items ( ) : <EOL> try : <EOL> count_s = ax_args [ '<STR_LIT>' + alias ] <EOL> except KeyError : <EOL> value = ax_args [ '<STR_LIT>' + alias ] <EOL> if value == '<STR_LIT>' : <EOL> values = [ ] <EOL> else : <EOL> values = [ value ] <EOL> else : <EOL> count = int ( count_s ) <EOL> values = [ ] <EOL> for i in range ( <NUM_LIT:1> , count + <NUM_LIT:1> ) : <EOL> value_key = '<STR_LIT>' % ( alias , i ) <EOL> value = ax_args [ value_key ] <EOL> values . append ( value ) <EOL> self . data [ type_uri ] = values <EOL> def getSingle ( self , type_uri , default = None ) : <EOL> """<STR_LIT>""" <EOL> values = self . data . get ( type_uri ) <EOL> if not values : <EOL> return default <EOL> elif len ( values ) == <NUM_LIT:1> : <EOL> return values [ <NUM_LIT:0> ] <EOL> else : <EOL> raise AXError ( <EOL> '<STR_LIT>' % ( type_uri , ) ) <EOL> def get ( self , type_uri ) : <EOL> """<STR_LIT>""" <EOL> return self . data [ type_uri ] <EOL> def count ( self , type_uri ) : <EOL> """<STR_LIT>""" <EOL> return len ( self . get ( type_uri ) ) <EOL> class FetchResponse ( AXKeyValueMessage ) : <EOL> """<STR_LIT>""" <EOL> mode = '<STR_LIT>' <EOL> def __init__ ( self , request = None , update_url = None ) : <EOL> """<STR_LIT>""" <EOL> AXKeyValueMessage . __init__ ( self ) <EOL> self . update_url = update_url <EOL> self . request = request <EOL> def getExtensionArgs ( self ) : <EOL> """<STR_LIT>""" <EOL> aliases = NamespaceMap ( ) <EOL> zero_value_types = [ ] <EOL> if self . request is not None : <EOL> for type_uri in self . data : <EOL> if type_uri not in self . request : <EOL> raise KeyError ( <EOL> '<STR_LIT>' <EOL> % ( type_uri , ) ) <EOL> for attr_info in self . request . iterAttrs ( ) : <EOL> if attr_info . alias is None : <EOL> aliases . add ( attr_info . type_uri ) <EOL> else : <EOL> aliases . addAlias ( attr_info . type_uri , attr_info . alias ) <EOL> try : <EOL> values = self . data [ attr_info . type_uri ] <EOL> except KeyError : <EOL> values = [ ] <EOL> zero_value_types . append ( attr_info ) <EOL> if ( attr_info . count != UNLIMITED_VALUES ) and ( attr_info . count < len ( values ) ) : <EOL> raise AXError ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' % ( attr_info . type_uri , ) ) <EOL> kv_args = self . _getExtensionKVArgs ( aliases ) <EOL> ax_args = self . _newArgs ( ) <EOL> for attr_info in zero_value_types : <EOL> alias = aliases . getAlias ( attr_info . type_uri ) <EOL> kv_args [ '<STR_LIT>' + alias ] = attr_info . type_uri <EOL> kv_args [ '<STR_LIT>' + alias ] = '<STR_LIT:0>' <EOL> update_url = ( ( self . request and self . request . update_url ) <EOL> or self . update_url ) <EOL> if update_url : <EOL> ax_args [ '<STR_LIT>' ] = update_url <EOL> ax_args . update ( kv_args ) <EOL> return ax_args <EOL> def parseExtensionArgs ( self , ax_args ) : <EOL> """<STR_LIT>""" <EOL> super ( FetchResponse , self ) . parseExtensionArgs ( ax_args ) <EOL> self . update_url = ax_args . get ( '<STR_LIT>' ) <EOL> def fromSuccessResponse ( cls , success_response , signed = True ) : <EOL> """<STR_LIT>""" <EOL> self = cls ( ) <EOL> ax_args = success_response . extensionResponse ( self . ns_uri , signed ) <EOL> try : <EOL> self . parseExtensionArgs ( ax_args ) <EOL> except NotAXMessage as err : <EOL> return None <EOL> else : <EOL> return self <EOL> fromSuccessResponse = classmethod ( fromSuccessResponse ) <EOL> class StoreRequest ( AXKeyValueMessage ) : <EOL> """<STR_LIT>""" <EOL> mode = '<STR_LIT>' <EOL> def __init__ ( self , aliases = None ) : <EOL> """<STR_LIT>""" <EOL> super ( StoreRequest , self ) . __init__ ( ) <EOL> self . aliases = aliases <EOL> def getExtensionArgs ( self ) : <EOL> """<STR_LIT>""" <EOL> ax_args = self . _newArgs ( ) <EOL> kv_args = self . _getExtensionKVArgs ( self . aliases ) <EOL> ax_args . update ( kv_args ) <EOL> return ax_args <EOL> class StoreResponse ( AXMessage ) : <EOL> """<STR_LIT>""" <EOL> SUCCESS_MODE = '<STR_LIT>' <EOL> FAILURE_MODE = '<STR_LIT>' <EOL> def __init__ ( self , succeeded = True , error_message = None ) : <EOL> AXMessage . __init__ ( self ) <EOL> if succeeded and error_message is not None : <EOL> raise AXError ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> if succeeded : <EOL> self . mode = self . SUCCESS_MODE <EOL> else : <EOL> self . mode = self . FAILURE_MODE <EOL> self . error_message = error_message <EOL> def succeeded ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . mode == self . SUCCESS_MODE <EOL> def getExtensionArgs ( self ) : <EOL> """<STR_LIT>""" <EOL> ax_args = self . _newArgs ( ) <EOL> if not self . succeeded ( ) and self . error_message : <EOL> ax_args [ '<STR_LIT:error>' ] = self . error_message <EOL> return ax_args </s>
