#!/usr/bin/env python3 from configparser import ConfigParser, NoOptionError, NoSectionError from os import chmod, environ, listdir, walk from os.path import expanduser, isdir, isfile, realpath from pathlib import Path from shutil import move 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 _ran, _width = False, get_terminal_size().columns print("\nfor a list of usable commands, run 'sshyp help'\n\n\u001b[38;5;0;48;5;15msshyp entries:\u001b[0m", end='') for _root, _dirs, _files in sorted(walk(_directory, topdown=True)): _color_alternator = 1 if _ran: print(f"\n\n\u001b[38;5;7;48;5;8m{_root.replace(f'{home}/.local/share/sshyp', '', 1)}/\u001b[0m") _char_counter = 0 for _filename in sorted(_files): if _color_alternator > 0: _print_string = _filename[:-4] else: _print_string = f"\u001b[38;5;8m{_filename[:-4]}\u001b[0m" # -3 instead of -4 to account for trailing space character _char_counter += len(_filename)-3 if _char_counter >= _width: # reset _char_counter to length of first entry in new line _char_counter = len(_filename)-3 print() print(_print_string + ' ', end='') _color_alternator = _color_alternator * -1 if _ran and _char_counter < 1: print('\u001b[38;5;9m-empty directory-\u001b[0m', end='') _ran = True print('\n') # displays the contents of an entry in a readable format def entry_reader(_decrypted_entry): _notes_flag = 0 if pass_show: _entry_password = f'\u001b[38;5;10m{_decrypted_entry[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(_decrypted_entry)): try: if _num == 0 and _decrypted_entry[1] != '': print(f"\u001b[38;5;7;48;5;8musername:\u001b[0m\n{_decrypted_entry[1]}\n") elif _num == 1 and _decrypted_entry[0] != '': print(f"\u001b[38;5;7;48;5;8mpassword:\u001b[0m\n{_entry_password}\n") elif _num == 2 and _decrypted_entry[2] != '': print(f"\u001b[38;5;7;48;5;8murl:\u001b[0m\n{_decrypted_entry[_num]}\n") elif _num >= 3 and _decrypted_entry[_num] != '' and _notes_flag != 1: _notes_flag = 1 print('\u001b[38;5;7;48;5;8mnotes:\u001b[0m\n' + _decrypted_entry[_num]) elif _num >= 3 and _notes_flag == 1: print(_decrypted_entry[_num]) if _notes_flag == 1: try: _line_test = _decrypted_entry[_num + 1] except IndexError: print() except IndexError: if _num == 0 and _decrypted_entry[0] != '': print(f"\u001b[38;5;7;48;5;8mpassword:\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('\npassword length: ')) except ValueError: continue else: if _length < 1: continue else: break _complexity = str(input('\npassword 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 file to allow notes to be edited by standard editors def edit_note(_note_lines, _exit_on_match=False): from tempfile import NamedTemporaryFile _joined_note_lines = '\n'.join(_note_lines) with NamedTemporaryFile(mode='w+') as _tmp: _tmp.write(_joined_note_lines) _tmp.seek(0) try: run((editor, _tmp.name)) except FileNotFoundError: print(f"\n\u001b[38;5;9merror: the configured text editor ({editor}) cannot be found on this system\n\n" f"please either install the editor or re-configure the active editor using 'sshyp tweak'\u001b[0m\n") _tmp.seek(0) _new_note = _tmp.read().rstrip() if _exit_on_match and _joined_note_lines == _new_note: s_exit(0) return _new_note # encrypts an entry and cleans up the temporary files def encrypt(_entry_data, _entry_dir, _gpg_id): _bytes_data = '\n'.join(_entry_data).rstrip().encode() _encrypted_data = run(('gpg', '-qr', str(_gpg_id), '-e'), input=_bytes_data, stdout=PIPE).stdout open(_entry_dir + '.gpg', 'wb').write(_encrypted_data) # decrypts an entry to a temporary directory def decrypt(_entry_dir, _quick_verify=None, _quick_pass=None): _contents = None # check quick-unlock status, fetch passphrase if _quick_verify: _quick_pass = whitelist_verify(port, username_ssh, ip, client_device_id, identity) else: if _quick_pass is None: _quick_pass = False # set decryption method based on quick-unlock availability if not isinstance(_quick_pass, bool): _cmd = ['gpg', '--pinentry-mode', 'loopback', '--passphrase', _quick_pass, '-qd'] else: _cmd = ['gpg', '-qd'] # set decryption target based on lock file availability if _entry_dir is None: _dec_target = [f"{home}/.config/sshyp/lock.gpg"] else: _dec_target = [f"{_entry_dir}.gpg"] # run decryption command try: _contents = run(_cmd + _dec_target, stderr=DEVNULL, stdout=PIPE, text=True, check=True).stdout 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: _contents = run(['gpg', '-qd'] + _dec_target, stderr=DEVNULL, stdout=PIPE, text=True, check=True).stdout 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) return _contents.rstrip().split('\n') # checks the user's whitelist status and fetches the full gpg key password if possible def whitelist_verify(_port, _username_ssh, _ip, _client_device_id, _identity): try: run(('gpg', '--pinentry-mode', 'cancel', '-qd', '--output', '/dev/null', f"{home}/.config/sshyp/lock.gpg"), stderr=DEVNULL, check=True) return False except CalledProcessError: _i, _full_password = 0, '' _server_whitelist = run(('ssh', '-i', _identity, '-p', _port, f"{_username_ssh}@{_ip}", f'python3 -c \'from os import listdir; print(listdir("/home/{_username_ssh}' f'/.config/sshyp/whitelist"))\''), stdout=PIPE, text=True).stdout.rstrip()[2:-2].split("', '") for _device_id in _server_whitelist: if _device_id == _client_device_id: from getpass import getpass _quick_unlock_password = getpass(prompt='\nquick-unlock pin: ') _quick_unlock_password_excluded = \ run(('ssh', '-i', _identity, '-p', _port, f"{_username_ssh}@{_ip}", f"gpg --pinentry-mode loopback --passphrase '{_quick_unlock_password}' " f"-qd ~/.config/sshyp/excluded.gpg"), stdout=PIPE, text=True).stdout.rstrip() while _i < len(_quick_unlock_password_excluded): try: _full_password += _quick_unlock_password_excluded[_i] except IndexError: pass try: _full_password += _quick_unlock_password[_i] except IndexError: pass _i += 1 break return _full_password # 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) # ensures an edited entry is optimized for best compatibility def line_edit(_lines, _edit_data, _edit_line): # ensure enough lines are present for edited field while len(_lines) < _edit_line + 1: _lines.append('') # write the edited field _lines[_edit_line] = _edit_data.rstrip() return _lines # attempts to connect to the user's server via ssh to register the device for synchronization def copy_id_check(_port, _username_ssh, _ip, _client_device_id, _identity, _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', _identity, '-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(f'\n\u001b[38;5;9mwarning: ssh connection could not be made - ensure the public key ({_identity}) is ' 'registered on the remote server and that the entered ip, port, and username are correct\n\n' 'synchronization functionality will be disabled until this is addressed\u001b[0m\n') _sshyp_data.set('CLIENT-ONLINE', 'ssh_error', 'true') write_config(_sshyp_data) return True _sshyp_data.set('CLIENT-ONLINE', 'ssh_error', 'false') 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' + 55*' ' + '\u001b[38;5;7;48;5;8m/\u001b[0m' _border = '\u001b[38;5;7;48;5;8m' + 14*'<>' + '-' + 14*'<>' + '\u001b[0m\n' print(f"""\nsshyp is a simple, self-hosted, sftp-synchronized\npassword manager for unix(-like) systems\n {9*' '}..{15*' '}\u001b[38;5;12m♥♥ \u001b[38;5;9m♥♥\u001b[0m{15*' '}.. {8*' '}/()\\''.''.{7*' '}\u001b[38;5;12m♥♥♥\u001b[0m♥♥♥♥\u001b[0m{7*' '}.''.''/()\\{3*' '}_) {5*' '}_.{3*' '}:{7*' '}*{7*' '}\u001b[38;5;9m♥♥♥♥♥\u001b[0m{7*' '}*{7*' '}:{3*' '}<[◎]|_|= }}-}}-*]{4*' '}`..'..'{9*' '}\u001b[0m♥♥♥\u001b[0m{9*' '}`..'..'{6*' '}| {4*' '}◎-◎{4*' '}//{3*' '}\\\\{10*' '}\u001b[38;5;9m♥\u001b[0m{10*' '}//{3*' '}\\\\{5*' '}/|\\""") print(f"{_border}{_blank}\n\u001b[38;5;7;48;5;8m\\\u001b[38;5;15;48;5;15m{18*' '}\u001b[38;5;15;48;5;8msshyp " f"version 1.5.2\u001b[38;5;15;48;5;15m{18*' '}\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{14*' '}\u001b[38;5;15;48;5;8mthe fortified flock" f" update\u001b[38;5;15;48;5;15m{15*' '}\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{9*' '}\u001b[38;5;15;48;5;8mcopyright 2021-2024 ", f"randall winkhart\u001b[38;5;15;48;5;15m{9*' '}\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 of\nversion 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.\n\nSee the GNU General Public License ' 'for more details:\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