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

Merge branch 'pwnagotchi-torch-testing' into pwnagotchi-torch-64

Browse Source 
# Conflicts:
#	requirements.txt
This commit is contained in:
Jeroen Oudshoorn
2023-10-15 22:38:18 +02:00
parent 096aa95afa a09415df67
commit bcc79a2457
35 changed files with 494 additions and 129 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
bin/pwnagotchi
View File

@ -10,6 +10,7 @@ import os
import pwnagotchi
from pwnagotchi import utils
from pwnagotchi.google import cmd as google_cmd
from pwnagotchi.plugins import cmd as plugins_cmd
from pwnagotchi import log
from pwnagotchi import restart
@ -96,9 +97,20 @@ def do_auto_mode(agent):
if __name__ == '__main__':
parser = argparse.ArgumentParser()
parser = plugins_cmd.add_parsers(parser)
def add_parsers(parser):
"""
Adds the plugins and google subcommands to a given argparse.ArgumentParser
"""
subparsers = parser.add_subparsers()
# Add parsers from plugins_cmd
plugins_cmd.add_parsers(subparsers)
# Add parsers from google_cmd
google_cmd.add_parsers(subparsers)
parser = argparse.ArgumentParser(prog="pwnagotchi")
# pwnagotchi --help
parser.add_argument('-C', '--config', action='store', dest='config', default='/etc/pwnagotchi/default.toml',
help='Main configuration file.')
parser.add_argument('-U', '--user-config', action='store', dest='user_config', default='/etc/pwnagotchi/config.toml',
@ -126,6 +138,8 @@ if __name__ == '__main__':
parser.add_argument('--donate', dest="donate", action="store_true", default=False,
help="How to donate to this project.")
# pwnagotchi plugins --help
add_parsers(parser)
args = parser.parse_args()
if plugins_cmd.used_plugin_cmd(args):
@ -133,6 +147,11 @@ if __name__ == '__main__':
log.setup_logging(args, config)
rc = plugins_cmd.handle_cmd(args, config)
sys.exit(rc)
if google_cmd.used_google_cmd(args):
config = utils.load_config(args)
log.setup_logging(args, config)
rc = google_cmd.handle_cmd(args)
sys.exit(rc)
if args.version:
print(pwnagotchi.__version__)

5
builder/data/etc/pwnagotchi/conf.d/age.toml Normal file
View File

@ -0,0 +1,5 @@
main.plugins.age.enabled = false
main.plugins.age.age_x_coord = 0
main.plugins.age.age_y_coord = 32
main.plugins.age.str_x_coord = 67
main.plugins.age.str_y_coord = 32

3
builder/data/etc/pwnagotchi/conf.d/auto-update.toml Normal file
View File

@ -0,0 +1,3 @@
main.plugins.auto-update.enabled = true
main.plugins.auto-update.install = true
main.plugins.auto-update.interval = 1

23
builder/data/etc/pwnagotchi/conf.d/bt-tether.toml Normal file
View File

@ -0,0 +1,23 @@
main.plugins.bt-tether.enabled = false
main.plugins.bt-tether.devices.android-phone.enabled = false
main.plugins.bt-tether.devices.android-phone.search_order = 1
main.plugins.bt-tether.devices.android-phone.mac = ""
main.plugins.bt-tether.devices.android-phone.ip = "192.168.44.44"
main.plugins.bt-tether.devices.android-phone.netmask = 24
main.plugins.bt-tether.devices.android-phone.interval = 1
main.plugins.bt-tether.devices.android-phone.scantime = 10
main.plugins.bt-tether.devices.android-phone.max_tries = 10
main.plugins.bt-tether.devices.android-phone.share_internet = false
main.plugins.bt-tether.devices.android-phone.priority = 1
main.plugins.bt-tether.devices.ios-phone.enabled = false
main.plugins.bt-tether.devices.ios-phone.search_order = 2
main.plugins.bt-tether.devices.ios-phone.mac = ""
main.plugins.bt-tether.devices.ios-phone.ip = "172.20.10.6"
main.plugins.bt-tether.devices.ios-phone.netmask = 24
main.plugins.bt-tether.devices.ios-phone.interval = 5
main.plugins.bt-tether.devices.ios-phone.scantime = 20
main.plugins.bt-tether.devices.ios-phone.max_tries = 0
main.plugins.bt-tether.devices.ios-phone.share_internet = false
main.plugins.bt-tether.devices.ios-phone.priority = 999

1
builder/data/etc/pwnagotchi/conf.d/fix_services.toml Normal file
View File

@ -0,0 +1 @@
main.plugins.fix_services.enabled = true

4
builder/data/etc/pwnagotchi/conf.d/gdrivesync.toml Normal file
View File

@ -0,0 +1,4 @@
main.plugins.gdrivesync.enabled = false
main.plugins.gdrivesync.backupfiles = ['']
main.plugins.gdrivesync.backup_folder = "PwnagotchiBackups"
main.plugin.gdrivesync.interval = 1

1
builder/data/etc/pwnagotchi/conf.d/gpio_buttons.toml Normal file
View File

@ -0,0 +1 @@
main.plugins.gpio_buttons.enabled = false

3
builder/data/etc/pwnagotchi/conf.d/gps.toml Normal file
View File

@ -0,0 +1,3 @@
main.plugins.gps.enabled = false
main.plugins.gps.speed = 19200
main.plugins.gps.device = "/dev/ttyUSB0"

5
builder/data/etc/pwnagotchi/conf.d/grid.toml Normal file
View File

@ -0,0 +1,5 @@
main.plugins.grid.enabled = true
main.plugins.grid.report = true
main.plugins.grid.exclude = [
"YourHomeNetworkHere"
]

2
builder/data/etc/pwnagotchi/conf.d/logtail.toml Normal file
View File

@ -0,0 +1,2 @@
main.plugins.logtail.enabled = false
main.plugins.logtail.max-lines = 10000

3
builder/data/etc/pwnagotchi/conf.d/memtemp.toml Normal file
View File

@ -0,0 +1,3 @@
main.plugins.memtemp.enabled = false
main.plugins.memtemp.scale = "celsius"
main.plugins.memtemp.orientation = "horizontal"

2
builder/data/etc/pwnagotchi/conf.d/net-pos.toml Normal file
View File

@ -0,0 +1,2 @@
main.plugins.net-pos.enabled = false
main.plugins.net-pos.api_key = "test"

5
builder/data/etc/pwnagotchi/conf.d/onlinehascrack.toml Normal file
View File

@ -0,0 +1,5 @@
main.plugins.onlinehashcrack.enabled = false
main.plugins.onlinehashcrack.email = ""
main.plugins.onlinehashcrack.dashboard = ""
main.plugins.onlinehashcrack.single_files = false
main.plugins.onlinehashcrack.whitelist = []

2
builder/data/etc/pwnagotchi/conf.d/paw-gps.toml Normal file
View File

@ -0,0 +1,2 @@
main.plugins.paw-gps.enabled = false
main.plugins.paw-gps.ip = "192.168.44.1:8080"

3
builder/data/etc/pwnagotchi/conf.d/pisugar2.toml Normal file
View File

@ -0,0 +1,3 @@
main.plugins.pisugar2.enabled = false
main.plugins.pisugar2.shutdown = 5
main.plugins.pisugar2.sync_rtc_on_boot = false

2
builder/data/etc/pwnagotchi/conf.d/session-stats.toml Normal file
View File

@ -0,0 +1,2 @@
main.plugins.session-stats.enabled = true
main.plugins.session-stats.save_directory = "/var/tmp/pwnagotchi/sessions/"

5
builder/data/etc/pwnagotchi/conf.d/ups_hat_c.toml Normal file
View File

@ -0,0 +1,5 @@
main.plugins.ups_hat_c.enabled = false
main.plugins.ups_hat_c.label_on = true # show BAT label or just percentage
main.plugins.ups_hat_c.shutdown = 5 # battery percent at which the device will turn off
main.plugins.ups_hat_c.bat_x_coord = 140
main.plugins.ups_hat_c.bat_y_coord = 0

2
builder/data/etc/pwnagotchi/conf.d/ups_lite.toml Normal file
View File

@ -0,0 +1,2 @@
main.plugins.ups_lite.enabled = false
main.plugins.ups_lite.shutdown = 2

1
builder/data/etc/pwnagotchi/conf.d/webcfg.toml Normal file
View File

@ -0,0 +1 @@
main.plugins.webcfg.enabled = true

1
builder/data/etc/pwnagotchi/conf.d/webgpsmap.toml Normal file
View File

@ -0,0 +1 @@
main.plugins.webgpsmap.enabled = false

4
builder/data/etc/pwnagotchi/conf.d/wigle.toml Normal file
View File

@ -0,0 +1,4 @@
main.plugins.wigle.enabled = false
main.plugins.wigle.api_key = ""
main.plugins.wigle.whitelist = []
main.plugins.wigle.donate = true

5
builder/data/etc/pwnagotchi/conf.d/wpa-sec.toml Normal file
View File

@ -0,0 +1,5 @@
main.plugins.wpa-sec.enabled = false
main.plugins.wpa-sec.api_key = ""
main.plugins.wpa-sec.api_url = "https://wpa-sec.stanev.org"
main.plugins.wpa-sec.download_results = false
main.plugins.wpa-sec.whitelist = []

2
builder/data/etc/systemd/system/pwnagotchi.service
View File

@ -6,7 +6,7 @@ After=pwngrid-peer.service
[Service]
Type=simple
WorkingDirectory=/tmp
WorkingDirectory=~
ExecStart=/usr/bin/pwnagotchi-launcher
Restart=always
RestartSec=30

4
builder/data/etc/update-motd.d/01-motd
View File

@ -23,9 +23,9 @@ echo " - sudo pwnagotchi --donate, to see how you can donate to this project"
echo " - sudo pwnagotchi --check-update, to see if there is a new version available"
echo
echo " If you want to know if I'm running, you can use"
echo " systemctl status pwnagotchi"
echo " sudo systemctl status pwnagotchi"
echo
echo " You can restart me using"
echo " systemctl restart pwnagotchi"
echo " sudo systemctl restart pwnagotchi"
echo
echo " You learn more about me at https://pwnagotchi.ai/"

0
builder/data/root/client_secrets.json Normal file
View File

0
builder/data/root/settings.yaml Normal file
View File

10
builder/pwnagotchi.yml
View File

@ -381,6 +381,11 @@
path: /usr/local/share/pwnagotchi/custom-plugins/
state: directory
- name: Create custom config directory
file:
path: /etc/pwnagotchi/conf.d/
state: directory
- name: clone pwnagotchi repository
git:
repo: https://github.com/jayofelony/pwnagotchi.git
@ -627,6 +632,11 @@
autoclean: true
when: removed.changed
- name: apt clean
shell: "apt clean"
args:
executable: /bin/bash
- name: remove dependencies that are no longer required
apt:
autoremove: yes

26
pwnagotchi/ai/__init__.py
View File

@ -3,7 +3,7 @@ import time
import logging
# https://stackoverflow.com/questions/40426502/is-there-a-way-to-suppress-the-messages-tensorflow-prints/40426709
os.environ['TF_CPP_MIN_LOG_LEVEL'] = '3' # or any {'0', '1', '2'}
# os.environ['TF_CPP_MIN_LOG_LEVEL'] = '3' # or any {'0', '1', '2'}
def load(config, agent, epoch, from_disk=True):
@ -15,47 +15,47 @@ def load(config, agent, epoch, from_disk=True):
try:
begin = time.time()
logging.info("[ai] bootstrapping dependencies ...")
logging.info("[AI] bootstrapping dependencies ...")
start = time.time()
SB_BACKEND = "stable_baselines3"
from stable_baselines3 import A2C
logging.debug("[ai] A2C imported in %.2fs" % (time.time() - start))
logging.debug("[AI] A2C imported in %.2fs" % (time.time() - start))
start = time.time()
from stable_baselines3.a2c import MlpPolicy
logging.debug("[ai] MlpPolicy imported in %.2fs" % (time.time() - start))
logging.debug("[AI] MlpPolicy imported in %.2fs" % (time.time() - start))
SB_A2C_POLICY = MlpPolicy
start = time.time()
from stable_baselines3.common.vec_env import DummyVecEnv
logging.debug("[ai] DummyVecEnv imported in %.2fs" % (time.time() - start))
logging.debug("[AI] DummyVecEnv imported in %.2fs" % (time.time() - start))
start = time.time()
import pwnagotchi.ai.gym as wrappers
logging.debug("[ai] gym wrapper imported in %.2fs" % (time.time() - start))
logging.debug("[AI] gym wrapper imported in %.2fs" % (time.time() - start))
env = wrappers.Environment(agent, epoch)
env = DummyVecEnv([lambda: env])
logging.info("[ai] creating model ...")
logging.info("[AI] creating model ...")
start = time.time()
a2c = A2C(SB_A2C_POLICY, env, **config['params'])
logging.debug("[ai] A2C created in %.2fs" % (time.time() - start))
logging.debug("[AI] A2C created in %.2fs" % (time.time() - start))
if from_disk and os.path.exists(config['path']):
logging.info("[ai] loading %s ..." % config['path'])
logging.info("[AI] loading %s ..." % config['path'])
start = time.time()
a2c.load(config['path'], env)
logging.debug("[ai] A2C loaded in %.2fs" % (time.time() - start))
logging.debug("[AI] A2C loaded in %.2fs" % (time.time() - start))
else:
logging.info("[ai] model created:")
logging.info("[AI] model created:")
for key, value in config['params'].items():
logging.info(" %s: %s" % (key, value))
logging.debug("[ai] total loading time is %.2fs" % (time.time() - begin))
logging.debug("[AI] total loading time is %.2fs" % (time.time() - begin))
return a2c
except Exception as e:
@ -63,5 +63,5 @@ def load(config, agent, epoch, from_disk=True):
logging.info("[AI] Deleting brain and restarting.")
os.system("rm /root/brain.nn && service pwnagotchi restart")
logging.warning("[ai] AI not loaded!")
logging.warning("[AI] AI not loaded!")
return False

8
pwnagotchi/ai/gym.py
View File

@ -138,13 +138,13 @@ class Environment(gym.Env):
self._last_render = self._epoch_num
logging.info("[ai] --- training epoch %d/%d ---" % (self._epoch_num, self._agent.training_epochs()))
logging.info("[ai] REWARD: %f" % self.last['reward'])
logging.info("[AI] --- training epoch %d/%d ---" % (self._epoch_num, self._agent.training_epochs()))
logging.info("[AI] REWARD: %f" % self.last['reward'])
logging.debug(
"[ai] policy: %s" % ', '.join("%s:%s" % (name, value) for name, value in self.last['params'].items()))
"[AI] policy: %s" % ', '.join("%s:%s" % (name, value) for name, value in self.last['params'].items()))
logging.info("[ai] observation:")
logging.info("[AI] observation:")
for name, value in self.last['state'].items():
if 'histogram' in name:
logging.info(" %s" % name.replace('_histogram', ''))

99
pwnagotchi/defaults.toml
View File

@ -7,9 +7,8 @@ main.custom_plugin_repos = [
"https://github.com/tisboyo/pwnagotchi-pisugar2-plugin/archive/master.zip",
"https://github.com/nullm0ose/pwnagotchi-plugin-pisugar3/archive/master.zip"
]
main.custom_plugins = "/usr/local/share/pwnagotchi/custom-plugins/"
main.plugins.fix_services.enabled = true
main.custom_plugins = "/usr/local/share/pwnagotchi/custom-plugins/"
main.iface = "wlan0mon"
main.mon_start_cmd = "/usr/bin/monstart"
@ -24,102 +23,6 @@ main.whitelist = [
]
main.filter = ""
main.plugins.age.enabled = false
main.plugins.age.age_x_coord = 0
main.plugins.age.age_y_coord = 32
main.plugins.age.str_x_coord = 67
main.plugins.age.str_y_coord = 32
main.plugins.ups_hat_c.enabled = false
main.plugins.ups_hat_c.label_on = true # show BAT label or just percentage
main.plugins.ups_hat_c.shutdown = 5 # battery percent at which the device will turn off
main.plugins.ups_hat_c.bat_x_coord = 140
main.plugins.ups_hat_c.bat_y_coord = 0
main.plugins.pisugar2.enabled = false
main.plugins.pisugar2.shutdown = 5
main.plugins.pisugar2.sync_rtc_on_boot = false
main.plugins.grid.enabled = true
main.plugins.grid.report = true
main.plugins.grid.exclude = [
"YourHomeNetworkHere"
]
main.plugins.auto-update.enabled = true
main.plugins.auto-update.install = true
main.plugins.auto-update.interval = 1
main.plugins.net-pos.enabled = false
main.plugins.net-pos.api_key = "test"
main.plugins.gps.enabled = false
main.plugins.gps.speed = 19200
main.plugins.gps.device = "/dev/ttyUSB0"
main.plugins.webgpsmap.enabled = false
main.plugins.onlinehashcrack.enabled = false
main.plugins.onlinehashcrack.email = ""
main.plugins.onlinehashcrack.dashboard = ""
main.plugins.onlinehashcrack.single_files = false
main.plugins.onlinehashcrack.whitelist = []
main.plugins.wpa-sec.enabled = false
main.plugins.wpa-sec.api_key = ""
main.plugins.wpa-sec.api_url = "https://wpa-sec.stanev.org"
main.plugins.wpa-sec.download_results = false
main.plugins.wpa-sec.whitelist = []
main.plugins.wigle.enabled = false
main.plugins.wigle.api_key = ""
main.plugins.wigle.whitelist = []
main.plugins.wigle.donate = true
main.plugins.bt-tether.enabled = false
main.plugins.bt-tether.devices.android-phone.enabled = false
main.plugins.bt-tether.devices.android-phone.search_order = 1
main.plugins.bt-tether.devices.android-phone.mac = ""
main.plugins.bt-tether.devices.android-phone.ip = "192.168.44.44"
main.plugins.bt-tether.devices.android-phone.netmask = 24
main.plugins.bt-tether.devices.android-phone.interval = 1
main.plugins.bt-tether.devices.android-phone.scantime = 10
main.plugins.bt-tether.devices.android-phone.max_tries = 10
main.plugins.bt-tether.devices.android-phone.share_internet = false
main.plugins.bt-tether.devices.android-phone.priority = 1
main.plugins.bt-tether.devices.ios-phone.enabled = false
main.plugins.bt-tether.devices.ios-phone.search_order = 2
main.plugins.bt-tether.devices.ios-phone.mac = ""
main.plugins.bt-tether.devices.ios-phone.ip = "172.20.10.6"
main.plugins.bt-tether.devices.ios-phone.netmask = 24
main.plugins.bt-tether.devices.ios-phone.interval = 5
main.plugins.bt-tether.devices.ios-phone.scantime = 20
main.plugins.bt-tether.devices.ios-phone.max_tries = 0
main.plugins.bt-tether.devices.ios-phone.share_internet = false
main.plugins.bt-tether.devices.ios-phone.priority = 999
main.plugins.memtemp.enabled = false
main.plugins.memtemp.scale = "celsius"
main.plugins.memtemp.orientation = "horizontal"
main.plugins.paw-gps.enabled = false
main.plugins.paw-gps.ip = "192.168.44.1:8080"
main.plugins.ups_lite.enabled = false
main.plugins.ups_lite.shutdown = 2
main.plugins.gpio_buttons.enabled = false
main.plugins.webcfg.enabled = true
main.plugins.logtail.enabled = false
main.plugins.logtail.max-lines = 10000
main.plugins.session-stats.enabled = true
main.plugins.session-stats.save_directory = "/var/tmp/pwnagotchi/sessions/"
main.log.path = "/var/log/pwnagotchi.log"
main.log.rotation.enabled = true
main.log.rotation.size = "10M"

101
pwnagotchi/google/cmd.py Normal file
View File

@ -0,0 +1,101 @@
# Handles the commandline stuff
import pydrive2
from pydrive2.auth import GoogleAuth
import logging
import os
def add_parsers(subparsers):
"""
Adds the plugins subcommand to a given argparse.ArgumentParser
"""
#subparsers = parser.add_subparsers()
# pwnagotchi google
parser_google = subparsers.add_parser('google')
google_subparsers = parser_google.add_subparsers(dest='googlecmd')
# pwnagotchi google auth
parser_google_auth = google_subparsers.add_parser('login', help='Login to Google')
# pwnagotchi google refresh token
parser_google_refresh = google_subparsers.add_parser('refresh', help="Refresh Google authentication token")
return subparsers
def used_google_cmd(args):
"""
Checks if the plugins subcommand was used
"""
return hasattr(args, 'googlecmd')
def handle_cmd(args):
"""
Parses the arguments and does the thing the user wants
"""
if args.googlecmd == 'login':
return auth()
elif args.googlecmd == 'refresh':
return refresh()
raise NotImplementedError()
def auth():
# start authentication process
user_input = input("By completing these steps you give pwnagotchi access to your personal Google Drive!\n"
"Personal credentials will be stored only locally for automated verification in the future.\n"
"No one else but you have access to these.\n"
"Do you agree? \n\n[y(es)/n(o)]\n"
"Answer: ")
if user_input.lower() in ('y', 'yes'):
if not os.path.exists("/root/client_secrets.json"):
logging.error("client_secrets.json not found in /root. Please RTFM!")
return 0
try:
gauth = GoogleAuth(settings_file="/root/settings.yaml")
print(gauth.GetAuthUrl())
user_input = input("Please copy this URL into a browser, "
"complete the verification and then copy/paste the code from addressbar.\n\n"
"Code: ")
gauth.Auth(user_input)
gauth.SaveCredentialsFile("/root/credentials.json")
except Exception as e:
logging.error(f"Error: {e}")
return 0
def refresh():
# refresh token for x amount of time (seconds)
gauth = GoogleAuth(settings_file="/root/settings.yaml")
try:
# Try to load saved client credentials
gauth.LoadCredentialsFile("/root/credentials.json")
if gauth.access_token_expired:
if gauth.credentials is not None:
try:
# Refresh the token
gauth.Refresh()
print("Succesfully refresh access token ..")
except pydrive2.auth.RefreshError:
print(gauth.GetAuthUrl())
user_input = input("Please copy this URL into a browser, "
"complete the verification and then copy/paste the code from addressbar.\n\n"
"Code: ")
gauth.Auth(user_input)
else:
print(gauth.GetAuthUrl())
user_input = input("Please copy this URL into a browser, "
"complete the verification and then copy/paste the code from addressbar.\n\n"
"Code: ")
gauth.Auth(user_input)
except pydrive2.auth.InvalidCredentialsError:
print(gauth.GetAuthUrl())
user_input = input("Please copy this URL into a browser, "
"complete the verification and then copy/paste the code from addressbar.\n\n"
"Code: ")
gauth.Auth(user_input)
gauth.SaveCredentialsFile("/root/credentials.json")
gauth.Authorize()
print("No refresh required ..")
return 0

10
pwnagotchi/plugins/cmd.py
View File

@ -14,11 +14,11 @@ SAVE_DIR = '/usr/local/share/pwnagotchi/available-plugins/'
DEFAULT_INSTALL_PATH = '/usr/local/share/pwnagotchi/installed-plugins/'
def add_parsers(parser):
def add_parsers(subparsers):
"""
Adds the plugins subcommand to a given argparse.ArgumentParser
"""
subparsers = parser.add_subparsers()
#subparsers = parser.add_subparsers()
# pwnagotchi plugins
parser_plugins = subparsers.add_parser('plugins')
plugin_subparsers = parser_plugins.add_subparsers(dest='plugincmd')
@ -58,7 +58,7 @@ def add_parsers(parser):
parser_plugins_edit = plugin_subparsers.add_parser('edit', help='Edit the options')
parser_plugins_edit.add_argument('name', type=str, help='Name of the plugin')
return parser
return subparsers
def used_plugin_cmd(args):
@ -75,7 +75,7 @@ def handle_cmd(args, config):
if args.plugincmd == 'update':
return update(config)
elif args.plugincmd == 'search':
args.installed = True # also search in installed plugins
args.installed = True # also search in installed plugins
return list_plugins(args, config, args.pattern)
elif args.plugincmd == 'install':
return install(args, config)
@ -271,7 +271,7 @@ def _get_installed(config):
Get all installed plugins
"""
installed = dict()
search_dirs = [ default_path, config['main']['custom_plugins'] ]
search_dirs = [default_path, config['main']['custom_plugins']]
for search_dir in search_dirs:
if search_dir:
for filename in glob.glob(os.path.join(search_dir, "*.py")):

248
pwnagotchi/plugins/default/gdrivesync.py Normal file
View File

@ -0,0 +1,248 @@
import logging
import os
import shutil
import time
import pwnagotchi.plugins as plugins
import pwnagotchi
import pydrive2
from pydrive2.auth import GoogleAuth
from pydrive2.drive import GoogleDrive
from threading import Lock
from pwnagotchi.utils import StatusFile
import zipfile
class GdriveSync(plugins.Plugin):
__author__ = '@jayofelony'
__version__ = '1.0'
__license__ = 'GPL3'
__description__ = 'A plugin to backup various pwnagotchi files and folders to Google Drive. Once every hour from loading plugin.'
__dependencies__ = {
'pip': ['pydrive2']
}
def __init__(self):
self.lock = Lock()
self.internet = False
self.ready = False
self.drive = None
self.status = StatusFile('/root/.gdrive-backup')
self.backup = True
self.backupfiles = [
'/root/brain.nn',
'/root/brain.json',
'/root/.api-report.json',
'/root/handshakes',
'/root/peers',
'/etc/pwnagotchi'
]
def on_loaded(self):
"""
Called when the plugin is loaded
"""
# client_secrets.json needs to be not empty
if os.stat("/root/client_secrets.json").st_size == 0:
logging.error("[gDriveSync] /root/client_secrets.json is empty. Please RTFM!")
return
# backup file, so we know if there has been a backup made at least once before.
if not os.path.exists("/root/.gdrive-backup"):
self.backup = False
try:
gauth = GoogleAuth(settings_file="/root/settings.yaml")
gauth.LoadCredentialsFile("/root/credentials.json")
if gauth.credentials is None:
# Authenticate if they're not there
gauth.LocalWebserverAuth()
elif gauth.access_token_expired:
# Refresh them if expired
gauth.Refresh()
gauth.SaveCredentialsFile("/root/credentials.json")
gauth.Authorize()
# Create GoogleDrive instance
self.drive = GoogleDrive(gauth)
# if backup file does not exist, we will check for backup folder on gdrive.
if not self.backup:
# Use self.options['backup_folder'] as the folder ID where backups are stored
backup_folder = self.create_folder_if_not_exists(self.options['backup_folder'])
# Continue with the rest of the code using backup_folder_id
backup_folder_file_list = self.drive.ListFile({'q': f"'{backup_folder}' in parents and mimeType = 'application/vnd.google-apps.folder' and trashed=false"}).GetList()
if not backup_folder_file_list:
# Handle the case where no files were found
# logging.warning(f"[gDriveSync] No files found in the folder with ID {root_file_list} and {pwnagotchi_file_list}")
if self.options['backupfiles'] is not None:
self.backupfiles = self.backupfiles + self.options['backupfiles']
self.backup_files(self.backupfiles, '/backup')
# Create a zip archive of the /backup folder
zip_file_path = os.path.join('/home/pi', 'backup.zip')
with zipfile.ZipFile(zip_file_path, 'w') as zip_ref:
for root, dirs, files in os.walk('/backup'):
for file in files:
file_path = os.path.join(root, file)
arcname = os.path.relpath(file_path, '/backup')
zip_ref.write(file_path, arcname=arcname)
# Upload the zip archive to Google Drive
self.upload_to_gdrive(zip_file_path, self.get_folder_id_by_name(self.drive, self.options['backup_folder']))
self.backup = True
self.status.update()
# Specify the local backup path
local_backup_path = '/'
# Download the zip archive from Google Drive
zip_file_id = self.get_latest_backup_file_id(self.options['backup_folder'])
if zip_file_id:
zip_file = self.drive.CreateFile({'id': zip_file_id})
zip_file.GetContentFile(os.path.join(local_backup_path, 'backup.zip'))
logging.info("[gDriveSync] Downloaded backup.zip from Google Drive")
# Extract the zip archive to the root directory
with zipfile.ZipFile(os.path.join(local_backup_path, 'backup.zip'), 'r') as zip_ref:
zip_ref.extractall('/')
self.status.update()
os.remove("/backup")
# Reboot so we can start opwngrid with the backup id
pwnagotchi.reboot()
# all set, gdriveSync is ready to run
self.ready = True
logging.info("[gdrivesync] loaded")
except Exception as e:
logging.error(f"Error: {e}")
self.ready = False
def get_latest_backup_file_id(self, backup_folder_id):
backup_folder_id = self.get_folder_id_by_name(self.drive, backup_folder_id)
# Retrieve the latest backup file in the Google Drive folder
file_list = self.drive.ListFile({'q': f"'{backup_folder_id}' in parents and trashed=false"}).GetList()
if file_list:
# Sort the files by creation date in descending order
latest_backup = max(file_list, key=lambda file: file['createdDate'])
return latest_backup['id']
else:
return None
def get_folder_id_by_name(self, drive, folder_name, parent_folder_id=None):
query = "mimeType='application/vnd.google-apps.folder' and trashed=false"
if parent_folder_id:
query += f" and '{parent_folder_id}' in parents"
file_list = drive.ListFile({'q': query}).GetList()
for file in file_list:
if file['title'] == folder_name:
return file['id']
return None
def create_folder_if_not_exists(self, backup_folder_name):
# First, try to retrieve the existing *BACKUP_FOLDER* folder
backup_folder_id = self.get_folder_id_by_name(self.drive, backup_folder_name)
if backup_folder_id is None:
# If not found, create *BACKUP_FOLDER*
backup_folder = self.drive.CreateFile(
{'title': backup_folder_name, 'mimeType': 'application/vnd.google-apps.folder'})
backup_folder.Upload()
backup_folder_id = backup_folder['id']
logging.info(f"[gDriveSync] Created folder '{backup_folder_name}' with ID: {backup_folder_id}")
return backup_folder_id
def on_unload(self, ui):
"""
Called when the plugin is unloaded
"""
logging.info("[gdrivesync] unloaded")
def on_internet_available(self, agent):
"""
Called when internet is available
"""
self.internet = True
def on_handshake(self, agent):
if not self.ready and not self.internet:
return
if self.lock.locked():
return
with self.lock:
if self.status.newer_then_hours(self.options['interval']):
logging.debug("[update] last check happened less than %d hours ago" % self.options['interval'])
return
logging.info("[gdrivesync] new handshake captured, backing up to gdrive")
if self.options['backupfiles'] is not None:
self.backupfiles = self.backupfiles + self.options['backupfiles']
self.backup_files(self.backupfiles, '/backup')
# Create a zip archive of the /backup folder
zip_file_path = os.path.join('/home/pi', 'backup.zip')
with zipfile.ZipFile(zip_file_path, 'w') as zip_ref:
for root, dirs, files in os.walk('/backup'):
for file in files:
file_path = os.path.join(root, file)
arcname = os.path.relpath(file_path, '/backup')
zip_ref.write(file_path, arcname=arcname)
# Upload the zip archive to Google Drive
self.upload_to_gdrive(zip_file_path, self.get_folder_id_by_name(self.drive, self.options['backup_folder']))
# Cleanup the local zip file
os.remove(zip_file_path)
os.remove("/backup")
self.status.update()
display = agent.view()
display.update(force=True, new_data={'Backing up to gdrive ...'})
def backup_files(self, paths, dest_path):
for src_path in paths:
try:
if os.path.exists(src_path):
dest_relative_path = os.path.relpath(src_path, '/')
dest = os.path.join(dest_path, dest_relative_path)
if os.path.isfile(src_path):
# If it's a file, copy it to the destination preserving the directory structure
os.makedirs(os.path.dirname(dest), exist_ok=True)
# Check if the destination file already exists
if os.path.exists(dest):
# If it exists, remove it to overwrite
os.remove(dest)
elif os.path.isdir(src_path):
# If it's a directory, copy the entire directory to the destination
shutil.copytree(src_path, dest)
except Exception as e:
logging.error(f"[gDriveSync] Error during backup_path: {e}")
def upload_to_gdrive(self, backup_path, gdrive_folder):
try:
# Upload zip-file to google drive
# Create a GoogleDriveFile instance for the zip file
zip_file = self.drive.CreateFile({'title': 'backup.zip', 'parents': [{'id': gdrive_folder}]})
# Set the content of the file to the zip file
zip_file.SetContentFile(backup_path)
# Upload the file to Google Drive
zip_file.Upload()
logging.info(f"[gDriveSync] Backup uploaded to Google Drive")
except pydrive2.files.ApiRequestError as api_error:
self.handle_upload_error(api_error, backup_path, gdrive_folder)
except Exception as e:
logging.error(f"[gDriveSync] Error during upload_to_gdrive: {e}")
def handle_upload_error(self, api_error, backup_path, gdrive_folder):
if 'Rate Limit Exceeded' in str(api_error):
logging.warning("[gDriveSync] Rate limit exceeded. Waiting for some time before retrying...")
# We set to 100 seconds, because there is a limit 20k requests per 100s per user
time.sleep(100) # You can adjust the sleep duration based on your needs
self.upload_to_gdrive(backup_path, gdrive_folder)
else:
logging.error(f"[gDriveSync] API Request Error: {api_error}")

7
pwnagotchi/plugins/default/webcfg.py
View File

@ -479,11 +479,13 @@ INDEX = """
{% endblock %}
"""
def serializer(obj):
if isinstance(obj, set):
return list(obj)
raise TypeError
class WebConfig(plugins.Plugin):
__author__ = '33197631+dadav@users.noreply.github.com'
__version__ = '1.0.0'
@ -513,7 +515,6 @@ class WebConfig(plugins.Plugin):
"""
logging.info("webcfg: Plugin loaded.")
def on_webhook(self, path, request):
"""
Serves the current configuration
@ -532,7 +533,7 @@ class WebConfig(plugins.Plugin):
elif request.method == "POST":
if path == "save-config":
try:
save_config(request.get_json(), '/etc/pwnagotchi/config.toml') # test
save_config(request.get_json(), '/etc/pwnagotchi/config.toml') # test
_thread.start_new_thread(restart, (self.mode,))
return "success"
except Exception as ex:
@ -547,7 +548,7 @@ class WebConfig(plugins.Plugin):
self._agent._config = merge_config(request.get_json(), self._agent._config)
logging.debug(" Agent CONFIG:\n%s" % repr(self._agent._config))
logging.debug(" Updated CONFIG:\n%s" % request.get_json())
save_config(request.get_json(), '/etc/pwnagotchi/config.toml') # test
save_config(request.get_json(), '/etc/pwnagotchi/config.toml') # test
return "success"
except Exception as ex:
logging.error("[webcfg mergesave] %s" % ex)

3
setup.py
View File

@ -11,7 +11,8 @@ import re
def install_file(source_filename, dest_filename):
# do not overwrite network configuration if it exists already
# https://github.com/evilsocket/pwnagotchi/issues/483
if dest_filename.startswith('/etc/network/interfaces.d/') and os.path.exists(dest_filename):
if (dest_filename.startswith('/etc/network/interfaces.d/') or dest_filename.startswith('/root/')
and os.path.exists(dest_filename)):
print("%s exists, skipping ..." % dest_filename)
return