Add to_rsa_format function to normalize private key

CHANGELOG.md

@@ -1,3 +1,7 @@
+# 1.8.0
+- Add `to_rsa_format` function to normalize private key
+- Update requests_mauth and httpx_mauth to support reading configuration from environment variables
+
 # 1.7.0
 - Add `MAuthHttpx` custom authentication scheme for HTTPX.
 - Remove Support for EOL Python 3.8

README.md

@@ -81,6 +81,13 @@ client = httpx.Client(auth=auth)
 response = client.get("https://api.example.com/endpoint")
 ```
 
+The following variables can be configured in the environment variables:
+
+| Key                                  | Value                              |
+| ------------------------------------ | ---------------------------------- |
+| `APP_UUID` or `MAUTH_APP_UUID`       | APP_UUID for signing requests      |
+| `PRIVATE_KEY` or `MAUTH_PRIVATE_KEY` | MAuth private key for the APP_UUID |
+
 The `mauth_sign_versions` option can be set as an environment variable to specify protocol versions to sign outgoing requests:
 
 | Key                   | Value                                                                                |
@@ -103,11 +110,11 @@ MAuth Client Python supports AWS Lambda functions and Flask applications to auth
 
 The following variables are **required** to be configured in the environment variables:
 
-| Key            | Value                                                         |
-| -------------- | ------------------------------------------------------------- |
-| `APP_UUID`     | APP_UUID for the AWS Lambda function                          |
-| `PRIVATE_KEY`  | Encrypted private key for the APP_UUID                        |
-| `MAUTH_URL`    | MAuth service URL (e.g. https://mauth-innovate.imedidata.com) |
+| Key                                  | Value                                                         |
+| ------------------------------------ | ------------------------------------------------------------- |
+| `APP_UUID` or `MAUTH_APP_UUID`       | APP_UUID for the AWS Lambda function                          |
+| `PRIVATE_KEY` or `MAUTH_PRIVATE_KEY` | Encrypted private key for the APP_UUID                        |
+| `MAUTH_URL`                          | MAuth service URL (e.g. https://mauth-innovate.imedidata.com) |
 
 
 The following variables can optionally be set in the environment variables:

mauth_client/config.py

@@ -1,11 +1,13 @@
 import os
+from .utils import to_rsa_format
 
 
 class Config:
-    APP_UUID = os.environ.get("APP_UUID")
-    MAUTH_URL = os.environ.get("MAUTH_URL")
-    MAUTH_API_VERSION = os.environ.get("MAUTH_API_VERSION", "v1")
-    MAUTH_MODE = os.environ.get("MAUTH_MODE", "local")
-    PRIVATE_KEY = os.environ.get("PRIVATE_KEY")
-    V2_ONLY_AUTHENTICATE = str(os.environ.get("V2_ONLY_AUTHENTICATE")).lower() == "true"
-    SIGN_VERSIONS = os.environ.get("MAUTH_SIGN_VERSIONS", "v1")
+    APP_UUID = os.getenv("APP_UUID", os.getenv("MAUTH_APP_UUID"))
+    MAUTH_URL = os.getenv("MAUTH_URL")
+    MAUTH_API_VERSION = os.getenv("MAUTH_API_VERSION", "v1")
+    MAUTH_MODE = os.getenv("MAUTH_MODE", "local")
+    _private_key_env = os.getenv("PRIVATE_KEY", os.getenv("MAUTH_PRIVATE_KEY", ""))
+    PRIVATE_KEY = to_rsa_format(_private_key_env) if _private_key_env else None
+    V2_ONLY_AUTHENTICATE = str(os.getenv("V2_ONLY_AUTHENTICATE")).lower() == "true"
+    SIGN_VERSIONS = os.getenv("MAUTH_SIGN_VERSIONS", "v1")

mauth_client/httpx_mauth/client.py

@@ -15,8 +15,8 @@ class MAuthHttpx(httpx.Auth):
 
     def __init__(
         self,
-        app_uuid: str,
-        private_key_data: str,
+        app_uuid: str = Config.APP_UUID,
+        private_key_data: str = Config.PRIVATE_KEY,
         sign_versions: str = Config.SIGN_VERSIONS,
     ):
         self.signer = Signer(app_uuid, private_key_data, sign_versions)

mauth_client/requests_mauth/client.py

@@ -9,7 +9,12 @@ class MAuth(requests.auth.AuthBase):
     Custom requests authorizer for MAuth
     """
 
-    def __init__(self, app_uuid, private_key_data, sign_versions=Config.SIGN_VERSIONS):
+    def __init__(
+        self,
+        app_uuid=Config.APP_UUID,
+        private_key_data=Config.PRIVATE_KEY,
+        sign_versions=Config.SIGN_VERSIONS
+    ):
         """
         Create a new MAuth Instance
 

mauth_client/utils.py

@@ -1,7 +1,11 @@
 import base64
 import charset_normalizer
+import re
 from hashlib import sha512
 
+HEADER = '-----BEGIN RSA PRIVATE KEY-----'
+FOOTER = '-----END RSA PRIVATE KEY-----'
+
 
 def make_bytes(val):
     """
@@ -32,3 +36,22 @@ def decode(byte_string: bytes) -> str:
     except UnicodeDecodeError:
         encoding = charset_normalizer.detect(byte_string)["encoding"]
         return byte_string.decode(encoding)
+
+
+def to_rsa_format(key: str) -> str:
+    """Convert a private key to RSA format with proper newlines."""
+
+    if "\n" in key and HEADER in key and FOOTER in key:
+        return key
+
+    body = key.strip()
+    body = body.replace(HEADER, "").replace(FOOTER, "").strip()
+
+    # Replace whitespace with newlines or chunk into 64-char lines
+    if " " in body or "\t" in body:
+        body = re.sub(r'\s+', '\n', body)
+    else:
+        # PEM-encoded keys are typically split into lines of 64 characters as per RFC 7468 (section 2)
+        body = '\n'.join(body[i:i + 64] for i in range(0, len(body), 64))
+
+    return f"{HEADER}\n{body}\n{FOOTER}"

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "mauth-client"
-version = "1.7.0"
+version = "1.8.0"
 description = "MAuth Client for Python"
 repository = "https://github.com/mdsol/mauth-client-python"
 authors = ["Medidata Solutions <support@mdsol.com>"]

tests/httpx_mauth/client_test.py

@@ -1,9 +1,10 @@
 import unittest
-import os
 import httpx
 from mauth_client.httpx_mauth import MAuthHttpx
+from ..common import load_key
 
 APP_UUID = "5ff4257e-9c16-11e0-b048-0026bbfffe5e"
+PRIVATE_KEY = load_key("priv")
 URL = "https://innovate.imedidata.com/api/v2/users/10ac3b0e-9fe2-11df-a531-12313900d531/studies.json"
 
 
@@ -12,28 +13,24 @@ def handler(request):
 
 
 class MAuthHttpxBaseTest(unittest.TestCase):
-    def setUp(self):
-        with open(os.path.join(os.path.dirname(__file__), "..", "keys", "fake_mauth.priv.key"), "r") as key_file:
-            self.example_private_key = key_file.read()
-
     def test_call(self):
-        auth = MAuthHttpx(APP_UUID, self.example_private_key, sign_versions="v1,v2")
+        auth = MAuthHttpx(APP_UUID, PRIVATE_KEY, sign_versions="v1,v2")
         with httpx.Client(transport=httpx.MockTransport(handler), auth=auth) as client:
             response = client.get(URL)
 
         for header in ["mcc-authentication", "mcc-time", "x-mws-authentication", "x-mws-time"]:
             self.assertIn(header, response.request.headers)
 
     def test_call_v1_only(self):
-        auth = MAuthHttpx(APP_UUID, self.example_private_key)
+        auth = MAuthHttpx(APP_UUID, PRIVATE_KEY)
         with httpx.Client(transport=httpx.MockTransport(handler), auth=auth) as client:
             response = client.get(URL)
 
         for header in ["x-mws-authentication", "x-mws-time"]:
             self.assertIn(header, response.request.headers)
 
     def test_call_v2_only(self):
-        auth = MAuthHttpx(APP_UUID, self.example_private_key, sign_versions="v2")
+        auth = MAuthHttpx(APP_UUID, PRIVATE_KEY, sign_versions="v2")
         with httpx.Client(transport=httpx.MockTransport(handler), auth=auth) as client:
             response = client.get(URL)
 

tests/utils_test.py

@@ -0,0 +1,22 @@
+import unittest
+
+from .common import load_key
+from mauth_client.utils import to_rsa_format
+
+PRIVATE_KEY = load_key("priv").strip()
+
+
+class TestToRsaFormat(unittest.TestCase):
+    def test_proper_format(self):
+        key = to_rsa_format(PRIVATE_KEY)
+        self.assertEqual(key, PRIVATE_KEY)
+
+    def test_newlines_replaced_with_spaces(self):
+        key_no_newlines = PRIVATE_KEY.replace("\n", " ")
+        key = to_rsa_format(key_no_newlines)
+        self.assertEqual(key, PRIVATE_KEY)
+
+    def test_newlines_removed(self):
+        key_no_newlines = PRIVATE_KEY.replace("\n", "")
+        key = to_rsa_format(key_no_newlines)
+        self.assertEqual(key, PRIVATE_KEY)