Last active
April 11, 2022 18:59
-
-
Save rafen/3f2045a13a226cc4c17632bd8fb992ae to your computer and use it in GitHub Desktop.
GitHub webhook django view + tests
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| from unittest.mock import patch | |
| from django.urls import reverse | |
| from django.tests import TestCase | |
| class TestGitHubWebhookView(TestCase): | |
| @patch('my_app.views.GitHubWebhookView.ping') | |
| def test_view_ping_event(self, mock_ping): | |
| # GIVEN a mock response for ping | |
| mock_ping.return_value = {'status': 'ok'} | |
| # WHEN webhook ping event is received | |
| response = self.client.post( | |
| reverse('landing_page_github_webhook'), | |
| data={ | |
| "ref":"refs/heads/main", | |
| "before":"0000000000000000000000000000000000000000", | |
| "after":"67fdba207ec25322be31ba9ddcdc0ada4e46614e", | |
| }, | |
| HTTP_X_HUB_SIGNATURE='sha1=849c611aae4d2b8a05bb815b42d201b2fc61a0e9', | |
| HTTP_X_GITHUB_EVENT='ping', | |
| content_type="application/json" | |
| ) | |
| # THEN, ping function is executed if all security checks pass | |
| mock_ping.assert_called_once() | |
| @patch('my_app.views.os.system') | |
| def test_view_push_event(self, mock_system): | |
| # GIVEN a mock response for system | |
| mock_system.return_value = 0 | |
| # WHEN webhook push event is received | |
| response = self.client.post( | |
| reverse('landing_page_github_webhook'), | |
| data={ | |
| "ref":"refs/heads/main", | |
| "before":"0000000000000000000000000000000000000000", | |
| "after":"67fdba207ec25322be31ba9ddcdc0ada4e46614e", | |
| }, | |
| HTTP_X_HUB_SIGNATURE='sha1=849c611aae4d2b8a05bb815b42d201b2fc61a0e9', | |
| HTTP_X_GITHUB_EVENT='push', | |
| content_type="application/json" | |
| ) | |
| # THEN, system function is executed if all security checks pass | |
| mock_system.assert_called_once() | |
| #TODO: add failure cases |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| path('webhooks/github-XXXXX/', GitHubWebhookView.as_view(), | |
| name='landing_page_github_webhook'), |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| import hashlib | |
| import hmac | |
| import json | |
| import os | |
| from django.views.generic import View | |
| from django.utils.decorators import method_decorator | |
| from django.views.decorators.csrf import csrf_exempt | |
| from django.http import HttpResponseBadRequest, JsonResponse | |
| class GitHubWebhookView(View): | |
| secret = 'XXXXX' | |
| allowed_events = [ | |
| 'ping', | |
| 'push', | |
| ] | |
| def get_secret(self): | |
| return self.secret | |
| def get_allowed_events(self): | |
| return self.allowed_events | |
| @method_decorator(csrf_exempt) | |
| def dispatch(self, request, *args, **kwargs): | |
| return super().dispatch(request, *args, **kwargs) | |
| def post(self, request, *args, **kwargs): | |
| secret = self.get_secret() | |
| if 'HTTP_X_HUB_SIGNATURE' not in request.META: | |
| return HttpResponseBadRequest('Request does not contain X-GITHUB-SIGNATURE header') | |
| if 'HTTP_X_GITHUB_EVENT' not in request.META: | |
| return HttpResponseBadRequest('Request does not contain X-GITHUB-EVENT header') | |
| digest_name, signature = request.META['HTTP_X_HUB_SIGNATURE'].split('=') | |
| if digest_name != 'sha1': | |
| return HttpResponseBadRequest('Unsupported X-HUB-SIGNATURE digest mode found: {}'.format( | |
| digest_name | |
| )) | |
| mac = hmac.new( | |
| secret.encode('utf-8'), | |
| msg=request.body, | |
| digestmod=hashlib.sha1 | |
| ) | |
| if not hmac.compare_digest(mac.hexdigest(), signature): | |
| return HttpResponseBadRequest('Invalid X-HUB-SIGNATURE header found') | |
| event = request.META['HTTP_X_GITHUB_EVENT'] | |
| if event not in self.get_allowed_events(): | |
| return HttpResponseBadRequest('Unsupported X-GITHUB-EVENT header found: {}'.format(event)) | |
| handler = getattr(self, event, None) | |
| if not handler: | |
| return HttpResponseBadRequest('Unsupported X-GITHUB-EVENT header found: {}'.format(event)) | |
| payload = json.loads(request.body.decode('utf-8')) | |
| response = handler(payload, request, *args, **kwargs) | |
| return JsonResponse(response) | |
| def ping(self, payload, request, *args, **kwargs): | |
| return { | |
| 'status': 'ok', | |
| } | |
| def push(self, payload, request, *args, **kwargs): | |
| if payload.get('ref') in ['refs/heads/main', 'refs/heads/develop']: | |
| # Update code of landing pages | |
| os.system('~/update_landing.sh') | |
| return { | |
| 'status': 'ok', | |
| } |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment