import json import logging import random import re import string import uuid from urllib.parse import urlparse, parse_qs from urllib.request import urlopen import requests from iact3.util import yaml, CustomSafeLoader from iact3.exceptions import Iact3Exception from iact3.plugin.ecs import EcsPlugin from iact3.plugin.oss import OssPlugin from iact3.plugin.ros import StackPlugin from iact3.plugin.vpc import VpcPlugin LOG = logging.getLogger(__name__) IAC_NAME = 'iact3' IAC_PACKAGE_NAME = 'alibabacloud-ros-iact3' class Selector: def __init__(self, key: str, original_value: any, allowed_values: list = None, parameters: dict = None): self.key = key self.original_value = original_value self.allowed_values = allowed_values or [] self.parameters = parameters self.current_value = allowed_values[0] if allowed_values else None self.next = None self.prev = None def refresh_parameters(self): self.parameters[self.key] = self.current_value class LinkedList: def __init__(self): self._head = None self._last = None def is_empty(self): return self._head is None def append(self, key, original_value, allowed_values=None, parameters=None): node = Selector(key, original_value, allowed_values, parameters) if self.is_empty(): self._head = node else: cur = self._last cur.next = node node.prev = cur self._last = node def first(self): return self._head def remove(self, key): if self.is_empty(): return cur = self.first() if cur.key == key: cur = cur.next if cur is None: self._head = None return cur.prev = None self._head = cur return while cur is not None: if cur.key == key: if cur.prev is not None: cur.prev.next = cur.next if cur.next is not None: cur.next.prev = cur.prev break cur = cur.next def __iter__(self): cur = self._head while cur is not None: value = cur.key cur = cur.next yield value def _error_message(key, value, msg): return f'Parsing pseudo parameter (Key: {key}, Value: {value}) error, {msg}' class ResolvedParameters: def __init__(self, name: str, region: str, parameters: dict, error=None): self.name = name self.region = region self.parameters = parameters self.error = error class ParamGenerator: RE_V_AUTO = re.compile(fr'\$\[{IAC_NAME}-auto]', re.I) RE_V_CURRENT_REGION = re.compile(fr'\$\[{IAC_NAME}-current[-_]region]', re.I) RE_K_ZONE_ID = re.compile(r'(\w*)zone(_|)id(_|)(\d*)', re.I) RE_K_VPC_ID = re.compile(r'(\w*)vpc(_|)id(_|)(\d*)', re.I) RE_K_VSW_ID = re.compile(r'(\w*)v(_|)switch(_|)id(_|)(\d*)', re.I) RE_K_SECURITY_GROUP = re.compile(r'(\w*)security(_|)group(_id|id)(_|)(\d*)', re.I) RE_K_COMMON_NAME = re.compile(r'(\w*)name(_|)(\d*)', re.I) RE_K_PASSWORD = re.compile(r'(\w*)password(_|)(\d*)', re.I) RE_K_UUID = re.compile(r'(\w*)uuid(_|)(\d*)', re.I) def __init__(self, config): self.config = config self.region = config.region self.parameters = config.parameters self.template_config = config.template_config self.parameters_order = config.parameters_order self.credential = config.auth.credential self.plugin = StackPlugin(region_id=self.region, credential=self.credential) self._vpc_id = None self._vsw_id = None self._not_support_keys = None self._linked_list: LinkedList = LinkedList() self._unresolved_parameters = {} @classmethod async def result(cls, config) -> ResolvedParameters: pg = cls(config) LOG.debug(f'start to generate parameters for {config.test_name}') try: await pg.resolve_auto_value() LOG.debug(f'resolve auto value result: {pg.parameters}') await pg.resolve_auto_key() resolved_parameters = ResolvedParameters(config.test_name, config.region, pg.parameters) LOG.debug( f'success generate parameters for {config.test_name}, parameters {resolved_parameters.parameters}') except Exception as ex: resolved_parameters = ResolvedParameters(config.test_name, config.region, pg.parameters, error=ex) LOG.debug(f'failed generate parameters for {config.test_name}, {ex}', exc_info=True) return resolved_parameters async def resolve_auto_key(self): for key, unresolved_value in self._unresolved_parameters.items(): if not isinstance(unresolved_value, str) or not self.RE_V_AUTO.fullmatch(unresolved_value): continue if self.RE_K_VSW_ID.fullmatch(key): if self._vsw_id is None: await self._gen_vpc_vsw_id(key, unresolved_value) self.parameters[key] = re.sub(self.RE_V_AUTO, self._vsw_id, unresolved_value) elif self.RE_K_VPC_ID.fullmatch(key): if self._vpc_id is None: await self._gen_vpc_vsw_id(key, unresolved_value) self.parameters[key] = re.sub(self.RE_V_AUTO, self._vpc_id, unresolved_value) elif self.RE_K_COMMON_NAME.fullmatch(key): value = self._gen_common_name() self.parameters[key] = re.sub(self.RE_V_AUTO, value, unresolved_value) elif self.RE_K_PASSWORD.fullmatch(key): value = self._gen_password() self.parameters[key] = re.sub(self.RE_V_AUTO, value, unresolved_value) elif self.RE_K_UUID.fullmatch(key): value = self._gen_uuid() self.parameters[key] = re.sub(self.RE_V_AUTO, value, unresolved_value) elif self.RE_K_SECURITY_GROUP.fullmatch(key): if self._vpc_id is None: self._vpc_id, self._vsw_id = await self._gen_vpc_vsw_id(key, unresolved_value) value = await self._gen_sg(key, unresolved_value) self.parameters[key] = re.sub(self.RE_V_AUTO, value, unresolved_value) return self.parameters async def resolve_auto_value(self): linked_list = LinkedList() resolved_parameters = {} parameters_order = self.parameters_order if not parameters_order: parameters_order = await self._get_parameters_order() or self.parameters.keys() for key in parameters_order: if key not in self.parameters: continue original_value = self.parameters[key] if not isinstance(original_value, str): resolved_parameters[key] = original_value continue if self.RE_V_AUTO.fullmatch(original_value): resolved_parameters[key] = None linked_list.append(key, original_value, parameters=resolved_parameters) elif self.RE_V_CURRENT_REGION.fullmatch(original_value): resolved_parameters[key] = self.region else: resolved_parameters[key] = original_value self._linked_list = linked_list first_selector = linked_list.first() if not first_selector: return resolved_parameters resolved_parameters = await self._select_value(first_selector) self.parameters.update(resolved_parameters) self.parameters.update(self._unresolved_parameters) return self.parameters async def _get_constraints(self, **kwargs): for _ in range(3): constraints = await self.plugin.get_parameter_constraints(**kwargs) behavior = constraints[0].get('Behavior') values = constraints[0].get('AllowedValues') reason = constraints[0].get('BehaviorReason') if behavior == 'QueryError' and reason and 'timeout' in reason: LOG.debug(f'get constraints timeout, {constraints}') continue if behavior == 'NotSupport': return return values else: return 'timeout' async def _select_value(self, selector: Selector, error_message=None) -> dict: key = selector.key parameters = selector.parameters allowed_values = selector.allowed_values current_value = selector.current_value error_msg = f'can not find any available value for {key} in {self.region} region ' \ f'in {allowed_values} for {self.config.test_name}' if allowed_values: next_selector = selector.next if not next_selector: return selector.parameters index = allowed_values.index(current_value) if index + 1 >= len(allowed_values): prev_selector = selector.prev if not prev_selector: raise Iact3Exception(error_message or error_msg) return await self._select_value(prev_selector, error_message=error_message) selector.current_value = allowed_values[index + 1] next_selector.parameters[key] = selector.current_value next_selector.allowed_values = [] return await self._select_value(next_selector, error_message=error_message) values = await self._get_constraints( parameters=parameters, **self.template_config.to_dict(), parameters_key_filter=[key], parameters_order=self.parameters_order ) if values is None: next_selector = selector.next self._unresolved_parameters[key] = selector.original_value self._linked_list.remove(key) if not next_selector: return selector.parameters return await self._select_value(next_selector, error_message=error_message) elif values == 'timeout': msg = f'get constraints timeout for {key} in {self.region} region for {self.config.test_name}' raise Iact3Exception(msg) elif not values: prev_selector = selector.prev if not prev_selector: param = json.dumps({k: v for k, v in parameters.items() if v is not None}) msg = (f'no available value found for {key} ' f'based on parameter {param} in {self.region} for {self.config.test_name}') raise Iact3Exception(msg) error_msg = f'no available value found for {key} in {self.region} region for {self.config.test_name}' return await self._select_value(prev_selector, error_message=error_msg) selector.allowed_values = values selector.current_value = values[0] selector.parameters[key] = selector.current_value next_selector = selector.next if not next_selector: return selector.parameters next_selector.parameters[key] = selector.current_value return await self._select_value(next_selector, error_message=error_message) async def _get_parameters_order(self): template = await self._get_template_body() if not template: raise Iact3Exception(f'failed to retrieve template by template config {self.template_config}') parsed_tpl = yaml.load(template, Loader=CustomSafeLoader) param_groups = parsed_tpl.get('Metadata', {}).get('ALIYUN::ROS::Interface', {}).get('ParameterGroups', []) if not param_groups: return params_in_metadata = [] for param_group in param_groups: params = param_group.get('Parameters', []) if param_group else None if not params: continue standard_params = [p for p in params if isinstance(p, str)] params_in_metadata += standard_params for key in self.parameters: if key not in params_in_metadata: params_in_metadata.append(key) self.parameters_order = params_in_metadata return params_in_metadata template_max_size = 524288 async def _get_template_body(self): template_body = self.template_config.template_body if template_body: return template_body template_id = self.template_config.template_id if template_id: try: template_info = await self.plugin.get_template( template_id=template_id, template_version=self.template_config.template_version ) return template_info['TemplateBody'] except Exception as ex: raise Iact3Exception(f'Failed to retrieve {template_id}: {ex}') template_url = self.template_config.template_url if template_url: components = urlparse(template_url) if components.scheme == 'oss': return self._get_template_from_oss(template_url, components) elif components.scheme == 'file': try: return urlopen(template_url).read() except Exception as ex: raise Iact3Exception(f'Failed to retrieve {template_url}: {ex}') else: try: resp = requests.get(template_url, timeout=10, stream=True) resp.raise_for_status() reader = resp.iter_content(chunk_size=1000) result = b'' max_size = self.template_max_size for chunk in reader: result += chunk if len(result) > max_size: raise Iact3Exception(f'template from {template_url}' f'exceeds maximum allowed size ({max_size} bytes)') return result except Exception as ex: raise Iact3Exception(f'Failed to retrieve {template_url}: {ex}') def _get_template_from_oss(self, template_url, components): bucket_name = components.netloc object_path = components.path.strip('/') if not bucket_name or not object_path: raise Iact3Exception(f'Invalid oss url {template_url}') region_id = self.region if components.query: t = parse_qs(components.query) region_ids = t.get('RegionId') region_id = region_ids[0] if region_ids else self.region oss_plugin = OssPlugin(region_id=region_id, bucket_name=bucket_name, credential=self.credential) try: object_meta = oss_plugin.get_object_meta(object_path) except Exception as ex: raise Iact3Exception(f'Oss failed: {ex}') if object_meta is None: raise Iact3Exception(f'Invalid oss url {template_url}') if object_meta.content_length > self.template_max_size: raise Iact3Exception(f'template from {template_url} exceeds maximum allowed size (524288 bytes)') try: r = oss_plugin.get_object_content(object_path) except Exception as ex: raise Iact3Exception(f'Oss failed: {ex}') if r is None: raise Iact3Exception(f'Invalid oss url {template_url}') return r.read() async def _gen_vpc_id(self, key, value): plugin = VpcPlugin(self.region, credential=self.credential) vpc = await plugin.get_one_vpc() if not vpc: msg = f'can not find any vpc in region {self.region}' raise Iact3Exception(_error_message(key, value, msg)) return vpc['VpcId'] async def _gen_vpc_vsw_id(self, key, value): zone_id = None for name, value in self.parameters.items(): if self.RE_K_ZONE_ID.fullmatch(name): zone_id = value break plugin = VpcPlugin(self.region, credential=self.credential) vsw = await plugin.get_one_vswitch(zone_id=zone_id) if not vsw: msg = f'can not find any vswitch in zone {zone_id}' raise Iact3Exception(_error_message(key, value, msg)) self._vpc_id = vsw['VpcId'] self._vsw_id = vsw['VSwitchId'] return self._vpc_id, self._vsw_id def _gen_common_name(self): return f'{IAC_NAME}-{uuid.uuid1().hex}'[:50] def _gen_password(self): special_chars = '!#$&{*:[=,]-_%@+' password_chars = [] for item in (string.ascii_lowercase, special_chars, string.digits, string.ascii_uppercase): password_chars.extend(random.sample(item, 4)) return ''.join(password_chars) def _gen_uuid(self): return str(uuid.uuid1()) async def _gen_sg(self, key, value): if self._vpc_id is None: await self._gen_vpc_vsw_id(key, value) plugin = EcsPlugin(region_id=self.region, credential=self.credential) sg = await plugin.get_security_group(vpc_id=self._vpc_id) if not sg: msg = f'can not find security group in vpc {self._vpc_id} in {self.region} region' raise Iact3Exception(_error_message(key, value, msg)) return sg['SecurityGroupId']