graph_cve_sync.py

#!/usr/bin/env python3

"""Script which synchronizes CVEs from upstream CVE database to graph."""


import os
import json
import subprocess
import requests
import tempfile
import logging
from urllib.parse import urljoin
from datetime import datetime
from snyk_feed import SnykDataFetcher

from victimsdb_lib import VictimsDB
from f8a_utils.versions import get_versions_for_ep


logging.basicConfig(format='%(asctime)s %(levelname)s:%(message)s', level=logging.INFO)
logger = logging.getLogger(__name__)


F8A_CVEDB_URL = os.environ.get(
    'F8A_CVEDB_URL',
    'https://github.com/fabric8-analytics/cvedb.git'
)


VICTIMS_CVEDB_URL = os.environ.get(
    'VICTIMS_CVEDB_URL',
    'https://github.com/victims/victims-cve-db.git'
)

ECOSYSTEM_MAPPING = {
    'java': 'maven',
    'python': 'pypi',
    'javascript': 'npm'
}


class GraphDatabase(object):
    """Despite the name, this class encapsulates data-importer's CVE-related API."""

    def __init__(self, host, port):
        self._url = 'http://{host}:{port}'.format(host=host, port=port)
        self._version_url = urljoin(self._url, 'api/v1/cvedb-version')
        self._cves_url = urljoin(self._url, 'api/v1/cves')

    def get_version(self):
        """Get CVE DB version."""
        response = requests.get(self._version_url)
        response.raise_for_status()
        return response.json().get('version')

    def set_version(self, version):
        """Set CVE DB version."""
        headers = {'Content-type': 'application/json'}
        response = requests.put(
            self._version_url, headers=headers, data=json.dumps({'version': version})
        )
        response.raise_for_status()
        return response.json().get('version')

    def get_cves(self, ecosystem):
        """Get all CVEs for given ecosystem."""
        graph_ecosystem = ECOSYSTEM_MAPPING[ecosystem]

        url = urljoin(self._cves_url + '/', graph_ecosystem)
        response = requests.get(url)
        response.raise_for_status()
        response_json = response.json()
        return response_json['count'], response_json['cve_ids']

    def delete_cve(self, cve_id, dry_run=None):
        """Delete given CVE."""
        headers = {'Content-type': 'application/json'}
        payload = {
            "cve_id": cve_id
        }

        dry_run = dry_run or is_dry_run()
        log_prefix = '' if not dry_run else '[DRY_RUN] '

        logger.info('{log_prefix}Deleting CVE: {cve_id}'.format(
            log_prefix=log_prefix,
            cve_id=cve_id)
        )

        if not dry_run:
            response = requests.delete(
                self._cves_url, headers=headers, data=json.dumps(payload)

            )
            response.raise_for_status()

    def put_cve(self,
                cve,
                affected_package,
                affected_versions,
                ecosystem,
                dry_run=None):
        """Insert given CVE to graph."""
        graph_ecosystem = ECOSYSTEM_MAPPING[ecosystem]

        fixed_in = [str(x) for x in affected_package.fixedin if x]
        headers = {'Content-type': 'application/json'}
        payload = {
            "cve_id": cve.cve_id,
            "description": cve.description or "",
            "cvss_v2": cve.cvss_v2,
            "ecosystem": graph_ecosystem,
            "affected": [{"name": affected_package.name, "version": x} for x in affected_versions],
            "nvd_status": "ANALYZED",  # we only support analyzed ATM: https://nvd.nist.gov/vuln
            "fixed_in": fixed_in
        }

        dry_run = dry_run or is_dry_run()
        log_prefix = '' if not dry_run else '[DRY_RUN] '

        logger.info(
            '{log_prefix}Ingesting {cve_id}: {name} {versions}, fixed in: {fixed_in}'.format(
                log_prefix=log_prefix,
                cve_id=cve.cve_id,
                name=affected_package.name,
                versions=affected_versions,
                fixed_in=fixed_in
            )
        )
        if not dry_run:
            response = requests.put(
                self._cves_url, headers=headers, data=json.dumps(payload)
            )
            response.raise_for_status()


class Git(object):
    """Git repository helpers."""

    def __init__(self, git_url, destination):
        self._git_url = git_url
        self._destination = destination

    def clone(self):
        """Clone Git repo."""
        subprocess.check_call(
            ['git', 'clone', '--single-branch', self._git_url, self._destination]
        )

    def get_last_hash(self):
        """Get last hash from the repo."""
        last_commit_hash = subprocess.check_output(
            ['git', 'rev-parse', 'HEAD'],
            universal_newlines=True, cwd=self._destination
        )
        return last_commit_hash.strip()

    def get_diff_since(self, since_hash, ecosystem):
        """Get CVE IDs of CVEs which were added/modified since `since_hash`."""
        script = os.path.join(os.path.dirname(os.path.realpath(__file__)), 'git-modified-since.sh')
        output = subprocess.check_output(
            [script, self._destination, since_hash, ecosystem],
            universal_newlines=True
        )
        result = []
        if output:
            result = [x.strip() for x in output.split('\n')]
        return result


def get_affected_versions(affected, ecosystem):
    """Get all affected versions for given package."""
    all_versions = get_versions_for_ep(ECOSYSTEM_MAPPING[ecosystem], affected.name)
    affected_versions = set()
    for version in all_versions:
        if affected.affects(affected.name, version=version):
            affected_versions.add(version)
    return affected_versions


def get_ecosystem_vulnerabilities(cve_db, ecosystem):
    """Get vulnerabilities for given ecosystem."""
    if ecosystem == 'java':
        return cve_db.java_vulnerabilities()
    if ecosystem == 'javascript':
        return cve_db.javascript_vulnerabilities()
    if ecosystem == 'python':
        return cve_db.python_vulnerabilities()
    raise ValueError('Unsupported ecosystem: {e}'.format(e=ecosystem))


def put_cve_to_graph(cve, graph_db, ecosystem, dry_run=None):
    """Ingest given CVE to graph."""
    for affected_package in cve.affected:
        affected_versions = get_affected_versions(affected_package, ecosystem)
        if affected_versions:
            graph_db.put_cve(
                cve, affected_package, affected_versions, ecosystem,
                dry_run=dry_run or is_dry_run()
            )


def perform_full_sync(cve_db, graph_db, ecosystem):
    """Ingest all CVEs to graph."""
    logger.info('Performing full sync...')
    for cve in cve_db:
        try:
            put_cve_to_graph(cve, graph_db, ecosystem, dry_run=is_dry_run())
        except Exception as e:
            logger.error('Failed to sync {cve_id}: {e}'.format(cve_id=cve.cve_id, e=str(e)))


def perform_diff_sync(cve_db, graph_db, diff, ecosystem):
    """Only sync what's missing."""
    logger.info('Performing diff sync...')
    for cve_id in diff:
        try:
            cve = cve_db[cve_id]
        except KeyError as e:
            logger.error(str(e))
            continue
        try:
            put_cve_to_graph(cve, graph_db, ecosystem, dry_run=is_dry_run())
        except Exception as e:
            logger.error('Failed to sync {cve_id}: {e}'.format(cve_id=cve_id, e=str(e)))


def delete_superfluous_cves_from_graph(cve_db, graph_db, ecosystem):
    """Delete superfluous (not present in upstream DB) CVEs from graph."""
    logger.info('Deleting superfluous CVEs from graph...')
    _, graph_cve_ids = graph_db.get_cves(ecosystem)

    f8a_cvedb_cve_ids = {x.cve_id for x in cve_db}
    graph_cve_ids_to_delete = set(graph_cve_ids) - f8a_cvedb_cve_ids
    for cve_id in graph_cve_ids_to_delete:
        try:
            graph_db.delete_cve(cve_id, dry_run=is_dry_run())
        except Exception as e:
            logger.error('Failed to delete {cve_id}: {e}'.format(cve_id=cve_id, e=str(e)))


def is_full_sync(graph_db):
    """Return True if user wants to force full sync."""
    graph_versions = graph_db.get_version()
    return os.environ.get('SYNC_MODE', 'diff').lower() == 'full' or graph_versions is None


def is_dry_run():
    """Return True if this is a dry run."""
    return os.environ.get('DRY_RUN', 'false').lower() in ('1', 'yes', 'true')


def set_version(f8a_commit, victims_commit, graph_db):
    """Construct and set version in graph."""
    version = '{f8a};{vic}'.format(f8a=f8a_commit, vic=victims_commit)
    logger.info('New graph version: {v}'.format(v=version))

    if not is_dry_run():
        try:
            graph_db.set_version(version)
        except Exception as e:
            logger.error('Failed to set CVE DB version in graph: {e}'.format(e=str(e)))


def sync(cve_db_all, graph_db, f8a_git, victims_git):
    """Sync upstream CVE DB to graph."""
    graph_versions = graph_db.get_version()
    logger.info('Graph version: {v}'.format(v=graph_versions))

    for ecosystem in ('java', 'javascript', 'python'):
        logger.info('Processing "{e}" ecosystem...'.format(e=ecosystem))

        cve_db = get_ecosystem_vulnerabilities(cve_db_all, ecosystem)

        if is_full_sync(graph_db):
            perform_full_sync(cve_db, graph_db, ecosystem)
        else:
            f8a_graph_ver, victims_graph_ver = graph_versions.split(';')
            diff = [x for x in f8a_git.get_diff_since(f8a_graph_ver, ecosystem) if x]
            diff_victims = [
                x for x in victims_git.get_diff_since(victims_graph_ver, ecosystem) if x
            ]
            diff.extend(diff_victims)
            if diff:
                perform_diff_sync(cve_db, graph_db, diff, ecosystem)

        try:
            delete_superfluous_cves_from_graph(cve_db, graph_db, ecosystem)
        except Exception as e:
            logger.error('Failed to delete superfluous CVEs: {e}'.format(e=str(e)))

    set_version(f8a_git.get_last_hash(), victims_git.get_last_hash(), graph_db)


def run():
    """Prepare before sync."""
    with tempfile.TemporaryDirectory() as f8a_dir:
        f8a_git = Git(F8A_CVEDB_URL, f8a_dir)
        f8a_git.clone()

        with tempfile.TemporaryDirectory() as victims_dir:
            victims_git = Git(VICTIMS_CVEDB_URL, victims_dir)
            victims_git.clone()

            f8a_yaml_dir = os.path.join(f8a_dir, 'database/')
            victims_yaml_dir = os.path.join(victims_dir, 'database/')

            f8a_db_all = VictimsDB.from_dir(f8a_yaml_dir)
            victims_db_all = VictimsDB.from_dir(victims_yaml_dir)

            f8a_db_all.merge(victims_db_all, keep_ours=True)

            graph_db = GraphDatabase(
                host=os.environ.get(
                    'BAYESIAN_DATA_IMPORTER_SERVICE_HOST', 'localhost'
                ),
                port=os.environ.get('BAYESIAN_DATA_IMPORTER_SERVICE_PORT', '9192')
            )
            sync(f8a_db_all, graph_db, f8a_git, victims_git)


if __name__ == '__main__':
    start = datetime.now()
    SnykDataFetcher().run_snyk_fetch()
    logger.info("Snyk sync operation completed. Exiting.")
    logger.info("Time taken {}".format(datetime.now() - start))
    run()
    logger.info("Sync operation for CRA completed. Exiting.")
    logger.info("Total time taken {}".format(datetime.now() - start))
    exit(0)