#!/usr/bin/env python3 from configparser import ConfigParser, NoSectionError from os import chmod, environ, listdir, remove, walk from os.path import exists, expanduser, isdir, isfile, realpath from pathlib import Path from random import randint from shutil import move, rmtree from sshync import delete as offline_delete, run_profile from subprocess import CalledProcessError, DEVNULL, PIPE, run from sys import argv, exit as s_exit # PORT START UNAME-IMPORT-SSHYP from os import uname # PORT END UNAME-IMPORT-SSHYP home = expanduser("~") # UTILITY FUNCTIONS # generates and prints full entry list def entry_list_gen(_directory=f"{home}/.local/share/sshyp/"): from shutil import get_terminal_size from textwrap import fill _ran = False print("\nfor a list of usable commands, run 'sshyp help'\n\n\u001b[38;5;0;48;5;15msshyp entries:\u001b[0m\n") for _root, _dirs, _files in sorted(walk(_directory, topdown=True)): _entry_list, _color_alternator = [], 1 if _ran: print(f"\u001b[38;5;15;48;5;238m{_root.replace(f'{home}/.local/share/sshyp', '', 1)}/\u001b[0m") for filename in sorted(_files): if _color_alternator > 0: _entry_list.append(filename[:-4]) else: _entry_list.append(f"\u001b[38;5;8m{filename[:-4]}\u001b[0m") _color_alternator = _color_alternator * -1 _real = len(' '.join(_entry_list)) - (5.5 * len(_entry_list)) if _real <= get_terminal_size()[0]: _width = len(' '.join(_entry_list)) else: _width = (len(' '.join(_entry_list)) / (_real / get_terminal_size()[0]) - 25) if len(_entry_list) > 0: print(fill(' '.join(_entry_list), width=_width) + '\n') elif _ran: print('\u001b[38;5;9m-empty directory-\u001b[0m\n') _ran = True # displays the contents of an entry in a readable format def entry_reader(_decrypted_entry): _entry_lines, _notes_flag = [_line.rstrip() for _line in open(_decrypted_entry, 'r').readlines()], 0 if pass_show: _entry_password = f'\u001b[38;5;10m{_entry_lines[0]}\u001b[0m' else: _entry_password = f'\u001b[38;5;3mend command in "--show" or "-s" to view\u001b[0m' print() for _num in range(len(_entry_lines)): try: if _num == 0 and _entry_lines[1] != '': print(f"\u001b[38;5;15;48;5;238musername:\u001b[0m\n{_entry_lines[1]}\n") elif _num == 1 and _entry_lines[0] != '': print(f"\u001b[38;5;15;48;5;238mpassword:\u001b[0m\n{_entry_password}\n") elif _num == 2 and _entry_lines[2] != '': print(f"\u001b[38;5;15;48;5;238murl:\u001b[0m\n{_entry_lines[_num]}\n") elif _num >= 3 and _entry_lines[_num] != '' and _notes_flag != 1: _notes_flag = 1 print('\u001b[38;5;15;48;5;238mnotes:\u001b[0m\n' + _entry_lines[_num]) elif _num >= 3 and _notes_flag == 1: print(_entry_lines[_num]) if _notes_flag == 1: try: _line_test = _entry_lines[_num + 1] except IndexError: print() except IndexError: if _num == 0: print(f"\u001b[38;5;15;48;5;238mpassword:\u001b[0m\n{_entry_password}\n") # generates and returns a random string based on input def string_gen(_complexity, _length): from random import SystemRandom import string if _complexity == 's': _character_pool = string.ascii_letters + string.digits elif _complexity == 'f': _character_pool = string.digits + string.ascii_letters + string.punctuation.replace('/', '').replace('\\', '')\ .replace("'", '').replace('"', '').replace('`', '').replace('~', '') else: _character_pool = string.digits + string.ascii_letters + string.punctuation _min_special, _special = round(.2 * _length), 0 while True: _gen = ''.join(SystemRandom().choice(_character_pool) for _ in range(_length)) for _character in _gen: if not _character.isalpha(): _special += 1 if _special >= _min_special: break return _gen # prompts the user for necessary information to generate a password and passes it to string_gen def pass_gen(): _length = 9 while True: try: _length = int(input('password length: ')) except ValueError: continue else: if _length < 1: continue else: break _complexity = str(input('password complexity - simple (for compatibility) or complex (for security)? (s/C) ')) if _complexity not in ('s', 'S'): _complexity = 'c' _gen = string_gen(_complexity.lower(), _length) return _gen # creates a temporary directory for entry editing def shm_gen(_tmp_dir=f"{home}/.config/sshyp/tmp/"): _shm_folder_gen = string_gen('f', randint(12, 48)) _shm_entry_gen = string_gen('f', randint(12, 48)) Path(_tmp_dir + _shm_folder_gen).mkdir(mode=0o700) return _shm_folder_gen, _shm_entry_gen # encrypts an entry and cleans up the temporary files def encrypt(_entry_dir, _shm_folder, _shm_entry, _gpg_id, _tmp_dir=f"{home}/.config/sshyp/tmp/"): run(('gpg', '-qr', str(_gpg_id), '-e', f"{_tmp_dir}{_shm_folder}/{_shm_entry}")) move(f"{_tmp_dir}{_shm_folder}/{_shm_entry}.gpg", f"{_entry_dir}.gpg") rmtree(f"{_tmp_dir}{_shm_folder}") # decrypts an entry to a temporary directory def decrypt(_entry_dir, _shm_folder, _shm_entry, _quick_pass, _tmp_dir=f"{home}/.config/sshyp/tmp/"): if not isinstance(_quick_pass, bool): _unlock_method = ['gpg', '--pinentry-mode', 'loopback', '--passphrase', _quick_pass, '-qd', '--output'] else: _unlock_method = ['gpg', '-qd', '--output'] if _shm_folder is None and _shm_entry is None: _output_target = ['/dev/null', f"{home}/.config/sshyp/lock.gpg"] else: _output_target = [f"{_tmp_dir}{_shm_folder}/{_shm_entry}", f"{_entry_dir}.gpg"] try: run(_unlock_method + _output_target, stderr=DEVNULL, check=True) except CalledProcessError: if not isinstance(_quick_pass, bool): print('\n\u001b[38;5;9merror: quick-unlock failed as a result of an incorrect passphrase, an unreachable ' 'sshyp server, or an invalid configuration\n\nfalling back to standard unlock\u001b[0m\n') try: run(('gpg', '-qd', '--output') + _output_target, stderr=DEVNULL, check=True) except CalledProcessError: print('\n\u001b[38;5;9merror: could not decrypt - ensure the correct gpg key is present\u001b[0m\n') s_exit(4) else: print('\n\u001b[38;5;9merror: could not decrypt - ensure the correct gpg key is present\u001b[0m\n') s_exit(4) # returns True if expected and reality align, otherwise error def target_exists_check(_target_name, _expected_presence): if isfile(f"{directory}{_target_name}.gpg") or isdir(f"{directory}{_target_name}"): if _expected_presence: return True else: print(f"\n\u001b[38;5;9merror: (/{_target_name}) already exists\u001b[0m\n") s_exit(3) else: if not _expected_presence: return True else: print(f"\n\u001b[38;5;9merror: (/{_target_name}) does not exist\u001b[0m\n") s_exit(2) # returns target type (entry == True, folder == False, null == None), optional errors def target_type_check(_target_name, _expected_type=True, _error=False): if isfile(f"{directory}{_target_name}.gpg"): if _error and not _expected_type: print(f"\n\u001b[38;5;9merror: (/{_target_name}) is an entry\u001b[0m\n") s_exit(2) return True elif isdir(f"{directory}{_target_name}"): if _error and _expected_type: print(f"\n\u001b[38;5;9merror: (/{_target_name}) is a folder\u001b[0m\n") s_exit(2) return False else: print(f"\n\u001b[38;5;9merror: (/{_target_name}) does not exist\u001b[0m\n") s_exit(2) # call decrypt() based on quick-unlock status def determine_decrypt(_entry_dir, _shm_folder, _shm_entry): if quick_unlock_enabled == 'true': decrypt(_entry_dir, _shm_folder, _shm_entry, whitelist_verify(port, username_ssh, ip, client_device_id)) else: decrypt(_entry_dir, _shm_folder, _shm_entry, False) # ensures an edited entry is optimized for best compatibility def optimized_edit(_lines, _edit_data, _edit_line): while len(_lines) < _edit_line + 1: _lines.append('\n') if _edit_data is not None: _lines[_edit_line] = _edit_data.strip('\n').rstrip() + '\n' for _num in range(len(_lines)): if not _lines[_num].endswith('\n'): _lines[_num] += '\n' for _num in reversed(range(len(_lines))): if _lines[_num] == '\n': _lines = _lines[:-1] elif _lines[_num].endswith('\n'): _lines[_num] = _lines[_num].rstrip() break else: break return _lines # edits the note attached to an entry def edit_note(_shm_folder, _shm_entry, _lines): _reg_lines = _lines[0:3] open(f"{tmp_dir}{_shm_folder}/{_shm_entry}-n", 'w').writelines(_lines[3:]) run((editor, f"{tmp_dir}{_shm_folder}/{_shm_entry}-n")) _new_notes = open(f"{tmp_dir}{_shm_folder}/{_shm_entry}-n").readlines() while len(_reg_lines) < 3: _reg_lines.append('\n') _noted_lines = _reg_lines + _new_notes return _noted_lines # attempts to connect to the user's server via ssh to register the device for syncing def copy_id_check(_port, _username_ssh, _ip, _client_device_id, _sshyp_data): from stweak import write_config if not _sshyp_data.has_section('CLIENT-ONLINE'): _sshyp_data.add_section('CLIENT-ONLINE') try: run(('ssh', '-o', 'ConnectTimeout=3', '-i', f"{home}/.ssh/sshyp", '-p', _port, f"{_username_ssh}@{_ip}", f'python3 -c \'from pathlib import Path; Path("/home/{_username_ssh}/.config/sshyp/devices/' f'{_client_device_id}").touch(mode=0o400, exist_ok=True)\''), stderr=DEVNULL, check=True) except CalledProcessError: print('\n\u001b[38;5;9mwarning: ssh connection could not be made - ensure the public key (~/.ssh/sshyp.pub) is ' 'registered on the remote server and that the entered ip, port, and username are correct\n\nsyncing ' 'functionality will be disabled until this is addressed\u001b[0m\n') _sshyp_data.set('CLIENT-ONLINE', 'ssh_error', '1') write_config(_sshyp_data) return True _sshyp_data.set('CLIENT-ONLINE', 'ssh_error', '0') write_config(_sshyp_data) return False # ARGUMENT-SPECIFIC FUNCTIONS # prints help text based on argument def print_info(): if arguments[0] in ('version', '-v'): _blank = '\u001b[38;5;7;48;5;8m/\u001b[38;5;15;48;5;15m' + 54*' ' + '\u001b[38;5;7;48;5;8m/\u001b[0m' _border = '\u001b[38;5;7;48;5;8m' + 28*'<>' + '\u001b[0m\n' print(f"""\nsshyp is a simple, self-hosted, sftp-synchronized\npassword manager for unix(-like) systems\n {16*' '}..{7*' '}\u001b[38;5;9m♥♥ ♥♥\u001b[0m{7*' '}.. {9*' '}.''.''/()\\{5*' '}\u001b[38;5;13m♥♥♥♥♥♥♥\u001b[0m{5*' '}/()\\''.''. {8*' '}*{7*' '}:{8*' '}\u001b[38;5;9m♥♥♥♥♥\u001b[0m{8*' '}:{7*' '}* {9*' '}`..'..'{10*' '}\u001b[38;5;13m♥♥♥\u001b[0m{10*' '}`..'..' {9*' '}//{3*' '}\\\\{11*' '}\u001b[38;5;9m♥\u001b[0m{11*' '}//{3*' '}\\\\""") print(f"{_border}{_blank}\n\u001b[38;5;7;48;5;8m/\u001b[38;5;15;48;5;15m{3*' '}\u001b[38;5;15;48;5;8m" f"sshyp ", f"copyright (c) 2021-2023 ", f"randall winkhart\u001b[38;5;15;48;5;15m{3*' '}" f"\u001b[38;5;7;48;5;8m/\u001b[0m\n{_blank}") print(f"\u001b[38;5;7;48;5;8m/\u001b[38;5;15;48;5;15m{20*' '}\u001b[38;5;15;48;5;8mversion 1.4.99" f"\u001b[38;5;15;48;5;15m{20*' '}\u001b[38;5;7;48;5;8m/\u001b[0m") print(f"\u001b[38;5;7;48;5;8m/\u001b[38;5;15;48;5;15m{9*' '}\u001b[38;5;15;48;5;8mthe argumentative " f"agronomist update\u001b[38;5;15;48;5;15m{10*' '}\u001b[38;5;7;48;5;8m/\u001b[0m") print(f"{_blank}\n{_border}\nsee https://github.com/rwinkhart/sshyp for more information\n") elif arguments[0] == 'license': print('\nThis program is free software: you can redistribute it and/or modify it under the terms\nof version 3 ' '(only) of the GNU General Public License as published by the Free Software Foundation.\n\nThis program ' 'is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY;\nwithout even the implied ' 'warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.\nSee the GNU General Public License for' ' more details.\n\nhttps://opensource.org/licenses/GPL-3.0\n') elif arguments[0] == 'add' and device_type == 'client': print(f"""\n\u001b[1musage:\u001b[0m sshyp add