/
common.py
466 lines (363 loc) · 15.8 KB
/
common.py
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
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
"""Plugin common functions."""
import logging
import os
import re
import shutil
import tempfile
import OpenSSL
import pkg_resources
import zope.interface
from acme.jose import util as jose_util
from certbot import constants
from certbot import crypto_util
from certbot import errors
from certbot import interfaces
from certbot import reverter
from certbot import util
logger = logging.getLogger(__name__)
def option_namespace(name):
"""ArgumentParser options namespace (prefix of all options)."""
return name + "-"
def dest_namespace(name):
"""ArgumentParser dest namespace (prefix of all destinations)."""
return name.replace("-", "_") + "_"
private_ips_regex = re.compile(
r"(^127\.0\.0\.1)|(^10\.)|(^172\.1[6-9]\.)|"
r"(^172\.2[0-9]\.)|(^172\.3[0-1]\.)|(^192\.168\.)")
hostname_regex = re.compile(
r"^(([a-z0-9]|[a-z0-9][a-z0-9\-]*[a-z0-9])\.)*[a-z]+$", re.IGNORECASE)
@zope.interface.implementer(interfaces.IPlugin)
class Plugin(object):
"""Generic plugin."""
# provider is not inherited, subclasses must define it on their own
# @zope.interface.provider(interfaces.IPluginFactory)
def __init__(self, config, name):
self.config = config
self.name = name
@jose_util.abstractclassmethod
def add_parser_arguments(cls, add):
"""Add plugin arguments to the CLI argument parser.
NOTE: If some of your flags interact with others, you can
use cli.report_config_interaction to register this to ensure
values are correctly saved/overridable during renewal.
:param callable add: Function that proxies calls to
`argparse.ArgumentParser.add_argument` prepending options
with unique plugin name prefix.
"""
@classmethod
def inject_parser_options(cls, parser, name):
"""Inject parser options.
See `~.IPlugin.inject_parser_options` for docs.
"""
# dummy function, doesn't check if dest.startswith(self.dest_namespace)
def add(arg_name_no_prefix, *args, **kwargs):
# pylint: disable=missing-docstring
return parser.add_argument(
"--{0}{1}".format(option_namespace(name), arg_name_no_prefix),
*args, **kwargs)
return cls.add_parser_arguments(add)
@property
def option_namespace(self):
"""ArgumentParser options namespace (prefix of all options)."""
return option_namespace(self.name)
def option_name(self, name):
"""Option name (include plugin namespace)."""
return self.option_namespace + name
@property
def dest_namespace(self):
"""ArgumentParser dest namespace (prefix of all destinations)."""
return dest_namespace(self.name)
def dest(self, var):
"""Find a destination for given variable ``var``."""
# this should do exactly the same what ArgumentParser(arg),
# does to "arg" to compute "dest"
return self.dest_namespace + var.replace("-", "_")
def conf(self, var):
"""Find a configuration value for variable ``var``."""
return getattr(self.config, self.dest(var))
# other
class Installer(Plugin):
"""An installer base class with reverter and ssl_dhparam methods defined.
Installer plugins do not have to inherit from this class.
"""
def __init__(self, *args, **kwargs):
super(Installer, self).__init__(*args, **kwargs)
self.reverter = reverter.Reverter(self.config)
def add_to_checkpoint(self, save_files, save_notes, temporary=False):
"""Add files to a checkpoint.
:param set save_files: set of filepaths to save
:param str save_notes: notes about changes during the save
:param bool temporary: True if the files should be added to a
temporary checkpoint rather than a permanent one. This is
usually used for changes that will soon be reverted.
:raises .errors.PluginError: when unable to add to checkpoint
"""
if temporary:
checkpoint_func = self.reverter.add_to_temp_checkpoint
else:
checkpoint_func = self.reverter.add_to_checkpoint
try:
checkpoint_func(save_files, save_notes)
except errors.ReverterError as err:
raise errors.PluginError(str(err))
def finalize_checkpoint(self, title):
"""Timestamp and save changes made through the reverter.
:param str title: Title describing checkpoint
:raises .errors.PluginError: when an error occurs
"""
try:
self.reverter.finalize_checkpoint(title)
except errors.ReverterError as err:
raise errors.PluginError(str(err))
def recovery_routine(self):
"""Revert all previously modified files.
Reverts all modified files that have not been saved as a checkpoint
:raises .errors.PluginError: If unable to recover the configuration
"""
try:
self.reverter.recovery_routine()
except errors.ReverterError as err:
raise errors.PluginError(str(err))
def revert_temporary_config(self):
"""Rollback temporary checkpoint.
:raises .errors.PluginError: when unable to revert config
"""
try:
self.reverter.revert_temporary_config()
except errors.ReverterError as err:
raise errors.PluginError(str(err))
def rollback_checkpoints(self, rollback=1):
"""Rollback saved checkpoints.
:param int rollback: Number of checkpoints to revert
:raises .errors.PluginError: If there is a problem with the input or
the function is unable to correctly revert the configuration
"""
try:
self.reverter.rollback_checkpoints(rollback)
except errors.ReverterError as err:
raise errors.PluginError(str(err))
def view_config_changes(self):
"""Show all of the configuration changes that have taken place.
:raises .errors.PluginError: If there is a problem while processing
the checkpoints directories.
"""
try:
self.reverter.view_config_changes()
except errors.ReverterError as err:
raise errors.PluginError(str(err))
@property
def ssl_dhparams(self):
"""Full absolute path to ssl_dhparams file."""
return os.path.join(self.config.config_dir, constants.SSL_DHPARAMS_DEST)
@property
def updated_ssl_dhparams_digest(self):
"""Full absolute path to digest of updated ssl_dhparams file."""
return os.path.join(self.config.config_dir, constants.UPDATED_SSL_DHPARAMS_DIGEST)
def install_ssl_dhparams(self):
"""Copy Certbot's ssl_dhparams file into the system's config dir if required."""
return install_version_controlled_file(
self.ssl_dhparams,
self.updated_ssl_dhparams_digest,
constants.SSL_DHPARAMS_SRC,
constants.ALL_SSL_DHPARAMS_HASHES)
class Addr(object):
r"""Represents an virtual host address.
:param str addr: addr part of vhost address
:param str port: port number or \*, or ""
"""
def __init__(self, tup, ipv6=False):
self.tup = tup
self.ipv6 = ipv6
@classmethod
def fromstring(cls, str_addr):
"""Initialize Addr from string."""
if str_addr.startswith('['):
# ipv6 addresses starts with [
endIndex = str_addr.rfind(']')
host = str_addr[:endIndex + 1]
port = ''
if len(str_addr) > endIndex + 2 and str_addr[endIndex + 1] == ':':
port = str_addr[endIndex + 2:]
return cls((host, port), ipv6=True)
else:
tup = str_addr.partition(':')
return cls((tup[0], tup[2]))
def __str__(self):
if self.tup[1]:
return "%s:%s" % self.tup
return self.tup[0]
def normalized_tuple(self):
"""Normalized representation of addr/port tuple
"""
if self.ipv6:
return (self.get_ipv6_exploded(), self.tup[1])
return self.tup
def __eq__(self, other):
if isinstance(other, self.__class__):
# compare normalized to take different
# styles of representation into account
return self.normalized_tuple() == other.normalized_tuple()
return False
def __hash__(self):
return hash(self.tup)
def get_addr(self):
"""Return addr part of Addr object."""
return self.tup[0]
def get_port(self):
"""Return port."""
return self.tup[1]
def get_addr_obj(self, port):
"""Return new address object with same addr and new port."""
return self.__class__((self.tup[0], port), self.ipv6)
def _normalize_ipv6(self, addr):
"""Return IPv6 address in normalized form, helper function"""
addr = addr.lstrip("[")
addr = addr.rstrip("]")
return self._explode_ipv6(addr)
def get_ipv6_exploded(self):
"""Return IPv6 in normalized form"""
if self.ipv6:
return ":".join(self._normalize_ipv6(self.tup[0]))
return ""
def _explode_ipv6(self, addr):
"""Explode IPv6 address for comparison"""
result = ['0', '0', '0', '0', '0', '0', '0', '0']
addr_list = addr.split(":")
if len(addr_list) > len(result):
# too long, truncate
addr_list = addr_list[0:len(result)]
append_to_end = False
for i in range(0, len(addr_list)):
block = addr_list[i]
if len(block) == 0:
# encountered ::, so rest of the blocks should be
# appended to the end
append_to_end = True
continue
elif len(block) > 1:
# remove leading zeros
block = block.lstrip("0")
if not append_to_end:
result[i] = str(block)
else:
# count the location from the end using negative indices
result[i-len(addr_list)] = str(block)
return result
class TLSSNI01(object):
"""Abstract base for TLS-SNI-01 challenge performers"""
def __init__(self, configurator):
self.configurator = configurator
self.achalls = []
self.indices = []
self.challenge_conf = os.path.join(
configurator.config.config_dir, "le_tls_sni_01_cert_challenge.conf")
# self.completed = 0
def add_chall(self, achall, idx=None):
"""Add challenge to TLSSNI01 object to perform at once.
:param .KeyAuthorizationAnnotatedChallenge achall: Annotated
TLSSNI01 challenge.
:param int idx: index to challenge in a larger array
"""
self.achalls.append(achall)
if idx is not None:
self.indices.append(idx)
def get_cert_path(self, achall):
"""Returns standardized name for challenge certificate.
:param .KeyAuthorizationAnnotatedChallenge achall: Annotated
tls-sni-01 challenge.
:returns: certificate file name
:rtype: str
"""
return os.path.join(self.configurator.config.work_dir,
achall.chall.encode("token") + ".crt")
def get_key_path(self, achall):
"""Get standardized path to challenge key."""
return os.path.join(self.configurator.config.work_dir,
achall.chall.encode("token") + '.pem')
def get_z_domain(self, achall):
"""Returns z_domain (SNI) name for the challenge."""
return achall.response(achall.account_key).z_domain.decode("utf-8")
def _setup_challenge_cert(self, achall, cert_key=None):
"""Generate and write out challenge certificate."""
cert_path = self.get_cert_path(achall)
key_path = self.get_key_path(achall)
# Register the path before you write out the file
self.configurator.reverter.register_file_creation(True, key_path)
self.configurator.reverter.register_file_creation(True, cert_path)
response, (cert, key) = achall.response_and_validation(
cert_key=cert_key)
cert_pem = OpenSSL.crypto.dump_certificate(
OpenSSL.crypto.FILETYPE_PEM, cert)
key_pem = OpenSSL.crypto.dump_privatekey(
OpenSSL.crypto.FILETYPE_PEM, key)
# Write out challenge cert and key
with open(cert_path, "wb") as cert_chall_fd:
cert_chall_fd.write(cert_pem)
with util.safe_open(key_path, 'wb', chmod=0o400) as key_file:
key_file.write(key_pem)
return response
def install_version_controlled_file(dest_path, digest_path, src_path, all_hashes):
"""Copy a file into an active location (likely the system's config dir) if required.
:param str dest_path: destination path for version controlled file
:param str digest_path: path to save a digest of the file in
:param str src_path: path to version controlled file found in distribution
:param list all_hashes: hashes of every released version of the file
"""
current_hash = crypto_util.sha256sum(src_path)
def _write_current_hash():
with open(digest_path, "w") as f:
f.write(current_hash)
def _install_current_file():
shutil.copyfile(src_path, dest_path)
_write_current_hash()
# Check to make sure options-ssl.conf is installed
if not os.path.isfile(dest_path):
_install_current_file()
return
# there's already a file there. if it's up to date, do nothing. if it's not but
# it matches a known file hash, we can update it.
# otherwise, print a warning once per new version.
active_file_digest = crypto_util.sha256sum(dest_path)
if active_file_digest == current_hash: # already up to date
return
elif active_file_digest in all_hashes: # safe to update
_install_current_file()
else: # has been manually modified, not safe to update
# did they modify the current version or an old version?
if os.path.isfile(digest_path):
with open(digest_path, "r") as f:
saved_digest = f.read()
# they modified it after we either installed or told them about this version, so return
if saved_digest == current_hash:
return
# there's a new version but we couldn't update the file, or they deleted the digest.
# save the current digest so we only print this once, and print a warning
_write_current_hash()
logger.warning("%s has been manually modified; updated file "
"saved to %s. We recommend updating %s for security purposes.",
dest_path, src_path, dest_path)
# test utils used by certbot_apache/certbot_nginx (hence
# "pragma: no cover") TODO: this might quickly lead to dead code (also
# c.f. #383)
def dir_setup(test_dir, pkg): # pragma: no cover
"""Setup the directories necessary for the configurator."""
def expanded_tempdir(prefix):
"""Return the real path of a temp directory with the specified prefix
Some plugins rely on real paths of symlinks for working correctly. For
example, certbot-apache uses real paths of configuration files to tell
a virtual host from another. On systems where TMP itself is a symbolic
link, (ex: OS X) such plugins will be confused. This function prevents
such a case.
"""
return os.path.realpath(tempfile.mkdtemp(prefix))
temp_dir = expanded_tempdir("temp")
config_dir = expanded_tempdir("config")
work_dir = expanded_tempdir("work")
os.chmod(temp_dir, constants.CONFIG_DIRS_MODE)
os.chmod(config_dir, constants.CONFIG_DIRS_MODE)
os.chmod(work_dir, constants.CONFIG_DIRS_MODE)
test_configs = pkg_resources.resource_filename(
pkg, os.path.join("testdata", test_dir))
shutil.copytree(
test_configs, os.path.join(temp_dir, test_dir), symlinks=True)
return temp_dir, config_dir, work_dir