Add custom header X-Perimeter-Token to bypass perimeter (#21)

Adds a custom HTTP header to enable requests to bypass the perimeter.

.travis.yml

@@ -1,3 +1,5 @@
+dist:
+    xenial  # required for SQLite3 3.8.3
 language:
     python
 python:

Pipfile

@@ -0,0 +1,19 @@
+[[source]]
+name = "pypi"
+url = "https://pypi.org/simple"
+verify_ssl = true
+
+[dev-packages]
+coverage = "*"
+black = "*"
+pylint = "*"
+
+[packages]
+# django-perimeter = {editable = true,path = "."}
+django = "*"
+
+[requires]
+python_version = "3"
+
+[pipenv]
+allow_prereleases = true

perimeter/forms.py

@@ -1,8 +1,8 @@
 from django import forms
 from django.core.exceptions import ValidationError
 
-from .middleware import set_request_token
 from .models import AccessToken
+from .settings import PERIMETER_SESSION_KEY
 
 
 class TokenGatewayForm(forms.Form):
@@ -27,8 +27,9 @@ def clean_token(self):
 
     def save(self, request):
         """Create a new AccessTokenUse object from the form."""
-        assert getattr(self, "token", None) is not None, "Form token attr is not set"
-        set_request_token(request, self.token.token)
+        if getattr(self, "token", None) is None:
+            raise ValueError("Form token attr is not set")
+        request.session[PERIMETER_SESSION_KEY] = self.token.token
         return self.token.record(
             user_email=self.cleaned_data.get("email"),
             user_name=self.cleaned_data.get("name"),

perimeter/middleware.py

@@ -4,43 +4,43 @@
 """
 from urllib.parse import urlencode
 
-from django.core.exceptions import MiddlewareNotUsed
+from django.core.exceptions import MiddlewareNotUsed, ImproperlyConfigured
 from django.http import HttpResponseRedirect
 from django.urls import reverse
 from django.utils.deprecation import MiddlewareMixin
 
 from .models import AccessToken
 from .settings import (
+    HTTP_X_PERIMETER_TOKEN,
     PERIMETER_SESSION_KEY,
     PERIMETER_ENABLED,
     PERIMETER_BYPASS_FUNCTION as bypass_perimeter,
 )
 
 
+def check_middleware(request):
+    """Check that Session middleware is installed."""
+    if not hasattr(request, "session"):
+        raise ImproperlyConfigured(
+            "Missing session attribute - please check MIDDLEWARE_CLASSES for "
+            "'django.contrib.sessions.middleware.SessionMiddleware'."
+        )
+
+
 def get_request_token(request):
-    """Returns AccessToken if found else EmptyToken."""
-    assert hasattr(request, "session"), (
-        "Missing session attribute - please check MIDDLEWARE_CLASSES for "
-        "'django.contrib.sessions.middleware.SessionMiddleware'."
+    """Extract token string from HTTP header or querystring."""
+    check_middleware(request)
+    return request.META.get(HTTP_X_PERIMETER_TOKEN, None) or request.session.get(
+        PERIMETER_SESSION_KEY, None
     )
-    token_value = request.session.get(PERIMETER_SESSION_KEY, None)
-    # NB this method implements caching, so is more performant
-    # than the straight get() alternative
-    return AccessToken.objects.get_access_token(token_value)
-
 
-def set_request_token(request, token_value):
-    """Sets the request.session token value.
 
-    Args:
-        token - string, the token value (not the token object, as that is
-            not serializable)
-    """
-    assert hasattr(request, "session"), (
-        "Missing session attribute - please check MIDDLEWARE_CLASSES for "
-        "'django.contrib.sessions.middleware.SessionMiddleware'."
-    )
-    request.session[PERIMETER_SESSION_KEY] = token_value
+def get_access_token(request):
+    """Returns AccessToken if found else EmptyToken."""
+    token = get_request_token(request)
+    # NB this method implements caching, so is more performant
+    # than the straight get() alternative
+    return AccessToken.objects.get_access_token(token)
 
 
 class PerimeterAccessMiddleware(MiddlewareMixin):
@@ -67,7 +67,7 @@ def process_request(self, request):
         if bypass_perimeter(request):
             return None
 
-        if get_request_token(request).is_valid:
+        if get_access_token(request).is_valid:
             return None
 
         # redirect to the gateway for validation,

perimeter/settings.py

@@ -27,6 +27,9 @@ def get_setting(setting_name, default_value, cast_func=lambda x: x):
     )
 
 
+# Name of HTTP header used to automatically bypass perimeter
+HTTP_X_PERIMETER_TOKEN = "HTTP_X_PERIMETER_TOKEN"
+
 # if False, the middleware will be disabled
 PERIMETER_ENABLED = get_setting("PERIMETER_ENABLED", False, cast_func=CAST_AS_BOOL)
 # request.session key used to store user's token

perimeter/tests/test_middleware.py

@@ -1,13 +1,16 @@
 from urllib.parse import urlparse
 
 from django.contrib.auth.models import User, AnonymousUser
+from django.core.exceptions import ImproperlyConfigured
 from django.test import TestCase, RequestFactory, override_settings
 from django.urls import reverse, resolve
 
 from ..middleware import (
     PerimeterAccessMiddleware,
     bypass_perimeter,
+    get_access_token,
     get_request_token,
+    HTTP_X_PERIMETER_TOKEN,
     PERIMETER_SESSION_KEY,
 )
 from ..models import AccessToken, EmptyToken
@@ -41,19 +44,34 @@ def test_bypass_perimeter_default(self):
         request = self.factory.get(reverse("perimeter:gateway"))
         self.assertTrue(bypass_perimeter(request))
 
-    def test_get_request_token(self):
+    def test_get_request_token_session(self):
         at = AccessToken.objects.create_access_token()
         self.request.session[PERIMETER_SESSION_KEY] = at.token
-        self.assertEqual(get_request_token(self.request), at)
+        self.assertEqual(get_request_token(self.request), at.token)
+
+    def test_get_request_token_http_header(self):
+        at = AccessToken.objects.create_access_token()
+        request = self.factory.get("/", HTTP_X_PERIMETER_TOKEN=at.token)
+        request.session = {}
+        self.assertEqual(get_request_token(request), at.token)
+
+    def test_get_access_token(self):
+        at = AccessToken.objects.create_access_token()
+        self.request.session[PERIMETER_SESSION_KEY] = at.token
+        self.assertEqual(get_access_token(self.request), at)
 
     def test_get_request_token_empty(self):
         token = get_request_token(self.request)
-        self.assertTrue(type(token) == EmptyToken)
+        self.assertIsNone(token)
+
+    def test_get_access_token_empty(self):
+        token = get_access_token(self.request)
+        self.assertIsInstance(token, EmptyToken)
 
     def test_missing_session(self):
         """Missing request.session should raise AssertionError."""
         del self.request.session
-        self.assertRaises(AssertionError, get_request_token, self.request)
+        self.assertRaises(ImproperlyConfigured, get_request_token, self.request)
 
     def test_missing_token(self):
         """AnonymousUser without a token should be denied."""
@@ -72,6 +90,14 @@ def test_valid_token(self):
         self.request.session["token"] = "foobar"
         self._assertRedirectsToGateway(self.request)
 
+    def test_perimeter_token_header(self):
+        """Test that the X-Perimeter-Token header works."""
+        AccessToken(token="foobar").save()
+        self.request.user = AnonymousUser()
+        self._assertRedirectsToGateway(self.request)
+        self.request.META["HTTP_X_PERIMETER_TOKEN"] = "foobar"
+        self.middleware.process_request(self.request)
+
     def test_next_query_string_set(self):
         """Check `next` query string param is properly encoded"""
 

runtests.py

@@ -0,0 +1,14 @@
+import os
+import sys
+
+import django
+from django.conf import settings
+from django.test.utils import get_runner
+
+if __name__ == "__main__":
+    os.environ["DJANGO_SETTINGS_MODULE"] = "test_app.settings"
+    django.setup()
+    TestRunner = get_runner(settings)
+    test_runner = TestRunner()
+    failures = test_runner.run_tests(["perimeter.tests"])
+    sys.exit(bool(failures))

setup.py

@@ -8,7 +8,7 @@
 
 setup(
     name="django-perimeter",
-    version="0.11",
+    version="0.12",
     packages=find_packages(),
     include_package_data=True,
     license="MIT",
@@ -25,11 +25,14 @@
         "Framework :: Django",
         "Framework :: Django :: 1.11",
         "Framework :: Django :: 2.0",
+        "Framework :: Django :: 2.1",
+        "Framework :: Django :: 2.2",
         "Intended Audience :: Developers",
         "License :: OSI Approved :: MIT License",
         "Operating System :: OS Independent",
         "Programming Language :: Python",
         "Programming Language :: Python :: 3.6",
+        "Programming Language :: Python :: 3.7",
         "Topic :: Internet :: WWW/HTTP",
         "Topic :: Internet :: WWW/HTTP :: Dynamic Content",
     ],

tox.ini

@@ -1,14 +1,15 @@
 [tox]
-envlist = py{36}-django{111,20}
+envlist = py{36,37}-django{111,20,21,22}
 
 [testenv]
 deps =
     coverage
     django111: Django==1.11
     django20: Django==2.0
+    django21: Django==2.1
+    django22: Django==2.2
 
 commands=
-    python --version
     coverage erase
-    coverage run --include=perimeter/* manage.py test perimeter
+    coverage run --include=perimeter/* runtests.py
     coverage report -m