harmacist/healthchecks
1
0
Fork 0
forked from GithubBackups/healthchecks
Code Issues Pull Requests Projects Releases Wiki Activity

Project code in URL for the "Add Pushbullet" page. cc: #336

Browse Source
This commit is contained in:
Pēteris Caune 2020-02-21 17:31:17 +02:00
parent 44819cb555
commit 38bd84cc91
No known key found for this signature in database
GPG Key ID: E28D7679E9A9EDE2
5 changed files with 122 additions and 74 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

46
hc/front/tests/test_add_pushbullet.py
View File

@ -1,14 +1,12 @@
import json
from django.test.utils import override_settings
from hc.api.models import Channel
from hc.test import BaseTestCase
from mock import patch
@override_settings(PUSHBULLET_CLIENT_ID="t1", PUSHBULLET_CLIENT_SECRET="s1")
class AddPushbulletTestCase(BaseTestCase):
url = "/integrations/add_pushbullet/"
def setUp(self):
super(AddPushbulletTestCase, self).setUp()
self.url = "/projects/%s/add_pushbullet/" % self.project.code
def test_instructions_work(self):
self.client.login(username="alice@example.org", password="password")
@ -17,46 +15,10 @@ class AddPushbulletTestCase(BaseTestCase):
self.assertContains(r, "Connect Pushbullet")
# There should now be a key in session
self.assertTrue("pushbullet" in self.client.session)
self.assertTrue("add_pushbullet" in self.client.session)
@override_settings(PUSHBULLET_CLIENT_ID=None)
def test_it_requires_client_id(self):
self.client.login(username="alice@example.org", password="password")
r = self.client.get(self.url)
self.assertEqual(r.status_code, 404)
@patch("hc.front.views.requests.post")
def test_it_handles_oauth_response(self, mock_post):
session = self.client.session
session["pushbullet"] = "foo"
session.save()
oauth_response = {"access_token": "test-token"}
mock_post.return_value.text = json.dumps(oauth_response)
mock_post.return_value.json.return_value = oauth_response
url = self.url + "?code=12345678&state=foo"
self.client.login(username="alice@example.org", password="password")
r = self.client.get(url, follow=True)
self.assertRedirects(r, "/integrations/")
self.assertContains(r, "The Pushbullet integration has been added!")
ch = Channel.objects.get()
self.assertEqual(ch.value, "test-token")
self.assertEqual(ch.project, self.project)
# Session should now be clean
self.assertFalse("pushbullet" in self.client.session)
def test_it_avoids_csrf(self):
session = self.client.session
session["pushbullet"] = "foo"
session.save()
url = self.url + "?code=12345678&state=bar"
self.client.login(username="alice@example.org", password="password")
r = self.client.get(url)
self.assertEqual(r.status_code, 400)

63
hc/front/tests/test_add_pushbullet_complete.py Normal file
View File

@ -0,0 +1,63 @@
import json
from django.test.utils import override_settings
from hc.api.models import Channel
from hc.test import BaseTestCase
from mock import patch
@override_settings(PUSHBULLET_CLIENT_ID="t1", PUSHBULLET_CLIENT_SECRET="s1")
class AddPushbulletTestCase(BaseTestCase):
url = "/integrations/add_pushbullet/"
@patch("hc.front.views.requests.post")
def test_it_handles_oauth_response(self, mock_post):
session = self.client.session
session["add_pushbullet"] = ("foo", str(self.project.code))
session.save()
oauth_response = {"access_token": "test-token"}
mock_post.return_value.text = json.dumps(oauth_response)
mock_post.return_value.json.return_value = oauth_response
url = self.url + "?code=12345678&state=foo&project=%s" % self.project.code
self.client.login(username="alice@example.org", password="password")
r = self.client.get(url, follow=True)
self.assertRedirects(r, self.channels_url)
self.assertContains(r, "The Pushbullet integration has been added!")
ch = Channel.objects.get()
self.assertEqual(ch.value, "test-token")
self.assertEqual(ch.project, self.project)
# Session should now be clean
self.assertFalse("add_pushbullet" in self.client.session)
def test_it_avoids_csrf(self):
session = self.client.session
session["pushbullet"] = ("foo", str(self.project.code))
session.save()
url = self.url + "?code=12345678&state=bar&project=%s" % self.project.code
self.client.login(username="alice@example.org", password="password")
r = self.client.get(url)
self.assertEqual(r.status_code, 403)
@patch("hc.front.views.requests.post")
def test_it_handles_denial(self, mock_post):
session = self.client.session
session["add_pushbullet"] = ("foo", str(self.project.code))
session.save()
self.client.login(username="alice@example.org", password="password")
r = self.client.get(self.url + "?error=access_denied", follow=True)
self.assertRedirects(r, self.channels_url)
self.assertContains(r, "Pushbullet setup was cancelled")
self.assertEqual(Channel.objects.count(), 0)
# Session should now be clean
self.assertFalse("add_pushbullet" in self.client.session)

7
hc/front/urls.py
View File

@ -28,7 +28,11 @@ channel_urls = [
path("add_pdc/<str:state>/", views.add_pdc, name="hc-add-pdc-state"),
path("add_slack/", views.add_slack, name="hc-add-slack"),
path("add_slack_btn/", views.add_slack_btn, name="hc-add-slack-btn"),
path("add_pushbullet/", views.add_pushbullet, name="hc-add-pushbullet"),
path(
"add_pushbullet/",
views.add_pushbullet_complete,
name="hc-add-pushbullet-complete",
),
path("add_discord/", views.add_discord, name="hc-add-discord"),
path("add_pushover/", views.add_pushover, name="hc-add-pushover"),
path("telegram/bot/", views.telegram_bot, name="hc-telegram-webhook"),
@ -60,6 +64,7 @@ project_urls = [
path("add_pagertree/", views.add_pagertree, name="hc-add-pagertree"),
path("add_pd/", views.add_pd, name="hc-add-pd"),
path("add_prometheus/", views.add_prometheus, name="hc-add-prometheus"),
path("add_pushbullet/", views.add_pushbullet, name="hc-add-pushbullet"),
path("add_shell/", views.add_shell, name="hc-add-shell"),
path("add_sms/", views.add_sms, name="hc-add-sms"),
path("add_victorops/", views.add_victorops, name="hc-add-victorops"),

78
hc/front/views.py
View File

@ -1101,45 +1101,20 @@ def add_slack_btn(request):
@login_required
def add_pushbullet(request):
def add_pushbullet(request, code):
if settings.PUSHBULLET_CLIENT_ID is None:
raise Http404("pushbullet integration is not available")
if "code" in request.GET:
code = _get_validated_code(request, "pushbullet")
if code is None:
return HttpResponseBadRequest()
project = _get_project_for_user(request, code)
redirect_uri = settings.SITE_ROOT + reverse("hc-add-pushbullet-complete")
result = requests.post(
"https://api.pushbullet.com/oauth2/token",
{
"client_id": settings.PUSHBULLET_CLIENT_ID,
"client_secret": settings.PUSHBULLET_CLIENT_SECRET,
"code": code,
"grant_type": "authorization_code",
},
)
doc = result.json()
if "access_token" in doc:
channel = Channel(kind="pushbullet", project=request.project)
channel.user = request.project.owner
channel.value = doc["access_token"]
channel.save()
channel.assign_all_checks()
messages.success(request, "The Pushbullet integration has been added!")
else:
messages.warning(request, "Something went wrong")
return redirect("hc-channels")
redirect_uri = settings.SITE_ROOT + reverse("hc-add-pushbullet")
state = get_random_string()
authorize_url = "https://www.pushbullet.com/authorize?" + urlencode(
{
"client_id": settings.PUSHBULLET_CLIENT_ID,
"redirect_uri": redirect_uri,
"response_type": "code",
"state": _prepare_state(request, "pushbullet"),
"state": state,
}
)
@ -1148,9 +1123,52 @@ def add_pushbullet(request):
"project": request.project,
"authorize_url": authorize_url,
}
request.session["add_pushbullet"] = (state, str(project.code))
return render(request, "integrations/add_pushbullet.html", ctx)
@login_required
def add_pushbullet_complete(request):
if settings.PUSHBULLET_CLIENT_ID is None:
raise Http404("pushbullet integration is not available")
if "add_pushbullet" not in request.session:
return HttpResponseForbidden()
state, code = request.session.pop("add_pushbullet")
project = _get_project_for_user(request, code)
if request.GET.get("error") == "access_denied":
messages.warning(request, "Pushbullet setup was cancelled")
return redirect("hc-p-channels", project.code)
if request.GET.get("state") != state:
return HttpResponseForbidden()
result = requests.post(
"https://api.pushbullet.com/oauth2/token",
{
"client_id": settings.PUSHBULLET_CLIENT_ID,
"client_secret": settings.PUSHBULLET_CLIENT_SECRET,
"code": request.GET.get("code"),
"grant_type": "authorization_code",
},
)
doc = result.json()
if "access_token" in doc:
channel = Channel(kind="pushbullet", project=project)
channel.value = doc["access_token"]
channel.save()
channel.assign_all_checks()
messages.success(request, "The Pushbullet integration has been added!")
else:
messages.warning(request, "Something went wrong")
return redirect("hc-p-channels", project.code)
@login_required
def add_discord(request):
if settings.DISCORD_CLIENT_ID is None:

2
templates/front/channels.html
View File

@ -297,7 +297,7 @@
<h2>Pushbullet</h2>
<p>Pushbullet connects your devices, making them feel like one.</p>
<a href="{% url 'hc-add-pushbullet' %}" class="btn btn-primary">Add Integration</a>
<a href="{% url 'hc-add-pushbullet' project.code %}" class="btn btn-primary">Add Integration</a>
</li>
{% endif %}