def __init__(self, role_to_assume):
self.role_to_assume = role_to_assume
boto3.setup_default_session(profile_name='default')
if role_to_assume:
self.client = boto3.client('sts')
response = self.client.assume_role(
RoleArn=role_to_assume, RoleSessionName="ecs_explorer")
creds = response['Credentials']
self.ecs = boto3.client('ecs',
aws_access_key_id=creds['AccessKeyId'],
aws_secret_access_key=creds['SecretAccessKey'],
aws_session_token=creds['SessionToken'])
self.ec2 = boto3.client('ec2',
aws_access_key_id=creds['AccessKeyId'],
aws_secret_access_key=creds['SecretAccessKey'],
aws_session_token=creds['SessionToken'])
else:
self.ecs = boto3.client('ecs')
self.ec2 = boto3.client('ec2')
python类setup_default_session()的实例源码
def run(self, command: str) -> None:
"""Execute one of the available commands.
Args:
command: Command in self.commands.
"""
boto3.setup_default_session(region_name=self._config.aws_region)
# Validate the configuration.
try:
if command not in {'compile_rules', 'configure', 'unit_test'}:
self._config.validate()
getattr(self, command)() # Command validation already happened in the ArgumentParser.
except InvalidConfigError as error:
sys.exit('ERROR: {}\nPlease run "python3 manage.py configure"'.format(error))
except TestFailureError as error:
sys.exit('TEST FAILED: {}'.format(error))
def get_service(hass, config):
"""Get the AWS SQS notification service."""
# pylint: disable=import-error
import boto3
aws_config = config.copy()
del aws_config[CONF_PLATFORM]
del aws_config[CONF_NAME]
profile = aws_config.get(CONF_PROFILE_NAME)
if profile is not None:
boto3.setup_default_session(profile_name=profile)
del aws_config[CONF_PROFILE_NAME]
sqs_client = boto3.client("sqs", **aws_config)
return AWSSQS(sqs_client)
def get_service(hass, config):
"""Get the AWS SNS notification service."""
# pylint: disable=import-error
import boto3
aws_config = config.copy()
del aws_config[CONF_PLATFORM]
del aws_config[CONF_NAME]
profile = aws_config.get(CONF_PROFILE_NAME)
if profile is not None:
boto3.setup_default_session(profile_name=profile)
del aws_config[CONF_PROFILE_NAME]
sns_client = boto3.client("sns", **aws_config)
return AWSSNS(sns_client)
def get_service(hass, config):
"""Get the AWS Lambda notification service."""
context_str = json.dumps({'hass': hass.config.as_dict(),
'custom': config[CONF_CONTEXT]})
context_b64 = base64.b64encode(context_str.encode("utf-8"))
context = context_b64.decode("utf-8")
# pylint: disable=import-error
import boto3
aws_config = config.copy()
del aws_config[CONF_PLATFORM]
del aws_config[CONF_NAME]
del aws_config[CONF_CONTEXT]
profile = aws_config.get(CONF_PROFILE_NAME)
if profile is not None:
boto3.setup_default_session(profile_name=profile)
del aws_config[CONF_PROFILE_NAME]
lambda_client = boto3.client("lambda", **aws_config)
return AWSLambda(lambda_client, context)
def clean_boto_configuration(monkeypatch):
monkeypatch.setenv('AWS_ACCESS_KEY_ID', 'AK000000000000000000')
monkeypatch.setenv('AWS_SECRET_ACCESS_KEY',
'0000000000000000000000000000000000000000')
monkeypatch.setenv('AWS_DEFAULT_REGION', 'us-east-1')
# Reset previously created default session (for credentials)
boto3.setup_default_session()
def __init__(self, debug=False, profile=None):
"""
Init
:param debug:
"""
self.debug = debug
self.profile = profile
if self.profile:
boto3.setup_default_session(profile_name=self.profile)
PrintMsg.debug = self.debug
self.client = boto3.client('iam')
def main():
artifacts_dir = "artifacts"
suite = unittest.TestSuite()
parser = argparse.ArgumentParser()
parser.add_argument("-c", "--config", type=str, required=True, help="Selects the config file")
parser.add_argument("-p", "--profile", type=str, help="Specifies the boto profile to choose from ~/.aws/config")
parser.add_argument("--report", default='html', help="Prints test execution on the console rather than generating a HTML report", choices=['text', 'html'])
args = parser.parse_args()
if args.profile :
boto3.setup_default_session(profile_name=args.profile)
testConfig = yaml.load(open(args.config, 'r'))
for testCategory, tests in testConfig.iteritems():
for test, enabled in tests.iteritems():
if enabled:
suite.addTest(eval(testCategory+"Audit")(test))
runner = ''
if args.report == 'text':
runner = unittest.TextTestRunner(verbosity=2)
elif args.report == 'html':
reportFile = open("test_results.html", "w")
runner = HTMLTestRunner.HTMLTestRunner(
stream=reportFile,
title='aws-security-test - Report',
verbosity=2
)
else:
print 'Invalid report type'
exit(1)
if not os.path.exists(artifacts_dir):
os.makedirs(artifacts_dir)
testExecution = runner.run(suite)
return len(testExecution.failures) + len(testExecution.errors)
def __init__(self, subject, email_from, email_to, credentials):
Email.__init__(self, subject, email_from, email_to)
boto3.setup_default_session(
aws_access_key_id=credentials['AWS_ID'],
aws_secret_access_key=credentials['AWS_PASS'],
region_name=credentials['region']
)
self.client = boto3.client('ses')
def __init__(self, type, service=None, region='us-west-2', profile='default'):
self.region = region
self.connection_type = type
self.service = service
self.client = None
self.resource = None
self.profile = profile
try:
boto3.setup_default_session(profile_name=self.profile)
except Exception as e:
logger.info("Problem setting default boto3 session: {}".format(e))
def pytest_configure():
boto3.setup_default_session(region_name='eu-west-1')
def main():
argparser = argparse.ArgumentParser()
argparser.add_argument('--site', help='Name of the Site VPC', required=True)
argparser.add_argument('--cidr', help='CIDR to add route for', required=True)
argparser.add_argument('--peeringid', help='Peering connection we are going to use', required=True)
argparser.add_argument('--profile', help='AWS Profile to use', required=True)
args = argparser.parse_args()
cidr = args.cidr
site = args.site
peeringid = args.peeringid
profile = args.profile
if profile != None: boto3.setup_default_session(profile_name=profile)
ec2resource = boto3.resource('ec2', region_name='eu-central-1')
ec2Client = boto3.client('ec2', region_name='eu-central-1')
vpcid = get_vpcid(site, client = ec2Client)
print('Looking up route tables for %s' % vpcid)
route_tables = get_route_tables(vpcid, ec2resource)
print ('Retrieved %s tables, proceeding to add tables' % len(route_tables))
for table_id in route_tables:
print('Adding route for %s to %s' % (cidr, peeringid))
try:
response = add_peering_route(table_id, cidr, peeringid, ec2resource)
if response:
continue
else:
print('Error occurred adding route to %s' % table_id)
except botocore.exceptions.ClientError as e:
if e.response['Error']['Code'] == 'RouteAlreadyExists':
print('Route already exists on %s, continuing' % table_id)
continue
else:
print('Unexpected error: %s' % e)
print('Routes have been added!')
def old(region, profile='default'):
project_details = json.load(open('project.json'))
boto3.setup_default_session(
profile_name=profile,
region_name=region
)
client = boto3.client('apigateway', region_name=region)
raml = ramlfications.parse('api_schema.raml')
api_name = raml.title
api_gateway = get_api_by_name(client, api_name)
if api_gateway is None:
api_gateway = client.create_rest_api(name=api_name)
aws_resources = client.get_resources(restApiId=api_gateway['id'])['items']
root = grab_root_resource(aws_resources)
resources = api.transform_resources(raml, raml.resources)
resources = associate_resources(aws_resources, resources)
aws_authorizers = client.get_authorizers(restApiId=api_gateway['id'])['items'] # NOQA
authorizers = associate_authorizers(aws_authorizers, raml.security_schemes or []) # NOQA
create_authorizer = functools.partial(
create_security_scheme,
client,
api_gateway['id'],
project_details['name']
)
authorizers = map(create_authorizer, authorizers)
for resource in resources:
print 'Creating Resource'
create_resource(
client,
api_gateway['id'],
root['id'],
resource,
project_details['name'],
authorizers
)
deployment = client.create_deployment(
restApiId=api_gateway['id'],
stageName=raml.base_uri
)
data = {
'deployment': deployment['id'],
'api': api_gateway['id'],
'uri': 'https://{}.execute-api.{}.amazonaws.com/{}/'.format(
api_gateway['id'],
region,
raml.base_uri
)
}
print data
