1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
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
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
|
# First Aid Kit - diagnostic and repair tool for Linux
# encoding: utf-8
# Copyright (C) 2009 Red Hat, Inc.
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
#
# Red Hat author: Miloslav Trmač <mitr@redhat.com>
import collections
import logging
from pyfirstaidkit.plugins import Plugin,Flow
from pyfirstaidkit.returns import *
from pyfirstaidkit.reporting import PLUGIN
import volume_key
# A hackish alternative way to initialize NSS:
# import rpm; rpm.ts()
# (currently conflicts with rpm.py in the current directory)
import nss.nss
nss.nss.nss_init_nodb()
class KeyRecoveryPlugin(Plugin):
"""This plugin allows restoring access to encrypted volumes using escrow
packets generated by volume_key."""
flows = {}
flows["recoverEncryptionKey"] = Flow({
Plugin.initial: {Return: "findEncryptedVolumes"},
"findEncryptedVolumes" : {ReturnSuccess: "openPacket",
ReturnFailure: Plugin.final,
None: Plugin.final},
"openPacket" : {ReturnSuccess: "addPassphrase",
ReturnFailure: Plugin.final,
None: Plugin.final},
"addPassphrase" : {ReturnSuccess: Plugin.final,
ReturnFailure: Plugin.final,
None: Plugin.final}
}, description="Add a password to an encrypted volume "
"using a provided escrow packet")
name = "Encryption key recovery plugin"
version = "0.0.1"
author = "Miloslav Trmač"
description = "Automates recovery of LUKS volume key"
@classmethod
def getDeps(cls):
return set(["root"])
def __init__(self, *args, **kwargs):
Plugin.__init__(self, *args, **kwargs)
self._ui = volume_key.UI()
self._ui.generic_cb = self._vk_ui_generic_cb
self._ui.passphrase_cb = self._vk_ui_passphrase_cb
def _vk_ui_generic_cb(self, prompt, echo):
prompt = "%s:" % (prompt,)
if echo:
r = self._reporting.text_question_wait(prompt, origin = self)
else:
r = self._reporting.password_question_wait(prompt, origin = self)
if r == '':
return None
return r
def _vk_ui_passphrase_cb(self, prompt, unused_failed_attempts):
r = self._reporting.password_question_wait("%s:" % (prompt,),
origin = self)
if r == '':
return None
return r
def findEncryptedVolumes(self):
# Note: This assumes "the system" has automatically set up RAID, LVM
# etc. to make the encrypted volumes visible.
self._volumes = collections.defaultdict(list)
with open("/proc/partitions") as f:
for line in f.readlines()[2:]:
(major, minor, blocks, name) = line.split()
if name.startswith("ram"):
continue
path = "/dev/%s" % name
try:
vol = volume_key.Volume.open(path)
except RuntimeError, e:
self._reporting.info("Error examining %s: %s" %
(path, str(e)), origin = self,
importance = logging.DEBUG)
else:
if vol.format in (volume_key.VOLUME_FORMAT_LUKS,):
self._volumes[vol.uuid].append(path)
if len(self._volumes) == 0:
self._reporting.info("No encrypted volume found", origin = self)
self._result = ReturnFailure
return
self._result = ReturnSuccess
def openPacket(self):
pack = self._openPacket()
if pack is None:
return
candidates = self._getCandidates(pack)
volume_path = self._chooseCandidate(candidates)
if volume_path is None:
return
self._packet = pack
self._volume_path = volume_path
self._result = ReturnSuccess
def _openPacket(self):
"""Ask for a packet and open it.
Return a packet if OK (setting self._packet_path), None otherwise.
"""
packet_path = self._reporting.filename_question_wait \
("Select an escrow packet:", origin = self)
try:
with open(packet_path, "rb") as f:
packet_data = f.read()
except IOError, e:
self._reporting.error("Error reading %s: %s" %
(packet_path, str(e)), origin = self)
return None
try:
fmt = volume_key.Packet.get_format(packet_data)
# In particular, PACKET_FORMAT_ASSYMETRIC requries a NSS db and is
# not supported
if fmt not in (volume_key.PACKET_FORMAT_CLEARTEXT,
volume_key.PACKET_FORMAT_PASSPHRASE):
self._reporting.error("Unsupported format %s of %s" %
(fmt, packet_path), origin = self)
return None
pack = volume_key.Packet.open(packet_data, self._ui)
except RuntimeError, e:
self._reporting.error("Error opening %s: %s" %
(packet_path, str(e)), origin = self)
return None
self._packet_path = packet_path
return pack
def _getCandidates(self, pack):
"""Return a list of volume paths that match pack."""
candidates = []
for path in self._volumes[pack.uuid]:
try:
vol = volume_key.Volume.open(path)
(unused_r, unused_warnings) = pack.packet_match_volume(vol)
except RuntimeError:
# volume_key.Volume.open() passed previously, so this exception
# probably means the packet did not match
pass
else:
candidates.append(path)
return candidates
def _chooseCandidate(self, candidates):
"""Let the user choose a volume out of candidates.
Return volume path, or None on error.
"""
if len(candidates) == 0:
self._reporting.error("Packet %s does not match any known volume" %
(self._packet_path,), origin = self)
self._result = ReturnFailure
return None
elif len(candidates) == 1:
volume_path = candidates[0]
if not self._reporting.choice_question_wait \
("This packet can be applied to volume %s. Continue?" %
volume_path, ((True, "Yes"), (False, "No")),
origin = self):
self._result = ReturnFailure
return None
else:
# This really shouldn't happen, that's what UUIDs are for...
choices = [(p, p) for p in candidates]
choices.append((None, "Cancel operation"))
volume_path = self._reporting.choice_question_wait \
("Which volume do you want to apply this packet to?", choices,
origin = self)
if volume_path is None:
self._result = ReturnFailure
return None
return volume_path
def addPassphrase(self):
try:
vol = volume_key.Volume.open(self._volume_path)
except RuntimeError, e:
self._reporting.error("Error opening %s: %s" %
(self._volume_path, str(e)), origin = self)
return
try:
(r, warnings) = self._packet.packet_match_volume(vol)
except RuntimeError, e:
self._reporting.error("%s does not match %s: %s" %
(self._packet_path, self._volume_path,
str(e)), origin = self)
return
if (r == volume_key.PACKET_MATCH_UNSURE and
not self._reporting.choice_question_wait \
("%s perhaps does not match %s:\n%s\nAre you sure you want "
" to use this packet?" %
(self._packet_path, self._volume_path, "\n".join(warnings)),
((True, "Yes"), (False, "No")), origin = self)):
self._result = ReturnFailure
return
self._reporting.info("Applying packet to set up a new passphrase...",
origin = self)
try:
vol.apply_packet(self._packet, volume_key.SECRET_DEFAULT, self._ui)
except RuntimeError, e:
self._reporting.error("Error applying %s to %s: %s" %
(self._packet_path, self._volume_path,
str(e)), origin = self)
self._result = ReturnFailure
return
self._reporting.info("Packet applied succesfully...", origin = self)
self._result = ReturnSuccess
return
def get_plugin():
return KeyRecoveryPlugin
|
