Newer
Older
"""Gestion centralisée des mots de passe avec chiffrement GPG
Copyright (C) 2010-2013 Cr@ns <roots@crans.org>
Authors : Daniel Stan <daniel.stan@crans.org>
Vincent Le Gallic <legallic@crans.org>
"""
Vincent Le gallic
committed
# Import builtins
Vincent Le gallic
committed
import time
Vincent Le gallic
committed
# Import de la config
# Oui, le nom de la commande est dans la config, mais on n'a pas encore accès à la config
bootstrap_cmd_name = os.path.split(sys.argv[0])[1]
sys.path.append(os.path.expanduser("~/.config/%s/" % (bootstrap_cmd_name,)))
import clientconfig as config
except ImportError:
Vincent Le gallic
committed
if sys.stderr.isatty() and not any([opt in sys.argv for opt in ["-q", "--quiet"]]):
sys.stderr.write(u"Va lire le fichier README.\n".encode("utf-8"))
#: pattern utilisé pour détecter la ligne contenant le mot de passe dans les fichiers
PASS = re.compile('[\t ]*pass(?:word)?[\t ]*:[\t ]*(.*)\r?\n?$',
Vincent Le gallic
committed
## Conf qu'il faudrait éliminer en passant ``parsed`` aux fonctions
#: Mode verbeux
VERB = False
#: Par défaut, place-t-on le mdp dans le presse-papier ?
CLIPBOARD = bool(os.getenv('DISPLAY')) and os.path.exists('/usr/bin/xclip')
#: Mode «ne pas demander confirmation»
FORCED = False
#: Droits à définir sur le fichier en édition
NEWROLES = None
#: Serveur à interroger (peuplée à l'exécution)
SERVER = None
Vincent Le gallic
committed
#: Path to gpg binary
Vincent Le gallic
committed
#: Paramètres à fournir à gpg en fonction de l'action désirée
'decrypt' : ['-d'],
'encrypt' : ['--armor', '-es'],
'receive-keys' : ['--recv-keys'],
Vincent Le gallic
committed
'list-keys' : ['--list-keys', '--with-colons', '--fixed-list-mode',
'--with-fingerprint', '--with-fingerprint'], # Ce n'est pas une erreur. Il faut 2 --with-fingerprint pour avoir les fingerprints des subkeys.
Vincent Le gallic
committed
#: Mapping (lettre de trustlevel) -> (signification, faut-il faire confiance à la clé)
Vincent Le gallic
committed
u"-" : (u"inconnue (pas de valeur assignée)", False),
u"o" : (u"inconnue (nouvelle clé)", False),
u"i" : (u"invalide (self-signature manquante ?)", False),
u"n" : (u"nulle", False),
u"m" : (u"marginale", True),
u"f" : (u"entière", True),
u"u" : (u"ultime", True),
u"r" : (u"révoquée", False),
u"e" : (u"expirée", False),
Vincent Le gallic
committed
u"q" : (u"non définie", False),
Vincent Le gallic
committed
def gpg(command, args=None, verbose=False):
Vincent Le gallic
committed
donnés. Renvoie son entrée standard et sa sortie standard."""
full_command = [GPG]
full_command.extend(GPG_ARGS[command])
if args:
full_command.extend(args)
Vincent Le gallic
committed
if verbose:
stderr = sys.stderr
stderr = subprocess.PIPE
proc = subprocess.Popen(full_command,
stdin = subprocess.PIPE,
stdout = subprocess.PIPE,
Vincent Le gallic
committed
if not verbose:
Vincent Le gallic
committed
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
def _parse_timestamp(string, canbenone=False):
"""Interprète ``string`` comme un timestamp depuis l'Epoch."""
if string == u'' and canbenone:
return None
return datetime.datetime(*time.localtime(int(string))[:7])
def _parse_pub(data):
"""Interprète une ligne ``pub:``"""
d = {
u'trustletter' : data[1],
u'length' : int(data[2]),
u'algorithm' : int(data[3]),
u'longid' : data[4],
u'signdate' : _parse_timestamp(data[5]),
u'expiredate' : _parse_timestamp(data[6], canbenone=True),
u'ownertrustletter' : data[8],
u'capabilities' : data[11],
}
return d
def _parse_uid(data):
"""Interprète une ligne ``uid:``"""
d = {
u'trustletter' : data[1],
u'signdate' : _parse_timestamp(data[5], canbenone=True),
u'hash' : data[7],
u'uid' : data[9],
}
return d
def _parse_fpr(data):
"""Interprète une ligne ``fpr:``"""
d = {
u'fpr' : data[9],
}
return d
def _parse_sub(data):
"""Interprète une ligne ``sub:``"""
d = {
u'trustletter' : data[1],
u'length' : int(data[2]),
u'algorithm' : int(data[3]),
u'longid' : data[4],
u'signdate' : _parse_timestamp(data[5]),
u'expiredate' : _parse_timestamp(data[6], canbenone=True),
u'capabilities' : data[11],
}
return d
#: Functions to parse the recognized fields
GPG_PARSERS = {
u'pub' : _parse_pub,
u'uid' : _parse_uid,
u'fpr' : _parse_fpr,
u'sub' : _parse_sub,
}
def _gpg_printdebug(d):
print("current_pub : %r" % d.get("current_pub", None))
print("current_sub : %r" % d.get("current_sub", None))
def parse_keys(gpgout, debug=False):
"""Parse l'output d'un listing de clés gpg."""
ring = {}
init_value = u"initialize" # Valeur utilisée pour dire "cet objet n'a pas encore été rencontré pendant le parsing"
current_pub = init_value
current_sub = init_value
for line in iter(gpgout.readline, ''):
# La doc dit que l'output est en UTF-8 «regardless of any --display-charset setting»
line = line.decode("utf-8")
line = line.split(":")
field = line[0]
if field in GPG_PARSERS.keys():
if debug:
print("\nbegin loop. met %s :" % (field))
_gpg_printdebug(locals())
try:
content = GPG_PARSERS[field](line)
except:
print("*** FAILED ***")
print(line)
raise
if field == u"pub":
# Nouvelle clé
# On sauvegarde d'abord le dernier sub (si il y en a un) dans son pub parent
if current_sub != init_value:
current_pub["subkeys"].append(current_sub)
# Ensuite on sauve le pub précédent (si il y en a un) dans le ring
if current_pub != init_value:
ring[current_pub[u"fpr"]] = current_pub
# On place le nouveau comme pub courant
current_pub = content
# Par défaut, il n'a ni subkeys, ni uids
current_pub[u"subkeys"] = []
current_pub[u"uids"] = []
# On oublié l'éventuel dernier sub rencontré
current_sub = init_value
elif field == u"fpr":
if current_sub != init_value:
# On a lu un sub depuis le dernier pub, donc le fingerprint est celui du dernier sub rencontré
current_sub[u"fpr"] = content[u"fpr"]
else:
# Alors c'est le fingerprint du pub
current_pub[u"fpr"] = content[u"fpr"]
elif field == u"uid":
current_pub[u"uids"].append(content)
elif field == u"sub":
# Nouvelle sous-clé
# D'abord on sauvegarde la précédente (si il y en a une) dans son pub parent
if current_sub != init_value:
current_pub[u"subkeys"].append(current_sub)
# On place la nouvelle comme sub courant
current_sub = content
if debug:
_gpg_printdebug(locals())
print("parsed object : %r" % content)
# À la fin, il faut sauvegarder les derniers (sub, pub) rencontrés,
# parce que leur sauvegarde n'a pas encore été déclenchée
if current_sub != init_value:
current_pub["subkeys"].append(current_sub)
if current_pub != init_value:
ring[current_pub[u"fpr"]] = current_pub
return ring
class simple_memoize(object):
""" Memoization/Lazy """
def __init__(self, f):
self.f = f
self.val = None
def __call__(self):
self.val = self.f()
return self.val
######
## Remote commands
def ssh(command, arg = None):
"""Lance ssh avec les arguments donnés. Renvoie son entrée
standard et sa sortie standard."""
full_command = list(SERVER['server_cmd'])
full_command.append(command)
if arg:
full_command.append(arg)
proc = subprocess.Popen(full_command,
stdin = subprocess.PIPE,
stdout = subprocess.PIPE,
stderr = sys.stderr,
close_fds = True)
return proc.stdin, proc.stdout
def remote_command(command, arg = None, stdin_contents = None):
"""Exécute la commande distante, et retourne la sortie de cette
commande"""
sshin, sshout = ssh(command, arg)
if not stdin_contents is None:
raw_out = sshout.read()
return json.loads(raw_out)
def all_keys():
"""Récupère les clés du serveur distant"""
return remote_command("listkeys")
def all_roles():
"""Récupère les roles du serveur distant"""
return remote_command("listroles")
def all_files():
"""Récupère les fichiers du serveur distant"""
return remote_command("listfiles")
def get_files(filenames):
"""Récupère le contenu des fichiers distants"""
return remote_command("getfiles", stdin_contents=filenames)
def put_files(files):
"""Dépose les fichiers sur le serveur distant"""
return remote_command("putfiles", stdin_contents=files)
def rm_file(filename):
"""Supprime le fichier sur le serveur distant"""
return remote_command("rmfile", filename)
"""Retourne la liste des rôles de l'utilisateur"""
return filter(lambda role: SERVER['user'] in allr[role], allr.keys())
random.seed(datetime.datetime.now().microsecond)
chars = string.letters + string.digits + '/=+*'
length = 15
return u''.join([random.choice(chars) for _ in xrange(length)])
######
## Local commands
def update_keys():
"""Met à jour les clés existantes"""
_, stdout = gpg("receive-keys", [key for _, key in keys.values() if key])
return stdout.read().decode("utf-8")
if VERB:
print("M : l'uid correspond au mail du fingerprint\nC : confiance OK (inclu la vérification de non expiration).\n")
keys = all_keys()
Vincent Le gallic
committed
_, gpgout = gpg('list-keys')
localring = parse_keys(gpgout)
failed = False
for (mail, fpr) in keys.values():
if fpr:
Vincent Le gallic
committed
if VERB:
print((u"Checking %s… " % (mail)).encode("utf-8"), end="")
Vincent Le gallic
committed
key = localring.get(fpr, None)
# On vérifie qu'on possède la clé…
Vincent Le gallic
committed
if not key is None:
# …qu'elle correspond au mail…
Vincent Le gallic
committed
if any([u"<%s>" % (mail,) in u["uid"] for u in key["uids"]]):
Vincent Le gallic
committed
meaning, trustvalue = GPG_TRUSTLEVELS[key["trustletter"]]
# … et qu'on lui fait confiance
if not trustvalue:
print((u"--> Fail on %s:%s\nLa confiance en la clé est : %s" % (fpr, mail, meaning,)).encode("utf-8"))
failed = True
print((u"--> Fail on %s:%s\n!! Le fingerprint et le mail ne correspondent pas !" % (fpr, mail)).encode("utf-8"))
failed = True
else:
print((u"--> Fail on %s:%s\nPas (ou trop) de clé avec ce fingerprint." % (fpr, mail)).encode("utf-8"))
failed = True
return not failed
def get_recipients_of_roles(roles):
"""Renvoie les destinataires d'un rôle"""
recipients = set()
allroles = all_roles()
for role in roles:
for recipient in allroles[role]:
recipients.add(recipient)
return recipients
def get_dest_of_roles(roles):
"""Renvoie la liste des "username : mail (fingerprint)" """
return [u"%s : %s (%s)" % (rec, allkeys[rec][0], allkeys[rec][1])
for rec in get_recipients_of_roles(roles) if allkeys[rec][1]]
"""Chiffre ``contents`` pour les ``roles`` donnés"""
recipients = get_recipients_of_roles(roles)
fpr = allkeys[recipient][1]
if fpr:
fpr_recipients.append("-r")
fpr_recipients.append(fpr)
stdin, stdout = gpg("encrypt", fpr_recipients)
stdin.write(contents.encode("utf-8"))
out = stdout.read().decode("utf-8")
return [False, u"Échec de chiffrement"]
def decrypt(contents):
"""Déchiffre le contenu"""
stdin, stdout = gpg("decrypt")
stdin.write(contents.encode("utf-8"))
return stdout.read().decode("utf-8")
def put_password(name, roles, contents):
"""Dépose le mot de passe après l'avoir chiffré pour les
destinataires donnés"""
success, enc_pwd_or_error = encrypt(roles, contents)
if NEWROLES != None:
roles = NEWROLES
if success:
enc_pwd = enc_pwd_or_error
return put_files([{'filename' : name, 'roles' : roles, 'contents' : enc_pwd}])[0]
error = enc_pwd_or_error
return [False, error]
def get_password(name):
"""Récupère le mot de passe donné par name"""
gotit, remotefile = get_files([name])[0]
if gotit:
remotefile = decrypt(remotefile['contents'])
return [gotit, remotefile]
def editor(texte, annotations=u""):
""" Lance $EDITOR sur texte.
Renvoie le nouveau texte si des modifications ont été apportées, ou None
"""
# Avoid syntax hilight with ".txt". Would be nice to have some colorscheme
# for annotations ...
f = tempfile.NamedTemporaryFile(suffix='.txt')
if annotations:
annotations = "# " + annotations.replace("\n", "\n# ")
f.write((texte + "\n" + annotations).encode("utf-8"))
proc = subprocess.Popen([os.getenv('EDITOR', '/usr/bin/editor'), f.name])
ntexte = f.read().decode("utf-8")
ntexte = u'\n'.join(filter(lambda l: not l.startswith('#'), ntexte.split('\n')))
"""Affiche la liste des fichiers disponibles sur le serveur distant"""
print(u"Liste des fichiers disponibles :".encode("utf-8"))
files = all_files()
keys = files.keys()
keys.sort()
for fname in keys:
froles = files[fname]
access = set(my_roles).intersection(froles) != set([])
print((u" %s %s (%s)" % ((access and '+' or '-'), fname, ", ".join(froles))).encode("utf-8"))
print((u"""--Mes roles: %s""" % (", ".join(my_roles),)).encode("utf-8"))
"""Affiche la liste des roles existants"""
print(u"Liste des roles disponibles".encode("utf-8"))
for (role, usernames) in all_roles().iteritems():
print((u" * %s : %s" % (role, ", ".join(usernames))).encode("utf-8"))
"""Affiche la liste des serveurs disponibles"""
print(u"Liste des serveurs disponibles".encode("utf-8"))
for server in config.servers.keys():
old_clipboard = None
def saveclipboard(restore=False):
"""Enregistre le contenu du presse-papier. Le rétablit si ``restore=True``"""
global old_clipboard
if restore and old_clipboard == None:
return
act = '-in' if restore else '-out'
proc = subprocess.Popen(['xclip', act, '-selection', 'clipboard'],
stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=sys.stderr)
if not restore:
old_clipboard = proc.stdout.read()
else:
raw_input(u"Appuyez sur Entrée pour récupérer le contenu précédent du presse papier.".encode("utf-8"))
proc.stdin.write(old_clipboard)
proc.stdin.close()
proc.stdout.close()
"""Place ``texte`` dans le presse-papier en mémorisant l'ancien contenu."""
proc =subprocess.Popen(['xclip', '-selection', 'clipboard'],\
stdin=subprocess.PIPE, stdout=sys.stdout, stderr=sys.stderr)
proc.stdin.write(texte.encode("utf-8"))
return u"[Le mot de passe a été mis dans le presse papier]"
gotit, value = get_files([fname])[0]
if not gotit:
print(value.encode("utf-8")) # value contient le message d'erreur
sin.write(value['contents'].encode("utf-8"))
texte = sout.read().decode("utf-8")
ntexte = u""
hidden = False # Est-ce que le mot de passe a été caché ?
lines = texte.split('\n')
for line in lines:
catchPass = PASS.match(line)
if catchPass != None and CLIPBOARD:
hidden = True
line = clipboard(catchPass.group(1))
ntexte += line + '\n'
showbin = "cat" if hidden else "less"
proc = subprocess.Popen([showbin], stdin=subprocess.PIPE)
raw = u"Fichier %s:\n\n%s-----\nVisible par: %s\n" % (fname, ntexte, ','.join(value['roles']))
out.write(raw.encode("utf-8"))
gotit, value = get_files([fname])[0]
if not gotit and not "pas les droits" in value:
annotations += u"""Ceci est un fichier initial contenant un mot de passe
aléatoire, pensez à rajouter une ligne "login: ${login}"
Enregistrez le fichier vide pour annuler.\n"""
texte = u"pass: %s\n" % gen_password()
roles = get_my_roles()
# Par défaut les roles d'un fichier sont ceux en écriture de son
# créateur
roles = [ r[:-2] for r in roles if r.endswith('-w') ]
print(u"Vous ne possédez aucun rôle en écriture ! Abandon.".encode("utf-8"))
elif not gotit:
print(value.encode("utf-8")) # value contient le message d'erreur
return
sin.write(value['contents'].encode("utf-8"))
texte = sout.read().decode("utf-8")
# On récupère les nouveaux roles si ils ont été précisés, sinon on garde les mêmes
value['roles'] = NEWROLES or value['roles']
annotations += u"""Ce fichier sera chiffré pour les rôles suivants :\n%s\n
C'est-à-dire pour les utilisateurs suivants :\n%s""" % (
', '.join(value['roles']),
'\n'.join(' %s' % rec for rec in get_dest_of_roles(value['roles']))
)
ntexte = editor(texte, annotations)
if ((not nfile and ntexte in [u'', texte] and NEWROLES == None) or # Fichier existant vidé ou inchangé
(nfile and ntexte == u'')): # Nouveau fichier créé vide
print(u"Pas de modification effectuée".encode("utf-8"))
success, message = put_password(fname, value['roles'], ntexte)
print(message.encode("utf-8"))
"""Demande confirmation, sauf si on est mode ``FORCED``"""
out = raw_input((text + u' (O/N)').encode("utf-8")).lower()
if out == 'o':
return True
elif out == 'n':
return False
def remove_file(fname):
if not confirm(u'Êtes-vous sûr de vouloir supprimer %s ?' % fname):
message = rm_file(fname)
print(message.encode("utf-8"))
"""Vérifie les clés et affiche un message en fonction du résultat"""
print(u"Vérification que les clés sont valides (uid correspondant au login) et de confiance.")
print((check_keys() and u"Base de clés ok" or u"Erreurs dans la base").encode("utf-8"))
"""Met à jour les clés existantes et affiche le résultat"""
def recrypt_files():
# Ici, la signification de NEWROLES est : on ne veut rechiffrer que les fichiers qui ont au moins un de ces roles
rechiffre_roles = NEWROLES
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
my_roles_w = [r for r in my_roles if r.endswith("-w")]
if rechiffre_roles == None:
# Sans précisions, on prend tous les roles qu'on peut
rechiffre_roles = my_roles
# On ne conserve que les rôles en écriture
rechiffre_roles = [ r[:-2] for r in rechiffre_roles if r.endswith('-w')]
# La liste des fichiers
allfiles = all_files()
# On ne demande que les fichiers dans lesquels on peut écrire
# et qui ont au moins un role dans ``roles``
askfiles = [filename for (filename, fileroles) in allfiles.iteritems()
if set(fileroles).intersection(roles) != set()
and set(fileroles).intersection(my_roles_w) != set()]
files = get_files(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:
print(message.encode("utf-8"))
return
# On rechiffre
to_put = [{'filename' : f['filename'],
'roles' : f['roles'],
'contents' : encrypt(f['roles'], decrypt(f['contents']))}
for f in files]
if to_put:
print((u"Rechiffrement de %s" % (", ".join([f['filename'] for f in to_put]))).encode("utf-8"))
results = put_files(to_put)
# On affiche les messages de retour
for i in range(len(results)):
Vincent Le gallic
committed
print(u"%s : %s" % (to_put[i]['filename'], results[i][1]))
else:
print(u"Aucun fichier n'a besoin d'être rechiffré".encode("utf-8"))
"""Interprête une liste de rôles fournie par l'utilisateur.
Renvoie ``False`` si au moins un de ces rôles pose problème."""
if strroles == None: return None
roles = all_roles()
my_roles = filter(lambda r: SERVER['user'] in roles[r], roles.keys())
my_roles_w = [ r[:-2] for r in my_roles if r.endswith('-w') ]
ret = set()
writable = False
for role in strroles.split(','):
if role not in roles.keys():
print((u"Le rôle %s n'existe pas !" % role).encode("utf-8"))
print((u"Le rôle %s ne devrait pas être utilisé ! (utilisez %s)")
% (role, role[:-2])).encode("utf-8")
return False
writable = writable or role in my_roles_w
ret.add(role)
if not FORCED and not writable:
if not confirm(u"""Vous vous apprêtez à perdre vos droits d'écritures\
(ROLES ne contient pas %s) sur ce fichier, continuer ?""" %
parser = argparse.ArgumentParser(description="trousseau crans")
parser.add_argument('-s', '--server', default='default',
help="Utilisation d'un serveur alternatif (test, backup, etc)")
parser.add_argument('-v', '--verbose', action='store_true', default=False,
Vincent Le gallic
committed
parser.add_argument('-q', '--quiet', action='store_true', default=False,
help="Mode silencieux. Cache les message d'erreurs (override --verbose).")
parser.add_argument('-c', '--clipboard', action='store_true', default=None,
help="Stocker le mot de passe dans le presse papier")
parser.add_argument('--no-clip', '--noclip', '--noclipboard', action='store_false', default=None,
dest='clipboard',
help="Ne PAS stocker le mot de passe dans le presse papier")
parser.add_argument('-f', '--force', action='store_true', default=False,
help="Ne pas demander confirmation")
# Actions possibles
action_grp = parser.add_mutually_exclusive_group(required=False)
action_grp.add_argument('-e', '--edit', action='store_const', dest='action',
default=show_file, const=edit_file,
action_grp.add_argument('--view', action='store_const', dest='action',
default=show_file, const=show_file,
help="Voir le fichier")
action_grp.add_argument('--remove', action='store_const', dest='action',
default=show_file, const=remove_file,
help="Effacer le fichier")
action_grp.add_argument('-l', '--list', action='store_const', dest='action',
default=show_file, const=show_files,
action_grp.add_argument('--check-keys', action='store_const', dest='action',
default=show_file, const=my_check_keys,
action_grp.add_argument('--update-keys', action='store_const', dest='action',
default=show_file, const=my_update_keys,
action_grp.add_argument('--list-roles', action='store_const', dest='action',
default=show_file, const=show_roles,
help="Lister les rôles existants")
action_grp.add_argument('--list-servers', action='store_const', dest='action',
default=show_file, const=show_servers,
help="Lister les serveurs")
action_grp.add_argument('--recrypt-files', action='store_const', dest='action',
default=show_file, const=recrypt_files,
help="Rechiffrer les mots de passe. (Avec les mêmes rôles qu'avant, sert à rajouter un lecteur)")
parser.add_argument('--roles', nargs='?', default=None,
help="""Liste de roles (séparés par des virgules).
Avec --edit, le fichier sera chiffré pour exactement ces roles
(par défaut, tous vos rôles en écriture seront utilisés).
Avec --recrypt-files, tous les fichiers ayant au moins un de ces roles (et pour lesquels vous avez le droit d'écriture) seront rechiffrés
(par défaut, tous les fichiers pour lesquels vous avez les droits en écriture sont rechiffrés).""")
parser.add_argument('fname', nargs='?', default=None,
Vincent Le gallic
committed
QUIET = parsed.quiet
VERB = parsed.verbose and not QUIET
if parsed.clipboard != None:
CLIPBOARD = parsed.clipboard
NEWROLES = parse_roles(parsed.roles)
if NEWROLES != False:
if parsed.action.func_code.co_argcount == 0:
parsed.action()
elif parsed.fname == None:
Vincent Le gallic
committed
if not QUIET:
print(u"Vous devez fournir un nom de fichier avec cette commande".encode("utf-8"))
Vincent Le gallic
committed
parser.print_help()
sys.exit(1)