rpm2tmpfiles

#!/usr/bin/python
# Copyright (c) 2018,2019 SUSE LLC
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.

import argparse
import logging
import rpm
import stat
import os
import sys
from glob import glob
from pprint import pprint

class TmpFileEntry(object):
    def __init__(self, tYpe, path, mode = None, uid = None, gid = None, age = None, argument = None):
        self._settings = [tYpe, path, mode, uid, gid, age, argument]
        self.file = None
        for i in range(0, len(self._settings)):
            if self._settings[i] is None:
                continue
            if self._settings[i] == '-':
                self._settings[i] = None
            elif i in (3, 4) and self._settings[i] == 'root':
                self._settings[i] = None

    @property
    def type(self):
        return self._settings[0]

    @property
    def path(self):
        return self._settings[1]

    @property
    def mode(self):
        return self._settings[2]

    @property
    def uid(self):
        return self._settings[3]

    @property
    def gid(self):
        return self._settings[4]

    @property
    def age(self):
        return self._settings[5]

    @property
    def argument(self):
        return self._settings[6]

    def __str__(self):
        last = 2
        for i in range(len(self._settings), 0, -1):
            if self._settings[i-1] is not None:
                last = i
                break

        l = []
        for i in range(0, last):
            l.append(self._settings[i] if self._settings[i] else '-')
        return ' '.join(l)

    def equal(self, b):
        for i in range(0, len(self._settings)):
            if self._settings[i] != b._settings[i]:
                return False

        return True

def b2s(s):
    return s.decode(encoding='UTF-8', errors='replace')

def readtmpfiles(tmpfilesdir):
    entries = {}
    for fn in glob(tmpfilesdir+'/*.conf'):
        with open(fn, 'r') as fh:
            for line in fh.readlines():
                line = line.strip()
                if line == "" or line.startswith('#'):
                    continue
                try:
                    t = TmpFileEntry(*line.split(None, 6))
                except Exception as e:
                    logging.error("%s: %s", fn, line)
                    raise
                t.file = fn
                if t.path in entries:
                    logging.warn("%s: %s", os.path.basename(t.file), t)
                    e = entries[t.path]
                    lvl = logging.WARN
                    if not t.equal(e):
                        lvl = logging.ERROR
                    logging.log(lvl, "%s duplicate: %s", os.path.basename(e.file), e)
                else:
                    entries[t.path] = t
    return entries


def main(args):

    tmpfiles = {}
    if args.tmpfiles:
        tmpfiles = readtmpfiles(args.tmpfiles)

    for name in args.package:
        ts = rpm.TransactionSet()
        mi = ts.dbMatch('name', name)
        if not mi:
            raise KeyError(name)
        try:
            hdr = next(mi)
        except StopIteration:
            raise KeyError(name)

        for i in range(0, len(hdr[rpm.RPMTAG_FILENAMES])):
            fn = b2s(hdr[rpm.RPMTAG_FILENAMES][i])
            mode = hdr[rpm.RPMTAG_FILEMODES][i]
            user = b2s(hdr[rpm.RPMTAG_FILEUSERNAME][i])
            group = b2s(hdr[rpm.RPMTAG_FILEGROUPNAME][i])
            if not args.path or fn.startswith(args.path):
                if stat.S_ISDIR(mode):
                    entry = TmpFileEntry('d', fn, '{:04o}'.format(mode&0o7777), user, group)
                elif stat.S_ISLNK(mode):
                    entry = TmpFileEntry('L', fn, None, None, None, None, b2s(hdr[rpm.RPMTAG_FILELINKTOS][i]))

                if entry.path in tmpfiles:
                    logging.warn("%s: %s", name+".rpm", entry)
                    e = tmpfiles[entry.path]
                    lvl = logging.WARN
                    if not entry.equal(e):
                        lvl = logging.ERROR
                    logging.log(lvl, "%s duplicate: %s", os.path.basename(tmpfiles[entry.path].file), tmpfiles[entry.path])
                else:
                    print(entry)

    return 0

if __name__ == '__main__':
    parser = argparse.ArgumentParser(description='boilerplate python commmand line program')
    parser.add_argument("--dry", action="store_true", help="dry run")
    parser.add_argument("--debug", action="store_true", help="debug output")
    parser.add_argument("--verbose", action="store_true", help="verbose")
    parser.add_argument("--path", metavar="directory", help="path prefix to handle")
    parser.add_argument("--tmpfiles", metavar="directory", help="directory with tmpfiles to compare against")
    parser.add_argument("package", nargs='*', help="some file name")

    args = parser.parse_args()

    if args.debug:
        level  = logging.DEBUG
    elif args.verbose:
        level = logging.INFO
    else:
        level = None

    logging.basicConfig(level=level, format='[%(levelname).1s] %(message)s')

    sys.exit(main(args))