osm_common/sol004_package.py - osm/common - Gitiles

 # -*- coding: utf-8 -*-

 # Copyright 2020 Whitestack, LLC
 # *************************************************************
 #
 # This file is part of OSM common repository.
 # All Rights Reserved to Whitestack, LLC
 #
 # 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.
 #
 # For those usages not covered by the Apache License, Version 2.0 please
 # contact: agarcia@whitestack.com
 ##

 """Python module for interacting with ETSI GS NFV-SOL004 compliant packages

 This module provides a SOL004Package class for validating and interacting with
 ETSI SOL004 packages. A valid SOL004 package may have its files arranged according
 to one of the following two structures:

 SOL004 with metadata directory    SOL004 without metadata directory

 native_charm_vnf/                 native_charm_vnf/
 ├── TOSCA-Metadata                ├── native_charm_vnfd.mf
 │   └── TOSCA.meta                ├── native_charm_vnfd.yaml
 ├── manifest.mf                   ├── ChangeLog.txt
 ├── Definitions                   ├── Licenses
 │   └── native_charm_vnfd.yaml    │   └── license.lic
 ├── Files                         ├── Files
 │   ├── icons                     │   └── icons
 │   │   └── osm.png               │       └── osm.png
 │   ├── Licenses                  └── Scripts
 │   │   └── license.lic               ├── cloud_init
 │   └── changelog.txt                 │   └── cloud-config.txt
 └── Scripts                           └── charms
     ├── cloud_init                        └── simple
     │   └── cloud-config.txt                  ├── config.yaml
     └── charms                                ├── hooks
         └── simple                            │   ├── install
             ├── config.yaml                  ...
             ├── hooks                         │
             │   ├── install                   └── src
            ...                                    └── charm.py
             └── src
                 └── charm.py
 """

 import yaml
 import os
 import hashlib


 _METADATA_FILE_PATH = "TOSCA-Metadata/TOSCA.meta"
 _METADATA_DESCRIPTOR_FIELD = "Entry-Definitions"
 _METADATA_MANIFEST_FIELD = "ETSI-Entry-Manifest"
 _METADATA_CHANGELOG_FIELD = "ETSI-Entry-Change-Log"
 _METADATA_LICENSES_FIELD = "ETSI-Entry-Licenses"
 _METADATA_DEFAULT_CHANGELOG_PATH = "ChangeLog.txt"
 _METADATA_DEFAULT_LICENSES_PATH = "Licenses"
 _MANIFEST_FILE_PATH_FIELD = "Source"
 _MANIFEST_FILE_HASH_ALGORITHM_FIELD = "Algorithm"
 _MANIFEST_FILE_HASH_DIGEST_FIELD = "Hash"


 class SOL004PackageException(Exception):
     pass


 class SOL004Package:
     def __init__(self, package_path=""):
         self._package_path = package_path
         self._package_metadata = self._parse_package_metadata()
         self._manifest_data = self._parse_manifest_data()

     def _parse_package_metadata(self):
         try:
             return self._parse_package_metadata_with_metadata_dir()
         except FileNotFoundError:
             return self._parse_package_metadata_without_metadata_dir()

     def _parse_package_metadata_with_metadata_dir(self):
         try:
             return self._parse_file_in_blocks(_METADATA_FILE_PATH)
         except FileNotFoundError as e:
             raise e
         except (Exception, OSError) as e:
             raise SOL004PackageException(
                 "Error parsing {}: {}".format(_METADATA_FILE_PATH, e)
             )

     def _parse_package_metadata_without_metadata_dir(self):
         package_root_files = {f for f in os.listdir(self._package_path)}
         package_root_yamls = [
             f for f in package_root_files if f.endswith(".yml") or f.endswith(".yaml")
         ]
         if len(package_root_yamls) != 1:
             error_msg = "Error parsing package metadata: there should be exactly 1 descriptor YAML, found {}"
             raise SOL004PackageException(error_msg.format(len(package_root_yamls)))
         # TODO: Parse extra metadata from descriptor YAML?
         return [
             {
                 _METADATA_DESCRIPTOR_FIELD: package_root_yamls[0],
                 _METADATA_MANIFEST_FIELD: "{}.mf".format(
                     os.path.splitext(package_root_yamls[0])[0]
                 ),
                 _METADATA_CHANGELOG_FIELD: _METADATA_DEFAULT_CHANGELOG_PATH,
                 _METADATA_LICENSES_FIELD: _METADATA_DEFAULT_LICENSES_PATH,
             }
         ]

     def _parse_manifest_data(self):
         manifest_path = None
         for tosca_meta in self._package_metadata:
             if _METADATA_MANIFEST_FIELD in tosca_meta:
                 manifest_path = tosca_meta[_METADATA_MANIFEST_FIELD]
                 break
         else:
             error_msg = "Error parsing {}: no {} field on path".format(
                 _METADATA_FILE_PATH, _METADATA_MANIFEST_FIELD
             )
             raise SOL004PackageException(error_msg)

         try:
             return self._parse_file_in_blocks(manifest_path)
         except (Exception, OSError) as e:
             raise SOL004PackageException(
                 "Error parsing {}: {}".format(manifest_path, e)
             )

     def _get_package_file_full_path(self, file_relative_path):
         return os.path.join(self._package_path, file_relative_path)

     def _parse_file_in_blocks(self, file_relative_path):
         file_path = self._get_package_file_full_path(file_relative_path)
         with open(file_path) as f:
             blocks = f.read().split("\n\n")
         parsed_blocks = map(yaml.safe_load, blocks)
         return [block for block in parsed_blocks if block is not None]

     def _get_package_file_manifest_data(self, file_relative_path):
         for file_data in self._manifest_data:
             if file_data.get(_MANIFEST_FILE_PATH_FIELD, "") == file_relative_path:
                 return file_data

         error_msg = (
             "Error parsing {} manifest data: file not found on manifest file".format(
                 file_relative_path
             )
         )
         raise SOL004PackageException(error_msg)

     def get_package_file_hash_digest_from_manifest(self, file_relative_path):
         """Returns the hash digest of a file inside this package as specified on the manifest file."""
         file_manifest_data = self._get_package_file_manifest_data(file_relative_path)
         try:
             return file_manifest_data[_MANIFEST_FILE_HASH_DIGEST_FIELD]
         except Exception as e:
             raise SOL004PackageException(
                 "Error parsing {} hash digest: {}".format(file_relative_path, e)
             )

     def get_package_file_hash_algorithm_from_manifest(self, file_relative_path):
         """Returns the hash algorithm of a file inside this package as specified on the manifest file."""
         file_manifest_data = self._get_package_file_manifest_data(file_relative_path)
         try:
             return file_manifest_data[_MANIFEST_FILE_HASH_ALGORITHM_FIELD]
         except Exception as e:
             raise SOL004PackageException(
                 "Error parsing {} hash digest: {}".format(file_relative_path, e)
             )

     @staticmethod
     def _get_hash_function_from_hash_algorithm(hash_algorithm):
         function_to_algorithm = {"SHA-256": hashlib.sha256, "SHA-512": hashlib.sha512}
         if hash_algorithm not in function_to_algorithm:
             error_msg = (
                 "Error checking hash function: hash algorithm {} not supported".format(
                     hash_algorithm
                 )
             )
             raise SOL004PackageException(error_msg)
         return function_to_algorithm[hash_algorithm]

     def _calculate_file_hash(self, file_relative_path, hash_algorithm):
         file_path = self._get_package_file_full_path(file_relative_path)
         hash_function = self._get_hash_function_from_hash_algorithm(hash_algorithm)
         try:
             with open(file_path, "rb") as f:
                 return hash_function(f.read()).hexdigest()
         except Exception as e:
             raise SOL004PackageException(
                 "Error hashing {}: {}".format(file_relative_path, e)
             )

     def validate_package_file_hash(self, file_relative_path):
         """Validates the integrity of a file using the hash algorithm and digest on the package manifest."""
         hash_algorithm = self.get_package_file_hash_algorithm_from_manifest(
             file_relative_path
         )
         file_hash = self._calculate_file_hash(file_relative_path, hash_algorithm)
         expected_file_hash = self.get_package_file_hash_digest_from_manifest(
             file_relative_path
         )
         if file_hash != expected_file_hash:
             error_msg = "Error validating {} hash: calculated hash {} is different than manifest hash {}"
             raise SOL004PackageException(
                 error_msg.format(file_relative_path, file_hash, expected_file_hash)
             )

     def validate_package_hashes(self):
         """Validates the integrity of all files listed on the package manifest."""
         for file_data in self._manifest_data:
             if _MANIFEST_FILE_PATH_FIELD in file_data:
                 file_relative_path = file_data[_MANIFEST_FILE_PATH_FIELD]
                 self.validate_package_file_hash(file_relative_path)

     def get_descriptor_location(self):
         """Returns this package descriptor location as a relative path from the package root."""
         for tosca_meta in self._package_metadata:
             if _METADATA_DESCRIPTOR_FIELD in tosca_meta:
                 return tosca_meta[_METADATA_DESCRIPTOR_FIELD]

         error_msg = "Error: no {} entry found on {}".format(
             _METADATA_DESCRIPTOR_FIELD, _METADATA_FILE_PATH
         )
         raise SOL004PackageException(error_msg)
	# -- coding: utf-8 --

	# Copyright 2020 Whitestack, LLC
	# *************************************************************
	#
	# This file is part of OSM common repository.
	# All Rights Reserved to Whitestack, LLC
	#
	# 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.
	#
	# For those usages not covered by the Apache License, Version 2.0 please
	# contact: agarcia@whitestack.com
	##

	"""Python module for interacting with ETSI GS NFV-SOL004 compliant packages

	This module provides a SOL004Package class for validating and interacting with
	ETSI SOL004 packages. A valid SOL004 package may have its files arranged according
	to one of the following two structures:

	SOL004 with metadata directory SOL004 without metadata directory

	native_charm_vnf/ native_charm_vnf/
	├── TOSCA-Metadata ├── native_charm_vnfd.mf
	│ └── TOSCA.meta ├── native_charm_vnfd.yaml
	├── manifest.mf ├── ChangeLog.txt
	├── Definitions ├── Licenses
	│ └── native_charm_vnfd.yaml │ └── license.lic
	├── Files ├── Files
	│ ├── icons │ └── icons
	│ │ └── osm.png │ └── osm.png
	│ ├── Licenses └── Scripts
	│ │ └── license.lic ├── cloud_init
	│ └── changelog.txt │ └── cloud-config.txt
	└── Scripts └── charms
	├── cloud_init └── simple
	│ └── cloud-config.txt ├── config.yaml
	└── charms ├── hooks
	└── simple │ ├── install
	├── config.yaml ...
	├── hooks │
	│ ├── install └── src
	... └── charm.py
	└── src
	└── charm.py
	"""

	import yaml
	import os
	import hashlib


	_METADATA_FILE_PATH = "TOSCA-Metadata/TOSCA.meta"
	_METADATA_DESCRIPTOR_FIELD = "Entry-Definitions"
	_METADATA_MANIFEST_FIELD = "ETSI-Entry-Manifest"
	_METADATA_CHANGELOG_FIELD = "ETSI-Entry-Change-Log"
	_METADATA_LICENSES_FIELD = "ETSI-Entry-Licenses"
	_METADATA_DEFAULT_CHANGELOG_PATH = "ChangeLog.txt"
	_METADATA_DEFAULT_LICENSES_PATH = "Licenses"
	_MANIFEST_FILE_PATH_FIELD = "Source"
	_MANIFEST_FILE_HASH_ALGORITHM_FIELD = "Algorithm"
	_MANIFEST_FILE_HASH_DIGEST_FIELD = "Hash"


	class SOL004PackageException(Exception):
	pass


	class SOL004Package:
	def __init__(self, package_path=""):
	self._package_path = package_path
	self._package_metadata = self._parse_package_metadata()
	self._manifest_data = self._parse_manifest_data()

	def _parse_package_metadata(self):
	try:
	return self._parse_package_metadata_with_metadata_dir()
	except FileNotFoundError:
	return self._parse_package_metadata_without_metadata_dir()

	def _parse_package_metadata_with_metadata_dir(self):
	try:
	return self._parse_file_in_blocks(_METADATA_FILE_PATH)
	except FileNotFoundError as e:
	raise e
	except (Exception, OSError) as e:
	raise SOL004PackageException(
	"Error parsing {}: {}".format(_METADATA_FILE_PATH, e)
	)

	def _parse_package_metadata_without_metadata_dir(self):
	package_root_files = {f for f in os.listdir(self._package_path)}
	package_root_yamls = [
	f for f in package_root_files if f.endswith(".yml") or f.endswith(".yaml")
	]
	if len(package_root_yamls) != 1:
	error_msg = "Error parsing package metadata: there should be exactly 1 descriptor YAML, found {}"
	raise SOL004PackageException(error_msg.format(len(package_root_yamls)))
	# TODO: Parse extra metadata from descriptor YAML?
	return [
	{
	_METADATA_DESCRIPTOR_FIELD: package_root_yamls[0],
	_METADATA_MANIFEST_FIELD: "{}.mf".format(
	os.path.splitext(package_root_yamls[0])[0]
	),
	_METADATA_CHANGELOG_FIELD: _METADATA_DEFAULT_CHANGELOG_PATH,
	_METADATA_LICENSES_FIELD: _METADATA_DEFAULT_LICENSES_PATH,
	}
	]

	def _parse_manifest_data(self):
	manifest_path = None
	for tosca_meta in self._package_metadata:
	if _METADATA_MANIFEST_FIELD in tosca_meta:
	manifest_path = tosca_meta[_METADATA_MANIFEST_FIELD]
	break
	else:
	error_msg = "Error parsing {}: no {} field on path".format(
	_METADATA_FILE_PATH, _METADATA_MANIFEST_FIELD
	)
	raise SOL004PackageException(error_msg)

	try:
	return self._parse_file_in_blocks(manifest_path)
	except (Exception, OSError) as e:
	raise SOL004PackageException(
	"Error parsing {}: {}".format(manifest_path, e)
	)

	def _get_package_file_full_path(self, file_relative_path):
	return os.path.join(self._package_path, file_relative_path)

	def _parse_file_in_blocks(self, file_relative_path):
	file_path = self._get_package_file_full_path(file_relative_path)
	with open(file_path) as f:
	blocks = f.read().split("\n\n")
	parsed_blocks = map(yaml.safe_load, blocks)
	return [block for block in parsed_blocks if block is not None]

	def _get_package_file_manifest_data(self, file_relative_path):
	for file_data in self._manifest_data:
	if file_data.get(_MANIFEST_FILE_PATH_FIELD, "") == file_relative_path:
	return file_data

	error_msg = (
	"Error parsing {} manifest data: file not found on manifest file".format(
	file_relative_path
	)
	)
	raise SOL004PackageException(error_msg)

	def get_package_file_hash_digest_from_manifest(self, file_relative_path):
	"""Returns the hash digest of a file inside this package as specified on the manifest file."""
	file_manifest_data = self._get_package_file_manifest_data(file_relative_path)
	try:
	return file_manifest_data[_MANIFEST_FILE_HASH_DIGEST_FIELD]
	except Exception as e:
	raise SOL004PackageException(
	"Error parsing {} hash digest: {}".format(file_relative_path, e)
	)

	def get_package_file_hash_algorithm_from_manifest(self, file_relative_path):
	"""Returns the hash algorithm of a file inside this package as specified on the manifest file."""
	file_manifest_data = self._get_package_file_manifest_data(file_relative_path)
	try:
	return file_manifest_data[_MANIFEST_FILE_HASH_ALGORITHM_FIELD]
	except Exception as e:
	raise SOL004PackageException(
	"Error parsing {} hash digest: {}".format(file_relative_path, e)
	)

	@staticmethod
	def _get_hash_function_from_hash_algorithm(hash_algorithm):
	function_to_algorithm = {"SHA-256": hashlib.sha256, "SHA-512": hashlib.sha512}
	if hash_algorithm not in function_to_algorithm:
	error_msg = (
	"Error checking hash function: hash algorithm {} not supported".format(
	hash_algorithm
	)
	)
	raise SOL004PackageException(error_msg)
	return function_to_algorithm[hash_algorithm]

	def _calculate_file_hash(self, file_relative_path, hash_algorithm):
	file_path = self._get_package_file_full_path(file_relative_path)
	hash_function = self._get_hash_function_from_hash_algorithm(hash_algorithm)
	try:
	with open(file_path, "rb") as f:
	return hash_function(f.read()).hexdigest()
	except Exception as e:
	raise SOL004PackageException(
	"Error hashing {}: {}".format(file_relative_path, e)
	)

	def validate_package_file_hash(self, file_relative_path):
	"""Validates the integrity of a file using the hash algorithm and digest on the package manifest."""
	hash_algorithm = self.get_package_file_hash_algorithm_from_manifest(
	file_relative_path
	)
	file_hash = self._calculate_file_hash(file_relative_path, hash_algorithm)
	expected_file_hash = self.get_package_file_hash_digest_from_manifest(
	file_relative_path
	)
	if file_hash != expected_file_hash:
	error_msg = "Error validating {} hash: calculated hash {} is different than manifest hash {}"
	raise SOL004PackageException(
	error_msg.format(file_relative_path, file_hash, expected_file_hash)
	)

	def validate_package_hashes(self):
	"""Validates the integrity of all files listed on the package manifest."""
	for file_data in self._manifest_data:
	if _MANIFEST_FILE_PATH_FIELD in file_data:
	file_relative_path = file_data[_MANIFEST_FILE_PATH_FIELD]
	self.validate_package_file_hash(file_relative_path)

	def get_descriptor_location(self):
	"""Returns this package descriptor location as a relative path from the package root."""
	for tosca_meta in self._package_metadata:
	if _METADATA_DESCRIPTOR_FIELD in tosca_meta:
	return tosca_meta[_METADATA_DESCRIPTOR_FIELD]

	error_msg = "Error: no {} entry found on {}".format(
	_METADATA_DESCRIPTOR_FIELD, _METADATA_FILE_PATH
	)
	raise SOL004PackageException(error_msg)