check_gitlab_vuln_report.py

#!/usr/bin/env python3
import argparse
import sys
from pprint import pprint
from pathlib import Path
from pathlib import PurePath
import os.path
import csv
import json
from operator import itemgetter
import requests

# https://www.peterbe.com/plog/best-practice-with-retries-with-requests
from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util.retry import Retry

def requests_retry_session(
    retries=8,
    backoff_factor=1,
    status_forcelist=(500, 502, 504, 404),
    session=None,
):
    session = session or requests.Session()
    retry = Retry(
        total=retries,
        read=retries,
        connect=retries,
        backoff_factor=backoff_factor,
        status_forcelist=status_forcelist,
    )
    adapter = HTTPAdapter(max_retries=retry)
    session.mount('http://', adapter)
    session.mount('https://', adapter)
    return session

def nagios_ok(*args):
    print("OK: " + "\n".join(args))
    sys.exit(0)

def nagios_warn(*args):
    print("WARNING: " + "\n".join(args))
    sys.exit(1)

def nagios_crit(*args):
    print("CRITICAL: " + "\n".join(args))
    sys.exit(2)

def nagios_unknown(*args):
    print("UNKNOWN: " + "\n".join(args))
    sys.exit(3)


# Debug
# import logging
# logging.basicConfig(level=logging.DEBUG)


try:
    parser = argparse.ArgumentParser(
            description='''
            Monitor the vulnerabilities in Gitlab generated reports.
            ''')
    parser.add_argument('--api',
            help='''
            The API URL to use. See
            https://docs.gitlab.com/ee/api/vulnerability_exports.html for
            examples. This can use project, group, or instance level reports.
            Example:
            "https://gitlab.my.org/api/v4/groups/234/vulnerability_exports".
            ''',
            required=True)
    parser.add_argument('--token',
            help='''
            Access token to use. This can be a personal access token, and
            requires the "api" scope. Read
            https://assets.nagios.com/downloads/nagioscore/docs/nagioscore/4/en/security.html#bestpractices
            when using Nagios.
            ''',
            required=True)
    parser.add_argument('--diff',
            help='''
            Instead of the absolute number of vulnerabilties in the report, use
            the difference compared to the previous report. This effectively
            means you just see the changes. A cache file will be used to store
            the relevant report details for comparison between checks. See also
            the '--cachedir' option.
            ''',
            action="store_true")
    parser.add_argument('--warn', '-w',
            help='''
            Number of found vulnerabilities that should result in a WARNING
            (default: 1)).
            ''',
            required=False,
            type=int,
            default=1)
    parser.add_argument('--crit', '-c',
            help='''
            Number of found vulnerabilities that should result in a CRITICAL
            (default: 3)).
            ''',
            required=False,
            type=int,
            default=3)
    parser.add_argument('--severity',
            help='''
            Comma separated list of vulnerability severities to take into
            account. Options: info, unknown, low, medium, high, criticial.
            Default: "high,critical".
            ''',
            required=False,
            default='high,critical')
    parser.add_argument('--status',
            help='''
            Comma separated list of vulnerability statuses to take into
            account. Options: detected, confirmed, dismissed, resolved.
            Default: "detected,confirmed".
            ''',
            default='detected,confirmed',
            required=False)
    parser.add_argument('--cachedir',
            help='''
            Which directory to use for storing cached content (default:
            "/tmp").
            ''',
            default='/tmp',
            required=False)
    parser.add_argument('--verbose',
            help='''
            Show verbose output.
            ''',
            action="store_true")
    args = parser.parse_args()


    # our arguments
    api = args.api
    diff = args.diff
    warn = args.warn
    crit = args.crit
    cachedir = args.cachedir
    verbose = args.verbose


    headers = { 'PRIVATE-TOKEN': args.token }
    # Split these into lists
    severity = args.severity.split(',')
    status = args.status.split(',')
    cachefile = PurePath(cachedir, PurePath( __file__).name).with_suffix('.cache.json')

    
    # Filter, dedupe, and sort
    def filter_vuls(
            vuls,
            severity=severity,
            status=status):
        filtered = filter(lambda d:
                d['Severity'] in severity and d['Status'] in status,
                vuls)
        deduped = [dict(t) for t in {tuple(d.items()) for d in filtered}]
        # Sort on all keys
        for sortkey in deduped[0].keys():
            deduped = sorted(deduped, key=itemgetter(sortkey))
        return deduped
    
    # write cache to disk (serialised as JSON)
    def write_cache(data, file=cachefile):
        jsondata = json.dumps(filter_vuls(data), indent=4)
        with open(file, "w") as c:
            c.write(jsondata)

    # Read JSON serialised cache from disk
    def read_cache(file=cachefile):
        with open(file, "r") as c:
            content = json.loads(c.read())
            return filter_vuls(content)

    # Initial request to the API
    r = requests.post(api, headers=headers)
    r.raise_for_status()

    # Fetch the download URI for the security report
    security_report_download_url = r.json()['_links']['download']

    # Request the report URL, using the retry logic, as the report will take some time
    # to generate.
    download = requests_retry_session().get(security_report_download_url, headers=headers)
    download.raise_for_status()

    # Generate dict structure from CSV data
    reader = csv.DictReader(download.text.splitlines())
    
    # Filter the results
    vuls = filter_vuls(reader)

    # Cosmetics
    report_line = "See the entire report at " + security_report_download_url

    # We are only interested in the difference with the previous report
    if diff:
        if os.path.isfile(cachefile):
            # Fetch old vulnerabilities
            old_vuls = read_cache()

            # write new data to disk 
            write_cache(vuls)

            if vuls != old_vuls:
                # Calculate the difference between the old and new vulnerability lists
                difference = [i for i in vuls if i not in old_vuls] + [j for j in old_vuls if j not in vuls]
                plural = '' if len(difference) == 1 else 's'
                if len(vuls) > len(old_vuls):
                    if len(difference) > crit:
                        nagios_crit(
                                str(len(difference)) + " new issue" + plural + " found",
                                json.dumps(difference, indent=4),
                                report_line
                                )
                    else:
                        nagios_warn(
                                str(len(difference)) + " new issue" + plural + " found",
                                json.dumps(difference, indent=4),
                                report_line
                                )
                elif len(vuls) < len(old_vuls):
                    nagios_ok(
                            str(len(difference)) + " issue" + plural + " less found",
                            json.dumps(difference, indent=4),
                            report_line
                            )
                else:
                    nagios_warn(
                            "Same amount of issues found, but different ones",
                            json.dumps(difference, indent=4),
                            report_line
                            )
            else:
                write_cache(vuls)
                nagios_ok(
                        "No changes compared to previous check",
                        report_line
                        )

        else:
            # Save the current output to disk for next time
            write_cache(vuls)
            nagios_unknown(
                    "No cache file yet, generating it for next use"
                    )


    found_items = len(vuls)
    plural = '' if found_items == 1 else 's'
    message = (
            "Found " + str(found_items) + " issue" + plural + " with " +
            "severity " + " or ".join(severity) +
            ", and " +  "status " + " or ".join(status)
            )

    if found_items > crit:
        nagios_crit(message, report_line)
    elif found_items > warn:
        nagios_warn(message, report_line)
    else:
        nagios_ok(message, report_line)

except Exception as e:
    nagios_unknown(e)