~ubuntu-security/ubuntu-security-tools/trunk

#!/usr/bin/python3

# Name:        debcompare
# Description: Compares different versions of binary packages
#              and enumerates changes. Useful for security
#              updates.
#
# Copyright (C) 2008-2014 Canonical Ltd.
# Authors:
#    Marc Deslauriers <marc.deslauriers@canonical.com>
#
# Based on original shell debcompare by:
#    Marc Deslauriers <marc.deslauriers@canonical.com>
#    Kees Cook <kees@ubuntu.com>
#    Jamie Strandboge <jamie@canonical.com>
#
# License: GPLv3
#

import difflib
import os
import re
import shutil
import signal
import subprocess
import sys
import tempfile

class DebFile:

    def __init__(self, debfile):

        self.deb_files = {}
        self.control_files = {}
        self.debfile = debfile
        self.lintian = None
        self.jarfiles = []

        self.temp_dir = tempfile.mkdtemp(prefix="debcompare-")
        self.files_dir = os.path.join(self.temp_dir, 'files')
        self.control_dir = os.path.join(self.temp_dir, 'control')

        self._parse_deb()

    def _get_md5(self, filename):
        '''Gets the md5sum of the file specified'''

        report = subprocess.check_output(["md5sum", "-b", filename])
        return report.split(b' ')[0].decode()

    def _get_filetype(self, filename):
        '''Gets the filetype of the file specified'''

        report = subprocess.check_output(["file", '-b', filename])
        report = report.decode().strip()

        # normalizes BuildID[sha1]=34ea62c7aad8f0e2ae79698bd360c819fbfc38a3,
        report = re.sub('BuildID\[sha1\]=[a-f0-9]{40}, ', '', report)

        # normalizes last modified: Mon Dec 23 11:39:49 2013,
        report = re.sub('last modified:.*, ', '', report)

        return report

    def _get_libdeps(self, filename):
        '''Gets the library dependencies of the file specified'''

        libdeps = []
        report = subprocess.check_output(["objdump", "-p", filename])

        for line in report.splitlines():
            if b'NEEDED' in line:
                libdeps.append(line.split()[1].decode().strip())

        return libdeps

    def _get_symbols(self, filename):
        '''Gets the symbols of the file specified'''

        symbols = []
        report = subprocess.check_output(["nm", "-PDC", filename])

        for line in report.splitlines():
            symbol = ' '.join(line.decode().split()[:2])
            symbols.append(symbol)

        return symbols

    def _extract_deb(self):
        '''Extracts a deb file into a directory'''

        subprocess.check_call(['dpkg-deb', '-x', self.debfile, self.files_dir])
        subprocess.check_call(['dpkg-deb', '-e', self.debfile, self.control_dir])

    def _run_lintian(self):
        '''Runs lintian on the deb file'''

        rc, report = cmd(["lintian", self.debfile])
        self.lintian = report.decode().splitlines()
        self.lintian.sort()

    def _get_file_list(self):
        '''Gets a list of files from a package'''

        out = subprocess.check_output(['dpkg-deb', '-c', self.debfile])

        for line in out.splitlines():
            file_details = {}
            file_details['perms'] = line.split()[0].decode()
            file_details['owner'] = line.split()[1].decode()
            file_details['size']  = line.split()[2].decode()
            file_details['date']  = "%s %s" % (line.split()[3].decode(),
                                               line.split()[4].decode())
            filename = " ".join(line.decode().split()[5:])  # handle files with spaces
            filename = filename.split(" -> ")[0]  # take only the source of symlinks
            # print(filename)
            self.deb_files[filename] = file_details

    def _analyze_directory(self):
        '''Analyzes files in the extracted directory'''

        for root, dirs, files in os.walk(self.files_dir):
            for filename in files:
                full_path = os.path.join(root, filename)
                self._analyze_file(full_path)

    def _analyze_control_files(self):
        '''Loads control files'''

        for root, dirs, files in os.walk(self.control_dir):
            for filename in files:
                # Skip md5sums as we compare the actual files later on
                if filename == "md5sums":
                    continue

                # TODO: special handling for 'control' that will parse it
                # and report depends changes better
                # TODO: special handling for 'conffiles' that will parse it
                # and report changes better

                full_path = os.path.join(root, filename)
                if os.path.islink(full_path):
                    continue
                contents = open(full_path, errors='replace').read()

                if filename == 'control':
                    contents = re.sub('\nVersion: .*\n',
                                      '\nVersion: <normalized>\n',
                                      contents)
                    contents = re.sub('\nInstalled-Size: .*\n',
                                      '\nInstalled-Size: <normalized>\n',
                                      contents)

                self.control_files[filename] = contents

    def _analyze_file(self, filename):
        '''Analyzes file in the extracted directory'''

        relative_name = filename.replace(self.files_dir, ".")
        filetype = self._get_filetype(filename)
        self.deb_files[relative_name]['filetype'] = filetype
        if os.path.islink(filename):
            return
        self.deb_files[relative_name]['md5'] = self._get_md5(filename)

        # If it's an ELF file, get library dependencies
        if "ELF" in filetype:
            self.deb_files[relative_name]['libdeps'] = self._get_libdeps(filename)

        # If it's a shared object, get symbols
        if "shared object" in filetype:
            # Not for debug packages though
            if not "-dbg_" in self.debfile:
                self.deb_files[relative_name]['symbols'] = self._get_symbols(filename)

        # If it's a jar file, add it to the list for further processing
        if "Zip archive" in filetype and filename.endswith('.jar'):
            self.jarfiles.append(filename)

    def _parse_deb(self):
        '''This extracts a deb file and fetches all the info about it'''

        self._get_file_list()
        self._extract_deb()
        self._analyze_directory()
        self._analyze_control_files()
        self._run_lintian()

# http://www.chiark.greenend.org.uk/ucgi/~cjwatson/blosxom/2009-07-02-python-sigpipe.html
# This is needed so that the subprocesses that produce endless output
# actually quit when the reader goes away.
def subprocess_setup():
    # Python installs a SIGPIPE handler by default. This is usually not what
    # non-Python subprocesses expect.
    signal.signal(signal.SIGPIPE, signal.SIG_DFL)

def cmd(command, input = None, stderr = subprocess.STDOUT,
        stdout = subprocess.PIPE, stdin = None, timeout = None):
    '''Try to execute given command (array) and return its stdout, or return
    a textual error if it failed.'''

    try:
        sp = subprocess.Popen(command, stdin=stdin, stdout=stdout,
                              stderr=stderr, close_fds=True,
                              preexec_fn=subprocess_setup)
    except OSError as e:
        return [127, str(e)]

    out, outerr = sp.communicate(input)
    # Handle redirection of stdout
    if out is None:
        out = b''
    # Handle redirection of stderr
    if outerr is None:
        outerr = b''

    return [sp.returncode,out+outerr]

def print_usage():
    '''Print usage'''
    print("Usage: debcompare <old file> <new file>", file=sys.stdout)

def check_required_tools():
    '''Checks if the required tools are installed'''

    # binary names, and package names
    tools = { 'dpkg-deb'  : 'dpkg',
              'nm'        : 'binutils',
              'diff'      : 'diffutils',
              'debdiff'   : 'devscripts',
              'lintian'   : 'lintian',
              'jardiff'   : 'jardiff' }

    missing_tools = []
    for tool in tools:
        ret = subprocess.call(['which', tool],
                              stdout=subprocess.DEVNULL,
                              stderr=subprocess.DEVNULL)
        if ret != 0:
            missing_tools += [tool]

    if missing_tools != []:
        print("Some tools are required for debcompare2 to function correctly, but\n" +
              "could not be found:\n\n%s" % " ".join(missing_tools), file=sys.stderr)

        print("\nPlease install the missing tools with the following command\n" +
              "and try again:\n", file=sys.stderr)

        package_list = set()
        for tool in missing_tools:
            package_list.add(tools[tool])

        print("sudo apt-get install %s" % " ".join(package_list), file=sys.stderr)
        sys.exit(1)

def print_diff(old, new, context=True, add_wdiff=False):
    if old == new:
        print("Contents are identical.")
    else:
        for line in difflib.unified_diff(old, new, lineterm=''):
            if line.startswith('--- '):
                continue
            if line.startswith('+++ '):
                continue
            if line.startswith('@@ '):
                continue
            if context == False:
                if not line.startswith('-') and not line.startswith('+'):
                    continue
            print(line)
        if add_wdiff:
            # difflib.unified_diff returns a generator, which means we
            # need to regenerate the difference list
            diff_lines = difflib.unified_diff(old, new, n=0, lineterm='')
            (rc, out) = cmd(['wdiff', '-d'], input=bytes('\n'.join(diff_lines), 'UTF-8'), stdin=subprocess.PIPE)
            print('\nwdiff alternate comparison:\n')
            for line in out.decode().splitlines():
                if line.startswith('[----]{++++}'):
                    continue
                if line.startswith('@@ '):
                    continue
                print(line)

def compare_lintian(old, new):
    print("\nlintian report:")
    print("---------------")

    print_diff(old, new, context=False)

def find_extra_files(old, new):
    '''finds files that are in new, but not in old'''
    extra_files = []
    for filename in new:
        if filename not in old:
            extra_files.append(filename)

    extra_files.sort()
    return extra_files

def find_different_libdeps(old, new):
    '''finds files that have different library dependencies'''
    different_libdeps = []
    for filename in old:
        if filename not in new:
            # Skip files that aren't common to both
            # TODO: should we be skipping this for libdeps also?
            continue
        if 'libdeps' not in old[filename]:
            continue
        # sort data structures now so we don't have to do it later
        old[filename]['libdeps'].sort()
        old_libdeps = old[filename]['libdeps']
        new[filename]['libdeps'].sort()
        new_libdeps = new[filename]['libdeps']
        if old_libdeps != new_libdeps:
            different_libdeps.append(filename)

    different_libdeps.sort()
    return different_libdeps

def find_different_symbols(old, new):
    '''finds files that have different symbols'''
    different_symbols = []
    for filename in old:
        if filename not in new:
            # Skip files that aren't common to both
            # TODO: should we be skipping this for symbols also?
            continue
        if 'symbols' not in old[filename]:
            continue
        # sort data structures now so we don't have to do it later
        old[filename]['symbols'].sort()
        old_symbols = old[filename]['symbols']
        new[filename]['symbols'].sort()
        new_symbols = new[filename]['symbols']
        if old_symbols != new_symbols:
            different_symbols.append(filename)

    different_symbols.sort()
    return different_symbols

def find_different_permissions(old, new):
    '''finds files that have different permissions'''
    different_perms = []
    for filename in old:
        if filename not in new:
            # Skip files that aren't common to both
            continue
        if ((old[filename]['perms'] != new[filename]['perms']) or
            (old[filename]['owner'] != new[filename]['owner'])):
            different_perms.append(filename)

    different_perms.sort()
    return different_perms

def find_different_hashes(old, new):
    '''finds files that have different hashes'''
    different_hashes = []
    for filename in old:
        if filename not in new:
            # Skip files that aren't common to both
            continue
        if 'md5' not in old[filename]:
            # Skip directories
            continue
        if filename.endswith('/changelog.Debian.gz'):
            # Ignore changelog files
            continue
        if old[filename].get('md5', '') != new[filename].get('md5', ''):
            different_hashes.append(filename)

    different_hashes.sort()
    return different_hashes

def find_different_filetypes(old, new):
    '''finds files that have different filetypes'''
    different_filetypes = []
    for filename in old:
        if filename not in new:
            # Skip files that aren't common to both
            continue
        if 'filetype' not in old[filename]:
            # Skip directories
            continue
        if old[filename]['filetype'] != new[filename]['filetype']:
            different_filetypes.append(filename)

    different_filetypes.sort()
    return different_filetypes

def print_file(filename, file_props, prefix=''):
    '''Formats and prints a file's properties'''

    print("%s %s %s %10s %s %s" % ( prefix,
                                    file_props['perms'],
                                    file_props['owner'],
                                    file_props['size'],
                                    file_props['date'],
                                    filename ))


def compare_files(old, new):
    '''Compare files from two packages'''

    print("\nFile owners/permissions/contents:")
    print("---------------------------------")

    identical = True

    extra_files = find_extra_files(old, new)
    if extra_files != []:
        identical = False
        print("\nThe new package has the following extra files:\n")
        for filename in extra_files:
            print_file(filename, new[filename])

    missing_files = find_extra_files(new, old)
    if missing_files != []:
        identical = False
        print("\nThe new package is missing the following files:\n")
        for filename in missing_files:
            print_file(filename, old[filename])

    different_perms = find_different_permissions(old, new)
    if different_perms != []:
        identical = False
        print("\nPermissions have changed on the following files:\n")
        for filename in different_perms:
            print_file(filename, old[filename], prefix='old:')
            print_file(filename, new[filename], prefix='new:')

    different_hashes = find_different_hashes(old, new)
    if different_hashes != []:
        identical = False
        print("\nContents have changed in the following files:\n")
        for filename in different_hashes:
            print(filename)

    different_filetypes = find_different_filetypes(old, new)
    if different_filetypes != []:
        identical = False
        print("\nFiletypes have changed on the following files:")
        for filename in different_filetypes:
            print("\n%s:" % filename)
            print("old filetype: %s" % old[filename]['filetype'])
            print("new filetype: %s" % new[filename]['filetype'])

    different_libdeps = find_different_libdeps(old, new)
    if different_libdeps != []:
        identical = False
        print("\nLibrary dependencies have changed in the following files:")
        for filename in different_libdeps:
            print("\n%s:" % filename)
            print_diff(old[filename]['libdeps'], new[filename]['libdeps'])

    different_symbols = find_different_symbols(old, new)
    if different_symbols != []:
        identical = False
        print("\nSymbols have changed in the following files:")
        for filename in different_symbols:
            print("\n%s:" % filename)
            print_diff(old[filename]['symbols'],
                       new[filename]['symbols'],
                       False)

    if identical == True:
        print("File contents are identical.")

def compare_jars(old, old_dir, new, new_dir):
    '''Use jardiff to compare files'''

    # We should probably reimplement this without using jardiff
    if old == []:
        return

    old.sort()
    for filename in old:
        if filename not in new:
            # Skip files that aren't common to both
            continue

        print("\nJar api differences for %s:\n" % filename)

        rc, report = cmd(["jardiff",
                          "-f", os.path.join(old_dir, old),
                          "-t", os.path.join(new_dir, new),
                          "o", "text"])
        print(report.decode())

def compare_control(old, new):
    '''Compare control files'''

    print("\nControl files:")
    print("--------------")

    identical = True

    missing = []
    for filename in old:
        if filename not in new:
            missing.append(filename)
    if missing != []:
        identical = False
        missing.sort()
        print("\nThe new package is missing the following control files:\n")
        for filename in missing:
            print(filename)

    extra = []
    for filename in new:
        if filename not in old:
            extra.append(filename)
    if extra != []:
        identical = False
        extra.sort()
        print("\nThe new package has the following new control files:\n")
        for filename in extra:
            print(filename)

    for filename in old:
        if filename not in new:
            # Skip files that aren't common to both
            continue
        old_split = old[filename].splitlines()
        new_split = new[filename].splitlines()

        # It makes sense to sort these before comparing them
        if filename in ['shlibs', 'conffiles']:
            old_split.sort()
            new_split.sort()

        if old_split != new_split:
            identical = False
            print("\nModified control file '%s':\n" % filename)
            if filename in ['control']:
                print_diff(old_split, new_split, False, add_wdiff=True)
            else:
                print_diff(old_split, new_split)

    if identical == True:
        print("Control files are identical.")

def compare_debs(old_deb, new_deb):
    '''Compare two packages'''

    vs_string = "%s vs %s" % (os.path.basename(old_deb.debfile),
                              os.path.basename(new_deb.debfile))
    num_stars = len(vs_string)


    print('# vim: set filetype=diff\n')
    print('*' * num_stars)
    print('Report for deb:')
    print(vs_string)
    print('*' * num_stars)

    compare_lintian(old_deb.lintian, new_deb.lintian)
    compare_control(old_deb.control_files, new_deb.control_files)
    compare_files(old_deb.deb_files, new_deb.deb_files)
    compare_jars(old_deb.jarfiles, old_deb.files_dir,
                 new_deb.jarfiles, new_deb.files_dir)

if len(sys.argv) < 3:
    print_usage()
    sys.exit(1)

oldfile = os.path.abspath(sys.argv[1])
newfile = os.path.abspath(sys.argv[2])

if not os.path.isfile(oldfile) or not os.path.isfile(newfile):
    print_usage()
    sys.exit(1)

check_required_tools()

old_deb = DebFile(oldfile)
new_deb = DebFile(newfile)

compare_debs(old_deb, new_deb)

# Clean up temp dirs
if old_deb.temp_dir and os.path.exists(old_deb.temp_dir):
    shutil.rmtree(old_deb.temp_dir, ignore_errors=True)
if new_deb.temp_dir and os.path.exists(new_deb.temp_dir):
    shutil.rmtree(new_deb.temp_dir, ignore_errors=True)

print("\n\n--------------")
print("End of report.")