用 sitecustomize.py 给 Python 应用做无侵入运行时补丁
什么是 sitecustomize.py
Python 解释器启动时，site 模块会在初始化的最后阶段尝试 import sitecustomize。只要这个模块出现在 sys.path 能搜到的任意位置（通常放进 site-packages，或直接扔到工作目录、再用 PYTHONPATH 指过去），它就会被自动执行——早于你的主程序、早于任何第三方库真正干活。
这带来一个非常有用的能力：
在不修改任何一行业务代码的前提下，往一个已经打好包、甚至被加密混淆过的 Python 应用里，注入你自己的运行时逻辑。
典型场景包括：
- 修复被打包工具搞丢的 sys.path；
- 替换某个第三方模块里的类 / 函数（比如授权校验、埋点、限流）；
- 给底层 HTTP 库统一注入超时、重试、代理配置；
- 在进程最开始就装好日志 / 监控 / 采样钩子。
下面这段脚本把上述几件事都做了。我们逐块拆开看它的设计。
完整脚本
# sitecustomize.py
import builtins
import sys
import threading
import os
# =============================================================================
# 1. 强力双重防重入锁机制
# =============================================================================
# 锁 A（全局骨架锁）：防止因为多次加载导致的多次 Hook
if not hasattr(builtins, "_original_import_backup"):
# 备份最原始的系统导入方法
builtins._original_import_backup = builtins.__import__
# 锁 B（线程函数锁）：防止解密过程中触发隐式导入造成的无限递归
_import_recursion_lock = threading.local()
print("[Patch] 正在注入终极防重入沙箱补丁...")
# =============================================================================
# 2. 自动修复路径
# =============================================================================
CRITICAL_PATHS = ["/opt/app", "/opt/app/service"]
for path in CRITICAL_PATHS:
if path not in sys.path:
sys.path.insert(0, path)
# =============================================================================
# 3. 构造伪造的 PRO 版授权类
# =============================================================================
class MockLicenseInfo:
def __init__(self, *args, **kwargs):
pass
def is_free(self):
return False # 核心：永远不触发免费版的 sys.exit(1)
def get_type(self):
return "PRO"
# =============================================================================
# 3.5 直连超时补丁(后端无关 + 自证后端)
# 根因:取 115 直连的请求读超时默认 60s;115 偶发不响应时单请求卡满 60s,
# 请求堆积 -> 占满 app 共享资源 -> 整个后端服务僵死(连 /api 都超时),只能重启。
# p115client 可用多种 HTTP 后端,默认 urllib3_future_request,也可能是 httpx_request。
# 这里把这两个后端的 request 函数各包一层:① 首次调用时打印实际命中哪个后端;
# ② 强制注入读超时,卡住的请求 ~Ns 内快速失败 -> 不再雪崩僵死。
# 可用环境变量 P115_READ_TIMEOUT 调整(秒),默认 8。
# =============================================================================
try:
_P115_READ_TIMEOUT = float(os.environ.get("P115_READ_TIMEOUT", "8"))
except Exception:
_P115_READ_TIMEOUT = 8.0
try:
_P115_RETRIES = int(os.environ.get("P115_RETRIES", "0"))
except Exception:
_P115_RETRIES = 0
_backend_logged = {}
def _mk_httpx_timeout():
from httpx import Timeout
return Timeout(connect=5, read=_P115_READ_TIMEOUT,
write=_P115_READ_TIMEOUT, pool=5)
def _mk_urllib3_timeout():
from urllib3_future import Timeout
return Timeout(connect=5, read=_P115_READ_TIMEOUT)
# 每后端额外强制的 kwargs(retries 只对 urllib3 有意义;httpx 不认这个参数)
def _extra_urllib3():
# 直接用 int(实测 retries=0 即可让读超时 8s 后立刻失败,不再 ×4 重试)
return {"retries": _P115_RETRIES}
def _extra_none():
return {}
def _wrap_backend_request(modname, make_timeout, make_extra):
try:
module = builtins._original_import_backup(modname)
except Exception as e:
print("[Patch] 后端 %s 未安装/导入失败: %s" % (modname, e))
return
orig = getattr(module, "request", None)
if orig is None or getattr(orig, "_p115_to_wrapped", False):
return
try:
to = make_timeout()
extra = make_extra()
except Exception as e:
print("[Patch] %s 构造 timeout/extra 失败: %s" % (modname, e))
return
def wrapped(*a, **kw):
if not _backend_logged.get(modname):
_backend_logged[modname] = True
print("[Patch] >>> 115 HTTP 后端实际命中 = %s (强制 read=%ss, retries=%s) <<<"
% (modname, _P115_READ_TIMEOUT, _P115_RETRIES))
kw["timeout"] = to # 强制覆盖,防 app/p115client 自己传 60s 或 None
for _k, _v in extra.items():
kw[_k] = _v
return orig(*a, **kw)
wrapped._p115_to_wrapped = True
try:
module.request = wrapped
print("[Patch] 已包裹后端 %s.request,注入读超时 %ss" % (modname, _P115_READ_TIMEOUT))
except Exception as e:
print("[Patch] 包裹 %s 失败: %s" % (modname, e))
# =============================================================================
# 4. 安全无感知的拦截核心
# =============================================================================
def patched_import(name, *args, **kwargs):
# 如果当前线程已经在执行导入流（说明是内部解密触发的二次导入），直接放行走原版
if getattr(_import_recursion_lock, "is_loading", False):
return builtins._original_import_backup(name, *args, **kwargs)
# 激活重入锁
_import_recursion_lock.is_loading = True
try:
# 调用原生的导入，让底层安全地完成它的解密和加载工作
module = builtins._original_import_backup(name, *args, **kwargs)
finally:
# 释放重入锁
_import_recursion_lock.is_loading = False
# 当模块安全加载完毕返回时，我们再悄悄修改属性
if module:
if hasattr(module, 'LicenseInfo'):
try:
setattr(module, 'LicenseInfo', MockLicenseInfo)
except:
pass
if hasattr(module, 'check_donate_code'):
try:
setattr(module, 'check_donate_code', lambda *args, **kwargs: None)
except:
pass
return module
# 将安全的拦截器回写至内置内核
builtins.__import__ = patched_import
# 直连超时补丁:预加载并包裹两个候选 HTTP 后端的 request(谁被命中谁生效)。
# 不依赖上面的 import hook(后端多为嵌套导入,会被防重入锁跳过)。
_wrap_backend_request("urllib3_future_request", _mk_urllib3_timeout, _extra_urllib3)
_wrap_backend_request("httpx_request", _mk_httpx_timeout, _extra_none)
print("[Patch] 路径修复与安全拦截沙箱部署完毕，等待主程序拉起...")
拆解四个关键技巧
技巧一：双重防重入锁
sitecustomize 有可能被多次加载（多进程、reload、被打包器重复触发），而脚本里又改写了 builtins.__import__ 这种全局对象。如果不加保护，就会出现「补丁打补丁」甚至无限递归。脚本用了两把锁来兜底：
-
锁 A（骨架锁）：
if not hasattr(builtins, "_original_import_backup")。第一次进来时把最原始的__import__备份到builtins上；之后再次加载看到这个属性已存在，整块逻辑直接跳过。这保证了「原始导入函数只被备份一次」，不会出现备份了「已经被包过一层的版本」的悲剧。 -
锁 B（线程递归锁）：
threading.local()。这是重点——一旦我们把__import__换成了自己的patched_import，那么在patched_import内部再调用底层导入时，如果底层库（比如被加密混淆的模块在解密阶段）又隐式触发了import，就会再次进入patched_import，形成递归。用线程局部变量is_loading作标记：进入导入流时置True，此时任何嵌套导入都直接走原版、不再做属性替换，出来时置False。既避免了递归，也避免了对解密过程中的中间态模块乱动手脚。
用 threading.local() 而不是普通全局布尔量，是为了让每个线程各自独立计数，多线程并发导入时互不干扰。
技巧二：修复 sys.path
CRITICAL_PATHS = ["/opt/app", "/opt/app/service"]
for path in CRITICAL_PATHS:
if path not in sys.path:
sys.path.insert(0, path)
打包 / 容器化后的应用经常出现「本地跑得好好的，进了容器就 ModuleNotFoundError」，根因往往是工作目录或源码目录没进 sys.path。因为 sitecustomize 执行时机早于主程序，在这里 insert(0, ...) 把关键目录塞到搜索路径最前面，后续所有 import 都能正确定位。这是最朴素、也最常见的一类用途。
技巧三：import hook —— 加载后替换模块属性
这是整段脚本的核心。思路是：不阻止导入，只在导入完成后动手改。
module = builtins._original_import_backup(name, *args, **kwargs)
...
if hasattr(module, 'LicenseInfo'):
setattr(module, 'LicenseInfo', MockLicenseInfo)
先让原生 __import__ 把模块正常加载完（对被加密混淆的模块尤其重要，解密逻辑必须让它自己跑完），拿到 module 对象后，再检查它身上有没有我们关心的符号——例如授权信息类 LicenseInfo、捐赠码校验函数 check_donate_code——有就替换成我们的桩实现。
桩类 MockLicenseInfo 的 is_free() 恒返回 False、get_type() 恒返回 "PRO"，从而绕开原逻辑里「免费版就 sys.exit(1)」的分支；check_donate_code 被换成一个什么都不做的 lambda。
这种「加载后打补丁」的模式比「拦截并伪造整个模块」要稳健得多：
- 不需要理解目标模块的内部实现，只在它的公开属性上做最小改动；
- 对混淆 / 加密代码友好，因为真正的加载、解密都交还给原生机制；
- try/except 包住每次 setattr，即使目标符号是只读的也不会把整个进程带崩。
注意
__import__hook 只能拦到「顶层由它触发的那一次导入」。很多库在函数内部才import，或者用 C 扩展、importlib直接加载，就绕过了这层钩子——这也正是下一个技巧存在的原因。
技巧四：monkey patch —— 给底层 HTTP 后端强制注入超时
脚本注释里记录了一个非常真实的线上事故：某网盘直连请求的读超时默认是 60s，对端偶发不响应时，单个请求会卡满 60s，请求不断堆积、占满共享资源，最终整个后端服务僵死，连健康检查接口都超时，只能重启。
修复的思路不是去改业务代码里每一处请求调用，而是在最底层的 HTTP 后端函数上包一层：
def _wrap_backend_request(modname, make_timeout, make_extra):
module = builtins._original_import_backup(modname)
orig = getattr(module, "request", None)
if orig is None or getattr(orig, "_p115_to_wrapped", False):
return
...
def wrapped(*a, **kw):
kw["timeout"] = to # 强制覆盖调用方传进来的 60s / None
for _k, _v in extra.items():
kw[_k] = _v
return orig(*a, **kw)
wrapped._p115_to_wrapped = True
module.request = wrapped
这里有几个值得学习的细节：
-
后端无关 + 自证后端。目标客户端可能用
urllib3_future也可能用httpx作为 HTTP 后端，脚本把两个候选后端的request都包一层，谁被真正调用谁生效；wrapped首次执行时打印一行日志，让你在运行时确认「究竟命中了哪个后端」，省掉大量猜测。 -
强制覆盖而非补默认值。
kw["timeout"] = to是直接覆盖，哪怕业务代码或库自己传了timeout=60或None，也会被我们的短超时顶掉。这才是「兜底」——不给上游留后门。 -
幂等标记
_p115_to_wrapped。包好后在函数上打个属性标记，重复执行时看到标记就跳过，防止「包了又包」叠加多层。这和技巧一的骨架锁是同一种防御思想。 -
参数可调。读超时、重试次数都从环境变量读取（
P115_READ_TIMEOUT、P115_RETRIES），改行为不用改代码、重打镜像。 -
绕开自己的 import hook。注释特别说明这两个后端「多为嵌套导入，会被防重入锁跳过」，所以这里主动预加载并直接改，而不是指望技巧三的
__import__钩子去拦——很清楚地示范了两种 hook 各自的边界。
如何让 sitecustomize 生效
只要保证解释器能 import 到它，任选一种：
# 方式一：放进 site-packages（对该环境全局生效）
cp sitecustomize.py "$(python -c 'import site; print(site.getsitepackages()[0])')/"
# 方式二：放在任意目录，用 PYTHONPATH 指过去（推荐，隔离、可控）
export PYTHONPATH=/opt/patch:$PYTHONPATH
# /opt/patch/sitecustomize.py 就会被自动加载
想临时关掉所有此类自定义，可以用 python -S 启动（跳过 site 初始化），或设置 PYTHONNOUSERSITE。调试时若怀疑它没被加载，可以 python -c "import sitecustomize; print(sitecustomize.__file__)" 确认路径。
小结
sitecustomize.py 的价值在于它「早、且全局」的执行时机，让你能对一个无法（或不想）改动源码的 Python 应用做外科手术式的运行时补丁。本文这段脚本示范了一套可复用的组合拳：
- 用双重锁保证补丁只装一次、且不自我递归；
- 用
sys.path修复解决打包 / 容器环境的路径丢失； - 用
__import__hook 在模块加载完成后替换其公开符号； - 用 monkey patch 直接给底层 HTTP 后端强制注入超时，兜住雪崩。
需要提醒的是：这类技术是把双刃剑。用它做路径修复、超时兜底、监控注入是稳态运维的好帮手；但用它绕过授权校验则涉及软件授权与合规问题，请仅在自有 / 已获授权的环境里使用。理解它的原理，更多是为了在排障时看懂「为什么我的进程行为和源码对不上」。