ui.py

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