# -*- 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