ui.py

   1 #!/usr/bin/env python
   2
   3 import curses
   4 import os
   5 import stat
   6 import traceback
   7
   8 from conf import config
   9 from output import logger
  10 import network
  11 import system
  12 import time
  13 import sys
  14
  15
  16 def init_routine():
  17     as_root_check()
  18     init_config()
  19
  20
  21 def as_root_check():
  22     if not system.as_root():
  23         logger.err("Must run as root", log=False)
  24         sys.exit(1)
  25
  26
  27 def init_config():
  28     try:
  29         # make sure config file is protected from user access
  30         mode = os.stat(config.configfile)[stat.ST_MODE]
  31         if mode & (stat.S_IRWXG | stat.S_IRWXO):
  32             logger.err(\
  33             "Config file %s should only be accessible by root (mode is: %s)"\
  34                 % (config.configfile, oct(stat.S_IMODE(mode))))
  35             sys.exit(1)
  36
  37         config.init_config()
  38     except Exception, msg:
  39         logger.err(msg)
  40         sys.exit(1)
  41
  42
  43 def tools_check():
  44     logger.display("Checking for missing tools...", log=False)
  45     for cmd_name in config.tools:
  46         var_name = cmd_name.replace("-", "_")
  47         path = config.__dict__[var_name]
  48         if os.path.isabs(path):
  49             logger.display(" [ok] %s" % path, log=False)
  50         else:
  51             logger.display(" [missing] %s" % cmd_name, log=False)
  52     sys.exit(0)
  53
  54
  55 def ip_check():
  56     logger.await("Checking for ip")
  57     ips = [(x, y) for (x, y) in
  58         [(i.interface, network.has_ip(i.interface))
  59             for i in config.interfaces]
  60         if y != None] # (iface, ip) pairs where ip is set
  61     if ips:
  62         logger.result(
  63             reduce(lambda x, y: "%s  %s" % (x, y),
  64             map(lambda (x, y): "%s (%s)" % (y, x), ips)))
  65         return True
  66     else:
  67         logger.result("none")
  68
  69
  70 def choose_medium():
  71     logger.await("Selecting preferred network medium")
  72     iface = config.interfaces.get_top()
  73     logger.result(iface.medium)
  74     if iface.medium == "wired": return True
  75
  76
  77 def check_wired_link():
  78     logger.await("Checking for wired link")
  79     ifaces = config.interfaces.get_all(pred=lambda x: x.medium == "wired")
  80     connected = [network.wire_connected(i.interface) for i in ifaces \
  81         if network.wire_connected(i.interface)]
  82     if connected:
  83         logger.result(reduce(lambda x,y: "%s %s" % (x,y),
  84             map(lambda x: "%s" % x, connected)))
  85         return True
  86     else:
  87         logger.result("none, use wireless")
  88
  89
  90 def request_ip(iface, net=None, tries=3):
  91     n = iface.interface
  92     if net:
  93         n = "%s (from %s)" % (n, net.essid)
  94         if not network.setup_wifi(iface.interface, net):
  95             logger.err("Failed to associate with access point for network %s"\
  96                 % net.essid)
  97             return
  98
  99     attempt = 0
 100     while attempt < tries:
 101         attempt += 1
 102         logger.await("(%d/%d) Request ip on %s" % (attempt, tries, n))
 103
 104         start_time = time.time()
 105         ip = network.request_ip(iface.interface)
 106         stop_time = time.time()
 107         duration = stop_time - start_time
 108
 109         if ip:
 110             logger.result("%s (%ds)" % (ip, duration))
 111             return True
 112         else: logger.result("failed (%ds)" % duration)
 113
 114
 115 def scan_wifi_networks(iface, wardrive=None):
 116     logger.await("Scan for wireless networks")
 117     nets = network.read_scan(network.normal_scan(iface.interface))
 118     nets = config.networks.merge(nets, 'essid')
 119     f = lambda x: x.priority != None and x.signal != None
 120     if wardrive:
 121         f = lambda x: x.signal != None and x.priority == None\
 122             and x.encrypted == None and x.essid != "<hidden>"
 123     nets = nets.sort(sort_key='quality').get_all(pred=f)
 124     if nets:
 125         logger.result(reduce(lambda x,y: "%s  %s" % (x,y),
 126             map(lambda x: "%s" % x.essid, nets[:3])))
 127         return nets
 128     else: logger.result("none")
 129
 130
 131 def display_live_networks(nets, field=None, column=None):
 132     if not nets:
 133         return "No networks detected"
 134
 135     keys = ['bssid', 'essid', 'channel', 'bitrate', 'sec', 'signal', 'quality']
 136     width = 19; mw = 6; sp = 2 # width: max width  mw : min width  sp : separate
 137
 138     if field:
 139         nets = nets.sort(sort_key=field)
 140     elif column and 0 < column <= len(keys):
 141         nets = nets.sort(sort_key=keys[column-1])
 142     else: nets = nets.sort(sort_key='signal')
 143
 144     # figure out column length to assign to fields based on content length
 145     def col_len(dicts, key):
 146         l = []
 147         for dict in dicts:
 148             if key in dict and dict[key]: l.append(len(dict[key])+sp)
 149             else: l.append(sp)
 150         return l
 151     ws = [max(min(max(col_len(nets, k)), width), mw) for k in keys]
 152
 153     s = ""
 154     for (i, key) in enumerate(keys):
 155         s += ("%s" % key[:ws[i]-sp]).ljust(ws[i])
 156     s = s[:-sp] + "\n"
 157     for net in nets:
 158         for (i, key) in enumerate(keys):
 159             if key in net: s += (net[key][:ws[i]-sp]).ljust(ws[i])
 160             else: s += "".ljust(ws[i])
 161         s = s[:-sp] + "\n"
 162     return s + "%d network(s) found" % len(nets)
 163
 164
 165 def curse(iface, timeout):
 166     def curses_init():
 167         logger.mute(quiet=True)
 168
 169         scr = curses.initscr()
 170         curses.noecho()
 171         curses.cbreak()
 172         scr.keypad(1)
 173         curses.curs_set(0)
 174         return scr
 175
 176     def curses_shutdown(scr):
 177         curses.nocbreak()
 178         scr.keypad(0)
 179         curses.echo()
 180         curses.endwin()
 181
 182         logger.unmute(quiet=True)
 183
 184     def display_list(scr, header, status, list_s):
 185         if list_s:
 186             ss = list_s.split("\n")
 187             scr.clear()
 188             try:
 189                 scr.addstr(header + " / Ctrl+C to exit\n")
 190                 if status:
 191                     scr.addstr(status + "\n")
 192                 scr.addstr("\n"+ss[0]+"\n", curses.A_REVERSE)
 193
 194                 # add one by one until possible exception is thrown
 195                 # when the list is too long
 196                 for s in ss[1:]:
 197                     scr.addstr(s+"\n")
 198             except:
 199                 pass
 200             scr.refresh()
 201
 202     def get_status(timeout, t):
 203         if timeout > 0:
 204             return "-> scanning for %ds (-%ds)" % (timeout, t-time.time())
 205         return ""
 206
 207     list_s = ""
 208     t = time.time()+timeout
 209     header = "%s scanning for wireless networks (%s)" % \
 210             (config.program_name, iface)
 211     status = get_status(timeout, t)
 212
 213     try:
 214         scr = curses_init()
 215         display_list(scr, header, status, display_live_networks(None))
 216
 217         while 1:
 218             status = get_status(timeout, t)
 219             if timeout > 0:
 220                 if t < time.time(): break
 221
 222             scan_data = network.single_scan(iface)
 223             nets = network.read_scan(scan_data)
 224             list_s = display_live_networks(nets)
 225
 226             display_list(scr, header, status, list_s)
 227
 228             curses.napms(500)
 229
 230     except KeyboardInterrupt:
 231         curses_shutdown(scr)
 232     except:
 233         curses_shutdown(scr)
 234         traceback.print_stack()
 235         traceback.print_exc()
 236     else:
 237         curses_shutdown(scr)
 238
 239     print list_s