postscriptdriver.prov

#!/usr/bin/python3

## Copyright (C) 2009-2020 Red Hat, Inc.
## Author: Tim Waugh <twaugh@redhat.com>

## This program is free software; you can redistribute it and/or modify
## it under the terms of the GNU General Public License as published by
## the Free Software Foundation; either version 2 of the License, or
## (at your option) any later version.

## This program is distributed in the hope that it will be useful,
## but WITHOUT ANY WARRANTY; without even the implied warranty of
## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
## GNU General Public License for more details.

## You should have received a copy of the GNU General Public License
## along with this program; if not, write to the Free Software
## Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.

import sys

try:
    import cups
    CAN_EXAMINE_PPDS = True
except:
    CAN_EXAMINE_PPDS = False

from getopt import getopt
import errno
import os
import posix
import re
import shlex
import signal
import subprocess
import sys
import tempfile

if len (sys.argv) > 1:
    RPM_BUILD_ROOT = sys.argv[1]
else:
    RPM_BUILD_ROOT = os.environ.get ("RPM_BUILD_ROOT")

class TimedOut(Exception):
    def __init__ (self):
        Exception.__init__ (self, "Timed out")

class DeviceIDs:
    def __init__ (self):
        self.ids = dict()

    def get_dict (self):
        return self.ids

    def get_tags (self):
        ret = []
        for mfg, mdlset in self.ids.items ():
            mfgl = mfg.lower ().replace (" ", "_")
            for mdl in mdlset:
                mdll = mdl.lower ().replace (" ", "_")
                ret.append ("postscriptdriver(%s;%s;)" % (mfgl,
                                                          mdll))

        return ret

    def __add__ (self, other):
        if isinstance(other, DeviceIDs):
            for omfg, omdlset in other.ids.items ():
                try:
                    mdlset = self.ids[omfg]
                except KeyError:
                    mdlset = set()
                    self.ids[omfg] = mdlset

                mdlset.update (omdlset)

            return self

        pieces = other.split (';')
        mfg = mdl = None
        for piece in pieces:
            s = piece.split (":")
            if len (s) != 2:
                continue
            key, value = s
            key = key.upper ().strip ()
            if key in ["MFG", "MANUFACTURER"]:
                mfg = value.strip ()
            elif key in ["MDL", "MODEL"]:
                mdl = value.strip ()

        if mfg and mdl:
            try:
                mdlset = self.ids[mfg]
            except KeyError:
                mdlset = set()
                self.ids[mfg] = mdlset

            mdlset.add (mdl)

        return self

class Driver:
    def __init__ (self):
        self.ids = DeviceIDs()

    def list (self):
        return self.ids

class PPDDriver(Driver):
    def __init__ (self, pathname=None):
        Driver.__init__ (self)
        self.pathname = pathname

    def list (self):
        if self.pathname != None:
            self.examine_file (self.pathname)

        return Driver.list (self)

    def examine_file (self, path):
        try:
            ppd = cups.PPD (path)
        except RuntimeError:
            # Not a PPD file.  Perhaps it's a drv file.
            drv = DrvDriver (path)
            self.ids += drv.list ()
            return

        attr = ppd.findAttr ('1284DeviceID')
        while attr:
            self.ids += attr.value
            attr = ppd.findNextAttr ('1284DeviceID')

class DynamicDriver(Driver):
    def __init__ (self, driver):
        Driver.__init__ (self)
        self.driver = driver
        signal.signal (signal.SIGALRM, self._alarm)

    def _alarm (self, sig, stack):
        raise TimedOut

    def list (self):
        signal.alarm (180)
        env = os.environ.copy ()
        if RPM_BUILD_ROOT:
            buildroot = RPM_BUILD_ROOT
            if not buildroot.endswith (os.path.sep):
                buildroot += os.path.sep

            env["DESTDIR"] = RPM_BUILD_ROOT
            env["LD_LIBRARY_PATH"] = "%susr/lib64:%susr/lib" % (buildroot,
                                                                buildroot)

        p = subprocess.Popen ([self.driver, "list"],
                              stdout=subprocess.PIPE,
                              stderr=subprocess.PIPE,
                              env=env)
        try:
            (stdout, stderr) = p.communicate ()
            signal.alarm (0)
        except TimedOut:
            posix.kill (p.pid, signal.SIGKILL)
            raise

        if stderr:
            print(stderr.decode (), file=sys.stderr)

        ppds = []
        lines = stdout.decode ().split ('\n')
        for line in lines:
            l = shlex.split (line)
            if len (l) < 5:
                continue
            self.ids += l[4]

        return Driver.list (self)

class DrvDriver(PPDDriver):
    def __init__ (self, pathname):
        PPDDriver.__init__ (self)
        self.drv = pathname

    def _alarm (self, sig, stack):
        raise TimedOut

    def list (self):
        tmpdir = os.environ.get ("TMPDIR", "/tmp") + os.path.sep
        outputdir = tempfile.mkdtemp (dir=tmpdir)
        
        argv = [ "ppdc",
                 "-d", outputdir,
                 "-I", "/usr/share/cups/ppdc",
                 self.drv ]

        signal.alarm (60)
        try:
            p = subprocess.Popen (argv,
                                  stdout=subprocess.PIPE,
                                  stderr=subprocess.PIPE)
        except OSError:
            # ppdc not available.
            os.rmdir (outputdir)
            return Driver.list (self)

        try:
            (stdout, stderr) = p.communicate ()
            signal.alarm (0)
        except TimedOut:
            posix.kill (p.pid, signal.SIGKILL)
            raise

        for (dirpath, dirnames, filenames) in os.walk (outputdir,
                                                       topdown=False):
            for filename in filenames:
                path = dirpath + os.path.sep + filename
                self.examine_file (path)
                try:
                    os.unlink (path)
                except OSError:
                    pass

            for dirname in dirnames:
                try:
                    os.rmdir (dirpath + os.paht.sep + dirname)
                except OSError:
                    pass

        try:
            os.rmdir (outputdir)
        except OSError:
            pass

        return Driver.list (self)

class TagBuilder:
    def __init__ (self, filelist=None):
        if filelist == None:
            filelist = sys.stdin

        paths = [x.rstrip () for x in filelist.readlines ()]
        self.ids = DeviceIDs ()

        for path in paths:
            if path.find ("/usr/lib/cups/driver/") != -1 and \
               path.find("driverless") == -1:
                try:
                    self.ids += DynamicDriver (path).list ()
                except TimedOut:
                    pass
                except OSError as exc:
                    (e, s) = exc.args
                    if e == errno.EACCES or e == errno.ENOENT:
                        # Not executable
                        pass
                    else:
                        raise

        if CAN_EXAMINE_PPDS:
            for path in paths:
                try:
                    self.ids += PPDDriver (path).list ()
                except TimedOut:
                    pass

    def get_tags (self):
        return self.ids.get_tags ()

if __name__ == "__main__":
    builder = TagBuilder ()
    tags = builder.get_tags ()
    for tag in tags:
        print (tag)