auth: support service principal sn+issuer auth (#7609)

src/azure-cli-core/HISTORY.rst

@@ -2,10 +2,9 @@
 
 Release History
 ===============
-
 2.0.50
 ++++++
-* Fix issue where update commands using `--remove` and `--ids` fail after first update is applied to first resource in ids list.
+* auth: support service principal sn+issuer auth
 
 2.0.49
 ++++++

src/azure-cli-core/azure/cli/core/_profile.py

@@ -10,6 +10,7 @@
 import json
 import os
 import os.path
+import re
 from copy import deepcopy
 from enum import Enum
 from six.moves import BaseHTTPServer
@@ -46,6 +47,7 @@
 _SERVICE_PRINCIPAL_TENANT = 'servicePrincipalTenant'
 _SERVICE_PRINCIPAL_CERT_FILE = 'certificateFile'
 _SERVICE_PRINCIPAL_CERT_THUMBPRINT = 'thumbprint'
+_SERVICE_PRINCIPAL_CERT_SN_ISSUER_AUTH = 'useCertSNIssuerAuth'
 _TOKEN_ENTRY_USER_ID = 'userId'
 _TOKEN_ENTRY_TOKEN_TYPE = 'tokenType'
 # This could mean either real access token, or client secret of a service principal
@@ -76,7 +78,6 @@ def load_subscriptions(cli_ctx, all_clouds=False, refresh=False):
 
 
 def _get_authority_url(cli_ctx, tenant):
-    import re
     authority_url = cli_ctx.cloud.endpoints.active_directory
     is_adfs = bool(re.match('.+(/adfs|/adfs/)$', authority_url, re.I))
     if is_adfs:
@@ -163,7 +164,8 @@ def find_subscriptions_on_login(self,
                                     tenant,
                                     use_device_code=False,
                                     allow_no_subscriptions=False,
-                                    subscription_finder=None):
+                                    subscription_finder=None,
+                                    use_cert_sn_issuer=None):
         from azure.cli.core._debug import allow_debug_adal_connection
         allow_debug_adal_connection()
         subscriptions = []
@@ -193,9 +195,10 @@ def find_subscriptions_on_login(self,
             if is_service_principal:
                 if not tenant:
                     raise CLIError('Please supply tenant using "--tenant"')
-                sp_auth = ServicePrincipalAuth(password)
+                sp_auth = ServicePrincipalAuth(password, use_cert_sn_issuer)
                 subscriptions = subscription_finder.find_from_service_principal_id(
                     username, sp_auth, tenant, self._ad_resource_uri)
+
             else:
                 subscriptions = subscription_finder.find_from_user_account(
                     username, password, tenant, self._ad_resource_uri)
@@ -219,13 +222,14 @@ def find_subscriptions_on_login(self,
             if not subscriptions:
                 return []
 
-        consolidated = self._normalize_properties(subscription_finder.user_id, subscriptions, is_service_principal)
+        consolidated = self._normalize_properties(subscription_finder.user_id, subscriptions,
+                                                  is_service_principal, bool(use_cert_sn_issuer))
 
         self._set_subscriptions(consolidated)
         # use deepcopy as we don't want to persist these changes to file.
         return deepcopy(consolidated)
 
-    def _normalize_properties(self, user, subscriptions, is_service_principal):
+    def _normalize_properties(self, user, subscriptions, is_service_principal, cert_sn_issuer_auth=None):
         consolidated = []
         for s in subscriptions:
             consolidated.append({
@@ -240,6 +244,8 @@ def _normalize_properties(self, user, subscriptions, is_service_principal):
                 _TENANT_ID: s.tenant_id,
                 _ENVIRONMENT_NAME: self.cli_ctx.cloud.name
             })
+            if cert_sn_issuer_auth:
+                consolidated[-1][_USER_ENTITY][_SERVICE_PRINCIPAL_CERT_SN_ISSUER_AUTH] = True
         return consolidated
 
     def _build_tenant_level_accounts(self, tenants):
@@ -492,7 +498,9 @@ def _retrieve_token():
                 if user_type == _USER:
                     return self._creds_cache.retrieve_token_for_user(username_or_sp_id,
                                                                      account[_TENANT_ID], resource)
-                return self._creds_cache.retrieve_token_for_service_principal(username_or_sp_id, resource)
+                use_cert_sn_issuer = account[_USER_ENTITY].get(_SERVICE_PRINCIPAL_CERT_SN_ISSUER_AUTH)
+                return self._creds_cache.retrieve_token_for_service_principal(username_or_sp_id, resource,
+                                                                              use_cert_sn_issuer)
 
             def _retrieve_tokens_from_external_tenants():
                 external_tokens = []
@@ -863,15 +871,16 @@ def retrieve_token_for_user(self, username, tenant, resource):
             self.persist_cached_creds()
         return (token_entry[_TOKEN_ENTRY_TOKEN_TYPE], token_entry[_ACCESS_TOKEN], token_entry)
 
-    def retrieve_token_for_service_principal(self, sp_id, resource):
+    def retrieve_token_for_service_principal(self, sp_id, resource, use_cert_sn_issuer=False):
         self.load_adal_token_cache()
         matched = [x for x in self._service_principal_creds if sp_id == x[_SERVICE_PRINCIPAL_ID]]
         if not matched:
             raise CLIError("Please run 'az account set' to select active account.")
         cred = matched[0]
         context = self._auth_ctx_factory(self._ctx, cred[_SERVICE_PRINCIPAL_TENANT], None)
         sp_auth = ServicePrincipalAuth(cred.get(_ACCESS_TOKEN, None) or
-                                       cred.get(_SERVICE_PRINCIPAL_CERT_FILE, None))
+                                       cred.get(_SERVICE_PRINCIPAL_CERT_FILE, None),
+                                       use_cert_sn_issuer)
         token_entry = sp_auth.acquire_token(context, resource, sp_id)
         return (token_entry[_TOKEN_ENTRY_TOKEN_TYPE], token_entry[_ACCESS_TOKEN], token_entry)
 
@@ -948,26 +957,33 @@ def remove_all_cached_creds(self):
 
 class ServicePrincipalAuth(object):
 
-    def __init__(self, password_arg_value):
+    def __init__(self, password_arg_value, use_cert_sn_issuer=None):
         if not password_arg_value:
             raise CLIError('missing secret or certificate in order to '
                            'authnenticate through a service principal')
         if os.path.isfile(password_arg_value):
             certificate_file = password_arg_value
             from OpenSSL.crypto import load_certificate, FILETYPE_PEM
             self.certificate_file = certificate_file
+            self.public_certificate = None
             with open(certificate_file, 'r') as file_reader:
                 self.cert_file_string = file_reader.read()
                 cert = load_certificate(FILETYPE_PEM, self.cert_file_string)
                 self.thumbprint = cert.digest("sha1").decode()
+                if use_cert_sn_issuer:
+                    # low-tech but safe parsing based on
+                    # https://github.com/libressl-portable/openbsd/blob/master/src/lib/libcrypto/pem/pem.h
+                    match = re.search(r'\-+BEGIN CERTIFICATE.+\-+(?P<public>[^-]+)\-+END CERTIFICATE.+\-+',
+                                      self.cert_file_string, re.I)
+                    self.public_certificate = match.group('public').strip()
         else:
             self.secret = password_arg_value
 
     def acquire_token(self, authentication_context, resource, client_id):
         if hasattr(self, 'secret'):
             return authentication_context.acquire_token_with_client_credentials(resource, client_id, self.secret)
         return authentication_context.acquire_token_with_client_certificate(resource, client_id, self.cert_file_string,
-                                                                            self.thumbprint)
+                                                                            self.thumbprint, self.public_certificate)
 
     def get_entry_to_persist(self, sp_id, tenant):
         entry = {

src/azure-cli-core/azure/cli/core/tests/test_profile.py

@@ -1165,7 +1165,34 @@ def test_find_subscriptions_from_service_principal_using_cert(self, mock_auth_co
         mock_arm_client.tenants.list.assert_not_called()
         mock_auth_context.acquire_token.assert_not_called()
         mock_auth_context.acquire_token_with_client_certificate.assert_called_once_with(
-            mgmt_resource, 'my app', mock.ANY, mock.ANY)
+            mgmt_resource, 'my app', mock.ANY, mock.ANY, None)
+
+    @mock.patch('adal.AuthenticationContext', autospec=True)
+    def test_find_subscriptions_from_service_principal_using_cert_sn_issuer(self, mock_auth_context):
+        cli = DummyCli()
+        mock_auth_context.acquire_token_with_client_certificate.return_value = self.token_entry1
+        mock_arm_client = mock.MagicMock()
+        mock_arm_client.subscriptions.list.return_value = [self.subscription1]
+        finder = SubscriptionFinder(cli, lambda _, _1, _2: mock_auth_context, None, lambda _: mock_arm_client)
+        mgmt_resource = 'https://management.core.windows.net/'
+
+        curr_dir = os.path.dirname(os.path.realpath(__file__))
+        test_cert_file = os.path.join(curr_dir, 'sp_cert.pem')
+        with open(test_cert_file) as cert_file:
+            cert_file_string = cert_file.read()
+        match = re.search(r'\-+BEGIN CERTIFICATE.+\-+(?P<public>[^-]+)\-+END CERTIFICATE.+\-+',
+                          cert_file_string, re.I)
+        public_certificate = match.group('public').strip()
+        # action
+        subs = finder.find_from_service_principal_id('my app', ServicePrincipalAuth(test_cert_file, use_cert_sn_issuer=True),
+                                                     self.tenant_id, mgmt_resource)
+
+        # assert
+        self.assertEqual([self.subscription1], subs)
+        mock_arm_client.tenants.list.assert_not_called()
+        mock_auth_context.acquire_token.assert_not_called()
+        mock_auth_context.acquire_token_with_client_certificate.assert_called_once_with(
+            mgmt_resource, 'my app', mock.ANY, mock.ANY, public_certificate)
 
     @mock.patch('adal.AuthenticationContext', autospec=True)
     def test_refresh_accounts_one_user_account(self, mock_auth_context):

src/azure-cli-core/setup.py

@@ -53,7 +53,7 @@
 
 # TODO These dependencies should be updated to reflect only what this package needs
 DEPENDENCIES = [
-    'adal>=1.0.2',
+    'adal>=1.2.0',
     'argcomplete>=1.8.0',
     'azure-cli-telemetry',
     'colorama>=0.3.9',

src/command_modules/azure-cli-profile/HISTORY.rst

@@ -2,10 +2,9 @@
 
 Release History
 ===============
-
 2.1.2
 ++++++
-* Minor fixes
+* az login: expose --use-cert-sn-issuer flag for service principal login with cert auto-rolls
 
 2.1.1
 ++++++

src/command_modules/azure-cli-profile/azure/cli/command_modules/profile/__init__.py

@@ -51,6 +51,7 @@ def load_arguments(self, command):
             c.argument('identity_port', type=int, help="the port to retrieve tokens for login. Default: 50342", arg_group='Managed Service Identity')
             c.argument('use_device_code', action='store_true',
                        help="Use CLI's old authentication flow based on device code. CLI will also use this if it can't launch a browser in your behalf, e.g. in remote SSH or Cloud Shell")
+            c.argument('use_cert_sn_issuer', action='store_true', help='used with a service principal configured with Subject Name and Issuer Authentication in order to support automatic certificate rolls')
 
         with self.argument_context('logout') as c:
             c.argument('username', help='account user, if missing, logout the current active account')

src/command_modules/azure-cli-profile/azure/cli/command_modules/profile/custom.py

@@ -86,7 +86,7 @@ def account_clear(cmd):
 
 # pylint: disable=inconsistent-return-statements
 def login(cmd, username=None, password=None, service_principal=None, tenant=None, allow_no_subscriptions=False,
-          identity=False, use_device_code=False):
+          identity=False, use_device_code=False, use_cert_sn_issuer=None):
     """Log in to access Azure subscriptions"""
     from adal.adal_error import AdalError
     import requests
@@ -96,6 +96,10 @@ def login(cmd, username=None, password=None, service_principal=None, tenant=None
         raise CLIError("usage error: '--identity' is not applicable with other arguments")
     if any([password, service_principal, username, identity]) and use_device_code:
         raise CLIError("usage error: '--use-device-code' is not applicable with other arguments")
+    if use_cert_sn_issuer and not service_principal:
+        raise CLIError("usage error: '--use-sn-issuer' is only applicable with a service principal")
+    if service_principal and not username:
+        raise CLIError('usage error: --service-principal --username NAME --password SECRET --tenant TENANT')
 
     interactive = False
 
@@ -125,7 +129,8 @@ def login(cmd, username=None, password=None, service_principal=None, tenant=None
             service_principal,
             tenant,
             use_device_code=use_device_code,
-            allow_no_subscriptions=allow_no_subscriptions)
+            allow_no_subscriptions=allow_no_subscriptions,
+            use_cert_sn_issuer=use_cert_sn_issuer)
     except AdalError as err:
         # try polish unfriendly server errors
         if username: