Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/OpenSSL/crypto.py: 34%

Shortcuts on this page

r m x   toggle line displays

j k   next/prev highlighted chunk

0   (zero) top of page

1   (one) first highlighted chunk

695 statements  

1from __future__ import annotations 

2 

3import calendar 

4import datetime 

5import functools 

6import sys 

7import typing 

8from base64 import b16encode 

9from collections.abc import Sequence 

10from functools import partial 

11from typing import ( 

12 Any, 

13 Callable, 

14 Union, 

15) 

16 

17if sys.version_info >= (3, 13): 

18 from warnings import deprecated 

19else: 

20 from typing_extensions import deprecated 

21 

22from cryptography import utils, x509 

23from cryptography.hazmat.primitives.asymmetric import ( 

24 dsa, 

25 ec, 

26 ed448, 

27 ed25519, 

28 rsa, 

29) 

30 

31from OpenSSL._util import StrOrBytesPath 

32from OpenSSL._util import ( 

33 byte_string as _byte_string, 

34) 

35from OpenSSL._util import ( 

36 exception_from_error_queue as _exception_from_error_queue, 

37) 

38from OpenSSL._util import ( 

39 ffi as _ffi, 

40) 

41from OpenSSL._util import ( 

42 lib as _lib, 

43) 

44from OpenSSL._util import ( 

45 make_assert as _make_assert, 

46) 

47from OpenSSL._util import ( 

48 path_bytes as _path_bytes, 

49) 

50 

51__all__ = [ 

52 "FILETYPE_ASN1", 

53 "FILETYPE_PEM", 

54 "FILETYPE_TEXT", 

55 "TYPE_DSA", 

56 "TYPE_RSA", 

57 "X509", 

58 "Error", 

59 "PKey", 

60 "X509Name", 

61 "X509Store", 

62 "X509StoreContext", 

63 "X509StoreContextError", 

64 "X509StoreFlags", 

65 "dump_certificate", 

66 "dump_privatekey", 

67 "dump_publickey", 

68 "get_elliptic_curve", 

69 "get_elliptic_curves", 

70 "load_certificate", 

71 "load_privatekey", 

72 "load_publickey", 

73] 

74 

75 

76_PrivateKey = Union[ 

77 dsa.DSAPrivateKey, 

78 ec.EllipticCurvePrivateKey, 

79 ed25519.Ed25519PrivateKey, 

80 ed448.Ed448PrivateKey, 

81 rsa.RSAPrivateKey, 

82] 

83_PublicKey = Union[ 

84 dsa.DSAPublicKey, 

85 ec.EllipticCurvePublicKey, 

86 ed25519.Ed25519PublicKey, 

87 ed448.Ed448PublicKey, 

88 rsa.RSAPublicKey, 

89] 

90_Key = Union[_PrivateKey, _PublicKey] 

91PassphraseCallableT = Union[bytes, Callable[..., bytes]] 

92 

93 

94FILETYPE_PEM: int = _lib.SSL_FILETYPE_PEM 

95FILETYPE_ASN1: int = _lib.SSL_FILETYPE_ASN1 

96 

97# TODO This was an API mistake. OpenSSL has no such constant. 

98FILETYPE_TEXT = 2**16 - 1 

99 

100TYPE_RSA: int = _lib.EVP_PKEY_RSA 

101TYPE_DSA: int = _lib.EVP_PKEY_DSA 

102TYPE_DH: int = _lib.EVP_PKEY_DH 

103TYPE_EC: int = _lib.EVP_PKEY_EC 

104 

105 

106class Error(Exception): 

107 """ 

108 An error occurred in an `OpenSSL.crypto` API. 

109 """ 

110 

111 

112_raise_current_error = partial(_exception_from_error_queue, Error) 

113_openssl_assert = _make_assert(Error) 

114 

115 

116def _new_mem_buf(buffer: bytes | None = None) -> Any: 

117 """ 

118 Allocate a new OpenSSL memory BIO. 

119 

120 Arrange for the garbage collector to clean it up automatically. 

121 

122 :param buffer: None or some bytes to use to put into the BIO so that they 

123 can be read out. 

124 """ 

125 if buffer is None: 

126 bio = _lib.BIO_new(_lib.BIO_s_mem()) 

127 free = _lib.BIO_free 

128 else: 

129 data = _ffi.new("char[]", buffer) 

130 bio = _lib.BIO_new_mem_buf(data, len(buffer)) 

131 

132 # Keep the memory alive as long as the bio is alive! 

133 def free(bio: Any, ref: Any = data) -> Any: 

134 return _lib.BIO_free(bio) 

135 

136 _openssl_assert(bio != _ffi.NULL) 

137 

138 bio = _ffi.gc(bio, free) 

139 return bio 

140 

141 

142def _bio_to_string(bio: Any) -> bytes: 

143 """ 

144 Copy the contents of an OpenSSL BIO object into a Python byte string. 

145 """ 

146 result_buffer = _ffi.new("char**") 

147 buffer_length = _lib.BIO_get_mem_data(bio, result_buffer) 

148 return _ffi.buffer(result_buffer[0], buffer_length)[:] 

149 

150 

151def _set_asn1_time(boundary: Any, when: bytes) -> None: 

152 """ 

153 The the time value of an ASN1 time object. 

154 

155 @param boundary: An ASN1_TIME pointer (or an object safely 

156 castable to that type) which will have its value set. 

157 @param when: A string representation of the desired time value. 

158 

159 @raise TypeError: If C{when} is not a L{bytes} string. 

160 @raise ValueError: If C{when} does not represent a time in the required 

161 format. 

162 @raise RuntimeError: If the time value cannot be set for some other 

163 (unspecified) reason. 

164 """ 

165 if not isinstance(when, bytes): 

166 raise TypeError("when must be a byte string") 

167 # ASN1_TIME_set_string validates the string without writing anything 

168 # when the destination is NULL. 

169 _openssl_assert(boundary != _ffi.NULL) 

170 

171 set_result = _lib.ASN1_TIME_set_string(boundary, when) 

172 if set_result == 0: 

173 raise ValueError("Invalid string") 

174 

175 

176def _new_asn1_time(when: bytes) -> Any: 

177 """ 

178 Behaves like _set_asn1_time but returns a new ASN1_TIME object. 

179 

180 @param when: A string representation of the desired time value. 

181 

182 @raise TypeError: If C{when} is not a L{bytes} string. 

183 @raise ValueError: If C{when} does not represent a time in the required 

184 format. 

185 @raise RuntimeError: If the time value cannot be set for some other 

186 (unspecified) reason. 

187 """ 

188 ret = _lib.ASN1_TIME_new() 

189 _openssl_assert(ret != _ffi.NULL) 

190 ret = _ffi.gc(ret, _lib.ASN1_TIME_free) 

191 _set_asn1_time(ret, when) 

192 return ret 

193 

194 

195def _get_asn1_time(timestamp: Any) -> bytes | None: 

196 """ 

197 Retrieve the time value of an ASN1 time object. 

198 

199 @param timestamp: An ASN1_GENERALIZEDTIME* (or an object safely castable to 

200 that type) from which the time value will be retrieved. 

201 

202 @return: The time value from C{timestamp} as a L{bytes} string in a certain 

203 format. Or C{None} if the object contains no time value. 

204 """ 

205 string_timestamp = _ffi.cast("ASN1_STRING*", timestamp) 

206 if _lib.ASN1_STRING_length(string_timestamp) == 0: 

207 return None 

208 elif ( 

209 _lib.ASN1_STRING_type(string_timestamp) == _lib.V_ASN1_GENERALIZEDTIME 

210 ): 

211 return _ffi.string(_lib.ASN1_STRING_get0_data(string_timestamp)) 

212 else: 

213 generalized_timestamp = _ffi.new("ASN1_GENERALIZEDTIME**") 

214 _lib.ASN1_TIME_to_generalizedtime(timestamp, generalized_timestamp) 

215 _openssl_assert(generalized_timestamp[0] != _ffi.NULL) 

216 

217 string_timestamp = _ffi.cast("ASN1_STRING*", generalized_timestamp[0]) 

218 string_data = _lib.ASN1_STRING_get0_data(string_timestamp) 

219 string_result = _ffi.string(string_data) 

220 _lib.ASN1_GENERALIZEDTIME_free(generalized_timestamp[0]) 

221 return string_result 

222 

223 

224class _X509NameInvalidator: 

225 def __init__(self) -> None: 

226 self._names: list[X509Name] = [] 

227 

228 def add(self, name: X509Name) -> None: 

229 self._names.append(name) 

230 

231 def clear(self) -> None: 

232 for name in self._names: 

233 # Breaks the object, but also prevents UAF! 

234 del name._name 

235 

236 

237class PKey: 

238 """ 

239 A class representing an DSA or RSA public key or key pair. 

240 """ 

241 

242 _only_public = False 

243 _initialized = True 

244 

245 def __init__(self) -> None: 

246 pkey = _lib.EVP_PKEY_new() 

247 self._pkey = _ffi.gc(pkey, _lib.EVP_PKEY_free) 

248 self._initialized = False 

249 

250 def to_cryptography_key(self) -> _Key: 

251 """ 

252 Export as a ``cryptography`` key. 

253 

254 :rtype: One of ``cryptography``'s `key interfaces`_. 

255 

256 .. _key interfaces: https://cryptography.io/en/latest/hazmat/\ 

257 primitives/asymmetric/rsa/#key-interfaces 

258 

259 .. versionadded:: 16.1.0 

260 """ 

261 from cryptography.hazmat.primitives.serialization import ( 

262 load_der_private_key, 

263 load_der_public_key, 

264 ) 

265 

266 if self._only_public: 

267 der = dump_publickey(FILETYPE_ASN1, self) 

268 return typing.cast(_Key, load_der_public_key(der)) 

269 else: 

270 der = _dump_privatekey_internal(FILETYPE_ASN1, self) 

271 return typing.cast(_Key, load_der_private_key(der, password=None)) 

272 

273 @classmethod 

274 def from_cryptography_key(cls, crypto_key: _Key) -> PKey: 

275 """ 

276 Construct based on a ``cryptography`` *crypto_key*. 

277 

278 :param crypto_key: A ``cryptography`` key. 

279 :type crypto_key: One of ``cryptography``'s `key interfaces`_. 

280 

281 :rtype: PKey 

282 

283 .. versionadded:: 16.1.0 

284 """ 

285 if not isinstance( 

286 crypto_key, 

287 ( 

288 dsa.DSAPrivateKey, 

289 dsa.DSAPublicKey, 

290 ec.EllipticCurvePrivateKey, 

291 ec.EllipticCurvePublicKey, 

292 ed25519.Ed25519PrivateKey, 

293 ed25519.Ed25519PublicKey, 

294 ed448.Ed448PrivateKey, 

295 ed448.Ed448PublicKey, 

296 rsa.RSAPrivateKey, 

297 rsa.RSAPublicKey, 

298 ), 

299 ): 

300 raise TypeError("Unsupported key type") 

301 

302 from cryptography.hazmat.primitives.serialization import ( 

303 Encoding, 

304 NoEncryption, 

305 PrivateFormat, 

306 PublicFormat, 

307 ) 

308 

309 if isinstance( 

310 crypto_key, 

311 ( 

312 dsa.DSAPublicKey, 

313 ec.EllipticCurvePublicKey, 

314 ed25519.Ed25519PublicKey, 

315 ed448.Ed448PublicKey, 

316 rsa.RSAPublicKey, 

317 ), 

318 ): 

319 return load_publickey( 

320 FILETYPE_ASN1, 

321 crypto_key.public_bytes( 

322 Encoding.DER, PublicFormat.SubjectPublicKeyInfo 

323 ), 

324 ) 

325 else: 

326 der = crypto_key.private_bytes( 

327 Encoding.DER, PrivateFormat.PKCS8, NoEncryption() 

328 ) 

329 return load_privatekey(FILETYPE_ASN1, der) 

330 

331 @deprecated( 

332 "PKey.generate_key is deprecated. You should use the key " 

333 "generation APIs in cryptography instead." 

334 ) 

335 def generate_key(self, type: int, bits: int) -> None: 

336 """ 

337 Generate a key pair of the given type, with the given number of bits. 

338 

339 This generates a key "into" the this object. 

340 

341 :param type: The key type. 

342 :type type: :py:data:`TYPE_RSA` or :py:data:`TYPE_DSA` 

343 :param bits: The number of bits. 

344 :type bits: :py:data:`int` ``>= 0`` 

345 :raises TypeError: If :py:data:`type` or :py:data:`bits` isn't 

346 of the appropriate type. 

347 :raises ValueError: If the number of bits isn't an integer of 

348 the appropriate size. 

349 :return: ``None`` 

350 """ 

351 if not isinstance(type, int): 

352 raise TypeError("type must be an integer") 

353 

354 if not isinstance(bits, int): 

355 raise TypeError("bits must be an integer") 

356 

357 if type == TYPE_RSA: 

358 if bits <= 0: 

359 raise ValueError("Invalid number of bits") 

360 

361 # TODO Check error return 

362 exponent = _lib.BN_new() 

363 exponent = _ffi.gc(exponent, _lib.BN_free) 

364 _lib.BN_set_word(exponent, _lib.RSA_F4) 

365 

366 rsa = _lib.RSA_new() 

367 

368 result = _lib.RSA_generate_key_ex(rsa, bits, exponent, _ffi.NULL) 

369 _openssl_assert(result == 1) 

370 

371 result = _lib.EVP_PKEY_assign_RSA(self._pkey, rsa) 

372 _openssl_assert(result == 1) 

373 

374 elif type == TYPE_DSA: 

375 dsa = _lib.DSA_new() 

376 _openssl_assert(dsa != _ffi.NULL) 

377 

378 dsa = _ffi.gc(dsa, _lib.DSA_free) 

379 res = _lib.DSA_generate_parameters_ex( 

380 dsa, bits, _ffi.NULL, 0, _ffi.NULL, _ffi.NULL, _ffi.NULL 

381 ) 

382 _openssl_assert(res == 1) 

383 

384 _openssl_assert(_lib.DSA_generate_key(dsa) == 1) 

385 _openssl_assert(_lib.EVP_PKEY_set1_DSA(self._pkey, dsa) == 1) 

386 else: 

387 raise Error("No such key type") 

388 

389 self._initialized = True 

390 

391 @deprecated( 

392 "PKey.check is deprecated. You should use the APIs in " 

393 "cryptography instead." 

394 ) 

395 def check(self) -> bool: 

396 """ 

397 Check the consistency of an RSA private key. 

398 

399 This is the Python equivalent of OpenSSL's ``RSA_check_key``. 

400 

401 :return: ``True`` if key is consistent. 

402 

403 :raise OpenSSL.crypto.Error: if the key is inconsistent. 

404 

405 :raise TypeError: if the key is of a type which cannot be checked. 

406 Only RSA keys can currently be checked. 

407 """ 

408 if self._only_public: 

409 raise TypeError("public key only") 

410 

411 if _lib.EVP_PKEY_type(self.type()) != _lib.EVP_PKEY_RSA: 

412 raise TypeError("Only RSA keys can currently be checked.") 

413 

414 rsa = _lib.EVP_PKEY_get1_RSA(self._pkey) 

415 rsa = _ffi.gc(rsa, _lib.RSA_free) 

416 result = _lib.RSA_check_key(rsa) 

417 if result == 1: 

418 return True 

419 _raise_current_error() 

420 

421 def type(self) -> int: 

422 """ 

423 Returns the type of the key 

424 

425 :return: The type of the key. 

426 """ 

427 return _lib.EVP_PKEY_id(self._pkey) 

428 

429 def bits(self) -> int: 

430 """ 

431 Returns the number of bits of the key 

432 

433 :return: The number of bits of the key. 

434 """ 

435 return _lib.EVP_PKEY_bits(self._pkey) 

436 

437 

438class _EllipticCurve: 

439 """ 

440 A representation of a supported elliptic curve. 

441 

442 @cvar _curves: :py:obj:`None` until an attempt is made to load the curves. 

443 Thereafter, a :py:type:`set` containing :py:type:`_EllipticCurve` 

444 instances each of which represents one curve supported by the system. 

445 @type _curves: :py:type:`NoneType` or :py:type:`set` 

446 """ 

447 

448 _curves = None 

449 

450 def __ne__(self, other: Any) -> bool: 

451 """ 

452 Implement cooperation with the right-hand side argument of ``!=``. 

453 

454 Python 3 seems to have dropped this cooperation in this very narrow 

455 circumstance. 

456 """ 

457 if isinstance(other, _EllipticCurve): 

458 return super().__ne__(other) 

459 return NotImplemented 

460 

461 @classmethod 

462 def _load_elliptic_curves(cls, lib: Any) -> set[_EllipticCurve]: 

463 """ 

464 Get the curves supported by OpenSSL. 

465 

466 :param lib: The OpenSSL library binding object. 

467 

468 :return: A :py:type:`set` of ``cls`` instances giving the names of the 

469 elliptic curves the underlying library supports. 

470 """ 

471 num_curves = lib.EC_get_builtin_curves(_ffi.NULL, 0) 

472 builtin_curves = _ffi.new("EC_builtin_curve[]", num_curves) 

473 # The return value on this call should be num_curves again. We 

474 # could check it to make sure but if it *isn't* then.. what could 

475 # we do? Abort the whole process, I suppose...? -exarkun 

476 lib.EC_get_builtin_curves(builtin_curves, num_curves) 

477 return set(cls.from_nid(lib, c.nid) for c in builtin_curves) 

478 

479 @classmethod 

480 def _get_elliptic_curves(cls, lib: Any) -> set[_EllipticCurve]: 

481 """ 

482 Get, cache, and return the curves supported by OpenSSL. 

483 

484 :param lib: The OpenSSL library binding object. 

485 

486 :return: A :py:type:`set` of ``cls`` instances giving the names of the 

487 elliptic curves the underlying library supports. 

488 """ 

489 if cls._curves is None: 

490 cls._curves = cls._load_elliptic_curves(lib) 

491 return cls._curves 

492 

493 @classmethod 

494 def from_nid(cls, lib: Any, nid: int) -> _EllipticCurve: 

495 """ 

496 Instantiate a new :py:class:`_EllipticCurve` associated with the given 

497 OpenSSL NID. 

498 

499 :param lib: The OpenSSL library binding object. 

500 

501 :param nid: The OpenSSL NID the resulting curve object will represent. 

502 This must be a curve NID (and not, for example, a hash NID) or 

503 subsequent operations will fail in unpredictable ways. 

504 :type nid: :py:class:`int` 

505 

506 :return: The curve object. 

507 """ 

508 return cls(lib, nid, _ffi.string(lib.OBJ_nid2sn(nid)).decode("ascii")) 

509 

510 def __init__(self, lib: Any, nid: int, name: str) -> None: 

511 """ 

512 :param _lib: The :py:mod:`cryptography` binding instance used to 

513 interface with OpenSSL. 

514 

515 :param _nid: The OpenSSL NID identifying the curve this object 

516 represents. 

517 :type _nid: :py:class:`int` 

518 

519 :param name: The OpenSSL short name identifying the curve this object 

520 represents. 

521 :type name: :py:class:`unicode` 

522 """ 

523 self._lib = lib 

524 self._nid = nid 

525 self.name = name 

526 

527 def __repr__(self) -> str: 

528 return f"<Curve {self.name!r}>" 

529 

530 def _to_EC_KEY(self) -> Any: 

531 """ 

532 Create a new OpenSSL EC_KEY structure initialized to use this curve. 

533 

534 The structure is automatically garbage collected when the Python object 

535 is garbage collected. 

536 """ 

537 key = self._lib.EC_KEY_new_by_curve_name(self._nid) 

538 return _ffi.gc(key, _lib.EC_KEY_free) 

539 

540 

541@deprecated( 

542 "get_elliptic_curves is deprecated. You should use the APIs in " 

543 "cryptography instead." 

544) 

545def get_elliptic_curves() -> set[_EllipticCurve]: 

546 """ 

547 Return a set of objects representing the elliptic curves supported in the 

548 OpenSSL build in use. 

549 

550 The curve objects have a :py:class:`unicode` ``name`` attribute by which 

551 they identify themselves. 

552 

553 The curve objects are useful as values for the argument accepted by 

554 :py:meth:`Context.set_tmp_ecdh` to specify which elliptical curve should be 

555 used for ECDHE key exchange. 

556 """ 

557 return _EllipticCurve._get_elliptic_curves(_lib) 

558 

559 

560@deprecated( 

561 "get_elliptic_curve is deprecated. You should use the APIs in " 

562 "cryptography instead." 

563) 

564def get_elliptic_curve(name: str) -> _EllipticCurve: 

565 """ 

566 Return a single curve object selected by name. 

567 

568 See :py:func:`get_elliptic_curves` for information about curve objects. 

569 

570 :param name: The OpenSSL short name identifying the curve object to 

571 retrieve. 

572 :type name: :py:class:`unicode` 

573 

574 If the named curve is not supported then :py:class:`ValueError` is raised. 

575 """ 

576 for curve in get_elliptic_curves(): 

577 if curve.name == name: 

578 return curve 

579 raise ValueError("unknown curve name", name) 

580 

581 

582@deprecated( 

583 "X509Name support in pyOpenSSL is deprecated. You should use the " 

584 "APIs in cryptography." 

585) 

586@functools.total_ordering 

587class X509Name: 

588 """ 

589 An X.509 Distinguished Name. 

590 

591 :ivar countryName: The country of the entity. 

592 :ivar C: Alias for :py:attr:`countryName`. 

593 

594 :ivar stateOrProvinceName: The state or province of the entity. 

595 :ivar ST: Alias for :py:attr:`stateOrProvinceName`. 

596 

597 :ivar localityName: The locality of the entity. 

598 :ivar L: Alias for :py:attr:`localityName`. 

599 

600 :ivar organizationName: The organization name of the entity. 

601 :ivar O: Alias for :py:attr:`organizationName`. 

602 

603 :ivar organizationalUnitName: The organizational unit of the entity. 

604 :ivar OU: Alias for :py:attr:`organizationalUnitName` 

605 

606 :ivar commonName: The common name of the entity. 

607 :ivar CN: Alias for :py:attr:`commonName`. 

608 

609 :ivar emailAddress: The e-mail address of the entity. 

610 """ 

611 

612 def __init__(self, name: X509Name) -> None: 

613 """ 

614 Create a new X509Name, copying the given X509Name instance. 

615 

616 :param name: The name to copy. 

617 :type name: :py:class:`X509Name` 

618 """ 

619 name = _lib.X509_NAME_dup(name._name) 

620 self._name: Any = _ffi.gc(name, _lib.X509_NAME_free) 

621 

622 def __setattr__(self, name: str, value: Any) -> None: 

623 if name.startswith("_"): 

624 return super().__setattr__(name, value) 

625 

626 # Note: we really do not want str subclasses here, so we do not use 

627 # isinstance. 

628 if type(name) is not str: 

629 raise TypeError( 

630 f"attribute name must be string, not " 

631 f"'{type(value).__name__:.200}'" 

632 ) 

633 

634 nid = _lib.OBJ_txt2nid(_byte_string(name)) 

635 if nid == _lib.NID_undef: 

636 try: 

637 _raise_current_error() 

638 except Error: 

639 pass 

640 raise AttributeError("No such attribute") 

641 

642 # If there's an old entry for this NID, remove it 

643 for i in range(_lib.X509_NAME_entry_count(self._name)): 

644 ent = _lib.X509_NAME_get_entry(self._name, i) 

645 ent_obj = _lib.X509_NAME_ENTRY_get_object(ent) 

646 ent_nid = _lib.OBJ_obj2nid(ent_obj) 

647 if nid == ent_nid: 

648 ent = _lib.X509_NAME_delete_entry(self._name, i) 

649 _lib.X509_NAME_ENTRY_free(ent) 

650 break 

651 

652 if isinstance(value, str): 

653 value = value.encode("utf-8") 

654 

655 add_result = _lib.X509_NAME_add_entry_by_NID( 

656 self._name, nid, _lib.MBSTRING_UTF8, value, len(value), -1, 0 

657 ) 

658 if not add_result: 

659 _raise_current_error() 

660 

661 def __getattr__(self, name: str) -> str | None: 

662 """ 

663 Find attribute. An X509Name object has the following attributes: 

664 countryName (alias C), stateOrProvince (alias ST), locality (alias L), 

665 organization (alias O), organizationalUnit (alias OU), commonName 

666 (alias CN) and more... 

667 """ 

668 nid = _lib.OBJ_txt2nid(_byte_string(name)) 

669 if nid == _lib.NID_undef: 

670 # This is a bit weird. OBJ_txt2nid indicated failure, but it seems 

671 # a lower level function, a2d_ASN1_OBJECT, also feels the need to 

672 # push something onto the error queue. If we don't clean that up 

673 # now, someone else will bump into it later and be quite confused. 

674 # See lp#314814. 

675 try: 

676 _raise_current_error() 

677 except Error: 

678 pass 

679 raise AttributeError("No such attribute") 

680 

681 entry_index = _lib.X509_NAME_get_index_by_NID(self._name, nid, -1) 

682 if entry_index == -1: 

683 return None 

684 

685 entry = _lib.X509_NAME_get_entry(self._name, entry_index) 

686 data = _lib.X509_NAME_ENTRY_get_data(entry) 

687 

688 result_buffer = _ffi.new("unsigned char**") 

689 data_length = _lib.ASN1_STRING_to_UTF8(result_buffer, data) 

690 _openssl_assert(data_length >= 0) 

691 

692 try: 

693 result = _ffi.buffer(result_buffer[0], data_length)[:].decode( 

694 "utf-8" 

695 ) 

696 finally: 

697 # XXX untested 

698 _lib.OPENSSL_free(result_buffer[0]) 

699 return result 

700 

701 def __eq__(self, other: Any) -> bool: 

702 if not isinstance(other, X509Name): 

703 return NotImplemented 

704 

705 return _lib.X509_NAME_cmp(self._name, other._name) == 0 

706 

707 def __lt__(self, other: Any) -> bool: 

708 if not isinstance(other, X509Name): 

709 return NotImplemented 

710 

711 return _lib.X509_NAME_cmp(self._name, other._name) < 0 

712 

713 def __repr__(self) -> str: 

714 """ 

715 String representation of an X509Name 

716 """ 

717 result_buffer = _ffi.new("char[]", 512) 

718 format_result = _lib.X509_NAME_oneline( 

719 self._name, result_buffer, len(result_buffer) 

720 ) 

721 _openssl_assert(format_result != _ffi.NULL) 

722 

723 return "<X509Name object '{}'>".format( 

724 _ffi.string(result_buffer).decode("utf-8"), 

725 ) 

726 

727 def hash(self) -> int: 

728 """ 

729 Return an integer representation of the first four bytes of the 

730 MD5 digest of the DER representation of the name. 

731 

732 This is the Python equivalent of OpenSSL's ``X509_NAME_hash``. 

733 

734 :return: The (integer) hash of this name. 

735 :rtype: :py:class:`int` 

736 """ 

737 return _lib.X509_NAME_hash(self._name) 

738 

739 def der(self) -> bytes: 

740 """ 

741 Return the DER encoding of this name. 

742 

743 :return: The DER encoded form of this name. 

744 :rtype: :py:class:`bytes` 

745 """ 

746 result_buffer = _ffi.new("unsigned char**") 

747 encode_result = _lib.i2d_X509_NAME(self._name, result_buffer) 

748 _openssl_assert(encode_result >= 0) 

749 

750 string_result = _ffi.buffer(result_buffer[0], encode_result)[:] 

751 _lib.OPENSSL_free(result_buffer[0]) 

752 return string_result 

753 

754 def get_components(self) -> list[tuple[bytes, bytes]]: 

755 """ 

756 Returns the components of this name, as a sequence of 2-tuples. 

757 

758 :return: The components of this name. 

759 :rtype: :py:class:`list` of ``name, value`` tuples. 

760 """ 

761 result = [] 

762 for i in range(_lib.X509_NAME_entry_count(self._name)): 

763 ent = _lib.X509_NAME_get_entry(self._name, i) 

764 

765 fname = _lib.X509_NAME_ENTRY_get_object(ent) 

766 fval = _lib.X509_NAME_ENTRY_get_data(ent) 

767 

768 nid = _lib.OBJ_obj2nid(fname) 

769 name = _lib.OBJ_nid2sn(nid) 

770 

771 # ffi.string does not handle strings containing NULL bytes 

772 # (which may have been generated by old, broken software) 

773 value = _ffi.buffer( 

774 _lib.ASN1_STRING_get0_data(fval), _lib.ASN1_STRING_length(fval) 

775 )[:] 

776 result.append((_ffi.string(name), value)) 

777 

778 return result 

779 

780 

781class X509: 

782 """ 

783 An X.509 certificate. 

784 """ 

785 

786 def __init__(self) -> None: 

787 x509 = _lib.X509_new() 

788 _openssl_assert(x509 != _ffi.NULL) 

789 self._x509 = _ffi.gc(x509, _lib.X509_free) 

790 

791 self._issuer_invalidator = _X509NameInvalidator() 

792 self._subject_invalidator = _X509NameInvalidator() 

793 

794 @classmethod 

795 def _from_raw_x509_ptr(cls, x509: Any) -> X509: 

796 cert = cls.__new__(cls) 

797 cert._x509 = _ffi.gc(x509, _lib.X509_free) 

798 cert._issuer_invalidator = _X509NameInvalidator() 

799 cert._subject_invalidator = _X509NameInvalidator() 

800 return cert 

801 

802 def to_cryptography(self) -> x509.Certificate: 

803 """ 

804 Export as a ``cryptography`` certificate. 

805 

806 :rtype: ``cryptography.x509.Certificate`` 

807 

808 .. versionadded:: 17.1.0 

809 """ 

810 from cryptography.x509 import load_der_x509_certificate 

811 

812 der = dump_certificate(FILETYPE_ASN1, self) 

813 return load_der_x509_certificate(der) 

814 

815 @classmethod 

816 def from_cryptography(cls, crypto_cert: x509.Certificate) -> X509: 

817 """ 

818 Construct based on a ``cryptography`` *crypto_cert*. 

819 

820 :param crypto_key: A ``cryptography`` X.509 certificate. 

821 :type crypto_key: ``cryptography.x509.Certificate`` 

822 

823 :rtype: X509 

824 

825 .. versionadded:: 17.1.0 

826 """ 

827 if not isinstance(crypto_cert, x509.Certificate): 

828 raise TypeError("Must be a certificate") 

829 

830 from cryptography.hazmat.primitives.serialization import Encoding 

831 

832 der = crypto_cert.public_bytes(Encoding.DER) 

833 return load_certificate(FILETYPE_ASN1, der) 

834 

835 @deprecated( 

836 "X509.set_version is deprecated. You should use " 

837 "cryptography's CertificateBuilder instead." 

838 ) 

839 def set_version(self, version: int) -> None: 

840 """ 

841 Set the version number of the certificate. Note that the 

842 version value is zero-based, eg. a value of 0 is V1. 

843 

844 :param version: The version number of the certificate. 

845 :type version: :py:class:`int` 

846 

847 :return: ``None`` 

848 """ 

849 if not isinstance(version, int): 

850 raise TypeError("version must be an integer") 

851 

852 _openssl_assert(_lib.X509_set_version(self._x509, version) == 1) 

853 

854 def get_version(self) -> int: 

855 """ 

856 Return the version number of the certificate. 

857 

858 :return: The version number of the certificate. 

859 :rtype: :py:class:`int` 

860 """ 

861 return _lib.X509_get_version(self._x509) 

862 

863 def get_pubkey(self) -> PKey: 

864 """ 

865 Get the public key of the certificate. 

866 

867 :return: The public key. 

868 :rtype: :py:class:`PKey` 

869 """ 

870 pkey = PKey.__new__(PKey) 

871 pkey._pkey = _lib.X509_get_pubkey(self._x509) 

872 if pkey._pkey == _ffi.NULL: 

873 _raise_current_error() 

874 pkey._pkey = _ffi.gc(pkey._pkey, _lib.EVP_PKEY_free) 

875 pkey._only_public = True 

876 return pkey 

877 

878 @deprecated( 

879 "X509.set_pubkey is deprecated. You should use " 

880 "cryptography's CertificateBuilder instead." 

881 ) 

882 def set_pubkey(self, pkey: PKey) -> None: 

883 """ 

884 Set the public key of the certificate. 

885 

886 :param pkey: The public key. 

887 :type pkey: :py:class:`PKey` 

888 

889 :return: :py:data:`None` 

890 """ 

891 if not isinstance(pkey, PKey): 

892 raise TypeError("pkey must be a PKey instance") 

893 

894 set_result = _lib.X509_set_pubkey(self._x509, pkey._pkey) 

895 _openssl_assert(set_result == 1) 

896 

897 @deprecated( 

898 "X509.sign is deprecated. You should use " 

899 "cryptography's CertificateBuilder instead." 

900 ) 

901 def sign(self, pkey: PKey, digest: str) -> None: 

902 """ 

903 Sign the certificate with this key and digest type. 

904 

905 :param pkey: The key to sign with. 

906 :type pkey: :py:class:`PKey` 

907 

908 :param digest: The name of the message digest to use. 

909 :type digest: :py:class:`str` 

910 

911 :return: :py:data:`None` 

912 """ 

913 if not isinstance(pkey, PKey): 

914 raise TypeError("pkey must be a PKey instance") 

915 

916 if pkey._only_public: 

917 raise ValueError("Key only has public part") 

918 

919 if not pkey._initialized: 

920 raise ValueError("Key is uninitialized") 

921 

922 evp_md = _lib.EVP_get_digestbyname(_byte_string(digest)) 

923 if evp_md == _ffi.NULL: 

924 raise ValueError("No such digest method") 

925 

926 sign_result = _lib.X509_sign(self._x509, pkey._pkey, evp_md) 

927 _openssl_assert(sign_result > 0) 

928 

929 def get_signature_algorithm(self) -> bytes: 

930 """ 

931 Return the signature algorithm used in the certificate. 

932 

933 :return: The name of the algorithm. 

934 :rtype: :py:class:`bytes` 

935 

936 :raises ValueError: If the signature algorithm is undefined. 

937 

938 .. versionadded:: 0.13 

939 """ 

940 sig_alg = _lib.X509_get0_tbs_sigalg(self._x509) 

941 alg = _ffi.new("ASN1_OBJECT **") 

942 _lib.X509_ALGOR_get0(alg, _ffi.NULL, _ffi.NULL, sig_alg) 

943 nid = _lib.OBJ_obj2nid(alg[0]) 

944 if nid == _lib.NID_undef: 

945 raise ValueError("Undefined signature algorithm") 

946 return _ffi.string(_lib.OBJ_nid2ln(nid)) 

947 

948 def digest(self, digest_name: str) -> bytes: 

949 """ 

950 Return the digest of the X509 object. 

951 

952 :param digest_name: The name of the digest algorithm to use. 

953 :type digest_name: :py:class:`str` 

954 

955 :return: The digest of the object, formatted as 

956 :py:const:`b":"`-delimited hex pairs. 

957 :rtype: :py:class:`bytes` 

958 """ 

959 digest = _lib.EVP_get_digestbyname(_byte_string(digest_name)) 

960 if digest == _ffi.NULL: 

961 raise ValueError("No such digest method") 

962 

963 result_buffer = _ffi.new("unsigned char[]", _lib.EVP_MAX_MD_SIZE) 

964 result_length = _ffi.new("unsigned int[]", 1) 

965 result_length[0] = len(result_buffer) 

966 

967 digest_result = _lib.X509_digest( 

968 self._x509, digest, result_buffer, result_length 

969 ) 

970 _openssl_assert(digest_result == 1) 

971 

972 return b":".join( 

973 [ 

974 b16encode(ch).upper() 

975 for ch in _ffi.buffer(result_buffer, result_length[0]) 

976 ] 

977 ) 

978 

979 def subject_name_hash(self) -> int: 

980 """ 

981 Return the hash of the X509 subject. 

982 

983 :return: The hash of the subject. 

984 :rtype: :py:class:`int` 

985 """ 

986 return _lib.X509_subject_name_hash(self._x509) 

987 

988 @deprecated( 

989 "X509.set_serial_number is deprecated. You should use " 

990 "cryptography's CertificateBuilder instead." 

991 ) 

992 def set_serial_number(self, serial: int) -> None: 

993 """ 

994 Set the serial number of the certificate. 

995 

996 :param serial: The new serial number. 

997 :type serial: :py:class:`int` 

998 

999 :return: :py:data`None` 

1000 """ 

1001 if not isinstance(serial, int): 

1002 raise TypeError("serial must be an integer") 

1003 

1004 hex_serial = hex(serial)[2:] 

1005 hex_serial_bytes = hex_serial.encode("ascii") 

1006 

1007 bignum_serial = _ffi.new("BIGNUM**") 

1008 

1009 # BN_hex2bn stores the result in &bignum. 

1010 result = _lib.BN_hex2bn(bignum_serial, hex_serial_bytes) 

1011 _openssl_assert(result != _ffi.NULL) 

1012 

1013 asn1_serial = _lib.BN_to_ASN1_INTEGER(bignum_serial[0], _ffi.NULL) 

1014 _lib.BN_free(bignum_serial[0]) 

1015 _openssl_assert(asn1_serial != _ffi.NULL) 

1016 asn1_serial = _ffi.gc(asn1_serial, _lib.ASN1_INTEGER_free) 

1017 set_result = _lib.X509_set_serialNumber(self._x509, asn1_serial) 

1018 _openssl_assert(set_result == 1) 

1019 

1020 def get_serial_number(self) -> int: 

1021 """ 

1022 Return the serial number of this certificate. 

1023 

1024 :return: The serial number. 

1025 :rtype: int 

1026 """ 

1027 asn1_serial = _lib.X509_get_serialNumber(self._x509) 

1028 bignum_serial = _lib.ASN1_INTEGER_to_BN(asn1_serial, _ffi.NULL) 

1029 try: 

1030 hex_serial = _lib.BN_bn2hex(bignum_serial) 

1031 try: 

1032 hexstring_serial = _ffi.string(hex_serial) 

1033 serial = int(hexstring_serial, 16) 

1034 return serial 

1035 finally: 

1036 _lib.OPENSSL_free(hex_serial) 

1037 finally: 

1038 _lib.BN_free(bignum_serial) 

1039 

1040 @deprecated( 

1041 "X509.gmtime_adj_notAfter is deprecated. You should use " 

1042 "cryptography's CertificateBuilder instead." 

1043 ) 

1044 def gmtime_adj_notAfter(self, amount: int) -> None: 

1045 """ 

1046 Adjust the time stamp on which the certificate stops being valid. 

1047 

1048 :param int amount: The number of seconds by which to adjust the 

1049 timestamp. 

1050 :return: ``None`` 

1051 """ 

1052 if not isinstance(amount, int): 

1053 raise TypeError("amount must be an integer") 

1054 

1055 notAfter = _lib.X509_getm_notAfter(self._x509) 

1056 _lib.X509_gmtime_adj(notAfter, amount) 

1057 

1058 @deprecated( 

1059 "X509.gmtime_adj_notBefore is deprecated. You should use " 

1060 "cryptography's CertificateBuilder instead." 

1061 ) 

1062 def gmtime_adj_notBefore(self, amount: int) -> None: 

1063 """ 

1064 Adjust the timestamp on which the certificate starts being valid. 

1065 

1066 :param amount: The number of seconds by which to adjust the timestamp. 

1067 :return: ``None`` 

1068 """ 

1069 if not isinstance(amount, int): 

1070 raise TypeError("amount must be an integer") 

1071 

1072 notBefore = _lib.X509_getm_notBefore(self._x509) 

1073 _lib.X509_gmtime_adj(notBefore, amount) 

1074 

1075 def has_expired(self) -> bool: 

1076 """ 

1077 Check whether the certificate has expired. 

1078 

1079 :return: ``True`` if the certificate has expired, ``False`` otherwise. 

1080 :rtype: bool 

1081 """ 

1082 time_bytes = self.get_notAfter() 

1083 if time_bytes is None: 

1084 raise ValueError("Unable to determine notAfter") 

1085 time_string = time_bytes.decode("utf-8") 

1086 not_after = datetime.datetime.strptime(time_string, "%Y%m%d%H%M%SZ") 

1087 

1088 UTC = datetime.timezone.utc 

1089 utcnow = datetime.datetime.now(UTC).replace(tzinfo=None) 

1090 return not_after < utcnow 

1091 

1092 def _get_boundary_time(self, which: Any) -> bytes | None: 

1093 return _get_asn1_time(which(self._x509)) 

1094 

1095 def get_notBefore(self) -> bytes | None: 

1096 """ 

1097 Get the timestamp at which the certificate starts being valid. 

1098 

1099 The timestamp is formatted as an ASN.1 TIME:: 

1100 

1101 YYYYMMDDhhmmssZ 

1102 

1103 :return: A timestamp string, or ``None`` if there is none. 

1104 :rtype: bytes or NoneType 

1105 """ 

1106 return self._get_boundary_time(_lib.X509_getm_notBefore) 

1107 

1108 def _set_boundary_time( 

1109 self, which: Callable[..., Any], when: bytes 

1110 ) -> None: 

1111 return _set_asn1_time(which(self._x509), when) 

1112 

1113 @deprecated( 

1114 "X509.set_notBefore is deprecated. You should use " 

1115 "cryptography's CertificateBuilder instead." 

1116 ) 

1117 def set_notBefore(self, when: bytes) -> None: 

1118 """ 

1119 Set the timestamp at which the certificate starts being valid. 

1120 

1121 The timestamp is formatted as an ASN.1 TIME:: 

1122 

1123 YYYYMMDDhhmmssZ 

1124 

1125 :param bytes when: A timestamp string. 

1126 :return: ``None`` 

1127 """ 

1128 return self._set_boundary_time(_lib.X509_getm_notBefore, when) 

1129 

1130 def get_notAfter(self) -> bytes | None: 

1131 """ 

1132 Get the timestamp at which the certificate stops being valid. 

1133 

1134 The timestamp is formatted as an ASN.1 TIME:: 

1135 

1136 YYYYMMDDhhmmssZ 

1137 

1138 :return: A timestamp string, or ``None`` if there is none. 

1139 :rtype: bytes or NoneType 

1140 """ 

1141 return self._get_boundary_time(_lib.X509_getm_notAfter) 

1142 

1143 @deprecated( 

1144 "X509.set_notAfter is deprecated. You should use " 

1145 "cryptography's CertificateBuilder instead." 

1146 ) 

1147 def set_notAfter(self, when: bytes) -> None: 

1148 """ 

1149 Set the timestamp at which the certificate stops being valid. 

1150 

1151 The timestamp is formatted as an ASN.1 TIME:: 

1152 

1153 YYYYMMDDhhmmssZ 

1154 

1155 :param bytes when: A timestamp string. 

1156 :return: ``None`` 

1157 """ 

1158 return self._set_boundary_time(_lib.X509_getm_notAfter, when) 

1159 

1160 def _get_name(self, which: Any) -> X509Name: 

1161 # Bypass X509Name.__new__, which warns that X509Name is deprecated; 

1162 # callers that should warn are decorated individually. 

1163 name = object.__new__(X509Name) 

1164 name._name = which(self._x509) 

1165 _openssl_assert(name._name != _ffi.NULL) 

1166 

1167 # The name is owned by the X509 structure. As long as the X509Name 

1168 # Python object is alive, keep the X509 Python object alive. 

1169 name._owner = self 

1170 

1171 return name 

1172 

1173 def _set_name(self, which: Any, name: X509Name) -> None: 

1174 if not isinstance(name, X509Name): 

1175 raise TypeError("name must be an X509Name") 

1176 set_result = which(self._x509, name._name) 

1177 _openssl_assert(set_result == 1) 

1178 

1179 @deprecated( 

1180 "X509.get_issuer is deprecated. You should use " 

1181 "cryptography's X.509 APIs instead." 

1182 ) 

1183 def get_issuer(self) -> X509Name: 

1184 """ 

1185 Return the issuer of this certificate. 

1186 

1187 This creates a new :class:`X509Name` that wraps the underlying issuer 

1188 name field on the certificate. Modifying it will modify the underlying 

1189 certificate, and will have the effect of modifying any other 

1190 :class:`X509Name` that refers to this issuer. 

1191 

1192 :return: The issuer of this certificate. 

1193 :rtype: :class:`X509Name` 

1194 """ 

1195 name = self._get_name(_lib.X509_get_issuer_name) 

1196 self._issuer_invalidator.add(name) 

1197 return name 

1198 

1199 @deprecated( 

1200 "X509.set_issuer is deprecated. You should use " 

1201 "cryptography's CertificateBuilder instead." 

1202 ) 

1203 def set_issuer(self, issuer: X509Name) -> None: 

1204 """ 

1205 Set the issuer of this certificate. 

1206 

1207 :param issuer: The issuer. 

1208 :type issuer: :py:class:`X509Name` 

1209 

1210 :return: ``None`` 

1211 """ 

1212 self._set_name(_lib.X509_set_issuer_name, issuer) 

1213 self._issuer_invalidator.clear() 

1214 

1215 @deprecated( 

1216 "X509.get_subject is deprecated. You should use " 

1217 "cryptography's X.509 APIs instead." 

1218 ) 

1219 def get_subject(self) -> X509Name: 

1220 """ 

1221 Return the subject of this certificate. 

1222 

1223 This creates a new :class:`X509Name` that wraps the underlying subject 

1224 name field on the certificate. Modifying it will modify the underlying 

1225 certificate, and will have the effect of modifying any other 

1226 :class:`X509Name` that refers to this subject. 

1227 

1228 :return: The subject of this certificate. 

1229 :rtype: :class:`X509Name` 

1230 """ 

1231 name = self._get_name(_lib.X509_get_subject_name) 

1232 self._subject_invalidator.add(name) 

1233 return name 

1234 

1235 @deprecated( 

1236 "X509.set_subject is deprecated. You should use " 

1237 "cryptography's CertificateBuilder instead." 

1238 ) 

1239 def set_subject(self, subject: X509Name) -> None: 

1240 """ 

1241 Set the subject of this certificate. 

1242 

1243 :param subject: The subject. 

1244 :type subject: :py:class:`X509Name` 

1245 

1246 :return: ``None`` 

1247 """ 

1248 self._set_name(_lib.X509_set_subject_name, subject) 

1249 self._subject_invalidator.clear() 

1250 

1251 def get_extension_count(self) -> int: 

1252 """ 

1253 Get the number of extensions on this certificate. 

1254 

1255 :return: The number of extensions. 

1256 :rtype: :py:class:`int` 

1257 

1258 .. versionadded:: 0.12 

1259 """ 

1260 return _lib.X509_get_ext_count(self._x509) 

1261 

1262 

1263class X509StoreFlags: 

1264 """ 

1265 Flags for X509 verification, used to change the behavior of 

1266 :class:`X509Store`. 

1267 

1268 See `OpenSSL Verification Flags`_ for details. 

1269 

1270 .. _OpenSSL Verification Flags: 

1271 https://www.openssl.org/docs/manmaster/man3/X509_VERIFY_PARAM_set_flags.html 

1272 """ 

1273 

1274 CRL_CHECK: int = _lib.X509_V_FLAG_CRL_CHECK 

1275 CRL_CHECK_ALL: int = _lib.X509_V_FLAG_CRL_CHECK_ALL 

1276 IGNORE_CRITICAL: int = _lib.X509_V_FLAG_IGNORE_CRITICAL 

1277 X509_STRICT: int = _lib.X509_V_FLAG_X509_STRICT 

1278 ALLOW_PROXY_CERTS: int = _lib.X509_V_FLAG_ALLOW_PROXY_CERTS 

1279 POLICY_CHECK: int = _lib.X509_V_FLAG_POLICY_CHECK 

1280 EXPLICIT_POLICY: int = _lib.X509_V_FLAG_EXPLICIT_POLICY 

1281 INHIBIT_MAP: int = _lib.X509_V_FLAG_INHIBIT_MAP 

1282 CHECK_SS_SIGNATURE: int = _lib.X509_V_FLAG_CHECK_SS_SIGNATURE 

1283 PARTIAL_CHAIN: int = _lib.X509_V_FLAG_PARTIAL_CHAIN 

1284 

1285 

1286class X509Store: 

1287 """ 

1288 An X.509 store. 

1289 

1290 An X.509 store is used to describe a context in which to verify a 

1291 certificate. A description of a context may include a set of certificates 

1292 to trust, a set of certificate revocation lists, verification flags and 

1293 more. 

1294 

1295 An X.509 store, being only a description, cannot be used by itself to 

1296 verify a certificate. To carry out the actual verification process, see 

1297 :class:`X509StoreContext`. 

1298 """ 

1299 

1300 def __init__(self) -> None: 

1301 store = _lib.X509_STORE_new() 

1302 self._store = _ffi.gc(store, _lib.X509_STORE_free) 

1303 

1304 def add_cert(self, cert: X509) -> None: 

1305 """ 

1306 Adds a trusted certificate to this store. 

1307 

1308 Adding a certificate with this method adds this certificate as a 

1309 *trusted* certificate. 

1310 

1311 :param X509 cert: The certificate to add to this store. 

1312 

1313 :raises TypeError: If the certificate is not an :class:`X509`. 

1314 

1315 :raises OpenSSL.crypto.Error: If OpenSSL was unhappy with your 

1316 certificate. 

1317 

1318 :return: ``None`` if the certificate was added successfully. 

1319 """ 

1320 if not isinstance(cert, X509): 

1321 raise TypeError() 

1322 

1323 res = _lib.X509_STORE_add_cert(self._store, cert._x509) 

1324 _openssl_assert(res == 1) 

1325 

1326 def add_crl(self, crl: x509.CertificateRevocationList) -> None: 

1327 """ 

1328 Add a certificate revocation list to this store. 

1329 

1330 The certificate revocation lists added to a store will only be used if 

1331 the associated flags are configured to check certificate revocation 

1332 lists. 

1333 

1334 .. versionadded:: 16.1.0 

1335 

1336 :param crl: The certificate revocation list to add to this store. 

1337 :type crl: ``cryptography.x509.CertificateRevocationList`` 

1338 :return: ``None`` if the certificate revocation list was added 

1339 successfully. 

1340 """ 

1341 if isinstance(crl, x509.CertificateRevocationList): 

1342 from cryptography.hazmat.primitives.serialization import Encoding 

1343 

1344 bio = _new_mem_buf(crl.public_bytes(Encoding.DER)) 

1345 openssl_crl = _lib.d2i_X509_CRL_bio(bio, _ffi.NULL) 

1346 _openssl_assert(openssl_crl != _ffi.NULL) 

1347 crl = _ffi.gc(openssl_crl, _lib.X509_CRL_free) 

1348 else: 

1349 raise TypeError( 

1350 "CRL must be of type " 

1351 "cryptography.x509.CertificateRevocationList" 

1352 ) 

1353 

1354 _openssl_assert(_lib.X509_STORE_add_crl(self._store, crl) != 0) 

1355 

1356 def set_flags(self, flags: int) -> None: 

1357 """ 

1358 Set verification flags to this store. 

1359 

1360 Verification flags can be combined by oring them together. 

1361 

1362 .. note:: 

1363 

1364 Setting a verification flag sometimes requires clients to add 

1365 additional information to the store, otherwise a suitable error will 

1366 be raised. 

1367 

1368 For example, in setting flags to enable CRL checking a 

1369 suitable CRL must be added to the store otherwise an error will be 

1370 raised. 

1371 

1372 .. versionadded:: 16.1.0 

1373 

1374 :param int flags: The verification flags to set on this store. 

1375 See :class:`X509StoreFlags` for available constants. 

1376 :return: ``None`` if the verification flags were successfully set. 

1377 """ 

1378 _openssl_assert(_lib.X509_STORE_set_flags(self._store, flags) != 0) 

1379 

1380 def set_time(self, vfy_time: datetime.datetime) -> None: 

1381 """ 

1382 Set the time against which the certificates are verified. 

1383 

1384 Normally the current time is used. 

1385 

1386 .. note:: 

1387 

1388 For example, you can determine if a certificate was valid at a given 

1389 time. 

1390 

1391 .. versionadded:: 17.0.0 

1392 

1393 :param datetime vfy_time: The verification time to set on this store. 

1394 :return: ``None`` if the verification time was successfully set. 

1395 """ 

1396 param = _lib.X509_VERIFY_PARAM_new() 

1397 param = _ffi.gc(param, _lib.X509_VERIFY_PARAM_free) 

1398 

1399 _lib.X509_VERIFY_PARAM_set_time( 

1400 param, calendar.timegm(vfy_time.timetuple()) 

1401 ) 

1402 _openssl_assert(_lib.X509_STORE_set1_param(self._store, param) != 0) 

1403 

1404 def load_locations( 

1405 self, 

1406 cafile: StrOrBytesPath | None, 

1407 capath: StrOrBytesPath | None = None, 

1408 ) -> None: 

1409 """ 

1410 Let X509Store know where we can find trusted certificates for the 

1411 certificate chain. Note that the certificates have to be in PEM 

1412 format. 

1413 

1414 If *capath* is passed, it must be a directory prepared using the 

1415 ``c_rehash`` tool included with OpenSSL. Either, but not both, of 

1416 *cafile* or *capath* may be ``None``. 

1417 

1418 .. note:: 

1419 

1420 Both *cafile* and *capath* may be set simultaneously. 

1421 

1422 Call this method multiple times to add more than one location. 

1423 For example, CA certificates, and certificate revocation list bundles 

1424 may be passed in *cafile* in subsequent calls to this method. 

1425 

1426 .. versionadded:: 20.0 

1427 

1428 :param cafile: In which file we can find the certificates (``bytes`` or 

1429 ``unicode``). 

1430 :param capath: In which directory we can find the certificates 

1431 (``bytes`` or ``unicode``). 

1432 

1433 :return: ``None`` if the locations were set successfully. 

1434 

1435 :raises OpenSSL.crypto.Error: If both *cafile* and *capath* is ``None`` 

1436 or the locations could not be set for any reason. 

1437 

1438 """ 

1439 if cafile is None: 

1440 cafile = _ffi.NULL 

1441 else: 

1442 cafile = _path_bytes(cafile) 

1443 

1444 if capath is None: 

1445 capath = _ffi.NULL 

1446 else: 

1447 capath = _path_bytes(capath) 

1448 

1449 load_result = _lib.X509_STORE_load_locations( 

1450 self._store, cafile, capath 

1451 ) 

1452 if not load_result: 

1453 _raise_current_error() 

1454 

1455 

1456class X509StoreContextError(Exception): 

1457 """ 

1458 An exception raised when an error occurred while verifying a certificate 

1459 using `OpenSSL.X509StoreContext.verify_certificate`. 

1460 

1461 :ivar certificate: The certificate which caused verificate failure. 

1462 :type certificate: :class:`X509` 

1463 """ 

1464 

1465 def __init__( 

1466 self, message: str, errors: list[Any], certificate: X509 

1467 ) -> None: 

1468 super().__init__(message) 

1469 self.errors = errors 

1470 self.certificate = certificate 

1471 

1472 

1473class X509StoreContext: 

1474 """ 

1475 An X.509 store context. 

1476 

1477 An X.509 store context is used to carry out the actual verification process 

1478 of a certificate in a described context. For describing such a context, see 

1479 :class:`X509Store`. 

1480 

1481 :param X509Store store: The certificates which will be trusted for the 

1482 purposes of any verifications. 

1483 :param X509 certificate: The certificate to be verified. 

1484 :param chain: List of untrusted certificates that may be used for building 

1485 the certificate chain. May be ``None``. 

1486 :type chain: :class:`list` of :class:`X509` 

1487 """ 

1488 

1489 def __init__( 

1490 self, 

1491 store: X509Store, 

1492 certificate: X509, 

1493 chain: Sequence[X509] | None = None, 

1494 ) -> None: 

1495 self._store = store 

1496 self._cert = certificate 

1497 self._chain = self._build_certificate_stack(chain) 

1498 

1499 @staticmethod 

1500 def _build_certificate_stack( 

1501 certificates: Sequence[X509] | None, 

1502 ) -> None: 

1503 def cleanup(s: Any) -> None: 

1504 # Equivalent to sk_X509_pop_free, but we don't 

1505 # currently have a CFFI binding for that available 

1506 for i in range(_lib.sk_X509_num(s)): 

1507 x = _lib.sk_X509_value(s, i) 

1508 _lib.X509_free(x) 

1509 _lib.sk_X509_free(s) 

1510 

1511 if certificates is None or len(certificates) == 0: 

1512 return _ffi.NULL 

1513 

1514 stack = _lib.sk_X509_new_null() 

1515 _openssl_assert(stack != _ffi.NULL) 

1516 stack = _ffi.gc(stack, cleanup) 

1517 

1518 for cert in certificates: 

1519 if not isinstance(cert, X509): 

1520 raise TypeError("One of the elements is not an X509 instance") 

1521 

1522 _openssl_assert(_lib.X509_up_ref(cert._x509) > 0) 

1523 if _lib.sk_X509_push(stack, cert._x509) <= 0: 

1524 _lib.X509_free(cert._x509) 

1525 _raise_current_error() 

1526 

1527 return stack 

1528 

1529 @staticmethod 

1530 def _exception_from_context(store_ctx: Any) -> X509StoreContextError: 

1531 """ 

1532 Convert an OpenSSL native context error failure into a Python 

1533 exception. 

1534 

1535 When a call to native OpenSSL X509_verify_cert fails, additional 

1536 information about the failure can be obtained from the store context. 

1537 """ 

1538 message = _ffi.string( 

1539 _lib.X509_verify_cert_error_string( 

1540 _lib.X509_STORE_CTX_get_error(store_ctx) 

1541 ) 

1542 ).decode("utf-8") 

1543 errors = [ 

1544 _lib.X509_STORE_CTX_get_error(store_ctx), 

1545 _lib.X509_STORE_CTX_get_error_depth(store_ctx), 

1546 message, 

1547 ] 

1548 # A context error should always be associated with a certificate, so we 

1549 # expect this call to never return :class:`None`. 

1550 _x509 = _lib.X509_STORE_CTX_get_current_cert(store_ctx) 

1551 _cert = _lib.X509_dup(_x509) 

1552 pycert = X509._from_raw_x509_ptr(_cert) 

1553 return X509StoreContextError(message, errors, pycert) 

1554 

1555 def _verify_certificate(self) -> Any: 

1556 """ 

1557 Verifies the certificate and runs an X509_STORE_CTX containing the 

1558 results. 

1559 

1560 :raises X509StoreContextError: If an error occurred when validating a 

1561 certificate in the context. Sets ``certificate`` attribute to 

1562 indicate which certificate caused the error. 

1563 """ 

1564 store_ctx = _lib.X509_STORE_CTX_new() 

1565 _openssl_assert(store_ctx != _ffi.NULL) 

1566 store_ctx = _ffi.gc(store_ctx, _lib.X509_STORE_CTX_free) 

1567 

1568 ret = _lib.X509_STORE_CTX_init( 

1569 store_ctx, self._store._store, self._cert._x509, self._chain 

1570 ) 

1571 _openssl_assert(ret == 1) 

1572 

1573 ret = _lib.X509_verify_cert(store_ctx) 

1574 if ret <= 0: 

1575 raise self._exception_from_context(store_ctx) 

1576 

1577 return store_ctx 

1578 

1579 def set_store(self, store: X509Store) -> None: 

1580 """ 

1581 Set the context's X.509 store. 

1582 

1583 .. versionadded:: 0.15 

1584 

1585 :param X509Store store: The store description which will be used for 

1586 the purposes of any *future* verifications. 

1587 """ 

1588 self._store = store 

1589 

1590 def verify_certificate(self) -> None: 

1591 """ 

1592 Verify a certificate in a context. 

1593 

1594 .. versionadded:: 0.15 

1595 

1596 :raises X509StoreContextError: If an error occurred when validating a 

1597 certificate in the context. Sets ``certificate`` attribute to 

1598 indicate which certificate caused the error. 

1599 """ 

1600 self._verify_certificate() 

1601 

1602 def get_verified_chain(self) -> list[X509]: 

1603 """ 

1604 Verify a certificate in a context and return the complete validated 

1605 chain. 

1606 

1607 :raises X509StoreContextError: If an error occurred when validating a 

1608 certificate in the context. Sets ``certificate`` attribute to 

1609 indicate which certificate caused the error. 

1610 

1611 .. versionadded:: 20.0 

1612 """ 

1613 store_ctx = self._verify_certificate() 

1614 

1615 # Note: X509_STORE_CTX_get1_chain returns a deep copy of the chain. 

1616 cert_stack = _lib.X509_STORE_CTX_get1_chain(store_ctx) 

1617 _openssl_assert(cert_stack != _ffi.NULL) 

1618 

1619 result = [] 

1620 for i in range(_lib.sk_X509_num(cert_stack)): 

1621 cert = _lib.sk_X509_value(cert_stack, i) 

1622 _openssl_assert(cert != _ffi.NULL) 

1623 pycert = X509._from_raw_x509_ptr(cert) 

1624 result.append(pycert) 

1625 

1626 # Free the stack but not the members which are freed by the X509 class. 

1627 _lib.sk_X509_free(cert_stack) 

1628 return result 

1629 

1630 

1631def load_certificate(type: int, buffer: bytes) -> X509: 

1632 """ 

1633 Load a certificate (X509) from the string *buffer* encoded with the 

1634 type *type*. 

1635 

1636 :param type: The file type (one of FILETYPE_PEM, FILETYPE_ASN1) 

1637 

1638 :param bytes buffer: The buffer the certificate is stored in 

1639 

1640 :return: The X509 object 

1641 """ 

1642 if isinstance(buffer, str): 

1643 buffer = buffer.encode("ascii") 

1644 

1645 bio = _new_mem_buf(buffer) 

1646 

1647 if type == FILETYPE_PEM: 

1648 x509 = _lib.PEM_read_bio_X509(bio, _ffi.NULL, _ffi.NULL, _ffi.NULL) 

1649 elif type == FILETYPE_ASN1: 

1650 x509 = _lib.d2i_X509_bio(bio, _ffi.NULL) 

1651 else: 

1652 raise ValueError("type argument must be FILETYPE_PEM or FILETYPE_ASN1") 

1653 

1654 if x509 == _ffi.NULL: 

1655 _raise_current_error() 

1656 

1657 return X509._from_raw_x509_ptr(x509) 

1658 

1659 

1660def dump_certificate(type: int, cert: X509) -> bytes: 

1661 """ 

1662 Dump the certificate *cert* into a buffer string encoded with the type 

1663 *type*. 

1664 

1665 :param type: The file type (one of FILETYPE_PEM, FILETYPE_ASN1, or 

1666 FILETYPE_TEXT) 

1667 :param cert: The certificate to dump 

1668 :return: The buffer with the dumped certificate in 

1669 """ 

1670 bio = _new_mem_buf() 

1671 

1672 if type == FILETYPE_PEM: 

1673 result_code = _lib.PEM_write_bio_X509(bio, cert._x509) 

1674 elif type == FILETYPE_ASN1: 

1675 result_code = _lib.i2d_X509_bio(bio, cert._x509) 

1676 elif type == FILETYPE_TEXT: 

1677 result_code = _lib.X509_print_ex(bio, cert._x509, 0, 0) 

1678 else: 

1679 raise ValueError( 

1680 "type argument must be FILETYPE_PEM, FILETYPE_ASN1, or " 

1681 "FILETYPE_TEXT" 

1682 ) 

1683 

1684 _openssl_assert(result_code == 1) 

1685 return _bio_to_string(bio) 

1686 

1687 

1688def dump_publickey(type: int, pkey: PKey) -> bytes: 

1689 """ 

1690 Dump a public key to a buffer. 

1691 

1692 :param type: The file type (one of :data:`FILETYPE_PEM` or 

1693 :data:`FILETYPE_ASN1`). 

1694 :param PKey pkey: The public key to dump 

1695 :return: The buffer with the dumped key in it. 

1696 :rtype: bytes 

1697 """ 

1698 bio = _new_mem_buf() 

1699 if type == FILETYPE_PEM: 

1700 write_bio = _lib.PEM_write_bio_PUBKEY 

1701 elif type == FILETYPE_ASN1: 

1702 write_bio = _lib.i2d_PUBKEY_bio 

1703 else: 

1704 raise ValueError("type argument must be FILETYPE_PEM or FILETYPE_ASN1") 

1705 

1706 result_code = write_bio(bio, pkey._pkey) 

1707 if result_code != 1: # pragma: no cover 

1708 _raise_current_error() 

1709 

1710 return _bio_to_string(bio) 

1711 

1712 

1713def dump_privatekey( 

1714 type: int, 

1715 pkey: PKey, 

1716 cipher: str | None = None, 

1717 passphrase: PassphraseCallableT | None = None, 

1718) -> bytes: 

1719 """ 

1720 Dump the private key *pkey* into a buffer string encoded with the type 

1721 *type*. Optionally (if *type* is :const:`FILETYPE_PEM`) encrypting it 

1722 using *cipher* and *passphrase*. 

1723 

1724 :param type: The file type (one of :const:`FILETYPE_PEM`, 

1725 :const:`FILETYPE_ASN1`, or :const:`FILETYPE_TEXT`) 

1726 :param PKey pkey: The PKey to dump 

1727 :param cipher: (optional) if encrypted PEM format, the cipher to use 

1728 :param passphrase: (optional) if encrypted PEM format, this can be either 

1729 the passphrase to use, or a callback for providing the passphrase. 

1730 

1731 :return: The buffer with the dumped key in 

1732 :rtype: bytes 

1733 

1734 .. deprecated:: 26.3.0 

1735 Use the serialization APIs on ``cryptography`` private key types 

1736 instead. 

1737 """ 

1738 bio = _new_mem_buf() 

1739 

1740 if not isinstance(pkey, PKey): 

1741 raise TypeError("pkey must be a PKey") 

1742 

1743 if cipher is not None: 

1744 if passphrase is None: 

1745 raise TypeError( 

1746 "if a value is given for cipher " 

1747 "one must also be given for passphrase" 

1748 ) 

1749 cipher_obj = _lib.EVP_get_cipherbyname(_byte_string(cipher)) 

1750 if cipher_obj == _ffi.NULL: 

1751 raise ValueError("Invalid cipher name") 

1752 else: 

1753 cipher_obj = _ffi.NULL 

1754 

1755 helper = _PassphraseHelper(type, passphrase) 

1756 if type == FILETYPE_PEM: 

1757 result_code = _lib.PEM_write_bio_PrivateKey( 

1758 bio, 

1759 pkey._pkey, 

1760 cipher_obj, 

1761 _ffi.NULL, 

1762 0, 

1763 helper.callback, 

1764 helper.callback_args, 

1765 ) 

1766 helper.raise_if_problem() 

1767 elif type == FILETYPE_ASN1: 

1768 result_code = _lib.i2d_PrivateKey_bio(bio, pkey._pkey) 

1769 elif type == FILETYPE_TEXT: 

1770 if _lib.EVP_PKEY_id(pkey._pkey) != _lib.EVP_PKEY_RSA: 

1771 raise TypeError("Only RSA keys are supported for FILETYPE_TEXT") 

1772 

1773 rsa = _ffi.gc(_lib.EVP_PKEY_get1_RSA(pkey._pkey), _lib.RSA_free) 

1774 result_code = _lib.RSA_print(bio, rsa, 0) 

1775 else: 

1776 raise ValueError( 

1777 "type argument must be FILETYPE_PEM, FILETYPE_ASN1, or " 

1778 "FILETYPE_TEXT" 

1779 ) 

1780 

1781 _openssl_assert(result_code != 0) 

1782 

1783 return _bio_to_string(bio) 

1784 

1785 

1786_dump_privatekey_internal = dump_privatekey 

1787 

1788utils.deprecated( 

1789 dump_privatekey, 

1790 __name__, 

1791 ( 

1792 "dump_privatekey is deprecated. You should use the APIs in " 

1793 "cryptography." 

1794 ), 

1795 DeprecationWarning, 

1796 name="dump_privatekey", 

1797) 

1798 

1799 

1800class _PassphraseHelper: 

1801 def __init__( 

1802 self, 

1803 type: int, 

1804 passphrase: PassphraseCallableT | None, 

1805 more_args: bool = False, 

1806 truncate: bool = False, 

1807 ) -> None: 

1808 if type != FILETYPE_PEM and passphrase is not None: 

1809 raise ValueError( 

1810 "only FILETYPE_PEM key format supports encryption" 

1811 ) 

1812 self._passphrase = passphrase 

1813 self._more_args = more_args 

1814 self._truncate = truncate 

1815 self._problems: list[Exception] = [] 

1816 

1817 @property 

1818 def callback(self) -> Any: 

1819 if self._passphrase is None: 

1820 return _ffi.NULL 

1821 elif isinstance(self._passphrase, bytes) or callable(self._passphrase): 

1822 return _ffi.callback("pem_password_cb", self._read_passphrase) 

1823 else: 

1824 raise TypeError( 

1825 "Last argument must be a byte string or a callable." 

1826 ) 

1827 

1828 @property 

1829 def callback_args(self) -> Any: 

1830 if self._passphrase is None: 

1831 return _ffi.NULL 

1832 elif isinstance(self._passphrase, bytes) or callable(self._passphrase): 

1833 return _ffi.NULL 

1834 else: 

1835 raise TypeError( 

1836 "Last argument must be a byte string or a callable." 

1837 ) 

1838 

1839 def raise_if_problem(self, exceptionType: type[Exception] = Error) -> None: 

1840 if self._problems: 

1841 # Flush the OpenSSL error queue 

1842 try: 

1843 _exception_from_error_queue(exceptionType) 

1844 except exceptionType: 

1845 pass 

1846 

1847 raise self._problems.pop(0) 

1848 

1849 def _read_passphrase( 

1850 self, buf: Any, size: int, rwflag: Any, userdata: Any 

1851 ) -> int: 

1852 try: 

1853 if callable(self._passphrase): 

1854 if self._more_args: 

1855 result = self._passphrase(size, rwflag, userdata) 

1856 else: 

1857 result = self._passphrase(rwflag) 

1858 else: 

1859 assert self._passphrase is not None 

1860 result = self._passphrase 

1861 if not isinstance(result, bytes): 

1862 raise ValueError("Bytes expected") 

1863 if len(result) > size: 

1864 if self._truncate: 

1865 result = result[:size] 

1866 else: 

1867 raise ValueError( 

1868 "passphrase returned by callback is too long" 

1869 ) 

1870 for i in range(len(result)): 

1871 buf[i] = result[i : i + 1] 

1872 return len(result) 

1873 except Exception as e: 

1874 self._problems.append(e) 

1875 return 0 

1876 

1877 

1878def load_publickey(type: int, buffer: str | bytes) -> PKey: 

1879 """ 

1880 Load a public key from a buffer. 

1881 

1882 :param type: The file type (one of :data:`FILETYPE_PEM`, 

1883 :data:`FILETYPE_ASN1`). 

1884 :param buffer: The buffer the key is stored in. 

1885 :type buffer: A Python string object, either unicode or bytestring. 

1886 :return: The PKey object. 

1887 :rtype: :class:`PKey` 

1888 """ 

1889 if isinstance(buffer, str): 

1890 buffer = buffer.encode("ascii") 

1891 

1892 bio = _new_mem_buf(buffer) 

1893 

1894 if type == FILETYPE_PEM: 

1895 evp_pkey = _lib.PEM_read_bio_PUBKEY( 

1896 bio, _ffi.NULL, _ffi.NULL, _ffi.NULL 

1897 ) 

1898 elif type == FILETYPE_ASN1: 

1899 evp_pkey = _lib.d2i_PUBKEY_bio(bio, _ffi.NULL) 

1900 else: 

1901 raise ValueError("type argument must be FILETYPE_PEM or FILETYPE_ASN1") 

1902 

1903 if evp_pkey == _ffi.NULL: 

1904 _raise_current_error() 

1905 

1906 pkey = PKey.__new__(PKey) 

1907 pkey._pkey = _ffi.gc(evp_pkey, _lib.EVP_PKEY_free) 

1908 pkey._only_public = True 

1909 return pkey 

1910 

1911 

1912def load_privatekey( 

1913 type: int, 

1914 buffer: str | bytes, 

1915 passphrase: PassphraseCallableT | None = None, 

1916) -> PKey: 

1917 """ 

1918 Load a private key (PKey) from the string *buffer* encoded with the type 

1919 *type*. 

1920 

1921 :param type: The file type (one of FILETYPE_PEM, FILETYPE_ASN1) 

1922 :param buffer: The buffer the key is stored in 

1923 :param passphrase: (optional) if encrypted PEM format, this can be 

1924 either the passphrase to use, or a callback for 

1925 providing the passphrase. 

1926 

1927 :return: The PKey object 

1928 """ 

1929 if isinstance(buffer, str): 

1930 buffer = buffer.encode("ascii") 

1931 

1932 bio = _new_mem_buf(buffer) 

1933 

1934 helper = _PassphraseHelper(type, passphrase) 

1935 if type == FILETYPE_PEM: 

1936 evp_pkey = _lib.PEM_read_bio_PrivateKey( 

1937 bio, _ffi.NULL, helper.callback, helper.callback_args 

1938 ) 

1939 helper.raise_if_problem() 

1940 elif type == FILETYPE_ASN1: 

1941 evp_pkey = _lib.d2i_PrivateKey_bio(bio, _ffi.NULL) 

1942 else: 

1943 raise ValueError("type argument must be FILETYPE_PEM or FILETYPE_ASN1") 

1944 

1945 if evp_pkey == _ffi.NULL: 

1946 _raise_current_error() 

1947 

1948 pkey = PKey.__new__(PKey) 

1949 pkey._pkey = _ffi.gc(evp_pkey, _lib.EVP_PKEY_free) 

1950 return pkey