Skip to content

Commit

Permalink
Merge pull request #8705 from kyleknap/ssm-session-profile
Browse files Browse the repository at this point in the history 
Only pass profile parameter in command to session manager plugin
  • Loading branch information
@kyleknap
kyleknap authored May 30, 2024
2 parents df124ea + e063856 commit 2ea405f
Show file tree
Hide file tree
Showing 4 changed files with 272 additions and 27 deletions.
5 changes: 5 additions & 0 deletions .changes/next-release/bugfix-ssmstartsession-37817.json
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
{
"type": "bugfix",
"category": "``ssm start-session``",
"description": "Only provide profile name to session-manager-plugin if provided using --profile flag"
}
9 changes: 5 additions & 4 deletions awscli/customizations/sessionmanager.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -104,11 +104,12 @@ def invoke(self, service_name, operation_name, parameters,
response = client.start_session(**parameters)
session_id = response['SessionId']
region_name = client.meta.region_name
# profile_name is used to passed on to session manager plugin
# Profile_name is used to passed on to session manager plugin
# to fetch same profile credentials to make an api call in the plugin.
# If no profile is passed then pass on empty string
profile_name = self._session.profile \
if self._session.profile is not None else ''
# If --profile flag is configured, pass it to Session Manager plugin.
# If not, set empty string.
profile_name = parsed_globals.profile \
if parsed_globals.profile is not None else ''
endpoint_url = client.meta.endpoint_url
ssm_env_name = self.DEFAULT_SSM_ENV_NAME

Expand Down
216 changes: 214 additions & 2 deletions tests/functional/ssm/test_start_session.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,8 @@

from awscli.testutils import BaseAWSCommandParamsTest
from awscli.testutils import BaseAWSHelpOutputTest
from awscli.testutils import mock
from awscli.testutils import create_clidriver, mock, temporary_file
from botocore.exceptions import ProfileNotFound


class TestSessionManager(BaseAWSCommandParamsTest):
Expand All @@ -41,7 +42,7 @@ def test_start_session_success(self, mock_check_output, mock_check_call):
json.dumps(expected_response),
mock.ANY,
"StartSession",
mock.ANY,
"",
json.dumps(start_session_params),
mock.ANY,
],
Expand Down Expand Up @@ -80,13 +81,224 @@ def test_start_session_with_new_version_plugin_success(
ssm_env_name,
mock.ANY,
"StartSession",
"",
json.dumps(start_session_params),
mock.ANY,
],
env=expected_env,
)

@mock.patch("awscli.customizations.sessionmanager.check_call")
@mock.patch("awscli.customizations.sessionmanager.check_output")
def test_profile_parameter_passed_to_sessionmanager_plugin(
self, mock_check_output, mock_check_call
):
cmdline = (
"ssm start-session --target instance-id "
"--profile user_profile"
)
mock_check_call.return_value = 0
mock_check_output.return_value = "1.2.500.0\n"
expected_response = {
"SessionId": "session-id",
"TokenValue": "token-value",
"StreamUrl": "stream-url",
}
self.parsed_responses = [expected_response]
ssm_env_name = "AWS_SSM_START_SESSION_RESPONSE"
start_session_params = {
"Target": "instance-id",
}

# We test this by creating 4 credentials
# env vars, default profile (default),
# env aws profile (local_profile)
# and StartSession command profile (user_profile)
# We want to make sure only profile name in command
# be set to session manager plugin as parameter
self.environ['AWS_ACCESS_KEY_ID'] = 'env_var_akid'
self.environ['AWS_SECRET_ACCESS_KEY'] = 'env_var_sak'

with temporary_file('w') as f:
f.write(
'[default]\n'
'aws_access_key_id = shared_default_akid\n'
'aws_secret_access_key = shared_default_sak\n'
'[local_profile]\n'
'aws_access_key_id = shared_local_akid\n'
'aws_secret_access_key = shared_local_sak\n'
'[user_profile]\n'
'aws_access_key_id = shared_user_akid\n'
'aws_secret_access_key = shared_user_sak\n'
)
f.flush()

self.environ['AWS_SHARED_CREDENTIALS_FILE'] = f.name
self.environ["AWS_PROFILE"] = "local_profile"

expected_env = self.environ.copy()
expected_env.update({ssm_env_name: json.dumps(expected_response)})

self.driver = create_clidriver()
self.run_cmd(cmdline, expected_rc=0)

self.assertEqual(self.operations_called[0][0].name,
'StartSession')
self.assertEqual(self.operations_called[0][1],
{'Target': 'instance-id'})
mock_check_call.assert_called_once_with(
[
"session-manager-plugin",
ssm_env_name,
mock.ANY,
"StartSession",
"user_profile",
json.dumps(start_session_params),
mock.ANY,
],
env=expected_env,
)

@mock.patch("awscli.customizations.sessionmanager.check_call")
@mock.patch("awscli.customizations.sessionmanager.check_output")
def test_profile_environment_not_passed_to_sessionmanager_plugin(
self, mock_check_output, mock_check_call
):
cmdline = "ssm start-session --target instance-id"
mock_check_call.return_value = 0
mock_check_output.return_value = "1.2.500.0\n"
expected_response = {
"SessionId": "session-id",
"TokenValue": "token-value",
"StreamUrl": "stream-url",
}
self.parsed_responses = [expected_response]
ssm_env_name = "AWS_SSM_START_SESSION_RESPONSE"
start_session_params = {
"Target": "instance-id",
}

with temporary_file('w') as f:
f.write(
'[default]\n'
'aws_access_key_id = shared_default_akid\n'
'aws_secret_access_key = shared_default_sak\n'
'[local_profile]\n'
'aws_access_key_id = shared_local_akid\n'
'aws_secret_access_key = shared_local_sak\n'
)
f.flush()

self.environ.pop("AWS_ACCESS_KEY_ID", None)
self.environ.pop("AWS_SECRET_ACCESS_KEY", None)
self.environ['AWS_SHARED_CREDENTIALS_FILE'] = f.name
self.environ["AWS_PROFILE"] = "local_profile"

expected_env = self.environ.copy()
expected_env.update({ssm_env_name: json.dumps(expected_response)})

self.driver = create_clidriver()
self.run_cmd(cmdline, expected_rc=0)

self.assertEqual(self.operations_called[0][0].name,
'StartSession')
self.assertEqual(self.operations_called[0][1],
{'Target': 'instance-id'})
mock_check_call.assert_called_once_with(
[
"session-manager-plugin",
ssm_env_name,
mock.ANY,
"StartSession",
"",
json.dumps(start_session_params),
mock.ANY,
],
env=expected_env,
)

@mock.patch("awscli.customizations.sessionmanager.check_call")
@mock.patch("awscli.customizations.sessionmanager.check_output")
def test_default_profile_used_and_not_passed_to_sessionmanager_plugin(
self, mock_check_output, mock_check_call
):
cmdline = "ssm start-session --target instance-id"
mock_check_call.return_value = 0
mock_check_output.return_value = "1.2.500.0\n"
expected_response = {
"SessionId": "session-id",
"TokenValue": "token-value",
"StreamUrl": "stream-url",
}
self.parsed_responses = [expected_response]
ssm_env_name = "AWS_SSM_START_SESSION_RESPONSE"
start_session_params = {
"Target": "instance-id",
}

with temporary_file('w') as f:
f.write(
'[default]\n'
'aws_access_key_id = shared_default_akid\n'
'aws_secret_access_key = shared_default_sak\n'
)
f.flush()

self.environ.pop("AWS_ACCESS_KEY_ID", None)
self.environ.pop("AWS_SECRET_ACCESS_KEY", None)
self.environ.pop("AWS_SHARED_CREDENTIALS_FILE", None)
self.environ.pop("AWS_PROFILE", None)

expected_env = self.environ.copy()
expected_env.update({ssm_env_name: json.dumps(expected_response)})

self.driver = create_clidriver()
self.run_cmd(cmdline, expected_rc=0)

self.assertEqual(self.operations_called[0][0].name,
'StartSession')
self.assertEqual(self.operations_called[0][1],
{'Target': 'instance-id'})
mock_check_call.assert_called_once_with(
[
"session-manager-plugin",
ssm_env_name,
mock.ANY,
"StartSession",
"",
json.dumps(start_session_params),
mock.ANY,
],
env=expected_env,
)

def test_start_session_with_user_profile_not_exist(self):
cmdline = (
"ssm start-session --target instance-id "
"--profile user_profile"
)
with temporary_file('w') as f:
f.write(
'[default]\n'
'aws_access_key_id = shared_default_akid\n'
'aws_secret_access_key = shared_default_sak\n'
)
f.flush()

self.environ.pop("AWS_ACCESS_KEY_ID", None)
self.environ.pop("AWS_SECRET_ACCESS_KEY", None)
self.environ.pop("AWS_PROFILE", None)
self.environ['AWS_SHARED_CREDENTIALS_FILE'] = f.name

try:
self.driver = create_clidriver()
self.run_cmd(cmdline, expected_rc=255)
except ProfileNotFound as e:
self.assertIn(
'The config profile (user_profile) could not be found',
str(e)
)

@mock.patch('awscli.customizations.sessionmanager.check_call')
@mock.patch("awscli.customizations.sessionmanager.check_output")
def test_start_session_fails(self, mock_check_output, mock_check_call):
Expand Down
Loading

0 comments on commit 2ea405f

Please sign in to comment.