|
7 | 7 | import contextlib |
8 | 8 | import os |
9 | 9 | import ssl |
| 10 | +import tempfile |
10 | 11 | import typing |
11 | 12 | from functools import lru_cache |
12 | 13 | from threading import RLock |
13 | 14 |
|
| 15 | +if typing.TYPE_CHECKING: |
| 16 | + from io import BufferedWriter |
| 17 | + |
14 | 18 | from ._version import VERSION, __version__ |
15 | 19 |
|
16 | 20 | #: Determine if we could load correctly the non-native rust module. |
|
24 | 28 | _USER_APPEND_CA_LOCK = RLock() |
25 | 29 |
|
26 | 30 |
|
| 31 | +@contextlib.contextmanager |
| 32 | +def _atomic_open(filename: str) -> typing.Generator[BufferedWriter, None, None]: |
| 33 | + """Write a file to the disk in an atomic fashion""" |
| 34 | + tmp_descriptor, tmp_name = tempfile.mkstemp(dir=os.path.dirname(filename)) |
| 35 | + try: |
| 36 | + with os.fdopen(tmp_descriptor, "wb") as tmp_handler: |
| 37 | + yield tmp_handler |
| 38 | + os.replace(tmp_name, filename) |
| 39 | + except BaseException: |
| 40 | + os.remove(tmp_name) |
| 41 | + raise |
| 42 | + |
| 43 | + |
| 44 | +def _extract_zipped_paths(path: str) -> str: |
| 45 | + """Replace nonexistent paths that look like they refer to a member of a zip |
| 46 | + archive with the location of an extracted copy of the target, or else |
| 47 | + just return the provided path unchanged. |
| 48 | + """ |
| 49 | + if os.path.exists(path): |
| 50 | + # this is already a valid path, no need to do anything further |
| 51 | + return path |
| 52 | + |
| 53 | + import zipfile |
| 54 | + |
| 55 | + # find the first valid part of the provided path and treat that as a zip archive |
| 56 | + # assume the rest of the path is the name of a member in the archive |
| 57 | + archive, member = os.path.split(path) |
| 58 | + while archive and not os.path.exists(archive): |
| 59 | + archive, prefix = os.path.split(archive) |
| 60 | + if not prefix: |
| 61 | + # If we don't check for an empty prefix after the split (in other words, archive remains unchanged after the split), |
| 62 | + # we _can_ end up in an infinite loop on a rare corner case affecting a small number of users |
| 63 | + break |
| 64 | + member = "/".join([prefix, member]) |
| 65 | + |
| 66 | + if not zipfile.is_zipfile(archive): |
| 67 | + return path |
| 68 | + |
| 69 | + zip_file = zipfile.ZipFile(archive) |
| 70 | + if member not in zip_file.namelist(): |
| 71 | + return path |
| 72 | + |
| 73 | + # we have a valid zip archive and a valid member of that archive |
| 74 | + tmp = tempfile.gettempdir() |
| 75 | + extracted_path = os.path.join(tmp, member.split("/")[-1]) |
| 76 | + |
| 77 | + if not os.path.exists(extracted_path): |
| 78 | + # use read + write to avoid the creating nested folders, we only want the file, avoids mkdir racing condition |
| 79 | + with _atomic_open(extracted_path) as file_handler: |
| 80 | + file_handler.write(zip_file.read(member)) |
| 81 | + |
| 82 | + return extracted_path |
| 83 | + |
| 84 | + |
27 | 85 | def _split_certifi_bundle(data: bytes) -> list[str]: |
28 | 86 | line_ending = b"\n" if b"-----\r\n" not in data else b"\r\n" |
29 | 87 | boundary = b"-----END CERTIFICATE-----" + line_ending |
@@ -66,7 +124,7 @@ def _certifi_fallback() -> list[bytes]: |
66 | 124 | certs: list[bytes] = [] |
67 | 125 |
|
68 | 126 | try: |
69 | | - with open(certifi.where(), "rb") as fp: |
| 127 | + with open(_extract_zipped_paths(certifi.where()), "rb") as fp: |
70 | 128 | for pem_cert in _split_certifi_bundle(fp.read()): |
71 | 129 | certs.append(ssl.PEM_cert_to_DER_cert(pem_cert)) |
72 | 130 | except (OSError, PermissionError) as e: |
|
0 commit comments