blob: 06608c7617f00831aa8c7984d0d2ce2956e552fc [file] [log] [blame]
"""ifstats: get information about a network interface."""
import os
import pipes
import re
class InterfaceStats(object):
"""InterfaceStats collects statistics from a network interface.
It's specialized for wireless interfaces on OS X and Linux, but will handle
wired interfaces too. This is useful when testing a router that you have
connected to over Ethernet.
def __init__(self, system=None, runner=None, interface=None):
self.system = system = runner
self.interface = interface
def IpAddr(self):
"""Invoke `ip addr` in one-line mode and return output as a string."""
return'ip -o -f inet addr')
def IwLink(self, devname):
"""Invoke `iw dev <devname> link` and return its output as a string."""
return'iw dev {} link'.format(pipes.quote(devname)))
def IwScan(self, devname):
"""Invoke `iw dev <devname> scan` and return its output as a string."""
devname_quoted = pipes.quote(devname)
text ='iw dev {} scan dump'.format(devname_quoted))
if not text:
text ='iw dev {} scan'.format(devname_quoted))
return text
def Ifconfig(self):
"""Gather information about known network interfaces from `ifconfig`."""
def AirportI(self):
"""Gather information about the current wireless network from `airport`."""
return'airport -I')
def AirportScan(self):
"""Gather information about other observable networks from `airport`."""
return'airport -s')
def SystemProfiler(self):
"""Gather information about wireless capabilities from `system_profiler`.
On a Mac with 802.11ac wireless, this will also contain a wireless network
scan result with more information than `airport -s` provides.
A string containing the output of `system_profiler`, and attributes as
described in the Fabric documentation.
return'system_profiler -xml SPAirPortDataType')
def GatherDarwin(self):
"""Gather wireless network information on Darwin (Mac OS X)."""
outputs = {
'ifconfig': self.Ifconfig(),
'airport': self.AirportI(),
'airportscan': self.AirportScan(),
'systemprofiler': self.SystemProfiler()
return outputs
def GatherLinux(self):
"""Gather wireless network information on Linux."""
outputs = {'ipaddr': self.IpAddr()}
if self.interface is not None:
'iwlink': self.IwLink(self.interface),
'iwscan': self.IwScan(self.interface)
return outputs
# If no interface was supplied, try to use the first connected one (like
# `airport` does for you automatically.)
iw_dev_raw ='iw dev')
for line in iw_dev_raw.splitlines():
tokens = line.split()
if len(tokens) >= 2 and tokens[0] == 'Interface':
interface = tokens[1]
'iwlink': self.IwLink(interface),
'iwscan': self.IwScan(interface)
except ValueError:
return outputs
def Gather(self):
"""Gather all the wireless network information we know how to."""
if self.system == 'Darwin':
return self.GatherDarwin()
elif self.system == 'Linux':
return self.GatherLinux()
raise OSError('System {} unsupported for InterfaceStats'.format(
def ParseIwLink(text):
"""Parse the text output by `iw dev <devname> link` into a dict."""
ol = text.splitlines()
if not ol:
return {}
# BSSID is in the first line, in an idiosyncratic format.
# sample: Connected to d8:c7:c8:d7:72:30 (on wlan0)
m ='(\w{2}:){5}\w{2}', ol[0])
if not m:
return {}
result = {'BSSID':}
for line in ol[1:]:
key, value = line.split(':', 1)
result[key.strip()] = value.strip()
except ValueError:
return result
def ParseIpAddr(text):
ol = text.splitlines()
result = {}
for line in ol:
_, interface, _, addr, _ = line.split(None, 4)
result[interface] = addr
return result
def ParseAirportI(text):
"""Parse the text output by `airport -i` into a dict."""
result = {}
for line in text.splitlines():
key, value = [cell.strip() for cell in line.split(':', 1)]
if key in ['agrCtlRSSI', 'agrCtlNoise']:
result[key] = int(value)
result[key] = value
except ValueError:
return result
def ParseAirportScan(text):
"""Parse the text output by `airport -S` into a table.
The table is similar to what would be returned by a csv.reader, and suitable
for output by a csv.writer.
text: a string containing the text output of `airport -s`
a list of lists, where the first list is a header row (all strings)
and subsequent rows are mixed strings and integers.
# This is a simple fixed-width format.
header = ['SSID', 'BSSID', 'RSSI', 'CHANNEL', 'HT', 'CC',
'SECURITY (auth/unicast/group)']
result = []
chre = re.compile(r'\d+(?:,\+|-\d+)?')
for line in text.splitlines():
ssid, bssid, rssi, channel, ht, cc, security = (
[cell.strip() for cell in (line[:32], line[33:50], line[51:55],
line[56:63], line[64:66], line[67:69],
# the scan sometimes includes comment lines. assume that anything that has
# a valid channel isn't a comment line.
if chre.match(channel):
result += [dict(zip(header,
[ssid, bssid, int(rssi), channel, ht, cc, security]))]
return result
def Parse(system, cache):
if system == 'Darwin':
return {
'link': ParseAirportI(cache.get('airport', '')),
'scan': ParseAirportScan(cache.get('airportscan', ''))
elif system == 'Linux':
return {
'link': ParseIwLink(cache.get('iwlink', '')),
'addr': ParseIpAddr(cache.get('ipaddr', ''))
raise OSError('Attempt to parse cache from '
'unrecognized system {}'.format(system))
def Restore(report_dir):
"""Restore an InterfaceStats cache from data on the filesystem."""
cache = {}
system_results = {
'Darwin': ['ifconfig', 'airport', 'airportscan', 'systemprofiler'],
'Linux': ['ipaddr', 'iwlink', 'iwscan'],
read = set()
for system, names in system_results.items():
for name in names:
with open(os.path.join(report_dir, name)) as infile:
cache[name] =
except IOError:
cache[name] = ''
if read:
return read, cache
raise IOError('Could not open report in {}'.format(report_dir))