path: root/nfvparser/toscaparser/prereq/csar.py

#    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