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
|
# Copyright (C) 2016 Canonical Ltd.
#
# Author: Scott Moser <scott.moser@canonical.com>
# Author: Christian Ehrhardt <christian.ehrhardt@canonical.com>
#
# This file is part of cloud-init. See LICENSE file for license information.
"""gpg.py - Collection of gpg key related functions"""
import logging
import os
import re
import signal
import time
from tempfile import TemporaryDirectory
from typing import Dict, Optional
from cloudinit import subp
LOG = logging.getLogger(__name__)
HOME = "GNUPGHOME"
class GPG:
def __init__(self):
self.gpg_started = False
self._env = {}
self.temp_dir = TemporaryDirectory()
def __enter__(self):
return self
@property
def env(self) -> Dict[str, str]:
"""when this env property gets invoked, set up our temporary
directory, and also set gpg_started to tell the cleanup()
method whether or not
why put this here and not in __init__? pytest seems unhappy
and it's not obvious how to work around it
"""
if self._env:
return self._env
self.gpg_started = True
self._env = {HOME: self.temp_dir.name}
return self._env
def __exit__(self, exc_typ, exc_value, traceback):
self.cleanup()
def cleanup(self) -> None:
"""cleanup the gpg temporary directory and kill gpg"""
self.kill_gpg()
if self.temp_dir and os.path.isdir(self.temp_dir.name):
self.temp_dir.cleanup()
def export_armour(self, key: str) -> Optional[str]:
"""Export gpg key, armoured key gets returned"""
try:
return subp.subp(
["gpg", "--export", "--armour", key],
capture=True,
update_env=self.env,
).stdout
except subp.ProcessExecutionError as error:
# debug, since it happens for any key not on the system initially
LOG.debug('Failed to export armoured key "%s": %s', key, error)
return None
def dearmor(self, key: str) -> str:
"""Dearmor gpg key, dearmored key gets returned
note: man gpg(1) makes no mention of an --armour spelling, only --armor
"""
return subp.subp(
["gpg", "--dearmor"], data=key, decode=False, update_env=self.env
).stdout
def list_keys(self, key_file: str, human_output=False) -> str:
"""List keys from a keyring with fingerprints. Default to a
stable machine parseable format.
@param key_file: a string containing a filepath to a key
@param human_output: return output intended for human parsing
"""
cmd = [
"gpg",
"--no-options",
"--with-fingerprint",
"--no-default-keyring",
"--list-keys",
"--keyring",
]
if not human_output:
cmd.append("--with-colons")
cmd.append(key_file)
stdout, stderr = subp.subp(cmd, update_env=self.env, capture=True)
if stderr:
LOG.warning(
'Failed to export armoured key "%s": %s', key_file, stderr
)
return stdout
def recv_key(self, key: str, keyserver: str, retries=(1, 1)) -> None:
"""Receive gpg key from the specified keyserver.
Retries are done by default because keyservers can be unreliable.
Additionally, there is no way to determine the difference between
a non-existent key and a failure. In both cases gpg (at least 2.2.4)
exits with status 2 and stderr: "keyserver receive failed: No data"
It is assumed that a key provided to cloud-init exists on the keyserver
so re-trying makes better sense than failing.
@param key: a string key fingerprint (as passed to gpg --recv-keys).
@param keyserver: the keyserver to request keys from.
@param retries: an iterable of sleep lengths for retries.
Use None to indicate no retries."""
LOG.debug("Importing key '%s' from keyserver '%s'", key, keyserver)
trynum = 0
error = None
sleeps = iter(retries or [])
while True:
trynum += 1
try:
subp.subp(
[
"gpg",
"--no-tty",
"--keyserver=%s" % keyserver,
"--recv-keys",
key,
],
capture=True,
update_env=self.env,
)
LOG.debug(
"Imported key '%s' from keyserver '%s' on try %d",
key,
keyserver,
trynum,
)
return
except subp.ProcessExecutionError as e:
error = e
try:
naplen = next(sleeps)
LOG.debug(
"Import failed with exit code %d, will try again in %ss",
error.exit_code,
naplen,
)
time.sleep(naplen)
except StopIteration as e:
raise ValueError(
"Failed to import key '%s' from keyserver '%s' "
"after %d tries: %s" % (key, keyserver, trynum, error)
) from e
def delete_key(self, key: str) -> None:
"""Delete the specified key from the local gpg ring"""
try:
subp.subp(
["gpg", "--batch", "--yes", "--delete-keys", key],
capture=True,
update_env=self.env,
)
except subp.ProcessExecutionError as error:
LOG.warning('Failed delete key "%s": %s', key, error)
def getkeybyid(
self, keyid: str, keyserver: str = "keyserver.ubuntu.com"
) -> Optional[str]:
"""get gpg keyid from keyserver"""
armour = self.export_armour(keyid)
if not armour:
try:
self.recv_key(keyid, keyserver=keyserver)
armour = self.export_armour(keyid)
except ValueError:
LOG.exception("Failed to obtain gpg key %s", keyid)
raise
finally:
# delete just imported key to leave environment as it
# was before
self.delete_key(keyid)
return armour
def kill_gpg(self) -> None:
"""killing with gpgconf is best practice, but when it isn't available
failover is possible
GH: 4344 - stop gpg-agent/dirmgr daemons spawned by gpg
key imports. Daemons spawned by cloud-config.service on systemd
v253 report (running)
"""
try:
if not self.gpg_started:
return
if subp.which("gpgconf"):
gpg_process_out = subp.subp(
["gpgconf", "--kill", "all"],
capture=True,
update_env=self.env,
).stdout
else:
gpg_process_out = subp.subp(
[
"ps",
"-o",
"ppid,pid",
"-C",
"keyboxd",
"-C",
"dirmngr",
"-C",
"gpg-agent",
],
capture=True,
rcs=[0, 1],
).stdout
gpg_pids = re.findall(
r"(?P<ppid>\d+)\s+(?P<pid>\d+)", gpg_process_out
)
root_gpg_pids = [
int(pid[1]) for pid in gpg_pids if pid[0] == "1"
]
if root_gpg_pids:
LOG.debug(
"Killing gpg-agent and dirmngr pids: %s", root_gpg_pids
)
for gpg_pid in root_gpg_pids:
os.kill(gpg_pid, signal.SIGKILL)
except subp.ProcessExecutionError as e:
LOG.warning("Failed to clean up gpg process: %s", e)
|
