wpaspy/wpaspy.py

   1 #!/usr/bin/python
   2 #
   3 # wpa_supplicant/hostapd control interface using Python
   4 # Copyright (c) 2013, Jouni Malinen <j@w1.fi>
   5 #
   6 # This software may be distributed under the terms of the BSD license.
   7 # See README for more details.
   8
   9 import os
  10 import socket
  11 import select
  12
  13 counter = 0
  14
  15 class Ctrl:
  16     def __init__(self, path):
  17         global counter
  18         self.started = False
  19         self.attached = False
  20         self.s = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
  21         self.dest = path
  22         self.local = "/tmp/wpa_ctrl_" + str(os.getpid()) + '-' + str(counter)
  23         counter += 1
  24         self.s.bind(self.local)
  25         try:
  26             self.s.connect(self.dest)
  27         except Exception, e:
  28             self.s.close()
  29             os.unlink(self.local)
  30             raise
  31         self.started = True
  32
  33     def __del__(self):
  34         self.close()
  35
  36     def close(self):
  37         if self.attached:
  38             try:
  39                 self.detach()
  40             except Exception, e:
  41                 # Need to ignore this allow the socket to be closed
  42                 pass
  43         if self.started:
  44             self.s.close()
  45             os.unlink(self.local)
  46             self.started = False
  47
  48     def request(self, cmd):
  49         self.s.send(cmd)
  50         [r, w, e] = select.select([self.s], [], [], 10)
  51         if r:
  52             return self.s.recv(4096)
  53         raise Exception("Timeout on waiting response")
  54
  55     def attach(self):
  56         if self.attached:
  57             return None
  58         res = self.request("ATTACH")
  59         if "OK" in res:
  60             self.attached = True
  61             return None
  62         raise Exception("ATTACH failed")
  63
  64     def detach(self):
  65         if not self.attached:
  66             return None
  67         res = self.request("DETACH")
  68         if "OK" in res:
  69             self.attached = False
  70             return None
  71         raise Exception("DETACH failed")
  72
  73     def pending(self, timeout=0):
  74         [r, w, e] = select.select([self.s], [], [], timeout)
  75         if r:
  76             return True
  77         return False
  78
  79     def recv(self):
  80         res = self.s.recv(4096)
  81         return res