diff --git a/README.rst b/README.rst index 87b444d..3e0953e 100644 --- a/README.rst +++ b/README.rst @@ -43,6 +43,10 @@ Install the app:: HEALTHPOINT_BASICAUTH_USERNAME = 'john' HEALTHPOINT_BASICAUTH_PASSWORD = 'doe' + # If set to True authentication is required for executing the + # health checks. + HEALTHPOINT_AUTH_REQUIRED = True + # urls.py urlpatterns = [ ... diff --git a/healthpoint/tests/test_views.py b/healthpoint/tests/test_views.py index c454af4..ac69ad9 100644 --- a/healthpoint/tests/test_views.py +++ b/healthpoint/tests/test_views.py @@ -1,5 +1,6 @@ from django.core.urlresolvers import reverse -from django.test import TestCase +import base64 +from django.test import TestCase, override_settings from django.utils.http import urlencode @@ -8,26 +9,54 @@ class HealthTestCase(TestCase): def test_health(self): resp = self.client.get(reverse('healthpoint_health')) self.assertEqual(resp.status_code, 503) + self.assertEqual(resp.json(), {}) def test_health_failure(self): resp = self._test_func('healthpoint.tests.health.failure') self.assertEqual(resp.status_code, 503) + self.assertEqual(resp.json(), {}) def test_health_success(self): resp = self._test_func('healthpoint.tests.health.success') self.assertEqual(resp.status_code, 200) + self.assertEqual(resp.json(), {}) def test_unknown_health(self): resp = self._test_func('does.not.exist') self.assertEqual(resp.status_code, 404) + self.assertEqual(resp.json(), {}) def test_unknown_health_and_success(self): resp = self._test_func('does.not.exist', 'healthpoint.tests.health.success') self.assertEqual(resp.status_code, 404) + self.assertEqual(resp.json(), {}) def test_unknown_health_and_failure(self): resp = self._test_func('does.not.exist', 'healthpoint.tests.health.failure') self.assertEqual(resp.status_code, 503) + self.assertEqual(resp.json(), {}) + + @override_settings( + HEALTHPOINT_BASICAUTH_USERNAME='john', + HEALTHPOINT_BASICAUTH_PASSWORD='doe' + ) + def test_health_with_basic_auth(self): + headers = { + 'HTTP_AUTHORIZATION': 'Basic ' + + base64.b64encode(b'john:doe').decode("ascii") + } + resp = self.client.get(reverse('healthpoint_health'), **headers) + self.assertEqual(resp.status_code, 503) + self.assertEqual( + resp.json(), + {'success': {'healthpoint.tests.health.success': 'This check always succeeds'}, + 'error': {'healthpoint.tests.health.failure': 'This check always fails'}} + ) + + @override_settings(HEALTHPOINT_AUTH_REQUIRED=True) + def test_health_with_auth_required(self): + resp = self.client.get(reverse('healthpoint_health')) + self.assertEqual(resp.status_code, 403) def _test_func(self, *func): params = urlencode({'test': func}, doseq=True) diff --git a/healthpoint/views.py b/healthpoint/views.py index 5a7179d..2600770 100644 --- a/healthpoint/views.py +++ b/healthpoint/views.py @@ -5,11 +5,13 @@ from healthpoint.registry import get_health_checks -def _show_health_details(request): - # Only staff members are allowed to see details... +def _is_authenticated_request(request): + # Superusers and staff members are always correctly authenticated. user = getattr(request, 'user', None) if user is not None and (user.is_staff or user.is_superuser): return True + # Requests with basic authentication credentials are only + # authenticated when they match the settings. ba_username = getattr(settings, 'HEALTHPOINT_BASICAUTH_USERNAME', None) ba_password = getattr(settings, 'HEALTHPOINT_BASICAUTH_PASSWORD', None) authorization = request.META.get('HTTP_AUTHORIZATION') @@ -27,23 +29,39 @@ def health(request): tests_left = set(tests) data = {'success': {}, 'error': {}} status = 200 - for health_check in get_health_checks(): - func = '.'.join([ - health_check.__module__, - health_check.__qualname__]) - if tests and func not in tests: - continue - tests_left.discard(func) - success, detail = health_check() - data['success' if success else 'error'][func] = detail - if not success: - status = 503 - if tests_left: - if status == 200: - status = 404 - for test in tests_left: - data['error'][test] = 'Unknown health check' - # Only staff members are allowed to see details... - if not _show_health_details(request): + + is_authenticated_request = _is_authenticated_request(request) + authentication_required = getattr(settings, 'HEALTHPOINT_AUTH_REQUIRED', False) + + if authentication_required and not is_authenticated_request: + # The settings require a succesfully authenticated request + # Abort the health checks if this is not the case... + status = 403 data = {} + else: + # Perform the actual health checks if the authentication is successful + # or not required. + for health_check in get_health_checks(): + func = '.'.join([ + health_check.__module__, + health_check.__qualname__]) + if tests and func not in tests: + continue + tests_left.discard(func) + success, detail = health_check() + data['success' if success else 'error'][func] = detail + if not success: + status = 503 + if tests_left: + if status == 200: + status = 404 + for test in tests_left: + data['error'][test] = 'Unknown health check' + + # Only successfully authenticated requests are allowed to see the + # full details of the results. + if not is_authenticated_request: + data = {} + + # Return the response return JsonResponse(data, status=status)