# Licensed under the Apache License, Version 2.0 (the "License"); you may # not use this file except in compliance with the License. You may obtain # a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the # License for the specific language governing permissions and limitations # under the License. import os.path import requests import shutil import six import tempfile import yaml import zipfile from toscaparser.common.exception import ExceptionCollector from toscaparser.common.exception import URLException from toscaparser.common.exception import ValidationError from toscaparser.imports import ImportsLoader from toscaparser.utils.gettextutils import _ from toscaparser.utils.urlutils import UrlUtils try: # Python 2.x from BytesIO import BytesIO except ImportError: # Python 3.x from io import BytesIO class CSAR(object): def __init__(self, csar_file, a_file=True): self.path = csar_file self.a_file = a_file self.is_validated = False self.error_caught = False self.csar = None self.temp_dir = None def validate(self): """Validate the provided CSAR file.""" self.is_validated = True # validate that the file or URL exists missing_err_msg = (_('"%s" does not exist.') % self.path) if self.a_file: if not os.path.isfile(self.path): ExceptionCollector.appendException( ValidationError(message=missing_err_msg)) return False else: self.csar = self.path else: # a URL if not UrlUtils.validate_url(self.path): ExceptionCollector.appendException( ValidationError(message=missing_err_msg)) return False else: response = requests.get(self.path) self.csar = BytesIO(response.content) # validate that it is a valid zip file if not zipfile.is_zipfile(self.csar): err_msg = (_('"%s" is not a valid zip file.') % self.path) ExceptionCollector.appendException( ValidationError(message=err_msg)) return False # validate that it contains the metadata file in the correct location self.zfile = zipfile.ZipFile(self.csar, 'r') filelist = self.zfile.namelist() if 'TOSCA-Metadata/TOSCA.meta' not in filelist: err_msg = (_('"%s" is not a valid CSAR as it does not contain the ' 'required file "TOSCA.meta" in the folder ' '"TOSCA-Metadata".') % self.path) ExceptionCollector.appendException( ValidationError(message=err_msg)) return False # validate that 'Entry-Definitions' property exists in TOSCA.meta data = self.zfile.read('TOSCA-Metadata/TOSCA.meta') invalid_yaml_err_msg = (_('The file "TOSCA-Metadata/TOSCA.meta" in ' 'the CSAR "%s" does not contain valid YAML ' 'content.') % self.path) try: meta = yaml.load(data) if type(meta) is dict: self.metadata = meta else: ExceptionCollector.appendException( ValidationError(message=invalid_yaml_err_msg)) return False except yaml.YAMLError: ExceptionCollector.appendException( ValidationError(message=invalid_yaml_err_msg)) return False if 'Entry-Definitions' not in self.metadata: err_msg = (_('The CSAR "%s" is missing the required metadata ' '"Entry-Definitions" in ' '"TOSCA-Metadata/TOSCA.meta".') % self.path) ExceptionCollector.appendException( ValidationError(message=err_msg)) return False # validate that 'Entry-Definitions' metadata value points to an # existing file in the CSAR entry = self.metadata.get('Entry-Definitions') if entry and entry not in filelist: err_msg = (_('The "Entry-Definitions" file defined in the ' 'CSAR "%s" does not exist.') % self.path) ExceptionCollector.appendException( ValidationError(message=err_msg)) return False # validate that external references in the main template actually # exist and are accessible self._validate_external_references() return not self.error_caught def get_metadata(self): """Return the metadata dictionary.""" # validate the csar if not already validated if not self.is_validated: self.validate() # return a copy to avoid changes overwrite the original return dict(self.metadata) if self.metadata else None def _get_metadata(self, key): if not self.is_validated: self.validate() return self.metadata.get(key) def get_author(self): return self._get_metadata('Created-By') def get_version(self): return self._get_metadata('CSAR-Version') def get_main_template(self): entry_def = self._get_metadata('Entry-Definitions') if entry_def in self.zfile.namelist(): return entry_def def get_main_template_yaml(self): main_template = self.get_main_template() if main_template: data = self.zfile.read(main_template) invalid_tosca_yaml_err_msg = ( _('The file "%(template)s" in the CSAR "%(csar)s" does not ' 'contain valid TOSCA YAML content.') % {'template': main_template, 'csar': self.path}) try: tosca_yaml = yaml.load(data) if type(tosca_yaml) is not dict: ExceptionCollector.appendException( ValidationError(message=invalid_tosca_yaml_err_msg)) return tosca_yaml except Exception: ExceptionCollector.appendException( ValidationError(message=invalid_tosca_yaml_err_msg)) def get_description(self): desc = self._get_metadata('Description') if desc is not None: return desc self.metadata['Description'] = \ self.get_main_template_yaml().get('description') return self.metadata['Description'] def decompress(self): if not self.is_validated: self.validate() self.temp_dir = tempfile.NamedTemporaryFile().name with zipfile.ZipFile(self.csar, "r") as zf: zf.extractall(self.temp_dir) def _validate_external_references(self): """Extracts files referenced in the main template These references are currently supported: * imports * interface implementations * artifacts """ try: self.decompress() main_tpl_file = self.get_main_template() if not main_tpl_file: return main_tpl = self.get_main_template_yaml() if 'imports' in main_tpl: ImportsLoader(main_tpl['imports'], os.path.join(self.temp_dir, main_tpl_file)) if 'topology_template' in main_tpl: topology_template = main_tpl['topology_template'] if 'node_templates' in topology_template: node_templates = topology_template['node_templates'] for node_template_key in node_templates: node_template = node_templates[node_template_key] if 'artifacts' in node_template: artifacts = node_template['artifacts'] for artifact_key in artifacts: artifact = artifacts[artifact_key] if isinstance(artifact, six.string_types): self._validate_external_reference( main_tpl_file, artifact) elif isinstance(artifact, dict): if 'file' in artifact: self._validate_external_reference( main_tpl_file, artifact['file']) else: ExceptionCollector.appendException( ValueError(_('Unexpected artifact ' 'definition for "%s".') % artifact_key)) self.error_caught = True if 'interfaces' in node_template: interfaces = node_template['interfaces'] for interface_key in interfaces: interface = interfaces[interface_key] for opertation_key in interface: operation = interface[opertation_key] if isinstance(operation, six.string_types): self._validate_external_reference( main_tpl_file, operation, False) elif isinstance(operation, dict): if 'implementation' in operation: self._validate_external_reference( main_tpl_file, operation['implementation']) finally: if self.temp_dir: shutil.rmtree(self.temp_dir) def _validate_external_reference(self, tpl_file, resource_file, raise_exc=True): """Verify that the external resource exists If resource_file is a URL verify that the URL is valid. If resource_file is a relative path verify that the path is valid considering base folder (self.temp_dir) and tpl_file. Note that in a CSAR resource_file cannot be an absolute path. """ if UrlUtils.validate_url(resource_file): msg = (_('The resource at "%s" cannot be accessed.') % resource_file) try: if UrlUtils.url_accessible(resource_file): return else: ExceptionCollector.appendException( URLException(what=msg)) self.error_caught = True except Exception: ExceptionCollector.appendException( URLException(what=msg)) self.error_caught = True if os.path.isfile(os.path.join(self.temp_dir, os.path.dirname(tpl_file), resource_file)): return if raise_exc: ExceptionCollector.appendException( ValueError(_('The resource "%s" does not exist.') % resource_file)) self.error_caught = True