Merge pull request #38 from ShipChain/feature/lambda-internal-request

Lambda Internal Request

Author: mlclay

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "shipchain-common"
-version = "1.0.23"
+version = "1.0.24"
 description = "A PyPI package containing shared code for ShipChain's Python/Django projects."
 
 license = "Apache-2.0"

src/shipchain_common/authentication.py

@@ -66,6 +66,10 @@ class TransmissionRequest(InternalRequest):
     SERVICE_NAME = 'transmission'
 
 
+class LambdaRequest(InternalRequest):
+    SERVICE_NAME = 'lambda'
+
+
 class PermissionedTokenUser(TokenUser):
     """
     This Requires the JWT from Profiles to have been generated with the `permissions` scope

tests/test_auth.py

@@ -14,8 +14,8 @@
 from rest_framework import exceptions
 from rest_framework_simplejwt.tokens import UntypedToken
 
-from src.shipchain_common.authentication import EngineRequest, passive_credentials_auth, PermissionedTokenUser,\
-    TransmissionRequest
+from src.shipchain_common.authentication import EngineRequest, passive_credentials_auth, PermissionedTokenUser, \
+    TransmissionRequest, LambdaRequest
 from src.shipchain_common.test_utils import get_jwt
 from src.shipchain_common.utils import random_id
 
@@ -43,6 +43,11 @@ def transmission_request():
     return TransmissionRequest()
 
 
+@pytest.fixture()
+def lambda_request():
+    return LambdaRequest()
+
+
 def test_passive_jwt_auth(username):
     with pytest.raises(exceptions.AuthenticationFailed):
         passive_credentials_auth('')
@@ -118,6 +123,35 @@ def test_transmission_auth_requires_header(transmission_request):
     assert transmission_request.has_permission(request, {})
 
 
+def test_lambda_auth_requires_header(lambda_request):
+    request = HttpRequest()
+
+    assert not lambda_request.has_permission(request, {})
+
+    request.META['X_NGINX_SOURCE'] = 'alb'
+    assert not lambda_request.has_permission(request, {})
+
+    request.META['X_NGINX_SOURCE'] = 'internal'
+    with pytest.raises(KeyError):
+        lambda_request.has_permission(request, {})
+
+    request.META['X_SSL_CLIENT_VERIFY'] = 'NONE'
+    assert not lambda_request.has_permission(request, {})
+
+    request.META['X_SSL_CLIENT_VERIFY'] = 'SUCCESS'
+    with pytest.raises(KeyError):
+        lambda_request.has_permission(request, {})
+
+    request.META['X_SSL_CLIENT_DN'] = '/CN=lambda.h4ck3d'
+    assert not lambda_request.has_permission(request, {})
+
+    request.META['X_SSL_CLIENT_DN'] = '/CN=profiles.test-internal'
+    assert not lambda_request.has_permission(request, {})
+
+    request.META['X_SSL_CLIENT_DN'] = '/CN=lambda.test-internal'
+    assert lambda_request.has_permission(request, {})
+
+
 def test_token_user_jti_cache_key():
     """By default, the jti is included in get_jwt and is used as cache key"""
     jwt = get_jwt()