Fix my FIC CLI

azure-devops/azext_devops/dev/team/_help.py

@@ -59,6 +59,27 @@ def load_team_help():
       environment variable. You can learn more about this at https://aka.ms/azure-devops-cli-service-endpoint
     """
 
+    helps['devops service-endpoint convert'] = """
+    type: command
+    short-summary: Convert a service endpoint to use external federated credentials.
+    long-summary: |
+      Sends a migration request to the Azure DevOps external federated credential migration API
+      for the service connection specified via --azdo-subject. The --azdo-subject value must be in
+      'sc://<organization>/<project>/<serviceConnectionName>' format. The organization URL is
+      automatically derived from --azdo-subject (using https://dev.azure.com) unless
+      --origin is explicitly provided.
+    examples:
+      - name: Convert using only the service connection (organization derived automatically)
+        text: >
+          az devops service-endpoint convert
+          --azdo-subject "sc://myorg/myproject/myserviceconnection"
+      - name: Convert with an explicit environment URL
+        text: >
+          az devops service-endpoint convert
+          --azdo-subject "sc://myorg/myproject/myserviceconnection"
+          --origin https://dev.azure.com/myorg
+    """
+
     helps['devops security'] = """
     type: group
     short-summary: Manage security related operations

azure-devops/azext_devops/dev/team/arguments.py

@@ -87,6 +87,17 @@ def load_team_arguments(self, _):
                          help='Allow all pipelines to access this service endpoint.',
                          arg_type=get_three_state_flag())
 
+    with self.argument_context('devops service-endpoint convert') as context:
+        context.argument('azdo_subject',
+                         options_list=['--azdo-subject'],
+                         help="Federated credential subject of the Azure DevOps issuer, in "
+                              "'sc://<organization>/<project>/<serviceConnectionName>' format. "
+                              "The organization URL is derived from --azdo-subject when --organization is not provided.")
+        context.argument('origin',
+                         options_list=['--origin'],
+                         help='Azure DevOps organization URL. If omitted, the organization is derived from '
+                              '--azdo-subject (assumes https://dev.azure.com).')
+
     with self.argument_context('devops invoke') as context:
         context.argument('route_parameters', nargs='*',
                          help='Specifies the list of route parameters')

azure-devops/azext_devops/dev/team/commands.py

@@ -118,6 +118,7 @@ def load_team_commands(self, _):
         g.command('github create', 'create_github_service_endpoint')
         g.command('delete', 'delete_service_endpoint',
                   confirmation='Are you sure you want to delete this service-endpoint?')
+        g.command('convert', 'migrate_external_federated_credential')
 
     with self.command_group('devops team', command_type=teamOps) as g:
         g.command('create', 'create_team', table_transformer=transform_team_table_output)

azure-devops/azext_devops/dev/team/service_endpoint.py

@@ -10,7 +10,11 @@
 from knack.prompting import prompt_pass
 from knack.util import CLIError
 from azext_devops.devops_sdk.v5_0.service_endpoint.models import ServiceEndpoint, EndpointAuthorization
-from azext_devops.dev.common.services import get_service_endpoint_client, resolve_instance_and_project
+import requests
+from azure.cli.core._profile import Profile
+from azext_devops.dev.common.services import (get_service_endpoint_client,
+                                              get_token_from_az_login,
+                                              resolve_instance, resolve_instance_and_project)
 from azext_devops.dev.common.const import CLI_ENV_VARIABLE_PREFIX, AZ_DEVOPS_GITHUB_PAT_ENVKEY
 from azext_devops.dev.common.prompting import verify_is_a_tty_or_raise_error
 
@@ -201,6 +205,69 @@ def create_service_endpoint(service_endpoint_configuration,
     return client.create_service_endpoint(service_endpoint_to_create, project)
 
 
+def migrate_external_federated_credential(azdo_subject, origin=None, detect=None):
+    """Migrate a service endpoint to use external federated credentials.
+    :param azdo_subject: Service connection in sc://<organization>/<project>/<serviceConnectionName> format.
+    :type azdo_subject: str
+    """
+    import json
+    import re
+
+    match = re.match(r'^sc://([^/]+)/([^/]+)/(.+)$', azdo_subject, re.IGNORECASE)
+    if not match:
+        raise CLIError(
+            "--azdo-subject must be in 'sc://<organization>/<project>/<serviceConnectionName>' format.")
+
+    if origin is None and not detect:
+        # Derive the organization URL from the sc:// input (assumes dev.azure.com)
+        org_name = match.group(1)
+        origin = 'https://dev.azure.com/{0}'.format(org_name)
+        logger.debug("Derived organization URL from --azdo-subject: %s", origin)
+    elif detect:
+        # Only use resolve_instance when auto-detection from git config is requested
+        origin = resolve_instance(detect=detect, organization=origin)
+    # else: --origin explicitly provided, use it directly (supports codedev.ms, etc.)
+
+    # Acquire an Entra Bearer token for Azure DevOps — the public migration
+    # endpoint requires Bearer auth, not the Basic-wrapped token the SDK normally sends.
+    profile = Profile()
+    profile.get_current_account_user()  # ensures cache is loaded
+    subscriptions = profile.load_cached_subscriptions(False)
+    tenant_id = next(
+        (s['tenantId'] for s in subscriptions if s.get('isDefault')),
+        next((s['tenantId'] for s in subscriptions), None)
+    )
+    if not tenant_id:
+        raise CLIError("No Azure login found. Run 'az login' and try again.")
+    bearer_token = get_token_from_az_login(profile, tenant_id)
+    if not bearer_token:
+        raise CLIError("Failed to acquire an Entra token. Run 'az login' and try again.")
+
+    api_version = "7.2-preview.1"
+    migrate_url = "{0}/_apis/public/serviceendpoint/externalfederatedcredentialmigration?api-version={1}".format(
+        origin.rstrip('/'), api_version)
+
+    logger.debug("POST %s", migrate_url)
+    response = requests.post(
+        migrate_url,
+        headers={
+            'Authorization': 'Bearer {0}'.format(bearer_token),
+            'Content-Type': 'application/json'
+        },
+        data=json.dumps({'serviceConnectionInput': azdo_subject}),
+        timeout=30
+    )
+
+    if not response.ok:
+        try:
+            error_detail = response.json()
+        except ValueError:
+            error_detail = response.text[:200] if response.text else '(no body)'
+        raise CLIError('Migration request failed ({0}): {1}'.format(
+            response.status_code, error_detail))
+    return response.json()
+
+
 def update_service_endpoint(id, enable_for_all=None, organization=None,  # pylint: disable=redefined-builtin
                             project=None, detect=None):
     """Update a service endpoint

azure-devops/azext_devops/tests/latest/team/test_service_endpoint.py

@@ -20,7 +20,8 @@
                                                     create_github_service_endpoint,
                                                     create_azurerm_service_endpoint,
                                                     delete_service_endpoint,
-                                                    update_service_endpoint)
+                                                    update_service_endpoint,
+                                                    migrate_external_federated_credential)
 
 from azext_devops.dev.common.services import clear_connection_cache
 from azext_devops.tests.utils.authentication import AuthenticatedTests
@@ -149,5 +150,119 @@ def test_create_service_endpoint_ttyi_exception_azure_se(self):
         except NoTTYException as ex:
             self.assertEqual(str(ex), 'Please specify azure service principal key in AZURE_DEVOPS_EXT_AZURE_RM_SERVICE_PRINCIPAL_KEY environment variable in non-interactive mode or use --azure-rm-service-principal-certificate-path.')
 
+
+class TestMigrateExternalFederatedCredential(unittest.TestCase):
+
+    _TEST_AZDO_SUBJECT = 'sc://myorg/myproject/myconnection'
+    _TEST_BEARER_TOKEN = 'fake-entra-token'
+    _TEST_SUBSCRIPTIONS = [{'tenantId': 'tenant-123', 'isDefault': True}]
+
+    def _make_profile_mock(self, mock_profile_cls):
+        mock_profile = mock_profile_cls.return_value
+        mock_profile.load_cached_subscriptions.return_value = self._TEST_SUBSCRIPTIONS
+        return mock_profile
+
+    def _make_response_mock(self, ok=True, json_data=None, status_code=200, text=''):
+        from unittest.mock import MagicMock
+        mock_resp = MagicMock()
+        mock_resp.ok = ok
+        mock_resp.status_code = status_code
+        mock_resp.text = text
+        mock_resp.json.return_value = json_data or {'status': 'success'}
+        return mock_resp
+
+    @patch('azext_devops.dev.team.service_endpoint.get_token_from_az_login', return_value=_TEST_BEARER_TOKEN)
+    @patch('azext_devops.dev.team.service_endpoint.Profile')
+    @patch('azext_devops.dev.team.service_endpoint.requests.post')
+    def test_convert_derives_org_from_subject(self, mock_post, mock_profile_cls, mock_get_token):
+        self._make_profile_mock(mock_profile_cls)
+        mock_post.return_value = self._make_response_mock()
+
+        result = migrate_external_federated_credential(azdo_subject=self._TEST_AZDO_SUBJECT)
+
+        mock_post.assert_called_once()
+        url = mock_post.call_args[0][0]
+        self.assertIn('https://dev.azure.com/myorg', url)
+        self.assertIn('externalfederatedcredentialmigration', url)
+        self.assertEqual(result, {'status': 'success'})
+
+    @patch('azext_devops.dev.team.service_endpoint.get_token_from_az_login', return_value=_TEST_BEARER_TOKEN)
+    @patch('azext_devops.dev.team.service_endpoint.Profile')
+    @patch('azext_devops.dev.team.service_endpoint.requests.post')
+    def test_convert_uses_explicit_origin(self, mock_post, mock_profile_cls, mock_get_token):
+        self._make_profile_mock(mock_profile_cls)
+        mock_post.return_value = self._make_response_mock()
+        explicit_origin = 'https://dev.azure.com/otherorg'
+
+        migrate_external_federated_credential(azdo_subject=self._TEST_AZDO_SUBJECT, origin=explicit_origin)
+
+        url = mock_post.call_args[0][0]
+        self.assertIn('otherorg', url)
+
+    @patch('azext_devops.dev.team.service_endpoint.get_token_from_az_login', return_value=_TEST_BEARER_TOKEN)
+    @patch('azext_devops.dev.team.service_endpoint.Profile')
+    @patch('azext_devops.dev.team.service_endpoint.requests.post')
+    def test_convert_sends_bearer_token(self, mock_post, mock_profile_cls, mock_get_token):
+        self._make_profile_mock(mock_profile_cls)
+        mock_post.return_value = self._make_response_mock()
+
+        migrate_external_federated_credential(azdo_subject=self._TEST_AZDO_SUBJECT)
+
+        headers = mock_post.call_args[1]['headers']
+        self.assertEqual(headers['Authorization'], 'Bearer {0}'.format(self._TEST_BEARER_TOKEN))
+        self.assertEqual(headers['Content-Type'], 'application/json')
+
+    @patch('azext_devops.dev.team.service_endpoint.get_token_from_az_login', return_value=_TEST_BEARER_TOKEN)
+    @patch('azext_devops.dev.team.service_endpoint.Profile')
+    @patch('azext_devops.dev.team.service_endpoint.requests.post')
+    def test_convert_sends_subject_in_body(self, mock_post, mock_profile_cls, mock_get_token):
+        import json
+        self._make_profile_mock(mock_profile_cls)
+        mock_post.return_value = self._make_response_mock()
+
+        migrate_external_federated_credential(azdo_subject=self._TEST_AZDO_SUBJECT)
+
+        body = json.loads(mock_post.call_args[1]['data'])
+        self.assertEqual(body['serviceConnectionInput'], self._TEST_AZDO_SUBJECT)
+
+    @patch('azext_devops.dev.team.service_endpoint.get_token_from_az_login', return_value=_TEST_BEARER_TOKEN)
+    @patch('azext_devops.dev.team.service_endpoint.Profile')
+    @patch('azext_devops.dev.team.service_endpoint.requests.post')
+    def test_convert_raises_on_http_error(self, mock_post, mock_profile_cls, mock_get_token):
+        self._make_profile_mock(mock_profile_cls)
+        mock_post.return_value = self._make_response_mock(ok=False, status_code=400, text='Bad Request')
+
+        with self.assertRaises(CLIError) as ctx:
+            migrate_external_federated_credential(azdo_subject=self._TEST_AZDO_SUBJECT)
+
+        self.assertIn('400', str(ctx.exception))
+        self.assertIn('Bad Request', str(ctx.exception))
+
+    def test_convert_raises_on_invalid_subject_format(self):
+        with self.assertRaises(CLIError) as ctx:
+            migrate_external_federated_credential(azdo_subject='not-a-valid-subject')
+
+        self.assertIn('--azdo-subject', str(ctx.exception))
+
+    @patch('azext_devops.dev.team.service_endpoint.get_token_from_az_login', return_value='')
+    @patch('azext_devops.dev.team.service_endpoint.Profile')
+    def test_convert_raises_when_no_entra_token(self, mock_profile_cls, mock_get_token):
+        self._make_profile_mock(mock_profile_cls)
+
+        with self.assertRaises(CLIError) as ctx:
+            migrate_external_federated_credential(azdo_subject=self._TEST_AZDO_SUBJECT)
+
+        self.assertIn('az login', str(ctx.exception))
+
+    @patch('azext_devops.dev.team.service_endpoint.Profile')
+    def test_convert_raises_when_no_subscriptions(self, mock_profile_cls):
+        mock_profile = mock_profile_cls.return_value
+        mock_profile.load_cached_subscriptions.return_value = []
+
+        with self.assertRaises(CLIError) as ctx:
+            migrate_external_federated_credential(azdo_subject=self._TEST_AZDO_SUBJECT)
+
+        self.assertIn('az login', str(ctx.exception))
+
 if __name__ == '__main__':
     unittest.main()