Runtime network egress control for Python

One function call. Zero dependencies. No infrastructure changes.

tethered is a lightweight, in-process policy check that hooks into Python's own socket layer to enforce your allow list before any packet leaves the machine. Use activate() to set a process-wide ceiling, and scope() to tighten individual code paths — request handlers, background jobs, library calls, AI-generated code. Everything runs locally within your process — works with requests, httpx, aiohttp, Django, Flask, FastAPI, and any library built on Python sockets.

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.test/exfil")            # raises tethered.EgressBlocked

Why tethered?

Your code, your dependencies, and AI coding agents all share the same Python process — and any of them can make network calls you didn't intend. A compromised dependency phones home. An AI coding agent writes tests that accidentally call live APIs. An AI-generated function calls an unauthorized endpoint. A misconfigured library hits production instead of staging.

Python has no built-in way to prevent this at runtime. Infrastructure-level controls (firewalls, network policies, proxies) require platform teams, separate services, or admin privileges. None of them give you a single line of Python that says "this process may only talk to these hosts."

tethered fills this gap at the application layer. One function call controls what any code in the process — yours, your dependencies', or AI-generated — can reach over the network. No proxies, no sidecars, no admin privileges. It's complementary to infrastructure controls, not a replacement.

Use cases

	Use case	How tethered helps
🔒	Supply chain defense	`activate()` locks the process to your known services — a compromised dependency can't phone home.
🔬	Scoped isolation	`scope()` restricts a specific code path — a request handler, a background job, a library call — to only the destinations it needs.
🤖	AI agent guardrails	Code generated by AI coding agents can't reach unauthorized endpoints — `activate()` enforces your allow list on any code running in the process.
🧪	Test isolation	`activate()` in your test setup ensures the suite never accidentally hits production services.
📋	Least-privilege networking	Combine `activate()` for the process boundary with `scope()` for per-function restrictions — declare your network surface like you declare your dependencies.

Install

uv add tethered

Or with pip:

pip install tethered

Requires Python 3.10+. Zero runtime dependencies. Pre-built wheels are available for Linux, macOS, and Windows. Source installs require a C compiler (the package includes a C extension for tamper-resistant locked mode).

Getting started

tethered has two complementary APIs. Both use the same Allow list syntax.

Process-wide ceiling: `activate()`

Call activate() as early as possible — before any library makes network connections:

# manage.py, wsgi.py, main.py, or your entrypoint
import tethered
tethered.activate(allow=["*.stripe.com:443", "db.internal:5432"])

# Then import and run your app
from myapp import create_app
app = create_app()

This pattern works the same for Django, Flask, FastAPI, scripts, and AI-assisted workflows — activate tethered before your application and its dependencies start making connections.

Existing connections (e.g., connection pools) established before activate() will continue to work — tethered intercepts at connect time, not at read/write time.

Use locked=True in production to prevent any code from replacing or disabling your policy. See Locked mode.

Context-local restriction: `scope()`

Use scope() to restrict egress for a specific code path — a request handler, a background job, a library call:

import tethered
import httpx

def charge(amount: int, token: str) -> dict:
    with tethered.scope(allow=["*.stripe.com:443"]):
        # ... validate input, call helper libraries, log analytics —
        # none of them can reach anything except *.stripe.com:443
        resp = httpx.post("https://api.stripe.com/v1/charges", ...)
        return resp.json()

Or as a decorator:

@tethered.scope(allow=["*.stripe.com:443"])
def charge(amount: int, token: str) -> dict:
    # ... validate input, call helper libraries, log analytics —
    # none of them can reach anything except *.stripe.com:443
    resp = httpx.post("https://api.stripe.com/v1/charges", ...)
    return resp.json()

No deactivate() needed — cleanup is automatic when the context exits or the decorated function returns. Works with both sync and async functions.

scope() works on its own — no activate() required. When used alone, the scope IS the policy for that code path. Code outside the scope is unaffected.

Scopes can only restrict, never widen. If the app also called activate(), the effective policy is the intersection — a connection must be allowed by both.

Package maintainers: Use scope(), never activate(). Your library doesn't own the process — the app does. activate() is a process-wide operation that would interfere with the host application's own policy. scope() is context-local and safe to use from any library.

How `activate()` and `scope()` work together

activate() sets a process-wide ceiling. No code anywhere in the process can reach destinations outside it.
scope() creates a temporary restriction within the current context. It can only narrow the effective policy, never widen it.
When both are active, the effective policy is the intersection — a connection must be allowed by both the global policy and every active scope.

# Process ceiling: allow Stripe and Twilio
tethered.activate(allow=["*.stripe.com:443", "*.twilio.com:443"])

# Payment endpoint: scope restricts to Stripe only
# (tethered logs a warning that *.sendgrid.com has no overlap with the global policy)
with tethered.scope(allow=["*.stripe.com:443", "*.sendgrid.com:443"]):
    # *.stripe.com:443   — allowed (in both global and scope)
    # *.sendgrid.com:443 — blocked (not in global policy — scope cannot widen)
    # *.twilio.com:443   — blocked (not in scope)
    httpx.post("https://api.stripe.com/v1/charges")  # works
    httpx.post("https://api.sendgrid.com/v3/mail")   # raises EgressBlocked

Nested scopes

Scopes nest naturally — each level further restricts:

tethered.activate(allow=["*.stripe.com:443", "*.twilio.com:443", "db.internal:5432"])

with tethered.scope(allow=["*.stripe.com:443", "*.twilio.com:443"]):
    # db.internal:5432 is excluded by this scope

    with tethered.scope(allow=["*.stripe.com:443"]):
        # Now only *.stripe.com:443 is allowed
        httpx.post("https://api.stripe.com/v1/charges")    # works
        httpx.post("https://api.twilio.com/v1/messages")   # raises EgressBlocked

Examples

Runnable examples covering each feature:

Example	Description
01_basic_activate.py	Process-wide allow list with `activate()`
02_scope_context_manager.py	`scope()` as a context manager
03_scope_decorator.py	`scope()` as a function decorator
04_global_with_scope.py	Global policy + scope — intersection semantics
05_global_with_nested_scopes.py	Global policy + nested scopes — progressive restriction
06_locked_mode.py	`locked=True` — prevent policy tampering
07_log_only.py	Monitor-only mode with `on_blocked` callback
08_scope_in_threads.py	Scoping inside thread pool workers
09_async_scope.py	Async decorator and context manager
10_package_maintainer.py	Library self-restricting with `scope()`

Allow list syntax

Pattern	Example	Matches
Exact hostname	`"api.stripe.com"`	`api.stripe.com` only
Wildcard subdomain	`"*.stripe.com"`	`api.stripe.com`, `dashboard.stripe.com` (not `stripe.com`)
Hostname + port	`"api.stripe.com:443"`	`api.stripe.com` on port 443 only
IPv4 address	`"198.51.100.1"`	That IP only
IPv4 CIDR range	`"10.0.0.0/8"`	Any IP in `10.x.x.x`
CIDR + port	`"10.0.0.0/8:5432"`	Any IP in `10.x.x.x` on port 5432
IPv6 address	`"2001:db8::1"` or `"[2001:db8::1]"`	That IPv6 address
IPv6 + port	`"[2001:db8::1]:443"`	That IPv6 address on port 443 only
IPv6 CIDR	`"[2001:db8::]/32"`	Any IP in that IPv6 prefix

Wildcard matching: Uses Python's fnmatch syntax. * matches any characters including dots, so *.stripe.com matches both api.stripe.com and a.b.stripe.com. This differs from TLS certificate wildcards. The characters ? (single character) and [seq] (character set) are also supported.

Localhost (127.0.0.0/8, ::1) is always allowed by default. The addresses 0.0.0.0 and :: (INADDR_ANY) are also treated as localhost.

Malformed hostnames containing whitespace, control characters, or invisible Unicode are rejected and never matched by wildcard rules.

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,
)

Parameter	Description
`allow`	Required. Allowed destinations — see Allow list syntax. Pass `[]` to block all non-localhost connections.
`log_only`	Log blocked connections instead of raising `EgressBlocked`. Default `False`.
`fail_closed`	Block when the policy check itself errors, instead of failing open. Default `False`.
`allow_localhost`	Allow loopback addresses (`127.0.0.0/8`, `::1`). Default `True`.
`on_blocked`	Callback `(host, port) -> None` invoked on every blocked connection, including in log-only mode.
`locked`	Enable tamper-resistant enforcement via C extension. Prevents `deactivate()` and `activate()` without the correct `lock_token`, and installs a C-level integrity verifier that blocks ALL network access on tamper detection. Default `False`. See Locked mode.
`lock_token`	Opaque, non-internable token required when `locked=True`. Must be an instance like `object()` — internable types (`str`, `int`, `float`, `bytes`, `bool`) are rejected with `TypeError`. Compared by identity (`is`), not equality.

Can be called multiple times to replace the active policy — calling activate() again does not require deactivate() first. If the current policy is locked, the correct lock_token must be provided. Each call creates a completely new policy; no parameters or state carry over from previous calls.

`tethered.scope()`

tethered.scope(
    *,
    allow: list[str],
    allow_localhost: bool = True,
    log_only: bool = False,
    fail_closed: bool = False,
    on_blocked: Callable[[str, int | None], None] | None = None,
)

Parameter	Description
`allow`	Required. Allowed destinations — see Allow list syntax.
`allow_localhost`	Allow loopback addresses. Default `True`.
`log_only`	Log blocked connections instead of raising. Default `False`.
`fail_closed`	Block when the policy check itself errors. Default `False`.
`on_blocked`	Callback `(host, port) -> None` on every blocked connection.

Use as a context manager (with tethered.scope(allow=[...]):) or a decorator (@tethered.scope(allow=[...])). Supports both sync and async functions. Cleanup is automatic — no deactivate() call needed.

Log-only mode

Monitor without blocking — useful for rollout or auditing:

tethered.activate(
    allow=["*.stripe.com"],
    log_only=True,
    on_blocked=lambda host, port: print(f"would block: {host}:{port}"),
)

tethered logs to the "tethered" logger via stdlib logging. To see log-only warnings, ensure your application has logging configured (e.g., logging.basicConfig()).

Locked mode

Tamper-resistant enforcement backed by a C extension:

secret = object()
tethered.activate(allow=["*.stripe.com:443"], locked=True, lock_token=secret)

# Both deactivate() and activate() require the correct token
tethered.deactivate(lock_token=secret)

Calling activate() or deactivate() without the correct lock_token raises TetheredLocked. See the Security model for the full threat analysis.

`tethered.deactivate()`

tethered.deactivate(*, lock_token: object | None = None)

Disable enforcement. All connections are allowed again. Internal state (IP-to-hostname mappings, callback references) is fully cleared — a subsequent activate() starts fresh.

If activated with locked=True, the matching lock_token must be provided or TetheredLocked is raised.

`tethered.EgressBlocked`

Raised when a connection is blocked. Subclass of RuntimeError.

try:
    urllib.request.urlopen("https://evil.test")
except tethered.EgressBlocked as e:
    print(e.host)           # "evil.test"
    print(e.port)           # 443
    print(e.resolved_from)  # original hostname if connecting by resolved IP

`tethered.TetheredLocked`

Raised when deactivate() or activate() is called on a locked policy without the correct token. Subclass of RuntimeError.

How it works

tethered uses sys.addaudithook (PEP 578) to intercept socket operations at the interpreter level:

socket.getaddrinfo — blocks DNS resolution for disallowed hostnames and records IP-to-hostname mappings for allowed hosts.
socket.gethostbyname / socket.gethostbyaddr — intercept alternative DNS resolution paths, including reverse-DNS lookups of raw IPs.
socket.connect (including connect_ex, which raises the socket.connect audit event in CPython) — enforces the allow list on TCP connections.
socket.sendto / socket.sendmsg — enforces the allow list on UDP datagrams.

When getaddrinfo resolves a hostname, tethered records the IP-to-hostname mapping in a bounded LRU cache. When a subsequent connect() targets that IP, tethered looks up the original hostname and checks it against the allow list. If denied, EgressBlocked is raised before any packet leaves the machine.

This works with libraries built on CPython sockets (requests, httpx, urllib3, aiohttp) and frameworks like Django, Flask, and FastAPI — they all call socket.getaddrinfo and socket.connect under the hood. Asyncio and async libraries using CPython sockets are supported: audit hooks fire at the C socket level, so asyncio, aiohttp, and httpx async use the same enforcement path as synchronous code.

scope() uses contextvars.ContextVar to push a per-context policy onto a stack. The audit hook checks the context-local scope stack in addition to the global policy. When a scope is active, a connection must pass both the global policy and every scope on the stack. When the context manager exits (or the decorated function returns), the scope is automatically popped. Because ContextVar is async-safe, scopes propagate correctly through await chains and asyncio.create_task(). Scopes do not automatically propagate to child threads — use scope() at the I/O point inside the thread, or use activate() for a process-wide ceiling.

The per-connection overhead is a Python function call with hostname normalization, a dictionary lookup, and pattern matching — designed to add minimal overhead relative to actual network I/O.

Security model

tethered is a defense-in-depth guardrail, not a security sandbox. It intercepts Python-level socket operations. Code that uses ctypes, cffi, subprocesses, or C extensions with direct syscalls can bypass it. For full process isolation, combine tethered with OS-level controls (containers, seccomp, network namespaces).

What tethered protects against

Trusted-but-buggy code and supply chain threats: dependencies that use Python's standard socket module (directly or through libraries like requests, urllib3, httpx, aiohttp). tethered prevents these from connecting to destinations not in your allow list.

What tethered does NOT protect against

ctypes / cffi / direct syscalls. Native code can call libc's connect() directly, bypassing the audit hook.
Subprocesses. subprocess.Popen, os.system, and os.exec* create new processes without the audit hook.
C extensions with raw socket calls. Extensions calling C-level socket functions are not intercepted.
In-process disabling (without locked=True). Code in the same interpreter can call deactivate() or activate() unless locked=True is used.
ctypes memory manipulation (with locked=True). Locked mode catches Python-level tampering: config replacement, method monkey-patching, frozen field mutation, bytecode swapping, and exception class replacement. sys.modules replacement is ineffective — the C guardian holds direct references to critical objects cached at activation time, so it never looks up modules through sys.modules. The remaining bypasses require ctypes to manipulate raw process memory — targeting CPython internals and/or the compiled C extension's private state. These attacks are version-specific, platform-specific, and fragile. They are not practical for opportunistic supply-chain attacks — they require a payload tailored to the exact Python version, OS, and tethered build.

Design trade-offs

Fail-open by default. If tethered's matching logic raises an unexpected exception, the connection is allowed and a warning is logged. A bug in tethered should not break your application. Use fail_closed=True for stricter environments.
Audit hooks are irremovable. sys.addaudithook has no remove function (by design — PEP 578). deactivate() makes the hook a no-op but cannot unregister it. This is per-process only — no persistent state, no system changes, everything is gone when the process exits.
IP-to-hostname mapping is bounded. The LRU cache holds up to 4096 entries. In long-running processes with many unique DNS lookups, older mappings are evicted. A connection to an evicted IP is checked against IP/CIDR rules only.
Direct IP connections skip hostname matching. Connecting to a raw IP without prior DNS resolution means only IP/CIDR rules apply — hostname wildcards won't match. On shared-IP infrastructure (CDNs, cloud hosting), multiple hostnames may resolve to the same IP. If an allowed hostname shares an IP with a disallowed one, a raw-IP connection to that address will pass hostname policy via the cached mapping. This is inherent to any system that cannot bind a socket to a specific hostname identity.
Localhost allows local relays. With the default allow_localhost=True, any proxy, tunnel, or forwarding agent listening on 127.0.0.1 or ::1 can relay traffic to external destinations, bypassing the intent of the egress policy. In high-security environments where local relays are a concern, set allow_localhost=False and explicitly allow only the loopback addresses and ports your application needs.

Recommendations

For defense-in-depth, combine tethered with:

OS-level sandboxing (containers, seccomp-bpf, network namespaces) for hard isolation.
Subprocess restrictions (audit hooks on subprocess.Popen events, or seccomp filters).
Import restrictions to prevent ctypes/cffi loading in untrusted code paths.

Handling blocked connections

EgressBlocked is a RuntimeError, not an OSError. This is intentional — a policy violation is not a network error and should not be silently caught by HTTP libraries or retry logic. You'll want to handle it explicitly at your application boundaries.

Django / FastAPI middleware

# 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 tasks

# EgressBlocked is a RuntimeError, so autoretry_for=(ConnectionError, TimeoutError)
# already won't retry it — the task fails immediately on a policy violation.
@app.task(autoretry_for=(ConnectionError, TimeoutError))
def sync_data():
    requests.post("https://api.stripe.com/v1/charges", ...)

Retry decorators

# Catch EgressBlocked before your retry logic — retrying a policy block is pointless
try:
    response = retry_with_backoff(make_request)
except tethered.EgressBlocked:
    raise  # don't retry policy violations
except ConnectionError:
    handle_network_failure()

Badge

Using tethered in your project? Add the badge to your README:

[![egress: tethered](https://img.shields.io/badge/egress-tethered-orange?labelColor=4B8BBE)](https://github.com/shcherbak-ai/tethered)

Contributing

See CONTRIBUTING.md for development setup and guidelines.

Security

See SECURITY.md for reporting vulnerabilities.

License

MIT

Name		Name	Last commit message	Last commit date
Latest commit History 37 Commits
.github		.github
assets		assets
examples		examples
scripts		scripts
src/tethered		src/tethered
tests		tests
tests_examples		tests_examples
.gitattributes		.gitattributes
.gitignore		.gitignore
.markdownlint-cli2.yaml		.markdownlint-cli2.yaml
.pre-commit-config.yaml		.pre-commit-config.yaml
AGENTS.md		AGENTS.md
CHANGELOG.md		CHANGELOG.md
CITATION.cff		CITATION.cff
CLAUDE.md		CLAUDE.md
CONTRIBUTING.md		CONTRIBUTING.md
LICENSE		LICENSE
README.md		README.md
SECURITY.md		SECURITY.md
pyproject.toml		pyproject.toml
setup.py		setup.py
uv.lock		uv.lock

Folders and files

Latest commit

History

Repository files navigation

Runtime network egress control for Python

Why tethered?

Use cases

Install

Getting started

Process-wide ceiling: activate()

Context-local restriction: scope()

How activate() and scope() work together

Nested scopes

Examples

Allow list syntax

API

tethered.activate()

tethered.scope()

Log-only mode

Locked mode

tethered.deactivate()

tethered.EgressBlocked

tethered.TetheredLocked

How it works

Security model

What tethered protects against

What tethered does NOT protect against

Design trade-offs

Recommendations

Handling blocked connections

Django / FastAPI middleware

Celery tasks

Retry decorators

Badge

Contributing

Security

License

About

Topics

Resources

License

Contributing

Security policy

Uh oh!

Stars

Watchers

Forks

Releases 4

Packages 0

Uh oh!

Uh oh!

Contributors 1

Languages

Process-wide ceiling: `activate()`

Context-local restriction: `scope()`

How `activate()` and `scope()` work together

`tethered.activate()`

`tethered.scope()`

`tethered.deactivate()`

`tethered.EgressBlocked`

`tethered.TetheredLocked`

Packages