calculate diff between file system state and installed packages - reproducible Arch - work in progress!

mtree.py

# coding: utf-8
'''
`man mtree`:

    Signature
        The first line of any mtree file must begin with “#mtree”.  If a file
        contains any full path entries, the first line should begin with
        “#mtree v2.0”, otherwise, the first line should begin with
        “#mtree v1.0”.
    Blank
        Blank lines are ignored.
    Comment
        Lines beginning with # are ignored.
    Special
        Lines beginning with / are special commands that influence the
        interpretation of later lines.
    Relative
        If the first whitespace-delimited word has no / characters, it is the
        name of a file in the current directory.  Any relative entry that
        describes a directory changes the current directory.
    dot-dot
        As a special case, a relative entry with the filename .. changes the
        current directory to the parent directory.  Options on dot-dot entries
        are always ignored.
    Full
        If the first whitespace-delimited word has a / character after the
        first character, it is the pathname of a file relative to the starting
        directory.  There can be multiple full entries describing the same file.
'''

import re
import os
import gzip
from glob import glob
import hashlib


def parse_keyword(word):
    key, sep, value = word.partition('=')
    return key.strip(), value.strip()


OCTALS_REFS = re.compile(r'\\([0-7][0-7][0-7])')

def octal_match_to_char(octal_match):
    return chr(int(octal_match.group(1), base=8))

def parse_path(word):
    return OCTALS_REFS.sub(octal_match_to_char, word)

assert parse_path(r'\033') == '\x1b'


open_mtree = gzip.open

def get_type(keywords):
    return keywords.get('type')

def parse_mtree(file_name, root='/'):
    global_keywords = {}

    with open_mtree(file_name) as mtree_file:
        header = next(mtree_file).lstrip()
        assert header.startswith('#mtree'), header
        for line in mtree_file:
            line = line.lstrip()
            if not line:
                pass
            elif line.startswith('#'):
                # comment
                pass
            else:
                words = line.split()
                first_word = words[0]
                parsed_keywords = dict(parse_keyword(word) for word in words[1:])
                if first_word == '/set':
                    global_keywords.update(parsed_keywords)
                elif first_word == '/unset':
                    for key in parsed_keywords and key in global_keywords:
                        del global_keywords[key]
                else:
                    keywords = global_keywords.copy()
                    keywords.update(parsed_keywords)
                    path = parse_path(first_word)
                    abspath = os.path.normpath(os.path.join(root, path))
                    if get_type(keywords) == 'dir':
                        if '/' not in path:
                            root = abspath
                    yield abspath, keywords


def read_all_mtrees():
    entries = {}
    for file_name in glob('/var/lib/pacman/local/*/mtree'):
        for path, keywords in parse_mtree(file_name):
            if path in entries:
                prev_type = get_type(entries[path])
                assert prev_type == get_type(keywords)
            entries[path] = keywords
    return entries


from pprint import pprint
# pprint(list(parse_mtree('mtree')))

def all_files():
    for dirpath, dirnames, filenames in os.walk('/'):
        for name in filenames + dirnames:
            yield os.path.join(dirpath, name)


SKIP_NEW = map(
    (lambda x: re.compile(x).search),
    (
        '^/home/', '^/tmp/',
        '^/dev/', '^/proc/', '^/sys/', '^/run/',
        '^/var/lib/pacman/', '^/var/cache/',
        # FIXME: package ca-certificates-utils
        '^/etc/ca-certificates/extracted/',
        # FIXME: package shared-mime-info
        '^/usr/share/mime/',
        # FIXME: package ca-certificates-utils, openssl
        '^/etc/ssl/certs/',
        # FIXME: ???
        '^/boot/EFI/BOOT/icons',
        # FIXME: package pacman-mirrorlist ?
        '^/etc/pacman.d/gnupg/',
        ))


def ignored_new(path):
    for filter in SKIP_NEW:
        if filter(path):
            return True

def get_hash(path, hash_class):
    if not os.path.isfile(path):
        return 'not a file'
    hash = hash_class()
    with open(path, 'rb') as f:
        hash.update(f.read())
    return hash.hexdigest().lower()

def type_eq(path, keywords):
    type = get_type(keywords)
    return (
        (type == 'file' and os.path.isfile(path)) or
        (type == 'dir'  and os.path.isdir(path)) or
        (type == 'link' and os.path.islink(path)))

def size_eq(path, keywords):
    assert os.path.isfile(path)
    return os.path.getsize(path) == int(keywords.get('size'))

def hash_eq(kwhash, realhash):
    if not kwhash:
        return True
    return kwhash.lower() == realhash

def same_as_installed(path, keywords):
    if not type_eq(path, keywords): return False
    if get_type(keywords) != 'file': return True
    return (
        size_eq(path, keywords) and
        hash_eq(keywords.get('md5digest'),    get_hash(path, hashlib.md5)) and
        hash_eq(keywords.get('sha256digest'), get_hash(path, hashlib.sha256)))

def progress(msg):
    print(msg)

def main():
    progress('reading mtrees')
    installed = read_all_mtrees()

    progress('reading files')
    real_files = set(all_files())

    new = sorted(
        path for path in real_files.difference(installed)
        if not ignored_new(path))
    pprint(new)
    print(len(new))

    missing = sorted(set(installed).difference(real_files))
    pprint(missing)
    print(len(missing))

    progress('verifying files')
    changed = sorted(
        path for path in real_files.intersection(installed)
        if not same_as_installed(path, installed[path]))
    pprint(changed)
    print(len(changed))

# new, missing, changed


if __name__ == '__main__':
    main()