AT POP for Public Client based on broker

msal/__init__.py

@@ -33,4 +33,5 @@
     )
 from .oauth2cli.oidc import Prompt
 from .token_cache import TokenCache, SerializableTokenCache
+from .auth_scheme import PopAuthScheme
 

msal/__main__.py

@@ -22,6 +22,11 @@
 
 _AZURE_CLI = "04b07795-8ddb-461a-bbee-02f9e1bf7b46"
 _VISUAL_STUDIO = "04f0c124-f2bc-4f59-8241-bf6df9866bbd"
+placeholder_auth_scheme = msal.PopAuthScheme(
+    http_method=msal.PopAuthScheme.HTTP_GET,
+    url="https://example.com/endpoint",
+    nonce="placeholder",
+    )
 
 def print_json(blob):
     print(json.dumps(blob, indent=2, sort_keys=True))
@@ -88,6 +93,9 @@ def _acquire_token_silent(app):
             _input_scopes(),
             account=account,
             force_refresh=_input_boolean("Bypass MSAL Python's token cache?"),
+            auth_scheme=placeholder_auth_scheme
+                if app.is_pop_supported() and _input_boolean("Acquire AT POP via Broker?")
+                else None,
             ))
 
 def _acquire_token_interactive(app, scopes=None, data=None):
@@ -117,6 +125,9 @@ def _acquire_token_interactive(app, scopes=None, data=None):
             ],  # Here this test app mimics the setting for some known MSA-PT apps
         port=1234,  # Hard coded for testing. Real app typically uses default value.
         prompt=prompt, login_hint=login_hint, data=data or {},
+        auth_scheme=placeholder_auth_scheme
+            if app.is_pop_supported() and _input_boolean("Acquire AT POP via Broker?")
+            else None,
         )
     if login_hint and "id_token_claims" in result:
         signed_in_user = result.get("id_token_claims", {}).get("preferred_username")

msal/application.py

@@ -182,6 +182,10 @@ class ClientApplication(object):
     _TOKEN_SOURCE_BROKER = "broker"
 
     _enable_broker = False
+    _AUTH_SCHEME_UNSUPPORTED = (
+        "auth_scheme is currently only available from broker. "
+        "You can enable broker by following these instructions. "
+        "https://msal-python.readthedocs.io/en/latest/#publicclientapplication")
 
     def __init__(
             self, client_id,
@@ -557,6 +561,10 @@ def _decide_broker(self, allow_broker, enable_pii_log):
                     "We will fallback to non-broker.")
         logger.debug("Broker enabled? %s", self._enable_broker)
 
+    def is_pop_supported(self):
+        """Returns True if this client supports Proof-of-Possession Access Token."""
+        return self._enable_broker
+
     def _decorate_scope(
             self, scopes,
             reserved_scope=frozenset(['openid', 'profile', 'offline_access'])):
@@ -1185,6 +1193,7 @@ def acquire_token_silent(
             authority=None,  # See get_authorization_request_url()
             force_refresh=False,  # type: Optional[boolean]
             claims_challenge=None,
+            auth_scheme=None,
             **kwargs):
         """Acquire an access token for given account, without user interaction.
 
@@ -1205,7 +1214,7 @@ def acquire_token_silent(
             return None  # A backward-compatible NO-OP to drop the account=None usage
         result = _clean_up(self._acquire_token_silent_with_error(
             scopes, account, authority=authority, force_refresh=force_refresh,
-            claims_challenge=claims_challenge, **kwargs))
+            claims_challenge=claims_challenge, auth_scheme=auth_scheme, **kwargs))
         return result if result and "error" not in result else None
 
     def acquire_token_silent_with_error(
@@ -1215,6 +1224,7 @@ def acquire_token_silent_with_error(
             authority=None,  # See get_authorization_request_url()
             force_refresh=False,  # type: Optional[boolean]
             claims_challenge=None,
+            auth_scheme=None,
             **kwargs):
         """Acquire an access token for given account, without user interaction.
 
@@ -1241,6 +1251,12 @@ def acquire_token_silent_with_error(
             in the form of a claims_challenge directive in the www-authenticate header to be
             returned from the UserInfo Endpoint and/or in the ID Token and/or Access Token.
             It is a string of a JSON object which contains lists of claims being requested from these locations.
+        :param object auth_scheme:
+            You can provide an ``msal.auth_scheme.PopAuthScheme`` object
+            so that MSAL will get a Proof-of-Possession (POP) token for you.
+
+            New in version 1.26.0.
+
         :return:
             - A dict containing no "error" key,
               and typically contains an "access_token" key,
@@ -1252,7 +1268,7 @@ def acquire_token_silent_with_error(
             return None  # A backward-compatible NO-OP to drop the account=None usage
         return _clean_up(self._acquire_token_silent_with_error(
             scopes, account, authority=authority, force_refresh=force_refresh,
-            claims_challenge=claims_challenge, **kwargs))
+            claims_challenge=claims_challenge, auth_scheme=auth_scheme, **kwargs))
 
     def _acquire_token_silent_with_error(
             self,
@@ -1261,6 +1277,7 @@ def _acquire_token_silent_with_error(
             authority=None,  # See get_authorization_request_url()
             force_refresh=False,  # type: Optional[boolean]
             claims_challenge=None,
+            auth_scheme=None,
             **kwargs):
         assert isinstance(scopes, list), "Invalid parameter type"
         self._validate_ssh_cert_input_data(kwargs.get("data", {}))
@@ -1276,6 +1293,7 @@ def _acquire_token_silent_with_error(
             scopes, account, self.authority, force_refresh=force_refresh,
             claims_challenge=claims_challenge,
             correlation_id=correlation_id,
+            auth_scheme=auth_scheme,
             **kwargs)
         if result and "error" not in result:
             return result
@@ -1298,6 +1316,7 @@ def _acquire_token_silent_with_error(
                 scopes, account, the_authority, force_refresh=force_refresh,
                 claims_challenge=claims_challenge,
                 correlation_id=correlation_id,
+                auth_scheme=auth_scheme,
                 **kwargs)
             if result:
                 if "error" not in result:
@@ -1322,12 +1341,13 @@ def _acquire_token_silent_from_cache_and_possibly_refresh_it(
             claims_challenge=None,
             correlation_id=None,
             http_exceptions=None,
+            auth_scheme=None,
             **kwargs):
         # This internal method has two calling patterns:
         # it accepts a non-empty account to find token for a user,
         # and accepts account=None to find a token for the current app.
         access_token_from_cache = None
-        if not (force_refresh or claims_challenge):  # Bypass AT when desired or using claims
+        if not (force_refresh or claims_challenge or auth_scheme):  # Then attempt AT cache
             query={
                     "client_id": self.client_id,
                     "environment": authority.instance,
@@ -1370,6 +1390,8 @@ def _acquire_token_silent_from_cache_and_possibly_refresh_it(
         try:
             data = kwargs.get("data", {})
             if account and account.get("authority_type") == _AUTHORITY_TYPE_CLOUDSHELL:
+                if auth_scheme:
+                    raise ValueError("auth_scheme is not supported in Cloud Shell")
                 return self._acquire_token_by_cloud_shell(scopes, data=data)
 
             if self._enable_broker and account and account.get("account_source") in (
@@ -1385,6 +1407,7 @@ def _acquire_token_silent_from_cache_and_possibly_refresh_it(
                     claims=_merge_claims_challenge_and_capabilities(
                         self._client_capabilities, claims_challenge),
                     correlation_id=correlation_id,
+                    auth_scheme=auth_scheme,
                     **data)
                 if response:  # Broker provides a decisive outcome
                     account_was_established_by_broker = account.get(
@@ -1393,6 +1416,8 @@ def _acquire_token_silent_from_cache_and_possibly_refresh_it(
                     if account_was_established_by_broker or broker_attempt_succeeded_just_now:
                         return self._process_broker_response(response, scopes, data)
 
+            if auth_scheme:
+                raise ValueError(self._AUTH_SCHEME_UNSUPPORTED)
             if account:
                 result = self._acquire_token_silent_by_finding_rt_belongs_to_me_or_my_family(
                     authority, self._decorate_scope(scopes), account,
@@ -1588,7 +1613,11 @@ def acquire_token_by_refresh_token(self, refresh_token, scopes, **kwargs):
         return response
 
     def acquire_token_by_username_password(
-            self, username, password, scopes, claims_challenge=None, **kwargs):
+            self, username, password, scopes, claims_challenge=None,
+            # Note: We shouldn't need to surface enable_msa_passthrough,
+            # because this ROPC won't work with MSA account anyway.
+            auth_scheme=None,
+            **kwargs):
         """Gets a token for a given resource via user credentials.
 
         See this page for constraints of Username Password Flow.
@@ -1604,6 +1633,12 @@ def acquire_token_by_username_password(
             returned from the UserInfo Endpoint and/or in the ID Token and/or Access Token.
             It is a string of a JSON object which contains lists of claims being requested from these locations.
 
+        :param object auth_scheme:
+            You can provide an ``msal.auth_scheme.PopAuthScheme`` object
+            so that MSAL will get a Proof-of-Possession (POP) token for you.
+
+            New in version 1.26.0.
+
         :return: A dict representing the json response from AAD:
 
             - A successful response would contain "access_token" key,
@@ -1623,9 +1658,12 @@ def acquire_token_by_username_password(
                     self.authority._is_known_to_developer
                     or self._instance_discovery is False) else None,
                 claims=claims,
+                auth_scheme=auth_scheme,
                 )
             return self._process_broker_response(response, scopes, kwargs.get("data", {}))
 
+        if auth_scheme:
+            raise ValueError(self._AUTH_SCHEME_UNSUPPORTED)
         scopes = self._decorate_scope(scopes)
         telemetry_context = self._build_telemetry_context(
             self.ACQUIRE_TOKEN_BY_USERNAME_PASSWORD_ID)
@@ -1768,6 +1806,7 @@ def acquire_token_interactive(
             max_age=None,
             parent_window_handle=None,
             on_before_launching_ui=None,
+            auth_scheme=None,
             **kwargs):
         """Acquire token interactively i.e. via a local browser.
 
@@ -1843,6 +1882,12 @@ def acquire_token_interactive(
 
             New in version 1.20.0.
 
+        :param object auth_scheme:
+            You can provide an ``msal.auth_scheme.PopAuthScheme`` object
+            so that MSAL will get a Proof-of-Possession (POP) token for you.
+
+            New in version 1.26.0.
+
         :return:
             - A dict containing no "error" key,
               and typically contains an "access_token" key.
@@ -1887,12 +1932,15 @@ def acquire_token_interactive(
                 claims,
                 data,
                 on_before_launching_ui,
+                auth_scheme,
                 prompt=prompt,
                 login_hint=login_hint,
                 max_age=max_age,
                 )
             return self._process_broker_response(response, scopes, data)
 
+        if auth_scheme:
+            raise ValueError("auth_scheme is currently only available from broker")
         on_before_launching_ui(ui="browser")
         telemetry_context = self._build_telemetry_context(
             self.ACQUIRE_TOKEN_INTERACTIVE)
@@ -1927,6 +1975,7 @@ def _acquire_token_interactive_via_broker(
             claims,  # type: str
             data,  # type: dict
             on_before_launching_ui,  # type: callable
+            auth_scheme,  # type: object
             prompt=None,
             login_hint=None,  # type: Optional[str]
             max_age=None,
@@ -1950,6 +1999,7 @@ def _acquire_token_interactive_via_broker(
                     accounts[0]["local_account_id"],
                     scopes,
                     claims=claims,
+                    auth_scheme=auth_scheme,
                     **data)
                 if response and "error" not in response:
                     return response
@@ -1962,6 +2012,7 @@ def _acquire_token_interactive_via_broker(
                 claims=claims,
                 max_age=max_age,
                 enable_msa_pt=enable_msa_passthrough,
+                auth_scheme=auth_scheme,
                 **data)
             is_wrong_account = bool(
                 # _signin_silently() only gets tokens for default account,
@@ -2002,6 +2053,7 @@ def _acquire_token_interactive_via_broker(
             claims=claims,
             max_age=max_age,
             enable_msa_pt=enable_msa_passthrough,
+            auth_scheme=auth_scheme,
             **data)
 
     def initiate_device_flow(self, scopes=None, **kwargs):

msal/auth_scheme.py

@@ -0,0 +1,34 @@
+try:
+    from urllib.parse import urlparse
+except ImportError:  # Fall back to Python 2
+    from urlparse import urlparse
+
+# We may support more auth schemes in the future
+class PopAuthScheme(object):
+    HTTP_GET = "GET"
+    HTTP_POST = "POST"
+    HTTP_PUT = "PUT"
+    HTTP_DELETE = "DELETE"
+    HTTP_PATCH = "PATCH"
+    _HTTP_METHODS = (HTTP_GET, HTTP_POST, HTTP_PUT, HTTP_DELETE, HTTP_PATCH)
+    # Internal design: https://identitydivision.visualstudio.com/DevEx/_git/AuthLibrariesApiReview?path=/PoPTokensProtocol/PopTokensProtocol.md
+    def __init__(self, http_method=None, url=None, nonce=None):
+        """Create an auth scheme which is needed to obtain a Proof-of-Possession token.
+
+        :param str http_method:
+            Its value is an uppercase http verb, such as "GET" and "POST".
+        :param str url:
+            The url to be signed.
+        :param str nonce:
+            The nonce came from resource's challenge.
+        """
+        if not (http_method and url and nonce):
+            # In the future, we may also support accepting an http_response as input
+            raise ValueError("All http_method, url and nonce are required parameters")
+        if http_method not in self._HTTP_METHODS:
+            raise ValueError("http_method must be uppercase, according to "
+                "https://datatracker.ietf.org/doc/html/draft-ietf-oauth-signed-http-request-03#section-3")
+        self._http_method = http_method
+        self._url = urlparse(url)
+        self._nonce = nonce
+

msal/broker.py

@@ -99,13 +99,17 @@ def _convert_result(result, client_id, expected_token_type=None):  # Mimic an on
     assert account, "Account is expected to be always available"
     # Note: There are more account attribute getters available in pymsalruntime 0.13+
     return_value = {k: v for k, v in {
-        "access_token": result.get_access_token(),
+        "access_token":
+            result.get_authorization_header()  # It returns "pop SignedHttpRequest"
+                .split()[1]
+            if result.is_pop_authorization() else result.get_access_token(),
         "expires_in": result.get_access_token_expiry_time() - int(time.time()),  # Convert epoch to count-down
         "id_token": result.get_raw_id_token(),  # New in pymsalruntime 0.8.1
         "id_token_claims": id_token_claims,
         "client_info": account.get_client_info(),
         "_account_id": account.get_account_id(),
-		"token_type": expected_token_type or "Bearer",  # Workaround its absence from broker
+		"token_type": "pop" if result.is_pop_authorization() else (
+            expected_token_type or "bearer"),  # Workaround "ssh-cert"'s absence from broker
         }.items() if v}
     likely_a_cert = return_value["access_token"].startswith("AAAA")  # Empirical observation
     if return_value["token_type"].lower() == "ssh-cert" and not likely_a_cert:
@@ -128,11 +132,16 @@ def _enable_msa_pt(params):
 def _signin_silently(
         authority, client_id, scopes, correlation_id=None, claims=None,
         enable_msa_pt=False,
+        auth_scheme=None,
         **kwargs):
     params = pymsalruntime.MSALRuntimeAuthParameters(client_id, authority)
     params.set_requested_scopes(scopes)
     if claims:
         params.set_decoded_claims(claims)
+    if auth_scheme:
+        params.set_pop_params(
+            auth_scheme._http_method, auth_scheme._url.netloc, auth_scheme._url.path,
+            auth_scheme._nonce)
     callback_data = _CallbackData()
     for k, v in kwargs.items():  # This can be used to support domain_hint, max_age, etc.
         if v is not None:
@@ -156,6 +165,7 @@ def _signin_interactively(
         claims=None,
         correlation_id=None,
         enable_msa_pt=False,
+        auth_scheme=None,
         **kwargs):
     params = pymsalruntime.MSALRuntimeAuthParameters(client_id, authority)
     params.set_requested_scopes(scopes)
@@ -178,6 +188,10 @@ def _signin_interactively(
         params.set_additional_parameter("msal_gui_thread", "true")  # Since pymsalruntime 0.8.1
     if enable_msa_pt:
         _enable_msa_pt(params)
+    if auth_scheme:
+        params.set_pop_params(
+            auth_scheme._http_method, auth_scheme._url.netloc, auth_scheme._url.path,
+            auth_scheme._nonce)
     for k, v in kwargs.items():  # This can be used to support domain_hint, max_age, etc.
         if v is not None:
             params.set_additional_parameter(k, str(v))
@@ -197,6 +211,7 @@ def _signin_interactively(
 
 def _acquire_token_silently(
         authority, client_id, account_id, scopes, claims=None, correlation_id=None,
+        auth_scheme=None,
         **kwargs):
     # For MSA PT scenario where you use the /organizations, yes,
     # acquireTokenSilently is expected to fail.  - Sam Wilson
@@ -208,6 +223,10 @@ def _acquire_token_silently(
     params.set_requested_scopes(scopes)
     if claims:
         params.set_decoded_claims(claims)
+    if auth_scheme:
+        params.set_pop_params(
+            auth_scheme._http_method, auth_scheme._url.netloc, auth_scheme._url.path,
+            auth_scheme._nonce)
     for k, v in kwargs.items():  # This can be used to support domain_hint, max_age, etc.
         if v is not None:
             params.set_additional_parameter(k, str(v))