tests/hwsim/wpasupplicant.py

   1 #!/usr/bin/python
   2 #
   3 # Python class for controlling wpa_supplicant
   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 time
  11 import logging
  12 import wpaspy
  13
  14 logger = logging.getLogger(__name__)
  15 wpas_ctrl = '/var/run/wpa_supplicant'
  16
  17 class WpaSupplicant:
  18     def __init__(self, ifname):
  19         self.ifname = ifname
  20         self.ctrl = wpaspy.Ctrl(os.path.join(wpas_ctrl, ifname))
  21         self.mon = wpaspy.Ctrl(os.path.join(wpas_ctrl, ifname))
  22         self.mon.attach()
  23
  24     def request(self, cmd):
  25         logger.debug(self.ifname + ": CTRL: " + cmd)
  26         return self.ctrl.request(cmd)
  27
  28     def ping(self):
  29         return "PONG" in self.request("PING")
  30
  31     def reset(self):
  32         self.request("P2P_STOP_FIND")
  33         self.request("P2P_FLUSH")
  34         self.request("P2P_GROUP_REMOVE *")
  35         self.request("REMOVE_NETWORK *")
  36         self.request("REMOVE_CRED *")
  37
  38     def get_status(self, field):
  39         res = self.request("STATUS")
  40         lines = res.splitlines()
  41         for l in lines:
  42             [name,value] = l.split('=', 1)
  43             if name == field:
  44                 return value
  45         return None
  46
  47     def p2p_dev_addr(self):
  48         return self.get_status("p2p_device_address")
  49
  50     def p2p_listen(self):
  51         return self.request("P2P_LISTEN")
  52
  53     def p2p_find(self, social=False):
  54         if social:
  55             return self.request("P2P_FIND type=social")
  56         return self.request("P2P_FIND")
  57
  58     def wps_read_pin(self):
  59         #TODO: make this random
  60         self.pin = "12345670"
  61         return self.pin
  62
  63     def peer_known(self, peer, full=True):
  64         res = self.request("P2P_PEER " + peer)
  65         if peer.lower() not in res.lower():
  66             return False
  67         if not full:
  68             return True
  69         return "[PROBE_REQ_ONLY]" not in res
  70
  71     def discover_peer(self, peer, full=True, timeout=15):
  72         logger.info(self.ifname + ": Trying to discover peer " + peer)
  73         if self.peer_known(peer, full):
  74             return True
  75         self.p2p_find()
  76         count = 0
  77         while count < timeout:
  78             time.sleep(1)
  79             count = count + 1
  80             if self.peer_known(peer, full):
  81                 return True
  82         return False
  83
  84     def p2p_go_neg_auth(self, peer, pin, method):
  85         if not self.discover_peer(peer):
  86             raise Exception("Peer " + peer + " not found")
  87         self.dump_monitor()
  88         cmd = "P2P_CONNECT " + peer + " " + pin + " " + method + " auth"
  89         if "OK" in self.request(cmd):
  90             return None
  91         raise Exception("P2P_CONNECT (auth) failed")
  92
  93     def p2p_go_neg_init(self, peer, pin, method, timeout=0):
  94         if not self.discover_peer(peer):
  95             raise Exception("Peer " + peer + " not found")
  96         self.dump_monitor()
  97         cmd = "P2P_CONNECT " + peer + " " + pin + " " + method
  98         if "OK" in self.request(cmd):
  99             if timeout == 0:
 100                 self.dump_monitor()
 101                 return None
 102             if self.wait_event("P2P-GROUP-STARTED", timeout):
 103                 self.dump_monitor()
 104                 return None
 105             raise Exception("Group formation timed out")
 106         raise Exception("P2P_CONNECT failed")
 107
 108     def wait_event(self, event, timeout):
 109         count = 0
 110         while count < timeout * 2:
 111             count = count + 1
 112             time.sleep(0.5)
 113             while self.mon.pending():
 114                 ev = self.mon.recv()
 115                 if event in ev:
 116                     return True
 117         return False
 118
 119     def dump_monitor(self):
 120         while self.mon.pending():
 121             ev = self.mon.recv()
 122             logger.debug(self.ifname + ": " + ev)
 123
 124     def remove_group(self, ifname=None):
 125         if ifname is None:
 126             ifname = self.ifname
 127         if "OK" not in self.request("P2P_GROUP_REMOVE " + ifname):
 128             raise Exception("Group could not be removed")