diff --git a/pwnagotchi/plugins/default/net-pos.py b/pwnagotchi/plugins/default/net-pos.py index 9f7fdd0c9..88400c00a 100644 --- a/pwnagotchi/plugins/default/net-pos.py +++ b/pwnagotchi/plugins/default/net-pos.py @@ -9,104 +9,96 @@ class NetPos(plugins.Plugin): - __author__ = 'zenzen san' - __version__ = '2.0.3' + __author__ = 'zenzen san, doki' + __version__ = '2.0.4' __license__ = 'GPL3' __description__ = """Saves a json file with the access points with more signal whenever a handshake is captured. - When internet is available the files are converted in geo locations - using Mozilla LocationService """ + When internet is available the files are converted to geo locations + using Google Geolocation API.""" - API_URL = 'https://location.services.mozilla.com/v1/geolocate?key={api}' + API_URL = 'https://www.googleapis.com/geolocation/v1/geolocate?key={api}' def __init__(self): - self.report = StatusFile('/root/.net_pos_saved', data_format='json') self.skip = list() self.ready = False self.lock = threading.Lock() + self.report_path = '/root/.net_pos_saved' + self.report = StatusFile(self.report_path, data_format='json') def on_loaded(self): - if 'api_key' not in self.options or ('api_key' in self.options and not self.options['api_key']): - logging.error("NET-POS: api_key isn't set. Can't use mozilla's api.") + if 'api_key' not in self.options or not self.options['api_key']: + logging.error("NET-POS: 'api_key' not set. Can't use Google's Geolocation API.") return if 'api_url' in self.options: self.API_URL = self.options['api_url'] + if 'status_file' in self.options: + self.report_path = self.options['status_file'] + self.report = StatusFile(self.report_path, data_format='json') + self.ready = True - logging.info("net-pos plugin loaded.") - logging.debug(f"net-pos: use api_url: {self.API_URL}"); - - def _append_saved(self, path): - to_save = list() - if isinstance(path, str): - to_save.append(path) - elif isinstance(path, list): - to_save += path - else: - raise TypeError("Expected list or str, got %s" % type(path)) - - with open('/root/.net_pos_saved', 'a') as saved_file: - for x in to_save: - saved_file.write(x + "\n") + logging.info("NET-POS: Plugin loaded successfully.") + logging.debug(f"NET-POS: Using API URL: {self.API_URL}") def on_internet_available(self, agent): if self.lock.locked(): return with self.lock: - if self.ready: - config = agent.config() - display = agent.view() - reported = self.report.data_field_or('reported', default=list()) - handshake_dir = config['bettercap']['handshakes'] - - all_files = os.listdir(handshake_dir) - all_np_files = [os.path.join(handshake_dir, filename) - for filename in all_files - if filename.endswith('.net-pos.json')] - new_np_files = set(all_np_files) - set(reported) - set(self.skip) - - if new_np_files: - logging.debug("NET-POS: Found %d new net-pos files. Fetching positions ...", len(new_np_files)) - display.set('status', f"Found {len(new_np_files)} new net-pos files. Fetching positions ...") - display.update(force=True) - for idx, np_file in enumerate(new_np_files): - - geo_file = np_file.replace('.net-pos.json', '.geo.json') - if os.path.exists(geo_file): - # got already the position - reported.append(np_file) - self.report.update(data={'reported': reported}) - continue - - try: - geo_data = self._get_geo_data(np_file) # returns json obj - except requests.exceptions.RequestException as req_e: - logging.error("NET-POS: %s - RequestException: %s", np_file, req_e) - self.skip += np_file - continue - except json.JSONDecodeError as js_e: - logging.error("NET-POS: %s - JSONDecodeError: %s, removing it...", np_file, js_e) - os.remove(np_file) - continue - except OSError as os_e: - logging.error("NET-POS: %s - OSError: %s", np_file, os_e) - self.skip += np_file - continue - - with open(geo_file, 'w+t') as sf: - json.dump(geo_data, sf) - + if not self.ready: + return + + config = agent.config() + display = agent.view() + reported = self.report.data_field_or('reported', default=list()) + handshake_dir = config['bettercap']['handshakes'] + + all_files = os.listdir(handshake_dir) + all_np_files = [os.path.join(handshake_dir, f) + for f in all_files if f.endswith('.net-pos.json')] + new_np_files = set(all_np_files) - set(reported) - set(self.skip) + + if new_np_files: + logging.debug("NET-POS: Found %d new net-pos files. Fetching positions ...", len(new_np_files)) + display.set('status', f"Found {len(new_np_files)} new net-pos files. Fetching positions ...") + display.update(force=True) + + for idx, np_file in enumerate(new_np_files): + geo_file = np_file.replace('.net-pos.json', '.geo.json') + if os.path.exists(geo_file): reported.append(np_file) self.report.update(data={'reported': reported}) - - display.set('status', f"Fetching positions ({idx + 1}/{len(new_np_files)})") - display.update(force=True) + continue + + try: + geo_data = self._get_geo_data(np_file) + except requests.exceptions.RequestException as req_e: + logging.error("NET-POS: %s - RequestException: %s", np_file, req_e) + self.skip.append(np_file) + continue + except json.JSONDecodeError as js_e: + logging.error("NET-POS: %s - JSONDecodeError: %s, removing it...", np_file, js_e) + os.remove(np_file) + continue + except OSError as os_e: + logging.error("NET-POS: %s - OSError: %s", np_file, os_e) + self.skip.append(np_file) + continue + + with open(geo_file, 'w+t') as sf: + json.dump(geo_data, sf) + + reported.append(np_file) + self.report.update(data={'reported': reported}) + + display.set('status', f"Fetching positions ({idx + 1}/{len(new_np_files)})") + display.update(force=True) def on_handshake(self, agent, filename, access_point, client_station): netpos = self._get_netpos(agent) if not netpos['wifiAccessPoints']: return - netpos["ts"] = int("%.0f" % time.time()) + netpos["ts"] = int(time.time()) netpos_filename = filename.replace('.pcap', '.net-pos.json') logging.debug("NET-POS: Saving net-location to %s", netpos_filename) @@ -116,15 +108,14 @@ def on_handshake(self, agent, filename, access_point, client_station): except OSError as os_e: logging.error("NET-POS: %s", os_e) - def _get_netpos(self, agent): aps = agent.get_access_points() - netpos = dict() - netpos['wifiAccessPoints'] = list() - # 6 seems a good number to save a wifi networks location + netpos = {'wifiAccessPoints': []} for access_point in sorted(aps, key=lambda i: i['rssi'], reverse=True)[:6]: - netpos['wifiAccessPoints'].append({'macAddress': access_point['mac'], - 'signalStrength': access_point['rssi']}) + netpos['wifiAccessPoints'].append({ + 'macAddress': access_point['mac'], + 'signalStrength': access_point['rssi'] + }) return netpos def _get_geo_data(self, path, timeout=30): @@ -133,17 +124,13 @@ def _get_geo_data(self, path, timeout=30): try: with open(path, "r") as json_file: data = json.load(json_file) - except json.JSONDecodeError as js_e: - raise js_e - except OSError as os_e: - raise os_e + except (json.JSONDecodeError, OSError) as e: + raise e try: - result = requests.post(geourl, - json=data, - timeout=timeout) + result = requests.post(geourl, json=data, timeout=timeout) return_geo = result.json() - if data["ts"]: + if data.get("ts"): return_geo["ts"] = data["ts"] return return_geo except requests.exceptions.RequestException as req_e: diff --git a/pwnagotchi/ui/display.py b/pwnagotchi/ui/display.py index ab0f3a15c..5eaad12f8 100644 --- a/pwnagotchi/ui/display.py +++ b/pwnagotchi/ui/display.py @@ -75,6 +75,9 @@ def is_waveshare35lcd(self): def is_spotpear24inch(self): return self._implementation.name == 'spotpear24inch' + + def is_displayhatmini(self): + return self._implementation.name == 'displayhatmini' def is_waveshare_any(self): return self.is_waveshare_v1() or self.is_waveshare_v2() diff --git a/pwnagotchi/ui/hw/__init__.py b/pwnagotchi/ui/hw/__init__.py index eb9beec04..f5ffe6f8c 100644 --- a/pwnagotchi/ui/hw/__init__.py +++ b/pwnagotchi/ui/hw/__init__.py @@ -15,6 +15,7 @@ from pwnagotchi.ui.hw.waveshare213bc import Waveshare213bc from pwnagotchi.ui.hw.waveshare35lcd import Waveshare35lcd from pwnagotchi.ui.hw.spotpear24inch import Spotpear24inch +from pwnagotchi.ui.hw.displayhatmini import DisplayHatMini def display_for(config): # config has been normalized already in utils.load_config @@ -68,3 +69,6 @@ def display_for(config): elif config['ui']['display']['type'] == 'spotpear24inch': return Spotpear24inch(config) + + elif config['ui']['display']['type'] == 'displayhatmini': + return DisplayHatMini(config) diff --git a/pwnagotchi/ui/hw/displayhatmini.py b/pwnagotchi/ui/hw/displayhatmini.py new file mode 100644 index 000000000..55cb86ef0 --- /dev/null +++ b/pwnagotchi/ui/hw/displayhatmini.py @@ -0,0 +1,44 @@ +import logging + +import pwnagotchi.ui.fonts as fonts +from pwnagotchi.ui.hw.base import DisplayImpl + + +class DisplayHatMini(DisplayImpl): + def __init__(self, config): + super(DisplayHatMini, self).__init__(config, 'displayhatmini') + self._display = None + + def layout(self): + fonts.setup(12, 10, 12, 70, 25, 9) + self._layout['width'] = 320 + self._layout['height'] = 240 + self._layout['face'] = (35, 50) + self._layout['name'] = (5, 20) + self._layout['channel'] = (0, 0) + self._layout['aps'] = (40, 0) + self._layout['uptime'] = (240, 0) + self._layout['line1'] = [0, 14, 320, 14] + self._layout['line2'] = [0, 220, 320, 220] + self._layout['friend_face'] = (0, 130) + self._layout['friend_name'] = (40, 135) + self._layout['shakes'] = (0, 220) + self._layout['mode'] = (280, 220) + self._layout['status'] = { + 'pos': (80, 160), + 'font': fonts.status_font(fonts.Medium), + 'max': 20 + } + + return self._layout + + def initialize(self): + logging.info("initializing Display Hat Mini") + from pwnagotchi.ui.hw.libs.pimoroni.displayhatmini.ST7789 import ST7789 + self._display = ST7789(0,1,9,13) + + def render(self, canvas): + self._display.display(canvas) + + def clear(self): + self._display.clear() diff --git a/pwnagotchi/ui/hw/libs/pimoroni/displayhatmini/ST7789.py b/pwnagotchi/ui/hw/libs/pimoroni/displayhatmini/ST7789.py new file mode 100644 index 000000000..37a53e12a --- /dev/null +++ b/pwnagotchi/ui/hw/libs/pimoroni/displayhatmini/ST7789.py @@ -0,0 +1,360 @@ +# Copyright (c) 2014 Adafruit Industries +# Author: Tony DiCola +# +# Permission is hereby granted, free of charge, to any person obtaining a copy +# of this software and associated documentation files (the "Software"), to deal +# in the Software without restriction, including without limitation the rights +# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +# copies of the Software, and to permit persons to whom the Software is +# furnished to do so, subject to the following conditions: +# +# The above copyright notice and this permission notice shall be included in +# all copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN +# THE SOFTWARE. +import numbers +import time +import numpy as np + +import spidev +import RPi.GPIO as GPIO + + +__version__ = '0.0.4' + +BG_SPI_CS_BACK = 0 +BG_SPI_CS_FRONT = 1 + +SPI_CLOCK_HZ = 16000000 + +ST7789_NOP = 0x00 +ST7789_SWRESET = 0x01 +ST7789_RDDID = 0x04 +ST7789_RDDST = 0x09 + +ST7789_SLPIN = 0x10 +ST7789_SLPOUT = 0x11 +ST7789_PTLON = 0x12 +ST7789_NORON = 0x13 + +ST7789_INVOFF = 0x20 +ST7789_INVON = 0x21 +ST7789_DISPOFF = 0x28 +ST7789_DISPON = 0x29 + +ST7789_CASET = 0x2A +ST7789_RASET = 0x2B +ST7789_RAMWR = 0x2C +ST7789_RAMRD = 0x2E + +ST7789_PTLAR = 0x30 +ST7789_MADCTL = 0x36 +ST7789_COLMOD = 0x3A + +ST7789_FRMCTR1 = 0xB1 +ST7789_FRMCTR2 = 0xB2 +ST7789_FRMCTR3 = 0xB3 +ST7789_INVCTR = 0xB4 +ST7789_DISSET5 = 0xB6 + +ST7789_GCTRL = 0xB7 +ST7789_GTADJ = 0xB8 +ST7789_VCOMS = 0xBB + +ST7789_LCMCTRL = 0xC0 +ST7789_IDSET = 0xC1 +ST7789_VDVVRHEN = 0xC2 +ST7789_VRHS = 0xC3 +ST7789_VDVS = 0xC4 +ST7789_VMCTR1 = 0xC5 +ST7789_FRCTRL2 = 0xC6 +ST7789_CABCCTRL = 0xC7 + +ST7789_RDID1 = 0xDA +ST7789_RDID2 = 0xDB +ST7789_RDID3 = 0xDC +ST7789_RDID4 = 0xDD + +ST7789_GMCTRP1 = 0xE0 +ST7789_GMCTRN1 = 0xE1 + +ST7789_PWCTR6 = 0xFC + + +class ST7789(object): + """Representation of an ST7789 TFT LCD.""" + + def __init__(self, port, cs, dc, backlight, rst=None, width=320, + height=240, rotation=0, invert=True, spi_speed_hz=60 * 1000 * 1000, + offset_left=0, + offset_top=0): + """Create an instance of the display using SPI communication. + + Must provide the GPIO pin number for the D/C pin and the SPI driver. + + Can optionally provide the GPIO pin number for the reset pin as the rst parameter. + + :param port: SPI port number + :param cs: SPI chip-select number (0 or 1 for BCM + :param backlight: Pin for controlling backlight + :param rst: Reset pin for ST7789 + :param width: Width of display connected to ST7789 + :param height: Height of display connected to ST7789 + :param rotation: Rotation of display connected to ST7789 + :param invert: Invert display + :param spi_speed_hz: SPI speed (in Hz) + + """ + if rotation not in [0, 90, 180, 270]: + raise ValueError("Invalid rotation {}".format(rotation)) + + if width != height and rotation in [90, 270]: + raise ValueError("Invalid rotation {} for {}x{} resolution".format(rotation, width, height)) + + GPIO.setwarnings(False) + GPIO.setmode(GPIO.BCM) + + self._spi = spidev.SpiDev(port, cs) + self._spi.mode = 0 + self._spi.lsbfirst = False + self._spi.max_speed_hz = spi_speed_hz + + self._dc = dc + self._rst = rst + self._width = width + self._height = height + self._rotation = rotation + self._invert = invert + + self._offset_left = offset_left + self._offset_top = offset_top + + # Set DC as output. + GPIO.setup(dc, GPIO.OUT) + + # Setup backlight as output (if provided). + self._backlight = backlight + if backlight is not None: + GPIO.setup(backlight, GPIO.OUT) + GPIO.output(backlight, GPIO.LOW) + time.sleep(0.1) + GPIO.output(backlight, GPIO.HIGH) + + # Setup reset as output (if provided). + if rst is not None: + GPIO.setup(self._rst, GPIO.OUT) + self.reset() + self._init() + + def send(self, data, is_data=True, chunk_size=4096): + """Write a byte or array of bytes to the display. Is_data parameter + controls if byte should be interpreted as display data (True) or command + data (False). Chunk_size is an optional size of bytes to write in a + single SPI transaction, with a default of 4096. + """ + # Set DC low for command, high for data. + GPIO.output(self._dc, is_data) + # Convert scalar argument to list so either can be passed as parameter. + if isinstance(data, numbers.Number): + data = [data & 0xFF] + # Write data a chunk at a time. + for start in range(0, len(data), chunk_size): + end = min(start + chunk_size, len(data)) + self._spi.xfer(data[start:end]) + + def set_backlight(self, value): + """Set the backlight on/off.""" + if self._backlight is not None: + GPIO.output(self._backlight, value) + + @property + def width(self): + return self._width if self._rotation == 0 or self._rotation == 180 else self._height + + @property + def height(self): + return self._height if self._rotation == 0 or self._rotation == 180 else self._width + + def command(self, data): + """Write a byte or array of bytes to the display as command data.""" + self.send(data, False) + + def data(self, data): + """Write a byte or array of bytes to the display as display data.""" + self.send(data, True) + + def reset(self): + """Reset the display, if reset pin is connected.""" + if self._rst is not None: + GPIO.output(self._rst, 1) + time.sleep(0.500) + GPIO.output(self._rst, 0) + time.sleep(0.500) + GPIO.output(self._rst, 1) + time.sleep(0.500) + + def _init(self): + # Initialize the display. + + self.command(ST7789_SWRESET) # Software reset + time.sleep(0.150) # delay 150 ms + + self.command(ST7789_MADCTL) + self.data(0x70) + + self.command(ST7789_FRMCTR2) # Frame rate ctrl - idle mode + self.data(0x0C) + self.data(0x0C) + self.data(0x00) + self.data(0x33) + self.data(0x33) + + self.command(ST7789_COLMOD) + self.data(0x05) + + self.command(ST7789_GCTRL) + self.data(0x14) + + self.command(ST7789_VCOMS) + self.data(0x37) + + self.command(ST7789_LCMCTRL) # Power control + self.data(0x2C) + + self.command(ST7789_VDVVRHEN) # Power control + self.data(0x01) + + self.command(ST7789_VRHS) # Power control + self.data(0x12) + + self.command(ST7789_VDVS) # Power control + self.data(0x20) + + self.command(0xD0) + self.data(0xA4) + self.data(0xA1) + + self.command(ST7789_FRCTRL2) + self.data(0x0F) + + self.command(ST7789_GMCTRP1) # Set Gamma + self.data(0xD0) + self.data(0x04) + self.data(0x0D) + self.data(0x11) + self.data(0x13) + self.data(0x2B) + self.data(0x3F) + self.data(0x54) + self.data(0x4C) + self.data(0x18) + self.data(0x0D) + self.data(0x0B) + self.data(0x1F) + self.data(0x23) + + self.command(ST7789_GMCTRN1) # Set Gamma + self.data(0xD0) + self.data(0x04) + self.data(0x0C) + self.data(0x11) + self.data(0x13) + self.data(0x2C) + self.data(0x3F) + self.data(0x44) + self.data(0x51) + self.data(0x2F) + self.data(0x1F) + self.data(0x1F) + self.data(0x20) + self.data(0x23) + + if self._invert: + self.command(ST7789_INVON) # Invert display + else: + self.command(ST7789_INVOFF) # Don't invert display + + self.command(ST7789_SLPOUT) + + self.command(ST7789_DISPON) # Display on + time.sleep(0.100) # 100 ms + + def begin(self): + """Set up the display + + Deprecated. Included in __init__. + + """ + pass + + def set_window(self, x0=0, y0=0, x1=None, y1=None): + """Set the pixel address window for proceeding drawing commands. x0 and + x1 should define the minimum and maximum x pixel bounds. y0 and y1 + should define the minimum and maximum y pixel bound. If no parameters + are specified the default will be to update the entire display from 0,0 + to width-1,height-1. + """ + if x1 is None: + x1 = self._width - 1 + + if y1 is None: + y1 = self._height - 1 + + y0 += self._offset_top + y1 += self._offset_top + + x0 += self._offset_left + x1 += self._offset_left + + self.command(ST7789_CASET) # Column addr set + self.data(x0 >> 8) + self.data(x0 & 0xFF) # XSTART + self.data(x1 >> 8) + self.data(x1 & 0xFF) # XEND + self.command(ST7789_RASET) # Row addr set + self.data(y0 >> 8) + self.data(y0 & 0xFF) # YSTART + self.data(y1 >> 8) + self.data(y1 & 0xFF) # YEND + self.command(ST7789_RAMWR) # write to RAM + + def display(self, image): + """Write the provided image to the hardware. + + :param image: Should be RGB format and the same dimensions as the display hardware. + + """ + # Set address bounds to entire display. + self.set_window() + + # Convert image to 16bit RGB565 format and + # flatten into bytes. + pixelbytes = self.image_to_data(image, self._rotation) + + # Write data to hardware. + for i in range(0, len(pixelbytes), 4096): + self.data(pixelbytes[i:i + 4096]) + + def image_to_data(self, image, rotation=0): + if not isinstance(image, np.ndarray): + image = np.array(image.convert('RGB')) + + # Rotate the image + pb = np.rot90(image, rotation // 90).astype('uint16') + + # Mask and shift the 888 RGB into 565 RGB + red = (pb[..., [0]] & 0xf8) << 8 + green = (pb[..., [1]] & 0xfc) << 3 + blue = (pb[..., [2]] & 0xf8) >> 3 + + # Stick 'em together + result = red | green | blue + + # Output the raw bytes + return result.byteswap().tobytes() diff --git a/pwnagotchi/utils.py b/pwnagotchi/utils.py index 29c363276..0a9a5cf0c 100644 --- a/pwnagotchi/utils.py +++ b/pwnagotchi/utils.py @@ -283,7 +283,10 @@ def load_config(args): elif config['ui']['display']['type'] in ('spotpear24inch'): config['ui']['display']['type'] = 'spotpear24inch' - + + elif config['ui']['display']['type'] in ('displayhatmini'): + config['ui']['display']['type'] = 'displayhatmini' + else: print("unsupported display type %s" % config['ui']['display']['type']) sys.exit(1)