refactor
This commit is contained in:
Simone Margaritelli
parent
1ebec7691f
commit
ba15d1940a
sdcard/rootfs/root/pwnagotchi/scripts
@ -1,48 +0,0 @@
|
||||
import glob
|
||||
import os
|
||||
import time
|
||||
import subprocess
|
||||
|
||||
|
||||
def secs_to_hhmmss(secs):
|
||||
mins, secs = divmod(secs, 60)
|
||||
hours, mins = divmod(mins, 60)
|
||||
return '%02d:%02d:%02d' % (hours, mins, secs)
|
||||
|
||||
|
||||
def total_unique_handshakes(path):
|
||||
expr = os.path.join(path, "*.pcap")
|
||||
return len(glob.glob(expr))
|
||||
|
||||
|
||||
def iface_address(ifname):
|
||||
output = subprocess.getoutput("/usr/sbin/ifconfig %s" % ifname)
|
||||
for line in output.split("\n"):
|
||||
line = line.strip()
|
||||
if line.startswith("inet "):
|
||||
return line.split(' ')[1].strip()
|
||||
return None
|
||||
|
||||
|
||||
def iface_channels(ifname):
|
||||
channels = []
|
||||
output = subprocess.getoutput("/sbin/iwlist %s freq" % ifname)
|
||||
for line in output.split("\n"):
|
||||
line = line.strip()
|
||||
if line.startswith("Channel "):
|
||||
channels.append(int(line.split()[1]))
|
||||
return channels
|
||||
|
||||
|
||||
def led(on=True):
|
||||
with open('/sys/class/leds/led0/brightness', 'w+t') as fp:
|
||||
fp.write("%d" % (0 if on is True else 1))
|
||||
|
||||
|
||||
def blink(times=1, delay=0.3):
|
||||
for t in range(0, times):
|
||||
led(True)
|
||||
time.sleep(delay)
|
||||
led(False)
|
||||
time.sleep(delay)
|
||||
led(True)
|
||||
@ -9,6 +9,7 @@ import _thread
|
||||
|
||||
import core
|
||||
|
||||
import pwnagotchi.utils as utils
|
||||
import pwnagotchi.plugins as plugins
|
||||
from bettercap.client import Client
|
||||
from pwnagotchi.mesh.utils import AsyncAdvertiser
|
||||
@ -30,7 +31,7 @@ class Agent(Client, AsyncAdvertiser, AsyncTrainer):
|
||||
self._started_at = time.time()
|
||||
self._filter = None if config['main']['filter'] is None else re.compile(config['main']['filter'])
|
||||
self._current_channel = 0
|
||||
self._supported_channels = core.iface_channels(config['main']['iface'])
|
||||
self._supported_channels = utils.iface_channels(config['main']['iface'])
|
||||
self._view = view
|
||||
self._access_points = []
|
||||
self._last_pwnd = None
|
||||
@ -252,7 +253,7 @@ class Agent(Client, AsyncAdvertiser, AsyncTrainer):
|
||||
|
||||
def _update_uptime(self, s):
|
||||
secs = time.time() - self._started_at
|
||||
self._view.set('uptime', core.secs_to_hhmmss(secs))
|
||||
self._view.set('uptime', utils.secs_to_hhmmss(secs))
|
||||
self._view.set('epoch', '%04d' % self._epoch.epoch)
|
||||
|
||||
def _update_counters(self):
|
||||
@ -272,7 +273,7 @@ class Agent(Client, AsyncAdvertiser, AsyncTrainer):
|
||||
if new_shakes > 0:
|
||||
self._epoch.track(handshake=True, inc=new_shakes)
|
||||
|
||||
tot = core.total_unique_handshakes(self._config['bettercap']['handshakes'])
|
||||
tot = utils.total_unique_handshakes(self._config['bettercap']['handshakes'])
|
||||
txt = '%d (%d)' % (len(self._handshakes), tot)
|
||||
|
||||
if self._last_pwnd is not None:
|
||||
@ -285,7 +286,7 @@ class Agent(Client, AsyncAdvertiser, AsyncTrainer):
|
||||
|
||||
def _update_advertisement(self, s):
|
||||
run_handshakes = len(self._handshakes)
|
||||
tot_handshakes = core.total_unique_handshakes(self._config['bettercap']['handshakes'])
|
||||
tot_handshakes = utils.total_unique_handshakes(self._config['bettercap']['handshakes'])
|
||||
started = s['started_at'].split('.')[0]
|
||||
started = datetime.strptime(started, '%Y-%m-%dT%H:%M:%S')
|
||||
started = time.mktime(started.timetuple())
|
||||
|
||||
@ -2,8 +2,8 @@ import time
|
||||
import threading
|
||||
import logging
|
||||
|
||||
import core
|
||||
import pwnagotchi
|
||||
import pwnagotchi.utils as utils
|
||||
import pwnagotchi.mesh.wifi as wifi
|
||||
|
||||
from pwnagotchi.ai.reward import RewardFunction
|
||||
@ -174,22 +174,22 @@ class Epoch(object):
|
||||
self._epoch_data_ready.set()
|
||||
|
||||
logging.info("[epoch %d] duration=%s slept_for=%s blind=%d inactive=%d active=%d hops=%d missed=%d "
|
||||
"deauths=%d assocs=%d handshakes=%d cpu=%d%% mem=%d%% temperature=%dC reward=%s" % (
|
||||
self.epoch,
|
||||
core.secs_to_hhmmss(self.epoch_duration),
|
||||
core.secs_to_hhmmss(self.num_slept),
|
||||
self.blind_for,
|
||||
self.inactive_for,
|
||||
self.active_for,
|
||||
self.num_hops,
|
||||
self.num_missed,
|
||||
self.num_deauths,
|
||||
self.num_assocs,
|
||||
self.num_shakes,
|
||||
cpu * 100,
|
||||
mem * 100,
|
||||
temp,
|
||||
self._epoch_data['reward']))
|
||||
"deauths=%d assocs=%d handshakes=%d cpu=%d%% mem=%d%% temperature=%dC reward=%s" % (
|
||||
self.epoch,
|
||||
utils.secs_to_hhmmss(self.epoch_duration),
|
||||
utils.secs_to_hhmmss(self.num_slept),
|
||||
self.blind_for,
|
||||
self.inactive_for,
|
||||
self.active_for,
|
||||
self.num_hops,
|
||||
self.num_missed,
|
||||
self.num_deauths,
|
||||
self.num_assocs,
|
||||
self.num_shakes,
|
||||
cpu * 100,
|
||||
mem * 100,
|
||||
temp,
|
||||
self._epoch_data['reward']))
|
||||
|
||||
self.epoch += 1
|
||||
self.epoch_started = now
|
||||
|
||||
@ -4,7 +4,7 @@ import time
|
||||
import logging
|
||||
from PIL import Image, ImageDraw
|
||||
|
||||
import core
|
||||
import pwnagotchi.utils as utils
|
||||
import pwnagotchi.plugins as plugins
|
||||
from pwnagotchi.voice import Voice
|
||||
|
||||
@ -86,8 +86,9 @@ class View(object):
|
||||
|
||||
'face': Text(value=faces.SLEEP, position=face_pos, color=BLACK, font=fonts.Huge),
|
||||
|
||||
'friend_face': Text(value=None, position=(0, (self._height * 0.88) - 15 ), font=fonts.Bold, color=BLACK),
|
||||
'friend_name': Text(value=None, position=(40,(self._height * 0.88) - 13 ), font=fonts.BoldSmall, color=BLACK),
|
||||
'friend_face': Text(value=None, position=(0, (self._height * 0.88) - 15), font=fonts.Bold, color=BLACK),
|
||||
'friend_name': Text(value=None, position=(40, (self._height * 0.88) - 13), font=fonts.BoldSmall,
|
||||
color=BLACK),
|
||||
|
||||
'name': Text(value='%s>' % 'pwnagotchi', position=name_pos, color=BLACK, font=fonts.Bold),
|
||||
|
||||
@ -166,7 +167,7 @@ class View(object):
|
||||
self.set('channel', '-')
|
||||
self.set('aps', "%d" % log.associated)
|
||||
self.set('shakes', '%d (%s)' % (log.handshakes, \
|
||||
core.total_unique_handshakes(self._config['bettercap']['handshakes'])))
|
||||
utils.total_unique_handshakes(self._config['bettercap']['handshakes'])))
|
||||
self.set_closest_peer(log.last_peer)
|
||||
|
||||
def is_normal(self):
|
||||
|
||||
@ -1,6 +1,9 @@
|
||||
import yaml
|
||||
import os
|
||||
import logging
|
||||
import glob
|
||||
import os
|
||||
import time
|
||||
import subprocess
|
||||
import yaml
|
||||
|
||||
|
||||
# https://stackoverflow.com/questions/823196/yaml-merge-in-python
|
||||
@ -40,3 +43,38 @@ def setup_logging(args, config):
|
||||
console_handler = logging.StreamHandler()
|
||||
console_handler.setFormatter(formatter)
|
||||
root.addHandler(console_handler)
|
||||
|
||||
|
||||
def secs_to_hhmmss(secs):
|
||||
mins, secs = divmod(secs, 60)
|
||||
hours, mins = divmod(mins, 60)
|
||||
return '%02d:%02d:%02d' % (hours, mins, secs)
|
||||
|
||||
|
||||
def total_unique_handshakes(path):
|
||||
expr = os.path.join(path, "*.pcap")
|
||||
return len(glob.glob(expr))
|
||||
|
||||
|
||||
def iface_channels(ifname):
|
||||
channels = []
|
||||
output = subprocess.getoutput("/sbin/iwlist %s freq" % ifname)
|
||||
for line in output.split("\n"):
|
||||
line = line.strip()
|
||||
if line.startswith("Channel "):
|
||||
channels.append(int(line.split()[1]))
|
||||
return channels
|
||||
|
||||
|
||||
def led(on=True):
|
||||
with open('/sys/class/leds/led0/brightness', 'w+t') as fp:
|
||||
fp.write("%d" % (0 if on is True else 1))
|
||||
|
||||
|
||||
def blink(times=1, delay=0.3):
|
||||
for t in range(0, times):
|
||||
led(True)
|
||||
time.sleep(delay)
|
||||
led(False)
|
||||
time.sleep(delay)
|
||||
led(True)
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user