snapscve.py

#!/usr/bin/python3
"""Check for unfixed CVE issues in desktop snaps"""
import argparse
import json
import os
import subprocess
import sys
import webbrowser

import yaml

import snaps

if sys.version_info >= (3, 0):
    import urllib.request

    import requests
else:
    import urllib2

# use existing cache
try:
    with open("cve.yml", "r") as cvereport:
        cvedict = yaml.load(cvereport, Loader=yaml.Loader)
except FileNotFoundError:
    cvedict = {}

parser = argparse.ArgumentParser()

parser.add_argument(
    "-v", "--verbose", help="display debug information", action="store_true"
)
parser.add_argument(
    "-i", "--interactive", help="prompt about actions", action="store_true"
)
parser.add_argument(
    "-n", "--nocache", help="don't cache the downloaded snaps", action="store_true"
)

arg = parser.parse_args()

#launchpad.people[...].memberships_details

def debug(text):
    """ Print when using the verbose option"""
    if arg.verbose:
        print(text)


STORE_URL = "https://api.snapcraft.io/api/v1/snaps/details/{snap}?channel={channel}"
STORE_HEADERS = {"X-Ubuntu-Series": "16", "X-Ubuntu-Architecture": "{arch}"}
CHECK_NOTICES_PATH = "/snap/bin/review-tools.check-notices"


def store_parse_versions(package):
    """Build a dictionnary of the channels and revisions of a snap in the store"""
    result = {}
    # the store wants a Snap-Device-Series header
    if sys.version_info >= (3, 0):
        req = urllib.request.Request(
            "http://api.snapcraft.io/v2/snaps/info/%s" % package,
            headers={"Content-Type": "application/json", "Snap-Device-Series": "16"},
        )
        snapdetails = urllib.request.urlopen(req)
    else:
        req = urllib2.Request(
            "http://api.snapcraft.io/v2/snaps/info/%s" % package,
            headers={"Content-Type": "application/json", "Snap-Device-Series": "16"},
        )
        snapdetails = urllib2.urlopen(req)
    report = json.load(snapdetails)
    for items in report["channel-map"]:
        larch = items["channel"]["architecture"]
        lchannel = items["channel"]["name"]
        lrev = items["revision"]
        if larch not in result:
            result[larch] = {}
        result[larch][lchannel] = lrev
    return result


def ask_yes_no(ynquestion):
    """ask a yes or no type of question"""
    options = ["yes", "no"]
    selected = None
    while selected not in options:
        try:
            selected = input("%s [y|N]? " % ynquestion)
        except (EOFError, KeyboardInterrupt):
            print("\nAborting as requested.")
            sys.exit(1)
        if selected == "":
            selected = "no"
        else:
            for option in options:
                # Example: User typed "y" instead of "yes".
                if selected == option[0]:
                    selected = option
        if selected not in options:
            print("Please answer the question with yes or no")
    print("")
    return selected


def get_store_snap(processor, snap, snapchannel):
    """Get the url from a snap in the store"""
    debug("Checking for snap %s on %s in channel %s" % (snap, processor, snapchannel))
    data = {
        "snap": snap,
        "channel": snapchannel,
        "arch": processor,
    }
    if sys.version_info >= (3, 0):
        req = urllib.request.Request(
            STORE_URL.format(**data),
            headers={k: v.format(**data) for k, v in STORE_HEADERS.items()},
        )
        snapdetails = urllib.request.urlopen(req)
    else:
        req = urllib2.Request(
            STORE_URL.format(**data),
            headers={k: v.format(**data) for k, v in STORE_HEADERS.items()},
        )
        snapdetails = urllib2.urlopen(req)

    try:
        result = json.load(snapdetails)
    except json.JSONDecodeError:
        print("Could not parse store response")
        return ""
    else:
        return result["anon_download_url"]


def fetch_url(entry):
    """Download the file from the given url"""
    dest, uri = entry
    req = requests.get(uri, stream=True)
    debug("Downloading %s to %s..." % (uri, dest))
    if req.status_code == 200:
        with open(dest, "wb") as download:
            for chunk in req:
                download.write(chunk)
    return dest


def is_lpbuildsnap_available():
    """Check if the lp-build-snap utility is available"""
    if os.path.isfile("/snap/bin/lp-build-snap"):
        return True
    debug("lp-build-snap not installed, no rebuld action")
    return False

# iterate over the list of snaps
for snapline in snaps.normalsnaps + snaps.specialsnaps:
    store_revisions = set()
    src = snapline[0]
    debug("* considering source %s" % src)
    if src not in cvedict:
        cvedict[src] = {}

    store_versions_table = store_parse_versions(src)
    debug("store versions")
    debug(store_versions_table)

    arches_needing_rebuild = set()
    arches_promote_candidate_to_stable = set()
    for architecture in store_versions_table:
        channel_is_fixed = set()
        for channel in store_versions_table[architecture]:
            rev = store_versions_table[architecture][channel]
            store_revisions.add(rev)

            debug("downloading %s, channel %s, revision %s" % (src, channel, rev))
            newenv = os.environ.copy()
            newenv["UBUNTU_STORE_ARCH"] = architecture
            snapfilename = "%s_%s.snap" % (src, rev)
            if not os.path.exists(snapfilename):
                fetch_url([snapfilename, get_store_snap(architecture, src, channel)])
            else:
                debug("target already exists, not downloading again")

            try:
                notices = subprocess.check_output(
                    [CHECK_NOTICES_PATH] + ["./%s_%s.snap" % (src, rev)],
                    encoding="UTF-8",
                )
            except subprocess.CalledProcessError as e:
                print("Failed to check notices")
                cvedict[src][rev] = []
                continue
            else:
                notices = json.loads(notices)
                if notices and notices[src][str(rev)]:
                    debug(", ".join(notices[src][str(rev)]))
                    cvedict[src][rev] = notices[src][str(rev)]
                else:
                    debug("no CVE found")
                    cvedict[src][rev] = []
                    channel_is_fixed.add(channel)
        if (
            "candidate" in channel_is_fixed
            and "stable" in store_versions_table[architecture].keys()
            and "stable" not in channel_is_fixed
        ):
            arches_promote_candidate_to_stable.add(architecture)
        if not channel_is_fixed:
            arches_needing_rebuild.add(architecture)
        if arg.nocache and os.path.exists(snapfilename):
            os.remove(snapfilename)
            debug("nocache, removing %s")

    if arches_promote_candidate_to_stable:
        if not set(store_versions_table.keys()) - arches_promote_candidate_to_stable:
            print(src, "candidate can be promoted to stable")
        else:
            print(
                src,
                "candidate can be promoted to stable on",
                ", ".join(arches_promote_candidate_to_stable),
            )
        print("-> https://snapcraft.io/%s/releases\n" % src)
        question = "Do you want to open the store webpage for %s now" % src
        if arg.interactive and ask_yes_no(question) == "yes":
            webbrowser.open("https://snapcraft.io/%s/releases" % src)
    if not set(store_versions_table.keys()) - arches_needing_rebuild:
        print(src, "needs a rebuild\n")
        if is_lpbuildsnap_available() and arg.interactive:
            for reference in snapline[1:4]:
                if reference:
                    pkgtobuild = reference.split("/")[-1]
                    owner = reference.split("~")[1].split("/")[0]
                    cmd = ["lp-build-snap", "--lpname", owner, pkgtobuild]
                    question = "Do you want to call '%s'" % " ".join(cmd)
                    if ask_yes_no(question) == "yes":
                        subprocess.run(cmd, check=True)
    elif arches_needing_rebuild:
        print(src, "needs a rebuild on %s\n" % ", ".join(arches_needing_rebuild))

    revisions_to_delete = set()
    for rev in cvedict[src]:
        if rev not in store_revisions:
            revisions_to_delete.add(rev)
    for rev in revisions_to_delete:
        details = ""
        snapfilename = "%s_%s.snap" % (src, rev)
        if os.path.exists(snapfilename):
            details = " and deleted %s" % snapfilename
            os.remove(snapfilename)
        debug("Cleaning out outdated revision %s from the cache%s" % (rev, details))
        del cvedict[src][rev]
    debug("")

# write updated cache
with open("cve.yml", "w") as outfile:
    yaml.dump(cvedict, outfile, default_flow_style=False)