From 5957307f2fbe87c5d200ed43d5ee236ddf8125aa Mon Sep 17 00:00:00 2001 From: Mathew Payne <2772944+GeekMasher@users.noreply.github.com> Date: Tue, 6 Aug 2024 09:53:29 +0000 Subject: [PATCH] feat: Fork Flask GitHubApp project --- ghasreview/flask_githubapp/README.md | 10 + ghasreview/flask_githubapp/__init__.py | 19 ++ ghasreview/flask_githubapp/core.py | 245 +++++++++++++++++++++++++ 3 files changed, 274 insertions(+) create mode 100644 ghasreview/flask_githubapp/README.md create mode 100644 ghasreview/flask_githubapp/__init__.py create mode 100644 ghasreview/flask_githubapp/core.py diff --git a/ghasreview/flask_githubapp/README.md b/ghasreview/flask_githubapp/README.md new file mode 100644 index 0000000..205c66b --- /dev/null +++ b/ghasreview/flask_githubapp/README.md @@ -0,0 +1,10 @@ +# Flask GitHub App + +This is a fork of the `flask-githubapp` project, which is a Flask extension for building GitHub Apps. + +The original project is available at: https://pypi.org/project/Flask-GitHubApp/ + +## The Issue + +- Deprecation of the `_app_ctx_stack.top` + - https://flask.palletsprojects.com/en/3.0.x/changes/#version-2-3-0 diff --git a/ghasreview/flask_githubapp/__init__.py b/ghasreview/flask_githubapp/__init__.py new file mode 100644 index 0000000..d14bbce --- /dev/null +++ b/ghasreview/flask_githubapp/__init__.py @@ -0,0 +1,19 @@ +from .core import ( + GitHubApp +) + +__version__ = "1.0.0" + +__all__ = ['GitHubApp'] + +# Set default logging handler to avoid "No handler found" warnings. +import logging +from logging import NullHandler + +# Set initial level to WARN. Users must manually enable logging for +# flask_githubapp to see our logging. +rootlogger = logging.getLogger(__name__) +rootlogger.addHandler(NullHandler()) + +if rootlogger.level == logging.NOTSET: + rootlogger.setLevel(logging.WARN) \ No newline at end of file diff --git a/ghasreview/flask_githubapp/core.py b/ghasreview/flask_githubapp/core.py new file mode 100644 index 0000000..8361978 --- /dev/null +++ b/ghasreview/flask_githubapp/core.py @@ -0,0 +1,245 @@ +"""Flask extension for rapid GitHub app development""" +import hmac +import logging + +from flask import abort, current_app, jsonify, make_response, request +from github3 import GitHub, GitHubEnterprise +from werkzeug.exceptions import BadRequest + +LOG = logging.getLogger(__name__) + +STATUS_FUNC_CALLED = 'HIT' +STATUS_NO_FUNC_CALLED = 'MISS' + + +class GitHubAppError(Exception): + pass + + +class GitHubAppValidationError(GitHubAppError): + pass + + +class GitHubApp(object): + """The GitHubApp object provides the central interface for interacting GitHub hooks + and creating GitHub app clients. + + GitHubApp object allows using the "on" decorator to make GitHub hooks to functions + and provides authenticated github3.py clients for interacting with the GitHub API. + + Keyword Arguments: + app {Flask object} -- App instance - created with Flask(__name__) (default: {None}) + """ + def __init__(self, app=None): + self._hook_mappings = {} + if app is not None: + self.init_app(app) + + def init_app(self, app): + """Initializes GitHubApp app by setting configuration variables. + + The GitHubApp instance is given the following configuration variables by calling on Flask's configuration: + + `GITHUBAPP_ID`: + + GitHub app ID as an int (required). + Default: None + + `GITHUBAPP_KEY`: + + Private key used to sign access token requests as bytes or utf-8 encoded string (required). + Default: None + + `GITHUBAPP_SECRET`: + + Secret used to secure webhooks as bytes or utf-8 encoded string (required). set to `False` to disable + verification (not recommended for production). + Default: None + + `GITHUBAPP_URL`: + + URL of GitHub API (used for GitHub Enterprise) as a string. + Default: None + + `GITHUBAPP_ROUTE`: + + Path used for GitHub hook requests as a string. + Default: '/' + """ + required_settings = ['GITHUBAPP_ID', 'GITHUBAPP_KEY', 'GITHUBAPP_SECRET'] + for setting in required_settings: + if not setting in app.config: + raise RuntimeError("Flask-GitHubApp requires the '%s' config var to be set" % setting) + + app.add_url_rule(app.config.get('GITHUBAPP_ROUTE', '/'), + view_func=self._flask_view_func, + methods=['POST']) + + @property + def id(self): + return current_app.config['GITHUBAPP_ID'] + + @property + def key(self): + key = current_app.config['GITHUBAPP_KEY'] + if hasattr(key, 'encode'): + key = key.encode('utf-8') + return key + + @property + def secret(self): + secret = current_app.config['GITHUBAPP_SECRET'] + if hasattr(secret, 'encode'): + secret = secret.encode('utf-8') + return secret + + @property + def _api_url(self): + return current_app.config['GITHUBAPP_URL'] + + @property + def client(self): + """Unauthenticated GitHub client""" + if current_app.config.get('GITHUBAPP_URL'): + return GitHubEnterprise(current_app.config['GITHUBAPP_URL']) + return GitHub() + + @property + def payload(self): + """GitHub hook payload""" + if request and request.json and 'installation' in request.json: + return request.json + + raise RuntimeError('Payload is only available in the context of a GitHub hook request') + + @property + def installation_client(self): + """GitHub client authenticated as GitHub app installation""" + # Get the current application context + ctx = current_app.app_context() + if ctx is not None: + if not hasattr(ctx, 'githubapp_installation'): + client = self.client + client.login_as_app_installation(self.key, + self.id, + self.payload['installation']['id']) + ctx.githubapp_installation = client + return ctx.githubapp_installation + + @property + def app_client(self): + """GitHub client authenticated as GitHub app""" + ctx = current_app.app_context() + if ctx is not None: + if not hasattr(ctx, 'githubapp_app'): + client = self.client + client.login_as_app(self.key, + self.id) + ctx.githubapp_app = client + return ctx.githubapp_app + + @property + def installation_token(self): + return self.installation_client.session.auth.token + + def on(self, event_action): + """Decorator routes a GitHub hook to the wrapped function. + + Functions decorated as a hook recipient are registered as the function for the given GitHub event. + + @github_app.on('issues.opened') + def cruel_closer(): + owner = github_app.payload['repository']['owner']['login'] + repo = github_app.payload['repository']['name'] + num = github_app.payload['issue']['id'] + issue = github_app.installation_client.issue(owner, repo, num) + issue.create_comment('Could not replicate.') + issue.close() + + Arguments: + event_action {str} -- Name of the event and optional action (separated by a period), e.g. 'issues.opened' or + 'pull_request' + """ + def decorator(f): + if event_action not in self._hook_mappings: + self._hook_mappings[event_action] = [f] + else: + self._hook_mappings[event_action].append(f) + + # make sure the function can still be called normally (e.g. if a user wants to pass in their + # own Context for whatever reason). + return f + + return decorator + + def _validate_request(self): + if not request.is_json: + raise GitHubAppValidationError('Invalid HTTP Content-Type header for JSON body ' + '(must be application/json or application/*+json).') + + try: + request.json + except BadRequest: + raise GitHubAppValidationError('Invalid HTTP body (must be JSON).') + + event = request.headers.get('X-GitHub-Event') + + if event is None: + raise GitHubAppValidationError('Missing X-GitHub-Event HTTP header.') + + action = request.json.get('action') + + return event, action + + def _flask_view_func(self): + functions_to_call = [] + calls = {} + + try: + event, action = self._validate_request() + except GitHubAppValidationError as e: + LOG.error(e) + error_response = make_response(jsonify(status='ERROR', description=str(e)), + 400) + return abort(error_response) + + if current_app.config['GITHUBAPP_SECRET'] is not False: + self._verify_webhook() + + if event in self._hook_mappings: + functions_to_call += self._hook_mappings[event] + + if action: + event_action = '.'.join([event, action]) + if event_action in self._hook_mappings: + functions_to_call += self._hook_mappings[event_action] + + if functions_to_call: + for function in functions_to_call: + calls[function.__name__] = function() + status = STATUS_FUNC_CALLED + else: + status = STATUS_NO_FUNC_CALLED + return jsonify({'status': status, + 'calls': calls}) + + def _verify_webhook(self): + signature_header ='X-Hub-Signature-256' + signature_header_legacy = 'X-Hub-Signature' + + if request.headers.get(signature_header): + signature = request.headers[signature_header].split('=')[1] + digestmod = 'sha256' + elif request.headers.get(signature_header_legacy): + signature = request.headers[signature_header_legacy].split('=')[1] + digestmod = 'sha1' + else: + LOG.warning('Signature header missing. Configure your GitHub App with a secret or set GITHUBAPP_SECRET' + 'to False to disable verification.') + return abort(400) + + mac = hmac.new(self.secret, msg=request.data, digestmod=digestmod) + + if not hmac.compare_digest(mac.hexdigest(), signature): + LOG.warning('GitHub hook signature verification failed.') + return abort(400) \ No newline at end of file