--- /dev/null
+# -*- coding: utf-8 -*-
+
+# Copyright 2021 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: fbravo@whitestack.com or agarcia@whitestack.com
+##
+
+import os
+import yaml
+import hashlib
+
+
+class SOLPackageException(Exception):
+ pass
+
+
+class SOLPackage:
+ _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"
+
+ _MANIFEST_ALL_FIELDS = []
+
+ def __init__(self, package_path=""):
+ self._package_path = package_path
+
+ self._package_metadata = self._parse_package_metadata()
+
+ try:
+ self._manifest_data = self._parse_manifest_data()
+ except Exception:
+ self._manifest_data = None
+
+ try:
+ self._manifest_metadata = self._parse_manifest_metadata()
+ except Exception:
+ self._manifest_metadata = None
+
+ 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(self._METADATA_FILE_PATH)
+ except FileNotFoundError as e:
+ raise e
+ except (Exception, OSError) as e:
+ raise SOLPackageException(
+ "Error parsing {}: {}".format(self._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 SOLPackageException(error_msg.format(len(package_root_yamls)))
+
+ base_manifest = [
+ {
+ SOLPackage._METADATA_DESCRIPTOR_FIELD: package_root_yamls[0],
+ SOLPackage._METADATA_MANIFEST_FIELD: "{}.mf".format(
+ os.path.splitext(package_root_yamls[0])[0]
+ ),
+ SOLPackage._METADATA_CHANGELOG_FIELD: SOLPackage._METADATA_DEFAULT_CHANGELOG_PATH,
+ SOLPackage._METADATA_LICENSES_FIELD: SOLPackage._METADATA_DEFAULT_LICENSES_PATH,
+ }
+ ]
+
+ return base_manifest
+
+ def _parse_manifest_data(self):
+ manifest_path = None
+ for tosca_meta in self._package_metadata:
+ if SOLPackage._METADATA_MANIFEST_FIELD in tosca_meta:
+ manifest_path = tosca_meta[SOLPackage._METADATA_MANIFEST_FIELD]
+ break
+ else:
+ error_msg = "Error parsing {}: no {} field on path".format(
+ self._METADATA_FILE_PATH, self._METADATA_MANIFEST_FIELD
+ )
+ raise SOLPackageException(error_msg)
+
+ try:
+ return self._parse_file_in_blocks(manifest_path)
+
+ except (Exception, OSError) as e:
+ raise SOLPackageException("Error parsing {}: {}".format(manifest_path, e))
+
+ def _parse_manifest_metadata(self):
+ try:
+ base_manifest = {}
+ manifest_file = os.open(
+ os.path.join(
+ self._package_path, base_manifest[self._METADATA_MANIFEST_FIELD]
+ ),
+ "rw",
+ )
+ for line in manifest_file:
+ fields_in_line = line.split(":", maxsplit=1)
+ fields_in_line[0] = fields_in_line[0].strip()
+ fields_in_line[1] = fields_in_line[1].strip()
+ if fields_in_line[0] in self._MANIFEST_ALL_FIELDS:
+ base_manifest[fields_in_line[0]] = fields_in_line[1]
+ return base_manifest
+ except (Exception, OSError) as e:
+ raise SOLPackageException(
+ "Error parsing {}: {}".format(
+ base_manifest[SOLPackage._METADATA_MANIFEST_FIELD], 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(SOLPackage._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 SOLPackageException(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[SOLPackage._MANIFEST_FILE_HASH_DIGEST_FIELD]
+ except Exception as e:
+ raise SOLPackageException(
+ "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[SOLPackage._MANIFEST_FILE_HASH_ALGORITHM_FIELD]
+ except Exception as e:
+ raise SOLPackageException(
+ "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 SOLPackageException(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 SOLPackageException(
+ "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 SOLPackageException(
+ 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 SOLPackage._MANIFEST_FILE_PATH_FIELD in file_data:
+ file_relative_path = file_data[SOLPackage._MANIFEST_FILE_PATH_FIELD]
+ self.validate_package_file_hash(file_relative_path)
+
+ def create_or_update_metadata_file(self):
+ """
+ Creates or updates the metadata file with the hashes calculated for each one of the package's files
+ """
+ if not self._manifest_metadata:
+ self.generate_manifest_data_from_descriptor()
+
+ self.write_manifest_data_into_file()
+
+ def generate_manifest_data_from_descriptor(self):
+ pass
+
+ def write_manifest_data_into_file(self):
+ with open(self.get_manifest_location(), "w") as metadata_file:
+ # Write manifest metadata
+ for metadata_entry in self._manifest_metadata:
+ metadata_file.write(
+ "{}: {}\n".format(
+ metadata_entry, self._manifest_metadata[metadata_entry]
+ )
+ )
+
+ # Write package's files hashes
+ file_hashes = {}
+ for root, dirs, files in os.walk(self._package_path):
+ for a_file in files:
+ file_path = os.path.join(root, a_file)
+ file_relative_path = file_path[len(self._package_path) :]
+ if file_relative_path.startswith("/"):
+ file_relative_path = file_relative_path[1:]
+ file_hashes[file_relative_path] = self._calculate_file_hash(
+ file_relative_path, "SHA-512"
+ )
+
+ for file, hash in file_hashes.items():
+ file_block = "Source: {}\nAlgorithm: SHA-512\nHash: {}\n\n".format(
+ file, hash
+ )
+ metadata_file.write(file_block)
+
+ 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 SOLPackage._METADATA_DESCRIPTOR_FIELD in tosca_meta:
+ return tosca_meta[SOLPackage._METADATA_DESCRIPTOR_FIELD]
+
+ error_msg = "Error: no {} entry found on {}".format(
+ SOLPackage._METADATA_DESCRIPTOR_FIELD, SOLPackage._METADATA_FILE_PATH
+ )
+ raise SOLPackageException(error_msg)
+
+ def get_manifest_location(self):
+ """Return the VNF/NS manifest location as a relative path from the package root."""
+ for tosca_meta in self._package_metadata:
+ if SOLPackage._METADATA_MANIFEST_FIELD in tosca_meta:
+ return tosca_meta[SOLPackage._METADATA_MANIFEST_FIELD]
+
+ raise SOLPackageException("No manifest file defined for this package")
projects / osm / common.git / blobdiff