mirror of
https://github.com/ansible-collections/community.general.git
synced 2024-09-14 20:13:21 +02:00
775be1d3f3
* multiple modules: removed unused imports * fixed ali_instance(_info) import check * add changelog fragment * Update changelogs/fragments/5240-unused-imports.yaml Co-authored-by: Felix Fontein <felix@fontein.de> Co-authored-by: Felix Fontein <felix@fontein.de>
292 lines
12 KiB
Python
292 lines
12 KiB
Python
# -*- coding: utf-8 -*-
|
|
# This code is part of Ansible, but is an independent component.
|
|
# This particular file snippet, and this file snippet only, is BSD licensed.
|
|
# Modules you write using this snippet, which is embedded dynamically by Ansible
|
|
# still belong to the author of the module, and may assign their own license
|
|
# to the complete work.
|
|
#
|
|
# Copyright (c) 2017-present Alibaba Group Holding Limited. He Guimin <heguimin36@163.com>
|
|
#
|
|
# Simplified BSD License (see LICENSES/BSD-2-Clause.txt or https://opensource.org/licenses/BSD-2-Clause)
|
|
# SPDX-License-Identifier: BSD-2-Clause
|
|
|
|
from __future__ import (absolute_import, division, print_function)
|
|
__metaclass__ = type
|
|
|
|
import os
|
|
import json
|
|
import traceback
|
|
from ansible.module_utils.basic import env_fallback
|
|
|
|
try:
|
|
import footmark
|
|
import footmark.ecs
|
|
import footmark.slb
|
|
import footmark.vpc
|
|
import footmark.rds
|
|
import footmark.ess
|
|
import footmark.sts
|
|
import footmark.dns
|
|
import footmark.ram
|
|
import footmark.market
|
|
|
|
FOOTMARK_IMP_ERR = None
|
|
HAS_FOOTMARK = True
|
|
except ImportError:
|
|
FOOTMARK_IMP_ERR = traceback.format_exc()
|
|
HAS_FOOTMARK = False
|
|
|
|
|
|
class AnsibleACSError(Exception):
|
|
pass
|
|
|
|
|
|
def acs_common_argument_spec():
|
|
return dict(
|
|
alicloud_access_key=dict(aliases=['access_key_id', 'access_key'], no_log=True,
|
|
fallback=(env_fallback, ['ALICLOUD_ACCESS_KEY', 'ALICLOUD_ACCESS_KEY_ID'])),
|
|
alicloud_secret_key=dict(aliases=['secret_access_key', 'secret_key'], no_log=True,
|
|
fallback=(env_fallback, ['ALICLOUD_SECRET_KEY', 'ALICLOUD_SECRET_ACCESS_KEY'])),
|
|
alicloud_security_token=dict(aliases=['security_token'], no_log=True,
|
|
fallback=(env_fallback, ['ALICLOUD_SECURITY_TOKEN'])),
|
|
ecs_role_name=dict(aliases=['role_name'], fallback=(env_fallback, ['ALICLOUD_ECS_ROLE_NAME']))
|
|
)
|
|
|
|
|
|
def ecs_argument_spec():
|
|
spec = acs_common_argument_spec()
|
|
spec.update(
|
|
dict(
|
|
alicloud_region=dict(required=True, aliases=['region', 'region_id'],
|
|
fallback=(env_fallback, ['ALICLOUD_REGION', 'ALICLOUD_REGION_ID'])),
|
|
alicloud_assume_role_arn=dict(fallback=(env_fallback, ['ALICLOUD_ASSUME_ROLE_ARN']),
|
|
aliases=['assume_role_arn']),
|
|
alicloud_assume_role_session_name=dict(fallback=(env_fallback, ['ALICLOUD_ASSUME_ROLE_SESSION_NAME']),
|
|
aliases=['assume_role_session_name']),
|
|
alicloud_assume_role_session_expiration=dict(type='int',
|
|
fallback=(env_fallback,
|
|
['ALICLOUD_ASSUME_ROLE_SESSION_EXPIRATION']),
|
|
aliases=['assume_role_session_expiration']),
|
|
alicloud_assume_role=dict(type='dict', aliases=['assume_role']),
|
|
profile=dict(fallback=(env_fallback, ['ALICLOUD_PROFILE'])),
|
|
shared_credentials_file=dict(fallback=(env_fallback, ['ALICLOUD_SHARED_CREDENTIALS_FILE']))
|
|
)
|
|
)
|
|
return spec
|
|
|
|
|
|
def get_acs_connection_info(params):
|
|
|
|
ecs_params = dict(acs_access_key_id=params.get('alicloud_access_key'),
|
|
acs_secret_access_key=params.get('alicloud_secret_key'),
|
|
security_token=params.get('alicloud_security_token'),
|
|
ecs_role_name=params.get('ecs_role_name'),
|
|
user_agent='Ansible-Provider-Alicloud')
|
|
return ecs_params
|
|
|
|
|
|
def connect_to_acs(acs_module, region, **params):
|
|
conn = acs_module.connect_to_region(region, **params)
|
|
if not conn:
|
|
if region not in [acs_module_region.id for acs_module_region in acs_module.regions()]:
|
|
raise AnsibleACSError(
|
|
"Region %s does not seem to be available for acs module %s." % (region, acs_module.__name__))
|
|
else:
|
|
raise AnsibleACSError(
|
|
"Unknown problem connecting to region %s for acs module %s." % (region, acs_module.__name__))
|
|
return conn
|
|
|
|
|
|
def get_assume_role(params):
|
|
""" Return new params """
|
|
sts_params = get_acs_connection_info(params)
|
|
assume_role = {}
|
|
if params.get('assume_role'):
|
|
assume_role['alicloud_assume_role_arn'] = params['assume_role'].get('role_arn')
|
|
assume_role['alicloud_assume_role_session_name'] = params['assume_role'].get('session_name')
|
|
assume_role['alicloud_assume_role_session_expiration'] = params['assume_role'].get('session_expiration')
|
|
assume_role['alicloud_assume_role_policy'] = params['assume_role'].get('policy')
|
|
|
|
assume_role_params = {
|
|
'role_arn': params.get('alicloud_assume_role_arn') if params.get('alicloud_assume_role_arn') else assume_role.get('alicloud_assume_role_arn'),
|
|
'role_session_name': params.get('alicloud_assume_role_session_name') if params.get('alicloud_assume_role_session_name')
|
|
else assume_role.get('alicloud_assume_role_session_name'),
|
|
'duration_seconds': params.get('alicloud_assume_role_session_expiration') if params.get('alicloud_assume_role_session_expiration')
|
|
else assume_role.get('alicloud_assume_role_session_expiration', 3600),
|
|
'policy': assume_role.get('alicloud_assume_role_policy', {})
|
|
}
|
|
|
|
try:
|
|
sts = connect_to_acs(footmark.sts, params.get('alicloud_region'), **sts_params).assume_role(**assume_role_params).read()
|
|
sts_params['acs_access_key_id'], sts_params['acs_secret_access_key'], sts_params['security_token'] \
|
|
= sts['access_key_id'], sts['access_key_secret'], sts['security_token']
|
|
except AnsibleACSError as e:
|
|
params.fail_json(msg=str(e))
|
|
return sts_params
|
|
|
|
|
|
def get_profile(params):
|
|
if not params['alicloud_access_key'] and not params['ecs_role_name'] and params['profile']:
|
|
path = params['shared_credentials_file'] if params['shared_credentials_file'] else os.getenv('HOME') + '/.aliyun/config.json'
|
|
auth = {}
|
|
with open(path, 'r') as f:
|
|
for pro in json.load(f)['profiles']:
|
|
if params['profile'] == pro['name']:
|
|
auth = pro
|
|
if auth:
|
|
if auth['mode'] == 'AK' and auth.get('access_key_id') and auth.get('access_key_secret'):
|
|
params['alicloud_access_key'] = auth.get('access_key_id')
|
|
params['alicloud_secret_key'] = auth.get('access_key_secret')
|
|
params['alicloud_region'] = auth.get('region_id')
|
|
params = get_acs_connection_info(params)
|
|
elif auth['mode'] == 'StsToken' and auth.get('access_key_id') and auth.get('access_key_secret') and auth.get('sts_token'):
|
|
params['alicloud_access_key'] = auth.get('access_key_id')
|
|
params['alicloud_secret_key'] = auth.get('access_key_secret')
|
|
params['security_token'] = auth.get('sts_token')
|
|
params['alicloud_region'] = auth.get('region_id')
|
|
params = get_acs_connection_info(params)
|
|
elif auth['mode'] == 'EcsRamRole':
|
|
params['ecs_role_name'] = auth.get('ram_role_name')
|
|
params['alicloud_region'] = auth.get('region_id')
|
|
params = get_acs_connection_info(params)
|
|
elif auth['mode'] == 'RamRoleArn' and auth.get('ram_role_arn'):
|
|
params['alicloud_access_key'] = auth.get('access_key_id')
|
|
params['alicloud_secret_key'] = auth.get('access_key_secret')
|
|
params['security_token'] = auth.get('sts_token')
|
|
params['ecs_role_name'] = auth.get('ram_role_name')
|
|
params['alicloud_assume_role_arn'] = auth.get('ram_role_arn')
|
|
params['alicloud_assume_role_session_name'] = auth.get('ram_session_name')
|
|
params['alicloud_assume_role_session_expiration'] = auth.get('expired_seconds')
|
|
params['alicloud_region'] = auth.get('region_id')
|
|
params = get_assume_role(params)
|
|
elif params.get('alicloud_assume_role_arn') or params.get('assume_role'):
|
|
params = get_assume_role(params)
|
|
else:
|
|
params = get_acs_connection_info(params)
|
|
return params
|
|
|
|
|
|
def ecs_connect(module):
|
|
""" Return an ecs connection"""
|
|
ecs_params = get_profile(module.params)
|
|
# If we have a region specified, connect to its endpoint.
|
|
region = module.params.get('alicloud_region')
|
|
if region:
|
|
try:
|
|
ecs = connect_to_acs(footmark.ecs, region, **ecs_params)
|
|
except AnsibleACSError as e:
|
|
module.fail_json(msg=str(e))
|
|
# Otherwise, no region so we fallback to the old connection method
|
|
return ecs
|
|
|
|
|
|
def slb_connect(module):
|
|
""" Return an slb connection"""
|
|
slb_params = get_profile(module.params)
|
|
# If we have a region specified, connect to its endpoint.
|
|
region = module.params.get('alicloud_region')
|
|
if region:
|
|
try:
|
|
slb = connect_to_acs(footmark.slb, region, **slb_params)
|
|
except AnsibleACSError as e:
|
|
module.fail_json(msg=str(e))
|
|
# Otherwise, no region so we fallback to the old connection method
|
|
return slb
|
|
|
|
|
|
def dns_connect(module):
|
|
""" Return an dns connection"""
|
|
dns_params = get_profile(module.params)
|
|
# If we have a region specified, connect to its endpoint.
|
|
region = module.params.get('alicloud_region')
|
|
if region:
|
|
try:
|
|
dns = connect_to_acs(footmark.dns, region, **dns_params)
|
|
except AnsibleACSError as e:
|
|
module.fail_json(msg=str(e))
|
|
# Otherwise, no region so we fallback to the old connection method
|
|
return dns
|
|
|
|
|
|
def vpc_connect(module):
|
|
""" Return an vpc connection"""
|
|
vpc_params = get_profile(module.params)
|
|
# If we have a region specified, connect to its endpoint.
|
|
region = module.params.get('alicloud_region')
|
|
if region:
|
|
try:
|
|
vpc = connect_to_acs(footmark.vpc, region, **vpc_params)
|
|
except AnsibleACSError as e:
|
|
module.fail_json(msg=str(e))
|
|
# Otherwise, no region so we fallback to the old connection method
|
|
return vpc
|
|
|
|
|
|
def rds_connect(module):
|
|
""" Return an rds connection"""
|
|
rds_params = get_profile(module.params)
|
|
# If we have a region specified, connect to its endpoint.
|
|
region = module.params.get('alicloud_region')
|
|
if region:
|
|
try:
|
|
rds = connect_to_acs(footmark.rds, region, **rds_params)
|
|
except AnsibleACSError as e:
|
|
module.fail_json(msg=str(e))
|
|
# Otherwise, no region so we fallback to the old connection method
|
|
return rds
|
|
|
|
|
|
def ess_connect(module):
|
|
""" Return an ess connection"""
|
|
ess_params = get_profile(module.params)
|
|
# If we have a region specified, connect to its endpoint.
|
|
region = module.params.get('alicloud_region')
|
|
if region:
|
|
try:
|
|
ess = connect_to_acs(footmark.ess, region, **ess_params)
|
|
except AnsibleACSError as e:
|
|
module.fail_json(msg=str(e))
|
|
# Otherwise, no region so we fallback to the old connection method
|
|
return ess
|
|
|
|
|
|
def sts_connect(module):
|
|
""" Return an sts connection"""
|
|
sts_params = get_profile(module.params)
|
|
# If we have a region specified, connect to its endpoint.
|
|
region = module.params.get('alicloud_region')
|
|
if region:
|
|
try:
|
|
sts = connect_to_acs(footmark.sts, region, **sts_params)
|
|
except AnsibleACSError as e:
|
|
module.fail_json(msg=str(e))
|
|
# Otherwise, no region so we fallback to the old connection method
|
|
return sts
|
|
|
|
|
|
def ram_connect(module):
|
|
""" Return an ram connection"""
|
|
ram_params = get_profile(module.params)
|
|
# If we have a region specified, connect to its endpoint.
|
|
region = module.params.get('alicloud_region')
|
|
if region:
|
|
try:
|
|
ram = connect_to_acs(footmark.ram, region, **ram_params)
|
|
except AnsibleACSError as e:
|
|
module.fail_json(msg=str(e))
|
|
# Otherwise, no region so we fallback to the old connection method
|
|
return ram
|
|
|
|
|
|
def market_connect(module):
|
|
""" Return an market connection"""
|
|
market_params = get_profile(module.params)
|
|
# If we have a region specified, connect to its endpoint.
|
|
region = module.params.get('alicloud_region')
|
|
if region:
|
|
try:
|
|
market = connect_to_acs(footmark.market, region, **market_params)
|
|
except AnsibleACSError as e:
|
|
module.fail_json(msg=str(e))
|
|
# Otherwise, no region so we fallback to the old connection method
|
|
return market
|