Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
# we don't want to include the port number in the signature
hostname = split.netloc.split(':')[0]
canonical_request = '{0}\n{1}\n\nhost:{2}\n\nhost\n'.format(
request.method,
split.path,
hostname)
logger.debug("Calculating signature using v4 auth.")
logger.debug('CanonicalRequest:\n%s', canonical_request)
string_to_sign = signer.string_to_sign(request, canonical_request)
logger.debug('StringToSign:\n%s', string_to_sign)
signature = signer.signature(string_to_sign, request)
logger.debug('Signature:\n%s', signature)
return '{0}Z{1}'.format(request.context['timestamp'], signature)
class CodeCommitCommand(BasicCommand):
NAME = 'credential-helper'
SYNOPSIS = ('aws codecommit credential-helper')
EXAMPLES = ''
SUBCOMMANDS = [
{'name': 'get', 'command_class': CodeCommitGetCommand},
{'name': 'store', 'command_class': CodeCommitNoOpStoreCommand},
{'name': 'erase', 'command_class': CodeCommitNoOpEraseCommand},
]
DESCRIPTION = ('Provide a SigV4 compatible user name and'
' password for git smart HTTP '
' These commands are consumed by git and'
' should not used directly. Erase and Store'
' are no-ops. Get is operation to generate'
' credentials to authenticate AWS CodeCommit.'
' Run \"aws codecommit credential-helper help\"'
This is a wrapper to make the plugin built-in to the cli as opposed
to specifiying it in the configuration file.
"""
awscli_initialize(event_handlers)
def add_s3(command_table, session, **kwargs):
"""
This creates a new service object for the s3 plugin. It sends the
old s3 commands to the namespace ``s3api``.
"""
utils.rename_command(command_table, 's3', 's3api')
command_table['s3'] = S3(session)
class S3(BasicCommand):
NAME = 's3'
DESCRIPTION = BasicCommand.FROM_FILE('s3/_concepts.rst')
SYNOPSIS = "aws s3 [ ...]"
SUBCOMMANDS = [
{'name': 'ls', 'command_class': ListCommand},
{'name': 'website', 'command_class': WebsiteCommand},
{'name': 'cp', 'command_class': CpCommand},
{'name': 'mv', 'command_class': MvCommand},
{'name': 'rm', 'command_class': RmCommand},
{'name': 'sync', 'command_class': SyncCommand},
{'name': 'mb', 'command_class': MbCommand},
{'name': 'rb', 'command_class': RbCommand},
{'name': 'presign', 'command_class': PresignCommand},
]
def _run_main(self, parsed_args, parsed_globals):
import sys
from datetime import datetime, timezone
from urllib import parse
from awscli.customizations.commands import BasicCommand
from awscli.customizations.utils import create_client_from_parsed_globals
from awscli.customizations.s3.subcommands import PAGE_SIZE
def awscli_initialize(cli):
cli.register('building-command-table.s3', _inject_touch)
def _inject_touch(command_table, session, **kwargs):
command_table['touch'] = S3Touch(session)
class S3Touch(BasicCommand):
"""Toch objects in S3 and trigger events"""
NAME = 'touch'
DESCRIPTION = BasicCommand.FROM_FILE('s3touch_description.rst', root_module=sys.modules[__name__])
EXAMPLES = BasicCommand.FROM_FILE('s3touch.rst', root_module=sys.modules[__name__])
ARG_TABLE = [
{
'name': 'bucket',
'help_text': 'Name of the bucket to list.',
'positional_arg': True,
},
{
'name': 'prefix',
'required': False,
def initialize(cli):
"""
The entry point for the credential helper
"""
cli.register('building-command-table.codecommit', inject_commands)
def inject_commands(command_table, session, **kwargs):
"""
Injects new commands into the codecommit subcommand.
"""
command_table['credential-helper'] = CodeCommitCommand(session)
class CodeCommitNoOpStoreCommand(BasicCommand):
NAME = 'store'
DESCRIPTION = ('This operation does nothing, credentials'
' are calculated each time')
SYNOPSIS = ('aws codecommit credential-helper store')
EXAMPLES = ''
_UNDOCUMENTED = True
def _run_main(self, args, parsed_globals):
return 0
class CodeCommitNoOpEraseCommand(BasicCommand):
NAME = 'erase'
DESCRIPTION = ('This operation does nothing, no credentials'
' are ever stored')
SYNOPSIS = ('aws codecommit credential-helper erase')
# or in the "license" file accompanying this file. This file is
# distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
# ANY KIND, either express or implied. See the License for the specific
# language governing permissions and limitations under the License.
import sys
import logging
from awscli.customizations.commands import BasicCommand
from awscli.compat import six
from . import PREDEFINED_SECTION_NAMES
LOG = logging.getLogger(__name__)
class ConfigureGetCommand(BasicCommand):
NAME = 'get'
DESCRIPTION = BasicCommand.FROM_FILE('configure', 'get',
'_description.rst')
SYNOPSIS = 'aws configure get varname [--profile profile-name]'
EXAMPLES = BasicCommand.FROM_FILE('configure', 'get', '_examples.rst')
ARG_TABLE = [
{'name': 'varname',
'help_text': 'The name of the config value to retrieve.',
'action': 'store',
'cli_type_name': 'string', 'positional_arg': True},
]
def __init__(self, session, stream=sys.stdout, error_stream=sys.stderr):
super(ConfigureGetCommand, self).__init__(session)
self._stream = stream
self._error_stream = error_stream
def inject_commands(command_table, session: Session, **kwargs):
"""
Used to inject top-level commands in the awscli command list.
"""
command_table['login'] = Login(session)
command_table['logout'] = Logout(session)
def inject_subcommands(command_table, session: Session, **kwargs):
"""
Used to inject subcommands into the aws login command list.
"""
command_table['configure'] = Configure(session)
class Login(BasicCommand):
NAME = 'login'
DESCRIPTION = ('is a plugin that manages retrieving and rotating'
' Amazon STS keys using the Shibboleth IdP and Duo'
' for authentication.')
SYNOPSIS = ('aws login [ ...]')
ARG_TABLE = [
{'name': 'entity-id', 'help_text': 'Entity ID of the IdP'},
{
'name': 'ecp-endpoint-url',
'help_text': 'ECP endpoint URL of the IdP'
},
{'name': 'username', 'help_text': 'Username to use on login to IdP'},
{'name': 'password', 'help_text': 'Password to use on login to IdP'},
{
'name': 'role-arn',
'name': 'verbose',
'action': 'count',
'default': 0,
'cli_type_name': 'integer',
'help_text': 'Display verbose output'
},
]
UPDATE = False
def _run_main(self, args: Namespace, parsed_globals):
logout(args, self._session)
return 0
class Configure(BasicCommand):
NAME = 'login'
DESCRIPTION = ('''
Configure LOGIN options. If this command is run with no arguments,
you will be prompted for configuration values such as your IdP's
entity ID and its ECP endpoint URL. You can configure a named
profile using the --profile argument. If your config file does not
exist (the default location is ~/.aws-login/config), it will be
created for you. To keep an existing value, hit enter when prompted
for the value. When you are prompted for information, the current
value will be dis- played in [brackets]. If the config item has
no value, it be displayed as [None].
=======================
Configuration Variables
=======================
{'name': 'security-configuration',
'help_text': helptext.SECURITY_CONFIG},
{'name': 'custom-ami-id',
'help_text' : helptext.CUSTOM_AMI_ID},
{'name': 'ebs-root-volume-size',
'help_text' : helptext.EBS_ROOT_VOLUME_SIZE},
{'name': 'repo-upgrade-on-boot',
'help_text' : helptext.REPO_UPGRADE_ON_BOOT},
{'name': 'kerberos-attributes',
'schema': argumentschema.KERBEROS_ATTRIBUTES_SCHEMA,
'help_text': helptext.KERBEROS_ATTRIBUTES},
{'name': 'step-concurrency-level',
'cli_type_name': 'integer',
'help_text': helptext.STEP_CONCURRENCY_LEVEL}
]
SYNOPSIS = BasicCommand.FROM_FILE('emr', 'create-cluster-synopsis.txt')
EXAMPLES = BasicCommand.FROM_FILE('emr', 'create-cluster-examples.rst')
def _run_main_command(self, parsed_args, parsed_globals):
params = {}
params['Name'] = parsed_args.name
self._validate_release_label_ami_version(parsed_args)
service_role_validation_message = (
" Either choose --use-default-roles or use both --service-role "
" and --ec2-attributes InstanceProfile=.")
if parsed_args.use_default_roles is True and \
parsed_args.service_role is not None:
raise exceptions.MutualExclusiveOptionError(
option1="--use-default-roles",
MSG_NO_EXECUTE_CHANGESET = \
("Changeset created successfully. Run the following command to "
"review changes:"
"\n"
"aws cloudformation describe-change-set --change-set-name "
"{changeset_id}"
"\n")
MSG_EXECUTE_SUCCESS = "Successfully created/updated stack - {stack_name}\n"
PARAMETER_OVERRIDE_CMD = "parameter-overrides"
TAGS_CMD = "tags"
NAME = 'deploy'
DESCRIPTION = BasicCommand.FROM_FILE("cloudformation",
"_deploy_description.rst")
ARG_TABLE = [
{
'name': 'template-file',
'required': True,
'help_text': (
'The path where your AWS CloudFormation'
' template is located.'
)
},
{
'name': 'stack-name',
'action': 'store',
'required': True,
'help_text': (
import sys
import json
from botocore.client import Config
from awscli.customizations.cloudformation.artifact_exporter import Template
from awscli.customizations.cloudformation.yamlhelper import yaml_dump
from awscli.customizations.cloudformation import exceptions
from awscli.customizations.commands import BasicCommand
from awscli.customizations.s3uploader import S3Uploader
LOG = logging.getLogger(__name__)
class PackageCommand(BasicCommand):
MSG_PACKAGED_TEMPLATE_WRITTEN = (
"Successfully packaged artifacts and wrote output template "
"to file {output_file_name}."
"\n"
"Execute the following command to deploy the packaged template"
"\n"
"aws cloudformation deploy --template-file {output_file_path} "
"--stack-name "
"\n")
NAME = "package"
DESCRIPTION = BasicCommand.FROM_FILE("cloudformation",
"_package_description.rst")