chrissewell/pwnagotchi
1
0
Fork 0
mirror of https://github.com/jayofelony/pwnagotchi.git synced 2025-07-01 18:37:27 -04:00
Code Issues Packages Projects Releases Wiki Activity

Version 2.3.8

Browse Source 
Signed-off-by: Jeroen Oudshoorn <oudshoorn.jeroen@gmail.com>
This commit is contained in:
Elvis
2023-09-10 09:13:57 +02:00
committed by Jeroen Oudshoorn
parent 38b4e4c334
commit bd586227b0
1 changed files with 37 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

47
pwnagotchi/bettercap.py
View File

@ -3,8 +3,9 @@ import requests
import websockets import websockets
from requests.auth import HTTPBasicAuth from requests.auth import HTTPBasicAuth
import asyncio # Add asyncio for async functionality from time import sleep
from time import sleep # Add sleep function
requests.adapters.DEFAULT_RETRIES = 5 # increase retries number
def decode(r, verbose_errors=True): def decode(r, verbose_errors=True):
@ -22,8 +23,7 @@ def decode(r, verbose_errors=True):
class Client(object): class Client(object):
def __init__(self, hostname='localhost', scheme='http', port=8081, username='user', password='pass', def __init__(self, hostname='localhost', scheme='http', port=8081, username='user', password='pass'):
max_websockets=5):
self.hostname = hostname self.hostname = hostname
self.scheme = scheme self.scheme = scheme
self.port = port self.port = port
@ -32,31 +32,46 @@ class Client(object):
self.url = "%s://%s:%d/api" % (scheme, hostname, port) self.url = "%s://%s:%d/api" % (scheme, hostname, port)
self.websocket = "ws://%s:%s@%s:%d/api" % (username, password, hostname, port) self.websocket = "ws://%s:%s@%s:%d/api" % (username, password, hostname, port)
self.auth = HTTPBasicAuth(username, password) self.auth = HTTPBasicAuth(username, password)
self.max_websockets = max_websockets
self.websocket_semaphore = asyncio.Semaphore(max_websockets)
# session takes optional argument to pull a sub-dictionary def session(self):
# ex.: "session/wifi", "session/ble" r = requests.get("%s/session" % self.url, auth=self.auth)
def session(self, sess="session"):
r = requests.get("%s/%s" % (self.url, sess), auth=self.auth)
return decode(r) return decode(r)
async def start_websocket(self, consumer): async def start_websocket(self, consumer):
s = "%s/events" % self.websocket s = "%s/events" % self.websocket
# while True:
# try:
# async with websockets.connect(s, ping_interval=60, ping_timeout=90) as ws:
# async for msg in ws:
# try:
# await consumer(msg)
# except Exception as ex:
# logging.debug("Error while parsing event (%s)", ex)
# except websockets.exceptions.ConnectionClosedError:
# logging.debug("Lost websocket connection. Reconnecting...")
# except websockets.exceptions.WebSocketException as wex:
# logging.debug("Websocket exception (%s)", wex)
# except OSError as e:
# logging.debug("Websocket OSError exception (%s) with parameter %s", e, s)
# except Exception as e:
# logging.debug("Other exception (%s) with parameter %s", e, s)
async with websockets.connect(s, ping_interval=60, ping_timeout=90) as ws:
while True: while True:
try: try:
async with websockets.connect(s, ping_interval=60, ping_timeout=90) as ws:
async for msg in ws: async for msg in ws:
try: try:
await consumer(msg) await consumer(msg)
except Exception as ex: except Exception as ex:
logging.debug("Error while parsing event (%s)", ex) logging.debug("Error while parsing event (%s)", ex)
except websockets.ConnectionClosedError: except websockets.exceptions.ConnectionClosedError:
logging.debug("Lost websocket connection. Reconnecting...") logging.error("Lost websocket connection. Reconnecting...")
except websockets.WebSocketException as wex: except websockets.exceptions.WebSocketException as wex:
logging.debug("Websocket exception (%s)", wex) logging.error("Websocket exception (%s)", wex)
except OSError as e:
logging.error("Websocket OSError exception (%s) with parameter %s", e, s)
except Exception as e: except Exception as e:
logging.exception("Other error while opening websocket (%s) with parameter %s", e, s) logging.error("Other exception (%s) with parameter %s", e, s)
def run(self, command, verbose_errors=True): def run(self, command, verbose_errors=True):
for _ in range(0,2): for _ in range(0,2):