Refactor req_ds_cnf to delegation_scope_key

Author: rayluo

msal/application.py

@@ -2359,7 +2359,8 @@ def _acquire_token_for_client(
         claims_challenge=None,
         *,
         delegation_constraints: Optional[list] = None,
-        req_ds_cnf: Optional[dict] = None,
+        delegation_scope_key=None,  # A Cyprtography's RSAPrivateKey-like object
+            # TODO: Support ECC key? https://github.com/pyca/cryptography/issues/4093
         **kwargs
     ):
         if self.authority.tenant.lower() in ["common", "organizations"]:
@@ -2371,17 +2372,20 @@ def _acquire_token_for_client(
         telemetry_context = self._build_telemetry_context(
             self.ACQUIRE_TOKEN_FOR_CLIENT_ID, refresh_reason=refresh_reason)
         client = self._regional_client or self.client
+        if delegation_constraints:
+            from .crypto import _generate_rsa_key, _convert_rsa_keys
+            _, jwk = _convert_rsa_keys(delegation_scope_key or _generate_rsa_key())
         response = client.obtain_token_for_client(
             scope=scopes,  # This grant flow requires no scope decoration
             headers=telemetry_context.generate_headers(),
             data=dict(
                 kwargs.pop("data", {}),
-                req_ds_cnf=_build_req_cnf(req_ds_cnf) if req_ds_cnf else None,
+                req_ds_cnf=_build_req_cnf(jwk) if delegation_constraints else None,
                 claims=_merge_claims_challenge_and_capabilities(
                     self._client_capabilities, claims_challenge)),
             **kwargs)
         if (
-            req_ds_cnf
+            delegation_constraints
             and not response.get("error") and not response.get("xms_ds_nonce")
         ):
             raise ValueError("Your app shall opt in to xms_ds_cnf claim first")

msal/crypto.py

@@ -0,0 +1,23 @@
+from base64 import urlsafe_b64encode
+
+from cryptography.hazmat.primitives.asymmetric import rsa
+
+
+def _urlsafe_b64encode(n:int, bit_size:int) -> str:
+    return urlsafe_b64encode(n.to_bytes(length=int(bit_size/8))).decode("utf-8")
+
+
+def _to_jwk(public_key: rsa.RSAPublicKey) -> dict:
+    numbers = public_key.public_numbers()
+    return {
+        "kty": "RSA",
+        "n": _urlsafe_b64encode(numbers.n, public_key.key_size),
+        "e": _urlsafe_b64encode(numbers.e, 24),  # TODO: TBD
+        }
+
+def _convert_rsa_keys(private_key: rsa.RSAPrivateKey):
+    return "pairs.private_bytes()", _to_jwk(private_key.public_key())
+
+def _generate_rsa_key() -> rsa.RSAPrivateKey:
+    return rsa.generate_private_key(public_exponent=65537, key_size=2048)
+

tests/test_crypto.py

@@ -0,0 +1,12 @@
+from unittest import TestCase
+
+from msal.crypto import _generate_rsa_key, _convert_rsa_keys
+
+
+class CryptoTestCase(TestCase):
+    def test_key_generation(self):
+        key = _generate_rsa_key()
+        _, jwk = _convert_rsa_keys(key)
+        self.assertEqual(jwk.get("kty"), "RSA")
+        self.assertIsNotNone(jwk.get("n") and jwk.get("e"))
+

tests/test_e2e.py

@@ -819,7 +819,7 @@ def test_user_account(self):
 
 
 class CdtTestCase(LabBasedTestCase):
-    _JWK1 = {"kty":"RSA", "n":"2tNr73xwcj6lH7bqRZrFzgSLj7OeLfbn8216uOMDHuaZ6TEUBDN8Uz0ve8jAlKsP9CQFCSVoSNovdE-fs7c15MxEGHjDcNKLWonznximj8pDGZQjVdfK-7mG6P6z-lgVcLuYu5JcWU_PeEqIKg5llOaz-qeQ4LEDS4T1D2qWRGpAra4rJX1-kmrWmX_XIamq30C9EIO0gGuT4rc2hJBWQ-4-FnE1NXmy125wfT3NdotAJGq5lMIfhjfglDbJCwhc8Oe17ORjO3FsB5CLuBRpYmP7Nzn66lRY3Fe11Xz8AEBl3anKFSJcTvlMnFtu3EpD-eiaHfTgRBU7CztGQqVbiQ", "e":"AQAB"}
+    #_JWK1 = {"kty":"RSA", "n":"2tNr73xwcj6lH7bqRZrFzgSLj7OeLfbn8216uOMDHuaZ6TEUBDN8Uz0ve8jAlKsP9CQFCSVoSNovdE-fs7c15MxEGHjDcNKLWonznximj8pDGZQjVdfK-7mG6P6z-lgVcLuYu5JcWU_PeEqIKg5llOaz-qeQ4LEDS4T1D2qWRGpAra4rJX1-kmrWmX_XIamq30C9EIO0gGuT4rc2hJBWQ-4-FnE1NXmy125wfT3NdotAJGq5lMIfhjfglDbJCwhc8Oe17ORjO3FsB5CLuBRpYmP7Nzn66lRY3Fe11Xz8AEBl3anKFSJcTvlMnFtu3EpD-eiaHfTgRBU7CztGQqVbiQ", "e":"AQAB"}
     def test_service_principal(self):
         """
         app = get_lab_app(
@@ -835,14 +835,14 @@ def test_service_principal(self):
         )
         from http.client import HTTPConnection
         HTTPConnection.debuglevel = 1
+        delegation_constraints = [
+            {"typ": "usr", "a": "C", "target": ["constraint1", "constraint4"]},
+            {"typ": "app", "a": "R", "target": ["constraint2", "constraint5"]},
+            {"typ": "subscription", "a": "U", "target": ["constraint3"]},
+        ]
         result = app.acquire_token_for_client(
             [f"{app.client_id}/.default"],
-            delegation_constraints=[
-                {"typ": "usr", "a": "C", "target": ["constraint1", "constraint4"]},
-                {"typ": "app", "a": "R", "target": ["constraint2", "constraint5"]},
-                {"typ": "subscription", "a": "U", "target": ["constraint3"]},
-            ],
-            req_ds_cnf=self._JWK1,
+            delegation_constraints=delegation_constraints,
         )
         self.assertIsNotNone(result.get("access_token"), "Encountered {}: {}".format(
             result.get("error"), result.get("error_description")))