Merge branch 'dev' into dharshanb/brokerSupportLinux

.github/workflows/RunIssueSentinel.yml

@@ -0,0 +1,17 @@
+name: Run issue sentinel
+on:
+  issues:
+    types: [opened, edited, closed]
+
+jobs:
+  Issue:
+    permissions:
+        issues: write
+    runs-on: ubuntu-latest
+    steps:
+      - name: Run Issue Sentinel
+        uses: Azure/issue-sentinel@v1
+        with:
+          password: ${{secrets.ISSUE_SENTINEL_PASSWORD}}
+          enable-similar-issues-scanning: true # Scan for similar issues
+          enable-security-issues-scanning: true # Scan for security issues

msal/__init__.py

@@ -26,12 +26,12 @@
 #------------------------------------------------------------------------------
 
 from .application import (
-    __version__,
     ClientApplication,
     ConfidentialClientApplication,
     PublicClientApplication,
     )
 from .oauth2cli.oidc import Prompt, IdTokenError
+from .sku import __version__
 from .token_cache import TokenCache, SerializableTokenCache
 from .auth_scheme import PopAuthScheme
 from .managed_identity import (

msal/application.py

@@ -6,6 +6,7 @@
 import warnings
 from threading import Lock
 from typing import Optional  # Needed in Python 3.7 & 3.8
+from urllib.parse import urlparse
 import os
 
 from .oauth2cli import Client, JwtAssertionCreator
@@ -19,10 +20,9 @@
 from .region import _detect_region
 from .throttled_http_client import ThrottledHttpClient
 from .cloudshell import _is_running_in_cloud_shell
+from .sku import SKU, __version__
 
 
-# The __init__.py will import this. Not the other way around.
-__version__ = "1.31.1"  # When releasing, also check and bump our dependencies's versions if needed
 
 logger = logging.getLogger(__name__)
 _AUTHORITY_TYPE_CLOUDSHELL = "CLOUDSHELL"
@@ -623,6 +623,9 @@ def __init__(
         # Here the self.authority will not be the same type as authority in input
         if oidc_authority and authority:
             raise ValueError("You can not provide both authority and oidc_authority")
+        if isinstance(authority, str) and urlparse(authority).path.startswith(
+            "/dstsv2"):  # dSTS authority's path always starts with "/dstsv2"
+            oidc_authority = authority  # So we treat it as if an oidc_authority
         try:
             authority_to_use = authority or "https://{}/common/".format(WORLD_WIDE)
             self.authority = Authority(
@@ -770,7 +773,7 @@ def _build_client(self, client_credential, authority, skip_regional_client=False
         client_assertion = None
         client_assertion_type = None
         default_headers = {
-            "x-client-sku": "MSAL.Python", "x-client-ver": __version__,
+            "x-client-sku": SKU, "x-client-ver": __version__,
             "x-client-os": sys.platform,
             "x-ms-lib-capability": "retry-after, h429",
         }

msal/broker.py

@@ -7,6 +7,7 @@
 import time
 import uuid
 
+from .sku import __version__, SKU
 
 logger = logging.getLogger(__name__)
 try:
@@ -136,13 +137,18 @@ def _get_new_correlation_id():
 def _enable_msa_pt(params):
     params.set_additional_parameter("msal_request_type", "consumer_passthrough")  # PyMsalRuntime 0.8+
 
+def _build_msal_runtime_auth_params(client_id, authority):
+    params = pymsalruntime.MSALRuntimeAuthParameters(client_id, authority)
+    params.set_additional_parameter("msal_client_sku", SKU)
+    params.set_additional_parameter("msal_client_ver", __version__)
+    return 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 = _build_msal_runtime_auth_params(client_id, authority)
     params.set_requested_scopes(scopes)
     if claims:
         params.set_decoded_claims(claims)
@@ -175,7 +181,7 @@ def _signin_interactively(
         enable_msa_pt=False,
         auth_scheme=None,
         **kwargs):
-    params = pymsalruntime.MSALRuntimeAuthParameters(client_id, authority)
+    params = _build_msal_runtime_auth_params(client_id, authority)
     params.set_requested_scopes(scopes)
     params.set_redirect_uri(
         _redirect_uri_on_mac if sys.platform == "darwin" else
@@ -231,7 +237,7 @@ def _acquire_token_silently(
     account = _read_account_by_id(account_id, correlation_id)
     if account is None:
         return
-    params = pymsalruntime.MSALRuntimeAuthParameters(client_id, authority)
+    params = _build_msal_runtime_auth_params(client_id, authority)
     params.set_requested_scopes(scopes)
     if claims:
         params.set_decoded_claims(claims)

msal/sku.py

@@ -0,0 +1,6 @@
+"""This module is from where we recieve the client sku name and version.
+"""
+
+# The __init__.py will import this. Not the other way around.
+__version__ = "1.31.1" # When releasing, also check and bump our dependencies's versions if needed
+SKU = "MSAL.Python"

tests/test_authority.py

@@ -104,32 +104,63 @@ def test_authority_with_path_should_be_used_as_is(self, oidc_discovery):
     "authorization_endpoint": "https://contoso.com/authorize",
     "token_endpoint": "https://contoso.com/token",
     })
-class TestOidcAuthority(unittest.TestCase):
+class OidcAuthorityTestCase(unittest.TestCase):
+    authority = "https://contoso.com/tenant"
+
+    def setUp(self):
+        # setUp() gives subclass a dynamic setup based on their authority
+        self.oidc_discovery_endpoint = (
+            # MSAL Python always does OIDC Discovery,
+            # not to be confused with Instance Discovery
+            # Here the test is to confirm the OIDC endpoint contains no "/v2.0"
+            self.authority + "/.well-known/openid-configuration")
+
     def test_authority_obj_should_do_oidc_discovery_and_skip_instance_discovery(
             self, oidc_discovery, instance_discovery):
         c = MinimalHttpClient()
-        a = Authority(None, c, oidc_authority_url="https://contoso.com/tenant")
+        a = Authority(None, c, oidc_authority_url=self.authority)
         instance_discovery.assert_not_called()
-        oidc_discovery.assert_called_once_with(
-            "https://contoso.com/tenant/.well-known/openid-configuration", c)
+        oidc_discovery.assert_called_once_with(self.oidc_discovery_endpoint, c)
         self.assertEqual(a.authorization_endpoint, 'https://contoso.com/authorize')
         self.assertEqual(a.token_endpoint, 'https://contoso.com/token')
 
     def test_application_obj_should_do_oidc_discovery_and_skip_instance_discovery(
             self, oidc_discovery, instance_discovery):
         app = msal.ClientApplication(
-            "id",
-            authority=None,
-            oidc_authority="https://contoso.com/tenant",
-            )
+            "id", authority=None, oidc_authority=self.authority)
         instance_discovery.assert_not_called()
         oidc_discovery.assert_called_once_with(
-            "https://contoso.com/tenant/.well-known/openid-configuration",
-            app.http_client)
+            self.oidc_discovery_endpoint, app.http_client)
         self.assertEqual(
             app.authority.authorization_endpoint, 'https://contoso.com/authorize')
         self.assertEqual(app.authority.token_endpoint, 'https://contoso.com/token')
 
+
+class DstsAuthorityTestCase(OidcAuthorityTestCase):
+    # Inherits OidcAuthority's test cases and run them with a dSTS authority
+    authority = (  # dSTS is single tenanted with a tenant placeholder
+        'https://test-instance1-dsts.dsts.core.azure-test.net/dstsv2/common')
+    authorization_endpoint = (
+        "https://some.url.dsts.core.azure-test.net/dstsv2/common/oauth2/authorize")
+    token_endpoint = (
+        "https://some.url.dsts.core.azure-test.net/dstsv2/common/oauth2/token")
+
+    @patch("msal.authority._instance_discovery")
+    @patch("msal.authority.tenant_discovery", return_value={
+        "authorization_endpoint": authorization_endpoint,
+        "token_endpoint": token_endpoint,
+    })  # We need to create new patches (i.e. mocks) for non-inherited test cases
+    def test_application_obj_should_accept_dsts_url_as_an_authority(
+            self, oidc_discovery, instance_discovery):
+        app = msal.ClientApplication("id", authority=self.authority)
+        instance_discovery.assert_not_called()
+        oidc_discovery.assert_called_once_with(
+            self.oidc_discovery_endpoint, app.http_client)
+        self.assertEqual(
+            app.authority.authorization_endpoint, self.authorization_endpoint)
+        self.assertEqual(app.authority.token_endpoint, self.token_endpoint)
+
+
 class TestAuthorityInternalHelperCanonicalize(unittest.TestCase):
 
     def test_canonicalize_tenant_followed_by_extra_paths(self):

tests/test_e2e.py

@@ -477,9 +477,10 @@ def get_lab_app(
     if os.getenv(env_client_id) and os.getenv(env_client_cert_path):
         # id came from https://docs.msidlab.com/accounts/confidentialclient.html
         client_id = os.getenv(env_client_id)
-        # Cert came from https://ms.portal.azure.com/#@microsoft.onmicrosoft.com/asset/Microsoft_Azure_KeyVault/Certificate/https://msidlabs.vault.azure.net/certificates/LabVaultAccessCert
         client_credential = {
-            "private_key_pfx_path": os.getenv(env_client_cert_path),
+            "private_key_pfx_path":
+                # Cert came from https://ms.portal.azure.com/#@microsoft.onmicrosoft.com/asset/Microsoft_Azure_KeyVault/Certificate/https://msidlabs.vault.azure.net/certificates/LabAuth
+                os.getenv(env_client_cert_path),
             "public_certificate": True,  # Opt in for SNI
             }
     elif os.getenv(env_client_id) and os.getenv(env_name2):
@@ -649,19 +650,27 @@ def _test_acquire_token_obo(self, config_pca, config_cca,
                 # Here we just test regional apps won't adversely break OBO
             http_client=None,
             ):
-        # 1. An app obtains a token representing a user, for our mid-tier service
-        pca = msal.PublicClientApplication(
-            config_pca["client_id"], authority=config_pca["authority"],
-            azure_region=azure_region,
-            http_client=http_client or MinimalHttpClient())
-        pca_result = pca.acquire_token_by_username_password(
-            config_pca["username"],
-            config_pca["password"],
-            scopes=config_pca["scope"],
-            )
-        self.assertIsNotNone(
-            pca_result.get("access_token"),
-            "PCA failed to get AT because %s" % json.dumps(pca_result, indent=2))
+        if "client_secret" not in config_pca:
+            # 1.a An app obtains a token representing a user, for our mid-tier service
+            result = msal.PublicClientApplication(
+                config_pca["client_id"], authority=config_pca["authority"],
+                azure_region=azure_region,
+                http_client=http_client or MinimalHttpClient(),
+                ).acquire_token_by_username_password(
+                    config_pca["username"], config_pca["password"],
+                    scopes=config_pca["scope"],
+                )
+        else:  # We repurpose the config_pca to contain client_secret for cca app 1
+            # 1.b An app obtains a token representing itself, for our mid-tier service
+            result = msal.ConfidentialClientApplication(
+                config_pca["client_id"], authority=config_pca["authority"],
+                client_credential=config_pca["client_secret"],
+                azure_region=azure_region,
+                http_client=http_client or MinimalHttpClient(),
+                ).acquire_token_for_client(scopes=config_pca["scope"])
+        assertion = result.get("access_token")
+        self.assertIsNotNone(assertion, "First app failed to get AT. {}".format(
+            json.dumps(result, indent=2)))
 
         # 2. Our mid-tier service uses OBO to obtain a token for downstream service
         cca = msal.ConfidentialClientApplication(
@@ -674,9 +683,9 @@ def _test_acquire_token_obo(self, config_pca, config_cca,
                 # That's fine if OBO app uses short-lived msal instance per session.
                 # Otherwise, the OBO app need to implement a one-cache-per-user setup.
             )
-        cca_result = cca.acquire_token_on_behalf_of(
-            pca_result['access_token'], config_cca["scope"])
-        self.assertNotEqual(None, cca_result.get("access_token"), str(cca_result))
+        cca_result = cca.acquire_token_on_behalf_of(assertion, config_cca["scope"])
+        self.assertIsNotNone(cca_result.get("access_token"), "OBO call failed: {}".format(
+            json.dumps(cca_result, indent=2)))
 
         # 3. Now the OBO app can simply store downstream token(s) in same session.
         #    Alternatively, if you want to persist the downstream AT, and possibly
@@ -685,13 +694,27 @@ def _test_acquire_token_obo(self, config_pca, config_cca,
         #    Assuming you already did that (which is not shown in this test case),
         #    the following part shows one of the ways to obtain an AT from cache.
         username = cca_result.get("id_token_claims", {}).get("preferred_username")
-        if username:  # It means CCA have requested an IDT w/ "profile" scope
-            self.assertEqual(config_cca["username"], username)
         accounts = cca.get_accounts(username=username)
-        assert len(accounts) == 1, "App is expected to partition token cache per user"
-        account = accounts[0]
+        if username is not None:  # It means CCA have requested an IDT w/ "profile" scope
+            assert config_cca["username"] == username, "Incorrect test case configuration"
+            self.assertEqual(1, len(accounts), "App is supposed to partition token cache per user")
+        account = accounts[0]  # Alternatively, cca app could just loop through each account
         result = cca.acquire_token_silent(config_cca["scope"], account)
-        self.assertEqual(cca_result["access_token"], result["access_token"])
+        self.assertTrue(
+            result and result.get("access_token") == cca_result["access_token"],
+            "CCA should hit an access token from cache: {}".format(
+                json.dumps(cca.token_cache._cache, indent=2)))
+        if "refresh_token" in cca_result:
+            result = cca.acquire_token_silent(
+                config_cca["scope"], account=account, force_refresh=True)
+            self.assertTrue(
+                result and "access_token" in result,
+                "CCA should get an AT silently, but we got this instead: {}".format(result))
+            self.assertNotEqual(
+                result["access_token"], cca_result["access_token"],
+                "CCA should get a new AT")
+        else:
+            logger.info("AAD did not issue a RT for OBO flow")
 
     def _test_acquire_token_by_client_secret(
             self, client_id=None, client_secret=None, authority=None, scope=None,
@@ -933,6 +956,31 @@ def test_acquire_token_obo(self):
 
         self._test_acquire_token_obo(config_pca, config_cca)
 
+    @unittest.skipUnless(
+        os.path.exists("tests/sp_obo.pem"),
+        "Need a 'tests/sp_obo.pem' private to run OBO for SP test")
+    def test_acquire_token_obo_for_sp(self):
+        authority = "https://login.windows-ppe.net/f686d426-8d16-42db-81b7-ab578e110ccd"
+        with open("tests/sp_obo.pem") as pem:
+            client_secret = {
+                "private_key": pem.read(),
+                "thumbprint": "378938210C976692D7F523B8C4FFBB645D17CE92",
+                }
+        midtier_app = {
+            "authority": authority,
+            "client_id": "c84e9c32-0bc9-4a73-af05-9efe9982a322",
+            "client_secret": client_secret,
+            "scope": ["23d08a1e-1249-4f7c-b5a5-cb11f29b6923/.default"],
+            #"username": "OBO-Client-PPE",  # We do NOT attempt locating initial_app by name
+            }
+        initial_app = {
+            "authority": authority,
+            "client_id": "9793041b-9078-4942-b1d2-babdc472cc0c",
+            "client_secret": client_secret,
+            "scope": [midtier_app["client_id"] + "/.default"],
+            }
+        self._test_acquire_token_obo(initial_app, midtier_app)
+
     def test_acquire_token_by_client_secret(self):
         # Vastly different than ArlingtonCloudTestCase.test_acquire_token_by_client_secret()
         _app = self.get_lab_app_object(
@@ -1323,7 +1371,7 @@ def test_at_pop_calling_pattern(self):
         # We skip it here because this test case has not yet initialize self.app
         # assert self.app.is_pop_supported()
         api_endpoint = "https://20.190.132.47/beta/me"
-        resp = requests.get(api_endpoint, verify=False)
+        resp = requests.get(api_endpoint, verify=False)  # @suppress py/bandit/requests-ssl-verify-disabled
         self.assertEqual(resp.status_code, 401, "Initial call should end with an http 401 error")
         result = self._get_shr_pop(**dict(
             self.get_lab_user(usertype="cloud"),  # This is generally not the current laptop's default AAD account
@@ -1334,6 +1382,9 @@ def test_at_pop_calling_pattern(self):
                 nonce=self._extract_pop_nonce(resp.headers.get("WWW-Authenticate")),
                 ),
             ))
+        # The api_endpoint is for test only and has no proper SSL certificate,
+        # so we suppress the CodeQL warning for disabling SSL certificate checks
+        # @suppress py/bandit/requests-ssl-verify-disabled
         resp = requests.get(api_endpoint, verify=False, headers={
             "Authorization": "pop {}".format(result["access_token"]),
             })