Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/filelock/_strict.py: 22%

1from __future__ import annotations 

2 

3import contextlib 

4import errno 

5import os 

6import secrets 

7import stat 

8import sys 

9import tempfile 

10import time 

11from dataclasses import dataclass 

12from errno import EACCES, EEXIST, ENOENT, ENOSYS, EPERM, ESTALE, EXDEV 

13from pathlib import Path 

14from typing import TYPE_CHECKING, Final, Literal, cast 

15 

16from ._api import BaseFileLock, _canonical, _raise_cleanup_errors 

17from ._error import SoftFileLockProtocolError 

18from ._identity import host_name, process_start_token 

19from ._soft_protocol import STRICT_SOFT_SENTINEL_RECORD 

20from ._util import ensure_directory_exists, write_all 

21 

22if TYPE_CHECKING: 

23 from collections.abc import Iterator 

24 

25StrictSoftFileClaimState = Literal["intent", "held"] 

26 

27_CLAIM_STATES: Final[frozenset[str]] = frozenset({"intent", "held"}) 

28_COORDINATION_SUFFIX: Final[str] = ".filelock" 

29_CLAIM_DIRECTORY_NAME: Final[str] = "claims" 

30_CLAIM_MAGIC: Final[str] = "filelock-strict-v1" 

31_CLAIM_RECORD_LIMIT: Final[int] = 1024 

32_CLAIM_NAME_PART_COUNT: Final[int] = 3 

33_TOKEN_HEX_LENGTH: Final[int] = 32 

34_PRIVATE_RECORD_MARKER: Final[str] = ".private-v1-" 

35_PRIVATE_RECORD_SUFFIX: Final[str] = ".tmp" 

36_PRIVATE_RECORD_RANDOM_HEX_LENGTH: Final[int] = 32 

37_PRIVATE_RECORD_GRACE: Final[float] = 2.0 

38_UNLINK_MAX_RETRIES: Final[int] = 10 

39#: How long a scan waits out a claim held in Windows' delete-pending state before treating it as unreadable. 

40_CLAIM_READ_GRACE: Final[float] = 0.5 

41_CLAIM_READ_RETRY: Final[float] = 0.002 

42#: Windows opens descriptors in text mode by default, which rewrites newlines and truncates a record at a control byte. 

43#: The claim and sentinel records are exact binary, so every record descriptor must be binary; POSIX ignores the flag. 

44_O_BINARY: Final[int] = getattr(os, "O_BINARY", 0) 

45_LEGACY_SENTINEL: Final[bytes] = STRICT_SOFT_SENTINEL_RECORD.encode() 

46_WINDOWS_HARD_LINK_UNSUPPORTED: Final[frozenset[int]] = frozenset({1, 17, 50}) 

47 

48# Termux/Android CPython ships without os.link (bionic long had only linkat), so the strict backend's whole hard-link 

49# mechanism is absent there. Probe once and gate every os.link reference on it, so importing filelock still works and 

50# only an actual StrictSoftFileLock acquire reports the missing capability. 

51_HAS_LINK: Final[bool] = hasattr(os, "link") 

52 

53# Probe dir_fd capability once at import. A per-call ``os.unlink in os.supports_dir_fd`` check flips to False the moment 

54# a test mocks os.unlink, silently diverting the code to a different branch than the one under test. 

55_OPEN_SUPPORTS_DIR_FD: Final[bool] = os.open in os.supports_dir_fd 

56_UNLINK_SUPPORTS_DIR_FD: Final[bool] = os.unlink in os.supports_dir_fd 

57_STAT_SUPPORTS_DIR_FD: Final[bool] = os.stat in os.supports_dir_fd 

58_LINK_SUPPORTS_DIR_FD: Final[bool] = _HAS_LINK and os.link in os.supports_dir_fd 

59 

60 

61def _probe_link_follow_symlinks() -> bool: 

62 # os.supports_follow_symlinks lists os.link on PyPy, but its linkat then rejects follow_symlinks=False with EINVAL, 

63 # and Windows raises NotImplementedError for the option outright. Link a throwaway file for real so the answer 

64 # reflects the runtime rather than its advertisement, and treat any failure as "not honored": the option only 

65 # hardens a source this process created with O_EXCL, so skipping it is safe, and a real environment fault surfaces 

66 # when the actual link runs. 

67 if not _HAS_LINK: 

68 return False 

69 try: 

70 with tempfile.TemporaryDirectory() as directory: 

71 source = Path(directory, "probe-source") 

72 source.touch() 

73 os.link(source, Path(directory, "probe-link"), follow_symlinks=False) 

74 except (OSError, NotImplementedError, ValueError): 

75 return False 

76 return True 

77 

78 

79_LINK_HONORS_FOLLOW_SYMLINKS: Final[bool] = _probe_link_follow_symlinks() 

80 

81 

82def _probe_hard_link_unsupported_errnos() -> frozenset[int]: 

83 # GraalPy's errno omits ENOTSUP, so importing the name outright breaks every runtime that ships without it. ENOTSUP 

84 # wins wherever it exists, leaving every runtime that names it unchanged. EOPNOTSUPP only stands in for the ones 

85 # that do not, and it approximates rather than matches: the two codes agree on Linux but differ on macOS/BSD and on 

86 # Windows. Where neither exists a runtime that cannot name "operation not supported" cannot raise it either, and 

87 # ENOSYS/EXDEV still classify the link failures it can raise. 

88 not_supported = getattr(errno, "ENOTSUP", getattr(errno, "EOPNOTSUPP", None)) 

89 return frozenset({ENOSYS, EXDEV} if not_supported is None else {ENOSYS, EXDEV, not_supported}) 

90 

91 

92_HARD_LINK_UNSUPPORTED_ERRNOS: Final[frozenset[int]] = _probe_hard_link_unsupported_errnos() 

93 

94 

95class StrictSoftFileLock(BaseFileLock): 

96 """Portable fail-closed lock based on immutable owner claims.""" 

97 

98 _preserve_lock_file_supported: bool = True 

99 _on_acquired_supported: bool = False 

100 #: Age cannot clear a strict claim: expiring one on a clock is the overlap the fail-closed contract exists to rule 

101 #: out, so only force_break() removes it. 

102 _lifetime_supported: bool = False 

103 _lifetime_unsupported_reason: str = "a strict claim is never broken by age, only by force_break()" 

104 #: The claim doorway publishes an intent and a held record per owner, so a shared instance must serialize them. 

105 _serialize_transitions: bool = True 

106 #: Contending processes each publish and rescan several files, so back their retries off across a jittered window 

107 #: rather than let them collide on every poll. Seconds; keeps a waiter responsive once it wins. 

108 _poll_backoff_cap: float = 0.05 

109 

110 def _acquire(self) -> None: 

111 # Resolve once per acquisition, not per poll: a waiter on a relative path must keep publishing into the 

112 # directory it started waiting in even when another thread changes the working directory mid-wait. 

113 if (claim_root := self._context.claim_root) is None: 

114 claim_root = self._context.claim_root = _canonical(self.lock_file) 

115 lock_path = Path(claim_root) 

116 coordination_directory = Path(f"{lock_path}{_COORDINATION_SUFFIX}") 

117 claim_directory = coordination_directory / _CLAIM_DIRECTORY_NAME 

118 ensure_directory_exists(os.fspath(lock_path)) 

119 _ensure_protocol_directory(self.lock_file, coordination_directory) 

120 _ensure_protocol_directory(self.lock_file, claim_directory) 

121 if (sentinel_fd := _open_or_create_sentinel(self.lock_file, lock_path, self._open_mode())) is None: 

122 return 

123 try: 

124 sentinel_identity = _file_identity(os.fstat(sentinel_fd)) 

125 except BaseException as inspection_error: # preserve inspection and descriptor cleanup errors 

126 try: 

127 os.close(sentinel_fd) 

128 except BaseException as close_error: # ruff:ignore[blind-except] # preserve inspection and descriptor cleanup errors 

129 _raise_cleanup_errors("strict sentinel inspection cleanup failed", inspection_error, close_error) 

130 raise 

131 self._mark_descriptor_pending(sentinel_fd, sentinel_identity) 

132 try: 

133 self._attempt_doorway(claim_directory, sentinel_fd, sentinel_identity) 

134 except BaseException: 

135 if self._context.pending_lock_file_fd == sentinel_fd: 

136 self._discard_doorway(sentinel_fd, sentinel_identity) 

137 raise 

138 

139 def _attempt_doorway(self, claim_directory: Path, sentinel_fd: int, sentinel_identity: tuple[int, int]) -> None: 

140 if _read_existing_claims(self.lock_file, claim_directory): 

141 self._discard_doorway(sentinel_fd, sentinel_identity) 

142 return 

143 

144 token = secrets.token_hex(_TOKEN_HEX_LENGTH // 2) 

145 intent_name = _claim_name("intent", token) 

146 intent_path = str(claim_directory / intent_name) 

147 try: 

148 publication_cleanup_error = _publish_record(intent_path, _claim_record(token), self._open_mode()) 

149 except _PrivateRecordReclaimedError: 

150 self._discard_doorway(sentinel_fd, sentinel_identity) 

151 return 

152 except (NotImplementedError, OSError) as error: 

153 _raise_if_hard_links_unsupported(self.lock_file, error) 

154 if isinstance(error, OSError) and error.errno == EEXIST: 

155 self._discard_doorway(sentinel_fd, sentinel_identity) 

156 return 

157 raise 

158 if publication_cleanup_error is not None: # pragma: needs dir-fd 

159 raise publication_cleanup_error 

160 self._context.owner_claim_paths = (intent_path,) 

161 

162 claims = _read_existing_claims(self.lock_file, claim_directory) 

163 if ( 

164 not claims 

165 or any(claim.state == "held" for claim in claims) 

166 or min(claim.name for claim in claims) != intent_name 

167 ): 

168 self._discard_doorway(sentinel_fd, sentinel_identity) 

169 return 

170 

171 held_name = _claim_name("held", token) 

172 held_path = str(claim_directory / held_name) 

173 try: 

174 link_cleanup_error = _link_no_replace(claim_directory, intent_name, held_name) 

175 except (NotImplementedError, OSError) as error: 

176 _raise_if_hard_links_unsupported(self.lock_file, error) 

177 raise 

178 self._context.owner_claim_paths = (held_path, intent_path) 

179 if link_cleanup_error is not None: # pragma: needs dir-fd 

180 self._context.owner_claim_paths = () 

181 raise link_cleanup_error 

182 

183 claims = _read_existing_claims(self.lock_file, claim_directory) 

184 if ( 

185 not {intent_name, held_name}.issubset(claim.name for claim in claims) 

186 or min(_claim_token_key(claim.name) for claim in claims) != f"v1-{token}.claim" 

187 ): 

188 self._discard_doorway(sentinel_fd, sentinel_identity) 

189 return 

190 # Keep the intent claim for the whole hold rather than unlinking it now. The intent has existed, unchanged, 

191 # since this owner published it, so a contender's os.scandir is guaranteed to return it (POSIX only leaves the 

192 # visibility of entries created or removed *during* a scan unspecified). The freshly linked held claim carries 

193 # no such guarantee: a scan that races its creation can miss it. Were the intent removed here, that scan could 

194 # observe neither claim and let a larger-token contender win over this owner. The stable intent is the witness 

195 # that keeps the phase-five min-token decision computed over the true set. Release unlinks both. 

196 self._mark_descriptor_owned(sentinel_fd, sentinel_identity) 

197 

198 @property 

199 def claims(self) -> tuple[StrictSoftFileClaim, ...]: 

200 """Published claims that block acquisition.""" 

201 return _read_existing_claims(self.lock_file, self._claim_directory) 

202 

203 def force_break(self, claim_name: str) -> None: 

204 """Remove one named claim, allowing overlap if its owner still holds the protected resource.""" 

205 _validate_force_break_name(claim_name) 

206 _require_exact_name(self._claim_directory, claim_name) 

207 if ( 

208 cleanup_error := _unlink_in_directory(self._claim_directory, claim_name) 

209 ) is not None: # pragma: needs dir-fd 

210 raise cleanup_error 

211 

212 def _rollback_failed_acquire(self, acquisition_error: BaseException) -> None: 

213 # _acquire already reconciles a failed doorway through _discard_doorway: it either closes the pending 

214 # descriptor or, when a held claim cannot be removed, commits it as owned so a later release retries and 

215 # raises the cleanup errors. A base rollback would release that owned descriptor again and report each 

216 # failure a second time, so leave the reconciled state alone. 

217 if self.is_locked: 

218 return 

219 super()._rollback_failed_acquire(acquisition_error) 

220 

221 def _reconcile_failed_acquire(self, canonical: str) -> None: 

222 # The acquisition is over, so the next one resolves the working directory again rather than reuse this one's. 

223 if not self.is_locked: 

224 self._context.claim_root = None 

225 super()._reconcile_failed_acquire(canonical) 

226 

227 def _release(self) -> None: 

228 fd = cast("int", self._context.lock_file_fd) 

229 self._context.claim_root = None 

230 remaining, errors = _unlink_owner_paths(self._context.owner_claim_paths) 

231 self._context.owner_claim_paths = tuple(remaining) 

232 if remaining: 

233 _raise_recorded_errors("strict claim release failed", errors) 

234 self._mark_descriptor_released() 

235 try: 

236 self._close_released_fd(fd, default_suppresses=False) 

237 except BaseException as close_error: # ruff:ignore[blind-except] # preserve claim and sentinel cleanup errors 

238 errors.append(close_error) 

239 if errors: 

240 _raise_recorded_errors("strict release cleanup failed", errors) 

241 

242 def _discard_doorway(self, fd: int, identity: tuple[int, int]) -> None: 

243 remaining, errors = _unlink_owner_paths(self._context.owner_claim_paths) 

244 self._context.owner_claim_paths = tuple(remaining) 

245 if remaining: 

246 self._mark_descriptor_owned(fd, identity) 

247 _raise_recorded_errors("strict doorway claim cleanup failed", errors) 

248 self._mark_descriptor_released() 

249 try: 

250 self._close_released_fd(fd, default_suppresses=False) 

251 except BaseException as close_error: # ruff:ignore[blind-except] # preserve claim and sentinel cleanup errors 

252 errors.append(close_error) 

253 if errors: 

254 _raise_recorded_errors("strict doorway cleanup failed", errors) 

255 

256 @property 

257 def _claim_directory(self) -> Path: 

258 if self._context.owner_claim_paths: 

259 return Path(self._context.owner_claim_paths[0]).parent 

260 return Path(f"{_canonical(self.lock_file)}{_COORDINATION_SUFFIX}") / _CLAIM_DIRECTORY_NAME 

261 

262 

263@dataclass(frozen=True) 

264class StrictSoftFileClaim: 

265 """One parsed strict soft-lock claim.""" 

266 

267 name: str 

268 state: StrictSoftFileClaimState 

269 token: str 

270 pid: int 

271 hostname: str 

272 #: The owner's process start token, or ``None`` when the platform exposes no proven start time. A strict lock never 

273 #: reclaims a claim on its own, so this identifies the owner for tooling rather than driving any automatic break. 

274 start: int | None = None 

275 

276 

277class _PrivateRecordReclaimedError(Exception): 

278 pass 

279 

280 

281def _open_or_create_sentinel(lock_file: str, path: Path, mode: int) -> int | None: 

282 try: 

283 return _open_sentinel(path) 

284 except FileNotFoundError: 

285 pass 

286 except OSError: 

287 return None 

288 

289 _reclaim_sentinel_private_records(path, time.time()) 

290 try: 

291 publication_cleanup_error = _publish_record(os.fspath(path), _LEGACY_SENTINEL, mode) 

292 except _PrivateRecordReclaimedError: 

293 return None 

294 except (NotImplementedError, OSError) as error: 

295 _raise_if_hard_links_unsupported(lock_file, error) 

296 if not isinstance(error, OSError) or error.errno != EEXIST: 

297 raise 

298 else: 

299 if publication_cleanup_error is not None: # pragma: needs dir-fd 

300 raise publication_cleanup_error 

301 try: 

302 return _open_sentinel(path) 

303 except OSError: 

304 return None 

305 

306 

307def _open_sentinel(path: Path) -> int | None: 

308 fd, record = _open_record(path, len(_LEGACY_SENTINEL)) 

309 if record == _LEGACY_SENTINEL: 

310 return fd 

311 os.close(fd) 

312 return None 

313 

314 

315def _read_claims(lock_file: str, directory: Path) -> tuple[StrictSoftFileClaim, ...]: 

316 try: 

317 with os.scandir(directory) as entries: 

318 names = _public_claim_names(directory, entries) 

319 except OSError as error: 

320 reason = f"cannot list claim directory: {error.strerror or type(error).__name__}" 

321 raise SoftFileLockProtocolError(lock_file, None, reason) from error 

322 

323 claims: list[StrictSoftFileClaim] = [] 

324 for name in names: 

325 if (name_parts := _parse_claim_name(name)) is None: 

326 raise SoftFileLockProtocolError(lock_file, name, "unknown claim name or protocol version") 

327 if (record := _read_claim_record(lock_file, directory, name)) is not None: 

328 claims.append(_parse_claim(lock_file, name, name_parts, record)) 

329 return tuple(claims) 

330 

331 

332def _read_claim_record(lock_file: str, directory: Path, name: str) -> bytes | None: 

333 # A contended scan can list a claim that is not yet cleanly readable, in two ways that both resolve on a brief 

334 # retry. Windows holds a claim mid-unlink in a delete-pending state that fails an open with EACCES until the unlink 

335 # completes. On NFS, a peer that unlinks its own claim leaves this client's cached filehandle stale, so the next 

336 # open returns ESTALE rather than a clean ENOENT until the lookup revalidates against the server. Retrying re-runs 

337 # the path lookup, which turns the vanished claim into ENOENT (skip) or reads it if it still exists. A genuinely 

338 # unreadable record (a locked-down file, an EIO fault) still fails closed. 

339 deadline = time.monotonic() + _CLAIM_READ_GRACE 

340 delaying = False 

341 while True: 

342 if delaying: 

343 time.sleep(_CLAIM_READ_RETRY) 

344 record, pending = _attempt_claim_read(lock_file, directory, name) 

345 if pending is None: 

346 return record 

347 if time.monotonic() >= deadline: 

348 if pending.errno == ESTALE: 

349 # A stale handle that outlives revalidation is a claim the server no longer has (RFC 1813 

350 # NFS3ERR_STALE): skip it like ENOENT. Skipping a peer's vanished claim can only overcount 

351 # contention, never free a held lock. 

352 return None 

353 reason = f"cannot read claim: {pending.strerror or str(pending) or type(pending).__name__}" 

354 raise SoftFileLockProtocolError(lock_file, name, reason) from pending 

355 delaying = True 

356 

357 

358def _attempt_claim_read(lock_file: str, directory: Path, name: str) -> tuple[bytes | None, OSError | None]: 

359 # Return the record, or (None, None) when the claim has already gone, or (None, error) for an open the caller may 

360 # retry: a Windows delete-pending EACCES that resolves to a clean removal, or an NFS ESTALE from a peer unlinking 

361 # its own claim out from under this client's cached filehandle. Any other OSError is a real fault and fails closed. 

362 try: 

363 return _read_record(directory / name, _CLAIM_RECORD_LIMIT), None 

364 except FileNotFoundError: 

365 return None, None 

366 except PermissionError as error: 

367 return None, error 

368 except OSError as error: 

369 if error.errno == ESTALE: 

370 return None, error 

371 reason = f"cannot read claim: {error.strerror or str(error) or type(error).__name__}" 

372 raise SoftFileLockProtocolError(lock_file, name, reason) from error 

373 

374 

375def _public_claim_names(directory: Path, entries: Iterator[os.DirEntry[str]]) -> list[str]: 

376 names: list[str] = [] 

377 now = time.time() 

378 for entry in entries: 

379 if not entry.name.startswith("."): 

380 names.append(entry.name) 

381 elif (public_name := _private_public_name(entry.name)) is not None and _parse_claim_name( 

382 public_name 

383 ) is not None: 

384 _reclaim_private_record((os.fspath(directory), None), entry.name, now) 

385 return sorted(names) 

386 

387 

388def _read_existing_claims(lock_file: str, directory: Path) -> tuple[StrictSoftFileClaim, ...]: 

389 if not directory.exists(): 

390 return () 

391 return _read_claims(lock_file, directory) 

392 

393 

394def _parse_claim( 

395 lock_file: str, 

396 name: str, 

397 name_parts: tuple[StrictSoftFileClaimState, str], 

398 record: bytes, 

399) -> StrictSoftFileClaim: 

400 try: 

401 magic, token, pid_text, hostname_hex, start_text, trailing = record.decode("ascii").split("\n") 

402 pid = int(pid_text) 

403 start = int(start_text) if start_text else None 

404 hostname = bytes.fromhex(hostname_hex).decode("utf-8") 

405 except (UnicodeDecodeError, ValueError) as error: 

406 raise SoftFileLockProtocolError(lock_file, name, "malformed claim record") from error 

407 if not all(( 

408 not trailing, 

409 magic == _CLAIM_MAGIC, 

410 token == name_parts[1], 

411 1 <= pid <= 2**31 - 1, 

412 str(pid) == pid_text, 

413 start is None or (start >= 0 and str(start) == start_text), 

414 hostname.encode().hex() == hostname_hex 

415 and hostname.isprintable() 

416 and not any(character.isspace() for character in hostname), 

417 )): 

418 raise SoftFileLockProtocolError(lock_file, name, "malformed claim record") 

419 return StrictSoftFileClaim(name=name, state=name_parts[0], token=token, pid=pid, hostname=hostname, start=start) 

420 

421 

422def _parse_claim_name(name: str) -> tuple[StrictSoftFileClaimState, str] | None: 

423 if not name.endswith(".claim"): 

424 return None 

425 parts = name.removesuffix(".claim").split("-") 

426 if len(parts) != _CLAIM_NAME_PART_COUNT or parts[0] not in _CLAIM_STATES or parts[1] != "v1": 

427 return None 

428 token = parts[2] 

429 if len(token) != _TOKEN_HEX_LENGTH or any(character not in "0123456789abcdef" for character in token): 

430 return None 

431 return cast("StrictSoftFileClaimState", parts[0]), token 

432 

433 

434def _claim_name(state: StrictSoftFileClaimState, token: str) -> str: 

435 return f"{state}-v1-{token}.claim" 

436 

437 

438def _claim_token_key(name: str) -> str: 

439 return name.removeprefix("held-").removeprefix("intent-") 

440 

441 

442def _claim_record(token: str) -> bytes: 

443 hostname_hex = host_name().encode().hex() 

444 start = process_start_token(os.getpid()) 

445 start_text = "" if start is None else str(start) 

446 return f"{_CLAIM_MAGIC}\n{token}\n{os.getpid()}\n{hostname_hex}\n{start_text}\n".encode("ascii") 

447 

448 

449def _publish_record( 

450 public_path: str, 

451 record: bytes, 

452 mode: int, 

453) -> BaseException | None: 

454 directory, public_name = os.path.split(public_path) 

455 directory = directory or os.curdir 

456 private_name = _private_record_name(public_name) 

457 directory_fd = _open_directory(directory) if _OPEN_SUPPORTS_DIR_FD else None 

458 directory_ref = directory, directory_fd 

459 try: 

460 _publish_record_in_directory(directory_ref, (private_name, public_name), mode, record) 

461 except BaseException as publication_error: # preserve publication and directory cleanup errors 

462 try: 

463 if directory_fd is not None: # pragma: needs dir-fd 

464 os.close(directory_fd) 

465 except BaseException as close_error: # ruff:ignore[blind-except] # pragma: needs dir-fd # preserve publication and directory cleanup errors 

466 _raise_cleanup_errors("strict publication directory cleanup failed", publication_error, close_error) 

467 raise 

468 if directory_fd is not None: # pragma: needs dir-fd 

469 try: 

470 os.close(directory_fd) 

471 except BaseException as close_error: # ruff:ignore[blind-except] # caller records the published path before raising 

472 return close_error 

473 return None 

474 

475 

476def _publish_record_in_directory( 

477 directory_ref: tuple[str, int | None], 

478 names: tuple[str, str], 

479 mode: int, 

480 record: bytes, 

481) -> None: 

482 flags = os.O_RDWR | os.O_CREAT | os.O_EXCL | _O_BINARY 

483 if (o_nofollow := getattr(os, "O_NOFOLLOW", None)) is not None: # pragma: needs o-nofollow 

484 flags |= o_nofollow 

485 private_fd = _open_relative(directory_ref, names[0], flags, mode) 

486 private_identity: tuple[int, int] | None = None 

487 try: 

488 private_identity = _file_identity(os.fstat(private_fd)) 

489 write_all(private_fd, record) 

490 _link_private_record(directory_ref, names, private_identity) 

491 except BaseException as publication_error: # preserve publication and cleanup errors 

492 close_error, unlink_error = _close_and_unlink_private_record( 

493 directory_ref, 

494 names[0], 

495 private_fd, 

496 private_identity, 

497 ) 

498 if close_error is not None or unlink_error is not None: 

499 _raise_cleanup_errors( 

500 "strict record publication cleanup failed", 

501 publication_error, 

502 close_error, 

503 unlink_error, 

504 ) 

505 raise 

506 close_error, unlink_error = _close_and_unlink_private_record( 

507 directory_ref, 

508 names[0], 

509 private_fd, 

510 private_identity, 

511 ) 

512 if close_error is not None or unlink_error is not None: 

513 _raise_record_finalization_errors(close_error, unlink_error) 

514 

515 

516def _close_and_unlink_private_record( 

517 directory_ref: tuple[str, int | None], 

518 private_name: str, 

519 private_fd: int, 

520 private_identity: tuple[int, int] | None, 

521) -> tuple[BaseException | None, BaseException | None]: 

522 close_error: BaseException | None = None 

523 try: 

524 os.close(private_fd) 

525 except BaseException as error: # ruff:ignore[blind-except] # returned for grouping with unlink failures 

526 close_error = error 

527 unlink_error: BaseException | None = None 

528 try: 

529 if private_identity is None: 

530 _unlink_relative(directory_ref, private_name) 

531 else: 

532 _unlink_relative_if_identity(directory_ref, private_name, private_identity) 

533 except FileNotFoundError: 

534 pass 

535 except BaseException as error: # ruff:ignore[blind-except] # returned for grouping with close failures 

536 unlink_error = error 

537 return close_error, unlink_error 

538 

539 

540def _link_private_record( 

541 directory_ref: tuple[str, int | None], 

542 names: tuple[str, str], 

543 private_identity: tuple[int, int], 

544) -> None: 

545 try: 

546 _link_relative(directory_ref, *names) 

547 except FileNotFoundError as error: 

548 if _relative_identity(directory_ref, names[0]) is not None: 

549 raise 

550 msg = "private publication record was reclaimed" 

551 raise _PrivateRecordReclaimedError(msg) from error 

552 if _relative_identity(directory_ref, names[1]) == private_identity: 

553 return 

554 msg_0 = "private publication record was replaced" 

555 raise _PrivateRecordReclaimedError(msg_0) 

556 

557 

558def _raise_record_finalization_errors( 

559 close_error: BaseException | None, 

560 unlink_error: BaseException | None, 

561) -> None: 

562 errors = [error for error in (close_error, unlink_error) if error is not None] 

563 if len(errors) > 1: 

564 _raise_cleanup_errors("strict record finalization failed", errors[0], *errors[1:]) 

565 raise errors[0] 

566 

567 

568def _private_record_name(public_name: str) -> str: 

569 return f".{public_name}{_PRIVATE_RECORD_MARKER}{secrets.token_hex(_PRIVATE_RECORD_RANDOM_HEX_LENGTH // 2)}.tmp" 

570 

571 

572def _private_public_name(private_name: str) -> str | None: 

573 if not private_name.startswith(".") or not private_name.endswith(_PRIVATE_RECORD_SUFFIX): 

574 return None 

575 public_name, marker, random_hex = private_name[1 : -len(_PRIVATE_RECORD_SUFFIX)].rpartition(_PRIVATE_RECORD_MARKER) 

576 if ( 

577 marker != _PRIVATE_RECORD_MARKER 

578 or len(random_hex) != _PRIVATE_RECORD_RANDOM_HEX_LENGTH 

579 or any(character not in "0123456789abcdef" for character in random_hex) 

580 ): 

581 return None 

582 return public_name 

583 

584 

585def _reclaim_sentinel_private_records(path: Path, now: float) -> None: 

586 directory_ref = os.fspath(path.parent), None 

587 with os.scandir(path.parent) as entries: 

588 for entry in entries: 

589 if _private_public_name(entry.name) == path.name: 

590 _reclaim_private_record(directory_ref, entry.name, now) 

591 

592 

593def _reclaim_private_record(directory_ref: tuple[str, int | None], private_name: str, now: float) -> None: 

594 directory, directory_fd = directory_ref 

595 try: 

596 private_stat = ( 

597 os.stat(private_name, dir_fd=directory_fd, follow_symlinks=False) 

598 if directory_fd is not None and _STAT_SUPPORTS_DIR_FD 

599 else Path(directory, private_name).lstat() 

600 ) 

601 except FileNotFoundError: 

602 return 

603 if not stat.S_ISREG(private_stat.st_mode): 

604 msg = f"{Path(directory, private_name)} is not a regular private record" 

605 raise OSError(msg) 

606 if private_stat.st_nlink == 1 and now - private_stat.st_mtime < _PRIVATE_RECORD_GRACE: 

607 return 

608 try: 

609 _unlink_private_record_once(directory_ref, private_name) 

610 except FileNotFoundError: 

611 pass 

612 except OSError as error: 

613 if error.errno not in {EACCES, EPERM}: 

614 raise 

615 

616 

617def _unlink_private_record_once(directory_ref: tuple[str, int | None], private_name: str) -> None: 

618 directory, directory_fd = directory_ref 

619 if ( 

620 directory_fd is not None and _UNLINK_SUPPORTS_DIR_FD 

621 ): # pragma: no cover # callers always pass directory_fd=None 

622 os.unlink(private_name, dir_fd=directory_fd) 

623 else: 

624 Path(directory, private_name).unlink() 

625 

626 

627def _relative_identity(directory_ref: tuple[str, int | None], name: str) -> tuple[int, int] | None: 

628 directory, directory_fd = directory_ref 

629 try: 

630 if directory_fd is not None and _STAT_SUPPORTS_DIR_FD: # pragma: needs dir-fd 

631 path_stat = os.stat(name, dir_fd=directory_fd, follow_symlinks=False) 

632 else: # pragma: win32 cover 

633 path_stat = Path(directory, name).lstat() 

634 except FileNotFoundError: 

635 return None 

636 return _file_identity(path_stat) 

637 

638 

639def _unlink_relative_if_identity( 

640 directory_ref: tuple[str, int | None], 

641 name: str, 

642 identity: tuple[int, int], 

643) -> None: 

644 if _relative_identity(directory_ref, name) == identity: 

645 with contextlib.suppress(FileNotFoundError): 

646 _unlink_relative(directory_ref, name) 

647 

648 

649def _open_relative(directory_ref: tuple[str, int | None], name: str, flags: int, mode: int) -> int: 

650 directory, directory_fd = directory_ref 

651 return ( 

652 os.open(name, flags, mode, dir_fd=directory_fd) 

653 if directory_fd is not None 

654 else os.open(Path(directory, name), flags, mode) 

655 ) 

656 

657 

658def _link_relative(directory_ref: tuple[str, int | None], source_name: str, destination_name: str) -> None: 

659 directory, directory_fd = directory_ref 

660 if directory_fd is not None and _LINK_SUPPORTS_DIR_FD: # pragma: needs dir-fd 

661 _link_no_follow(source_name, destination_name, src_dir_fd=directory_fd, dst_dir_fd=directory_fd) 

662 return 

663 _link_no_follow(Path(directory, source_name), Path(directory, destination_name)) # pragma: win32 cover 

664 

665 

666def _link_no_follow( 

667 source: str | Path, 

668 destination: str | Path, 

669 *, 

670 src_dir_fd: int | None = None, 

671 dst_dir_fd: int | None = None, 

672) -> None: 

673 if not _HAS_LINK: 

674 # No os.link at all (Termux/Android): report it as unsupported like a filesystem that refuses hard links, so 

675 # the acquire path raises SoftFileLockProtocolError instead of a bare AttributeError. 

676 msg = "os.link is unavailable on this platform" 

677 raise NotImplementedError(msg) 

678 # The source is a private record this process created with O_CREAT | O_EXCL, so follow_symlinks guards nothing an 

679 # attacker can reach. Pass the option only when the runtime honors it: PyPy advertises it through 

680 # os.supports_follow_symlinks yet its linkat rejects it with EINVAL, so probe once rather than trust the set. 

681 if _LINK_HONORS_FOLLOW_SYMLINKS: 

682 os.link(source, destination, src_dir_fd=src_dir_fd, dst_dir_fd=dst_dir_fd, follow_symlinks=False) 

683 else: # pragma: lacks link-follow-symlinks 

684 os.link(source, destination, src_dir_fd=src_dir_fd, dst_dir_fd=dst_dir_fd) 

685 

686 

687def _unlink_relative(directory_ref: tuple[str, int | None], name: str) -> None: 

688 directory, directory_fd = directory_ref 

689 if directory_fd is not None and _UNLINK_SUPPORTS_DIR_FD: # pragma: needs dir-fd 

690 os.unlink(name, dir_fd=directory_fd) 

691 elif sys.platform == "win32": # pragma: win32 cover 

692 _unlink_in_directory(Path(directory), name) 

693 else: 

694 Path(directory, name).unlink() # pragma: win32 no cover 

695 

696 

697def _link_no_replace(directory: Path, source_name: str, destination_name: str) -> BaseException | None: 

698 if _LINK_SUPPORTS_DIR_FD: # pragma: needs dir-fd 

699 directory_fd = _open_directory(str(directory)) 

700 try: 

701 _link_no_follow(source_name, destination_name, src_dir_fd=directory_fd, dst_dir_fd=directory_fd) 

702 except BaseException as link_error: # preserve link and directory cleanup errors 

703 try: 

704 os.close(directory_fd) 

705 except BaseException as close_error: # ruff:ignore[blind-except] # preserve link and directory cleanup errors 

706 _raise_cleanup_errors("strict link directory cleanup failed", link_error, close_error) 

707 raise 

708 try: 

709 os.close(directory_fd) 

710 except BaseException as close_error: # ruff:ignore[blind-except] # caller records the held path before raising 

711 return close_error 

712 return None 

713 _link_no_follow(directory / source_name, directory / destination_name) # pragma: win32 cover 

714 return None # pragma: win32 cover 

715 

716 

717def _unlink_owner_path(path: str) -> BaseException | None: 

718 directory, name = os.path.split(path) 

719 return _unlink_in_directory(Path(directory or os.curdir), name) 

720 

721 

722def _unlink_owner_paths(paths: tuple[str, ...]) -> tuple[list[str], list[BaseException]]: 

723 results = [(path, _unlink_owner_path_result(path)) for path in paths] 

724 return [path for path, (removed, _) in results if not removed], [ 

725 error for _, (_, error) in results if error is not None 

726 ] 

727 

728 

729def _unlink_owner_path_result(path: str) -> tuple[bool, BaseException | None]: 

730 try: 

731 cleanup_error = _unlink_owner_path(path) 

732 except FileNotFoundError: 

733 return True, None 

734 except BaseException as error: # ruff:ignore[blind-except] # keep ownership when unlink did not commit 

735 return False, error 

736 return True, cleanup_error 

737 

738 

739def _raise_recorded_errors(message: str, errors: list[BaseException]) -> None: 

740 if len(errors) > 1: 

741 _raise_cleanup_errors(message, errors[0], *errors[1:])