Viewing file:  CheckNetworkServerSanity.py (7.82 KB) -rw-r--r--
Select action/file-type:
 (+) |  (+) |  (+) | Code (+) | Session (+) |  (+) | SDB (+) |  (+) |  (+) |  (+) |  (+) |  (+) |
#!/usr/bin/env python
## Printing troubleshooter
## Copyright (C) 2008, 2009, 2010 Red Hat, Inc.
## Authors:
## 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., 675 Mass Ave, Cambridge, MA 02139, USA.
import cups
import os
import smburi
import socket
import subprocess
from timedops import TimedSubprocess, TimedOperation
from base import *
try:
import smbc
except:
pass
class CheckNetworkServerSanity(Question):
def __init__ (self, troubleshooter):
Question.__init__ (self, troubleshooter, "Check network server sanity")
troubleshooter.new_page (gtk.Label (), self)
def display (self):
# Collect useful information.
self.answers = {}
answers = self.troubleshooter.answers
if (not answers.has_key ('remote_server_name') and
not answers.has_key ('remote_server_ip_address')):
return False
parent = self.troubleshooter.get_window ()
server_name = answers['remote_server_name']
server_port = answers.get('remote_server_port', 631)
if server_name:
# Try resolving the hostname.
try:
ai = socket.getaddrinfo (server_name, server_port)
resolves = map (lambda (family, socktype,
proto, canonname, sockaddr):
sockaddr[0], ai)
except socket.gaierror:
resolves = False
self.answers['remote_server_name_resolves'] = resolves
if resolves:
ipaddr = answers.get ('remote_server_ip_address', '')
if ipaddr:
try:
resolves.index (ipaddr)
except ValueError:
# The IP address given doesn't match the server name.
# Use the IP address instead of the name.
server_name = ipaddr
else:
server_name = answers['remote_server_ip_address']
# Validate it.
try:
ai = socket.getaddrinfo (server_name, server_port)
resolves = map (lambda (family, socktype,
proto, canonname, sockaddr):
sockaddr[0], ai)
except socket.gaierror:
resolves = False
self.answers['remote_server_name_resolves'] = resolves
self.answers['remote_server_try_connect'] = server_name
if (self.answers['remote_server_name_resolves'] and
answers.get ('cups_device_uri_scheme', 'ipp') in ['ipp',
'http',
'https']):
if answers.get ('cups_device_uri_scheme') == 'https':
encryption = cups.HTTP_ENCRYPT_REQUIRED
else:
encryption = cups.HTTP_ENCRYPT_IF_REQUESTED
try:
self.op = TimedOperation (cups.Connection,
kwargs={"host": server_name,
"port": server_port,
"encryption": encryption},
parent=parent)
c = self.op.run ()
ipp_connect = True
except RuntimeError:
ipp_connect = False
self.answers['remote_server_connect_ipp'] = ipp_connect
if ipp_connect:
try:
self.op = TimedOperation (c.getPrinters, parent=parent)
self.op.run ()
cups_server = True
except:
cups_server = False
self.answers['remote_server_cups'] = cups_server
if cups_server:
cups_printer_dict = answers.get ('cups_printer_dict', {})
uri = cups_printer_dict.get ('device-uri', None)
if uri:
try:
self.op = TimedOperation (c.getPrinterAttributes,
kwargs={"uri": uri},
parent=parent)
attr = self.op.run ()
self.answers['remote_cups_queue_attributes'] = attr
except:
pass
if (self.answers['remote_server_name_resolves'] and
not self.answers.get ('remote_server_connect_ipp', False)):
# Try to see if we can connect using smbc.
context = None
try:
context = smbc.Context ()
name = self.answers['remote_server_try_connect']
self.op = TimedOperation (context.opendir,
args=("smb://%s/" % name,),
parent=parent)
dir = self.op.run ()
self.op = TimedOperation (dir.getdents, parent=parent)
shares = self.op.run ()
self.answers['remote_server_smb'] = True
self.answers['remote_server_smb_shares'] = shares
except NameError:
# No smbc support
pass
except RuntimeError, (e, s):
self.answers['remote_server_smb_shares'] = (e, s)
if context != None and answers.has_key ('cups_printer_dict'):
uri = answers['cups_printer_dict'].get ('device-uri', '')
u = smburi.SMBURI (uri)
(group, host, share, user, password) = u.separate ()
accessible = False
try:
self.op = TimedOperation (context.open,
args=("smb://%s/%s" % (host,
share),
os.O_RDWR,
0777),
parent=parent)
f = self.op.run ()
accessible = True
except RuntimeError, (e, s):
accessible = (e, s)
self.answers['remote_server_smb_share_anon_access'] = accessible
# Try traceroute if we haven't already.
if (self.answers['remote_server_name_resolves'] and
not answers.has_key ('remote_server_traceroute')):
try:
self.op = TimedSubprocess (parent=parent, close_fds=True,
args=['traceroute', '-w', '1',
server_name],
stdin=file("/dev/null"),
stdout=subprocess.PIPE,
stderr=subprocess.PIPE)
self.answers['remote_server_traceroute'] = self.op.run ()
except:
# Problem executing command.
pass
return False
def collect_answer (self):
return self.answers
def cancel_operation (self):
self.op.cancel ()
|