PKS%Jv>Xn]puwH print('No action available for the webhook :', ':',payload)vw5_ Xss>v>Xn]WY* # TODO: Extract fom X-GitHub-Event5_ vss>v>Xn] import re import os import hmacimport tornado.webimport tornado.httpserverimport tornado.ioloop from .utils import Authenticator%from yieldbreaker import YieldBreaker class Config: botname = None integration_id = None key = None botname = None at_botname = None integration_id = None webhook_secret = None! def __init__(self, **kwargs):$ self.__dict__.update(kwargs) def validate(self):E missing = [attr for attr in dir(self) if not attr.startswith(1 '_') and getattr(self, attr) is None] if missing: raise ValueError(X 'The followingg configuration options are missing : {}'.format(missing)) return self1def verify_signature(payload, signature, secret): """) Make sure hooks are encoded correctly """9 expected = 'sha1=' + hmac.new(secret.encode('ascii'),> payload, 'sha1').hexdigest()3 return hmac.compare_digest(signature, expected).class BaseHandler(tornado.web.RequestHandler): def error(self, message): self.set_status(500); self.write({'status': 'error', 'message': message}). def success(self, message='', payload={}):N self.write({'status': 'success', 'message': message, 'data': payload})class MainHandler(BaseHandler): def get(self): self.finish('No').def process_mentionning_comment(body, bot_re): """W Given a comment body and a bot name parse this into a tuple of (command, arguments) """ lines = body.splitlines(): lines = [l.strip() for l in lines if bot_re.search(l)]8 lines = [bot_re.split(l)[-1].strip() for l in lines]3 command_args = [l.split(' ', 1) for l in lines]L command_args = [c if len(c) > 1 else (c[0], None) for c in command_args] return command_args"class WebHookHandler(MainHandler):A def initialize(self, actions, config, auth, *args, **kwargs): self.actions = actions self.config = config self.auth = auth+ super().initialize(*args, **kwargs)5 print('Webhook initialize got', args, kwargs) def get(self):5 self.getfinish("Webhook alive and listening") def post(self):9 if not 'X-Hub-Signature' in self.request.headers:C return self.error('WebHook not configured with secret')+ # TODO: Extract from X-GitHub-Event2 if not verify_signature(self.request.body,H self.request.headers['X-Hub-Signature'],< self.config.webhook_secret):D return self.error('Cannot validate GitHub payload with '8 'provided WebHook secret')? payload = tornado.escape.json_decode(self.request.body)I org = payload.get('repository', {}).get('owner', {}).get('login')\ if hasattr(self.config, 'org_whitelist') and (org not in self.config.org_whitelist):* print('Non allowed org:', org)* self.error('Not allowed org.'); sender = payload.get('sender', {}).get('login', {})a if hasattr(self.config, 'user_whitelist') and (sender not in self.config.user_whitelist):. print('Not allowed user:', sender)+ self.error('Not allowed user.'), action = payload.get("action", None)" if payload.get('commits'): # TODO3 print("commits were likely pushed....") return if action:+ print('## dispatching request',@ self.request.headers.get('X-GitHub-Delivery'))8 return self.dispatch_action(action, payload) else:x print('No action available for the webhook :', self.request.headers.get('X-GitHub-Delivery'), ':',payload) @property def mention_bot_re(self):% botname = self.config.botnameT return re.compile('@?' + re.escape(botname) + '(?:\[bot\])?', re.IGNORECASE). def dispatch_action(self, type_, payload):% botname = self.config.botname # new issue/PR opened if type_ == 'opened':. issue = payload.get('issue', None) if not issue:; print('request has no issue key:', payload)J return self.error('Not really good, request has no issue') if issue:8 user = payload['issue']['user']['login']A if user == self.config.botname.lower() + '[bot]':@ return self.finish("Not responding to self")& # todo dispatch on on-open # new comment created elif type_ == 'created':2 comment = payload.get('comment', None)< installation = payload.get('installation', None) if comment:: user = payload['comment']['user']['login']5 if user == botname.lower() + '[bot]':3 print('Not responding to self')@ return self.finish("Not responding to self")# if '[bot]' in user:: print('Not responding to another bot')G return self.finish("Not responding to another bot")1 body = payload['comment']['body'], print('Got a comment', body)5 if self.mention_bot_re.findall(body):A self.dispatch_on_mention(body, payload, user) else:. print('Was not mentioned',? self.config.botname, body, '|', user)> elif installation and installation.get('account'):D print('we got a new installation maybe ?!', payload)$ return self.finish() else:- print('not handled', payload) else:3 print("can't deal with ", type_, "yet")7 def dispatch_on_mention(self, body, payload, user):! # to dispatch to commands7 installation_id = payload['installation']['id']. org = payload['organization']['login'], repo = payload['repository']['name']4 session = self.auth.session(installation_id); is_admin = session.is_collaborator(org, repo, user)M command_args = process_mentionning_comment(body, self.mention_bot_re)1 for (command, arguments) in command_args:8 print(" :: treating", command, arguments)5 handler = self.actions.get(command, None) if handler:B print(" :: testing who can use ", str(handler))^ if ((handler.scope == 'admin') and is_admin) or (handler.scope == 'everyone'):I print(" :: authorisation granted ", handler.scope)( maybe_gen = handler(N session=session, payload=payload, arguments=arguments) import types> if type(maybe_gen) == types.GeneratorType:5 gen = YieldBreaker(maybe_gen), for org_repo in gen:= torg, trepo = org_repo.split('/')F session_id = self.auth.idmap.get(org_repo)* if session_id:N target_session = self.auth.session(session_id)U if target_session.is_collaborator(torg, trepo, user):< gen.send(target_session)% else:2 gen.send(None)! else:W print('org/repo not found', org_repo, self.auth.id_map). gen.send(None) else:5 print('I Cannot let you do that') else:2 print('unnknown command', command)class MeeseeksBox:) def __init__(self, commands, config): self.commands = commands5 self.port = int(os.environ.get('PORT', 5000)) self.application = None self.config = configN self.auth = Authenticator(self.config.integration_id, self.config.key)* self.auth._build_auth_id_mapping() def start(self):4 self.application = tornado.web.Application([ (r"/", MainHandler),) (r"/webhook", WebHookHandler,R {'actions': self.commands, 'config': self.config, 'auth': self.auth}) ])I tornado.httpserver.HTTPServer(self.application).listen(self.port)0 tornado.ioloop.IOLoop.instance().start()5_ ;ss>v>Xn]:<1def process_mentionning_cogitmment(body, bot_re):5_v;Xn]avwuw' print('No action available for the webhook :',{'sender': {'id': 2680980, 'organizations_url': 'https://api.github.com/users/willingc/orgs', 'url': 'https://api.github.com/users/willingc', 'received_events_url': 'https://api.github.com/users/willingc/received_events', 'html_url': 'https://github.com/willingc', 'site_admin': False, 'following_url': 'https://api.github.com/users/willingc/following{/other_user}', 'followers_url': 'https://api.github.com/users/willingc/followers', 'gravatar_id': '', 'starred_url': 'https://api.github.com/users/willingc/starred{/owner}{/repo}', 'type': 'User', 'events_url': 'https://api.github.com/users/willingc/events{/privacy}', 'repos_url': 'https://api.github.com/users/willingc/repos', 'avatar_url': 'https://avatars.githubusercontent.com/u/2680980?v=3', 'subscriptions_url': 'https://api.github.com/users/willingc/subscriptions', 'gists_url': 'https://api.github.com/users/willingc/gists{/gist_id}', 'login': 'willingc'}, 'organization': {'events_url': 'https://api.github.com/orgs/ipython/events', 'id': 230453, 'repos_url': 'https://api.github.com/orgs/ipython/repos', 'url': 'https://api.github.com/orgs/ipython', 'members_url': 'https://api.github.com/orgs/ipython/members{/member}', 'description': 'interactive computing in Python', 'hooks_url': 'https://api.github.com/orgs/ipython/hooks', 'issues_url': 'https://api.github.com/orgs/ipython/issues', 'avatar_url': 'https://avatars.githubusercontent.com/u/230453?v=3', 'public_members_url': 'https://api.github.com/orgs/ipython/public_members{/member}', 'login': 'ipython'}, 'sha': 'bda4ba81e9292cffb9852ef22c918d4e48bc5f64', 'id': 941431245, 'name': 'ipython/ipython', 'updated_at': '2017-01-05T14:49:49Z', 'state': 'success', 'created_at': '2017-01-05T14:49:49Z', 'repository': {'keys_url': 'https://api.github.com/repos/ipython/ipython/keys{/key_id}', 'git_commits_url': 'https://api.github.com/repos/ipython/ipython/git/commits{/sha}', 'has_downloads': True, 'name': 'ipython', 'url': 'https://api.github.com/repos/ipython/ipython', 'watchers_count': 10903, 'has_wiki': True, 'language': 'Python', 'deployments_url': 'https://api.github.com/repos/ipython/ipython/deployments', 'pushed_at': '2017-01-05T14:36:22Z', 'events_url': 'https://api.github.com/repos/ipython/ipython/events', 'downloads_url': 'https://api.github.com/repos/ipython/ipython/downloads', 'stargazers_url': 'https://api.github.com/repos/ipython/ipython/stargazers', 'subscription_url': 'https://api.github.com/repos/ipython/ipython/subscription', 'git_tags_url': 'https://api.github.com/repos/ipython/ipython/git/tags{/sha}', 'has_issues': True, 'compare_url': 'https://api.github.com/repos/ipython/ipython/compare/{base}...{head}', 'forks_count': 3221, 'branches_url': 'https://api.github.com/repos/ipython/ipython/branches{/branch}', 'issue_events_url': 'https://api.github.com/repos/ipython/ipython/issues/events{/number}', 'archive_url': 'https://api.github.com/repos/ipython/ipython/{archive_format}{/ref}', 'created_at': '2010-05-10T04:46:06Z', 'open_issues_count': 996, 'open_issues': 996, 'contributors_url': 'https://api.github.com/repos/ipython/ipython/contributors', 'blobs_url': 'https://api.github.com/repos/ipython/ipython/git/blobs{/sha}', 'hooks_url': 'https://api.github.com/repos/ipython/ipython/hooks', 'owner': {'id': 230453, 'organizations_url': 'https://api.github.com/users/ipython/orgs', 'url': 'https://api.github.com/users/ipython', 'received_events_url': 'https://api.github.com/users/ipython/received_events', 'html_url': 'https://github.com/ipython', 'site_admin': False, 'following_url': 'https://api.github.com/users/ipython/following{/other_user}', 'followers_url': 'https://api.github.com/users/ipython/followers', 'gravatar_id': '', 'starred_url': 'https://api.github.com/users/ipython/starred{/owner}{/repo}', 'type': 'Organization', 'events_url': 'https://api.github.com/users/ipython/events{/privacy}', 'repos_url': 'https://api.github.com/users/ipython/repos', 'avatar_url': 'https://avatars.githubusercontent.com/u/230453?v=3', 'subscriptions_url': 'https://api.github.com/users/ipython/subscriptions', 'gists_url': 'https://api.github.com/users/ipython/gists{/gist_id}', 'login': 'ipython'}, 'statuses_url': 'https://api.github.com/repos/ipython/ipython/statuses/{sha}', 'pulls_url': 'https://api.github.com/repos/ipython/ipython/pulls{/number}', 'commits_url': 'https://api.github.com/repos/ipython/ipython/commits{/sha}', 'notifications_url': 'https://api.github.com/repos/ipython/ipython/notifications{?since,all,participating}', 'git_refs_url': 'https://api.github.com/repos/ipython/ipython/git/refs{/sha}', 'tags_url': 'https://api.github.com/repos/ipython/ipython/tags', 'forks': 3221, 'watchers': 10903, 'issue_comment_url': 'https://api.github.com/repos/ipython/ipython/issues/comments{/number}', 'updated_at': '2017-01-05T14:03:25Z', 'svn_url': 'https://github.com/ipython/ipython', 'mirror_url': None, 'issues_url': 'https://api.github.com/repos/ipython/ipython/issues{/number}', 'ssh_url': 'git@github.com:ipython/ipython.git', 'assignees_url': 'https://api.github.com/repos/ipython/ipython/assignees{/user}', 'default_branch': 'master', 'languages_url': 'https://api.github.com/repos/ipython/ipython/languages', 'labels_url': 'https://api.github.com/repos/ipython/ipython/labels{/name}', 'id': 658518, 'merges_url': 'https://api.github.com/repos/ipython/ipython/merges', 'git_url': 'git://github.com/ipython/ipython.git', 'private': False, 'releases_url': 'https://api.github.com/repos/ipython/ipython/releases{/id}', 'subscribers_url': 'https://api.github.com/repos/ipython/ipython/subscribers', 'size': 69027, 'contents_url': 'https://api.github.com/repos/ipython/ipython/contents/{+path}', 'has_pages': False, 'full_name': 'ipython/ipython', 'trees_url': 'https://api.github.com/repos/ipython/ipython/git/trees{/sha}', 'html_url': 'https://github.com/ipython/ipython', 'teams_url': 'https://api.github.com/repos/ipython/ipython/teams', 'milestones_url': 'https://api.github.com/repos/ipython/ipython/milestones{/number}', 'stargazers_count': 10903, 'collaborators_url': 'https://api.github.com/repos/ipython/ipython/collaborators{/collaborator}', 'forks_url': 'https://api.github.com/repos/ipython/ipython/forks', 'comments_url': 'https://api.github.com/repos/ipython/ipython/comments{/number}', 'description': 'Official repository for IPython itself. Other repos in the IPython organization contain things like the website, documentation builds, etc.', 'fork': False, 'clone_url': 'https://github.com/ipython/ipython.git', 'homepage': 'http://ipython.org'}, 'commit': {'parents': [{'sha': 'af798bd3daf94ee116d5be6dfea9893fe9460091', 'html_url': 'https://github.com/ipython/ipython/commit/af798bd3daf94ee116d5be6dfea9893fe9460091', 'url': 'https://api.github.com/repos/ipython/ipython/commits/af798bd3daf94ee116d5be6dfea9893fe9460091'}], 'sha': 'bda4ba81e9292cffb9852ef22c918d4e48bc5f64', 'author': {'id': 198396, 'organizations_url': 'https://api.github.com/users/srinivasreddy/orgs', 'url': 'https://api.github.com/users/srinivasreddy', 'received_events_url': 'https://api.github.com/users/srinivasreddy/received_events', 'html_url': 'https://github.com/srinivasreddy', 'site_admin': False, 'following_url': 'https://api.github.com/users/srinivasreddy/following{/other_user}', 'followers_url': 'https://api.github.com/users/srinivasreddy/followers', 'gravatar_id': '', 'starred_url': 'https://api.github.com/users/srinivasreddy/starred{/owner}{/repo}', 'type': 'User', 'events_url': 'https://api.github.com/users/srinivasreddy/events{/privacy}', 'repos_url': 'https://api.github.com/users/srinivasreddy/repos', 'avatar_url': 'https://avatars.githubusercontent.com/u/198396?v=3', 'subscriptions_url': 'https://api.github.com/users/srinivasreddy/subscriptions', 'gists_url': 'https://api.github.com/users/srinivasreddy/gists{/gist_id}', 'login': 'srinivasreddy'}, 'committer': {'id': 19864447, 'organizations_url': 'https://api.github.com/users/web-flow/orgs', 'url': 'https://api.github.com/users/web-flow', 'received_events_url': 'https://api.github.com/users/web-flow/received_events', 'html_url': 'https://github.com/web-flow', 'site_admin': False, 'following_url': 'https://api.github.com/users/web-flow/following{/other_user}', 'followers_url': 'https://api.github.com/users/web-flow/followers', 'gravatar_id': '', 'starred_url': 'https://api.github.com/users/web-flow/starred{/owner}{/repo}', 'type': 'User', 'events_url': 'https://api.github.com/users/web-flow/events{/privacy}', 'repos_url': 'https://api.github.com/users/web-flow/repos', 'avatar_url': 'https://avatars.githubusercontent.com/u/19864447?v=3', 'subscriptions_url': 'https://api.github.com/users/web-flow/subscriptions', 'gists_url': 'https://api.github.com/users/web-flow/gists{/gist_id}', 'login': 'web-flow'}, 'url': 'https://api.github.com/repos/ipython/ipython/commits/bda4ba81e9292cffb9852ef22c918d4e48bc5f64', 'comments_url': 'https://api.github.com/repos/ipython/ipython/commits/bda4ba81e9292cffb9852ef22c918d4e48bc5f64/comments', 'html_url': 'https://github.com/ipython/ipython/commit/bda4ba81e9292cffb9852ef22c918d4e48bc5f64', 'commit': {'message': 'remove python2 import statement', 'committer': {'name': 'GitHub', 'email': 'noreply@github.com', 'date': '2017-01-05T14:36:01Z'}, 'url': 'https://api.github.com/repos/ipython/ipython/git/commits/bda4ba81e9292cffb9852ef22c918d4e48bc5f64', 'comment_count': 0, 'tree': {'sha': '7093d0fff2138bd5eefc0931f23565f8d4fc31f1', 'url': 'https://api.github.com/repos/ipython/ipython/git/trees/7093d0fff2138bd5eefc0931f23565f8d4fc31f1'}, 'author': {'name': 'Srinivas Reddy Thatiparthy', 'email': 'srinivasreddy@users.noreply.github.com', 'date': '2017-01-05T14:36:01Z'}}}, 'context': 'codecov/patch', 'target_url': 'https://codecov.io/gh/ipython/ipython/compare/af798bd3daf94ee116d5be6dfea9893fe9460091...bda4ba81e9292cffb9852ef22c918d4e48bc5f64', 'branches': [], 'installation': {'id': 5138}, 'description': '0.00% of diff hit (target 0.00%)'} ':',payload)5PK%JAHdPPmeeseeksbox/.utils.py.swpb0VIM 8.0BdnXK $bussonniermatthiasMacBook-Pro.local~bussonniermatthias/dev/meeseeksbox/framework/meeseeksbox/utils.pyutf-8 3210#"! Utp ]Xhads "OL73 5 0 x g  { z "  ) $ Z  Ig[B6mayT,}IHGFgA65{N prepared = req.prepare() req = requests.Request(method, url, headers=headers) 'User-Agent': 'python/requests'} 'Host': 'api.github.com', 'Accept': ACCEPT_HEADER, headers = {'Authorization': 'Bearer {}'.format(tok.decode()), tok = jwt.encode(payload, key=self.rsadata, algorithm='RS256') }) 'iss': self.integration_id, 'exp': self.since + self.duration, 'iat': self.since, payload = dict({ self.since= int(datetime.datetime.now().timestamp()) def _integration_authenticated_request(self, method, url): self.idmap[repo['full_name']] = iid for repo in repositories['repositories']: 'GET', installation['repositories_url'], json=None).json() repositories = session.ghrequest( session = self.session(iid) iid = installation['id'] for installation in installations: installations = self.list_installations() """ to do cross repository operations. Build an organisation/repo -> installation_id mappingg in order to be able """ def _build_auth_id_mapping(self): return response.json() 'GET', "https://api.github.com/integration/installations") response = self._integration_authenticated_request( """ Todo: Pagination """ def list_installations(self): return Session(self.integration_id, self.rsadata, installation_id) def session(self, installation_id): self.idmap = {} # have new / deleted installations # TODO: this mapping is built at startup, we should update it when we self.rsadata = rsadata self.integration_id = integration_id self._token = None self.duration = 60*10 self.since = int(datetime.datetime.now().timestamp()) def __init__(self, integration_id, rsadata): class Authenticator: return """[`@{op}` commented]({original_url}): {body}""".format(op=original_poster, original_url=original_url, body=body) body = RELINK_RE.sub('{org}/{repo}\\1'.format(org=original_org, repo=original_repo), body) """ This should be improved to quote mention of people This, for now does only simple fixes, like link to the original comment. """def fix_comment_body(body, original_poster, original_url, original_org, original_repo): """.format(org=original_org, repo=original_repo, number=original_number, reporter=original_poster, requester=migration_requester) \nOriginally opened as {org}/{repo}#{number} by @{reporter}, migration requested by @{requester} """\n\n---- return body + \ body = RELINK_RE.sub('{org}/{repo}\\1'.format(org=original_org, repo=original_repo), body) """ This should be improved to quote mention of people This, for now does only simple fixes, like link to the original issue. """def fix_issue_body(body, original_poster, original_repo, original_org, original_number, migration_requester):RELINK_RE = re.compile('(?:(?<=[:,\s])|(?<=^))(#\d+)\\b')"""specific repository.PayACCEPT_HEADER = 'application/vnd.github.machine-man-preview+json,application/vnd.github.korra-preview'API_AACCEPT_HEADER = 'application/vnd.github.machine-man-preview+json,application/vnd.github.korra-preview'API_COLLABORATORS_TEMPLATE = 'https://api.github.com/repos/{org}/{repo}/collaborators/{username}/permission'import reimport requestsimport jsonimport datetimeimport jwt"""Utility functions to work with github."""ad\XKW8 B 5 ~ g H  n "  b 5 u l 4 N z:9 N* s]A'OB,na json=arguments) json json=arguments) return self.ghrequest('POST', 'https://api.github.com/repos/{}/{}/issues'.format(org, repo), raise ValueError('Assignees must be a list or a tuple') else: arguments['assignees'] = assignees if type(assignees) in (list, tuple): if assignees: raise ValueError('Labels must be a list of a tuple') else: arguments['labels'] = labels if type(labels) in (list, tuple): if labels: } "body": body, "title": title, arguments = { def create_issue(self, org:str, repo:str , title:str, body:str, *, labels=None, assignees=None): resp.raise_for_status() else: return resp.json() if resp.status_code == 200: resp = self.ghrequest('GET', get_collaborators_query, None) get_collaborators_query = 'https://api.github.com/repos/{org}/{repo}/collaborators'.format(org=org, repo=repo) def get_collaborator_list(self, org, repo): self.ghrequest('POST', comment_url, json={"body":body}) def post_comment(self, comment_url, body): return resp.json()['permission'] in ('admin', 'write') resp.raise_for_status() resp = self.ghrequest('GET', get_collaborators_query, None) get_collaborators_query = API_COLLABORATORS_TEMPLATE.format(org=org, repo=repo, username=username) """ finer grained decision. (application/vnd.github.korra-preview) with github which allows to get Right now this is a boolean, there is a new API Check if a user is collaborator on this repository """ def is_collaborator(self, org, repo, username): return response response.raise_for_status() response = s.send(prepare()) self.regen_token() if response.status_code == 401: response = s.send(prepare()) with requests.Session() as s: return req.prepare() req = requests.Request(method, url, headers=headers, json=json) 'User-Agent': 'python/requests'} 'Host': 'api.github.com', 'Accept': ACCEPT_HEADER, headers = {'Authorization': 'Bearer {}'.format(atk), atk = self.token() def prepare(): def ghrequest(self, method, url, json=None): raise ValueError(resp.content, url) except: self._token = json.loads(resp.content.decode())['token'] try: resp = self._integration_authenticated_request(method, url) url = 'https://api.github.com/installations/%s/access_tokens'%self.installation_id method = 'POST' def regen_token(self): return self._token self.regen_token() if (now > self.since + self.duration-60) or (self._token is None): now = datetime.datetime.now().timestamp() def token(self): self.installation_id = installation_id super().__init__(integration_id, rsadata) def __init__(self, integration_id, rsadata, installation_id):class Session(Authenticator): return s.send(prepared) with requests.Session() as s:ad#]u`\"! ^Y" + K C R M E ' r G /  ^R9-dXpK#t@?>=^8-,rE prepared = req.prepare() req = requests.Request(method, url, headers=headers) 'User-Agent': 'python/requests'} 'Host': 'api.github.com', 'Accept': ACCEPT_HEADER, headers = {'Authorization': 'Bearer {}'.format(tok.decode()), tok = jwt.encode(payload, key=self.rsadata, algorithm='RS256') }) 'iss': self.integration_id, 'exp': self.since + self.duration, 'iat': self.since, payload = dict({ self.since= int(datetime.datetime.now().timestamp()) def _integration_authenticated_request(self, method, url): self.idmap[repo['full_name']] = iid for repo in repositories['repositories']: 'GET', installation['repositories_url'], json=None).json() repositories = session.ghrequest( session = self.session(iid) iid = installation['id'] for installation in installations: installations = self.list_installations() """ to do cross repository operations. Build an organisation/repo -> installation_id mappingg in order to be able """ def _build_auth_id_mapping(self): return response.json() 'GET', "https://api.github.com/integration/installations") response = self._integration_authenticated_request( """ Todo: Pagination """ def list_installations(self): return self._session_class(self.integration_id, self.rsadata, installation_id) def session(self, installation_id): self._session_class = Session self.idmap = {} # have new / deleted installations # TODO: this mapping is built at startup, we should update it when we self.rsadata = rsadata self.integration_id = integration_id self._token = None self.duration = 60*10 self.since = int(datetime.datetime.now().timestamp()) def __init__(self, integration_id, rsadata): class Authenticator: return """[`@{op}` commented]({original_url}): {body}""".format(op=original_poster, original_url=original_url, body=body) body = RELINK_RE.sub('{org}/{repo}\\1'.format(org=original_org, repo=original_repo), body) """ This should be improved to quote mention of people This, for now does only simple fixes, like link to the original comment. """def fix_comment_body(body, original_poster, original_url, original_org, original_repo): """.format(org=original_org, repo=original_repo, number=original_number, reporter=original_poster, requester=migration_requester) \nOriginally opened as {org}/{repo}#{number} by @{reporter}, migration requested by @{requester} """\n\n---- return body + \ body = RELINK_RE.sub('{org}/{repo}\\1'.format(org=original_org, repo=original_repo), body) """ This should be improved to quote mention of people This, for now does only simple fixes, like link to the original issue. """def fix_issue_body(body, original_poster, original_repo, original_org, original_number, migration_requester):RELINK_RE = re.compile('(?:(?<=[:,\s])|(?<=^))(#\d+)\\b')"""specific repository.Pay attention to not relink things like foo#23 as they already point to aRegular expression to relink issues/pr comments correctly."""PK%JmwQBQBmeeseeksbox/.utils.py.un~VimUnDom5-Đ6]wXYgakK,lf$$$$XndC _ A";v;Xn^ AACCEPT_HEADER = 'application/vnd.github.machine-man-preview+json'5_ ";v;Xn^ PERMISSION_ACCEPT = '' 5_ ";v;Xn^ :PERMISSION_ACCEPT = 'application/vnd.github.korra-preview'5_   ?v?Xn^ APERMISSION_ACCEPT_HEADER = 'application/vnd.github.korra-preview'5_ @  ?v?Xn^ AACCEPT_HEADER = 'application/vnd.github.machine-man-preview+json' 5_ B  ?v?Xn^ gACCEPT_HEADER = 'application/vnd.github.machine-man-preview+json, application/vnd.github.korra-preview'5_   ?v?Xn^ PERMISSION_ACCEPT_HEADER = ''5_    ?v?Xn_#    import re5_ ?v?Xn_% 6/repos/:owner/:repo/collaborators/:username/permission5_ `?v?Xn_, aAPI_COLLABORATORS_TEMPLATE = 'https://api.github.com/repos/{org}/{repo}/collaborators/{username}'5_ 4?v?Xn_8 j /repos/:owner/:repo/collaborators/:username/permission5_    ?v?Xn_8 5_  Xn_g# if resp.status_code == 204:5_ Xn_g return True5_Xn_g% elif resp.status_code == 404:5_ Xn_h return False5_Xn_h else:5_ Xn_i# resp.raise_for_status()5_Xn_k resp.raise_for_status()5_Xn_l5_Xn_u resp.json()[]5_Xn_u resp.json()['']5_!Xn_y! resp.json()['permission']5_&Xn_|' resp.json()['permission'] in ()5_'Xn_}) resp.json()['permission'] in ('')5_-Xn_~. resp.json()['permission'] in ('admin')5_0Xn_2 resp.json()['permission'] in ('admin', '')5_Xn_7 resp.json()['permission'] in ('admin', 'write')5_9*Xnd% 8;* # have new / deleted installations5_ :Xnd, 9; self._session_class5_! ='Xnd3<?' def session(self, installation_id):5_ "!???vXnd: >@J return Session(self.integration_id, self.rsadata, installation_id)?@5_!#">??vXnd< =>5_"$#:>>vXnd@9:% self._session_class = Session5_#$:==vXndB :<:;5_  Xn_e" f resp.status_code == 204: return True$ lif resp.status_code == 404: return False lse:5PK%J3meeseeksbox/__init__.py""" MeeseeksBox Base of a framework to write stateless bots on GitHub. Mainly writte to use the (currently Beta) new GitHub "Integration" API, and handle authencation of user. """ import os import base64 from .core import Config from .core import MeeseeksBox version_info = (0, 0, 3) __version__ = '.'.join(map(str,version_info)) def load_config_from_env(): """ Load the configuration, for now stored in the environment """ config={} integration_id = os.environ.get('GITHUB_INTEGRATION_ID') botname = os.environ.get('GITHUB_BOT_NAME', None) if not integration_id: raise ValueError('Please set GITHUB_INTEGRATION_ID') if not botname: raise ValueError('Need to set a botname') if "@" in botname: print("Don't include @ in the botname !") botname = botname.replace('@','') at_botname = '@'+botname integration_id = int(integration_id) config['key'] = base64.b64decode(bytes(os.environ.get('B64KEY'), 'ASCII')) config['botname'] = botname config['at_botname'] = at_botname config['integration_id'] = integration_id config['webhook_secret'] = os.environ.get('WEBHOOK_SECRET') return Config(**config).validate() PKrfI00meeseeksbox/commands.py""" Define a few commands """ import random import os import subprocess import git import pipes import mock import sys #from friendlyautopep8 import run_on_cwd from .utils import Session, fix_issue_body, fix_comment_body from .scopes import admin, everyone @everyone def replyuser(*, session, payload, arguments): print("I'm replying to a user, look at me.") comment_url = payload['issue']['comments_url'] user = payload['comment']['user']['login'] c = random.choice( ("Helloooo @{user}, I'm Mr. Meeseeks! Look at me!", "Look at me, @{user}, I'm Mr. Meeseeks! ", "I'm Mr. Meeseek, @{user}, Look at meee ! ", ) ) session.post_comment(comment_url, c.format(user=user)) from textwrap import dedent @everyone def zen(*, session, payload, arguments): comment_url = payload['issue']['comments_url'] session.post_comment(comment_url, dedent( """ Zen of Pyton ([pep 20](https://www.python.org/dev/peps/pep-0020/)) ``` >>> import this Beautiful is better than ugly. Explicit is better than implicit. Simple is better than complex. Complex is better than complicated. Flat is better than nested. Sparse is better than dense. Readability counts. Special cases aren't special enough to break the rules. Although practicality beats purity. Errors should never pass silently. Unless explicitly silenced. In the face of ambiguity, refuse the temptation to guess. There should be one-- and preferably only one --obvious way to do it. Although that way may not be obvious at first unless you're Dutch. Now is better than never. Although never is often better than *right* now. If the implementation is hard to explain, it's a bad idea. If the implementation is easy to explain, it may be a good idea. Namespaces are one honking great idea -- let's do more of those! ``` """ )) @admin def replyadmin(*, session, payload, arguments): comment_url = payload['issue']['comments_url'] user = payload['issue']['user']['login'] session.post_comment(comment_url, "Hello @{user}. Waiting for your orders.".format(user=user)) @admin def pep8ify(*, session, payload, arguments): target_branch = arguments # collect initial payload prnumber = payload['issue']['number'] prtitle = payload['issue']['title'] org_name = payload['organization']['login'] repo_name = payload['repository']['name'] # collect extended payload on the PR print('== Collecting data on Pull-request...') r = session.ghrequest('GET', 'https://api.github.com/repos/{}/{}/pulls/{}'.format( org_name, repo_name, prnumber), json=None) pr_data = r.json() merge_sha = pr_data['merge_commit_sha'] # body = pr_data['body'] # clone locally # this process can take some time, regen token atk = session.token() if os.path.exists(repo_name): print('== Cleaning up previsous work... ') subprocess.run('rm -rf {}'.format(repo_name).split(' ')) print('== Done cleaning ') print('== Cloning current repository, this can take some time..') process = subprocess.run( ['git', 'clone', 'https://x-access-token:{}@github.com/{}/{}'.format(atk, org_name, repo_name)]) print('== Cloned..') process.check_returncode() subprocess.run('git config --global user.email ipy.bot@bot.com'.split(' ')) subprocess.run('git config --global user.name FriendlyBot'.split(' ')) # do the backport on local filesystem repo = git.Repo(repo_name) print('== Fetching branch to backport on ...') repo.remotes.origin.fetch('refs/heads/{}:workbranch'.format(target_branch)) repo.git.checkout('workbranch') print('== Fetching Commits to backport...') repo.remotes.origin.fetch('{mergesha}'.format( num=prnumber, mergesha=merge_sha)) print('== All has been fetched correctly') # write the commit message msg = "Autofix pep 8 of #%i: %s" % (prnumber, prtitle) + '\n\n' repo.git.commit('-m', msg) # Push the backported work remote_submit_branch = 'auto-backport-of-pr-{}'.format(prnumber) print("== Pushing work....:") repo.remotes.origin.push('workbranch:{}'.format(remote_submit_branch)) repo.git.checkout('master') repo.branches.workbranch.delete(repo, 'workbranch', force=True) # ToDO checkout master and get rid of branch # Make the PR on GitHub new_pr = session.ghrequest('POST', 'https://api.github.com/repos/{}/{}/pulls'.format(org_name, repo_name), json={ "title": "Backport PR #%i on branch %s" % (prnumber, target_branch), "body": msg, "head": "{}:{}".format(org_name, remote_submit_branch), "base": target_branch }) new_number = new_pr.json().get('number', None) print('Backported as PR', new_number) return new_pr.json() @admin def backport(session, payload, arguments): target_branch = arguments # collect initial payload prnumber = payload['issue']['number'] prtitle = payload['issue']['title'] org_name = payload['organization']['login'] repo_name = payload['repository']['name'] # collect extended payload on the PR print('== Collecting data on Pull-request...') r = session.ghrequest('GET', 'https://api.github.com/repos/{}/{}/pulls/{}'.format( org_name, repo_name, prnumber), json=None) pr_data = r.json() merge_sha = pr_data['merge_commit_sha'] body = pr_data['body'] # clone locally # this process can take some time, regen token atk = session.token() if os.path.exists(repo_name): print('== Cleaning up previsous work... ') subprocess.run('rm -rf {}'.format(repo_name).split(' ')) print('== Done cleaning ') print('== Cloning current repository, this can take some time..') process = subprocess.run( ['git', 'clone', 'https://x-access-token:{}@github.com/{}/{}'.format(atk, org_name, repo_name)]) print('== Cloned..') process.check_returncode() subprocess.run('git config --global user.email ipy.bot@bot.com'.split(' ')) subprocess.run('git config --global user.name FriendlyBot'.split(' ')) # do the backport on local filesystem repo = git.Repo(repo_name) print('== Fetching branch to backport on ...') repo.remotes.origin.fetch('refs/heads/{}:workbranch'.format(target_branch)) repo.git.checkout('workbranch') print('== Fetching Commits to backport...') repo.remotes.origin.fetch('{mergesha}'.format( num=prnumber, mergesha=merge_sha)) print('== All has been fetched correctly') # remove mentions from description, to avoid pings: description = body.replace('@', ' ').replace('#', ' ') print("Cherry-picking %s" % merge_sha) args = ('-m', '1', merge_sha) try: with mock.patch.dict('os.environ', {'GIT_EDITOR': 'true'}): repo.git.cherry_pick(*args) except Exception as e: print('\n' + e.stderr.decode('utf8', 'replace'), file=sys.stderr) print('\n' + repo.git.status(), file=sys.stderr) cmd = ' '.join(pipes.quote(arg) for arg in sys.argv) print('\nPatch did not apply. Resolve conflicts (add, not commit), then re-run `%s`' % cmd, file=sys.stderr) # write the commit message msg = "Backport PR #%i: %s" % (prnumber, prtitle) + '\n\n' + description repo.git.commit('--amend', '-m', msg) print("== PR #%i applied, with msg:" % prnumber) print() print(msg) print("== ") # Push the backported work remote_submit_branch = 'auto-backport-of-pr-{}'.format(prnumber) print("== Pushing work....:") repo.remotes.origin.push('workbranch:{}'.format(remote_submit_branch)) repo.git.checkout('master') repo.branches.workbranch.delete(repo, 'workbranch', force=True) # ToDO checkout master and get rid of branch # Make the PR on GitHub new_pr = session.ghrequest('POST', 'https://api.github.com/repos/{}/{}/pulls'.format(org_name, repo_name), json={ "title": "Backport PR #%i on branch %s" % (prnumber, target_branch), "body": msg, "head": "{}:{}".format(org_name, remote_submit_branch), "base": target_branch }) new_number = new_pr.json().get('number', None) print('Backported as PR', new_number) return new_pr.json() @admin def tag(*, session, payload, arguments): org = payload['organization']['login'] repo = payload['repository']['name'] num = payload.get('issue').get('number') url = "https://api.github.com/repos/{org}/{repo}/issues/{num}/labels".format(**locals()) tags = [arg.strip() for arg in arguments.split(',')] session.ghrequest('POST', url, json=tags) @admin def untag(*, session, payload, arguments): org = payload['organization']['login'] repo = payload['repository']['name'] num = payload.get('issue').get('number') tags = [arg.strip() for arg in arguments.split(',')] name = '{name}' url = "https://api.github.com/repos/{org}/{repo}/issues/{num}/labels/{name}".format(**locals()) for tag in tags: session.ghrequest('DELETE', url.format(name=tag)) @admin def migrate_issue_request(*, session:Session, payload:dict, arguments:str): """Todo: - Works through pagination of comments - Works through pagination of labels Link to non-migrated labels. """ if arguments.startswith('to '): arguments = arguments[3:] org_repo = arguments org, repo = arguments.split('/') target_session = yield org_repo if not target_session: session.post_comment(payload['issue']['comments_url'], "It appears that I can't do that") return issue_title = payload['issue']['title'] issue_body = payload['issue']['body'] original_org = payload['organization']['login'] original_repo = payload['repository']['name'] original_poster = payload['issue']['user']['login'] original_number = payload['issue']['number'] migration_requester = payload['comment']['user']['login'] request_id = payload['comment']['id'] original_labels = [l['name'] for l in payload['issue']['labels']] if original_labels: available_labels = target_session.ghrequest('GET', 'https://api.github.com/repos/{org}/{repo}/labels'.format( org=org, repo=repo), None).json() available_labels = [l['name'] for l in available_labels] migrate_labels = [l for l in original_labels if l in available_labels] not_set_labels = [l for l in original_labels if l not in available_labels] new_response = target_session.create_issue(org, repo, issue_title, fix_issue_body( issue_body, original_poster, original_repo, original_org, original_number, migration_requester), labels=migrate_labels ) new_issue = new_response.json() new_comment_url = new_issue['comments_url'] original_comments = session.ghrequest( 'GET', payload['issue']['comments_url'], None).json() for comment in original_comments: if comment['id'] == request_id: continue body = comment['body'] op = comment['user']['login'] url = comment['html_url'] target_session.post_comment(new_comment_url, body=fix_comment_body( body, op, url, original_org, original_repo)) if not_set_labels: body = "I was not able to apply the following label(s): %s " % ','.join( not_set_labels) target_session.post_comment(new_comment_url, body=body) session.post_comment(payload['issue'][ 'comments_url'], body='Done as {}/{}#{}.'.format(org, repo, new_issue['number'])) session.ghrequest('PATCH', payload['issue'][ 'url'], json={'state': 'closed'}) PK~%J!!meeseeksbox/core.pyimport re import os import hmac import tornado.web import tornado.httpserver import tornado.ioloop from .utils import Authenticator from yieldbreaker import YieldBreaker class Config: botname = None integration_id = None key = None botname = None at_botname = None integration_id = None webhook_secret = None def __init__(self, **kwargs): self.__dict__.update(kwargs) def validate(self): missing = [attr for attr in dir(self) if not attr.startswith( '_') and getattr(self, attr) is None] if missing: raise ValueError( 'The followingg configuration options are missing : {}'.format(missing)) return self def verify_signature(payload, signature, secret): """ Make sure hooks are encoded correctly """ expected = 'sha1=' + hmac.new(secret.encode('ascii'), payload, 'sha1').hexdigest() return hmac.compare_digest(signature, expected) class BaseHandler(tornado.web.RequestHandler): def error(self, message): self.set_status(500) self.write({'status': 'error', 'message': message}) def success(self, message='', payload={}): self.write({'status': 'success', 'message': message, 'data': payload}) class MainHandler(BaseHandler): def get(self): self.finish('No') def process_mentionning_comment(body, bot_re): """ Given a comment body and a bot name parse this into a tuple of (command, arguments) """ lines = body.splitlines() lines = [l.strip() for l in lines if bot_re.search(l)] lines = [bot_re.split(l)[-1].strip() for l in lines] command_args = [l.split(' ', 1) for l in lines] command_args = [c if len(c) > 1 else (c[0], None) for c in command_args] return command_args class WebHookHandler(MainHandler): def initialize(self, actions, config, auth, *args, **kwargs): self.actions = actions self.config = config self.auth = auth super().initialize(*args, **kwargs) print('Webhook initialize got', args, kwargs) def get(self): self.getfinish("Webhook alive and listening") def post(self): if not 'X-Hub-Signature' in self.request.headers: return self.error('WebHook not configured with secret') # TODO: Extract from X-GitHub-Event if not verify_signature(self.request.body, self.request.headers['X-Hub-Signature'], self.config.webhook_secret): return self.error('Cannot validate GitHub payload with ' 'provided WebHook secret') payload = tornado.escape.json_decode(self.request.body) org = payload.get('repository', {}).get('owner', {}).get('login') if hasattr(self.config, 'org_whitelist') and (org not in self.config.org_whitelist): print('Non allowed org:', org) self.error('Not allowed org.') sender = payload.get('sender', {}).get('login', {}) if hasattr(self.config, 'user_whitelist') and (sender not in self.config.user_whitelist): print('Not allowed user:', sender) self.error('Not allowed user.') action = payload.get("action", None) if payload.get('commits'): # TODO print("commits were likely pushed....") return if action: print('## dispatching request', self.request.headers.get('X-GitHub-Delivery')) return self.dispatch_action(action, payload) else: print('No action available for the webhook :', self.request.headers.get('X-GitHub-Delivery'), ':', payload) @property def mention_bot_re(self): botname = self.config.botname return re.compile('@?' + re.escape(botname) + '(?:\[bot\])?', re.IGNORECASE) def dispatch_action(self, type_, payload): botname = self.config.botname # new issue/PR opened if type_ == 'opened': issue = payload.get('issue', None) if not issue: print('request has no issue key:', payload) return self.error('Not really good, request has no issue') if issue: user = payload['issue']['user']['login'] if user == self.config.botname.lower() + '[bot]': return self.finish("Not responding to self") # todo dispatch on on-open # new comment created elif type_ == 'created': comment = payload.get('comment', None) installation = payload.get('installation', None) if comment: user = payload['comment']['user']['login'] if user == botname.lower() + '[bot]': print('Not responding to self') return self.finish("Not responding to self") if '[bot]' in user: print('Not responding to another bot') return self.finish("Not responding to another bot") body = payload['comment']['body'] print('Got a comment', body) if self.mention_bot_re.findall(body): self.dispatch_on_mention(body, payload, user) else: print('Was not mentioned', self.config.botname, body, '|', user) elif installation and installation.get('account'): print('we got a new installation maybe ?!', payload) return self.finish() else: print('not handled', payload) else: print("can't deal with ", type_, "yet") def dispatch_on_mention(self, body, payload, user): # to dispatch to commands installation_id = payload['installation']['id'] org = payload['organization']['login'] repo = payload['repository']['name'] session = self.auth.session(installation_id) is_admin = session.is_collaborator(org, repo, user) command_args = process_mentionning_comment(body, self.mention_bot_re) for (command, arguments) in command_args: print(" :: treating", command, arguments) handler = self.actions.get(command, None) if handler: print(" :: testing who can use ", str(handler)) if ((handler.scope == 'admin') and is_admin) or (handler.scope == 'everyone'): print(" :: authorisation granted ", handler.scope) maybe_gen = handler( session=session, payload=payload, arguments=arguments) import types if type(maybe_gen) == types.GeneratorType: gen = YieldBreaker(maybe_gen) for org_repo in gen: torg, trepo = org_repo.split('/') session_id = self.auth.idmap.get(org_repo) if session_id: target_session = self.auth.session(session_id) if target_session.is_collaborator(torg, trepo, user): gen.send(target_session) else: gen.send(None) else: print('org/repo not found', org_repo, self.auth.id_map) gen.send(None) else: print('I Cannot let you do that') else: print('unnknown command', command) class MeeseeksBox: def __init__(self, commands, config): self.commands = commands self.port = int(os.environ.get('PORT', 5000)) self.application = None self.config = config self.auth = Authenticator(self.config.integration_id, self.config.key) self.auth._build_auth_id_mapping() def start(self): self.application = tornado.web.Application([ (r"/", MainHandler), (r"/webhook", WebHookHandler, {'actions': self.commands, 'config': self.config, 'auth': self.auth}) ]) tornado.httpserver.HTTPServer(self.application).listen(self.port) tornado.ioloop.IOLoop.instance().start() PKrfI(]meeseeksbox/scopes.py""" Define various scopes """ def admin(function): function.scope='admin' return function def everyone(function): function.scope='everyone' return function PK%Jm!'meeseeksbox/utils.py""" Utility functions to work with github. """ import jwt import datetime import json import requests import re API_COLLABORATORS_TEMPLATE = 'https://api.github.com/repos/{org}/{repo}/collaborators/{username}/permission' ACCEPT_HEADER = 'application/vnd.github.machine-man-preview+json,application/vnd.github.korra-preview' """ Regular expression to relink issues/pr comments correctly. Pay attention to not relink things like foo#23 as they already point to a specific repository. """ RELINK_RE = re.compile('(?:(?<=[:,\s])|(?<=^))(#\d+)\\b') def fix_issue_body(body, original_poster, original_repo, original_org, original_number, migration_requester): """ This, for now does only simple fixes, like link to the original issue. This should be improved to quote mention of people """ body = RELINK_RE.sub('{org}/{repo}\\1'.format(org=original_org, repo=original_repo), body) return body + \ """\n\n---- \nOriginally opened as {org}/{repo}#{number} by @{reporter}, migration requested by @{requester} """.format(org=original_org, repo=original_repo, number=original_number, reporter=original_poster, requester=migration_requester) def fix_comment_body(body, original_poster, original_url, original_org, original_repo): """ This, for now does only simple fixes, like link to the original comment. This should be improved to quote mention of people """ body = RELINK_RE.sub('{org}/{repo}\\1'.format(org=original_org, repo=original_repo), body) return """[`@{op}` commented]({original_url}): {body}""".format(op=original_poster, original_url=original_url, body=body) class Authenticator: def __init__(self, integration_id, rsadata): self.since = int(datetime.datetime.now().timestamp()) self.duration = 60*10 self._token = None self.integration_id = integration_id self.rsadata = rsadata # TODO: this mapping is built at startup, we should update it when we # have new / deleted installations self.idmap = {} self._session_class = Session def session(self, installation_id): return self._session_class(self.integration_id, self.rsadata, installation_id) def list_installations(self): """ Todo: Pagination """ response = self._integration_authenticated_request( 'GET', "https://api.github.com/integration/installations") return response.json() def _build_auth_id_mapping(self): """ Build an organisation/repo -> installation_id mappingg in order to be able to do cross repository operations. """ installations = self.list_installations() for installation in installations: iid = installation['id'] session = self.session(iid) repositories = session.ghrequest( 'GET', installation['repositories_url'], json=None).json() for repo in repositories['repositories']: self.idmap[repo['full_name']] = iid def _integration_authenticated_request(self, method, url): self.since= int(datetime.datetime.now().timestamp()) payload = dict({ 'iat': self.since, 'exp': self.since + self.duration, 'iss': self.integration_id, }) tok = jwt.encode(payload, key=self.rsadata, algorithm='RS256') headers = {'Authorization': 'Bearer {}'.format(tok.decode()), 'Accept': ACCEPT_HEADER, 'Host': 'api.github.com', 'User-Agent': 'python/requests'} req = requests.Request(method, url, headers=headers) prepared = req.prepare() with requests.Session() as s: return s.send(prepared) class Session(Authenticator): def __init__(self, integration_id, rsadata, installation_id): super().__init__(integration_id, rsadata) self.installation_id = installation_id def token(self): now = datetime.datetime.now().timestamp() if (now > self.since + self.duration-60) or (self._token is None): self.regen_token() return self._token def regen_token(self): method = 'POST' url = 'https://api.github.com/installations/%s/access_tokens'%self.installation_id resp = self._integration_authenticated_request(method, url) try: self._token = json.loads(resp.content.decode())['token'] except: raise ValueError(resp.content, url) def ghrequest(self, method, url, json=None): def prepare(): atk = self.token() headers = {'Authorization': 'Bearer {}'.format(atk), 'Accept': ACCEPT_HEADER, 'Host': 'api.github.com', 'User-Agent': 'python/requests'} req = requests.Request(method, url, headers=headers, json=json) return req.prepare() with requests.Session() as s: response = s.send(prepare()) if response.status_code == 401: self.regen_token() response = s.send(prepare()) response.raise_for_status() return response def is_collaborator(self, org, repo, username): """ Check if a user is collaborator on this repository Right now this is a boolean, there is a new API (application/vnd.github.korra-preview) with github which allows to get finer grained decision. """ get_collaborators_query = API_COLLABORATORS_TEMPLATE.format(org=org, repo=repo, username=username) resp = self.ghrequest('GET', get_collaborators_query, None) resp.raise_for_status() return resp.json()['permission'] in ('admin', 'write') def post_comment(self, comment_url, body): self.ghrequest('POST', comment_url, json={"body":body}) def get_collaborator_list(self, org, repo): get_collaborators_query = 'https://api.github.com/repos/{org}/{repo}/collaborators'.format(org=org, repo=repo) resp = self.ghrequest('GET', get_collaborators_query, None) if resp.status_code == 200: return resp.json() else: resp.raise_for_status() def create_issue(self, org:str, repo:str , title:str, body:str, *, labels=None, assignees=None): arguments = { "title": title, "body": body, } if labels: if type(labels) in (list, tuple): arguments['labels'] = labels else: raise ValueError('Labels must be a list of a tuple') if assignees: if type(assignees) in (list, tuple): arguments['assignees'] = assignees else: raise ValueError('Assignees must be a list or a tuple') return self.ghrequest('POST', 'https://api.github.com/repos/{}/{}/issues'.format(org, repo), json=arguments) PK!H;@QP!meeseeksbox-0.0.3.dist-info/WHEEL1 0 RZq+D-Dv;_[*7Fp ܦpv/fݞoL(*IPK!HVA$meeseeksbox-0.0.3.dist-info/METADATAW]o}_1@ݤ8Gca+r)5evRآ=$eJM 83gfxL+봩赸ZPS٭W{}Fo]SҶ e.!єj^5nqr>oVQbʓZ%ֿDU\h|n,Js-]6Ι檔XjemOKrg7B:3YPo [E][Q)_-/>h䍭dj[*u:\o,&y>|ضfme2UlUVEB&%$Ɍ%Wr~cK#!m.=iG+;"Y$ǪOC=lۨ5oV*V$ylo,~z% y1߻(ĸTgx{Zhuș85S.76Nb_[2jZ|GHgVV~FړMSb.B\L7%dlEWkTa?E`ŒQAr h[+i(̈r;{wb}TEa/Zq \DߍS?TL_-'%]ٝאP y 5U9h`;smo[F-3 kYs:RٵJgϫN4Gdw0K폑$L/ieekVxݛ"'⯘☓: qmԜ)6^7.K|FmmPFqQ`^BLuA0^5jj>{& :oHPjEs_z+{hX4S"&U4*hH@mKXP]oE:nQr-W\܁x `pBKt3vE "k2 n[Aᕚu*=qmqPLe2 hx!xa^ațχPpn[&(wXՌ78W˶0]7EGE;?LBV&!,4y D?2ӑNwhHlקxp]js槧OOqwvv6ǽa%7ഭ0긎KYs:u(B/{pQfĐ0`2TޞWVs*?[e›@@^ghRJ|6q<3KcUb.Ќ3t>qLf+:}Xttj4ѻG$p:oDnnc=Y0ZG¤s)qrEI"meeseeksbox-0.0.3.dist-info/RECORD}ɲ@}ȸȂIA@AEy UTM%/벤eYьat?0mRgx7)+A8@Z} ɭ>"&Gq[a/]:R}ižw P~;$횑\iGN/\*lՐy3jhHp3ϽeEaXIKJk ՐJq"Bqmeeseeksbox-0.0.3.dist-info/RECORDPK _s