API Reference

Every public name, generated from the package's own docstrings.

The same surface in prose, with the bounds and the gotchas, is the API header that ships inside the installed package, netproxy/AGENTS.md, readable with importlib.resources.files("netproxy") and printed in the Guide of this site.

A rule-routed asyncio proxy server with per-destination mTLS.

TLSSource = Union[ssl.SSLContext, Credential] module-attribute

A client context used as is, or a :class:Credential that builds one.

Upstream = Union[str, Any, None] module-attribute

None: the server's egress; a string: a proxy URL, "direct", or a PAC result such as "PROXY a:3128; DIRECT"; anything else: a proxylib ProxyMap (indexed with the destination URL).

Connection(host, port, scheme='https', path=None, client=None, listener='') dataclass

A connection the proxy is about to make, as rules see it.

scheme is "https" for a CONNECT tunnel, "http" for a forwarded plain-HTTP request, "tcp" for a tunnel listener. path is set only for forwarded HTTP requests. client is the peer address, listener the name of the listener that accepted it.

authority property

host:port, with IPv6 hosts bracketed.

url property

The destination as a URL, the key an upstream ProxyMap is asked; the scheme's default port is left out, as clients and PAC scripts write it.

Credential(identity=None, *, backend=None, pem=None, options=None, query=None, verify=None, cafile=None, capath=None, cadata=None, minimum_version=None, name='')

A client identity for mTLS, turned into a client SSLContext lazily.

Give one source: identity (an :class:sslcontext.Identity, or a callable returning one); pem, a (certificate, key, password) tuple for :func:sslcontext.load_pem (see :meth:from_pem); or a keystore backend name with its constructor options (store, module, pin, ...) and a query for :meth:sslcontext.backends.Backend.find_identity (subject, thumbprint, label, ...). verify, cafile, capath, cadata and minimum_version shape how the destination's certificate is checked, as in :func:sslcontext.create_context.

Building touches the keystore, so it may block: the server calls :meth:context from a worker thread. The context is cached until :meth:invalidate; a failure is not cached, so the next connection tries again.

Source code in src/netproxy/_credential.py

def __init__(
    self,
    identity: Optional[IdentitySource] = None,
    *,
    backend: Optional[str] = None,
    pem: Optional[Tuple[Any, Any, Any]] = None,
    options: Optional[Mapping[str, Any]] = None,
    query: Optional[Mapping[str, Any]] = None,
    verify: Union[bool, str, None] = None,
    cafile: Optional[str] = None,
    capath: Optional[str] = None,
    cadata: Union[str, bytes, None] = None,
    minimum_version: Optional[ssl.TLSVersion] = None,
    name: str = "",
) -> None:
    if sum(source is not None for source in (identity, backend, pem)) != 1:
        raise ValueError("give exactly one of identity=, backend= or pem=")
    self._pem = pem
    if backend is not None:
        get_backend(backend)  # an unknown name is a caller's error, now
    self._identity = identity
    self.backend = backend
    self.options = dict(options or {})
    self.query = dict(query or {})
    self._context_options = {
        "verify": verify,
        "cafile": cafile,
        "capath": capath,
        "cadata": cadata,
        "minimum_version": minimum_version,
    }
    self.name = name
    self._lock = threading.Lock()
    self._context: Optional[ssl.SSLContext] = None

from_pem(certificate, key=None, password=None, *, name='', **context_options) classmethod

A credential from a certificate (+ chain) and key held in memory, as :func:sslcontext.load_pem takes them: PEM or DER (key omitted: both are in certificate), or a PKCS#12 bundle as certificate with no key. password decrypts the key or bundle (a value, a file, or a callable).

Source code in src/netproxy/_credential.py

@classmethod
def from_pem(
    cls,
    certificate: Union[str, bytes],
    key: Union[str, bytes, None] = None,
    password: "sslcontext.PasswordSource" = None,
    *,
    name: str = "",
    **context_options: Any,
) -> "Credential":
    """A credential from a certificate (+ chain) and key held in memory,
    as :func:`sslcontext.load_pem` takes them: PEM or DER (``key``
    omitted: both are in ``certificate``), or a PKCS#12 bundle as
    ``certificate`` with no ``key``. ``password`` decrypts the key or
    bundle (a value, a file, or a callable)."""
    return cls(pem=(certificate, key, password), name=name, **context_options)

invalidate()

Forget the built context; the next use looks the identity up again.

Source code in src/netproxy/_credential.py

def invalidate(self) -> None:
    """Forget the built context; the next use looks the identity up again."""
    with self._lock:
        self._context = None

context()

The client context, built on first call.

Raises :class:~netproxy.exceptions.CredentialUnavailableError when the identity cannot be found or loaded.

Source code in src/netproxy/_credential.py

def context(self) -> ssl.SSLContext:
    """The client context, built on first call.

    Raises :class:`~netproxy.exceptions.CredentialUnavailableError` when
    the identity cannot be found or loaded.
    """
    with self._lock:
        if self._context is None:
            self._context = self._build()
        return self._context

AsyncProxyServer(router=None, *, listen=None, egress=None, tunnels=(), timeout=30.0, idle_timeout=None, allow_remote=False, ca=None, recorders=(), keylog=None)

A proxy server on the running event loop.

router decides every connection (default: everything direct). listen is where the HTTP proxy accepts, in netimps.parse_listen syntax ("127.0.0.1:3128", a list of such); None means 127.0.0.1 on a free port, () no HTTP proxy listener at all. egress is a proxylib ProxyMap (or proxy URL string) for routes that set no upstream. tunnels are extra :class:Tunnel listeners. timeout bounds each connection attempt and handshake, idle_timeout ends a relayed direction that stays silent that long. Listening beyond loopback needs allow_remote=True. ca (an sslcontext.issuing.CertificateAuthority or a :class:~netproxy.intercept.Interceptor) enables interception of CONNECT requests whose route is :attr:~netproxy.Action.TLS. recorders (see :mod:netproxy.recording) see every relayed connection. keylog names a file that receives the TLS secrets of every TLS connection the proxy makes or ends, in the SSLKEYLOGFILE format Wireshark reads; treat it as a secret.

The constructor does no I/O; :meth:start binds, :meth:aclose stops and waits for every connection to end.

Source code in src/netproxy/_server.py

def __init__(
    self,
    router: Optional[Routing] = None,
    *,
    listen: Any = None,
    egress: Any = None,
    tunnels: Iterable[Tunnel] = (),
    timeout: float = 30.0,
    idle_timeout: Optional[float] = None,
    allow_remote: bool = False,
    ca: Any = None,
    recorders: Iterable[Any] = (),
    keylog: Union[str, "os.PathLike[str]", None] = None,
) -> None:
    self.router: Routing = router if router is not None else Router()
    self.recorders = tuple(recorders)
    # What the server built itself (from a configuration file) and closes.
    self._owned: List[Callable[[], None]] = []
    if keylog is not None and not KEYLOG_SUPPORTED:
        # Elsewhere setting keylog_filename would only add an attribute.
        raise ValueError(
            f"keylog needs a TLS library that logs keys (OpenSSL 1.1.1 or "
            f"later); this Python uses {ssl.OPENSSL_VERSION}"
        )
    self.keylog = os.fspath(keylog) if keylog is not None else None
    self.interceptor: Any = None
    if ca is not None:
        from .intercept import Interceptor

        self.interceptor = ca if isinstance(ca, Interceptor) else Interceptor(ca)
        self.interceptor.keylog = self.keylog
    self.listen = listen
    self.egress = egress
    self.tunnels = tuple(tunnels)
    self.timeout = timeout
    self.idle_timeout = idle_timeout
    self.allow_remote = allow_remote
    self._servers: List[asyncio.AbstractServer] = []
    self._proxy_servers: List[asyncio.AbstractServer] = []
    self._tasks: Set["asyncio.Task[Any]"] = set()
    self._closed = False

addresses property

(host, port) the HTTP proxy listens on, ports read back.

port property

The HTTP proxy's first port (the free one picked for port 0).

start() async

Bind every listener. Raises OSError (address in use) or ValueError (a remote address without allow_remote).

Source code in src/netproxy/_server.py

async def start(self) -> None:
    """Bind every listener. Raises ``OSError`` (address in use) or
    ``ValueError`` (a remote address without ``allow_remote``)."""
    if self._servers:
        return
    try:
        if self.listen != ():
            for sock in _listen_sockets(self.listen, self.allow_remote):
                server = await asyncio.start_server(self._accept_proxy, sock=sock)
                self._servers.append(server)
                self._proxy_servers.append(server)
        for tunnel in self.tunnels:
            for sock in _listen_sockets(tunnel.listen, self.allow_remote):
                server = await asyncio.start_server(
                    lambda r, w, t=tunnel: self._accept_tunnel(t, r, w), sock=sock
                )
                self._servers.append(server)
    except BaseException:
        await self.aclose()
        raise

tunnel_address(tunnel)

Where tunnel (one of tunnels) listens, port read back.

Source code in src/netproxy/_server.py

def tunnel_address(self, tunnel: Tunnel) -> Tuple[str, int]:
    """Where ``tunnel`` (one of ``tunnels``) listens, port read back."""
    index = self.tunnels.index(tunnel)
    offset = len(self._proxy_servers)
    return self._servers[offset + index].sockets[0].getsockname()[:2]

serve_forever() async

Start if needed, then wait until :meth:aclose (or cancellation).

Source code in src/netproxy/_server.py

async def serve_forever(self) -> None:
    """Start if needed, then wait until :meth:`aclose` (or cancellation)."""
    await self.start()
    await asyncio.gather(*(s.serve_forever() for s in self._servers))

aclose() async

Stop listening, end every relayed connection and wait for them.

Source code in src/netproxy/_server.py

async def aclose(self) -> None:
    """Stop listening, end every relayed connection and wait for them."""
    self._closed = True
    for server in self._servers:
        server.close()
    for task in list(self._tasks):
        task.cancel()
    if self._tasks:
        await asyncio.gather(*self._tasks, return_exceptions=True)
    for server in self._servers:
        await server.wait_closed()
    self._servers.clear()
    self._proxy_servers.clear()
    for close in self._owned:
        close()
    self._owned.clear()

open_upstream(connection, route, *, forward_http=False) async

:func:netproxy.upstream.connect with this server's egress, timeout and key log.

Source code in src/netproxy/_server.py

async def open_upstream(
    self, connection: Connection, route: Route, *, forward_http: bool = False
) -> UpstreamStream:
    """:func:`netproxy.upstream.connect` with this server's egress,
    timeout and key log."""
    return await connect(
        connection,
        route,
        egress=self.egress,
        timeout=self.timeout,
        forward_http=forward_http,
        keylog=self.keylog,
    )

start_recording(connection, route, client, upstream)

A recording of one session, or None with no recorders.

Source code in src/netproxy/_server.py

def start_recording(
    self,
    connection: Connection,
    route: Route,
    client: asyncio.StreamWriter,
    upstream: UpstreamStream,
) -> Optional[_Recording]:
    """A recording of one session, or ``None`` with no recorders."""
    if not self.recorders:
        return None
    return _Recording.start(
        self.recorders,
        connection,
        route,
        client.get_extra_info("peername"),
        upstream.writer.get_extra_info("peername"),
    )

relay_session(connection, route, reader, writer, upstream, *, recording=None, before_server=None) async

Relay reader/writer with upstream, recorded when the server has recorders (recording passes one already started).

Source code in src/netproxy/_server.py

async def relay_session(
    self,
    connection: Connection,
    route: Route,
    reader: asyncio.StreamReader,
    writer: asyncio.StreamWriter,
    upstream: UpstreamStream,
    *,
    recording: Optional[_Recording] = None,
    before_server: Any = None,
) -> None:
    """Relay ``reader``/``writer`` with ``upstream``, recorded when the
    server has recorders (``recording`` passes one already started)."""
    if recording is None:
        recording = self.start_recording(connection, route, writer, upstream)
    try:
        await relay(
            reader,
            writer,
            upstream.reader,
            upstream.writer,
            idle_timeout=self.idle_timeout,
            on_data=recording.data if recording else None,
            on_end=recording.ended if recording else None,
            before_server=before_server,
        )
    finally:
        if recording is not None:
            recording.closed()

ProxyServer(router=None, *, server=None, **options)

:class:AsyncProxyServer for synchronous code: the server runs on its own event loop in a daemon thread. Same arguments.

:meth:start returns once every listener is bound (or raises what binding raised); :meth:close stops the server and joins the thread.

Source code in src/netproxy/_server.py

def __init__(
    self,
    router: Optional[Routing] = None,
    *,
    server: Optional[AsyncProxyServer] = None,
    **options: Any,
) -> None:
    if server is not None and (router is not None or options):
        raise TypeError("pass server= alone, or router and options, not both")
    self._server = (
        server if server is not None else AsyncProxyServer(router, **options)
    )
    self._loop: Optional[asyncio.AbstractEventLoop] = None
    self._thread: Optional[threading.Thread] = None

Tunnel(listen, target, route=None, name='') dataclass

A local port whose connections go to one fixed target.

listen is where to accept ("127.0.0.1:15432"); target is host:port. Each connection is routed as a Connection with scheme "tcp" and listener name (default "tunnel:<target>"); route overrides the router for this tunnel.

Action

Bases: Enum

What the proxy does with a connection.

DIRECT = 'direct' class-attribute instance-attribute

Relay bytes to the destination unchanged.

TLS = 'tls' class-attribute instance-attribute

The proxy makes the TLS connection to the destination with the route's client context. Plain HTTP is upgraded to HTTPS; a CONNECT tunnel is intercepted (the client's TLS ends at the proxy).

REJECT = 'reject' class-attribute instance-attribute

Refuse the connection.

Route(action=Action.DIRECT, tls=None, upstream=None, name='') dataclass

The decision for one connection.

tls is required for, and only allowed with, :attr:Action.TLS. upstream overrides the server's egress for this route (see :data:Upstream). name labels the route in logs and check.

Router(rules=(), *, default=None)

Ordered rules; the first that matches decides, else default.

Thread-safe for reading; :meth:add is meant for set-up.

Source code in src/netproxy/_route.py

def __init__(
    self, rules: Iterable[RuleLike] = (), *, default: Optional[Route] = None
) -> None:
    self.default = default if default is not None else Route.direct(name="default")
    self._rules: List[Rule] = []
    for rule in rules:
        self.add(rule)

add(rule, route=None)

Append rule: a :class:Rule, a (match, route) pair, or add(match, route). Returns the stored :class:Rule.

Source code in src/netproxy/_route.py

def add(self, rule: Union[RuleLike, Any], route: Optional[Route] = None) -> Rule:
    """Append ``rule``: a :class:`Rule`, a ``(match, route)`` pair, or
    ``add(match, route)``. Returns the stored :class:`Rule`."""
    if route is not None:
        rule = Rule(rule, route)
    elif isinstance(rule, tuple):
        rule = Rule(*rule)
    if not isinstance(rule, Rule):
        raise TypeError(
            "add() takes a Rule, a (match, route) pair, or match and route"
        )
    self._rules.append(rule)
    return rule

explain(connection)

The rule that matched (None for the default) and its route.

Source code in src/netproxy/_route.py

def explain(self, connection: Connection) -> Tuple[Optional[Rule], Route]:
    """The rule that matched (``None`` for the default) and its route."""
    for rule in self._rules:
        if rule.match(connection):
            return rule, rule.route
    return None, self.default

Routing

Bases: Protocol

Anything that routes connections; :class:Router is the built-in one.

Rule(match, route, name='') dataclass

route applies to connections match accepts.

match is a :class:~netproxy.rules.Matcher, any callable taking a :class:~netproxy.Connection, or a pattern string for :func:~netproxy.rules.parse_matcher.

ConfigError(msg, path=None, lineno=None, colno=None)

Bases: NetProxyError, ValueError

A configuration document is malformed or inconsistent.

path, lineno and colno locate the problem when known; the message never quotes the document.

Source code in src/netproxy/exceptions.py

def __init__(
    self,
    msg: str,
    path: str | None = None,
    lineno: int | None = None,
    colno: int | None = None,
) -> None:
    where = ":".join(str(p) for p in (path, lineno, colno) if p is not None)
    super().__init__(f"{where}: {msg}" if where else msg)
    self.msg = msg
    self.path = path
    self.lineno = lineno
    self.colno = colno

CredentialUnavailableError

Bases: NetProxyError

A route's client credential could not be loaded (keystore entry missing, no PIN, keystore failure).

NetProxyError

Bases: Exception

Base class for every error this package raises.

RejectedError

Bases: NetProxyError

A rule rejected the connection.

UpstreamError

Bases: NetProxyError, OSError

The destination, or an upstream proxy on the way to it, could not be reached or refused the connection.

Upstream connections

Reaching a destination: directly or through upstream proxies, then TLS if the route says so.

Upstream proxies come from a proxylib ProxyMap (the route's upstream, else the server's egress). Its answers are tried in order, so a PAC result such as PROXY a:3128; DIRECT fails over. HTTP and HTTPS proxies are supported (CONNECT, Basic credentials from the proxy URL).

UpstreamStream(reader, writer, via, tls, forward=False, proxy_authorization=None) dataclass

An open connection to a destination.

via instance-attribute

host:port of the upstream proxy used, or None for direct.

tls instance-attribute

Whether the proxy made a TLS connection to the destination.

forward = False class-attribute instance-attribute

The stream reaches an upstream HTTP proxy without a tunnel: send it absolute-form requests (forward_http=True and a proxy answer).

proxy_authorization = field(default=None, repr=False) class-attribute instance-attribute

Proxy-Authorization value for a forward stream, from the proxy URL's credentials.

resolve_upstream(connection, route, egress=None) async

The upstream answers to try for connection: proxylib Proxy objects and None (direct), in order.

Source code in src/netproxy/upstream.py

async def resolve_upstream(
    connection: Connection, route: Route, egress: Any = None
) -> List[Any]:
    """The upstream answers to try for ``connection``: proxylib ``Proxy``
    objects and ``None`` (direct), in order."""
    source = route.upstream if route.upstream is not None else egress
    if source is None:
        return [None]
    from proxylib import alookup

    try:
        return list(await alookup(_as_proxymap(source), connection.url))
    except KeyError:  # no opinion
        return [None]

connect(connection, route, *, egress=None, timeout=30.0, forward_http=False, keylog=None) async

Open the connection route describes to connection's destination.

Raises :class:~netproxy.exceptions.UpstreamError when no upstream answer works (the message lists each attempt, never proxy credentials) and :class:~netproxy.exceptions.CredentialUnavailableError when a TLS route's credential cannot be loaded. A REJECT route is a ValueError: the caller should not have asked.

forward_http=True (a plain http:// request being forwarded) makes a proxy answer return a stream to the proxy itself, flagged forward, instead of a CONNECT tunnel: proxies commonly refuse CONNECT to port 80. keylog is set as the TLS context's keylog_filename.

Source code in src/netproxy/upstream.py

async def connect(
    connection: Connection,
    route: Route,
    *,
    egress: Any = None,
    timeout: float = 30.0,
    forward_http: bool = False,
    keylog: Optional[str] = None,
) -> UpstreamStream:
    """Open the connection ``route`` describes to ``connection``'s destination.

    Raises :class:`~netproxy.exceptions.UpstreamError` when no upstream
    answer works (the message lists each attempt, never proxy credentials)
    and :class:`~netproxy.exceptions.CredentialUnavailableError` when a TLS
    route's credential cannot be loaded. A REJECT route is a ``ValueError``:
    the caller should not have asked.

    ``forward_http=True`` (a plain ``http://`` request being forwarded) makes
    a proxy answer return a stream to the proxy itself, flagged ``forward``,
    instead of a CONNECT tunnel: proxies commonly refuse CONNECT to port 80.
    ``keylog`` is set as the TLS context's ``keylog_filename``.
    """
    if route.action is Action.REJECT:
        raise ValueError("a REJECT route opens no connection")
    context = await _client_context(route) if route.action is Action.TLS else None
    if context is not None and keylog is not None:
        context.keylog_filename = keylog
    failures = []
    for answer in await resolve_upstream(connection, route, egress):
        where = "direct" if answer is None else _describe(answer)
        try:
            if answer is None:
                reader, writer = await _open_direct(connection, timeout)
            elif forward_http and context is None:
                reader, writer = await _open_proxy(answer, timeout)
                return UpstreamStream(
                    reader,
                    writer,
                    _describe(answer),
                    False,
                    forward=True,
                    proxy_authorization=_authorization(answer),
                )
            else:
                reader, writer = await _open_via_proxy(connection, answer, timeout)
        except (OSError, asyncio.TimeoutError, asyncio.IncompleteReadError) as error:
            failures.append(f"{where}: {error or type(error).__name__}")
            _log.debug(
                "netproxy: %s via %s failed: %s", connection.authority, where, error
            )
            continue
        if context is not None:
            try:
                await start_tls(
                    reader,
                    writer,
                    context,
                    server_hostname=connection.host,
                    timeout=timeout,
                )
            except (OSError, ssl.SSLError, asyncio.TimeoutError) as error:
                await close_stream(writer)
                raise UpstreamError(
                    f"TLS to {connection.authority} via {where} failed: {error}"
                ) from error
            except BaseException:  # cancelled: no await is safe, nothing was sent
                writer.transport.abort()
                raise
        return UpstreamStream(
            reader,
            writer,
            None if answer is None else _describe(answer),
            context is not None,
        )
    raise UpstreamError(f"cannot reach {connection.authority}: " + "; ".join(failures))

TLS interception

TLS interception: the proxy ends the client's TLS with a certificate it mints.

A CONNECT whose route is :attr:~netproxy.Action.TLS is intercepted: the proxy connects to the destination first (with the route's client credential), answers 200, then completes the client's handshake with a leaf issued by its CA for the CONNECT host, and relays the plaintext between the two TLS connections. Clients must trust the CA (:attr:CertificateAuthority.certificate).

Create the CA with :func:create_ca so it is name-constrained to exactly the hosts the TLS rules intercept: a leaked key then cannot vouch for anything else.

Interceptor(ca, *, cache_size=256, leaf_days=7)

Server-side TLS for intercepted hosts, from ca.

Leaves (one per host, valid leaf_days) are cached, at most cache_size of them, least recently used first out.

Source code in src/netproxy/intercept.py

def __init__(
    self, ca: CertificateAuthority, *, cache_size: int = 256, leaf_days: float = 7
) -> None:
    if not isinstance(ca, CertificateAuthority):
        raise TypeError("ca must be an sslcontext.issuing.CertificateAuthority")
    self.ca = ca
    self.cache_size = cache_size
    self.leaf_days = leaf_days
    self.keylog: Optional[str] = None
    """Where minted contexts log TLS secrets (set by the server's ``keylog``)."""
    self._contexts: "OrderedDict[str, ssl.SSLContext]" = OrderedDict()
    self._lock = threading.Lock()

keylog = None instance-attribute

Where minted contexts log TLS secrets (set by the server's keylog).

server_context(host)

The context presenting a leaf for host (minted on first use).

Source code in src/netproxy/intercept.py

def server_context(self, host: str) -> ssl.SSLContext:
    """The context presenting a leaf for ``host`` (minted on first use)."""
    with self._lock:
        context = self._contexts.get(host)
        if context is not None:
            self._contexts.move_to_end(host)
            return context
    context = create_context("server")
    context.set_alpn_protocols(["http/1.1"])
    if self.keylog is not None:
        context.keylog_filename = self.keylog
    self.ca.issue([host], days=self.leaf_days).load_into(context)
    with self._lock:
        self._contexts[host] = context
        while len(self._contexts) > self.cache_size:
            self._contexts.popitem(last=False)
    return context

intercept(server, connection, route, reader, writer) async

Serve one intercepted CONNECT (the request head already read).

Source code in src/netproxy/intercept.py

async def intercept(
    self,
    server: Any,
    connection: Connection,
    route: Route,
    reader: asyncio.StreamReader,
    writer: asyncio.StreamWriter,
) -> None:
    """Serve one intercepted CONNECT (the request head already read)."""
    from ._http import _fail, _respond

    try:
        upstream = await server.open_upstream(connection, route)
    except (NetProxyError, OSError, asyncio.TimeoutError) as error:
        await _fail(writer, error, connection.authority)
        return
    relayed = False
    try:
        if reader._buffer:  # type: ignore[attr-defined]
            # Bytes sent before our 200 would bypass the TLS handshake.
            await _respond(writer, 400, "Bad Request", "data sent before 200\n")
            return
        context = await asyncio.to_thread(self.server_context, connection.host)
        # The client sends its ClientHello as soon as it reads the 200; stop
        # reading plaintext first, or it lands in the stream's buffer where
        # the TLS layer never sees it.
        writer.transport.pause_reading()
        writer.write(b"HTTP/1.1 200 Connection established\r\n\r\n")
        await start_tls(
            reader, writer, context, server_side=True, timeout=server.timeout
        )
        relayed = True
    except (OSError, ssl.SSLError, asyncio.TimeoutError) as error:
        _log.info(
            "netproxy: client TLS for %s failed: %s", connection.authority, error
        )
        return
    finally:
        if not relayed:
            # Never used, so nothing to flush; aborting is safe even when
            # cancelled, where awaiting a close would not be.
            upstream.writer.transport.abort()
    await server.relay_session(connection, route, reader, writer, upstream)

permitted_names(router)

The hosts and networks router's TLS routes can intercept, or None when some TLS rule (or the default) is not bounded by Host/Network matchers and no name constraint can cover it.

Source code in src/netproxy/intercept.py

def permitted_names(router: Router) -> Optional[List[str]]:
    """The hosts and networks ``router``'s TLS routes can intercept, or
    ``None`` when some TLS rule (or the default) is not bounded by
    ``Host``/``Network`` matchers and no name constraint can cover it."""
    if router.default.action is Action.TLS:
        return None
    names: List[str] = []
    for rule in router.rules:
        if rule.route.action is not Action.TLS:
            continue
        found = _names(rule.match)
        if found is None:
            return None
        names += [n for n in found if n not in names]
    return names

create_ca(router, common_name='netproxy local CA', *, days=3650, constrain=True)

A new CA for intercepting router's TLS routes.

With constrain (the default) it is name-constrained to :func:permitted_names; when the rules cannot be bounded that way this raises ValueError rather than create an unconstrained CA. Pass constrain=False to accept an unconstrained one knowingly.

Source code in src/netproxy/intercept.py

def create_ca(
    router: Router,
    common_name: str = "netproxy local CA",
    *,
    days: float = 3650,
    constrain: bool = True,
) -> CertificateAuthority:
    """A new CA for intercepting ``router``'s TLS routes.

    With ``constrain`` (the default) it is name-constrained to
    :func:`permitted_names`; when the rules cannot be bounded that way this
    raises ``ValueError`` rather than create an unconstrained CA. Pass
    ``constrain=False`` to accept an unconstrained one knowingly.
    """
    permitted: List[str] = []
    if constrain:
        names = permitted_names(router)
        if names is None:
            raise ValueError(
                "some TLS rule is not bounded by Host/Network matchers, so the CA "
                "cannot be name-constrained; pass constrain=False to accept that"
            )
        permitted = [n.lstrip(".") if n.startswith(".") else n for n in names]
    return CertificateAuthority.create(common_name, permitted=permitted, days=days)

Recording

Recording what passes through the proxy: a hook, a log, and pcap files.

A :class:Recorder sees each relayed connection as a :class:Session: when it opens, every chunk each side sends, when each direction ends, and when it closes. What it sees is what crossed the proxy: the plaintext of a TLS route (the proxy made, or ended, that TLS) and of the request and response heads it rewrote; the ciphertext of a tunnel it only relayed. Plaintext of an mTLS session is as sensitive as the credentials that opened it; store it as such.

A recorder that raises is logged and stops receiving that session; traffic is never affected.

Session(id, connection, route, client, server, started) dataclass

One relayed connection, as recorders see it.

client instance-attribute

The client's address.

server instance-attribute

The address the proxy connected to: the destination, or the upstream proxy that tunnelled to it.

started instance-attribute

Seconds since the epoch.

Recorder

Bases: Protocol

What the server calls for each relayed connection. Calls come from the event loop's thread, one session's in order; each must return quickly.

data(session, sender, data)

sender is "client" or "server".

Source code in src/netproxy/recording.py

def data(self, session: Session, sender: str, data: bytes) -> None:
    """``sender`` is ``"client"`` or ``"server"``."""

ended(session, sender)

sender's direction has no more data.

Source code in src/netproxy/recording.py

def ended(self, session: Session, sender: str) -> None:
    """``sender``'s direction has no more data."""

LogRecorder(logger='netproxy.traffic', level=logging.INFO)

Logs each session's opening and closing, with the octets each way, at level on logger (default netproxy.traffic, INFO).

Source code in src/netproxy/recording.py

def __init__(
    self,
    logger: Union[str, logging.Logger] = "netproxy.traffic",
    level: int = logging.INFO,
) -> None:
    self.logger = logging.getLogger(logger) if isinstance(logger, str) else logger
    self.level = level
    self._counts: Dict[int, List[int]] = {}

PcapRecorder(path, *, format='pcap')

Writes every session to a capture file as a TCP connection.

Each session becomes one connection between the client's address and the address the proxy connected to, in the file at path (pcap, or pcapng when format="pcapng"). The octets are those the proxy relayed, so a TLS route is recorded in plaintext on its TLS port: in Wireshark, Decode As HTTP. Needs the capture extra (pktcap). Thread-safe; close it (or use with) to flush the file.

Source code in src/netproxy/recording.py

def __init__(
    self, path: Union[str, "os.PathLike[str]"], *, format: str = "pcap"
) -> None:
    try:
        import pktcap
    except ImportError:
        raise ImportError(
            'recording to a capture needs the "capture" extra: '
            'pip install "netproxy[capture]"'
        ) from None
    if format not in ("pcap", "pcapng"):
        raise ValueError(f"format must be 'pcap' or 'pcapng', not {format!r}")
    writer_class = pktcap.PcapWriter if format == "pcap" else pktcap.PcapngWriter
    self._pktcap = pktcap
    self._writer = writer_class(os.fspath(path))
    self._flows: Dict[int, Any] = {}
    self._lock = threading.Lock()

Configuration files

Configuration files: read a document, then build a server from it.

The file is netproxy.cfg; its first line may name its format (#!toml, #!yaml, #!json, #!ini), and without one it is TOML (see :func:loads). Documents are read by anycfg. :func:loads and :func:load return plain data. :func:build_server turns that data into an :class:~netproxy.AsyncProxyServer; :func:load_server does both. Relative paths in the document are relative to the file's directory. Secrets never live in the file: PINs and key passwords are named by environment variable (pin_env, password_env), read when the credential is first used.

Document shape, in TOML (every table optional)::

[server]
listen = "127.0.0.1:3128"            # or a list
egress = "system"                     # "system" | "env" | "auto" | "direct" | "pac:<url>" | a proxy URL
timeout = 30.0
idle_timeout = 300.0
allow_remote = false
record = "proxy.pcap"                 # every relayed session, as a capture (extra "capture")
keylog = "tls-keys.log"               # TLS secrets for Wireshark; a secret itself
log_traffic = true                    # a log line per session opened and closed
ca = { certificate = "ca.pem", key = "ca.key", password_env = "NETPROXY_CA_PASSWORD" }

[credentials.work]                    # backend = "cng" | "pkcs11" | "pem"
backend = "cng"
options = { store = "MY" }            # the backend's constructor options
query = { subject = "alice" }         # find_identity() criteria
pin_env = "NETPROXY_WORK_PIN"         # or pin = "keystore": the OS dialog / PIN pad
cafile = "corp-ca.pem"                # how the destination is verified (also verify = false)

[[rules]]
match = "*.corp.example"              # a pattern or a list of patterns (any of)
port = 443                            # optional
path = "^/api/"                       # optional regex, forwarded HTTP only
action = "tls"                        # "tls" | "direct" | "reject"
credential = "work"                   # required for "tls"
upstream = "direct"                   # optional: "direct", a proxy URL or a PAC result
name = "corp"

[default]
action = "direct"

[[tunnels]]
listen = "127.0.0.1:15432"
target = "db.corp.example:5432"
credential = "work"                   # optional: makes the tunnel's route TLS

In INI a dotted section nests ([credentials.work], [server.ca]), [rules.<name>] and [tunnels.<name>] are list entries in file order with that name, and values are typed: true/false (also yes/no/on/off), integers, decimals, and JSON lists, objects or strings (match = ["*.corp.example", "corp"]); anything else is a string.

loads(text, *, path=None, format=None)

Parse a configuration document into plain data. path labels errors and, by its suffix, may name the format.

The format is the one a #!<format> first line names (#!toml, #!yaml, #!json, #!ini; the line is not part of the document), else format, else the one path's suffix selects (.toml, .yaml/.yml, .json, .ini; not .cfg), else TOML. An INI document is read as described in the module documentation. Bytes are strict UTF-8 (one leading BOM ignored). Raises :class:ConfigError with lineno/colno when known; the message never quotes the document.

Source code in src/netproxy/config.py

def loads(
    text: Union[str, bytes], *, path: Optional[str] = None, format: Optional[str] = None
) -> Dict[str, Any]:
    """Parse a configuration document into plain data. ``path`` labels errors
    and, by its suffix, may name the format.

    The format is the one a ``#!<format>`` first line names (``#!toml``,
    ``#!yaml``, ``#!json``, ``#!ini``; the line is not part of the document),
    else ``format``, else the one ``path``'s suffix selects (``.toml``,
    ``.yaml``/``.yml``, ``.json``, ``.ini``; not ``.cfg``), else TOML. An INI
    document is read as described in the module documentation. Bytes are strict UTF-8 (one leading BOM
    ignored). Raises :class:`ConfigError` with ``lineno``/``colno`` when
    known; the message never quotes the document.
    """
    import anycfg
    from anycfg import formats

    if isinstance(text, bytes):
        try:
            text = text.decode("utf-8")
        except UnicodeDecodeError as error:
            raise ConfigError(f"not UTF-8 at byte {error.start}", path) from None
    if text.startswith("\ufeff"):
        text = text[1:]
    declared, _ = formats.declared(text)
    chosen = declared or format or _format_for(path) or "toml"
    try:
        if declared:
            data = anycfg.loads(text)  # removes the #! line
        else:
            data = anycfg.loads(text, format=chosen)
    except anycfg.MissingDependencyError as error:
        raise ConfigError(f"{chosen}: {error}", path) from None
    except anycfg.UnsupportedFormatError as error:
        raise ConfigError(str(error), path) from None
    except anycfg.ParseError as error:
        raise ConfigError(error.msg, path, error.lineno, error.colno) from None
    except anycfg.ConfigError as error:
        raise ConfigError(str(error), path) from None
    if formats.get(chosen).name == "ini":
        return _from_ini(data, path)
    if not isinstance(data, dict):
        raise ConfigError("the document is not a table", path)
    return data

load(source, *, format=None)

Read and parse the configuration file at source (OSError if unreadable); the format is chosen as :func:loads chooses it.

Source code in src/netproxy/config.py

def load(
    source: Union[str, "os.PathLike[str]"], *, format: Optional[str] = None
) -> Dict[str, Any]:
    """Read and parse the configuration file at ``source`` (``OSError`` if
    unreadable); the format is chosen as :func:`loads` chooses it."""
    path = FilePath(source)
    return loads(path.read_bytes(), path=str(path), format=format)

default_path()

NETPROXY_CONFIG, else netproxy/netproxy.cfg in the platform's configuration directory (%APPDATA% on Windows, $XDG_CONFIG_HOME or ~/.config elsewhere).

Source code in src/netproxy/config.py

def default_path() -> FilePath:
    """``NETPROXY_CONFIG``, else ``netproxy/netproxy.cfg`` in the platform's
    configuration directory (``%APPDATA%`` on Windows, ``$XDG_CONFIG_HOME`` or
    ``~/.config`` elsewhere)."""
    configured = os.environ.get(_CONFIG_ENV)
    if configured:
        return FilePath(configured)
    if sys.platform == "win32" and os.environ.get("APPDATA"):
        base = FilePath(os.environ["APPDATA"])
    else:
        base = FilePath(
            os.environ.get("XDG_CONFIG_HOME") or FilePath.home() / ".config"
        )
    return base / "netproxy" / "netproxy.cfg"

build_router(document, *, base_dir=None)

The :class:~netproxy.Router the document's rules and default describe.

Source code in src/netproxy/config.py

def build_router(
    document: Mapping[str, Any],
    *,
    base_dir: Union[str, "os.PathLike[str]", None] = None,
) -> Router:
    """The :class:`~netproxy.Router` the document's rules and default describe."""
    builder = _Builder(document, FilePath(base_dir or "."), "<document>")
    return builder.router(builder.credentials())

build_server(document, *, base_dir=None, label='<document>')

An :class:~netproxy.AsyncProxyServer for document (from :func:loads).

Raises :class:ConfigError naming the table and key at fault, or OSError for an egress PAC that cannot be fetched.

Source code in src/netproxy/config.py

def build_server(
    document: Mapping[str, Any],
    *,
    base_dir: Union[str, "os.PathLike[str]", None] = None,
    label: str = "<document>",
) -> AsyncProxyServer:
    """An :class:`~netproxy.AsyncProxyServer` for ``document`` (from :func:`loads`).

    Raises :class:`ConfigError` naming the table and key at fault, or
    ``OSError`` for an egress PAC that cannot be fetched.
    """
    return _Builder(document, FilePath(base_dir or "."), label).server()

load_server(source)

:func:load then :func:build_server, paths relative to the file.

Source code in src/netproxy/config.py

def load_server(source: Union[str, "os.PathLike[str]"]) -> AsyncProxyServer:
    """:func:`load` then :func:`build_server`, paths relative to the file."""
    path = FilePath(source)
    return build_server(load(path), base_dir=path.parent, label=str(path))

PAC scripts

A proxy auto-config (PAC) script generated from a router's rules.

Clients that read a PAC (browsers, OS proxy settings) then send only what needs the proxy through it. A rule PAC JavaScript cannot express exactly (a path, an IPv6 network, a custom callable) sends everything still undecided to the proxy, which routes it exactly: the script errs towards the proxy, never away from it.

generate_pac(router, proxy, *, egress=None)

A PAC script sending to proxy (host:port) every request router does not send straight to its destination.

A route goes DIRECT in the script only when the proxy itself would connect directly: a DIRECT route whose upstream is "direct", or none with no egress. Rules are evaluated in order, the default last.

Source code in src/netproxy/pac.py

def generate_pac(router: Router, proxy: str, *, egress: Any = None) -> str:
    """A PAC script sending to ``proxy`` (``host:port``) every request
    ``router`` does not send straight to its destination.

    A route goes DIRECT in the script only when the proxy itself would connect
    directly: a DIRECT route whose upstream is ``"direct"``, or none with no
    ``egress``. Rules are evaluated in order, the default last.
    """
    if not isinstance(router, Router):
        raise TypeError("generate_pac needs a netproxy Router")
    lines: List[str] = [
        "// Generated by netproxy from its routing rules.",
        _HELPERS,
        "function FindProxyForURL(url, host) {",
        "  host = host.toLowerCase();",
    ]
    for rule in router.rules:
        outcome = _outcome(rule.route, proxy, egress)
        expression = _expression(rule.match)
        if expression is None:
            lines.append(
                f"  return {_js('PROXY ' + proxy)};  // a rule PAC cannot express"
            )
            lines.append("}")
            return "\n".join(lines) + "\n"
        lines.append(f"  if ({expression}) return {_js(outcome)};")
    lines.append(f"  return {_js(_outcome(router.default, proxy, egress))};")
    lines.append("}")
    return "\n".join(lines) + "\n"

Exceptions

Errors raised by :mod:netproxy.

A caller's bad argument raises plain :class:TypeError or :class:ValueError; everything here is a failure found at run time.

NetProxyError

Bases: Exception

Base class for every error this package raises.

UpstreamError

Bases: NetProxyError, OSError

The destination, or an upstream proxy on the way to it, could not be reached or refused the connection.

RejectedError

Bases: NetProxyError

A rule rejected the connection.

CredentialUnavailableError

Bases: NetProxyError

A route's client credential could not be loaded (keystore entry missing, no PIN, keystore failure).

ConfigError(msg, path=None, lineno=None, colno=None)

Bases: NetProxyError, ValueError

A configuration document is malformed or inconsistent.

path, lineno and colno locate the problem when known; the message never quotes the document.

Source code in src/netproxy/exceptions.py

def __init__(
    self,
    msg: str,
    path: str | None = None,
    lineno: int | None = None,
    colno: int | None = None,
) -> None:
    where = ":".join(str(p) for p in (path, lineno, colno) if p is not None)
    super().__init__(f"{where}: {msg}" if where else msg)
    self.msg = msg
    self.path = path
    self.lineno = lineno
    self.colno = colno