path: root/meta-arm/ci/patchreview

#! /usr/bin/env python3
#
# SPDX-License-Identifier: GPL-2.0-only
#

# TODO
# - option to just list all broken files
# - test suite
# - validate signed-off-by

import argparse
import collections
import json
import os
import re
import subprocess

status_values = (
    "accepted",
    "pending",
    "inappropriate",
    "backport",
    "submitted",
    "denied",
)


class PatchResult:
    # Whether the patch has an Upstream-Status or not
    missing_upstream_status = False
    # If the Upstream-Status tag is malformed in some way (string for bad bit)
    malformed_upstream_status = None
    # If the Upstream-Status value is unknown (boolean)
    unknown_upstream_status = False
    # The upstream status value (Pending, etc)
    upstream_status = None
    # Whether the patch has a Signed-off-by or not
    missing_sob = False
    # Whether the Signed-off-by tag is malformed in some way
    malformed_sob = False
    # The Signed-off-by tag value
    sob = None
    # Whether a patch looks like a CVE but doesn't have a CVE tag
    missing_cve = False


class Summary:
    total = 0
    cve_missing = 0
    sob_missing = 0
    sob_malformed = 0
    status_missing = 0
    status_malformed = 0
    status_pending = 0

def blame_patch(patch):
    """
    From a patch filename, return a list of "commit summary (author name <author
    email>)" strings representing the history.
    """
    return subprocess.check_output(("git", "log",
                                    "--follow", "--find-renames", "--diff-filter=A",
                                    "--format=%s (%aN <%aE>)",
                                    "--", patch)).decode("utf-8").splitlines()

def patchreview(patches):
    # General pattern: start of line, optional whitespace, tag with optional
    # hyphen or spaces, maybe a colon, some whitespace, then the value, all case
    # insensitive.
    sob_re = re.compile(r"^[\t ]*(Signed[-_ ]off[-_ ]by:?)[\t ]*(.+)", re.IGNORECASE | re.MULTILINE)
    status_re = re.compile(r"^[\t ]*(Upstream[-_ ]Status:?)[\t ]*(\w*)", re.IGNORECASE | re.MULTILINE)
    cve_tag_re = re.compile(r"^[\t ]*(CVE:)[\t ]*(.*)", re.IGNORECASE | re.MULTILINE)
    cve_re = re.compile(r"cve-[0-9]{4}-[0-9]{4,6}", re.IGNORECASE)

    results = {}

    for patch in patches:

        result = PatchResult()
        results[patch] = result

        content = open(patch, encoding="ascii", errors="ignore").read()

        # Find the Signed-off-by tag
        match = sob_re.search(content)
        if match:
            value = match.group(1)
            if value != "Signed-off-by:":
                result.malformed_sob = value
            result.sob = match.group(2)
        else:
            result.missing_sob = True

        # Find the Upstream-Status tag
        match = status_re.search(content)
        if match:
            value = match.group(1)
            if value != "Upstream-Status:":
                result.malformed_upstream_status = value

            value = match.group(2).lower()
            # TODO: check case
            if value not in status_values:
                result.unknown_upstream_status = True
            result.upstream_status = value
        else:
            result.missing_upstream_status = True

        # Check that patches which looks like CVEs have CVE tags
        if cve_re.search(patch) or cve_re.search(content):
            if not cve_tag_re.search(content):
                result.missing_cve = True
        # TODO: extract CVE list

    return results


def analyse(results, want_blame=False, verbose=True):
    """
    want_blame: display blame data for each malformed patch
    verbose: display per-file results instead of just summary
    """

    # want_blame requires verbose, so disable blame if we're not verbose
    if want_blame and not verbose:
        want_blame = False

    summary = Summary()

    for patch in sorted(results):
        r = results[patch]
        summary.total += 1
        need_blame = False

        # Build statistics
        if r.missing_sob:
            summary.sob_missing += 1
        if r.malformed_sob:
            summary.sob_malformed += 1
        if r.missing_upstream_status:
            summary.status_missing += 1
        if r.malformed_upstream_status or r.unknown_upstream_status:
            summary.status_malformed += 1
            # Count patches with no status as pending
            summary.status_pending += 1
        if r.missing_cve:
            summary.cve_missing += 1
        if r.upstream_status == "pending":
            summary.status_pending += 1

        # Output warnings
        if r.missing_sob:
            need_blame = True
            if verbose:
                print("Missing Signed-off-by tag (%s)" % patch)
        if r.malformed_sob:
            need_blame = True
            if verbose:
                print("Malformed Signed-off-by '%s' (%s)" % (r.malformed_sob, patch))
        if r.missing_cve:
            need_blame = True
            if verbose:
                print("Missing CVE tag (%s)" % patch)
        if r.missing_upstream_status:
            need_blame = True
            if verbose:
                print("Missing Upstream-Status tag (%s)" % patch)
        if r.malformed_upstream_status:
            need_blame = True
            if verbose:
                print("Malformed Upstream-Status '%s' (%s)" % (r.malformed_upstream_status, patch))
        if r.unknown_upstream_status:
            need_blame = True
            if verbose:
                print("Unknown Upstream-Status value '%s' (%s)" % (r.upstream_status, patch))

        if want_blame and need_blame:
            print("\n".join(blame_patch(patch)) + "\n")

    return summary


def display_summary(summary, verbose):
    def percent(num):
        try:
            return "%d (%d%%)" % (num, round(num * 100.0 / summary.total))
        except ZeroDivisionError:
            return "N/A"

    if verbose:
        print()

    print("""Total patches found: %d
Patches missing Signed-off-by: %s
Patches with malformed Signed-off-by: %s
Patches missing CVE: %s
Patches missing Upstream-Status: %s
Patches with malformed Upstream-Status: %s
Patches in Pending state: %s""" % (summary.total,
                                   percent(summary.sob_missing),
                                   percent(summary.sob_malformed),
                                   percent(summary.cve_missing),
                                   percent(summary.status_missing),
                                   percent(summary.status_malformed),
                                   percent(summary.status_pending)))


def generate_metrics(summary, output):
    # https://github.com/OpenObservability/OpenMetrics/blob/main/specification/OpenMetrics.md
    # Summary attribute name, MetricPoint help
    mapping = (
        ("total", "Total patches"),
        ("cve_missing", "Patches missing CVE tag"),
        ("sob_malformed", "Patches with malformed Signed-off-by"),
        ("sob_missing", "Patches with missing Signed-off-by"),
        ("status_malformed", "Patches with malformed Upstream-Status"),
        ("status_missing", "Patches with missing Upstream-Status"),
        ("status_pending", "Patches with Pending Upstream-Status")
    )
    for attr, help in mapping:
        metric = f"patch_check_{attr}"
        value = getattr(summary, attr)
        output.write(f"""
# TYPE {metric} gauge
# HELP {help}
{metric} {value}
""")
    output.write("\n# EOF\n")

def histogram(results):
    import math

    from toolz import dicttoolz, recipes
    counts = recipes.countby(lambda r: r.upstream_status, results.values())
    bars = dicttoolz.valmap(lambda v: "#" * int(math.ceil(float(v) / len(results) * 100)), counts)
    for k in bars:
        print("%-20s %s (%d)" % (k.capitalize() if k else "No status", bars[k], counts[k]))

def gather_patches(directories):
    patches = []
    for directory in directories:
        filenames = subprocess.check_output(("git", "-C", directory, "ls-files", "recipes-*/**/*.patch", "recipes-*/**/*.diff")).decode("utf-8").split()
        patches += [os.path.join(directory, f) for f in filenames]
    return patches

if __name__ == "__main__":
    args = argparse.ArgumentParser(description="Patch Review Tool")
    args.add_argument("-b", "--blame", action="store_true", help="show blame for malformed patches")
    args.add_argument("-v", "--verbose", action="store_true", help="show per-patch results")
    args.add_argument("-g", "--histogram", action="store_true", help="show patch histogram")
    args.add_argument("-j", "--json", help="update JSON")
    args.add_argument("-m", "--metrics", type=argparse.FileType('w'), help="write OpenMetrics")
    args.add_argument("dirs", metavar="DIRECTORY", nargs="+", help="directory to scan")
    args = args.parse_args()

    patches = gather_patches(args.dirs)
    results = patchreview(patches)
    summary = analyse(results, want_blame=args.blame, verbose=args.verbose)
    display_summary(summary, verbose=args.verbose)

    if args.json:
        if os.path.isfile(args.json):
            data = json.load(open(args.json))
        else:
            data = []

        row = collections.Counter()
        row["total"] = len(results)
        row["date"] = subprocess.check_output(["git", "-C", args.dirs[0], "show", "-s", "--pretty=format:%cd", "--date=format:%s"]).decode("utf-8").strip()
        for r in results.values():
            if r.upstream_status in status_values:
                row[r.upstream_status] += 1
            if r.malformed_upstream_status or r.missing_upstream_status:
                row["malformed-upstream-status"] += 1
            if r.malformed_sob or r.missing_sob:
                row["malformed-sob"] += 1

        data.append(row)
        json.dump(data, open(args.json, "w"))

    if args.metrics:
        generate_metrics(summary, args.metrics)

    if args.histogram:
        print()
        histogram(results)