vulnerability_metadata.py

#!/usr/bin/env python3
# Copyright © 2021 Red Hat Inc.
#
# 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.
# Author: Yusuf Zainee <yzainee@redhat.com>
#

"""Script which generates the vulnerability metadata file for the air gapped solution."""

from datetime import datetime
import json
from helper import Helper
import logging


logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
SUPPORTED_ECOSYSTEMS = ['maven', 'npm', 'pypi', 'golang']
today = datetime.now().strftime('%d-%m-%Y')


class Vulnerability:
    """Populate minimal viable vuln information."""

    def __init__(self):
        """Init method for Vulnerability class."""
        self.helper = Helper()

    def get_vuln_data(self):
        """Get data from s3 or default."""
        # Will be True if we need the entire feed to run. It will recreate vuln file.
        if self.helper.is_complete_vuln_mode():
            with open('data/default_vulnerability.json', encoding='utf-8') as f:
                data = json.load(f)
        else:
            # in normal mode will add delta info into existing file.
            data = self.helper.read_data_from_s3("vulnerability-data", "snyk-feed/")
        return data

    def save_vuln_json(self, data):
        """Save the json in the vuln file in S3."""
        self.helper.store_json_content(data, "snyk-feed/vulnerability-data.json")

    def generate_vuln_json(self, vuln_data, pkg):
        """Generate a json from the vulnerability data."""
        # Convert list to string.
        ver_str = "" + str(vuln_data['affected']).replace(' ', '')\
            .replace('[', '').replace(']', '').replace('\'', '')

        return {
            "snyk_vuln_id": vuln_data['id'],
            "package_name": pkg,
            "vulnerable_versions": ver_str,
            "severity": vuln_data['severity'],
            "title": vuln_data['title'],
            "url": vuln_data['url'],
            "fixed_in": vuln_data['initiallyFixedIn'],
            "updated_on": today
        }

    def extract_info(self, snyk_data, json_data):
        """Extract the vuln info and modify the existing json."""
        for eco in SUPPORTED_ECOSYSTEMS:
            if eco in snyk_data:
                logger.info("Adding data for {} from the parsed snyk feed.".format(eco))
                eco_data = snyk_data[eco]
                for pkg in eco_data:
                    if pkg not in json_data[eco]:
                        json_data[eco][pkg] = {}
                    for vuln in eco_data[pkg]['vulnerabilities']:
                        id = vuln['id']
                        logger.info("Adding {}.".format(id))
                        json_data[eco][pkg][id] = self.generate_vuln_json(vuln, pkg)
            else:
                logger.info("No data for {} present in the parsed snyk feed.".format(eco))
        return json_data

    def process(self, snyk_data):
        """Process the vulnerability operation for air gapped or cache."""
        json_data = self.get_vuln_data()
        json_data = self.extract_info(snyk_data, json_data)
        self.save_vuln_json(json_data)