/usr/bin/mountpy.py

#!/usr/bin/python


import os, glob, sys, gzip, re, traceback, string, commands

from commands import getstatusoutput

from optparse import OptionParser

#default configuration

config_file = '/etc/mountpy.conf'
#device prefix 
device_prefix = '/dev'

probe_devices = [
'sd[a-z]',
'sd[a-z][1-9]',
'fd[0-1]',
'cdrom',
'sr[0-9]'
]

blacklist_devices = [
#'fd[0-1]'
]

whitelist_devices = [
]

default_options = 'users,noatime,sync,dirsync'
default_options2 = 'users,noatime,dirsync'

# ('filesystem_name', 'filesystem_options')
fs_options = {
'vfat': default_options2+',uid=%(uid)s,gid=%(gid)s,utf8',
'msdos': default_options2+',uid=%(uid)s,gid=%(gid)s,utf8',
'iso9660': 'users,ro,utf8,unhide',
'ntfs': 'users,ro,nls=utf8,uid=%(uid)s,gid=%(gid)s',
'auto': default_options
}

try_filesystems = ['vfat','msdos','iso9660','ntfs','auto']

mount_command = '/bin/mount'
umount_command = '/bin/umount'

do_sync = True

device_map = {

}

mntdir = '/media/'

check_removables = "no"

mtab = '/etc/mtab'

# end of default options

VERSION='0.8.1'

if not os.path.isfile(config_file):
    print "Missing configuration file", config_file
    sys.exit()
    
execfile(config_file)


#safety checks
if ( not os.path.isabs(mntdir) or
     len(os.path.normpath(mntdir))<2  # empty or root /
     ):
    print "mntdir is set to %s" % `mntdir`
    print "This looks insane to me, refusing to continue"
    print "Please edit /etc/mountpy.conf"
    print "If you have no idea what this means, contact your system administrator."
    print "If you have no idea what this means and YOU are "
    print "an administrator, go read some basic literature about UNIX"
    print "and then try again."
    sys.exit()
     
colours = {
            'none'       :    "",
            'default'    :    "\033[0m",
            'bold'       :    "\033[1m",
            'underline'  :    "\033[4m",
            'blink'      :    "\033[5m",
            'reverse'    :    "\033[7m",
            'concealed'  :    "\033[8m",

            'black'      :    "\033[30m",
            'red'        :    "\033[31m",
            'green'      :    "\033[32m",
            'yellow'     :    "\033[33m",
            'blue'       :    "\033[34m",
            'magenta'    :    "\033[35m",
            'cyan'       :    "\033[36m",
            'white'      :    "\033[37m",

            'on_black'   :    "\033[40m",
            'on_red'     :    "\033[41m",
            'on_green'   :    "\033[42m",
            'on_yellow'  :    "\033[43m",
            'on_blue'    :    "\033[44m",
            'on_magenta' :    "\033[45m",
            'on_cyan'    :    "\033[46m",
            'on_white'   :    "\033[47m",

            'beep'       :    "\007",
            }

alert_colour = colours['red']+colours['bold']
info_colour = colours['green']
default_colour = colours['default']


def alert(*text):
    sys.stdout.write(alert_colour)
    sys.stdout.write(' '.join(text))
    sys.stdout.write(' '*20)
    sys.stdout.write('\n')
    sys.stdout.write(default_colour)
    sys.stdout.flush()

def info(*text):
    sys.stdout.write(info_colour)
    sys.stdout.write(' '.join(text))
    sys.stdout.write(' '*20)
    sys.stdout.write('\n')
    sys.stdout.write(default_colour)
    sys.stdout.flush()

def debug(*text):
    sys.stdout.write(' '.join(text))
    sys.stdout.write(' '*20)
    sys.stdout.write('\n')

def msg(*text):
    sys.stdout.write(' '.join(text))
    sys.stdout.write(' '*20)
    sys.stdout.write('\r')
    sys.stdout.flush()

def sync_it():
    os.system('/bin/sync &')

def safe_rmdir(path):
    "remove directory, but do some safety checks before"
    if ( (os.path.isdir(path) or os.path.islink(path)) and 
         os.path.isabs(path) and 
         not os.path.ismount(path) and
         path.startswith(mntdir) and
         path == os.path.normpath(path) and
         path.startswith(mntdir)
          ):

        if os.path.islink(path):
            os.remove(path)
        else:
            os.rmdir(path)
    else:
        raise IOError, "Trying to remove suspicious directory: "+`path`

def get_all_devices(list_to_probe):
    r = []
    for i in list_to_probe:
        path = os.path.join(device_prefix, i)
        path = os.path.normpath(path)
        for j in glob.glob(path):
            x = os.path.normpath(j)
            if x.startswith(device_prefix):
                r.append(j)
            else:
                alert("Discarding suspicious device "+`j`)
    return r

def check_device(devicepath):
    "check if device exists at all"
    try:
        os.open(devicepath, os.O_RDONLY)
    except OSError:
        return False
    return True

def is_removable(device):
    "test if device is removable"
    "possible return values are: 'no', 'dunno', 'yes'"
    "test is done by reading /sys/block/hda/removable"
    
    drive = list(device)
    while len(drive)>=1 and drive[-1] in string.digits:
        del drive[-1]
    drive = ''.join(drive)
    path = '/asys/block/%s/removable' % drive
    try:
        val = file(path).read()
    except IOError:
        return 'dunno'
    val = val.strip()
    if val == '0':
        return 'no'
    elif val == '1':
        return 'yes'
    return 'dunno'
    
    
def try_mount(devicepath, fs):
    dir_was_created = False
    device = os.path.split(devicepath)[1]
        
    mountpoint = os.path.join(mntdir, device)
    if os.path.islink(mountpoint) and not os.path.exists(mountpoint):
        # broken symlink
        info(mountpoint + " is a broken symlink - removing")
        safe_rmdir(mountpoint)
    if os.path.exists(mountpoint):
        verbose(mountpoint + " already exists")
        if os.path.ismount(mountpoint):
            verbose(mountpoint + " is already mounted - skipping")
            return
    else:
        os.makedirs(mountpoint)
        dir_was_created = True
    d = {'uid':uid, 'gid':gid}
    mount_options = fs_options.get(fs, default_options) % d
    command = '%s "%s" "%s" -t %s -o %s' % (mount_command, devicepath, mountpoint, fs, mount_options)
    status, output = getstatusoutput(command)
    if status == 0:
        return True
    else:
        verbose("%s %s" %(status, output))
        if dir_was_created:
            safe_rmdir(mountpoint)
        return False

def try_umount(devicepath):
    mountpoint = os.path.join(mntdir, os.path.split(devicepath)[1])
    if os.path.islink(mountpoint):
        #verbose(mountpoint + " is a symlink - removing")
        #safe_rmdir(mountpoint)
        verbose(mountpoint + " is a symlink - skipping")
        return None
    if not os.path.exists(mountpoint):
        verbose(mountpoint + " does not exists")
        return None
    if not os.path.ismount(mountpoint):
        verbose(mountpoint + " is not a mountpoint")
        return None
        
#    if not os.path.ismount(mountpoint):
#        verbose(mountpoint + " is not mounted - removing")
#        safe_rmdir(mountpoint)
#        return None
    command = '%s "%s"' % (umount_command, mountpoint)
    status, output = getstatusoutput(command)
    if status == 0:
#        safe_rmdir(mountpoint)
        return True
    else:
        alert("Could not umount %s, reason: %s\n" %(devicepath, output))
        return False # could not umount

def get_mounted_fs(mtab):
    "try to get a list of already mounted devices"
    "return dictionary of device: mountpoint"
    devices = {}
    if os.path.exists(mtab):
        try:
            lines = file(mtab).readlines()
            devices = [x.split()[0:2] for x in lines]
            devices = dict(devices)
        except IOError:
            devices = {}
    return devices

def call_sync():
    verbose('Calling sync')
    os.system('sync')

how_to_run = '''
mountpy is not being run with enough privileges.
There are 3 ways how to make mountpy run comfortably with root privileges:
1) install sudo, add this line to /etc/sudoers:
joesmith ALL = NOPASSWD: /usr/bin/mountpy, usr/bin/umountpy
   to allow user joesmith use mountpy.
   It you want to allow all the users use mountpy, change the line into:
 ALL ALL = NOPASSWD: /usr/bin/mountpy, usr/bin/umountpy
   and then run "sudo mountpy" and "sudo umountpy"
2) install super, add these lines to /etc/super.tab:
   mountpy /usr/bin/mountpy.py joesmith
umountpy /usr/bin/umountpy joesmith
   to allow user joesmith to run mountpy, or:
mountpy /usr/bin/mountpy.py .*
umountpy /usr/bin/umountpy .*
   to give access to all the users, and run "super mountpy" or "super umountpy"
3) make the C wrapper script setuid root
   log in as root and type:
chmod u+s /usr/bin/mountpy

Be aware that giving all the users rights to execute mountpy can be a
security risk, if an exploitable bug is found in mountpy.  This is
especially true in the 3rd case. Therefore, it is recommended to allow
only selected users to run mountpy, and use super or sudo if possible.
'''
    
def verbose(text, verbosity = 1):
    if options.verbosity >= verbosity:
        debug(text)

parser = OptionParser(usage="usage: %prog [options] arg")
parser.add_option("-u", "--umount",
      action="store_const", const='umount', dest="action", default="mount",
      help="Umount")
parser.add_option("-m", "--mount",
      action="store_const", const='mount', dest="action", default="mount",
      help="Mount")

parser.add_option("-v", "--verbose",
      action="count", dest="verbosity",
      default=0,
      help="Increase verbosity")

      
      
(options, args) = parser.parse_args()


if options.verbosity > 0:
    verbose("Action " + options.action)

if len(args)==0:
    verbose("Will try all possible devices")
    if options.action == 'mount':
        devices = get_all_devices(probe_devices)
    elif options.action == 'umount':
        devices = os.listdir(mntdir)
    else:
        raise
else:
    devices = get_all_devices(args)

blacklist_devices = get_all_devices(blacklist_devices)
wihtelist_devices = get_all_devices(whitelist_devices)


processed = [] # list of succesfully mounted/umounted devices
unsuccessful = [] # list of devices that could not have been umounted

uid = os.getuid()
gid = os.getgid()
euid = os.geteuid()

# test if we are running with sudo
if uid==0:
    if os.environ.has_key('SUDO_UID'):
        uid = int(os.environ['SUDO_UID'])
        verbose("sudo detected")

verbose("Detected UID %s, GID %s, EUID %s" % (uid, gid, euid))

if euid != 0:
    msg(how_to_run)
    raw_input("Press ENTER to try anyway")

if options.action == "umount" and do_sync:
    call_sync()

os.setuid(0)


mounted_devicepaths = get_mounted_fs(mtab)

for devicepath in devices:
    if devicepath in blacklist_devices:
        verbose("Blacklisted "+ devicepath, 2)
        continue

    device = os.path.split(devicepath)[1]

    if options.action == "mount":

        if device not in whitelist_devices and check_removables != 'no':
            removable = is_removable(device)
            if check_removables == 'strict' and removable != 'yes':
                verbose(device + " is not known to be removable and you have strict checking on")
                continue
            elif check_removables == 'relaxed' and removable == 'no':
                verbose(device + " is not removable and you have relaxed checking on")
                continue

        if devicepath in mounted_devicepaths:
            info(devicepath, 'seems to be already mounted on', mounted_devicepaths[devicepath])
            continue

        msg("Checking", devicepath)
        if not check_device(devicepath):
            verbose("Device not present "+ devicepath, 2)
            continue
        for fs in try_filesystems:
            verbose("Trying to mount " + devicepath +' '+fs, 2)
            msg("Trying to mount", devicepath, fs)
            r = try_mount(devicepath, fs)
            if r:
                info("Mounted", devicepath, fs)
                processed.append(devicepath)
                break

    elif options.action == "umount":
        verbose("Trying to umount " + devicepath, 2)
        msg("Trying to umount", devicepath)
        r = try_umount(devicepath)
        if r:
            info("Umounted", devicepath)
            processed.append(devicepath)
        if r==False:
            unsuccessful.append(devicepath)
            
if options.action == "mount": 
    if processed:
        info('List of mounted devices: ', ' '.join(processed))
    else:
        info('Found nothing to mount')
    
if options.action == "umount":
    if processed:
        info('List of umounted devices: ', ' '.join(processed))
    if unsuccessful:
        sync_it()
        alert('WARNING: these devices remained mounted: ', ' '.join(unsuccessful))
    if not processed and not unsuccessful:
        info('Found nothing to umount')
mountpy 0.8.1build1 / usr / bin / mountpy.py
