import json import os import time from subprocess import run from tempfile import NamedTemporaryFile import requests import jsonref from alibabacloud_ros20190910 import models as ros_models from alibabacloud_ros20190910.client import Client as RosClient from Tea.exceptions import TeaException from tools import exceptions class RosResource: def __init__(self, client: RosClient, resource_type: str): self.resource_type = resource_type self.client = client self._rt_info = None def fetch(self): req = ros_models.GetResourceTypeRequest(self.resource_type) resp = None for i in range(4): try: resp = self.client.get_resource_type(req) break except TeaException as e: if e.code != 'Throttling.User' or i == 3: raise e time.sleep(10) if resp: self._rt_info = resp.body @property def rt_info(self) -> ros_models.GetResourceTypeResponseBody: if not self._rt_info: self.fetch() return self._rt_info def properties(self): return self.rt_info.properties def attributes(self): return self.rt_info.attributes class TerraformResource: ALICLOUD_SRC_URL = "https://raw.githubusercontent.com/aliyun/terraform-provider-alicloud/master/alicloud/resource_{}.go" AST_FUNC_RETURN = { "NodeType": "StarExpr", "X": { "NodeType": "SelectorExpr", "X": {"NodeType": "Ident", "Name": "schema"}, "Sel": {"NodeType": "Ident", "Name": "Resource"}, }, } AST_RESOURCE_TYPE = { "NodeType": "SelectorExpr", "X": {"NodeType": "Ident", "Name": "schema"}, "Sel": {"NodeType": "Ident", "Name": "Resource"}, } AST_SCHEMA_TYPE = { "NodeType": "StarExpr", "X": { "NodeType": "SelectorExpr", "X": {"NodeType": "Ident", "Name": "schema"}, "Sel": {"NodeType": "Ident", "Name": "Schema"}, }, } AST_D_SET_FUNC = { "NodeType": "SelectorExpr", "X": {"NodeType": "Ident", "Name": "d"}, "Sel": {"NodeType": "Ident", "Name": "Set"}, } TYPE_MAPPING = { "TypeString": "string", "TypeBool": "boolean", "TypeSet": "list", "TypeList": "list", "TypeInt": "integer", "TypeFloat": "number", "TypeMap": "map", } KEYS_USING_DEFAULT_TAGS_SCHEMA = ("tags", "volume_tags", "template_tags") DEFAULT_TAGS_SCHEMA = { "Handler": "tags_dict_to_list" # 'Type': 'list', # 'Schema': { # '*': { # "Type": "map", # "Required": False, # "Schema": { # "Value": { # "Type": "string", # "Required": False, # }, # "Key": { # "Type": "string", # "Required": True, # } # }, # } # } } def __init__(self, resource_type: str, resource_filename: str = None): self.resource_type = resource_type self.resource_filename = resource_filename or resource_type self._ast = None def path_or_url(self): alicloud_local = os.getenv("TERRAFORM_PROVIDER_ALICLOUD") if alicloud_local: path = os.path.join( alicloud_local, "alicloud", f"resource_{self.resource_filename}.go" ) return path, "path" url = self.ALICLOUD_SRC_URL.format(self.resource_filename) return url, "url" def fetch(self): path_or_url, t = self.path_or_url() if t == "path": with open(path_or_url) as f: code = f.read().encode("utf-8") else: resp = requests.get(path_or_url) if resp.status_code != 200: raise exceptions.RosToolWarning( message=f"Cannot fetch {path_or_url}. Reason: {resp.text}" ) code = resp.text.encode("utf-8") with NamedTemporaryFile() as f: p = run("asty go2json", input=code, shell=True, stdout=f) if p.returncode != 0: raise exceptions.RosToolWarning( message=f'Run `{p.args}` failed. Reason: {p.stderr.decode("utf-8")}' ) f.seek(0) self._ast = json.loads(f.read()) @property def ast(self): if not self._ast: self.fetch() return self._ast def properties(self): ast_resource = self._find_ast_resource(self.ast) ast_schema = self._find_ast_schema(ast_resource) properties = self._get_properties_schema(ast_schema) return properties def attributes(self): decls = self.ast.get("Decls") if not decls: return attributes = {} for decl in decls: if decl["NodeType"] != "FuncDecl": continue decl_type = decl["Type"] if decl_type["NodeType"] != "FuncType": continue body = decl["Body"] if body["NodeType"] != "BlockStmt": continue self._fetch_attributes(body["List"], attributes) return attributes @classmethod def _fetch_attributes(cls, block, attributes): if isinstance(block, dict): x = block.get("X") if x and x.get("Fun") == cls.AST_D_SET_FUNC: args = x["Args"] if args and args[0]["Kind"] == "STRING": attr_name = args[0]["Value"].strip('"') attributes[attr_name] = {} else: for val in block.values(): cls._fetch_attributes(val, attributes) elif isinstance(block, list): for val in block: cls._fetch_attributes(val, attributes) @classmethod def _find_ast_resource(cls, ast): decls = ast.get("Decls") if not decls: return for decl in decls: if decl["NodeType"] != "FuncDecl": continue decl_type = decl["Type"] if decl_type["NodeType"] != "FuncType": continue if decl_type["Results"]["NodeType"] != "FieldList": continue for result in decl_type["Results"]["List"]: if result["Type"] != cls.AST_FUNC_RETURN: continue for block in decl["Body"]["List"]: if block["NodeType"] != "ReturnStmt": continue for result in block["Results"]: x = result.get("X") if x and x["Type"] == cls.AST_RESOURCE_TYPE: return x @classmethod def _find_ast_schema(cls, ast_resource): if not ast_resource: return for elt in ast_resource["Elts"]: if elt["NodeType"] != "KeyValueExpr": continue elt_value = elt["Value"] if ( elt_value["NodeType"] == "CompositeLit" and elt_value["Type"]["Value"] == cls.AST_SCHEMA_TYPE ): return elt_value def _get_properties_schema(self, ast_schema, prop_path=""): if not ast_schema: return {} properties_schema = {} for elt in ast_schema["Elts"]: prop_name = eval(elt["Key"]["Value"]) new_prop_path = f"{prop_path}.{prop_name}" if prop_path else prop_name try: prop = self._get_property_schema(elt["Value"], new_prop_path) except ValueError as e: if prop_name in self.KEYS_USING_DEFAULT_TAGS_SCHEMA: prop = self.DEFAULT_TAGS_SCHEMA else: path_or_url, t = self.path_or_url() raise ValueError(f"{e}. Path: {path_or_url}") properties_schema[prop_name] = prop return properties_schema def _get_property_schema(self, ast_prop, path): if ast_prop["NodeType"] == "CallExpr": path_or_url, t = self.path_or_url() raise ValueError( f'Found function call in schema "{path}", not supported. Path: {path_or_url}' ) prop = {"Required": True} for elt in ast_prop["Elts"]: elt_key_name = elt["Key"]["Name"] elt_value = elt["Value"] if elt_key_name == "Type": prop["Type"] = self.TYPE_MAPPING[elt_value["Sel"]["Name"]] elif elt_key_name == "Optional": prop["Required"] = elt_value["Name"] == '"True"' elif elt_key_name == "Elem": node_type = elt_value["NodeType"] if node_type == "UnaryExpr": if elt_value["Op"] == "&": x = elt_value["X"] x_name = x["Type"]["Sel"]["Name"] new_prop_path = f"{path}.*" if x_name == "Schema": sub_prop = self._get_property_schema(x, new_prop_path) prop["Schema"] = {"*": sub_prop} elif x_name == "Resource": sub_props = self._get_properties_schema( x["Elts"][0]["Value"], new_prop_path ) prop["Schema"] = { "*": { "Required": False, "Type": "map", "Schema": sub_props, } } return prop class CloudFormationResource: def __init__(self, client, resource_type: str): self.client = client self.resource_type = resource_type self._rt_info = None self._schema = None def fetch(self): resp = self.client.describe_type( Type="RESOURCE", TypeName=self.resource_type, ) self._rt_info = resp @property def rt_info(self) -> dict: if not self._rt_info: self.fetch() return self._rt_info @property def schema(self) -> dict: if not self._schema: s = self.rt_info["Schema"] # remove resource-schema.json due to can not resolve s = s.replace("resource-schema.json#", "#") self._schema = jsonref.loads(s, proxies=False, lazy_load=False) self._schema["type"] = "object" self._handle_prop(self._schema) return self._schema def properties(self): return self._get_props(read_only=False) def attributes(self): return self._get_props(read_only=True) def _get_props(self, read_only=False): properties = self.schema["Properties"] read_only_properties = self.schema.get("ReadOnlyProperties") if read_only_properties: read_only_names = set( n.replace("/properties/", "") for n in read_only_properties ) else: read_only_names = set() if read_only: props = {n: properties[n] for n in properties if n in read_only_names} else: props = {n: properties[n] for n in properties if n not in read_only_names} return props def _handle_props(self, props: dict, required_names: list = None): for name, prop in props.items(): if required_names and name in required_names: prop["Required"] = True else: prop["Required"] = False self._handle_prop(prop) def _handle_prop(self, prop: dict): for k in list(prop): if k == "required" and prop.get("type") == "object": nk = ".RequiredList" else: nk = "{}{}".format(k[0].upper(), k[1:]) prop[nk] = prop.pop(k) t = prop.get("Type") if isinstance(t, list): if "object" in t: t = "object" elif "array" in t: t = "array" else: t = t[-1] prop["Type"] = t if t == "array": prop["Type"] = "list" item_schema = prop.get("Items") self._handle_prop(item_schema) prop["Schema"] = {"*": item_schema} elif t == "object": prop["Type"] = "map" props = prop.get("Properties") if props: required_names = prop.pop(".RequiredList", None) self._handle_props(props, required_names) prop["Schema"] = props elif not t and "oneof" in prop: for each_prop in prop["oneof"]: self._handle_prop(each_prop)