Hide keyboard shortcuts

Hot-keys on this page

r m x p   toggle line displays

j k   next/prev highlighted chunk

0   (zero) top of page

1   (one) first highlighted chunk

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

467

468

469

470

471

472

473

474

475

476

477

478

479

480

481

482

483

484

485

486

487

488

489

490

491

492

493

494

495

496

"""Plugin common functions.""" 

import logging 

import os 

import re 

import shutil 

import tempfile 

 

import OpenSSL 

import pkg_resources 

import zope.interface 

 

from josepy import util as jose_util 

 

from acme.magic_typing import List # pylint: disable=unused-import, no-name-in-module 

from certbot import achallenges # pylint: disable=unused-import 

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 

 

from certbot.plugins.storage import PluginStorage 

 

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)) 

 

 

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.storage = PluginStorage(self.config, self.name) 

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 ChallengePerformer(object): 

"""Abstract base for challenge performers. 

 

:ivar configurator: Authenticator and installer plugin 

:ivar achalls: Annotated challenges 

:vartype achalls: `list` of `.KeyAuthorizationAnnotatedChallenge` 

:ivar indices: Holds the indices of challenges from a larger array 

so the user of the class doesn't have to. 

:vartype indices: `list` of `int` 

 

""" 

 

def __init__(self, configurator): 

self.configurator = configurator 

self.achalls = [] # type: List[achallenges.KeyAuthorizationAnnotatedChallenge] 

self.indices = [] # type: List[int] 

 

def add_chall(self, achall, idx=None): 

"""Store challenge to be performed when perform() is called. 

 

:param .KeyAuthorizationAnnotatedChallenge achall: Annotated 

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 perform(self): 

"""Perform all added challenges. 

 

:returns: challenge respones 

:rtype: `list` of `acme.challenges.KeyAuthorizationChallengeResponse` 

 

 

""" 

raise NotImplementedError() 

 

 

class TLSSNI01(ChallengePerformer): 

# pylint: disable=abstract-method 

"""Abstract base for TLS-SNI-01 challenge performers""" 

 

def __init__(self, configurator): 

super(TLSSNI01, self).__init__(configurator) 

self.challenge_conf = os.path.join( 

configurator.config.config_dir, "le_tls_sni_01_cert_challenge.conf") 

# self.completed = 0 

 

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