Metadata-Version: 2.4
Name: tethered
Version: 0.1.3
Summary: Runtime network egress control for Python
Project-URL: Homepage, https://github.com/shcherbak-ai/tethered
Project-URL: Repository, https://github.com/shcherbak-ai/tethered
Project-URL: Issues, https://github.com/shcherbak-ai/tethered/issues
Author: Sergii Shcherbak
License-Expression: MIT
License-File: LICENSE
Keywords: audit,egress,network,security,supply-chain
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Topic :: Security
Classifier: Topic :: System :: Networking
Classifier: Typing :: Typed
Requires-Python: >=3.10
Description-Content-Type: text/markdown

# 🪁 tethered

[![CI](https://github.com/shcherbak-ai/tethered/actions/workflows/ci.yml/badge.svg)](https://github.com/shcherbak-ai/tethered/actions/workflows/ci.yml)
[![CodeQL](https://github.com/shcherbak-ai/tethered/actions/workflows/codeql.yml/badge.svg)](https://github.com/shcherbak-ai/tethered/actions/workflows/codeql.yml)
[![PyPI](https://img.shields.io/pypi/v/tethered?v=1)](https://pypi.org/project/tethered/)
[![Python](https://img.shields.io/pypi/pyversions/tethered?v=1)](https://pypi.org/project/tethered/)
[![coverage](https://img.shields.io/endpoint?url=https://gist.githubusercontent.com/SergiiShcherbak/20432f86c9102aa2b77ad9e4d4c21aa6/raw/tethered-coverage.json)](https://github.com/shcherbak-ai/tethered/actions/workflows/ci.yml)
[![License: MIT](https://img.shields.io/badge/license-MIT-blue.svg)](https://github.com/shcherbak-ai/tethered/blob/main/LICENSE)
[![Ruff](https://img.shields.io/endpoint?url=https://raw.githubusercontent.com/astral-sh/ruff/main/assets/badge/v2.json)](https://github.com/astral-sh/ruff)
[![uv](https://img.shields.io/endpoint?url=https://raw.githubusercontent.com/astral-sh/uv/main/assets/badge/v0.json)](https://github.com/astral-sh/uv)
[![Checked with pyright](https://microsoft.github.io/pyright/img/pyright_badge.svg)](https://microsoft.github.io/pyright/)

Runtime network egress control for Python. One function call to restrict which hosts your code can connect to.

```python
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 is a lightweight in-process policy check — not a proxy, not a firewall. It intercepts Python socket operations via [`sys.addaudithook`](https://docs.python.org/3/library/sys.html#sys.addaudithook) ([PEP 578](https://peps.python.org/pep-0578/)) and blocks disallowed connections before any packet leaves the machine. No admin privileges, no infrastructure changes, no effect on other processes.

## Why

Python has no built-in way to restrict outbound network access at runtime. Infrastructure-level controls (firewalls, network policies, proxies) operate outside your application — they 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. It's complementary to infrastructure controls, not a replacement.

### Use cases

- **Supply chain defense.** A compromised dependency can't phone home if egress is locked to your known services.
- **AI agent guardrails.** Constrain LLM-powered agents to only the APIs they need.
- **Test isolation.** Ensure your test suite never accidentally hits production endpoints.
- **Least-privilege networking.** Declare your app's network surface the same way you declare its dependencies.

## Install

```bash
uv add tethered
```

Or with pip:

```bash
pip install tethered
```

Requires Python 3.10+. Zero runtime dependencies.

## Getting started

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

```python
# 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 agents — 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.

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

## API

### `tethered.activate()`

```python
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](#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` | Prevent `deactivate()` without the correct `lock_token`. Default `False`. |
| `lock_token` | Opaque token required to `deactivate()` when locked. Compared by identity (`is`), not equality. |

Can be called multiple times to replace the active policy — calling `activate()` again does not require `deactivate()` first. Each call creates a completely new policy; no parameters or state carry over from previous calls.

#### Log-only mode

Monitor without blocking — useful for rollout or auditing:

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

#### Locked mode

Prevent in-process code from disabling enforcement:

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

# Only works with the correct token
tethered.deactivate(lock_token=secret)
```

### `tethered.deactivate(*, lock_token=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`.

```python
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`

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

## How it works

Tethered uses [`sys.addaudithook`](https://docs.python.org/3/library/sys.html#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`** — intercepts alternative DNS resolution paths.
- **`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.

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.** Code in the same interpreter can call `deactivate()` unless `locked=True` is used. Even locked mode can be bypassed by code that modifies module state — Python has no true encapsulation.

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

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

```python
# 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

```python
# 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

```python
# 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:

```markdown
[![egress: 🪁 tethered](https://img.shields.io/badge/egress-%F0%9F%AA%81%20tethered-4B8BBE.svg)](https://github.com/shcherbak-ai/tethered)
```

[![egress: 🪁 tethered](https://img.shields.io/badge/egress-%F0%9F%AA%81%20tethered-4B8BBE.svg)](https://github.com/shcherbak-ai/tethered)

## Contributing

See [CONTRIBUTING.md](CONTRIBUTING.md) for development setup and guidelines.

## Security

See [SECURITY.md](SECURITY.md) for reporting vulnerabilities.

## License

MIT
