gerrit.onap Code Review - sdc/sdc-distribution-client.git/blob

   1 #    Licensed under the Apache License, Version 2.0 (the "License"); you may
   2 #    not use this file except in compliance with the License. You may obtain
   3 #    a copy of the License at
   4 #
   5 #         http://www.apache.org/licenses/LICENSE-2.0
   6 #
   7 #    Unless required by applicable law or agreed to in writing, software
   8 #    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
   9 #    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
  10 #    License for the specific language governing permissions and limitations
  11 #    under the License.
  12
  13 import os.path
  14 import requests
  15 import shutil
  16 import six
  17 import tempfile
  18 import yaml
  19 import zipfile
  20
  21 from toscaparser.common.exception import ExceptionCollector
  22 from toscaparser.common.exception import URLException
  23 from toscaparser.common.exception import ValidationError
  24 from toscaparser.imports import ImportsLoader
  25 from toscaparser.utils.gettextutils import _
  26 from toscaparser.utils.urlutils import UrlUtils
  27
  28 try:  # Python 2.x
  29     from BytesIO import BytesIO
  30 except ImportError:  # Python 3.x
  31     from io import BytesIO
  32
  33
  34 class CSAR(object):
  35
  36     def __init__(self, csar_file, a_file=True):
  37         self.path = csar_file
  38         self.a_file = a_file
  39         self.is_validated = False
  40         self.error_caught = False
  41         self.csar = None
  42         self.temp_dir = None
  43
  44     def validate(self):
  45         """Validate the provided CSAR file."""
  46
  47         self.is_validated = True
  48
  49         # validate that the file or URL exists
  50         missing_err_msg = (_('"%s" does not exist.') % self.path)
  51         if self.a_file:
  52             if not os.path.isfile(self.path):
  53                 ExceptionCollector.appendException(
  54                     ValidationError(message=missing_err_msg))
  55                 return False
  56             else:
  57                 self.csar = self.path
  58         else:  # a URL
  59             if not UrlUtils.validate_url(self.path):
  60                 ExceptionCollector.appendException(
  61                     ValidationError(message=missing_err_msg))
  62                 return False
  63             else:
  64                 response = requests.get(self.path)
  65                 self.csar = BytesIO(response.content)
  66
  67         # validate that it is a valid zip file
  68         if not zipfile.is_zipfile(self.csar):
  69             err_msg = (_('"%s" is not a valid zip file.') % self.path)
  70             ExceptionCollector.appendException(
  71                 ValidationError(message=err_msg))
  72             return False
  73
  74         # validate that it contains the metadata file in the correct location
  75         self.zfile = zipfile.ZipFile(self.csar, 'r')
  76         filelist = self.zfile.namelist()
  77         if 'TOSCA-Metadata/TOSCA.meta' not in filelist:
  78             err_msg = (_('"%s" is not a valid CSAR as it does not contain the '
  79                          'required file "TOSCA.meta" in the folder '
  80                          '"TOSCA-Metadata".') % self.path)
  81             ExceptionCollector.appendException(
  82                 ValidationError(message=err_msg))
  83             return False
  84
  85         # validate that 'Entry-Definitions' property exists in TOSCA.meta
  86         data = self.zfile.read('TOSCA-Metadata/TOSCA.meta')
  87         invalid_yaml_err_msg = (_('The file "TOSCA-Metadata/TOSCA.meta" in '
  88                                   'the CSAR "%s" does not contain valid YAML '
  89                                   'content.') % self.path)
  90         try:
  91             meta = yaml.load(data)
  92             if type(meta) is dict:
  93                 self.metadata = meta
  94             else:
  95                 ExceptionCollector.appendException(
  96                     ValidationError(message=invalid_yaml_err_msg))
  97                 return False
  98         except yaml.YAMLError:
  99             ExceptionCollector.appendException(
 100                 ValidationError(message=invalid_yaml_err_msg))
 101             return False
 102
 103         if 'Entry-Definitions' not in self.metadata:
 104             err_msg = (_('The CSAR "%s" is missing the required metadata '
 105                          '"Entry-Definitions" in '
 106                          '"TOSCA-Metadata/TOSCA.meta".')
 107                        % self.path)
 108             ExceptionCollector.appendException(
 109                 ValidationError(message=err_msg))
 110             return False
 111
 112         # validate that 'Entry-Definitions' metadata value points to an
 113         # existing file in the CSAR
 114         entry = self.metadata.get('Entry-Definitions')
 115         if entry and entry not in filelist:
 116             err_msg = (_('The "Entry-Definitions" file defined in the '
 117                          'CSAR "%s" does not exist.') % self.path)
 118             ExceptionCollector.appendException(
 119                 ValidationError(message=err_msg))
 120             return False
 121
 122         # validate that external references in the main template actually
 123         # exist and are accessible
 124         self._validate_external_references()
 125         return not self.error_caught
 126
 127     def get_metadata(self):
 128         """Return the metadata dictionary."""
 129
 130         # validate the csar if not already validated
 131         if not self.is_validated:
 132             self.validate()
 133
 134         # return a copy to avoid changes overwrite the original
 135         return dict(self.metadata) if self.metadata else None
 136
 137     def _get_metadata(self, key):
 138         if not self.is_validated:
 139             self.validate()
 140         return self.metadata.get(key)
 141
 142     def get_author(self):
 143         return self._get_metadata('Created-By')
 144
 145     def get_version(self):
 146         return self._get_metadata('CSAR-Version')
 147
 148     def get_main_template(self):
 149         entry_def = self._get_metadata('Entry-Definitions')
 150         if entry_def in self.zfile.namelist():
 151             return entry_def
 152
 153     def get_main_template_yaml(self):
 154         main_template = self.get_main_template()
 155         if main_template:
 156             data = self.zfile.read(main_template)
 157             invalid_tosca_yaml_err_msg = (
 158                 _('The file "%(template)s" in the CSAR "%(csar)s" does not '
 159                   'contain valid TOSCA YAML content.') %
 160                 {'template': main_template, 'csar': self.path})
 161             try:
 162                 tosca_yaml = yaml.load(data)
 163                 if type(tosca_yaml) is not dict:
 164                     ExceptionCollector.appendException(
 165                         ValidationError(message=invalid_tosca_yaml_err_msg))
 166                 return tosca_yaml
 167             except Exception:
 168                 ExceptionCollector.appendException(
 169                     ValidationError(message=invalid_tosca_yaml_err_msg))
 170
 171     def get_description(self):
 172         desc = self._get_metadata('Description')
 173         if desc is not None:
 174             return desc
 175
 176         self.metadata['Description'] = \
 177             self.get_main_template_yaml().get('description')
 178         return self.metadata['Description']
 179
 180     def decompress(self):
 181         if not self.is_validated:
 182             self.validate()
 183         # Although in python this works well - jython insists on getting a directory here, not a file:
 184         #self.temp_dir = tempfile.NamedTemporaryFile().name
 185         self.temp_dir = tempfile.mkdtemp()
 186
 187         with zipfile.ZipFile(self.csar, "r") as zf:
 188             zf.extractall(self.temp_dir)
 189
 190     def _validate_external_references(self):
 191         """Extracts files referenced in the main template
 192
 193         These references are currently supported:
 194         * imports
 195         * interface implementations
 196         * artifacts
 197         """
 198         try:
 199             self.decompress()
 200             main_tpl_file = self.get_main_template()
 201             if not main_tpl_file:
 202                 return
 203             main_tpl = self.get_main_template_yaml()
 204
 205             if 'imports' in main_tpl:
 206                 ImportsLoader(main_tpl['imports'],
 207                               os.path.join(self.temp_dir, main_tpl_file))
 208
 209             if 'topology_template' in main_tpl:
 210                 topology_template = main_tpl['topology_template']
 211
 212                 if 'node_templates' in topology_template:
 213                     node_templates = topology_template['node_templates']
 214
 215                     for node_template_key in node_templates:
 216                         node_template = node_templates[node_template_key]
 217                         if 'artifacts' in node_template:
 218                             artifacts = node_template['artifacts']
 219                             for artifact_key in artifacts:
 220                                 artifact = artifacts[artifact_key]
 221                                 if isinstance(artifact, six.string_types):
 222                                     self._validate_external_reference(
 223                                         main_tpl_file,
 224                                         artifact)
 225                                 elif isinstance(artifact, dict):
 226                                     if 'file' in artifact:
 227                                         self._validate_external_reference(
 228                                             main_tpl_file,
 229                                             artifact['file'])
 230                                 else:
 231                                     ExceptionCollector.appendException(
 232                                         ValueError(_('Unexpected artifact '
 233                                                      'definition for "%s".')
 234                                                    % artifact_key))
 235                                     self.error_caught = True
 236                         if 'interfaces' in node_template:
 237                             interfaces = node_template['interfaces']
 238                             for interface_key in interfaces:
 239                                 interface = interfaces[interface_key]
 240                                 for opertation_key in interface:
 241                                     operation = interface[opertation_key]
 242                                     if isinstance(operation, six.string_types):
 243                                         self._validate_external_reference(
 244                                             main_tpl_file,
 245                                             operation,
 246                                             False)
 247                                     elif isinstance(operation, dict):
 248                                         if 'implementation' in operation:
 249                                             self._validate_external_reference(
 250                                                 main_tpl_file,
 251                                                 operation['implementation'])
 252         finally:
 253             if self.temp_dir:
 254                 shutil.rmtree(self.temp_dir)
 255
 256     def _validate_external_reference(self, tpl_file, resource_file,
 257                                      raise_exc=True):
 258         """Verify that the external resource exists
 259
 260         If resource_file is a URL verify that the URL is valid.
 261         If resource_file is a relative path verify that the path is valid
 262         considering base folder (self.temp_dir) and tpl_file.
 263         Note that in a CSAR resource_file cannot be an absolute path.
 264         """
 265         if UrlUtils.validate_url(resource_file):
 266             msg = (_('The resource at "%s" cannot be accessed.') %
 267                    resource_file)
 268             try:
 269                 if UrlUtils.url_accessible(resource_file):
 270                     return
 271                 else:
 272                     ExceptionCollector.appendException(
 273                         URLException(what=msg))
 274                     self.error_caught = True
 275             except Exception:
 276                 ExceptionCollector.appendException(
 277                     URLException(what=msg))
 278                 self.error_caught = True
 279
 280         if os.path.isfile(os.path.join(self.temp_dir,
 281                                        os.path.dirname(tpl_file),
 282                                        resource_file)):
 283             return
 284
 285         if raise_exc:
 286             ExceptionCollector.appendException(
 287                 ValueError(_('The resource "%s" does not exist.')
 288                            % resource_file))
 289             self.error_caught = True