[Key Vault] az keyvault create: Support new Managed HSM SKUs Standard_B1v2, Standard_B5v2, Standard_B10v2, Standard_B15v2, Standard_B20v2

src/azure-cli-testsdk/azure/cli/testsdk/base.py

@@ -24,7 +24,8 @@
                       patch_progress_controller, patch_get_current_system_username)
 from .exceptions import CliExecutionError
 from .utilities import (find_recording_dir, StorageAccountKeyReplacer, GraphClientPasswordReplacer,
-                        MSGraphClientPasswordReplacer, AADAuthRequestFilter, EmailAddressReplacer)
+                        MSGraphClientPasswordReplacer, MSGraphUserProfileReplacer, AADAuthRequestFilter,
+                        EmailAddressReplacer)
 from .reverse_dependency import get_dummy_cli
 
 logger = logging.getLogger('azure.cli.testsdk')
@@ -89,7 +90,7 @@ def __init__(self, method_name, config_file=None, recording_name=None,
         self.kwargs = {}
         self.test_guid_count = 0
         self._processors_to_reset = [StorageAccountKeyReplacer(), GraphClientPasswordReplacer(),
-                                     MSGraphClientPasswordReplacer()]
+                                     MSGraphClientPasswordReplacer(), MSGraphUserProfileReplacer()]
         default_recording_processors = [
             SubscriptionRecordingProcessor(MOCKED_SUBSCRIPTION_ID),
             AADAuthRequestFilter(),

src/azure-cli-testsdk/azure/cli/testsdk/scenario_tests/tests/test_msgraph_user_profile_replacer.py

@@ -0,0 +1,104 @@
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+
+import json
+import unittest
+
+from vcr.request import Request
+
+from azure.cli.testsdk.utilities import MSGraphUserProfileReplacer
+
+ME_ID = '11111111-2222-3333-4444-555555555555'
+
+
+def _user(display_name='Jane Li', given_name='Jane', surname='Li', user_id=ME_ID):
+    return {'@odata.type': '#microsoft.graph.user', 'id': user_id, 'businessPhones': ['+1 555 0100'],
+            'displayName': display_name, 'givenName': given_name, 'jobTitle': 'Engineer',
+            'mail': 'test@example.com', 'mobilePhone': '+1 555 0101', 'officeLocation': 'Building 1',
+            'surname': surname, 'userPrincipalName': 'test@example.com'}
+
+
+def _response(obj, raw=None):
+    return {'body': {'string': raw if raw is not None else json.dumps(obj)}}
+
+
+class TestMSGraphUserProfileReplacer(unittest.TestCase):
+    def _process(self, processor, uri, body, method='GET'):
+        processor.process_request(Request(method, uri, None, {}))
+        return processor.process_response(_response(body))['body']['string']
+
+    def test_me_response_is_sanitized(self):
+        rp = MSGraphUserProfileReplacer()
+        result = json.loads(self._process(rp, 'https://graph.microsoft.com/v1.0/me', _user()))
+        self.assertEqual(result['displayName'], 'Test User')
+        self.assertEqual(result['givenName'], 'Test')
+        self.assertEqual(result['surname'], 'User')
+        self.assertEqual(result['businessPhones'], [])
+        self.assertIsNone(result['mobilePhone'])
+        self.assertIsNone(result['jobTitle'])
+        self.assertIsNone(result['officeLocation'])
+        self.assertEqual(result['id'], ME_ID)
+
+    def test_short_surname_does_not_change_keys_or_identifiers(self):
+        rp = MSGraphUserProfileReplacer()
+        self._process(rp, 'https://graph.microsoft.com/v1.0/me', _user())
+        page = {'@odata.nextLink': 'https://graph.microsoft.com/v1.0/users?$skiptoken=Li42',
+                'value': [_user(display_name='Linda', given_name='Linda', surname='Lin', user_id='other')]}
+        result = json.loads(self._process(rp, 'https://graph.microsoft.com/v1.0/users', page))
+        self.assertIn('@odata.nextLink', result)
+        self.assertEqual(result['@odata.nextLink'], page['@odata.nextLink'])
+        self.assertEqual(result['value'][0]['displayName'], 'Linda')
+        self.assertEqual(result['value'][0]['surname'], 'Lin')
+        self.assertEqual(result['value'][0]['businessPhones'], [])
+
+    def test_escaped_names_are_replaced(self):
+        rp = MSGraphUserProfileReplacer()
+        me = _user(display_name='José Núñez', given_name='José', surname='Núñez')
+        raw = json.dumps(me)  # ensure_ascii escapes the names, e.g. Jos\u00e9
+        self.assertIn('\\u00e9', raw)
+        rp.process_request(Request('GET', 'https://graph.microsoft.com/v1.0/me', None, {}))
+        result = rp.process_response(_response(None, raw=raw))['body']['string']
+        self.assertNotIn('Jos', result)
+        self.assertNotIn('N\\u00fa\\u00f1ez', result)
+
+        other = {'value': [{'principalDisplayName': 'José Núñez', 'note': 'José Núñez owns this'}]}
+        result = json.loads(self._process(rp, 'https://management.azure.com/x', other))
+        self.assertEqual(result['value'][0]['principalDisplayName'], 'Test User')
+        # only exact values are replaced
+        self.assertEqual(result['value'][0]['note'], 'José Núñez owns this')
+
+    def test_signed_in_user_in_directory_objects(self):
+        rp = MSGraphUserProfileReplacer()
+        self._process(rp, 'https://graph.microsoft.com/v1.0/me', _user())
+        body = {'value': [_user(display_name='Jane L.', given_name='J', surname='L')]}
+        result = json.loads(self._process(rp, 'https://graph.microsoft.com/v1.0/directoryObjects/getByIds',
+                                          body, method='POST'))
+        self.assertEqual(result['value'][0]['displayName'], 'Test User')
+        self.assertEqual(result['value'][0]['surname'], 'User')
+
+    def test_sovereign_cloud_me_endpoints(self):
+        for uri in ('https://graph.microsoft.us/v1.0/me', 'https://dod-graph.microsoft.us/v1.0/me',
+                    'https://microsoftgraph.chinacloudapi.cn/v1.0/me'):
+            rp = MSGraphUserProfileReplacer()
+            self._process(rp, uri, _user())
+            result = json.loads(self._process(rp, 'https://management.azure.com/x', {'name': 'Jane Li'}))
+            self.assertEqual(result['name'], 'Test User', uri)
+
+    def test_non_json_and_unrelated_bodies_are_untouched(self):
+        rp = MSGraphUserProfileReplacer()
+        self.assertEqual(rp.process_response(_response(None, raw='not json'))['body']['string'], 'not json')
+        raw = '{"a":  "b"}'
+        self.assertEqual(rp.process_response(_response(None, raw=raw))['body']['string'], raw)
+
+    def test_reset(self):
+        rp = MSGraphUserProfileReplacer()
+        self._process(rp, 'https://graph.microsoft.com/v1.0/me', _user())
+        rp.reset()
+        result = json.loads(self._process(rp, 'https://management.azure.com/x', {'name': 'Jane Li'}))
+        self.assertEqual(result['name'], 'Jane Li')
+
+
+if __name__ == '__main__':
+    unittest.main()

src/azure-cli-testsdk/azure/cli/testsdk/utilities.py

@@ -242,6 +242,96 @@ def process_response(self, response):
         return response
 
 
+class MSGraphUserProfileReplacer(RecordingProcessor):
+    """Remove personal profile data of Microsoft Graph user objects from JSON responses.
+
+    - Contact fields (phones, job title, office location, address) are cleared on every Graph user object.
+    - The signed-in user is learned from the `/me` response of any built-in cloud's Graph endpoint. Its name fields
+      are replaced with mock values, and so is any later JSON string value that exactly equals one of its names.
+
+    Only decoded JSON values are changed; keys, other values and non-JSON bodies are left untouched.
+    """
+
+    MOCKED_DISPLAY_NAME = 'Test User'
+    MOCKED_GIVEN_NAME = 'Test'
+    MOCKED_SURNAME = 'User'
+    _NAME_FIELDS = (('displayName', MOCKED_DISPLAY_NAME), ('givenName', MOCKED_GIVEN_NAME),
+                    ('surname', MOCKED_SURNAME))
+    _CLEARED_FIELDS = {'businessPhones': [], 'mobilePhone': None, 'faxNumber': None, 'jobTitle': None,
+                       'officeLocation': None, 'department': None, 'companyName': None, 'streetAddress': None,
+                       'city': None, 'state': None, 'postalCode': None, 'country': None}
+    # Hosts of `microsoft_graph_resource_id` for the built-in clouds in azure.cli.core.cloud
+    GRAPH_HOSTS = ('graph.microsoft.com', 'graph.microsoft.us', 'dod-graph.microsoft.us',
+                   'microsoftgraph.chinacloudapi.cn', 'graph.microsoft.de', 'graph.svc.sovcloud.fr')
+
+    def __init__(self):
+        self._is_me_request = False
+        self._me_id = None
+        self._names = {}
+
+    def reset(self):
+        self._is_me_request = False
+        self._me_id = None
+        self._names = {}
+
+    def process_request(self, request):
+        from urllib.parse import urlparse
+        url = urlparse(request.uri)
+        self._is_me_request = ((url.hostname or '').lower() in self.GRAPH_HOSTS and
+                               url.path.rstrip('/').lower().endswith('/me') and request.method.upper() == 'GET')
+        return request
+
+    def _learn_me(self, me):
+        if not isinstance(me, dict):
+            return
+        if isinstance(me.get('id'), str):
+            self._me_id = me['id']
+        for key, mock in self._NAME_FIELDS:
+            if isinstance(me.get(key), str) and me[key]:
+                self._names[me[key]] = mock
+
+    def _scrub(self, obj):
+        if isinstance(obj, list):
+            return [self._scrub(item) for item in obj]
+        if isinstance(obj, dict):
+            result = {key: self._scrub(value) for key, value in obj.items()}
+            if 'userPrincipalName' in obj:
+                for key, value in self._CLEARED_FIELDS.items():
+                    if key in result:
+                        result[key] = value
+                if self._me_id is not None and obj.get('id') == self._me_id:
+                    for key, mock in self._NAME_FIELDS:
+                        if result.get(key):
+                            result[key] = mock
+            return result
+        if isinstance(obj, str):
+            return self._names.get(obj, obj)
+        return obj
+
+    def process_response(self, response):
+        import json
+
+        is_me_response, self._is_me_request = self._is_me_request, False
+        body = response['body']['string']
+        if not body:
+            return response
+        try:
+            body = _byte_to_str(body)
+            data = json.loads(body)
+        except (UnicodeDecodeError, ValueError):
+            return response
+
+        if is_me_response:
+            self._learn_me(data)
+        if not self._names and '"userPrincipalName"' not in body:
+            return response
+
+        scrubbed = self._scrub(data)
+        if scrubbed != data:
+            response['body']['string'] = json.dumps(scrubbed)
+        return response
+
+
 class AADAuthRequestFilter(RecordingProcessor):
     """Remove oauth authentication requests and responses from recording.
     This is derived from OAuthRequestResponsesFilter.