Util to check if request is Hawk request (#43)

hawkrest/__init__.py

@@ -17,6 +17,8 @@
 from rest_framework.authentication import BaseAuthentication
 from rest_framework.exceptions import AuthenticationFailed
 
+from hawkrest.util import get_auth_header, is_hawk_request
+
 
 log = logging.getLogger(__name__)
 # Number of seconds until a Hawk message expires.
@@ -67,11 +69,11 @@ def authenticate(self, request):
         # pollution of META.
         request.META['hawk.receiver'] = None
 
-        http_authorization = request.META.get('HTTP_AUTHORIZATION')
+        http_authorization = get_auth_header(request)
         if not http_authorization:
             log.debug('no authorization header in request')
             return None
-        elif not http_authorization.startswith('Hawk '):
+        elif not is_hawk_request(request):
             log.debug('ignoring non-Hawk authorization header: {} '
                       .format(http_authorization))
             return None

hawkrest/middleware.py

@@ -6,23 +6,21 @@
 except ImportError:  # Django version < 1.11
     middleware_cls = object
 
+from hawkrest.util import is_hawk_request
+
 
 log = logging.getLogger(__name__)
 
 
 class HawkResponseMiddleware(middleware_cls):
 
     def process_response(self, request, response):
-        is_hawk_request = False
-        if request.META.get('HTTP_AUTHORIZATION', '').startswith('Hawk'):
-            is_hawk_request = True
-
         hawk_auth_was_processed = 'hawk.receiver' in request.META
         receiver = request.META.get('hawk.receiver', None)
 
         log.debug('receiver? {rec}; hawk auth processed? {auth}'
                   .format(rec=receiver, auth=hawk_auth_was_processed))
-        if is_hawk_request and not hawk_auth_was_processed:
+        if is_hawk_request(request) and not hawk_auth_was_processed:
             # This is a paranoid check to make sure Django
             # isn't misconfigured.
             raise RuntimeError('Django did not handle an incoming '

hawkrest/util.py

@@ -0,0 +1,7 @@
+def get_auth_header(request):
+    return request.META.get('HTTP_AUTHORIZATION', '')
+
+
+def is_hawk_request(request):
+    auth_header = get_auth_header(request)
+    return auth_header.startswith('Hawk ')

tests/test_util.py

@@ -0,0 +1,38 @@
+from django.test import RequestFactory
+
+from tests.base import BaseTest
+from hawkrest.util import get_auth_header, is_hawk_request
+
+
+class TestGetAuthHeader(BaseTest):
+
+    def test_func_gets_http_authorization_header(self):
+        sender = self._sender()
+        factory_obj = RequestFactory(HTTP_AUTHORIZATION=sender.request_header)
+        request = factory_obj.request()
+        auth_header = get_auth_header(request)
+        self.assertEqual(auth_header, sender.request_header)
+
+    def test_func_returns_empty_string_if_no_auth_header(self):
+        factory_obj = RequestFactory()
+        request = factory_obj.request()
+        auth_header = get_auth_header(request)
+        self.assertEqual(auth_header, '')
+
+
+class TestIsHawkRequest(BaseTest):
+
+    def test_func_returns_true_when_auth_header_begins_with_hawk(self):
+        factory_obj = RequestFactory(HTTP_AUTHORIZATION='Hawk ')
+        request = factory_obj.request()
+        self.assertTrue(is_hawk_request(request))
+
+    def test_func_returns_false_if_no_trailing_whitespace(self):
+        factory_obj = RequestFactory(HTTP_AUTHORIZATION='Hawk')
+        request = factory_obj.request()
+        self.assertFalse(is_hawk_request(request))
+
+    def test_func_returns_false_if_auth_header_missing(self):
+        factory_obj = RequestFactory()
+        request = factory_obj.request()
+        self.assertFalse(is_hawk_request(request))