Files
sshyp/lib/stweak.py
T
RandyTheSilly d031e3c7ab Use tempfile.NamedTemporaryFile() for editing notes, removing the need for shm_gen()
Former-commit-id: 2d2e733945ba2748a24ed93f861145cf26ac1f13
Former-commit-id: e3b5add08675438e1a08607d081a9a44f426425e
2023-06-18 14:10:25 -04:00

501 lines
21 KiB
Python

from configparser import ConfigParser
from curses import A_REVERSE, echo, KEY_DOWN, KEY_UP, cbreak, curs_set, endwin, initscr, newwin, noecho, nocbreak
from curses.textpad import rectangle, Textbox
from os import environ, listdir, remove
from os.path import exists, expanduser, isfile
from pathlib import Path
from random import randint
from shutil import get_terminal_size, which
from sshyp import copy_id_check, string_gen
from subprocess import PIPE, run
# PORT START UNAME-IMPORT-STWEAK
from os import uname
# PORT END UNAME-IMPORT-STWEAK
home, stdscr, sshyp_data = expanduser("~"), initscr(), ConfigParser()
if isfile(f"{home}/.config/sshyp/sshyp.ini"):
_exists_flag = True
sshyp_data.read(f"{home}/.config/sshyp/sshyp.ini")
else:
_exists_flag = False
# writes data stored in ConfigParser to the correct config file
def write_config(_sshyp_data=sshyp_data):
with open(f"{home}/.config/sshyp/sshyp.ini", 'w') as configfile:
_sshyp_data.write(configfile)
# creates a radio selection between the provided options
def curses_radio(_options, _pretext):
curs_set(0)
_selected = 0
while True:
stdscr.clear()
stdscr.addstr(0, 0, _pretext)
for _i, _option in enumerate(_options):
_y = _i + 2
if _i == _selected:
stdscr.addstr(_y, 0, "[*] " + _option, A_REVERSE)
else:
stdscr.addstr(_y, 0, "[ ] " + _option)
stdscr.refresh()
_key = stdscr.getch()
# update _selected based on user input
if _key == KEY_UP:
_selected = (_selected - 1) % len(_options)
elif _key == KEY_DOWN:
_selected = (_selected + 1) % len(_options)
elif _key == ord('\n'):
break
stdscr.refresh()
curs_set(1)
return _selected
# creates a text-box input
def curses_text(_pretext):
stdscr.clear()
stdscr.addstr(0, 0, _pretext)
_term_columns = get_terminal_size()[0]
_editwin = newwin(1, _term_columns - 2, 3, 1)
rectangle(stdscr, 2, 0, 4, _term_columns - 1)
stdscr.refresh()
_box = Textbox(_editwin)
# let the user edit until ctrl+g/enter is struck
_box.edit()
# return resulting contents
return _box.gather().strip()
# cleanly exit curses and optionally prints an exit message
def curses_terminate(_term_message):
nocbreak()
echo()
endwin()
if _term_message:
print(_term_message)
# device+sync type selection
def install_type():
_offline_mode = 'false'
# PORT START TWEAK-DEVTYPE
_install_type = curses_radio(('client (ssh-synchronized)', 'client (offline)', 'server'),
'device + sync type configuration')
# PORT END TWEAK-DEVTYPE
if _install_type == 2:
_dev_type = 'server'
Path(f"{home}/.config/sshyp/deleted").mkdir(mode=0o700, exist_ok=True)
Path(f"{home}/.config/sshyp/whitelist").mkdir(mode=0o700, exist_ok=True)
curses_terminate('\nmake sure the ssh service is running and properly configured\n')
else:
_dev_type = 'client'
if _install_type == 1:
_offline_mode = 'true'
if not sshyp_data.has_section('CLIENT-GENERAL'):
sshyp_data.add_section('CLIENT-GENERAL')
sshyp_data.set('CLIENT-GENERAL', 'offline_mode_enabled', _offline_mode)
if not sshyp_data.has_section('GENERAL'):
sshyp_data.add_section('GENERAL')
sshyp_data.set('GENERAL', 'device_type', _dev_type)
write_config()
return _dev_type, _offline_mode
# gpg configuration
def gpg_config():
# gpg key selection
_uid_list = [_item for _item in run(('gpg', '-k', '--with-colons'),
stdout=PIPE, text=True).stdout.splitlines() if _item.startswith('uid')]
_named_uid_list = []
for _uid in _uid_list:
_named_uid_list.append(_uid.split(':')[9])
_named_uid_list.append('auto-generate')
_gpg_id_sel = curses_radio(_named_uid_list, 'gpg key selection')
if _gpg_id_sel == len(_named_uid_list) - 1:
if not isfile(f"{home}/.config/sshyp/gpg-gen"):
open(f"{home}/.config/sshyp/gpg-gen", 'w').writelines([
'Key-Type: 1\n', 'Key-Length: 4096\n', 'Key-Usage: sign encrypt\n', 'Name-Real: sshyp\n',
'Name-Comment: gpg-sshyp\n', 'Name-Email: github.com/rwinkhart/sshyp\n',
'Expire-Date: 0'])
run(('gpg', '-q', '--batch', '--generate-key', f"{home}/.config/sshyp/gpg-gen"))
remove(f"{home}/.config/sshyp/gpg-gen")
_gpg_id = run(('gpg', '-k', '--with-colons'), stdout=PIPE, text=True).stdout.splitlines()[-1].split(':')[9]
else:
_gpg_id = _named_uid_list[_gpg_id_sel]
# lock file generation
if isfile(f"{home}/.config/sshyp/lock.gpg"):
remove(f"{home}/.config/sshyp/lock.gpg")
open(f"{home}/.config/sshyp/lock", 'w')
run(('gpg', '-qr', _gpg_id, '-e', f"{home}/.config/sshyp/lock"))
remove(f"{home}/.config/sshyp/lock")
if not sshyp_data.has_section('CLIENT-GENERAL'):
sshyp_data.add_section('CLIENT-GENERAL')
sshyp_data.set('CLIENT-GENERAL', 'gpg_id', _gpg_id)
write_config()
# text editor configuration
def editor_config(_env_mode):
if _env_mode:
# set default text editor to value of EDITOR environment variable, otherwise default to nano
if 'EDITOR' in environ:
_editor = environ['EDITOR']
else:
_editor = 'nano'
else:
_editor = curses_text('enter the name of your preferred text editor:\n\n\n\n\n'
'(ctrl+g/enter to confirm)\n\nexample input: vim')
if not sshyp_data.has_section('CLIENT-GENERAL'):
sshyp_data.add_section('CLIENT-GENERAL')
sshyp_data.set('CLIENT-GENERAL', 'text_editor', _editor)
write_config()
# ssh+sshync configuration
def ssh_config():
# private key selection/generation
_keys = []
for _file in listdir(f"{home}/.ssh"):
if not _file.startswith('.') and _file not in ('known_hosts', 'authorized_keys') \
and not _file.endswith('.pub') and isfile(f"{home}/.ssh/{_file}"):
_keys.append(f"{home}/.ssh/{_file}")
_keys.extend(['auto-generate', 'other (type the location)'])
_key_selected_num = curses_radio(_keys, 'which private ssh key would you like to use for sshyp?')
_gen_index = len(_keys)-2
if _key_selected_num >= _gen_index:
_ssh_key = expanduser(curses_text('enter the location for your private ssh key:\n\n\n\n\n(ctrl+g/enter to '
'confirm)\n\nexample input:\n\n~/.ssh/privkey'))
if _key_selected_num == _gen_index:
_passphrase = curses_text('enter your desired ssh keyfile passphrase:\n\n\n\n\n(ctrl+g/enter to confirm)'
'\n\ntip: you can leave this blank to use the keyfile without a passphrase')
run(('ssh-keygen', '-q', '-t', 'ed25519', '-N', _passphrase, '-f', _ssh_key))
else:
_ssh_key = _keys[_key_selected_num]
# ssh+sshync configuration
_uiport = curses_text('enter the username, ip, and ssh port of your sshyp server:\n\n\n\n\n(ctrl+g/enter to '
'confirm)\n\nexample inputs:\n\n ipv4: user@10.10.10.10:22\n ipv6: user@[2000:2000:2000:2000'
':2000:2000:2000:2000]:22\n domain: user@mydomain.com:22')
_uiport_split = _uiport.split('@')
_username_ssh = _uiport_split[0]
_iport = _uiport_split[1].lstrip('[').replace(']', '').rsplit(':', 1)
if not sshyp_data.has_section('SSHYNC'):
sshyp_data.add_section('SSHYNC')
sshyp_data.set('SSHYNC', 'user', _username_ssh)
sshyp_data.set('SSHYNC', 'ip', _iport[0])
sshyp_data.set('SSHYNC', 'port', _iport[1])
sshyp_data.set('SSHYNC', 'local_dir', f"{home}/.local/share/sshyp/")
sshyp_data.set('SSHYNC', 'remote_dir', f"/home/{_username_ssh}/.local/share/sshyp/")
sshyp_data.set('SSHYNC', 'identity_file', _ssh_key)
write_config()
return _iport[1], _username_ssh, _iport[0]
# device id configuration
def dev_id_config(_ip, _username_ssh, _port):
_device_id_prefix = curses_text('name this device:\n\n\n\n\n(ctrl+g/enter to confirm)\n\nimportant: this '
'id must be unique amongst your client devices\n\nthis is used to keep track of '
'database syncing and quick-unlock permissions\n')
_device_id_suffix = string_gen('f', randint(24, 48))
_device_id = _device_id_prefix + '-' + _device_id_suffix
# remove existing device ids
for _id in listdir(f"{home}/.config/sshyp/devices"):
remove(f"{home}/.config/sshyp/devices/{_id}")
open(f"{home}/.config/sshyp/devices/{_device_id}", 'w')
# test server connection and attempt to register device id
copy_id_check(_ip, _username_ssh, _port, _device_id, sshyp_data)
# quick-unlock configuration
def quick_unlock_config(_default):
if _default:
_enabled = 'false'
else:
_quick_unlock_sel = curses_radio(('yes', 'no'), 'enable quick-unlock?\n\n\n\n\nquick-unlock allows you to use '
'a shorter version of your gpg key passphrase and\nrequires a '
'constant connection to your sshyp server to authenticate\n\n'
'WARNING: quick-unlock is only as secure as the environment'
' you use it in\n\na compromised program on your computer could'
' scan the process list in the\nbrief period during decryption '
'to retrieve the necessary information to decrypt your entries')
if _quick_unlock_sel == 0:
_enabled = 'true'
else:
_enabled = 'false'
if not sshyp_data.has_section('CLIENT-ONLINE'):
sshyp_data.add_section('CLIENT-ONLINE')
sshyp_data.set('CLIENT-ONLINE', 'quick_unlock_enabled', _enabled)
write_config()
return _enabled
# re-encrypt/optimize all entries
def refresh_encryption():
_directory = sshyp_data.get('SSHYNC', 'local_dir').rstrip('/')
# warn the user of potential data loss and prompt to continue
_proceed = curses_radio(('yes', 'no'), "are you sure you wish to re-encrypt all entries with this key?"
"\n\n\n\n\nWARNING: proceeding with this action will remove/overwrite"
" any directories matching the following:"
f"\n\n{home}/.local/share/sshyp.old\n{home}/.local/share/sshyp.new\n\n")
if _proceed != 0:
return 3
# set new gpg key
gpg_config()
from os import walk
from os.path import isdir
from shutil import move, rmtree
from sshyp import decrypt, encrypt, optimized_edit
# remove existing conflicts
for _extension in ('.new', '.old'):
if exists(f"{_directory}{_extension}"):
rmtree(f"{_directory}{_extension}")
# decrypt, optimize, and re-encrypt each entry with the newly selected key
if isdir(_directory):
curses_terminate('\noptimizing and re-encrypting entries... please wait - do not terminate this process')
for _root, _dirs, _files in sorted(walk(_directory, topdown=True)):
for _filename in _files:
Path(_root.replace(_directory, _directory + '.new', 1)).mkdir(0o700, parents=True, exist_ok=True)
_new_lines = optimized_edit(decrypt(f"{_root}/{_filename[:-4]}"), None, -1)
encrypt(_new_lines, f"{_root.replace(_directory, _directory + '.new', 1)}/{_filename[:-4]}",
sshyp_data.get('CLIENT-GENERAL', 'gpg_id'))
# create a backup of the original version and activate the new version
move(_directory, _directory + '.old')
move(_directory + '.new', _directory)
return 1
else:
return 2
# PORT START WHITELIST-SERVER
# takes input from the user to set up quick-unlock pin
def whitelist_setup():
_gpg_password_temp = str(curses_text('full gpg passphrase:\n\n\n\n\n(ctrl+g/enter to confirm)'))
_half_length = int(len(_gpg_password_temp)/2)
try:
_short_password_length = int(curses_text(f"quick unlock pin length ({_half_length}):\n\n\n\n\n(ctrl+g/enter "
"to confirm)\n\npin must be half the length of the gpg passphrase "
"or less\n\ncannot be a negative number"))
if not 0 <= _short_password_length <= _half_length:
_short_password_length = _half_length
except ValueError:
_short_password_length = _half_length
_i, _quick_unlock_password, _quick_unlock_password_excluded = 0, '', ''
for _char in _gpg_password_temp:
if _i % 2 == 1 and _i < _short_password_length*2:
_quick_unlock_password += _char
else:
_quick_unlock_password_excluded += _char
_i += 1
# create assembly key
open(f"{home}/.config/sshyp/gpg-gen", 'w').writelines([
'Key-Type: 1\n', 'Key-Length: 4096\n', 'Key-Usage: sign encrypt\n', 'Name-Real: sshyp\n',
'Name-Comment: gpg-sshyp-whitelist\n', 'Name-Email: github.com/rwinkhart/sshyp\n', 'Expire-Date: 0'])
run(('gpg', '-q', '--pinentry-mode', 'loopback', '--batch', '--generate-key', '--passphrase',
_quick_unlock_password, f"{home}/.config/sshyp/gpg-gen"))
remove(f"{home}/.config/sshyp/gpg-gen")
_gpg_id = run(('gpg', '-k', '--with-colons'), stdout=PIPE, text=True).stdout.splitlines()[-1].split(':')[9]
# encrypt excluded with the assembly key
from sshyp import encrypt
encrypt(_quick_unlock_password_excluded, f"{home}/.config/sshyp/excluded", _gpg_id)
curses_radio(['okay, I have it memorized'], f"your quick-unlock pin: {_quick_unlock_password}")
# adds or removes quick-unlock whitelisted device ids
def whitelist_manage(_action):
_whitelisted_ids = listdir(f"{home}/.config/sshyp/whitelist")
_device_ids = listdir(f"{home}/.config/sshyp/devices")
# a value of True indicates adding
if _action:
_unwhitelisted_ids = []
for _id in _device_ids:
if _id not in _whitelisted_ids:
_unwhitelisted_ids.append(_id)
_unwhitelisted_ids.append('cancel')
_add_id = curses_radio(_unwhitelisted_ids, 'id to add to whitelist:')
if _add_id == len(_unwhitelisted_ids)-1:
return
open(f"{home}/.config/sshyp/whitelist/{_unwhitelisted_ids[_add_id]}", 'w').write('')
else:
_whitelisted_choices = _whitelisted_ids + ['cancel']
_del_id = curses_radio(_whitelisted_choices, 'id to remove from whitelist:')
if _del_id == len(_whitelisted_choices)-1:
return
remove(f"{home}/.config/sshyp/whitelist/{_whitelisted_ids[_del_id]}")
# prune deleted device ids from whitelist
for _id in _whitelisted_ids:
if _id not in _device_ids:
remove(f"{home}/.config/sshyp/whitelist/{_id}")
# runs quick-unlock configuration menu
def whitelist_menu():
while True:
_choice = curses_radio(('setup/create pin', 'add to whitelist', 'remove from whitelist',
'exit/done'), 'quick-unlock/whitelist management')
if _choice == 0:
whitelist_setup()
elif _choice == 1:
whitelist_manage(True)
elif _choice == 2:
whitelist_manage(False)
else:
break
# PORT END WHITELIST-SERVER
# runs secondary configuration menu
def global_menu(_device_type, _top_message):
while True:
# curses initialization
noecho()
cbreak()
stdscr.keypad(True)
_options, _choice, _term_message, _exit_signal = ['change device/synchronization types'], 0, False, False
try:
if _device_type == 'client':
_options.extend(['change gpg key', 're-configure ssh(ync)', 'change device name',
'[OPTIONAL, RECOMMENDED] set custom text editor',
'[OPTIONAL] enable/disable quick-unlock',
'[OPTIONAL] re-encrypt/optimize entries',
'[OPTIONAL, NOT IMPLEMENTED] extensions and updates'])
else:
_options.extend(['manage quick-unlock/whitelist'])
_options.extend(['exit/done'])
_choice += curses_radio(_options, _top_message)
if _choice == 0:
_dev_sync_types = install_type()
# if not running in server or offline mode and a sshync config has not been made
if _dev_sync_types[0] != 'server' and _dev_sync_types[1] != 'true' \
and not sshyp_data.has_section('SSHYNC'):
_ip, _username_ssh, _port = ssh_config()
# if no device id is set
if not listdir(f"{home}/.config/sshyp/devices"):
dev_id_config(_ip, _username_ssh, _port)
_device_type = _dev_sync_types[0]
elif _choice == 1:
if _device_type == 'client':
gpg_config()
else:
whitelist_menu()
elif _choice == 2:
if _device_type == 'client':
ssh_config()
else:
_exit_signal = True
elif _choice == 3:
if not sshyp_data.has_section('SSHYNC'):
ssh_config()
dev_id_config(sshyp_data.get('SSHYNC', 'ip'), sshyp_data.get('SSHYNC', 'user'),
sshyp_data.get('SSHYNC', 'port'))
elif _choice == 4:
editor_config(False)
elif _choice == 5:
_enabled = quick_unlock_config(False)
if _enabled == 'true':
_term_message = ('\nquick-unlock has been enabled client-side - in order for this feature to '
'function,\nyou must first log in to the sshyp server and run:\n\nsshyp tweak\n\n'
'from there you can create a quick-unlock pin and add this device to the '
'whitelist\n')
elif _choice == 6:
_success = refresh_encryption()
if _success == 1:
_term_message = "\na backup of your previous entry directory has been created:\n\n" \
f"{home}/.local/share/sshyp.old\n"
elif _success == 2:
_term_message = '\n\u001b[38;5;9merror: re-encryption failed: ' \
'entry directory not found\u001b[0m\n'
elif _choice > 6: # TODO update once displayed options are all implemented
_exit_signal = True
curses_terminate(_term_message)
except KeyboardInterrupt:
curses_terminate(False)
_exit_signal = True
if _exit_signal:
break
# runs initial configuration wizard
def initial_setup():
# required directory creation
Path(f"{home}/.config/sshyp/devices").mkdir(mode=0o700, parents=True, exist_ok=True)
Path(f"{home}/.local/share/sshyp").mkdir(mode=0o700, parents=True, exist_ok=True)
# removal of old config files
if _exists_flag:
sshyp_data.clear()
# curses initialization
noecho()
cbreak()
stdscr.keypad(True)
# curses menu tree
try:
# device+sync type selection
_dev_sync_types = install_type()
if _dev_sync_types[0] == 'client':
# gpg configuration
gpg_config()
# text editor configuration (automated)
editor_config(True)
# quick-unlock configuration (disabled by default)
quick_unlock_config(True)
# online (synchronized mode) configuration
if _dev_sync_types[1] != 'true':
# ssh+sshync configuration
_ip, _username_ssh, _port = ssh_config()
# device id configuration
dev_id_config(_ip, _username_ssh, _port)
# cleanly exit curses
curses_terminate(False)
else:
# cleanly exit curses
curses_terminate(False)
# PORT START CLIPTOOL
# check for clipboard tool and display warning if missing
if uname()[0] in ('Linux', 'FreeBSD') and 'WSL_DISTRO_NAME' not in environ \
and not exists("/data/data/com.termux"):
_display_server, _clipboard_tool, _clipboard_package = None, None, None
if 'WAYLAND_DISPLAY' in environ:
_display_server, _clipboard_tool, _clipboard_package = 'Wayland', 'wl-copy', 'wl-clipboard'
elif 'DISPLAY' in environ:
_display_server, _clipboard_tool, _clipboard_package = 'X11', 'xclip', 'xclip'
if _display_server is not None and which(_clipboard_tool) is None:
print(f'\n\u001b[38;5;9mwarning: you are using {_display_server} and "{_clipboard_tool}" is not '
f'present - \ncopying entry fields will not function until '
f'"{_clipboard_package}" is installed\u001b[0m')
# PORT END CLIPTOOL
global_menu(_dev_sync_types[0], 'additional configuration options:')
except KeyboardInterrupt:
# cleanly exit curses
curses_terminate(False)