This repository was archived by the owner on Jan 21, 2026. It is now read-only.
-
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathsitecustomize.py
More file actions
77 lines (56 loc) · 2.14 KB
/
sitecustomize.py
File metadata and controls
77 lines (56 loc) · 2.14 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
"""Project-wide runtime tweaks loaded by Python at startup.
Currently forces urllib3 clients (e.g. pip-audit via requests) to avoid
opportunistic HTTP/3 upgrades that intermittently fail against some
vulnerability APIs. Set ``ACB_HTTP3_DISABLE=0`` to opt back in.
"""
from __future__ import annotations
import os
from typing import Any
_PATCH_FLAG = "_acb_http3_patch"
_HTTP3_SENTINEL = None
def _should_disable_http3() -> bool:
raw_value = os.environ.get("ACB_HTTP3_DISABLE", "1").strip().lower()
return raw_value not in {"0", "false", "no", "off"}
def _with_h3_disabled(values: dict[str, Any]) -> dict[str, Any]:
sentinel = _HTTP3_SENTINEL
if sentinel is None:
return values
disabled = values.get("disabled_svn")
disabled_set = set(disabled or ())
disabled_set.add(sentinel)
values["disabled_svn"] = disabled_set
return values
def _patch_init(cls: type) -> None:
original_init = getattr(cls, "__init__", None)
if original_init is None or getattr(original_init, _PATCH_FLAG, False):
return
def wrapped(self, *args: Any, **kwargs: Any) -> None:
kwargs = dict(kwargs)
_with_h3_disabled(kwargs)
original_init(self, *args, **kwargs)
setattr(wrapped, _PATCH_FLAG, True)
cls.__init__ = wrapped # type: ignore[assignment]
if _should_disable_http3():
try:
from urllib3 import connectionpool, poolmanager
from urllib3.backend import HttpVersion
except Exception:
pass
else:
_HTTP3_SENTINEL = HttpVersion.h3
for _candidate in (
poolmanager.PoolManager,
connectionpool.HTTPConnectionPool,
connectionpool.HTTPSConnectionPool,
):
_patch_init(_candidate)
def _patch_crackerjack_skylos_timeout() -> None:
try:
from crackerjack.config import hooks as cj_hooks
except Exception:
return
timeout_value = int(os.environ.get("ACB_CRACKERJACK_SKYLOS_TIMEOUT", "600"))
for hook in cj_hooks.COMPREHENSIVE_HOOKS:
if hook.name == "skylos" and hook.timeout < timeout_value:
hook.timeout = timeout_value
_patch_crackerjack_skylos_timeout()