~ubuntu-security/ubuntu-cve-tracker/master

#!/usr/bin/env python

# Copyright 2007-2017, Canonical, Ltd.

# Author: Kees Cook <kees@ubuntu.com>

#         Jamie Strandboge <jamie@canonical.com>

#         Marc Deslauriers <marc.deslauriers@canonical.com>

# License: GPLv3

#

# Extract/download list of changes file links from a given LP name, pkg, version

#

# TODO: need to handle multiple orig tarballs for 3.0 format

# http://wiki.debian.org/Projects/DebSrc3.0

from __future__ import print_function

import copy

import optparse

import os.path

import progressbar

import re

import shutil

import sys

import tempfile

import urllib

import cve_lib

from source_map import version_compare

try:

    import lpl_common

except ImportError as e:

    print("lpl_common.py seems to be missing. Please create a symlink from $UQT/common/lpl_common.py to $UCT/scripts/", file=sys.stderr)

    print(e, file=sys.stderr)

    sys.exit(1)

# import warnings

# warnings.filterwarnings('ignore', 'apt API not stable yet', FutureWarning)

import apt

def download(url):

    # Download file to tmpdir

    if not os.path.exists(tmpdir):

        print("Failed: '%s' does not exist" % (tmpdir), file=sys.stderr)

        sys.exit(1)

    # Initialize progressbar so it has a real view of the time taken to fetch

    widgets = [progressbar.Percentage(),

               ' ', progressbar.Bar(marker='=', left='[', right=']'),

               ' ', progressbar.FileTransferSpeed(),

               ' ', progressbar.ETA()]

    bar = progressbar.ProgressBar(widgets=widgets).start()

    # Open the URL

    urlfile = lpl_common.open_url(opener, url)

    received = 0

    # Extract expected file size, updating progress bar and widgets

    size = int(urlfile.info().getheader('Content-Length').strip())

    bar.maxval = size

    widgets.insert(1, ' of %d' % (size))

    bar.widgets = widgets

    bar.update(received)

    # See 'if opt.action == changes' section when adding replace() characters

    name = urllib.unquote(os.path.join(tmpdir, os.path.basename(url)))

    try:

        tmp, tmpname = tempfile.mkstemp()

    except Exception:

        raise

    # Fetch data, updating progressbar in minimum 100K chunks

    while True:

        data = urlfile.read(1024 * 100)

        if not data:

            break

        received += len(data)

        os.write(tmp, data)

        bar.update(received)

        if received == size:

            bar.finish()

    # Close and rename

    os.close(tmp)

    shutil.move(tmpname, name)

    return name

def download_url(url):

    '''Display URL, and optionally download it, if requested and matches the re'''

    if url is None:

        print("download_url(): passed an empty url, skipping...", file=sys.stderr)

        return None

    filename = os.path.basename(url)

    if not opt.re or re.search(opt.re, filename):

        print(url)

        if opt.download:

            return download(url)

    return None

#

# START SCRIPT

#

parser = optparse.OptionParser()

parser.add_option("--action", help="What action to take: 'changes'(default), 'check-build', 'binaries', 'source', 'buildlogs', 'list'", metavar="NAME", action='store', default='changes')

parser.add_option("--ppa", help="Which PPA to use (default is 'ubuntu-security/ppa')", metavar="PERSON[/PPA]", action='store', default='ubuntu-security/ppa')

parser.add_option("--pocket", help="Which pocket to use (valid values are: 'Release', 'Security', 'Updates', 'Proposed', 'Backports')", metavar="POCKET", action='store', default=None)

parser.add_option("--superseded-name", help="Name of superseded source package", metavar="SRCPKG", action='store')

parser.add_option("--superseded-version", help="Version of superseded files", metavar="NAME", action='store')

parser.add_option("--debug", help="Show debug output", action='store_true')

parser.add_option("--verbose", help="Verbose output", action='store_true')

parser.add_option("--uri", help="Use specific URI for API", action='store', default=None, metavar="URI")

parser.add_option("--beta", help="Use beta API instead of 1.0 LP API", action='store_true', default=False)

parser.add_option("-r", "--release", help="Limit to a specific set of comma-separate releases", metavar="SERIES", action='store', default=None)

parser.add_option("--skip-build-check", help="Skip binary package build check", action='store_true', default=False)

# Action-specific options

#   'changes'

parser.add_option("--dsc", help="Toggle fetching source .dsc files (default is True)", action='store_false', default=True)

#   'binaries'

parser.add_option("--arch", help="Limit 'binaries' and 'changes' action to comma-separated list of archs", metavar="ARCH[,ARCH...]", action='store')

parser.add_option("--re", help="When handling binaries, only include those matching this regular expression", metavar="RE", action='store')

parser.add_option("--include-debug", help="When handling binaries, skip .udeb, -dbg, -dbgsym and non-English -locale packages", action='store_true', default=False)

#   'changes', 'binaries', 'source'

parser.add_option("--download", help="Download to DIR", metavar="DIR", action='store', default='')

parser.add_option("--force-download", help="Force download to DIR if it exists (removes old DIR)", action='store_true', default=False)

#   'source'

parser.add_option("--fetch-orig", help="Download the orig.tar.gz when fetching source", action='store_true', default=False)

#   'include-devel'

parser.add_option("--include-devel", help="Include development release", action='store_true', default=False)

parser.add_option("--include-eol", help="Include end of life releases", action='store_true', default=False)

parser.add_option("--distribution", help="Distribution to use (eg, 'ubuntu-rtm')", metavar="DIST", action='store', default=None)

(opt, args) = parser.parse_args()

# Load configuration

cve_lib.read_config()

# API interface

lp = lpl_common.connect(beta=opt.beta, uri=opt.uri)

# Get authenticated URL fetcher

opener = lpl_common.opener_with_cookie(cve_lib.config["plb_authentication"])

if not opener:

    raise ValueError("Could not open cookies")

if len(args) < 1 and not opt.superseded_name:

    print("Usage: %s [--download <dir>] SRCPKG" % (sys.argv[0]), file=sys.stderr)

    sys.exit(1)

serieses = []

if opt.release:

    for r in opt.release.split(','):

        serieses.append(r.lower())

download_dir = ""

if opt.download:

    if opt.download == '':

        print("Must specify a directory with '--download'", file=sys.stderr)

        sys.exit(1)

    else:

        download_dir = opt.download

        if os.path.exists(download_dir):

            if opt.force_download:

                cve_lib.recursive_rm(download_dir)

            else:

                print("Specified download directory exists:\n %s" % (download_dir), file=sys.stderr)

                print("\nPlease remove (or use --force-download) and try again.", file=sys.stderr)

                sys.exit(1)

        tmpdir = tempfile.mkdtemp(prefix='sis-changes-download-')

# split_package -> pkg_name, arch

def split_package(pkg):

    tmp = pkg.split('_')

    arch = tmp[-1].split('.')[0]

    pkg_name = tmp[-3].split('/')[-1]

    return (pkg_name, arch)

def get_arch_from_dsc(dsc):

    f = open(dsc, 'r')

    for line in f:

        vals = line.split(':')

        if len(vals) == 2 and vals[0] == 'Architecture':

            f.close()

            return vals[1].strip()

    f.close()

    return None

# pkg -> { release, release -> { version } }

def load_pkg_details_from_lp(pkgs, pkg, item):

    rel = item.distro_series.name

    if opt.debug:

        print("Processing %s" % (rel), file=sys.stderr)

    if opt.distribution is None and rel not in cve_lib.releases:

        raise ValueError("Unknown release '%s':\n" % (rel))

    if serieses and rel not in serieses:

        if opt.debug:

            print("Skipping %s: not in %s" % (rel, serieses), file=sys.stderr)

        return

    version = item.source_package_version

    if opt.superseded_version and version != opt.superseded_version:

        print("Skipping %s: %s %s (we need %s)" % (rel, pkg, version, opt.superseded_version), file=sys.stderr)

        return

    if pkg in pkgs and rel in pkgs[pkg]:

        state = version_compare(version, pkgs[pkg][rel]['source']['version'])

        if state < 0:

            print("Skipping %s: %s %s (already have %s)" % (rel, pkg, version, pkgs[pkg][rel]['source']['version']), file=sys.stderr)

            return

        elif state == 0:

            if opt.verbose:

                print("Skipping %s: %s %s (same as %s)" % (rel, pkg, version, pkgs[pkg][rel]['source']['version']), file=sys.stderr)

            return

        else:

            print("Forgetting %s: %s %s (now have %s)" % (rel, pkg, pkgs[pkg][rel]['source']['version'], version), file=sys.stderr)

            pkgs[pkg][rel] = dict()

    pkgs.setdefault(pkg, dict())

    pkgs[pkg].setdefault(rel, dict())

    if opt.debug:

        print("Source(%s): %s %s" % (rel, pkg, version), file=sys.stderr)

    # Source details

    pkgs[pkg][rel].setdefault('source', dict())

    pkgs[pkg][rel]['source'].setdefault('version', version)

    # Handle transition to method (LP: #474876)

    if hasattr(item, 'changes_file_url'):

        src_changes = item.changes_file_url

    else:

        src_changes = item.changesFileUrl()

    pkgs[pkg][rel]['source'].setdefault('changes', src_changes)

    if opt.debug:

        print("Source(%s) changes: %s" % (rel, src_changes), file=sys.stderr)

    # Get per-build items

    build = None

    for build in item.getBuilds():

        arch = build.arch_tag

        pkgs[pkg][rel].setdefault(arch, dict())

        state = build.buildstate

        if opt.debug:

            print("Build(%s,%s) %s" % (rel, arch, state), file=sys.stderr)

        # Work around LP: #559591

        if state == 'Successful build':

            state = 'Successfully built'

        pkgs[pkg][rel][arch].setdefault('build_state', state)

        bin_changes = build.changesfile_url

        pkgs[pkg][rel][arch].setdefault('changes', bin_changes)

        if opt.debug:

            print("Build(%s,%s) changes: %s" % (rel, arch, bin_changes), file=sys.stderr)

        build_log = build.build_log_url

        pkgs[pkg][rel][arch].setdefault('build_log', build_log)

    # If we didn't find a build, we're in trouble.

    # This can happen if something was pocket-copied from a different release

    # See LP: #783613

    if build is None:

        raise ValueError("Could not find any builds for %s." % (pkg))

    # Diff (we don't use this yet...)

    # diff_url = item.packageDiffUrl()

    # pkgs[pkg][rel]['source'].setdefault('ancestor-diff', diff_url)

    # if opt.debug:

    #     print("Diff(%s) URL: %s" % (rel, diff_url), file=sys.stderr)

    # Binary outputs

    # Handle transition to method (LP: #474876)

    if hasattr(item, 'binary_file_url'):

        bin_files = item.binary_file_urls

    else:

        bin_files = item.binaryFileUrls()

    for file_url in bin_files:

        if file_url.endswith('deb'):

            name, arch = split_package(file_url)

            if opt.debug:

                print("Binary(%s,%s) URL: %s" % (rel, arch, file_url), file=sys.stderr)

            # hack for "all": attach to all_arch

            if arch == 'all':

                all_arch = cve_lib.get_all_arch(rel)

                # if only building for one arch that's not the default

                # all arch, the all packages will be built under that arch

                # so check the all_arch has binary pkgs

                archs = [x for x in pkgs[pkg][rel].keys() if x != 'source']

                if all_arch in archs:

                    arch = all_arch

                elif len(archs) == 1:

                    arch = archs[0]

                elif opt.debug:

                    print("Couldn't find 'all' arch for %s in %s/%s" % (name, arch, rel), file=sys.stderr)

            if not cve_lib.arch_is_valid_for_release(arch, rel):

                if opt.debug:

                    print("Skipping %s binary because %s is not a valid arch in %s" % (name, arch, rel), file=sys.stderr)

                continue

            pkgs[pkg][rel][arch].setdefault('binaries', dict())

            pkgs[pkg][rel][arch]['binaries'].setdefault(name, file_url)

        else:

            raise ValueError("Unknown downloadable binary file from %s %s '%s'" % (pkg, version, file_url))

    # Source inputs

    # Handle transition to method (LP: #474876)

    if hasattr(item, 'source_file_url'):

        src_files = item.source_file_urls

    else:

        src_files = item.sourceFileUrls()

    for file_url in src_files:

        if file_url.endswith('.dsc'):

            pkgs[pkg][rel]['source'].setdefault('dsc', file_url)

            if opt.debug:

                print("Source(%s) dsc URL: %s" % (rel, file_url), file=sys.stderr)

        elif re.search('\.(diff\.gz|debian\.tar\.(gz|bz2|lzma|xz))$', file_url):

            pkgs[pkg][rel]['source'].setdefault('diff', file_url)

            if opt.debug:

                print("Source(%s) debian differences URL: %s" % (rel, file_url), file=sys.stderr)

        elif re.search('\.tar\.(gz|bz2|lzma|xz)$', file_url):

            pkgs[pkg][rel]['source'].setdefault('orig', file_url)

            if opt.debug:

                print("Source(%s) orig URL: %s" % (rel, file_url), file=sys.stderr)

        elif file_url.endswith('.asc'):

            pkgs[pkg][rel]['source'].setdefault('asc', file_url)

            if opt.debug:

                print("Source(%s) asc URL: %s" % (rel, file_url), file=sys.stderr)

        else:

            raise ValueError("Unknown downloadable source file from %s %s '%s'" % (pkg, version, file_url))

    # Check that all built binaries have actually published into the PPA

    if opt.skip_build_check is True:

        print("WARNING: skipping binary publication check. Please check manually.", file=sys.stderr)

    else:

        for binary in item.getPublishedBinaries():

            if binary.status != 'Published':

                if opt.debug:

                    print("BinaryPublication(%s,%s,%s) state: %s" % (rel, binary.distro_arch_series.architecture_tag, binary.binary_package_name, binary.status), file=sys.stderr)

                arch = binary.distro_arch_series.architecture_tag

                # Override binary target in the case of "all"

                if 'all' in pkgs[pkg][rel]:

                    arch = 'all'

                pkgs[pkg][rel][arch]['build_state'] = 'Binaries pending'

def is_debug_pkg(name):

    if name.endswith("-dbg") or name.endswith("-dbgsym") or name.endswith("-mozsymbols"):

        return True

    if name.endswith("-source") and name.startswith("openjdk-"):

        return True

    if ("-locale-" in name and not name.endswith("-locale-en") and

            (name.startswith("firefox") or name.startswith("thunderbird"))):

        return True

if opt.distribution is None:

    # We could default to this, but it would require changes elsewhere

    # distribution = lp.distributions['ubuntu']

    distribution = opt.distribution

else:

    distribution = lp.distributions[opt.distribution]

archive, group, ppa = lpl_common.get_archive(opt.ppa, lp, opt.debug, distribution=distribution)

pkgs = dict()

if opt.superseded_version:

    status = "Superseded"

else:

    status = "Published"

for pkg_name in args:

    params = dict(source_name=pkg_name,

                  exact_match=True,

                  status=status)

    if opt.pocket:

        params['pocket'] = opt.pocket

    for item in archive.getPublishedSources(**params):

        load_pkg_details_from_lp(pkgs, pkg_name, item)

if opt.action == 'changes':

    for pkg in args:

        if pkg not in pkgs:

            msg = "Source package '%s' not found in group %s PPA %s" % (pkg, group, ppa)

            if opt.pocket:

                msg += " pocket %s" % (opt.pocket)

            raise ValueError(msg)

        for rel in sorted(pkgs[pkg].keys()):

            if not opt.include_devel and rel == cve_lib.devel_release:

                print("Skipping '%s' (use --include-devel)" % (rel), file=sys.stderr)

                continue

            if not opt.include_eol and rel in cve_lib.eol_releases:

                print("Skipping '%s' (use --include-eol)" % (rel), file=sys.stderr)

                continue

            version = pkgs[pkg][rel]['source']['version']

            if ':' in version and not version.endswith(':'):

                # strip out epoch, if it exists

                version = version[(version.find(':') + 1):]

            if opt.debug:

                print("Fetching %s %s ..." % (pkg, version), file=sys.stderr)

            download_url(pkgs[pkg][rel]['source']['changes'])

            archs = sorted(pkgs[pkg][rel].keys())

            if opt.arch:

                archs = archlist = opt.arch.split(',')

            if opt.dsc:

                dsc = download_url(pkgs[pkg][rel]['source']['dsc'])

                dsc_arch = get_arch_from_dsc(dsc)

                if dsc_arch == 'all':

                    archs = [cve_lib.get_all_arch(rel)]

                    print("Skipping non-i386 builds for 'Architecture: all' package %s %s" % (pkg, rel), file=sys.stderr)

            for arch in archs:

                # Ignore 'source' and 'item' for build states

                if arch in ['source', 'item']:

                    continue

                if pkgs[pkg][rel][arch]['build_state'] != 'Successfully built':

                    print("Skipping '%s' build for %s %s %s" % (pkgs[pkg][rel][arch]['build_state'], pkg, rel, arch), file=sys.stderr)

                    continue

                download_url(pkgs[pkg][rel][arch]['changes'])

elif opt.action == 'binaries':

    for pkg in args:

        if pkg not in pkgs:

            raise ValueError("Source package '%s' not found in PPA" % (pkg))

        for rel in sorted(pkgs[pkg].keys()):

            if not opt.include_devel and rel == cve_lib.devel_release:

                print("Skipping '%s' (use --include-devel)" % (rel), file=sys.stderr)

                continue

            if not opt.include_eol and rel in cve_lib.eol_releases:

                print("Skipping '%s' (use --include-eol)" % (rel), file=sys.stderr)

                continue

            version = pkgs[pkg][rel]['source']['version']

            archlist = sorted(pkgs[pkg][rel].keys())

            if opt.arch:

                archlist = opt.arch.split(',')

            for arch in archlist:

                if arch not in pkgs[pkg][rel]:

                    continue

                if 'binaries' not in pkgs[pkg][rel][arch]:

                    continue

                for name in sorted(pkgs[pkg][rel][arch]['binaries'].keys()):

                    # If --include-debug is not specified, don't download:

                    # -dbg

                    # -dbgsym

                    # -mozsymbols

                    # non-english firefox-locale-*

                    # non-english thunderbird-locale-*

                    # openjdk-*-source

                    # .udeb

                    if not opt.include_debug and \

                            (is_debug_pkg(name) or

                             pkgs[pkg][rel][arch]['binaries'][name].endswith(".udeb")):

                        print("Skipping '%s' (use --include-debug)" % (name), file=sys.stderr)

                        continue

                    download_url(pkgs[pkg][rel][arch]['binaries'][name])

elif opt.action == 'buildlogs':

    for pkg in args:

        if pkg not in pkgs:

            raise ValueError("Source package '%s' not found in PPA" % (pkg))

        for rel in sorted(pkgs[pkg].keys()):

            if not opt.include_devel and rel == cve_lib.devel_release:

                print("Skipping '%s' (use --include-devel)" % (rel), file=sys.stderr)

                continue

            if not opt.include_eol and rel in cve_lib.eol_releases:

                print("Skipping '%s' (use --include-eol)" % (rel), file=sys.stderr)

                continue

            archlist = sorted(pkgs[pkg][rel].keys())

            if opt.arch:

                archlist = opt.arch.split(',')

            for arch in archlist:

                if opt.debug:

                    print("Fetching build log for %s %s %s ..." % (pkg, rel, arch), file=sys.stderr)

                if not cve_lib.arch_is_valid_for_release(arch, rel):

                    continue

                if arch not in pkgs[pkg][rel]:

                    continue

                if 'build_log' not in pkgs[pkg][rel][arch]:

                    continue

                download_url(pkgs[pkg][rel][arch]['build_log'])

elif opt.action == 'list':

    for pkg in sorted(pkgs.keys()):

        print(pkg)

elif opt.action == 'source':

    for pkg in args:

        if pkg not in pkgs:

            raise ValueError("Source package '%s' not found in PPA" % (pkg))

        for rel in sorted(pkgs[pkg].keys()):

            if not opt.include_devel and rel == cve_lib.devel_release:

                print("Skipping '%s' (use --include-devel)" % (rel), file=sys.stderr)

                continue

            if not opt.include_eol and rel in cve_lib.eol_releases:

                print("Skipping '%s' (use --include-eol)" % (rel), file=sys.stderr)

                continue

            version = pkgs[pkg][rel]['source']['version']

            download_url(pkgs[pkg][rel]['source']['dsc'])