proxy_server.py

"""Lightweight HTTP and SOCKS5 proxy server.

Environment variables:
    LISTEN_HOST  - Interface to bind (default: 0.0.0.0)
    HTTP_PORT    - HTTP proxy port (default: 8080)
    SOCKS_PORT   - SOCKS5 proxy port (default: 1080)
    PROXY_USER   - Optional username for auth
    PROXY_PASS   - Optional password for auth
"""

import asyncio
import logging
import os
import socket

import pproxy

DEFAULT_LISTEN_HOST = "0.0.0.0"
DEFAULT_HTTP_PORT = 8080
DEFAULT_SOCKS_PORT = 1080


def _build_auth_fragment() -> str:
    user = os.getenv("PROXY_USER")
    password = os.getenv("PROXY_PASS")
    if user and password:
        return f"#{user}:{password}"
    if user or password:
        raise ValueError("Both PROXY_USER and PROXY_PASS must be set together.")
    return ""


def _env_int(name: str, default: int) -> int:
    raw = os.getenv(name, str(default))
    try:
        return int(raw)
    except ValueError as exc:
        raise ValueError(f"{name} must be an integer (got {raw!r}).") from exc


def _local_ipv4_addresses() -> list[str]:
    addresses: set[str] = set()

    try:
        hostname = socket.gethostname()
        for info in socket.getaddrinfo(hostname, None, family=socket.AF_INET):
            addr = info[4][0]
            if addr and not addr.startswith("127."):
                addresses.add(addr)
    except OSError:
        pass

    try:
        with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
            sock.connect(("8.8.8.8", 80))
            addr = sock.getsockname()[0]
            if addr and not addr.startswith("127."):
                addresses.add(addr)
    except OSError:
        pass

    return sorted(addresses)


def _configure_logging() -> None:
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s | %(levelname)s | %(message)s",
    )


async def main() -> None:
    _configure_logging()

    host = os.getenv("LISTEN_HOST", DEFAULT_LISTEN_HOST)
    http_port = _env_int("HTTP_PORT", DEFAULT_HTTP_PORT)
    socks_port = _env_int("SOCKS_PORT", DEFAULT_SOCKS_PORT)
    auth = _build_auth_fragment()

    http_uri = f"http://{host}:{http_port}{auth}"
    socks_uri = f"socks5://{host}:{socks_port}{auth}"

    args = {
        "rserver": [],  # direct outbound (uses system routing / VPN)
        "verbose": lambda message: logging.info("pproxy: %s", message),
    }

    http_server = pproxy.Server(http_uri)
    socks_server = pproxy.Server(socks_uri)

    http_handler = await http_server.start_server(args)
    socks_handler = await socks_server.start_server(args)

    logging.info("HTTP proxy listening on %s:%s", host, http_port)
    logging.info("SOCKS5 proxy listening on %s:%s", host, socks_port)
    for addr in _local_ipv4_addresses():
        logging.info("Use this LAN IP on other devices: %s", addr)

    try:
        await asyncio.Event().wait()
    except KeyboardInterrupt:
        logging.info("Shutdown requested by user.")
    finally:
        http_handler.close()
        socks_handler.close()
        await http_handler.wait_closed()
        await socks_handler.wait_closed()


if __name__ == "__main__":
    asyncio.run(main())