<s> import itertools <EOL> import os <EOL> import urlparse <EOL> from github3 import GitHubError <EOL> import markupsafe <EOL> from modularodm import fields <EOL> from framework . auth import Auth <EOL> from website import settings <EOL> from website . util import web_url_for <EOL> from website . addons . base import exceptions <EOL> from website . addons . base import AddonOAuthUserSettingsBase , AddonOAuthNodeSettingsBase <EOL> from website . addons . base import StorageAddonBase <EOL> from website . addons . github import utils <EOL> from website . addons . github . api import GitHubClient <EOL> from website . addons . github . serializer import GitHubSerializer <EOL> from website . addons . github import settings as github_settings <EOL> from website . addons . github . exceptions import ApiError , NotFoundError <EOL> from website . oauth . models import ExternalProvider <EOL> hook_domain = github_settings . HOOK_DOMAIN or settings . DOMAIN <EOL> class GitHubProvider ( ExternalProvider ) : <EOL> name = '<STR_LIT>' <EOL> short_name = '<STR_LIT>' <EOL> client_id = github_settings . CLIENT_ID <EOL> client_secret = github_settings . CLIENT_SECRET <EOL> auth_url_base = github_settings . OAUTH_AUTHORIZE_URL <EOL> callback_url = github_settings . OAUTH_ACCESS_TOKEN_URL <EOL> default_scopes = github_settings . SCOPE <EOL> def handle_callback ( self , response ) : <EOL> """<STR_LIT>""" <EOL> client = GitHubClient ( <EOL> access_token = response [ '<STR_LIT>' ] <EOL> ) <EOL> user_info = client . user ( ) <EOL> return { <EOL> '<STR_LIT>' : str ( user_info . id ) , <EOL> '<STR_LIT>' : user_info . html_url , <EOL> '<STR_LIT>' : user_info . login <EOL> } <EOL> class GitHubUserSettings ( AddonOAuthUserSettingsBase ) : <EOL> """<STR_LIT>""" <EOL> oauth_provider = GitHubProvider <EOL> serializer = GitHubSerializer <EOL> def revoke_remote_oauth_access ( self , external_account ) : <EOL> """<STR_LIT>""" <EOL> connection = GitHubClient ( external_account = external_account ) <EOL> try : <EOL> connection . revoke_token ( ) <EOL> except GitHubError : <EOL> pass <EOL> @ property <EOL> def public_id ( self ) : <EOL> gh_accounts = [ <EOL> a for a in self . owner . external_accounts <EOL> if a . provider == self . oauth_provider . short_name <EOL> ] <EOL> if gh_accounts : <EOL> return gh_accounts [ <NUM_LIT:0> ] . display_name <EOL> return None <EOL> class GitHubNodeSettings ( StorageAddonBase , AddonOAuthNodeSettingsBase ) : <EOL> oauth_provider = GitHubProvider <EOL> serializer = GitHubSerializer <EOL> user = fields . StringField ( ) <EOL> repo = fields . StringField ( ) <EOL> hook_id = fields . StringField ( ) <EOL> hook_secret = fields . StringField ( ) <EOL> registration_data = fields . DictionaryField ( ) <EOL> @ property <EOL> def folder_id ( self ) : <EOL> return self . repo or None <EOL> @ property <EOL> def folder_name ( self ) : <EOL> if self . complete : <EOL> return '<STR_LIT>' . format ( self . user , self . repo ) <EOL> return None <EOL> @ property <EOL> def folder_path ( self ) : <EOL> return self . repo or None <EOL> @ property <EOL> def has_auth ( self ) : <EOL> return bool ( self . user_settings and self . user_settings . has_auth ) <EOL> @ property <EOL> def complete ( self ) : <EOL> return self . has_auth and self . repo is not None and self . user is not None <EOL> def authorize ( self , user_settings , save = False ) : <EOL> self . user_settings = user_settings <EOL> self . owner . add_log ( <EOL> action = '<STR_LIT>' , <EOL> params = { <EOL> '<STR_LIT>' : self . owner . parent_id , <EOL> '<STR_LIT>' : self . owner . _id , <EOL> } , <EOL> auth = Auth ( user_settings . owner ) , <EOL> ) <EOL> if save : <EOL> self . save ( ) <EOL> def clear_settings ( self ) : <EOL> self . user = None <EOL> self . repo = None <EOL> self . hook_id = None <EOL> self . hook_secret = None <EOL> self . registration_data = None <EOL> def deauthorize ( self , auth = None , log = True ) : <EOL> self . delete_hook ( save = False ) <EOL> self . clear_settings ( ) <EOL> if log : <EOL> self . owner . add_log ( <EOL> action = '<STR_LIT>' , <EOL> params = { <EOL> '<STR_LIT>' : self . owner . parent_id , <EOL> '<STR_LIT>' : self . owner . _id , <EOL> } , <EOL> auth = auth , <EOL> ) <EOL> self . clear_auth ( ) <EOL> def delete ( self , save = False ) : <EOL> super ( GitHubNodeSettings , self ) . delete ( save = False ) <EOL> self . deauthorize ( log = False ) <EOL> if save : <EOL> self . save ( ) <EOL> @ property <EOL> def repo_url ( self ) : <EOL> if self . user and self . repo : <EOL> return '<STR_LIT>' . format ( <EOL> self . user , self . repo <EOL> ) <EOL> @ property <EOL> def short_url ( self ) : <EOL> if self . user and self . repo : <EOL> return '<STR_LIT:/>' . join ( [ self . user , self . repo ] ) <EOL> @ property <EOL> def is_private ( self ) : <EOL> connection = GitHubClient ( external_account = self . external_account ) <EOL> return connection . repo ( user = self . user , repo = self . repo ) . private <EOL> def to_json ( self , user ) : <EOL> ret = super ( GitHubNodeSettings , self ) . to_json ( user ) <EOL> user_settings = user . get_addon ( '<STR_LIT>' ) <EOL> ret . update ( { <EOL> '<STR_LIT>' : user_settings and user_settings . has_auth , <EOL> '<STR_LIT>' : self . owner . is_registration , <EOL> } ) <EOL> if self . user_settings and self . user_settings . has_auth : <EOL> valid_credentials = False <EOL> owner = self . user_settings . owner <EOL> connection = GitHubClient ( external_account = self . external_account ) <EOL> valid_credentials = True <EOL> try : <EOL> repos = itertools . chain . from_iterable ( ( connection . repos ( ) , connection . my_org_repos ( ) ) ) <EOL> repo_names = [ <EOL> '<STR_LIT>' . format ( repo . owner . login , repo . name ) <EOL> for repo in repos <EOL> ] <EOL> except GitHubError : <EOL> repo_names = [ ] <EOL> valid_credentials = False <EOL> if owner == user : <EOL> ret . update ( { '<STR_LIT>' : repo_names } ) <EOL> ret . update ( { <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : self . user or '<STR_LIT>' , <EOL> '<STR_LIT>' : self . repo or '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' . format ( self . user , self . repo ) if ( self . user and self . repo ) else '<STR_LIT>' , <EOL> '<STR_LIT>' : owner . fullname , <EOL> '<STR_LIT>' : owner . url , <EOL> '<STR_LIT>' : owner . _id , <EOL> '<STR_LIT>' : self . external_account . display_name , <EOL> '<STR_LIT>' : self . external_account . profile_url , <EOL> '<STR_LIT>' : owner == user , <EOL> '<STR_LIT>' : valid_credentials , <EOL> '<STR_LIT>' : web_url_for ( '<STR_LIT>' ) , <EOL> '<STR_LIT>' : self . owner . web_url_for ( '<STR_LIT>' ) <EOL> } ) <EOL> return ret <EOL> def serialize_waterbutler_credentials ( self ) : <EOL> if not self . complete or not self . repo : <EOL> raise exceptions . AddonError ( '<STR_LIT>' ) <EOL> return { '<STR_LIT>' : self . external_account . oauth_key } <EOL> def serialize_waterbutler_settings ( self ) : <EOL> if not self . complete : <EOL> raise exceptions . AddonError ( '<STR_LIT>' ) <EOL> return { <EOL> '<STR_LIT>' : self . user , <EOL> '<STR_LIT>' : self . repo , <EOL> } <EOL> def create_waterbutler_log ( self , auth , action , metadata ) : <EOL> path = metadata [ '<STR_LIT:path>' ] <EOL> url = self . owner . web_url_for ( '<STR_LIT>' , path = path , provider = '<STR_LIT>' ) <EOL> if not metadata . get ( '<STR_LIT>' ) : <EOL> sha = None <EOL> urls = { } <EOL> else : <EOL> sha = metadata [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> urls = { <EOL> '<STR_LIT>' : '<STR_LIT>' . format ( url , sha ) , <EOL> '<STR_LIT>' : '<STR_LIT>' . format ( url , sha ) <EOL> } <EOL> self . owner . add_log ( <EOL> '<STR_LIT>' . format ( action ) , <EOL> auth = auth , <EOL> params = { <EOL> '<STR_LIT>' : self . owner . parent_id , <EOL> '<STR_LIT>' : self . owner . _id , <EOL> '<STR_LIT:path>' : path , <EOL> '<STR_LIT>' : urls , <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT:user>' : self . user , <EOL> '<STR_LIT>' : self . repo , <EOL> '<STR_LIT>' : sha , <EOL> } , <EOL> } , <EOL> ) <EOL> def before_page_load ( self , node , user ) : <EOL> """<STR_LIT>""" <EOL> messages = [ ] <EOL> if not node . is_contributor ( user ) : <EOL> return messages <EOL> if self . user is None or self . repo is None : <EOL> return messages <EOL> if self . user_settings is None : <EOL> return messages <EOL> connect = GitHubClient ( external_account = self . external_account ) <EOL> try : <EOL> repo = connect . repo ( self . user , self . repo ) <EOL> except ( ApiError , GitHubError ) : <EOL> return <EOL> node_permissions = '<STR_LIT>' if node . is_public else '<STR_LIT>' <EOL> repo_permissions = '<STR_LIT>' if repo . private else '<STR_LIT>' <EOL> if repo_permissions != node_permissions : <EOL> message = ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' . format ( <EOL> category = markupsafe . escape ( node . project_or_component ) , <EOL> node_perm = markupsafe . escape ( node_permissions ) , <EOL> repo_perm = markupsafe . escape ( repo_permissions ) , <EOL> user = markupsafe . escape ( self . user ) , <EOL> repo = markupsafe . escape ( self . repo ) , <EOL> ) <EOL> ) <EOL> if repo_permissions == '<STR_LIT>' : <EOL> message += ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> ) <EOL> else : <EOL> message += ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> ) . format ( <EOL> user = self . user , <EOL> repo = self . repo , <EOL> ) <EOL> messages . append ( message ) <EOL> return messages <EOL> def before_remove_contributor_message ( self , node , removed ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> message = ( super ( GitHubNodeSettings , self ) . before_remove_contributor_message ( node , removed ) + <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' . format ( <EOL> url = node . api_url + '<STR_LIT>' <EOL> ) ) <EOL> except TypeError : <EOL> return None <EOL> else : <EOL> return message <EOL> before_remove_contributor = before_remove_contributor_message <EOL> def after_remove_contributor ( self , node , removed , auth = None ) : <EOL> """<STR_LIT>""" <EOL> if self . user_settings and self . user_settings . owner == removed : <EOL> self . user_settings = None <EOL> self . save ( ) <EOL> message = ( <EOL> u'<STR_LIT>' <EOL> u'<STR_LIT>' <EOL> ) . format ( <EOL> category = markupsafe . escape ( node . category_display ) , <EOL> title = markupsafe . escape ( node . title ) , <EOL> user = markupsafe . escape ( removed . fullname ) <EOL> ) <EOL> if not auth or auth . user != removed : <EOL> url = node . web_url_for ( '<STR_LIT>' ) <EOL> message += ( <EOL> u'<STR_LIT>' <EOL> ) . format ( url = url ) <EOL> return message <EOL> def after_fork ( self , node , fork , user , save = True ) : <EOL> """<STR_LIT>""" <EOL> clone , _ = super ( GitHubNodeSettings , self ) . after_fork ( <EOL> node , fork , user , save = False <EOL> ) <EOL> if self . user_settings and self . user_settings . owner == user : <EOL> clone . user_settings = self . user_settings <EOL> message = ( <EOL> '<STR_LIT>' <EOL> ) . format ( <EOL> cat = markupsafe . escape ( fork . project_or_component ) , <EOL> ) <EOL> else : <EOL> message = ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> ) . format ( <EOL> cat = markupsafe . escape ( fork . project_or_component ) , <EOL> url = fork . url + '<STR_LIT>' <EOL> ) <EOL> if save : <EOL> clone . save ( ) <EOL> return clone , message <EOL> def before_make_public ( self , node ) : <EOL> try : <EOL> is_private = self . is_private <EOL> except NotFoundError : <EOL> return None <EOL> if is_private : <EOL> return ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> ) . format ( <EOL> cat = node . project_or_component , <EOL> ) <EOL> def after_delete ( self , node , user ) : <EOL> self . deauthorize ( Auth ( user = user ) , log = True ) <EOL> def add_hook ( self , save = True ) : <EOL> if self . user_settings : <EOL> connect = GitHubClient ( external_account = self . external_account ) <EOL> secret = utils . make_hook_secret ( ) <EOL> hook = connect . add_hook ( <EOL> self . user , self . repo , <EOL> '<STR_LIT>' , <EOL> { <EOL> '<STR_LIT:url>' : urlparse . urljoin ( <EOL> hook_domain , <EOL> os . path . join ( <EOL> self . owner . api_url , '<STR_LIT>' , '<STR_LIT>' <EOL> ) <EOL> ) , <EOL> '<STR_LIT>' : github_settings . HOOK_CONTENT_TYPE , <EOL> '<STR_LIT>' : secret , <EOL> } , <EOL> events = github_settings . HOOK_EVENTS , <EOL> ) <EOL> if hook : <EOL> self . hook_id = hook . id <EOL> self . hook_secret = secret <EOL> if save : <EOL> self . save ( ) <EOL> def delete_hook ( self , save = True ) : <EOL> """<STR_LIT>""" <EOL> if self . user_settings and self . hook_id : <EOL> connection = GitHubClient ( external_account = self . external_account ) <EOL> try : <EOL> response = connection . delete_hook ( self . user , self . repo , self . hook_id ) <EOL> except ( GitHubError , NotFoundError ) : <EOL> return False <EOL> if response : <EOL> self . hook_id = None <EOL> if save : <EOL> self . save ( ) <EOL> return True <EOL> return False </s>
