feat: Support Bearer Token Authentication

.pre-commit-config.yaml

@@ -17,6 +17,7 @@ repos:
         args:
         - --disable=W,R,C
         - --rcfile=.pylintrc
+        - --jobs=1
         name: Pylint
         language: system
         types: [python]

CHANGELOG.rst

@@ -1,6 +1,11 @@
 Changelog
 =========
 
+Added
+~~~~~
+
+- Support bearer token authentication
+
 Updated
 ~~~~~~~
 

test/config_test.py

@@ -3,6 +3,7 @@
 import tempfile
 import re
 
+import pytest
 import requests
 from email.utils import formatdate
 
@@ -170,3 +171,79 @@ def test_signing(self):
             set(re.search(r'headers="(.+?)"', post_sig).group(1).split(" ")),
             {"(request-target)", "date", "host", "content-length", "digest"},
         )
+
+    def test_bearer_token(self):
+        """Verify that the authorization header is set when a bearer token is provided"""
+
+        bearer_token = (
+            "Bearer eyJraWQiOiJWcmVsOE9zZ0JXaUpHeEpMeFJ4bE1UaVwvbjgyc1hwWktUaTd2UExUNFQ0T"
+            "T0iLCJhbGciOiJSUzI1NiJ9.eyJzdWIiOiJoMTBlM2hwajliNjc4bXMwOG8zbGlibHQ2IiwidG9r"
+            "ZW5fdXNlIjoiYWNjZXNzIiwic2NvcGUiOiJ3ZWJcL2dldCB3ZWJcL3Bvc3QiLCJhdXRoX3RpbWUi"
+            "OjE1OTM3MjM1NDgsImlzcyI6Imh0dHBzOlwvXC9jb2duaXRvLWlkcC51cy1lYXN0LTEuYW1hem9u"
+            "YXdzLmNvbVwvdXMtZWFzdC0xX1d6aEZzTGlPRyIsImV4cCI6MTU5MzcyNzE0OCwiaWF0IjoxNTkz"
+            "NzIzNTQ4LCJ2ZXJzaW9uIjoyLCJqdGkiOiI4Njk5ZDEwYy05Mjg4LTQ0YmEtODIxNi01OTJjZGU3"
+            "MDBhY2MiLCJjbGllbnRfaWQiOiJoMTBlM2hwajliNjc4bXMwOG8zbGlibHQ2In0.YA_yiD-x6UuB"
+            "MShprUbUKuB_DO6ogCtd5srfgpJA6Ve_qsf8n19nVMmFsZBy3GxzN92P1ZXiFY99FfNPohhQtaRR"
+            "hpeUkir08hgJN2bEHCJ5Ym8r9mr9mlwSG6FoiedgLaUVGwJujD9c2rcA83NEo8ayTyfCynF2AZ2p"
+            "MxLHvqOYtvscGMiMzIwlZfJV301iKUVgPODJM5lpJ4iKCpOy2ByCl2_KL1uxIxgMkglpB-i7kgJc"
+            "-WmYoJFoN88D89ugnEoAxNfK14N4_RyEkrLNGape9kew79nUeR6fWbVFLiGDDu25_9z-7VB-GGGk"
+            "7L_Hb7YgVJ5W2FwESnkDvV1T4Q"
+        )
+
+        connection = transcriptic.Connection(
+            email="[email protected]",
+            bearer_token=bearer_token,
+            organization_id="transcriptic",
+            api_root="http://foo:5555",
+            user_id="ufoo2",
+        )
+
+        get_request = requests.Request("GET", "http://foo:5555/get")
+        prepared_get = connection.session.prepare_request(get_request)
+
+        authorization_header_value = prepared_get.headers["authorization"]
+        self.assertEqual(bearer_token, authorization_header_value)
+
+    def test_malformed_bearer_token(self):
+        """Verify that an exception is thrown when a malformed JWT bearer token is provided"""
+
+        bearer_token = "Bearer myBigBadBearerToken"
+
+        with pytest.raises(ValueError, match="Malformed JWT Bearer Token"):
+            transcriptic.Connection(
+                email="[email protected]",
+                bearer_token=bearer_token,
+                organization_id="transcriptic",
+                api_root="http://foo:5555",
+                user_id="ufoo2",
+            )
+
+    def test_user_token_supersedes_bearer_token(self):
+        """Verify that the user token and bearer token are mutually exclusive and that
+        user token supersedes bearer token"""
+
+        user_token = "userTokenFoo"
+        with tempfile.NamedTemporaryFile() as config_file:
+            with open(config_file.name, "w") as f:
+                json.dump(
+                    {
+                        "email": "[email protected]",
+                        "token": user_token,
+                        "bearer_token": "bearerTokenBar",
+                        "organization_id": "transcriptic",
+                        "api_root": "http://foo:5555",
+                        "analytics": True,
+                        "user_id": "ufoo2",
+                        "feature_groups": [
+                            "can_submit_autoprotocol",
+                            "can_upload_packages",
+                        ],
+                    },
+                    f,
+                )
+            connection = transcriptic.config.Connection.from_file(config_file.name)
+
+        get_request = requests.Request("GET", "http://foo:5555/get")
+        prepared_get = connection.session.prepare_request(get_request)
+        self.assertFalse("authorization" in prepared_get.headers)
+        self.assertTrue("X-User-Email" in prepared_get.headers)

transcriptic/config.py

@@ -11,7 +11,8 @@
 import zipfile
 
 from . import routes
-from .signing import StrateosSign
+from .signing import StrateosSign, BearerAuth
+from .util import is_valid_jwt_token
 from .version import __version__
 
 try:
@@ -111,6 +112,7 @@ def __init__(
         feature_groups=[],
         rsa_key=None,
         session=None,
+        bearer_token=None,
     ):
         # Initialize environment args used for computing routes
         self.env_args = dict()
@@ -122,6 +124,8 @@ def __init__(
             session = initialize_default_session()
         self.session = session
 
+        self._bearer_token = None
+
         # Initialize RSA props
         self._rsa_key = None
         self._rsa_key_path = None
@@ -149,7 +153,15 @@ def __init__(
                 )
             self.session.headers["Cookie"] = None
             self.email = email
-            self.token = token
+            if token is not None:
+                self.token = token
+                if bearer_token is not None:
+                    warnings.warn(
+                        "User token and bearer token authentication"
+                        "is mutually exclusive. Ignoring bearer token"
+                    )
+            elif bearer_token is not None:
+                self.bearer_token = bearer_token
             self.update_session_auth()
 
         # Initialize feature groups
@@ -240,6 +252,17 @@ def email(self, value):
         self.update_headers(**{"X-User-Email": value})
         self.update_session_auth()
 
+    @property
+    def bearer_token(self):
+        return self._bearer_token
+
+    @bearer_token.setter
+    def bearer_token(self, value):
+        if is_valid_jwt_token(value):
+            self._bearer_token = value
+        else:
+            raise ValueError("Malformed JWT Bearer Token")
+
     @property
     def token(self):
         try:
@@ -331,6 +354,8 @@ def update_session_auth(self, use_signature=True):
             and "X-User-Email" in self.session.headers
         ):
             self.session.auth = StrateosSign(self.email, self._rsa_secret)
+        elif self.bearer_token:
+            self.session.auth = BearerAuth(self.bearer_token)
         else:
             self.session.auth = None
 
@@ -1081,7 +1106,7 @@ def get_route(self, method, **kwargs):
                 input_args.append(arg_dict[arg])
             else:
                 raise Exception(
-                    f"For route: {method}, argument {arg} needs " f"to be provided."
+                    f"For route: {method}, argument {arg} needs to be provided."
                 )
         return route_method(  # pylint: disable=no-value-for-parameter
             *tuple(input_args)

transcriptic/signing.py

@@ -46,3 +46,12 @@ def __call__(self, request):
             return self.body_auth(request)
 
         return self.auth(request)
+
+
+class BearerAuth(AuthBase):
+    def __init__(self, token):
+        self.token = token
+
+    def __call__(self, r):
+        r.headers["authorization"] = self.token
+        return r

transcriptic/util.py

@@ -92,3 +92,8 @@ def makedirs(name, mode=None, exist_ok=False):
 
     mode = mode if mode is not None else 0o777
     makedirs(name, mode, exist_ok)
+
+
+def is_valid_jwt_token(token: str):
+    regex = r"Bearer ([a-zA-Z0-9_=]+)\.([a-zA-Z0-9_=]+)\.([a-zA-Z0-9_\-\+\/=]*)"
+    return re.fullmatch(regex, token) is not None