osm_common/sol004_package.py

# -*- 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)