forked from GithubBackups/healthchecks
Stricter UUID validation.
This commit is contained in:
parent
dc76e4bdde
commit
47d93c2522
@ -1,4 +1,5 @@
|
|||||||
import json
|
import json
|
||||||
|
import re
|
||||||
import uuid
|
import uuid
|
||||||
from functools import wraps
|
from functools import wraps
|
||||||
|
|
||||||
@ -7,13 +8,13 @@ from django.http import (HttpResponseBadRequest, HttpResponseForbidden,
|
|||||||
JsonResponse)
|
JsonResponse)
|
||||||
from hc.lib.jsonschema import ValidationError, validate
|
from hc.lib.jsonschema import ValidationError, validate
|
||||||
|
|
||||||
|
RE_UUID = re.compile("^[a-fA-F0-9]{8}-[a-fA-F0-9]{4}-4[a-fA-F0-9]{3}-[8|9|aA|bB][a-fA-F0-9]{3}-[a-fA-F0-9]{12}$")
|
||||||
|
|
||||||
|
|
||||||
def uuid_or_400(f):
|
def uuid_or_400(f):
|
||||||
@wraps(f)
|
@wraps(f)
|
||||||
def wrapper(request, *args, **kwds):
|
def wrapper(request, *args, **kwds):
|
||||||
try:
|
if not RE_UUID.match(args[0]):
|
||||||
uuid.UUID(args[0])
|
|
||||||
except ValueError:
|
|
||||||
return HttpResponseBadRequest()
|
return HttpResponseBadRequest()
|
||||||
|
|
||||||
return f(request, *args, **kwds)
|
return f(request, *args, **kwds)
|
||||||
|
@ -11,30 +11,30 @@ class PingTestCase(TestCase):
|
|||||||
|
|
||||||
def test_it_works(self):
|
def test_it_works(self):
|
||||||
r = self.client.get("/ping/%s/" % self.check.code)
|
r = self.client.get("/ping/%s/" % self.check.code)
|
||||||
assert r.status_code == 200
|
self.assertEqual(r.status_code, 200)
|
||||||
|
|
||||||
self.check.refresh_from_db()
|
self.check.refresh_from_db()
|
||||||
self.assertEqual(self.check.status, "up")
|
self.assertEqual(self.check.status, "up")
|
||||||
self.assertEqual(self.check.alert_after, self.check.get_alert_after())
|
self.assertEqual(self.check.alert_after, self.check.get_alert_after())
|
||||||
|
|
||||||
ping = Ping.objects.latest("id")
|
ping = Ping.objects.latest("id")
|
||||||
assert ping.scheme == "http"
|
self.assertEqual(ping.scheme, "http")
|
||||||
|
|
||||||
def test_it_changes_status_of_paused_check(self):
|
def test_it_changes_status_of_paused_check(self):
|
||||||
self.check.status = "paused"
|
self.check.status = "paused"
|
||||||
self.check.save()
|
self.check.save()
|
||||||
|
|
||||||
r = self.client.get("/ping/%s/" % self.check.code)
|
r = self.client.get("/ping/%s/" % self.check.code)
|
||||||
assert r.status_code == 200
|
self.assertEqual(r.status_code, 200)
|
||||||
|
|
||||||
self.check.refresh_from_db()
|
self.check.refresh_from_db()
|
||||||
assert self.check.status == "up"
|
self.assertEqual(self.check.status, "up")
|
||||||
|
|
||||||
def test_post_works(self):
|
def test_post_works(self):
|
||||||
csrf_client = Client(enforce_csrf_checks=True)
|
csrf_client = Client(enforce_csrf_checks=True)
|
||||||
r = csrf_client.post("/ping/%s/" % self.check.code, "hello world",
|
r = csrf_client.post("/ping/%s/" % self.check.code, "hello world",
|
||||||
content_type="text/plain")
|
content_type="text/plain")
|
||||||
assert r.status_code == 200
|
self.assertEqual(r.status_code, 200)
|
||||||
|
|
||||||
self.check.refresh_from_db()
|
self.check.refresh_from_db()
|
||||||
self.assertEqual(self.check.last_ping_body, "hello world")
|
self.assertEqual(self.check.last_ping_body, "hello world")
|
||||||
@ -45,16 +45,21 @@ class PingTestCase(TestCase):
|
|||||||
def test_head_works(self):
|
def test_head_works(self):
|
||||||
csrf_client = Client(enforce_csrf_checks=True)
|
csrf_client = Client(enforce_csrf_checks=True)
|
||||||
r = csrf_client.head("/ping/%s/" % self.check.code)
|
r = csrf_client.head("/ping/%s/" % self.check.code)
|
||||||
assert r.status_code == 200
|
self.assertEqual(r.status_code, 200)
|
||||||
assert Ping.objects.count() == 1
|
self.assertEqual(Ping.objects.count(), 1)
|
||||||
|
|
||||||
def test_it_handles_bad_uuid(self):
|
def test_it_handles_bad_uuid(self):
|
||||||
r = self.client.get("/ping/not-uuid/")
|
r = self.client.get("/ping/not-uuid/")
|
||||||
assert r.status_code == 400
|
self.assertEqual(r.status_code, 400)
|
||||||
|
|
||||||
|
def test_it_rejects_alternative_uuid_formats(self):
|
||||||
|
# This uuid is missing separators. uuid.UUID() would accept it.
|
||||||
|
r = self.client.get("/ping/07c2f54898504b27af5d6c9dc157ec02/")
|
||||||
|
self.assertEqual(r.status_code, 400)
|
||||||
|
|
||||||
def test_it_handles_missing_check(self):
|
def test_it_handles_missing_check(self):
|
||||||
r = self.client.get("/ping/07c2f548-9850-4b27-af5d-6c9dc157ec02/")
|
r = self.client.get("/ping/07c2f548-9850-4b27-af5d-6c9dc157ec02/")
|
||||||
assert r.status_code == 404
|
self.assertEqual(r.status_code, 404)
|
||||||
|
|
||||||
def test_it_handles_120_char_ua(self):
|
def test_it_handles_120_char_ua(self):
|
||||||
ua = ("Mozilla/5.0 (Macintosh; Intel Mac OS X 10_10_4) "
|
ua = ("Mozilla/5.0 (Macintosh; Intel Mac OS X 10_10_4) "
|
||||||
@ -62,19 +67,19 @@ class PingTestCase(TestCase):
|
|||||||
"Chrome/44.0.2403.89 Safari/537.36")
|
"Chrome/44.0.2403.89 Safari/537.36")
|
||||||
|
|
||||||
r = self.client.get("/ping/%s/" % self.check.code, HTTP_USER_AGENT=ua)
|
r = self.client.get("/ping/%s/" % self.check.code, HTTP_USER_AGENT=ua)
|
||||||
assert r.status_code == 200
|
self.assertEqual(r.status_code, 200)
|
||||||
|
|
||||||
ping = Ping.objects.latest("id")
|
ping = Ping.objects.latest("id")
|
||||||
assert ping.ua == ua
|
self.assertEqual(ping.ua, ua)
|
||||||
|
|
||||||
def test_it_truncates_long_ua(self):
|
def test_it_truncates_long_ua(self):
|
||||||
ua = "01234567890" * 30
|
ua = "01234567890" * 30
|
||||||
|
|
||||||
r = self.client.get("/ping/%s/" % self.check.code, HTTP_USER_AGENT=ua)
|
r = self.client.get("/ping/%s/" % self.check.code, HTTP_USER_AGENT=ua)
|
||||||
assert r.status_code == 200
|
self.assertEqual(r.status_code, 200)
|
||||||
|
|
||||||
ping = Ping.objects.latest("id")
|
ping = Ping.objects.latest("id")
|
||||||
assert len(ping.ua) == 200
|
self.assertEqual(len(ping.ua), 200)
|
||||||
assert ua.startswith(ping.ua)
|
assert ua.startswith(ping.ua)
|
||||||
|
|
||||||
def test_it_reads_forwarded_ip(self):
|
def test_it_reads_forwarded_ip(self):
|
||||||
@ -82,22 +87,22 @@ class PingTestCase(TestCase):
|
|||||||
r = self.client.get("/ping/%s/" % self.check.code,
|
r = self.client.get("/ping/%s/" % self.check.code,
|
||||||
HTTP_X_FORWARDED_FOR=ip)
|
HTTP_X_FORWARDED_FOR=ip)
|
||||||
ping = Ping.objects.latest("id")
|
ping = Ping.objects.latest("id")
|
||||||
assert r.status_code == 200
|
self.assertEqual(r.status_code, 200)
|
||||||
assert ping.remote_addr == "1.1.1.1"
|
self.assertEqual(ping.remote_addr, "1.1.1.1")
|
||||||
|
|
||||||
ip = "1.1.1.1, 2.2.2.2"
|
ip = "1.1.1.1, 2.2.2.2"
|
||||||
r = self.client.get("/ping/%s/" % self.check.code,
|
r = self.client.get("/ping/%s/" % self.check.code,
|
||||||
HTTP_X_FORWARDED_FOR=ip, REMOTE_ADDR="3.3.3.3")
|
HTTP_X_FORWARDED_FOR=ip, REMOTE_ADDR="3.3.3.3")
|
||||||
ping = Ping.objects.latest("id")
|
ping = Ping.objects.latest("id")
|
||||||
assert r.status_code == 200
|
self.assertEqual(r.status_code, 200)
|
||||||
assert ping.remote_addr == "1.1.1.1"
|
self.assertEqual(ping.remote_addr, "1.1.1.1")
|
||||||
|
|
||||||
def test_it_reads_forwarded_protocol(self):
|
def test_it_reads_forwarded_protocol(self):
|
||||||
r = self.client.get("/ping/%s/" % self.check.code,
|
r = self.client.get("/ping/%s/" % self.check.code,
|
||||||
HTTP_X_FORWARDED_PROTO="https")
|
HTTP_X_FORWARDED_PROTO="https")
|
||||||
ping = Ping.objects.latest("id")
|
ping = Ping.objects.latest("id")
|
||||||
assert r.status_code == 200
|
self.assertEqual(r.status_code, 200)
|
||||||
assert ping.scheme == "https"
|
self.assertEqual(ping.scheme, "https")
|
||||||
|
|
||||||
def test_it_never_caches(self):
|
def test_it_never_caches(self):
|
||||||
r = self.client.get("/ping/%s/" % self.check.code)
|
r = self.client.get("/ping/%s/" % self.check.code)
|
||||||
|
Loading…
x
Reference in New Issue
Block a user