diff --git a/azure-devops/azext_devops/dev/team/_help.py b/azure-devops/azext_devops/dev/team/_help.py index 43aec8fa..a8b54141 100644 --- a/azure-devops/azext_devops/dev/team/_help.py +++ b/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:////' 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 diff --git a/azure-devops/azext_devops/dev/team/arguments.py b/azure-devops/azext_devops/dev/team/arguments.py index 0ba64cd0..8e7e0118 100644 --- a/azure-devops/azext_devops/dev/team/arguments.py +++ b/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:////' 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') diff --git a/azure-devops/azext_devops/dev/team/commands.py b/azure-devops/azext_devops/dev/team/commands.py index f341c818..71f55bca 100644 --- a/azure-devops/azext_devops/dev/team/commands.py +++ b/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) diff --git a/azure-devops/azext_devops/dev/team/service_endpoint.py b/azure-devops/azext_devops/dev/team/service_endpoint.py index c91815d5..24b52458 100644 --- a/azure-devops/azext_devops/dev/team/service_endpoint.py +++ b/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://// 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:////' 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 diff --git a/azure-devops/azext_devops/tests/latest/team/test_service_endpoint.py b/azure-devops/azext_devops/tests/latest/team/test_service_endpoint.py index 5fa73ff9..68b18d38 100644 --- a/azure-devops/azext_devops/tests/latest/team/test_service_endpoint.py +++ b/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() \ No newline at end of file