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

"""Certbot client crypto utility functions. 

 

.. todo:: Make the transition to use PSS rather than PKCS1_v1_5 when the server 

is capable of handling the signatures. 

 

""" 

import hashlib 

import logging 

import os 

import warnings 

 

import pyrfc3339 

import six 

import zope.component 

from cryptography.exceptions import InvalidSignature 

from cryptography.hazmat.backends import default_backend 

from cryptography.hazmat.primitives.asymmetric.ec import ECDSA 

from cryptography.hazmat.primitives.asymmetric.ec import EllipticCurvePublicKey 

from cryptography.hazmat.primitives.asymmetric.padding import PKCS1v15 

from cryptography.hazmat.primitives.asymmetric.rsa import RSAPublicKey 

# https://github.com/python/typeshed/tree/master/third_party/2/cryptography 

from cryptography import x509 # type: ignore 

from OpenSSL import crypto 

from OpenSSL import SSL # type: ignore 

 

from acme import crypto_util as acme_crypto_util 

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

from certbot import compat 

from certbot import errors 

from certbot import interfaces 

from certbot import util 

 

 

logger = logging.getLogger(__name__) 

 

 

# High level functions 

def init_save_key(key_size, key_dir, keyname="key-certbot.pem"): 

"""Initializes and saves a privkey. 

 

Inits key and saves it in PEM format on the filesystem. 

 

.. note:: keyname is the attempted filename, it may be different if a file 

already exists at the path. 

 

:param int key_size: RSA key size in bits 

:param str key_dir: Key save directory. 

:param str keyname: Filename of key 

 

:returns: Key 

:rtype: :class:`certbot.util.Key` 

 

:raises ValueError: If unable to generate the key given key_size. 

 

""" 

try: 

key_pem = make_key(key_size) 

except ValueError as err: 

logger.error("", exc_info=True) 

raise err 

 

config = zope.component.getUtility(interfaces.IConfig) 

# Save file 

util.make_or_verify_dir(key_dir, 0o700, compat.os_geteuid(), 

config.strict_permissions) 

key_f, key_path = util.unique_file( 

os.path.join(key_dir, keyname), 0o600, "wb") 

with key_f: 

key_f.write(key_pem) 

logger.debug("Generating key (%d bits): %s", key_size, key_path) 

 

return util.Key(key_path, key_pem) 

 

 

def init_save_csr(privkey, names, path): 

"""Initialize a CSR with the given private key. 

 

:param privkey: Key to include in the CSR 

:type privkey: :class:`certbot.util.Key` 

 

:param set names: `str` names to include in the CSR 

 

:param str path: Certificate save directory. 

 

:returns: CSR 

:rtype: :class:`certbot.util.CSR` 

 

""" 

config = zope.component.getUtility(interfaces.IConfig) 

 

csr_pem = acme_crypto_util.make_csr( 

privkey.pem, names, must_staple=config.must_staple) 

 

# Save CSR 

util.make_or_verify_dir(path, 0o755, compat.os_geteuid(), 

config.strict_permissions) 

csr_f, csr_filename = util.unique_file( 

os.path.join(path, "csr-certbot.pem"), 0o644, "wb") 

with csr_f: 

csr_f.write(csr_pem) 

logger.debug("Creating CSR: %s", csr_filename) 

 

return util.CSR(csr_filename, csr_pem, "pem") 

 

 

# WARNING: the csr and private key file are possible attack vectors for TOCTOU 

# We should either... 

# A. Do more checks to verify that the CSR is trusted/valid 

# B. Audit the parsing code for vulnerabilities 

 

def valid_csr(csr): 

"""Validate CSR. 

 

Check if `csr` is a valid CSR for the given domains. 

 

:param str csr: CSR in PEM. 

 

:returns: Validity of CSR. 

:rtype: bool 

 

""" 

try: 

req = crypto.load_certificate_request( 

crypto.FILETYPE_PEM, csr) 

return req.verify(req.get_pubkey()) 

except crypto.Error: 

logger.debug("", exc_info=True) 

return False 

 

 

def csr_matches_pubkey(csr, privkey): 

"""Does private key correspond to the subject public key in the CSR? 

 

:param str csr: CSR in PEM. 

:param str privkey: Private key file contents (PEM) 

 

:returns: Correspondence of private key to CSR subject public key. 

:rtype: bool 

 

""" 

req = crypto.load_certificate_request( 

crypto.FILETYPE_PEM, csr) 

pkey = crypto.load_privatekey(crypto.FILETYPE_PEM, privkey) 

try: 

return req.verify(pkey) 

except crypto.Error: 

logger.debug("", exc_info=True) 

return False 

 

 

def import_csr_file(csrfile, data): 

"""Import a CSR file, which can be either PEM or DER. 

 

:param str csrfile: CSR filename 

:param str data: contents of the CSR file 

 

:returns: (`crypto.FILETYPE_PEM`, 

util.CSR object representing the CSR, 

list of domains requested in the CSR) 

:rtype: tuple 

 

""" 

PEM = crypto.FILETYPE_PEM 

load = crypto.load_certificate_request 

try: 

# Try to parse as DER first, then fall back to PEM. 

csr = load(crypto.FILETYPE_ASN1, data) 

except crypto.Error: 

try: 

csr = load(PEM, data) 

except crypto.Error: 

raise errors.Error("Failed to parse CSR file: {0}".format(csrfile)) 

 

domains = _get_names_from_loaded_cert_or_req(csr) 

# Internally we always use PEM, so re-encode as PEM before returning. 

data_pem = crypto.dump_certificate_request(PEM, csr) 

return PEM, util.CSR(file=csrfile, data=data_pem, form="pem"), domains 

 

 

def make_key(bits): 

"""Generate PEM encoded RSA key. 

 

:param int bits: Number of bits, at least 1024. 

 

:returns: new RSA key in PEM form with specified number of bits 

:rtype: str 

 

""" 

assert bits >= 1024 # XXX 

key = crypto.PKey() 

key.generate_key(crypto.TYPE_RSA, bits) 

return crypto.dump_privatekey(crypto.FILETYPE_PEM, key) 

 

 

def valid_privkey(privkey): 

"""Is valid RSA private key? 

 

:param str privkey: Private key file contents in PEM 

 

:returns: Validity of private key. 

:rtype: bool 

 

""" 

try: 

return crypto.load_privatekey( 

crypto.FILETYPE_PEM, privkey).check() 

except (TypeError, crypto.Error): 

return False 

 

 

def verify_renewable_cert(renewable_cert): 

"""For checking that your certs were not corrupted on disk. 

 

Several things are checked: 

1. Signature verification for the cert. 

2. That fullchain matches cert and chain when concatenated. 

3. Check that the private key matches the certificate. 

 

:param `.storage.RenewableCert` renewable_cert: cert to verify 

 

:raises errors.Error: If verification fails. 

""" 

verify_renewable_cert_sig(renewable_cert) 

verify_fullchain(renewable_cert) 

verify_cert_matches_priv_key(renewable_cert.cert, renewable_cert.privkey) 

 

 

def verify_renewable_cert_sig(renewable_cert): 

""" Verifies the signature of a `.storage.RenewableCert` object. 

 

:param `.storage.RenewableCert` renewable_cert: cert to verify 

 

:raises errors.Error: If signature verification fails. 

""" 

try: 

with open(renewable_cert.chain, 'rb') as chain_file: # type: IO[bytes] 

chain = x509.load_pem_x509_certificate(chain_file.read(), default_backend()) 

with open(renewable_cert.cert, 'rb') as cert_file: # type: IO[bytes] 

cert = x509.load_pem_x509_certificate(cert_file.read(), default_backend()) 

pk = chain.public_key() 

with warnings.catch_warnings(): 

warnings.simplefilter("ignore") 

if isinstance(pk, RSAPublicKey): 

# https://github.com/python/typeshed/blob/master/third_party/2/cryptography/hazmat/primitives/asymmetric/rsa.pyi 

verifier = pk.verifier( # type: ignore 

cert.signature, PKCS1v15(), cert.signature_hash_algorithm 

) 

verifier.update(cert.tbs_certificate_bytes) 

verifier.verify() 

elif isinstance(pk, EllipticCurvePublicKey): 

verifier = pk.verifier( 

cert.signature, ECDSA(cert.signature_hash_algorithm) 

) 

verifier.update(cert.tbs_certificate_bytes) 

verifier.verify() 

else: 

raise errors.Error("Unsupported public key type") 

except (IOError, ValueError, InvalidSignature) as e: 

error_str = "verifying the signature of the cert located at {0} has failed. \ 

Details: {1}".format(renewable_cert.cert, e) 

logger.exception(error_str) 

raise errors.Error(error_str) 

 

 

def verify_cert_matches_priv_key(cert_path, key_path): 

""" Verifies that the private key and cert match. 

 

:param str cert_path: path to a cert in PEM format 

:param str key_path: path to a private key file 

 

:raises errors.Error: If they don't match. 

""" 

try: 

context = SSL.Context(SSL.SSLv23_METHOD) 

context.use_certificate_file(cert_path) 

context.use_privatekey_file(key_path) 

context.check_privatekey() 

except (IOError, SSL.Error) as e: 

error_str = "verifying the cert located at {0} matches the \ 

private key located at {1} has failed. \ 

Details: {2}".format(cert_path, 

key_path, e) 

logger.exception(error_str) 

raise errors.Error(error_str) 

 

 

def verify_fullchain(renewable_cert): 

""" Verifies that fullchain is indeed cert concatenated with chain. 

 

:param `.storage.RenewableCert` renewable_cert: cert to verify 

 

:raises errors.Error: If cert and chain do not combine to fullchain. 

""" 

try: 

with open(renewable_cert.chain) as chain_file: # type: IO[str] 

chain = chain_file.read() 

with open(renewable_cert.cert) as cert_file: # type: IO[str] 

cert = cert_file.read() 

with open(renewable_cert.fullchain) as fullchain_file: # type: IO[str] 

fullchain = fullchain_file.read() 

if (cert + chain) != fullchain: 

error_str = "fullchain does not match cert + chain for {0}!" 

error_str = error_str.format(renewable_cert.lineagename) 

raise errors.Error(error_str) 

except IOError as e: 

error_str = "reading one of cert, chain, or fullchain has failed: {0}".format(e) 

logger.exception(error_str) 

raise errors.Error(error_str) 

except errors.Error as e: 

raise e 

 

 

def pyopenssl_load_certificate(data): 

"""Load PEM/DER certificate. 

 

:raises errors.Error: 

 

""" 

 

openssl_errors = [] 

 

for file_type in (crypto.FILETYPE_PEM, crypto.FILETYPE_ASN1): 

try: 

return crypto.load_certificate(file_type, data), file_type 

except crypto.Error as error: # TODO: other errors? 

openssl_errors.append(error) 

raise errors.Error("Unable to load: {0}".format(",".join( 

str(error) for error in openssl_errors))) 

 

 

def _load_cert_or_req(cert_or_req_str, load_func, 

typ=crypto.FILETYPE_PEM): 

try: 

return load_func(typ, cert_or_req_str) 

except crypto.Error: 

logger.error("", exc_info=True) 

raise 

 

 

def _get_sans_from_cert_or_req(cert_or_req_str, load_func, 

typ=crypto.FILETYPE_PEM): 

# pylint: disable=protected-access 

return acme_crypto_util._pyopenssl_cert_or_req_san(_load_cert_or_req( 

cert_or_req_str, load_func, typ)) 

 

 

def get_sans_from_cert(cert, typ=crypto.FILETYPE_PEM): 

"""Get a list of Subject Alternative Names from a certificate. 

 

:param str cert: Certificate (encoded). 

:param typ: `crypto.FILETYPE_PEM` or `crypto.FILETYPE_ASN1` 

 

:returns: A list of Subject Alternative Names. 

:rtype: list 

 

""" 

return _get_sans_from_cert_or_req( 

cert, crypto.load_certificate, typ) 

 

 

def _get_names_from_cert_or_req(cert_or_req, load_func, typ): 

loaded_cert_or_req = _load_cert_or_req(cert_or_req, load_func, typ) 

return _get_names_from_loaded_cert_or_req(loaded_cert_or_req) 

 

 

def _get_names_from_loaded_cert_or_req(loaded_cert_or_req): 

# pylint: disable=protected-access 

return acme_crypto_util._pyopenssl_cert_or_req_all_names(loaded_cert_or_req) 

 

 

def get_names_from_cert(csr, typ=crypto.FILETYPE_PEM): 

"""Get a list of domains from a cert, including the CN if it is set. 

 

:param str cert: Certificate (encoded). 

:param typ: `crypto.FILETYPE_PEM` or `crypto.FILETYPE_ASN1` 

 

:returns: A list of domain names. 

:rtype: list 

 

""" 

return _get_names_from_cert_or_req( 

csr, crypto.load_certificate, typ) 

 

 

def dump_pyopenssl_chain(chain, filetype=crypto.FILETYPE_PEM): 

"""Dump certificate chain into a bundle. 

 

:param list chain: List of `crypto.X509` (or wrapped in 

:class:`josepy.util.ComparableX509`). 

 

""" 

# XXX: returns empty string when no chain is available, which 

# shuts up RenewableCert, but might not be the best solution... 

return acme_crypto_util.dump_pyopenssl_chain(chain, filetype) 

 

 

def notBefore(cert_path): 

"""When does the cert at cert_path start being valid? 

 

:param str cert_path: path to a cert in PEM format 

 

:returns: the notBefore value from the cert at cert_path 

:rtype: :class:`datetime.datetime` 

 

""" 

return _notAfterBefore(cert_path, crypto.X509.get_notBefore) 

 

 

def notAfter(cert_path): 

"""When does the cert at cert_path stop being valid? 

 

:param str cert_path: path to a cert in PEM format 

 

:returns: the notAfter value from the cert at cert_path 

:rtype: :class:`datetime.datetime` 

 

""" 

return _notAfterBefore(cert_path, crypto.X509.get_notAfter) 

 

 

def _notAfterBefore(cert_path, method): 

"""Internal helper function for finding notbefore/notafter. 

 

:param str cert_path: path to a cert in PEM format 

:param function method: one of ``crypto.X509.get_notBefore`` 

or ``crypto.X509.get_notAfter`` 

 

:returns: the notBefore or notAfter value from the cert at cert_path 

:rtype: :class:`datetime.datetime` 

 

""" 

# pylint: disable=redefined-outer-name 

with open(cert_path) as f: 

x509 = crypto.load_certificate(crypto.FILETYPE_PEM, 

f.read()) 

# pyopenssl always returns bytes 

timestamp = method(x509) 

reformatted_timestamp = [timestamp[0:4], b"-", timestamp[4:6], b"-", 

timestamp[6:8], b"T", timestamp[8:10], b":", 

timestamp[10:12], b":", timestamp[12:]] 

timestamp_str = b"".join(reformatted_timestamp) 

# pyrfc3339 uses "native" strings. That is, bytes on Python 2 and unicode 

# on Python 3 

if six.PY3: 

timestamp_str = timestamp_str.decode('ascii') 

return pyrfc3339.parse(timestamp_str) 

 

 

def sha256sum(filename): 

"""Compute a sha256sum of a file. 

 

NB: In given file, platform specific newlines characters will be converted 

into their equivalent unicode counterparts before calculating the hash. 

 

:param str filename: path to the file whose hash will be computed 

 

:returns: sha256 digest of the file in hexadecimal 

:rtype: str 

""" 

sha256 = hashlib.sha256() 

with open(filename, 'r') as file_d: 

sha256.update(file_d.read().encode('UTF-8')) 

return sha256.hexdigest() 

 

def cert_and_chain_from_fullchain(fullchain_pem): 

"""Split fullchain_pem into cert_pem and chain_pem 

 

:param str fullchain_pem: concatenated cert + chain 

 

:returns: tuple of string cert_pem and chain_pem 

:rtype: tuple 

 

""" 

cert = crypto.dump_certificate(crypto.FILETYPE_PEM, 

crypto.load_certificate(crypto.FILETYPE_PEM, fullchain_pem)).decode() 

chain = fullchain_pem[len(cert):].lstrip() 

return (cert, chain)