diff mbox series

[2/7] wpaspy: add option to use wpa_cli

Message ID 20240309194228.4186699-2-janusz.dziedzic@gmail.com
State Changes Requested
Headers show
Series [1/7] tests: remotehost: extend proc api | expand

Commit Message

Janusz Dziedzic March 9, 2024, 7:42 p.m. UTC 
This one allow connection to remote/local
host via SSH and use wpa_cli.

Signed-off-by: Janusz Dziedzic <janusz.dziedzic@gmail.com>
---
 wpaspy/wpaspy.py | 52 +++++++++++++++++++++++++++++++++++++++++++++++-
 1 file changed, 51 insertions(+), 1 deletion(-)

Comments

Jouni Malinen March 23, 2024, 8:02 p.m. UTC | #1 
On Sat, Mar 09, 2024 at 08:42:23PM +0100, Janusz Dziedzic wrote:
> This one allow connection to remote/local
> host via SSH and use wpa_cli.

> diff --git a/wpaspy/wpaspy.py b/wpaspy/wpaspy.py
> @@ -10,16 +10,26 @@ import os
>  import stat
>  import socket
>  import select
> +import remotehost

That feels quite undesired. remotehost.py is part of the tests/hwsim
test setup while wpaspy.py is supposed to be standalone module that can
be used to communicate with wpa_supplicant and hostapd for various needs
that are certainly not limited to just testing.

Maybe this should be in a new class within tests/hwsim that inherits
class Ctrl and extends it for this purpose.
diff mbox series

Patch

diff --git a/wpaspy/wpaspy.py b/wpaspy/wpaspy.py
index 5b8140b7c..ae305c8d5 100644
--- a/wpaspy/wpaspy.py
+++ b/wpaspy/wpaspy.py
@@ -10,16 +10,26 @@  import os
 import stat
 import socket
 import select
+import remotehost
 
 counter = 0
 
 class Ctrl:
-    def __init__(self, path, port=9877):
+    def __init__(self, path, port=9877, hostname=None, ifname=None):
         global counter
         self.started = False
         self.attached = False
         self.path = path
         self.port = port
+        self.ifname = ifname
+        self.host = None
+        self.proc = None
+        self.hostname = hostname
+
+        if hostname:
+            self.host = remotehost.Host(hostname)
+            self.started = True
+            return
 
         self.udp = False
         if not path.startswith('/'):
@@ -74,6 +84,11 @@  class Ctrl:
                 # Need to ignore this allow the socket to be closed
                 self.attached = False
                 pass
+
+        if self.host and self.started:
+            self.started = False
+            return
+
         if self.started:
             self.s.close()
             if not self.udp:
@@ -81,6 +96,15 @@  class Ctrl:
             self.started = False
 
     def request(self, cmd, timeout=10):
+        if self.host:
+            cmd = '\'' + cmd + '\''
+            if self.ifname:
+                _cmd = ['wpa_cli', '-p', self.path, '-i', self.ifname, "raw " + cmd]
+            else:
+                _cmd = ['wpa_cli', '-g', self.path, "raw " + cmd]
+            status, buf = self.host.execute(_cmd)
+            return buf
+
         if type(cmd) == str:
             try:
                 cmd2 = cmd.encode()
@@ -104,6 +128,16 @@  class Ctrl:
     def attach(self):
         if self.attached:
             return None
+
+        if self.host:
+            if self.ifname:
+                _cmd = [ "wpa_cli", "-p", self.path, "-i", self.ifname ]
+            else:
+                _cmd = [ "wpa_cli", '-g', self.path]
+            self.proc = self.host.proc_run(_cmd)
+            self.attached = True
+            return
+
         res = self.request("ATTACH")
         if "OK" in res:
             self.attached = True
@@ -113,6 +147,15 @@  class Ctrl:
     def detach(self):
         if not self.attached:
             return None
+
+        if self.hostname and self.proc:
+            self.request("DETACH")
+            self.request("QUIT")
+            self.host.proc_stop(self.proc)
+            self.attached = False
+            self.proc = None
+            return None
+
         if self.s.fileno() == -1:
             self.attached = False
             return None
@@ -135,12 +178,19 @@  class Ctrl:
         self.close()
 
     def pending(self, timeout=0):
+        if self.host and self.proc:
+            return self.host.proc_pending(self.proc, timeout=timeout)
+
         [r, w, e] = select.select([self.s], [], [], timeout)
         if r:
             return True
         return False
 
     def recv(self):
+        if self.host and self.proc:
+            res = self.host.proc_read(self.proc)
+            return res
+
         res = self.s.recv(4096).decode()
         try:
             r = str(res)