python-foomatic 0.7.9.5build1 / usr / share / pyshared / foomatic / detect.py

#!/usr/bin/python
#
# detect.py - Autodetection routines for various printer interfaces
#
#     foomatic-gui - GNOME2 interface to the foomatic printing system
#     Copyright (C) 2002-04 Chris Lawrence <lawrencc@debian.org>
#
#     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., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
#
# $Id: detect.py,v 1.28 2010/05/23 00:08:14 lordsutch Exp $

# Some of the code here is derived(*) from code from the
# Common Unix Printing System, (C) 1997-2002 Easy Software Products
# (Licensed under GPL v2 exclusively.)

# (*) derived in the sense that I looked at the code to figure out how
# to implement autodetection for parallel and USB devices.

import os
import fcntl
import socket
import string
import urllib2
import urlparse
import xml.sax

import pysmb
from IPy import IP

# Suppress the damn annoying FutureWarning
import warnings
warnings.filterwarnings('ignore', category=FutureWarning)

USB_MAX = 16
NETWORK_MAX = 16
PARALLEL_MAX = 4
SERIAL_MAX = 8

# IOCTL definitions for LPIOC_GET_DEVICE_ID
_IOC_NRBITS = 8
_IOC_TYPEBITS = 8
_IOC_SIZEBITS = 14
_IOC_DIRBITS = 2

_IOC_NRSHIFT = 0
_IOC_TYPESHIFT = _IOC_NRSHIFT + _IOC_NRBITS
_IOC_SIZESHIFT = _IOC_TYPESHIFT + _IOC_TYPEBITS
_IOC_DIRSHIFT = _IOC_SIZESHIFT + _IOC_SIZEBITS

_IOC_NONE = 0
_IOC_WRITE = 1
_IOC_READ = 2

def _IOC(dir, otype, nr, size):
    return ((dir << _IOC_DIRSHIFT) | (ord(otype) << _IOC_TYPESHIFT) |
            (nr << _IOC_NRSHIFT) | (size << _IOC_SIZESHIFT))

IOCNR_GET_DEVICE_ID = 1
def LPIOC_GET_DEVICE_ID(length):
    return _IOC(_IOC_READ, 'P', IOCNR_GET_DEVICE_ID, length)

# This remaps the raw USB info to what is in foomatic-configure -O
usb_remap_key = {
    'mdl' : 'model', 'mfg' : 'manufacturer', 'cmd' : 'commandset',
    'des' : 'description', 'cls' : 'class',
    }
parallel_remap_key = {
    'command set' : 'commandset',
    }

def format_printinfo(printinfo):
    if printinfo.get('description'):
        desc = printinfo['description']
    elif 'manufacturer' in printinfo and 'model' in printinfo:
        if printinfo['model'].startswith(printinfo['manufacturer']):
            desc = printinfo['model']
        else:
            desc = '%(manufacturer)s %(model)s' % printinfo
    else:
        desc = 'Unknown'

    if 'sern' in printinfo:
        desc += ' (S/N: %(sern)s)' % printinfo

    return desc

def read_ieee1284_data(device):
    from struct import pack, unpack

    devstr = ' '*1024
    op = LPIOC_GET_DEVICE_ID(len(devstr))

    # Get this into the right (signed) format for sending to ioctl
    op = unpack('i', pack('I', op))[0]
    
    #print op, hex(op)

    try:
        fd = os.open(device, os.O_RDWR | os.O_EXCL)
    except:
        return

    printerinfo = {}
    if fd >= 0:
        try:
            ret = fcntl.ioctl(fd, long(op), devstr)
        except (OverflowError, IOError):
            # Work around #249113 and Savannah #10274
            return
        
        if ret:
            length = unpack('>h', ret[:2])[0]
            data = ret[2:length]
            if data:
                # Record the raw IEEE 1284 data
                printerinfo['ieee1284'] = data
                for entry in data.split(';'):
                    if not ':' in entry:
                        continue
                    key, val = entry.split(':', 1)
                    key = key.lower().strip()
                    printerinfo[usb_remap_key.get(key, key)] = val.strip()
    else:
        return

    os.close(fd)
    return printerinfo

def detect_usb_printers():
    conns = []
    # Figure out what special files are used on this system
    format = None
    int = 0
    for fmt in ('/dev/usb/lp%d', '/dev/usb/usblp%d', '/dev/usblp%d'):
	if os.path.exists(fmt % 0):
            format = fmt
            break
	else:
	    for i in range(USB_MAX):
		if os.path.exists(fmt % i):
	            format = fmt
		    int = i
		    break
	if format:
		break

    if not format:

        return conns

    for i in range(USB_MAX):
	if i < int:
		continue
        device = format % i
        if not os.path.exists(device): continue
        data = read_ieee1284_data(device)
        # Don't need to check device class here; that's done by the low-level
        # USB driver
        if isinstance(data, dict):
            conns.append( ('usb:'+device, data, 'USB Printer #%d' % (i+1),
                           format_printinfo(data)) )
## Disabled; if there's not a USB printer autodetected, it's not a printer...
##         else:
##             conns.append( ('usb:'+device, {},
##                            'USB Printer #%d' % (i+1),
##                            '(No printer detected)') )
    return conns

def read_parport_data(num):
    printinfo = {}
    for fmt in ('/proc/parport/%d/autoprobe',
                '/proc/sys/dev/parport/parport%d/autoprobe'):
        procfile = fmt % num
        if os.path.exists(procfile):
            probe = open(procfile)
            data = ''
            for line in probe:
                data += line
                line = line.strip()
                if line.endswith(';'):
                    line = line[:-1]
                key, val = line.split(':', 1)
                key = key.lower()
                printinfo[parallel_remap_key.get(key, key)] = val
            if data:
                printinfo['ieee1284'] = data
            return printinfo
    return None

def detect_parallel_printers():
    conns = []
    # Figure out what special files are used
    format = None
    for fmt in ('/dev/parallel/%d', '/dev/lp%d', '/dev/par%d',
                '/dev/printers/%d'):
        if os.path.exists(fmt % 0):
            format = fmt
            break
    if not format:
        return conns

    for i in range(PARALLEL_MAX):
        device = format % i
        if not os.path.exists(device): continue
        data = read_parport_data(i)
        if isinstance(data, dict) and \
               data.get('class', 'PRINTER') == 'PRINTER':
            conns.append( ('parallel:'+device, data,
                           'Parallel Printer #%d' % (i+1),
                           format_printinfo(data)) )
        else:
            conns.append( ('parallel:'+device, {},
                           'Parallel Printer #%d' % (i+1),
                           '(No printer detected)') )

    return conns

# No real detection, just list the found device special files we can open
def detect_serial_printers():
    conns = []
    for devfmt in ('/dev/ttyS%d', '/dev/usb/ttyUSB%d'):
        for i in range(SERIAL_MAX):
            device = devfmt % i
            if os.path.exists(device):
                try:
                    fd = os.open(device,
                                 os.O_WRONLY | os.O_NOCTTY | os.O_NDELAY)
                except OSError:
                    continue
                if fd >= 0:
                    os.close(fd)
                    conns.append( ('file:'+device, {},
                                   'Serial Port #%d' % (i+1),
                                   format_printinfo({})) )
    return conns

def detect_smb_printers():
    conns = []
    hosts = pysmb.get_host_list()
    for (host, hinfo) in hosts.iteritems():
        printers = pysmb.get_printer_list(hinfo)
        group = hinfo.get('GROUP')
        hostname = hinfo['NAME']
        for (printer, comment) in printers.iteritems():
            if group:
                smbname = 'smb://%s/%s/%s' % (group, hostname, printer)
            else:
                smbname = 'smb://%s/%s' % (hostname, printer)
            conns.append( (smbname, {}, 
                           'SMB Printer %s on %s' % (printer, hostname),
                           comment or 'Unknown') )
    return conns

CLASS_D = IP('224.0.0.0/3')
def get_local_hosts(compact=False):
    hosts = {}
    
    route = os.popen('/sbin/route -n', 'r')
    for line in route:
        line = line.rstrip()
        if line[0] not in string.digits: continue
        dest, gw, netmask, flags, metric, ref, use, iface = line.split()
        # Ignore ptp connections and route entries
        if 'H' in flags or 'G' in flags:
            continue
        addr = IP('%s/%s' % (dest, netmask))
        # Ignore multicast connections
        if addr in CLASS_D:
            continue
        
        if compact:
            hosts[str(addr)] = True
        else:
            for h in addr:
                hosts[str(h)] = True

    route.close()
    return hosts.keys()

class NMapXMLHandler(xml.sax.ContentHandler):
    def __init__(self):
        xml.sax.ContentHandler.__init__(self)
        self.hosts = {}
        self._data = ''
        self._ports = []
        self._addr = ''
        
    def characters(self, characters):
        self._data += characters

    def startElement(self, name, attrs):
        self._data = ''
        if name == 'host':
            self._addr = ''
            self._ports = []
            self._port = None
        elif name == 'address':
            if attrs['addrtype'] == 'ipv4':
                self._addr = attrs['addr']
        elif name == 'port':
            if attrs['protocol'] == 'tcp':
                self._port = int(attrs['portid'])
        elif name == 'state':
            if attrs['state'] == 'open' and self._port:
                self._ports.append(self._port)
                self._port = None

    def endElement(self, name):
        if name == 'host':
            if self._addr:
                self.hosts[self._addr] = self._ports

    def endDocument(self):
        del self._data
        del self._ports
        del self._addr

def detect_hosts(ports, hosts=None):
    if not hosts:
        hosts = get_local_hosts(compact=True)

    handler = NMapXMLHandler()
    pipe = os.popen('nmap -n -p %s -oX - %s 2>/dev/null' %
                    (','.join([str(x) for x in ports]), ' '.join(hosts)), 'r')
    try:
        xml.sax.parse(pipe, handler)
    except xml.sax.SAXParseException:
        return {}
    pipe.close()

    portinfo = {}
    for (host, ports) in handler.hosts.iteritems():
        for port in ports:
            portinfo.setdefault(port, []).append(host)
    return portinfo

def detect_lpd_printers(hosts=None):
    PORT = 515
    conns = []

    if not hosts:
        hosts = get_local_hosts()

    for host in hosts:
        for res in socket.getaddrinfo(host, PORT, socket.AF_UNSPEC,
                                      socket.SOCK_STREAM):
            af, socktype, proto, canonname, sa = res
            try:
                s = socket.socket(af, socktype, proto)
            except socket.error:
                s = None
                continue
            # Classic LPD requires a client port in this range
            for port in range(721, 732):
                for lres in socket.getaddrinfo(None, port, af, socktype):
                    local_sa = lres[-1]
                    try:
                        s.bind(local_sa)
                    except socket.error:
                        pass
            try:
                s.connect(sa)
            except socket.error:
                s.close()
                s = None
                continue
            break

        if s:
            cmd = '\x04 \n'
            s.sendall(cmd)
            while 1:
                data = s.recv(1024)
                if not data:
                    break
                print repr(data)
            s.close()

    return conns

def detect_ipp_queues(hosts):
    PORT=631
    conns = []
    for host in hosts:
        hport = '%s:%d' % (host, PORT)
        url = urlparse.urlunsplit( ('http', hport, '/', None, None) )
        req = urllib2.Request(url)
    
    return conns

def detect_jetdirect_queues(hosts):
    PORT = 9100
    conns = []
    for host in hosts:
        try:
            data = os.popen('pconf_detect -m NETWORK -i "%s"' % host, 'r')
        except (OSError, IOError):
            continue
        
        info = data.read().strip()
        data.close()
        pinfo = {}
        if info:
            bits = info.split(';')
            for (k, v) in [x.split('=') for x in bits]:
                if k == 'vendor':
                    pinfo['manufacturer'] = v
                elif k == 'model':
                    pinfo[k] = v
            
        conns.append( ('socket://%s:%d' % (host, PORT), pinfo,
                       'JetDirect Printer at %s' % host,
                       format_printinfo(pinfo)) )
    return conns

_portmap = {
    9100 : detect_jetdirect_queues,
    #515 : detect_lpd_queues,
    #631 : detect_ipp_queues, # Possibly doable
    }

def detect_tcp_printers(hosts=None):
    conns = []
    printerinfo = detect_hosts(hosts=hosts, ports=_portmap.keys())
    for (port, hosts) in printerinfo.iteritems():
        func = _portmap.get(port)
        if func:
            conns += func(hosts)

    return conns

remote_detect = [
    detect_tcp_printers,
    detect_smb_printers,
    ]

def detect_remote_printers():
    conns = []
    
    # Could scan the network for open IPP and LPD ports.
    # smbtree seems to output some printers, but I'm not sure if there's an
    # easy way to figure out what's a printer and what's not.

    for routine in remote_detect:
        conns += routine()
    
    return conns


def printer_connections(remote=False):
    "Return a list of possible printer connection points."
    conns = []
    fallback_conns = []

    for routine in (detect_usb_printers, detect_parallel_printers,
                    detect_serial_printers):
        conns += routine()

    if remote:
        conns += detect_remote_printers()

    conns.sort(lambda x, y: cmp(x[2], y[2]))

    return conns + fallback_conns

def test():
    #print get_local_hosts()
    #print printer_connections(remote=False)
    print printer_connections(remote=True)
    #print detect_remote_printers()

if __name__ == '__main__':
    test()