The `proxy-protocol` library provides an implementation of the PROXY protocol, including an `asyncio` server implementation, for Python. It currently supports versions 0.11.3 and is actively maintained with regular minor releases, focusing on improvements, bug fixes, and broader platform support.
Install & Compatibility
Where this runs
tested against v0.11.3 · pip install
no network on importno background threads
Install × environment matrix
Each cell = how many times install + import succeeded across repeated harness runs. Partial = flaky.
glibc = Debian/Ubuntu slim · musl = Alpine Linux
muslpy 3.10–3.920 runs
installs and imports cleanly · install 0.0s · import 0.068s · 18.4MB
glibcpy 3.10–3.920 runs
installs and imports cleanly · install 1.7s · import 0.060s · 19MB
16MB installed
● package 16MB
Code
Verified usage
Verified import paths — ran on the pinned version, not inferred.
from proxyprotocol.detect import ProxyProtocolDetect
from proxyprotocol.reader import ProxyProtocolReader
from proxyprotocol.sock import SocketInfo
from proxyprotocol.version import ProxyProtocolVersion
This example sets up an asyncio server that listens for incoming connections. It uses `ProxyProtocolDetect` to automatically detect PROXY protocol v1 or v2 headers. Once a connection is established and the header is parsed (if present), the `on_connection` callback receives a `SocketInfo` object containing the original client and destination address details, and then echoes any received data back to the client. This demonstrates how to integrate `proxy-protocol` with a standard `asyncio.start_server` setup.
import asyncio
from asyncio import StreamReader, StreamWriter
from proxyprotocol.detect import ProxyProtocolDetect
from proxyprotocol.reader import ProxyProtocolReader
from proxyprotocol.sock import SocketInfo
async def on_connection(reader: StreamReader, writer: StreamWriter, info: SocketInfo) -> None:
print(f"Connection from: {info.family.name} {info.peername}")
print(f"Original client: {info.source_addr}:{info.source_port} -> {info.dest_addr}:{info.dest_port}")
# Echo back received data (optional, for demonstration)
while True:
data = await reader.read(1024)
if not data: break
writer.write(data)
await writer.drain()
writer.close()
await writer.wait_closed()
async def main(host: str, port: int) -> None:
pp_detect = ProxyProtocolDetect()
callback = ProxyProtocolReader(pp_detect).get_callback(on_connection)
server = await asyncio.start_server(callback, host, port)
async with server:
await server.serve_forever()
if __name__ == '__main__':
# To test, configure your proxy (e.g., HAProxy, NGINX) to send PROXY protocol
# to localhost:10007. Then connect to the proxy directly.
# Example with `netcat` after a proxy is set up:
# echo 'hello' | nc -q 1 localhost 10007
try:
asyncio.run(main('127.0.0.1', 10007))
except KeyboardInterrupt:
print("Server stopped.")
Upgrade
Version history
Breaking-change detection hasn't run for this library yet.
Audit
Security & dependencies
CVE tracking and dependency tree are planned for a later release.