#!python """ Group password manager client Copyright (C) 2010-2020 Cr@ns <roots@crans.org> Authors : Daniel Stan <daniel.stan@crans.org> Vincent Le Gallic <legallic@crans.org> Alexandre Iooss <erdnaxe@crans.org> SPDX-License-Identifier: GPL-3.0-or-later """ # Import builtins import sys import subprocess import os import argparse import re import copy import logging from tempfile import NamedTemporaryFile from configparser import ConfigParser from secrets import token_urlsafe # Import thirdparty from gpg.errors import KeyNotFound import pyperclip # Import modules from .crypto import decrypt, encrypt, receive_key, get_key_from_fingerprint, check_key_validity from .locale import _ from .remote import remote_command # Configuration loading # On n'a pas encore accès à la config donc on devine le nom bootstrap_cmd_name = os.path.split(sys.argv[0])[1] default_config_path = os.path.expanduser("~/.config/" + bootstrap_cmd_name) config_path = os.getenv( "CRANSPASSWORDS_CLIENT_CONFIG_DIR", default_config_path) config = ConfigParser() if not config.read(config_path + "/clientconfig.ini"): # If config could not be imported, display an error if required ducktape_display_error = sys.stderr.isatty() and \ not any([opt in sys.argv for opt in ["-q", "--quiet"]]) if ducktape_display_error: # Do not use logger as it has not been initialized yet print(_("%s/clientconfig.ini could not be found or read.\n" "Please copy `docs/clientconfig.example.ini` from the source " "repository and customize.") % config_path) exit(1) # Local logger log = logging.getLogger(__name__) #: Pattern utilisé pour détecter la ligne contenant le mot de passe dans les fichiers pass_regexp = re.compile('[\t ]*pass(?:word)?[\t ]*:[\t ]*(.*)\r?\n?$', flags=re.IGNORECASE) class SimpleMemoize(object): """ Memoization/Lazy """ def __init__(self, f): self.f = f self.val = None def __call__(self, *args, **kwargs): """Attention ! On peut fournir des paramètres, mais comme on mémorise pour la prochaine fois, si on rappelle avec des paramètres différents, on aura quand même la même réponse. Pour l'instant, on s'en fiche puisque les paramètres ne changent pas d'un appel au suivant, mais il faudra s'en préoccuper si un jour on veut changer le comportement.""" if self.val is None: self.val = self.f(*args, **kwargs) # On évite de tout deepcopier. Typiquement, un SSHClient # ne devrait pas l'être (comme dans create_ssh_client) if type(self.val) in [dict, list]: return copy.deepcopy(self.val) else: return self.val ###### # Remote commands @SimpleMemoize def all_keys(options): """Récupère les clés du serveur distant""" return remote_command(options, "listkeys") @SimpleMemoize def all_roles(options): """ Fetch all roles and their users from server """ return remote_command(options, "listroles") @SimpleMemoize def all_files(options): """Récupère les fichiers du serveur distant""" return remote_command(options, "listfiles") @SimpleMemoize def restore_all_files(options): """Récupère les fichiers du serveur distant""" return remote_command(options, "restorefiles") @SimpleMemoize def get_file(options, filename): """ Get the content of one remote file """ return remote_command(options, "getfile", filename) def put_files(options, files): """Dépose les fichiers sur le serveur distant""" return remote_command(options, "putfiles", stdin_contents=files) def rm_file(options, filename): """Supprime le fichier sur le serveur distant""" return remote_command(options, "rmfile", filename) @SimpleMemoize def get_my_roles(options): """Retourne la liste des rôles de l'utilisateur, et également la liste des rôles dont il possède le role-w.""" allroles = all_roles(options) distant_username = allroles.pop("whoami") my_roles = [r for (r, users) in allroles.items() if distant_username in users] my_roles_w = [r[:-2] for r in my_roles if r.endswith("-w")] return (my_roles, my_roles_w) def gen_password(length=15): """ Generate a random password """ return token_urlsafe(length) ###### # Local commands def check_keys(options, recipients=None, quiet=False): """Vérifie les clés, c'est-à-dire, si le mail est présent dans les identités du fingerprint, et que la clé est de confiance (et non expirée/révoquée). * Si ``recipients`` est fourni, vérifie seulement ces recipients. Renvoie la liste de ceux qu'on n'a pas droppés. * Si ``options.force=False``, demandera confirmation pour dropper un recipient dont la clé est invalide. * Sinon, et si ``options.drop_invalid=True``, droppe les recipients automatiquement. * Si rien n'est fourni, vérifie toutes les clés et renvoie juste un booléen disant si tout va bien. """ trusted_recipients = [] # fetch all known keys from server keys = all_keys(options) # check only recipients if recipients is not None: keys = {u: val for (u, val) in keys.items() if u in recipients} for (user, (email, fpr)) in keys.items(): # fetch key # TODO: use ignore-invalid option try: key = get_key_from_fingerprint(fpr) except KeyNotFound: log.error( _("key correponding to fingerprint %s and mail %s not found") % (fpr, email)) # ask to download key res = confirm(options, _("Do you want to try to import it now ?")) if res: receive_key(fpr) try: key = get_key_from_fingerprint(fpr) except KeyNotFound: # this time ignore log.warn(_("Dropping key from %s") % email) continue else: # ignore log.warn(_("Dropping key from %s") % email) continue # check that we should encrypt for this key should_trust = check_key_validity(key, email) if should_trust: log.info("Trusting %s" % email) else: log.warn("Not trusting %s" % email) message = "Abandonner le chiffrement pour cette clé ? (Si vous la conservez, il est possible que gpg crashe)" if confirm(options, message, ('drop', fpr, email)): # ignore key continue trusted_recipients.append(user) if recipients is None: return set(keys.keys()).issubset(trusted_recipients) else: return trusted_recipients def get_recipients_of_roles(options, roles): """Renvoie les destinataires d'une liste de rôles""" recipients = set() allroles = all_roles(options) for role in roles: if role == "whoami": continue for recipient in allroles[role]: recipients.add(recipient) return recipients def get_dest_of_roles(options, roles): """Renvoie la liste des "username : mail (fingerprint)" """ allkeys = all_keys(options) return ["%s : %s (%s)" % (rec, allkeys[rec][0], allkeys[rec][1]) for rec in get_recipients_of_roles(options, roles) if allkeys[rec][1]] def my_encrypt(options, roles, contents): """Chiffre le contenu pour les roles donnés""" allkeys = all_keys(options) recipients = get_recipients_of_roles(options, roles) recipients = check_keys(options, recipients=recipients, quiet=True) keys = [] for recipient in recipients: fpr = allkeys[recipient][1] if fpr: keys.append(get_key_from_fingerprint(fpr)) out = encrypt(contents, keys) if out == '': return False, "Échec de chiffrement" else: return True, out def put_password(options, roles, contents): """Dépose le mot de passe après l'avoir chiffré pour les destinataires dans ``roles``.""" success, enc_pwd_or_error = my_encrypt(options, roles, contents) if success: enc_pwd = enc_pwd_or_error put_files(options, [{'filename': options.filename, 'roles': roles, 'contents': enc_pwd}]) return [True, ""] else: error = enc_pwd_or_error return [False, error] ###### # Interface NEED_FILENAME = [] def need_filename(f): """Décorateur qui ajoutera la fonction à la liste des fonctions qui attendent un filename.""" NEED_FILENAME.append(f) return f def editor(text, annotations=""): """ Lance $EDITOR sur texte. Renvoie le nouveau texte si des modifications ont été apportées, ou None """ if annotations: annotations = "# " + annotations.replace("\n", "\n# ") # Usually, there is already an ending newline in a text document if text and text[-1] != '\n': annotations = '\n' + annotations else: annotations += '\n' text += annotations # use tempory file with .yml to get syntax coloration new_text = "" with NamedTemporaryFile(suffix='.yml') as f: f.write(text.encode("utf-8")) f.flush() subprocess.run( [os.getenv('EDITOR', '/usr/bin/editor'), f.name], check=True) f.seek(0) for line in f.readlines(): # remove comment lines if not line.startswith(b'#'): new_text += line.decode("utf-8", errors='ignore') return new_text def show_files(options): """ Action to list files by role """ my_roles, my_roles_w = get_my_roles(options) files = all_files(options) keys = list(files.keys()) keys.sort() print(_("Available files:")) for fname in keys: froles = files[fname] access = set(my_roles).intersection(froles) != set([]) print((" %s %s (%s)" % ((access and '+' or '-'), fname, ", ".join(froles)))) print((_("""--Mes roles: %s""") % ", ".join(my_roles))) def restore_files(options): """ Action to restore corrumpted files """ print(_("Fichier corrompus :")) files = restore_all_files(options) keys = files.keys() keys.sort() for fname in keys: print(" %s (%s)" % (fname, files[fname])) def show_roles(options): """ Action to list roles and their users """ print(_("Listing available roles and their users")) allroles = all_roles(options) for role, usernames in allroles.items(): if role == "whoami": continue if not role.endswith('-w'): print("-> %s : %s" % (role, ", ".join(usernames))) def list_servers(options): """ Action to list all configured servers """ print(_("List of configured servers")) for name, param in config.items(): host = param.get("host", _("undefined")) remote_cmd = param.get("remote_cmd", _("undefined")) print(f"-> {name:8}\t{host:16}\t{remote_cmd}") def saveclipboard(restore=False, old_clipboard=None): """Enregistre le contenu du presse-papier. Le rétablit si ``restore=True``""" if restore and old_clipboard is None: return if not restore: old_clipboard = pyperclip.paste() else: input(_("Appuyez sur Entrée pour récupérer le contenu précédent du presse papier.")) pyperclip.copy(old_clipboard) return old_clipboard @need_filename def show_file(options): """ Action that decrypt file content """ fname = options.filename gotit, value = get_file(options, fname) if not gotit: log.warn(value) # value contient le message d'erreur return passfile = value content = passfile['contents'] # Kludge (broken db ?) if type(content) == list: log.warn("Eau dans le gaz") content = content[-1] # Déchiffre le contenu texte = decrypt(content) # Est-ce une clé ssh ? is_key = texte.startswith('-----BEGIN RSA PRIVATE KEY-----') # Est-ce que le mot de passe a été caché ? (si non, on utilisera less) is_hidden = is_key # Texte avec mdp caché filtered = "" # Ancien contenu du press papier old_clipboard = None # Essaie de planquer le mot de passe for line in texte.split('\n'): catch_pass = None # On essaie de trouver le pass pour le cacher dans le clipboard # si ce n'est déjà fait et si c'est voulu if not is_hidden and options.clipboard: catch_pass = pass_regexp.match(line) if catch_pass is not None: is_hidden = True # On met le mdp dans le clipboard en mémorisant son ancien contenu old_clipboard = saveclipboard() pyperclip.copy(catch_pass.group(1)) # Et donc on override l'affichage line = "[Le mot de passe a été mis dans le presse papier]" filtered += line + '\n' if is_key: filtered = _("La clé a été mise dans l'agent ssh") shown = "Fichier %s:\n\n%s-----\nVisible par: %s\n" % ( fname, filtered, ','.join(passfile['roles'])) if is_key: with NamedTemporaryFile(suffix='') as key_file: # Génère la clé publique correspondante key_file.write(texte.encode('utf-8')) key_file.flush() pub = subprocess.check_output( ['ssh-keygen', '-y', '-f', key_file.name]) # Charge en mémoire subprocess.check_call(['ssh-add', key_file.name]) # On attend (hors tmpfile) print(shown) input() with NamedTemporaryFile(suffix='') as pub_file: # On met la clé publique en fichier pour suppression pub_file.write(pub) pub_file.flush() subprocess.check_call(['ssh-add', '-d', pub_file.name]) else: # Le binaire à utiliser showbin = "cat" if is_hidden else "less" proc = subprocess.Popen([showbin], stdin=subprocess.PIPE) out = proc.stdin out.write(shown.encode("utf-8")) out.close() os.waitpid(proc.pid, 0) # Repope ancien pass if old_clipboard is not None: saveclipboard(restore=True, old_clipboard=old_clipboard) @need_filename def edit_file(options): """ Action to decrypt, edit a file and encrypt """ # fetch ciphertext from server file_exist, value = get_file(options, options.filename) # fetch my roles from server my_roles, _a = get_my_roles(options) nfile = False comments = "" new_roles = options.roles # Cas du nouveau fichier if not file_exist and "pas les droits" not in value: nfile = True log.info(_("File not found on server")) if not confirm(options, "Créer fichier ?"): exit(1) comments += ("Ceci est un fichier initial contenant un mot de passe\n" "aléatoire, pensez à rajouter une ligne `login: ${login}`\n" "Enregistrez le fichier vide pour annuler.\n") texte = "pass: %s\n" % gen_password() if new_roles is None: new_roles = parse_roles(options, cast=True) passfile = {'roles': new_roles} elif not file_exist: log.warn(value) exit(1) else: passfile = value contents = passfile['contents'] # <ducktape> (waddle waddle) if isinstance(contents, list): contents = contents[-1] # </ducktape> texte = decrypt(contents) if new_roles is None: new_roles = passfile['roles'] # On vérifie qu'on a le droit actuellement (plutôt que de se faire jeter # plus tard) if not any(r + '-w' in my_roles for r in passfile['roles']): log.warn("Aucun rôle en écriture pour ce fichier ! Abandon.") return # On peut vouloir chiffrer un fichier sans avoir la possibilité de le lire # dans le futur, mais dans ce cas on préfère demander confirmation if not any(r + '-w' in my_roles for r in new_roles): message = """Vous vous apprêtez à perdre vos droits en écriture""" + \ """(ROLES ne contient rien parmi : %s) sur ce fichier, continuer ?""" message = message % (", ".join(r[:-2] for r in my_roles if '-w' in r),) if not confirm(options, message): return comments += ("Ce fichier sera chiffré pour les rôles suivants :\n%s\n" "C'est-à-dire pour les utilisateurs suivants :\n%s""" % ( ', '.join(new_roles), '\n'.join(' %s' % rec for rec in get_dest_of_roles(options, new_roles)) )) ntexte = editor(texte, comments) if ((not nfile and ntexte in ['', texte] # pas nouveau, vidé ou pas modifié and set(new_roles) == set(passfile['roles'])) # et on n'a même pas touché à ses rôles, or (nfile and ntexte == '')): # ou alors on a créé un fichier vide. message = "Pas de modification à enregistrer.\n" message += "Si ce n'est pas un nouveau fichier, il a été vidé ou n'a pas été modifié (même pas ses rôles).\n" message += "Si c'est un nouveau fichier, vous avez tenté de le créer vide." if not options.quiet: print(message) else: ntexte = texte if ntexte is None else ntexte success, message = put_password(options, new_roles, ntexte) print(message) _remember_dict = {} def confirm(options, text, remember_key=None): """Demande confirmation, sauf si on est mode ``--force``. Si ``remember_key`` est fourni, il doit correspondre à un objet hashable qui permettra de ne pas poser deux fois les mêmes questions. """ global _remember_dict if options.force: return True if remember_key in _remember_dict: return _remember_dict[remember_key] while True: out = input((text + ' (o/n)')).lower() if out == 'o': res = True break elif out == 'n': res = False break # Remember the answer if remember_key is not None: _remember_dict[remember_key] = res return res @need_filename def remove_file(options): """ Action to delete a file """ fname = options.filename if not confirm(options, 'Êtes-vous sûr de vouloir supprimer %s ?' % (fname,)): return message = rm_file(options, fname) print(message) def my_check_keys(options): """ Action to check all keys """ print(_("Vérification que les clés sont valides (uid correspondant au login) et de confiance.")) print(check_keys(options) and "Base de clés ok" or "Erreurs dans la base") def update_keys(options): """ Action to receive server known keys """ # fetch all keys from server keys = all_keys(options) # receive each key for name, fpr in keys.values(): receive_key(fpr) def recrypt_files(options, strict=False): """Rechiffre les fichiers. Ici, la signification de ``options.roles`` est : strict => on chiffre les fichiers dont *tous* les rôles sont dans la liste non strict => on ne veut rechiffrer que les fichiers qui ont au moins un de ces roles. """ rechiffre_roles = options.roles _, my_roles_w = get_my_roles(options) if rechiffre_roles is None: # Sans précisions, on prend tous les roles qu'on peut rechiffre_roles = my_roles_w # La liste des fichiers allfiles = all_files(options) def is_wanted(fileroles): # On drope ce qui ne peut être écrit if not set(fileroles).intersection(my_roles_w): return False if strict: return set(fileroles).issubset(rechiffre_roles) else: return bool(set(fileroles).intersection(rechiffre_roles)) askfiles = [filename for (filename, fileroles) in allfiles.iteritems() if is_wanted(fileroles)] files = [get_file(options, f) for f in askfiles] # Au cas où on aurait échoué à récupérer ne serait-ce qu'un de ces fichiers, # on affiche le message d'erreur correspondant et on abandonne. for (success, message) in files: if not success: log.warn(message) return # On informe l'utilisateur et on demande confirmation avant de rechiffrer # Si il a précisé --force, on ne lui demandera rien. filenames = ", ".join(askfiles) message = _( "Vous vous apprêtez à rechiffrer les fichiers suivants :\n%s") % filenames if not confirm(options, message + "\nConfirmer"): exit(2) # On rechiffre to_put = [{'filename': f['filename'], 'roles': f['roles'], 'contents': my_encrypt(options, f['roles'], decrypt(f['contents']))[-1]} for [success, f] in files] if to_put: if not options.quiet: print("Rechiffrement de %s" % (", ".join([f['filename'] for f in to_put]))) put_files(options, to_put) else: log.warn(_("Aucun fichier n'a besoin d'être rechiffré")) def parse_roles(options, cast=False): """Interprête la liste de rôles fournie par l'utilisateur. Si il n'en a pas fourni, c'est-à-dire que roles vaut None, alors on considère cette valeur comme valide. Cependant, si ``cast`` est vraie, cette valeur est remplacée par tous les roles en écriture (*-w) de l'utilisateur. Renvoie ``False`` si au moins un de ces rôles pose problème. poser problème, c'est : * être un role-w (il faut utiliser le role sans le w) * ne pas exister dans la config du serveur """ if options.roles is None and not cast: return options.roles strroles = options.roles allroles = all_roles(options) _, my_roles_w = get_my_roles(options) if strroles is None: # L'utilisateur n'a rien donné, on lui donne les rôles (non -w) dont il possède le -w return my_roles_w ret = set() for role in strroles.split(','): if role not in allroles.keys(): log.warn("role %s do not exists" % role) exit(1) if role.endswith('-w'): log.warn("role %s should not be used, rather use %s" % (role, role[:-2])) exit(1) ret.add(role) return list(ret) def insult_on_nofilename(options, parser): """Insulte (si non quiet) et quitte si aucun nom de fichier n'a été fourni en commandline.""" if options.filename is None: log.warn(_("You need to provide a filename with this command")) if not options.quiet: parser.print_help() exit(1) def main(): # Gestion des arguments parser = argparse.ArgumentParser( description=_("Group passwords manager based on GPG."), ) parser.add_argument( 'filename', nargs='?', default=None, help=_("name of file to show or edit") ) parser.add_argument( '-v', '--verbose', action='count', default=1, help=_("verbose mode, multiple -v options increase verbosity"), ) parser.add_argument( '-q', '--quiet', action='store_true', default=False, help=_("silent mode: hide errors, overrides verbosity"), ) parser.add_argument( '-s', '--server', default='DEFAULT', metavar="SRV", help=_("select another server than DEFAULT"), ) parser.add_argument( '--no-clipboard', action='store_false', default=None, dest='clipboard', help=_("do not try to store password in clipboard"), ) parser.add_argument( '-f', '--force', action='store_true', default=False, help=_("do not prompt confirmation"), ) parser.add_argument( '--drop-invalid', action='store_true', default=False, dest='drop_invalid', help=_("need --force, drop untrusted keys without confirmation."), ) parser.add_argument( '--roles', nargs='?', default=None, help=_("specify for which roles to crypt (default to all roles, or do " "not change if editing)"), ) # Actions possibles arg_grp = parser.add_argument_group("actions") action_grp = arg_grp.add_mutually_exclusive_group(required=False) action_grp.add_argument( '--view', action='store_const', dest='action', const=show_file, help=_("read file (default)"), ) action_grp.add_argument( '-e', '--edit', action='store_const', dest='action', const=edit_file, help=_("edit (or create) file"), ) action_grp.add_argument( '--remove', action='store_const', dest='action', const=remove_file, help=_("erase file"), ) action_grp.add_argument( '-l', '--list', action='store_const', dest='action', const=show_files, help=_("list files"), ) action_grp.add_argument( '-r', '--restore', action='store_const', dest='action', const=restore_files, help=_("restore corrumpted files"), ) action_grp.add_argument( '--check-keys', action='store_const', dest='action', const=my_check_keys, help=_("check keys"), ) action_grp.add_argument( '--update-keys', action='store_const', dest='action', const=update_keys, help=_("update keys"), ) action_grp.add_argument( '--list-roles', action='store_const', dest='action', const=show_roles, help=_("list existing roles"), ) action_grp.add_argument( '--list-servers', action='store_const', dest='action', const=list_servers, help=_("list servers"), ) action_grp.add_argument( '--recrypt', action='store_const', dest='action', const=recrypt_files, help=_("recrypt all files having a role listed in --roles"), ) action_grp.set_defaults(action=show_file) # On parse les options fournies en commandline options = parser.parse_args() # Active le logger avec des couleurs # Par défaut on affiche >= WARNING options.verbose = 40 - (10 * options.verbose) if not options.quiet else 0 logging.addLevelName(logging.INFO, "\033[1;36mINFO\033[1;0m") logging.addLevelName(logging.WARNING, "\033[1;33mWARNING\033[1;0m") logging.addLevelName(logging.ERROR, "\033[1;91mERROR\033[1;0m") logging.addLevelName(logging.DEBUG, "\033[1;37mDEBUG\033[1;0m") logging.basicConfig( level=options.verbose, format='\033[90m%(asctime)s\033[1;0m %(name)s %(levelname)s %(message)s' ) # On calcule les options qui dépendent des autres. # C'est un peu un hack, peut-être que la méthode propre serait de surcharger argparse.ArgumentParser # et argparse.Namespace, mais j'ai pas réussi à comprendre commenr m'en sortir. # ** Presse papier ** # Si l'utilisateur n'a rien dit (ni option --clipboard ni --noclipboard), # on active le clipboard par défaut, à la condition # que xclip existe et qu'il a un serveur graphique auquel parler. if options.clipboard is None: options.clipboard = bool( os.getenv('DISPLAY')) and os.path.exists('/usr/bin/xclip') # On récupère les données du serveur à partir du nom fourni options.serverdata = config[options.server] # On parse les roles fournis, et il doivent exister, ne pas être -w… # parse_roles s'occupe de ça # NB : ça nécessite de se connecter au serveur, or, pour list_servers on n'en a pas besoin # Il faudrait ptêtre faire ça plus proprement, en attendant, je ducktape. if options.action != list_servers: options.roles = parse_roles(options) # Si l'utilisateur a demandé une action qui nécessite un nom de fichier, # on vérifie qu'il a bien fourni un nom de fichier. if options.action in NEED_FILENAME: insult_on_nofilename(options, parser) # On exécute l'action demandée options.action(options) if __name__ == "__main__": main()